From 098cdda9bf05fedac6f4a820319f86e14817fa42 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Sat, 19 Sep 2026 01:42:55 +0200
Subject: [PATCH 1/2] Suspend the cursor join within a key

The cursor join tactic played a key's whole cross product into its container
builder before checking whether a container had completed, so a key large
enough to outrun the driver's fuel buffered its entire output before the
iterator returned anything. A single-key join of 10,000 by 10,000 values
peaked at 3.2 GB of resident memory for its 100,000,000 matches. The driver's
budget never got a chance to act, because `next` did not return until the key
was done.

Keys now take one of two paths, chosen once the key's edits are loaded. A key
whose cross product fits `KEY_WORK_LIMIT`, half the driver's fuel, runs to
completion in one call over histories of borrowed values, as before. A larger
key is reloaded into histories of value indices, which the iterator owns and
which therefore survive a suspension, and is then replayed a container at a
time. The indices are what make the state storable: a history of
`Cursor::Val<'a>` cannot outlive the storage the iterator owns, so it cannot
be held across calls.

The same key now peaks at 97 MB. The benchmark added here shows the shapes
that still take the uninterrupted path are unaffected: one value per key and
a handful of values per key both land within noise of the previous code.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../src/operators/cursor/history.rs           |  39 ++
 .../src/operators/cursor/join.rs              | 409 +++++++++++++++++-
 .../src/operators/history.rs                  |  19 +-
 differential-dataflow/src/operators/join.rs   |  25 +-
 differential-dataflow/tests/join_bench.rs     |  84 ++++
 5 files changed, 555 insertions(+), 21 deletions(-)
 create mode 100644 differential-dataflow/tests/join_bench.rs

diff --git a/differential-dataflow/src/operators/cursor/history.rs b/differential-dataflow/src/operators/cursor/history.rs
index f4592b7de..b489858b5 100644
--- a/differential-dataflow/src/operators/cursor/history.rs
+++ b/differential-dataflow/src/operators/cursor/history.rs
@@ -3,6 +3,7 @@
 use crate::lattice::Lattice;
 use crate::operators::history::{EditList, HistoryReplay, ValueHistory};
 use crate::trace::Cursor;
+use crate::trace::implementations::containers::BatchContainer;
 
 /// Walks the cursor's values at the current key into `target`, advancing times by `meet` if supplied.
 fn load_values<'a, V, T, D, C>(
@@ -47,6 +48,44 @@ where
     load_values(history.edits_mut(), cursor, storage, meet);
 }
 
+/// Loads the cursor's values at its current key into `values`, and their edits into `history`
+/// keyed by each value's index in `values`.
+///
+/// The indices are what let the history outlive the borrow of `storage`: a history of
+/// `Cursor::Val<'a>` can only live as long as the storage it reads from, and so cannot be held
+/// across calls by an iterator that owns that storage. A history of indices can, at the cost of
+/// copying the key's values into a container the iterator owns too. Callers that replay a key in
+/// one uninterrupted pass should prefer `load_current`, which copies nothing.
+///
+/// The cursor is left with its values exhausted, as `load_current` leaves it.
+pub(super) fn load_current_indexed<T, D, C>(
+    values: &mut C::ValContainer,
+    history: &mut ValueHistory<usize, T, D>,
+    cursor: &mut C,
+    storage: &C::Storage,
+    meet: Option<&T>,
+)
+where
+    T: Ord + Clone + Lattice,
+    D: crate::difference::Semigroup,
+    C: Cursor<Time = T, Diff = D>,
+{
+    history.clear();
+    values.clear();
+    let edits = history.edits_mut();
+    while let Some(val) = cursor.get_val(storage) {
+        let index = values.len();
+        values.push_ref(val);
+        cursor.map_times(storage, |time, diff| {
+            let mut time = C::owned_time(time);
+            if let Some(meet) = meet { time.join_assign(meet); }
+            edits.push(time, C::owned_diff(diff));
+        });
+        edits.seal(index);
+        cursor.step_val(storage);
+    }
+}
+
 /// Loads and replays a specified key.
 ///
 /// If the key is absent, the replayed history will be empty.
diff --git a/differential-dataflow/src/operators/cursor/join.rs b/differential-dataflow/src/operators/cursor/join.rs
index cf4e41326..890c01ace 100644
--- a/differential-dataflow/src/operators/cursor/join.rs
+++ b/differential-dataflow/src/operators/cursor/join.rs
@@ -9,11 +9,11 @@ use timely::ContainerBuilder;
 use timely::dataflow::Stream;
 use timely::progress::Timestamp;
 
-use super::history::load_current;
+use super::history::{load_current, load_current_indexed};
 use crate::lattice::Lattice;
 use crate::operators::arrange::Arranged;
 use crate::operators::history::ValueHistory;
-use crate::operators::join::{Fresh, JoinTactic, join_with_tactic};
+use crate::operators::join::{Fresh, JoinTactic, KEY_WORK_LIMIT, join_with_tactic};
 use crate::trace::{BatchCursor, BatchDiff, BatchVal, Cursor, Navigable, TraceReader};
 use crate::trace::cursor::cursor_list;
 use crate::trace::implementations::containers::BatchContainer;
@@ -117,6 +117,11 @@ where
             logic: Rc::clone(&self.logic),
             builder: CB::default(),
             ready: VecDeque::new(),
+            vals1: BatchContainer::with_capacity(0),
+            vals2: BatchContainer::with_capacity(0),
+            history1: ValueHistory::new(),
+            history2: ValueHistory::new(),
+            resume: None,
             done: false,
         })
     }
@@ -130,6 +135,14 @@ where
 /// 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.
+///
+/// A container boundary can fall in the middle of a key, and a single key's cross product can be
+/// arbitrarily larger than the driver's whole budget, so suspending only between keys would let one
+/// key buffer unboundedly. Keys are therefore handled by one of two strategies, chosen once the key's
+/// edits are loaded and its work is known: a key whose cross product fits [`KEY_WORK_LIMIT`] runs to
+/// completion in a single call, over histories of borrowed values; a larger key is reloaded into the
+/// iterator's own `vals`/`history` fields, which are indexed rather than borrowed and so survive a
+/// suspension, and is then replayed by [`replay_suspending`] a container at a time.
 struct DeferredIter<T, C1, C2, L, CB>
 where
     T: Timestamp+Lattice,
@@ -152,9 +165,36 @@ where
     builder: CB,
     /// Completed containers awaiting a `next` call.
     ready: VecDeque<CB::Container>,
