From 552143c37718a6fac90e6d854bba4196627f661f Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 29 Jul 2026 12:48:10 -0400
Subject: [PATCH 1/3] int_proxy join: draw the proxy collection lazily, a block
 at a time

The proxy join performed the whole of a unit's work inside `prep`, and returned an
iterator that spilled the finished containers.
The driver's fuel budget was therefore spent after the fact.

`present0`/`present1` are replaced by a single `advance`, which populates both bridges
with the intersecting keys from `from` onward and updates `from`.
This removes the two-phase presentation, in which the whole fresh side was materialized
to derive the key filter for the other side.
A unit's progress through the key space rides in `from` rather than in the backend, so
the harness can interleave the calls of the many units it has in flight.

`cross` populates a `Vec<Self::Output>` rather than returning one container.
Both granularities are the backend's: `advance` sizes the block it draws, `cross` the
containers it cuts.
The operator's own `JOIN_CHUNK` policy is gone, along with the mid-key flushing it
needed; the operator draws a block, merge-matches it whole, and hands the matches back.
A single key's matches are held whole, as under the cursor tactic.

`JoinInstance` owns its batches, which drops its lifetime parameter and lets the
iterator hold it as a single field.

Debug assertions enforce the `advance` contract: bridges sorted and consolidated, `from`
strictly increasing or exhausted, and each reported key hash confined to the block that
first mentions it.
The last guards the case that would otherwise be silently wrong rather than a panic, and
the `from` check also guards liveness.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ErrnMtBsvrvncVeYL51TMv
---
 .../src/operators/int_proxy/join.rs           | 274 ++++++++++++------
 1 file changed, 183 insertions(+), 91 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index 7a02d1b1f..05d6b1a9d 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -3,8 +3,10 @@
 //! A conventional differential join against `(u64, u64)` values, which are provided by
 //! and then interpreted by a backend, who is relieved of lattice-time reasoning.
 
+use std::cell::RefCell;
+use std::rc::Rc;
+
 use timely::progress::{Antichain, Timestamp};
-use timely::progress::frontier::AntichainRef;
 
 use crate::difference::{Multiply, Semigroup};
 use crate::lattice::Lattice;
@@ -15,19 +17,27 @@ use crate::operators::join::{Fresh, JoinTactic};
 use super::history::IdHistory;
 
 /// A unit of proxied join work, presented to the backend.
-pub struct JoinInstance<'a, B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
+pub struct JoinInstance<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
     /// The first input's batches.
-    pub batches0: &'a [B0],
+    pub batches0: Vec<B0>,
     /// The second input's batches.
-    pub batches1: &'a [B1],
+    pub batches1: Vec<B1>,
     /// The compaction frontier for loading (the unit's capability time).
-    pub lower: AntichainRef<'a, B0::Time>,
+    pub lower: Antichain<B0::Time>,
 }
 
 /// A type that can interpret and retire pairs of batches, joined by key hashes.
 ///
-/// The protocol invokes the `present*` methods with the instance to produce the proxy collection,
-/// and then returns with any number (including zero) calls to `cross` to produce outputs.
+/// The protocol repeatedly invokes [`advance`](Self::advance) to draw a block of the proxy
+/// collection, then [`cross`](Self::cross) to turn that block's matches into output containers.
+/// The unit is retired when `advance` reports the key space exhausted.
+///
+/// The two granularities are the backend's, and independent: `advance` sizes the input it draws,
+/// `cross` sizes the containers it produces from the matches that block yields.
+///
+/// The backend holds no state across the calls of one unit: its progress lives in the `from`
+/// argument, which the harness owns. This is load-bearing, as the harness may have many units
+/// in flight against the same backend, and interleaves their calls freely.
 pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
     /// Diff type presented for the first input.
     type R0: Semigroup + Multiply<Self::R1, Output = Self::ROut>;
@@ -38,104 +48,181 @@ pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
     /// The output container built from matched value ids.
     type Output;
 
-    /// Prepare a proxy bridge from the instance's first input, optionally key-hash restricted.
+    /// Populates two bridges with intersecting keys from `from` onward, updating `from` to the new limit.
     ///
