diff --git a/differential-dataflow/examples/cursors.rs b/differential-dataflow/examples/cursors.rs index 8dcfc739d..f4038b65c 100644 --- a/differential-dataflow/examples/cursors.rs +++ b/differential-dataflow/examples/cursors.rs @@ -94,7 +94,7 @@ 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 (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..5732e019b 100644 --- a/differential-dataflow/examples/multitemporal.rs +++ b/differential-dataflow/examples/multitemporal.rs @@ -100,7 +100,7 @@ fn main() { 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 (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/src/algorithms/graphs/bfs.rs b/differential-dataflow/src/algorithms/graphs/bfs.rs index 1a419b8a5..8667deeaa 100644 --- a/differential-dataflow/src/algorithms/graphs/bfs.rs +++ b/differential-dataflow/src/algorithms/graphs/bfs.rs @@ -25,7 +25,7 @@ use crate::operators::arrange::Arranged; pub fn bfs_arranged<'scope, N, Tr>(edges: Arranged<'scope, Tr>, roots: VecCollection<'scope, Tr::Time, N>) -> VecCollection<'scope, Tr::Time, (N, u32)> where N: ExchangeData+Hash, - Tr: TraceReader+Clone+'static, + Tr: TraceReader+Clone+'static, for<'a> BatchCursor: Cursor=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>, { // initialize roots as reaching themselves at distance 0 diff --git a/differential-dataflow/src/algorithms/graphs/bijkstra.rs b/differential-dataflow/src/algorithms/graphs/bijkstra.rs index 4d2034382..ffdf55ee4 100644 --- a/differential-dataflow/src/algorithms/graphs/bijkstra.rs +++ b/differential-dataflow/src/algorithms/graphs/bijkstra.rs @@ -40,7 +40,7 @@ pub fn bidijkstra_arranged<'scope, N, Tr>( ) -> VecCollection<'scope, Tr::Time, ((N,N), u32)> where N: ExchangeData+Hash, - Tr: TraceReader+Clone+'static, + Tr: TraceReader+Clone+'static, for<'a> BatchCursor: Cursor=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>, { let outer = forward.stream.scope(); diff --git a/differential-dataflow/src/algorithms/graphs/propagate.rs b/differential-dataflow/src/algorithms/graphs/propagate.rs index a691b3279..e865233ef 100644 --- a/differential-dataflow/src/algorithms/graphs/propagate.rs +++ b/differential-dataflow/src/algorithms/graphs/propagate.rs @@ -58,7 +58,7 @@ where R: Multiply, R: From, L: ExchangeData, - Tr: TraceReader+Clone+'static, + Tr: TraceReader+Clone+'static, for<'a> BatchCursor: Cursor=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=R>, F: Fn(&L)->u64+Clone+'static, { diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs index b01b92ca6..1236586a4 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -783,9 +783,9 @@ pub mod vec { /// ``` pub fn reduce_abelian(self, name: &str, mut logic: L) -> Arranged<'scope, TraceAgent> where - T2: Trace+'static, + 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 = crate::trace::BatchOf> + 'static, L: FnMut(&K, &[(&V, R)], &mut Vec<(V, BatchDiff)>)+'static, { self.reduce_core::<_,Bu,T2>(name, move |key, input, output, change| { @@ -803,9 +803,9 @@ pub mod vec { pub fn reduce_core(self, name: &str, logic: L) -> Arranged<'scope, TraceAgent> where V: Clone+'static, - T2: Trace+'static, + T2: Trace+'static, for<'a> BatchCursor: Cursor=&'a K, ValOwn = V, Time = T2::Time>, - Bu: Builder)>, Output = T2::Batch> + 'static, + Bu: Builder)>, Output = crate::trace::BatchOf> + 'static, L: FnMut(&K, &[(&V, R)], &mut Vec<(V,BatchDiff)>, &mut Vec<(V, BatchDiff)>)+'static, { self.arrange_by_key_named(&format!("Arrange: {}", name)) @@ -966,9 +966,9 @@ pub mod vec { pub fn consolidate_named(self, name: &str, reify: F) -> Self where Ba: crate::trace::Batcher, Time=T> + 'static, - Tr: crate::trace::Trace+'static, + Tr: crate::trace::Trace+'static, for<'a> BatchCursor: Cursor, - Bu: crate::trace::Builder, Output=Tr::Batch>, + Bu: crate::trace::Builder, Output=crate::trace::BatchOf>, 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 = crate::trace::BatchOf>, 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 = crate::trace::BatchOf>, Tr: crate::trace::Trace + 'static, { let exchange = timely::dataflow::channels::pact::Exchange::new(move |update: &((K,()),T,R)| (update.0).0.hashed().into()); @@ -1244,7 +1244,7 @@ pub mod vec { /// ``` pub fn join_core (self, stream2: Arranged<'scope, Tr2>, result: L) -> Collection<'scope, T,I::Item,>::Output> where - Tr2: crate::trace::TraceReader+Clone+'static, + Tr2: crate::trace::TraceReader+Clone+'static, for<'a> BatchCursor: Cursor=&'a K>, // Pin the cursor diff to a named param `R2`: a `Multiply` bound on a projection does not // connect to its use-site (the solver normalizes the use but not the bound's subject). 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..3e7ac3ee0 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(); @@ -798,7 +791,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 +807,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 +824,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 +855,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..2db1177e7 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::{Batch, Trace, TraceReader}; use timely::scheduling::Activator; @@ -38,7 +38,7 @@ pub struct TraceAgent { impl TraceReader for TraceAgent { type Time = Tr::Time; - type Batch = Tr::Batch; + type Payload = Tr::Payload; fn set_logical_compaction(&mut self, frontier: AntichainRef) { // This method does not enforce that `frontier` is greater or equal to `self.logical_compaction`. @@ -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> { + fn batches_through(&mut self, frontier: AntichainRef<'_, Tr::Time>) -> Option>> { self.trace.borrow_mut().trace.batches_through(frontier) } - fn map_batches(&self, f: F) { self.trace.borrow().trace.map_batches(f) } + fn map_batches)>(&self, f: F) { self.trace.borrow().trace.map_batches(f) } } impl TraceAgent { @@ -441,7 +441,11 @@ impl TraceAgent { TraceReplayInstruction::Batch(batch, hint) => { if !hint.is_empty() && !batch.is_empty() { let delayed = capabilities.delayed_stamp(&hint); - output.session(&delayed).give(BatchFrontier::make_from(batch, since.borrow(), until.borrow())); + let wrapped = Batch::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..0cb441c0c 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, BatchOf, Trace, TraceReader, Navigable, Batcher, Builder, Cursor, BatchCursor, BatchDiff, BatchKey, BatchVal, BatchValOwn}; -use trace::wrappers::enter::{TraceEnter, BatchEnter,}; +use trace::wrappers::enter::{TraceEnter, enter_batch}; 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(|bw| enter_batch(bw)), 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(BatchOf) -> I+'static, { self.stream.unary(Pipeline, "AsContainer", move |_,_| move |input, output| { input.for_each(|time, data| { @@ -120,7 +120,7 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { /// 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> where - Tr::Batch: Navigable, + Tr::Payload: Navigable, BatchCursor: Cursor