From 54f47ce5fe008b59bdfe2aa14628c592d6354077 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 4 Aug 2026 06:17:49 -0400 Subject: [PATCH 1/4] dogsdogsdogs: one lookup primitive under the delta-join operators `half_join`, `lookup_map`, `count`, `propose` and `validate` were five hand-rolled `binary_frontier` operators, each carrying its own copy of the stash-by-capability, frontier, cursor-seek and compaction logic. They are one operator with five output functions: for each (request, matching value): here are the consolidated (time, diff) arrangement updates the cut admits. Do what you like with them. `operators::lookup` is that operator. Two behaviours sit on it: `lookup_map` sums the admitted updates into one diff and emits once (`count`), and `lookup_join` visits them individually, joining each time into a time the request carries (`propose`, `validate`). `half_join2` is promoted to `half_join`, whose 366-line body becomes a 155-line derivation; the previous implementation is removed. Source is roughly flat at +903/-864 -- this consolidates duplicated operator logic rather than shrinking the tree, and a little over half of the new `lookup.rs` is doc comments. The +804 under `tests/` is the bulk of the change. Cuts are derived, not chosen. `Cut::for_positions(seed, atom)` returns the strict or lax comparison from the two atoms' positions: rule `seed` claims a combination exactly when it is the largest index achieving the `Ord`-maximum of the times, so exactly one rule claims each. Ordering by `Ord` rather than `PartialOrder` is what makes that maximum exist -- under the partial order two incomparable times have no maximum, no rule claims the pair, and the match is lost. Every other value a caller could pass either double counts or drops matches. The order time stays the dataflow timestamp and the join time rides in the payload. Progress tracking must see the time the cuts compare against, and since order <= join, a frontier over join times would not bound the order times still in flight. `propose` and `validate` therefore emit per admitted update rather than accumulating, which matters as soon as a cut admits an update incomparable to the request -- that is, inside a nested scope. `CollectionIndex` holds local `Arranged` values rather than trace handles. Re-importing an arrangement the same dataflow produced forfeits the scope's progress tracking, and inside a recursive scope the timestamp then counts upward without end. Documented on `TraceAgent::import`. Two behaviour changes worth review: - `validate` multiplies by the matched diff instead of passing the prefix diff through. An atom contributes to the product whether it proposed an extension or merely validated it. For set-valued relations the matched diff is one, so nothing observable moves. - `lookup_map` prunes a request only when the cut admits nothing, never when the admitted diffs sum to zero. Those updates can carry different join times, so their sum is not the count at any one of them; dropping on it removes the request from the collection outright, so no atom proposes it and the extension is lost. An inaccurate non-zero count merely picks a worse proposer. `count` still reports distinct-extension cardinality, which tracks changes but measures the wrong thing: the work is the post-consolidation updates a walk encounters. Reporting that needs a cursor hook that does not exist yet, so it is left alone. Tests, 11 in total, every case at one and four workers: - `delta_join_property`, `wcoj_triangle_property`: two- and three-atom joins against brute-force oracles over random `Product` updates. - `half_join_total_order`: exactly-once pinned from all three sides, on incomparable and on equal times. - `ktruss_iterative`: a worst-case-optimal join inside a recursive scope, maintained incrementally across insertions and deletions. Created using Claude Code --- .../src/operators/arrange/agent.rs | 18 + dogsdogsdogs/examples/delta_query.rs | 36 +- dogsdogsdogs/examples/delta_query2.rs | 6 +- dogsdogsdogs/examples/delta_query_wcoj.rs | 13 +- dogsdogsdogs/examples/dogsdogsdogs.rs | 28 +- dogsdogsdogs/src/lib.rs | 161 +++++-- dogsdogsdogs/src/operators/count.rs | 29 +- dogsdogsdogs/src/operators/half_join.rs | 366 +++----------- dogsdogsdogs/src/operators/half_join2.rs | 366 -------------- dogsdogsdogs/src/operators/lookup.rs | 446 ++++++++++++++++++ dogsdogsdogs/src/operators/lookup_join.rs | 116 +++++ dogsdogsdogs/src/operators/lookup_map.rs | 208 ++++---- dogsdogsdogs/src/operators/mod.rs | 6 +- dogsdogsdogs/src/operators/propose.rs | 43 +- dogsdogsdogs/src/operators/validate.rs | 26 +- dogsdogsdogs/tests/delta_join_property.rs | 208 ++++++++ dogsdogsdogs/tests/half_join_total_order.rs | 182 +++++++ dogsdogsdogs/tests/ktruss_iterative.rs | 216 +++++++++ dogsdogsdogs/tests/lookup_map_regression.rs | 13 +- dogsdogsdogs/tests/wcoj_triangle_property.rs | 191 ++++++++ 20 files changed, 1780 insertions(+), 898 deletions(-) delete mode 100644 dogsdogsdogs/src/operators/half_join2.rs create mode 100644 dogsdogsdogs/src/operators/lookup.rs create mode 100644 dogsdogsdogs/src/operators/lookup_join.rs create mode 100644 dogsdogsdogs/tests/delta_join_property.rs create mode 100644 dogsdogsdogs/tests/half_join_total_order.rs create mode 100644 dogsdogsdogs/tests/ktruss_iterative.rs create mode 100644 dogsdogsdogs/tests/wcoj_triangle_property.rs diff --git a/differential-dataflow/src/operators/arrange/agent.rs b/differential-dataflow/src/operators/arrange/agent.rs index a310be8bb..8485eed68 100644 --- a/differential-dataflow/src/operators/arrange/agent.rs +++ b/differential-dataflow/src/operators/arrange/agent.rs @@ -173,6 +173,24 @@ impl TraceAgent { /// responsibility this should be (the trace/batch should only reveal these times, or an operator should know /// to advance times before using them). /// + /// # Import a trace from elsewhere, not one this dataflow just built + /// + /// This method exists to replay a trace captured *some other way* — typically by a dataflow that has already + /// been built, as in the example below. Re-importing an arrangement produced by the same dataflow is an + /// antipattern, and inside a recursive scope it does not merely cost extra, it fails to terminate. + /// + /// An imported stream conveys progress by in-line progress statements rather than by participating in the + /// scope's progress tracking. Ordinary progress tracking can observe that a whole iterative subgraph is done + /// and let it exit; in-line statements have no whole-scope perspective, so capabilities advance only by + /// repeated frontier advancement — counting the timestamp upward round after round, and concluding only when + /// the timestamp would overflow. The symptom is a loop that emits a little output and then spins at full CPU + /// indefinitely, which reads like a hang but is really a count toward `u64::MAX`. There is also a risk that + /// times simply fail to advance. + /// + /// If the arrangement was built in this dataflow, keep the [`Arranged`] and share it: it is `Clone`, so any + /// number of operators can read it over ordinary dataflow edges, which is both correct and cheaper than a + /// replay operator per consumer. To carry an arrangement into a nested scope, use `enter`, not `import`. + /// /// # Examples /// /// ``` diff --git a/dogsdogsdogs/examples/delta_query.rs b/dogsdogsdogs/examples/delta_query.rs index bf5388fce..bef2094cf 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; @@ -72,22 +74,38 @@ fn main() { use differential_dogs3::operators::propose; use differential_dogs3::operators::validate; + // The alt/neu distinction rides on the arrangement times here, so every atom + // reads at the same cut; `AltNeu` is totally ordered, so the identity + // compaction bound is sound (see `operators::lookup::identity_frontier`). + use differential_dogs3::operators::{identity_frontier, Cut}; + + // Entering the delta region: each update carries its own time as the initial + // join time, while its dataflow timestamp stays the order time the cuts compare + // against. `leave_region` below turns the accumulated join time back into the + // update's own time. (`extend` brackets this for you; these calls are raw.) + let seeded = 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(seeded.clone(), forward_key_neu.clone(), Cut::AtOrBefore, identity_frontier, key2.clone()); + let changes1 = validate(changes1, forward_self_neu.clone(), Cut::AtOrBefore, identity_frontier, key1.clone()); + let changes1 = changes1 + .inner.map(|((data, carried), _order, r)| (data, carried, r)).as_collection() + .map(|((a,b),c)| (a,b,c)); // 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(seeded.clone(), reverse_key_alt.clone(), Cut::AtOrBefore, identity_frontier, key1.clone()); + let changes2 = validate(changes2, reverse_self_neu.clone(), Cut::AtOrBefore, identity_frontier, key2.clone()); + let changes2 = changes2 + .inner.map(|((data, carried), _order, r)| (data, carried, r)).as_collection() + .map(|((b,c),a)| (a,b,c)); // 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 changes3 = propose(seeded, forward_key_alt.clone(), Cut::AtOrBefore, identity_frontier, key1.clone()); + let changes3 = validate(changes3, reverse_self_alt.clone(), Cut::AtOrBefore, identity_frontier, key2.clone()); + let changes3 = changes3 + .inner.map(|((data, carried), _order, r)| (data, carried, r)).as_collection() + .map(|((a,c),b)| (a,b,c)); let prev_changes = changes1.concat(changes2).concat(changes3).leave(scope); diff --git a/dogsdogsdogs/examples/delta_query2.rs b/dogsdogsdogs/examples/delta_query2.rs index f29a95a1a..111eff4f1 100644 --- a/dogsdogsdogs/examples/delta_query2.rs +++ b/dogsdogsdogs/examples/delta_query2.rs @@ -31,7 +31,7 @@ fn main() { let changes1 = edges1.inner.map(|((k,v),t,r)| ((k,v,t.clone()),t,r)).as_collection(); let changes2 = edges2.inner.map(|((k,v),t,r)| ((k,v,t.clone()),t,r)).as_collection(); - use differential_dogs3::operators::half_join; + use differential_dogs3::operators::{half_join, Cut}; // pick a frontier that will not mislead TOTAL ORDER comparisons. let closure = |time: &Product, antichain: &mut timely::progress::Antichain>| { @@ -43,7 +43,7 @@ fn main() { changes1, forward2, closure, - |t1,t2| t1.lt(t2), // This one ignores concurrent updates. + Cut::Before, // This one ignores concurrent updates. |key, val1, val2| (key.clone(), (val1.clone(), val2.clone())), ); @@ -52,7 +52,7 @@ fn main() { changes2, forward1, closure, - |t1,t2| t1.le(t2), // This one can "see" concurrent updates. + Cut::AtOrBefore, // This one can "see" concurrent updates. |key, val1, val2| (key.clone(), (val2.clone(), val1.clone())), ); diff --git a/dogsdogsdogs/examples/delta_query_wcoj.rs b/dogsdogsdogs/examples/delta_query_wcoj.rs index cba34a71f..c222c521e 100644 --- a/dogsdogsdogs/examples/delta_query_wcoj.rs +++ b/dogsdogsdogs/examples/delta_query_wcoj.rs @@ -2,6 +2,7 @@ use timely::dataflow::operators::probe::Handle; use differential_dataflow::input::Input; use graph_map::GraphMMap; +use differential_dogs3::operators::{identity_frontier, Cut}; use differential_dogs3::{CollectionIndex, altneu::AltNeu}; use differential_dogs3::{ProposeExtensionMethod}; @@ -56,8 +57,8 @@ fn main() { forward .clone() .extend(&mut [ - &mut neu_forward.extend_using(|(_a,b)| *b), - &mut neu_forward.extend_using(|(a,_b)| *a), + &mut neu_forward.extend_using(|(_a,b)| *b, Cut::AtOrBefore, identity_frontier), + &mut neu_forward.extend_using(|(a,_b)| *a, Cut::AtOrBefore, identity_frontier), ]) .map(|((a,b),c)| (a,b,c)); @@ -66,16 +67,16 @@ fn main() { forward .clone() .extend(&mut [ - &mut alt_reverse.extend_using(|(b,_c)| *b), - &mut neu_reverse.extend_using(|(_b,c)| *c), + &mut alt_reverse.extend_using(|(b,_c)| *b, Cut::AtOrBefore, identity_frontier), + &mut neu_reverse.extend_using(|(_b,c)| *c, Cut::AtOrBefore, identity_frontier), ]) .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), + &mut alt_forward.extend_using(|(a,_c)| *a, Cut::AtOrBefore, identity_frontier), + &mut alt_reverse.extend_using(|(_a,c)| *c, Cut::AtOrBefore, identity_frontier), ]) .map(|((a,c),b)| (a,b,c)); diff --git a/dogsdogsdogs/examples/dogsdogsdogs.rs b/dogsdogsdogs/examples/dogsdogsdogs.rs index 110ec3922..b0026cc7c 100644 --- a/dogsdogsdogs/examples/dogsdogsdogs.rs +++ b/dogsdogsdogs/examples/dogsdogsdogs.rs @@ -1,10 +1,11 @@ -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; use graph_map::GraphMMap; use differential_dogs3::{CollectionIndex, PrefixExtender}; +use differential_dogs3::operators::{identity_frontier, Cut}; fn main() { @@ -31,26 +32,31 @@ fn main() { println!("loaded {} nodes, {} edges", nodes, edges.len()); - let index = worker.dataflow::(|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(); let mut edges = worker.dataflow::(|scope| { + // The index is built in the *same* dataflow that reads it, so both extenders share + // one local arrangement over an ordinary dataflow edge. Building it in a separate + // dataflow would force the extenders to re-import it, which costs a replay operator + // per use and, inside a recursive scope, prevents the loop from ever concluding. + let index = CollectionIndex::index(Collection::new(edges.to_stream(scope))); + let mut index_xz = index.extend_using(|&(ref x, ref _y)| *x, Cut::AtOrBefore, identity_frontier); + let mut index_yz = index.extend_using(|&(ref _x, ref y)| *y, Cut::AtOrBefore, identity_frontier); + let (edges_input, edges) = scope.new_collection(); + // Entering the delta region: carry each update's own time as the initial join + // time. The dataflow timestamp stays the order time the cuts compare against. + let seeded = edges.inner.map(|(d, t, r)| ((d, 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 = seeded.map(|(p, carried)| ((p, usize::MAX, usize::MAX), carried)); 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), carried),t,d)| (i as u64,((p, carried),t,d))); // propose extensions using relation based on index. let propose0 = index_xz.propose(parts[0].clone().as_collection()); @@ -62,6 +68,8 @@ fn main() { validate0 .concat(validate1) + // Leaving the delta region: the carried join time becomes the update's time. + .inner.map(|((data, carried), _order, r)| (data, carried, r)).as_collection() .inner .count() .inspect(move |x| println!("{:?}", x)) diff --git a/dogsdogsdogs/src/lib.rs b/dogsdogsdogs/src/lib.rs index ce484e3b7..3d41a64c8 100644 --- a/dogsdogsdogs/src/lib.rs +++ b/dogsdogsdogs/src/lib.rs @@ -1,7 +1,8 @@ use std::hash::Hash; +use std::rc::Rc; -use timely::progress::Timestamp; -use timely::dataflow::operators::vec::Partition; +use timely::progress::{Antichain, Timestamp}; +use timely::dataflow::operators::vec::{Map, Partition}; use timely::dataflow::operators::Concatenate; use differential_dataflow::{ExchangeData, VecCollection, AsCollection}; @@ -25,11 +26,16 @@ pub trait PrefixExtender<'scope, T: Timestamp, R: Monoid+Multiply> { /// 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>; - /// 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>; - /// 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>; + /// + /// Prefixes carry a *join* time alongside the payload; `count` passes it through untouched, + /// as a routing decision contributes no record to the output tuple. + fn count(&mut self, prefixes: VecCollection<'scope, T, ((Self::Prefix, usize, usize), T), R>, index: usize) -> VecCollection<'scope, T, ((Self::Prefix, usize, usize), T), R>; + /// Extends each prefix with corresponding extensions, joining the matched times in. + fn propose(&mut self, prefixes: VecCollection<'scope, T, (Self::Prefix, T), R>) -> VecCollection<'scope, T, ((Self::Prefix, Self::Extension), T), R>; + /// Restricts proposed extensions by those the extender would have proposed, joining the + /// matched times in — a validating atom contributes to the output time exactly as a + /// proposing one does. + fn validate(&mut self, extensions: VecCollection<'scope, T, ((Self::Prefix, Self::Extension), T), R>) -> VecCollection<'scope, T, ((Self::Prefix, Self::Extension), T), R>; } pub trait ProposeExtensionMethod<'scope, T: Timestamp, P: ExchangeData+Ord, R: Monoid+Multiply> { @@ -47,23 +53,30 @@ where where PE: PrefixExtender<'scope, T, R, Prefix=P> { - extender.propose(self) + let seeded = self.inner.map(|(p, t, r)| ((p, t.clone()), t, r)).as_collection(); + extender.propose(seeded) + .inner.map(|((data, carried), _order, diff)| (data, carried, diff)).as_collection() } fn extend(self, extenders: &mut [&mut dyn PrefixExtender<'scope, T,R,Prefix=P,Extension=E>]) -> VecCollection<'scope, T, (P, E), R> where E: ExchangeData+Ord { - if extenders.len() == 1 { - extenders[0].propose(self) + // Entering the delta region: each update carries its own time as the initial join + // time, while its dataflow timestamp stays the order time every cut compares against. + let seeded = self.inner.map(|(p, t, r)| ((p, t.clone()), t, r)).as_collection(); + + let extended = if extenders.len() == 1 { + extenders[0].propose(seeded) } else { - let mut counts = self.clone().map(|p| (p, 1 << 31, 0)); + let seeded_scope = seeded.scope(); + let mut counts = seeded.map(|(p, carried)| ((p, 1 << 31, 0), carried)); 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), carried),t,d)| (i as u64, ((p, carried),t,d))); let mut results = Vec::new(); for (index, nominations) in parts.into_iter().enumerate() { @@ -75,8 +88,12 @@ where results.push(extensions.inner); // save extensions } - self.scope().concatenate(results).as_collection() - } + seeded_scope.concatenate(results).as_collection() + }; + + // Leaving the delta region: the carried join time becomes the update's own time, and + // the order time — scaffolding for the cuts — is discarded. + extended.inner.map(|((data, carried), _order, diff)| (data, carried, diff)).as_collection() } } @@ -86,33 +103,58 @@ pub trait ValidateExtensionMethod<'scope, T: Timestamp, R: Monoid+Multiply, P, E> ValidateExtensionMethod<'scope, T, R, P, E> for VecCollection<'scope, T, (P, E), R> { fn validate_using>(self, extender: &mut PE) -> VecCollection<'scope, T, (P, E), R> { - extender.validate(self) + let seeded = self.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection(); + extender.validate(seeded) + .inner.map(|((data, carried), _order, diff)| (data, carried, diff)).as_collection() } } // These are all defined here so that users can be assured a common layout. use differential_dataflow::trace::implementations::{KeySpine, ValSpine}; -type TraceValHandle = TraceAgent>; -type TraceKeyHandle = TraceAgent>; +use differential_dataflow::operators::arrange::Arranged; +type ArrangedVal<'scope, K,V,T,R> = Arranged<'scope, TraceAgent>>; +type ArrangedKey<'scope, K,T,R> = Arranged<'scope, TraceAgent>>; -pub struct CollectionIndex +/// The three arrangements an atom is read through, held as *local* arrangements. +/// +/// # Why arrangements and not trace handles +/// +/// The obvious alternative is to store `TraceAgent` handles, which carry no scope lifetime and +/// so make an index portable anywhere. Getting an operator input back out of a handle means +/// [`TraceAgent::import`], and re-importing an arrangement produced by the *same* dataflow is +/// an antipattern: the imported stream carries in-line progress statements rather than +/// participating in the scope's progress tracking. Outside a loop that costs only a redundant +/// replay operator per use. Inside a recursive scope it is fatal — progress has no whole-scope +/// view, so it advances capabilities by repeated frontier advancement and simply counts the +/// timestamp upward, never concluding the loop is done. A three-atom join built this way spins +/// at full CPU rather than converging. +/// +/// Holding `Arranged` instead means every extender shares one stream and one trace by an +/// ordinary dataflow edge. `Arranged` is `Clone`, so sharing is free, and the redundant replay +/// operators disappear along with the hazard. The cost is the `'scope` lifetime: an index may +/// only be used in the scope that built it, which is what every caller already does. +/// +/// Genuinely external arrangements — captured and replayed from another dataflow — would want +/// a second variant here, as `import`'s legitimate use. Crossing into a nested scope is +/// `enter`'s job, not `import`'s. Neither is modelled: this is local-only. +pub struct CollectionIndex<'scope, K, V, T, R> where K: ExchangeData, V: ExchangeData, T: Lattice+ExchangeData+Timestamp, R: Monoid+Multiply+ExchangeData, { - /// A trace of type (K, ()), used to count extensions for each prefix. - count_trace: TraceKeyHandle, + /// An arrangement of `(K, ())`, used to count extensions for each prefix. + count: ArrangedKey<'scope, K, T, isize>, - /// A trace of type (K, V), used to propose extensions for each prefix. - propose_trace: TraceValHandle, + /// An arrangement of `(K, V)`, used to propose extensions for each prefix. + propose: ArrangedVal<'scope, K, V, T, R>, - /// A trace of type ((K, V), ()), used to validate proposed extensions. - validate_trace: TraceKeyHandle<(K, V), T, R>, + /// An arrangement of `((K, V), ())`, used to validate proposed extensions. + validate: ArrangedKey<'scope, (K, V), T, R>, } -impl Clone for CollectionIndex +impl<'scope, K, V, T, R> Clone for CollectionIndex<'scope, K, V, T, R> where K: ExchangeData+Hash, V: ExchangeData+Hash, @@ -121,14 +163,14 @@ where { fn clone(&self) -> Self { CollectionIndex { - count_trace: self.count_trace.clone(), - propose_trace: self.propose_trace.clone(), - validate_trace: self.validate_trace.clone(), + count: self.count.clone(), + propose: self.propose.clone(), + validate: self.validate.clone(), } } } -impl CollectionIndex +impl<'scope, K, V, T, R> CollectionIndex<'scope, K, V, T, R> where K: ExchangeData+Hash, V: ExchangeData+Hash, @@ -136,7 +178,7 @@ where R: Monoid+Multiply+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,27 +188,35 @@ 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(); - CollectionIndex { - count_trace: counts, - propose_trace: propose, - validate_trace: validate, - } + CollectionIndex { count: counts, propose, validate: arranged } } - pub fn extend_usingK+Clone>(&self, logic: F) -> CollectionExtender { + /// An extender reading this index at `cut`. + /// + /// The cut and its compaction bound are fixed here, on the extender, rather than at each + /// of `count` / `propose` / `validate`. That is deliberate: all three must read the *same* + /// cut relation. If `count` sizes an atom over a different cut than `propose` enumerates, + /// a prefix can be routed to the atom that offers the fewest extensions and then find + /// none — and since every other atom only validates, the extension is lost with nothing to + /// recover it. Binding the cut to the atom makes that unrepresentable. + pub fn extend_using(&self, logic: F, cut: operators::lookup::Cut, frontier_func: FF) -> CollectionExtender<'scope, K, V, T, R, P, F> + where + F: Fn(&P)->K+Clone, + FF: Fn(&T, &mut Antichain) + 'static, + { CollectionExtender { phantom: std::marker::PhantomData, indices: self.clone(), key_selector: logic, + cut, + frontier_func: Rc::new(frontier_func), } } } -pub struct CollectionExtender +pub struct CollectionExtender<'scope, K, V, T, R, P, F> where K: ExchangeData, V: ExchangeData, @@ -175,11 +225,15 @@ where F: Fn(&P)->K+Clone, { phantom: std::marker::PhantomData

, - indices: CollectionIndex, + indices: CollectionIndex<'scope, K, V, T, R>, key_selector: F, + /// The cut this atom is read at, shared by `count`, `propose`, and `validate`. + cut: operators::lookup::Cut, + /// The compaction bound the cut requires; see [`operators::lookup::identity_frontier`]. + frontier_func: Rc)>, } -impl<'scope, T, K, V, R, P, F> PrefixExtender<'scope, T, R> for CollectionExtender +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, @@ -191,18 +245,21 @@ where 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> { - let counts = self.indices.count_trace.import(prefixes.scope()); - operators::count::count(prefixes, counts, self.key_selector.clone(), index) + fn count(&mut self, prefixes: VecCollection<'scope, T, ((P, usize, usize), T), R>, index: usize) -> VecCollection<'scope, T, ((P, usize, usize), T), R> { + let counts = self.indices.count.clone(); + let ff = Rc::clone(&self.frontier_func); + operators::count::count(prefixes, counts, self.cut, move |t, a| ff(t, a), 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()) + fn propose(&mut self, prefixes: VecCollection<'scope, T, (P, T), R>) -> VecCollection<'scope, T, ((P, V), T), R> { + let propose = self.indices.propose.clone(); + let ff = Rc::clone(&self.frontier_func); + operators::propose::propose(prefixes, propose, self.cut, move |t, a| ff(t, a), 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()) + fn validate(&mut self, extensions: VecCollection<'scope, T, ((P, V), T), R>) -> VecCollection<'scope, T, ((P, V), T), R> { + let validate = self.indices.validate.clone(); + let ff = Rc::clone(&self.frontier_func); + operators::validate::validate(extensions, validate, self.cut, move |t, a| ff(t, a), self.key_selector.clone()) } } diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs index c46be8c25..9917f1d9e 100644 --- a/dogsdogsdogs/src/operators/count.rs +++ b/dogsdogsdogs/src/operators/count.rs @@ -3,39 +3,46 @@ use differential_dataflow::difference::{Semigroup, Monoid, Multiply}; use differential_dataflow::operators::arrange::Arranged; use differential_dataflow::trace::{BatchCursor, BatchDiff, BatchDiffGat, Cursor, Navigable, TraceReader}; +use crate::operators::lookup::Cut; + /// Reports a number of extensions to a stream of prefixes. /// /// 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>, +pub fn count<'scope, Tr, K, R, F, FF, P>( + prefixes: VecCollection<'scope, Tr::Time, ((P, usize, usize), Tr::Time), R>, arrangement: Arranged<'scope, Tr>, + cut: Cut, + frontier_func: FF, key_selector: F, index: usize, -) -> VecCollection<'scope, Tr::Time, (P, usize, usize), R> +) -> VecCollection<'scope, Tr::Time, ((P, usize, usize), Tr::Time), R> where - Tr: TraceReader+Clone+'static, + Tr: TraceReader+Clone+'static, BatchCursor: Cursor