From 03710d438d6341b032d1b7df74af5c4dc17c63fc Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 15:34:15 -0400
Subject: [PATCH 1/8] Reduce proxy pending-state rebuild overhead

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

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index 1e46d2c4f..52108087d 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -3,7 +3,10 @@
 //! A conventional differential reduce against `(u64, u64)`, where the backend supplies the
 //! implementation of the interpretation of the integers.
 
+use std::cell::RefCell;
 use std::collections::BTreeMap;
+use std::sync::OnceLock;
+use std::time::{Duration, Instant};
 
 use timely::PartialOrder;
 use timely::progress::{Antichain, Timestamp};
@@ -16,6 +19,72 @@ use super::ProxyBridge;
 use crate::operators::reduce::{sort_dedup, ReduceTactic};
 use crate::operators::ValueHistory;
 
+/// One opt-in phase measurement from the generic proxy-reduce driver.
+#[derive(Clone, Debug)]
+pub struct ProxyReducePhaseStat {
+    /// Stable phase label.
+    pub phase: &'static str,
+    /// Number of timed invocations.
+    pub calls: u64,
+    /// Phase-specific row or slot count, useful only within the named phase.
+    pub work: u64,
+    /// Total wall-clock time spent in the phase.
+    pub elapsed: Duration,
+}
+
+#[derive(Default)]
+struct ProxyReducePhaseAccum {
+    calls: u64,
+    work: u64,
+    elapsed: Duration,
+}
+
+thread_local! {
+    static PROXY_REDUCE_PROFILE: RefCell<BTreeMap<&'static str, ProxyReducePhaseAccum>> = RefCell::new(BTreeMap::new());
+}
+
+fn proxy_profile_enabled() -> bool {
+    static ENABLED: OnceLock<bool> = OnceLock::new();
+    *ENABLED.get_or_init(|| std::env::var("PROXY_REDUCE_PROFILE").is_ok_and(|x| x != "0"))
+}
+
+#[inline]
+fn phase_start() -> Option<Instant> {
+    proxy_profile_enabled().then(Instant::now)
+}
+
+#[inline]
+fn phase_finish(phase: &'static str, start: Option<Instant>, work: usize) {
+    if let Some(start) = start {
+        PROXY_REDUCE_PROFILE.with(|profile| {
+            let mut profile = profile.borrow_mut();
+            let stat = profile.entry(phase).or_default();
+            stat.calls += 1;
+            stat.work += work as u64;
+            stat.elapsed += start.elapsed();
+        });
+    }
+}
+
+/// Clear proxy-reduce phase counters for the current worker thread.
+pub fn reset_phase_profile() {
+    PROXY_REDUCE_PROFILE.with(|profile| profile.borrow_mut().clear());
+}
+
+/// Snapshot current-thread proxy-reduce phase counters, slowest phase first.
+pub fn phase_profile() -> Vec<ProxyReducePhaseStat> {
+    PROXY_REDUCE_PROFILE.with(|profile| {
+        let mut stats: Vec<_> = profile.borrow().iter().map(|(&phase, stat)| ProxyReducePhaseStat {
+            phase,
+            calls: stat.calls,
+            work: stat.work,
+            elapsed: stat.elapsed,
+        }).collect();
+        stats.sort_unstable_by_key(|stat| std::cmp::Reverse(stat.elapsed));
+        stats
+    })
+}
+
 /// A unit of proxied reduce work, presented to the backend.
 pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
     /// The accumulated input history.
@@ -162,7 +231,9 @@ where
         upper: &Antichain<B1::Time>,
         held: &Antichain<B1::Time>,
     ) -> (Vec<(B1::Time, B2)>, Antichain<B1::Time>) {
+        let retire_phase = phase_start();
         if held.elements().iter().all(|t| upper.less_equal(t)) {
+            phase_finish("retire.total_inclusive", retire_phase, 0);
             return (Vec::new(), held.clone());
         }
 
@@ -175,14 +246,24 @@ where
 
         // Split the carried interesting times against `upper`.
         // A time below it is DUE: its key must be re-evaluated this retire, so the key is `changed`.
-        // A time at or beyond it is carried forward untouched, seeding `new_pending`.
+        // A time at or beyond it remains carried in `self.pending`. Partition in place: rebuilding
+        // the whole map cloned every carried timestamp on every retire, even though only the small
+        // due subset changes.
         let mut due: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
-        let mut new_pending: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
-        for (key, times) in self.pending.iter() {
-            let (carried, ready): (Vec<_>, Vec<_>) = times.iter().cloned().partition(|t| upper.less_equal(t));
-            if !ready.is_empty() { due.insert(*key, ready); }
-            if !carried.is_empty() { new_pending.insert(*key, carried); }
+        let mut pending_frontier = Antichain::new();
+        let phase = phase_start();
+        for (&key, times) in self.pending.iter_mut() {
+            let mut ready = Vec::new();
+            times.retain(|time| {
+                let carried = upper.less_equal(time);
+                if carried { pending_frontier.insert_ref(time); }
+                else { ready.push(time.clone()); }
+                carried
+            });
+            if !ready.is_empty() { due.insert(key, ready); }
         }
+        self.pending.retain(|_, times| !times.is_empty());
+        phase_finish("retire.partition_pending", phase, self.pending.len());
         // The keys the harness knows must be revisited. The backend adds those its novel batches
         // touch, which it discovers while reading them; neither side scans the whole key space.
         let changed: Vec<u64> = due.keys().copied().collect();
@@ -193,20 +274,17 @@ where
         // beyond `upper` can remain when nothing is due, and releasing their capabilities would
         // strand them (see the frontier clause of the `ReduceTactic::retire` contract).
         if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) {
-            let mut frontier = Antichain::new();
-            for times in self.pending.values() {
-                for time in times {
-                    frontier.insert_ref(time);
-                }
-            }
-            return (Vec::new(), frontier);
+            phase_finish("retire.total_inclusive", retire_phase, 0);
+            return (Vec::new(), pending_frontier);
         }
 
         // The output tiling (identical to the Abelian tactic): one tile per held time, keeping
         // non-degenerate intervals; `tile_of[i]` maps held time `i` to its tile.
         let held_elems: Vec<B1::Time> = held.elements().to_vec();
+        let phase = phase_start();
         let (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems);
         self.backend.begin(&tile_descs);
+        phase_finish("retire.describe_and_begin", phase, tile_descs.len());
 
         // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
         // once the backend reports the space covered.
@@ -216,6 +294,7 @@ where
         // Retire-wide reusable scratch: cleared per window or wave, never reallocated. Fresh
         // per-key/per-wave `Vec`s were once the dominant cost here, which is why the slots and the
         // staging buffers are held across the whole retire rather than built where they are used.
+        let phase = phase_start();
         let mut slots: Vec<KeySweep<B1::Time, Bk::RIn, Bk::ROut>> = Vec::new();
         let mut live: Vec<usize> = Vec::new();
         let mut tile_deltas: Vec<Vec<((u64, u64), B1::Time, Bk::ROut)>> = (0..held_elems.len()).map(|_| Vec::new()).collect();
@@ -227,11 +306,14 @@ where
         let mut active: Vec<(usize, B1::Time)> = Vec::new();
         let mut in_accum: Vec<(u64, Bk::RIn)> = Vec::new();
         let mut cur_out: Vec<(u64, Bk::ROut)> = Vec::new();
+        phase_finish("retire.allocate_scratch", phase, held_elems.len());
 
         while from.is_some() {
             let before = from;
             window.clear();
+            let phase = phase_start();
             self.backend.next_window(&instance, &changed, &mut from, &mut window);
+            phase_finish("window.backend_present", phase, window.input.len() + window.output.len() + window.seeds.len());
             let p_in = &window.input;
             let seeds = &window.seeds;
             let p_out = &window.output;
@@ -274,6 +356,7 @@ where
             let mut n_slots = 0usize;
             let (mut is, mut ns, mut os) = (0usize, 0usize, 0usize);
             live.clear();
+            let phase = phase_start();
             // Mapped to hashes before the min: the sources differ in shape.
             while let Some(key) = [
                 p_in.get(is).map(|record| record.0.0),
@@ -294,7 +377,7 @@ where
                 let slot = &mut slots[n_slots];
                 slot.key = key;
                 slot.pended.clear();
-                // Only the DUE times seed the sweep; the carried ones are already in `new_pending`.
+                // Only the DUE times seed the sweep; the carried ones remain in `self.pending`.
                 let owed = due.get(&key).map(|p| &p[..]).unwrap_or(&[]);
                 slot.sweep.load(
                     owed,
@@ -305,10 +388,12 @@ where
                 slot.at = slot.sweep.next_crossing(upper, &mut slot.pended);
                 if slot.at.is_some() { live.push(n_slots); }
                 else if !slot.pended.is_empty() {
-                    new_pending.entry(key).or_default().append(&mut slot.pended);
+                    for time in &slot.pended { pending_frontier.insert_ref(time); }
+                    self.pending.entry(key).or_default().append(&mut slot.pended);
                 }
                 n_slots += 1;
             }
+            phase_finish("window.initialize_sweeps", phase, p_in.len() + seeds.len() + p_out.len());
 
             // Each wave: read every suspended key's accumulations, cross the non-empty ones in one
             // call, hand the corrections back, and step every live key on. A key retires when its
@@ -321,6 +406,7 @@ where
                 out_all.clear();
                 active.clear();
 
+                let phase = phase_start();
                 for &si in live.iter() {
                     let at = slots[si].at.clone().expect("live slots are suspended at a time");
                     in_accum.clear();
@@ -337,9 +423,13 @@ where
                     out_ends.push(out_all.len());
                     active.push((si, at));
                 }
+                phase_finish("wave.accumulate", phase, live.len());
 
                 if !batch_keys.is_empty() {
+                    let phase = phase_start();
                     let (corr, corr_ends) = self.backend.reduce_corrections(&batch_keys, &in_ends, &in_all, &out_ends, &out_all);
+                    phase_finish("wave.backend_corrections", phase, in_all.len() + out_all.len());
+                    let phase = phase_start();
                     let mut cstart = 0usize;
                     for (bi, (si, at)) in active.iter().enumerate() {
                         let cend = corr_ends[bi];
@@ -352,41 +442,45 @@ where
                         }
                         cstart = cend;
                     }
+                    phase_finish("wave.commit_corrections", phase, corr.len());
                 }
 
                 // Step every live key past the time it was suspended at, and retire the spent ones.
+                let phase = phase_start();
                 for &si in live.iter() {
                     let slot = &mut slots[si];
                     slot.at = slot.sweep.next_crossing(upper, &mut slot.pended);
                     if slot.at.is_none() && !slot.pended.is_empty() {
-                        let entry = new_pending.entry(slot.key).or_default();
+                        for time in &slot.pended { pending_frontier.insert_ref(time); }
+                        let entry = self.pending.entry(slot.key).or_default();
                         entry.append(&mut slot.pended);
                         crate::operators::reduce::sort_dedup(entry);
                     }
                 }
                 live.retain(|&si| slots[si].at.is_some());
+                phase_finish("wave.advance_sweeps", phase, live.len());
             }
 
+            let phase = phase_start();
+            let mut emit_work = 0usize;
             for (held_index, deltas) in tile_deltas.iter_mut().enumerate() {
                 if deltas.is_empty() {
                     continue;
                 }
                 if let Some(tile) = tile_of[held_index] {
+                    emit_work += deltas.len();
                     crate::consolidation::consolidate_updates(deltas);
                     self.backend.emit(tile, &deltas[..]);
                 }
             }
+            phase_finish("window.consolidate_and_emit", phase, emit_work);
         }
 
-        self.pending = new_pending;
+        let phase = phase_start();
         let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
-        let mut frontier = Antichain::new();
-        for times in self.pending.values() {
-            for t in times {
-                frontier.insert_ref(t);
-            }
-        }
-        (produced, frontier)
+        phase_finish("retire.backend_finish", phase, produced.len());
+        phase_finish("retire.total_inclusive", retire_phase, produced.len());
+        (produced, pending_frontier)
     }
 }
 

From 53785738c72d35dfb3b63439b1e1b6567619ca93 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 15:34:15 -0400
Subject: [PATCH 2/8] Merge ordered Corgi reduce presentations

---
 interactive/src/corgi/reduce.rs | 208 +++++++++++++++++++++++++++++---
 1 file changed, 193 insertions(+), 15 deletions(-)

diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 12b01738f..81fd6c127 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -26,10 +26,14 @@
 //! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes
 //! (a columnar semijoin — matching the row-wise tactic's read).
 
-use std::collections::HashMap;
+use std::cell::RefCell;
+use std::cmp::Reverse;
+use std::collections::BinaryHeap;
+use std::collections::{BTreeMap, HashMap};
 use std::hash::{BuildHasherDefault, Hasher};
 use std::rc::Rc;
-
+use std::sync::OnceLock;
+use std::time::{Duration, Instant};
 
 use differential_dataflow::consolidation::consolidate_updates;
 use differential_dataflow::trace::Description;
@@ -47,6 +51,73 @@ use crate::parse::Reducer;
 
 type CBatch<T> = Rc<ChunkBatch<CorgiChunk<T, Diff>>>;
 
