This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cdc5eee955 [spark] Resolve primary-key and bucket-key properties like
other identifiers (#9372)
cdc5eee955 is described below
commit cdc5eee9553c2782e7cce06e37ffd377e3e9a238
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Tue Aug 25 09:18:51 2026 +0800
[spark] Resolve primary-key and bucket-key properties like other
identifiers (#9372)
---
.../spark/sql/KeyPropertyCaseResolutionTest.scala | 21 ++++
.../spark/sql/KeyPropertyCaseResolutionTest.scala | 21 ++++
.../spark/sql/KeyPropertyCaseResolutionTest.scala | 21 ++++
.../spark/sql/KeyPropertyCaseResolutionTest.scala | 21 ++++
.../spark/sql/KeyPropertyCaseResolutionTest.scala | 21 ++++
.../spark/sql/KeyPropertyCaseResolutionTest.scala | 21 ++++
.../java/org/apache/paimon/spark/SparkCatalog.java | 68 ++++++++++++-
.../sql/KeyPropertyCaseResolutionTestBase.scala | 112 +++++++++++++++++++++
8 files changed, 304 insertions(+), 2 deletions(-)
diff --git
a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
new file mode 100644
index 0000000000..bd2b6c98f2
--- /dev/null
+++
b/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
diff --git
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
new file mode 100644
index 0000000000..bd2b6c98f2
--- /dev/null
+++
b/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
diff --git
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
new file mode 100644
index 0000000000..bd2b6c98f2
--- /dev/null
+++
b/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
diff --git
a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
new file mode 100644
index 0000000000..bd2b6c98f2
--- /dev/null
+++
b/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
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
new file mode 100644
index 0000000000..bd2b6c98f2
--- /dev/null
+++
b/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
diff --git
a/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTest.scala
new file mode 100644
index 0000000000..bd2b6c98f2
--- /dev/null
+++
b/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
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
index 16fc1806d1..e9f737ace3 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -74,6 +74,7 @@ import org.apache.spark.sql.connector.expressions.Transform;
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.Arrays;
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 class SparkCatalog extends SparkBaseCatalog
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 @@ public class SparkCatalog extends SparkBaseCatalog
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 @@ public class SparkCatalog extends SparkBaseCatalog
: 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 @@ public class SparkCatalog extends SparkBaseCatalog
"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");
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/KeyPropertyCaseResolutionTestBase.scala
new file mode 100644
index 0000000000..42d9abbb21
--- /dev/null
+++
b/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)
+ }
+ }
+ }
+}
