Struct differential_dataflow::collection::Collection
source · Expand description
A mutable collection of values of type D
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 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.
Fields
inner: Stream<G, (D, G::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.
Implementations
sourceimpl<G: Scope, D: Data, R: Diff> Collection<G, D, R>where
G::Timestamp: Data,
impl<G: Scope, D: Data, R: Diff> Collection<G, D, R>where
G::Timestamp: Data,
sourcepub fn new(stream: Stream<G, (D, G::Timestamp, R)>) -> Collection<G, D, R>
pub fn new(stream: Stream<G, (D, G::Timestamp, R)>) -> Collection<G, D, R>
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
idiomatic approach to convert timely streams to collections. Also, the input::Input
trait
provides a new_collection
method which will create a new collection for you without exposing
the underlying timely stream at all.
sourcepub fn map<D2, L>(&self, logic: L) -> Collection<G, D2, R>where
D2: Data,
L: Fn(D) -> D2 + 'static,
pub 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();
});
}
sourcepub fn map_in_place<L>(&self, logic: L) -> Collection<G, D, R>where
L: Fn(&mut D) + 'static,
pub 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;
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();
});
}
sourcepub 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,
pub 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;
fn main() {
::timely::example(|scope| {
scope.new_collection_from(1 .. 10).1
.flat_map(|x| 0 .. x);
});
}
sourcepub fn negate(&self) -> Collection<G, D, R>
pub 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;
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);
});
}
sourcepub fn filter<L>(&self, logic: L) -> Collection<G, D, R>where
L: Fn(&D) -> bool + 'static,
pub 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;
fn main() {
::timely::example(|scope| {
scope.new_collection_from(1 .. 10).1
.map(|x| x * 2)
.filter(|x| x % 2 == 1)
.assert_empty();
});
}
sourcepub fn concat(&self, other: &Collection<G, D, R>) -> Collection<G, D, R>
pub 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;
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);
});
}
sourcepub 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,
pub 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;
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);
});
}
sourcepub fn enter<'a, T>(
&self,
child: &Child<'a, G, T>
) -> Collection<Child<'a, G, T>, D, R>where
T: Refines<<G as ScopeParent>::Timestamp>,
pub fn enter<'a, T>(
&self,
child: &Child<'a, G, T>
) -> Collection<Child<'a, G, T>, D, R>where
T: Refines<<G as ScopeParent>::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;
fn main() {
::timely::example(|scope| {
let data = scope.new_collection_from(1 .. 10).1;
let result = scope.region(|child| {
data.enter(child)
.leave()
});
data.assert_eq(&result);
});
}
sourcepub 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: Fn(&D) -> T + 'static,
G::Timestamp: Hash,
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: Fn(&D) -> T + '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
extern crate timely;
extern crate differential_dataflow;
use timely::dataflow::Scope;
use differential_dataflow::input::Input;
fn main() {
::timely::example(|scope| {
let data = scope.new_collection_from(1 .. 10).1;
let result = scope.iterative::<u64,_,_>(|child| {
data.enter_at(child, |x| *x)
.leave()
});
data.assert_eq(&result);
});
}
sourcepub fn delay<F>(&self, func: F) -> Collection<G, D, R>where
F: Fn(&G::Timestamp) -> G::Timestamp + 'static,
pub fn delay<F>(&self, func: F) -> Collection<G, D, R>where
F: Fn(&G::Timestamp) -> G::Timestamp + '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 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).
sourcepub fn inspect<F>(&self, func: F) -> Collection<G, D, R>where
F: FnMut(&(D, G::Timestamp, R)) + 'static,
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
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_in_place(|x| *x *= 2)
.filter(|x| x % 2 == 1)
.inspect(|x| println!("error: {:?}", x));
});
}
sourcepub fn inspect_batch<F>(&self, func: F) -> Collection<G, D, R>where
F: FnMut(&G::Timestamp, &[(D, G::Timestamp, R)]) + 'static,
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
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_in_place(|x| *x *= 2)
.filter(|x| x % 2 == 1)
.inspect_batch(|t,xs| println!("errors @ {:?}: {:?}", t, xs));
});
}
sourcepub fn probe(&self) -> Handle<G::Timestamp>
pub 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.
sourcepub fn probe_with(
&self,
handle: &mut Handle<G::Timestamp>
) -> Collection<G, D, R>
pub 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.
sourcepub fn assert_eq(&self, other: &Self)where
D: Data + Hashable,
G::Timestamp: Lattice + Ord,
pub 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;
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);
});
}
sourcepub fn assert_empty(&self)where
D: Data + Hashable,
G::Timestamp: Lattice + Ord,
pub 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;
fn main() {
::timely::example(|scope| {
scope.new_collection_from(1 .. 10).1
.map(|x| x * 2)
.filter(|x| x % 2 == 1)
.assert_empty();
});
}
sourceimpl<'a, G: Scope, T: Timestamp, D: Data, R: Diff> Collection<Child<'a, G, T>, D, R>where
T: Refines<<G as ScopeParent>::Timestamp>,
impl<'a, G: Scope, T: Timestamp, D: Data, R: Diff> Collection<Child<'a, G, T>, D, R>where
T: Refines<<G as ScopeParent>::Timestamp>,
sourcepub fn leave(&self) -> Collection<G, D, R>
pub 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;
fn main() {
::timely::example(|scope| {
let data = scope.new_collection_from(1 .. 10).1;
let result = scope.region(|child| {
data.enter(child)
.leave()
});
data.assert_eq(&result);
});
}
Trait Implementations
sourceimpl<G: Scope, K: Data + Hashable, R: Diff> Arrange<G, K, (), R> for Collection<G, K, R>where
G::Timestamp: Lattice + Ord,
impl<G: Scope, K: Data + Hashable, R: Diff> Arrange<G, K, (), R> for Collection<G, K, R>where
G::Timestamp: Lattice + Ord,
sourcefn arrange_named<T>(
&self,
name: &str
) -> Arranged<G, K, (), R, TraceAgent<K, (), G::Timestamp, R, T>>where
T: Trace<K, (), G::Timestamp, R> + 'static,
T::Batch: Batch<K, (), G::Timestamp, R>,
fn arrange_named<T>(
&self,
name: &str
) -> Arranged<G, K, (), R, TraceAgent<K, (), G::Timestamp, R, T>>where
T: Trace<K, (), G::Timestamp, R> + 'static,
T::Batch: Batch<K, (), G::Timestamp, R>,
(Key, Val)
updates by Key
. Accepts an empty instance of the trace type. Read moresourceimpl<G: Scope, K: Data + Hashable, V: Data, R: Diff> Arrange<G, K, V, R> for Collection<G, (K, V), R>where
G::Timestamp: Lattice + Ord,
impl<G: Scope, K: Data + Hashable, V: Data, R: Diff> Arrange<G, K, V, R> for Collection<G, (K, V), R>where
G::Timestamp: Lattice + Ord,
sourceimpl<G: Scope, K: Data + Hashable, V: Data, R: Diff> ArrangeByKey<G, K, V, R> for Collection<G, (K, V), R>where
G::Timestamp: Lattice + Ord,
impl<G: Scope, K: Data + Hashable, V: Data, R: Diff> ArrangeByKey<G, K, V, R> for Collection<G, (K, V), R>where
G::Timestamp: Lattice + Ord,
sourcefn arrange_by_key(
&self
) -> Arranged<G, K, V, R, TraceAgent<K, V, G::Timestamp, R, DefaultValTrace<K, V, G::Timestamp, R>>>
fn arrange_by_key(
&self
) -> Arranged<G, K, V, R, TraceAgent<K, V, G::Timestamp, R, DefaultValTrace<K, V, G::Timestamp, R>>>
sourceimpl<G: Scope, K: Data + Hashable, R: Diff> ArrangeBySelf<G, K, R> for Collection<G, K, R>where
G::Timestamp: Lattice + Ord,
impl<G: Scope, K: Data + Hashable, R: Diff> ArrangeBySelf<G, K, R> for Collection<G, K, R>where
G::Timestamp: Lattice + Ord,
sourcefn arrange_by_self(
&self
) -> Arranged<G, K, (), R, TraceAgent<K, (), G::Timestamp, R, DefaultKeyTrace<K, G::Timestamp, R>>>
fn arrange_by_self(
&self
) -> Arranged<G, K, (), R, TraceAgent<K, (), G::Timestamp, R, DefaultKeyTrace<K, G::Timestamp, R>>>
sourceimpl<G: Clone + Scope, D: Clone, R: Clone + Diff> Clone for Collection<G, D, R>where
G::Timestamp: Clone,
impl<G: Clone + Scope, D: Clone, R: Clone + Diff> Clone for Collection<G, D, R>where
G::Timestamp: Clone,
sourcefn clone(&self) -> Collection<G, D, R>
fn clone(&self) -> Collection<G, D, R>
1.0.0 · sourcefn clone_from(&mut self, source: &Self)
fn clone_from(&mut self, source: &Self)
source
. Read moresourceimpl<G: Scope, D, R> Consolidate<D> for Collection<G, D, R>where
D: Data + Hashable,
R: Diff,
G::Timestamp: Lattice + Ord,
impl<G: Scope, D, R> Consolidate<D> for Collection<G, D, R>where
D: Data + Hashable,
R: Diff,
G::Timestamp: Lattice + Ord,
sourcefn consolidate(&self) -> Self
fn consolidate(&self) -> Self
sourceimpl<G: Scope, D, R> ConsolidateStream<D> for Collection<G, D, R>where
D: Data + Hashable,
R: Diff,
G::Timestamp: Lattice + Ord,
impl<G: Scope, D, R> ConsolidateStream<D> for Collection<G, D, R>where
D: Data + Hashable,
R: Diff,
G::Timestamp: Lattice + Ord,
sourcefn consolidate_stream(&self) -> Self
fn consolidate_stream(&self) -> Self
sourceimpl<G: Scope, K: Data + Hashable, R: Diff> Count<G, K, R> for Collection<G, K, R>where
G::Timestamp: Lattice + Ord,
impl<G: Scope, K: Data + Hashable, R: Diff> Count<G, K, R> for Collection<G, K, R>where
G::Timestamp: Lattice + Ord,
sourceimpl<G: Scope, K: Data + Hashable, R: Diff> CountTotal<G, K, R> for Collection<G, K, R>where
G::Timestamp: TotalOrder + Lattice + Ord,
impl<G: Scope, K: Data + Hashable, R: Diff> CountTotal<G, K, R> for Collection<G, K, R>where
G::Timestamp: TotalOrder + Lattice + Ord,
sourcefn count_total(&self) -> Collection<G, (K, R), isize>
fn count_total(&self) -> Collection<G, (K, R), isize>
sourceimpl<G, K, V, R> Group<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
G::Timestamp: Lattice + Ord,
K: Data + Hashable,
V: Data,
R: Diff,
impl<G, K, V, R> Group<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
G::Timestamp: Lattice + Ord,
K: Data + Hashable,
V: Data,
R: Diff,
sourceimpl<G, K, V, R> GroupArranged<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
G::Timestamp: Lattice + Ord,
K: Data + Hashable,
V: Data,
R: Diff,
impl<G, K, V, R> GroupArranged<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
G::Timestamp: Lattice + Ord,
K: Data + Hashable,
V: Data,
R: Diff,
sourcefn group_arranged<L, V2, T2, R2>(
&self,
logic: L
) -> Arranged<G, K, V2, R2, TraceAgent<K, V2, G::Timestamp, R2, T2>>where
V2: Data,
R2: Diff,
T2: Trace<K, V2, G::Timestamp, R2> + 'static,
T2::Batch: Batch<K, V2, G::Timestamp, R2>,
L: Fn(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,
fn group_arranged<L, V2, T2, R2>(
&self,
logic: L
) -> Arranged<G, K, V2, R2, TraceAgent<K, V2, G::Timestamp, R2, T2>>where
V2: Data,
R2: Diff,
T2: Trace<K, V2, G::Timestamp, R2> + 'static,
T2::Batch: Batch<K, V2, G::Timestamp, R2>,
L: Fn(&K, &[(&V, R)], &mut Vec<(V2, R2)>) + 'static,
group
to arranged data, and returns an arrangement of output data. Read moresourceimpl<G, D, R> Identifiers<G, D, R> for Collection<G, D, R>where
G: Scope,
G::Timestamp: Lattice,
D: Data + Hash,
R: Diff,
impl<G, D, R> Identifiers<G, D, R> for Collection<G, D, R>where
G: Scope,
G::Timestamp: Lattice,
D: Data + Hash,
R: Diff,
sourcefn identifiers(&self) -> Collection<G, (D, u64), R>
fn identifiers(&self) -> Collection<G, (D, u64), R>
sourceimpl<G: Scope, D: Ord + Data + Debug, R: Diff> Iterate<G, D, R> for Collection<G, D, R>
impl<G: Scope, D: Ord + Data + Debug, R: Diff> Iterate<G, D, R> for Collection<G, D, R>
sourcefn iterate<F>(&self, logic: F) -> Collection<G, D, R>where
G::Timestamp: Lattice,
for<'a> F: FnOnce(&Collection<Iterative<'a, G, u64>, D, R>) -> Collection<Iterative<'a, G, u64>, D, R>,
fn iterate<F>(&self, logic: F) -> Collection<G, D, R>where
G::Timestamp: Lattice,
for<'a> F: FnOnce(&Collection<Iterative<'a, G, u64>, D, R>) -> Collection<Iterative<'a, G, u64>, D, R>,
logic
to the source collection until convergence. Read moresourceimpl<G, K, V, R> Join<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
K: Data + Hashable,
V: Data,
R: Diff,
G::Timestamp: Lattice + Ord,
impl<G, K, V, R> Join<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
K: Data + Hashable,
V: Data,
R: Diff,
G::Timestamp: Lattice + Ord,
sourcefn join_map<V2: Data, R2: Diff, D: Data, L>(
&self,
other: &Collection<G, (K, V2), R2>,
logic: L
) -> Collection<G, D, <R as Mul<R2>>::Output>where
R: Mul<R2>,
<R as Mul<R2>>::Output: Diff,
L: Fn(&K, &V, &V2) -> D + 'static,
fn join_map<V2: Data, R2: Diff, D: Data, L>(
&self,
other: &Collection<G, (K, V2), R2>,
logic: L
) -> Collection<G, D, <R as Mul<R2>>::Output>where
R: Mul<R2>,
<R as Mul<R2>>::Output: Diff,
L: Fn(&K, &V, &V2) -> D + 'static,
sourcefn semijoin<R2: Diff>(
&self,
other: &Collection<G, K, R2>
) -> Collection<G, (K, V), <R as Mul<R2>>::Output>where
R: Mul<R2>,
<R as Mul<R2>>::Output: Diff,
fn semijoin<R2: Diff>(
&self,
other: &Collection<G, K, R2>
) -> Collection<G, (K, V), <R as Mul<R2>>::Output>where
R: Mul<R2>,
<R as Mul<R2>>::Output: Diff,
(key, val)
and key
based on key
, producing the former with frequencies multiplied. Read moresourcefn antijoin<R2: Diff>(
&self,
other: &Collection<G, K, R2>
) -> Collection<G, (K, V), R>where
R: Mul<R2, Output = R>,
fn antijoin<R2: Diff>(
&self,
other: &Collection<G, K, R2>
) -> Collection<G, (K, V), R>where
R: Mul<R2, Output = R>,
(key, val)
and key
based on key
, discarding values
in the first collection if their key is present in the second. Read moresourceimpl<G, K, V, R> JoinCore<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
K: Data + Hashable,
V: Data,
R: Diff,
G::Timestamp: Lattice + Ord,
impl<G, K, V, R> JoinCore<G, K, V, R> for Collection<G, (K, V), R>where
G: Scope,
K: Data + Hashable,
V: Data,
R: Diff,
G::Timestamp: Lattice + Ord,
sourcefn join_core<V2, T2, R2, I, L>(
&self,
stream2: &Arranged<G, K, V2, R2, T2>,
result: L
) -> Collection<G, I::Item, <R as Mul<R2>>::Output>where
V2: Ord + Clone + Debug + 'static,
T2: TraceReader<K, V2, G::Timestamp, R2> + Clone + 'static,
T2::Batch: BatchReader<K, V2, G::Timestamp, R2> + 'static,
R2: Diff,
R: Mul<R2>,
<R as Mul<R2>>::Output: Diff,
I: IntoIterator,
I::Item: Data,
L: Fn(&K, &V, &V2) -> I + 'static,
fn join_core<V2, T2, R2, I, L>(
&self,
stream2: &Arranged<G, K, V2, R2, T2>,
result: L
) -> Collection<G, I::Item, <R as Mul<R2>>::Output>where
V2: Ord + Clone + Debug + 'static,
T2: TraceReader<K, V2, G::Timestamp, R2> + Clone + 'static,
T2::Batch: BatchReader<K, V2, G::Timestamp, R2> + 'static,
R2: Diff,
R: Mul<R2>,
<R as Mul<R2>>::Output: Diff,
I: IntoIterator,
I::Item: Data,
L: Fn(&K, &V, &V2) -> I + 'static,
sourceimpl<G, K, D> PrefixSum<G, K, D> for Collection<G, ((usize, K), D)>where
G: Scope,
G::Timestamp: Lattice,
K: Data + Hash,
D: Data + Hash,
impl<G, K, D> PrefixSum<G, K, D> for Collection<G, ((usize, K), D)>where
G: Scope,
G::Timestamp: Lattice,
K: Data + Hash,
D: Data + Hash,
sourcefn prefix_sum<F>(&self, zero: D, combine: F) -> Selfwhere
F: Fn(&K, &D, &D) -> D + 'static,
fn prefix_sum<F>(&self, zero: D, combine: F) -> Selfwhere
F: Fn(&K, &D, &D) -> D + 'static,
sourcefn prefix_sum_at<F>(
&self,
locations: Collection<G, (usize, K)>,
zero: D,
combine: F
) -> Selfwhere
F: Fn(&K, &D, &D) -> D + 'static,
fn prefix_sum_at<F>(
&self,
locations: Collection<G, (usize, K)>,
zero: D,
combine: F
) -> Selfwhere
F: Fn(&K, &D, &D) -> D + 'static,
location
.