+    /// The suspending strategy's copy of the open key's values, which its histories index.
+    vals1: C1::ValContainer,
+    vals2: C2::ValContainer,
+    /// The suspending strategy's histories for the open key, over indices into `vals1`/`vals2`.
+    history1: ValueHistory<usize, T, C1::Diff>,
+    history2: ValueHistory<usize, T, C2::Diff>,
+    /// Where in the open key's replay this iterator suspended; `None` when no key is open.
+    resume: Option<Resume>,
     done: bool,
 }
 
+/// Progress through one key, for the strategy that suspends within a key.
+///
+/// The variants mirror the two strategies of [`JoinThinker::think`]: the nested cross product it
+/// uses for small histories, and the time-ordered wave it uses for large ones. A wave step advances
+/// one history's buffer before emitting against it, so the step it is part-way through is part of
+/// the state: resuming must not advance that buffer a second time, which would consolidate it and
+/// invalidate the offset resumed from.
+#[derive(Clone, Copy)]
+enum Resume {
+    /// The nested cross product, at edit indices `(index1, index2)` of the two histories.
+    Simple(usize, usize),
+    /// The wave, between steps; the next step follows from the histories' least times.
+    WaveChoose,
+    /// The wave, emitting `history2`'s buffer against `history1`'s next edit, from this offset.
+    WaveStep1(usize),
+    /// The wave, emitting `history1`'s buffer against `history2`'s next edit, from this offset.
+    WaveStep2(usize),
+}
+
 impl<T, C1, C2, L, CB> Iterator for DeferredIter<T, C1, C2, L, CB>
 where
     T: Timestamp+Lattice,
@@ -182,6 +222,11 @@ where
         let storage2 = &self.storage2;
         let cursor1 = &mut self.cursor1;
         let cursor2 = &mut self.cursor2;
+        let vals1 = &mut self.vals1;
+        let vals2 = &mut self.vals2;
+        let history1 = &mut self.history1;
+        let history2 = &mut self.history2;
+        let resume = &mut self.resume;
         let builder = &mut self.builder;
         let ready = &mut self.ready;
         let mut logic = self.logic.borrow_mut();
@@ -197,23 +242,76 @@ where
                     Ordering::Greater => cursor2.seek_key(storage2, key1),
                     Ordering::Equal => {
 
-                        load_current(&mut thinker.history1, cursor1, storage1, meet1);
-                        load_current(&mut thinker.history2, cursor2, storage2, meet2);
+                        // A key is opened once and then either retired in this iteration or suspended
+                        // part-way; the cursors are not stepped until it retires, so a suspended key is
+                        // still the key both cursors report on re-entry.
+                        if resume.is_none() {
+                            load_current(&mut thinker.history1, cursor1, storage1, meet1);
+                            load_current(&mut thinker.history2, cursor2, storage2, meet2);
 
-                        thinker.think(|v1,v2,t,r1,r2| {
-                            logic(key1, v1, v2, t, r1, r2, builder);
-                        });
+                            let len1 = thinker.history1.edit_len();
+                            let len2 = thinker.history2.edit_len();
+                            if len1.saturating_mul(len2) <= KEY_WORK_LIMIT {
+                                thinker.think(|v1,v2,t,r1,r2| {
+                                    logic(key1, v1, v2, t, r1, r2, builder);
+                                });
 
-                        cursor1.step_key(storage1);
-                        cursor2.step_key(storage2);
+                                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() {
-                            // Avoiding the mem::take would require a non-iterator trait.
-                            ready.push_back(std::mem::take(container));
+                                // Move any completed containers aside; we yield them one at a time.
+                                drain_builder(builder, ready);
+                                continue;
+                            }
+
+                            // The key can outrun the budget, so reload it into histories that survive a
+                            // suspension. The strategy is chosen from the same edit counts `think` uses,
+                            // so both paths visit a key's matches in the same order.
+                            let simple = len1 < 10 || len2 < 10;
+                            thinker.history1.clear();
+                            thinker.history2.clear();
+                            cursor1.rewind_vals(storage1);
+                            cursor2.rewind_vals(storage2);
+                            load_current_indexed(vals1, history1, cursor1, storage1, meet1);
+                            load_current_indexed(vals2, history2, cursor2, storage2, meet2);
+                            *resume = Some(if simple {
+                                Resume::Simple(0, 0)
+                            } else {
+                                history1.build();
+                                history2.build();
+                                Resume::WaveChoose
+                            });
+                        }
+
+                        let state = resume.as_mut().expect("key is open");
+                        let retired = {
+                            let (loaded1, loaded2) = (&*vals1, &*vals2);
+                            replay_suspending(history1, history2, state, |index1, index2, t, r1, r2| {
+                                // `logic` takes its key and its values at one lifetime, and the values
+                                // here are borrowed from `vals1`/`vals2` rather than from `storage1`.
+                                // Reborrowing the key narrows it to match, instead of holding the
+                                // values' borrow open for as long as the storage's.
+                                let key1 = <C1::KeyContainer as BatchContainer>::reborrow(key1);
+                                logic(key1, loaded1.index(index1), loaded2.index(index2), t, r1, r2, builder);
+                                // Move any completed containers aside; we yield them one at a time.
+                                drain_builder(builder, ready)
+                            })
+                        };
+
+                        if retired {
+                            cursor1.step_key(storage1);
+                            cursor2.step_key(storage2);
+
+                            history1.clear();
+                            history2.clear();
+                            vals1.clear();
+                            vals2.clear();
+                            *resume = None;
+
+                            drain_builder(builder, ready);
                         }
                     }
                 },
@@ -235,6 +333,110 @@ where
     }
 }
 
