From 5de4f02b6752b41037d63f2c142093dfea40a804 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 14:47:19 -0400
Subject: [PATCH 01/10] reduce: add TimeReplay, a time-only non-destructive
 ValueHistory walk

TimeReplay mirrors the time-facing half of HistoryReplay (time / meet / step /
step_while_time_is / advance_buffer_by / buffer) but carries only *times* and walks
a shrinking view of the built history rather than popping it, so the underlying
history stays intact for a later value walk. advance_buffer_by compacts by join +
dedup -- the time-only analogue of consolidation -- which keeps a join-closure over
the times an antichain, hence non-quadratic. Adds ValueHistory::{walk, replay_times}.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 differential-dataflow/src/operators/mod.rs | 57 ++++++++++++++++++++++
 1 file changed, 57 insertions(+)

diff --git a/differential-dataflow/src/operators/mod.rs b/differential-dataflow/src/operators/mod.rs
index 4260c2b95..746781daa 100644
--- a/differential-dataflow/src/operators/mod.rs
+++ b/differential-dataflow/src/operators/mod.rs
@@ -91,6 +91,7 @@ struct ValueHistory<V, T, D> {
     edits: EditList<V, T, D>,
     history: Vec<(T, T, usize, usize)>,     // (time, meet, value_index, edit_offset)
     buffer: Vec<((V, T), D)>,               // where we accumulate / collapse updates.
+    time_buffer: Vec<T>,                    // like `buffer`, but times only, for `TimeReplay`.
 }
 
 impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> ValueHistory<V, T, D> {
@@ -99,12 +100,14 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
             edits: EditList::new(),
             history: Vec::new(),
             buffer: Vec::new(),
+            time_buffer: Vec::new(),
         }
     }
     fn clear(&mut self) {
         self.edits.clear();
         self.history.clear();
         self.buffer.clear();
+        self.time_buffer.clear();
     }
 
     /// Loads and replays a specified key.
@@ -125,6 +128,23 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
         self.replay()
     }
 
+    /// Wraps the already-built, sorted `history` for a fresh walk, WITHOUT rebuilding or re-sorting
+    /// it. Valid whenever `history` is intact — e.g. after a `replay_key` whose returned replay was
+    /// only read through [`HistoryReplay::times`] (which does not step it). Used by reduce's
+    /// `reference` tactic, whose determination reads times and whose application then walks values.
+    fn walk<'history>(&'history mut self) -> HistoryReplay<'history, V, T, D> {
+        self.buffer.clear();
+        HistoryReplay { replay: self }
+    }
+
+    /// A time-only, non-destructive walk over the already-built `history` (see [`TimeReplay`]). It
+    /// reads times and their precomputed meets and accumulates only *times*, leaving `history`
+    /// intact for a later [`walk`](Self::walk) over values.
+    fn replay_times<'history>(&'history mut self) -> TimeReplay<'history, T> {
+        self.time_buffer.clear();
+        TimeReplay { history: &self.history[..], buffer: &mut self.time_buffer }
+    }
+
     /// Organizes history based on current contents of edits.
     fn replay<'history>(&'history mut self) -> HistoryReplay<'history, V, T, D> {
 
@@ -181,3 +201,40 @@ impl<'history, V: Copy + Ord, T: Ord + Clone + Lattice, D: Clone + crate::differ
     }
     fn is_done(&self) -> bool { self.replay.history.is_empty() }
 }
+
+/// A time-only, non-destructive walk over an already-built [`ValueHistory`] history.
+///
+/// It mirrors the time-facing half of [`HistoryReplay`] — `time`/`meet`/`step`/`step_while_time_is`
+/// /`advance_buffer_by`/`buffer` — but it carries only *times*, and it walks a shrinking *view* of
+/// `history` rather than popping it, so the underlying history stays intact for a later value walk.
+/// `advance_buffer_by` compacts the accumulated times by joining with `meet` and deduplicating (the
+/// time-only analogue of consolidation), which is what keeps a join-closure over these times an
+/// antichain — and hence non-quadratic.
+struct TimeReplay<'history, T> {
+    history: &'history [(T, T, usize, usize)],   // shrinking view; `last()` is the least time
+    buffer: &'history mut Vec<T>,                 // accumulated (and compacted) times seen so far
+}
+
+impl<'history, T: Ord + Clone + Lattice> TimeReplay<'history, T> {
+    fn time(&self) -> Option<&T> { self.history.last().map(|entry| &entry.0) }
+    fn meet(&self) -> Option<&T> { self.history.last().map(|entry| &entry.1) }
+    fn step(&mut self) {
+        let last = self.history.len() - 1;
+        self.buffer.push(self.history[last].0.clone());
+        self.history = &self.history[..last];
+    }
+    fn step_while_time_is(&mut self, time: &T) -> bool {
+        let mut found = false;
+        while self.time() == Some(time) {
+            found = true;
+            self.step();
+        }
+        found
+    }
+    fn advance_buffer_by(&mut self, meet: &T) {
+        for time in self.buffer.iter_mut() { time.join_assign(meet); }
+        self.buffer.sort();
+        self.buffer.dedup();
+    }
+    fn buffer(&self) -> &[T] { &self.buffer[..] }
+}

From c15b63c6a0440719a00c776ba5488b4267c31c53 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 14:47:19 -0400
Subject: [PATCH 02/10] reduce: add model-derived 'reference' tactic

Adds mod reference, a second ReduceTactic whose determination computes reduce's
interesting times directly as formal/Differential/Model.lean's Reached -- the
truncated join-closure over the stored input/output/seed times -- the non-quadratic
way: an ordered walk that keeps its live join-partners an antichain via
meet-compaction, over TimeReplay so it is time-only and leaves the histories intact
for the value walk. It runs at parity with the default cursor tactic, and exists as
a differential oracle for that tactic and an executable form of the model.

Also splits the cursor tactic's per-key compute into the same two phases
(determination, then evaluation); semantically identical, validated by the suite.

Entry points: Collection::{reduce_reference, reduce_named_reference},
Arranged::{reduce_abelian_reference, reduce_core_reference}, reduce_trace_reference.
Optional interesting-time counters in operators::reduce::metrics behind the
reduce-metrics feature.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 differential-dataflow/Cargo.toml              |   2 +
 differential-dataflow/src/collection.rs       |  21 +
 .../src/operators/arrange/arrangement.rs      |  33 +
 differential-dataflow/src/operators/reduce.rs | 600 ++++++++++++++++--
 4 files changed, 619 insertions(+), 37 deletions(-)

diff --git a/differential-dataflow/Cargo.toml b/differential-dataflow/Cargo.toml
index 3dbace3c7..ae9a2b0da 100644
--- a/differential-dataflow/Cargo.toml
+++ b/differential-dataflow/Cargo.toml
@@ -39,6 +39,8 @@ timely = {workspace = true}
 
 [features]
 default = ["timely/getopts"]
+# Enables the reduce interesting-time counters in `operators::reduce::metrics`.
+reduce-metrics = []
 
 [[bench]]
 name = "chunk_bench"
diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs
index ebf634ff2..6eed46760 100644
--- a/differential-dataflow/src/collection.rs
+++ b/differential-dataflow/src/collection.rs
@@ -757,6 +757,27 @@ pub mod vec {
                 .as_collection(|k,v| (k.clone(), v.clone()))
         }
 
+        /// As [`reduce`](Self::reduce), but driven by the model-derived reference tactic. Same result;
+        /// intended for differential testing against the default reduce.
+        pub fn reduce_reference<L, V2: crate::Data, R2: Ord+Abelian+'static>(self, logic: L) -> Collection<'scope, T, (K, V2), R2>
+        where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>)+'static {
+            self.reduce_named_reference("Reduce", logic)
+        }
+
+        /// As [`reduce_named`](Self::reduce_named), but driven by the model-derived reference tactic.
+        pub fn reduce_named_reference<L, V2: crate::Data, R2: Ord+Abelian+'static>(self, name: &str, logic: L) -> Collection<'scope, T, (K, V2), R2>
+        where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>)+'static {
+            use crate::trace::implementations::{ValBuilder, ValSpine};
+
+            self.arrange_by_key_named(&format!("Arrange: {}", name))
+                .reduce_abelian_reference::<_,ValBuilder<_,_,_,_>,ValSpine<K,V2,_,_>,_>(
+                    name,
+                    logic,
+                    |vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
+                )
+                .as_collection(|k,v| (k.clone(), v.clone()))
+        }
+
         /// Applies `reduce` to arranged data, and returns an arrangement of output data.
         ///
         /// This method is used by the more ergonomic `reduce`, `distinct`, and `count` methods, although
diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index f6c4e52a4..9445f95bc 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -294,6 +294,39 @@ impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
         use crate::operators::reduce::reduce_trace;
         reduce_trace::<_,Bu,_,_,_>(self, name, logic, push)
     }
