From 11ce61094b73fd0baee2961ccceaa91df64e50c2 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 29 Jul 2026 19:37:16 -0400
Subject: [PATCH 1/4] int_proxy reduce: split retire into a named phase struct

`retire` held fourteen scratch buffers as inline locals and distinguished its
two phases by comment alone. Those buffers and the times a retire reasons
against now live on a `Retire` struct, whose `run` walks the windows and whose
`determine` / `apply` / `flush` are the phases the comments named.

Pure motion; the only behavioral difference is that `tile_deltas` is cleared at
the start of `apply` rather than the start of the window, which nothing between
the two observes. Also drops a vestigial `'static` on `ProxyReduceBackend::ROut`
and corrects a comment that credited `drain` for what `append` does.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../src/operators/int_proxy/reduce.rs         | 442 +++++++++++-------
 1 file changed, 267 insertions(+), 175 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index ee206a904..eec397353 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -51,7 +51,7 @@ pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     /// Diff type presented for the input.
     type RIn: Semigroup;
     /// Diff type of the output.
-    type ROut: Semigroup + 'static;
+    type ROut: Semigroup;
 
     /// Hash keys and associated times in the instance's novel input batches.
     ///
@@ -152,201 +152,293 @@ where
         let (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems);
         self.backend.begin(&tile_descs);
 
-        let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
+        let mut retire = Retire::new(&mut self.backend, instance, upper, held_elems, tile_of, &self.pending);
+        retire.run(&changed, &seeds);
+        let new_pending = retire.new_pending;
 