+/// One opt-in `CorgiReduceBackend` phase measurement. `work` is the phase's natural row count;
+/// it is diagnostic context, not a quantity comparable across differently named phases.
+#[derive(Clone, Debug)]
+pub struct ReducePhaseStat {
+    pub phase: &'static str,
+    pub calls: u64,
+    pub work: u64,
+    pub elapsed: Duration,
+}
+
+#[derive(Default)]
+struct ReducePhaseAccum {
+    calls: u64,
+    work: u64,
+    elapsed: Duration,
+}
+
+thread_local! {
+    static REDUCE_PROFILE: RefCell<BTreeMap<&'static str, ReducePhaseAccum>> = RefCell::new(BTreeMap::new());
+}
+
+fn reduce_profile_enabled() -> bool {
+    static ENABLED: OnceLock<bool> = OnceLock::new();
+    *ENABLED.get_or_init(|| std::env::var("CORGI_REDUCE_PROFILE").is_ok_and(|x| x != "0"))
+}
+
+#[inline]
+fn phase_start() -> Option<Instant> {
+    reduce_profile_enabled().then(Instant::now)
+}
+
+#[inline]
+fn phase_finish(phase: &'static str, start: Option<Instant>, work: usize) {
+    if let Some(start) = start {
+        REDUCE_PROFILE.with(|profile| {
+            let mut profile = profile.borrow_mut();
+            let stat = profile.entry(phase).or_default();
+            stat.calls += 1;
+            stat.work += work as u64;
+            stat.elapsed += start.elapsed();
+        });
+    }
+}
+
+/// Clear phase counters for the current worker thread.
+pub fn reset_phase_profile() {
+    REDUCE_PROFILE.with(|profile| profile.borrow_mut().clear());
+}
+
+/// Snapshot the current worker thread's phase counters, slowest phase first.
+pub fn phase_profile() -> Vec<ReducePhaseStat> {
+    REDUCE_PROFILE.with(|profile| {
+        let mut stats: Vec<_> = profile
+            .borrow()
+            .iter()
+            .map(|(&phase, stat)| ReducePhaseStat {
+                phase,
+                calls: stat.calls,
+                work: stat.work,
+                elapsed: stat.elapsed,
+            })
+            .collect();
+        stats.sort_unstable_by_key(|stat| std::cmp::Reverse(stat.elapsed));
+        stats
+    })
+}
+
 /// An identity `Hasher` for the id-index maps: their keys are already well-distributed 64-bit
 /// content hashes (`hash_rows`), so passing the id straight through avoids re-hashing it (siphash
 /// on `register_keys`/lookups was ~7% of the reduce in profiling). Only `write_u64` is used.
@@ -206,7 +277,7 @@ fn seek_needles(sample: &CValue, changed: &[u64]) -> Option<CValue> {
 /// TODO: the scan's per-row work can still batch: `ids` re-derives (and copies) each chunk's
 /// key hashes every retire (memoize per chunk, or a stored hash column), and each hit
 /// materializes an owned time (`times().get`); kept RANGES could move via `push_range`.
-fn collect_present<T>(chunks: &[&CorgiChunk<T, Diff>], changed: &[u64]) -> (CValue, CValue, Vec<u64>, Vec<T>, Vec<Diff>)
+fn collect_present<T>(chunks: &[&CorgiChunk<T, Diff>], changed: &[u64]) -> (CValue, CValue, Vec<u64>, Vec<T>, Vec<Diff>, Vec<usize>)
 where
     T: ColTime,
 {
@@ -215,22 +286,30 @@ where
     /// (~log(rows) compares, each far costlier than the scan's flat membership test), so
     /// marginal seeks LOSE to the scan — measured, not modeled; 16 regressed load-shaped
     /// retires before this was widened.
-    const SEEK_ADVANTAGE: usize = 64;
+    fn seek_advantage() -> usize {
+        static ADVANTAGE: OnceLock<usize> = OnceLock::new();
+        *ADVANTAGE.get_or_init(|| {
+            std::env::var("CORGI_SEEK_ADVANTAGE").ok().and_then(|x| x.parse().ok()).unwrap_or(64)
+        })
+    }
 
     let key_srcs: Vec<Option<&CValue>> = chunks.iter().map(|c| Some(c.keys())).collect();
     let val_srcs: Vec<Option<&CValue>> = chunks.iter().map(|c| Some(c.vals())).collect();
     let (mut tags, mut offs) = (Vec::new(), Vec::new());
     let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new());
+    let mut run_ends = Vec::new();
     let total: usize = chunks.iter().map(|c| c.diffs().len()).sum();
-    let needles = if changed.len().saturating_mul(SEEK_ADVANTAGE) < total {
+    let needles = if changed.len().saturating_mul(seek_advantage()) < total {
         chunks.iter().find(|c| c.diffs().len() > 0).and_then(|c| seek_needles(c.keys(), changed))
     } else {
         None
     };
     if let Some(needles) = needles {
+        let phase = phase_start();
         // Narrow changed set over seekable keys: gallop each chunk once per changed key.
         // Chunks are key-ordered and `changed` ascends, so emission order matches the scan's.
         for (ci, ch) in chunks.iter().enumerate() {
+            let before = khs.len();
             if ch.diffs().is_empty() {
                 continue;
             }
@@ -244,9 +323,13 @@ where
                     diffs.push(ch.diffs()[i]);
                 }
             }
+            if khs.len() > before { run_ends.push(khs.len()); }
         }
+        phase_finish("present.seek_chunks", phase, total);
     } else {
+        let phase = phase_start();
         for (ci, ch) in chunks.iter().enumerate() {
+            let before = khs.len();
             // Borrow the key leaf when there is one (`ids`' value-as-id fast paths); only
             // structural keys need the hash, and only they pay a materialization. A shared
             // column's `Arc` cannot be unwrapped, so `ids` would copy the whole key column
@@ -266,14 +349,66 @@ where
                     diffs.push(ch.diffs()[i]);
                 }
             }
+            if khs.len() > before { run_ends.push(khs.len()); }
         }
+        phase_finish("present.scan_chunks", phase, total);
     }
     if tags.is_empty() {
-        return (CValue::Unit(0), CValue::Unit(0), khs, times, diffs);
+        return (CValue::Unit(0), CValue::Unit(0), khs, times, diffs, run_ends);
     }
+    let phase = phase_start();
     let keys_col = gather_lanes(&key_srcs, &tags, &offs);
     let vals_col = gather_lanes(&val_srcs, &tags, &offs);
-    (keys_col, vals_col, khs, times, diffs)
+    phase_finish("present.gather_columns", phase, tags.len());
+    (keys_col, vals_col, khs, times, diffs, run_ends)
+}
+
+fn merge_present_enabled() -> bool {
+    static ENABLED: OnceLock<bool> = OnceLock::new();
+    *ENABLED.get_or_init(|| std::env::var("CORGI_MERGE_PRESENT").map_or(true, |x| x != "0"))
+}
+
+/// Merge already-ordered selected chunk runs directly into a consolidated proxy bridge. Returns
+/// false when a run's structural value order does not agree with its proxy-id order.
+fn merge_present<T: Ord + Clone>(
+    khs: &[u64], vids: &[u64], times: &[T], diffs: &[Diff], run_ends: &[usize],
+    bridge: &mut ProxyBridge<T, Diff>,
+) -> bool {
+    let mut start = 0usize;
+    for &end in run_ends {
+        if (start + 1..end).any(|i| (khs[i - 1], vids[i - 1], &times[i - 1]) > (khs[i], vids[i], &times[i])) {
+            return false;
+        }
+        start = end;
+    }
+
+    bridge.clear();
+    let mut heap: BinaryHeap<Reverse<((u64, u64), T, usize, usize)>> = BinaryHeap::new();
+    let mut lo = 0usize;
+    for (run, &hi) in run_ends.iter().enumerate() {
+        heap.push(Reverse(((khs[lo], vids[lo]), times[lo].clone(), run, lo)));
+        lo = hi;
+    }
+    let mut current: Option<((u64, u64), T, Diff)> = None;
+    while let Some(Reverse((kv, time, run, index))) = heap.pop() {
+        if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == &time) {
+            current.as_mut().unwrap().2 += diffs[index];
+        } else {
+            if let Some(record) = current.take() {
+                if record.2 != 0 { bridge.push(record); }
+            }
+            current = Some((kv, time, diffs[index]));
+        }
+        let end = run_ends[run];
+        if index + 1 < end {
+            let next = index + 1;
+            heap.push(Reverse(((khs[next], vids[next]), times[next].clone(), run, next)));
+        }
+    }
+    if let Some(record) = current {
+        if record.2 != 0 { bridge.push(record); }
+    }
+    true
 }
 
 /// All chunks of a batch list, flattened (empty chunks included — `hash_rows` yields nothing for them).
@@ -302,17 +437,27 @@ where
         len: &mut usize,
         bridge: &mut ProxyBridge<T, Diff>,
     ) {
-        let (p_keys, p_vals, khs, times, diffs) = collect_present(chunks, keys);
+        let (p_keys, p_vals, khs, times, diffs, run_ends) = collect_present(chunks, keys);
         if khs.is_empty() {
             return;
         }
+        let phase = phase_start();
         let vids = ids(&p_vals);
         for (row, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(*len + row); }
         *len += p_vals.len();
         blocks.push(p_vals);
         self.register_keys(p_keys, &khs);
-        bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i])));
-        consolidate_updates(bridge);
+        phase_finish("present.register_input", phase, khs.len());
+        let phase = phase_start();
+        if merge_present_enabled() && merge_present(&khs, &vids, &times, &diffs, &run_ends, bridge) {
+            phase_finish("present.merge_input", phase, khs.len());
+        } else {
+            bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i])));
+            phase_finish("present.build_input_bridge", phase, khs.len());
+            let phase = phase_start();
+            consolidate_updates(bridge);
+            phase_finish("present.consolidate_input", phase, bridge.len());
+        }
     }
 
     /// The one value crossing for a retire: every `(key, time)` bracket at once. Builds the output
@@ -492,6 +637,7 @@ where
         // The seeds are the novel batches' RAW (key_hash, time) support, recorded here — before the
         // merged presentation below, whose consolidation may net a novel record away entirely. The
         // key hashes come from the scan the key list needs anyway.
+        let phase = phase_start();
         let mut seeds: Vec<(u64, T)> = Vec::new();
         for ch in novel_chunks.iter() {
             let khs = ids(ch.keys());
@@ -523,6 +669,7 @@ where
             }
             keys = merged;
         }
+        phase_finish("window.seed_and_keys", phase, novel_chunks.iter().map(|c| c.diffs().len()).sum::<usize>() + changed.len());
         if keys.is_empty() {
             self.in_vals = CValue::Unit(0);
             self.in_index = IdMap::default();
@@ -538,28 +685,50 @@ where
         let mut in_len = 0usize;
         let mut in_chunks = chunks_of(instance.source_batches);
         in_chunks.extend(novel_chunks.iter().copied());
+        let phase = phase_start();
         self.present_input(&in_chunks, &keys, &mut in_blocks, &mut in_len, &mut window.input);
+        phase_finish("window.present_input", phase, in_chunks.iter().map(|c| c.diffs().len()).sum());
+        let phase = phase_start();
         self.in_vals = concat_columns(&in_blocks);
+        phase_finish("window.concat_input", phase, in_len);
 
         // Output-history presentation, same keys (register keys + values for correction resolution).
-        let (o_keys, o_vals, o_khs, o_times, o_diffs) = collect_present(&chunks_of(instance.output_batches), &keys);
+        let output_chunks = chunks_of(instance.output_batches);
+        let phase = phase_start();
+        let (o_keys, o_vals, o_khs, o_times, o_diffs, o_run_ends) = collect_present(&output_chunks, &keys);
         if !o_khs.is_empty() {
+            let subphase = phase_start();
             let vids = ids(&o_vals);
             self.register_keys(o_keys, &o_khs);
             self.register_vals(o_vals, &vids);
-            window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])));
-            consolidate_updates(&mut window.output);
+            phase_finish("present.register_output", subphase, o_khs.len());
+            let subphase = phase_start();
+            if merge_present_enabled() && merge_present(&o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output) {
+                phase_finish("present.merge_output", subphase, o_khs.len());
+            } else {
+                window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])));
+                phase_finish("present.build_output_bridge", subphase, o_khs.len());
+                let subphase = phase_start();
+                consolidate_updates(&mut window.output);
+                phase_finish("present.consolidate_output", subphase, window.output.len());
+            }
         }
+        phase_finish("window.present_output", phase, output_chunks.iter().map(|c| c.diffs().len()).sum());
     }
 
     fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec<usize>) {
         // Resolve input value_ids to `in_vals` rows, reduce (desired output), then difference the
         // desired against the presented current output per key: correction = desired − current.
+        let phase = phase_start();
         let in_rows: Vec<(usize, Diff)> = input.iter()
             .map(|&(vid, d)| (*self.in_index.get(&vid).expect("input value_id presented this window"), d))
             .collect();
+        phase_finish("corrections.resolve_input", phase, input.len());
+        let phase = phase_start();
         let (desired, desired_ends) = self.reduce_brackets(in_ends, &in_rows);
+        phase_finish("corrections.reduce_brackets", phase, in_rows.len());
 
+        let phase = phase_start();
         let mut corr: Vec<(u64, Diff)> = Vec::new();
         let mut corr_ends: Vec<usize> = Vec::with_capacity(keys.len());
         let (mut ds, mut os) = (0usize, 0usize);
@@ -582,11 +751,13 @@ where
             ds = de;
             os = oe;
         }