+
+    /// As [`reduce_abelian`](Self::reduce_abelian), but using the model-derived reference tactic.
+    pub fn reduce_abelian_reference<L, Bu, Tr2, P>(self, name: &str, mut logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
+    where
+        Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
+        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time, Diff: Abelian>,
+        Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default> + 'static,
+        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
+    {
+        self.reduce_core_reference::<_,Bu,Tr2,_>(name, move |key, input, output, change| {
+            if !input.is_empty() {
+                logic(key, input, change);
+            }
+            change.extend(output.drain(..).map(|(x,mut d)| { d.negate(); (x, d) }));
+            crate::consolidation::consolidate(change);
+        }, push)
+    }
+
+    /// As [`reduce_core`](Self::reduce_core), but using the model-derived reference tactic.
+    pub fn reduce_core_reference<L, Bu, Tr2, P>(self, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
+    where
+        Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
+        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
+        Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default> + 'static,
+        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
+    {
+        use crate::operators::reduce::reduce_trace_reference;
+        reduce_trace_reference::<_,Bu,_,_,_>(self, name, logic, push)
+    }
 }
 
 
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index 08c99dba5..280a7b5e7 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -24,6 +24,25 @@ use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, Batc
 use crate::trace::cursor::cursor_list;
 use crate::trace::implementations::containers::BatchContainer;
 
+/// Optional interesting-time counters. They measure how many in-band interesting times each tactic
+/// evaluates, and hence how much the value-blind `reference` over-derives relative to the value-aware
+/// cursor (whose consolidation drops the zero-debt addresses the reference keeps). Off unless built
+/// with `--features reduce-metrics`; the increments compile to nothing otherwise.
+#[cfg(feature = "reduce-metrics")]
+pub mod metrics {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    /// In-band interesting times evaluated by the default `cursors::CursorTactic`.
+    pub static CURSOR: AtomicUsize = AtomicUsize::new(0);
+    /// In-band interesting times evaluated by the model-derived `reference::ReferenceTactic`.
+    pub static REFERENCE: AtomicUsize = AtomicUsize::new(0);
+    /// Reset both counters to zero.
+    pub fn reset() { CURSOR.store(0, Ordering::Relaxed); REFERENCE.store(0, Ordering::Relaxed); }
+    /// Read the cursor tactic's count.
+    pub fn cursor() -> usize { CURSOR.load(Ordering::Relaxed) }
+    /// Read the reference tactic's count.
+    pub fn reference() -> usize { REFERENCE.load(Ordering::Relaxed) }
+}
+
 /// A type that resolves a key-wise reduction over batches arriving on the input.
 ///
 /// Unlike join, reduce does not suspend: its output is at most linear in its input, so a single
@@ -73,6 +92,22 @@ where
     reduce_with_tactic(trace, name, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
 }
 
+/// As [`reduce_trace`], but driven by the model-derived [`reference::ReferenceTactic`] instead of the
+/// default `cursors::CursorTactic`. Same result contract; intended for differential testing of the
+/// two tactics against each other.
+pub fn reduce_trace_reference<'scope, Tr1, Bu, Tr2, L, P>(trace: Arranged<'scope, Tr1>, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
+where
+    Tr1: TraceReader<Batch: Navigable> + 'static,
+    Tr2: Trace<Batch: Navigable, Time = Tr1::Time> + 'static,
+    BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+    for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
+    Bu: Builder<Time=Tr2::Time, Output = Tr2::Batch, Input: Default> + 'static,
+    L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+    P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
+{
+    reduce_with_tactic(trace, name, reference::ReferenceTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
+}
+
 /// Drives a key-wise reduction using a supplied [`ReduceTactic`].
 ///
 /// This is the general reduce operator: it does the dataflow plumbing (frontiers, capabilities, output
@@ -620,33 +655,76 @@ mod cursors {
                     // before we add the time to `synth_times`.
                     if !upper_limit.less_equal(&next_time) {
 
+                        // DETERMINATION (times only). Determine synthetic interesting times.
+                        //
+                        // Synthetic interesting times are produced differently for interesting and uninteresting
+                        // times. An uninteresting time must join with an interesting time to become interesting,
+                        // which means joins with `self.batch_history` and  `self.times_current`. I think we can
+                        // skip `self.synth_times` as we haven't gotten to them yet, but we will and they will be
+                        // joined against everything.
+
+                        // Any time, even uninteresting times, must be joined with the current accumulation of
+                        // batch times as well as the current accumulation of `times_current`.
+                        self.temporary.extend(batch_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                        self.temporary.extend(self.times_current.iter().filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+
+                        // An interesting time additionally joins with `input` and `output` history and this round's
+                        // produced output: it carries the seed, so those joins stay interesting (an uninteresting
+                        // time does not, as `input`/`output` times are not themselves seeds). We advance the buffers
+                        // by `meet` first, exactly as evaluation reads them below; by join preservation the advanced
+                        // and unadvanced times spawn the same synthetics, so this matches the pre-split behavior.
+                        if interesting {
+                            if let Some(meet) = meet.as_ref() { input_replay.advance_buffer_by(meet) };
+                            if let Some(meet) = meet.as_ref() { output_replay.advance_buffer_by(meet) };
+                            self.temporary.extend(input_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                            self.temporary.extend(output_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                            self.temporary.extend(self.output_produced.iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
+                        }
+                        sort_dedup(&mut self.temporary);
+
+                        // Introduce synthetic times, and re-organize if we add any.
+                        let synth_len = self.synth_times.len();
+                        for time in self.temporary.drain(..) {
+                            // We can either service `join` now, or must delay for the future.
+                            if upper_limit.less_equal(&time) {
+                                debug_assert!(outputs.iter().any(|(t,_)| t.less_equal(&time)));
+                                new_interesting.push(time);
+                            }
+                            else {
+                                self.synth_times.push(time);
+                            }
+                        }
+                        if self.synth_times.len() > synth_len {
+                            self.synth_times.sort_by(|x,y| y.cmp(x));
+                            self.synth_times.dedup();
+                        }
+
+                        // EVALUATION (values only).
                         // We should re-evaluate the computation if this is an interesting time.
                         // If the time is uninteresting (and our logic is sound) it is not possible for there to be
                         // output produced. This sounds like a good test to have for debug builds!
                         if interesting {
 
+                            #[cfg(feature = "reduce-metrics")]
+                            crate::operators::reduce::metrics::CURSOR.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+
                             // Assemble the input collection at `next_time`. (`self.input_buffer` cleared just after use).
+                            // The buffers were advanced by `meet` in the determination step above.
                             debug_assert!(self.input_buffer.is_empty());
-                            if let Some(meet) = meet.as_ref() { input_replay.advance_buffer_by(meet) };
                             for ((value, time), diff) in input_replay.buffer().iter() {
                                 if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             for ((value, time), diff) in batch_replay.buffer().iter() {
                                 if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             crate::consolidation::consolidate(&mut self.input_buffer);
 
                             // Assemble the output collection at `next_time`. (`self.output_buffer` cleared just after use).
-                            if let Some(meet) = meet.as_ref() { output_replay.advance_buffer_by(meet) };
                             for ((value, time), diff) in output_replay.buffer().iter() {
                                 if time.less_equal(&next_time) { self.output_buffer.push((C2::owned_val(*value), diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             for ((value, time), diff) in self.output_produced.iter() {
                                 if time.less_equal(&next_time) { self.output_buffer.push(((*value).to_owned(), diff.clone())); }
-                                else { self.temporary.push(next_time.join(time)); }
                             }
                             crate::consolidation::consolidate(&mut self.output_buffer);
 
@@ -686,37 +764,6 @@ mod cursors {
                                 }
                             }
                         }
-
-                        // Determine synthetic interesting times.
-                        //
-                        // Synthetic interesting times are produced differently for interesting and uninteresting
-                        // times. An uninteresting time must join with an interesting time to become interesting,
-                        // which means joins with `self.batch_history` and  `self.times_current`. I think we can
-                        // skip `self.synth_times` as we haven't gotten to them yet, but we will and they will be
-                        // joined against everything.
-
-                        // Any time, even uninteresting times, must be joined with the current accumulation of
-                        // batch times as well as the current accumulation of `times_current`.
-                        self.temporary.extend(batch_replay.buffer().iter().map(|((_,time),_)| time).filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
-                        self.temporary.extend(self.times_current.iter().filter(|time| !time.less_equal(&next_time)).map(|time| time.join(&next_time)));
-                        sort_dedup(&mut self.temporary);
-
-                        // Introduce synthetic times, and re-organize if we add any.
-                        let synth_len = self.synth_times.len();
-                        for time in self.temporary.drain(..) {
-                            // We can either service `join` now, or must delay for the future.
-                            if upper_limit.less_equal(&time) {
-                                debug_assert!(outputs.iter().any(|(t,_)| t.less_equal(&time)));
-                                new_interesting.push(time);
-                            }
-                            else {
-                                self.synth_times.push(time);
-                            }
-                        }
-                        if self.synth_times.len() > synth_len {
-                            self.synth_times.sort_by(|x,y| y.cmp(x));
-                            self.synth_times.dedup();
-                        }
                     }
                     else if interesting {
                         // We cannot process `next_time` now, and must delay it.
@@ -760,3 +807,482 @@ mod cursors {
         }
     }
 }
+
+/// A second [`ReduceTactic`], written directly from the incremental model in
+/// `formal/Differential/Model.lean`.
+///
+/// Per key it runs two phases over one cursor walk. Phase 1 (determination) computes the
+/// interesting times as the truncated join-closure over {input, output, seeds} — the model's
+/// `Reached` — advancing by meets so the synthetic set stays bounded. Phase 2 (application) walks
+/// exactly those times in order, maintaining tight input/output accumulations by meets, and emits
+/// the corrections — the model's `emit_correct`. Determination never consults the output produced
+/// this round (it finishes first), so this tactic embodies the proven algorithm exactly and is the
+/// clean subject for differential testing against [`cursors::CursorTactic`].
+mod reference {
+
+    use super::*;
+    use crate::lattice::Lattice;
+    use crate::operators::ValueHistory;
+
+    /// Sorts and deduplicates, matching `cursors::sort_dedup`.
+    #[inline(never)]
+    fn sort_dedup<T: Ord>(list: &mut Vec<T>) {
+        list.dedup();
+        list.sort();
+        list.dedup();
+    }
+
+    /// Updates an optional meet by an optional time.
+    fn update_meet<T: Lattice+Clone>(meet: &mut Option<T>, other: Option<&T>) {
+        if let Some(time) = other {
+            if let Some(meet) = meet.as_mut() { meet.meet_assign(time); }
+            else { *meet = Some(time.clone()); }
+        }
+    }
+
+    /// The model-derived [`ReduceTactic`]. Structurally a twin of [`cursors::CursorTactic`]; only the
+    /// per-key engine differs.
+    pub struct ReferenceTactic<B1, B2, Bu, L, P>
+    where
+        B1: BatchReader + Navigable,
+        B2: BatchReader<Time = B1::Time> + Navigable,
+        B1::Cursor: Cursor<Time = B1::Time>,
+        for<'a> B2::Cursor: Cursor<Key<'a> = <B1::Cursor as Cursor>::Key<'a>, ValOwn: Data, Time = B1::Time>,
+    {
+        logic: L,
+        push: P,
+        pending_keys: <B1::Cursor as Cursor>::KeyContainer,
+        pending_time: <B1::Cursor as Cursor>::TimeContainer,
+        next_pending_keys: <B1::Cursor as Cursor>::KeyContainer,
+        next_pending_time: <B1::Cursor as Cursor>::TimeContainer,
+        interesting_times: Vec<B1::Time>,
+        new_interesting_times: Vec<B1::Time>,
+        output_upper: Antichain<B1::Time>,
+        output_lower: Antichain<B1::Time>,
+        _marker: PhantomData<(B2, Bu)>,
+    }
+
+    impl<B1, B2, Bu, L, P> ReferenceTactic<B1, B2, Bu, L, P>
+    where
+        B1: BatchReader + Navigable,
+        B2: BatchReader<Time = B1::Time> + Navigable,
+        B1::Cursor: Cursor<Time = B1::Time>,
+        for<'a> B2::Cursor: Cursor<Key<'a> = <B1::Cursor as Cursor>::Key<'a>, ValOwn: Data, Time = B1::Time>,
+    {
+        /// Construct a tactic that applies `logic` to each key and shapes output with `push`.
+        pub fn new(logic: L, push: P) -> Self {
+            ReferenceTactic {
+                logic,
+                push,
+                pending_keys: <B1::Cursor as Cursor>::KeyContainer::with_capacity(0),
+                pending_time: <B1::Cursor as Cursor>::TimeContainer::with_capacity(0),
+                next_pending_keys: <B1::Cursor as Cursor>::KeyContainer::with_capacity(0),
+                next_pending_time: <B1::Cursor as Cursor>::TimeContainer::with_capacity(0),
+                interesting_times: Vec::new(),
+                new_interesting_times: Vec::new(),
+                output_upper: Antichain::from_elem(<B1::Time as Timestamp>::minimum()),
+                output_lower: Antichain::from_elem(<B1::Time as Timestamp>::minimum()),
+                _marker: PhantomData,
+            }
+        }
+    }
+
+    impl<B1, B2, Bu, L, P> ReduceTactic<B1, B2> for ReferenceTactic<B1, B2, Bu, L, P>
+    where
+        B1: BatchReader + Navigable,
+        B2: BatchReader<Time = B1::Time> + Navigable,
+        B1::Cursor: Cursor<Time = B1::Time>,
+        for<'a> B2::Cursor: Cursor<Key<'a> = <B1::Cursor as Cursor>::Key<'a>, ValOwn: Data, Time = B1::Time>,
+        Bu: Builder<Time = B1::Time, Output = B2, Input: Default>,
+        L: FnMut(<B1::Cursor as Cursor>::Key<'_>, &[(<B1::Cursor as Cursor>::Val<'_>, <B1::Cursor as Cursor>::Diff)], &mut Vec<(<B2::Cursor as Cursor>::ValOwn, <B2::Cursor as Cursor>::Diff)>, &mut Vec<(<B2::Cursor as Cursor>::ValOwn, <B2::Cursor as Cursor>::Diff)>),
+        P: FnMut(&mut Bu::Input, <B1::Cursor as Cursor>::Key<'_>, &mut Vec<(<B2::Cursor as Cursor>::ValOwn, B1::Time, <B2::Cursor as Cursor>::Diff)>),
+    {
+        fn retire(
+            &mut self,
+            source_batches: Vec<B1>,
+            output_batches: Vec<B2>,
+            input_batches: Vec<B1>,
+            lower: &Antichain<B1::Time>,
+            upper: &Antichain<B1::Time>,
+            held: &Antichain<B1::Time>,
+        ) -> (Vec<(B1::Time, B2)>, Antichain<B1::Time>)
+        {
+            let mut produced = Vec::new();
+
+            if held.elements().iter().any(|time| !upper.less_equal(time)) {
+
+                let (mut source_cursor, ref source_storage) = cursor_list(source_batches);
+                let (mut output_cursor, ref output_storage) = cursor_list(output_batches);
+                let (mut batch_cursor, ref batch_storage) = cursor_list(input_batches);
+
+                let mut buffers = Vec::<(B1::Time, Vec<(<B2::Cursor as Cursor>::ValOwn, B1::Time, <B2::Cursor as Cursor>::Diff)>)>::new();
+                let mut builders = Vec::new();
+                for time in held.elements().iter() {
+                    buffers.push((time.clone(), Vec::new()));
+                    builders.push(Bu::new());
+                }
+                let mut buffer = Bu::Input::default();
+
+                // Reuseable state for performing the computation.
+                let mut thinker = ReferenceThinker::new();
+
+                let mut pending_pos = 0;
+                while batch_cursor.key_valid(batch_storage) || pending_pos < self.pending_keys.len() {
+
+                    let key1 = self.pending_keys.get(pending_pos);
+                    let key2 = batch_cursor.get_key(batch_storage);
+                    let key = match (key1, key2) {
+                        (Some(key1), Some(key2)) => ::std::cmp::min(key1, key2),
+                        (Some(key1), None)       => key1,
+                        (None, Some(key2))       => key2,
+                        (None, None)             => unreachable!(),
+                    };
+
+                    let prior_pos = pending_pos;
+                    self.interesting_times.clear();
+                    while self.pending_keys.get(pending_pos) == Some(key) {
+                        let owned_time = <B1::Cursor as Cursor>::owned_time(self.pending_time.index(pending_pos));
+                        if !upper.less_equal(&owned_time) { self.interesting_times.push(owned_time); }
+                        pending_pos += 1;
+                    }
+
+                    sort_dedup(&mut self.interesting_times);
+
+                    if batch_cursor.get_key(batch_storage) == Some(key) || !self.interesting_times.is_empty() {
+
+                        thinker.compute(
+                            key,
+                            (&mut source_cursor, source_storage),
+                            (&mut output_cursor, output_storage),
+                            (&mut batch_cursor, batch_storage),
+                            &self.interesting_times,
+                            &mut self.logic,
+                            upper,
+                            &mut buffers[..],
+                            &mut self.new_interesting_times,
+                        );
+
+                        if batch_cursor.get_key(batch_storage) == Some(key) { batch_cursor.step_key(batch_storage); }
+
+                        for pos in prior_pos .. pending_pos {
+                            let owned_time = <B1::Cursor as Cursor>::owned_time(self.pending_time.index(pos));
+                            if upper.less_equal(&owned_time) { self.new_interesting_times.push(owned_time); }
+                        }
+                        sort_dedup(&mut self.new_interesting_times);
+                        for time in self.new_interesting_times.drain(..) {
+                            self.next_pending_keys.push_ref(key);
+                            self.next_pending_time.push_own(&time);
+                        }
+
+                        for index in 0 .. buffers.len() {
+                            buffers[index].1.sort_by(|x,y| x.0.cmp(&y.0));
+                            (self.push)(&mut buffer, key, &mut buffers[index].1);
+                            buffers[index].1.clear();
+                            builders[index].push(&mut buffer);
+
+                        }
+                    }
+                    else {
+                        for pos in prior_pos .. pending_pos {
+                            self.next_pending_keys.push_ref(self.pending_keys.index(pos));
+                            self.next_pending_time.push_ref(self.pending_time.index(pos));
+                        }
+                    }
+                }
+                drop(thinker);
+
+                self.output_lower.clear();
+                self.output_lower.extend(lower.borrow().iter().cloned());
+
+                for (index, builder) in builders.drain(..).enumerate() {
+
+                    self.output_upper.clear();
+                    self.output_upper.extend(upper.borrow().iter().cloned());
+                    for time in &held.elements()[index + 1 ..] {
+                        self.output_upper.insert_ref(time);
+                    }
+
+                    if self.output_upper.borrow() != self.output_lower.borrow() {
+
+                        let description = Description::new(self.output_lower.clone(), self.output_upper.clone(), Antichain::from_elem(<B1::Time as Timestamp>::minimum()));
+                        let batch = builder.done(description);
+
+                        produced.push((held.elements()[index].clone(), batch));
+
+                        self.output_lower.clear();
+                        self.output_lower.extend(self.output_upper.borrow().iter().cloned());
+                    }
+                }
+                assert!(self.output_upper.borrow() == upper.borrow());
+
+                self.pending_keys.clear(); std::mem::swap(&mut self.next_pending_keys, &mut self.pending_keys);
+                self.pending_time.clear(); std::mem::swap(&mut self.next_pending_time, &mut self.pending_time);
+
+                let mut frontier = Antichain::<B1::Time>::new();
+                let mut owned_time = <B1::Time as Timestamp>::minimum();
+                for pos in 0 .. self.pending_time.len() {
+                    <B1::Cursor as Cursor>::clone_time_onto(self.pending_time.index(pos), &mut owned_time);
+                    frontier.insert_ref(&owned_time);
+                }
+
+                (produced, frontier)
+            }
+            else {
+                (produced, held.clone())
+            }
+        }
+    }
+
+    /// The two-phase per-key engine.
+    ///
+    /// Phase 1 (determination) reads the input/output/seed *times* and closes them into `active`
+    /// (the interesting times) and the pended set — Model.lean's `Reached`, directly. Phase 2
+    /// (application) walks the same, still-loaded histories for *values* and emits corrections.
+    pub struct ReferenceThinker<V1, V2, V, T, D1, D2> {
+        input_history: ValueHistory<V1, T, D1>,
+        output_history: ValueHistory<V2, T, D2>,
+        batch_history: ValueHistory<V1, T, D1>,
+        input_buffer: Vec<(V1, D1)>,
+        output_buffer: Vec<(V, D2)>,
+        update_buffer: Vec<(V, D2)>,
+        output_produced: Vec<((V, T), D2)>,
+        // Phase 1 (the compacted closure): synthetic reached times still to visit, the reached times
+        // in play as join partners, scratch for the joins, and suffix-meets of the supplied times.
+        synth_times: Vec<T>,
+        times_current: Vec<T>,
+        temporary: Vec<T>,
+        meets: Vec<T>,
+        // The interesting (in-band reached) times, handed from phase 1 to phase 2.
+        active: Vec<T>,
+    }
+
+    impl<V1, V2, V, T, D1, D2> ReferenceThinker<V1, V2, V, T, D1, D2>
+    where
+        V1: Copy + Ord,
+        V2: Copy + Ord,
+        V: Clone + Ord,
+        T: Ord + Clone + Lattice + 'static,
+        D1: Clone + crate::difference::Semigroup,
+        D2: Clone + crate::difference::Semigroup,
+    {
+        pub fn new() -> Self {
+            ReferenceThinker {
+                input_history: ValueHistory::new(),
+                output_history: ValueHistory::new(),
+                batch_history: ValueHistory::new(),
+                input_buffer: Vec::new(),
+                output_buffer: Vec::new(),
+                update_buffer: Vec::new(),
+                output_produced: Vec::new(),
+                synth_times: Vec::new(),
+                times_current: Vec::new(),
+                temporary: Vec::new(),
+                meets: Vec::new(),
+                active: Vec::new(),
+            }
+        }
+
+        #[inline(never)]
+        pub fn compute<'a, K, C1, C2, C3, L>(
+            &mut self,
+            key: K,
+            (source_cursor, source_storage): (&mut C1, &'a C1::Storage),
+            (output_cursor, output_storage): (&mut C2, &'a C2::Storage),
+            (batch_cursor, batch_storage): (&mut C3, &'a C3::Storage),
+            times: &Vec<T>,
+            logic: &mut L,
+            upper_limit: &Antichain<T>,
+            outputs: &mut [(T, Vec<(V, T, D2)>)],
+            new_interesting: &mut Vec<T>)
+        where
+            C1: Cursor<Key<'a> = K, Val<'a> = V1, Time = T, Diff = D1>,
+            C2: Cursor<Key<'a> = K, Val<'a> = V2, ValOwn = V, Time = T, Diff = D2>,
+            C3: Cursor<Key<'a> = K, Val<'a> = V1, Time = T, Diff = D1>,
+            K: Copy + Ord,
+            L: FnMut(K, &[(V1, D1)], &mut Vec<(V, D2)>, &mut Vec<(V, D2)>),
+        {
+            // ================== PHASE 1 — DETERMINATION (`Reached`, compacted) ==================
+            // The interesting times are Model.lean's `Reached` — but computed the non-quadratic way.
+            // Walk the input/output/seed times in increasing order; a time is *reached* only via a
+            // seed (a batch update, a due pending time, or a synthetic join of earlier reached times);
+            // a reached in-band time joins against the live partners to spawn more reached times, and
+            // a join beyond `upper` is pended. The live partner sets are kept an antichain by
+            // `advance_buffer_by(meet)` — coincident times collapse under the running meet — so this is
+            // the closure without the all-pairs blow-up. Time-only: `TimeReplay` reads the histories
+            // without touching values or stepping the underlying `history`, so phase 2 can walk them.
+            {
+                // Suffix-meets of the supplied (due pending) times, consumed as we pass them.
+                self.meets.clear();
+                self.meets.extend(times.iter().cloned());
+                for index in (1 .. self.meets.len()).rev() {
+                    self.meets[index-1] = self.meets[index-1].meet(&self.meets[index]);
+                }
+
+                // Build each history, then read it time-only (leaving it intact for phase 2).
+                drop(self.batch_history.replay_key(batch_cursor, batch_storage, key, None));
+                drop(self.input_history.replay_key(source_cursor, source_storage, key, None));
+                drop(self.output_history.replay_key(output_cursor, output_storage, key, None));
+                let mut batch_replay = self.batch_history.replay_times();
+                let mut input_replay = self.input_history.replay_times();
+                let mut output_replay = self.output_history.replay_times();
+
+                self.synth_times.clear();
+                self.times_current.clear();
+                self.temporary.clear();
+                self.active.clear();
+
+                let mut times_slice = &times[..];
+                let mut meets_slice = &self.meets[..];
+                let mut meet: Option<T> = None;
+
+                while let Some(next_time) = [   batch_replay.time(),
+                                                times_slice.first(),
+                                                input_replay.time(),
+                                                output_replay.time(),
+                                                self.synth_times.last(),
+                                            ].into_iter().flatten().min().cloned() {
+
+                    input_replay.step_while_time_is(&next_time);
+                    output_replay.step_while_time_is(&next_time);
+
+                    // Reached via a seed: a batch update, a due pending time, or a synthetic join.
+                    // (Input/output times alone are not reached — they are only join partners.)
+                    let mut interesting = batch_replay.step_while_time_is(&next_time);
+                    if interesting { if let Some(m) = meet.as_ref() { batch_replay.advance_buffer_by(m); } }
+                    while self.synth_times.last() == Some(&next_time) {
+                        self.times_current.push(self.synth_times.pop().unwrap());
+                        interesting = true;
+                    }
+                    while times_slice.first() == Some(&next_time) {
+                        self.times_current.push(next_time.clone());
+                        times_slice = &times_slice[1..];
+                        meets_slice = &meets_slice[1..];
+                        interesting = true;
+                    }
+                    // Absorb: a time at or above a reached time is itself reached.
+                    interesting = interesting
+                        || batch_replay.buffer().iter().any(|t| t.less_equal(&next_time))
+                        || self.times_current.iter().any(|t| t.less_equal(&next_time));
+
+                    if !upper_limit.less_equal(&next_time) {
+                        // A reached in-band time is `active`; join it against the live partners —
+                        // input/output (`joinBase`) and reached-so-far (`joinActive`) — for new times.
+                        if interesting {
+                            self.active.push(next_time.clone());
+                            if let Some(m) = meet.as_ref() { input_replay.advance_buffer_by(m); output_replay.advance_buffer_by(m); }
+                            self.temporary.extend(input_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                            self.temporary.extend(output_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                        }
+                        self.temporary.extend(batch_replay.buffer().iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                        self.temporary.extend(self.times_current.iter().filter(|t| !t.less_equal(&next_time)).map(|t| t.join(&next_time)));
+                        sort_dedup(&mut self.temporary);
+
+                        let synth_len = self.synth_times.len();
+                        for time in self.temporary.drain(..) {
+                            if upper_limit.less_equal(&time) { new_interesting.push(time); }  // pended
+                            else { self.synth_times.push(time); }                             // reached, later
+                        }
+                        if self.synth_times.len() > synth_len {
+                            self.synth_times.sort_by(|x,y| y.cmp(x));
+                            self.synth_times.dedup();
+                        }
+                    }
+                    else if interesting {
+                        new_interesting.push(next_time.clone());
+                    }
+
+                    // Running meet (a lower bound on every time still to visit); compact the reached
+                    // partners `times_current` with it.
+                    meet = None;
+                    update_meet(&mut meet, batch_replay.meet());
+                    update_meet(&mut meet, input_replay.meet());
+                    update_meet(&mut meet, output_replay.meet());
+                    for time in self.synth_times.iter() { update_meet(&mut meet, Some(time)); }
+                    update_meet(&mut meet, meets_slice.first());
+                    if let Some(m) = meet.as_ref() {
+                        for time in self.times_current.iter_mut() { *time = time.join(m); }
+                    }
+                    sort_dedup(&mut self.times_current);
+                }
+
+                sort_dedup(&mut self.active);
+
+                #[cfg(feature = "reduce-metrics")]
+                crate::operators::reduce::metrics::REFERENCE.fetch_add(self.active.len(), std::sync::atomic::Ordering::Relaxed);
+            }
+
+            // ===================== PHASE 2 — APPLICATION (`emit_correct`) =====================
+            // Walk `self.active` in order over the SAME per-key edits (via `replay`, no cursor), keep
+            // the input/output accumulations tight by advancing to the meet of the times still to be
+            // produced, apply `logic`, and emit corrections.
+            {
+                self.meets.clear();
+                self.meets.extend(self.active.iter().cloned());
+                for index in (1 .. self.meets.len()).rev() {
+                    self.meets[index-1] = self.meets[index-1].meet(&self.meets[index]);
+                }
+
+                // Walk the histories loaded (and left intact) by phase 1 — no cursor re-read, no
+                // rebuild, no re-sort; just a fresh walk of the same sorted `history` for values.
+                let mut batch_replay = self.batch_history.walk();
+                let mut input_replay = self.input_history.walk();
+                let mut output_replay = self.output_history.walk();
+
+                self.output_produced.clear();
+
+                for index in 0 .. self.active.len() {
+                    let next_time = self.active[index].clone();
+                    let meet = self.meets[index].clone();
+
+                    // Phase 2 visits only the active times, so we must consume ALL history edits at or
+                    // below `next_time` (not just those exactly at it, as a full time-order walk would
+                    // reach incrementally); edits at non-active times still contribute to the
+                    // accumulation here.
+                    while input_replay.time().map_or(false, |t| t.less_equal(&next_time)) { input_replay.step(); }
+                    while batch_replay.time().map_or(false, |t| t.less_equal(&next_time)) { batch_replay.step(); }
+                    while output_replay.time().map_or(false, |t| t.less_equal(&next_time)) { output_replay.step(); }
+                    input_replay.advance_buffer_by(&meet);
+                    batch_replay.advance_buffer_by(&meet);
+                    output_replay.advance_buffer_by(&meet);
+
+                    debug_assert!(self.input_buffer.is_empty());
+                    for ((value, time), diff) in input_replay.buffer().iter() {
+                        if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
+                    }
+                    for ((value, time), diff) in batch_replay.buffer().iter() {
+                        if time.less_equal(&next_time) { self.input_buffer.push((*value, diff.clone())); }
+                    }
+                    crate::consolidation::consolidate(&mut self.input_buffer);
+
+                    for ((value, time), diff) in output_replay.buffer().iter() {
+                        if time.less_equal(&next_time) { self.output_buffer.push((C2::owned_val(*value), diff.clone())); }
+                    }
+                    for ((value, time), diff) in self.output_produced.iter() {
+                        if time.less_equal(&next_time) { self.output_buffer.push((value.clone(), diff.clone())); }
+                    }
+                    crate::consolidation::consolidate(&mut self.output_buffer);
+
+                    if !self.input_buffer.is_empty() || !self.output_buffer.is_empty() {
+                        logic(key, &self.input_buffer[..], &mut self.output_buffer, &mut self.update_buffer);
+                        self.input_buffer.clear();
+                        self.output_buffer.clear();
+
+                        crate::consolidation::consolidate(&mut self.update_buffer);
+                        if !self.update_buffer.is_empty() {
+
+                            let idx = outputs.iter().rev().position(|(time, _)| time.less_equal(&next_time));
+                            let idx = outputs.len() - idx.expect("failed to find index") - 1;
+                            for (val, diff) in self.update_buffer.drain(..) {
+                                self.output_produced.push(((val.clone(), next_time.clone()), diff.clone()));
+                                outputs[idx].1.push((val, next_time.clone(), diff));
+                            }
+
+                            for entry in &mut self.output_produced { (entry.0).1.join_assign(&meet); }
+                            crate::consolidation::consolidate(&mut self.output_produced);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

From b3a2d7b2f5064e11b86c3d7b4c77e10e0c793e0e Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 14:47:19 -0400
Subject: [PATCH 03/10] reduce: differential + oracle tests and benchmark for
 the reference tactic

tests/reduce_reference.rs: a differential harness (cursor vs reference, assert_eq at
every time) over Pair<u64,u64> partial-order random input with retractions and an
iterative BFS; an oracle checking each tactic against a from-scratch f(acc input)
(the executable form of Model.lean's stream-correctness); and a feature-gated
over-derivation metric -- measured 1.00x, i.e. the value-blind reference derives the
same interesting-time set as the value-aware cursor. examples/reduce_bench.rs: a
cursor-vs-reference benchmark, at parity across churn / wide-interval / iterative
workloads.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../examples/reduce_bench.rs                  | 188 +++++++++++
 .../tests/reduce_reference.rs                 | 316 ++++++++++++++++++
 2 files changed, 504 insertions(+)
 create mode 100644 differential-dataflow/examples/reduce_bench.rs
 create mode 100644 differential-dataflow/tests/reduce_reference.rs

diff --git a/differential-dataflow/examples/reduce_bench.rs b/differential-dataflow/examples/reduce_bench.rs
new file mode 100644
index 000000000..5a74b0719
--- /dev/null
+++ b/differential-dataflow/examples/reduce_bench.rs
@@ -0,0 +1,188 @@
+//! Wall-clock comparison of the default cursor reduce tactic vs the model-derived reference tactic.
+//!
+//!   cargo run --release --example reduce_bench
+//!
+//! Two workloads: a reduce-dominated 1-D (total-order) churn, which isolates the base per-key engine
+//! overhead (determination is trivial with total orders), and an iterative BFS with Product times,
+//! where the synthetic-time determination actually does work.
+
+use std::time::Instant;
+
+use rand::{Rng, SeedableRng, StdRng};
+
+use differential_dataflow::input::Input;
+use differential_dataflow::VecCollection;
+use differential_dataflow::operators::*;
+use differential_dataflow::lattice::Lattice;
+
+type Node = usize;
+type Edge = (Node, Node);
+
+fn main() {
+    for _ in 0 .. 2 {
+        let c = reduce_churn(false);
+        let r = reduce_churn(true);
+        println!("reduce_churn  cursor {:>8.3}s   reference {:>8.3}s   ratio {:.2}x",
+            c.as_secs_f64(), r.as_secs_f64(), r.as_secs_f64() / c.as_secs_f64());
+    }
+    for _ in 0 .. 2 {
+        let c = fewkeys(false);
+        let r = fewkeys(true);
+        println!("fewkeys       cursor {:>8.3}s   reference {:>8.3}s   ratio {:.2}x",
+            c.as_secs_f64(), r.as_secs_f64(), r.as_secs_f64() / c.as_secs_f64());
+    }
+    for _ in 0 .. 2 {
+        let c = bfs_bench(false);
+        let r = bfs_bench(true);
+        println!("bfs           cursor {:>8.3}s   reference {:>8.3}s   ratio {:.2}x",
+            c.as_secs_f64(), r.as_secs_f64(), r.as_secs_f64() / c.as_secs_f64());
+    }
+}
+
+// Reduce-dominated, totally-ordered time: many keys churn values across many rounds; reduce counts
+// distinct values per key. Determination is trivial here, so this measures base engine overhead.
+fn reduce_churn(reference: bool) -> std::time::Duration {
+    let keys = 20000usize;
+    let vals = 100usize;
+    let rounds = 500usize;
+    let per_round = 5000usize;
+
+    let seed: &[_] = &[1, 2, 3, 4];
+    let mut rng: StdRng = SeedableRng::from_seed(seed);
+    let mut batches: Vec<Vec<(usize, usize, isize)>> = Vec::with_capacity(rounds);
+    for _ in 0 .. rounds {
+        let mut batch = Vec::with_capacity(per_round);
+        for _ in 0 .. per_round {
+            let k = rng.gen_range(0, keys);
+            let v = rng.gen_range(0, vals);
+            let d = if rng.gen::<bool>() { 1 } else { -1 };
+            batch.push((k, v, d));
+        }
+        batches.push(batch);
+    }
+
+    let start = Instant::now();
+    timely::execute_directly(move |worker| {
+        let mut input = worker.dataflow::<usize, _, _>(|scope| {
+            let (input, coll) = scope.new_collection::<(usize, usize), isize>();
+            let reduced = if reference {
+                coll.reduce_reference(|_k, s, t| t.push((s.len() as isize, 1)))
+            } else {
+                coll.reduce(|_k, s, t| t.push((s.len() as isize, 1)))
+            };
+            reduced.inspect(|_| {});
+            input
+        });
+        let mut round = 0usize;
+        for batch in batches {
+            for (k, v, d) in batch { input.update((k, v), d); }
+            round += 1;
+            input.advance_to(round);
+        }
+    });
+    start.elapsed()
+}
+
+// Few keys, wide processing intervals (via `update_at` + coarse `advance_to`), so each key's
+// per-round history is long — the regime where the phase-2 re-sort would actually cost.
+fn fewkeys(reference: bool) -> std::time::Duration {
+    let keys = 8usize;
+    let vals = 40usize;
+    let times = 24000usize;
+    let step = 6000usize;
+
+    let seed: &[_] = &[7, 7, 7, 7];
+    let mut rng: StdRng = SeedableRng::from_seed(seed);
+    let mut ups: Vec<(usize, usize, usize, isize)> = Vec::new();
+    for t in 1 ..= times {
+        for k in 0 .. keys {
+            ups.push((k, rng.gen_range(0, vals), t, if rng.gen::<bool>() { 1 } else { -1 }));
+        }
+    }
+
+    let start = Instant::now();
+    timely::execute_directly(move |worker| {
+        let mut input = worker.dataflow::<usize, _, _>(|scope| {
+            let (input, coll) = scope.new_collection::<(usize, usize), isize>();
+            let reduced = if reference {
+                coll.reduce_reference(|_k, s, t| t.push((s.len() as isize, 1)))
+            } else {
+                coll.reduce(|_k, s, t| t.push((s.len() as isize, 1)))
+            };
+            reduced.inspect(|_| {});
+            input
+        });
+        let mut next_flush = step;
+        for (k, v, t, d) in ups {
+            input.update_at((k, v), t, d);
+            if t >= next_flush { input.advance_to(next_flush); next_flush += step; }
+        }
+        input.advance_to(times + 1);
+    });
+    start.elapsed()
+}
+
+fn bfs_bench(reference: bool) -> std::time::Duration {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::Arc;
+
+    let nodes = 4000usize;
+    let edges = 32000usize;
+    let rounds = 400usize;
+
+    let seed: &[_] = &[9, 8, 7, 6];
+    let mut rng1: StdRng = SeedableRng::from_seed(seed);
+    let mut rng2: StdRng = SeedableRng::from_seed(seed);
+    let mut edge_list: Vec<((usize, usize), usize, isize)> = Vec::new();
+    for _ in 0 .. edges { edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), 0, 1)); }
+    for round in 1 .. rounds {
+        edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), round, 1));
+        edge_list.push(((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes)), round, -1));
+    }
+    let root_list = vec![(1usize, 0usize, 1isize)];
+    let counter = Arc::new(AtomicUsize::new(0));
+
+    let start = Instant::now();
+    let c = counter.clone();
+    timely::execute_directly(move |worker| {
+        let mut roots_list = root_list.clone();
+        let mut edges_list = edge_list.clone();
+        let (mut roots, mut edges) = worker.dataflow::<usize, _, _>(|scope| {
+            let (root_input, roots) = scope.new_collection();
+            let (edge_input, edges) = scope.new_collection();
+            let c = c.clone();
+            bfs(edges, roots, reference).inspect(move |_| { c.fetch_add(1, Ordering::Relaxed); });
+            (root_input, edge_input)
+        });
+        roots_list.sort_by(|x, y| y.1.cmp(&x.1));
+        edges_list.sort_by(|x, y| y.1.cmp(&x.1));
+        let mut round = 0;
+        while !roots_list.is_empty() || !edges_list.is_empty() {
+            while roots_list.last().map(|x| x.1) == Some(round) { let (n, _, d) = roots_list.pop().unwrap(); roots.update(n, d); }
+            while edges_list.last().map(|x| x.1) == Some(round) { let ((s, t), _, d) = edges_list.pop().unwrap(); edges.update((s, t), d); }
+            round += 1;
+            roots.advance_to(round);
+            edges.advance_to(round);
+        }
+    });
+    let elapsed = start.elapsed();
+    eprintln!("    (bfs {} output updates)", counter.load(std::sync::atomic::Ordering::Relaxed));
+    elapsed
+}
+
+fn bfs<'scope, T>(edges: VecCollection<'scope, T, Edge>, roots: VecCollection<'scope, T, Node>, reference: bool) -> VecCollection<'scope, T, (Node, usize)>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let nodes = roots.map(|x| (x, 0));
+    nodes.clone().iterate(|scope, inner| {
+        let edges = edges.enter(scope);
+        let nodes = nodes.enter(scope);
+        let combined = inner.join_map(edges, |_k, l, d| (*d, l + 1)).concat(nodes);
+        if reference {
+            combined.reduce_reference(|_, s, t| t.push((*s[0].0, 1)))
+        } else {
+            combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
+        }
+    })
+}
diff --git a/differential-dataflow/tests/reduce_reference.rs b/differential-dataflow/tests/reduce_reference.rs
new file mode 100644
index 000000000..f4720f63c
--- /dev/null
+++ b/differential-dataflow/tests/reduce_reference.rs
@@ -0,0 +1,316 @@
+//! Test harness for the model-derived `reference` reduce tactic. Four things:
+//!
+//! * `differential_*` — run the default (cursor) reduce and the `reference` reduce on the SAME random
+//!   input and `assert_eq` their outputs at every time. The drift detector: if either tactic changes
+//!   behavior, it fails. Inputs use `Pair<u64,u64>` (product / partial order, so joins and synthetic
+//!   times occur) with diffs ±1 (so updates cancel — the corner where the value-blind reference could
+//!   over-derive).
+//!
+//! * `oracle_*` — run one tactic and check its output against a from-scratch recomputation of
+//!   `f(accumulated input)` at many times. The correctness check: it does not rely on the two tactics
+//!   agreeing, so it catches bugs in code they share (the driver, `ValueHistory`, …). It is the
+//!   executable form of `formal/Differential/Model.lean`'s stream-correctness statement.
+//!
+//! * `bfs_*` — an iterative BFS (product time via iteration; a real computation shape) computed both
+//!   ways over a graph that grows and shrinks; another differential check.
+//!
+//! * `over_derivation` — with `--features reduce-metrics`, count how many interesting times each
+//!   tactic evaluates on a cancellation-heavy input, to measure the value-blind over-derivation.
+
+use std::collections::BTreeMap;
+
+use rand::{Rng, SeedableRng, StdRng};
+
+use timely::PartialOrder;
+use timely::dataflow::operators::capture::{Capture, Extract};
+use timely::dataflow::operators::vec::unordered_input::UnorderedInput;
+
+use differential_dataflow::AsCollection;
+use differential_dataflow::VecCollection;
+use differential_dataflow::input::Input;
+use differential_dataflow::lattice::Lattice;
+use differential_dataflow::operators::*;
+
+use pair::Pair;
+type Time = Pair<u64, u64>;
+
+// ---- the reduce logics under test (plain `fn`s, so they are `Copy` and can feed both tactics) ----
+
+fn count(_key: &u64, input: &[(&u64, isize)], output: &mut Vec<(i64, isize)>) {
+    output.push((input.len() as i64, 1));
+}
+fn sum(_key: &u64, input: &[(&u64, isize)], output: &mut Vec<(i64, isize)>) {
+    let s: i64 = input.iter().map(|(v, d)| **v as i64 * *d as i64).sum();
+    output.push((s, 1));
+}
+fn min(_key: &u64, input: &[(&u64, isize)], output: &mut Vec<(i64, isize)>) {
+    if let Some(m) = input.iter().map(|(v, _)| **v).min() {
+        output.push((m as i64, 1));
+    }
+}
+
+type Logic = fn(&u64, &[(&u64, isize)], &mut Vec<(i64, isize)>);
+
+// A random input: `((key, val), Pair(a, b), diff)`, diffs ±1 so updates cancel.
+fn random_input(seed: usize, n: usize, keys: u64, vals: u64, span: u64) -> Vec<((u64, u64), Time, isize)> {
+    let s: &[usize] = &[seed, 7, 13, 29];
+    let mut rng: StdRng = SeedableRng::from_seed(s);
+    (0..n).map(|_| {
+        let k = rng.gen_range(0, keys);
+        let v = rng.gen_range(0, vals);
+        let t = Pair::new(rng.gen_range(0, span), rng.gen_range(0, span));
+        let d = if rng.gen::<bool>() { 1 } else { -1 };
+        ((k, v), t, d)
+    }).collect()
+}
+
+// ======================= differential: cursor vs reference, all times =======================
+
+#[test] fn differential_count() { for s in 0..25 { differential(s, count); } }
+#[test] fn differential_sum()   { for s in 0..25 { differential(s, sum); } }
+#[test] fn differential_min()   { for s in 0..25 { differential(s, min); } }
+
+fn differential(seed: usize, logic: Logic) {
+    let span = 6u64;
+    let updates = random_input(seed, 400, 12, 8, span);
+    timely::execute_directly(move |worker| {
+        let (mut input, capability) = worker.dataflow::<Time, _, _>(|scope| {
+            let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
+            let collection = stream.as_collection();
+            let via_cursor    = collection.clone().reduce(logic);
+            let via_reference = collection.reduce_reference(logic);
+            via_cursor.assert_eq(via_reference);
+            (input, capability)
+        });
+        for u in updates { input.activate().session(&capability).give(u); }
+        drop(capability); // release the input frontier so everything finalizes
+    });
+}
+
+// ======================= oracle: a tactic vs from-scratch f(acc input) =======================
+
+#[test] fn oracle_cursor_count()    { for s in 0..15 { oracle(s, false, count); } }
+#[test] fn oracle_reference_count() { for s in 0..15 { oracle(s, true,  count); } }
+#[test] fn oracle_reference_sum()   { for s in 0..15 { oracle(s, true,  sum); } }
+#[test] fn oracle_reference_min()   { for s in 0..15 { oracle(s, true,  min); } }
+
+fn oracle(seed: usize, reference: bool, logic: Logic) {
+    let span = 6u64;
+    let updates = random_input(seed, 300, 10, 8, span);
+
+    // Capture the reduce output as `((key, out_val), time, diff)` updates.
+    let (send, recv) = std::sync::mpsc::channel();
+    let send = std::sync::Arc::new(std::sync::Mutex::new(send));
+    let drive = updates.clone();
+    timely::execute_directly(move |worker| {
+        let send = send.lock().unwrap().clone();
+        let (mut input, capability) = worker.dataflow::<Time, _, _>(|scope| {
+            let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
+            let collection = stream.as_collection();
+            let out = if reference { collection.reduce_reference(logic) } else { collection.reduce(logic) };
+            out.inner.capture_into(send);
+            (input, capability)
+        });
+        for u in drive { input.activate().session(&capability).give(u); }
+        drop(capability);
+    });
+    let output: Vec<((u64, i64), Time, isize)> =
+        recv.extract().into_iter().flat_map(|(_, batch)| batch).collect();
+
+    // Everything is finalized, so `acc(output, q) == f(acc(input, q))` must hold at every `q`. Check
+    // at a rich set: all input times, all output times, and their pairwise joins (where reduce's
+    // output can change).
+    let mut times: Vec<Time> = Vec::new();
+    times.extend(updates.iter().map(|(_, t, _)| t.clone()));
+    times.extend(output.iter().map(|(_, t, _)| t.clone()));
+    let base = times.clone();
+    for a in &base { for b in &base { times.push(a.join(b)); } }
+    times.sort(); times.dedup();
+
+    for q in &times {
+        // expected: per key, f applied to the accumulated input at `q`.
+        let mut expected: BTreeMap<(u64, i64), i64> = BTreeMap::new();
+        let mut per_key: BTreeMap<u64, BTreeMap<u64, isize>> = BTreeMap::new();
+        for ((k, v), t, d) in &updates {
+            if t.less_equal(q) { *per_key.entry(*k).or_default().entry(*v).or_default() += *d; }
+        }
+        for (k, vals) in &per_key {
+            let acc: Vec<(&u64, isize)> = vals.iter().filter(|(_, d)| **d != 0).map(|(v, d)| (v, *d)).collect();
+            if acc.is_empty() { continue; }
+            let mut out = Vec::new();
+            logic(k, &acc, &mut out);
+            for (ov, od) in out { *expected.entry((*k, ov)).or_default() += od as i64; }
+        }
+        expected.retain(|_, d| *d != 0);
+
+        // actual: the accumulated reduce output at `q`.
+        let mut actual: BTreeMap<(u64, i64), i64> = BTreeMap::new();
+        for ((k, ov), t, d) in &output {
+            if t.less_equal(q) { *actual.entry((*k, *ov)).or_default() += *d as i64; }
+        }
+        actual.retain(|_, d| *d != 0);
+
+        assert_eq!(expected, actual, "seed {seed} reference {reference} at time {q:?}");
+    }
+}
+
+// ============= over-derivation: interesting-time counts, cursor vs reference =============
+// Run with:  cargo test -p differential-dataflow --features reduce-metrics over_derivation -- --nocapture
+
+#[cfg(feature = "reduce-metrics")]
+#[test]
+fn over_derivation() {
+    use differential_dataflow::operators::reduce::metrics;
+    // Cancellation-heavy configs: few keys/values over dense product-time grids with ±1 diffs, so many
+    // updates advance-cancel — the cursor drops those (zero-debt) addresses, the reference keeps them.
+    // Sweep densities to look for any case where the value-blind reference derives more.
+    let configs = [
+        (1, 1200usize, 4u64, 3u64, 10u64),
+        (2, 1500, 3, 2, 12),
+        (3, 2000, 2, 2, 8),
+        (4, 1000, 6, 4, 10),
+        (5, 1800, 2, 3, 14),
+    ];
+    println!();
+    for (seed, n, keys, vals, span) in configs {
+        let updates = random_input(seed, n, keys, vals, span);
+        metrics::reset();
+        timely::execute_directly(move |worker| {
+            let (mut input, capability) = worker.dataflow::<Time, _, _>(|scope| {
+                let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
+                let collection = stream.as_collection();
+                collection.clone().reduce(count).inspect(|_| {});     // drive the cursor tactic
+                collection.reduce_reference(count).inspect(|_| {});   // drive the reference tactic
+                (input, capability)
+            });
+            for u in updates { input.activate().session(&capability).give(u); }
+            drop(capability);
+        });
+        let (c, r) = (metrics::cursor(), metrics::reference());
+        println!("  keys={keys} vals={vals} span={span} n={n}: cursor {c}, reference {r}  ({:.2}x)",
+                 r as f64 / (c.max(1)) as f64);
+        assert!(r >= c, "value-blind reference should derive at least as many interesting times as the cursor");
+    }
+}
+
+// ======================= iterative BFS differential (a real shape) =======================
+
+#[test] fn bfs_tiny_a()   { bfs_differential(6, 10, 4, &[1, 2, 3, 4]); }
+#[test] fn bfs_tiny_b()   { bfs_differential(5, 8, 5, &[5, 6, 7, 8]); }
+#[test] fn bfs_30_60_20() { bfs_differential(30, 60, 20, &[1, 2, 3, 4]); }
+#[test] fn bfs_15_40_30() { bfs_differential(15, 40, 30, &[5, 6, 7, 8]); }
+#[test] fn bfs_sweep() {
+    for s in 0u8 .. 12 { bfs_differential(40, 120, 30, &[s as usize, (s as usize) + 7, 13, 29]); }
+    for s in 0u8 .. 6  { bfs_differential(60, 200, 25, &[3, (s as usize) + 1, (s as usize) * 5 + 2, 11]); }
+}
+
+type Node = usize;
+type Edge = (Node, Node);
+
+fn bfs_differential(nodes: usize, edges: usize, rounds: usize, seed: &[usize]) {
+    let mut rng1: StdRng = SeedableRng::from_seed(seed);
+    let mut rng2: StdRng = SeedableRng::from_seed(seed);
+    let mut edge_list: Vec<((usize, usize), usize, isize)> = Vec::new();
+    for _ in 0 .. edges { edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), 0, 1)); }
+    for round in 1 .. rounds {
+        edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), round, 1));
+        edge_list.push(((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes)), round, -1));
+    }
+    let root_list = vec![(1usize, 0usize, 1isize)];
+
+    timely::execute_directly(move |worker| {
+        let mut roots_list = root_list.clone();
+        let mut edges_list = edge_list.clone();
+        let (mut roots, mut edges) = worker.dataflow::<usize, _, _>(|scope| {
+            let (root_input, roots) = scope.new_collection();
+            let (edge_input, edges) = scope.new_collection();
+            let via_cursor    = bfs(edges.clone(), roots.clone(), false);
+            let via_reference = bfs(edges, roots, true);
+            via_cursor.assert_eq(via_reference);
+            (root_input, edge_input)
+        });
+        roots_list.sort_by(|x, y| y.1.cmp(&x.1));
+        edges_list.sort_by(|x, y| y.1.cmp(&x.1));
+        let mut round = 0;
+        while !roots_list.is_empty() || !edges_list.is_empty() {
+            while roots_list.last().map(|x| x.1) == Some(round) { let (n, _, d) = roots_list.pop().unwrap(); roots.update(n, d); }
+            while edges_list.last().map(|x| x.1) == Some(round) { let ((s, t), _, d) = edges_list.pop().unwrap(); edges.update((s, t), d); }
+            round += 1;
+            roots.advance_to(round);
+            edges.advance_to(round);
+        }
+    });
+}
+
+fn bfs<'scope, T>(edges: VecCollection<'scope, T, Edge>, roots: VecCollection<'scope, T, Node>, reference: bool) -> VecCollection<'scope, T, (Node, usize)>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let nodes = roots.map(|x| (x, 0));
+    nodes.clone().iterate(|scope, inner| {
+        let edges = edges.enter(scope);
+        let nodes = nodes.enter(scope);
+        let combined = inner.join_map(edges, |_k, l, d| (*d, l + 1)).concat(nodes);
+        if reference {
+            combined.reduce_reference(|_, s, t| t.push((*s[0].0, 1)))
+        } else {
+            combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
+        }
+    })
+}
+
+/// A minimal product-order (partially ordered) timestamp, so the harness exercises the synthetic
+/// interesting times determination is about. Copied from `examples/multitemporal.rs`.
+mod pair {
+
+    #[derive(Hash, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
+    pub struct Pair<S, T> { pub first: S, pub second: T }
+
+    impl<S, T> Pair<S, T> {
+        pub fn new(first: S, second: T) -> Self { Pair { first, second } }
+    }
+
+    use timely::order::PartialOrder;
+    impl<S: PartialOrder, T: PartialOrder> PartialOrder for Pair<S, T> {
+        fn less_equal(&self, other: &Self) -> bool {
+            self.first.less_equal(&other.first) && self.second.less_equal(&other.second)
+        }
+    }
+
+    use timely::progress::timestamp::Refines;
+    impl<S: Timestamp, T: Timestamp> Refines<()> for Pair<S, T> {
+        fn to_inner(_outer: ()) -> Self { Self::minimum() }
+        fn to_outer(self) -> () { () }
+        fn summarize(_summary: <Self>::Summary) -> () { () }
+    }
+
+    use timely::progress::PathSummary;
+    impl<S: Timestamp, T: Timestamp> PathSummary<Pair<S,T>> for () {
+        fn results_in(&self, timestamp: &Pair<S, T>) -> Option<Pair<S,T>> { Some(timestamp.clone()) }
+        fn followed_by(&self, other: &Self) -> Option<Self> { Some(other.clone()) }
+    }
+
+    use timely::progress::Timestamp;
+    impl<S: Timestamp, T: Timestamp> Timestamp for Pair<S, T> {
+        fn minimum() -> Self { Pair { first: S::minimum(), second: T::minimum() }}
+        type Summary = ();
+    }
+
+    use differential_dataflow::lattice::Lattice;
+    impl<S: Lattice, T: Lattice> Lattice for Pair<S, T> {
+        fn join(&self, other: &Self) -> Self {
+            Pair { first: self.first.join(&other.first), second: self.second.join(&other.second) }
+        }
+        fn meet(&self, other: &Self) -> Self {
+            Pair { first: self.first.meet(&other.first), second: self.second.meet(&other.second) }
+        }
+    }
+
+    use std::fmt::{Formatter, Error, Debug};
+    use serde::{Deserialize, Serialize};
+    impl<TOuter: Debug, TInner: Debug> Debug for Pair<TOuter, TInner> {
+        fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
+            f.write_str(&format!("({:?}, {:?})", self.first, self.second))
+        }
+    }
+}

From 7499d38c2fa0aa97c073c6e1f84d45a0bc1872e4 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 18:57:21 -0400
Subject: [PATCH 04/10] reduce: move reference tactic's time buffer out of
 ValueHistory

The reference tactic's time-only walk needs a scratch buffer, but it was
added as a field on the shared ValueHistory, which the standard cursor
tactic allocates and clears per key. Move the buffer to ReferenceThinker
(pooled across keys, as before) and pass it into replay_times, which now
borrows &self. ValueHistory is byte-identical to before the reference tactic.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/src/operators/mod.rs    | 14 ++++++--------
 differential-dataflow/src/operators/reduce.rs | 14 +++++++++++---
 2 files changed, 17 insertions(+), 11 deletions(-)

diff --git a/differential-dataflow/src/operators/mod.rs b/differential-dataflow/src/operators/mod.rs
index 746781daa..a2e0a32b2 100644
--- a/differential-dataflow/src/operators/mod.rs
+++ b/differential-dataflow/src/operators/mod.rs
@@ -91,7 +91,6 @@ struct ValueHistory<V, T, D> {
     edits: EditList<V, T, D>,
     history: Vec<(T, T, usize, usize)>,     // (time, meet, value_index, edit_offset)
     buffer: Vec<((V, T), D)>,               // where we accumulate / collapse updates.
-    time_buffer: Vec<T>,                    // like `buffer`, but times only, for `TimeReplay`.
 }
 
 impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> ValueHistory<V, T, D> {
@@ -100,14 +99,12 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
             edits: EditList::new(),
             history: Vec::new(),
             buffer: Vec::new(),
-            time_buffer: Vec::new(),
         }
     }
     fn clear(&mut self) {
         self.edits.clear();
         self.history.clear();
         self.buffer.clear();
-        self.time_buffer.clear();
     }
 
     /// Loads and replays a specified key.
@@ -138,11 +135,12 @@ impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> V
     }
 
     /// A time-only, non-destructive walk over the already-built `history` (see [`TimeReplay`]). It
-    /// reads times and their precomputed meets and accumulates only *times*, leaving `history`
-    /// intact for a later [`walk`](Self::walk) over values.
-    fn replay_times<'history>(&'history mut self) -> TimeReplay<'history, T> {
-        self.time_buffer.clear();
-        TimeReplay { history: &self.history[..], buffer: &mut self.time_buffer }
+    /// reads times and their precomputed meets and accumulates only *times* into the caller-supplied
+    /// `buffer`, leaving `history` intact for a later [`walk`](Self::walk) over values. The buffer is
+    /// owned by the caller so the standard value walk pays nothing for this reference-only capability.
+    fn replay_times<'history>(&'history self, buffer: &'history mut Vec<T>) -> TimeReplay<'history, T> {
+        buffer.clear();
+        TimeReplay { history: &self.history[..], buffer }
     }
 
     /// Organizes history based on current contents of edits.
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index 280a7b5e7..882ab9b30 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -1052,6 +1052,11 @@ mod reference {
         times_current: Vec<T>,
         temporary: Vec<T>,
         meets: Vec<T>,
+        // Reusable time-only buffers for phase 1's `TimeReplay` walks. Pooled here (rather than in
+        // `ValueHistory`) so the reference tactic pays for them and the standard value walk does not.
+        batch_times: Vec<T>,
+        input_times: Vec<T>,
+        output_times: Vec<T>,
         // The interesting (in-band reached) times, handed from phase 1 to phase 2.
         active: Vec<T>,
     }
@@ -1078,6 +1083,9 @@ mod reference {
                 times_current: Vec::new(),
                 temporary: Vec::new(),
                 meets: Vec::new(),
+                batch_times: Vec::new(),
+                input_times: Vec::new(),
+                output_times: Vec::new(),
                 active: Vec::new(),
             }
         }
@@ -1122,9 +1130,9 @@ mod reference {
                 drop(self.batch_history.replay_key(batch_cursor, batch_storage, key, None));
                 drop(self.input_history.replay_key(source_cursor, source_storage, key, None));
                 drop(self.output_history.replay_key(output_cursor, output_storage, key, None));
-                let mut batch_replay = self.batch_history.replay_times();
-                let mut input_replay = self.input_history.replay_times();
-                let mut output_replay = self.output_history.replay_times();
+                let mut batch_replay = self.batch_history.replay_times(&mut self.batch_times);
+                let mut input_replay = self.input_history.replay_times(&mut self.input_times);
+                let mut output_replay = self.output_history.replay_times(&mut self.output_times);
 
                 self.synth_times.clear();
                 self.times_current.clear();

From d56cf94d034944e8f71a38dc7445d442fca1912b Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 19:09:47 -0400
Subject: [PATCH 05/10] reduce: hide reference entry points from the public API

The reference tactic is a testing and demonstration oracle, not something to
build on. Mark its entry points `#[doc(hidden)]` so they leave the documented
surface without inviting downstream reliance: Collection::{reduce_reference,
reduce_named_reference}, Arranged::{reduce_abelian_reference,
reduce_core_reference}, and reduce_trace_reference. Still callable, so the
differential and oracle tests are unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/src/collection.rs                   | 8 ++++++++
 .../src/operators/arrange/arrangement.rs                  | 8 ++++++++
 differential-dataflow/src/operators/reduce.rs             | 4 ++++
 3 files changed, 20 insertions(+)

diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs
index 6eed46760..ffb5f5b83 100644
--- a/differential-dataflow/src/collection.rs
+++ b/differential-dataflow/src/collection.rs
@@ -759,12 +759,20 @@ pub mod vec {
 
         /// As [`reduce`](Self::reduce), but driven by the model-derived reference tactic. Same result;
         /// intended for differential testing against the default reduce.
+        ///
+        /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
+        /// not a stable entry point to build on.
+        #[doc(hidden)]
         pub fn reduce_reference<L, V2: crate::Data, R2: Ord+Abelian+'static>(self, logic: L) -> Collection<'scope, T, (K, V2), R2>
         where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>)+'static {
             self.reduce_named_reference("Reduce", logic)
         }
 
         /// As [`reduce_named`](Self::reduce_named), but driven by the model-derived reference tactic.
+        ///
+        /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
+        /// not a stable entry point to build on.
+        #[doc(hidden)]
         pub fn reduce_named_reference<L, V2: crate::Data, R2: Ord+Abelian+'static>(self, name: &str, logic: L) -> Collection<'scope, T, (K, V2), R2>
         where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>)+'static {
             use crate::trace::implementations::{ValBuilder, ValSpine};
diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index 9445f95bc..01d290fb1 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -296,6 +296,10 @@ impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
     }
 
     /// As [`reduce_abelian`](Self::reduce_abelian), but using the model-derived reference tactic.
+    ///
+    /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
+    /// not a stable entry point to build on.
+    #[doc(hidden)]
     pub fn reduce_abelian_reference<L, Bu, Tr2, P>(self, name: &str, mut logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
     where
         Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
@@ -315,6 +319,10 @@ impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
     }
 
     /// As [`reduce_core`](Self::reduce_core), but using the model-derived reference tactic.
+    ///
+    /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
+    /// not a stable entry point to build on.
+    #[doc(hidden)]
     pub fn reduce_core_reference<L, Bu, Tr2, P>(self, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
     where
         Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index 882ab9b30..e6a00d987 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -95,6 +95,10 @@ where
 /// As [`reduce_trace`], but driven by the model-derived [`reference::ReferenceTactic`] instead of the
 /// default `cursors::CursorTactic`. Same result contract; intended for differential testing of the
 /// two tactics against each other.
+///
+/// Hidden from the public API: the reference tactic is a testing and demonstration oracle, not a
+/// stable entry point to build on.
+#[doc(hidden)]
 pub fn reduce_trace_reference<'scope, Tr1, Bu, Tr2, L, P>(trace: Arranged<'scope, Tr1>, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
 where
     Tr1: TraceReader<Batch: Navigable> + 'static,

From 9ef46017769f0888209a02ae10e80eb0f94e05af Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 19:24:30 -0400
Subject: [PATCH 06/10] reduce: drop reference tactic's inherent methods from
 Collection/Arranged

The reference tactic is a testing and demonstration oracle. Inherent methods
on the stable public types (Collection::{reduce_reference, reduce_named_reference},
Arranged::{reduce_abelian_reference, reduce_core_reference}) commit us to them
under semver and invite reliance, so remove them. The one standalone entry
reduce_trace_reference stays (doc-hidden). The arrange / abelian-negate /
as_collection glue those methods provided moves into a local `reduce_reference`
helper in the test and the benchmark, the only callers.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../examples/reduce_bench.rs                  | 47 +++++++++++++++--
 differential-dataflow/src/collection.rs       | 29 -----------
 .../src/operators/arrange/arrangement.rs      | 40 --------------
 .../tests/reduce_reference.rs                 | 52 +++++++++++++++++--
 4 files changed, 91 insertions(+), 77 deletions(-)

diff --git a/differential-dataflow/examples/reduce_bench.rs b/differential-dataflow/examples/reduce_bench.rs
index 5a74b0719..b91b136bd 100644
--- a/differential-dataflow/examples/reduce_bench.rs
+++ b/differential-dataflow/examples/reduce_bench.rs
@@ -10,14 +10,55 @@ use std::time::Instant;
 
 use rand::{Rng, SeedableRng, StdRng};
 
+use differential_dataflow::{Data, ExchangeData, Hashable};
 use differential_dataflow::input::Input;
 use differential_dataflow::VecCollection;
+use differential_dataflow::difference::{Abelian, Semigroup};
 use differential_dataflow::operators::*;
+use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+use differential_dataflow::operators::reduce::reduce_trace_reference;
+use differential_dataflow::trace::implementations::{ValBuilder, ValSpine};
 use differential_dataflow::lattice::Lattice;
 
 type Node = usize;
 type Edge = (Node, Node);
 
+// Stand-alone equivalent of the default `reduce`, driven by the model-derived reference tactic. The
+// library exposes only the low-level, doc-hidden `reduce_trace_reference`; this helper supplies the
+// same arrange / abelian-negate / as_collection glue `reduce_named` uses, so the benchmark drives the
+// reference tactic without it living on the public `Collection`/`Arranged` API.
+fn reduce_reference<'scope, Ti, K, V, R, V2, R2, L>(
+    collection: VecCollection<'scope, Ti, (K, V), R>,
+    mut logic: L,
+) -> VecCollection<'scope, Ti, (K, V2), R2>
+where
+    Ti: timely::progress::Timestamp + Lattice + Ord,
+    K: ExchangeData + Hashable,
+    V: ExchangeData,
+    R: ExchangeData + Semigroup,
+    V2: Data,
+    R2: Ord + Abelian + 'static,
+    L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,
+{
+    // See tests/reduce_reference.rs for why `Tr1` and the spine/builder types are pinned explicitly:
+    // as a free function the output-key bound is checked before the argument would infer `Tr1`, and
+    // the closure params are annotated to the concrete `&K`/`&V` types so the user-logic coercion
+    // normalizes here, exactly as the default `reduce_named` gets for free.
+    let arranged: Arranged<'scope, TraceAgent<ValSpine<K, V, Ti, R>>> =
+        collection.arrange_by_key_named("Arrange: ReduceReference");
+    reduce_trace_reference::<TraceAgent<ValSpine<K, V, Ti, R>>, ValBuilder<K, V2, Ti, R2>, ValSpine<K, V2, Ti, R2>, _, _>(
+        arranged,
+        "ReduceReference",
+        move |key: &K, input: &[(&V, R)], output: &mut Vec<(V2, R2)>, change: &mut Vec<(V2, R2)>| {
+            if !input.is_empty() { logic(key, input, change); }
+            change.extend(output.drain(..).map(|(x, mut d)| { d.negate(); (x, d) }));
+            differential_dataflow::consolidation::consolidate(change);
+        },
+        |vec, key: &K, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v, t, r)| ((key.clone(), v), t, r))); },
+    )
+    .as_collection(|k, v| (k.clone(), v.clone()))
+}
+
 fn main() {
     for _ in 0 .. 2 {
         let c = reduce_churn(false);
@@ -66,7 +107,7 @@ fn reduce_churn(reference: bool) -> std::time::Duration {
         let mut input = worker.dataflow::<usize, _, _>(|scope| {
             let (input, coll) = scope.new_collection::<(usize, usize), isize>();
             let reduced = if reference {
-                coll.reduce_reference(|_k, s, t| t.push((s.len() as isize, 1)))
+                reduce_reference(coll, |_k, s, t| t.push((s.len() as isize, 1)))
             } else {
                 coll.reduce(|_k, s, t| t.push((s.len() as isize, 1)))
             };
@@ -105,7 +146,7 @@ fn fewkeys(reference: bool) -> std::time::Duration {
         let mut input = worker.dataflow::<usize, _, _>(|scope| {
             let (input, coll) = scope.new_collection::<(usize, usize), isize>();
             let reduced = if reference {
-                coll.reduce_reference(|_k, s, t| t.push((s.len() as isize, 1)))
+                reduce_reference(coll, |_k, s, t| t.push((s.len() as isize, 1)))
             } else {
                 coll.reduce(|_k, s, t| t.push((s.len() as isize, 1)))
             };
@@ -180,7 +221,7 @@ where
         let nodes = nodes.enter(scope);
         let combined = inner.join_map(edges, |_k, l, d| (*d, l + 1)).concat(nodes);
         if reference {
-            combined.reduce_reference(|_, s, t| t.push((*s[0].0, 1)))
+            reduce_reference(combined, |_, s, t| t.push((*s[0].0, 1)))
         } else {
             combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
         }
diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs
index ffb5f5b83..ebf634ff2 100644
--- a/differential-dataflow/src/collection.rs
+++ b/differential-dataflow/src/collection.rs
@@ -757,35 +757,6 @@ pub mod vec {
                 .as_collection(|k,v| (k.clone(), v.clone()))
         }
 
-        /// As [`reduce`](Self::reduce), but driven by the model-derived reference tactic. Same result;
-        /// intended for differential testing against the default reduce.
-        ///
-        /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
-        /// not a stable entry point to build on.
-        #[doc(hidden)]
-        pub fn reduce_reference<L, V2: crate::Data, R2: Ord+Abelian+'static>(self, logic: L) -> Collection<'scope, T, (K, V2), R2>
-        where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>)+'static {
-            self.reduce_named_reference("Reduce", logic)
-        }
-
-        /// As [`reduce_named`](Self::reduce_named), but driven by the model-derived reference tactic.
-        ///
-        /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
-        /// not a stable entry point to build on.
-        #[doc(hidden)]
-        pub fn reduce_named_reference<L, V2: crate::Data, R2: Ord+Abelian+'static>(self, name: &str, logic: L) -> Collection<'scope, T, (K, V2), R2>
-        where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>)+'static {
-            use crate::trace::implementations::{ValBuilder, ValSpine};
-
-            self.arrange_by_key_named(&format!("Arrange: {}", name))
-                .reduce_abelian_reference::<_,ValBuilder<_,_,_,_>,ValSpine<K,V2,_,_>,_>(
-                    name,
-                    logic,
-                    |vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v,t,r)| ((key.clone(), v),t,r))); },
-                )
-                .as_collection(|k,v| (k.clone(), v.clone()))
-        }
-
         /// Applies `reduce` to arranged data, and returns an arrangement of output data.
         ///
         /// This method is used by the more ergonomic `reduce`, `distinct`, and `count` methods, although
diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index 01d290fb1..b5a8ea344 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -295,46 +295,6 @@ impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
         reduce_trace::<_,Bu,_,_,_>(self, name, logic, push)
     }
 
-    /// As [`reduce_abelian`](Self::reduce_abelian), but using the model-derived reference tactic.
-    ///
-    /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
-    /// not a stable entry point to build on.
-    #[doc(hidden)]
-    pub fn reduce_abelian_reference<L, Bu, Tr2, P>(self, name: &str, mut logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
-    where
-        Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
-        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
-        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time, Diff: Abelian>,
-        Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default> + 'static,
-        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
-        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
-    {
-        self.reduce_core_reference::<_,Bu,Tr2,_>(name, move |key, input, output, change| {
-            if !input.is_empty() {
-                logic(key, input, change);
-            }
-            change.extend(output.drain(..).map(|(x,mut d)| { d.negate(); (x, d) }));
-            crate::consolidation::consolidate(change);
-        }, push)
-    }
-
-    /// As [`reduce_core`](Self::reduce_core), but using the model-derived reference tactic.
-    ///
-    /// Hidden from the public API: the reference tactic is a testing and demonstration oracle,
-    /// not a stable entry point to build on.
-    #[doc(hidden)]
-    pub fn reduce_core_reference<L, Bu, Tr2, P>(self, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
-    where
-        Tr2: Trace<Batch: Navigable, Time=Tr1::Time>+'static,
-        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
-        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
-        Bu: Builder<Time=Tr1::Time, Output = Tr2::Batch, Input: Default> + 'static,
-        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
-        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
-    {
-        use crate::operators::reduce::reduce_trace_reference;
-        reduce_trace_reference::<_,Bu,_,_,_>(self, name, logic, push)
-    }
 }
 
 
diff --git a/differential-dataflow/tests/reduce_reference.rs b/differential-dataflow/tests/reduce_reference.rs
index f4720f63c..6f8a1fb04 100644
--- a/differential-dataflow/tests/reduce_reference.rs
+++ b/differential-dataflow/tests/reduce_reference.rs
@@ -25,11 +25,15 @@ use timely::PartialOrder;
 use timely::dataflow::operators::capture::{Capture, Extract};
 use timely::dataflow::operators::vec::unordered_input::UnorderedInput;
 
-use differential_dataflow::AsCollection;
+use differential_dataflow::{AsCollection, Data, ExchangeData, Hashable};
 use differential_dataflow::VecCollection;
+use differential_dataflow::difference::{Abelian, Semigroup};
 use differential_dataflow::input::Input;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::*;
+use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+use differential_dataflow::operators::reduce::reduce_trace_reference;
+use differential_dataflow::trace::implementations::{ValBuilder, ValSpine};
 
 use pair::Pair;
 type Time = Pair<u64, u64>;
@@ -51,6 +55,44 @@ fn min(_key: &u64, input: &[(&u64, isize)], output: &mut Vec<(i64, isize)>) {
 
 type Logic = fn(&u64, &[(&u64, isize)], &mut Vec<(i64, isize)>);
 
+// Stand-alone equivalent of the default `reduce`, driven by the model-derived reference tactic. The
+// library exposes only the low-level, doc-hidden `reduce_trace_reference`; this helper supplies the
+// same arrange / abelian-negate / as_collection glue `reduce_named` uses, so the tests compare the
+// two tactics without the reference living on the public `Collection`/`Arranged` API.
+fn reduce_reference<'scope, Ti, K, V, R, V2, R2, L>(
+    collection: VecCollection<'scope, Ti, (K, V), R>,
+    mut logic: L,
+) -> VecCollection<'scope, Ti, (K, V2), R2>
+where
+    Ti: timely::progress::Timestamp + Lattice + Ord,
+    K: ExchangeData + Hashable,
+    V: ExchangeData,
+    R: ExchangeData + Semigroup,
+    V2: Data,
+    R2: Ord + Abelian + 'static,
+    L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,
+{
+    // Bind the input arrangement to an explicitly-typed local so `Tr1` is concrete going into
+    // `reduce_trace_reference` (otherwise its output-key bound is checked before the argument pins it).
+    let arranged: Arranged<'scope, TraceAgent<ValSpine<K, V, Ti, R>>> =
+        collection.arrange_by_key_named("Arrange: ReduceReference");
+
+    // The closure params are annotated to the concrete `&K` / `&V` / owned-output types (rather than
+    // left to infer the cursor's `Key<'_>` projection) so the coercion to the user's `&K` logic
+    // normalizes here, exactly as the default `reduce_named` gets for free by passing its logic through.
+    reduce_trace_reference::<TraceAgent<ValSpine<K, V, Ti, R>>, ValBuilder<K, V2, Ti, R2>, ValSpine<K, V2, Ti, R2>, _, _>(
+        arranged,
+        "ReduceReference",
+        move |key: &K, input: &[(&V, R)], output: &mut Vec<(V2, R2)>, change: &mut Vec<(V2, R2)>| {
+            if !input.is_empty() { logic(key, input, change); }
+            change.extend(output.drain(..).map(|(x, mut d)| { d.negate(); (x, d) }));
+            differential_dataflow::consolidation::consolidate(change);
+        },
+        |vec, key: &K, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v, t, r)| ((key.clone(), v), t, r))); },
+    )
+    .as_collection(|k, v| (k.clone(), v.clone()))
+}
+
 // A random input: `((key, val), Pair(a, b), diff)`, diffs ±1 so updates cancel.
 fn random_input(seed: usize, n: usize, keys: u64, vals: u64, span: u64) -> Vec<((u64, u64), Time, isize)> {
     let s: &[usize] = &[seed, 7, 13, 29];
@@ -78,7 +120,7 @@ fn differential(seed: usize, logic: Logic) {
             let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
             let collection = stream.as_collection();
             let via_cursor    = collection.clone().reduce(logic);
-            let via_reference = collection.reduce_reference(logic);
+            let via_reference = reduce_reference(collection, logic);
             via_cursor.assert_eq(via_reference);
             (input, capability)
         });
