Native Iceberg scan reads every field of a nested column even when Spark prunes it #6724

What is the problem the feature request solves?

The native Iceberg scan reads and decodes every leaf of a projected struct, list, or map column, even when Spark's nested schema pruning keeps only some of its fields. Results are correct, because the batch adaptation after the read drops the extra fields. The cost is I/O and decode work for data the query never uses.

Cause. For a task without deletes, CometIcebergNativeScan.serializePartitions sends iceberg-rust the full metadata.tableSchema as the FileScanTask schema, with top-level project_field_ids. iceberg-rust's include_leaf_field_id (arrow/reader/projection.rs) expands a struct, list, or map field id to every leaf it finds in that schema. The pruned read schema, metadata.scanSchema (SparkScan.expectedSchema()), is used only when VERSION AS OF reads a dropped column. A task with deletes uses the full FileScanTask.schema().

The code passed scanSchema until #2729. That PR switched to tableSchema because a count(*) scan schema lacked the partition source columns that constants_map needs. schemaWithRequiredFields now adds partition source columns explicitly, so that reason no longer applies. CometIcebergNativeSuite already notes the behavior: "The native read asks for the column's full nested type even when Spark prunes it."

Impact. These numbers come from Spark event logs of TPC-H SF1000 with a nested schema on Iceberg (Spark 4.0, 32 executors x 16 cores). In that schema each table keeps its non-key columns in an array<struct> column, such as partsupp_data.

• Q20 needs only ps_availqty from partsupp_data, and Spark prunes the struct to it. Comet's partsupp scan still reads 26.4 GiB, which is every byte of the table's files. Q16 reads only the top-level keys of the same table and scans 1.2 GiB. The partsupp scan is about 62% of Comet's Q20 time, and its task time is only 14% below Spark's. In that run Q20 was level with Spark (28.6 s vs 28.1 s), while Comet was faster on most other queries.
• The gap matters only where Spark itself prunes a nested column. An explode adds a size(col) > 0 filter on the whole column, which blocks Spark's pruning unless the explode was rewritten to a single field. Across the 22 queries, Spark prunes 12 of 66 scans. They carry 18% of Comet's scanned bytes. On them Comet's scan-stage task time is 0.53x Spark's, against 0.38x on scans where both engines read the whole column and 0.25x on flat scans.
• Q2 scans partsupp twice, once pruned by Spark and once not. Spark spends 724 s of task time on the pruned scan and 1,309 s on the other. Comet spends about 695 s on each.
• The affected scan is on the critical path of Q14 (part, 6.7 of 7.6 s), Q18 (orders), and Q20 (partsupp).

Describe the potential solution

Pass the pruned scan schema to iceberg-rust instead of the table schema. I have a working prototype:

• The task schema is metadata.scanSchema without its metadata columns, for every task, including tasks with deletes. Metadata columns are field ids from Integer.MAX_VALUE - 200 up, which a table schema never has. Partition source columns and equality-delete keys are still added through schemaWithRequiredFields.
• schemaWithRequiredFields is memoized by base schema and field ids. Without that, a partition source the query does not project makes every task build its own schema, and the schema pool grows to one entry per task.
• A task keeps the full schema when a missing partition source or equality-delete key is nested inside a struct the query pruned. That field would otherwise be appended at the top level. A table with a top-level region and an s.region partition source then fails with Invalid schema: multiple fields for name region.
• A new config, spark.comet.scan.icebergNative.nestedSchemaPruning.enabled (default true), turns it off.

The nested-evolution fallback from #6543 stays as it is. It checks the full column type in the table schema, so it still falls back when the evolved field is pruned away. A follow-up could check only the pruned fields and keep more queries native.

Prototype results.

• Four new tests pass. They cover struct, nested struct, list, and map value pruning with NULL validity, checked against Spark. They also cover deletes, partitions, VERSION AS OF, the nested partition source case, one schema in the pool, and data files without field ids.
• CometIcebergNativeSuite passes 120 of 121, with one version-gated cancel. CometFuzzIcebergSuite, CometIcebergResidualPushdownSuite, and CometIcebergRewriteActionSuite pass.
• A new case in CometIcebergReadBenchmark reads a 4M-row nested partsupp table (zstd, comments of about 128 characters). It ran on one local core with a warm page cache, using a release native build:

Bytes scanned for the first query drop from 124.0 MiB to 7.5 MiB. Locally the gain is all decode work, since the data sits in the page cache. On object storage the smaller read should add to it. That part is not measured yet.

Additional context

• iceberg-rust matches nested children by position rather than by field id (Type casting error when reading files persisted with old schema for complex type iceberg-rust#2617). The pruned read projects the same leaves from the file as the task schema holds, so positions line up for every file the fix: fall back from the native Iceberg scan when a nested field was added or renamed #6543 fallback lets through.
• For data files without field ids and no name mapping, iceberg-rust projects whole top-level columns by position. Arrow's struct cast then narrows each one to the pruned type by field name. Results are correct, but there is no I/O saving.
• iceberg-rust merges reads whose gaps are under 1 MiB (DEFAULT_RANGE_COALESCE_BYTES). On tables with small files the I/O saving is smaller than the size of the pruned fields suggests. The decode saving remains.
• The cluster check still to do is TPC-H Q14, Q18, and Q20 on the nested schema, expecting Q20's partsupp scan to drop from 26.4 GiB to a few GiB.

Willingness to contribute

I can contribute a fix for this bug independently