diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs index b7c04f6e9..2235d0aae 100644 --- a/differential-dataflow/src/operators/int_proxy/reduce.rs +++ b/differential-dataflow/src/operators/int_proxy/reduce.rs @@ -237,9 +237,9 @@ where let before = from; window.clear(); self.backend.next_window(&instance, &changed, &mut from, &mut window); - let p_in = &window.input; + let p_in = &mut window.input; let seeds = &window.seeds; - let p_out = &window.output; + 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"); debug_assert!( @@ -303,14 +303,27 @@ where while due_pos < due.rows.len() && due.rows[due_pos].0 < key { due_pos += 1; } let start = due_pos; while due_pos < due.rows.len() && due.rows[due_pos].0 == key { due_pos += 1; } - let owed = due.rows[start..due_pos].iter().map(|&(_, row)| due.times[row].clone()); - slot.sweep.load( - owed, - (n0..n1).map(|n| seeds[n].1.clone()), - (i0..i1).map(|i| (p_in[i].0.1, p_in[i].1.clone(), p_in[i].2.clone())), - (o0..o1).map(|o| (p_out[o].0.1, p_out[o].1.clone(), p_out[o].2.clone())), - ); - slot.at = slot.sweep.next_crossing(upper, &mut slot.pended); + let owed = &due.rows[start..due_pos]; + let novel = &seeds[n0..n1]; + // One seed dominating every record has exactly one possible evaluation. + 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))); + 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 } + else { Some(at.clone()) } + } else { + 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())), + ); + slot.sweep.next_crossing(upper, &mut slot.pended) + }; if slot.at.is_some() { live.push(n_slots); } else if !slot.pended.is_empty() { deferred.extend(slot.pended.drain(..).map(|time| (time, key))); @@ -334,8 +347,15 @@ where let at = slots[si].at.clone().expect("live slots are suspended at a time"); in_accum.clear(); cur_out.clear(); - slots[si].sweep.input_at(&at, &mut in_accum); - slots[si].sweep.output_at(&at, &mut cur_out); + 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); + } else { + slots[si].sweep.input_at(&at, &mut in_accum); + slots[si].sweep.output_at(&at, &mut cur_out); + } // 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; } @@ -357,7 +377,9 @@ where for (vid, d) in &corr[cstart..cend] { deltas.push(((slots[*si].key, *vid), at.clone(), d.clone())); } - slots[*si].sweep.commit(at, corr[cstart..cend].iter().cloned()); + if slots[*si].direct.is_none() { + slots[*si].sweep.commit(at, corr[cstart..cend].iter().cloned()); + } } cstart = cend; } @@ -366,7 +388,8 @@ where // Step every live key past the time it was suspended at, and retire the spent ones. for &si in live.iter() { let slot = &mut slots[si]; - slot.at = slot.sweep.next_crossing(upper, &mut slot.pended); + slot.at = if slot.direct.is_some() { None } + else { slot.sweep.next_crossing(upper, &mut slot.pended) }; if slot.at.is_none() && !slot.pended.is_empty() { deferred.extend(slot.pended.drain(..).map(|time| (time, slot.key))); } @@ -394,6 +417,7 @@ where struct KeySweep { key: u64, sweep: Sweep, + direct: Option<(std::ops::Range, std::ops::Range)>, /// Times at or beyond `upper` the sweep has reached; carried forward when the slot retires. pended: Vec, /// The time the sweep last suspended at, or `None` once it is spent. @@ -402,7 +426,7 @@ struct KeySweep { impl KeySweep { fn empty() -> Self { - KeySweep { key: 0, sweep: Sweep::new(), pended: Vec::new(), at: None } + KeySweep { key: 0, sweep: Sweep::new(), direct: None, pended: Vec::new(), at: None } } } diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index 91dab3427..691e5535d 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -519,3 +519,66 @@ fn bounded_sweeps_match_partial_order_oracle() { assert_eq!(hread(&all), expected, "limit {limit}"); } } + +/// A single seed stays pending beyond upper, then evaluates with no novel input. +#[test] +fn reduce_single_seed_deferred_then_due() { + let time = Product::::new; + let lower = Antichain::from_elem(time(0, 0)); + let middle = Antichain::from_elem(time(1, 1)); + let seed = time(2, 2); + let input = hbatch(vec![((7u64, 3u64), seed, 1i64)], time(0, 0), time(3, 3)); + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(max_logic)); + let (first, pending) = tactic.retire( + vec![], vec![], vec![input.clone()], &lower, &middle, &lower, + ); + let outputs: Vec<_> = first.into_iter().filter_map(|s| s.inner).collect(); + assert!(hread(&outputs).is_empty(), "a seed beyond upper must not evaluate"); + assert_eq!(pending, Antichain::from_elem(seed)); + let (second, pending) = tactic.retire( + vec![input], outputs, vec![], &middle, &Antichain::from_elem(time(3, 3)), &pending, + ); + let outputs: Vec<_> = second.into_iter().filter_map(|s| s.inner).collect(); + assert_eq!(hread(&outputs), vec![((7, 3), seed, 1)], "the due seed needs no novel support"); + assert!(pending.is_empty()); +} + +/// Input and prior output at exactly the seed time participate in direct reconciliation. +#[test] +fn reduce_single_seed_includes_equal_times() { + let time = Product::::new; + let zero = time(0, 0); + let seed = time(1, 1); + let prior = hbatch(vec![((7u64, 3u64), zero, 1i64)], zero, seed); + let input = hbatch(vec![((7u64, 9u64), seed, 1i64)], seed, time(2, 2)); + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(max_logic)); + // The backend advances both the prior input and output to lower = seed. + let (result, pending) = tactic.retire( + vec![prior.clone()], vec![prior], vec![input], + &Antichain::from_elem(seed), &Antichain::from_elem(time(2, 2)), &Antichain::from_elem(seed), + ); + let outputs: Vec<_> = result.into_iter().filter_map(|s| s.inner).collect(); + assert_eq!(hread(&outputs), vec![((7, 3), seed, -1), ((7, 9), seed, 1)]); + assert!(pending.is_empty()); +} + +/// One incomparable record requires both the seed evaluation and its later join. +#[test] +fn reduce_single_seed_with_incomparable_record() { + let time = Product::::new; + let zero = time(0, 0); + let seed = time(2, 0); + let mut lower = Antichain::from_elem(time(0, 2)); + lower.insert(seed); + let prior = hbatch(vec![((7u64, 9u64), time(1, 1), 1i64)], zero, time(1, 1)); + let input = hbatch(vec![((7u64, 3u64), seed, 1i64)], zero, time(2, 2)); + let mut tactic = ProxyReduceTactic::new(VecReduceBackend::new(max_logic)); + let (result, pending) = tactic.retire( + vec![prior.clone()], vec![prior], vec![input], + &lower, &Antichain::from_elem(time(3, 3)), &lower, + ); + let outputs: Vec<_> = result.into_iter().filter_map(|s| s.inner).collect(); + assert_eq!(hread(&outputs), vec![((7, 3), seed, 1), ((7, 3), time(2, 1), -1)], + "including the incomparable value at the seed, or skipping the join, is incorrect"); + assert!(pending.is_empty()); +} diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 294434821..b36b804b8 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -293,9 +293,9 @@ where /// 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( +fn merge_present( keys_col: &CValue, vals_col: &CValue, - khs: &[u64], vids: &[u64], times: &[T], diffs: &[Diff], run_ends: &[usize], + khs: &[u64], vids: &[u64], times: &mut [T], diffs: &[Diff], run_ends: &[usize], bridge: &mut ProxyBridge, ) -> bool { let ordered_keys = corgi::arrange::leaf_slice(keys_col).is_some() || { @@ -328,20 +328,20 @@ fn merge_present( }, "identity ids do not preserve selected chunk order"); let mut current: Option<((u64, u64), T, Diff)> = None; - let mut accumulate = |kv, time: &T, diff| { - if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == time) { + let mut accumulate = |kv, time: T, diff| { + if current.as_ref().is_some_and(|(ckv, ct, _)| ckv == &kv && ct == &time) { current.as_mut().unwrap().2 += diff; } else { if let Some(record) = current.take() { if record.2 != 0 { bridge.push(record); } } - current = Some((kv, time.clone(), diff)); + current = Some((kv, time, diff)); } }; 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]), std::mem::replace(&mut times[index], T::minimum()), diffs[index]); } drop(accumulate); if let Some(record) = current { @@ -350,19 +350,20 @@ fn merge_present( return true; } - let mut heap: BinaryHeap> = BinaryHeap::new(); + 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]), std::mem::replace(&mut times[lo], T::minimum()), run, lo))); lo = hi; } while let Some(mut head) = heap.peek_mut() { - let Reverse((kv, time, run, index)) = *head; + let Reverse((kv, _, run, index)) = *head; + let time = std::mem::replace(&mut head.0.1, T::minimum()); accumulate(kv, time, diffs[index]); 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]), std::mem::replace(&mut times[next], T::minimum()), run, next)); } else { std::collections::binary_heap::PeekMut::pop(head); } @@ -400,18 +401,18 @@ where len: &mut usize, bridge: &mut ProxyBridge, ) { - let (p_keys, p_vals, khs, times, diffs, run_ends) = collect_present(chunks, keys); + let (p_keys, p_vals, khs, mut times, diffs, run_ends) = collect_present(chunks, keys); if khs.is_empty() { return; } let vids = ids(&p_vals); - let merged = merge_present(&p_keys, &p_vals, &khs, &vids, ×, &diffs, &run_ends, bridge); + let merged = merge_present(&p_keys, &p_vals, &khs, &vids, &mut times, &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); self.register_keys(p_keys, &khs); if !merged { - bridge.extend((0..khs.len()).map(|i| ((khs[i], vids[i]), times[i].clone(), diffs[i]))); + bridge.extend(times.into_iter().enumerate().map(|(i, time)| ((khs[i], vids[i]), time, diffs[i]))); consolidate_updates(bridge); } } @@ -648,14 +649,14 @@ where self.in_vals = concat_columns(&in_blocks); // 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); + let (o_keys, o_vals, o_khs, mut 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 merged = merge_present(&o_keys, &o_vals, &o_khs, &vids, &o_times, &o_diffs, &o_run_ends, &mut window.output); + let merged = merge_present(&o_keys, &o_vals, &o_khs, &vids, &mut o_times, &o_diffs, &o_run_ends, &mut window.output); self.register_keys(o_keys, &o_khs); self.register_vals(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]))); + 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); } } @@ -758,7 +759,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], &[10, 20], &mut [0u64, 0], &[1, 1], &[2], &mut bridge, )); assert_eq!(bridge.len(), 2); } @@ -789,11 +790,11 @@ mod tests { } 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 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(); assert!(merge_present(&CValue::u64(khs.clone()), &CValue::u64(vids.clone()), - &khs, &vids, ×, &diffs, &ends, &mut bridge)); + &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}"); @@ -809,7 +810,7 @@ mod tests { &vals, &[1, 1], &[10, 20], - &[0u64, 0], + &mut [0u64, 0], &[1, 1], &[2], &mut Vec::new(),