POC: batch-granular parquet scan scheduling (and two row-group-granular alternatives) - #24086
Draft
adriangb wants to merge 9 commits into
Draft
POC: batch-granular parquet scan scheduling (and two row-group-granular alternatives)#24086adriangb wants to merge 9 commits into
adriangb wants to merge 9 commits into
Conversation
…the parquet push decoder Experimental alternative to apache#23492: instead of widening the blocking fetch, spawn a background fetch of upcoming row groups' projected ranges (same byte budget) while the current row group decodes, so I/O overlaps decode. Env-var switched (DF_FETCH_POLICY / DF_FETCH_BUDGET) for A/B benchmarking; not intended to merge in this form. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ader Adds DF_FETCH_POLICY=streaming: instead of buffering whole row groups, one long-lived sync ParquetRecordBatchReader pulls bytes through a shared in-memory ChunkReader. The stream driver computes from the offset index exactly which page ranges the next batch needs, awaits their fetch, and keeps up to DF_FETCH_BUDGET bytes of background readahead in flight; page bytes are dropped as soon as the decode cursor passes them. Because the reader persists for the whole file, dictionary pages are fetched and decoded exactly once, there is no per-row-group reader rebuild, and resident memory is bounded by the readahead window rather than row-group size. Supports projections, RowSelections (page skipping preserved via selected-row prefix sums), limits, and multi-RG files; falls back to the push-decoder path when row filters are active or the offset index is unavailable. Local simulated-latency results (300MB single-row-group file, 50ms latency, 100MB window, streaming scan): time-to-first-batch 112ms vs ~700ms for all row-group-granular policies, total 708ms vs ~845ms, peak staged 100MB vs 283MB. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
adriangb
changed the title
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "pipelined"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "4G"
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "pipelined"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "pipelined"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
The required-only first fetch keeps TTFB low on large files, but for a file whose whole plan fits the readahead window it turns one fetch wave into two serial waves — an extra round trip per file, which dominates many-small-file workloads (TPC-DS under simulated latency regressed 1.3x). Fetch the entire plan in one wave when it fits the window; required-only kicks in only for plans larger than the window, preserving the giant-row-group TTFB and memory wins. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (5ff7af6) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "pipelined"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (075e816) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (075e816) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (075e816) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24086 +/- ##
==========================================
- Coverage 80.85% 80.80% -0.05%
==========================================
Files 1099 1102 +3
Lines 374304 376803 +2499
Branches 374304 376803 +2499
==========================================
+ Hits 302642 304476 +1834
- Misses 53607 54204 +597
- Partials 18055 18123 +68 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
…efault Three fixes from investigating why clickbench showed no change: - Log when the streaming policy is requested but cannot run (no offset index, or pushdown filters active). The fallback was silent, which is why it took three benchmark rounds to notice that the published ClickBench files carry no page index at all (0 of 105 columns) and the path was never executing. - Install every planned page a fetched blob covers, not just the current wave's. Coalescing over-fetches the gaps between pages and those gaps can contain pages planned for a later wave. Measured worth only ~7MB of 157MB on clickbench, so this was not the source of the over-fetch I suspected, but re-fetching bytes already in hand is still wrong. - Lower the default coalesce gap from 4MB to 1MB, matching object_store's OBJECT_STORE_COALESCE_DEFAULT. 4MB was a guess and measurement contradicts it: on the page-index ClickBench copy it merged away 59 requests but pulled 157MB of unprojected columns with them and ran slower (q22 204ms/236MB vs 170ms/212MB at 1MB). With several partitions fetching concurrently a saved round trip is worth far less than latency x bandwidth suggests. At 1MB, bytes fetched return to parity with the unscheduled path (1655MB vs 1650MB) and the speedup holds at 1.53x. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run benchmark tpch10 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch10
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
`ObjectStore::get_ranges` already coalesces ranges within 1MB for every store using the default implementation (S3, GCS, Azure), while LocalFileSystem and the other overriding stores coalesce not at all. A second pass in the scheduler can therefore only raise the effective threshold, never lower it, and it makes the merge decision without knowing the medium — so DF_FETCH_COALESCE was both redundant on real object stores and unable to express "don't merge" there. It also never earned its place. Removing it entirely is marginally faster than any setting of it on the page-index ClickBench copy (50ms latency, 8 partitions, medians of 2): 1.54x vs 1.53x at 1MB and 1.51x at 4MB, with bytes fetched back near parity with the unscheduled path (1680MB vs 1650MB, against 1807MB at 4MB). Per query: 35 faster, 0 slower, 1 neutral. The scan's own wins do not depend on it: the tpch 1.69x and tpch10 6.4x results were measured at bcbbda1, before this code existed. What remains is the actual idea — batch-granular readiness, so decode starts on the first pages instead of waiting for a whole row group. If a scheduler-side merge is wanted later, it should be driven by characteristics the store reports rather than a constant, so that the two layers stop duplicating a decision neither can make alone. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
Author
|
run benchmarks |
Contributor
Author
|
run benchmark tpch10 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch10
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch10
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (ee1db7b) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark tpcds
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark tpch10
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing claude/datafusion-rowgroup-buffering-ec9abc (7f3e448) to 39d5064 (merge-base) diff Run configurationrun benchmark clickbench_partitioned
env:
DF_FETCH_BUDGET: "104857600"
DF_FETCH_POLICY: "streaming"
SIMULATE_LATENCY: "true"CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
adriangb
changed the title
Contributor
Author
Benchmark summaryAll GKE runs are adriangbot on With simulated latencyCommit
Neither policy dominates. Without simulated latency (local NVMe)Every suite neutral on wall time for both policies — per-batch scheduling costs nothing on fast storage — with memory consistently lower for ClickBench: the files have no page indexclickbench reported all 43 queries as "no change", which is the signature of a code path not executing rather than executing without benefit. It isn't: the published So page-granular planning cannot run there by construction, and only row-group-granular prefetch can help — which is exactly why Adding a page index changes the answer completely. I downloaded 3 of the 100 files, rewrote them with
Per query on the page-index copy: 35 faster, 0 slower, 1 neutral. The The mechanism is not fetching less. For most queries the GET count and bytes fetched are identical between Caveats: 3 of 100 files, so absolute numbers do not extrapolate; Synthetic: where the ceiling isA 300MB file written as a single 8M-row row group — the case row-group-granular policies cannot help at all (50ms latency, 100MB window):
Same effect the ClickBench page-index experiment later reproduced on real data. On a 4 × 71.5MB row-group file, Hypotheses that measurement killedRecorded because the wrong ones cost real time:
That last one has a layering lesson worth stating: ReproducingLocal experiments used a benchmark harness with an |
The comment said the streaming path "bypasses the push decoder entirely", which is true of the decoder but reads as contradictory now that the path imports `plan_scan_ranges` from arrow-rs's `push_decoder` module. Only the plan comes from there; the decoder is not constructed, because `NeedsData` resolves at row-group granularity and so cannot express what the next batch needs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
Today a Parquet scan buffers a whole row group before it decodes anything: the push decoder's
NeedsDatadoes not resolve until every projected byte of the row group has landed. Two consequences, both visible under object-store latency:#23492 addresses (1) by widening the blocking fetch: when the decoder asks for row group N, it appends upcoming row groups' whole projected column chunks to the same call. That amortizes round trips well, but I/O and decode still alternate, and it does nothing about (2).
This PR explores what happens if you attack the granularity instead, and includes the intermediate schedules so the comparison is concrete rather than asserted.
What changes are included in this PR?
Three fetch-scheduling policies behind
DF_FETCH_POLICY, with a byte budget inDF_FETCH_BUDGET(default 20MB). Default unset = current behavior, unchanged.batched— a faithful reimplementation of Prefetch Parquet row groups with a bounded I/O budget #23492's logic, so it can be A/B'd here on identical hardware.pipelined— when a row group's reader is handed over for decode, spawn a background fetch of upcoming row groups' projected ranges under the same byte budget, so I/O overlaps decode. This is the shape of alamb's earlier POC Prefetch Row Groups usingnext_readerAPI in parquet-rs #18391, but memory-bounded.streaming— batch-granular readiness. One long-lived syncParquetRecordBatchReaderpulls bytes through a shared in-memoryChunkReader; the stream driver asks arrow-rs which page ranges the next batch needs, awaits exactly those, and keeps bounded readahead in flight. Pages are dropped as the decode cursor passes them.streamingis the interesting one. Because the reader persists for the whole file, dictionary pages are fetched and decoded once, there is no per-row-group reader rebuild, resident bytes are bounded by the readahead window rather than by row-group size, and the first batch is emitted after its first pages land rather than after a whole row group.The arrow-rs side
The demand set — which pages this scan reads, in what order, and when each can be dropped — is computed by a new upstream API,
parquet::arrow::push_decoder::plan_scan_ranges(apache/arrow-rs#10555). An earlier revision of this PR reimplemented that walk inside DataFusion, duplicating logicInMemoryRowGroup::fetch_rangesalready owns; that was the wrong crate for it. DataFusion now keeps only what is genuinely scheduling: readahead budget, wave sizing, and eviction.Because that API is unreleased,
[patch.crates-io]pins every arrow crate to the branch carrying it (pydantic/arrow-rsclaude/push-decoder-peek-59= apache/arrow-rs 59.1.0 plus that one additive module). That pin is why this cannot merge as-is; the intended sequence is landing the arrow-rs side first, then dropping the patch.What this does not do
RowFilterpredicates eagerly over a whole row group insidebuild(), so incremental filter evaluation is a genuine upstream gap, not something schedulable from here.SortExecpushes down) is bypassed on the streaming path.MemoryPoolreservation, so the real bound istarget_partitions × budget. Making it a pool reservation is the obvious productionization step, and the SF10 memory numbers below show why it matters.Are these changes tested?
Benchmarked as below;
PEAK_STAGED_BYTESverifies the policies stay within their configured budget; correctness spot-checked against theoffpath for full scans, page-index-selection queries andLIMIT. No new unit tests — this is a measurement POC, not a merge candidate, and the default path is untouched.Full benchmark results, methodology and the several hypotheses that measurement killed are in the summary comment below.
Headline results
adriangbot GKE runs vs merge-base,
SIMULATE_LATENCY, 100MB budget, commit7f3e448:streamingWithout simulated latency every suite is neutral on wall time, with memory consistently lower.
pipelinedis the better policy for clickbench (445.1 → 335.1s, 1.33x) and never regresses anywhere — the two policies win on different workload shapes, which is the main design finding here.ClickBench is unchanged for a specific and fixable reason: the published
hits_*.parquetfiles contain no page index at all — 0 of 105 columns carry an offset index. Rewriting three of them withwrite_page_index=True, changing nothing else, makes the same 43 queries 1.54x faster (35 faster, 0 slower, 1 neutral). Details in the comment below.Are there any user-facing changes?
No.
DF_FETCH_POLICYunset is today's behavior; the policies are opt-in env vars for benchmarking. The[patch.crates-io]pin means this is not mergeable in its current form by design.🤖 Generated with Claude Code