+        phase_finish("corrections.diff_output", phase, desired.len() + output.len());
         (corr, corr_ends)
     }
 
     fn emit(&mut self, tile: usize, records: &[((u64, u64), T, Diff)]) {
         // Resolve each correction's key/value proxies to pool rows and accumulate into the tile.
+        let phase = phase_start();
         for rec in records {
             let ((kh, vid), t, d) = (rec.0, &rec.1, rec.2);
             let kr = *self.key_index.get(&kh).expect("key resolvable this retire");
@@ -597,18 +768,25 @@ where
             times.push(t.clone());
             diffs.push(d);
         }
+        phase_finish("emit.resolve_and_buffer", phase, records.len());
     }
 
     fn finish(&mut self) -> Vec<CBatch<T>> {
         // Seal each tile: gather its accumulated (key, val) pool rows into columns, one CorgiChunk batch.
+        let phase = phase_start();
         let key_pool = concat_columns(&self.key_blocks);
         let val_pool = concat_columns(&self.val_blocks);
+        phase_finish("finish.concat_pools", phase, self.key_len + self.val_len);
         let tiles = std::mem::take(&mut self.tiles);
         let tile_rows = std::mem::take(&mut self.tile_rows);
-        tiles.into_iter().zip(tile_rows).map(|(desc, (krows, vrows, times, diffs))| {
+        let output_rows: usize = tile_rows.iter().map(|rows| rows.0.len()).sum();
+        let phase = phase_start();
+        let result = tiles.into_iter().zip(tile_rows).map(|(desc, (krows, vrows, times, diffs))| {
             let keys = gather(&key_pool, &krows);
             let vals = gather(&val_pool, &vrows);
             Rc::new(columns_to_batch(keys, vals, times, diffs, desc))
-        }).collect()
+        }).collect();
+        phase_finish("finish.gather_and_seal", phase, output_rows);
+        result
     }
 }

From c6986463f2c0b39f5838d8f8f0a84b45d7cccb66 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 15:34:15 -0400
Subject: [PATCH 3/8] Add SCC backend comparison benchmark

---
 interactive/examples/scc_compare.rs | 502 ++++++++++++++++++++++++++++
 1 file changed, 502 insertions(+)
 create mode 100644 interactive/examples/scc_compare.rs

diff --git a/interactive/examples/scc_compare.rs b/interactive/examples/scc_compare.rs
new file mode 100644
index 000000000..0609c5b6f
--- /dev/null
+++ b/interactive/examples/scc_compare.rs
@@ -0,0 +1,502 @@
+//! Apples-to-apples SCC benchmark: compiled Differential vs DDIR Vec vs DDIR Corgi.
+//!
+//! All three variants consume the same deterministic graph and update stream. The compiled
+//! baseline uses `strongly_connected_at(_, |x| *x as u64)`, matching the logarithmic label staging
+//! performed by DDIR's `enter_at($1[0])`. The measured output is only probed; it is not inspected or
+//! captured. DDIR parsing, lowering, optimization, and dataflow construction are outside the load
+//! and update timers.
+//!
+//! ```text
+//! N=100000 E=200000 BATCH=1000 ROUNDS=20 ITERS=3 \
+//!   cargo run --release -p interactive --example scc_compare
+//! ```
+//!
+//! `BACKENDS=compiled,vec,corgi` selects variants. `CHECK=1` first checks Vec/Corgi result parity
+//! on a bounded prefix of the graph (default on). Corgi is currently single-worker, so this
+//! benchmark intentionally uses `timely::execute_directly` for every variant.
+
+use std::cell::RefCell;
+use std::collections::BTreeMap;
+use std::rc::Rc;
+use std::time::{Duration, Instant};
+
+use differential_dataflow::algorithms::graphs::scc::strongly_connected_at;
+use differential_dataflow::dynamic::pointstamp::PointStamp;
+use differential_dataflow::input::Input;
+use interactive::backend::{corgi, vec};
+use interactive::ir::{Diff, Value};
+use interactive::scope_ir as st;
+use interactive::{lower, parse};
+use timely::dataflow::operators::probe::Handle;
+use timely::logging::{StartStop, TimelyEvent, TimelyEventBuilder};
+
+#[global_allocator]
+static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
+const SCC: &str = r#"
+let edges = input 0 | key($0[0] ; $0[1]);
+let trans = edges | key($1 ; $0);
+
+outer: {
+    let scc = edges + trim;
+    fwd: {
+        let nodes = edges | key($1 ; $1) | enter_at($1[0]);
+        let labels = proposals + nodes | min;
+        var proposals = labels | join(scc, ($2 ; $1));
+    }
+    let trim_fwd = edges
+        | join(fwd::labels, ($1 ; $0, $2))
+        | join(fwd::labels, ($0 ; $1, $2))
+        | filter($1[1] == $1[2])
+        | key($0 ; $1[0]);
+    bwd: {
+        let nodes = trans | key($1 ; $1) | enter_at($1[0]);
+        let labels = proposals + nodes | min;
+        var proposals = labels | join(trim_fwd, ($2 ; $1));
+    }
+    let trim_bwd = trans
+        | join(bwd::labels, ($1 ; $0, $2))
+        | join(bwd::labels, ($0 ; $1, $2))
+        | filter($1[1] == $1[2])
+        | key($0 ; $1[0]);
+    var trim = trim_bwd - edges;
+}
+
+export "result" = outer::scc | map(;) | arrange;
+"#;
+
+#[derive(Clone, Copy, Debug)]
+enum Backend {
+    Compiled,
+    Vec,
+    Corgi,
+}
+
+impl Backend {
+    fn name(self) -> &'static str {
+        match self {
+            Backend::Compiled => "compiled",
+            Backend::Vec => "vec",
+            Backend::Corgi => "corgi",
+        }
+    }
+}
+
+#[derive(Debug)]
+struct Stats {
+    load: Duration,
+    rounds: Vec<Duration>,
+    operators: Vec<(String, Duration)>,
+    reduce_phases: Vec<interactive::corgi::reduce::ReducePhaseStat>,
+    proxy_reduce_phases:
+        Vec<differential_dataflow::operators::int_proxy::reduce::ProxyReducePhaseStat>,
+}
+
+#[derive(Default)]
+struct OperatorProfile {
+    operators: BTreeMap<usize, (String, Vec<usize>)>,
+    elapsed: BTreeMap<usize, Duration>,
+}
+
+macro_rules! install_operator_profile {
+    ($worker:expr) => {{
+        let profile = Rc::new(RefCell::new(OperatorProfile::default()));
+        if std::env::var("PROFILE_OPS").is_ok_and(|x| x != "0") {
+            let target = Rc::clone(&profile);
+            let mut starts = BTreeMap::new();
+            $worker
+                .log_register()
+                .expect("timely logging registry")
+                .insert::<TimelyEventBuilder, _>("timely", move |_time, data| {
+                    if let Some(events) = data {
+                        for (event_time, event) in events.iter() {
+                            match event {
+                                TimelyEvent::Operates(event) => {
+                                    target
+                                        .borrow_mut()
+                                        .operators
+                                        .insert(event.id, (event.name.clone(), event.addr.clone()));
+                                }
+                                TimelyEvent::Schedule(event) => match event.start_stop {
+                                    StartStop::Start => {
+                                        starts.insert(event.id, *event_time);
+                                    }
+                                    StartStop::Stop => {
+                                        if let Some(start) = starts.remove(&event.id) {
+                                            *target
+                                                .borrow_mut()
+                                                .elapsed
+                                                .entry(event.id)
+                                                .or_default() += event_time.saturating_sub(start);
+                                        }
+                                    }
+                                },
+                                _ => {}
+                            }
+                        }
+                    }
+                });
+        }
+        profile
+    }};
+}
+
+fn finish_operator_profile(profile: Rc<RefCell<OperatorProfile>>) -> Vec<(String, Duration)> {
+    let profile = profile.borrow();
+    let addresses: Vec<&Vec<usize>> = profile
+        .operators
+        .values()
+        .map(|(_, address)| address)
+        .collect();
+    let mut operators: Vec<_> = profile
+        .elapsed
+        .iter()
+        .filter_map(|(id, elapsed)| {
+            let (name, address) = profile
+                .operators
+                .get(id)
+                .cloned()
+                .unwrap_or_else(|| (format!("operator-{id}"), Vec::new()));
+            let has_child = addresses
+                .iter()
+                .any(|other| other.len() > address.len() && other.starts_with(&address));
+            (!has_child).then_some((name, *elapsed))
+        })
+        .collect();
+    operators.sort_unstable_by_key(|(_, elapsed)| std::cmp::Reverse(*elapsed));
+    operators
+}
+
+fn env_usize(name: &str, default: usize) -> usize {
+    std::env::var(name)
+        .ok()
+        .and_then(|x| x.parse().ok())
+        .unwrap_or(default)
+}
+
+fn mix(mut x: u64) -> u64 {
+    x = x.wrapping_add(0x9e3779b97f4a7c15);
+    x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
+    x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
+    x ^ (x >> 31)
+}
+
+fn edge(index: usize, nodes: usize) -> (usize, usize) {
+    let first = mix(index as u64);
+    let second = mix(first);
+    ((first as usize) % nodes, (second as usize) % nodes)
+}
+
+fn row_edge(index: usize, nodes: usize) -> (Value, Value) {
+    let (src, dst) = edge(index, nodes);
+    (
+        Value::Tuple(vec![Value::Int(src as i64), Value::Int(dst as i64)]),
+        Value::unit(),
+    )
+}
+
+fn compiled(nodes: usize, edges: usize, batch: usize, rounds: usize) -> Stats {
+    timely::execute_directly(move |worker| {
+        let operator_profile = install_operator_profile!(worker);
+        let mut probe = Handle::new();
+        let mut input = worker.dataflow::<usize, _, _>(|scope| {
+            let (input, graph) = scope.new_collection::<(usize, usize), isize>();
+            strongly_connected_at(graph, |x| *x as u64).probe_with(&mut probe);
+            input
+        });
+
+        let timer = Instant::now();
+        for index in 0..edges {
+            input.insert(edge(index, nodes));
+        }
+        input.advance_to(1);
+        input.flush();
+        while probe.less_than(input.time()) {
+            worker.step();
+        }
+        let load = timer.elapsed();
+
+        let mut times = Vec::with_capacity(rounds);
+        for round in 0..rounds {
+            let timer = Instant::now();
+            for offset in 0..batch {
+                let index = round * batch + offset;
+                input.remove(edge(index, nodes));
+                input.insert(edge(edges + index, nodes));
+            }
+            input.advance_to(round + 2);
+            input.flush();
+            while probe.less_than(input.time()) {
+                worker.step();
+            }
+            times.push(timer.elapsed());
+        }
+        worker
+            .log_register()
+            .expect("timely logging registry")
+            .flush();
+        let operators = finish_operator_profile(operator_profile);
+        Stats {
+            load,
+            rounds: times,
+            operators,
+            reduce_phases: Vec::new(),
+            proxy_reduce_phases: Vec::new(),
+        }
+    })
+}
+
+fn interpreted(
+    backend: Backend,
+    program: &st::Program,
+    nodes: usize,
+    edges: usize,
+    batch: usize,
+    rounds: usize,
+) -> Stats {
+    let program = program.clone();
+    timely::execute_directly(move |worker| {
+        if matches!(backend, Backend::Corgi) {
+            interactive::corgi::reduce::reset_phase_profile();
+            differential_dataflow::operators::int_proxy::reduce::reset_phase_profile();
+        }
+        let operator_profile = install_operator_profile!(worker);
+        let mut probe = Handle::new();
+        let mut input = worker.dataflow::<u64, _, _>(|scope| {
+            let (input, collection) = scope.new_collection::<(Value, Value), Diff>();
+            let output = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
+                let entered = collection.clone().enter(inner);
+                let imports = program
+                    .root
+                    .imports
+                    .iter()
+                    .map(|import| match &import.from {
+                        st::Source::Input(0) => entered.clone(),
+                        ref other => panic!("unexpected SCC import {other:?}"),
+                    })
+                    .collect();
+                let exports = match backend {
+                    Backend::Vec => vec::render_tree(&program.root, inner.clone(), 0, imports),
+                    Backend::Corgi => {
+                        corgi::render_tree_rows(&program.root, inner.clone(), 0, imports)
+                    }
+                    Backend::Compiled => unreachable!(),
+                };
+                exports
+                    .into_iter()
+                    .next()
+                    .expect("SCC result export")
+                    .leave(scope)
+            });
+            output.probe_with(&mut probe);
+            input
+        });
+
+        let timer = Instant::now();
+        for index in 0..edges {
+            input.update(row_edge(index, nodes), 1);
+        }
+        input.advance_to(1);
+        input.flush();
+        while probe.less_than(input.time()) {
+            worker.step();
+        }
+        let load = timer.elapsed();
+
+        let mut times = Vec::with_capacity(rounds);
+        for round in 0..rounds {
+            let timer = Instant::now();
+            for offset in 0..batch {
+                let index = round * batch + offset;
+                input.update(row_edge(index, nodes), -1);
+                input.update(row_edge(edges + index, nodes), 1);
+            }
+            input.advance_to((round + 2) as u64);
+            input.flush();
+            while probe.less_than(input.time()) {
+                worker.step();
+            }
+            times.push(timer.elapsed());
+        }
+        worker
+            .log_register()
+            .expect("timely logging registry")
+            .flush();
+        let operators = finish_operator_profile(operator_profile);
+        let reduce_phases = if matches!(backend, Backend::Corgi) {
+            interactive::corgi::reduce::phase_profile()
+        } else {
+            Vec::new()
+        };
+        let proxy_reduce_phases = if matches!(backend, Backend::Corgi) {
+            differential_dataflow::operators::int_proxy::reduce::phase_profile()
+        } else {
+            Vec::new()
+        };
+        Stats {
+            load,
+            rounds: times,
+            operators,
+            reduce_phases,
+            proxy_reduce_phases,
+        }
+    })
+}
+
+fn median(mut values: Vec<Duration>) -> Duration {
+    values.sort_unstable();
+    values[values.len() / 2]
+}
+
+fn print_operator_profile(backend: Backend, operators: &[(String, Duration)]) {
+    if operators.is_empty() {
+        return;
+    }
+    let total: Duration = operators.iter().map(|(_, elapsed)| *elapsed).sum();
+    eprintln!(
+        "{} operator schedule profile (sum {:?}):",
+        backend.name(),
+        total
+    );
+    for (name, elapsed) in operators.iter().take(20) {
+        eprintln!(
+            "  {:>8.2}%  {:>12?}  {}",
+            100.0 * elapsed.as_secs_f64() / total.as_secs_f64(),
+            elapsed,
+            name,
+        );
+    }
+}
+
+fn print_reduce_profile(stats: &[interactive::corgi::reduce::ReducePhaseStat]) {
+    if stats.is_empty() {
+        return;
+    }
+    let measured: Duration = stats.iter().map(|stat| stat.elapsed).sum();
+    eprintln!("CorgiReduce exclusive phase profile (sum {measured:?}):");
+    for stat in stats {
+        eprintln!(
+            "  {:>8.2}%  {:>12?}  calls {:>8}  work {:>12}  {}",
+            100.0 * stat.elapsed.as_secs_f64() / measured.as_secs_f64(),
+            stat.elapsed,
+            stat.calls,
+            stat.work,
+            stat.phase,
+        );
+    }
+}
+
+fn print_proxy_reduce_profile(
+    stats: &[differential_dataflow::operators::int_proxy::reduce::ProxyReducePhaseStat],
+) {
+    if stats.is_empty() {
+        return;
+    }
+    let measured: Duration = stats.iter().map(|stat| stat.elapsed).sum();
+    eprintln!("ProxyReduce exclusive driver profile (sum {measured:?}):");
+    for stat in stats {
+        eprintln!(
+            "  {:>8.2}%  {:>12?}  calls {:>8}  work {:>12}  {}",
+            100.0 * stat.elapsed.as_secs_f64() / measured.as_secs_f64(),
+            stat.elapsed,
+            stat.calls,
+            stat.work,
+            stat.phase,
+        );
+    }
+}
+
+fn main() {
+    let nodes = env_usize("N", 100_000);
+    let edges = env_usize("E", 2 * nodes);
+    let batch = env_usize("BATCH", 1_000);
+    let rounds = env_usize("ROUNDS", 20);
+    let iters = env_usize("ITERS", 3);
+    assert!(nodes > 0, "N must be nonzero");
+    assert!(rounds > 0, "ROUNDS must be nonzero");
+    assert!(iters > 0, "ITERS must be nonzero");
+    assert!(
+        batch.saturating_mul(rounds) <= edges,
+        "BATCH * ROUNDS must not exceed E"
+    );
+
+    let selected = std::env::var("BACKENDS").unwrap_or_else(|_| "compiled,vec,corgi".to_owned());
+    let backends: Vec<Backend> = selected
+        .split(',')
+        .map(|name| match name.trim() {
+            "compiled" => Backend::Compiled,
+            "vec" => Backend::Vec,
+            "corgi" => Backend::Corgi,
+            other => panic!("unknown backend {other:?}"),
+        })
+        .collect();
+
+    let mut program = lower::lower_tree(parse::pipe::parse(SCC));
+    program.optimize();
+
+    if std::env::var("CHECK").map_or(true, |x| x != "0") {
+        let check_nodes = nodes.min(2_000);
+        let check_edges = edges.min(2 * check_nodes);
+        let rows: Vec<_> = (0..check_edges).map(|i| row_edge(i, check_nodes)).collect();
+        assert_eq!(
+            vec::evaluate(&program, std::slice::from_ref(&rows)),
+            corgi::evaluate(&program, &[rows]),
+            "DDIR Vec and Corgi disagree",
+        );
+        eprintln!("checked Vec/Corgi parity at n={check_nodes}, e={check_edges}");
+    }
+
+    println!("SCC n={nodes} e={edges} batch={batch} rounds={rounds} iterations={iters} workers=1");
+    println!("backend\tload_median\tround_median\trounds_total");
+    let mut results: BTreeMap<&str, (Duration, Duration, Duration)> = BTreeMap::new();
+    for backend in backends {
+        let mut loads = Vec::with_capacity(iters);
+        let mut round_medians = Vec::with_capacity(iters);
+        let mut round_totals = Vec::with_capacity(iters);
+        for iteration in 0..iters {
+            let stats = match backend {
+                Backend::Compiled => compiled(nodes, edges, batch, rounds),
+                Backend::Vec | Backend::Corgi => {
+                    interpreted(backend, &program, nodes, edges, batch, rounds)
+                }
+            };
+            let round_total: Duration = stats.rounds.iter().sum();
+            eprintln!(
+                "{} iteration {}: load {:?}, median round {:?}, rounds total {:?}",
+                backend.name(),
+                iteration + 1,
+                stats.load,
+                median(stats.rounds.clone()),
+                round_total,
+            );
+            print_operator_profile(backend, &stats.operators);
+            print_reduce_profile(&stats.reduce_phases);
+            print_proxy_reduce_profile(&stats.proxy_reduce_phases);
+            loads.push(stats.load);
+            round_medians.push(median(stats.rounds));
+            round_totals.push(round_total);
+        }
+        let result = (median(loads), median(round_medians), median(round_totals));
+        println!(
+            "{}\t{:?}\t{:?}\t{:?}",
+            backend.name(),
+            result.0,
+            result.1,
+            result.2
+        );
+        results.insert(backend.name(), result);
+    }
+
+    if let Some(compiled) = results.get("compiled") {
+        for name in ["vec", "corgi"] {
+            if let Some(result) = results.get(name) {
+                println!(
+                    "{name}/compiled\tload {:.2}x\tround {:.2}x\trounds_total {:.2}x",
+                    result.0.as_secs_f64() / compiled.0.as_secs_f64(),
+                    result.1.as_secs_f64() / compiled.1.as_secs_f64(),
+                    result.2.as_secs_f64() / compiled.2.as_secs_f64(),
+                );
+            }
+        }
+    }
+}

