Spark: fix NPE thrown for MAP/LIST columns on DELETE, UPDATE, and MERGE operations

[antonlin1]

Partitioned tables with MAP or LIST columns throws NPE on DELETE, UPDATE, and MERGE operations when running Spark 4.1 + Iceberg 1.11 (reproduced with iceberg-spark-runtime-4.1_2.13:1.11.0-20260306.003105-70):

Cannot invoke "org.apache.iceberg.types.Types$NestedField.type()" because "expected" is null
	at org.apache.iceberg.parquet.PruneColumns.isStruct(PruneColumns.java:173)
	at org.apache.iceberg.parquet.PruneColumns.message(PruneColumns.java:61)
	at org.apache.iceberg.parquet.PruneColumns.message(PruneColumns.java:38)
	at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visit(TypeWithSchemaVisitor.java:48)
	at org.apache.iceberg.parquet.ParquetSchemaUtil.pruneColumns(ParquetSchemaUtil.java:134)
	at org.apache.iceberg.parquet.ReadConf.<init>(ReadConf.java:82)
	at org.apache.iceberg.parquet.VectorizedParquetReader.init(VectorizedParquetReader.java:90)
	at org.apache.iceberg.parquet.VectorizedParquetReader.iterator(VectorizedParquetReader.java:99)
	at org.apache.iceberg.io.CloseableIterable$7$1.<init>(CloseableIterable.java:205)
	at org.apache.iceberg.io.CloseableIterable$7.iterator(CloseableIterable.java:204)
	at org.apache.iceberg.spark.source.BatchDataReader.open(BatchDataReader.java:120)
	at org.apache.iceberg.spark.source.BatchDataReader.open(BatchDataReader.java:42)
	at org.apache.iceberg.spark.source.BaseReader.next(BaseReader.java:140)
	at org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:148)
	at org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:186)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:72)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:72)
	at scala.Option.exists(Option.scala:406)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:72)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:103)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:72)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$$anon$9.hasNext(Iterator.scala:593)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:50)
	at scala.collection.Iterator$$anon$9.hasNext(Iterator.scala:593)
	at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:195)
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:57)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:111)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:180)
	at org.apache.spark.scheduler.Task.run(Task.scala:147)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$5(Executor.scala:716)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:86)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:83)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:97)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:719)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)

allUsedFieldIds() (introduced in #15297) uses TypeUtil.getProjectedIds() to build the set of field IDs already in use. This omits MAP and LIST container IDs, creating gaps in the ID space. A _partition child field gets reassigned into one of these gaps, landing on the same ID as a MAP or LIST column.

During the Parquet scan, PruneColumns finds this ID in selectedIds (via _partition) and looks it up in the expected Iceberg schema — but the ID belongs to a nested _partition child, not a top-level data column, so expected.field(id) returns null. isStruct() then NPEs on that null.

This regression did not exist in 1.10.x where the equivalent code used TypeUtil.indexById() which correctly indexes all field IDs including MAP/LIST containers.

Fix

Replace TypeUtil.getProjectedIds() with TypeUtil.indexById().keySet() in allUsedFieldIds() in order to take MAP + LIST columns into account

[huaxingao]

LGTM

[szehon-ho]

fyi @aokolnychyi

[szehon-ho]

Oh, also can you add a test for List as well?

[singhpk234]

LGTM too

+1 to @szehon-ho's suggestion

[antonlin1]

@szehon-ho added list column unit test here: afb97ac

[singhpk234]

Thanks for the change @antonlin1 ! Thank you for review @huaxingao @szehon-ho !