Repository navigation
Plan a V2 scan as a chain of planners from one prepared scan - #10694
joseph-isaacs wants to merge 41 commits into
Conversation
Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> (cherry picked from commit 6e27e30)
The V2 opener gave every partition its own VortexFile, so each partition read and decoded chunks straddling its boundaries and the file's metadata segments on its own. V1 avoids this through the shared layout reader. Keep a weak per-path map of V2 segment sources beside `layout_readers` so concurrent partitions read through one source. TPC-H SF1 q4 went from 196 segment reads (52.4 MB) to 115 (34.8 MB) per query, matching V1. Hot SF10 geomean against V1 moved from 1.01 to about 0.96 on this machine. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> (cherry picked from commit bec59fe)
Prefetch needs the segments a stage will read before the stage runs. Compiling already registers every segment scan and does no IO, so `Scan::segments` compiles a plan over rows under a mask and lists them, and the per-plan `reach` walk that re-derived the same answer goes, with the `Reach` coordinates it carried. `Shares` used the walk to decode a segment once when several splits or readers of one scan read it, and goes with it. The V2 executor runs one split per scan and shares reads in its IO session, so a scan's splits now each read the segments they overlap, and a plan reading one segment through two paths reads it through each. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
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>
Merging this PR will regress 2 benchmarks4 warnings, 1 note
Warning Please fix the performance issues or acknowledge them on CodSpeed. Performance Changes
Tip Investigate this regression by commenting Comparing Footnotes
|
Polar Signals Profiling ResultsLatest Run
Previous Runs (14)
Powered by Polar Signals Cloud |
Benchmarks: PolarSignals Profiling 📖Commits: PR datafusion / vortex-file-compressed / ns (1.010x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: FineWeb NVMe 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.957x ➖, 2↑ 0↓)
datafusion / parquet / ns (0.987x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.003x ➖, 0↑ 0↓)
duckdb / parquet / ns (1.001x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: TPC-H SF=1 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.120x ❌, 0↑ 10↓)
datafusion / parquet / ns (1.002x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.003x ➖, 0↑ 0↓)
duckdb / parquet / ns (1.002x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: TPC-H SF=10 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.084x ➖, 0↑ 7↓)
datafusion / parquet / ns (1.000x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.999x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.990x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: Clickbench on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.000x ➖, 0↑ 0↓)
datafusion / parquet / ns (0.994x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.992x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.991x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: TPC-DS SF=1 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.005x ➖, 0↑ 1↓)
datafusion / parquet / ns (0.995x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.994x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.994x ➖, 0↑ 0↓)
No file size changes detected. |
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>
Benchmarks: TPC-H SF=1 on S3 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.799x ➖, 5↑ 1↓)
datafusion / parquet / ns (1.073x ➖, 0↑ 3↓)
duckdb / vortex-file-compressed / ns (1.162x ➖, 0↑ 4↓)
duckdb / parquet / ns (1.048x ➖, 1↑ 3↓)
|
…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>
A planner says what it needs, compute or reads. Where either runs is an engine's decision, so this crate stops driving: `ScanBuilder::prepare` lowers and plans a scan without IO into a `PreparedScan`, which exposes its rows, selection, filter plans and IO service, and `PrunePlanner` and `ChunkPlanner` are public for an engine to compose, the prune planner continuing into whatever it is built with and passing `None` when no proof ran. The builder's ordering, concurrency, mapping and streams go with the driver, and the layout's tests drive their scans on a run of their own. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The opener composes a scan's planners, a prune planner into a chunk planner when the scan has a filter and the chunk planner alone when it does not, and a driver of this crate runs them: the root run stepped on `spawn_cpu` tasks as DataFusion polls the stream, each spawned chunk on a task of its own up to the concurrency, or on the root run when the engine's partitions already fill the cores, with the concurrency derived from cores per partition unless configured. Batches are mapped to Arrow on the task that produced them. Every spawning decision of a V2 scan under DataFusion is in this crate. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
`vortex_file::v2::ScanDriver` composes a scan's planners and runs them on the session's runtime, the way an engine's driver does, for the file API's callers: the compress and random-access benchmarks move to it. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The opener prepares the scan where compute runs, composes its planners itself, a prune planner continuing into a chunk planner when there is a filter and the chunk planner alone when there is not, and hands the driver the first planner. The driver only drives. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The prune planner skipped prefetching the zone maps of proofs an earlier scan of the file had proved. The segment list now comes from compiling the proof, and a proved proof compiles to its kept result without a segment scan, so the check repeated what compile already knew. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
The scan code named the filter's zone-map plans proofs. They are the filter's pruning expressions evaluated over the file's zone map, so the layout, file and DataFusion code now says pruning, matching prune_zones, PrunedZones and the pruning field they already used. Behaviour is unchanged. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
…steps A planner that waits can now say what it will read later: the run asks every parked planner to speculate when nothing else can compute, and registers the answer with its root's session as reads that are never delivered. The prune planner speculates the predicate columns of the next chunk of its partition on each ask, so they are read while the zone maps are, as its filter planner would read them once pruning keeps the chunk; a scan selecting rows by index speculates nothing. The segment IO service keeps speculated reads for whichever session fetches a segment first, so a chunk on its own session takes the read the prune planner started. The root run's CPU step now runs until the run has a batch, waits, or is done, collecting every chunk it spawns, instead of one spawn_cpu hop per chunk. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
The prune planner ran its pruning plans one after another, each over the rows the previous one kept, so a filter with three plans waited for three reads in turn, each wait costing a round trip and a hop of the root run. It now builds every plan's adaptor in its first compute, publishes all their zone-map reads as one batch, waits once, and intersects the kept rows. Each adaptor numbers its requests from its own range, so the planner routes deliveries by request id. The separate compile of each plan to prefetch its segments is gone: the adaptors' fetches are the reads. Debug events mark when the pruning reads go out, when a chunk's reads are speculated, and when a chunk's run starts. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
The root run was stepped on a spawn_cpu task per step, so a partition paid a hop to take its zone-map completion, another to step the run, and a task spawn for its chunk, each a few hundred microseconds when every core is busy. The root run now runs on one task that awaits its reads in place, like a chunk's, and the first chunk a partition spawns runs on that run as another root. Later chunks are spawned as before, so one partition with many chunks stays parallel. 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>
Planning cuts a partition's rows from the footer's row count on the writer's 8192-row grid, but the writer buffers chunks to megabytes, so at fourteen partitions nearly every boundary fell inside a chunk and both neighbours decoded it: q19 at SF1 retired 17% more instructions than the stacked V2 scan, which aligned partitions to natural splits, and none more at one partition. The opener now snaps the planning-time rows up to the natural split boundaries of the columns the scan reads, computed once per file as the byte-range fallback already does, so the partitions still tile the file and no chunk is decoded twice. 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>
…reads Speculation prefetched only the predicate columns of the next chunk, so the projection columns interleaved between them in the file were read as gap bytes and then read again on their own: q6 at SF1 issued 23 physical reads for 164 requests where the stacked scan issued 14, and 14% more bytes. The prune planner now speculates the batch the chunk's filter planner would publish, predicate prefetches and projection announcements together, and the segment IO service keeps speculated announcements, requested and never polled, for whichever session fetches the segment first. 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>
…erlap The session polled a fetch as it was submitted, which marked it wanted and let the read driver dispatch the zone map alone before the run had speculated the chunk's reads microseconds later: q6 at SF1 still issued 25 physical reads for the stacked scan's 14. The run now speculates before it polls its sources, and the session registers a fetch on submit and polls it when the run polls, so what a planner speculates is registered beside the reads it waits for and a coalescing source folds them into one read. 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>
A chunk's filter planner announced its projection through its own session, which registered a second request for a segment speculation had already announced beside the chunk's reads. The second request was dispatched alone when the projection fetched, after the chunk's read had gone: two of each partition's eleven requests came back as a separate read inside the same range. A session announcement now takes the speculated announcement, and a fetch prefers it too. 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>
A chunk straddling two partitions is fetched by both. A partition's chunk took its speculated read out of the service's map, so when the neighbouring partition speculated the same segment a moment later it started another read, and once the first had completed the file's shared source no longer folded them: on q6 at SF1, 26 segments were read twice. A speculated read now stays in the service's map for as long as the service lives, cloned by every session that fetches it. Trace events name each new segment request by the path that made it. 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>
Filtered chunks were cut on a fixed 1M-row grid over the file, so a partition of 430K rows straddling a grid line got a small second chunk, speculated only on the next idle wait, after its neighbour's read had gone: on q6 at SF1 those tails came back as separate reads of ranges already read. Chunks now start at the partition's own first row, so a partition shorter than a chunk is one chunk and its reads go out together. 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>
One chunk per idle wait left a partition of five chunks at SF10 reading its last three only once their tasks started, where the stacked scan had every split prefetching from the start. The prune planner now speculates every chunk's predicate prefetches and projection announcements in one batch on the first ask, registered together so the partition's reads coalesce as one scan's. 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>
Two chunks of one partition can straddle a writer chunk. The first chunk's session took the speculated announcement for it and read it; the second found nothing speculated, announced a request of its own and read the segment again at the end of the scan: q12 at SF10 issued 30 single-request reads of ranges already read. A speculated announcement now stays with the service, a session's own announcement of the segment adds nothing, and the first fetch turns it into the shared read every later session finds. 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>
A projection-only partition of seventeen 64K-row chunks handed each chunk through the channel on its own, each needing a poll, a hop and a task spawn, so its chunks started their first reads a few hundred microseconds apart and the last ones no longer coalesced: q12 at SF10 issued 27 single-request reads of the orders table at the end of the scan, where the stacked scan spawned every split of a partition within microseconds. The root task now collects every chunk the run spawns until it has a batch, waits, or finishes, and sends them as one group the consumer spawns at once. 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>
…ments A read only announced is not started until the projection fetches it, after the filter, with nothing to overlap, and in a file whose columns are written in bursts a wide projection column shares no read with the narrow predicate beside it. On q14 at SF1 the projection waited 2.4 to 4 ms on reads of its own while its CPU was under 0.1 ms, the whole gap to V1, which polls all of a split's reads at open. The prune planner now speculates the projection's prefetches after the predicates' in the same batch, and the session registers a batch's announcements before it starts any read, so what is announced is dispatched with the reads it coalesces with. Debug events mark the projection's start, reads and finish, and each batch handed to the engine. 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>
A prepared scan carries a set of metrics its planners and morsels update, registered with the engine's registry under the partition's labels like the IO metrics: pruning duration and rows, speculated requests, chunks planned and skipped, filter duration, conjunct evaluations and their rows, projection duration and the wait from its reads going out to its first piece, rows projected and batches handed over. The DataFusion opener registers them, so they print beside the engine's own per-operator metrics. 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>
…hunks A partition without a filter had each 64K-row chunk fetch its own segments as it started, so the coalescer saw them piecemeal: q9's lineitem scan at SF1 issued 62 reads and 18% more bytes than V1, which registers a split's reads at open. The chunk planner now prefetches the partition's projection segments before it emits a chunk, and a prefetch is shared by every session of the service, like a speculated read, since chunks run on sessions of their own. Listing a plan's segments now shares the scan's value cache, so a dictionary's values already held list no segment and are not read ahead again. Timer metrics gain a total. 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>
Debug events now bracket each phase a partition goes through: open prepared and stream returned, pruning started, reads out, speculated and finished, filter started and each conjunct evaluated, projection started, reads out, first piece and finished, and each batch sent, so a trace gives the timeline of a scan without instrumenting it again. 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>
The driver takes driver metrics from the opener and records, under the partition's labels, the time from the stream's creation to its first batch and to its end, each spawned chunk's run, and how many chunks it spawned, so --show-metrics gives a partition's timeline end to end beside the stages' own durations. 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>
…ions waiting for it Every partition of a file rebuilt the filter plans and optimized the projection on every query: all fourteen open together before any has filled the file's cache, so the cache never hit within a query. On q6 at SF1 that was 448 us of filter plans and 92 us of projection per partition, on q14 177 and 142, at open, on every partition, fourteen times over. Each plan is now a once-cell in the cache: the first partition to reach it builds it and the others wait for that plan alone, not for the cache. Debug events time prepare's steps. 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>
…nd values The scan cache keys plans, pruned zones and values by the identity of a bound expression's tree, and every DataFusion partition binds the filter and projection on its own, so no two partitions ever hit the same entry: every partition built every plan, proved its own zones and decoded every dictionary's values for itself. A prepared scan now interns its filter and projection in the file's cache, structurally, and uses the shared tree, so the once-cells build once and the values are decoded once per file. 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>
A filtered partition of fewer rows than a chunk ran as one chunk, on one task from its first read to its last batch, and an engine's partitions started together straggled on uneven cores: identical 430K-row filter chunks took 0.5 to 5 ms at SF1, and the query waited for the slowest while V1's finer futures were stolen by the idle cores. A filtered chunk is now a quarter of the partition's rows on the row grid, between 256K and 1M rows, so a partition's chunks after its first are spawned where the runtime has room. SF1 partitions get two chunks; SF10 partitions keep their 1M-row chunks. A fixed 256K cost SF10 q1 8%. Local, 14 partitions, against the previous head: SF1 0.993 median (q15 0.976, q19 0.982), SF10 1.008. 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>
Interning hashes an expression's whole tree, and under a mutex the partitions opening a file together serialized on it: 3 ms of lock spinning over 15 runs of q15. Readers probe concurrently; only a miss takes the write lock, and checks again under it. 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>
… beside A zone-map fetch is the read everything else in a partition waits on, yet the file's read driver coalesced it with the chunk reads speculated beside it, so pruning waited for a 3 MB read from the page cache and, over an object store, for a 9.5 MB one behind 150 others. An urgent intent reads a segment on its own as soon as it is published, coalesced only with other urgent reads: a segment source may request a segment alone, the read driver keeps such requests apart, and the prune planner's adaptors publish their reads urgent. Over MinIO with 20 ms of round trip, SF10 q6 pruning finished at 25 to 50 ms instead of 320 to 390. Local: neutral (SF1 0.996, SF10 1.005 against the previous commit). MinIO with 20 ms round trip and 100 MB/s per connection: SF1 0.60 of V1, SF10 1.2 (the SF10 loss is addressed next). 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>
A partition speculated every chunk's reads at once, and the file's read driver dispatches in registration order, so with an engine's partitions started together the last partitions' first chunks queued behind the first partitions' last: over MinIO at SF10 every partition's zone maps and first batches came at once, and the batches of the rest tailed for 300 ms while V1, reading one split at a time per partition, shared the bandwidth evenly. Each partition now reads ahead only its first chunks when the store is not a local file, two by default; the later chunks fetch their own reads as their tasks start, interleaved across partitions. A local file keeps reading every chunk ahead: reading two ahead there cost SF10 6%. MinIO, 20 ms round trip, 100 MB/s per connection, 10 partitions at SF10: 0.90 of V1 at the median (q1 0.79, q19 0.66, q6 1.23 bimodal), from 1.27 with everything read ahead; SF1 at 14 partitions 0.61 of V1. Local unchanged (SF1 1.001, SF10 1.001). 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>
A V2 scan cache lived only as long as the scans of one query holding it, so every query lowered the file's layout, optimized its projection and filter plans, proved its zones and decoded its dictionary values again: q2 at SF1 ran 35 prepares a query, 1.6 ms of CPU, with the first per table on the critical path of a five-table join chain. The cache now lives beside the footer in DataFusion's session-level file metadata cache, so a later query of the file plans nothing again; partitions opening a file the cache misses still share one cache, put with the entry the first of them writes, and a rewritten file invalidates both. The entry's reported size does not count the cache, which grows after the entry is put. Local, 14 partitions, against the previous commit: SF1 0.965 (q19 0.90, q6 0.93, q2 0.96, q13 0.96), SF10 0.962 (q6 0.95, q12 0.95, q19 0.94). Against V1: SF1 0.955, SF10 0.81 on the targeted sets. 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>
Draft, against develop. Replaces the stack #10641 → #10648 → #10664 → #10669 → #10678 → #10670 with one branch, none of whose commits adds a plan walk.
What
A V2 scan is planned once per file and run as a chain of planners, each emitting the next through the planning protocol. No stage derives anything a stage before it already knew. The engine composes the planners and owns every decision about where work runs: all CPU is a planner's
compute, all IO is aspawn_ioread.Protocol additions in
vortex-scanandvortex-io:Planner::speculate(default none), asked by the run of every parked planner when nothing can compute, before the run polls its sources;IoSource::speculate(default: submit), which a source may serve for every session of its service. A speculated fetch is refused.The DataFusion driver (
persistent/scan_v2.rs) runs the root planner on one task that awaits its reads in place, runs the partition's first chunk on that run, and hands the chunks spawned between two hand-offs to the consumer as one group, each on a task of its own up to the concurrency, or on the root run at a concurrency of one. The same driver serves the file API (vortex_file::v2::ScanDriver).The segment IO service keeps speculated reads for whichever session fetches a segment, for as long as the service lives, since chunks straddling two partitions or two chunks are fetched by both.
Benchmarks
Local, hot page cache, 14 DataFusion partitions, interleaved A B B A per query, two rounds, 20 iterations at SF1 and 8 at SF10, pooled. Ratios are this head over the baseline; below 1.0 is faster.
SF1 wins: q12 0.77, q6 0.83, q19 0.82, q14 0.95, q11 0.95, q15 0.96; losses q13 1.03, q8/q9/q21 1.02. SF10 wins: q12 0.75, q6 0.78, q19 0.83, q1 0.88, q15 0.92, q14 0.93; losses q13 1.05, q22 1.03, q10 1.02. The head before this round (141570f) measured SF1 0.999 and SF10 0.973 against V1, and SF1 0.979 / SF10 1.000 against #10664.
Targeted, three rounds, q4/q6/q8/q9/q14/q15/q19/q22 at SF1 and q1/q6/q9/q12/q19 at SF10, median / min geomean against V1: SF1 0.984 / 0.963, SF10 0.884 / 0.894.
Planning once per file, across queries
Every query re-planned each table: the scan cache died with the last scan of a query, so the next query lowered the layout, optimized the projection and filter plans, proved the zones and decoded dictionary values again. q2 at SF1 ran 35 prepares per query. The cache now lives beside the parsed footer in DataFusion's session-level file metadata cache, invalidated with it when the file changes. Local, 14 partitions, against the commit before: SF1 0.965 (q19 0.90, q6 0.93, q2 0.96, q13 0.96), SF10 0.962. Against V1 on the targeted sets: SF1 0.955, SF10 0.81.
Object store: MinIO behind toxiproxy
Local MinIO holding the same SF1 and SF10 files, reached through toxiproxy with a latency toxic of 10 ms ± 4 ms jitter in each direction (about 20 to 28 ms per request) and, for the capped rows, a bandwidth toxic of 100 MB/s per connection. Ratios are this branch over V1; below 1.0 is faster. SF10 ran on 10 partitions so MinIO and the proxy had cores left; SF1 on 14.
q9 stays at parity because its three scan phases are serialized by the join tree and each is a couple of round trips in both scans. SF10 over MinIO is bandwidth bound (400 MB per q6 through one box) and V2's runs there are bimodal between about 250 and 800 ms; the median is what the table shows. The slow mode is state a previous V2 query leaves in the process: six single-iteration processes all ran q6 in 360 to 436 ms, while six-iteration processes alternate 230 to 260 ms and 530 to 785 ms, and V1 is stable at about 340 ms after a cold first run. Partition streams and chunk tasks do end within their query (checked with end events on q13: every stream of the orders scan ends 130 to 155 ms after it starts, before the next query begins), so the leftover is not a task or a stream; the scan stream timer's 2.1 s reading on a 200 ms query is a metrics artifact. What a finished V2 query leaves behind that slows the next one over the object store is still open; the HTTP connection pool and the proxy are the remaining suspects.
Two of this round's commits are for that path. Zone maps are read on their own (an urgent intent, coalesced only with other urgent reads) so pruning no longer waits for the data read beside them: over MinIO SF10 pruning finished at 25 to 50 ms instead of 320 to 390. And a file in an object store reads ahead only its first two chunks per partition; the read driver dispatches in registration order, so reading every chunk ahead made the last partitions' first chunks queue behind the first partitions' last (SF10 1.27 to 0.90). A local file keeps reading every chunk ahead, where reading two ahead cost SF10 6%.
The other two are local: a filtered partition is cut into chunks of a quarter of its rows (256K to 1M), so SF1 partitions have a second chunk to spawn and stragglers on the slow cores gate the query less (SF1 0.993, SF10 1.008 against the previous head); and interning probes under a read lock, since partitions opening together serialized on hashing the expression tree.
To rerun:
brew install minio minio-mc toxiproxy,minio server <dir>on 127.0.0.1:9000,mc cp --recursive vortex-bench/data/tpch/<sf>/vortex-file-compressed local/tpch/<sf>/,toxiproxy-cli create -l 127.0.0.1:9100 -u 127.0.0.1:9000 minio, add the toxics, then the bench withAWS_ENDPOINT=http://127.0.0.1:9100 AWS_ALLOW_HTTP=trueand--opt remote-data-dir=s3://tpch/<sf>/.What was slower, and why
Found by instruction counts (
/usr/bin/time -l) at 1 and 14 partitions and the read driver's trace, not by timing:Metrics and phase timing
A prepared scan carries
vortex.scan.*metrics its stages update, registered under the partition's labels like the IO metrics, so--show-metricsprints them beside DataFusion's: pruning duration and rows, speculated requests, chunks planned and skipped, filter duration and conjunct rows, projection duration and the wait from its reads going out to its first piece, rows and batches. Timers also report_total. Debug events bracket every phase of a partition (RUST_LOG=vortex_layout::scan=debug,vortex_datafusion::persistent=debug): open prepared and stream returned, pruning started, reads out, speculated and finished, filter started and each conjunct, projection started, reads out, first piece and finished, each batch sent.Median per lineitem partition at SF1, 14 partitions, microseconds from the partition's open: open prepared 500, stream 720, pruning reads out 1000, speculated 1260, pruning finished 1700 to 2000, filter done 4400 to 8300, projection done 300 later, first batch 400 later. The zone map is the first read dispatched, alone, 1.36 ms after it is polled: that is the read driver's dispatch latency, the same on V1's path.
What was tried and backed out
Branch
ji/scan-v2-read-ahead-experimentsholds a read-ahead planner after pruning (a window of chunks prefetched, the next announced, intervals listed while the zone maps are in flight), a file announcer that registers a scan's whole interest once per file, an urgent intent that reads zone maps on their own, a bounded drain in the read driver, and speculated reads shared per file rather than per scan. Each was measured against this head and V1 on q4/q6/q8/q9/q14/q15/q19/q22 at SF1 and q1/q6/q9/q12/q19 at SF10. The announcer halved reads on projection-only scans (q9 orders 11 to 2) but, for filtered scans, widened zone-map reads into neighbours' data; the rest left filtered scans issuing two to three times V1's reads, and every combination was 1 to 3% behind this head at SF1 and 4 to 7% at SF10. The IO-level pieces (urgent intent, read-alone segment requests, bounded driver drain) are correct and tested but did not pay on this workload.Checks
cargo clippy --all-targets --all-features -- -D warningson vortex-scan, vortex-io, vortex-layout, vortex-file, vortex-datafusion;cargo nextest runon the same (1812 tests on vortex-scan and vortex-io, 1569 on the others);cargo +nightly-2026-09-10 fmt. Not run: the workspace-wide checks CI runs.🤖 Generated with Claude Code
https://claude.ai/code/session_01Vr65ZwPZz4LPjkvDA6ZUuE