@@ -107,7 +149,7 @@ fn oracle(seed: usize, reference: bool, logic: Logic) {
         let (mut input, capability) = worker.dataflow::<Time, _, _>(|scope| {
             let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
             let collection = stream.as_collection();
-            let out = if reference { collection.reduce_reference(logic) } else { collection.reduce(logic) };
+            let out = if reference { reduce_reference(collection, logic) } else { collection.reduce(logic) };
             out.inner.capture_into(send);
             (input, capability)
         });
@@ -180,7 +222,7 @@ fn over_derivation() {
                 let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
                 let collection = stream.as_collection();
                 collection.clone().reduce(count).inspect(|_| {});     // drive the cursor tactic
-                collection.reduce_reference(count).inspect(|_| {});   // drive the reference tactic
+                reduce_reference(collection, count).inspect(|_| {});  // drive the reference tactic
                 (input, capability)
             });
             for u in updates { input.activate().session(&capability).give(u); }
@@ -252,7 +294,7 @@ where
         let nodes = nodes.enter(scope);
         let combined = inner.join_map(edges, |_k, l, d| (*d, l + 1)).concat(nodes);
         if reference {
-            combined.reduce_reference(|_, s, t| t.push((*s[0].0, 1)))
+            reduce_reference(combined, |_, s, t| t.push((*s[0].0, 1)))
         } else {
             combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
         }

