Extract CollectionCore from Collection

Author: frankmcsherry

differential-dataflow/examples/graspan.rs

@@ -12,7 +12,7 @@ use differential_dataflow::Collection;
 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::VariableRow;
 use differential_dataflow::operators::Threshold;
 
 type Node = usize;
@@ -83,7 +83,7 @@ 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>,
+    variable: VariableRow<G, Edge, Diff>,
     current: Collection<G, Edge, Diff>,
     forward: Option<Arrange<G, Node, Node, Diff>>,
     reverse: Option<Arrange<G, Node, Node, Diff>>,
@@ -92,7 +92,7 @@ pub struct EdgeVariable<G: Scope<Timestamp: Lattice>> {
 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);
+        let variable = VariableRow::new(&mut source.scope(), step);
         EdgeVariable {
             variable: variable,
             current: source.clone(),
@@ -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/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};
+use differential_dataflow::{AsCollection, collection::CollectionCore};
 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: Collection<_, u32, isize, _>  = wrap(&numbers.inner).as_collection();
+        let numbers: CollectionCore<_, _>  = 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: CollectionCore<_, _> = variable.inner.unary(Pipeline, "Map", |_,_| {
                 |input, output| {
                     input.for_each(|time, data| {
                         let mut session = output.session(&time);

differential-dataflow/src/collection.rs

@@ -37,8 +37,15 @@ use crate::hashable::Hashable;
 /// parameter is the type of data in your collection, for example `String`, or `(u32, Vec<Option<()>>)`.
 /// The `R` parameter represents the types of changes that the data undergo, and is most commonly (and
 /// defaults to) `isize`, representing changes to the occurrence count of each record.
+pub type Collection<G, D, R = isize> = CollectionCore<G, Vec<(D, <G as ScopeParent>::Timestamp, R)>>;
+
+/// A collection represented by a stream of abstract containers.
+///
+/// The containers purport to reperesent changes to a collection, and they must implement various traits
+/// in order to expose some of this functionality (e.g. negation, timestamp manipulation). Other actions
+/// on the containers, and streams of containers, are left to the container implementor to describe.
 #[derive(Clone)]
-pub struct Collection<G: Scope, D, R = isize, C = Vec<(D, <G as ScopeParent>::Timestamp, R)>> {
+pub struct CollectionCore<G: Scope, C> {
     /// The underlying timely dataflow stream.
     ///
     /// This field is exposed to support direct timely dataflow manipulation when required, but it is
@@ -48,11 +55,9 @@ pub struct Collection<G: Scope, D, R = isize, C = Vec<(D, <G as ScopeParent>::Ti
     /// the timely-dataflow sense. If this invariant is not upheld, differential operators may behave
     /// unexpectedly.
     pub inner: timely::dataflow::StreamCore<G, C>,
-    /// Phantom data for unreferenced `D` and `R` types.
-    phantom: std::marker::PhantomData<(D, R)>,
 }
 
-impl<G: Scope, D, R, C> Collection<G, D, R, C> {
+impl<G: Scope, C> CollectionCore<G, C> {
     /// Creates a new Collection from a timely dataflow stream.
     ///
     /// This method seems to be rarely used, with the `as_collection` method on streams being a more
@@ -62,11 +67,9 @@ impl<G: Scope, D, R, C> Collection<G, D, R, C> {
     ///
     /// This stream should satisfy the timestamp invariant as documented on [Collection]; this
     /// method does not check it.
-    pub fn new(stream: StreamCore<G, C>) -> Collection<G, D, R, C> {
-        Collection { inner: stream, phantom: std::marker::PhantomData }
-    }
+    pub fn new(stream: StreamCore<G, C>) -> Self { Self { inner: stream } }
 }
-impl<G: Scope, D, R, C: Container> Collection<G, D, R, C> {
+impl<G: Scope, C: Container> CollectionCore<G, C> {
     /// Creates a new collection accumulating the contents of the two collections.
     ///
     /// Despite the name, differential dataflow collections are unordered. This method is so named because the
@@ -128,7 +131,7 @@ impl<G: Scope, D, R, C: Container> Collection<G, D, R, C> {
     ///
     /// This method is a specialization of `enter` to the case where the nested scope is a region.
     /// It removes the need for an operator that adjusts the timestamp.
-    pub fn enter_region<'a>(&self, child: &Child<'a, G, <G as ScopeParent>::Timestamp>) -> Collection<Child<'a, G, <G as ScopeParent>::Timestamp>, D, R, C> {
+    pub fn enter_region<'a>(&self, child: &Child<'a, G, <G as ScopeParent>::Timestamp>) -> CollectionCore<Child<'a, G, <G as ScopeParent>::Timestamp>, C> {
         self.inner
             .enter(child)
             .as_collection()
@@ -204,7 +207,7 @@ impl<G: Scope, D, R, C: Container> Collection<G, D, R, C> {
     ///         .assert_eq(&evens);
     /// });
     /// ```
-    pub fn negate(&self) -> Collection<G, D, R, C> where C: containers::Negate {
+    pub fn negate(&self) -> Self where C: containers::Negate {
         use timely::dataflow::channels::pact::Pipeline;
         self.inner
             .unary(Pipeline, "Negate", move |_,_| move |input, output| {
@@ -233,7 +236,7 @@ impl<G: Scope, D, R, C: Container> Collection<G, D, R, C> {
     ///     data.assert_eq(&result);
     /// });
     /// ```
-    pub fn enter<'a, T>(&self, child: &Child<'a, G, T>) -> Collection<Child<'a, G, T>, D, R, <C as containers::Enter<<G as ScopeParent>::Timestamp, T>>::InnerContainer>
+    pub fn enter<'a, T>(&self, child: &Child<'a, G, T>) -> CollectionCore<Child<'a, G, T>, <C as containers::Enter<<G as ScopeParent>::Timestamp, T>>::InnerContainer>
     where
         C: containers::Enter<<G as ScopeParent>::Timestamp, T, InnerContainer: Container>,
         T: Refines<<G as ScopeParent>::Timestamp>,
@@ -594,7 +597,7 @@ use timely::dataflow::scopes::ScopeParent;
 use timely::progress::timestamp::Refines;
 
 /// Methods requiring a nested scope.
-impl<'a, G: Scope, T: Timestamp, D: Clone+'static, R: Clone+'static, C: Container> Collection<Child<'a, G, T>, D, R, C>
+impl<'a, G: Scope, T: Timestamp, C: Container> CollectionCore<Child<'a, G, T>, C>
 where
     C: containers::Leave<T, G::Timestamp, OuterContainer: Container>,
     T: Refines<<G as ScopeParent>::Timestamp>,
@@ -619,7 +622,7 @@ where
     ///     data.assert_eq(&result);
     /// });
     /// ```
-    pub fn leave(&self) -> Collection<G, D, R, <C as containers::Leave<T, G::Timestamp>>::OuterContainer> {
+    pub fn leave(&self) -> CollectionCore<G, <C as containers::Leave<T, G::Timestamp>>::OuterContainer> {
         use timely::dataflow::channels::pact::Pipeline;
         self.inner
             .leave()
@@ -631,13 +634,13 @@ where
 }
 
 /// Methods requiring a region as the scope.
-impl<G: Scope, D, R, C: Container+Data> Collection<Child<'_, G, G::Timestamp>, D, R, C>
+impl<G: Scope, C: Container+Data> CollectionCore<Child<'_, G, G::Timestamp>, C>
 {
     /// Returns the value of a Collection from a nested region to its containing scope.
     ///
     /// This method is a specialization of `leave` to the case that of a nested region.
     /// It removes the need for an operator that adjusts the timestamp.
-    pub fn leave_region(&self) -> Collection<G, D, R, C> {
+    pub fn leave_region(&self) -> CollectionCore<G, C> {
         self.inner
             .leave()
             .as_collection()
@@ -682,18 +685,18 @@ impl<G: Scope<Timestamp: Data>, D: Clone+'static, R: Abelian+'static> Collection
 }
 
 /// Conversion to a differential dataflow Collection.
-pub trait AsCollection<G: Scope, D, R, C> {
+pub trait AsCollection<G: Scope, C> {
     /// Converts the type to a differential dataflow collection.
-    fn as_collection(&self) -> Collection<G, D, R, C>;
+    fn as_collection(&self) -> CollectionCore<G, C>;
 }
 
-impl<G: Scope, D, R, C: Clone> AsCollection<G, D, R, C> for StreamCore<G, C> {
+impl<G: Scope, C: Clone> AsCollection<G, C> for StreamCore<G, C> {
     /// Converts the type to a differential dataflow collection.
     ///
     /// By calling this method, you guarantee that the timestamp invariant (as documented on
     /// [Collection]) is upheld. This method will not check it.
-    fn as_collection(&self) -> Collection<G, D, R, C> {
-        Collection::<G,D,R,C>::new(self.clone())
+    fn as_collection(&self) -> CollectionCore<G, C> {
+        CollectionCore::<G,C>::new(self.clone())
     }
 }
 
@@ -718,13 +721,11 @@ impl<G: Scope, D, R, C: Clone> AsCollection<G, D, R, C> for StreamCore<G, C> {
 ///         .assert_eq(&data);
 /// });
 /// ```
-pub fn concatenate<G, D, R, C, I>(scope: &mut G, iterator: I) -> Collection<G, D, R, C>
+pub fn concatenate<G, C, I>(scope: &mut G, iterator: I) -> CollectionCore<G, C>
 where
     G: Scope,
-    D: Data,
-    R: Semigroup + 'static,
     C: Container,
-    I: IntoIterator<Item=Collection<G, D, R, C>>,
+    I: IntoIterator<Item=CollectionCore<G, C>>,
 {
     scope
         .concatenate(iterator.into_iter().map(|x| x.inner))

differential-dataflow/src/operators/iterate.rs

@@ -42,7 +42,7 @@ use timely::dataflow::scopes::child::Iterative;
 use timely::dataflow::operators::{Feedback, ConnectLoop};
 use timely::dataflow::operators::feedback::Handle;
 
-use crate::{Data, Collection};
+use crate::{Data, Collection, collection::CollectionCore};
 use crate::difference::{Semigroup, Abelian};
 use crate::lattice::Lattice;
 
@@ -151,20 +151,21 @@ impl<G: Scope<Timestamp: Lattice>, D: Ord+Data+Debug, R: Semigroup+'static> Iter
 ///     });
 /// })
 /// ```
-pub struct Variable<G, D, R, C = Vec<(D, <G as ScopeParent>::Timestamp, R)>>
+pub struct Variable<G, C>
 where
     G: Scope<Timestamp: Lattice>,
-    D: Data,
-    R: Abelian + 'static,
     C: Container,
 {
-    collection: Collection<G, D, R, C>,
+    collection: CollectionCore<G, C>,
     feedback: Handle<G, C>,
-    source: Option<Collection<G, D, R, C>>,
+    source: Option<CollectionCore<G, C>>,
     step: <G::Timestamp as Timestamp>::Summary,
 }
 
-impl<G, D: Data, R: Abelian, C: Container> Variable<G, D, R, C>
+/// A `Variable` specialized to a vector container of update triples (data, time, diff).
+pub type VariableRow<G, D, R> = Variable<G, Vec<(D, <G as ScopeParent>::Timestamp, R)>>;
+
+impl<G, C: Container> Variable<G, C>
 where
     G: Scope<Timestamp: Lattice>,
     C: crate::collection::containers::Negate + crate::collection::containers::ResultsIn<<G::Timestamp as Timestamp>::Summary>,
@@ -175,22 +176,22 @@ where
     /// be used whenever the variable has an empty input.
     pub fn new(scope: &mut G, step: <G::Timestamp as Timestamp>::Summary) -> Self {
         let (feedback, updates) = scope.feedback(step.clone());
-        let collection = Collection::<G, D, R, C>::new(updates);
+        let collection = CollectionCore::<G, C>::new(updates);
         Self { collection, feedback, source: None, step }
     }
 
     /// Creates a new `Variable` from a supplied `source` stream.
-    pub fn new_from(source: Collection<G, D, R, C>, step: <G::Timestamp as Timestamp>::Summary) -> Self {
+    pub fn new_from(source: CollectionCore<G, C>, step: <G::Timestamp as Timestamp>::Summary) -> Self {
         let (feedback, updates) = source.inner.scope().feedback(step.clone());
-        let collection = Collection::<G, D, R, C>::new(updates).concat(&source);
+        let collection = CollectionCore::<G, C>::new(updates).concat(&source);
         Variable { collection, feedback, source: Some(source), step }
     }
 
     /// Set the definition of the `Variable` to a collection.
     ///
     /// This method binds the `Variable` to be equal to the supplied collection,
     /// which may be recursively defined in terms of the variable itself.
-    pub fn set(self, result: &Collection<G, D, R, C>) -> Collection<G, D, R, C> {
+    pub fn set(self, result: &CollectionCore<G, C>) -> CollectionCore<G, C> {
         let mut in_result = result.clone();
         if let Some(source) = &self.source {
             in_result = in_result.concat(&source.negate());
@@ -207,7 +208,7 @@ where
     ///
     /// This behavior can also be achieved by using `new` to create an empty initial
     /// collection, and then using `self.set(self.concat(result))`.
-    pub fn set_concat(self, result: &Collection<G, D, R, C>) -> Collection<G, D, R, C> {
+    pub fn set_concat(self, result: &CollectionCore<G, C>) -> CollectionCore<G, C> {
         let step = self.step;
         result
             .results_in(step)
@@ -218,8 +219,8 @@ where
     }
 }
 
-impl<G: Scope<Timestamp: Lattice>, D: Data, R: Abelian, C: Container> Deref for Variable<G, D, R, C> {
-    type Target = Collection<G, D, R, C>;
+impl<G: Scope<Timestamp: Lattice>, C: Container> Deref for Variable<G, C> {
+    type Target = CollectionCore<G, C>;
     fn deref(&self) -> &Self::Target {
         &self.collection
     }
@@ -231,32 +232,30 @@ impl<G: Scope<Timestamp: Lattice>, D: Data, R: Abelian, C: Container> Deref for
 /// that do not implement `Abelian` and only implement `Semigroup`. This means
 /// that it can be used in settings where the difference type does not support
 /// negation.
-pub struct SemigroupVariable<G, D, R, C = Vec<(D, <G as ScopeParent>::Timestamp, R)>>
+pub struct SemigroupVariable<G, C>
 where
     G: Scope<Timestamp: Lattice>,
-    D: Data,
-    R: Semigroup + 'static,
     C: Container,
 {
-    collection: Collection<G, D, R, C>,
+    collection: CollectionCore<G, C>,
     feedback: Handle<G, C>,
     step: <G::Timestamp as Timestamp>::Summary,
 }
 
-impl<G, D: Data, R: Semigroup, C: Container> SemigroupVariable<G, D, R, C>
+impl<G, C: Container> SemigroupVariable<G, C>
 where
     G: Scope<Timestamp: Lattice>,
     C: crate::collection::containers::ResultsIn<<G::Timestamp as Timestamp>::Summary>,
 {
     /// Creates a new initially empty `SemigroupVariable`.
     pub fn new(scope: &mut G, step: <G::Timestamp as Timestamp>::Summary) -> Self {
         let (feedback, updates) = scope.feedback(step.clone());
-        let collection = Collection::<G,D,R,C>::new(updates);
+        let collection = CollectionCore::<G,C>::new(updates);
         SemigroupVariable { collection, feedback, step }
     }
 
     /// Adds a new source of data to `self`.
-    pub fn set(self, result: &Collection<G, D, R, C>) -> Collection<G, D, R, C> {
+    pub fn set(self, result: &CollectionCore<G, C>) -> CollectionCore<G, C> {
         let step = self.step;
         result
             .results_in(step)
@@ -267,8 +266,8 @@ where
     }
 }
 
-impl<G: Scope, D: Data, R: Semigroup, C: Container> Deref for SemigroupVariable<G, D, R, C> where G::Timestamp: Lattice {
-    type Target = Collection<G, D, R, C>;
+impl<G: Scope, C: Container> Deref for SemigroupVariable<G, C> where G::Timestamp: Lattice {
+    type Target = CollectionCore<G, C>;
     fn deref(&self) -> &Self::Target {
         &self.collection
     }

doop/src/main.rs

@@ -11,7 +11,7 @@ use differential_dataflow::{AsCollection, Collection, Hashable};
 use differential_dataflow::ExchangeData as Data;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::input::Input;
-use differential_dataflow::operators::iterate::Variable;
+use differential_dataflow::operators::iterate::VariableRow;
 use differential_dataflow::operators::{Threshold, Join, JoinCore};
 use differential_dataflow::operators::arrange::ArrangeByKey;
 
@@ -56,7 +56,7 @@ type MethodInvocation = Instruction;
 
 /// Set-valued collection.
 pub struct Relation<'a, G: Scope, D: Data+Hashable> where G::Timestamp : Lattice {
-    variable: Variable<Child<'a, G, Iter>, D, Diff>,
+    variable: VariableRow<Child<'a, G, Iter>, D, Diff>,
     current: Collection<Child<'a, G, Iter>, D, Diff>,
 }
 
@@ -68,7 +68,7 @@ impl<'a, G: Scope, D: Data+Hashable> Relation<'a, G, D> where G::Timestamp : Lat
     /// Creates a new variable initialized with `source`.
     pub fn new_from(source: &Collection<Child<'a, G, Iter>, D, Diff>) -> Self {
         use ::timely::order::Product;
-        let variable = Variable::new_from(source.clone(), Product::new(Default::default(), 1));
+        let variable = VariableRow::new_from(source.clone(), Product::new(Default::default(), 1));
         Relation {
             variable: variable,
             current: source.clone(),

mdbook/src/chapter_2/chapter_2_7.md

@@ -94,7 +94,7 @@ As an example, the implementation of the `iterate` operator looks something like
 # {
 #     (*variable).clone()
 # }
-# fn example<'a, G: Scope<Timestamp=u64>>(collection: &Collection<G, (u64, u64)>) //, logic: impl Fn(&Variable<Child<'a, G, G::Timestamp>, (u64, u64), isize>) -> Collection<Child<'a, G, G::Timestamp>, (u64, u64)>)
+# fn example<'a, G: Scope<Timestamp=u64>>(collection: &Collection<G, (u64, u64)>) //, logic: impl Fn(&VariableRow<Child<'a, G, G::Timestamp>, (u64, u64), isize>) -> Collection<Child<'a, G, G::Timestamp>, (u64, u64)>)
 #    where G::Timestamp: Lattice
 # {
     collection.scope().scoped("inner", |subgraph| {