P2: pipeline executor

[tamnd]

Spec: perf/03. Needs the P1 types.

Fixes R3 and R7. ensure_flat forces the driving side row-at-a-time (exec.rs:1164,2494-2496). Only [Source,Scan]-headed first stages parallelize, which gets us ~1.8x on 8 cores (exec.rs:3865). Sinks are serial, and each worker reopens the file with cold caches.

• zu-exec crate: pipeline DAG, push-based PhysOperator (execute/finish per vector)
• morsel scheduler v2, group-aligned morsels (131072 rows), work stealing
• parallel sinks: partitioned join build / agg partial / topn / sort
• one shared Arc across workers, no per-worker reopen
• skew split when a worker sits > 4ms in a morsel while others idle
• LIMIT early stop through the pipeline
• keep old exec.rs around as the differential oracle until parity, then delete it

done when:

• B4 two-hop SF1 <= 0.4ms (today 1.6-2.8ms)
• >= 6x scaling at 8 cores, all pipelines parallel
• alloc counter still 0 (P8)

[tamnd]

P2 landed in #83. Here is where each box stands and why three of them are still open.

Checked off:

zu-exec crate. New crate with a plan compiler and a push driver. The compiler walks one LogicalPlan chain and produces a single pipeline of push operators over factorized ChunkSets; the driver splits the driving scan into morsels and pushes each one all the way to a per worker sink partial. Coverage today is the linear read pipeline: one required node scan, filters compiled to vector kernel programs with zone pushdown, single hop expands off the newest level, and a final Project or Aggregate that absorbs Distinct, Skip, and Limit. Anything else returns Ok(None) and the caller runs the old engine.

One handle per worker with no reopen. Not literally one Arc, and on purpose: a Snapshot is stateful, it carries decode scratch and reader positions, so sharing one behind a lock would just move the contention. Instead every worker gets its own fork that shares the block cache, the decoded pools, and a pool of reusable fork handles with the caller. Nobody reopens the file and a fork is warm on its first read, which is what the box was actually asking for.

LIMIT early stop. A shared quota that only counts the contiguous completed prefix of morsels. A worker heading that prefix stops mid morsel once its own rows close the quota, and the stop flag drains everyone at the next chunk boundary. Rows past the cut can only come from morsels after the prefix, and the final truncate drops those, so early stop cannot change an answer.

Old exec.rs stays as the oracle. The parity suite in crates/zu/src/exec2_tests.rs runs every claimed shape on both engines over one real zu1 file and compares rows exactly, order included, at 1 thread and at auto. It also asserts that the unclaimed shapes come back None.

B4 two hop SF1 at 0.4 ms. 7622269 paths in 0.243 to 0.249 ms locally, 0.172 ms on gamingpc under WSL, 0.558 ms on server3 while it sits at load 2. The win is a fusion: trailing expands whose only consumer is a bare count never build a level, the count comes straight off the CSR offsets, one step for expand then count and several for the hub shape the optimizer picks for an unseeded two hop. ldbc_two_hop_p50_ms comes down from 10 to 8 rather than near the measurement, because the two loaded hosts still have to pass it.

Alloc counter still 0. The gate passes, warm group pins and degree batches allocate nothing.

Still open:

Work stealing. Morsels are group aligned as asked, and there are eight per worker rather than four, because the tail decides the query: on a machine with fast and slow cores the slow worker picks up a last morsel nobody can finish for it, so halving the morsel halves that tail. Sixteen gave nothing back over eight. Stealing itself has nothing to steal from yet: workers pull from one shared claim counter, so a worker that finishes early already takes the next morsel. Leaving this unchecked until there is a shape where a per worker queue beats the shared counter.

Parallel sinks beyond agg partials. Agg partials are per worker and merge at the end, and the row sink is per worker with deterministic stitching by morsel index. Partitioned join build, topn, and sort are not here because neither joins nor ORDER BY compile yet; those come with the P3 operator set.

Skew split. Needs a workload where one morsel genuinely runs long while the others idle, and the cardinality work in P3 is where that comes from. Nothing on the current bench set produces it.

The 6x at 8 cores done when is the one I cannot prove on this fleet, and I want to be straight about why rather than move the number. Measured 1 worker against 8 on the expand filter count over a million person graph: local M4 4.5x and 4.8x on the scan query, server3 4.4x, gamingpc under WSL 3.8x and 4.0x pinned to its 8 performance cores, gamingpc on native Windows 1.0 to 1.9x and very noisy, server1 has 4 cores so the gate only prints there. No machine on the fleet is 8 quiet equal cores. The M series has 4 performance plus 6 efficiency cores, so 4.5x is about 86 percent of what those cores can add up to. gamingpc has 8 performance plus 16 efficiency cores and a scheduler that moves threads between them mid query. Both servers are shared vCPUs that never sit still.

What made me stop blaming the executor: a trace on gamingpc showed all 8 workers starting within 20 us of each other and morsels balancing 5/5/5/5 on the performance cores and 3/3/3/3 on the efficiency cores. That is the scheduler working. Four separate bugs did come out of chasing this, all fixed in the PR:

