M2b: Chunked execution engine

Tracking issue for **M2b**. Specification: [`engine/`](../blob/main/docs/specs/engine), in particular [`02-execution-model.md`](../blob/main/docs/specs/engine/02-execution-model.md), [`03-memory-and-spilling.md`](../blob/main/docs/specs/engine/03-memory-and-spilling.md) and [`04-operator-plan.md`](../blob/main/docs/specs/engine/04-operator-plan.md).

Numbered M2b rather than getting a number of its own, so that the nine milestones after it keep their numbers.

### Why this exists

firepanda runs one kernel over one whole column and writes a whole new column. That is pandas' execution model. At ten million int64 rows a column is eighty megabytes, so every operator boundary is a full round trip through DRAM that a chunked engine does not do at all.

DuckDB never had this model. Polars had it, replaced it with a streaming engine, and measures three to seven times on PDS-H from the change alone. We are five to nine times slower than DuckDB on joins with individually competitive kernels, and no amount of further kernel work closes that.

### What it moves from where

M5 was parallel execution and M10 was streaming and out of core. This takes the morsel scheduler from M5 and the spilling from M10 and does them together, because they are one design and doing them separately means building the scheduler twice. M5 and M10 keep the wider operator coverage and the distributed pieces.

### Scope: foundations

- [x] `ChunkedArray` becomes what a `DataFrame` column is, replacing the bare `AnyArray` list
- [ ] Every kernel gains a chunked entry point that walks chunks and calls the existing contiguous kernel per chunk, inner signatures unchanged
- [x] Chunk size when we create chunks, reader boundaries preserved when we do not. The read takes a height and there is deliberately no default, because the best one is 16k to 32k on thirty two threads and a quarter of a million on eight, and a number baked into the reader would be wrong on one of them
- [x] A sortedness flag on a column, set by sort and by readers that know

### Scope: scheduling

- [x] `MorselQueue` with an atomic cursor, on top of the existing `parallel_for`
- [x] `parallel_morsels(body, total, morsel_rows)` starting `worker_count()` tasks that each loop taking morsels
- [x] Every kernel that currently slices into equal per worker pieces converted to it
- [x] The worker count sweep rerun, to confirm the optimum moves from 24 back to 32 on a 32 thread machine

### Scope: pipelines

- [x] A `Node` interface of `update_state`, `process`, `finish`
- [x] A `Pipeline` of source, operators and sink, executed by pushing chunks
- [x] Pipeline cutting at breakers, found by one traversal
- [x] A `Materialize` fallback node that collects chunks, calls the whole column kernel and emits chunks, so anything not yet ported still works
- [x] The driver runs the elementwise operators at the front of the line on every core, a batch of one chunk per worker at a time, the rest of the line staying in chunk order on the calling thread
- [x] Chunks handed out by a shared counter rather than one task each. The second half of this line turned out to be wrong: measured with the counter in place, a projection over sixty four chunks was still forty per cent slower spread out, so the gate that keeps a projection and a cast on one core stays

### Scope: operators off the fallback

- [x] Elementwise: filter, select, arithmetic, comparison, cast, `with_column`
- [x] Group by that folds partials, one row of state per group, chunks released as they go
- [x] Group by keyed on a map that outlives the chunk, so absorbing a chunk costs the chunk rather than the group count
- [x] The pass that lays a group's values out for a median, a quantile or a distinct count is partitioned and parallel, so a high cardinality key is not paid for one row at a time on one core
- [ ] Group by, thread local partitioned tables merged partition wise
- [x] Sorted group by, taken when the key is flagged sorted
- [x] A reduction over the whole input, folding each chunk into a running row, which is what a join in a pipeline hands its chunks to
- [ ] Inner and left join, build side consuming chunks, probe emitting as it goes
- [x] Inner, left, semi and anti join as a pipeline operator, build side held whole and hashed once in `bind`, probe emitting a chunk per chunk
- [ ] Sort, consuming chunks and emitting chunks

### Scope: selection vectors

