Remove Data trait

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

Author: antiguru

timely/src/dataflow/channels/pact.rs

@@ -18,7 +18,6 @@ use crate::dataflow::channels::Message;
 use crate::logging::{TimelyLogger as Logger, MessagesEvent};
 use crate::progress::Timestamp;
 use crate::worker::AsWorker;
-use crate::Data;
 
 /// A `ParallelizationContract` allocates paired `Push` and `Pull` implementors.
 pub trait ParallelizationContract<T, C> {
@@ -84,7 +83,7 @@ impl<T: Timestamp, CB, H: 'static> ParallelizationContract<T, CB::Container> for
 where
     CB: ContainerBuilder,
     CB: for<'a> PushInto<<CB::Container as Container>::Item<'a>>,
-    CB::Container: Data + Send + SizableContainer + crate::dataflow::channels::ContainerBytes,
+    CB::Container: Send + SizableContainer + crate::dataflow::channels::ContainerBytes,
     for<'a> H: FnMut(&<CB::Container as Container>::Item<'a>) -> u64
 {
     type Pusher = ExchangePusher<T, CB, LogPusher<T, CB::Container, Box<dyn Push<Message<T, CB::Container>>>>, H>;

timely/src/dataflow/channels/pushers/exchange.rs

@@ -3,7 +3,7 @@
 use crate::communication::Push;
 use crate::container::{ContainerBuilder, SizableContainer, PushInto};
 use crate::dataflow::channels::Message;
-use crate::{Container, Data};
+use crate::Container;
 
 // TODO : Software write combining
 /// Distributes records among target pushees according to a distribution function.
@@ -50,7 +50,7 @@ where
     }
 }
 
-impl<T: Eq+Data, CB, P, H> Push<Message<T, CB::Container>> for Exchange<T, CB, P, H>
+impl<T: Eq+Clone, CB, P, H> Push<Message<T, CB::Container>> for Exchange<T, CB, P, H>
 where
     CB: ContainerBuilder,
     CB::Container: SizableContainer,

timely/src/dataflow/channels/pushers/owned.rs

@@ -5,7 +5,6 @@ use std::fmt;
 use std::rc::Rc;
 
 use timely_communication::Push;
-use crate::{Data, Container};
 use crate::dataflow::channels::Message;
 
 /// A pusher that can bind to a single downstream pusher.
@@ -37,7 +36,7 @@ impl<T, D> Clone for PushOwned<T, D> {
     }
 }
 
