Balance spine batch events; document the stamp invariant

differential-dataflow/src/collection.rs

@@ -20,15 +20,19 @@ use crate::difference::Abelian;
 /// The containers purport to reperesent changes to a collection, and they must implement various traits
 /// in order to expose some of this functionality (e.g. negation, timestamp manipulation). Other actions
 /// on the containers, and streams of containers, are left to the container implementor to describe.
+///
+/// Unchecked assumption: each update in a container has a time greater or equal to some time in the
+/// container's stamp. In particular, a container with an empty stamp holds no updates. Differential
+/// operators rely on this to retire updates as frontiers advance, and may behave unexpectedly if it fails.
 #[derive(Clone)]
 pub struct Collection<'scope, T: Timestamp, C: 'static> {
     /// The underlying timely dataflow stream.
     ///
     /// This field is exposed to support direct timely dataflow manipulation when required, but it is
     /// not intended to be the idiomatic way to work with the collection.
     ///
-    /// The timestamp in the data is required to always be at least the timestamp _of_ the data, in
-    /// the timely-dataflow sense. If this invariant is not upheld, differential operators may behave
+    /// Each update's time is required to be greater or equal to some time in its container's stamp,
+    /// as documented on [Collection]. If this invariant is not upheld, differential operators may behave
     /// unexpectedly.
     pub inner: Stream<'scope, T, C>,
 }

differential-dataflow/src/logging.rs

@@ -22,6 +22,11 @@ where
 }
 
 /// Possible different differential events.
+///
+/// The spine reports batches through `BatchEvent`, `MergeEvent`, and `DropEvent`, which balance:
+/// a consumer that adds one batch per `BatchEvent`, and removes one per `MergeEvent` with `complete`
+/// set and per `DropEvent`, counts the batches the trace holds. Batches without updates are counted,
+/// and report a length of zero.
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
 pub enum DifferentialEvent {
     /// Batch creation.
@@ -38,12 +43,14 @@ pub enum DifferentialEvent {
     Batcher(BatcherEvent),
 }
 
-/// Either the start or end of a merge event.
+/// A batch introduced to a trace.
+///
+/// See [`DifferentialEvent`] for how batch, merge, and drop events balance.
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
 pub struct BatchEvent {
     /// Operator identifier.
     pub operator: usize,
-    /// Which order of magnitude.
+    /// Number of updates in the batch.
     pub length: usize,
 }
 
@@ -67,18 +74,22 @@ pub struct BatcherEvent {
 
 impl From<BatcherEvent> for DifferentialEvent { fn from(e: BatcherEvent) -> Self { DifferentialEvent::Batcher(e) } }
 
-/// Either the start or end of a merge event.
+/// A batch dropped by a trace.
+///
+/// See [`DifferentialEvent`] for how batch, merge, and drop events balance.
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
 pub struct DropEvent {
     /// Operator identifier.
     pub operator: usize,
-    /// Which order of magnitude.
+    /// Number of updates in the batch.
     pub length: usize,
 }
 
 impl From<DropEvent> for DifferentialEvent { fn from(e: DropEvent) -> Self { DifferentialEvent::Drop(e) } }
 
-/// Either the start or end of a merge event.
+/// Either the start or end of a merge of two batches into one.
+///
+/// See [`DifferentialEvent`] for how batch, merge, and drop events balance.
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
 pub struct MergeEvent {
     /// Operator identifier.
@@ -89,7 +100,7 @@ pub struct MergeEvent {
     pub length1: usize,
     /// Length of second trace.
     pub length2: usize,
-    /// None implies a start.
+    /// None implies a start; otherwise the number of updates in the merged batch.
     pub complete: Option<usize>,
 }
 

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

@@ -49,6 +49,9 @@
     /// 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.
+    ///
+    /// Unchecked assumption: each update in a batch has a time greater or equal to some time in the
+    /// stamp of the message carrying it. In particular, a message with an empty stamp carries no updates.
     pub stream: Stream<'scope, Tr::Time, Vec<SpanOf<Tr>>>,
     /// A shared trace, updated by the `Arrange` operator and readable by others.
     pub trace: Tr,
@@ -186,7 +189,7 @@
                     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)));
                                 });
                             }
@@ -375,10 +378,8 @@
                 batcher.insert(data);
             });
 
-            // The frontier may have advanced by multiple elements, which is an issue because
-            // timely dataflow currently only allows one capability per message. This means we
-            // must pretend to process the frontier advances one element at a time, batching
-            // and sending smaller bites than we might have otherwise done.
+            // The frontier may have advanced by multiple elements. We extract one batch for the
+            // whole advance, and send it stamped with every held capability it retires.
 
             // Assert that the frontier never regresses.
             assert!(PartialOrder::less_equal(&prev_frontier.borrow(), &frontier.frontier()));

differential-dataflow/src/operators/cursor/join.rs

@@ -141,7 +141,7 @@ where
     storage1: C1::Storage,
     cursor2: C2,
     storage2: C2::Storage,
-    /// The capability's time, at which this unit's output ships; the lower envelope for consolidation.
+    /// The meet of the stamp's times, a lower bound on this unit's output times; the lower envelope for consolidation.
     meet: T,
     /// Whether to advance each side's history by `meet` before consolidation.
     advance1: bool,

