Extract VecCollection from Collection

advent_of_code_2017/src/bin/day_07.rs

@@ -3,7 +3,7 @@ extern crate differential_dataflow;
 
 // taken from: https://adventofcode.com/2017/day/6
 
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::*;
 
@@ -1074,7 +1074,7 @@ tvhftq (35)";
         let index = worker.index();
         let peers = worker.peers();
 
-        let worker_input = 
+        let worker_input =
         input
             .split('\n')
             .enumerate()
@@ -1102,7 +1102,7 @@ tvhftq (35)";
             let weights = input.explode(|(name,weight,_)| Some((name, weight as isize)));
             let parents = input.flat_map(|(name,_,links)| links.into_iter().map(move |link| (link,name.to_string())));
 
-            let total_weights: Collection<_,String> = weights
+            let total_weights: VecCollection<_,String> = weights
                 .iterate(|inner| {
                     parents.enter(&inner.scope())
                            .semijoin(&inner)

advent_of_code_2017/src/bin/day_08.rs

@@ -3,7 +3,7 @@ extern crate differential_dataflow;
 
 // taken from: https://adventofcode.com/2017/day/8
 
-// use differential_dataflow::Collection;
+// use differential_dataflow::VecCollection;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::*;
 use differential_dataflow::algorithms::prefix_sum::PrefixSum;
@@ -1068,7 +1068,7 @@ wui inc -120 if i > -2038";
             let edits = data.map(|(pos, line)| {
 
                 // Each line has a read location with a condition, and a write location with an operation.
-                // E.g., 
+                // E.g.,
                 //
                 //      kxm dec 986 if xbh <= -2515
                 //      obc inc 63 if wui >= 2800

advent_of_code_2017/src/bin/day_09.rs

@@ -20,7 +20,7 @@ fn main() {
         let index = worker.index();
         let peers = worker.peers();
 
-        let worker_input = 
+        let worker_input =
         input
             .chars()
             .enumerate()
@@ -37,7 +37,7 @@ fn main() {
             //   { next_invalid, garbage }
             //
             // where the first bool indicates that the next character should be ignored,
-            // and the second bool indicates that we are in a garbage scope. We will 
+            // and the second bool indicates that we are in a garbage scope. We will
             // encode this as the values 0 .. 4, where
             //
             //  0: valid, non-garbage
@@ -48,7 +48,7 @@ fn main() {
             // Each character initially describes a substring of length one, but we will
             // build up the state transition for larger substrings, iteratively.
 
-            let transitions = 
+            let transitions =
             input
                 .map(|(pos, character)|
                     (pos, match character {
@@ -95,7 +95,7 @@ fn main() {
 }
 
 /// Accumulate data in `collection` into all powers-of-two intervals containing them.
-fn pp_aggregate<G, D, F>(collection: Collection<G, (usize, D)>, combine: F) -> Collection<G, ((usize, usize), D)>
+fn pp_aggregate<G, D, F>(collection: VecCollection<G, (usize, D)>, combine: F) -> VecCollection<G, ((usize, usize), D)>
 where
     G: Scope<Timestamp: Lattice>,
     D: Data,
@@ -105,7 +105,7 @@ where
     let unit_ranges = collection.map(|(index, data)| ((index, 0), data));
 
     unit_ranges
-        .iterate(|ranges| 
+        .iterate(|ranges|
 
             // Each available range, of size less than usize::max_value(), advertises itself as the range
             // twice as large, aligned to integer multiples of its size. Each range, which may contain at
@@ -126,26 +126,26 @@ where
 
 /// Produces the accumulated values at each of the `usize` locations in `aggregates` (and others).
 fn pp_broadcast<G, D, B, F>(
-    ranges: Collection<G, ((usize, usize), D)>, 
+    ranges: VecCollection<G, ((usize, usize), D)>,
     seed: B,
     zero: D,
-    combine: F) -> Collection<G, (usize, B)>
+    combine: F) -> VecCollection<G, (usize, B)>
 where
     G: Scope<Timestamp: Lattice+Ord+::std::fmt::Debug>,
     D: Data,
     B: Data+::std::hash::Hash,
     F: Fn(&B, &D) -> B + 'static,
 {
     // Each range proposes an empty first child, to provide for its second child if it has no sibling.
-    // This is important if we want to reconstruct 
+    // This is important if we want to reconstruct
     let zero_ranges =
         ranges
             .map(move |((pos, log),_)| ((pos, if log > 0 { log - 1 } else { 0 }), zero.clone()))
             .antijoin(&ranges.map(|((pos, log),_)| (pos, log)));
 
     let aggregates = ranges.concat(&zero_ranges);
 
-    let init_state = 
+    let init_state =
     Some(((0, seed), Default::default(), 1))
         .to_stream(&mut aggregates.scope())
         .as_collection();

advent_of_code_2017/src/bin/day_10.rs

@@ -3,7 +3,7 @@ extern crate differential_dataflow;
 
 // taken from: https://adventofcode.com/2017/day/8
 
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::*;
 
@@ -61,4 +61,4 @@ fn knot_step<I: Iterator<Item=u8>+Clone>(iter: I, rounds: usize) -> Vec<u8> {
     }
 
     state
-}
+}

advent_of_code_2017/src/bin/day_14.rs

@@ -3,7 +3,7 @@ extern crate differential_dataflow;
 
 // taken from: https://adventofcode.com/2017/day/8
 
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::*;
 
@@ -31,8 +31,8 @@ fn main() {
                     // println!("{:?}: \t{:?}", row, hash);
 
                     (0 .. 128)
-                        .map(move |col| 
-                            if hash[col/8] & (1 << (7-(col % 8))) != 0 { 
+                        .map(move |col|
+                            if hash[col/8] & (1 << (7-(col % 8))) != 0 {
                                 (row, col, '#')
                             }
                             else {
@@ -92,4 +92,4 @@ fn knot_step<I: Iterator<Item=u8>+Clone>(iter: I, rounds: usize) -> Vec<u8> {
     }
 
     state
-}
+}

advent_of_code_2017/src/bin/day_15.rs

@@ -3,7 +3,7 @@ extern crate differential_dataflow;
 
 // taken from: https://adventofcode.com/2017/day/8
 
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::*;
 
@@ -43,4 +43,4 @@ fn main() {
 
     println!("part2: {:?}", equal);
 
-}
+}

differential-dataflow/examples/bfs.rs

@@ -4,7 +4,7 @@ use timely::dataflow::*;
 use timely::dataflow::operators::probe::Handle;
 
 use differential_dataflow::input::Input;
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::operators::*;
 use differential_dataflow::lattice::Lattice;
 
@@ -91,9 +91,9 @@ fn main() {
 }
 
 // returns pairs (n, s) indicating node n can be reached from a root in s steps.
-fn bfs<G>(edges: &Collection<G, Edge>, roots: &Collection<G, Node>) -> Collection<G, (Node, u32)>
+fn bfs<G>(edges: &VecCollection<G, Edge>, roots: &VecCollection<G, Node>) -> VecCollection<G, (Node, u32)>
 where
-    G: Scope<Timestamp: Lattice+Ord>, 
+    G: Scope<Timestamp: Lattice+Ord>,
 {
     // initialize roots as reaching themselves at distance 0
     let nodes = roots.map(|x| (x, 0));
@@ -108,4 +108,4 @@ where
              .concat(&nodes)
              .reduce(|_, s, t| t.push((*s[0].0, 1)))
      })
-}
+}

differential-dataflow/examples/dynamic.rs

@@ -4,7 +4,7 @@ use timely::dataflow::*;
 use timely::dataflow::operators::probe::Handle;
 
 use differential_dataflow::input::Input;
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::operators::*;
 use differential_dataflow::lattice::Lattice;
 
@@ -91,9 +91,9 @@ fn main() {
 }
 
 // returns pairs (n, s) indicating node n can be reached from a root in s steps.
-fn bfs<G>(edges: &Collection<G, Edge>, roots: &Collection<G, Node>) -> Collection<G, (Node, u32)>
+fn bfs<G>(edges: &VecCollection<G, Edge>, roots: &VecCollection<G, Node>) -> VecCollection<G, (Node, u32)>
 where
-    G: Scope<Timestamp: Lattice+Ord>, 
+    G: Scope<Timestamp: Lattice+Ord>,
 {
     use timely::order::Product;
     use iterate::Variable;
@@ -115,7 +115,7 @@ where
         let inner = feedback_summary::<usize>(1, 1);
         let label = Variable::new_from(nodes.clone(), Product { outer: Default::default(), inner });
 
-        let next = 
+        let next =
         label
             .join_map(&edges, |_k,l,d| (*d, l+1))
             .concat(&nodes)
@@ -130,4 +130,4 @@ where
             .leave()
     })
 
-}
+}

differential-dataflow/examples/graspan.rs

@@ -8,11 +8,11 @@ use timely::order::Product;
 use timely::dataflow::Scope;
 use timely::dataflow::scopes::ScopeParent;
 
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::input::{Input, InputSession};
 use differential_dataflow::operators::arrange::{ArrangeByKey, ArrangeBySelf};
-use differential_dataflow::operators::iterate::Variable;
+use differential_dataflow::operators::iterate::VecVariable;
 use differential_dataflow::operators::Threshold;
 
 type Node = usize;
@@ -83,16 +83,16 @@ type Arrange<G,K,V,R> = Arranged<G, TraceValHandle<K, V, <G as ScopeParent>::Tim
 /// An edge variable provides arranged representations of its contents, even before they are
 /// completely defined, in support of recursively defined productions.
 pub struct EdgeVariable<G: Scope<Timestamp: Lattice>> {
-    variable: Variable<G, Edge, Diff>,
-    current: Collection<G, Edge, Diff>,
+    variable: VecVariable<G, Edge, Diff>,
+    current: VecCollection<G, Edge, Diff>,
     forward: Option<Arrange<G, Node, Node, Diff>>,
     reverse: Option<Arrange<G, Node, Node, Diff>>,
 }
 
 impl<G: Scope<Timestamp: Lattice>> EdgeVariable<G> {
     /// Creates a new variable initialized with `source`.
-    pub fn from(source: &Collection<G, Edge>, step: <G::Timestamp as Timestamp>::Summary) -> Self {
-        let variable = Variable::new(&mut source.scope(), step);
+    pub fn from(source: &VecCollection<G, Edge>, step: <G::Timestamp as Timestamp>::Summary) -> Self {
+        let variable = VecVariable::new(&mut source.scope(), step);
         EdgeVariable {
             variable: variable,
             current: source.clone(),
@@ -101,7 +101,7 @@ impl<G: Scope<Timestamp: Lattice>> EdgeVariable<G> {
         }
     }
     /// Concatenates `production` into the definition of the variable.
-    pub fn add_production(&mut self, production: &Collection<G, Edge, Diff>) {
+    pub fn add_production(&mut self, production: &VecCollection<G, Edge, Diff>) {
         self.current = self.current.concat(production);
     }
     /// Finalizes the variable, connecting its recursive definition.
@@ -153,7 +153,7 @@ impl Query {
 
     /// Creates a dataflow implementing the query, and returns input and trace handles.
     pub fn render_in<G>(&self, scope: &mut G) -> IndexMap<String, RelationHandles<G::Timestamp>>
-    where 
+    where
         G: Scope<Timestamp: Lattice+::timely::order::TotalOrder>,
     {
         // Create new input (handle, stream) pairs

differential-dataflow/examples/interpreted.rs

@@ -2,7 +2,7 @@ use std::hash::Hash;
 use timely::dataflow::*;
 use timely::dataflow::operators::*;
 
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::*;
 
@@ -35,13 +35,13 @@ fn main() {
         println!("loaded {} nodes, {} edges", nodes, edges.len());
 
         worker.dataflow::<(),_,_>(|scope| {
-            interpret(&Collection::new(edges.to_stream(scope)), &[(0,2), (1,2)]);
+            interpret(&VecCollection::new(edges.to_stream(scope)), &[(0,2), (1,2)]);
         });
 
     }).unwrap();
 }
 
-fn interpret<G>(edges: &Collection<G, Edge>, relations: &[(usize, usize)]) -> Collection<G, Vec<Node>>
+fn interpret<G>(edges: &VecCollection<G, Edge>, relations: &[(usize, usize)]) -> VecCollection<G, Vec<Node>>
 where
     G: Scope<Timestamp: Lattice+Hash+Ord>,
 {
@@ -103,4 +103,4 @@ where
     }
 
     results
-}
+}

differential-dataflow/examples/iterate_container.rs

@@ -52,12 +52,12 @@ fn main() {
     timely::example(|scope| {
 
         let numbers = scope.new_collection_from(1 .. 10u32).1;
-        let numbers: Collection<_, u32, isize, _>  = wrap(&numbers.inner).as_collection();
+        let numbers: Collection<_, _>  = wrap(&numbers.inner).as_collection();
 
         scope.iterative::<u64,_,_>(|nested| {
             let summary = Product::new(Default::default(), 1);
             let variable = Variable::new_from(numbers.enter(nested), summary);
-            let mapped: Collection<_, u32, isize, _> = variable.inner.unary(Pipeline, "Map", |_,_| {
+            let mapped: Collection<_, _> = variable.inner.unary(Pipeline, "Map", |_,_| {
                 |input, output| {
                     input.for_each(|time, data| {
                         let mut session = output.session(&time);

differential-dataflow/examples/monoid-bfs.rs

@@ -5,7 +5,7 @@ use timely::dataflow::*;
 use timely::dataflow::operators::probe::Handle;
 
 use differential_dataflow::input::Input;
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::operators::*;
 use differential_dataflow::lattice::Lattice;
 
@@ -123,7 +123,7 @@ fn main() {
 }
 
 // returns pairs (n, s) indicating node n can be reached from a root in s steps.
-fn bfs<G>(edges: &Collection<G, Edge, MinSum>, roots: &Collection<G, Node, MinSum>) -> Collection<G, Node, MinSum>
+fn bfs<G>(edges: &VecCollection<G, Edge, MinSum>, roots: &VecCollection<G, Node, MinSum>) -> VecCollection<G, Node, MinSum>
 where
     G: Scope<Timestamp: Lattice+Ord>,
 {

differential-dataflow/examples/pagerank.rs

@@ -1,7 +1,7 @@
 use timely::order::Product;
 use timely::dataflow::{*, operators::Filter};
 
-use differential_dataflow::Collection;
+use differential_dataflow::VecCollection;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::operators::{*, iterate::Variable};
 use differential_dataflow::input::InputSession;
@@ -77,7 +77,7 @@ fn main() {
 
 // Returns a weighted collection in which the weight of each node is proportional
 // to its PageRank in the input graph `edges`.
-fn pagerank<G>(iters: Iter, edges: &Collection<G, Edge, Diff>) -> Collection<G, Node, Diff>
+fn pagerank<G>(iters: Iter, edges: &VecCollection<G, Edge, Diff>) -> VecCollection<G, Node, Diff>
 where
     G: Scope<Timestamp: Lattice>,
 {