diff --git a/differential-dataflow/src/operators/arrange/upsert.rs b/differential-dataflow/src/operators/arrange/upsert.rs index 5d1f4f1ff..466ca4c5f 100644 --- a/differential-dataflow/src/operators/arrange/upsert.rs +++ b/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 = as Cursor>::KeyContainer::with_capacity(1); for (key, mut list) in to_process { diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs index a2d7f63a2..6d5eb734b 100644 --- a/differential-dataflow/src/operators/reduce.rs +++ b/differential-dataflow/src/operators/reduce.rs @@ -335,7 +335,7 @@ mod cursors { // 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::<(::ValOwn, TimeOf, ::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 @@ pub(crate) mod reference { let (mut batch_cursor, ref batch_storage) = cursor_list(input_batches); let mut output_updates = Vec::<(::ValOwn, TimeOf, ::Diff)>::new(); - let mut builder = Bu::new(); + let mut builder = Bu::default(); let mut buffer = Bu::Input::default(); // Reuseable state for performing the computation. diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index ccc126b37..ac9f32378 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -656,6 +656,12 @@ pub struct ChunkBatchBuilder { output: VecDeque, } +impl Default for ChunkBatchBuilder { + fn default() -> Self { + Self { input: VecDeque::new(), output: VecDeque::new() } + } +} + impl crate::trace::Builder for ChunkBatchBuilder where C: Chunk + Default + 'static, @@ -665,10 +671,6 @@ where type Time = C::Time; type Output = ChunkBatch; - 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 crate::trace::Sealer for ChunkBatchBuilder +where + C: Chunk + Default + 'static, + C::Time: timely::progress::Timestamp, +{ + type Output = ChunkBatch; + fn seal(chain: &mut Vec) -> Option { // We settle the chain because we are not guaranteed to received pre-settled data. // This should be efficient on pre-settled data. diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index b668d32d5..4ad588776 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/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 { +pub struct MergeBatcher { /// 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 { /// Timely operator ID. operator_id: usize, /// Seals each extracted chain into a batch. - builder: std::marker::PhantomData, + sealer: std::marker::PhantomData, } -impl Batcher for MergeBatcher +impl Batcher for MergeBatcher where M: Merger, Chu: ContainerBuilder + for<'a> PushInto<&'a mut C>, - Bu: Builder, + S: Sealer, { 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, AntichainRef<'a, M::Time>) { + fn extract<'a>(&'a mut self, upper: AntichainRef<'_, M::Time>) -> (Option, 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 MergeBatcher { +impl MergeBatcher { /// Allocates a new empty batcher. /// /// The logger and operator identifier are used to report the batcher's memory footprint, @@ -112,12 +112,12 @@ impl MergeBatcher { chains: Vec::new(), stash: Vec::new(), frontier: Antichain::new(), - builder: std::marker::PhantomData, + sealer: std::marker::PhantomData, } } } -impl MergeBatcher { +impl MergeBatcher { /// 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) { @@ -192,7 +192,7 @@ impl MergeBatcher { } } -impl Drop for MergeBatcher { +impl Drop for MergeBatcher { fn drop(&mut self) { // Cleanup chain to retract accounting information. while self.chain_pop().is_some() {} diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index 89ed2d15f..13012d5a5 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/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, } - impl Builder for OrdValBuilder + impl OrdValBuilder where - L: for<'a> Layout< - KeyContainer: PushInto>, - ValContainer: PushInto>, - >, - CI: for<'a> BuilderInput, Diff=layout::Diff>, + L: Layout, { - - type Input = CI; - type Time = layout::Time; - type Output = OrdValBatch; - + /// 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 Default for OrdValBuilder { + fn default() -> Self { Self::with_capacity(0, 0, 0) } + } + + impl Builder for OrdValBuilder + where + L: for<'a> Layout< + KeyContainer: PushInto>, + ValContainer: PushInto>, + >, + CI: for<'a> BuilderInput, Diff=layout::Diff>, + { + + type Input = CI; + type Time = layout::Time; + type Output = OrdValBatch; #[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) -> Option { - let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]); + } + + impl Sealer for OrdValBuilder + where + L: for<'a> Layout< + KeyContainer: PushInto>, + ValContainer: PushInto>, + >, + CI: for<'a> BuilderInput, Diff=layout::Diff>, + { + type Output = OrdValBatch; + + fn seal(chain: &mut Vec) -> Option { + 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, } - impl Builder for OrdKeyBuilder - where - L: for<'a> Layout>>, - L: Layout>, - CI: BuilderInput, Diff=layout::Diff>, - { - - type Input = CI; - type Time = layout::Time; - type Output = OrdKeyBatch; - + impl OrdKeyBuilder { + /// 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 Default for OrdKeyBuilder { + fn default() -> Self { Self::with_capacity(0, 0, 0) } + } + + impl Builder for OrdKeyBuilder + where + L: for<'a> Layout>>, + L: Layout>, + CI: BuilderInput, Diff=layout::Diff>, + { + + type Input = CI; + type Time = layout::Time; + type Output = OrdKeyBatch; #[inline] fn push(&mut self, chunk: &mut Self::Input) { @@ -1047,8 +1080,18 @@ pub mod key_batch { }) } - fn seal(chain: &mut Vec) -> Option { - let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]); + } + + impl Sealer for OrdKeyBuilder + where + L: for<'a> Layout>>, + L: Layout>, + CI: BuilderInput, Diff=layout::Diff>, + { + type Output = OrdKeyBatch; + + fn seal(chain: &mut Vec) -> Option { + 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); diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index 75a82c4a2..7b4e8f677 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -258,7 +258,10 @@ pub trait Batcher { } /// 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. @@ -266,28 +269,31 @@ pub trait Builder: Sized { /// 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; +} + +/// Forms a batch from a whole chain of updates at once. +/// +/// Named rather than a bare `fn(&mut Vec) -> Option` 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 { + /// 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) -> Option; + /// + /// Having the whole chain in hand, an implementor can size itself before it fills. + fn seal(chain: &mut Vec) -> Option; } /// Blanket implementations for reference counted batches.