From bf93ec2ed38126426cc15c80eb98c4c6a42089ab Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 19:34:09 -0400
Subject: [PATCH 07/10] reduce: consolidate reference tactic into mod
 reference; demote metrics

Move reduce_trace_reference into `pub(crate) mod reference` alongside the tactic
it constructs, and re-export it (doc-hidden) from the parent as the sole public
handle its out-of-crate tests need. Move the reduce-metrics counters off the top
of the file to the bottom; they are an optional diagnostic, not the headline.
Also drop a stray blank line left in arrangement.rs by the earlier removal.

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

diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs
index b5a8ea344..bb7f74f6f 100644
--- a/differential-dataflow/src/operators/arrange/arrangement.rs
+++ b/differential-dataflow/src/operators/arrange/arrangement.rs
@@ -294,10 +294,8 @@ impl<'scope, Tr1: TraceReader<Batch: Navigable>+'static> Arranged<'scope, Tr1> {
         use crate::operators::reduce::reduce_trace;
         reduce_trace::<_,Bu,_,_,_>(self, name, logic, push)
     }
-
 }
 
-
 impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
     /// Brings an arranged collection out of a nested region.
     ///
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index e6a00d987..c682452e5 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -24,25 +24,6 @@ use crate::trace::{BatchCursor, BatchDiff, BatchKey, BatchReader, BatchVal, Batc
 use crate::trace::cursor::cursor_list;
 use crate::trace::implementations::containers::BatchContainer;
 
-/// Optional interesting-time counters. They measure how many in-band interesting times each tactic
-/// evaluates, and hence how much the value-blind `reference` over-derives relative to the value-aware
-/// cursor (whose consolidation drops the zero-debt addresses the reference keeps). Off unless built
-/// with `--features reduce-metrics`; the increments compile to nothing otherwise.
-#[cfg(feature = "reduce-metrics")]
-pub mod metrics {
-    use std::sync::atomic::{AtomicUsize, Ordering};
-    /// In-band interesting times evaluated by the default `cursors::CursorTactic`.
-    pub static CURSOR: AtomicUsize = AtomicUsize::new(0);
-    /// In-band interesting times evaluated by the model-derived `reference::ReferenceTactic`.
-    pub static REFERENCE: AtomicUsize = AtomicUsize::new(0);
-    /// Reset both counters to zero.
-    pub fn reset() { CURSOR.store(0, Ordering::Relaxed); REFERENCE.store(0, Ordering::Relaxed); }
-    /// Read the cursor tactic's count.
-    pub fn cursor() -> usize { CURSOR.load(Ordering::Relaxed) }
-    /// Read the reference tactic's count.
-    pub fn reference() -> usize { REFERENCE.load(Ordering::Relaxed) }
-}
-
 /// A type that resolves a key-wise reduction over batches arriving on the input.
 ///
 /// Unlike join, reduce does not suspend: its output is at most linear in its input, so a single
@@ -92,25 +73,10 @@ where
     reduce_with_tactic(trace, name, cursors::CursorTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
 }
 
-/// As [`reduce_trace`], but driven by the model-derived [`reference::ReferenceTactic`] instead of the
-/// default `cursors::CursorTactic`. Same result contract; intended for differential testing of the
-/// two tactics against each other.
-///
-/// Hidden from the public API: the reference tactic is a testing and demonstration oracle, not a
-/// stable entry point to build on.
+// The model-derived reference tactic and its entry point live in `mod reference`; re-exported here
+// (doc-hidden) as the sole public handle for its differential and oracle tests.
 #[doc(hidden)]
-pub fn reduce_trace_reference<'scope, Tr1, Bu, Tr2, L, P>(trace: Arranged<'scope, Tr1>, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
-where
-    Tr1: TraceReader<Batch: Navigable> + 'static,
-    Tr2: Trace<Batch: Navigable, Time = Tr1::Time> + 'static,
-    BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
-    for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
-    Bu: Builder<Time=Tr2::Time, Output = Tr2::Batch, Input: Default> + 'static,
-    L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
-    P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
-{
-    reduce_with_tactic(trace, name, reference::ReferenceTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
-}
+pub use reference::reduce_trace_reference;
 
 /// Drives a key-wise reduction using a supplied [`ReduceTactic`].
 ///
@@ -822,12 +788,30 @@ mod cursors {
 /// the corrections — the model's `emit_correct`. Determination never consults the output produced
 /// this round (it finishes first), so this tactic embodies the proven algorithm exactly and is the
 /// clean subject for differential testing against [`cursors::CursorTactic`].
-mod reference {
+pub(crate) mod reference {
 
     use super::*;
     use crate::lattice::Lattice;
     use crate::operators::ValueHistory;
 
+    /// Drives a key-wise reduction with the model-derived [`ReferenceTactic`], the analogue of the
+    /// default [`super::reduce_trace`]. Same result contract; intended for differential testing of the
+    /// two tactics against each other. Re-exported (doc-hidden) from the parent module as the sole
+    /// public handle: the reference tactic is a testing and demonstration oracle, not a stable entry
+    /// point to build on.
+    pub fn reduce_trace_reference<'scope, Tr1, Bu, Tr2, L, P>(trace: Arranged<'scope, Tr1>, name: &str, logic: L, push: P) -> Arranged<'scope, TraceAgent<Tr2>>
+    where
+        Tr1: TraceReader<Batch: Navigable> + 'static,
+        Tr2: Trace<Batch: Navigable, Time = Tr1::Time> + 'static,
+        BatchCursor<Tr1>: Cursor<Time = Tr1::Time>,
+        for<'a> BatchCursor<Tr2>: Cursor<Key<'a> = BatchKey<'a, Tr1>, ValOwn: Data, Time = Tr2::Time>,
+        Bu: Builder<Time=Tr2::Time, Output = Tr2::Batch, Input: Default> + 'static,
+        L: FnMut(BatchKey<'_, Tr1>, &[(BatchVal<'_, Tr1>, BatchDiff<Tr1>)], &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>, &mut Vec<(BatchValOwn<Tr2>, BatchDiff<Tr2>)>)+'static,
+        P: FnMut(&mut Bu::Input, BatchKey<'_, Tr1>, &mut Vec<(BatchValOwn<Tr2>, Tr2::Time, BatchDiff<Tr2>)>) + 'static,
+    {
+        reduce_with_tactic(trace, name, ReferenceTactic::<Tr1::Batch, Tr2::Batch, Bu, L, P>::new(logic, push))
+    }
+
     /// Sorts and deduplicates, matching `cursors::sort_dedup`.
     #[inline(never)]
     fn sort_dedup<T: Ord>(list: &mut Vec<T>) {
@@ -1298,3 +1282,22 @@ mod reference {
         }
     }
 }
+
+/// Optional interesting-time counters. They measure how many in-band interesting times each tactic
+/// evaluates, and hence how much the value-blind `reference` over-derives relative to the value-aware
+/// cursor (whose consolidation drops the zero-debt addresses the reference keeps). Off unless built
+/// with `--features reduce-metrics`; the increments compile to nothing otherwise.
+#[cfg(feature = "reduce-metrics")]
+pub mod metrics {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    /// In-band interesting times evaluated by the default `cursors::CursorTactic`.
+    pub static CURSOR: AtomicUsize = AtomicUsize::new(0);
+    /// In-band interesting times evaluated by the model-derived `reference::ReferenceTactic`.
+    pub static REFERENCE: AtomicUsize = AtomicUsize::new(0);
+    /// Reset both counters to zero.
+    pub fn reset() { CURSOR.store(0, Ordering::Relaxed); REFERENCE.store(0, Ordering::Relaxed); }
+    /// Read the cursor tactic's count.
+    pub fn cursor() -> usize { CURSOR.load(Ordering::Relaxed) }
+    /// Read the reference tactic's count.
+    pub fn reference() -> usize { REFERENCE.load(Ordering::Relaxed) }
+}

From 52f504d83ceb723a9e438bec36b3fc3ad3c4a6d6 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 19:45:25 -0400
Subject: [PATCH 08/10] reduce: remove the reduce-metrics counters and
 over_derivation test

The interesting-time counters served only the `over_derivation` characterization
test, which measured (not verified) the value-blind reference's over-derivation
and asserted a non-correctness property. They cost a Cargo feature, two global
AtomicUsize statics, and two fetch_add lines in both tactics' per-key hot loops
(and would silently sum across workers). Correctness is already covered by the
differential and oracle tests. Drop the feature, the metrics module, the
instrumentation, and the test; the measurement stays reproducible via ad-hoc
instrumentation when wanted.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/Cargo.toml              |  2 -
 differential-dataflow/src/operators/reduce.rs | 25 -----------
 .../tests/reduce_reference.rs                 | 44 +------------------
 3 files changed, 1 insertion(+), 70 deletions(-)

diff --git a/differential-dataflow/Cargo.toml b/differential-dataflow/Cargo.toml
index ae9a2b0da..3dbace3c7 100644
--- a/differential-dataflow/Cargo.toml
+++ b/differential-dataflow/Cargo.toml
@@ -39,8 +39,6 @@ timely = {workspace = true}
 
 [features]
 default = ["timely/getopts"]
-# Enables the reduce interesting-time counters in `operators::reduce::metrics`.
-reduce-metrics = []
 
 [[bench]]
 name = "chunk_bench"
diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index c682452e5..59aa17880 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -675,9 +675,6 @@ mod cursors {
                         // output produced. This sounds like a good test to have for debug builds!
                         if interesting {
 
-                            #[cfg(feature = "reduce-metrics")]
-                            crate::operators::reduce::metrics::CURSOR.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-
                             // Assemble the input collection at `next_time`. (`self.input_buffer` cleared just after use).
                             // The buffers were advanced by `meet` in the determination step above.
                             debug_assert!(self.input_buffer.is_empty());
@@ -1202,9 +1199,6 @@ pub(crate) mod reference {
                 }
 
                 sort_dedup(&mut self.active);
-
-                #[cfg(feature = "reduce-metrics")]
-                crate::operators::reduce::metrics::REFERENCE.fetch_add(self.active.len(), std::sync::atomic::Ordering::Relaxed);
             }
 
             // ===================== PHASE 2 — APPLICATION (`emit_correct`) =====================
@@ -1282,22 +1276,3 @@ pub(crate) mod reference {
         }
     }
 }
-
-/// Optional interesting-time counters. They measure how many in-band interesting times each tactic
-/// evaluates, and hence how much the value-blind `reference` over-derives relative to the value-aware
-/// cursor (whose consolidation drops the zero-debt addresses the reference keeps). Off unless built
-/// with `--features reduce-metrics`; the increments compile to nothing otherwise.
-#[cfg(feature = "reduce-metrics")]
-pub mod metrics {
-    use std::sync::atomic::{AtomicUsize, Ordering};
-    /// In-band interesting times evaluated by the default `cursors::CursorTactic`.
-    pub static CURSOR: AtomicUsize = AtomicUsize::new(0);
-    /// In-band interesting times evaluated by the model-derived `reference::ReferenceTactic`.
-    pub static REFERENCE: AtomicUsize = AtomicUsize::new(0);
-    /// Reset both counters to zero.
-    pub fn reset() { CURSOR.store(0, Ordering::Relaxed); REFERENCE.store(0, Ordering::Relaxed); }
-    /// Read the cursor tactic's count.
-    pub fn cursor() -> usize { CURSOR.load(Ordering::Relaxed) }
-    /// Read the reference tactic's count.
-    pub fn reference() -> usize { REFERENCE.load(Ordering::Relaxed) }
-}
diff --git a/differential-dataflow/tests/reduce_reference.rs b/differential-dataflow/tests/reduce_reference.rs
index 6f8a1fb04..f8f65ff40 100644
--- a/differential-dataflow/tests/reduce_reference.rs
+++ b/differential-dataflow/tests/reduce_reference.rs
@@ -1,4 +1,4 @@
-//! Test harness for the model-derived `reference` reduce tactic. Four things:
+//! Test harness for the model-derived `reference` reduce tactic. Three things:
 //!
 //! * `differential_*` — run the default (cursor) reduce and the `reference` reduce on the SAME random
 //!   input and `assert_eq` their outputs at every time. The drift detector: if either tactic changes
@@ -13,9 +13,6 @@
 //!
 //! * `bfs_*` — an iterative BFS (product time via iteration; a real computation shape) computed both
 //!   ways over a graph that grows and shrinks; another differential check.
-//!
-//! * `over_derivation` — with `--features reduce-metrics`, count how many interesting times each
-//!   tactic evaluates on a cancellation-heavy input, to measure the value-blind over-derivation.
 
 use std::collections::BTreeMap;
 
@@ -196,45 +193,6 @@ fn oracle(seed: usize, reference: bool, logic: Logic) {
     }
 }
 