- [x] A chunk carries an optional selection vector
- [ ] Kernels read through it, in the shape of DuckDB's unified vector format
- [ ] `filter`, `take` and join output produce selections rather than copies
- [ ] A flatten threshold, chosen by measurement, for when a sparse selection is worse than a copy

### Scope: memory and spilling

- [ ] A memory manager with a limit, an atomic total, `reserve`, `release` and `pressure`, counting only
- [ ] External sort
- [ ] External group by, partitions spilled and merged
- [ ] External join, build partitions spilled, probe rows routed and replayed
- [ ] A bail out for the skewed key case, where more radix bits do not separate a partition

### Ahead of all of it, as ordinary pull requests

- [x] ~~Salt bytes beside the pointer in the hash table, so a probe miss costs one cache line instead of two~~ Does not apply, see the comment below: the slot already holds the hash and the ordinal in sixteen bytes, so a miss is already one cache line
- [x] Build side chosen by size rather than by argument order
- [x] The slab a grouped median, quantile or distinct count lays its values out in is no longer zeroed before it is filled, the fill covering every element of it
- [x] A grouped correlation or covariance over two columns of the same dtype dispatches on the pair instead of casting both to float64 first
- [x] Grouping on several keys widens the first key's ordinals inside the first packing step rather than in a pass of its own
- [x] The merge that folds the thread local tables at the end of a grouped min or max is vectorized, the reduction's identity standing in for a group a worker never reached
- [x] A key tuple of integers of one dtype, each in a range narrow enough to index, packs straight out of the columns with no per key factorize at all
- [x] Grouping on several keys packs the ordinals into the combined key in a single weighted sum per row rather than one fold pass per key, the packed column four bytes wide where the product of the counts fits
- [x] A factorize no longer zeroes the ordinals it is about to write, on any of its eight routes
- [x] A string factorize settles a hash match against a view it kept when it handed the ordinal out, rather than reading the representative row back out of a views buffer sixteen bytes a row wide
- [x] A join builds only the output columns it was asked for, so a column the query drops is never gathered
- [x] The join's build side is a value the probe reads rather than a phase inside one function, which is what a streaming join needs
- [x] The join's built table carries its key dtype as a field rather than as a type parameter, so a node in a `Variant` can hold one, and probing it in pieces writes what probing it in one go writes
- [x] The join's code to row lists are a value too, and the walk that pairs against them takes a stretch of rows rather than a frame, so pairing a side in pieces gives what pairing it once gives
- [x] A reduction immediately behind the parallel prefix folds each chunk on the core that produced it, one partial row per chunk, merged after
- [x] A gather no longer zeroes the column it is about to fill, so the memset that faulted every output page onto one core is gone
- [x] The join's shared ordinal list is not zeroed either, the build side writing its own stretch and the probe side the rest
- [x] A grouped mean takes its count out of the same scatter as its sum, and the variance, the standard deviation and the skewness take theirs from that one call rather than asking again
- [x] The merge that folds the workers' tables at the end of a factorize stays on the calling thread when there are only a few thousand entries to fold, rather than waking every core to move a few kilobytes
- [x] The scan that decides whether an integer key gets a table indexed by its value reads its first sixty five thousand rows where it stands and hands the rest out one morsel a core, so a column that qualifies is no longer read to the end on one thread in front of two passes that are on all of them
- [x] The check that no group ordinal names a group that was never allocated runs once for a grouping rather than once for every reduction of it, the callers that made the ordinals themselves saying so
- [x] The discovery pass of a factorize on a narrow integer key stops as soon as its table is full, which on a direct table means there is nothing left in the range to find
- [x] A group by asking for several reductions of one grouping runs them in one pass over the ordinals, each worker cutting its slice into blocks and running every reduction over a block before moving on to the next one
- [x] A text key of few enough values and no nulls is factorized against one dictionary built from the front of the column and probed read only by every worker, so there is no table per worker, no merge and no pass to renumber local ordinals into global ones
- [x] A grouped pair reduction has a microbenchmark row of its own, so db-benchmark q9 can be worked on without running a whole benchmark suite at a hundred million rows to see whether anything moved
- [x] A grouped correlation or covariance reads its two columns once rather than twice, centring each group on the first pair it is seen with instead of on its means, with every worker's origin moved onto one in the merge
- [x] The eight accumulators a pair reduction needs for a group live in one sixty four byte slot rather than in eight tables, so a row touches one cache line instead of eight
- [x] A fused group by has a microbenchmark row on the key shape db-benchmark q3 and q5 group on, and the two changes it was built to measure were both losses and are recorded as such
- [x] A factorize on an integer key whose range is too wide for the workers' tables to be merged discovers its groups on one thread and numbers its rows on every core, rather than running the whole pass on one
- [x] A text factorize that partitions its rows by hash carries each row's view into the partition beside the hash and the row, so settling a hash match compares two views the build was handed rather than reading a scattered one back out of the column
- [x] A join whose build side is as tall as its probe side has a microbenchmark row of its own, so db-benchmark j5 can be worked on without running a hundred million row suite to find out whether anything moved
- [x] The chunk a join is handed is measured rather than assumed, and the answer is thirty two thousand rows and not a hundred and twenty eight thousand, which is worth two times on every pipelined join query
- [x] Three further ideas on the partitioned text factorize were measured and all three are recorded in the function rather than shipped, including one whose obvious probe reports the opposite of the truth
- [x] An outer join pairs on every core rather than on one, marking a byte per key code instead of a bit per built row, and reading the byte before writing it so the cores are not passing the marks around, which is three times on `join/outer`
- [x] The morsel size is measured rather than inherited from DuckDB, and unlike the join's chunk it turns out to be right where it is, with the ten million row measurement that said otherwise recorded beside it because it is the more useful half
- [x] The join microbenchmarks join on text as well as on integers, which no row did before, and the text key turns out to be four and a half times the integer key on the same shape
- [x] A join on a text key builds its table on the smaller side and probes it read only, instead of concatenating both key columns and factorizing the pair, which is 1.56 times against a dimension of a thousand and 2.28 against one of a hundred thousand, and is declined when the two sides are close in height because the build is still serial
- [x] The gather issues a prefetch eight rows ahead of itself, the way the hash table's probe already did, which is three per cent on the big join where the source is eight hundred megabytes against a thirty six megabyte L3 and nothing measurable where the source is L3 resident
- [x] A gather whose indices ascend by one is taken as the copy it is, which is what an inner join that matches every probe row once hands its probe side, and is seven per cent on the join microbenchmarks that gather four columns and four per cent of the CPU on j1 and j2
- [ ] The string build spreads across cores, which is what would let two sides of the same height take the probe route as well and would remove the threshold rather than moving it
- [ ] j2's 2.44x against DuckDB is measured against a plan DuckDB pessimizes with a dynamic filter that rejects nothing, so the honest medium join comparison is j3's 1.08x and the three integer joins on the small and medium tables are one problem
- [x] The benchmark suite runs the join queries the public suite runs, with the character join upstream has and none of ours had, so j4 is the medium inner join on a text key and pairs the same rows as j2 on an integer one
- [x] A text key can be a streaming join key, and the reason j4 still runs the whole frame shape is not the node any more, it is that j4's build side at 5GB is forty eight megabytes against a thirty six megabyte L3, where the pipeline loses 1.25x, while at 0.5GB the build side is five megabytes and the pipeline wins 1.29x
- [ ] j3 against DuckDB, which is the one join in the set where we are level rather than ahead, and where DuckDB's outer join is faster than its own inner join on the same tables
- [x] A filter compacts on every core, the per morsel kept counts prefix summed into the output position each morsel begins at, which is three times on q6's projection and takes the query from about forty three milliseconds to about thirty one
- [x] A join on more than one integer key column packs the tuple into one uint32 that both sides agree on and takes the single key route, dictionary on the smaller side and all, which is 1.27 times on a compound key join and 2.21 times when the two sides are the same height
- [x] The five text operators are handed out to the cores, which they were refusing on two arguments that both turn out to be wrong, and which is worth 13.8 times on `length`, 13.2 on `substring`, 12.0 on `trim`, 8.0 on `upper` and 7.5 on `position`
- [x] `x IN (a, b, c)` is one set lookup rather than a compare per member joined by `or`, which is 1.89 times at two members and 6.02 at thirty two
- [x] The linear route of the set lookup keeps eight SIMD blocks live and walks the set once for the group rather than once per block, which is three times at every set size, and its threshold against the hash table is remeasured with the table lifted out so it could be run below it
- [x] A case change over ASCII text is one SIMD pass that flips the letters and checks for a byte at or above 0x80 at the same time, written straight into the payload of the column being built, which is sixty times and takes `upper` from four hundred nanoseconds a row to 6.6
- [x] A `strip` or a `trim` walks the ends of a row in byte offsets rather than in character ordinals, so it stops looking after at most three bytes instead of rescanning from the front of the element on every character, which is 2.04 times on text with nothing to come off and 4.83 times on text with three spaces on each end
- [x] A filter counts its mask once for the chunk rather than once per column, since the count is a property of the mask and nothing between the columns can change it, which is 1.22 times on a four million row line read as one chunk and about five per cent on the same line in chunks
- [x] A scan cuts a chunk taller than a morsel into morsel sized windows that share the parent's allocation instead of copying it, so a frame that arrived in one chunk takes the batched prefix like any other, which is 2.5 times on a four million row line and puts the one chunk row inside the spread of the chunked one

