Repository navigation
feat(layout): execute a plan into a stream, add a Query plan, and benchmark against V1 - #10419
joseph-isaacs wants to merge 18 commits into
Conversation
…he graph Add `plan::exec::execute`, which runs a plan over a row range and selection against a `SegmentSource` and yields the selected rows as a `Stream` of arrays. It drives an `ExecGraph`, issues reads as the graph discovers them, and delivers them as they complete in any order. The graph emits in row order, so the stream is a plain drain. Add the `exec_graph` benchmark. `split` lowers a struct of chunked columns written to in-memory segments and drives a graph per iteration, answering reads at once, over full, sparse, and single-chunk selections. `scheduling` runs the same shapes over hand-built sources that emit tiny arrays, with and without a read round trip, so the per-node cost of the graph itself is what is measured. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…d-project query Add `scan_v1`, which runs `ScanBuilder` over a `LayoutReader` on a struct of eight chunked `i64` columns in in-memory segments: a predicate over two columns keeping about a tenth of the rows, a projection of three others, one split per chunk, drained on a single-thread runtime. Add `query` to `exec_graph`, which runs the same fixture and query on the exec graph: per chunk, the optimized filter plan over every row to a mask, then the optimized projection plan over the kept rows, reads answered from memory as they are published. The two benchmarks share nothing but build the same data from the same function, so only the executor differs. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…ke file Lower zoned and legacy stats layouts to their data child, so a file written by the default strategy can be planned. Zone statistics only drive pruning, which plans do not do yet. Add `query_v1` and `query_exec` under `vortex-file/benches`. Each writes the same lineitem-like table (2M rows: sorted keys, uniform integers and floats, ship dates clustered by row, dictionary strings, a high cardinality string) through the default write strategy into memory, opens it, and runs three queries over the file's natural splits: a TPC-H Q6 style date, discount, and quantity filter projecting two columns; a dictionary string equality with a quantity bound projecting three; and a filterless projection of three. `query_v1` drains a `ScanBuilder`; `query_exec` runs the optimized filter plan per split to a mask and the optimized projection plan under it, through the file's own segment source. Both assert the row count the plain columns give. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
… zone maps The first comparison measured planner features as much as executors: the V1 scan narrows the mask conjunct by conjunct and prunes zones, and its reader decodes a segment once per split, while the exec graph evaluates the whole predicate at once, does not prune, and decodes once per query. Make the comparison isolate the infrastructure: - `WriteStrategyBuilder::with_zone_maps(false)` writes a file without zone maps, so neither side has anything to prune or any statistics to evaluate. Both benches run every query on a zoned and a plain file. - `vortex_layout::scan::filter` is public, so the exec bench can run the V1 scan's own `FilterExpr` conjunct scheduler. - The exec bench runs each query under three algorithms: the whole predicate as one plan; V1's algorithm on the graph, conjuncts in the order `FilterExpr` chooses, each narrowing the mask exactly as the V1 flat reader does; and that again with segments decoded once per graph, as the V1 reader decodes them once per split. With the last on the plain file, both executors do the same reads, decodes, comparisons, and filters, and what remains is how they drive the work. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
The values of a Take run over their whole domain whatever rows the take is executed with, so they depend only on the plan. Keep the joined values on the plan the first time any execution produces them, as a `SharedArray` so they are canonicalized once, and have later executions spawn only the codes. This is what the V1 dictionary reader does with its cached values and its per-expression evaluations over them: a predicate the optimizer pushed onto the values is the values child of its own Take plan, so it is cached with it. Values evaluated with a dynamic comparison are not kept, since the engine updates the comparison between executions. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…ariants The conjunct mirror of the V1 scan wrapped a split's pieces in a chunked array before filtering, which does not push the filter into the lazy predicate the way the V1 flat reader's per-chunk evaluation does. Evaluate each piece on its own, as the V1 chunked reader evaluates each chunk, and concatenate the masks. Add `plain-2k` and `plain-256` file variants, written with small row blocks and no byte coalescing, so a split is one small block of every column. The work per split shrinks until each executor's own cost per split and per node is what is measured, on the same layouts and plans. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
`query_profile` builds one file variant of the lineitem-like table at a quarter of the benchmark's rows and runs one query on one executor, so a profiler can be pointed at `run_v1` or `run_exec` alone. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…rojection as one graph A Query plans each conjunct of a filter and the projection over one source. Its node evaluates the conjuncts one at a time in the order the shared FilterExpr scheduler prefers, each under the rows the earlier ones kept, then streams the projection under the rows that passed them all, so one graph per split does what the V1 scan's split task does. The exec benchmark and the profiling example gain a `Query` algorithm running one such graph per split, and the example can print the optimized plan and the plan-build time. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…ut copying The Pack expression rule now drops a residual that only re-packs the rewritten fields in order, so a projection of plain columns runs the Pack directly instead of applying an identity `pack` to every piece. Expressions stay above a Concat whose chunks are scans, since the exec node applies them to each chunk's rows as they arrive anyway; pushing them down only multiplied the plan nodes by the chunk count. The optimizer memoizes the subtrees a query shares between its conjuncts and its projection, so each is optimized once. Pass-through nodes pop arrays from their ports one at a time instead of draining them into a vector, and a graph's arena is sized for a typical split up front. The exec benchmark and the profiling example now plan the `Query` algorithm inside the timed run, over the file's layout lowered once per file, as a V1 scan prepares every scan over the file's shared reader tree. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…eration A `Zoned` plan now stands for a zoned layout: it produces its data child's rows and keeps the zone table beside them. The optimizer pushes an expression over it into the data, keeping a boolean one as the plan's predicate, and a Query turns each such predicate into a pruning plan. A ZonePrune node tells, for every row, whether the row's zone may hold a match, from the zone table alone: the table is read once per zoned plan and the zones each predicate prunes are kept on it, shared by every plan derived from it, as the default scan's zoned reader keeps them. A Query first narrows its rows with each conjunct's pruning plan, then evaluates the conjuncts. A conjunct over a dense mask now runs over the chunks holding a selected row, rather than over every row, so the chunks the zones pruned are never read while whole chunks are still evaluated and intersected rather than filtered. A DecodeCache can be advanced a generation between the row ranges an owner runs in order; a segment no graph of the last two generations used is dropped, so a segment spanning consecutive splits is decoded once without the cache holding a whole file. The exec benchmark and the profiling example gain a `QueryStreaming` algorithm that drives one such cache per run. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
… schemes The benchmark sessions had no compression schemes registered, so the default strategy wrote every column uncompressed. Both benchmarks and the profiling example now register them, as the default session does, so the files carry the encodings a real file would and both executors decode them. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
The exec benchmark and the profiling example now run the `Query` algorithms with up to four splits per core in flight, as the V1 scan buffers its split tasks, so the reads of later splits overlap the compute of earlier ones. Two file variants are written to disk and read through the file IO path. A decode cache takes a window of generations to cover the splits in flight, a pruning plan keeps its proof rather than looking it up per split, and a segment scan whose bytes arrive after another graph decoded the same segment reuses that array. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
A file in memory answers every read at once, so holding several splits' graphs alive only costs memory; the exec benchmark and the profiling example now pipeline splits only for the on-disk variants, where reads overlap compute. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
… arrive The Query node executed a conjunct's arrays once its child had closed, joining them into one lazy array first. It now executes each array to bits as it arrives, so no lazy result and none of the decoded segments it refers to outlive their piece. The profiling example counts allocations and segment reads for both executors, writes its on-disk file to a fixed path, and takes the split rule and the splits in flight from the environment. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…hmarks Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…e runs in the example The selected fraction at which a Query runs a conjunct over whole chunks rather than the selected rows is now a plan option, defaulting to the scan's 0.2. The profiling example alternates the two executors for a number of runs and reports their medians, so drift between runs affects both alike, and takes the threshold from the environment. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Design document for the push-based pipeline executor that replaces the node-based plan executor: traits, ports, driver, scheduler, masks, the Query stage planner, the Share node, scan integration, open decisions, and an implementation order. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Merging this PR will degrade performance by 35.12%
|
| Mode | Benchmark | BASE |
HEAD |
Efficiency | |
|---|---|---|---|---|---|
| ❌ | Simulation | decode_primitives[i64, (1000, 8)] |
24.6 µs | 37.9 µs | -35.12% |
| 🆕 | Simulation | exec[0, dict@plain-256@64k] |
N/A | 1.3 s | N/A |
| 🆕 | Simulation | exec[0, dict@plain-256@layout] |
N/A | 1.6 s | N/A |
| 🆕 | Simulation | exec[0, dict@plain-2k@64k] |
N/A | 473 ms | N/A |
| 🆕 | Simulation | exec[0, dict@plain-2k@layout] |
N/A | 512.4 ms | N/A |
| 🆕 |
Simulation | exec[0, dict@plain-disk@64k] |
N/A | 376.7 ms | N/A |
| 🆕 |
Simulation | exec[0, dict@plain-disk@layout] |
N/A | 397.6 ms | N/A |
| 🆕 | Simulation | exec[0, dict@plain@64k] |
N/A | 370.8 ms | N/A |
| 🆕 | Simulation | exec[0, dict@plain@layout] |
N/A | 387.9 ms | N/A |
| 🆕 |
Simulation | exec[0, dict@zoned-disk@64k] |
N/A | 376.9 ms | N/A |
| 🆕 |
Simulation | exec[0, dict@zoned-disk@layout] |
N/A | 399.9 ms | N/A |
| 🆕 | Simulation | exec[0, dict@zoned@64k] |
N/A | 371 ms | N/A |
| 🆕 |
Simulation | exec[0, dict@zoned@layout] |
N/A | 390.3 ms | N/A |
| 🆕 | Simulation | exec[0, project@plain-256@64k] |
N/A | 517.1 ms | N/A |
| 🆕 | Simulation | exec[0, project@plain-256@layout] |
N/A | 520.6 ms | N/A |
| 🆕 | Simulation | exec[0, project@plain-2k@64k] |
N/A | 162.9 ms | N/A |
| 🆕 | Simulation | exec[0, project@plain-2k@layout] |
N/A | 165.3 ms | N/A |
| 🆕 |
Simulation | exec[0, project@plain-disk@64k] |
N/A | 115.7 ms | N/A |
| 🆕 |
Simulation | exec[0, project@plain-disk@layout] |
N/A | 115.7 ms | N/A |
| 🆕 | Simulation | exec[0, project@plain@64k] |
N/A | 113.7 ms | N/A |
| ... | ... | ... | ... | ... | ... |
ℹ️ Only the first 20 benchmarks are displayed. Go to the app to view all benchmarks.
Tip
Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.
Comparing ji/exec-6-execute-stream (6eb349c) with ji/exec-5-list-pack-node (152edd4)
Footnotes
-
381 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩
Summary
Part of the exec plan stack extracted from #10349. Stacked on #10418.
This PR finishes the node-based executor and measures it against the V1 scan (
ScanBuilder/LayoutReader) like for like. It also adds the design document for the pipeline executor that the next PR implements.Changes
executeruns a plan over a row range and yields arrays in row order.Takekeeps its values across executions.vortex-file/benches/query_exec.rsandquery_v1.rsrun on a lineitem-like file in several block-size and on-disk variants, split by layout or by fixed row ranges.vortex-file/examples/query_profile.rsruns interleaved A/B comparisons and counts allocations and reads.vortex-layout/docs/pipeline-exec-design.mddescribes the push-based pipeline executor.🤖 Generated with Claude Code
https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Generated by Claude Code