From c8bf9873c004de5853dcb362c86e1fcb912fffb9 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 26 Aug 2026 16:20:52 -0400 Subject: [PATCH 1/2] Split `Builder` along the seam its two users already follow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Builder` had two consumers using disjoint halves of it. `MergeBatcher` called only `seal`, and through it `with_capacity`, because it holds a whole chain and can count keys, values, and updates before allocating. Reduce and upsert called only `new`, `push`, and `done`, discovering their output key by key and never sizing anything. `seal` moves to a trait of its own: pub trait Sealer { type Output; fn seal(chain: &mut Vec) -> Option; } which is a `fn(&mut Vec) -> Option` in a form that can name `B`. It has no receiver, so there is nowhere for an update to be retained between the chain going in and the batch coming out. `with_capacity` becomes inherent on the two `ord_neu` builders. It had no callers outside `seal` and `new`, so it was never vocabulary; `new` now carries its own one-line body instead of a default. `MergeBatcher` bounds `Bu: Sealer` and stops naming `Builder`, which also drops `Builder::Time` from its vocabulary — it never wanted it. The three implementors implement both traits and share their machinery. No call site changes: the aliases name the same types, and those types satisfy both bounds. This splits one question into two. Whether reduce and upsert should take a `Batcher` instead, and whether `Sealer` should be a function value rather than a trait, can now be answered independently. Co-Authored-By: Claude Opus 5 --- differential-dataflow/src/trace/chunk/mod.rs | 11 ++- .../trace/implementations/merge_batcher.rs | 10 +- .../src/trace/implementations/ord_neu.rs | 93 +++++++++++++------ differential-dataflow/src/trace/mod.rs | 23 +++-- 4 files changed, 95 insertions(+), 42 deletions(-) diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index ccc126b37..2d6d4264d 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -665,7 +665,7 @@ where type Time = C::Time; type Output = ChunkBatch; - fn with_capacity(_keys: usize, _vals: usize, _upds: usize) -> Self { + fn new() -> Self { Self { input: VecDeque::new(), output: VecDeque::new() } } @@ -684,6 +684,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..c2dee2cc3 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -13,11 +13,11 @@ 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 +/// Chunking input is `Chu`'s business, merging chunks is `M`'s, and sealing the extracted chain /// into a batch is `Bu`'s; the batcher's own work is the geometric ladder of chains and the /// carve-by-frontier. pub struct MergeBatcher { @@ -42,14 +42,14 @@ 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 where M: Merger, Chu: ContainerBuilder + for<'a> PushInto<&'a mut C>, - Bu: Builder, + Bu: Sealer, { type Time = M::Time; type Output = Bu::Output; @@ -112,7 +112,7 @@ impl MergeBatcher { chains: Vec::new(), stash: Vec::new(), frontier: Antichain::new(), - builder: std::marker::PhantomData, + sealer: std::marker::PhantomData, } } } diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index 89ed2d15f..97aecf76c 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,22 @@ pub mod val_batch { _marker: PhantomData, } } + } + + 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; + + fn new() -> Self { Self::with_capacity(0, 0, 0) } #[inline] fn push(&mut self, chunk: &mut Self::Input) { @@ -680,8 +690,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 +722,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,6 +1017,22 @@ pub mod key_batch { _marker: PhantomData, } + 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 { + keys: L::KeyContainer::with_capacity(keys), + upds: Upds::with_capacity(keys+1, upds), + }, + staging: UpdsBuilder::default(), + _marker: PhantomData, + } + } + } + impl Builder for OrdKeyBuilder where L: for<'a> Layout>>, @@ -1006,16 +1044,7 @@ pub mod key_batch { type Time = layout::Time; type Output = OrdKeyBatch; - fn with_capacity(keys: usize, _vals: usize, upds: usize) -> Self { - Self { - result: OrdKeyStorage { - keys: L::KeyContainer::with_capacity(keys), - upds: Upds::with_capacity(keys+1, upds), - }, - staging: UpdsBuilder::default(), - _marker: PhantomData, - } - } + fn new() -> Self { Self::with_capacity(0, 0, 0) } #[inline] fn push(&mut self, chunk: &mut Self::Input) { @@ -1047,8 +1076,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..70505e521 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -267,27 +267,32 @@ pub trait Builder: Sized { 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; + fn new() -> 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. From ebd8e495bc339612e1d7ecd6a4af3eab93b2f3b3 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 26 Aug 2026 16:27:53 -0400 Subject: [PATCH 2/2] `Builder: Default`, retiring `new` `new` was a second name for the empty builder, kept because the trait once offered `with_capacity` to prefer over it. It no longer does: sizing happens inside `Sealer::seal`, which owns its own allocation. So the preference the comment on `new` expressed is no longer expressible, or needed. `Default` says the same thing in the standard vocabulary. Reduce and upsert construct with `Bu::default()`. Co-Authored-By: Claude Opus 5 --- .../src/operators/arrange/upsert.rs | 2 +- differential-dataflow/src/operators/reduce.rs | 4 ++-- differential-dataflow/src/trace/chunk/mod.rs | 10 +++++---- .../trace/implementations/merge_batcher.rs | 22 +++++++++---------- .../src/trace/implementations/ord_neu.rs | 12 ++++++---- differential-dataflow/src/trace/mod.rs | 7 +++--- 6 files changed, 32 insertions(+), 25 deletions(-) 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 2d6d4264d..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 new() -> 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 { diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index c2dee2cc3..4ad588776 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -18,9 +18,9 @@ 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 sealing the extracted chain -/// into a batch is `Bu`'s; the batcher's own work is the geometric ladder of chains and the +/// 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. - sealer: 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: Sealer, + 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, @@ -117,7 +117,7 @@ impl MergeBatcher { } } -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 97aecf76c..13012d5a5 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -634,6 +634,10 @@ pub mod val_batch { } } + impl Default for OrdValBuilder { + fn default() -> Self { Self::with_capacity(0, 0, 0) } + } + impl Builder for OrdValBuilder where L: for<'a> Layout< @@ -647,8 +651,6 @@ pub mod val_batch { type Time = layout::Time; type Output = OrdValBatch; - fn new() -> Self { Self::with_capacity(0, 0, 0) } - #[inline] fn push(&mut self, chunk: &mut Self::Input) { for item in chunk.drain() { @@ -1033,6 +1035,10 @@ pub mod key_batch { } } + impl Default for OrdKeyBuilder { + fn default() -> Self { Self::with_capacity(0, 0, 0) } + } + impl Builder for OrdKeyBuilder where L: for<'a> Layout>>, @@ -1044,8 +1050,6 @@ pub mod key_batch { type Time = layout::Time; type Output = OrdKeyBatch; - fn new() -> Self { Self::with_capacity(0, 0, 0) } - #[inline] fn push(&mut self, chunk: &mut Self::Input) { for item in chunk.drain() { diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index 70505e521..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,8 +269,6 @@ pub trait Builder: Sized { /// Output batch type. type Output; - /// Allocates an empty builder. - fn new() -> Self; /// Adds a chunk of elements to the batch. /// /// Adds all elements from `chunk` to the builder and leaves `chunk` in an undefined state.