+        self.pending = new_pending;
+        let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
+        let mut frontier = Antichain::new();
+        for times in self.pending.values() {
+            for t in times {
+                frontier.insert_ref(t);
+            }
+        }
+        (produced, frontier)
+    }
+}
+
+/// One retire in progress: the backend session it drives, the times it reasons against, and the
+/// scratch its two phases reuse.
+///
+/// The retire proceeds window by window (see [`run`](Self::run)), each window determined then
+/// applied then flushed. All buffers below are cleared per window, round, or moment rather than
+/// reallocated. See the profiling note on [`DiscoverScratch`]: fresh per-key and per-round `Vec`s
+/// were the dominant cost.
+struct Retire<'a, B1, B2, Bk>
+where
+    B1: BatchReader,
+    B2: BatchReader<Time = B1::Time>,
+    Bk: ProxyReduceBackend<B1, B2>,
+{
+    /// The backend, between its `begin` and its `finish`.
+    backend: &'a mut Bk,
+    /// The batches under retirement, as presented to the backend.
+    instance: ReduceInstance<'a, B1, B2>,
+    /// The retire's upper bound: times at or beyond it are withheld rather than applied.
+    upper: &'a Antichain<B1::Time>,
+    /// The held capability times, ascending; `tile_of[i]` is `held[i]`'s tile, if it has one.
+    held: Vec<B1::Time>,
+    tile_of: Vec<Option<usize>>,
+    /// Interesting times withheld by earlier retires, by key hash.
+    pending: &'a BTreeMap<u64, Vec<B1::Time>>,
+    /// Interesting times this retire withholds, by key hash.
+    new_pending: BTreeMap<u64, Vec<B1::Time>>,
+    /// Output updates by held index, accumulated across a window and emitted at its end.
+    tile_deltas: Vec<Vec<((u64, u64), B1::Time, Bk::ROut)>>,
+
+    /// Per-key application state; `n_states` is the prefix [`determine`](Self::determine) filled
+    /// this window. Higher slots persist, retaining their capacity for a later, wider window.
+    states: Vec<KeyState<B1::Time, Bk::RIn, Bk::ROut>>,
+    n_states: usize,
+    /// Time-discovery scratch, and the two lists one key's discovery writes.
+    discover_scratch: DiscoverScratch<B1::Time, Bk::RIn>,
+    moments: Vec<B1::Time>,
+    pended: Vec<B1::Time>,
+
+    /// One round's crossing, assembled across all active keys: their hashes, and their input and
+    /// output accumulations as runs delimited by `in_ends` and `out_ends` (see `reduce_corrections`).
+    batch_keys: Vec<u64>,
+    in_ends: Vec<usize>,
+    in_all: Vec<(u64, Bk::RIn)>,
+    out_ends: Vec<usize>,
+    out_all: Vec<(u64, Bk::ROut)>,
+    /// The key slot and moment each of `batch_keys` was assembled for.
+    active: Vec<(usize, B1::Time)>,
+    /// One key's accumulations, before they are appended to `in_all` and `out_all`.
+    in_accum: Vec<(u64, Bk::RIn)>,
+    cur_out: Vec<(u64, Bk::ROut)>,
+}
+
+impl<'a, B1, B2, Bk> Retire<'a, B1, B2, Bk>
+where
+    B1: BatchReader,
+    B2: BatchReader<Time = B1::Time>,
+    Bk: ProxyReduceBackend<B1, B2>,
+{
+    /// A retire against a backend whose session has begun, with empty scratch.
+    fn new(
+        backend: &'a mut Bk,
+        instance: ReduceInstance<'a, B1, B2>,
+        upper: &'a Antichain<B1::Time>,
+        held: Vec<B1::Time>,
+        tile_of: Vec<Option<usize>>,
+        pending: &'a BTreeMap<u64, Vec<B1::Time>>,
+    ) -> Self {
+        let tile_deltas = (0..held.len()).map(|_| Vec::new()).collect();
+        Retire {
+            backend, instance, upper, held, tile_of, pending,
+            new_pending: BTreeMap::new(),
+            tile_deltas,
+            states: Vec::new(),
+            n_states: 0,
+            discover_scratch: DiscoverScratch::new(),
+            moments: Vec::new(),
+            pended: Vec::new(),
+            batch_keys: Vec::new(),
+            in_ends: Vec::new(),
+            in_all: Vec::new(),
+            out_ends: Vec::new(),
+            out_all: Vec::new(),
+            active: Vec::new(),
+            in_accum: Vec::new(),
+            cur_out: Vec::new(),
+        }
+    }
+
+    /// Retire every window the backend offers, in ascending key-hash order.
+    fn run(&mut self, changed: &[u64], seeds: &[(u64, B1::Time)]) {
         let mut cursor = 0usize;
         let mut ns = 0usize;
+        while let Some(window) = self.backend.next_window(&self.instance, changed, &mut cursor) {
+            super::debug_assert_sorted_bridge(&window.input, "next_window.input");
+            super::debug_assert_sorted_bridge(&window.output, "next_window.output");
+            self.determine(&window, seeds, &mut ns);
+            self.apply();
+            self.flush();
+        }
+    }
 
-        // Retire-wide reusable scratch (cleared per window/round/moment, not reallocated). See the
-        // profiling note on `DiscoverScratch`: fresh per-key/per-round `Vec`s were the dominant cost.
-        let mut discover_scratch: DiscoverScratch<B1::Time, Bk::RIn> = DiscoverScratch::new();
-        let mut states: Vec<KeyState<B1::Time, Bk::RIn, Bk::ROut>> = Vec::new();
-        let mut tile_deltas: Vec<Vec<((u64, u64), B1::Time, Bk::ROut)>> = (0..held_elems.len()).map(|_| Vec::new()).collect();
-        let mut batch_keys: Vec<u64> = Vec::new();
-        let mut in_ends: Vec<usize> = Vec::new();
-        let mut in_all: Vec<(u64, Bk::RIn)> = Vec::new();
-        let mut out_ends: Vec<usize> = Vec::new();
-        let mut out_all: Vec<(u64, Bk::ROut)> = Vec::new();
-        let mut active: Vec<(usize, B1::Time)> = Vec::new();
-        let mut in_accum: Vec<(u64, Bk::RIn)> = Vec::new();
-        let mut cur_out: Vec<(u64, Bk::ROut)> = Vec::new();
-        let mut moments_scratch: Vec<B1::Time> = Vec::new();
-        let mut pended_scratch: Vec<B1::Time> = Vec::new();
-
-        while let Some(window) = self.backend.next_window(&instance, &changed, &mut cursor) {
-            let p_in = &window.input;
-            let p_out = &window.output;
-            super::debug_assert_sorted_bridge(p_in, "next_window.input");
-            super::debug_assert_sorted_bridge(p_out, "next_window.output");
-
-            for deltas in tile_deltas.iter_mut() { deltas.clear(); }
-
-            // Phase 1 (determination): for every key in the window, discover its interesting times
-            // (times only — no accumulation) and stand up its per-moment replays. Peak state is
-            // O(window presentation), bounded by the window `next_window` already materialized.
-            // `states` is a long-lived buffer reloaded slot-by-slot (not cleared/rebuilt): a slot's
-            // `Vec`s and replays are allocated once and reused, so keys cost no per-key alloc/free.
-            // `n_states` is the live prefix this window; higher slots persist (retaining capacity).
-            let mut n_states = 0usize;
-            let (mut is, mut os) = (0usize, 0usize);
-            for &key in &window.keys {
-                while is < p_in.len() && p_in[is].0.0 < key { is += 1; }
-                let i0 = is;
-                while is < p_in.len() && p_in[is].0.0 == key { is += 1; }
-                let i1 = is;
-                while os < p_out.len() && p_out[os].0.0 < key { os += 1; }
-                let o0 = os;
-                while os < p_out.len() && p_out[os].0.0 == key { os += 1; }
-                let o1 = os;
-                while ns < seeds.len() && seeds[ns].0 < key { ns += 1; }
-                let n0 = ns;
-                while ns < seeds.len() && seeds[ns].0 == key { ns += 1; }
-                let n1 = ns;
-
-                moments_scratch.clear();
-                pended_scratch.clear();
-                {
-                    let pending = self.pending.get(&key).map(|p| &p[..]).unwrap_or(&[]);
-                    let seed_times = seeds[n0..n1].iter().map(|(_, t)| t.clone());
-                    let out_times = (o0..o1).map(|o| p_out[o].1.clone());
-                    discover_times(
-                        KeyView { p_in: &p_in[..], i0, i1, pending },
-                        seed_times, out_times, upper,
-                        &mut discover_scratch,
-                        &mut moments_scratch, &mut pended_scratch,
-                    );
-                }
-                if !pended_scratch.is_empty() {
-                    new_pending.insert(key, std::mem::take(&mut pended_scratch));
-                }
-                if moments_scratch.is_empty() {
-                    continue;
-                }
-
-                // Reload slot `n_states` in place (grow the buffer by one only when a window is wider
-                // than any before). `drain` moves the discovered moments in without copy or realloc.
-                if n_states == states.len() {
-                    states.push(KeyState::empty());
-                }
-                let st = &mut states[n_states];
-                st.key = key;
-                st.cursor = 0;
-                st.produced.clear();
-                st.moments.clear();
-                st.moments.append(&mut moments_scratch);
-                st.meets.clear();
-                st.meets.extend(st.moments.iter().cloned());
-                for i in (1..st.meets.len()).rev() {
-                    let m = st.meets[i].clone();
-                    st.meets[i - 1].meet_assign(&m);
-                }
-                st.in_replay.load_iter((i0..i1).map(|i| (p_in[i].0.1, p_in[i].1.clone(), p_in[i].2.clone())), st.meets.first());
-                st.out_replay.load_iter((o0..o1).map(|o| (p_out[o].0.1, p_out[o].1.clone(), p_out[o].2.clone())), st.meets.first());
-                n_states += 1;
+    /// Phase 1 (determination): for every key in the window, discover its interesting times (times
+    /// only — no accumulation) and stand up its per-moment replays, leaving them in `states[..n_states]`.
+    ///
+    /// Peak state is O(window presentation), bounded by the window `next_window` already materialized.
+    /// `states` is a long-lived buffer reloaded slot-by-slot (not cleared/rebuilt): a slot's `Vec`s
+    /// and replays are allocated once and reused, so keys cost no per-key alloc/free.
+    ///
+    /// `ns` is the caller's cursor into `seeds`, which the windows consume in key order.
+    fn determine(&mut self, window: &ReduceWindow<B1::Time, Bk::RIn, Bk::ROut>, seeds: &[(u64, B1::Time)], ns: &mut usize) {
+        let p_in = &window.input;
+        let p_out = &window.output;
+
+        self.n_states = 0;
+        let (mut is, mut os) = (0usize, 0usize);
+        for &key in &window.keys {
+            while is < p_in.len() && p_in[is].0.0 < key { is += 1; }
+            let i0 = is;
+            while is < p_in.len() && p_in[is].0.0 == key { is += 1; }
+            let i1 = is;
+            while os < p_out.len() && p_out[os].0.0 < key { os += 1; }
+            let o0 = os;
+            while os < p_out.len() && p_out[os].0.0 == key { os += 1; }
+            let o1 = os;
+            while *ns < seeds.len() && seeds[*ns].0 < key { *ns += 1; }
+            let n0 = *ns;
+            while *ns < seeds.len() && seeds[*ns].0 == key { *ns += 1; }
+            let n1 = *ns;
+
+            self.moments.clear();
+            self.pended.clear();
+            {
+                let pending = self.pending.get(&key).map(|p| &p[..]).unwrap_or(&[]);
+                let seed_times = seeds[n0..n1].iter().map(|(_, t)| t.clone());
+                let out_times = (o0..o1).map(|o| p_out[o].1.clone());
+                discover_times(
+                    KeyView { p_in: &p_in[..], i0, i1, pending },
+                    seed_times, out_times, self.upper,
+                    &mut self.discover_scratch,
+                    &mut self.moments, &mut self.pended,
+                );
+            }
+            if !self.pended.is_empty() {
+                self.new_pending.insert(key, std::mem::take(&mut self.pended));
+            }
+            if self.moments.is_empty() {
+                continue;
             }
 
-            // Phase 2 (application): walk all keys' moments in ROUNDS. Each round assembles every
-            // active key's one-moment-deep input and current-output accumulations and crosses them in
-            // a SINGLE `reduce_corrections` — batching across keys (a key's own moments stay
-            // sequential, each seeing its earlier corrections via `produced`). This caps the backend
-            // call count at O(max moments over keys), not O(sum of moments), with peak materialization
-            // one moment deep per key. `produced` is meet-collapsed each round, exactly like the
-            // reference — bounded, not the O(times × values) delta history.
-            loop {
-                batch_keys.clear();
-                in_ends.clear();
-                in_all.clear();
-                out_ends.clear();
-                out_all.clear();
-                active.clear();
-                let mut advanced = false;
-                for (si, st) in states[..n_states].iter_mut().enumerate() {
-                    if st.cursor >= st.moments.len() {
-                        continue;
-                    }
-                    advanced = true;
-                    let j = st.cursor;
-                    st.cursor += 1;
-                    let t = st.moments[j].clone();
-                    st.in_replay.step_through(&t);
-                    st.out_replay.step_through(&t);
-                    st.in_replay.advance_buffer_by(&st.meets[j]);
-                    st.out_replay.advance_buffer_by(&st.meets[j]);
-                    for ((_, et), _) in st.produced.iter_mut() {
-                        *et = et.join(&st.meets[j]);
-                    }
-                    crate::consolidation::consolidate(&mut st.produced);
-
-                    in_accum.clear();
-                    for ((vid, et), d) in st.in_replay.buffer().iter() {
-                        if et.less_equal(&t) {
-                            in_accum.push((*vid, d.clone()));
-                        }
-                    }
-                    crate::consolidation::consolidate(&mut in_accum);
-                    cur_out.clear();
-                    for ((vid, et), d) in st.out_replay.buffer().iter().chain(st.produced.iter()) {
-                        if et.less_equal(&t) {
-                            cur_out.push((*vid, d.clone()));
-                        }
-                    }
-                    crate::consolidation::consolidate(&mut cur_out);
+            // Reload slot `n_states` in place (grow the buffer by one only when a window is wider
+            // than any before). `append` moves the discovered moments in without copy or realloc.
+            if self.n_states == self.states.len() {
+                self.states.push(KeyState::empty());
+            }
+            let st = &mut self.states[self.n_states];
+            st.key = key;
+            st.cursor = 0;
+            st.produced.clear();
+            st.moments.clear();
+            st.moments.append(&mut self.moments);
+            st.meets.clear();
+            st.meets.extend(st.moments.iter().cloned());
+            for i in (1..st.meets.len()).rev() {
+                let m = st.meets[i].clone();
+                st.meets[i - 1].meet_assign(&m);
+            }
+            st.in_replay.load_iter((i0..i1).map(|i| (p_in[i].0.1, p_in[i].1.clone(), p_in[i].2.clone())), st.meets.first());
+            st.out_replay.load_iter((o0..o1).map(|o| (p_out[o].0.1, p_out[o].1.clone(), p_out[o].2.clone())), st.meets.first());
+            self.n_states += 1;
+        }
+    }
 
-                    if in_accum.is_empty() && cur_out.is_empty() {
-                        continue;
-                    }
-                    batch_keys.push(st.key);
-                    in_all.append(&mut in_accum);
-                    in_ends.push(in_all.len());
-                    out_all.append(&mut cur_out);
-                    out_ends.push(out_all.len());
-                    active.push((si, t));
-                }
-                // Terminate only when every key is EXHAUSTED — not merely when this round produced no
-                // crossing. A round can be empty because every key's current moment is empty-gated
-                // while keys still have later (non-empty) moments; breaking here would drop them.
-                if !advanced {
-                    break;
-                }
-                if batch_keys.is_empty() {
+    /// Phase 2 (application): walk all keys' moments in ROUNDS, accumulating output updates into
+    /// `tile_deltas`.
+    ///
+    /// Each round assembles every active key's one-moment-deep input and current-output accumulations
+    /// and crosses them in a SINGLE `reduce_corrections` — batching across keys (a key's own moments
+    /// stay sequential, each seeing its earlier corrections via `produced`). This caps the backend
+    /// call count at O(max moments over keys), not O(sum of moments), with peak materialization one
+    /// moment deep per key. `produced` is meet-collapsed each round, exactly like the reference —
+    /// bounded, not the O(times × values) delta history.
+    fn apply(&mut self) {
+        for deltas in self.tile_deltas.iter_mut() { deltas.clear(); }
+
+        loop {
+            self.batch_keys.clear();
+            self.in_ends.clear();
+            self.in_all.clear();
+            self.out_ends.clear();
+            self.out_all.clear();
+            self.active.clear();
+            let mut advanced = false;
+            for (si, st) in self.states[..self.n_states].iter_mut().enumerate() {
+                if st.cursor >= st.moments.len() {
                     continue;
                 }
+                advanced = true;
+                let j = st.cursor;
+                st.cursor += 1;
+                let t = st.moments[j].clone();
+                st.in_replay.step_through(&t);
+                st.out_replay.step_through(&t);
+                st.in_replay.advance_buffer_by(&st.meets[j]);
+                st.out_replay.advance_buffer_by(&st.meets[j]);
+                for ((_, et), _) in st.produced.iter_mut() {
+                    *et = et.join(&st.meets[j]);
+                }
+                crate::consolidation::consolidate(&mut st.produced);
 
-                let (corr, corr_ends) = self.backend.reduce_corrections(&batch_keys, &in_ends, &in_all, &out_ends, &out_all);
-                let mut cstart = 0usize;
-                for (bi, (si, t)) in active.iter().enumerate() {
-                    let cend = corr_ends[bi];
-                    if cstart != cend {
-                        let idx = held_elems.iter().rposition(|h| h.less_equal(t)).expect("no held capability <= active time");
-                        for (vid, d) in &corr[cstart..cend] {
-                            states[*si].produced.push(((*vid, t.clone()), d.clone()));
-                            tile_deltas[idx].push(((states[*si].key, *vid), t.clone(), d.clone()));
-                        }
+                self.in_accum.clear();
+                for ((vid, et), d) in st.in_replay.buffer().iter() {
+                    if et.less_equal(&t) {
+                        self.in_accum.push((*vid, d.clone()));
                     }
-                    cstart = cend;
                 }
-            }
+                crate::consolidation::consolidate(&mut self.in_accum);
+                self.cur_out.clear();
+                for ((vid, et), d) in st.out_replay.buffer().iter().chain(st.produced.iter()) {
+                    if et.less_equal(&t) {
+                        self.cur_out.push((*vid, d.clone()));
+                    }
+                }
+                crate::consolidation::consolidate(&mut self.cur_out);
 
-            for (held_index, deltas) in tile_deltas.iter_mut().enumerate() {
-                if deltas.is_empty() {
+                if self.in_accum.is_empty() && self.cur_out.is_empty() {
                     continue;
                 }
-                if let Some(tile) = tile_of[held_index] {
-                    crate::consolidation::consolidate_updates(deltas);
-                    self.backend.emit(tile, &deltas[..]);
+                self.batch_keys.push(st.key);
+                self.in_all.append(&mut self.in_accum);
+                self.in_ends.push(self.in_all.len());
+                self.out_all.append(&mut self.cur_out);
+                self.out_ends.push(self.out_all.len());
+                self.active.push((si, t));
+            }
+            // Terminate only when every key is EXHAUSTED — not merely when this round produced no
+            // crossing. A round can be empty because every key's current moment is empty-gated
+            // while keys still have later (non-empty) moments; breaking here would drop them.
+            if !advanced {
+                break;
+            }
+            if self.batch_keys.is_empty() {
+                continue;
+            }
+
+            let (corr, corr_ends) = self.backend.reduce_corrections(&self.batch_keys, &self.in_ends, &self.in_all, &self.out_ends, &self.out_all);
+            let mut cstart = 0usize;
+            for (bi, (si, t)) in self.active.iter().enumerate() {
+                let cend = corr_ends[bi];
+                if cstart != cend {
+                    let idx = self.held.iter().rposition(|h| h.less_equal(t)).expect("no held capability <= active time");
+                    for (vid, d) in &corr[cstart..cend] {
+                        self.states[*si].produced.push(((*vid, t.clone()), d.clone()));
+                        self.tile_deltas[idx].push(((self.states[*si].key, *vid), t.clone(), d.clone()));
+                    }
                 }
+                cstart = cend;
             }
         }
+    }
 
-        self.pending = new_pending;
-        let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
-        let mut frontier = Antichain::new();
-        for times in self.pending.values() {
-            for t in times {
-                frontier.insert_ref(t);
+    /// Hand the window's output updates to the backend, one call per tile they land in.
+    fn flush(&mut self) {
+        for (held_index, deltas) in self.tile_deltas.iter_mut().enumerate() {
+            if deltas.is_empty() {
+                continue;
+            }
+            if let Some(tile) = self.tile_of[held_index] {
+                crate::consolidation::consolidate_updates(deltas);
+                self.backend.emit(tile, &deltas[..]);
             }
         }
-        (produced, frontier)
     }
 }
 

From 233c1f700c56009d55b3e56c6b688eb8d2afa395 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 29 Jul 2026 19:39:07 -0400
Subject: [PATCH 2/4] int_proxy reduce: name the reconciliation buffers,
 populate corrections in place

`reduce_corrections` took five undocumented parallel arrays and returned two
more. The inputs are now a `ReduceRound` and the outputs a `ReduceCorrections`,
each documenting how its lists segment by key.

The corrections are populated in place rather than returned owned, so the
harness's allocation survives the round instead of being replaced by a fresh
pair on every one of them.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../src/operators/int_proxy/mod.rs            |   2 +-
 .../src/operators/int_proxy/reduce.rs         | 121 ++++++++++++------
 2 files changed, 86 insertions(+), 37 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs
index 9c72ffeab..bc6581280 100644
--- a/differential-dataflow/src/operators/int_proxy/mod.rs
+++ b/differential-dataflow/src/operators/int_proxy/mod.rs
@@ -64,4 +64,4 @@ pub(crate) fn debug_assert_sorted_bridge<T: Ord, R>(bridge: &ProxyBridge<T, R>,
 }
 
 pub use join::{JoinInstance, ProxyJoinBackend, ProxyJoinTactic};
-pub use reduce::{ProxyReduceBackend, ProxyReduceTactic, ReduceInstance, ReduceWindow};
+pub use reduce::{ProxyReduceBackend, ProxyReduceTactic, ReduceCorrections, ReduceInstance, ReduceRound, ReduceWindow};
diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index eec397353..b21dbad91 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -42,6 +42,67 @@ pub struct ReduceWindow<T, RIn, ROut> {
     pub output: ProxyBridge<T, ROut>,
 }
 
+/// One round of reconciliation: for each of several keys, an accumulated input and tentative output.
+///
+/// The lists are segmented, parallel to `keys`: key `keys[i]` has input `input[..]` over the range
+/// ending at `in_ends[i]` and starting at `in_ends[i-1]` (at zero for `i == 0`), and output `output[..]`
+/// over the range likewise delimited by `out_ends`. So `in_ends` and `out_ends` have `keys`' length,
+/// ascend, and end at the lengths of `input` and `output`. Within a key's range the `u64` are value
+/// ids, ascending and consolidated.
+pub struct ReduceRound<RIn, ROut> {
+    /// The keys reconciled this round, ascending.
+    pub keys: Vec<u64>,
+    /// One past each key's last input record.
+    pub in_ends: Vec<usize>,
+    /// The accumulated input, by key.
+    pub input: Vec<(u64, RIn)>,
+    /// One past each key's last output record.
+    pub out_ends: Vec<usize>,
+    /// The tentative accumulated output, by key.
+    pub output: Vec<(u64, ROut)>,
+}
+
+impl<RIn, ROut> Default for ReduceRound<RIn, ROut> {
+    fn default() -> Self {
+        Self { keys: vec![], in_ends: vec![], input: vec![], out_ends: vec![], output: vec![] }
+    }
+}
+
+impl<RIn, ROut> ReduceRound<RIn, ROut> {
+    /// Empties the round, retaining its allocations for the next one.
+    pub fn clear(&mut self) {
+        self.keys.clear();
+        self.in_ends.clear();
+        self.input.clear();
+        self.out_ends.clear();
+        self.output.clear();
+    }
+}
+
+/// The output updates a backend supplies for a [`ReduceRound`], as one segmented list.
+///
+/// `ends` is parallel to the round's `keys`: key `keys[i]`'s corrections are the `updates` ending at
+/// `ends[i]` and starting at `ends[i-1]` (at zero for `i == 0`). A key needing no correction gets an
+/// empty range, not an omitted one, so `ends` has `keys`' length and ends at `updates`' length.
+pub struct ReduceCorrections<ROut> {
+    /// The corrections, by key. The `u64` are value ids.
+    pub updates: Vec<(u64, ROut)>,
+    /// One past each key's last correction.
+    pub ends: Vec<usize>,
+}
+
+impl<ROut> Default for ReduceCorrections<ROut> {
+    fn default() -> Self { Self { updates: vec![], ends: vec![] } }
+}
+
+impl<ROut> ReduceCorrections<ROut> {
+    /// Empties the corrections, retaining their allocations for the next round.
+    pub fn clear(&mut self) {
+        self.updates.clear();
+        self.ends.clear();
+    }
+}
+
 /// The reduce backend: value semantics for a proxy-space reduction, driven by [`ProxyReduceTactic`].
 ///
 /// The protocol is currently (temporarily) for each round of invocation:
@@ -75,16 +136,13 @@ pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     /// A wave of input-output reconciliation, in which the backend supplies necessary edits.
     ///
     /// Multiple keys are provided concurrently, for each an accumulated input and tentative output.
-    /// The backend should provide for each key the necessary output updates to bring the output in
-    /// with its desires. The `usize` integers upper bound the range for the corresponding key.
+    /// The backend should populate `corrections`, which arrives empty, with for each key the output
+    /// updates that bring the output in line with its desires.
     fn reduce_corrections(
         &mut self,
-        keys: &[u64],
-        in_ends: &[usize],
-        input: &[(u64, Self::RIn)],
-        out_ends: &[usize],
-        output: &[(u64, Self::ROut)],
-    ) -> (Vec<(u64, Self::ROut)>, Vec<usize>);
+        round: &ReduceRound<Self::RIn, Self::ROut>,
+        corrections: &mut ReduceCorrections<Self::ROut>,
+    );
 
     /// Commit to a collection of updates at a specific batch in progress.
     ///
@@ -206,16 +264,12 @@ where
     moments: Vec<B1::Time>,
     pended: Vec<B1::Time>,
 
-    /// One round's crossing, assembled across all active keys: their hashes, and their input and
-    /// output accumulations as runs delimited by `in_ends` and `out_ends` (see `reduce_corrections`).
-    batch_keys: Vec<u64>,
-    in_ends: Vec<usize>,
-    in_all: Vec<(u64, Bk::RIn)>,
-    out_ends: Vec<usize>,
-    out_all: Vec<(u64, Bk::ROut)>,
-    /// The key slot and moment each of `batch_keys` was assembled for.
+    /// One round's crossing, assembled across all active keys, and the corrections it draws.
+    round: ReduceRound<Bk::RIn, Bk::ROut>,
+    corrections: ReduceCorrections<Bk::ROut>,
+    /// The key slot and moment each of the round's keys was assembled for.
     active: Vec<(usize, B1::Time)>,
-    /// One key's accumulations, before they are appended to `in_all` and `out_all`.
+    /// One key's accumulations, before they are appended to the round.
     in_accum: Vec<(u64, Bk::RIn)>,
     cur_out: Vec<(u64, Bk::ROut)>,
 }
@@ -245,11 +299,8 @@ where
             discover_scratch: DiscoverScratch::new(),
             moments: Vec::new(),
             pended: Vec::new(),
-            batch_keys: Vec::new(),
-            in_ends: Vec::new(),
-            in_all: Vec::new(),
-            out_ends: Vec::new(),
-            out_all: Vec::new(),
+            round: ReduceRound::default(),
+            corrections: ReduceCorrections::default(),
             active: Vec::new(),
             in_accum: Vec::new(),
             cur_out: Vec::new(),
@@ -353,11 +404,7 @@ where
         for deltas in self.tile_deltas.iter_mut() { deltas.clear(); }
 
         loop {
-            self.batch_keys.clear();
-            self.in_ends.clear();
-            self.in_all.clear();
-            self.out_ends.clear();
-            self.out_all.clear();
+            self.round.clear();
             self.active.clear();
             let mut advanced = false;
             for (si, st) in self.states[..self.n_states].iter_mut().enumerate() {
@@ -395,11 +442,11 @@ where
                 if self.in_accum.is_empty() && self.cur_out.is_empty() {
                     continue;
                 }
-                self.batch_keys.push(st.key);
-                self.in_all.append(&mut self.in_accum);
-                self.in_ends.push(self.in_all.len());
-                self.out_all.append(&mut self.cur_out);
-                self.out_ends.push(self.out_all.len());
+                self.round.keys.push(st.key);
+                self.round.input.append(&mut self.in_accum);
+                self.round.in_ends.push(self.round.input.len());
+                self.round.output.append(&mut self.cur_out);
+                self.round.out_ends.push(self.round.output.len());
                 self.active.push((si, t));
             }
             // Terminate only when every key is EXHAUSTED — not merely when this round produced no
@@ -408,17 +455,19 @@ where
             if !advanced {
                 break;
             }
-            if self.batch_keys.is_empty() {
+            if self.round.keys.is_empty() {
                 continue;
             }
 
-            let (corr, corr_ends) = self.backend.reduce_corrections(&self.batch_keys, &self.in_ends, &self.in_all, &self.out_ends, &self.out_all);
+            self.corrections.clear();
+            self.backend.reduce_corrections(&self.round, &mut self.corrections);
+            debug_assert_eq!(self.corrections.ends.len(), self.round.keys.len(), "corrections must delimit one run per key");
             let mut cstart = 0usize;
             for (bi, (si, t)) in self.active.iter().enumerate() {
-                let cend = corr_ends[bi];
+                let cend = self.corrections.ends[bi];
                 if cstart != cend {
                     let idx = self.held.iter().rposition(|h| h.less_equal(t)).expect("no held capability <= active time");
-                    for (vid, d) in &corr[cstart..cend] {
+                    for (vid, d) in &self.corrections.updates[cstart..cend] {
                         self.states[*si].produced.push(((*vid, t.clone()), d.clone()));
                         self.tile_deltas[idx].push(((self.states[*si].key, *vid), t.clone(), d.clone()));
                     }

From 5b51e7d60a06194fc3fbf953a2250d1a4e728caf Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 29 Jul 2026 19:44:52 -0400
Subject: [PATCH 3/4] int_proxy reduce: draw windows with an advance, dropping
 seed_times

`seed_times` scanned the whole instance up front and returned a time for every
changed key, so a retire materialized its entire seed set before doing any work.
The harness then derived the changed keys from it and handed them back to the
backend to window.

`advance` replaces both it and `next_window`, on join's model: progress is an
inclusive lower bound on key hash that the backend advances to the exclusive
upper bound of the range it covers, and the window it populates now carries the
seeds for its own keys. What the harness contributes is the times earlier retires
withheld, presented flat and key-ordered for the backend to move through and
block its windows against.

The window is populated in place rather than returned owned, so its three
buffers survive from one window to the next.

The output session now opens on the first window with work, preserving the
behavior of the old changed-is-empty short circuit: a retire with nothing to do
ships no batches rather than an empty one per tile. `advance` correspondingly
draws a non-empty window unless it reports exhaustion.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../src/operators/int_proxy/reduce.rs         | 184 +++++++++++++-----
 1 file changed, 134 insertions(+), 50 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index b21dbad91..ae8ebc52d 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -3,7 +3,7 @@
 //! A conventional differential reduce against `(u64, u64)`, where the backend supplies the
 //! implementation of the interpretation of the integers.
 
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::BTreeMap;
 
 use timely::PartialOrder;
 use timely::progress::{Antichain, Timestamp};
@@ -32,14 +32,36 @@ pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>>
 
 /// One window of a retire's changed keys: a bounded, hash-contiguous snip the backend sizes.
 ///
-/// The window has the input (old and new) and output histories, restricted to the window's keys.
+/// The window has the input (old and new) and output histories, restricted to the window's keys,
+/// and the times the new input makes interesting for them.
 pub struct ReduceWindow<T, RIn, ROut> {
-    /// The window's key hashes: a contiguous, ascending slice of the retire's `changed` keys.
+    /// The window's key hashes: a contiguous, ascending slice of the retire's changed keys.
     pub keys: Vec<u64>,
     /// Input presentation for `keys`, sorted & consolidated by `((key_hash, value_id), time)`.
     pub input: ProxyBridge<T, RIn>,
     /// Output-history presentation for `keys`, same ordering.
     pub output: ProxyBridge<T, ROut>,
+    /// Times the instance's *new* input batches carry for `keys`, sorted by key hash.
+    ///
+    /// These seed each key's interesting times, together with the times held over from earlier
+    /// retires. A key needs an entry here or in those held times to belong in `keys` at all.
+    pub seeds: Vec<(u64, T)>,
+}
+
+impl<T, RIn, ROut> Default for ReduceWindow<T, RIn, ROut> {
+    fn default() -> Self {
+        Self { keys: vec![], input: vec![], output: vec![], seeds: vec![] }
+    }
+}
+
+impl<T, RIn, ROut> ReduceWindow<T, RIn, ROut> {
+    /// Empties the window, retaining its allocations for the next one.
+    pub fn clear(&mut self) {
+        self.keys.clear();
+        self.input.clear();
+        self.output.clear();
+        self.seeds.clear();
+    }
 }
 
 /// One round of reconciliation: for each of several keys, an accumulated input and tentative output.
@@ -105,20 +127,18 @@ impl<ROut> ReduceCorrections<ROut> {
 
 /// The reduce backend: value semantics for a proxy-space reduction, driven by [`ProxyReduceTactic`].
 ///
-/// The protocol is currently (temporarily) for each round of invocation:
-/// `seed_times begin [ next_window reduce_correction* emit ]* finish`
-/// This should be improved to put the `seed_times` in the per-window loop, or remove it entirely.
+/// The harness repeatedly invokes [`advance`](Self::advance) to draw a window of changed keys, then
+/// reconciles that window's keys against their output with [`reduce_corrections`](Self::reduce_corrections),
+/// handing the resulting updates back with [`emit`](Self::emit), until `advance` reports the key space
+/// exhausted. The output session brackets whatever windows have work:
+/// `advance ( begin [ reduce_corrections* emit* advance ]* finish )?`
+/// A retire with no changed keys draws one window, finds it empty, and never opens the session.
 pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
     /// Diff type presented for the input.
     type RIn: Semigroup;
     /// Diff type of the output.
     type ROut: Semigroup;
 
-    /// Hash keys and associated times in the instance's novel input batches.
-    ///
-    /// This is used (with held times) to seed the interesting times for each key.
-    fn seed_times(&self, instance: &ReduceInstance<'_, B1, B2>) -> Vec<(u64, B1::Time)>;
-
     /// Initiate a session to create batches for these descriptions, which span `[lower, upper)`.
     ///
     /// It is the backend's job to prepare output batches for each of these descriptions.
@@ -126,12 +146,26 @@ pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     /// work in progress, until `finish()` is called.
     fn begin(&mut self, tiles: &[Description<B1::Time>]);
 
-    /// Produce the next window, resticted to `changed[cursor..]`, and update `cursor` to track.
+    /// Populates `window` with the next range of changed keys, and everything known about them.
+    ///
+    /// 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 covers in this call. The
+    /// `None` value indicates the keys are exhausted. A key is *changed* if the instance's new input
+    /// batches carry a time for it, or if it appears in `pending`: the interesting times earlier
+    /// retires withheld, ascending by key hash, which the implementor moves through alongside its own
+    /// input. The window must hold every changed key in the range, with all of both inputs' updates
+    /// for those keys, and it must be non-empty unless `from` is returned as `None`.
     ///
     /// The size of the window is up to the backend, where the window should be large enough to
     /// amortize the crossings between the harness and the backend. The proxy bridges for the
     /// whole window will be active at the same time, so tighter windows reduce the required state.
-    fn next_window(&mut self, instance: &ReduceInstance<'_, B1, B2>, changed: &[u64], cursor: &mut usize) -> Option<ReduceWindow<B1::Time, Self::RIn, Self::ROut>>;
+    fn advance(
+        &mut self,
+        instance: &ReduceInstance<'_, B1, B2>,
+        from: &mut Option<u64>,
+        pending: &[(u64, B1::Time)],
+        window: &mut ReduceWindow<B1::Time, Self::RIn, Self::ROut>,
+    );
 
     /// A wave of input-output reconciliation, in which the backend supplies necessary edits.
     ///
@@ -194,27 +228,27 @@ where
             lower: lower.borrow(),
         };
 
-        let seeds = self.backend.seed_times(&instance);
-        debug_assert!(seeds.windows(2).all(|w| w[0].0 <= w[1].0), "seed_times must be sorted by key_hash");
-        let mut changed: BTreeSet<u64> = seeds.iter().map(|(k, _)| *k).collect();
-        changed.extend(self.pending.keys().copied());
-        if changed.is_empty() {
-            self.pending.clear();
-            return (Vec::new(), Antichain::new());
-        }
-        let changed: Vec<u64> = changed.into_iter().collect();
-
         // The output tiling (identical to the Abelian tactic): one tile per held time, keeping
         // non-degenerate intervals; `tile_of[i]` maps held time `i` to its tile.
         let held_elems: Vec<B1::Time> = held.elements().to_vec();
         let (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems);
-        self.backend.begin(&tile_descs);
 
-        let mut retire = Retire::new(&mut self.backend, instance, upper, held_elems, tile_of, &self.pending);
-        retire.run(&changed, &seeds);
+        // The withheld times as the backend sees them: one flat, key-ordered list to move through.
+        // The harness keeps its own map, which it indexes by key as it determines each one.
+        let pending_flat: Vec<(u64, B1::Time)> = self.pending.iter()
+            .flat_map(|(key, times)| times.iter().map(move |t| (*key, t.clone())))
+            .collect();
+
+        let mut retire = Retire::new(&mut self.backend, instance, upper, held_elems, tile_of, &self.pending, &pending_flat);
+        let began = retire.run(&tile_descs);
         let new_pending = retire.new_pending;
 
         self.pending = new_pending;
+        // No window had work, so no session was opened and there is nothing to finish or to withhold.
+        if !began {
+            debug_assert!(self.pending.is_empty(), "a retire that determined no key cannot withhold a time");
+            return (Vec::new(), Antichain::new());
+        }
         let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
         let mut frontier = Antichain::new();
         for times in self.pending.values() {
@@ -239,7 +273,7 @@ where
     B2: BatchReader<Time = B1::Time>,
     Bk: ProxyReduceBackend<B1, B2>,
 {
-    /// The backend, between its `begin` and its `finish`.
+    /// The backend, whose output session [`run`](Self::run) opens and the caller closes.
     backend: &'a mut Bk,
     /// The batches under retirement, as presented to the backend.
     instance: ReduceInstance<'a, B1, B2>,
@@ -248,13 +282,21 @@ where
     /// The held capability times, ascending; `tile_of[i]` is `held[i]`'s tile, if it has one.
     held: Vec<B1::Time>,
     tile_of: Vec<Option<usize>>,
-    /// Interesting times withheld by earlier retires, by key hash.
+    /// Interesting times withheld by earlier retires, by key hash, and the flat presentation of
+    /// the same times the backend moves through.
     pending: &'a BTreeMap<u64, Vec<B1::Time>>,
+    pending_flat: &'a [(u64, B1::Time)],
     /// Interesting times this retire withholds, by key hash.
     new_pending: BTreeMap<u64, Vec<B1::Time>>,
     /// Output updates by held index, accumulated across a window and emitted at its end.
     tile_deltas: Vec<Vec<((u64, u64), B1::Time, Bk::ROut)>>,
 
+    /// Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
+    /// once the backend reports the keys exhausted.
+    from: Option<u64>,
+    /// The window last drawn, held across draws to keep its allocations.
+    window: ReduceWindow<B1::Time, Bk::RIn, Bk::ROut>,
+
     /// Per-key application state; `n_states` is the prefix [`determine`](Self::determine) filled
     /// this window. Higher slots persist, retaining their capacity for a later, wider window.
     states: Vec<KeyState<B1::Time, Bk::RIn, Bk::ROut>>,
@@ -280,7 +322,7 @@ where
     B2: BatchReader<Time = B1::Time>,
     Bk: ProxyReduceBackend<B1, B2>,
 {
-    /// A retire against a backend whose session has begun, with empty scratch.
+    /// A retire at the start of the key space, with empty scratch and no session yet opened.
     fn new(
         backend: &'a mut Bk,
         instance: ReduceInstance<'a, B1, B2>,
@@ -288,12 +330,15 @@ where
         held: Vec<B1::Time>,
         tile_of: Vec<Option<usize>>,
         pending: &'a BTreeMap<u64, Vec<B1::Time>>,
+        pending_flat: &'a [(u64, B1::Time)],
     ) -> Self {
         let tile_deltas = (0..held.len()).map(|_| Vec::new()).collect();
         Retire {
-            backend, instance, upper, held, tile_of, pending,
+            backend, instance, upper, held, tile_of, pending, pending_flat,
             new_pending: BTreeMap::new(),
             tile_deltas,
+            from: Some(0),
+            window: ReduceWindow::default(),
             states: Vec::new(),
             n_states: 0,
             discover_scratch: DiscoverScratch::new(),
@@ -307,34 +352,73 @@ where
         }
     }
 
-    /// Retire every window the backend offers, in ascending key-hash order.
-    fn run(&mut self, changed: &[u64], seeds: &[(u64, B1::Time)]) {
-        let mut cursor = 0usize;
-        let mut ns = 0usize;
-        while let Some(window) = self.backend.next_window(&self.instance, changed, &mut cursor) {
-            super::debug_assert_sorted_bridge(&window.input, "next_window.input");
-            super::debug_assert_sorted_bridge(&window.output, "next_window.output");
-            self.determine(&window, seeds, &mut ns);
+    /// Retire every window the backend offers, in ascending key-hash order, opening the output
+    /// session on the first window that has work. Reports whether the session was opened.
+    fn run(&mut self, tiles: &[Description<B1::Time>]) -> bool {
+        let mut began = false;
+        while self.from.is_some() {
+            self.draw();
+            // The backend should only report an empty window when it reports exhaustion, but skipping
+            // one costs nothing, where terminating on one would abandon the keys beyond it.
+            if self.window.keys.is_empty() { continue; }
+            if !began {
+                self.backend.begin(tiles);
+                began = true;
+            }
+            self.determine();
             self.apply();
             self.flush();
         }
+        began
+    }
+
+    /// Draw the next window from the backend.
+    fn draw(&mut self) {
+        self.window.clear();
+        let before = self.from;
+        self.backend.advance(&self.instance, &mut self.from, self.pending_flat, &mut self.window);
+        // Without progress the retire would never terminate, 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 keys exhausted",
+        );
+        debug_assert!(
+            self.from.is_none() || !self.window.keys.is_empty(),
+            "advance must draw a non-empty window unless it reports the keys exhausted",
+        );
+        super::debug_assert_sorted_bridge(&self.window.input, "advance (input)");
+        super::debug_assert_sorted_bridge(&self.window.output, "advance (output)");
+        debug_assert!(self.window.keys.windows(2).all(|w| w[0] < w[1]), "a window's keys must ascend");
+        debug_assert!(self.window.seeds.windows(2).all(|w| w[0].0 <= w[1].0), "a window's seeds must be sorted by key hash");
+        // A changed key outside `[before, from)` is either one an earlier window already retired, or
+        // one a later window may yet report: both split a key's times across windows, which reconciles
+        // it against an input it is not yet whole.
+        debug_assert!(
+            {
+                let mut keys = self.window.keys.iter().copied()
+                    .chain(self.window.input.iter().map(|r| r.0.0))
+                    .chain(self.window.output.iter().map(|r| r.0.0))
+                    .chain(self.window.seeds.iter().map(|s| s.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 window that first mentions it",
+        );
     }
 
     /// Phase 1 (determination): for every key in the window, discover its interesting times (times
     /// only — no accumulation) and stand up its per-moment replays, leaving them in `states[..n_states]`.
     ///
-    /// Peak state is O(window presentation), bounded by the window `next_window` already materialized.
+    /// Peak state is O(window presentation), bounded by the window `advance` already materialized.
     /// `states` is a long-lived buffer reloaded slot-by-slot (not cleared/rebuilt): a slot's `Vec`s
     /// and replays are allocated once and reused, so keys cost no per-key alloc/free.
-    ///
-    /// `ns` is the caller's cursor into `seeds`, which the windows consume in key order.
-    fn determine(&mut self, window: &ReduceWindow<B1::Time, Bk::RIn, Bk::ROut>, seeds: &[(u64, B1::Time)], ns: &mut usize) {
-        let p_in = &window.input;
-        let p_out = &window.output;
+    fn determine(&mut self) {
+        let p_in = &self.window.input;
+        let p_out = &self.window.output;
+        let seeds = &self.window.seeds;
 
         self.n_states = 0;
-        let (mut is, mut os) = (0usize, 0usize);
-        for &key in &window.keys {
+        let (mut is, mut os, mut ns) = (0usize, 0usize, 0usize);
+        for &key in &self.window.keys {
             while is < p_in.len() && p_in[is].0.0 < key { is += 1; }
             let i0 = is;
             while is < p_in.len() && p_in[is].0.0 == key { is += 1; }
@@ -343,10 +427,10 @@ where
             let o0 = os;
             while os < p_out.len() && p_out[os].0.0 == key { os += 1; }
             let o1 = os;
-            while *ns < seeds.len() && seeds[*ns].0 < key { *ns += 1; }
-            let n0 = *ns;
-            while *ns < seeds.len() && seeds[*ns].0 == key { *ns += 1; }
-            let n1 = *ns;
+            while ns < seeds.len() && seeds[ns].0 < key { ns += 1; }
+            let n0 = ns;
+            while ns < seeds.len() && seeds[ns].0 == key { ns += 1; }
+            let n1 = ns;
 
             self.moments.clear();
             self.pended.clear();

From f84688f350a3ccae828b489f45cf46d7f47c8a52 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Wed, 29 Jul 2026 19:55:46 -0400
Subject: [PATCH 4/4] int_proxy reduce: drop three of Retire's members

Twenty-one fields to eighteen, none of them costing an allocation.

`pended` becomes a local. It looked like a twin of `moments`, but the two are
drained differently: `append` leaves `moments`' capacity behind for the next key,
where `new_pending` takes `pended`'s allocation whole and leaves nothing. So it
either never allocated or was carried off, and a local does the same work. The
`moments` field now documents that asymmetry, since it is the reason only one of
the pair is a field.

`n_states` becomes a value returned by `determine` and passed to `apply`. It was
per-window state in a per-retire struct, and the dependency between the phases
was invisible.

`held` and `tile_of` were parallel by held index, and become one `tiles` list.

Also renames `cur_out` to `out_accum`, matching `in_accum` beside it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../src/operators/int_proxy/reduce.rs         | 91 ++++++++++---------
 1 file changed, 49 insertions(+), 42 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index ae8ebc52d..700293dc7 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -232,6 +232,7 @@ where
         // non-degenerate intervals; `tile_of[i]` maps held time `i` to its tile.
         let held_elems: Vec<B1::Time> = held.elements().to_vec();
         let (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems);
+        let tiles: Vec<(B1::Time, Option<usize>)> = held_elems.into_iter().zip(tile_of).collect();
 
         // The withheld times as the backend sees them: one flat, key-ordered list to move through.
         // The harness keeps its own map, which it indexes by key as it determines each one.
@@ -239,7 +240,7 @@ where
             .flat_map(|(key, times)| times.iter().map(move |t| (*key, t.clone())))
             .collect();
 
-        let mut retire = Retire::new(&mut self.backend, instance, upper, held_elems, tile_of, &self.pending, &pending_flat);
+        let mut retire = Retire::new(&mut self.backend, instance, upper, tiles, &self.pending, &pending_flat);
         let began = retire.run(&tile_descs);
         let new_pending = retire.new_pending;
 
@@ -279,9 +280,10 @@ where
     instance: ReduceInstance<'a, B1, B2>,
     /// The retire's upper bound: times at or beyond it are withheld rather than applied.
     upper: &'a Antichain<B1::Time>,
-    /// The held capability times, ascending; `tile_of[i]` is `held[i]`'s tile, if it has one.
-    held: Vec<B1::Time>,
-    tile_of: Vec<Option<usize>>,
+    /// The held capability times, ascending, each with the tile it commits to.
+    ///
+    /// A held time whose interval was degenerate has no tile; `tile_descriptions` skipped it.
+    tiles: Vec<(B1::Time, Option<usize>)>,
     /// Interesting times withheld by earlier retires, by key hash, and the flat presentation of
     /// the same times the backend moves through.
     pending: &'a BTreeMap<u64, Vec<B1::Time>>,
@@ -297,23 +299,27 @@ where
     /// The window last drawn, held across draws to keep its allocations.
     window: ReduceWindow<B1::Time, Bk::RIn, Bk::ROut>,
 
-    /// Per-key application state; `n_states` is the prefix [`determine`](Self::determine) filled
-    /// this window. Higher slots persist, retaining their capacity for a later, wider window.
+    /// Per-key application state, in slots [`determine`](Self::determine) refills window by window.
+    /// It reports how many it filled; higher slots persist, retaining their capacity for a later,
+    /// wider window.
     states: Vec<KeyState<B1::Time, Bk::RIn, Bk::ROut>>,
-    n_states: usize,
-    /// Time-discovery scratch, and the two lists one key's discovery writes.
+    /// Time-discovery scratch.
     discover_scratch: DiscoverScratch<B1::Time, Bk::RIn>,
+    /// One key's discovered times inside `upper`, before they move into its state slot.
+    ///
+    /// A field rather than a local because `append` drains it and leaves its capacity behind, so the
+    /// next key reuses it. Its counterpart — the times at or beyond `upper` — is a local, because
+    /// `new_pending` takes that allocation whole and leaves nothing to reuse.
     moments: Vec<B1::Time>,
-    pended: Vec<B1::Time>,
 
     /// One round's crossing, assembled across all active keys, and the corrections it draws.
     round: ReduceRound<Bk::RIn, Bk::ROut>,
     corrections: ReduceCorrections<Bk::ROut>,
-    /// The key slot and moment each of the round's keys was assembled for.
+    /// The state slot and moment each of the round's keys was assembled for, parallel to its `keys`.
     active: Vec<(usize, B1::Time)>,
-    /// One key's accumulations, before they are appended to the round.
+    /// One key's accumulations, before `append` drains them into the round and leaves their capacity.
     in_accum: Vec<(u64, Bk::RIn)>,
-    cur_out: Vec<(u64, Bk::ROut)>,
+    out_accum: Vec<(u64, Bk::ROut)>,
 }
 
 impl<'a, B1, B2, Bk> Retire<'a, B1, B2, Bk>
@@ -327,28 +333,25 @@ where
         backend: &'a mut Bk,
         instance: ReduceInstance<'a, B1, B2>,
         upper: &'a Antichain<B1::Time>,
-        held: Vec<B1::Time>,
-        tile_of: Vec<Option<usize>>,
+        tiles: Vec<(B1::Time, Option<usize>)>,
         pending: &'a BTreeMap<u64, Vec<B1::Time>>,
         pending_flat: &'a [(u64, B1::Time)],
     ) -> Self {
-        let tile_deltas = (0..held.len()).map(|_| Vec::new()).collect();
+        let tile_deltas = (0..tiles.len()).map(|_| Vec::new()).collect();
         Retire {
-            backend, instance, upper, held, tile_of, pending, pending_flat,
+            backend, instance, upper, tiles, pending, pending_flat,
             new_pending: BTreeMap::new(),
             tile_deltas,
             from: Some(0),
             window: ReduceWindow::default(),
             states: Vec::new(),
-            n_states: 0,
             discover_scratch: DiscoverScratch::new(),
             moments: Vec::new(),
-            pended: Vec::new(),
             round: ReduceRound::default(),
             corrections: ReduceCorrections::default(),
             active: Vec::new(),
             in_accum: Vec::new(),
-            cur_out: Vec::new(),
+            out_accum: Vec::new(),
         }
     }
 
@@ -365,8 +368,8 @@ where
                 self.backend.begin(tiles);
                 began = true;
             }
-            self.determine();
-            self.apply();
+            let live = self.determine();
+            self.apply(live);
             self.flush();
         }
         began
@@ -406,17 +409,20 @@ where
     }
 
     /// Phase 1 (determination): for every key in the window, discover its interesting times (times
-    /// only — no accumulation) and stand up its per-moment replays, leaving them in `states[..n_states]`.
+    /// only — no accumulation) and stand up its per-moment replays. Returns how many `states` slots
+    /// it filled, which is what [`apply`](Self::apply) then walks.
     ///
     /// Peak state is O(window presentation), bounded by the window `advance` already materialized.
     /// `states` is a long-lived buffer reloaded slot-by-slot (not cleared/rebuilt): a slot's `Vec`s
     /// and replays are allocated once and reused, so keys cost no per-key alloc/free.
-    fn determine(&mut self) {
+    fn determine(&mut self) -> usize {
         let p_in = &self.window.input;
         let p_out = &self.window.output;
         let seeds = &self.window.seeds;
 
-        self.n_states = 0;
+        // The times at or beyond `upper`, which `new_pending` takes whole; see the `moments` field.
+        let mut withheld: Vec<B1::Time> = Vec::new();
+        let mut live = 0usize;
         let (mut is, mut os, mut ns) = (0usize, 0usize, 0usize);
         for &key in &self.window.keys {
             while is < p_in.len() && p_in[is].0.0 < key { is += 1; }
@@ -433,7 +439,7 @@ where
             let n1 = ns;
 
             self.moments.clear();
-            self.pended.clear();
+            withheld.clear();
             {
                 let pending = self.pending.get(&key).map(|p| &p[..]).unwrap_or(&[]);
                 let seed_times = seeds[n0..n1].iter().map(|(_, t)| t.clone());
@@ -442,22 +448,22 @@ where
                     KeyView { p_in: &p_in[..], i0, i1, pending },
                     seed_times, out_times, self.upper,
                     &mut self.discover_scratch,
-                    &mut self.moments, &mut self.pended,
+                    &mut self.moments, &mut withheld,
                 );
             }
-            if !self.pended.is_empty() {
-                self.new_pending.insert(key, std::mem::take(&mut self.pended));
+            if !withheld.is_empty() {
+                self.new_pending.insert(key, std::mem::take(&mut withheld));
             }
             if self.moments.is_empty() {
                 continue;
             }
 
-            // Reload slot `n_states` in place (grow the buffer by one only when a window is wider
-            // than any before). `append` moves the discovered moments in without copy or realloc.
-            if self.n_states == self.states.len() {
+            // Reload slot `live` in place (grow the buffer by one only when a window is wider than
+            // any before). `append` moves the discovered moments in without copy or realloc.
+            if live == self.states.len() {
                 self.states.push(KeyState::empty());
             }
-            let st = &mut self.states[self.n_states];
+            let st = &mut self.states[live];
             st.key = key;
             st.cursor = 0;
             st.produced.clear();
@@ -471,8 +477,9 @@ where
             }
             st.in_replay.load_iter((i0..i1).map(|i| (p_in[i].0.1, p_in[i].1.clone(), p_in[i].2.clone())), st.meets.first());
             st.out_replay.load_iter((o0..o1).map(|o| (p_out[o].0.1, p_out[o].1.clone(), p_out[o].2.clone())), st.meets.first());
-            self.n_states += 1;
+            live += 1;
         }
+        live
     }
 
     /// Phase 2 (application): walk all keys' moments in ROUNDS, accumulating output updates into
@@ -484,14 +491,14 @@ where
     /// call count at O(max moments over keys), not O(sum of moments), with peak materialization one
     /// moment deep per key. `produced` is meet-collapsed each round, exactly like the reference —
     /// bounded, not the O(times × values) delta history.
-    fn apply(&mut self) {
+    fn apply(&mut self, live: usize) {
         for deltas in self.tile_deltas.iter_mut() { deltas.clear(); }
 
         loop {
             self.round.clear();
             self.active.clear();
             let mut advanced = false;
-            for (si, st) in self.states[..self.n_states].iter_mut().enumerate() {
+            for (si, st) in self.states[..live].iter_mut().enumerate() {
                 if st.cursor >= st.moments.len() {
                     continue;
                 }
@@ -515,21 +522,21 @@ where
                     }
                 }
                 crate::consolidation::consolidate(&mut self.in_accum);
-                self.cur_out.clear();
+                self.out_accum.clear();
                 for ((vid, et), d) in st.out_replay.buffer().iter().chain(st.produced.iter()) {
                     if et.less_equal(&t) {
-                        self.cur_out.push((*vid, d.clone()));
+                        self.out_accum.push((*vid, d.clone()));
                     }
                 }
-                crate::consolidation::consolidate(&mut self.cur_out);
+                crate::consolidation::consolidate(&mut self.out_accum);
 
-                if self.in_accum.is_empty() && self.cur_out.is_empty() {
+                if self.in_accum.is_empty() && self.out_accum.is_empty() {
                     continue;
                 }
                 self.round.keys.push(st.key);
                 self.round.input.append(&mut self.in_accum);
                 self.round.in_ends.push(self.round.input.len());
-                self.round.output.append(&mut self.cur_out);
+                self.round.output.append(&mut self.out_accum);
                 self.round.out_ends.push(self.round.output.len());
                 self.active.push((si, t));
             }
@@ -550,7 +557,7 @@ where
             for (bi, (si, t)) in self.active.iter().enumerate() {
                 let cend = self.corrections.ends[bi];
                 if cstart != cend {
-                    let idx = self.held.iter().rposition(|h| h.less_equal(t)).expect("no held capability <= active time");
+                    let idx = self.tiles.iter().rposition(|(h, _)| h.less_equal(t)).expect("no held capability <= active time");
                     for (vid, d) in &self.corrections.updates[cstart..cend] {
                         self.states[*si].produced.push(((*vid, t.clone()), d.clone()));
                         self.tile_deltas[idx].push(((self.states[*si].key, *vid), t.clone(), d.clone()));
@@ -567,7 +574,7 @@ where
             if deltas.is_empty() {
                 continue;
             }
-            if let Some(tile) = self.tile_of[held_index] {
+            if let Some(tile) = self.tiles[held_index].1 {
                 crate::consolidation::consolidate_updates(deltas);
                 self.backend.emit(tile, &deltas[..]);
             }