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 000000000000..bd2b6c98f210 --- /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 000000000000..bd2b6c98f210 --- /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 000000000000..bd2b6c98f210 --- /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 000000000000..bd2b6c98f210 --- /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 000000000000..bd2b6c98f210 --- /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 000000000000..bd2b6c98f210 --- /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 16fc1806d1e2..e9f737ace3b4 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.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 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 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 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 resolveColumnNames(List names, List fieldNames) { + if (names.isEmpty() || SQLConf.get().caseSensitiveAnalysis()) { + return names; + } + Map 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 fieldNames) { + List 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 000000000000..42d9abbb21f1 --- /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) + } + } + } +}