[BUG FIX] Iceberg: fix Missing required field for newly-added nested MAP/LIST

When an Iceberg table evolves via ALTER TABLE ... ADD COLUMN with a nested
optional MAP/LIST and a pre-evolution data file is read, the GPU post-
processor used to throw `IllegalArgumentException: Missing required field`
before the missing container's own handler could emit FillNull. Iceberg's
SchemaWithPartnerVisitor walks post-order, so primitive children of a missing
container are visited first; a required descendant (map key is intrinsically
required, list elements may be required) was dispatched to
MissingFieldActionBuilder.buildAction with isOptional=false and threw.

Track each field's partner type on the ActionBuildingVisitor's field stack
and expose a strict-ancestor `isInsideMissingContainer` flag (excludes the
current field itself, so the missing container's own struct/list/map handler
still runs the regular partner==null logic). Treat the new flag exactly like
the existing `isInsideConstantStruct` short-circuit in the four
partner==null branches: descendants of a missing container fall through to
FillNull (which the parent's own FillNull discards anyway).

Adds three regression tests covering:
- nested optional MAP<STRING, BIGINT> missing from file (the reproducer)
- nested optional LIST<required element> missing from file
- doubly-nested missing struct containing a missing map (exercises the
  strict-ancestor logic across multiple levels)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Signed-off-by: Chong Gao <res_life@163.com>

Author: Chong Gao

iceberg/common/src/main/scala/com/nvidia/spark/rapids/iceberg/parquet/GpuParquetReaderPostProcessor.scala

@@ -396,12 +396,23 @@ private class ActionBuildingVisitor(
     idToConstant: JMap[Integer, _]
 ) extends SchemaWithPartnerVisitor[Type, ColumnAction] {
 
-  // Track the current field and whether we are inside a constant struct.
-  private val fieldStack = Stack.empty[(Types.NestedField, Boolean)]
+  // Track the current field, its partner type in the file schema (null when the
+  // field is missing from the file), and whether we are inside a constant struct.
+  // The partner is stored so that strict-ancestor "inside a missing container"
+  // can be detected by descendants — see isInsideMissingContainer below.
+  private val fieldStack = Stack.empty[(Types.NestedField, Type, Boolean)]
   private def currentField: Types.NestedField =
     fieldStack.headOption.map(_._1).orNull
   private def isInsideConstantStruct: Boolean =
-    fieldStack.headOption.exists(_._2)
+    fieldStack.headOption.exists(_._3)
+  // True iff a STRICT ancestor of the current field is missing from the file
+  // schema (partner == null at an ancestor level). The current field's own
+  // missing-ness is reflected by the visitor method's `partner` parameter and
+  // is intentionally excluded here, so the missing container's own handler
+  // still runs the normal partner==null logic (constant lookup, fill-null for
+  // optional, or throw for required).
+  private def isInsideMissingContainer: Boolean =
+    fieldStack.size > 1 && fieldStack.tail.exists(_._2 == null)
 
   override def schema(
       schema: Schema,
@@ -424,7 +435,11 @@ private class ActionBuildingVisitor(
     }
 
     if (partner == null) {
-      if (isInsideConstantStruct) {
+      // Inside a missing ancestor: the parent's own handler will emit
+      // FillNull for the whole container and discard this result, so any
+      // safe placeholder works — FillNull mirrors the isInsideConstantStruct
+      // path and never requires an input column at execute time.
+      if (isInsideConstantStruct || isInsideMissingContainer) {
         return FillNull(sparkType)
       }
       return MissingFieldActionBuilder.buildAction(
@@ -463,7 +478,7 @@ private class ActionBuildingVisitor(
   }
 
   override def beforeField(field: Types.NestedField, partner: Type): Unit = {
-    fieldStack.push((field,
+    fieldStack.push((field, partner,
       isInsideConstantStruct ||
         (field.`type`().isStructType &&
           idToConstant.containsKey(field.fieldId()))))
@@ -484,7 +499,7 @@ private class ActionBuildingVisitor(
       elementResult: ColumnAction): ColumnAction = {
     if (partner == null) {
       val sparkType = SparkSchemaUtil.convert(list)
-      if (isInsideConstantStruct) {
+      if (isInsideConstantStruct || isInsideMissingContainer) {
         return FillNull(sparkType)
       }
       return MissingFieldActionBuilder.buildAction(
@@ -493,7 +508,7 @@ private class ActionBuildingVisitor(
         currentField.isOptional,
         idToConstant)
     }
-    
+
     if (elementResult == PassThrough) {
       PassThrough
     } else {
@@ -508,7 +523,7 @@ private class ActionBuildingVisitor(
       valueResult: ColumnAction): ColumnAction = {
     if (partner == null) {
       val sparkType = SparkSchemaUtil.convert(map)
-      if (isInsideConstantStruct) {
+      if (isInsideConstantStruct || isInsideMissingContainer) {
         return FillNull(sparkType)
       }
       return MissingFieldActionBuilder.buildAction(
@@ -517,7 +532,7 @@ private class ActionBuildingVisitor(
         currentField.isOptional,
         idToConstant)
     }
-    
+
     if (keyResult == PassThrough && valueResult == PassThrough) {
       PassThrough
     } else {
@@ -538,7 +553,8 @@ private class ActionBuildingVisitor(
       } else {
         UpCast(fileType, expectedType)
       }
-    } else if (isInsideConstantStruct) {
+    } else if (isInsideConstantStruct || isInsideMissingContainer) {
+      // Children of a missing container — see struct() for rationale.
       FillNull(expectedType)
     } else {
       MissingFieldActionBuilder.buildAction(

tests/src/test/spark350/scala/com/nvidia/spark/rapids/iceberg/GpuPostProcessorSuite.scala

@@ -935,4 +935,131 @@ class GpuPostProcessorSuite extends AnyFunSuite with BeforeAndAfterAll {
       s"FetchConstant(fieldId=$structFieldId, struct<"))
   }
 
+  // Regression: ALTER TABLE ADD COLUMN of an optional nested MAP must not
+  // throw when reading old data files that pre-date the ADD. Iceberg map
+  // keys are intrinsically required, so the post-order visitor used to call
+  // MissingFieldActionBuilder for the key with isOptional=false and throw
+  // "Missing required field" before the missing-container's own map() handler
+  // could emit FillNull.
+  test("Missing nested optional map with required key does not throw") {
+    val infoStructId = 1
+    val nameFieldId = 2
+    val scoreFieldId = 3
+    val propsMapId = 4  // newly-added MAP, missing from file
+    val propsKeyId = 5
+    val propsValueId = 6
+
+    // Parquet file: info STRUCT<name: STRING, score: BIGINT> — no props
+    val nameType =
+      ShadedTypes.primitive(ShadedPrimitiveTypeName.BINARY, ShadedRepetition.OPTIONAL)
+      .id(nameFieldId).named("name")
+    val scoreType =
+      ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.OPTIONAL)
+      .id(scoreFieldId).named("score")
+    val infoStruct = ShadedTypes.optionalGroup().addField(nameType).addField(scoreType)
+      .id(infoStructId).named("info")
+    val parquetSchema = new ShadedMessageType("test",
+      Seq[ShadedType](infoStruct).asJava)
+
+    // Expected: info STRUCT<name, score, props: MAP<STRING, BIGINT>>
+    val expectedSchema = new Schema(
+      Types.NestedField.optional(infoStructId, "info",
+        Types.StructType.of(
+          Types.NestedField.optional(nameFieldId, "name", Types.StringType.get()),
+          Types.NestedField.optional(scoreFieldId, "score", Types.LongType.get()),
+          Types.NestedField.optional(propsMapId, "props",
+            Types.MapType.ofOptional(propsKeyId, propsValueId,
+              Types.StringType.get(), Types.LongType.get()))
+        ))
+    )
+
+    val (parquetInfo, shadedSchema) = createParquetInfo(parquetSchema)
+    val processor = new GpuParquetReaderPostProcessor(
+      parquetInfo,
+      new JHashMap[Integer, Any](),
+      expectedSchema,
+      shadedSchema,
+      Map.empty)
+
+    val plan = processor.displayActionPlan()
+    // The missing map itself must emit FillNull (it's the only valid behavior
+    // for an optional missing field). Children are discarded by the parent.
+    assert(plan.contains("FillNull(map<"),
+      s"expected FillNull for missing map in plan:\n$plan")
+  }
+
+  // Same defect, with LIST<required element> as the missing container.
+  test("Missing nested optional list with required element does not throw") {
+    val outerStructId = 1
+    val keepFieldId = 2
+    val newListId = 3   // newly-added list, missing from file
+    val newListElementId = 4
+
+    val keepType =
+      ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.OPTIONAL)
+      .id(keepFieldId).named("keep")
+    val outerStruct = ShadedTypes.optionalGroup().addField(keepType)
+      .id(outerStructId).named("outer")
+    val parquetSchema = new ShadedMessageType("test",
+      Seq[ShadedType](outerStruct).asJava)
+
+    val expectedSchema = new Schema(
+      Types.NestedField.optional(outerStructId, "outer",
+        Types.StructType.of(
+          Types.NestedField.optional(keepFieldId, "keep", Types.LongType.get()),
+          // List with a REQUIRED element — the bug also fires here because the
+          // required element primitive is visited before the parent list().
+          Types.NestedField.optional(newListId, "new_list",
+            Types.ListType.ofRequired(newListElementId, Types.LongType.get()))
+        ))
+    )
+
+    val (parquetInfo, shadedSchema) = createParquetInfo(parquetSchema)
+    val processor = new GpuParquetReaderPostProcessor(
+      parquetInfo,
+      new JHashMap[Integer, Any](),
+      expectedSchema,
+      shadedSchema,
+      Map.empty)
+
+    val plan = processor.displayActionPlan()
+    assert(plan.contains("FillNull(array<"),
+      s"expected FillNull for missing list in plan:\n$plan")
+  }
+
+  // Doubly-nested: a missing struct that itself contains a missing map with
+  // a required key. Exercises strict-ancestor isInsideMissingContainer when
+  // multiple levels above also have partner == null.
+  test("Doubly nested missing container with required descendant does not throw") {
+    val metaStructId = 1   // missing struct
+    val propsMapId = 2     // missing map (inside metaStruct)
+    val propsKeyId = 3
+    val propsValueId = 4
+
+    val parquetSchema = new ShadedMessageType("test",
+      Seq.empty[ShadedType].asJava)
+    val expectedSchema = new Schema(
+      Types.NestedField.optional(metaStructId, "meta",
+        Types.StructType.of(
+          Types.NestedField.optional(propsMapId, "props",
+            Types.MapType.ofOptional(propsKeyId, propsValueId,
+              Types.StringType.get(), Types.LongType.get()))
+        ))
+    )
+
+    val (parquetInfo, shadedSchema) = createParquetInfo(parquetSchema)
+    val processor = new GpuParquetReaderPostProcessor(
+      parquetInfo,
+      new JHashMap[Integer, Any](),
+      expectedSchema,
+      shadedSchema,
+      Map.empty)
+
+    val plan = processor.displayActionPlan()
+    // Outer struct is missing; its child map is also missing — only the
+    // outer FillNull(struct<...>) is composed into the plan.
+    assert(plan.contains("FillNull(struct<"),
+      s"expected FillNull for outer missing struct in plan:\n$plan")
+  }
+
 }