-    /// The returned bridge **must be sorted and consolidated** by `((key_hash, value_id), time)`.
-    fn present0(&mut self, instance: &JoinInstance<'_, B0, B1>, filter: Option<&[u64]>) -> ProxyBridge<B0::Time, Self::R0>;
-    /// Prepare a proxy bridge from the instance's second input, optionally key-hash restricted.
+    /// The `from` argument is an inclusive lower bound on key identifier, initially `Some(0)`, with `None`
+    /// used to indicate that the keyspace has been exhausted.
     ///
-    /// The returned bridge **must be sorted and consolidated** by `((key_hash, value_id), time)`.
-    fn present1(&mut self, instance: &JoinInstance<'_, B0, B1>, filter: Option<&[u64]>) -> ProxyBridge<B0::Time, Self::R1>;
+    /// A key hash must be reported entirely by the call that first mentions it: all of its updates,
+    /// from both inputs, and none in any other block. Equivalently, each reported key hash lies in
+    /// `[from, from')` for the argument's values before and after the call.
+    fn advance(
+        &mut self,
+        instance: &JoinInstance<B0, B1>,
+        from: &mut Option<u64>,
+        bridge0: &mut ProxyBridge<B0::Time, Self::R0>,
+        bridge1: &mut ProxyBridge<B0::Time, Self::R1>,
+    );
+
     /// From a list of left and right identifiers, and corresponding times and diffs, the output.
-    fn cross(&mut self, instance: &JoinInstance<'_, B0, B1>, left: &[(u64, u64)], right: &[(u64, u64)], times: Vec<B0::Time>, diffs: Vec<Self::ROut>) -> Self::Output;
+    ///
+    /// The four are parallel arrays with one entry per matched pair: `left[i]` and `right[i]` are the
+    /// `(key_hash, value_id)` of the pair's two records, which share a key hash, and `times[i]` and
+    /// `diffs[i]` are the join of their times and the product of their diffs. They are not
+    /// consolidated, and a pair can repeat at distinct times. Drain `times` and `diffs` to take their
+    /// contents; the harness keeps their allocations.
+    ///
+    /// The containers are pushed to `output`, which arrives empty, and are shipped one at a time:
+    /// their sizing is the backend's, independent of the block `advance` drew the matches from.
+    fn cross(&mut self, instance: &JoinInstance<B0, B1>, left: &[(u64, u64)], right: &[(u64, u64)], times: &mut Vec<B0::Time>, diffs: &mut Vec<Self::ROut>, output: &mut Vec<Self::Output>);
 }
 
-/// A proxy-space [`JoinTactic`]: matches records of the two presented runs by `key_hash`,
+/// A proxy-space [`JoinTactic`]: matches records of the two drawn runs by `key_hash`.
+///
+/// The backend is shared across all outstanding units (an `Rc<RefCell<_>>`), as each prepared unit
+/// is a self-contained `'static` iterator and cannot borrow the tactic. Units are safe to interleave
+/// because a unit's progress through the key space rides in its own `from`, not in the backend.
 pub struct ProxyJoinTactic<B0, B1, Bk> {
-    backend: Bk,
+    backend: Rc<RefCell<Bk>>,
     _marker: std::marker::PhantomData<(B0, B1)>,
 }
 
 impl<B0, B1, Bk> ProxyJoinTactic<B0, B1, Bk> {
     /// A join tactic deferring all value semantics to `backend`.
     pub fn new(backend: Bk) -> Self {
-        ProxyJoinTactic { backend, _marker: std::marker::PhantomData }
+        ProxyJoinTactic { backend: Rc::new(RefCell::new(backend)), _marker: std::marker::PhantomData }
     }
 }
 
 impl<B0, B1, Bk> JoinTactic<B0, B1, Bk::Output> for ProxyJoinTactic<B0, B1, Bk>
 where
-    B0: BatchReader,
-    B1: BatchReader<Time = B0::Time>,
-    Bk: ProxyJoinBackend<B0, B1>,
+    B0: BatchReader + 'static,
+    B1: BatchReader<Time = B0::Time> + 'static,
+    Bk: ProxyJoinBackend<B0, B1> + 'static,
     Bk::Output: 'static,
 {
-    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = Bk::Output>> {
-        Box::new(join_prep(&mut self.backend, input0, input1, fresh, meet).into_iter())
+    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, _fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = Bk::Output>> {
+        Box::new(ProxyJoinIter {
+            backend: Rc::clone(&self.backend),
+            instance: JoinInstance { batches0: input0, batches1: input1, lower: Antichain::from_elem(meet) },
+            from: Some(0),
+            p0: Vec::new(),
+            p1: Vec::new(),
+            h0: IdHistory::new(),
+            h1: IdHistory::new(),
+            li: Vec::new(),
+            ri: Vec::new(),
+            ot: Vec::new(),
+            od: Vec::new(),
+            ready: Vec::new(),
+        })
     }
 }
 
-/// Update count threshold used to return to the backend with join output to instantiate.
-const JOIN_CHUNK: usize = 1 << 20;
+/// Deferred proxy join computation, as an iterator of output containers.
+///
+/// The unit draws the proxy collection a block at a time (`Bk::advance`), merge-matches the two
+/// drawn runs by `key_hash`, and hands the matches back to be cut into containers (`Bk::cross`).
+/// Both granularities are the backend's: `advance` sizes the block, `cross` the containers. One
+/// key's matches are held whole however the block is sized, as the replay wave cannot be
+/// interrupted. The driver stops pulling once its budget is spent and resumes the same iterator on
+/// the next activation.
+///
+/// `meet` is a lower bound on the fresh side's times, and `lower` presents it to the backend
+/// (see [`JoinInstance`]). Loaded times can be advanced by it on either input: the fresh side's
+/// times dominate `meet`, so the joined time does too.
+struct ProxyJoinIter<B0, B1, Bk>
+where
+    B0: BatchReader,
+    B1: BatchReader<Time = B0::Time>,
+    Bk: ProxyJoinBackend<B0, B1>,
+{
+    /// The backend, shared across all outstanding units.
+    backend: Rc<RefCell<Bk>>,
+    /// The unit's inputs, and the compaction frontier at which they load.
+    instance: JoinInstance<B0, B1>,
+    /// Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
+    /// once the backend reports the unit drawn.
+    from: Option<u64>,
+    /// The current block: the two runs `advance` last drew, which one `next` consumes entirely.
+    p0: ProxyBridge<B0::Time, Bk::R0>,
+    p1: ProxyBridge<B0::Time, Bk::R1>,
+    /// Per-key replay histories, held across the unit and reloaded per key (only the >=16/>=16
+    /// replay-wave path touches them), so a high-fanout join pays no per-key `IdHistory` alloc.
+    h0: IdHistory<B0::Time, Bk::R0>,
+    h1: IdHistory<B0::Time, Bk::R1>,
+    /// The block's matched records, as the four parallel arrays `cross` reads. Held across blocks
+    /// only to keep their allocations.
+    li: Vec<(u64, u64)>,
+    ri: Vec<(u64, u64)>,
+    ot: Vec<B0::Time>,
+    od: Vec<Bk::ROut>,
+    /// The last block's containers, in reverse, served from the back one `next` at a time.
+    ready: Vec<Bk::Output>,
+}
 
-/// Prepare one join unit: present both sides, merge-match key runs over the sorted hashes, and
-/// cross-product matched records into a sequence of output containers — one per ~[`JOIN_CHUNK`]
-/// matches, flushed wherever the batch fills (including mid-key; parity: their concatenation is the
-/// single-container output). `meet` is a lower bound on the fresh side's times, so the accumulated
-/// side loads compacted (see [`JoinInstance`]); every output is produced under the capability at
-/// `meet`, so advancing loaded times by it leaves the output unchanged.
-fn join_prep<B0, B1, Bk>(backend: &mut Bk, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: B0::Time) -> Vec<Bk::Output>
+impl<B0, B1, Bk> Iterator for ProxyJoinIter<B0, B1, Bk>
 where
     B0: BatchReader,
     B1: BatchReader<Time = B0::Time>,
     Bk: ProxyJoinBackend<B0, B1>,
 {
-    let lower = Antichain::from_elem(meet);
-    let instance = JoinInstance { batches0: &input0, batches1: &input1, lower: lower.borrow() };
-
-    // Prepare work using the keys of the just-received side.
-    // FIXME: Use the smaller of the two instead.
-    let (p0, p1) = match fresh {
-        Fresh::Input0 => {
-            let p0 = backend.present0(&instance, None);
-            if p0.is_empty() { return Vec::new(); }
-            let mut keys: Vec<u64> = p0.iter().map(|r| r.0.0).collect();
-            keys.dedup();
-            let p1 = backend.present1(&instance, Some(&keys));
-            (p0, p1)
-        }
-        Fresh::Input1 => {
-            let p1 = backend.present1(&instance, None);
-            if p1.is_empty() { return Vec::new(); }
-            let mut keys: Vec<u64> = p1.iter().map(|r| r.0.0).collect();
-            keys.dedup();
-            let p0 = backend.present0(&instance, Some(&keys));
-            (p0, p1)
+    type Item = Bk::Output;
+
+    /// Serve a ready container, else draw and cross blocks until one yields any.
+    ///
+    /// A block that yields no matches is not an empty container; it is a region of the key space
+    /// with no intersection, and the unit moves on to the next block.
+    fn next(&mut self) -> Option<Bk::Output> {
+        while self.ready.is_empty() && self.from.is_some() {
+            self.refill();
+            self.work();
+            if !self.li.is_empty() { self.cross(); }
         }
-    };
-    if p0.is_empty() || p1.is_empty() { return Vec::new(); }
-    super::debug_assert_sorted_bridge(&p0, "present0");
-    super::debug_assert_sorted_bridge(&p1, "present1");
-
-    // Merge-join the two presented runs on `key_hash` (both sorted) and cross matched pairs. `flush`
-    // ships the accumulated batch as a container and resets the buffers; `join_key` calls it at each
-    // `JOIN_CHUNK` boundary — including *within* a high-fanout key's wave — so no single key
-    // materializes more than a chunk of its cross product. The closure scope releases its
-    // `backend`/`out` borrow before `out` is returned.
-    let mut out: Vec<Bk::Output> = Vec::new();
-    {
-        let mut flush = |li: &mut Vec<(u64, u64)>, ri: &mut Vec<(u64, u64)>, ot: &mut Vec<B0::Time>, od: &mut Vec<Bk::ROut>| {
-            out.push(backend.cross(&instance, li.as_slice(), ri.as_slice(), std::mem::take(ot), std::mem::take(od)));
-            li.clear();
-            ri.clear();
-        };
-
-        let (mut li, mut ri) = (Vec::new(), Vec::new());
-        let (mut ot, mut od) = (Vec::new(), Vec::new());
-        // Per-key replay histories, held across the unit and reloaded per key (only the >=16/>=16
-        // replay-wave path touches them), so a high-fanout join pays no per-key `IdHistory` alloc.
-        let mut h0 = IdHistory::new();
-        let mut h1 = IdHistory::new();
+        self.ready.pop()
+    }
+}
+
+impl<B0, B1, Bk> ProxyJoinIter<B0, B1, Bk>
+where
+    B0: BatchReader,
+    B1: BatchReader<Time = B0::Time>,
+    Bk: ProxyJoinBackend<B0, B1>,
+{
+    /// Draw the next block from the backend.
+    fn refill(&mut self) {
+        self.p0.clear();
+        self.p1.clear();
+        let before = self.from;
+        self.backend.borrow_mut().advance(&self.instance, &mut self.from, &mut self.p0, &mut self.p1);
+        // Without progress the unit would never retire, so this guards liveness as well as contract.
+        debug_assert!(
+            self.from.is_none() || self.from > before,
+            "advance must either strictly increase `from` or report the unit drawn",
+        );
+        super::debug_assert_sorted_bridge(&self.p0, "advance (bridge0)");
+        super::debug_assert_sorted_bridge(&self.p1, "advance (bridge1)");
+        // A key hash outside `[before, from)` is either one an earlier block already retired, or one
+        // a later block may yet report: both split a key across blocks, which silently drops the
+        // matches that would have crossed the split.
+        debug_assert!(
+            {
+                let mut keys = self.p0.iter().map(|r| r.0.0).chain(self.p1.iter().map(|r| r.0.0));
+                keys.all(|k| before.is_none_or(|b| b <= k) && self.from.is_none_or(|f| k < f))
+            },
+            "advance must report a key hash entirely within the block that first mentions it",
+        );
+    }
+
+    /// Merge-match the whole of the current block into the match buffers.
+    ///
+    /// Both runs are sorted by `key_hash`, so matching keys are found by a merge.
+    fn work(&mut self) {
+        // Disjoint field borrows, as `join_key` holds the bridges and the buffers at once.
+        let (p0, p1) = (&self.p0, &self.p1);
+        let (li, ri, ot, od) = (&mut self.li, &mut self.ri, &mut self.ot, &mut self.od);
+        let (h0, h1) = (&mut self.h0, &mut self.h1);
+
         let (mut i, mut j) = (0usize, 0usize);
         while i < p0.len() && j < p1.len() {
             let (ki, kj) = (p0[i].0.0, p1[j].0.0);
@@ -148,16 +235,30 @@ where
                 while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
                 let mut e1 = j;
                 while e1 < p1.len() && p1[e1].0.0 == ki { e1 += 1; }
-                join_key(ki, &p0, i..e0, &p1, j..e1, &mut h0, &mut h1, &mut li, &mut ri, &mut ot, &mut od, &mut flush);
+                join_key(ki, p0, i..e0, p1, j..e1, h0, h1, li, ri, ot, od);
                 i = e0;
                 j = e1;
             }
         }
-        if !li.is_empty() {
-            flush(&mut li, &mut ri, &mut ot, &mut od);
-        }
     }
-    out
+
+    /// Turn the block's matches into containers, ready to be served one at a time.
+    fn cross(&mut self) {
+        self.backend.borrow_mut().cross(
+            &self.instance,
+            self.li.as_slice(),
+            self.ri.as_slice(),
+            &mut self.ot,
+            &mut self.od,
+            &mut self.ready,
+        );
+        // `next` serves from the back, so reverse to ship in the order the backend produced.
+        self.ready.reverse();
+        self.li.clear();
+        self.ri.clear();
+        self.ot.clear();
+        self.od.clear();
+    }
 }
 
 /// Match one key's records across the two presented runs.
@@ -166,7 +267,7 @@ where
 /// If both histories are large, this replays the histories compacting as it goes in
 /// order to (potentially) avoid quadratic blow-up.
 #[allow(clippy::too_many_arguments)]
-fn join_key<T, R0, R1, RO, F>(
+fn join_key<T, R0, R1, RO>(
     kh: u64,
     p0: &ProxyBridge<T, R0>,
     r0: std::ops::Range<usize>,
@@ -178,16 +279,11 @@ fn join_key<T, R0, R1, RO, F>(
     ri: &mut Vec<(u64, u64)>,
     ot: &mut Vec<T>,
     od: &mut Vec<RO>,
-    flush: &mut F,
 ) where
     T: Lattice + Timestamp,
     R0: Semigroup + Multiply<R1, Output = RO> + Clone,
     R1: Semigroup + Clone,
-    F: FnMut(&mut Vec<(u64, u64)>, &mut Vec<(u64, u64)>, &mut Vec<T>, &mut Vec<RO>),
 {
-    // `flush` ships the accumulated batch once it reaches `JOIN_CHUNK` and resets the buffers. It's
-    // checked after every produced pair — *including inside the replay wave below* — so even a key
-    // whose fanout dwarfs `JOIN_CHUNK` never materializes more than a chunk of its cross product.
     if r0.len() < 16 || r1.len() < 16 {
         for a in r0 {
             for b in r1.clone() {
@@ -195,7 +291,6 @@ fn join_key<T, R0, R1, RO, F>(
                 ri.push((kh, p1[b].0.1));
                 ot.push(p0[a].1.join(&p1[b].1));
                 od.push(p0[a].2.clone().multiply(&p1[b].2));
-                if li.len() >= JOIN_CHUNK { flush(li, ri, ot, od); }
             }
         }
         return;
@@ -211,8 +306,5 @@ fn join_key<T, R0, R1, RO, F>(
         ri.push((kh, v1));
         ot.push(t);
         od.push(d);
-        if li.len() >= JOIN_CHUNK {
-            flush(li, ri, ot, od);
-        }
     });
 }

From fc62d8b6d9bf96ce89de3b65b1f4b690875d5ef3 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 29 Jul 2026 14:26:00 -0400
Subject: [PATCH 2/3] int_proxy join: simplify the match presentation and
 tighten the key contract

`cross` takes a single `JoinMatches`, whose `ids` carry `(key, (val0, val1))` rather
than repeating the key hash in two parallel arrays of pairs.
`join_key` fills it directly, in place of four loose vectors threaded through every
call.

`JoinInstance::lower` is a time rather than a one-element antichain.
It was only ever built from the capability's time, and a backend consolidating loaded
times wants `join_assign`, not `advance_by` against a singleton frontier.

`work` walks the two bridges in lockstep rather than merging them.
`advance` reports the intersecting keys, entirely and from both inputs, so the runs
carry the same keys in the same order; the previous merge silently skipped a
mismatch, which is a protocol violation that produces a wrong answer rather than a
failure.
Debug assertions now report it, and the loop keeps both bounds so a violating backend
truncates instead of indexing out of bounds.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../src/operators/int_proxy/join.rs           | 183 ++++++++----------
 1 file changed, 79 insertions(+), 104 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index 05d6b1a9d..a791fbc1b 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -6,7 +6,7 @@
 use std::cell::RefCell;
 use std::rc::Rc;
 
-use timely::progress::{Antichain, Timestamp};
+use timely::progress::Timestamp;
 
 use crate::difference::{Multiply, Semigroup};
 use crate::lattice::Lattice;
@@ -16,46 +16,24 @@ use crate::operators::join::{Fresh, JoinTactic};
 
 use super::history::IdHistory;
 
-/// A unit of proxied join work, presented to the backend.
-pub struct JoinInstance<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
-    /// The first input's batches.
-    pub batches0: Vec<B0>,
-    /// The second input's batches.
-    pub batches1: Vec<B1>,
-    /// The compaction frontier for loading (the unit's capability time).
-    pub lower: Antichain<B0::Time>,
-}
-
-/// A type that can interpret and retire pairs of batches, joined by key hashes.
-///
-/// The protocol repeatedly invokes [`advance`](Self::advance) to draw a block of the proxy
-/// collection, then [`cross`](Self::cross) to turn that block's matches into output containers.
-/// The unit is retired when `advance` reports the key space exhausted.
+/// A type that can interpret and retire pairs of lists of batches, joined by key hashes.
 ///
-/// The two granularities are the backend's, and independent: `advance` sizes the input it draws,
-/// `cross` sizes the containers it produces from the matches that block yields.
-///
-/// The backend holds no state across the calls of one unit: its progress lives in the `from`
-/// argument, which the harness owns. This is load-bearing, as the harness may have many units
-/// in flight against the same backend, and interleaves their calls freely.
+/// The harness repeatedly invokes [`advance`](Self::advance) to draw a block of the proxy collection,
+/// then [`cross`](Self::cross) to turn that block's matches into output containers, until `advance` reports the key space exhausted.
 pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
-    /// Diff type presented for the first input.
+    /// Diff type of the first input.
     type R0: Semigroup + Multiply<Self::R1, Output = Self::ROut>;
-    /// Diff type presented for the second input.
+    /// Diff type of the second input.
     type R1: Semigroup;
-    /// Diff type of matched records (`R0 * R1`), computed by the tactic.
+    /// Diff type of matched records (`R0 * R1`), computed by the harness.
     type ROut: Semigroup;
     /// The output container built from matched value ids.
     type Output;
 
-    /// Populates two bridges with intersecting keys from `from` onward, updating `from` to the new limit.
-    ///
-    /// The `from` argument is an inclusive lower bound on key identifier, initially `Some(0)`, with `None`
-    /// used to indicate that the keyspace has been exhausted.
+    /// Populates two bridges with all updates of a range of matched keys.
     ///
-    /// A key hash must be reported entirely by the call that first mentions it: all of its updates,
-    /// from both inputs, and none in any other block. Equivalently, each reported key hash lies in
-    /// `[from, from')` for the argument's values before and after the call.
+    /// The key range should start with `from`, and be updated to reflect the new inclusive lower bound,
+    /// with `None` used to indicate completion.
     fn advance(
         &mut self,
         instance: &JoinInstance<B0, B1>,
@@ -64,17 +42,41 @@ pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
         bridge1: &mut ProxyBridge<B0::Time, Self::R1>,
     );
 
-    /// From a list of left and right identifiers, and corresponding times and diffs, the output.
-    ///
-    /// The four are parallel arrays with one entry per matched pair: `left[i]` and `right[i]` are the
-    /// `(key_hash, value_id)` of the pair's two records, which share a key hash, and `times[i]` and
-    /// `diffs[i]` are the join of their times and the product of their diffs. They are not
-    /// consolidated, and a pair can repeat at distinct times. Drain `times` and `diffs` to take their
-    /// contents; the harness keeps their allocations.
+    /// From a list of matching identifiers, interpret and populate `output` with resulting containers.
+    fn cross(
+        &mut self,
+        instance: &JoinInstance<B0, B1>,
+        matches: &mut JoinMatches<B0::Time, Self::ROut>,
+        output: &mut Vec<Self::Output>,
+    );
+}
+
+/// A unit of proxied join work, presented to the backend.
+pub struct JoinInstance<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
+    /// The first input's batches.
+    pub batches0: Vec<B0>,
+    /// The second input's batches.
+    pub batches1: Vec<B1>,
+    /// A lower bound on the meet of pairs of update times.
     ///
-    /// The containers are pushed to `output`, which arrives empty, and are shipped one at a time:
-    /// their sizing is the backend's, independent of the block `advance` drew the matches from.
-    fn cross(&mut self, instance: &JoinInstance<B0, B1>, left: &[(u64, u64)], right: &[(u64, u64)], times: &mut Vec<B0::Time>, diffs: &mut Vec<Self::ROut>, output: &mut Vec<Self::Output>);
+    /// This can be applied when loading updates to consolidate on load.
+    pub lower: B0::Time,
+}
+
+/// Presentation of discovered join matches.
+///
+/// The arrays have common lengths, and are in key order but may not be consolidated.
+pub struct JoinMatches<T, R> {
+    /// Triples of `(key, (val0, val1))` of matches.
+    pub ids: Vec<(u64, (u64, u64))>,
+    /// Times of the updates.
+    pub times: Vec<T>,
+    /// Diffs of the updates.
+    pub diffs: Vec<R>,
+}
+
+impl<T, R> Default for JoinMatches<T, R> {
+    fn default() -> Self { Self { ids: vec![], times: vec![], diffs: vec![] } }
 }
 
 /// A proxy-space [`JoinTactic`]: matches records of the two drawn runs by `key_hash`.
@@ -104,16 +106,13 @@ where
     fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, _fresh: Fresh, meet: B0::Time) -> Box<dyn Iterator<Item = Bk::Output>> {
         Box::new(ProxyJoinIter {
             backend: Rc::clone(&self.backend),
-            instance: JoinInstance { batches0: input0, batches1: input1, lower: Antichain::from_elem(meet) },
+            instance: JoinInstance { batches0: input0, batches1: input1, lower: meet },
             from: Some(0),
             p0: Vec::new(),
             p1: Vec::new(),
             h0: IdHistory::new(),
             h1: IdHistory::new(),
-            li: Vec::new(),
-            ri: Vec::new(),
-            ot: Vec::new(),
-            od: Vec::new(),
+            matches: JoinMatches::default(),
             ready: Vec::new(),
         })
     }
@@ -139,7 +138,7 @@ where
 {
     /// The backend, shared across all outstanding units.
     backend: Rc<RefCell<Bk>>,
-    /// The unit's inputs, and the compaction frontier at which they load.
+    /// The unit's inputs, and the time at which they can consolidate as they load.
     instance: JoinInstance<B0, B1>,
     /// Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
     /// once the backend reports the unit drawn.
@@ -151,12 +150,8 @@ where
     /// replay-wave path touches them), so a high-fanout join pays no per-key `IdHistory` alloc.
     h0: IdHistory<B0::Time, Bk::R0>,
     h1: IdHistory<B0::Time, Bk::R1>,
-    /// The block's matched records, as the four parallel arrays `cross` reads. Held across blocks
-    /// only to keep their allocations.
-    li: Vec<(u64, u64)>,
-    ri: Vec<(u64, u64)>,
-    ot: Vec<B0::Time>,
-    od: Vec<Bk::ROut>,
+    /// The block's matched records, held across blocks to keep their allocations.
+    matches: JoinMatches<B0::Time, Bk::ROut>,
     /// The last block's containers, in reverse, served from the back one `next` at a time.
     ready: Vec<Bk::Output>,
 }
@@ -177,7 +172,7 @@ where
         while self.ready.is_empty() && self.from.is_some() {
             self.refill();
             self.work();
-            if !self.li.is_empty() { self.cross(); }
+            if !self.matches.ids.is_empty() { self.cross(); }
         }
         self.ready.pop()
     }
@@ -214,59 +209,47 @@ where
         );
     }
 