+/// Moves every container the builder has completed into `ready`, reporting whether any moved.
+fn drain_builder<CB>(builder: &mut CB, ready: &mut VecDeque<CB::Container>) -> bool
+where
+    CB: ContainerBuilder<Container: Default>,
+{
+    let mut extracted = false;
+    while let Some(container) = builder.extract() {
+        // Avoiding the mem::take would require a non-iterator trait.
+        ready.push_back(std::mem::take(container));
+        extracted = true;
+    }
+    extracted
+}
+
+/// Replays one key's loaded histories, emitting each match, until the key is done or `emit` stops it.
+///
+/// This produces the same matches in the same order as [`JoinThinker::think`], over histories of value
+/// indices rather than of values. It differs in that it can be interrupted: `emit` returns `true` to
+/// suspend, `replay_suspending` then returns `false` having recorded in `resume` where it stopped, and
+/// a later call resumes from there. It returns `true` once the key is fully replayed, at which point
+/// `resume` is spent and the caller may retire the key.
+fn replay_suspending<T, D1, D2, F>(
+    history1: &mut ValueHistory<usize, T, D1>,
+    history2: &mut ValueHistory<usize, T, D2>,
+    resume: &mut Resume,
+    mut emit: F,
+) -> bool
+where
+    T: Ord + Clone + Lattice,
+    D1: Clone + crate::difference::Semigroup,
+    D2: Clone + crate::difference::Semigroup,
+    F: FnMut(usize, usize, T, &D1, &D2) -> bool,
+{
+    loop {
+        match *resume {
+            Resume::Simple(mut index1, mut index2) => {
+                let len1 = history1.edit_len();
+                let len2 = history2.edit_len();
+                while index1 < len1 {
+                    while index2 < len2 {
+                        let (val1, time1, diff1) = history1.edit_at(index1);
+                        let (val2, time2, diff2) = history2.edit_at(index2);
+                        let suspend = emit(val1, val2, time1.join(time2), diff1, diff2);
+                        index2 += 1;
+                        if suspend {
+                            *resume = Resume::Simple(index1, index2);
+                            return false;
+                        }
+                    }
+                    index2 = 0;
+                    index1 += 1;
+                }
+                return true;
+            }
+            Resume::WaveChoose => {
+                // Step whichever history holds the earlier un-replayed time, and once one is spent,
+                // whichever remains. A tie steps the second, as `think`'s comparison does.
+                let step1 = match (history1.time(), history2.time()) {
+                    (Some(time1), Some(time2)) => time1.cmp(time2) == Ordering::Less,
+                    (Some(_), None) => true,
+                    (None, Some(_)) => false,
+                    (None, None) => return true,
+                };
+                if step1 {
+                    history2.advance_buffer_by(history1.meet().unwrap());
+                    *resume = Resume::WaveStep1(0);
+                }
+                else {
+                    history1.advance_buffer_by(history2.meet().unwrap());
+                    *resume = Resume::WaveStep2(0);
+                }
+            }
+            Resume::WaveStep1(mut offset) => {
+                while offset < history2.buffer().len() {
+                    let (val1, time1, diff1) = history1.edit().unwrap();
+                    let ((val2, time2), diff2) = &history2.buffer()[offset];
+                    let suspend = emit(val1, *val2, time1.join(time2), diff1, diff2);
+                    offset += 1;
+                    if suspend {
+                        *resume = Resume::WaveStep1(offset);
+                        return false;
+                    }
+                }
+                history1.step();
+                *resume = Resume::WaveChoose;
+            }
+            Resume::WaveStep2(mut offset) => {
+                while offset < history1.buffer().len() {
+                    let ((val1, time1), diff1) = &history1.buffer()[offset];
+                    let (val2, time2, diff2) = history2.edit().unwrap();
+                    let suspend = emit(*val1, val2, time1.join(time2), diff1, diff2);
+                    offset += 1;
+                    if suspend {
+                        *resume = Resume::WaveStep2(offset);
+                        return false;
+                    }
+                }
+                history2.step();
+                *resume = Resume::WaveChoose;
+            }
+        }
+    }
+}
+
 struct JoinThinker<V1, V2, T, D1, D2> {
     pub history1: ValueHistory<V1, T, D1>,
     pub history2: ValueHistory<V2, T, D2>,
@@ -315,3 +517,180 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+
+    use std::cell::Cell;
+    use std::rc::Rc;
+
+    use timely::container::CapacityContainerBuilder;
+    use timely::container::PushInto;
+
+    use super::{CursorTactic, KEY_WORK_LIMIT};
+    use crate::operators::join::{Fresh, JoinTactic};
+    use crate::trace::Builder;
+    use crate::trace::implementations::ValBuilder;
+    use crate::trace::implementations::ord_neu::OrdValBatch;
+    use crate::trace::implementations::Vector;
+
+    type Update = ((u64, u64), u64, isize);
+    type Batch = Rc<OrdValBatch<Vector<Update>>>;
+    type Output = CapacityContainerBuilder<Vec<Update>>;
+
+    /// A batch holding `vals` values under the single key `0`, each at time `0` with diff `1`.
+    fn one_key(vals: u64) -> Batch {
+        let mut updates: Vec<Update> = (0..vals).map(|val| ((0, val), 0, 1)).collect();
+        let mut builder = <ValBuilder<u64, u64, u64, isize>>::default();
+        builder.push(&mut updates);
+        Rc::new(builder.done().expect("non-empty batch"))
+    }
+
+    /// Joins two single-key batches, reporting the first container, the total matches produced by
+    /// the time it arrived, and the whole run's matches as `(pair count, pair checksum)`.
+    fn join_one_key(vals1: u64, vals2: u64) -> (usize, usize, (usize, u64)) {
+        let produced = Rc::new(Cell::new(0));
+        let counter = Rc::clone(&produced);
+        let mut tactic = CursorTactic::<_, _, _, Output>::new(
+            move |_key: &u64, val1: &u64, val2: &u64, time, diff1: &isize, diff2: &isize, out: &mut Output| {
+                counter.set(counter.get() + 1);
+                out.push_into(((*val1, *val2), time, diff1 * diff2));
+            },
+        );
+
+        let mut work = tactic.prep(vec![one_key(vals1)], vec![one_key(vals2)], Fresh::Input1, 0);
+        let first = work.next().expect("at least one container");
+        let (first_len, produced_by_first) = (first.len(), produced.get());
+
+        let mut count = 0;
+        let mut checksum = 0;
+        for container in std::iter::once(first).chain(work) {
+            for ((val1, val2), time, diff) in container {
+                assert_eq!((time, diff), (0, 1));
+                count += 1;
+                checksum += val1 * vals2 + val2;
+            }
+        }
+        (first_len, produced_by_first, (count, checksum))
+    }
+
+    /// The matches of a `vals1` by `vals2` cross product of distinct values: one per pair, and each
+    /// pair exactly once, which the sum over `val1 * vals2 + val2` pins down.
+    fn expected(vals1: u64, vals2: u64) -> (usize, u64) {
+        let pairs = vals1 * vals2;
+        (pairs as usize, pairs * (pairs - 1) / 2)
+    }
+
+    /// A key large enough to outrun the driver's budget suspends within itself, rather than
+    /// buffering its whole cross product before the first container is served. Both sides carry
+    /// enough edits for the time-ordered wave.
+    #[test]
+    fn wave_suspends_within_a_key() {
+        let vals = (KEY_WORK_LIMIT as f64).sqrt() as u64 + 1;
+        assert!((vals * vals) as usize > KEY_WORK_LIMIT);
+
+        let (first_len, produced_by_first, totals) = join_one_key(vals, vals);
+        assert!(produced_by_first <= first_len + 1, "produced {produced_by_first} for a container of {first_len}");
+        assert_eq!(totals, expected(vals, vals));
+    }
+
+    /// The same, for the nested cross product the join uses when one side is small: few edits on a
+    /// side does not bound the work, only the other side's edit count does.
+    #[test]
+    fn nested_suspends_within_a_key() {
+        let vals1 = 5;
+        let vals2 = KEY_WORK_LIMIT as u64 / vals1 + 1;
+        assert!((vals1 * vals2) as usize > KEY_WORK_LIMIT);
+
+        let (first_len, produced_by_first, totals) = join_one_key(vals1, vals2);
+        assert!(produced_by_first <= first_len + 1, "produced {produced_by_first} for a container of {first_len}");
+        assert_eq!(totals, expected(vals1, vals2));
+    }
+
+    /// A batch holding `vals` values under the single key `0`, each present at `times` distinct
+    /// times, with diffs that differ per value and time so a dropped or duplicated match shows up.
+    fn one_key_over_time(vals: u64, times: u64) -> Batch {
+        let mut updates: Vec<Update> = Vec::new();
+        for val in 0..vals {
+            for time in 0..times {
+                updates.push(((0, val), time, 1 + isize::try_from(val + time).unwrap() % 3));
+            }
+        }
+        let mut builder = <ValBuilder<u64, u64, u64, isize>>::default();
+        builder.push(&mut updates);
+        Rc::new(builder.done().expect("non-empty batch"))
+    }
+
+    /// The `(value, time, diff)` edits `one_key_over_time` produces, as the join sees them.
+    fn edits_over_time(vals: u64, times: u64) -> Vec<(u64, u64, isize)> {
+        (0..vals)
+            .flat_map(|val| (0..times).map(move |time| (val, time, 1 + isize::try_from(val + time).unwrap() % 3)))
+            .collect()
+    }
+
+    /// Accumulates updates into a sorted, consolidated list, dropping those that cancel.
+    fn consolidate(mut updates: Vec<Update>) -> Vec<Update> {
+        updates.sort();
+        let mut out: Vec<Update> = Vec::new();
+        for (data, time, diff) in updates {
+            match out.last_mut() {
+                Some(last) if (last.0, last.1) == (data, time) => last.2 += diff,
+                _ => out.push((data, time, diff)),
+            }
+        }
+        out.retain(|update| update.2 != 0);
+        out
+    }
+
+    /// A key with many distinct times exercises the wave's buffer advance and consolidation, which
+    /// a suspension must resume without repeating. The join's output must consolidate to the same
+    /// collection as the cross product of the two sides' edits, with times joined and diffs
+    /// multiplied.
+    #[test]
+    fn suspended_wave_matches_the_cross_product() {
+        let (vals, times) = (200, 8);
+        assert!(((vals * times) * (vals * times)) as usize > KEY_WORK_LIMIT);
+
+        let mut tactic = CursorTactic::<_, _, _, Output>::new(
+            |_key: &u64, val1: &u64, val2: &u64, time, diff1: &isize, diff2: &isize, out: &mut Output| {
+                out.push_into(((*val1, *val2), time, diff1 * diff2));
+            },
+        );
+        let work = tactic.prep(
+            vec![one_key_over_time(vals, times)],
+            vec![one_key_over_time(vals, times)],
+            Fresh::Input1,
+            0,
+        );
+        let produced: Vec<Update> = work.flatten().collect();
+
+        let mut expected = Vec::new();
+        for (val1, time1, diff1) in edits_over_time(vals, times) {
+            for (val2, time2, diff2) in edits_over_time(vals, times) {
+                expected.push(((val1, val2), time1.max(time2), diff1 * diff2));
+            }
+        }
+
+        // The collections run to millions of updates, so report the first difference rather than
+        // both of them in full.
+        let (produced, expected) = (consolidate(produced), consolidate(expected));
+        let differs = produced.iter().zip(expected.iter()).position(|(x, y)| x != y);
+        assert_eq!(
+            differs.map(|at| (at, produced[at], expected[at])),
+            None,
+            "consolidated join output differs from the cross product",
+        );
+        assert_eq!(produced.len(), expected.len());
+    }
+
+    /// A key whose cross product fits the budget runs to completion in one call, and still produces
+    /// exactly its cross product.
+    #[test]
+    fn small_key_runs_in_one_call() {
+        let vals = 64;
+        assert!((vals * vals) as usize <= KEY_WORK_LIMIT);
+
+        let (_, _, totals) = join_one_key(vals, vals);
+        assert_eq!(totals, expected(vals, vals));
+    }
+}
diff --git a/differential-dataflow/src/operators/history.rs b/differential-dataflow/src/operators/history.rs
index c21386dc4..f4120c071 100644
--- a/differential-dataflow/src/operators/history.rs
+++ b/differential-dataflow/src/operators/history.rs
@@ -39,6 +39,15 @@ impl<V: Copy, T: Ord + Lattice, D: crate::difference::Semigroup> EditList<V, T,
             self.values.push((value, self.edits.len()));
         }
     }
