From 32c8ea856e35ac4c522789c7c976ac7ff83dc54e Mon Sep 17 00:00:00 2001
From: Dustin Smith <dustin.william.smith@gmail.com>
Date: Tue, 20 Oct 2026 16:14:26 +0700
Subject: [PATCH 1/5] fix: match Iceberg column names exactly in the native
 scan

The native Iceberg scan folded column names in two places, though both
sides of each match already carry the table schema's names. The schema
adapter turned a table with columns `a` and `A` into a duplicate-field
error. FileScanTasks were built case-insensitive, and iceberg-rust uses
that flag only to bind equality-delete predicates, so `a` and `A`
resolved to one field id picked by HashMap order and deletes hit the
wrong column.

Both now match exactly. iceberg-rust relabels batches and equality-delete
columns to the task schema by field id, so an exact match is a field-id
match. A top-level column missing from the batch is now an error rather
than a column of nulls.
---
 docs/source/user-guide/latest/iceberg.md      |   3 -
 .../src/execution/operators/iceberg_scan.rs   | 121 +++++++--
 native/core/src/execution/planner.rs          |   5 +-
 .../comet/CometIcebergNativeSuite.scala       | 250 ++++++++++++++++++
 4 files changed, 359 insertions(+), 20 deletions(-)

diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index 7f251a2c2d8..71f716e3f92 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -823,9 +823,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..0d988a3f57d 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,39 @@ impl fmt::Debug for RedactedProperties<'_> {
     }
 }
 
+/// Adapter options for iceberg-rust batches. Both the batches and Spark's scan attributes carry
+/// the table schema's names, so matching is exact: folding case would only report columns that
+/// differ in case as duplicates.
+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.
+///
+/// iceberg-rust emits every projected top-level column, present even when the file lacks it, so
+/// a top-level target column absent from `file_schema` is an error rather than a null column.
 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 +676,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 +687,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 +719,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 +801,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 69e4738dde3..31a654a06b0 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -2669,7 +2669,10 @@ 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 equality-delete predicates are bound with this flag, and iceberg-rust has
+                // already renamed their columns to the task schema by field id, so an exact bind
+                // is a field-id bind. Folding 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/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index e859b23be08..8ef87ffb866 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -1272,6 +1272,256 @@ class CometIcebergNativeSuite
     }
   }
 
