fix: frontier freezes when updates in ValDistributor consolidate away

differential-dataflow/src/columnar/collection/exchange.rs

@@ -22,6 +22,7 @@ pub struct ValDistributor<U: Update, H> {
     marker: std::marker::PhantomData<U>,
     hashfunc: H,
     pre_lens: Vec<usize>,
+    worker: usize,
 }
 
 impl<U: Update, H: for<'a> FnMut(columnar::Ref<'a, U::Key>)->u64> Distributor<RecordedUpdates<U>> for ValDistributor<U, H> {
@@ -68,6 +69,21 @@ impl<U: Update, H: for<'a> FnMut(columnar::Ref<'a, U::Key>)->u64> Distributor<Re
         // Distribute the input's record count across non-empty outputs.
         let total_records = container.records;
         let non_empty: usize = outputs.iter().filter(|o| !o.keys.values.is_empty()).count();
+        // N.B. An input whose updates all consolidated away needs to emit a message
+        // carrying a `records` count for timely's bookkeeping.
+        if non_empty == 0 && total_records > 0 {
+            let mut recorded = RecordedUpdates::<U> {
+                updates: Default::default(),
+                records: total_records,
+                consolidated: true,
+            };
+            // Push the empty update to the worker that produced the original
+            // values so the send stays local. Not needed for correctness, but
+            // a reasonable choice.
+            Message::push_at(&mut recorded, time.clone(), &mut pushers[self.worker % pushers.len()]);
+            return;
+        }
+
         let mut first_records = total_records.saturating_sub(non_empty.saturating_sub(1));
         for (pusher, output) in pushers.iter_mut().zip(outputs) {
             if !output.keys.values.is_empty() {
@@ -108,6 +124,7 @@ where
             marker: std::marker::PhantomData,
             hashfunc: self.hashfunc,
             pre_lens: Vec::new(),
+            worker: worker.index(),
         };
         (Exchange::new(senders, distributor), LogPuller::new(receiver, worker.index(), identifier, logging.clone()))
     }

differential-dataflow/tests/columnar.rs

@@ -0,0 +1,80 @@
+use std::time::{Duration, Instant};
+
+use timely::dataflow::operators::{Input, Probe};
+use timely::dataflow::{InputHandle, ProbeHandle};
+use timely::Config;
+
+use differential_dataflow::columnar::collection;
+use differential_dataflow::columnar::trace::{Batcher, Builder, Chunker, Spine};
+use differential_dataflow::operators::arrange::arrangement::arrange_core;
+
+type Upd = (u64, (), u64, i64);
+
+/// Feed the dataflow with a single (key, val) at the provided `diffs` the
+/// advance the frontier to 1.
+///
+/// Returns false if the dataflow hangs for longer than 5s.
+fn arrange_reaches_one(config: Config, diffs: &'static [i64]) -> bool {
+    let guards = timely::execute(config, move |worker| {
+        let index = worker.index();
+        let mut probe = ProbeHandle::new();
+        let mut input = <InputHandle<u64, collection::Builder<Upd>>>::new_with_builder();
+
+        worker.dataflow::<u64, _, _>(|scope| {
+            let stream = scope.input_from(&mut input);
+            let pact = collection::Pact {
+                hashfunc: |k: columnar::Ref<'_, u64>| *k,
+            };
+
+            arrange_core::<
+                _,
+                _,
+                Chunker<Upd>,
+                Batcher<u64, (), u64, i64>,
+                Builder<u64, (), u64, i64>,
+                Spine<u64, (), u64, i64>,
+            >(stream, pact, "Arrange")
+            .stream
+            .probe_with(&mut probe);
+        });
+
+        for &diff in diffs {
+            input.send((index as u64, (), 0, diff));
+        }
+        input.advance_to(1);
+        input.flush();
+
+        let start = Instant::now();
+        while probe.less_than(&1) {
+            worker.step();
+            if start.elapsed() > Duration::from_secs(5) {
+                for id in worker.installed_dataflows() {
+                    worker.drop_dataflow(id);
+                }
+                return false;
+            }
+        }
+        true
+    })
+    .expect("timely execute");
+
+    guards.join().into_iter().all(|r| r.unwrap_or(false))
+}
+
+#[test]
+fn columnar_exchange_net_empty_container() {
+    // [+1, -1] diffs should consolidate to nothing, but the exchange must
+    // still deliver a message with `records = 2`.
+    //
+    // Previously the exchange did not emit a message and the dataflow froze.
+    assert!(
+        arrange_reaches_one(Config::process(2), &[1, -1]),
+        "frontier stalled: record count lost for an all-cancelled container"
+    );
+}
+
+#[test]
+fn columnar_exchange_consolidated_container() {
+    assert!(arrange_reaches_one(Config::process(2), &[1, 1, 1]));
+    assert!(arrange_reaches_one(Config::process(3), &[2, -1, 1]));
+}