Type Alias Collection 

pub type Collection<G, D, R = isize> = Collection<G, Vec<(D, <G as ScopeParent>::Timestamp, R)>>;

An evolving collection of values of type D, backed by Rust Vec types as containers.

The Collection type is the core abstraction in differential dataflow programs. As you write your differential dataflow computation, you write as if the collection is a static dataset to which you apply functional transformations, creating new collections. Once your computation is written, you are able to mutate the collection (by inserting and removing elements); differential dataflow will propagate changes through your functional computation and report the corresponding changes to the output collections.

Each vec collection has three generic parameters. The parameter G is for the scope in which the collection exists; as you write more complicated programs you may wish to introduce nested scopes (e.g. for iteration) and this parameter tracks the scope (for timely dataflow’s benefit). The D 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.

This type definition instantiates the Collection type with a Vec<(D, G::Timestamp, R)>.

Aliased Type

pub struct Collection<G, D, R = isize> {
    pub inner: Stream<G, Vec<(D, <G as ScopeParent>::Timestamp, R)>>,
}

Fields

inner: Stream<G, Vec<(D, <G as ScopeParent>::Timestamp, R)>>

The underlying timely dataflow stream.

This field is exposed to support direct timely dataflow manipulation when required, but it is not intended to be the idiomatic way to work with the collection.

The timestamp in the data is required to always be at least the timestamp of the data, in the timely-dataflow sense. If this invariant is not upheld, differential operators may behave unexpectedly.

Implementations

impl<G, D, R, T, TOuter> VecCollection<G, D, R>

where G: Scope<Timestamp = Product<TOuter, PointStamp<T>>>, D: Data, R: Semigroup + 'static, T: Timestamp + Default, TOuter: Timestamp,

pub fn enter_dynamic(self, _level: usize) -> Self

Enters a dynamically created scope which has level timestamp coordinates.

pub fn leave_dynamic(self, level: usize) -> Self

Leaves a dynamically created scope which has level timestamp coordinates.

impl<G: Scope, D: Clone + 'static, R: Clone + 'static> Collection<G, D, R>

pub fn map<D2, L>(self, logic: L) -> Collection<G, D2, R>

where D2: Clone + 'static, L: FnMut(D) -> D2 + 'static,

Creates a new collection by applying the supplied function to each input element.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    scope.new_collection_from(1 .. 10).1
         .map(|x| x * 2)
         .filter(|x| x % 2 == 1)
         .assert_empty();
});

pub fn map_in_place<L>(self, logic: L) -> Collection<G, D, R>

where L: FnMut(&mut D) + 'static,

Creates a new collection by applying the supplied function to each input element.

Although the name suggests in-place mutation, this function does not change the source collection, but rather re-uses the underlying allocations in its implementation. The method is semantically equivalent to map, but can be more efficient.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    scope.new_collection_from(1 .. 10).1
         .map_in_place(|x| *x *= 2)
         .filter(|x| x % 2 == 1)
         .assert_empty();
});

pub fn flat_map<I, L>(self, logic: L) -> Collection<G, I::Item, R>

where G::Timestamp: Clone, I: IntoIterator<Item: Clone + 'static>, L: FnMut(D) -> I + 'static,

Creates a new collection by applying the supplied function to each input element and accumulating the results.

This method extracts an iterator from each input element, and extracts the full contents of the iterator. Be warned that if the iterators produce substantial amounts of data, they are currently fully drained before attempting to consolidate the results.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    scope.new_collection_from(1 .. 10).1
         .flat_map(|x| 0 .. x);
});

pub fn filter<L>(self, logic: L) -> Collection<G, D, R>

where L: FnMut(&D) -> bool + 'static,

Creates a new collection containing those input records satisfying the supplied predicate.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    scope.new_collection_from(1 .. 10).1
         .map(|x| x * 2)
         .filter(|x| x % 2 == 1)
         .assert_empty();
});

pub fn explode<D2, R2, I, L>( self, logic: L, ) -> Collection<G, D2, <R2 as Multiply<R>>::Output>

where D2: Clone + 'static, R2: Semigroup + Multiply<R, Output: Semigroup + 'static>, I: IntoIterator<Item = (D2, R2)>, L: FnMut(D) -> I + 'static,