+  /**
+   * Runs each query through [[checkIcebergNativeScan]] and returns one message per query that
+   * failed, so every query is reported rather than only the first.
+   */
+  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"<plan failed: ${pe.getMessage}>" }
+          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)
+
+  test("issue 5540: case-sensitive read of columns a and A") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    withTempIcebergDir { warehouseDir =>
+      withCaseCatalog(warehouseDir, caseSensitive = true) {
+        val t = "case_cat.db.case_pair"
+        spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
+        spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
+        val failures = failingNativeScanChecks(
+          Seq(
+            s"SELECT a, A FROM $t ORDER BY a",
+            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 * FROM $t ORDER BY a"))
+        spark.sql(s"DROP TABLE $t")
+        assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+      }
+    }
+  }
+
+  test("issue 5540: case-sensitive filters on columns a and A") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    withTempIcebergDir { warehouseDir =>
+      withCaseCatalog(warehouseDir, caseSensitive = true) {
+        val t = "case_cat.db.case_pair_filter"
+        spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
+        spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
+        // a spans [1, 3] and A spans [20, 40]: a filter on one column bound to the other would
+        // prune the whole row group or keep the wrong row.
+        val failures = failingNativeScanChecks(
+          Seq(
+            s"SELECT a FROM $t WHERE A = 40",
+            s"SELECT A FROM $t WHERE a = 3",
+            s"SELECT A FROM $t WHERE A = 40",
+            s"SELECT a FROM $t WHERE a = 3",
+            s"SELECT a, A FROM $t WHERE A = 40",
+            s"SELECT a, A FROM $t WHERE a = 3",
+            s"SELECT a FROM $t WHERE A > 30",
+            s"SELECT a FROM $t WHERE A IN (20)",
+            s"SELECT A FROM $t WHERE a < 2"))
+        spark.sql(s"DROP TABLE $t")
+        assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
+      }
+    }
+  }
+
+  test("issue 5540: case-sensitive equality deletes keyed on a and on A") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    withTempIcebergDir { warehouseDir =>
+      withCaseCatalog(warehouseDir, caseSensitive = true) {
+        val t = "case_cat.db.case_pair_eq"
+        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)")
+        // One delete keyed on `a` and one keyed on `A`: binding either key by a folded name
+        // misreads one of them whichever field the fold resolves to. Only (5, 60) survives.
+        commitEqualityDelete("case_cat", "db", "case_pair_eq", "a", 1, warehouseDir)
+        commitEqualityDelete("case_cat", "db", "case_pair_eq", "A", 40, warehouseDir)
+        val 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"))
+        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: case-insensitive read of columns a and A with equality deletes") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    withTempIcebergDir { warehouseDir =>
+      val t = "case_cat.db.case_pair_ci"
+      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)")
+        commitEqualityDelete("case_cat", "db", "case_pair_ci", "a", 1, warehouseDir)
+        commitEqualityDelete("case_cat", "db", "case_pair_ci", "A", 40, warehouseDir)
+      }
+      // `a` is ambiguous here, so the queries name no column: only (5, 60) survives. Spark 4.1+
+      // rejects the duplicate columns itself before any scan runs.
+      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")
+      }
+    }
+  }
+
+  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, which computes the file's column metrics by field id through
+        // this mapping, so `WHERE id = 2` reads a file with stats.
+        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 {
+          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("parquet_case_mapped"),
+              icebergTable,
+              s"${warehouseDir.getAbsolutePath}/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: 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<x: INT, y: INT>) 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")
+      }
+      // A nested rename falls back, so the native reader never sees `s.x` under the name `s.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("MOR table with multiple delete operations - mixed delete types") {
     assume(icebergAvailable, "Iceberg not available in classpath")
 

From 60b4745fd2d3b0b2df18a6a579a55144e94dcefa Mon Sep 17 00:00:00 2001
From: Dustin Smith <dustin.william.smith@gmail.com>
Date: Tue, 20 Oct 2026 18:28:58 +0700
Subject: [PATCH 2/5] test: cover nested name-mapped children and metadata
 columns under exact Iceberg matching

A name-mapped file whose nested children differ in case from the table
reads the same as before: iceberg-rust hands the adapter a batch with the
table's nested names. Metadata columns resolve by reserved field id, so
the exact top-level check leaves them alone under either case setting.
---
 .../comet/CometIcebergNativeSuite.scala       | 118 ++++++++++++++++++
 1 file changed, 118 insertions(+)

diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index 8ef87ffb866..dc816c9e476 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -1498,6 +1498,124 @@ class CometIcebergNativeSuite
     }
   }
 
+  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]
+    // (struct, x, y) names as written in the file, mapped by field id onto `s STRUCT<x, y>`. The
+    // first shape matches the table's names 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<x: INT, y: INT>) 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"]}]}]"""
+          // Set before the import, which computes the file's column metrics through the mapping.
+          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 {
+            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(s"parquet_$name"),
+                icebergTable,
+                s"${warehouseDir.getAbsolutePath}/staging")
+            // Spark itself reads the mapped children, so matching Spark means reading 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"
+                  // iceberg-rust applies a name mapping to top-level fields only, so two reads
+                  // return wrong values whatever the case setting: any read that also projects a
+                  // top-level column (`s` comes back NULL, #6790), and a pruned `s.y` over
+                  // physical `X`, `Y` (it reads `X`, #7134). Neither is checked here.
+                  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 rather than through the task schema, so
+      // the exact top-level name check must leave them alone.
+      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 =>

