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..5e2749eb0 --- /dev/null +++ b/differential-dataflow/src/operators/int_proxy/diffs.rs @@ -0,0 +1,180 @@ +//! A prototype of collective difference operations. +//! +//! 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. + +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]); + + /// 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); +} + +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 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)); + } +} + +/// 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]))); + } +} + +/// 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, +} + +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); + } +} + +/// 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 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() } + } + /// 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 81a98c9ba..79185b8d4 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 ordered 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<(V, T), C>, + history: Vec<(T, T, usize)>, // (time, suffix meet, row) + pub buffer: Records<(V, T), C>, + scratch: Consolidation<(V, 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<((K, V), 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 15dbf0fe5..1546760af 100644 --- a/differential-dataflow/src/operators/int_proxy/mod.rs +++ b/differential-dataflow/src/operators/int_proxy/mod.rs @@ -1,7 +1,9 @@ //! Backend-agnostic operator tactics using ordered, copyable proxies. //! //! The tactics support custom operator implementations without rebuilding the non-trivial and often non-obvious time-based logic that supports them. -//! They run DD's operator logic over consolidated `[((key, val), time, diff)]` lists. +//! They run DD's operator logic over consolidated updates with `(key, val)` proxies. +//! Reduce keeps `((key, val), time)` metadata beside a [`diffs::DiffContainer`]. +//! Join still exchanges scalar-difference [`ProxyBridge`] lists. //! Keys identify independent groups; values identify data within a group. //! Both require only `Copy + Ord` and need not be the types presented to user logic. //! The tactics elicit proxies, perform time and difference computations, and return proxies to the backend for interpretation. @@ -44,6 +46,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; diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs index 9107ff350..6c6abf4fe 100644 --- a/differential-dataflow/src/operators/int_proxy/reduce.rs +++ b/differential-dataflow/src/operators/int_proxy/reduce.rs @@ -8,12 +8,12 @@ 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::{KeyPosition, ProxyBridge}; +use super::history::DiffHistory; +use super::KeyPosition; 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 +40,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, value_id), time)`. /// May be advanced to the compaction frontier. - pub input: ProxyBridge, + pub input: Records<((K, VIn), T), RIn>, /// The RAW novel time support: `(key, time)` pairs sorted by `(key, time)` and deduplicated. /// Record these from the novel batches BEFORE any consolidation or advancement. /// A netted-away record's time must still appear here. pub seeds: Vec<(K, T)>, /// Accumulated output preceding the retire's interval, same ordering as `input`. - pub output: ProxyBridge, + pub output: Records<((K, VOut), 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,7 +66,7 @@ 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 { /// Independent groups, with identities stable across retires. @@ -74,10 +75,10 @@ pub trait ProxyReduceBackend { type VIn: Copy + Ord; /// Output value proxies, including newly produced values, valid throughout a window. type VOut: Copy + Ord; - /// 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)`. /// @@ -86,6 +87,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 `Start` or an inclusive `At(key)` lower bound on keys still to be covered. @@ -120,13 +125,13 @@ pub trait ProxyReduceBackend { &mut self, keys: &[Self::Key], in_ends: &[usize], - input: &[(Self::VIn, Self::RIn)], + input: &Records, out_ends: &[usize], - output: &[(Self::VOut, Self::ROut)], - ) -> (Vec<(Self::VOut, Self::ROut)>, Vec); + output: &Records, + ) -> (Records, Vec); /// Commit a collection of updates to the batch in progress. - fn emit(&mut self, records: &[((Self::Key, Self::VOut), T, Self::ROut)]); + fn emit(&mut self, records: &Records<((Self::Key, Self::VOut), T), Self::ROut>); /// Complete the session matching `begin`, yielding the batch it described, /// or `None` when the span it described carries no updates. @@ -218,32 +223,39 @@ where // Start before every key; each window advances to an inclusive key bound or the end. let mut from = KeyPosition::Start; - 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<((K, Bk::VOut), 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<(Bk::VIn, 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<(Bk::VOut, 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<(Bk::VIn, Bk::RIn)> = Vec::new(); - let mut cur_out: Vec<(Bk::VOut, 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 != KeyPosition::End { 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, time) and deduplicated", @@ -257,7 +269,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 <= KeyPosition::At(k) && KeyPosition::At(k) < from) }, "next_window must report a key entirely within the window that first mentions it", @@ -283,21 +295,21 @@ where live.clear(); // Mapped to keys 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(key)); } + if n_slots == slots.len() { slots.push(KeySweep::empty(key, &p_in.diffs, &p_out.diffs)); } let slot = &mut slots[n_slots]; slot.key = key; slot.pended.clear(); @@ -311,8 +323,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 } @@ -321,8 +333,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) }; @@ -350,21 +362,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)); } @@ -376,11 +388,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; @@ -403,8 +413,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); } } @@ -426,9 +436,9 @@ struct KeySweep { at: Option, } -impl KeySweep { - fn empty(key: K) -> Self { - KeySweep { key, sweep: Sweep::new(), direct: None, pended: Vec::new(), at: None } +impl KeySweep { + fn empty(key: K, input: &RIn, output: &ROut) -> Self { + KeySweep { key, sweep: Sweep::new(input, output), direct: None, pended: Vec::new(), at: None } } } @@ -469,8 +479,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 @@ -488,7 +498,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<((VOut, T), ROut)>, + produced: Records<(VOut, T), ROut>, + produced_scratch: Consolidation<(VOut, 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. @@ -507,14 +518,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, } } @@ -526,12 +538,14 @@ impl( &mut self, owed: impl Iterator, novel_times: impl Iterator, - input: impl Iterator, - output: impl Iterator, + input: &Records<((K, VIn), T), RIn>, + in_rows: std::ops::Range, + output: &Records<((K, VOut), T), ROut>, + out_rows: std::ops::Range, ) { // Merge the two ascending seed sources, deduplicated. self.seeds.clear(); @@ -566,8 +580,8 @@ impl = 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; } @@ -679,11 +693,11 @@ impl) { - 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<(VOut, 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); } } @@ -740,3 +751,82 @@ impl, 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 dc8d08a34..79cbf25cd 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::{KeyPosition, ProxyReduceBackend, ReduceInstance, ReduceWindow}; /// The batch type of a hash-keyed [`ChunkSpine`](crate::trace::chunk::vec::ChunkSpine): updates @@ -103,8 +104,10 @@ where type Key = u64; type VIn = u64; type VOut = u64; - 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(); @@ -116,7 +119,7 @@ where instance: &ReduceInstance<'_, T, VBatch<(K, V), T, R>, VBatch<(K, W), T, R>>, changed: &[u64], from: &mut KeyPosition, - window: &mut ReduceWindow, + window: &mut ReduceWindow, Vec>, ) { let start = match *from { KeyPosition::Start => 0, KeyPosition::At(key) => key, KeyPosition::End => return }; @@ -160,16 +163,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(); @@ -208,7 +213,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. @@ -228,11 +234,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(); @@ -243,15 +249,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); @@ -262,16 +268,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())); } @@ -292,7 +299,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); } } } @@ -304,9 +312,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 893202f64..49f7da32f 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -28,10 +28,10 @@ use std::collections::HashMap; use std::hash::{BuildHasherDefault, Hasher}; use std::rc::Rc; -use differential_dataflow::consolidation::consolidate_updates; use differential_dataflow::trace::Description; use differential_dataflow::trace::chunk::ChunkBatch; -use differential_dataflow::operators::int_proxy::{KeyPosition, ProxyBridge}; +use differential_dataflow::operators::int_proxy::diffs::{consolidate, Records}; +use differential_dataflow::operators::int_proxy::KeyPosition; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; use corgi::arrange::{compare_at, gather, gather_lanes, sort_blocks}; @@ -289,7 +289,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 +326,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 +338,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 +363,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 +390,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 +401,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 +437,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 +461,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 +490,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 +526,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())); @@ -576,8 +578,10 @@ where type Key = u64; type VIn = u64; type VOut = u64; - 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. @@ -587,7 +591,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 KeyPosition, window: &mut ReduceWindow) { + fn next_window(&mut self, instance: &ReduceInstance<'_, T, CBatch, CBatch>, changed: &[u64], from: &mut KeyPosition, 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 @@ -658,16 +662,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. @@ -681,12 +686,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; @@ -695,15 +700,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); } } @@ -729,7 +733,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]); @@ -788,7 +794,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, )); @@ -823,12 +829,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}"); } } @@ -844,7 +851,7 @@ mod tests { &mut [0u64, 0], &[1, 1], &[2], - &mut Vec::new(), + &mut Records::new(Vec::new()), )); } }