From 9099aad9727b6e834e4d7c874967021f4e747194 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 15 Oct 2026 23:08:37 -0400
Subject: [PATCH 1/3] corgi: a half join batcher and tactic
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`HalfJoinBatcher` holds a lifted stream's updates in a `MergeLadder` of chains in time order, sorted and consolidated by corgi.
An update is released once the arrangement's frontier unblocks it: its time is before the 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 part of at most one more, and arrivals already unblocked skip the ladder.
The bound the batcher reports is the minimal times of what it keeps, in the partial order, cached per chunk.
What one extraction releases leaves as runs of at most 64K rows, each sorted by identifier within one window of its source.

`CorgiHalfJoin` merges a unit's runs a block at a time, below one identifier horizon per block, so each run is 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.
A pair meets when the arranged time precedes the update's in the total order, and gives the projection of `(key, val, v1)` with the data time `s ∨ t2`, at the update's time and capability.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 interactive/Cargo.toml                     |   1 +
 interactive/src/corgi/chunk.rs             |   2 +-
 interactive/src/corgi/col_times.rs         |   5 +
 interactive/src/corgi/half_join/batcher.rs | 387 +++++++++++++++++++++
 interactive/src/corgi/half_join/mod.rs     | 157 +++++++++
 interactive/src/corgi/half_join/tactic.rs  | 349 +++++++++++++++++++
 interactive/src/corgi/mod.rs               |   5 +-
 7 files changed, 903 insertions(+), 3 deletions(-)
 create mode 100644 interactive/src/corgi/half_join/batcher.rs
 create mode 100644 interactive/src/corgi/half_join/mod.rs
 create mode 100644 interactive/src/corgi/half_join/tactic.rs

diff --git a/interactive/Cargo.toml b/interactive/Cargo.toml
index a6d3781f4..705fa0a8b 100644
--- a/interactive/Cargo.toml
+++ b/interactive/Cargo.toml
@@ -16,6 +16,7 @@ columnar = { workspace = true }
 # The columnar kernels for the interpreted backend, pinned by git rev.
 corgi = { git = "https://github.com/frankmcsherry/WIP", rev = "41e873124a850772a9c4f7df88be94e3163512ef", features = ["serde"] }
 differential-dataflow = { workspace = true }
+differential-dogs3 = { path = "../dogsdogsdogs", default-features = false }
 serde = { version = "1.0", features = ["derive"] }
 smallvec = "1.15.1"
 timely = { workspace = true }
diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs
index eb50b47d9..bb59da9e9 100644
--- a/interactive/src/corgi/chunk.rs
+++ b/interactive/src/corgi/chunk.rs
@@ -640,7 +640,7 @@ pub fn recover_key_ref(keys: &CValue) -> &CValue {
 }
 
 /// Concatenate column blocks into one column (multi-source `gather_lanes`, no sort).
