From cc495df1233c1b77f52286fbf70e689adf4a5c3a Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 8 Jul 2026 13:04:54 -0400
Subject: [PATCH 1/5] join: reshape JoinTactic to an iterator of output
 containers

Replace the two-method JoinTactic (defer + fuel-metered work) with a single
`prep(input0, input1, fresh, meet) -> Box<dyn Iterator<Item = CB::Container>>`.

The tactic now maps two batch lists to an iterator of output containers and
holds neither capabilities nor a fuel budget, mirroring ReduceTactic's
data-to-data division of labor. The driver owns the two per-direction queues,
pairs each iterator with its shipping capability, pulls under a split fuel
budget, ships each yielded container, and drops a unit when its iterator goes
dry. "Work remains" becomes driver-observable (an iterator has yet to yield
None) rather than a fuel-sign protocol the tactic must self-report.

Join closures now push into a bare `&mut EffortBuilder<CB>` rather than a
timely `JoinSession`. The session is bound to a live operator output and is
not locally constructible, so it could not back a self-contained iterator;
the arbitrary-container capability #478 wanted comes from the ContainerBuilder,
not the session, so a bare builder suffices. `JoinSession` is removed.

The cursor tactic shares its logic closure across units via Rc<RefCell<_>>
(each unit is a 'static iterator and cannot borrow the tactic), preserving the
single-mutable-state semantics of one closure threaded through every match.
Deferred becomes DeferredIter: next() plays the same merge-join loop forward,
suspending at container boundaries instead of under a fuel budget.

Open for review:
- EffortBuilder's record counter is now vestigial; the type reduces to a bare
  give-providing sink, or could be removed for push_into on CB directly.
- DeferredIter::next uses mem::take on extract (needs Container: Default),
  stealing the builder's recycled allocation; a stash-swap would avoid it.
- logic's public sink type changed (JoinSession -> EffortBuilder<CB>); Mz
  exposure unverified.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/src/operators/join.rs | 335 +++++++++++---------
 1 file changed, 184 insertions(+), 151 deletions(-)

diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index 72b2085e9..efe27be81 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -4,29 +4,43 @@
 //! the multiplication distributes over addition. That is, we will repeatedly evaluate (a + b) * c as (a * c)
 //! + (b * c), and if this is not equal to the former term, little is known about the actual output.
 use std::cmp::Ordering;
+use std::cell::RefCell;
+use std::collections::VecDeque;
+use std::rc::Rc;
 
 use timely::{Accountable, ContainerBuilder};
 use timely::container::PushInto;
 use timely::order::PartialOrder;
 use timely::progress::Timestamp;
 use timely::dataflow::Stream;
-use timely::dataflow::operators::generic::{Operator, OutputBuilderSession, Session};
+use timely::dataflow::operators::generic::{Operator, OutputBuilderSession};
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::Capability;
 
 use crate::lattice::Lattice;
 use crate::operators::arrange::Arranged;
 use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, Cursor, Navigable, TraceReader};
-use crate::trace::cursor::{cursor_list, CursorList};
+use crate::trace::cursor::cursor_list;
 use crate::operators::ValueHistory;
 
-/// The session passed to join closures.
-pub type JoinSession<'a, 'b, T, CB, CT> = Session<'a, 'b, T, EffortBuilder<CB>, CT>;
-
-/// A container builder that tracks the length of outputs to estimate the effort of join closures.
+/// The sink presented to join closures: a thin [`ContainerBuilder`] wrapper providing [`give`](Self::give).
+///
+/// The wrapped `usize` cell once counted built records, so the driver could meter fuel from a closure's
+/// output volume. The driver now meters fuel from the record counts of the containers each prepared unit
+/// yields, so the counter is no longer read. This leaves the type a candidate for reduction to a bare
+/// sink, or removal in favor of pushing into `CB` directly (at the cost of the `give` ergonomics that
+/// closures rely on). Left in place pending review.
 #[derive(Default, Debug)]
 pub struct EffortBuilder<CB>(pub std::cell::Cell<usize>, pub CB);
 
+impl<CB: ContainerBuilder> EffortBuilder<CB> {
+    /// Push one output record into the builder. This is the sink presented to join closures.
+    #[inline]
+    pub fn give<D>(&mut self, item: D) where Self: PushInto<D> {
+        self.push_into(item);
+    }
+}
+
 impl<CB: ContainerBuilder> timely::container::ContainerBuilder for EffortBuilder<CB> {
     type Container = CB::Container;
 
@@ -53,22 +67,21 @@ impl<CB: PushInto<D>, D> PushInto<D> for EffortBuilder<CB> {
 }
 
 /// A type that can manage the joining of lists of batches.
+///
+/// A tactic maps two lists of batches to an iterator of output containers; it holds neither
+/// capabilities nor a fuel budget. The driver ([`join_with_tactic`]) pairs the returned iterator with
+/// the capability under which to ship its output, pulls it under a fuel budget, ships each yielded
+/// container, and drops the unit when the iterator goes dry. The iterator is the suspension mechanism:
+/// a container the driver is free to stop reading. Because "work remains" is just "the iterator has
+/// not yet yielded `None`," dryness is driver-observable rather than a protocol the tactic reports.
 pub(crate) trait JoinTactic<B0: BatchReader, B1: BatchReader<Time = B0::Time>, CB: ContainerBuilder> {
-    /// Prepare for work the join of two lists of corresponding batches, against a sufficient capability.
+    /// Prepare the join of two lists of corresponding batches into an iterator of output containers.
     ///
     /// `fresh` names which input contributed the freshly-arrived batch; its times all lie at or beyond
-    /// the capability, so a tactic need not advance that side by the capability's meet.
-    fn defer(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, capability: Capability<B0::Time>);
-    /// Perform an amount of work that just barely exceeds `fuel`, which is decremented.
-    ///
-    /// **Fuel protocol.** On return, `fuel` is non-negative *iff* all outstanding work is exhausted,
-    /// and left negative exactly when work remains. The driver ([`join_with_tactic`]) reschedules the
-    /// operator iff `fuel < 0`, so returning non-negative with work still queued silently drops it
-    /// (and returning negative with nothing to do spins). The protocol is not driver-checkable — the
-    /// work-remaining state is tactic-internal — so derive the sign *from* that state rather than
-    /// tracking it separately, and it holds by construction (as the in-tree tactic does, setting
-    /// `fuel` from whether its queues are empty).
-    fn work(&mut self, fuel: &mut isize, output: &mut OutputBuilderSession<B0::Time, EffortBuilder<CB>>);
+    /// `meet`, so a tactic need not advance that side by `meet`. `meet` is the time of the capability
+    /// the driver will ship this unit's output under — the lower envelope at which output is produced,
+    /// which the tactic may use to consolidate the accumulated side before the cross-product.
+    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = CB::Container>>;
 }
 
 /// Which input contributed the freshly-arrived batch of a deferred join unit.
@@ -102,8 +115,8 @@ where
     Tr2: TraceReader<Batch: Navigable, Time = Tr1::Time>+'static,
     BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
     for<'a> BatchCursor<Tr2>: Cursor<Key<'a>=BatchKey<'a, Tr1>, Time = Tr1::Time>,
-    L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff<Tr1>,&BatchDiff<Tr2>,&mut JoinSession<Tr1::Time, CB, Capability<Tr1::Time>>)+'static,
-    CB: ContainerBuilder,
+    L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff<Tr1>,&BatchDiff<Tr2>,&mut EffortBuilder<CB>)+'static,
+    CB: ContainerBuilder<Container: Default> + 'static,
 {
     join_with_tactic(arranged1, arranged2, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, _>::new(result))
 }
