Schedule generic planners and morsels, register IO batches, route completions, and clean up retired work. Add a separate blocking byte-source adapter and scripted driver tests. Existing scan execution and IO adapters are unchanged.

Signed-off-by: "Joe Isaacs" <[email protected]>

Add separate FooterOpen and FooterPrune stages and a caller-supplied continuation. Exercise real footer IO and conservative file-statistics pruning with a diagnostic range morsel confined to test support. Keep existing file and scan entry points unchanged.

Signed-off-by: "Joe Isaacs" <[email protected]>

Import-order only; output of cargo +nightly-2026-09-10 fmt -p vortex-file.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Fable 5.1 <[email protected]>

Add a filter-and-project stage that emits one morsel per natural split
and a PollingSegmentSource behind the unchanged LayoutReader. The morsel
holds the reader's future, polls it once per compute with a no-op
waker, and reports the segments that poll missed as NeedsIO keyed by
segment id. Pending leaves store the waker they were polled with and
deliver() wakes them, so the reader's combinators re-poll only what
changed and nothing above the awaits runs twice. A poll that is pending
without a recorded miss is an error, so a reader awaiting anything but
a segment cannot park forever. plan_scan composes footer open, pruning,
and this stage for one file.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Fable 5.1 <[email protected]>

… vortex-layout

SplitMorsel and PollingSegmentSource only need a LayoutReader, a
SegmentSource, and the planning protocol, all of which live in or below
vortex-layout. Moving them there lets the layout scan path drive them.

The polling source takes SegmentLocation values instead of the file
footer's SegmentSpec. The FilterProject planner stays in vortex-file,
where it reads the footer, and builds morsels through SplitMorsel::new.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Add vortex_layout::scan::v2 with prepare and into_stream, drop-in
replacements for ScanBuilder::prepare and ScanBuilder::into_stream that
return the same types. RepeatedScanV2 starts as a copy of RepeatedScan so
its split execution can be replaced without touching the default path.

DuckDB's reader_initialize and DataFusion's VortexOpener select the
replacement when VORTEX_SCAN_V2=1; nothing changes otherwise. The builder
exposes its configuration to the replacement through a crate-private
into_parts.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

RepeatedScanV2 now turns each split into a SplitMorsel over a fresh
PollingSegmentSource and layout reader, and drives it to completion: the
protocol reports which segments the split needs, and the file's own
segment source fetches them by id. Each split gets its own polling
source because the source's miss list is not safe to share between
morsels that compute concurrently.

SplitMorsel takes an initial mask, so scan selections and filterless
limits reach it the same way they reach split_exec. Row-index
expressions wrap the per-split reader as the default path does.

v2::prepare and v2::into_stream take a ScanFile: the root layout, the
segment locations, and the segment source. vortex_file::planning::scan_file
builds one from a VortexFile. DataFusion's opener already holds the
file; DuckDB's OpenFileReader now keeps it.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Port the uncommitted ExecGraph work from ji/plan-exec unchanged. A
PlanRef becomes an arena of exec nodes for one row range and selection;
SegmentScan, Concat, and Pack nodes push row-tagged pieces to their
parents and publish segment reads for the owner to fulfil.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Replace ExecNode's start, is_ready, step, on_input, and on_io with one
compute method. The graph buffers child pieces, closed ports, and IO
results in a per-node inbox, and the node takes them from its StepCx.
compute returns whether the node waits for input, wants to run again,
or is done. The first compute still runs when a node is spawned, so the
first graph compute still publishes every leaf's read in one batch.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

EvalNode runs its child over the same rows and selection and applies the
bound expression to each piece, so expressions only see selected rows.
RowIdxNode emits the global index of every selected row; row-index plans
sit in the root row domain, so the graph carries the row offset of its
first plan row and ExecGraph::try_new takes it.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Split one split's work into separate protocol stages. FilterPlanner runs
the filter plan over the split's rows and incoming mask and narrows the
mask to the rows whose value is true; if any survive, it hands them to
a Next, by default a ProjectionPlanner. ProjectionPlanner emits a
ProjectionMorsel, which runs the projection plan over the selected rows
and returns them as one batch in row order. plan_split composes them,
going straight to projection when there is no filter.

Both stages execute physical plans through ExecGraph, wrapped in a
ProtocolGraph that publishes the graph's segment reads as byte-range
Fetch requests and routes deliveries back, realigned to each segment's
alignment. Neither calls a LayoutReader.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

…on planners

v2::prepare lowers the file's layout into a physical plan once, lowering
zoned layouts to their data, and plans and optimizes the filter and the
projection over it, including row-index expressions. Each split then
runs plan_split on the planning driver instead of polling a
LayoutReader: the filter and projection are separate stages executed
through the exec graph, and the builder's reader only chooses the
splits.

The driver blocks while waiting for reads, so it runs on the runtime's
blocking pool, and a SegmentIoSource serves its byte-range requests
from the file's segment source on the runtime. This needs a runtime
whose blocking work runs off its async threads, as Tokio and the
current-thread runtime do; the V2 tests therefore run on Tokio.

