differential_dogs3/operators/
count.rs

Help
1use timely::dataflow::Scope;
2
3use differential_dataflow::{ExchangeData, Collection, Hashable};
4use differential_dataflow::difference::{Semigroup, Monoid, Multiply};
5use differential_dataflow::operators::arrange::Arranged;
6use differential_dataflow::trace::TraceReader;
7use differential_dataflow::trace::cursor::IntoOwned;
8
9/// Reports a number of extensions to a stream of prefixes.
10///
11/// This method takes as input a stream of `(prefix, count, index)` triples.
12/// For each triple, it extracts a key using `key_selector`, and finds the
13/// associated count in `arrangement`. If the found count is less than `count`,
14/// the `count` and `index` fields are overwritten with their new values.
15pub fn count<G, Tr, K, R, F, P>(
16    prefixes: &Collection<G, (P, usize, usize), R>,
17    arrangement: Arranged<G, Tr>,
18    key_selector: F,
19    index: usize,
20) -> Collection<G, (P, usize, usize), R>
21where
22    G: Scope<Timestamp=Tr::Time>,
23    Tr: TraceReader<Diff=isize>+Clone+'static,
24    for<'a> Tr::Key<'a>: IntoOwned<'a, Owned = K>,
25    for<'a> Tr::Diff : Semigroup<Tr::DiffGat<'a>>,
26    K: Hashable + Ord + Default + 'static,
27    R: Monoid+Multiply<Output = R>+ExchangeData,
28    F: Fn(&P)->K+Clone+'static,
29    P: ExchangeData,
30{
31    crate::operators::lookup_map(
32        prefixes,
33        arrangement,
34        move |p: &(P,usize,usize), k: &mut K| { *k = key_selector(&p.0); },
35        move |(p,c,i), r, _, s| {
36            let s = *s as usize;
37            if *c < s { ((p.clone(), *c, *i), r.clone()) }
38            else      { ((p.clone(), s, index), r.clone()) }
39        },
40        Default::default(),
41        Default::default(),
42        Default::default(),
43    )
44}
differential_dogs3/operators/count.rs

differential_dogs3/operators/
count.rs
