From 2d11f8bb0559ead1cee6e9b36f7091c75a584164 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 19 Aug 2026 16:35:03 -0400
Subject: [PATCH 1/4] Preliminary HalfJoin tactic

---
 dogsdogsdogs/examples/delta_query2.rs   |   4 +-
 dogsdogsdogs/src/operators/count.rs     |  14 +-
 dogsdogsdogs/src/operators/half_join.rs | 606 ++++++++++++++++--------
 dogsdogsdogs/src/operators/propose.rs   |  14 +-
 dogsdogsdogs/src/operators/validate.rs  |  14 +-
 5 files changed, 419 insertions(+), 233 deletions(-)

diff --git a/dogsdogsdogs/examples/delta_query2.rs b/dogsdogsdogs/examples/delta_query2.rs
index f29a95a1a..b9c53d0a8 100644
--- a/dogsdogsdogs/examples/delta_query2.rs
+++ b/dogsdogsdogs/examples/delta_query2.rs
@@ -43,7 +43,7 @@ fn main() {
                     changes1,
                     forward2,
                     closure,
-                    |t1,t2| t1.lt(t2),  // This one ignores concurrent updates.
+                    true,   // This one ignores concurrent updates.
                     |key, val1, val2| (key.clone(), (val1.clone(), val2.clone())),
                 );
 
@@ -52,7 +52,7 @@ fn main() {
                     changes2,
                     forward1,
                     closure,
-                    |t1,t2| t1.le(t2),  // This one can "see" concurrent updates.
+                    false,  // This one can "see" concurrent updates.
                     |key, val1, val2| (key.clone(), (val2.clone(), val1.clone())),
                 );
 
diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs
index 8aa0eebc7..a40c0155a 100644
--- a/dogsdogsdogs/src/operators/count.rs
+++ b/dogsdogsdogs/src/operators/count.rs
@@ -64,15 +64,9 @@ where
     };
 
     use crate::operators::half_join::half_join_internal_unsafe as half_join_unsafe;
-    // Branch once here, so that each comparison monomorphizes rather than testing `strict` at
-    // every timestamp. The cost is instantiating `half_join` twice.
-    if strict {
-        half_join_unsafe::<_, _, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
-            requests, arrangement, frontier_func, |t1, t2| t1 < t2, |_timer, _count| false, output_func)
-    }
-    else {
-        half_join_unsafe::<_, _, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
-            requests, arrangement, frontier_func, |t1, t2| t1 <= t2, |_timer, _count| false, output_func)
-    }
+    // `strict` now reaches the join as a value rather than as a comparison closure, so there is
+    // nothing left to monomorphize by branching here; the test is made per arrangement time.
+    half_join_unsafe::<_, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
+        requests, arrangement, frontier_func, strict, |_timer, _count| false, output_func)
     .as_collection()
 }
diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
index 102a0f108..908cfff1d 100644
--- a/dogsdogsdogs/src/operators/half_join.rs
+++ b/dogsdogsdogs/src/operators/half_join.rs
@@ -1,8 +1,8 @@
 /// Streaming asymmetric join between updates (K,V1) and an arrangement on (K,V2).
 ///
 /// The asymmetry is that the join only responds to streamed updates, not to changes in the arrangement.
-/// Streamed updates join only with matching arranged updates at lesser times *in the total order*, and
-/// subject to a predicate supplied by the user (roughly: strictly less, or not).
+/// Streamed updates join only with matching arranged updates at lesser times *in the total order*,
+/// where a `strict` flag says whether an equal time counts as lesser.
 ///
 /// This behavior can ensure that any pair of matching updates interact exactly once.
 ///
@@ -20,20 +20,22 @@
 use std::collections::VecDeque;
 use std::ops::Mul;
 
-use timely::ContainerBuilder;
-use timely::container::CapacityContainerBuilder;
+use timely::{Container, ContainerBuilder};
+use timely::container::{CapacityContainerBuilder, DrainContainer, NoopBuilder, PushInto, SizableContainer};
 use timely::dataflow::Stream;
-use timely::dataflow::channels::pact::{Pipeline, Exchange};
+use timely::dataflow::channels::ContainerBytes;
+use timely::dataflow::channels::pact::{Pipeline, ExchangeCore};
 use timely::dataflow::operators::Operator;
+use timely::dataflow::operators::generic::OutputBuilderSession;
 use timely::PartialOrder;
-use timely::progress::{Antichain, ChangeBatch, Timestamp};
+use timely::progress::{Antichain, Timestamp};
 use timely::progress::frontier::MutableAntichain;
 
 use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable};
-use differential_dataflow::difference::{Monoid, Semigroup};
+use differential_dataflow::difference::Semigroup;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchTimeGat, BatchVal, Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::{BatchReader, BatchCursor, BatchDiff, BatchVal, Cursor, Navigable, TraceReader};
 use differential_dataflow::trace::cursor::cursor_list;
 use differential_dataflow::consolidation::{consolidate, consolidate_updates};
 use differential_dataflow::trace::implementations::BatchContainer;
@@ -59,26 +61,29 @@ use timely::dataflow::operators::CapabilitySet;
 /// This last constraint is important to ensure that we correctly produce
 /// all pairs of output updates across multiple `half_join` operators.
 ///
+/// The `strict` argument selects the comparison: `true` requires `time2` to be strictly less than
+/// `initial_time`, and `false` also admits `time2` equal to it. A delta query pairs a strict operator
+/// with a non-strict one, which is what makes each pair of matching updates interact exactly once.
+///
 /// Notice that the time is hoisted up into data. The expectation is that
 /// once out of the "delta flow region", the updates will be `delay`d to the
 /// times specified in the payloads.
-pub fn half_join<'scope, K, V, R, Tr, FF, CF, DOut, S>(
+pub fn half_join<'scope, K, V, R, Tr, FF, DOut, S>(
     stream: VecCollection<'scope, Tr::Time, (K, V, Tr::Time), R>,
     arrangement: Arranged<'scope, Tr>,
     frontier_func: FF,
-    comparison: CF,
+    strict: bool,
     mut output_func: S,
 ) -> VecCollection<'scope, Tr::Time, (DOut, Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
 where
     K: Hashable + ExchangeData,
     V: ExchangeData,
-    R: ExchangeData + Monoid,
-    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    R: ExchangeData + Semigroup,
+    Tr: TraceReader<Batch: Navigable>+Clone+'static,
     BatchCursor<Tr>: Cursor<Time = Tr::Time>,
     <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
     R: Mul<BatchDiff<Tr>, Output: Semigroup>,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
     DOut: Clone+'static,
     S: FnMut(&K, &V, BatchVal<'_, Tr>)->DOut+'static,
 {
@@ -90,7 +95,7 @@ where
             builder.push_into((dout, initial.clone(), diff));
         }
     };
-    half_join_internal_unsafe::<_, _, _, _, _, _,_,_, CapacityContainerBuilder<Vec<_>>>(stream, arrangement, frontier_func, comparison, |_timer, _count| false, output_func)
+    half_join_internal_unsafe::<_, _, _, _, _, _, _, CapacityContainerBuilder<Vec<_>>>(stream, arrangement, frontier_func, strict, |_timer, _count| false, output_func)
         .as_collection()
 }
 
@@ -115,254 +120,453 @@ where
 /// for each `((key, val2), time2, diff2)` present in `arrangement`, where
 /// `time2` is less than `initial_time` *UNDER THE TOTAL ORDER ON TIMES*.
 ///
+/// The `strict` argument selects the comparison: `true` requires `time2` to be strictly less than
+/// `initial_time`, and `false` also admits `time2` equal to it. A delta query pairs a strict operator
+/// with a non-strict one, which is what makes each pair of matching updates interact exactly once.
+///
 /// The `yield_function` allows the caller to indicate when the operator should
-/// yield control, as a function of the elapsed time and the number of matched
-/// records. Note this is not the number of *output* records, owing mainly to
-/// the number of matched records being easiest to record with low overhead.
-pub fn half_join_internal_unsafe<'scope, K, V, R, Tr, FF, CF, Y, S, CB>(
+/// yield control, as a function of the elapsed time and the number of records in
+/// the output containers shipped so far. Joined work is suspended at container
+/// boundaries, so the count advances a container at a time rather than a record
+/// at a time.
+pub fn half_join_internal_unsafe<'scope, K, V, R, Tr, FF, Y, S, CB>(
     stream: VecCollection<'scope, Tr::Time, (K, V, Tr::Time), R>,
-    mut arrangement: Arranged<'scope, Tr>,
+    arrangement: Arranged<'scope, Tr>,
     frontier_func: FF,
-    comparison: CF,
+    strict: bool,
     yield_function: Y,
-    mut output_func: S,
+    output_func: S,
 ) -> Stream<'scope, Tr::Time, CB::Container>
 where
     K: Hashable + ExchangeData,
     V: ExchangeData,