-fn concat_blocks(blocks: &[CValue]) -> CValue {
+pub(crate) fn concat_blocks(blocks: &[CValue]) -> CValue {
     if blocks.len() == 1 {
         return blocks[0].clone();
     }
diff --git a/interactive/src/corgi/col_times.rs b/interactive/src/corgi/col_times.rs
index d5613c7aa..d32f0359d 100644
--- a/interactive/src/corgi/col_times.rs
+++ b/interactive/src/corgi/col_times.rs
@@ -209,6 +209,11 @@ impl<T> ColTimes<T> {
         Ordering::Equal
     }
 
+    /// Whether this column's row `i` is less than or equal to `other`'s row `j` in the partial order.
+    pub fn less_equal_cross(&self, i: usize, other: &ColTimes<T>, j: usize) -> bool {
+        (0..self.lanes.len().max(other.lanes.len())).all(|k| self.at(k, i) <= other.at(k, j))
+    }
+
     /// Whether row `i` is less than or equal to row `j` in the partial order.
     fn less_equal(&self, i: usize, j: usize) -> bool {
         self.lanes.iter().all(|lane| lane[i] <= lane[j])
diff --git a/interactive/src/corgi/half_join/batcher.rs b/interactive/src/corgi/half_join/batcher.rs
new file mode 100644
index 000000000..11c6a8454
--- /dev/null
+++ b/interactive/src/corgi/half_join/batcher.rs
@@ -0,0 +1,387 @@
+//! The half join's batcher: arrivals wait in a merge ladder of chains in time order, and what
+//! the frontier unblocks leaves the fronts of the chains as runs in identifier order.
+
+use std::borrow::Cow;
+use std::cmp::Ordering;
+
+use timely::progress::Antichain;
+use timely::progress::frontier::AntichainRef;
+
+use differential_dataflow::batcher::Batcher;
+use differential_dataflow::batcher::merge::{MergeLadder, Merger};
+
+use corgi::arrange::{gather, gather_lanes, sort_blocks, survey_groups, GroupRun};
+use corgi::Value as CValue;
+
+use crate::corgi::chunk::concat_blocks;
+use crate::corgi::col_times::{ColTimes, Lanes};
+use crate::ir::{Diff, Time};
+
+use super::{gather_rows, Part, PartTimes, Run, CC};
+
+/// The most rows in one [`Run`]. Small under test, so that units hold many runs.
+const RUN: usize = if cfg!(test) { 4 } else { 1 << 16 };
+
+/// The most rows a merge puts in one [`TimeChunk`]. Releasing part of a chunk recomputes its
+/// lower bound, which costs its length. Small under test, so that chains hold many chunks.
+const TIME_TARGET: usize = if cfg!(test) { 5 } else { 1 << 16 };
+
+/// Updates in the total order on their times: rows `(t, key, val)`, where `t` is the update's own
+/// time as a `Prod` of `Int` lanes, sorted in corgi's structural order and consolidated. The
+/// structural order on the lanes is lexicographic, which is the total order on times.
+pub struct TimeChunk {
+    /// `Prod([t, key, val])`.
+    rows: CValue,
+    diffs: Vec<Diff>,
+    /// Rows before `start` have been released.
+    start: usize,
+    /// The minimal times among the rows from `start` on, in the partial order.
+    lower: Antichain<Time>,
+}
+
+impl Default for TimeChunk {
+    fn default() -> Self { TimeChunk { rows: CValue::Unit(0), diffs: Vec::new(), start: 0, lower: Antichain::new() } }
+}
+
+impl TimeChunk {
+    fn new(rows: CValue, diffs: Vec<Diff>) -> Self {
+        let mut chunk = TimeChunk { rows, diffs, start: 0, lower: Antichain::new() };
+        chunk.set_lower();
+        chunk
+    }
+
+    /// Sorts and consolidates rows `(t, key, val)`.
+    fn form(rows: CValue, diffs: Vec<Diff>) -> Option<Self> {
+        let (perm, labels) = sort_blocks(&[], &rows);
+        let (mut keep, mut kept) = (Vec::new(), Vec::new());
+        let mut i = 0;
+        while i < perm.len() {
+            let mut j = i + 1;
+            while j < perm.len() && labels[j] == labels[i] { j += 1; }
+            let d: Diff = perm[i..j].iter().map(|&p| diffs[p]).sum();
+            if d != 0 {
+                keep.push(perm[i]);
+                kept.push(d);
+            }
+            i = j;
+        }
+        if kept.is_empty() { return None; }
+        Some(TimeChunk::new(gather(&rows, &keep), kept))
+    }
+
+    /// The number of rows not yet released.
+    fn len(&self) -> usize { self.diffs.len() - self.start }
+
+    /// The time lanes, as integers.
+    fn lanes(&self) -> Vec<Cow<'_, [i64]>> { time_lanes(&self.rows) }
+
+    /// Recompute `lower` from the rows not yet released. Equal times are adjacent, so each distinct
+    /// time is offered once.
+    fn set_lower(&mut self) {
+        let lanes = self.lanes();
+        let mut lower = Antichain::new();
+        for i in self.start..self.diffs.len() {
+            if i > self.start && lanes.iter().all(|lane| lane[i - 1] == lane[i]) { continue; }
+            lower.insert(Time::from_coordinates(lanes.iter().map(|lane| lane[i] as u64)));
+        }
+        drop(lanes);
+        self.lower = lower;
+    }
+
+    /// The end of the rows `bound` unblocks: those before the bound in the total order, and those
+    /// equal to it when `strict`. No bound unblocks every row.
+    fn unblocked(&self, bound: Option<&[i64]>, strict: bool) -> usize {
+        let Some(bound) = bound else { return self.diffs.len() };
+        let lanes = self.lanes();
+        let (mut lo, mut hi) = (self.start, self.diffs.len());
+        while lo < hi {
+            let mid = (lo + hi) / 2;
+            if unblocked_at(&lanes, mid, bound, strict) { lo = mid + 1 } else { hi = mid }
+        }
+        lo
+    }
+
+    /// Release the rows up to `end`, which no longer count towards `lower`.
+    fn release_to(&mut self, end: usize) -> TimeChunk {
+        let rows: Vec<usize> = (self.start..end).collect();
+        let released = TimeChunk { rows: gather(&self.rows, &rows), diffs: self.diffs[self.start..end].to_vec(), start: 0, lower: Antichain::new() };
+        self.start = end;
+        self.set_lower();
+        released
+    }
+}
+
+/// Staged updates as rows `(t, key, val)`, their times written as `width` lanes, and their diffs.
+fn stage_rows(stage: Vec<CC>, width: usize) -> Option<(CValue, Vec<Diff>)> {
+    if stage.is_empty() { return None; }
+    assert!(stage.iter().all(|c| c.times.width() <= width), "half_join: a time wider than its scope");
+    let keys = concat_blocks(&stage.iter().map(|c| c.keys.clone()).collect::<Vec<_>>());
+    let vals = concat_blocks(&stage.iter().map(|c| c.vals.clone()).collect::<Vec<_>>());
+    let lanes = (0..width).map(|k| CValue::i64(stage.iter().flat_map(|c| c.times.lane(k)).map(|x| x as i64).collect())).collect();
+    let diffs: Vec<Diff> = stage.iter().flat_map(|c| c.diffs.iter().copied()).collect();
+    Some((CValue::Prod(vec![CValue::Prod(lanes), keys, vals]), diffs))
+}
+
+/// The time lanes of rows `(t, key, val)`, as integers.
+fn time_lanes(rows: &CValue) -> Vec<Cow<'_, [i64]>> {
+    let CValue::Prod(fields) = rows else { panic!("half_join: rows that are not (t, key, val)") };
+    let CValue::Prod(lanes) = &fields[0] else { panic!("half_join: a time that is not a tuple") };
+    lanes.iter().map(|lane| lane.as_i64("half_join time lane").unwrap()).collect()
+}
+
+/// Whether row `i`'s time, as `lanes`, is unblocked by `bound`: before it in the total order, or
+/// equal to it when `strict`.
+fn unblocked_at(lanes: &[Cow<'_, [i64]>], i: usize, bound: &[i64], strict: bool) -> bool {
+    match lanes.iter().zip(bound).map(|(lane, b)| lane[i].cmp(b)).find(|o| o.is_ne()).unwrap_or(Ordering::Equal) {
+        Ordering::Less => true,
+        Ordering::Equal => strict,
+        Ordering::Greater => false,
+    }
+}
+
+/// Rows `start..` of each part `(rows, start, diffs)`, in order, as one column and their diffs.
+fn concat_parts(parts: &[(&CValue, usize, &[Diff])]) -> (CValue, Vec<Diff>) {
+    if let [(rows, 0, diffs)] = parts { return ((*rows).clone(), diffs.to_vec()); }
+    let sources: Vec<Option<&CValue>> = parts.iter().map(|p| Some(p.0)).collect();
+    let (mut tags, mut offs, mut diffs) = (Vec::new(), Vec::new(), Vec::new());
+    for (tag, (_, start, part)) in parts.iter().enumerate() {
+        tags.resize(tags.len() + (part.len() - start), tag);
+        offs.extend(*start..part.len());
+        diffs.extend_from_slice(&part[*start..]);
+    }
+    (gather_lanes(&sources, &tags, &offs), diffs)
+}
+
+/// The unreleased rows of `chunks`, in order, as one column and their diffs.
+fn live_rows(chunks: &[TimeChunk]) -> (CValue, Vec<Diff>) {
+    concat_parts(&chunks.iter().map(|c| (&c.rows, c.start, &c.diffs[..])).collect::<Vec<_>>())
+}
+
+/// Merges chains of [`TimeChunk`]s, consolidating equal rows.
+#[derive(Default)]
+pub struct TimeMerger;
+
+impl Merger for TimeMerger {
+    type Chunk = TimeChunk;
+
+    fn merge(&mut self, list1: Vec<TimeChunk>, list2: Vec<TimeChunk>, output: &mut Vec<TimeChunk>, _stash: &mut Vec<TimeChunk>) {
+        let (a, ad) = live_rows(&list1);
+        let (b, bd) = live_rows(&list2);
+        let (mut tags, mut offs, mut diffs) = (Vec::new(), Vec::new(), Vec::new());
+        for run in survey_groups(&a, &b) {
+            match run {
+                GroupRun::A(lo, hi) => {
+                    tags.resize(tags.len() + (hi - lo), 0);
+                    offs.extend(lo..hi);
+                    diffs.extend_from_slice(&ad[lo..hi]);
+                }
+                GroupRun::B(lo, hi) => {
+                    tags.resize(tags.len() + (hi - lo), 1);
+                    offs.extend(lo..hi);
+                    diffs.extend_from_slice(&bd[lo..hi]);
+                }
+                GroupRun::Both(alo, ahi, blo, bhi) => {
+                    let d: Diff = ad[alo..ahi].iter().chain(&bd[blo..bhi]).sum();
+                    if d != 0 {
+                        tags.push(0);
+                        offs.push(alo);
+                        diffs.push(d);
+                    }
+                }
+            }
+        }
+        for lo in (0..diffs.len()).step_by(TIME_TARGET) {
+            let hi = (lo + TIME_TARGET).min(diffs.len());
+            output.push(TimeChunk::new(gather_lanes(&[Some(&a), Some(&b)], &tags[lo..hi], &offs[lo..hi]), diffs[lo..hi].to_vec()));
+        }
+    }
+
+    fn len(chunk: &TimeChunk) -> usize { chunk.len() }
+}
+
+/// Stages a lifted stream's updates and releases those the arrangement's frontier has unblocked.
+///
+/// An update at `t` is unblocked once no arrangement time that precedes it in the total order can
+/// still arrive: the frontier's least element `b` bounds every time yet to come, so `t ≤ b` suffices
+/// when only strictly earlier times meet `t`, and `t < b` when equal ones do too. An empty frontier
+/// unblocks everything.
+///
+/// Updates wait in a merge ladder of chains in time order, so what a frontier releases from each
+/// chain is a prefix: whole chunks from its front, and part of at most one more. Arrivals the
+/// frontier already unblocks skip the ladder. What one extraction releases is put in identifier
+/// order, ready to meet the arrangement in its order.
+pub struct HalfJoinBatcher {
+    strict: bool,
+    /// The number of lanes in a time: the scope depth, plus one.
+    width: usize,
+    /// Arrivals not yet sorted.
+    stage: Vec<CC>,
+    ladder: MergeLadder<TimeMerger>,
+    /// The minimal times among retained updates.
+    lower: Antichain<Time>,
+}
+
+impl HalfJoinBatcher {
+    pub fn new(strict: bool, width: usize) -> Self {
+        HalfJoinBatcher { strict, width, stage: Vec::new(), ladder: MergeLadder::new(None, 0), lower: Antichain::new() }
+    }
+}
+
+impl Batcher<CC> for HalfJoinBatcher {
+    type Time = Time;
+    type Output = Vec<Run>;
+
+    fn insert(&mut self, container: &mut CC) {
+        if !container.times.is_empty() {
+            self.stage.push(std::mem::take(container));
+        }
+    }
+
+    fn extract<'a>(&'a mut self, frontier: AntichainRef<'_, Time>) -> (Option<Vec<Run>>, AntichainRef<'a, Time>) {
+        let width = self.width;
+        let least = frontier.iter().min();
+        let bound: Option<Vec<i64>> = least.map(|b| {
+            let mut lanes: Vec<i64> = b.coordinates().map(|x| x as i64).collect();
+            assert!(lanes.len() <= width, "half_join: a frontier wider than its scope");
+            lanes.resize(width, 0);
+            lanes
+        });
+        let bound_col: Option<ColTimes<Time>> = least.map(|b| std::iter::once(b.clone()).collect());
+        let strict = self.strict;
+
+        // Arrivals the bound already unblocks are released without entering the ladder.
+        let (mut staged, mut blocked) = (Vec::new(), Vec::new());
+        for c in std::mem::take(&mut self.stage) {
+            let Some(b) = &bound_col else { staged.push(c); continue };
+            let (now, later): (Vec<usize>, Vec<usize>) = (0..c.times.len()).partition(|&i| match c.times.cmp_cross(i, b, 0) {
+                Ordering::Less => true,
+                Ordering::Equal => strict,
+                Ordering::Greater => false,
+            });
+            if later.is_empty() { staged.push(c); }
+            else if now.is_empty() { blocked.push(c); }
+            else {
+                staged.push(gather_rows(&c, &now));
+                blocked.push(gather_rows(&c, &later));
+            }
+        }
+        if let Some(chunk) = stage_rows(blocked, width).and_then(|(rows, diffs)| TimeChunk::form(rows, diffs)) {
+            self.ladder.insert_chain(vec![chunk]);
+        }
+
+        let mut released = Vec::new();
+        self.ladder.retain_chains(|chain| {
+            let whole = chain.iter().take_while(|c| c.unblocked(bound.as_deref(), strict) == c.diffs.len()).count();
+            released.extend(chain.drain(..whole));
+            if let Some(front) = chain.first_mut() {
+                let end = front.unblocked(bound.as_deref(), strict);
+                if end > front.start { released.push(front.release_to(end)); }
+            }
+        });
+        self.lower = Antichain::new();
+        for chunk in self.ladder.chains().flatten() {
+            for t in chunk.lower.elements() { self.lower.insert(t.clone()); }
+        }
+        let released = (!released.is_empty() || !staged.is_empty()).then(|| into_runs(released, staged, width));
+        (released, self.lower.borrow())
+    }
+}
+
+/// Released rows, from chunks and from arrivals released on arrival, as runs in identifier order.
+/// Each source is dropped once its runs are formed.
+fn into_runs(released: Vec<TimeChunk>, staged: Vec<CC>, width: usize) -> Vec<Run> {
+    let mut runs = Vec::new();
+    for chunk in released {
+        let CValue::Prod(fields) = &chunk.rows else { panic!("half_join: a time chunk that is not (t, key, val)") };
+        Part { keys: fields[1].clone(), vals: &fields[2], times: PartTimes::Lanes(chunk.lanes()), diffs: &chunk.diffs, rows: chunk.start..chunk.diffs.len() }.runs(RUN, width, &mut runs);
+    }
+    for c in staged {
+        Part { keys: c.keys.clone(), vals: &c.vals, times: PartTimes::Col(&c.times), diffs: &c.diffs, rows: 0..c.diffs.len() }.runs(RUN, width, &mut runs);
+    }
+    runs
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use super::super::test_util::*;
+    use crate::corgi::chunk::{ident, recover_key};
+    use crate::corgi::container::CorgiContainer;
+    use crate::ir::Value as V;
+
+    /// The batcher releases what the frontier's least element unblocks, and bounds what it keeps.
+    #[test]
+    fn releases_by_the_least_frontier_element() {
+        let released = |strict: bool, frontier: &[Time]| -> (Vec<Time>, Vec<Time>) {
+            let mut batcher = HalfJoinBatcher::new(strict, 2);
+            let times = [t(1, 0), t(0, 5), t(0, 2)];
+            batcher.insert(&mut container(&[1, 2, 3], &[0, 0, 0], &[(0, 0); 3], &times, &[1, 1, 1]));
+            let frontier: Antichain<Time> = frontier.iter().cloned().collect();
+            let (out, lower) = batcher.extract(frontier.borrow());
+            let mut out: Vec<Time> = out.unwrap_or_default().iter().flat_map(|c| c.times().to_vec()).collect();
+            out.sort();
+            let mut lower = lower.to_vec();
+            lower.sort();
+            (out, lower)
+        };
+        assert_eq!(released(true, &[t(1, 0)]), (vec![t(0, 2), t(0, 5), t(1, 0)], vec![]));
+        assert_eq!(released(false, &[t(1, 0)]), (vec![t(0, 2), t(0, 5)], vec![t(1, 0)]));
+        // (0, [5]) and (1, [0]) are incomparable, so both bound what is kept.
+        assert_eq!(released(false, &[t(0, 3), t(2, 0)]), (vec![t(0, 2)], vec![t(0, 5), t(1, 0)]));
+        assert_eq!(released(true, &[]), (vec![t(0, 2), t(0, 5), t(1, 0)], vec![]));
+    }
+
+    /// Against a model that holds each update until the frontier unblocks it: over rounds of random
+    /// arrivals, some already unblocked, and an advancing frontier, each extraction releases exactly
+    /// the unblocked updates, in identifier order, and reports a bound that covers every update it
+    /// keeps and holds only times it keeps.
+    #[test]
+    fn releases_what_the_model_releases() {
+        use std::collections::BTreeMap;
+        for strict in [false, true] {
+            for seed in 0..40 {
+                let mut rng = Rng(seed);
+                let mut batcher = HalfJoinBatcher::new(strict, 2);
+                let mut pending: BTreeMap<(Time, i64, i64), Diff> = BTreeMap::new();
+                for round in 0..14u64 {
+                    for _ in 0..rng.below(4) {
+                        let n = rng.below(9) as usize;
+                        let keys: Vec<i64> = (0..n).map(|_| rng.below(4) as i64).collect();
+                        let vals: Vec<i64> = (0..n).map(|_| rng.below(3) as i64).collect();
+                        let times: Vec<Time> = (0..n).map(|_| t(round / 3 + rng.below(3), rng.below(3))).collect();
+                        let diffs: Vec<Diff> = (0..n).map(|_| rng.diff()).collect();
+                        for i in 0..n { *pending.entry((times[i].clone(), keys[i], vals[i])).or_default() += diffs[i]; }
+                        batcher.insert(&mut container(&keys, &vals, &vec![(0, 0); n], &times, &diffs));
+                    }
+                    let frontier: Antichain<Time> = if round == 13 { Antichain::new() } else { [t(round / 3, round % 3), t(round / 3 + 1, 0)].into_iter().collect() };
+                    let bound = frontier.iter().min().cloned();
+                    let unblocked = |time: &Time| bound.as_ref().is_none_or(|b| time < b || (strict && time == b));
+
+                    let (out, lower) = batcher.extract(frontier.borrow());
+                    let lower = lower.to_owned();
+                    let mut got = Vec::new();
+                    for chunk in out.unwrap_or_default() {
+                        assert!(ident(chunk.keys()).windows(2).all(|w| w[0] <= w[1]), "released updates are in identifier order");
+                        let c = CorgiContainer { keys: recover_key(chunk.keys()), vals: chunk.vals().clone(), times: chunk.times().clone(), diffs: chunk.diffs().to_vec() };
+                        got.extend(c.into_updates());
+                    }
+                    let released: Vec<_> = pending.keys().filter(|(time, _, _)| unblocked(time)).cloned().collect();
+                    let mut expect = Vec::new();
+                    for entry in released {
+                        let d = pending.remove(&entry).unwrap();
+                        let (time, k, v) = entry;
+                        if d != 0 { expect.push(((V::Int(k), tup(vec![V::Int(v), tup(vec![V::Int(0), V::Int(0)])])), time, d)); }
+                    }
+                    assert_eq!(consolidated(got), consolidated(expect), "seed {seed}, round {round}, strict {strict}: the released updates");
+                    for ((time, _, _), d) in pending.iter() {
+                        assert!(*d == 0 || lower.less_equal(time), "seed {seed}, round {round}: the bound covers a kept update at {time:?}");
+                    }
+                    for b in lower.elements() {
+                        assert!(pending.keys().any(|(time, _, _)| time == b), "seed {seed}, round {round}: the bound holds {b:?}, which no kept update has");
+                    }
+                }
+                assert!(pending.values().all(|&d| d == 0), "an empty frontier releases everything");
+            }
+        }
+    }
+}
diff --git a/interactive/src/corgi/half_join/mod.rs b/interactive/src/corgi/half_join/mod.rs
new file mode 100644
index 000000000..bdc471739
--- /dev/null
+++ b/interactive/src/corgi/half_join/mod.rs
@@ -0,0 +1,157 @@
+//! The corgi half join: a [`Batcher`](differential_dataflow::batcher::Batcher) that releases a
+//! lifted stream's updates once the arrangement can no longer gain rows that precede them, and a
+//! [`HalfJoinTactic`](differential_dogs3::operators::half_join::HalfJoinTactic) that meets
+//! released updates with the arrangement's chunks, column at a time. Both plug into dogs3's
+//! `half_join_with_tactic`, which owns progress, capabilities, and yielding.
+//!
+//! An update is `(key ; (val, s))` at time `t`: `s` is a time as data, a `Prod` of `Int` lanes as
+//! `lift` writes it, and `t` the update's own time. It meets each arranged row `(key, v1)` at time
+//! `t2` with `t2 < t` in the total order on times (`≤` unless strict), giving
+//! `(key' ; (val', s ∨ t2))` at time `t` with the product of the diffs, where `(key', val')` is the
+//! projection over `(key, val, v1)`. The total order is the lexicographic order of the time lanes,
+//! which is DD's `Ord` on these times.
+//!
+//! The batcher keeps arrivals in time order and releases them as runs in identifier order; the
+//! tactic merges a unit's runs and meets them with the arrangement a block at a time. A [`Run`] is
+//! what passes between them.
+
+use std::borrow::Cow;
+use std::rc::Rc;
+
+use differential_dataflow::trace::chunk::ChunkBatch;
+
+use corgi::arrange::gather;
+use corgi::Value as CValue;
+
+use crate::corgi::chunk::{ident, present_key, CorgiChunk};
+use crate::corgi::col_times::ColTimes;
+use crate::corgi::container::CorgiContainer;
+use crate::ir::{Diff, Time};
+
+mod batcher;
+mod tactic;
+
+pub use batcher::HalfJoinBatcher;
+pub use tactic::CorgiHalfJoin;
+
+type CC = CorgiContainer<Time, Diff>;
+type NativeBatch = Rc<ChunkBatch<CorgiChunk<Time, Diff>>>;
+
+/// A run of released updates, as the batcher hands them to the tactic: in identifier order, as the
+/// arrangement's chunks are, with keys as [`present_key`] presents them. One extraction releases
+/// several runs, each sorted within a window of what was released, and the tactic merges them as
+/// it works. Runs are not consolidated: what one extraction releases from different chains may
+/// repeat a row.
+#[derive(Clone)]
+pub struct Run {
+    keys: CValue,
+    vals: CValue,
+    times: ColTimes<Time>,
+    diffs: Vec<Diff>,
+}
+
+impl Run {
+    /// Updates put in identifier order, as one run.
+    pub fn new(keys: CValue, vals: CValue, times: ColTimes<Time>, diffs: Vec<Diff>) -> Self {
+        let n = diffs.len();
+        let width = times.width();
+        let part = Part { keys, vals: &vals, times: PartTimes::Col(&times), diffs: &diffs, rows: 0..n };
+        let mut runs = Vec::new();
+        part.runs(n.max(1), width, &mut runs);
+        runs.pop().unwrap_or_else(|| Run { keys: present_key(part.keys.clone()), vals: vals.clone(), times: times.clone(), diffs: Vec::new() })
+    }
+    pub fn len(&self) -> usize { self.diffs.len() }
+    pub fn keys(&self) -> &CValue { &self.keys }
+    pub fn vals(&self) -> &CValue { &self.vals }
+    pub fn times(&self) -> &ColTimes<Time> { &self.times }
+    pub fn diffs(&self) -> &[Diff] { &self.diffs }
+}
+
+/// Rows `rows` of released columns, from a time chunk or from arrivals released on arrival.
+struct Part<'a> {
+    keys: CValue,
+    vals: &'a CValue,
+    times: PartTimes<'a>,
+    diffs: &'a [Diff],
+    rows: std::ops::Range<usize>,
+}
+
+impl Part<'_> {
+    /// The part's rows as runs of at most `size` rows, their times written as `width` lanes. Each
+    /// run is sorted and gathered from one window of the part, so its reads stay within it.
+    fn runs(&self, size: usize, width: usize, out: &mut Vec<Run>) {
+        let keys = present_key(self.keys.clone());
+        let ids = ident(&keys);
+        for lo in self.rows.clone().step_by(size) {
+            let hi = (lo + size).min(self.rows.end);
+            let mut order: Vec<(i64, usize)> = (lo..hi).map(|i| (ids[i], i)).collect();
+            order.sort_unstable_by_key(|&(id, _)| id);
+            let rows: Vec<usize> = order.into_iter().map(|(_, i)| i).collect();
+            let lanes = (0..width).map(|k| rows.iter().map(|&i| self.times.at(k, i)).collect()).collect();
+            out.push(Run {
+                keys: gather(&keys, &rows),
+                vals: gather(self.vals, &rows),
+                times: ColTimes::from_raw_lanes(lanes, rows.len()),
+                diffs: rows.iter().map(|&i| self.diffs[i]).collect(),
+            });
+        }
+    }
+}
+
+/// A part's times: a time chunk's lanes, or an arrival's column.
+enum PartTimes<'a> {
+    Lanes(Vec<Cow<'a, [i64]>>),
+    Col(&'a ColTimes<Time>),
+}
+
+impl PartTimes<'_> {
+    /// Coordinate `k` of row `i`, zero beyond the times' width.
+    fn at(&self, k: usize, i: usize) -> u64 {
+        match self {
+            PartTimes::Lanes(lanes) => lanes.get(k).map_or(0, |lane| lane[i] as u64),
+            PartTimes::Col(times) => times.raw_lanes().get(k).map_or(0, |lane| lane[i]),
+        }
+    }
+}
+/// The rows of `c` at `rows`, in that order.
+fn gather_rows(c: &CC, rows: &[usize]) -> CC {
+    CorgiContainer {
+        keys: gather(&c.keys, rows),
+        vals: gather(&c.vals, rows),
+        times: c.times.gather(rows),
+        diffs: rows.iter().map(|&i| c.diffs[i]).collect(),
+    }
+}
+
+#[cfg(test)]
+mod test_util {
+    use differential_dataflow::dynamic::pointstamp::PointStamp;
+    use timely::order::Product;
+
+    use super::*;
+    use crate::ir::Value as V;
+
+    /// The time `(outer, [inner])`, at scope depth one.
+    pub(super) fn t(outer: u64, inner: u64) -> Time { Product::new(outer, PointStamp::new(smallvec::smallvec![inner])) }
+    pub(super) fn ints(xs: &[i64]) -> CValue { CValue::i64(xs.to_vec()) }
+    pub(super) fn lifted(s: &[(i64, i64)]) -> CValue { CValue::Prod(vec![ints(&s.iter().map(|p| p.0).collect::<Vec<_>>()), ints(&s.iter().map(|p| p.1).collect::<Vec<_>>())]) }
+    pub(super) fn container(keys: &[i64], vals: &[i64], s: &[(i64, i64)], times: &[Time], diffs: &[Diff]) -> CC {
+        CorgiContainer { keys: ints(keys), vals: CValue::Prod(vec![ints(vals), lifted(s)]), times: times.iter().cloned().collect(), diffs: diffs.to_vec() }
+    }
+    pub(super) fn tup(xs: Vec<V>) -> V { V::Tuple(xs) }
+
+    /// A deterministic source of small numbers.
+    pub(super) struct Rng(pub(super) u64);
+    impl Rng {
+        pub(super) fn below(&mut self, n: u64) -> u64 {
+            self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
+            (self.0 >> 33) % n
+        }
+        pub(super) fn diff(&mut self) -> Diff { if self.below(3) == 0 { -1 } else { 1 } }
+    }
+
+    pub(super) fn consolidated(mut rows: Vec<((V, V), Time, Diff)>) -> Vec<((V, V), Time, Diff)> {
+        differential_dataflow::consolidation::consolidate_updates(&mut rows);
+        rows
+    }
+}
diff --git a/interactive/src/corgi/half_join/tactic.rs b/interactive/src/corgi/half_join/tactic.rs
new file mode 100644
index 000000000..15bace195
--- /dev/null
+++ b/interactive/src/corgi/half_join/tactic.rs
@@ -0,0 +1,349 @@
+//! The half join's tactic: a unit's runs, merged by identifier a block at a time, meet the
+//! arrangement's chunks, each read forward from where the last block left it.
+
+use std::borrow::Cow;
+use std::cell::RefCell;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::rc::Rc;
+
+use timely::progress::Antichain;
+
+use differential_dataflow::trace::chunk::Chunk;
+use differential_dogs3::operators::half_join::HalfJoinTactic;
+
+use corgi::arrange::{compare_at, gather_lanes};
+use corgi::{Graph, NumOp, Shape, Value as CValue};
+
+use crate::corgi::chunk::{ident, key_is_hashed, recover_key_ref, CorgiChunk};
+use crate::corgi::col_times::ColTimes;
+use crate::corgi::container::CorgiContainer;
+use crate::corgi::logic::compile_join_projection;
+use crate::corgi::search::matching_ranges;
+use crate::ir::{Diff, Term, Time};
+
+use super::{gather_rows, NativeBatch, Run, CC};
+
+/// Released updates are met with the arrangement this many at a time, so that the driver can yield
+/// between blocks. Small under test, so that tests span many blocks.
+const BLOCK: usize = if cfg!(test) { 3 } else { 1 << 14 };
+
+/// Meets released updates with an arrangement's chunks. The projection compiles once, at the
+/// shapes of the first block that meets anything, and is shared by every unit of work.
+pub struct CorgiHalfJoin {
+    key: Term,
+    val: Term,
+    strict: bool,
+    /// The number of lanes in a time as data: the scope depth, plus one.
+    width: usize,
+    compiled: Rc<RefCell<Option<([Shape; 3], Graph<NumOp>)>>>,
+}
+
+impl CorgiHalfJoin {
+    pub fn new(key: Term, val: Term, strict: bool, width: usize) -> Self {
+        CorgiHalfJoin { key, val, strict, width, compiled: Rc::new(RefCell::new(None)) }
+    }
+}
+
+impl HalfJoinTactic<Time, NativeBatch, Run, CC> for CorgiHalfJoin {
+    fn prep(&mut self, runs: Vec<Run>, batches: Vec<NativeBatch>, lower: Antichain<Time>) -> Box<dyn Iterator<Item = (CC, Time)>> {
+        let arranged: Vec<CorgiChunk<Time, Diff>> = batches.iter().flat_map(|b| b.chunks.iter().filter(|c| c.len() > 0).cloned()).collect();
+        let lower: Vec<Time> = lower.elements().to_vec();
+        debug_assert!(runs.iter().all(|r| ident(r.keys()).windows(2).all(|w| w[0] <= w[1])), "half_join: a run out of identifier order");
+        Box::new(Work {
+            starts: vec![0; runs.len()],
+            runs,
+            seek: vec![0; arranged.len()],
+            arranged,
+            lower_times: lower.iter().cloned().collect(),
+            lower,
+            key: self.key.clone(),
+            val: self.val.clone(),
+            strict: self.strict,
+            width: self.width,
+            compiled: Rc::clone(&self.compiled),
+            ready: VecDeque::new(),
+        })
+    }
+}
+
+/// One unit of half-join work: released runs, merged and met a block at a time.
+///
+/// Runs and arranged chunks are all in identifier order. Each block takes the rows of every run
+/// below one identifier horizon, so the runs are read front to back, and the blocks meet each
+/// arranged chunk at positions that only move forward: `seek` records where each arranged chunk's
+/// next search starts, and each is walked at most once per unit.
+struct Work {
+    runs: Vec<Run>,
+    /// For each run, the first row not yet met.
+    starts: Vec<usize>,
+    arranged: Vec<CorgiChunk<Time, Diff>>,
+    /// For each arranged chunk, the first row later blocks could meet.
+    seek: Vec<usize>,
+    /// The capabilities output may ship at, and the same as a column.
+    lower: Vec<Time>,
+    lower_times: ColTimes<Time>,
+    key: Term,
+    val: Term,
+    strict: bool,
+    width: usize,
+    compiled: Rc<RefCell<Option<([Shape; 3], Graph<NumOp>)>>>,
+    /// Output met but not yet handed over.
+    ready: VecDeque<(CC, Time)>,
+}
+
+impl Iterator for Work {
+    type Item = (CC, Time);
+
+    fn next(&mut self) -> Option<(CC, Time)> {
+        while self.ready.is_empty() {
+            let live = self.runs.iter().zip(&self.starts).filter(|(r, &s)| s < r.len()).count();
+            if live == 0 { return None; }
+            let ends = self.block_ends((BLOCK / live).max(1));
+            self.meet(&ends);
+            self.starts = ends;
+        }
+        self.ready.pop_front()
+    }
+}
+
+impl Work {
+    /// Where each run's next block ends: below the least identifier `budget` rows into a run with
+    /// more than that left, past that identifier's whole run of rows, so that no identifier is
+    /// split between blocks. Every run with no more than `budget` rows left runs to its end.
+    fn block_ends(&self, budget: usize) -> Vec<usize> {
+        let mut horizon: Option<i64> = None;
+        for (run, &s) in self.runs.iter().zip(&self.starts) {
+            let ids = ident(run.keys());
+            if ids.len() - s <= budget { continue; }
+            // `None` on overflow means nothing can exceed it: run to the end.
+            let Some(bound) = ids[s + budget].checked_add(1) else { continue };
+            horizon = Some(horizon.map_or(bound, |h| h.min(bound)));
+        }
+        self.runs.iter().map(|run| match horizon {
+            Some(h) => ident(run.keys()).partition_point(|&x| x < h),
+            None => run.len(),
+        }).collect()
+    }
+
+    /// Meet each run's rows from its start to `ends` with the arrangement, queueing the output by
+    /// capability.
+    fn meet(&mut self, ends: &[usize]) {
+        // The block's rows as (identifier, run, row), in identifier order; its distinct identifiers,
+        // ascending, and the rows holding each.
+        let mut rows: Vec<(i64, usize, usize)> = Vec::new();
+        for (r, (run, (&lo, &hi))) in self.runs.iter().zip(self.starts.iter().zip(ends)).enumerate() {
+            let ids = ident(run.keys());
+            rows.extend((lo..hi).map(|u| (ids[u], r, u)));
+        }
+        rows.sort_unstable_by_key(|&(id, _, _)| id);
+        let (mut needles, mut groups) = (Vec::new(), Vec::new());
+        for (pos, &(id, _, _)) in rows.iter().enumerate() {
+            if needles.last() != Some(&id) {
+                needles.push(id);
+                groups.push(pos..pos);
+            }
+            groups.last_mut().unwrap().end = pos + 1;
+        }
+        let (Some(&first), Some(&last)) = (needles.first(), needles.last()) else { return };
+        let hashed = key_is_hashed(self.runs[rows[0].1].keys());
+        let keys: Vec<&CValue> = self.runs.iter().map(|run| recover_key_ref(run.keys())).collect();
+
+        // The pairs that meet: (run, update row, arranged chunk, arranged row).
+        let (mut rs, mut us, mut cs, mut js) = (Vec::new(), Vec::new(), Vec::new(), Vec::new());
+        let (mut scratch, mut found) = (Vec::new(), Vec::new());
+        for (cid, chunk) in self.arranged.iter().enumerate() {
+            let hay = &ident(chunk.keys())[self.seek[cid]..];
+            let base = self.seek[cid];
+            // Later blocks hold identifiers at least `last`, so they need nothing before it.
+            self.seek[cid] += hay.partition_point(|&x| x < last);
+            if hay.first().is_none_or(|&x| x > last) || hay.last().is_some_and(|&x| x < first) { continue; }
+            matching_ranges(&needles, hay, &mut scratch, &mut found);
+            let chunk_key = recover_key_ref(chunk.keys());
+            for (needle, range) in found.iter() {
+                for &(_, r, u) in &rows[groups[*needle].clone()] {
+                    for j in range.start + base..range.end + base {
+                        if hashed && compare_at(keys[r], u, chunk_key, j).is_ne() { continue; }
+                        let precedes = match chunk.times().cmp_cross(j, self.runs[r].times(), u) {
+                            Ordering::Less => true,
+                            Ordering::Equal => !self.strict,
+                            Ordering::Greater => false,
+                        };
+                        if precedes {
+                            rs.push(r);
+                            us.push(u);
+                            cs.push(cid);
+                            js.push(j);
+                        }
+                    }
+                }
+            }
+        }
+        if us.is_empty() { return; }
+
+        // The projection over (key, val, arranged val).
+        let fields: Vec<&[CValue]> = self.runs.iter().map(|run| match run.vals() {
+            CValue::Prod(fields) => &fields[..],
+            _ => panic!("half_join: a stream value that is not (val, time)"),
+        }).collect();
+        let keys = gather_lanes(&keys.iter().map(|&k| Some(k)).collect::<Vec<_>>(), &rs, &us);
+        let vals = gather_lanes(&fields.iter().map(|f| Some(&f[0])).collect::<Vec<_>>(), &rs, &us);
+        let sources: Vec<Option<&CValue>> = self.arranged.iter().map(|c| Some(c.vals())).collect();
+        let arranged_vals = gather_lanes(&sources, &cs, &js);
+        let shapes = [corgi::shape_of_value(&keys), corgi::shape_of_value(&vals), corgi::shape_of_value(&arranged_vals)];
+        let mut compiled = self.compiled.borrow_mut();
+        if compiled.is_none() {
+            let g = compile_join_projection(&self.key, &self.val, &shapes[0], &shapes[1], &shapes[2])
+                .unwrap_or_else(|e| panic!("half_join: type error at shapes ({}, {}, {}): {e}", shapes[0], shapes[1], shapes[2]));
+            *compiled = Some((shapes.clone(), g));
+        }
+        let (pinned, graph) = compiled.as_ref().unwrap();
+        assert!(*pinned == shapes, "half_join: a block of shape {shapes:?} reached an operator pinned at {pinned:?}");
+        let mut out = corgi::eval_graph(graph, CValue::Prod(vec![keys, vals, arranged_vals])).into_prod("half_join projection").unwrap();
+        let out_val = out.pop().unwrap();
+        let out_key = out.pop().unwrap();
+
+        // The update's time as data, joined lane by lane with the arranged row's time.
+        let data_times: Vec<Vec<Cow<'_, [i64]>>> = fields.iter().map(|f| {
+            let CValue::Prod(lanes) = &f[1] else { panic!("half_join: a time as data that is not a tuple") };
+            assert_eq!(lanes.len(), self.width, "half_join: a time as data of the wrong width");
+            lanes.iter().map(|lane| lane.as_i64("half_join time lane").unwrap()).collect()
+        }).collect();
+        let joined: Vec<CValue> = (0..self.width).map(|l| {
+            CValue::i64(rs.iter().zip(&us).zip(cs.iter().zip(&js)).map(|((&r, &u), (&c, &j))| {
+                let t2 = self.arranged[c].times().raw_lanes().get(l).map_or(0, |lane| lane[j]) as i64;
+                data_times[r][l][u].max(t2)
+            }).collect())
+        }).collect();
+        let mut times = ColTimes::new();
+        for (&r, &u) in rs.iter().zip(&us) { times.push_ref(self.runs[r].times(), u); }
+        let met = CorgiContainer {
+            keys: out_key,
+            vals: CValue::Prod(vec![out_val, CValue::Prod(joined)]),
+            times,
+            diffs: rs.iter().zip(&us).zip(cs.iter().zip(&js)).map(|((&r, &u), (&c, &j))| self.runs[r].diffs()[u] * self.arranged[c].diffs()[j]).collect(),
+        };
+
+        // Each row ships at the first capability at or before its own time, as the driver requires.
+        let mut by_cap: Vec<Vec<usize>> = vec![Vec::new(); self.lower.len()];
+        for row in 0..met.times.len() {
+            let cap = (0..self.lower.len()).find(|&c| self.lower_times.less_equal_cross(c, &met.times, row))
+                .expect("half_join: no capability covers a released update");
+            by_cap[cap].push(row);
+        }
+        for (cap, rows) in by_cap.into_iter().enumerate() {
+            if !rows.is_empty() {
+                self.ready.push_back((gather_rows(&met, &rows), self.lower[cap].clone()));
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use super::super::test_util::*;
+    use crate::corgi::chunk::{columns_to_batch, present_key};
+    use crate::ir::Value as V;
+
+    /// An update meets the arranged rows before it in the total order on times, strictly or not:
+    /// never a later one, even one beyond it in a later coordinate alone. Its data time joins the
+    /// arranged row's, coordinate by coordinate, and each output ships at the capability covering it.
+    #[test]
+    fn meets_what_precedes_in_the_total_order() {
+        let arranged = Rc::new(columns_to_batch(
+            present_key(ints(&[1, 1, 1, 1, 1])),
+            ints(&[50, 100, 200, 300, 400]),
+            [t(0, 1), t(0, 5), t(1, 0), t(2, 0), t(1, 2)].into_iter().collect(),
+            vec![1, 1, 1, 1, 1],
+        ));
+        // Updates of key 1 at (1, [0]) and at (0, [3]), each with its own time as data.
+        let updates = container(&[1, 1], &[7, 8], &[(1, 0), (0, 3)], &[t(1, 0), t(0, 3)], &[2, 1]);
+        let updates = Run::new(updates.keys, updates.vals, updates.times, updates.diffs);
+        let lower: Antichain<Time> = [t(0, 3), t(1, 0)].into_iter().collect();
+        let projection = (Term::Var(0), Term::Tuple(vec![Term::Var(1), Term::Var(2)]));
+        let rows = |strict: bool| -> Vec<(Time, ((V, V), Time, Diff))> {
+            let mut tactic = CorgiHalfJoin::new(projection.0.clone(), projection.1.clone(), strict, 2);
+            let mut rows: Vec<_> = tactic.prep(vec![updates.clone()], vec![Rc::clone(&arranged)], lower.clone())
+                .flat_map(|(c, cap)| c.into_updates().into_iter().map(move |u| (cap.clone(), u)))
+                .collect();
+            rows.sort();
+            rows
+        };
+        let row = |cap: Time, x: i64, y: i64, data: (i64, i64), time: Time, diff: Diff| {
+            (cap, ((V::Int(1), tup(vec![tup(vec![V::Int(x), V::Int(y)]), tup(vec![V::Int(data.0), V::Int(data.1)])])), time, diff))
+        };
+        let mut strict = vec![
+            row(t(1, 0), 7, 50, (1, 1), t(1, 0), 2),
+            row(t(1, 0), 7, 100, (1, 5), t(1, 0), 2),
+            row(t(0, 3), 8, 50, (0, 3), t(0, 3), 1),
+        ];
+        strict.sort();
+        assert_eq!(rows(true), strict);
+        // Not strict, the update at (1, [0]) also meets the arranged row at (1, [0]).
+        let mut loose = strict.clone();
+        loose.push(row(t(1, 0), 7, 200, (1, 0), t(1, 0), 2));
+        loose.sort();
+        assert_eq!(rows(false), loose);
+    }
+
+    /// Against the definition, over several arranged chunks and released updates in several runs
+    /// spanning many blocks, with keys that are integers and keys that are hashed: each update
+    /// meets exactly the arranged rows of its key that precede it, and ships at a capability at or
+    /// before its time.
+    #[test]
+    fn meets_what_the_definition_meets() {
+        for hashed in [false, true] {
+            for strict in [false, true] {
+                for seed in 0..40 {
+                    let mut rng = Rng(seed);
+                    let key_col = |ks: &[i64]| if hashed { CValue::Prod(vec![ints(ks), ints(&ks.iter().map(|k| k * 7).collect::<Vec<_>>())]) } else { ints(ks) };
+                    let key_val = |k: i64| if hashed { tup(vec![V::Int(k), V::Int(k * 7)]) } else { V::Int(k) };
+                    let mut arranged_rows = Vec::new();
+                    let batches: Vec<NativeBatch> = (0..3).map(|_| {
+                        let n = 1 + rng.below(20) as usize;
+                        let ks: Vec<i64> = (0..n).map(|_| rng.below(6) as i64).collect();
+                        let vs: Vec<i64> = (0..n).map(|_| rng.below(100) as i64).collect();
+                        let ts: Vec<Time> = (0..n).map(|_| t(rng.below(4), rng.below(3))).collect();
+                        let ds: Vec<Diff> = (0..n).map(|_| rng.diff()).collect();
+                        for i in 0..n { arranged_rows.push((ks[i], vs[i], ts[i].clone(), ds[i])); }
+                        Rc::new(columns_to_batch(present_key(key_col(&ks)), ints(&vs), ts.into_iter().collect(), ds))
+                    }).collect();
+                    let n = 1 + rng.below(40) as usize;
+                    let ks: Vec<i64> = (0..n).map(|_| rng.below(8) as i64).collect();
+                    let vs: Vec<i64> = (0..n).map(|_| rng.below(10) as i64).collect();
+                    let ss: Vec<(i64, i64)> = (0..n).map(|_| (rng.below(4) as i64, rng.below(3) as i64)).collect();
+                    let ts: Vec<Time> = (0..n).map(|_| t(rng.below(4), rng.below(3))).collect();
+                    let ds: Vec<Diff> = (0..n).map(|_| rng.diff()).collect();
+                    // The updates as several runs, of random lengths, as one extraction releases them.
+                    let mut runs = Vec::new();
+                    let mut lo = 0;
+                    while lo < n {
+                        let hi = (lo + 1 + rng.below(6) as usize).min(n);
+                        runs.push(Run::new(key_col(&ks[lo..hi]), CValue::Prod(vec![ints(&vs[lo..hi]), lifted(&ss[lo..hi])]), ts[lo..hi].iter().cloned().collect(), ds[lo..hi].to_vec()));
+                        lo = hi;
+                    }
+                    let lower: Antichain<Time> = ts.iter().cloned().collect();
+
+                    let mut expect = Vec::new();
+                    for i in 0..n {
+                        for (k2, v1, t2, d2) in arranged_rows.iter() {
+                            if *k2 != ks[i] || !(*t2 < ts[i] || (!strict && *t2 == ts[i])) { continue; }
+                            let s = tup(vec![V::Int(ss[i].0.max(t2.outer as i64)), V::Int(ss[i].1.max(t2.inner.first().copied().unwrap_or(0) as i64))]);
+                            expect.push(((key_val(ks[i]), tup(vec![tup(vec![V::Int(vs[i]), V::Int(*v1)]), s])), ts[i].clone(), ds[i] * d2));
+                        }
+                    }
+                    let mut tactic = CorgiHalfJoin::new(Term::Var(0), Term::Tuple(vec![Term::Var(1), Term::Var(2)]), strict, 2);
+                    let mut got = Vec::new();
+                    for (c, cap) in tactic.prep(runs, batches, lower) {
+                        for row in c.into_updates() {
+                            assert!(timely::PartialOrder::less_equal(&cap, &row.1), "a row ships at a capability at or before its time");
+                            got.push(row);
+                        }
+                    }
+                    assert_eq!(consolidated(got), consolidated(expect), "seed {seed}, strict {strict}, hashed {hashed}");
+                }
+            }
+        }
+    }
+}
diff --git a/interactive/src/corgi/mod.rs b/interactive/src/corgi/mod.rs
index 8f862e4ed..2d773fbbd 100644
--- a/interactive/src/corgi/mod.rs
+++ b/interactive/src/corgi/mod.rs
@@ -1,8 +1,8 @@
 //! The corgi rendering substrate's internals: the columnar container on dataflow
 //! edges ([`container`]) and its wire format ([`bytes`]), the columnar chunk and
 //! its arrangement plumbing ([`chunk`]), the key-hash shuffle ([`exchange`]),
-//! the DDIR-term compiler ([`logic`]), and the join/reduce tactic bindings ([`join`],
-//! [`reduce`]). The `Backend` impl wiring these into rendering is
+//! the DDIR-term compiler ([`logic`]), and the join/half-join/reduce tactic bindings ([`join`],
+//! [`half_join`], [`reduce`]). The `Backend` impl wiring these into rendering is
 //! [`crate::backend::corgi`].
 
 pub mod bytes;
@@ -10,6 +10,7 @@ pub mod chunk;
 pub mod col_times;
 pub mod container;
 pub mod exchange;
+pub mod half_join;
 pub mod join;
 pub mod logic;
 pub mod reduce;

From 2ff0abe6cf6f7d3dad9945a439dfbbe13f5544ab Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 15 Oct 2026 23:08:49 -0400
Subject: [PATCH 2/3] DDIR: half_join, on the corgi backend
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`stream | half_join(other, strict, (k, v, v1) -> (k', v'))` meets each update of a lifted stream, `(k ; (v, time))` at its own time `t`, with `other`'s rows at its key whose times 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 product of the diffs; the lambda sees no times, and `other`'s changes produce nothing.

The operator is dogs3's `half_join_with_tactic`, which owns progress, capabilities and yielding, with the corgi batcher and tactic.
The arrangement compacts to the stream frontier stepped back one in each coordinate.
The vec backend and explain do not support it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 interactive/src/backend/corgi.rs | 30 +++++++++++
 interactive/src/backend/mod.rs   |  9 ++++
 interactive/src/backend/vec.rs   |  3 ++
 interactive/src/explain/mod.rs   |  3 ++
 interactive/src/lower.rs         | 10 +++-
 interactive/src/parse/mod.rs     |  5 +-
 interactive/src/parse/pipe.rs    | 17 ++++++
 interactive/src/scope_ir.rs      | 10 ++++
 interactive/src/shapes.rs        | 12 +++++
 interactive/tests/half_join.rs   | 89 ++++++++++++++++++++++++++++++++
 10 files changed, 184 insertions(+), 4 deletions(-)
 create mode 100644 interactive/tests/half_join.rs

diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 62a527391..53d56a498 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -32,7 +32,9 @@ use crate::corgi::chunk::{recover_key, CorgiChunk, CorgiChunker};
 use crate::corgi::col_times::ColTime;
 use crate::corgi::container::CorgiContainer;
 use crate::corgi::exchange::CorgiPact;
+use crate::corgi::half_join::{CorgiHalfJoin, HalfJoinBatcher};
 use crate::corgi::join::CorgiJoinBackend;
+use differential_dogs3::operators::half_join::{half_join_with_tactic, HalfJoinTactic};
 use crate::corgi::reduce::CorgiReduceBackend;
 use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic};
 use crate::corgi::logic::{compile_flatmap, compile_join_function, compile_predicate, compile_projection, compile_scalar, shape_of_row};
