ddshow 0.2.2

Timely and Differential dataflow log analysis and vizualization
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

use differential_dataflow::{
    difference::{Abelian, Semigroup},
    lattice::Lattice,
    operators::iterate::{SemigroupVariable, Variable},
    Collection, Data,
};
use timely::{
    dataflow::{scopes::child::Iterative, Scope},
    order::Product,
};

/// An extension trait for the `iterate` method.
pub trait IterateExt<S, D, R>
where
    S: Scope,
    D: Data,
    R: Semigroup,
{
    fn iterate_named<F>(&self, name: &str, logic: F) -> Collection<S, D, R>
    where
        S::Timestamp: Lattice,
        for<'a> F: FnOnce(
            &Collection<Iterative<'a, S, u64>, D, R>,
        ) -> Collection<Iterative<'a, S, u64>, D, R>;
}

impl<G, D, R> IterateExt<G, D, R> for Collection<G, D, R>
where
    G: Scope,
    D: Ord + Data,
    R: Abelian,
{
    fn iterate_named<F>(&self, name: &str, 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>,
    {
        self.inner.scope().scoped(name, |subgraph| {
            // create a new variable, apply logic, bind variable, return.
            //
            // this could be much more succinct if we returned the collection
            // wrapped by `variable`, but it also results in substantially more
            // diffs produced; `result` is post-consolidation, and means fewer
            // records are yielded out of the loop.
            let variable =
                Variable::new_from(self.enter(subgraph), Product::new(Default::default(), 1));
            let result = logic(&variable);
            variable.set(&result);
            result.leave()
        })
    }
}

impl<G, D, R> IterateExt<G, D, R> for G
where
    G: Scope,
    D: Ord + Data,
    R: Semigroup,
{
    fn iterate_named<F>(&self, name: &str, 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>,
    {
        // TODO: This makes me think we have the wrong ownership pattern here.
        let mut clone = self.clone();
        clone.scoped(name, |subgraph| {
            // create a new variable, apply logic, bind variable, return.
            //
            // this could be much more succinct if we returned the collection
            // wrapped by `variable`, but it also results in substantially more
            // diffs produced; `result` is post-consolidation, and means fewer
            // records are yielded out of the loop.
            let variable = SemigroupVariable::new(subgraph, Product::new(Default::default(), 1));
            let result = logic(&variable);
            variable.set(&result);
            result.leave()
        })
    }
}