adriangb opened a new pull request, #24086: URL: https://github.com/apache/datafusion/pull/24086
## Which issue does this PR close? - Part of #18470. Alternative / companion experiment to #23492. Related: apache/arrow-rs#6946. ## Rationale for this change #23492 amortizes round trips by widening the *blocking* fetch: when the decoder asks for the current row group's ranges, upcoming row groups' projected column-chunk ranges are appended to the same `get_byte_ranges` call, within a byte budget. That helps a lot on high-latency stores, but I/O and decode still strictly alternate per stream — while a gulp is in flight nothing decodes, and while a row group decodes nothing is fetched. Issue #18470 asks for exactly that overlap. @alamb's earlier overlap POC (#18391) demonstrated the win but had no memory bound. This POC combines both ideas: while the current row group decodes, a *background* task prefetches upcoming row groups' projected ranges under the same byte budget as #23492. The point of putting it up as a draft is to make the batched-vs-pipelined comparison concrete for the discussion on #23492, with numbers from the same harness and the same budget accounting. ## What changes are included in this PR? An env-var-switchable fetch-scheduling policy for the parquet push-decoder stream (`DF_FETCH_POLICY=off|batched|pipelined`, `DF_FETCH_BUDGET` in bytes, default 20MB): - **`off`** — current `main` behavior: fetch exactly what the decoder asks for, when it asks for it. - **`batched`** — #23492 semantics: append upcoming row groups' complete projected column-chunk ranges to the same blocking `get_byte_ranges` call, as long as `buffered + staged <= budget`. - **`pipelined`** (the new part) — when a row group's reader is handed over for decode, spawn a tokio background task that fetches upcoming row groups' projected ranges under the same byte budget, overlapping I/O with decode. A half-budget hysteresis keeps the background gulps large (without it, a nearly-full budget degrades to one round trip per row group). A `ReaderSlot` enum (`Idle` / `Busy` / `Empty`) lends the `AsyncFileReader` to the background task and reclaims it (landing any fetched bytes) at the next `NeedsData`. Files: - `datafusion/datasource-parquet/src/push_decoder.rs` — `FetchPolicy`, `ReaderSlot`, the budgeted range computation shared by both policies, the batched extension of the blocking fetch, and the pipelined background prefetch + reclaim in the stream driver. Also a `PEAK_STAGED_BYTES` atomic so benchmarks can observe peak staged bytes and verify both policies respect the same budget. - `datafusion/datasource-parquet/src/opener/mod.rs` — wraps the file reader in `ReaderSlot::Idle` and threads the policy + metadata into the stream state. - `datafusion/datasource-parquet/src/mod.rs` — re-exports `PEAK_STAGED_BYTES` for the benchmark harness. **This is a benchmarking POC and is NOT proposed for merge as-is**: configuration is via env vars, the budget is not integrated with the `MemoryPool`, and the background `JoinHandle` detaches on stream drop. It exists to make the batched-vs-pipelined comparison concrete. ### Benchmark results Simulated-latency `ObjectStore` over a 335MB / 64-row-group file, ~5.5MB projected per row group, single partition unless noted, medians of 3, aggregate query: | latency | budget | off | batched (#23492 semantics) | pipelined (this POC) | |---|---|---|---|---| | 0ms | 20MB | 1230ms | 1008ms | 914ms | | 10ms | 20MB | 2113ms | 1306ms | 1248ms | | 50ms | 20MB | 4738ms | 2329ms | 2613ms | | 50ms | 100MB | 4730ms | 1209ms | 1217ms | | 50ms, 8 partitions | 20MB | 714ms | 377ms | 498ms | Time-to-first-batch on a streaming scan at 50ms latency: off 130ms / batched 324ms (budget 100MB; the first batch waits for the whole first gulp) / pipelined 123ms. **Interpretation:** at equal budget, batched and pipelined total times are within ~15% of each other and flip depending on the latency:transfer ratio — batched does fewer, larger blocking gulps; pipelined pays ~1.4x the round trips in exchange for overlap under the same memory bound. Pipelined never regresses time-to-first-batch, while batched's TTFB penalty grows with the budget. Both remain limited by a single fetch in flight per stream; concurrent budgeted readahead would dominate both, and is the suggested longer-term direction. ## Are these changes tested? Benchmarked as above (and `PEAK_STAGED_BYTES` used to verify both policies stay within the configured budget). No new unit tests — this is a POC for A/B measurement, not a merge candidate; existing tests cover the default `off` path, which is unchanged behavior. ## Are there any user-facing changes? No. The default (`DF_FETCH_POLICY` unset) is the current behavior; the new policies are opt-in via env vars for benchmarking only. 🤖 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