@@ -275,6 +277,23 @@ impl<B0: Native, B1: Native, J: JoinTactic<Time, NativeBatch, NativeBatch, CC>>
     }
 }
 
+/// A half-join tactic over native batches, accepting an imported arrangement.
+struct RetimeHalfJoin<J>(J);
+impl<B: Native, M, J: HalfJoinTactic<Time, NativeBatch, M, CC>> HalfJoinTactic<Time, B, M, CC> for RetimeHalfJoin<J> {
+    fn prep(&mut self, chunks: Vec<M>, batches: Vec<B>, lower: Antichain<Time>) -> Box<dyn Iterator<Item = (CC, Time)>> {
+        self.0.prep(chunks, batches.into_iter().map(Native::native).collect(), lower)
+    }
+}
+
+/// The time each coordinate of `t` less one, at least zero: the half join's compaction of its
+/// arrangement. Every later stream time is at or beyond some such frontier element, and advancing
+/// arranged times to it keeps their order against those stream times, strict and not.
+fn step_back(t: &Time, out: &mut Antichain<Time>) {
+    use differential_dataflow::dynamic::pointstamp::PointStamp;
+    let inner = PointStamp::new(t.inner.iter().map(|x| x.saturating_sub(1)).collect());
+    out.insert(timely::order::Product::new(t.outer.saturating_sub(1), inner));
+}
+
 /// A reduce tactic over native batches, accepting an imported input. (Its output is its own.)
 struct RetimeReduce<R>(R);
 impl<B1: Native, R: ReduceTactic<Time, NativeBatch, NativeBatch>> ReduceTactic<Time, B1, NativeBatch> for RetimeReduce<R> {
@@ -404,6 +423,17 @@ impl Backend for CorgiBackend {
         }
     }
 
