Struct differential_dataflow::operators::arrange::arrangement::Arranged

pub struct Arranged<G: Scope, Tr> where
    G::Timestamp: Lattice + Ord,
    Tr: TraceReader + Clone,  {
    pub stream: Stream<G, Tr::Batch>,
    pub trace: Tr,
}

An arranged collection of (K,V) values.

An Arranged allows multiple differential operators to share the resources (communication, computation, memory) required to produce and maintain an indexed representation of a collection.

Fields

stream: Stream<G, Tr::Batch>

A stream containing arranged updates.

This stream contains the same batches of updates the trace itself accepts, so there should be no additional overhead to receiving these records. The batches can be navigated just as the batches in the trace, by key and by value.

trace: Tr

A shared trace, updated by the Arrange operator and readable by others.

Implementations

impl<G: Scope, Tr> Arranged<G, Tr> where
    G::Timestamp: Lattice + Ord,
    Tr: TraceReader<Time = G::Timestamp> + Clone,
    Tr::Batch: BatchReader<Tr::Key, Tr::Val, G::Timestamp, Tr::R>,
    Tr::Cursor: Cursor<Tr::Key, Tr::Val, G::Timestamp, Tr::R>, 

pub fn enter<'a, TInner>(
    &self,
    child: &Child<'a, G, TInner>
) -> Arranged<Child<'a, G, TInner>, TraceEnter<Tr, TInner>> where
    Tr::Key: 'static,
    Tr::Val: 'static,
    Tr::R: 'static,
    G::Timestamp: Clone + 'static,
    TInner: Refines<G::Timestamp> + Lattice + Timestamp + Clone + 'static, 

Brings an arranged collection into a nested scope.

This method produces a proxy trace handle that uses the same backing data, but acts as if the timestamps have all been extended with an additional coordinate with the default value. The resulting collection does not vary with the new timestamp coordinate.

pub fn enter_region<'a>(
    &self,
    child: &Child<'a, G, G::Timestamp>
) -> Arranged<Child<'a, G, G::Timestamp>, Tr> where
    Tr::Key: 'static,
    Tr::Val: 'static,
    Tr::R: 'static,
    G::Timestamp: Clone + 'static, 

Brings an arranged collection into a nested region.

This method only applies to regions, which are subscopes with the same timestamp as their containing scope. In this case, the trace type does not need to change.

pub fn enter_at<'a, TInner, F, P>(
    &self,
    child: &Child<'a, G, TInner>,
    logic: F,
    prior: P
) -> Arranged<Child<'a, G, TInner>, TraceEnterAt<Tr, TInner, F, P>> where
    Tr::Key: 'static,
    Tr::Val: 'static,
    Tr::R: 'static,
    G::Timestamp: Clone + 'static,
    TInner: Refines<G::Timestamp> + Lattice + Timestamp + Clone + 'static,
    F: FnMut(&Tr::Key, &Tr::Val, &G::Timestamp) -> TInner + Clone + 'static,
    P: FnMut(&TInner) -> Tr::Time + Clone + 'static, 

Brings an arranged collection into a nested scope.

This method produces a proxy trace handle that uses the same backing data, but acts as if the timestamps have all been extended with an additional coordinate with the default value. The resulting collection does not vary with the new timestamp coordinate.

pub fn filter<F>(&self, logic: F) -> Arranged<G, TraceFilter<Tr, F>> where
    Tr::Key: 'static,
    Tr::Val: 'static,
    Tr::R: 'static,
    G::Timestamp: Clone + 'static,
    F: FnMut(&Tr::Key, &Tr::Val) -> bool + Clone + 'static, 

Filters an arranged collection.

This method produces a new arrangement backed by the same shared arrangement as self, paired with user-specified logic that can filter by key and value. The resulting collection is restricted to the keys and values that return true under the user predicate.

Examples

extern crate timely;
extern crate differential_dataflow;

use differential_dataflow::input::Input;
use differential_dataflow::operators::arrange::ArrangeByKey;

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

        let arranged =
        scope.new_collection_from(0 .. 10).1
             .map(|x| (x, x+1))
             .arrange_by_key();

        arranged
            .filter(|k,v| k == v)
            .as_collection(|k,v| (*k,*v))
            .assert_empty();
    });
}

