diff --git a/Cargo.toml b/Cargo.toml index 88102a7e3..c5d0b71c9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,9 +23,9 @@ rust-version = "1.86" [workspace.dependencies] differential-dataflow = { path = "differential-dataflow", default-features = false, version = "0.25.1" } -timely = { version = "0.31", default-features = false } +#timely = { version = "0.31", default-features = false } columnar = { version = "0.13", default-features = false } -#timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false } +timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false } #timely = { path = "../timely-dataflow/timely/", default-features = false } [workspace.lints.clippy] diff --git a/diagnostics/src/logging.rs b/diagnostics/src/logging.rs index cc4b8331a..979ddece8 100644 --- a/diagnostics/src/logging.rs +++ b/diagnostics/src/logging.rs @@ -72,7 +72,7 @@ impl ClientInput { pub fn connect(&mut self, client_id: usize, elapsed: Duration) { let _ = self .sender - .send(Event::Messages(self.time, vec![(client_id, elapsed, 1)])); + .send(Event::Messages(timely::progress::Stamp::from_elem(self.time), vec![(client_id, elapsed, 1)])); self.advance(elapsed); } @@ -80,7 +80,7 @@ impl ClientInput { pub fn disconnect(&mut self, client_id: usize, elapsed: Duration) { let _ = self .sender - .send(Event::Messages(self.time, vec![(client_id, elapsed, -1)])); + .send(Event::Messages(timely::progress::Stamp::from_elem(self.time), vec![(client_id, elapsed, -1)])); self.advance(elapsed); } diff --git a/differential-dataflow/examples/scc_bench.rs b/differential-dataflow/examples/scc_bench.rs new file mode 100644 index 000000000..b88a03395 --- /dev/null +++ b/differential-dataflow/examples/scc_bench.rs @@ -0,0 +1,109 @@ +//! Timed strongly connected components over streamed rounds of edge changes. +//! +//! Usage: scc_bench [-w] +//! +//! All rounds are introduced before stepping, so many epochs are concurrently +//! live and arrangements hold many incomparable capabilities in the nested +//! iterative scopes. + +use std::mem; +use std::hash::Hash; + +use rand::{Rng, SeedableRng, StdRng}; + +use differential_dataflow::VecCollection; +use differential_dataflow::input::Input; +use differential_dataflow::operators::*; +use differential_dataflow::lattice::Lattice; + +type Node = usize; +type Edge = (Node, Node); + +fn main() { + let nodes: usize = std::env::args().nth(1).unwrap().parse().unwrap(); + let edges: usize = std::env::args().nth(2).unwrap().parse().unwrap(); + let rounds: usize = std::env::args().nth(3).unwrap().parse().unwrap(); + + timely::execute_from_args(std::env::args(), move |worker| { + + let timer = std::time::Instant::now(); + let mut probe = timely::dataflow::ProbeHandle::new(); + + let mut edge_input = worker.dataflow(|scope| { + let (edge_input, graph) = scope.new_collection(); + _strongly_connected(graph) + .consolidate() + .probe_with(&mut probe) + ; + edge_input + }); + + let seed: &[_] = &[1, 2, 3, 4]; + let mut rng1: StdRng = SeedableRng::from_seed(seed); + let mut rng2: StdRng = SeedableRng::from_seed(seed); + + if worker.index() == 0 { + for _ in 0..edges { + edge_input.insert((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes))); + } + for round in 1..rounds { + edge_input.advance_to(round); + edge_input.insert((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes))); + edge_input.remove((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes))); + } + } + edge_input.advance_to(rounds); + edge_input.flush(); + edge_input.close(); + + while worker.step() { } + + if worker.index() == 0 { + println!("elapsed: {:?}", timer.elapsed()); + } + }).unwrap(); +} + +fn _strongly_connected<'scope, T>(graph: VecCollection<'scope, T, Edge>) -> VecCollection<'scope, T, Edge> +where + T: timely::progress::Timestamp + Lattice + Ord + Hash, +{ + graph.clone().iterate(|scope, inner| { + let edges = graph.enter(scope); + let trans = edges.clone().map_in_place(|x| mem::swap(&mut x.0, &mut x.1)); + _trim_edges(_trim_edges(inner, edges), trans) + }) +} + +fn _trim_edges<'scope, T>(cycle: VecCollection<'scope, T, Edge>, edges: VecCollection<'scope, T, Edge>) -> VecCollection<'scope, T, Edge> +where + T: timely::progress::Timestamp + Lattice + Ord + Hash, +{ + let nodes = edges.clone() + .map_in_place(|x| x.0 = x.1) + .consolidate(); + + let labels = _reachability(cycle, nodes); + + edges.consolidate() + .join_map(labels.clone(), |&e1,&e2,&l1| (e2,(e1,l1))) + .join_map(labels.clone(), |&e2,&(e1,l1),&l2| ((e1,e2),(l1,l2))) + .filter(|&(_,(l1,l2))| l1 == l2) + .map(|((x1,x2),_)| (x2,x1)) +} + +fn _reachability<'scope, T>(edges: VecCollection<'scope, T, Edge>, nodes: VecCollection<'scope, T, (Node, Node)>) -> VecCollection<'scope, T, Edge> +where + T: timely::progress::Timestamp + Lattice + Ord + Hash, +{ + edges.clone() + .filter(|_| false) + .iterate(|scope, inner| { + let edges = edges.enter(scope); + let nodes = nodes.enter_at(scope, |r| 256 * (64 - (r.0 as u64).leading_zeros() as u64)); + + inner.join_map(edges, |_k,l,d| (*d,*l)) + .concat(nodes) + .reduce(|_, s, t| t.push((*s[0].0, 1))) + }) +} diff --git a/differential-dataflow/src/columnar/collection/exchange.rs b/differential-dataflow/src/columnar/collection/exchange.rs index a6f5173a9..9fae470b6 100644 --- a/differential-dataflow/src/columnar/collection/exchange.rs +++ b/differential-dataflow/src/columnar/collection/exchange.rs @@ -30,7 +30,7 @@ impl FnMut(columnar::Ref<'a, U::Key>)->u64> Distributor>>>(&mut self, container: &mut RecordedUpdates, time: &T, pushers: &mut [P]) { + fn partition>>>(&mut self, container: &mut RecordedUpdates, stamp: &timely::progress::Stamp, pushers: &mut [P]) { use crate::columnar::updates::child_range; let view = container.updates.view(); @@ -80,7 +80,7 @@ impl FnMut(columnar::Ref<'a, U::Key>)->u64> Distributor FnMut(columnar::Ref<'a, U::Key>)->u64> Distributor>>>(&mut self, _time: &T, _pushers: &mut [P]) { } + fn flush>>>(&mut self, _stamp: &timely::progress::Stamp, _pushers: &mut [P]) { } fn relax(&mut self) { } } diff --git a/differential-dataflow/src/operators/arrange/agent.rs b/differential-dataflow/src/operators/arrange/agent.rs index a310be8bb..4da615ce9 100644 --- a/differential-dataflow/src/operators/arrange/agent.rs +++ b/differential-dataflow/src/operators/arrange/agent.rs @@ -118,7 +118,7 @@ impl TraceAgent { .borrow_mut() .trace .map_batches(|batch| { - new_queue.push_back(TraceReplayInstruction::Batch(batch.clone(), Some(Tr::Time::minimum()))); + new_queue.push_back(TraceReplayInstruction::Batch(batch.clone(), timely::progress::Stamp::from_elem(Tr::Time::minimum()))); upper = Some(batch.upper().clone()); }); @@ -300,11 +300,9 @@ impl TraceAgent { capabilities.downgrade(&frontier.borrow()[..]); }, TraceReplayInstruction::Batch(batch, hint) => { - if let Some(time) = hint { - if !batch.is_empty() { - let delayed = capabilities.delayed(&time); - output.session(&delayed).give(batch); - } + if !hint.is_empty() && !batch.is_empty() { + let delayed = capabilities.delayed_stamp(&hint); + output.session(&delayed).give(batch); } } } @@ -441,11 +439,9 @@ impl TraceAgent { } }, TraceReplayInstruction::Batch(batch, hint) => { - if let Some(time) = hint { - if !batch.is_empty() { - let delayed = capabilities.delayed(&time); - output.session(&delayed).give(BatchFrontier::make_from(batch, since.borrow(), until.borrow())); - } + 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())); } } } diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index 22d2efc99..dc8e07cfd 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -25,7 +25,8 @@ use timely::dataflow::channels::pact::{ParallelizationContract, Pipeline}; use timely::progress::Timestamp; use timely::progress::Antichain; use timely::container::{ContainerBuilder, PushInto}; -use timely::dataflow::operators::Capability; +use timely::dataflow::operators::{Capability, CapabilitySet}; +use timely::progress::Stamp; use crate::{Data, VecCollection, AsCollection}; use crate::difference::Semigroup; @@ -392,7 +393,9 @@ where // when we realize that time intervals are complete. input.for_each(|cap, data| { - capabilities.insert(cap.retain(0)); + for capability in cap.retain_stamp(0).iter() { + capabilities.insert(capability.clone()); + } chunker.push_into(data); while let Some(chunk) = chunker.extract() { batcher.push_into(std::mem::take(chunk)); @@ -422,8 +425,7 @@ where // // 1. If any held capabilities are not in advance of the new input frontier, // we must carve out updates now in advance of the new input frontier and - // transmit them as batches, which requires appropriate *single* capabilities; - // Until timely dataflow supports multiple capabilities on messages, at least. + // transmit them as a batch, stamped with the capabilities they retire. // // 2. If there are no held capabilities in advance of the new input frontier, // then there are no updates not in advance of the new input frontier and @@ -433,37 +435,30 @@ where // If there is at least one capability not in advance of the input frontier ... if capabilities.elements().iter().any(|c| !frontier.less_equal(c.time())) { - let mut upper = Antichain::new(); // re-used allocation for sealing batches. - - // For each capability not in advance of the input frontier ... - for (index, capability) in capabilities.elements().iter().enumerate() { - - if !frontier.less_equal(capability.time()) { - - // Assemble the upper bound on times we can commit with this capabilities. - // We must respect the input frontier, and *subsequent* capabilities, as - // we are pretending to retire the capability changes one by one. - upper.clear(); - for time in frontier.frontier().iter() { - upper.insert(time.clone()); - } - for other_capability in &capabilities.elements()[(index + 1) .. ] { - upper.insert(other_capability.time().clone()); - } - - // Extract updates not in advance of `upper`. - let (mut chain, description) = batcher.seal(upper.clone()); - let batch = Bu::seal(&mut chain, description); - - writer.insert(batch.clone(), Some(capability.time().clone())); - - // send the batch to downstream consumers, empty or not. - output.session(&capabilities.elements()[index]).give(batch); - } - } - - // Having extracted and sent batches between each capability and the input frontier, - // we should downgrade all capabilities to match the batcher's lower update frontier. + // The capabilities to retire: those not in advance of the input frontier. + // Each update sealed below is greater or equal to one of them, as updates + // supported only by the remaining capabilities are in advance of the input + // frontier and remain in the batcher. + let retired = capabilities + .elements() + .iter() + .filter(|c| !frontier.less_equal(c.time())) + .cloned() + .collect::>(); + + // Extract all updates not in advance of the input frontier, as one batch. + let (mut chain, description) = batcher.seal(frontier.frontier().to_owned()); + let batch = Bu::seal(&mut chain, description); + + let stamp = retired.iter().map(|c| c.time().clone()).collect::>(); + writer.insert(batch.clone(), stamp); + + // send the batch to downstream consumers, empty or not. + output.session(&retired).give(batch); + + // Having extracted and sent the batch of updates not in advance of the input + // frontier, we should downgrade all capabilities to match the batcher's lower + // update frontier. // This may involve discarding capabilities, which is fine as any new updates arrive // in messages with new capabilities. diff --git a/differential-dataflow/src/operators/arrange/mod.rs b/differential-dataflow/src/operators/arrange/mod.rs index fedbfdc23..e864cfda3 100644 --- a/differential-dataflow/src/operators/arrange/mod.rs +++ b/differential-dataflow/src/operators/arrange/mod.rs @@ -51,8 +51,10 @@ use crate::trace::TraceReader; pub enum TraceReplayInstruction { /// Describes a frontier advance. Frontier(Antichain), - /// Describes a batch of data and a capability hint. - Batch(Tr::Batch, Option), + /// Describes a batch of data and the stamp justifying its contents. + /// + /// The stamp is empty exactly when the batch is empty. + Batch(Tr::Batch, timely::progress::Stamp), } // Short names for strongly and weakly owned activators and shared queues. diff --git a/differential-dataflow/src/operators/arrange/upsert.rs b/differential-dataflow/src/operators/arrange/upsert.rs index 8d9d32a6c..bc48bd1d3 100644 --- a/differential-dataflow/src/operators/arrange/upsert.rs +++ b/differential-dataflow/src/operators/arrange/upsert.rs @@ -282,7 +282,7 @@ where prev_frontier.clone_from(&upper); // Communicate `batch` to the arrangement and the stream. - writer.insert(batch.clone(), Some(capability.time().clone())); + writer.insert(batch.clone(), timely::progress::Stamp::from_elem(capability.time().clone())); output.session(&capabilities.elements()[index]).give(batch); } } diff --git a/differential-dataflow/src/operators/arrange/writer.rs b/differential-dataflow/src/operators/arrange/writer.rs index 4d3a2d031..21761c05e 100644 --- a/differential-dataflow/src/operators/arrange/writer.rs +++ b/differential-dataflow/src/operators/arrange/writer.rs @@ -52,7 +52,7 @@ impl TraceWriter { /// The `hint` argument is either `None` in the case of an empty batch, /// or is `Some(time)` for a time less or equal to all updates in the /// batch and which is suitable for use as a capability. - pub fn insert(&mut self, batch: Tr::Batch, hint: Option) { + pub fn insert(&mut self, batch: Tr::Batch, hint: timely::progress::Stamp) { // Something is wrong if not a sequence. if !(&self.upper == batch.lower()) { @@ -84,7 +84,7 @@ impl TraceWriter { /// Inserts an empty batch up to `upper`. pub fn seal(&mut self, upper: Antichain) { if self.upper != upper { - self.insert(Tr::Batch::empty(self.upper.clone(), upper), None); + self.insert(Tr::Batch::empty(self.upper.clone(), upper), timely::progress::Stamp::new()); } } } diff --git a/differential-dataflow/src/operators/count.rs b/differential-dataflow/src/operators/count.rs index 9a9506afc..37586e0a7 100644 --- a/differential-dataflow/src/operators/count.rs +++ b/differential-dataflow/src/operators/count.rs @@ -81,10 +81,10 @@ where lower_limit.clear(); lower_limit.extend(upper_limit.borrow().iter().cloned()); - let mut cap = None; + let mut caps = timely::dataflow::operators::CapabilitySet::new(); input.for_each(|capability, batches| { - if cap.is_none() { // NB: Assumes batches are in-order - cap = Some(capability.retain(0)); + for capability in capability.retain_stamp(0).iter() { + caps.insert(capability.clone()); } for batch in batches.drain(..) { upper_limit.clone_from(batch.upper()); // NB: Assumes batches are in-order @@ -92,9 +92,9 @@ where } }); - if let Some(capability) = cap { + if !caps.is_empty() { - let mut session = output.session(&capability); + let mut session = output.session(&caps); let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage); let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap()); diff --git a/differential-dataflow/src/operators/int_proxy/reduce.rs b/differential-dataflow/src/operators/int_proxy/reduce.rs index 345a4a0ac..f988beb08 100644 --- a/differential-dataflow/src/operators/int_proxy/reduce.rs +++ b/differential-dataflow/src/operators/int_proxy/reduce.rs @@ -79,7 +79,7 @@ pub trait ProxyReduceBackend> /// It is the backend's job to prepare output batches for each of these descriptions. /// The computation proceeds in windows of keys, where only the backend maintains this /// work in progress, until `finish()` is called. - fn begin(&mut self, tiles: &[Description]); + fn begin(&mut self, description: Description); /// Present the next window of the key space, and advance `from` past it. /// @@ -123,14 +123,11 @@ pub trait ProxyReduceBackend> output: &[(u64, Self::ROut)], ) -> (Vec<(u64, Self::ROut)>, Vec); - /// Commit to a collection of updates at a specific batch in progress. - /// - /// The `tile: usize` indexes the list of descriptions provided to `begin()`, and these updates - /// are aimed at that batch in progress. - fn emit(&mut self, tile: usize, records: &[((u64, u64), B1::Time, Self::ROut)]); + /// Commit a collection of updates to the batch in progress. + fn emit(&mut self, records: &[((u64, u64), B1::Time, Self::ROut)]); - /// Complete the session matching `begin`. The outputs correspond to the descriptions it was provided. - fn finish(&mut self) -> Vec; + /// Complete the session matching `begin`, yielding the batch it described. + fn finish(&mut self) -> B2; } /// A proxy-space [`ReduceTactic`]: matches input and output records by `key_hash`. @@ -170,13 +167,13 @@ where lower: &Antichain, upper: &Antichain, held: &Antichain, - ) -> (Vec<(B1::Time, B2)>, Antichain) { + ) -> (Option, Antichain) { if held.elements().iter().all(|t| upper.less_equal(t)) { debug_assert!( self.pending.values().flatten().all(|time| held.less_equal(time)), "held capabilities do not cover pending times", ); - return (Vec::new(), held.clone()); + return (None, held.clone()); } let instance = ReduceInstance { @@ -215,14 +212,12 @@ where // strand them (see the frontier clause of the `ReduceTactic::retire` contract). if changed.is_empty() && instance.input_batches.iter().all(|b| b.is_empty()) { debug_assert_pending_frontier(&self.pending, &pending_frontier); - return (Vec::new(), pending_frontier); + return (None, pending_frontier); } - // The output tiling (identical to the Abelian tactic): one tile per held time, keeping - // non-degenerate intervals; `tile_of[i]` maps held time `i` to its tile. - let held_elems: Vec = held.elements().to_vec(); - let (tile_descs, tile_held, tile_of) = tile_descriptions(lower, upper, &held_elems); - self.backend.begin(&tile_descs); + // The single output batch spans the retired interval. + let description = Description::new(lower.clone(), upper.clone(), Antichain::from_elem(B1::Time::minimum())); + self.backend.begin(description); // Progress through the key space: `Some(h)` for key hashes at or above `h` remaining, `None` // once the backend reports the space covered. @@ -234,7 +229,7 @@ where // staging buffers are held across the whole retire rather than built where they are used. let mut slots: Vec> = Vec::new(); let mut live: Vec = Vec::new(); - let mut tile_deltas: Vec> = (0..held_elems.len()).map(|_| Vec::new()).collect(); + let mut deltas: Vec<((u64, u64), B1::Time, Bk::ROut)> = Vec::new(); let mut batch_keys: Vec = Vec::new(); let mut in_ends: Vec = Vec::new(); let mut in_all: Vec<(u64, Bk::RIn)> = Vec::new(); @@ -272,7 +267,7 @@ where "next_window must report a key hash entirely within the window that first mentions it", ); - for deltas in tile_deltas.iter_mut() { deltas.clear(); } + deltas.clear(); // The window's keys are the hashes its presentations mention: the least of the three // heads, each iteration, until all three are drained. A `changed` key that appears in @@ -361,9 +356,9 @@ where for (bi, (si, at)) in active.iter().enumerate() { let cend = corr_ends[bi]; if cstart != cend { - let idx = held_elems.iter().rposition(|h| h.less_equal(at)).expect("no held capability <= active time"); + debug_assert!(held.elements().iter().any(|h| h.less_equal(at)), "no held capability <= active time"); for (vid, d) in &corr[cstart..cend] { - tile_deltas[idx].push(((slots[*si].key, *vid), at.clone(), d.clone())); + deltas.push(((slots[*si].key, *vid), at.clone(), d.clone())); } slots[*si].sweep.commit(at, corr[cstart..cend].iter().cloned()); } @@ -385,18 +380,13 @@ where live.retain(|&si| slots[si].at.is_some()); } - for (held_index, deltas) in tile_deltas.iter_mut().enumerate() { - if deltas.is_empty() { - continue; - } - if let Some(tile) = tile_of[held_index] { - crate::consolidation::consolidate_updates(deltas); - self.backend.emit(tile, &deltas[..]); - } + if !deltas.is_empty() { + crate::consolidation::consolidate_updates(&mut deltas); + self.backend.emit(&deltas[..]); } } - let produced: Vec<(B1::Time, B2)> = tile_held.into_iter().zip(self.backend.finish()).collect(); + let produced = Some(self.backend.finish()); debug_assert_pending_frontier(&self.pending, &pending_frontier); (produced, pending_frontier) } @@ -420,37 +410,6 @@ impl Ke } } -/// Cuts the interval `[lower, upper)` into consecutive batch descriptions along `held`, which -/// must be sorted: the `i`-th cut point is the frontier formed by inserting `held[i+1..]` into -/// `upper`, so description `i` covers the part of the interval not greater-or-equal any held -/// time after `held[i]` (and not covered by an earlier description). Descriptions whose -/// interval is empty are skipped. Returns the descriptions, the held time associated with -/// each, and, per held index, the index of its description (`None` if skipped). A batch built -/// to description `i` can be committed at the capability `held[i]`. -fn tile_descriptions( - lower: &Antichain, - upper: &Antichain, - held: &[T], -) -> (Vec>, Vec, Vec>) { - let mut tile_descs: Vec> = Vec::new(); - let mut tile_held: Vec = Vec::new(); - let mut tile_of: Vec> = vec![None; held.len()]; - let mut out_lower = lower.clone(); - for index in 0..held.len() { - let mut out_upper = upper.clone(); - for t in &held[index + 1..] { - out_upper.insert(t.clone()); - } - if out_upper != out_lower { - tile_of[index] = Some(tile_descs.len()); - tile_descs.push(Description::new(out_lower.clone(), out_upper.clone(), Antichain::from_elem(T::minimum()))); - tile_held.push(held[index].clone()); - out_lower = out_upper; - } - } - (tile_descs, tile_held, tile_of) -} - /// Updates an optional meet by an optional time. fn update_meet(meet: &mut Option, other: Option<&T>) { if let Some(time) = other { diff --git a/differential-dataflow/src/operators/int_proxy/vec_backend.rs b/differential-dataflow/src/operators/int_proxy/vec_backend.rs index d5818a3f0..8a2c13361 100644 --- a/differential-dataflow/src/operators/int_proxy/vec_backend.rs +++ b/differential-dataflow/src/operators/int_proxy/vec_backend.rs @@ -63,9 +63,9 @@ pub struct VecReduceBackend { /// Lookup-only: the non-determinism of the map's iteration order is never observed. out_ids: HashMap<(K, W), u64>, - /// The retire's output tile descriptions, and the chunks accumulated for each. - tiles: Vec>, - tile_chunks: Vec>>, + /// The retire's output batch description, and the chunks accumulated for it. + description: Option>, + chunks: Vec>, /// Scratch to re-order one `emit`'s output by types, rather than transient identifiers. stage: Vec<((u64, (K, W)), T, R)>, } @@ -85,8 +85,8 @@ impl VecReduceBackend { in_pool: Vec::new(), out_pool: Vec::new(), out_ids: HashMap::new(), - tiles: Vec::new(), - tile_chunks: Vec::new(), + description: None, + chunks: Vec::new(), stage: Vec::new(), } } @@ -105,9 +105,9 @@ where type RIn = R; type ROut = R; - fn begin(&mut self, tiles: &[Description]) { - self.tiles = tiles.to_vec(); - self.tile_chunks = (0..tiles.len()).map(|_| Vec::new()).collect(); + fn begin(&mut self, description: Description) { + self.description = Some(description); + self.chunks.clear(); } #[inline(never)] @@ -304,7 +304,7 @@ where } #[inline(never)] - fn emit(&mut self, tile: usize, records: &[((u64, u64), T, R)]) { + fn emit(&mut self, records: &[((u64, u64), T, R)]) { self.stage.clear(); for ((h, vid), t, d) in records { let row = self.out_pool[*vid as usize].clone(); @@ -312,7 +312,7 @@ where } // TODO: could consolidate only within a hash key, rather than the whole chunk. consolidate_updates(&mut self.stage); - let chunks = &mut self.tile_chunks[tile]; + let chunks = &mut self.chunks; for update in self.stage.drain(..) { if chunks.last().is_none_or(|c| c.as_slice().len() >= as crate::trace::chunk::Chunk>::TARGET) { chunks.push(VecChunk::default()); @@ -322,17 +322,13 @@ where } #[inline(never)] - fn finish(&mut self) -> Vec> { + fn finish(&mut self) -> VBatch<(K, W), T, R> { self.in_pool.clear(); self.out_pool.clear(); self.out_ids.clear(); - let tiles = std::mem::take(&mut self.tiles); - let tile_chunks = std::mem::take(&mut self.tile_chunks); - tiles - .into_iter() - .zip(tile_chunks) - .map(|(desc, chunks)| Rc::new(ChunkBatch::new(chunks, desc))) - .collect() + let description = self.description.take().expect("finish without begin"); + let chunks = std::mem::take(&mut self.chunks); + Rc::new(ChunkBatch::new(chunks, description)) } } diff --git a/differential-dataflow/src/operators/join.rs b/differential-dataflow/src/operators/join.rs index 0d03d6b20..a10fba48a 100644 --- a/differential-dataflow/src/operators/join.rs +++ b/differential-dataflow/src/operators/join.rs @@ -13,7 +13,7 @@ use timely::progress::Timestamp; use timely::dataflow::Stream; use timely::dataflow::operators::generic::{Operator, OutputBuilderSession}; use timely::dataflow::channels::pact::Pipeline; -use timely::dataflow::operators::Capability; +use timely::dataflow::operators::CapabilitySet; use crate::lattice::Lattice; use crate::operators::arrange::Arranged; @@ -117,8 +117,8 @@ where // batch (so a burst on one input cannot starve the other). The driver owns the capabilities and // the fuel budget; each iterator, prepared by the tactic, yields the output containers to ship // under its paired capability, and is dropped once it goes dry. - let mut todo0: VecDeque<(Capability, Box>)> = VecDeque::new(); - let mut todo1: VecDeque<(Capability, Box>)> = VecDeque::new(); + let mut todo0: VecDeque<(CapabilitySet, Box>)> = VecDeque::new(); + let mut todo1: VecDeque<(CapabilitySet, Box>)> = VecDeque::new(); // We'll unload the initial batches here, to put ourselves in a less non-deterministic state to start. trace1.map_batches(|batch1| { @@ -168,7 +168,7 @@ where // in `batch2.upper()`. Only necessary for non-empty batches, as empty batches may not have // that property. let work = tactic.prep(trace1_storage, vec![batch2], Fresh::Input1, capability.time().clone()); - todo1.push_back((capability.clone(), work)); + todo1.push_back((CapabilitySet::from_elem(capability.clone()), work)); } // Droppable handles to shared trace data structures. @@ -194,7 +194,10 @@ where input1.for_each(|capability, data| { // This test *should* always pass, as we only drop a trace in response to the other input emptying. if let Some(ref mut trace2) = trace2_option { - let capability = capability.retain(0); + let capability = capability.retain_stamp(0); + // The lattice meet of the stamp's elements lower bounds all output + // times this batch can produce. + let meet = capability.iter().map(|c| c.time().clone()).reduce(|a, b| a.meet(&b)); for batch1 in data.drain(..) { // An arriving batch must lie wholly on one side of the preload boundary, // and wholly on one side of `acknowledged1`: both frontiers are drawn from @@ -223,7 +226,7 @@ where // It is safe to ask for `ack2` as we validated that it was at least `get_physical_compaction()` // at start-up, and have held back physical compaction ever since. let trace2_storage = trace2.batches_through(acknowledged2.borrow()).unwrap(); - let work = tactic.prep(vec![batch1.clone()], trace2_storage, Fresh::Input0, capability.time().clone()); + let work = tactic.prep(vec![batch1.clone()], trace2_storage, Fresh::Input0, meet.clone().expect("non-empty stamp")); todo0.push_back((capability.clone(), work)); } @@ -245,7 +248,10 @@ where input2.for_each(|capability, data| { // This test *should* always pass, as we only drop a trace in response to the other input emptying. if let Some(ref mut trace1) = trace1_option { - let capability = capability.retain(0); + let capability = capability.retain_stamp(0); + // The lattice meet of the stamp's elements lower bounds all output + // times this batch can produce. + let meet = capability.iter().map(|c| c.time().clone()).reduce(|a, b| a.meet(&b)); for batch2 in data.drain(..) { // An arriving batch must lie wholly on one side of the preload boundary, // and wholly on one side of `acknowledged2`: both frontiers are drawn from @@ -274,7 +280,7 @@ where // It is safe to ask for `ack1` as we validated that it was at least `get_physical_compaction()` // at start-up, and have held back physical compaction ever since. let trace1_storage = trace1.batches_through(acknowledged1.borrow()).unwrap(); - let work = tactic.prep(trace1_storage, vec![batch2.clone()], Fresh::Input1, capability.time().clone()); + let work = tactic.prep(trace1_storage, vec![batch2.clone()], Fresh::Input1, meet.clone().expect("non-empty stamp")); todo1.push_back((capability.clone(), work)); } @@ -319,7 +325,7 @@ where // pins the operator output to `NoopBuilder` — the builder for exactly this "containers ready // to go" case, which is a `ContainerBuilder` for any `C` without further bounds. let output: &mut OutputBuilderSession<'_, Tr1::Time, NoopBuilder> = output; - let mut drain = |queue: &mut VecDeque<(Capability, Box>)>, mut fuel: isize| { + let mut drain = |queue: &mut VecDeque<(CapabilitySet, Box>)>, mut fuel: isize| { while fuel >= 0 { let Some((capability, work)) = queue.front_mut() else { break }; match work.next() { diff --git a/differential-dataflow/src/operators/reduce.rs b/differential-dataflow/src/operators/reduce.rs index 83238105a..b57ee92e2 100644 --- a/differential-dataflow/src/operators/reduce.rs +++ b/differential-dataflow/src/operators/reduce.rs @@ -40,26 +40,22 @@ pub(crate) fn sort_dedup(list: &mut Vec) { /// `retire` runs the whole `[lower, upper)` interval to completion rather than yielding under a fuel /// budget. pub trait ReduceTactic> { - /// Retire the interval `[lower, upper)`, producing the output batches it informs. + /// Retire the interval `[lower, upper)`, producing the output batch it informs. /// /// It is presented with the pre-existing input batches and output batches (those before `lower`), /// the new input batches, and `held`: the times the operator currently holds capabilities for. It - /// reasons only about times, returning the output batches to ship — each tagged with the time at - /// which to ship it — and the new frontier of interesting times for the operator to hold. + /// reasons only about times, returning the output batch to ship — `None` when the interval holds + /// no work at all — and the new frontier of interesting times for the operator to hold. /// /// # Contract /// - /// The driver ([`reduce_with_tactic`]) relies on the following; the first two are cheap to check - /// and are `debug_assert!`ed there. + /// The driver ([`reduce_with_tactic`]) relies on the following; the first is cheap to check + /// and is `debug_assert!`ed there. /// - /// * **Ordered, tiling output.** The returned `(time, batch)` pairs are in ascending order and - /// their descriptions *tile* `[lower, upper)`: the first batch's lower is `lower`, each batch's - /// upper is the next batch's lower, and the last batch's upper is `upper` — no gaps, no overlaps. - /// Sub-intervals with no updates are skipped; the next batch's lower simply picks up where the - /// last left off. Producing *in order* is a requirement, not a convenience — it is what lets the - /// driver check the tiling with a single linear scan. - /// * **Shipped at a held time.** Each batch's `time` tag is an element of `held`; the driver mints - /// a capability at it, which is only valid for a held time. + /// * **Spanning output.** The returned batch's description is exactly `[lower, upper)`. The + /// driver ships it stamped with the held times not in advance of `upper`, which justify its + /// contents: every update time lies at or beyond one of them (a time at or beyond only the + /// remaining held times would be in advance of `upper`, and so outside the interval). /// * **Frontier bounds withheld work, and collapses to empty when there is none.** The returned /// frontier must be at-or-below every time the tactic defers, so the driver knows what is safe to /// release. In particular, with no work to defer it must be the *empty* antichain. Derive it from @@ -74,7 +70,7 @@ pub trait ReduceTactic> { lower: &Antichain, upper: &Antichain, held: &Antichain, - ) -> (Vec<(B1::Time, B2)>, Antichain); + ) -> (Option, Antichain); } /// A key-wise reduction of values in an input trace. @@ -172,7 +168,9 @@ where // Drain input batches in order, capturing capabilities and the last upper. input.for_each(|capability, batches| { - capabilities.insert(capability.retain(0)); + for capability in capability.retain_stamp(0).iter() { + capabilities.insert(capability.clone()); + } for batch in batches.drain(..) { upper_limit.clone_from(batch.upper()); batch_storage.push(batch); @@ -197,36 +195,28 @@ where // The times the operator currently holds capabilities for, as an antichain. let held: Antichain = capabilities.iter().map(|c| c.time().clone()).collect(); - // Retire the interval. The tactic reasons only about times: it returns output batches - // each tagged with the time to ship it at, and the new frontier of interesting times. + // Retire the interval. The tactic reasons only about times: it returns the output + // batch to ship, if any, and the new frontier of interesting times. let (produced, new_frontier) = tactic.retire(source_batches, output_batches, batch_storage, &lower_limit, &upper_limit, &held); - // Contract checks (see `ReduceTactic::retire`). Cheap, debug-only. - debug_assert!( - produced.iter().all(|(time, _)| held.elements().contains(time)), - "ReduceTactic::retire shipped a batch at a time not held as a capability", - ); + // Contract check (see `ReduceTactic::retire`). Cheap, debug-only. debug_assert!( - { - // Ordered output makes tiling a single linear scan: each description's lower - // must meet the previous upper (starting at `lower_limit`), ending at `upper_limit`. - let mut edge = lower_limit.clone(); - let abutting = produced.iter().all(|(_, batch)| { - let matches = batch.description().lower() == &edge; - edge.clone_from(batch.description().upper()); - matches - }); - abutting && (produced.is_empty() || edge == upper_limit) - }, - "ReduceTactic::retire output must be ordered and tile [lower, upper)", + produced.as_ref().is_none_or(|batch| batch.description().lower() == &lower_limit && batch.description().upper() == &upper_limit), + "ReduceTactic::retire output must span [lower, upper)", ); - // Ship each batch at a capability minted from the set at its time, and commit it to the - // output trace. The times are elements of `held`, so they stay valid until we downgrade. - for (time, batch) in produced { - let capability = capabilities.delayed(&time); - output.session(&capability).give(batch.clone()); - output_writer.insert(batch, Some(time)); + // Ship the batch stamped with the capabilities it retires — those not in advance of + // the upper limit — and commit it to the output trace. The times are elements of + // `held`, so they stay valid until we downgrade. + if let Some(batch) = produced { + let retiring = capabilities + .iter() + .filter(|c| !upper_limit.less_equal(c.time())) + .cloned() + .collect::>(); + output.session(&retiring).give(batch.clone()); + let stamp = retiring.iter().map(|c| c.time().clone()).collect::>(); + output_writer.insert(batch, stamp); } // Downgrade to the frontier the tactic handed back (a no-op when it found no work). @@ -284,8 +274,6 @@ mod cursors { interesting_times: Vec, new_interesting_times: Vec, // Output batches may need to be built piecemeal, and these temp storage help there. - output_upper: Antichain, - output_lower: Antichain, _marker: PhantomData<(B2, Bu)>, } @@ -307,8 +295,6 @@ mod cursors { next_pending_time: ::TimeContainer::with_capacity(0), interesting_times: Vec::new(), new_interesting_times: Vec::new(), - output_upper: Antichain::from_elem(::minimum()), - output_lower: Antichain::from_elem(::minimum()), _marker: PhantomData, } } @@ -332,9 +318,9 @@ mod cursors { lower: &Antichain, upper: &Antichain, held: &Antichain, - ) -> (Vec<(B1::Time, B2)>, Antichain) + ) -> (Option, Antichain) { - let mut produced = Vec::new(); + let mut produced = None; // We have compute needs only if we hold a time in the interval [lower, upper); otherwise we // could not transmit outputs even if they were (incorrectly) non-zero, and we leave the held @@ -346,15 +332,10 @@ mod cursors { let (mut output_cursor, ref output_storage) = cursor_list(output_batches); let (mut batch_cursor, ref batch_storage) = cursor_list(input_batches); - // Prepare an output buffer and builder for each held time. - // TODO: It would be better if all updates went into one batch, but timely dataflow prevents - // this as long as it requires that there is only one capability for each message. - let mut buffers = Vec::<(B1::Time, Vec<(::ValOwn, B1::Time, ::Diff)>)>::new(); - let mut builders = Vec::new(); - for time in held.elements().iter() { - buffers.push((time.clone(), Vec::new())); - builders.push(Bu::new()); - } + // Prepare one output buffer and builder: the batch spans [lower, upper) and + // ships stamped with the held times that justify its contents. + let mut output_updates = Vec::<(::ValOwn, B1::Time, ::Diff)>::new(); + let mut builder = Bu::new(); // Temporary staging for output building. let mut buffer = Bu::Input::default(); @@ -401,7 +382,8 @@ mod cursors { &self.interesting_times, &mut self.logic, upper, - &mut buffers[..], + &mut output_updates, + held.elements(), &mut self.new_interesting_times, ); @@ -420,17 +402,14 @@ mod cursors { self.next_pending_time.push_own(&time); } - // Sort each buffer by value and move into the corresponding builder. + // Sort the buffer by value and move into the builder. // TODO: This makes assumptions about at least one of (i) the stability of `sort_by`, - // (ii) that the buffers are time-ordered, and (iii) that the builders accept + // (ii) that the buffer is time-ordered, and (iii) that the builders accept // arbitrarily ordered times. - for index in 0 .. buffers.len() { - buffers[index].1.sort_by(|x,y| x.0.cmp(&y.0)); - (self.push)(&mut buffer, key, &mut buffers[index].1); - buffers[index].1.clear(); - builders[index].push(&mut buffer); - - } + output_updates.sort_by(|x,y| x.0.cmp(&y.0)); + (self.push)(&mut buffer, key, &mut output_updates); + output_updates.clear(); + builder.push(&mut buffer); } else { // copy over the pending key and times. @@ -443,38 +422,10 @@ mod cursors { // Drop to avoid lifetime issues that would lock `pending_{keys, time}`. drop(thinker); - // We start sealing output batches from the lower limit (previous upper limit). - // In principle, we could update `lower` itself, and it should arrive at `upper` by the - // end of the process. - self.output_lower.clear(); - self.output_lower.extend(lower.borrow().iter().cloned()); - - // build each batch (because only one capability per message). - for (index, builder) in builders.drain(..).enumerate() { - - // Form the upper limit of the next batch, which includes all times greater - // than the input batch, or the held times from i + 1 onward. - self.output_upper.clear(); - self.output_upper.extend(upper.borrow().iter().cloned()); - for time in &held.elements()[index + 1 ..] { - self.output_upper.insert_ref(time); - } - - if self.output_upper.borrow() != self.output_lower.borrow() { - - let description = Description::new(self.output_lower.clone(), self.output_upper.clone(), Antichain::from_elem(::minimum())); - let batch = builder.done(description); - - // hand the batch back to the driver to ship and commit, tagged with its time. - produced.push((held.elements()[index].clone(), batch)); - - self.output_lower.clear(); - self.output_lower.extend(self.output_upper.borrow().iter().cloned()); - } - } - // This should be true, as the final iteration introduces no held times, and - // uses exactly `upper` to determine the upper bound. Good to check though. - assert!(self.output_upper.borrow() == upper.borrow()); + // Build the batch spanning the interval, and hand it back to the driver + // to ship and commit. + let description = Description::new(lower.clone(), upper.clone(), Antichain::from_elem(::minimum())); + produced = Some(builder.done(description)); // Refresh pending keys and times. self.pending_keys.clear(); std::mem::swap(&mut self.next_pending_keys, &mut self.pending_keys); @@ -559,7 +510,8 @@ mod cursors { times: &Vec, logic: &mut L, upper_limit: &Antichain, - outputs: &mut [(T, Vec<(V, T, D2)>)], + outputs: &mut Vec<(V, T, D2)>, + held: &[T], new_interesting: &mut Vec) where C1: Cursor = K, Val<'a> = V1, Time = T, Diff = D1>, @@ -702,7 +654,7 @@ mod cursors { for time in self.temporary.drain(..) { // We can either service `join` now, or must delay for the future. if upper_limit.less_equal(&time) { - debug_assert!(outputs.iter().any(|(t,_)| t.less_equal(&time))); + debug_assert!(held.iter().any(|t| t.less_equal(&time))); new_interesting.push(time); } else { @@ -760,11 +712,10 @@ mod cursors { // We *should* be able to find a capability for `next_time`. Any thing else would // indicate a logical error somewhere along the way; either we release a capability // we should have kept, or we have computed the output incorrectly (or both!) - let idx = outputs.iter().rev().position(|(time, _)| time.less_equal(&next_time)); - let idx = outputs.len() - idx.expect("failed to find index") - 1; + assert!(held.iter().any(|time| time.less_equal(&next_time)), "failed to find capability"); for (val, diff) in self.update_buffer.drain(..) { self.output_produced.push(((val.clone(), next_time.clone()), diff.clone())); - outputs[idx].1.push((val, next_time.clone(), diff)); + outputs.push((val, next_time.clone(), diff)); } // Advance times in `self.output_produced` and consolidate the representation. @@ -784,7 +735,7 @@ mod cursors { // as initial interesting times are filtered to be in interval, and synthetic times are also // filtered before introducing them to `self.synth_times`. new_interesting.push(next_time.clone()); - debug_assert!(outputs.iter().any(|(t,_)| t.less_equal(&next_time))) + debug_assert!(held.iter().any(|t| t.less_equal(&next_time))) } // Update `meet` to track the meet of each source of times. @@ -880,8 +831,6 @@ pub(crate) mod reference { next_pending_time: ::TimeContainer, interesting_times: Vec, new_interesting_times: Vec, - output_upper: Antichain, - output_lower: Antichain, _marker: PhantomData<(B2, Bu)>, } @@ -903,8 +852,6 @@ pub(crate) mod reference { next_pending_time: ::TimeContainer::with_capacity(0), interesting_times: Vec::new(), new_interesting_times: Vec::new(), - output_upper: Antichain::from_elem(::minimum()), - output_lower: Antichain::from_elem(::minimum()), _marker: PhantomData, } } @@ -928,9 +875,9 @@ pub(crate) mod reference { lower: &Antichain, upper: &Antichain, held: &Antichain, - ) -> (Vec<(B1::Time, B2)>, Antichain) + ) -> (Option, Antichain) { - let mut produced = Vec::new(); + let mut produced = None; if held.elements().iter().any(|time| !upper.less_equal(time)) { @@ -938,12 +885,8 @@ pub(crate) mod reference { let (mut output_cursor, ref output_storage) = cursor_list(output_batches); let (mut batch_cursor, ref batch_storage) = cursor_list(input_batches); - let mut buffers = Vec::<(B1::Time, Vec<(::ValOwn, B1::Time, ::Diff)>)>::new(); - let mut builders = Vec::new(); - for time in held.elements().iter() { - buffers.push((time.clone(), Vec::new())); - builders.push(Bu::new()); - } + let mut output_updates = Vec::<(::ValOwn, B1::Time, ::Diff)>::new(); + let mut builder = Bu::new(); let mut buffer = Bu::Input::default(); // Reuseable state for performing the computation. @@ -981,7 +924,8 @@ pub(crate) mod reference { &self.interesting_times, &mut self.logic, upper, - &mut buffers[..], + &mut output_updates, + held.elements(), &mut self.new_interesting_times, ); @@ -997,13 +941,10 @@ pub(crate) mod reference { self.next_pending_time.push_own(&time); } - for index in 0 .. buffers.len() { - buffers[index].1.sort_by(|x,y| x.0.cmp(&y.0)); - (self.push)(&mut buffer, key, &mut buffers[index].1); - buffers[index].1.clear(); - builders[index].push(&mut buffer); - - } + output_updates.sort_by(|x,y| x.0.cmp(&y.0)); + (self.push)(&mut buffer, key, &mut output_updates); + output_updates.clear(); + builder.push(&mut buffer); } else { for pos in prior_pos .. pending_pos { @@ -1014,29 +955,8 @@ pub(crate) mod reference { } drop(thinker); - self.output_lower.clear(); - self.output_lower.extend(lower.borrow().iter().cloned()); - - for (index, builder) in builders.drain(..).enumerate() { - - self.output_upper.clear(); - self.output_upper.extend(upper.borrow().iter().cloned()); - for time in &held.elements()[index + 1 ..] { - self.output_upper.insert_ref(time); - } - - if self.output_upper.borrow() != self.output_lower.borrow() { - - let description = Description::new(self.output_lower.clone(), self.output_upper.clone(), Antichain::from_elem(::minimum())); - let batch = builder.done(description); - - produced.push((held.elements()[index].clone(), batch)); - - self.output_lower.clear(); - self.output_lower.extend(self.output_upper.borrow().iter().cloned()); - } - } - assert!(self.output_upper.borrow() == upper.borrow()); + let description = Description::new(lower.clone(), upper.clone(), Antichain::from_elem(::minimum())); + produced = Some(builder.done(description)); self.pending_keys.clear(); std::mem::swap(&mut self.next_pending_keys, &mut self.pending_keys); self.pending_time.clear(); std::mem::swap(&mut self.next_pending_time, &mut self.pending_time); @@ -1123,7 +1043,8 @@ pub(crate) mod reference { times: &Vec, logic: &mut L, upper_limit: &Antichain, - outputs: &mut [(T, Vec<(V, T, D2)>)], + outputs: &mut Vec<(V, T, D2)>, + held: &[T], new_interesting: &mut Vec) where C1: Cursor = K, Val<'a> = V1, Time = T, Diff = D1>, @@ -1304,11 +1225,10 @@ pub(crate) mod reference { crate::consolidation::consolidate(&mut self.update_buffer); if !self.update_buffer.is_empty() { - let idx = outputs.iter().rev().position(|(time, _)| time.less_equal(&next_time)); - let idx = outputs.len() - idx.expect("failed to find index") - 1; + assert!(held.iter().any(|time| time.less_equal(&next_time)), "failed to find capability"); for (val, diff) in self.update_buffer.drain(..) { self.output_produced.push(((val.clone(), next_time.clone()), diff.clone())); - outputs[idx].1.push((val, next_time.clone(), diff)); + outputs.push((val, next_time.clone(), diff)); } for entry in &mut self.output_produced { (entry.0).1.join_assign(&meet); } diff --git a/differential-dataflow/src/operators/threshold.rs b/differential-dataflow/src/operators/threshold.rs index d46906638..0ab6e35e8 100644 --- a/differential-dataflow/src/operators/threshold.rs +++ b/differential-dataflow/src/operators/threshold.rs @@ -131,10 +131,10 @@ where lower_limit.clear(); lower_limit.extend(upper_limit.borrow().iter().cloned()); - let mut cap = None; + let mut caps = timely::dataflow::operators::CapabilitySet::new(); input.for_each(|capability, batches| { - if cap.is_none() { // NB: Assumes batches are in-order - cap = Some(capability.retain(0)); + for capability in capability.retain_stamp(0).iter() { + caps.insert(capability.clone()); } for batch in batches.drain(..) { upper_limit.clone_from(batch.upper()); // NB: Assumes batches are in-order @@ -142,9 +142,9 @@ where } }); - if let Some(capability) = cap { + if !caps.is_empty() { - let mut session = output.session(&capability); + let mut session = output.session(&caps); let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage); let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap()); diff --git a/differential-dataflow/tests/int_proxy.rs b/differential-dataflow/tests/int_proxy.rs index ef8299d52..1d8947492 100644 --- a/differential-dataflow/tests/int_proxy.rs +++ b/differential-dataflow/tests/int_proxy.rs @@ -97,7 +97,7 @@ fn reduce_one_retire() { &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), ); assert!(frontier.is_empty()); - let out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + let out: Vec<_> = produced.into_iter().flat_map(|b| hread(&[b])).collect(); // Output values are `(key, max)`: (7,5) and (9,2). assert_eq!(out, vec![((7u64, 5u64), 0u64, 1i64), ((9, 2), 0, 1)]); } @@ -126,7 +126,7 @@ fn reduce_collision_correct() { vec![], vec![], vec![input], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), ); - let out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + let out: Vec<_> = produced.into_iter().flat_map(|b| hread(&[b])).collect(); assert_eq!(out, vec![((Collide(1), 9u64), 0u64, 1i64), ((Collide(2), 9), 0, 1)], "collision must not merge keys"); } @@ -269,7 +269,7 @@ fn reduce_collision_across_retires() { vec![], vec![], vec![b0.clone()], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), ); - let out0: Vec<_> = p0.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + let out0: Vec<_> = p0.into_iter().flat_map(|b| hread(&[b])).collect(); assert_eq!(out0, vec![((Collide(1), 5u64), 0u64, 1i64), ((Collide(2), 7), 0, 1)], "retire 1"); // Retire 2: a novel update to the LOWER key only, so the id order is [C1(hist), C2(hist), C1(novel)]. @@ -277,14 +277,14 @@ fn reduce_collision_across_retires() { let mut t2 = ProxyReduceTactic::new(VecReduceBackend::new(logic)); let b = hbatch::(vec![((Collide(1), 5), 0, 1), ((Collide(2), 7), 0, 1)], 0, 1); let (p, _) = t2.retire(vec![], vec![], vec![b], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64)); - p.into_iter().map(|(_t, b)| b).collect() + p.into_iter().collect() }; let b1 = hbatch::(vec![((Collide(1), 9), 1, 1)], 1, 2); let (p1, _f) = tactic.retire( vec![b0], out_batches, vec![b1], &Antichain::from_elem(1u64), &Antichain::from_elem(2u64), &Antichain::from_elem(1u64), ); - let out1: Vec<_> = p1.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + let out1: Vec<_> = p1.into_iter().flat_map(|b| hread(&[b])).collect(); // C1's max rises 5 -> 9; C2 is untouched and must NOT be disturbed. assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 9), 1, 1)], "retire 2 must not disturb C2"); } @@ -360,7 +360,7 @@ fn reduce_collision_multiwindow() { vec![], vec![], vec![input], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), ); - let mut out: Vec<_> = produced.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + let mut out: Vec<_> = produced.into_iter().flat_map(|b| hread(&[b])).collect(); out.sort(); let want: Vec<_> = (0..5u64).map(|k| ((Bucket(k), 10 * k + 1), 0u64, 1i64)).collect(); assert_eq!(out, want, "each real key keeps its own maximum"); @@ -388,7 +388,7 @@ fn reduce_collision_fastpath_endpoints() { vec![], vec![], vec![b0.clone()], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), ); - let outs: Vec<_> = p0.into_iter().map(|(_t, b)| b).collect(); + let outs: Vec<_> = p0.into_iter().collect(); assert_eq!(outs.iter().flat_map(|b| hread(std::slice::from_ref(b))).collect::>(), vec![((Collide(1), 5u64), 0u64, 1i64)], "retire 1: only C1 emits"); @@ -397,7 +397,7 @@ fn reduce_collision_fastpath_endpoints() { vec![b0], outs, vec![b1], &Antichain::from_elem(1u64), &Antichain::from_elem(2u64), &Antichain::from_elem(1u64), ); - let out1: Vec<_> = p1.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + let out1: Vec<_> = p1.into_iter().flat_map(|b| hread(&[b])).collect(); // C1's values are {5, 6}: the max becomes 6. C2's 900 belongs to a different real key. assert_eq!(out1, vec![((Collide(1), 5u64), 1u64, -1i64), ((Collide(1), 6), 1, 1)], "C2's value must not enter C1's reduction"); @@ -433,7 +433,7 @@ fn reduce_seed_survives_cancellation() { vec![], vec![], vec![b0.clone()], &Antichain::from_elem(0u64), &Antichain::from_elem(1u64), &Antichain::from_elem(0u64), ); - let outs: Vec<_> = p0.into_iter().map(|(_t, b)| b).collect(); + let outs: Vec<_> = p0.into_iter().collect(); assert_eq!( outs.iter().flat_map(|b| hread(std::slice::from_ref(b))).collect::>(), vec![((7u64, 3u64), 0u64, 1i64)], @@ -448,6 +448,6 @@ fn reduce_seed_survives_cancellation() { vec![b0], outs, vec![b1], &Antichain::from_elem(1u64), &Antichain::from_elem(2u64), &Antichain::from_elem(1u64), ); - let out1: Vec<_> = p1.into_iter().flat_map(|(_t, b)| hread(&[b])).collect(); + let out1: Vec<_> = p1.into_iter().flat_map(|b| hread(&[b])).collect(); assert_eq!(out1, vec![((7u64, 3u64), 1u64, -1i64)], "the stale output must be retracted"); } diff --git a/differential-dataflow/tests/stamps.rs b/differential-dataflow/tests/stamps.rs new file mode 100644 index 000000000..b066ec856 --- /dev/null +++ b/differential-dataflow/tests/stamps.rs @@ -0,0 +1,221 @@ +//! Tests for arrangement batches stamped by multiple capabilities. + +use std::cell::RefCell; +use std::rc::Rc; + +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::Probe; +use timely::dataflow::operators::generic::Operator; +use timely::dataflow::operators::generic::builder_rc::OperatorBuilder; +use timely::dataflow::operators::vec::unordered_input::UnorderedInput; +use timely::dataflow::{ProbeHandle, Stream}; +use timely::progress::Stamp; + +use differential_dataflow::AsCollection; +use differential_dataflow::trace::BatchReader; + +use pair::Pair; + +type Time = Pair; + +/// Attaches an observer to a stream of batches, recording for each message its +/// stamp and the record counts of the batches it contains. +fn observe_frames<'scope, B: BatchReader + Clone + 'static>( + stream: &Stream<'scope, Time, Vec>, + name: &str, + seen: Rc, Vec)>>>, +) { + let mut builder = OperatorBuilder::new(name.to_owned(), stream.scope()); + let mut input = builder.new_input(stream.clone(), Pipeline); + builder.build(move |_init_caps| { + move |_frontiers| { + input.for_each(|cap, data| { + let lens = data.iter().map(|batch| batch.len()).collect::>(); + seen.borrow_mut().push((cap.stamp().clone(), lens)); + }); + } + }); +} + +/// An arrangement that retires two incomparable capabilities at once ships a +/// single batch stamped by both times, and a concurrently imported copy of the +/// trace replays it identically. +#[test] +fn fused_arrange_stamps() { + let (arranged_seen, imported_seen, joined_seen, reduced_seen) = timely::execute_directly(move |worker| { + let arranged_seen = Rc::new(RefCell::new(Vec::new())); + let imported_seen = Rc::new(RefCell::new(Vec::new())); + let arranged_rec = Rc::clone(&arranged_seen); + let imported_rec = Rc::clone(&imported_seen); + + let mut probe = ProbeHandle::new(); + + let joined_seen = Rc::new(RefCell::new(Vec::new())); + let reduced_seen = Rc::new(RefCell::new(Vec::new())); + let joined_rec = Rc::clone(&joined_seen); + let reduced_rec = Rc::clone(&reduced_seen); + + // Arrange some data, observing the stamps on the batch stream, and + // driving `join` and `reduce` from the fused (multi-stamp) batches. + let (mut input, capability, mut trace) = worker.dataflow::(|scope| { + let ((input, capability), data) = scope.new_unordered_input(); + let collection = data.as_collection(); + let arranged = collection.clone().arrange_by_key(); + observe_frames(&arranged.stream, "observe-arranged", arranged_rec); + arranged.stream.clone().probe_with(&mut probe); + + arranged + .clone() + .join_core(arranged.clone(), |key: &u64, val1: &u64, val2: &u64| Some((*key, *val1, *val2))) + .inner + .unary::>, _, _, _>(Pipeline, "record-joined", move |_, _| move |input, output| { + input.for_each(|cap, data| { + joined_rec.borrow_mut().extend(data.drain(..)); + output.session(&cap).give(0u8); + }); + }) + .probe_with(&mut probe); + + collection + .reduce(|_key, input, output| output.push((input.len(), 1isize))) + .inner + .unary::>, _, _, _>(Pipeline, "record-reduced", move |_, _| move |input, output| { + input.for_each(|cap, data| { + reduced_rec.borrow_mut().extend(data.drain(..)); + output.session(&cap).give(0u8); + }); + }) + .probe_with(&mut probe); + + (input, capability, arranged.trace) + }); + + // Import the trace concurrently, observing the stamps on the replayed stream. + worker.dataflow::(|scope| { + let imported = trace.import(scope); + observe_frames(&imported.stream, "observe-imported", imported_rec); + imported.stream.probe_with(&mut probe); + }); + drop(trace); + + // Two incomparable open times. + let time1 = Pair::new(0, 1); + let time2 = Pair::new(1, 0); + let cap1 = capability.delayed(&time1); + let cap2 = capability.delayed(&time2); + drop(capability); + + input.activate().session(&cap1).give(((10u64, 100u64), time1.clone(), 1isize)); + input.activate().session(&cap2).give(((20u64, 200u64), time2.clone(), 1isize)); + + // Ensure the arrangement has ingested both updates before it can seal, + // so that both capabilities retire in the same frontier advance. + for _ in 0..10 { worker.step(); } + drop(cap1); + drop(cap2); + while worker.step() { } + + ( + Rc::try_unwrap(arranged_seen).unwrap().into_inner(), + Rc::try_unwrap(imported_seen).unwrap().into_inner(), + Rc::try_unwrap(joined_seen).unwrap().into_inner(), + Rc::try_unwrap(reduced_seen).unwrap().into_inner(), + ) + }); + + let expected_frame: Stamp