fn eval(&mut self, delta: &Z, trace: &T) -> Z {
        //eprintln!("distinct_trace {}", self.time);

        self.empty_input = delta.is_zero();

        // Make sure we have enough room in `future_updates` to
        // accommodate the largest timestamp in the trace, so we don't
        // need to worry about growing `future_updates` later on.
        let mut new_len: u32 = self.time + 1;
        trace.map_batches(|batch| {
            for ts in batch.upper().elements().iter() {
                new_len = max(new_len, ts.inner() + 1);
            }
        });

        self.future_updates
            .resize(new_len as usize, BTreeSet::new());

        let mut batch = Vec::with_capacity(delta.len());

        let mut trace_cursor = trace.cursor();

        // For all keys in delta, for all keys in future_updates[time].
        let mut delta_cursor = delta.cursor();
        let candidates = take(&mut self.future_updates[self.time as usize]);
        let mut cand_iterator = candidates.iter();

        let mut candidate = cand_iterator.next();

        // Iterate over keys that appear in either `future_updates[self.time]` or
        // `delta`.
        while delta_cursor.key_valid(delta) && candidate.is_some() {
            let cand_val = candidate.unwrap();
            let k = delta_cursor.key(delta);
            let w = delta_cursor.weight(delta);
            match k.cmp(cand_val) {
                // Key only appears in `delta`.
                Ordering::Less => {
                    self.eval_value(&mut trace_cursor, trace, k, w, &mut batch);
                    delta_cursor.step_key(delta);
                }
                // Key only appears in `future_updates`.
                Ordering::Greater => {
                    self.eval_value(
                        &mut trace_cursor,
                        trace,
                        cand_val,
                        HasZero::zero(),
                        &mut batch,
                    );
                    candidate = cand_iterator.next();
                }
                // Key appears in both `delta` and `future_updates`.
                Ordering::Equal => {
                    self.eval_value(&mut trace_cursor, trace, k, w, &mut batch);
                    delta_cursor.step_key(delta);
                    candidate = cand_iterator.next();
                }
            }
        }

        // One of the cursors is empty; iterate over whatever remains in the other
        // cursor.

        while delta_cursor.key_valid(delta) {
            let k = delta_cursor.key(delta);
            let w = delta_cursor.weight(delta);

            self.eval_value(&mut trace_cursor, trace, k, w, &mut batch);
            delta_cursor.step_key(delta);
        }
        while candidate.is_some() {
            self.eval_value(
                &mut trace_cursor,
                trace,
                candidate.unwrap(),
                HasZero::zero(),
                &mut batch,
            );
            candidate = cand_iterator.next();
        }

        self.time += 1;

        let result = Z::from_tuples((), batch);
        self.empty_output = result.is_zero();
        result
    }
// Distinct does not have nice properties like linearity that help with
// incremental evaluation: computing an update for each key requires inspecting
// the entire history of this key.  Fortunately, we do not need to look at all
// keys in the trace.  In particular, only those keys that appeared in input
// deltas during the current clock epoch can appear in the output of this
// operator with non-zero weights.  So, as a simple heuristic, we could track
// the set of keys that appeared since the start of the epoch and only consider
// those.
//
// But we can restrict the set of keys to consider even more by using a stronger
// property: only the keys `k` that satisfy one of the following conditions can
// appear in the output of the operator at local time `t`:
// * keys in the current input `delta`
// * keys from earlier inputs observed during the current clock epoch that
//   appeared in one of the previous epochs at time `t`.
//
// To efficiently compute keys that satisfy the second condition, we use the
// `future_updates` vector, where for each key observed in the current epoch
// at time `t1` we lookup the smallest time `t2 > t1` (if any) at which we saw
// the key during any previous epochs and record this key in
// `future_updates[t2]`. Then when evaluating an operatpr at time `t` we
// simply scan all keys in `delta` and all keys in `future_updates[t]` for
// candidates.
//
// TODO: Add example.
//
Distinct does not have nice properties like linearity that help with
incremental evaluation: computing an update for each key requires inspecting
the entire history of this key.  Fortunately, we do not need to look at all
keys in the trace.  In particular, only those keys that appeared in input
deltas during the current clock epoch can appear in the output of this
operator with non-zero weights.  So, as a simple heuristic, we could track
the set of keys that appeared since the start of the epoch and only consider
those.





