From fa0be326dd46eeb693d8ecc72209b36adc3f66cf Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Mon, 10 Aug 2026 11:06:02 -0400
Subject: [PATCH 1/3] Retire seed_times; present the novel run apart from the
 history
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`seed_times` existed because `next_window` merged the accumulated input and the
novel delta into one bridge, which loses which records were novel.
A novel record that nets to zero against a history record compaction has
advanced onto its time vanishes, and its interesting time with it, so the
seeds had to be derived from a second, separate read of the delta.

The window now carries the two runs apart, mirroring the three cursors of the
conventional reduce, which has always held them apart for exactly this reason.
The seeds are the novel run's own times, read where they already are.
The two runs combine only in the per-moment accumulation, never as
presentations.

`next_window` also takes `from: &mut Option<u64>` — a key hash rather than an
index into a harness-built list — on the same contract as the join tactic's
`advance`: strictly increase it or report the key space exhausted, and report a
key entirely within the window that first mentions it.
The window is caller-owned and refilled, so its three runs keep their
allocations.

Neither side now scans the whole key space.
The harness contributes the keys it knows must be revisited, and the backend
adds the ones its novel batches touch, which it discovers in the scan the
presentation needs anyway.

That change to `changed` is where the time went, and it was a real redundancy
rather than a constant factor.
`changed` was every pending key; it is now only the keys with an interesting
time DUE in this interval.
A key whose times all lie at or beyond `upper` was being presented and walked
through determination on every retire, only to produce no moments — and its
history and output rows were gathered into the presentation to do it.
The conventional tactic never did this: it filters to the due times and takes
the no-work branch when there are none.
The closure is not lost by skipping, because a join of two not-yet-due times is
itself not yet due, and is synthesized when they become due.

Carrying the times of unvisited keys is now explicit, since a key that is not
`changed` is never presented and would otherwise be forgotten.
The split against `upper` happens up front: due times drive determination,
carried times seed the next round's pending set directly, and a key with both
unions them.

Measured on the corgi backend, both binaries built in the same target directory:

  scc n=500  e=1000 batch=100 rounds=60     4.07s -> 1.92s
  scc n=1000 e=2000 batch=100 rounds=100   12.33s -> 4.95s
  scc n=2000 e=4000 batch=100 rounds=60    18.83s -> 5.72s
  reach n=20000 e=40000 batch=200 rounds=100  flat

The gap grows with n, as it should if the old cost tracked the whole pending
set and the new one tracks the due set.
Program output is byte-identical across scc, reach, tour, adt, binders and
unnest, on both the corgi and vec backends.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/src/operators/common.rs |  18 +-
 .../src/operators/int_proxy/reduce.rs         | 190 +++++++++++++-----
 interactive/src/corgi/reduce.rs               | 138 +++++++------
 3 files changed, 234 insertions(+), 112 deletions(-)

diff --git a/differential-dataflow/src/operators/common.rs b/differential-dataflow/src/operators/common.rs
index aea8e598b..069846f12 100644
--- a/differential-dataflow/src/operators/common.rs
+++ b/differential-dataflow/src/operators/common.rs
@@ -179,9 +179,10 @@ pub fn tile_descriptions<T: Timestamp + Lattice>(
     (tile_descs, tile_held, tile_of)
 }
 
-/// A one-key view into an input presentation: the read-only arguments [`discover_times`] needs
-/// about a single key — its slice `[i0, i1)` of the merged `(id, time, diff)` run `p_in` and
-/// the carried `pending` times.
+/// A one-key view into the ACCUMULATED input presentation: the read-only arguments
+/// [`discover_times`] needs about a single key — its slice `[i0, i1)` of the `(id, time, diff)` run
+/// `p_in` and the carried `pending` times. The novel run is not here; it arrives as `seed_times`,
+/// and the two are deliberately never merged (see the note on `seed_times` below).
 pub struct KeyView<'a, T, RIn> {
     /// The presented `((key_hash, value_id), time, diff)` run the key's records live in.
     pub p_in: &'a [((u64, u64), T, RIn)],
@@ -236,15 +237,18 @@ impl<T: Timestamp + Lattice, RIn: Semigroup + Clone> Default for DiscoverScratch
 
 /// Determines the times in `[lower, upper)` at which a key's reduction must be re-evaluated
 /// (`moments`), and the times at or beyond `upper` to carry into the next invocation