-impl<T: Data, D: Container> Push<Message<T, D>> for PushOwned<T, D> {
+impl<T, D> Push<Message<T, D>> for PushOwned<T, D> {
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, D>>) {
         let mut pusher = self.0.borrow_mut();

timely/src/dataflow/channels/pushers/tee.rs

@@ -7,7 +7,7 @@ use std::rc::Rc;
 use crate::dataflow::channels::Message;
 
 use crate::communication::Push;
-use crate::{Container, Data};
+use crate::Container;
 
 type PushList<T, C> = Rc<RefCell<Vec<Box<dyn Push<Message<T, C>>>>>>;
 
@@ -17,7 +17,7 @@ pub struct Tee<T, C> {
     shared: PushList<T, C>,
 }
 
-impl<T: Data, C: Container + Data> Push<Message<T, C>> for Tee<T, C> {
+impl<T: Clone + 'static, C: Container + Clone + 'static> Push<Message<T, C>> for Tee<T, C> {
     #[inline]
     fn push(&mut self, message: &mut Option<Message<T, C>>) {
         let mut pushers = self.shared.borrow_mut();

timely/src/dataflow/operators/aggregation/aggregate.rs

@@ -2,7 +2,7 @@
 use std::hash::Hash;
 use std::collections::HashMap;
 
-use crate::{Data, ExchangeData};
+use crate::ExchangeData;
 use crate::dataflow::{Scope, StreamLike, OwnedStream};
 use crate::dataflow::operators::generic::operator::Operator;
 use crate::dataflow::channels::pact::Exchange;
@@ -60,7 +60,7 @@ pub trait Aggregate<S: Scope, K: ExchangeData+Hash, V: ExchangeData> {
     ///         .inspect(|x| assert!(*x == (0, 5) || *x == (1, 5)));
     /// });
     /// ```
-    fn aggregate<R: Data, D: Default+'static, F: Fn(&K, V, &mut D)+'static, E: Fn(K, D)->R+'static, H: Fn(&K)->u64+'static>(
+    fn aggregate<R: 'static, D: Default+'static, F: Fn(&K, V, &mut D)+'static, E: Fn(K, D)->R+'static, H: Fn(&K)->u64+'static>(
         self,
         fold: F,
         emit: E,
@@ -75,7 +75,7 @@ where
     S: StreamLike<G, Vec<(K, V)>>,
 {
 
-    fn aggregate<R: Data, D: Default+'static, F: Fn(&K, V, &mut D)+'static, E: Fn(K, D)->R+'static, H: Fn(&K)->u64+'static>(
+    fn aggregate<R: 'static, D: Default+'static, F: Fn(&K, V, &mut D)+'static, E: Fn(K, D)->R+'static, H: Fn(&K)->u64+'static>(
         self,
         fold: F,
         emit: E,

timely/src/dataflow/operators/aggregation/state_machine.rs

@@ -2,7 +2,7 @@
 use std::hash::Hash;
 use std::collections::HashMap;
 
-use crate::{Data, ExchangeData};
+use crate::ExchangeData;
 use crate::dataflow::{OwnedStream, Scope, StreamLike};
 use crate::dataflow::operators::generic::operator::Operator;
 use crate::dataflow::channels::pact::Exchange;
@@ -46,7 +46,7 @@ pub trait StateMachine<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> {
     /// });
     /// ```
     fn state_machine<
-        R: Data,                                    // output type
+        R: 'static,                                 // output type
         D: Default+'static,                         // per-key state (data)
         I: IntoIterator<Item=R>,                    // type of output iterator
         F: Fn(&K, V, &mut D)->(bool, I)+'static,    // state update logic
@@ -62,7 +62,7 @@ where
     S: StreamLike<G, Vec<(K, V)>>,
 {
     fn state_machine<
-            R: Data,                                    // output type
+            R: 'static,                                 // output type
             D: Default+'static,                         // per-key state (data)
             I: IntoIterator<Item=R>,                    // type of output iterator
             F: Fn(&K, V, &mut D)->(bool, I)+'static,    // state update logic

timely/src/dataflow/operators/branch.rs

@@ -3,10 +3,10 @@
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
 use crate::dataflow::{Scope, OwnedStream, StreamLike};
-use crate::{Container, Data};
+use crate::Container;
 
 /// Extension trait for `Stream`.
-pub trait Branch<S: Scope, D: Data> {
+pub trait Branch<S: Scope, D> {
     /// Takes one input stream and splits it into two output streams.
     /// For each record, the supplied closure is called with a reference to
     /// the data and its time. If it returns `true`, the record will be sent
@@ -34,7 +34,7 @@ pub trait Branch<S: Scope, D: Data> {
     ) -> (OwnedStream<S, Vec<D>>, OwnedStream<S, Vec<D>>);
 }
 
-impl<G: Scope, D: Data, S: StreamLike<G, Vec<D>>> Branch<G, D> for S {
+impl<G: Scope, D: 'static, S: StreamLike<G, Vec<D>>> Branch<G, D> for S {
     fn branch(
         self,
         condition: impl Fn(&G::Timestamp, &D) -> bool + 'static,
@@ -92,7 +92,7 @@ pub trait BranchWhen<G: Scope, C: Container>: Sized {
     fn branch_when(self, condition: impl Fn(&G::Timestamp) -> bool + 'static) -> (OwnedStream<G, C>, OwnedStream<G, C>);
 }
 
-impl<G: Scope, C: Container + Data, S: StreamLike<G, C>> BranchWhen<G, C> for S {
+impl<G: Scope, C: Container + 'static, S: StreamLike<G, C>> BranchWhen<G, C> for S {
     fn branch_when(self, condition: impl Fn(&G::Timestamp) -> bool + 'static) -> (OwnedStream<G, C>, OwnedStream<G, C>) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
 

timely/src/dataflow/operators/broadcast.rs

@@ -21,7 +21,7 @@ pub trait Broadcast<G: Scope, D: ExchangeData> {
     fn broadcast(self) -> OwnedStream<G, Vec<D>>;
 }
 
-impl<G: Scope, D: ExchangeData, S: StreamLike<G, Vec<D>>> Broadcast<G, D> for S {
+impl<G: Scope, D: ExchangeData + Clone, S: StreamLike<G, Vec<D>>> Broadcast<G, D> for S {
     fn broadcast(self) -> OwnedStream<G, Vec<D>> {
 
         // NOTE: Simplified implementation due to underlying motion

timely/src/dataflow/operators/core/enterleave.rs

@@ -25,7 +25,7 @@ use crate::logging::{TimelyLogger, MessagesEvent};
 use crate::progress::Timestamp;
 use crate::progress::timestamp::Refines;
 use crate::progress::{Source, Target};
-use crate::{Container, Data};
+use crate::Container;
 use crate::communication::Push;
 use crate::dataflow::channels::pushers::{Counter, PushOwned};
 use crate::dataflow::channels::Message;
@@ -53,7 +53,7 @@ pub trait Enter<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container> {
     fn enter<'a>(self, _: &Child<'a, G, T>) -> OwnedStream<Child<'a, G, T>, C>;
 }
 
-impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Data+Container, S: StreamLike<G, C>> Enter<G, T, C> for S {
+impl<G: Scope, T: Timestamp+Refines<G::Timestamp>, C: Container + 'static, S: StreamLike<G, C>> Enter<G, T, C> for S {
     fn enter<'a>(self, scope: &Child<'a, G, T>) -> OwnedStream<Child<'a, G, T>, C> {
 
         use crate::scheduling::Scheduler;
@@ -130,14 +130,14 @@ impl<'a, G: Scope, C: Container + 'static, T: Timestamp+Refines<G::Timestamp>, S
 }
 
 
-struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container + Data> {
+struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> {
     targets: Counter<TInner, TContainer, PushOwned<TInner, TContainer>>,
     phantom: PhantomData<TOuter>,
     activator: crate::scheduling::Activator,
     active: bool,
 }
 
-impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container + Data> Push<Message<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
+impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> Push<Message<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
     fn push(&mut self, element: &mut Option<Message<TOuter, TContainer>>) {
         if let Some(outer_message) = element {
             let data = ::std::mem::take(&mut outer_message.data);

timely/src/dataflow/operators/core/feedback.rs

@@ -1,6 +1,6 @@
 //! Create cycles in a timely dataflow graph.
 
-use crate::{Container, Data};
+use crate::Container;
 use crate::container::CapacityContainerBuilder;
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::channels::pushers::PushOwned;
@@ -85,7 +85,7 @@ impl<'a, G: Scope, T: Timestamp> LoopVariable<'a, G, T> for Iterative<'a, G, T>
 }
 
 /// Connect a `Stream` to the input of a loop variable.
-pub trait ConnectLoop<G: Scope, C: Container + Data> {
+pub trait ConnectLoop<G: Scope, C: Container> {
     /// Connect a `Stream` to be the input of a loop variable.
     ///
     /// # Examples
@@ -106,7 +106,7 @@ pub trait ConnectLoop<G: Scope, C: Container + Data> {
     fn connect_loop(self, handle: Handle<G, C>);
 }
 
-impl<G: Scope, C: Container + Data, S: StreamLike<G, C>> ConnectLoop<G, C> for S {
+impl<G: Scope, C: Container + 'static, S: StreamLike<G, C>> ConnectLoop<G, C> for S {
     fn connect_loop(self, handle: Handle<G, C>) {
 
         let mut builder = handle.builder;