Dictionary layouts lower to Take, which has no exec node yet, so V2
fails on them explicitly.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

TakeNode runs the codes over the selection and the values over their
whole domain, holds codes pieces until every value has arrived, and
emits each as a DictArray over the joined values. With it, VORTEX_SCAN_V2
reads dictionary-encoded columns instead of failing.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Add a Filter plan operator. It holds no predicate: it keeps the rows of
its child selected by the mask it is executed with, and its child
produces every row of the same domain. Flat layouts now lower to a
Filter over their SegmentScan, in both the plan fixture lowering and
the V2 file lowering, so every parent still receives only selected rows.

At execution a filtered scan and a bare scan are one SegmentScanNode.
The node is always given a selection saying which rows its parent cares
about, which it may use or ignore, and holds an optional filter: a
Filter(SegmentScan) passes the selection as the filter and gets the
selected rows, and a bare SegmentScan gets every row of its range.
Other children of Filter have no exec implementation yet.

Plan tests and snapshots now expect the Filter above each segment scan;
a new exec test checks that a bare scan is dense and a filtered scan
keeps only the selection.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Each split's driver blocks while it waits for reads, and V2 ran it with
spawn_blocking. ClickBench under DataFusion hung: more than 512 splits
filled Tokio's blocking pool with waiting drivers, and the reads they
waited for need that same pool, because object_store's local
filesystem reads through spawn_blocking.

Drivers now run on a small pool of threads owned by V2, sized to four
per core. The driver thread never starts a read: SegmentIoSource sends
each request to the split's own future, which starts and polls the
reads on whatever runtime drives the scan and sends completions back.
Waiting drivers therefore never hold a thread that a read needs, and no
read is spawned from outside a runtime.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Signed-off-by: Joe Isaacs <[email protected]>

The zoned plan tests were written before flat layouts lowered to a
Filter over their segment scan. Update their snapshots for that shape;
nothing else in the trees changes.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

A pruning Zoned plan now runs as a ZonePruneNode. It reads the column's
zone table, evaluates the pruning proof with ZoneMap::prune, and emits
for every selected row whether the proof holds for the row's zone. The
per-zone result is cached on the plan, so later executions of the same
plan only expand it to their rows and issue no reads. A data Zoned plan
runs as its data child.

Zoned plans now carry the zone map's aggregate functions, and pruning
plans remember the column dtype their proof is bound to, which the zone
map needs.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Add a pruning stage in front of the filter. v2::prepare falsifies the
filter into a predicate over statistics and applies it to a copy of the
file's plan that keeps its zoned layouts. The optimizer pushes the
predicate down to the zoned columns it reads and rewrites it into
pruning plans over their zone tables. If any part of the proof would
still read column data, the scan skips pruning.

Each split then runs a pruning FilterPlanner, which drops the rows whose
zones prove the filter false, before the filter and projection stages.
Rows it drops are never read. The V2 lowering gains a mode that keeps
zoned layouts for the pruning copy; data plans still lower them to their
data.

A new test writes a zoned column and checks that V2 returns what the
default path returns while never reading the pruned chunk.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Each V2 split handed its planning driver to a shared pool of driver
threads and parked the awaiting thread until the driver finished. The
hand-off and the pool's shared job queue cost CPU on every split:
profiles of DuckDB ClickBench q14 showed the DuckDB threads parked in
block_on and 56 pool threads contending on the queue's mutex.

The planning Driver gains a non-blocking Run: Driver::start returns it,
Run::advance runs ready work until everything waits for IO, and
Run::complete delivers a completion. Driver::run is now a blocking loop
over the same Run, so its behaviour is unchanged. V2's split future
advances its Run inline and awaits the next read itself; the
SegmentIoSource starts each read when it is submitted and polls them
without blocking. The driver pool and its channels are gone.

The Run is not Send, since the protocol allows thread-local planners.
The split wraps it in a type asserting Send, justified by static
assertions that the only planners plan_split builds are Send, which
needs ExecNode to be Send.

Also fixes a pre-existing clippy absolute_paths error in the driver
tests.

Interleaved V1/V2 medians, V2/V1 ratio before -> after:
DuckDB ClickBench q13 1.145 -> 1.051, q14 1.181 -> 1.138;
DataFusion TPC-H SF1 1.214 -> 1.116; DuckDB TPC-H SF1 0.989 -> 0.963.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

A TakeNode ran its values subtree over the whole dictionary on every
execution, so each V2 split, and both the filter and the projection of
a split, read, decoded and evaluated the same dictionary again. The
layout reader (V1) does this once per file. Profiles of DuckDB
ClickBench q14 showed two costs: the filter's pushed-down predicate ran
over the SearchPhrase dictionary for every split, and each split's
projection handed DuckDB a new values array, so the exporter's
per-dictionary cache never hit and it re-exported the dictionary each
time.

