differential_dataflow/trace/cursor/wrappers/
frontier.rs

1//! Cursor for the `frontier` batch wrapper.
2
3use crate::trace::implementations::BatchContainer;
4use crate::trace::wrappers::frontier::BatchFrontier;
5use crate::trace::Navigable;
6use crate::trace::cursor::Cursor;
7
8impl<B> Navigable for BatchFrontier<B, <B::Cursor as Cursor>::Time>
9where
10    B: Navigable,
11{
12    type Cursor = BatchCursorFrontier<B::Cursor>;
13
14    fn cursor(&self) -> Self::Cursor {
15        BatchCursorFrontier { cursor: self.inner().cursor() }
16    }
17}
18
19/// Wrapper to provide cursor to nested scope.
20///
21/// The wrapper's `since` and `until` frontiers stay with the batch, which applies them through
22/// [`BatchFrontier::advance_time`]; the cursor holds only the cursor it forwards to.
23pub struct BatchCursorFrontier<C> {
24    cursor: C,
25}
26
27impl<C> Cursor for BatchCursorFrontier<C>
28where
29    C: Cursor,
30{
31    type Storage = BatchFrontier<C::Storage, C::Time>;
32
33    type Key<'a> = C::Key<'a>;
34    type ValOwn = C::ValOwn;
35    type Val<'a> = C::Val<'a>;
36    type KeyContainer = C::KeyContainer;
37    type ValContainer = C::ValContainer;
38    type DiffContainer = C::DiffContainer;
39    type Diff = C::Diff;
40    type DiffGat<'a> = C::DiffGat<'a>;
41    type TimeContainer = Vec<C::Time>;
42    type Time = <Vec<C::Time> as BatchContainer>::Owned;
43    type TimeGat<'a> = <Vec<C::Time> as BatchContainer>::ReadItem<'a>;
44
45    #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage.inner()) }
46    #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage.inner()) }
47
48    #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage.inner()) }
49    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage.inner()) }
50
51    #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { self.cursor.get_key(storage.inner()) }
52    #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(storage.inner()) }
53
54    #[inline]
55    fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, mut logic: L) {
56        let mut temp: C::Time = <C::Time as timely::progress::Timestamp>::minimum();
57        self.cursor.map_times(storage.inner(), |time, diff| {
58            C::clone_time_onto(time, &mut temp);
59            if storage.advance_time(&mut temp) {
60                logic(&temp, diff);
61            }
62        })
63    }
64
65    #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage.inner()) }
66    #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage.inner(), key) }
67
68    #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage.inner()) }
69    #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage.inner(), val) }
70
71    #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage.inner()) }
72    #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage.inner()) }
73}
differential_dataflow/trace/cursor/wrappers/frontier.rs

differential_dataflow/trace/cursor/wrappers/
frontier.rs
