wip

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

Author: antiguru

container/src/lib.rs

@@ -11,12 +11,7 @@ pub mod columnation;
 ///
 /// A container must implement default. The default implementation is not required to allocate
 /// memory for variable-length components.
-///
-/// We require the container to be cloneable to enable efficient copies when providing references
-/// of containers to operators. Care must be taken that the type's `clone_from` implementation
-/// is efficient (which is not necessarily the case when deriving `Clone`.)
-/// TODO: Don't require `Container: Clone`
-pub trait Container: Default + Clone + 'static {
+pub trait Container: Default + 'static {
     /// The type of elements this container holds.
     type Item;
 
@@ -40,7 +35,7 @@ pub trait Container: Default + Clone + 'static {
     fn clear(&mut self);
 }
 
-impl<T: Clone + 'static> Container for Vec<T> {
+impl<T: 'static> Container for Vec<T> {
     type Item = T;
 
     fn len(&self) -> usize {
@@ -118,7 +113,7 @@ pub trait PushPartitioned: Container {
         F: FnMut(usize, &mut Self);
 }
 
-impl<T: Clone + 'static> PushPartitioned for Vec<T> {
+impl<T: 'static> PushPartitioned for Vec<T> {
     fn push_partitioned<I, F>(&mut self, buffers: &mut [Self], mut index: I, mut flush: F)
     where
         I: FnMut(&Self::Item) -> usize,

kafkaesque/src/kafka_source.rs

@@ -5,7 +5,7 @@ use timely::dataflow::operators::generic::OutputHandle;
 
 use rdkafka::Message;
 use rdkafka::consumer::{ConsumerContext, BaseConsumer};
-use timely::dataflow::channels::pushers::tee::PushOwned;
+use timely::dataflow::channels::pushers::PushOwned;
 use timely::dataflow::stream::OwnedStream;
 
 /// Constructs a stream of data from a Kafka consumer.

timely/src/dataflow/channels/pact.rs

@@ -33,8 +33,8 @@ pub trait ParallelizationContractCore<T, D> {
 
 /// A `ParallelizationContractCore` specialized for `Vec` containers
 /// TODO: Use trait aliases once stable.
-pub trait ParallelizationContract<T, D: Clone>: ParallelizationContractCore<T, Vec<D>> { }
-impl<T, D: Clone, P: ParallelizationContractCore<T, Vec<D>>> ParallelizationContract<T, D> for P { }
+pub trait ParallelizationContract<T, D>: ParallelizationContractCore<T, Vec<D>> { }
+impl<T, D, P: ParallelizationContractCore<T, Vec<D>>> ParallelizationContract<T, D> for P { }
 
 /// A direct connection
 #[derive(Debug)]
@@ -71,7 +71,7 @@ impl<C, D, F: FnMut(&D)->u64+'static> ExchangeCore<C, D, F> {
 // Exchange uses a `Box<Pushable>` because it cannot know what type of pushable will return from the allocator.
 impl<T: Timestamp, C, D: Data+Clone, F: FnMut(&D)->u64+'static> ParallelizationContractCore<T, C> for ExchangeCore<C, D, F>
 where
-    C: Data + Container + PushPartitioned<Item=D>,
+    C: Data + Clone + PushPartitioned<Item=D>,
 {
     type Pusher = ExchangePusher<T, C, D, LogPusher<T, C, Box<dyn Push<BundleCore<T, C>>>>, F>;
     type Puller = LogPuller<T, C, Box<dyn Pull<BundleCore<T, C>>>>;

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

@@ -40,7 +40,7 @@ impl<T: Clone, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) ->
     }
 }
 
