From 3f22b6b2579e2d01125483c2c87839b6b3a38eb4 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 27 Aug 2026 18:00:30 -0400 Subject: [PATCH 1/5] Relocate Batcher to batcher/ --- differential-dataflow/benches/chunk_bench.rs | 2 +- differential-dataflow/examples/spines.rs | 2 +- .../merge}/chunker.rs | 2 +- .../src/batcher/merge/mod.rs | 466 ++++++++++++++++++ differential-dataflow/src/batcher/mod.rs | 37 ++ differential-dataflow/src/collection.rs | 6 +- differential-dataflow/src/lib.rs | 1 + .../src/operators/arrange/arrangement.rs | 3 +- differential-dataflow/src/trace/chunk/mod.rs | 12 +- differential-dataflow/src/trace/chunk/vec.rs | 2 +- .../implementations/merge_batcher/mod.rs | 458 +---------------- .../src/trace/implementations/mod.rs | 2 +- .../src/trace/implementations/ord_neu.rs | 10 +- differential-dataflow/src/trace/mod.rs | 33 +- differential-dataflow/tests/trace.rs | 3 +- dogsdogsdogs/src/operators/half_join.rs | 3 +- 16 files changed, 532 insertions(+), 510 deletions(-) rename differential-dataflow/src/{trace/implementations/merge_batcher => batcher/merge}/chunker.rs (96%) create mode 100644 differential-dataflow/src/batcher/merge/mod.rs create mode 100644 differential-dataflow/src/batcher/mod.rs diff --git a/differential-dataflow/benches/chunk_bench.rs b/differential-dataflow/benches/chunk_bench.rs index 073fec3e3..140610a24 100644 --- a/differential-dataflow/benches/chunk_bench.rs +++ b/differential-dataflow/benches/chunk_bench.rs @@ -26,7 +26,7 @@ use differential_dataflow::trace::chunk::{merge_chains, Chunk, NavigableChunk}; use differential_dataflow::trace::chunk::vec::VecChunk; use differential_dataflow::columnar::trace::ColChunk; use differential_dataflow::trace::cursor::Cursor; -use differential_dataflow::trace::implementations::merge_batcher::chunker::ContainerChunker; +use differential_dataflow::batcher::merge::chunker::ContainerChunker; /// A global allocator that tracks currently-resident bytes, so we can snapshot /// the heap footprint of a built chunk chain. diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs index ac10400dc..17732826c 100644 --- a/differential-dataflow/examples/spines.rs +++ b/differential-dataflow/examples/spines.rs @@ -81,7 +81,7 @@ fn main() { use differential_dataflow::Hashable; use differential_dataflow::columnar::trace::{Spine, ColChunk}; use differential_dataflow::trace::chunk::ChunkBatcher; - use differential_dataflow::trace::implementations::merge_batcher::chunker::ContainerChunker; + use differential_dataflow::batcher::merge::chunker::ContainerChunker; use differential_dataflow::operators::arrange::arrangement::arrange_core; use timely::dataflow::channels::pact::Exchange; diff --git a/differential-dataflow/src/trace/implementations/merge_batcher/chunker.rs b/differential-dataflow/src/batcher/merge/chunker.rs similarity index 96% rename from differential-dataflow/src/trace/implementations/merge_batcher/chunker.rs rename to differential-dataflow/src/batcher/merge/chunker.rs index 7925e75ed..75d5c1c64 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher/chunker.rs +++ b/differential-dataflow/src/batcher/merge/chunker.rs @@ -1,4 +1,4 @@ -//! Organizes streams of data into sorted chunks for a merge batcher. +//! Organizes streams of data into sorted chunks for a [`MergeBatcher`](super::MergeBatcher). use std::collections::VecDeque; diff --git a/differential-dataflow/src/batcher/merge/mod.rs b/differential-dataflow/src/batcher/merge/mod.rs new file mode 100644 index 000000000..4884e4c9d --- /dev/null +++ b/differential-dataflow/src/batcher/merge/mod.rs @@ -0,0 +1,466 @@ +//! A `Batcher` implementation based on merge sort. +//! +//! The `MergeBatcher` requires a "merger" that implements the [`Merger`] trait, which provides +//! hooks for manipulating sorted "chains" of chunks as needed by the merge batcher: merging +//! chunks and also splitting them apart based on time. +//! +//! 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. + +pub mod chunker; + +use timely::container::{ContainerBuilder, PushInto}; +use timely::progress::frontier::AntichainRef; +use timely::progress::{frontier::Antichain, Timestamp}; + +use crate::logging::{BatcherEvent, Logger}; +use crate::batcher::Batcher; + +/// Creates batches from chunks of sorted, consolidated tuples. +/// +/// Chunking input is `Chu`'s business, merging chunks is `M`'s, and sealing the extracted chain +/// into a batch is `S`'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 + /// chains by updates, not chunk counts, since regrading decouples the two. A + /// chain is immutable until merged, so the weight is computed once at push. + /// + /// Do not push/pop directly but use the corresponding functions ([`Self::chain_push`]/[`Self::chain_pop`]). + chains: Vec<(usize, Vec)>, + /// Stash of empty chunks, recycled through the merging process. + stash: Vec, + /// Merges consolidated chunks, and extracts the subset of an update chain that lies in an interval of time. + merger: M, + /// The lower-bound frontier of the data, after the last call to extract. + frontier: Antichain, + /// Logger for size accounting. + logger: Option, + /// Timely operator ID. + operator_id: usize, + /// Seals each extracted chain into a batch. + sealer: std::marker::PhantomData, +} + +impl Batcher for MergeBatcher +where + M: Merger, + Chu: ContainerBuilder + for<'a> PushInto<&'a mut C>, + S: Sealer, +{ + type Time = M::Time; + type Output = S::Output; + + 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 + // `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>) { + // 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(); + let list2 = self.chain_pop().unwrap(); + let merged = self.merge_by(list1, list2); + self.chain_push(merged); + } + let merged = self.chain_pop().unwrap_or_default(); + + // Extract readied data. + let mut kept = Vec::new(); + let mut readied = Vec::new(); + self.frontier.clear(); + + self.merger.extract(merged, upper, &mut self.frontier, &mut readied, &mut kept, &mut self.stash); + + if !kept.is_empty() { + self.chain_push(kept); + } + + self.stash.clear(); + + (S::seal(&mut readied), self.frontier.borrow()) + } +} + +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(), + chunker: Chu::default(), + chains: Vec::new(), + stash: Vec::new(), + frontier: Antichain::new(), + sealer: 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) { + if !chain.is_empty() { + self.chain_push(chain); + while self.chains.len() > 1 && (self.chains[self.chains.len() - 1].0 >= self.chains[self.chains.len() - 2].0 / 2) { + let list1 = self.chain_pop().unwrap(); + let list2 = self.chain_pop().unwrap(); + let merged = self.merge_by(list1, list2); + self.chain_push(merged); + } + } + } + + // merges two sorted input lists into one sorted output list. + fn merge_by(&mut self, list1: Vec, list2: Vec) -> Vec { + // TODO: `list1` and `list2` get dropped; would be better to reuse? + let mut output = Vec::with_capacity(list1.len() + list2.len()); + self.merger.merge(list1, list2, &mut output, &mut self.stash); + + output + } + + /// Pop a chain and account size changes. + #[inline] + fn chain_pop(&mut self) -> Option> { + let (_weight, chain) = self.chains.pop()?; + self.account(chain.iter().map(Self::record), -1); + Some(chain) + } + + /// Push a chain and account size changes. + /// + /// Caches the chain's summed update count alongside it for the ladder. + #[inline] + fn chain_push(&mut self, chain: Vec) { + let weight = chain.iter().map(M::len).sum(); + self.account(chain.iter().map(Self::record), 1); + self.chains.push((weight, chain)); + } + + /// The `(records, size, capacity, allocations)` logger tuple for one chunk, + /// assembled from the two focused `Merger` methods. + #[inline] + fn record(chunk: &M::Chunk) -> (usize, usize, usize, usize) { + let (size, capacity, allocations) = M::allocation(chunk); + (M::len(chunk), size, capacity, allocations) + } + + /// Account size changes. Only performs work if a logger exists. + /// + /// Calculate the size based on the iterator passed along, with each attribute + /// multiplied by `diff`. Usually, one wants to pass 1 or -1 as the diff. + #[inline] + fn account>(&self, items: I, diff: isize) { + if let Some(logger) = &self.logger { + let (mut records, mut size, mut capacity, mut allocations) = (0isize, 0isize, 0isize, 0isize); + for (records_, size_, capacity_, allocations_) in items { + records = records.saturating_add_unsigned(records_); + size = size.saturating_add_unsigned(size_); + capacity = capacity.saturating_add_unsigned(capacity_); + allocations = allocations.saturating_add_unsigned(allocations_); + } + logger.log(BatcherEvent { + operator: self.operator_id, + records_diff: records * diff, + size_diff: size * diff, + capacity_diff: capacity * diff, + allocations_diff: allocations * diff, + }) + } + } +} + +impl Drop for MergeBatcher { + fn drop(&mut self) { + // Cleanup chain to retract accounting information. + while self.chain_pop().is_some() {} + } +} + +/// A trait to describe interesting moments in a merge batcher. +pub trait Merger: Default { + /// The internal representation of chunks of data. + type Chunk: Default; + /// The type of time in frontiers to extract updates. + type Time; + /// Merge chains into an output chain. + fn merge(&mut self, list1: Vec, list2: Vec, output: &mut Vec, stash: &mut Vec); + /// Extract ready updates based on the `upper` frontier. + fn extract( + &mut self, + merged: Vec, + upper: AntichainRef, + frontier: &mut Antichain, + readied: &mut Vec, + kept: &mut Vec, + stash: &mut Vec, + ); + + /// The number of updates in a chunk. + /// + /// Drives the geometric ladder (chains are weighed by summed updates, not chunk + /// counts, since regrading decouples the two) and the `records` field of the + /// size logger. + fn len(chunk: &Self::Chunk) -> usize; + + /// Backing-allocation figures for a chunk: `(size, capacity, allocations)`, for + /// the size logger's memory telemetry. + /// + /// Defaults to zero — most chunk types do not track this. Override to report + /// real figures (e.g. Materialize's memory accounting). + fn allocation(_chunk: &Self::Chunk) -> (usize, usize, usize) { (0, 0, 0) } +} + +/// Forms a batch from a whole chain of updates at once. +/// +/// Named rather than a bare `fn(&mut Vec) -> Option` so that implementors can name the +/// batch they produce. There is no receiver: the chain goes in and the batch comes out, leaving +/// nowhere for an update to be retained. +pub trait Sealer { + /// Output batch type. + type Output; + + /// Builds a batch from a chain of updates. + /// + /// This method relies on the chain only containing updates greater or equal to the lower frontier, + /// and not greater or equal to the upper frontier, of the interval the caller means to describe. + /// Chains must also be sorted and consolidated. + /// + /// Having the whole chain in hand, an implementor can size itself before it fills. + fn seal(chain: &mut Vec) -> Option; +} + +/// A `Merger` implementation for vector update containers. +pub mod vec { + + use std::marker::PhantomData; + use timely::container::SizableContainer; + use timely::progress::frontier::{Antichain, AntichainRef}; + use timely::PartialOrder; + use crate::batcher::merge::Merger; + + /// A `Merger` implementation for `Vec<(D, T, R)>` that drains owned inputs. + pub struct VecMerger { + _marker: PhantomData<(D, T, R)>, + } + + impl Default for VecMerger { + fn default() -> Self { Self { _marker: PhantomData } } + } + + impl VecMerger { + /// The target capacity for output buffers, as a power of two. + /// + /// This amount is used to size vectors, where vectors not exactly this capacity are dropped. + /// If this is mis-set, there is the potential for more memory churn than anticipated. + fn target_capacity() -> usize { + timely::container::buffer::default_capacity::<(D, T, R)>().next_power_of_two() + } + /// Acquire a buffer with the target capacity. + fn empty(&self, stash: &mut Vec>) -> Vec<(D, T, R)> { + let target = Self::target_capacity(); + let mut container = stash.pop().unwrap_or_default(); + container.clear(); + // Reuse if at target; otherwise allocate fresh. + if container.capacity() != target { + container = Vec::with_capacity(target); + } + container + } + /// Refill `queue` from `iter` if empty. Recycles drained queues into `stash`. + fn refill(queue: &mut std::collections::VecDeque<(D, T, R)>, iter: &mut impl Iterator>, stash: &mut Vec>) { + if queue.is_empty() { + let target = Self::target_capacity(); + if stash.len() < 2 { + let mut recycled = Vec::from(std::mem::take(queue)); + recycled.clear(); + if recycled.capacity() == target { + stash.push(recycled); + } + } + if let Some(chunk) = iter.next() { + *queue = std::collections::VecDeque::from(chunk); + } + } + } + } + + impl Merger for VecMerger + where + D: Ord + Clone + 'static, + T: Ord + Clone + PartialOrder + 'static, + R: crate::difference::Semigroup + 'static, + { + type Chunk = Vec<(D, T, R)>; + type Time = T; + + fn merge( + &mut self, + list1: Vec>, + list2: Vec>, + output: &mut Vec>, + stash: &mut Vec>, + ) { + use std::cmp::Ordering; + use std::collections::VecDeque; + + let mut iter1 = list1.into_iter(); + let mut iter2 = list2.into_iter(); + let mut q1 = VecDeque::<(D,T,R)>::from(iter1.next().unwrap_or_default()); + let mut q2 = VecDeque::<(D,T,R)>::from(iter2.next().unwrap_or_default()); + + let mut result = self.empty(stash); + + // Merge while both queues are non-empty. + while let (Some((d1, t1, _)), Some((d2, t2, _))) = (q1.front(), q2.front()) { + match (d1, t1).cmp(&(d2, t2)) { + Ordering::Less => { + result.push(q1.pop_front().unwrap()); + } + Ordering::Greater => { + result.push(q2.pop_front().unwrap()); + } + Ordering::Equal => { + let (d, t, mut r1) = q1.pop_front().unwrap(); + let (_, _, r2) = q2.pop_front().unwrap(); + r1.plus_equals(&r2); + if !r1.is_zero() { + result.push((d, t, r1)); + } + } + } + + if result.at_capacity() { + output.push(std::mem::take(&mut result)); + result = self.empty(stash); + } + + // Refill emptied queues from their chains. + if q1.is_empty() { Self::refill(&mut q1, &mut iter1, stash); } + if q2.is_empty() { Self::refill(&mut q2, &mut iter2, stash); } + } + + // Push partial result and remaining data from both sides. + if !result.is_empty() { output.push(result); } + for q in [q1, q2] { + if !q.is_empty() { output.push(Vec::from(q)); } + } + output.extend(iter1); + output.extend(iter2); + } + + fn extract( + &mut self, + merged: Vec>, + upper: AntichainRef, + frontier: &mut Antichain, + ship: &mut Vec>, + kept: &mut Vec>, + stash: &mut Vec>, + ) { + let mut keep = self.empty(stash); + let mut ready = self.empty(stash); + + for mut chunk in merged { + // Go update-by-update to swap out full containers. + for (data, time, diff) in chunk.drain(..) { + if upper.less_equal(&time) { + frontier.insert_with(&time, |time| time.clone()); + keep.push((data, time, diff)); + } else { + ready.push((data, time, diff)); + } + if keep.at_capacity() { + kept.push(std::mem::take(&mut keep)); + keep = self.empty(stash); + } + if ready.at_capacity() { + ship.push(std::mem::take(&mut ready)); + ready = self.empty(stash); + } + } + // Recycle the now-empty chunk if it has the right capacity. + if chunk.capacity() == Self::target_capacity() { + stash.push(chunk); + } + } + if !keep.is_empty() { kept.push(keep); } + if !ready.is_empty() { ship.push(ready); } + } + + fn len(chunk: &Vec<(D, T, R)>) -> usize { chunk.len() } + } +} + +#[cfg(test)] +mod test { + use timely::progress::frontier::Antichain; + use crate::batcher::Batcher; + use super::{MergeBatcher, Sealer}; + use super::vec::VecMerger; + use crate::batcher::merge::chunker::ContainerChunker; + + type In = Vec<((u64, ()), u64, i64)>; + type Bt = MergeBatcher, VecMerger<(u64, ()), u64, i64>, TestSealer>; + + struct TestSealer; + + impl Sealer for TestSealer { + type Output = Vec; + + fn seal(chain: &mut Vec) -> Option { + (!chain.is_empty()).then(|| std::mem::take(chain)) + } + } + + /// 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 `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] + 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)]]); + 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:?})"); + } + + /// Sanity: with no cross-chain cancellation, the frontier is the minimal kept time. + #[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)]]); + 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/batcher/mod.rs b/differential-dataflow/src/batcher/mod.rs new file mode 100644 index 000000000..d5e7a629a --- /dev/null +++ b/differential-dataflow/src/batcher/mod.rs @@ -0,0 +1,37 @@ +//! Traits and implementations for forming batches from streams of updates. + +use timely::progress::frontier::AntichainRef; + +pub mod merge; + +/// 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 +/// `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 { + /// 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 + /// 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 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<'_, Self::Time>) -> (Option, AntichainRef<'a, Self::Time>); +} diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs index abe41b3fc..356f1346f 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -965,7 +965,7 @@ pub mod vec { /// 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 where - Ba: crate::trace::Batcher, Time = T, Output: Into> + 'static, + Ba: crate::batcher::Batcher, Time = T, Output: Into> + 'static, Tr: crate::trace::Trace+'static, for<'a> BatchCursor: Cursor, F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static, @@ -1036,7 +1036,7 @@ pub mod vec { /// directly. pub fn arrange(self, batcher: impl FnOnce(Option, usize) -> Ba) -> Arranged<'scope, TraceAgent> where - Ba: crate::trace::Batcher, Time = T, Output: Into> + 'static, + Ba: crate::batcher::Batcher, Time = T, Output: Into> + 'static, Tr: crate::trace::Trace + 'static, { self.arrange_named::("Arrange", batcher) @@ -1045,7 +1045,7 @@ pub mod vec { /// 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> where - Ba: crate::trace::Batcher, Time = T, Output: Into> + 'static, + Ba: crate::batcher::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()); diff --git a/differential-dataflow/src/lib.rs b/differential-dataflow/src/lib.rs index 81db82af7..26a74c4f8 100644 --- a/differential-dataflow/src/lib.rs +++ b/differential-dataflow/src/lib.rs @@ -96,6 +96,7 @@ pub mod hashable; pub mod operators; pub mod algorithms; pub mod lattice; +pub mod batcher; pub mod trace; pub mod input; pub mod difference; diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index c553aeac5..75eba8177 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -32,7 +32,8 @@ use crate::{Data, VecCollection, AsCollection}; use crate::difference::Semigroup; use crate::lattice::Lattice; use crate::logging::Logger; -use crate::trace::{self, Description, SpanOf, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; +use crate::batcher::Batcher; +use crate::trace::{self, Description, SpanOf, Trace, TraceReader, Navigable, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; use trace::wrappers::enter::{TraceEnter, enter_span}; diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index 496a79dda..977747f93 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -21,7 +21,7 @@ //! These are the `Batcher` / `Builder` / `Spine` to hand to //! [`arrange_core`](crate::operators::arrange::arrangement::arrange_core), along with a //! chunker that forms `C` from the input stream — typically -//! [`ContainerChunker`](crate::trace::implementations::merge_batcher::chunker::ContainerChunker). +//! [`ContainerChunker`](crate::batcher::merge::chunker::ContainerChunker). //! Trace *maintenance* needs only [`Chunk`]; cursor-driven *consumption* of the //! arrangement additionally asks `C` for the [`NavigableChunk`] capability. //! Everything else here ([`ChunkBatch`], [`ChunkMerger`], [`ChunkBatchMerger`], @@ -270,14 +270,14 @@ where fn len(&self) -> usize { self.chunks.iter().map(C::len).sum() } } -/// A merge-batcher [`Merger`](crate::trace::implementations::merge_batcher::Merger) +/// A merge-batcher [`Merger`](crate::batcher::merge::Merger) /// over chains of [`Chunk`]s. /// /// `merge` runs the whole-chain binary merger; `extract` splits by the seal frontier /// 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::batcher::merge::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>>; @@ -470,7 +470,7 @@ impl Cursor for ChunkBatchCursor { } } -/// A merge-batcher [`Merger`](crate::trace::implementations::merge_batcher::Merger) +/// A merge-batcher [`Merger`](crate::batcher::merge::Merger) /// over chains of [`Chunk`]s. /// /// `merge` runs the whole-chain binary merger; `extract` splits by the seal frontier @@ -485,7 +485,7 @@ impl Default for ChunkMerger { fn default() -> Self { Self { _marker: std::marker::PhantomData } } } -impl crate::trace::implementations::merge_batcher::Merger for ChunkMerger +impl crate::batcher::merge::Merger for ChunkMerger where C: Chunk + Default + 'static, C::Time: Clone + timely::PartialOrder + 'static, @@ -688,7 +688,7 @@ where } -impl crate::trace::implementations::merge_batcher::Sealer for ChunkBatchBuilder +impl crate::batcher::merge::Sealer for ChunkBatchBuilder where C: Chunk + Default + 'static, C::Time: timely::progress::Timestamp, diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs index df9c6c51e..e0969d57f 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>, VecChunk>; +pub type ChunkBatcher = super::ChunkBatcher>, VecChunk>; /// Batch builder. pub type ChunkBuilder = super::ChunkBuilder>; diff --git a/differential-dataflow/src/trace/implementations/merge_batcher/mod.rs b/differential-dataflow/src/trace/implementations/merge_batcher/mod.rs index 1f9182630..02375efcf 100644 --- a/differential-dataflow/src/trace/implementations/merge_batcher/mod.rs +++ b/differential-dataflow/src/trace/implementations/merge_batcher/mod.rs @@ -1,457 +1,3 @@ -//! A `Batcher` implementation based on merge sort. -//! -//! The `MergeBatcher` requires a "merger" that implements the [`Merger`] trait, which provides -//! hooks for manipulating sorted "chains" of chunks as needed by the merge batcher: merging -//! chunks and also splitting them apart based on time. -//! -//! 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. +//! Compatibility re-exports for the merge batcher implementation. -pub mod chunker; - -use timely::container::{ContainerBuilder, PushInto}; -use timely::progress::frontier::AntichainRef; -use timely::progress::{frontier::Antichain, Timestamp}; - -use crate::logging::{BatcherEvent, Logger}; -use crate::trace::Batcher; - -/// Creates batches from chunks of sorted, consolidated tuples. -/// -/// Chunking input is `Chu`'s business, merging chunks is `M`'s, and sealing the extracted chain -/// into a batch is `S`'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 - /// chains by updates, not chunk counts, since regrading decouples the two. A - /// chain is immutable until merged, so the weight is computed once at push. - /// - /// Do not push/pop directly but use the corresponding functions ([`Self::chain_push`]/[`Self::chain_pop`]). - chains: Vec<(usize, Vec)>, - /// Stash of empty chunks, recycled through the merging process. - stash: Vec, - /// Merges consolidated chunks, and extracts the subset of an update chain that lies in an interval of time. - merger: M, - /// The lower-bound frontier of the data, after the last call to extract. - frontier: Antichain, - /// Logger for size accounting. - logger: Option, - /// Timely operator ID. - operator_id: usize, - /// Seals each extracted chain into a batch. - sealer: std::marker::PhantomData, -} - -impl Batcher for MergeBatcher -where - M: Merger, - Chu: ContainerBuilder + for<'a> PushInto<&'a mut C>, - S: Sealer, -{ - type Time = M::Time; - type Output = S::Output; - - 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 - // `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>) { - // 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(); - let list2 = self.chain_pop().unwrap(); - let merged = self.merge_by(list1, list2); - self.chain_push(merged); - } - let merged = self.chain_pop().unwrap_or_default(); - - // Extract readied data. - let mut kept = Vec::new(); - let mut readied = Vec::new(); - self.frontier.clear(); - - self.merger.extract(merged, upper, &mut self.frontier, &mut readied, &mut kept, &mut self.stash); - - if !kept.is_empty() { - self.chain_push(kept); - } - - self.stash.clear(); - - (S::seal(&mut readied), self.frontier.borrow()) - } -} - -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(), - chunker: Chu::default(), - chains: Vec::new(), - stash: Vec::new(), - frontier: Antichain::new(), - sealer: 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) { - if !chain.is_empty() { - self.chain_push(chain); - while self.chains.len() > 1 && (self.chains[self.chains.len() - 1].0 >= self.chains[self.chains.len() - 2].0 / 2) { - let list1 = self.chain_pop().unwrap(); - let list2 = self.chain_pop().unwrap(); - let merged = self.merge_by(list1, list2); - self.chain_push(merged); - } - } - } - - // merges two sorted input lists into one sorted output list. - fn merge_by(&mut self, list1: Vec, list2: Vec) -> Vec { - // TODO: `list1` and `list2` get dropped; would be better to reuse? - let mut output = Vec::with_capacity(list1.len() + list2.len()); - self.merger.merge(list1, list2, &mut output, &mut self.stash); - - output - } - - /// Pop a chain and account size changes. - #[inline] - fn chain_pop(&mut self) -> Option> { - let (_weight, chain) = self.chains.pop()?; - self.account(chain.iter().map(Self::record), -1); - Some(chain) - } - - /// Push a chain and account size changes. - /// - /// Caches the chain's summed update count alongside it for the ladder. - #[inline] - fn chain_push(&mut self, chain: Vec) { - let weight = chain.iter().map(M::len).sum(); - self.account(chain.iter().map(Self::record), 1); - self.chains.push((weight, chain)); - } - - /// The `(records, size, capacity, allocations)` logger tuple for one chunk, - /// assembled from the two focused `Merger` methods. - #[inline] - fn record(chunk: &M::Chunk) -> (usize, usize, usize, usize) { - let (size, capacity, allocations) = M::allocation(chunk); - (M::len(chunk), size, capacity, allocations) - } - - /// Account size changes. Only performs work if a logger exists. - /// - /// Calculate the size based on the iterator passed along, with each attribute - /// multiplied by `diff`. Usually, one wants to pass 1 or -1 as the diff. - #[inline] - fn account>(&self, items: I, diff: isize) { - if let Some(logger) = &self.logger { - let (mut records, mut size, mut capacity, mut allocations) = (0isize, 0isize, 0isize, 0isize); - for (records_, size_, capacity_, allocations_) in items { - records = records.saturating_add_unsigned(records_); - size = size.saturating_add_unsigned(size_); - capacity = capacity.saturating_add_unsigned(capacity_); - allocations = allocations.saturating_add_unsigned(allocations_); - } - logger.log(BatcherEvent { - operator: self.operator_id, - records_diff: records * diff, - size_diff: size * diff, - capacity_diff: capacity * diff, - allocations_diff: allocations * diff, - }) - } - } -} - -impl Drop for MergeBatcher { - fn drop(&mut self) { - // Cleanup chain to retract accounting information. - while self.chain_pop().is_some() {} - } -} - -/// A trait to describe interesting moments in a merge batcher. -pub trait Merger: Default { - /// The internal representation of chunks of data. - type Chunk: Default; - /// The type of time in frontiers to extract updates. - type Time; - /// Merge chains into an output chain. - fn merge(&mut self, list1: Vec, list2: Vec, output: &mut Vec, stash: &mut Vec); - /// Extract ready updates based on the `upper` frontier. - fn extract( - &mut self, - merged: Vec, - upper: AntichainRef, - frontier: &mut Antichain, - readied: &mut Vec, - kept: &mut Vec, - stash: &mut Vec, - ); - - /// The number of updates in a chunk. - /// - /// Drives the geometric ladder (chains are weighed by summed updates, not chunk - /// counts, since regrading decouples the two) and the `records` field of the - /// size logger. - fn len(chunk: &Self::Chunk) -> usize; - - /// Backing-allocation figures for a chunk: `(size, capacity, allocations)`, for - /// the size logger's memory telemetry. - /// - /// Defaults to zero — most chunk types do not track this. Override to report - /// real figures (e.g. Materialize's memory accounting). - fn allocation(_chunk: &Self::Chunk) -> (usize, usize, usize) { (0, 0, 0) } -} - -/// Forms a batch from a whole chain of updates at once. -/// -/// Named rather than a bare `fn(&mut Vec) -> Option` so that implementors can name the -/// batch they produce. There is no receiver: the chain goes in and the batch comes out, leaving -/// nowhere for an update to be retained. -pub trait Sealer { - /// Output batch type. - type Output; - - /// Builds a batch from a chain of updates. - /// - /// This method relies on the chain only containing updates greater or equal to the lower frontier, - /// and not greater or equal to the upper frontier, of the interval the caller means to describe. - /// Chains must also be sorted and consolidated. - /// - /// Having the whole chain in hand, an implementor can size itself before it fills. - fn seal(chain: &mut Vec) -> Option; -} - -/// A `Merger` implementation for vector update containers. -pub mod vec { - - use std::marker::PhantomData; - use timely::container::SizableContainer; - use timely::progress::frontier::{Antichain, AntichainRef}; - use timely::PartialOrder; - use crate::trace::implementations::merge_batcher::Merger; - - /// A `Merger` implementation for `Vec<(D, T, R)>` that drains owned inputs. - pub struct VecMerger { - _marker: PhantomData<(D, T, R)>, - } - - impl Default for VecMerger { - fn default() -> Self { Self { _marker: PhantomData } } - } - - impl VecMerger { - /// The target capacity for output buffers, as a power of two. - /// - /// This amount is used to size vectors, where vectors not exactly this capacity are dropped. - /// If this is mis-set, there is the potential for more memory churn than anticipated. - fn target_capacity() -> usize { - timely::container::buffer::default_capacity::<(D, T, R)>().next_power_of_two() - } - /// Acquire a buffer with the target capacity. - fn empty(&self, stash: &mut Vec>) -> Vec<(D, T, R)> { - let target = Self::target_capacity(); - let mut container = stash.pop().unwrap_or_default(); - container.clear(); - // Reuse if at target; otherwise allocate fresh. - if container.capacity() != target { - container = Vec::with_capacity(target); - } - container - } - /// Refill `queue` from `iter` if empty. Recycles drained queues into `stash`. - fn refill(queue: &mut std::collections::VecDeque<(D, T, R)>, iter: &mut impl Iterator>, stash: &mut Vec>) { - if queue.is_empty() { - let target = Self::target_capacity(); - if stash.len() < 2 { - let mut recycled = Vec::from(std::mem::take(queue)); - recycled.clear(); - if recycled.capacity() == target { - stash.push(recycled); - } - } - if let Some(chunk) = iter.next() { - *queue = std::collections::VecDeque::from(chunk); - } - } - } - } - - impl Merger for VecMerger - where - D: Ord + Clone + 'static, - T: Ord + Clone + PartialOrder + 'static, - R: crate::difference::Semigroup + 'static, - { - type Chunk = Vec<(D, T, R)>; - type Time = T; - - fn merge( - &mut self, - list1: Vec>, - list2: Vec>, - output: &mut Vec>, - stash: &mut Vec>, - ) { - use std::cmp::Ordering; - use std::collections::VecDeque; - - let mut iter1 = list1.into_iter(); - let mut iter2 = list2.into_iter(); - let mut q1 = VecDeque::<(D,T,R)>::from(iter1.next().unwrap_or_default()); - let mut q2 = VecDeque::<(D,T,R)>::from(iter2.next().unwrap_or_default()); - - let mut result = self.empty(stash); - - // Merge while both queues are non-empty. - while let (Some((d1, t1, _)), Some((d2, t2, _))) = (q1.front(), q2.front()) { - match (d1, t1).cmp(&(d2, t2)) { - Ordering::Less => { - result.push(q1.pop_front().unwrap()); - } - Ordering::Greater => { - result.push(q2.pop_front().unwrap()); - } - Ordering::Equal => { - let (d, t, mut r1) = q1.pop_front().unwrap(); - let (_, _, r2) = q2.pop_front().unwrap(); - r1.plus_equals(&r2); - if !r1.is_zero() { - result.push((d, t, r1)); - } - } - } - - if result.at_capacity() { - output.push(std::mem::take(&mut result)); - result = self.empty(stash); - } - - // Refill emptied queues from their chains. - if q1.is_empty() { Self::refill(&mut q1, &mut iter1, stash); } - if q2.is_empty() { Self::refill(&mut q2, &mut iter2, stash); } - } - - // Push partial result and remaining data from both sides. - if !result.is_empty() { output.push(result); } - for q in [q1, q2] { - if !q.is_empty() { output.push(Vec::from(q)); } - } - output.extend(iter1); - output.extend(iter2); - } - - fn extract( - &mut self, - merged: Vec>, - upper: AntichainRef, - frontier: &mut Antichain, - ship: &mut Vec>, - kept: &mut Vec>, - stash: &mut Vec>, - ) { - let mut keep = self.empty(stash); - let mut ready = self.empty(stash); - - for mut chunk in merged { - // Go update-by-update to swap out full containers. - for (data, time, diff) in chunk.drain(..) { - if upper.less_equal(&time) { - frontier.insert_with(&time, |time| time.clone()); - keep.push((data, time, diff)); - } else { - ready.push((data, time, diff)); - } - if keep.at_capacity() { - kept.push(std::mem::take(&mut keep)); - keep = self.empty(stash); - } - if ready.at_capacity() { - ship.push(std::mem::take(&mut ready)); - ready = self.empty(stash); - } - } - // Recycle the now-empty chunk if it has the right capacity. - if chunk.capacity() == Self::target_capacity() { - stash.push(chunk); - } - } - if !keep.is_empty() { kept.push(keep); } - if !ready.is_empty() { ship.push(ready); } - } - - fn len(chunk: &Vec<(D, T, R)>) -> usize { chunk.len() } - } -} - -#[cfg(test)] -mod test { - use timely::progress::frontier::Antichain; - use crate::trace::Batcher; - use super::MergeBatcher; - use super::vec::VecMerger; - use crate::trace::implementations::ord_neu::VecOrdKeyBuilder; - use crate::trace::implementations::merge_batcher::chunker::ContainerChunker; - - 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 - /// 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] - 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)]]); - 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:?})"); - } - - /// Sanity: with no cross-chain cancellation, the frontier is the minimal kept time. - #[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)]]); - let (_, retained) = Batcher::::extract(&mut b, Antichain::from_elem(3).borrow()); - let got: Vec = retained.iter().cloned().collect(); - assert_eq!(got, vec![5u64]); - } -} +pub use crate::batcher::merge::{MergeBatcher, Merger, Sealer, chunker, vec}; diff --git a/differential-dataflow/src/trace/implementations/mod.rs b/differential-dataflow/src/trace/implementations/mod.rs index 57e35deb1..00b85eedf 100644 --- a/differential-dataflow/src/trace/implementations/mod.rs +++ b/differential-dataflow/src/trace/implementations/mod.rs @@ -44,7 +44,7 @@ pub mod merge_batcher; pub mod ord_neu; // Opinionated takes on default spines. -pub use self::merge_batcher::chunker::ContainerChunker; +pub use crate::batcher::merge::chunker::ContainerChunker; pub use self::ord_neu::OrdValSpine as ValSpine; pub use self::ord_neu::OrdValBatcher as ValBatcher; pub use self::ord_neu::VecOrdValBuilder as ValBuilder; diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index fa4136892..88fa4d446 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -11,9 +11,9 @@ use std::rc::Rc; use crate::trace::implementations::spine_fueled::Spine; -use crate::trace::implementations::merge_batcher::chunker::ContainerChunker; -use crate::trace::implementations::merge_batcher::MergeBatcher; -use crate::trace::implementations::merge_batcher::vec::VecMerger; +use crate::batcher::merge::chunker::ContainerChunker; +use crate::batcher::merge::MergeBatcher; +use crate::batcher::merge::vec::VecMerger; use super::{Layout, Vector}; @@ -694,7 +694,7 @@ pub mod val_batch { } - impl crate::trace::implementations::merge_batcher::Sealer for OrdValBuilder + impl crate::batcher::merge::Sealer for OrdValBuilder where L: for<'a> Layout< KeyContainer: PushInto>, @@ -728,7 +728,7 @@ pub mod key_batch { use crate::trace::implementations::spine_fueled::{SpineBatch, Merger}; use crate::trace::implementations::BatchContainer; use crate::trace::implementations::layout; - use crate::trace::implementations::merge_batcher::Sealer; + use crate::batcher::merge::Sealer; use super::{Layout, Upds, layers::UpdsBuilder, BuilderInput}; diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index f4cd5fd26..a6a8a521b 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -21,6 +21,7 @@ pub use self::cursor::Cursor; pub use self::cursor::Navigable; pub use self::cursor::{BatchCursor, BatchKey, BatchVal, BatchValOwn, BatchDiff, BatchDiffGat, BatchTimeGat}; pub use self::description::Description; +pub use crate::batcher::Batcher; /// A type used to express how much effort a trace should exert even in the absence of updates. pub type ExertionLogic = std::sync::Arc Fn(&'a [(usize, usize, usize)])->Option+Send+Sync>; @@ -225,38 +226,6 @@ pub trait Trace : TraceReader { fn close(&mut self); } -/// 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 -/// `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 { - /// 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 - /// 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 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<'_, Self::Time>) -> (Option, AntichainRef<'a, Self::Time>); -} - /// Functionality for building batches from ordered update sequences. /// /// `Default` is the empty builder; a builder discovers its output as it is pushed, and so has diff --git a/differential-dataflow/tests/trace.rs b/differential-dataflow/tests/trace.rs index 1fd9129f5..be9196d2f 100644 --- a/differential-dataflow/tests/trace.rs +++ b/differential-dataflow/tests/trace.rs @@ -2,7 +2,8 @@ use timely::dataflow::operators::generic::OperatorInfo; use timely::progress::{Antichain, frontier::AntichainRef}; use differential_dataflow::trace::implementations::{ValBatcher, ValSpine}; -use differential_dataflow::trace::{Description, Span, Trace, TraceReader, Batcher}; +use differential_dataflow::batcher::Batcher; +use differential_dataflow::trace::{Description, Span, Trace, TraceReader}; use differential_dataflow::trace::cursor::{Cursor, cursor_list}; type IntegerTrace = ValSpine; diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs index 588a6f571..72ceb53e5 100644 --- a/dogsdogsdogs/src/operators/half_join.rs +++ b/dogsdogsdogs/src/operators/half_join.rs @@ -33,7 +33,8 @@ 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, Batcher, Cursor, Navigable, TraceReader}; +use differential_dataflow::batcher::Batcher; +use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchVal, Cursor, Navigable, TraceReader}; use differential_dataflow::trace::cursor::cursor_list; use differential_dataflow::consolidation::{consolidate, consolidate_updates}; use differential_dataflow::trace::implementations::BatchContainer; From 5b52d3359519804252d5911c98ec401280a5306a Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Thu, 27 Aug 2026 19:02:24 -0400 Subject: [PATCH 2/5] Relocate cursor operator tactics --- .../src/operators/cursor/join.rs | 316 ++++++++++ .../src/operators/cursor/mod.rs | 4 + .../src/operators/cursor/reduce.rs | 559 +++++++++++++++++ differential-dataflow/src/operators/join.rs | 316 +--------- differential-dataflow/src/operators/mod.rs | 1 + differential-dataflow/src/operators/reduce.rs | 561 +----------------- 6 files changed, 889 insertions(+), 868 deletions(-) create mode 100644 differential-dataflow/src/operators/cursor/join.rs create mode 100644 differential-dataflow/src/operators/cursor/mod.rs create mode 100644 differential-dataflow/src/operators/cursor/reduce.rs diff --git a/differential-dataflow/src/operators/cursor/join.rs b/differential-dataflow/src/operators/cursor/join.rs new file mode 100644 index 000000000..ee6225c8b --- /dev/null +++ b/differential-dataflow/src/operators/cursor/join.rs @@ -0,0 +1,316 @@ +//! Cursor-based join implementation. + +use std::cell::RefCell; +use std::cmp::Ordering; +use std::collections::VecDeque; +use std::rc::Rc; + +use timely::ContainerBuilder; +use timely::dataflow::Stream; +use timely::progress::Timestamp; + +use crate::lattice::Lattice; +use crate::operators::ValueHistory; +use crate::operators::arrange::Arranged; +use crate::operators::join::{Fresh, JoinTactic, join_with_tactic}; +use crate::trace::{BatchCursor, BatchDiff, BatchVal, Cursor, Navigable, TraceReader}; +use crate::trace::cursor::cursor_list; +use crate::trace::implementations::containers::BatchContainer; + +/// An equijoin of two traces, sharing a common key type. +/// +/// This method exists to provide join functionality without opinions on the specific input types, keys and values, +/// that should be presented. The two traces here can have arbitrary key and value types, which can be unsized and +/// even potentially unrelated to the input collection data. Importantly, the key and value types could be generic +/// associated types (GATs) of the traces, and we would seemingly struggle to frame these types as trait arguments. +/// +/// The implementation produces a caller-specified container. Implementations can use [`AsCollection`] to wrap the +/// output stream in a collection. +/// +/// The "correctness" of this method depends heavily on the behavior of the supplied `result` function. +/// +/// [`AsCollection`]: crate::collection::AsCollection +pub fn join_traces<'scope, Tr1, Tr2, KC, L, CB>(arranged1: Arranged<'scope, Tr1>, arranged2: Arranged<'scope, Tr2>, name: &str, result: L) -> Stream<'scope, Tr1::Time, CB::Container> +where + Tr1: TraceReader+'static, + Tr2: TraceReader+'static, + KC: BatchContainer, + BatchCursor: Cursor