From f6124cb3fe2de616406b3a687ab6414f04bdfe28 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 16:38:36 -0400
Subject: [PATCH 4/8] Remove profiling and benchmark scaffolding

---
 .../src/operators/int_proxy/reduce.rs         |  97 ----
 interactive/examples/scc_compare.rs           | 502 ------------------
 interactive/src/corgi/reduce.rs               | 141 +----
 3 files changed, 8 insertions(+), 732 deletions(-)
 delete mode 100644 interactive/examples/scc_compare.rs

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index 52108087d..16a526cf1 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -3,10 +3,7 @@
 //! A conventional differential reduce against `(u64, u64)`, where the backend supplies the
 //! implementation of the interpretation of the integers.
 
-use std::cell::RefCell;
 use std::collections::BTreeMap;
-use std::sync::OnceLock;
-use std::time::{Duration, Instant};
 
 use timely::PartialOrder;
 use timely::progress::{Antichain, Timestamp};
@@ -19,72 +16,6 @@ use super::ProxyBridge;
 use crate::operators::reduce::{sort_dedup, ReduceTactic};
 use crate::operators::ValueHistory;
 
-/// One opt-in phase measurement from the generic proxy-reduce driver.
-#[derive(Clone, Debug)]
-pub struct ProxyReducePhaseStat {
-    /// Stable phase label.
-    pub phase: &'static str,
-    /// Number of timed invocations.
-    pub calls: u64,
-    /// Phase-specific row or slot count, useful only within the named phase.
-    pub work: u64,
-    /// Total wall-clock time spent in the phase.
-    pub elapsed: Duration,
-}
-
-#[derive(Default)]
-struct ProxyReducePhaseAccum {
-    calls: u64,
-    work: u64,
-    elapsed: Duration,
-}
-
-thread_local! {
-    static PROXY_REDUCE_PROFILE: RefCell<BTreeMap<&'static str, ProxyReducePhaseAccum>> = RefCell::new(BTreeMap::new());
-}
-
-fn proxy_profile_enabled() -> bool {
-    static ENABLED: OnceLock<bool> = OnceLock::new();
-    *ENABLED.get_or_init(|| std::env::var("PROXY_REDUCE_PROFILE").is_ok_and(|x| x != "0"))
-}
-
-#[inline]
-fn phase_start() -> Option<Instant> {
-    proxy_profile_enabled().then(Instant::now)
-}
-
-#[inline]
-fn phase_finish(phase: &'static str, start: Option<Instant>, work: usize) {
-    if let Some(start) = start {
-        PROXY_REDUCE_PROFILE.with(|profile| {
-            let mut profile = profile.borrow_mut();
-            let stat = profile.entry(phase).or_default();
-            stat.calls += 1;
-            stat.work += work as u64;
-            stat.elapsed += start.elapsed();
-        });
-    }
-}
-
-/// Clear proxy-reduce phase counters for the current worker thread.
-pub fn reset_phase_profile() {
-    PROXY_REDUCE_PROFILE.with(|profile| profile.borrow_mut().clear());
-}
-
-/// Snapshot current-thread proxy-reduce phase counters, slowest phase first.
-pub fn phase_profile() -> Vec<ProxyReducePhaseStat> {
-    PROXY_REDUCE_PROFILE.with(|profile| {
-        let mut stats: Vec<_> = profile.borrow().iter().map(|(&phase, stat)| ProxyReducePhaseStat {
-            phase,
-            calls: stat.calls,
-            work: stat.work,
-            elapsed: stat.elapsed,
-        }).collect();
-        stats.sort_unstable_by_key(|stat| std::cmp::Reverse(stat.elapsed));
-        stats
-    })
-}
-
 /// A unit of proxied reduce work, presented to the backend.
 pub struct ReduceInstance<'a, B1: BatchReader, B2: BatchReader<Time = B1::Time>> {
     /// The accumulated input history.
