differential_dataflow/trace/wrappers/
filter.rs

1//! Wrapper for filtered trace.
2
3use timely::progress::Timestamp;
4use timely::progress::frontier::AntichainRef;
5
6use crate::trace::{TraceReader, BatchReader, Description};
7use crate::trace::cursor::Cursor;
8
9/// Wrapper to provide trace to nested scope.
10pub struct TraceFilter<Tr, F> {
11    trace: Tr,
12    logic: F,
13}
14
15impl<Tr,F> Clone for TraceFilter<Tr, F>
16where
17    Tr: TraceReader+Clone,
18    F: Clone,
19{
20    fn clone(&self) -> Self {
21        TraceFilter {
22            trace: self.trace.clone(),
23            logic: self.logic.clone(),
24        }
25    }
26}
27
28impl<Tr, F> TraceReader for TraceFilter<Tr, F>
29where
30    Tr: TraceReader,
31    Tr::Batch: Clone,
32    Tr::Time: Timestamp,
33    Tr::Diff: 'static,
34    F: FnMut(Tr::Key<'_>, Tr::Val<'_>)->bool+Clone+'static,
35{
36    type Key<'a> = Tr::Key<'a>;
37    type KeyOwned = Tr::KeyOwned;
38    type Val<'a> = Tr::Val<'a>;
39    type ValOwned = Tr::ValOwned;
40    type Time = Tr::Time;
41    type Diff = Tr::Diff;
42
43    type Batch = BatchFilter<Tr::Batch, F>;
44    type Storage = Tr::Storage;
45    type Cursor = CursorFilter<Tr::Cursor, F>;
46
47    fn map_batches<F2: FnMut(&Self::Batch)>(&self, mut f: F2) {
48        let logic = self.logic.clone();
49        self.trace
50            .map_batches(|batch| f(&Self::Batch::make_from(batch.clone(), logic.clone())))
51    }
52
53    fn set_logical_compaction(&mut self, frontier: AntichainRef<Tr::Time>) { self.trace.set_logical_compaction(frontier) }
54    fn get_logical_compaction(&mut self) -> AntichainRef<Tr::Time> { self.trace.get_logical_compaction() }
55
56    fn set_physical_compaction(&mut self, frontier: AntichainRef<Tr::Time>) { self.trace.set_physical_compaction(frontier) }
57    fn get_physical_compaction(&mut self) -> AntichainRef<Tr::Time> { self.trace.get_physical_compaction() }
58
59    fn cursor_through(&mut self, upper: AntichainRef<Tr::Time>) -> Option<(Self::Cursor, Self::Storage)> {
60        self.trace.cursor_through(upper).map(|(x,y)| (CursorFilter::new(x, self.logic.clone()), y))
61    }
62}
63
64impl<Tr, F> TraceFilter<Tr, F>
65where
66    Tr: TraceReader,
67    Tr::Time: Timestamp,
68{
69    /// Makes a new trace wrapper
70    pub fn make_from(trace: Tr, logic: F) -> Self {
71        TraceFilter {
72            trace,
73            logic,
74        }
75    }
76}
77
78
79/// Wrapper to provide batch to nested scope.
80#[derive(Clone)]
81pub struct BatchFilter<B, F> {
82    batch: B,
83    logic: F,
84}
85
86impl<B, F> BatchReader for BatchFilter<B, F>
87where
88    B: BatchReader,
89    B::Time: Timestamp,
90    F: FnMut(B::Key<'_>, B::Val<'_>)->bool+Clone+'static
91{
92    type Key<'a> = B::Key<'a>;
93    type KeyOwned = B::KeyOwned;
94    type Val<'a> = B::Val<'a>;
95    type ValOwned = B::ValOwned;
96    type Time = B::Time;
97    type Diff = B::Diff;
98
99    type Cursor = BatchCursorFilter<B::Cursor, F>;
100
101    fn cursor(&self) -> Self::Cursor {
102        BatchCursorFilter::new(self.batch.cursor(), self.logic.clone())
103    }
104    fn len(&self) -> usize { self.batch.len() }
105    fn description(&self) -> &Description<B::Time> { self.batch.description() }
106}
107
108impl<B, F> BatchFilter<B, F>
109where
110    B: BatchReader,
111    B::Time: Timestamp,
112{
113    /// Makes a new batch wrapper
114    pub fn make_from(batch: B, logic: F) -> Self {
115        BatchFilter {
116            batch,
117            logic,
118        }
119    }
120}
121
122/// Wrapper to provide cursor to nested scope.
123pub struct CursorFilter<C, F> {
124    cursor: C,
125    logic: F,
126}
127
128impl<C, F> CursorFilter<C, F> {
129    fn new(cursor: C, logic: F) -> Self {
130        CursorFilter {
131            cursor,
132            logic,
133        }
134    }
135}
136
137impl<C, F> Cursor for CursorFilter<C, F>
138where
139    C: Cursor,
140    C::Time: Timestamp,
141    F: FnMut(C::Key<'_>, C::Val<'_>)->bool+'static
142{
143    type Key<'a> = C::Key<'a>;
144    type KeyOwned = C::KeyOwned;
145    type Val<'a> = C::Val<'a>;
146    type ValOwned = C::ValOwned;
147    type Time = C::Time;
148    type Diff = C::Diff;
149
150    type Storage = C::Storage;
151
152    #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(storage) }
153    #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(storage) }
154
155    #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(storage) }
156    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(storage) }
157
158    #[inline]
159    fn map_times<L: FnMut(&Self::Time,&Self::Diff)>(&mut self, storage: &Self::Storage, logic: L) {
160        let key = self.key(storage);
161        let val = self.val(storage);
162        if (self.logic)(key, val) {
163            self.cursor.map_times(storage, logic)
164        }
165    }
166
167    #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(storage) }
168    #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(storage, key) }
169
170    #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(storage) }
171    #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(storage, val) }
172
173    #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(storage) }
174    #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(storage) }
175}
176
177
178
179/// Wrapper to provide cursor to nested scope.
180pub struct BatchCursorFilter<C, F> {
181    cursor: C,
182    logic: F,
183}
184
185impl<C, F> BatchCursorFilter<C, F> {
186    fn new(cursor: C, logic: F) -> Self {
187        BatchCursorFilter {
188            cursor,
189            logic,
190        }
191    }
192}
193
194impl<C: Cursor, F> Cursor for BatchCursorFilter<C, F>
195where
196    C::Time: Timestamp,
197    F: FnMut(C::Key<'_>, C::Val<'_>)->bool+'static,
198{
199    type Key<'a> = C::Key<'a>;
200    type KeyOwned = C::KeyOwned;
201    type Val<'a> = C::Val<'a>;
202    type ValOwned = C::ValOwned;
203    type Time = C::Time;
204    type Diff = C::Diff;
205
206    type Storage = BatchFilter<C::Storage, F>;
207
208    #[inline] fn key_valid(&self, storage: &Self::Storage) -> bool { self.cursor.key_valid(&storage.batch) }
209    #[inline] fn val_valid(&self, storage: &Self::Storage) -> bool { self.cursor.val_valid(&storage.batch) }
210
211    #[inline] fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { self.cursor.key(&storage.batch) }
212    #[inline] fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { self.cursor.val(&storage.batch) }
213
214    #[inline]
215    fn map_times<L: FnMut(&Self::Time,&Self::Diff)>(&mut self, storage: &Self::Storage, logic: L) {
216        let key = self.key(storage);
217        let val = self.val(storage);
218        if (self.logic)(key, val) {
219            self.cursor.map_times(&storage.batch, logic)
220        }
221    }
222
223    #[inline] fn step_key(&mut self, storage: &Self::Storage) { self.cursor.step_key(&storage.batch) }
224    #[inline] fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { self.cursor.seek_key(&storage.batch, key) }
225
226    #[inline] fn step_val(&mut self, storage: &Self::Storage) { self.cursor.step_val(&storage.batch) }
227    #[inline] fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { self.cursor.seek_val(&storage.batch, val) }
228
229    #[inline] fn rewind_keys(&mut self, storage: &Self::Storage) { self.cursor.rewind_keys(&storage.batch) }
230    #[inline] fn rewind_vals(&mut self, storage: &Self::Storage) { self.cursor.rewind_vals(&storage.batch) }
231}
differential_dataflow/trace/wrappers/filter.rs

differential_dataflow/trace/wrappers/
filter.rs