+    /// The `index`th edit, as `(value, time, diff)`, in the order `map` visits them.
+    ///
+    /// The value is found by binary search over the per-value edit bounds, which `seal` leaves
+    /// ascending.
+    fn index(&self, index: usize) -> (V, &T, &D) {
+        let value_index = self.values.partition_point(|&(_, upper)| upper <= index);
+        let (time, diff) = &self.edits[index];
+        (self.values[value_index].0, time, diff)
+    }
     fn map<F: FnMut(V, &T, &D)>(&self, mut logic: F) {
         for index in 0 .. self.values.len() {
             let lower = if index == 0 { 0 } else { self.values[index-1].1 };
@@ -89,8 +98,16 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
         self.edits.map(logic)
     }
 
+    /// The `index`th edit, as `(value, time, diff)`, in the order `map_edits` visits them.
+    ///
+    /// Indexed access lets a reader replay the edits a few at a time, rather than in one
+    /// closure-driven sweep it cannot interrupt.
+    pub(in crate::operators) fn edit_at(&self, index: usize) -> (V, &T, &D) {
+        self.edits.index(index)
+    }
+
     /// Organizes history based on current contents of edits (sort + suffix meets).
-    fn build(&mut self) {
+    pub(in crate::operators) fn build(&mut self) {
         self.buffer.clear();
         self.history.clear();
         for value_index in 0 .. self.edits.values.len() {
diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs
index 761a500be..82bd25bb7 100644
--- a/differential-dataflow/src/operators/join.rs
+++ b/differential-dataflow/src/operators/join.rs
@@ -32,6 +32,23 @@ pub trait JoinTactic<T, B0, B1, C> {
     fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: T) -> Box<dyn Iterator<Item = C>>;
 }
 
+/// Records a join activation may ship before it yields to the timely runtime.
+///
+/// The budget is split evenly between the two input queues, preserving the historical
+/// `1_000_000` of progress per input each activation. Tactics consult it to decide how far
+/// they may run before suspending.
+pub(crate) const FUEL: usize = 2_000_000;
+
+/// Matches a tactic may produce for one key before it must be able to suspend within that key.
+///
+/// A key replayed in one uninterrupted pass buffers its whole cross product, so the bound is what a
+/// join is willing to buffer: the driver's per-queue budget, which is the most one activation ships
+/// from a single iterator anyway. Tactics measure a key by the product of its two sides' edit
+/// counts, an upper bound on its matches. The bound is exact for a nested cross product and
+/// conservative for a replay that nets its inputs down as it goes, so a key a replay would have
+/// kept under the budget is only handled more carefully than it needed to be, never wrongly.
+pub(crate) const KEY_WORK_LIMIT: usize = FUEL / 2;
+
 /// Which input contributed the freshly-arrived batch of a deferred join unit.
 ///
 /// The fresh batch's times all lie at or beyond the capability, so its side is not advanced by the
@@ -293,8 +310,7 @@ where
             // 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.
+            // which is observable directly: an iterator has yet to yield `None`.
             // 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.
@@ -311,9 +327,8 @@ where
                     }
                 }
             };
