Use Batcher to implement all of Arrange

differential-dataflow/examples/columnar/main.rs

@@ -98,7 +98,7 @@ mod reachability {
     use differential_dataflow::operators::arrange::arrangement::arrange_core;
     use differential_dataflow::operators::join::join_traces;
 
-    use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Builder as ValBuilder, Chunker as ValChunker, Spine as ValSpine};
+    use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Builder as ValBuilder, Spine as ValSpine};
     use differential_dataflow::columnar::collection::{Builder as ValColBuilder, Pact as ValPact, RecordedUpdates, as_recorded_updates};
 
     type Node = u32;
@@ -128,16 +128,12 @@ mod reachability {
             let reach_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 };
 
             let edges_arr = arrange_core::<_, _,
-                ValChunker<(Node, Node, IterTime, Diff)>,
                 _,
-                ValBuilder<Node, Node, IterTime, Diff>,
                 ValSpine<Node, Node, IterTime, Diff>,
             >(edges_inner.inner, edges_pact, "Edges", ValBatcher::new);
 
             let reach_arr = arrange_core::<_, _,
-                ValChunker<(Node, (), IterTime, Diff)>,
                 _,
-                ValBuilder<Node, (), IterTime, Diff>,
                 ValSpine<Node, (), IterTime, Diff>,
             >(reach.inner, reach_pact, "Reach", ValBatcher::new);
 
@@ -161,9 +157,7 @@ mod reachability {
             // Arrange for reduce.
             let combined_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 };
             let combined_arr = arrange_core::<_, _,
-                ValChunker<(Node, (), IterTime, Diff)>,
                 _,
-                ValBuilder<Node, (), IterTime, Diff>,
                 ValSpine<Node, (), IterTime, Diff>,
             >(combined.inner, combined_pact, "Combined", ValBatcher::new);
 

differential-dataflow/examples/columnar_spill.rs

@@ -50,7 +50,7 @@ fn reset_stats() {
     BYTES_COMPRESSED.store(0, Ordering::Relaxed);
 }
 
-use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Builder as ValBuilder, Chunker as ValChunker, Spine as ValSpine};
+use differential_dataflow::columnar::trace::{Batcher as ValBatcher, Spine as ValSpine};
 use differential_dataflow::columnar::collection::Builder as ValColBuilder;
 use differential_dataflow::columnar::trace::spill::{self, BytesSource, BytesStore, SpillStats};
 use differential_dataflow::columnar::updates::{Updates, UpdatesTyped};
