Skip to content

dogsdogsdogs: one lookup primitive under the delta-join operators #818

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

Open
frankmcsherry wants to merge 3 commits into
base: master-next
Choose a base branch
from
+1,876 −1,223
Open

dogsdogsdogs: one lookup primitive under the delta-join operators #818

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
54f47ce
dogsdogsdogs: one lookup primitive under the delta-join operators
frankmcsherry Aug 4, 2026
1212832
dogsdogsdogs: retire AltNeu now that cuts are explicit
frankmcsherry Aug 4, 2026
7a19405
dogsdogsdogs: disambiguate intra-doc links
frankmcsherry Aug 5, 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
18 changes: 18 additions & 0 deletions differential-dataflow/src/operators/arrange/agent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,24 @@
/// 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
///
/// ```
Expand Down Expand Up @@ -283,7 +301,7 @@
let activator = scope.activator_for(Rc::clone(&info.address));
let queue = self.new_listener(activator);

let activator = scope.activator_for(info.address);

Check warning on line 304 in differential-dataflow/src/operators/arrange/agent.rs

View workflow job for this annotation

GitHub Actions / Cargo clippy 

`activator` shadows a previous, unrelated binding
*shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));

capabilities.borrow_mut().as_mut().unwrap().insert(capability);
Expand Down Expand Up @@ -416,7 +434,7 @@
let activator = scope.activator_for(Rc::clone(&info.address));
let queue = self.new_listener(activator);

let activator = scope.activator_for(info.address);

Check warning on line 437 in differential-dataflow/src/operators/arrange/agent.rs

View workflow job for this annotation

GitHub Actions / Cargo clippy 

`activator` shadows a previous, unrelated binding
*shutdown_button_ref = Some(ShutdownButton::new(Rc::clone(&capabilities), activator));

capabilities.borrow_mut().as_mut().unwrap().insert(capability);
Expand Down
130 changes: 48 additions & 82 deletions dogsdogsdogs/examples/delta_query.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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<usize>) {
antichain.insert(time.saturating_sub(1));
}

fn main() {

Expand Down Expand Up @@ -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::<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.
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)
Expand Down
6 changes: 3 additions & 3 deletions dogsdogsdogs/examples/delta_query2.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<usize, usize>, antichain: &mut timely::progress::Antichain<Product<usize, usize>>| {
Expand All @@ -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())),
);

Expand All @@ -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())),
);

Expand Down
104 changes: 48 additions & 56 deletions dogsdogsdogs/examples/delta_query_wcoj.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<usize>) {
antichain.insert(time.saturating_sub(1));
}

fn main() {

Expand All @@ -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::<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)
});
//
// 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)
Expand Down
28 changes: 18 additions & 10 deletions dogsdogsdogs/examples/dogsdogsdogs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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() {

Expand All @@ -31,26 +32,31 @@ 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();

let mut edges = worker.dataflow::<usize,_,_>(|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());
Expand All @@ -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))
Expand Down
Loading
Loading