-/// (`pended`). Replays the key's `seed_times` and `pending` times in ascending order, marking
+/// (`pended`). Replays the key's `seed_times` (the novel run) and `pending` times in ascending
+/// order, marking
 /// those that carry updates and closing the set under joins with the input and output
 /// histories' times and with each other. No input collection is materialized, so peak memory
 /// is O(times); buffers are advanced by the meet of the times still to come, keeping a key
 /// with many distinct times linear rather than quadratic.
 ///
-/// `seed_times` must be the novel batch's own time support for this key. Seeding from a
-/// consolidated view is unsound: compaction may advance a history record onto a novel time,
-/// where consolidation cancels the novel update and its interesting time is missed.
+/// `seed_times` must be the novel batch's own time support for this key, and `key.p_in` the
+/// accumulated input WITHOUT it. Seeding from a view of the two consolidated together is unsound:
+/// compaction may advance a history record onto a novel time, where consolidation cancels the novel
+/// update and its interesting time is missed. This is why the caller keeps the two runs apart and
+/// combines them only when accumulating.
 #[allow(clippy::too_many_arguments)]
 pub fn discover_times<T, RIn>(
     key: KeyView<'_, T, RIn>,
diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index ee206a904..ae9e90253 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};
@@ -30,34 +30,51 @@ pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>>
     pub lower: AntichainRef<'a, B1::Time>,
 }
 
-/// One window of a retire's changed keys: a bounded, hash-contiguous snip the backend sizes.
+/// One window of the key space: the presentations a bounded, hash-contiguous snip needs.
 ///
-/// The window has the input (old and new) and output histories, restricted to the window's keys.
+/// The three runs mirror the three cursors of the conventional reduce, and are held apart for the
+/// same reason it holds them apart: `novel` is the *seed* of interesting times, and consolidating it
+/// into `history` would lose which records were novel. A novel record that nets to zero against a
+/// history record compaction has advanced onto its time would vanish, and its interesting time with
+/// it. The runs are combined only at the accumulation, never as presentations.
+///
+/// Owned by the harness and refilled by [`ProxyReduceBackend::next_window`], so the allocations
+/// survive the window loop.
 pub struct ReduceWindow<T, RIn, ROut> {
-    /// 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.
+    /// Accumulated input preceding the retire's interval, sorted & consolidated by
+    /// `((key_hash, value_id), time)`.
+    pub history: ProxyBridge<T, RIn>,
+    /// The retire's novel input delta, same ordering. Its times per key are the interesting-time
+    /// seeds, so it must be the batch's own time support — not a view netted against `history`.
+    pub novel: ProxyBridge<T, RIn>,
+    /// Accumulated output preceding the retire's interval, same ordering.
     pub output: ProxyBridge<T, ROut>,
 }
 
+impl<T, RIn, ROut> Default for ReduceWindow<T, RIn, ROut> {
+    fn default() -> Self { ReduceWindow { history: Vec::new(), novel: Vec::new(), output: Vec::new() } }
+}
+
+impl<T, RIn, ROut> ReduceWindow<T, RIn, ROut> {
+    /// Empty the runs, keeping their allocations.
+    pub fn clear(&mut self) {
+        self.history.clear();
+        self.novel.clear();
+        self.output.clear();
+    }
+}
+
 /// 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 protocol for each round of invocation is
+/// `begin [ next_window reduce_corrections* emit ]* finish`,
+/// where the window loop runs until `next_window` reports the key space exhausted.
 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;
 
-    /// 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.
@@ -65,12 +82,29 @@ 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.
+    /// Present the next window of the key space, and advance `from` past it.
+    ///
+    /// On entry `*from` is the inclusive lower bound on key hashes still to be covered. The backend
+    /// chooses the window's exclusive upper bound and writes it back, or writes `None` to report the
+    /// key space exhausted — matching [`ProxyJoinBackend::advance`](super::ProxyJoinBackend::advance).
+    /// A call that leaves `*from` unchanged does not terminate, so it is a contract violation.
     ///
