fix: match Iceberg column names exactly in the native scan

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)).

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<dyn PhysicalExprAdapter>,
 ) -> DFResult<Vec<Arc<dyn PhysicalExpr>>> {
     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<dyn PhysicalExpr> = 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::<Vec<_>>(),
+            ));
+            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<i64> = output
+                .columns()
+                .iter()
+                .map(|c| c.as_any().downcast_ref::<Int64Array>().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<FileScanTaskDeleteFile>) -> 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()
     }

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))

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)
 

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