From 2446126819e2adc495e27ef90aea199b14bafdc2 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 21 Sep 2026 13:31:27 -0400 Subject: [PATCH 1/3] Prototype collective difference containers --- .../src/operators/int_proxy/diffs.rs | 154 ++++++++++++++++++ .../src/operators/int_proxy/mod.rs | 1 + 2 files changed, 155 insertions(+) create mode 100644 differential-dataflow/src/operators/int_proxy/diffs.rs diff --git a/differential-dataflow/src/operators/int_proxy/diffs.rs b/differential-dataflow/src/operators/int_proxy/diffs.rs new file mode 100644 index 000000000..d4dd05e37 --- /dev/null +++ b/differential-dataflow/src/operators/int_proxy/diffs.rs @@ -0,0 +1,154 @@ +//! A prototype of collective difference operations. +//! +//! The proxy tactics still use scalar differences. +//! This module isolates the proposed storage boundary and exercises it without changing their time logic. +//! Containers own accumulation and movement; callers supply row selections and groups. +//! No scalar difference type, random access, ordering, or constructible zero is required. + +use crate::difference::{Multiply, Semigroup}; + +/// Storage for differences, with collective movement and accumulation. +/// +/// Row numbers refer to logical positions, not to a required physical layout. +/// Destinations must have the same runtime schema and accumulation semantics as their sources. +/// Operations append to their destinations, allowing several sources to contribute to one collection. +pub trait DiffContainer: Sized { + /// The number of differences. + fn len(&self) -> usize; + /// Whether there are no differences. + fn is_empty(&self) -> bool { self.len() == 0 } + /// Empty storage with this container's schema and accumulation semantics. + fn empty(&self) -> Self; + /// Discard differences while retaining the schema and reusable storage. + fn clear(&mut self); + + /// Append the selected source rows in the supplied order, including repetitions. + fn copy_from(&mut self, source: &Self, rows: &[usize]); + + /// Append one sum for each nonempty group of selected source rows. + /// + /// Group `g` occupies `rows[ends[g - 1]..ends[g]]`, with the first group starting at zero. + /// Ends must strictly increase and the last end must equal `rows.len()`. + /// Empty `rows` and `ends` describe no groups. + /// Sums follow the selected row order and have the semantics of repeated `Semigroup::plus_equals`. + /// Zero results are retained so that composed containers preserve group correspondence. + fn sum_from(&mut self, source: &Self, rows: &[usize], ends: &[usize]); + + /// Append the positions of nonzero differences in increasing order. + /// The test has the semantics of `IsZero`; a semigroup without zero reports every position. + fn nonzero(&self, into: &mut Vec); +} + +impl DiffContainer for Vec { + fn len(&self) -> usize { Vec::len(self) } + fn empty(&self) -> Self { Vec::new() } + fn clear(&mut self) { Vec::clear(self); } + + fn copy_from(&mut self, source: &Self, rows: &[usize]) { + self.extend(rows.iter().map(|&row| source[row].clone())); + } + + fn sum_from(&mut self, source: &Self, rows: &[usize], ends: &[usize]) { + let mut start = 0; + for &end in ends { + assert!(start < end, "sum groups must be nonempty"); + let mut sum = source[rows[start]].clone(); + for &row in &rows[start + 1..end] { sum.plus_equals(&source[row]); } + self.push(sum); + start = end; + } + assert_eq!(start, rows.len(), "sum groups must cover the selection"); + } + + fn nonzero(&self, into: &mut Vec) { + into.extend(self.iter().enumerate().filter(|(_, diff)| !diff.is_zero()).map(|(row, _)| row)); + } +} + +/// Bilinear multiplication of selected differences, separate from accumulation. +/// +/// Callers may submit bounded blocks of pairs rather than materialize a complete cross product. +/// The backend supplies output storage with the appropriate schema. +pub trait MultiplyContainer: DiffContainer { + /// Storage for the product differences. + type Output: DiffContainer; + /// Append products in pair order, including zero products. + fn multiply_into(&self, rhs: &Rhs, pairs: &[(usize, usize)], into: &mut Self::Output); +} + +impl MultiplyContainer> for Vec +where + R0: Semigroup + Multiply, + R1: Semigroup, + RO: Semigroup, +{ + type Output = Vec; + fn multiply_into(&self, rhs: &Vec, pairs: &[(usize, usize)], into: &mut Vec) { + into.extend(pairs.iter().map(|&(a, b)| self[a].clone().multiply(&rhs[b]))); + } +} + +/// Consolidate aligned data and differences through collective operations. +/// +/// This is an executable consumer of the proposed interface, not a replacement for row consolidation. +/// Sorting touches only data and row numbers; differences decide how to accumulate each group. +/// Data can be `(key, value, time)` or just `value` for an accumulation at a selected time. +/// A production integration should reuse the temporary storage and allow fused implementations. +pub fn consolidate(data: &mut Vec, diffs: &mut C) { + assert_eq!(data.len(), diffs.len()); + let mut rows: Vec<_> = (0..data.len()).collect(); + rows.sort_unstable_by(|&a, &b| data[a].cmp(&data[b])); + let mut ends = Vec::new(); + for i in 1..rows.len() { + if data[rows[i - 1]] != data[rows[i]] { ends.push(i); } + } + if !rows.is_empty() { ends.push(rows.len()); } + + let mut sums = diffs.empty(); + sums.sum_from(diffs, &rows, &ends); + let mut kept = Vec::new(); + sums.nonzero(&mut kept); + let output = kept.iter().map(|&group| { + let start = if group == 0 { 0 } else { ends[group - 1] }; + data[rows[start]].clone() + }).collect(); + diffs.clear(); + diffs.copy_from(&sums, &kept); + *data = output; +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Two independent columns; neither storage nor a scalar row implements Semigroup here. + struct Pair { left: Vec, right: Vec } + + impl DiffContainer for Pair { + fn len(&self) -> usize { self.left.len() } + fn empty(&self) -> Self { Self { left: vec![], right: vec![] } } + fn clear(&mut self) { self.left.clear(); self.right.clear(); } + fn copy_from(&mut self, source: &Self, rows: &[usize]) { + DiffContainer::copy_from(&mut self.left, &source.left, rows); + DiffContainer::copy_from(&mut self.right, &source.right, rows); + } + fn sum_from(&mut self, source: &Self, rows: &[usize], ends: &[usize]) { + self.left.sum_from(&source.left, rows, ends); + self.right.sum_from(&source.right, rows, ends); + } + fn nonzero(&self, into: &mut Vec) { + into.extend(self.left.iter().zip(&self.right).enumerate() + .filter(|(_, (a, b))| **a != 0 || **b != 0).map(|(row, _)| row)); + } + } + + #[test] + fn component_zeroes_are_retained_until_the_whole_diff_is_tested() { + let mut data = vec!["a", "b", "a", "b", "c", "c"]; + let mut diffs = Pair { left: vec![1, 0, -1, 0, 1, -1], right: vec![0, 1, 2, -1, 0, 0] }; + consolidate(&mut data, &mut diffs); + assert_eq!(data, vec!["a"]); + assert_eq!((diffs.left, diffs.right), (vec![0], vec![2])); + } + +} diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs index 50764452a..1cf85fccf 100644 --- a/differential-dataflow/src/operators/int_proxy/mod.rs +++ b/differential-dataflow/src/operators/int_proxy/mod.rs @@ -42,6 +42,7 @@ //! Both are welcome to efficiently notice that there have been no collisions and optimize, //! or to ignore the risk entirely and live dangerously. +pub mod diffs; mod history; pub mod join; From 09bc0c2532d7ad3e4cbf86f647cc7ccf160b180a Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 21 Sep 2026 13:53:35 -0400 Subject: [PATCH 2/3] Integrate collective differences through proxy reduce replay Carry aligned metadata and opaque diff containers through history maintenance and correction feedback. Adapt Vec and Corgi reduce backends, preserving the timestamp schedule. Extend the composed-diff example through partial-order replay. This remains an integration experiment: scalar benchmarks regress, and flat metadata, gathered selections, and per-key scratch need further attention. --- .../src/operators/int_proxy/diffs.rs | 142 ++++++---- .../src/operators/int_proxy/history.rs | 45 ++++ .../src/operators/int_proxy/mod.rs | 8 +- .../src/operators/int_proxy/reduce.rs | 255 ++++++++++++------ .../src/operators/int_proxy/vec_backend.rs | 50 ++-- interactive/src/corgi/reduce.rs | 84 +++--- 6 files changed, 381 insertions(+), 203 deletions(-) diff --git a/differential-dataflow/src/operators/int_proxy/diffs.rs b/differential-dataflow/src/operators/int_proxy/diffs.rs index d4dd05e37..5e2749eb0 100644 --- a/differential-dataflow/src/operators/int_proxy/diffs.rs +++ b/differential-dataflow/src/operators/int_proxy/diffs.rs @@ -1,7 +1,6 @@ //! A prototype of collective difference operations. //! -//! The proxy tactics still use scalar differences. -//! This module isolates the proposed storage boundary and exercises it without changing their time logic. +//! Proxy reduce uses this boundary throughout history replay and correction feedback. //! Containers own accumulation and movement; callers supply row selections and groups. //! No scalar difference type, random access, ordering, or constructible zero is required. @@ -34,6 +33,14 @@ pub trait DiffContainer: Sized { /// Zero results are retained so that composed containers preserve group correspondence. fn sum_from(&mut self, source: &Self, rows: &[usize], ends: &[usize]); + /// Keep exactly these positions, which must be strictly increasing. + /// Containers can override this to compact in place without copying survivors. + fn retain(&mut self, rows: &[usize]) { + let mut kept = self.empty(); + kept.copy_from(self, rows); + *self = kept; + } + /// Append the positions of nonzero differences in increasing order. /// The test has the semantics of `IsZero`; a semigroup without zero reports every position. fn nonzero(&self, into: &mut Vec); @@ -60,6 +67,11 @@ impl DiffContainer for Vec { assert_eq!(start, rows.len(), "sum groups must cover the selection"); } + fn retain(&mut self, rows: &[usize]) { + for (dest, &source) in rows.iter().enumerate() { self.swap(dest, source); } + self.truncate(rows.len()); + } + fn nonzero(&self, into: &mut Vec) { into.extend(self.iter().enumerate().filter(|(_, diff)| !diff.is_zero()).map(|(row, _)| row)); } @@ -88,67 +100,81 @@ where } } -/// Consolidate aligned data and differences through collective operations. -/// -/// This is an executable consumer of the proposed interface, not a replacement for row consolidation. -/// Sorting touches only data and row numbers; differences decide how to accumulate each group. -/// Data can be `(key, value, time)` or just `value` for an accumulation at a selected time. -/// A production integration should reuse the temporary storage and allow fused implementations. -pub fn consolidate(data: &mut Vec, diffs: &mut C) { - assert_eq!(data.len(), diffs.len()); - let mut rows: Vec<_> = (0..data.len()).collect(); - rows.sort_unstable_by(|&a, &b| data[a].cmp(&data[b])); - let mut ends = Vec::new(); - for i in 1..rows.len() { - if data[rows[i - 1]] != data[rows[i]] { ends.push(i); } - } - if !rows.is_empty() { ends.push(rows.len()); } - - let mut sums = diffs.empty(); - sums.sum_from(diffs, &rows, &ends); - let mut kept = Vec::new(); - sums.nonzero(&mut kept); - let output = kept.iter().map(|&group| { - let start = if group == 0 { 0 } else { ends[group - 1] }; - data[rows[start]].clone() - }).collect(); - diffs.clear(); - diffs.copy_from(&sums, &kept); - *data = output; +/// Row metadata aligned with a container of differences. +/// The backend owns the interpretation of differences; the tactic only reads metadata. +pub struct Records { + /// One metadata entry per logical difference. + pub data: Vec, + /// Differences aligned with `data`. + pub diffs: C, + selection: Vec, } -#[cfg(test)] -mod tests { - use super::*; +impl Records { + /// Empty records with the supplied difference storage and schema. + pub fn new(diffs: C) -> Self { + assert!(diffs.is_empty()); + Self { data: Vec::new(), diffs, selection: Vec::new() } + } + /// Number of records. + pub fn len(&self) -> usize { self.data.len() } + /// Whether there are no records. + pub fn is_empty(&self) -> bool { self.data.is_empty() } + /// Discard records while retaining storage and schema. + pub fn clear(&mut self) { self.data.clear(); self.diffs.clear(); } + /// Append selected records, transforming their metadata without inspecting differences. + pub fn extend(&mut self, source: &Records, rows: impl IntoIterator, mut map: impl FnMut(&S) -> D) { + self.selection.clear(); + self.selection.extend(rows); + self.data.extend(self.selection.iter().map(|&row| map(&source.data[row]))); + self.diffs.copy_from(&source.diffs, &self.selection); + } +} - /// Two independent columns; neither storage nor a scalar row implements Semigroup here. - struct Pair { left: Vec, right: Vec } +/// Reusable storage for collective consolidation. +/// Sorting touches only metadata and row numbers; containers accumulate the resulting groups. +/// Data can be `(key, value, time)` or just `value` for an accumulation at a selected time. +pub struct Consolidation { + rows: Vec, + ends: Vec, + kept: Vec, + data: Vec, + sums: C, +} - impl DiffContainer for Pair { - fn len(&self) -> usize { self.left.len() } - fn empty(&self) -> Self { Self { left: vec![], right: vec![] } } - fn clear(&mut self) { self.left.clear(); self.right.clear(); } - fn copy_from(&mut self, source: &Self, rows: &[usize]) { - DiffContainer::copy_from(&mut self.left, &source.left, rows); - DiffContainer::copy_from(&mut self.right, &source.right, rows); - } - fn sum_from(&mut self, source: &Self, rows: &[usize], ends: &[usize]) { - self.left.sum_from(&source.left, rows, ends); - self.right.sum_from(&source.right, rows, ends); - } - fn nonzero(&self, into: &mut Vec) { - into.extend(self.left.iter().zip(&self.right).enumerate() - .filter(|(_, (a, b))| **a != 0 || **b != 0).map(|(row, _)| row)); - } +impl Consolidation { + /// Scratch with the same schema as `diffs`. + pub fn new(diffs: &C) -> Self { + Self { rows: Vec::new(), ends: Vec::new(), kept: Vec::new(), data: Vec::new(), sums: diffs.empty() } } - - #[test] - fn component_zeroes_are_retained_until_the_whole_diff_is_tested() { - let mut data = vec!["a", "b", "a", "b", "c", "c"]; - let mut diffs = Pair { left: vec![1, 0, -1, 0, 1, -1], right: vec![0, 1, 2, -1, 0, 0] }; - consolidate(&mut data, &mut diffs); - assert_eq!(data, vec!["a"]); - assert_eq!((diffs.left, diffs.right), (vec![0], vec![2])); + /// Sort, sum equal metadata, and discard zero sums, retaining scratch allocations. + pub fn consolidate(&mut self, data: &mut Vec, diffs: &mut C) { + assert_eq!(data.len(), diffs.len()); + self.rows.clear(); + self.rows.extend(0..data.len()); + self.rows.sort_unstable_by(|&a, &b| data[a].cmp(&data[b])); + self.ends.clear(); + for i in 1..self.rows.len() { + if data[self.rows[i - 1]] != data[self.rows[i]] { self.ends.push(i); } + } + if !self.rows.is_empty() { self.ends.push(self.rows.len()); } + + self.sums.sum_from(diffs, &self.rows, &self.ends); + self.kept.clear(); + self.sums.nonzero(&mut self.kept); + self.data.extend(self.kept.iter().map(|&group| { + let start = if group == 0 { 0 } else { self.ends[group - 1] }; + data[self.rows[start]].clone() + })); + self.sums.retain(&self.kept); + std::mem::swap(data, &mut self.data); + std::mem::swap(diffs, &mut self.sums); + self.data.clear(); + self.sums.clear(); } +} +/// Consolidate aligned data and differences through collective operations. +pub fn consolidate(data: &mut Vec, diffs: &mut C) { + Consolidation::new(diffs).consolidate(data, diffs); } diff --git a/differential-dataflow/src/operators/int_proxy/history.rs b/differential-dataflow/src/operators/int_proxy/history.rs index 83d8369fc..8cc21b0a5 100644 --- a/differential-dataflow/src/operators/int_proxy/history.rs +++ b/differential-dataflow/src/operators/int_proxy/history.rs @@ -2,3 +2,48 @@ /// A value history suitable for integer proxy values. pub(in crate::operators) type IdHistory = crate::operators::history::ValueHistory; + +use crate::lattice::Lattice; +use super::diffs::{Consolidation, DiffContainer, Records}; + +/// Reduce history with opaque differences and a separate time-ordered replay index. +/// Stepping gathers all edits at a time; advancement consolidates the buffered metadata. +pub(super) struct DiffHistory { + edits: Records<(u64, T), C>, + history: Vec<(T, T, usize)>, // (time, suffix meet, row) + pub buffer: Records<(u64, T), C>, + scratch: Consolidation<(u64, T), C>, +} + +impl DiffHistory { + pub fn new(diffs: &C) -> Self { + Self { + edits: Records::new(diffs.empty()), history: Vec::new(), + buffer: Records::new(diffs.empty()), scratch: Consolidation::new(diffs), + } + } + pub fn load(&mut self, source: &Records<((u64, u64), T), C>, rows: std::ops::Range, meet: Option<&T>) { + self.edits.clear(); + self.edits.extend(source, rows, |((_, id), time)| { + (*id, meet.map_or_else(|| time.clone(), |meet| time.join(meet))) + }); + self.scratch.consolidate(&mut self.edits.data, &mut self.edits.diffs); + self.buffer.clear(); + self.history.clear(); + self.history.extend(self.edits.data.iter().enumerate().map(|(row, (_, time))| (time.clone(), time.clone(), row))); + self.history.sort_unstable_by(|a, b| b.cmp(a)); + self.history.iter_mut().reduce(|prev, cur| { cur.1.meet_assign(&prev.1); cur }); + } + pub fn time(&self) -> Option<&T> { self.history.last().map(|r| &r.0) } + pub fn meet(&self) -> Option<&T> { self.history.last().map(|r| &r.1) } + pub fn step_while_time_is(&mut self, time: &T) { + let mut start = self.history.len(); + while start > 0 && &self.history[start - 1].0 == time { start -= 1; } + self.buffer.extend(&self.edits, self.history[start..].iter().rev().map(|r| r.2), Clone::clone); + self.history.truncate(start); + } + pub fn advance_buffer_by(&mut self, meet: &T) { + for (_, time) in &mut self.buffer.data { time.join_assign(meet); } + self.scratch.consolidate(&mut self.buffer.data, &mut self.buffer.diffs); + } +} diff --git a/differential-dataflow/src/operators/int_proxy/mod.rs b/differential-dataflow/src/operators/int_proxy/mod.rs index 1cf85fccf..0c9341ca9 100644 --- a/differential-dataflow/src/operators/int_proxy/mod.rs +++ b/differential-dataflow/src/operators/int_proxy/mod.rs @@ -3,9 +3,11 @@ //! The tactics are intended to support custom operator implementations without rebuilding //! the non-trivial and often non-obvious time-based logic that supports them. //! -//! The tactics here run DD's operator logic over consolidated `[((u64, u64), time, diff)]` -//! lists, the first integer a hash of the "key" and granule of independence, the second an -//! ephemeral data identifier understood by the backend but opaque to the operator harness. +//! The tactics here run DD's operator logic over consolidated updates with `(u64, u64)` proxies. +//! The first integer is a hash of the "key" and granule of independence. +//! The second is an ephemeral data identifier understood by the backend but opaque to the operator harness. +//! Reduce keeps `((key, id), time)` metadata beside a [`diffs::DiffContainer`]. +//! Join still exchanges scalar-difference [`ProxyBridge`] lists. //! The tactics first elicit proxy identifiers from the backends, perform their necessary time //! and difference based computations to stage integer collections, and then re-invoke the //! backends with those same identifiers to produce the necessary output. diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs index 2235d0aae..5c4115046 100644 --- a/differential-dataflow/src/operators/int_proxy/reduce.rs +++ b/differential-dataflow/src/operators/int_proxy/reduce.rs @@ -8,12 +8,11 @@ use super::pending::Pending; use timely::progress::{Antichain, Timestamp}; use timely::progress::frontier::AntichainRef; -use crate::difference::Semigroup; +use super::diffs::{Consolidation, DiffContainer, Records}; use crate::lattice::Lattice; use crate::trace::{Span, Description}; -use super::ProxyBridge; +use super::history::DiffHistory; use crate::operators::reduce::{sort_dedup, ReduceTactic}; -use crate::operators::history::ValueHistory; /// A unit of proxied reduce work, presented to the backend. pub struct ReduceInstance<'a, T, B1, B2> { @@ -40,20 +39,21 @@ pub struct ReduceInstance<'a, T, B1, B2> { pub struct ReduceWindow { /// The key's full input — novel and prior merged, netted — sorted & consolidated by /// `((key_hash, value_id), time)`. May be advanced to the compaction frontier. - pub input: ProxyBridge, + pub input: Records<((u64, u64), T), RIn>, /// The RAW novel time support: `(key_hash, time)` pairs sorted by `(key_hash, time)` and /// deduplicated, recorded from the novel batches BEFORE any consolidation or advancement — /// a netted-away record's time must still appear here. pub seeds: Vec<(u64, T)>, /// Accumulated output preceding the retire's interval, same ordering as `input`. - pub output: ProxyBridge, + pub output: Records<((u64, u64), T), ROut>, } -impl Default for ReduceWindow { - fn default() -> Self { ReduceWindow { input: Vec::new(), seeds: Vec::new(), output: Vec::new() } } -} +impl ReduceWindow { + /// Empty presentations with backend-supplied difference storage. + pub fn new(input: RIn, output: ROut) -> Self { + Self { input: Records::new(input), seeds: Vec::new(), output: Records::new(output) } + } -impl ReduceWindow { /// Clear the presentations, keeping their allocations. pub fn clear(&mut self) { self.input.clear(); @@ -65,13 +65,13 @@ impl ReduceWindow { /// The reduce backend: value semantics for a proxy-space reduction, driven by [`ProxyReduceTactic`]. /// /// The protocol for each round of invocation is -/// `begin [ next_window reduce_corrections* emit ]* finish`, +/// `begin new_diffs [ next_window reduce_corrections* emit ]* finish`, /// where the window loop runs until `next_window` reports the key space exhausted. pub trait ProxyReduceBackend { - /// Diff type presented for the input. - type RIn: Semigroup; - /// Diff type of the output. - type ROut: Semigroup + 'static; + /// Difference storage presented for the input. + type RIn: DiffContainer; + /// Difference storage for the output. + type ROut: DiffContainer; /// Initiate a session to create batches for these descriptions, which span `[lower, upper)`. /// @@ -80,6 +80,10 @@ pub trait ProxyReduceBackend { /// work in progress, until `finish()` is called. fn begin(&mut self, description: Description); + /// Empty input and output storage with this session's schemas and accumulation semantics. + /// Called after `begin`; all presentations and corrections must use these schemas. + fn new_diffs(&self) -> (Self::RIn, Self::ROut); + /// Present the next window of the key space, and advance `from` past it. /// /// On entry `from` is the inclusive lower bound on key hashes still to be covered. The backend @@ -117,13 +121,13 @@ pub trait ProxyReduceBackend { &mut self, keys: &[u64], in_ends: &[usize], - input: &[(u64, Self::RIn)], + input: &Records, out_ends: &[usize], - output: &[(u64, Self::ROut)], - ) -> (Vec<(u64, Self::ROut)>, Vec); + output: &Records, + ) -> (Records, Vec); /// Commit a collection of updates to the batch in progress. - fn emit(&mut self, records: &[((u64, u64), T, Self::ROut)]); + fn emit(&mut self, records: &Records<((u64, u64), T), Self::ROut>); /// Complete the session matching `begin`, yielding the batch it described, /// or `None` when the span it described carries no updates. @@ -216,32 +220,39 @@ where // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None` // once the backend reports the space covered. let mut from = Some(0u64); - let mut window: ReduceWindow = ReduceWindow::default(); + let (input_diffs, output_diffs) = self.backend.new_diffs(); + let mut window = ReduceWindow::new(input_diffs, output_diffs); // Retire-wide reusable scratch: cleared per group, window or wave, retaining capacity. Fresh // per-key/per-wave `Vec`s were once the dominant cost here, which is why the slots and the // staging buffers are held across the whole retire rather than built where they are used. let mut slots: Vec> = Vec::new(); let mut live: Vec = Vec::new(); - let mut deltas: Vec<((u64, u64), T, Bk::ROut)> = Vec::new(); + let mut deltas = Records::new(window.output.diffs.empty()); + let mut delta_scratch = Consolidation::new(&window.output.diffs); let mut batch_keys: Vec = Vec::new(); let mut in_ends: Vec = Vec::new(); - let mut in_all: Vec<(u64, Bk::RIn)> = Vec::new(); + let mut in_all = Records::new(window.input.diffs.empty()); let mut out_ends: Vec = Vec::new(); - let mut out_all: Vec<(u64, Bk::ROut)> = Vec::new(); + let mut out_all = Records::new(window.output.diffs.empty()); let mut active: Vec<(usize, T)> = Vec::new(); - let mut in_accum: Vec<(u64, Bk::RIn)> = Vec::new(); - let mut cur_out: Vec<(u64, Bk::ROut)> = Vec::new(); + let mut in_accum = Records::new(window.input.diffs.empty()); + let mut cur_out = Records::new(window.output.diffs.empty()); + + let mut in_scratch = Consolidation::new(&window.input.diffs); + let mut out_scratch = Consolidation::new(&window.output.diffs); while from.is_some() { let before = from; window.clear(); self.backend.next_window(&instance, &changed, &mut from, &mut window); - let p_in = &mut window.input; + let p_in = &window.input; let seeds = &window.seeds; - let p_out = &mut window.output; - super::debug_assert_sorted_bridge(p_in, "next_window.input"); - super::debug_assert_sorted_bridge(p_out, "next_window.output"); + let p_out = &window.output; + debug_assert_eq!(p_in.len(), p_in.diffs.len()); + debug_assert!(p_in.data.windows(2).all(|w| w[0] < w[1]), "next_window.input must be sorted and consolidated"); + debug_assert_eq!(p_out.len(), p_out.diffs.len()); + debug_assert!(p_out.data.windows(2).all(|w| w[0] < w[1]), "next_window.output must be sorted and consolidated"); debug_assert!( seeds.windows(2).all(|w| w[0] < w[1]), "next_window.seeds must be sorted by (key_hash, time) and deduplicated", @@ -255,7 +266,7 @@ where ); debug_assert!( { - let mut keys = p_in.iter().map(|r| r.0.0).chain(seeds.iter().map(|s| s.0)).chain(p_out.iter().map(|r| r.0.0)); + let mut keys = p_in.data.iter().map(|r| r.0.0).chain(seeds.iter().map(|s| s.0)).chain(p_out.data.iter().map(|r| r.0.0)); keys.all(|k| before.is_none_or(|b| b <= k) && from.is_none_or(|f| k < f)) }, "next_window must report a key hash entirely within the window that first mentions it", @@ -281,21 +292,21 @@ where live.clear(); // Mapped to hashes before the min: the sources differ in shape. while let Some(key) = [ - p_in.get(is).map(|record| record.0.0), + p_in.data.get(is).map(|record| record.0.0), seeds.get(ns).map(|seed| seed.0), - p_out.get(os).map(|record| record.0.0), + p_out.data.get(os).map(|record| record.0.0), ].into_iter().flatten().min() { let i0 = is; - while is < p_in.len() && p_in[is].0.0 == key { is += 1; } + while is < p_in.len() && p_in.data[is].0.0 == key { is += 1; } let i1 = is; let n0 = ns; while ns < seeds.len() && seeds[ns].0 == key { ns += 1; } let n1 = ns; let o0 = os; - while os < p_out.len() && p_out[os].0.0 == key { os += 1; } + while os < p_out.len() && p_out.data[os].0.0 == key { os += 1; } let o1 = os; - if n_slots == slots.len() { slots.push(KeySweep::empty()); } + if n_slots == slots.len() { slots.push(KeySweep::empty(&p_in.diffs, &p_out.diffs)); } let slot = &mut slots[n_slots]; slot.key = key; slot.pended.clear(); @@ -309,8 +320,8 @@ where let single = owed.first().map(|r| &due.times[r.1]).or_else(|| novel.first().map(|r| &r.1)) .filter(|at| owed.len() <= 1 && novel.len() <= 1 && novel.first().is_none_or(|r| &r.1 == *at) - && p_in[i0..i1].iter().all(|r| r.1.less_equal(at)) - && p_out[o0..o1].iter().all(|r| r.1.less_equal(at))); + && p_in.data[i0..i1].iter().all(|r| r.1.less_equal(at)) + && p_out.data[o0..o1].iter().all(|r| r.1.less_equal(at))); slot.direct = single.map(|_| (i0..i1, o0..o1)); slot.at = if let Some(at) = single { if upper.less_equal(at) { slot.pended.push(at.clone()); None } @@ -319,8 +330,8 @@ where slot.sweep.load( owed.iter().map(|&(_, row)| due.times[row].clone()), novel.iter().map(|r| r.1.clone()), - p_in[i0..i1].iter_mut().map(|r| (r.0.1, std::mem::replace(&mut r.1, T::minimum()), r.2.clone())), - p_out[o0..o1].iter_mut().map(|r| (r.0.1, std::mem::replace(&mut r.1, T::minimum()), r.2.clone())), + p_in, i0..i1, + p_out, o0..o1, ); slot.sweep.next_crossing(upper, &mut slot.pended) }; @@ -348,21 +359,21 @@ where in_accum.clear(); cur_out.clear(); if let Some((ir, or)) = &slots[si].direct { - in_accum.extend(p_in[ir.clone()].iter().map(|r| (r.0.1, r.2.clone()))); - cur_out.extend(p_out[or.clone()].iter().map(|r| (r.0.1, r.2.clone()))); - crate::consolidation::consolidate(&mut in_accum); - crate::consolidation::consolidate(&mut cur_out); + in_accum.extend(p_in, ir.clone(), |r| r.0.1); + cur_out.extend(p_out, or.clone(), |r| r.0.1); } else { slots[si].sweep.input_at(&at, &mut in_accum); slots[si].sweep.output_at(&at, &mut cur_out); } + in_scratch.consolidate(&mut in_accum.data, &mut in_accum.diffs); + out_scratch.consolidate(&mut cur_out.data, &mut cur_out.diffs); // An interesting time can still reach the gate with nothing to read; the // conventional reduce skips user logic there and so do we. if in_accum.is_empty() && cur_out.is_empty() { continue; } batch_keys.push(slots[si].key); - in_all.append(&mut in_accum); + in_all.extend(&in_accum, 0..in_accum.len(), |id| *id); in_ends.push(in_all.len()); - out_all.append(&mut cur_out); + out_all.extend(&cur_out, 0..cur_out.len(), |id| *id); out_ends.push(out_all.len()); active.push((si, at)); } @@ -374,11 +385,9 @@ where let cend = corr_ends[bi]; if cstart != cend { debug_assert!(held.elements().iter().any(|h| h.less_equal(at)), "no held capability <= active time"); - for (vid, d) in &corr[cstart..cend] { - deltas.push(((slots[*si].key, *vid), at.clone(), d.clone())); - } + deltas.extend(&corr, cstart..cend, |vid| ((slots[*si].key, *vid), at.clone())); if slots[*si].direct.is_none() { - slots[*si].sweep.commit(at, corr[cstart..cend].iter().cloned()); + slots[*si].sweep.commit(at, &corr, cstart..cend); } } cstart = cend; @@ -401,8 +410,8 @@ where } if !deltas.is_empty() { - crate::consolidation::consolidate_updates(&mut deltas); - self.backend.emit(&deltas[..]); + delta_scratch.consolidate(&mut deltas.data, &mut deltas.diffs); + self.backend.emit(&deltas); } } @@ -424,9 +433,9 @@ struct KeySweep { at: Option, } -impl KeySweep { - fn empty() -> Self { - KeySweep { key: 0, sweep: Sweep::new(), direct: None, pended: Vec::new(), at: None } +impl KeySweep { + fn empty(input: &RIn, output: &ROut) -> Self { + KeySweep { key: 0, sweep: Sweep::new(input, output), direct: None, pended: Vec::new(), at: None } } } @@ -467,8 +476,8 @@ struct Sweep { /// The accumulated input (novel and prior, merged and netted) and output: join partners, and /// the accumulations to evaluate over. Both may be advanced freely — witness duty lives in /// `seeds`, not in any record. - input: ValueHistory, - output: ValueHistory, + input: DiffHistory, + output: DiffHistory, /// The key's seed times — the harness's due (warned) times merged with the raw novel time /// support — ascending and deduplicated, with their suffix meets; `seed_pos` consumes them. /// These are the ONLY source of interest: the schedule is stated over them, so they are held @@ -486,7 +495,8 @@ struct Sweep { temporary: Vec, /// Corrections emitted so far this sweep, meet-collapsed; both a join partner and part of the /// output accumulation. - produced: Vec<((u64, T), ROut)>, + produced: Records<(u64, T), ROut>, + produced_scratch: Consolidation<(u64, T), ROut>, /// The meet of every time still to come. meet: Option, /// Whether the last `next_crossing` returned a time whose step is not yet settled. @@ -505,14 +515,15 @@ enum Tick { Done, } -impl Sweep { +impl Sweep { /// An empty sweep, to be `load`ed and reused for successive keys. - fn new() -> Self { + fn new(input: &RIn, output: &ROut) -> Self { Sweep { - input: ValueHistory::new(), output: ValueHistory::new(), + input: DiffHistory::new(input), output: DiffHistory::new(output), seeds: Vec::new(), seed_meets: Vec::new(), seed_pos: 0, synth: Vec::new(), reached: Vec::new(), temporary: Vec::new(), - produced: Vec::new(), meet: None, suspended: false, + produced: Records::new(output.empty()), produced_scratch: Consolidation::new(output), + meet: None, suspended: false, } } @@ -528,8 +539,10 @@ impl Sw &mut self, owed: impl Iterator, novel_times: impl Iterator, - input: impl Iterator, - output: impl Iterator, + input: &Records<((u64, u64), T), RIn>, + in_rows: std::ops::Range, + output: &Records<((u64, u64), T), ROut>, + out_rows: std::ops::Range, ) { // Merge the two ascending seed sources, deduplicated. self.seeds.clear(); @@ -564,8 +577,8 @@ impl Sw // be advanced by it at load. let mut meet: Option = None; update_meet(&mut meet, self.seed_meets.first()); - self.input.load_iter(input, meet.as_ref()); - self.output.load_iter(output, meet.as_ref()); + self.input.load(input, in_rows, meet.as_ref()); + self.output.load(output, out_rows, meet.as_ref()); self.meet = meet; } @@ -677,11 +690,11 @@ impl Sw self.input.advance_buffer_by(meet); self.output.advance_buffer_by(meet); } - self.temporary.extend(self.input.buffer().iter().map(|((_, t), _)| t) + self.temporary.extend(self.input.buffer.data.iter().map(|(_, t)| t) .filter(|t| !t.less_equal(at)).map(|t| t.join(at))); - self.temporary.extend(self.output.buffer().iter().map(|((_, t), _)| t) + self.temporary.extend(self.output.buffer.data.iter().map(|(_, t)| t) .filter(|t| !t.less_equal(at)).map(|t| t.join(at))); - self.temporary.extend(self.produced.iter().map(|((_, t), _)| t) + self.temporary.extend(self.produced.data.iter().map(|(_, t)| t) .filter(|t| !t.less_equal(at)).map(|t| t.join(at))); } sort_dedup(&mut self.temporary); @@ -695,31 +708,28 @@ impl Sw } } - /// The input accumulation at the suspended time. - fn input_at(&self, at: &T, into: &mut Vec<(u64, RIn)>) { - for ((id, time), diff) in self.input.buffer().iter() { - if time.less_equal(at) { into.push((*id, diff.clone())); } - } - crate::consolidation::consolidate(into); + /// The input accumulation at the suspended time, to be consolidated by the caller. + fn input_at(&self, at: &T, into: &mut Records) { + let buffer = &self.input.buffer; + into.extend(buffer, (0..buffer.len()).filter(|&row| buffer.data[row].1.less_equal(at)), |r| r.0); } /// The tentative output accumulation at the suspended time, including this sweep's corrections. - fn output_at(&self, at: &T, into: &mut Vec<(u64, ROut)>) { - for ((id, time), diff) in self.output.buffer().iter().chain(self.produced.iter()) { - if time.less_equal(at) { into.push((*id, diff.clone())); } + /// The caller consolidates the combined selection. + fn output_at(&self, at: &T, into: &mut Records) { + for buffer in [&self.output.buffer, &self.produced] { + into.extend(buffer, (0..buffer.len()).filter(|&row| buffer.data[row].1.less_equal(at)), |r| r.0); } - crate::consolidation::consolidate(into); } /// Record the corrections evaluated at the suspended time, and collapse them by the meet. - fn commit(&mut self, at: &T, corrections: impl Iterator) { - let before = self.produced.len(); - for (id, diff) in corrections { self.produced.push(((id, at.clone()), diff)); } - if self.produced.len() > before { + fn commit(&mut self, at: &T, corrections: &Records, rows: std::ops::Range) { + if !rows.is_empty() { + self.produced.extend(corrections, rows, |id| (*id, at.clone())); if let Some(meet) = self.meet.as_ref() { - for entry in self.produced.iter_mut() { (entry.0).1.join_assign(meet); } + for (_, time) in &mut self.produced.data { time.join_assign(meet); } } - crate::consolidation::consolidate(&mut self.produced); + self.produced_scratch.consolidate(&mut self.produced.data, &mut self.produced.diffs); } } @@ -738,3 +748,82 @@ impl Sw self.meet = meet; } } + +#[cfg(test)] +mod tests { + use super::*; + use super::super::diffs::consolidate; + use timely::order::Product; + + /// Two independent columns; neither storage nor a scalar row implements Semigroup here. + struct Pair { left: Vec, right: Vec } + + impl DiffContainer for Pair { + fn len(&self) -> usize { self.left.len() } + fn empty(&self) -> Self { Self { left: vec![], right: vec![] } } + fn clear(&mut self) { self.left.clear(); self.right.clear(); } + fn copy_from(&mut self, source: &Self, rows: &[usize]) { + DiffContainer::copy_from(&mut self.left, &source.left, rows); + DiffContainer::copy_from(&mut self.right, &source.right, rows); + } + fn sum_from(&mut self, source: &Self, rows: &[usize], ends: &[usize]) { + self.left.sum_from(&source.left, rows, ends); + self.right.sum_from(&source.right, rows, ends); + } + fn nonzero(&self, into: &mut Vec) { + into.extend(self.left.iter().zip(&self.right).enumerate() + .filter(|(_, (a, b))| **a != 0 || **b != 0).map(|(row, _)| row)); + } + } + + #[test] + fn columnar_diffs_survive_consolidation_and_history_replay() { + let mut data = vec!["a", "b", "a", "b", "c", "c"]; + let mut diffs = Pair { left: vec![1, 0, -1, 0, 1, -1], right: vec![0, 1, 2, -1, 0, 0] }; + consolidate(&mut data, &mut diffs); + assert_eq!(data, vec!["a"]); + assert_eq!((diffs.left, diffs.right), (vec![0], vec![2])); + + let schema = Pair { left: vec![], right: vec![] }; + let mut input = Records::new(schema.empty()); + input.data = vec![((0, 7), Product::new(0u64, 1u64)), ((0, 7), Product::new(1, 0)), ((0, 7), Product::new(2, 2))]; + input.diffs = Pair { left: vec![1, -1, 3], right: vec![0, 2, -2] }; + let output = Records::new(schema.empty()); + let mut sweep = Sweep::new(&schema, &schema); + sweep.load(std::iter::empty(), input.data.iter().map(|r| r.1), &input, 0..3, &output, 0..0); + let mut pended = Vec::new(); + let mut accum = Records::new(schema.empty()); + let mut current = Records::new(schema.empty()); + // Identity reduction: corrections from earlier crossings must enter later output sums. + for (time, expected_input, expected_output) in [ + (Product::new(0, 1), (1, 0), None), + (Product::new(1, 0), (-1, 2), None), + (Product::new(1, 1), (0, 2), Some((0, 2))), + (Product::new(2, 2), (3, 0), Some((0, 2))), + ] { + assert_eq!(sweep.next_crossing(&Antichain::new(), &mut pended), Some(time)); + accum.clear(); + current.clear(); + sweep.input_at(&time, &mut accum); + sweep.output_at(&time, &mut current); + consolidate(&mut accum.data, &mut accum.diffs); + consolidate(&mut current.data, &mut current.diffs); + assert_eq!(accum.data, [7]); + assert_eq!((accum.diffs.left[0], accum.diffs.right[0]), expected_input); + if let Some(expected) = expected_output { + assert_eq!(current.data, [7]); + assert_eq!((current.diffs.left[0], current.diffs.right[0]), expected); + } else { + assert!(current.is_empty()); + } + let previous = expected_output.unwrap_or((0, 0)); + accum.diffs.left[0] -= previous.0; + accum.diffs.right[0] -= previous.1; + consolidate(&mut accum.data, &mut accum.diffs); + sweep.commit(&time, &accum, 0..accum.len()); + } + assert_eq!(sweep.next_crossing(&Antichain::new(), &mut pended), None); + assert!(pended.is_empty()); + } + +} diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 253534169..029bd955e 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -34,6 +34,7 @@ use crate::trace::chunk::ChunkBatch; use crate::trace::chunk::vec::VecChunk; use crate::trace::Description; +use super::diffs::Records; use super::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; /// The batch type of a hash-keyed [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine): updates @@ -100,8 +101,10 @@ where R: Semigroup + Ord + Clone + 'static, L: FnMut(&K, &[(V, R)], &mut Vec<(W, R)>, &mut Vec<(W, R)>), { - type RIn = R; - type ROut = R; + type RIn = Vec; + type ROut = Vec; + + fn new_diffs(&self) -> (Vec, Vec) { (Vec::new(), Vec::new()) } fn begin(&mut self, _description: Description) { self.chunks.clear(); @@ -113,7 +116,7 @@ where instance: &ReduceInstance<'_, T, VBatch<(K, V), T, R>, VBatch<(K, W), T, R>>, changed: &[u64], from: &mut Option, - window: &mut ReduceWindow, + window: &mut ReduceWindow, Vec>, ) { let Some(start) = *from else { return }; @@ -157,16 +160,18 @@ where } // Novel and prior share the id space and arrive (id, time)-adjacent, so an exactly // cancelling pair meets here and nets away; its time survives in the seeds. - if let Some(((k2, i2), t2, d2)) = window.input.last_mut() { + if let Some((((k2, i2), t2), d2)) = window.input.data.last().zip(window.input.diffs.last_mut()) { if *k2 == key && *i2 == id && *t2 == time { d2.plus_equals(&diff); if d2.is_zero() { - window.input.pop(); + window.input.data.pop(); + window.input.diffs.pop(); } continue; } } - window.input.push(((key, id), time, diff)); + window.input.data.push(((key, id), time)); + window.input.diffs.push(diff); } // Per key, seeds arrive in (value, time) order; the contract wants (key, time) order. seed_scratch.sort(); @@ -205,7 +210,8 @@ where last = Some(data); } let id = (self.out_pool.len() - 1) as u64; - window.output.push(((key, id), time, diff)); + window.output.data.push(((key, id), time)); + window.output.diffs.push(diff); } // The two passes agree or the hash collides. Neither can see this alone: a hash whose // input has fully cancelled for one real key still carries that key's stale output. @@ -225,11 +231,11 @@ where &mut self, keys: &[u64], in_ends: &[usize], - input: &[(u64, R)], + input: &Records>, out_ends: &[usize], - output: &[(u64, R)], - ) -> (Vec<(u64, R)>, Vec) { - let mut corr: Vec<(u64, R)> = Vec::new(); + output: &Records>, + ) -> (Records>, Vec) { + let mut corr = Records::new(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(); @@ -240,15 +246,15 @@ where // No hash collision fast path, expected to be the most common case. let collides = !self.collisions.is_empty() && self.collisions.binary_search(&keys[i]).is_ok(); let single_key: Option = if collides { None } else { - input[is..ie].first().map(|(vid, _)| self.in_pool[*vid as usize].0.clone()) - .or_else(|| output[os..oe].first().map(|(vid, _)| self.out_pool[*vid as usize].0.clone())) + input.data[is..ie].first().map(|vid| self.in_pool[*vid as usize].0.clone()) + .or_else(|| output.data[os..oe].first().map(|vid| self.out_pool[*vid as usize].0.clone())) }; 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()))); + input_vals.extend(input.data[is..ie].iter().zip(&input.diffs[is..ie]).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()))); + current.extend(output.data[os..oe].iter().zip(&output.diffs[os..oe]).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); @@ -259,16 +265,17 @@ where self.out_pool.push(key_w); (self.out_pool.len() - 1) as u64 }); - corr.push((id, d)); + corr.data.push(id); + corr.diffs.push(d); } } else { let mut ins: BTreeMap> = BTreeMap::new(); - for (vid, d) in &input[is..ie] { + for (vid, d) in input.data[is..ie].iter().zip(&input.diffs[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] { + for (vid, d) in output.data[os..oe].iter().zip(&output.diffs[os..oe]) { let (k, w) = &self.out_pool[*vid as usize]; outs.entry(k.clone()).or_default().push((w.clone(), d.clone())); } @@ -289,7 +296,8 @@ where self.out_pool.push(key_w); (self.out_pool.len() - 1) as u64 }); - corr.push((id, d)); + corr.data.push(id); + corr.diffs.push(d); } } } @@ -301,9 +309,9 @@ where } #[inline(never)] - fn emit(&mut self, records: &[((u64, u64), T, R)]) { + fn emit(&mut self, records: &Records<((u64, u64), T), Vec>) { self.stage.clear(); - for ((h, vid), t, d) in records { + for (((h, vid), t), d) in records.data.iter().zip(&records.diffs) { let row = self.out_pool[*vid as usize].clone(); self.stage.push(((*h, row), t.clone(), d.clone())); } diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 26a02ecd5..ecda353b6 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -28,10 +28,11 @@ use std::collections::HashMap; use std::hash::{BuildHasherDefault, Hasher}; use std::rc::Rc; +#[cfg(test)] use differential_dataflow::consolidation::consolidate_updates; use differential_dataflow::trace::Description; use differential_dataflow::trace::chunk::ChunkBatch; -use differential_dataflow::operators::int_proxy::ProxyBridge; +use differential_dataflow::operators::int_proxy::diffs::{consolidate, Records}; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; use corgi::arrange::{compare_at, gather, gather_lanes, sort_blocks}; @@ -289,7 +290,7 @@ where fn merge_present( keys_col: &CValue, vals_col: &CValue, khs: &[u64], vids: &[u64], times: &mut [T], diffs: &[Diff], run_ends: &[usize], - bridge: &mut ProxyBridge, + bridge: &mut Records<((u64, u64), T), Vec>, ) -> bool { let ordered_keys = corgi::arrange::leaf_slice(keys_col).is_some() || { let mut start = 0usize; @@ -326,7 +327,7 @@ fn merge_present( current.as_mut().unwrap().2 += diff; } else { if let Some(record) = current.take() { - if record.2 != 0 { bridge.push(record); } + if record.2 != 0 { bridge.data.push((record.0, record.1)); bridge.diffs.push(record.2); } } current = Some((kv, time, diff)); } @@ -338,7 +339,7 @@ fn merge_present( } drop(accumulate); if let Some(record) = current { - if record.2 != 0 { bridge.push(record); } + if record.2 != 0 { bridge.data.push((record.0, record.1)); bridge.diffs.push(record.2); } } return true; } @@ -363,7 +364,7 @@ fn merge_present( } drop(accumulate); if let Some(record) = current { - if record.2 != 0 { bridge.push(record); } + if record.2 != 0 { bridge.data.push((record.0, record.1)); bridge.diffs.push(record.2); } } true } @@ -390,7 +391,7 @@ where &mut self, chunks: &[&CorgiChunk], keys: &[u64], - bridge: &mut ProxyBridge, + bridge: &mut Records<((u64, u64), T), Vec>, ) { let (p_keys, p_vals, khs, mut times, diffs, run_ends) = collect_present(chunks, keys); if khs.is_empty() { @@ -401,22 +402,24 @@ where self.input.register(p_vals, &vids); self.keys.register(p_keys, &khs); if !merged { - bridge.extend(times.into_iter().enumerate().map(|(i, time)| ((khs[i], vids[i]), time, diffs[i]))); - consolidate_updates(bridge); + bridge.data.extend(times.into_iter().enumerate().map(|(i, time)| ((khs[i], vids[i]), time))); + bridge.diffs.extend(diffs); + consolidate(&mut bridge.data, &mut bridge.diffs); } } /// The one value crossing for a retire: every `(key, time)` bracket at once. Builds the output /// value COLUMN directly per reducer, registers it (id → row) into the val pool, and returns the - /// proxy `(value_id, diff)` deltas with per-bracket ends. `input[k] = (value_id, accumulated diff)`; the bracket `i` is `input[ends[i-1]..ends[i]]`, non-empty. - fn reduce_brackets(&mut self, ends: &[usize], input: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { + /// proxy `(value_id, diff)` deltas with per-bracket ends. + /// Input value IDs and accumulated diffs are aligned; ends delimit the brackets. + fn reduce_brackets(&mut self, ends: &[usize], input: &Records>) -> (Vec<(u64, Diff)>, Vec) { // Primitive IDs contain the signed integer itself; a segmented minimum // needs neither payload resolution nor a structural sort. if matches!(self.reducer, Reducer::Min) && self.input.depth.is_some() { let (mut values, mut output_ends) = (Vec::new(), Vec::with_capacity(ends.len())); let mut start = 0; for &end in ends { - if let Some(&(id, _)) = input[start..end].iter().filter(|r| r.1 != 0).min_by_key(|r| r.0 as i64) { + if let Some((&id, _)) = input.data[start..end].iter().zip(&input.diffs[start..end]).filter(|r| *r.1 != 0).min_by_key(|r| *r.0 as i64) { values.push((id, 1)); } output_ends.push(values.len()); @@ -435,7 +438,7 @@ where let mut sums: Vec = Vec::new(); let mut start = 0; for &end in ends { - let c: Diff = input[start..end].iter().map(|&(_, d)| d).sum(); + let c: Diff = input.diffs[start..end].iter().sum(); if c > 0 { sums.push(c as u64); out_diffs.push(1); @@ -459,7 +462,7 @@ where let mut present = 0usize; let mut start = 0; for &end in ends { - if input[start..end].iter().any(|&(_, d)| d != 0) { + if input.diffs[start..end].iter().any(|&d| d != 0) { present += 1; out_diffs.push(1); } @@ -488,8 +491,8 @@ where let lo = cand_reps.len(); let seg = block_starts.len() as u64; for k in start..end { - if input[k].1 != 0 { - cand_reps.push(input[k].0); + if input.diffs[k] != 0 { + cand_reps.push(input.data[k]); labels.push(seg); } } @@ -524,11 +527,11 @@ where let mut blocks: Vec<(usize, usize)> = Vec::with_capacity(ends.len()); let mut start = 0; for (bi, &end) in ends.iter().enumerate() { - if input[start..end].iter().any(|&(_, d)| d != 0) { + if input.diffs[start..end].iter().any(|&d| d != 0) { let lo = entry_reps.len(); for k in start..end { - entry_reps.push(input[k].0); - entry_diffs.push(input[k].1); + entry_reps.push(input.data[k]); + entry_diffs.push(input.diffs[k]); labels.push(bi as u64); } blocks.push((lo, entry_reps.len())); @@ -573,8 +576,10 @@ impl ProxyReduceBackend, CBatch> for CorgiReduceBackend where T: ColTime + Ord, { - type RIn = Diff; - type ROut = Diff; + type RIn = Vec; + type ROut = Vec; + + fn new_diffs(&self) -> (Vec, Vec) { (Vec::new(), Vec::new()) } fn begin(&mut self, _description: Description) { // Open the output session for this retire; reset the per-retire resolution pools. @@ -584,7 +589,7 @@ where self.rows = (Vec::new(), Vec::new(), ColTimes::default(), Vec::new()); } - fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch, CBatch>, changed: &[u64], from: &mut Option, window: &mut ReduceWindow) { + fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch, CBatch>, changed: &[u64], from: &mut Option, window: &mut ReduceWindow, Vec>) { // Single window: present the WHOLE key space at once, and report it covered. This is NOT a // deferred refinement — bounded windows were measured and rejected: at WINDOW = 1<<14, scc // (100 rounds x batch 100) cost 84.4s against 63.7s, a 33% regression, while peak RSS @@ -655,16 +660,17 @@ where self.keys.register(o_keys, &o_khs); self.vals.register(o_vals, &vids); if !merged { - window.output.extend(o_times.into_iter().enumerate().map(|(i, time)| ((o_khs[i], vids[i]), time, o_diffs[i]))); - consolidate_updates(&mut window.output); + window.output.data.extend(o_times.into_iter().enumerate().map(|(i, time)| ((o_khs[i], vids[i]), time))); + window.output.diffs.extend(o_diffs); + consolidate(&mut window.output.data, &mut window.output.diffs); } } } - fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { + fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &Records>, out_ends: &[usize], output: &Records>) -> (Records>, Vec) { let (desired, desired_ends) = self.reduce_brackets(in_ends, input); - let mut corr: Vec<(u64, Diff)> = Vec::new(); + let mut corr = Records::new(Vec::new()); let mut corr_ends: Vec = Vec::with_capacity(keys.len()); let (mut ds, mut os) = (0usize, 0usize); // Scratch for netting, cleared per key rather than allocated per key. @@ -678,12 +684,12 @@ where for &(vid, d) in &desired[ds..de] { if let Some(x) = net.get_mut(&vid) { *x += d; } else { net.insert(vid, d); order.push(vid); } } - for &(vid, d) in &output[os..oe] { + for (&vid, &d) in output.data[os..oe].iter().zip(&output.diffs[os..oe]) { if let Some(x) = net.get_mut(&vid) { *x -= d; } else { net.insert(vid, -d); order.push(vid); } } for &vid in &order { let d = net[&vid]; - if d != 0 { corr.push((vid, d)); } + if d != 0 { corr.data.push(vid); corr.diffs.push(d); } } corr_ends.push(corr.len()); ds = de; @@ -692,15 +698,14 @@ where (corr, corr_ends) } - fn emit(&mut self, records: &[((u64, u64), T, Diff)]) { + fn emit(&mut self, records: &Records<((u64, u64), T), Vec>) { // Accumulate IDs; resolve columns once at the output boundary. - for rec in records { - let ((kh, vid), t, d) = (rec.0, &rec.1, rec.2); + for (((kh, vid), t), d) in records.data.iter().zip(&records.diffs) { let (krows, vrows, times, diffs) = &mut self.rows; - krows.push(kh); - vrows.push(vid); + krows.push(*kh); + vrows.push(*vid); times.push(t); - diffs.push(d); + diffs.push(*d); } } @@ -726,7 +731,9 @@ mod tests { for _ in 0..depth { col = CValue::Prod(vec![col]); } let input_ids = ids(&col); backend.input.register(col, &input_ids); - let input = [(0, 1), (u64::MAX, -1), (i64::MIN as u64, 0), (i64::MAX as u64, -1), (i64::MIN as u64, -2)]; + let mut input = Records::new(Vec::new()); + input.data = vec![0, u64::MAX, i64::MIN as u64, i64::MAX as u64, i64::MIN as u64]; + input.diffs = vec![1, -1, 0, -1, -2]; let (values, ends) = backend.reduce_brackets(&[0, 3, 5], &input); assert_eq!(values, vec![(u64::MAX, 1), (i64::MIN as u64, 1)]); assert_eq!(ends, vec![0, 1, 2]); @@ -785,7 +792,7 @@ mod tests { fn merge_present_accepts_ordered_compound_keys() { let keys = compound_keys(vec![1, 2], vec![7, 8]); let vals = CValue::u64(vec![10, 20]); - let mut bridge = Vec::new(); + let mut bridge = Records::new(Vec::new()); assert!(merge_present( &keys, &vals, &[1, 2], &[10, 20], &mut [0u64, 0], &[1, 1], &[2], &mut bridge, )); @@ -820,12 +827,13 @@ mod tests { let vids: Vec<_> = rows.iter().map(|r| r.0.1).collect(); let mut times: Vec<_> = rows.iter().map(|r| r.1).collect(); let diffs: Vec<_> = rows.iter().map(|r| r.2).collect(); - let mut bridge = Vec::new(); + let mut bridge = Records::new(Vec::new()); assert!(merge_present(&CValue::u64(khs.clone()), &CValue::u64(vids.clone()), &khs, &vids, &mut times, &diffs, &ends, &mut bridge)); let expected: Vec<_> = expected.into_iter().filter(|(_, d)| *d != 0) .map(|((kv, time), diff)| (kv, time, diff)).collect(); - assert_eq!(bridge, expected, "run count: {count}"); + let actual: Vec<_> = bridge.data.into_iter().zip(bridge.diffs).map(|((kv, t), d)| (kv, t, d)).collect(); + assert_eq!(actual, expected, "run count: {count}"); } } @@ -841,7 +849,7 @@ mod tests { &mut [0u64, 0], &[1, 1], &[2], - &mut Vec::new(), + &mut Records::new(Vec::new()), )); } } From 9db0ea3ebd4e0be2f53c364e6c3228f8722aca9f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 21 Sep 2026 15:20:41 -0400 Subject: [PATCH 3/3] Remove unused consolidation import from Corgi reduce --- interactive/src/corgi/reduce.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index ecda353b6..1b3d1158f 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -28,8 +28,6 @@ use std::collections::HashMap; use std::hash::{BuildHasherDefault, Hasher}; use std::rc::Rc; -#[cfg(test)] -use differential_dataflow::consolidation::consolidate_updates; use differential_dataflow::trace::Description; use differential_dataflow::trace::chunk::ChunkBatch; use differential_dataflow::operators::int_proxy::diffs::{consolidate, Records};