@@ -119,7 +132,7 @@ where
     Tr1: TraceReader+'static,
     Tr2: TraceReader<Time = Tr1::Time>+'static,
     T: JoinTactic<Tr1::Batch, Tr2::Batch, CB>+'static,
-    CB: ContainerBuilder,
+    CB: ContainerBuilder + 'static,
 {
     // Rename traces for symmetry from here on out.
     let mut trace1 = arranged1.trace;
@@ -147,6 +160,13 @@ where
         let mut acknowledged1 = Antichain::from_elem(Tr1::Time::minimum());
         let mut acknowledged2 = Antichain::from_elem(Tr1::Time::minimum());
 
+        // Deferred work, as `(capability, iterator)` pairs bucketed by which input carried the fresh
+        // batch (so a burst on one input cannot starve the other). The driver owns the capabilities and
+        // the fuel budget; each iterator, prepared by the tactic, yields the output containers to ship
+        // under its paired capability, and is dropped once it goes dry.
+        let mut todo0: VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = CB::Container>>)> = VecDeque::new();
+        let mut todo1: VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = CB::Container>>)> = VecDeque::new();
+
         // We'll unload the initial batches here, to put ourselves in a less non-deterministic state to start.
         trace1.map_batches(|batch1| {
             acknowledged1.clone_from(batch1.upper());
@@ -183,7 +203,8 @@ where
             // TODO: downgrade the capability by searching out the one time in `batch2.lower()` and not
             // in `batch2.upper()`. Only necessary for non-empty batches, as empty batches may not have
             // that property.
-            tactic.defer(trace1_storage, vec![batch2], Fresh::Input1, capability.clone());
+            let work = tactic.prep(trace1_storage, vec![batch2], Fresh::Input1, capability.time().clone());
+            todo1.push_back((capability.clone(), work));
         }
 
         // Droppable handles to shared trace data structures.
@@ -217,7 +238,8 @@ where
                                 // It is safe to ask for `ack2` as we validated that it was at least `get_physical_compaction()`
                                 // at start-up, and have held back physical compaction ever since.
                                 let trace2_storage = trace2.batches_through(acknowledged2.borrow()).unwrap();
-                                tactic.defer(vec![batch1.clone()], trace2_storage, Fresh::Input0, capability.clone());
+                                let work = tactic.prep(vec![batch1.clone()], trace2_storage, Fresh::Input0, capability.time().clone());
+                                todo0.push_back((capability.clone(), work));
                             }
 
                             // To update `acknowledged1` we might presume that `batch1.lower` should equal it, but we
@@ -243,7 +265,8 @@ where
                                 // It is safe to ask for `ack1` as we validated that it was at least `get_physical_compaction()`
                                 // at start-up, and have held back physical compaction ever since.
                                 let trace1_storage = trace1.batches_through(acknowledged1.borrow()).unwrap();
-                                tactic.defer(trace1_storage, vec![batch2.clone()], Fresh::Input1, capability.clone());
+                                let work = tactic.prep(trace1_storage, vec![batch2.clone()], Fresh::Input1, capability.time().clone());
+                                todo1.push_back((capability.clone(), work));
                             }
 
                             // To update `acknowledged2` we might presume that `batch2.lower` should equal it, but we
@@ -275,13 +298,28 @@ where
             // which results in unintentionally quadratic processing time (each batch of either
             // input must scan all batches from the other input).
 
-            // Perform some amount of outstanding work. The tactic decrements `fuel` as it works, and
-            // leaves it negative exactly when work remains; we reschedule the operator in that case.
-            // The two inputs' work shares this budget, so we set it to `2_000_000` to preserve the
-            // historical `1_000_000` of progress per input each activation.
-            let mut fuel = 2_000_000;
-            tactic.work(&mut fuel, output);
-            if fuel < 0 {
+            // Perform some amount of outstanding work by pulling the deferred iterators and shipping the
+            // containers they yield. Each direction drains against its own half of the budget, so a burst
+            // on one input cannot starve the other. We reschedule the operator whenever any work remains,
+            // which is observable directly: an iterator has yet to yield `None`. The budget is split from
+            // `2_000_000` to preserve the historical `1_000_000` of progress per input each activation.
+            let output: &mut OutputBuilderSession<'_, Tr1::Time, CB> = output;
+            let mut drain = |queue: &mut VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = CB::Container>>)>, mut fuel: isize| {
+                while fuel >= 0 {
+                    let Some((capability, work)) = queue.front_mut() else { break };
+                    match work.next() {
+                        Some(mut container) => {
+                            fuel -= container.record_count() as isize;
+                            output.session_with_builder(&*capability).give_container(&mut container);
+                        }
+                        None => { queue.pop_front(); }
+                    }
+                }
+            };
+            let fuel = 2_000_000;
+            drain(&mut todo0, fuel / 2);
+            drain(&mut todo1, fuel / 2);
+            if !todo0.is_empty() || !todo1.is_empty() {
                 activator.activate();
             }
 