@@ -238,9 +238,7 @@ fn run_timely_dataflow(times: u64, keys_per_time: u64, workers: usize, sample_se
             let stream = scope.input_from(&mut input);
             let arranged = arrange_core::<
                 _, _,
-                ValChunker<(u64, u64, u64, i64)>,
                 _,
-                ValBuilder<u64, u64, u64, i64>,
                 ValSpine<u64, u64, u64, i64>,
             >(stream, Pipeline, "ColumnarSpillArrange", ValBatcher::new);
             arranged.stream.probe_with(&mut probe);

differential-dataflow/examples/spines.rs

@@ -55,21 +55,21 @@ fn main() {
 
             match mode.as_str() {
                 "key" => {
-                    use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatcher, VecOrdKeyBuilder, OrdKeySpine};
+                    use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatcher, OrdKeySpine};
                     let (data_input, data) = scope.new_collection::<String, isize>();
                     let (keys_input, keys) = scope.new_collection::<String, isize>();
-                    let data = data.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder<String,_,isize>, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
-                    let keys = keys.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder<String,_,isize>, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
+                    let data = data.map(|k| (k, ())).arrange::<_, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
+                    let keys = keys.map(|k| (k, ())).arrange::<_, OrdKeySpine<String,_,isize>>(OrdKeyBatcher::new);
                     keys.join_core(data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                     Workload { data_input, keys_input }
                 },
                 "val" => {
-                    use differential_dataflow::trace::implementations::ord_neu::{OrdValBatcher, VecOrdValBuilder, OrdValSpine};
+                    use differential_dataflow::trace::implementations::ord_neu::{OrdValBatcher, OrdValSpine};
                     let (data_input, data) = scope.new_collection::<String, isize>();
                     let (keys_input, keys) = scope.new_collection::<String, isize>();
-                    let data = data.map(|x| (x, ())).arrange::<_, VecOrdValBuilder<String,(),_,isize>, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
-                    let keys = keys.map(|x| (x, ())).arrange::<_, VecOrdValBuilder<String,(),_,isize>, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
+                    let data = data.map(|x| (x, ())).arrange::<_, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
+                    let keys = keys.map(|x| (x, ())).arrange::<_, OrdValSpine<String,(),_,isize>>(OrdValBatcher::new);
                     keys.join_core(data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                     Workload { data_input, keys_input }
@@ -79,7 +79,8 @@ fn main() {
                     // the same generic `Chunk` harness as `vec` via a
                     // `ContainerChunker<ColChunk>`.
                     use differential_dataflow::Hashable;
-                    use differential_dataflow::columnar::trace::{Batcher, Builder, Spine, ColChunk};
+                    use differential_dataflow::columnar::trace::{Spine, ColChunk};
+                    use differential_dataflow::trace::chunk::ChunkBatcher;
                     use differential_dataflow::trace::implementations::chunker::ContainerChunker;
                     use differential_dataflow::operators::arrange::arrangement::arrange_core;
                     use timely::dataflow::channels::pact::Exchange;
@@ -89,13 +90,12 @@ fn main() {
                     let data = data.map(|x| (x, ()));
                     let keys = keys.map(|x| (x, ()));
 
-                    type Ba = Batcher<String, (), u64, isize>;
-                    type Bu = Builder<String, (), u64, isize>;
-                    type Sp = Spine<String, (), u64, isize>;
                     type Chu = ContainerChunker<ColChunk<(String, (), u64, isize)>>;
+                    type Ba = ChunkBatcher<Chu, ColChunk<(String, (), u64, isize)>>;
+                    type Sp = Spine<String, (), u64, isize>;
                     let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into());
-                    let data = arrange_core::<_, _, Chu, Ba, Bu, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
-                    let keys = arrange_core::<_, _, Chu, Ba, Bu, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
+                    let data = arrange_core::<_, _, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
+                    let keys = arrange_core::<_, _, Ba, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
                     // `ColChunk`'s cursor yields `Val = columnar::Ref<()> = ()`, not `&()`.
                     keys.join_core(data, |_k, _, _| Option::<()>::None)
                         .probe_with(&mut probe);
@@ -106,8 +106,7 @@ fn main() {
                     // insert allocates a `String`) but arranged through the `Chunk`
                     // harness via a `ContainerChunker<VecChunk>`.
                     use differential_dataflow::Hashable;
-                    use differential_dataflow::trace::chunk::vec::{ChunkBatcher, ChunkBuilder, ChunkSpine, VecChunk};
-                    use differential_dataflow::trace::implementations::chunker::ContainerChunker;
+                    use differential_dataflow::trace::chunk::vec::{ChunkBatcher, ChunkSpine};
                     use differential_dataflow::operators::arrange::arrangement::arrange_core;
                     use timely::dataflow::channels::pact::Exchange;
 
@@ -117,12 +116,10 @@ fn main() {
                     let keys = keys.map(|x| (x, ()));
 
                     type Ba = ChunkBatcher<String, (), u64, isize>;
-                    type Bu = ChunkBuilder<String, (), u64, isize>;
                     type Sp = ChunkSpine<String, (), u64, isize>;
-                    type Chu = ContainerChunker<VecChunk<String, (), u64, isize>>;
                     let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into());
-                    let data = arrange_core::<_, _, Chu, Ba, Bu, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
-                    let keys = arrange_core::<_, _, Chu, Ba, Bu, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
+                    let data = arrange_core::<_, _, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new);
+                    let keys = arrange_core::<_, _, Ba, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new);
                     keys.join_core(data, |_k, &(), &()| Option::<()>::None)
                         .probe_with(&mut probe);
                     Workload { data_input, keys_input }

differential-dataflow/src/collection.rs

@@ -957,22 +957,21 @@ pub mod vec {
         /// });
         /// ```
         pub fn consolidate(self) -> Self {
-            use crate::trace::implementations::{KeyBatcher, KeyBuilder, KeySpine};
-            self.consolidate_named::<_,KeyBuilder<_,_,_>, KeySpine<_,_,_>,_>("Consolidate", KeyBatcher::new, |key,&()| key.clone())
+            use crate::trace::implementations::{KeyBatcher, KeySpine};
+            self.consolidate_named::<_, KeySpine<_,_,_>,_>("Consolidate", KeyBatcher::new, |key,&()| key.clone())
         }
 
         /// As `consolidate` but with the ability to name the operator, specify the trace type,
         /// and provide the function `reify` to produce owned keys and values..
-        pub fn consolidate_named<Ba, Bu, Tr, F>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba, reify: F) -> Self
+        pub fn consolidate_named<Ba, Tr, F>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba, reify: F) -> Self
         where
-            Ba: crate::trace::Batcher<T, Vec<((D, ()), T, R)>, Vec<Vec<((D, ()), T, R)>>> + 'static,
+            Ba: crate::trace::Batcher<Vec<((D, ()), T, R)>, Time = T, Output: Into<Tr::Batch>> + 'static,
             Tr: crate::trace::Trace<Batch: Navigable, Time=T>+'static,
             for<'a> BatchCursor<Tr>: Cursor<Time=Tr::Time, Diff=R>,
-            Bu: crate::trace::Builder<Time=Tr::Time, Input=Vec<((D, ()), T, R)>, Output: Into<Tr::Batch>>,
             F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static,
         {
             self.map(|k| (k, ()))
-                .arrange_named::<Ba, Bu, Tr>(name, batcher)
+                .arrange_named::<Ba, Tr>(name, batcher)
                 .as_collection(reify)
         }
 
@@ -1020,9 +1019,8 @@ pub mod vec {
         }
     }
 
-    use crate::trace::implementations::{ValSpine, ValBatcher, ValBuilder};
-    use crate::trace::implementations::{KeySpine, KeyBatcher, KeyBuilder};
-    use crate::trace::implementations::ContainerChunker;
+    use crate::trace::implementations::{ValSpine, ValBatcher};
+    use crate::trace::implementations::{KeySpine, KeyBatcher};
     impl<'scope, T, K, V, R> Collection<'scope, T, (K, V), R>
     where
         T: Timestamp + Lattice,
@@ -1036,24 +1034,22 @@ pub mod vec {
         /// same-type containers. For chunker setups that convert between container types (e.g.
         /// columnar layouts), call [`arrange_core`](crate::operators::arrange::arrangement::arrange_core)
         /// directly.
-        pub fn arrange<Ba, Bu, Tr>(self, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
+        pub fn arrange<Ba, Tr>(self, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
         where
-            Ba: crate::trace::Batcher<T, Vec<((K, V), T, R)>, Vec<Bu::Input>> + 'static,
-            Bu: crate::trace::Builder<Time=T, Output: Into<Tr::Batch>>,
+            Ba: crate::trace::Batcher<Vec<((K, V), T, R)>, Time = T, Output: Into<Tr::Batch>> + 'static,
             Tr: crate::trace::Trace<Time=T> + 'static,
         {
-            self.arrange_named::<Ba, Bu, Tr>("Arrange", batcher)
+            self.arrange_named::<Ba, Tr>("Arrange", batcher)
         }
 
         /// As [`Collection::arrange`] but with the ability to name the operator.
-        pub fn arrange_named<Ba, Bu, Tr>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
+        pub fn arrange_named<Ba, Tr>(self, name: &str, batcher: impl FnOnce(Option<crate::logging::Logger>, usize) -> Ba) -> Arranged<'scope, TraceAgent<Tr>>
         where
-            Ba: crate::trace::Batcher<T, Vec<((K, V), T, R)>, Vec<Bu::Input>> + 'static,
-            Bu: crate::trace::Builder<Time=T, Output: Into<Tr::Batch>>,
+            Ba: crate::trace::Batcher<Vec<((K, V), T, R)>, Time = T, Output: Into<Tr::Batch>> + 'static,
             Tr: crate::trace::Trace<Time=T> + 'static,
         {
             let exchange = timely::dataflow::channels::pact::Exchange::new(move |update: &((K,V),T,R)| (update.0).0.hashed().into());
-            crate::operators::arrange::arrangement::arrange_core::<_, _, ContainerChunker<Vec<((K, V), T, R)>>, Ba, Bu, _>(self.inner, exchange, name, batcher)
+            crate::operators::arrange::arrangement::arrange_core::<_, _, Ba, _>(self.inner, exchange, name, batcher)
         }
     }
 
@@ -1072,7 +1068,7 @@ pub mod vec {
 
         /// As `arrange_by_key` but with the ability to name the arrangement.
         pub fn arrange_by_key_named(self, name: &str) -> Arranged<'scope, TraceAgent<ValSpine<K, V, T, R>>> {
-            self.arrange_named::<_,ValBuilder<_,_,_,_>,_>(name, ValBatcher::new)
+            self.arrange_named::<_,_>(name, ValBatcher::new)
         }
     }
 
@@ -1092,7 +1088,7 @@ pub mod vec {
         /// As `arrange_by_self` but with the ability to name the arrangement.
         pub fn arrange_by_self_named(self, name: &str) -> Arranged<'scope, TraceAgent<KeySpine<K, T, R>>> {
             self.map(|k| (k, ()))
-                .arrange_named::<_,KeyBuilder<_,_,_>,_>(name, KeyBatcher::new)
+                .arrange_named::<_,_>(name, KeyBatcher::new)
         }
     }
 

differential-dataflow/src/columnar/trace/mod.rs

@@ -21,7 +21,7 @@ pub use chunk::ColChunk;
 /// The columnar trace: a spine of `Rc`-shared [`ColChunk`] batches.
 pub type Spine<K, V, T, R> = crate::trace::chunk::ChunkSpine<ColChunk<(K, V, T, R)>>;
 /// The columnar merge batcher (the chunk harness's `MergeBatcher` over `ColChunk`).
-pub type Batcher<K, V, T, R> = crate::trace::chunk::ChunkBatcher<ColChunk<(K, V, T, R)>>;
+pub type Batcher<K, V, T, R> = crate::trace::chunk::ChunkBatcher<Chunker<(K, V, T, R)>, ColChunk<(K, V, T, R)>>;
 /// The columnar batch builder.
 pub type Builder<K, V, T, R> = crate::trace::chunk::ChunkBuilder<ColChunk<(K, V, T, R)>>;
 /// The input chunker: melds `RecordedUpdates<U>` streams into `ColChunk<U>` batches.

differential-dataflow/src/operators/arrange/arrangement.rs

@@ -24,7 +24,7 @@ use timely::dataflow::operators::generic::Operator;
 use timely::dataflow::channels::pact::{ParallelizationContract, Pipeline};
 use timely::progress::Timestamp;
 use timely::progress::Antichain;
-use timely::container::{ContainerBuilder, PushInto};
+use timely::container::PushInto;
 use timely::dataflow::operators::{Capability, CapabilitySet};
 use timely::progress::Stamp;
 
@@ -303,7 +303,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> {
 /// This operator arranges a stream of values into a shared trace, whose contents it maintains.
 /// It uses the supplied parallelization contract to distribute the data, which does not need to
 /// be consistently by key (though this is the most common).
-pub fn arrange_core<'scope, P, C, Chu, Ba, Bu, Tr>(
+pub fn arrange_core<'scope, P, C, Ba, Tr>(
     stream: Stream<'scope, Tr::Time, C>,
     pact: P,
     name: &str,
@@ -312,9 +312,7 @@ pub fn arrange_core<'scope, P, C, Chu, Ba, Bu, Tr>(
 where
     C: Container + Clone + 'static,
     P: ParallelizationContract<Tr::Time, C>,
-    Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static,
-    Ba: Batcher<Tr::Time, Chu::Container, Vec<Bu::Input>> + 'static,
-    Bu: Builder<Time=Tr::Time, Output: Into<Tr::Batch>>,
+    Ba: Batcher<C, Time = Tr::Time, Output: Into<Tr::Batch>> + 'static,
     Tr: Trace+'static,
 {
     // The `Arrange` operator is tasked with reacting to an advancing input
@@ -363,8 +361,6 @@ where
         // Initialize to the minimal input frontier.
         let mut prev_frontier = Antichain::from_elem(Tr::Time::minimum());
 
-        let mut chunker = Chu::default();
-
         move |(input, frontier), output| {
 
             // As we receive data, we need to (i) stash the data and (ii) keep *enough* capabilities.
@@ -375,10 +371,7 @@ where
                 for capability in cap.retain_stamp(0).iter() {
                     capabilities.insert(capability.clone());
                 }
-                chunker.push_into(data);
-                while let Some(chunk) = chunker.extract() {
-                    batcher.insert(chunk);
-                }
+                batcher.insert(data);
             });
 
             // The frontier may have advanced by multiple elements, which is an issue because
@@ -393,13 +386,6 @@ where
             // frontier isn't equal to the previous. It is only in this case that we have any
             // data processing to do.
             if prev_frontier.borrow() != frontier.frontier() {
-                // Flush any data the chunker is still accumulating into the batcher before we
-                // seal. The batcher only sees chunks the chunker has emitted; without this drain
-                // a partial final chunk would never reach the batcher.
-                while let Some(chunk) = chunker.finish() {
-                    batcher.insert(chunk);
-                }
-
                 // There are two cases to handle with some care:
                 //
                 // 1. If any held capabilities are not in advance of the new input frontier,
@@ -433,9 +419,9 @@ where
                         frontier.frontier().to_owned(),
                         Antichain::from_elem(Tr::Time::minimum()),
                     );
-                    let (chain, retained) = batcher.extract(frontier.frontier());
+                    let (extracted, retained) = batcher.extract(frontier.frontier());
 
-                    let batch = trace::Span::new(description, chain.and_then(|mut chain| Bu::seal(&mut chain)).map(Into::into));
+                    let batch = trace::Span::new(description, extracted.map(Into::into));
 
                     let stamp = retired.iter().map(|c| c.time().clone()).collect::<Stamp<_>>();
                     writer.insert(batch.clone(), stamp);
@@ -444,11 +430,10 @@ where
                     output.session(&retired).give(batch);
 
                     // Having extracted and sent the batch of updates not in advance of the input
-                    // frontier, we should downgrade all capabilities to match the batcher's lower
-                    // update frontier.
+                    // frontier, we downgrade all capabilities to match the batcher's lower update
+                    // frontier.
                     // This may involve discarding capabilities, which is fine as any new updates arrive
                     // in messages with new capabilities.
-
                     let mut new_capabilities = Antichain::new();
                     for time in retained.iter() {
                         if let Some(capability) = capabilities.elements().iter().find(|c| c.time().less_equal(time)) {
@@ -458,7 +443,6 @@ where
                             panic!("failed to find capability");
                         }
                     }
-
                     capabilities = new_capabilities;
                 }
                 else {