-    /// 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>>;
+    /// The window must present, for every key hash in `[from_before, from_after)` that either
+    /// carries an update in the instance's novel batches or appears in `changed`, that key's novel
+    /// updates, its accumulated input, and its accumulated output. A key must be reported entirely
+    /// within the window that first mentions it: splitting one across windows drops the interaction
+    /// between the halves. `changed` is ascending; the harness reads no key outside the window's
+    /// range, so a backend that keeps its own key order need not consult the whole space.
+    ///
+    /// The size of the window is up to the backend: large enough to amortize the crossings, small
+    /// enough that the three presentations are affordable, as all are live at once.
+    fn next_window(
+        &mut self,
+        instance: &ReduceInstance<'_, B1, B2>,
+        changed: &[u64],
+        from: &mut Option<u64>,
+        window: &mut ReduceWindow<B1::Time, Self::RIn, Self::ROut>,
+    );
 
     /// A wave of input-output reconciliation, in which the backend supplies necessary edits.
     ///
@@ -136,15 +170,21 @@ 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());
+        // Split the carried interesting times against `upper`. A time below it is DUE: its key must
+        // be re-evaluated this retire, so the key is `changed`. A time at or beyond it is carried
+        // forward untouched, seeding `new_pending` — and this is the only thing that carries it,
+        // because a key whose times are all beyond `upper` is not `changed`, is never presented in a
+        // window, and would otherwise be silently forgotten.
+        let mut due: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
+        let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
+        for (key, times) in self.pending.iter() {
+            let (carried, ready): (Vec<_>, Vec<_>) = times.iter().cloned().partition(|t| upper.less_equal(t));
+            if !ready.is_empty() { due.insert(*key, ready); }
+            if !carried.is_empty() { new_pending.insert(*key, carried); }
         }
-        let changed: Vec<u64> = changed.into_iter().collect();
+        // The keys the harness knows must be revisited. The backend adds those its novel batches
+        // touch, which it discovers while reading them; neither side scans the whole key space.
+        let changed: Vec<u64> = due.keys().copied().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.
@@ -152,10 +192,13 @@ 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 cursor = 0usize;
-        let mut ns = 0usize;
+        // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
+        // once the backend reports the space covered.
+        let mut from = Some(0u64);
+        // Monotone cursor into `changed`, consumed as the windows pass over it.
+        let mut cs = 0usize;
+        let mut window: ReduceWindow<B1::Time, Bk::RIn, Bk::ROut> = ReduceWindow::default();
+        let mut window_keys: Vec<u64> = Vec::new();
 
         // 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.
@@ -173,11 +216,49 @@ where
         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;