+    fn half_join<'s>(stream: Collection<'s, Time, CC>, arrangement: Self::Arr<'s>, strict: bool, projection: &Projection, level: usize, _shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, CC> {
+        // dogs3's driver owns progress and yielding; corgi supplies when an update is unblocked
+        // (`HalfJoinBatcher`) and how released updates meet the arrangement (`CorgiHalfJoin`).
+        // `CorgiPact` places each update on the worker holding its key, as `arrange` placed the rows.
+        let tactic = || RetimeHalfJoin(CorgiHalfJoin::new(projection.key.clone(), projection.val.clone(), strict, level + 1));
+        let yielding = |_timer: std::time::Instant, count: usize| count > 1_000_000;
+        match arrangement {
+            Arr::Local(a) => half_join_with_tactic(stream.inner, a, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic()).as_collection(),
+            Arr::Imported(a) => half_join_with_tactic(stream.inner, a, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic()).as_collection(),
+        }
+    }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         // Amortize columnar corrections while reusing sweep scratch across wide presentations.
         let tactic = RetimeReduce(ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone())).with_key_batch_size(256));
diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs
index 28eaa8351..5c69fca01 100644
--- a/interactive/src/backend/mod.rs
+++ b/interactive/src/backend/mod.rs
@@ -49,6 +49,9 @@ pub trait Backend {
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, Self::Container>;
     /// `shapes` are the two inputs' row shapes, where they are known before any data arrives.
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container>;
+    /// See `scope_ir::Node::HalfJoin`. `level` is the scope depth, which fixes the width of a time
+    /// as data; `shapes` are the stream's and the arrangement's row shapes, where known.
+    fn half_join<'s>(stream: Collection<'s, Time, Self::Container>, arrangement: Self::Arr<'s>, strict: bool, projection: &Projection, level: usize, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container>;
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s>;
     fn inspect<'s>(c: Collection<'s, Time, Self::Container>, label: String) -> Collection<'s, Time, Self::Container>;
     fn leave_dynamic<'s>(c: Collection<'s, Time, Self::Container>, depth: usize) -> Collection<'s, Time, Self::Container>;
