Address feedback

Signed-off-by: Moritz Hoffmann <antiguru@gmail.com>

src/operators/arrange/arrangement.rs

@@ -75,7 +75,8 @@ where
 
 use ::timely::dataflow::scopes::Child;
 use ::timely::progress::timestamp::Refines;
-use timely::container::{PushContainer, PushInto};
+use timely::Container;
+use timely::container::PushInto;
 
 impl<G, Tr> Arranged<G, Tr>
 where
@@ -293,7 +294,7 @@ where
         F: Fn(T2::Val<'_>) -> V + 'static,
         T2::Diff: Abelian,
         T2::Batch: Batch,
-        <T2::Builder as Builder>::Input: PushContainer,
+        <T2::Builder as Builder>::Input: Container,
         ((T1::KeyOwned, V), T2::Time, T2::Diff): PushInto<<T2::Builder as Builder>::Input>,
         L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(V, T2::Diff)>)+'static,
     {
@@ -313,7 +314,7 @@ where
         V: Data,
         F: Fn(T2::Val<'_>) -> V + 'static,
         T2::Batch: Batch,
-        <T2::Builder as Builder>::Input: PushContainer,
+        <T2::Builder as Builder>::Input: Container,
         ((T1::KeyOwned, V), T2::Time, T2::Diff): PushInto<<T2::Builder as Builder>::Input>,
         L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(V, T2::Diff)>, &mut Vec<(V, T2::Diff)>)+'static,
     {

src/operators/reduce.rs

@@ -5,7 +5,8 @@
 //! to the key and the list of values.
 //! The function is expected to populate a list of output values.
 
-use timely::container::{PushContainer, PushInto};
+use timely::Container;
+use timely::container::PushInto;
 use crate::hashable::Hashable;
 use crate::{Data, ExchangeData, Collection};
 use crate::difference::{Semigroup, Abelian};
@@ -313,7 +314,7 @@ where
     V: Data,
     F: Fn(T2::Val<'_>) -> V + 'static,
     T2::Batch: Batch,
-    <T2::Builder as Builder>::Input: PushContainer,
+    <T2::Builder as Builder>::Input: Container,
     ((T1::KeyOwned, V), T2::Time, T2::Diff): PushInto<<T2::Builder as Builder>::Input>,
     L: FnMut(T1::Key<'_>, &[(T1::Val<'_>, T1::Diff)], &mut Vec<(V,T2::Diff)>, &mut Vec<(V, T2::Diff)>)+'static,
 {
@@ -450,12 +451,14 @@ where
                         // TODO: It would be better if all updates went into one batch, but timely dataflow prevents
                         //       this as long as it requires that there is only one capability for each message.
                         let mut buffers = Vec::<(G::Timestamp, Vec<(V, G::Timestamp, T2::Diff)>)>::new();
-                        let mut chains = Vec::new();
+                        let mut builders = Vec::new();
                         for cap in capabilities.iter() {
                             buffers.push((cap.time().clone(), Vec::new()));
-                            chains.push(Default::default());
+                            builders.push(T2::Builder::new());
                         }
 
+                        let mut buffer = Default::default();
+
                         // cursors for navigating input and output traces.
                         let (mut source_cursor, source_storage): (T1::Cursor, _) = source_trace.cursor_through(lower_limit.borrow()).expect("failed to acquire source cursor");
                         let source_storage = &source_storage;
@@ -533,8 +536,9 @@ where
                             for index in 0 .. buffers.len() {
                                 buffers[index].1.sort_by(|x,y| x.0.cmp(&y.0));
                                 for (val, time, diff) in buffers[index].1.drain(..) {
-                                    // TODO(antiguru): This is dumb. Need to stage data and then reveal it.
-                                    ((key.into_owned(), val), time, diff).push_into(&mut chains[index]);
+                                    ((key.into_owned(), val), time, diff).push_into(&mut buffer);
+                                    builders[index].push(&mut buffer);
+                                    buffer.clear();
                                 }
                             }
                         }
@@ -546,10 +550,8 @@ where
                         output_lower.extend(lower_limit.borrow().iter().cloned());
 
                         // build and ship each batch (because only one capability per message).
-                        for (index, mut chain) in chains.drain(..).enumerate() {
-                            let mut builder = T2::Builder::new();
-                            // TODO(antiguru): Form actual chains.
-                            builder.push(&mut chain);
+                        for (index, builder) in builders.drain(..).enumerate() {
+
                             // Form the upper limit of the next batch, which includes all times greater
                             // than the input batch, or the capabilities from i + 1 onward.
                             output_upper.clear();

src/trace/implementations/mod.rs

@@ -161,8 +161,8 @@ impl<K,V,T,R> Update for Preferred<K, V, T, R>
 where
     K: ToOwned + ?Sized,
     K::Owned: Ord+Clone+'static,
-    V: ToOwned + ?Sized + 'static,
-    V::Owned: Ord+Clone,
+    V: ToOwned + ?Sized,
+    V::Owned: Ord+Clone+'static,
     T: Ord+Lattice+timely::progress::Timestamp+Clone,
     R: Semigroup+Clone,
 {
@@ -177,8 +177,8 @@ where
     K: Ord+ToOwned+PreferredContainer + ?Sized,
     K::Owned: Ord+Clone+'static,
     // for<'a> K::Container: BatchContainer<ReadItem<'a> = &'a K>,
-    V: Ord+ToOwned+PreferredContainer + ?Sized + 'static,
-    V::Owned: Ord+Clone,
+    V: Ord+ToOwned+PreferredContainer + ?Sized,
+    V::Owned: Ord+Clone+'static,
     T: Ord+Lattice+timely::progress::Timestamp+Clone,
     D: Semigroup+Clone,
 {
@@ -192,6 +192,7 @@ where
 use std::convert::TryInto;
 use std::ops::Deref;
 use abomonation_derive::Abomonation;
+use timely::Container;
 use timely::container::PushInto;
 use timely::progress::Timestamp;
 use crate::trace::cursor::MyTrait;
@@ -364,9 +365,7 @@ impl BatchContainer for OffsetList {
 }
 
 /// Behavior to split an update into principal components.
-pub trait BuilderInput<L: Layout> {
-    /// The item to break apart.
-    type Item<'a>;
+pub trait BuilderInput<L: Layout>: Container {
     /// Key portion
     type Key<'a>: Ord;
     /// Value portion
@@ -393,7 +392,6 @@ where
     T: Timestamp + Lattice + Clone + 'static,
     R: Semigroup + Clone + 'static,
 {
-    type Item<'a> = ((K, V), T, R);
     type Key<'a> = K;
     type Val<'a> = V;
     type Time = T;
@@ -419,7 +417,6 @@ where
     T: Timestamp + Lattice + Columnation + Clone + 'static,
     R: Semigroup + Columnation + Clone + 'static,
 {
-    type Item<'a> = &'a ((K, V), T, R);
     type Key<'a> = &'a K;
     type Val<'a> = &'a V;
     type Time = T;
@@ -442,12 +439,11 @@ impl<K,V,T,R> BuilderInput<Preferred<K, V, T, R>> for TimelyStack<((<K as ToOwne
 where
     K: Ord+ToOwned+PreferredContainer + ?Sized,
     K::Owned: Columnation + Ord+Clone+'static,
-    V: Ord+ToOwned+PreferredContainer + ?Sized + 'static,
-    V::Owned: Columnation + Ord+Clone,
-    T: Columnation + Ord+Lattice+timely::progress::Timestamp+Clone,
+    V: Ord+ToOwned+PreferredContainer + ?Sized,
+    V::Owned: Columnation + Ord+Clone+'static,
+    T: Columnation + Ord+Lattice+Timestamp+Clone,
     R: Columnation + Semigroup+Clone,
 {
-    type Item<'a> = &'a ((<K as ToOwned>::Owned, <V as ToOwned>::Owned), T, R);
     type Key<'a> = &'a K::Owned;
     type Val<'a> = &'a V::Owned;
     type Time = T;

src/trace/implementations/ord_neu.rs

@@ -69,7 +69,6 @@ mod val_batch {
 
     use std::marker::PhantomData;
     use abomonation_derive::Abomonation;
-    use timely::Container;
     use timely::container::PushInto;
     use timely::progress::{Antichain, frontier::AntichainRef};
 
@@ -544,7 +543,7 @@ mod val_batch {
     impl<L, CI> Builder for OrdValBuilder<L, CI>
     where
         L: Layout,
-        CI: Container + for<'a> BuilderInput<L, Item<'a> = <CI as Container>::Item<'a>, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
+        CI: for<'a> BuilderInput<L, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
         for<'a> CI::Key<'a>: PushInto<L::KeyContainer>,
         for<'a> CI::Val<'a>: PushInto<L::ValContainer>,
     {
@@ -618,7 +617,6 @@ mod key_batch {
 
     use std::marker::PhantomData;
     use abomonation_derive::Abomonation;
-    use timely::Container;
     use timely::container::PushInto;
     use timely::progress::{Antichain, frontier::AntichainRef};
 
@@ -990,7 +988,7 @@ mod key_batch {
     impl<L: Layout, CI> Builder for OrdKeyBuilder<L, CI>
     where
         L: Layout,
-        CI: Container + for<'a> BuilderInput<L, Item<'a> = <CI as Container>::Item<'a>, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
+        CI: for<'a> BuilderInput<L, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
         for<'a> CI::Key<'a>: PushInto<L::KeyContainer>,
     {
 

src/trace/implementations/rhh.rs

@@ -80,7 +80,6 @@ mod val_batch {
     use std::convert::TryInto;
     use std::marker::PhantomData;
     use abomonation_derive::Abomonation;
-    use timely::Container;
     use timely::container::PushInto;
     use timely::progress::{Antichain, frontier::AntichainRef};
 
@@ -740,7 +739,7 @@ mod val_batch {
     where
         <L::Target as Update>::Key: Default + HashOrdered,
         // RhhValBatch<L>: Batch<Key=<L::Target as Update>::Key, Val=<L::Target as Update>::Val, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
-        CI: Container + for<'a> BuilderInput<L, Item<'a> = <CI as Container>::Item<'a>, Key<'a> = <L::Target as Update>::Key, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
+        CI: for<'a> BuilderInput<L, Key<'a> = <L::Target as Update>::Key, Time=<L::Target as Update>::Time, Diff=<L::Target as Update>::Diff>,
         for<'a> CI::Val<'a>: PushInto<L::ValContainer>,
     {
         type Input = CI;

src/trace/mod.rs

@@ -336,10 +336,9 @@ pub trait Builder: Sized {
     ///
     /// They represent respectively the number of distinct `key`, `(key, val)`, and total updates.
     fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self;
-    /// Adds an element to the batch.
+    /// Adds a chunk of elements to the batch.
     ///
-    /// The default implementation uses `self.copy` with references to the owned arguments.
-    /// One should override it if the builder can take advantage of owned arguments.
+    /// Adds all elements from `chunk` to the builder and leaves `chunk` in an undefined state.
     fn push(&mut self, chunk: &mut Self::Input);
     /// Completes building and returns the batch.
     fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Self::Output;