-    /// Merge-match the whole of the current block into the match buffers.
-    ///
-    /// Both runs are sorted by `key_hash`, so matching keys are found by a merge.
+    /// Match the whole of the current block into the match buffers.
     fn work(&mut self) {
         // Disjoint field borrows, as `join_key` holds the bridges and the buffers at once.
         let (p0, p1) = (&self.p0, &self.p1);
-        let (li, ri, ot, od) = (&mut self.li, &mut self.ri, &mut self.ot, &mut self.od);
         let (h0, h1) = (&mut self.h0, &mut self.h1);
 
         let (mut i, mut j) = (0usize, 0usize);
         while i < p0.len() && j < p1.len() {
-            let (ki, kj) = (p0[i].0.0, p1[j].0.0);
-            if ki < kj {
-                i += 1;
-            } else if kj < ki {
-                j += 1;
-            } else {
-                let mut e0 = i;
-                while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
-                let mut e1 = j;
-                while e1 < p1.len() && p1[e1].0.0 == ki { e1 += 1; }
-                join_key(ki, p0, i..e0, p1, j..e1, h0, h1, li, ri, ot, od);
-                i = e0;
-                j = e1;
-            }
+            let ki = p0[i].0.0;
+            debug_assert_eq!(ki, p1[j].0.0, "advance must report common keys");
+            let mut e0 = i;
+            while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
+            let mut e1 = j;
+            while e1 < p1.len() && p1[e1].0.0 == ki { e1 += 1; }
+            join_key(ki, p0, i..e0, p1, j..e1, h0, h1, &mut self.matches);
+            i = e0;
+            j = e1;
         }
+        debug_assert!(i == p0.len() && j == p1.len(), "both bridges must drain together");
     }
 
     /// Turn the block's matches into containers, ready to be served one at a time.
     fn cross(&mut self) {
         self.backend.borrow_mut().cross(
             &self.instance,
-            self.li.as_slice(),
-            self.ri.as_slice(),
-            &mut self.ot,
-            &mut self.od,
+            &mut self.matches,
             &mut self.ready,
         );
         // `next` serves from the back, so reverse to ship in the order the backend produced.
         self.ready.reverse();
-        self.li.clear();
-        self.ri.clear();
-        self.ot.clear();
-        self.od.clear();
+        self.matches.ids.clear();
+        self.matches.times.clear();
+        self.matches.diffs.clear();
     }
 }
 
 /// Match one key's records across the two presented runs.
 ///
