Cleanup, use MergerChunk to clear in reduce

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

Author: antiguru

differential-dataflow/examples/columnar.rs

@@ -392,7 +392,7 @@ pub mod batcher {
         ready: VecDeque<C>,
     }
 
-    impl<C: Container + Clone + 'static> ContainerBuilder for Chunker<C> {
+    impl<C: Container> ContainerBuilder for Chunker<C> {
         type Container = C;
 
         fn extract(&mut self) -> Option<&mut Self::Container> {

differential-dataflow/src/collection.rs

@@ -68,7 +68,7 @@ impl<G: Scope, D, R, C> Collection<G, D, R, C> {
         Collection { inner: stream, phantom: std::marker::PhantomData }
     }
 }
-impl<G: Scope, D, R, C: Container + Clone + 'static> Collection<G, D, R, C> {
+impl<G: Scope, D, R, C: Container> Collection<G, D, R, C> {
     /// Creates a new collection accumulating the contents of the two collections.
     ///
     /// Despite the name, differential dataflow collections are unordered. This method is so named because the
@@ -686,7 +686,7 @@ where
     G: Scope,
     D: Data,
     R: Semigroup + 'static,
-    C: Container + Clone + 'static,
+    C: Container,
     I: IntoIterator<Item=Collection<G, D, R, C>>,
 {
     scope

differential-dataflow/src/operators/arrange/arrangement.rs

@@ -31,6 +31,7 @@ use crate::difference::Semigroup;
 use crate::lattice::Lattice;
 use crate::trace::{self, Trace, TraceReader, BatchReader, Batcher, Builder, Cursor};
 use crate::trace::implementations::{KeyBatcher, KeyBuilder, KeySpine, ValBatcher, ValBuilder, ValSpine};
+use crate::trace::implementations::merge_batcher::container::MergerChunk;
 
 use trace::wrappers::enter::{TraceEnter, BatchEnter,};
 use trace::wrappers::enter_at::TraceEnter as TraceEnterAt;
@@ -293,7 +294,7 @@ where
             Time=T1::Time,
             Diff: Abelian,
         >+'static,
-        Bu: Builder<Time=G::Timestamp, Output = T2::Batch, Input: Container + PushInto<((T1::KeyOwn, T2::ValOwn), T2::Time, T2::Diff)>>,
+        Bu: Builder<Time=G::Timestamp, Output = T2::Batch, Input: MergerChunk + PushInto<((T1::KeyOwn, T2::ValOwn), T2::Time, T2::Diff)>>,
         L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(T2::ValOwn, T2::Diff)>)+'static,
     {
         self.reduce_core::<_,Bu,T2>(name, move |key, input, output, change| {
@@ -315,7 +316,7 @@ where
             ValOwn: Data,
             Time=T1::Time,
         >+'static,
-        Bu: Builder<Time=G::Timestamp, Output = T2::Batch, Input: Container + PushInto<((T1::KeyOwn, T2::ValOwn), T2::Time, T2::Diff)>>,
+        Bu: Builder<Time=G::Timestamp, Output = T2::Batch, Input: MergerChunk + PushInto<((T1::KeyOwn, T2::ValOwn), T2::Time, T2::Diff)>>,
         L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(T2::ValOwn, T2::Diff)>, &mut Vec<(T2::ValOwn, T2::Diff)>)+'static,
     {
         use crate::operators::reduce::reduce_trace;
@@ -393,7 +394,7 @@ pub fn arrange_core<G, P, Ba, Bu, Tr>(stream: &StreamCore<G, Ba::Input>, pact: P
 where
     G: Scope<Timestamp: Lattice>,
     P: ParallelizationContract<G::Timestamp, Ba::Input>,
-    Ba: Batcher<Time=G::Timestamp,Input: Container + Clone + 'static> + 'static,
+    Ba: Batcher<Time=G::Timestamp,Input: Container> + 'static,
     Bu: Builder<Time=G::Timestamp, Input=Ba::Output, Output = Tr::Batch>,
     Tr: Trace<Time=G::Timestamp>+'static,
 {

differential-dataflow/src/operators/iterate.rs

@@ -156,15 +156,15 @@ where
     G: Scope<Timestamp: Lattice>,
     D: Data,
     R: Abelian + 'static,
-    C: Container + Clone + 'static,
+    C: Container,
 {
     collection: Collection<G, D, R, C>,
     feedback: Handle<G, C>,
     source: Option<Collection<G, D, R, C>>,
     step: <G::Timestamp as Timestamp>::Summary,
 }
 
-impl<G, D: Data, R: Abelian, C: Container + Clone + 'static> Variable<G, D, R, C>
+impl<G, D: Data, R: Abelian, C: Container> Variable<G, D, R, C>
 where
     G: Scope<Timestamp: Lattice>,
     StreamCore<G, C>: crate::operators::Negate<G, C> + ResultsIn<G, C>,
@@ -218,7 +218,7 @@ where
     }
 }
 
-impl<G: Scope<Timestamp: Lattice>, D: Data, R: Abelian, C: Container + Clone + 'static> Deref for Variable<G, D, R, C> {
+impl<G: Scope<Timestamp: Lattice>, D: Data, R: Abelian, C: Container> Deref for Variable<G, D, R, C> {
     type Target = Collection<G, D, R, C>;
     fn deref(&self) -> &Self::Target {
         &self.collection
@@ -236,14 +236,14 @@ where
     G: Scope<Timestamp: Lattice>,
     D: Data,
     R: Semigroup + 'static,
-    C: Container + Clone + 'static,
+    C: Container,
 {
     collection: Collection<G, D, R, C>,
     feedback: Handle<G, C>,
     step: <G::Timestamp as Timestamp>::Summary,
 }
 
-impl<G, D: Data, R: Semigroup, C: Container+Clone> SemigroupVariable<G, D, R, C>
+impl<G, D: Data, R: Semigroup, C: Container> SemigroupVariable<G, D, R, C>
 where
     G: Scope<Timestamp: Lattice>,
     StreamCore<G, C>: ResultsIn<G, C>,
@@ -267,7 +267,7 @@ where
     }
 }
 
-impl<G: Scope, D: Data, R: Semigroup, C: Container+Clone+'static> Deref for SemigroupVariable<G, D, R, C> where G::Timestamp: Lattice {
+impl<G: Scope, D: Data, R: Semigroup, C: Container> Deref for SemigroupVariable<G, D, R, C> where G::Timestamp: Lattice {
     type Target = Collection<G, D, R, C>;
     fn deref(&self) -> &Self::Target {
         &self.collection

differential-dataflow/src/operators/join.rs

@@ -362,7 +362,7 @@ where
     T1: TraceReader+Clone+'static,
     T2: for<'a> TraceReader<Key<'a>=T1::Key<'a>, Time=T1::Time>+Clone+'static,
     L: FnMut(T1::Key<'_>,T1::Val<'_>,T2::Val<'_>,&G::Timestamp,&T1::Diff,&T2::Diff,&mut JoinSession<T1::Time, CB, CB::Container>)+'static,
-    CB: ContainerBuilder + 'static,
+    CB: ContainerBuilder,
 {
     // Rename traces for symmetry from here on out.
     let mut trace1 = arranged1.trace.clone();

differential-dataflow/src/operators/reduce.rs

@@ -5,7 +5,6 @@
 //! to the key and the list of values.
 //! The function is expected to populate a list of output values.
 
-use timely::Container;
 use timely::container::PushInto;
 use crate::hashable::Hashable;
 use crate::{Data, ExchangeData, Collection};
@@ -25,7 +24,7 @@ use crate::trace::{BatchReader, Cursor, Trace, Builder, ExertionLogic, Descripti
 use crate::trace::cursor::CursorList;
 use crate::trace::implementations::{KeySpine, KeyBuilder, ValSpine, ValBuilder};
 use crate::trace::implementations::containers::BatchContainer;
-
+use crate::trace::implementations::merge_batcher::container::MergerChunk;
 use crate::trace::TraceReader;
 
 /// Extension trait for the `reduce` differential dataflow method.
@@ -315,7 +314,7 @@ where
     G: Scope<Timestamp=T1::Time>,
     T1: TraceReader<KeyOwn: Ord> + Clone + 'static,
     T2: for<'a> Trace<Key<'a>=T1::Key<'a>, KeyOwn=T1::KeyOwn, ValOwn: Data, Time=T1::Time> + 'static,
-    Bu: Builder<Time=T2::Time, Output = T2::Batch, Input: Container + PushInto<((T1::KeyOwn, T2::ValOwn), T2::Time, T2::Diff)>>,
+    Bu: Builder<Time=T2::Time, Output = T2::Batch, Input: MergerChunk + PushInto<((T1::KeyOwn, T2::ValOwn), T2::Time, T2::Diff)>>,
     L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(T2::ValOwn,T2::Diff)>, &mut Vec<(T2::ValOwn, T2::Diff)>)+'static,
 {
     let mut result_trace = None;
@@ -533,8 +532,7 @@ where
                                 for (val, time, diff) in buffers[index].1.drain(..) {
                                     buffer.push_into(((T1::owned_key(key), val), time, diff));
                                     builders[index].push(&mut buffer);
-                                    // TODO: Clear!
-                                    // buffer.clear();
+                                    buffer.clear();
                                 }
                             }
                         }

differential-dataflow/src/trace/implementations/merge_batcher.rs

@@ -48,7 +48,7 @@ pub struct MergeBatcher<Input, C, M: Merger> {
 
 impl<Input, C, M> Batcher for MergeBatcher<Input, C, M>
 where
-    C: ContainerBuilder<Container=M::Chunk> + Default + for<'a> PushInto<&'a mut Input>,
+    C: ContainerBuilder<Container=M::Chunk> + for<'a> PushInto<&'a mut Input>,
     M: Merger<Time: Timestamp>,
 {
     type Input = Input;
@@ -325,7 +325,7 @@ pub mod container {
 
     impl<MC, CQ> Merger for ContainerMerger<MC, CQ>
     where
-        for<'a> MC: MergerChunk<TimeOwned: Ord + PartialOrder + Data> + Clone + Default + PushInto<<MC as DrainContainer>::Item<'a>> + 'static,
+        for<'a> MC: MergerChunk<TimeOwned: Ord + PartialOrder + Data> + Clone + PushInto<<MC as DrainContainer>::Item<'a>> + 'static,
         CQ: ContainerQueue<MC>,
     {
         type Time = MC::TimeOwned;