+        while from.is_some() {
+            let before = from;
+            window.clear();
+            self.backend.next_window(&instance, &changed, &mut from, &mut window);
+            let p_in = &window.history;
+            let p_nv = &window.novel;
             let p_out = &window.output;
-            super::debug_assert_sorted_bridge(p_in, "next_window.input");
+            super::debug_assert_sorted_bridge(p_in, "next_window.history");
+            super::debug_assert_sorted_bridge(p_nv, "next_window.novel");
             super::debug_assert_sorted_bridge(p_out, "next_window.output");
+            // Without progress the window loop would never retire, so this guards liveness as well
+            // as contract; the range check catches a key reported outside the window that owns it,
+            // which would silently drop the interaction between its halves.
+            debug_assert!(
+                from.is_none() || from > before,
+                "next_window must either advance `from` or report the key space exhausted",
+            );
+            debug_assert!(
+                {
+                    let mut keys = p_in.iter().chain(p_nv.iter()).map(|r| r.0.0).chain(p_out.iter().map(|r| r.0.0));
+                    keys.all(|k| before.is_none_or(|b| b <= k) && from.is_none_or(|f| k < f))
+                },
+                "next_window must report a key hash entirely within the window that first mentions it",
+            );
+
+            // The window's keys: those its novel updates touch, merged with the `changed` keys it
+            // covers. A `changed` key can appear in no presentation at all — a due interesting time
+            // is reason enough to re-evaluate it — so the key list cannot be read off the bridges.
+            window_keys.clear();
+            let mut nk = 0usize;
+            loop {
+                let from_novel = p_nv.get(nk).map(|r| r.0.0);
+                let from_changed = changed.get(cs).copied().filter(|k| from.is_none_or(|f| *k < f));
+                let key = match (from_novel, from_changed) {
+                    (Some(a), Some(b)) => std::cmp::min(a, b),
+                    (Some(a), None) => a,
+                    (None, Some(b)) => b,
+                    (None, None) => break,
+                };
+                while nk < p_nv.len() && p_nv[nk].0.0 == key { nk += 1; }
+                if changed.get(cs) == Some(&key) { cs += 1; }
+                window_keys.push(key);
+            }
 
             for deltas in tile_deltas.iter_mut() { deltas.clear(); }
 
@@ -188,26 +269,30 @@ where
             // `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 {
+            let (mut is, mut ns, mut os) = (0usize, 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 ns < p_nv.len() && p_nv[ns].0.0 < key { ns += 1; }
+                let n0 = ns;
+                while ns < p_nv.len() && p_nv[ns].0.0 == key { ns += 1; }
+                let n1 = ns;
                 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());
+                    // Only the DUE times seed determination; the carried ones are already in
+                    // `new_pending` and must not be re-derived (nor re-pended) here.
+                    let pending = due.get(&key).map(|p| &p[..]).unwrap_or(&[]);
+                    // The seeds are the novel run's own time support for this key — the whole point
+                    // of presenting it apart from `history`.
+                    let seed_times = (n0..n1).map(|n| p_nv[n].1.clone());
                     let out_times = (o0..o1).map(|o| p_out[o].1.clone());
                     discover_times(
                         KeyView { p_in: &p_in[..], i0, i1, pending },
@@ -217,7 +302,10 @@ where
                     );
                 }
                 if !pended_scratch.is_empty() {
-                    new_pending.insert(key, std::mem::take(&mut pended_scratch));
+                    // The key may already carry times beyond `upper` from the split above; union.
+                    let entry = new_pending.entry(key).or_default();
+                    entry.append(&mut pended_scratch);
+                    crate::operators::reduce::sort_dedup(entry);
                 }
                 if moments_scratch.is_empty() {
                     continue;
@@ -241,6 +329,7 @@ where
                     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.nv_replay.load_iter((n0..n1).map(|n| (p_nv[n].0.1, p_nv[n].1.clone(), p_nv[n].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;
             }
@@ -269,16 +358,19 @@ where
                     st.cursor += 1;
                     let t = st.moments[j].clone();
                     st.in_replay.step_through(&t);
+                    st.nv_replay.step_through(&t);
                     st.out_replay.step_through(&t);
                     st.in_replay.advance_buffer_by(&st.meets[j]);
+                    st.nv_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);
 
+                    // The two input runs meet HERE, at the accumulation — never as presentations.
                     in_accum.clear();
-                    for ((vid, et), d) in st.in_replay.buffer().iter() {
+                    for ((vid, et), d) in st.in_replay.buffer().iter().chain(st.nv_replay.buffer().iter()) {
                         if et.less_equal(&t) {
                             in_accum.push((*vid, d.clone()));
                         }
@@ -351,7 +443,8 @@ where
 }
 
 /// Per-key application state for [`ProxyReduceTactic`]'s round-batched walk: the key's ordered
-/// interesting `moments` and their suffix `meets`, its input and output replays (meet-collapsed),
+/// interesting `moments` and their suffix `meets`, its accumulated-input, novel-input, and output
+/// replays (meet-collapsed; the two input replays combine only in the per-moment accumulation),
 /// the corrections `produced` this round so far, and a `cursor` into `moments`. Held for all of a
 /// window's keys at once so each round's crossing batches across keys — a key's own moments stay
 /// sequential (each sees its earlier corrections via `produced`), but distinct keys are independent.
@@ -360,6 +453,7 @@ struct KeyState<T, RIn, ROut> {
     moments: Vec<T>,
     meets: Vec<T>,
     in_replay: IdHistory<T, RIn>,
+    nv_replay: IdHistory<T, RIn>,
     out_replay: IdHistory<T, ROut>,
     produced: Vec<((u64, T), ROut)>,
     cursor: usize,
@@ -370,7 +464,7 @@ impl<T: Timestamp + Lattice, RIn: Semigroup, ROut: Semigroup> KeyState<T, RIn, R
     /// vector holds these across windows and reloads them in place, so a key's buffers are allocated
     /// once (per slot) and reused — never dropped per key (which was ~18% of load in `free`).
     fn empty() -> Self {
-        KeyState { key: 0, moments: Vec::new(), meets: Vec::new(), in_replay: IdHistory::new(), out_replay: IdHistory::new(), produced: Vec::new(), cursor: 0 }
+        KeyState { key: 0, moments: Vec::new(), meets: Vec::new(), in_replay: IdHistory::new(), nv_replay: IdHistory::new(), out_replay: IdHistory::new(), produced: Vec::new(), cursor: 0 }
     }
 }
 
diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 68ceacd21..1accf08b2 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -288,6 +288,33 @@ impl<T> CorgiReduceBackend<T>
 where
     T: ColTime + Ord,
 {
+    /// Present one input run — the accumulated history or the novel delta — restricted to `keys`.
+    ///
+    /// Fills `bridge`, registers the run's representative keys, and extends the shared value pool
+    /// (`blocks`/`len`, concatenated into `in_vals`) with its values, so `in_index` resolves a value
+    /// id from EITHER run to a row. The two runs stay apart as presentations and meet only in the
+    /// tactic's accumulation; the pool is shared because a value id means the same thing in both.
+    fn present_input(
+        &mut self,
+        chunks: &[&CorgiChunk<T, Diff>],
+        keys: &[u64],
+        blocks: &mut Vec<CValue>,
+        len: &mut usize,
+        bridge: &mut ProxyBridge<T, Diff>,
+    ) {
+        let (p_keys, p_vals, khs, times, diffs) = collect_present(chunks, keys);
+        if khs.is_empty() {
+            return;
+        }
+        let vids = ids(&p_vals);
+        for (row, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(*len + row); }
+        *len += p_vals.len();
+        blocks.push(p_vals);
+        self.register_keys(p_keys, &khs);
+        bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i])));
+        consolidate_updates(bridge);
+    }
+
     /// The one value crossing for a retire: every `(key, time)` bracket at once. Builds the output
     /// value COLUMN directly per reducer, registers it (id → row) into the val pool, and returns the
     /// proxy `(value_id, diff)` deltas with per-bracket ends. `input[k] = (rep index into the input