-// ============= over-derivation: interesting-time counts, cursor vs reference =============
-// Run with:  cargo test -p differential-dataflow --features reduce-metrics over_derivation -- --nocapture
-
-#[cfg(feature = "reduce-metrics")]
-#[test]
-fn over_derivation() {
-    use differential_dataflow::operators::reduce::metrics;
-    // Cancellation-heavy configs: few keys/values over dense product-time grids with ±1 diffs, so many
-    // updates advance-cancel — the cursor drops those (zero-debt) addresses, the reference keeps them.
-    // Sweep densities to look for any case where the value-blind reference derives more.
-    let configs = [
-        (1, 1200usize, 4u64, 3u64, 10u64),
-        (2, 1500, 3, 2, 12),
-        (3, 2000, 2, 2, 8),
-        (4, 1000, 6, 4, 10),
-        (5, 1800, 2, 3, 14),
-    ];
-    println!();
-    for (seed, n, keys, vals, span) in configs {
-        let updates = random_input(seed, n, keys, vals, span);
-        metrics::reset();
-        timely::execute_directly(move |worker| {
-            let (mut input, capability) = worker.dataflow::<Time, _, _>(|scope| {
-                let ((input, capability), stream) = scope.new_unordered_input::<((u64, u64), Time, isize)>();
-                let collection = stream.as_collection();
-                collection.clone().reduce(count).inspect(|_| {});     // drive the cursor tactic
-                reduce_reference(collection, count).inspect(|_| {});  // drive the reference tactic
-                (input, capability)
-            });
-            for u in updates { input.activate().session(&capability).give(u); }
-            drop(capability);
-        });
-        let (c, r) = (metrics::cursor(), metrics::reference());
-        println!("  keys={keys} vals={vals} span={span} n={n}: cursor {c}, reference {r}  ({:.2}x)",
-                 r as f64 / (c.max(1)) as f64);
-        assert!(r >= c, "value-blind reference should derive at least as many interesting times as the cursor");
-    }
-}
-
 // ======================= iterative BFS differential (a real shape) =======================
 
 #[test] fn bfs_tiny_a()   { bfs_differential(6, 10, 4, &[1, 2, 3, 4]); }