-    R: ExchangeData + Monoid,
-    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
+    R: ExchangeData + Semigroup,
+    Tr: TraceReader<Batch: Navigable>+Clone+'static,
     BatchCursor<Tr>: Cursor<Time = Tr::Time>,
     <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
     Y: Fn(std::time::Instant, usize) -> bool + 'static,
     S: FnMut(&mut CB, &K, &V, BatchVal<'_, Tr>, &Tr::Time, &R, &mut Vec<(Tr::Time, BatchDiff<Tr>)>) + 'static,
     CB: ContainerBuilder,
+{
+    // Updates are staged in a `BlobList` and joined by a cursor walk; the driver owns progress.
+    let batcher = BlobList::<(K, V, Tr::Time), Tr::Time, R>::default();
+    let tactic = cursors::CursorTactic::<K, V, R, Tr::Batch, S, CB>::new(output_func, strict);
+    let route = |update: &((K, V, Tr::Time), Tr::Time, R)| (update.0).0.hashed().into();
+    half_join_with_tactic(stream.inner, arrangement, route, frontier_func, strict, yield_function, batcher, tactic)
+}
+
+pub trait HalfJoinTactic<B: BatchReader, C0, C1> {
+    /// Converts a list of chunks and a list of batches to a list of outputs, each with a time suitable as a capability.
+    ///
+    /// The `lower` argument lower bounds the times in `chunks`, and may be used to load `batches`
+    /// compacted, or to bound the arrangement times the join must consider.
+    fn prep(&mut self, chunks: Vec<C0>, batches: Vec<B>, lower: Antichain<B::Time>) -> Box<dyn Iterator<Item = (C1, B::Time)>>;
+}
+
+/// A type capable of accepting containers of updates, and carving them out by time.
+///
+/// The implementor is able to determine the meaning of extraction by a frontier;
+/// it is not required to be by antichain partial order.
+pub trait BatcherAlt<C, T: Timestamp> {
+    /// Moves responsibility for `container` into the implementor.
+    fn insert(&mut self, container: C);
+    /// Extracts containers on the basis of `bound`, and the lower bounds of the times retained.
+    ///
+    /// The returned bound is reported at the one moment it is well defined: an implementor is not
+    /// required to account for inserted updates until it is asked to extract, so this is the only
+    /// bound a caller can safely downgrade capabilities to.
+    fn extract(&mut self, bound: Eligibility<T>) -> (Vec<C>, &MutableAntichain<T>);
+}
+
+pub enum Eligibility<T> {
+    All,
+    Le(T),
+    Lt(T),
+}
+
+struct BlobList<D, T: Timestamp, R> {
+    stage: Vec<(T, D, R)>,
+    blobs: Vec<VecDeque<(T, D, R)>>,
+    lower: MutableAntichain<T>,
+}
+
+impl<D, T: Timestamp, R> Default for BlobList<D, T, R> {
+    fn default() -> Self {
+        BlobList { stage: Vec::new(), blobs: Vec::new(), lower: MutableAntichain::new() }
+    }
+}
+
+impl<D: Ord, T: Timestamp, R: Semigroup> BatcherAlt<Vec<(D, T, R)>, T> for BlobList<D, T, R> {
+    fn insert(&mut self, container: Vec<(D, T, R)>) {
+        self.stage.extend(container.into_iter().map(|(d,t,r)| (t,d,r)));
+    }
+    fn extract(&mut self, bound: Eligibility<T>) -> (Vec<Vec<(D, T, R)>>, &MutableAntichain<T>) {
+        // Handle any staged updates first.
+        consolidate_updates(&mut self.stage);
+        if !self.stage.is_empty() {
+            let blob: VecDeque<_> = std::mem::take(&mut self.stage).into();
+            self.lower.update_iter(blob.iter().map(|x| (x.0.clone(), 1)));
+            self.blobs.push(blob);
+        }
+
+        let mut result = Vec::new();
+        for blob in self.blobs.iter_mut() {
+            let mut list = Vec::new();
+            match &bound {
+                Eligibility::Le(bound) => {
+                    while blob.front().map(|(t,_,_)| t.le(bound)).unwrap_or(false) {
+                        let (time, data, diff) = blob.pop_front().unwrap();
+                        list.push((data, time, diff));
+                    }
+                }
+                Eligibility::Lt(bound) => {
+                    while blob.front().map(|(t,_,_)| t.lt(bound)).unwrap_or(false) {
+                        let (time, data, diff) = blob.pop_front().unwrap();
+                        list.push((data, time, diff));
+                    }
+                }
+                Eligibility::All => {
+                    list.extend(blob.drain(..).map(|(t,d,r)| (d,t,r)));
+                }
+            }
+            if !list.is_empty() { result.push(list); }
+        }
+
+        self.blobs.retain(|b| !b.is_empty());
+
+        self.lower.update_iter(result.iter().flat_map(|l| l.iter().map(|x| (x.1.clone(), -1))));
+        (result, &self.lower)
+    }
+}
+
+/// A `half_join` driven by a [`BatcherAlt`] and a [`HalfJoinTactic`].
+///
+/// This is the analogue of [`half_join_internal_unsafe`], with the join itself lifted out. The
+/// driver owns progress and nothing else: it stages arriving updates in `batcher`, releases those
+/// the arrangement frontier has unblocked, and hands each release to `tactic` as a unit of deferred
+/// work paired with the capabilities required to ship its output. It holds no cursor, and has no
+/// opinion on how the join is performed.
+///
+/// The `strict` argument plays the role of the `comparison` closure: an update is unblocked once
+/// no arrangement time can still precede its initial time under the total order, and `strict` says
+/// whether an equal time counts as preceding. It reaches the join itself through `tactic`, which
+/// must be built with the same value.
+pub fn half_join_with_tactic<'scope, Tr, FF, RF, Y, Bat, Tac, CIn, C>(
+    stream: Stream<'scope, Tr::Time, CIn>,
+    mut arrangement: Arranged<'scope, Tr>,
+    route: RF,
+    frontier_func: FF,
+    strict: bool,
+    yield_function: Y,
+    mut batcher: Bat,
+    mut tactic: Tac,
+) -> Stream<'scope, Tr::Time, C>
+where
+    Tr: TraceReader + Clone + 'static,
+    FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
+    for<'a> RF: FnMut(&CIn::Item<'a>) -> u64 + 'static,
+    Y: Fn(std::time::Instant, usize) -> bool + 'static,
+    Bat: BatcherAlt<CIn, Tr::Time> + 'static,
+    Tac: HalfJoinTactic<Tr::Batch, CIn, C> + 'static,
+    CIn: Container + SizableContainer + DrainContainer + ContainerBytes + Send,
+    for<'a> CIn: PushInto<<CIn as DrainContainer>::Item<'a>>,
+    C: Container + 'static,
 {
     // No need to block physical merging for this operator.
     arrangement.trace.set_physical_compaction(Antichain::new().borrow());
     let mut arrangement_trace = Some(arrangement.trace);
     let arrangement_stream = arrangement.stream;
 
-    let exchange = Exchange::new(move |update: &((K, V, Tr::Time),Tr::Time,R)| (update.0).0.hashed().into());
-
-    // Stash for (time, diff) accumulation.
-    let mut output_buffer = Vec::new();
-
-    // Unified blobs: each blob holds data in (T, D, R) order, with a stuck_count
-    // tracking how many elements at the back are not yet eligible for processing.
-    // The ready prefix is sorted by (D, T, R) for cursor traversal.
-    let mut blobs: Vec<Blob<(K, V, Tr::Time), Tr::Time, R>> = Vec::new();
+    let exchange = ExchangeCore::<CapacityContainerBuilder<CIn>, _>::new(route);
 
     let scope = stream.scope();
-    stream.inner.binary_frontier(arrangement_stream, exchange, Pipeline, "HalfJoin", move |_,info| {
+    stream.binary_frontier(arrangement_stream, exchange, Pipeline, "HalfJoin", move |_, info| {
 
         // Acquire an activator to reschedule the operator when it has unfinished work.
         let activator = scope.activator_for(info.address);
 
+        // Capabilities covering everything `batcher` retains. Downgraded after each extraction to
+        // the bound it reports, and cloned into each unit of deferred work before that.
+        let mut caps = CapabilitySet::new();
+
+        // Deferred work, as `(capabilities, iterator)` pairs. Each iterator, prepared by the
+        // tactic, yields output containers and the time at which each may be shipped; the paired
+        // capability set covers those times. An iterator is dropped once it goes dry.
+        let mut todo: VecDeque<(CapabilitySet<Tr::Time>, Box<dyn Iterator<Item = (C, Tr::Time)>>)> = VecDeque::new();
+
         move |(input1, frontier1), (input2, frontier2), output| {
 
-            // Drain all input into a single buffer.
-            let mut arriving: Vec<(Tr::Time, (K, V, Tr::Time), R)> = Vec::new();
-            let mut caps = CapabilitySet::new();
+            // The driver only ships finished containers (`give_container`), never pushing records,
+            // so it pins the operator output to `NoopBuilder<C>`.
+            let output: &mut OutputBuilderSession<'_, Tr::Time, NoopBuilder<C>> = output;
+
+            // Stage all arriving updates, retaining capabilities that cover them.
             input1.for_each(|capability, data| {
                 caps.insert(capability.retain(0));
-                arriving.extend(data.drain(..).map(|(d, t, r)| (t, d, r)));
+                batcher.insert(std::mem::take(data));
             });
 
             // Drain input batches; although we do not observe them, we want access to the input
             // to observe the frontier and to drive scheduling.
             input2.for_each(|_, _| { });
 
-            // Local variables to track if and when we should exit early.
-            let mut yielded = false;
-            let timer = std::time::Instant::now();
-            let mut work = 0;
-
             if let Some(ref mut trace) = arrangement_trace {
 
-                let frontier = frontier2.frontier();
-
-                // Determine the total-order minimum of the arrangement frontier,
-                // used to partition arrivals into immediately-eligible vs stuck.
-                let mut time_con = <BatchCursor<Tr> as Cursor>::TimeContainer::with_capacity(1);
-                if let Some(min_time) = frontier.iter().min() {
-                    time_con.push_own(min_time);
-                }
-                let eligible = |initial: &Tr::Time| -> bool {
-                    !(0..time_con.len()).any(|i| comparison(time_con.index(i), initial))
+                // An update is unblocked once no arrangement time can still precede its initial
+                // time. The arrangement frontier's total-order minimum bounds those times, and
+                // `strict` says whether an equal time still counts as preceding.
+                let eligibility = match frontier2.frontier().iter().min() {
+                    None => Eligibility::All,
+                    Some(bound) if strict => Eligibility::Le(bound.clone()),
+                    Some(bound) => Eligibility::Lt(bound.clone()),
                 };
 
-                // Form a new blob from arrivals.
-                // consolidate_updates sorts by (T, D, R) — the stuck order.
-                consolidate_updates(&mut arriving);
-
-                if !arriving.is_empty() {
-                    let mut lower = MutableAntichain::new();
-                    lower.update_iter(arriving.iter().map(|(t, _, _)| (t.clone(), 1)));
-                    let mut blob_caps = CapabilitySet::new();
-                    for time in lower.frontier().iter() {
-                        blob_caps.insert(caps.delayed(time));
-                    }
-
-                    // Determine how many records are stuck (ineligible).
-                    // Data is sorted by (T, D, R) and eligibility is monotone in T,
-                    // so stuck records form a suffix.
-                    let stuck_count = arriving.iter().rev()
-                        .take_while(|(t, _, _)| !eligible(t))
-                        .count();
-
-                    let mut data: VecDeque<_> = arriving.into();
-
-                    // Sort the ready prefix by (D, T, R) for cursor traversal.
-                    let ready_len = data.len() - stuck_count;
-                    if ready_len > 0 {
-                        // VecDeque slices: make_contiguous then sort the prefix.
-                        let slice = data.make_contiguous();
-                        slice[..ready_len].sort_by(|(t1, d1, r1), (t2, d2, r2)| {
-                            (d1, t1, r1).cmp(&(d2, t2, r2))
-                        });
-                    }
-
-                    blobs.push(Blob {
-                        caps: blob_caps,
-                        lower,
-                        data,
-                        stuck_count,
-                    });
-                }
-
-                // Nibble: only when all ready elements have been drained (stuck_count == len),
-                // check if stuck records have become eligible and promote them.
-                for blob in blobs.iter_mut().filter(|b| b.stuck_count == b.data.len()) {
-
-                    // Count how many stuck records (from the front, which has the
-                    // lowest initial times) are now eligible.
-                    let newly_ready = blob.data.iter().take_while(|(t, _, _)| eligible(t)).count();
-
-                    if newly_ready > 0 {
-                        blob.stuck_count -= newly_ready;
-
-                        // Sort the newly-ready prefix by (D, T, R) for cursor traversal.
-                        let slice = blob.data.make_contiguous();
-                        slice[..newly_ready].sort_by(|(t1, d1, r1), (t2, d2, r2)| {
-                            (d1, t1, r1).cmp(&(d2, t2, r2))
-                        });
-
-                        // Downgrade capabilities.
-                        let mut new_lower = MutableAntichain::new();
-                        new_lower.update_iter(blob.data.iter().map(|(t, _, _)| (t.clone(), 1)));
-                        blob.lower = new_lower;
-                        blob.caps.downgrade(&blob.lower.frontier());
-                    }
-                }
-
-                // Process ready elements from blobs.
-                for blob in blobs.iter_mut().filter(|b| b.data.len() > b.stuck_count) {
-                    if yielded { break; }
-
-                    let mut builders = (0..blob.caps.len()).map(|_| CB::default()).collect::<Vec<_>>();
+                // `caps` covers everything `batcher` holds: capabilities retained for arriving
+                // updates, downgraded after each extraction to what was left behind. It therefore
+                // lower bounds whatever the extraction below releases, and can both mint the
+                // capabilities that work will ship under and bound the times it will contain.
+                let lower: Antichain<Tr::Time> = caps.iter().map(|c| c.time().clone()).collect();
 
+                let (chunks, retained) = batcher.extract(eligibility);
+                if !chunks.is_empty() {
+                    // The batches are handed to the tactic, which holds them for as long as the
+                    // work item lives; what it joins against cannot change underneath it.
                     let batches = trace.batches_through(Antichain::new().borrow()).unwrap();
-                    let (mut cursor, storage) = cursor_list(batches);
-                    let mut key_con = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
-                    let mut removals: ChangeBatch<Tr::Time> = ChangeBatch::new();
-
-                    // Process ready elements from the front.
-                    while blob.data.len() > blob.stuck_count {
-                        yielded = yielded || yield_function(timer, work);
-                        if yielded { break; }
-
-                        // Peek at the front element. It's in (T, D, R) storage order,
-                        // but the ready prefix has been sorted by (D, T, R).
-                        let (ref initial, (ref key, ref val1, ref time), ref diff1) = blob.data[0];
-
-                        let builder_idx = blob.caps.iter().position(|c| c.time().less_equal(initial)).unwrap();
-
-                        key_con.clear(); key_con.push_own(&key);
-                        cursor.seek_key(&storage, key_con.index(0));
-                        if cursor.get_key(&storage) == key_con.get(0) {
-                            while let Some(val2) = cursor.get_val(&storage) {
-                                cursor.map_times(&storage, |t, d| {
-                                    if comparison(t, initial) {
-                                        let mut t = <BatchCursor<Tr> as Cursor>::owned_time(t);
-                                        t.join_assign(time);
-                                        output_buffer.push((t, <BatchCursor<Tr> as Cursor>::owned_diff(d)))
-                                    }
-                                });
-                                consolidate(&mut output_buffer);
-                                work += output_buffer.len();
-                                output_func(&mut builders[builder_idx], key, val1, val2, initial, diff1, &mut output_buffer);
-                                output_buffer.clear();
-                                cursor.step_val(&storage);
-                            }
-                            cursor.rewind_vals(&storage);
-                        }
-
-                        while let Some(container) = builders[builder_idx].extract() {
-                            output.session(&blob.caps[builder_idx]).give_container(container);
-                        }
-
-                        let (initial, _, _) = blob.data.pop_front().unwrap();
-                        removals.update(initial, -1);
-                    }
-
-                    for builder_idx in 0 .. blob.caps.len() {
-                        while let Some(container) = builders[builder_idx].finish() {
-                            output.session(&blob.caps[builder_idx]).give_container(container);
-                        }
-                    }
-
-                    // Apply all removals in bulk and downgrade once.
-                    if blob.data.is_empty() {
-                        // Eagerly release the blob's resources.
-                        blob.lower = MutableAntichain::new();
-                        blob.caps = CapabilitySet::new();
-                        blob.data = VecDeque::default();
-                    } else  {
-                        blob.lower.update_iter(removals.drain());
-                        blob.caps.downgrade(&blob.lower.frontier());
-                    }
+                    let work = tactic.prep(chunks, batches, lower);
+                    todo.push_back((caps.clone(), work));
                 }
 
-                // Remove fully-consumed blobs.
-                blobs.retain(|blob| !blob.data.is_empty());
+                // Release capabilities for everything `batcher` no longer holds. The bound comes
+                // from the extraction itself, as an implementor need not account for inserted
+                // updates until it is asked to extract, and a bound read before that could discard
+                // a capability the extraction then needs.
+                caps.downgrade(retained.frontier().iter());
             }
 
-            // Re-activate if we have blobs with ready elements to process.
-            if blobs.iter().any(|b| b.data.len() > b.stuck_count) {
-                activator.activate();
+            // Perform some amount of outstanding work, shipping each container at a capability
+            // covering the time the tactic reports for it. Work is measured in output records,
+            // matching `half_join_internal_unsafe`, whose counter is also read after the join's
+            // time filter and consolidation.
+            let timer = std::time::Instant::now();
+            let mut work = 0;
+            while !yield_function(timer, work) {
+                let Some((caps, iter)) = todo.front_mut() else { break };
+                match iter.next() {
+                    Some((mut container, time)) => {
+                        work += container.record_count() as usize;
+                        let cap = caps.iter().find(|c| c.time().less_equal(&time)).expect("no capability covers a produced container");
+                        output.session_with_builder(cap).give_container(&mut container);
+                    }
+                    None => { todo.pop_front(); }
+                }
             }
+            if !todo.is_empty() { activator.activate(); }
 
-            // The logical merging frontier depends on input1 and all blobs.
+            // The logical merging frontier depends on input1, on staged updates, and on the
+            // deferred work that has already been released to the tactic.
             let mut frontier = Antichain::new();
             for time in frontier1.frontier().iter() {
                 frontier_func(time, &mut frontier);
             }
-            for blob in blobs.iter() {
-                for cap in blob.caps.iter() {
+            for cap in caps.iter() {
+                frontier_func(cap.time(), &mut frontier);
+            }
+            for (caps, _) in todo.iter() {
+                for cap in caps.iter() {
                     frontier_func(cap.time(), &mut frontier);
                 }
             }
             arrangement_trace.as_mut().map(|trace| trace.set_logical_compaction(frontier.borrow()));
 
-            if frontier1.is_empty() && blobs.is_empty() {
+            if frontier1.is_empty() && caps.is_empty() && todo.is_empty() {
                 arrangement_trace = None;
             }
         }
     })
 }
 
-/// A unified blob of updates. Data is stored as `(T, D, R)` tuples in a VecDeque.
-/// The last `stuck_count` elements are stuck (not yet eligible for processing),
-/// sorted by `(T, D, R)` from consolidation. The ready prefix (everything before
-/// the stuck tail) is sorted by `(D, T, R)` for efficient cursor traversal.
-/// Ready elements are consumed from the front via `pop_front`.
-struct Blob<D, T: Timestamp, R> {
-    caps: CapabilitySet<T>,
-    lower: MutableAntichain<T>,
-    data: VecDeque<(T, D, R)>,
-    /// Number of stuck (ineligible) elements at the back of `data`.
-    stuck_count: usize,
+/// Cursor-based half join: the conventional [`HalfJoinTactic`] implementation and its worker.
+mod cursors {
+
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
+    use super::*;
+
+    /// The cursor of a batch.
+    type BCursor<B> = <B as Navigable>::Cursor;
+
+    /// The conventional cursor-based [`HalfJoinTactic`].
+    ///
+    /// It builds a cursor over the batches it is handed and walks the released updates against it,
+    /// at whatever rate the driver allows. `logic` is shared across all outstanding units (an
+    /// `Rc<RefCell<_>>`), preserving the single mutable-state semantics of one closure threaded
+    /// through every match: each unit is a self-contained `'static` iterator, so it cannot borrow
+    /// the tactic.
+    pub struct CursorTactic<K, V, R, B, L, CB> {
+        logic: Rc<RefCell<L>>,
+        /// Whether an arrangement time equal to an update's initial time is excluded.
+        strict: bool,
+        _marker: std::marker::PhantomData<(K, V, R, B, CB)>,
+    }
+
+    impl<K, V, R, B, L, CB> CursorTactic<K, V, R, B, L, CB> {
+        /// Construct a tactic that applies `logic` to each matched `(key, val1, val2)`.
+        pub fn new(logic: L, strict: bool) -> Self {
+            CursorTactic { logic: Rc::new(RefCell::new(logic)), strict, _marker: std::marker::PhantomData }
+        }
+    }
+
+    impl<K, V, R, B, L, CB> HalfJoinTactic<B, Vec<((K, V, B::Time), B::Time, R)>, CB::Container> for CursorTactic<K, V, R, B, L, CB>
+    where
+        B: BatchReader + Navigable + 'static,
+        BCursor<B>: Cursor<Time = B::Time>,
+        <BCursor<B> as Cursor>::KeyContainer: BatchContainer<Owned = K>,
+        K: Ord + 'static,
+        V: Ord + 'static,
+        R: 'static,
+        L: for<'a> FnMut(&mut CB, &K, &V, <BCursor<B> as Cursor>::Val<'a>, &B::Time, &R, &mut Vec<(B::Time, <BCursor<B> as Cursor>::Diff)>) + 'static,
+        CB: ContainerBuilder,
+    {
+        fn prep(&mut self, chunks: Vec<Vec<((K, V, B::Time), B::Time, R)>>, batches: Vec<B>, lower: Antichain<B::Time>) -> Box<dyn Iterator<Item = (CB::Container, B::Time)>> {
+            // The batcher releases updates in time order, but the cursor is walked in data order.
+            let mut updates: Vec<_> = chunks.into_iter().flatten().collect();
+            updates.sort_by(|(d1, t1, _), (d2, t2, _)| (d1, t1).cmp(&(d2, t2)));
+
+            let (cursor, storage) = cursor_list(batches);
+            let lower = lower.elements().to_vec();
+            let builders = (0 .. lower.len()).map(|_| CB::default()).collect::<Vec<_>>();
+
+            Box::new(DeferredIter {
+                updates,
+                index: 0,
+                cursor,
+                storage,
+                key_con: BatchContainer::with_capacity(1),
+                time_con: BatchContainer::with_capacity(1),
+                lower,
+                builders,
+                ready: VecDeque::new(),
+                output_buffer: Vec::new(),
+                logic: Rc::clone(&self.logic),
+                strict: self.strict,
+                done: false,
+            })
+        }
+    }
+
+    /// Deferred half-join computation, as an iterator of output containers and the times they ship at.
+    ///
+    /// Each `next` walks the released updates forward until a builder yields a container, or the
+    /// updates run dry. The driver stops pulling once it has done enough work, and resumes the same
+    /// iterator on the next activation.
+    struct DeferredIter<K, V, R, C: Cursor, L, CB: ContainerBuilder> {
+        /// Released updates, sorted by data for cursor traversal.
+        updates: Vec<((K, V, C::Time), C::Time, R)>,
+        /// How far through `updates` we have walked.
+        index: usize,
+        cursor: C,
+        storage: C::Storage,
+        key_con: C::KeyContainer,
+        time_con: C::TimeContainer,
+        /// The times output ships at, one per builder. An update is assigned the first that is
+        /// less or equal to its initial time, mirroring the capability the driver will find.
+        lower: Vec<C::Time>,
+        builders: Vec<CB>,
+        /// Completed containers awaiting a `next` call.
+        ready: VecDeque<(CB::Container, C::Time)>,
+        /// Stash for (time, diff) accumulation.
+        output_buffer: Vec<(C::Time, C::Diff)>,
+        /// The output closure, shared across all outstanding units.
+        logic: Rc<RefCell<L>>,
+        /// Whether an arrangement time equal to an update's initial time is excluded.
+        strict: bool,
+        done: bool,
+    }
+
+    impl<K, V, R, C, L, CB> Iterator for DeferredIter<K, V, R, C, L, CB>
+    where
+        C: Cursor,
+        C::KeyContainer: BatchContainer<Owned = K>,
+        CB: ContainerBuilder,
+        L: for<'a> FnMut(&mut CB, &K, &V, C::Val<'a>, &C::Time, &R, &mut Vec<(C::Time, C::Diff)>),
+    {
+        type Item = (CB::Container, C::Time);
+
+        fn next(&mut self) -> Option<Self::Item> {
+
+            // Serve any container completed on an earlier call first.
+            if let Some(item) = self.ready.pop_front() { return Some(item); }
+            if self.done { return None; }
+
+            {
+                let updates = &self.updates;
+                let cursor = &mut self.cursor;
+                let storage = &self.storage;
+                let key_con = &mut self.key_con;
+                let time_con = &mut self.time_con;
+                let lower = &self.lower;
+                let builders = &mut self.builders;
+                let ready = &mut self.ready;
+                let output_buffer = &mut self.output_buffer;
+                let strict = self.strict;
+                let mut logic = self.logic.borrow_mut();
+                let logic = &mut *logic;
+
+                while ready.is_empty() && self.index < updates.len() {
+
+                    let ((key, val1, time), initial, diff1) = &updates[self.index];
+                    self.index += 1;
+
+                    // Each capability has its own builder, so that output shipped at one is not
+                    // mixed with output that requires another.
+                    let builder_idx = lower.iter().position(|t| t.less_equal(initial)).expect("no capability covers a released update");
+
+                    key_con.clear();
+                    key_con.push_own(key);
+                    cursor.seek_key(storage, key_con.index(0));
+                    if cursor.get_key(storage) == key_con.get(0) {
+                        time_con.clear();
+                        time_con.push_own(initial);
+                        let bound = time_con.index(0);
+                        while let Some(val2) = cursor.get_val(storage) {
+                            cursor.map_times(storage, |t, d| {
+                                // `bound` and `t` are read from different containers, and their
+                                // borrowed times only compare once narrowed to a common lifetime.
+                                let bound = <C::TimeContainer as BatchContainer>::reborrow(bound);
+                                let t = <C::TimeContainer as BatchContainer>::reborrow(t);
+                                if if strict { t < bound } else { t <= bound } {
+                                    let mut t = C::owned_time(t);
+                                    t.join_assign(time);
+                                    output_buffer.push((t, C::owned_diff(d)))
+                                }
+                            });
+                            consolidate(output_buffer);
+                            logic(&mut builders[builder_idx], key, val1, val2, initial, diff1, output_buffer);
+                            output_buffer.clear();
+                            cursor.step_val(storage);
+                        }
+                        cursor.rewind_vals(storage);
+                    }
+
+                    while let Some(container) = builders[builder_idx].extract() {
+                        ready.push_back((std::mem::take(container), lower[builder_idx].clone()));
+                    }
+                }
+            }
+
+            // Flush the final partial containers once the updates are exhausted.
+            if self.index == self.updates.len() {
+                self.done = true;
+                for builder_idx in 0 .. self.builders.len() {
+                    while let Some(container) = self.builders[builder_idx].finish() {
+                        self.ready.push_back((std::mem::take(container), self.lower[builder_idx].clone()));
+                    }
+                }
+            }
+
+            self.ready.pop_front()
+        }
+    }
 }
diff --git a/dogsdogsdogs/src/operators/propose.rs b/dogsdogsdogs/src/operators/propose.rs
index f6db7c1a6..bbc709fca 100644
--- a/dogsdogsdogs/src/operators/propose.rs
+++ b/dogsdogsdogs/src/operators/propose.rs
@@ -39,14 +39,8 @@ where
     for<'a, 'b> BatchTimeGat<'a, Tr>: PartialOrd<&'b Tr::Time>,
 {
     let requests = prefixes.map(move |(prefix, payload)| (key_selector(&prefix), prefix, payload));
-    // Branch once here, so that each comparison monomorphizes rather than testing `strict` at
-    // every timestamp. The cost is instantiating `half_join` twice.
-    if strict {
-        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 < t2,
-            |_key, prefix, value| (prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)))
-    }
-    else {
-        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 <= t2,
-            |_key, prefix, value| (prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)))
-    }
+    // `strict` now reaches the join as a value rather than as a comparison closure, so there is
+    // nothing left to monomorphize by branching here; the test is made per arrangement time.
+    crate::operators::half_join(requests, arrangement, frontier_func, strict,
+        |_key, prefix, value| (prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)))
 }
diff --git a/dogsdogsdogs/src/operators/validate.rs b/dogsdogsdogs/src/operators/validate.rs
index 5e61f942f..e12832eb7 100644
--- a/dogsdogsdogs/src/operators/validate.rs
+++ b/dogsdogsdogs/src/operators/validate.rs
@@ -41,14 +41,8 @@ where
     let requests = extensions.map(move |((prefix, extension), payload)| {
         ((key_selector(&prefix), extension.clone()), (prefix, extension), payload)
     });
-    // Branch once here, so that each comparison monomorphizes rather than testing `strict` at
-    // every timestamp. The cost is instantiating `half_join` twice.
-    if strict {
-        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 < t2,
-            |_key, extended, _value| extended.clone())
-    }
-    else {
-        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 <= t2,
-            |_key, extended, _value| extended.clone())
-    }
+    // `strict` now reaches the join as a value rather than as a comparison closure, so there is
+    // nothing left to monomorphize by branching here; the test is made per arrangement time.
+    crate::operators::half_join(requests, arrangement, frontier_func, strict,
+        |_key, extended, _value| extended.clone())
 }