@@ -330,13 +368,16 @@ where
 /// Cursor-based join: the conventional [`JoinTactic`] implementation and its per-batch worker.
 mod cursors {
     use super::*;
+    use timely::container::ContainerBuilder as _;
 
     /// The conventional cursor-based [`JoinTactic`].
     ///
     /// It builds a [`CursorList`] over each input batch list and plays the merge-join out at whatever rate
-    /// the operator's fuel allows. Each unit joins a `B0`-side cursor against a `B1`-side cursor, emitting
-    /// `(val0, val1)` to `logic`. The two arrival directions are queued separately and drained against
-    /// independent fuel, so a burst on one input cannot starve the other.
+    /// the driver's fuel allows. Each prepared unit joins a `B0`-side cursor against a `B1`-side cursor,
+    /// emitting `(val0, val1)` to `logic` and yielding the output containers `logic` fills. `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<B0, B1, L>
     where
         B0: BatchReader + Navigable,
@@ -344,11 +385,8 @@ mod cursors {
         B0::Cursor: Cursor<Time = B0::Time>,
         B1::Cursor: for<'a> Cursor<Key<'a> = <B0::Cursor as Cursor>::Key<'a>, Time = B0::Time>,
     {
-        logic: L,
-        /// Units whose fresh batch arrived on the first input (`B0`); their accumulated side is `B1`.
-        todo0: std::collections::VecDeque<Deferred<B0::Time, CursorList<B0::Cursor>, CursorList<B1::Cursor>>>,
-        /// Units whose fresh batch arrived on the second input (`B1`); their accumulated side is `B0`.
-        todo1: std::collections::VecDeque<Deferred<B0::Time, CursorList<B0::Cursor>, CursorList<B1::Cursor>>>,
+        logic: Rc<RefCell<L>>,
+        _marker: std::marker::PhantomData<(B0, B1)>,
     }
 
     impl<B0, B1, L> CursorTactic<B0, B1, L>
@@ -360,167 +398,162 @@ mod cursors {
     {
         /// Construct a tactic that applies `logic` to each matched `(key, val0, val1)`.
         pub fn new(logic: L) -> Self {
-            CursorTactic { logic, todo0: std::collections::VecDeque::new(), todo1: std::collections::VecDeque::new() }
+            CursorTactic { logic: Rc::new(RefCell::new(logic)), _marker: std::marker::PhantomData }
         }
     }
 
     impl<B0, B1, L, CB> JoinTactic<B0, B1, CB> for CursorTactic<B0, B1, L>
     where
-        B0: BatchReader + Navigable,
-        B1: BatchReader<Time = B0::Time> + Navigable,
+        B0: BatchReader + Navigable + 'static,
+        B1: BatchReader<Time = B0::Time> + Navigable + 'static,
         B0::Cursor: Cursor<Time = B0::Time>,
         B1::Cursor: for<'a> Cursor<Key<'a> = <B0::Cursor as Cursor>::Key<'a>, Time = B0::Time>,
-        CB: ContainerBuilder,
-        L: for<'a> FnMut(<B0::Cursor as Cursor>::Key<'a>, <B0::Cursor as Cursor>::Val<'a>, <B1::Cursor as Cursor>::Val<'a>, B0::Time, &<B0::Cursor as Cursor>::Diff, &<B1::Cursor as Cursor>::Diff, &mut JoinSession<B0::Time, CB, Capability<B0::Time>>),
+        CB: ContainerBuilder<Container: Default> + 'static,
+        L: for<'a> FnMut(<B0::Cursor as Cursor>::Key<'a>, <B0::Cursor as Cursor>::Val<'a>, <B1::Cursor as Cursor>::Val<'a>, B0::Time, &<B0::Cursor as Cursor>::Diff, &<B1::Cursor as Cursor>::Diff, &mut EffortBuilder<CB>) + 'static,
     {
-        fn defer(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, capability: Capability<B0::Time>) {
-            // Advance the accumulated trace's history by `meet` to consolidate it before the cross-product;
-            // leave the fresh batch, whose times already lie at or beyond `meet` (the capability is held at
-            // or below them). `fresh` also selects the queue, so the two arrival directions drain against
-            // independent fuel and neither can starve the other.
-            //
-            // We advance the accumulated side unconditionally. The advance is output-neutral either way (the
-            // fresh side's times are at or beyond `meet`, so the joined time is too), so this is purely a
-            // consolidation: it pays off when the accumulated side carries times below `meet`, and is a wasted
-            // scan when it does not. A more precise rule would skip the scan when the side is already entirely
-            // at or beyond `meet`, but detecting that needs both frontiers, not just `lower`: a batch's times
-            // lie at or beyond both its `lower` and its `since`, so the side is entirely beyond `meet` exactly
-            // when `meet <= lower` or `meet <= since`. A fresh batch is caught by `lower` (its `since` is
-            // `minimum`), a compacted trace by `since` (its `lower` is `minimum`); checking `lower` alone would
-            // wrongly advance a compacted trace whose times are all already at or beyond `meet`. We keep the
-            // simpler fresh-based choice and accept the occasional no-op scan.
-            let (cursor0, storage0) = cursor_list(input0);
-            let (cursor1, storage1) = cursor_list(input1);
-            match fresh {
-                Fresh::Input0 => {
-                    let deferred = Deferred::new(cursor0, storage0, cursor1, storage1, capability, false, true);
-                    self.todo0.push_back(deferred);
-                }
-                Fresh::Input1 => {
-                    let deferred = Deferred::new(cursor0, storage0, cursor1, storage1, capability, true, false);
-                    self.todo1.push_back(deferred);
-                }
-            }
-        }
-
-        fn work(&mut self, fuel: &mut isize, output: &mut OutputBuilderSession<B0::Time, EffortBuilder<CB>>) {
-            // Drain each direction against its own half of the budget, so a burst on one input cannot starve
-            // the other. Within a direction the front unit decrements its sub-budget (possibly below zero)
-            // and we stop that direction once it goes negative. If either queue still has work afterward we
-            // drive `fuel` negative so the operator reschedules; an empty pair leaves it non-negative.
-            let mut fuel0 = *fuel / 2;
-            while fuel0 >= 0 {
-                let Some(front) = self.todo0.front_mut() else { break };
-                front.work(output, &mut self.logic, &mut fuel0);
-                if !front.work_remains() { self.todo0.pop_front(); }
-            }
-            let mut fuel1 = *fuel / 2;
-            while fuel1 >= 0 {
-                let Some(front) = self.todo1.front_mut() else { break };
-                front.work(output, &mut self.logic, &mut fuel1);
-                if !front.work_remains() { self.todo1.pop_front(); }
-            }
-            *fuel = if self.todo0.is_empty() && self.todo1.is_empty() { 0 } else { -1 };
+        fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = CB::Container>> {
+            // The accumulated side's history is advanced by `meet` to consolidate it before the
+            // cross-product; the fresh side is left, as its times already lie at or beyond `meet`. `fresh`
+            // fixes which side is which. The advance is output-neutral either way (the fresh side's times are
+            // at or beyond `meet`, so the joined time is too), so it is purely a consolidation: it pays off
+            // when the accumulated side carries times below `meet`, and is a wasted scan when it does not. A
+            // more precise rule would skip the scan when the side is already entirely at or beyond `meet`,
+            // but detecting that needs both frontiers, not just `lower`: a batch's times lie at or beyond
+            // both its `lower` and its `since`, so the side is entirely beyond `meet` exactly when
+            // `meet <= lower` or `meet <= since`. A fresh batch is caught by `lower` (its `since` is
+            // `minimum`), a compacted trace by `since` (its `lower` is `minimum`); checking `lower` alone
+            // would wrongly advance a compacted trace whose times are all already at or beyond `meet`. We
+            // keep the simpler fresh-based choice and accept the occasional no-op scan.
+            let (cursor1, storage1) = cursor_list(input0);
+            let (cursor2, storage2) = cursor_list(input1);
+            let (advance1, advance2) = match fresh {
+                Fresh::Input0 => (false, true),
+                Fresh::Input1 => (true, false),
+            };
+            Box::new(DeferredIter {
+                cursor1,
+                storage1,
+                cursor2,
+                storage2,
+                meet,
+                advance1,
+                advance2,
+                logic: Rc::clone(&self.logic),
+                builder: EffortBuilder::default(),
+                ready: VecDeque::new(),
+                done: false,
+            })
         }
     }
 
-    /// Deferred join computation.
+    /// Deferred join computation, as an iterator of output containers.
     ///
     /// The structure wraps cursors which allow us to play out join computation at whatever rate we like.
     /// This allows us to avoid producing and buffering massive amounts of data, without giving the timely
-    /// dataflow system a chance to run operators that can consume and aggregate the data.
-    struct Deferred<T, C1, C2>
+    /// dataflow system a chance to run operators that can consume and aggregate the data. Each `next` plays
+    /// the merge-join forward until the builder yields a container (or the cursors run dry), matching the
+    /// former per-unit `work` loop but suspending at container boundaries rather than under a fuel budget:
+    /// the driver stops pulling once its budget is spent and resumes the same iterator next activation.
+    struct DeferredIter<T, C1, C2, L, CB>
     where
         T: Timestamp+Lattice,
         C1: Cursor<Time=T>,
         C2: for<'a> Cursor<Key<'a>=C1::Key<'a>, Time=T>,
+        CB: ContainerBuilder,
     {
         cursor1: C1,
         storage1: C1::Storage,
         cursor2: C2,
         storage2: C2::Storage,
-        capability: Capability<T>,
-        /// Whether to advance each side's history by the capability's meet before consolidation.
+        /// The capability's time, at which this unit's output ships; the lower envelope for consolidation.
+        meet: T,
+        /// Whether to advance each side's history by `meet` before consolidation.
         advance1: bool,
         advance2: bool,
+        /// The output closure, shared across all outstanding units.
+        logic: Rc<RefCell<L>>,
+        /// The builder `logic` fills; drained into `ready` as containers complete.
+        builder: EffortBuilder<CB>,
+        /// Completed containers awaiting a `next` call.
+        ready: VecDeque<CB::Container>,
         done: bool,
     }
 
-    impl<T, C1, C2> Deferred<T, C1, C2>
+    impl<T, C1, C2, L, CB> Iterator for DeferredIter<T, C1, C2, L, CB>
     where
+        T: Timestamp+Lattice,
         C1: Cursor<Time=T>,
         C2: for<'a> Cursor<Key<'a>=C1::Key<'a>, Time=T>,
-        T: Timestamp+Lattice,
+        CB: ContainerBuilder<Container: Default>,
+        L: for<'a> FnMut(C1::Key<'a>, C1::Val<'a>, C2::Val<'a>, T, &C1::Diff, &C2::Diff, &mut EffortBuilder<CB>),
     {
-        fn new(cursor1: C1, storage1: C1::Storage, cursor2: C2, storage2: C2::Storage, capability: Capability<T>, advance1: bool, advance2: bool) -> Self {
-            Deferred {
-                cursor1,
-                storage1,
-                cursor2,
-                storage2,
-                capability,
-                advance1,
-                advance2,
-                done: false,
-            }
-        }
-
-        fn work_remains(&self) -> bool {
-            !self.done
-        }
+        type Item = CB::Container;
 
-        /// Process keys until `fuel` is driven below zero, or the work is exhausted.
+        /// Play the merge-join forward until a container is ready, or the cursors run dry.
         #[inline(never)]
-        fn work<L, CB: ContainerBuilder>(&mut self, output: &mut OutputBuilderSession<T, EffortBuilder<CB>>, logic: &mut L, fuel: &mut isize)
-        where
-            L: for<'a> FnMut(C1::Key<'a>, C1::Val<'a>, C2::Val<'a>, T, &C1::Diff, &C2::Diff, &mut JoinSession<T, CB, Capability<T>>),
-        {
-
-            let meet = self.capability.time();
+        fn next(&mut self) -> Option<CB::Container> {
+            // Serve any container completed on an earlier call first.
+            if let Some(container) = self.ready.pop_front() { return Some(container); }
+            if self.done { return None; }
 
-            // Advance the accumulated side by `meet` to consolidate its history; leave the fresh side, whose
-            // times already lie at or beyond `meet`. The choice was fixed per side at construction, from
-            // which input carried the fresh batch.
-            let meet1 = if self.advance1 { Some(meet) } else { None };
-            let meet2 = if self.advance2 { Some(meet) } else { None };
-
-            let mut session = output.session_with_builder(&self.capability);
+            // The accumulated side is advanced by `meet` to consolidate its history; the fresh side is left,
+            // as its times already lie at or beyond `meet`. The choice was fixed per side at construction,
+            // from which input carried the fresh batch.
+            let meet1 = if self.advance1 { Some(&self.meet) } else { None };
+            let meet2 = if self.advance2 { Some(&self.meet) } else { None };
 
             let storage1 = &self.storage1;
             let storage2 = &self.storage2;
-
             let cursor1 = &mut self.cursor1;
             let cursor2 = &mut self.cursor2;
+            let builder = &mut self.builder;
+            let ready = &mut self.ready;
+            let mut logic = self.logic.borrow_mut();
+            let logic = &mut *logic;
 
             let mut thinker = JoinThinker::new();
+            let mut exhausted = false;
 
-            while let (Some(key1), Some(key2), true) = (cursor1.get_key(storage1), cursor2.get_key(storage2), *fuel >= 0) {
+            while ready.is_empty() {
+                match (cursor1.get_key(storage1), cursor2.get_key(storage2)) {
+                    (Some(key1), Some(key2)) => match key1.cmp(&key2) {
+                        Ordering::Less => cursor1.seek_key(storage1, key2),
+                        Ordering::Greater => cursor2.seek_key(storage2, key1),
+                        Ordering::Equal => {
 
-                match key1.cmp(&key2) {
-                    Ordering::Less => cursor1.seek_key(storage1, key2),
-                    Ordering::Greater => cursor2.seek_key(storage2, key1),
-                    Ordering::Equal => {
+                            thinker.history1.edits.load(cursor1, storage1, meet1);
+                            thinker.history2.edits.load(cursor2, storage2, meet2);
 
-                        thinker.history1.edits.load(cursor1, storage1, meet1);
-                        thinker.history2.edits.load(cursor2, storage2, meet2);
+                            thinker.think(|v1,v2,t,r1,r2| {
+                                logic(key1, v1, v2, t, r1, r2, builder);
+                            });
 
-                        // populate `temp` with the results in the best way we know how.
-                        thinker.think(|v1,v2,t,r1,r2| {
-                            logic(key1, v1, v2, t, r1, r2, &mut session);
-                        });
+                            cursor1.step_key(storage1);
+                            cursor2.step_key(storage2);
 
-                        // TODO: Effort isn't perfectly tracked as we might still have some data in the
-                        // session at the moment it's dropped.
-                        *fuel -= session.builder().0.take() as isize;
-                        cursor1.step_key(storage1);
-                        cursor2.step_key(storage2);
+                            thinker.history1.clear();
+                            thinker.history2.clear();
 
-                        thinker.history1.clear();
-                        thinker.history2.clear();
-                    }
+                            // Move any completed containers aside; we yield them one at a time.
+                            while let Some(container) = builder.extract() {
+                                ready.push_back(std::mem::take(container));
+                            }
+                        }
+                    },
+                    // One side is exhausted; no further keys can match.
+                    _ => { exhausted = true; break; }
+                }
+            }
+
+            if exhausted {
+                self.done = true;
+                // Flush the final partial container.
+                while let Some(container) = builder.finish() {
+                    ready.push_back(std::mem::take(container));
                 }
             }
-            self.done = !cursor1.key_valid(storage1) || !cursor2.key_valid(storage2);
+
+            ready.pop_front()
         }
     }
 

From c80c4546320c48c5bc79a2225e8dff62e51ccec7 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 8 Jul 2026 13:15:47 -0400
Subject: [PATCH 2/5] join: remove the now-vestigial EffortBuilder

Its only remaining job was renaming `push_into` to `give`; the record counter
that justified the wrapper went dead when the driver began metering fuel from
shipped containers' `record_count()`. Join closures now receive `&mut CB` and
call `push_into` directly. `EffortBuilder` is deleted.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../src/operators/arrange/arrangement.rs      |  2 +-
 differential-dataflow/src/operators/join.rs   | 55 ++-----------------
 2 files changed, 6 insertions(+), 51 deletions(-)

diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index bb7f74f6f..359ad8945 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -251,7 +251,7 @@ impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
             other,
             move |k, v1, v2, t, d1, d2, c| {
                 for datum in result(k, v1, v2, t, d1, d2) {
-                    c.give(datum);
+                    c.push_into(datum);
                 }
             }
         )
diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index efe27be81..6160a5ab2 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -9,7 +9,6 @@ use std::collections::VecDeque;
 use std::rc::Rc;
 
 use timely::{Accountable, ContainerBuilder};
-use timely::container::PushInto;
 use timely::order::PartialOrder;
 use timely::progress::Timestamp;
 use timely::dataflow::Stream;
@@ -23,49 +22,6 @@ use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, Curs
 use crate::trace::cursor::cursor_list;
 use crate::operators::ValueHistory;
 
-/// The sink presented to join closures: a thin [`ContainerBuilder`] wrapper providing [`give`](Self::give).
-///
-/// The wrapped `usize` cell once counted built records, so the driver could meter fuel from a closure's
-/// output volume. The driver now meters fuel from the record counts of the containers each prepared unit
-/// yields, so the counter is no longer read. This leaves the type a candidate for reduction to a bare
-/// sink, or removal in favor of pushing into `CB` directly (at the cost of the `give` ergonomics that
-/// closures rely on). Left in place pending review.
-#[derive(Default, Debug)]
-pub struct EffortBuilder<CB>(pub std::cell::Cell<usize>, pub CB);
-
-impl<CB: ContainerBuilder> EffortBuilder<CB> {
-    /// Push one output record into the builder. This is the sink presented to join closures.
-    #[inline]
-    pub fn give<D>(&mut self, item: D) where Self: PushInto<D> {
-        self.push_into(item);
-    }
-}
-
-impl<CB: ContainerBuilder> timely::container::ContainerBuilder for EffortBuilder<CB> {
-    type Container = CB::Container;
-
-    #[inline]
-    fn extract(&mut self) -> Option<&mut Self::Container> {
-        let extracted = self.1.extract();
-        self.0.replace(self.0.take() + extracted.as_ref().map_or(0, |e| e.record_count() as usize));
-        extracted
-    }
-
-    #[inline]
-    fn finish(&mut self) -> Option<&mut Self::Container> {
-        let finished = self.1.finish();
-        self.0.replace(self.0.take() + finished.as_ref().map_or(0, |e| e.record_count() as usize));
-        finished
-    }
-}
-
-impl<CB: PushInto<D>, D> PushInto<D> for EffortBuilder<CB> {
-    #[inline]
-    fn push_into(&mut self, item: D) {
-        self.1.push_into(item);
-    }
-}
-
 /// A type that can manage the joining of lists of batches.
 ///
 /// A tactic maps two lists of batches to an iterator of output containers; it holds neither
@@ -115,7 +71,7 @@ where
     Tr2: TraceReader<Batch: Navigable, Time = Tr1::Time>+'static,
     BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
     for<'a> BatchCursor<Tr2>: Cursor<Key<'a>=BatchKey<'a, Tr1>, Time = Tr1::Time>,
-    L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff<Tr1>,&BatchDiff<Tr2>,&mut EffortBuilder<CB>)+'static,
+    L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff<Tr1>,&BatchDiff<Tr2>,&mut CB)+'static,
     CB: ContainerBuilder<Container: Default> + 'static,
 {
     join_with_tactic(arranged1, arranged2, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, _>::new(result))
@@ -368,7 +324,6 @@ where
 /// Cursor-based join: the conventional [`JoinTactic`] implementation and its per-batch worker.
 mod cursors {
     use super::*;
-    use timely::container::ContainerBuilder as _;
 
     /// The conventional cursor-based [`JoinTactic`].
     ///
@@ -409,7 +364,7 @@ mod cursors {
         B0::Cursor: Cursor<Time = B0::Time>,
         B1::Cursor: for<'a> Cursor<Key<'a> = <B0::Cursor as Cursor>::Key<'a>, Time = B0::Time>,
         CB: ContainerBuilder<Container: Default> + 'static,
-        L: for<'a> FnMut(<B0::Cursor as Cursor>::Key<'a>, <B0::Cursor as Cursor>::Val<'a>, <B1::Cursor as Cursor>::Val<'a>, B0::Time, &<B0::Cursor as Cursor>::Diff, &<B1::Cursor as Cursor>::Diff, &mut EffortBuilder<CB>) + 'static,
+        L: for<'a> FnMut(<B0::Cursor as Cursor>::Key<'a>, <B0::Cursor as Cursor>::Val<'a>, <B1::Cursor as Cursor>::Val<'a>, B0::Time, &<B0::Cursor as Cursor>::Diff, &<B1::Cursor as Cursor>::Diff, &mut CB) + 'static,
     {
         fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = CB::Container>> {
             // The accumulated side's history is advanced by `meet` to consolidate it before the
@@ -439,7 +394,7 @@ mod cursors {
                 advance1,
                 advance2,
                 logic: Rc::clone(&self.logic),
-                builder: EffortBuilder::default(),
+                builder: CB::default(),
                 ready: VecDeque::new(),
                 done: false,
             })
@@ -473,7 +428,7 @@ mod cursors {
         /// The output closure, shared across all outstanding units.
         logic: Rc<RefCell<L>>,
         /// The builder `logic` fills; drained into `ready` as containers complete.
-        builder: EffortBuilder<CB>,
+        builder: CB,
         /// Completed containers awaiting a `next` call.
         ready: VecDeque<CB::Container>,
         done: bool,
@@ -485,7 +440,7 @@ mod cursors {
         C1: Cursor<Time=T>,
         C2: for<'a> Cursor<Key<'a>=C1::Key<'a>, Time=T>,
         CB: ContainerBuilder<Container: Default>,
-        L: for<'a> FnMut(C1::Key<'a>, C1::Val<'a>, C2::Val<'a>, T, &C1::Diff, &C2::Diff, &mut EffortBuilder<CB>),
+        L: for<'a> FnMut(C1::Key<'a>, C1::Val<'a>, C2::Val<'a>, T, &C1::Diff, &C2::Diff, &mut CB),
     {
         type Item = CB::Container;
 

From 449d2a4141eeb1aed28932a7b9736157fe584d5d Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 8 Jul 2026 13:22:52 -0400
Subject: [PATCH 3/5] join: fix the columnar example's join sink caller

The reachability example pushed into the join sink via `session.give(..)`, an
inherent method on the removed `EffortBuilder`. The sink is now `&mut CB`, so it
calls `push_into` directly and imports `timely::container::PushInto` in the
`reachability` module (the old inherent `give` needed no trait in scope).

Only surfaces under `--all-targets`, which is why `cargo build --workspace`
missed it but CI did not.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/examples/columnar/main.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs
index 692f1fd27..d8775f44e 100644
--- a/differential-dataflow/examples/columnar/main.rs
+++ b/differential-dataflow/examples/columnar/main.rs
@@ -90,6 +90,7 @@ fn main() {
 /// Push on UpdatesTyped for the reduce builder path.
 mod reachability {
 
+    use timely::container::PushInto;
     use timely::order::Product;
     use differential_dataflow::Collection;
     use differential_dataflow::AsCollection;
@@ -149,7 +150,7 @@ mod reachability {
                         use differential_dataflow::difference::Multiply;
                         let dst: Node = *dst;
                         let diff: Diff = d1.clone().multiply(d2);
-                        session.give::<(Node, (), IterTime, Diff)>((dst, (), time.clone(), diff));
+                        session.push_into((dst, (), time.clone(), diff));
                     },
                 ).as_collection();
 

From 318530a30e6fe3d228aaa2cc1ce84c13ec951d91 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 8 Jul 2026 14:43:17 -0400
Subject: [PATCH 4/5] Tidy some structure

---
 differential-dataflow/src/operators/join.rs | 23 ++++++++-------------
 1 file changed, 9 insertions(+), 14 deletions(-)

diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index 6160a5ab2..d1d4e8efd 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -4,9 +4,7 @@
 //! the multiplication distributes over addition. That is, we will repeatedly evaluate (a + b) * c as (a * c)
 //! + (b * c), and if this is not equal to the former term, little is known about the actual output.
 use std::cmp::Ordering;
-use std::cell::RefCell;
 use std::collections::VecDeque;
-use std::rc::Rc;
 
 use timely::{Accountable, ContainerBuilder};
 use timely::order::PartialOrder;
@@ -23,20 +21,11 @@ use crate::trace::cursor::cursor_list;
 use crate::operators::ValueHistory;
 
 /// A type that can manage the joining of lists of batches.
-///
-/// A tactic maps two lists of batches to an iterator of output containers; it holds neither
-/// capabilities nor a fuel budget. The driver ([`join_with_tactic`]) pairs the returned iterator with
-/// the capability under which to ship its output, pulls it under a fuel budget, ships each yielded
-/// container, and drops the unit when the iterator goes dry. The iterator is the suspension mechanism:
-/// a container the driver is free to stop reading. Because "work remains" is just "the iterator has
-/// not yet yielded `None`," dryness is driver-observable rather than a protocol the tactic reports.
 pub(crate) trait JoinTactic<B0: BatchReader, B1: BatchReader<Time = B0::Time>, CB: ContainerBuilder> {
-    /// Prepare the join of two lists of corresponding batches into an iterator of output containers.
+    /// Prepare the join of two lists of batches into an iterator of output containers.
     ///
-    /// `fresh` names which input contributed the freshly-arrived batch; its times all lie at or beyond
-    /// `meet`, so a tactic need not advance that side by `meet`. `meet` is the time of the capability
-    /// the driver will ship this unit's output under — the lower envelope at which output is produced,
-    /// which the tactic may use to consolidate the accumulated side before the cross-product.
+    /// The supplied `fresh` and `meet` indicate respectively which input is "novel", and should drive the
+    /// join, as well as a lower bound on that input's times, so that the other input can be loaded compacted.
     fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = CB::Container>>;
 }
 
@@ -323,6 +312,10 @@ where
 
 /// Cursor-based join: the conventional [`JoinTactic`] implementation and its per-batch worker.
 mod cursors {
+
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
     use super::*;
 
     /// The conventional cursor-based [`JoinTactic`].
@@ -491,6 +484,7 @@ mod cursors {
 
                             // Move any completed containers aside; we yield them one at a time.
                             while let Some(container) = builder.extract() {
+                                // Avoiding the mem::take would require a non-iterator trait.
                                 ready.push_back(std::mem::take(container));
                             }
                         }
@@ -504,6 +498,7 @@ mod cursors {
                 self.done = true;
                 // Flush the final partial container.
                 while let Some(container) = builder.finish() {
+                    // Avoiding the mem::take would require a non-iterator trait.
                     ready.push_back(std::mem::take(container));
                 }
             }

From ee68e79854d5ed707a3e251d0654da9a79140363 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 8 Jul 2026 14:48:28 -0400
Subject: [PATCH 5/5] join: parameterize JoinTactic by its output container,
 not the builder
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The trait used `CB` only as `CB::Container` (the iterator's Item); it never
constructs or pushes into a builder. Parameterizing by the container `C`
instead keeps the builder where it is actually used — a private concern of the
cursor tactic, which now carries it as `CursorTactic<B0, B1, L, CB>` and
implements `JoinTactic<B0, B1, CB::Container>`.

Two things fall out. The `Container: Default` bound (needed only for the cursor
tactic's `mem::take`) leaves the trait for that impl. And the driver, which
only ships finished containers via `give_container`, no longer needs the
caller's builder: it pins the operator output to `NoopBuilder<C>` (a builder
for any `C`, no bounds) and requires only `C: Container`. This is the groundwork
for tactics that produce containers without a push-builder (proxy `cross`,
columnar bulk), which would otherwise carry a builder they never use.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/src/operators/join.rs | 41 +++++++++++++--------
 1 file changed, 26 insertions(+), 15 deletions(-)

diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index d1d4e8efd..3201c3176 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -6,7 +6,8 @@
 use std::cmp::Ordering;
 use std::collections::VecDeque;
 
-use timely::{Accountable, ContainerBuilder};
+use timely::{Container, ContainerBuilder};
+use timely::container::NoopBuilder;
 use timely::order::PartialOrder;
 use timely::progress::Timestamp;
 use timely::dataflow::Stream;
@@ -21,12 +22,16 @@ use crate::trace::cursor::cursor_list;
 use crate::operators::ValueHistory;
 
 /// A type that can manage the joining of lists of batches.
-pub(crate) trait JoinTactic<B0: BatchReader, B1: BatchReader<Time = B0::Time>, CB: ContainerBuilder> {
+///
+/// The trait is parameterized by the output container `C`, not by the builder that assembles it: a tactic
+/// yields finished containers, and how it produces them (pushing records into a [`ContainerBuilder`], or
+/// otherwise) is its own concern.
+pub(crate) trait JoinTactic<B0: BatchReader, B1: BatchReader<Time = B0::Time>, C> {
     /// Prepare the join of two lists of batches into an iterator of output containers.
     ///
     /// The supplied `fresh` and `meet` indicate respectively which input is "novel", and should drive the
     /// join, as well as a lower bound on that input's times, so that the other input can be loaded compacted.
-    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = CB::Container>>;
+    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = C>>;
 }
 
 /// Which input contributed the freshly-arrived batch of a deferred join unit.
@@ -63,7 +68,7 @@ where
     L: FnMut(BatchKey<'_, Tr1>,BatchVal<'_, Tr1>,BatchVal<'_, Tr2>,Tr1::Time,&BatchDiff<Tr1>,&BatchDiff<Tr2>,&mut CB)+'static,
     CB: ContainerBuilder<Container: Default> + 'static,
 {
-    join_with_tactic(arranged1, arranged2, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, _>::new(result))
+    join_with_tactic(arranged1, arranged2, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, _, CB>::new(result))
 }
 
 /// Drives an equijoin of two traces using a supplied [`JoinTactic`].
@@ -72,12 +77,12 @@ where
 /// compaction) and routes the per-batch work through the tactic. It requires only `TraceReader` of its
 /// inputs, never `Navigable`: it extracts trace batches via `batches_through`, and building cursors over
 /// them (if that is how the join proceeds) is the tactic's concern.
-pub(crate) fn join_with_tactic<'scope, Tr1, Tr2, T, CB>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, mut tactic: T) -> Stream<'scope, Tr1::Time, CB::Container>
+pub(crate) fn join_with_tactic<'scope, Tr1, Tr2, T, C>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, mut tactic: T) -> Stream<'scope, Tr1::Time, C>
 where
     Tr1: TraceReader+'static,
     Tr2: TraceReader<Time = Tr1::Time>+'static,
-    T: JoinTactic<Tr1::Batch, Tr2::Batch, CB>+'static,
-    CB: ContainerBuilder + 'static,
+    T: JoinTactic<Tr1::Batch, Tr2::Batch, C>+'static,
+    C: Container + 'static,
 {
     // Rename traces for symmetry from here on out.
     let mut trace1 = arranged1.trace;
@@ -109,8 +114,8 @@ where
         // batch (so a burst on one input cannot starve the other). The driver owns the capabilities and
         // the fuel budget; each iterator, prepared by the tactic, yields the output containers to ship
         // under its paired capability, and is dropped once it goes dry.
-        let mut todo0: VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = CB::Container>>)> = VecDeque::new();
-        let mut todo1: VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = CB::Container>>)> = VecDeque::new();
+        let mut todo0: VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = C>>)> = VecDeque::new();
+        let mut todo1: VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = C>>)> = VecDeque::new();
 
         // We'll unload the initial batches here, to put ourselves in a less non-deterministic state to start.
         trace1.map_batches(|batch1| {
@@ -248,8 +253,11 @@ where
             // on one input cannot starve the other. We reschedule the operator whenever any work remains,
             // which is observable directly: an iterator has yet to yield `None`. The budget is split from
             // `2_000_000` to preserve the historical `1_000_000` of progress per input each activation.
-            let output: &mut OutputBuilderSession<'_, Tr1::Time, CB> = output;
-            let mut drain = |queue: &mut VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = CB::Container>>)>, mut fuel: isize| {
+            // The driver only ships finished containers (`give_container`), never pushing records, so it
+            // pins the operator output to `NoopBuilder<C>` — the builder for exactly this "containers ready
+            // to go" case, which is a `ContainerBuilder` for any `C` without further bounds.
+            let output: &mut OutputBuilderSession<'_, Tr1::Time, NoopBuilder<C>> = output;
+            let mut drain = |queue: &mut VecDeque<(Capability<Tr1::Time>, Box<dyn Iterator<Item = C>>)>, mut fuel: isize| {
                 while fuel >= 0 {
                     let Some((capability, work)) = queue.front_mut() else { break };
                     match work.next() {
@@ -326,7 +334,10 @@ mod cursors {
     /// 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<B0, B1, L>
+    ///
+    /// It is parameterized by the builder `CB` into which `logic` pushes output; the [`JoinTactic`] it
+    /// implements is over the container `CB` yields (`CB::Container`).
+    pub struct CursorTactic<B0, B1, L, CB>
     where
         B0: BatchReader + Navigable,
         B1: BatchReader<Time = B0::Time> + Navigable,
@@ -334,10 +345,10 @@ mod cursors {
         B1::Cursor: for<'a> Cursor<Key<'a> = <B0::Cursor as Cursor>::Key<'a>, Time = B0::Time>,
     {
         logic: Rc<RefCell<L>>,
-        _marker: std::marker::PhantomData<(B0, B1)>,
+        _marker: std::marker::PhantomData<(B0, B1, CB)>,
     }
 
-    impl<B0, B1, L> CursorTactic<B0, B1, L>
+    impl<B0, B1, L, CB> CursorTactic<B0, B1, L, CB>
     where
         B0: BatchReader + Navigable,
         B1: BatchReader<Time = B0::Time> + Navigable,
@@ -350,7 +361,7 @@ mod cursors {
         }
     }
 
-    impl<B0, B1, L, CB> JoinTactic<B0, B1, CB> for CursorTactic<B0, B1, L>
+    impl<B0, B1, L, CB> JoinTactic<B0, B1, CB::Container> for CursorTactic<B0, B1, L, CB>
     where
         B0: BatchReader + Navigable + 'static,
         B1: BatchReader<Time = B0::Time> + Navigable + 'static,