Repository navigation
[bench only] #10707 with the previous V2 chunk concurrency default - #10759
Draft
joseph-isaacs wants to merge 13 commits into
Draft
joseph-isaacs wants to merge 13 commits into
joseph-isaacs wants to merge 13 commits into
Conversation
A `PreparedScan` holds a scan's plans over one file, built once from the footer, and makes its planners; nothing in it touches a segment and no planner plans again. The planners form a chain, each emitting the next through the planning protocol: - `PrunePlanner` proves the filter's zone maps over the scan's rows, prefetching the zone tables together, and emits one chunk planner for the rows no proof excludes. Rows a proof excludes are never planned. - `ChunkPlanner` cuts those rows into chunks and emits a filter or a projection planner per chunk, skipping chunks nothing is selected or kept of. The run drives an emitted chunk ahead of the next, so chunks waiting on reads overlap. - `FilterPlanner` prefetches its chunk's conjunct segments and announces the projection's at its first compute, runs the conjuncts in the adaptive order over the surviving rows, and emits the projection. - `ProjectionPlanner` emits one morsel per piece of the rows. A stage with nothing to do is never made: a filter without proofs starts at the chunk planner, a scan without a filter at the projection. A scan runs as one task owning one planning run, instead of one task and run per split plus a private run for the proofs; `with_concurrency` sets nothing. The IO session keeps every read it starts, so a later stage or morsel of a chunk that reads the segment again shares the read. The segments a stage prefetches come from compiling its plans. DataFusion attaches each partition's rows from the cached footer when it repartitions a file, so the V2 opener reads exactly those rows instead of translating a byte range through the layout. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
A scan reads a zoned column through the zoned node's data child and never its zone table, yet the node and its zone table stayed in every optimized data plan. A new `ZonedDataRule` replaces a zoned data plan by its data, so an optimized plan carries a zone table only where a pruning proof reads it. Legacy stats get the same treatment through `LegacyStatsRule`; the rule that dropped legacy stats beneath a zoned plan is subsumed. A zone proof over legacy stats depended on the converted zoned node surviving child optimization. `ExpressionLegacyStatsRule` now converts the legacy child under a proof directly, through the shared `is_zone_proof` check, so the pruning rewrite still meets a zoned plan. With zone tables gone from data plans, the identity projection is optimized and cached like every other projection: chunked columns stay lazy through `Concat::map_chunks`, and a projection-only scan no longer lowers the zone tables it never reads. (cherry picked from commit 91be25d) Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
DataFusion opens a file once per partition, so each partition lowered the layout, planned the filter and proved the zone maps on its own. A scan cache now keeps each distinct filter's plans along with the zones their proofs excluded, and the opener keeps one cache per file that concurrent partitions find through a weak handle and that the last scan holding it frees. A prune planner skips the zone maps of a proof an earlier scan of the file already proved: they are not read again. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The stream a scan returns now owns the planning run and drives it as it is polled. Every CPU step of the run, the planners' and morsels' compute, is a `spawn_cpu` task; every read is already a `spawn_io` task of the segment source; the stream's poll awaits one or the other and does no scan work on the engine's thread. The scan advances only as fast as its consumer pulls, so the task and channel that ran ahead of it go. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
A prune planner no longer knows about chunks or selections. It is built with the planner it continues into, as a function of the rows its proofs keep, and emits that planner with the kept mask once every proof has run. `PreparedScan::planner` supplies the chunk planner. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…h other A planner may now emit a child with `PlannerOutput::Spawn`: the run does not queue it but hands it to its owner as `Progress::Spawn`, and the planner continues at once. The owner runs it anywhere; the blocking driver admits it as another root of the same run. The chunk planner spawns every chunk. The scan's stream runs each on a run and a session of its own, stepped on `spawn_cpu`, up to `concurrency` chunks per worker thread at once, merging their batches as they finish. A scan with fewer engine partitions than cores had lost the parallelism of one task per split, 2 to 4 times slower on selective queries with one partition; it is back, without a planner knowing about threads. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Hints were deduplicated by scanning the batch for each segment, which is quadratic in a chunk's segments: a wide projection over a filtered chunk announces hundreds. A set of the segments seen replaces the scan. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
…es not A spawned chunk runs on one task of the runtime that drives its run to the end and awaits only its reads, instead of a `spawn_cpu` task per step: a chunk's run waits for reads many times, and a hop at each wait was latency on its critical path that no other chunk's work filled. At a concurrency of one, chunks run on the scan's own run as further roots: the run's steps already interleave their work, and a run, a session and a task per chunk only added a hand-off per chunk, measured at 2 to 3 percent on TPC-H, for parallelism the engine provides. DataFusion sets the concurrency, unless configured, to the cores per partition, so a scan over one partition per core runs its chunks inline and a single-partition scan runs them beside each other. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
rustdoc rejects the explicit target as redundant now that `Run` is imported here. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Vr65ZwPZz4LPjkvDA6ZUuE Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
…-planners Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
…ist test #10533 added `nested_empty_list_roundtrip` calling `ListArray::try_new` without the `&mut ExecutionCtx` argument it now takes, which broke the `vortex-file` test build on develop. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BL868qtvZQwUKzuhNuWAS2
…-planners Picks up the ctutils bump for RUSTSEC-2026-0385. Conflicts with the Eval(Eval()) and Eval(pack()) reduce rules from #10732 keep this branch's ZonedDataRule and ExpressionLegacyStatsRule beside them, and the legacy stats pruning test now expects the zones plan to reduce to a Pack. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BL868qtvZQwUKzuhNuWAS2
The opener defaulted the V2 scan's concurrency to available cores divided by the engine's partitions, which is 1 at one partition per core. The scan multiplies concurrency by the core count again, so this never ran chunks inline as intended; it only cut the chunks in flight per partition from 16 to 4. Fewer concurrent chunks left the segment source less to coalesce: on TPC-H SF=1 q18 the scan made 2-3x as many reads and read about 25% more bytes. Leave the builder's default unless `scan_concurrency` is set. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KbaPHjo8XVLKjhGy9xoUMC
joseph-isaacs
added
changelog/performance
Claude
This branch has not been deployed
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
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.
Summary
Benchmark-only draft, not for merging. It is #10707 plus one commit (
451ce0b), so its SQL benchmark comments compare directly with that PR's. The question is whether the commit recovers the TPC-H SF=1 DataFusion hot-run regression reported on #10707 (+9.3%, worst on q19, q06 and q10). If it does, the commit should be cherry-picked ontoji/scan-v2-chain-2-planners.Changes
#10707 made the DataFusion opener default the V2 scan's concurrency to
available_parallelism / partitions, which is 1 at one partition per core. The V2 scan multiplies concurrency by the core count again, so this never ran chunks inline as its comment says. It only cut the chunks in flight per partition from 16 to 4. TPC-H runs V2 by default indatafusion-bench, so this affects the benchmark.With fewer chunks in flight, the segment source has fewer reads to coalesce. TPC-H SF=1 q18, measured locally:
The commit keeps the builder's default unless
scan_concurrencyis set and removes thepartitionsplumbing.Local wall-clock timing on a 4-core VM was too noisy to show a difference either way: rounds varied by ±25%, and the geomeans were within noise. That is why this needs the CI benchmarks.
Checks:
datafusion-benchbuilds with therelease_debugprofile with no warnings. Clippy and tests were not run.Generated by Claude Code