From 6e8c03d5825ebad742b8f973043872d437e2aa7c Mon Sep 17 00:00:00 2001
From: LsomeYeah <94825748+LsomeYeah@users.noreply.github.com>
Date: Wed, 20 Nov 2024 16:05:51 +0800
Subject: [PATCH] [core] all catalogs use 'allow-upper-case' to control
 case-sensitive (#4555)

---
 .../catalog/FileSystemCatalogOptions.java     |  1 +
 .../paimon/catalog/FileSystemCatalogTest.java | 39 ++++++++++++++++++-
 2 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogOptions.java b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogOptions.java
index 962b249bac67..e656742b42e9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogOptions.java
@@ -28,6 +28,7 @@ public final class FileSystemCatalogOptions {
             ConfigOptions.key("case-sensitive")
                     .booleanType()
                     .defaultValue(true)
+                    .withFallbackKeys("allow-upper-case")
                     .withDescription(
                             "Is case sensitive. If case insensitive, you need to set this option to false, and the table name and fields be converted to lowercase.");
 
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
index 35a5c06ffbc6..798402c6d7db 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
@@ -19,8 +19,15 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.types.DataTypes;
 
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 /** Tests for {@link FileSystemCatalog}. */
 public class FileSystemCatalogTest extends CatalogTestBase {
@@ -28,6 +35,36 @@ public class FileSystemCatalogTest extends CatalogTestBase {
     @BeforeEach
     public void setUp() throws Exception {
         super.setUp();
-        catalog = new FileSystemCatalog(fileIO, new Path(warehouse));
+        Options catalogOptions = new Options();
+        catalogOptions.set(CatalogOptions.ALLOW_UPPER_CASE, false);
+        catalog = new FileSystemCatalog(fileIO, new Path(warehouse), catalogOptions);
+    }
+
+    @Test
+    public void testCreateTableAllowUpperCase() throws Exception {
+        catalog.createDatabase("test_db", false);
+        Identifier identifier = Identifier.create("test_db", "new_table");
+        Schema schema =
+                Schema.newBuilder()
+                        .column("Pk1", DataTypes.INT())
+                        .column("pk2", DataTypes.STRING())
+                        .column("pk3", DataTypes.STRING())
+                        .column(
+                                "Col1",
+                                DataTypes.ROW(
+                                        DataTypes.STRING(),
+                                        DataTypes.BIGINT(),
+                                        DataTypes.TIMESTAMP(),
+                                        DataTypes.ARRAY(DataTypes.STRING())))
+                        .column("col2", DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT()))
+                        .column("col3", DataTypes.ARRAY(DataTypes.ROW(DataTypes.STRING())))
+                        .partitionKeys("Pk1", "pk2")
+                        .primaryKey("Pk1", "pk2", "pk3")
+                        .build();
+
+        // Create table throws Exception when table is system table
+        assertThatExceptionOfType(IllegalArgumentException.class)
+                .isThrownBy(() -> catalog.createTable(identifier, schema, false))
+                .withMessage("Field name [Pk1, Col1] cannot contain upper case in the catalog.");
     }
 }