Move reduce timestamps and directly evaluate single dominating seeds

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<T, RIn, ROut> {
     key: u64,
     sweep: Sweep<T, RIn, ROut>,
+    direct: Option<(std::ops::Range<usize>, std::ops::Range<usize>)>,
     /// Times at or beyond `upper` the sweep has reached; carried forward when the slot retires.
     pended: Vec<T>,
     /// The time the sweep last suspended at, or `None` once it is spent.
@@ -402,7 +426,7 @@ struct KeySweep<T, RIn, ROut> {
 
 impl<T: Timestamp + Lattice, RIn: Semigroup + Clone, ROut: Semigroup + Clone> KeySweep<T, RIn, ROut> {
     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 }
     }
 }
 

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::<u64, u64>::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::<u64, u64>::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::<u64, u64>::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());
+}

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<T: Ord + Clone>(
+fn merge_present<T: timely::progress::Timestamp>(
     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<T, Diff>,
 ) -> bool {
     let ordered_keys = corgi::arrange::leaf_slice(keys_col).is_some() || {
@@ -328,20 +328,20 @@ fn merge_present<T: Ord + Clone>(
     }, "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]), &times[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<T: Ord + Clone>(
         return true;
     }
 
-    let mut heap: BinaryHeap<Reverse<((u64, u64), &T, usize, usize)>> = BinaryHeap::new();
+    let mut heap: BinaryHeap<Reverse<((u64, u64), T, usize, usize)>> = BinaryHeap::new();
     let mut lo = 0usize;
     for (run, &hi) in run_ends.iter().enumerate() {
-        heap.push(Reverse(((khs[lo], vids[lo]), &times[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]), &times[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<T, Diff>,
     ) {
-        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, &times, &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, &times, &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(),