• a uniform spin before parking regressed local scaling from 4.4x to 3.6x, because outside Windows a parked thread wakes in microseconds and the spin only steals a core from the query still running. The spin is now Windows only.
• the caller parked on the completion condvar while every worker had already finished. On gamingpc the workers were done at 1.3 ms and the query still reported 4.5 ms. The latch is now an atomic the caller spins on, with the mutex taken in count_down before notify_all so a wakeup cannot slip between the check and the wait.
• the pool never grew past a couple of workers. Submit spawned only when it saw zero idle workers, but idle drops when a worker dequeues, long after the loop has pushed everything, so one parked worker served a whole batch alone. Spawning while queued jobs outnumber free workers took 4 and 8 threads from a flat 1.68 ms to 0.91 and 0.70 ms. This was the dominant one, and it only became visible after the bench started printing the whole thread curve instead of a single speedup.
• morsels per worker went from four to eight for the tail reason above.

exec_scale_8x goes in at 2.0, under server3's worst run, with the reasoning written into budgets.toml. Raise it as soon as a quiet 8 core box joins the fleet.

One more thing worth recording. A point predicate on the driving slot's id now declines and falls back. The old engine folds that filter into the scan and seeks the primary key index, so it touches one row, while the pipeline has no seek source and was scanning the whole table to find it. That cost 21 us on the session bench against 1.5 us on the old path and broke the 20 us ceiling. The seek belongs with the P3 operator work; until then the fallback is the correct answer and the point read is back to 1.46 us locally, 0.88 us on gamingpc under WSL, 2.00 us on native Windows, 2.92 us on server1, and 3.69 us on server3.

ZU_GATE=1 is green on this head locally, on server1, on server3, and on gamingpc both native and under WSL.

[tamnd]

Measured where the 100k group aggregation actually spends its time at eight workers, because the scaling on that shape is about 1.7x while the twelve group shape beside it scales 9.4x, and the two numbers had to have one explanation.

Split the query into the parallel drive and the serial finish behind it. At one worker the drive is 92 to 147 ms and the finish 14 to 33. At eight the drive is 35 to 49 and the finish is 24 to 64, so the tail is as large as the scan it is waiting on, and that is the whole of the missing scaling. The twelve group shape has a finish of 0.0 ms, which is why it scales. Inside the finish at eight workers the four parts read merge 8 to 18 ms, drain 2 to 12, sort 7 to 19 and row build 7 to 17.

#766 takes three of those four. The sink used to drain the folded table into a vector of key and state pairs, which is two small vectors per group built before anything is known about where any of them goes, sort that by comparing the values the groups decoded to, and then walk the result a third time to put keys and aggregates back in RETURN order. A hundred thousand groups is three hundred thousand allocations made and thrown away in the tail. The table has the packed key words, so it orders an index vector against those and decodes each group once, straight into its finished row. The word order is the value order for every kind a key part can be, and there is a test holding the new path to the old drain and sort on a key with one part of every kind in it, both string forms included.

100k groups went 136.7 ms to 111.9 at one worker and 80.6 to 64.9 at eight, medians of six paired alternating runs of two separately built binaries, branch faster in every pair. The twelve group shape and the thousand group string shape did not move, which is the check that this was the tail and not the scan.

The merge is the part left, and it is the one that needs the pool rather than a better data structure: every worker sees keys from all over the scan, so each partial is about as wide as the answer and folding eight of them into the first is seven full width merges in a row. Folding them pairwise up a tree is the same seven merges in three rounds. That is written and it passes, but this Mac has not been quiet enough to say whether it is a win, and an unmeasured perf change is not worth merging, so it is waiting on a quiet host rather than going in on the argument.

Not ticking the parallel sinks box. Making the serial tail cheaper is not the same as making it parallel, and the join build, the topn and the sort halves of that box have not been touched.

[tamnd]

Ticked the morsel scheduler line. It is done and has been for a while, and the box was open because nobody went back to check it against what the code does.

Morsels are group aligned at 131072 rows, which is GROUP_ROWS in zu-common, and they never cross a storage group so a morsel's CSR pins and zone reads stay inside one group. Sizes are chunk multiples targeting eight morsels per worker rather than four, because the tail is what decides the query: on a machine with slow and fast cores the slow worker picks up a last morsel nobody can finish for it, so halving the morsel halves that wait. Sixteen per worker gave nothing back. morsels_cover_rows_and_stay_group_aligned holds the tiling, the chunk start and the group containment.

The stealing half is a single shared claim counter that every worker fetch_adds against, rather than a deque per worker with stealing off the back. That is the same guarantee and a cheaper one: with one queue there is nothing to steal, because no worker can be idle while a morsel is unclaimed. Per worker deques exist to cut contention on a shared queue, and one relaxed fetch_add per 131072 rows is not contention worth cutting.

