diff --git a/differential-dataflow/src/collection.rs b/differential-dataflow/src/collection.rs index 356f1346f..3500aa9cb 100644 --- a/differential-dataflow/src/collection.rs +++ b/differential-dataflow/src/collection.rs @@ -1252,6 +1252,42 @@ pub mod vec { } } +/// Routes the records of one message to singleton capabilities when the message's stamp has +/// several elements. +/// +/// A batch is shipped under the set of capabilities it retires, so a message of batches can +/// carry several timestamps. Records each have one time, and an operator over records reads a +/// message's time as one (`InputCapability::time`); an operator that turns batches into records +/// keeps that true by sending each record under the first element of the stamp at or before its +/// time. The elements cover every record: a batch's times are at or beyond one of the +/// capabilities it retired under. +pub struct StampRouter { + caps: Vec>, +} + +impl StampRouter { + /// One capability per element of the message's stamp, for output `port`. + pub fn new(cap: &timely::dataflow::operators::InputCapability, port: usize) -> Self { + Self { caps: cap.stamp().iter().map(|t| cap.delayed(t, port)).collect() } + } + /// One capability per element of a set. + pub fn from_set(set: &timely::dataflow::operators::CapabilitySet) -> Self { + Self { caps: set.iter().cloned().collect() } + } + /// The capabilities, in the order `index` names them. + pub fn capabilities(&self) -> &[timely::dataflow::operators::Capability] { + &self.caps + } + /// Which capability a record at `time` goes under. + pub fn index(&self, time: &T) -> usize { + self.caps.iter().position(|c| c.time().less_equal(time)).expect("a record's time is at or beyond an element of its message's stamp") + } + /// Empty buffers, one per capability, to route records into and then give under each. + pub fn buffers(&self) -> Vec> { + (0..self.caps.len()).map(|_| Vec::new()).collect() + } +} + /// Conversion to a differential dataflow Collection. pub trait AsCollection<'scope, T: Timestamp, C> { /// Converts the type to a differential dataflow collection. diff --git a/differential-dataflow/src/columnar/collection/operators.rs b/differential-dataflow/src/columnar/collection/operators.rs index ff7b3f21a..57880ed8a 100644 --- a/differential-dataflow/src/columnar/collection/operators.rs +++ b/differential-dataflow/src/columnar/collection/operators.rs @@ -116,12 +116,19 @@ where move |_frontier| { let mut output = output.activate(); op_input.for_each(|cap, data| { - // Truncate the capability's timestamp. - let mut new_time = cap.time().clone(); - let mut vec = std::mem::take(&mut new_time.inner).into_inner(); - vec.truncate(level - 1); - new_time.inner = PointStamp::new(vec); - let new_cap = cap.delayed(&new_time, 0); + // A message may carry several timestamps (a multi-element stamp): hold a + // capability for each, truncated exactly as the updates are. + let new_cap: timely::dataflow::operators::CapabilitySet<_> = cap + .stamp() + .iter() + .map(|t| { + let mut new_time = t.clone(); + let mut vec = std::mem::take(&mut new_time.inner).into_inner(); + vec.truncate(level - 1); + new_time.inner = PointStamp::new(vec); + cap.delayed(&new_time, 0) + }) + .collect(); // Push updates with truncated times into the builder. // The builder's form call on flush sorts and consolidates, // handling the duplicate times that truncation can produce. @@ -167,20 +174,30 @@ where .unary::, _, _, _>(Pipeline, "AsRecordedUpdates", |_, _| { move |input, output| { 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) { - let key = cursor.key(&batch); - let val = cursor.val(&batch); - cursor.map_times(&batch, |time, diff| { - session.give((key, val, time, diff)); - }); - cursor.step_val(&batch); + // A message of batches may carry several timestamps; each record goes out + // under the one at or before its time (a pass per element, the rare case), + // so the collection's messages carry one each. + let router = crate::collection::StampRouter::new(&time, 0); + let batches: Vec<_> = batches.drain(..).filter_map(|b| b.inner).collect(); + let mut owned_time = U::Time::default(); + for (index, cap) in router.capabilities().iter().enumerate() { + let mut session = output.session_with_builder(cap); + for batch in batches.iter() { + let mut cursor = batch.cursor(); + while cursor.key_valid(batch) { + while cursor.val_valid(batch) { + let key = cursor.key(batch); + let val = cursor.val(batch); + cursor.map_times(batch, |time, diff| { + columnar::Columnar::copy_from(&mut owned_time, time); + if router.index(&owned_time) == index { + session.give((key, val, time, diff)); + } + }); + cursor.step_val(batch); + } + cursor.step_key(batch); } - cursor.step_key(&batch); } } }); diff --git a/differential-dataflow/src/dynamic/mod.rs b/differential-dataflow/src/dynamic/mod.rs index 0a615ffbf..1020b33af 100644 --- a/differential-dataflow/src/dynamic/mod.rs +++ b/differential-dataflow/src/dynamic/mod.rs @@ -16,6 +16,7 @@ pub mod pointstamp; use timely::order::Product; use timely::progress::Timestamp; use timely::dataflow::operators::generic::{OutputBuilder, builder_rc::OperatorBuilder}; +use timely::dataflow::operators::CapabilitySet; use timely::dataflow::channels::pact::Pipeline; use timely::progress::Antichain; @@ -47,17 +48,21 @@ where builder.build(move |_capability| move |_frontier| { let mut output = output.activate(); input.for_each(|cap, data| { - let mut new_time = cap.time().clone(); - let mut vec = std::mem::take(&mut new_time.inner).into_inner(); - vec.truncate(level - 1); - new_time.inner = PointStamp::new(vec); - let new_cap = cap.delayed(&new_time, 0); - for (_data, time, _diff) in data.iter_mut() { - let mut vec = std::mem::take(&mut time.inner).into_inner(); + // A message may carry several timestamps (a multi-element stamp, e.g. late + // iterations of one epoch alongside early ones of the next): hold a capability + // for each, truncated exactly as the records are. + let truncate = |time: &Product>| { + let mut new_time = time.clone(); + let mut vec = std::mem::take(&mut new_time.inner).into_inner(); vec.truncate(level - 1); - time.inner = PointStamp::new(vec); + new_time.inner = PointStamp::new(vec); + new_time + }; + let caps: CapabilitySet<_> = cap.stamp().iter().map(|t| cap.delayed(&truncate(t), 0)).collect(); + for (_data, time, _diff) in data.iter_mut() { + *time = truncate(time); } - output.session(&new_cap).give_container(data); + output.session(&caps).give_container(data); }); }); diff --git a/differential-dataflow/src/operators/arrange/arrangement.rs b/differential-dataflow/src/operators/arrange/arrangement.rs index 75eba8177..a7d0d817e 100644 --- a/differential-dataflow/src/operators/arrange/arrangement.rs +++ b/differential-dataflow/src/operators/arrange/arrangement.rs @@ -179,15 +179,20 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { { stream.unary(Pipeline, "AsCollection", move |_,_| move |input, output| { input.for_each(|time, data| { - let mut session = output.session(&time); + // A message of batches may carry several timestamps; each record goes out under + // the one at or before its time, so the collection's messages carry one each. + let router = crate::collection::StampRouter::new(&time, 0); + let mut buffers = router.buffers(); for wrapper in data.iter() { let Some(batch) = wrapper.inner.as_ref() else { continue }; let mut cursor = batch.cursor(); while let Some(key) = cursor.get_key(batch) { while let Some(val) = cursor.get_val(batch) { for datum in logic(key, val) { - cursor.map_times(batch, |time, diff| { - session.give((datum.clone(), as Cursor>::owned_time(time), as Cursor>::owned_diff(diff))); + cursor.map_times(batch, |t, diff| { + let t = as Cursor>::owned_time(t); + let index = router.index(&t); + buffers[index].push((datum.clone(), t, as Cursor>::owned_diff(diff))); }); } cursor.step_val(batch); @@ -195,6 +200,11 @@ impl<'scope, Tr: TraceReader> Arranged<'scope, Tr> { cursor.step_key(batch); } } + for (cap, mut buffer) in router.capabilities().iter().zip(buffers) { + if !buffer.is_empty() { + output.session(cap).give_container(&mut buffer); + } + } }); }) .as_collection() diff --git a/differential-dataflow/src/operators/count.rs b/differential-dataflow/src/operators/count.rs index 1f707c3b5..a3c1d1727 100644 --- a/differential-dataflow/src/operators/count.rs +++ b/differential-dataflow/src/operators/count.rs @@ -94,7 +94,14 @@ where if !caps.is_empty() { - let mut session = output.session(&caps); + // The batches' capabilities may span several timestamps; each record goes out + // under the one at or before its time, so the collection's messages carry one. + let router = crate::collection::StampRouter::from_set(&caps); + let mut buffers = router.buffers(); + let mut give = |record: (_, Tr::Time, _)| { + let index = router.index(&record.1); + buffers[index].push(record); + }; let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage.into_iter().filter_map(|b| b.inner).collect()); let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap()); @@ -114,20 +121,25 @@ where if let Some(count) = count.as_ref() { if !count.is_zero() { - session.give(((key.clone(), count.clone()), as Cursor>::owned_time(time), R2::from(-1i8))); + give(((key.clone(), count.clone()), as Cursor>::owned_time(time), R2::from(-1i8))); } } count.as_mut().map(|c| c.plus_equals(&diff)); if count.is_none() { count = Some( as Cursor>::owned_diff(diff)); } if let Some(count) = count.as_ref() { if !count.is_zero() { - session.give(((key.clone(), count.clone()), as Cursor>::owned_time(time), R2::from(1i8))); + give(((key.clone(), count.clone()), as Cursor>::owned_time(time), R2::from(1i8))); } } }); batch_cursor.step_key(&batch_storage); } + for (cap, mut buffer) in router.capabilities().iter().zip(buffers) { + if !buffer.is_empty() { + output.session(cap).give_container(&mut buffer); + } + } } // tidy up the shared input trace. diff --git a/differential-dataflow/src/operators/threshold.rs b/differential-dataflow/src/operators/threshold.rs index c9502976f..abba9fe60 100644 --- a/differential-dataflow/src/operators/threshold.rs +++ b/differential-dataflow/src/operators/threshold.rs @@ -144,7 +144,14 @@ where if !caps.is_empty() { - let mut session = output.session(&caps); + // The batches' capabilities may span several timestamps; each record goes out + // under the one at or before its time, so the collection's messages carry one. + let router = crate::collection::StampRouter::from_set(&caps); + let mut buffers = router.buffers(); + let mut give = |record: (_, Tr::Time, _)| { + let index = router.index(&record.1); + buffers[index].push(record); + }; let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage.into_iter().filter_map(|b| b.inner).collect()); let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap()); @@ -185,13 +192,18 @@ where if let Some(difference) = difference { if !difference.is_zero() { - session.give((key.clone(), as Cursor>::owned_time(time), difference)); + give((key.clone(), as Cursor>::owned_time(time), difference)); } } }); batch_cursor.step_key(&batch_storage); } + for (cap, mut buffer) in router.capabilities().iter().zip(buffers) { + if !buffer.is_empty() { + output.session(cap).give_container(&mut buffer); + } + } } // tidy up the shared input trace. diff --git a/differential-dataflow/tests/dynamic.rs b/differential-dataflow/tests/dynamic.rs new file mode 100644 index 000000000..05e896bb7 --- /dev/null +++ b/differential-dataflow/tests/dynamic.rs @@ -0,0 +1,158 @@ +//! Leaving a dynamic scope on a message that carries two timestamps. +//! +//! Since timely's stamps became multisets, a message can carry several timestamps, and DD's +//! batch-shipping operators make such messages: `arrange` ships a batch under the set of +//! capabilities it retires, `reduce` likewise, and `join` forwards its input batch's set. In an +//! iterative scope with two epochs in flight, an arrange holding `(1, [18])` (epoch 1, round 18) +//! and `(2, [])` (epoch 2, just entered) retires both in one batch when its input frontier passes +//! both at once. `as_collection` then forwards the batch's records under that same set, and the +//! feedback delays it element-wise, so the set reaches whatever reads the *message's* time +//! rather than the records' — which `leave_dynamic` did, to truncate it, and so panicked on +//! "expected a singleton stamp". (Observed on a DDIR program at 20k nodes with four workers; +//! the first multi-element stamp came from an arrange, then a join, then a reduce, then the +//! arrange whose batch reached the scope's exit.) +//! +//! Which retirements coincide depends on scheduling, so this test makes the message directly: an +//! operator that ships its input under a capability for epoch `e` round 3 and one for epoch +//! `e + 1` round 0, into `leave_dynamic`. + +use differential_dataflow::collection::AsCollection; +use differential_dataflow::dynamic::pointstamp::PointStamp; +use differential_dataflow::input::Input; +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::generic::{builder_rc::OperatorBuilder, OutputBuilder}; +use timely::dataflow::operators::CapabilitySet; +use timely::order::Product; + +type Time = Product>; + +#[test] +fn leave_dynamic_on_a_message_over_two_epochs() { + let received = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let seen = std::sync::Arc::clone(&received); + timely::execute_directly(move |worker| { + let mut probe = timely::dataflow::ProbeHandle::new(); + let mut input = worker.dataflow::(|scope| { + let (input, data) = scope.new_collection::(); + let left = scope.iterative::, _, _>(|inner| { + let entered = data.enter(inner); + let mut builder = OperatorBuilder::new("TwoEpochs".to_string(), inner); + let (output, stream) = builder.new_output(); + let mut output = OutputBuilder::from(output); + let mut input = builder.new_input(entered.inner, Pipeline); + builder.build(move |mut caps| { + // the initial capability, at (0, []), can be delayed to any later time; it is + // dropped once the input is done, so the computation can finish + let mut root = caps.pop(); + move |frontier| { + let mut output = output.activate(); + input.for_each(|cap, data| { + let Some(root) = root.as_ref() else { return }; + let epoch = cap.time().outer; + let t1 = Product::new(epoch, PointStamp::new([3].into_iter().collect())); + let t2 = Product::new(epoch + 1, PointStamp::new([0].into_iter().collect())); + let caps: CapabilitySet