@@ -430,22 +457,6 @@ where
     type RIn = Diff;
     type ROut = Diff;
 
-    fn seed_times(&self, instance: &ReduceInstance<'_, CBatch<T>, CBatch<T>>) -> Vec<(u64, T)> {
-        // The batch's raw (key_hash, time) support — hash the novel KEY columns only, one entry per
-        // record, sorted by key_hash. Seeds may over-derive (a non-changing seed yields a zero delta),
-        // so this superset of b.support suffices; `instance.lower` is not applied (see ReduceInstance).
-        let mut out: Vec<(u64, T)> = Vec::new();
-        for ch in chunks_of(instance.input_batches) {
-            let kh = ids(ch.keys());
-            let times = ch.times();
-            for (i, h) in kh.into_iter().enumerate() {
-                out.push((h, times.get(i)));
-            }
-        }
-        out.sort_by_key(|(k, _)| *k);
-        out
-    }
-
     fn begin(&mut self, tiles: &[Description<T>]) {
         // Open a tiled output session for this retire; reset the per-retire resolution pools.
         self.reset_pools();
@@ -453,57 +464,70 @@ where
         self.tile_rows = (0..tiles.len()).map(|_| (Vec::new(), Vec::new(), Vec::new(), Vec::new())).collect();
     }
 
