Struct differential_dataflow::operators::iterate::Variable

pub struct Variable<'a, G: Scope, D: Data, R: Diff> where
    G::Timestamp: Lattice,  { /* fields omitted */ }

A differential dataflow collection variable

The Variable struct allows differential dataflow programs requiring more sophisticated iterative patterns than singly recursive iteration. For example: in mutual recursion two collections evolve simultaneously.

Examples

The following example is equivalent to the example for the Iterate trait.

extern crate timely;
extern crate differential_dataflow;

use timely::dataflow::Scope;

use differential_dataflow::input::Input;
use differential_dataflow::operators::iterate::Variable;
use differential_dataflow::operators::Consolidate;

fn main() {
    ::timely::example(|scope| {

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

        scope.scoped(|nested| {
            let variable = Variable::from(numbers.enter(nested));
            let result = variable.map(|x| if x % 2 == 0 { x/2 } else { x })
                                 .consolidate();
            variable.set(&result)
                    .leave()
        });
    })
}

Methods

impl<'a, G: Scope, D: Data, R: Diff> Variable<'a, G, D, R> where
    G::Timestamp: Lattice, 

fn from(source: Collection<Child<'a, G, u64>, D, R>) -> Variable<'a, G, D, R>

Creates a new Variable and a Stream representing its output, from a supplied source stream.

fn set(
    self,
    result: &Collection<Child<'a, G, u64>, D, R>
) -> Collection<Child<'a, G, u64>, D, R>

Adds a new source of data to the Variable.

Methods from Deref<Target = Collection<Child<'a, G, u64>, D, R>>

fn map<D2, L>(&self, logic: L) -> Collection<G, D2, R> where
    D2: Data,
    L: Fn(D) -> D2 + 'static, 

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

Examples

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;

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

fn map_in_place<L>(&self, logic: L) -> Collection<G, D, R> where
    L: Fn(&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

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

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

fn flat_map<I, L>(&self, logic: L) -> Collection<G, I::Item, R> where
    G::Timestamp: Clone,
    I: IntoIterator,
    I::Item: Data,
    L: Fn(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

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

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

fn negate(&self) -> Collection<G, D, R>

Creates a new collection whose counts are the negation of those in the input.

This method is most commonly used with concat to get those element in one collection but not another. However, differential dataflow computations are still defined for all values of the difference type R, including negative counts.

Examples

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

fn main() {
    ::timely::example(|scope| {

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

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

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

fn filter<L>(&self, logic: L) -> Collection<G, D, R> where
    L: Fn(&D) -> bool + 'static, 

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

Examples

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

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

fn concat(&self, other: &Collection<G, D, R>) -> Collection<G, D, R>

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 implementation is the concatenation of the stream of updates, but it corresponds to the addition of the two collections.

Examples

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

fn main() {
    ::timely::example(|scope| {

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

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

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

fn explode<D2, R2, I, L>(
    &self,
    logic: L
) -> Collection<G, D2, <R2 as Mul<R>>::Output> where
    D2: Data,
    R2: Diff + Mul<R>,
    <R2 as Mul<R>>::Output: Data + Diff,
    I: IntoIterator<Item = (D2, R2)>,
    L: Fn(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

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

fn main() {
    ::timely::example(|scope| {

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

        x1.assert_eq(&x2);
    });
}

fn enter<'a, T>(
    &self,
    child: &Child<'a, G, T>
) -> Collection<Child<'a, G, T>, D, R> where
    T: Timestamp, 

Brings a Collection into a nested scope.

Examples

extern crate timely;
extern crate differential_dataflow;

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

fn main() {
    ::timely::example(|scope| {

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

        let result = scope.scoped::<(),_,_>(|child| {
            data.enter(child)
                .leave()
        });

        data.assert_eq(&result);
    });
}

fn enter_at<'a, T, F>(
    &self,
    child: &Child<'a, G, T>,
    initial: F
) -> Collection<Child<'a, G, T>, D, R> where
    T: Timestamp,
    F: Fn(&D) -> T + 'static,
    G::Timestamp: Hash,
    T: 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

extern crate timely;
extern crate differential_dataflow;

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

fn main() {
    ::timely::example(|scope| {

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

        let result = scope.scoped(|child| {
            data.enter_at(child, |x| *x)
                .leave()
        });

        data.assert_eq(&result);
    });
}

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

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

fn main() {
    ::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));
    });
}

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

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

fn main() {
    ::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));
    });
}

fn probe(&self) -> Handle<G::Timestamp>

Attaches a timely dataflow probe to the output of a Collection.

This probe is used to determine when the state of the Collection has stabilized and can be read out.

fn probe_with(&self, handle: &mut Handle<G::Timestamp>) -> Collection<G, D, R>

Attaches a timely dataflow probe to the output of a Collection.

This probe is used to determine when the state of the Collection has stabilized and all updates observed. In addition, a probe is also often use to limit the number of rounds of input in flight at any moment; a computation can wait until the probe has caught up to the input before introducing more rounds of data, to avoid swamping the system.

fn assert_eq(&self, other: &Self) where
    D: Data + 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

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

fn main() {
    ::timely::example(|scope| {

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

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

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

fn assert_empty(&self) where
    D: Data + Hashable,
    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

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::*;

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

fn scope(&self) -> G

The scope containing the underlying timely dataflow stream.

fn leave(&self) -> Collection<G, D, R>

Returns the final value of a Collection from a nested scope to its containing scope.

Examples

extern crate timely;
extern crate differential_dataflow;

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

fn main() {
    ::timely::example(|scope| {

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

        let result = scope.scoped::<(),_,_>(|child| {
            data.enter(child)
                .leave()
        });

        data.assert_eq(&result);
    });
}

Trait Implementations

impl<'a, G: Scope, D: Data, R: Diff> Deref for Variable<'a, G, D, R> where
    G::Timestamp: Lattice, 

type Target = Collection<Child<'a, G, u64>, D, R>

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.