differential_dataflow/trace/wrappers/
frontier.rs

1//! Wrapper for frontiered trace.
2//!
3//! Wraps a trace with `since` and `upper` frontiers so that all exposed timestamps are first advanced
4//! by the `since` frontier and restricted by the `upper` frontier. This presents a deterministic trace
5//! on the interval `[since, upper)`, presenting only accumulations up to `since` (rather than partially
6//! accumulated updates) and no updates at times greater or equal to `upper` (even as parts of batches
7//! that span that time).
8
9use timely::progress::{Antichain, frontier::AntichainRef};
10
11use crate::trace::{BatchReader, Description, Navigable, TraceReader};
12use crate::trace::cursor::Cursor;
13use crate::lattice::Lattice;
14
15/// Wrapper to provide trace to nested scope.
16pub struct TraceFrontier<Tr: TraceReader> {
17    trace: Tr,
18    /// Frontier to which all update times will be advanced.
19    since: Antichain<Tr::Time>,
20    /// Frontier after which all update times will be suppressed.
21    until: Antichain<Tr::Time>,
22}
23
24impl<Tr: TraceReader + Clone> Clone for TraceFrontier<Tr> {
25    fn clone(&self) -> Self {
26        TraceFrontier {
27            trace: self.trace.clone(),
28            since: self.since.clone(),
29            until: self.until.clone(),
30        }
31    }
32}
33
34impl<Tr: TraceReader> TraceReader for TraceFrontier<Tr> {
35
36    type Time = Tr::Time;
37    type Batch = BatchFrontier<Tr::Batch>;
38
39    fn map_batches<F: FnMut(&Self::Batch)>(&self, mut f: F) {
40        let since = self.since.borrow();
41        let until = self.until.borrow();
42        self.trace.map_batches(|batch| f(&Self::Batch::make_from(batch.clone(), since, until)))
43    }
44
45    fn set_logical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) { self.trace.set_logical_compaction(frontier) }
46    fn get_logical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { self.trace.get_logical_compaction() }
47
48    fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) { self.trace.set_physical_compaction(frontier) }
49    fn get_physical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { self.trace.get_physical_compaction() }
50
51    fn batches_through(&mut self, upper: AntichainRef<'_, Tr::Time>) -> Option<Vec<Self::Batch>> {
52        let storage = self.trace.batches_through(upper)?;
53        let since = self.since.borrow();
54        let until = self.until.borrow();
55        Some(storage.into_iter().map(|batch| BatchFrontier::make_from(batch, since, until)).collect())
56    }
57}
58
59impl<Tr: TraceReader> TraceFrontier<Tr> {
60    /// Makes a new trace wrapper
61    pub fn make_from(trace: Tr, since: AntichainRef<'_, Tr::Time>, until: AntichainRef<'_, Tr::Time>) -> Self {
62        TraceFrontier {
63            trace,
64            since: since.to_owned(),
65            until: until.to_owned(),
66        }
67    }
68}
69
70
71/// Wrapper to provide batch to nested scope.
72#[derive(Clone)]
73pub struct BatchFrontier<B: BatchReader> {
74    batch: B,
75    since: Antichain<B::Time>,
76    until: Antichain<B::Time>,
77}
78
79impl<B> Navigable for BatchFrontier<B>
80where
81    B: BatchReader + Navigable,
82    B::Cursor: Cursor<Time = B::Time>,
83{
84
85    type Cursor = BatchCursorFrontier<B::Cursor>;
86
87    fn cursor(&self) -> Self::Cursor {
88        BatchCursorFrontier::new(self.batch.cursor(), self.since.borrow(), self.until.borrow())
89    }
90}
91
92impl<B: BatchReader> BatchReader for BatchFrontier<B> {
93    type Time = B::Time;
94    fn len(&self) -> usize { self.batch.len() }
95    fn description(&self) -> &Description<B::Time> { self.batch.description() }
96}
97
98impl<B: BatchReader> BatchFrontier<B> {
99    /// Makes a new batch wrapper
100    pub fn make_from(batch: B, since: AntichainRef<B::Time>, until: AntichainRef<B::Time>) -> Self {
101        BatchFrontier {
102            batch,
103            since: since.to_owned(),
104            until: until.to_owned(),
105        }
106    }
107}
108
109use crate::trace::implementations::BatchContainer;
110
111/// Wrapper to provide cursor to nested scope.
112pub struct BatchCursorFrontier<C: Cursor> {
113    cursor: C,
114    since: Antichain<C::Time>,
115    until: Antichain<C::Time>,
116}
117
118impl<C: Cursor> BatchCursorFrontier<C> {
119    fn new(cursor: C, since: AntichainRef<C::Time>, until: AntichainRef<C::Time>) -> Self {
120        BatchCursorFrontier {
121            cursor,
122            since: since.to_owned(),
123            until: until.to_owned(),
124        }
125    }
126}
127
128impl<C: Cursor<Storage: BatchReader>> Cursor for BatchCursorFrontier<C> {
129
130    type Storage = BatchFrontier<C::Storage>;
131
132    type Key<'a> = C::Key<'a>;
133    type ValOwn = C::ValOwn;
134    type Val<'a> = C::Val<'a>;
135    type KeyContainer = C::KeyContainer;
136    type ValContainer = C::ValContainer;
137    type DiffContainer = C::DiffContainer;
138    type Diff = C::Diff;
139    type DiffGat<'a> = C::DiffGat<'a>;
140    type TimeContainer = Vec<C::Time>;
141    type Time = <Vec<C::Time> as BatchContainer>::Owned;
142    type TimeGat<'a> = <Vec<C::Time> as BatchContainer>::ReadItem<'a>;
143
144    #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(&storage.batch) }
145    #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(&storage.batch) }
146
147    #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(&storage.batch) }
148    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(&storage.batch) }
149
150    #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { self.cursor.get_key(&storage.batch) }
151    #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(&storage.batch) }
152
153    #[inline]
154    fn map_times<L: FnMut(Self::TimeGat<'_>, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, mut logic: L) {
155        let since = self.since.borrow();
156        let until = self.until.borrow();
157        let mut temp: C::Time = <C::Time as timely::progress::Timestamp>::minimum();
158        self.cursor.map_times(&storage.batch, |time, diff| {
159            C::clone_time_onto(time, &mut temp);
160            temp.advance_by(since);
161            if !until.less_equal(&temp) {
162                logic(&temp, diff);
163            }
164        })
165    }
166
167    #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(&storage.batch) }
168    #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(&storage.batch, key) }
169
170    #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(&storage.batch) }
171    #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(&storage.batch, val) }
172
173    #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(&storage.batch) }
174    #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(&storage.batch) }
175}
differential_dataflow/trace/wrappers/frontier.rs

differential_dataflow/trace/wrappers/
frontier.rs