pub fn as_collection<D: Data, L>(&self, logic: L) -> Collection<G, D, Tr::R> where
    Tr::R: Semigroup,
    L: FnMut(&Tr::Key, &Tr::Val) -> D + 'static, 

Flattens the stream into a Collection.

The underlying Stream<G, BatchWrapper<T::Batch>> is a much more efficient way to access the data, and this method should only be used when the data need to be transformed or exchanged, rather than supplied as arguments to an operator using the same key-value structure.

pub fn flat_map_ref<I, L>(&self, logic: L) -> Collection<G, I::Item, Tr::R> where
    Tr::R: Semigroup,
    I: IntoIterator,
    I::Item: Data,
    L: FnMut(&Tr::Key, &Tr::Val) -> I + 'static, 

Extracts elements from an arrangement as a collection.

The supplied logic may produce an iterator over output values, allowing either filtering or flat mapping as part of the extraction.

pub fn flat_map_batches<I, L>(
    stream: &Stream<G, Tr::Batch>,
    logic: L
) -> Collection<G, I::Item, Tr::R> where
    Tr::R: Semigroup,
    I: IntoIterator,
    I::Item: Data,
    L: FnMut(&Tr::Key, &Tr::Val) -> I + 'static, 

Extracts elements from a stream of batches as a collection.

The supplied logic may produce an iterator over output values, allowing either filtering or flat mapping as part of the extraction.

This method exists for streams of batches without the corresponding arrangement. If you have the arrangement, its flat_map_ref method is equivalent to this.

pub fn lookup(
    &self,
    queries: &Stream<G, (Tr::Key, G::Timestamp)>
) -> Stream<G, (Tr::Key, Tr::Val, G::Timestamp, Tr::R)> where
    G::Timestamp: Data + Lattice + Ord + TotalOrder,
    Tr::Key: ExchangeData + Hashable,
    Tr::Val: ExchangeData,
    Tr::R: ExchangeData + Semigroup,
    Tr: 'static, 

Report values associated with keys at certain times.

This method consumes a stream of (key, time) queries and reports the corresponding stream of (key, value, time, diff) accumulations in the self trace.

impl<'a, G: Scope, Tr> Arranged<Child<'a, G, G::Timestamp>, Tr> where
    G::Timestamp: Lattice + Ord,
    Tr: TraceReader<Time = G::Timestamp> + Clone,
    Tr::Batch: BatchReader<Tr::Key, Tr::Val, G::Timestamp, Tr::R>,
    Tr::Cursor: Cursor<Tr::Key, Tr::Val, G::Timestamp, Tr::R>, 

pub fn leave_region(&self) -> Arranged<G, Tr>

Brings an arranged collection out of a nested region.

This method only applies to regions, which are subscopes with the same timestamp as their containing scope. In this case, the trace type does not need to change.

Trait Implementations

impl<G: Scope, Tr> Clone for Arranged<G, Tr> where
    G::Timestamp: Lattice + Ord,
    Tr: TraceReader<Time = G::Timestamp> + Clone,
    Tr::Batch: BatchReader<Tr::Key, Tr::Val, G::Timestamp, Tr::R>,
    Tr::Cursor: Cursor<Tr::Key, Tr::Val, G::Timestamp, Tr::R>, 

fn clone(&self) -> Self

Returns a copy of the value.

pub fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<G: Scope, K: Data, T1, R: Semigroup> Count<G, K, R> for Arranged<G, T1> where
    G::Timestamp: Lattice + Ord,
    T1: TraceReader<Key = K, Val = (), Time = G::Timestamp, R = R> + Clone + 'static,
    T1::Batch: BatchReader<K, (), G::Timestamp, R>,
    T1::Cursor: Cursor<K, (), G::Timestamp, R>, 

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

Counts the number of occurrences of each element.

impl<G: Scope, T1> CountTotal<G, <T1 as TraceReader>::Key, <T1 as TraceReader>::R> for Arranged<G, T1> where
    G::Timestamp: TotalOrder + Lattice + Ord,
    T1: TraceReader<Val = (), Time = G::Timestamp> + Clone + 'static,
    T1::Key: ExchangeData,
    T1::R: ExchangeData + Semigroup,
    T1::Batch: BatchReader<T1::Key, (), G::Timestamp, T1::R>,
    T1::Cursor: Cursor<T1::Key, (), G::Timestamp, T1::R>, 

