diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 3520eccbc..8b3f22682 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -27,6 +27,7 @@ use std::collections::BinaryHeap; use std::collections::HashMap; use std::hash::{BuildHasherDefault, Hasher}; use std::rc::Rc; +use std::sync::LazyLock; use differential_dataflow::consolidation::consolidate_updates; use differential_dataflow::trace::Description; @@ -171,6 +172,21 @@ impl CorgiReduceBackend { } } + /// Whether this reducer's output values are always unit. + fn unit_output(&self) -> bool { + match self.reducer { + Reducer::Distinct => true, + Reducer::Count | Reducer::Min | Reducer::Collect => false, + } + } + + /// Whether the current reduction path resolves input IDs to payload rows. + fn resolves_input_payloads(&self) -> bool { + // Count only consumes diffs, but still uses the general row-resolution path. + // A diff-only Count path can opt out here without changing presentation. + !self.unit_output() + } + /// Clear the resolution pools at the start of a retire (called from `next_window`'s first call). fn reset_pools(&mut self) { self.key_index.clear(); @@ -198,6 +214,14 @@ impl CorgiReduceBackend { self.val_len += col.len(); self.val_blocks.push(col); } + + /// Register presented values, using just the first representative for a unit column. + fn register_vals_ids(&mut self, col: CValue, ids: &PresentedIds) { + match ids { + PresentedIds::Unit(id) => self.register_vals(col, std::slice::from_ref(id)), + PresentedIds::Rows(ids) => self.register_vals(col, ids), + } + } } /// Concatenate corgi columns (skipping empties, which contribute no rows and so don't shift the @@ -244,6 +268,33 @@ fn ids(col: &CValue) -> Vec { corgi::hash(col) } +// Use Corgi's canonical ID at every unit presentation and correction site. +static UNIT_ID: LazyLock = LazyLock::new(|| corgi::hash(&CValue::Unit(1))[0]); + +/// Unit columns need one ID, independent of the number of presented rows. +/// Inspect each column anew; an empty presentation never fixes a future shape. +enum PresentedIds { + Unit(u64), + Rows(Vec), +} +impl PresentedIds { + fn new(col: &CValue) -> Self { + match col { + CValue::Unit(_) => Self::Unit(*UNIT_ID), + _ => Self::Rows(ids(col)), + } + } +} +impl std::ops::Index for PresentedIds { + type Output = u64; + fn index(&self, row: usize) -> &u64 { + match self { + Self::Unit(id) => id, + Self::Rows(ids) => &ids[row], + } + } +} + /// Concatenate the records of the `changed` keys across a run of chunks into parallel /// `(keys_col, vals_col)` corgi columns plus per-record `(key_hash, time, diff)`. `changed` is the /// ASCENDING set of changed key ids; a row is kept iff its key id is in it. @@ -286,15 +337,15 @@ where (keys_col, vals_col, khs, times, diffs, run_ends) } -/// Merge already-ordered selected chunk runs directly into an empty proxy bridge. Leaf values -/// preserve value-id order. Keys may either be identity-id leaves or carried-hash columns, provided +/// Merge already-ordered selected chunk runs directly into an empty proxy bridge. Leaf and unit +/// values preserve value-id order. Keys may either be identity-id leaves or carried-hash columns, provided /// no one chunk run contains two real keys under the same hash; in the latter case the real-key /// tie-break would interrupt proxy `(key_id, value_id, time)` order, so we fall back to ordinary /// consolidation. A debug assertion audits the inferred order. Returns false when the inference /// does not hold or the bridge is nonempty. fn merge_present( keys_col: &CValue, vals_col: &CValue, - khs: &[u64], vids: &[u64], times: &[T], diffs: &[Diff], run_ends: &[usize], + khs: &[u64], vids: &PresentedIds, times: &[T], diffs: &[Diff], run_ends: &[usize], bridge: &mut ProxyBridge, ) -> bool { let ordered_keys = corgi::arrange::leaf_slice(keys_col).is_some() || { @@ -308,17 +359,30 @@ fn merge_present( one_real_key_per_id }) && start == khs.len() }; - let ordered_ids = ordered_keys && corgi::arrange::leaf_slice(vals_col).is_some(); + let ordered_ids = ordered_keys + && (matches!(vals_col, CValue::Unit(_)) || corgi::arrange::leaf_slice(vals_col).is_some()); if !ordered_ids || !bridge.is_empty() { return false; } + // Select once per presentation. The merge loop sees either a constant ID or a + // slice lookup, without a unit-versus-row choice at each element. + match vids { + PresentedIds::Unit(id) => merge_present_ids(khs, |_| *id, times, diffs, run_ends, bridge), + PresentedIds::Rows(ids) => merge_present_ids(khs, |row| ids[row], times, diffs, run_ends, bridge), + } +} + +fn merge_present_ids( + khs: &[u64], vids: impl Fn(usize) -> u64, times: &[T], diffs: &[Diff], run_ends: &[usize], + bridge: &mut ProxyBridge, +) -> bool { debug_assert!({ let mut start = 0usize; let sorted = run_ends.iter().all(|&end| { let sorted = (start + 1..end).all(|i| { - (khs[i - 1], vids[i - 1], ×[i - 1]) - <= (khs[i], vids[i], ×[i]) + (khs[i - 1], vids(i - 1), ×[i - 1]) + <= (khs[i], vids(i), ×[i]) }); start = end; sorted @@ -340,7 +404,7 @@ fn merge_present( if run_ends.len() == 1 { for index in 0..run_ends[0] { - accumulate((khs[index], vids[index]), ×[index], diffs[index]); + accumulate((khs[index], vids(index)), ×[index], diffs[index]); } drop(accumulate); if let Some(record) = current { @@ -352,7 +416,7 @@ fn merge_present( let mut heap: BinaryHeap> = BinaryHeap::new(); let mut lo = 0usize; for (run, &hi) in run_ends.iter().enumerate() { - heap.push(Reverse(((khs[lo], vids[lo]), ×[lo], run, lo))); + heap.push(Reverse(((khs[lo], vids(lo)), ×[lo], run, lo))); lo = hi; } while let Some(mut head) = heap.peek_mut() { @@ -361,7 +425,7 @@ fn merge_present( let end = run_ends[run]; if index + 1 < end { let next = index + 1; - *head = Reverse(((khs[next], vids[next]), ×[next], run, next)); + *head = Reverse(((khs[next], vids(next)), ×[next], run, next)); } else { std::collections::binary_heap::PeekMut::pop(head); } @@ -403,11 +467,19 @@ where if khs.is_empty() { return; } - let vids = ids(&p_vals); + let vids = PresentedIds::new(&p_vals); let merged = merge_present(&p_keys, &p_vals, &khs, &vids, ×, &diffs, &run_ends, bridge); - for (row, &vid) in vids.iter().enumerate() { self.in_index.entry(vid).or_insert(*len + row); } - *len += p_vals.len(); - blocks.push(p_vals); + // Distinct needs value identity in the sweep, but never resolves input payloads. + if self.resolves_input_payloads() { + match &vids { + PresentedIds::Unit(id) => { self.in_index.entry(*id).or_insert(*len); } + PresentedIds::Rows(ids) => { + for (row, &id) in ids.iter().enumerate() { self.in_index.entry(id).or_insert(*len + row); } + } + } + *len += p_vals.len(); + blocks.push(p_vals); + } self.register_keys(p_keys, &khs); if !merged { bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i]))); @@ -445,29 +517,7 @@ where out_ids = ids(&col); self.register_vals(col, &out_ids); } - Reducer::Distinct => { - // Present iff any value has NON-ZERO net -- the sign does not matter. DD's `reduce` - // presents every value whose accumulation is non-zero, negatives included, and - // `backend::vec`'s Distinct then emits `1` without looking at the diffs at all. A - // `> 0` test here silently drops a key whose values all accumulate negative, which - // is exactly what a negated collection produces. Output value is unit (a `Unit` column). - let mut present = 0usize; - let mut start = 0; - for &end in ends { - if input[start..end].iter().any(|&(_, d)| d != 0) { - present += 1; - out_diffs.push(1); - } - out_ends.push(out_diffs.len()); - start = end; - } - if present == 0 { - return (Vec::new(), out_ends); - } - let col = CValue::Unit(present); - out_ids = ids(&col); // all equal (unit content hash) - self.register_vals(col, &out_ids); - } + Reducer::Distinct => unreachable!("unit_output() reducers are handled directly in reduce_corrections"), Reducer::Min => { // The structural minimum over values with NON-ZERO net. The sign does not select // candidates: DD presents every non-zero accumulation. Filtering to `> 0` here both @@ -649,10 +699,12 @@ where // Output-history presentation, same keys (register keys + values for correction resolution). let (o_keys, o_vals, o_khs, o_times, o_diffs, o_run_ends) = collect_present(&chunks_of(instance.output_batches), &keys); if !o_khs.is_empty() { - let vids = ids(&o_vals); + let vids = PresentedIds::new(&o_vals); let merged = merge_present(&o_keys, &o_vals, &o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output); self.register_keys(o_keys, &o_khs); - self.register_vals(o_vals, &vids); + if !self.unit_output() { + self.register_vals_ids(o_vals, &vids); + } if !merged { window.output.extend((0..o_khs.len()).map(|i| ((o_khs[i], vids[i]), o_times[i].clone(), o_diffs[i]))); consolidate_updates(&mut window.output); @@ -661,6 +713,29 @@ where } fn reduce_corrections(&mut self, keys: &[u64], in_ends: &[usize], input: &[(u64, Diff)], out_ends: &[usize], output: &[(u64, Diff)]) -> (Vec<(u64, Diff)>, Vec) { + debug_assert_eq!(in_ends.len(), keys.len()); + debug_assert_eq!(out_ends.len(), keys.len()); + if self.unit_output() { + // Distinct emits unit for any non-empty accumulation. DD reduce includes + // values with negative accumulated diffs: a negated collection is all-negative + // but must still be present, so testing > 0 would be wrong. Input identities + // remain separate in the sweep: a:+1, b:-1 is also present despite its zero sum. + let unit_id = *UNIT_ID; + let mut corr = Vec::new(); + let mut ends = Vec::with_capacity(keys.len()); + let (mut is, mut os) = (0, 0); + for (&ie, &oe) in in_ends.iter().zip(out_ends) { + let desired = Diff::from(input[is..ie].iter().any(|&(_, d)| d != 0)); + let current: Diff = output[os..oe].iter().map(|&(_, d)| d).sum(); + let diff = desired - current; + if diff != 0 { corr.push((unit_id, diff)); } + ends.push(corr.len()); + is = ie; + os = oe; + } + return (corr, ends); + } + // Resolve input value_ids to `in_vals` rows, reduce (desired output), then difference the // desired against the presented current output per key: correction = desired − current. let in_rows: Vec<(usize, Diff)> = input.iter() @@ -694,14 +769,17 @@ where } fn emit(&mut self, records: &[((u64, u64), T, Diff)]) { - // Resolve each correction's key/value proxies to pool rows and accumulate. + // Unit output has no value rows to resolve or retain. + if !self.unit_output() { + self.rows.1.extend(records.iter().map(|rec| { + *self.val_index.get(&rec.0.1).expect("value resolvable this retire") + })); + } for rec in records { - let ((kh, vid), t, d) = (rec.0, &rec.1, rec.2); + let ((kh, _), t, d) = (rec.0, &rec.1, rec.2); let kr = *self.key_index.get(&kh).expect("key resolvable this retire"); - let vr = *self.val_index.get(&vid).expect("value resolvable this retire"); - let (krows, vrows, times, diffs) = &mut self.rows; + let (krows, _, times, diffs) = &mut self.rows; krows.push(kr); - vrows.push(vr); times.push(t.clone()); diffs.push(d); } @@ -710,11 +788,14 @@ where fn finish(&mut self) -> Option> { // Seal the batch: gather the accumulated (key, val) pool rows into columns, one CorgiChunk batch. let key_pool = concat_columns(&self.key_blocks); - let val_pool = concat_columns(&self.val_blocks); let (krows, vrows, times, diffs) = std::mem::take(&mut self.rows); if times.is_empty() { return None; } let keys = gather(&key_pool, &krows); - let vals = gather(&val_pool, &vrows); + let vals = if self.unit_output() { + CValue::Unit(times.len()) + } else { + gather(&concat_columns(&self.val_blocks), &vrows) + }; Some(Rc::new(columns_to_batch(keys, vals, times, diffs))) } } @@ -754,7 +835,7 @@ mod tests { let vals = CValue::u64(vec![10, 20]); let mut bridge = Vec::new(); assert!(merge_present( - &keys, &vals, &[1, 2], &[10, 20], &[0u64, 0], &[1, 1], &[2], &mut bridge, + &keys, &vals, &[1, 2], &PresentedIds::new(&vals), &[0u64, 0], &[1, 1], &[2], &mut bridge, )); assert_eq!(bridge.len(), 2); } @@ -773,42 +854,56 @@ mod tests { ((4, 40), Product::new(0, 0), -1)], vec![((1, 10), Product::new(1, 0), 2)], ]; - for count in 1..=runs.len() { - let (mut rows, mut ends) = (Vec::new(), Vec::new()); - let mut expected = BTreeMap::new(); - for run in &runs[..count] { - rows.extend_from_slice(run); - ends.push(rows.len()); - for &(kv, time, diff) in run { - *expected.entry((kv, time)).or_insert(0) += diff; + for unit in [false, true] { + let runs = runs.clone().map(|mut run| { + if unit { + for row in &mut run { row.0.1 = *UNIT_ID; } + run.sort_unstable(); } + run + }); + for count in 1..=runs.len() { + let (mut rows, mut ends) = (Vec::new(), Vec::new()); + let mut expected = BTreeMap::new(); + for run in &runs[..count] { + rows.extend_from_slice(run); + ends.push(rows.len()); + for &(kv, time, diff) in run { + *expected.entry((kv, time)).or_insert(0) += diff; + } + } + let khs: Vec<_> = rows.iter().map(|r| r.0.0).collect(); + let vids: Vec<_> = rows.iter().map(|r| r.0.1).collect(); + let 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 vals = if unit { CValue::Unit(vids.len()) } else { CValue::u64(vids) }; + assert!(merge_present(&CValue::u64(khs.clone()), &vals, + &khs, &PresentedIds::new(&vals), ×, &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}, unit: {unit}"); } - let khs: Vec<_> = rows.iter().map(|r| r.0.0).collect(); - let vids: Vec<_> = rows.iter().map(|r| r.0.1).collect(); - let times: Vec<_> = rows.iter().map(|r| r.1).collect(); - let diffs: Vec<_> = rows.iter().map(|r| r.2).collect(); - let mut bridge = Vec::new(); - assert!(merge_present(&CValue::u64(khs.clone()), &CValue::u64(vids.clone()), - &khs, &vids, ×, &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}"); } } #[test] fn merge_present_rejects_a_compound_hash_collision_within_a_run() { let keys = compound_keys(vec![1, 1], vec![7, 8]); - let vals = CValue::u64(vec![10, 20]); - assert!(!merge_present( - &keys, - &vals, - &[1, 1], - &[10, 20], - &[0u64, 0], - &[1, 1], - &[2], - &mut Vec::new(), - )); + for vals in [CValue::u64(vec![10, 20]), CValue::Unit(2)] { + assert!(!merge_present( + &keys, + &vals, + &[1, 1], + &PresentedIds::new(&vals), + &[0u64, 0], + &[1, 1], + &[2], + &mut Vec::new(), + )); + } } } + +#[cfg(test)] +mod unit_value_tests; diff --git a/interactive/src/corgi/reduce/unit_value_tests.rs b/interactive/src/corgi/reduce/unit_value_tests.rs new file mode 100644 index 000000000..d12313e1e --- /dev/null +++ b/interactive/src/corgi/reduce/unit_value_tests.rs @@ -0,0 +1,112 @@ +//! Unit-value diversion through the existing tactic, checked against row-level answers. +use super::*; +use crate::corgi::chunk::{present_key, recover_key}; +use crate::corgi::logic::{shape_of_row, transcode, untranscode}; +use crate::ir::Value; +use differential_dataflow::operators::int_proxy::reduce::ProxyReduceTactic; +use differential_dataflow::operators::reduce::ReduceTactic; +use timely::order::Product; +use timely::progress::Antichain; +use timely::PartialOrder; + +#[test] +fn distinct_with_unit_and_nonunit_values_matches_partial_order_oracle() { + type T = Product; + let t = T::new; + // Incomparable updates require a correction at their join. Some later novel + // records cancel source records at the same time; their raw seeds must survive. + let rounds = [ + vec![], // No value shape can be inferred from an empty first retire. + vec![(7, 10, t(0, 2), 1), (7, 10, t(2, 0), 1), + (8, 10, t(0, 0), -3), (8, 11, t(0, 0), 3)], + vec![(7, 10, t(0, 2), -1), (8, 10, t(1, 1), 3), (8, 11, t(1, 1), -3)], + vec![(7, 10, t(3, 1), -1), (7, 11, t(3, 3), 2)], + vec![(7, 11, t(4, 4), -2)], + ]; + for key_shape in 0..3 { + let key = |k| match key_shape { + 0 => Value::Int(k), + 1 => Value::Tuple(vec![Value::Int(k), Value::Int(0)]), + _ => Value::List(vec![Value::Int(k), Value::Int(0)]), + }; + let shape = shape_of_row(&key(7)).unwrap(); + for unit in [false, true] { + let mut tactic = ProxyReduceTactic::new(CorgiReduceBackend::new(Reducer::Distinct)); + let (mut source, mut output, mut original) = (Vec::new(), Vec::new(), Vec::new()); + for novel in &rounds { + let input = Rc::new(columns_to_batch( + present_key(transcode(&novel.iter().map(|r| key(r.0)).collect::>(), &shape)), + if unit { CValue::Unit(novel.len()) } + else { CValue::u64(novel.iter().map(|r| r.1).collect()) }, + novel.iter().map(|r| r.2).collect(), + novel.iter().map(|r| r.3).collect(), + )); + let lower = Antichain::from_elem(t(0, 0)); + let upper = Antichain::from_elem(t(2, 2)); + let (result, pending) = tactic.retire(source.clone(), output.clone(), vec![input.clone()], &lower, &upper, &lower); + if let Some(batch) = result.and_then(|s| s.inner) { output.push(batch); } + source.push(input); + original.extend_from_slice(novel); + // Drain pending joins with no novel input. + let (result, pending) = tactic.retire(source.clone(), output.clone(), vec![], &upper, &Antichain::new(), &pending); + assert!(pending.is_empty()); + if let Some(batch) = result.and_then(|s| s.inner) { output.push(batch); } + let mut actual = Vec::new(); + for chunk in chunks_of(&output) { + assert!(matches!(chunk.vals(), CValue::Unit(_))); + for (i, key) in untranscode(recover_key(chunk.keys()), &shape).into_iter().enumerate() { + actual.push((key, chunk.times().get(i), chunk.diffs()[i])); + } + } + for x in 0..=5 { + for y in 0..=5 { + for k in [7, 8] { + let time = t(x, y); + let mut counts = std::collections::BTreeMap::::new(); + for &(rk, value, rt, diff) in &original { + if rk == k && rt.less_equal(&time) { + *counts.entry(if unit { 0 } else { value }).or_default() += diff; + } + } + let expected = Diff::from(counts.values().any(|&d| d != 0)); + let observed: Diff = actual.iter().filter(|r| r.0 == key(k) && r.1.less_equal(&time)).map(|r| r.2).sum(); + assert_eq!(observed, expected, "key shape {key_shape}, unit {unit}, key {k}, time {time:?}"); + } + } + } + } + } + } +} + +#[test] +fn unit_input_remains_resolvable_for_other_reducers() { + for reducer in [Reducer::Min, Reducer::Count, Reducer::Collect] { + let mut tactic = ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone())); + let (mut source, mut output) = (Vec::new(), Vec::new()); + let mut count = 0; + for (epoch, diff) in [2, -2, -3, 4].into_iter().enumerate() { + let input = Rc::new(columns_to_batch(CValue::u64(vec![7]), CValue::Unit(1), vec![epoch as u64], vec![diff])); + let lower = Antichain::from_elem(epoch as u64); + let upper = Antichain::from_elem(epoch as u64 + 1); + let (result, pending) = tactic.retire(source.clone(), output.clone(), vec![input.clone()], &lower, &upper, &lower); + assert!(pending.is_empty()); + source.push(input); + if let Some(batch) = result.and_then(|s| s.inner) { output.push(batch); } + let mut actual = Vec::new(); + for chunk in chunks_of(&output) { + let values = untranscode(chunk.vals().clone(), &corgi::shape_of_value(chunk.vals())); + actual.extend(values.into_iter().zip(chunk.diffs().iter().copied())); + } + differential_dataflow::consolidation::consolidate(&mut actual); + count += diff; + let expected = match reducer { + Reducer::Count if count > 0 => vec![(Value::Tuple(vec![Value::Int(count)]), 1)], + Reducer::Min if count != 0 => vec![(Value::unit(), 1)], + Reducer::Collect if count != 0 => vec![(Value::List(vec![Value::unit(); count.max(0) as usize]), 1)], + _ => vec![], + }; + assert_eq!(actual, expected, "epoch {epoch}"); + } + } +}