@@ -231,9 +162,7 @@ where
         upper: &Antichain<B1::Time>,
         held: &Antichain<B1::Time>,
     ) -> (Vec<(B1::Time, B2)>, Antichain<B1::Time>) {
-        let retire_phase = phase_start();
         if held.elements().iter().all(|t| upper.less_equal(t)) {
-            phase_finish("retire.total_inclusive", retire_phase, 0);
             return (Vec::new(), held.clone());
         }
 
@@ -251,7 +180,6 @@ where
         // due subset changes.
         let mut due: BTreeMap<u64, Vec<B1::Time>> = BTreeMap::new();
         let mut pending_frontier = Antichain::new();
-        let phase = phase_start();
         for (&key, times) in self.pending.iter_mut() {
             let mut ready = Vec::new();
             times.retain(|time| {
@@ -263,7 +191,6 @@ where
             if !ready.is_empty() { due.insert(key, ready); }
         }
         self.pending.retain(|_, times| !times.is_empty());
-        phase_finish("retire.partition_pending", phase, self.pending.len());
         // The keys the harness knows must be revisited. The backend adds those its novel batches
         // touch, which it discovers while reading them; neither side scans the whole key space.
         let changed: Vec<u64> = due.keys().copied().collect();
@@ -274,17 +201,14 @@ where
         // beyond `upper` can remain when nothing is due, and releasing their capabilities would
         // strand them (see the frontier clause of the `ReduceTactic::retire` contract).
         if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) {
-            phase_finish("retire.total_inclusive", retire_phase, 0);
             return (Vec::new(), pending_frontier);
         }
 
         // The output tiling (identical to the Abelian tactic): one tile per held time, keeping
         // non-degenerate intervals; `tile_of[i]` maps held time `i` to its tile.
         let held_elems: Vec<B1::Time> = held.elements().to_vec();
-        let phase = phase_start();
         let (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems);
         self.backend.begin(&tile_descs);
-        phase_finish("retire.describe_and_begin", phase, tile_descs.len());
 
         // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None`
         // once the backend reports the space covered.
@@ -294,7 +218,6 @@ where
         // Retire-wide reusable scratch: cleared per window or wave, never reallocated. Fresh
         // per-key/per-wave `Vec`s were once the dominant cost here, which is why the slots and the
         // staging buffers are held across the whole retire rather than built where they are used.
-        let phase = phase_start();
         let mut slots: Vec<KeySweep<B1::Time, Bk::RIn, Bk::ROut>> = Vec::new();
         let mut live: Vec<usize> = Vec::new();
         let mut tile_deltas: Vec<Vec<((u64, u64), B1::Time, Bk::ROut)>> = (0..held_elems.len()).map(|_| Vec::new()).collect();
@@ -306,14 +229,11 @@ where
         let mut active: Vec<(usize, B1::Time)> = Vec::new();
         let mut in_accum: Vec<(u64, Bk::RIn)> = Vec::new();
         let mut cur_out: Vec<(u64, Bk::ROut)> = Vec::new();
-        phase_finish("retire.allocate_scratch", phase, held_elems.len());
 
         while from.is_some() {
             let before = from;
             window.clear();
-            let phase = phase_start();
             self.backend.next_window(&instance, &changed, &mut from, &mut window);
-            phase_finish("window.backend_present", phase, window.input.len() + window.output.len() + window.seeds.len());
             let p_in = &window.input;
             let seeds = &window.seeds;
             let p_out = &window.output;
@@ -356,7 +276,6 @@ where
             let mut n_slots = 0usize;
             let (mut is, mut ns, mut os) = (0usize, 0usize, 0usize);
             live.clear();
-            let phase = phase_start();
             // Mapped to hashes before the min: the sources differ in shape.
             while let Some(key) = [
                 p_in.get(is).map(|record| record.0.0),
@@ -393,7 +312,6 @@ where
                 }
                 n_slots += 1;
             }
-            phase_finish("window.initialize_sweeps", phase, p_in.len() + seeds.len() + p_out.len());
 
             // Each wave: read every suspended key's accumulations, cross the non-empty ones in one
             // call, hand the corrections back, and step every live key on. A key retires when its
@@ -406,7 +324,6 @@ where
                 out_all.clear();
                 active.clear();
 
-                let phase = phase_start();
                 for &si in live.iter() {
                     let at = slots[si].at.clone().expect("live slots are suspended at a time");
                     in_accum.clear();
@@ -423,13 +340,9 @@ where
                     out_ends.push(out_all.len());
                     active.push((si, at));
                 }
-                phase_finish("wave.accumulate", phase, live.len());
 
                 if !batch_keys.is_empty() {
-                    let phase = phase_start();
                     let (corr, corr_ends) = self.backend.reduce_corrections(&batch_keys, &in_ends, &in_all, &out_ends, &out_all);
-                    phase_finish("wave.backend_corrections", phase, in_all.len() + out_all.len());
-                    let phase = phase_start();
                     let mut cstart = 0usize;
                     for (bi, (si, at)) in active.iter().enumerate() {
                         let cend = corr_ends[bi];
@@ -442,11 +355,9 @@ where
                         }
                         cstart = cend;
                     }
-                    phase_finish("wave.commit_corrections", phase, corr.len());
                 }
 
                 // Step every live key past the time it was suspended at, and retire the spent ones.
-                let phase = phase_start();
                 for &si in live.iter() {
                     let slot = &mut slots[si];
                     slot.at = slot.sweep.next_crossing(upper, &mut slot.pended);
@@ -458,28 +369,20 @@ where
                     }
                 }
                 live.retain(|&si| slots[si].at.is_some());
-                phase_finish("wave.advance_sweeps", phase, live.len());
             }
 
-            let phase = phase_start();
-            let mut emit_work = 0usize;
             for (held_index, deltas) in tile_deltas.iter_mut().enumerate() {
                 if deltas.is_empty() {
                     continue;
                 }
                 if let Some(tile) = tile_of[held_index] {
-                    emit_work += deltas.len();
                     crate::consolidation::consolidate_updates(deltas);
                     self.backend.emit(tile, &deltas[..]);
                 }
             }
-            phase_finish("window.consolidate_and_emit", phase, emit_work);
         }
 
-        let phase = phase_start();
         let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
-        phase_finish("retire.backend_finish", phase, produced.len());
-        phase_finish("retire.total_inclusive", retire_phase, produced.len());
         (produced, pending_frontier)
     }
 }