differential-dataflow/src/trace/implementations/ord_neu.rs

@@ -612,7 +612,7 @@ pub mod val_batch {
         /// Allocates a builder with capacity for the specified keys, values, and updates.
         ///
         /// They represent respectively the number of distinct `key`, `(key, val)`, and total updates.
-        fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self {
+        pub fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self {
             Self {
                 result: OrdValStorage {
                     keys: L::KeyContainer::with_capacity(keys),
@@ -1006,7 +1006,7 @@ pub mod key_batch {
         /// Allocates a builder with capacity for the specified keys and updates.
         ///
         /// They represent respectively the number of distinct `key` and total updates.
-        fn with_capacity(keys: usize, _vals: usize, upds: usize) -> Self {
+        pub fn with_capacity(keys: usize, _vals: usize, upds: usize) -> Self {
             Self {
                 result: OrdKeyStorage {
                     keys: L::KeyContainer::with_capacity(keys),

differential-dataflow/src/trace/implementations/spine_fueled.rs

@@ -371,7 +371,7 @@ impl<B: SpineBatch> Drop for Spine<B> {
 
 
 impl<B: SpineBatch> Spine<B> {
-    /// Drops and logs batches. Used in `set_logical_compaction` and drop.
+    /// Drops and logs batches. Used in drop.
     fn drop_batches(&mut self) {
         if let Some(logger) = &self.logger {
             for batch in self.merging.drain(..) {
@@ -682,6 +682,21 @@ impl<B: SpineBatch> Spine<B> {
                         complete: None,
                     }
                 ));
+                // Two present spans merge without work when at least one carries no batch. Such a
+                // merge is complete at once, and `complete_at` will not log it, so we do so here.
+                if let (Some(old), Some(batch)) = (&old, &batch) {
+                    if old.inner.is_none() || batch.inner.is_none() {
+                        self.logger.as_ref().map(|l| l.log(
+                            crate::logging::MergeEvent {
+                                operator: self.operator.global_id,
+                                scale: index,
+                                length1: span_len(old),
+                                length2: span_len(batch),
+                                complete: Some(span_len(old) + span_len(batch)),
+                            }
+                        ));
+                    }
+                }
                 let compaction_frontier = self.logical_frontier.borrow();
                 self.merging[index] = MergeState::begin_merge(old, batch, compaction_frontier);
             }

differential-dataflow/tests/logging.rs

@@ -0,0 +1,69 @@
+use std::sync::{Arc, Mutex};
+
+use timely::dataflow::operators::Probe;
+
+use differential_dataflow::input::InputSession;
+use differential_dataflow::logging::{DifferentialEvent, DifferentialEventBuilder};
+
+/// The spine's batch events balance: adding one per `BatchEvent` and removing one per completed
+/// `MergeEvent` and per `DropEvent` counts the batches the trace holds, and returns to zero once the
+/// trace is dropped. Rounds without updates insert batches with no updates, which merge without work.
+///
+/// A consumer that counts only batches with updates also balances.
+#[test]
+fn spine_batch_events_balance() {
+
+    let batches = Arc::new(Mutex::new((0i64, 0i64)));
+    let batches2 = Arc::clone(&batches);
+
+    timely::execute_directly(move |worker| {
+
+        worker.log_register().unwrap().insert::<DifferentialEventBuilder, _>("differential/arrange", move |_time, data| {
+            if let Some(data) = data {
+                let (all, nonempty) = &mut *batches2.lock().unwrap();
+                let count = |length: usize| (length > 0) as i64;
+                for (_, event) in data.iter() {
+                    match event {
+                        DifferentialEvent::Batch(e) => {
+                            *all += 1;
+                            *nonempty += count(e.length);
+                        }
+                        DifferentialEvent::Merge(e) => if let Some(complete) = e.complete {
+                            *all -= 1;
+                            *nonempty += count(complete) - count(e.length1) - count(e.length2);
+                        }
+                        DifferentialEvent::Drop(e) => {
+                            *all -= 1;
+                            *nonempty -= count(e.length);
+                        }
+                        _ => { }
+                    }
+                }
+            }
+        });
+
+        let mut input = InputSession::new();
+        // Hold the trace, as the arrangement maintains its spine only while a handle exists.
+        let (probe, trace) = worker.dataflow(|scope| {
+            let arranged = input.to_collection(scope).arrange_by_self();
+            let (probe, _) = arranged.stream.probe();
+            (probe, arranged.trace)
+        });
+
+        for round in 0u64 .. 100 {
+            // Updates in some rounds, retractions in others, and nothing in most.
+            if round % 10 == 0 { input.insert(round); }
+            if round % 10 == 5 { input.remove(round - 5); }
+            input.advance_to(round + 1);
+            input.flush();
+            worker.step_while(|| probe.less_than(input.time()));
+        }
+
+        drop(trace);
+        let dataflow = worker.installed_dataflows()[0];
+        worker.drop_dataflow(dataflow);
+        worker.step();
+    });
+
+    assert_eq!(*batches.lock().unwrap(), (0, 0));
+}