From 9b7c2cb458d07ea068df57c9843ce1a7372e42a6 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 26 Aug 2026 10:46:40 -0400 Subject: [PATCH 1/3] The batcher seals its own batches `arrange_core` used to take both a batcher and a builder, pull a chain of chunks out of the batcher, and call `Bu::seal` on it to get a batch. The batcher named the batch type in its output, but did not produce one. Now `MergeBatcher` carries the builder and seals internally, so `extract` returns a batch. `arrange_core` drops its `Bu` parameter: the bounds go from Ba: Batcher>, Bu: Builder>, to Ba: Batcher, B: Into, Arrange no longer knows that builders exist; it asks for updates and gets something it can turn into a batch. Call sites stop naming a builder, which is the bulk of the diff. The builder is `PhantomData` on `MergeBatcher` because `seal` is an associated function. Sealing still happens at extraction with the whole chain in hand, so `Builder::with_capacity` still pre-sizes from the full key, value, and update counts. Co-Authored-By: Claude Opus 5 --- .../examples/columnar/main.rs | 6 ++-- .../examples/columnar_spill.rs | 4 +-- differential-dataflow/examples/spines.rs | 26 +++++++------- differential-dataflow/src/collection.rs | 36 +++++++++---------- .../src/operators/arrange/arrangement.rs | 16 ++++----- differential-dataflow/src/trace/chunk/mod.rs | 2 +- .../trace/implementations/merge_batcher.rs | 33 ++++++++++------- .../src/trace/implementations/ord_neu.rs | 4 +-- differential-dataflow/tests/columnar.rs | 4 +-- differential-dataflow/tests/int_proxy.rs | 10 +++--- .../tests/int_proxy_bench.rs | 6 ++-- differential-dataflow/tests/trace.rs | 8 ++--- experiments/src/bin/deals.rs | 10 +++--- experiments/src/bin/graspan1.rs | 6 ++-- experiments/src/bin/graspan2.rs | 32 ++++++++--------- interactive/src/backend/corgi.rs | 4 +-- 16 files changed, 104 insertions(+), 103 deletions(-) diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs index d566c2424..8323653d0 100644 --- a/differential-dataflow/examples/columnar/main.rs +++ b/differential-dataflow/examples/columnar/main.rs @@ -130,14 +130,14 @@ mod reachability { let edges_arr = arrange_core::<_, _, ValChunker<(Node, Node, IterTime, Diff)>, _, - ValBuilder, + _, ValSpine, >(edges_inner.inner, edges_pact, "Edges", ValBatcher::new); let reach_arr = arrange_core::<_, _, ValChunker<(Node, (), IterTime, Diff)>, _, - ValBuilder, + _, ValSpine, >(reach.inner, reach_pact, "Reach", ValBatcher::new); @@ -163,7 +163,7 @@ mod reachability { let combined_arr = arrange_core::<_, _, ValChunker<(Node, (), IterTime, Diff)>, _, - ValBuilder, + _, ValSpine, >(combined.inner, combined_pact, "Combined", ValBatcher::new); diff --git a/differential-dataflow/examples/columnar_spill.rs b/differential-dataflow/examples/columnar_spill.rs index c5619a08b..712955bb4 100644 --- a/differential-dataflow/examples/columnar_spill.rs +++ b/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, Chunker as ValChunker, 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}; @@ -240,7 +240,7 @@ fn run_timely_dataflow(times: u64, keys_per_time: u64, workers: usize, sample_se _, _, ValChunker<(u64, u64, u64, i64)>, _, - ValBuilder, + _, ValSpine, >(stream, Pipeline, "ColumnarSpillArrange", ValBatcher::new); arranged.stream.probe_with(&mut probe); diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs index e7bdb630f..62a0cf77c 100644 --- a/differential-dataflow/examples/spines.rs +++ b/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::(); let (keys_input, keys) = scope.new_collection::(); - let data = data.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder, OrdKeySpine>(OrdKeyBatcher::new); - let keys = keys.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder, OrdKeySpine>(OrdKeyBatcher::new); + let data = data.map(|k| (k, ())).arrange::<_, _, OrdKeySpine>(OrdKeyBatcher::new); + let keys = keys.map(|k| (k, ())).arrange::<_, _, OrdKeySpine>(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::(); let (keys_input, keys) = scope.new_collection::(); - let data = data.map(|x| (x, ())).arrange::<_, VecOrdValBuilder, OrdValSpine>(OrdValBatcher::new); - let keys = keys.map(|x| (x, ())).arrange::<_, VecOrdValBuilder, OrdValSpine>(OrdValBatcher::new); + let data = data.map(|x| (x, ())).arrange::<_, _, OrdValSpine>(OrdValBatcher::new); + let keys = keys.map(|x| (x, ())).arrange::<_, _, OrdValSpine>(OrdValBatcher::new); keys.join_core(data, |_k, &(), &()| Option::<()>::None) .probe_with(&mut probe); Workload { data_input, keys_input } @@ -79,7 +79,7 @@ fn main() { // the same generic `Chunk` harness as `vec` via a // `ContainerChunker`. use differential_dataflow::Hashable; - use differential_dataflow::columnar::trace::{Batcher, Builder, Spine, ColChunk}; + use differential_dataflow::columnar::trace::{Batcher, Spine, ColChunk}; use differential_dataflow::trace::implementations::chunker::ContainerChunker; use differential_dataflow::operators::arrange::arrangement::arrange_core; use timely::dataflow::channels::pact::Exchange; @@ -90,12 +90,11 @@ fn main() { let keys = keys.map(|x| (x, ())); type Ba = Batcher; - type Bu = Builder; type Sp = Spine; type Chu = ContainerChunker>; 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::<_, _, Chu, Ba, _, Sp>(data.inner, exchange(), "DataArrange", Ba::new); + let keys = arrange_core::<_, _, Chu, 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,7 +105,7 @@ fn main() { // insert allocates a `String`) but arranged through the `Chunk` // harness via a `ContainerChunker`. use differential_dataflow::Hashable; - use differential_dataflow::trace::chunk::vec::{ChunkBatcher, ChunkBuilder, ChunkSpine, VecChunk}; + use differential_dataflow::trace::chunk::vec::{ChunkBatcher, ChunkSpine, VecChunk}; use differential_dataflow::trace::implementations::chunker::ContainerChunker; use differential_dataflow::operators::arrange::arrangement::arrange_core; use timely::dataflow::channels::pact::Exchange; @@ -117,12 +116,11 @@ fn main() { let keys = keys.map(|x| (x, ())); type Ba = ChunkBatcher; - type Bu = ChunkBuilder; type Sp = ChunkSpine; type Chu = ContainerChunker>; 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::<_, _, Chu, Ba, _, Sp>(data.inner, exchange(), "DataArrange", Ba::new); + let keys = arrange_core::<_, _, Chu, Ba, _, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new); keys.join_core(data, |_k, &(), &()| Option::<()>::None) .probe_with(&mut probe); Workload { data_input, keys_input } diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs index a77b05267..4e2be6e8e 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -957,22 +957,22 @@ 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(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba, reify: F) -> Self + pub fn consolidate_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba, reify: F) -> Self where - Ba: crate::trace::Batcher, Vec>> + 'static, + Ba: crate::trace::Batcher, B> + 'static, + B: Into, Tr: crate::trace::Trace+'static, for<'a> BatchCursor: Cursor, - Bu: crate::trace::Builder, Output: Into>, F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static, { self.map(|k| (k, ())) - .arrange_named::(name, batcher) + .arrange_named::(name, batcher) .as_collection(reify) } @@ -1020,8 +1020,8 @@ pub mod vec { } } - use crate::trace::implementations::{ValSpine, ValBatcher, ValBuilder}; - use crate::trace::implementations::{KeySpine, KeyBatcher, KeyBuilder}; + use crate::trace::implementations::{ValSpine, ValBatcher}; + use crate::trace::implementations::{KeySpine, KeyBatcher}; use crate::trace::implementations::ContainerChunker; impl<'scope, T, K, V, R> Collection<'scope, T, (K, V), R> where @@ -1036,24 +1036,24 @@ 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(self, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> + pub fn arrange(self, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> where - Ba: crate::trace::Batcher, Vec> + 'static, - Bu: crate::trace::Builder>, + Ba: crate::trace::Batcher, B> + 'static, + B: Into, Tr: crate::trace::Trace + 'static, { - self.arrange_named::("Arrange", batcher) + self.arrange_named::("Arrange", batcher) } /// As [`Collection::arrange`] but with the ability to name the operator. - pub fn arrange_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> + pub fn arrange_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> where - Ba: crate::trace::Batcher, Vec> + 'static, - Bu: crate::trace::Builder>, + Ba: crate::trace::Batcher, B> + 'static, + B: Into, Tr: crate::trace::Trace + '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>, Ba, Bu, _>(self.inner, exchange, name, batcher) + crate::operators::arrange::arrangement::arrange_core::<_, _, ContainerChunker>, Ba, B, _>(self.inner, exchange, name, batcher) } } @@ -1072,7 +1072,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>> { - self.arrange_named::<_,ValBuilder<_,_,_,_>,_>(name, ValBatcher::new) + self.arrange_named::<_,_,_>(name, ValBatcher::new) } } @@ -1092,7 +1092,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>> { self.map(|k| (k, ())) - .arrange_named::<_,KeyBuilder<_,_,_>,_>(name, KeyBatcher::new) + .arrange_named::<_,_,_>(name, KeyBatcher::new) } } diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index 3d1d9264e..db8fcc4c2 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -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, Chu, Ba, B, Tr>( stream: Stream<'scope, Tr::Time, C>, pact: P, name: &str, @@ -313,8 +313,8 @@ where C: Container + Clone + 'static, P: ParallelizationContract, Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static, - Ba: Batcher> + 'static, - Bu: Builder>, + Ba: Batcher + 'static, + B: Into, Tr: Trace+'static, { // The `Arrange` operator is tasked with reacting to an advancing input @@ -433,9 +433,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::>(); writer.insert(batch.clone(), stamp); @@ -444,11 +444,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 +457,6 @@ where panic!("failed to find capability"); } } - capabilities = new_capabilities; } else { diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index b1aecc244..f900778ec 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -277,7 +277,7 @@ where /// using [`Chunk::extract`]. The batcher consolidates equal `(data, time)` updates /// but does *not* advance times — time advancement is advance's job, handled later in /// the trace. Both settle their output, since the batcher's chains want to be graded. -pub type ChunkBatcher = crate::trace::implementations::merge_batcher::MergeBatcher>; +pub type ChunkBatcher = crate::trace::implementations::merge_batcher::MergeBatcher, ChunkBuilder>; /// A spine of `Rc`-shared [`ChunkBatch`]s of type `C`: the trace type for `arrange`. pub type ChunkSpine = crate::trace::implementations::spine_fueled::Spine>>; diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index 1606a15cc..7720c14e5 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -12,10 +12,13 @@ use timely::progress::frontier::AntichainRef; use timely::progress::{frontier::Antichain, Timestamp}; use crate::logging::{BatcherEvent, Logger}; -use crate::trace::Batcher; +use crate::trace::{Batcher, Builder}; /// Creates batches from chunks of sorted, consolidated tuples. -pub struct MergeBatcher { +/// +/// Merging is `M`'s business and building 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 { /// Sorted, consolidated chains, each paired with its cached summed update count. /// /// The cached count is the chain's *merge weight*: the geometric ladder weighs @@ -34,11 +37,14 @@ pub struct MergeBatcher { logger: Option, /// Timely operator ID. operator_id: usize, + /// Seals each extracted chain into a batch. + builder: std::marker::PhantomData, } -impl Batcher> for MergeBatcher +impl Batcher for MergeBatcher where M: Merger, + Bu: Builder, { fn insert(&mut self, chunk: &mut M::Chunk) { self.insert_chain(vec![std::mem::take(chunk)]); @@ -48,7 +54,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>) { // Merge all remaining chains into a single chain. while self.chains.len() > 1 { let list1 = self.chain_pop().unwrap(); @@ -71,12 +77,11 @@ where self.stash.clear(); - let readied = if readied.is_empty() { None } else { Some(readied) }; - (readied, self.frontier.borrow()) + (Bu::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, @@ -89,6 +94,7 @@ impl MergeBatcher { chains: Vec::new(), stash: Vec::new(), frontier: Antichain::new(), + builder: std::marker::PhantomData, } } @@ -166,7 +172,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() {} @@ -375,8 +381,9 @@ mod test { use crate::trace::Batcher; use super::MergeBatcher; use super::vec::VecMerger; + use crate::trace::implementations::ord_neu::VecOrdKeyBuilder; - type Bt = MergeBatcher>; + type Bt = MergeBatcher, VecOrdKeyBuilder>; /// The sealed frontier must reflect the POST-CONSOLIDATION set of distinct kept times: /// two chains carry cancelling updates at a kept time (`t=5`), plus a survivor at a later @@ -386,8 +393,8 @@ mod test { #[test] fn frontier_is_post_consolidation() { let mut b = Bt::new(None, 0); - b.chain_push(vec![vec![(100u64, 5u64, 1i64), (200u64, 7u64, 1i64)]]); - b.chain_push(vec![vec![(100u64, 5u64, -1i64)]]); + b.chain_push(vec![vec![((100u64, ()), 5u64, 1i64), ((200u64, ()), 7u64, 1i64)]]); + b.chain_push(vec![vec![((100u64, ()), 5u64, -1i64)]]); let (_, retained) = b.extract(Antichain::from_elem(3).borrow()); let got: Vec = retained.iter().cloned().collect(); assert_eq!(got, vec![7u64], @@ -398,8 +405,8 @@ mod test { #[test] fn frontier_survivor_minimum() { let mut b = Bt::new(None, 0); - b.chain_push(vec![vec![(100u64, 5u64, 1i64)]]); - b.chain_push(vec![vec![(200u64, 7u64, 1i64)]]); + b.chain_push(vec![vec![((100u64, ()), 5u64, 1i64)]]); + b.chain_push(vec![vec![((200u64, ()), 7u64, 1i64)]]); let (_, retained) = b.extract(Antichain::from_elem(3).borrow()); let got: Vec = retained.iter().cloned().collect(); assert_eq!(got, vec![5u64]); diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index 81e396b9e..5ab540db0 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -22,14 +22,14 @@ pub use self::key_batch::{OrdKeyBatch, OrdKeyBuilder}; /// A trace implementation using a spine of ordered lists. pub type OrdValSpine = Spine>>>; /// A batcher using ordered lists. -pub type OrdValBatcher = MergeBatcher>; +pub type OrdValBatcher = MergeBatcher, VecOrdValBuilder>; /// A builder using ordered lists. pub type VecOrdValBuilder = OrdValBuilder, Vec<((K,V),T,R)>>; /// A trace implementation using a spine of ordered lists. pub type OrdKeySpine = Spine>>>; /// A batcher for ordered lists. -pub type OrdKeyBatcher = MergeBatcher>; +pub type OrdKeyBatcher = MergeBatcher, VecOrdKeyBuilder>; /// A builder for ordered lists. pub type VecOrdKeyBuilder = OrdKeyBuilder, Vec<((K,()),T,R)>>; diff --git a/differential-dataflow/tests/columnar.rs b/differential-dataflow/tests/columnar.rs index 5001d01be..3428464c3 100644 --- a/differential-dataflow/tests/columnar.rs +++ b/differential-dataflow/tests/columnar.rs @@ -5,7 +5,7 @@ use timely::dataflow::{InputHandle, ProbeHandle}; use timely::Config; use differential_dataflow::columnar::collection; -use differential_dataflow::columnar::trace::{Batcher, Builder, Chunker, Spine}; +use differential_dataflow::columnar::trace::{Batcher, Chunker, Spine}; use differential_dataflow::operators::arrange::arrangement::arrange_core; type Upd = (u64, (), u64, i64); @@ -31,7 +31,7 @@ fn arrange_reaches_one(config: Config, diffs: &'static [i64]) -> bool { _, Chunker, _, - Builder, + _, Spine, >(stream, pact, "Arrange", Batcher::new) .stream diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index a71fca87d..4aeccf15e 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -24,7 +24,7 @@ use differential_dataflow::operators::int_proxy::vec_backend::VecReduceBackend; use differential_dataflow::operators::iterate::Iterate; use differential_dataflow::operators::reduce::{reduce_with_tactic, ReduceTactic}; use differential_dataflow::trace::chunk::vec::{ - ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine, VecChunk, + ChunkBatcher as VChunkBatcher, ChunkSpine as VChunkSpine, VecChunk, }; use differential_dataflow::trace::chunk::ChunkBatch; use differential_dataflow::trace::cursor::Cursor; @@ -139,7 +139,7 @@ fn proxy_matches_mainline(updates: Vec<((u64, u64), u64, i64)>, window: usize) { worker.dataflow::(|scope| { let coll = updates.clone().to_stream(scope).as_collection(); let hashed = coll.clone().map(|(k, v)| (k.hashed(), (k, v))); - let arr = arrange_core::, ContainerChunker>, _, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); + let arr = arrange_core::, ContainerChunker>, _, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, window))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) .inspect(move |(d, t, r)| ts.lock().unwrap().push((*d, *t, *r))); @@ -202,7 +202,7 @@ fn reduce_string_values_matches_mainline() { worker.dataflow::(|scope| { let coll = updates.clone().to_stream(scope).as_collection(); let hashed = coll.clone().map(|(k, v): (u64, String)| (k.hashed(), (k, v))); - let arr = arrange_core::, ContainerChunker>, _, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); + let arr = arrange_core::, ContainerChunker>, _, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduceStr", ProxyReduceTactic::new(VecReduceBackend::with_window( |_k: &u64, input: &[(String, i64)], current: &mut Vec<(String, i64)>, updates: &mut Vec<(String, i64)>| { if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { updates.push((m, 1)); } @@ -236,7 +236,7 @@ fn reduce_inside_iterate() { let input = updates.clone().to_stream(scope).as_collection(); let result = input.iterate(|_scope, inner| { let hashed = inner.map(|(k, v)| (k.hashed(), (k, v))); - let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkBuilder, i64>, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "IterReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) }); @@ -311,7 +311,7 @@ fn reduce_collision_inside_iterate() { let input = updates.clone().to_stream(scope).as_collection(); let result = input.iterate(|_scope, inner| { let hashed = inner.map(|(k, v)| (k % 2, (k, v))); - let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkBuilder, i64>, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "CollideReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) }); diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs index c3942300b..9c21dccfb 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -58,7 +58,7 @@ macro_rules! harrange { Vec<((u64, (u64, u64)), $t, isize)>, ContainerChunker>, _, - VChunkBuilder, + _, VChunkSpine, >(hashed.inner, Pipeline, $name, VChunkBatcher::new) }}; @@ -149,7 +149,7 @@ fn run_wide(mode: Mode) -> f64 { let arr = arrange_core::, ContainerChunker>, _, - VChunkBuilder, + _, VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); arr.reduce_core::<_, VChunkBuilder, VChunkSpine, @@ -173,7 +173,7 @@ fn run_wide(mode: Mode) -> f64 { let arr = arrange_core::, ContainerChunker>, _, - VChunkBuilder, + _, VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>( arr, "ProxyWide", diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs index 7a9aea117..98abf69ee 100644 --- a/differential-dataflow/tests/trace.rs +++ b/differential-dataflow/tests/trace.rs @@ -1,12 +1,11 @@ use timely::dataflow::operators::generic::OperatorInfo; use timely::progress::{Antichain, frontier::AntichainRef}; -use differential_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine}; -use differential_dataflow::trace::{Description, Span, Trace, TraceReader, Batcher, Builder}; +use differential_dataflow::trace::implementations::{ValBatcher, ValSpine}; +use differential_dataflow::trace::{Description, Span, Trace, TraceReader, Batcher}; use differential_dataflow::trace::cursor::{Cursor, cursor_list}; type IntegerTrace = ValSpine; -type IntegerBuilder = ValBuilder; fn get_trace() -> ValSpine { let op_info = OperatorInfo::new(0, 0, [].into()); @@ -24,9 +23,8 @@ fn get_trace() -> ValSpine { let mut lower = Antichain::from_elem(0); for i in batch_ts { let upper = Antichain::from_elem(*i); - let (chain, _retained) = batcher.extract(upper.borrow()); + let (batch, _retained) = batcher.extract(upper.borrow()); let description = Description::new(lower, upper.clone(), Antichain::from_elem(0)); - let batch = chain.and_then(|mut chain| IntegerBuilder::seal(&mut chain)); trace.insert(Span::new(description, batch.map(Into::into))); lower = upper; } diff --git a/experiments/src/bin/deals.rs b/experiments/src/bin/deals.rs index a6502baf4..392ecec5d 100644 --- a/experiments/src/bin/deals.rs +++ b/experiments/src/bin/deals.rs @@ -4,7 +4,7 @@ use differential_dataflow::input::Input; use differential_dataflow::VecCollection; use differential_dataflow::operators::*; -use differential_dataflow::trace::implementations::{ValSpine, ValBatcher, ValBuilder}; +use differential_dataflow::trace::implementations::{ValSpine, ValBatcher}; use differential_dataflow::operators::arrange::TraceAgent; use differential_dataflow::operators::arrange::Arranged; use differential_dataflow::operators::iterate::Variable; @@ -39,7 +39,7 @@ fn main() { let (input, graph) = scope.new_collection(); // each edge should exist in both directions. - let graph = graph.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + let graph = graph.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); match program.as_str() { "tc" => tc(graph.clone()).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("tc count: {:?}", x)).probe(), @@ -93,7 +93,7 @@ fn tc<'s, T: timely::progress::Timestamp + Lattice + Default + timely::order::Em let result = inner_collection .map(|(x,y)| (y,x)) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(edges.clone(), |_y,&x,&z| Some((x, z))) .concat(edges.as_collection(|&k,&v| (k,v))) .arrange_by_self() @@ -121,9 +121,9 @@ fn sg<'s, T: timely::progress::Timestamp + Lattice + Default + timely::order::Em let result = inner_collection - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(edges.clone(), |_,&x,&z| Some((x, z))) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(edges, |_,&x,&z| Some((x, z))) .concat(peers) .arrange_by_self() diff --git a/experiments/src/bin/graspan1.rs b/experiments/src/bin/graspan1.rs index 28e96c818..4b443f6a9 100644 --- a/experiments/src/bin/graspan1.rs +++ b/experiments/src/bin/graspan1.rs @@ -5,7 +5,7 @@ use timely::order::Product; use differential_dataflow::difference::Present; use differential_dataflow::input::Input; -use differential_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine}; +use differential_dataflow::trace::implementations::{ValBatcher, ValSpine}; use differential_dataflow::operators::*; use differential_dataflow::operators::iterate::Variable; @@ -29,7 +29,7 @@ fn main() { let (n_handle, nodes) = scope.new_collection(); let (e_handle, edges) = scope.new_collection(); - let edges = edges.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + let edges = edges.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); // a N c <- a N b && b E c // N(a,c) <- N(a,b), E(b, c) @@ -46,7 +46,7 @@ fn main() { labels_collection.join_core(edges, |_b, a, c| Some((*c, *a))) .concat(nodes) .map(|k| (k, ())) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) // .distinct_total_core::(); .threshold_semigroup(|_,_,x: Option<&Present>| if x.is_none() { Some(Present) } else { None }); diff --git a/experiments/src/bin/graspan2.rs b/experiments/src/bin/graspan2.rs index d087c7a7b..71621b94c 100644 --- a/experiments/src/bin/graspan2.rs +++ b/experiments/src/bin/graspan2.rs @@ -8,7 +8,7 @@ use differential_dataflow::operators::iterate::Variable; use differential_dataflow::VecCollection; use differential_dataflow::input::Input; use differential_dataflow::operators::*; -use differential_dataflow::trace::implementations::{ValSpine, ValBatcher, ValBuilder}; +use differential_dataflow::trace::implementations::{ValSpine, ValBatcher}; use differential_dataflow::difference::Present; type Node = u32; @@ -45,7 +45,7 @@ fn unoptimized() { .flat_map(|(a,b)| vec![a,b]) .concat(dereference.clone().flat_map(|(a,b)| vec![a,b])); - let dereference = dereference.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + let dereference = dereference.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); let (value_flow, memory_alias, value_alias) = scope @@ -58,14 +58,14 @@ fn unoptimized() { let (value_flow, value_flow_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); let (memory_alias, memory_alias_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); - let value_flow_arranged = value_flow_collection.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); - let memory_alias_arranged = memory_alias_collection.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + let value_flow_arranged = value_flow_collection.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let memory_alias_arranged = memory_alias_collection.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); // VA(a,b) <- VF(x,a),VF(x,b) // VA(a,b) <- VF(x,a),MA(x,y),VF(y,b) let value_alias_next = value_flow_arranged.clone().join_core(value_flow_arranged.clone(), |_,&a,&b| Some((a,b))); let value_alias_next = value_flow_arranged.clone().join_core(memory_alias_arranged.clone(), |_,&a,&b| Some((b,a))) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_arranged.clone(), |_,&a,&b| Some((a,b))) .concat(value_alias_next); @@ -75,10 +75,10 @@ fn unoptimized() { let value_flow_next = assignment.clone() .map(|(a,b)| (b,a)) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(memory_alias_arranged, |_,&a,&b| Some((b,a))) .concat(assignment.map(|(a,b)| (b,a))) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_arranged, |_,&a,&b| Some((a,b))) .concat(nodes.map(|n| (n,n))); @@ -93,7 +93,7 @@ fn unoptimized() { let memory_alias_next: VecCollection<_,_,Present> = value_alias_next.clone() .join_core(dereference.clone(), |_x,&y,&a| Some((y,a))) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(dereference, |_y,&a,&b| Some((a,b))); let memory_alias_next: VecCollection<_,_,Present> = @@ -170,7 +170,7 @@ fn optimized() { .flat_map(|(a,b)| vec![a,b]) .concat(dereference.clone().flat_map(|(a,b)| vec![a,b])); - let dereference = dereference.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + let dereference = dereference.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); let (value_flow, memory_alias) = scope @@ -183,8 +183,8 @@ fn optimized() { let (value_flow, value_flow_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); let (memory_alias, memory_alias_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); - let value_flow_arranged = value_flow_collection.clone().arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); - let memory_alias_arranged = memory_alias_collection.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + let value_flow_arranged = value_flow_collection.clone().arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let memory_alias_arranged = memory_alias_collection.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); // VF(a,a) <- // VF(a,b) <- A(a,x),VF(x,b) @@ -192,10 +192,10 @@ fn optimized() { let value_flow_next = assignment.clone() .map(|(a,b)| (b,a)) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(memory_alias_arranged.clone(), |_,&a,&b| Some((b,a))) .concat(assignment.map(|(a,b)| (b,a))) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_arranged, |_,&a,&b| Some((a,b))) .concat(nodes.map(|n| (n,n))) .arrange_by_self() @@ -207,9 +207,9 @@ fn optimized() { let value_flow_deref = value_flow_collection .map(|(a,b)| (b,a)) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(dereference, |_x,&a,&b| Some((a,b))) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); // MA(a,b) <- VFD(x,a),VFD(y,b) // MA(a,b) <- VFD(x,a),MA(x,y),VFD(y,b) @@ -220,7 +220,7 @@ fn optimized() { let memory_alias_next = memory_alias_arranged .join_core(value_flow_deref.clone(), |_x,&y,&a| Some((y,a))) - .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_deref, |_y,&a,&b| Some((a,b))) .concat(memory_alias_next) .arrange_by_self() diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index d3cedb101..5f06ad5e6 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -18,7 +18,7 @@ use differential_dataflow::operators::join::join_with_tactic; use differential_dataflow::operators::reduce::reduce_with_tactic; use differential_dataflow::operators::arrange::arrangement::arrange_core; use differential_dataflow::operators::arrange::{Arranged, TraceAgent}; -use differential_dataflow::trace::chunk::{Chunk, ChunkBatcher, ChunkBuilder}; +use differential_dataflow::trace::chunk::{Chunk, ChunkBatcher}; use corgi::arrange::gather; use corgi::Value as CValue; @@ -295,7 +295,7 @@ impl Backend for CorgiBackend { // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher + // builder. No columns→rows→columns round-trip at the arrangement boundary. - arrange_core::<_, CC, CorgiChunker, _, ChunkBuilder>, CTrace>( + arrange_core::<_, CC, CorgiChunker, _, _, CTrace>( c.inner, Pipeline, "CorgiArrange", From f74efcf5d6f75f76a464c960ac21dd36e979fe8f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 26 Aug 2026 10:56:53 -0400 Subject: [PATCH 2/3] `Batcher`'s time and output become associated types The input a batcher accepts stays a parameter, because a batcher may reasonably accept several shapes of container. What it produces, and the timestamps it carves by, are facts about the batcher, so they become associated types. Neither impl in the tree implements `Batcher` twice for one type, so the parameters were buying no polymorphism. -pub trait Batcher { +pub trait Batcher { + type Time; + type Output; Callers that want to pin the output still can, as half_join does with `Output = Vec`. Callers that only need it convertible say so directly: - Ba: Batcher + 'static, - B: Into, + Ba: Batcher> + 'static, That drops a type parameter from `arrange_core`, `Collection::arrange`, `arrange_named`, and `consolidate_named`, and with it the anonymous `_` that every call site had to write in its turbofish. Co-Authored-By: Claude Opus 5 --- .../examples/columnar/main.rs | 3 -- .../examples/columnar_spill.rs | 1 - differential-dataflow/examples/spines.rs | 16 +++++----- differential-dataflow/src/collection.rs | 27 ++++++++--------- .../src/operators/arrange/arrangement.rs | 5 ++-- .../trace/implementations/merge_batcher.rs | 5 +++- differential-dataflow/src/trace/mod.rs | 15 ++++++---- differential-dataflow/tests/columnar.rs | 1 - differential-dataflow/tests/int_proxy.rs | 8 ++--- .../tests/int_proxy_bench.rs | 3 -- dogsdogsdogs/src/operators/half_join.rs | 7 +++-- experiments/src/bin/deals.rs | 8 ++--- experiments/src/bin/graspan1.rs | 4 +-- experiments/src/bin/graspan2.rs | 30 +++++++++---------- interactive/src/backend/corgi.rs | 2 +- 15 files changed, 67 insertions(+), 68 deletions(-) diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs index 8323653d0..2f5fd2f9c 100644 --- a/differential-dataflow/examples/columnar/main.rs +++ b/differential-dataflow/examples/columnar/main.rs @@ -130,14 +130,12 @@ mod reachability { let edges_arr = arrange_core::<_, _, ValChunker<(Node, Node, IterTime, Diff)>, _, - _, ValSpine, >(edges_inner.inner, edges_pact, "Edges", ValBatcher::new); let reach_arr = arrange_core::<_, _, ValChunker<(Node, (), IterTime, Diff)>, _, - _, ValSpine, >(reach.inner, reach_pact, "Reach", ValBatcher::new); @@ -163,7 +161,6 @@ mod reachability { let combined_arr = arrange_core::<_, _, ValChunker<(Node, (), IterTime, Diff)>, _, - _, ValSpine, >(combined.inner, combined_pact, "Combined", ValBatcher::new); diff --git a/differential-dataflow/examples/columnar_spill.rs b/differential-dataflow/examples/columnar_spill.rs index 712955bb4..f12bff8de 100644 --- a/differential-dataflow/examples/columnar_spill.rs +++ b/differential-dataflow/examples/columnar_spill.rs @@ -240,7 +240,6 @@ fn run_timely_dataflow(times: u64, keys_per_time: u64, workers: usize, sample_se _, _, ValChunker<(u64, u64, u64, i64)>, _, - _, ValSpine, >(stream, Pipeline, "ColumnarSpillArrange", ValBatcher::new); arranged.stream.probe_with(&mut probe); diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs index 62a0cf77c..e4ed0bf39 100644 --- a/differential-dataflow/examples/spines.rs +++ b/differential-dataflow/examples/spines.rs @@ -58,8 +58,8 @@ fn main() { use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatcher, OrdKeySpine}; let (data_input, data) = scope.new_collection::(); let (keys_input, keys) = scope.new_collection::(); - let data = data.map(|k| (k, ())).arrange::<_, _, OrdKeySpine>(OrdKeyBatcher::new); - let keys = keys.map(|k| (k, ())).arrange::<_, _, OrdKeySpine>(OrdKeyBatcher::new); + let data = data.map(|k| (k, ())).arrange::<_, OrdKeySpine>(OrdKeyBatcher::new); + let keys = keys.map(|k| (k, ())).arrange::<_, OrdKeySpine>(OrdKeyBatcher::new); keys.join_core(data, |_k, &(), &()| Option::<()>::None) .probe_with(&mut probe); Workload { data_input, keys_input } @@ -68,8 +68,8 @@ fn main() { use differential_dataflow::trace::implementations::ord_neu::{OrdValBatcher, OrdValSpine}; let (data_input, data) = scope.new_collection::(); let (keys_input, keys) = scope.new_collection::(); - let data = data.map(|x| (x, ())).arrange::<_, _, OrdValSpine>(OrdValBatcher::new); - let keys = keys.map(|x| (x, ())).arrange::<_, _, OrdValSpine>(OrdValBatcher::new); + let data = data.map(|x| (x, ())).arrange::<_, OrdValSpine>(OrdValBatcher::new); + let keys = keys.map(|x| (x, ())).arrange::<_, OrdValSpine>(OrdValBatcher::new); keys.join_core(data, |_k, &(), &()| Option::<()>::None) .probe_with(&mut probe); Workload { data_input, keys_input } @@ -93,8 +93,8 @@ fn main() { type Sp = Spine; type Chu = ContainerChunker>; let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into()); - let data = arrange_core::<_, _, Chu, Ba, _, Sp>(data.inner, exchange(), "DataArrange", Ba::new); - let keys = arrange_core::<_, _, Chu, Ba, _, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new); + let data = arrange_core::<_, _, Chu, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new); + let keys = arrange_core::<_, _, Chu, 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); @@ -119,8 +119,8 @@ fn main() { type Sp = ChunkSpine; type Chu = ContainerChunker>; let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into()); - let data = arrange_core::<_, _, Chu, Ba, _, Sp>(data.inner, exchange(), "DataArrange", Ba::new); - let keys = arrange_core::<_, _, Chu, Ba, _, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new); + let data = arrange_core::<_, _, Chu, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new); + let keys = arrange_core::<_, _, Chu, Ba, Sp>(keys.inner, exchange(), "KeysArrange", Ba::new); keys.join_core(data, |_k, &(), &()| Option::<()>::None) .probe_with(&mut probe); Workload { data_input, keys_input } diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs index 4e2be6e8e..daed2c29a 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -958,21 +958,20 @@ pub mod vec { /// ``` pub fn consolidate(self) -> Self { use crate::trace::implementations::{KeyBatcher, KeySpine}; - self.consolidate_named::<_,_, KeySpine<_,_,_>,_>("Consolidate", KeyBatcher::new, |key,&()| key.clone()) + 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(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba, reify: F) -> Self + pub fn consolidate_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba, reify: F) -> Self where - Ba: crate::trace::Batcher, B> + 'static, - B: Into, + Ba: crate::trace::Batcher, Time = T, Output: Into> + 'static, Tr: crate::trace::Trace+'static, for<'a> BatchCursor: Cursor, F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static, { self.map(|k| (k, ())) - .arrange_named::(name, batcher) + .arrange_named::(name, batcher) .as_collection(reify) } @@ -1036,24 +1035,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(self, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> + pub fn arrange(self, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> where - Ba: crate::trace::Batcher, B> + 'static, - B: Into, + Ba: crate::trace::Batcher, Time = T, Output: Into> + 'static, Tr: crate::trace::Trace + 'static, { - self.arrange_named::("Arrange", batcher) + self.arrange_named::("Arrange", batcher) } /// As [`Collection::arrange`] but with the ability to name the operator. - pub fn arrange_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> + pub fn arrange_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> where - Ba: crate::trace::Batcher, B> + 'static, - B: Into, + Ba: crate::trace::Batcher, Time = T, Output: Into> + 'static, Tr: crate::trace::Trace + '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>, Ba, B, _>(self.inner, exchange, name, batcher) + crate::operators::arrange::arrangement::arrange_core::<_, _, ContainerChunker>, Ba, _>(self.inner, exchange, name, batcher) } } @@ -1072,7 +1069,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>> { - self.arrange_named::<_,_,_>(name, ValBatcher::new) + self.arrange_named::<_,_>(name, ValBatcher::new) } } @@ -1092,7 +1089,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>> { self.map(|k| (k, ())) - .arrange_named::<_,_,_>(name, KeyBatcher::new) + .arrange_named::<_,_>(name, KeyBatcher::new) } } diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index db8fcc4c2..906a81807 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -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, B, Tr>( +pub fn arrange_core<'scope, P, C, Chu, Ba, Tr>( stream: Stream<'scope, Tr::Time, C>, pact: P, name: &str, @@ -313,8 +313,7 @@ where C: Container + Clone + 'static, P: ParallelizationContract, Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static, - Ba: Batcher + 'static, - B: Into, + Ba: Batcher> + 'static, Tr: Trace+'static, { // The `Arrange` operator is tasked with reacting to an advancing input diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index 7720c14e5..c3c1f8236 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -41,11 +41,14 @@ pub struct MergeBatcher { builder: std::marker::PhantomData, } -impl Batcher for MergeBatcher +impl Batcher for MergeBatcher where M: Merger, Bu: Builder, { + type Time = M::Time; + type Output = Bu::Output; + fn insert(&mut self, chunk: &mut M::Chunk) { self.insert_chain(vec![std::mem::take(chunk)]); } diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index 3c114d282..75a82c4a2 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -228,13 +228,18 @@ pub trait Trace : TraceReader { /// A type capable of accepting containers of updates, and carving them out by time as batches. /// /// Updates are accepted as `C0`, the containers that arrive on the dataflow edge, and released as -/// `C1`, whatever the consumer means by a batch. The two need not agree: an implementor staging -/// updates in a form of its own can release that form directly. A consumer whose batch is a -/// sequence of chunks names a sequence as `C1`. +/// `Output`, whatever the implementor means by a batch. The two need not agree: an implementor +/// staging updates in a form of its own can release that form directly, and one whose batch is a +/// sequence of chunks names a sequence as its output. /// /// The implementor determines the meaning of extraction by a frontier; it is not required to be by /// antichain partial order. -pub trait Batcher { +pub trait Batcher { + /// The timestamps by which updates are carved out. + type Time; + /// The batches released by extraction. + type Output; + /// Takes the updates in `container`, leaving it in an undefined state. /// /// The implementor decides whether to claim the container's allocation or to drain it and @@ -249,7 +254,7 @@ pub trait Batcher { /// The reported lower bound should accurately reflect the times of all accepted updates that /// have not been extracted. Over approximation can result in stalling dataflows, and under /// approximation is simply incorrect. - fn extract<'a>(&'a mut self, upper: AntichainRef<'_, T>) -> (Option, AntichainRef<'a, T>); + fn extract<'a>(&'a mut self, upper: AntichainRef<'_, Self::Time>) -> (Option, AntichainRef<'a, Self::Time>); } /// Functionality for building batches from ordered update sequences. diff --git a/differential-dataflow/tests/columnar.rs b/differential-dataflow/tests/columnar.rs index 3428464c3..d8583d3be 100644 --- a/differential-dataflow/tests/columnar.rs +++ b/differential-dataflow/tests/columnar.rs @@ -31,7 +31,6 @@ fn arrange_reaches_one(config: Config, diffs: &'static [i64]) -> bool { _, Chunker, _, - _, Spine, >(stream, pact, "Arrange", Batcher::new) .stream diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index 4aeccf15e..13e89e21f 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -139,7 +139,7 @@ fn proxy_matches_mainline(updates: Vec<((u64, u64), u64, i64)>, window: usize) { worker.dataflow::(|scope| { let coll = updates.clone().to_stream(scope).as_collection(); let hashed = coll.clone().map(|(k, v)| (k.hashed(), (k, v))); - let arr = arrange_core::, ContainerChunker>, _, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); + let arr = arrange_core::, ContainerChunker>, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, window))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) .inspect(move |(d, t, r)| ts.lock().unwrap().push((*d, *t, *r))); @@ -202,7 +202,7 @@ fn reduce_string_values_matches_mainline() { worker.dataflow::(|scope| { let coll = updates.clone().to_stream(scope).as_collection(); let hashed = coll.clone().map(|(k, v): (u64, String)| (k.hashed(), (k, v))); - let arr = arrange_core::, ContainerChunker>, _, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); + let arr = arrange_core::, ContainerChunker>, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduceStr", ProxyReduceTactic::new(VecReduceBackend::with_window( |_k: &u64, input: &[(String, i64)], current: &mut Vec<(String, i64)>, updates: &mut Vec<(String, i64)>| { if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { updates.push((m, 1)); } @@ -236,7 +236,7 @@ fn reduce_inside_iterate() { let input = updates.clone().to_stream(scope).as_collection(); let result = input.iterate(|_scope, inner| { let hashed = inner.map(|(k, v)| (k.hashed(), (k, v))); - let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "IterReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) }); @@ -311,7 +311,7 @@ fn reduce_collision_inside_iterate() { let input = updates.clone().to_stream(scope).as_collection(); let result = input.iterate(|_scope, inner| { let hashed = inner.map(|(k, v)| (k % 2, (k, v))); - let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "CollideReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) }); diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs index 9c21dccfb..ca7f68779 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -58,7 +58,6 @@ macro_rules! harrange { Vec<((u64, (u64, u64)), $t, isize)>, ContainerChunker>, _, - _, VChunkSpine, >(hashed.inner, Pipeline, $name, VChunkBatcher::new) }}; @@ -149,7 +148,6 @@ fn run_wide(mode: Mode) -> f64 { let arr = arrange_core::, ContainerChunker>, _, - _, VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); arr.reduce_core::<_, VChunkBuilder, VChunkSpine, @@ -173,7 +171,6 @@ fn run_wide(mode: Mode) -> f64 { let arr = arrange_core::, ContainerChunker>, _, - _, VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>( arr, "ProxyWide", diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs index 77cd816b4..588a6f571 100644 --- a/dogsdogsdogs/src/operators/half_join.rs +++ b/dogsdogsdogs/src/operators/half_join.rs @@ -77,7 +77,7 @@ where FF: Fn(&Tr::Time, &mut Antichain) + 'static, P: ParallelizationContract, Y: Fn(std::time::Instant, usize) -> bool + 'static, - Bat: Batcher> + 'static, + Bat: Batcher> + 'static, Tac: HalfJoinTactic + 'static, CIn: Container, C: Container + 'static, @@ -484,7 +484,10 @@ pub mod cursors { } } - impl Batcher, Vec>> for BlobList { + impl Batcher> for BlobList { + type Time = T; + type Output = Vec>; + fn insert(&mut self, container: &mut Vec<(D, T, R)>) { self.stage.extend(container.drain(..).map(|(d,t,r)| (t,d,r))); } diff --git a/experiments/src/bin/deals.rs b/experiments/src/bin/deals.rs index 392ecec5d..dfbec4768 100644 --- a/experiments/src/bin/deals.rs +++ b/experiments/src/bin/deals.rs @@ -39,7 +39,7 @@ fn main() { let (input, graph) = scope.new_collection(); // each edge should exist in both directions. - let graph = graph.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let graph = graph.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); match program.as_str() { "tc" => tc(graph.clone()).filter(move |_| inspect).map(|_| ()).consolidate().inspect(|x| println!("tc count: {:?}", x)).probe(), @@ -93,7 +93,7 @@ fn tc<'s, T: timely::progress::Timestamp + Lattice + Default + timely::order::Em let result = inner_collection .map(|(x,y)| (y,x)) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(edges.clone(), |_y,&x,&z| Some((x, z))) .concat(edges.as_collection(|&k,&v| (k,v))) .arrange_by_self() @@ -121,9 +121,9 @@ fn sg<'s, T: timely::progress::Timestamp + Lattice + Default + timely::order::Em let result = inner_collection - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(edges.clone(), |_,&x,&z| Some((x, z))) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(edges, |_,&x,&z| Some((x, z))) .concat(peers) .arrange_by_self() diff --git a/experiments/src/bin/graspan1.rs b/experiments/src/bin/graspan1.rs index 4b443f6a9..443b64e07 100644 --- a/experiments/src/bin/graspan1.rs +++ b/experiments/src/bin/graspan1.rs @@ -29,7 +29,7 @@ fn main() { let (n_handle, nodes) = scope.new_collection(); let (e_handle, edges) = scope.new_collection(); - let edges = edges.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let edges = edges.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); // a N c <- a N b && b E c // N(a,c) <- N(a,b), E(b, c) @@ -46,7 +46,7 @@ fn main() { labels_collection.join_core(edges, |_b, a, c| Some((*c, *a))) .concat(nodes) .map(|k| (k, ())) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) // .distinct_total_core::(); .threshold_semigroup(|_,_,x: Option<&Present>| if x.is_none() { Some(Present) } else { None }); diff --git a/experiments/src/bin/graspan2.rs b/experiments/src/bin/graspan2.rs index 71621b94c..40d20eba1 100644 --- a/experiments/src/bin/graspan2.rs +++ b/experiments/src/bin/graspan2.rs @@ -45,7 +45,7 @@ fn unoptimized() { .flat_map(|(a,b)| vec![a,b]) .concat(dereference.clone().flat_map(|(a,b)| vec![a,b])); - let dereference = dereference.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let dereference = dereference.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); let (value_flow, memory_alias, value_alias) = scope @@ -58,14 +58,14 @@ fn unoptimized() { let (value_flow, value_flow_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); let (memory_alias, memory_alias_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); - let value_flow_arranged = value_flow_collection.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); - let memory_alias_arranged = memory_alias_collection.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let value_flow_arranged = value_flow_collection.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); + let memory_alias_arranged = memory_alias_collection.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); // VA(a,b) <- VF(x,a),VF(x,b) // VA(a,b) <- VF(x,a),MA(x,y),VF(y,b) let value_alias_next = value_flow_arranged.clone().join_core(value_flow_arranged.clone(), |_,&a,&b| Some((a,b))); let value_alias_next = value_flow_arranged.clone().join_core(memory_alias_arranged.clone(), |_,&a,&b| Some((b,a))) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_arranged.clone(), |_,&a,&b| Some((a,b))) .concat(value_alias_next); @@ -75,10 +75,10 @@ fn unoptimized() { let value_flow_next = assignment.clone() .map(|(a,b)| (b,a)) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(memory_alias_arranged, |_,&a,&b| Some((b,a))) .concat(assignment.map(|(a,b)| (b,a))) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_arranged, |_,&a,&b| Some((a,b))) .concat(nodes.map(|n| (n,n))); @@ -93,7 +93,7 @@ fn unoptimized() { let memory_alias_next: VecCollection<_,_,Present> = value_alias_next.clone() .join_core(dereference.clone(), |_x,&y,&a| Some((y,a))) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(dereference, |_y,&a,&b| Some((a,b))); let memory_alias_next: VecCollection<_,_,Present> = @@ -170,7 +170,7 @@ fn optimized() { .flat_map(|(a,b)| vec![a,b]) .concat(dereference.clone().flat_map(|(a,b)| vec![a,b])); - let dereference = dereference.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let dereference = dereference.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); let (value_flow, memory_alias) = scope @@ -183,8 +183,8 @@ fn optimized() { let (value_flow, value_flow_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); let (memory_alias, memory_alias_collection) = Variable::new(inner_scope, Product::new(Default::default(), 1)); - let value_flow_arranged = value_flow_collection.clone().arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); - let memory_alias_arranged = memory_alias_collection.arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + let value_flow_arranged = value_flow_collection.clone().arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); + let memory_alias_arranged = memory_alias_collection.arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); // VF(a,a) <- // VF(a,b) <- A(a,x),VF(x,b) @@ -192,10 +192,10 @@ fn optimized() { let value_flow_next = assignment.clone() .map(|(a,b)| (b,a)) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(memory_alias_arranged.clone(), |_,&a,&b| Some((b,a))) .concat(assignment.map(|(a,b)| (b,a))) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_arranged, |_,&a,&b| Some((a,b))) .concat(nodes.map(|n| (n,n))) .arrange_by_self() @@ -207,9 +207,9 @@ fn optimized() { let value_flow_deref = value_flow_collection .map(|(a,b)| (b,a)) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(dereference, |_x,&a,&b| Some((a,b))) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new); + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new); // MA(a,b) <- VFD(x,a),VFD(y,b) // MA(a,b) <- VFD(x,a),MA(x,y),VFD(y,b) @@ -220,7 +220,7 @@ fn optimized() { let memory_alias_next = memory_alias_arranged .join_core(value_flow_deref.clone(), |_x,&y,&a| Some((y,a))) - .arrange::<_, _, ValSpine<_,_,_,_>>(ValBatcher::new) + .arrange::<_, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(value_flow_deref, |_y,&a,&b| Some((a,b))) .concat(memory_alias_next) .arrange_by_self() diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index 5f06ad5e6..c0c16dd98 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -295,7 +295,7 @@ impl Backend for CorgiBackend { // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher + // builder. No columns→rows→columns round-trip at the arrangement boundary. - arrange_core::<_, CC, CorgiChunker, _, _, CTrace>( + arrange_core::<_, CC, CorgiChunker, _, CTrace>( c.inner, Pipeline, "CorgiArrange", From 373b1176994b9abec62896396026b0776523bd6f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Wed, 26 Aug 2026 11:23:46 -0400 Subject: [PATCH 3/3] The batcher chunks its own input `arrange_core` used to hold a chunker, push each incoming container into it, drain the chunks it emitted into the batcher, and drain it again before sealing so a partial final chunk would not be stranded. That last part needed a comment explaining a seam the operator had no business knowing about. `MergeBatcher` now holds the chunker, so `insert` takes the container from the wire and `extract` flushes before merging. `arrange_core` drops its `Chu` parameter and ingest becomes `batcher.insert(data)`. -pub fn arrange_core<'scope, P, C, Chu, Ba, Tr>( +pub fn arrange_core<'scope, P, C, Ba, Tr>( - Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static, - Ba: Batcher> + 'static, + Ba: Batcher> + 'static, The chunker is not a separate axis from the batcher: it adapts whatever container is on the wire to the merger's preferred representation, so it is part of what selects a batcher rather than a tax on everyone downstream. The columnar batcher pairs with `TrieChunker` and with `ContainerChunker` depending on its input, and those are now two batchers rather than one batcher and two chunkers. Corgi keeps its column-native ingest by naming its chunker where it names its batcher. One consequence: `MergeBatcher` implements `Batcher` for every `C` its chunker accepts, and `extract` does not mention `C`, so a caller holding a concrete batcher must say which implementation it means. Generic callers pin `C` in their bounds and are unaffected; the three sites that needed a turbofish are tests driving a batcher by hand. Co-Authored-By: Claude Opus 5 --- .../examples/columnar/main.rs | 5 +- .../examples/columnar_spill.rs | 3 +- differential-dataflow/examples/spines.rs | 19 ++++---- differential-dataflow/src/collection.rs | 3 +- .../src/columnar/trace/mod.rs | 2 +- .../src/operators/arrange/arrangement.rs | 21 ++------- differential-dataflow/src/trace/chunk/mod.rs | 2 +- differential-dataflow/src/trace/chunk/vec.rs | 2 +- .../trace/implementations/merge_batcher.rs | 47 +++++++++++++------ .../src/trace/implementations/ord_neu.rs | 5 +- differential-dataflow/tests/columnar.rs | 3 +- differential-dataflow/tests/int_proxy.rs | 9 ++-- .../tests/int_proxy_bench.rs | 6 +-- differential-dataflow/tests/trace.rs | 7 +-- interactive/src/backend/corgi.rs | 4 +- 15 files changed, 67 insertions(+), 71 deletions(-) diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs index 2f5fd2f9c..d85d572f5 100644 --- a/differential-dataflow/examples/columnar/main.rs +++ b/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,13 +128,11 @@ mod reachability { let reach_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 }; let edges_arr = arrange_core::<_, _, - ValChunker<(Node, Node, IterTime, Diff)>, _, ValSpine, >(edges_inner.inner, edges_pact, "Edges", ValBatcher::new); let reach_arr = arrange_core::<_, _, - ValChunker<(Node, (), IterTime, Diff)>, _, ValSpine, >(reach.inner, reach_pact, "Reach", ValBatcher::new); @@ -159,7 +157,6 @@ 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)>, _, ValSpine, >(combined.inner, combined_pact, "Combined", ValBatcher::new); diff --git a/differential-dataflow/examples/columnar_spill.rs b/differential-dataflow/examples/columnar_spill.rs index f12bff8de..611b8602a 100644 --- a/differential-dataflow/examples/columnar_spill.rs +++ b/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, 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,7 +238,6 @@ 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)>, _, ValSpine, >(stream, Pipeline, "ColumnarSpillArrange", ValBatcher::new); diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs index e4ed0bf39..f6606afdd 100644 --- a/differential-dataflow/examples/spines.rs +++ b/differential-dataflow/examples/spines.rs @@ -79,7 +79,8 @@ fn main() { // the same generic `Chunk` harness as `vec` via a // `ContainerChunker`. use differential_dataflow::Hashable; - use differential_dataflow::columnar::trace::{Batcher, 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,12 +90,12 @@ fn main() { let data = data.map(|x| (x, ())); let keys = keys.map(|x| (x, ())); - type Ba = Batcher; - type Sp = Spine; type Chu = ContainerChunker>; + type Ba = ChunkBatcher>; + type Sp = Spine; let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into()); - let data = arrange_core::<_, _, Chu, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new); - let keys = arrange_core::<_, _, Chu, Ba, 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); @@ -105,8 +106,7 @@ fn main() { // insert allocates a `String`) but arranged through the `Chunk` // harness via a `ContainerChunker`. use differential_dataflow::Hashable; - use differential_dataflow::trace::chunk::vec::{ChunkBatcher, 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,10 +117,9 @@ fn main() { type Ba = ChunkBatcher; type Sp = ChunkSpine; - type Chu = ContainerChunker>; let exchange = || Exchange::new(|u: &((String, ()), u64, isize)| (u.0).0.hashed().into()); - let data = arrange_core::<_, _, Chu, Ba, Sp>(data.inner, exchange(), "DataArrange", Ba::new); - let keys = arrange_core::<_, _, Chu, Ba, 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 } diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs index daed2c29a..abe41b3fc 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -1021,7 +1021,6 @@ pub mod vec { use crate::trace::implementations::{ValSpine, ValBatcher}; use crate::trace::implementations::{KeySpine, KeyBatcher}; - use crate::trace::implementations::ContainerChunker; impl<'scope, T, K, V, R> Collection<'scope, T, (K, V), R> where T: Timestamp + Lattice, @@ -1050,7 +1049,7 @@ pub mod vec { Tr: crate::trace::Trace + '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>, Ba, _>(self.inner, exchange, name, batcher) + crate::operators::arrange::arrangement::arrange_core::<_, _, Ba, _>(self.inner, exchange, name, batcher) } } diff --git a/differential-dataflow/src/columnar/trace/mod.rs b/differential-dataflow/src/columnar/trace/mod.rs index eb2e35740..5b6a987e3 100644 --- a/differential-dataflow/src/columnar/trace/mod.rs +++ b/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 = crate::trace::chunk::ChunkSpine>; /// The columnar merge batcher (the chunk harness's `MergeBatcher` over `ColChunk`). -pub type Batcher = crate::trace::chunk::ChunkBatcher>; +pub type Batcher = crate::trace::chunk::ChunkBatcher, ColChunk<(K, V, T, R)>>; /// The columnar batch builder. pub type Builder = crate::trace::chunk::ChunkBuilder>; /// The input chunker: melds `RecordedUpdates` streams into `ColChunk` batches. diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index 906a81807..c553aeac5 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/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, Tr>( +pub fn arrange_core<'scope, P, C, Ba, Tr>( stream: Stream<'scope, Tr::Time, C>, pact: P, name: &str, @@ -312,8 +312,7 @@ pub fn arrange_core<'scope, P, C, Chu, Ba, Tr>( where C: Container + Clone + 'static, P: ParallelizationContract, - Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static, - Ba: Batcher> + 'static, + Ba: Batcher> + 'static, Tr: Trace+'static, { // The `Arrange` operator is tasked with reacting to an advancing input @@ -362,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. @@ -374,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 @@ -392,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, diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index f900778ec..ccc126b37 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -277,7 +277,7 @@ where /// using [`Chunk::extract`]. The batcher consolidates equal `(data, time)` updates /// but does *not* advance times — time advancement is advance's job, handled later in /// the trace. Both settle their output, since the batcher's chains want to be graded. -pub type ChunkBatcher = crate::trace::implementations::merge_batcher::MergeBatcher, ChunkBuilder>; +pub type ChunkBatcher = crate::trace::implementations::merge_batcher::MergeBatcher, ChunkBuilder>; /// A spine of `Rc`-shared [`ChunkBatch`]s of type `C`: the trace type for `arrange`. pub type ChunkSpine = crate::trace::implementations::spine_fueled::Spine>>; diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs index 315a96138..df9c6c51e 100644 --- a/differential-dataflow/src/trace/chunk/vec.rs +++ b/differential-dataflow/src/trace/chunk/vec.rs @@ -53,7 +53,7 @@ impl Default for VecChunk { pub type ChunkSpine = super::ChunkSpine>; /// Merge batcher over `VecChunk`s; a `ContainerChunker` at the /// `arrange_core` callsite forms the chunks it merges (via the container traits below). -pub type ChunkBatcher = super::ChunkBatcher>; +pub type ChunkBatcher = super::ChunkBatcher>, VecChunk>; /// Batch builder. pub type ChunkBuilder = super::ChunkBuilder>; diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index c3c1f8236..b668d32d5 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -4,10 +4,11 @@ //! hooks for manipulating sorted "chains" of chunks as needed by the merge batcher: merging //! chunks and also splitting them apart based on time. //! -//! Callers feed already-chunked, sorted-and-consolidated input into the batcher via [`Batcher::insert`]. -//! Forming such chunks from raw data is the responsibility of the caller (typically a chunker -//! living in the surrounding dataflow operator). +//! Raw input containers are fed to the batcher via [`Batcher::insert`], which chunks them with +//! its `Chu` before merging: forming sorted, consolidated chunks is the first stage of the +//! batcher's own work rather than something a caller arranges. +use timely::container::{ContainerBuilder, PushInto}; use timely::progress::frontier::AntichainRef; use timely::progress::{frontier::Antichain, Timestamp}; @@ -16,9 +17,12 @@ use crate::trace::{Batcher, Builder}; /// Creates batches from chunks of sorted, consolidated tuples. /// -/// Merging is `M`'s business and building 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 { +/// 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 +/// carve-by-frontier. +pub struct MergeBatcher { + /// Melds input containers into sorted, consolidated chunks. + chunker: Chu, /// Sorted, consolidated chains, each paired with its cached summed update count. /// /// The cached count is the chain's *merge weight*: the geometric ladder weighs @@ -41,16 +45,20 @@ pub struct MergeBatcher { builder: std::marker::PhantomData, } -impl Batcher for MergeBatcher +impl Batcher for MergeBatcher where M: Merger, + Chu: ContainerBuilder + for<'a> PushInto<&'a mut C>, Bu: Builder, { type Time = M::Time; type Output = Bu::Output; - fn insert(&mut self, chunk: &mut M::Chunk) { - self.insert_chain(vec![std::mem::take(chunk)]); + fn insert(&mut self, container: &mut C) { + self.chunker.push_into(container); + while let Some(chunk) = self.chunker.extract().map(std::mem::take) { + self.insert_chain(vec![chunk]); + } } // Extraction means finding those updates with times not greater or equal to any time in @@ -58,6 +66,12 @@ where // 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>) { + // 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) { + self.insert_chain(vec![chunk]); + } + // Merge all remaining chains into a single chain. while self.chains.len() > 1 { let list1 = self.chain_pop().unwrap(); @@ -84,7 +98,7 @@ where } } -impl MergeBatcher { +impl MergeBatcher { /// Allocates a new empty batcher. /// /// The logger and operator identifier are used to report the batcher's memory footprint, @@ -94,13 +108,16 @@ impl MergeBatcher { logger, operator_id, merger: M::default(), + chunker: Chu::default(), chains: Vec::new(), stash: Vec::new(), frontier: Antichain::new(), builder: std::marker::PhantomData, } } +} +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) { @@ -175,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() {} @@ -385,8 +402,10 @@ mod test { use super::MergeBatcher; use super::vec::VecMerger; use crate::trace::implementations::ord_neu::VecOrdKeyBuilder; + use crate::trace::implementations::chunker::ContainerChunker; - type Bt = MergeBatcher, VecOrdKeyBuilder>; + type In = Vec<((u64, ()), u64, i64)>; + type Bt = MergeBatcher, VecMerger<(u64, ()), u64, i64>, VecOrdKeyBuilder>; /// The sealed frontier must reflect the POST-CONSOLIDATION set of distinct kept times: /// two chains carry cancelling updates at a kept time (`t=5`), plus a survivor at a later @@ -398,7 +417,7 @@ mod test { let mut b = Bt::new(None, 0); b.chain_push(vec![vec![((100u64, ()), 5u64, 1i64), ((200u64, ()), 7u64, 1i64)]]); b.chain_push(vec![vec![((100u64, ()), 5u64, -1i64)]]); - let (_, retained) = b.extract(Antichain::from_elem(3).borrow()); + let (_, retained) = Batcher::::extract(&mut b, Antichain::from_elem(3).borrow()); let got: Vec = retained.iter().cloned().collect(); assert_eq!(got, vec![7u64], "frontier held a capability at t=5, which consolidates to zero (got {got:?})"); @@ -410,7 +429,7 @@ mod test { let mut b = Bt::new(None, 0); b.chain_push(vec![vec![((100u64, ()), 5u64, 1i64)]]); b.chain_push(vec![vec![((200u64, ()), 7u64, 1i64)]]); - let (_, retained) = b.extract(Antichain::from_elem(3).borrow()); + let (_, retained) = Batcher::::extract(&mut b, Antichain::from_elem(3).borrow()); let got: Vec = retained.iter().cloned().collect(); assert_eq!(got, vec![5u64]); } diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index 5ab540db0..89ed2d15f 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -11,6 +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::MergeBatcher; use crate::trace::implementations::merge_batcher::vec::VecMerger; @@ -22,14 +23,14 @@ pub use self::key_batch::{OrdKeyBatch, OrdKeyBuilder}; /// A trace implementation using a spine of ordered lists. pub type OrdValSpine = Spine>>>; /// A batcher using ordered lists. -pub type OrdValBatcher = MergeBatcher, VecOrdValBuilder>; +pub type OrdValBatcher = MergeBatcher>, VecMerger<(K, V), T, R>, VecOrdValBuilder>; /// A builder using ordered lists. pub type VecOrdValBuilder = OrdValBuilder, Vec<((K,V),T,R)>>; /// A trace implementation using a spine of ordered lists. pub type OrdKeySpine = Spine>>>; /// A batcher for ordered lists. -pub type OrdKeyBatcher = MergeBatcher, VecOrdKeyBuilder>; +pub type OrdKeyBatcher = MergeBatcher>, VecMerger<(K, ()), T, R>, VecOrdKeyBuilder>; /// A builder for ordered lists. pub type VecOrdKeyBuilder = OrdKeyBuilder, Vec<((K,()),T,R)>>; diff --git a/differential-dataflow/tests/columnar.rs b/differential-dataflow/tests/columnar.rs index d8583d3be..5f7fa642a 100644 --- a/differential-dataflow/tests/columnar.rs +++ b/differential-dataflow/tests/columnar.rs @@ -5,7 +5,7 @@ use timely::dataflow::{InputHandle, ProbeHandle}; use timely::Config; use differential_dataflow::columnar::collection; -use differential_dataflow::columnar::trace::{Batcher, Chunker, Spine}; +use differential_dataflow::columnar::trace::{Batcher, Spine}; use differential_dataflow::operators::arrange::arrangement::arrange_core; type Upd = (u64, (), u64, i64); @@ -29,7 +29,6 @@ fn arrange_reaches_one(config: Config, diffs: &'static [i64]) -> bool { arrange_core::< _, _, - Chunker, _, Spine, >(stream, pact, "Arrange", Batcher::new) diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index 13e89e21f..8618bf1e5 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -28,7 +28,6 @@ use differential_dataflow::trace::chunk::vec::{ }; use differential_dataflow::trace::chunk::ChunkBatch; use differential_dataflow::trace::cursor::Cursor; -use differential_dataflow::trace::implementations::ContainerChunker; use differential_dataflow::trace::{Navigable}; use differential_dataflow::AsCollection; @@ -139,7 +138,7 @@ fn proxy_matches_mainline(updates: Vec<((u64, u64), u64, i64)>, window: usize) { worker.dataflow::(|scope| { let coll = updates.clone().to_stream(scope).as_collection(); let hashed = coll.clone().map(|(k, v)| (k.hashed(), (k, v))); - let arr = arrange_core::, ContainerChunker>, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); + let arr = arrange_core::, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, window))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) .inspect(move |(d, t, r)| ts.lock().unwrap().push((*d, *t, *r))); @@ -202,7 +201,7 @@ fn reduce_string_values_matches_mainline() { worker.dataflow::(|scope| { let coll = updates.clone().to_stream(scope).as_collection(); let hashed = coll.clone().map(|(k, v): (u64, String)| (k.hashed(), (k, v))); - let arr = arrange_core::, ContainerChunker>, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); + let arr = arrange_core::, _, VChunkSpine>(hashed.inner, Pipeline, "Arrange", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>(arr, "VecReduceStr", ProxyReduceTactic::new(VecReduceBackend::with_window( |_k: &u64, input: &[(String, i64)], current: &mut Vec<(String, i64)>, updates: &mut Vec<(String, i64)>| { if let Some(m) = input.iter().filter(|(_, d)| *d > 0).map(|(v, _)| v.clone()).max() { updates.push((m, 1)); } @@ -236,7 +235,7 @@ fn reduce_inside_iterate() { let input = updates.clone().to_stream(scope).as_collection(); let result = input.iterate(|_scope, inner| { let hashed = inner.map(|(k, v)| (k.hashed(), (k, v))); - let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new); + let arr = arrange_core::, i64)>, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "IterReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) }); @@ -311,7 +310,7 @@ fn reduce_collision_inside_iterate() { let input = updates.clone().to_stream(scope).as_collection(); let result = input.iterate(|_scope, inner| { let hashed = inner.map(|(k, v)| (k % 2, (k, v))); - let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new); + let arr = arrange_core::, i64)>, _, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, i64>, _>(arr, "CollideReduce", ProxyReduceTactic::new(VecReduceBackend::with_window(max_logic, 1))) .as_collection(|_h, kw: &(u64, u64)| (kw.0, kw.1)) }); diff --git a/differential-dataflow/tests/int_proxy_bench.rs b/differential-dataflow/tests/int_proxy_bench.rs index ca7f68779..094d5361e 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -34,10 +34,9 @@ use differential_dataflow::operators::iterate::Iterate; use differential_dataflow::operators::reduce::reduce_with_tactic; use differential_dataflow::trace::chunk::vec::{ ChunkBatcher as VChunkBatcher, ChunkBuilder as VChunkBuilder, ChunkSpine as VChunkSpine, - VecChunk, VecChunkCursor, + VecChunkCursor, }; use differential_dataflow::trace::cursor::Cursor; -use differential_dataflow::trace::implementations::ContainerChunker; #[derive(Clone, Copy, PartialEq, Eq, Debug)] enum Mode { @@ -56,7 +55,6 @@ macro_rules! harrange { arrange_core::< Pipeline, Vec<((u64, (u64, u64)), $t, isize)>, - ContainerChunker>, _, VChunkSpine, >(hashed.inner, Pipeline, $name, VChunkBatcher::new) @@ -146,7 +144,6 @@ fn run_wide(mode: Mode) -> f64 { Mode::CursorSame => { let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); let arr = arrange_core::, - ContainerChunker>, _, VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); arr.reduce_core::<_, VChunkBuilder, @@ -169,7 +166,6 @@ fn run_wide(mode: Mode) -> f64 { Mode::Proxy => { let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); let arr = arrange_core::, - ContainerChunker>, _, VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>( diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs index 98abf69ee..1fd9129f5 100644 --- a/differential-dataflow/tests/trace.rs +++ b/differential-dataflow/tests/trace.rs @@ -13,17 +13,18 @@ fn get_trace() -> ValSpine { { let mut batcher = ValBatcher::::new(None, 0); - batcher.insert(&mut vec![ + let mut input: Vec<((u64, u64), usize, i64)> = vec![ ((1, 2), 0, 1), ((2, 3), 1, 1), ((2, 3), 2, -1), - ]); + ]; + batcher.insert(&mut input); let batch_ts = &[1, 2, 3]; let mut lower = Antichain::from_elem(0); for i in batch_ts { let upper = Antichain::from_elem(*i); - let (batch, _retained) = batcher.extract(upper.borrow()); + let (batch, _retained) = Batcher::>::extract(&mut batcher, upper.borrow()); let description = Description::new(lower, upper.clone(), Antichain::from_elem(0)); trace.insert(Span::new(description, batch.map(Into::into))); lower = upper; diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index c0c16dd98..81cdaeff8 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -295,11 +295,11 @@ impl Backend for CorgiBackend { // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher + // builder. No columns→rows→columns round-trip at the arrangement boundary. - arrange_core::<_, CC, CorgiChunker, _, CTrace>( + arrange_core::<_, CC, _, CTrace>( c.inner, Pipeline, "CorgiArrange", - ChunkBatcher::new, + ChunkBatcher::, _>::new, ) }