Updates for timely 0.25 (#647)

* Updates for timely 0.25

* Restore proper Cargo.toml

Signed-off-by: Moritz Hoffmann <[email protected]>

* Update columnar example to timely 0.25

---------

Signed-off-by: Moritz Hoffmann <[email protected]>
Co-authored-by: Moritz Hoffmann <[email protected]>

Author: frankmcsherry

Cargo.toml

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

differential-dataflow/examples/columnar.rs

@@ -298,7 +298,7 @@ pub mod storage {
     pub mod val {
 
         use std::fmt::Debug;
-        use columnar::{Container, ContainerOf, Index, Len, Push};
+        use columnar::{Borrow, Container, ContainerOf, Index, Len, Push};
         use columnar::Vecs;
 
         use crate::layout::ColumnarUpdate as Update;
@@ -406,7 +406,7 @@ pub mod storage {
 
     pub mod key {
 
-        use columnar::{Container, ContainerOf, Index, Len, Push};
+        use columnar::{Borrow, Container, ContainerOf, Index, Len, Push};
         use columnar::Vecs;
 
         use crate::layout::ColumnarUpdate as Update;
@@ -532,7 +532,7 @@ mod column_builder {
                 self.current.push(item);
                 if self.current.len() > 1024 * 1024 {
                     // TODO: Consolidate the batch?
-                    use columnar::{Container, Index};
+                    use columnar::{Borrow, Index};
                     let mut refs = self.current.borrow().into_index_iter().collect::<Vec<_>>();
                     refs.sort();
                     let storage = ValStorage::form(refs.into_iter());
@@ -570,7 +570,7 @@ mod column_builder {
             fn finish(&mut self) -> Option<&mut Self::Container> {
                 if !self.current.is_empty() {
                     // TODO: Consolidate the batch?
-                    use columnar::{Container, Index};
+                    use columnar::{Borrow, Index};
                     let mut refs = self.current.borrow().into_index_iter().collect::<Vec<_>>();
                     refs.sort();
                     let storage = ValStorage::form(refs.into_iter());
@@ -612,7 +612,7 @@ mod column_builder {
                 self.current.push(item);
                 if self.current.len() > 1024 * 1024 {
                     // TODO: Consolidate the batch?
-                    use columnar::{Container, Index};
+                    use columnar::{Borrow, Index};
                     let mut refs = self.current.borrow().into_index_iter().collect::<Vec<_>>();
                     refs.sort();
                     let storage = KeyStorage::form(refs.into_iter());
@@ -642,7 +642,7 @@ mod column_builder {
             fn finish(&mut self) -> Option<&mut Self::Container> {
                 if !self.current.is_empty() {
                     // TODO: Consolidate the batch?
-                    use columnar::{Container, Index};
+                    use columnar::{Borrow, Index};
                     let mut refs = self.current.borrow().into_index_iter().collect::<Vec<_>>();
                     refs.sort();
                     let storage = KeyStorage::form(refs.into_iter());
@@ -763,7 +763,7 @@ pub mod arrangement {
     pub use batch_container::Coltainer;
     pub mod batch_container {
 
-        use columnar::{Columnar, Container, Clear, Push, Index, Len};
+        use columnar::{Borrow, Columnar, Container, Clear, Push, Index, Len};
         use differential_dataflow::trace::implementations::BatchContainer;
 
         /// Container, anchored by `C` to provide an owned type.
@@ -815,7 +815,7 @@ pub mod arrangement {
     pub mod batcher {
 
         use std::ops::Range;
-        use columnar::{Columnar, Container, Index, Len, Push};
+        use columnar::{Borrow, Columnar, Container, Index, Len, Push};
         use differential_dataflow::trace::implementations::chainless_batcher as chainless;
         use differential_dataflow::difference::{Semigroup, IsZero};
         use timely::progress::frontier::{Antichain, AntichainRef};

differential-dataflow/examples/iterate_container.rs

@@ -40,8 +40,7 @@ fn wrap<G: Scope, C: timely::Container>(stream: &StreamCore<G, C>) -> StreamCore
     builder.build(move |_capability| move |_frontier| {
         let mut output = output.activate();
         input.for_each(|time, data| {
-            let mut session = output.session(&time);
-            session.give_container(&mut ContainerWrapper(std::mem::take(data)));
+            output.give(&time, &mut ContainerWrapper(std::mem::take(data)));
         });
     });
     stream_out

differential-dataflow/examples/multitemporal.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/progress.rs

@@ -121,7 +121,7 @@ fn frontier<G, T>(
 ) -> VecCollection<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/src/algorithms/graphs/propagate.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: &VecCollection<G, (N,N), R>, nodes: &VecCollection<G,(N,L),R>) -> VecCollection<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: &VecCollection<G, (N,N), R>, nodes: &VecCollection<G,(N,L),R>, logic: F) -> VecCollection<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/scc.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: &VecCollection<G, (N,N), R>) -> VecCollection<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: &VecCollection<G, (N,N), R>, edges: &VecCollection<G, (N,N), R>)
     -> VecCollection<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/sequential.rs

@@ -11,7 +11,7 @@ use crate::hashable::Hashable;
 
 fn _color<G, N>(edges: &VecCollection<G, (N,N)>) -> VecCollection<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/capture.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/collection.rs

@@ -496,6 +496,7 @@ impl<G: Scope, D: Clone+'static, R: Clone+'static> VecCollection<G, D, R> {
     /// to all of the data timestamps).
     pub fn delay<F>(&self, func: F) -> VecCollection<G, D, R>
     where
+        G::Timestamp: Hash,
         F: FnMut(&G::Timestamp) -> G::Timestamp + Clone + 'static,
     {
         let mut func1 = func.clone();