[spark] Resolve primary-key and bucket-key properties like other identifiers

paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+class KeyPropertyCaseResolutionTest extends KeyPropertyCaseResolutionTestBase

paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+class KeyPropertyCaseResolutionTest extends KeyPropertyCaseResolutionTestBase

paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+class KeyPropertyCaseResolutionTest extends KeyPropertyCaseResolutionTestBase

paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+class KeyPropertyCaseResolutionTest extends KeyPropertyCaseResolutionTestBase

paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+class KeyPropertyCaseResolutionTest extends KeyPropertyCaseResolutionTestBase

paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala

@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+class KeyPropertyCaseResolutionTest extends KeyPropertyCaseResolutionTestBase

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java

@@ -74,6 +74,7 @@
 import org.apache.spark.sql.execution.datasources.DataSource;
 import org.apache.spark.sql.execution.datasources.FileFormat;
 import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2;
+import org.apache.spark.sql.internal.SQLConf;
 import org.apache.spark.sql.types.ArrayType;
 import org.apache.spark.sql.types.BinaryType;
 import org.apache.spark.sql.types.StructField;
@@ -89,6 +90,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -351,7 +353,9 @@ public SparkTable loadTable(Identifier ident, long timestamp) throws NoSuchTable
     public org.apache.spark.sql.connector.catalog.Table alterTable(
             Identifier ident, TableChange... changes) throws NoSuchTableException {
         List<SchemaChange> schemaChanges =
-                Arrays.stream(changes).map(this::toSchemaChange).collect(Collectors.toList());
+                Arrays.stream(changes)
+                        .map(change -> toSchemaChange(ident, change))
+                        .collect(Collectors.toList());
         try {
             catalog.alterTable(toIdentifier(ident, catalogName), schemaChanges, false);
             return loadTable(ident);
@@ -475,12 +479,17 @@ private StagedTable stageReplaceByDropAndCreate(
         return new RollbackStagedTable(loadTable(ident), () -> {});
     }
 
-    private SchemaChange toSchemaChange(TableChange change) {
+    private SchemaChange toSchemaChange(Identifier ident, TableChange change) {
         if (change instanceof TableChange.SetProperty) {
             TableChange.SetProperty set = (TableChange.SetProperty) change;
             validateAlterProperty(set.property());
             if (set.property().equals(TableCatalog.PROP_COMMENT)) {
                 return SchemaChange.updateComment(set.value());
+            } else if (set.property().equals(CoreOptions.BUCKET_KEY.key())) {
+                // Same reasoning as in `toInitialSchema`; `primary-key` cannot reach here because
+                // `validateAlterProperty` rejects altering it.
+                return SchemaChange.setOption(
+                        set.property(), resolveColumnNameList(set.value(), fieldNames(ident)));
             } else {
                 return SchemaChange.setOption(set.property(), set.value());
             }
@@ -597,6 +606,21 @@ private Schema toInitialSchema(
                         : Arrays.stream(pkAsString.split(","))
                                 .map(String::trim)
                                 .collect(Collectors.toList());
+
+        // `primary-key` and `bucket-key` are plain strings in TBLPROPERTIES, so unlike the
+        // partition columns -- which Spark's analyzer has already resolved -- they reach Paimon
+        // exactly as the user typed them and are then matched against the schema exactly. Resolve
+        // them here so that they follow the session's `spark.sql.caseSensitive` semantics, the same
+        // way every other identifier in the statement does.
+        List<String> fieldNames =
+                Arrays.stream(schema.fields()).map(StructField::name).collect(Collectors.toList());
+        primaryKeys = resolveColumnNames(primaryKeys, fieldNames);
+        String bucketKey = normalizedProperties.get(CoreOptions.BUCKET_KEY.key());
+        if (bucketKey != null) {
+            normalizedProperties.put(
+                    CoreOptions.BUCKET_KEY.key(), resolveColumnNameList(bucketKey, fieldNames));
+        }
+
         Schema.Builder schemaBuilder =
                 Schema.newBuilder()
                         .options(normalizedProperties)
@@ -692,6 +716,46 @@ private static DataType toBlobType(StructField field, boolean allowNested) {
                 "The type of blob field must be binary, array of binary, or map with binary values");
     }
 
+    /** Column names of the table as it exists now, or empty if it cannot be loaded. */
+    private List<String> fieldNames(Identifier ident) {
+        if (SQLConf.get().caseSensitiveAnalysis()) {
+            // The names are used verbatim in that case, so skip loading the table.
+            return Collections.emptyList();
+        }
+        try {
+            return catalog.getTable(toIdentifier(ident, catalogName)).rowType().getFieldNames();
+        } catch (Catalog.TableNotExistException e) {
+            return Collections.emptyList();
+        }
+    }
+
+    /**
+     * Rewrites {@code names} to the way the columns are actually spelled in the schema, following
+     * the session's case-sensitivity setting. A name that matches no column is passed through
+     * untouched so that Paimon's own validation reports it.
+     */
+    private static List<String> resolveColumnNames(List<String> names, List<String> fieldNames) {
+        if (names.isEmpty() || SQLConf.get().caseSensitiveAnalysis()) {
+            return names;
+        }
+        Map<String, String> lowerToActual = new HashMap<>();
+        for (String fieldName : fieldNames) {
+            // A schema with two columns differing only in case cannot exist under
+            // case-insensitive analysis, so the first match is the only match.
+            lowerToActual.putIfAbsent(fieldName.toLowerCase(Locale.ROOT), fieldName);
+        }
+        return names.stream()
+                .map(name -> lowerToActual.getOrDefault(name.toLowerCase(Locale.ROOT), name))
+                .collect(Collectors.toList());
+    }
+
+    /** {@link #resolveColumnNames} for a comma separated option value such as `bucket-key`. */
+    private static String resolveColumnNameList(String value, List<String> fieldNames) {
+        List<String> names =
+                Arrays.stream(value.split(",")).map(String::trim).collect(Collectors.toList());
+        return String.join(",", resolveColumnNames(names, fieldNames));
+    }
+
     private void validateAlterProperty(String alterKey) {
         if (PRIMARY_KEY_IDENTIFIER.equals(alterKey)) {
             throw new UnsupportedOperationException("Alter primary key is not supported");

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTestBase.scala

@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import org.apache.spark.sql.Row
+
+import scala.collection.JavaConverters._
+
+/**
+ * `primary-key` and `bucket-key` are plain strings in TBLPROPERTIES, so they used to be matched
+ * against the schema exactly while every other identifier in the same statement -- notably the
+ * partition columns -- followed `spark.sql.caseSensitive`.
+ */
+abstract class KeyPropertyCaseResolutionTestBase extends PaimonSparkTestBase {
+
+  test("Paimon DDL: primary-key property resolves like every other identifier") {
+    withTable("t") {
+      spark.sql("""CREATE TABLE t (Id INT, Name STRING)
+                  | TBLPROPERTIES ('primary-key'='id', 'bucket'='1')""".stripMargin)
+      // The key is stored under the column's real spelling, not the one the user typed.
+      assert(loadTable("t").primaryKeys().asScala == Seq("Id"))
+
+      spark.sql("INSERT INTO t VALUES (1, 'a')")
+      spark.sql("INSERT INTO t VALUES (1, 'b')")
+      checkAnswer(spark.sql("SELECT * FROM t"), Row(1, "b") :: Nil)
+    }
+  }
+
+  test("Paimon DDL: bucket-key property resolves like every other identifier") {
+    withTable("t") {
+      spark.sql("""CREATE TABLE t (Id INT, Name STRING)
+                  | TBLPROPERTIES ('bucket-key'='id', 'bucket'='2')""".stripMargin)
+      assert(loadTable("t").schema().options().get("bucket-key") == "Id")
+
+      spark.sql("INSERT INTO t VALUES (1, 'a'), (2, 'b')")
+      checkAnswer(spark.sql("SELECT * FROM t ORDER BY Id"), Row(1, "a") :: Row(2, "b") :: Nil)
+    }
+  }
+
+  test("Paimon DDL: multi column and partitioned keys resolve") {
+    withTable("t") {
+      spark.sql("""CREATE TABLE t (Id INT, Sub INT, Pt STRING, V STRING)
+                  | PARTITIONED BY (pt)
+                  | TBLPROPERTIES ('primary-key'='id,sub,pt', 'bucket'='1')""".stripMargin)
+      assert(loadTable("t").primaryKeys().asScala == Seq("Id", "Sub", "Pt"))
+      assert(loadTable("t").partitionKeys().asScala == Seq("Pt"))
+
+      spark.sql("INSERT INTO t VALUES (1, 1, 'p1', 'a')")
+      spark.sql("INSERT INTO t VALUES (1, 1, 'p1', 'b')")
+      checkAnswer(spark.sql("SELECT V FROM t"), Row("b") :: Nil)
+    }
+  }
+
+  test("Paimon DDL: ALTER TABLE SET TBLPROPERTIES resolves bucket-key") {
+    withTable("t") {
+      spark.sql("""CREATE TABLE t (Id INT, Name STRING)
+                  | TBLPROPERTIES ('bucket-key'='Id', 'bucket'='2')""".stripMargin)
+      spark.sql("ALTER TABLE t SET TBLPROPERTIES ('bucket-key'='id')")
+      assert(loadTable("t").schema().options().get("bucket-key") == "Id")
+    }
+  }
+
+  test("Paimon DDL: an exact spelling keeps working") {
+    withTable("t") {
+      spark.sql(
+        """CREATE TABLE t (Id INT, Name STRING)
+          | TBLPROPERTIES ('primary-key'='Id', 'bucket-key'='Id', 'bucket'='2')""".stripMargin)
+      assert(loadTable("t").primaryKeys().asScala == Seq("Id"))
+      assert(loadTable("t").schema().options().get("bucket-key") == "Id")
+    }
+  }
+
+  test("Paimon DDL: an unknown column still reports the original name") {
+    withTable("t") {
+      val e = intercept[Exception] {
+        spark.sql("""CREATE TABLE t (Id INT, Name STRING)
+                    | TBLPROPERTIES ('primary-key'='nope', 'bucket'='1')""".stripMargin)
+      }
+      assert(e.getMessage.contains("nope"), e.getMessage)
+    }
+  }
+
+  test("Paimon DDL: case sensitive analysis keeps the exact match requirement") {
+    withSparkSQLConf("spark.sql.caseSensitive" -> "true") {
+      withTable("t") {
+        val e = intercept[Exception] {
+          spark.sql("""CREATE TABLE t (Id INT, Name STRING)
+                      | TBLPROPERTIES ('primary-key'='id', 'bucket'='1')""".stripMargin)
+        }
+        assert(e.getMessage.contains("id"), e.getMessage)
+      }
+    }
+  }
+}