Replaces each record with another, with a new difference type.

This method is most commonly used to take records containing aggregatable data (e.g. numbers to be summed) and move the data into the difference component. This will allow differential dataflow to update in-place.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let nums = scope.new_collection_from(0 .. 10).1;
    let x1 = nums.clone().flat_map(|x| 0 .. x);
    let x2 = nums.map(|x| (x, 9 - x))
                 .explode(|(x,y)| Some((x,y)));

    x1.assert_eq(x2);
});

pub fn join_function<D2, R2, I, L>( self, logic: L, ) -> Collection<G, D2, <R2 as Multiply<R>>::Output>

where G::Timestamp: Lattice, D2: Clone + 'static, R2: Semigroup + Multiply<R, Output: Semigroup + 'static>, I: IntoIterator<Item = (D2, G::Timestamp, R2)>, L: FnMut(D) -> I + 'static,

Joins each record against a collection defined by the function logic.

This method performs what is essentially a join with the collection of records (x, logic(x)). Rather than materialize this second relation, logic is applied to each record and the appropriate modifications made to the results, namely joining timestamps and multiplying differences.

#Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    // creates `x` copies of `2*x` from time `3*x` until `4*x`,
    // for x from 0 through 9.
    scope.new_collection_from(0 .. 10isize).1
         .join_function(|x|
             //   data      time      diff
             vec![(2*x, (3*x) as u64,  x),
                  (2*x, (4*x) as u64, -x)]
          );
});

pub fn enter_at<'a, T, F>( self, child: &Iterative<'a, G, T>, initial: F, ) -> Collection<Iterative<'a, G, T>, D, R>

where T: Timestamp + Hash, F: FnMut(&D) -> T + Clone + 'static, G::Timestamp: Hash,

Brings a Collection into a nested scope, at varying times.

The initial function indicates the time at which each element of the Collection should appear.

Examples

use timely::dataflow::Scope;
use differential_dataflow::input::Input;

::timely::example(|scope| {

    let data = scope.new_collection_from(1 .. 10).1;

    let result = scope.iterative::<u64,_,_>(|child| {
        data.clone()
            .enter_at(child, |x| *x)
            .leave()
    });

    data.assert_eq(result);
});

pub fn delay<F>(self, func: F) -> Collection<G, D, R>

where G::Timestamp: Hash, F: FnMut(&G::Timestamp) -> G::Timestamp + Clone + 'static,

Delays each difference by a supplied function.

It is assumed that func only advances timestamps; this is not verified, and things may go horribly wrong if that assumption is incorrect. It is also critical that func be monotonic: if two times are ordered, they should have the same order or compare equal once func is applied to them (this is because we advance the timely capability with the same logic, and it must remain less_equal to all of the data timestamps).

pub fn inspect<F>(self, func: F) -> Collection<G, D, R>

where F: FnMut(&(D, G::Timestamp, R)) + 'static,

Applies a supplied function to each update.

This method is most commonly used to report information back to the user, often for debugging purposes. Any function can be used here, but be warned that the incremental nature of differential dataflow does not guarantee that it will be called as many times as you might expect.

The (data, time, diff) triples indicate a change diff to the frequency of data which takes effect at the logical time time. When times are totally ordered (for example, usize), these updates reflect the changes along the sequence of collections. For partially ordered times, the mathematics are more interesting and less intuitive, unfortunately.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    scope.new_collection_from(1 .. 10).1
         .map_in_place(|x| *x *= 2)
         .filter(|x| x % 2 == 1)
         .inspect(|x| println!("error: {:?}", x));
});

pub fn inspect_batch<F>(self, func: F) -> Collection<G, D, R>

where F: FnMut(&G::Timestamp, &[(D, G::Timestamp, R)]) + 'static,

Applies a supplied function to each batch of updates.

This method is analogous to inspect, but operates on batches and reveals the timestamp of the timely dataflow capability associated with the batch of updates. The observed batching depends on how the system executes, and may vary run to run.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    scope.new_collection_from(1 .. 10).1
         .map_in_place(|x| *x *= 2)
         .filter(|x| x % 2 == 1)
         .inspect_batch(|t,xs| println!("errors @ {:?}: {:?}", t, xs));
});

