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;
9pub mod wrappers;
10
11pub use self::cursor_list::CursorList;
12
13use timely::progress::Timestamp;
14use crate::lattice::Lattice;
15use crate::difference::Semigroup;
16use crate::trace::implementations::containers::BatchContainer;
17
18/// A batch type that has a cursor for navigation.
19///
20/// This is the entry point for accessing batch data through cursors, and the place that opinions
21/// about keys and values are introduced (via the `Cursor` associated type). Cut-and-merge assembly
22/// is the trace's concern: [`TraceReader::spans_through`](crate::trace::TraceReader::spans_through)
23/// selects the batches, and [`cursor_list()`] merges their cursors.
24pub trait Navigable {
25
26    /// The cursor type.
27    ///
28    /// The cursor carries the layout opinions (keys, values, containers); a `Navigable` type only
29    /// promises that it can produce one.
30    type Cursor: Cursor<Storage = Self>;
31
32    /// Acquire a cursor suitable for the instance.
33    fn cursor(&self) -> Self::Cursor;
34}
35
36/// The cursor type for the updates in a trace's batches.
37pub type BatchCursor<Tr> = <<Tr as crate::trace::TraceReader>::Batch as Navigable>::Cursor;
38
39/// The borrowed key type of a trace's batch cursor.
40pub type BatchKey<'a, Tr> = <BatchCursor<Tr> as Cursor>::Key<'a>;
41/// The borrowed val type of a trace's batch cursor.
42pub type BatchVal<'a, Tr> = <BatchCursor<Tr> as Cursor>::Val<'a>;
43/// The owned val type of a trace's batch cursor.
44pub type BatchValOwn<Tr> = <BatchCursor<Tr> as Cursor>::ValOwn;
45/// The borrowed diff type of a trace's batch cursor.
46pub type BatchDiffGat<'a, Tr> = <BatchCursor<Tr> as Cursor>::DiffGat<'a>;
47/// The owned diff type of a trace's batch cursor.
48pub type BatchDiff<Tr> = <BatchCursor<Tr> as Cursor>::Diff;
49/// The borrowed time type of a trace's batch cursor.
50pub type BatchTimeGat<'a, Tr> = <BatchCursor<Tr> as Cursor>::TimeGat<'a>;
51
52/// Assembles a merged cursor over several batches' updates.
53///
54/// The updates become the cursor's storage and are returned alongside the cursor; they must be
55/// kept alive and handed to the cursor's navigation methods.
56pub fn cursor_list<B: Navigable>(updates: Vec<B>) -> (CursorList<B::Cursor>, Vec<B>) {
57    let cursors = updates.iter().map(|u| u.cursor()).collect::<Vec<_>>();
58    let cursor = CursorList::new(cursors, &updates);
59    (cursor, updates)
60}
61
62/// A cursor for navigating ordered `(key, val, time, diff)` updates.
63pub trait Cursor {
64
65    /// Storage required by the cursor.
66    type Storage;
67
68    /// Alias for a borrowed key.
69    type Key<'a>: Copy + Ord;
70    /// Alias for an owned val.
71    type ValOwn: Clone + Ord;
72    /// Alias for a borrowed val.
73    type Val<'a>: Copy + Ord;
74    /// Alias for an owned time.
75    type Time: Lattice + Timestamp;
76    /// Alias for a borrowed time.
77    type TimeGat<'a>: Copy + Ord;
78    /// Alias for an owned diff.
79    type Diff: Semigroup + 'static;
80    /// Alias for a borrowed diff.
81    type DiffGat<'a>: Copy + Ord;
82
83    /// Container for update keys.
84    type KeyContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Key<'a>>;
85    /// Container for update vals.
86    type ValContainer: for<'a> BatchContainer<ReadItem<'a> = Self::Val<'a>, Owned = Self::ValOwn>;
87    /// Container for times.
88    type TimeContainer: for<'a> BatchContainer<ReadItem<'a> = Self::TimeGat<'a>, Owned = Self::Time>;
89    /// Container for diffs.
90    type DiffContainer: for<'a> BatchContainer<ReadItem<'a> = Self::DiffGat<'a>, Owned = Self::Diff>;
91
92    /// Construct an owned val from a reference.
93    #[inline(always)] fn owned_val(val: Self::Val<'_>) -> Self::ValOwn { <Self::ValContainer as BatchContainer>::into_owned(val) }
94    /// Construct an owned time from a reference.
95    #[inline(always)] fn owned_time(time: Self::TimeGat<'_>) -> Self::Time { <Self::TimeContainer as BatchContainer>::into_owned(time) }
96    /// Construct an owned diff from a reference.
97    #[inline(always)] fn owned_diff(diff: Self::DiffGat<'_>) -> Self::Diff { <Self::DiffContainer as BatchContainer>::into_owned(diff) }
98    /// Clones a reference time onto an owned time.
99    #[inline(always)] fn clone_time_onto(time: Self::TimeGat<'_>, onto: &mut Self::Time) { <Self::TimeContainer as BatchContainer>::clone_onto(time, onto) }
100
101    /// Indicates if the current key is valid.
102    ///
103    /// A value of `false` indicates that the cursor has exhausted all keys.
104    fn key_valid(&self, storage: &Self::Storage) -> bool;
105    /// Indicates if the current value is valid.
106    ///
107    /// A value of `false` indicates that the cursor has exhausted all values for this key.
108    fn val_valid(&self, storage: &Self::Storage) -> bool;
109
110    /// A reference to the current key. Asserts if invalid.
111    fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a>;
112    /// A reference to the current value. Asserts if invalid.
113    fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a>;
114
115    /// Returns a reference to the current key, if valid.
116    fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>>;
117    /// Returns a reference to the current value, if valid.
118    fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>>;
119
120    /// Applies `logic` to each pair of time and difference. Intended for mutation of the
121    /// closure's scope.
122    fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L);
123
124    /// Advances the cursor to the next key.
125    fn step_key(&mut self, storage: &Self::Storage);
126    /// Advances the cursor to the specified key.
127    fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>);
128
129    /// Advances the cursor to the next value.
130    fn step_val(&mut self, storage: &Self::Storage);
131    /// Advances the cursor to the specified value.
132    fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>);
133
134    /// Rewinds the cursor to the first key.
135    fn rewind_keys(&mut self, storage: &Self::Storage);
136    /// Rewinds the cursor to the first value for current key.
137    fn rewind_vals(&mut self, storage: &Self::Storage);
138
139    /// Rewinds the cursor and outputs its contents to a Vec
140    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)>)>
141    where
142        IK: for<'a> Fn(Self::Key<'a>) -> K,
143        IV: for<'a> Fn(Self::Val<'a>) -> V,
144    {
145        let mut out = Vec::new();
146        self.rewind_keys(storage);
147        while let Some(key) = self.get_key(storage) {
148            self.rewind_vals(storage);
149            while let Some(val) = self.get_val(storage) {
150                let mut kv_out = Vec::new();
151                self.map_times(storage, |ts, r| {
152                    kv_out.push((Self::owned_time(ts), Self::owned_diff(r)));
153                });
154                out.push(((into_key(key), into_val(val)), kv_out));
155                self.step_val(storage);
156            }
157            self.step_key(storage);
158        }
159        out
160    }
161}
162
163
164/// Blanket implementations for reference counted batches.
165pub mod rc_blanket_impls {
166
167    use std::rc::Rc;
168
169    use super::{Navigable, Cursor};
170
171    impl<B: Navigable> Navigable for Rc<B> {
172        /// The type used to enumerate the batch's contents.
173        type Cursor = RcBatchCursor<B::Cursor>;
174        /// Acquires a cursor to the batch's contents.
175        fn cursor(&self) -> Self::Cursor {
176            RcBatchCursor::new((**self).cursor())
177        }
178    }
179
180    /// Wrapper to provide cursor to nested scope.
181    pub struct RcBatchCursor<C> {
182        cursor: C,
183    }
184
185    impl<C> RcBatchCursor<C> {
186        fn new(cursor: C) -> Self {
187            RcBatchCursor {
188                cursor,
189            }
190        }
191    }
192
193    impl<C: Cursor> Cursor for RcBatchCursor<C> {
194
195        type Storage = Rc<C::Storage>;
196
197        type Key<'a> = C::Key<'a>;
198        type ValOwn = C::ValOwn;
199        type Val<'a> = C::Val<'a>;
200        type Time = C::Time;
201        type TimeGat<'a> = C::TimeGat<'a>;
202        type Diff = C::Diff;
203        type DiffGat<'a> = C::DiffGat<'a>;
204        type KeyContainer = C::KeyContainer;
205        type ValContainer = C::ValContainer;
206        type TimeContainer = C::TimeContainer;
207        type DiffContainer = C::DiffContainer;
208
209        #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) }
210        #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) }
211
212        #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) }
213        #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) }
214
215        #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { self.cursor.get_key(storage) }
216        #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(storage) }
217
218        #[inline]
219        fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, logic: L) {
220            self.cursor.map_times(storage, logic)
221        }
222
223        #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) }
224        #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) }
225
226        #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) }
227        #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) }
228
229        #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) }
230        #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) }
231    }
232
233}
differential_dataflow/trace/cursor/mod.rs

differential_dataflow/trace/cursor/
mod.rs
