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

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

@@ -204,10 +204,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 where no schema in the table's history gives a partition source or equality-delete key a
   name that the scan's other columns don't use (every task of a native scan must give such a
   column the same name)
@@ -779,14 +779,9 @@ Which entries apply depends on which native paths are enabled:
 
 ### Reads that differ from Spark
 
-- **Migrated tables that use a name mapping.** The `migrate`, `snapshot` and `add_files`
-  procedures set `schema.name-mapping.default` for files without Iceberg field ids. When a query
-  also reads a top-level primitive column, the native scan returns NULL for nested fields (struct
-  fields, list elements and map values). Filters on those fields then select the wrong rows
-  ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The fix needs an iceberg-rust
-  upgrade ([#6934](https://github.com/apache/datafusion-comet/issues/6934)).
 - **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]

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala

@@ -1304,15 +1304,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 +1324,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

@@ -789,16 +789,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
@@ -808,18 +809,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 " +