The take plan now keeps the joined values in its plan data, shared by
every execution of the plan and reset when the plan is rebuilt with new
children. Later executions skip the values subtree entirely. Boolean
values, which only arise from a predicate pushed onto the dictionary,
are evaluated before they are kept, so the predicate runs once.

Interleaved V1/V2 medians, V2/V1 ratio before -> after:
DuckDB ClickBench q14 1.138 -> 1.075, q13 1.051 -> 1.019.
V2 instructions per run: q14 99.0G -> 91.6G, q13 152.4G -> 140.9G.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

The take plan's cached dictionary values were plain arrays, and a
pushed-down predicate was evaluated eagerly before caching. Splits that
start together, as a file's splits do under DuckDB, all miss the cache,
so each evaluated the predicate itself, and every consumer that
canonicalized the values did so again: DuckDB's exporter races on its
per-dictionary cache, and each miss decompressed the FSST values anew.

The values are now kept as a SharedArray, as the layout reader keeps
them, and the predicate is left lazy. Whichever consumer canonicalizes
them first does so once for all, and concurrent consumers wait for it.

Interleaved V1/V2 medians, V2/V1 ratio before -> after:
DuckDB ClickBench q14 1.075 -> 1.064, q13 1.019 -> 0.986.
V2 instructions per run: q14 91.6G -> 86.3G, q13 140.9G -> 133.5G.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

FilterPlanner wrapped the filter's pieces in a ChunkedArray even when
there was only one, so turning the filter result into a mask went
through the generic builder (Dict::append_to_builder on ClickBench q14)
instead of the piece's own kernels. A lone piece is now executed
directly, as ProjectionMorsel already does for its output.

V2 instructions per run: DuckDB ClickBench q14 86.3G -> 85.9G,
q13 133.5G -> 132.9G.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

The layout reader (V1) builds every split's futures when the scan is
executed, which registers their segments with the file's segment
source. The source coalesces registered segments into any nearby read,
so V1 serves TPC-H q06 in 31 reads. V2 only asked for a segment when a
split's graph needed it, so nothing was registered ahead and the same
query took 184 reads. Under DataFusion each local read is a
spawn_blocking, and the extra reads showed up as contention on Tokio's
blocking pool, the largest part of V2's extra CPU on q06.

When V2 builds its split tasks it now walks the filter and projection
plans over each split's rows, registers the segments they are likely to
read, and holds the registrations until the split finishes. Fetches go
through the source as before; while a registration is alive, a shared
segment source serves them through it, so a column read by both the
filter and the projection is also read once. The walk is a hint: it
follows concat, take, pack, filter and eval plans and stops elsewhere.

q06 on DataFusion: 184 -> 63 reads, 47.1 MB -> 43.1 MB.
Interleaved V1/V2 medians, V2/V1 ratio before -> after:
DataFusion TPC-H SF1 1.131 -> 1.012 (V2 CPU per run 20.4 s -> 17.9 s,
V1 15.6 s); DuckDB ClickBench q14 1.048 -> 1.024.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

V2 now registers every segment a split may read before the split runs,
so the source can coalesce them. Registration alone reads nothing, so
the test records the segments whose reads are awaited rather than every
request: the pruned chunk is registered but never read.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

FilterPlanner now evaluates a filter one conjunct at a time, as the
default scan does. v2::prepare splits the filter with FilterExpr and
plans each conjunct on its own. Each split's planner runs the next
conjunct over the rows the previous ones kept, stops as soon as none
remain, and reports every conjunct's selectivity to the scan's shared
FilterExpr, which reorders them most selective first. A conjunct only
reads and evaluates the rows still selected.

The conjuncts travel as FilterPlans, which the pruning stage also uses
with its single plan and no ordering. Because one planner now runs
several graphs, ProtocolGraph numbers its requests from a base the
owner carries forward, so request ids never repeat within the planner.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

Evaluating two conjuncts separately reads and decodes their shared
columns twice. V2 now merges the conjuncts FilterExpr splits out back
into one per group of conjuncts that read a common field, transitively,
before planning them, following the shared-field grouping in #9877.
Groups and their members keep their input order.

FilterExpr gains a crate-private from_conjuncts constructor for
conjuncts that are already split; FilterExpr::new calls it, so the
default path is unchanged.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

A split runs up to three kinds of graph over the same rows: pruning,
one per filter conjunct, and projection. Each decoded every segment it
read, so a column read by the filter and the projection, or by two
conjuncts, was fetched and decoded again.

ExecGraph now takes a DecodeCache, and plan_split gives each split one
shared by all its graphs. A SegmentScan that finds its segment there
neither reads nor decodes it; otherwise it stores the whole decoded
segment before slicing and filtering it to its own rows. The cache
lives as long as the split.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

ZonePruneNode::expand checked the selection and the zone result one row
at a time, and was the largest single cost V2 added on TPC-H SF1 under
DuckDB. Every row of a zone shares the zone's value, so fill the rows a
zone at a time and apply the selection once with a filter.

Signed-off-by: "Joe Isaacs" <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>