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/operators/cursor/history.rs b/differential-dataflow/src/operators/cursor/history.rs new file mode 100644 index 000000000..f4592b7de --- /dev/null +++ b/differential-dataflow/src/operators/cursor/history.rs @@ -0,0 +1,73 @@ +//! Loading adapters between cursors and value histories. + +use crate::lattice::Lattice; +use crate::operators::history::{EditList, HistoryReplay, ValueHistory}; +use crate::trace::Cursor; + +/// Walks the cursor's values at the current key into `target`, advancing times by `meet` if supplied. +fn load_values<'a, V, T, D, C>( + target: &mut EditList, + cursor: &mut C, + storage: &'a C::Storage, + meet: Option<&T>, +) +where + V: Copy + Ord, + T: Ord + Clone + Lattice, + D: crate::difference::Semigroup, + C: Cursor = V, Time = T, Diff = D>, +{ + while let Some(val) = cursor.get_val(storage) { + cursor.map_times(storage, |time, diff| { + let mut time = C::owned_time(time); + if let Some(meet) = meet { time.join_assign(meet); } + target.push(time, C::owned_diff(diff)); + }); + target.seal(val); + cursor.step_val(storage); + } +} + +/// Loads the cursor's values at its current key into `history`. +/// +/// This avoids a redundant seek in the merge-join inner loop, where the cursor is positioned by the upstream merge step. +pub(super) fn load_current<'a, V, T, D, C>( + history: &mut ValueHistory, + cursor: &mut C, + storage: &'a C::Storage, + meet: Option<&T>, +) +where + V: Copy + Ord, + T: Ord + Clone + Lattice, + D: crate::difference::Semigroup, + C: Cursor = V, Time = T, Diff = D>, +{ + history.clear(); + load_values(history.edits_mut(), cursor, storage, meet); +} + +/// Loads and replays a specified key. +/// +/// If the key is absent, the replayed history will be empty. +pub(super) fn replay_key<'a, 'history, V, T, D, C>( + history: &'history mut ValueHistory, + cursor: &mut C, + storage: &'a C::Storage, + key: C::Key<'a>, + meet: Option<&T>, +) -> HistoryReplay<'history, V, T, D> +where + V: Copy + Ord, + T: Ord + Clone + Lattice, + D: crate::difference::Semigroup, + C: Cursor = V, Time = T, Diff = D>, +{ + history.clear(); + cursor.seek_key(storage, key); + if cursor.get_key(storage) == Some(key) { + cursor.rewind_vals(storage); + load_values(history.edits_mut(), cursor, storage, meet); + } + history.replay() +} diff --git a/differential-dataflow/src/operators/cursor/join.rs b/differential-dataflow/src/operators/cursor/join.rs new file mode 100644 index 000000000..cf4e41326 --- /dev/null +++ b/differential-dataflow/src/operators/cursor/join.rs @@ -0,0 +1,317 @@ +//! 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 super::history::load_current; +use crate::lattice::Lattice; +use crate::operators::arrange::Arranged; +use crate::operators::history::ValueHistory; +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