Batcher as subtrait

differential-dataflow/src/batcher/merge/mod.rs

@@ -18,7 +18,7 @@ use timely::progress::frontier::AntichainRef;
 use timely::progress::{frontier::Antichain, Timestamp};
 
 use crate::logging::{BatcherEvent, Logger};
-use crate::batcher::Batcher;
+use crate::batcher::{Batcher, Extract};
 
 /// Creates batches from chunks of sorted, consolidated tuples.
 ///
@@ -42,15 +42,22 @@ where
     Chu: ContainerBuilder<Container = M::Chunk> + for<'a> PushInto<&'a mut C>,
     S: Sealer<M::Chunk>,
 {
-    type Time = M::Time;
-    type Output = S::Output;
-
     fn insert(&mut self, container: &mut C) {
         self.chunker.push_into(container);
         while let Some(chunk) = self.chunker.extract().map(std::mem::take) {
             self.ladder.insert_chain(vec![chunk]);
         }
     }
+}
+
+impl<Chu, M, S> Extract for MergeBatcher<Chu, M, S>
+where
+    M: Extractor<Time: Timestamp>,
+    Chu: ContainerBuilder<Container = M::Chunk>,
+    S: Sealer<M::Chunk>,
+{
+    type Time = M::Time;
+    type Output = S::Output;
 
     // Extraction means finding those updates with times not greater or equal to any time in
     // `upper`. All updates must have time greater or equal to the previously used `upper`, by
@@ -480,7 +487,7 @@ pub mod vec {
 #[cfg(test)]
 mod test {
     use timely::progress::frontier::Antichain;
-    use crate::batcher::Batcher;
+    use crate::batcher::Extract;
     use super::{MergeBatcher, Sealer};
     use super::vec::VecMerger;
     use crate::batcher::merge::chunker::ContainerChunker;
@@ -508,7 +515,7 @@ mod test {
         let mut b = Bt::new(None, 0);
         b.ladder.chain_push(vec![vec![((100u64, ()), 5u64, 1i64), ((200u64, ()), 7u64, 1i64)]]);
         b.ladder.chain_push(vec![vec![((100u64, ()), 5u64, -1i64)]]);
-        let (_, retained) = Batcher::<In>::extract(&mut b, Antichain::from_elem(3).borrow());
+        let (_, retained) = Extract::extract(&mut b, Antichain::from_elem(3).borrow());
         let got: Vec<u64> = retained.iter().cloned().collect();
         assert_eq!(got, vec![7u64],
             "frontier held a capability at t=5, which consolidates to zero (got {got:?})");
@@ -520,7 +527,7 @@ mod test {
         let mut b = Bt::new(None, 0);
         b.ladder.chain_push(vec![vec![((100u64, ()), 5u64, 1i64)]]);
         b.ladder.chain_push(vec![vec![((200u64, ()), 7u64, 1i64)]]);
-        let (_, retained) = Batcher::<In>::extract(&mut b, Antichain::from_elem(3).borrow());
+        let (_, retained) = Extract::extract(&mut b, Antichain::from_elem(3).borrow());
         let got: Vec<u64> = retained.iter().cloned().collect();
         assert_eq!(got, vec![5u64]);
     }

differential-dataflow/src/batcher/mod.rs

@@ -13,17 +13,21 @@ pub mod merge;
 ///
 /// The implementor determines the meaning of extraction by a frontier; it is not required to be by
 /// antichain partial order.
-pub trait Batcher<C0> {
-    /// The timestamps by which updates are carved out.
-    type Time;
-    /// The batches released by extraction.
-    type Output;
-
+pub trait Batcher<C0> : Extract {
     /// Takes the updates in `container`, leaving it in an undefined state.
     ///
     /// The implementor decides whether to claim the container's allocation or to drain it and
     /// leave the allocation with the caller, who is free to reuse the container either way.
     fn insert(&mut self, container: &mut C0);
+}
+
+/// Supertrait for `Batcher` that ensures a consistent time and output across `C0` implementations.
+pub trait Extract {
+    /// The timestamps by which updates are carved out.
+    type Time;
+    /// The batches released by extraction.
+    type Output;
+
     /// Extracts the updates `upper` unblocks as a batch, and lower bounds the times of those retained.
     ///
     /// What `upper` unblocks is the implementor's to decide. It can be based on the antichain up

differential-dataflow/tests/trace.rs

@@ -2,7 +2,7 @@ use timely::dataflow::operators::generic::OperatorInfo;
 use timely::progress::{Antichain, frontier::AntichainRef};
 
 use differential_dataflow::trace::implementations::{ValBatcher, ValSpine};
-use differential_dataflow::batcher::Batcher;
+use differential_dataflow::batcher::{Batcher, Extract};
 use differential_dataflow::trace::{Description, Span, Trace, TraceReader};
 use differential_dataflow::trace::cursor::{Cursor, cursor_list};
 
@@ -25,7 +25,7 @@ fn get_trace() -> ValSpine<u64, u64, usize, i64> {
         let mut lower = Antichain::from_elem(0);
         for i in batch_ts {
             let upper = Antichain::from_elem(*i);
-            let (batch, _retained) = Batcher::<Vec<((u64, u64), usize, i64)>>::extract(&mut batcher, upper.borrow());
+            let (batch, _retained) = batcher.extract(upper.borrow());
             let description = Description::new(lower, upper.clone(), Antichain::from_elem(0));
             trace.insert(Span::new(description, batch.map(Into::into)));
             lower = upper;

dogsdogsdogs/src/operators/half_join.rs

@@ -32,7 +32,7 @@ use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable}
 use differential_dataflow::difference::Semigroup;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::batcher::Batcher;
+use differential_dataflow::batcher::{Batcher, Extract};
 use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchVal, Cursor, Navigable, TraceReader};
 use differential_dataflow::trace::cursor::cursor_list;
 use differential_dataflow::consolidation::{consolidate, consolidate_updates};
@@ -488,12 +488,15 @@ pub mod cursors {
     }
 
     impl<D: Ord, T: Timestamp, R: Semigroup> Batcher<Vec<(D, T, R)>> for BlobList<D, T, R> {
-        type Time = T;
-        type Output = Vec<Vec<(D, T, R)>>;
-
         fn insert(&mut self, container: &mut Vec<(D, T, R)>) {
             self.stage.extend(container.drain(..).map(|(d,t,r)| (t,d,r)));
         }
+    }
+
+    impl<D: Ord, T: Timestamp, R: Semigroup> Extract for BlobList<D, T, R> {
+        type Time = T;
+        type Output = Vec<Vec<(D, T, R)>>;
+
         fn extract<'a>(&'a mut self, frontier: AntichainRef<'_, T>) -> (Option<Vec<Vec<(D, T, R)>>>, AntichainRef<'a, T>) {
             // Handle any staged updates first.
             consolidate_updates(&mut self.stage);

interactive/src/runtime/half_join/batcher.rs

@@ -7,7 +7,7 @@ use std::cmp::Ordering;
 use timely::progress::Antichain;
 use timely::progress::frontier::AntichainRef;
 
-use differential_dataflow::batcher::Batcher;
+use differential_dataflow::batcher::{Batcher, Extract};
 use differential_dataflow::batcher::merge::{MergeLadder, Merger};
 
 use corgi::arrange::{gather, gather_lanes, sort_blocks, survey_groups, GroupRun};
@@ -228,14 +228,15 @@ impl HalfJoinBatcher {
 }
 
 impl Batcher<CC> for HalfJoinBatcher {
-    type Time = Time;
-    type Output = Vec<Run>;
-
     fn insert(&mut self, container: &mut CC) {
         if !container.times.is_empty() {
             self.stage.push(std::mem::take(container));
         }
     }
+}
+impl Extract for HalfJoinBatcher {
+    type Time = Time;
+    type Output = Vec<Run>;
 
     fn extract<'a>(&'a mut self, frontier: AntichainRef<'_, Time>) -> (Option<Vec<Run>>, AntichainRef<'a, Time>) {
         let width = self.width;