From b8518b8deae498152a1df204dabf158b7dd620c3 Mon Sep 17 00:00:00 2001
From: Dustin Smith <dustin.william.smith@gmail.com>
Date: Wed, 21 Oct 2026 10:34:36 +0700
Subject: [PATCH 3/5] test: move Iceberg case-sensitive read checks into a SQL
 fixture

Read and filter checks for case-variant columns now live in
sql-tests/iceberg/case_sensitive_columns.sql. The suite keeps the
equality-delete coverage in one test for both key columns and shares
a helper for importing Parquet files into a table.
---
 .../iceberg/case_sensitive_columns.sql        |  91 +++++++++++
 .../comet/CometIcebergNativeSuite.scala       | 148 +++++-------------
 2 files changed, 132 insertions(+), 107 deletions(-)
 create mode 100644 spark/src/test/resources/sql-tests/iceberg/case_sensitive_columns.sql

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..ea5a51e2849
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/iceberg/case_sensitive_columns.sql
@@ -0,0 +1,91 @@
+-- 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.
+
+-- Columns `a` and `A` in one Iceberg table under spark.sql.caseSensitive=true. The native scan
+-- has to bind each projected column and each filter to its own field rather than match names
+-- case-insensitively (https://github.com/apache/datafusion-comet/issues/5540).
+-- CometIcebergNativeSuite covers equality deletes, name mapping, and metadata columns.
+
+-- 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
+
+-- `a` spans [1, 3] and `A` spans [20, 40], so a filter bound to the other column prunes the
+-- whole row group or keeps the wrong row.
+statement
+INSERT INTO test_cat.db.case_pair VALUES (1, 20), (3, 40)
+
+-- A plain query fails on any operator that is not Comet's, so each one also checks that the
+-- Iceberg scan ran natively.
+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 dc816c9e476..b7faa7387df 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -1300,57 +1300,32 @@ class CometIcebergNativeSuite
       CometConf.COMET_EXEC_ENABLED.key -> "true",
       CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")(f)
 
-  test("issue 5540: case-sensitive read of columns a and A") {
-    assume(icebergAvailable, "Iceberg not available in classpath")
-    withTempIcebergDir { warehouseDir =>
-      withCaseCatalog(warehouseDir, caseSensitive = true) {
-        val t = "case_cat.db.case_pair"
-        spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
-        spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
-        val failures = failingNativeScanChecks(
-          Seq(
-            s"SELECT a, A FROM $t ORDER BY a",
-            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 * FROM $t ORDER BY a"))
-        spark.sql(s"DROP TABLE $t")
-        assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
-      }
-    }
-  }
-
-  test("issue 5540: case-sensitive filters on columns a and A") {
-    assume(icebergAvailable, "Iceberg not available in classpath")
-    withTempIcebergDir { warehouseDir =>
-      withCaseCatalog(warehouseDir, caseSensitive = true) {
-        val t = "case_cat.db.case_pair_filter"
-        spark.sql(s"CREATE TABLE $t (a INT, A INT) USING iceberg")
-        spark.sql(s"INSERT INTO $t VALUES (1, 20), (3, 40)")
-        // a spans [1, 3] and A spans [20, 40]: a filter on one column bound to the other would
-        // prune the whole row group or keep the wrong row.
-        val failures = failingNativeScanChecks(
-          Seq(
-            s"SELECT a FROM $t WHERE A = 40",
-            s"SELECT A FROM $t WHERE a = 3",
-            s"SELECT A FROM $t WHERE A = 40",
-            s"SELECT a FROM $t WHERE a = 3",
-            s"SELECT a, A FROM $t WHERE A = 40",
-            s"SELECT a, A FROM $t WHERE a = 3",
-            s"SELECT a FROM $t WHERE A > 30",
-            s"SELECT a FROM $t WHERE A IN (20)",
-            s"SELECT A FROM $t WHERE a < 2"))
-        spark.sql(s"DROP TABLE $t")
-        assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
-      }
-    }
-  }
-
-  test("issue 5540: case-sensitive equality deletes keyed on a and on A") {
+  /** Imports the files of session table `sourceTable` into `table` with SparkTableUtil. */
+  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) {
-        val t = "case_cat.db.case_pair_eq"
         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)")
@@ -1358,15 +1333,29 @@ class CometIcebergNativeSuite
         // misreads one of them whichever field the fold resolves to. Only (5, 60) survives.
         commitEqualityDelete("case_cat", "db", "case_pair_eq", "a", 1, warehouseDir)
         commitEqualityDelete("case_cat", "db", "case_pair_eq", "A", 40, warehouseDir)
-        val failures = failingNativeScanChecks(
+        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 when case-insensitive, so these queries name no column. Spark 4.1+
+      // rejects the duplicate columns itself before any scan runs.
+      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"))
       }
+      assert(failures.isEmpty, failures.mkString("\n\n=====\n\n"))
     }
   }
 
@@ -1391,35 +1380,6 @@ class CometIcebergNativeSuite
     }
   }
 
