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::{Span, TraceReader};
12use crate::lattice::Lattice;
13
14/// Wrapper to provide trace to nested scope.
15pub struct TraceFrontier<Tr: TraceReader> {
16    trace: Tr,
17    /// Frontier to which all update times will be advanced.
18    since: Antichain<Tr::Time>,
19    /// Frontier after which all update times will be suppressed.
20    until: Antichain<Tr::Time>,
21}
22
23impl<Tr: TraceReader + Clone> Clone for TraceFrontier<Tr> {
24    fn clone(&self) -> Self {
25        TraceFrontier {
26            trace: self.trace.clone(),
27            since: self.since.clone(),
28            until: self.until.clone(),
29        }
30    }
31}
32
33impl<Tr: TraceReader> TraceReader for TraceFrontier<Tr> {
34
35    type Time = Tr::Time;
36    type Batch = BatchFrontier<Tr::Batch, Tr::Time>;
37
38    fn map_spans<F: FnMut(&Span<Tr::Time, Self::Batch>)>(&self, mut f: F) {
39        let since = self.since.borrow();
40        let until = self.until.borrow();
41        self.trace.map_spans(|span| {
42            let wrapped = Span::new(
43                span.desc.clone(),
44                span.inner.clone().map(|b| BatchFrontier::make_from(b, since, until)),
45            );
46            f(&wrapped)
47        })
48    }
49
50    fn set_logical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) { self.trace.set_logical_compaction(frontier) }
51    fn get_logical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { self.trace.get_logical_compaction() }
52
53    fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) { self.trace.set_physical_compaction(frontier) }
54    fn get_physical_compaction(&mut self) -> AntichainRef<'_, Tr::Time> { self.trace.get_physical_compaction() }
55
56    fn spans_through(&mut self, upper: AntichainRef<'_, Tr::Time>) -> Option<Vec<Span<Tr::Time, Self::Batch>>> {
57        let storage = self.trace.spans_through(upper)?;
58        let since = self.since.borrow();
59        let until = self.until.borrow();
60        Some(storage.into_iter().map(|span| {
61            Span::new(span.desc, span.inner.map(|b| BatchFrontier::make_from(b, since, until)))
62        }).collect())
63    }
64}
65
66impl<Tr: TraceReader> TraceFrontier<Tr> {
67    /// Makes a new trace wrapper
68    pub fn make_from(trace: Tr, since: AntichainRef<'_, Tr::Time>, until: AntichainRef<'_, Tr::Time>) -> Self {
69        TraceFrontier {
70            trace,
71            since: since.to_owned(),
72            until: until.to_owned(),
73        }
74    }
75}
76
77
78/// Wrapper to restrict a batch's times to a frontier range.
79#[derive(Clone)]
80pub struct BatchFrontier<B, T> {
81    batch: B,
82    since: Antichain<T>,
83    until: Antichain<T>,
84}
85
86impl<B, T> BatchFrontier<B, T> {
87    /// Makes a new wrapper
88    pub fn make_from(batch: B, since: AntichainRef<T>, until: AntichainRef<T>) -> Self
89    where T: Clone,
90    {
91        BatchFrontier {
92            batch,
93            since: since.to_owned(),
94            until: until.to_owned(),
95        }
96    }
97
98    /// The wrapped batch, whose times are neither advanced nor suppressed.
99    ///
100    /// Its times must be presented through [`BatchFrontier::advance_time`], which is the whole of
101    /// the wrapper's read-side semantics.
102    pub fn inner(&self) -> &B { &self.batch }
103
104    /// Applies the wrapper's time semantics to a time of the wrapped batch.
105    ///
106    /// The time is advanced by `since`, which accumulates the updates at or before `since` rather
107    /// than presenting them partially accumulated. The method returns whether the advanced time
108    /// should be presented at all: times at or after `until` are suppressed, even when they are
109    /// part of a batch that spans `until`.
110    #[inline]
111    pub fn advance_time(&self, time: &mut T) -> bool
112    where T: Lattice + timely::progress::Timestamp,
113    {
114        time.advance_by(self.since.borrow());
115        !self.until.less_equal(time)
116    }
117}
differential_dataflow/trace/wrappers/frontier.rs

differential_dataflow/trace/wrappers/
frontier.rs
