feat: prune unread Parquet leaves when a nested column is cast to a narrower type

[mbutrovich]

Which issue does this PR close?

• Related to Comet native Parquet scan reads full nested columns despite pruned Spark ReadSchema datafusion-comet#4859. No DataFusion issue is filed for this one.

Rationale for this change

When a table declares a nested column narrower than the Parquet file's physical type, DataFusion reads every leaf of the column and drops the extra subfields in memory instead of skipping them at read time.

• DefaultPhysicalExprAdapter rewrites the projected column into CAST(col AS narrow_type).
• Projection mask derivation only understands literal get_field chains, so the cast's inner column expands to every physical leaf via ProjectionMask::roots.
• Comet reported a production query reading 1.35 TB where plain Spark read 30.9 GB for the same pruned ReadSchema (datafusion-comet#4859).
• Any embedder that hands DataFusion a pre-pruned schema (Comet, delta-rs, Iceberg integrations) hits the same gap.

This is a port of #23398 onto current main. #23398 (stacked on the merged #23396 and #23397, superseding an earlier attempt at #23392) implements the fix and was reviewed favorably, but has merge conflicts against main since a follow-up refactor moved PushdownChecker and PushdownColumns into projection_read_plan.rs, and has four unanswered review comments. This PR reimplements the same approach against current main and resolves those four comments by construction:

• Drops the nested_projection_pruning config flag; the clip's total fallback design makes a kill switch unnecessary.
• Drops an unreachable zero-name-overlap code path; validate_struct_compatibility already rejects that case during physical planning.
• Drops the untested union of a cast clip with a sibling get_field access on the same root, in favor of a full read fallback for that case.
• Notes that ListView, LargeListView, and Dictionary wrappers are conservatively left unclipped, a candidate follow up.

What changes are included in this PR?

• New module datafusion/datasource-parquet/src/nested_schema_pruning.rs. clip_for_cast walks the physical and cast target type trees together, matches struct fields by name, recurses through List and LargeList, and returns the kept leaf offsets plus the pruned Arrow type in one pass. The clip is total: maps, dictionaries, wrapper-kind mismatches, and any shape it does not understand keep every leaf, so the worst case is today's full read.
• PushdownChecker in projection_read_plan.rs now also collects CastColumnAccess entries for a CastExpr over a plain Column where the cast requires nested struct handling. Collection is opt in and only enabled for projection analysis, so filter pushdown is unchanged.
• build_projection_read_plan routes to a new build_read_plan_with_cast_clipping when any cast access survives. It partitions referenced roots into whole column reads, cast clipped reads, get_field only reads, and full read fallbacks, including a fallback for a root reached by both a cast and a separate get_field access.
• Fixed a pre-existing bug in the has_struct_columns fast path: it only tested the top level field type against Struct, so a LIST STRUCT root such as events was misclassified as containing no struct and skipped PushdownChecker entirely, meaning cast based clipping never fired for that shape. Replaced with a recursive contains_struct check through List, LargeList, ListView, LargeListView, FixedSizeList, Map, Dictionary, and RunEndEncoded.
• No new configuration option.

Are these changes tested?

• 15 unit tests in nested_schema_pruning.rs: struct subset, reordering, leaf promotion, missing field null fill, nested struct in struct, list of struct, two levels of list of struct nesting, maps, dictionaries, wrapper mismatches, and an arrow-rs roundtrip test pinning that ProjectionMask::leaves over a subset of List Struct leaves emits exactly the predicted type.
• 2 new unit tests in projection_read_plan.rs, plus the pre-existing struct preservation test, unchanged.
• 8 integration tests in datafusion/core/tests/parquet/expr_adapter.rs, asserting identical results and a bytes_scanned drop of more than 2x: list of struct narrowing, top level struct, struct level nullability, get_field on a narrowed struct, mixed whole column and subfield access, filter pushdown enabled, a scan mixing a physically narrow and a wide file, and a regression test built from the ReadSchema and InputSchema shapes reported in datafusion-comet#4859 (a two level list STRUCT column with a dropped struct sibling, a dropped map sibling, and dropped top level columns).
• New datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt covering the end to end SQL path.
• The existing benchmark (already merged as part of bench: parquet scan with a table schema narrower than a nested column #23397) had an assertion documenting the pre-fix baseline, narrow equals full. That assertion now fails as expected and has been flipped to assert narrow reads less than half of full. Measured for the top level struct case: narrow_schema=1081 bytes vs full_schema=8398289 bytes, matching the physically_narrow floor of 1081 bytes.

Are there any user-facing changes?

No API changes and no new configuration option. Behavior is IO reduction only, results are unchanged.

[adriangb]

Can we somehow assert that the pruning worked / the pad columns where not read?

[mbutrovich]

Added a full_schema table (no cast) and pinned bytes_scanned as a literal: 172 narrow vs 312 full. Rest stays masked, like limit_pruning.slt:103.

[adriangb]

Thank you. I'll try to check that we have full coverage via SLT tests and if there is any low hanging fruit etsting we can add as SLTs.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 99.43630% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.02%. Comparing base (3e3a92d) to head (bd09d48).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
...ion/datasource-parquet/src/projection_read_plan.rs	99.32%	2 Missing and 1 partial ⚠️
...on/datasource-parquet/src/nested_schema_pruning.rs	99.55%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24090      +/-   ##
==========================================
+ Coverage   80.91%   81.02%   +0.10%     
==========================================
  Files        1103     1105       +2     
  Lines      377219   379704    +2485     
  Branches   377219   379704    +2485     
==========================================
+ Hits       305244   307663    +2419     
- Misses      53775    53821      +46     
- Partials    18200    18220      +20     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[adriangb]

This generally looks good.

One high level concern I had was why this doesn't work / isn't enabled for filter pushdown (it would seem valuable there, although maybe Spark/Comet don't use filter pushdown so it's a mute point). TLDR is filter pushdown would not accept a filter like struct_col = struct(...). It will however accept s['x'] = 5 and if s has different fields in the logical and physical schema we end up with get_field(cast(s as <logical type>), 'x') = 5 which (1) disables the optimization where we do leaf = 5 directly and (2) brings us back to the cast piercing this PR is doing. But it feels like we can handle that in a followup, even if it results in some large refactoring of this PRs code.

I've kicked off some checks for coverage and a couple review areas, I'll report back tomorrow. We should leave this open for feedback for a bit longer as well.

[adriangb]

Is it intentional that this not work for filter push down? I don't immediately see any reason why it couldn't work.

[adriangb]

Thank you. I'll try to check that we have full coverage via SLT tests and if there is any low hanging fruit etsting we can add as SLTs.

[adriangb]

I spent some time stress-testing this (mutation testing, randomised differential testing against the arrow-rs reader, and benchmarks on wide schemas) and found one correctness bug plus a few smaller things. I've opened mbutrovich#1 against your branch with fixes and tests — fold it in however you like.

Two cast targets on one column read too few leaves

build_read_plan_with_cast_clipping skips every cast access on a root it has already clipped:

if whole_roots.contains(&root)
    || fallback_roots.contains(&root)
    || clipped_by_root.contains_key(&root)   // <-- second cast on this root: dropped
{ continue; }

If a projection consumes one column through two different narrowing casts, only the first target's leaves reach the mask, and the second cast then evaluates against a struct missing the children it names:

• overlapping targets → cast_column null-fills them → silently wrong results
• disjoint targets → validate_struct_compatibility rejects it → the query fails where it succeeds without pruning

I originally assumed this needed a custom adapter, but it's reachable from plain SQL: ProjectionExec is merged into the scan via ParquetSource::try_pushdown_projection, so a query-level CAST(col AS STRUCT<...>) reaches the same analysis as an adapter-inserted one. Against a table whose schema is inferred from the file (so no adapter cast is interposed):

-- file: s Struct<x Int64, y Utf8, pad Utf8>
SELECT CAST(s AS STRUCT<x BIGINT>), CAST(s AS STRUCT<x BIGINT, y VARCHAR>) FROM t;
-- {x: 100} {x: 100, y: NULL}   <- y should be 's1'

SELECT CAST(s AS STRUCT<x BIGINT>), CAST(s AS STRUCT<pad VARCHAR>) FROM t;
-- Error: Cannot cast struct with 1 fields to 1 fields because there is no field name overlap

Both return correct results with with_cast_collection disabled, so they're regressions rather than pre-existing behaviour. The fix in the linked PR demotes the root to a full read when a second cast has a different target; identical repeated targets (what the expression adapter actually produces) still clip.

Smaller things

• A zero-overlap nested struct level predicts a type the reader never emits. clip_type can return Struct([]) for a name-matched child whose own children share no name with their target, while keeping that child in the emitted type — but arrow-rs drops a struct child whose leaves are all masked out (complex.rs, if children.is_empty() { return Ok(None) }). I couldn't reach it through the default stack (validate_struct_compatibility and the logical planner both reject the cast first), so it isn't live today. But the module doc's safety argument rests on a caller invariant and the motivating use case is a custom PhysicalExprAdapter, so I'd rather clip_for_cast enforce it itself. Test added pinning the arrow-rs behaviour the argument depends on.
• leaves_by_root[root] is a panicking BTreeMap index, on a path where the same function already guards the "root with no parquet leaves" case twenty lines earlier. Latent — I couldn't build a file that reaches it — but a one-line fix.

Performance

Three things weren't O(projected_columns):

• clip_type matches struct fields with a linear find per physical child — Θ(N·M) string comparisons per struct level, ~375k for a 1000→500 subfield struct. Spark's ParquetReadSupport.clipParquetGroupFields uses a name→field map for exactly this; the PR does the same above a small width.
• Widening the fast-path gate to contains_struct correctly fixes the List<Struct> hole, but it also routes Map, Dictionary<_, Struct> and every array-of-records column through PushdownChecker, whose Schema::index_of is a linear name scan per column node — O(|exprs| · |schema|), paid once per file opened. Gating on the projected roots instead keeps the List<Struct> fix and is O(projected).
• An O(G²) field_with_name re-lookup for get_field roots (also picks the wrong field under duplicate root names).

New benchmark parquet_wide_schema_read_plan — 1000-column schemas and 1000-subfield structs over 32 small files, so per-file planning dominates. Before → after, back to back, with two untouched-path controls so the numbers mean something:

case	change
flat_wide/column_projection (control)	−0.1% (p = 0.93)
wide_struct/full_schema (control)	+1.1% (p = 0.63)
wide_struct/unrelated_column — wide struct present but not projected	−13.6% (p = 0.01)
wide_struct/narrowed_schema — 1000 → 500 subfield clip	−7.2% (p = 0.01)

The feature holds up well at that width: 90.8 ms narrowed vs 166.2 ms full, ~45% less wall time.

Coverage

cargo-mutants over the two files: 121 mutants, 82 viable, 16 survivors, of which 8 were real gaps — including that start + o (rebasing clip offsets onto absolute leaf indices) survives start - o, because every clip test casts to the struct's first field where the two are identical; and that the whole get_field_accesses branch of build_read_plan_with_cast_clipping is never executed by any test. All eight now have tests, each verified by re-applying the mutation.

I also added two seeded randomised differential harnesses — one checking clip_for_cast's predicted type and kept leaves against the real reader, one comparing a clipped scan against cast_column over an unclipped one end to end. ~63k and ~25k cases respectively, no further discrepancies. They're cheap by default (~3.5 s / ~1.2 s) and widen by env var.

SLT coverage went from 110 to ~380 lines, moving most of the Rust-only assertions into SQL: SELECT *, mixed whole-column and field access, aggregation and filtering over a clipped column, a declared field order differing from the file's, struct-in-struct, the two-level ARRAY<STRUCT<..., ARRAY<STRUCT<..>>>> shape from the Comet issue, a MAP sibling, a scan mixing narrow and wide files, and the zero-overlap rejection.

One thing that isn't yours

While writing the SLT tests I hit a pre-existing soundness bug: with pushdown_filters = true, a get_field predicate is silently dropped whenever the file needs schema adaptation, so the query returns every row. It's not caused by this PR (row_filter.rs is untouched), but every table this feature targets carries exactly the cast that triggers it, so it's worth knowing about before this lands. Filed separately as #24109.

---

Generated by Claude Code

[mbutrovich]

I spent some time stress-testing this (mutation testing, randomised differential testing against the arrow-rs reader, and benchmarks on wide schemas) and found one correctness bug plus a few smaller things. I've opened mbutrovich#1 against your branch with fixes and tests — fold it in however you like.

Does it address the earlier review comments?

[adriangb]

I spent some time stress-testing this (mutation testing, randomised differential testing against the arrow-rs reader, and benchmarks on wide schemas) and found one correctness bug plus a few smaller things. I've opened mbutrovich#1 against your branch with fixes and tests — fold it in however you like.

Does it address the earlier review comments?

It addresses the coverage aspect, it doesn't implement this for filter pushdown, I think we should defer that to followups. If we merge mbutrovich#1 into here we can then merge this PR into main and track a couple of followups, namely:

• Getting this optimization to work for filter pushdown
• Taking the union of fields from a cast instead of reading the whole column

[adriangb]

Thanks @mbutrovich !