-/// If either history is small, this performs a simple cross product.
+/// If either history is small, this performs a direct cross product.
 /// If both histories are large, this replays the histories compacting as it goes in
 /// order to (potentially) avoid quadratic blow-up.
-#[allow(clippy::too_many_arguments)]
 fn join_key<T, R0, R1, RO>(
     kh: u64,
     p0: &ProxyBridge<T, R0>,
@@ -275,10 +258,7 @@ fn join_key<T, R0, R1, RO>(
     r1: std::ops::Range<usize>,
     h0: &mut IdHistory<T, R0>,
     h1: &mut IdHistory<T, R1>,
-    li: &mut Vec<(u64, u64)>,
-    ri: &mut Vec<(u64, u64)>,
-    ot: &mut Vec<T>,
-    od: &mut Vec<RO>,
+    matches: &mut JoinMatches<T, RO>,
 ) where
     T: Lattice + Timestamp,
     R0: Semigroup + Multiply<R1, Output = RO> + Clone,
@@ -287,24 +267,19 @@ fn join_key<T, R0, R1, RO>(
     if r0.len() < 16 || r1.len() < 16 {
         for a in r0 {
             for b in r1.clone() {
-                li.push((kh, p0[a].0.1));
-                ri.push((kh, p1[b].0.1));
-                ot.push(p0[a].1.join(&p1[b].1));
-                od.push(p0[a].2.clone().multiply(&p1[b].2));
+                matches.ids.push((kh, (p0[a].0.1, p1[b].0.1)));
+                matches.times.push(p0[a].1.join(&p1[b].1));
+                matches.diffs.push(p0[a].2.clone().multiply(&p1[b].2));
             }
         }
-        return;
     }
-
-    // Reusable replay scratch, reloaded per key (`load_iter` clears + rebuilds, keeping capacity);
-    // the caller holds `h0`/`h1` across the unit so a high-fanout join allocates no per-key history.
-    h0.load_iter(r0.map(|i| (p0[i].0.1, p0[i].1.clone(), p0[i].2.clone())), None);
-    h1.load_iter(r1.map(|i| (p1[i].0.1, p1[i].1.clone(), p1[i].2.clone())), None);
-
-    crate::operators::common::bilinear_wave(h0, h1, |v0, v1, t, d| {
-        li.push((kh, v0));
-        ri.push((kh, v1));
-        ot.push(t);
-        od.push(d);
-    });
+    else {
+        h0.load_iter(r0.map(|i| (p0[i].0.1, p0[i].1.clone(), p0[i].2.clone())), None);
+        h1.load_iter(r1.map(|i| (p1[i].0.1, p1[i].1.clone(), p1[i].2.clone())), None);
+        crate::operators::common::bilinear_wave(h0, h1, |v0, v1, t, d| {
+            matches.ids.push((kh, (v0, v1)));
+            matches.times.push(t);
+            matches.diffs.push(d);
+        });
+    }
 }

From 1dd6ad9c3f56cec95979b8fce77f94bb45e46d9a Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 29 Jul 2026 17:13:27 -0400
Subject: [PATCH 3/3] improve documentation

---
 .../src/operators/int_proxy/join.rs           | 45 +++++++------------
 1 file changed, 16 insertions(+), 29 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index a791fbc1b..7a024d596 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -30,10 +30,12 @@ pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
     /// The output container built from matched value ids.
     type Output;
 
-    /// Populates two bridges with all updates of a range of matched keys.
+    /// Populates the two bridges with all updates for all keys that match in a returned range.
     ///
-    /// The key range should start with `from`, and be updated to reflect the new inclusive lower bound,
-    /// with `None` used to indicate completion.
+    /// The `from` indicates an inclusive lower bound on key hash, and should be updated by the implementor to an exclusive
+    /// upper bound for the range of keys it intends to return in this call. The `None` value indicates the keys are exhausted.
+    /// The returned bridges must contain all updates from both `instance` inputs for keys that are present in both inputs, and
+    /// which are greater or equal to the initial `from`, and not greater or equal to its value when returned.
     fn advance(
         &mut self,
         instance: &JoinInstance<B0, B1>,
@@ -42,7 +44,7 @@ pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
         bridge1: &mut ProxyBridge<B0::Time, Self::R1>,
     );
 
-    /// From a list of matching identifiers, interpret and populate `output` with resulting containers.
+    /// Interpret a list of matching identifiers, translate them to outputs, and place them in `output`.
     fn cross(
         &mut self,
         instance: &JoinInstance<B0, B1>,
@@ -51,7 +53,7 @@ pub trait ProxyJoinBackend<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
     );
 }
 
