differential_dataflow/operators/
threshold.rs

1//! Reduce the collection to one occurrence of each distinct element.
2//!
3//! The `distinct_total` and `distinct_total_u` operators are optimizations of the more general
4//! `distinct` and `distinct_u` operators for the case in which time is totally ordered.
5
6use timely::order::TotalOrder;
7use timely::progress::Timestamp;
8use timely::dataflow::operators::Operator;
9use timely::dataflow::channels::pact::Pipeline;
10
11use crate::lattice::Lattice;
12use crate::{ExchangeData, VecCollection};
13use crate::difference::{Semigroup, Abelian};
14use crate::hashable::Hashable;
15use crate::collection::AsCollection;
16use crate::operators::arrange::Arranged;
17use crate::trace::{BatchCursor, BatchDiff, BatchDiffGat, BatchKey, Cursor, Navigable, TraceReader};
18
19/// Extension trait for the `distinct` differential dataflow method.
20pub trait ThresholdTotal<'scope, T: Timestamp + TotalOrder + Lattice, K: ExchangeData, R: ExchangeData+Semigroup> : Sized {
21    /// Reduces the collection to one occurrence of each distinct element.
22    fn threshold_semigroup<R2, F>(self, thresh: F) -> VecCollection<'scope, T, K, R2>
23    where
24        R2: Semigroup+'static,
25        F: FnMut(&K,&R,Option<&R>)->Option<R2>+'static,
26        ;
27    /// Reduces the collection to one occurrence of each distinct element.
28    ///
29    /// As with `threshold`, a zero count thresholds to zero, whatever `thresh` returns for it.
30    ///
31    /// # Examples
32    ///
33    /// ```
34    /// use differential_dataflow::input::Input;
35    /// use differential_dataflow::operators::ThresholdTotal;
36    ///
37    /// ::timely::example(|scope| {
38    ///     // report the number of occurrences of each key
39    ///     scope.new_collection_from(1 .. 10).1
40    ///          .map(|x| x / 3)
41    ///          .threshold_total(|_,c| c % 2);
42    /// });
43    /// ```
44    fn threshold_total<R2: Abelian+'static, F: FnMut(&K,&R)->R2+'static>(self, mut thresh: F) -> VecCollection<'scope, T, K, R2> {
45        self.threshold_semigroup(move |key, new, old| {
46            // A key whose count is zero may or may not still have updates in the trace, so
47            // treat a zero count as absent rather than call `thresh` on it.
48            let mut new = if new.is_zero() { R2::zero() } else { thresh(key, new) };
49            if let Some(old) = old.filter(|old| !old.is_zero()) {
50                let mut add = thresh(key, old);
51                add.negate();
52                new.plus_equals(&add);
53            }
54            if !new.is_zero() { Some(new) } else { None }
55        })
56    }
57    /// Reduces the collection to one occurrence of each distinct element.
58    ///
59    /// This reduction only tests whether the weight associated with a record is non-zero, and otherwise
60    /// ignores its specific value. To take more general actions based on the accumulated weight, consider
61    /// the `threshold` method.
62    ///
63    /// # Examples
64    ///
65    /// ```
66    /// use differential_dataflow::input::Input;
67    /// use differential_dataflow::operators::ThresholdTotal;
68    ///
69    /// ::timely::example(|scope| {
70    ///     // report the number of occurrences of each key
71    ///     scope.new_collection_from(1 .. 10).1
72    ///          .map(|x| x / 3)
73    ///          .distinct_total();
74    /// });
75    /// ```
76    fn distinct_total(self) -> VecCollection<'scope, T, K, isize> {
77        self.distinct_total_core()
78    }
79
80    /// Distinct for general integer differences.
81    ///
82    /// This method allows `distinct` to produce collections whose difference
83    /// type is something other than an `isize` integer, for example perhaps an
84    /// `i32`.
85    fn distinct_total_core<R2: Abelian+From<i8>+'static>(self) -> VecCollection<'scope, T, K, R2> {
86        self.threshold_total(|_,_| R2::from(1i8))
87    }
88
89}
90
91impl<'scope, T, K: ExchangeData+Hashable, R: ExchangeData+Semigroup> ThresholdTotal<'scope, T, K, R> for VecCollection<'scope, T, K, R>
92where
93    T: Timestamp + TotalOrder + Lattice,
94{
95    fn threshold_semigroup<R2, F>(self, thresh: F) -> VecCollection<'scope, T, K, R2>
96    where
97        R2: Semigroup+'static,
98        F: FnMut(&K,&R,Option<&R>)->Option<R2>+'static,
99    {
100        self.arrange_by_self_named("Arrange: ThresholdTotal")
101            .threshold_semigroup(thresh)
102    }
103}
104
105impl<'scope, K, Tr> ThresholdTotal<'scope, Tr::Time, K, BatchDiff<Tr>> for Arranged<'scope, Tr>
106where
107    Tr: TraceReader<Batch: Navigable, Time: TotalOrder> + Clone + 'static,
108    for<'a> BatchCursor<Tr>: Cursor<
109        Key<'a>=&'a K,
110        Val<'a>=&'a (),
111        Time = Tr::Time,
112        Diff : ExchangeData + Semigroup<BatchDiffGat<'a, Tr>>,
113    >,
114    K: ExchangeData,
115{
116    fn threshold_semigroup<R2, F>(self, mut thresh: F) -> VecCollection<'scope, Tr::Time, K, R2>
117    where
118        R2: Semigroup+'static,
119        F: for<'a> FnMut(BatchKey<'a, Tr>,&BatchDiff<Tr>,Option<&BatchDiff<Tr>>)->Option<R2>+'static,
120    {
121
122        let mut trace = self.trace.clone();
123
124        self.stream.unary_frontier(Pipeline, "ThresholdTotal", move |_,_| {
125
126            // tracks the lower and upper limit of received batches.
127            let mut lower_limit = timely::progress::frontier::Antichain::from_elem(Tr::Time::minimum());
128            let mut upper_limit = timely::progress::frontier::Antichain::from_elem(Tr::Time::minimum());
129
130            move |(input, _frontier), output| {
131
132                let mut batch_storage = Vec::new();
133
134                // Downgrde previous upper limit to be current lower limit.
135                lower_limit.clear();
136                lower_limit.extend(upper_limit.borrow().iter().cloned());
137
138                let mut caps = timely::dataflow::operators::CapabilitySet::new();
139                input.for_each(|capability, batches| {
140                    for capability in capability.retain_stamp(0).iter() {
141                        caps.insert(capability.clone());
142                    }
143                    for batch in batches.drain(..) {
144                        upper_limit.clone_from(batch.upper());  // NB: Assumes batches are in-order
145                        batch_storage.push(batch);
146                    }
147                });
148
149                if !caps.is_empty() {
150
151                    let mut session = output.session(&caps);
152
153                    let (mut batch_cursor, batch_storage) = crate::trace::cursor::cursor_list(batch_storage.into_iter().filter_map(|b| b.inner).collect());
154                    let (mut trace_cursor, trace_storage) = crate::trace::cursor::cursor_list(trace.batches_through(lower_limit.borrow()).unwrap());
155
156                    while let Some(key) = batch_cursor.get_key(&batch_storage) {
157                        let mut count: Option<BatchDiff<Tr>> = None;
158
159                        // Compute the multiplicity of this key before the current batch.
160                        trace_cursor.seek_key(&trace_storage, key);
161                        if trace_cursor.get_key(&trace_storage) == Some(key) {
162                            trace_cursor.map_times(&trace_storage, |_, diff| {
163                                count.as_mut().map(|c| c.plus_equals(&diff));
164                                if count.is_none() { count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff)); }
165                            });
166                        }
167
168                        // Apply `thresh` both before and after `diff` is applied to `count`.
169                        // If the result is non-zero, send it along.
170                        batch_cursor.map_times(&batch_storage, |time, diff| {
171
172                            let difference =
173                            match &count {
174                                Some(old) => {
175                                    let mut temp = old.clone();
176                                    temp.plus_equals(&diff);
177                                    thresh(key, &temp, Some(old))
178                                },
179                                None => { thresh(key, &<BatchCursor<Tr> as Cursor>::owned_diff(diff), None) },
180                            };
181
182                            // Either add or assign `diff` to `count`.
183                            if let Some(count) = &mut count {
184                                count.plus_equals(&diff);
185                            }
186                            else {
187                                count = Some(<BatchCursor<Tr> as Cursor>::owned_diff(diff));
188                            }
189
190                            if let Some(difference) = difference {
191                                if !difference.is_zero() {
192                                    session.give((key.clone(), <BatchCursor<Tr> as Cursor>::owned_time(time), difference));
193                                }
194                            }
195                        });
196
197                        batch_cursor.step_key(&batch_storage);
198                    }
199                }
200
201                // tidy up the shared input trace.
202                trace.advance_upper(&mut upper_limit);
203                trace.set_logical_compaction(upper_limit.borrow());
204                trace.set_physical_compaction(upper_limit.borrow());
205            }
206        })
207        .as_collection()
208    }
209}
differential_dataflow/operators/threshold.rs

differential_dataflow/operators/
threshold.rs