-            let fuel = 2_000_000;
-            drain(&mut todo0, fuel / 2);
-            drain(&mut todo1, fuel / 2);
+            drain(&mut todo0, (FUEL / 2) as isize);
+            drain(&mut todo1, (FUEL / 2) as isize);
             if !todo0.is_empty() || !todo1.is_empty() {
                 activator.activate();
             }
diff --git a/differential-dataflow/tests/join_bench.rs b/differential-dataflow/tests/join_bench.rs
new file mode 100644
index 000000000..3f4ba7bd3
--- /dev/null
+++ b/differential-dataflow/tests/join_bench.rs
@@ -0,0 +1,84 @@
+//! Benchmarks the cursor join tactic across the key shapes its two strategies cover.
+//!
+//! The tactic runs a key to completion in one call when the key's cross product is small, and
+//! otherwise reloads the key and replays it a container at a time so that it can suspend within the
+//! key. The workloads here bracket that choice:
+//! * `unique` — one value per key on each side: the per-key overhead, where a join spends its time
+//!   opening and retiring keys rather than crossing them.
+//! * `fanout` — a handful of values per key on each side: still a key per call, with real work in it.
+//! * `warm-key` — a single key just under the threshold: the largest key run in one call.
+//! * `hot-key` — a single key well over the threshold: the suspending path, and the shape that
+//!   buffered its whole cross product before the tactic could suspend within a key.
+//!
+//! Each workload loads its inputs, then measures the round that joins them. Run with:
+//! ```text
+//! cargo test --release --test join_bench -- --ignored --nocapture
+//! ```
+
+use std::time::{Duration, Instant};
+
+use timely::dataflow::operators::probe::Handle as ProbeHandle;
+use timely::dataflow::operators::Probe;
+
+use differential_dataflow::input::InputSession;
+
+/// The `(key, value)` pairs of one side of a join: `keys` keys, each carrying `vals` values.
+fn side(keys: u64, vals: u64) -> impl Iterator<Item = (u64, u64)> {
+    (0..keys).flat_map(move |key| (0..vals).map(move |val| (key, val)))
+}
+
+/// Joins `keys` keys of `vals1` by `vals2` values, returning how long the joining round took.
+///
+/// The inputs are loaded and flushed at time `0`, and the round timed is the one that advances to
+/// time `1`, so the measurement covers the join and not the arrangement of its inputs.
+fn join_round(keys: u64, vals1: u64, vals2: u64) -> Duration {
+    let elapsed = timely::execute_directly(move |worker| {
+        let probe = ProbeHandle::new();
+        let mut input1 = <InputSession<u64, (u64, u64), isize>>::new();
+        let mut input2 = <InputSession<u64, (u64, u64), isize>>::new();
+        worker.dataflow(|scope| {
+            let collection1 = input1.to_collection(scope);
+            let collection2 = input2.to_collection(scope);
+            collection1
+                .arrange_by_key()
+                .join_core(collection2.arrange_by_key(), |key, val1: &u64, val2: &u64| {
+                    Some((*key, val1 ^ val2))
+                })
+                .inner
+                .probe_with(&probe);
+        });
+
+        // Load both sides at time `0`, and let the join of an empty side against them settle.
+        for (key, val) in side(keys, vals1) { input1.insert((key, val)); }
+        for (key, val) in side(keys, vals2) { input2.insert((key, val)); }
+        input1.advance_to(1);
+        input2.advance_to(1);
+        input1.flush();
+        input2.flush();
+
+        let start = Instant::now();
+        worker.step_while(|| probe.less_than(input1.time()));
+        start.elapsed()
+    });
+    elapsed
+}
+
+/// Runs one workload and reports its shape, its match count, and the time the joining round took.
+fn report(name: &str, keys: u64, vals1: u64, vals2: u64) {
+    let elapsed = join_round(keys, vals1, vals2);
+    let matches = keys * vals1 * vals2;
+    println!(
+        "{name:<10} keys {keys:>8}  vals {vals1:>7} x {vals2:<7}  matches {matches:>10}  {elapsed:>12.3?}",
+    );
+}
+
+#[test]
+#[ignore]
+fn join_bench() {
+    // Sized against the tactic's threshold, which is half the join driver's fuel: 1,000,000 matches.
+    report("unique", 1_000_000, 1, 1);
+    report("fanout", 100_000, 8, 8);
+    report("warm-key", 1, 900, 900);
+    report("hot-key", 1, 1_500, 1_500);
+    report("hot-key", 1, 3_000, 3_000);
+}

From 329727aada0bf92c2dae075ae3c26491e52dc797 Mon Sep 17 00:00:00 2001
From: Moritz Hoffmann <antiguru@gmail.com>
Date: Sat, 19 Sep 2026 01:42:58 +0200
Subject: [PATCH 2/2] Suspend the proxy join within a key

The proxy join tactic matched a whole block into its match buffers before
handing any of it to the backend. `advance` reports a key entirely within the
block that first mentions it, so a block is at least one whole key, and one
key's cross product is bounded by nothing; a hot key buffered its entire
output the same way the cursor tactic did.

