From 61b1308dd9bb34504b64481a2c8a2952e9ed59cf Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 21 Aug 2026 17:17:11 -0400 Subject: [PATCH] Batches as data: Batch { desc, inner: Option } An exploratory replacement of the BatchReader and Batch traits with a concrete struct: a batch is a Description plus an optional payload, absent exactly when there are no updates. Payloads are unconstrained; reading them is the Navigable capability, and everything else is the business of whichever harness holds them. Consequences, in rough order of significance: * Empty batches are constructible by anyone (a description with no payload): Batch::empty is a constructor, not a capability, and the minting question (TraceWriter::seal, Spine::close) dissolves. * Rc sharing is a payload type parameter, not a forwarding-impl family: rc_blanket_impls shrinks to the Navigable/cursor forwarding, and RcBuilder disappears (builders emit Rc payloads directly). External Arc newtypes would need no batch impls at all. * Description algebra hoists out of the payload mergers into the spine: MergeState computes each merge result's description (adjacent lower/ upper, sinces joined with the compaction frontier), and mergers become pure payload mergers handed a since. A merged payload that cancels entirely becomes an absent payload. * len() leaves the public batch surface: its only consumers were the spine's accounting and logging, so it lives on SpinePayload (the renamed SpineBatch), alongside the Merger opinion. * TraceReader grows a Payload associated type; Tr::Batch becomes the BatchOf alias. The join/reduce/half-join tactics take payloads (empty batches have nothing to join) with an explicit time parameter, and the drivers keep descriptions for their frontier accounting. The two kinds of nothing remain deliberately distinct in the spine: MergeState::Single(None) is structural fuel bookkeeping with no description at all; Batch { inner: None } is a recorded empty interval. Co-Authored-By: Claude Fable 5 --- differential-dataflow/examples/cursors.rs | 2 +- .../examples/multitemporal.rs | 2 +- .../src/algorithms/graphs/bfs.rs | 2 +- .../src/algorithms/graphs/bijkstra.rs | 2 +- .../src/algorithms/graphs/propagate.rs | 2 +- differential-dataflow/src/collection.rs | 18 +- .../src/columnar/collection/operators.rs | 1 + .../src/columnar/trace/chunk.rs | 17 +- .../src/operators/arrange/agent.rs | 14 +- .../src/operators/arrange/arrangement.rs | 42 ++-- .../src/operators/arrange/mod.rs | 2 +- .../src/operators/arrange/upsert.rs | 6 +- .../src/operators/arrange/writer.rs | 8 +- differential-dataflow/src/operators/count.rs | 8 +- .../src/operators/int_proxy/join.rs | 60 +++-- .../src/operators/int_proxy/reduce.rs | 56 ++--- .../src/operators/int_proxy/vec_backend.rs | 16 +- differential-dataflow/src/operators/join.rs | 67 ++--- differential-dataflow/src/operators/reduce.rs | 133 +++++----- .../src/operators/threshold.rs | 8 +- differential-dataflow/src/trace/chunk/mod.rs | 71 +++--- differential-dataflow/src/trace/chunk/vec.rs | 26 +- differential-dataflow/src/trace/cursor/mod.rs | 18 +- .../src/trace/cursor/wrappers/enter.rs | 7 +- .../src/trace/cursor/wrappers/frontier.rs | 11 +- .../src/trace/implementations/ord_neu.rs | 154 ++++-------- .../src/trace/implementations/spine_fueled.rs | 238 ++++++++++-------- differential-dataflow/src/trace/mod.rs | 138 +++++----- .../src/trace/wrappers/enter.rs | 64 +++-- .../src/trace/wrappers/frontier.rs | 50 ++-- differential-dataflow/tests/int_proxy.rs | 27 +- differential-dataflow/tests/trace.rs | 8 +- dogsdogsdogs/src/operators/count.rs | 2 +- dogsdogsdogs/src/operators/half_join.rs | 31 +-- dogsdogsdogs/src/operators/propose.rs | 2 +- dogsdogsdogs/src/operators/validate.rs | 2 +- interactive/src/backend/corgi.rs | 3 +- interactive/src/corgi/chunk.rs | 15 +- interactive/src/corgi/join.rs | 19 +- interactive/src/corgi/reduce.rs | 19 +- 40 files changed, 628 insertions(+), 743 deletions(-) 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