Split Builder along the seam its two users already follow

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

@@ -233,7 +233,7 @@ where
                                 // new stuff that we add.
                                 let batches = reader_local.batches_through(Antichain::new().borrow()).unwrap();
                                 let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(batches);
-                                let mut builder = Bu::new();
+                                let mut builder = Bu::default();
                                 let mut key_con = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
                                 for (key, mut list) in to_process {
 

differential-dataflow/src/operators/reduce.rs

@@ -171,7 +171,7 @@
 
                 // Drain input batches in order, capturing capabilities and the last upper.
                 input.for_each(|capability, batches| {
                     for capability in capability.retain_stamp(0).iter() {
                         capabilities.insert(capability.clone());
                     }
                     for batch in batches.drain(..) {
@@ -335,7 +335,7 @@
                 // Prepare one output buffer and builder: the batch spans [lower, upper) and
                 // ships stamped with the held times that justify its contents.
                 let mut output_updates = Vec::<(<B2::Cursor as Cursor>::ValOwn, TimeOf<B1>, <B2::Cursor as Cursor>::Diff)>::new();
-                let mut builder = Bu::new();
+                let mut builder = Bu::default();
                 // Temporary staging for output building.
                 let mut buffer = Bu::Input::default();
 
@@ -883,7 +883,7 @@
                 let (mut batch_cursor, ref batch_storage) = cursor_list(input_batches);
 
                 let mut output_updates = Vec::<(<B2::Cursor as Cursor>::ValOwn, TimeOf<B1>, <B2::Cursor as Cursor>::Diff)>::new();
-                let mut builder = Bu::new();
+                let mut builder = Bu::default();
                 let mut buffer = Bu::Input::default();
 
                 // Reuseable state for performing the computation.

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

@@ -656,6 +656,12 @@ pub struct ChunkBatchBuilder<C: Chunk> {
     output: VecDeque<C>,
 }
 
+impl<C: Chunk> Default for ChunkBatchBuilder<C> {
+    fn default() -> Self {
+        Self { input: VecDeque::new(), output: VecDeque::new() }
+    }
+}
+
 impl<C> crate::trace::Builder for ChunkBatchBuilder<C>
 where
     C: Chunk + Default + 'static,
@@ -665,10 +671,6 @@ where
     type Time = C::Time;
     type Output = ChunkBatch<C>;
 
-    fn with_capacity(_keys: usize, _vals: usize, _upds: usize) -> Self {
-        Self { input: VecDeque::new(), output: VecDeque::new() }
-    }
-
     fn push(&mut self, chunk: &mut C) {
         let chunk = std::mem::take(chunk);
         if chunk.len() > 0 {
@@ -684,6 +686,15 @@ where
         wrap(chunks)
     }
 
+}
+
+impl<C> crate::trace::Sealer<C> for ChunkBatchBuilder<C>
+where
+    C: Chunk + Default + 'static,
+    C::Time: timely::progress::Timestamp,
+{
+    type Output = ChunkBatch<C>;
+
     fn seal(chain: &mut Vec<C>) -> Option<Self::Output> {
         // We settle the chain because we are not guaranteed to received pre-settled data.
         // This should be efficient on pre-settled data.

differential-dataflow/src/trace/implementations/merge_batcher.rs

@@ -13,14 +13,14 @@ use timely::progress::frontier::AntichainRef;
 use timely::progress::{frontier::Antichain, Timestamp};
 
 use crate::logging::{BatcherEvent, Logger};
-use crate::trace::{Batcher, Builder};
+use crate::trace::{Batcher, Sealer};
 
 /// Creates batches from chunks of sorted, consolidated tuples.
 ///
-/// Chunking input is `Chu`'s business, merging chunks is `M`'s, and building the extracted chain
-/// into a batch is `Bu`'s; the batcher's own work is the geometric ladder of chains and the
+/// Chunking input is `Chu`'s business, merging chunks is `M`'s, and sealing the extracted chain
+/// into a batch is `S`'s; the batcher's own work is the geometric ladder of chains and the
 /// carve-by-frontier.
-pub struct MergeBatcher<Chu, M: Merger, Bu> {
+pub struct MergeBatcher<Chu, M: Merger, S> {
     /// Melds input containers into sorted, consolidated chunks.
     chunker: Chu,
     /// Sorted, consolidated chains, each paired with its cached summed update count.
@@ -42,17 +42,17 @@ pub struct MergeBatcher<Chu, M: Merger, Bu> {
     /// Timely operator ID.
     operator_id: usize,
     /// Seals each extracted chain into a batch.
-    builder: std::marker::PhantomData<Bu>,
+    sealer: std::marker::PhantomData<S>,
 }
 
-impl<C, Chu, M, Bu> Batcher<C> for MergeBatcher<Chu, M, Bu>
+impl<C, Chu, M, S> Batcher<C> for MergeBatcher<Chu, M, S>
 where
     M: Merger<Time: Timestamp>,
     Chu: ContainerBuilder<Container = M::Chunk> + for<'a> PushInto<&'a mut C>,
-    Bu: Builder<Input = M::Chunk>,
+    S: Sealer<M::Chunk>,
 {
     type Time = M::Time;
-    type Output = Bu::Output;
+    type Output = S::Output;
 
     fn insert(&mut self, container: &mut C) {
         self.chunker.push_into(container);
@@ -65,7 +65,7 @@ where
     // `upper`. All updates must have time greater or equal to the previously used `upper`, by
     // assumption that after extracting from a batcher we receive no more updates with times not
     // greater or equal to `upper`.
-    fn extract<'a>(&'a mut self, upper: AntichainRef<'_, M::Time>) -> (Option<Bu::Output>, AntichainRef<'a, M::Time>) {
+    fn extract<'a>(&'a mut self, upper: AntichainRef<'_, M::Time>) -> (Option<S::Output>, AntichainRef<'a, M::Time>) {
         // Flush whatever the chunker is still accumulating: a partial final chunk would
         // otherwise never reach the merge ladder.
         while let Some(chunk) = self.chunker.finish().map(std::mem::take) {
@@ -94,11 +94,11 @@ where
 
         self.stash.clear();
 
-        (Bu::seal(&mut readied), self.frontier.borrow())
+        (S::seal(&mut readied), self.frontier.borrow())
     }
 }
 
-impl<Chu: Default, M: Merger, Bu> MergeBatcher<Chu, M, Bu> {
+impl<Chu: Default, M: Merger, S> MergeBatcher<Chu, M, S> {
     /// Allocates a new empty batcher.
     ///
     /// The logger and operator identifier are used to report the batcher's memory footprint,
@@ -112,12 +112,12 @@ impl<Chu: Default, M: Merger, Bu> MergeBatcher<Chu, M, Bu> {
             chains: Vec::new(),
             stash: Vec::new(),
             frontier: Antichain::new(),
-            builder: std::marker::PhantomData,
+            sealer: std::marker::PhantomData,
         }
     }
 }
 
-impl<Chu, M: Merger, Bu> MergeBatcher<Chu, M, Bu> {
+impl<Chu, M: Merger, S> MergeBatcher<Chu, M, S> {
     /// Insert a chain and maintain chain properties: Chains are geometrically sized
     /// (by summed updates) and ordered by decreasing update weight.
     fn insert_chain(&mut self, chain: Vec<M::Chunk>) {
@@ -192,7 +192,7 @@ impl<Chu, M: Merger, Bu> MergeBatcher<Chu, M, Bu> {
     }
 }
 
-impl<Chu, M: Merger, Bu> Drop for MergeBatcher<Chu, M, Bu> {
+impl<Chu, M: Merger, S> Drop for MergeBatcher<Chu, M, S> {
     fn drop(&mut self) {
         // Cleanup chain to retract accounting information.
         while self.chain_pop().is_some() {}

differential-dataflow/src/trace/implementations/ord_neu.rs

@@ -248,7 +248,7 @@ pub mod val_batch {
     use timely::container::PushInto;
     use timely::progress::{Antichain, frontier::AntichainRef};
 
-    use crate::trace::{Builder, Cursor};
+    use crate::trace::{Builder, Sealer, Cursor};
     use crate::trace::implementations::spine_fueled::{SpineBatch, Merger};
     use crate::trace::implementations::{BatchContainer, BuilderInput};
     use crate::trace::implementations::layout;
@@ -614,19 +614,13 @@ pub mod val_batch {
         _marker: PhantomData<CI>,
     }
 
-    impl<L, CI> Builder for OrdValBuilder<L, CI>
+    impl<L, CI> OrdValBuilder<L, CI>
     where
-        L: for<'a> Layout<
-            KeyContainer: PushInto<CI::Key<'a>>,
-            ValContainer: PushInto<CI::Val<'a>>,
-        >,
-        CI: for<'a> BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
+        L: Layout,
     {
-
-        type Input = CI;
-        type Time = layout::Time<L>;
-        type Output = OrdValBatch<L>;
-
+        /// Allocates a builder with capacity for the specified keys, values, and updates.
+        ///
+        /// They represent respectively the number of distinct `key`, `(key, val)`, and total updates.
         fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self {
             Self {
                 result: OrdValStorage {
@@ -638,6 +632,24 @@ pub mod val_batch {
                 _marker: PhantomData,
             }
         }
+    }
+
+    impl<L: Layout, CI> Default for OrdValBuilder<L, CI> {
+        fn default() -> Self { Self::with_capacity(0, 0, 0) }
+    }
+
+    impl<L, CI> Builder for OrdValBuilder<L, CI>
+    where
+        L: for<'a> Layout<
+            KeyContainer: PushInto<CI::Key<'a>>,
+            ValContainer: PushInto<CI::Val<'a>>,
+        >,
+        CI: for<'a> BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
+    {
+
+        type Input = CI;
+        type Time = layout::Time<L>;
+        type Output = OrdValBatch<L>;
 
         #[inline]
         fn push(&mut self, chunk: &mut Self::Input) {
@@ -680,8 +692,20 @@ pub mod val_batch {
             (updates > 0).then(|| OrdValBatch { updates, storage: self.result })
         }
 
-        fn seal(chain: &mut Vec<Self::Input>) -> Option<Self::Output> {
-            let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);
+    }
+
+    impl<L, CI> Sealer<CI> for OrdValBuilder<L, CI>
+    where
+        L: for<'a> Layout<
+            KeyContainer: PushInto<CI::Key<'a>>,
+            ValContainer: PushInto<CI::Val<'a>>,
+        >,
+        CI: for<'a> BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
+    {
+        type Output = OrdValBatch<L>;
+
+        fn seal(chain: &mut Vec<CI>) -> Option<Self::Output> {
+            let (keys, vals, upds) = CI::key_val_upd_counts(&chain[..]);
             let mut builder = Self::with_capacity(keys, vals, upds);
             for mut chunk in chain.drain(..) {
                 builder.push(&mut chunk);
@@ -700,7 +724,7 @@ pub mod key_batch {
     use timely::container::PushInto;
     use timely::progress::{Antichain, frontier::AntichainRef};
 
-    use crate::trace::{Builder, Cursor};
+    use crate::trace::{Builder, Sealer, Cursor};
     use crate::trace::implementations::spine_fueled::{SpineBatch, Merger};
     use crate::trace::implementations::{BatchContainer, BuilderInput};
     use crate::trace::implementations::layout;
@@ -995,17 +1019,10 @@ pub mod key_batch {
         _marker: PhantomData<CI>,
     }
 
-    impl<L: Layout, CI> Builder for OrdKeyBuilder<L, CI>
-    where
-        L: for<'a> Layout<KeyContainer: PushInto<CI::Key<'a>>>,
-        L: Layout<ValContainer: BatchContainer<Owned: Default>>,
-        CI: BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
-    {
-
-        type Input = CI;
-        type Time = layout::Time<L>;
-        type Output = OrdKeyBatch<L>;
-
+    impl<L: Layout, CI> OrdKeyBuilder<L, CI> {
+        /// Allocates a builder with capacity for the specified keys and updates.
+        ///
+        /// They represent respectively the number of distinct `key` and total updates.
         fn with_capacity(keys: usize, _vals: usize, upds: usize) -> Self {
             Self {
                 result: OrdKeyStorage {
@@ -1016,6 +1033,22 @@ pub mod key_batch {
                 _marker: PhantomData,
             }
         }
+    }
+
+    impl<L: Layout, CI> Default for OrdKeyBuilder<L, CI> {
+        fn default() -> Self { Self::with_capacity(0, 0, 0) }
+    }
+
+    impl<L: Layout, CI> Builder for OrdKeyBuilder<L, CI>
+    where
+        L: for<'a> Layout<KeyContainer: PushInto<CI::Key<'a>>>,
+        L: Layout<ValContainer: BatchContainer<Owned: Default>>,
+        CI: BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
+    {
+
+        type Input = CI;
+        type Time = layout::Time<L>;
+        type Output = OrdKeyBatch<L>;
 
         #[inline]
         fn push(&mut self, chunk: &mut Self::Input) {
@@ -1047,8 +1080,18 @@ pub mod key_batch {
             })
         }
 
-        fn seal(chain: &mut Vec<Self::Input>) -> Option<Self::Output> {
-            let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);
+    }
+
+    impl<L: Layout, CI> Sealer<CI> for OrdKeyBuilder<L, CI>
+    where
+        L: for<'a> Layout<KeyContainer: PushInto<CI::Key<'a>>>,
+        L: Layout<ValContainer: BatchContainer<Owned: Default>>,
+        CI: BuilderInput<L::KeyContainer, L::ValContainer, Time=layout::Time<L>, Diff=layout::Diff<L>>,
+    {
+        type Output = OrdKeyBatch<L>;
+
+        fn seal(chain: &mut Vec<CI>) -> Option<Self::Output> {
+            let (keys, vals, upds) = CI::key_val_upd_counts(&chain[..]);
             let mut builder = Self::with_capacity(keys, vals, upds);
             for mut chunk in chain.drain(..) {
                 builder.push(&mut chunk);

differential-dataflow/src/trace/mod.rs

@@ -258,36 +258,42 @@ pub trait Batcher<C0> {
 }
 
 /// Functionality for building batches from ordered update sequences.
-pub trait Builder: Sized {
+///
+/// `Default` is the empty builder; a builder discovers its output as it is pushed, and so has
+/// no opportunity to size itself in advance.
+pub trait Builder: Default {
     /// Input item type.
     type Input;
     /// Timestamp type.
     type Time: Timestamp;
     /// Output batch type.
     type Output;
 
-    /// Allocates an empty builder.
-    ///
-    /// Ideally we deprecate this and insist all non-trivial building happens via `with_capacity()`.
-    // #[deprecated]
-    fn new() -> Self { Self::with_capacity(0, 0, 0) }
-    /// Allocates an empty builder with capacity for the specified keys, values, and updates.
-    ///
-    /// They represent respectively the number of distinct `key`, `(key, val)`, and total updates.
-    fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self;
     /// Adds a chunk of elements to the batch.
     ///
     /// Adds all elements from `chunk` to the builder and leaves `chunk` in an undefined state.
     fn push(&mut self, chunk: &mut Self::Input);
     /// Completes building and returns the batch, absent if no updates were pushed.
     fn done(self) -> Option<Self::Output>;
+}
+
+/// Forms a batch from a whole chain of updates at once.
+///
+/// Named rather than a bare `fn(&mut Vec<C>) -> Option<B>` so that implementors can name the
+/// batch they produce. There is no receiver: the chain goes in and the batch comes out, leaving
+/// nowhere for an update to be retained.
+pub trait Sealer<C> {
+    /// Output batch type.
+    type Output;
 
     /// Builds a batch from a chain of updates.
     ///
     /// This method relies on the chain only containing updates greater or equal to the lower frontier,
     /// and not greater or equal to the upper frontier, of the interval the caller means to describe.
     /// Chains must also be sorted and consolidated.
-    fn seal(chain: &mut Vec<Self::Input>) -> Option<Self::Output>;
+    ///
+    /// Having the whole chain in hand, an implementor can size itself before it fills.
+    fn seal(chain: &mut Vec<C>) -> Option<Self::Output>;
 }
 
 /// Blanket implementations for reference counted batches.