diff --git a/interactive/examples/scc_compare.rs b/interactive/examples/scc_compare.rs
deleted file mode 100644
index 0609c5b6f..000000000
--- a/interactive/examples/scc_compare.rs
+++ /dev/null
@@ -1,502 +0,0 @@
-//! Apples-to-apples SCC benchmark: compiled Differential vs DDIR Vec vs DDIR Corgi.
-//!
-//! All three variants consume the same deterministic graph and update stream. The compiled
-//! baseline uses `strongly_connected_at(_, |x| *x as u64)`, matching the logarithmic label staging
-//! performed by DDIR's `enter_at($1[0])`. The measured output is only probed; it is not inspected or
-//! captured. DDIR parsing, lowering, optimization, and dataflow construction are outside the load
-//! and update timers.
-//!
-//! ```text
-//! N=100000 E=200000 BATCH=1000 ROUNDS=20 ITERS=3 \
-//!   cargo run --release -p interactive --example scc_compare
-//! ```
-//!
-//! `BACKENDS=compiled,vec,corgi` selects variants. `CHECK=1` first checks Vec/Corgi result parity
-//! on a bounded prefix of the graph (default on). Corgi is currently single-worker, so this
-//! benchmark intentionally uses `timely::execute_directly` for every variant.
-
-use std::cell::RefCell;
-use std::collections::BTreeMap;
-use std::rc::Rc;
-use std::time::{Duration, Instant};
-
-use differential_dataflow::algorithms::graphs::scc::strongly_connected_at;
-use differential_dataflow::dynamic::pointstamp::PointStamp;
-use differential_dataflow::input::Input;
-use interactive::backend::{corgi, vec};
-use interactive::ir::{Diff, Value};
-use interactive::scope_ir as st;
-use interactive::{lower, parse};
-use timely::dataflow::operators::probe::Handle;
-use timely::logging::{StartStop, TimelyEvent, TimelyEventBuilder};
-
-#[global_allocator]
-static ALLOC: mimalloc::MiMalloc = mimalloc::MiMalloc;
-
-const SCC: &str = r#"
-let edges = input 0 | key($0[0] ; $0[1]);
-let trans = edges | key($1 ; $0);
-
-outer: {
-    let scc = edges + trim;
-    fwd: {
-        let nodes = edges | key($1 ; $1) | enter_at($1[0]);
-        let labels = proposals + nodes | min;
-        var proposals = labels | join(scc, ($2 ; $1));
-    }
-    let trim_fwd = edges
-        | join(fwd::labels, ($1 ; $0, $2))
-        | join(fwd::labels, ($0 ; $1, $2))
-        | filter($1[1] == $1[2])
-        | key($0 ; $1[0]);
-    bwd: {
-        let nodes = trans | key($1 ; $1) | enter_at($1[0]);
-        let labels = proposals + nodes | min;
-        var proposals = labels | join(trim_fwd, ($2 ; $1));
-    }
-    let trim_bwd = trans
-        | join(bwd::labels, ($1 ; $0, $2))
-        | join(bwd::labels, ($0 ; $1, $2))
-        | filter($1[1] == $1[2])
-        | key($0 ; $1[0]);
-    var trim = trim_bwd - edges;
-}
-
-export "result" = outer::scc | map(;) | arrange;
-"#;
-
-#[derive(Clone, Copy, Debug)]
-enum Backend {
-    Compiled,
-    Vec,
-    Corgi,
-}
-
-impl Backend {
-    fn name(self) -> &'static str {
-        match self {
-            Backend::Compiled => "compiled",
-            Backend::Vec => "vec",
-            Backend::Corgi => "corgi",
-        }
-    }
-}
-
-#[derive(Debug)]
-struct Stats {
-    load: Duration,
-    rounds: Vec<Duration>,
-    operators: Vec<(String, Duration)>,
-    reduce_phases: Vec<interactive::corgi::reduce::ReducePhaseStat>,
-    proxy_reduce_phases:
-        Vec<differential_dataflow::operators::int_proxy::reduce::ProxyReducePhaseStat>,
-}
-
-#[derive(Default)]
-struct OperatorProfile {
-    operators: BTreeMap<usize, (String, Vec<usize>)>,
-    elapsed: BTreeMap<usize, Duration>,
-}
-
-macro_rules! install_operator_profile {
-    ($worker:expr) => {{
-        let profile = Rc::new(RefCell::new(OperatorProfile::default()));
-        if std::env::var("PROFILE_OPS").is_ok_and(|x| x != "0") {
-            let target = Rc::clone(&profile);
-            let mut starts = BTreeMap::new();
-            $worker
-                .log_register()
-                .expect("timely logging registry")
-                .insert::<TimelyEventBuilder, _>("timely", move |_time, data| {
-                    if let Some(events) = data {
-                        for (event_time, event) in events.iter() {
-                            match event {
-                                TimelyEvent::Operates(event) => {
-                                    target
-                                        .borrow_mut()
-                                        .operators
-                                        .insert(event.id, (event.name.clone(), event.addr.clone()));
-                                }
-                                TimelyEvent::Schedule(event) => match event.start_stop {
-                                    StartStop::Start => {
-                                        starts.insert(event.id, *event_time);
-                                    }
-                                    StartStop::Stop => {
-                                        if let Some(start) = starts.remove(&event.id) {
-                                            *target
-                                                .borrow_mut()
-                                                .elapsed
-                                                .entry(event.id)
-                                                .or_default() += event_time.saturating_sub(start);
-                                        }
-                                    }
-                                },
-                                _ => {}
-                            }
-                        }
-                    }
-                });
-        }
-        profile
-    }};
-}
-
-fn finish_operator_profile(profile: Rc<RefCell<OperatorProfile>>) -> Vec<(String, Duration)> {
-    let profile = profile.borrow();
-    let addresses: Vec<&Vec<usize>> = profile
-        .operators
-        .values()
-        .map(|(_, address)| address)
-        .collect();
-    let mut operators: Vec<_> = profile
-        .elapsed
-        .iter()
-        .filter_map(|(id, elapsed)| {
-            let (name, address) = profile
-                .operators
-                .get(id)
-                .cloned()
-                .unwrap_or_else(|| (format!("operator-{id}"), Vec::new()));
-            let has_child = addresses
-                .iter()
-                .any(|other| other.len() > address.len() && other.starts_with(&address));
-            (!has_child).then_some((name, *elapsed))
-        })
-        .collect();
-    operators.sort_unstable_by_key(|(_, elapsed)| std::cmp::Reverse(*elapsed));
-    operators
-}
-
-fn env_usize(name: &str, default: usize) -> usize {
-    std::env::var(name)
-        .ok()
-        .and_then(|x| x.parse().ok())
-        .unwrap_or(default)
-}
-
-fn mix(mut x: u64) -> u64 {
-    x = x.wrapping_add(0x9e3779b97f4a7c15);
-    x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
-    x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
-    x ^ (x >> 31)
-}
-
-fn edge(index: usize, nodes: usize) -> (usize, usize) {
-    let first = mix(index as u64);
-    let second = mix(first);
-    ((first as usize) % nodes, (second as usize) % nodes)
-}
-
-fn row_edge(index: usize, nodes: usize) -> (Value, Value) {
-    let (src, dst) = edge(index, nodes);
-    (
-        Value::Tuple(vec![Value::Int(src as i64), Value::Int(dst as i64)]),
-        Value::unit(),
-    )
-}
-
-fn compiled(nodes: usize, edges: usize, batch: usize, rounds: usize) -> Stats {
-    timely::execute_directly(move |worker| {
-        let operator_profile = install_operator_profile!(worker);
-        let mut probe = Handle::new();
-        let mut input = worker.dataflow::<usize, _, _>(|scope| {
-            let (input, graph) = scope.new_collection::<(usize, usize), isize>();
-            strongly_connected_at(graph, |x| *x as u64).probe_with(&mut probe);
-            input
-        });
-
-        let timer = Instant::now();
-        for index in 0..edges {
-            input.insert(edge(index, nodes));
-        }
-        input.advance_to(1);
-        input.flush();
-        while probe.less_than(input.time()) {
-            worker.step();
-        }
-        let load = timer.elapsed();
-
-        let mut times = Vec::with_capacity(rounds);
-        for round in 0..rounds {
-            let timer = Instant::now();
-            for offset in 0..batch {
-                let index = round * batch + offset;
-                input.remove(edge(index, nodes));
-                input.insert(edge(edges + index, nodes));
-            }
-            input.advance_to(round + 2);
-            input.flush();
-            while probe.less_than(input.time()) {
-                worker.step();
-            }
-            times.push(timer.elapsed());
-        }
-        worker
-            .log_register()
-            .expect("timely logging registry")
-            .flush();
-        let operators = finish_operator_profile(operator_profile);
-        Stats {
-            load,
-            rounds: times,
-            operators,
-            reduce_phases: Vec::new(),
-            proxy_reduce_phases: Vec::new(),
-        }
-    })
-}
-
-fn interpreted(
-    backend: Backend,
-    program: &st::Program,
-    nodes: usize,
-    edges: usize,
-    batch: usize,
-    rounds: usize,
-) -> Stats {
-    let program = program.clone();
-    timely::execute_directly(move |worker| {
-        if matches!(backend, Backend::Corgi) {
-            interactive::corgi::reduce::reset_phase_profile();
-            differential_dataflow::operators::int_proxy::reduce::reset_phase_profile();
-        }
-        let operator_profile = install_operator_profile!(worker);
-        let mut probe = Handle::new();
-        let mut input = worker.dataflow::<u64, _, _>(|scope| {
-            let (input, collection) = scope.new_collection::<(Value, Value), Diff>();
-            let output = scope.iterative::<PointStamp<u64>, _, _>(|inner| {
-                let entered = collection.clone().enter(inner);
-                let imports = program
-                    .root
-                    .imports
-                    .iter()
-                    .map(|import| match &import.from {
-                        st::Source::Input(0) => entered.clone(),
-                        ref other => panic!("unexpected SCC import {other:?}"),
-                    })
-                    .collect();
-                let exports = match backend {
-                    Backend::Vec => vec::render_tree(&program.root, inner.clone(), 0, imports),
-                    Backend::Corgi => {
-                        corgi::render_tree_rows(&program.root, inner.clone(), 0, imports)
-                    }
-                    Backend::Compiled => unreachable!(),
-                };
-                exports
-                    .into_iter()
-                    .next()
-                    .expect("SCC result export")
-                    .leave(scope)
-            });
-            output.probe_with(&mut probe);
-            input
-        });
-
-        let timer = Instant::now();
-        for index in 0..edges {
-            input.update(row_edge(index, nodes), 1);
-        }
-        input.advance_to(1);
-        input.flush();
-        while probe.less_than(input.time()) {
-            worker.step();
-        }
-        let load = timer.elapsed();
-
-        let mut times = Vec::with_capacity(rounds);
-        for round in 0..rounds {
-            let timer = Instant::now();
-            for offset in 0..batch {
-                let index = round * batch + offset;
-                input.update(row_edge(index, nodes), -1);
-                input.update(row_edge(edges + index, nodes), 1);
-            }
-            input.advance_to((round + 2) as u64);
-            input.flush();
-            while probe.less_than(input.time()) {
-                worker.step();
-            }
-            times.push(timer.elapsed());
-        }
-        worker
-            .log_register()
-            .expect("timely logging registry")
-            .flush();
-        let operators = finish_operator_profile(operator_profile);
-        let reduce_phases = if matches!(backend, Backend::Corgi) {
-            interactive::corgi::reduce::phase_profile()
-        } else {
-            Vec::new()
-        };
-        let proxy_reduce_phases = if matches!(backend, Backend::Corgi) {
-            differential_dataflow::operators::int_proxy::reduce::phase_profile()
-        } else {
-            Vec::new()
-        };
-        Stats {
-            load,
-            rounds: times,
-            operators,
-            reduce_phases,
-            proxy_reduce_phases,
-        }
-    })
-}
-
-fn median(mut values: Vec<Duration>) -> Duration {
-    values.sort_unstable();
-    values[values.len() / 2]
-}
-
-fn print_operator_profile(backend: Backend, operators: &[(String, Duration)]) {
-    if operators.is_empty() {
-        return;
-    }
-    let total: Duration = operators.iter().map(|(_, elapsed)| *elapsed).sum();
-    eprintln!(
-        "{} operator schedule profile (sum {:?}):",
-        backend.name(),
-        total
-    );
-    for (name, elapsed) in operators.iter().take(20) {
-        eprintln!(
-            "  {:>8.2}%  {:>12?}  {}",
-            100.0 * elapsed.as_secs_f64() / total.as_secs_f64(),
-            elapsed,
-            name,
-        );
-    }
-}
-
-fn print_reduce_profile(stats: &[interactive::corgi::reduce::ReducePhaseStat]) {
-    if stats.is_empty() {
-        return;
-    }
-    let measured: Duration = stats.iter().map(|stat| stat.elapsed).sum();
-    eprintln!("CorgiReduce exclusive phase profile (sum {measured:?}):");
-    for stat in stats {
-        eprintln!(
-            "  {:>8.2}%  {:>12?}  calls {:>8}  work {:>12}  {}",
-            100.0 * stat.elapsed.as_secs_f64() / measured.as_secs_f64(),
-            stat.elapsed,
-            stat.calls,
-            stat.work,
-            stat.phase,
-        );
-    }
-}
-
-fn print_proxy_reduce_profile(
-    stats: &[differential_dataflow::operators::int_proxy::reduce::ProxyReducePhaseStat],
-) {
-    if stats.is_empty() {
-        return;
-    }
-    let measured: Duration = stats.iter().map(|stat| stat.elapsed).sum();
-    eprintln!("ProxyReduce exclusive driver profile (sum {measured:?}):");
-    for stat in stats {
-        eprintln!(
-            "  {:>8.2}%  {:>12?}  calls {:>8}  work {:>12}  {}",
-            100.0 * stat.elapsed.as_secs_f64() / measured.as_secs_f64(),
-            stat.elapsed,
-            stat.calls,
-            stat.work,
-            stat.phase,
-        );
-    }
-}
-
-fn main() {
-    let nodes = env_usize("N", 100_000);
-    let edges = env_usize("E", 2 * nodes);
-    let batch = env_usize("BATCH", 1_000);
-    let rounds = env_usize("ROUNDS", 20);
-    let iters = env_usize("ITERS", 3);
-    assert!(nodes > 0, "N must be nonzero");
-    assert!(rounds > 0, "ROUNDS must be nonzero");
-    assert!(iters > 0, "ITERS must be nonzero");
-    assert!(
-        batch.saturating_mul(rounds) <= edges,
-        "BATCH * ROUNDS must not exceed E"
-    );
-
-    let selected = std::env::var("BACKENDS").unwrap_or_else(|_| "compiled,vec,corgi".to_owned());
-    let backends: Vec<Backend> = selected
-        .split(',')
-        .map(|name| match name.trim() {
-            "compiled" => Backend::Compiled,
-            "vec" => Backend::Vec,
-            "corgi" => Backend::Corgi,
-            other => panic!("unknown backend {other:?}"),
-        })
-        .collect();
-
-    let mut program = lower::lower_tree(parse::pipe::parse(SCC));
-    program.optimize();
-
-    if std::env::var("CHECK").map_or(true, |x| x != "0") {
-        let check_nodes = nodes.min(2_000);
-        let check_edges = edges.min(2 * check_nodes);
-        let rows: Vec<_> = (0..check_edges).map(|i| row_edge(i, check_nodes)).collect();
-        assert_eq!(
-            vec::evaluate(&program, std::slice::from_ref(&rows)),
-            corgi::evaluate(&program, &[rows]),
-            "DDIR Vec and Corgi disagree",
-        );
-        eprintln!("checked Vec/Corgi parity at n={check_nodes}, e={check_edges}");
-    }
-
-    println!("SCC n={nodes} e={edges} batch={batch} rounds={rounds} iterations={iters} workers=1");
-    println!("backend\tload_median\tround_median\trounds_total");
-    let mut results: BTreeMap<&str, (Duration, Duration, Duration)> = BTreeMap::new();
-    for backend in backends {
-        let mut loads = Vec::with_capacity(iters);
-        let mut round_medians = Vec::with_capacity(iters);
-        let mut round_totals = Vec::with_capacity(iters);
-        for iteration in 0..iters {
-            let stats = match backend {
-                Backend::Compiled => compiled(nodes, edges, batch, rounds),
-                Backend::Vec | Backend::Corgi => {
-                    interpreted(backend, &program, nodes, edges, batch, rounds)
-                }
-            };
-            let round_total: Duration = stats.rounds.iter().sum();
-            eprintln!(
-                "{} iteration {}: load {:?}, median round {:?}, rounds total {:?}",
-                backend.name(),
-                iteration + 1,
-                stats.load,
-                median(stats.rounds.clone()),
-                round_total,
-            );
-            print_operator_profile(backend, &stats.operators);
-            print_reduce_profile(&stats.reduce_phases);
-            print_proxy_reduce_profile(&stats.proxy_reduce_phases);
-            loads.push(stats.load);
-            round_medians.push(median(stats.rounds));
-            round_totals.push(round_total);
-        }
-        let result = (median(loads), median(round_medians), median(round_totals));
-        println!(
-            "{}\t{:?}\t{:?}\t{:?}",
-            backend.name(),
-            result.0,
-            result.1,
-            result.2
-        );
-        results.insert(backend.name(), result);
-    }
-
-    if let Some(compiled) = results.get("compiled") {
-        for name in ["vec", "corgi"] {
-            if let Some(result) = results.get(name) {
-                println!(
-                    "{name}/compiled\tload {:.2}x\tround {:.2}x\trounds_total {:.2}x",
-                    result.0.as_secs_f64() / compiled.0.as_secs_f64(),
-                    result.1.as_secs_f64() / compiled.1.as_secs_f64(),
-                    result.2.as_secs_f64() / compiled.2.as_secs_f64(),
-                );
-            }
-        }
-    }
-}
diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 81fd6c127..089a2ff5e 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -26,14 +26,11 @@
 //! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes
 //! (a columnar semijoin — matching the row-wise tactic's read).
 
-use std::cell::RefCell;
 use std::cmp::Reverse;
 use std::collections::BinaryHeap;
-use std::collections::{BTreeMap, HashMap};
+use std::collections::HashMap;
 use std::hash::{BuildHasherDefault, Hasher};
 use std::rc::Rc;
-use std::sync::OnceLock;
-use std::time::{Duration, Instant};
 
 use differential_dataflow::consolidation::consolidate_updates;
 use differential_dataflow::trace::Description;
@@ -51,73 +48,6 @@ use crate::parse::Reducer;
 
 type CBatch<T> = Rc<ChunkBatch<CorgiChunk<T, Diff>>>;
 
-/// One opt-in `CorgiReduceBackend` phase measurement. `work` is the phase's natural row count;
-/// it is diagnostic context, not a quantity comparable across differently named phases.
-#[derive(Clone, Debug)]
-pub struct ReducePhaseStat {
-    pub phase: &'static str,
-    pub calls: u64,
-    pub work: u64,
-    pub elapsed: Duration,
-}
-
-#[derive(Default)]
-struct ReducePhaseAccum {
-    calls: u64,
-    work: u64,
-    elapsed: Duration,
-}
-
-thread_local! {
-    static REDUCE_PROFILE: RefCell<BTreeMap<&'static str, ReducePhaseAccum>> = RefCell::new(BTreeMap::new());
-}
-
-fn reduce_profile_enabled() -> bool {
-    static ENABLED: OnceLock<bool> = OnceLock::new();
-    *ENABLED.get_or_init(|| std::env::var("CORGI_REDUCE_PROFILE").is_ok_and(|x| x != "0"))
-}
-
-#[inline]
-fn phase_start() -> Option<Instant> {
-    reduce_profile_enabled().then(Instant::now)
-}
-
-#[inline]
-fn phase_finish(phase: &'static str, start: Option<Instant>, work: usize) {
-    if let Some(start) = start {
-        REDUCE_PROFILE.with(|profile| {
-            let mut profile = profile.borrow_mut();
-            let stat = profile.entry(phase).or_default();
-            stat.calls += 1;
-            stat.work += work as u64;
-            stat.elapsed += start.elapsed();
-        });
-    }
-}
-
-/// Clear phase counters for the current worker thread.
-pub fn reset_phase_profile() {
-    REDUCE_PROFILE.with(|profile| profile.borrow_mut().clear());
-}
-
-/// Snapshot the current worker thread's phase counters, slowest phase first.
-pub fn phase_profile() -> Vec<ReducePhaseStat> {
-    REDUCE_PROFILE.with(|profile| {
-        let mut stats: Vec<_> = profile
-            .borrow()
-            .iter()
-            .map(|(&phase, stat)| ReducePhaseStat {
-                phase,
-                calls: stat.calls,
-                work: stat.work,
-                elapsed: stat.elapsed,
-            })
-            .collect();
-        stats.sort_unstable_by_key(|stat| std::cmp::Reverse(stat.elapsed));
-        stats
-    })
-}
-
 /// An identity `Hasher` for the id-index maps: their keys are already well-distributed 64-bit
 /// content hashes (`hash_rows`), so passing the id straight through avoids re-hashing it (siphash
 /// on `register_keys`/lookups was ~7% of the reduce in profiling). Only `write_u64` is used.
@@ -286,12 +216,7 @@ where
     /// (~log(rows) compares, each far costlier than the scan's flat membership test), so
     /// marginal seeks LOSE to the scan — measured, not modeled; 16 regressed load-shaped
     /// retires before this was widened.
