differential_dataflow/trace/wrappers/
enter_at.rs

1//! Wrappers to provide trace access to nested scopes.
2
3use timely::progress::timestamp::Refines;
4use timely::progress::{Antichain, frontier::AntichainRef};
5
6use crate::lattice::Lattice;
7use crate::trace::{BatchKey, BatchReader, BatchTimeGat, BatchVal, Description, Navigable, TraceReader};
8use crate::trace::cursor::Cursor;
9
10/// Wrapper to provide trace to nested scope.
11///
12/// Each wrapped update is presented with a timestamp determined by `logic`.
13///
14/// At the same time, we require a method `prior` that can "invert" timestamps,
15/// and which will be applied to compaction frontiers as they are communicated
16/// back to the wrapped traces. A better explanation is pending, and until that
17/// happens use this construct at your own peril!
18pub struct TraceEnter<Tr: TraceReader, TInner, F, G> {
19    trace: Tr,
20    stash1: Antichain<Tr::Time>,
21    stash2: Antichain<TInner>,
22    logic: F,
23    prior: G,
24}
25
26impl<Tr,TInner,F,G> Clone for TraceEnter<Tr, TInner, F, G>
27where
28    Tr: TraceReader+Clone,
29    F: Clone,
30    G: Clone,
31{
32    fn clone(&self) -> Self {
33        TraceEnter {
34            trace: self.trace.clone(),
35            stash1: Antichain::new(),
36            stash2: Antichain::new(),
37            logic: self.logic.clone(),
38            prior: self.prior.clone(),
39        }
40    }
41}
42
43impl<Tr, TInner, F, G> TraceReader for TraceEnter<Tr, TInner, F, G>
44where
45    Tr: TraceReader<Batch: Navigable>,
46    TInner: Refines<Tr::Time>+Lattice,
47    F: 'static,
48    F: FnMut(BatchKey<'_, Tr>, BatchVal<'_, Tr>, BatchTimeGat<'_, Tr>)->TInner+Clone,
49    G: FnMut(&TInner)->Tr::Time+Clone+'static,
50{
51    type Time = TInner;
52    type Batch = BatchEnter<Tr::Batch, TInner,F>;
53
54    fn map_batches<F2: FnMut(&Self::Batch)>(&self, mut f: F2) {
55        let logic = self.logic.clone();
56        self.trace.map_batches(|batch| {
57            f(&Self::Batch::make_from(batch.clone(), logic.clone()));
58        })
59    }
60
61    fn set_logical_compaction(&mut self, frontier: AntichainRef<'_, TInner>) {
62        self.stash1.clear();
63        for time in frontier.iter() {
64            self.stash1.insert((self.prior)(time));
65        }
66        self.trace.set_logical_compaction(self.stash1.borrow());
67    }
68    fn get_logical_compaction(&mut self) -> AntichainRef<'_, TInner> {
69        self.stash2.clear();
70        for time in self.trace.get_logical_compaction().iter() {
71            self.stash2.insert(TInner::to_inner(time.clone()));
72        }
73        self.stash2.borrow()
74    }
75
76    fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, TInner>) {
77        self.stash1.clear();
78        for time in frontier.iter() {
79            self.stash1.insert((self.prior)(time));
80        }
81        self.trace.set_physical_compaction(self.stash1.borrow());
82    }
83    fn get_physical_compaction(&mut self) -> AntichainRef<'_, TInner> {
84        self.stash2.clear();
85        for time in self.trace.get_physical_compaction().iter() {
86            self.stash2.insert(TInner::to_inner(time.clone()));
87        }
88        self.stash2.borrow()
89    }
90
91    fn batches_through(&mut self, upper: AntichainRef<TInner>) -> Option<Vec<Self::Batch>> {
92        self.stash1.clear();
93        for time in upper.iter() {
94            self.stash1.insert(time.clone().to_outer());
95        }
96        let logic = self.logic.clone();
97        let storage = self.trace.batches_through(self.stash1.borrow())?;
98        Some(storage.into_iter().map(|batch| BatchEnter::make_from(batch, logic.clone())).collect())
99    }
100}
101
102impl<Tr, TInner, F, G> TraceEnter<Tr, TInner, F, G>
103where
104    Tr: TraceReader,
105    TInner: Refines<Tr::Time>+Lattice,
106{
107    /// Makes a new trace wrapper
108    pub fn make_from(trace: Tr, logic: F, prior: G) -> Self {
109        TraceEnter {
110            trace,
111            stash1: Antichain::new(),
112            stash2: Antichain::new(),
113            logic,
114            prior,
115        }
116    }
117}
118
119
120/// Wrapper to provide batch to nested scope.
121#[derive(Clone)]
122pub struct BatchEnter<B, TInner, F> {
123    batch: B,
124    description: Description<TInner>,
125    logic: F,
126}
127
128impl<B, TInner, F> Navigable for BatchEnter<B, TInner, F>
129where
130    B: BatchReader + Navigable,
131    TInner: Refines<B::Time>+Lattice,
132    TInner: Refines<<B::Cursor as Cursor>::Time>,
133    F: FnMut(<B::Cursor as Cursor>::Key<'_>, <B::Cursor as Cursor>::Val<'_>, <B::Cursor as Cursor>::TimeGat<'_>)->TInner+Clone,
134{
135    type Cursor = BatchCursorEnter<B::Cursor, TInner, F>;
136
137    fn cursor(&self) -> Self::Cursor {
138        BatchCursorEnter::new(self.batch.cursor(), self.logic.clone())
139    }
140}
141
142impl<B, TInner, F> BatchReader for BatchEnter<B, TInner, F>
143where
144    B: BatchReader + Navigable,
145    TInner: Refines<B::Time>+Lattice,
146    F: FnMut(<B::Cursor as Cursor>::Key<'_>, <B::Cursor as Cursor>::Val<'_>, <B::Cursor as Cursor>::TimeGat<'_>)->TInner+Clone,
147{
148    type Time = TInner;
149    fn len(&self) -> usize { self.batch.len() }
150    fn description(&self) -> &Description<TInner> { &self.description }
151}
152
153impl<B, TInner, F> BatchEnter<B, TInner, F>
154where
155    B: BatchReader,
156    TInner: Refines<B::Time>+Lattice,
157{
158    /// Makes a new batch wrapper
159    pub fn make_from(batch: B, logic: F) -> Self {
160        let lower: Vec<_> = batch.description().lower().elements().iter().map(|x| TInner::to_inner(x.clone())).collect();
161        let upper: Vec<_> = batch.description().upper().elements().iter().map(|x| TInner::to_inner(x.clone())).collect();
162        let since: Vec<_> = batch.description().since().elements().iter().map(|x| TInner::to_inner(x.clone())).collect();
163
164        BatchEnter {
165            batch,
166            description: Description::new(Antichain::from(lower), Antichain::from(upper), Antichain::from(since)),
167            logic,
168        }
169    }
170}
171
172use crate::trace::implementations::BatchContainer;
173
174/// Wrapper to provide cursor to nested scope.
175pub struct BatchCursorEnter<C, TInner, F> {
176    phantom: ::std::marker::PhantomData<TInner>,
177    cursor: C,
178    logic: F,
179}
180
181impl<C, TInner, F> BatchCursorEnter<C, TInner, F> {
182    fn new(cursor: C, logic: F) -> Self {
183        BatchCursorEnter {
184            phantom: ::std::marker::PhantomData,
185            cursor,
186            logic,
187        }
188    }
189}
190
191impl<TInner, C: Cursor, F> Cursor for BatchCursorEnter<C, TInner, F>
192where
193    TInner: Refines<C::Time>+Lattice,
194    F: FnMut(C::Key<'_>, C::Val<'_>, C::TimeGat<'_>)->TInner,
195{
196    type Storage = BatchEnter<C::Storage, TInner, F>;
197
198    type Key<'a> = C::Key<'a>;
199    type ValOwn = C::ValOwn;
200    type Val<'a> = C::Val<'a>;
201    type KeyContainer = C::KeyContainer;
202    type ValContainer = C::ValContainer;
203    type DiffContainer = C::DiffContainer;
204    type Diff = C::Diff;
205    type DiffGat<'a> = C::DiffGat<'a>;
206    type TimeContainer = Vec<TInner>;
207    type Time = <Vec<TInner> as BatchContainer>::Owned;
208    type TimeGat<'a> = <Vec<TInner> as BatchContainer>::ReadItem<'a>;
209
210    #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(&storage.batch) }
211    #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(&storage.batch) }
212
213    #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(&storage.batch) }
214    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(&storage.batch) }
215
216    #[inline] fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Key<'a>> { self.cursor.get_key(&storage.batch) }
217    #[inline] fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option<Self::Val<'a>> { self.cursor.get_val(&storage.batch) }
218
219    #[inline]
220    fn map_times<L: FnMut(&TInner, Self::DiffGat<'_>)>(&mut self, storage: &Self::Storage, mut logic: L) {
221        let key = self.key(storage);
222        let val = self.val(storage);
223        let logic2 = &mut self.logic;
224        self.cursor.map_times(&storage.batch, |time, diff| {
225            logic(&logic2(key, val, time), diff)
226        })
227    }
228
229    #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(&storage.batch) }
230    #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(&storage.batch, key) }
231
232    #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(&storage.batch) }
233    #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(&storage.batch, val) }
234
235    #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(&storage.batch) }
236    #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(&storage.batch) }
237}
differential_dataflow/trace/wrappers/enter_at.rs

differential_dataflow/trace/wrappers/
enter_at.rs