From de1b5e5723434bfa03a96694a5bf7434e61dd2a9 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 19 Aug 2026 18:43:58 -0400
Subject: [PATCH 2/4] Tighten up interfaces

---
 dogsdogsdogs/src/operators/half_join.rs | 83 +++++++++++--------------
 1 file changed, 37 insertions(+), 46 deletions(-)

diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
index 908cfff1d..e7b6c3ec8 100644
--- a/dogsdogsdogs/src/operators/half_join.rs
+++ b/dogsdogsdogs/src/operators/half_join.rs
@@ -29,7 +29,7 @@ use timely::dataflow::operators::Operator;
 use timely::dataflow::operators::generic::OutputBuilderSession;
 use timely::PartialOrder;
 use timely::progress::{Antichain, Timestamp};
-use timely::progress::frontier::MutableAntichain;
+use timely::progress::frontier::{AntichainRef, MutableAntichain};
 
 use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable};
 use differential_dataflow::difference::Semigroup;
@@ -150,10 +150,10 @@ where
     CB: ContainerBuilder,
 {
     // Updates are staged in a `BlobList` and joined by a cursor walk; the driver owns progress.
-    let batcher = BlobList::<(K, V, Tr::Time), Tr::Time, R>::default();
+    let batcher = BlobList::<(K, V, Tr::Time), Tr::Time, R>::new(strict);
     let tactic = cursors::CursorTactic::<K, V, R, Tr::Batch, S, CB>::new(output_func, strict);
     let route = |update: &((K, V, Tr::Time), Tr::Time, R)| (update.0).0.hashed().into();
-    half_join_with_tactic(stream.inner, arrangement, route, frontier_func, strict, yield_function, batcher, tactic)
+    half_join_with_tactic(stream.inner, arrangement, route, frontier_func, yield_function, batcher, tactic)
 }
 
 pub trait HalfJoinTactic<B: BatchReader, C0, C1> {
@@ -171,29 +171,30 @@ pub trait HalfJoinTactic<B: BatchReader, C0, C1> {
 pub trait BatcherAlt<C, T: Timestamp> {
     /// Moves responsibility for `container` into the implementor.
     fn insert(&mut self, container: C);
-    /// Extracts containers on the basis of `bound`, and the lower bounds of the times retained.
+    /// Extracts the containers `frontier` unblocks, and the lower bounds of the times retained.
+    ///
+    /// What `frontier` unblocks is the implementor's to decide. It is handed the frontier itself
+    /// rather than a bound derived from it, so that a rule other than the antichain partial order
+    /// (`half_join` releases by a total order on times) remains expressible.
     ///
     /// The returned bound is reported at the one moment it is well defined: an implementor is not
     /// required to account for inserted updates until it is asked to extract, so this is the only
     /// bound a caller can safely downgrade capabilities to.
-    fn extract(&mut self, bound: Eligibility<T>) -> (Vec<C>, &MutableAntichain<T>);
-}
-
-pub enum Eligibility<T> {
-    All,
-    Le(T),
-    Lt(T),
+    fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec<C>, &MutableAntichain<T>);
 }
 
 struct BlobList<D, T: Timestamp, R> {
     stage: Vec<(T, D, R)>,
     blobs: Vec<VecDeque<(T, D, R)>>,
     lower: MutableAntichain<T>,
+    /// Whether an arrangement time equal to an update's time still blocks it.
+    strict: bool,
 }
 
-impl<D, T: Timestamp, R> Default for BlobList<D, T, R> {
-    fn default() -> Self {
-        BlobList { stage: Vec::new(), blobs: Vec::new(), lower: MutableAntichain::new() }
+impl<D, T: Timestamp, R> BlobList<D, T, R> {
+    /// Allocates an empty list that releases updates under the `strict` comparison.
+    fn new(strict: bool) -> Self {
+        BlobList { stage: Vec::new(), blobs: Vec::new(), lower: MutableAntichain::new(), strict }
     }
 }
 
@@ -201,7 +202,7 @@ impl<D: Ord, T: Timestamp, R: Semigroup> BatcherAlt<Vec<(D, T, R)>, T> for BlobL
     fn insert(&mut self, container: Vec<(D, T, R)>) {
         self.stage.extend(container.into_iter().map(|(d,t,r)| (t,d,r)));
     }
-    fn extract(&mut self, bound: Eligibility<T>) -> (Vec<Vec<(D, T, R)>>, &MutableAntichain<T>) {
+    fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec<Vec<(D, T, R)>>, &MutableAntichain<T>) {
         // Handle any staged updates first.
         consolidate_updates(&mut self.stage);
         if !self.stage.is_empty() {
@@ -210,25 +211,25 @@ impl<D: Ord, T: Timestamp, R: Semigroup> BatcherAlt<Vec<(D, T, R)>, T> for BlobL
             self.blobs.push(blob);
         }
 
+        // An update is unblocked once no arrangement time can still precede it. The times the
+        // frontier may yet produce are bounded below in the total order by its minimum, and
+        // `strict` says whether a time equal to an update's own still counts as preceding. An
+        // empty frontier will produce no times at all, and unblocks everything.
+        let bound = frontier.iter().min();
+        let strict = self.strict;
+        let eligible = |time: &T| match bound {
+            None => true,
+            Some(bound) => if strict { time.le(bound) } else { time.lt(bound) },
+        };
+
+        // Blobs are stored in time order, and eligibility is monotone in time, so what each
+        // releases is a prefix.
         let mut result = Vec::new();
         for blob in self.blobs.iter_mut() {
             let mut list = Vec::new();
-            match &bound {
-                Eligibility::Le(bound) => {
-                    while blob.front().map(|(t,_,_)| t.le(bound)).unwrap_or(false) {
-                        let (time, data, diff) = blob.pop_front().unwrap();
-                        list.push((data, time, diff));
-                    }
-                }
-                Eligibility::Lt(bound) => {
-                    while blob.front().map(|(t,_,_)| t.lt(bound)).unwrap_or(false) {
-                        let (time, data, diff) = blob.pop_front().unwrap();
-                        list.push((data, time, diff));
-                    }
-                }
-                Eligibility::All => {
-                    list.extend(blob.drain(..).map(|(t,d,r)| (d,t,r)));
-                }
+            while blob.front().map(|(t,_,_)| eligible(t)).unwrap_or(false) {
+                let (time, data, diff) = blob.pop_front().unwrap();
+                list.push((data, time, diff));
             }
             if !list.is_empty() { result.push(list); }
         }
@@ -248,16 +249,15 @@ impl<D: Ord, T: Timestamp, R: Semigroup> BatcherAlt<Vec<(D, T, R)>, T> for BlobL
 /// work paired with the capabilities required to ship its output. It holds no cursor, and has no
 /// opinion on how the join is performed.
 ///
-/// The `strict` argument plays the role of the `comparison` closure: an update is unblocked once
-/// no arrangement time can still precede its initial time under the total order, and `strict` says
-/// whether an equal time counts as preceding. It reaches the join itself through `tactic`, which
-/// must be built with the same value.
+/// The driver holds no opinion on when an update is unblocked: it hands `batcher` the arrangement
+/// frontier and ships whatever `batcher` releases. The rule that decides this — in particular
+/// whether an arrangement time equal to an update's initial time still blocks it — is shared
+/// between `batcher` and `tactic`, which must be built to agree.
 pub fn half_join_with_tactic<'scope, Tr, FF, RF, Y, Bat, Tac, CIn, C>(
     stream: Stream<'scope, Tr::Time, CIn>,
     mut arrangement: Arranged<'scope, Tr>,
     route: RF,
     frontier_func: FF,
-    strict: bool,
     yield_function: Y,
     mut batcher: Bat,
     mut tactic: Tac,
@@ -313,22 +313,13 @@ where
 
             if let Some(ref mut trace) = arrangement_trace {
 
-                // An update is unblocked once no arrangement time can still precede its initial
-                // time. The arrangement frontier's total-order minimum bounds those times, and
-                // `strict` says whether an equal time still counts as preceding.
-                let eligibility = match frontier2.frontier().iter().min() {
-                    None => Eligibility::All,
-                    Some(bound) if strict => Eligibility::Le(bound.clone()),
-                    Some(bound) => Eligibility::Lt(bound.clone()),
-                };
-
                 // `caps` covers everything `batcher` holds: capabilities retained for arriving
                 // updates, downgraded after each extraction to what was left behind. It therefore
                 // lower bounds whatever the extraction below releases, and can both mint the
                 // capabilities that work will ship under and bound the times it will contain.
                 let lower: Antichain<Tr::Time> = caps.iter().map(|c| c.time().clone()).collect();
 
-                let (chunks, retained) = batcher.extract(eligibility);
+                let (chunks, retained) = batcher.extract(frontier2.frontier());
                 if !chunks.is_empty() {
                     // The batches are handed to the tactic, which holds them for as long as the
                     // work item lives; what it joins against cannot change underneath it.

From 519c88d50ddfc871ee7c8e2fd386f23654d2de7d Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 20 Aug 2026 11:38:55 -0400
Subject: [PATCH 3/4] reorganize and tighten

---
 dogsdogsdogs/src/operators/count.rs     |   2 +-
 dogsdogsdogs/src/operators/half_join.rs | 480 ++++++++++++------------
 dogsdogsdogs/src/operators/mod.rs       |   2 +-
 3 files changed, 234 insertions(+), 250 deletions(-)

diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs
index a40c0155a..3ab2bfbb5 100644
--- a/dogsdogsdogs/src/operators/count.rs
+++ b/dogsdogsdogs/src/operators/count.rs
@@ -63,7 +63,7 @@ where
         builder.push_into(((triple, payload.clone()), initial.clone(), diff1.clone()));
     };
 
-    use crate::operators::half_join::half_join_internal_unsafe as half_join_unsafe;
+    use crate::operators::half_join::cursors::half_join_internal_unsafe as half_join_unsafe;
     // `strict` now reaches the join as a value rather than as a comparison closure, so there is
     // nothing left to monomorphize by branching here; the test is made per arrangement time.
     half_join_unsafe::<_, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
index e7b6c3ec8..c1a72572a 100644
--- a/dogsdogsdogs/src/operators/half_join.rs
+++ b/dogsdogsdogs/src/operators/half_join.rs
@@ -1,30 +1,28 @@
-/// Streaming asymmetric join between updates (K,V1) and an arrangement on (K,V2).
+/// Streaming asymmetric join between an update stream and a maintained arrangement.
 ///
 /// The asymmetry is that the join only responds to streamed updates, not to changes in the arrangement.
-/// Streamed updates join only with matching arranged updates at lesser times *in the total order*,
-/// where a `strict` flag says whether an equal time counts as lesser.
+/// The operator is the basis of several multi-way join implementations, which use networks of stateless
+/// operators rather than sequences of stateful operators.
 ///
-/// This behavior can ensure that any pair of matching updates interact exactly once.
+/// The standard recipe for a multi-way join among collections A, B, .. Z is to create a dataflow path for
+/// each collection, which responds to changes in the collection. Each path uses a sequence of half join
+/// operators to prompt the response to each streamed input change when joined with "prior" updates from
+/// the other collections. A total order on update times, often just the `Ord` implementation, can then be
+/// extended to all relations to impose a total order on updates to any of the collections, which can then
+/// ensure that each tuple of updates is processed exactly once (usually at the time of the "last" update).
 ///
-/// There are various forms of this operator with tangled closures about how to emit the outputs and
-/// wrangle the logical compaction frontier in order to preserve the distinctions around times that are
-/// strictly less (conventional compaction logic would collapse unequal times to the frontier, and lose
-/// the distiction).
-///
-/// The methods also carry an auxiliary time next to the value, which is used to advance the joined times.
-/// This is .. a byproduct of wanting to allow advancing times a la `join_function`, without breaking the
-/// coupling by total order on "initial time".
-///
-/// The doccomments for individual methods are a bit of a mess. Sorry.
+/// The operator design is foremost as a "tactic", which allows an implementor to supply the understanding
+/// of the streamed update containers and the maintained trace batches, without complicating the operator.
+/// The implementation requires an explanation of how to mark updates as "eligible", via the `Batcher` trait,
+/// and then what to do with the eligible updates and existing batches, through the `HalfJoinTactic` trait.
 
 use std::collections::VecDeque;
 use std::ops::Mul;
 
 use timely::{Container, ContainerBuilder};
-use timely::container::{CapacityContainerBuilder, DrainContainer, NoopBuilder, PushInto, SizableContainer};
+use timely::container::{CapacityContainerBuilder, NoopBuilder};
 use timely::dataflow::Stream;
-use timely::dataflow::channels::ContainerBytes;
-use timely::dataflow::channels::pact::{Pipeline, ExchangeCore};
+use timely::dataflow::channels::pact::{ParallelizationContract, Pipeline};
 use timely::dataflow::operators::Operator;
 use timely::dataflow::operators::generic::OutputBuilderSession;
 use timely::PartialOrder;
@@ -42,120 +40,9 @@ use differential_dataflow::trace::implementations::BatchContainer;
 
 use timely::dataflow::operators::CapabilitySet;
 
-/// A binary equijoin that responds to updates on only its first input.
-///
-/// This operator responds to inputs of the form
-///
-/// ```ignore
-/// ((key, val1, time1), initial_time, diff1)
-/// ```
-///
-/// where `initial_time` is less or equal to `time1`, and produces as output
-///
-/// ```ignore
-/// ((output_func(key, val1, val2), lub(time1, time2)), initial_time, diff1 * diff2)
-/// ```
-///
-/// for each `((key, val2), time2, diff2)` present in `arrangement`, where
-/// `time2` is less than `initial_time` *UNDER THE TOTAL ORDER ON TIMES*.
-/// This last constraint is important to ensure that we correctly produce
-/// all pairs of output updates across multiple `half_join` operators.
+/// An implementation suitable to define half-join behavior among its referenced types.
 ///
-/// The `strict` argument selects the comparison: `true` requires `time2` to be strictly less than
-/// `initial_time`, and `false` also admits `time2` equal to it. A delta query pairs a strict operator
-/// with a non-strict one, which is what makes each pair of matching updates interact exactly once.
-///
-/// Notice that the time is hoisted up into data. The expectation is that
-/// once out of the "delta flow region", the updates will be `delay`d to the
-/// times specified in the payloads.
-pub fn half_join<'scope, K, V, R, Tr, FF, DOut, S>(
-    stream: VecCollection<'scope, Tr::Time, (K, V, Tr::Time), R>,
-    arrangement: Arranged<'scope, Tr>,
-    frontier_func: FF,
-    strict: bool,
-    mut output_func: S,
-) -> VecCollection<'scope, Tr::Time, (DOut, Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
-where
-    K: Hashable + ExchangeData,
-    V: ExchangeData,
-    R: ExchangeData + Semigroup,
-    Tr: TraceReader<Batch: Navigable>+Clone+'static,
-    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
-    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
-    R: Mul<BatchDiff<Tr>, Output: Semigroup>,
-    FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    DOut: Clone+'static,
-    S: FnMut(&K, &V, BatchVal<'_, Tr>)->DOut+'static,
-{
-    let output_func = move |builder: &mut CapacityContainerBuilder<Vec<_>>, k: &K, v1: &V, v2: BatchVal<'_, Tr>, initial: &Tr::Time, diff1: &R, output: &mut Vec<(Tr::Time, BatchDiff<Tr>)>| {
-        for (time, diff2) in output.drain(..) {
-            let diff = diff1.clone() * diff2.clone();
-            let dout = (output_func(k, v1, v2), time.clone());
-            use timely::container::PushInto;
-            builder.push_into((dout, initial.clone(), diff));
-        }
-    };
-    half_join_internal_unsafe::<_, _, _, _, _, _, _, CapacityContainerBuilder<Vec<_>>>(stream, arrangement, frontier_func, strict, |_timer, _count| false, output_func)
-        .as_collection()
-}
-
-/// An unsafe variant of `half_join` where the `output_func` closure takes
-/// additional arguments a vector of `time` and `diff` tuples as input and
-/// writes its outputs at a container builder. The container builder
-/// can, but isn't required to, accept `(data, time, diff)` triplets.
-/// This allows for more flexibility, but is more error-prone.
-///
-/// This operator responds to inputs of the form
-///
-/// ```ignore
-/// ((key, val1, time1), initial_time, diff1)
-/// ```
-///
-/// where `initial_time` is less or equal to `time1`, and produces as output
-///
-/// ```ignore
-/// output_func(session, key, val1, val2, initial_time, diff1, &[lub(time1, time2), diff2])
-/// ```
-///
-/// for each `((key, val2), time2, diff2)` present in `arrangement`, where
-/// `time2` is less than `initial_time` *UNDER THE TOTAL ORDER ON TIMES*.
-///
-/// The `strict` argument selects the comparison: `true` requires `time2` to be strictly less than
-/// `initial_time`, and `false` also admits `time2` equal to it. A delta query pairs a strict operator
-/// with a non-strict one, which is what makes each pair of matching updates interact exactly once.
-///
-/// The `yield_function` allows the caller to indicate when the operator should
-/// yield control, as a function of the elapsed time and the number of records in
-/// the output containers shipped so far. Joined work is suspended at container
-/// boundaries, so the count advances a container at a time rather than a record
-/// at a time.
-pub fn half_join_internal_unsafe<'scope, K, V, R, Tr, FF, Y, S, CB>(
-    stream: VecCollection<'scope, Tr::Time, (K, V, Tr::Time), R>,
-    arrangement: Arranged<'scope, Tr>,
-    frontier_func: FF,
-    strict: bool,
-    yield_function: Y,
-    output_func: S,
-) -> Stream<'scope, Tr::Time, CB::Container>
-where
-    K: Hashable + ExchangeData,
-    V: ExchangeData,
-    R: ExchangeData + Semigroup,
-    Tr: TraceReader<Batch: Navigable>+Clone+'static,
-    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
-    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
-    FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    Y: Fn(std::time::Instant, usize) -> bool + 'static,
-    S: FnMut(&mut CB, &K, &V, BatchVal<'_, Tr>, &Tr::Time, &R, &mut Vec<(Tr::Time, BatchDiff<Tr>)>) + 'static,
-    CB: ContainerBuilder,
-{
-    // Updates are staged in a `BlobList` and joined by a cursor walk; the driver owns progress.
-    let batcher = BlobList::<(K, V, Tr::Time), Tr::Time, R>::new(strict);
-    let tactic = cursors::CursorTactic::<K, V, R, Tr::Batch, S, CB>::new(output_func, strict);
-    let route = |update: &((K, V, Tr::Time), Tr::Time, R)| (update.0).0.hashed().into();
-    half_join_with_tactic(stream.inner, arrangement, route, frontier_func, yield_function, batcher, tactic)
-}
-
+/// The implementor can half-join streams of `C0` with batches `B`, producing output streams of `C1`.
 pub trait HalfJoinTactic<B: BatchReader, C0, C1> {
     /// Converts a list of chunks and a list of batches to a list of outputs, each with a time suitable as a capability.
     ///
@@ -168,95 +55,36 @@ pub trait HalfJoinTactic<B: BatchReader, C0, C1> {
 ///
 /// The implementor is able to determine the meaning of extraction by a frontier;
 /// it is not required to be by antichain partial order.
-pub trait BatcherAlt<C, T: Timestamp> {
+pub trait Batcher<C, T: Timestamp> {
     /// Moves responsibility for `container` into the implementor.
     fn insert(&mut self, container: C);
-    /// Extracts the containers `frontier` unblocks, and the lower bounds of the times retained.
+    /// Extracts updates `frontier` unblocks, and lower bounds the time of retained updates.
     ///
-    /// What `frontier` unblocks is the implementor's to decide. It is handed the frontier itself
-    /// rather than a bound derived from it, so that a rule other than the antichain partial order
-    /// (`half_join` releases by a total order on times) remains expressible.
+    /// What `frontier` unblocks is the implementor's to decide. It can be based on the antichain up set,
+    /// or it can be based on the total order of times (as used in delta join constructions).
     ///
-    /// The returned bound is reported at the one moment it is well defined: an implementor is not
-    /// required to account for inserted updates until it is asked to extract, so this is the only
-    /// bound a caller can safely downgrade capabilities to.
+    /// The reported lower bound antichain should accurately reflect the times of all accepted updates
+    /// that have not been extracted. Over approximation can result in stalling dataflows, and under
+    /// approximation is simply incorrect.
     fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec<C>, &MutableAntichain<T>);
 }
 
-struct BlobList<D, T: Timestamp, R> {
-    stage: Vec<(T, D, R)>,
-    blobs: Vec<VecDeque<(T, D, R)>>,
-    lower: MutableAntichain<T>,
-    /// Whether an arrangement time equal to an update's time still blocks it.
-    strict: bool,
-}
-
-impl<D, T: Timestamp, R> BlobList<D, T, R> {
-    /// Allocates an empty list that releases updates under the `strict` comparison.
-    fn new(strict: bool) -> Self {
-        BlobList { stage: Vec::new(), blobs: Vec::new(), lower: MutableAntichain::new(), strict }
-    }
-}
-
-impl<D: Ord, T: Timestamp, R: Semigroup> BatcherAlt<Vec<(D, T, R)>, T> for BlobList<D, T, R> {
-    fn insert(&mut self, container: Vec<(D, T, R)>) {
-        self.stage.extend(container.into_iter().map(|(d,t,r)| (t,d,r)));
-    }
-    fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec<Vec<(D, T, R)>>, &MutableAntichain<T>) {
-        // Handle any staged updates first.
-        consolidate_updates(&mut self.stage);
-        if !self.stage.is_empty() {
-            let blob: VecDeque<_> = std::mem::take(&mut self.stage).into();
-            self.lower.update_iter(blob.iter().map(|x| (x.0.clone(), 1)));
-            self.blobs.push(blob);
-        }
-
-        // An update is unblocked once no arrangement time can still precede it. The times the
-        // frontier may yet produce are bounded below in the total order by its minimum, and
-        // `strict` says whether a time equal to an update's own still counts as preceding. An
-        // empty frontier will produce no times at all, and unblocks everything.
-        let bound = frontier.iter().min();
-        let strict = self.strict;
-        let eligible = |time: &T| match bound {
-            None => true,
-            Some(bound) => if strict { time.le(bound) } else { time.lt(bound) },
-        };
-
-        // Blobs are stored in time order, and eligibility is monotone in time, so what each
-        // releases is a prefix.
-        let mut result = Vec::new();
-        for blob in self.blobs.iter_mut() {
-            let mut list = Vec::new();
-            while blob.front().map(|(t,_,_)| eligible(t)).unwrap_or(false) {
-                let (time, data, diff) = blob.pop_front().unwrap();
-                list.push((data, time, diff));
-            }
-            if !list.is_empty() { result.push(list); }
-        }
-
-        self.blobs.retain(|b| !b.is_empty());
-
-        self.lower.update_iter(result.iter().flat_map(|l| l.iter().map(|x| (x.1.clone(), -1))));
-        (result, &self.lower)
-    }
-}
-
-/// A `half_join` driven by a [`BatcherAlt`] and a [`HalfJoinTactic`].
+/// A `half_join` driven by a [`Batcher`] and a [`HalfJoinTactic`].
 ///
-/// This is the analogue of [`half_join_internal_unsafe`], with the join itself lifted out. The
-/// driver owns progress and nothing else: it stages arriving updates in `batcher`, releases those
-/// the arrangement frontier has unblocked, and hands each release to `tactic` as a unit of deferred
-/// work paired with the capabilities required to ship its output. It holds no cursor, and has no
-/// opinion on how the join is performed.
+/// The operator introduces all streamed updates to the `batcher`, and then extracts all updates
+/// unlocked by the trace frontier. The `tactic` converts unlocked updates into output containers,
+/// which are produced lazily as the operator's yield logic permits.
 ///
 /// The driver holds no opinion on when an update is unblocked: it hands `batcher` the arrangement
-/// frontier and ships whatever `batcher` releases. The rule that decides this — in particular
-/// whether an arrangement time equal to an update's initial time still blocks it — is shared
-/// between `batcher` and `tactic`, which must be built to agree.
-pub fn half_join_with_tactic<'scope, Tr, FF, RF, Y, Bat, Tac, CIn, C>(
+/// frontier and ships whatever `batcher` releases to `tactic`. The pair of `batcher` and `tactic`
+/// should agree with each other for meaningful outcomes.
+///
+/// The `pact` distributes the streamed updates, which should land each at the appropriate worker.
+/// A likely implementation is an exchange by a hash of a common key.
+pub fn half_join_with_tactic<'scope, Tr, FF, P, Y, Bat, Tac, CIn, C>(
     stream: Stream<'scope, Tr::Time, CIn>,
     mut arrangement: Arranged<'scope, Tr>,
-    route: RF,
+    pact: P,
     frontier_func: FF,
     yield_function: Y,
     mut batcher: Bat,
@@ -265,12 +93,11 @@ pub fn half_join_with_tactic<'scope, Tr, FF, RF, Y, Bat, Tac, CIn, C>(
 where
     Tr: TraceReader + Clone + 'static,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    for<'a> RF: FnMut(&CIn::Item<'a>) -> u64 + 'static,
+    P: ParallelizationContract<Tr::Time, CIn>,
     Y: Fn(std::time::Instant, usize) -> bool + 'static,
-    Bat: BatcherAlt<CIn, Tr::Time> + 'static,
+    Bat: Batcher<CIn, Tr::Time> + 'static,
     Tac: HalfJoinTactic<Tr::Batch, CIn, C> + 'static,
-    CIn: Container + SizableContainer + DrainContainer + ContainerBytes + Send,
-    for<'a> CIn: PushInto<<CIn as DrainContainer>::Item<'a>>,
+    CIn: Container,
     C: Container + 'static,
 {
     // No need to block physical merging for this operator.
@@ -278,10 +105,8 @@ where
     let mut arrangement_trace = Some(arrangement.trace);
     let arrangement_stream = arrangement.stream;
 
-    let exchange = ExchangeCore::<CapacityContainerBuilder<CIn>, _>::new(route);
-
     let scope = stream.scope();
-    stream.binary_frontier(arrangement_stream, exchange, Pipeline, "HalfJoin", move |_, info| {
+    stream.binary_frontier(arrangement_stream, pact, Pipeline, "HalfJoin", move |_, info| {
 
         // Acquire an activator to reschedule the operator when it has unfinished work.
         let activator = scope.activator_for(info.address);
@@ -292,53 +117,43 @@ where
 
         // Deferred work, as `(capabilities, iterator)` pairs. Each iterator, prepared by the
         // tactic, yields output containers and the time at which each may be shipped; the paired
-        // capability set covers those times. An iterator is dropped once it goes dry.
+        // capability set covers those times.
         let mut todo: VecDeque<(CapabilitySet<Tr::Time>, Box<dyn Iterator<Item = (C, Tr::Time)>>)> = VecDeque::new();
 
         move |(input1, frontier1), (input2, frontier2), output| {
 
-            // The driver only ships finished containers (`give_container`), never pushing records,
-            // so it pins the operator output to `NoopBuilder<C>`.
+            // The driver ships only finished containers, so it pins the operator output to `NoopBuilder<C>`.
             let output: &mut OutputBuilderSession<'_, Tr::Time, NoopBuilder<C>> = output;
 
             // Stage all arriving updates, retaining capabilities that cover them.
+            // TODO: Tolerate multi-capability inputs.
             input1.for_each(|capability, data| {
                 caps.insert(capability.retain(0));
                 batcher.insert(std::mem::take(data));
             });
 
-            // Drain input batches; although we do not observe them, we want access to the input
-            // to observe the frontier and to drive scheduling.
+            // Drain input batches. We do not capture the batches, but we do use the frontier.
             input2.for_each(|_, _| { });
 
             if let Some(ref mut trace) = arrangement_trace {
 
-                // `caps` covers everything `batcher` holds: capabilities retained for arriving
-                // updates, downgraded after each extraction to what was left behind. It therefore
-                // lower bounds whatever the extraction below releases, and can both mint the
-                // capabilities that work will ship under and bound the times it will contain.
-                let lower: Antichain<Tr::Time> = caps.iter().map(|c| c.time().clone()).collect();
-
+                // Look for updates that are newly eligible against the current frontier.
                 let (chunks, retained) = batcher.extract(frontier2.frontier());
                 if !chunks.is_empty() {
                     // The batches are handed to the tactic, which holds them for as long as the
                     // work item lives; what it joins against cannot change underneath it.
                     let batches = trace.batches_through(Antichain::new().borrow()).unwrap();
+                    let lower: Antichain<Tr::Time> = caps.iter().map(|c| c.time().clone()).collect();
                     let work = tactic.prep(chunks, batches, lower);
                     todo.push_back((caps.clone(), work));
                 }
 
-                // Release capabilities for everything `batcher` no longer holds. The bound comes
-                // from the extraction itself, as an implementor need not account for inserted
-                // updates until it is asked to extract, and a bound read before that could discard
-                // a capability the extraction then needs.
+                // Downgrade capabilities to those held by `batcher`.
                 caps.downgrade(retained.frontier().iter());
             }
 
             // Perform some amount of outstanding work, shipping each container at a capability
-            // covering the time the tactic reports for it. Work is measured in output records,
-            // matching `half_join_internal_unsafe`, whose counter is also read after the join's
-            // time filter and consolidation.
+            // covering the time the tactic reports for it. Work is measured in output records.
             let timer = std::time::Instant::now();
             let mut work = 0;
             while !yield_function(timer, work) {
@@ -357,34 +172,144 @@ where
             // The logical merging frontier depends on input1, on staged updates, and on the
             // deferred work that has already been released to the tactic.
             let mut frontier = Antichain::new();
-            for time in frontier1.frontier().iter() {
-                frontier_func(time, &mut frontier);
-            }
-            for cap in caps.iter() {
-                frontier_func(cap.time(), &mut frontier);
-            }
-            for (caps, _) in todo.iter() {
-                for cap in caps.iter() {
-                    frontier_func(cap.time(), &mut frontier);
-                }
-            }
+            for time in frontier1.frontier().iter() { frontier_func(time, &mut frontier); }
+            for cap in caps.iter() { frontier_func(cap.time(), &mut frontier); }
+            for (caps, _) in todo.iter() { for cap in caps.iter() { frontier_func(cap.time(), &mut frontier); } }
             arrangement_trace.as_mut().map(|trace| trace.set_logical_compaction(frontier.borrow()));
 
-            if frontier1.is_empty() && caps.is_empty() && todo.is_empty() {
-                arrangement_trace = None;
-            }
+            // If no updates incoming, no updates held in `batcher`, and no work in `todo`, we can drop the trace.
+            if frontier1.is_empty() && caps.is_empty() && todo.is_empty() { arrangement_trace = None; }
         }
     })
 }
 
-/// Cursor-based half join: the conventional [`HalfJoinTactic`] implementation and its worker.
-mod cursors {
+/// Cursor-based half join: the conventional [`HalfJoinTactic`] implementation, its worker, and the
+/// `half_join` entry points built on them.
+pub mod cursors {
 
     use std::cell::RefCell;
     use std::rc::Rc;
 
+    use timely::dataflow::channels::pact::Exchange;
+
     use super::*;
 
+    /// A binary equijoin that responds to updates on only its first input.
+    ///
+    /// This operator responds to inputs of the form
+    ///
+    /// ```ignore
+    /// ((key, val1, time1), initial_time, diff1)
+    /// ```
+    ///
+    /// where `initial_time` is less or equal to `time1`, and produces as output
+    ///
+    /// ```ignore
+    /// ((output_func(key, val1, val2), lub(time1, time2)), initial_time, diff1 * diff2)
+    /// ```
+    ///
+    /// for each `((key, val2), time2, diff2)` present in `arrangement`, where
+    /// `time2` is less than `initial_time` *UNDER THE TOTAL ORDER ON TIMES*.
+    /// This last constraint is important to ensure that we correctly produce
+    /// all pairs of output updates across multiple `half_join` operators.
+    ///
+    /// The `strict` argument selects the comparison: `true` requires `time2` to be strictly less than
+    /// `initial_time`, and `false` also admits `time2` equal to it. A delta query pairs a strict operator
+    /// with a non-strict one, which is what makes each pair of matching updates interact exactly once.
+    ///
+    /// Notice that the time is hoisted up into data. The expectation is that
+    /// once out of the "delta flow region", the updates will be `delay`d to the
+    /// times specified in the payloads.
+    pub fn half_join<'scope, K, V, R, Tr, FF, DOut, S>(
+        stream: VecCollection<'scope, Tr::Time, (K, V, Tr::Time), R>,
+        arrangement: Arranged<'scope, Tr>,
+        frontier_func: FF,
+        strict: bool,
+        mut output_func: S,
+    ) -> VecCollection<'scope, Tr::Time, (DOut, Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
+    where
+        K: Hashable + ExchangeData,
+        V: ExchangeData,
+        R: ExchangeData + Semigroup,
+        Tr: TraceReader<Batch: Navigable>+Clone+'static,
+        BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+        <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
+        R: Mul<BatchDiff<Tr>, Output: Semigroup>,
+        FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
+        DOut: Clone+'static,
+        S: FnMut(&K, &V, BatchVal<'_, Tr>)->DOut+'static,
+    {
+        let output_func = move |builder: &mut CapacityContainerBuilder<Vec<_>>, k: &K, v1: &V, v2: BatchVal<'_, Tr>, initial: &Tr::Time, diff1: &R, output: &mut Vec<(Tr::Time, BatchDiff<Tr>)>| {
+            for (time, diff2) in output.drain(..) {
+                let diff = diff1.clone() * diff2.clone();
+                let dout = (output_func(k, v1, v2), time.clone());
+                use timely::container::PushInto;
+                builder.push_into((dout, initial.clone(), diff));
+            }
+        };
+        half_join_internal_unsafe::<_, _, _, _, _, _, _, CapacityContainerBuilder<Vec<_>>>(stream, arrangement, frontier_func, strict, |_timer, _count| false, output_func)
+            .as_collection()
+    }
+
+    /// An unsafe variant of `half_join` where the `output_func` closure takes
+    /// additional arguments a vector of `time` and `diff` tuples as input and
+    /// writes its outputs at a container builder. The container builder
+    /// can, but isn't required to, accept `(data, time, diff)` triplets.
+    /// This allows for more flexibility, but is more error-prone.
+    ///
+    /// This operator responds to inputs of the form
+    ///
+    /// ```ignore
+    /// ((key, val1, time1), initial_time, diff1)
+    /// ```
+    ///
+    /// where `initial_time` is less or equal to `time1`, and produces as output
+    ///
+    /// ```ignore
+    /// output_func(session, key, val1, val2, initial_time, diff1, &[lub(time1, time2), diff2])
+    /// ```
+    ///
+    /// for each `((key, val2), time2, diff2)` present in `arrangement`, where
+    /// `time2` is less than `initial_time` *UNDER THE TOTAL ORDER ON TIMES*.
+    ///
+    /// The `strict` argument selects the comparison: `true` requires `time2` to be strictly less than
+    /// `initial_time`, and `false` also admits `time2` equal to it. A delta query pairs a strict operator
+    /// with a non-strict one, which is what makes each pair of matching updates interact exactly once.
+    ///
+    /// The `yield_function` allows the caller to indicate when the operator should
+    /// yield control, as a function of the elapsed time and the number of records in
+    /// the output containers shipped so far. Joined work is suspended at container
+    /// boundaries, so the count advances a container at a time rather than a record
+    /// at a time.
+    pub fn half_join_internal_unsafe<'scope, K, V, R, Tr, FF, Y, S, CB>(
+        stream: VecCollection<'scope, Tr::Time, (K, V, Tr::Time), R>,
+        arrangement: Arranged<'scope, Tr>,
+        frontier_func: FF,
+        strict: bool,
+        yield_function: Y,
+        output_func: S,
+    ) -> Stream<'scope, Tr::Time, CB::Container>
+    where
+        K: Hashable + ExchangeData,
+        V: ExchangeData,
+        R: ExchangeData + Semigroup,
+        Tr: TraceReader<Batch: Navigable>+Clone+'static,
+        BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+        <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
+        FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
+        Y: Fn(std::time::Instant, usize) -> bool + 'static,
+        S: FnMut(&mut CB, &K, &V, BatchVal<'_, Tr>, &Tr::Time, &R, &mut Vec<(Tr::Time, BatchDiff<Tr>)>) + 'static,
+        CB: ContainerBuilder,
+    {
+        // Updates are staged in a `BlobList` and joined by a cursor walk; the driver owns progress.
+        let batcher = BlobList::<(K, V, Tr::Time), Tr::Time, R>::new(strict);
+        let tactic = CursorTactic::<K, V, R, Tr::Batch, S, CB>::new(output_func, strict);
+        // Updates are routed by key hash, matching how `arrangement` itself is distributed.
+        let route = |update: &((K, V, Tr::Time), Tr::Time, R)| (update.0).0.hashed().into();
+        let pact = Exchange::new(route);
+        half_join_with_tactic(stream.inner, arrangement, pact, frontier_func, yield_function, batcher, tactic)
+    }
+
     /// The cursor of a batch.
     type BCursor<B> = <B as Navigable>::Cursor;
 
@@ -560,4 +485,63 @@ mod cursors {
             self.ready.pop_front()
         }
     }
+
+    struct BlobList<D, T: Timestamp, R> {
+        stage: Vec<(T, D, R)>,
+        blobs: Vec<VecDeque<(T, D, R)>>,
+        lower: MutableAntichain<T>,
+        /// Whether an arrangement time equal to an update's time still blocks it.
+        strict: bool,
+    }
+
+    impl<D, T: Timestamp, R> BlobList<D, T, R> {
+        /// Allocates an empty list that releases updates under the `strict` comparison.
+        fn new(strict: bool) -> Self {
+            BlobList { stage: Vec::new(), blobs: Vec::new(), lower: MutableAntichain::new(), strict }
+        }
+    }
+
+    impl<D: Ord, T: Timestamp, R: Semigroup> Batcher<Vec<(D, T, R)>, T> for BlobList<D, T, R> {
+        fn insert(&mut self, container: Vec<(D, T, R)>) {
+            self.stage.extend(container.into_iter().map(|(d,t,r)| (t,d,r)));
+        }
+        fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec<Vec<(D, T, R)>>, &MutableAntichain<T>) {
+            // Handle any staged updates first.
+            consolidate_updates(&mut self.stage);
+            if !self.stage.is_empty() {
+                let blob: VecDeque<_> = std::mem::take(&mut self.stage).into();
+                self.lower.update_iter(blob.iter().map(|x| (x.0.clone(), 1)));
+                self.blobs.push(blob);
+            }
+
+            // An update is unblocked once no arrangement time can still precede it. The times the
+            // frontier may yet produce are bounded below in the total order by its minimum, and
+            // `strict` says whether a time equal to an update's own still counts as preceding. An
+            // empty frontier will produce no times at all, and unblocks everything.
+            let bound = frontier.iter().min();
+            let strict = self.strict;
+            let eligible = |time: &T| match bound {
+                None => true,
+                Some(bound) => if strict { time.le(bound) } else { time.lt(bound) },
+            };
+
+            // Blobs are stored in time order, and eligibility is monotone in time, so what each
+            // releases is a prefix.
+            let mut result = Vec::new();
+            for blob in self.blobs.iter_mut() {
+                let mut list = Vec::new();
+                while blob.front().map(|(t,_,_)| eligible(t)).unwrap_or(false) {
+                    let (time, data, diff) = blob.pop_front().unwrap();
+                    list.push((data, time, diff));
+                }
+                if !list.is_empty() { result.push(list); }
+            }
+
+            self.blobs.retain(|b| !b.is_empty());
+
+            self.lower.update_iter(result.iter().flat_map(|l| l.iter().map(|x| (x.1.clone(), -1))));
+            (result, &self.lower)
+        }
+    }
+
 }
diff --git a/dogsdogsdogs/src/operators/mod.rs b/dogsdogsdogs/src/operators/mod.rs
index 6da214107..f35f58ca8 100644
--- a/dogsdogsdogs/src/operators/mod.rs
+++ b/dogsdogsdogs/src/operators/mod.rs
@@ -4,7 +4,7 @@ pub mod count;
 pub mod propose;
 pub mod validate;
 
-pub use self::half_join::half_join;
+pub use self::half_join::cursors::half_join;
 pub use self::count::count;
 pub use self::propose::propose;
 pub use self::validate::validate;

From fd32a902507fdd1f87b8e8861c3dc5f2dbd76e1d Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 20 Aug 2026 12:51:11 -0400
Subject: [PATCH 4/4] Generalize to allow a different mid container type

---
 dogsdogsdogs/src/operators/half_join.rs | 18 +++++++++++-------
 1 file changed, 11 insertions(+), 7 deletions(-)

diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
index c1a72572a..f1ee2a07d 100644
--- a/dogsdogsdogs/src/operators/half_join.rs
+++ b/dogsdogsdogs/src/operators/half_join.rs
@@ -55,9 +55,13 @@ pub trait HalfJoinTactic<B: BatchReader, C0, C1> {
 ///
 /// The implementor is able to determine the meaning of extraction by a frontier;
 /// it is not required to be by antichain partial order.
-pub trait Batcher<C, T: Timestamp> {
+///
+/// Updates are accepted as `C0`, the containers that arrive on the dataflow edge, and released as
+/// `C1`, whatever a [`HalfJoinTactic`] would rather consume. The two need not agree: an implementor
+/// staging updates in a form of its own can release that form directly.
+pub trait Batcher<T: Timestamp, C0, C1> {
     /// Moves responsibility for `container` into the implementor.
-    fn insert(&mut self, container: C);
+    fn insert(&mut self, container: C0);
     /// Extracts updates `frontier` unblocks, and lower bounds the time of retained updates.
     ///
     /// What `frontier` unblocks is the implementor's to decide. It can be based on the antichain up set,
@@ -66,7 +70,7 @@ pub trait Batcher<C, T: Timestamp> {
     /// The reported lower bound antichain should accurately reflect the times of all accepted updates
     /// that have not been extracted. Over approximation can result in stalling dataflows, and under
     /// approximation is simply incorrect.
-    fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec<C>, &MutableAntichain<T>);
+    fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec<C1>, &MutableAntichain<T>);
 }
 
 /// A `half_join` driven by a [`Batcher`] and a [`HalfJoinTactic`].
@@ -81,7 +85,7 @@ pub trait Batcher<C, T: Timestamp> {
 ///
 /// The `pact` distributes the streamed updates, which should land each at the appropriate worker.
 /// A likely implementation is an exchange by a hash of a common key.
-pub fn half_join_with_tactic<'scope, Tr, FF, P, Y, Bat, Tac, CIn, C>(
+pub fn half_join_with_tactic<'scope, Tr, FF, P, Y, Bat, Tac, CIn, CMid, C>(
     stream: Stream<'scope, Tr::Time, CIn>,
     mut arrangement: Arranged<'scope, Tr>,
     pact: P,
@@ -95,8 +99,8 @@ where
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
     P: ParallelizationContract<Tr::Time, CIn>,
     Y: Fn(std::time::Instant, usize) -> bool + 'static,
-    Bat: Batcher<CIn, Tr::Time> + 'static,
-    Tac: HalfJoinTactic<Tr::Batch, CIn, C> + 'static,
+    Bat: Batcher<Tr::Time, CIn, CMid> + 'static,
+    Tac: HalfJoinTactic<Tr::Batch, CMid, C> + 'static,
     CIn: Container,
     C: Container + 'static,
 {
@@ -501,7 +505,7 @@ pub mod cursors {
         }
     }
 
-    impl<D: Ord, T: Timestamp, R: Semigroup> Batcher<Vec<(D, T, R)>, T> for BlobList<D, T, R> {
+    impl<D: Ord, T: Timestamp, R: Semigroup> Batcher<T, Vec<(D, T, R)>, Vec<(D, T, R)>> for BlobList<D, T, R> {
         fn insert(&mut self, container: Vec<(D, T, R)>) {
             self.stage.extend(container.into_iter().map(|(d,t,r)| (t,d,r)));
         }