Use consolidating builder in consolidate_stream (#488)

This switches from consolidating each input individually to
consolidating at the output. The benefits are that it can yield better
consolidation performance because it can consolidate across input
containers instead of only being able to consolidate each individual
input container.

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

src/operators/consolidate.rs

@@ -9,6 +9,7 @@
 use timely::dataflow::Scope;
 
 use crate::{Collection, ExchangeData, Hashable};
+use crate::consolidation::ConsolidatingContainerBuilder;
 use crate::difference::Semigroup;
 
 use crate::Data;
@@ -92,14 +93,13 @@ where
         use crate::collection::AsCollection;
 
         self.inner
-            .unary(Pipeline, "ConsolidateStream", |_cap, _info| {
+            .unary::<ConsolidatingContainerBuilder<_>, _, _, _>(Pipeline, "ConsolidateStream", |_cap, _info| {
 
                 let mut vector = Vec::new();
                 move |input, output| {
                     input.for_each(|time, data| {
                         data.swap(&mut vector);
-                        crate::consolidation::consolidate_updates(&mut vector);
-                        output.session(&time).give_container(&mut vector);
+                        output.session_with_builder(&time).give_container(&mut vector);
                     })
                 }
             })