Updates for timely 0.25

Author: frankmcsherry

Cargo.toml

@@ -17,9 +17,9 @@ resolver = "2"
 
 [workspace.dependencies]
 differential-dataflow = { path = "differential-dataflow", default-features = false, version = "0.17.0" }
-timely = { version = "0.24", default-features = false }
+#timely = { version = "0.24", default-features = false }
 columnar = { version = "0.10", default-features = false }
-#timely = { path = "../timely-dataflow/timely/", default-features = false }
+timely = { path = "../timely-dataflow/timely/", default-features = false }
 
 [profile.release]
 opt-level = 3

differential-dataflow/examples/columnar.rs

@@ -59,7 +59,8 @@ fn main() {
                             let time = Pair::new(arguments[1], arguments[2]);
                             if capability.time().less_equal(&time) {
                                 input
-                                    .session(capability.clone())
+                                    .activate()
+                                    .session(&capability)
                                     .give((arguments[0], time, arguments[3]));
                             } else {
                                 println!("Requested time {:?} no longer open (input from {:?})", time, capability.time());

differential-dataflow/examples/multitemporal.rs

@@ -121,7 +121,7 @@ fn frontier<G, T>(
 ) -> Collection<G, (Location, T)>
 where
     G: Scope<Timestamp: Lattice+Ord>,
-    T: Timestamp<Summary: differential_dataflow::ExchangeData>,
+    T: Timestamp<Summary: differential_dataflow::ExchangeData>+std::hash::Hash,
 {
     // Translate node and edge transitions into a common Location to Location edge with an associated Summary.
     let nodes = nodes.map(|(target, source, summary)| (Location::from(target), (Location::from(source), summary)));

differential-dataflow/examples/progress.rs

@@ -16,7 +16,7 @@ use crate::operators::arrange::arrangement::ArrangeByKey;
 /// method to limit the introduction of labels.
 pub fn propagate<G, N, L, R>(edges: &Collection<G, (N,N), R>, nodes: &Collection<G,(N,L),R>) -> Collection<G,(N,L),R>
 where
-    G: Scope<Timestamp: Lattice+Ord>,
+    G: Scope<Timestamp: Lattice+Ord+Hash>,
     N: ExchangeData+Hash,
     R: ExchangeData+Abelian,
     R: Multiply<R, Output=R>,
@@ -33,7 +33,7 @@ where
 /// method to limit the introduction of labels.
 pub fn propagate_at<G, N, L, F, R>(edges: &Collection<G, (N,N), R>, nodes: &Collection<G,(N,L),R>, logic: F) -> Collection<G,(N,L),R>
 where
-    G: Scope<Timestamp: Lattice+Ord>,
+    G: Scope<Timestamp: Lattice+Ord+Hash>,
     N: ExchangeData+Hash,
     R: ExchangeData+Abelian,
     R: Multiply<R, Output=R>,
@@ -60,7 +60,7 @@ where
     R: Multiply<R, Output=R>,
     R: From<i8>,
     L: ExchangeData,
-    Tr: for<'a> TraceReader<Key<'a>=&'a N, Val<'a>=&'a N, Diff=R>+Clone+'static,
+    Tr: for<'a> TraceReader<Key<'a>=&'a N, Val<'a>=&'a N, Time:Hash, Diff=R>+Clone+'static,
     F: Fn(&L)->u64+Clone+'static,
 {
     // Morally the code performs the following iterative computation. However, in the interest of a simplified

differential-dataflow/src/algorithms/graphs/propagate.rs

@@ -35,7 +35,7 @@ where
 /// Returns the subset of edges in the same strongly connected component.
 pub fn strongly_connected<G, N, R>(graph: &Collection<G, (N,N), R>) -> Collection<G, (N,N), R>
 where
-    G: Scope<Timestamp: Lattice + Ord>,
+    G: Scope<Timestamp: Lattice+Ord+Hash>,
     N: ExchangeData + Hash,
     R: ExchangeData + Abelian,
     R: Multiply<R, Output=R>,
@@ -51,7 +51,7 @@ where
 fn trim_edges<G, N, R>(cycle: &Collection<G, (N,N), R>, edges: &Collection<G, (N,N), R>)
     -> Collection<G, (N,N), R>
 where
-    G: Scope<Timestamp: Lattice+Ord>,
+    G: Scope<Timestamp: Lattice+Ord+Hash>,
     N: ExchangeData + Hash,
     R: ExchangeData + Abelian,
     R: Multiply<R, Output=R>,

differential-dataflow/src/algorithms/graphs/scc.rs

@@ -11,7 +11,7 @@ use crate::hashable::Hashable;
 
 fn _color<G, N>(edges: &Collection<G, (N,N)>) -> Collection<G,(N,Option<u32>)>
 where
-    G: Scope<Timestamp: Lattice+Ord>,
+    G: Scope<Timestamp: Lattice+Ord+Hash>,
     N: ExchangeData+Hash,
 {
     // need some bogus initial values.

differential-dataflow/src/algorithms/graphs/sequential.rs

@@ -228,6 +228,7 @@ pub mod source {
     use std::marker::{Send, Sync};
     use std::sync::Arc;
     use timely::dataflow::{Scope, Stream, operators::{Capability, CapabilitySet}};
+    use timely::dataflow::operators::generic::OutputBuilder;
     use timely::progress::Timestamp;
     use timely::scheduling::SyncActivator;
 
@@ -313,8 +314,11 @@ pub mod source {
         let activator2 = scope.activator_for(Rc::clone(&address));
         let drop_activator = DropActivator { activator: Arc::new(scope.sync_activator_for(address.to_vec())) };
         let mut source = source_builder(activator);
-        let (mut updates_out, updates) = messages_op.new_output();
-        let (mut progress_out, progress) = messages_op.new_output();
+        let (updates_out, updates) = messages_op.new_output();
+        let mut updates_out = OutputBuilder::from(updates_out);
+        let (progress_out, progress) = messages_op.new_output();
+        let mut progress_out = OutputBuilder::from(progress_out);
+
         messages_op.build(|capabilities| {
 
             // A Weak that communicates whether the returned token has been dropped.
@@ -387,8 +391,11 @@ pub mod source {
         // Step 2: The UPDATES operator.
         let mut updates_op = OperatorBuilder::new("CDCV2_Updates".to_string(), scope.clone());
         let mut input = updates_op.new_input(&updates, Exchange::new(|x: &(D, T, R)| x.hashed()));
-        let (mut changes_out, changes) = updates_op.new_output();
-        let (mut counts_out, counts) = updates_op.new_output();
+        let (changes_out, changes) = updates_op.new_output();
+        let mut changes_out = OutputBuilder::from(changes_out);
+        let (counts_out, counts) = updates_op.new_output();
+        let mut counts_out = OutputBuilder::from(counts_out);
+
         updates_op.build(move |_capability| {
             // Deduplicates updates, and ships novel updates and the counts for each time.
             // For simplicity, this operator ships updates as they are discovered to be new.
@@ -438,7 +445,8 @@ pub mod source {
         );
         let mut counts =
             progress_op.new_input(&counts, Exchange::new(|x: &(T, i64)| (x.0).hashed()));
-        let (mut frontier_out, frontier) = progress_op.new_output();
+        let (frontier_out, frontier) = progress_op.new_output();
+        let mut frontier_out = OutputBuilder::from(frontier_out);
         progress_op.build(move |_capability| {
             // Receive progress statements, deduplicated counts. Track lower frontier of both and broadcast changes.
 
@@ -554,7 +562,7 @@ pub mod sink {
     use timely::progress::{Antichain, ChangeBatch, Timestamp};
     use timely::dataflow::{Scope, Stream};
     use timely::dataflow::channels::pact::{Exchange, Pipeline};
-    use timely::dataflow::operators::generic::{FrontieredInputHandle, builder_rc::OperatorBuilder};
+    use timely::dataflow::operators::generic::{builder_rc::OperatorBuilder, OutputBuilder};
 
     use crate::{lattice::Lattice, ExchangeData};
     use super::{Writer, Message, Progress};
@@ -583,7 +591,8 @@ pub mod sink {
         let mut builder = OperatorBuilder::new("UpdatesWriter".to_owned(), stream.scope());
         let reactivator = stream.scope().activator_for(builder.operator_info().address);
         let mut input = builder.new_input(stream, Pipeline);
-        let (mut updates_out, updates) = builder.new_output();
+        let (updates_out, updates) = builder.new_output();
+        let mut updates_out = OutputBuilder::from(updates_out);
 
         builder.build_reschedule(
             move |_capability| {
@@ -650,7 +659,6 @@ pub mod sink {
 
         builder.build_reschedule(|_capabilities| {
             move |frontiers| {
-                let mut input = FrontieredInputHandle::new(&mut input, &frontiers[0]);
 
                 // We want to drain inputs no matter what.
                 // We could do this after the next step, as we are certain these timestamps will
@@ -667,9 +675,9 @@ pub mod sink {
                         // If our frontier advances strictly, we have the opportunity to issue a progress statement.
                         if <_ as PartialOrder>::less_than(
                             &frontier.borrow(),
-                            &input.frontier.frontier(),
+                            &frontiers[0].frontier(),
                         ) {
-                            let new_frontier = input.frontier.frontier();
+                            let new_frontier = frontiers[0].frontier();
 
                             // Extract the timestamp counts to announce.
                             let mut announce = Vec::new();
@@ -691,7 +699,7 @@ pub mod sink {
                             send_queue.push_back(Message::Progress(progress));
 
                             // Advance our frontier to track our progress utterance.
-                            frontier = input.frontier.frontier().to_owned();
+                            frontier = frontiers[0].frontier().to_owned();
 
                             while let Some(message) = send_queue.front() {
                                 if let Some(duration) = sink.poll(message) {

differential-dataflow/src/capture.rs

@@ -207,7 +207,7 @@ impl<G: Scope, D, R, C: Container> Collection<G, D, R, C> {
     // TODO: Removing this function is possible, but breaks existing callers of `negate` who expect
     //       an inherent method on `Collection`.
     pub fn negate(&self) -> Collection<G, D, R, C>
-    where 
+    where
         StreamCore<G, C>: crate::operators::Negate<G, C>
     {
         crate::operators::Negate::negate(&self.inner).as_collection()
@@ -455,6 +455,7 @@ impl<G: Scope, D: Clone+'static, R: Clone+'static> Collection<G, D, R> {
     /// to all of the data timestamps).
     pub fn delay<F>(&self, func: F) -> Collection<G, D, R>
     where
+        G::Timestamp: Hash,
         F: FnMut(&G::Timestamp) -> G::Timestamp + Clone + 'static,
     {
         let mut func1 = func.clone();

differential-dataflow/src/collection.rs

@@ -268,8 +268,6 @@ impl<T: Columnation> PushInto<&&T> for TimelyStack<T> {
 }
 
 mod container {
-    use std::ops::Deref;
-
     use columnation::Columnation;
 
     use crate::containers::TimelyStack;
@@ -278,11 +276,6 @@ mod container {
         #[inline] fn record_count(&self) -> i64 { i64::try_from(self.local.len()).unwrap() }
         #[inline] fn is_empty(&self) -> bool { self.local.is_empty() }
     }
-    impl<T: Columnation> timely::container::IterContainer for TimelyStack<T> {
-        type ItemRef<'a> = &'a T where Self: 'a;
-        type Iter<'a> = std::slice::Iter<'a, T> where Self: 'a;
-        #[inline] fn iter(&self) -> Self::Iter<'_> { self.deref().iter() }
-    }
     impl<T: Columnation> timely::container::DrainContainer for TimelyStack<T> {
         type Item<'a> = &'a T where Self: 'a;
         type DrainIter<'a> = std::slice::Iter<'a, T> where Self: 'a;