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

Fixes #14879.

### Description

When an Iceberg table evolves via `ALTER TABLE ... ADD COLUMN` with a
nested optional `MAP` or `LIST`, and a pre-evolution data file is read
on GPU, `GpuParquetReaderPostProcessor` used to throw
`IllegalArgumentException: Missing required field` before the missing
container's own `map()`/`list()` handler could emit `FillNull`. The CPU
reader handles this case correctly; the GPU read fails.

**User experience after this change:** queries against Iceberg tables
that have schema-evolved to add a nested `MAP`/`LIST` now succeed
end-to-end on GPU; pre-evolution rows materialize the new column as
`NULL`, matching CPU behavior. No new configs or user-facing knobs.

**Technical fix:** 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 at `GpuParquetReaderPostProcessor.scala:385` before the
container's own handler ran.

The fix extends `ActionBuildingVisitor`'s field stack to track each
field's partner type and adds a strict-ancestor
`isInsideMissingContainer` flag (excludes the current field itself, so
the missing container's own `struct`/`list`/`map` handler still runs the
normal `partner == null` logic). The new flag is treated 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).

**Tests added** (`GpuPostProcessorSuite`):
- `Missing nested optional map with required key does not throw` — the
reproducer; verifies both action-plan construction and `process()`
execution
- `Missing nested optional list with required element does not throw`
- `Doubly nested missing container with required descendant does not
throw` — exercises the strict-ancestor logic across multiple levels

All 15 `GpuPostProcessorSuite` tests pass under buildver=356.

### Checklists

Documentation
- [ ] Updated for new or modified user-facing features or behaviors
- [x] No user-facing change

Testing
- [x] Added or modified tests to cover new code paths
- [ ] Covered by existing tests
- [ ] Not required

Performance
- [ ] Tests ran and results are added in the PR description
- [ ] Issue filed with a link in the PR description
- [x] Not required

---------

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

Author: res-life

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,157 @@ 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 scoreFieldId = 2
+    val propsMapId = 3  // newly-added MAP, missing from file
+    val propsKeyId = 4
+    val propsValueId = 5
+
+    // Parquet file: info STRUCT<score: BIGINT> — no props
+    val scoreType =
+      ShadedTypes.primitive(ShadedPrimitiveTypeName.INT64, ShadedRepetition.OPTIONAL)
+      .id(scoreFieldId).named("score")
+    val infoStruct = ShadedTypes.optionalGroup().addField(scoreType)
+      .id(infoStructId).named("info")
+    val parquetSchema = new ShadedMessageType("test",
+      Seq[ShadedType](infoStruct).asJava)
+
+    // Expected: info STRUCT<score, props: MAP<STRING, BIGINT>>
+    val expectedSchema = new Schema(
+      Types.NestedField.optional(infoStructId, "info",
+        Types.StructType.of(
+          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 rowCount = 3
+    val (parquetInfo, shadedSchema) = createParquetInfo(parquetSchema, rowCount.toLong)
+    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")
+
+    // Exercise process() too: action-tree construction succeeding is not the
+    // same guarantee as execute() succeeding for FillNull(map<...>).
+    import com.nvidia.spark.rapids.{FuzzerUtils, GpuColumnVector, SpillableColumnarBatch}
+    import com.nvidia.spark.rapids.SpillPriorities
+    import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
+    import org.apache.spark.sql.types.{LongType, StructField, StructType => SparkStructType}
+
+    val inputSparkSchema = SparkStructType(Array(StructField(
+      "info",
+      SparkStructType(Seq(StructField("score", LongType, true))),
+      true)))
+    val inputBatch = FuzzerUtils.createColumnarBatch(inputSparkSchema, rowCount, seed = 42)
+    val spillable = closeOnExcept(inputBatch) { batch =>
+      SpillableColumnarBatch(batch, SpillPriorities.ACTIVE_ON_DECK_PRIORITY)
+    }
+    withResource(spillable) { _ =>
+      withResource(processor.process(spillable.getColumnarBatch())) { outputBatch =>
+        assert(outputBatch.numRows() == rowCount)
+        assert(outputBatch.numCols() == 1)
+        val infoCol = outputBatch.column(0).asInstanceOf[GpuColumnVector].getBase
+        // info struct gains props as a second child.
+        assert(infoCol.getNumChildren == 2,
+          s"expected info struct to have 2 children, got ${infoCol.getNumChildren}")
+        withResource(infoCol.getChildColumnView(1)) { propsCol =>
+          assert(propsCol.getNullCount == rowCount,
+            s"props should be all-null; nullCount=${propsCol.getNullCount} rowCount=$rowCount")
+        }
+      }
+    }
+  }
+
+  // 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")
+  }
+
 }