Repository navigation
feat(layout): add push-based pipelines and the scan that runs them - #10391
Conversation
Merging this PR will degrade performance by 15.74%
|
| Mode | Benchmark | BASE |
HEAD |
Efficiency | |
|---|---|---|---|---|---|
| ❌ | Simulation | decode_primitives[i64, (1000, 32)] |
24.7 µs | 38.3 µs | -35.49% |
| ❌ | WallTime | threshold_neon[8] |
2.9 µs | 3.8 µs | -22.88% |
| ❌ | WallTime | threshold_neon[64] |
4.3 µs | 4.9 µs | -12.21% |
| ❌ | WallTime | threshold_neon[80] |
4.7 µs | 5.3 µs | -11.65% |
| ⚡ | WallTime | words_gather_dispatch_neon[65536] |
2.3 µs | 2.1 µs | +10.05% |
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-2-exec-node (b966c27) with develop (e277ccd)
Add `plan::exec`, a push-based executor interface for physical plans. A node is built by `PlanVTable::exec` for a range of its plan's rows and a selection over them. It has input ports, one output, and two entry points: `start`, called once to spawn children and request segments, and `compute`, called whenever something changed for the node and its readiness rule holds. A port is an ordered queue of arrays fed by one child plus a closed flag, owned by the graph and read by the node through `StepCx::input`. Every node emits its selected rows in row order as non-empty arrays and returns `Done` to close its own output; there are no close events, no empty arrays, and no row ranges on the data. `Ready` is the readiness rule: `Any` for nodes that stream input through, `Closed(ports)` for nodes that need one input whole before using the others, as a dictionary needs its values before its codes, and `AllClosed` for nodes that need everything. Chains of single-port `Any` nodes are pipelines; the other rules are the barriers between them. `ExecGraph` is the flat node arena. It wires ports, routes segment deliveries to the leaves that requested them, runs ready nodes last-in first-out so data drains toward the root, and hands its owner the root's arrays and batches of `IoRequest`s. IO may complete in any order; a node whose children are backed by independent reads holds the early ones in their ports and emits the ordered prefix. `plan::exec::synthetic` has plans whose nodes are built by hand, so the graph can be exercised without layouts. The scheduling tests drive sources that cut their rows into pieces, request in-memory segments, and yield, under a probe node that records when its readiness rule let it run. 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
75184f0 to
8e49f35
Compare
Replaces the `ExecNode` graph with `plan::pipeline`, the executor a plan now runs on: - `Operator` and `Source`: one `compute` method returning a `Step` (`More`, `Last`, `Consumed`, `Blocked`, `Finished`). Only a source blocks, on a read, an empty inlet, or a full outlet. - Ports with one writer and one reader, in an arena the scan owns; the reading source declares each inlet's capacity, and `Inlet` exposes `peek_mut`, `take`, and `closed`. - A driver that pushes each batch through a pipeline's stages into its outlets, and a scheduler that wakes a blocked pipeline only when its condition clears, checking only the ports a run touched. - `Scan`, which compiles a plan per split, runs several splits at once, and returns reads to its owner as `Turn::Read`. - `PlanVTable::compile` and `PlanVTable::reach` replace `PlanVTable::exec`: the hooks a plan implements to compile into a chain and to report the segments it reads. Synthetic plans exercise the runtime without layouts. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
`Cx::spawn` let a source ask for a plan to be compiled into a new inlet while the scan ran. Every plan is now compiled up front: what a stage reads is decided by its plan and its masks before it runs, so the runtime never plans mid-run. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…et a reader waits on Ports allocate their buffer once, at their capacity, and keep it when a freed slot is reused. The default capacity rises from 2 to 8 batches, which halves scheduler round trips on narrow pipelines. A stage's output port is bounded like any other, so it needs no growth either; only a share's port, whose reader may be compiled later, can grow past the default. A reader blocked on inlet i is woken only by a batch or close on that inlet, and the inlets a run touched are tracked with a flag per inlet, so a source reading thousands of inlets pays per inlet it read, not per wake. Pipelines and ports live in a slab that never moves its entries. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
A source could ask for one segment at a time: the driver issued its read and did not compute the source until the bytes came back, so a source reading several segments waited for each in turn. Now the driver asks the source for segments until it has none to ask for and issues every read at once. The source is computed as each arrives and takes the bytes tagged with their segment through Cx::take_segment, oldest arrival first. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
The scan's docs linked QueryPlan, which a later change adds, and failed the docs build with broken intra-doc links denied. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Ports are named by their ends: a pipeline's sink pushes into its outlets
and another pipeline's source reads them as inlets (SourceEnd replaces
Reader). Source and Stage replace Operator and Input: a source returns
Step::{Batch, Blocked, Finished}, a stage pushes one batch and flushes on
finish, and the driver owns full outlets. ConcatSource::new(1) replaces
PortSource, and OnceSource moves to its own module.
The driver now:
- cancels the reads a retiring pipeline left in flight, so their bytes
are dropped instead of reaching a pipeline that reuses its id;
- reports Done once no split is live, whatever reads are left over;
- finishes stages one at a time behind the outlets' room check, so a
flush never overfills a port, and closes at once when nothing is left.
Debug builds validate the graph after every compile (port ends agree,
sources read the inlets they declare, every port has a reader, fan-out
is unbounded so no diamond can deadlock) and each run's report (a source
waits only on an open inlet with room, or on a read in flight, and
finishes only once its inlets close). "Scan is stuck" names the port and
sink each waiting source depends on.
Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…only Queue::unbounded and Queue::is_free, and the test sources that break the scheduling contract, are used only under debug_assertions, so release builds (cargo check --release in CI) rejected them as dead code. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
… Input Stage::push and Stage::finish become one Operator::compute, handed Input::Batch for each batch and Input::End once, last. The driver still hands End one operator at a time behind the outlets' room check, and an operator still never blocks. Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Source is now one method, ready(), returning Step::{Batch, Read,
Blocked, Finished}. A source asks for a segment by returning
Step::Read; the driver issues the read once the run ends and asks the
source again at once, so it can still have several reads in flight
before it answers Blocked::Io. Bytes already there, as of a segment read
ahead, run the source again at once.
The inlet count comes from the chain, and capacities are the joining
plan's: Compiler::join holds DEFAULT_CAPACITY batches per inlet, and
Compiler::join_with takes a capacity per inlet.
Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
…uations Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Adds the pipeline runtime and plan hooks so an in-memory
PlanRefcan execute over row ranges. Tracking: #10349. Concrete plan operators follow in #10392 and #10393; #10389 is independent.A pipeline has one
Source: Operator, followed by zero or more operators.Source::readychecks inlet/IO readiness and queues segment reads throughCx::request_read, returningReadyorBlocked. Both newly queued and outstanding reads useBlocked(Io); the driver issues queued requests before waiting for bytes. Decoding and transformation stay in non-blockingOperator::compute.computeacceptsInput::{None, Batch, End}and returnsStep::{End, More, Final}:More(batch)produces a batch and requests another compute call withNone, without new input or another source readiness check.Final(batch)produces the last batch for this input. A stage waits for its next input; a source checks readiness again.Endproduces no output for the current input. It finishes a source, or finishes a stage when draining upstream'sInput::End.The driver drains downstream continuations first, preserving their state across outlet backpressure. It delivers
Input::Endonce per stage and drains all flush output before closing outlets.Ports have one writer and reader.
Compiler::join_withsets their capacities. The scheduler wakes pipelines when their blocked condition clears, tracks touched inlets in a compact set, and reports changed inlet indices to source readiness.Scan::stepreturns read requests and output;Scan::deliversupplies bytes tagged with their segment ID. Retiring a pipeline cancels outstanding reads.Synthetic tests cover input/output order, nested continuations, multi-batch flushing, bounded outlets, batched read requests, and readiness that consumes an inlet before requesting IO.
Physical plans and their optimizer live under
vortex_layout::plan::physical. The siblingoperatormodule definesOperator,Source, and the execution context;operatorscontains concrete implementations;pipelineowns compilation, scheduling, and ports.Validation:
cargo +nightly-2026-09-10 fmt -p vortex-layout; scopedcargo clippy -p vortex-layout --all-targets --all-features -- -D warnings; targeted physical-plan tests (52 passed).