Skip to content

Correct dogs3 worst-case optimal tools #821

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
frankmcsherry merged 5 commits into from
Aug 7, 2026
Merged

Correct dogs3 worst-case optimal tools #821

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
865bb73
Avoid non-termination that results from in-band progress tracking
frankmcsherry Aug 7, 2026
1482236
Retarget count/propose/validate onto half_join
frankmcsherry Aug 7, 2026
02b75c0
Test the WCOJ path over partially ordered times
frankmcsherry Aug 7, 2026
3a474d0
State delta-query strictness directly, rather than in the timestamp
frankmcsherry Aug 7, 2026
4cb5ddc
Delete altneu and calculus, which no longer have consumers
frankmcsherry Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
139 changes: 58 additions & 81 deletions dogsdogsdogs/examples/delta_query.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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() {

Expand Down Expand Up @@ -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::<AltNeu<usize>,_,_>("DeltaQuery (Triangles)", |inner| {

// Grab the stream of changes.
let changes = edges.clone().enter(inner);

// Each relation we'll need.
let forward_key_alt = forward_key.clone().enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let reverse_key_alt = reverse_key.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let forward_key_neu = forward_key.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
// let reverse_key_neu = reverse_key.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));

// let forward_self_alt = forward_self.enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let reverse_self_alt = reverse_self.clone().enter_at(inner, |_,_,t| AltNeu::alt(t.clone()), |t| t.time.saturating_sub(1));
let forward_self_neu = forward_self.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));
let reverse_self_neu = reverse_self.enter_at(inner, |_,_,t| AltNeu::neu(t.clone()), |t| t.time.saturating_sub(1));

// For each relation, we form a delta query driven by changes to that relation.
//
// The sequence of joined relations are such that we only introduce relations
// which share some bound attributes with the current stream of deltas.
// Each joined relation is delayed { alt -> neu } if its position in the
// sequence is greater than the delta stream.
// Each joined relation is directed { forward, reverse } by whether the
// bound variable occurs in the first or second position.

let key1 = |x: &(u32, u32)| x.0;
let key2 = |x: &(u32, u32)| x.1;

use differential_dogs3::operators::propose;
use differential_dogs3::operators::validate;

// 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<usize>| {
antichain.insert(time.saturating_sub(1));
};

// Stash each delta's own time as its payload, to be advanced by the times of the
// records it matches, and delayed to once we are done extending.
let deltas = edges.clone().inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();

// dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
let changes1 = propose(deltas.clone(), forward_key.clone(), key2.clone(), frontier_func, true);
let changes1 = validate(changes1, forward_self.clone(), key1.clone(), frontier_func, true);
let changes1 = changes1.map(|(((a,b),c), payload)| ((a,b,c), payload));

// dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
let changes2 = propose(deltas.clone(), reverse_key.clone(), key1.clone(), frontier_func, false);
let changes2 = validate(changes2, reverse_self.clone(), key2.clone(), frontier_func, true);
let changes2 = changes2.map(|(((b,c),a), payload)| ((a,b,c), payload));

// dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
let changes3 = propose(deltas, forward_key.clone(), key1.clone(), frontier_func, false);
let changes3 = validate(changes3, reverse_self.clone(), key2.clone(), frontier_func, false);
let changes3 = changes3.map(|(((a,c),b), payload)| ((a,b,c), payload));

// Delay updates to the payload time worked out while extending.
let triangles_prev = changes1.concat(changes2).concat(changes3)
.inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection();

// The same query as a conventional three-way join, which shares no machinery with
// the delta fragments above and so is an independent answer rather than a second
// opinion from the same method.
let triangles_next =
edges
.map(|(x,y)| (y,x))
.join_core(forward_key, |b,a,c| Some(((*a, *c), *b)))
.join_core(forward_self, |(a,c), b, &()| Some((*a,*b,*c)));

// Test that the concatenated delta fragments equal the conventional join.
triangles_prev.clone().assert_eq(triangles_next);

triangles_prev
Expand Down
113 changes: 61 additions & 52 deletions dogsdogsdogs/examples/delta_query_wcoj.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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() {
Expand All @@ -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::<AltNeu<usize>,_,_>("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<usize>| {
antichain.insert(time.saturating_sub(1));
};

// One index per orientation. The "old" and "new" copies these replace differed
// only in whether a lookup could see updates at the delta's own time, which each
// `extend_using` below now states for itself.
let index_forward = CollectionIndex::index(forward.clone(), frontier_func);
let index_reverse = CollectionIndex::index(reverse.clone(), frontier_func);

// Stash each delta's own time as its payload, to be advanced by the times of the
// records it matches, and delayed to once we are done extending.
let deltas = forward.inner.map(|(d, t, r)| ((d, t.clone()), t, r)).as_collection();

// For each relation, we form a delta query driven by changes to that relation.
//
// The sequence of joined relations are such that we only introduce relations
// which share some bound attributes with the current stream of deltas.
// Each lookup is strict exactly when it reaches a relation later in the sequence
// than the delta stream, which is what stops a pair of updates matching twice.
// Each joined relation is directed { forward, reverse } by whether the
// bound variable occurs in the first or second position.

// dQ/dE1 := dE1(a,b), E2(b,c), E3(a,c)
let changes1 =
deltas
.clone()
.extend(&mut [
&mut index_forward.extend_using(|(_a,b)| *b, true),
&mut index_forward.extend_using(|(a,_b)| *a, true),
])
.map(|(((a,b),c), payload)| ((a,b,c), payload));

// dQ/dE2 := dE2(b,c), E1(a,b), E3(a,c)
let changes2 =
deltas
.clone()
.extend(&mut [
&mut index_reverse.extend_using(|(b,_c)| *b, false),
&mut index_reverse.extend_using(|(_b,c)| *c, true),
])
.map(|(((b,c),a), payload)| ((a,b,c), payload));

// dQ/dE3 := dE3(a,c), E1(a,b), E2(b,c)
let changes3 = deltas
.extend(&mut [
&mut index_forward.extend_using(|(a,_c)| *a, false),
&mut index_reverse.extend_using(|(_a,c)| *c, false),
])
.map(|(((a,c),b), payload)| ((a,b,c), payload));

// Delay updates to the payload time worked out while extending.
let triangles = changes1.concat(changes2).concat(changes3)
.inner.map(|((d, payload), _time, r)| (d, payload, r)).as_collection();

triangles
.filter(move |_| inspect)
Expand Down
33 changes: 22 additions & 11 deletions dogsdogsdogs/examples/dogsdogsdogs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -31,26 +31,36 @@ fn main() {

println!("loaded {} nodes, {} edges", nodes, edges.len());

let index = worker.dataflow::<usize,_,_>(|scope| {
CollectionIndex::index(Collection::new(edges.to_stream(scope)))
});

let mut index_xz = index.extend_using(|&(ref x, ref _y)| *x);
let mut index_yz = index.extend_using(|&(ref _x, ref y)| *y);

let mut probe = Handle::new();

// The index and its readers must share a dataflow: the extenders hold scope-bound
// arrangements rather than exported traces.
let mut edges = worker.dataflow::<usize,_,_>(|scope| {

// Hold compaction back one step, and let a prefix see arranged updates at its own
// time and earlier. The index is static here, so every prefix sees all of it.
let frontier_func = |time: &usize, antichain: &mut timely::progress::Antichain<usize>| {
antichain.insert(time.saturating_sub(1));
};

let 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());
Expand All @@ -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))
Expand Down
Loading
Loading