pub fn assert_empty(self)

where D: ExchangeData + Hashable, R: ExchangeData + Hashable + Semigroup, G::Timestamp: Lattice + Ord,

Assert if the collection is ever non-empty.

Because this is a dataflow fragment, the test is only applied as the computation is run. If the computation is not run, or not run to completion, there may be un-exercised times at which the collection could be non-empty. Typically, a timely dataflow computation runs to completion on drop, and so clean exit from a program should indicate that this assertion never found cause to complain.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    scope.new_collection_from(1 .. 10).1
         .map(|x| x * 2)
         .filter(|x| x % 2 == 1)
         .assert_empty();
});

impl<G: Scope<Timestamp: Clone + 'static>, D: Clone + 'static, R: Abelian + 'static> Collection<G, D, R>

Methods requiring an Abelian difference, to support negation.

pub fn assert_eq(self, other: Self)

where D: ExchangeData + Hashable, R: ExchangeData + Hashable, G::Timestamp: Lattice + Ord,

Assert if the collections are ever different.

Because this is a dataflow fragment, the test is only applied as the computation is run. If the computation is not run, or not run to completion, there may be un-exercised times at which the collections could vary. Typically, a timely dataflow computation runs to completion on drop, and so clean exit from a program should indicate that this assertion never found cause to complain.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let data = scope.new_collection_from(1 .. 10).1;

    let odds = data.clone().filter(|x| x % 2 == 1);
    let evens = data.clone().filter(|x| x % 2 == 0);

    odds.concat(evens)
        .assert_eq(data);
});

impl<G, K, V, R> Collection<G, (K, V), R>

where G: Scope<Timestamp: Lattice + Ord>, K: ExchangeData + Hashable, V: ExchangeData, R: ExchangeData + Semigroup,

pub fn reduce<L, V2: Data, R2: Ord + Abelian + 'static>( self, logic: L, ) -> Collection<G, (K, V2), R2>

where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,

Applies a reduction function on records grouped by key.

Input data must be structured as (key, val) pairs. The user-supplied reduction function takes as arguments

1. a reference to the key,
2. a reference to the slice of values and their accumulated updates,
3. a mutuable reference to a vector to populate with output values and accumulated updates.

The user logic is only invoked for non-empty input collections, and it is safe to assume that the slice of input values is non-empty. The values are presented in sorted order, as defined by their Ord implementations.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    // report the smallest value for each group
    scope.new_collection_from(1 .. 10).1
         .map(|x| (x / 3, x))
         .reduce(|_key, input, output| {
             output.push((*input[0].0, 1))
         });
});

pub fn reduce_named<L, V2: Data, R2: Ord + Abelian + 'static>( self, name: &str, logic: L, ) -> Collection<G, (K, V2), R2>

