diff --git a/Cargo.toml b/Cargo.toml index ad3af021b..88102a7e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,4 +97,5 @@ debug = true rpath = false lto = true debug-assertions = false -codegen-units = 4 +# One unit, so that A/B benchmarking is maximally accurate. +codegen-units = 1 diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index 75f1bccfa..d955addd9 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -12,13 +12,11 @@ //! the `u64` key hash and whose value is the full `(key, val)` pair. //! //! Per window, the backend presents three proxy bridges — novel input, prior input, and prior output. -//! The identifiers chosen are based on ordinal position. Unfortunately, they are chosen independently -//! for the novel and prior inputs, meaning that empty input collections may be supplied to the logic -//! for evaluation, where the logic should be ignored and zero -> zero enforced. This could be fixed -//! with a more attentive identifier selection. +//! Identifiers are ordinal positions minted by a single [`MergedWalk`] over the novel batches and the +//! prior ones together, so a value carries ONE identifier whichever run it appears in and a novel +//! retraction cancels against its own history. They are minted after the walk consolidates, so a value +//! whose records all cancel never reaches the tactic and never spends an identifier. //! -//! The backend crawls all keys at once, rather than respecting the window setting. -//! This is a defect that should be fixed, but the backend shouldn't be used at scale. //! The backend clones keys and values like it is being paid to waste cycles. //! Generally, this is not a high-performance backend, but shouldn't be abysmal. @@ -27,6 +25,7 @@ use std::rc::Rc; use timely::container::PushInto; use timely::progress::Timestamp; +use timely::progress::frontier::AntichainRef; use crate::consolidation::{consolidate, consolidate_updates}; use crate::difference::Semigroup; @@ -45,14 +44,16 @@ type VBatch = Rc>>; pub struct VecReduceBackend { /// User supplied reduce closure. logic: L, - /// Configuration: keys per window, to size the steps the backend performs. + /// Configuration: records per window, bounding what the three presentations cost at once. window_size: usize, - /// All active keys, either novel input or supplied as externally changed. - /// This is *not* windowed, which is a defect to fix. + /// The current window's active keys, in ascending order: the input pass records them, and the + /// output pass replays them as its own active set. keys_cache: Vec, - /// A state bit indicating that the keys cache should be rebuild (each begin). - keys_stale: bool, + /// The current window's hashes carrying more than one real key, ascending. + collisions: Vec, + /// Per entry of `keys_cache`, the identifier of that key's first input value, if it had one. + reps: Vec>, /// Resolves an input value id (a per-window ordinal) to its `(key, val)` row. in_pool: Vec<(K, V)>, @@ -73,13 +74,14 @@ impl VecReduceBackend { /// A backend deferring value semantics to `logic`, covering the key space in windows. pub fn new(logic: L) -> Self { Self::with_window(logic, 1 << 12) } - /// A backend with an explicit window size, in keys. + /// A backend with an explicit window budget, in presented records. pub fn with_window(logic: L, window_size: usize) -> Self { VecReduceBackend { logic, window_size: window_size.max(1), keys_cache: Vec::new(), - keys_stale: true, + collisions: Vec::new(), + reps: Vec::new(), in_pool: Vec::new(), out_pool: Vec::new(), out_ids: HashMap::new(), @@ -104,7 +106,6 @@ where type ROut = R; fn begin(&mut self, tiles: &[Description]) { - self.keys_stale = true; self.tiles = tiles.to_vec(); self.tile_chunks = (0..tiles.len()).map(|_| Vec::new()).collect(); } @@ -119,79 +120,85 @@ where ) { let Some(start) = *from else { return }; - // If the first window: form a list of all active keys, novel or changed. - // TODO: this is wasteful; the in-order chunk keys could be merged instead. - if self.keys_stale { - self.keys_stale = false; - self.keys_cache.clear(); - for batch in instance.input_batches.iter() { - for chunk in batch.chunks.iter() { - self.keys_cache.extend(chunk.as_slice().iter().map(|r| r.0.0)); - } - } - self.keys_cache.sort_unstable(); - self.keys_cache.dedup(); - if !changed.is_empty() { - let mut merged = Vec::with_capacity(self.keys_cache.len() + changed.len()); - let (mut a, mut b) = (0usize, 0usize); - while a < self.keys_cache.len() || b < changed.len() { - let key = match (self.keys_cache.get(a), changed.get(b)) { - (Some(x), Some(y)) => *x.min(y), - (Some(x), None) => *x, - (None, Some(y)) => *y, - (None, None) => unreachable!("loop condition ensures one is present"), - }; - if self.keys_cache.get(a) == Some(&key) { a += 1; } - if changed.get(b) == Some(&key) { b += 1; } - merged.push(key); + // Populating the novel and prior input bridges. + self.in_pool.clear(); + self.keys_cache.clear(); + self.collisions.clear(); + self.reps.clear(); + let mut walk = MergedWalk::new( + instance.input_batches, + instance.source_batches, + changed, + start, + instance.lower, + ); + let mut drawn: Vec> = Vec::new(); + let mut budget = self.window_size; + while let Some(key) = walk.advance(&mut drawn) { + self.keys_cache.push(key); + budget = budget.saturating_sub(drawn.len()); + let rep = self.in_pool.len() as u64; + let (mut single, mut collides): (Option<&K>, bool) = (None, false); + let mut last: Option<&(K, V)> = None; + for ((data, novel, time), diff) in drawn.drain(..) { + if last != Some(data) { + match single { + None => single = Some(&data.0), + Some(only) => collides |= *only != data.0, + } + self.in_pool.push(data.clone()); + last = Some(data); } - self.keys_cache = merged; + let id = (self.in_pool.len() - 1) as u64; + let bridge = if novel { &mut window.novel } else { &mut window.history }; + bridge.push(((key, id), time, diff)); } + self.reps.push(single.map(|_| rep)); + if collides { self.collisions.push(key); } + if budget == 0 { break; } } + *from = walk.due(); - // Determine the range of active keys to process in this window. - let lo = self.keys_cache.partition_point(|k| *k < start); - if lo == self.keys_cache.len() { - *from = None; - return; - } - let hi = (lo + self.window_size).min(self.keys_cache.len()); - *from = if hi == self.keys_cache.len() { None } else { Some(self.keys_cache[hi]) }; - let keys = &self.keys_cache[lo..hi]; - - // The two input runs, presented apart on ordinal ids from a shared pool. Ids ascend with - // the walk, so each bridge emerges sorted by `((hash, id), time)` with nothing to sort or - // consolidate; a value in both runs gets two ids, reconciled by value in the corrections. - self.in_pool.clear(); - let pool = &mut self.in_pool; - let mut last: Option = None; - merged_run(instance.source_batches, keys, |hash, data, time, diff| { - if last != Some(hash) || pool.last() != Some(data) { - pool.push(data.clone()); - last = Some(hash); - } - window.history.push(((hash, (pool.len() - 1) as u64), time.clone(), diff.clone())); - }); - let mut last: Option = None; - merged_run(instance.input_batches, keys, |hash, data, time, diff| { - if last != Some(hash) || pool.last() != Some(data) { - pool.push(data.clone()); - last = Some(hash); - } - window.novel.push(((hash, (pool.len() - 1) as u64), time.clone(), diff.clone())); - }); - - // The output history, interned into the id namespace corrections mint into. + // Populating the output bridge. self.out_ids.clear(); self.out_pool.clear(); - let (out_pool, out_ids) = (&mut self.out_pool, &mut self.out_ids); - merged_run(instance.output_batches, keys, |hash, data, time, diff| { - let id = *out_ids.entry(data.clone()).or_insert_with(|| { - out_pool.push(data.clone()); - (out_pool.len() - 1) as u64 - }); - window.output.push(((hash, id), time.clone(), diff.clone())); - }); + let mut owalk = MergedWalk::new( + &[], + instance.output_batches, + &self.keys_cache, + start, + instance.lower, + ); + let mut odrawn: Vec> = Vec::new(); + let mut index = 0; + while let Some(key) = owalk.advance(&mut odrawn) { + debug_assert_eq!(self.keys_cache.get(index), Some(&key), "the output pass replays the input pass's keys"); + let (mut single, mut collides): (Option<&K>, bool) = (None, false); + let mut last: Option<&(K, W)> = None; + for ((data, _, time), diff) in odrawn.drain(..) { + if last != Some(data) { + match single { + None => single = Some(&data.0), + Some(only) => collides |= *only != data.0, + } + self.out_ids.insert(data.clone(), self.out_pool.len() as u64); + self.out_pool.push(data.clone()); + last = Some(data); + } + let id = (self.out_pool.len() - 1) as u64; + window.output.push(((key, id), time, 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. + if let (Some(out_key), Some(rep)) = (single, self.reps.get(index).copied().flatten()) { + collides |= self.in_pool[rep as usize].0 != *out_key; + } + if collides { self.collisions.push(key); } + index += 1; + } + // The two passes each contribute in ascending order, but one after the other. + self.collisions.sort_unstable(); + self.collisions.dedup(); } #[inline(never)] @@ -211,36 +218,11 @@ where let mut current: Vec<(W, R)> = Vec::new(); for i in 0..keys.len() { let (ie, oe) = (in_ends[i], out_ends[i]); - // No-collision fast path: when the whole bracket is one real key, the per-key grouping - // below can be skipped. Testing only the bracket's ENDPOINTS would be wrong: neither id - // space is key-ordered across a bracket. Input ids are ordinals minted history-run - // first and then novel, so the order is `history by key, then novel by key`; output ids - // are interned, with corrections appended after the presentation. Either can read - // `[A, B, A]`, whose endpoints agree while its interior does not. So resolve every id - // and require them all to agree — a linear pass over data the fast path is about to - // clone anyway, against the general path's per-key maps. - let single_key: Option = { - let mut only: Option<&K> = None; - let mut agree = true; - for (vid, _) in &input[is..ie] { - let k = &self.in_pool[*vid as usize].0; - match only { - None => only = Some(k), - Some(prev) if prev == k => {} - Some(_) => { agree = false; break } - } - } - if agree { - for (vid, _) in &output[os..oe] { - let k = &self.out_pool[*vid as usize].0; - match only { - None => only = Some(k), - Some(prev) if prev == k => {} - Some(_) => { agree = false; break } - } - } - } - if agree { only.cloned() } else { None } + // 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())) }; if let Some(key) = single_key { input_vals.clear(); @@ -332,110 +314,188 @@ where } } -/// Merge-walks `batches`, restricted to `keys`, invoking `logic` on each consolidated non-zero update. +/// One record drawn by [`MergedWalk::advance`], keyed for consolidation. /// -/// The merge-walk is in order of `(hash, data, time)`. +/// The key is `(data, novel, time)`: grouping by data is what mints one identifier per distinct +/// value, and the `novel` flag separates the two roles *within* a value so that consolidation +/// never cancels a novel update against a prior one. The data is borrowed from the batch — it is +/// cloned once per surviving value, when its identifier is minted, and never per record. +type Drawn<'a, D, T, R> = ((&'a D, bool, T), R); + +/// The length of `slice`'s prefix of records whose hash is `key`, which must be non-empty. /// -/// TODO: Not actually correct at the moment, in that the consolidation does not yet occur. -#[inline(never)] -fn merged_run( - batches: &[VBatch], - keys: &[u64], - mut logic: impl FnMut(u64, &D, &T, &R), -) where +/// Found by doubling rather than by bisecting the whole slice: a key's run within one chunk is +/// usually a handful of records, and a binary search over the thousands that follow it would cost +/// several times the run it is measuring. +fn run_len(slice: &[((u64, D), T, R)], key: u64) -> usize { + debug_assert!(slice.first().is_some_and(|r| r.0.0 == key), "the run must start at `key`"); + let mut step = 1; + while step < slice.len() && slice[step].0.0 == key { step <<= 1; } + let lower = step >> 1; + let upper = step.min(slice.len()); + lower + slice[lower..upper].partition_point(|r| r.0.0 <= key) +} + +/// A resumable merge-walk over a list of batches, delivering one key at a time in `(hash, data, +/// time)` order. +/// +/// The list is logically one, with batches `[0, novel)` the retire's novel input and the rest its +/// prior history. Only the novel batches and `changed` make a key ACTIVE. The prior batches are +/// seeked to whatever key the active set names and are never walked in search of one, which is what +/// keeps a window proportional to the work asked for rather than to the accumulated trace: on an +/// iterative computation, where a retire asks for a scattered handful of keys, the difference is +/// between presenting a window and re-reading the whole trace. +/// +/// The state is a `(chunk, offset)` cursor per batch, an index into `changed`, and the next active +/// hash. Nothing is materialized ahead of the walk, so a caller can stop at any key boundary and +/// resume from the hash left in [`due`](Self::due) — which is exactly the windowing the backend owes +/// [`ProxyReduceBackend::next_window`]. +struct MergedWalk<'a, D: Ord + Clone + 'static, T: Lattice + Timestamp, R: Semigroup + Ord + Clone + 'static> { + /// The novel batches, then the prior ones; [`batch`](Self::batch) reads the pair as one list. + novel: &'a [VBatch], + prior: &'a [VBatch], + /// Externally flagged keys, ascending: active whatever the batches hold. + changed: &'a [u64], + /// Cursor into `changed`. + ci: usize, + /// Per-batch `(chunk, offset)` cursor, in `batch` order. + pos: Vec<(usize, usize)>, + /// The least active hash at or above the cursors, or `None` once the key space is spent. + due: Option, + /// The compaction frontier prior records are advanced to as they are drawn. Advancing before + /// the sort is what lets the consolidation that follows collapse a long-but-quiet history to one + /// record per value: advancement is only useful because something merges the collisions it + /// creates. An empty frontier advances nothing. + frontier: AntichainRef<'a, T>, +} + +impl<'a, D, T, R> MergedWalk<'a, D, T, R> +where D: Ord + Clone + 'static, T: Lattice + Timestamp, R: Semigroup + Ord + Clone + 'static, { - if keys.is_empty() { - return; + /// A walk over `novel ++ prior`, positioned at the first active key at or above `from`. + fn new( + novel: &'a [VBatch], + prior: &'a [VBatch], + changed: &'a [u64], + from: u64, + frontier: AntichainRef<'a, T>, + ) -> Self { + let count = novel.len() + prior.len(); + let mut walk = MergedWalk { + novel, + prior, + changed, + ci: changed.partition_point(|k| *k < from), + pos: vec![(0, 0); count], + due: None, + frontier, + }; + for b in 0..count { + walk.seek_to(b, from); + } + walk.refresh(); + walk } - let n = batches.len(); - let (mut ci, mut oi) = (vec![0usize; n], vec![0usize; n]); - let mut cur: Vec<&[((u64, D), T, R)]> = vec![&[]; n]; - // Seek each batch to the first record with hash at or above the window's first key. - let k0 = keys[0]; - for b in 0..n { - let chunks = &batches[b].chunks; - ci[b] = chunks.partition_point(|c| c.as_slice().last().is_some_and(|r| r.0.0 < k0)); - cur[b] = chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); - oi[b] = cur[b].partition_point(|r| r.0.0 < k0); + + /// The number of batches, novel and prior together. + fn count(&self) -> usize { self.novel.len() + self.prior.len() } + + /// The `b`-th batch of the logical list. + fn batch(&self, b: usize) -> &'a ChunkBatch> { + if b < self.novel.len() { &self.novel[b] } else { &self.prior[b - self.novel.len()] } } - let mut scratch: Vec<(&D, &T, &R)> = Vec::new(); - let mut ki = 0usize; - loop { - // Least hash among the batch heads. - let mut minh: Option = None; - for b in 0..n { - if let Some(r) = cur[b].get(oi[b]) { - if minh.is_none_or(|m| r.0.0 < m) { - minh = Some(r.0.0); - } - } - } - let Some(h) = minh else { break }; - while ki < keys.len() && keys[ki] < h { - ki += 1; - } - if ki >= keys.len() { - break; + + /// The hash the `b`-th batch's cursor sits on, or `None` if it is drained. + fn head(&self, b: usize) -> Option { + let (chunk, offset) = self.pos[b]; + self.batch(b).chunks.get(chunk).and_then(|c| c.as_slice().get(offset)).map(|r| r.0.0) + } + + /// The least active hash at or above the cursors, or `None` once the key space is spent. + /// + /// A key is active if a novel batch holds a record for it or it appears in `changed`. Prior + /// history alone never activates a key: with no novel update and nothing due, its reduction has + /// the same input and the same output as it did last round, so there is nothing to reconcile. + fn due(&self) -> Option { self.due } + + /// Recompute the next active hash from the novel heads and the `changed` cursor. + fn refresh(&mut self) { + let from_novel = (0..self.novel.len()).filter_map(|b| self.head(b)).min(); + let from_changed = self.changed.get(self.ci).copied(); + self.due = [from_novel, from_changed].into_iter().flatten().min(); + } + + /// Seek the `b`-th batch's cursor to the first record with hash at or above `key`. + /// + /// Both steps are binary searches: over the chunk last-keys, which ascend across a batch and so + /// form a sparse index that never touches a record, and then over the landed chunk's records. + fn seek_to(&mut self, b: usize, key: u64) { + // Already positioned, or drained. Keys are drawn in ascending order, so a cursor sitting at + // or above `key` is already at the FIRST record at or above it: everything it passed was at + // or below the previous key. This is the dense case — most keys, most batches — and paying + // two binary searches per batch per key there costs more than the whole walk. + if self.head(b).is_none_or(|h| h >= key) { return; } + let chunks = &self.batch(b).chunks; + let (mut chunk, mut offset) = self.pos[b]; + let skip = chunks[chunk.min(chunks.len())..] + .partition_point(|c| c.as_slice().last().is_some_and(|r| r.0.0 < key)); + if skip > 0 { + chunk += skip; + offset = 0; } - if keys[ki] != h { - // `h` is not wanted. Seeking every batch to the next key that IS wanted costs a binary - // search per batch and lands at or above it, so at most one unwanted hash is visited - // per wanted key. Walking `h`'s records instead would make the whole pass linear in the - // accumulated history rather than in what is asked for — which on an iterative - // computation, where a retire asks for a handful of scattered keys, is the difference - // between presenting the window and re-reading the trace. When the key set is dense - // this branch is simply never taken, so there is no threshold to tune. - let next = keys[ki]; - for b in 0..n { - loop { - let Some(chunk) = batches[b].chunks.get(ci[b]) else { cur[b] = &[]; break }; - let slice = chunk.as_slice(); - if slice.last().is_some_and(|r| r.0.0 < next) { - ci[b] += 1; - oi[b] = 0; - continue; - } - cur[b] = slice; - oi[b] += slice[oi[b]..].partition_point(|r| r.0.0 < next); - // Chunks are non-empty (`ChunkBatch::new` asserts it) and the guard above - // skipped any whose last key is below `next`, so this chunk holds a record at - // or above `next` and the search cannot land at the end. The branch below is - // unreachable; it is kept so a violated invariant degrades to a slower walk - // rather than to a batch that silently reads as drained. - debug_assert!(oi[b] < slice.len(), "a chunk kept by the skip guard must hold a record at or above the sought key"); - if oi[b] >= slice.len() { - ci[b] += 1; - oi[b] = 0; - continue; - } - break; - } - } - continue; + if let Some(slice) = chunks.get(chunk).map(|c| c.as_slice()) { + offset += slice[offset..].partition_point(|r| r.0.0 < key); } - scratch.clear(); - for b in 0..n { - while let Some(r) = cur[b].get(oi[b]) { - if r.0.0 != h { - break; - } - scratch.push((&r.0.1, &r.1, &r.2)); - oi[b] += 1; - if oi[b] >= cur[b].len() { - ci[b] += 1; - oi[b] = 0; - cur[b] = batches[b].chunks.get(ci[b]).map(|c| c.as_slice()).unwrap_or(&[]); - } + self.pos[b] = (chunk, offset); + } + + /// Draw the `b`-th batch's run of records for `key` into `into`, advancing its cursor past them. + /// + /// A key's run can straddle chunks, so this walks chunks until one ends the run; within each it + /// takes the whole run at once rather than a record at a time. + fn take_run(&mut self, b: usize, key: u64, into: &mut Vec>) { + let novel = b < self.novel.len(); + let chunks = &self.batch(b).chunks; + while let Some(slice) = chunks.get(self.pos[b].0).map(|c| c.as_slice()) { + let start = self.pos[b].1; + // The cursor is at or above `key`; if it is above, this batch holds nothing for it. + // Most batches hold nothing for most keys, so this is the common exit and it must cost + // one comparison rather than a search. + if slice.get(start).is_none_or(|r| r.0.0 != key) { return; } + let end = start + run_len(&slice[start..], key); + into.extend(slice[start..end].iter().map(|record| { + let mut time = record.1.clone(); + // Novel times are the interesting-time seeds the tactic's schedule is stated over, + // so they are drawn unadvanced; only the prior history is lifted. + if !novel { time.advance_by(self.frontier); } + ((&record.0.1, novel, time), record.2.clone()) + })); + if end < slice.len() { + self.pos[b].1 = end; + return; } + self.pos[b] = (self.pos[b].0 + 1, 0); } - { - scratch.sort_by(|a, b| (a.0, a.1).cmp(&(b.0, b.1))); - for (d, t, r) in scratch.drain(..) { - logic(h, d, t, r); - } + } + + /// Draw every record of the next active key, and advance to the one after it. + /// + /// Returns the key drawn, or `None` once the key space is spent. `into` is left sorted by + /// `(data, novel, time)` and consolidated, so a value whose records all cancel is gone before + /// the caller ever sees it — and so never spends an identifier. + fn advance(&mut self, into: &mut Vec>) -> Option { + let key = self.due?; + into.clear(); + for b in 0..self.count() { + self.seek_to(b, key); + self.take_run(b, key, into); } + consolidate(into); + if self.changed.get(self.ci) == Some(&key) { self.ci += 1; } + self.refresh(); + Some(key) } } diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index 37836c677..d4e8825e9 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -289,6 +289,83 @@ fn reduce_collision_across_retires() { assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 9), 1, 1)], "retire 2 must not disturb C2"); } +/// Colliding hashes inside an `iterate`, where a correction can be deferred past the retire's +/// upper bound and applied in a later one. +/// +/// The hash is `k % 2` rather than a real hash, so every bucket holds several real keys throughout. +/// A round's reduction retracts the values it replaces, so a real key's input can cancel in one +/// retire while the output it no longer justifies is corrected in another — the window where a +/// hash's input mentions one real key and its output another. Under `Product` times a synthesized +/// time can land at or beyond `upper`, which is what defers the correction to produce that window. +#[test] +fn reduce_collision_inside_iterate() { + let out = Arc::new(Mutex::new(Vec::<((u64, u64), u64, i64)>::new())); + let os = out.clone(); + // Key `k` holds `{10k, 10k + 7, 10k + 3}`, so its maximum is `10k + 7`; a key borrowing from + // the neighbour it shares a bucket with would land on a different, visibly wrong maximum. + let updates: Vec<((u64, u64), u64, i64)> = (0..6u64) + .flat_map(|k| [((k, 10 * k), 0u64, 1i64), ((k, 10 * k + 7), 0, 1), ((k, 10 * k + 3), 0, 1)]) + .collect(); + + timely::execute_directly(move |worker| { + worker.dataflow::(|scope| { + let input = updates.clone().to_stream(scope).as_collection(); + let result = input.iterate(|_scope, inner| { + let hashed = inner.map(|(k, v)| (k % 2, (k, v))); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, VChunkBatcher, i64>, VChunkBuilder, i64>, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide"); + reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "CollideReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) + .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) + }); + result.inspect(move |(d, t, r)| os.lock().unwrap().push((*d, *t, *r))); + }); + }); + + let mut got = out.lock().unwrap().clone(); + got.sort(); + let want: Vec<_> = (0..6u64).map(|k| ((k, 10 * k + 7), 0u64, 1i64)).collect(); + assert_eq!(got, want, "each real key reaches its own maximum"); +} + +/// A key landing in one of two hash buckets, so several real keys share each hash and several +/// hashes share a window. `Collide` puts everything in one bucket, which cannot show that the +/// backend rebuilds its collision bookkeeping per window rather than carrying it between them. +#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)] +struct Bucket(u64); +impl std::hash::Hash for Bucket { + fn hash(&self, state: &mut H) { state.write_u64(self.0 % 2) } +} + +/// Two colliding hashes, each holding several real keys, presented across several windows. +/// +/// The window budget of one record forces a window per hash, so the collision set is built, used +/// and discarded twice within the retire. A set that leaked between windows, or a representative +/// misaligned with the window's key list, merges two real keys' values into one reduction — which +/// shows up here as a key taking a neighbour's maximum. +#[test] +fn reduce_collision_multiwindow() { + let logic = |_k: &Bucket, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>| { + if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| *v).max() { + updates.push((m, 1)); + } + for (w, d) in current.iter() { updates.push((*w, -*d)); } + }; + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::with_window(logic, 1)); + // Key `k` carries values `10k` and `10k + 1`, so its maximum is `10k + 1` and borrowing from a + // neighbour would be visible. Keys 0, 2, 4 share one hash; keys 1, 3 share the other. + let rows: Vec<((Bucket, u64), u64, i64)> = (0..5u64) + .flat_map(|k| [((Bucket(k), 10 * k), 0u64, 1i64), ((Bucket(k), 10 * k + 1), 0, 1)]) + .collect(); + let input = hbatch::(rows, 0, 1); + let (produced, _f) = tactic.retire( + vec![], vec![], vec![input], + &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), + ); + let mut out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + out.sort(); + let want: Vec<_> = (0..5u64).map(|k| ((Bucket(k), 10 * k + 1), 0u64, 1i64)).collect(); + assert_eq!(out, want, "each real key keeps its own maximum"); +} + /// A reduction that emits only for `Collide(1)`, so `Collide(2)` has input but never any output. fn only_first(k: &Collide, input: &[(u64, i64)], current: &mut Vec<(u64, i64)>, updates: &mut Vec<(u64, i64)>) { if k.0 == 1 { diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs index 6872c0ed4..ff95586a1 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -307,14 +307,14 @@ fn run_flat(mode: Mode, keys: u64, sub: u64) -> f64 { }) } -const PROP_NODES: u64 = 2_000; -const PROP_EDGES: usize = 4_000; -const PROP_CHURN: usize = 50; -const PROP_ROUNDS: u64 = 40; -const PROP_WARMUP: u64 = 5; +fn prop_nodes() -> u64 { sized("PROP_NODES", 2_000) } +fn prop_edges() -> usize { sized("PROP_EDGES", 4_000) as usize } +fn prop_churn() -> usize { sized("PROP_CHURN", 50) as usize } +fn prop_rounds() -> u64 { sized("PROP_ROUNDS", 40) } +fn prop_warmup() -> u64 { sized("PROP_WARMUP", 5) } fn edge(i: usize) -> (u64, u64) { - let n = PROP_NODES; + let n = prop_nodes(); ((i as u64).wrapping_mul(2654435761) % n, (i as u64).wrapping_mul(40503).wrapping_add(7) % n) } @@ -322,6 +322,10 @@ fn edge(i: usize) -> (u64, u64) { /// carried-interesting-times shape. Returns averaged microseconds per round and a checksum of the /// produced label updates, which must agree across modes. fn run_propagate(mode: Mode) -> (f64, i64) { + assert!( + prop_rounds() as usize * prop_churn() <= prop_edges(), + "the churn retracts edge `r * PROP_CHURN + c`, which must be one of the PROP_EDGES inserted", + ); let check = Arc::new(Mutex::new(0i64)); let sum = check.clone(); let avg = timely::execute_directly(move |worker| { @@ -364,10 +368,10 @@ fn run_propagate(mode: Mode) -> (f64, i64) { }); nodes.advance_to(0); edges.advance_to(0); - for k in 0..PROP_NODES { + for k in 0..prop_nodes() { nodes.insert((k, k)); } - for i in 0..PROP_EDGES { + for i in 0..prop_edges() { let (s, d) = edge(i); edges.insert((s, d)); } @@ -379,7 +383,7 @@ fn run_propagate(mode: Mode) -> (f64, i64) { worker.step(); } let mut times = Vec::new(); - for r in 0..PROP_ROUNDS { + for r in 0..prop_rounds() { let t = r + 1; // Perturb a low-numbered label each round: low labels spread widely under `min`, so // withdrawing one forces its reach to relabel — real iterate work, with interesting @@ -390,11 +394,11 @@ fn run_propagate(mode: Mode) -> (f64, i64) { nodes.insert((p, p)); } nodes.remove((m, m)); - for c in 0..PROP_CHURN { - let dead = (r as usize) * PROP_CHURN + c; + for c in 0..prop_churn() { + let dead = (r as usize) * prop_churn() + c; let (s, d) = edge(dead); edges.remove((s, d)); - let (s, d) = edge(PROP_EDGES + dead); + let (s, d) = edge(prop_edges() + dead); edges.insert((s, d)); } nodes.advance_to(t + 1); @@ -405,7 +409,7 @@ fn run_propagate(mode: Mode) -> (f64, i64) { while probe.less_than(&(t + 1)) { worker.step(); } - if r >= PROP_WARMUP { + if r >= prop_warmup() { times.push(start.elapsed().as_micros()); } }