differential_dataflow/trace/wrappers/
enter.rs

1//! Wrappers to provide trace access to nested scopes.
2
3use std::marker::PhantomData;
4
5use timely::progress::timestamp::Refines;
6use timely::progress::{Antichain, frontier::AntichainRef};
7
8use crate::lattice::Lattice;
9use crate::trace::{Span, Description, TraceReader};
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
28/// Converts a description of outer times to one of inner times.
29pub fn enter_description<T: timely::progress::Timestamp, TInner: Refines<T>+Lattice>(desc: &Description<T>) -> Description<TInner> {
30    let lower: Vec<_> = desc.lower().elements().iter().map(|x| TInner::to_inner(x.clone())).collect();
31    let upper: Vec<_> = desc.upper().elements().iter().map(|x| TInner::to_inner(x.clone())).collect();
32    let since: Vec<_> = desc.since().elements().iter().map(|x| TInner::to_inner(x.clone())).collect();
33    Description::new(Antichain::from(lower), Antichain::from(upper), Antichain::from(since))
34}
35
36/// Converts an outer span to an inner one: the description enters the scope, and the batch is
37/// wrapped so that its readers do the same.
38pub fn enter_span<T: timely::progress::Timestamp, TInner: Refines<T>+Lattice, B>(span: Span<T, B>) -> Span<TInner, BatchEnter<B, TInner>> {
39    Span::new(enter_description(&span.desc), span.inner.map(BatchEnter::make_from))
40}
41
42impl<Tr, TInner> TraceReader for TraceEnter<Tr, TInner>
43where
44    Tr: TraceReader,
45    TInner: Refines<Tr::Time>+Lattice,
46{
47    type Time = TInner;
48    type Batch = BatchEnter<Tr::Batch, TInner>;
49
50    fn map_spans<F: FnMut(&Span<TInner, Self::Batch>)>(&self, mut f: F) {
51        self.trace.map_spans(|span| {
52            f(&enter_span(span.clone()));
53        })
54    }
55
56    fn set_logical_compaction(&mut self, frontier: AntichainRef<'_, TInner>) {
57        self.stash1.clear();
58        for time in frontier.iter() {
59            self.stash1.insert(time.clone().to_outer());
60        }
61        self.trace.set_logical_compaction(self.stash1.borrow());
62    }
63    fn get_logical_compaction(&mut self) -> AntichainRef<'_, TInner> {
64        self.stash2.clear();
65        for time in self.trace.get_logical_compaction().iter() {
66            self.stash2.insert(TInner::to_inner(time.clone()));
67        }
68        self.stash2.borrow()
69    }
70
71    fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, TInner>) {
72        self.stash1.clear();
73        for time in frontier.iter() {
74            self.stash1.insert(time.clone().to_outer());
75        }
76        self.trace.set_physical_compaction(self.stash1.borrow());
77    }
78    fn get_physical_compaction(&mut self) -> AntichainRef<'_, TInner> {
79        self.stash2.clear();
80        for time in self.trace.get_physical_compaction().iter() {
81            self.stash2.insert(TInner::to_inner(time.clone()));
82        }
83        self.stash2.borrow()
84    }
85
86    fn spans_through(&mut self, upper: AntichainRef<TInner>) -> Option<Vec<Span<TInner, Self::Batch>>> {
87        self.stash1.clear();
88        for time in upper.iter() {
89            self.stash1.insert(time.clone().to_outer());
90        }
91        let storage = self.trace.spans_through(self.stash1.borrow())?;
92        Some(storage.into_iter().map(enter_span).collect())
93    }
94}
95
96impl<Tr, TInner> TraceEnter<Tr, TInner>
97where
98    Tr: TraceReader,
99    TInner: Refines<Tr::Time>+Lattice,
100{
101    /// Makes a new trace wrapper
102    pub fn make_from(trace: Tr) -> Self {
103        TraceEnter {
104            trace,
105            stash1: Antichain::new(),
106            stash2: Antichain::new(),
107        }
108    }
109}
110
111
112/// Wrapper to provide a batch to a nested scope.
113#[derive(Clone)]
114pub struct BatchEnter<B, TInner> {
115    batch: B,
116    phantom: PhantomData<TInner>,
117}
118
119impl<B, TInner> BatchEnter<B, TInner> {
120    /// The wrapped batch, whose times are those of the containing scope.
121    ///
122    /// Each of its times enters the nested scope as `TInner::to_inner(time)`; that rule is the
123    /// whole of the wrapper's read-side semantics, and any reader of the wrapped batch must
124    /// apply it.
125    pub fn inner(&self) -> &B { &self.batch }
126
127    /// Makes a new wrapper
128    pub fn make_from(batch: B) -> Self {
129        BatchEnter { batch, phantom: PhantomData }
130    }
131}
differential_dataflow/trace/wrappers/enter.rs

differential_dataflow/trace/wrappers/
enter.rs