Add example.
Distinct does not have nice properties like linearity that help with incremental evaluation : computing an update for each key requires inspecting the entire history of this key . Fortunately we do not need to look at all keys in the trace . In particular only those keys that appeared in input deltas during the current clock epoch can appear in the output of this operator with non - zero weights . So as a simple heuristic we could track the set of keys that appeared since the start of the epoch and only consider those . Add example .
fn eval(&mut self, delta: &Z, trace: &T) -> Z {
        //eprintln!("distinct_trace {}", self.time);
        self.empty_input = delta.is_zero();
        // Make sure we have enough room in `future_updates` to
        // accommodate the largest timestamp in the trace, so we don't
        // need to worry about growing `future_updates` later on.
        let mut new_len: u32 = self.time + 1;
        trace.map_batches(|batch| {
            for ts in batch.upper().elements().iter() {
                new_len = max(new_len, ts.inner() + 1);
            }
        });
        self.future_updates
            .resize(new_len as usize, BTreeSet::new());
        let mut batch = Vec::with_capacity(delta.len());
        let mut trace_cursor = trace.cursor();
        // For all keys in delta, for all keys in future_updates[time].
        let mut delta_cursor = delta.cursor();
        let candidates = take(&mut self.future_updates[self.time as usize]);
        let mut cand_iterator = candidates.iter();
        let mut candidate = cand_iterator.next();
        // Iterate over keys that appear in either `future_updates[self.time]` or
        // `delta`.
        while delta_cursor.key_valid(delta) && candidate.is_some() {
            let cand_val = candidate.unwrap();
            let k = delta_cursor.key(delta);
            let w = delta_cursor.weight(delta);
            match k.cmp(cand_val) {
                // Key only appears in `delta`.
                Ordering::Less => {
                    self.eval_value(&mut trace_cursor, trace, k, w, &mut batch);
                    delta_cursor.step_key(delta);
                }
                // Key only appears in `future_updates`.
                Ordering::Greater => {
                    self.eval_value(
                        &mut trace_cursor,
                        trace,
                        cand_val,
                        HasZero::zero(),
                        &mut batch,
                    );
                    candidate = cand_iterator.next();
                }
                // Key appears in both `delta` and `future_updates`.
                Ordering::Equal => {
                    self.eval_value(&mut trace_cursor, trace, k, w, &mut batch);
                    delta_cursor.step_key(delta);
                    candidate = cand_iterator.next();
                }
            }
        }
        // One of the cursors is empty; iterate over whatever remains in the other
        // cursor.
        while delta_cursor.key_valid(delta) {
            let k = delta_cursor.key(delta);
            let w = delta_cursor.weight(delta);
            self.eval_value(&mut trace_cursor, trace, k, w, &mut batch);
            delta_cursor.step_key(delta);
        }
        while candidate.is_some() {
            self.eval_value(
                &mut trace_cursor,
                trace,
                candidate.unwrap(),
                HasZero::zero(),
                &mut batch,
            );
            candidate = cand_iterator.next();
        }
        self.time += 1;
        let result = Z::from_tuples((), batch);
        self.empty_output = result.is_zero();
        result
    }
fn eval ( & mut self , delta : & Z , trace : & T ) -> Z { //eprintln!("distinct_trace {}", self.time); self . empty_input = delta . is_zero ( ) ; // Make sure we have enough room in `future_updates` to // accommodate the largest timestamp in the trace, so we don't // need to worry about growing `future_updates` later on. let mut new_len : u32 = self . time + 1 ; trace . map_batches ( | batch | { for ts in batch . upper ( ) . elements ( ) . iter ( ) { new_len = max ( new_len , ts . inner ( ) + 1 ) ; } } ) ; self . future_updates . resize ( new_len as usize , BTreeSet :: new ( ) ) ; let mut batch = Vec :: with_capacity ( delta . len ( ) ) ; let mut trace_cursor = trace . cursor ( ) ; // For all keys in delta, for all keys in future_updates[time]. let mut delta_cursor = delta . cursor ( ) ; let candidates = take ( & mut self . future_updates [ self . time as usize ] ) ; let mut cand_iterator = candidates . iter ( ) ; let mut candidate = cand_iterator . next ( ) ; // Iterate over keys that appear in either `future_updates[self.time]` or // `delta`. while delta_cursor . key_valid ( delta ) && candidate . is_some ( ) { let cand_val = candidate . unwrap ( ) ; let k = delta_cursor . key ( delta ) ; let w = delta_cursor . weight ( delta ) ; match k . cmp ( cand_val ) { // Key only appears in `delta`. Ordering :: Less => { self . eval_value ( & mut trace_cursor , trace , k , w , & mut batch ) ; delta_cursor . step_key ( delta ) ; } // Key only appears in `future_updates`. Ordering :: Greater => { self . eval_value ( & mut trace_cursor , trace , cand_val , HasZero :: zero ( ) , & mut batch , ) ; candidate = cand_iterator . next ( ) ; } // Key appears in both `delta` and `future_updates`. Ordering :: Equal => { self . eval_value ( & mut trace_cursor , trace , k , w , & mut batch ) ; delta_cursor . step_key ( delta ) ; candidate = cand_iterator . next ( ) ; } } } // One of the cursors is empty; iterate over whatever remains in the other // cursor. while delta_cursor . key_valid ( delta ) { let k = delta_cursor . key ( delta ) ; let w = delta_cursor . weight ( delta ) ; self . eval_value ( & mut trace_cursor , trace , k , w , & mut batch ) ; delta_cursor . step_key ( delta ) ; } while candidate . is_some ( ) { self . eval_value ( & mut trace_cursor , trace , candidate . unwrap ( ) , HasZero :: zero ( ) , & mut batch , ) ; candidate = cand_iterator . next ( ) ; } self . time += 1 ; let result = Z :: from_tuples ( ( ) , batch ) ; self . empty_output = result . is_zero ( ) ; result }
Distinct does not have nice properties like linearity that help with
incremental evaluation: computing an update for each key requires inspecting
the entire history of this key.
Distinct does not have nice properties like linearity that help with incremental evaluation : computing an update for each key requires inspecting the entire history of this key .
&mut self
delta
&Z
trace
&T