-    fn seek_advantage() -> usize {
-        static ADVANTAGE: OnceLock<usize> = OnceLock::new();
-        *ADVANTAGE.get_or_init(|| {
-            std::env::var("CORGI_SEEK_ADVANTAGE").ok().and_then(|x| x.parse().ok()).unwrap_or(64)
-        })
-    }
+    const SEEK_ADVANTAGE: usize = 64;
 
     let key_srcs: Vec<Option<&CValue>> = chunks.iter().map(|c| Some(c.keys())).collect();
     let val_srcs: Vec<Option<&CValue>> = chunks.iter().map(|c| Some(c.vals())).collect();
@@ -299,13 +224,12 @@ where
     let (mut khs, mut times, mut diffs) = (Vec::new(), Vec::new(), Vec::new());
     let mut run_ends = Vec::new();
     let total: usize = chunks.iter().map(|c| c.diffs().len()).sum();
-    let needles = if changed.len().saturating_mul(seek_advantage()) < total {
+    let needles = if changed.len().saturating_mul(SEEK_ADVANTAGE) < total {
         chunks.iter().find(|c| c.diffs().len() > 0).and_then(|c| seek_needles(c.keys(), changed))
     } else {
         None
     };
     if let Some(needles) = needles {
-        let phase = phase_start();
         // Narrow changed set over seekable keys: gallop each chunk once per changed key.
         // Chunks are key-ordered and `changed` ascends, so emission order matches the scan's.
         for (ci, ch) in chunks.iter().enumerate() {
@@ -325,9 +249,7 @@ where
             }
             if khs.len() > before { run_ends.push(khs.len()); }
         }
-        phase_finish("present.seek_chunks", phase, total);
     } else {
-        let phase = phase_start();
         for (ci, ch) in chunks.iter().enumerate() {
             let before = khs.len();
             // Borrow the key leaf when there is one (`ids`' value-as-id fast paths); only
@@ -351,23 +273,15 @@ where
             }
             if khs.len() > before { run_ends.push(khs.len()); }
         }
-        phase_finish("present.scan_chunks", phase, total);
     }
     if tags.is_empty() {
         return (CValue::Unit(0), CValue::Unit(0), khs, times, diffs, run_ends);
     }
-    let phase = phase_start();
     let keys_col = gather_lanes(&key_srcs, &tags, &offs);
     let vals_col = gather_lanes(&val_srcs, &tags, &offs);
-    phase_finish("present.gather_columns", phase, tags.len());
     (keys_col, vals_col, khs, times, diffs, run_ends)
 }
 
-fn merge_present_enabled() -> bool {
-    static ENABLED: OnceLock<bool> = OnceLock::new();
-    *ENABLED.get_or_init(|| std::env::var("CORGI_MERGE_PRESENT").map_or(true, |x| x != "0"))
-}
-
 /// Merge already-ordered selected chunk runs directly into a consolidated proxy bridge. Returns
 /// false when a run's structural value order does not agree with its proxy-id order.
 fn merge_present<T: Ord + Clone>(
@@ -441,22 +355,14 @@ where
         if khs.is_empty() {
             return;
         }
-        let phase = phase_start();
         let vids = ids(&p_vals);
         for (row, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(*len + row); }
         *len += p_vals.len();
         blocks.push(p_vals);
         self.register_keys(p_keys, &khs);
-        phase_finish("present.register_input", phase, khs.len());
-        let phase = phase_start();
-        if merge_present_enabled() && merge_present(&khs, &vids, &times, &diffs, &run_ends, bridge) {
-            phase_finish("present.merge_input", phase, khs.len());
-        } else {
+        if !merge_present(&khs, &vids, &times, &diffs, &run_ends, bridge) {
             bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i])));
-            phase_finish("present.build_input_bridge", phase, khs.len());
-            let phase = phase_start();
             consolidate_updates(bridge);
-            phase_finish("present.consolidate_input", phase, bridge.len());
         }
     }
 
@@ -637,7 +543,6 @@ where
         // The seeds are the novel batches' RAW (key_hash, time) support, recorded here — before the
         // merged presentation below, whose consolidation may net a novel record away entirely. The
         // key hashes come from the scan the key list needs anyway.
-        let phase = phase_start();
         let mut seeds: Vec<(u64, T)> = Vec::new();
         for ch in novel_chunks.iter() {
             let khs = ids(ch.keys());
@@ -669,7 +574,6 @@ where
             }
             keys = merged;
         }
-        phase_finish("window.seed_and_keys", phase, novel_chunks.iter().map(|c| c.diffs().len()).sum::<usize>() + changed.len());
         if keys.is_empty() {
             self.in_vals = CValue::Unit(0);
             self.in_index = IdMap::default();
@@ -685,50 +589,30 @@ where
         let mut in_len = 0usize;
         let mut in_chunks = chunks_of(instance.source_batches);
         in_chunks.extend(novel_chunks.iter().copied());
-        let phase = phase_start();
         self.present_input(&in_chunks, &keys, &mut in_blocks, &mut in_len, &mut window.input);
-        phase_finish("window.present_input", phase, in_chunks.iter().map(|c| c.diffs().len()).sum());
-        let phase = phase_start();
         self.in_vals = concat_columns(&in_blocks);
-        phase_finish("window.concat_input", phase, in_len);
 
         // Output-history presentation, same keys (register keys + values for correction resolution).
-        let output_chunks = chunks_of(instance.output_batches);
-        let phase = phase_start();
-        let (o_keys, o_vals, o_khs, o_times, o_diffs, o_run_ends) = collect_present(&output_chunks, &keys);
+        let (o_keys, o_vals, o_khs, o_times, o_diffs, o_run_ends) = collect_present(&chunks_of(instance.output_batches), &keys);
         if !o_khs.is_empty() {
-            let subphase = phase_start();
             let vids = ids(&o_vals);
             self.register_keys(o_keys, &o_khs);
             self.register_vals(o_vals, &vids);
-            phase_finish("present.register_output", subphase, o_khs.len());
-            let subphase = phase_start();
-            if merge_present_enabled() && merge_present(&o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output) {
-                phase_finish("present.merge_output", subphase, o_khs.len());
-            } else {
+            if !merge_present(&o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output) {
                 window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])));
-                phase_finish("present.build_output_bridge", subphase, o_khs.len());
-                let subphase = phase_start();
                 consolidate_updates(&mut window.output);
-                phase_finish("present.consolidate_output", subphase, window.output.len());
             }
         }
-        phase_finish("window.present_output", phase, output_chunks.iter().map(|c| c.diffs().len()).sum());
     }
 
     fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec<usize>) {
         // Resolve input value_ids to `in_vals` rows, reduce (desired output), then difference the
         // desired against the presented current output per key: correction = desired − current.
-        let phase = phase_start();
         let in_rows: Vec<(usize, Diff)> = input.iter()
             .map(|&(vid, d)| (*self.in_index.get(&vid).expect("input value_id presented this window"), d))
             .collect();
-        phase_finish("corrections.resolve_input", phase, input.len());
-        let phase = phase_start();
         let (desired, desired_ends) = self.reduce_brackets(in_ends, &in_rows);
-        phase_finish("corrections.reduce_brackets", phase, in_rows.len());
 
-        let phase = phase_start();
         let mut corr: Vec<(u64, Diff)> = Vec::new();
         let mut corr_ends: Vec<usize> = Vec::with_capacity(keys.len());
         let (mut ds, mut os) = (0usize, 0usize);
@@ -751,13 +635,11 @@ where
             ds = de;
             os = oe;
         }
-        phase_finish("corrections.diff_output", phase, desired.len() + output.len());
         (corr, corr_ends)
     }
 
     fn emit(&mut self, tile: usize, records: &[((u64, u64), T, Diff)]) {
         // Resolve each correction's key/value proxies to pool rows and accumulate into the tile.
-        let phase = phase_start();
         for rec in records {
             let ((kh, vid), t, d) = (rec.0, &rec.1, rec.2);
             let kr = *self.key_index.get(&kh).expect("key resolvable this retire");
@@ -768,25 +650,18 @@ where
             times.push(t.clone());
             diffs.push(d);
         }
-        phase_finish("emit.resolve_and_buffer", phase, records.len());
     }
 
     fn finish(&mut self) -> Vec<CBatch<T>> {
         // Seal each tile: gather its accumulated (key, val) pool rows into columns, one CorgiChunk batch.
-        let phase = phase_start();
         let key_pool = concat_columns(&self.key_blocks);
         let val_pool = concat_columns(&self.val_blocks);
-        phase_finish("finish.concat_pools", phase, self.key_len + self.val_len);
         let tiles = std::mem::take(&mut self.tiles);
         let tile_rows = std::mem::take(&mut self.tile_rows);
-        let output_rows: usize = tile_rows.iter().map(|rows| rows.0.len()).sum();
-        let phase = phase_start();
-        let result = tiles.into_iter().zip(tile_rows).map(|(desc, (krows, vrows, times, diffs))| {
+        tiles.into_iter().zip(tile_rows).map(|(desc, (krows, vrows, times, diffs))| {
             let keys = gather(&key_pool, &krows);
             let vals = gather(&val_pool, &vrows);
             Rc::new(columns_to_batch(keys, vals, times, diffs, desc))
-        }).collect();
-        phase_finish("finish.gather_and_seal", phase, output_rows);
-        result
+        }).collect()
     }
 }

From 482f6075baf1f509c0972876c7572e5e4c3f1a09 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 20:40:13 -0400
Subject: [PATCH 5/8] Assert proxy pending frontier invariant

---
 .../src/operators/int_proxy/reduce.rs            | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index 16a526cf1..f453616f1 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -147,6 +147,12 @@ impl<T, Bk> ProxyReduceTactic<T, Bk> {
     }
 }
 
+fn frontier_of_pending<T: PartialOrder + Clone>(pending: &BTreeMap<u64, Vec<T>>) -> Antichain<T> {
+    let mut frontier = Antichain::new();
+    for time in pending.values().flatten() { frontier.insert_ref(time); }
+    frontier
+}
+
 impl<B1, B2, Bk> ReduceTactic<B1, B2> for ProxyReduceTactic<B1::Time, Bk>
 where
     B1: BatchReader,
@@ -201,6 +207,11 @@ where
         // beyond `upper` can remain when nothing is due, and releasing their capabilities would
         // strand them (see the frontier clause of the `ReduceTactic::retire` contract).
         if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) {
+            debug_assert!({
+                let expected = frontier_of_pending(&self.pending);
+                expected.elements().iter().all(|t| pending_frontier.less_equal(t))
+                    && pending_frontier.elements().iter().all(|t| expected.less_equal(t))
+            }, "maintained pending frontier differs from pending times");
             return (Vec::new(), pending_frontier);
         }
 
@@ -383,6 +394,11 @@ where
         }
 
         let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
+        debug_assert!({
+            let expected = frontier_of_pending(&self.pending);
+            expected.elements().iter().all(|t| pending_frontier.less_equal(t))
+                && pending_frontier.elements().iter().all(|t| expected.less_equal(t))
+        }, "maintained pending frontier differs from pending times");
         (produced, pending_frontier)
     }
 }

From fba14a2a50161f2991650ebb98e7d44cec71894b Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 20:59:01 -0400
Subject: [PATCH 6/8] Move due proxy timestamps without cloning