-  test("issue 5540: case-insensitive read of columns a and A with equality deletes") {
-    assume(icebergAvailable, "Iceberg not available in classpath")
-    withTempIcebergDir { warehouseDir =>
-      val t = "case_cat.db.case_pair_ci"
-      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)")
-        commitEqualityDelete("case_cat", "db", "case_pair_ci", "a", 1, warehouseDir)
-        commitEqualityDelete("case_cat", "db", "case_pair_ci", "A", 40, warehouseDir)
-      }
-      // `a` is ambiguous here, so the queries name no column: only (5, 60) survives. Spark 4.1+
-      // rejects the duplicate columns itself before any scan runs.
-      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")
-      }
-    }
-  }
-
   test("issue 5540: equality delete survives a case-only column rename") {
     assume(icebergAvailable, "Iceberg not available in classpath")
     withTempIcebergDir { warehouseDir =>
@@ -1470,20 +1430,7 @@ class CometIcebergNativeSuite
           .table()
         spark.sql(s"CREATE TABLE parquet_case_mapped USING parquet LOCATION '$dataPath'")
         try {
-          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("parquet_case_mapped"),
-              icebergTable,
-              s"${warehouseDir.getAbsolutePath}/staging")
+          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")
@@ -1530,20 +1477,7 @@ class CometIcebergNativeSuite
           icebergTable.refresh()
           spark.sql(s"CREATE TABLE parquet_$name USING parquet LOCATION '$dataPath'")
           try {
-            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(s"parquet_$name"),
-                icebergTable,
-                s"${warehouseDir.getAbsolutePath}/staging")
+            importParquetInto(icebergTable, s"parquet_$name", s"$warehouseDir/staging")
             // Spark itself reads the mapped children, so matching Spark means reading values.
             withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
               assert(

From 5ac663aed43988437508ad4eac435cfdc85fe8f0 Mon Sep 17 00:00:00 2001
From: Dustin Smith <dustin.william.smith@gmail.com>
Date: Thu, 22 Oct 2026 10:31:41 +0700
Subject: [PATCH 4/5] docs: keep only the comments that explain exact Iceberg
 column matching

Drop comments that restate the code or a test's name, and shorten the
rest to why names match exactly and what the tests guard.
---
 .../src/execution/operators/iceberg_scan.rs   |  7 ++---
 native/core/src/execution/planner.rs          |  5 ++-
 .../iceberg/case_sensitive_columns.sql        | 10 ++----
 .../comet/CometIcebergNativeSuite.scala       | 31 ++++++-------------
 4 files changed, 15 insertions(+), 38 deletions(-)

diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs
index 0d988a3f57d..640c7896a98 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -576,9 +576,7 @@ impl fmt::Debug for RedactedProperties<'_> {
     }
 }
 
-/// Adapter options for iceberg-rust batches. Both the batches and Spark's scan attributes carry
-/// the table schema's names, so matching is exact: folding case would only report columns that
-/// differ in case as duplicates.
+/// 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;
@@ -590,8 +588,7 @@ fn iceberg_parquet_options() -> SparkParquetOptions {
 /// The returned expressions can be cached and reused across multiple batches
 /// that share the same file schema, avoiding repeated expression construction.
 ///
-/// iceberg-rust emits every projected top-level column, present even when the file lacks it, so
-/// a top-level target column absent from `file_schema` is an error rather than a null column.
+/// 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,
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 1d33b60f567..954b0bd0b41 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -2670,9 +2670,8 @@ 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)
-                // Only equality-delete predicates are bound with this flag, and iceberg-rust has
-                // already renamed their columns to the task schema by field id, so an exact bind
-                // is a field-id bind. Folding would merge columns that differ only in case.
+                // 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.
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
index ea5a51e2849..8b06567796d 100644
--- a/spark/src/test/resources/sql-tests/iceberg/case_sensitive_columns.sql
+++ b/spark/src/test/resources/sql-tests/iceberg/case_sensitive_columns.sql
@@ -15,10 +15,7 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- Columns `a` and `A` in one Iceberg table under spark.sql.caseSensitive=true. The native scan
--- has to bind each projected column and each filter to its own field rather than match names
--- case-insensitively (https://github.com/apache/datafusion-comet/issues/5540).
--- CometIcebergNativeSuite covers equality deletes, name mapping, and metadata columns.
+-- `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.
@@ -38,13 +35,10 @@ DROP TABLE IF EXISTS test_cat.db.case_pair
 statement
 CREATE TABLE test_cat.db.case_pair (a INT, A INT) USING iceberg
 
--- `a` spans [1, 3] and `A` spans [20, 40], so a filter bound to the other column prunes the
--- whole row group or keeps the wrong row.
+-- 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)
 
--- A plain query fails on any operator that is not Comet's, so each one also checks that the
--- Iceberg scan ran natively.
 query
 SELECT a, A FROM test_cat.db.case_pair ORDER BY a
 
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index b7faa7387df..bcaaca0d96e 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -1272,10 +1272,7 @@ class CometIcebergNativeSuite
     }
   }
 
-  /**
-   * Runs each query through [[checkIcebergNativeScan]] and returns one message per query that
-   * failed, so every query is reported rather than only the first.
-   */
+  /** Returns one message per query failing [[checkIcebergNativeScan]], so all are reported. */
   private def failingNativeScanChecks(queries: Seq[String]): Seq[String] =
     queries.flatMap { q =>
       try {
@@ -1300,7 +1297,6 @@ class CometIcebergNativeSuite
       CometConf.COMET_EXEC_ENABLED.key -> "true",
       CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true")(f)
 
-  /** Imports the files of session table `sourceTable` into `table` with SparkTableUtil. */
   private def importParquetInto(
       table: org.apache.iceberg.Table,
       sourceTable: String,
@@ -1329,8 +1325,7 @@ class CometIcebergNativeSuite
         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)")
-        // One delete keyed on `a` and one keyed on `A`: binding either key by a folded name
-        // misreads one of them whichever field the fold resolves to. Only (5, 60) survives.
+        // 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(
@@ -1340,8 +1335,7 @@ class CometIcebergNativeSuite
             s"SELECT a FROM $t ORDER BY a",
             s"SELECT count(*) FROM $t"))
       }
-      // `a` is ambiguous when case-insensitive, so these queries name no column. Spark 4.1+
-      // rejects the duplicate columns itself before any scan runs.
+      // `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) {
@@ -1416,8 +1410,7 @@ class CometIcebergNativeSuite
           .write
           .parquet(dataPath)
         spark.sql(s"CREATE TABLE $t (id INT, val INT) USING iceberg")
-        // Set before the import, which computes the file's column metrics by field id through
-        // this mapping, so `WHERE id = 2` reads a file with stats.
+        // 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"]}]')""")
@@ -1450,8 +1443,7 @@ class CometIcebergNativeSuite
     val name = "case_mapped_nested"
     val t = s"case_cat.db.$name"
     var failures = Seq.empty[String]
-    // (struct, x, y) names as written in the file, mapped by field id onto `s STRUCT<x, y>`. The
-    // first shape matches the table's names exactly, as a control.
+    // 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"
@@ -1471,14 +1463,13 @@ class CometIcebergNativeSuite
               s"""{"field-id":${fieldId("s")},"names":["$sName"],"fields":[""" +
               s"""{"field-id":${fieldId("s.x")},"names":["$xName"]},""" +
               s"""{"field-id":${fieldId("s.y")},"names":["$yName"]}]}]"""
-          // Set before the import, which computes the file's column metrics through the mapping.
           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 itself reads the mapped children, so matching Spark means reading values.
+            // 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 ==
@@ -1493,10 +1484,8 @@ class CometIcebergNativeSuite
                   CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> pruning.toString) {
                   val label = s"file $sName<$xName, $yName>, caseSensitive=$caseSensitive, " +
                     s"nestedPruning=$pruning"
-                  // iceberg-rust applies a name mapping to top-level fields only, so two reads
-                  // return wrong values whatever the case setting: any read that also projects a
-                  // top-level column (`s` comes back NULL, #6790), and a pruned `s.y` over
-                  // physical `X`, `Y` (it reads `X`, #7134). Neither is checked here.
+                  // 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",
@@ -1529,8 +1518,7 @@ class CometIcebergNativeSuite
             "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 rather than through the task schema, so
-      // the exact top-level name check must leave them alone.
+      // Metadata columns resolve by reserved field id, so the exact name check must skip them.
       Seq(true, false).foreach { caseSensitive =>
         withCaseCatalog(warehouseDir, caseSensitive) {
           failures ++= failingNativeScanChecks(Seq(
@@ -1561,7 +1549,6 @@ class CometIcebergNativeSuite
             "(2, named_struct('x', 20, 'y', 200))")
         spark.sql(s"ALTER TABLE $t RENAME COLUMN s.x TO X")
       }
-      // A nested rename falls back, so the native reader never sees `s.x` under the name `s.X`.
       withCaseCatalog(warehouseDir, caseSensitive = false) {
         Seq(
           s"SELECT id, s FROM $t ORDER BY id",

From 74979463a9dff1607d780386eb112195fd9cd3c9 Mon Sep 17 00:00:00 2001
From: Dustin Smith <dustin.william.smith@gmail.com>
Date: Fri, 23 Oct 2026 09:49:58 +0700
Subject: [PATCH 5/5] fix: look up projected Iceberg field ids by exact name in
 CometScanRule

With spark.sql.caseSensitive=false the rule matched projected columns to
field ids through the session resolver, so in a table with columns a and
A both resolved to a's id. The nested-field check then skipped A, and on
Spark 3.5 and 4.0 a field renamed under A read back as NULL from files
written before the rename. Look ids up by exact name, as the serde does.
---
 .../apache/comet/rules/CometScanRule.scala    |  8 +---
 .../comet/CometIcebergNativeSuite.scala       | 40 ++++++++++++++++++-
 2 files changed, 41 insertions(+), 7 deletions(-)

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/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index bcaaca0d96e..bc0ef4bc10c 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -1518,7 +1518,8 @@ class CometIcebergNativeSuite
             "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, so the exact name check must skip them.
+      // 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(
@@ -1561,6 +1562,43 @@ class CometIcebergNativeSuite
     }
   }
 
+  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 <x, z>, since case-insensitive schema
+        // pruning requests `A` with the type of `a`.
+        spark.sql(
+          s"CREATE TABLE $t (a STRUCT<x: INT, z: INT>, A STRUCT<x: INT, y: INT>) 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")