From 2b40d56a3fc254ab9b02e1220b0663f73c7a6d5e Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 25 Aug 2026 19:28:10 -0400 Subject: [PATCH] One `Batcher` trait, shaped like the one half_join had to write `dogsdogsdogs::half_join` declared its own `Batcher` trait, because `trace::Batcher` did not fit a consumer that stages updates in a form of its own and releases them by a total order rather than an antichain. Two traits with the same name and the same job, one of them half the size. This adopts the smaller one, and deletes the duplicate. pub trait Batcher { fn insert(&mut self, container: &mut C0); fn extract<'a>(&'a mut self, upper: AntichainRef<'_, T>) -> (Option, AntichainRef<'a, T>); } Four consequences, in rough order of how much they matter. Input and output types are now distinct. `trace::Batcher` pinned them to one associated type, which is why `arrange_core` carried a chunker whose container had to equal `Ba::Output`. A batcher may now release whatever its consumer means by a batch: half_join's is a sequence of chunks, and arrange's is a chain it still hands to a `Builder`, which a following commit narrows to the batch itself. Construction leaves the trait. It has to: a trait meant to cover both a merge batcher and, later, a builder-shaped one cannot mandate a `new(logger, operator_id)` constructor, since only the arrange path has a differential logger or an operator id to give. `arrange_core` takes the constructor as an argument and `MergeBatcher::new` is inherent. Call sites move the batcher from turbofish to argument position and are otherwise unchanged. The `Description` leaves too. `MergeBatcher::lower` existed only to manufacture one, duplicating `arrange_core`'s `prev_frontier`; the operator now builds it from the frontiers it already tracks. `seal` and `frontier` collapse into one call, so the ordering invariant between them ("the frontier after the most recent seal") is gone. The retained frontier is reported by the extraction that determines it. `insert` takes its container by reference, as `Builder::push` does, and the implementor decides whether to claim the allocation or drain it. The recycling decision belongs there rather than at the call site, which no longer writes `std::mem::take`; `arrange_core` drops a `Default` bound as a result. This also keeps the door open for a builder-shaped batcher: `reduce` allocates one buffer outside its per-key loop and hands it over by reference every key, which a by-value `insert` would cost it. One behaviour change worth naming: the empty-batch branch used to call `seal` and discard the result, to advance the lower bound that no longer exists. Dropping that call also drops the full chain merge it performed on every empty frontier advance. No held capability precedes the input frontier there, so no update does either, and nothing was extractable. Co-Authored-By: Claude Opus 5 --- .../examples/columnar/main.rs | 12 +-- .../examples/columnar_spill.rs | 4 +- differential-dataflow/examples/spines.rs | 16 ++-- differential-dataflow/src/collection.rs | 36 ++++----- .../src/operators/arrange/arrangement.rs | 45 +++++++---- .../trace/implementations/merge_batcher.rs | 78 ++++++++----------- differential-dataflow/src/trace/mod.rs | 42 +++++----- differential-dataflow/tests/columnar.rs | 4 +- differential-dataflow/tests/int_proxy.rs | 8 +- .../tests/int_proxy_bench.rs | 12 +-- differential-dataflow/tests/trace.rs | 14 ++-- dogsdogsdogs/src/operators/half_join.rs | 43 +++------- experiments/src/bin/deals.rs | 8 +- experiments/src/bin/graspan1.rs | 4 +- experiments/src/bin/graspan2.rs | 30 +++---- interactive/src/backend/corgi.rs | 3 +- 16 files changed, 174 insertions(+), 185 deletions(-) diff --git a/differential-dataflow/examples/columnar/main.rs b/differential-dataflow/examples/columnar/main.rs index dad44cf86..d566c2424 100644 --- a/differential-dataflow/examples/columnar/main.rs +++ b/differential-dataflow/examples/columnar/main.rs @@ -129,17 +129,17 @@ mod reachability { let edges_arr = arrange_core::<_, _, ValChunker<(Node, Node, IterTime, Diff)>, - ValBatcher, + _, ValBuilder, ValSpine, - >(edges_inner.inner, edges_pact, "Edges"); + >(edges_inner.inner, edges_pact, "Edges", ValBatcher::new); let reach_arr = arrange_core::<_, _, ValChunker<(Node, (), IterTime, Diff)>, - ValBatcher, + _, ValBuilder, ValSpine, - >(reach.inner, reach_pact, "Reach"); + >(reach.inner, reach_pact, "Reach", ValBatcher::new); // join_traces with ValColBuilder: produces Stream<_, RecordedUpdates<...>>. let proposed = @@ -162,10 +162,10 @@ mod reachability { let combined_pact = ValPact { hashfunc: |k: columnar::Ref<'_, Node>| *k as u64 }; let combined_arr = arrange_core::<_, _, ValChunker<(Node, (), IterTime, Diff)>, - ValBatcher, + _, ValBuilder, ValSpine, - >(combined.inner, combined_pact, "Combined"); + >(combined.inner, combined_pact, "Combined", ValBatcher::new); // reduce_abelian on the columnar arrangement. let result = combined_arr.reduce_abelian::<_, diff --git a/differential-dataflow/examples/columnar_spill.rs b/differential-dataflow/examples/columnar_spill.rs index a8ad36719..c5619a08b 100644 --- a/differential-dataflow/examples/columnar_spill.rs +++ b/differential-dataflow/examples/columnar_spill.rs @@ -239,10 +239,10 @@ fn run_timely_dataflow(times: u64, keys_per_time: u64, workers: usize, sample_se let arranged = arrange_core::< _, _, ValChunker<(u64, u64, u64, i64)>, - ValBatcher, + _, ValBuilder, ValSpine, - >(stream, Pipeline, "ColumnarSpillArrange"); + >(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 ff17a281a..e7bdb630f 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, VecOrdKeyBuilder, OrdKeySpine}; let (data_input, data) = scope.new_collection::(); let (keys_input, keys) = scope.new_collection::(); - let data = data.map(|k| (k, ())).arrange::, VecOrdKeyBuilder, OrdKeySpine>(); - let keys = keys.map(|k| (k, ())).arrange::, VecOrdKeyBuilder, OrdKeySpine>(); + let data = data.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder, OrdKeySpine>(OrdKeyBatcher::new); + let keys = keys.map(|k| (k, ())).arrange::<_, VecOrdKeyBuilder, 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, VecOrdValBuilder, OrdValSpine}; let (data_input, data) = scope.new_collection::(); let (keys_input, keys) = scope.new_collection::(); - let data = data.map(|x| (x, ())).arrange::, VecOrdValBuilder, OrdValSpine>(); - let keys = keys.map(|x| (x, ())).arrange::, VecOrdValBuilder, OrdValSpine>(); + let data = data.map(|x| (x, ())).arrange::<_, VecOrdValBuilder, OrdValSpine>(OrdValBatcher::new); + let keys = keys.map(|x| (x, ())).arrange::<_, VecOrdValBuilder, OrdValSpine>(OrdValBatcher::new); keys.join_core(data, |_k, &(), &()| Option::<()>::None) .probe_with(&mut probe); Workload { data_input, keys_input } @@ -94,8 +94,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, Bu, Sp>(data.inner, exchange(), "DataArrange"); - let keys = arrange_core::<_, _, Chu, Ba, Bu, Sp>(keys.inner, exchange(), "KeysArrange"); + 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); // `ColChunk`'s cursor yields `Val = columnar::Ref<()> = ()`, not `&()`. keys.join_core(data, |_k, _, _| Option::<()>::None) .probe_with(&mut probe); @@ -121,8 +121,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, Bu, Sp>(data.inner, exchange(), "DataArrange"); - let keys = arrange_core::<_, _, Chu, Ba, Bu, Sp>(keys.inner, exchange(), "KeysArrange"); + 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); 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 bab5ae4a6..a77b05267 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -958,21 +958,21 @@ pub mod vec { /// ``` pub fn consolidate(self) -> Self { use crate::trace::implementations::{KeyBatcher, KeyBuilder, KeySpine}; - self.consolidate_named::,KeyBuilder<_,_,_>, KeySpine<_,_,_>,_>("Consolidate", |key,&()| key.clone()) + self.consolidate_named::<_,KeyBuilder<_,_,_>, 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, reify: F) -> Self + pub fn consolidate_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba, reify: F) -> Self where - Ba: crate::trace::Batcher, Time=T> + 'static, + Ba: crate::trace::Batcher, Vec>> + 'static, 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) + .arrange_named::(name, batcher) .as_collection(reify) } @@ -1032,28 +1032,28 @@ pub mod vec { { /// Arranges updates into a shared trace, exchanged by a hash of the key. /// - /// The batcher's output container must equal the stream container; the default chunker - /// only consolidates 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) -> Arranged<'scope, TraceAgent> + /// The batcher must accept the stream container; the default chunker only consolidates + /// 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> where - Ba: crate::trace::Batcher, Time=T> + 'static, - Bu: crate::trace::Builder, Output: Into>, + Ba: crate::trace::Batcher, Vec> + 'static, + Bu: crate::trace::Builder>, Tr: crate::trace::Trace + 'static, { - self.arrange_named::("Arrange") + self.arrange_named::("Arrange", batcher) } /// As [`Collection::arrange`] but with the ability to name the operator. - pub fn arrange_named(self, name: &str) -> Arranged<'scope, TraceAgent> + pub fn arrange_named(self, name: &str, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> where - Ba: crate::trace::Batcher, Time=T> + 'static, - Bu: crate::trace::Builder, Output: Into>, + Ba: crate::trace::Batcher, Vec> + 'static, + Bu: crate::trace::Builder>, 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) + crate::operators::arrange::arrangement::arrange_core::<_, _, ContainerChunker>, Ba, Bu, _>(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) + self.arrange_named::<_,ValBuilder<_,_,_,_>,_>(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) + .arrange_named::<_,KeyBuilder<_,_,_>,_>(name, KeyBatcher::new) } } diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index 795411c11..3d1d9264e 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -31,7 +31,8 @@ use timely::progress::Stamp; use crate::{Data, VecCollection, AsCollection}; use crate::difference::Semigroup; use crate::lattice::Lattice; -use crate::trace::{self, SpanOf, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; +use crate::logging::Logger; +use crate::trace::{self, Description, SpanOf, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; use trace::wrappers::enter::{TraceEnter, enter_span}; @@ -302,13 +303,18 @@ 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>(stream: Stream<'scope, Tr::Time, C>, pact: P, name: &str) -> Arranged<'scope, TraceAgent> +pub fn arrange_core<'scope, P, C, Chu, Ba, Bu, Tr>( + stream: Stream<'scope, Tr::Time, C>, + pact: P, + name: &str, + batcher: impl FnOnce(Option, usize) -> Ba, +) -> Arranged<'scope, TraceAgent> where C: Container + Clone + 'static, P: ParallelizationContract, - Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static, - Ba: Batcher + 'static, - Bu: Builder>, + Chu: ContainerBuilder + for<'a> PushInto<&'a mut C> + 'static, + Ba: Batcher> + 'static, + Bu: Builder>, Tr: Trace+'static, { // The `Arrange` operator is tasked with reacting to an advancing input @@ -338,7 +344,7 @@ where let logger = scope.worker().logger_for::("differential/arrange").map(Into::into); // Where we will deposit received updates, and from which we extract batches. - let mut batcher = Ba::new(logger.clone(), info.global_id); + let mut batcher = batcher(logger.clone(), info.global_id); // Capabilities for the lower envelope of updates in `batcher`. let mut capabilities = Antichain::>::new(); @@ -371,7 +377,7 @@ where } chunker.push_into(data); while let Some(chunk) = chunker.extract() { - batcher.push_into(std::mem::take(chunk)); + batcher.insert(chunk); } }); @@ -391,7 +397,7 @@ where // 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.push_into(std::mem::take(chunk)); + batcher.insert(chunk); } // There are two cases to handle with some care: @@ -409,7 +415,7 @@ where if capabilities.elements().iter().any(|c| !frontier.less_equal(c.time())) { // The capabilities to retire: those not in advance of the input frontier. - // Each update sealed below is greater or equal to one of them, as updates + // Each update extracted below is greater or equal to one of them, as updates // supported only by the remaining capabilities are in advance of the input // frontier and remain in the batcher. let retired = capabilities @@ -420,8 +426,16 @@ where .collect::>(); // Extract all updates not in advance of the input frontier, as one batch. - let (mut chain, description) = batcher.seal(frontier.frontier().to_owned()); - let batch = trace::Span::new(description, Bu::seal(&mut chain).map(Into::into)); + // The batch spans the interval from the previously reported frontier to the + // current one, which is exactly the interval the batcher carves out. + let description = Description::new( + prev_frontier.clone(), + frontier.frontier().to_owned(), + Antichain::from_elem(Tr::Time::minimum()), + ); + let (chain, retained) = batcher.extract(frontier.frontier()); + + let batch = trace::Span::new(description, chain.and_then(|mut chain| Bu::seal(&mut chain)).map(Into::into)); let stamp = retired.iter().map(|c| c.time().clone()).collect::>(); writer.insert(batch.clone(), stamp); @@ -436,7 +450,7 @@ where // in messages with new capabilities. let mut new_capabilities = Antichain::new(); - for time in batcher.frontier().iter() { + for time in retained.iter() { if let Some(capability) = capabilities.elements().iter().find(|c| c.time().less_equal(time)) { new_capabilities.insert(capability.delayed(time)); } @@ -448,10 +462,9 @@ where capabilities = new_capabilities; } else { - // Announce progress updates, even without data. We seal the batcher to - // advance its lower bound and frontier, but discard the readied updates - // rather than building a batch we would immediately drop. - let _ = batcher.seal(frontier.frontier().to_owned()); + // Announce progress updates, even without data. No held capability precedes + // the input frontier, so no update does either, and the batcher has nothing + // to extract. writer.seal(frontier.frontier().to_owned()); } diff --git a/differential-dataflow/src/trace/implementations/merge_batcher.rs b/differential-dataflow/src/trace/implementations/merge_batcher.rs index 83947eee1..1606a15cc 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher.rs @@ -4,16 +4,15 @@ //! 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 [`PushInto`]. +//! 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). use timely::progress::frontier::AntichainRef; use timely::progress::{frontier::Antichain, Timestamp}; -use timely::container::PushInto; use crate::logging::{BatcherEvent, Logger}; -use crate::trace::{Batcher, Description}; +use crate::trace::Batcher; /// Creates batches from chunks of sorted, consolidated tuples. pub struct MergeBatcher { @@ -29,9 +28,7 @@ pub struct MergeBatcher { stash: Vec, /// Merges consolidated chunks, and extracts the subset of an update chain that lies in an interval of time. merger: M, - /// Current lower frontier, we sealed up to here. - lower: Antichain, - /// The lower-bound frontier of the data, after the last call to seal. + /// The lower-bound frontier of the data, after the last call to extract. frontier: Antichain, /// Logger for size accounting. logger: Option, @@ -39,30 +36,19 @@ pub struct MergeBatcher { operator_id: usize, } -impl Batcher for MergeBatcher +impl Batcher> for MergeBatcher where M: Merger, { - type Time = M::Time; - type Output = M::Chunk; - - fn new(logger: Option, operator_id: usize) -> Self { - Self { - logger, - operator_id, - merger: M::default(), - chains: Vec::new(), - stash: Vec::new(), - frontier: Antichain::new(), - lower: Antichain::from_elem(M::Time::minimum()), - } + fn insert(&mut self, chunk: &mut M::Chunk) { + self.insert_chain(vec![std::mem::take(chunk)]); } - // Sealing a batch means finding those updates with times not greater or equal to any time - // in `upper`. All updates must have time greater or equal to the previously used `upper`, - // which we call `lower`, by assumption that after sealing a batcher we receive no more - // updates with times not greater or equal to `upper`. - fn seal(&mut self, upper: Antichain) -> (Vec, Description) { + // Extraction means finding those updates with times not greater or equal to any time in + // `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>) { // Merge all remaining chains into a single chain. while self.chains.len() > 1 { let list1 = self.chain_pop().unwrap(); @@ -77,7 +63,7 @@ where let mut readied = Vec::new(); self.frontier.clear(); - self.merger.extract(merged, upper.borrow(), &mut self.frontier, &mut readied, &mut kept, &mut self.stash); + self.merger.extract(merged, upper, &mut self.frontier, &mut readied, &mut kept, &mut self.stash); if !kept.is_empty() { self.chain_push(kept); @@ -85,25 +71,27 @@ where self.stash.clear(); - let description = Description::new(self.lower.clone(), upper.clone(), Antichain::from_elem(M::Time::minimum())); - self.lower = upper; - (readied, description) - } - - /// The frontier of elements remaining after the most recent call to `self.seal`. - #[inline] - fn frontier(&mut self) -> AntichainRef<'_, M::Time> { - self.frontier.borrow() + let readied = if readied.is_empty() { None } else { Some(readied) }; + (readied, self.frontier.borrow()) } } -impl PushInto for MergeBatcher { - fn push_into(&mut self, chunk: M::Chunk) { - self.insert_chain(vec![chunk]); +impl MergeBatcher { + /// Allocates a new empty batcher. + /// + /// The logger and operator identifier are used to report the batcher's memory footprint, + /// attributed to the operator that owns it. + pub fn new(logger: Option, operator_id: usize) -> Self { + Self { + logger, + operator_id, + merger: M::default(), + chains: Vec::new(), + stash: Vec::new(), + frontier: Antichain::new(), + } } -} -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) { @@ -392,7 +380,7 @@ mod test { /// 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 - /// kept time (`t=7`). After `seal(upper=[3])` the frontier must be `{7}` — `(100, 5)` nets + /// kept time (`t=7`). After `extract(upper=[3])` the frontier must be `{7}` — `(100, 5)` nets /// to zero and needs no capability. (A per-chain extract that folds the frontier before /// consolidating would wrongly report `{5}`.) #[test] @@ -400,8 +388,8 @@ 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 _ = b.seal(Antichain::from_elem(3)); - let got: Vec = b.frontier().iter().cloned().collect(); + let (_, retained) = b.extract(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:?})"); } @@ -412,8 +400,8 @@ 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 _ = b.seal(Antichain::from_elem(3)); - let got: Vec = b.frontier().iter().cloned().collect(); + 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/mod.rs b/differential-dataflow/src/trace/mod.rs index 9b57c229c..3c114d282 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -13,12 +13,10 @@ pub mod description; pub mod implementations; pub mod wrappers; -use timely::container::PushInto; use timely::progress::{Antichain, frontier::AntichainRef}; use timely::progress::Timestamp; use crate::lattice::Lattice; -use crate::logging::Logger; pub use self::cursor::Cursor; pub use self::cursor::Navigable; pub use self::cursor::{BatchCursor, BatchKey, BatchVal, BatchValOwn, BatchDiff, BatchDiffGat, BatchTimeGat}; @@ -227,25 +225,31 @@ pub trait Trace : TraceReader { fn close(&mut self); } -/// Functionality for collecting and batching updates. +/// A type capable of accepting containers of updates, and carving them out by time as batches. /// -/// Accepts containers of type `Output` via [`PushInto`] and produces output batches of the same -/// type. Callers are responsible for converting raw input data into `Output` containers (e.g. -/// using a chunker) before pushing into the batcher. -pub trait Batcher: PushInto { - /// Type produced by the batcher, and also the type it consumes. - type Output: Default; - /// Times at which batches are formed. - type Time: Timestamp; - /// Allocates a new empty batcher. - fn new(logger: Option, operator_id: usize) -> Self; - /// Returns all updates not greater or equal to an element of `upper`, as a sorted and - /// consolidated chain together with the description that bounds them. +/// 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`. +/// +/// The implementor determines the meaning of extraction by a frontier; it is not required to be by +/// antichain partial order. +pub trait Batcher { + /// 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 + /// leave the allocation with the caller, who is free to reuse the container either way. + fn insert(&mut self, container: &mut C0); + /// Extracts the updates `upper` unblocks as a batch, and lower bounds the times of those retained. + /// + /// What `upper` unblocks is the implementor's to decide. It can be based on the antichain up + /// set, or it can be based on the total order of times (as used in delta join constructions). + /// Absent a batch, `upper` unblocked no updates. /// - /// The returned chain is suitable to hand directly to [`Builder::seal`]. - fn seal(&mut self, upper: Antichain) -> (Vec, Description); - /// Returns the lower envelope of contained update times. - fn frontier(&mut self) -> AntichainRef<'_, Self::Time>; + /// 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>); } /// Functionality for building batches from ordered update sequences. diff --git a/differential-dataflow/tests/columnar.rs b/differential-dataflow/tests/columnar.rs index 84288ad56..5001d01be 100644 --- a/differential-dataflow/tests/columnar.rs +++ b/differential-dataflow/tests/columnar.rs @@ -30,10 +30,10 @@ fn arrange_reaches_one(config: Config, diffs: &'static [i64]) -> bool { _, _, Chunker, - Batcher, + _, Builder, Spine, - >(stream, pact, "Arrange") + >(stream, pact, "Arrange", Batcher::new) .stream .probe_with(&mut probe); }); diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index f55ad2d4e..a71fca87d 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>, VChunkBatcher, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange"); + let arr = arrange_core::, ContainerChunker>, _, VChunkBuilder, 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>, VChunkBatcher, VChunkBuilder, VChunkSpine>(hashed.inner, Pipeline, "Arrange"); + let arr = arrange_core::, ContainerChunker>, _, VChunkBuilder, 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>>, VChunkBatcher, i64>, VChunkBuilder, i64>, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrIter"); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkBuilder, 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>>, VChunkBatcher, i64>, VChunkBuilder, i64>, VChunkSpine, i64>>(hashed.inner, Pipeline, "ArrCollide"); + let arr = arrange_core::, i64)>, ContainerChunker, i64>>, _, VChunkBuilder, 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 ff95586a1..c3942300b 100644 --- a/differential-dataflow/tests/int_proxy_bench.rs +++ b/differential-dataflow/tests/int_proxy_bench.rs @@ -57,10 +57,10 @@ macro_rules! harrange { Pipeline, Vec<((u64, (u64, u64)), $t, isize)>, ContainerChunker>, - VChunkBatcher, + _, VChunkBuilder, VChunkSpine, - >(hashed.inner, Pipeline, $name) + >(hashed.inner, Pipeline, $name, VChunkBatcher::new) }}; } @@ -148,9 +148,9 @@ fn run_wide(mode: Mode) -> f64 { let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); let arr = arrange_core::, ContainerChunker>, - VChunkBatcher, + _, VChunkBuilder, - VChunkSpine>(hashed.inner, Pipeline, "ArrW"); + VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); arr.reduce_core::<_, VChunkBuilder, VChunkSpine, as Cursor>::KeyContainer, _>( @@ -172,9 +172,9 @@ fn run_wide(mode: Mode) -> f64 { let hashed = coll.map(|(k, v)| (k.hashed(), (k, v))); let arr = arrange_core::, ContainerChunker>, - VChunkBatcher, + _, VChunkBuilder, - VChunkSpine>(hashed.inner, Pipeline, "ArrW"); + VChunkSpine>(hashed.inner, Pipeline, "ArrW", VChunkBatcher::new); reduce_with_tactic::<_, VChunkSpine, _>( arr, "ProxyWide", ProxyReduceTactic::new(VecReduceBackend::new( diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs index 49d9a5ca0..7a9aea117 100644 --- a/differential-dataflow/tests/trace.rs +++ b/differential-dataflow/tests/trace.rs @@ -1,9 +1,8 @@ -use timely::container::PushInto; use timely::dataflow::operators::generic::OperatorInfo; use timely::progress::{Antichain, frontier::AntichainRef}; use differential_dataflow::trace::implementations::{ValBatcher, ValBuilder, ValSpine}; -use differential_dataflow::trace::{Span, Trace, TraceReader, Batcher, Builder}; +use differential_dataflow::trace::{Description, Span, Trace, TraceReader, Batcher, Builder}; use differential_dataflow::trace::cursor::{Cursor, cursor_list}; type IntegerTrace = ValSpine; @@ -15,16 +14,21 @@ fn get_trace() -> ValSpine { { let mut batcher = ValBatcher::::new(None, 0); - batcher.push_into(vec![ + batcher.insert(&mut vec![ ((1, 2), 0, 1), ((2, 3), 1, 1), ((2, 3), 2, -1), ]); let batch_ts = &[1, 2, 3]; + let mut lower = Antichain::from_elem(0); for i in batch_ts { - let (mut chain, description) = batcher.seal(Antichain::from_elem(*i)); - trace.insert(Span::new(description, IntegerBuilder::seal(&mut chain).map(Into::into))); + let upper = Antichain::from_elem(*i); + let (chain, _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; } } trace diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs index c155a9303..77cd816b4 100644 --- a/dogsdogsdogs/src/operators/half_join.rs +++ b/dogsdogsdogs/src/operators/half_join.rs @@ -33,7 +33,7 @@ use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable} use differential_dataflow::difference::Semigroup; use differential_dataflow::lattice::Lattice; use differential_dataflow::operators::arrange::Arranged; -use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchVal, Cursor, Navigable, TraceReader}; +use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchVal, Batcher, Cursor, Navigable, TraceReader}; use differential_dataflow::trace::cursor::cursor_list; use differential_dataflow::consolidation::{consolidate, consolidate_updates}; use differential_dataflow::trace::implementations::BatchContainer; @@ -51,28 +51,6 @@ pub trait HalfJoinTactic { fn prep(&mut self, chunks: Vec, batches: Vec, lower: Antichain) -> Box>; } -/// A type capable of accepting containers of updates, and carving them out by time. -/// -/// The implementor is able to determine the meaning of extraction by a frontier; -/// it is not required to be by antichain partial order. -/// -/// Updates are accepted as `C0`, the containers that arrive on the dataflow edge, and released as -/// `C1`, whatever a [`HalfJoinTactic`] would rather consume. The two need not agree: an implementor -/// staging updates in a form of its own can release that form directly. -pub trait Batcher { - /// Moves responsibility for `container` into the implementor. - fn insert(&mut self, container: C0); - /// Extracts updates `frontier` unblocks, and lower bounds the time of retained updates. - /// - /// What `frontier` unblocks is the implementor's to decide. It can be based on the antichain up set, - /// or it can be based on the total order of times (as used in delta join constructions). - /// - /// The reported lower bound antichain 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(&mut self, frontier: AntichainRef<'_, T>) -> (Vec, &MutableAntichain); -} - /// A `half_join` driven by a [`Batcher`] and a [`HalfJoinTactic`]. /// /// The operator introduces all streamed updates to the `batcher`, and then extracts all updates @@ -99,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, @@ -133,7 +111,7 @@ where // TODO: Tolerate multi-capability inputs. input1.for_each(|capability, data| { caps.insert(capability.retain(0)); - batcher.insert(std::mem::take(data)); + batcher.insert(data); }); // Drain input batches. We do not capture the batches, but we do use the frontier. @@ -143,7 +121,7 @@ where // Look for updates that are newly eligible against the current frontier. let (chunks, retained) = batcher.extract(frontier2.frontier()); - if !chunks.is_empty() { + if let Some(chunks) = chunks { // The batches are handed to the tactic, which holds them for as long as the // work item lives; what it joins against cannot change underneath it. let batches = trace.batches_through(Antichain::new().borrow()).unwrap(); @@ -153,7 +131,7 @@ where } // Downgrade capabilities to those held by `batcher`. - caps.downgrade(retained.frontier().iter()); + caps.downgrade(retained.iter()); } // Perform some amount of outstanding work, shipping each container at a capability @@ -506,11 +484,11 @@ pub mod cursors { } } - impl Batcher, Vec<(D, T, R)>> for BlobList { - fn insert(&mut self, container: Vec<(D, T, R)>) { - self.stage.extend(container.into_iter().map(|(d,t,r)| (t,d,r))); + impl Batcher, Vec>> for BlobList { + fn insert(&mut self, container: &mut Vec<(D, T, R)>) { + self.stage.extend(container.drain(..).map(|(d,t,r)| (t,d,r))); } - fn extract(&mut self, frontier: AntichainRef<'_, T>) -> (Vec>, &MutableAntichain) { + fn extract<'a>(&'a mut self, frontier: AntichainRef<'_, T>) -> (Option>>, AntichainRef<'a, T>) { // Handle any staged updates first. consolidate_updates(&mut self.stage); if !self.stage.is_empty() { @@ -545,7 +523,8 @@ pub mod cursors { self.blobs.retain(|b| !b.is_empty()); self.lower.update_iter(result.iter().flat_map(|l| l.iter().map(|x| (x.1.clone(), -1)))); - (result, &self.lower) + let result = if result.is_empty() { None } else { Some(result) }; + (result, self.lower.frontier()) } } diff --git a/experiments/src/bin/deals.rs b/experiments/src/bin/deals.rs index 045dd623a..a6502baf4 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::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(); + let graph = graph.arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(edges.clone(), |_,&x,&z| Some((x, z))) - .arrange::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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 b7b7cb356..28e96c818 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::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(); + let edges = edges.arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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 07783d244..d087c7a7b 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::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(); + let dereference = dereference.arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>(); - let memory_alias_arranged = memory_alias_collection.arrange::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(); + let value_flow_arranged = value_flow_collection.arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new); + let memory_alias_arranged = memory_alias_collection.arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(memory_alias_arranged, |_,&a,&b| Some((b,a))) .concat(assignment.map(|(a,b)| (b,a))) - .arrange::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>(); + let dereference = dereference.arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>(); - let memory_alias_arranged = memory_alias_collection.arrange::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(); + 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); // 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(memory_alias_arranged.clone(), |_,&a,&b| Some((b,a))) .concat(assignment.map(|(a,b)| (b,a))) - .arrange::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(ValBatcher::new) .join_core(dereference, |_x,&a,&b| Some((a,b))) - .arrange::, ValBuilder<_,_,_,_>, ValSpine<_,_,_,_>>(); + .arrange::<_, ValBuilder<_,_,_,_>, 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<_,_,_,_>>() + .arrange::<_, ValBuilder<_,_,_,_>, 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 385ee9918..d3cedb101 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -295,10 +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, ChunkBatcher>, ChunkBuilder>, CTrace>( + arrange_core::<_, CC, CorgiChunker, _, ChunkBuilder>, CTrace>( c.inner, Pipeline, "CorgiArrange", + ChunkBatcher::new, ) }