where L: FnMut(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,

As reduce with the ability to name the operator.

pub fn reduce_abelian<L, Bu, T2>( self, name: &str, logic: L, ) -> Arranged<G, TraceAgent<T2>>

where T2: for<'a> Trace<Key<'a> = &'a K, KeyOwn = K, ValOwn = V, Time = G::Timestamp, Diff: Abelian> + 'static, Bu: Builder<Time = T2::Time, Input = Vec<((K, V), T2::Time, T2::Diff)>, Output = T2::Batch>, L: FnMut(&K, &[(&V, R)], &mut Vec<(V, T2::Diff)>) + 'static,

Applies reduce to arranged data, and returns an arrangement of output data.

This method is used by the more ergonomic reduce, distinct, and count methods, although it can be very useful if one needs to manually attach and re-use existing arranged collections.

Examples

use differential_dataflow::input::Input;
use differential_dataflow::trace::Trace;
use differential_dataflow::trace::implementations::{ValBuilder, ValSpine};

::timely::example(|scope| {

    let trace =
    scope.new_collection_from(1 .. 10u32).1
         .map(|x| (x, x))
         .reduce_abelian::<_,ValBuilder<_,_,_,_>,ValSpine<_,_,_,_>>(
            "Example",
             move |_key, src, dst| dst.push((*src[0].0, 1))
         )
         .trace;
});

pub fn reduce_core<L, Bu, T2>( self, name: &str, logic: L, ) -> Arranged<G, TraceAgent<T2>>

where V: Clone + 'static, T2: for<'a> Trace<Key<'a> = &'a K, KeyOwn = K, ValOwn = V, Time = G::Timestamp> + 'static, Bu: Builder<Time = T2::Time, Input = Vec<((K, V), T2::Time, T2::Diff)>, Output = T2::Batch>, L: FnMut(&K, &[(&V, R)], &mut Vec<(V, T2::Diff)>, &mut Vec<(V, T2::Diff)>) + 'static,

Solves for output updates when presented with inputs and would-be outputs.

Unlike reduce_arranged, this method may be called with an empty input, and it may not be safe to index into the first element. At least one of the two collections will be non-empty.

impl<G, K, R1> Collection<G, K, R1>

where G: Scope<Timestamp: Lattice + Ord>, K: ExchangeData + Hashable, R1: ExchangeData + Semigroup,

pub fn distinct(self) -> Collection<G, K, isize>

Reduces the collection to one occurrence of each distinct element.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    // report at most one of each key.
    scope.new_collection_from(1 .. 10).1
         .map(|x| x / 3)
         .distinct();
});

pub fn distinct_core<R2: Ord + Abelian + 'static + From<i8>>( self, ) -> Collection<G, K, R2>

Distinct for general integer differences.

This method allows distinct to produce collections whose difference type is something other than an isize integer, for example perhaps an i32.

pub fn threshold<R2: Ord + Abelian + 'static, F: FnMut(&K, &R1) -> R2 + 'static>( self, thresh: F, ) -> Collection<G, K, R2>

Transforms the multiplicity of records.

The threshold function is obliged to map R1::zero to R2::zero, or at least the computation may behave as if it does. Otherwise, the transformation can be nearly arbitrary: the code does not assume any properties of threshold.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    // report at most one of each key.
    scope.new_collection_from(1 .. 10).1
         .map(|x| x / 3)
         .threshold(|_,c| c % 2);
});

pub fn threshold_named<R2: Ord + Abelian + 'static, F: FnMut(&K, &R1) -> R2 + 'static>( self, name: &str, thresh: F, ) -> Collection<G, K, R2>

A threshold with the ability to name the operator.

impl<G, K, R> Collection<G, K, R>

where G: Scope<Timestamp: Lattice + Ord>, K: ExchangeData + Hashable, R: ExchangeData + Semigroup,

pub fn count(self) -> Collection<G, (K, R), isize>

Counts the number of occurrences of each element.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {
    // report the number of occurrences of each key
    scope.new_collection_from(1 .. 10).1
         .map(|x| x / 3)
         .count();
});

pub fn count_core<R2: Ord + Abelian + From<i8> + 'static>( self, ) -> Collection<G, (K, R), R2>

Count for general integer differences.

This method allows count to produce collections whose difference type is something other than an isize integer, for example perhaps an i32.

impl<G, D, R> Collection<G, D, R>

where G: Scope<Timestamp: Clone + 'static + Lattice>, D: ExchangeData + Hashable, R: ExchangeData + Semigroup,

Methods which require data be arrangeable.

pub fn consolidate(self) -> Self

Aggregates the weights of equal records into at most one record.

This method uses the type D’s hashed() method to partition the data. The data are accumulated in place, each held back until their timestamp has completed.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let x = scope.new_collection_from(1 .. 10u32).1;

    x.clone()
     .negate()
     .concat(x)
     .consolidate() // <-- ensures cancellation occurs
     .assert_empty();
});

pub fn consolidate_named<Ba, Bu, Tr, F>(self, name: &str, reify: F) -> Self

where Ba: Batcher<Input = Vec<((D, ()), G::Timestamp, R)>, Time = G::Timestamp> + 'static, Tr: for<'a> Trace<Time = G::Timestamp, Diff = R> + 'static, Bu: Builder<Time = Tr::Time, Input = Ba::Output, Output = Tr::Batch>, F: Fn(Tr::Key<'_>, Tr::Val<'_>) -> D + 'static,

