diff --git a/differential-dataflow/examples/cursors.rs b/differential-dataflow/examples/cursors.rs index 8dcfc739d..f4d209284 100644 --- a/differential-dataflow/examples/cursors.rs +++ b/differential-dataflow/examples/cursors.rs @@ -93,8 +93,8 @@ fn main() { } /* Return trace content after the last round. */ - let batches = graph_trace.batches_through(Antichain::new().borrow()).unwrap(); - let (mut cursor, storage) = cursor_list(batches); + let batches = graph_trace.spans_through(Antichain::new().borrow()).unwrap(); + let (mut cursor, storage) = cursor_list(batches.into_iter().filter_map(|b| b.inner).collect()); cursor.to_vec(&storage, |k| k.clone(), |v| v.clone()) }) .unwrap().join(); diff --git a/differential-dataflow/examples/multitemporal.rs b/differential-dataflow/examples/multitemporal.rs index 0850745c1..ffc9b2e06 100644 --- a/differential-dataflow/examples/multitemporal.rs +++ b/differential-dataflow/examples/multitemporal.rs @@ -99,8 +99,8 @@ fn main() { else { println!("Report at {:?}", query_time); // enumerate the contents of `trace` at `query_time`. - let batches = trace.batches_through(Antichain::new().borrow()).unwrap(); - let (mut cursor, storage) = cursor_list(batches); + let batches = trace.spans_through(Antichain::new().borrow()).unwrap(); + let (mut cursor, storage) = cursor_list(batches.into_iter().filter_map(|b| b.inner).collect()); while let Some(key) = cursor.get_key(&storage) { while let Some(_val) = cursor.get_val(&storage) { let mut sum = 0; diff --git a/differential-dataflow/examples/spines.rs b/differential-dataflow/examples/spines.rs index e90f12e1c..6bfe3b224 100644 --- a/differential-dataflow/examples/spines.rs +++ b/differential-dataflow/examples/spines.rs @@ -56,21 +56,21 @@ fn main() { match mode.as_str() { "key" => { - use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatcher, RcOrdKeyBuilder, OrdKeySpine}; + 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.arrange::, RcOrdKeyBuilder, OrdKeySpine>(); - let keys = keys.arrange::, RcOrdKeyBuilder, OrdKeySpine>(); + let data = data.arrange::, VecOrdKeyBuilder, OrdKeySpine>(); + let keys = keys.arrange::, VecOrdKeyBuilder, OrdKeySpine>(); keys.join_core(data, |_k, &(), &()| Option::<()>::None) .probe_with(&mut probe); Workload { data_input, keys_input } }, "val" => { - use differential_dataflow::trace::implementations::ord_neu::{OrdValBatcher, RcOrdValBuilder, OrdValSpine}; + 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::, RcOrdValBuilder, OrdValSpine>(); - let keys = keys.map(|x| (x, ())).arrange::, RcOrdValBuilder, OrdValSpine>(); + let data = data.map(|x| (x, ())).arrange::, VecOrdValBuilder, OrdValSpine>(); + let keys = keys.map(|x| (x, ())).arrange::, VecOrdValBuilder, OrdValSpine>(); 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 b01b92ca6..53ce59fdb 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -785,7 +785,7 @@ pub mod vec { where T2: Trace+'static, for<'a> BatchCursor: Cursor= &'a K, ValOwn = V, Time = T2::Time, Diff: Abelian>, - Bu: Builder)>, Output = T2::Batch> + 'static, + Bu: Builder)>, Output: Into> + 'static, L: FnMut(&K, &[(&V, R)], &mut Vec<(V, BatchDiff)>)+'static, { self.reduce_core::<_,Bu,T2>(name, move |key, input, output, change| { @@ -805,7 +805,7 @@ pub mod vec { V: Clone+'static, T2: Trace+'static, for<'a> BatchCursor: Cursor=&'a K, ValOwn = V, Time = T2::Time>, - Bu: Builder)>, Output = T2::Batch> + 'static, + Bu: Builder)>, Output: Into> + 'static, L: FnMut(&K, &[(&V, R)], &mut Vec<(V,BatchDiff)>, &mut Vec<(V, BatchDiff)>)+'static, { self.arrange_by_key_named(&format!("Arrange: {}", name)) @@ -968,7 +968,7 @@ pub mod vec { Ba: crate::trace::Batcher, Time=T> + 'static, Tr: crate::trace::Trace+'static, for<'a> BatchCursor: Cursor, - Bu: crate::trace::Builder, Output=Tr::Batch>, + Bu: crate::trace::Builder, Output: Into>, F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static, { use crate::operators::arrange::arrangement::Arrange; @@ -1036,7 +1036,7 @@ pub mod vec { fn arrange_named(self, name: &str) -> Arranged<'scope, TraceAgent> where Ba: crate::trace::Batcher, Time=T> + 'static, - Bu: crate::trace::Builder, Output = Tr::Batch>, + Bu: crate::trace::Builder, Output: Into>, Tr: crate::trace::Trace + 'static, { let exchange = timely::dataflow::channels::pact::Exchange::new(move |update: &((K,V),T,R)| (update.0).0.hashed().into()); @@ -1051,7 +1051,7 @@ pub mod vec { fn arrange_named(self, name: &str) -> Arranged<'scope, TraceAgent> where Ba: crate::trace::Batcher, Time=T> + 'static, - Bu: crate::trace::Builder, Output = Tr::Batch>, + Bu: crate::trace::Builder, Output: Into>, Tr: crate::trace::Trace + 'static, { let exchange = timely::dataflow::channels::pact::Exchange::new(move |update: &((K,()),T,R)| (update.0).0.hashed().into()); diff --git a/differential-dataflow/src/columnar/collection/operators.rs b/differential-dataflow/src/columnar/collection/operators.rs index c5f00d540..ff7b3f21a 100644 --- a/differential-dataflow/src/columnar/collection/operators.rs +++ b/differential-dataflow/src/columnar/collection/operators.rs @@ -169,6 +169,7 @@ where input.for_each(|time, batches| { let mut session = output.session_with_builder(&time); for batch in batches.drain(..) { + let Some(batch) = batch.inner else { continue }; let mut cursor = batch.cursor(); while cursor.key_valid(&batch) { while cursor.val_valid(&batch) { diff --git a/differential-dataflow/src/columnar/trace/chunk.rs b/differential-dataflow/src/columnar/trace/chunk.rs index 393058920..89c6e0fbc 100644 --- a/differential-dataflow/src/columnar/trace/chunk.rs +++ b/differential-dataflow/src/columnar/trace/chunk.rs @@ -546,6 +546,7 @@ fn advance_trie( #[cfg(test)] mod test { + use timely::progress::Antichain; use std::collections::VecDeque; use columnar::Push; use super::{ColChunk, Chunk}; @@ -647,18 +648,14 @@ mod test { #[test] fn cursor_handles_straddle() { use crate::trace::cursor::Cursor; - use crate::trace::Description; use crate::trace::chunk::ChunkBatch; - use timely::progress::Antichain; let chunks = vec![ chunk(vec![(0, 0, 0, 1), (1, 0, 0, 1), (1, 1, 0, 1)]), chunk(vec![(1, 1, 1, 1), (1, 2, 0, 1)]), chunk(vec![(2, 0, 0, 1)]), ]; - let desc = Description::new( - Antichain::from_elem(0u64), Antichain::from_elem(2u64), Antichain::from_elem(0u64)); - let batch = ChunkBatch::new(chunks, desc); + let batch = ChunkBatch::new(chunks); let mut cursor = batch.cursor(); let got = cursor.to_vec(&batch, |k| *k, |v| *v); @@ -678,12 +675,10 @@ mod test { // resumable merge -> advance -> settle pipeline end to end. #[test] fn batch_merger_resumable_matches_reference() { - use crate::trace::Description; use crate::trace::implementations::spine_fueled::Merger; use crate::trace::chunk::{ChunkBatch, ChunkBatchMerger, is_graded}; use crate::trace::cursor::Cursor; use crate::consolidation::consolidate_updates; - use timely::progress::Antichain; let mut seed = 0x9E3779B97F4A7C15u64; let mut rng = move || { seed ^= seed << 13; seed ^= seed >> 7; seed ^= seed << 17; seed }; @@ -700,9 +695,7 @@ mod test { } fn batch(updates: &[Upd], sz: usize) -> ChunkBatch> { let chunks: Vec<_> = updates.chunks(sz).map(|c| chunk(c.to_vec())).collect(); - let desc = Description::new( - Antichain::from_elem(0u64), Antichain::from_elem(10u64), Antichain::from_elem(0u64)); - ChunkBatch::new(chunks, desc) + ChunkBatch::new(chunks) } fn read(b: &ChunkBatch>) -> Vec { let mut out = Vec::new(); @@ -737,13 +730,16 @@ mod test { } let result = merger.done(); - assert!(is_graded(&result.chunks), "ungraded result: {:?}", - result.chunks.iter().map(Chunk::len).collect::>()); - let got = read(&result); + let chunks: &[ColChunk] = result.as_ref().map_or(&[], |b| &b.chunks[..]); + assert!(is_graded(chunks), "ungraded result: {:?}", + chunks.iter().map(Chunk::len).collect::>()); + let got = result.as_ref().map_or_else(Vec::new, read); let mut want: Vec<((u64, u64), u64, i64)> = u1.iter().chain(u2.iter()).map(|&(k, v, t, d)| ((k, v), t.max(f), d)).collect(); consolidate_updates(&mut want); let want: Vec = want.into_iter().map(|((k, v), t, d)| (k, v, t, d)).collect(); + // A merge that cancels to nothing reports an absent updates. + assert_eq!(result.is_none(), want.is_empty(), "absence must track emptiness\n u1={u1:?}\n u2={u2:?}\n f={f}"); assert_eq!(got, want, "fuel-driven merge mismatch\n u1={u1:?}\n u2={u2:?}\n f={f}"); } } @@ -798,7 +794,6 @@ mod test { // the right advanced-and-consolidated result. #[test] fn advance_single_key_spanning_pushes() { - use timely::progress::Antichain; let frontier = Antichain::from_elem(100u64); let n = 50u64; let mut q = VecDeque::new(); @@ -815,7 +810,6 @@ mod test { // withholding the (possibly-growing) last group as the carry when not `done`. #[test] fn advance_emits_complete_groups_eagerly() { - use timely::progress::Antichain; let frontier = Antichain::from_elem(5u64); // Group (0,0) is complete within this chunk; group (1,0) might still grow. let mut q = VecDeque::from([chunk(vec![(0, 0, 0, 1), (0, 0, 1, 1), (1, 0, 0, 1)])]); @@ -833,7 +827,6 @@ mod test { // group boundaries. #[test] fn advance_resumable_matches_oneshot() { - use timely::progress::Antichain; let frontier = Antichain::from_elem(3u64); // Groups span chunk boundaries and carry several times each. let input = || vec![ @@ -865,7 +858,6 @@ mod test { // boundaries, exercising the meld / withhold / split path. #[test] fn advance_matches_row_reference() { - use timely::progress::Antichain; use crate::consolidation::consolidate_updates; let mut seed = 0x2545F4914F6CDD1Du64; diff --git a/differential-dataflow/src/operators/arrange/agent.rs b/differential-dataflow/src/operators/arrange/agent.rs index 4da615ce9..8850feecc 100644 --- a/differential-dataflow/src/operators/arrange/agent.rs +++ b/differential-dataflow/src/operators/arrange/agent.rs @@ -10,7 +10,7 @@ use timely::progress::Timestamp; use timely::progress::{Antichain, frontier::AntichainRef}; use timely::dataflow::operators::CapabilitySet; -use crate::trace::{Trace, TraceReader, BatchReader}; +use crate::trace::{Span, Trace, TraceReader}; use timely::scheduling::Activator; @@ -62,10 +62,10 @@ impl TraceReader for TraceAgent { fn get_physical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { self.physical_compaction.borrow() } - fn batches_through(&mut self, frontier: AntichainRef<'_, Tr::Time>) -> Option> { - self.trace.borrow_mut().trace.batches_through(frontier) + fn spans_through(&mut self, frontier: AntichainRef<'_, Tr::Time>) -> Option>> { + self.trace.borrow_mut().trace.spans_through(frontier) } - fn map_batches(&self, f: F) { self.trace.borrow().trace.map_batches(f) } + fn map_spans)>(&self, f: F) { self.trace.borrow().trace.map_spans(f) } } impl TraceAgent { @@ -117,9 +117,9 @@ impl TraceAgent { self.trace .borrow_mut() .trace - .map_batches(|batch| { - new_queue.push_back(TraceReplayInstruction::Batch(batch.clone(), timely::progress::Stamp::from_elem(Tr::Time::minimum()))); - upper = Some(batch.upper().clone()); + .map_spans(|span| { + new_queue.push_back(TraceReplayInstruction::Span(span.clone(), timely::progress::Stamp::from_elem(Tr::Time::minimum()))); + upper = Some(span.upper().clone()); }); if let Some(upper) = upper { @@ -299,8 +299,8 @@ impl TraceAgent { TraceReplayInstruction::Frontier(frontier) => { capabilities.downgrade(&frontier.borrow()[..]); }, - TraceReplayInstruction::Batch(batch, hint) => { - if !hint.is_empty() && !batch.is_empty() { + TraceReplayInstruction::Span(batch, hint) => { + if !hint.is_empty() && batch.has_updates() { let delayed = capabilities.delayed_stamp(&hint); output.session(&delayed).give(batch); } @@ -438,10 +438,14 @@ impl TraceAgent { capabilities.downgrade(&frontier.borrow()[..]); } }, - TraceReplayInstruction::Batch(batch, hint) => { - if !hint.is_empty() && !batch.is_empty() { + TraceReplayInstruction::Span(batch, hint) => { + if !hint.is_empty() && batch.has_updates() { let delayed = capabilities.delayed_stamp(&hint); - output.session(&delayed).give(BatchFrontier::make_from(batch, since.borrow(), until.borrow())); + let wrapped = Span::new( + batch.desc, + batch.inner.map(|p| BatchFrontier::make_from(p, since.borrow(), until.borrow())), + ); + output.session(&delayed).give(wrapped); } } } diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index dc8e07cfd..97d81f7dd 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -31,9 +31,9 @@ use timely::progress::Stamp; use crate::{Data, VecCollection, AsCollection}; use crate::difference::Semigroup; use crate::lattice::Lattice; -use crate::trace::{self, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; +use crate::trace::{self, SpanOf, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; -use trace::wrappers::enter::{TraceEnter, BatchEnter,}; +use trace::wrappers::enter::{TraceEnter, enter_span}; use super::TraceAgent; @@ -47,7 +47,7 @@ pub struct Arranged<'scope, Tr: TraceReader> { /// This stream contains the same batches of updates the trace itself accepts, so there should /// be no additional overhead to receiving these records. The batches can be navigated just as /// the batches in the trace, by key and by value. - pub stream: Stream<'scope, Tr::Time, Vec>, + pub stream: Stream<'scope, Tr::Time, Vec>>, /// A shared trace, updated by the `Arrange` operator and readable by others. pub trace: Tr, } @@ -75,7 +75,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { TInner: Refines+Lattice, { Arranged { - stream: self.stream.enter(child).map(|bw| BatchEnter::make_from(bw)), + stream: self.stream.enter(child).map(enter_span), trace: TraceEnter::make_from(self.trace), } } @@ -98,7 +98,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { pub fn as_container(self, mut logic: L) -> crate::Collection<'scope, Tr::Time, I::Item> where I: IntoIterator, - L: FnMut(Tr::Batch) -> I+'static, + L: FnMut(SpanOf) -> I+'static, { self.stream.unary(Pipeline, "AsContainer", move |_,_| move |input, output| { input.for_each(|time, data| { @@ -115,7 +115,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { /// Flattens the stream into a `VecCollection`. /// - /// The underlying `Stream>>` is a much more efficient way to access the data, + /// The underlying `Stream>>` is a much more efficient way to access the data, /// and this method should only be used when the data need to be transformed or exchanged, rather than /// supplied as arguments to an operator using the same key-value structure. pub fn as_collection(self, mut logic: L) -> VecCollection<'scope, Tr::Time, D, BatchDiff> @@ -129,7 +129,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { /// Flattens the stream into a `VecCollection`. /// - /// The underlying `Stream>>` is a much more efficient way to access the data, + /// The underlying `Stream>>` is a much more efficient way to access the data, /// and this method should only be used when the data need to be transformed or exchanged, rather than /// supplied as arguments to an operator using the same key-value structure. /// @@ -168,7 +168,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { /// /// This method exists for streams of batches without the corresponding arrangement. /// If you have the arrangement, its `flat_map_ref` method is equivalent to this. - pub fn flat_map_batches(stream: Stream<'scope, Tr::Time, Vec>, mut logic: L) -> VecCollection<'scope, Tr::Time, I::Item, BatchDiff> + pub fn flat_map_batches(stream: Stream<'scope, Tr::Time, Vec>>, mut logic: L) -> VecCollection<'scope, Tr::Time, I::Item, BatchDiff> where Tr::Batch: Navigable, BatchCursor: Cursor