### Holes closed along the way

- [x] Top n per group, beside `group_ordinals`, which unblocks db-benchmark q8
- [ ] Merge join on two sorted inputs

### Exit criteria

- [ ] Every existing test passes with the engine on, and the `Materialize` fallback still present and still correct
- [ ] db-benchmark at 0.5GB and at 5GB on gamingpc, with the 5GB numbers reported as the headline, because the effect grows with size
- [ ] Group by queries at 5GB at least 1.5x their pre M2b selves
- [ ] Joins at 5GB at least 2x their pre M2b selves
- [ ] A group by over an input larger than the memory limit completes rather than failing
- [ ] A join over an input larger than the memory limit completes rather than failing
- [x] q8 answered
- [ ] The worker count sweep shows no interior optimum

### What is deliberately not in it

**A buffer manager.** Operator controlled spilling instead, which is three operators rather than a subsystem every allocation goes through. The Kuiper VLDB paper makes the case against fine grained buffer manager driven spilling from inside DuckDB, and Polars chose the other side.

**A lazy frame.** The public API stays eager and the pipeline is built and run inside each call. Letting a user build a plan across calls is M3 and M4, and this design is what makes that a front end change rather than a second engine.

**An optimizer.** Polars measured about ten percent from cost based planning, and it needs a plan to optimize.

**Window functions**, and **distributed and GPU**. The honest single GPU number on a bandwidth bound workload is about three times with a mature engine underneath.

### The shape of the work

Not a rewrite with a flag day. The `Materialize` fallback means the system works from the first pull request, and every entry in the operator list is one pull request that removes one fallback and has its own benchmark number.

That is how Polars did it. Their releases from December 2025 to April 2026 are almost entirely one more operator lowered to streaming, one more sink made streaming, one more thing made out of core, twelve releases in a row, with the old engine as the fallback throughout. It is still the fallback at 1.39 and the streaming engine is still not the default. Plan for the fallback to be permanent rather than temporary.

### Depends on

M2. A streaming engine with no streaming source is a benchmark harness, and the Parquet reader is the first real source. This is also why Polars did sinks and sources before joins.

Start with docs/specs/engine/02-execution-model.md, 03-memory-and-spilling.md, and 04-operator-plan.md, then inspect the unchecked items under engine/. Done means implementing and validating the remaining chunked execution, selection-vector, operator, and spilling work described by the specification.

Written by the indexing model from the issue text.