differential_dataflow/trace/wrappers/
enter.rs

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

differential_dataflow/trace/wrappers/
enter.rs
