@@ -192,6 +192,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
 
 Writes are not covered by this list. By default Iceberg writes use Spark's own writer; see
 [Iceberg Writes](iceberg-writes.md) for the experimental native writer and when it applies.
 
@@ -1149,6 +1149,74 @@ object IcebergReflection extends Logging {
     here ++ nested
   }
 
+  /**
+   * 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
+   * reflection failure so the caller can fall back.
+   */
+  def nestedFieldsAddedOrRenamed(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] =
+      Option(findField.invoke(s, id.asInstanceOf[AnyRef]))
+    lazy val history = getMethod(table.getClass, "schemas")
+      .invoke(table)
+      .asInstanceOf[java.util.Map[_, _]]
+      .values()
+      .asScala
+      .toSeq
+    fieldIds.toSeq.flatMap { id =>
+      fieldById(schema, id).filter(childFields(_).nonEmpty).toSeq.flatMap { field =>
+        history.flatMap(fieldById(_, id)).flatMap { older =>
+          childFieldsChangedFrom(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] = {
+    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)")
+      }
+    }
+  }
+
+  /** 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._
+    val fieldType = getMethod(field.getClass, "type").invoke(field)
+    if (getMethod(fieldType.getClass, "isNestedType").invoke(fieldType).asInstanceOf[Boolean]) {
+      val nestedType = getMethod(fieldType.getClass, "asNestedType").invoke(fieldType)
+      getMethod(nestedType.getClass, "fields")
+        .invoke(nestedType)
+        .asInstanceOf[java.util.List[_]]
+        .asScala
+        .toSeq
+    } else {
+      Nil
+    }
+  }
+
+  private def fieldName(field: Any): String =
+    getMethod(field.getClass, "name").invoke(field).asInstanceOf[String]
+
+  private def fieldIdOf(field: Any): Int =
+    getMethod(field.getClass, "fieldId").invoke(field).asInstanceOf[Int]
+
   /**
    * Converts an Iceberg `Schema` to the Spark `StructType` it reads as, via
    * `SparkSchemaUtil.convert`. Comet serializes the whole table/scan schema to native (not just
 
@@ -727,16 +727,27 @@ 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)
+
         // The whole Iceberg table schema is serialized to native, but iceberg-rust can represent
-        // Variant in that schema as long as no projected field contains one. Match projected
-        // roots by field ID so historical snapshots still identify renamed columns, check them
-        // strictly, and allow Variant only under entirely unprojected roots. Other unsupported
-        // types still fail closed everywhere. An empty data projection is also strict because
-        // iceberg-rust currently interprets an empty field-id list as a request for every column.
+        // Variant in that schema as long as no projected field contains one. Check projected
+        // roots strictly, and allow Variant only under entirely unprojected roots. Other
+        // unsupported types still fail closed everywhere. An empty data projection is also strict
+        // because iceberg-rust currently interprets an empty field-id list as a request for every
+        // column.
         val schemaTypesSupported =
           try {
             val fullSchema = IcebergReflection.toSparkSchema(metadata.tableSchema)
-            val projectedDataColumns = scanExec.output.filterNot(_.isMetadataCol)
             // DataTypeSupport recursively dispatches back to this override for struct fields,
             // array elements, and map entries, so Variant is allowed at any nesting depth only
             // when its entire top-level Iceberg field is unprojected.
@@ -747,15 +758,7 @@ case class CometScanRule(session: SparkSession)
                   reasons: ListBuffer[String]): Boolean =
                 isVariantType(dt) || super.isTypeSupported(dt, name, reasons)
             }
-            val resolver = session.sessionState.conf.resolver
             val tableFieldIds = IcebergReflection.buildFieldIdMapping(metadata.tableSchema)
-            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)
 
             fullSchema.fields.forall { field =>
               val isProjected = projectedDataColumns.isEmpty ||
@@ -771,6 +774,44 @@ 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. The native read asks for a
+        // projected column's full nested type even when Spark prunes it, taken from the current
+        // table schema, or from the scan schema when VERSION AS OF reads a dropped column (see
+        // CometIcebergNativeScan). 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.
+        val nestedFieldsSupported =
+          try {
+            val schemas = Seq(metadata.tableSchema, metadata.scanSchema)
+            val fieldIds =
+              if (hasUnresolvedProjectedFieldIds) {
+                schemas.flatMap(IcebergReflection.buildFieldIdMapping(_).values).toSet
+              } else {
+                resolvedProjectedFieldIds
+              }
+            val changed = schemas
+              .flatMap { schema =>
+                IcebergReflection.nestedFieldsAddedOrRenamed(metadata.table, schema, 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(", ")}"
+            }
+            changed.isEmpty
+          } catch {
+            case e: Exception =>
+              fallbackReasons += "Iceberg reflection failure: could not compare nested fields " +
+                s"with the table's schema history: ${e.getMessage}"
+              false
+          }
+
         // The native Parquet reader decrypts 128-bit and 256-bit AES-GCM data keys (16- or
         // 32-byte). 192-bit is unsupported because the underlying crypto has no AES-192-GCM. Fall
         // back for anything else. None means the table is unencrypted. Reflection failure also
@@ -1006,7 +1047,8 @@ case class CometScanRule(session: SparkSession)
           defaultValuesSupported && schemaTypesSupported && encryptionKeyLengthSupported &&
           taskValidation.allParquet && allSupportedFilesystems && allLocationsOpenable &&
           metadataSchemeSupported && partitionTypesSupported && unifiedPartitionTypeSupported &&
-          transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported) {
+          transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported &&
+          nestedFieldsSupported) {
           CometBatchScanExec(
             scanExec.clone().asInstanceOf[BatchScanExec],
             runtimeFilters = scanExec.runtimeFilters,