What is still open on the balance side is the box below it, the skew split, and that one is real. A worker that has claimed a morsel runs it to the end, so a morsel that is pathologically slow for reasons the row count did not predict, a hub expansion say, leaves the others idle at the end of the query with no way to help. That is the case the four millisecond rule is for and nothing implements it yet.

[tamnd]

The tree merge is in, #771, together with a second half that was not written when the last comment went up.

Both halves are the same idea. After #766 the hundred thousand group aggregation at eight workers is about 37 ms of parallel scan and 20 ms of serial tail, and the tail is two pieces of parallel work being done on one thread while the pool sits idle. Folding the eight workers' group tables into one is seven full width merges one after another, because the scheduler hands every worker morsels from all over the scan and so each partial ends up about as wide as the answer. Building the answer rows out of the folded table is a hundred thousand independent decodes.

The fold now goes pairwise up a tree, three rounds instead of seven steps, each round on the pool the scan has just finished with. An odd table carries to the next round rather than joining a pair so every merge in a round is the same size. The row build now takes slices of the settled order, one per hand, each hand reading accumulators out of the shared table rather than moving them away first.

Hands are per worker that ran, not per core. A query asked for on one thread has to be answered on one thread, tail included, or exec_group_mrows_s_core stops measuring a core. Under four thousand groups a hand it stays on this thread, since a latch and a lock cost more than the decoding they would split.

Measured on gamingpc, quiet, medians of six paired alternating runs of two separately built binaries. Hundred thousand groups at eight workers went 40.8 ms to 33.9, 243 to 296 M rows/s, faster in all six pairs. At one worker it read 99.5 against 99.4, unmoved, which is the check on the hand count. Twelve groups and the thousand group string shape did not move either, the first because it has no tail and the second because a thousand groups is under the split.

That shape scales 2.9x from one worker to eight now, against 2.4x before and 1.7x before #766. Still short of the 6x line on this checklist, and the remaining gap is the scan itself rather than the tail, so the next thing to look at is the probe.

[tamnd]

Two more of the P2 tail landed, both on gamingpc (32 cores, WSL Ubuntu), paired alternating runs of two separately built binaries.

#780 orders the groups of a wide aggregation on the pool. After #766 and #771 the fold and the row build were both parallel and only the sort of the group keys was still on one thread, which on a hundred thousand groups is most of what is left. GroupTable::rows now splits before it sorts: it samples every nth group, sorts the sample, and uses it as pivots to cut the group index vector into one ascending key range per hand, then each hand sorts its own range and builds its rows and the ranges are laid end to end. The split pass costs a few compares a group against a pivot list that stays in cache, against the seventeen a full sort of a hundred thousand costs. Groups sit in first seen order before this, so a stride samples fairly whether the column arrived shuffled or already sorted, and an uneven range costs a hand some idle time and nothing else. Timed in place over 24 runs of each binary the tail went 3.53 ms to 2.07 ms at eight workers and was unmoved at one. Whole query 28.3 ms to 25.8 ms at eight, faster in 8 of 10 pairs, 354 to 388 M rows/s, 1t unmoved at 91 ms. Scaling 3.2x to 3.6x.

#781 is the one that mattered more. The scale gate kept coming in under 6x and no amount of looking at the query explained it, so I timed each worker in place instead of timing the query around them, printing when each one began and ended. A query asking for eight hands was regularly getting six or seven. Two defects in the pool. First, submit pushed the jobs one at a time with a notify_one each, and a wakeup is not addressed to a job: a worker already awake takes the job the notify was meant for, the notified worker finds the queue empty and parks again, and the orphaned job sits until somebody loops back for it, which for a morsel scan is the end of the query. The whole batch now goes under one lock with one broadcast, and the spawn decision compares queued jobs against workers actually free rather than against zero. Second, the spin before a worker parks was Windows only, on the reading that a parked thread wakes in microseconds elsewhere. On this box it does not, it takes about a tenth of a millisecond, which against a query that runs in one is not a rounding error. Non Windows now spins 100 microseconds before parking and the submitting thread spins 200 on the latch. With both, all eight workers start within 0.13 ms of the first and every one of them is inside the scan. Over nine pairs the expand query went 4.9x to 5.4x scaling and 1.12 ms to 1.02 ms at eight workers, faster in 9 of 9 pairs, and the scan query 4.2x to 5.3x and 0.80 ms to 0.67 ms. Single worker moved 2 percent the wrong way, which is the spin. Groupby unmoved, it was never losing a worker.

Worth saying what is still in the way of 6x on the hundred thousand group shape, since it is not a tail problem any more. The same scan with twelve groups scales 6.4x. The difference is eight private counting tables of about four megabytes apiece and a random probe into thirty two megabytes of them, which is a bandwidth wall rather than anything schedulable. Fixing it means radix partitioning the rows before the aggregate the way DuckDB and Umbra do, or one shared lock free table. Both are large and neither is started. Written up in bench/budgets.toml next to the floor it explains.

Next in P2 is the morsel granularity, which is being re measured now because the note in make_morsels saying sixteen morsels a worker gave nothing over eight was written while the pool was still losing a worker, so it was measuring the wrong thing.