Correct example, improve chunking

Author: frankmcsherry

differential-dataflow/examples/columnar.rs

@@ -94,7 +94,7 @@ fn main() {
                 buffer.clear();
                 i += worker.peers();
             }
-            data_input.send_batch(&mut container);
+            keys_input.send_batch(&mut container);
             container.clear();
             queries += size;
             data_input.advance_to(data_input.time() + 1);
@@ -414,7 +414,7 @@ pub mod batcher {
         for<'b> T::Ref<'b>: Ord + Copy,
         R: Columnar + Semigroup + for<'b> Semigroup<R::Ref<'b>>,
         for<'b> R::Ref<'b>: Ord,
-        C2: Container + for<'b> PushInto<&'b (D, T, R)>,
+        C2: Container + for<'b, 'c> PushInto<(D::Ref<'b>, T::Ref<'b>, &'c R)>,
     {
         fn push_into(&mut self, container: &'a mut Column<(D, T, R)>) {
 
@@ -431,9 +431,6 @@ pub mod batcher {
                 let mut iter = permutation.drain(..);
                 if let Some((data, time, diff)) = iter.next() {
 
-                    let mut owned_data = D::into_owned(data);
-                    let mut owned_time = T::into_owned(time);
-
                     let mut prev_data = data;
                     let mut prev_time = time;
                     let mut prev_diff = <R as Columnar>::into_owned(diff);
@@ -444,12 +441,8 @@ pub mod batcher {
                         }
                         else {
                             if !prev_diff.is_zero() {
-                                D::copy_from(&mut owned_data, prev_data);
-                                T::copy_from(&mut owned_time, prev_time);
-                                let tuple = (owned_data, owned_time, prev_diff);
-                                self.empty.push_into(&tuple);
-                                owned_data = tuple.0;
-                                owned_time = tuple.1;
+                                let tuple = (prev_data, prev_time, &prev_diff);
+                                self.empty.push_into(tuple);
                             }
                             prev_data = data;
                             prev_time = time;
@@ -458,10 +451,8 @@ pub mod batcher {
                     }
 
                     if !prev_diff.is_zero() {
-                        D::copy_from(&mut owned_data, prev_data);
-                        T::copy_from(&mut owned_time, prev_time);
-                        let tuple = (owned_data, owned_time, prev_diff);
-                        self.empty.push_into(&tuple);
+                        let tuple = (prev_data, prev_time, &prev_diff);
+                        self.empty.push_into(tuple);
                     }
                 }
             }
@@ -661,6 +652,7 @@ pub mod dd_builder {
         <L::ValContainer as BatchContainer>::Owned: Columnar,
         <L::TimeContainer as BatchContainer>::Owned: Columnar,
         <L::DiffContainer as BatchContainer>::Owned: Columnar,
+        // These two constraints seem .. like we could potentially replace by `Columnar::Ref<'a>`.
         for<'a> L::KeyContainer: PushInto<&'a <L::KeyContainer as BatchContainer>::Owned>,
         for<'a> L::ValContainer: PushInto<&'a <L::ValContainer as BatchContainer>::Owned>,
         for<'a> <L::TimeContainer as BatchContainer>::ReadItem<'a> : IntoOwned<'a, Owned = <L::Target as Update>::Time>,