Skip to content

feat(layout): execute a plan into a stream, add a Query plan, and benchmark against V1 - #10419

Closed
joseph-isaacs wants to merge 18 commits into
ji/exec-5-list-pack-nodefrom
ji/exec-6-execute-stream
Closed

joseph-isaacs wants to merge 18 commits into
ji/exec-5-list-pack-nodefrom
ji/exec-6-execute-stream

Conversation

@joseph-isaacs

Copy link
Copy Markdown
Contributor

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

  • Streaming. execute runs a plan over a row range and yields arrays in row order.
  • Query plan. It prunes zones, evaluates the conjuncts one at a time under the rows the earlier ones kept, then projects. Conjunct results are folded to bits as they arrive. The dense threshold is configurable.
  • Planning. The optimizer memoizes shared subtrees. Arrays pass through nodes without copying. A Take keeps its values across executions.
  • Decode cache. A cache bounded by generation lets segments that span splits be decoded once.
  • Benchmarks. vortex-file/benches/query_exec.rs and query_v1.rs run 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.rs runs interleaved A/B comparisons and counts allocations and reads.
  • Design. vortex-layout/docs/pipeline-exec-design.md describes the push-based pipeline executor.

🤖 Generated with Claude Code

https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3


Generated by Claude Code

joseph-isaacs and others added 18 commits October 8, 2026 20:16
…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
@codspeed

codspeed Bot commented Oct 9, 2026 •

Copy link
Copy Markdown

Merging this PR will degrade performance by 35.12%

⚠️ Unknown Walltime execution environment detected

Using the Walltime instrument on standard Hosted Runners will lead to inconsistent data.

For the most accurate results, we recommend using CodSpeed Macro Runners: bare-metal machines fine-tuned for performance measurement consistency.

⚠️ 117 benchmarks spent significant time in system calls

System calls cannot be consistently instrumented, so they are not included in the measure, which understates the real cost. Please switch to the Walltime instrument to accurately measure system calls.

Measurement and system calls

❌ 1 regressed benchmark
✅ 2152 untouched benchmarks
🆕 310 new benchmarks
⏩ 381 skipped benchmarks1

Warning

Please fix the performance issues or acknowledge them on CodSpeed.

Performance Changes

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)

Open in CodSpeed

Footnotes

  1. 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. ↩

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants