differential_dataflow/dynamic/
pointstamp.rs

1//! A timestamp type as in Naiad, where a vector of timestamps of different lengths are comparable.
2//!
3//! This type compares using "standard" tuple logic as if each timestamp were extended indefinitely with minimal elements.
4//!
5//! The path summary for this type allows *run-time* rather than *type-driven* iterative scopes.
6//! Each summary represents some journey within and out of some number of scopes, followed by entry
7//! into and iteration within some other number of scopes.
8//!
9//! As a result, summaries describe some number of trailing coordinates to truncate, and some increments
10//! to the resulting vector. Structurally, the increments can only be to one non-truncated coordinate
11//! (as iteration within a scope requires leaving contained scopes), and then to any number of appended
12//! default coordinates (which is effectively just *setting* the coordinate).
13
14use columnar::Columnar;
15use serde::{Deserialize, Serialize};
16use smallvec::SmallVec;
17
18/// A sequence of timestamps, partially ordered by the product order.
19///
20/// Sequences of different lengths are compared as if extended indefinitely by `T::minimum()`.
21/// `N` coordinates are stored inline (one by default); longer sequences spill to the heap.
22/// Sequences are guaranteed to be "minimal", and may not end with `T::minimum()` entries.
23#[derive(Hash, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, Serialize, Columnar)]
24#[columnar(derive(Eq, PartialEq, Ord, PartialOrd))]
25pub struct PointStamp<T, const N: usize = 1> {
26    /// A sequence of timestamps corresponding to timestamps in a sequence of nested scopes.
27    vector: SmallVec<[T; N]>,
28}
29
30impl<T: Timestamp, const N: usize> PartialEq<[T]> for PointStamp<T, N> {
31    fn eq(&self, other: &[T]) -> bool {
32        self.vector.iter()
33            .zip(other.iter().chain(std::iter::repeat(&T::minimum())))
34            .all(|(t1, t2)| t1.eq(t2))
35    }
36}
37
38impl<T: Timestamp, const N: usize> PartialEq<PointStamp<T, N>> for [T] {
39    fn eq(&self, other: &PointStamp<T, N>) -> bool {
40        self.iter()
41            .zip(other.vector.iter().chain(std::iter::repeat(&T::minimum())))
42            .all(|(t1, t2)| t1.eq(t2))
43    }
44}
45
46impl<T: Timestamp, const N: usize> PartialOrder<[T]> for PointStamp<T, N> {
47    fn less_equal(&self, other: &[T]) -> bool {
48        self.vector.iter()
49            .zip(other.iter().chain(std::iter::repeat(&T::minimum())))
50            .all(|(t1, t2)| t1.less_equal(t2))
51    }
52}
53
54impl<T: Timestamp, const N: usize> PartialOrder<PointStamp<T, N>> for [T] {
55    fn less_equal(&self, other: &PointStamp<T, N>) -> bool {
56        self.iter()
57            .zip(other.vector.iter().chain(std::iter::repeat(&T::minimum())))
58            .all(|(t1, t2)| t1.less_equal(t2))
59    }
60}
61
62impl<T: Timestamp, const N: usize> PointStamp<T, N> {
63    /// Create a new sequence.
64    ///
65    /// This method will modify `vector` to ensure it does not end with `T::minimum()`.
66    pub fn new(mut vector: SmallVec<[T; N]>) -> Self {
67        while vector.last() == Some(&T::minimum()) {
68            vector.pop();
69        }
70        PointStamp { vector }
71    }
72    /// Returns the wrapped small vector.
73    ///
74    /// This method is the support way to mutate the contents of `self`, by extracting
75    /// the vector and then re-introducing it with `PointStamp::new` to re-establish
76    /// the invariant that the vector not end with `T::minimum`.
77    pub fn into_inner(self) -> SmallVec<[T; N]> {
78        self.vector
79    }
80}
81
82// Deserialized sequences go through `new`, so they are minimal whatever the input was;
83// `less_equal` relies on it.
84impl<'de, T: Timestamp, const N: usize> Deserialize<'de> for PointStamp<T, N> {
85    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
86        // The serialized form, field for field, before trimming.
87        #[derive(Deserialize)]
88        struct PointStamp<T, const N: usize> { vector: SmallVec<[T; N]> }
89        let PointStamp { vector } = PointStamp::<T, N>::deserialize(deserializer)?;
90        Ok(Self::new(vector))
91    }
92}
93
94impl<T, const N: usize> std::ops::Deref for PointStamp<T, N> {
95    type Target = [T];
96    fn deref(&self) -> &Self::Target {
97        &self.vector
98    }
99}
100
101// Implement timely dataflow's `PartialOrder` trait.
102use timely::order::PartialOrder;
103impl<T: Timestamp, const N: usize> PartialOrder for PointStamp<T, N> {
104    #[inline]
105    fn less_equal(&self, other: &Self) -> bool {
106        // Every present coordinate must be less-equal the corresponding coordinate,
107        // where absent corresponding coordinates are `T::minimum()`. Coordinates
108        // absent from `self.vector` are themselves `T::minimum()` and are less-equal
109        // any corresponding coordinate in `other.vector`. A longer `self` is never
110        // less-equal: sequences are minimal, so its last coordinate is not `T::minimum()`,
111        // and it would have to be less-equal the absent, minimal coordinate of `other`.
112        let (this, that) = (&self.vector[..], &other.vector[..]);
113        this.len() <= that.len() && this.iter().zip(that.iter()).all(|(t1, t2)| t1.less_equal(t2))
114    }
115}
116
117use timely::progress::timestamp::Refines;
118impl<T: Timestamp, const N: usize> Refines<()> for PointStamp<T, N> {
119    fn to_inner(_outer: ()) -> Self {
120        Self { vector: Default::default() }
121    }
122    fn to_outer(self) -> () {
123        ()
124    }
125    fn summarize(_summary: <Self>::Summary) -> () {
126        ()
127    }
128}
129
130// Implement timely dataflow's `PathSummary` trait.
131// This is preparation for the `Timestamp` implementation below.
132use timely::progress::PathSummary;
133
134/// Describes an action on a `PointStamp`: truncation to `length` followed by `actions`.
135#[derive(Hash, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, Serialize, Deserialize)]
136pub struct PointStampSummary<TS> {
137    /// Number of leading coordinates to retain.
138    ///
139    /// A `None` value indicates that all coordinates should be retained.
140    pub retain: Option<usize>,
141    /// Summary actions to apply to all coordinates.
142    ///
143    /// If `actions.len()` is greater than `retain`, a timestamp should be extended by
144    /// `T::minimum()` in order to be subjected to `actions`.
145    pub actions: Vec<TS>,
146}
147
148impl<T: Timestamp, const N: usize> PathSummary<PointStamp<T, N>> for PointStampSummary<T::Summary> {
149    fn results_in(&self, timestamp: &PointStamp<T, N>) -> Option<PointStamp<T, N>> {
150        // Get a slice of timestamp coordinates appropriate for consideration.
151        let timestamps = if let Some(retain) = self.retain {
152            if retain < timestamp.vector.len() {
153                &timestamp.vector[..retain]
154            } else {
155                &timestamp.vector[..]
156            }
157        } else {
158            &timestamp.vector[..]
159        };
160
161        let mut vector = Vec::with_capacity(std::cmp::max(timestamps.len(), self.actions.len()));
162        // Introduce elements where both timestamp and action exist.
163        let min_len = std::cmp::min(timestamps.len(), self.actions.len());
164        for (action, timestamp) in self.actions.iter().zip(timestamps.iter()) {
165            vector.push(action.results_in(timestamp)?);
166        }
167        // Any remaining timestamps should be copied in.
168        for timestamp in timestamps.iter().skip(min_len) {
169            vector.push(timestamp.clone());
170        }
171        // Any remaining actions should be applied to the empty timestamp.
172        for action in self.actions.iter().skip(min_len) {
173            vector.push(action.results_in(&T::minimum())?);
174        }
175
176        Some(PointStamp::new(vector.into()))
177    }
178    fn followed_by(&self, other: &Self) -> Option<Self> {
179        // The output `retain` will be the minimum of the two inputs.
180        let retain = match (self.retain, other.retain) {
181            (Some(x), Some(y)) => Some(std::cmp::min(x, y)),
182            (Some(x), None) => Some(x),
183            (None, Some(y)) => Some(y),
184            (None, None) => None,
185        };
186
187        // The output `actions` will depend on the relative sizes of the input `retain`s.
188        let self_actions = if let Some(retain) = other.retain {
189            if retain < self.actions.len() {
190                &self.actions[..retain]
191            } else {
192                &self.actions[..]
193            }
194        } else {
195            &self.actions[..]
196        };
197
198        let mut actions = Vec::with_capacity(std::cmp::max(self_actions.len(), other.actions.len()));
199        // Introduce actions where both input actions apply.
200        let min_len = std::cmp::min(self_actions.len(), other.actions.len());
201        for (action1, action2) in self_actions.iter().zip(other.actions.iter()) {
202            actions.push(action1.followed_by(action2)?);
203        }
204        // Append any remaining self actions.
205        actions.extend(self_actions.iter().skip(min_len).cloned());
206        // Append any remaining other actions.
207        actions.extend(other.actions.iter().skip(min_len).cloned());
208
209        Some(Self { retain, actions })
210    }
211}
212
213impl<TS: PartialOrder> PartialOrder for PointStampSummary<TS> {
214    fn less_equal(&self, other: &Self) -> bool {
215        // If the `retain`s are not the same, there is some coordinate which
216        // could either be bigger or smaller as the timestamp or the replacement.
217        // In principle, a `T::minimum()` extension could break this rule, and
218        // we could tighten this logic if needed; I think it is fine not to though.
219        self.retain == other.retain
220            && self.actions.len() <= other.actions.len()
221            && self
222                .actions
223                .iter()
224                .zip(other.actions.iter())
225                .all(|(t1, t2)| t1.less_equal(t2))
226    }
227}
228
229// Implement timely dataflow's `Timestamp` trait.
230use timely::progress::Timestamp;
231impl<T: Timestamp, const N: usize> Timestamp for PointStamp<T, N> {
232    fn minimum() -> Self {
233        Self::new(Default::default())
234    }
235    type Summary = PointStampSummary<T::Summary>;
236}
237
238// Implement differential dataflow's `Lattice` trait.
239// This extends the `PartialOrder` implementation with additional structure.
240use crate::lattice::Lattice;
241impl<T: Lattice + Timestamp, const N: usize> Lattice for PointStamp<T, N> {
242    #[inline(always)]
243    fn join(&self, other: &Self) -> Self {
244        // The longer input's coordinates, joined with the shorter input's where it has them.
245        // The result is minimal without trimming: its last coordinate is the longer input's
246        // last, which is not `T::minimum()`, or that coordinate joined with another.
247        let (this, that) = (&self.vector[..], &other.vector[..]);
248        // An empty input (the minimum time) leaves the other unchanged.
249        if this.is_empty() { return other.clone(); }
250        if that.is_empty() { return self.clone(); }
251        let (long, short) = if this.len() >= that.len() { (this, that) } else { (that, this) };
252        let vector = long
253            .iter()
254            .enumerate()
255            .map(|(index, time)| if index < short.len() { time.join(&short[index]) } else { time.clone() })
256            .collect();
257        Self { vector }
258    }
259    #[inline]
260    fn join_assign(&mut self, other: &Self) {
261        let that = &other.vector[..];
262        let my_len = self.vector.len();
263        for (this, that) in self.vector.iter_mut().zip(that.iter()) {
264            this.join_assign(that);
265        }
266        if that.len() > my_len {
267            self.vector.extend(that[my_len..].iter().cloned());
268        }
269    }
270    #[inline(always)]
271    fn meet(&self, other: &Self) -> Self {
272        // Coordinates both inputs have, met; the rest are `T::minimum()` in one input, and so
273        // in the output, which `new` then trims from the result.
274        let vector = self.vector.iter().zip(other.vector.iter()).map(|(this, that)| this.meet(that)).collect();
275        Self::new(vector)
276    }
277    #[inline]
278    fn meet_assign(&mut self, other: &Self) {
279        let that = &other.vector[..];
280        self.vector.truncate(that.len());
281        for (this, that) in self.vector.iter_mut().zip(that.iter()) {
282            this.meet_assign(that);
283        }
284        while self.vector.last() == Some(&T::minimum()) { self.vector.pop(); }
285    }
286}
287
288mod columnation {
289    use columnation::{Columnation, Region};
290    use smallvec::SmallVec;
291    use crate::dynamic::pointstamp::PointStamp;
292
293    impl<T: Columnation+Clone, const N: usize> Columnation for PointStamp<T, N> {
294        type InnerRegion = PointStampStack<T::InnerRegion, N>;
295    }
296
297    /// Stack for PointStamp. Part of Columnation implementation.
298    pub struct PointStampStack<R: Region<Item: Columnation+Clone>, const N: usize = 1>(<SmallVec<[R::Item; N]> as Columnation>::InnerRegion);
299
300    impl<R: Region<Item: Columnation+Clone>, const N: usize> Default for PointStampStack<R, N> {
301        #[inline]
302        fn default() -> Self {
303            Self(Default::default())
304        }
305    }
306
307    impl<R: Region<Item: Columnation+Clone>, const N: usize> Region for PointStampStack<R, N> {
308        type Item = PointStamp<R::Item, N>;
309
310        #[inline]
311        unsafe fn copy(&mut self, item: &Self::Item) -> Self::Item {
312            Self::Item { vector: self.0.copy(&item.vector) }
313        }
314
315        fn clear(&mut self) {
316            self.0.clear();
317        }
318
319        fn reserve_items<'a, I>(&mut self, items: I) where Self: 'a, I: Iterator<Item=&'a Self::Item> + Clone {
320            self.0.reserve_items(items.map(|x| &x.vector));
321        }
322
323        fn reserve_regions<'a, I>(&mut self, regions: I) where Self: 'a, I: Iterator<Item=&'a Self> + Clone {
324            self.0.reserve_regions(regions.map(|r| &r.0));
325        }
326
327        fn heap_size(&self, callback: impl FnMut(usize, usize)) {
328            self.0.heap_size(callback);
329        }
330    }
331}
differential_dataflow/dynamic/pointstamp.rs

differential_dataflow/dynamic/
pointstamp.rs
