POC: page-granular scan planning for the Parquet push decoder

[adriangb]

Which issue does this PR close?

• Part of Avoid buffering the entire row group in Async parquet reader #6946 (Avoid buffering the entire row group in Async parquet reader)
• Related: [Parquet] PushDecoder: Add a peek API to support pre-fetching #8668 (PushDecoder: Add a peek API to support pre-fetching)

Rationale for this change

ParquetPushDecoder reports what it needs one row group at a time: DecodeResult::NeedsData does not resolve until every projected byte of the row group is buffered. That is the right granularity when the caller wants a whole row group's reader, but for a caller scheduling its own I/O it means:

• resident bytes are bounded by row-group size, not by anything the caller chooses;
• decoding cannot start until the last byte of the row group lands, so time-to-first-batch is a whole row group of latency; and
• there is no way to ask "what does the next batch need?", which is the question a scheduler actually has.

This adds plan_scan_ranges, which decomposes a scan into the individual pages it will read, in the order decoding needs them, each tagged with the span of selected rows it serves. A caller can then fetch only what the next batch needs, drop pages once its decode cursor passes them, and read ahead as far as its own byte budget allows.

The plan is demand, not schedule: it says which bytes the query will read and when they are first needed. Request sizing, readahead depth and whether to merge nearby ranges stay caller policy, because they depend on the storage medium rather than on the file. (ObjectStore::get_ranges already coalesces for stores that use the default implementation, which is one reason this deliberately does not.)

Units: why row tags rather than a fixed quantum

The obvious shape would be next_batch_ranges() or ranges_for(n_rows). Both bake in a unit the caller may not want. A fixed row count can imply an unbounded byte count — 8192 rows of 1MB values is 8GB — while a pure byte budget cannot express "enough to make progress".

So each planned range carries first_row / last_row in selected-row space instead. A caller budgeting in bytes takes pages until the budget fills; a caller thinking in rows takes pages until the tags cover its window; and last_row is what makes eviction possible at all.

One floor remains and is documented: ParquetRecordBatchReader decodes batch_size rows at a time, so no fetch plan can subdivide a batch. For schemas with very large values the lever is a smaller batch_size, or byte-based batch sizing, which is a separate gap.

What changes are included in this PR?

A single new module, parquet::arrow::push_decoder::scan_plan, exporting plan_scan_ranges, ScanPlan and PlannedRange. No existing code path is touched — nothing calls it inside the crate yet, so this is purely additive.

plan_scan_ranges(metadata, row_groups, projection, selection) returns None when the file has no offset index, since page locations are what make page-granular planning possible; callers fall back to row-group-granular fetching.

The natural follow-up, if this shape is acceptable, is to use it inside the decoder itself so NeedsData can resolve at batch granularity rather than row-group granularity, and to make RowFilter evaluation incremental (today predicates are evaluated eagerly over a whole row group inside build(), which is why the DataFusion side below falls back to row-group granularity whenever filter pushdown is on).

Are these changes tested?

Yes — six unit tests covering page ordering, projection, selection-driven page skipping (including selected-row-space tagging), row-group subsets, the no-offset-index fallback, and that planned bytes exactly cover the projected column chunks when nothing is skipped.

It is also exercised end to end by a DataFusion POC that schedules I/O against this plan: apache/datafusion#24086. Measured there against DataFusion main, with simulated object-store latency:

benchmark	baseline	with batch-granular scheduling
TPC-H SF10	635.1s	100.0s (22 of 22 queries faster)
TPC-H SF1	110.0s	65.0s (21 of 22 faster, 0 slower)
TPC-DS SF1	380.1s	380.1s, peak memory −18%
single 300MB row group	845ms, 700ms TTFB, 283MB peak	708ms, 112ms TTFB, 100MB peak

ClickBench is unchanged there — because the published hits_*.parquet files contain no page index at all (0 of 105 columns), so page-granular planning cannot run on them. Rewriting three of those files with write_page_index=True and changing nothing else makes the same queries 1.54x faster (36 queries: 35 faster, 0 slower, 1 neutral), which is a reasonable argument that the capability is worth having even where files do not use it yet.

Are there any user-facing changes?

New public API only (plan_scan_ranges, ScanPlan, PlannedRange); no behavior change to existing paths. Marked draft because the API shape is the thing worth discussing — in particular whether this belongs as a free function or as a method on the decoder that can also advance it.

🤖 Generated with Claude Code