feat: bump iceberg-rust past apache/iceberg-rust#3255 and narrow the nested schema evolution fallback

docs/source/user-guide/latest/iceberg.md

@@ -217,10 +217,10 @@ The following scenarios will fall back to the JVM Iceberg reader:
 - Scans with residual filters using `truncate`, `bucket`, `year`, `month`, `day`, or `hour`
   transform functions (partition pruning still works, but row-level filtering of these
   transforms falls back)
-- Scans that read a struct, array, or map column with a nested field that schema evolution added
-  or renamed. The native reader cannot yet match such a field to data files written before the
-  change. The check uses the table's schema history, so the fallback stays after those files are
-  rewritten
+- Scans that read a struct column, or a struct inside an array or map, whose fields were all added
+  by schema evolution after some data file was written. For such a file the native reader returns a
+  struct of NULLs or NULL, depending on what else the scan reads, where Spark does not. The check
+  uses the table's schema history, so the fallback stays after those files are rewritten
 - Scans that read a struct, array, or map column from a table with
   `schema.name-mapping.default`, including tables imported with `snapshot`, `migrate`, or
   `add_files` ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The native reader
@@ -801,7 +801,8 @@ Which entries apply depends on which native paths are enabled:
 ### Reads that differ from Spark
 
 - **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such
-  a file as NULL, and the native scan returns the values stored in the file
+  a file as NULL, and the native scan returns the values stored in the file. A read of only some
+  of a struct's fields fails instead
   ([#6794](https://github.com/apache/datafusion-comet/issues/6794)).
 - **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as
   NULL, and two columns that swapped names return each other's values

native/Cargo.toml

@@ -85,8 +85,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false }
 # Pinned to a revision rather than a release because no published iceberg-rust
 # version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain
 # version requirement once a release ships on arrow 59. Bump policy: #5645.
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
 reqsign-core = "3"
 
 [profile.release]

native/core/src/execution/planner.rs

@@ -2739,7 +2739,11 @@ 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)
+                // The only consumer is the binding of an equality delete's predicate, whose
+                // column names iceberg-rust takes from the task schema. They match it exactly,
+                // and a case-insensitive binding rejects a schema with two names that differ
+                // only by 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/iceberg/IcebergReflection.scala

@@ -1283,6 +1283,24 @@ object IcebergReflection extends Logging {
     columns.asScala.flatMap(walkFieldForDefault).toList
   }
 
