From 67c0d5cb9fe8eea1b357d7aa0f500dec0ec5c93e Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 10 Aug 2026 13:44:11 -0400 Subject: [PATCH 01/11] A reference VecReduceBackend, with tests and a tactic-vs-tactic benchmark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A ProxyReduceBackend over hash-keyed VecChunk storage, the plain-rows counterpart to the corgi backend. It covers the key space in bounded windows, so the harness's multi-window path — the code #824 noted had never executed — now runs under its contract assertions in every test. Input value ids are per-run ordinals from a shared pool: the history and novel runs mint independently, and a value present in both gets two ids, reconciled because reduce_corrections resolves ids to values and consolidates by value before applying logic. Output ids are interned, sharing the namespace corrections mint into. Collisions re-group each hash bracket by real key, so a 64-bit collision is an inefficiency rather than an error (tested by forcing every key to one hash). tests/int_proxy.rs gives the proxy tactic its first in-repo tests: a direct retire, collision correctness, cursor-reduce comparisons (flat, String values, multi-moment, one-key windows), and reduce inside iterate. tests/int_proxy_bench.rs benchmarks the cursor tactic against the proxy tactic over the SAME hash-keyed arrangement (with the inherent reduce as reference) across three regimes: streaming churn, multi-time batches, and label propagation inside iterate with churn — the carried-interesting-times shape of #824. Current readings, single worker, release: churn 1.03x cursor-same; multimoment 1.84x; propagate 3.39x (output checksums asserted equal across all three modes). Also adds VecChunk::as_slice, a read-only accessor for the sorted records, which the backend's merge walks. Co-Authored-By: Claude Fable 5 --- .../src/operators/int_proxy/mod.rs | 1 + .../src/operators/int_proxy/vec_backend.rs | 386 ++++++++++++++++++ differential-dataflow/src/trace/chunk/vec.rs | 5 + differential-dataflow/tests/int_proxy.rs | 249 +++++++++++ .../tests/int_proxy_bench.rs | 338 +++++++++++++++ 5 files changed, 979 insertions(+) create mode 100644 differential-dataflow/src/operators/int_proxy/vec_backend.rs create mode 100644 differential-dataflow/tests/int_proxy.rs create mode 100644 differential-dataflow/tests/int_proxy_bench.rs diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs index 9c72ffeab..e247fc44a 100644 --- a/differential-dataflow/src/operators/int_proxy/mod.rs +++ b/differential-dataflow/src/operators/int_proxy/mod.rs @@ -46,6 +46,7 @@ mod history; pub mod join; pub mod reduce; +pub mod vec_backend; /// Integer-only exchange medium: a consolidated collection of `[((hash, id), time, diff)]`. /// diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs new file mode 100644 index 000000000..f9d35ab60 --- /dev/null +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -0,0 +1,386 @@ +//! A reference [`ProxyReduceBackend`] over [`VecChunk`] storage. +//! +//! This backend exists to demonstrate the backend recipe in plain `Vec`-of-rows form, and to give +//! the proxy tactic a counterpart that can be tested and benchmarked against the cursor tactic on +//! identical storage: both can drive a reduction over the same hash-keyed +//! [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine) arrangement (see `tests/int_proxy.rs` and +//! `tests/int_proxy_bench.rs`). Unlike the corgi backend it covers the key space in bounded +//! windows, exercising the harness's multi-window path. +//! +//! # The recipe +//! +//! The wiring arranges `coll.map(|(k, v)| (k.hashed(), (k, v)))` into a `ChunkSpine` whose key is +//! the `u64` key hash and whose value is the full `(key, val)` pair. Retaining the real key as +//! data is what makes hash collisions survivable: the operator isolates work by hash, and this +//! backend re-groups each hash bracket by the real key before applying `logic` (see +//! [`the module docs`](super) on the two integers). +//! +//! Per window, the backend presents three runs — accumulated history, the novel delta, and the +//! output history — as `((hash, value_id), time, diff)` bridges. Input value ids are **ordinals**: +//! minted in presentation order (which is `(hash, value, time)` order, so bridges emerge sorted), +//! resolved through a per-window pool, and never persisted. The history and novel runs mint ids +//! independently; a value present in both gets two ids, which is harmless because +//! [`reduce_corrections`](ProxyReduceBackend::reduce_corrections) resolves ids to values and +//! consolidates *by value* before applying `logic`. Output ids are interned (value -> id) instead, +//! because corrections mint values that must share the namespace of the presented output history. +//! +//! Time handling remains zero lines: the tactic owns all lattice logic, and this backend only ever +//! clones times through. + +use std::collections::BTreeMap; +use std::rc::Rc; + +use timely::container::PushInto; +use timely::progress::Timestamp; + +use crate::consolidation::{consolidate, consolidate_updates}; +use crate::difference::Semigroup; +use crate::lattice::Lattice; +use crate::trace::chunk::ChunkBatch; +use crate::trace::chunk::vec::VecChunk; +use crate::trace::Description; + +use super::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; + +/// The batch type of a hash-keyed [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine): payload +/// `D` is `(K, V)` on the input side and `(K, W)` on the output side. +type VBatch = Rc>>; + +/// Walks `batches` restricted to the ascending `keys`, emitting each record as +/// `(hash, &payload, &time, &diff)` in `(hash, payload, time)` order. +/// +/// Per hash bracket, the batches' contiguous runs are gathered and sorted by `(payload, time)`, so +/// equal payloads meet across batches and each payload's times arrive in order (batch intervals +/// are disjoint, so cross-batch times need ordering but never summing). The walk seeks to the +/// first requested key and stops after the last, so a bounded window pays for its own range. +fn merged_run( + batches: &[VBatch], + keys: &[u64], + mut sink: impl FnMut(u64, &D, &T, &R), +) where + D: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Semigroup + Ord + Clone + 'static, +{ + if keys.is_empty() { + return; + } + let n = batches.len(); + let (mut ci, mut oi) = (vec![0usize; n], vec![0usize; n]); + let mut cur: Vec<&[((u64, D), T, R)]> = vec![&[]; n]; + // Seek each batch to the first record with hash at or above the window's first key. + let k0 = keys[0]; + for b in 0..n { + let chunks = &batches[b].chunks; + ci[b] = chunks.partition_point(|c| c.as_slice().last().is_some_and(|r| r.0.0 < k0)); + cur[b] = chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); + oi[b] = cur[b].partition_point(|r| r.0.0 < k0); + } + let mut scratch: Vec<(&D, &T, &R)> = Vec::new(); + let mut ki = 0usize; + loop { + // Least hash among the batch heads. + let mut minh: Option = None; + for b in 0..n { + if let Some(r) = cur[b].get(oi[b]) { + if minh.is_none_or(|m| r.0.0 < m) { + minh = Some(r.0.0); + } + } + } + let Some(h) = minh else { break }; + while ki < keys.len() && keys[ki] < h { + ki += 1; + } + if ki >= keys.len() { + break; + } + let want = keys[ki] == h; + scratch.clear(); + for b in 0..n { + while let Some(r) = cur[b].get(oi[b]) { + if r.0.0 != h { + break; + } + if want { + scratch.push((&r.0.1, &r.1, &r.2)); + } + oi[b] += 1; + if oi[b] >= cur[b].len() { + ci[b] += 1; + oi[b] = 0; + cur[b] = batches[b].chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); + } + } + } + if want { + scratch.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1))); + for (d, t, r) in scratch.drain(..) { + sink(h, d, t, r); + } + } + } +} + +/// A reference [`ProxyReduceBackend`] over hash-keyed [`VecChunk`] storage. +/// +/// `logic` is differential's general four-argument reduce closure: it receives the real key, the +/// accumulated input values, the tentative accumulated output, and appends the output updates it +/// deems necessary. +pub struct VecReduceBackend { + logic: L, + /// Keys per window: bounds the live presentation, and exercises the multi-window path. + window_size: usize, + /// The retire's relevant keys — those the novel batches touch, merged with `changed` — built + /// on the retire's first window and consumed by hash range from there. + keys_cache: Vec, + /// Resolves an input value id (a per-window ordinal) to its `(key, val)` row. + in_pool: Vec<(K, V)>, + /// Resolves an output value id to its `(key, out)` row; spans the whole retire, because + /// corrections mint ids that later windows' presentations and `emit` must agree on. + out_pool: Vec<(K, W)>, + /// Interns `(key, out)` rows to their output id. + out_ids: BTreeMap<(K, W), u64>, + /// The retire's output tile descriptions, and the rows accumulated for each. + tiles: Vec>, + tile_rows: Vec>, +} + +impl VecReduceBackend { + /// A backend deferring value semantics to `logic`, covering the key space in default-sized + /// windows. + pub fn new(logic: L) -> Self { + Self::with_window(logic, 1 << 12) + } + + /// A backend with an explicit window size, in keys. Small sizes exercise the harness's + /// multi-window path; `usize::MAX` presents a single window, like the corgi backend. + pub fn with_window(logic: L, window_size: usize) -> Self { + VecReduceBackend { + logic, + window_size: window_size.max(1), + keys_cache: Vec::new(), + in_pool: Vec::new(), + out_pool: Vec::new(), + out_ids: BTreeMap::new(), + tiles: Vec::new(), + tile_rows: Vec::new(), + } + } +} + +impl ProxyReduceBackend, VBatch<(K, W), T, R>> + for VecReduceBackend +where + K: Ord + Clone + 'static, + V: Ord + Clone + 'static, + W: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Semigroup + Ord + Clone + 'static, + L: FnMut(&K, &[(V, R)], &mut Vec<(W, R)>, &mut Vec<(W, R)>), +{ + type RIn = R; + type ROut = R; + + fn begin(&mut self, tiles: &[Description]) { + self.out_pool.clear(); + self.out_ids.clear(); + self.tiles = tiles.to_vec(); + self.tile_rows = (0..tiles.len()).map(|_| Vec::new()).collect(); + } + + fn next_window( + &mut self, + instance: &ReduceInstance<'_, VBatch<(K, V), T, R>, VBatch<(K, W), T, R>>, + changed: &[u64], + from: &mut Option, + window: &mut ReduceWindow, + ) { + let Some(start) = *from else { return }; + + // First window of the retire: gather the relevant keys — those the novel batches touch, + // merged with the `changed` keys the harness supplies. Discovered in the scan the + // presentation needs anyway; later windows slice this by hash range. + if start == 0 { + self.keys_cache.clear(); + for batch in instance.input_batches.iter() { + for chunk in batch.chunks.iter() { + self.keys_cache.extend(chunk.as_slice().iter().map(|r| r.0.0)); + } + } + self.keys_cache.sort_unstable(); + self.keys_cache.dedup(); + if !changed.is_empty() { + let mut merged = Vec::with_capacity(self.keys_cache.len() + changed.len()); + let (mut a, mut b) = (0usize, 0usize); + while a < self.keys_cache.len() || b < changed.len() { + let key = match (self.keys_cache.get(a), changed.get(b)) { + (Some(x), Some(y)) => *x.min(y), + (Some(x), None) => *x, + (None, Some(y)) => *y, + (None, None) => unreachable!("loop condition ensures one is present"), + }; + if self.keys_cache.get(a) == Some(&key) { a += 1; } + if changed.get(b) == Some(&key) { b += 1; } + merged.push(key); + } + self.keys_cache = merged; + } + } + + let lo = self.keys_cache.partition_point(|k| *k < start); + if lo == self.keys_cache.len() { + *from = None; + return; + } + let hi = (lo + self.window_size).min(self.keys_cache.len()); + *from = if hi == self.keys_cache.len() { None } else { Some(self.keys_cache[hi]) }; + let keys = &self.keys_cache[lo..hi]; + + // The two input runs, presented apart on ordinal ids from a shared pool. Ids ascend with + // the walk, so each bridge emerges sorted by `((hash, id), time)` with nothing to sort or + // consolidate; a value in both runs gets two ids, reconciled by value in the corrections. + self.in_pool.clear(); + let pool = &mut self.in_pool; + let mut last: Option = None; + merged_run(instance.source_batches, keys, |h, d, t, r| { + if last != Some(h) || pool.last() != Some(d) { + pool.push(d.clone()); + last = Some(h); + } + window.history.push(((h, (pool.len() - 1) as u64), t.clone(), r.clone())); + }); + let mut last: Option = None; + merged_run(instance.input_batches, keys, |h, d, t, r| { + if last != Some(h) || pool.last() != Some(d) { + pool.push(d.clone()); + last = Some(h); + } + window.novel.push(((h, (pool.len() - 1) as u64), t.clone(), r.clone())); + }); + + // The output history, interned into the id namespace corrections mint into. + let (out_pool, out_ids) = (&mut self.out_pool, &mut self.out_ids); + merged_run(instance.output_batches, keys, |h, d, t, r| { + let id = *out_ids.entry(d.clone()).or_insert_with(|| { + out_pool.push(d.clone()); + (out_pool.len() - 1) as u64 + }); + window.output.push(((h, id), t.clone(), r.clone())); + }); + consolidate_updates(&mut window.output); + } + + fn reduce_corrections( + &mut self, + keys: &[u64], + in_ends: &[usize], + input: &[(u64, R)], + out_ends: &[usize], + output: &[(u64, R)], + ) -> (Vec<(u64, R)>, Vec) { + let mut corr: Vec<(u64, R)> = Vec::new(); + let mut corr_ends: Vec = Vec::with_capacity(keys.len()); + let (mut is, mut os) = (0usize, 0usize); + let mut updates: Vec<(W, R)> = Vec::new(); + let mut input_vals: Vec<(V, R)> = Vec::new(); + let mut current: Vec<(W, R)> = Vec::new(); + for i in 0..keys.len() { + let (ie, oe) = (in_ends[i], out_ends[i]); + // No-collision fast path: if the bracket's endpoints agree on the real key, the whole + // bracket is one key (ids resolve in key-sorted order) and the per-key grouping below + // can be skipped. The general path re-groups by real key, which is what makes a 64-bit + // hash collision an inefficiency rather than an error. + let ik0 = (is < ie).then(|| &self.in_pool[input[is].0 as usize].0); + let ikn = (is < ie).then(|| &self.in_pool[input[ie - 1].0 as usize].0); + let ok0 = (os < oe).then(|| &self.out_pool[output[os].0 as usize].0); + let okn = (os < oe).then(|| &self.out_pool[output[oe - 1].0 as usize].0); + let single = ik0 == ikn && ok0 == okn && (ik0.is_none() || ok0.is_none() || ik0 == ok0); + if single && (ik0.is_some() || ok0.is_some()) { + let key: K = ik0.or(ok0).expect("bracket has an endpoint").clone(); + input_vals.clear(); + input_vals.extend(input[is..ie].iter().map(|(vid, d)| (self.in_pool[*vid as usize].1.clone(), d.clone()))); + consolidate(&mut input_vals); + current.clear(); + current.extend(output[os..oe].iter().map(|(vid, d)| (self.out_pool[*vid as usize].1.clone(), d.clone()))); + consolidate(&mut current); + updates.clear(); + (self.logic)(&key, &input_vals, &mut current, &mut updates); + consolidate(&mut updates); + for (w, d) in updates.drain(..) { + let key_w = (key.clone(), w); + let id = *self.out_ids.entry(key_w.clone()).or_insert_with(|| { + self.out_pool.push(key_w); + (self.out_pool.len() - 1) as u64 + }); + corr.push((id, d)); + } + } else { + let mut ins: BTreeMap> = BTreeMap::new(); + for (vid, d) in &input[is..ie] { + let (k, v) = &self.in_pool[*vid as usize]; + ins.entry(k.clone()).or_default().push((v.clone(), d.clone())); + } + let mut outs: BTreeMap> = BTreeMap::new(); + for (vid, d) in &output[os..oe] { + let (k, w) = &self.out_pool[*vid as usize]; + outs.entry(k.clone()).or_default().push((w.clone(), d.clone())); + } + let mut real_keys: Vec = ins.keys().chain(outs.keys()).cloned().collect(); + real_keys.sort(); + real_keys.dedup(); + for key in real_keys { + let mut ivals = ins.remove(&key).unwrap_or_default(); + consolidate(&mut ivals); + let mut cur = outs.remove(&key).unwrap_or_default(); + consolidate(&mut cur); + updates.clear(); + (self.logic)(&key, &ivals, &mut cur, &mut updates); + consolidate(&mut updates); + for (w, d) in updates.drain(..) { + let key_w = (key.clone(), w); + let id = *self.out_ids.entry(key_w.clone()).or_insert_with(|| { + self.out_pool.push(key_w); + (self.out_pool.len() - 1) as u64 + }); + corr.push((id, d)); + } + } + } + corr_ends.push(corr.len()); + is = ie; + os = oe; + } + (corr, corr_ends) + } + + fn emit(&mut self, tile: usize, records: &[((u64, u64), T, R)]) { + for ((h, vid), t, d) in records { + let row = self.out_pool[*vid as usize].clone(); + self.tile_rows[tile].push(((*h, row), t.clone(), d.clone())); + } + } + + fn finish(&mut self) -> Vec> { + 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, mut rows)| { + consolidate_updates(&mut rows); + let chunks: Vec> = rows + .chunks( as crate::trace::chunk::Chunk>::TARGET) + .map(|piece| { + let mut chunk = VecChunk::default(); + for update in piece { + chunk.push_into(update.clone()); + } + chunk + }) + .collect(); + Rc::new(ChunkBatch::new(chunks, desc)) + }) + .collect() + } +} diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs index 1a6b9aff8..18dc05d78 100644 --- a/differential-dataflow/src/trace/chunk/vec.rs +++ b/differential-dataflow/src/trace/chunk/vec.rs @@ -37,6 +37,11 @@ const TARGET: usize = 8192; /// A sorted, consolidated run of `((key, val), time, diff)`, shared via `Rc`. pub struct VecChunk(Rc>); +impl VecChunk { + /// The chunk's records as a sorted, consolidated slice. + pub fn as_slice(&self) -> &[((K, V), T, R)] { &self.0 } +} + impl Clone for VecChunk { fn clone(&self) -> Self { VecChunk(Rc::clone(&self.0)) } } diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs new file mode 100644 index 000000000..40e0c1609 --- /dev/null +++ b/differential-dataflow/tests/int_proxy.rs @@ -0,0 +1,249 @@ +//! Tests for the proxy reduce tactic over the [`VecReduceBackend`] reference backend. +//! +//! The backend reads a HASH-KEYED arrangement (`VecChunk`, key = hash(K)), so +//! the caller maps `(k, v) -> (hash(k), (k, v))` before arranging. Every dataflow test compares +//! against the cursor-tactic reduce on the same input; run in debug so the tactic's +//! `debug_assert_sorted_bridge` and window-contract assertions are live. + +use std::rc::Rc; +use std::sync::{Arc, Mutex}; + +use timely::container::PushInto; +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::ToStream; +use timely::order::Product; +use timely::progress::{Antichain, Timestamp}; + +use differential_dataflow::consolidation::consolidate_updates; +use differential_dataflow::difference::Semigroup; +use differential_dataflow::hashable::Hashable; +use differential_dataflow::lattice::Lattice; +use differential_dataflow::operators::arrange::arrangement::arrange_core; +use differential_dataflow::operators::int_proxy::reduce::ProxyReduceTactic; +use differential_dataflow::operators::int_proxy::vec_backend::VecReduceBackend; +use differential_dataflow::operators::iterate::Iterate; +use differential_dataflow::operators::reduce::{reduce_with_tactic, ReduceTactic}; +use differential_dataflow::trace::chunk::vec::{ + ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine, VecChunk, +}; +use differential_dataflow::trace::chunk::ChunkBatch; +use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::implementations::ContainerChunker; +use differential_dataflow::trace::{Description, Navigable}; +use differential_dataflow::AsCollection; + +type Batch = Rc>>; + +/// Read a `u64`-keyed batch, dropping the hash key; returns `(value, time, diff)`. +fn hread(batches: &[Batch]) -> Vec<(KV, T, R)> +where + KV: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Ord + Semigroup + 'static, +{ + let mut out = Vec::new(); + for b in batches { + for chunk in &b.chunks { + let mut c = chunk.cursor(); + while c.key_valid(chunk) { + while c.val_valid(chunk) { + let kv = c.val(chunk).clone(); + c.map_times(chunk, |t, d| out.push((kv.clone(), t.clone(), d.clone()))); + c.step_val(chunk); + } + c.step_key(chunk); + } + } + } + consolidate_updates(&mut out); + out +} + +/// Build a HASH-KEYED input batch from `((K, V), T, R)` rows: `((hash(K), (K, V)), T, R)`. +fn hbatch(rows: Vec<((K, V), T, R)>, lower: T, upper: T) -> Batch +where + K: Hashable + Ord + Clone + 'static, + K::Output: Into, + V: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Ord + Semigroup + 'static, +{ + let mut hrows: Vec<((u64, (K, V)), T, R)> = + rows.into_iter().map(|((k, v), t, r)| ((k.hashed().into(), (k, v)), t, r)).collect(); + consolidate_updates(&mut hrows); + let mut chunk = VecChunk::default(); + for u in hrows { + chunk.push_into(u); + } + let desc = Description::new(Antichain::from_elem(lower), Antichain::from_elem(upper), Antichain::from_elem(T::minimum())); + Rc::new(ChunkBatch::new(vec![chunk], desc)) +} + +fn max_logic(_k: &u64, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>) { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { + updates.push((*w, -*d)); + } +} + +#[test] +fn reduce_one_retire() { + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(max_logic)); + let input = hbatch::(vec![((7, 3), 0, 1), ((7, 5), 0, 1), ((7, 4), 0, 1), ((9, 2), 0, 1)], 0, 1); + let (produced, frontier) = tactic.retire( + vec![], vec![], vec![input], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + assert!(frontier.is_empty()); + let out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + // Output values are `(key, max)`: (7,5) and (9,2). + assert_eq!(out, vec![((7u64, 5u64), 0u64, 1i64), ((9, 2), 0, 1)]); +} + +#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)] +struct Collide(u64); +impl Hashable for Collide { + type Output = u64; + fn hashed(&self) -> u64 { 0 } +} + +#[test] +fn reduce_collision_correct() { + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new( + |_k: &Collide, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { updates.push((*w, -*d)); } + }, + )); + let input = hbatch::(vec![((Collide(1), 9), 0, 1), ((Collide(1), 4), 0, 1), ((Collide(2), 9), 0, 1)], 0, 1); + let (produced, _f) = tactic.retire( + vec![], vec![], vec![input], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + let out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + assert_eq!(out, vec![((Collide(1), 9u64), 0u64, 1i64), ((Collide(2), 9), 0, 1)], "collision must not merge keys"); +} + +/// Run the proxy reduce (with the given backend window size) and the cursor reduce over the same +/// updates; both outputs must agree. +fn proxy_matches_mainline(updates: Vec<((u64, u64), u64, i64)>, window: usize) { + let t_out = Arc::new(Mutex::new(Vec::<((u64, u64), u64, i64)>::new())); + let r_out = Arc::new(Mutex::new(Vec::<((u64, u64), u64, i64)>::new())); + let (ts, rs) = (t_out.clone(), r_out.clone()); + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let coll = updates.clone().to_stream(scope).as_collection(); + let hashed = coll.clone().map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, ContainerChunker>, VChunkBatcher, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange"); + reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, window))) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + .inspect(move |(d, t, r)| ts.lock().unwrap().push((*d, *t, *r))); + coll.reduce(|_k, input, output| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { output.push((*m, 1)); } + }) + .inspect(move |(d, t, r)| rs.lock().unwrap().push((*d, *t, *r))); + }); + }); + let (mut got, mut want) = (t_out.lock().unwrap().clone(), r_out.lock().unwrap().clone()); + got.sort(); + want.sort(); + assert_eq!(got, want, "proxy-reduce must match the cursor reduce"); +} + +#[test] +fn reduce_dataflow_matches_mainline() { + proxy_matches_mainline( + vec![((7, 3), 0, 1), ((7, 5), 0, 1), ((7, 4), 0, 1), ((9, 2), 0, 1), ((7, 5), 1, -1)], + usize::MAX, + ); +} + +#[test] +fn reduce_multiwindow_matches_mainline() { + // One key per window: every window-loop edge (the `from` cursor, the window-key merge, and the + // in-window `changed` consumption) runs many times, under the harness's contract asserts. + let mut updates = Vec::new(); + for k in 0..64u64 { + updates.push(((k, k), 0u64, 1i64)); + updates.push(((k, 100 + k), 0, 1)); + updates.push(((k, 100 + k), 1, -1)); + } + proxy_matches_mainline(updates, 1); +} + +#[test] +fn reduce_multimoment_matches_mainline() { + // Four distinct times in ONE batch: multi-moment keys exercise the round loop, including + // produced-corrections feedback (the max changes at t1, reverts at t2, changes at t3). + proxy_matches_mainline( + vec![ + ((7, 3), 0, 1), ((7, 5), 1, 1), ((7, 5), 2, -1), ((7, 9), 3, 1), + ((9, 2), 0, 1), ((9, 1), 2, 1), ((9, 2), 3, -1), + ], + usize::MAX, + ); +} + +#[test] +fn reduce_string_values_matches_mainline() { + let t_out = Arc::new(Mutex::new(Vec::<((u64, String), u64, i64)>::new())); + let r_out = Arc::new(Mutex::new(Vec::<((u64, String), u64, i64)>::new())); + let (ts, rs) = (t_out.clone(), r_out.clone()); + let updates: Vec<((u64, String), u64, i64)> = vec![ + ((7, "b".into()), 0, 1), ((7, "d".into()), 0, 1), ((7, "c".into()), 0, 1), ((9, "a".into()), 0, 1), + ((7, "d".into()), 1, -1), ((9, "z".into()), 1, 1), + ]; + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let coll = updates.clone().to_stream(scope).as_collection(); + let hashed = coll.clone().map(|(k, v): (u64, String)| (k.hashed(), (k, v))); + let arr = arrange_core::, ContainerChunker>, VChunkBatcher, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange"); + reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduceStr", ProxyReduceTactic::new(VecReduceBackend::with_window( + |_k: &u64, input: &[(String, i64)], current: &mut Vec<(String, i64)>, updates: &mut Vec<(String, i64)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { updates.push((m, 1)); } + for (w, d) in current.iter() { updates.push((w.clone(), -*d)); } + }, + 1, + ))) + .as_collection(|_h, kw: &(u64, String)| (kw.0, kw.1.clone())) + .inspect(move |(d, t, r)| ts.lock().unwrap().push((d.clone(), *t, *r))); + coll.reduce(|_k, input, output| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| (*v).clone()).max() { output.push((m, 1)); } + }) + .inspect(move |(d, t, r)| rs.lock().unwrap().push((d.clone(), *t, *r))); + }); + }); + let (mut got, mut want) = (t_out.lock().unwrap().clone(), r_out.lock().unwrap().clone()); + got.sort(); + want.sort(); + assert_eq!(got, want, "proxy-reduce must match the cursor reduce for String values"); +} + +#[test] +fn reduce_inside_iterate() { + let out = Arc::new(Mutex::new(Vec::<((u64, u64), u64, i64)>::new())); + let os = out.clone(); + let updates: Vec<((u64, u64), u64, i64)> = + vec![((7, 3), 0, 1), ((7, 5), 0, 1), ((7, 4), 0, 1), ((9, 2), 0, 1), ((9, 8), 0, 1)]; + + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let input = updates.clone().to_stream(scope).as_collection(); + let result = input.iterate(|_scope, inner| { + let hashed = inner.map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, VChunkBatcher, i64>, VChunkBuilder, i64>, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter"); + reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "IterReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + }); + result.inspect(move |(d, t, r)| os.lock().unwrap().push((*d, *t, *r))); + }); + }); + + let mut got = out.lock().unwrap().clone(); + got.sort(); + assert_eq!(got, vec![((7u64, 5u64), 0u64, 1i64), ((9, 8), 0, 1)], "max-per-key fixpoint"); +} diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs new file mode 100644 index 000000000..714cb4519 --- /dev/null +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -0,0 +1,338 @@ +//! Benchmarks the cursor tactic against the proxy tactic over IDENTICAL storage. +//! +//! Both tactics drive a reduction over the same hash-keyed `ChunkSpine` arrangement — the cursor +//! tactic through [`Arranged::reduce_core`], the proxy tactic through `reduce_with_tactic` with the +//! [`VecReduceBackend`] — so their difference is the tactic (and its backend boundary), not the +//! storage. The inherent `reduce` (its own `OrdVal`-style arrangement) is carried as a familiar +//! reference point. +//! +//! Three workloads cover the regimes that have historically mattered: +//! * `churn` — bounded per-key history, one distinct time per flush: the streaming steady state. +//! * `multimoment` — several distinct times per flush: the batched/throughput regime. +//! * `propagate` — label propagation inside `iterate`: `Product` times and carried interesting +//! times across retires, the shape on which visiting non-due keys was quadratic (PR #824). +//! +//! Run with: +//! ```text +//! cargo test --release --test int_proxy_bench -- --ignored --nocapture +//! ``` + +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::probe::Handle as ProbeHandle; +use timely::dataflow::operators::Probe; +use timely::order::Product; + +use differential_dataflow::hashable::Hashable; +use differential_dataflow::input::InputSession; +use differential_dataflow::operators::arrange::arrangement::arrange_core; +use differential_dataflow::operators::int_proxy::reduce::ProxyReduceTactic; +use differential_dataflow::operators::int_proxy::vec_backend::VecReduceBackend; +use differential_dataflow::operators::iterate::Iterate; +use differential_dataflow::operators::reduce::reduce_with_tactic; +use differential_dataflow::trace::chunk::vec::{ + ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine, + VecChunk, VecChunkCursor, +}; +use differential_dataflow::trace::cursor::Cursor; +use differential_dataflow::trace::implementations::ContainerChunker; +use differential_dataflow::AsCollection; + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +enum Mode { + /// The inherent `reduce`: cursor tactic over its own conventional arrangement. + Mainline, + /// The cursor tactic over the hash-keyed chunk arrangement. + CursorSame, + /// The proxy tactic + `VecReduceBackend` over the hash-keyed chunk arrangement. + Proxy, +} + +/// Arrange `(k, v) -> (hash(k), (k, v))` into a hash-keyed `ChunkSpine`. +macro_rules! harrange { + ($coll:expr, $t:ty, $name:expr) => {{ + let hashed = $coll.map(|(k, v)| (k.hashed(), (k, v))); + arrange_core::< + Pipeline, + Vec<((u64, (u64, u64)), $t, isize)>, + ContainerChunker>, + VChunkBatcher, + VChunkBuilder, + VChunkSpine, + >(hashed.inner, Pipeline, $name) + }}; +} + +/// The cursor tactic over a hash-keyed chunk arrangement, with `$fold` computing the desired +/// output value from the accumulated `(key, val)` pairs. +macro_rules! cursor_reduce { + ($arr:expr, $t:ty, $name:expr, $fold:expr) => { + $arr.reduce_core::< + _, + VChunkBuilder, + VChunkSpine, + as Cursor>::KeyContainer, + _, + >( + $name, + |_h, input, current, updates| { + if let Some(kw) = $fold(input.iter().filter(|(_, d)| *d > 0).map(|(kv, _)| **kv)) { + updates.push((kw, 1)); + } + for (w, d) in current.iter() { + updates.push((w.clone(), -*d)); + } + }, + |chunk, key, list| { + use timely::container::PushInto; + *chunk = Default::default(); + for (v, t, d) in list.drain(..) { + chunk.push_into(((*key, v), t, d)); + } + }, + ) + }; +} + +/// The proxy tactic + backend over a hash-keyed chunk arrangement, same `$fold`. +macro_rules! proxy_reduce { + ($arr:expr, $t:ty, $name:expr, $fold:expr) => { + reduce_with_tactic::<_, VChunkSpine, _>( + $arr, + $name, + ProxyReduceTactic::new(VecReduceBackend::new( + |k: &u64, input: &[(u64, isize)], current: &mut Vec<(u64, isize)>, updates: &mut Vec<(u64, isize)>| { + if let Some((_, w)) = $fold(input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| (*k, *v))) { + updates.push((w, 1)); + } + for (w, d) in current.iter() { + updates.push((*w, -*d)); + } + }, + )), + ) + }; +} + +fn max_fold(iter: impl Iterator) -> Option<(u64, u64)> { + iter.max_by_key(|kv| kv.1) +} + +fn min_fold(iter: impl Iterator) -> Option<(u64, u64)> { + iter.min_by_key(|kv| kv.1) +} + +const CHURN_KEYS: u64 = 50_000; +const MM_KEYS: u64 = 10_000; +const MM_SUB: u64 = 4; +const ROUNDS: u64 = 16; +const WARMUP: u64 = 6; + +/// Bounded churn, one distinct time per flush; returns averaged microseconds per round. +fn run_churn(mode: Mode) -> f64 { + run_flat(mode, CHURN_KEYS, 1) +} + +/// Bounded churn, `MM_SUB` distinct times per flush. +fn run_multimoment(mode: Mode) -> f64 { + run_flat(mode, MM_KEYS, MM_SUB) +} + +fn run_flat(mode: Mode, keys: u64, sub: u64) -> f64 { + timely::execute_directly(move |worker| { + let mut input: InputSession = InputSession::new(); + let probe = worker.dataflow::(|scope| { + let coll = input.to_collection(scope); + let mut ph = ProbeHandle::new(); + match mode { + Mode::Mainline => { + coll.reduce(|_k, i: &[(&u64, isize)], o: &mut Vec<(u64, isize)>| { + if let Some(m) = i.iter().filter(|(_, d)| *d > 0).map(|(v, _)| **v).max() { + o.push((m, 1)); + } + }) + .inner + .probe_with(&mut ph); + } + Mode::CursorSame => { + let arr = harrange!(coll, u64, "Arrange"); + cursor_reduce!(arr, u64, "CursorReduce", max_fold).stream.probe_with(&mut ph); + } + Mode::Proxy => { + let arr = harrange!(coll, u64, "Arrange"); + proxy_reduce!(arr, u64, "ProxyReduce", max_fold).stream.probe_with(&mut ph); + } + } + ph + }); + input.advance_to(0); + for k in 0..keys { + input.insert((k, 0)); + input.insert((k, 1)); + } + input.advance_to(1); + input.flush(); + while probe.less_than(&1) { + worker.step(); + } + let mut times = Vec::new(); + for r in 0..ROUNDS { + let base = 1 + r * sub; + for s in 0..sub { + let t = base + s; + input.advance_to(t); + for k in 0..keys { + if t > 1 { + input.remove((k, 1000 + t - 1)); + } + input.insert((k, 1000 + t)); + } + } + input.advance_to(base + sub); + input.flush(); + let start = Instant::now(); + while probe.less_than(&(base + sub)) { + worker.step(); + } + if r >= WARMUP { + times.push(start.elapsed().as_micros()); + } + } + times.iter().sum::() as f64 / times.len() as f64 + }) +} + +const PROP_NODES: u64 = 2_000; +const PROP_EDGES: usize = 4_000; +const PROP_CHURN: usize = 50; +const PROP_ROUNDS: u64 = 40; +const PROP_WARMUP: u64 = 5; + +fn edge(i: usize) -> (u64, u64) { + let n = PROP_NODES; + ((i as u64).wrapping_mul(2654435761) % n, (i as u64).wrapping_mul(40503).wrapping_add(7) % n) +} + +/// Label propagation (min over self and in-neighbors) inside `iterate`, with edge churn: the +/// carried-interesting-times shape. Returns averaged microseconds per round and a checksum of the +/// produced label updates, which must agree across modes. +fn run_propagate(mode: Mode) -> (f64, i64) { + let check = Arc::new(Mutex::new(0i64)); + let sum = check.clone(); + let avg = timely::execute_directly(move |worker| { + let mut nodes: InputSession = InputSession::new(); + let mut edges: InputSession = InputSession::new(); + let probe = worker.dataflow::(|scope| { + let nodes = nodes.to_collection(scope); + let edges = edges.to_collection(scope); + let mut ph = ProbeHandle::new(); + let labels = nodes.clone().iterate(move |_scope, inner| { + let scope = inner.scope(); + let nodes_in = nodes.enter(scope); + let edges_in = edges.enter(scope); + let prop = edges_in.join_map(inner, |_src, dst, lbl| (*dst, *lbl)).concat(nodes_in); + match mode { + Mode::Mainline => prop.reduce(|_k, i: &[(&u64, isize)], o: &mut Vec<(u64, isize)>| { + if let Some(m) = i.iter().filter(|(_, d)| *d > 0).map(|(v, _)| **v).min() { + o.push((m, 1)); + } + }), + Mode::CursorSame => { + let arr = harrange!(prop, Product, "ArrProp"); + cursor_reduce!(arr, Product, "CursorProp", min_fold) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + } + Mode::Proxy => { + let arr = harrange!(prop, Product, "ArrProp"); + proxy_reduce!(arr, Product, "ProxyProp", min_fold) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + } + } + }); + labels + .inspect(move |((_, lbl), _, d)| { + *sum.lock().unwrap() += (*lbl as i64) * (*d as i64); + }) + .inner + .probe_with(&mut ph); + ph + }); + nodes.advance_to(0); + edges.advance_to(0); + for k in 0..PROP_NODES { + nodes.insert((k, k)); + } + for i in 0..PROP_EDGES { + let (s, d) = edge(i); + edges.insert((s, d)); + } + nodes.advance_to(1); + nodes.flush(); + edges.advance_to(1); + edges.flush(); + while probe.less_than(&1) { + worker.step(); + } + let mut times = Vec::new(); + for r in 0..PROP_ROUNDS { + let t = r + 1; + // Perturb a low-numbered label each round: low labels spread widely under `min`, so + // withdrawing one forces its reach to relabel — real iterate work, with interesting + // times carried across retires. + let m = r % 50; + if r > 0 { + let p = (r - 1) % 50; + nodes.insert((p, p)); + } + nodes.remove((m, m)); + for c in 0..PROP_CHURN { + let dead = (r as usize) * PROP_CHURN + c; + let (s, d) = edge(dead); + edges.remove((s, d)); + let (s, d) = edge(PROP_EDGES + dead); + edges.insert((s, d)); + } + nodes.advance_to(t + 1); + nodes.flush(); + edges.advance_to(t + 1); + edges.flush(); + let start = Instant::now(); + while probe.less_than(&(t + 1)) { + worker.step(); + } + if r >= PROP_WARMUP { + times.push(start.elapsed().as_micros()); + } + } + times.iter().sum::() as f64 / times.len() as f64 + }); + let total = *check.lock().unwrap(); + (avg, total) +} + +fn report(name: &str, mainline: f64, cursor: f64, proxy: f64) { + eprintln!(" {name}:"); + eprintln!(" inherent reduce (own storage) : {:9.0}us/round", mainline); + eprintln!(" cursor tactic (chunk storage) : {:9.0}us/round", cursor); + eprintln!(" proxy tactic (chunk storage) : {:9.0}us/round ({:.2}x cursor-same, {:.2}x inherent)", proxy, proxy / cursor, proxy / mainline); +} + +#[test] +#[ignore] +fn bench_reduce_tactics() { + eprintln!(); + for (name, run) in [ + ("churn (50k keys, 1 time/flush)", run_churn as fn(Mode) -> f64), + ("multimoment (10k keys, 4 times/flush)", run_multimoment), + ] { + report(name, run(Mode::Mainline), run(Mode::CursorSame), run(Mode::Proxy)); + } + let (mainline, sum_m) = run_propagate(Mode::Mainline); + let (cursor, sum_c) = run_propagate(Mode::CursorSame); + let (proxy, sum_p) = run_propagate(Mode::Proxy); + assert_eq!(sum_m, sum_c, "cursor-tactic propagate output must match the inherent reduce"); + assert_eq!(sum_m, sum_p, "proxy-tactic propagate output must match the inherent reduce"); + report("propagate (2k nodes, 4k edges, iterate)", mainline, cursor, proxy); +} From 8dc2cdca1b6b26d689ec209c940eb74bea84d57f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 10 Aug 2026 14:03:58 -0400 Subject: [PATCH 02/11] Verify the single-key bracket rather than testing its endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fast path in `reduce_corrections` skipped the per-key regrouping when a bracket's first and last input ids resolved to the same real key, and likewise for its output ids, on the stated grounds that "ids resolve in key-sorted order". Neither id space is key-ordered across a bracket. Input ids are ordinals minted from one pool, the history run first and then the novel run, so the order is `history by key, then novel by key` rather than globally by key. Output ids are interned, with the corrections a crossing mints appended after the presentation's. Either can read `[A, B, A]` by id, whose endpoints agree while its interior does not, and the bracket is then reduced as if it were all `A` — so a hash collision became an error rather than the inefficiency the general path exists to make it. Resolving every id in the bracket and requiring them all to agree is a linear pass over data the fast path clones immediately afterwards, against the general path's per-key maps. The three benchmark workloads are unchanged by it (churn 1.03x, multimoment 1.79x, propagate 3.93x against the cursor tactic, all within run-to-run noise of the previous numbers). `reduce_collision_fastpath_endpoints` is the regression test: a reduction that emits only for `Collide(1)` leaves `Collide(2)` with input but never any output, so retire 2 presents the input bracket as `[C1, C2, C1]` and the output bracket as `[C1]`, and both endpoint tests passed. Before this commit it produced `(Collide(1), 900)` — the other key's value — where `(Collide(1), 6)` was correct. `reduce_collision_across_retires` covers the neighbouring shape that does reach the general path. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/operators/int_proxy/vec_backend.rs | 43 ++++++++--- differential-dataflow/tests/int_proxy.rs | 76 +++++++++++++++++++ 2 files changed, 108 insertions(+), 11 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index f9d35ab60..2dff62498 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -287,17 +287,38 @@ where let mut current: Vec<(W, R)> = Vec::new(); for i in 0..keys.len() { let (ie, oe) = (in_ends[i], out_ends[i]); - // No-collision fast path: if the bracket's endpoints agree on the real key, the whole - // bracket is one key (ids resolve in key-sorted order) and the per-key grouping below - // can be skipped. The general path re-groups by real key, which is what makes a 64-bit - // hash collision an inefficiency rather than an error. - let ik0 = (is < ie).then(|| &self.in_pool[input[is].0 as usize].0); - let ikn = (is < ie).then(|| &self.in_pool[input[ie - 1].0 as usize].0); - let ok0 = (os < oe).then(|| &self.out_pool[output[os].0 as usize].0); - let okn = (os < oe).then(|| &self.out_pool[output[oe - 1].0 as usize].0); - let single = ik0 == ikn && ok0 == okn && (ik0.is_none() || ok0.is_none() || ik0 == ok0); - if single && (ik0.is_some() || ok0.is_some()) { - let key: K = ik0.or(ok0).expect("bracket has an endpoint").clone(); + // No-collision fast path: when the whole bracket is one real key, the per-key grouping + // below can be skipped. Testing only the bracket's ENDPOINTS would be wrong: neither id + // space is key-ordered across a bracket. Input ids are ordinals minted history-run + // first and then novel, so the order is `history by key, then novel by key`; output ids + // are interned, with corrections appended after the presentation. Either can read + // `[A, B, A]`, whose endpoints agree while its interior does not. So resolve every id + // and require them all to agree — a linear pass over data the fast path is about to + // clone anyway, against the general path's per-key maps. + let single_key: Option = { + let mut only: Option<&K> = None; + let mut agree = true; + for (vid, _) in &input[is..ie] { + let k = &self.in_pool[*vid as usize].0; + match only { + None => only = Some(k), + Some(prev) if prev == k => {} + Some(_) => { agree = false; break } + } + } + if agree { + for (vid, _) in &output[os..oe] { + let k = &self.out_pool[*vid as usize].0; + match only { + None => only = Some(k), + Some(prev) if prev == k => {} + Some(_) => { agree = false; break } + } + } + } + if agree { only.cloned() } else { None } + }; + if let Some(key) = single_key { input_vals.clear(); input_vals.extend(input[is..ie].iter().map(|(vid, d)| (self.in_pool[*vid as usize].1.clone(), d.clone()))); consolidate(&mut input_vals); diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index 40e0c1609..da72e2859 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -247,3 +247,79 @@ fn reduce_inside_iterate() { got.sort(); assert_eq!(got, vec![((7u64, 5u64), 0u64, 1i64), ((9, 8), 0, 1)], "max-per-key fixpoint"); } + +/// Two colliding keys, but the collision spans the HISTORY/NOVEL boundary: retire 1 inserts both +/// keys, retire 2 touches only the lower one. Input ids are ordinals minted history-first, so the +/// bracket reads `[C1, C2, C1]` by id and its endpoints agree even though its interior does not. +#[test] +fn reduce_collision_across_retires() { + let logic = |_k: &Collide, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { updates.push((*w, -*d)); } + }; + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(logic)); + + // Retire 1: both colliding keys arrive. + let b0 = hbatch::(vec![((Collide(1), 5), 0, 1), ((Collide(2), 7), 0, 1)], 0, 1); + let (p0, _f) = tactic.retire( + vec![], vec![], vec![b0.clone()], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + let out0: Vec<_> = p0.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + assert_eq!(out0, vec![((Collide(1), 5u64), 0u64, 1i64), ((Collide(2), 7), 0, 1)], "retire 1"); + + // Retire 2: a novel update to the LOWER key only, so the id order is [C1(hist), C2(hist), C1(novel)]. + let out_batches: Vec<_> = { + let mut t2 = ProxyReduceTactic::new(VecReduceBackend::new(logic)); + let b = hbatch::(vec![((Collide(1), 5), 0, 1), ((Collide(2), 7), 0, 1)], 0, 1); + let (p, _) = t2.retire(vec![], vec![], vec![b], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64)); + p.into_iter().map(|(_t, b)| b).collect() + }; + let b1 = hbatch::(vec![((Collide(1), 9), 1, 1)], 1, 2); + let (p1, _f) = tactic.retire( + vec![b0], out_batches, vec![b1], + &Antichain::from_elem(1u64), &Antichain::from_elem(2u64), &Antichain::from_elem(1u64), + ); + let out1: Vec<_> = p1.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + // C1's max rises 5 -> 9; C2 is untouched and must NOT be disturbed. + assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 9), 1, 1)], "retire 2 must not disturb C2"); +} + +/// A reduction that emits only for `Collide(1)`, so `Collide(2)` has input but never any output. +fn only_first(k: &Collide, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>) { + if k.0 == 1 { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + } + for (w, d) in current.iter() { updates.push((*w, -*d)); } +} + +/// Input ids are ordinals minted history-first, so the id order across a hash bracket is +/// `[history by key, then novel by key]` — NOT globally key-sorted. Here that makes the bracket +/// read `[C1, C2, C1]` by id while its endpoints agree, and the output bracket holds only `C1` +/// (C2 never emits), so the endpoint test concludes the bracket is a single key. +#[test] +fn reduce_collision_fastpath_endpoints() { + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(only_first)); + let b0 = hbatch::(vec![((Collide(1), 5), 0, 1), ((Collide(2), 900), 0, 1)], 0, 1); + let (p0, _) = tactic.retire( + vec![], vec![], vec![b0.clone()], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + let outs: Vec<_> = p0.into_iter().map(|(_t, b)| b).collect(); + assert_eq!(outs.iter().flat_map(|b| hread(std::slice::from_ref(b))).collect::>(), + vec![((Collide(1), 5u64), 0u64, 1i64)], "retire 1: only C1 emits"); + + let b1 = hbatch::(vec![((Collide(1), 6), 1, 1)], 1, 2); + let (p1, _) = tactic.retire( + vec![b0], outs, vec![b1], + &Antichain::from_elem(1u64), &Antichain::from_elem(2u64), &Antichain::from_elem(1u64), + ); + let out1: Vec<_> = p1.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + // C1's values are {5, 6}: the max becomes 6. C2's 900 belongs to a different real key. + assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 6), 1, 1)], + "C2's value must not enter C1's reduction"); +} From 7f7326e4c3b97d46221ca4d8859ab7f4e3aa0878 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 10 Aug 2026 15:48:45 -0400 Subject: [PATCH 03/11] Cover keys whose input cancels away entirely Every key's input nets to zero by time 2, so from then on it holds no records in any of the three presentations while still being a key the retire must consider: its stale output has to be retracted. With one key per window this is also the case where a window's key list and the `changed` set disagree, which the harness's window-key derivation depends on. Co-Authored-By: Claude Opus 5 (1M context) --- differential-dataflow/tests/int_proxy.rs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index da72e2859..e90c8a3b4 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -323,3 +323,19 @@ fn reduce_collision_fastpath_endpoints() { assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 6), 1, 1)], "C2's value must not enter C1's reduction"); } + +#[test] +fn reduce_cancelling_keys_matches_mainline() { + // Every key's input cancels completely by time 2, so from then on it has no records in any of + // the three presentations while still being a key the retire must consider — its stale output + // has to be retracted. With one key per window this is also the case where a window's key list + // and the `changed` set disagree. + let mut updates = Vec::new(); + for k in 0..16u64 { + updates.push(((k, 10 + k), 0u64, 1i64)); + updates.push(((k, 20 + k), 1, 1)); + updates.push(((k, 10 + k), 2, -1)); + updates.push(((k, 20 + k), 2, -1)); + } + proxy_matches_mainline(updates, 1); +} From 822e02be629c7cd94c1c0585bc39b2ccba226865 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 07:41:49 -0400 Subject: [PATCH 04/11] Seek past unwanted keys in the reference backend's presentation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `merged_run` seeked each batch once, to the window's first key, and then walked forward record by record — stepping over every record of every hash it did not want. So a call cost the accumulated history, not what it was asked for. On `propagate` that is 64,436,421 records scanned to present 441,218, a factor of 146, across 11,739 calls asking 63,318 keys in total: about five scattered keys per call, each call re-reading the whole trace. `churn` and `multimoment` ask for nearly every key at once, so scanning everything IS presenting everything and they waste nothing — which is why the shortfall looked like a property of the tactic and tracked moments-per-key. Skipping now seeks: when the least hash present is not wanted, every batch binary-searches to the next key that is. That lands at or above it, so at most one unwanted hash is visited per wanted key. No threshold to tune, because a dense key set never takes the branch — unlike corgi's `collect_present`, which chooses between seeking and scanning on a measured ratio. propagate 10361us -> 3124us/round 3.76x -> 1.12x of the cursor tactic churn 0.95x -> 0.94x multimoment 1.66x -> 1.59x This is the yardstick, not the thing being measured: `merged_run` is the reference backend's, and every proxy-vs-cursor number taken before this was reading its cost as the tactic's. It belongs on `vec-backend-recipe`, where `vec_backend.rs` lives; it is here because that branch is cherry-picked in and this is where it was measured. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/operators/int_proxy/vec_backend.rs | 37 ++++++++++++++++--- 1 file changed, 32 insertions(+), 5 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 2dff62498..484ccc1a8 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -95,16 +95,43 @@ fn merged_run( if ki >= keys.len() { break; } - let want = keys[ki] == h; + if keys[ki] != h { + // `h` is not wanted. Seeking every batch to the next key that IS wanted costs a binary + // search per batch and lands at or above it, so at most one unwanted hash is visited + // per wanted key. Walking `h`'s records instead would make the whole pass linear in the + // accumulated history rather than in what is asked for — which on an iterative + // computation, where a retire asks for a handful of scattered keys, is the difference + // between presenting the window and re-reading the trace. When the key set is dense + // this branch is simply never taken, so there is no threshold to tune. + let next = keys[ki]; + for b in 0..n { + loop { + let Some(chunk) = batches[b].chunks.get(ci[b]) else { cur[b] = &[]; break }; + let slice = chunk.as_slice(); + if slice.last().is_some_and(|r| r.0.0 < next) { + ci[b] += 1; + oi[b] = 0; + continue; + } + cur[b] = slice; + oi[b] += slice[oi[b]..].partition_point(|r| r.0.0 < next); + if oi[b] >= slice.len() { + ci[b] += 1; + oi[b] = 0; + continue; + } + break; + } + } + continue; + } scratch.clear(); for b in 0..n { while let Some(r) = cur[b].get(oi[b]) { if r.0.0 != h { break; } - if want { - scratch.push((&r.0.1, &r.1, &r.2)); - } + scratch.push((&r.0.1, &r.1, &r.2)); oi[b] += 1; if oi[b] >= cur[b].len() { ci[b] += 1; @@ -113,7 +140,7 @@ fn merged_run( } } } - if want { + { scratch.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1))); for (d, t, r) in scratch.drain(..) { sink(h, d, t, r); From fe1fe47df42b64ea9e46b1794aa34a55989616d0 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 08:17:29 -0400 Subject: [PATCH 05/11] Intern output ids through a hash map, not a B-tree MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `reduce_corrections` minted an id for every emitted correction by looking `(key, value)` up in a `BTreeMap`. Carving the call into phases put 435 of its 513 samples there — more, on its own, than the whole of the conventional tactic's `compute`. The clones are not the problem; for these workloads the key is two `u64`s and copying it is free. The B-tree descent is. churn 0.94x -> 0.58x of the cursor tactic (47.5ms -> 29.1ms/round) multimoment 1.59x -> 1.09x (32.1ms -> 21.0ms/round) propagate 1.12x -> 1.13x (unchanged) Propagate is untouched because its corrections are few; churn and multimoment emit one per key per moment, which is where a per-correction map lookup lands. The backend now asks `Hash` of its key and output types, which is what a backend that interns needs and no burden on one that does not — the corgi backend derives ids by content-hashing a column and keeps no map at all. `Collide` states its collision as a `Hash` that writes a constant, rather than as a `Hashable` impl, since `Hashable` has a blanket impl for `T: Hash` and the two cannot coexist. Third finding in a row in the reference backend rather than the tactic. The benchmark was built to compare tactics over identical storage and has been comparing a young backend against a mature cursor implementation; ratios taken before these three fixes read its cost as the tactic's. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/operators/int_proxy/vec_backend.rs | 10 +++++----- differential-dataflow/tests/int_proxy.rs | 8 +++++--- 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 484ccc1a8..14ef1206c 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -27,7 +27,7 @@ //! Time handling remains zero lines: the tactic owns all lattice logic, and this backend only ever //! clones times through. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::rc::Rc; use timely::container::PushInto; @@ -167,7 +167,7 @@ pub struct VecReduceBackend { /// corrections mint ids that later windows' presentations and `emit` must agree on. out_pool: Vec<(K, W)>, /// Interns `(key, out)` rows to their output id. - out_ids: BTreeMap<(K, W), u64>, + out_ids: HashMap<(K, W), u64>, /// The retire's output tile descriptions, and the rows accumulated for each. tiles: Vec>, tile_rows: Vec>, @@ -189,7 +189,7 @@ impl VecReduceBackend { keys_cache: Vec::new(), in_pool: Vec::new(), out_pool: Vec::new(), - out_ids: BTreeMap::new(), + out_ids: HashMap::new(), tiles: Vec::new(), tile_rows: Vec::new(), } @@ -199,9 +199,9 @@ impl VecReduceBackend { impl ProxyReduceBackend, VBatch<(K, W), T, R>> for VecReduceBackend where - K: Ord + Clone + 'static, + K: Ord + Clone + std::hash::Hash + 'static, V: Ord + Clone + 'static, - W: Ord + Clone + 'static, + W: Ord + Clone + std::hash::Hash + 'static, T: Lattice + Timestamp, R: Semigroup + Ord + Clone + 'static, L: FnMut(&K, &[(V, R)], &mut Vec<(W, R)>, &mut Vec<(W, R)>), diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index e90c8a3b4..37836c677 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -104,9 +104,11 @@ fn reduce_one_retire() { #[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)] struct Collide(u64); -impl Hashable for Collide { - type Output = u64; - fn hashed(&self) -> u64 { 0 } +/// Every `Collide` hashes alike, so they share a `key_hash` and exercise the collision paths. +/// Written as `Hash` rather than `Hashable` because the latter has a blanket impl for `T: Hash`, +/// and the backend's id interning needs `Hash` too. +impl std::hash::Hash for Collide { + fn hash(&self, state: &mut H) { state.write_u64(0); } } #[test] From b6b26048b7a216afdcbf8a07b2841f5fbb645770 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 10:28:36 -0400 Subject: [PATCH 06/11] Review notes: pin an invariant, and rebuild the key set on `begin` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three small things a reviewer raised. The continuation after `partition_point` in the seek branch is unreachable: chunks are non-empty and the guard above skipped any whose last key is below the sought one, so the chunk holds a record at or above it. A `debug_assert` states that, and the branch stays, so a violated invariant degrades to a slower walk rather than to a batch that silently reads as drained. `out_ids` says it is lookup-only and that its iteration order is never observed, which is what makes a `HashMap` safe here. The key set was rebuilt when `from` opened at `Some(0)`, which is true today but is not something the trait promises. It now rebuilds on a flag that `begin` sets — `begin` runs once per retire, before any window — so the backend no longer depends on the value the harness opens with. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/operators/int_proxy/vec_backend.rs | 20 ++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 14ef1206c..c077da3fd 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -115,6 +115,12 @@ fn merged_run( } cur[b] = slice; oi[b] += slice[oi[b]..].partition_point(|r| r.0.0 < next); + // Chunks are non-empty (`ChunkBatch::new` asserts it) and the guard above + // skipped any whose last key is below `next`, so this chunk holds a record at + // or above `next` and the search cannot land at the end. The branch below is + // unreachable; it is kept so a violated invariant degrades to a slower walk + // rather than to a batch that silently reads as drained. + debug_assert!(oi[b] < slice.len(), "a chunk kept by the skip guard must hold a record at or above the sought key"); if oi[b] >= slice.len() { ci[b] += 1; oi[b] = 0; @@ -159,14 +165,19 @@ pub struct VecReduceBackend { /// Keys per window: bounds the live presentation, and exercises the multi-window path. window_size: usize, /// The retire's relevant keys — those the novel batches touch, merged with `changed` — built - /// on the retire's first window and consumed by hash range from there. + /// once per retire and consumed by hash range from there. keys_cache: Vec, + /// Whether `keys_cache` is still the previous retire's. Set by `begin`, which runs once per + /// retire before any window, so the rebuild does not depend on the value the harness opens + /// `from` with. + keys_stale: bool, /// Resolves an input value id (a per-window ordinal) to its `(key, val)` row. in_pool: Vec<(K, V)>, /// Resolves an output value id to its `(key, out)` row; spans the whole retire, because /// corrections mint ids that later windows' presentations and `emit` must agree on. out_pool: Vec<(K, W)>, - /// Interns `(key, out)` rows to their output id. + /// Interns `(key, out)` rows to their output id. Lookup-only: the map's iteration order is + /// never observed, so the hasher cannot affect what the operator produces. out_ids: HashMap<(K, W), u64>, /// The retire's output tile descriptions, and the rows accumulated for each. tiles: Vec>, @@ -187,6 +198,7 @@ impl VecReduceBackend { logic, window_size: window_size.max(1), keys_cache: Vec::new(), + keys_stale: true, in_pool: Vec::new(), out_pool: Vec::new(), out_ids: HashMap::new(), @@ -210,6 +222,7 @@ where type ROut = R; fn begin(&mut self, tiles: &[Description]) { + self.keys_stale = true; self.out_pool.clear(); self.out_ids.clear(); self.tiles = tiles.to_vec(); @@ -228,7 +241,8 @@ where // First window of the retire: gather the relevant keys — those the novel batches touch, // merged with the `changed` keys the harness supplies. Discovered in the scan the // presentation needs anyway; later windows slice this by hash range. - if start == 0 { + if self.keys_stale { + self.keys_stale = false; self.keys_cache.clear(); for batch in instance.input_batches.iter() { for chunk in batch.chunks.iter() { From 55229409ac5b77bb78c0afebeb4c5f7f9e5cf3a9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 10:39:58 -0400 Subject: [PATCH 07/11] Drop an unused import in the benchmark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI denies warnings; my local runs did not, so this only surfaced there. Checked the whole workspace under `-D warnings` rather than just the reported line — this was the only one. Co-Authored-By: Claude Opus 5 (1M context) --- differential-dataflow/tests/int_proxy_bench.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs index 714cb4519..4e24a9bf8 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -38,7 +38,6 @@ use differential_dataflow::trace::chunk::vec::{ }; use differential_dataflow::trace::cursor::Cursor; use differential_dataflow::trace::implementations::ContainerChunker; -use differential_dataflow::AsCollection; #[derive(Clone, Copy, PartialEq, Eq, Debug)] enum Mode { From e928c4dae52c1be5178cf97ccc63c90c49387991 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 10:46:26 -0400 Subject: [PATCH 08/11] Consolidate a tile per emit, not once over everything MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `finish` consolidated each tile's whole accumulation, sorting an array that was already in key-hash order. It only ever needed ordering within a hash: interned ids are in first-seen order rather than value order, so a hash's records can need reordering by their real `(key, out)` value, but no record ever needs to move past a different hash. `emit` is the place for it. A call carries the whole of every hash it mentions — the harness consolidates a tile's deltas per window, and a hash belongs to exactly one window — and calls arrive in ascending hash order, so consolidating the run just appended leaves the tile ordered. A `debug_assert` states the ascending-arrival property the backend now relies on. The difference is a sort per emit against one sort of everything a retire produced, which is invisible at benchmark scale and is not at the scale this is meant for. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/operators/int_proxy/vec_backend.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index c077da3fd..4eca101cf 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -33,7 +33,7 @@ use std::rc::Rc; use timely::container::PushInto; use timely::progress::Timestamp; -use crate::consolidation::{consolidate, consolidate_updates}; +use crate::consolidation::{consolidate, consolidate_updates, consolidate_updates_from}; use crate::difference::Semigroup; use crate::lattice::Lattice; use crate::trace::chunk::ChunkBatch; @@ -417,10 +417,21 @@ where } fn emit(&mut self, tile: usize, records: &[((u64, u64), T, R)]) { + // A call carries the whole of every key hash it mentions, and calls arrive in ascending + // hash order, so the tile stays hash-ordered and only the run just appended can need + // reordering — by the real `(key, out)` value, since interned ids are in first-seen order + // rather than value order. Consolidating here rather than over the whole tile at `finish` + // is the difference between a sort per emit and one sort of everything the retire produced. + let mark = self.tile_rows[tile].len(); + debug_assert!( + self.tile_rows[tile].last().is_none_or(|last| records.first().is_none_or(|r| last.0.0 <= r.0.0)), + "emit must arrive in ascending key-hash order", + ); for ((h, vid), t, d) in records { let row = self.out_pool[*vid as usize].clone(); self.tile_rows[tile].push(((*h, row), t.clone(), d.clone())); } + consolidate_updates_from(&mut self.tile_rows[tile], mark); } fn finish(&mut self) -> Vec> { @@ -429,8 +440,8 @@ where tiles .into_iter() .zip(tile_rows) - .map(|(desc, mut rows)| { - consolidate_updates(&mut rows); + .map(|(desc, rows)| { + // Already ordered and consolidated: `emit` did it a run at a time. let chunks: Vec> = rows .chunks( as crate::trace::chunk::Chunk>::TARGET) .map(|piece| { From 0a40770662ca46df48fe9ba5ad3bd0baa0e78fce Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 10:53:55 -0400 Subject: [PATCH 09/11] Let the benchmark be run at a scale where these costs show MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The workload sizes and round counts come from the environment, and `PROG`/`MODE` run a single cell of the matrix. Costs that are structural but invisible at a few megabytes — a sort over a whole retire's output rather than over each key's run, say — need a run that moves gigabytes to register, and the suite could not be asked for one without editing it. CHURN_KEYS=4000000 ROUNDS=6 WARMUP=2 PROG=churn MODE=proxy \ cargo test --release --test int_proxy_bench -- --ignored --nocapture Defaults unchanged, so the reported matrix is the same. Co-Authored-By: Claude Opus 5 (1M context) --- .../tests/int_proxy_bench.rs | 47 ++++++++++++++----- 1 file changed, 34 insertions(+), 13 deletions(-) diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs index 4e24a9bf8..b3e14b467 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -123,20 +123,25 @@ fn min_fold(iter: impl Iterator) -> Option<(u64, u64)> { iter.min_by_key(|kv| kv.1) } -const CHURN_KEYS: u64 = 50_000; -const MM_KEYS: u64 = 10_000; -const MM_SUB: u64 = 4; -const ROUNDS: u64 = 16; -const WARMUP: u64 = 6; +/// Workload sizes, overridable so the suite can be run at a scale where costs that are invisible +/// at a few megabytes show up. `churn_keys()=4000000 ROUNDS=6 WARMUP=2` moves gigabytes per round. +fn sized(var: &str, default: u64) -> u64 { + std::env::var(var).ok().and_then(|v| v.parse().ok()).unwrap_or(default) +} +fn churn_keys() -> u64 { sized("CHURN_KEYS", 50_000) } +fn mm_keys() -> u64 { sized("MM_KEYS", 10_000) } +fn mm_sub() -> u64 { sized("MM_SUB", 4) } +fn rounds() -> u64 { sized("ROUNDS", 16) } +fn warmup() -> u64 { sized("WARMUP", 6) } /// Bounded churn, one distinct time per flush; returns averaged microseconds per round. fn run_churn(mode: Mode) -> f64 { - run_flat(mode, CHURN_KEYS, 1) + run_flat(mode, churn_keys(), 1) } -/// Bounded churn, `MM_SUB` distinct times per flush. +/// Bounded churn, `mm_sub()` distinct times per flush. fn run_multimoment(mode: Mode) -> f64 { - run_flat(mode, MM_KEYS, MM_SUB) + run_flat(mode, mm_keys(), mm_sub()) } fn run_flat(mode: Mode, keys: u64, sub: u64) -> f64 { @@ -177,7 +182,7 @@ fn run_flat(mode: Mode, keys: u64, sub: u64) -> f64 { worker.step(); } let mut times = Vec::new(); - for r in 0..ROUNDS { + for r in 0..rounds() { let base = 1 + r * sub; for s in 0..sub { let t = base + s; @@ -195,7 +200,7 @@ fn run_flat(mode: Mode, keys: u64, sub: u64) -> f64 { while probe.less_than(&(base + sub)) { worker.step(); } - if r >= WARMUP { + if r >= warmup() { times.push(start.elapsed().as_micros()); } } @@ -322,11 +327,27 @@ fn report(name: &str, mainline: f64, cursor: f64, proxy: f64) { #[ignore] fn bench_reduce_tactics() { eprintln!(); + // `PROG=churn MODE=proxy` runs one cell of the matrix, which is what a large run wants. + if let (Some(prog), Some(mode)) = (std::env::var("PROG").ok(), match std::env::var("MODE").ok().as_deref() { + Some("proxy") => Some(Mode::Proxy), + Some("cursor") => Some(Mode::CursorSame), + Some("mainline") => Some(Mode::Mainline), + _ => None, + }) { + let us = match prog.as_str() { + "churn" => run_churn(mode), + "multimoment" => run_multimoment(mode), + "propagate" => run_propagate(mode).0, + other => panic!("unknown PROG {other:?}"), + }; + eprintln!(" {prog} {:?}: {us:.0}us/round", mode); + return; + } for (name, run) in [ - ("churn (50k keys, 1 time/flush)", run_churn as fn(Mode) -> f64), - ("multimoment (10k keys, 4 times/flush)", run_multimoment), + (format!("churn ({} keys, 1 time/flush)", churn_keys()), run_churn as fn(Mode) -> f64), + (format!("multimoment ({} keys, {} times/flush)", mm_keys(), mm_sub()), run_multimoment), ] { - report(name, run(Mode::Mainline), run(Mode::CursorSame), run(Mode::Proxy)); + report(&name, run(Mode::Mainline), run(Mode::CursorSame), run(Mode::Proxy)); } let (mainline, sum_m) = run_propagate(Mode::Mainline); let (cursor, sum_c) = run_propagate(Mode::CursorSame); From 711f72d82235106f4717d7c4ba9baed3a08bcdec Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 13:16:04 -0400 Subject: [PATCH 10/11] Restructure nonsense --- .../src/operators/int_proxy/vec_backend.rs | 346 +++++++++--------- .../tests/int_proxy_bench.rs | 100 +++++ 2 files changed, 264 insertions(+), 182 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 4eca101cf..08307c4ef 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -4,28 +4,23 @@ //! the proxy tactic a counterpart that can be tested and benchmarked against the cursor tactic on //! identical storage: both can drive a reduction over the same hash-keyed //! [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine) arrangement (see `tests/int_proxy.rs` and -//! `tests/int_proxy_bench.rs`). Unlike the corgi backend it covers the key space in bounded -//! windows, exercising the harness's multi-window path. +//! `tests/int_proxy_bench.rs`). //! //! # The recipe //! //! The wiring arranges `coll.map(|(k, v)| (k.hashed(), (k, v)))` into a `ChunkSpine` whose key is -//! the `u64` key hash and whose value is the full `(key, val)` pair. Retaining the real key as -//! data is what makes hash collisions survivable: the operator isolates work by hash, and this -//! backend re-groups each hash bracket by the real key before applying `logic` (see -//! [`the module docs`](super) on the two integers). +//! the `u64` key hash and whose value is the full `(key, val)` pair. //! -//! Per window, the backend presents three runs — accumulated history, the novel delta, and the -//! output history — as `((hash, value_id), time, diff)` bridges. Input value ids are **ordinals**: -//! minted in presentation order (which is `(hash, value, time)` order, so bridges emerge sorted), -//! resolved through a per-window pool, and never persisted. The history and novel runs mint ids -//! independently; a value present in both gets two ids, which is harmless because -//! [`reduce_corrections`](ProxyReduceBackend::reduce_corrections) resolves ids to values and -//! consolidates *by value* before applying `logic`. Output ids are interned (value -> id) instead, -//! because corrections mint values that must share the namespace of the presented output history. +//! Per window, the backend presents three proxy bridges — novel input, prior input, and prior output. +//! The identifiers chosen are based on ordinal position. Unfortunately, they are chosen independently +//! for the novel and prior inputs, meaning that empty input collections may be supplied to the logic +//! for evaluation, where the logic should be ignored and zero -> zero enforced. This could be fixed +//! with a more attentive identifier selection. //! -//! Time handling remains zero lines: the tactic owns all lattice logic, and this backend only ever -//! clones times through. +//! The backend crawls all keys at once, rather than respecting the window setting. +//! This is a defect that should be fixed, but the backend shouldn't be used at scale. +//! The backend clones keys and values like it is being paid to waste cycles. +//! Generally, this is not a high-performance backend, but shouldn't be abysmal. use std::collections::{BTreeMap, HashMap}; use std::rc::Rc; @@ -33,7 +28,7 @@ use std::rc::Rc; use timely::container::PushInto; use timely::progress::Timestamp; -use crate::consolidation::{consolidate, consolidate_updates, consolidate_updates_from}; +use crate::consolidation::{consolidate, consolidate_updates_from}; use crate::difference::Semigroup; use crate::lattice::Lattice; use crate::trace::chunk::ChunkBatch; @@ -46,131 +41,19 @@ use super::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; /// `D` is `(K, V)` on the input side and `(K, W)` on the output side. type VBatch = Rc>>; -/// Walks `batches` restricted to the ascending `keys`, emitting each record as -/// `(hash, &payload, &time, &diff)` in `(hash, payload, time)` order. -/// -/// Per hash bracket, the batches' contiguous runs are gathered and sorted by `(payload, time)`, so -/// equal payloads meet across batches and each payload's times arrive in order (batch intervals -/// are disjoint, so cross-batch times need ordering but never summing). The walk seeks to the -/// first requested key and stops after the last, so a bounded window pays for its own range. -fn merged_run( - batches: &[VBatch], - keys: &[u64], - mut sink: impl FnMut(u64, &D, &T, &R), -) where - D: Ord + Clone + 'static, - T: Lattice + Timestamp, - R: Semigroup + Ord + Clone + 'static, -{ - if keys.is_empty() { - return; - } - let n = batches.len(); - let (mut ci, mut oi) = (vec![0usize; n], vec![0usize; n]); - let mut cur: Vec<&[((u64, D), T, R)]> = vec![&[]; n]; - // Seek each batch to the first record with hash at or above the window's first key. - let k0 = keys[0]; - for b in 0..n { - let chunks = &batches[b].chunks; - ci[b] = chunks.partition_point(|c| c.as_slice().last().is_some_and(|r| r.0.0 < k0)); - cur[b] = chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); - oi[b] = cur[b].partition_point(|r| r.0.0 < k0); - } - let mut scratch: Vec<(&D, &T, &R)> = Vec::new(); - let mut ki = 0usize; - loop { - // Least hash among the batch heads. - let mut minh: Option = None; - for b in 0..n { - if let Some(r) = cur[b].get(oi[b]) { - if minh.is_none_or(|m| r.0.0 < m) { - minh = Some(r.0.0); - } - } - } - let Some(h) = minh else { break }; - while ki < keys.len() && keys[ki] < h { - ki += 1; - } - if ki >= keys.len() { - break; - } - if keys[ki] != h { - // `h` is not wanted. Seeking every batch to the next key that IS wanted costs a binary - // search per batch and lands at or above it, so at most one unwanted hash is visited - // per wanted key. Walking `h`'s records instead would make the whole pass linear in the - // accumulated history rather than in what is asked for — which on an iterative - // computation, where a retire asks for a handful of scattered keys, is the difference - // between presenting the window and re-reading the trace. When the key set is dense - // this branch is simply never taken, so there is no threshold to tune. - let next = keys[ki]; - for b in 0..n { - loop { - let Some(chunk) = batches[b].chunks.get(ci[b]) else { cur[b] = &[]; break }; - let slice = chunk.as_slice(); - if slice.last().is_some_and(|r| r.0.0 < next) { - ci[b] += 1; - oi[b] = 0; - continue; - } - cur[b] = slice; - oi[b] += slice[oi[b]..].partition_point(|r| r.0.0 < next); - // Chunks are non-empty (`ChunkBatch::new` asserts it) and the guard above - // skipped any whose last key is below `next`, so this chunk holds a record at - // or above `next` and the search cannot land at the end. The branch below is - // unreachable; it is kept so a violated invariant degrades to a slower walk - // rather than to a batch that silently reads as drained. - debug_assert!(oi[b] < slice.len(), "a chunk kept by the skip guard must hold a record at or above the sought key"); - if oi[b] >= slice.len() { - ci[b] += 1; - oi[b] = 0; - continue; - } - break; - } - } - continue; - } - scratch.clear(); - for b in 0..n { - while let Some(r) = cur[b].get(oi[b]) { - if r.0.0 != h { - break; - } - scratch.push((&r.0.1, &r.1, &r.2)); - oi[b] += 1; - if oi[b] >= cur[b].len() { - ci[b] += 1; - oi[b] = 0; - cur[b] = batches[b].chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); - } - } - } - { - scratch.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1))); - for (d, t, r) in scratch.drain(..) { - sink(h, d, t, r); - } - } - } -} - -/// A reference [`ProxyReduceBackend`] over hash-keyed [`VecChunk`] storage. -/// -/// `logic` is differential's general four-argument reduce closure: it receives the real key, the -/// accumulated input values, the tentative accumulated output, and appends the output updates it -/// deems necessary. +/// A reference [`ProxyReduceBackend`] over [`VBatch`] storage. pub struct VecReduceBackend { + /// User supplied reduce closure. logic: L, - /// Keys per window: bounds the live presentation, and exercises the multi-window path. + /// Configuration: keys per window, to size the steps the backend performs. window_size: usize, - /// The retire's relevant keys — those the novel batches touch, merged with `changed` — built - /// once per retire and consumed by hash range from there. + + /// All active keys, either novel input or supplied as externally changed. + /// This is *not* windowed, which is a defect to fix. keys_cache: Vec, - /// Whether `keys_cache` is still the previous retire's. Set by `begin`, which runs once per - /// retire before any window, so the rebuild does not depend on the value the harness opens - /// `from` with. + /// A state bit indicating that the keys cache should be rebuild (each begin). keys_stale: bool, + /// Resolves an input value id (a per-window ordinal) to its `(key, val)` row. in_pool: Vec<(K, V)>, /// Resolves an output value id to its `(key, out)` row; spans the whole retire, because @@ -179,20 +62,17 @@ pub struct VecReduceBackend { /// Interns `(key, out)` rows to their output id. Lookup-only: the map's iteration order is /// never observed, so the hasher cannot affect what the operator produces. out_ids: HashMap<(K, W), u64>, + /// The retire's output tile descriptions, and the rows accumulated for each. tiles: Vec>, tile_rows: Vec>, } impl VecReduceBackend { - /// A backend deferring value semantics to `logic`, covering the key space in default-sized - /// windows. - pub fn new(logic: L) -> Self { - Self::with_window(logic, 1 << 12) - } + /// A backend deferring value semantics to `logic`, covering the key space in windows. + pub fn new(logic: L) -> Self { Self::with_window(logic, 1 << 12) } - /// A backend with an explicit window size, in keys. Small sizes exercise the harness's - /// multi-window path; `usize::MAX` presents a single window, like the corgi backend. + /// A backend with an explicit window size, in keys. pub fn with_window(logic: L, window_size: usize) -> Self { VecReduceBackend { logic, @@ -229,6 +109,7 @@ where self.tile_rows = (0..tiles.len()).map(|_| Vec::new()).collect(); } + #[inline(never)] fn next_window( &mut self, instance: &ReduceInstance<'_, VBatch<(K, V), T, R>, VBatch<(K, W), T, R>>, @@ -238,9 +119,8 @@ where ) { let Some(start) = *from else { return }; - // First window of the retire: gather the relevant keys — those the novel batches touch, - // merged with the `changed` keys the harness supplies. Discovered in the scan the - // presentation needs anyway; later windows slice this by hash range. + // If the first window: form a list of all active keys, novel or changed. + // TODO: this is wasteful; the in-order chunk keys could be merged instead. if self.keys_stale { self.keys_stale = false; self.keys_cache.clear(); @@ -269,6 +149,7 @@ where } } + // Determine the range of active keys to process in this window. let lo = self.keys_cache.partition_point(|k| *k < start); if lo == self.keys_cache.len() { *from = None; @@ -284,34 +165,34 @@ where self.in_pool.clear(); let pool = &mut self.in_pool; let mut last: Option = None; - merged_run(instance.source_batches, keys, |h, d, t, r| { - if last != Some(h) || pool.last() != Some(d) { - pool.push(d.clone()); - last = Some(h); + merged_run(instance.source_batches, keys, |hash, data, time, diff| { + if last != Some(hash) || pool.last() != Some(data) { + pool.push(data.clone()); + last = Some(hash); } - window.history.push(((h, (pool.len() - 1) as u64), t.clone(), r.clone())); + window.history.push(((hash, (pool.len() - 1) as u64), time.clone(), diff.clone())); }); let mut last: Option = None; - merged_run(instance.input_batches, keys, |h, d, t, r| { - if last != Some(h) || pool.last() != Some(d) { - pool.push(d.clone()); - last = Some(h); + merged_run(instance.input_batches, keys, |hash, data, time, diff| { + if last != Some(hash) || pool.last() != Some(data) { + pool.push(data.clone()); + last = Some(hash); } - window.novel.push(((h, (pool.len() - 1) as u64), t.clone(), r.clone())); + window.novel.push(((hash, (pool.len() - 1) as u64), time.clone(), diff.clone())); }); // The output history, interned into the id namespace corrections mint into. let (out_pool, out_ids) = (&mut self.out_pool, &mut self.out_ids); - merged_run(instance.output_batches, keys, |h, d, t, r| { - let id = *out_ids.entry(d.clone()).or_insert_with(|| { - out_pool.push(d.clone()); + merged_run(instance.output_batches, keys, |hash, data, time, diff| { + let id = *out_ids.entry(data.clone()).or_insert_with(|| { + out_pool.push(data.clone()); (out_pool.len() - 1) as u64 }); - window.output.push(((h, id), t.clone(), r.clone())); + window.output.push(((hash, id), time.clone(), diff.clone())); }); - consolidate_updates(&mut window.output); } + #[inline(never)] fn reduce_corrections( &mut self, keys: &[u64], @@ -416,17 +297,9 @@ where (corr, corr_ends) } + #[inline(never)] fn emit(&mut self, tile: usize, records: &[((u64, u64), T, R)]) { - // A call carries the whole of every key hash it mentions, and calls arrive in ascending - // hash order, so the tile stays hash-ordered and only the run just appended can need - // reordering — by the real `(key, out)` value, since interned ids are in first-seen order - // rather than value order. Consolidating here rather than over the whole tile at `finish` - // is the difference between a sort per emit and one sort of everything the retire produced. let mark = self.tile_rows[tile].len(); - debug_assert!( - self.tile_rows[tile].last().is_none_or(|last| records.first().is_none_or(|r| last.0.0 <= r.0.0)), - "emit must arrive in ascending key-hash order", - ); for ((h, vid), t, d) in records { let row = self.out_pool[*vid as usize].clone(); self.tile_rows[tile].push(((*h, row), t.clone(), d.clone())); @@ -434,6 +307,7 @@ where consolidate_updates_from(&mut self.tile_rows[tile], mark); } + #[inline(never)] fn finish(&mut self) -> Vec> { let tiles = std::mem::take(&mut self.tiles); let tile_rows = std::mem::take(&mut self.tile_rows); @@ -441,19 +315,127 @@ where .into_iter() .zip(tile_rows) .map(|(desc, rows)| { - // Already ordered and consolidated: `emit` did it a run at a time. - let chunks: Vec> = rows - .chunks( as crate::trace::chunk::Chunk>::TARGET) - .map(|piece| { - let mut chunk = VecChunk::default(); - for update in piece { - chunk.push_into(update.clone()); - } - chunk - }) - .collect(); + let mut chunks: Vec> = Vec::default(); + let mut iter = rows.into_iter(); + while iter.len() > 0 { + let mut chunk = VecChunk::default(); + for update in (&mut iter).take( as crate::trace::chunk::Chunk>::TARGET) { + chunk.push_into(update); + } + chunks.push(chunk); + } Rc::new(ChunkBatch::new(chunks, desc)) }) .collect() } } + +/// Walks `batches` restricted to the ascending `keys`, emitting each record as +/// `(hash, &payload, &time, &diff)` in `(hash, payload, time)` order. +/// +/// Per hash bracket, the batches' contiguous runs are gathered and sorted by `(payload, time)`, so +/// equal payloads meet across batches and each payload's times arrive in order (batch intervals +/// are disjoint, so cross-batch times need ordering but never summing). The walk seeks to the +/// first requested key and stops after the last, so a bounded window pays for its own range. +#[inline(never)] +fn merged_run( + batches: &[VBatch], + keys: &[u64], + mut sink: impl FnMut(u64, &D, &T, &R), +) where + D: Ord + Clone + 'static, + T: Lattice + Timestamp, + R: Semigroup + Ord + Clone + 'static, +{ + if keys.is_empty() { + return; + } + let n = batches.len(); + let (mut ci, mut oi) = (vec![0usize; n], vec![0usize; n]); + let mut cur: Vec<&[((u64, D), T, R)]> = vec![&[]; n]; + // Seek each batch to the first record with hash at or above the window's first key. + let k0 = keys[0]; + for b in 0..n { + let chunks = &batches[b].chunks; + ci[b] = chunks.partition_point(|c| c.as_slice().last().is_some_and(|r| r.0.0 < k0)); + cur[b] = chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); + oi[b] = cur[b].partition_point(|r| r.0.0 < k0); + } + let mut scratch: Vec<(&D, &T, &R)> = Vec::new(); + let mut ki = 0usize; + loop { + // Least hash among the batch heads. + let mut minh: Option = None; + for b in 0..n { + if let Some(r) = cur[b].get(oi[b]) { + if minh.is_none_or(|m| r.0.0 < m) { + minh = Some(r.0.0); + } + } + } + let Some(h) = minh else { break }; + while ki < keys.len() && keys[ki] < h { + ki += 1; + } + if ki >= keys.len() { + break; + } + if keys[ki] != h { + // `h` is not wanted. Seeking every batch to the next key that IS wanted costs a binary + // search per batch and lands at or above it, so at most one unwanted hash is visited + // per wanted key. Walking `h`'s records instead would make the whole pass linear in the + // accumulated history rather than in what is asked for — which on an iterative + // computation, where a retire asks for a handful of scattered keys, is the difference + // between presenting the window and re-reading the trace. When the key set is dense + // this branch is simply never taken, so there is no threshold to tune. + let next = keys[ki]; + for b in 0..n { + loop { + let Some(chunk) = batches[b].chunks.get(ci[b]) else { cur[b] = &[]; break }; + let slice = chunk.as_slice(); + if slice.last().is_some_and(|r| r.0.0 < next) { + ci[b] += 1; + oi[b] = 0; + continue; + } + cur[b] = slice; + oi[b] += slice[oi[b]..].partition_point(|r| r.0.0 < next); + // Chunks are non-empty (`ChunkBatch::new` asserts it) and the guard above + // skipped any whose last key is below `next`, so this chunk holds a record at + // or above `next` and the search cannot land at the end. The branch below is + // unreachable; it is kept so a violated invariant degrades to a slower walk + // rather than to a batch that silently reads as drained. + debug_assert!(oi[b] < slice.len(), "a chunk kept by the skip guard must hold a record at or above the sought key"); + if oi[b] >= slice.len() { + ci[b] += 1; + oi[b] = 0; + continue; + } + break; + } + } + continue; + } + scratch.clear(); + for b in 0..n { + while let Some(r) = cur[b].get(oi[b]) { + if r.0.0 != h { + break; + } + scratch.push((&r.0.1, &r.1, &r.2)); + oi[b] += 1; + if oi[b] >= cur[b].len() { + ci[b] += 1; + oi[b] = 0; + cur[b] = batches[b].chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); + } + } + } + { + scratch.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1))); + for (d, t, r) in scratch.drain(..) { + sink(h, d, t, r); + } + } + } +} diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs index b3e14b467..6872c0ed4 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -115,6 +115,105 @@ macro_rules! proxy_reduce { }; } +/// A `String`-valued churn: the same shape as `churn`, but every value clone allocates. The +/// `u64` workloads cannot show what the proxy boundary costs to marshal owned data, because +/// cloning a `u64` is free. +fn wide_keys() -> u64 { sized("WIDE_KEYS", 20_000) } +fn wide_len() -> usize { sized("WIDE_LEN", 48) as usize } + +fn wide_value(k: u64, t: u64, len: usize) -> String { + let mut s = String::with_capacity(len); + s.push_str(&format!("{k:016x}{t:016x}")); + while s.len() < len { s.push('x'); } + s +} + +fn run_wide(mode: Mode) -> f64 { + let keys = wide_keys(); + let len = wide_len(); + timely::execute_directly(move |worker| { + let mut input: InputSession = InputSession::new(); + let probe = worker.dataflow::(|scope| { + let coll = input.to_collection(scope); + let mut ph = ProbeHandle::new(); + match mode { + Mode::Mainline => { + coll.reduce(|_k, i: &[(&String, isize)], o: &mut Vec<(String, isize)>| { + if let Some(m) = i.iter().filter(|(_, d)| *d > 0).map(|(v, _)| (*v).clone()).max() { + o.push((m, 1)); + } + }).inner.probe_with(&mut ph); + } + Mode::CursorSame => { + let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, + ContainerChunker>, + VChunkBatcher, + VChunkBuilder, + VChunkSpine>(hashed.inner, Pipeline, "ArrW"); + arr.reduce_core::<_, VChunkBuilder, + VChunkSpine, + as Cursor>::KeyContainer, _>( + "CursorWide", + |_h, input, current, updates| { + if let Some(kv) = input.iter().filter(|(_, d)| *d > 0).map(|(kv, _)| (*kv).clone()).max_by(|a, b| a.1.cmp(&b.1)) { + updates.push((kv, 1)); + } + for (w, d) in current.iter() { updates.push((w.clone(), -*d)); } + }, + |chunk, key, list| { + use timely::container::PushInto; + *chunk = Default::default(); + for (v, t, d) in list.drain(..) { chunk.push_into(((*key, v), t, d)); } + }, + ).stream.probe_with(&mut ph); + } + Mode::Proxy => { + let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); + let arr = arrange_core::, + ContainerChunker>, + VChunkBatcher, + VChunkBuilder, + VChunkSpine>(hashed.inner, Pipeline, "ArrW"); + reduce_with_tactic::<_, VChunkSpine, _>( + arr, "ProxyWide", + ProxyReduceTactic::new(VecReduceBackend::new( + |_k: &u64, input: &[(String, isize)], current: &mut Vec<(String, isize)>, updates: &mut Vec<(String, isize)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { updates.push((w.clone(), -*d)); } + }, + )), + ).stream.probe_with(&mut ph); + } + } + ph + }); + input.advance_to(0); + for k in 0..keys { input.insert((k, wide_value(k, 0, len))); } + input.advance_to(1); + input.flush(); + while probe.less_than(&1) { worker.step(); } + let mut times = Vec::new(); + let warm = warmup(); + for r in 0..rounds() { + let t = 1 + r; + input.advance_to(t); + for k in 0..keys { + if t > 1 { input.remove((k, wide_value(k, t - 1, len))); } + input.insert((k, wide_value(k, t, len))); + } + input.advance_to(t + 1); + input.flush(); + let start = Instant::now(); + while probe.less_than(&(t + 1)) { worker.step(); } + if r >= warm { times.push(start.elapsed().as_micros()); } + } + times.iter().sum::() as f64 / times.len() as f64 + }) +} + fn max_fold(iter: impl Iterator) -> Option<(u64, u64)> { iter.max_by_key(|kv| kv.1) } @@ -338,6 +437,7 @@ fn bench_reduce_tactics() { "churn" => run_churn(mode), "multimoment" => run_multimoment(mode), "propagate" => run_propagate(mode).0, + "wide" => run_wide(mode), other => panic!("unknown PROG {other:?}"), }; eprintln!(" {prog} {:?}: {us:.0}us/round", mode); From 4befef76d9ced7ad2ced9f12825ffc61e5949a4e Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 11 Aug 2026 13:54:28 -0400 Subject: [PATCH 11/11] flush unneeded state --- .../src/operators/int_proxy/vec_backend.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 08307c4ef..426391864 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -56,11 +56,10 @@ pub struct VecReduceBackend { /// Resolves an input value id (a per-window ordinal) to its `(key, val)` row. in_pool: Vec<(K, V)>, - /// Resolves an output value id to its `(key, out)` row; spans the whole retire, because - /// corrections mint ids that later windows' presentations and `emit` must agree on. + /// Resolves an output value id to its `(key, out)` row, for the current window only. out_pool: Vec<(K, W)>, - /// Interns `(key, out)` rows to their output id. Lookup-only: the map's iteration order is - /// never observed, so the hasher cannot affect what the operator produces. + /// Interns `(key, out)` rows to their output id, for the current window only. + /// Lookup-only: the non-determinism of the map's iteration order is never observed. out_ids: HashMap<(K, W), u64>, /// The retire's output tile descriptions, and the rows accumulated for each. @@ -103,8 +102,6 @@ where fn begin(&mut self, tiles: &[Description]) { self.keys_stale = true; - self.out_pool.clear(); - self.out_ids.clear(); self.tiles = tiles.to_vec(); self.tile_rows = (0..tiles.len()).map(|_| Vec::new()).collect(); } @@ -182,6 +179,8 @@ where }); // The output history, interned into the id namespace corrections mint into. + self.out_ids.clear(); + self.out_pool.clear(); let (out_pool, out_ids) = (&mut self.out_pool, &mut self.out_ids); merged_run(instance.output_batches, keys, |hash, data, time, diff| { let id = *out_ids.entry(data.clone()).or_insert_with(|| { @@ -309,6 +308,9 @@ where #[inline(never)] fn finish(&mut self) -> Vec> { + self.in_pool.clear(); + self.out_pool.clear(); + self.out_ids.clear(); let tiles = std::mem::take(&mut self.tiles); let tile_rows = std::mem::take(&mut self.tile_rows); tiles