-    fn next_window(&mut self, instance: &ReduceInstance<'_, CBatch<T>, CBatch<T>>, changed: &[u64], cursor: &mut usize) -> Option<ReduceWindow<T, Diff, Diff>> {
-        // Single window: present ALL remaining changed keys at once. This is NOT a deferred
-        // refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc
+    fn next_window(&mut self, instance: &ReduceInstance<'_, CBatch<T>, CBatch<T>>, changed: &[u64], from: &mut Option<u64>, window: &mut ReduceWindow<T, Diff, Diff>) {
+        // Single window: present the WHOLE key space at once, and report it covered. This is NOT a
+        // deferred refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc
         // (100 rounds x batch 100) cost 84.4s against 63.7s, a 33% regression, while peak RSS
         // fell only 356MB -> 340MB. Two reasons: the per-window, per-chunk seek setup is a
         // fixed cost that multiplies by the window count, and the presentation is not the
-        // memory peak in the first place (the trace is). `changed` is ascending, so
-        // `binary_search` is the changed-key filter.
-        if *cursor >= changed.len() {
-            return None;
+        // memory peak in the first place (the trace is).
+        if from.is_none() {
+            return;
         }
-        let keys: Vec<u64> = changed[*cursor..].to_vec();
-        *cursor = changed.len();
-        let present = |chunks: &[&CorgiChunk<T, Diff>]| collect_present(chunks, &keys);
-
-        // Input presentation: accumulated history ∪ novel delta, restricted to the window's keys.
-        // value_id = content hash of the value (equal values share an id → the tactic nets them);
-        // `in_index` resolves an id back to a representative in_vals row for `reduce_corrections`.
-        let mut in_chunks = chunks_of(instance.input_batches);
-        in_chunks.extend(chunks_of(instance.source_batches));
-        let (in_keys, in_vals, in_khs, in_times, in_diffs) = present(&in_chunks);
-        self.in_index = IdMap::default();
-        let input: ProxyBridge<T, Diff> = if in_khs.is_empty() {
+        *from = None;
+
+        // The window's keys: the hashes the novel batches touch, merged with the `changed` set the
+        // harness supplies. The novel hashes come from the scan the presentation needs anyway — the
+        // separate seeding pass this replaced read the delta a second time to derive them.
+        let novel_chunks = chunks_of(instance.input_batches);
+        let mut keys: Vec<u64> = Vec::new();
+        for ch in novel_chunks.iter() {
+            keys.extend(ids(ch.keys()));
+        }
+        keys.sort_unstable();
+        keys.dedup();
+        if !changed.is_empty() {
+            // Both sides ascend, so this is a merge.
+            let mut merged: Vec<u64> = Vec::with_capacity(keys.len() + changed.len());
+            let (mut a, mut b) = (0usize, 0usize);
+            while a < keys.len() || b < changed.len() {
+                let key = match (keys.get(a), changed.get(b)) {
+                    (Some(x), Some(y)) => *x.min(y),
+                    (Some(x), None) => *x,
+                    (None, Some(y)) => *y,
+                    (None, None) => unreachable!("loop condition ensures one is present"),
+                };
+                if keys.get(a) == Some(&key) { a += 1; }
+                if changed.get(b) == Some(&key) { b += 1; }
+                merged.push(key);
+            }
+            keys = merged;
+        }
+        if keys.is_empty() {
             self.in_vals = CValue::Unit(0);
-            Vec::new()
-        } else {
-            let vids = ids(&in_vals);
-            for (r, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(r); }
-            self.in_vals = in_vals;
-            self.register_keys(in_keys, &in_khs);
-            let mut b: ProxyBridge<T, Diff> =
-                (0..in_khs.len()).map(|i| ((in_khs[i], vids[i]), in_times[i].clone(), in_diffs[i])).collect();
-            consolidate_updates(&mut b);
-            b
-        };
+            self.in_index = IdMap::default();
+            return;
+        }
+
+        // The two input presentations, held apart: `history` is the accumulated input and `novel`
+        // the delta whose times seed the interesting times. `in_index` resolves a value id back to a
+        // representative row of `in_vals`, which spans BOTH runs, for `reduce_corrections`.
+        self.in_index = IdMap::default();
+        let mut in_blocks: Vec<CValue> = Vec::new();
+        let mut in_len = 0usize;
+        self.present_input(&chunks_of(instance.source_batches), &keys, &mut in_blocks, &mut in_len, &mut window.history);
+        self.present_input(&novel_chunks, &keys, &mut in_blocks, &mut in_len, &mut window.novel);
+        self.in_vals = concat_columns(&in_blocks);
 
         // Output-history presentation, same keys (register keys + values for correction resolution).
-        let (o_keys, o_vals, o_khs, o_times, o_diffs) = present(&chunks_of(instance.output_batches));
-        let output: ProxyBridge<T, Diff> = if o_khs.is_empty() {
-            Vec::new()
-        } else {
+        let (o_keys, o_vals, o_khs, o_times, o_diffs) = collect_present(&chunks_of(instance.output_batches), &keys);
+        if !o_khs.is_empty() {
             let vids = ids(&o_vals);
             self.register_keys(o_keys, &o_khs);
             self.register_vals(o_vals, &vids);
-            let mut b: ProxyBridge<T, Diff> =
-                (0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])).collect();
-            consolidate_updates(&mut b);
-            b
-        };
-
-        Some(ReduceWindow { keys, input, output })
+            window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])));
+            consolidate_updates(&mut window.output);
+        }
     }
 
     fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec<usize>) {

From b9606dfe523a94e8d85ef549953ff6f74da753c4 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Mon, 10 Aug 2026 12:04:26 -0400
Subject: [PATCH 2/3] Restore the no-work early return, correctly guarded
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Narrowing `changed` to the due keys made the old early return unsafe, so it was
dropped along with `seed_times`.
It read

    if changed.is_empty() { self.pending.clear(); return (Vec::new(), Antichain::new()) }

which was sound only because `changed` then held every pending key, so an empty
`changed` implied an empty pending set.
It no longer does: times at or beyond `upper` remain when nothing is due, and
clearing them while returning an empty frontier releases the capabilities that
were holding them.

The guarded form also requires the novel batches to be empty, and returns the
frontier over the untouched pending set rather than an empty one.
Without it a retire with nothing to do still opened a tiled output session and
shipped empty batches.

Also measured and rejected: advancing the accumulated input and output
presentations by `instance.lower` in the corgi backend, as its join backend
already does in `stage_runs`.
The reasoning is sound — every moment is at or above `lower`, so advancing
preserves each record's downset while letting the consolidation net more — but
it costs more than it saves, because the driver already sets logical compaction
to `upper` on both traces each activation, so the times largely arrive advanced
and the pass is work for nothing.
On scc (n=1000, corgi) it was a stable 5.09-5.10s against 4.85-4.89s, about 4%
worse, and flat everywhere else.
`ReduceInstance::lower` therefore remains unread by the only backend.

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

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index ae9e90253..8dffb83a4 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -186,6 +186,21 @@ where
         // touch, which it discovers while reading them; neither side scans the whole key space.
         let changed: Vec<u64> = due.keys().copied().collect();
 
+        // Nothing due and nothing novel: no time in the interval can be interesting, so there is no
+        // work and no output. Return the frontier bounding the times still withheld — NOT an empty
+        // one. This is exactly where a due-only `changed` differs from the whole pending set: times
+        // beyond `upper` can remain when nothing is due, and releasing their capabilities would
+        // strand them (see the frontier clause of the `ReduceTactic::retire` contract).
+        if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) {
+            let mut frontier = Antichain::new();
+            for times in self.pending.values() {
+                for time in times {
+                    frontier.insert_ref(time);
+                }
+            }
+            return (Vec::new(), frontier);
+        }
+
         // 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();

