diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index 972ceea6fbf..2f9da1c31c7 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -832,9 +832,6 @@ The physical differences that every native data file has, and why they are safe, ([#7033](https://github.com/apache/datafusion-comet/issues/7033)). - **`VERSION AS OF` reads of a column whose type was promoted after the snapshot** ([#6993](https://github.com/apache/datafusion-comet/issues/6993)). -- **Column names that only newer Unicode case data folds together**, such as a pair of Vithkuqi - letters, which Unicode 14 added. Queries on such a table fail with a duplicate-field error - ([#5540](https://github.com/apache/datafusion-comet/issues/5540)). - **Iceberg tables whose metadata location has no host.** The native Iceberg scan skips the credential provider configured with `s3.comet.credential.provider.class` ([#6536](https://github.com/apache/datafusion-comet/issues/6536)). diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index 3065c75aef5..640c7896a98 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -236,8 +236,7 @@ impl IcebergScanExec { let scan_metrics = scan_result.metrics().clone(); let stream = scan_result.stream(); - let spark_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); - let adapter_factory = SparkPhysicalExprAdapterFactory::new(spark_options, None); + let adapter_factory = SparkPhysicalExprAdapterFactory::new(iceberg_parquet_options(), None); let adapted_stream = stream.map_err(|e| DataFusionError::Execution(format!("Iceberg scan error: {}", e))); @@ -488,12 +487,13 @@ where .adapter_factory .create(Arc::clone(&self.schema), Arc::clone(&file_schema))?; let exprs = - build_projection_expressions(&self.schema, &adapter).map_err(|e| { - DataFusionError::Execution(format!( - "Failed to build projection expressions: {}", - e - )) - })?; + build_projection_expressions(&self.schema, &file_schema, &adapter) + .map_err(|e| { + DataFusionError::Execution(format!( + "Failed to build projection expressions: {}", + e + )) + })?; self.cached = Some(CachedProjection { file_schema, projection_exprs: exprs, @@ -576,21 +576,36 @@ impl fmt::Debug for RedactedProperties<'_> { } } +/// Exact name matching: iceberg-rust batches and Spark's attributes carry the table's own names. +fn iceberg_parquet_options() -> SparkParquetOptions { + let mut options = SparkParquetOptions::new(EvalMode::Legacy, "UTC"); + options.case_sensitive = true; + options +} + /// Build projection expressions that adapt batches from a file schema to the target schema. /// /// The returned expressions can be cached and reused across multiple batches /// that share the same file schema, avoiding repeated expression construction. +/// +/// A top-level column absent from `file_schema` is a bug: iceberg-rust emits every one it projects. fn build_projection_expressions( target_schema: &SchemaRef, + file_schema: &SchemaRef, adapter: &Arc, ) -> DFResult>> { target_schema .fields() .iter() - .enumerate() - .map(|(i, _field)| { + .map(|field| { + if file_schema.fields().find(field.name()).is_none() { + return Err(DataFusionError::Execution(format!( + "Iceberg scan column '{}' is missing from the batch read by iceberg-rust", + field.name() + ))); + } let col_expr: Arc = Arc::new(Column::new_with_schema( - target_schema.field(i).name(), + field.name(), target_schema.as_ref(), )?); adapter.rewrite(col_expr) @@ -658,9 +673,8 @@ mod tests { Field::new("a", DataType::Int64, false), Field::new("b", DataType::Int64, false), ])); - let mut options = super::SparkParquetOptions::new(super::EvalMode::Legacy, "UTC"); - options.case_sensitive = true; - let factory = super::SparkPhysicalExprAdapterFactory::new(options, None); + let factory = + super::SparkPhysicalExprAdapterFactory::new(super::iceberg_parquet_options(), None); for name in ["a", "b"] { let target = Arc::new(ArrowSchema::new(vec![Field::new( name, @@ -670,7 +684,7 @@ mod tests { let adapter = factory .create(Arc::clone(&target), Arc::clone(&physical)) .unwrap(); - let result = super::build_projection_expressions(&target, &adapter); + let result = super::build_projection_expressions(&target, &physical, &adapter); if name == "a" { let error = result .expect_err("selected root must be ambiguous") @@ -702,6 +716,78 @@ mod tests { } } + #[test] + fn projection_distinguishes_case_variant_roots() { + use arrow::array::Int64Array; + use arrow::datatypes::{DataType, Field, Schema as ArrowSchema}; + use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; + + let physical = Arc::new(ArrowSchema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("A", DataType::Int64, false), + ])); + let batch = super::RecordBatch::try_new( + Arc::clone(&physical), + vec![ + Arc::new(Int64Array::from(vec![1])), + Arc::new(Int64Array::from(vec![2])), + ], + ) + .unwrap(); + let factory = + super::SparkPhysicalExprAdapterFactory::new(super::iceberg_parquet_options(), None); + for (names, expected) in [ + (vec!["a", "A"], vec![1, 2]), + (vec!["A"], vec![2]), + (vec!["a"], vec![1]), + ] { + let target = Arc::new(ArrowSchema::new( + names + .iter() + .map(|name| Field::new(*name, DataType::Int64, false)) + .collect::>(), + )); + let adapter = factory + .create(Arc::clone(&target), Arc::clone(&physical)) + .unwrap(); + let exprs = super::build_projection_expressions(&target, &physical, &adapter) + .unwrap_or_else(|e| panic!("{names:?}: {e}")); + let output = + super::adapt_batch_with_expressions(batch.clone(), &target, &exprs).unwrap(); + let values: Vec = output + .columns() + .iter() + .map(|c| c.as_any().downcast_ref::().unwrap().value(0)) + .collect(); + assert_eq!(values, expected, "{names:?}"); + } + } + + #[test] + fn projection_rejects_missing_top_level_column() { + use arrow::datatypes::{DataType, Field, Schema as ArrowSchema}; + use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; + + let physical = Arc::new(ArrowSchema::new(vec![Field::new( + "a", + DataType::Int64, + true, + )])); + let target = Arc::new(ArrowSchema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new("b", DataType::Int64, true), + ])); + let factory = + super::SparkPhysicalExprAdapterFactory::new(super::iceberg_parquet_options(), None); + let adapter = factory + .create(Arc::clone(&target), Arc::clone(&physical)) + .unwrap(); + let error = super::build_projection_expressions(&target, &physical, &adapter) + .expect_err("a missing top-level column must not read as null") + .to_string(); + assert!(error.contains("'b' is missing"), "{error}"); + } + fn task_with_deletes(deletes: Vec) -> FileScanTask { FileScanTask::builder() .with_file_size_in_bytes(0) @@ -712,7 +798,7 @@ mod tests { .with_schema(Arc::new(Schema::builder().build().unwrap())) .with_project_field_ids(vec![]) .with_deletes(deletes) - .with_case_sensitive(false) + .with_case_sensitive(true) .build() .unwrap() } diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index a47c19d1a23..6247744b85a 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -2732,7 +2732,9 @@ fn parse_file_scan_tasks_from_common( .with_partition_spec(partition_spec) .with_name_mapping(name_mapping) .with_unified_partition_type(unified_partition_type_for_task) - .with_case_sensitive(false) + // Only the equality-delete bind reads this, and iceberg-rust already renamed those + // columns by field id; folding case would merge columns that differ only in case. + .with_case_sensitive(true) // Plaintext StandardKeyMetadata forwarded verbatim from the JVM; decoded by // iceberg-rust with no KMS unwrap. None for unencrypted data files. .with_key_metadata(proto_task.key_metadata.clone().map(Vec::into_boxed_slice)) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index 3c511b4f378..db591ddf26b 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -772,12 +772,8 @@ case class CometScanRule(session: SparkSession) // Projected roots are matched by field ID so historical snapshots still identify renamed // columns. Both schema checks below use them. val projectedDataColumns = scanExec.output.filterNot(_.isMetadataCol) - val resolver = session.sessionState.conf.resolver - val projectedFieldIds = projectedDataColumns.map { attr => - metadata.globalFieldIdMapping.collectFirst { - case (fieldName, fieldId) if resolver(fieldName, attr.name) => fieldId - } - } + val projectedFieldIds = + projectedDataColumns.map(attr => metadata.globalFieldIdMapping.get(attr.name)) val resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty) diff --git a/spark/src/test/resources/sql-tests/iceberg/case_sensitive_columns.sql b/spark/src/test/resources/sql-tests/iceberg/case_sensitive_columns.sql new file mode 100644 index 00000000000..8b06567796d --- /dev/null +++ b/spark/src/test/resources/sql-tests/iceberg/case_sensitive_columns.sql @@ -0,0 +1,85 @@ +-- 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. + +-- `a` and `A` each bind their own field: https://github.com/apache/datafusion-comet/issues/5540 + +-- No Iceberg spark-runtime is published for Spark 4.2 yet, and the 4.0 runtime the build reuses +-- is binary-incompatible with it. See https://github.com/apache/datafusion-comet/issues/4969. +-- MaxSparkVersion: 4.1 + +-- Config: spark.sql.catalog.test_cat=org.apache.iceberg.spark.SparkCatalog +-- Config: spark.sql.catalog.test_cat.type=hadoop +-- Config: spark.sql.catalog.test_cat.warehouse=/tmp/comet-iceberg-sql-test +-- Config: spark.comet.enabled=true +-- Config: spark.comet.exec.enabled=true +-- Config: spark.comet.scan.icebergNative.enabled=true +-- Config: spark.sql.caseSensitive=true + +statement +DROP TABLE IF EXISTS test_cat.db.case_pair + +statement +CREATE TABLE test_cat.db.case_pair (a INT, A INT) USING iceberg + +-- Disjoint ranges: a filter bound to the other column prunes the row group or keeps wrong rows. +statement +INSERT INTO test_cat.db.case_pair VALUES (1, 20), (3, 40) + +query +SELECT a, A FROM test_cat.db.case_pair ORDER BY a + +query +SELECT A, a FROM test_cat.db.case_pair ORDER BY a + +query +SELECT A FROM test_cat.db.case_pair ORDER BY A + +query +SELECT a FROM test_cat.db.case_pair ORDER BY a + +query +SELECT * FROM test_cat.db.case_pair ORDER BY a + +query +SELECT a FROM test_cat.db.case_pair WHERE A = 40 + +query +SELECT A FROM test_cat.db.case_pair WHERE a = 3 + +query +SELECT A FROM test_cat.db.case_pair WHERE A = 40 + +query +SELECT a FROM test_cat.db.case_pair WHERE a = 3 + +query +SELECT a, A FROM test_cat.db.case_pair WHERE A = 40 + +query +SELECT a, A FROM test_cat.db.case_pair WHERE a = 3 + +query +SELECT a FROM test_cat.db.case_pair WHERE A > 30 + +query +SELECT a FROM test_cat.db.case_pair WHERE A IN (20) + +query +SELECT A FROM test_cat.db.case_pair WHERE a < 2 + +statement +DROP TABLE test_cat.db.case_pair diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index e859b23be08..bc0ef4bc10c 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -1272,6 +1272,333 @@ class CometIcebergNativeSuite } } + /** Returns one message per query failing [[checkIcebergNativeScan]], so all are reported. */ + private def failingNativeScanChecks(queries: Seq[String]): Seq[String] = + queries.flatMap { q => + try { + checkIcebergNativeScan(q) + None + } catch { + case scala.util.control.NonFatal(e) => + val plan = + try spark.sql(q).queryExecution.executedPlan.toString + catch { case scala.util.control.NonFatal(pe) => s"" } + Some(s"QUERY: $q\nERROR: ${e.getMessage}\nPLAN:\n$plan") + } + } + + private def withCaseCatalog(warehouseDir: File, caseSensitive: Boolean)(f: => Unit): Unit = + withSQLConf( + "spark.sql.catalog.case_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.case_cat.type" -> "hadoop", + "spark.sql.catalog.case_cat.warehouse" -> warehouseDir.getAbsolutePath, + SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")(f) + + private def importParquetInto( + table: org.apache.iceberg.Table, + sourceTable: String, + stagingDir: String): Unit = + Class + .forName("org.apache.iceberg.spark.SparkTableUtil") + .getMethod( + "importSparkTable", + classOf[org.apache.spark.sql.SparkSession], + classOf[org.apache.spark.sql.catalyst.TableIdentifier], + classOf[org.apache.iceberg.Table], + classOf[String]) + .invoke( + null, + spark, + new org.apache.spark.sql.catalyst.TableIdentifier(sourceTable), + table, + stagingDir) + + test("issue 5540: equality deletes keyed on columns a and A") { + assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => + val t = "case_cat.db.case_pair_eq" + var failures = Seq.empty[String] + withCaseCatalog(warehouseDir, caseSensitive = true) { + spark.sql( + s"CREATE TABLE $t (a INT, A INT) USING iceberg TBLPROPERTIES ('format-version' = '2')") + spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40), (5, 60)") + // Binding either delete key by a folded name would misread one of them. + commitEqualityDelete("case_cat", "db", "case_pair_eq", "a", 1, warehouseDir) + commitEqualityDelete("case_cat", "db", "case_pair_eq", "A", 40, warehouseDir) + failures = failingNativeScanChecks( + Seq( + s"SELECT a, A FROM $t ORDER BY a", + s"SELECT A FROM $t ORDER BY A", + s"SELECT a FROM $t ORDER BY a", + s"SELECT count(*) FROM $t")) + } + // `a` is ambiguous here, so name no column; Spark 4.1+ rejects the duplicates itself. + withCaseCatalog(warehouseDir, caseSensitive = false) { + Seq(s"SELECT * FROM $t", s"SELECT count(*) FROM $t").foreach { query => + if (isSpark41Plus) { + val e = intercept[org.apache.spark.sql.AnalysisException](spark.sql(query).collect()) + assert(e.getMessage.contains("COLUMN_ALREADY_EXISTS"), e.getMessage) + } else { + checkIcebergNativeScan(query) + } + } + } + withCaseCatalog(warehouseDir, caseSensitive = true) { + spark.sql(s"DROP TABLE $t") + } + assert(failures.isEmpty, failures.mkString("\n\n=====\n\n")) + } + } + + test("issue 5540: case-insensitive control on mixed-case column names") { + assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => + withCaseCatalog(warehouseDir, caseSensitive = false) { + val t = "case_cat.db.mixed_case" + spark.sql( + s"CREATE TABLE $t (Id INT, Val INT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '2')") + spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40), (5, 60)") + commitEqualityDelete("case_cat", "db", "mixed_case", "Id", 1, warehouseDir) + val failures = failingNativeScanChecks( + Seq( + s"SELECT id, VAL FROM $t ORDER BY ID", + s"SELECT ID FROM $t WHERE val = 40", + s"SELECT * FROM $t WHERE ID > 2 ORDER BY id")) + spark.sql(s"DROP TABLE $t") + assert(failures.isEmpty, failures.mkString("\n\n=====\n\n")) + } + } + } + + test("issue 5540: equality delete survives a case-only column rename") { + assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => + val t = "case_cat.db.case_rename" + withCaseCatalog(warehouseDir, caseSensitive = true) { + spark.sql( + s"CREATE TABLE $t (id INT, val INT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '2')") + spark.sql(s"INSERT INTO $t VALUES (1, 10), (2, 20), (3, 30)") + commitEqualityDelete("case_cat", "db", "case_rename", "id", 2, warehouseDir) + spark.sql(s"ALTER TABLE $t RENAME COLUMN id TO ID") + } + Seq(true, false).foreach { caseSensitive => + withCaseCatalog(warehouseDir, caseSensitive) { + checkIcebergNativeScan(s"SELECT ID, val FROM $t ORDER BY ID") + checkIcebergNativeScan(s"SELECT count(*) FROM $t") + } + } + withCaseCatalog(warehouseDir, caseSensitive = true) { + spark.sql(s"DROP TABLE $t") + } + } + } + + test("issue 5540: name mapping onto physical columns that differ in case") { + assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => + val t = "case_cat.db.case_mapped" + val dataPath = s"${warehouseDir.getAbsolutePath}/case_mapped_src" + withCaseCatalog(warehouseDir, caseSensitive = false) { + spark + .sql("SELECT CAST(id AS INT) AS ID, CAST(id * 10 AS INT) AS Val FROM range(1, 4)") + .coalesce(1) + .write + .parquet(dataPath) + spark.sql(s"CREATE TABLE $t (id INT, val INT) USING iceberg") + // Set before the import so the file's column metrics are computed through the mapping. + spark.sql( + s"ALTER TABLE $t SET TBLPROPERTIES ('schema.name-mapping.default' = " + + """'[{"field-id":1,"names":["ID"]},{"field-id":2,"names":["Val"]}]')""") + val icebergTable = spark.sessionState.catalogManager + .catalog("case_cat") + .asInstanceOf[org.apache.iceberg.spark.SparkCatalog] + .loadTable( + org.apache.spark.sql.connector.catalog.Identifier.of(Array("db"), "case_mapped")) + .asInstanceOf[org.apache.iceberg.spark.source.SparkTable] + .table() + spark.sql(s"CREATE TABLE parquet_case_mapped USING parquet LOCATION '$dataPath'") + try { + importParquetInto(icebergTable, "parquet_case_mapped", s"$warehouseDir/staging") + Seq(true, false).foreach { caseSensitive => + withCaseCatalog(warehouseDir, caseSensitive) { + checkIcebergNativeScan(s"SELECT id, val FROM $t ORDER BY id") + checkIcebergNativeScan(s"SELECT val FROM $t WHERE id = 2") + } + } + } finally { + spark.sql("DROP TABLE IF EXISTS parquet_case_mapped") + spark.sql(s"DROP TABLE IF EXISTS $t") + } + } + } + } + + test("issue 5540: name mapping onto nested children that differ in case") { + assume(icebergAvailable, "Iceberg not available in classpath") + val name = "case_mapped_nested" + val t = s"case_cat.db.$name" + var failures = Seq.empty[String] + // Physical (s, x, y) names; the first matches the table exactly, as a control. + Seq(("s", "x", "y"), ("S", "X", "Y"), ("s", "X", "y")).foreach { case (sName, xName, yName) => + withTempIcebergDir { warehouseDir => + val dataPath = s"${warehouseDir.getAbsolutePath}/${name}_src" + withCaseCatalog(warehouseDir, caseSensitive = false) { + spark + .sql(s"SELECT CAST(id AS INT) AS id, named_struct('$xName', CAST(id * 10 AS INT), " + + s"'$yName', CAST(id * 100 AS INT)) AS $sName FROM range(1, 4)") + .coalesce(1) + .write + .parquet(dataPath) + spark.sql(s"CREATE TABLE $t (id INT, s STRUCT) USING iceberg") + val icebergTable = loadIcebergTable(spark, "case_cat", "db", name) + .asInstanceOf[org.apache.iceberg.Table] + def fieldId(path: String): Int = icebergTable.schema().findField(path).fieldId() + val mapping = + s"""[{"field-id":${fieldId("id")},"names":["id"]},""" + + s"""{"field-id":${fieldId("s")},"names":["$sName"],"fields":[""" + + s"""{"field-id":${fieldId("s.x")},"names":["$xName"]},""" + + s"""{"field-id":${fieldId("s.y")},"names":["$yName"]}]}]""" + spark.sql( + s"ALTER TABLE $t SET TBLPROPERTIES ('schema.name-mapping.default' = '$mapping')") + icebergTable.refresh() + spark.sql(s"CREATE TABLE parquet_$name USING parquet LOCATION '$dataPath'") + try { + importParquetInto(icebergTable, s"parquet_$name", s"$warehouseDir/staging") + // Spark reads the mapped children, so the checks below compare against real values. + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + assert( + spark.sql(s"SELECT id, s FROM $t ORDER BY id").collect().toSeq == + (1 to 3).map(i => Row(i, Row(i * 10, i * 100)))) + } + for { + caseSensitive <- Seq(false, true) + pruning <- Seq(true, false) + } { + withCaseCatalog(warehouseDir, caseSensitive) { + withSQLConf( + CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> pruning.toString) { + val label = s"file $sName<$xName, $yName>, caseSensitive=$caseSensitive, " + + s"nestedPruning=$pruning" + // Unchecked, as iceberg-rust maps top-level names only: `s` beside `id` is + // NULL (#6790), and a pruned `s.y` over physical `X`, `Y` reads `X` (#7134). + val skipsPrunedY = pruning && xName == "X" && yName == "Y" + val queries = Seq( + s"SELECT s FROM $t ORDER BY s.x", + s"SELECT s.x FROM $t ORDER BY s.x", + s"SELECT s.y FROM $t ORDER BY s.y", + s"SELECT s.y FROM $t WHERE s.x = 20") + .filterNot(q => skipsPrunedY && q.endsWith("ORDER BY s.y")) + failures ++= failingNativeScanChecks(queries).map(f => s"[$label]\n$f") + } + } + } + } finally { + spark.sql(s"DROP TABLE IF EXISTS parquet_$name") + spark.sql(s"DROP TABLE IF EXISTS $t") + } + } + } + } + assert(failures.isEmpty, failures.mkString("\n\n=====\n\n")) + } + + test("issue 5540: metadata columns beside data columns under case sensitivity") { + assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => + val t = "case_cat.db.case_meta" + var failures = Seq.empty[String] + withCaseCatalog(warehouseDir, caseSensitive = true) { + spark.sql( + s"CREATE TABLE $t (id INT, Region STRING, val INT) USING iceberg " + + "PARTITIONED BY (Region) TBLPROPERTIES ('format-version' = '2')") + spark.sql(s"INSERT INTO $t VALUES (1, 'a', 10), (2, 'a', 20), (3, 'b', 30)") + } + // Metadata columns resolve by reserved field id and keep Iceberg's reserved names, so the + // native scan's exact name check finds them too. + Seq(true, false).foreach { caseSensitive => + withCaseCatalog(warehouseDir, caseSensitive) { + failures ++= failingNativeScanChecks(Seq( + s"SELECT _file, id FROM $t ORDER BY id", + s"SELECT _pos, id FROM $t ORDER BY id", + s"SELECT _partition, id FROM $t ORDER BY id", + s"SELECT _partition.Region, id FROM $t ORDER BY id", + s"SELECT _spec_id, id FROM $t ORDER BY id", + s"SELECT id, Region, val, _file, _pos, _spec_id, _partition FROM $t ORDER BY id")) + .map(f => s"[caseSensitive=$caseSensitive]\n$f") + } + } + withCaseCatalog(warehouseDir, caseSensitive = true) { + spark.sql(s"DROP TABLE $t") + } + assert(failures.isEmpty, failures.mkString("\n\n=====\n\n")) + } + } + + test("issue 5540: nested struct field renamed by case only") { + assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => + val t = "case_cat.db.case_nested" + withCaseCatalog(warehouseDir, caseSensitive = true) { + spark.sql(s"CREATE TABLE $t (id INT, s STRUCT) USING iceberg") + spark.sql( + s"INSERT INTO $t VALUES (1, named_struct('x', 10, 'y', 100)), " + + "(2, named_struct('x', 20, 'y', 200))") + spark.sql(s"ALTER TABLE $t RENAME COLUMN s.x TO X") + } + withCaseCatalog(warehouseDir, caseSensitive = false) { + Seq( + s"SELECT id, s FROM $t ORDER BY id", + s"SELECT id, s.X FROM $t ORDER BY id", + s"SELECT id FROM $t WHERE s.X = 20").foreach { query => + checkNestedFieldEvolutionFallback(query, "s.X (renamed from x)") + } + spark.sql(s"DROP TABLE $t") + } + } + } + + test("issue 5540: nested field renamed under A beside a") { + assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => + val t = "case_cat.db.case_pair_nested" + withCaseCatalog(warehouseDir, caseSensitive = true) { + // After the rename `a` and `A` share the shape , since case-insensitive schema + // pruning requests `A` with the type of `a`. + spark.sql( + s"CREATE TABLE $t (a STRUCT, A STRUCT) USING iceberg") + spark.sql( + s"INSERT INTO $t VALUES " + + "(named_struct('x', 1, 'z', 2), named_struct('x', 10, 'y', 100)), " + + "(named_struct('x', 3, 'z', 4), " + + "named_struct('x', CAST(NULL AS INT), 'y', CAST(NULL AS INT)))") + spark.sql(s"ALTER TABLE $t RENAME COLUMN A.y TO z") + spark.sql( + s"INSERT INTO $t VALUES " + + "(named_struct('x', 5, 'z', 6), named_struct('x', 30, 'z', 300))") + } + // A case-insensitive lookup of `A` finds `a` first and would skip the check on `A.z`. + withCaseCatalog(warehouseDir, caseSensitive = false) { + val query = s"SELECT * FROM $t" + if (isSpark41Plus) { + val e = intercept[org.apache.spark.sql.AnalysisException](spark.sql(query).collect()) + assert(e.getMessage.contains("COLUMN_ALREADY_EXISTS"), e.getMessage) + } else { + checkNestedFieldEvolutionFallback(query, "A.z (renamed from y)") + } + } + withCaseCatalog(warehouseDir, caseSensitive = true) { + checkIcebergNativeScan(s"SELECT a FROM $t") + checkNestedFieldEvolutionFallback(s"SELECT A FROM $t", "A.z (renamed from y)") + spark.sql(s"DROP TABLE $t") + } + } + } + test("MOR table with multiple delete operations - mixed delete types") { assume(icebergAvailable, "Iceberg not available in classpath")