Stamps at the batch-to-records seams, leave_dynamic under multi-element stamps, and unconsolidated feedback

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<T: Timestamp> {
+    caps: Vec<timely::dataflow::operators::Capability<T>>,
+}
+
+impl<T: Timestamp> StampRouter<T> {
+    /// One capability per element of the message's stamp, for output `port`.
+    pub fn new(cap: &timely::dataflow::operators::InputCapability<T>, 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<T>) -> Self {
+        Self { caps: set.iter().cloned().collect() }
+    }
+    /// The capabilities, in the order `index` names them.
+    pub fn capabilities(&self) -> &[timely::dataflow::operators::Capability<T>] {
+        &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<D>(&self) -> Vec<Vec<D>> {
+        (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.

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::<Builder<U>, _, _, _>(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);
                         }
                     }
                 });

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<TOuter, PointStamp<T>>| {
+                    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);
             });
         });
 

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

@@ -179,22 +179,32 @@ 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(), <BatchCursor<Tr> as Cursor>::owned_time(time), <BatchCursor<Tr> as Cursor>::owned_diff(diff)));
+                                cursor.map_times(batch, |t, diff| {
+                                    let t = <BatchCursor<Tr> as Cursor>::owned_time(t);
+                                    let index = router.index(&t);
+                                    buffers[index].push((datum.clone(), t, <BatchCursor<Tr> as Cursor>::owned_diff(diff)));
                                 });
                             }
                             cursor.step_val(batch);
                         }
                         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()

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()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(-1i8)));
+                                    give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(-1i8)));
                                 }
                             }
                             count.as_mut().map(|c| c.plus_equals(&diff));
                             if count.is_none() { count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff)); }
                             if let Some(count) = count.as_ref() {
                                 if !count.is_zero() {
-                                    session.give(((key.clone(), count.clone()), <BatchCursor<Tr> as Cursor>::owned_time(time), R2::from(1i8)));
+                                    give(((key.clone(), count.clone()), <BatchCursor<Tr> 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.

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(), <BatchCursor<Tr> as Cursor>::owned_time(time), difference));
+                                    give((key.clone(), <BatchCursor<Tr> 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.