feat: prune nested Parquet leaves when the projected schema narrows a nested column

[mbutrovich]

Which issue does this PR close?

• Relates to Introduce ProjectionMask To Allow Nested Projection Pushdown #2581. (A step toward general nested projection pushdown, not a full close.)

Rationale for this change

build_projection_read_plan prunes Parquet leaves only for get_field chains on struct columns. Every other projection falls back to ProjectionMask::roots and reads all leaves of the top-level column. A projection that requests a narrower nested type for a whole column (for example array<struct<a,b>> narrowed to array<struct<a>>, or a narrowed map value struct) has no get_field form and reads the entire column.

This is the shape produced when a caller hands DataFusion a pre-pruned nested schema (Spark nested schema pruning via DataFusion Comet, and similarly delta-rs and Iceberg integrations). The schema adapter rewrites the logical-vs-physical type mismatch into a whole-column CastExpr(Column, narrow_type), so the pruning survives only as a type on the projection, not as expressions the mask derivation understands. The full column chunks are then fetched and decoded and the cast drops the unwanted subfields in memory, so bytes_scanned does not improve even though the output is correct. The parquet reader is not the limit: ProjectionMask::leaves already handles arbitrary nesting. The gap is that nothing translates a schema-level narrowing into a leaf mask.

What changes are included in this PR?

Add try_nested_projection_leaves, run after the plain-column fast path in build_projection_read_plan. For each projection expression referencing a single file column, it compares the expression output type (expr.data_type) to the file column type. When the output type is a nested type narrower than the file's, prune_and_collect walks the two type trees, matches fields by name, and emits a ProjectionMask::leaves of the reached leaves plus the pruned schema (built from the file type so names and nullability match the decoder output). This is the equivalent of Spark's ParquetReadSupport.clipParquetSchema. It returns None, deferring to the existing path, when any expression is not a single-column reference, a root repeats, or the requested type is not a structural subtree, so get_field and all current behavior are unchanged. Per-root leaf offsets come from SchemaDescriptor::get_column_root_idx and pruned fields are rebuilt with Field::with_data_type.

Alternative considered: trigger on the logical-vs-physical file schema diff in the opener instead of expr.data_type. That is more general (it does not depend on an adapter-inserted cast) but requires threading the logical file schema into the mask derivation. Keying off expr.data_type needs no signature change and covers the cast the adapter already produces. Open to either.

Not addressed here, worth follow-ups: map key/value handling, case sensitivity, and field-id matching for Iceberg. A pluggable clipping policy (mirroring the existing expr-adapter factory) would let embedders supply the matching rule.

Are these changes tested?

Unit tests in row_filter.rs cover prune_and_collect for array<struct>, array<struct<struct>>, and the defer case (a primitive get_field-style request). Existing datafusion-datasource-parquet tests pass. Validated end to end in DataFusion Comet against Spark output (nested reader and fuzz suites), where bytes_scanned for a single nested leaf drops to a fraction of the full column. A DataFusion-native integration test is worth adding: CREATE EXTERNAL TABLE over a file with a wide nested column declaring a narrower nested type, select it, and assert bytes_scanned drops.

Are there any user-facing changes?

Fewer Parquet leaves are read for projections that narrow a nested column. Results are unchanged and there is no API change.

[kosiew]

@mbutrovich
Thanks for working on this. The core direction makes sense, but I think we should add an integration-level regression test.

[kosiew]

The added tests exercise prune_and_collect directly, which is helpful, but they do not cover the main behavior this PR changes: recognizing a real CastExpr(Column, narrow_type) projection and turning it into the correct read plan.

A regression in try_nested_projection_leaves dispatch, expr.data_type(file_schema), ProjectionMask::leaves, or projected_schema construction would not be caught by the current tests.

Could you please add at least one integration-level regression test through build_projection_read_plan or DecoderProjection::try_new using the real CastExpr(Column, narrow_type) shape, and assert the leaf mask and projected schema for array<struct<a,b>> -> array<struct<a>>?

Since the PR explicitly handles maps, it would also be good to include a narrowed map value struct case. Otherwise, we should probably defer map support if it is not intended to be covered here.

[kosiew]

Small style thought: consider importing HashSet with the existing collection imports, or using BTreeSet consistently in this file. Not a blocker, but it would keep this helper visually aligned with the surrounding projection-pruning code.