-impl<T: Eq+Data, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) -> u64> Push<BundleCore<T, C>> for Exchange<T, C, D, P, H>
+impl<T: Eq+Data, C: Container + Data, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) -> u64> Push<BundleCore<T, C>> for Exchange<T, C, D, P, H>
 where
     C: PushPartitioned<Item=D>
 {

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

@@ -1,7 +1,9 @@
+pub use self::owned::PushOwned;
 pub use self::tee::{Tee, TeeCore, TeeHelper};
 pub use self::exchange::Exchange;
 pub use self::counter::{Counter, CounterCore};
 
+pub mod owned;
 pub mod tee;
 pub mod exchange;
 pub mod counter;

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

@@ -0,0 +1,49 @@
+//! A `Push` implementor with a single target.
+
+use std::cell::RefCell;
+use std::fmt;
+use std::rc::Rc;
+
+use timely_communication::Push;
+use crate::{Data, Container};
+
+use crate::dataflow::channels::BundleCore;
+
+/// A pusher that can bind to a single downstream pusher.
+pub struct PushOwned<T, D>(Rc<RefCell<Option<Box<dyn Push<BundleCore<T, D>>>>>>);
+
+impl<T, D> PushOwned<T, D> {
+    /// Create a new `PushOwned`. Similarly to `Tee`, it returns a pair where either element
+    /// can be used as pusher or registrar.
+    pub fn new() -> (Self, Self) {
+        let zelf = Self(Rc::new(RefCell::new(None)));
+        (zelf.clone(), zelf)
+    }
+
+    /// Set the downstream pusher.
+    pub fn set<P: Push<BundleCore<T, D>> + 'static>(self, pusher: P) {
+        *self.0.borrow_mut() = Some(Box::new(pusher));
+    }
+}
+
+impl<T, D> fmt::Debug for PushOwned<T, D> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("PushOwned").finish_non_exhaustive()
+    }
+}
+
+impl<T, D> Clone for PushOwned<T, D> {
+    fn clone(&self) -> Self {
+        Self(Rc::clone(&self.0))
+    }
+}
+
+impl<T: Data, D: Container> Push<BundleCore<T, D>> for PushOwned<T, D> {
+    #[inline]
+    fn push(&mut self, message: &mut Option<BundleCore<T, D>>) {
+        let mut pusher = self.0.borrow_mut();
+        if let Some(pusher) = pusher.as_mut() {
+            pusher.push(message);
+        }
+    }
+}

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

@@ -1,7 +1,7 @@
 //! A `Push` implementor with a list of `Box<Push>` to forward pushes to.
 
 use std::cell::RefCell;
-use std::fmt::{self, Debug, Formatter};
+use std::fmt::{self, Debug};
 use std::rc::Rc;
 
 use crate::dataflow::channels::{BundleCore, Message};
@@ -10,39 +10,6 @@ use crate::communication::Push;
 use crate::{Container, Data};
 
 type PushList<T, D> = Rc<RefCell<Vec<Box<dyn Push<BundleCore<T, D>>>>>>;
