differential_dataflow/trace/cursor/
mod.rs

1//! Traits and types for navigating order sequences of update tuples.
2//!
3//! The `Cursor` trait contains several methods for efficiently navigating ordered collections
4//! of tuples of the form `(key, val, time, diff)`. The cursor is different from an iterator
5//! both because it allows navigation on multiple levels (key and val), but also because it
6//! supports efficient seeking (via the `seek_key` and `seek_val` methods).
7
8pub mod cursor_list;
9
10pub use self::cursor_list::CursorList;
11
12use timely::progress::Timestamp;
13use crate::lattice::Lattice;
14use crate::difference::Semigroup;
15use crate::trace::implementations::containers::BatchContainer;
16
17/// A batch type that has a cursor for navigation.
18///
19/// This is the entry point for accessing batch data through cursors, and the place that opinions
20/// about keys and values are introduced (via the `Cursor` associated type). Cut-and-merge assembly
21/// is the trace's concern: [`TraceReader::batches_through`](crate::trace::TraceReader::batches_through)
22/// selects the batches and the defaulted
23/// [`TraceReader::cursor_through`](crate::trace::TraceReader::cursor_through) builds a [`CursorList`]
24/// over their per-batch cursors.
25pub trait Navigable {
26
27    /// The cursor type.
28    ///
29    /// The cursor carries the layout opinions (keys, values, containers); a `Navigable` type only
30    /// promises that it can produce one.
31    type Cursor: Cursor<Storage = Self>;
32
33    /// Acquire a cursor suitable for the instance.
34    fn cursor(&self) -> Self::Cursor;
35}
36
37/// The cursor type for a trace's batches.
38pub type BatchCursor<Tr> = <<Tr as crate::trace::TraceReader>::Batch as Navigable>::Cursor;
39
40/// The borrowed key type of a trace's batch cursor.
41pub type BatchKey<'a, Tr> = <BatchCursor<Tr> as Cursor>::Key<'a>;
42/// The borrowed val type of a trace's batch cursor.
43pub type BatchVal<'a, Tr> = <BatchCursor<Tr> as Cursor>::Val<'a>;
44/// The owned val type of a trace's batch cursor.
45pub type BatchValOwn<Tr> = <BatchCursor<Tr> as Cursor>::ValOwn;
46/// The borrowed diff type of a trace's batch cursor.
47pub type BatchDiffGat<'a, Tr> = <BatchCursor<Tr> as Cursor>::DiffGat<'a>;
48/// The owned diff type of a trace's batch cursor.
49pub type BatchDiff<Tr> = <BatchCursor<Tr> as Cursor>::Diff;
50/// The borrowed time type of a trace's batch cursor.
51pub type BatchTimeGat<'a, Tr> = <BatchCursor<Tr> as Cursor>::TimeGat<'a>;
52
53/// Assembles a merged cursor over a sequence of batches.
54///
55/// The batches become the cursor's storage and are returned alongside the cursor; they must be kept
56/// alive and handed to the cursor's navigation methods. This is the shared assembly behind
57/// `TraceReader::cursor_through` and the per-round input cursors in `reduce` / `count` / `threshold`.
58pub fn cursor_list<B: crate::trace::BatchReader + Navigable>(batches: Vec<B>) -> (CursorList<B::Cursor>, Vec<B>) {
59    let cursors = batches.iter().map(|batch| batch.cursor()).collect::<Vec<_>>();
60    let cursor = CursorList::new(cursors, &batches);
61    (cursor, batches)
62}
63
64/// A cursor for navigating ordered `(key, val, time, diff)` updates.
65pub trait Cursor {
66
67    /// Storage required by the cursor.
68    type Storage;
69
70    /// Alias for a borrowed key.
71    type Key<'a>: Copy + Ord;
72    /// Alias for an owned val.
73    type ValOwn: Clone + Ord;
74    /// Alias for a borrowed val.
75    type Val<'a>: Copy + Ord;
76    /// Alias for an owned time.
77    type Time: Lattice + Timestamp;
78    /// Alias for a borrowed time.
79    type TimeGat<'a>: Copy + Ord;
80    /// Alias for an owned diff.
81    type Diff: Semigroup + 'static;
82    /// Alias for a borrowed diff.
83    type DiffGat<'a>: Copy + Ord;
84
85    /// Container for update keys.
86    type KeyContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Key<'a>>;
87    /// Container for update vals.
88    type ValContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Val<'a>, Owned = Self::ValOwn>;
89    /// Container for times.
90    type TimeContainer: for<'a> BatchContainer<ReadItem<'a> = Self::TimeGat<'a>, Owned = Self::Time>;
91    /// Container for diffs.
92    type DiffContainer: for<'a> BatchContainer<ReadItem<'a> = Self::DiffGat<'a>, Owned = Self::Diff>;
93
94    /// Construct an owned val from a reference.
95    #[inline(always)] fn owned_val(val: Self::Val<'_>) -> Self::ValOwn { <Self::ValContainer as BatchContainer>::into_owned(val) }
96    /// Construct an owned time from a reference.
97    #[inline(always)] fn owned_time(time: Self::TimeGat<'_>) -> Self::Time { <Self::TimeContainer as BatchContainer>::into_owned(time) }
98    /// Construct an owned diff from a reference.
99    #[inline(always)] fn owned_diff(diff: Self::DiffGat<'_>) -> Self::Diff { <Self::DiffContainer as BatchContainer>::into_owned(diff) }
100    /// Clones a reference time onto an owned time.
101    #[inline(always)] fn clone_time_onto(time: Self::TimeGat<'_>, onto: &mut Self::Time) { <Self::TimeContainer as BatchContainer>::clone_onto(time, onto) }
102
103    /// Indicates if the current key is valid.
104    ///
105    /// A value of `false` indicates that the cursor has exhausted all keys.
106    fn key_valid(&self, storage: &Self::Storage) -> bool;
107    /// Indicates if the current value is valid.
108    ///
109    /// A value of `false` indicates that the cursor has exhausted all values for this key.
110    fn val_valid(&self, storage: &Self::Storage) -> bool;
111
112    /// A reference to the current key. Asserts if invalid.
113    fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a>;
114    /// A reference to the current value. Asserts if invalid.
115    fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a>;
116
117    /// Returns a reference to the current key, if valid.
118    fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>>;
119    /// Returns a reference to the current value, if valid.
120    fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>>;
121
122    /// Applies `logic` to each pair of time and difference. Intended for mutation of the
123    /// closure's scope.
124    fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L);
125
126    /// Advances the cursor to the next key.
127    fn step_key(&mut self, storage: &Self::Storage);
128    /// Advances the cursor to the specified key.
129    fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>);
130
131    /// Advances the cursor to the next value.
132    fn step_val(&mut self, storage: &Self::Storage);
133    /// Advances the cursor to the specified value.
134    fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>);
135
136    /// Rewinds the cursor to the first key.
137    fn rewind_keys(&mut self, storage: &Self::Storage);
138    /// Rewinds the cursor to the first value for current key.
139    fn rewind_vals(&mut self, storage: &Self::Storage);
140
141    /// Loads `target` with all updates associated with the supplied `key`.
142    ///
143    /// First `target` is cleared, and then if we find `key` we populated it with each of its `(val, time, diff)` updates.
144    /// If `meet` is supplied, the time is joined with each `time` in the updates, to advance the times before consolidation.
145    fn populate_key<'a>(&mut self, storage: &'a Self::Storage, key: Self::Key<'a>, meet: Option<&Self::Time>, target: &mut crate::operators::EditList<Self::Val<'a>, Self::Time, Self::Diff>) {
146        target.clear();
147        self.seek_key(storage, key);
148        if self.get_key(storage) == Some(key) {
149            self.rewind_vals(storage);
150            while let Some(val) = self.get_val(storage) {
151                self.map_times(storage, |time, diff| {
152                    use crate::lattice::Lattice;
153                    let mut time = Self::owned_time(time);
154                    if let Some(meet) = meet { time.join_assign(meet); }
155                    target.push(time, Self::owned_diff(diff))
156                });
157                target.seal(val);
158                self.step_val(storage);
159            }
160        }
161    }
162
163    /// Rewinds the cursor and outputs its contents to a Vec
164    fn to_vec<K, IK, V, IV>(&mut self, storage: &Self::Storage, into_key: IK, into_val: IV) -> Vec<((K, V), Vec<(Self::Time, Self::Diff)>)>
165    where
166        IK: for<'a> Fn(Self::Key<'a>) -> K,
167        IV: for<'a> Fn(Self::Val<'a>) -> V,
168    {
169        let mut out = Vec::new();
170        self.rewind_keys(storage);
171        while let Some(key) = self.get_key(storage) {
172            self.rewind_vals(storage);
173            while let Some(val) = self.get_val(storage) {
174                let mut kv_out = Vec::new();
175                self.map_times(storage, |ts, r| {
176                    kv_out.push((Self::owned_time(ts), Self::owned_diff(r)));
177                });
178                out.push(((into_key(key), into_val(val)), kv_out));
179                self.step_val(storage);
180            }
181            self.step_key(storage);
182        }
183        out
184    }
185}
differential_dataflow/trace/cursor/mod.rs

differential_dataflow/trace/cursor/
mod.rs
