adriangb opened a new pull request, #10555: URL: https://github.com/apache/arrow-rs/pull/10555
# Which issue does this PR close? - Part of #6946 (Avoid buffering the entire row group in Async parquet reader) - Related: #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](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
