Repository navigation
Fix and speed up the V2 scan's planner chain - #10789
Open
joseph-isaacs wants to merge 30 commits into
Open
joseph-isaacs wants to merge 30 commits into
joseph-isaacs wants to merge 30 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
…n IO session At one partition per core the opener passes a concurrency of one, but inline was gated on concurrency times the core count, so every partition ran a core's worth of chunks at once. Chunks admitted inline or spawned inside a chunk now each get their own IO session, since retiring a root clears its session's reads. Also logs how long preparing a scan takes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
Polling each read as it was registered let the segment source issue a prefetch before the announcements it could coalesce with existed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
…gn projection windows MaskScan drives a boolean plan and combines its pieces, taking a single piece without copying. Dense projections are cut on the file's row grid instead of from the trimmed chunk start, so writer chunks are not split between two morsels. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
An Eval rebuilding every root field of a Pack in order is the identity. Legacy stats only need converting to zones under a proof, which the expression rule already handles; a data read reduces straight to data. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
Each 64Ki-row window of a chunk was a morsel with a scan of its own, so a segment spanning several windows was decoded once per window. One morsel now runs one scan per chunk with a split per window: segments are decoded once, kernels run a window at a time, and a window's batch is yielded as soon as it has all its rows. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
…57ni Signed-off-by: Claude <noreply@anthropic.com> # Conflicts: # vortex-layout/src/scan/execution/filter.rs
joseph-isaacs
added
changelog/performance
Claude
github-actions
Bot
removed
the
action/bench-sql
Contributor
Polar Signals Profiling ResultsLatest Run
Previous Runs (43)
Powered by Polar Signals Cloud |
Contributor
Benchmarks: PolarSignals Profiling 📖Commits: PR datafusion / vortex-file-compressed / ns (0.990x ➖, 0↑ 0↓)
No file size changes detected. |
Contributor
Benchmarks: FineWeb NVMe 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.017x ➖, 0↑ 1↓)
datafusion / parquet / ns (1.001x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.003x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.997x ➖, 0↑ 0↓)
No file size changes detected. |
Contributor
Benchmarks: TPC-H SF=1 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.886x ✅, 9↑ 0↓)
datafusion / parquet / ns (0.999x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.989x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.997x ➖, 0↑ 0↓)
No file size changes detected. |
Contributor
Benchmarks: TPC-H SF=10 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.993x ➖, 1↑ 0↓)
datafusion / parquet / ns (0.999x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.998x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.996x ➖, 0↑ 0↓)
No file size changes detected. |
Contributor
Benchmarks: TPC-DS SF=1 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.001x ➖, 0↑ 1↓)
datafusion / parquet / ns (0.996x ➖, 0↑ 1↓)
duckdb / vortex-file-compressed / ns (1.001x ➖, 0↑ 1↓)
duckdb / parquet / ns (1.001x ➖, 1↑ 1↓)
No file size changes detected. |
Contributor
Benchmarks: Clickbench on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.992x ➖, 0↑ 0↓)
datafusion / parquet / ns (0.995x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.001x ➖, 0↑ 1↓)
duckdb / parquet / ns (1.003x ➖, 0↑ 0↓)
No file size changes detected. |
Contributor
Benchmarks: TPC-H SF=1 on S3 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.012x ➖, 1↑ 4↓)
datafusion / parquet / ns (0.948x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.925x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.966x ➖, 0↑ 0↓)
|
Merging this PR will degrade performance by 8.36%3 warnings
Warning Please fix the performance issues or acknowledge them on CodSpeed. Performance Changes
Tip Investigate this regression by commenting Comparing 
Footnotes
|
Concurrency 1 made a scan run its chunks inline, and the opener derived it from the engine's partition count. A file the engine does not split, such as a small dimension table, was then read by one partition with its chunks run one after another while other cores idled. The opener now divides the cores by the partitions that split this file, so a whole-file scan runs its chunks beside each other and a file split a partition per core still runs inline. A V2 scan's concurrency now counts its chunks in flight, one per worker thread by default, instead of per worker thread. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
joseph-isaacs
added
the
action/bench-sql
Claude
github-actions
Bot
removed
the
action/bench-sql
On a many-core machine DataFusion split SF1 lineitem into a partition per core, about 62K rows each, so every partition paid the V2 scan's fixed work (planning, zone proofs, compiling each conjunct) for one output window of rows. A V2 scan now caps the engine's target partitions by the files' row counts, so each partition reads at least 256Ki rows; the engine repartitions after the scan as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
joseph-isaacs
added
the
action/bench-sql
Claude
github-actions
Bot
removed
the
action/bench-sql
A filtered chunk was always 1Mi rows, so a partition of a few hundred thousand rows ran one chunk at a time however many chunks its concurrency allowed, leaving cores idle on CPU-heavy predicates. A filtered chunk now takes the scan's rows divided by its concurrency, rounded to the row grid and kept between 64Ki and 1Mi rows. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX Signed-off-by: Claude <noreply@anthropic.com>
joseph-isaacs
added
the
action/bench-sql
Claude
github-actions
Bot
removed
the
action/bench-sql
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_01GwhHqFVpQqPgQQkkK2y9kX
develop now carries #10707 as a squash; keep this branch's changes on top of it. 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_01GwhHqFVpQqPgQQkkK2y9kX
Resolve conflicts with #10708, which moved V2 scan driving out of ScanBuilder into ScanDrivers in vortex-datafusion and vortex-file, and renamed zone proofs to pruning plans. - Port per-root IO sessions and the inline-at-concurrency-one gate into both ScanDrivers. The DataFusion driver takes concurrency as the scan's total chunks in flight; the vortex-file driver keeps its per-worker-thread meaning. - Restore ScanBuilder::with_concurrency, used only to size filtered chunks, and set it from both drivers. - Keep the OnceCell scan cache, MaskScan and the is_zone_pruning dtype check under develop's pruning names. - Port dense_chunk_yields_a_batch_per_window onto PreparedScan; drop unordered_yields_before_the_split_finishes with the code develop removed. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJp2eSZRzHhv4cVCUNo6nV
joseph-isaacs
added
the
action/bench-sql
Claude
github-actions
Bot
removed
the
action/bench-sql
A partition keeps at least 256Ki rows, so a small, wide table such as TPC-H SF10's customer (1.5M rows) is read by 6 partitions instead of one per core. Each ran 4 projection chunks of 64Ki rows, so only 24 chunks decoded the table on 96 cores, and Q10 ran 19-24% slower than develop. A projection chunk is now cut like a filtered one: small enough that every chunk the scan runs at once has rows, down to one row block, and never larger than before. Also correct ProjectionMorsel's doc: each split decodes the segments it reads; a chunk's splits share the reads and dictionary values, not the decoded segments. 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_01GwhHqFVpQqPgQQkkK2y9kX
joseph-isaacs
added
the
action/bench-sql
Claude
github-actions
Bot
removed
the
action/bench-sql
The minimum rows per V2 partition cut TPC-H SF10's customer from one partition per core to 6 and part to 8. Every row of an unfiltered scan flows into the operators above it, which run a task per partition, so Q10 and Q14 ran 13-17% slower than develop on 96 cores however many chunks each partition ran at once. Forcing customer and part to one partition reproduces it locally: Q10 and Q14 at SF10 are 13% slower on 4 cores. The minimum now applies only to a filtered scan, whose partitions each prove and plan the filter and emit a fraction of the rows they read. Locally the SF10 slowdown is gone, and TPC-H SF1 split 96 ways keeps most of what the minimum saves: a geomean of 87 ms, against 83 ms with the minimum on every scan and 104 ms without it. 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_01GwhHqFVpQqPgQQkkK2y9kX
joseph-isaacs
added
the
action/bench-sql
Claude
github-actions
Bot
removed
the
action/bench-sql
Resolve vortex-file's ScanDriver against #10861, which maps batches on the driver's tasks: keep develop's map_fn plumbing and reapply the per-root IO sessions, the inline-at-concurrency-one gate and the scan builder's concurrency. 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_01GwhHqFVpQqPgQQkkK2y9kX
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
Fixes and perf work on the planner-driven V2 scan, found by dumping and timing real V2 scan plans for every TPC-H SF1 and SF10 query. The diff against
developis only this PR's changes.What the plans showed:
WHEREpredicate. A proof costs about 1 ms per partition. Join-time dynamic filters are not pushed into Vortex at all (Support datafusion dynamic filter expressions #4034), which is a separate pruning opportunity (for example Q18 over thel_orderkey-sortedlineitem).Changes
concurrency * cores <= 1, so each partition still ran a core's worth of chunks, about cores² chunks in flight overall.dense_chunk_yields_a_batch_per_window.OnceCells. The worst case drops from 14 ms to about 1 ms.MaskScan. A mask that is a single piece is no longer copied.Evals left after pushing a projection into aPack.is_zone_proof's boolean dtype before walking the expression.V2_MIN_ROWS_PER_PARTITION). An unfiltered scan keeps the engine's partitions, because every row it reads flows into the operators above it. Cappingcustomerandpartat SF10 made Q10 and Q14 13–17% slower; forcing them to one partition reproduces this locally. Test:v2_scan_keeps_partitions_unless_filtered.Benchmarks (CI, 96 cores,
6a4c019vsdevelop)DataFusion and DuckDB both run TPC-H on V2. The other suites run V1 and are neutral as expected.
DuckDB on Vortex is flat on TPC-H at both scales.
Checks run locally:
cargo clippy -p vortex-layout -p vortex-datafusion --all-targets --all-features -- -D warningscargo fmt --checkforvortex-layoutandvortex-datafusioncargo test -p vortex-layout --lib: 675 passed, also with the default concurrency forced to 96cargo test -p vortex-datafusion --lib: 774 passed🤖 Generated with Claude Code
https://claude.ai/code/session_01GwhHqFVpQqPgQQkkK2y9kX