Matching now stops once the buffered matches reach `KEY_WORK_LIMIT`, even
part-way through a key, and resumes from `at0`, `at1` and `resume` on the next
call. Unlike the cursor tactic nothing has to be reloaded: both strategies
already address their inputs by offset, and the histories hold identifiers
rather than borrowed values, so the state the iterator owns is enough.

A backend may now see several `cross` calls for one `advance`, which the trait
documents. The corgi backend's block-local `colliding` set is written by
`advance` and only read by `cross`, so it stays correct across the extra calls.

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

diff --git a/differential-dataflow/src/operators/int_proxy/join.rs b/differential-dataflow/src/operators/int_proxy/join.rs
index 31986ca00..8b6715338 100644
--- a/differential-dataflow/src/operators/int_proxy/join.rs
+++ b/differential-dataflow/src/operators/int_proxy/join.rs
@@ -11,7 +11,7 @@ use timely::progress::Timestamp;
 use crate::difference::{Multiply, Semigroup};
 use crate::lattice::Lattice;
 use super::ProxyBridge;
-use crate::operators::join::{Fresh, JoinTactic};
+use crate::operators::join::{Fresh, JoinTactic, KEY_WORK_LIMIT};
 use crate::operators::history::ValueHistory;
 
 use super::history::IdHistory;
@@ -44,10 +44,11 @@ pub trait ProxyJoinBackend<T, B0, B1> {
         bridge1: &mut ProxyBridge<T, Self::R1>,
     );
 
-    /// Interpret matches derived from the immediately preceding [`Self::advance`] call and place
-    /// them in `output`. The iterator calls `cross` before another `advance`, so a backend may keep
-    /// block-local interpretation state between the two calls. `cross` may be skipped when the
-    /// block produced no matches, in which case the next `advance` may overwrite that state.
+    /// Interpret matches derived from the preceding [`Self::advance`] call and place them in
+    /// `output`. A block is matched in bounded steps, so `cross` may be called several times for
+    /// one `advance`, and is skipped for a step that produced no matches. The iterator always makes
+    /// those calls before the next `advance`, so a backend may keep block-local interpretation state
+    /// across them, and the next `advance` may overwrite it.
     fn cross(
         &mut self,
         instance: &JoinInstance<T, B0, B1>,
@@ -114,6 +115,9 @@ where
             p1: Vec::new(),
             h0: IdHistory::new(),
             h1: IdHistory::new(),
+            at0: 0,
+            at1: 0,
+            resume: None,
             matches: JoinMatches::default(),
             ready: Vec::new(),
         })
@@ -125,6 +129,13 @@ where
 /// 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.
+///
+/// A block is not matched in one go. `advance` reports a key entirely within the block that first
+/// mentions it, so a block is at least one whole key, and one key's cross product is bounded by
+/// nothing; matching a whole block would buffer it all before the back-end saw any of it. Matching
+/// therefore stops once the buffered matches reach [`KEY_WORK_LIMIT`], even part-way through a key,
+/// and `at0`/`at1`/`resume` record where to pick up. The state is all identifiers, times and diffs
+/// the iterator owns, so unlike the cursor tactic it needs nothing reloaded to survive a suspension.
 struct ProxyJoinIter<T, B0, B1, Bk>
 where
     Bk: ProxyJoinBackend<T, B0, B1>,
@@ -142,6 +153,11 @@ where
     /// Per-key replay histories, held across the iterator and reloaded per key when needed.
     h0: IdHistory<T, Bk::R0>,
     h1: IdHistory<T, Bk::R1>,
+    /// How far into the current block's two runs the matching has come.
+    at0: usize,
+    at1: usize,
+    /// Where in the open key's replay the matching suspended; `None` when no key is open.
+    resume: Option<Resume>,
     /// The block's matched records, held across blocks to keep their allocations.
     matches: JoinMatches<T, Bk::ROut>,
     /// The last block's containers, in reverse, served from the back one `next` at a time.