+  /**
+   * [[columnsWithInitialDefault]] for only the top-level columns of `schema` whose ids are in
+   * `columnIds`. A scan reads every nested field of such a column that its schema holds, so this
+   * is for the schema the scan reads when that is wider than the query's, as the table's is while
+   * nested schema pruning is off.
+   */
+  def columnsWithInitialDefault(schema: Any, columnIds: Set[Int]): List[String] = {
+    import scala.jdk.CollectionConverters._
+    val columns =
+      getMethod(schema.getClass, "columns").invoke(schema).asInstanceOf[java.util.List[_]]
+    columns.asScala
+      .filter(column => columnIds.contains(fieldIdOf(column)))
+      .flatMap {
+        walkFieldForDefault
+      }
+      .toList
+  }
+
   private def walkFieldForDefault(field: Any): List[String] = {
     import scala.jdk.CollectionConverters._
     val name = getMethod(field.getClass, "name").invoke(field).asInstanceOf[String]
@@ -1304,15 +1322,15 @@ object IcebergReflection extends Logging {
   }
 
   /**
-   * Returns the nested fields of the `fieldIds` columns of `schema` that some schema in the
-   * table's history (`table.schemas()`) lacks or names differently. Each is a dotted path with
-   * the change, for example `items.element.z (renamed from a)`. A data file written under that
-   * older schema lacks the field, or has it under the old name. Fields are matched by id at every
-   * level (struct fields, list elements, map keys and values), so a reorder or a type promotion
-   * is not a change, and an older schema that lacks the column itself is skipped. Throws on
+   * Returns the structs under the `fieldIds` columns of `schema`, at any depth and including the
+   * columns themselves, that share none of their fields with the same struct in some schema in
+   * the table's history (`table.schemas()`). Each is a dotted path, for example
+   * `items.element.s`. A data file written under that older schema has none of the struct's
+   * current fields. Structs are matched by field id at every level (struct fields, list elements,
+   * map keys and values), so an older schema that lacks the column itself is skipped. Throws on
    * reflection failure so the caller can fall back.
    */
-  def nestedFieldsAddedOrRenamed(table: Any, schema: Any, fieldIds: Set[Int]): Seq[String] = {
+  def structsWithNoFieldsInCommon(table: Any, schema: Any, fieldIds: Set[Int]): Seq[String] = {
     import scala.jdk.CollectionConverters._
     val findField = getMethod(schema.getClass, "findField", classOf[Int])
     def fieldById(s: Any, id: Int): Option[Any] =
@@ -1324,31 +1342,33 @@ object IcebergReflection extends Logging {
       .asScala
       .toSeq
     fieldIds.toSeq.flatMap { id =>
-      fieldById(schema, id).filter(childFields(_).nonEmpty).toSeq.flatMap { field =>
+      fieldById(schema, id).toSeq.flatMap { field =>
         history.flatMap(fieldById(_, id)).flatMap { older =>
-          childFieldsChangedFrom(field, older, fieldName(field))
+          structsWithNoFieldsInCommon(field, older, fieldName(field))
         }
       }
     }.distinct
   }
 
-  /** The nested fields under `field`, at any depth, that `older` lacks or names differently. */
-  private def childFieldsChangedFrom(field: Any, older: Any, path: String): Seq[String] = {
+  /** `field` if it is a struct sharing no field with `older`, plus the same for its children. */
+  private def structsWithNoFieldsInCommon(field: Any, older: Any, path: String): Seq[String] = {
+    val children = childFields(field)
     val olderChildren = childFields(older).map(child => fieldIdOf(child) -> child).toMap
-    childFields(field).flatMap { child =>
-      val name = fieldName(child)
-      val childPath = s"$path.$name"
-      olderChildren.get(fieldIdOf(child)) match {
-        case Some(olderChild) =>
-          val olderName = fieldName(olderChild)
-          val renamed =
-            if (olderName != name) Seq(s"$childPath (renamed from $olderName)") else Nil
-          renamed ++ childFieldsChangedFrom(child, olderChild, childPath)
-        case None => Seq(s"$childPath (added)")
+    val disjoint = isStruct(field) && children.nonEmpty &&
+      !children.exists(child => olderChildren.contains(fieldIdOf(child)))
+    val here = if (disjoint) Seq(path) else Nil
+    here ++ children.flatMap { child =>
+      olderChildren.get(fieldIdOf(child)).toSeq.flatMap { olderChild =>
+        structsWithNoFieldsInCommon(child, olderChild, s"$path.${fieldName(child)}")
       }
     }
   }
 
+  private def isStruct(field: Any): Boolean = {
+    val fieldType = getMethod(field.getClass, "type").invoke(field)
+    getMethod(fieldType.getClass, "isStructType").invoke(fieldType).asInstanceOf[Boolean]
+  }
+
   /** A struct's fields, a list's element, or a map's key and value. Empty for other types. */
   private def childFields(field: Any): Seq[Any] = {
     import scala.jdk.CollectionConverters._

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -743,17 +743,41 @@ case class CometScanRule(session: SparkSession)
             false
         }
 
+        // 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 resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet
+        val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty)
+
         // Column default values are a V3 feature, so V1/V2 tables cannot have them and need no
         // check. Only inspect V3 tables. iceberg-rust does not synthesize a V3 initial-default for
-        // a column absent from a data file, so a scan projecting such a column must fall back. A V3
-        // table implies an Iceberg version new enough to expose initialDefault(), so a reflection
-        // failure here is unexpected and also falls back rather than risk a crash.
+        // a column absent from a data file (it fails a nested one), so a scan reading such a
+        // column must fall back. That is the query's schema and, while nested schema pruning is
+        // off, the whole of each projected column in the task schema, which holds nested fields
+        // the query's does not. A V3 table implies an Iceberg version new enough to expose
+        // initialDefault(), so a reflection failure here is unexpected and also falls back rather
+        // than risk a crash.
         val defaultValuesSupported =
           if (!formatVersion.exists(_ >= 3)) {
             true
           } else {
             try {
-              val defaulted = IcebergReflection.columnsWithInitialDefault(metadata.scanSchema)
+              val readDefaults = metadata.taskSchema.toSeq.flatMap { schema =>
+                if (hasUnresolvedProjectedFieldIds) {
+                  IcebergReflection.columnsWithInitialDefault(schema)
+                } else {
+                  IcebergReflection.columnsWithInitialDefault(schema, resolvedProjectedFieldIds)
+                }
+              }
+              val defaulted =
+                (IcebergReflection.columnsWithInitialDefault(
+                  metadata.scanSchema) ++ readDefaults).distinct
               if (defaulted.nonEmpty) {
                 fallbackReasons += "Iceberg column(s) with V3 default values are not yet " +
                   s"supported by Comet's native reader: ${defaulted.mkString(", ")}"
@@ -769,18 +793,6 @@ 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 resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet
-        val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty)
-
         // iceberg-rust applies name mappings only at the root (#6790). Nested columns can
         // disappear during projection, or reach the adapter with physical field names that
         // cannot distinguish a mapped alias from a genuinely missing field. Schema history
@@ -831,16 +843,17 @@ case class CometScanRule(session: SparkSession)
               false
           }
 
-        // Neither iceberg-rust nor the batch adaptation after it matches a data file's nested
-        // fields to the table's by field id (apache/iceberg-rust#2617). A file written before a
-        // nested field was added (to a struct, or to a struct inside a list or map) fails the
-        // native scan with "Incorrect number of arrays for StructArray fields", and a nested field
-        // renamed since the file was written can read back as NULL. This check walks the schema
-        // the scan reads (see CometIcebergNativeScanMetadata.taskSchema), which for VERSION AS OF
-        // can be older than the table's, and also the table and scan schemas, so it can fall back
-        // when the changed field is pruned away. A FileScanTask does not record which schema wrote
-        // its file, so fall back when any schema in the table's history lacks a nested field of a
-        // projected column or names it differently.
+        // iceberg-rust matches a data file's nested fields to the table's by field id, filling in
+        // a field the file lacks with NULLs and applying renames, so schema evolution that adds,
+        // renames, drops or reorders nested fields reads correctly. One shape still differs from
+        // Spark: a struct (at any depth) whose fields in the schema the scan reads were all added
+        // after a file was written. The native read returns a struct of NULLs or NULL there,
+        // depending on what else the scan reads, where Spark returns NULL for a top-level struct
+        // and a struct of NULLs for a nested one. The schema the scan reads is the task schema (see
+        // CometIcebergNativeScanMetadata.taskSchema), which holds only the nested fields Spark
+        // keeps while nested schema pruning is on, and which for VERSION AS OF can be older than
+        // the table's. A FileScanTask does not record which schema wrote its file, so fall back
+        // when any schema in the table's history has such a struct in a projected column.
         val nestedFieldsSupported =
           try {
             val schemas = Seq(metadata.tableSchema, metadata.scanSchema) ++ metadata.taskSchema
@@ -850,18 +863,17 @@ case class CometScanRule(session: SparkSession)
               } else {
                 resolvedProjectedFieldIds
               }
-            val changed = schemas
-              .flatMap { schema =>
-                IcebergReflection.nestedFieldsAddedOrRenamed(metadata.table, schema, fieldIds)
-              }
+            val disjoint = metadata.taskSchema.toSeq
+              .flatMap(IcebergReflection.structsWithNoFieldsInCommon(metadata.table, _, fieldIds))
               .distinct
               .sorted
-            if (changed.nonEmpty) {
-              fallbackReasons += "Nested fields added or renamed by Iceberg schema evolution are " +
-                "not yet supported by Comet's native reader, which cannot match them to data " +
-                s"files written before the change: ${changed.mkString(", ")}"
+            if (disjoint.nonEmpty) {
+              fallbackReasons += "Structs whose fields were all added by Iceberg schema " +
+                "evolution are not yet supported by Comet's native reader, which reads data " +
+                "files written before the change differently from Spark: " +
+                disjoint.mkString(", ")
             }
-            changed.isEmpty
+            disjoint.isEmpty
           } catch {
             case e: Exception =>
               fallbackReasons += "Iceberg reflection failure: could not compare nested fields " +