dogs^3 compaction improvement (#457)

* Allow frontier function to populate antichain * Clean up dependencies
TimelyDataflow · Jan 25, 2024 · e153706 · e153706
1 parent dc5ebef
commit e153706
Show file tree

Hide file tree

Showing 6 changed files with 10 additions and 12 deletions.
diff --git a/dogsdogsdogs/examples/delta_query2.rs b/dogsdogsdogs/examples/delta_query2.rs
@@ -36,13 +36,15 @@ fn main() {
                 use dogsdogsdogs::operators::half_join;
 
                 // pick a frontier that will not mislead TOTAL ORDER comparisons.
-                let closure = |time: &Product<usize, usize>| Product::new(time.outer.saturating_sub(1), time.inner.saturating_sub(1));
+                let closure = |time: &Product<usize, usize>, antichain: &mut timely::progress::Antichain<Product<usize, usize>>| { 
+                    antichain.insert(Product::new(time.outer.saturating_sub(1), time.inner.saturating_sub(1))); 
+                };
 
                 let path1 =
                 half_join(
                     &changes1,
                     forward2,
-                    closure.clone(),
+                    closure,
                     |t1,t2| t1.lt(t2),  // This one ignores concurrent updates.
                     |key, val1, val2| (key.clone(), (val1.clone(), val2.clone())),
                 );
@@ -51,7 +53,7 @@ fn main() {
                 half_join(
                     &changes2,
                     forward1,
-                    closure.clone(),
+                    closure,
                     |t1,t2| t1.le(t2),  // This one can "see" concurrent updates.
                     |key, val1, val2| (key.clone(), (val2.clone(), val1.clone())),
                 );

diff --git a/dogsdogsdogs/src/operators/count.rs b/dogsdogsdogs/src/operators/count.rs
@@ -2,7 +2,6 @@ use timely::dataflow::Scope;
 
 use differential_dataflow::{ExchangeData, Collection, Hashable};
 use differential_dataflow::difference::{Monoid, Multiply};
-use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::trace::TraceReader;
 

diff --git a/dogsdogsdogs/src/operators/half_join.rs b/dogsdogsdogs/src/operators/half_join.rs
@@ -83,7 +83,7 @@ where
     Tr: TraceReader+Clone+'static,
     R: Mul<Tr::Diff>,
     <R as Mul<Tr::Diff>>::Output: Semigroup,
-    FF: Fn(&G::Timestamp) -> G::Timestamp + 'static,
+    FF: Fn(&G::Timestamp, &mut Antichain<G::Timestamp>) + 'static,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,
     DOut: Clone+'static,
     S: FnMut(&Tr::KeyOwned, &V, Tr::Val<'_>)->DOut+'static,
@@ -134,7 +134,7 @@ where
     V: ExchangeData,
     R: ExchangeData + Monoid,
     Tr: TraceReader+Clone+'static,
-    FF: Fn(&G::Timestamp) -> G::Timestamp + 'static,
+    FF: Fn(&G::Timestamp, &mut Antichain<G::Timestamp>) + 'static,
     CF: Fn(&G::Timestamp, &G::Timestamp) -> bool + 'static,
     DOut: Clone+'static,
     ROut: Semigroup,
@@ -281,10 +281,10 @@ where
             // The logical merging frontier depends on both input1 and stash.
             let mut frontier = timely::progress::frontier::Antichain::new();
             for time in input1.frontier().frontier().iter() {
-                frontier.insert(frontier_func(time));
+                frontier_func(time, &mut frontier);
             }
-            for key in stash.keys() {
-                frontier.insert(frontier_func(key.time()));
+            for time in stash.keys() {
+                frontier_func(time, &mut frontier);
             }
             arrangement_trace.as_mut().map(|trace| trace.set_logical_compaction(frontier.borrow()));
 

diff --git a/dogsdogsdogs/src/operators/lookup_map.rs b/dogsdogsdogs/src/operators/lookup_map.rs
@@ -8,7 +8,6 @@ use timely::progress::Antichain;
 
 use differential_dataflow::{ExchangeData, Collection, AsCollection, Hashable};
 use differential_dataflow::difference::{Semigroup, Monoid};
-use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::trace::{Cursor, TraceReader};
 

diff --git a/dogsdogsdogs/src/operators/propose.rs b/dogsdogsdogs/src/operators/propose.rs
@@ -2,7 +2,6 @@ use timely::dataflow::Scope;
 
 use differential_dataflow::{ExchangeData, Collection, Hashable};
 use differential_dataflow::difference::{Monoid, Multiply};
-use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::trace::TraceReader;
 use differential_dataflow::trace::cursor::MyTrait;

diff --git a/dogsdogsdogs/src/operators/validate.rs b/dogsdogsdogs/src/operators/validate.rs
@@ -4,7 +4,6 @@ use timely::dataflow::Scope;
 
 use differential_dataflow::{ExchangeData, Collection};
 use differential_dataflow::difference::{Monoid, Multiply};
-use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::arrange::Arranged;
 use differential_dataflow::trace::TraceReader;