@@ -133,6 +136,7 @@ fn reads_rows(s: &st::Scope, k: usize) -> bool {
     let is_k = |r: &st::Ref| matches!(r, st::Ref::Import(i) if *i == k);
     s.items.iter().any(|item| match item {
         st::Item::Op(st::Node::Linear { input, .. } | st::Node::Inspect { input, .. }) => is_k(input),
+        st::Item::Op(st::Node::HalfJoin { stream, .. }) => is_k(stream),
         st::Item::Op(st::Node::Concat(refs)) => refs.iter().any(is_k),
         st::Item::Op(st::Node::Arrange(_) | st::Node::Join { .. } | st::Node::Reduce { .. }) => false,
         st::Item::Sub(_) => false,
@@ -189,6 +193,11 @@ pub fn render_tree<'s, B: Backend>(
                         let r = resolve(&items, &imports, &var_cols, right).arrange();
                         Rendered::Collection(B::join(l, r, projection, shape_of(left).zip(shape_of(right))))
                     },
+                    st::Node::HalfJoin { stream, arrangement, strict, projection } => {
+                        let c = resolve(&items, &imports, &var_cols, stream).collection();
+                        let a = resolve(&items, &imports, &var_cols, arrangement).arrange();
+                        Rendered::Collection(B::half_join(c, a, *strict, projection, depth, shape_of(stream).zip(shape_of(arrangement))))
+                    },
                     st::Node::Reduce { input, reducer } => {
                         let a = resolve(&items, &imports, &var_cols, input).arrange();
                         Rendered::Arrangement(B::reduce(a, reducer))
diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs
index 5dea98f7b..8a918f6b4 100644
--- a/interactive/src/backend/vec.rs
+++ b/interactive/src/backend/vec.rs
@@ -207,6 +207,9 @@ impl Backend for VecBackend {
             (Arr::Imported(l), Arr::Imported(r)) => l.join_core(r, move |k, v1, v2| f(k, v1, v2)),
         }
     }
+    fn half_join<'s>(_stream: Collection<'s, Time, Self::Container>, _arrangement: Self::Arr<'s>, _strict: bool, _projection: &Projection, _level: usize, _shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container> {
+        panic!("half_join runs on the corgi backend only")
+    }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         let f: Arc<dyn Fn(&Row, &[(&Row, Diff)], &mut Vec<(Row, Diff)>) + Send + Sync> = match reducer {
             Reducer::Min => Arc::new(|_key, vals, output| { if let Some(min) = vals.iter().map(|(v, _)| (*v).clone()).min() { output.push((min, 1)); } }),
diff --git a/interactive/src/explain/mod.rs b/interactive/src/explain/mod.rs
index 0bb2e56ac..5eb9e888f 100644
--- a/interactive/src/explain/mod.rs
+++ b/interactive/src/explain/mod.rs
@@ -77,6 +77,7 @@ fn clone_node(node: &Node, m: impl Fn(&Ref) -> Ref) -> Node {
         Node::Concat(refs) => Node::Concat(refs.iter().map(&m).collect()),
         Node::Arrange(r) => Node::Arrange(m(r)),
         Node::Join { left, right, projection } => Node::Join { left: m(left), right: m(right), projection: projection.clone() },
+        Node::HalfJoin { stream, arrangement, strict, projection } => Node::HalfJoin { stream: m(stream), arrangement: m(arrangement), strict: *strict, projection: projection.clone() },
         Node::Reduce { input, reducer } => Node::Reduce { input: m(input), reducer: reducer.clone() },
         Node::Inspect { input, label } => Node::Inspect { input: m(input), label: label.clone() },
     }
@@ -914,6 +915,8 @@ impl<'a> Reverse<'a> {
                 let contrib = ex.emit_lookup_keyed(dep_this, &side, out_user_len, reducer);
                 self.contribs.entry(target).or_default().push(contrib);
             }
+            // It reads its stream's whole times, as `lift` does.
+            Node::HalfJoin { .. } => panic!("explain: half_join is not supported"),
             Node::Join { left, right, projection } => {
                 let lt = resolve(self.orig, path, left);
                 let rt = resolve(self.orig, path, right);
diff --git a/interactive/src/lower.rs b/interactive/src/lower.rs
index 4ce009fbc..e9d295bfd 100644
--- a/interactive/src/lower.rs
+++ b/interactive/src/lower.rs
@@ -51,7 +51,7 @@ fn expr_free_names<'a>(expr: &'a Expr, out: &mut BTreeSet<&'a str>) {
             | Expr::Negate(e) | Expr::EnterAt(e, _) | Expr::LiftIter(e)
             | Expr::FlatMap(e, _) | Expr::Lift(e) | Expr::JoinFunction(e, _)
             | Expr::Inspect(e, _) | Expr::Arrange(e) => expr_free_names(e, out),
-        Expr::Join(l, r, _) => { expr_free_names(l, out); expr_free_names(r, out); },
+        Expr::Join(l, r, _) | Expr::HalfJoin(l, r, _, _) => { expr_free_names(l, out); expr_free_names(r, out); },
         Expr::Concat(es) => { for e in es { expr_free_names(e, out); } },
     }
 }
@@ -168,6 +168,12 @@ impl ScopeLower {
                 let rr = self.lower_expr(r); let ra = self.push(st::Node::Arrange(rr));
                 self.push(st::Node::Join { left: la, right: ra, projection: p.clone() })
             },
+            // A half join reads its stream as updates and only its second input arranged.
+            Expr::HalfJoin(s, r, strict, p) => {
+                let stream = self.lower_expr(s);
+                let rr = self.lower_expr(r); let ra = self.push(st::Node::Arrange(rr));
+                self.push(st::Node::HalfJoin { stream, arrangement: ra, strict: *strict, projection: p.clone() })
+            },
             Expr::Reduce(e, red)=> {
                 let r = self.lower_expr(e); let a = self.push(st::Node::Arrange(r));
                 self.push(st::Node::Reduce { input: a, reducer: red.clone() })
@@ -316,7 +322,7 @@ fn collect_qualified(e: &Expr, scope: &str, out: &mut Vec<String>) {
         Expr::TypedSource(e, _, _) | Expr::Map(e, _) | Expr::Filter(e, _) | Expr::Negate(e) | Expr::EnterAt(e, _)
         | Expr::FlatMap(e, _) | Expr::Lift(e) | Expr::JoinFunction(e, _)
         | Expr::LiftIter(e) | Expr::Reduce(e, _) | Expr::Inspect(e, _) | Expr::Arrange(e) => collect_qualified(e, scope, out),
-        Expr::Join(l, r, _) => { collect_qualified(l, scope, out); collect_qualified(r, scope, out); },
+        Expr::Join(l, r, _) | Expr::HalfJoin(l, r, _, _) => { collect_qualified(l, scope, out); collect_qualified(r, scope, out); },
         Expr::Concat(es) => for e in es { collect_qualified(e, scope, out); },
         Expr::Input(_) | Expr::Import(_) | Expr::Name(_) => {}
     }
diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs
index 7b7332c75..ddd60b357 100644
--- a/interactive/src/parse/mod.rs
+++ b/interactive/src/parse/mod.rs
@@ -25,6 +25,8 @@ pub enum Expr {
     Qualified(String, String),
     Map(Box<Expr>, Projection),
     Join(Box<Expr>, Box<Expr>, Projection),
+    /// `stream | half_join(arrangement, strict, projection)`: see `scope_ir::Node::HalfJoin`.
+    HalfJoin(Box<Expr>, Box<Expr>, bool, Projection),
     Reduce(Box<Expr>, Reducer),
     Filter(Box<Expr>, Term),
     Negate(Box<Expr>),
@@ -32,8 +34,7 @@ pub enum Expr {
     /// Append the current user-iter coord (at the operator's scope depth)
     /// to each row's value. Time itself is unchanged.
     ///
-    /// Discipline (post-lowering check, see `lower::validate_lift_iter`):
-    /// the result of `LiftIter` must not be referenced inside the same
+    /// Discipline, which nothing checks: the result of `LiftIter` should not be referenced inside the same
     /// scope it appears in — only from an enclosing scope, after the
     /// implicit leave. This preserves the "loop body is a time-invariant
     /// function" property; in-scope use risks defeating the fixpoint.
diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs
index 4af9fb6cf..cf69d0792 100644
--- a/interactive/src/parse/pipe.rs
+++ b/interactive/src/parse/pipe.rs
@@ -36,6 +36,10 @@
 //!   of the epoch and every scope coordinate.
 //! - `| join_function(term)` — replace each row by the `term`'s list of
 //!   `((key, val), time, diff)`, at the row's time joined with `time`, diffs multiplied.
+//! - `| half_join(other, strict, (k, v, v1) -> (k', v'))` — each update of a lifted stream,
+//!   `(k ; (v, time))`, meets `other`'s rows at its key whose times precede the update's own in the
+//!   total order on times (strictly when `strict` is 1, or also equal when 0), giving
+//!   `(k' ; (v', time ∨ t2))`. The lambda does not see the times. `other`'s changes produce nothing.
 //! - `| delay(term)` — `join_function` with the list `[((key, val), term, 1)]`: advance each
 //!   row to at least the time `term`.
 //! - `a + b` (concat), `a - b` (concat with `b` negated).
@@ -621,6 +625,19 @@ impl Parser {
             Token::EnterAt => { self.next(); self.expect(&Token::LParen); let f = self.parse_slot_term(); self.expect(&Token::RParen); Expr::EnterAt(Box::new(lhs), f) },
             Token::LiftIter => { self.next(); Expr::LiftIter(Box::new(lhs)) },
             Token::Ident(name) if name == "lift" => { self.next(); Expr::Lift(Box::new(lhs)) },
+            Token::Ident(name) if name == "half_join" => {
+                self.next(); self.expect(&Token::LParen);
+                let r = self.parse_join_arg(); self.expect(&Token::Comma);
+                let strict = match self.next() {
+                    Token::Int(0) => false,
+                    Token::Int(1) => true,
+                    other => panic!("half_join: strict is 0 or 1, not {other:?}"),
+                };
+                self.expect(&Token::Comma);
+                let p = match self.lambda(3) { Some(l) => Self::split(l), None => self.parse_projection() };
+                self.expect(&Token::RParen);
+                Expr::HalfJoin(Box::new(lhs), Box::new(r), strict, p)
+            },
             Token::Ident(name) if name == "join_function" => {
                 self.next(); self.expect(&Token::LParen);
                 let t = self.parse_slot_term(); self.expect(&Token::RParen);
diff --git a/interactive/src/scope_ir.rs b/interactive/src/scope_ir.rs
index 138393117..44e9f27d0 100644
--- a/interactive/src/scope_ir.rs
+++ b/interactive/src/scope_ir.rs
@@ -117,6 +117,13 @@ pub enum Node {
     Concat(Vec<Ref>),
     Arrange(Ref),
     Join { left: Ref, right: Ref, projection: Projection },
+    /// Each update of `stream`, `(key ; (val, time))` with `time` a time as data (as `lift` writes
+    /// it), meets the rows of `arrangement` at its key whose times are less than the update's own
+    /// in the total order on times, or less or equal unless `strict`. Each pair gives
+    /// `(key' ; (val', time ∨ t2))`: `projection` over `(key, val, arranged val)`, and the update's
+    /// data time joined with the arranged row's. The output keeps the update's own time, and its
+    /// diff is the product of the two. The arrangement's changes produce no output.
+    HalfJoin { stream: Ref, arrangement: Ref, strict: bool, projection: Projection },
     Reduce { input: Ref, reducer: Reducer },
     Inspect { input: Ref, label: String },
 }
@@ -208,6 +215,7 @@ impl Scope {
                     Node::Linear { input, .. } | Node::Arrange(input)
                     | Node::Reduce { input, .. } | Node::Inspect { input, .. } => f(input),
                     Node::Join { left, right, .. } => { f(left); f(right); },
+                    Node::HalfJoin { stream, arrangement, .. } => { f(stream); f(arrangement); },
                     Node::Concat(refs) => for r in refs { f(r); },
                 },
                 Item::Sub(child) => for imp in child.imports.iter_mut() {
@@ -239,6 +247,7 @@ impl Scope {
                     Node::Linear { input, .. } | Node::Arrange(input)
                     | Node::Reduce { input, .. } | Node::Inspect { input, .. } => tally(input),
                     Node::Join { left, right, .. } => { tally(left); tally(right); },
+                    Node::HalfJoin { stream, arrangement, .. } => { tally(stream); tally(arrangement); },
                     Node::Concat(refs) => for r in refs { tally(r); },
                 },
                 Item::Sub(child) => for imp in &child.imports {
@@ -387,6 +396,7 @@ fn dump_scope_body(s: &Scope, indent: usize) {
                     Node::Concat(refs) => refs.iter().map(fmt_ref).collect::<Vec<_>>().join(" + "),
                     Node::Arrange(r) => format!("{} | arrange", fmt_ref(r)),
                     Node::Join { left, right, .. } => format!("join({}, {})", fmt_ref(left), fmt_ref(right)),
+                    Node::HalfJoin { stream, arrangement, strict, .. } => format!("{} | half_join({}, {})", fmt_ref(stream), fmt_ref(arrangement), *strict as u8),
                     Node::Reduce { input, reducer } => format!("{} | {:?}", fmt_ref(input), reducer),
                     Node::Inspect { input, label } => format!("{} | inspect({})", fmt_ref(input), label),
                 };
diff --git a/interactive/src/shapes.rs b/interactive/src/shapes.rs
index b30dc310e..52f360eda 100644
--- a/interactive/src/shapes.rs
+++ b/interactive/src/shapes.rs
@@ -246,6 +246,18 @@ fn infer_node(node: &st::Node, shapes: &ScopeShapes, depth: usize, at: &str, pro
             let v = match shape_of_term(&projection.val, &env, None) { Ok(s) => s, Err(e) => return fail(format!("join val: {e}")) };
             Some(CollShape::new(k, v, depth))
         }
+        st::Node::HalfJoin { stream, arrangement, projection, .. } => {
+            let (s, a) = (shapes.of(stream)?, shapes.of(arrangement)?);
+            if s.key != a.key { return fail(format!("half_join keys {} and {}", s.key, a.key)); }
+            let v = match s.val {
+                Shape::Prod(fs) if fs.len() == 2 && fs[1] == time_shape(depth) => fs[0].clone(),
+                other => return fail(format!("half_join: a stream value of shape {other}, not (val, {}) as `lift` makes it", time_shape(depth))),
+            };
+            let env = [s.key, v, a.val];
+            let k = match shape_of_term(&projection.key, &env, None) { Ok(s) => s, Err(e) => return fail(format!("half_join key: {e}")) };
+            let v = match shape_of_term(&projection.val, &env, None) { Ok(s) => s, Err(e) => return fail(format!("half_join val: {e}")) };
+            Some(CollShape::new(k, Shape::Prod(vec![v, time_shape(depth)]), depth))
+        }
         st::Node::Reduce { input, reducer } => {
             let c = shapes.of(input)?;
             let v = match reducer {
diff --git a/interactive/tests/half_join.rs b/interactive/tests/half_join.rs
new file mode 100644
index 000000000..b181aa10e
--- /dev/null
+++ b/interactive/tests/half_join.rs
@@ -0,0 +1,89 @@
+//! `half_join`: a delta join, two half joins whose results are delayed to their data times, is the
+//! join, at the root and inside an iterative scope, as inputs change over several epochs.
+
+use interactive::ir::Value;
+use interactive::server::{RenderBackend, Server};
+use interactive::{lower, parse};
+
+const PROGRAM: &str = r#"
+    let a = input 0 : ((int, int) ; ()) | key(((k, x), ()) -> (k, x));
+    let b = input 1 : ((int, int) ; ()) | key(((k, y), ()) -> (k, y));
+
+    -- Each of `a`'s updates meets `b`'s at or before it, and each of `b`'s meets `a`'s strictly
+    -- before it, so every pair meets once, at the later of its two times.
+    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));
+    export "join" = a | join(b, (k, x, y) -> (k, (x, y)));
+
+    -- Reachability from `a`'s keys along `b`'s edges, joining by each method.
+    let roots = a | key((k, _) -> (k, ()));
+    joined: {
+        let step = reach | join(b, (_, (), dst) -> (dst, ()));
+        var reach = roots + step | distinct;
+    }
+    halved: {
+        let fwd = reach | lift | half_join(b, 0, (_, (), dst) -> (dst, ()));
+        let rev = b | lift | half_join(reach, 1, (_, dst, ()) -> (dst, ()));
+        let step = (fwd + rev) | delay((_, (_, t)) -> t) | key((n, _) -> (n, ()));
+        var reach = roots + step | distinct;
+    }
+    export "reach_join" = joined::reach;
+    export "reach_delta" = halved::reach;
+"#;
+
+const TICKS: u64 = 24;
+
+/// The rows fed at each tick, as `(input, key, value, diff)`: a few pseudo-random rows of small
+/// keys and values each tick, each added and, a few ticks later, retracted. Enough ticks that the
+/// arrangements' batches merge, and so compact.
+fn feeds(tick: u64) -> Vec<(usize, i64, i64, i64)> {
+    let row = |n: u64| {
+        let h = n.wrapping_mul(0x9E37_79B9_7F4A_7C15).rotate_left(17).wrapping_mul(0xBF58_476D_1CE4_E5B9);
+        ((h % 2) as usize, ((h >> 8) % 6) as i64, ((h >> 16) % 6) as i64)
+    };
+    let mut rows = Vec::new();
+    for i in 0..4 {
+        let (input, k, v) = row(4 * tick + i);
+        rows.push((input, k, v, 1));
+        if tick >= 3 {
+            let (input, k, v) = row(4 * (tick - 3) + i);
+            rows.push((input, k, v, -1));
+        }
+    }
+    rows
+}
+
+/// Every export after each tick, by name.
+fn run(workers: usize) -> Vec<[Vec<(Value, Value, i64)>; 4]> {
+    let guards = timely::execute(timely::Config::process(workers), move |worker| {
+        let mut server = Server::with_backend(RenderBackend::Corgi);
+        let mut program = lower::lower_tree(parse::pipe::parse(PROGRAM));
+        program.optimize();
+        server.install(worker, "p", &program).unwrap();
+        let mut seen = Vec::new();
+        for tick in 0..TICKS {
+            if worker.index() == 0 {
+                for (input, k, v, d) in feeds(tick) {
+                    server.feed("p", input, Value::Tuple(vec![Value::Int(k), Value::Int(v)]), Value::unit(), None, d).unwrap();
+                }
+            }
+            server.tick(worker);
+            seen.push(["delta", "join", "reach_delta", "reach_join"].map(|name| server.snapshot(worker, name).unwrap()));
+        }
+        seen
+    })
+    .unwrap();
+    guards.join().into_iter().next().unwrap().unwrap()
+}
+
+#[test]
+fn delta_joins_are_joins() {
+    for workers in [1, 3] {
+        for (tick, [delta, join, reach_delta, reach_join]) in run(workers).into_iter().enumerate() {
+            assert!(!join.is_empty() && !reach_join.is_empty(), "tick {tick}: the joins have rows");
+            assert_eq!(delta, join, "tick {tick}, {workers} workers: the delta join");
+            assert_eq!(reach_delta, reach_join, "tick {tick}, {workers} workers: reachability");
+        }
+    }
+}

From 044035ccb1a71a3d5bd36a40ecd7c6abb8077dec Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 15 Oct 2026 23:08:49 -0400
Subject: [PATCH 3/3] interactive: a triangles benchmark, by binary joins and
 by a delta query

`triangles_bench` counts triangles in a random graph by binary joins, which arrange the two-paths, or by a delta query of half joins, which arranges only the edges.
It reports the time and memory to load the graph and the time per tick as edges are replaced, and counts heap allocations to report the heap's peak.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 interactive/examples/triangles_bench.rs | 133 ++++++++++++++++++++++++
 1 file changed, 133 insertions(+)
 create mode 100644 interactive/examples/triangles_bench.rs

diff --git a/interactive/examples/triangles_bench.rs b/interactive/examples/triangles_bench.rs
new file mode 100644
index 000000000..94b7d5596
--- /dev/null
+++ b/interactive/examples/triangles_bench.rs
@@ -0,0 +1,133 @@
+//! Triangles in a random graph, by binary joins or by a delta query of half joins. The binary
+//! joins arrange the two-paths; the delta query arranges only the edges, by source, by
+//! destination, and by themselves. Reports the time and memory to load the graph, and then the
+//! time per tick as edges are replaced.
+//!
+//! ```text
+//! cargo run --release --example triangles_bench -- [join|delta] [edges] [nodes] [churn] [ticks] [workers]
+//! ```
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Instant;
+
+/// Counts live heap bytes and their peak, to tell the heap's size from the process's.
+struct Counting;
+static LIVE: AtomicUsize = AtomicUsize::new(0);
+static PEAK: AtomicUsize = AtomicUsize::new(0);
+unsafe impl GlobalAlloc for Counting {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        let live = LIVE.fetch_add(layout.size(), Ordering::Relaxed) + layout.size();
+        PEAK.fetch_max(live, Ordering::Relaxed);
+        unsafe { System.alloc(layout) }
+    }
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        LIVE.fetch_sub(layout.size(), Ordering::Relaxed);
+        unsafe { System.dealloc(ptr, layout) }
+    }
+}
+#[global_allocator]
+static ALLOC: Counting = Counting;
+
+fn heap() -> String {
+    format!("heap {:.1} MiB (peak {:.1} MiB)", LIVE.load(Ordering::Relaxed) as f64 / 1048576.0, PEAK.load(Ordering::Relaxed) as f64 / 1048576.0)
+}
+
+use interactive::server::{RenderBackend, Server};
+use interactive::{lower, parse};
+
+const EDGES: &str = r#"
+    let e = input 0 : ((int, int) ; ()) | key(((x, y), ()) -> (x, y));
+    let r = e | key((x, y) -> (y, x));
+    let s = e | key((x, y) -> ((x, y), ()));
+"#;
+
+/// Each triangle `(a, b, c)` has edges `(a, b)`, `(b, c)`, and `(a, c)`.
+const JOIN: &str = r#"
+    let paths = r | join(e, (b, a, c) -> ((a, c), b));
+    let tri = paths | join(s, ((a, c), b, ()) -> ((a, b, c), ()));
+    export "count" = tri | key((_, _) -> ((), ())) | count;
+"#;
+
+/// One path of half joins per edge of the triangle, each meeting the earlier edges' changes at or
+/// before its own time and the later edges' strictly before.
+const DELTA: &str = r#"
+    let d1 = r | lift | half_join(e, 1, (b, a, c) -> ((a, c), b)) | half_join(s, 1, ((a, c), b, ()) -> ((a, b, c), ()));
+    let d2 = e | lift | half_join(r, 0, (b, c, a) -> ((a, c), b)) | half_join(s, 1, ((a, c), b, ()) -> ((a, b, c), ()));
+    let d3 = e | lift | half_join(e, 0, (a, c, b) -> ((b, c), a)) | half_join(s, 0, ((b, c), a, ()) -> ((a, b, c), ()));
+    let tri = (d1 + d2 + d3) | delay((_, (_, t)) -> t) | key((abc, (v, _)) -> (abc, v));
+    export "count" = tri | key((_, _) -> ((), ())) | count;
+"#;
+
+/// This process's memory in MiB: on macOS its `footprint`, elsewhere its RSS.
+fn memory_mib() -> Option<f64> {
+    let out = std::process::Command::new("footprint").args(["-p", &std::process::id().to_string()]).output();
+    if let Ok(out) = out {
+        let text = String::from_utf8_lossy(&out.stdout).into_owned();
+        let mut words = text.split("Footprint:").nth(1)?.split_whitespace();
+        let n: f64 = words.next()?.parse().ok()?;
+        return match words.next()? {
+            "B" => Some(n / (1024.0 * 1024.0)),
+            "KB" => Some(n / 1024.0),
+            "MB" => Some(n),
+            "GB" => Some(n * 1024.0),
+            _ => None,
+        };
+    }
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    let kib: f64 = status.lines().find(|l| l.starts_with("VmRSS:"))?.split_whitespace().nth(1)?.parse().ok()?;
+    Some(kib / 1024.0)
+}
+
+fn mem() -> String {
+    memory_mib().map_or_else(|| "n/a".to_string(), |m| format!("{m:>7.1} MiB"))
+}
+
+fn main() {
+    let mut args = std::env::args().skip(1);
+    let method = args.next().unwrap_or("delta".into());
+    let nums: Vec<u64> = args.map(|a| a.parse().expect("integer argument")).collect();
+    let edges = nums.first().copied().unwrap_or(1_000_000);
+    let nodes = nums.get(1).copied().unwrap_or(100_000);
+    let churn = nums.get(2).copied().unwrap_or(1_000);
+    let ticks = nums.get(3).copied().unwrap_or(20);
+    let workers = nums.get(4).copied().unwrap_or(1) as usize;
+    let query = match method.as_str() {
+        "join" => JOIN,
+        "delta" => DELTA,
+        other => panic!("unknown method {other:?}: join or delta"),
+    };
+    let source = format!("{EDGES}{query}");
+
+    timely::execute(timely::Config::process(workers), move |worker| {
+        let index = worker.index();
+        let mut server = Server::with_backend(RenderBackend::Corgi);
+        let mut program = lower::lower_tree(parse::pipe::parse(&source));
+        program.optimize();
+        server.install(worker, "p", &program).unwrap();
+        server.load(worker, "p", 0, &format!("random:nodes={nodes},edges={edges},churn={churn}")).unwrap();
+        let loaded = Instant::now();
+        server.tick(worker);
+        let count = |server: &mut Server, worker: &mut timely::worker::Worker| -> i64 {
+            server.snapshot(worker, "count").unwrap().iter().map(|(_, v, d)| match v { interactive::ir::Value::Int(n) => n * d, _ => 0 }).sum()
+        };
+        // A snapshot is a dataflow, so every worker takes part; worker 0 receives the rows.
+        let elapsed = loaded.elapsed();
+        let n = count(&mut server, worker);
+        if index == 0 {
+            println!("{method}: {edges} edges over {nodes} nodes, {workers} workers");
+            println!("load:  {:>9.1} ms  mem {}  {}  count {n}", elapsed.as_secs_f64() * 1e3, mem(), heap());
+        }
+        let churned = Instant::now();
+        for _ in 0..ticks { server.tick(worker); }
+        let elapsed = churned.elapsed();
+        let n = count(&mut server, worker);
+        if index == 0 {
+            println!(
+                "churn: {:>9.2} ms/tick over {ticks} ticks ({churn} edges replaced per tick)  mem {}  count {n}",
+                elapsed.as_secs_f64() * 1e3 / ticks as f64, mem(),
+            );
+        }
+    })
+    .unwrap();
+}