@@ -155,10 +171,13 @@ where
 {
     type Item = Bk::Output;
 
-    /// Serve a ready container, else draw and cross blocks until one yields any.
+    /// Serve a ready container, else match and cross until one yields any.
     fn next(&mut self) -> Option<Bk::Output> {
-        while self.ready.is_empty() && self.from.is_some() {
-            self.refill();
+        while self.ready.is_empty() {
+            if self.block_done() {
+                if self.from.is_none() { break; }
+                self.refill();
+            }
             self.work();
             if !self.matches.ids.is_empty() { self.cross(); }
         }
@@ -171,10 +190,18 @@ where
     T: Timestamp + Lattice,
     Bk: ProxyJoinBackend<T, B0, B1>,
 {
+    /// Whether the current block has been matched in full.
+    fn block_done(&self) -> bool {
+        self.resume.is_none() && (self.at0 >= self.p0.len() || self.at1 >= self.p1.len())
+    }
+
     /// Draw the next block from the backend.
     fn refill(&mut self) {
+        debug_assert!(self.block_done(), "a block is drawn only once the last one is matched");
         self.p0.clear();
         self.p1.clear();
+        self.at0 = 0;
+        self.at1 = 0;
         let before = self.from;
         self.backend.borrow_mut().advance(&self.instance, &mut self.from, &mut self.p0, &mut self.p1);
         // Without progress the iterator would never retire, so this guards liveness as well as contract.
@@ -196,25 +223,27 @@ where
         );
     }
 
-    /// Match the whole of the current block into the match buffers.
+    /// Match the current block into the match buffers, until it is drained or the buffers are full.
     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 (h0, h1) = (&mut self.h0, &mut self.h1);
 
-        let (mut i, mut j) = (0usize, 0usize);
-        while i < p0.len() && j < p1.len() {
-            let ki = p0[i].0.0;
-            debug_assert_eq!(ki, p1[j].0.0, "advance must report common keys");
-            let mut e0 = i;
+        while self.at0 < p0.len() && self.at1 < p1.len() {
+            let ki = p0[self.at0].0.0;
+            debug_assert_eq!(ki, p1[self.at1].0.0, "advance must report common keys");
+            let mut e0 = self.at0;
             while e0 < p0.len() && p0[e0].0.0 == ki { e0 += 1; }
-            let mut e1 = j;
+            let mut e1 = self.at1;
             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;
+            let retired = join_key(ki, p0, self.at0..e0, p1, self.at1..e1, h0, h1, &mut self.matches, &mut self.resume);
+            // A key that suspended stays open, so its runs must still be the ones `at0`/`at1` name.
+            if !retired { return; }
+            self.at0 = e0;
+            self.at1 = e1;
+            if self.matches.ids.len() >= KEY_WORK_LIMIT { return; }
         }
-        debug_assert!(i == p0.len() && j == p1.len(), "both bridges must drain together");
+        debug_assert!(self.at0 == p0.len() && self.at1 == p1.len(), "both bridges must drain together");
     }
 
     /// Turn the block's matches into containers, ready to be served one at a time.
@@ -232,11 +261,35 @@ where
     }
 }
 
-/// Match one key's records across the two presented runs.
+/// Progress through one key, for the matching that suspends within a key.
+///
+/// The variants mirror the two strategies [`join_key`] chooses between: the direct cross product it
+/// uses when a run is short, and the time-ordered wave it uses when both are long. A wave step
+/// advances one history's buffer before emitting against it, so the step it is part-way through is
+/// part of the state: resuming must not advance that buffer a second time, which would consolidate
+/// it and invalidate the offset resumed from.
+#[derive(Clone, Copy)]
+enum Resume {
+    /// The direct cross product, at offsets `(offset0, offset1)` into the key's two runs.
+    Nested(usize, usize),
+    /// The wave, between steps; the next step follows from the histories' least times.
+    WaveChoose,
+    /// The wave, emitting `h1`'s buffer against `h0`'s next edit, from this offset.
+    WaveStep0(usize),
+    /// The wave, emitting `h0`'s buffer against `h1`'s next edit, from this offset.
+    WaveStep1(usize),
+}
+
+/// Match one key's records across the two presented runs, reporting whether the key is retired.
 ///
 /// 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.
+///
+/// A key whose cross product fits [`KEY_WORK_LIMIT`] is matched in one uninterrupted pass. A larger
+/// one stops once `matches` reaches the limit, records where it stopped in `resume`, and returns
+/// `false`; a later call with the same runs and the same `resume` picks up from there. Both
+/// strategies address their inputs by offset, so nothing has to be reloaded across a suspension.
 fn join_key<T, R0, R1, RO>(
     kh: u64,
     p0: &ProxyBridge<T, R0>,
@@ -246,29 +299,124 @@ fn join_key<T, R0, R1, RO>(
     h0: &mut IdHistory<T, R0>,
     h1: &mut IdHistory<T, R1>,
     matches: &mut JoinMatches<T, RO>,
-) where
+    resume: &mut Option<Resume>,
+) -> bool
+where
     T: Lattice + Timestamp,
     R0: Semigroup + Multiply<R1, Output = RO> + Clone,
     R1: Semigroup + Clone,
 {
-    if r0.len() < 16 || r1.len() < 16 {
-        for a in r0 {
-            for b in r1.clone() {
-                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));
+    let nested = r0.len() < 16 || r1.len() < 16;
+
+    if resume.is_none() {
+        if r0.len().saturating_mul(r1.len()) <= KEY_WORK_LIMIT {
+            if nested {
+                for a in r0 {
+                    for b in r1.clone() {
+                        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));
+                    }
+                }
+            }
+            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);
+                bilinear_wave(h0, h1, |v0, v1, t, d| {
+                    matches.ids.push((kh, (v0, v1)));
+                    matches.times.push(t);
+                    matches.diffs.push(d);
+                });
             }
+            return true;
         }
-    }
-    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);
-        bilinear_wave(h0, h1, |v0, v1, t, d| {
-            matches.ids.push((kh, (v0, v1)));
-            matches.times.push(t);
-            matches.diffs.push(d);
+
+        // The key can outrun the budget, so open it for the matching that suspends within it. The
+        // strategy is the one the uninterrupted pass would have taken, so both visit a key's
+        // matches in the same order.
+        *resume = Some(if nested {
+            Resume::Nested(r0.start, r1.start)
+        }
+        else {
+            h0.load_iter(r0.clone().map(|i| (p0[i].0.1, p0[i].1.clone(), p0[i].2.clone())), None);
+            h1.load_iter(r1.clone().map(|i| (p1[i].0.1, p1[i].1.clone(), p1[i].2.clone())), None);
+            Resume::WaveChoose
         });
     }
+
+    let mut state = resume.expect("key is open");
+    loop {
+        match state {
+            Resume::Nested(mut offset0, mut offset1) => {
+                while offset0 < r0.end {
+                    while offset1 < r1.end {
+                        matches.ids.push((kh, (p0[offset0].0.1, p1[offset1].0.1)));
+                        matches.times.push(p0[offset0].1.join(&p1[offset1].1));
+                        matches.diffs.push(p0[offset0].2.clone().multiply(&p1[offset1].2));
+                        offset1 += 1;
+                        if matches.ids.len() >= KEY_WORK_LIMIT {
+                            *resume = Some(Resume::Nested(offset0, offset1));
+                            return false;
+                        }
+                    }
+                    offset1 = r1.start;
+                    offset0 += 1;
+                }
+                *resume = None;
+                return true;
+            }
+            Resume::WaveChoose => {
+                // Step whichever history holds the earlier un-replayed time, and once one is spent,
+                // whichever remains. A tie steps the second, as `bilinear_wave`'s comparison does.
+                let step0 = match (h0.time(), h1.time()) {
+                    (Some(time0), Some(time1)) => time0 < time1,
+                    (Some(_), None) => true,
+                    (None, Some(_)) => false,
+                    (None, None) => { *resume = None; return true; }
+                };
+                if step0 {
+                    h1.advance_buffer_by(h0.meet().unwrap());
+                    state = Resume::WaveStep0(0);
+                }
+                else {
+                    h0.advance_buffer_by(h1.meet().unwrap());
+                    state = Resume::WaveStep1(0);
+                }
+            }
+            Resume::WaveStep0(mut offset) => {
+                while offset < h1.buffer().len() {
+                    let (val0, time0, diff0) = h0.edit().unwrap();
+                    let ((val1, time1), diff1) = &h1.buffer()[offset];
+                    matches.ids.push((kh, (val0, *val1)));
+                    matches.times.push(time0.join(time1));
+                    matches.diffs.push(diff0.clone().multiply(diff1));
+                    offset += 1;
+                    if matches.ids.len() >= KEY_WORK_LIMIT {
+                        *resume = Some(Resume::WaveStep0(offset));
+                        return false;
+                    }
+                }
+                h0.step();
+                state = Resume::WaveChoose;
+            }
+            Resume::WaveStep1(mut offset) => {
+                while offset < h0.buffer().len() {
+                    let ((val0, time0), diff0) = &h0.buffer()[offset];
+                    let (val1, time1, diff1) = h1.edit().unwrap();
+                    matches.ids.push((kh, (*val0, val1)));
+                    matches.times.push(time0.join(time1));
+                    matches.diffs.push(diff0.clone().multiply(diff1));
+                    offset += 1;
+                    if matches.ids.len() >= KEY_WORK_LIMIT {
+                        *resume = Some(Resume::WaveStep1(offset));
+                        return false;
+                    }
+                }
+                h1.step();
+                state = Resume::WaveChoose;
+            }
+        }
+    }
 }
 
 /// Produces the join of two histories: every pair of edits, diffs multiplied and times
