From 865bb7343241c9905b19331a5317c30aadbe6d27 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 7 Aug 2026 11:02:41 -0400
Subject: [PATCH 1/5] Avoid non-termination that results from in-band progress
 tracking

---
 dogsdogsdogs/examples/dogsdogsdogs.rs | 14 +++----
 dogsdogsdogs/src/lib.rs               | 60 ++++++++++-----------------
 2 files changed, 30 insertions(+), 44 deletions(-)

diff --git a/dogsdogsdogs/examples/dogsdogsdogs.rs b/dogsdogsdogs/examples/dogsdogsdogs.rs
index 110ec3922..9e3a542eb 100644
--- a/dogsdogsdogs/examples/dogsdogsdogs.rs
+++ b/dogsdogsdogs/examples/dogsdogsdogs.rs
@@ -31,17 +31,17 @@ fn main() {
 
         println!("loaded {} nodes, {} edges", nodes, edges.len());
 
-        let index = worker.dataflow::<usize,_,_>(|scope| {
-            CollectionIndex::index(Collection::new(edges.to_stream(scope)))
-        });
-
-        let mut index_xz = index.extend_using(|&(ref x, ref _y)| *x);
-        let mut index_yz = index.extend_using(|&(ref _x, ref y)| *y);
-
         let mut probe = Handle::new();
 
+        // The index and its readers must share a dataflow: the extenders hold scope-bound
+        // arrangements rather than exported traces.
         let mut edges = worker.dataflow::<usize,_,_>(|scope| {
 
+            let index = CollectionIndex::index(Collection::new(edges.to_stream(scope)));
+
+            let mut index_xz = index.extend_using(|&(ref x, ref _y)| *x);
+            let mut index_yz = index.extend_using(|&(ref _x, ref y)| *y);
+
             let (edges_input, edges) = scope.new_collection();
 
             // determine stream of (prefix, count, index) indicating relation with fewest extensions.
diff --git a/dogsdogsdogs/src/lib.rs b/dogsdogsdogs/src/lib.rs
index ce484e3b7..b370d0d6d 100644
--- a/dogsdogsdogs/src/lib.rs
+++ b/dogsdogsdogs/src/lib.rs
@@ -91,11 +91,18 @@ impl<'scope, T: Timestamp, R: Monoid+Multiply<Output = R>, P, E> ValidateExtensi
 }
 
 // These are all defined here so that users can be assured a common layout.
+use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::trace::implementations::{KeySpine, ValSpine};
 type TraceValHandle<K,V,T,R> = TraceAgent<ValSpine<K,V,T,R>>;
 type TraceKeyHandle<K,T,R> = TraceAgent<KeySpine<K,T,R>>;
 
-pub struct CollectionIndex<K, V, T, R>
+/// The three arrangements a relation must present to extend prefixes.
+///
+/// The arrangements are scope-bound rather than exported traces, so that the operators
+/// reading them observe timely's own progress tracking. An imported trace instead reports
+/// its frontier in-band, and in a cycle those statements circulate without ever settling.
+#[derive(Clone)]
+pub struct CollectionIndex<'scope, K, V, T, R>
 where
     K: ExchangeData,
     V: ExchangeData,
@@ -103,40 +110,23 @@ where
     R: Monoid+Multiply<Output = R>+ExchangeData,
 {
     /// A trace of type (K, ()), used to count extensions for each prefix.
-    count_trace: TraceKeyHandle<K, T, isize>,
+    count_trace: Arranged<'scope, TraceKeyHandle<K, T, isize>>,
 
     /// A trace of type (K, V), used to propose extensions for each prefix.
-    propose_trace: TraceValHandle<K, V, T, R>,
+    propose_trace: Arranged<'scope, TraceValHandle<K, V, T, R>>,
 
     /// A trace of type ((K, V), ()), used to validate proposed extensions.
-    validate_trace: TraceKeyHandle<(K, V), T, R>,
+    validate_trace: Arranged<'scope, TraceKeyHandle<(K, V), T, R>>,
 }
 
-impl<K, V, T, R> Clone for CollectionIndex<K, V, T, R>
+impl<'scope, K, V, T, R> CollectionIndex<'scope, K, V, T, R>
 where
     K: ExchangeData+Hash,
     V: ExchangeData+Hash,
     T: Lattice+ExchangeData+Timestamp,
     R: Monoid+Multiply<Output = R>+ExchangeData,
 {
-    fn clone(&self) -> Self {
-        CollectionIndex {
-            count_trace: self.count_trace.clone(),
-            propose_trace: self.propose_trace.clone(),
-            validate_trace: self.validate_trace.clone(),
-        }
-    }
-}
-
-impl<K, V, T, R> CollectionIndex<K, V, T, R>
-where
-    K: ExchangeData+Hash,
-    V: ExchangeData+Hash,
-    T: Lattice+ExchangeData+Timestamp,
-    R: Monoid+Multiply<Output = R>+ExchangeData,
-{
-
-    pub fn index<'scope>(collection: VecCollection<'scope, T, (K, V), R>) -> Self {
+    pub fn index(collection: VecCollection<'scope, T, (K, V), R>) -> Self {
         // We need to count the number of (k, v) pairs and not rely on the given Monoid R and its binary addition operation.
         // counts and validate can share the base arrangement
         let arranged = collection.clone().arrange_by_self();
@@ -146,10 +136,9 @@ where
             .as_collection(|k,_v| k.clone())
             .distinct()
             .map(|(k, _v)| k)
-            .arrange_by_self()
-            .trace;
-        let propose = collection.arrange_by_key().trace;
-        let validate = arranged.trace;
+            .arrange_by_self();
+        let propose = collection.arrange_by_key();
+        let validate = arranged;
 
         CollectionIndex {
             count_trace: counts,
@@ -157,7 +146,7 @@ where
             validate_trace: validate,
         }
     }
-    pub fn extend_using<P, F: Fn(&P)->K+Clone>(&self, logic: F) -> CollectionExtender<K, V, T, R, P, F> {
+    pub fn extend_using<P, F: Fn(&P)->K+Clone>(&self, logic: F) -> CollectionExtender<'scope, K, V, T, R, P, F> {
         CollectionExtender {
             phantom: std::marker::PhantomData,
             indices: self.clone(),
@@ -166,7 +155,7 @@ where
     }
 }
 
-pub struct CollectionExtender<K, V, T, R, P, F>
+pub struct CollectionExtender<'scope, K, V, T, R, P, F>
 where
     K: ExchangeData,
     V: ExchangeData,
@@ -175,11 +164,11 @@ where
     F: Fn(&P)->K+Clone,
 {
     phantom: std::marker::PhantomData<P>,
-    indices: CollectionIndex<K, V, T, R>,
+    indices: CollectionIndex<'scope, K, V, T, R>,
     key_selector: F,
 }
 
-impl<'scope, T, K, V, R, P, F> PrefixExtender<'scope, T, R> for CollectionExtender<K, V, T, R, P, F>
+impl<'scope, T, K, V, R, P, F> PrefixExtender<'scope, T, R> for CollectionExtender<'scope, K, V, T, R, P, F>
 where
     T: Timestamp + Lattice + ExchangeData + Hash,
     K: ExchangeData+Hash+Default,
@@ -192,17 +181,14 @@ where
     type Extension = V;
 
     fn count(&mut self, prefixes: VecCollection<'scope, T, (P, usize, usize), R>, index: usize) -> VecCollection<'scope, T, (P, usize, usize), R> {
-        let counts = self.indices.count_trace.import(prefixes.scope());
-        operators::count::count(prefixes, counts, self.key_selector.clone(), index)
+        operators::count::count(prefixes, self.indices.count_trace.clone(), self.key_selector.clone(), index)
     }
 
     fn propose(&mut self, prefixes: VecCollection<'scope, T, P, R>) -> VecCollection<'scope, T, (P, V), R> {
-        let propose = self.indices.propose_trace.import(prefixes.scope());
-        operators::propose::propose(prefixes, propose, self.key_selector.clone())
+        operators::propose::propose(prefixes, self.indices.propose_trace.clone(), self.key_selector.clone())
     }
 
     fn validate(&mut self, extensions: VecCollection<'scope, T, (P, V), R>) -> VecCollection<'scope, T, (P, V), R> {
-        let validate = self.indices.validate_trace.import(extensions.scope());
-        operators::validate::validate(extensions, validate, self.key_selector.clone())
+        operators::validate::validate(extensions, self.indices.validate_trace.clone(), self.key_selector.clone())
     }
 }

From 148223607b2b257d2ac781ed5af401316702ff70 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 7 Aug 2026 15:17:10 -0400
Subject: [PATCH 2/5] Retarget count/propose/validate onto half_join

The three shared `lookup_map`, which matched arranged updates at times less-or-equal in the *partial* order, and emitted at the update's own time.
That is a correct lookup but not a correct delta join.
Pairs at incomparable times never met, and a chained stage compared against the previous stage's time rather than the original.

They now sit on half_join, which compares in a caller-supplied total order and carries the original time beside a payload time advanced by lattice join.
`propose` and `validate` are wrappers over the safe entry point.
`count` uses the unsafe one, to sum a scalar and annotate the prefix without joining times.
`lookup_map` and `propose_distinct` are deleted; the latter proposed presence without maintaining an indicator to test it against.

Differences are pinned to `isize` and the `Multiply` bounds dropped.
BiGJoin (Ammar et al., VLDB 2018) is stated over sets: its extension indices are set-valued and its intersection step is an existence test.
Above arity two an extension index is a projection, whose multiplicity counts completions rather than the record's own annotation, so carrying annotations would need indicator projections and a contributes-once rule.
That is InsideOut, a different algorithm with different indices.
The set expectation is documented rather than enforced, so callers whose data are already distinct pay nothing.

`count` routes a prefix with a zero count rather than dropping it, because the sum spans an interval and zero there does not mean empty at every time within it.
It sums absolute values, since magnitude carries the extension count and polarity only churn.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 dogsdogsdogs/examples/delta_query.rs        |  38 ++--
 dogsdogsdogs/examples/delta_query_wcoj.rs   |  41 +++--
 dogsdogsdogs/examples/dogsdogsdogs.rs       |  22 ++-
 dogsdogsdogs/src/lib.rs                     | 183 ++++++++++++++-----
 dogsdogsdogs/src/operators/count.rs         |  90 ++++++---
 dogsdogsdogs/src/operators/lookup_map.rs    | 145 ---------------
 dogsdogsdogs/src/operators/mod.rs           |   4 +-
 dogsdogsdogs/src/operators/propose.rs       |  92 ++++------
 dogsdogsdogs/src/operators/validate.rs      |  62 ++++---
 dogsdogsdogs/tests/lookup_map_regression.rs | 108 -----------
 dogsdogsdogs/tests/wcoj_triangle.rs         | 193 ++++++++++++++++++++
 11 files changed, 538 insertions(+), 440 deletions(-)
 delete mode 100644 dogsdogsdogs/src/operators/lookup_map.rs
 delete mode 100644 dogsdogsdogs/tests/lookup_map_regression.rs
 create mode 100644 dogsdogsdogs/tests/wcoj_triangle.rs

diff --git a/dogsdogsdogs/examples/delta_query.rs b/dogsdogsdogs/examples/delta_query.rs
index bf5388fce..d35b18713 100644
--- a/dogsdogsdogs/examples/delta_query.rs
+++ b/dogsdogsdogs/examples/delta_query.rs
@@ -1,4 +1,6 @@
 use timely::dataflow::operators::probe::Handle;
+use timely::dataflow::operators::vec::Map;
+use differential_dataflow::AsCollection;
 use differential_dataflow::input::Input;
 use graph_map::GraphMMap;
 
@@ -73,23 +75,37 @@ fn main() {
                 use differential_dogs3::operators::propose;
                 use differential_dogs3::operators::validate;
 
+                // Hold compaction back one base-time step, so the alt/neu distinction survives,
+                // and compare in the total order on `AltNeu` (lexicographic on `(time, neu)`).
+                let frontier_func = |time: &AltNeu<usize>, antichain: &mut timely::progress::Antichain<AltNeu<usize>>| {
+                    antichain.insert(AltNeu::alt(time.time.saturating_sub(1)));
+                };
+                let comparison = |t1: &AltNeu<usize>, t2: &AltNeu<usize>| t1 <= t2;
+
+                // Stash each delta's own time as its payload, to be advanced by the times of
+                // the records it matches, and delayed to once we leave the delta region.
+                let deltas = changes.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
+
                 // Prior technology
                 //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
-                let changes1 = propose(changes.clone(), forward_key_neu.clone(), key2.clone());
-                let changes1 = validate(changes1, forward_self_neu.clone(), key1.clone());
-                let changes1 = changes1.map(|((a,b),c)| (a,b,c));
+                let changes1 = propose(deltas.clone(), forward_key_neu.clone(), key2.clone(), frontier_func, comparison);
+                let changes1 = validate(changes1, forward_self_neu.clone(), key1.clone(), frontier_func, comparison);
+                let changes1 = changes1.map(|(((a,b),c), payload)| ((a,b,c), payload));
 
                 //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
-                let changes2 = propose(changes.clone(), reverse_key_alt.clone(), key1.clone());
-                let changes2 = validate(changes2, reverse_self_neu.clone(), key2.clone());
-                let changes2 = changes2.map(|((b,c),a)| (a,b,c));
+                let changes2 = propose(deltas.clone(), reverse_key_alt.clone(), key1.clone(), frontier_func, comparison);
+                let changes2 = validate(changes2, reverse_self_neu.clone(), key2.clone(), frontier_func, comparison);
+                let changes2 = changes2.map(|(((b,c),a), payload)| ((a,b,c), payload));
 
                 //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                let changes3 = propose(changes, forward_key_alt.clone(), key1.clone());
-                let changes3 = validate(changes3, reverse_self_alt.clone(), key2.clone());
-                let changes3 = changes3.map(|((a,c),b)| (a,b,c));
-
-                let prev_changes = changes1.concat(changes2).concat(changes3).leave(scope);
+                let changes3 = propose(deltas, forward_key_alt.clone(), key1.clone(), frontier_func, comparison);
+                let changes3 = validate(changes3, reverse_self_alt.clone(), key2.clone(), frontier_func, comparison);
+                let changes3 = changes3.map(|(((a,c),b), payload)| ((a,b,c), payload));
+
+                // Delay updates to the payload time worked out while extending.
+                let prev_changes = changes1.concat(changes2).concat(changes3)
+                    .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
+                    .leave(scope);
 
                 // New ideas
                 let d_edges = edges.differentiate(inner);
diff --git a/dogsdogsdogs/examples/delta_query_wcoj.rs b/dogsdogsdogs/examples/delta_query_wcoj.rs
index cba34a71f..6539f8ce6 100644
--- a/dogsdogsdogs/examples/delta_query_wcoj.rs
+++ b/dogsdogsdogs/examples/delta_query_wcoj.rs
@@ -1,4 +1,6 @@
 use timely::dataflow::operators::probe::Handle;
+use timely::dataflow::operators::vec::Map;
+use differential_dataflow::AsCollection;
 use differential_dataflow::input::Input;
 use graph_map::GraphMMap;
 
@@ -36,11 +38,25 @@ fn main() {
                 let forward = forward.enter(inner);
                 let reverse = reverse.enter(inner);
 
+                // Hold compaction back one base-time step, so that the alt/neu distinction
+                // survives, and compare in the total order on `AltNeu` (lexicographic on
+                // `(time, neu)`). Delta streams enter at `alt`, so an `alt` arrangement is
+                // seen at the delta's own time and a `neu` one only strictly after it: the
+                // tag is what makes the comparison strict or not.
+                let frontier_func = |time: &AltNeu<usize>, antichain: &mut timely::progress::Antichain<AltNeu<usize>>| {
+                    antichain.insert(AltNeu::alt(time.time.saturating_sub(1)));
+                };
+                let comparison = |t1: &AltNeu<usize>, t2: &AltNeu<usize>| t1 <= t2;
+
                 // Without using wrappers yet, maintain an "old" and a "new" copy of edges.
-                let alt_forward = CollectionIndex::index(forward.clone());
-                let alt_reverse = CollectionIndex::index(reverse.clone());
-                let neu_forward = CollectionIndex::index(forward.clone().delay(|time| AltNeu::neu(time.time.clone())));
-                let neu_reverse = CollectionIndex::index(reverse.clone().delay(|time| AltNeu::neu(time.time.clone())));
+                let alt_forward = CollectionIndex::index(forward.clone(), frontier_func, comparison);
+                let alt_reverse = CollectionIndex::index(reverse.clone(), frontier_func, comparison);
+                let neu_forward = CollectionIndex::index(forward.clone().delay(|time| AltNeu::neu(time.time.clone())), frontier_func, comparison);
+                let neu_reverse = CollectionIndex::index(reverse.clone().delay(|time| AltNeu::neu(time.time.clone())), frontier_func, comparison);
+
+                // Stash each delta's own time as its payload, to be advanced by the times of
+                // the records it matches, and delayed to once we leave the delta region.
+                let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
 
                 // For each relation, we form a delta query driven by changes to that relation.
                 //
@@ -53,33 +69,36 @@ fn main() {
 
                 //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
                 let changes1 =
-                forward
+                deltas
                     .clone()
                     .extend(&mut [
                         &mut neu_forward.extend_using(|(_a,b)| *b),
                         &mut neu_forward.extend_using(|(a,_b)| *a),
                     ])
-                    .map(|((a,b),c)| (a,b,c));
+                    .map(|(((a,b),c), payload)| ((a,b,c), payload));
 
                 //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
                 let changes2 =
-                forward
+                deltas
                     .clone()
                     .extend(&mut [
                         &mut alt_reverse.extend_using(|(b,_c)| *b),
                         &mut neu_reverse.extend_using(|(_b,c)| *c),
                     ])
-                    .map(|((b,c),a)| (a,b,c));
+                    .map(|(((b,c),a), payload)| ((a,b,c), payload));
 
                 //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                let changes3 = forward
+                let changes3 = deltas
                     .extend(&mut [
                         &mut alt_forward.extend_using(|(a,_c)| *a),
                         &mut alt_reverse.extend_using(|(_a,c)| *c),
                     ])
-                    .map(|((a,c),b)| (a,b,c));
+                    .map(|(((a,c),b), payload)| ((a,b,c), payload));
 
-                changes1.concat(changes2).concat(changes3).leave(scope)
+                // Delay updates to the payload time worked out while extending.
+                changes1.concat(changes2).concat(changes3)
+                    .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
+                    .leave(scope)
             });
 
             triangles
diff --git a/dogsdogsdogs/examples/dogsdogsdogs.rs b/dogsdogsdogs/examples/dogsdogsdogs.rs
index 9e3a542eb..58822152b 100644
--- a/dogsdogsdogs/examples/dogsdogsdogs.rs
+++ b/dogsdogsdogs/examples/dogsdogsdogs.rs
@@ -1,4 +1,4 @@
-use timely::dataflow::operators::{ToStream, vec::{Partition, count::Accumulate}, Inspect, Probe};
+use timely::dataflow::operators::{ToStream, vec::{Map, Partition, count::Accumulate}, Inspect, Probe};
 use timely::dataflow::operators::probe::Handle;
 use differential_dataflow::{Collection, AsCollection};
 use differential_dataflow::input::Input;
@@ -37,20 +37,31 @@ fn main() {
         // arrangements rather than exported traces.
         let mut edges = worker.dataflow::<usize,_,_>(|scope| {
 
-            let index = CollectionIndex::index(Collection::new(edges.to_stream(scope)));
+            // Hold compaction back one step, and let a prefix see arranged updates at its own
+            // time and earlier. The index is static here, so every prefix sees all of it.
+            let frontier_func = |time: &usize, antichain: &mut timely::progress::Antichain<usize>| {
+                antichain.insert(time.saturating_sub(1));
+            };
+            let comparison = |t1: &usize, t2: &usize| t1 <= t2;
+
+            let index = CollectionIndex::index(Collection::new(edges.to_stream(scope)), frontier_func, comparison);
 
             let mut index_xz = index.extend_using(|&(ref x, ref _y)| *x);
             let mut index_yz = index.extend_using(|&(ref _x, ref y)| *y);
 
             let (edges_input, edges) = scope.new_collection();
 
+            // Stash each prefix's own time as its payload, to be advanced by the times of the
+            // records it matches, and delayed to once we are done extending.
+            let prefixes = edges.inner.map(|(p, t, r): ((u32, u32), usize, isize)| ((p, t.clone()), t, r)).as_collection();
+
             // determine stream of (prefix, count, index) indicating relation with fewest extensions.
-            let counts  = edges.map(|p| (p, usize::MAX, usize::MAX));
+            let counts  = prefixes.map(|(p, payload)| ((p, usize::MAX, usize::MAX), payload));
             let counts0 = index_xz.count(counts,  0);
             let counts1 = index_yz.count(counts0, 1);
 
             // partition by index.
-            let parts = counts1.inner.partition(2, |((p, _c, i),t,d)| (i as u64,(p,t,d)));
+            let parts = counts1.inner.partition(2, |(((p, _c, i), payload),t,d)| (i as u64,((p, payload),t,d)));
 
             // propose extensions using relation based on index.
             let propose0 = index_xz.propose(parts[0].clone().as_collection());
@@ -62,7 +73,8 @@ fn main() {
 
             validate0
                 .concat(validate1)
-                .inner
+                // Delay updates to the payload time worked out while extending.
+                .inner.map(|((extended, payload), _time, r)| (extended, payload, r))
                 .count()
                 .inspect(move |x| println!("{:?}", x))
                 // .inspect(move |x| println!("{:?}:\t{:?}", timer.elapsed(), x))
diff --git a/dogsdogsdogs/src/lib.rs b/dogsdogsdogs/src/lib.rs
index b370d0d6d..afedd39cd 100644
--- a/dogsdogsdogs/src/lib.rs
+++ b/dogsdogsdogs/src/lib.rs
@@ -1,11 +1,41 @@
+//! Worst-case optimal joins as differential dataflows.
+//!
+//! This crate implements the BiGJoin / Delta-GJ algorithms of Ammar, McSherry, Salihoglu and
+//! Joglekar, "Distributed Evaluation of Subgraph Queries Using Worst-case Optimal and Low-Memory
+//! Dataflows" (VLDB 2018). Prefixes are extended one attribute at a time: each relation binding
+//! the next attribute reports how many extensions it would propose, the smallest proposes them,
+//! and the others intersect against their own extensions.
+//!
+//! # Set semantics
+//!
+//! The algorithm is stated over *sets*. Its extension indices are set-valued, its intersection
+//! step is an existence test, and its count minimization ranks by set cardinality. Accordingly,
+//! the differences here are `isize` and the relations are **expected to be sets**: each record
+//! present with multiplicity one.
+//!
+//! This expectation is documented rather than enforced, because a caller whose data are already
+//! distinct should not pay for a `distinct()` that does nothing. A caller who is unsure should
+//! apply `distinct()` to the collection before handing it to [`CollectionIndex::index`].
+//!
+//! Feeding a multiset in does not produce the multiset join. `propose` and `validate` multiply
+//! the matched record's multiplicity into the output, so multiplicities scale rather than filter,
+//! and `count` reports set cardinalities that no longer describe the proposals. Nor is this
+//! repairable by adjusting the operators: with relations of arity above two, an extension index
+//! is a *projection* of its relation, and a projection's multiplicity is a count of completions
+//! rather than the record's own annotation. Carrying annotations correctly through a worst-case
+//! optimal join requires indicator projections and a rule that each relation contributes its
+//! annotation exactly once, when its last attribute binds — that is InsideOut (Abo Khamis, Ngo,
+//! Rudra, "FAQ: Questions Asked Frequently", PODS 2016), a different algorithm with different
+//! indices, not a tuning of this one.
+
 use std::hash::Hash;
+use std::rc::Rc;
 
-use timely::progress::Timestamp;
+use timely::progress::{Antichain, Timestamp};
 use timely::dataflow::operators::vec::Partition;
 use timely::dataflow::operators::Concatenate;
 
 use differential_dataflow::{ExchangeData, VecCollection, AsCollection};
-use differential_dataflow::difference::{Monoid, Multiply};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::TraceAgent;
 
@@ -13,43 +43,60 @@ pub mod altneu;
 pub mod calculus;
 pub mod operators;
 
+/// Holds back logical compaction so that total-order time comparisons stay meaningful.
+///
+/// Conventional compaction collapses unequal times to the frontier, which would lose the
+/// distinction between "strictly before" and "at the same time" that the delta discipline
+/// rests on. See [`crate::operators::half_join`].
+pub type FrontierFunc<T> = Rc<dyn Fn(&T, &mut Antichain<T>)>;
+
+/// Compares an arranged record's time against a prefix's own time, in the *total order*.
+///
+/// The two useful choices are strictly-less and less-or-equal, which is how a delta query
+/// decides whether a stage sees updates concurrent with the delta it is responding to.
+pub type Comparison<T> = Rc<dyn Fn(&T, &T) -> bool>;
+
 /// A type capable of extending a stream of prefixes.
 ///
 /**
     Implementors of `PrefixExtension` provide types and methods for extending a differential dataflow collection,
     via the three methods `count`, `propose`, and `validate`.
+
+    Each prefix travels with a payload time alongside it. The payload starts as the prefix's own
+    time, accumulates the times of the records it matches, and is delayed to once the delta region
+    is left. The update itself stays at the time it entered on, which is what lets the total-order
+    comparison decide exactly once which stage produces each output.
 **/
-pub trait PrefixExtender<'scope, T: Timestamp, R: Monoid+Multiply<Output = R>> {
+pub trait PrefixExtender<'scope, T: Timestamp> {
     /// The required type of prefix to extend.
     type Prefix;
     /// The type to be produced as extension.
     type Extension;
     /// Annotates prefixes with the number of extensions the relation would propose.
-    fn count(&mut self, prefixes: VecCollection<'scope, T, (Self::Prefix, usize, usize), R>, index: usize) -> VecCollection<'scope, T, (Self::Prefix, usize, usize), R>;
+    fn count(&mut self, prefixes: VecCollection<'scope, T, ((Self::Prefix, usize, usize), T), isize>, index: usize) -> VecCollection<'scope, T, ((Self::Prefix, usize, usize), T), isize>;
     /// Extends each prefix with corresponding extensions.
-    fn propose(&mut self, prefixes: VecCollection<'scope, T, Self::Prefix, R>) -> VecCollection<'scope, T, (Self::Prefix, Self::Extension), R>;
+    fn propose(&mut self, prefixes: VecCollection<'scope, T, (Self::Prefix, T), isize>) -> VecCollection<'scope, T, ((Self::Prefix, Self::Extension), T), isize>;
     /// Restricts proposed extensions by those the extender would have proposed.
-    fn validate(&mut self, extensions: VecCollection<'scope, T, (Self::Prefix, Self::Extension), R>) -> VecCollection<'scope, T, (Self::Prefix, Self::Extension), R>;
+    fn validate(&mut self, extensions: VecCollection<'scope, T, ((Self::Prefix, Self::Extension), T), isize>) -> VecCollection<'scope, T, ((Self::Prefix, Self::Extension), T), isize>;
 }
 
-pub trait ProposeExtensionMethod<'scope, T: Timestamp, P: ExchangeData+Ord, R: Monoid+Multiply<Output = R>> {
-    fn propose_using<PE: PrefixExtender<'scope, T, R, Prefix=P>>(self, extender: &mut PE) -> VecCollection<'scope, T, (P, PE::Extension), R>;
-    fn extend<E: ExchangeData+Ord>(self, extenders: &mut [&mut dyn PrefixExtender<'scope, T,R,Prefix=P,Extension=E>]) -> VecCollection<'scope, T, (P, E), R>;
+pub trait ProposeExtensionMethod<'scope, T: Timestamp, P: ExchangeData+Ord> {
+    fn propose_using<PE: PrefixExtender<'scope, T, Prefix=P>>(self, extender: &mut PE) -> VecCollection<'scope, T, ((P, PE::Extension), T), isize>;
+    fn extend<E: ExchangeData+Ord>(self, extenders: &mut [&mut dyn PrefixExtender<'scope, T, Prefix=P, Extension=E>]) -> VecCollection<'scope, T, ((P, E), T), isize>;
 }
 
-impl<'scope, T, P, R> ProposeExtensionMethod<'scope, T, P, R> for VecCollection<'scope, T, P, R>
+impl<'scope, T, P> ProposeExtensionMethod<'scope, T, P> for VecCollection<'scope, T, (P, T), isize>
 where
     T: Timestamp,
     P: ExchangeData+Ord,
-    R: Monoid+Multiply<Output = R>+'static,
 {
-    fn propose_using<PE>(self, extender: &mut PE) -> VecCollection<'scope, T, (P, PE::Extension), R>
+    fn propose_using<PE>(self, extender: &mut PE) -> VecCollection<'scope, T, ((P, PE::Extension), T), isize>
     where
-        PE: PrefixExtender<'scope, T, R, Prefix=P>
+        PE: PrefixExtender<'scope, T, Prefix=P>
     {
         extender.propose(self)
     }
-    fn extend<E>(self, extenders: &mut [&mut dyn PrefixExtender<'scope, T,R,Prefix=P,Extension=E>]) -> VecCollection<'scope, T, (P, E), R>
+    fn extend<E>(self, extenders: &mut [&mut dyn PrefixExtender<'scope, T, Prefix=P, Extension=E>]) -> VecCollection<'scope, T, ((P, E), T), isize>
     where
         E: ExchangeData+Ord
     {
@@ -58,12 +105,12 @@ where
             extenders[0].propose(self)
         }
         else {
-            let mut counts = self.clone().map(|p| (p, 1 << 31, 0));
+            let mut counts = self.clone().map(|(p, payload)| ((p, 1 << 31, 0), payload));
             for (index,extender) in extenders.iter_mut().enumerate() {
                 counts = extender.count(counts, index);
             }
 
-            let parts = counts.inner.partition(extenders.len() as u64, |((p, _, i),t,d)| (i as u64, (p,t,d)));
+            let parts = counts.inner.partition(extenders.len() as u64, |(((p, _, i), payload),t,d)| (i as u64, ((p, payload),t,d)));
 
             let mut results = Vec::new();
             for (index, nominations) in parts.into_iter().enumerate() {
@@ -80,12 +127,12 @@ where
     }
 }
 
-pub trait ValidateExtensionMethod<'scope, T: Timestamp, R: Monoid+Multiply<Output = R>, P, E> {
-    fn validate_using<PE: PrefixExtender<'scope, T, R, Prefix=P, Extension=E>>(self, extender: &mut PE) -> VecCollection<'scope, T, (P, E), R>;
+pub trait ValidateExtensionMethod<'scope, T: Timestamp, P, E> {
+    fn validate_using<PE: PrefixExtender<'scope, T, Prefix=P, Extension=E>>(self, extender: &mut PE) -> VecCollection<'scope, T, ((P, E), T), isize>;
 }
 
-impl<'scope, T: Timestamp, R: Monoid+Multiply<Output = R>, P, E> ValidateExtensionMethod<'scope, T, R, P, E> for VecCollection<'scope, T, (P, E), R> {
-    fn validate_using<PE: PrefixExtender<'scope, T, R, Prefix=P, Extension=E>>(self, extender: &mut PE) -> VecCollection<'scope, T, (P, E), R> {
+impl<'scope, T: Timestamp, P, E> ValidateExtensionMethod<'scope, T, P, E> for VecCollection<'scope, T, ((P, E), T), isize> {
+    fn validate_using<PE: PrefixExtender<'scope, T, Prefix=P, Extension=E>>(self, extender: &mut PE) -> VecCollection<'scope, T, ((P, E), T), isize> {
         extender.validate(self)
     }
 }
@@ -93,40 +140,67 @@ impl<'scope, T: Timestamp, R: Monoid+Multiply<Output = R>, P, E> ValidateExtensi
 // These are all defined here so that users can be assured a common layout.
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::trace::implementations::{KeySpine, ValSpine};
-type TraceValHandle<K,V,T,R> = TraceAgent<ValSpine<K,V,T,R>>;
-type TraceKeyHandle<K,T,R> = TraceAgent<KeySpine<K,T,R>>;
+type TraceValHandle<K,V,T> = TraceAgent<ValSpine<K,V,T,isize>>;
+type TraceKeyHandle<K,T> = TraceAgent<KeySpine<K,T,isize>>;
 
 /// The three arrangements a relation must present to extend prefixes.
 ///
 /// The arrangements are scope-bound rather than exported traces, so that the operators
 /// reading them observe timely's own progress tracking. An imported trace instead reports
 /// its frontier in-band, and in a cycle those statements circulate without ever settling.
-#[derive(Clone)]
-pub struct CollectionIndex<'scope, K, V, T, R>
+///
+/// The indexed collection is expected to be a set; see the note on set semantics in [`crate`].
+pub struct CollectionIndex<'scope, K, V, T>
 where
     K: ExchangeData,
     V: ExchangeData,
     T: Lattice+ExchangeData+Timestamp,
-    R: Monoid+Multiply<Output = R>+ExchangeData,
 {
     /// A trace of type (K, ()), used to count extensions for each prefix.
-    count_trace: Arranged<'scope, TraceKeyHandle<K, T, isize>>,
+    count_trace: Arranged<'scope, TraceKeyHandle<K, T>>,
 
     /// A trace of type (K, V), used to propose extensions for each prefix.
-    propose_trace: Arranged<'scope, TraceValHandle<K, V, T, R>>,
+    propose_trace: Arranged<'scope, TraceValHandle<K, V, T>>,
 
     /// A trace of type ((K, V), ()), used to validate proposed extensions.
-    validate_trace: Arranged<'scope, TraceKeyHandle<(K, V), T, R>>,
+    validate_trace: Arranged<'scope, TraceKeyHandle<(K, V), T>>,
+
+    /// Holds back compaction; see [`FrontierFunc`].
+    frontier_func: FrontierFunc<T>,
+
+    /// Decides which arranged times a prefix sees; see [`Comparison`].
+    comparison: Comparison<T>,
 }
 
-impl<'scope, K, V, T, R> CollectionIndex<'scope, K, V, T, R>
+impl<'scope, K, V, T> Clone for CollectionIndex<'scope, K, V, T>
+where
+    K: ExchangeData,
+    V: ExchangeData,
+    T: Lattice+ExchangeData+Timestamp,
+{
+    fn clone(&self) -> Self {
+        CollectionIndex {
+            count_trace: self.count_trace.clone(),
+            propose_trace: self.propose_trace.clone(),
+            validate_trace: self.validate_trace.clone(),
+            frontier_func: Rc::clone(&self.frontier_func),
+            comparison: Rc::clone(&self.comparison),
+        }
+    }
+}
+
+impl<'scope, K, V, T> CollectionIndex<'scope, K, V, T>
 where
     K: ExchangeData+Hash,
     V: ExchangeData+Hash,
     T: Lattice+ExchangeData+Timestamp,
-    R: Monoid+Multiply<Output = R>+ExchangeData,
 {
-    pub fn index(collection: VecCollection<'scope, T, (K, V), R>) -> Self {
+
+    pub fn index<FF, CF>(collection: VecCollection<'scope, T, (K, V), isize>, frontier_func: FF, comparison: CF) -> Self
+    where
+        FF: Fn(&T, &mut Antichain<T>) + 'static,
+        CF: Fn(&T, &T) -> bool + 'static,
+    {
         // We need to count the number of (k, v) pairs and not rely on the given Monoid R and its binary addition operation.
         // counts and validate can share the base arrangement
         let arranged = collection.clone().arrange_by_self();
@@ -144,9 +218,11 @@ where
             count_trace: counts,
             propose_trace: propose,
             validate_trace: validate,
+            frontier_func: Rc::new(frontier_func),
+            comparison: Rc::new(comparison),
         }
     }
-    pub fn extend_using<P, F: Fn(&P)->K+Clone>(&self, logic: F) -> CollectionExtender<'scope, K, V, T, R, P, F> {
+    pub fn extend_using<P, F: Fn(&P)->K+Clone>(&self, logic: F) -> CollectionExtender<'scope, K, V, T, P, F> {
         CollectionExtender {
             phantom: std::marker::PhantomData,
             indices: self.clone(),
@@ -155,40 +231,57 @@ where
     }
 }
 
-pub struct CollectionExtender<'scope, K, V, T, R, P, F>
+pub struct CollectionExtender<'scope, K, V, T, P, F>
 where
     K: ExchangeData,
     V: ExchangeData,
     T: Lattice+ExchangeData+Timestamp,
-    R: Monoid+Multiply<Output = R>+ExchangeData,
     F: Fn(&P)->K+Clone,
 {
     phantom: std::marker::PhantomData<P>,
-    indices: CollectionIndex<'scope, K, V, T, R>,
+    indices: CollectionIndex<'scope, K, V, T>,
     key_selector: F,
 }
 
-impl<'scope, T, K, V, R, P, F> PrefixExtender<'scope, T, R> for CollectionExtender<'scope, K, V, T, R, P, F>
+impl<'scope, K, V, T, P, F> CollectionExtender<'scope, K, V, T, P, F>
+where
+    K: ExchangeData,
+    V: ExchangeData,
+    T: Lattice+ExchangeData+Timestamp,
+    F: Fn(&P)->K+Clone,
+{
+    /// The index's time closures, as plain callables the operators can accept.
+    fn time_logic(&self) -> (impl Fn(&T, &mut Antichain<T>) + 'static, impl Fn(&T, &T) -> bool + 'static) {
+        let frontier_func = Rc::clone(&self.indices.frontier_func);
+        let comparison = Rc::clone(&self.indices.comparison);
+        (move |t: &T, a: &mut Antichain<T>| frontier_func(t, a), move |t1: &T, t2: &T| comparison(t1, t2))
+    }
+}
+
+impl<'scope, T, K, V, P, F> PrefixExtender<'scope, T> for CollectionExtender<'scope, K, V, T, P, F>
 where
     T: Timestamp + Lattice + ExchangeData + Hash,
-    K: ExchangeData+Hash+Default,
-    V: ExchangeData+Hash+Default,
+    K: ExchangeData+Hash,
+    V: ExchangeData+Hash,
     P: ExchangeData,
-    R: Monoid+Multiply<Output = R>+ExchangeData,
     F: Fn(&P)->K+Clone+'static,
 {
     type Prefix = P;
     type Extension = V;
 
-    fn count(&mut self, prefixes: VecCollection<'scope, T, (P, usize, usize), R>, index: usize) -> VecCollection<'scope, T, (P, usize, usize), R> {
-        operators::count::count(prefixes, self.indices.count_trace.clone(), self.key_selector.clone(), index)
+    fn count(&mut self, prefixes: VecCollection<'scope, T, ((P, usize, usize), T), isize>, index: usize) -> VecCollection<'scope, T, ((P, usize, usize), T), isize> {
+        let (frontier_func, comparison) = self.time_logic();
+        operators::count::count(prefixes, self.indices.count_trace.clone(), self.key_selector.clone(), index, frontier_func, comparison)
     }
 
-    fn propose(&mut self, prefixes: VecCollection<'scope, T, P, R>) -> VecCollection<'scope, T, (P, V), R> {
-        operators::propose::propose(prefixes, self.indices.propose_trace.clone(), self.key_selector.clone())
+    fn propose(&mut self, prefixes: VecCollection<'scope, T, (P, T), isize>) -> VecCollection<'scope, T, ((P, V), T), isize> {
+        let (frontier_func, comparison) = self.time_logic();
+        operators::propose::propose(prefixes, self.indices.propose_trace.clone(), self.key_selector.clone(), frontier_func, comparison)
     }
 
-    fn validate(&mut self, extensions: VecCollection<'scope, T, (P, V), R>) -> VecCollection<'scope, T, (P, V), R> {
-        operators::validate::validate(extensions, self.indices.validate_trace.clone(), self.key_selector.clone())
+    fn validate(&mut self, extensions: VecCollection<'scope, T, ((P, V), T), isize>) -> VecCollection<'scope, T, ((P, V), T), isize> {
+        let (frontier_func, comparison) = self.time_logic();
+        let key_selector = self.key_selector.clone();
+        operators::validate::validate(extensions, self.indices.validate_trace.clone(), key_selector, frontier_func, comparison)
     }
 }
diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs
index c46be8c25..44756dd67 100644
--- a/dogsdogsdogs/src/operators/count.rs
+++ b/dogsdogsdogs/src/operators/count.rs
@@ -1,41 +1,75 @@
-use differential_dataflow::{ExchangeData, VecCollection, Hashable};
-use differential_dataflow::difference::{Semigroup, Monoid, Multiply};
+use timely::container::CapacityContainerBuilder;
+use timely::container::PushInto;
+use timely::progress::Antichain;
+
+use differential_dataflow::{AsCollection, ExchangeData, VecCollection, Hashable};
+use differential_dataflow::difference::Monoid;
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::{BatchCursor, BatchTimeGat, BatchVal, Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::implementations::BatchContainer;
 
-/// Reports a number of extensions to a stream of prefixes.
+/// Updates a stream of prefix routing judgements based on approximate counts.
+///
+/// Each prefix observes the changes in distinct values over time, and treats this as a lower
+/// bound on the count that will be experienced. When the lower bound improves on the routing
+/// judgement's current count, it is overwritten and the `index` argument is substituted in.
 ///
-/// This method takes as input a stream of `(prefix, count, index)` triples.
-/// For each triple, it extracts a key using `key_selector`, and finds the
-/// associated count in `arrangement`. If the found count is less than `count`,
-/// the `count` and `index` fields are overwritten with their new values.
-pub fn count<'scope, Tr, K, R, F, P>(
-    prefixes: VecCollection<'scope, Tr::Time, (P, usize, usize), R>,
+/// A prefix is dropped only when its key is absent from `arrangement` entirely, which is only
+/// expected to happen when there have never been counts (they are meant to be non-negative).
+pub fn count<'scope, Tr, K, F, P, R, FF, CF>(
+    prefixes: VecCollection<'scope, Tr::Time, ((P, usize, usize), Tr::Time), R>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
     index: usize,
-) -> VecCollection<'scope, Tr::Time, (P, usize, usize), R>
+    frontier_func: FF,
+    comparison: CF,
+) -> VecCollection<'scope, Tr::Time, ((P, usize, usize), Tr::Time), R>
 where
     Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
-    BatchCursor<Tr>: Cursor<Time = Tr::Time, Diff=isize>,
-    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
-    for<'a> BatchDiff<Tr> : Semigroup<BatchDiffGat<'a, Tr>>,
-    K: Hashable + Ord + Default + 'static,
-    R: Monoid+Multiply<Output = R>+ExchangeData,
-    F: Fn(&P)->K+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time, Diff = isize>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
+    K: Hashable + ExchangeData,
+    R: ExchangeData + Monoid,
+    F: Fn(&P)->K+'static,
     P: ExchangeData,
+    FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
 {
-    crate::operators::lookup_map(
-        prefixes,
+    // The payload time is carried in the record as well as in the half-join's own payload
+    // slot, because the output closure is handed the joined times rather than the payload.
+    let requests = prefixes.map(move |(triple, payload)| {
+        (key_selector(&triple.0), (triple, payload.clone()), payload)
+    });
+
+    type Output<P, T, R> = CapacityContainerBuilder<Vec<(((P, usize, usize), T), T, R)>>;
+
+    let output_func = move |
+        builder: &mut Output<P, Tr::Time, R>,
+        _key: &K,
+        val1: &((P, usize, usize), Tr::Time),
+        _val2: BatchVal<'_, Tr>,
+        initial: &Tr::Time,
+        diff1: &R,
+        output: &mut Vec<(Tr::Time, isize)>,
+    | {
+        // Each diff is a number of distinct values in/out, so sum the absolute values.
+        // A zero count does not mean no changes; it could mean swaps of values, so we
+        // do not drop prefixes with zero sums. We may inappropriately favor them, but
+        // it is a performance issue not a correctness issue.
+        let found: usize = output.iter().map(|(_,diff)| diff.abs()).sum::<isize>() as usize;
+        let ((prefix, count, best), payload) = val1;
+        let triple = if *count < found { (prefix.clone(), *count, *best) }
+                     else              { (prefix.clone(), found, index) };
+        builder.push_into(((triple, payload.clone()), initial.clone(), diff1.clone()));
+    };
+
+    crate::operators::half_join::half_join_internal_unsafe::<_, _, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
+        requests,
         arrangement,
-        move |p: &(P,usize,usize), k: &mut K| { *k = key_selector(&p.0); },
-        move |(p,c,i), r, _, s| {
-            let s = *s as usize;
-            if *c < s { ((p.clone(), *c, *i), r.clone()) }
-            else      { ((p.clone(), s, index), r.clone()) }
-        },
-        Default::default(),
-        Default::default(),
-        Default::default(),
+        frontier_func,
+        comparison,
+        |_timer, _count| false,
+        output_func,
     )
+    .as_collection()
 }
diff --git a/dogsdogsdogs/src/operators/lookup_map.rs b/dogsdogsdogs/src/operators/lookup_map.rs
deleted file mode 100644
index dc2a3f494..000000000
--- a/dogsdogsdogs/src/operators/lookup_map.rs
+++ /dev/null
@@ -1,145 +0,0 @@
-use std::collections::HashMap;
-
-use timely::dataflow::channels::pact::{Pipeline, Exchange};
-use timely::dataflow::operators::Operator;
-use timely::PartialOrder;
-use timely::progress::Antichain;
-
-use differential_dataflow::{ExchangeData, VecCollection, AsCollection, Hashable};
-use differential_dataflow::difference::{IsZero, Semigroup, Monoid};
-use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, BatchVal, Cursor, Navigable, TraceReader};
-use differential_dataflow::trace::cursor::cursor_list;
-use differential_dataflow::trace::implementations::BatchContainer;
-
-/// Proposes extensions to a stream of prefixes.
-///
-/// This method takes a stream of prefixes and for each determines a
-/// key with `key_selector` and then proposes all pair af the prefix
-/// and values associated with the key in `arrangement`.
-pub fn lookup_map<'scope, D, K, R, Tr, F, DOut, ROut, S>(
-    prefixes: VecCollection<'scope, Tr::Time, D, R>,
-    mut arrangement: Arranged<'scope, Tr>,
-    key_selector: F,
-    mut output_func: S,
-    supplied_key0: K,
-    supplied_key1: K,
-    supplied_key2: K,
-) -> VecCollection<'scope, Tr::Time, DOut, ROut>
-where
-    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
-    for<'a> BatchCursor<Tr>: Cursor<
-        Time = Tr::Time,
-        Diff : Semigroup<BatchDiffGat<'a, Tr>>+Monoid+ExchangeData,
-    >,
-    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
-    K: Hashable + Ord + 'static,
-    F: FnMut(&D, &mut K)+Clone+'static,
-    D: ExchangeData,
-    R: ExchangeData+Monoid,
-    DOut: Clone+'static,
-    ROut: Monoid + 'static,
-    S: FnMut(&D, &R, BatchVal<'_, Tr>, &BatchDiff<Tr>)->(DOut, ROut)+'static,
-{
-    // No need to block physical merging for this operator.
-    arrangement.trace.set_physical_compaction(Antichain::new().borrow());
-    let mut propose_trace = Some(arrangement.trace);
-    let propose_stream = arrangement.stream;
-
-    let mut stash = HashMap::new();
-    let mut logic1 = key_selector.clone();
-    let mut logic2 = key_selector.clone();
-
-    let mut key: K = supplied_key0;
-    let exchange = Exchange::new(move |update: &(D,Tr::Time,R)| {
-        logic1(&update.0, &mut key);
-        key.hashed().into()
-    });
-
-    let mut key1: K = supplied_key1;
-    let mut key2: K = supplied_key2;
-
-    prefixes.inner.binary_frontier(propose_stream, exchange, Pipeline, "LookupMap", move |_,_| move |(input1, frontier1), (input2, frontier2), output| {
-
-        // drain the first input, stashing requests.
-        input1.for_each(|capability, data| {
-            stash.entry(capability.retain(0))
-                 .or_insert(Vec::new())
-                 .extend(data.drain(..))
-        });
-
-        // Drain input batches; although we do not observe them, we want access to the input
-        // to observe the frontier and to drive scheduling.
-        input2.for_each(|_, _| { });
-
-        if let Some(ref mut trace) = propose_trace {
-
-            for (capability, prefixes) in stash.iter_mut() {
-
-                // defer requests at incomplete times.
-                // NOTE: not all updates may be at complete times, but if this test fails then none of them are.
-                if !frontier2.less_equal(capability.time()) {
-
-                    let mut session = output.session(capability);
-
-                    // sort requests for in-order cursor traversal. could consolidate?
-                    prefixes.sort_by(|x,y| {
-                        logic2(&x.0, &mut key1);
-                        logic2(&y.0, &mut key2);
-                        key1.cmp(&key2)
-                    });
-
-                    let batches = trace.batches_through(Antichain::new().borrow()).unwrap();
-                    let (mut cursor, storage) = cursor_list(batches);
-                    // Key container to stage keys for comparison.
-                    let mut key_con = <BatchCursor<Tr> as Cursor>::KeyContainer::with_capacity(1);
-                    for &mut (ref prefix, ref time, ref mut diff) in prefixes.iter_mut() {
-                        if !frontier2.less_equal(time) {
-                            logic2(prefix, &mut key1);
-                            key_con.clear(); key_con.push_own(&key1);
-                            cursor.seek_key(&storage, key_con.index(0));
-                            if cursor.get_key(&storage) == Some(key_con.index(0)) {
-                                while let Some(value) = cursor.get_val(&storage) {
-                                    let mut count = BatchDiff::<Tr>::zero();
-                                    cursor.map_times(&storage, |t, d| {
-                                        if <BatchCursor<Tr> as Cursor>::owned_time(t).less_equal(time) { count.plus_equals(&d); }
-                                    });
-                                    if !count.is_zero() {
-                                        let (dout, rout) = output_func(prefix, diff, value, &count);
-                                        if !rout.is_zero() {
-                                            session.give((dout, time.clone(), rout));
-                                        }
-                                    }
-                                    cursor.step_val(&storage);
-                                }
-                                cursor.rewind_vals(&storage);
-                            }
-                            *diff = R::zero();
-                        }
-                    }
-
-                    prefixes.retain(|ptd| !ptd.2.is_zero());
-                }
-            }
-
-        }
-
-        // drop fully processed capabilities.
-        stash.retain(|_,prefixes| !prefixes.is_empty());
-
-        // The logical merging frontier depends on both input1 and stash.
-        let mut frontier = timely::progress::frontier::Antichain::new();
-        for time in frontier1.frontier().to_vec() {
-            frontier.insert(time);
-        }
-        for key in stash.keys() {
-            frontier.insert(key.time().clone());
-        }
-        propose_trace.as_mut().map(|trace| trace.set_logical_compaction(frontier.borrow()));
-
-        if frontier1.is_empty() && stash.is_empty() {
-            propose_trace = None;
-        }
-
-    }).as_collection()
-}
diff --git a/dogsdogsdogs/src/operators/mod.rs b/dogsdogsdogs/src/operators/mod.rs
index feba05f1c..6da214107 100644
--- a/dogsdogsdogs/src/operators/mod.rs
+++ b/dogsdogsdogs/src/operators/mod.rs
@@ -1,12 +1,10 @@
 pub mod half_join;
-pub mod lookup_map;
 
 pub mod count;
 pub mod propose;
 pub mod validate;
 
 pub use self::half_join::half_join;
-pub use self::lookup_map::lookup_map;
 pub use self::count::count;
-pub use self::propose::{propose, propose_distinct};
+pub use self::propose::propose;
 pub use self::validate::validate;
diff --git a/dogsdogsdogs/src/operators/propose.rs b/dogsdogsdogs/src/operators/propose.rs
index 9415985bd..008576184 100644
--- a/dogsdogsdogs/src/operators/propose.rs
+++ b/dogsdogsdogs/src/operators/propose.rs
@@ -1,75 +1,49 @@
+use std::ops::Mul;
+
+use timely::progress::Antichain;
+
 use differential_dataflow::{ExchangeData, VecCollection, Hashable};
-use differential_dataflow::difference::{Semigroup, Monoid, Multiply};
+use differential_dataflow::difference::{Monoid, Semigroup};
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchTimeGat, Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::implementations::BatchContainer;
 
 /// Proposes extensions to a prefix stream.
 ///
-/// This method takes a collection `prefixes` and an arrangement `arrangement` and for each
-/// update in the collection joins with the accumulated arranged records at a time less or
-/// equal to that of the update. Note that this is not a join by itself, but can be used to
-/// create a join if the `prefixes` collection is also arranged and responds to changes that
-/// `arrangement` undergoes. More complicated patterns are also appropriate, as in the case
-/// of delta queries.
-pub fn propose<'scope, Tr, K, F, P, V>(
-    prefixes: VecCollection<'scope, Tr::Time, P, BatchDiff<Tr>>,
-    arrangement: Arranged<'scope, Tr>,
-    key_selector: F,
-) -> VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>
-where
-    Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
-    for<'a> BatchCursor<Tr>: Cursor<
-        Time = Tr::Time,
-        ValOwn = V,
-        Diff: Monoid+Multiply<Output = BatchDiff<Tr>>+ExchangeData+Semigroup<BatchDiffGat<'a, Tr>>,
-    >,
-    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
-    K: Hashable + Default + Ord + 'static,
-    F: Fn(&P)->K+Clone+'static,
-    P: ExchangeData,
-    V: Clone + 'static,
-{
-    crate::operators::lookup_map(
-        prefixes,
-        arrangement,
-        move |p: &P, k: &mut K | { *k = key_selector(p); },
-        move |prefix, diff, value, sum| ((prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)), diff.clone().multiply(sum)),
-        Default::default(),
-        Default::default(),
-        Default::default(),
-    )
-}
-
-/// Proposes distinct extensions to a prefix stream.
+/// This operator matches streamed updates with arranged updates, and pairs the streamed updates
+/// with arranged updates whose times are less or equal under the *total order* on timestamps.
+/// This inequality is allowed to either be strict or non-strict, as determined by `comparison`.
+/// The total order allows the caller to ensure that each pair of updates match exactly once.
+/// The streamed updates also carry a time as data, and that time is advanced (by lattice join)
+/// by the time of the arranged update. The time of the streamed update cannot be advanced, as
+/// it needs to stay put to ensure the total order math works out.
 ///
-/// Unlike `propose`, this method does not scale the multiplicity of matched
-/// prefixes by the number of matches in `arrangement`. This can be useful to
-/// avoid the need to prepare an arrangement of distinct extensions.
-pub fn propose_distinct<'scope, Tr, K, F, P, V>(
-    prefixes: VecCollection<'scope, Tr::Time, P, BatchDiff<Tr>>,
+/// The arrangement is expected to hold a *set*: see the note on set semantics in [`crate`].
+pub fn propose<'scope, Tr, K, F, P, V, R, FF, CF>(
+    prefixes: VecCollection<'scope, Tr::Time, (P, Tr::Time), R>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
-) -> VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>
+    frontier_func: FF,
+    comparison: CF,
+) -> VecCollection<'scope, Tr::Time, ((P, V), Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
 where
     Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
-    for<'a> BatchCursor<Tr>: Cursor<
-        Time = Tr::Time,
-        ValOwn = V,
-        Diff : Semigroup<BatchDiffGat<'a, Tr>>+Monoid+Multiply<Output = BatchDiff<Tr>>+ExchangeData,
-    >,
-    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=K>,
-    K: Hashable + Default + Ord + 'static,
-    F: Fn(&P)->K+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time, ValOwn = V>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=K>,
+    K: Hashable + ExchangeData,
+    R: ExchangeData + Monoid + Mul<BatchDiff<Tr>, Output: Semigroup>,
+    F: Fn(&P)->K+'static,
     P: ExchangeData,
     V: Clone + 'static,
+    FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
 {
-    crate::operators::lookup_map(
-        prefixes,
+    let requests = prefixes.map(move |(prefix, payload)| (key_selector(&prefix), prefix, payload));
+    crate::operators::half_join(
+        requests,
         arrangement,
-        move |p: &P, k: &mut K| { *k = key_selector(p); },
-        move |prefix, diff, value, _sum| ((prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)), diff.clone()),
-        Default::default(),
-        Default::default(),
-        Default::default(),
+        frontier_func,
+        comparison,
+        |_key, prefix, value| (prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)),
     )
 }
diff --git a/dogsdogsdogs/src/operators/validate.rs b/dogsdogsdogs/src/operators/validate.rs
index 262957f71..bb5f40273 100644
--- a/dogsdogsdogs/src/operators/validate.rs
+++ b/dogsdogsdogs/src/operators/validate.rs
@@ -1,39 +1,51 @@
-use std::hash::Hash;
+use std::ops::Mul;
+
+use timely::progress::Antichain;
 
 use differential_dataflow::{ExchangeData, VecCollection};
-use differential_dataflow::difference::{Semigroup, Monoid, Multiply};
+use differential_dataflow::difference::{Monoid, Semigroup};
 use differential_dataflow::operators::arrange::Arranged;
-use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchTimeGat, Cursor, Navigable, TraceReader};
+use differential_dataflow::trace::implementations::BatchContainer;
 
-/// Proposes extensions to a stream of prefixes.
+/// Restricts proposed extensions to those the arrangement would also have proposed.
+///
+/// This operator matches streamed updates with arranged updates, and pairs the streamed updates
+/// with arranged updates whose times are less or equal under the *total order* on timestamps.
+/// This inequality is allowed to either be strict or non-strict, as determined by `comparison`.
+/// The total order allows the caller to ensure that each pair of updates match exactly once.
+/// The streamed updates also carry a time as data, and that time is advanced (by lattice join)
+/// by the time of the arranged update. The time of the streamed update cannot be advanced, as
+/// it needs to stay put to ensure the total order math works out.
 ///
-/// This method takes a stream of prefixes and for each determines a
-/// key with `key_selector` and then proposes all pair af the prefix
-/// and values associated with the key in `arrangement`.
-pub fn validate<'scope, K, V, Tr, F, P>(
-    extensions: VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>,
+/// The arrangement is expected to hold a *set*: see the note on set semantics in [`crate`].
+pub fn validate<'scope, Tr, K, V, F, P, R, FF, CF>(
+    extensions: VecCollection<'scope, Tr::Time, ((P, V), Tr::Time), R>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
-) -> VecCollection<'scope, Tr::Time, (P, V), BatchDiff<Tr>>
+    frontier_func: FF,
+    comparison: CF,
+) -> VecCollection<'scope, Tr::Time, ((P, V), Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
 where
     Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
-    for<'a> BatchCursor<Tr>: Cursor<
-        Time = Tr::Time,
-        Diff : Semigroup<BatchDiffGat<'a, Tr>>+Monoid+Multiply<Output = BatchDiff<Tr>>+ExchangeData,
-    >,
-    <BatchCursor<Tr> as Cursor>::KeyContainer: differential_dataflow::trace::implementations::BatchContainer<Owned=(K,V)>,
-    K: Ord+Hash+Clone+Default + 'static,
-    V: ExchangeData+Hash+Default,
-    F: Fn(&P)->K+Clone+'static,
+    BatchCursor<Tr>: Cursor<Time = Tr::Time>,
+    <BatchCursor<Tr> as Cursor>::KeyContainer: BatchContainer<Owned=(K,V)>,
+    K: ExchangeData + std::hash::Hash,
+    V: ExchangeData + std::hash::Hash,
+    R: ExchangeData + Monoid + Mul<BatchDiff<Tr>, Output: Semigroup>,
+    F: Fn(&P)->K+'static,
     P: ExchangeData,
+    FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
+    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
 {
-    crate::operators::lookup_map(
-        extensions,
+    let requests = extensions.map(move |((prefix, extension), payload)| {
+        ((key_selector(&prefix), extension.clone()), (prefix, extension), payload)
+    });
+    crate::operators::half_join(
+        requests,
         arrangement,
-        move |(pre,val),key| { *key = (key_selector(pre), val.clone()); },
-        |(pre,val),r,_,_| ((pre.clone(), val.clone()), r.clone()),
-        Default::default(),
-        Default::default(),
-        Default::default(),
+        frontier_func,
+        comparison,
+        |_key, extended, _value| extended.clone(),
     )
 }
diff --git a/dogsdogsdogs/tests/lookup_map_regression.rs b/dogsdogsdogs/tests/lookup_map_regression.rs
deleted file mode 100644
index a48b5513c..000000000
--- a/dogsdogsdogs/tests/lookup_map_regression.rs
+++ /dev/null
@@ -1,108 +0,0 @@
-//! Regression test for the `lookup_map` off-by-one (`key_con.index(1)` -> `index(0)`).
-//!
-//! `lookup_map` stages the probe key into a capacity-1 `KeyContainer`, pushes one
-//! element (valid index 0), and (before the fix) reads it back at index 1 — out of
-//! bounds, panicking on *every* probe. This test drives a triangle worst-case-optimal
-//! join (`extend` -> `propose`/`count`/`validate` -> `lookup_map`) over a single
-//! in-memory triangle: before the fix it panics; after the fix it finds (0, 1, 2).
-
-use std::sync::{Arc, Mutex};
-
-use timely::dataflow::operators::probe::Handle;
-use differential_dataflow::input::Input;
-
-use differential_dogs3::{CollectionIndex, altneu::AltNeu, ProposeExtensionMethod};
-
-#[test]
-fn lookup_map_triangle_wcoj_finds_triangle() {
-    let found: Arc<Mutex<Vec<(u32, u32, u32)>>> = Arc::new(Mutex::new(Vec::new()));
-    let found_outer = found.clone();
-
-    timely::execute(timely::Config::thread(), move |worker| {
-        let found_outer = found_outer.clone();
-        let mut probe = Handle::new();
-
-        let mut input = worker.dataflow::<usize, _, _>(|scope| {
-            let (input, edges) = scope.new_collection::<(u32, u32), isize>();
-
-            let forward = edges.clone();
-            let reverse = edges.map(|(x, y)| (y, x));
-
-            // Q(a,b,c) := E1(a,b), E2(b,c), E3(a,c), via the dogsdogsdogs WCOJ `extend`
-            // path, which routes through propose/count/validate -> lookup_map.
-            let triangles = scope.scoped::<AltNeu<usize>, _, _>("Triangles", |inner| {
-                let forward = forward.enter(inner);
-                let reverse = reverse.enter(inner);
-
-                let alt_forward = CollectionIndex::index(forward.clone());
-                let alt_reverse = CollectionIndex::index(reverse.clone());
-                let neu_forward = CollectionIndex::index(forward.clone().delay(|t| AltNeu::neu(t.time.clone())));
-                let neu_reverse = CollectionIndex::index(reverse.clone().delay(|t| AltNeu::neu(t.time.clone())));
-
-                //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
-                let changes1 = forward.clone()
-                    .extend(&mut [
-                        &mut neu_forward.extend_using(|(_a, b)| *b),
-                        &mut neu_forward.extend_using(|(a, _b)| *a),
-                    ])
-                    .map(|((a, b), c)| (a, b, c));
-
-                //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
-                let changes2 = forward.clone()
-                    .extend(&mut [
-                        &mut alt_reverse.extend_using(|(b, _c)| *b),
-                        &mut neu_reverse.extend_using(|(_b, c)| *c),
-                    ])
-                    .map(|((b, c), a)| (a, b, c));
-
-                //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                let changes3 = forward
-                    .extend(&mut [
-                        &mut alt_forward.extend_using(|(a, _c)| *a),
-                        &mut alt_reverse.extend_using(|(_a, c)| *c),
-                    ])
-                    .map(|((a, c), b)| (a, b, c));
-
-                changes1.concat(changes2).concat(changes3).leave(scope)
-            });
-
-            triangles
-                .inspect_batch(move |_t, xs| {
-                    let mut v = found_outer.lock().unwrap();
-                    // Only collect records present with positive multiplicity, so a
-                    // retraction (negative diff) or a net-zero update can't make the
-                    // test pass and weaken the regression signal.
-                    for (data, _t, r) in xs {
-                        if *r > 0 {
-                            v.push(*data);
-                        }
-                    }
-                })
-                .probe_with(&mut probe);
-
-            input
-        });
-
-        // A single triangle: directed edges 0->1, 1->2, 0->2.
-        input.advance_to(0);
-        input.insert((0, 1));
-        input.insert((1, 2));
-        input.insert((0, 2));
-        input.advance_to(1);
-        input.flush();
-        while probe.less_than(input.time()) {
-            worker.step();
-        }
-    })
-    .unwrap();
-
-    let mut got = found.lock().unwrap().clone();
-    got.sort();
-    got.dedup();
-    assert_eq!(
-        got,
-        vec![(0, 1, 2)],
-        "expected exactly the triangle (0,1,2); got {:?}",
-        got
-    );
-}
diff --git a/dogsdogsdogs/tests/wcoj_triangle.rs b/dogsdogsdogs/tests/wcoj_triangle.rs
new file mode 100644
index 000000000..8e84f2cea
--- /dev/null
+++ b/dogsdogsdogs/tests/wcoj_triangle.rs
@@ -0,0 +1,193 @@
+//! End-to-end checks of the worst-case-optimal `extend` path.
+//!
+//! Both tests drive a triangle query through `extend` -> `count`/`propose`/`validate` ->
+//! `half_join`. That path carries a payload time: each delta enters holding its own time, the
+//! extension stages join the times of the records they match into it, and the result is delayed
+//! to that payload on the way out.
+
+use std::collections::BTreeSet;
+use std::sync::{Arc, Mutex};
+
+use timely::dataflow::operators::probe::Handle;
+use timely::dataflow::operators::vec::Map;
+use timely::progress::Antichain;
+use differential_dataflow::AsCollection;
+use differential_dataflow::input::Input;
+
+use differential_dogs3::{CollectionIndex, altneu::AltNeu, ProposeExtensionMethod};
+
+/// Runs `Q(a,b,c) := E(a,b), E(b,c), E(a,c)` over `deltas`, and returns the output updates.
+///
+/// Updates are applied at the times they carry, and the computation is run out to `rounds`.
+fn triangles(deltas: &[((u32, u32), usize, isize)], rounds: usize) -> Vec<((u32, u32, u32), usize, isize)> {
+    let found = Arc::new(Mutex::new(Vec::new()));
+    let found_outer = found.clone();
+    let deltas = deltas.to_vec();
+
+    timely::execute(timely::Config::thread(), move |worker| {
+        let found_outer = found_outer.clone();
+        let deltas = deltas.clone();
+        let mut probe = Handle::new();
+
+        let mut input = worker.dataflow::<usize, _, _>(|scope| {
+            let (input, edges) = scope.new_collection::<(u32, u32), isize>();
+
+            let forward = edges.clone();
+            let reverse = edges.map(|(x, y)| (y, x));
+
+            let triangles = scope.scoped::<AltNeu<usize>, _, _>("Triangles", |inner| {
+                let forward = forward.enter(inner);
+                let reverse = reverse.enter(inner);
+
+                // Hold compaction back one base-time step so the alt/neu distinction survives,
+                // and compare in the total order on `AltNeu` (lexicographic on `(time, neu)`).
+                let frontier_func = |time: &AltNeu<usize>, antichain: &mut Antichain<AltNeu<usize>>| {
+                    antichain.insert(AltNeu::alt(time.time.saturating_sub(1)));
+                };
+                let comparison = |t1: &AltNeu<usize>, t2: &AltNeu<usize>| t1 <= t2;
+
+                let alt_forward = CollectionIndex::index(forward.clone(), frontier_func, comparison);
+                let alt_reverse = CollectionIndex::index(reverse.clone(), frontier_func, comparison);
+                let neu_forward = CollectionIndex::index(forward.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
+                let neu_reverse = CollectionIndex::index(reverse.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
+
+                let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
+
+                //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
+                let changes1 = deltas.clone()
+                    .extend(&mut [
+                        &mut neu_forward.extend_using(|(_a, b)| *b),
+                        &mut neu_forward.extend_using(|(a, _b)| *a),
+                    ])
+                    .map(|(((a, b), c), payload)| ((a, b, c), payload));
+
+                //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
+                let changes2 = deltas.clone()
+                    .extend(&mut [
+                        &mut alt_reverse.extend_using(|(b, _c)| *b),
+                        &mut neu_reverse.extend_using(|(_b, c)| *c),
+                    ])
+                    .map(|(((b, c), a), payload)| ((a, b, c), payload));
+
+                //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
+                let changes3 = deltas
+                    .extend(&mut [
+                        &mut alt_forward.extend_using(|(a, _c)| *a),
+                        &mut alt_reverse.extend_using(|(_a, c)| *c),
+                    ])
+                    .map(|(((a, c), b), payload)| ((a, b, c), payload));
+
+                // Delay updates to the payload time worked out while extending.
+                changes1.concat(changes2).concat(changes3)
+                    .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
+                    .leave(scope)
+            });
+
+            triangles
+                .inspect_batch(move |_t, xs| {
+                    let mut v = found_outer.lock().unwrap();
+                    v.extend(xs.iter().cloned());
+                })
+                .probe_with(&mut probe);
+
+            input
+        });
+
+        for round in 0..rounds {
+            input.advance_to(round);
+            for (edge, _, diff) in deltas.iter().filter(|(_, time, _)| *time == round) {
+                input.update(*edge, *diff);
+            }
+        }
+        input.advance_to(rounds);
+        input.flush();
+        while probe.less_than(input.time()) {
+            worker.step();
+        }
+    })
+    .unwrap();
+
+    let updates = found.lock().unwrap().clone();
+    updates
+}
+
+/// Accumulates `updates` as of `time`, discarding anything that cancels.
+fn accumulate<D: Ord + Clone>(updates: &[(D, usize, isize)], time: usize) -> BTreeSet<D> {
+    let mut counts = std::collections::BTreeMap::new();
+    for (data, _, diff) in updates.iter().filter(|(_, when, _)| *when <= time) {
+        *counts.entry(data.clone()).or_insert(0isize) += diff;
+    }
+    counts.into_iter().filter(|(_, c)| *c != 0).map(|(d, _)| d).collect()
+}
+
+/// The triangles of `edges`, by exhaustive search.
+fn triangles_of(edges: &BTreeSet<(u32, u32)>) -> BTreeSet<(u32, u32, u32)> {
+    let mut found = BTreeSet::new();
+    for &(a, b) in edges.iter() {
+        for &(b2, c) in edges.iter() {
+            if b2 == b && edges.contains(&(a, c)) {
+                found.insert((a, b, c));
+            }
+        }
+    }
+    found
+}
+
+#[test]
+fn wcoj_extend_finds_triangle() {
+    // A single triangle: directed edges 0->1, 1->2, 0->2, all at time 0.
+    let deltas = vec![((0, 1), 0, 1), ((1, 2), 0, 1), ((0, 2), 0, 1)];
+    let mut got: Vec<_> = triangles(&deltas, 1).into_iter().filter(|(_, _, r)| *r > 0).map(|(d, _, _)| d).collect();
+    got.sort();
+    // Not deduplicated: all three edges arrive at the same time, so every pair of them ties
+    // and the alt/neu discipline is what keeps exactly one delta fragment from producing the
+    // triangle. Deduplicating here would hide a fragment producing it a second time.
+    assert_eq!(got, vec![(0, 1, 2)], "expected the triangle (0,1,2) exactly once; got {:?}", got);
+}
+
+/// A deterministic xorshift, so a failing seed can be replayed.
+fn next(state: &mut u64) -> u64 {
+    *state ^= *state << 13;
+    *state ^= *state >> 7;
+    *state ^= *state << 17;
+    *state
+}
+
+#[test]
+fn wcoj_matches_exhaustive_search_under_churn() {
+    const NODES: u32 = 5;
+    const ROUNDS: usize = 6;
+
+    let universe: Vec<(u32, u32)> = (0..NODES).flat_map(|a| (0..NODES).filter(move |b| *b != a).map(move |b| (a, b))).collect();
+
+    for seed in 1..=20u64 {
+        let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15);
+
+        // Draw a fresh edge set for each round, and record the differences between rounds.
+        // Drawing states rather than differences keeps the collection a set at every time,
+        // which is what these operators are documented to expect, while still exercising
+        // edges that arrive and depart -- including within a single comparison interval.
+        let mut deltas = Vec::new();
+        let mut previous: BTreeSet<(u32, u32)> = BTreeSet::new();
+        let mut states = Vec::new();
+        for round in 0..ROUNDS {
+            let current: BTreeSet<(u32, u32)> = universe.iter().filter(|_| next(&mut state) % 3 != 0).cloned().collect();
+            for edge in current.difference(&previous) { deltas.push((*edge, round, 1)); }
+            for edge in previous.difference(&current) { deltas.push((*edge, round, -1)); }
+            states.push(current.clone());
+            previous = current;
+        }
+
+        let updates = triangles(&deltas, ROUNDS);
+
+        for round in 0..ROUNDS {
+            let expected = triangles_of(&states[round]);
+            let actual = accumulate(&updates, round);
+            assert_eq!(
+                actual, expected,
+                "seed {} round {}: edges {:?}\n  extend gave {:?}\n  search gave {:?}",
+                seed, round, states[round], actual, expected,
+            );
+        }
+    }
+}

From 02b75c0bf0db65b25b0b3f8a0e5cdc5d3e49f3d1 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 7 Aug 2026 15:17:26 -0400
Subject: [PATCH 3/5] Test the WCOJ path over partially ordered times

`wcoj_triangle.rs` cannot see the payload time do any work.
For a totally ordered time the total order used by `comparison` coincides with the lattice order, so every match satisfies `lub(t2, payload) == payload` and all matches collapse onto one output time.

This runs the same triangle query over `Product<usize, usize>` with unordered input, against an exhaustive-search oracle, at every input time and every pairwise join of them.
It adds a constructed case where an extension is born at (0,0) and retracted at (0,2), both preceding a delta at (1,0) in the total order while (0,2) stays incomparable to (1,0) in the lattice.
The extension is therefore live at (1,0) while its updates sum to zero, and a `count` that gates on a non-zero sum drops the prefix and loses the triangle.

Randomized generation does not reach that configuration: `count` is per key, so every extension of a key must be born and die inside the window, which twelve seeds never produced.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 dogsdogsdogs/tests/wcoj_partial_order.rs | 227 +++++++++++++++++++++++
 1 file changed, 227 insertions(+)
 create mode 100644 dogsdogsdogs/tests/wcoj_partial_order.rs

diff --git a/dogsdogsdogs/tests/wcoj_partial_order.rs b/dogsdogsdogs/tests/wcoj_partial_order.rs
new file mode 100644
index 000000000..28b1c6310
--- /dev/null
+++ b/dogsdogsdogs/tests/wcoj_partial_order.rs
@@ -0,0 +1,227 @@
+//! The worst-case-optimal `extend` path over a *partially ordered* time.
+//!
+//! This is the case the payload time exists for. `comparison` orders times totally, so a prefix
+//! at `initial` matches records at times incomparable to it, and `lub(t2, payload)` then differs
+//! from match to match. Two updates to the same extension therefore land at two *different*
+//! output times rather than collapsing onto one, and no longer cancel.
+//!
+//! With a totally ordered time the two coincide and the distinction is invisible, which is why
+//! `wcoj_triangle.rs` cannot see it.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::sync::{Arc, Mutex};
+
+use timely::order::{PartialOrder, Product};
+use timely::dataflow::operators::probe::Handle;
+use timely::dataflow::operators::vec::{Map, UnorderedInput};
+use timely::progress::Antichain;
+use differential_dataflow::AsCollection;
+use differential_dataflow::lattice::Lattice;
+
+use differential_dogs3::{CollectionIndex, altneu::AltNeu, ProposeExtensionMethod};
+
+type Time = Product<usize, usize>;
+
+/// Runs `Q(a,b,c) := E(a,b), E(b,c), E(a,c)` over `deltas`, and returns the output updates.
+fn triangles(deltas: &[((u32, u32), Time, isize)]) -> Vec<((u32, u32, u32), Time, isize)> {
+    let found = Arc::new(Mutex::new(Vec::new()));
+    let found_outer = found.clone();
+    let deltas = deltas.to_vec();
+
+    timely::execute(timely::Config::thread(), move |worker| {
+        let found_outer = found_outer.clone();
+        let deltas = deltas.clone();
+        let mut probe = Handle::new();
+
+        // A `Product` root doesn't refine `()`, so the partially ordered times live one scope in.
+        let (mut input, capability) = worker.dataflow::<usize, _, _>(|scope| {
+            let (handles, output) = scope.scoped::<Time, _, _>("Times", |outer| {
+                let ((input, capability), stream) = outer.new_unordered_input::<((u32, u32), Time, isize)>();
+                let edges = stream.as_collection();
+
+                let forward = edges.clone();
+                let reverse = edges.map(|(x, y)| (y, x));
+
+                let triangles = outer.scoped::<AltNeu<Time>, _, _>("Triangles", |inner| {
+                    let forward = forward.enter(inner);
+                    let reverse = reverse.enter(inner);
+
+                    // Hold compaction back one step in each coordinate, so that total-order
+                    // comparisons are not misled, and compare in the total order on `AltNeu`
+                    // (lexicographic on `(time, neu)`, and `time` lexicographic on `Product`).
+                    let frontier_func = |time: &AltNeu<Time>, antichain: &mut Antichain<AltNeu<Time>>| {
+                        antichain.insert(AltNeu::alt(Product::new(time.time.outer.saturating_sub(1), time.time.inner.saturating_sub(1))));
+                    };
+                    let comparison = |t1: &AltNeu<Time>, t2: &AltNeu<Time>| t1 <= t2;
+
+                    let alt_forward = CollectionIndex::index(forward.clone(), frontier_func, comparison);
+                    let alt_reverse = CollectionIndex::index(reverse.clone(), frontier_func, comparison);
+                    let neu_forward = CollectionIndex::index(forward.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
+                    let neu_reverse = CollectionIndex::index(reverse.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
+
+                    let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
+
+                    //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
+                    let changes1 = deltas.clone()
+                        .extend(&mut [
+                            &mut neu_forward.extend_using(|(_a, b)| *b),
+                            &mut neu_forward.extend_using(|(a, _b)| *a),
+                        ])
+                        .map(|(((a, b), c), payload)| ((a, b, c), payload));
+
+                    //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
+                    let changes2 = deltas.clone()
+                        .extend(&mut [
+                            &mut alt_reverse.extend_using(|(b, _c)| *b),
+                            &mut neu_reverse.extend_using(|(_b, c)| *c),
+                        ])
+                        .map(|(((b, c), a), payload)| ((a, b, c), payload));
+
+                    //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
+                    let changes3 = deltas
+                        .extend(&mut [
+                            &mut alt_forward.extend_using(|(a, _c)| *a),
+                            &mut alt_reverse.extend_using(|(_a, c)| *c),
+                        ])
+                        .map(|(((a, c), b), payload)| ((a, b, c), payload));
+
+                    // Delay updates to the payload time worked out while extending.
+                    changes1.concat(changes2).concat(changes3)
+                        .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
+                        .leave(outer)
+                });
+
+                let left = triangles
+                    .inspect_batch(move |_t, xs| {
+                        let mut v = found_outer.lock().unwrap();
+                        v.extend(xs.iter().cloned());
+                    })
+                    .leave(scope);
+
+                ((input, capability), left)
+            });
+            output.probe_with(&mut probe);
+            handles
+        });
+
+        input.activate().session(&capability).give_iterator(deltas.iter().cloned().map(|(e, t, r)| (e, t, r)));
+        drop(capability);
+        while worker.step() { }
+    })
+    .unwrap();
+
+    let updates = found.lock().unwrap().clone();
+    updates
+}
+
+/// Accumulates `updates` as of `time`, in the *lattice* order, discarding anything that cancels.
+fn accumulate<D: Ord + Clone>(updates: &[(D, Time, isize)], time: &Time) -> BTreeSet<D> {
+    let mut counts = BTreeMap::new();
+    for (data, _, diff) in updates.iter().filter(|(_, when, _)| when.less_equal(time)) {
+        *counts.entry(data.clone()).or_insert(0isize) += diff;
+    }
+    counts.into_iter().filter(|(_, c)| *c != 0).map(|(d, _)| d).collect()
+}
+
+/// The triangles of `edges`, by exhaustive search.
+fn triangles_of(edges: &BTreeSet<(u32, u32)>) -> BTreeSet<(u32, u32, u32)> {
+    let mut found = BTreeSet::new();
+    for &(a, b) in edges.iter() {
+        for &(b2, c) in edges.iter() {
+            if b2 == b && edges.contains(&(a, c)) {
+                found.insert((a, b, c));
+            }
+        }
+    }
+    found
+}
+
+/// A deterministic xorshift, so a failing seed can be replayed.
+fn next(state: &mut u64) -> u64 {
+    *state ^= *state << 13;
+    *state ^= *state >> 7;
+    *state ^= *state << 17;
+    *state
+}
+
+#[test]
+fn wcoj_matches_exhaustive_search_at_incomparable_times() {
+    const NODES: u32 = 4;
+    const COORD: usize = 3;
+
+    let universe: Vec<(u32, u32)> = (0..NODES).flat_map(|a| (0..NODES).filter(move |b| *b != a).map(move |b| (a, b))).collect();
+
+    // Times spread over both coordinates, so that many pairs are incomparable.
+    let times: Vec<Time> = (0..COORD).flat_map(|o| (0..COORD).map(move |i| Product::new(o, i))).collect();
+
+    for seed in 1..=12u64 {
+        let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15);
+
+        // Each edge gets an independent life: a time it arrives, and possibly a later time in
+        // the total order at which it departs. An edge that comes and goes between two times
+        // incomparable to a third is exactly the case a summed count cannot see.
+        let mut deltas = Vec::new();
+        for edge in universe.iter() {
+            if next(&mut state) % 3 == 0 { continue; }
+            let birth = times[(next(&mut state) as usize) % times.len()].clone();
+            deltas.push((*edge, birth.clone(), 1));
+            if next(&mut state) % 2 == 0 {
+                let death = times[(next(&mut state) as usize) % times.len()].clone();
+                // Retract only strictly later in the lattice, to keep the collection a set.
+                if birth.less_than(&death) {
+                    deltas.push((*edge, death, -1));
+                }
+            }
+        }
+
+        let updates = triangles(&deltas);
+
+        // Check at every input time and at every pairwise join of them, which is where the
+        // payload times land.
+        let mut probes: BTreeSet<Time> = deltas.iter().map(|(_, t, _)| t.clone()).collect();
+        for t1 in probes.clone().iter() {
+            for t2 in probes.clone().iter() {
+                probes.insert(t1.join(t2));
+            }
+        }
+
+        for time in probes.iter() {
+            let expected = triangles_of(&accumulate(&deltas, time));
+            let actual = accumulate(&updates, time);
+            assert_eq!(
+                actual, expected,
+                "seed {} time {:?}\n  deltas {:?}\n  extend gave {:?}\n  search gave {:?}",
+                seed, time, deltas, actual, expected,
+            );
+        }
+    }
+}
+
+#[test]
+fn extension_that_lives_and_dies_inside_the_comparison_interval() {
+    // Times: (0,0) < (0,2) < (1,0) in the *total* order, but (0,2) and (1,0) are incomparable
+    // in the lattice. So an edge born at (0,0) and retracted at (0,2) is still present at (1,0),
+    // while both of its updates lie inside the comparison interval of a delta at (1,0).
+    //
+    // Node 0's only in-edge into 2 is (0,2), so the count for that key sums to +1 - 1 = 0 even
+    // though the edge is live at (1,0). A count that gates on a non-zero sum drops the prefix
+    // and loses the triangle.
+    let deltas = vec![
+        ((0, 1), Product::new(0, 0), 1),
+        ((0, 2), Product::new(0, 0), 1),
+        ((0, 2), Product::new(0, 2), -1),
+        ((1, 2), Product::new(1, 0), 1),
+    ];
+
+    let updates = triangles(&deltas);
+
+    for time in [Product::new(0, 0), Product::new(0, 2), Product::new(1, 0), Product::new(1, 2)] {
+        let expected = triangles_of(&accumulate(&deltas, &time));
+        let actual = accumulate(&updates, &time);
+        assert_eq!(
+            actual, expected,
+            "time {:?}: extend gave {:?}, search gave {:?}\n  all updates {:?}",
+            time, actual, expected, updates,
+        );
+    }
+}

From 3a474d0febd20a79503265bdb1dbc2287070e78e Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 7 Aug 2026 15:58:31 -0400
Subject: [PATCH 4/5] State delta-query strictness directly, rather than in the
 timestamp

`count`, `propose` and `validate` take `strict: bool` in place of a comparison closure.
Only two comparisons were ever valid, and which one follows from the query: a relation looking up in a later one is strict, and cannot see updates concurrent with the delta it is responding to; looking up in an earlier one is non-strict, and can.
A relation never looks up in itself.
The closure existed for monomorphization rather than for generality, so the operators branch on `strict` once while building the dataflow and instantiate `half_join` twice, leaving nothing to test at each timestamp.
`half_join` keeps its closure: it is the externally relied-on surface, and it knows nothing about which relations are being joined.

That subsumes `AltNeu`, whose `neu` flag was a stand-in for the same choice.
A delta enters at `alt`, so an arrangement left at `alt` is visible at the delta's own time and one delayed to `neu` is not; with strictness stated directly the tag carries no information.
The four tagged indexes in the worst-case-optimal example and its tests collapse to one per orientation, and the nested scope goes with them.

`delta_query.rs` compared two delta implementations, agreeing only after concatenating all three fragments.
It now compares the concatenated fragments against a conventional three-way join, which shares no machinery with them and so is an independent answer rather than a second opinion from the same method.
That also removes the last use of `Arranged::enter_at` in the repository.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 dogsdogsdogs/examples/delta_query.rs      | 151 ++++++++--------------
 dogsdogsdogs/examples/delta_query_wcoj.rs | 128 +++++++++---------
 dogsdogsdogs/examples/dogsdogsdogs.rs     |   7 +-
 dogsdogsdogs/src/lib.rs                   |  41 +++---
 dogsdogsdogs/src/operators/count.rs       |  25 ++--
 dogsdogsdogs/src/operators/propose.rs     |  25 ++--
 dogsdogsdogs/src/operators/validate.rs    |  25 ++--
 dogsdogsdogs/tests/wcoj_partial_order.rs  |  91 ++++++-------
 dogsdogsdogs/tests/wcoj_triangle.rs       |  87 ++++++-------
 9 files changed, 256 insertions(+), 324 deletions(-)

diff --git a/dogsdogsdogs/examples/delta_query.rs b/dogsdogsdogs/examples/delta_query.rs
index d35b18713..b087bcb8e 100644
--- a/dogsdogsdogs/examples/delta_query.rs
+++ b/dogsdogsdogs/examples/delta_query.rs
@@ -4,8 +4,6 @@ use differential_dataflow::AsCollection;
 use differential_dataflow::input::Input;
 use graph_map::GraphMMap;
 
-use differential_dogs3::altneu::AltNeu;
-use differential_dogs3::calculus::{Differentiate, Integrate};
 
 fn main() {
 
@@ -44,99 +42,62 @@ fn main() {
             // let reverse_count = edges.map(|(x,y)| y).arrange_by_self();
 
             // Q(a,b,c) :=  E1(a,b),  E2(b,c),  E3(a,c)
-            let (triangles_prev, triangles_next) = scope.scoped::<AltNeu<usize>,_,_>("DeltaQuery (Triangles)", |inner| {
-
-                // Grab the stream of changes.
-                let changes = edges.clone().enter(inner);
-
-                // Each relation we'll need.
-                let forward_key_alt = forward_key.clone().enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
-                let reverse_key_alt = reverse_key.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
-                let forward_key_neu = forward_key.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
-                // let reverse_key_neu = reverse_key.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
-
-                // let forward_self_alt = forward_self.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
-                let reverse_self_alt = reverse_self.clone().enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
-                let forward_self_neu = forward_self.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
-                let reverse_self_neu = reverse_self.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
-
-                // For each relation, we form a delta query driven by changes to that relation.
-                //
-                // The sequence of joined relations are such that we only introduce relations
-                // which share some bound attributes with the current stream of deltas.
-                // Each joined relation is delayed { alt -> neu } if its position in the
-                // sequence is greater than the delta stream.
-                // Each joined relation is directed { forward, reverse } by whether the
-                // bound variable occurs in the first or second position.
-
-                let key1 = |x: &(u32, u32)| x.0;
-                let key2 = |x: &(u32, u32)| x.1;
-
-                use differential_dogs3::operators::propose;
-                use differential_dogs3::operators::validate;
-
-                // Hold compaction back one base-time step, so the alt/neu distinction survives,
-                // and compare in the total order on `AltNeu` (lexicographic on `(time, neu)`).
-                let frontier_func = |time: &AltNeu<usize>, antichain: &mut timely::progress::Antichain<AltNeu<usize>>| {
-                    antichain.insert(AltNeu::alt(time.time.saturating_sub(1)));
-                };
-                let comparison = |t1: &AltNeu<usize>, t2: &AltNeu<usize>| t1 <= t2;
-
-                // Stash each delta's own time as its payload, to be advanced by the times of
-                // the records it matches, and delayed to once we leave the delta region.
-                let deltas = changes.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
-
-                // Prior technology
-                //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
-                let changes1 = propose(deltas.clone(), forward_key_neu.clone(), key2.clone(), frontier_func, comparison);
-                let changes1 = validate(changes1, forward_self_neu.clone(), key1.clone(), frontier_func, comparison);
-                let changes1 = changes1.map(|(((a,b),c), payload)| ((a,b,c), payload));
-
-                //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
-                let changes2 = propose(deltas.clone(), reverse_key_alt.clone(), key1.clone(), frontier_func, comparison);
-                let changes2 = validate(changes2, reverse_self_neu.clone(), key2.clone(), frontier_func, comparison);
-                let changes2 = changes2.map(|(((b,c),a), payload)| ((a,b,c), payload));
-
-                //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                let changes3 = propose(deltas, forward_key_alt.clone(), key1.clone(), frontier_func, comparison);
-                let changes3 = validate(changes3, reverse_self_alt.clone(), key2.clone(), frontier_func, comparison);
-                let changes3 = changes3.map(|(((a,c),b), payload)| ((a,b,c), payload));
-
-                // Delay updates to the payload time worked out while extending.
-                let prev_changes = changes1.concat(changes2).concat(changes3)
-                    .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
-                    .leave(scope);
-
-                // New ideas
-                let d_edges = edges.differentiate(inner);
-
-                //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
-                let changes1 =
-                d_edges
-                    .clone()
-                    .map(|(x,y)| (y,x))
-                    .join_core(forward_key_neu, |b,a,c| Some(((*a, *c), *b)))
-                    .join_core(forward_self_neu.clone(), |(a,c), b, &()| Some((*a,*b,*c)));
-
-                //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
-                let changes2 =
-                d_edges
-                    .clone()
-                    .join_core(reverse_key_alt, |b,c,a| Some(((*a, *c), *b)))
-                    .join_core(forward_self_neu, |(a,c), b, &()| Some((*a,*b,*c)));
-
-                //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                let changes3 =
-                d_edges
-                    .join_core(forward_key_alt, |a,c,b| Some(((*c, *b), *a)))
-                    .join_core(reverse_self_alt, |(c,b), a, &()| Some((*a,*b,*c)));
-
-                let next_changes = changes1.concat(changes2).concat(changes3).integrate(scope);
-
-                (prev_changes, next_changes)
-            });
-
-            // Test if our two methods do the same thing.
+            //
+            // For each relation, we form a delta query driven by changes to that relation.
+            //
+            // The sequence of joined relations are such that we only introduce relations
+            // which share some bound attributes with the current stream of deltas.
+            // Each joined relation is directed { forward, reverse } by whether the
+            // bound variable occurs in the first or second position.
+            //
+            // Each lookup is strict exactly when it reaches a relation later in the sequence
+            // than the delta stream, which is what stops a pair of updates matching twice.
+
+            let key1 = |x: &(u32, u32)| x.0;
+            let key2 = |x: &(u32, u32)| x.1;
+
+            use differential_dogs3::operators::propose;
+            use differential_dogs3::operators::validate;
+
+            // Hold compaction back one step, so that we do not lose the distinction between
+            // "strictly before" and "at the same time".
+            let frontier_func = |time: &usize, antichain: &mut timely::progress::Antichain<usize>| {
+                antichain.insert(time.saturating_sub(1));
+            };
+
+            // Stash each delta's own time as its payload, to be advanced by the times of the
+            // records it matches, and delayed to once we are done extending.
+            let deltas = edges.clone().inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
+
+            //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
+            let changes1 = propose(deltas.clone(), forward_key.clone(), key2.clone(), frontier_func, true);
+            let changes1 = validate(changes1, forward_self.clone(), key1.clone(), frontier_func, true);
+            let changes1 = changes1.map(|(((a,b),c), payload)| ((a,b,c), payload));
+
+            //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
+            let changes2 = propose(deltas.clone(), reverse_key.clone(), key1.clone(), frontier_func, false);
+            let changes2 = validate(changes2, reverse_self.clone(), key2.clone(), frontier_func, true);
+            let changes2 = changes2.map(|(((b,c),a), payload)| ((a,b,c), payload));
+
+            //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
+            let changes3 = propose(deltas, forward_key.clone(), key1.clone(), frontier_func, false);
+            let changes3 = validate(changes3, reverse_self.clone(), key2.clone(), frontier_func, false);
+            let changes3 = changes3.map(|(((a,c),b), payload)| ((a,b,c), payload));
+
+            // Delay updates to the payload time worked out while extending.
+            let triangles_prev = changes1.concat(changes2).concat(changes3)
+                .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection();
+
+            // The same query as a conventional three-way join, which shares no machinery with
+            // the delta fragments above and so is an independent answer rather than a second
+            // opinion from the same method.
+            let triangles_next =
+            edges
+                .map(|(x,y)| (y,x))
+                .join_core(forward_key, |b,a,c| Some(((*a, *c), *b)))
+                .join_core(forward_self, |(a,c), b, &()| Some((*a,*b,*c)));
+
+            // Test that the concatenated delta fragments equal the conventional join.
             triangles_prev.clone().assert_eq(triangles_next);
 
             triangles_prev
diff --git a/dogsdogsdogs/examples/delta_query_wcoj.rs b/dogsdogsdogs/examples/delta_query_wcoj.rs
index 6539f8ce6..b751b2959 100644
--- a/dogsdogsdogs/examples/delta_query_wcoj.rs
+++ b/dogsdogsdogs/examples/delta_query_wcoj.rs
@@ -4,7 +4,7 @@ use differential_dataflow::AsCollection;
 use differential_dataflow::input::Input;
 use graph_map::GraphMMap;
 
-use differential_dogs3::{CollectionIndex, altneu::AltNeu};
+use differential_dogs3::CollectionIndex;
 use differential_dogs3::{ProposeExtensionMethod};
 
 fn main() {
@@ -31,75 +31,65 @@ fn main() {
             let forward = edges.clone();
             let reverse = edges.map(|(x,y)| (y,x));
 
+
             // Q(a,b,c) :=  E1(a,b),  E2(b,c),  E3(a,c)
-            let triangles = scope.scoped::<AltNeu<usize>,_,_>("DeltaQuery (Triangles)", |inner| {
-
-                // Each relation we'll need.
-                let forward = forward.enter(inner);
-                let reverse = reverse.enter(inner);
-
-                // Hold compaction back one base-time step, so that the alt/neu distinction
-                // survives, and compare in the total order on `AltNeu` (lexicographic on
-                // `(time, neu)`). Delta streams enter at `alt`, so an `alt` arrangement is
-                // seen at the delta's own time and a `neu` one only strictly after it: the
-                // tag is what makes the comparison strict or not.
-                let frontier_func = |time: &AltNeu<usize>, antichain: &mut timely::progress::Antichain<AltNeu<usize>>| {
-                    antichain.insert(AltNeu::alt(time.time.saturating_sub(1)));
-                };
-                let comparison = |t1: &AltNeu<usize>, t2: &AltNeu<usize>| t1 <= t2;
-
-                // Without using wrappers yet, maintain an "old" and a "new" copy of edges.
-                let alt_forward = CollectionIndex::index(forward.clone(), frontier_func, comparison);
-                let alt_reverse = CollectionIndex::index(reverse.clone(), frontier_func, comparison);
-                let neu_forward = CollectionIndex::index(forward.clone().delay(|time| AltNeu::neu(time.time.clone())), frontier_func, comparison);
-                let neu_reverse = CollectionIndex::index(reverse.clone().delay(|time| AltNeu::neu(time.time.clone())), frontier_func, comparison);
-
-                // Stash each delta's own time as its payload, to be advanced by the times of
-                // the records it matches, and delayed to once we leave the delta region.
-                let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
-
-                // For each relation, we form a delta query driven by changes to that relation.
-                //
-                // The sequence of joined relations are such that we only introduce relations
-                // which share some bound attributes with the current stream of deltas.
-                // Each joined relation is delayed { alt -> neu } if its position in the
-                // sequence is greater than the delta stream.
-                // Each joined relation is directed { forward, reverse } by whether the
-                // bound variable occurs in the first or second position.
-
-                //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
-                let changes1 =
-                deltas
-                    .clone()
-                    .extend(&mut [
-                        &mut neu_forward.extend_using(|(_a,b)| *b),
-                        &mut neu_forward.extend_using(|(a,_b)| *a),
-                    ])
-                    .map(|(((a,b),c), payload)| ((a,b,c), payload));
-
-                //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
-                let changes2 =
-                deltas
-                    .clone()
-                    .extend(&mut [
-                        &mut alt_reverse.extend_using(|(b,_c)| *b),
-                        &mut neu_reverse.extend_using(|(_b,c)| *c),
-                    ])
-                    .map(|(((b,c),a), payload)| ((a,b,c), payload));
-
-                //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                let changes3 = deltas
-                    .extend(&mut [
-                        &mut alt_forward.extend_using(|(a,_c)| *a),
-                        &mut alt_reverse.extend_using(|(_a,c)| *c),
-                    ])
-                    .map(|(((a,c),b), payload)| ((a,b,c), payload));
-
-                // Delay updates to the payload time worked out while extending.
-                changes1.concat(changes2).concat(changes3)
-                    .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
-                    .leave(scope)
-            });
+
+            // Hold compaction back one step, so that we do not lose the distinction between
+            // "strictly before" and "at the same time".
+            let frontier_func = |time: &usize, antichain: &mut timely::progress::Antichain<usize>| {
+                antichain.insert(time.saturating_sub(1));
+            };
+
+            // One index per orientation. The "old" and "new" copies these replace differed
+            // only in whether a lookup could see updates at the delta's own time, which each
+            // `extend_using` below now states for itself.
+            let index_forward = CollectionIndex::index(forward.clone(), frontier_func);
+            let index_reverse = CollectionIndex::index(reverse.clone(), frontier_func);
+
+            // Stash each delta's own time as its payload, to be advanced by the times of the
+            // records it matches, and delayed to once we are done extending.
+            let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
+
+            // For each relation, we form a delta query driven by changes to that relation.
+            //
+            // The sequence of joined relations are such that we only introduce relations
+            // which share some bound attributes with the current stream of deltas.
+            // Each lookup is strict exactly when it reaches a relation later in the sequence
+            // than the delta stream, which is what stops a pair of updates matching twice.
+            // Each joined relation is directed { forward, reverse } by whether the
+            // bound variable occurs in the first or second position.
+
+            //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
+            let changes1 =
+            deltas
+                .clone()
+                .extend(&mut [
+                    &mut index_forward.extend_using(|(_a,b)| *b, true),
+                    &mut index_forward.extend_using(|(a,_b)| *a, true),
+                ])
+                .map(|(((a,b),c), payload)| ((a,b,c), payload));
+
+            //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
+            let changes2 =
+            deltas
+                .clone()
+                .extend(&mut [
+                    &mut index_reverse.extend_using(|(b,_c)| *b, false),
+                    &mut index_reverse.extend_using(|(_b,c)| *c, true),
+                ])
+                .map(|(((b,c),a), payload)| ((a,b,c), payload));
+
+            //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
+            let changes3 = deltas
+                .extend(&mut [
+                    &mut index_forward.extend_using(|(a,_c)| *a, false),
+                    &mut index_reverse.extend_using(|(_a,c)| *c, false),
+                ])
+                .map(|(((a,c),b), payload)| ((a,b,c), payload));
+
+            // Delay updates to the payload time worked out while extending.
+            let triangles = changes1.concat(changes2).concat(changes3)
+                .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection();
 
             triangles
                 .filter(move |_| inspect)
diff --git a/dogsdogsdogs/examples/dogsdogsdogs.rs b/dogsdogsdogs/examples/dogsdogsdogs.rs
index 58822152b..49da755a2 100644
--- a/dogsdogsdogs/examples/dogsdogsdogs.rs
+++ b/dogsdogsdogs/examples/dogsdogsdogs.rs
@@ -42,12 +42,11 @@ fn main() {
             let frontier_func = |time: &usize, antichain: &mut timely::progress::Antichain<usize>| {
                 antichain.insert(time.saturating_sub(1));
             };
-            let comparison = |t1: &usize, t2: &usize| t1 <= t2;
 
-            let index = CollectionIndex::index(Collection::new(edges.to_stream(scope)), frontier_func, comparison);
+            let index = CollectionIndex::index(Collection::new(edges.to_stream(scope)), frontier_func);
 
-            let mut index_xz = index.extend_using(|&(ref x, ref _y)| *x);
-            let mut index_yz = index.extend_using(|&(ref _x, ref y)| *y);
+            let mut index_xz = index.extend_using(|&(ref x, ref _y)| *x, false);
+            let mut index_yz = index.extend_using(|&(ref _x, ref y)| *y, false);
 
             let (edges_input, edges) = scope.new_collection();
 
diff --git a/dogsdogsdogs/src/lib.rs b/dogsdogsdogs/src/lib.rs
index afedd39cd..089c3cd18 100644
--- a/dogsdogsdogs/src/lib.rs
+++ b/dogsdogsdogs/src/lib.rs
@@ -50,12 +50,6 @@ pub mod operators;
 /// rests on. See [`crate::operators::half_join`].
 pub type FrontierFunc<T> = Rc<dyn Fn(&T, &mut Antichain<T>)>;
 
-/// Compares an arranged record's time against a prefix's own time, in the *total order*.
-///
-/// The two useful choices are strictly-less and less-or-equal, which is how a delta query
-/// decides whether a stage sees updates concurrent with the delta it is responding to.
-pub type Comparison<T> = Rc<dyn Fn(&T, &T) -> bool>;
-
 /// A type capable of extending a stream of prefixes.
 ///
 /**
@@ -167,9 +161,6 @@ where
 
     /// Holds back compaction; see [`FrontierFunc`].
     frontier_func: FrontierFunc<T>,
-
-    /// Decides which arranged times a prefix sees; see [`Comparison`].
-    comparison: Comparison<T>,
 }
 
 impl<'scope, K, V, T> Clone for CollectionIndex<'scope, K, V, T>
@@ -184,7 +175,6 @@ where
             propose_trace: self.propose_trace.clone(),
             validate_trace: self.validate_trace.clone(),
             frontier_func: Rc::clone(&self.frontier_func),
-            comparison: Rc::clone(&self.comparison),
         }
     }
 }
@@ -196,10 +186,9 @@ where
     T: Lattice+ExchangeData+Timestamp,
 {
 
-    pub fn index<FF, CF>(collection: VecCollection<'scope, T, (K, V), isize>, frontier_func: FF, comparison: CF) -> Self
+    pub fn index<FF>(collection: VecCollection<'scope, T, (K, V), isize>, frontier_func: FF) -> Self
     where
         FF: Fn(&T, &mut Antichain<T>) + 'static,
-        CF: Fn(&T, &T) -> bool + 'static,
     {
         // We need to count the number of (k, v) pairs and not rely on the given Monoid R and its binary addition operation.
         // counts and validate can share the base arrangement
@@ -219,14 +208,20 @@ where
             propose_trace: propose,
             validate_trace: validate,
             frontier_func: Rc::new(frontier_func),
-            comparison: Rc::new(comparison),
         }
     }
-    pub fn extend_using<P, F: Fn(&P)->K+Clone>(&self, logic: F) -> CollectionExtender<'scope, K, V, T, P, F> {
+    /// Prepares to extend prefixes by this relation, using `logic` to find the key.
+    ///
+    /// For a delta query, `strict` follows from the positions of the two relations: a relation
+    /// looking up in a *later* one is strict, and cannot see updates concurrent with the delta
+    /// it is responding to; looking up in an *earlier* one is non-strict, and can. A relation
+    /// never looks up in itself.
+    pub fn extend_using<P, F: Fn(&P)->K+Clone>(&self, logic: F, strict: bool) -> CollectionExtender<'scope, K, V, T, P, F> {
         CollectionExtender {
             phantom: std::marker::PhantomData,
             indices: self.clone(),
             key_selector: logic,
+            strict,
         }
     }
 }
@@ -241,6 +236,7 @@ where
     phantom: std::marker::PhantomData<P>,
     indices: CollectionIndex<'scope, K, V, T>,
     key_selector: F,
+    strict: bool,
 }
 
 impl<'scope, K, V, T, P, F> CollectionExtender<'scope, K, V, T, P, F>
@@ -250,11 +246,10 @@ where
     T: Lattice+ExchangeData+Timestamp,
     F: Fn(&P)->K+Clone,
 {
-    /// The index's time closures, as plain callables the operators can accept.
-    fn time_logic(&self) -> (impl Fn(&T, &mut Antichain<T>) + 'static, impl Fn(&T, &T) -> bool + 'static) {
+    /// The index's compaction closure, as a plain callable the operators can accept.
+    fn frontier_func(&self) -> impl Fn(&T, &mut Antichain<T>) + 'static {
         let frontier_func = Rc::clone(&self.indices.frontier_func);
-        let comparison = Rc::clone(&self.indices.comparison);
-        (move |t: &T, a: &mut Antichain<T>| frontier_func(t, a), move |t1: &T, t2: &T| comparison(t1, t2))
+        move |t: &T, a: &mut Antichain<T>| frontier_func(t, a)
     }
 }
 
@@ -270,18 +265,14 @@ where
     type Extension = V;
 
     fn count(&mut self, prefixes: VecCollection<'scope, T, ((P, usize, usize), T), isize>, index: usize) -> VecCollection<'scope, T, ((P, usize, usize), T), isize> {
-        let (frontier_func, comparison) = self.time_logic();
-        operators::count::count(prefixes, self.indices.count_trace.clone(), self.key_selector.clone(), index, frontier_func, comparison)
+        operators::count::count(prefixes, self.indices.count_trace.clone(), self.key_selector.clone(), index, self.frontier_func(), self.strict)
     }
 
     fn propose(&mut self, prefixes: VecCollection<'scope, T, (P, T), isize>) -> VecCollection<'scope, T, ((P, V), T), isize> {
-        let (frontier_func, comparison) = self.time_logic();
-        operators::propose::propose(prefixes, self.indices.propose_trace.clone(), self.key_selector.clone(), frontier_func, comparison)
+        operators::propose::propose(prefixes, self.indices.propose_trace.clone(), self.key_selector.clone(), self.frontier_func(), self.strict)
     }
 
     fn validate(&mut self, extensions: VecCollection<'scope, T, ((P, V), T), isize>) -> VecCollection<'scope, T, ((P, V), T), isize> {
-        let (frontier_func, comparison) = self.time_logic();
-        let key_selector = self.key_selector.clone();
-        operators::validate::validate(extensions, self.indices.validate_trace.clone(), key_selector, frontier_func, comparison)
+        operators::validate::validate(extensions, self.indices.validate_trace.clone(), self.key_selector.clone(), self.frontier_func(), self.strict)
     }
 }
diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs
index 44756dd67..8aa0eebc7 100644
--- a/dogsdogsdogs/src/operators/count.rs
+++ b/dogsdogsdogs/src/operators/count.rs
@@ -16,13 +16,13 @@ use differential_dataflow::trace::implementations::BatchContainer;
 ///
 /// A prefix is dropped only when its key is absent from `arrangement` entirely, which is only
 /// expected to happen when there have never been counts (they are meant to be non-negative).
-pub fn count<'scope, Tr, K, F, P, R, FF, CF>(
+pub fn count<'scope, Tr, K, F, P, R, FF>(
     prefixes: VecCollection<'scope, Tr::Time, ((P, usize, usize), Tr::Time), R>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
     index: usize,
     frontier_func: FF,
-    comparison: CF,
+    strict: bool,
 ) -> VecCollection<'scope, Tr::Time, ((P, usize, usize), Tr::Time), R>
 where
     Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
@@ -33,7 +33,7 @@ where
     F: Fn(&P)->K+'static,
     P: ExchangeData,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
+    for<'a, 'b> BatchTimeGat<'a, Tr>: PartialOrd<&'b Tr::Time>,
 {
     // The payload time is carried in the record as well as in the half-join's own payload
     // slot, because the output closure is handed the joined times rather than the payload.
@@ -63,13 +63,16 @@ where
         builder.push_into(((triple, payload.clone()), initial.clone(), diff1.clone()));
     };
 
-    crate::operators::half_join::half_join_internal_unsafe::<_, _, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
-        requests,
-        arrangement,
-        frontier_func,
-        comparison,
-        |_timer, _count| false,
-        output_func,
-    )
+    use crate::operators::half_join::half_join_internal_unsafe as half_join_unsafe;
+    // Branch once here, so that each comparison monomorphizes rather than testing `strict` at
+    // every timestamp. The cost is instantiating `half_join` twice.
+    if strict {
+        half_join_unsafe::<_, _, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
+            requests, arrangement, frontier_func, |t1, t2| t1 < t2, |_timer, _count| false, output_func)
+    }
+    else {
+        half_join_unsafe::<_, _, _, _, _, _, _, _, Output<P, Tr::Time, R>>(
+            requests, arrangement, frontier_func, |t1, t2| t1 <= t2, |_timer, _count| false, output_func)
+    }
     .as_collection()
 }
diff --git a/dogsdogsdogs/src/operators/propose.rs b/dogsdogsdogs/src/operators/propose.rs
index 008576184..f6db7c1a6 100644
--- a/dogsdogsdogs/src/operators/propose.rs
+++ b/dogsdogsdogs/src/operators/propose.rs
@@ -12,19 +12,19 @@ use differential_dataflow::trace::implementations::BatchContainer;
 ///
 /// This operator matches streamed updates with arranged updates, and pairs the streamed updates
 /// with arranged updates whose times are less or equal under the *total order* on timestamps.
-/// This inequality is allowed to either be strict or non-strict, as determined by `comparison`.
+/// This inequality is allowed to either be strict or non-strict, as determined by `strict`.
 /// The total order allows the caller to ensure that each pair of updates match exactly once.
 /// The streamed updates also carry a time as data, and that time is advanced (by lattice join)
 /// by the time of the arranged update. The time of the streamed update cannot be advanced, as
 /// it needs to stay put to ensure the total order math works out.
 ///
 /// The arrangement is expected to hold a *set*: see the note on set semantics in [`crate`].
-pub fn propose<'scope, Tr, K, F, P, V, R, FF, CF>(
+pub fn propose<'scope, Tr, K, F, P, V, R, FF>(
     prefixes: VecCollection<'scope, Tr::Time, (P, Tr::Time), R>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
     frontier_func: FF,
-    comparison: CF,
+    strict: bool,
 ) -> VecCollection<'scope, Tr::Time, ((P, V), Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
 where
     Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
@@ -36,14 +36,17 @@ where
     P: ExchangeData,
     V: Clone + 'static,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
+    for<'a, 'b> BatchTimeGat<'a, Tr>: PartialOrd<&'b Tr::Time>,
 {
     let requests = prefixes.map(move |(prefix, payload)| (key_selector(&prefix), prefix, payload));
-    crate::operators::half_join(
-        requests,
-        arrangement,
-        frontier_func,
-        comparison,
-        |_key, prefix, value| (prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)),
-    )
+    // Branch once here, so that each comparison monomorphizes rather than testing `strict` at
+    // every timestamp. The cost is instantiating `half_join` twice.
+    if strict {
+        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 < t2,
+            |_key, prefix, value| (prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)))
+    }
+    else {
+        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 <= t2,
+            |_key, prefix, value| (prefix.clone(), <BatchCursor<Tr> as Cursor>::owned_val(value)))
+    }
 }
diff --git a/dogsdogsdogs/src/operators/validate.rs b/dogsdogsdogs/src/operators/validate.rs
index bb5f40273..5e61f942f 100644
--- a/dogsdogsdogs/src/operators/validate.rs
+++ b/dogsdogsdogs/src/operators/validate.rs
@@ -12,19 +12,19 @@ use differential_dataflow::trace::implementations::BatchContainer;
 ///
 /// This operator matches streamed updates with arranged updates, and pairs the streamed updates
 /// with arranged updates whose times are less or equal under the *total order* on timestamps.
-/// This inequality is allowed to either be strict or non-strict, as determined by `comparison`.
+/// This inequality is allowed to either be strict or non-strict, as determined by `strict`.
 /// The total order allows the caller to ensure that each pair of updates match exactly once.
 /// The streamed updates also carry a time as data, and that time is advanced (by lattice join)
 /// by the time of the arranged update. The time of the streamed update cannot be advanced, as
 /// it needs to stay put to ensure the total order math works out.
 ///
 /// The arrangement is expected to hold a *set*: see the note on set semantics in [`crate`].
-pub fn validate<'scope, Tr, K, V, F, P, R, FF, CF>(
+pub fn validate<'scope, Tr, K, V, F, P, R, FF>(
     extensions: VecCollection<'scope, Tr::Time, ((P, V), Tr::Time), R>,
     arrangement: Arranged<'scope, Tr>,
     key_selector: F,
     frontier_func: FF,
-    comparison: CF,
+    strict: bool,
 ) -> VecCollection<'scope, Tr::Time, ((P, V), Tr::Time), <R as Mul<BatchDiff<Tr>>>::Output>
 where
     Tr: TraceReader<Batch: Navigable, Time: std::hash::Hash>+Clone+'static,
@@ -36,16 +36,19 @@ where
     F: Fn(&P)->K+'static,
     P: ExchangeData,
     FF: Fn(&Tr::Time, &mut Antichain<Tr::Time>) + 'static,
-    CF: Fn(BatchTimeGat<'_, Tr>, &Tr::Time) -> bool + 'static,
+    for<'a, 'b> BatchTimeGat<'a, Tr>: PartialOrd<&'b Tr::Time>,
 {
     let requests = extensions.map(move |((prefix, extension), payload)| {
         ((key_selector(&prefix), extension.clone()), (prefix, extension), payload)
     });
-    crate::operators::half_join(
-        requests,
-        arrangement,
-        frontier_func,
-        comparison,
-        |_key, extended, _value| extended.clone(),
-    )
+    // Branch once here, so that each comparison monomorphizes rather than testing `strict` at
+    // every timestamp. The cost is instantiating `half_join` twice.
+    if strict {
+        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 < t2,
+            |_key, extended, _value| extended.clone())
+    }
+    else {
+        crate::operators::half_join(requests, arrangement, frontier_func, |t1, t2| t1 <= t2,
+            |_key, extended, _value| extended.clone())
+    }
 }
diff --git a/dogsdogsdogs/tests/wcoj_partial_order.rs b/dogsdogsdogs/tests/wcoj_partial_order.rs
index 28b1c6310..5a001854d 100644
--- a/dogsdogsdogs/tests/wcoj_partial_order.rs
+++ b/dogsdogsdogs/tests/wcoj_partial_order.rs
@@ -1,6 +1,6 @@
 //! The worst-case-optimal `extend` path over a *partially ordered* time.
 //!
-//! This is the case the payload time exists for. `comparison` orders times totally, so a prefix
+//! This is the case the payload time exists for. Times are compared totally, so a prefix
 //! at `initial` matches records at times incomparable to it, and `lub(t2, payload)` then differs
 //! from match to match. Two updates to the same extension therefore land at two *different*
 //! output times rather than collapsing onto one, and no longer cancel.
@@ -18,7 +18,7 @@ use timely::progress::Antichain;
 use differential_dataflow::AsCollection;
 use differential_dataflow::lattice::Lattice;
 
-use differential_dogs3::{CollectionIndex, altneu::AltNeu, ProposeExtensionMethod};
+use differential_dogs3::{CollectionIndex, ProposeExtensionMethod};
 
 type Time = Product<usize, usize>;
 
@@ -42,54 +42,45 @@ fn triangles(deltas: &[((u32, u32), Time, isize)]) -> Vec<((u32, u32, u32), Time
                 let forward = edges.clone();
                 let reverse = edges.map(|(x, y)| (y, x));
 
-                let triangles = outer.scoped::<AltNeu<Time>, _, _>("Triangles", |inner| {
-                    let forward = forward.enter(inner);
-                    let reverse = reverse.enter(inner);
-
-                    // Hold compaction back one step in each coordinate, so that total-order
-                    // comparisons are not misled, and compare in the total order on `AltNeu`
-                    // (lexicographic on `(time, neu)`, and `time` lexicographic on `Product`).
-                    let frontier_func = |time: &AltNeu<Time>, antichain: &mut Antichain<AltNeu<Time>>| {
-                        antichain.insert(AltNeu::alt(Product::new(time.time.outer.saturating_sub(1), time.time.inner.saturating_sub(1))));
-                    };
-                    let comparison = |t1: &AltNeu<Time>, t2: &AltNeu<Time>| t1 <= t2;
-
-                    let alt_forward = CollectionIndex::index(forward.clone(), frontier_func, comparison);
-                    let alt_reverse = CollectionIndex::index(reverse.clone(), frontier_func, comparison);
-                    let neu_forward = CollectionIndex::index(forward.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
-                    let neu_reverse = CollectionIndex::index(reverse.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
-
-                    let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
-
-                    //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
-                    let changes1 = deltas.clone()
-                        .extend(&mut [
-                            &mut neu_forward.extend_using(|(_a, b)| *b),
-                            &mut neu_forward.extend_using(|(a, _b)| *a),
-                        ])
-                        .map(|(((a, b), c), payload)| ((a, b, c), payload));
-
-                    //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
-                    let changes2 = deltas.clone()
-                        .extend(&mut [
-                            &mut alt_reverse.extend_using(|(b, _c)| *b),
-                            &mut neu_reverse.extend_using(|(_b, c)| *c),
-                        ])
-                        .map(|(((b, c), a), payload)| ((a, b, c), payload));
-
-                    //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                    let changes3 = deltas
-                        .extend(&mut [
-                            &mut alt_forward.extend_using(|(a, _c)| *a),
-                            &mut alt_reverse.extend_using(|(_a, c)| *c),
-                        ])
-                        .map(|(((a, c), b), payload)| ((a, b, c), payload));
-
-                    // Delay updates to the payload time worked out while extending.
-                    changes1.concat(changes2).concat(changes3)
-                        .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
-                        .leave(outer)
-                });
+                // Hold compaction back one step in each coordinate, so that total-order
+                // comparisons are not misled. Times compare in the total order on `Product`,
+                // which is lexicographic and therefore a linear extension of the lattice.
+                let frontier_func = |time: &Time, antichain: &mut Antichain<Time>| {
+                    antichain.insert(Product::new(time.outer.saturating_sub(1), time.inner.saturating_sub(1)));
+                };
+
+                let index_forward = CollectionIndex::index(forward.clone(), frontier_func);
+                let index_reverse = CollectionIndex::index(reverse.clone(), frontier_func);
+
+                let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
+
+                //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
+                let changes1 = deltas.clone()
+                    .extend(&mut [
+                        &mut index_forward.extend_using(|(_a, b)| *b, true),
+                        &mut index_forward.extend_using(|(a, _b)| *a, true),
+                    ])
+                    .map(|(((a, b), c), payload)| ((a, b, c), payload));
+
+                //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
+                let changes2 = deltas.clone()
+                    .extend(&mut [
+                        &mut index_reverse.extend_using(|(b, _c)| *b, false),
+                        &mut index_reverse.extend_using(|(_b, c)| *c, true),
+                    ])
+                    .map(|(((b, c), a), payload)| ((a, b, c), payload));
+
+                //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
+                let changes3 = deltas
+                    .extend(&mut [
+                        &mut index_forward.extend_using(|(a, _c)| *a, false),
+                        &mut index_reverse.extend_using(|(_a, c)| *c, false),
+                    ])
+                    .map(|(((a, c), b), payload)| ((a, b, c), payload));
+
+                // Delay updates to the payload time worked out while extending.
+                let triangles = changes1.concat(changes2).concat(changes3)
+                    .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection();
 
                 let left = triangles
                     .inspect_batch(move |_t, xs| {
diff --git a/dogsdogsdogs/tests/wcoj_triangle.rs b/dogsdogsdogs/tests/wcoj_triangle.rs
index 8e84f2cea..ee5f61d0c 100644
--- a/dogsdogsdogs/tests/wcoj_triangle.rs
+++ b/dogsdogsdogs/tests/wcoj_triangle.rs
@@ -14,7 +14,7 @@ use timely::progress::Antichain;
 use differential_dataflow::AsCollection;
 use differential_dataflow::input::Input;
 
-use differential_dogs3::{CollectionIndex, altneu::AltNeu, ProposeExtensionMethod};
+use differential_dogs3::{CollectionIndex, ProposeExtensionMethod};
 
 /// Runs `Q(a,b,c) := E(a,b), E(b,c), E(a,c)` over `deltas`, and returns the output updates.
 ///
@@ -35,53 +35,44 @@ fn triangles(deltas: &[((u32, u32), usize, isize)], rounds: usize) -> Vec<((u32,
             let forward = edges.clone();
             let reverse = edges.map(|(x, y)| (y, x));
 
-            let triangles = scope.scoped::<AltNeu<usize>, _, _>("Triangles", |inner| {
-                let forward = forward.enter(inner);
-                let reverse = reverse.enter(inner);
-
-                // Hold compaction back one base-time step so the alt/neu distinction survives,
-                // and compare in the total order on `AltNeu` (lexicographic on `(time, neu)`).
-                let frontier_func = |time: &AltNeu<usize>, antichain: &mut Antichain<AltNeu<usize>>| {
-                    antichain.insert(AltNeu::alt(time.time.saturating_sub(1)));
-                };
-                let comparison = |t1: &AltNeu<usize>, t2: &AltNeu<usize>| t1 <= t2;
-
-                let alt_forward = CollectionIndex::index(forward.clone(), frontier_func, comparison);
-                let alt_reverse = CollectionIndex::index(reverse.clone(), frontier_func, comparison);
-                let neu_forward = CollectionIndex::index(forward.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
-                let neu_reverse = CollectionIndex::index(reverse.clone().delay(|t| AltNeu::neu(t.time.clone())), frontier_func, comparison);
-
-                let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
-
-                //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
-                let changes1 = deltas.clone()
-                    .extend(&mut [
-                        &mut neu_forward.extend_using(|(_a, b)| *b),
-                        &mut neu_forward.extend_using(|(a, _b)| *a),
-                    ])
-                    .map(|(((a, b), c), payload)| ((a, b, c), payload));
-
-                //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
-                let changes2 = deltas.clone()
-                    .extend(&mut [
-                        &mut alt_reverse.extend_using(|(b, _c)| *b),
-                        &mut neu_reverse.extend_using(|(_b, c)| *c),
-                    ])
-                    .map(|(((b, c), a), payload)| ((a, b, c), payload));
-
-                //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
-                let changes3 = deltas
-                    .extend(&mut [
-                        &mut alt_forward.extend_using(|(a, _c)| *a),
-                        &mut alt_reverse.extend_using(|(_a, c)| *c),
-                    ])
-                    .map(|(((a, c), b), payload)| ((a, b, c), payload));
-
-                // Delay updates to the payload time worked out while extending.
-                changes1.concat(changes2).concat(changes3)
-                    .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection()
-                    .leave(scope)
-            });
+            // Hold compaction back one step, so that we do not lose the distinction between
+            // "strictly before" and "at the same time".
+            let frontier_func = |time: &usize, antichain: &mut Antichain<usize>| {
+                antichain.insert(time.saturating_sub(1));
+            };
+
+            let index_forward = CollectionIndex::index(forward.clone(), frontier_func);
+            let index_reverse = CollectionIndex::index(reverse.clone(), frontier_func);
+
+            let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();
+
+            //   dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
+            let changes1 = deltas.clone()
+                .extend(&mut [
+                    &mut index_forward.extend_using(|(_a, b)| *b, true),
+                    &mut index_forward.extend_using(|(a, _b)| *a, true),
+                ])
+                .map(|(((a, b), c), payload)| ((a, b, c), payload));
+
+            //   dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
+            let changes2 = deltas.clone()
+                .extend(&mut [
+                    &mut index_reverse.extend_using(|(b, _c)| *b, false),
+                    &mut index_reverse.extend_using(|(_b, c)| *c, true),
+                ])
+                .map(|(((b, c), a), payload)| ((a, b, c), payload));
+
+            //   dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
+            let changes3 = deltas
+                .extend(&mut [
+                    &mut index_forward.extend_using(|(a, _c)| *a, false),
+                    &mut index_reverse.extend_using(|(_a, c)| *c, false),
+                ])
+                .map(|(((a, c), b), payload)| ((a, b, c), payload));
+
+            // Delay updates to the payload time worked out while extending.
+            let triangles = changes1.concat(changes2).concat(changes3)
+                .inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection();
 
             triangles
                 .inspect_batch(move |_t, xs| {

From 4cb5ddc0e390c871b711b8bbb38f29162f32fac7 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 7 Aug 2026 15:59:39 -0400
Subject: [PATCH 5/5] Delete altneu and calculus, which no longer have
 consumers

Both modules are gone from the worst-case-optimal path, which now states strictness at each lookup rather than encoding it in the timestamp.

`calculus` cannot outlive `altneu`.
`Differentiate` is defined by emitting a change at `alt` and retracting it at `neu`, and `Integrate` recovers the collection by filtering on `!t.neu`, so the tag is its representation rather than an implementation detail.

This removes two public modules from the crate.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 dogsdogsdogs/src/altneu.rs   | 100 -----------------------------------
 dogsdogsdogs/src/calculus.rs |  67 -----------------------
 dogsdogsdogs/src/lib.rs      |   2 -
 3 files changed, 169 deletions(-)
 delete mode 100644 dogsdogsdogs/src/altneu.rs
 delete mode 100644 dogsdogsdogs/src/calculus.rs

diff --git a/dogsdogsdogs/src/altneu.rs b/dogsdogsdogs/src/altneu.rs
deleted file mode 100644
index 0ddb9572a..000000000
--- a/dogsdogsdogs/src/altneu.rs
+++ /dev/null
@@ -1,100 +0,0 @@
-//! A lexicographically ordered pair of timestamps.
-//!
-//! Two timestamps (s1, t1) and (s2, t2) are ordered either if
-//! s1 and s2 are ordered, or if s1 equals s2 and t1 and t2 are
-//! ordered.
-//!
-//! The join of two timestamps should have as its first coordinate
-//! the join of the first coordinates, and for its second coordinate
-//! the join of the second coordinates for elements whose first
-//! coordinate equals the computed join. That may be the minimum
-//! element of the second lattice, if neither first element equals
-//! the join.
-
-use serde::{Deserialize, Serialize};
-
-/// A pair of timestamps, partially ordered by the product order.
-#[derive(Debug, Hash, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
-pub struct AltNeu<T> {
-    pub time: T,
-    pub neu: bool,  // alt < neu in timestamp comparisons.
-}
-
-impl<T> AltNeu<T> {
-    pub fn alt(time: T) -> Self { AltNeu { time, neu: false } }
-    pub fn neu(time: T) -> Self { AltNeu { time, neu: true } }
-}
-
-// Implement timely dataflow's `PartialOrder` trait.
-use timely::order::PartialOrder;
-impl<T: PartialOrder> PartialOrder for AltNeu<T> {
-    fn less_equal(&self, other: &Self) -> bool {
-        if self.time.eq(&other.time) {
-            self.neu <= other.neu
-        }
-        else {
-            self.time.less_equal(&other.time)
-        }
-    }
-}
-
-// Implement timely dataflow's `PathSummary` trait.
-// This is preparation for the `Timestamp` implementation below.
-use timely::progress::PathSummary;
-impl<T: Timestamp> PathSummary<AltNeu<T>> for () {
-    fn results_in(&self, timestamp: &AltNeu<T>) -> Option<AltNeu<T>> {
-        Some(timestamp.clone())
-    }
-    fn followed_by(&self, other: &Self) -> Option<Self> {
-        Some(other.clone())
-    }
-}
-
-// Implement timely dataflow's `Timestamp` trait.
-use timely::progress::Timestamp;
-impl<T: Timestamp> Timestamp for AltNeu<T> {
-    type Summary = ();
-    fn minimum() -> Self { AltNeu::alt(T::minimum()) }
-}
-
-use timely::progress::timestamp::Refines;
-
-impl<T: Timestamp> Refines<T> for AltNeu<T> {
-    fn to_inner(other: T) -> Self {
-        AltNeu::alt(other)
-    }
-    fn to_outer(self: AltNeu<T>) -> T {
-        self.time
-    }
-    fn summarize(_path: ()) -> T::Summary {
-        Default::default()
-    }
-}
-
-// Implement differential dataflow's `Lattice` trait.
-// This extends the `PartialOrder` implementation with additional structure.
-use differential_dataflow::lattice::Lattice;
-impl<T: Lattice> Lattice for AltNeu<T> {
-    fn join(&self, other: &Self) -> Self {
-        let time = self.time.join(&other.time);
-        let mut neu = false;
-        if time == self.time {
-            neu = neu || self.neu;
-        }
-        if time == other.time {
-            neu = neu || other.neu;
-        }
-        AltNeu { time, neu }
-    }
-    fn meet(&self, other: &Self) -> Self {
-        let time = self.time.meet(&other.time);
-        let mut neu = true;
-        if time == self.time {
-            neu = neu && self.neu;
-        }
-        if time == other.time {
-            neu = neu && other.neu;
-        }
-        AltNeu { time, neu }
-    }
-}
diff --git a/dogsdogsdogs/src/calculus.rs b/dogsdogsdogs/src/calculus.rs
deleted file mode 100644
index f7697e1e6..000000000
--- a/dogsdogsdogs/src/calculus.rs
+++ /dev/null
@@ -1,67 +0,0 @@
-//! Traits and implementations for differentiating and integrating collections.
-//!
-//! The `Differentiate` and `Integrate` traits allow us to move between standard differential
-//! dataflow collections, and collections that describe their instantaneous change. The first
-//! trait converts a collection to one that contains each change at the moment it occurs, but
-//! then immediately retracting it. The second trait takes such a representation are recreates
-//! the collection from its instantaneous changes.
-//!
-//! These two traits together allow us to build dataflows that maintain computates over inputs
-//! that are the instantaneous changes, and then to reconstruct collections from them. The most
-//! clear use case for this are "delta query" implementations of relational joins, where linearity
-//! allows us to write dataflows based on instantaneous changes, whose "accumluated state" is
-//! almost everywhere empty (and so has a low memory footprint, if the system works as planned).
-
-use timely::dataflow::Scope;
-use timely::progress::Timestamp;
-use timely::dataflow::operators::vec::{Filter, Map};
-use differential_dataflow::{AsCollection, VecCollection, Data};
-use differential_dataflow::difference::Abelian;
-
-use crate::altneu::AltNeu;
-
-/// Produce a collection containing the changes at the moments they happen.
-pub trait Differentiate<'scope, T: Timestamp, D: Data, R: Abelian> {
-    fn differentiate<'inner>(self, child: Scope<'inner, AltNeu<T>>) -> VecCollection<'inner, AltNeu<T>, D, R>;
-}
-
-/// Collect instantaneous changes back in to a collection.
-pub trait Integrate<'scope, T: Timestamp, D: Data, R: Abelian> {
-    fn integrate<'outer>(self, outer: Scope<'outer, T>) -> VecCollection<'outer, T, D, R>;
-}
-
-impl<'scope, T, D, R> Differentiate<'scope, T, D, R> for VecCollection<'scope, T, D, R>
-where
-    T: Timestamp,
-    D: Data,
-    R: Abelian + 'static,
-{
-    // For each (data, Alt(time), diff) we add a (data, Neu(time), -diff).
-    fn differentiate<'inner>(self, child: Scope<'inner, AltNeu<T>>) -> VecCollection<'inner, AltNeu<T>, D, R> {
-        self.enter(child)
-            .inner
-            .flat_map(|(data, time, diff)| {
-                let mut neg_diff = diff.clone();
-                neg_diff.negate();
-                let neu = (data.clone(), AltNeu::neu(time.time.clone()), neg_diff);
-                let alt = (data, time, diff);
-                Some(alt).into_iter().chain(Some(neu))
-            })
-            .as_collection()
-    }
-}
-
-impl<'scope, T, D, R> Integrate<'scope, T, D, R> for VecCollection<'scope, AltNeu<T>, D, R>
-where
-    T: Timestamp,
-    D: Data,
-    R: Abelian + 'static,
-{
-    // We discard each `neu` variant and strip off the `alt` wrapper.
-    fn integrate<'outer>(self, outer: Scope<'outer, T>) -> VecCollection<'outer, T, D, R> {
-        self.inner
-            .filter(|(_d,t,_r)| !t.neu)
-            .as_collection()
-            .leave(outer)
-    }
-}
diff --git a/dogsdogsdogs/src/lib.rs b/dogsdogsdogs/src/lib.rs
index 089c3cd18..6736f8df5 100644
--- a/dogsdogsdogs/src/lib.rs
+++ b/dogsdogsdogs/src/lib.rs
@@ -39,8 +39,6 @@ use differential_dataflow::{ExchangeData, VecCollection, AsCollection};
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::TraceAgent;
 
-pub mod altneu;
-pub mod calculus;
 pub mod operators;
 
 /// Holds back logical compaction so that total-order time comparisons stay meaningful.