---
 differential-dataflow/src/operators/int_proxy/reduce.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index f453616f1..5ab8f22e6 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -188,10 +188,10 @@ where
         let mut pending_frontier = Antichain::new();
         for (&key, times) in self.pending.iter_mut() {
             let mut ready = Vec::new();
-            times.retain(|time| {
+            times.retain_mut(|time| {
                 let carried = upper.less_equal(time);
                 if carried { pending_frontier.insert_ref(time); }
-                else { ready.push(time.clone()); }
+                else { ready.push(std::mem::replace(time, B1::Time::minimum())); }
                 carried
             });
             if !ready.is_empty() { due.insert(key, ready); }

From fac98b4509fb8b4fb121f5c83c072eac7275623e Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 21:23:47 -0400
Subject: [PATCH 7/8] Tighten Corgi presentation merge contract

---
 interactive/src/corgi/reduce.rs | 51 +++++++++++++++++++++++----------
 1 file changed, 36 insertions(+), 15 deletions(-)

diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 089a2ff5e..016400590 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -282,41 +282,58 @@ where
     (keys_col, vals_col, khs, times, diffs, run_ends)
 }
 
-/// Merge already-ordered selected chunk runs directly into a consolidated proxy bridge. Returns
-/// false when a run's structural value order does not agree with its proxy-id order.
+/// Merge already-ordered selected chunk runs directly into an empty proxy bridge. Returns false
+/// when proxy ids do not preserve the chunks' structural order, or when the bridge already holds
+/// records; the caller can then append the records and consolidate the whole bridge normally.
 fn merge_present<T: Ord + Clone>(
+    ordered_ids: bool,
     khs: &[u64], vids: &[u64], times: &[T], diffs: &[Diff], run_ends: &[usize],
     bridge: &mut ProxyBridge<T, Diff>,
 ) -> bool {
-    let mut start = 0usize;
-    for &end in run_ends {
-        if (start + 1..end).any(|i| (khs[i - 1], vids[i - 1], &times[i - 1]) > (khs[i], vids[i], &times[i])) {
-            return false;
+    if !ordered_ids || !bridge.is_empty() {
+        return false;
+    }
+
+    if run_ends.len() == 1 {
+        let mut current: Option<((u64, u64), T, Diff)> = None;
+        for index in 0..run_ends[0] {
+            let kv = (khs[index], vids[index]);
+            let time = &times[index];
+            if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == time) {
+                current.as_mut().unwrap().2 += diffs[index];
+            } else {
+                if let Some(record) = current.take() {
+                    if record.2 != 0 { bridge.push(record); }
+                }
+                current = Some((kv, time.clone(), diffs[index]));
+            }
+        }
+        if let Some(record) = current {
+            if record.2 != 0 { bridge.push(record); }
         }
-        start = end;
+        return true;
     }
 
-    bridge.clear();
-    let mut heap: BinaryHeap<Reverse<((u64, u64), T, usize, usize)>> = BinaryHeap::new();
+    let mut heap: BinaryHeap<Reverse<((u64, u64), &T, usize, usize)>> = BinaryHeap::new();
     let mut lo = 0usize;
     for (run, &hi) in run_ends.iter().enumerate() {
-        heap.push(Reverse(((khs[lo], vids[lo]), times[lo].clone(), run, lo)));
+        heap.push(Reverse(((khs[lo], vids[lo]), &times[lo], run, lo)));
         lo = hi;
     }
     let mut current: Option<((u64, u64), T, Diff)> = None;
     while let Some(Reverse((kv, time, run, index))) = heap.pop() {
-        if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == &time) {
+        if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == time) {
             current.as_mut().unwrap().2 += diffs[index];
         } else {
             if let Some(record) = current.take() {
                 if record.2 != 0 { bridge.push(record); }
             }
-            current = Some((kv, time, diffs[index]));
+            current = Some((kv, time.clone(), diffs[index]));
         }
         let end = run_ends[run];
         if index + 1 < end {
             let next = index + 1;
-            heap.push(Reverse(((khs[next], vids[next]), times[next].clone(), run, next)));
+            heap.push(Reverse(((khs[next], vids[next]), &times[next], run, next)));
         }
     }
     if let Some(record) = current {
@@ -355,12 +372,14 @@ where
         if khs.is_empty() {
             return;
         }
+        let ordered_ids = corgi::arrange::leaf_slice(&p_keys).is_some()
+            && corgi::arrange::leaf_slice(&p_vals).is_some();
         let vids = ids(&p_vals);
         for (row, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(*len + row); }
         *len += p_vals.len();
         blocks.push(p_vals);
         self.register_keys(p_keys, &khs);
-        if !merge_present(&khs, &vids, &times, &diffs, &run_ends, bridge) {
+        if !merge_present(ordered_ids, &khs, &vids, &times, &diffs, &run_ends, bridge) {
             bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i])));
             consolidate_updates(bridge);
         }
@@ -595,10 +614,12 @@ where
         // Output-history presentation, same keys (register keys + values for correction resolution).
         let (o_keys, o_vals, o_khs, o_times, o_diffs, o_run_ends) = collect_present(&chunks_of(instance.output_batches), &keys);
         if !o_khs.is_empty() {
+            let ordered_ids = corgi::arrange::leaf_slice(&o_keys).is_some()
+                && corgi::arrange::leaf_slice(&o_vals).is_some();
             let vids = ids(&o_vals);
             self.register_keys(o_keys, &o_khs);
             self.register_vals(o_vals, &vids);
-            if !merge_present(&o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output) {
+            if !merge_present(ordered_ids, &o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output) {
                 window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])));
                 consolidate_updates(&mut window.output);
             }

From c9824fb892a9518ad746e715b7d2dd2f29c58756 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 15 Aug 2026 21:52:48 -0400
Subject: [PATCH 8/8] Make proxy reduce invariants explicit

---
 .../src/operators/int_proxy/reduce.rs         | 27 ++++---
 interactive/src/corgi/reduce.rs               | 76 +++++++++++--------
 2 files changed, 57 insertions(+), 46 deletions(-)

diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs
index 5ab8f22e6..75bc95450 100644
--- a/differential-dataflow/src/operators/int_proxy/reduce.rs
+++ b/differential-dataflow/src/operators/int_proxy/reduce.rs
@@ -147,10 +147,13 @@ impl<T, Bk> ProxyReduceTactic<T, Bk> {
     }
 }
 
-fn frontier_of_pending<T: PartialOrder + Clone>(pending: &BTreeMap<u64, Vec<T>>) -> Antichain<T> {
-    let mut frontier = Antichain::new();
-    for time in pending.values().flatten() { frontier.insert_ref(time); }
-    frontier
+fn assert_pending_frontier<T: PartialOrder + Clone>(pending: &BTreeMap<u64, Vec<T>>, maintained: &Antichain<T>) {
+    debug_assert!({
+        let mut expected = Antichain::new();
+        for time in pending.values().flatten() { expected.insert_ref(time); }
+        expected.elements().iter().all(|t| maintained.less_equal(t))
+            && maintained.elements().iter().all(|t| expected.less_equal(t))
+    }, "maintained pending frontier differs from pending times");
 }
 
 impl<B1, B2, Bk> ReduceTactic<B1, B2> for ProxyReduceTactic<B1::Time, Bk>
@@ -169,6 +172,10 @@ where
         held: &Antichain<B1::Time>,
     ) -> (Vec<(B1::Time, B2)>, Antichain<B1::Time>) {
         if held.elements().iter().all(|t| upper.less_equal(t)) {
+            debug_assert!(
+                self.pending.values().flatten().all(|time| held.less_equal(time)),
+                "held capabilities do not cover pending times",
+            );
             return (Vec::new(), held.clone());
         }
 
@@ -207,11 +214,7 @@ where
         // beyond `upper` can remain when nothing is due, and releasing their capabilities would
         // strand them (see the frontier clause of the `ReduceTactic::retire` contract).
         if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) {
-            debug_assert!({
-                let expected = frontier_of_pending(&self.pending);
-                expected.elements().iter().all(|t| pending_frontier.less_equal(t))
-                    && pending_frontier.elements().iter().all(|t| expected.less_equal(t))
-            }, "maintained pending frontier differs from pending times");
+            assert_pending_frontier(&self.pending, &pending_frontier);
             return (Vec::new(), pending_frontier);
         }
 
@@ -394,11 +397,7 @@ where
         }
 
         let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect();
-        debug_assert!({
-            let expected = frontier_of_pending(&self.pending);
-            expected.elements().iter().all(|t| pending_frontier.less_equal(t))
-                && pending_frontier.elements().iter().all(|t| expected.less_equal(t))
-        }, "maintained pending frontier differs from pending times");
+        assert_pending_frontier(&self.pending, &pending_frontier);
         (produced, pending_frontier)
     }
 }
diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs
index 016400590..c53d882ab 100644
--- a/interactive/src/corgi/reduce.rs
+++ b/interactive/src/corgi/reduce.rs
@@ -165,8 +165,10 @@ fn concat_columns(blocks: &[CValue]) -> CValue {
 /// width-blind and consistent-with-equality) — not the branch-local `arrange::hash_rows`; DDIR
 /// transcodes every leaf to `u64`, so width-blindness is a no-op for us and there is no cross-path
 /// hash comparison (value-as-id and native hash are never used for the same value: shape is uniform
-/// per column). The id is used ONLY as an identity for netting/dedup — its numeric order is never
-/// relied upon — so the raw two's-complement `u64` is correct even for negative ints (no swizzle).
+/// per column). Compound ids are used only for identity, but the leaf fast path additionally relies
+/// on raw-id order matching corgi's unsigned leaf order: `seek_needles` searches in that order and
+/// `merge_present` merges chunk runs in it. Raw two's-complement `u64` therefore remains correct for
+/// negative ints (no swizzle); changing the leaf encoding must also revisit those ordered paths.
 /// Applied CONSISTENTLY at every id site (both value presentations AND the freshly-produced
 /// `reduce_brackets` outputs), else `desired − current` nets across mismatched ids for the same value.
 fn ids(col: &CValue) -> Vec<u64> {
@@ -282,32 +284,51 @@ where
     (keys_col, vals_col, khs, times, diffs, run_ends)
 }
 
-/// Merge already-ordered selected chunk runs directly into an empty proxy bridge. Returns false
-/// when proxy ids do not preserve the chunks' structural order, or when the bridge already holds
-/// records; the caller can then append the records and consolidate the whole bridge normally.
+/// Merge already-ordered selected chunk runs directly into an empty proxy bridge. Identity-id leaf
+/// columns preserve the chunks' structural order; a debug assertion audits the resulting
+/// `(key_id, value_id, time)` order against that inference. Returns false for structural columns or
+/// a nonempty bridge, so the caller can append and consolidate the whole bridge normally.
 fn merge_present<T: Ord + Clone>(
-    ordered_ids: bool,
+    keys_col: &CValue, vals_col: &CValue,
     khs: &[u64], vids: &[u64], times: &[T], diffs: &[Diff], run_ends: &[usize],
     bridge: &mut ProxyBridge<T, Diff>,
 ) -> bool {
+    let ordered_ids = corgi::arrange::leaf_slice(keys_col).is_some()
+        && corgi::arrange::leaf_slice(vals_col).is_some();
     if !ordered_ids || !bridge.is_empty() {
         return false;
     }
 
+    debug_assert!({
+        let mut start = 0usize;
+        let sorted = run_ends.iter().all(|&end| {
+            let sorted = (start + 1..end).all(|i| {
+                (khs[i - 1], vids[i - 1], &times[i - 1])
+                    <= (khs[i], vids[i], &times[i])
+            });
+            start = end;
+            sorted
+        });
+        sorted && start == khs.len()
+    }, "identity ids do not preserve selected chunk order");
+
+    let mut current: Option<((u64, u64), T, Diff)> = None;
+    let mut accumulate = |kv, time: &T, diff| {
+        if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == time) {
+            current.as_mut().unwrap().2 += diff;
+        } else {
+            if let Some(record) = current.take() {
+                if record.2 != 0 { bridge.push(record); }
+            }
+            current = Some((kv, time.clone(), diff));
+        }
+    };
+
     if run_ends.len() == 1 {
-        let mut current: Option<((u64, u64), T, Diff)> = None;
         for index in 0..run_ends[0] {
-            let kv = (khs[index], vids[index]);
-            let time = &times[index];
-            if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == time) {
-                current.as_mut().unwrap().2 += diffs[index];
-            } else {
-                if let Some(record) = current.take() {
-                    if record.2 != 0 { bridge.push(record); }
-                }
-                current = Some((kv, time.clone(), diffs[index]));
-            }
+            accumulate((khs[index], vids[index]), &times[index], diffs[index]);
         }
+        drop(accumulate);
         if let Some(record) = current {
             if record.2 != 0 { bridge.push(record); }
         }
@@ -320,22 +341,15 @@ fn merge_present<T: Ord + Clone>(
         heap.push(Reverse(((khs[lo], vids[lo]), &times[lo], run, lo)));
         lo = hi;
     }
-    let mut current: Option<((u64, u64), T, Diff)> = None;
     while let Some(Reverse((kv, time, run, index))) = heap.pop() {
-        if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == time) {
-            current.as_mut().unwrap().2 += diffs[index];
-        } else {
-            if let Some(record) = current.take() {
-                if record.2 != 0 { bridge.push(record); }
-            }
-            current = Some((kv, time.clone(), diffs[index]));
-        }
+        accumulate(kv, time, diffs[index]);
         let end = run_ends[run];
         if index + 1 < end {
             let next = index + 1;
             heap.push(Reverse(((khs[next], vids[next]), &times[next], run, next)));
         }
     }
+    drop(accumulate);
     if let Some(record) = current {
         if record.2 != 0 { bridge.push(record); }
     }
@@ -372,14 +386,13 @@ where
         if khs.is_empty() {
             return;
         }
-        let ordered_ids = corgi::arrange::leaf_slice(&p_keys).is_some()
-            && corgi::arrange::leaf_slice(&p_vals).is_some();
         let vids = ids(&p_vals);
+        let merged = merge_present(&p_keys, &p_vals, &khs, &vids, &times, &diffs, &run_ends, bridge);
         for (row, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(*len + row); }
         *len += p_vals.len();
         blocks.push(p_vals);
         self.register_keys(p_keys, &khs);
-        if !merge_present(ordered_ids, &khs, &vids, &times, &diffs, &run_ends, bridge) {
+        if !merged {
             bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i])));
             consolidate_updates(bridge);
         }
@@ -614,12 +627,11 @@ where
         // Output-history presentation, same keys (register keys + values for correction resolution).
         let (o_keys, o_vals, o_khs, o_times, o_diffs, o_run_ends) = collect_present(&chunks_of(instance.output_batches), &keys);
         if !o_khs.is_empty() {
-            let ordered_ids = corgi::arrange::leaf_slice(&o_keys).is_some()
-                && corgi::arrange::leaf_slice(&o_vals).is_some();
             let vids = ids(&o_vals);
+            let merged = merge_present(&o_keys, &o_vals, &o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output);
             self.register_keys(o_keys, &o_khs);
             self.register_vals(o_vals, &vids);
-            if !merge_present(ordered_ids, &o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output) {
+            if !merged {
                 window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i])));
                 consolidate_updates(&mut window.output);
             }