From bc6b9c9aeffc9c0b477e06bc1460187b34d6c163 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 26 Aug 2026 19:48:54 -0400 Subject: [PATCH 1/4] Move Sealer to merge_batcher.rs --- differential-dataflow/src/trace/chunk/mod.rs | 2 +- .../trace/implementations/merge_batcher.rs | 21 ++++++++++++++++++- .../src/trace/implementations/ord_neu.rs | 7 ++++--- differential-dataflow/src/trace/mod.rs | 19 ----------------- 4 files changed, 25 insertions(+), 24 deletions(-) diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index ac9f32378..6bdb1ce71 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -688,7 +688,7 @@ where } -impl crate::trace::Sealer for ChunkBatchBuilder +impl crate::trace::implementations::merge_batcher::Sealer for ChunkBatchBuilder where C: Chunk + Default + 'static, C::Time: timely::progress::Timestamp, diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index 4ad588776..91695f58f 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -13,7 +13,7 @@ use timely::progress::frontier::AntichainRef; use timely::progress::{frontier::Antichain, Timestamp}; use crate::logging::{BatcherEvent, Logger}; -use crate::trace::{Batcher, Sealer}; +use crate::trace::Batcher; /// Creates batches from chunks of sorted, consolidated tuples. /// @@ -233,6 +233,25 @@ pub trait Merger: Default { fn allocation(_chunk: &Self::Chunk) -> (usize, usize, usize) { (0, 0, 0) } } +/// 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. + /// + /// Having the whole chain in hand, an implementor can size itself before it fills. + fn seal(chain: &mut Vec) -> Option; +} + /// A `Merger` implementation for vector update containers. pub mod vec { diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index 13012d5a5..e8a9cd628 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, Sealer, Cursor}; + use crate::trace::{Builder, Cursor}; use crate::trace::implementations::spine_fueled::{SpineBatch, Merger}; use crate::trace::implementations::{BatchContainer, BuilderInput}; use crate::trace::implementations::layout; @@ -694,7 +694,7 @@ pub mod val_batch { } - impl Sealer for OrdValBuilder + impl crate::trace::implementations::merge_batcher::Sealer for OrdValBuilder where L: for<'a> Layout< KeyContainer: PushInto>, @@ -724,10 +724,11 @@ pub mod key_batch { use timely::container::PushInto; use timely::progress::{Antichain, frontier::AntichainRef}; - use crate::trace::{Builder, Sealer, Cursor}; + use crate::trace::{Builder, Cursor}; use crate::trace::implementations::spine_fueled::{SpineBatch, Merger}; use crate::trace::implementations::{BatchContainer, BuilderInput}; use crate::trace::implementations::layout; + use crate::trace::implementations::merge_batcher::Sealer; use super::{Layout, Upds, layers::UpdsBuilder}; diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index 7b4e8f677..2a0b31140 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -277,25 +277,6 @@ pub trait Builder: Default { 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. - /// - /// 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. pub mod rc_blanket_impls { From bb95c036ef7da711d02e8a4c2afd95f3d9b2b006 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 26 Aug 2026 19:54:17 -0400 Subject: [PATCH 2/4] Move Rc cursor impls to cursor/mod.rs --- differential-dataflow/src/trace/cursor/mod.rs | 72 +++++++++++++++++++ differential-dataflow/src/trace/mod.rs | 71 ------------------ 2 files changed, 72 insertions(+), 71 deletions(-) diff --git a/differential-dataflow/src/trace/cursor/mod.rs b/differential-dataflow/src/trace/cursor/mod.rs index 4fc3c65d8..c82a2be41 100644 --- a/differential-dataflow/src/trace/cursor/mod.rs +++ b/differential-dataflow/src/trace/cursor/mod.rs @@ -181,3 +181,75 @@ pub trait Cursor { out } } + + +/// Blanket implementations for reference counted batches. +pub mod rc_blanket_impls { + + use std::rc::Rc; + + use super::{Navigable, Cursor}; + + impl Navigable for Rc { + /// The type used to enumerate the batch's contents. + type Cursor = RcBatchCursor; + /// Acquires a cursor to the batch's contents. + fn cursor(&self) -> Self::Cursor { + RcBatchCursor::new((**self).cursor()) + } + } + + /// Wrapper to provide cursor to nested scope. + pub struct RcBatchCursor { + cursor: C, + } + + impl RcBatchCursor { + fn new(cursor: C) -> Self { + RcBatchCursor { + cursor, + } + } + } + + impl Cursor for RcBatchCursor { + + type Storage = Rc; + + type Key<'a> = C::Key<'a>; + type ValOwn = C::ValOwn; + type Val<'a> = C::Val<'a>; + type Time = C::Time; + type TimeGat<'a> = C::TimeGat<'a>; + type Diff = C::Diff; + type DiffGat<'a> = C::DiffGat<'a>; + type KeyContainer = C::KeyContainer; + type ValContainer = C::ValContainer; + type TimeContainer = C::TimeContainer; + type DiffContainer = C::DiffContainer; + + #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) } + #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) } + + #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) } + #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) } + + #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_key(storage) } + #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_val(storage) } + + #[inline] + fn map_times, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L) { + self.cursor.map_times(storage, logic) + } + + #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) } + #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) } + + #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) } + #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) } + + #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) } + #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) } + } + +} diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index 2a0b31140..f4cd5fd26 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -276,74 +276,3 @@ pub trait Builder: Default { /// Completes building and returns the batch, absent if no updates were pushed. fn done(self) -> Option; } - -/// Blanket implementations for reference counted batches. -pub mod rc_blanket_impls { - - use std::rc::Rc; - - use super::{Navigable, Cursor}; - - impl Navigable for Rc { - /// The type used to enumerate the batch's contents. - type Cursor = RcBatchCursor; - /// Acquires a cursor to the batch's contents. - fn cursor(&self) -> Self::Cursor { - RcBatchCursor::new((**self).cursor()) - } - } - - /// Wrapper to provide cursor to nested scope. - pub struct RcBatchCursor { - cursor: C, - } - - impl RcBatchCursor { - fn new(cursor: C) -> Self { - RcBatchCursor { - cursor, - } - } - } - - impl Cursor for RcBatchCursor { - - type Storage = Rc; - - type Key<'a> = C::Key<'a>; - type ValOwn = C::ValOwn; - type Val<'a> = C::Val<'a>; - type Time = C::Time; - type TimeGat<'a> = C::TimeGat<'a>; - type Diff = C::Diff; - type DiffGat<'a> = C::DiffGat<'a>; - type KeyContainer = C::KeyContainer; - type ValContainer = C::ValContainer; - type TimeContainer = C::TimeContainer; - type DiffContainer = C::DiffContainer; - - #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) } - #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) } - - #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) } - #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) } - - #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_key(storage) } - #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option> { self.cursor.get_val(storage) } - - #[inline] - fn map_times, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L) { - self.cursor.map_times(storage, logic) - } - - #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) } - #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) } - - #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) } - #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) } - - #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) } - #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) } - } - -} From 743f989cd672c90a7170357ffd25068e6882e26a Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 27 Aug 2026 11:09:47 -0400 Subject: [PATCH 3/4] Relocate BuilderInput to ord_neu.rs --- .../src/trace/implementations/mod.rs | 79 +--------------- .../src/trace/implementations/ord_neu.rs | 90 ++++++++++++++++++- 2 files changed, 87 insertions(+), 82 deletions(-) diff --git a/differential-dataflow/src/trace/implementations/mod.rs b/differential-dataflow/src/trace/implementations/mod.rs index e6baa739d..50cd60779 100644 --- a/differential-dataflow/src/trace/implementations/mod.rs +++ b/differential-dataflow/src/trace/implementations/mod.rs @@ -56,8 +56,7 @@ pub use self::ord_neu::VecOrdKeyBuilder as KeyBuilder; use std::convert::TryInto; use serde::{Deserialize, Serialize}; -use timely::container::{DrainContainer, PushInto}; -use timely::progress::Timestamp; +use timely::container::PushInto; use crate::lattice::Lattice; use crate::difference::Semigroup; @@ -289,82 +288,6 @@ impl BatchContainer for OffsetList { } } -/// Behavior to split an update into principal components. -pub trait BuilderInput: DrainContainer + Sized { - /// Key portion - type Key<'a>: Ord; - /// Value portion - type Val<'a>: Ord; - /// Time - type Time; - /// Diff - type Diff; - - /// Split an item into separate parts. - fn into_parts<'a>(item: Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff); - - /// Test that the key equals a key in the layout's key container. - fn key_eq(this: &Self::Key<'_>, other: K::ReadItem<'_>) -> bool; - - /// Test that the value equals a key in the layout's value container. - fn val_eq(this: &Self::Val<'_>, other: V::ReadItem<'_>) -> bool; - - /// Count the number of distinct keys, (key, val) pairs, and total updates. - fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize); -} - -impl BuilderInput for Vec<((K, V), T, R)> -where - K: Ord + Clone + 'static, - KBC: for<'a> BatchContainer: PartialEq<&'a K>>, - V: Ord + Clone + 'static, - VBC: for<'a> BatchContainer: PartialEq<&'a V>>, - T: Timestamp + Lattice + 'static, - R: Ord + Semigroup + 'static, -{ - type Key<'a> = K; - type Val<'a> = V; - type Time = T; - type Diff = R; - - fn into_parts<'a>(((key, val), time, diff): Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff) { - (key, val, time, diff) - } - - fn key_eq(this: &K, other: KBC::ReadItem<'_>) -> bool { - KBC::reborrow(other) == this - } - - fn val_eq(this: &V, other: VBC::ReadItem<'_>) -> bool { - VBC::reborrow(other) == this - } - - fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize) { - let mut keys = 0; - let mut vals = 0; - let mut upds = 0; - let mut prev_keyval = None; - for link in chain.iter() { - for ((key, val), _, _) in link.iter() { - if let Some((p_key, p_val)) = prev_keyval { - if p_key != key { - keys += 1; - vals += 1; - } else if p_val != val { - vals += 1; - } - } else { - keys += 1; - vals += 1; - } - upds += 1; - prev_keyval = Some((key, val)); - } - } - (keys, vals, upds) - } -} - pub use self::containers::{BatchContainer, SliceContainer}; /// Containers for data that resemble `Vec`, with leaner implementations. diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index e8a9cd628..8e64ed589 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -250,10 +250,10 @@ pub mod val_batch { use crate::trace::{Builder, Cursor}; use crate::trace::implementations::spine_fueled::{SpineBatch, Merger}; - use crate::trace::implementations::{BatchContainer, BuilderInput}; + use crate::trace::implementations::{BatchContainer}; use crate::trace::implementations::layout; - use super::{Layout, Vals, Upds, layers::UpdsBuilder}; + use super::{Layout, Vals, Upds, layers::UpdsBuilder, BuilderInput}; /// An immutable collection of update tuples, from a contiguous interval of logical times. #[derive(Debug, Serialize, Deserialize)] @@ -726,11 +726,11 @@ pub mod key_batch { use crate::trace::{Builder, Cursor}; use crate::trace::implementations::spine_fueled::{SpineBatch, Merger}; - use crate::trace::implementations::{BatchContainer, BuilderInput}; + use crate::trace::implementations::BatchContainer; use crate::trace::implementations::layout; use crate::trace::implementations::merge_batcher::Sealer; - use super::{Layout, Upds, layers::UpdsBuilder}; + use super::{Layout, Upds, layers::UpdsBuilder, BuilderInput}; /// An immutable collection of update tuples, from a contiguous interval of logical times. #[derive(Debug, Serialize, Deserialize)] @@ -1103,3 +1103,85 @@ pub mod key_batch { } } + +use timely::container::DrainContainer; +use timely::progress::Timestamp; +use crate::trace::Lattice; +use crate::difference::Semigroup; +use crate::trace::implementations::BatchContainer; + +/// Behavior to split an update into principal components. +pub trait BuilderInput: DrainContainer + Sized { + /// Key portion + type Key<'a>: Ord; + /// Value portion + type Val<'a>: Ord; + /// Time + type Time; + /// Diff + type Diff; + + /// Split an item into separate parts. + fn into_parts<'a>(item: Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff); + + /// Test that the key equals a key in the layout's key container. + fn key_eq(this: &Self::Key<'_>, other: K::ReadItem<'_>) -> bool; + + /// Test that the value equals a key in the layout's value container. + fn val_eq(this: &Self::Val<'_>, other: V::ReadItem<'_>) -> bool; + + /// Count the number of distinct keys, (key, val) pairs, and total updates. + fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize); +} + +impl BuilderInput for Vec<((K, V), T, R)> +where + K: Ord + Clone + 'static, + KBC: for<'a> BatchContainer: PartialEq<&'a K>>, + V: Ord + Clone + 'static, + VBC: for<'a> BatchContainer: PartialEq<&'a V>>, + T: Timestamp + Lattice + 'static, + R: Ord + Semigroup + 'static, +{ + type Key<'a> = K; + type Val<'a> = V; + type Time = T; + type Diff = R; + + fn into_parts<'a>(((key, val), time, diff): Self::Item<'a>) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff) { + (key, val, time, diff) + } + + fn key_eq(this: &K, other: KBC::ReadItem<'_>) -> bool { + KBC::reborrow(other) == this + } + + fn val_eq(this: &V, other: VBC::ReadItem<'_>) -> bool { + VBC::reborrow(other) == this + } + + fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize) { + let mut keys = 0; + let mut vals = 0; + let mut upds = 0; + let mut prev_keyval = None; + for link in chain.iter() { + for ((key, val), _, _) in link.iter() { + if let Some((p_key, p_val)) = prev_keyval { + if p_key != key { + keys += 1; + vals += 1; + } else if p_val != val { + vals += 1; + } + } else { + keys += 1; + vals += 1; + } + upds += 1; + prev_keyval = Some((key, val)); + } + } + (keys, vals, upds) + } +} From 148dd1612d2cf89d9afbb35715e66a9216aa307e Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 27 Aug 2026 11:58:36 -0400 Subject: [PATCH 4/4] Relocate chunker.rs under merge_batcher/ --- differential-dataflow/benches/chunk_bench.rs | 2 +- differential-dataflow/examples/spines.rs | 2 +- differential-dataflow/src/trace/chunk/mod.rs | 2 +- .../src/trace/implementations/{ => merge_batcher}/chunker.rs | 2 +- .../{merge_batcher.rs => merge_batcher/mod.rs} | 4 +++- differential-dataflow/src/trace/implementations/mod.rs | 3 +-- differential-dataflow/src/trace/implementations/ord_neu.rs | 2 +- 7 files changed, 9 insertions(+), 8 deletions(-) rename differential-dataflow/src/trace/implementations/{ => merge_batcher}/chunker.rs (97%) rename differential-dataflow/src/trace/implementations/{merge_batcher.rs => merge_batcher/mod.rs} (99%) diff --git a/differential-dataflow/benches/chunk_bench.rs b/differential-dataflow/benches/chunk_bench.rs index 33e34f9cd..073fec3e3 100644 --- a/differential-dataflow/benches/chunk_bench.rs +++ b/differential-dataflow/benches/chunk_bench.rs @@ -26,7 +26,7 @@ use differential_dataflow::trace::chunk::{merge_chains, Chunk, NavigableChunk}; use differential_dataflow::trace::chunk::vec::VecChunk; use differential_dataflow::columnar::trace::ColChunk; use differential_dataflow::trace::cursor::Cursor; -use differential_dataflow::trace::implementations::chunker::ContainerChunker; +use differential_dataflow::trace::implementations::merge_batcher::chunker::ContainerChunker; /// A global allocator that tracks currently-resident bytes, so we can snapshot /// the heap footprint of a built chunk chain. diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs index f6606afdd..ac10400dc 100644 --- a/differential-dataflow/examples/spines.rs +++ b/differential-dataflow/examples/spines.rs @@ -81,7 +81,7 @@ fn main() { use differential_dataflow::Hashable; use differential_dataflow::columnar::trace::{Spine, ColChunk}; use differential_dataflow::trace::chunk::ChunkBatcher; - use differential_dataflow::trace::implementations::chunker::ContainerChunker; + use differential_dataflow::trace::implementations::merge_batcher::chunker::ContainerChunker; use differential_dataflow::operators::arrange::arrangement::arrange_core; use timely::dataflow::channels::pact::Exchange; diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index 6bdb1ce71..496a79dda 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -21,7 +21,7 @@ //! These are the `Batcher` / `Builder` / `Spine` to hand to //! [`arrange_core`](crate::operators::arrange::arrangement::arrange_core), along with a //! chunker that forms `C` from the input stream — typically -//! [`ContainerChunker`](crate::trace::implementations::chunker::ContainerChunker). +//! [`ContainerChunker`](crate::trace::implementations::merge_batcher::chunker::ContainerChunker). //! Trace *maintenance* needs only [`Chunk`]; cursor-driven *consumption* of the //! arrangement additionally asks `C` for the [`NavigableChunk`] capability. //! Everything else here ([`ChunkBatch`], [`ChunkMerger`], [`ChunkBatchMerger`], diff --git a/differential-dataflow/src/trace/implementations/chunker.rs b/differential-dataflow/src/trace/implementations/merge_batcher/chunker.rs similarity index 97% rename from differential-dataflow/src/trace/implementations/chunker.rs rename to differential-dataflow/src/trace/implementations/merge_batcher/chunker.rs index 4737838f2..7925e75ed 100644 --- a/differential-dataflow/src/trace/implementations/chunker.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher/chunker.rs @@ -1,4 +1,4 @@ -//! Organize streams of data into sorted chunks. +//! Organizes streams of data into sorted chunks for a merge batcher. use std::collections::VecDeque; diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher/mod.rs similarity index 99% rename from differential-dataflow/src/trace/implementations/merge_batcher.rs rename to differential-dataflow/src/trace/implementations/merge_batcher/mod.rs index 91695f58f..1f9182630 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher/mod.rs @@ -8,6 +8,8 @@ //! its `Chu` before merging: forming sorted, consolidated chunks is the first stage of the //! batcher's own work rather than something a caller arranges. +pub mod chunker; + use timely::container::{ContainerBuilder, PushInto}; use timely::progress::frontier::AntichainRef; use timely::progress::{frontier::Antichain, Timestamp}; @@ -421,7 +423,7 @@ mod test { use super::MergeBatcher; use super::vec::VecMerger; use crate::trace::implementations::ord_neu::VecOrdKeyBuilder; - use crate::trace::implementations::chunker::ContainerChunker; + use crate::trace::implementations::merge_batcher::chunker::ContainerChunker; type In = Vec<((u64, ()), u64, i64)>; type Bt = MergeBatcher, VecMerger<(u64, ()), u64, i64>, VecOrdKeyBuilder>; diff --git a/differential-dataflow/src/trace/implementations/mod.rs b/differential-dataflow/src/trace/implementations/mod.rs index 50cd60779..57e35deb1 100644 --- a/differential-dataflow/src/trace/implementations/mod.rs +++ b/differential-dataflow/src/trace/implementations/mod.rs @@ -42,10 +42,9 @@ pub mod spine_fueled; pub mod merge_batcher; pub mod ord_neu; -pub mod chunker; // Opinionated takes on default spines. -pub use self::chunker::ContainerChunker; +pub use self::merge_batcher::chunker::ContainerChunker; pub use self::ord_neu::OrdValSpine as ValSpine; pub use self::ord_neu::OrdValBatcher as ValBatcher; pub use self::ord_neu::VecOrdValBuilder as ValBuilder; diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index 8e64ed589..fa4136892 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -11,7 +11,7 @@ use std::rc::Rc; use crate::trace::implementations::spine_fueled::Spine; -use crate::trace::implementations::chunker::ContainerChunker; +use crate::trace::implementations::merge_batcher::chunker::ContainerChunker; use crate::trace::implementations::merge_batcher::MergeBatcher; use crate::trace::implementations::merge_batcher::vec::VecMerger;