From 4e03613f071ce55a19e61e45221b3c47f004b593 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 19:53:19 -0400
Subject: [PATCH 09/10] reduce: drop the reduce_bench example

It existed only to race the default cursor tactic against the reference tactic.
The reference is a testing and demonstration oracle, not a shipped code path, so
a standing benchmark of it does not belong in `examples/`. Remove it.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../examples/reduce_bench.rs                  | 229 ------------------
 1 file changed, 229 deletions(-)
 delete mode 100644 differential-dataflow/examples/reduce_bench.rs

diff --git a/differential-dataflow/examples/reduce_bench.rs b/differential-dataflow/examples/reduce_bench.rs
deleted file mode 100644
index b91b136bd..000000000
--- a/differential-dataflow/examples/reduce_bench.rs
+++ /dev/null
@@ -1,229 +0,0 @@
-//! Wall-clock comparison of the default cursor reduce tactic vs the model-derived reference tactic.
-//!
-//!   cargo run --release --example reduce_bench
-//!
-//! Two workloads: a reduce-dominated 1-D (total-order) churn, which isolates the base per-key engine
-//! overhead (determination is trivial with total orders), and an iterative BFS with Product times,
-//! where the synthetic-time determination actually does work.
-
-use std::time::Instant;
-
-use rand::{Rng, SeedableRng, StdRng};
-
-use differential_dataflow::{Data, ExchangeData, Hashable};
-use differential_dataflow::input::Input;
-use differential_dataflow::VecCollection;
-use differential_dataflow::difference::{Abelian, Semigroup};
-use differential_dataflow::operators::*;
-use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
-use differential_dataflow::operators::reduce::reduce_trace_reference;
-use differential_dataflow::trace::implementations::{ValBuilder, ValSpine};
-use differential_dataflow::lattice::Lattice;
-
-type Node = usize;
-type Edge = (Node, Node);
-
-// Stand-alone equivalent of the default `reduce`, driven by the model-derived reference tactic. The
-// library exposes only the low-level, doc-hidden `reduce_trace_reference`; this helper supplies the
-// same arrange / abelian-negate / as_collection glue `reduce_named` uses, so the benchmark drives the
-// reference tactic without it living on the public `Collection`/`Arranged` API.
-fn reduce_reference<'scope, Ti, K, V, R, V2, R2, L>(
-    collection: VecCollection<'scope, Ti, (K, V), R>,
-    mut logic: L,
-) -> VecCollection<'scope, Ti, (K, V2), R2>
-where
-    Ti: timely::progress::Timestamp + Lattice + Ord,
-    K: ExchangeData + Hashable,
-    V: ExchangeData,
-    R: ExchangeData + Semigroup,
-    V2: Data,
-    R2: Ord + Abelian + 'static,
-    L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,
-{
-    // See tests/reduce_reference.rs for why `Tr1` and the spine/builder types are pinned explicitly:
-    // as a free function the output-key bound is checked before the argument would infer `Tr1`, and
-    // the closure params are annotated to the concrete `&K`/`&V` types so the user-logic coercion
-    // normalizes here, exactly as the default `reduce_named` gets for free.
-    let arranged: Arranged<'scope, TraceAgent<ValSpine<K, V, Ti, R>>> =
-        collection.arrange_by_key_named("Arrange: ReduceReference");
-    reduce_trace_reference::<TraceAgent<ValSpine<K, V, Ti, R>>, ValBuilder<K, V2, Ti, R2>, ValSpine<K, V2, Ti, R2>, _, _>(
-        arranged,
-        "ReduceReference",
-        move |key: &K, input: &[(&V, R)], output: &mut Vec<(V2, R2)>, change: &mut Vec<(V2, R2)>| {
-            if !input.is_empty() { logic(key, input, change); }
-            change.extend(output.drain(..).map(|(x, mut d)| { d.negate(); (x, d) }));
-            differential_dataflow::consolidation::consolidate(change);
-        },
-        |vec, key: &K, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v, t, r)| ((key.clone(), v), t, r))); },
-    )
-    .as_collection(|k, v| (k.clone(), v.clone()))
-}
-
-fn main() {
-    for _ in 0 .. 2 {
-        let c = reduce_churn(false);
-        let r = reduce_churn(true);
-        println!("reduce_churn  cursor {:>8.3}s   reference {:>8.3}s   ratio {:.2}x",
-            c.as_secs_f64(), r.as_secs_f64(), r.as_secs_f64() / c.as_secs_f64());
-    }
-    for _ in 0 .. 2 {
-        let c = fewkeys(false);
-        let r = fewkeys(true);
-        println!("fewkeys       cursor {:>8.3}s   reference {:>8.3}s   ratio {:.2}x",
-            c.as_secs_f64(), r.as_secs_f64(), r.as_secs_f64() / c.as_secs_f64());
-    }
-    for _ in 0 .. 2 {
-        let c = bfs_bench(false);
-        let r = bfs_bench(true);
-        println!("bfs           cursor {:>8.3}s   reference {:>8.3}s   ratio {:.2}x",
-            c.as_secs_f64(), r.as_secs_f64(), r.as_secs_f64() / c.as_secs_f64());
-    }
-}
-
-// Reduce-dominated, totally-ordered time: many keys churn values across many rounds; reduce counts
-// distinct values per key. Determination is trivial here, so this measures base engine overhead.
-fn reduce_churn(reference: bool) -> std::time::Duration {
-    let keys = 20000usize;
-    let vals = 100usize;
-    let rounds = 500usize;
-    let per_round = 5000usize;
-
-    let seed: &[_] = &[1, 2, 3, 4];
-    let mut rng: StdRng = SeedableRng::from_seed(seed);
-    let mut batches: Vec<Vec<(usize, usize, isize)>> = Vec::with_capacity(rounds);
-    for _ in 0 .. rounds {
-        let mut batch = Vec::with_capacity(per_round);
-        for _ in 0 .. per_round {
-            let k = rng.gen_range(0, keys);
-            let v = rng.gen_range(0, vals);
-            let d = if rng.gen::<bool>() { 1 } else { -1 };
-            batch.push((k, v, d));
-        }
-        batches.push(batch);
-    }
-
-    let start = Instant::now();
-    timely::execute_directly(move |worker| {
-        let mut input = worker.dataflow::<usize, _, _>(|scope| {
-            let (input, coll) = scope.new_collection::<(usize, usize), isize>();
-            let reduced = if reference {
-                reduce_reference(coll, |_k, s, t| t.push((s.len() as isize, 1)))
-            } else {
-                coll.reduce(|_k, s, t| t.push((s.len() as isize, 1)))
-            };
-            reduced.inspect(|_| {});
-            input
-        });
-        let mut round = 0usize;
-        for batch in batches {
-            for (k, v, d) in batch { input.update((k, v), d); }
-            round += 1;
-            input.advance_to(round);
-        }
-    });
-    start.elapsed()
-}
-
-// Few keys, wide processing intervals (via `update_at` + coarse `advance_to`), so each key's
-// per-round history is long — the regime where the phase-2 re-sort would actually cost.
-fn fewkeys(reference: bool) -> std::time::Duration {
-    let keys = 8usize;
-    let vals = 40usize;
-    let times = 24000usize;
-    let step = 6000usize;
-
-    let seed: &[_] = &[7, 7, 7, 7];
-    let mut rng: StdRng = SeedableRng::from_seed(seed);
-    let mut ups: Vec<(usize, usize, usize, isize)> = Vec::new();
-    for t in 1 ..= times {
-        for k in 0 .. keys {
-            ups.push((k, rng.gen_range(0, vals), t, if rng.gen::<bool>() { 1 } else { -1 }));
-        }
-    }
-
-    let start = Instant::now();
-    timely::execute_directly(move |worker| {
-        let mut input = worker.dataflow::<usize, _, _>(|scope| {
-            let (input, coll) = scope.new_collection::<(usize, usize), isize>();
-            let reduced = if reference {
-                reduce_reference(coll, |_k, s, t| t.push((s.len() as isize, 1)))
-            } else {
-                coll.reduce(|_k, s, t| t.push((s.len() as isize, 1)))
-            };
-            reduced.inspect(|_| {});
-            input
-        });
-        let mut next_flush = step;
-        for (k, v, t, d) in ups {
-            input.update_at((k, v), t, d);
-            if t >= next_flush { input.advance_to(next_flush); next_flush += step; }
-        }
-        input.advance_to(times + 1);
-    });
-    start.elapsed()
-}
-
-fn bfs_bench(reference: bool) -> std::time::Duration {
-    use std::sync::atomic::{AtomicUsize, Ordering};
-    use std::sync::Arc;
-
-    let nodes = 4000usize;
-    let edges = 32000usize;
-    let rounds = 400usize;
-
-    let seed: &[_] = &[9, 8, 7, 6];
-    let mut rng1: StdRng = SeedableRng::from_seed(seed);
-    let mut rng2: StdRng = SeedableRng::from_seed(seed);
-    let mut edge_list: Vec<((usize, usize), usize, isize)> = Vec::new();
-    for _ in 0 .. edges { edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), 0, 1)); }
-    for round in 1 .. rounds {
-        edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), round, 1));
-        edge_list.push(((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes)), round, -1));
-    }
-    let root_list = vec![(1usize, 0usize, 1isize)];
-    let counter = Arc::new(AtomicUsize::new(0));
-
-    let start = Instant::now();
-    let c = counter.clone();
-    timely::execute_directly(move |worker| {
-        let mut roots_list = root_list.clone();
-        let mut edges_list = edge_list.clone();
-        let (mut roots, mut edges) = worker.dataflow::<usize, _, _>(|scope| {
-            let (root_input, roots) = scope.new_collection();
-            let (edge_input, edges) = scope.new_collection();
-            let c = c.clone();
-            bfs(edges, roots, reference).inspect(move |_| { c.fetch_add(1, Ordering::Relaxed); });
-            (root_input, edge_input)
-        });
-        roots_list.sort_by(|x, y| y.1.cmp(&x.1));
-        edges_list.sort_by(|x, y| y.1.cmp(&x.1));
-        let mut round = 0;
-        while !roots_list.is_empty() || !edges_list.is_empty() {
-            while roots_list.last().map(|x| x.1) == Some(round) { let (n, _, d) = roots_list.pop().unwrap(); roots.update(n, d); }
-            while edges_list.last().map(|x| x.1) == Some(round) { let ((s, t), _, d) = edges_list.pop().unwrap(); edges.update((s, t), d); }
-            round += 1;
-            roots.advance_to(round);
-            edges.advance_to(round);
-        }
-    });
-    let elapsed = start.elapsed();
-    eprintln!("    (bfs {} output updates)", counter.load(std::sync::atomic::Ordering::Relaxed));
-    elapsed
-}
-
-fn bfs<'scope, T>(edges: VecCollection<'scope, T, Edge>, roots: VecCollection<'scope, T, Node>, reference: bool) -> VecCollection<'scope, T, (Node, usize)>
-where
-    T: timely::progress::Timestamp + Lattice + Ord,
-{
-    let nodes = roots.map(|x| (x, 0));
-    nodes.clone().iterate(|scope, inner| {
-        let edges = edges.enter(scope);
-        let nodes = nodes.enter(scope);
-        let combined = inner.join_map(edges, |_k, l, d| (*d, l + 1)).concat(nodes);
-        if reference {
-            reduce_reference(combined, |_, s, t| t.push((*s[0].0, 1)))
-        } else {
-            combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
-        }
-    })
-}