As consolidate but with the ability to name the operator, specify the trace type, and provide the function reify to produce owned keys and values..

pub fn consolidate_stream(self) -> Self

Aggregates the weights of equal records.

Unlike consolidate, this method does not exchange data and does not ensure that at most one copy of each (data, time) pair exists in the results. Instead, it acts on each batch of data and collapses equivalent (data, time) pairs found therein, suppressing any that accumulate to zero.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let x = scope.new_collection_from(1 .. 10u32).1;

    // nothing to assert, as no particular guarantees.
    x.clone()
     .negate()
     .concat(x)
     .consolidate_stream();
});

impl<G, K: ExchangeData + Hashable, V: ExchangeData, R: ExchangeData + Semigroup> Collection<G, (K, V), R>

where G: Scope<Timestamp: Lattice + Ord>,

pub fn arrange_by_key( self, ) -> Arranged<G, TraceAgent<ValSpine<K, V, G::Timestamp, R>>>

Arranges a collection of (Key, Val) records by Key.

This operator arranges a stream of values into a shared trace, whose contents it maintains. This trace is current for all times completed by the output stream, which can be used to safely identify the stable times and values in the trace.

pub fn arrange_by_key_named( self, name: &str, ) -> Arranged<G, TraceAgent<ValSpine<K, V, G::Timestamp, R>>>

As arrange_by_key but with the ability to name the arrangement.

impl<G, K: ExchangeData + Hashable, R: ExchangeData + Semigroup> Collection<G, K, R>

where G: Scope<Timestamp: Lattice + Ord>,

pub fn arrange_by_self( self, ) -> Arranged<G, TraceAgent<KeySpine<K, G::Timestamp, R>>>

Arranges a collection of Key records by Key.

This operator arranges a collection of records into a shared trace, whose contents it maintains. This trace is current for all times complete in the output stream, which can be used to safely identify the stable times and values in the trace.

pub fn arrange_by_self_named( self, name: &str, ) -> Arranged<G, TraceAgent<KeySpine<K, G::Timestamp, R>>>

As arrange_by_self but with the ability to name the arrangement.

impl<G, K, V, R> Collection<G, (K, V), R>

where G: Scope<Timestamp: Lattice + Ord>, K: ExchangeData + Hashable, V: ExchangeData, R: ExchangeData + Semigroup,

pub fn join<V2, R2>( self, other: Collection<G, (K, V2), R2>, ) -> Collection<G, (K, (V, V2)), <R as Multiply<R2>>::Output>

where K: ExchangeData, V2: ExchangeData, R2: ExchangeData + Semigroup, R: Multiply<R2, Output: Semigroup + 'static>,

Matches pairs (key,val1) and (key,val2) based on key and yields pairs (key, (val1, val2)).

The join_map method may be more convenient for non-trivial processing pipelines.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let x = scope.new_collection_from(vec![(0, 1), (1, 3)]).1;
    let y = scope.new_collection_from(vec![(0, 'a'), (1, 'b')]).1;
    let z = scope.new_collection_from(vec![(0, (1, 'a')), (1, (3, 'b'))]).1;

    x.join(y)
     .assert_eq(z);
});

pub fn join_map<V2: ExchangeData, R2: ExchangeData + Semigroup, D: Data, L>( self, other: Collection<G, (K, V2), R2>, logic: L, ) -> Collection<G, D, <R as Multiply<R2>>::Output>

where R: Multiply<R2, Output: Semigroup + 'static>, L: FnMut(&K, &V, &V2) -> D + 'static,

Matches pairs (key,val1) and (key,val2) based on key and then applies a function.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let x = scope.new_collection_from(vec![(0, 1), (1, 3)]).1;
    let y = scope.new_collection_from(vec![(0, 'a'), (1, 'b')]).1;
    let z = scope.new_collection_from(vec![(1, 'a'), (3, 'b')]).1;

    x.join_map(y, |_key, &a, &b| (a,b))
     .assert_eq(z);
});

pub fn semijoin<R2: ExchangeData + Semigroup>( self, other: Collection<G, K, R2>, ) -> Collection<G, (K, V), <R as Multiply<R2>>::Output>