From 7ba9477c6df742f7fbb8495338aa525c146b9a32 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Mon, 10 Aug 2026 16:08:57 -0400
Subject: [PATCH 3/3] Improve docs; remove unneeded logic

---
 .../src/operators/int_proxy/reduce.rs         | 66 +++++++------------
 1 file changed, 24 insertions(+), 42 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index 8dffb83a4..27668ae49 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -32,14 +32,11 @@ pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>>
 
 /// One window of the key space: the presentations a bounded, hash-contiguous snip needs.
 ///
-/// The three runs mirror the three cursors of the conventional reduce, and are held apart for the
-/// same reason it holds them apart: `novel` is the *seed* of interesting times, and consolidating it
-/// into `history` would lose which records were novel. A novel record that nets to zero against a
-/// history record compaction has advanced onto its time would vanish, and its interesting time with
-/// it. The runs are combined only at the accumulation, never as presentations.
+/// The two input runs are held apart because the novel and prior data have different roles:
+/// the novel data seed interesting times, and the prior data may benefit from historical rollup.
+/// Their updates are combined after we are able to see the distinction between the two.
 ///
-/// Owned by the harness and refilled by [`ProxyReduceBackend::next_window`], so the allocations
-/// survive the window loop.
+/// Owned by the harness and refilled by [`ProxyReduceBackend::next_window`].
 pub struct ReduceWindow<T, RIn, ROut> {
     /// Accumulated input preceding the retire's interval, sorted & consolidated by
     /// `((key_hash, value_id), time)`.
@@ -56,7 +53,7 @@ impl<T, RIn, ROut> Default for ReduceWindow<T, RIn, ROut> {
 }
 
 impl<T, RIn, ROut> ReduceWindow<T, RIn, ROut> {
-    /// Empty the runs, keeping their allocations.
+    /// Clear the three proxy bridges.
     pub fn clear(&mut self) {
         self.history.clear();
         self.novel.clear();
@@ -84,10 +81,9 @@ pub trait ProxyReduceBackend<B1: BatchReader, B2: BatchReader<Time = B1::Time>>
 
     /// Present the next window of the key space, and advance `from` past it.
     ///
-    /// On entry `*from` is the inclusive lower bound on key hashes still to be covered. The backend
+    /// On entry `from` is the inclusive lower bound on key hashes still to be covered. The backend
     /// chooses the window's exclusive upper bound and writes it back, or writes `None` to report the
-    /// key space exhausted — matching [`ProxyJoinBackend::advance`](super::ProxyJoinBackend::advance).
-    /// A call that leaves `*from` unchanged does not terminate, so it is a contract violation.
+    /// key space exhausted. An implementor must advance `from`, as it is guaranteed to be non-`None`.
     ///
     /// The window must present, for every key hash in `[from_before, from_after)` that either
     /// carries an update in the instance's novel batches or appears in `changed`, that key's novel
@@ -170,11 +166,9 @@ where
             lower: lower.borrow(),
         };
 
-        // Split the carried interesting times against `upper`. A time below it is DUE: its key must
-        // be re-evaluated this retire, so the key is `changed`. A time at or beyond it is carried
-        // forward untouched, seeding `new_pending` — and this is the only thing that carries it,
-        // because a key whose times are all beyond `upper` is not `changed`, is never presented in a
-        // window, and would otherwise be silently forgotten.
+        // Split the carried interesting times against `upper`.
+        // A time below it is DUE: its key must be re-evaluated this retire, so the key is `changed`.
+        // A time at or beyond it is carried forward untouched, seeding `new_pending`.
         let mut due: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
         let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
         for (key, times) in self.pending.iter() {
@@ -210,10 +204,7 @@ where
         // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
         // once the backend reports the space covered.
         let mut from = Some(0u64);
-        // Monotone cursor into `changed`, consumed as the windows pass over it.
-        let mut cs = 0usize;
         let mut window: ReduceWindow<B1::Time, Bk::RIn, Bk::ROut> = ReduceWindow::default();
-        let mut window_keys: Vec<u64> = Vec::new();
 
         // 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.
@@ -256,25 +247,6 @@ where
                 "next_window must report a key hash entirely within the window that first mentions it",
             );
 
-            // The window's keys: those its novel updates touch, merged with the `changed` keys it
-            // covers. A `changed` key can appear in no presentation at all — a due interesting time
-            // is reason enough to re-evaluate it — so the key list cannot be read off the bridges.
-            window_keys.clear();
-            let mut nk = 0usize;
-            loop {
-                let from_novel = p_nv.get(nk).map(|r| r.0.0);
-                let from_changed = changed.get(cs).copied().filter(|k| from.is_none_or(|f| *k < f));
-                let key = match (from_novel, from_changed) {
-                    (Some(a), Some(b)) => std::cmp::min(a, b),
-                    (Some(a), None) => a,
-                    (None, Some(b)) => b,
-                    (None, None) => break,
-                };
-                while nk < p_nv.len() && p_nv[nk].0.0 == key { nk += 1; }
-                if changed.get(cs) == Some(&key) { cs += 1; }
-                window_keys.push(key);
-            }
-
             for deltas in tile_deltas.iter_mut() { deltas.clear(); }
 
             // Phase 1 (determination): for every key in the window, discover its interesting times
@@ -283,18 +255,28 @@ where
             // `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).
+            // The window's keys are the hashes its presentations mention: the least of the three
+            // heads, each iteration, until all three are drained. A `changed` key that appears in
+            // none of them has no records at all, so its reduction has nothing to read and nothing
+            // to retract — the moment its due time would raise reaches the evaluation gate with an
+            // empty input and an empty output, and produces nothing. Skipping it is exactly what
+            // visiting it would do. (`changed` is still the backend's instruction about which keys
+            // to present; it is just not a source of keys here.)
             let mut n_states = 0usize;
             let (mut is, mut ns, mut os) = (0usize, 0usize, 0usize);
-            for &key in &window_keys {
-                while is < p_in.len() && p_in[is].0.0 < key { is += 1; }
+            loop {
+                // Mapped to hashes before the min: the three runs differ in their diff type.
+                let Some(key) = [
+                    p_in.get(is).map(|record| record.0.0),
+                    p_nv.get(ns).map(|record| record.0.0),
+                    p_out.get(os).map(|record| record.0.0),
+                ].into_iter().flatten().min() else { break };
                 let i0 = is;
                 while is < p_in.len() && p_in[is].0.0 == key { is += 1; }
                 let i1 = is;
-                while ns < p_nv.len() && p_nv[ns].0.0 < key { ns += 1; }
                 let n0 = ns;
                 while ns < p_nv.len() && p_nv[ns].0.0 == key { ns += 1; }
                 let n1 = ns;
-                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;