-/// A unit of proxied join work, presented to the backend.
+/// A unit of proxied join work, for presentation to the backend.
 pub struct JoinInstance<B0: BatchReader, B1: BatchReader<Time = B0::Time>> {
     /// The first input's batches.
     pub batches0: Vec<B0>,
@@ -80,10 +82,6 @@ impl<T, R> Default for JoinMatches<T, R> {
 }
 
 /// A proxy-space [`JoinTactic`]: matches records of the two drawn runs by `key_hash`.
-///
-/// The backend is shared across all outstanding units (an `Rc<RefCell<_>>`), as each prepared unit
-/// is a self-contained `'static` iterator and cannot borrow the tactic. Units are safe to interleave
-/// because a unit's progress through the key space rides in its own `from`, not in the backend.
 pub struct ProxyJoinTactic<B0, B1, Bk> {
     backend: Rc<RefCell<Bk>>,
     _marker: std::marker::PhantomData<(B0, B1)>,
@@ -120,34 +118,26 @@ where
 
 /// Deferred proxy join computation, as an iterator of output containers.
 ///
-/// The unit draws the proxy collection a block at a time (`Bk::advance`), merge-matches the two
-/// drawn runs by `key_hash`, and hands the matches back to be cut into containers (`Bk::cross`).
-/// Both granularities are the backend's: `advance` sizes the block, `cross` the containers. One
-/// key's matches are held whole however the block is sized, as the replay wave cannot be
-/// interrupted. The driver stops pulling once its budget is spent and resumes the same iterator on
-/// the next activation.
-///
-/// `meet` is a lower bound on the fresh side's times, and `lower` presents it to the backend
-/// (see [`JoinInstance`]). Loaded times can be advanced by it on either input: the fresh side's
-/// times dominate `meet`, so the joined time does too.
+/// The iterator draws the proxy collection from the back-end a block at a time (`Bk::advance`).
+/// Each block is then translated to output updates with joined times and multiplied differences,
+/// which are provided to the back-end to translate into output containers, which are then returned.
 struct ProxyJoinIter<B0, B1, Bk>
 where
     B0: BatchReader,
     B1: BatchReader<Time = B0::Time>,
     Bk: ProxyJoinBackend<B0, B1>,
 {
-    /// The backend, shared across all outstanding units.
+    /// The backend, shared across all outstanding iterators.
     backend: Rc<RefCell<Bk>>,
-    /// The unit's inputs, and the time at which they can consolidate as they load.
+    /// The iterator's inputs, and the time at which they can consolidate as they load.
     instance: JoinInstance<B0, B1>,
     /// Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
-    /// once the backend reports the unit drawn.
+    /// once the backend reports the iteration is complete.
     from: Option<u64>,
     /// The current block: the two runs `advance` last drew, which one `next` consumes entirely.
     p0: ProxyBridge<B0::Time, Bk::R0>,
     p1: ProxyBridge<B0::Time, Bk::R1>,
-    /// Per-key replay histories, held across the unit and reloaded per key (only the >=16/>=16
-    /// replay-wave path touches them), so a high-fanout join pays no per-key `IdHistory` alloc.
+    /// Per-key replay histories, held across the iterator and reloaded per key when needed.
     h0: IdHistory<B0::Time, Bk::R0>,
     h1: IdHistory<B0::Time, Bk::R1>,
     /// The block's matched records, held across blocks to keep their allocations.
@@ -165,9 +155,6 @@ where
     type Item = Bk::Output;
 
     /// Serve a ready container, else draw and cross blocks until one yields any.
-    ///
-    /// A block that yields no matches is not an empty container; it is a region of the key space
-    /// with no intersection, and the unit moves on to the next block.
     fn next(&mut self) -> Option<Bk::Output> {
         while self.ready.is_empty() && self.from.is_some() {
             self.refill();
@@ -190,10 +177,10 @@ where
         self.p1.clear();
         let before = self.from;
         self.backend.borrow_mut().advance(&self.instance, &mut self.from, &mut self.p0, &mut self.p1);
-        // Without progress the unit would never retire, so this guards liveness as well as contract.
+        // Without progress the iterator would never retire, so this guards liveness as well as contract.
         debug_assert!(
             self.from.is_none() || self.from > before,
-            "advance must either strictly increase `from` or report the unit drawn",
+            "advance must either strictly increase `from` or report the iteration complete",
         );
         super::debug_assert_sorted_bridge(&self.p0, "advance (bridge0)");
         super::debug_assert_sorted_bridge(&self.p1, "advance (bridge1)");