where R: Multiply<R2, Output: Semigroup + 'static>,

Matches pairs (key, val) and key based on key, producing the former with frequencies multiplied.

When the second collection contains frequencies that are either zero or one this is the more traditional relational semijoin. When the second collection may contain multiplicities, this operation may scale up the counts of the records in the first input.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let x = scope.new_collection_from(vec![(0, 1), (1, 3)]).1;
    let y = scope.new_collection_from(vec![0, 2]).1;
    let z = scope.new_collection_from(vec![(0, 1)]).1;

    x.semijoin(y)
     .assert_eq(z);
});

pub fn antijoin<R2: ExchangeData + Semigroup>( self, other: Collection<G, K, R2>, ) -> Collection<G, (K, V), R>

where R: Multiply<R2, Output = R> + Abelian + 'static,

Subtracts the semijoin with other from self.

In the case that other has multiplicities zero or one this results in a relational antijoin, in which we discard input records whose key is present in other. If the multiplicities could be other than zero or one, the semantic interpretation of this operator is less clear.

In almost all cases, you should ensure that other has multiplicities that are zero or one, perhaps by using the distinct operator.

Examples

use differential_dataflow::input::Input;

::timely::example(|scope| {

    let x = scope.new_collection_from(vec![(0, 1), (1, 3)]).1;
    let y = scope.new_collection_from(vec![0, 2]).1;
    let z = scope.new_collection_from(vec![(1, 3)]).1;

    x.antijoin(y)
     .assert_eq(z);
});

pub fn join_core<Tr2, I, L>( self, stream2: Arranged<G, Tr2>, result: L, ) -> Collection<G, I::Item, <R as Multiply<Tr2::Diff>>::Output>

where Tr2: for<'a> TraceReader<Key<'a> = &'a K, Time = G::Timestamp> + Clone + 'static, R: Multiply<Tr2::Diff, Output: Semigroup + 'static>, I: IntoIterator<Item: Data>, L: FnMut(&K, &V, Tr2::Val<'_>) -> I + 'static,

Joins two arranged collections with the same key type.

Each matching pair of records (key, val1) and (key, val2) are subjected to the result function, which produces something implementing IntoIterator, where the output collection will have an entry for every value returned by the iterator.

This trait is implemented for arrangements (Arranged<G, T>) rather than collections. The Join trait contains the implementations for collections.

Examples

use differential_dataflow::input::Input;
use differential_dataflow::trace::Trace;

::timely::example(|scope| {

    let x = scope.new_collection_from(vec![(0u32, 1), (1, 3)]).1
                 .arrange_by_key();
    let y = scope.new_collection_from(vec![(0, 'a'), (1, 'b')]).1
                 .arrange_by_key();

    let z = scope.new_collection_from(vec![(1, 'a'), (3, 'b')]).1;

    x.join_core(y, |_key, &a, &b| Some((a, b)))
     .assert_eq(z);
});

Trait Implementations

impl<G, K: ExchangeData + Hashable, R: ExchangeData + Semigroup> Arrange<G, Vec<((K, ()), <G as ScopeParent>::Timestamp, R)>> for Collection<G, K, R>

where G: Scope<Timestamp: Lattice + Ord>,

fn arrange_named<Ba, Bu, Tr>(self, name: &str) -> Arranged<G, TraceAgent<Tr>>

where Ba: Batcher<Input = Vec<((K, ()), G::Timestamp, R)>, Time = G::Timestamp> + 'static, Bu: Builder<Time = G::Timestamp, Input = Ba::Output, Output = Tr::Batch>, Tr: Trace<Time = G::Timestamp> + 'static,

Arranges updates into a shared trace, with a supplied name.

fn arrange<Ba, Bu, Tr>(self) -> Arranged<G, TraceAgent<Tr>>

where Ba: Batcher<Input = C, Time = G::Timestamp> + 'static, Bu: Builder<Time = G::Timestamp, Input = Ba::Output, Output = Tr::Batch>, Tr: Trace<Time = G::Timestamp> + 'static,

Arranges updates into a shared trace.

