DDIR: half_join, on the corgi backend

[frankmcsherry]

Half joins for DDIR, on the corgi backend, following dogs3 and Materialize's delta joins.
Builds on #938, whose MergeLadder the batcher uses.

Surface

stream | half_join(other, strict, (k, v, v1) -> (k', v'))

• The stream's value must be (v, time), as lift makes it; time is a time as data.
• Each update at its own time t meets other's rows at its key whose times t2 precede t in the total order on times: strictly when strict is 1, or also equal when 0.
• Each pair gives (k' ; (v', time ∨ t2)) at time t, with the diffs multiplied. The lambda sees (k, v, v1) and no times.
• other's changes produce no output.
• Install checks the shapes: equal keys, and a stream time of the scope's width.

A delta join is two half joins and a delay:

let ab = a | lift | half_join(b, 0, (k, x, y) -> (k, (x, y)));
let ba = b | lift | half_join(a, 1, (k, y, x) -> (k, (x, y)));
export "delta" = (ab + ba) | delay((_, (_, t)) -> t) | key((k, (xy, _)) -> (k, xy));

Commits

1. corgi: a half join batcher and tactic (corgi/half_join/), unused until the next commit.
2. DDIR: half_join, on the corgi backend: the node, parser, shapes, backend wiring, vec and explain panics, and the dataflow tests.
3. interactive: a triangles benchmark: binary joins against a delta query.

Implementation

The operator is dogs3's half_join_with_tactic, which owns progress, capabilities and yielding.
corgi/half_join/ supplies the two pieces it takes, which meet only at Run.

HalfJoinBatcher (batcher.rs):

• Arrivals wait in a MergeLadder of chains in time order, as TimeChunks: rows (t, key, (v, s)), sorted and consolidated by corgi (sort_blocks, merged with survey_groups).
• An update is unblocked once its time is before the arrangement frontier's least element in the total order, or equal to it when strict.
• Release takes whole chunks from the front of each chain and a prefix of at most one more; arrivals already unblocked skip the ladder.
• The reported bound is the minimal times kept, in the partial order, cached per chunk and recomputed only for a chunk partly released.
• What one extraction releases leaves as Runs of at most 64K rows, each sorted by identifier within one window of one source; each source is dropped once its runs are formed.

CorgiHalfJoin (tactic.rs):

• Each block takes the rows of every run below one identifier horizon (as corgi/join.rs does), so runs are read front to back.
• Each block meets every arranged chunk from where the previous block left it, so a chunk is walked at most once per unit, and chunks whose identifiers miss the block are skipped.
• Pairs pass a real-key check when keys are hashed and the lexicographic time comparison, then the compiled projection, time ∨ t2 as a lane-wise max, and grouping by the capability covering each row.

backend/corgi.rs places stream updates with CorgiPact, retimes imported batches as RetimeJoin does, yields after 1M records as Materialize does, and compacts the arrangement to the stream frontier stepped back one in each coordinate (saturating).

The vec backend panics on half_join, and explain does too, as for lift.

Also: a comment in parse/mod.rs cited a validate_lift_iter check that does not exist; it now says nothing checks that discipline.

Measurements

cargo run --release --example triangles_bench -- [join|delta] 1000000 100000 1000 20 [workers]: 1M random edges over 100K nodes, then 20 ticks each replacing 1,000 edges, with codegen-units = 1 and runs interleaved.

	load	churn / tick	memory after load	peak heap
1 worker, binary joins	1.76–1.84 s	15.5–16.8 ms	2584 MiB	not measured
1 worker, delta query	1.86–1.88 s	7.5–7.8 ms	395 MiB	240 MiB
4 workers, binary joins	0.80 s	5.06 ms	2045 MiB	not measured
4 workers, delta query	0.60 s	3.47 ms	552–561 MiB	420 MiB

Memory after load is the process's footprint.
The binary-join rows come from an earlier round of runs than the delta-query rows.

Against this PR's first batcher and tactic, the delta query is about 10% slower to load at 1 worker, at par at 4 workers, 31% faster per tick at 4 workers, and lower in peak heap at both.
At 1 worker about 0.5 s of the load sorts each block's rows from its runs, and about 0.3 s forms the runs.

Tests

• tests/half_join.rs, at 1 and 3 workers, after each of 24 ticks of pseudo-random additions and retractions: the delta join above equals a | join(b, …), and reachability computed by half joins in an iterative scope equals reachability by join.
• The batcher against a model over 40 seeds of random arrivals and an advancing frontier, strict and not: each extraction releases exactly the unblocked updates, in identifier order, and reports a bound that covers every kept update and holds only kept times.
• The tactic against the definition over 40 seeds, with several arranged chunks, updates in several runs spanning many blocks, and integer and hashed keys.
• Two fixed cases: an arranged row later in one coordinate but earlier in the total order meets; the batcher releases and bounds by the frontier's least element, incomparable times included.
• Under test a block is 3 rows, a run 4, and a time chunk 5, so the tests span many blocks, runs, and chunks.

Mutations that each break a test: both half joins strict; ignoring the time order; dropping the time join; flipping strictness in the batcher, for arrivals or for chunks; a seek that skips a key straddling two blocks; a merge that drops one side's diff; a bound from first chunks only, or stale after a partial release; a wrong chunk skip; a lost released chunk or arrival; a wrong split of an arrival; a lossy time lane; data or update times read from the wrong run; a skewed block end.
Compacting the arrangement to the stream frontier rather than a step behind it makes the iterative reach test hang.
A run out of identifier order also hangs the merge, so a debug assertion checks each run's order.

cargo test --workspace passes, and the workspace builds without warnings.

Not here

• Each extraction releases everything it unblocks. Bounding that, so that a bounded number of chunks is active, needs the dogs3 driver to reactivate after an extraction that releases something.
• Released runs are not consolidated across chains, or with arrivals released on arrival.
• Nothing guards a half join used outside the lift … delay pattern.

🤖 Generated with Claude Code