Exploration: batches as data — Batch<T, B> { desc, inner: Option<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();

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;

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<Batch: Navigable>+Clone+'static,
+    Tr: TraceReader<Payload: Navigable>+Clone+'static,
     for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>,
 {
     // initialize roots as reaching themselves at distance 0

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<Batch: Navigable>+Clone+'static,
+    Tr: TraceReader<Payload: Navigable>+Clone+'static,
     for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=isize>,
 {
     let outer = forward.stream.scope();

differential-dataflow/src/algorithms/graphs/propagate.rs

@@ -58,7 +58,7 @@ where
     R: Multiply<R, Output=R>,
     R: From<i8>,
     L: ExchangeData,
-    Tr: TraceReader<Batch: Navigable, Time: Hash>+Clone+'static,
+    Tr: TraceReader<Payload: Navigable, Time: Hash>+Clone+'static,
     for<'a> BatchCursor<Tr>: Cursor<Key<'a>=&'a N, Val<'a>=&'a N, Time=Tr::Time, Diff=R>,
     F: Fn(&L)->u64+Clone+'static,
 {

differential-dataflow/src/collection.rs

@@ -783,9 +783,9 @@ pub mod vec {
         /// ```
         pub fn reduce_abelian<L, Bu, T2>(self, name: &str, mut logic: L) -> Arranged<'scope, TraceAgent<T2>>
         where
-            T2: Trace<Batch: Navigable, Time=T>+'static,
+            T2: Trace<Payload: Navigable, Time=T>+'static,
             for<'a> BatchCursor<T2>: Cursor<Key<'a>= &'a K, ValOwn = V, Time = T2::Time, Diff: Abelian>,
-            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, BatchDiff<T2>)>, Output = T2::Batch> + 'static,
+            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, BatchDiff<T2>)>, Output = crate::trace::BatchOf<T2>> + 'static,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(V, BatchDiff<T2>)>)+'static,
         {
             self.reduce_core::<_,Bu,T2>(name, move |key, input, output, change| {
@@ -803,9 +803,9 @@ pub mod vec {
         pub fn reduce_core<L, Bu, T2>(self, name: &str, logic: L) -> Arranged<'scope, TraceAgent<T2>>
         where
             V: Clone+'static,
-            T2: Trace<Batch: Navigable, Time=T>+'static,
+            T2: Trace<Payload: Navigable, Time=T>+'static,
             for<'a> BatchCursor<T2>: Cursor<Key<'a>=&'a K, ValOwn = V, Time = T2::Time>,
-            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, BatchDiff<T2>)>, Output = T2::Batch> + 'static,
+            Bu: Builder<Time=T2::Time, Input = Vec<((K, V), T2::Time, BatchDiff<T2>)>, Output = crate::trace::BatchOf<T2>> + 'static,
             L: FnMut(&K, &[(&V, R)], &mut Vec<(V,BatchDiff<T2>)>, &mut Vec<(V, BatchDiff<T2>)>)+'static,
         {
             self.arrange_by_key_named(&format!("Arrange: {}", name))
@@ -966,9 +966,9 @@ pub mod vec {
         pub fn consolidate_named<Ba, Bu, Tr, F>(self, name: &str, reify: F) -> Self
         where
             Ba: crate::trace::Batcher<Output=Vec<((D, ()), T, R)>, Time=T> + 'static,
-            Tr: crate::trace::Trace<Batch: Navigable, Time=T>+'static,
+            Tr: crate::trace::Trace<Payload: Navigable, Time=T>+'static,
             for<'a> BatchCursor<Tr>: Cursor<Time=Tr::Time, Diff=R>,
-            Bu: crate::trace::Builder<Time=Tr::Time, Input=Vec<((D, ()), T, R)>, Output=Tr::Batch>,
+            Bu: crate::trace::Builder<Time=Tr::Time, Input=Vec<((D, ()), T, R)>, Output=crate::trace::BatchOf<Tr>>,
             F: Fn(BatchKey<'_, Tr>, BatchVal<'_, Tr>) -> D + 'static,
         {
             use crate::operators::arrange::arrangement::Arrange;
@@ -1036,7 +1036,7 @@ pub mod vec {
         fn arrange_named<Ba, Bu, Tr>(self, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
         where
             Ba: crate::trace::Batcher<Output=Vec<((K, V), T, R)>, Time=T> + 'static,
-            Bu: crate::trace::Builder<Time=T, Input=Vec<((K, V), T, R)>, Output = Tr::Batch>,
+            Bu: crate::trace::Builder<Time=T, Input=Vec<((K, V), T, R)>, Output = crate::trace::BatchOf<Tr>>,
             Tr: crate::trace::Trace<Time=T> + '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<Ba, Bu, Tr>(self, name: &str) -> Arranged<'scope, TraceAgent<Tr>>
         where
             Ba: crate::trace::Batcher<Output=Vec<((K, ()), T, R)>, Time=T> + 'static,
-            Bu: crate::trace::Builder<Time=T, Input=Vec<((K, ()), T, R)>, Output = Tr::Batch>,
+            Bu: crate::trace::Builder<Time=T, Input=Vec<((K, ()), T, R)>, Output = crate::trace::BatchOf<Tr>>,
             Tr: crate::trace::Trace<Time=T> + '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<Tr2,I,L,R2> (self, stream2: Arranged<'scope, Tr2>, result: L) -> Collection<'scope, T,I::Item,<R as Multiply<R2>>::Output>
         where
-            Tr2: crate::trace::TraceReader<Batch: Navigable, Time=T>+Clone+'static,
+            Tr2: crate::trace::TraceReader<Payload: Navigable, Time=T>+Clone+'static,
             for<'a> BatchCursor<Tr2>: Cursor<Key<'a>=&'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).

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) {

differential-dataflow/src/columnar/trace/chunk.rs

@@ -546,6 +546,7 @@ fn advance_trie<U: ColumnarUpdate>(
 
 #[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<ColChunk<Upd>> {
             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<ColChunk<Upd>>) -> Vec<Upd> {
             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;

differential-dataflow/src/operators/arrange/agent.rs

@@ -10,7 +10,7 @@
 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 @@
 impl<Tr: TraceReader> TraceReader for TraceAgent<Tr> {
 
     type Time = Tr::Time;
-    type Batch = Tr::Batch;
+    type Payload = Tr::Payload;
 
     fn set_logical_compaction(&mut self, frontier: AntichainRef<Tr::Time>) {
         // This method does not enforce that `frontier` is greater or equal to `self.logical_compaction`.
@@ -62,10 +62,10 @@
     fn get_physical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> {
         self.physical_compaction.borrow()
     }
-    fn batches_through(&mut self, frontier: AntichainRef<'_, Tr::Time>) -> Option<Vec<Self::Batch>> {
+    fn batches_through(&mut self, frontier: AntichainRef<'_, Tr::Time>) -> Option<Vec<Batch<Tr::Time, Tr::Payload>>> {
         self.trace.borrow_mut().trace.batches_through(frontier)
     }
-    fn map_batches<F: FnMut(&Self::Batch)>(&self, f: F) { self.trace.borrow().trace.map_batches(f) }
+    fn map_batches<F: FnMut(&Batch<Tr::Time, Tr::Payload>)>(&self, f: F) { self.trace.borrow().trace.map_batches(f) }
 }
 
 impl<Tr: TraceReader> TraceAgent<Tr> {
@@ -283,7 +283,7 @@
                 let activator = scope.activator_for(Rc::clone(&info.address));
                 let queue = self.new_listener(activator);
 
                 let activator = scope.activator_for(info.address);
                 *shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));
 
                 capabilities.borrow_mut().as_mut().unwrap().insert(capability);
@@ -414,7 +414,7 @@
                 let activator = scope.activator_for(Rc::clone(&info.address));
                 let queue = self.new_listener(activator);
 
                 let activator = scope.activator_for(info.address);
                 *shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));
 
                 capabilities.borrow_mut().as_mut().unwrap().insert(capability);
@@ -441,7 +441,11 @@
                                     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);
                                         }
                                     }
                                 }