diff --git a/dogsdogsdogs/examples/delta_query.rs b/dogsdogsdogs/examples/delta_query.rs index bf5388fce..b087bcb8e 100644 --- a/dogsdogsdogs/examples/delta_query.rs +++ b/dogsdogsdogs/examples/delta_query.rs @@ -1,9 +1,9 @@ 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}; fn main() { @@ -42,85 +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::,_,_>("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. + // + // 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| { + 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 cba34a71f..b751b2959 100644 --- a/dogsdogsdogs/examples/delta_query_wcoj.rs +++ b/dogsdogsdogs/examples/delta_query_wcoj.rs @@ -1,8 +1,10 @@ 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::{CollectionIndex, altneu::AltNeu}; +use differential_dogs3::CollectionIndex; use differential_dogs3::{ProposeExtensionMethod}; fn main() { @@ -29,58 +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::,_,_>("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) - }); + + // 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| { + 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 110ec3922..49da755a2 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; @@ -31,26 +31,36 @@ 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(); + // The index and its readers must share a dataflow: the extenders hold scope-bound + // arrangements rather than exported traces. let mut edges = worker.dataflow::(|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| { + antichain.insert(time.saturating_sub(1)); + }; + + let index = CollectionIndex::index(Collection::new(edges.to_stream(scope)), frontier_func); + + 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(); + // 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 +72,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/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..6736f8df5 100644 --- a/dogsdogsdogs/src/lib.rs +++ b/dogsdogsdogs/src/lib.rs @@ -1,55 +1,94 @@ +//! 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; -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 = Rc)>; + /// 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> { +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> { - fn propose_using>(self, extender: &mut PE) -> VecCollection<'scope, T, (P, PE::Extension), R>; - fn extend(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>(self, extender: &mut PE) -> VecCollection<'scope, T, ((P, PE::Extension), T), isize>; + fn extend(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+'static, { - fn propose_using(self, extender: &mut PE) -> VecCollection<'scope, T, (P, PE::Extension), R> + fn propose_using(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(self, extenders: &mut [&mut dyn PrefixExtender<'scope, T,R,Prefix=P,Extension=E>]) -> VecCollection<'scope, T, (P, E), R> + fn extend(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 +97,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,63 +119,75 @@ where } } -pub trait ValidateExtensionMethod<'scope, T: Timestamp, R: Monoid+Multiply, P, E> { - fn validate_using>(self, extender: &mut PE) -> VecCollection<'scope, T, (P, E), R>; +pub trait ValidateExtensionMethod<'scope, T: Timestamp, P, E> { + fn validate_using>(self, extender: &mut PE) -> VecCollection<'scope, T, ((P, E), T), isize>; } -impl<'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> { +impl<'scope, T: Timestamp, P, E> ValidateExtensionMethod<'scope, T, P, E> for VecCollection<'scope, T, ((P, E), T), isize> { + fn validate_using>(self, extender: &mut PE) -> VecCollection<'scope, T, ((P, E), T), isize> { extender.validate(self) } } // 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 = TraceAgent>; -type TraceKeyHandle = TraceAgent>; +type TraceValHandle = TraceAgent>; +type TraceKeyHandle = TraceAgent>; -pub struct CollectionIndex +/// 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. +/// +/// 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+ExchangeData, { /// A trace of type (K, ()), used to count extensions for each prefix. - count_trace: TraceKeyHandle, + count_trace: Arranged<'scope, TraceKeyHandle>, /// A trace of type (K, V), used to propose extensions for each prefix. - propose_trace: TraceValHandle, + propose_trace: Arranged<'scope, TraceValHandle>, /// 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>>, + + /// Holds back compaction; see [`FrontierFunc`]. + frontier_func: FrontierFunc, } -impl Clone for CollectionIndex +impl<'scope, K, V, T> Clone for CollectionIndex<'scope, K, V, T> where - K: ExchangeData+Hash, - V: ExchangeData+Hash, + K: ExchangeData, + V: ExchangeData, T: Lattice+ExchangeData+Timestamp, - R: Monoid+Multiply+ExchangeData, { 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), } } } -impl CollectionIndex +impl<'scope, K, V, T> CollectionIndex<'scope, K, V, T> where K: ExchangeData+Hash, V: ExchangeData+Hash, T: Lattice+ExchangeData+Timestamp, - 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), isize>, frontier_func: FF) -> Self + where + FF: Fn(&T, &mut Antichain) + '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(); @@ -146,63 +197,80 @@ 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, propose_trace: propose, validate_trace: validate, + frontier_func: Rc::new(frontier_func), } } - pub fn extend_usingK+Clone>(&self, logic: F) -> CollectionExtender { + /// 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_usingK+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, } } } -pub struct CollectionExtender +pub struct CollectionExtender<'scope, K, V, T, P, F> where K: ExchangeData, V: ExchangeData, T: Lattice+ExchangeData+Timestamp, - R: Monoid+Multiply+ExchangeData, F: Fn(&P)->K+Clone, { phantom: std::marker::PhantomData

, - indices: CollectionIndex, + 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> +where + K: ExchangeData, + V: ExchangeData, + T: Lattice+ExchangeData+Timestamp, + F: Fn(&P)->K+Clone, +{ + /// The index's compaction closure, as a plain callable the operators can accept. + fn frontier_func(&self) -> impl Fn(&T, &mut Antichain) + 'static { + let frontier_func = Rc::clone(&self.indices.frontier_func); + move |t: &T, a: &mut Antichain| frontier_func(t, a) + } } -impl<'scope, T, K, V, R, P, F> PrefixExtender<'scope, T, R> for CollectionExtender +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+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> { - 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), isize>, index: usize) -> VecCollection<'scope, T, ((P, usize, usize), T), isize> { + 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, 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), isize>) -> VecCollection<'scope, T, ((P, V), T), isize> { + 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), 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), isize>) -> VecCollection<'scope, T, ((P, V), T), isize> { + 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 c46be8c25..8aa0eebc7 100644 --- a/dogsdogsdogs/src/operators/count.rs +++ b/dogsdogsdogs/src/operators/count.rs @@ -1,41 +1,78 @@ -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>( + 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, + strict: bool, +) -> VecCollection<'scope, Tr::Time, ((P, usize, usize), Tr::Time), R> where Tr: TraceReader+Clone+'static, - BatchCursor: Cursor