VecCollection

Author: frankmcsherry

advent_of_code_2017/src/bin/day_07.rs

@@ -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_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();

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::VariableRow;
+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: VariableRow<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 = VariableRow::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.

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

@@ -5,7 +5,7 @@ use timely::dataflow::operators::Operator;
 use timely::order::Product;
 use timely::dataflow::{Scope, StreamCore};
 use timely::dataflow::operators::generic::builder_rc::OperatorBuilder;
-use differential_dataflow::{AsCollection, collection::CollectionCore};
+use differential_dataflow::{AsCollection, Collection};
 use differential_dataflow::input::Input;
 use differential_dataflow::operators::iterate::Variable;
 use differential_dataflow::collection::containers::{Enter, Leave, Negate, ResultsIn};
@@ -52,12 +52,12 @@ fn main() {
     timely::example(|scope| {
 
         let numbers = scope.new_collection_from(1 .. 10u32).1;
-        let numbers: CollectionCore<_, _>  = 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: CollectionCore<_, _> = 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>,
 {

differential-dataflow/examples/progress.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;
@@ -115,18 +115,18 @@ fn main() {
 /// The computation to determine this, and to maintain it as times change, is an iterative
 /// computation that propagates times and maintains the minimal elements at each location.
 fn frontier<G, T>(
-    nodes: Collection<G, (Target, Source, T::Summary)>,
-    edges: Collection<G, (Source, Target)>,
-    times: Collection<G, (Location, T)>,
-) -> Collection<G, (Location, T)>
+    nodes: VecCollection<G, (Target, Source, T::Summary)>,
+    edges: VecCollection<G, (Source, Target)>,
+    times: VecCollection<G, (Location, T)>,
+) -> VecCollection<G, (Location, T)>
 where
     G: Scope<Timestamp: Lattice+Ord>,
     T: Timestamp<Summary: differential_dataflow::ExchangeData>,
 {
     // Translate node and edge transitions into a common Location to Location edge with an associated Summary.
     let nodes = nodes.map(|(target, source, summary)| (Location::from(target), (Location::from(source), summary)));
     let edges = edges.map(|(source, target)| (Location::from(source), (Location::from(target), Default::default())));
-    let transitions: Collection<G, (Location, (Location, T::Summary))> = nodes.concat(&edges);
+    let transitions: VecCollection<G, (Location, (Location, T::Summary))> = nodes.concat(&edges);
 
     times
         .iterate(|reach| {
@@ -149,9 +149,9 @@ where
 
 /// Summary paths from locations to operator zero inputs.
 fn summarize<G, T>(
-    nodes: Collection<G, (Target, Source, T::Summary)>,
-    edges: Collection<G, (Source, Target)>,
-) -> Collection<G, (Location, (Location, T::Summary))>
+    nodes: VecCollection<G, (Target, Source, T::Summary)>,
+    edges: VecCollection<G, (Source, Target)>,
+) -> VecCollection<G, (Location, (Location, T::Summary))>
 where
     G: Scope<Timestamp: Lattice+Ord>,
     T: Timestamp<Summary: differential_dataflow::ExchangeData+std::hash::Hash>,
@@ -166,7 +166,7 @@ where
     // Retain node connections along "default" timestamp summaries.
     let nodes = nodes.map(|(target, source, summary)| (Location::from(source), (Location::from(target), summary)));
     let edges = edges.map(|(source, target)| (Location::from(target), (Location::from(source), Default::default())));
-    let transitions: Collection<G, (Location, (Location, T::Summary))> = nodes.concat(&edges);
+    let transitions: VecCollection<G, (Location, (Location, T::Summary))> = nodes.concat(&edges);
 
     zero_inputs
         .iterate(|summaries| {
@@ -192,9 +192,9 @@ where
 
 /// Identifies cycles along paths that do not increment timestamps.
 fn find_cycles<G: Scope, T: Timestamp>(
-    nodes: Collection<G, (Target, Source, T::Summary)>,
-    edges: Collection<G, (Source, Target)>,
-) -> Collection<G, (Location, Location)>
+    nodes: VecCollection<G, (Target, Source, T::Summary)>,
+    edges: VecCollection<G, (Source, Target)>,
+) -> VecCollection<G, (Location, Location)>
 where
     G: Scope<Timestamp: Lattice+Ord>,
     T: Timestamp<Summary: differential_dataflow::ExchangeData>,
@@ -209,7 +209,7 @@ where
         }
     });
     let edges = edges.map(|(source, target)| (Location::from(source), Location::from(target)));
-    let transitions: Collection<G, (Location, Location)> = nodes.concat(&edges);
+    let transitions: VecCollection<G, (Location, Location)> = nodes.concat(&edges);
 
     // Repeatedly restrict to locations with an incoming path.
     transitions
@@ -223,4 +223,4 @@ where
                 .semijoin(&active)
         })
         .consolidate()
-}
+}