differential_dataflow/operators/
history.rs

1//! Shared organization and replay of value histories.
2
3use crate::lattice::Lattice;
4
5/// An accumulation of (value, time, diff) updates.
6pub struct EditList<V, T, D> {
7    values: Vec<(V, usize)>,
8    edits: Vec<(T, D)>,
9}
10
11impl<V: Copy, T: Ord + Lattice, D: crate::difference::Semigroup> EditList<V, T, D> {
12    /// Creates an empty list of edits.
13    #[inline]
14    fn new() -> Self {
15        EditList {
16            values: Vec::new(),
17            edits: Vec::new(),
18        }
19    }
20    /// Clears the list of edits.
21    #[inline]
22    pub fn clear(&mut self) {
23        self.values.clear();
24        self.edits.clear();
25    }
26    fn len(&self) -> usize { self.edits.len() }
27    /// Inserts a new edit for an as-yet undetermined value.
28    #[inline]
29    pub fn push(&mut self, time: T, diff: D) {
30        // TODO: Could attempt "insertion-sort" like behavior here, where we collapse if possible.
31        self.edits.push((time, diff));
32    }
33    /// Associates all edits pushed since the previous `seal_value` call with `value`.
34    #[inline]
35    pub fn seal(&mut self, value: V) {
36        let prev = self.values.last().map(|x| x.1).unwrap_or(0);
37        crate::consolidation::consolidate_from(&mut self.edits, prev);
38        if self.edits.len() > prev {
39            self.values.push((value, self.edits.len()));
40        }
41    }
42    fn map<F: FnMut(V, &T, &D)>(&self, mut logic: F) {
43        for index in 0 .. self.values.len() {
44            let lower = if index == 0 { 0 } else { self.values[index-1].1 };
45            let upper = self.values[index].1;
46            for edit in lower .. upper {
47                logic(self.values[index].0, &self.edits[edit].0, &self.edits[edit].1);
48            }
49        }
50    }
51}
52
53/// A loaded, time-ordered replay of one key's `(value, time, diff)` edits, with meet-advanced
54/// buffer collapse — the shared machinery under replay-based tactics.
55/// Its local contract: after
56/// `advance_buffer_by(meet)` with the meet of the un-replayed times, the buffer is consolidated
57/// and replay cost stays linear in edits rather than quadratic.
58pub struct ValueHistory<V, T, D> {
59    edits: EditList<V, T, D>,
60    history: Vec<(T, T, usize, usize)>,     // (time, meet, value_index, edit_offset)
61    buffer: Vec<((V, T), D)>,               // where we accumulate / collapse updates.
62}
63
64impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: crate::difference::Semigroup> ValueHistory<V, T, D> {
65    /// Creates an empty history.
66    pub fn new() -> Self {
67        ValueHistory {
68            edits: EditList::new(),
69            history: Vec::new(),
70            buffer: Vec::new(),
71        }
72    }
73    /// Discards all loaded state (capacity retained).
74    pub fn clear(&mut self) {
75        self.edits.clear();
76        self.history.clear();
77        self.buffer.clear();
78    }
79
80    pub(in crate::operators) fn edits_mut(&mut self) -> &mut EditList<V, T, D> {
81        &mut self.edits
82    }
83
84    pub(in crate::operators) fn edit_len(&self) -> usize {
85        self.edits.len()
86    }
87
88    pub(in crate::operators) fn map_edits<F: FnMut(V, &T, &D)>(&self, logic: F) {
89        self.edits.map(logic)
90    }
91
92    /// Organizes history based on current contents of edits (sort + suffix meets).
93    fn build(&mut self) {
94        self.buffer.clear();
95        self.history.clear();
96        for value_index in 0 .. self.edits.values.len() {
97            let lower = if value_index > 0 { self.edits.values[value_index-1].1 } else { 0 };
98            let upper = self.edits.values[value_index].1;
99            for edit_index in lower .. upper {
100                let time = self.edits.edits[edit_index].0.clone();
101                self.history.push((time.clone(), time, value_index, edit_index));
102            }
103        }
104
105        self.history.sort_by(|x,y| y.cmp(x));
106        self.history.iter_mut().reduce(|prev, cur| { cur.1.meet_assign(&prev.1); cur });
107    }
108
109    /// Organizes history based on current contents of edits, returning a fresh replay.
110    pub(in crate::operators) fn replay<'history>(&'history mut self) -> HistoryReplay<'history, V, T, D> {
111        self.build();
112        HistoryReplay { replay: self }
113    }
114
115    /// Loads `edits` from a plain iterator (grouped by consecutive value — the presentation
116    /// order), advancing each time by `advance_by` if supplied, then organizes. This is the
117    /// iterator ingestion path: the `int_proxy` tactics present `(value_id, time, diff)` runs
118    /// directly and share this machinery instead of re-implementing it.
119    /// Ungrouped input is still correct, only less compact.
120    pub fn load_iter(&mut self, edits: impl Iterator<Item = (V, T, D)>, advance_by: Option<&T>) {
121        self.edits.clear();
122        let mut cur: Option<V> = None;
123        for (v, mut time, diff) in edits {
124            if cur != Some(v) {
125                if let Some(pv) = cur { self.edits.seal(pv); }
126                cur = Some(v);
127            }
128            if let Some(m) = advance_by { time.join_assign(m); }
129            self.edits.push(time, diff);
130        }
131        if let Some(pv) = cur { self.edits.seal(pv); }
132        self.build();
133    }
134}
135
136impl<V: Copy + Ord, T: Ord + Clone + Lattice, D: Clone + crate::difference::Semigroup> ValueHistory<V, T, D> {
137    /// The next (least) un-replayed time.
138    pub fn time(&self) -> Option<&T> { self.history.last().map(|x| &x.0) }
139    /// The meet of all un-replayed times.
140    pub fn meet(&self) -> Option<&T> { self.history.last().map(|x| &x.1) }
141    /// The next un-replayed edit, as `(value, time, diff)`.
142    pub fn edit(&self) -> Option<(V, &T, &D)> {
143        self.history.last().map(|&(ref t, _, v, e)| (self.edits.values[v].0, t, &self.edits.edits[e].1))
144    }
145    /// The buffered (stepped-in, advanced, consolidated) edits.
146    pub fn buffer(&self) -> &[((V, T), D)] { &self.buffer[..] }
147    /// Move the next edit into the buffer.
148    pub fn step(&mut self) {
149        let (time, _, value_index, edit_offset) = self.history.pop().unwrap();
150        self.buffer.push(((self.edits.values[value_index].0, time), self.edits.edits[edit_offset].1.clone()));
151    }
152    /// Step edits while the next time equals `time`; true iff any did.
153    pub fn step_while_time_is(&mut self, time: &T) -> bool {
154        let mut found = false;
155        while self.time() == Some(time) { found = true; self.step(); }
156        found
157    }
158    /// Step edits while the next time is `<= time` in the TOTAL order (a superset of the
159    /// partially-ordered downset; readers filter the buffer by `less_equal` themselves).
160    pub fn step_through(&mut self, time: &T) {
161        while self.time().is_some_and(|t| t <= time) { self.step(); }
162    }
163    /// Advance buffered times by `meet` and consolidate — the collapse that keeps replay linear.
164    pub fn advance_buffer_by(&mut self, meet: &T) {
165        for element in self.buffer.iter_mut() { (element.0).1.join_assign(meet); }
166        crate::consolidation::consolidate(&mut self.buffer);
167    }
168    /// True when every edit has been replayed.
169    pub fn is_done(&self) -> bool { self.history.is_empty() }
170}
171
172pub(in crate::operators) struct HistoryReplay<'history, V, T, D> {
173    replay: &'history mut ValueHistory<V, T, D>,
174}
175
176// A `HistoryReplay` is a thin handle over a `ValueHistory`.
177// The replay machinery lives on `ValueHistory` itself, and these methods forward to it.
178impl<'history, V: Copy + Ord, T: Ord + Clone + Lattice, D: Clone + crate::difference::Semigroup> HistoryReplay<'history, V, T, D> {
179    pub(in crate::operators) fn time(&self) -> Option<&T> { self.replay.time() }
180    pub(in crate::operators) fn meet(&self) -> Option<&T> { self.replay.meet() }
181    pub(in crate::operators) fn edit(&self) -> Option<(V, &T, &D)> { self.replay.edit() }
182    pub(in crate::operators) fn buffer(&self) -> &[((V, T), D)] { self.replay.buffer() }
183    pub(in crate::operators) fn step(&mut self) { self.replay.step() }
184    pub(in crate::operators) fn step_while_time_is(&mut self, time: &T) -> bool { self.replay.step_while_time_is(time) }
185    pub(in crate::operators) fn advance_buffer_by(&mut self, meet: &T) { self.replay.advance_buffer_by(meet) }
186    pub(in crate::operators) fn is_done(&self) -> bool { self.replay.is_done() }
187}
differential_dataflow/operators/history.rs

differential_dataflow/operators/
history.rs
