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..acae26347 100644 --- a/dogsdogsdogs/examples/delta_query.rs +++ b/dogsdogsdogs/examples/delta_query.rs @@ -1,9 +1,18 @@ 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; -use differential_dogs3::altneu::AltNeu; -use differential_dogs3::calculus::{Differentiate, Integrate}; +use timely::progress::Antichain; + +use differential_dogs3::operators::{propose, validate, Cut}; + +/// `Cut::Before` is strict, and logical compaction destroys strictness, so the bound must sit +/// strictly below every time still held. Sound for the lax cuts here too, just conservative. +fn step_back(time: &usize, antichain: &mut Antichain) { + antichain.insert(time.saturating_sub(1)); +} fn main() { @@ -42,86 +51,43 @@ 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::,_,_>("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; - - // 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)); - - // 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)); - - // 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); - - // 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. - triangles_prev.clone().assert_eq(triangles_next); + // + // One delta query per relation, driven by changes to that relation. Each rule + // proposes from one atom and validates against the other; which side of a tie + // each is read at follows from the atoms' positions (E1 = 0, E2 = 1, E3 = 2). + // + // These are the raw `propose` / `validate` operators rather than `extend`, so the + // delta region is bracketed by hand: each update carries its own time as the + // initial join time while its dataflow timestamp stays the order time the cuts + // compare against, and the carried time becomes the update's own time on the way + // out. `extend` does both for you. + let key1 = |x: &(u32, u32)| x.0; + let key2 = |x: &(u32, u32)| x.1; + + let seeded = edges.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(seeded.clone(), forward_key.clone(), Cut::for_positions(0, 1), step_back, key2); + let changes1 = validate(changes1, forward_self.clone(), Cut::for_positions(0, 2), step_back, key1); + 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(seeded.clone(), reverse_key.clone(), Cut::for_positions(1, 0), step_back, key1); + let changes2 = validate(changes2, reverse_self.clone(), Cut::for_positions(1, 2), step_back, key2); + 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(seeded, forward_key, Cut::for_positions(2, 0), step_back, key1); + let changes3 = validate(changes3, reverse_self, Cut::for_positions(2, 1), step_back, key2); + let changes3 = changes3 + .inner.map(|((data, carried), _order, r)| (data, carried, r)).as_collection() + .map(|((a, c), b)| (a, b, c)); + + let triangles_prev = changes1.concat(changes2).concat(changes3); triangles_prev .filter(move |_| inspect) 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..56b24dc64 100644 --- a/dogsdogsdogs/examples/delta_query_wcoj.rs +++ b/dogsdogsdogs/examples/delta_query_wcoj.rs @@ -2,8 +2,17 @@ use timely::dataflow::operators::probe::Handle; use differential_dataflow::input::Input; use graph_map::GraphMMap; -use differential_dogs3::{CollectionIndex, altneu::AltNeu}; -use differential_dogs3::{ProposeExtensionMethod}; +use timely::progress::Antichain; + +use differential_dogs3::operators::Cut; +use differential_dogs3::CollectionIndex; +use differential_dogs3::ProposeExtensionMethod; + +/// `Cut::Before` is strict, and logical compaction destroys strictness, so the bound must sit +/// strictly below every time still held. +fn step_back(time: &usize, antichain: &mut Antichain) { + antichain.insert(time.saturating_sub(1)); +} fn main() { @@ -26,61 +35,44 @@ fn main() { let (edges_input, edges) = scope.new_collection(); - 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::,_,_>("DeltaQuery (Triangles)", |inner| { - - // Each relation we'll need. - let forward = forward.enter(inner); - let reverse = reverse.enter(inner); - - // 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()))); - - // 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 = - 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) - }); + // + // One delta query per relation, driven by changes to that relation. Relations are + // sequenced so each introduces a variable sharing a bound attribute with the + // running prefix, and directed { forward, reverse } by whether the bound variable + // occurs first or second. + // + // Which side of a tie each atom reads at is derived from the atoms' positions + // (E1 = 0, E2 = 1, E3 = 2), so two indices suffice. Encoding the same distinction + // in the timestamp needed an "old" and a "new" copy of each, and four indices. + let forward = CollectionIndex::index(edges.clone()); + let reverse = CollectionIndex::index(edges.clone().map(|(x, y)| (y, x))); + + // dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c); bind (a,b), extend by c + let changes1 = edges.clone() + .extend(&mut [ + &mut forward.extend_using(|(_a, b): &(u32, u32)| *b, Cut::for_positions(0, 1), step_back), + &mut forward.extend_using(|(a, _b): &(u32, u32)| *a, Cut::for_positions(0, 2), step_back), + ]) + .map(|((a, b), c)| (a, b, c)); + + // dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c); bind (b,c), extend by a + let changes2 = edges.clone() + .extend(&mut [ + &mut reverse.extend_using(|(b, _c): &(u32, u32)| *b, Cut::for_positions(1, 0), step_back), + &mut reverse.extend_using(|(_b, c): &(u32, u32)| *c, Cut::for_positions(1, 2), step_back), + ]) + .map(|((b, c), a)| (a, b, c)); + + // dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c); bind (a,c), extend by b + let changes3 = edges + .extend(&mut [ + &mut forward.extend_using(|(a, _c): &(u32, u32)| *a, Cut::for_positions(2, 0), step_back), + &mut reverse.extend_using(|(_a, c): &(u32, u32)| *c, Cut::for_positions(2, 1), step_back), + ]) + .map(|((a, c), b)| (a, b, c)); + + let triangles = changes1.concat(changes2).concat(changes3); triangles .filter(move |_| inspect) 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/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 { - pub time: T, - pub neu: bool, // alt < neu in timestamp comparisons. -} - -impl AltNeu { - 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 PartialOrder for AltNeu { - 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 PathSummary> for () { - fn results_in(&self, timestamp: &AltNeu) -> Option> { - Some(timestamp.clone()) - } - fn followed_by(&self, other: &Self) -> Option { - Some(other.clone()) - } -} - -// Implement timely dataflow's `Timestamp` trait. -use timely::progress::Timestamp; -impl Timestamp for AltNeu { - type Summary = (); - fn minimum() -> Self { AltNeu::alt(T::minimum()) } -} - -use timely::progress::timestamp::Refines; - -impl Refines for AltNeu { - fn to_inner(other: T) -> Self { - AltNeu::alt(other) - } - fn to_outer(self: AltNeu) -> 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 Lattice for AltNeu { - 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>) -> VecCollection<'inner, AltNeu, 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>) -> VecCollection<'inner, AltNeu, 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, 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 ce484e3b7..180a8bc16 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}; @@ -9,8 +10,6 @@ use differential_dataflow::difference::{Monoid, Multiply}; use differential_dataflow::lattice::Lattice; use differential_dataflow::operators::arrange::TraceAgent; -pub mod altneu; -pub mod calculus; pub mod operators; /// A type capable of extending a stream of prefixes. @@ -25,11 +24,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 +51,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 +86,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 +101,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 +161,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 +176,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 +186,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 +223,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 +243,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