@@ -324,3 +472,151 @@ fn bilinear_wave<V, T, R0, R1, RO>(
         h1.step();
     }
 }
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+
+    /// One `(key hash, id0, id1)` match, at a time and with a diff.
+    type Match = ((u64, u64, u64), u64, isize);
+
+    /// A backend that presents its two sides as a single block, and hands matches back verbatim.
+    ///
+    /// The single block is the shape the suspension has to cope with: `advance` must report a key
+    /// entirely within one block, so a hot key cannot be split across blocks however a backend
+    /// draws them.
+    struct OneBlock {
+        side0: ProxyBridge<u64, isize>,
+        side1: ProxyBridge<u64, isize>,
+    }
+
+    impl ProxyJoinBackend<u64, (), ()> for OneBlock {
+        type R0 = isize;
+        type R1 = isize;
+        type ROut = isize;
+        type Output = Vec<Match>;
+
+        fn advance(
+            &mut self,
+            _instance: &JoinInstance<u64, (), ()>,
+            from: &mut Option<u64>,
+            bridge0: &mut ProxyBridge<u64, Self::R0>,
+            bridge1: &mut ProxyBridge<u64, Self::R1>,
+        ) {
+            if from.is_some() {
+                bridge0.clone_from(&self.side0);
+                bridge1.clone_from(&self.side1);
+                *from = None;
+            }
+        }
+
+        fn cross(
+            &mut self,
+            _instance: &JoinInstance<u64, (), ()>,
+            matches: &mut JoinMatches<u64, Self::ROut>,
+            output: &mut Vec<Self::Output>,
+        ) {
+            let crossed = (0..matches.ids.len())
+                .map(|at| {
+                    let (key, (id0, id1)) = matches.ids[at];
+                    ((key, id0, id1), matches.times[at], matches.diffs[at])
+                })
+                .collect();
+            output.push(crossed);
+        }
+    }
+
+    /// One side of a join: `vals` values under the single key hash `0`, each at `times` distinct
+    /// times, with diffs that differ per value and time so a dropped or duplicated match shows up.
+    fn side(vals: u64, times: u64) -> ProxyBridge<u64, isize> {
+        let mut bridge = ProxyBridge::new();
+        for val in 0..vals {
+            for time in 0..times {
+                bridge.push(((0, val), time, 1 + isize::try_from(val + time).unwrap() % 3));
+            }
+        }
+        bridge
+    }
+
+    /// Joins two single-key sides, returning the containers the tactic yields.
+    fn join(side0: ProxyBridge<u64, isize>, side1: ProxyBridge<u64, isize>) -> Vec<Vec<Match>> {
+        let mut tactic = ProxyJoinTactic::new(OneBlock { side0, side1 });
+        tactic.prep(vec![], vec![], Fresh::Input1, 0).collect()
+    }
+
+    /// Accumulates matches into a sorted, consolidated list, dropping those that cancel.
+    fn consolidate(mut matches: Vec<Match>) -> Vec<Match> {
+        matches.sort();
+        let mut out: Vec<Match> = Vec::new();
+        for (data, time, diff) in matches {
+            match out.last_mut() {
+                Some(last) if (last.0, last.1) == (data, time) => last.2 += diff,
+                _ => out.push((data, time, diff)),
+            }
+        }
+        out.retain(|update| update.2 != 0);
+        out
+    }
+
+    /// The cross product of two sides, as the join should produce it.
+    fn cross(side0: &ProxyBridge<u64, isize>, side1: &ProxyBridge<u64, isize>) -> Vec<Match> {
+        let mut out = Vec::new();
+        for ((_, id0), time0, diff0) in side0 {
+            for ((_, id1), time1, diff1) in side1 {
+                out.push(((0, *id0, *id1), *time0.max(time1), diff0 * diff1));
+            }
+        }
+        out
+    }
+
+    /// A key large enough to outrun the driver's budget suspends within itself, rather than
+    /// matching its whole cross product before the backend is handed any of it. Both sides carry
+    /// enough records for the time-ordered wave.
+    #[test]
+    fn wave_suspends_within_a_key() {
+        let vals = 1_100;
+        assert!((vals * vals) as usize > KEY_WORK_LIMIT);
+
+        let containers = join(side(vals, 1), side(vals, 1));
+        assert!(containers.len() > 1, "the key was matched in one go");
+        assert!(containers[0].len() <= KEY_WORK_LIMIT);
+        assert_eq!(containers.iter().map(Vec::len).sum::<usize>(), (vals * vals) as usize);
+    }
+
+    /// The same, for the direct cross product used when one side's run is short: few records on a
+    /// side does not bound the work, only the other side's run length does.
+    #[test]
+    fn nested_suspends_within_a_key() {
+        let (vals0, vals1) = (5, 250_000);
+        assert!((vals0 * vals1) as usize > KEY_WORK_LIMIT);
+
+        let containers = join(side(vals0, 1), side(vals1, 1));
+        assert!(containers.len() > 1, "the key was matched in one go");
+        assert!(containers[0].len() <= KEY_WORK_LIMIT);
+        assert_eq!(containers.iter().map(Vec::len).sum::<usize>(), (vals0 * vals1) as usize);
+    }
+
+    /// A key with many distinct times exercises the wave's buffer advance and consolidation, which
+    /// a suspension must resume without repeating. The matches must consolidate to the same
+    /// collection as the cross product of the two sides, with times joined and diffs multiplied.
+    #[test]
+    fn suspended_wave_matches_the_cross_product() {
+        let (vals, times) = (200, 8);
+        assert!(((vals * times) * (vals * times)) as usize > KEY_WORK_LIMIT);
+
+        let (side0, side1) = (side(vals, times), side(vals, times));
+        let expected = consolidate(cross(&side0, &side1));
+        let produced = consolidate(join(side0, side1).concat());
+
+        // The collections run to millions of matches, so report the first difference rather than
+        // both of them in full.
+        let differs = produced.iter().zip(expected.iter()).position(|(x, y)| x != y);
+        assert_eq!(
+            differs.map(|at| (at, produced[at], expected[at])),
+            None,
+            "consolidated matches differ from the cross product",
+        );
+        assert_eq!(produced.len(), expected.len());
+    }
+}