fn count_total(&self) -> Collection<G, (T1::Key, T1::R), isize>

Counts the number of occurrences of each element.

impl<G, Tr> Join<G, <Tr as TraceReader>::Key, <Tr as TraceReader>::Val, <Tr as TraceReader>::R> for Arranged<G, Tr> where
    G: Scope,
    G::Timestamp: Lattice + Ord,
    Tr: TraceReader<Time = G::Timestamp> + Clone + 'static,
    Tr::Key: Data + Hashable,
    Tr::Val: Data,
    Tr::R: Semigroup,
    Tr::Batch: BatchReader<Tr::Key, Tr::Val, G::Timestamp, Tr::R> + 'static,
    Tr::Cursor: Cursor<Tr::Key, Tr::Val, G::Timestamp, Tr::R> + 'static, 

fn join_map<V2: ExchangeData, R2: ExchangeData + Semigroup, D: Data, L>(
    &self,
    other: &Collection<G, (Tr::Key, V2), R2>,
    logic: L
) -> Collection<G, D, <Tr::R as Mul<R2>>::Output> where
    Tr::Key: ExchangeData,
    Tr::R: Mul<R2>,
    <Tr::R as Mul<R2>>::Output: Semigroup,
    L: FnMut(&Tr::Key, &Tr::Val, &V2) -> D + 'static, 

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

fn semijoin<R2: ExchangeData + Semigroup>(
    &self,
    other: &Collection<G, Tr::Key, R2>
) -> Collection<G, (Tr::Key, Tr::Val), <Tr::R as Mul<R2>>::Output> where
    Tr::Key: ExchangeData,
    Tr::R: Mul<R2>,
    <Tr::R as Mul<R2>>::Output: Semigroup, 

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

fn antijoin<R2: ExchangeData + Semigroup>(
    &self,
    other: &Collection<G, Tr::Key, R2>
) -> Collection<G, (Tr::Key, Tr::Val), Tr::R> where
    Tr::Key: ExchangeData,
    Tr::R: Mul<R2, Output = Tr::R>,
    Tr::R: Abelian, 

Subtracts the semijoin with other from self.

fn join<V2, R2>(
    &self,
    other: &Collection<G, (K, V2), R2>
) -> Collection<G, (K, (V, V2)), <R as Mul<R2>>::Output> where
    K: ExchangeData,
    V2: ExchangeData,
    R2: ExchangeData + Semigroup,
    R: Mul<R2>,
    <R as Mul<R2>>::Output: Semigroup, 

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

impl<G, T1> JoinCore<G, <T1 as TraceReader>::Key, <T1 as TraceReader>::Val, <T1 as TraceReader>::R> for Arranged<G, T1> where
    G: Scope,
    G::Timestamp: Lattice + Ord + Debug,
    T1: TraceReader<Time = G::Timestamp> + Clone + 'static,
    T1::Key: Ord + Debug + 'static,
    T1::Val: Ord + Clone + Debug + 'static,
    T1::R: Semigroup,
    T1::Batch: BatchReader<T1::Key, T1::Val, G::Timestamp, T1::R> + 'static,
    T1::Cursor: Cursor<T1::Key, T1::Val, G::Timestamp, T1::R> + 'static, 

fn join_core<Tr2, I, L>(
    &self,
    other: &Arranged<G, Tr2>,
    result: L
) -> Collection<G, I::Item, <T1::R as Mul<Tr2::R>>::Output> where
    Tr2::Val: Ord + Clone + Debug + 'static,
    Tr2: TraceReader<Key = T1::Key, Time = G::Timestamp> + Clone + 'static,
    Tr2::Batch: BatchReader<T1::Key, Tr2::Val, G::Timestamp, Tr2::R> + 'static,
    Tr2::Cursor: Cursor<T1::Key, Tr2::Val, G::Timestamp, Tr2::R> + 'static,
    Tr2::R: Semigroup,
    T1::R: Mul<Tr2::R>,
    <T1::R as Mul<Tr2::R>>::Output: Semigroup,
    I: IntoIterator,
    I::Item: Data,
    L: FnMut(&T1::Key, &T1::Val, &Tr2::Val) -> I + 'static, 

