Pushdown Joins & Aggregates
Aggregates: the math happens in the VM
Section titled “Aggregates: the math happens in the VM”SELECT SUM(price) FROM t WHERE qty > 0 used to mean streaming every
matching row offset back to the orchestrator. It now compiles to
ACC_SUM into a reserved register: the worker folds every block of the
table into a worker-wide state and replies with one terminal frame — a
64-byte header plus one 16-byte {value, valid} record per accumulator.
| Query | Rows | Worker output on the wire | If it were offsets |
|---|---|---|---|
COUNT(*) … WHERE | 10,000,000 | 80 bytes, one frame | 160 MB, 2,443 frames |
SUM(price) WHERE qty > 0, 5 containers | 5,000,000 | 80 bytes, 27.7 ms end to end | 72 MB |
SUM, MIN and MAX of no rows are NULL; COUNT is 0. AVG rides as
SUM and COUNT and is divided at the orchestrator as f64. The
bytes_sent counter on the worker is what the tests assert.
GROUP BY: an in-place hash table
Section titled “GROUP BY: an in-place hash table”HASH_ACC hashes the key register into a flat open-addressing table of
64-byte GroupSlots — one cache line each: key, state, four accumulators,
four validity bytes — in the connection’s pre-allocated scratch. Slot 0 is
reserved for the NULL-key group. The table accumulates across every block
of the frame and the occupied slots go out verbatim as
RESULT_GROUPED_AGG_CHUNKs. A full table (4,096 keys) is refused, never
merged. Proven across three containers with SUM, COUNT, MAX and
AVG per key and a NULL-key group, one chunk for 101 groups.
Joins: the two-stage pipeline
Section titled “Joins: the two-stage pipeline”A SELECT … FROM a JOIN b ON … [WHERE …] whose tables live on different
clusters enters the pipeline where the dispatcher used to refuse it.
Stage 0 — measure, don’t guess
Section titled “Stage 0 — measure, don’t guess”The catalog carries no row counts, so the pipeline ships COUNT(*) with
each side’s pushed-down filters — 80 bytes a side, microseconds — and the
smaller filtered side becomes the build side. Legacy engines pick the build
side from stale statistics because their network overhead is too high to
ask; the 80-byte aggregate makes asking practically free.
For a LEFT JOIN the preserved (FROM) side is always the probe side, so
Stage 0 is skipped.
Stage 1 — build
Section titled “Stage 1 — build”The smaller side’s worker runs a materialize program that yields its key
columns, every build-side column the query touches (in the SELECT list or a
cross-side conjunct), and each row’s offset (WITH_OFFSETS). The
orchestrator builds a fat 1:N table:
dir_slots | record_count | stride (24-byte preamble)directory: KeyOffset { composite_key, start << 32 | count } (open addressing)records: [build_row_offset, valid_mask, key cols…, payload cols…]Keys are the composite mix of the key columns (hash_table::mix, the same
fold the VM’s HASH_MIX performs); duplicate keys become one directory
entry pointing at a contiguous run of records, so a foreign key with N
build rows yields N pairs. Rows with a NULL key never enter the table. Both
halves are 8-byte words viewed in place on the worker, never parsed.
Stage 2 — broadcast and probe
Section titled “Stage 2 — broadcast and probe”The table goes to the larger side’s worker as HASH_TABLE_LOAD and is
acknowledged before the probe program is sent. The program mixes the probe
key columns, probes, verifies every key column against the record
(LOAD_BUILD + EQ — a 64-bit collision cannot yield a false pair),
evaluates the cross-side conjuncts as post-probe bytecode, and materialises
the SELECT list from both sides. Result rows follow the SELECT list; a
LEFT JOIN miss yields the row with the build columns NULL.
What was proven from SQL text, across two workers
Section titled “What was proven from SQL text, across two workers”| Query | Exercises |
|---|---|
SELECT o.amount FROM orders o JOIN customers c ON o.cid = c.id WHERE c.tier > 1 AND o.amount > 900 | filters pushed to each side, 1:N keys (30,000 × 2,000 rows) |
SELECT o.amount, c.tier … WHERE o.amount + c.tier > 1000 | build-side projection from the embedded record, cross-side arithmetic post-probe |
… ON o.cid = c.id AND o.amount = c.tier | composite key, mixed and verified |
SELECT o.cid, c.tier FROM orders o LEFT JOIN customers c ON o.cid = c.id WHERE o.amount > 990 | NULL-extended rows for a third of the orders |
… WHERE (o.amount * 2 - c.tier) / 3 > 660 | the full ALU across sides |
Every row set is compared to a Rust oracle, and the counters for the table load, the broadcast and the JOIN-loop run each move by one.
The coordinator-side cross join still exists
Section titled “The coordinator-side cross join still exists”Before joins compiled to bytecode, the orchestrator matched two workers’
offset streams itself (sorted merge or hash probe) and fetched survivors.
That path remains and was made streaming in the same arc: its
MatchedPair arena is now a chunk buffer, sorted_merge and probe_hash
are resumable cursors, each chunk’s survivors are fetched before the
matcher resumes, and a remote sink sends JOIN_STREAM_CHUNKs by viewing
the pairs as bytes — zero heap allocations per flush, asserted with a
counting allocator over a real socket.
Semantics worth knowing
Section titled “Semantics worth knowing”WHEREconjuncts that reference only one side run on that side before the probe; conjuncts that reference both run after the hit. ForLEFT JOIN, a miss skips the cross-side conjuncts and yields the row — the directive’s chosen semantics, recorded because standard SQL would drop the NULL-extended row when such a conjunct is NULL.- Joins are INNER or LEFT, two tables, an
ANDof equalities.RIGHT,FULL, non-equi and three-way joins are refused with the reason. - Integer columns only in the VM.