From 616fb792639932d0c2f7a3fe3b0ab633305ed4ba Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Thu, 2 Jul 2026 20:48:49 -0400
Subject: [PATCH 10/10] reduce: fix reference tactic's phase-2 catch-up walk
 (partial vs total order)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Phase 2 of the reference tactic visits only the active interesting times, so at
each it catches the input/output histories up to `next_time`. `history` is sorted
by the total `Ord` and `step` pops the least, but the walk stopped on the partial
`t.less_equal(next_time)`. The partial order interleaves with the sort: an
`Ord`-earlier time incomparable to `next_time` halts the walk early and strands
later edits that are genuinely `<= next_time`. The reduce then accumulates an
incomplete input and emits a transiently wrong output, self-correcting one
interesting-time later. Non-iterative and single-iterate (two-coordinate) shapes
never observe the in-between times, so it stayed hidden; SCC's nested iteration
(three coordinates) feeds the transient error back and it compounds into a wrong
result.

Fix: step the `Ord`-prefix `*t <= next_time`; the existing `less_equal` filter
when assembling the buffers selects the true `<= next_time` edits. This matches
the value-aware cursor, which walks every time in `Ord` order and filters by
`less_equal`. The Model.lean stream-correctness statement is unaffected — the
traversal was simply unfaithful to the partial-order accumulation it realizes.

Adds `scc_*` to tests/reduce_reference.rs: a differential SCC (two nested
product-timed fixpoints) that reaches the three-coordinate depth the prior tests
did not. The buggy tactic also thrashed the SCC fixpoint; with the fix it
converges (scc_sweep ~0.26s).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 differential-dataflow/src/operators/reduce.rs | 20 ++--
 .../tests/reduce_reference.rs                 | 95 ++++++++++++++++++-
 2 files changed, 107 insertions(+), 8 deletions(-)

diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs
index 59aa17880..55210ace4 100644
--- a/differential-dataflow/src/operators/reduce.rs
+++ b/differential-dataflow/src/operators/reduce.rs
@@ -1224,13 +1224,19 @@ pub(crate) mod reference {
                     let next_time = self.active[index].clone();
                     let meet = self.meets[index].clone();
 
-                    // Phase 2 visits only the active times, so we must consume ALL history edits at or
-                    // below `next_time` (not just those exactly at it, as a full time-order walk would
-                    // reach incrementally); edits at non-active times still contribute to the
-                    // accumulation here.
-                    while input_replay.time().map_or(false, |t| t.less_equal(&next_time)) { input_replay.step(); }
-                    while batch_replay.time().map_or(false, |t| t.less_equal(&next_time)) { batch_replay.step(); }
-                    while output_replay.time().map_or(false, |t| t.less_equal(&next_time)) { output_replay.step(); }
+                    // Phase 2 visits only the active times, so at each we must catch up the histories to
+                    // include every edit that will contribute to the accumulation at `next_time` (edits
+                    // at non-active times count too). `history` is sorted by the total `Ord` and `step`
+                    // pops the least, so we step the `Ord`-prefix `t <= next_time`, NOT the partial
+                    // `t.less_equal(next_time)`: the partial order interleaves with the sort, so an
+                    // `Ord`-earlier time incomparable to `next_time` would halt a `less_equal` walk early
+                    // and strand later edits that *are* `less_equal(next_time)`. Stepping the `Ord`-prefix
+                    // takes a superset; the `less_equal` filter below then selects the true `<= next_time`
+                    // edits. (Assembling with `less_equal` while stepping with `<=` is what the value-aware
+                    // cursor does via its full time-order frontier walk.)
+                    while input_replay.time().map_or(false, |t| *t <= next_time) { input_replay.step(); }
+                    while batch_replay.time().map_or(false, |t| *t <= next_time) { batch_replay.step(); }
+                    while output_replay.time().map_or(false, |t| *t <= next_time) { output_replay.step(); }
                     input_replay.advance_buffer_by(&meet);
                     batch_replay.advance_buffer_by(&meet);
                     output_replay.advance_buffer_by(&meet);
diff --git a/differential-dataflow/tests/reduce_reference.rs b/differential-dataflow/tests/reduce_reference.rs
index f8f65ff40..6046660ae 100644
--- a/differential-dataflow/tests/reduce_reference.rs
+++ b/differential-dataflow/tests/reduce_reference.rs
@@ -1,4 +1,4 @@
-//! Test harness for the model-derived `reference` reduce tactic. Three things:
+//! Test harness for the model-derived `reference` reduce tactic. Four things:
 //!
 //! * `differential_*` — run the default (cursor) reduce and the `reference` reduce on the SAME random
 //!   input and `assert_eq` their outputs at every time. The drift detector: if either tactic changes
@@ -13,6 +13,12 @@
 //!
 //! * `bfs_*` — an iterative BFS (product time via iteration; a real computation shape) computed both
 //!   ways over a graph that grows and shrinks; another differential check.
+//!
+//! * `scc_*` — strongly connected components: two nested product-timed fixpoints (edge trimming around
+//!   iterative label propagation), so the reduce runs at three nested time coordinates. This depth is
+//!   what exposed the phase-2 catch-up walk stepping the partial-order `less_equal` set instead of the
+//!   total-`Ord` prefix (see the note at that walk in `operators/reduce.rs`); the two-coordinate shapes
+//!   above never reached it.
 
 use std::collections::BTreeMap;
 
@@ -259,6 +265,93 @@ where
     })
 }
 
+// ===================== iterative SCC differential (a deeper shape) =====================
+// SCC nests two Product-timed fixpoints: an outer edge-trimming loop and, inside each trim, an
+// iterative label propagation whose `reduce` is the tactic under test. The doubly-nested product
+// time produces far more synthetic interesting times than BFS, over joins / concats / filters.
+
+#[test] fn scc_tiny_a() { scc_differential(6, 12, 4, &[1, 2, 3, 4]); }
+#[test] fn scc_tiny_b() { scc_differential(8, 16, 4, &[5, 6, 7, 8]); }
+#[test] fn scc_12_24()  { scc_differential(12, 24, 5, &[2, 4, 6, 8]); }
+#[test] fn scc_sweep()  { for s in 0u8 .. 6 { scc_differential(15, 30, 5, &[s as usize, (s as usize) + 7, 13, 29]); } }
+
+fn scc_differential(nodes: usize, edges: usize, rounds: usize, seed: &[usize]) {
+    let mut rng1: StdRng = SeedableRng::from_seed(seed);
+    let mut rng2: StdRng = SeedableRng::from_seed(seed);
+    let mut edge_list: Vec<((usize, usize), usize, isize)> = Vec::new();
+    for _ in 0 .. edges { edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), 0, 1)); }
+    for round in 1 .. rounds {
+        edge_list.push(((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)), round, 1));
+        edge_list.push(((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes)), round, -1));
+    }
+
+    timely::execute_directly(move |worker| {
+        let mut edges_list = edge_list.clone();
+        let mut edges = worker.dataflow::<usize, _, _>(|scope| {
+            let (edge_input, edges) = scope.new_collection();
+            let via_cursor    = scc(edges.clone(), false);
+            let via_reference = scc(edges, true);
+            via_cursor.assert_eq(via_reference);
+            edge_input
+        });
+        edges_list.sort_by(|x, y| y.1.cmp(&x.1));
+        let mut round = 0;
+        while !edges_list.is_empty() {
+            while edges_list.last().map(|x| x.1) == Some(round) { let ((s, t), _, d) = edges_list.pop().unwrap(); edges.update((s, t), d); }
+            round += 1;
+            edges.advance_to(round);
+        }
+    });
+}
+
+// The subset of edges within a strongly connected component, mirroring
+// `algorithms::graphs::scc::strongly_connected` but with the propagation `reduce` swappable.
+fn scc<'scope, T>(graph: VecCollection<'scope, T, Edge>, reference: bool) -> VecCollection<'scope, T, Edge>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let graph2 = graph.clone();
+    graph.iterate(move |scope, inner| {
+        let edges = graph2.enter(scope);
+        let trans = edges.clone().map(|(a, b)| (b, a));
+        let trimmed = trim_edges(inner, edges, reference);
+        trim_edges(trimmed, trans, reference)
+    })
+}
+
+// Keep only edges whose endpoints propagate to the same label; run twice (forward then transposed)
+// this converges on the SCC edges.
+fn trim_edges<'scope, T>(cycle: VecCollection<'scope, T, Edge>, edges: VecCollection<'scope, T, Edge>, reference: bool) -> VecCollection<'scope, T, Edge>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let nodes = edges.clone().map(|(_a, b)| (b, b));
+    let labels = propagate(cycle, nodes, reference);
+    edges.join_map(labels.clone(), |&src, &dst, &l1| (dst, (src, l1)))
+         .join_map(labels, |&dst, &(src, l1), &l2| ((src, dst), (l1, l2)))
+         .filter(|(_, (l1, l2))| l1 == l2)
+         .map(|((src, dst), _)| (dst, src))
+}
+
+// Forward min-label propagation along `edges` from per-node seed labels; the inner `reduce` (min
+// label per node) is the tactic under test.
+fn propagate<'scope, T>(edges: VecCollection<'scope, T, Edge>, nodes: VecCollection<'scope, T, (Node, Node)>, reference: bool) -> VecCollection<'scope, T, (Node, Node)>
+where
+    T: timely::progress::Timestamp + Lattice + Ord,
+{
+    let seed = nodes.clone();
+    nodes.filter(|_| false).iterate(move |scope, inner| {
+        let edges = edges.enter(scope);
+        let seed = seed.enter(scope);
+        let combined = inner.join_map(edges, |&_n, &l, &d| (d, l)).concat(seed);
+        if reference {
+            reduce_reference(combined, |_, s, t| t.push((*s[0].0, 1)))
+        } else {
+            combined.reduce(|_, s, t| t.push((*s[0].0, 1)))
+        }
+    })
+}
+
 /// A minimal product-order (partially ordered) timestamp, so the harness exercises the synthetic
 /// interesting times determination is about. Copied from `examples/multitemporal.rs`.
 mod pair {