Joins two arranged collections with the same key type.

impl<G: Scope, K: Data, V: Data, T1, R: Semigroup> Reduce<G, K, V, R> for Arranged<G, T1> where
    G::Timestamp: Lattice + Ord,
    T1: TraceReader<Key = K, Val = V, Time = G::Timestamp, R = R> + Clone + 'static,
    T1::Batch: BatchReader<K, V, G::Timestamp, R>,
    T1::Cursor: Cursor<K, V, G::Timestamp, R>, 

fn reduce_named<L, V2: Data, R2: Abelian>(
    &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.

fn reduce<L, V2: Data, R2: Abelian>(
    &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.

impl<G: Scope, K: Data, V: Data, T1, R: Semigroup> ReduceCore<G, K, V, R> for Arranged<G, T1> where
    G::Timestamp: Lattice + Ord,
    T1: TraceReader<Key = K, Val = V, Time = G::Timestamp, R = R> + Clone + 'static,
    T1::Batch: BatchReader<K, V, G::Timestamp, R>,
    T1::Cursor: Cursor<K, V, G::Timestamp, R>, 

fn reduce_core<L, T2>(
    &self,
    name: &str,
    logic: L
) -> Arranged<G, TraceAgent<T2>> where
    T2: Trace + TraceReader<Key = K, Time = G::Timestamp> + 'static,
    T2::Val: Data,
    T2::R: Semigroup,
    T2::Batch: Batch<K, T2::Val, G::Timestamp, T2::R>,
    T2::Cursor: Cursor<K, T2::Val, G::Timestamp, T2::R>,
    L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val, T2::R)>, &mut Vec<(T2::Val, T2::R)>) + 'static, 

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

fn reduce_abelian<L, T2>(
    &self,
    name: &str,
    logic: L
) -> Arranged<G, TraceAgent<T2>> where
    T2: Trace + TraceReader<Key = K, Time = G::Timestamp> + 'static,
    T2::Val: Data,
    T2::R: Abelian,
    T2::Batch: Batch<K, T2::Val, G::Timestamp, T2::R>,
    T2::Cursor: Cursor<K, T2::Val, G::Timestamp, T2::R>,
    L: FnMut(&K, &[(&V, R)], &mut Vec<(T2::Val, T2::R)>) + 'static, 

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

impl<G: Scope, K: Data, T1, R1: Semigroup> Threshold<G, K, R1> for Arranged<G, T1> where
    G::Timestamp: Lattice + Ord,
    T1: TraceReader<Key = K, Val = (), Time = G::Timestamp, R = R1> + Clone + 'static,
    T1::Batch: BatchReader<K, (), G::Timestamp, R1>,
    T1::Cursor: Cursor<K, (), G::Timestamp, R1>, 

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

A threshold with the ability to name the operator.

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

Transforms the multiplicity of records.

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

Reduces the collection to one occurrence of each distinct element.

fn distinct_core<R2: Abelian + From<i8>>(&self) -> Collection<G, K, R2>

Distinct for general integer differences.

impl<G: Scope, T1> ThresholdTotal<G, <T1 as TraceReader>::Key, <T1 as TraceReader>::R> for Arranged<G, T1> where
    G::Timestamp: TotalOrder + Lattice + Ord,
    T1: TraceReader<Val = (), Time = G::Timestamp> + Clone + 'static,
    T1::Key: ExchangeData,
    T1::R: ExchangeData + Semigroup,
    T1::Batch: BatchReader<T1::Key, (), G::Timestamp, T1::R>,
    T1::Cursor: Cursor<T1::Key, (), G::Timestamp, T1::R>, 

fn threshold_semigroup<R2, F>(&self, thresh: F) -> Collection<G, T1::Key, R2> where
    R2: Semigroup,
    F: FnMut(&T1::Key, &T1::R, Option<&T1::R>) -> Option<R2> + 'static, 

Reduces the collection to one occurrence of each distinct element.

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

Reduces the collection to one occurrence of each distinct element.

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

Reduces the collection to one occurrence of each distinct element.

fn distinct_total_core<R2: Abelian + From<i8>>(&self) -> Collection<G, K, R2>

Distinct for general integer differences.