reduce_core: present referenced data to logic function

`reduce_core` was previously forced to clone the output data before
presenting it to the logic function for inspection.

Simply annotating the type with a reference was not possible due to the
buffer being held in the same struct as some of the data it contained,
making it a self-referencial struct which the borrow checker doesn't
allow.

However, the only reason this buffer was held in the same struct was to
reuse its allocation. This patch therefore creates a brand new vector
every time we need to process data that contains references instead of
cloned data. The expectation is that in the general case it's better to
allocate a vector of references than clone a bunch of values.

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

Author: petrosagg

src/operators/reduce.rs

@@ -281,7 +281,7 @@ pub trait ReduceCore<G: Scope, K: Data, V: Data, R: Semigroup> where G::Timestam
                 if !input.is_empty() {
                     logic(key, input, change);
                 }
-                change.extend(output.drain(..).map(|(x,d)| (x, d.negate())));
+                change.extend(output.drain(..).map(|(x,d)| (x.clone(), d.negate())));
                 crate::consolidation::consolidate(change);
             })
         }
@@ -298,7 +298,7 @@ pub trait ReduceCore<G: Scope, K: Data, V: Data, R: Semigroup> where G::Timestam
             T2::R: Semigroup,
             T2::Batch: Batch<K, T2::Val, G::Timestamp, T2::R>,
             T2::Cursor: Cursor<K, T2::Val, G::Timestamp, T2::R>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val,T2::R)>)+'static
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(&T2::Val,T2::R)>, &mut Vec<(T2::Val,T2::R)>)+'static
             ;
 }
 
@@ -317,7 +317,7 @@ where
             T2: Trace+TraceReader<Key=K, Time=G::Timestamp>+'static,
             T2::Batch: Batch<K, T2::Val, G::Timestamp, T2::R>,
             T2::Cursor: Cursor<K, T2::Val, G::Timestamp, T2::R>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val, T2::R)>)+'static
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(&T2::Val,T2::R)>, &mut Vec<(T2::Val, T2::R)>)+'static
     {
         self.arrange_by_key_named(&format!("Arrange: {}", name))
             .reduce_core(name, logic)
@@ -338,7 +338,7 @@ where
             T2::R: Semigroup,
             T2::Batch: Batch<K, T2::Val, G::Timestamp, T2::R>,
             T2::Cursor: Cursor<K, T2::Val, G::Timestamp, T2::R>,
-            L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val,T2::R)>, &mut Vec<(T2::Val, T2::R)>)+'static {
+            L: FnMut(&K, &[(&V, R)], &mut Vec<(&T2::Val,T2::R)>, &mut Vec<(T2::Val, T2::R)>)+'static {
 
         let mut result_trace = None;
 
@@ -676,7 +676,7 @@ where
         C1: Cursor<K, V1, T, R1>,
         C2: Cursor<K, V2, T, R2>,
         C3: Cursor<K, V1, T, R1>,
-        L: FnMut(&K, &[(&V1, R1)], &mut Vec<(V2, R2)>, &mut Vec<(V2, R2)>);
+        L: FnMut(&K, &[(&V1, R1)], &mut Vec<(&V2, R2)>, &mut Vec<(V2, R2)>);
 }
 
 
@@ -705,7 +705,7 @@ mod history_replay {
         input_history: ValueHistory<'a, V1, T, R1>,
         output_history: ValueHistory<'a, V2, T, R2>,
         input_buffer: Vec<(&'a V1, R1)>,
-        output_buffer: Vec<(V2, R2)>,
+        output_buffer_capacity: usize,
         update_buffer: Vec<(V2, R2)>,
         output_produced: Vec<((V2, T), R2)>,
         synth_times: Vec<T>,
@@ -728,7 +728,7 @@ mod history_replay {
                 input_history: ValueHistory::new(),
                 output_history: ValueHistory::new(),
                 input_buffer: Vec::new(),
-                output_buffer: Vec::new(),
+                output_buffer_capacity: 0,
                 update_buffer: Vec::new(),
                 output_produced: Vec::new(),
                 synth_times: Vec::new(),
@@ -754,7 +754,7 @@ mod history_replay {
             C1: Cursor<K, V1, T, R1>,
             C2: Cursor<K, V2, T, R2>,
             C3: Cursor<K, V1, T, R1>,
-            L: FnMut(&K, &[(&V1, R1)], &mut Vec<(V2, R2)>, &mut Vec<(V2, R2)>)
+            L: FnMut(&K, &[(&V1, R1)], &mut Vec<(&V2, R2)>, &mut Vec<(V2, R2)>)
         {
 
             // The work we need to perform is at times defined principally by the contents of `batch_cursor`
@@ -913,30 +913,31 @@ mod history_replay {
                         }
                         crate::consolidation::consolidate(&mut self.input_buffer);
 
+                        let mut output_buffer = Vec::with_capacity(self.output_buffer_capacity);
                         meet.as_ref().map(|meet| output_replay.advance_buffer_by(&meet));
-                        for &((ref value, ref time), ref diff) in output_replay.buffer().iter() {
+                        for &((value, ref time), ref diff) in output_replay.buffer().iter() {
                             if time.less_equal(&next_time) {
-                                self.output_buffer.push(((*value).clone(), diff.clone()));
+                                output_buffer.push((value, diff.clone()));
                             }
                             else {
                                 self.temporary.push(next_time.join(time));
                             }
                         }
                         for &((ref value, ref time), ref diff) in self.output_produced.iter() {
                             if time.less_equal(&next_time) {
-                                self.output_buffer.push(((*value).clone(), diff.clone()));
+                                output_buffer.push((value, diff.clone()));
                             }
                             else {
                                 self.temporary.push(next_time.join(&time));
                             }
                         }
-                        crate::consolidation::consolidate(&mut self.output_buffer);
+                        crate::consolidation::consolidate(&mut output_buffer);
 
                         // Apply user logic if non-empty input and see what happens!
-                        if self.input_buffer.len() > 0 || self.output_buffer.len() > 0 {
-                            logic(key, &self.input_buffer[..], &mut self.output_buffer, &mut self.update_buffer);
+                        if self.input_buffer.len() > 0 || output_buffer.len() > 0 {
+                            logic(key, &self.input_buffer[..], &mut output_buffer, &mut self.update_buffer);
                             self.input_buffer.clear();
-                            self.output_buffer.clear();
+                            self.output_buffer_capacity = output_buffer.capacity();
                         }
 
                         // output_replay.advance_buffer_by(&meet);