mz_timestamp_oracle/
lib.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! An interface/trait that provides write and read timestamps, reads observe
11//! exactly their preceding writes.
12//!
13//! Specifically, all read timestamps will be greater or equal to all previously
14//! reported completed write timestamps, and strictly less than all subsequently
15//! emitted write timestamps.
16
17use async_trait::async_trait;
18use mz_ore::now::{EpochMillis, NowFn};
19
20pub mod batching_oracle;
21pub mod config;
22pub mod metrics;
23pub mod postgres_oracle;
24pub mod retry;
25
26pub use config::TimestampOracleConfig;
27
28/// Timestamps used by writes in an Append command.
29#[derive(Debug)]
30pub struct WriteTimestamp<T = mz_repr::Timestamp> {
31    /// Timestamp that the write will take place on.
32    pub timestamp: T,
33    /// Timestamp to advance the appended table to.
34    pub advance_to: T,
35}
36
37/// A type that provides write and read timestamps, reads observe exactly their
38/// preceding writes.
39///
40/// Specifically, all read timestamps will be greater or equal to all previously
41/// reported completed write timestamps, and strictly less than all subsequently
42/// emitted write timestamps.
43#[async_trait]
44pub trait TimestampOracle<T>: std::fmt::Debug {
45    /// Acquire a new timestamp for writing.
46    ///
47    /// This timestamp will be strictly greater than all prior values of
48    /// `self.read_ts()` and `self.write_ts()`.
49    async fn write_ts(&self) -> WriteTimestamp<T>;
50
51    /// Peek the current write timestamp.
52    async fn peek_write_ts(&self) -> T;
53
54    /// Acquire a new timestamp for reading.
55    ///
56    /// This timestamp will be greater or equal to all prior values of
57    /// `self.apply_write(write_ts)`, and strictly less than all subsequent
58    /// values of `self.write_ts()`.
59    async fn read_ts(&self) -> T;
60
61    /// Mark a write at `write_ts` completed.
62    ///
63    /// All subsequent values of `self.read_ts()` will be greater or equal to
64    /// `write_ts`.
65    async fn apply_write(&self, lower_bound: T);
66}
67
68/// A [`NowFn`] that is generic over the timestamp.
69///
70/// The oracle operations work in terms of [`mz_repr::Timestamp`] and we could
71/// work around it by bridging between the two in the oracle implementation
72/// itself. This wrapper type makes that slightly easier, though.
73pub trait GenericNowFn<T>: Clone + Send + Sync {
74    fn now(&self) -> T;
75}
76
77impl GenericNowFn<mz_repr::Timestamp> for NowFn<EpochMillis> {
78    fn now(&self) -> mz_repr::Timestamp {
79        (self)().into()
80    }
81}
82
83impl<T: Clone + Send + Sync> GenericNowFn<T> for NowFn<T> {
84    fn now(&self) -> T {
85        (self)()
86    }
87}
88
89// TODO: Gate this with a `#[cfg(test)]` again once the legacy catalog impl goes
90// away.
91pub mod tests {
92    use std::sync::Arc;
93
94    use futures::Future;
95    use mz_repr::Timestamp;
96
97    use super::*;
98
99    // These test methods are meant to be used by tests for timestamp oracle
100    // implementations.
101
102    pub async fn timestamp_oracle_impl_test<F, NewFn>(
103        mut new_fn: NewFn,
104    ) -> Result<(), anyhow::Error>
105    where
106        F: Future<Output = Arc<dyn TimestampOracle<Timestamp> + Send + Sync>>,
107        NewFn: FnMut(String, NowFn, Timestamp) -> F,
108    {
109        // Normally, these could all be separate test methods but we bundle them
110        // all together so that it's easier to call this one test method from
111        // the implementation tests.
112
113        // Timestamp::MIN as initial timestamp
114        let timeline = uuid::Uuid::new_v4().to_string();
115        let oracle = new_fn(timeline, NowFn::from(|| 0u64), Timestamp::MIN).await;
116        assert_eq!(oracle.read_ts().await, Timestamp::MIN);
117        assert_eq!(oracle.peek_write_ts().await, Timestamp::MIN);
118
119        // Timestamp::MAX as initial timestamp
120        let timeline = uuid::Uuid::new_v4().to_string();
121        let oracle = new_fn(timeline, NowFn::from(|| 0u64), Timestamp::MAX).await;
122        assert_eq!(oracle.read_ts().await, Timestamp::MAX);
123        assert_eq!(oracle.peek_write_ts().await, Timestamp::MAX);
124
125        // Timestamp::MAX-1 from NowFn. We have to step back by one, otherwise
126        // `write_ts` can't determine the "advance_to" timestamp.
127        let timeline = uuid::Uuid::new_v4().to_string();
128        let oracle = new_fn(
129            timeline,
130            NowFn::from(|| Timestamp::MAX.step_back().expect("known to work").into()),
131            Timestamp::MIN,
132        )
133        .await;
134        // At first, read_ts and peek_write_ts stay where they are.
135        assert_eq!(oracle.read_ts().await, Timestamp::MIN);
136        assert_eq!(oracle.peek_write_ts().await, Timestamp::MIN);
137        assert_eq!(
138            oracle.write_ts().await.timestamp,
139            Timestamp::MAX.step_back().expect("known to work")
140        );
141        // Now peek_write_ts jump to MAX-1 but read_ts stays.
142        assert_eq!(oracle.read_ts().await, Timestamp::MIN);
143        assert_eq!(
144            oracle.peek_write_ts().await,
145            Timestamp::MAX.step_back().expect("known to work")
146        );
147
148        // Repeated write_ts calls advance the timestamp.
149        let timeline = uuid::Uuid::new_v4().to_string();
150        let oracle = new_fn(timeline, NowFn::from(|| 0u64), Timestamp::MIN).await;
151        assert_eq!(oracle.write_ts().await.timestamp, Timestamp::from(1u64));
152        assert_eq!(oracle.write_ts().await.timestamp, Timestamp::from(2u64));
153
154        // Repeated peek_write_ts calls _DON'T_ advance the timestamp.
155        let timeline = uuid::Uuid::new_v4().to_string();
156        let oracle = new_fn(timeline, NowFn::from(|| 0u64), Timestamp::MIN).await;
157        assert_eq!(oracle.peek_write_ts().await, Timestamp::from(0u64));
158        assert_eq!(oracle.peek_write_ts().await, Timestamp::from(0u64));
159
160        // Interesting scenarios around apply_write, from its rustdoc.
161        //
162        // Scenario #1:
163        // input <= r_0 <= w_0 -> r_1 = r_0 and w_1 = w_0
164        let timeline = uuid::Uuid::new_v4().to_string();
165        let oracle = new_fn(timeline, NowFn::from(|| 0u64), 10u64.into()).await;
166        oracle.apply_write(5u64.into()).await;
167        assert_eq!(oracle.peek_write_ts().await, Timestamp::from(10u64));
168        assert_eq!(oracle.read_ts().await, Timestamp::from(10u64));
169
170        // Scenario #2:
171        // r_0 <= input <= w_0 -> r_1 = input and w_1 = w_0
172        let timeline = uuid::Uuid::new_v4().to_string();
173        let oracle = new_fn(timeline, NowFn::from(|| 0u64), 0u64.into()).await;
174        // Have to bump the write_ts up manually:
175        assert_eq!(oracle.write_ts().await.timestamp, Timestamp::from(1u64));
176        assert_eq!(oracle.write_ts().await.timestamp, Timestamp::from(2u64));
177        assert_eq!(oracle.write_ts().await.timestamp, Timestamp::from(3u64));
178        assert_eq!(oracle.write_ts().await.timestamp, Timestamp::from(4u64));
179        oracle.apply_write(2u64.into()).await;
180        assert_eq!(oracle.peek_write_ts().await, Timestamp::from(4u64));
181        assert_eq!(oracle.read_ts().await, Timestamp::from(2u64));
182
183        // Scenario #3:
184        // r_0 <= w_0 <= input -> r_1 = input and w_1 = input
185        let timeline = uuid::Uuid::new_v4().to_string();
186        let oracle = new_fn(timeline, NowFn::from(|| 0u64), 0u64.into()).await;
187        oracle.apply_write(2u64.into()).await;
188        assert_eq!(oracle.peek_write_ts().await, Timestamp::from(2u64));
189        assert_eq!(oracle.read_ts().await, Timestamp::from(2u64));
190        oracle.apply_write(4u64.into()).await;
191        assert_eq!(oracle.peek_write_ts().await, Timestamp::from(4u64));
192        assert_eq!(oracle.read_ts().await, Timestamp::from(4u64));
193
194        Ok(())
195    }
196}
mz_timestamp_oracle/lib.rs

mz_timestamp_oracle/
lib.rs