-/// TODO
-pub struct PushOwned<T, D>(Rc<RefCell<Option<Box<dyn Push<BundleCore<T, D>>>>>>);
-
-impl<T, D> PushOwned<T, D> {
-    /// TODO
-    pub fn new() -> (Self, Self) {
-        let zelf = Self(Rc::new(RefCell::new(None)));
-        (zelf.clone(), zelf)
-    }
-
-    /// TODO
-    pub fn set<P: Push<BundleCore<T, D>> + 'static>(&self, pusher: P) {
-        *self.0.borrow_mut() = Some(Box::new(pusher));
-    }
-}
-
-impl<T, D> Default for PushOwned<T, D> {
-    fn default() -> Self {
-        Self(Rc::new(RefCell::new(None)))
-    }
-}
-
-impl<T, D> Debug for PushOwned<T, D> {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        f.debug_struct("PushOwned").finish_non_exhaustive()
-    }
-}
-
-impl<T, D> Clone for PushOwned<T, D> {
-    fn clone(&self) -> Self {
-        Self(Rc::clone(&self.0))
-    }
-}
 
 /// Wraps a shared list of `Box<Push>` to forward pushes to. Owned by `Stream`.
 pub struct TeeCore<T, D> {
@@ -53,18 +20,7 @@ pub struct TeeCore<T, D> {
 /// [TeeCore] specialized to `Vec`-based container.
 pub type Tee<T, D> = TeeCore<T, Vec<D>>;
 
-impl<T: Data, D: Container> Push<BundleCore<T, D>> for PushOwned<T, D> {
-    #[inline]
-    fn push(&mut self, message: &mut Option<BundleCore<T, D>>) {
-        let mut pusher = self.0.borrow_mut();
-        if let Some(pusher) = pusher.as_mut() {
-            pusher.push(message);
-        }
-    }
-}
-
-
-impl<T: Data, D: Container> Push<BundleCore<T, D>> for TeeCore<T, D> {
+impl<T: Data, D: Container+Clone> Push<BundleCore<T, D>> for TeeCore<T, D> {
     #[inline]
     fn push(&mut self, message: &mut Option<BundleCore<T, D>>) {
         let mut pushers = self.shared.borrow_mut();

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

@@ -3,10 +3,9 @@ use std::hash::Hash;
 use std::collections::HashMap;
 
 use crate::{Data, ExchangeData};
-use crate::dataflow::{Stream, Scope};
+use crate::dataflow::{Scope, StreamLike, OwnedStream};
 use crate::dataflow::operators::generic::operator::Operator;
 use crate::dataflow::channels::pact::Exchange;
-use crate::dataflow::stream::OwnedStream;
 
 /// Generic intra-timestamp aggregation
 ///
@@ -68,13 +67,19 @@ pub trait Aggregate<S: Scope, K: ExchangeData+Hash, V: ExchangeData> {
         hash: H) -> OwnedStream<S, Vec<R>> where S::Timestamp: Eq;
 }
 
-impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> Aggregate<S, K, V> for Stream<S, (K, V)> {
+impl<G, K, V, S> Aggregate<G, K, V> for S
+where
+    G: Scope,
+    K: ExchangeData + Hash + Eq + Clone,
+    V: ExchangeData,
+    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>(
         self,
         fold: F,
         emit: E,
-        hash: H) -> OwnedStream<S, Vec<R>> where S::Timestamp: Eq {
+        hash: H) -> OwnedStream<G, Vec<R>> where G::Timestamp: Eq {
 
         let mut aggregates = HashMap::new();
         let mut vector = Vec::new();

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

@@ -3,10 +3,9 @@ use std::hash::Hash;
 use std::collections::HashMap;
 
 use crate::{Data, ExchangeData};
-use crate::dataflow::{Stream, Scope};
+use crate::dataflow::{OwnedStream, Scope, StreamLike};
 use crate::dataflow::operators::generic::operator::Operator;
 use crate::dataflow::channels::pact::Exchange;
-use crate::dataflow::stream::OwnedStream;
 
 /// Generic state-transition machinery: each key has a state, and receives a sequence of events.
 /// Events are applied in time-order, but no other promises are made. Each state transition can
@@ -55,14 +54,20 @@ pub trait StateMachine<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> {
     >(self, fold: F, hash: H) -> OwnedStream<S, Vec<R>> where S::Timestamp : Hash+Eq ;
 }
 
-impl<S: Scope, K: ExchangeData+Hash+Eq, V: ExchangeData> StateMachine<S, K, V> for Stream<S, (K, V)> {
+impl<G, K, V, S> StateMachine<G, K, V> for S
+where
+    G: Scope,
+    K: ExchangeData + Hash + Eq + Clone,
+    V: ExchangeData,
+    S: StreamLike<G, Vec<(K, V)>>,
+{
     fn state_machine<
             R: Data,                                    // 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
             H: Fn(&K)->u64+'static,                     // "hash" function for keys
-        >(self, fold: F, hash: H) -> OwnedStream<S, Vec<R>> where S::Timestamp : Hash+Eq {
+        >(self, fold: F, hash: H) -> OwnedStream<G, Vec<R>> where G::Timestamp : Hash+Eq {
 
         let mut pending: HashMap<_, Vec<(K, V)>> = HashMap::new();   // times -> (keys -> state)
         let mut states = HashMap::new();    // keys -> state

timely/src/dataflow/operators/branch.rs

@@ -2,9 +2,8 @@
 
 use crate::dataflow::channels::pact::Pipeline;
 use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
-use crate::dataflow::{Scope, Stream, StreamCore};
+use crate::dataflow::{Scope, OwnedStream, StreamLike};
 use crate::{Container, Data};
-use crate::dataflow::stream::{OwnedStream, StreamLike};
 
 /// Extension trait for `Stream`.
 pub trait Branch<S: Scope, D: Data> {
@@ -35,11 +34,11 @@ pub trait Branch<S: Scope, D: Data> {
     ) -> (OwnedStream<S, Vec<D>>, OwnedStream<S, Vec<D>>);
 }
 
-impl<S: Scope, D: Data> Branch<S, D> for Stream<S, D> {
+impl<G: Scope, D: Data, S: StreamLike<G, Vec<D>>> Branch<G, D> for S {
     fn branch(
         self,
-        condition: impl Fn(&S::Timestamp, &D) -> bool + 'static,
-    ) -> (OwnedStream<S, Vec<D>>, OwnedStream<S, Vec<D>>) {
+        condition: impl Fn(&G::Timestamp, &D) -> bool + 'static,
+    ) -> (OwnedStream<G, Vec<D>>, OwnedStream<G, Vec<D>>) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
 
         let mut input = builder.new_input(self, Pipeline);
@@ -95,7 +94,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, S: StreamLike<G, C>> BranchWhen<G, C> for S {
+impl<G: Scope, C: Container + Clone, 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());