impl<G, K, V, R> Arrange<G, Vec<((K, V), <G as ScopeParent>::Timestamp, R)>> for Collection<G, (K, V), R>

where G: Scope<Timestamp: Lattice>, K: ExchangeData + Hashable, V: ExchangeData, R: ExchangeData + Semigroup,

fn arrange_named<Ba, Bu, Tr>(self, name: &str) -> Arranged<G, TraceAgent<Tr>>

where Ba: Batcher<Input = Vec<((K, V), G::Timestamp, R)>, Time = G::Timestamp> + 'static, Bu: Builder<Time = G::Timestamp, Input = Ba::Output, Output = Tr::Batch>, Tr: Trace<Time = G::Timestamp> + 'static,

Arranges updates into a shared trace, with a supplied name.

fn arrange<Ba, Bu, Tr>(self) -> Arranged<G, TraceAgent<Tr>>

where Ba: Batcher<Input = C, Time = G::Timestamp> + 'static, Bu: Builder<Time = G::Timestamp, Input = Ba::Output, Output = Tr::Batch>, Tr: Trace<Time = G::Timestamp> + 'static,

Arranges updates into a shared trace.

impl<G, K: ExchangeData + Hashable, R: ExchangeData + Semigroup> CountTotal<G, K, R> for VecCollection<G, K, R>

where G: Scope<Timestamp: TotalOrder + Lattice + Ord>,

fn count_total_core<R2: Semigroup + From<i8> + 'static>( self, ) -> VecCollection<G, (K, R), R2>

Count for general integer differences.

fn count_total(self) -> VecCollection<G, (K, R), isize>

Counts the number of occurrences of each element.

impl<G, D, R> Identifiers<G, D, R> for VecCollection<G, D, R>

where G: Scope<Timestamp: Lattice>, D: ExchangeData + Hash, R: ExchangeData + Abelian,

fn identifiers(self) -> VecCollection<G, (D, u64), R>

Assign unique identifiers to elements of a collection.

impl<G: Scope<Timestamp: Lattice>, D: Ord + Data + Debug, R: Abelian + 'static> Iterate<G, D, R> for VecCollection<G, D, R>

fn iterate<F>(self, logic: F) -> VecCollection<G, D, R>

where for<'a> F: FnOnce(Iterative<'a, G, u64>, VecCollection<Iterative<'a, G, u64>, D, R>) -> VecCollection<Iterative<'a, G, u64>, D, R>,

Iteratively apply logic to the source collection until convergence.

impl<G, K, D> PrefixSum<G, K, D> for VecCollection<G, ((usize, K), D)>

where G: Scope<Timestamp: Lattice>, K: ExchangeData + Hash, D: ExchangeData + Hash,

fn prefix_sum<F>(self, zero: D, combine: F) -> Self

where F: Fn(&K, &D, &D) -> D + 'static,

Computes the prefix sum for each element in the collection.

fn prefix_sum_at<F>( self, locations: VecCollection<G, (usize, K)>, zero: D, combine: F, ) -> Self

where F: Fn(&K, &D, &D) -> D + 'static,

Determine the prefix sum at each element of location.

impl<G, K: ExchangeData + Hashable, R: ExchangeData + Semigroup> ThresholdTotal<G, K, R> for VecCollection<G, K, R>

where G: Scope<Timestamp: TotalOrder + Lattice + Ord> + Scope,

fn threshold_semigroup<R2, F>(self, thresh: F) -> VecCollection<G, K, R2>

where R2: Semigroup + 'static, F: FnMut(&K, &R, Option<&R>) -> Option<R2> + 'static,

Reduces the collection to one occurrence of each distinct element.

fn threshold_total<R2: Abelian + 'static, F: FnMut(&K, &R) -> R2 + 'static>( self, thresh: F, ) -> VecCollection<G, K, R2>

Reduces the collection to one occurrence of each distinct element.

fn distinct_total(self) -> VecCollection<G, K, isize>

Reduces the collection to one occurrence of each distinct element.

fn distinct_total_core<R2: Abelian + From<i8> + 'static>( self, ) -> VecCollection<G, K, R2>

Distinct for general integer differences.