mz_repr/
timestamp.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
10use std::convert::TryFrom;
11use std::num::TryFromIntError;
12use std::time::Duration;
13
14use dec::TryFromDecimalError;
15use mz_proto::{RustType, TryFromProtoError};
16use mz_timely_util::temporal::BucketTimestamp;
17#[cfg(any(test, feature = "proptest"))]
18use proptest_derive::Arbitrary;
19use serde::{Deserialize, Serialize, Serializer};
20
21use chrono::{DateTime, Utc};
22
23use crate::adt::numeric::Numeric;
24use crate::adt::timestamp::CheckedTimestamp;
25use crate::refresh_schedule::RefreshSchedule;
26use crate::strconv::{ParseError, parse_timestamptz, parse_timestamptz_legacy};
27
28include!(concat!(env!("OUT_DIR"), "/mz_repr.timestamp.rs"));
29
30/// System-wide timestamp type.
31#[derive(
32    Clone,
33    // TODO: De-implement Copy, which is widely used.
34    Copy,
35    PartialEq,
36    Eq,
37    PartialOrd,
38    Ord,
39    Hash,
40    Default,
41    bytemuck::AnyBitPattern,
42    bytemuck::NoUninit,
43)]
44#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
45#[repr(transparent)]
46pub struct Timestamp {
47    /// note no `pub`.
48    internal: u64,
49}
50
51impl PartialEq<&Timestamp> for Timestamp {
52    fn eq(&self, other: &&Timestamp) -> bool {
53        self.eq(*other)
54    }
55}
56
57impl PartialEq<Timestamp> for &Timestamp {
58    fn eq(&self, other: &Timestamp) -> bool {
59        self.internal.eq(&other.internal)
60    }
61}
62
63impl RustType<ProtoTimestamp> for Timestamp {
64    fn into_proto(&self) -> ProtoTimestamp {
65        ProtoTimestamp {
66            internal: self.into(),
67        }
68    }
69
70    fn from_proto(proto: ProtoTimestamp) -> Result<Self, TryFromProtoError> {
71        Ok(Timestamp::new(proto.internal))
72    }
73}
74
75mod columnar_timestamp {
76    use crate::Timestamp;
77    use columnar::Columnar;
78    use mz_ore::cast::CastFrom;
79    use std::ops::Range;
80
81    /// A newtype wrapper for a vector of `Timestamp` values.
82    #[derive(Clone, Copy, Default, Debug)]
83    pub struct Timestamps<T>(T);
84    impl<D, T: columnar::Push<D>> columnar::Push<D> for Timestamps<T> {
85        #[inline(always)]
86        fn push(&mut self, item: D) {
87            self.0.push(item)
88        }
89    }
90    impl<T: columnar::Clear> columnar::Clear for Timestamps<T> {
91        #[inline(always)]
92        fn clear(&mut self) {
93            self.0.clear()
94        }
95    }
96    impl<T: columnar::Len> columnar::Len for Timestamps<T> {
97        #[inline(always)]
98        fn len(&self) -> usize {
99            self.0.len()
100        }
101    }
102    impl<'a> columnar::Index for Timestamps<&'a [Timestamp]> {
103        type Ref = Timestamp;
104
105        #[inline(always)]
106        fn get(&self, index: usize) -> Self::Ref {
107            self.0[index]
108        }
109    }
110
111    impl Columnar for Timestamp {
112        #[inline(always)]
113        fn into_owned<'a>(other: columnar::Ref<'a, Self>) -> Self {
114            other
115        }
116        type Container = Timestamps<Vec<Timestamp>>;
117        #[inline(always)]
118        fn reborrow<'b, 'a: 'b>(thing: columnar::Ref<'a, Self>) -> columnar::Ref<'b, Self>
119        where
120            Self: 'a,
121        {
122            thing
123        }
124    }
125
126    impl columnar::Borrow for Timestamps<Vec<Timestamp>> {
127        type Ref<'a> = Timestamp;
128        type Borrowed<'a>
129            = Timestamps<&'a [Timestamp]>
130        where
131            Self: 'a;
132        #[inline(always)]
133        fn borrow<'a>(&'a self) -> Self::Borrowed<'a> {
134            Timestamps(self.0.as_slice())
135        }
136        #[inline(always)]
137        fn reborrow<'b, 'a: 'b>(item: Self::Borrowed<'a>) -> Self::Borrowed<'b>
138        where
139            Self: 'a,
140        {
141            Timestamps(item.0)
142        }
143
144        #[inline(always)]
145        fn reborrow_ref<'b, 'a: 'b>(item: Self::Ref<'a>) -> Self::Ref<'b>
146        where
147            Self: 'a,
148        {
149            item
150        }
151    }
152
153    impl columnar::Container for Timestamps<Vec<Timestamp>> {
154        #[inline(always)]
155        fn extend_from_self(&mut self, other: Self::Borrowed<'_>, range: Range<usize>) {
156            self.0.extend_from_self(other.0, range)
157        }
158        #[inline(always)]
159        fn reserve_for<'a, I>(&mut self, selves: I)
160        where
161            Self: 'a,
162            I: Iterator<Item = Self::Borrowed<'a>> + Clone,
163        {
164            self.0.reserve_for(selves.map(|s| s.0));
165        }
166    }
167
168    impl<'a> columnar::AsBytes<'a> for Timestamps<&'a [Timestamp]> {
169        const SLICE_COUNT: usize = 1;
170        #[inline(always)]
171        fn get_byte_slice(&self, index: usize) -> (u64, &'a [u8]) {
172            mz_ore::soft_assert_no_log!(index < Self::SLICE_COUNT);
173            (
174                u64::cast_from(align_of::<Timestamp>()),
175                bytemuck::cast_slice(self.0),
176            )
177        }
178        #[inline(always)]
179        fn as_bytes(&self) -> impl Iterator<Item = (u64, &'a [u8])> {
180            std::iter::once((
181                u64::cast_from(align_of::<Timestamp>()),
182                bytemuck::cast_slice(self.0),
183            ))
184        }
185    }
186    impl<'a> columnar::FromBytes<'a> for Timestamps<&'a [Timestamp]> {
187        const SLICE_COUNT: usize = 1;
188        #[inline(always)]
189        fn from_bytes(bytes: &mut impl Iterator<Item = &'a [u8]>) -> Self {
190            Timestamps(bytemuck::cast_slice(
191                bytes.next().expect("Iterator exhausted prematurely"),
192            ))
193        }
194    }
195}
196
197impl BucketTimestamp for Timestamp {
198    fn advance_by_power_of_two(&self, exponent: u32) -> Option<Self> {
199        let rhs = 1_u64.checked_shl(exponent)?;
200        Some(self.internal.checked_add(rhs)?.into())
201    }
202}
203
204pub trait TimestampManipulation:
205    timely::progress::Timestamp
206    + timely::order::TotalOrder
207    + differential_dataflow::lattice::Lattice
208    + std::fmt::Debug
209    + mz_persist_types::StepForward
210    + Sync
211{
212    /// Advance a timestamp by the least amount possible such that
213    /// `ts.less_than(ts.step_forward())` is true. Panic if unable to do so.
214    fn step_forward(&self) -> Self;
215
216    /// Advance a timestamp forward by the given `amount`. Panic if unable to do so.
217    fn step_forward_by(&self, amount: &Self) -> Self;
218
219    /// Advance a timestamp forward by the given `amount`. Return `None` if unable to do so.
220    fn try_step_forward_by(&self, amount: &Self) -> Option<Self>;
221
222    /// Advance a timestamp by the least amount possible such that `ts.less_than(ts.step_forward())`
223    /// is true. Return `None` if unable to do so.
224    fn try_step_forward(&self) -> Option<Self>;
225
226    /// Retreat a timestamp by the least amount possible such that
227    /// `ts.step_back().unwrap().less_than(ts)` is true. Return `None` if unable,
228    /// which must only happen if the timestamp is `Timestamp::minimum()`.
229    fn step_back(&self) -> Option<Self>;
230
231    /// Return the maximum value for this timestamp.
232    fn maximum() -> Self;
233
234    /// Rounds up the timestamp to the time of the next refresh according to the given schedule.
235    /// Returns None if there is no next refresh.
236    fn round_up(&self, schedule: &RefreshSchedule) -> Option<Self>;
237
238    /// Rounds down `timestamp - 1` to the time of the previous refresh according to the given
239    /// schedule.
240    /// Returns None if there is no previous refresh.
241    fn round_down_minus_1(&self, schedule: &RefreshSchedule) -> Option<Self>;
242}
243
244impl TimestampManipulation for Timestamp {
245    fn step_forward(&self) -> Self {
246        self.step_forward()
247    }
248
249    fn step_forward_by(&self, amount: &Self) -> Self {
250        self.step_forward_by(amount)
251    }
252
253    fn try_step_forward(&self) -> Option<Self> {
254        self.try_step_forward()
255    }
256
257    fn try_step_forward_by(&self, amount: &Self) -> Option<Self> {
258        self.try_step_forward_by(amount)
259    }
260
261    fn step_back(&self) -> Option<Self> {
262        self.step_back()
263    }
264
265    fn maximum() -> Self {
266        Self::MAX
267    }
268
269    fn round_up(&self, schedule: &RefreshSchedule) -> Option<Self> {
270        schedule.round_up_timestamp(*self)
271    }
272
273    fn round_down_minus_1(&self, schedule: &RefreshSchedule) -> Option<Self> {
274        schedule.round_down_timestamp_m1(*self)
275    }
276}
277
278impl mz_persist_types::StepForward for Timestamp {
279    fn step_forward(&self) -> Self {
280        self.step_forward()
281    }
282}
283
284impl Timestamp {
285    pub const MAX: Self = Self { internal: u64::MAX };
286    pub const MIN: Self = Self { internal: u64::MIN };
287
288    pub const fn new(timestamp: u64) -> Self {
289        Self {
290            internal: timestamp,
291        }
292    }
293
294    pub fn to_bytes(&self) -> [u8; 8] {
295        self.internal.to_le_bytes()
296    }
297
298    pub fn from_bytes(bytes: [u8; 8]) -> Self {
299        Self {
300            internal: u64::from_le_bytes(bytes),
301        }
302    }
303
304    pub fn saturating_sub<I: Into<Self>>(self, rhs: I) -> Self {
305        Self {
306            internal: self.internal.saturating_sub(rhs.into().internal),
307        }
308    }
309
310    pub fn saturating_add<I: Into<Self>>(self, rhs: I) -> Self {
311        Self {
312            internal: self.internal.saturating_add(rhs.into().internal),
313        }
314    }
315
316    pub fn saturating_mul<I: Into<Self>>(self, rhs: I) -> Self {
317        Self {
318            internal: self.internal.saturating_mul(rhs.into().internal),
319        }
320    }
321
322    pub fn checked_add<I: Into<Self>>(self, rhs: I) -> Option<Self> {
323        self.internal
324            .checked_add(rhs.into().internal)
325            .map(|internal| Self { internal })
326    }
327
328    pub fn checked_sub<I: Into<Self>>(self, rhs: I) -> Option<Self> {
329        self.internal
330            .checked_sub(rhs.into().internal)
331            .map(|internal| Self { internal })
332    }
333
334    /// Advance a timestamp by the least amount possible such that
335    /// `ts.less_than(ts.step_forward())` is true. Panic if unable to do so.
336    pub fn step_forward(&self) -> Self {
337        match self.checked_add(1) {
338            Some(ts) => ts,
339            None => panic!("could not step forward"),
340        }
341    }
342
343    /// Advance a timestamp forward by the given `amount`. Panic if unable to do so.
344    pub fn step_forward_by(&self, amount: &Self) -> Self {
345        match self.checked_add(*amount) {
346            Some(ts) => ts,
347            None => panic!("could not step {self} forward by {amount}"),
348        }
349    }
350
351    /// Advance a timestamp by the least amount possible such that `ts.less_than(ts.step_forward())`
352    /// is true. Return `None` if unable to do so.
353    pub fn try_step_forward(&self) -> Option<Self> {
354        self.checked_add(1)
355    }
356
357    /// Advance a timestamp forward by the given `amount`. Return `None` if unable to do so.
358    pub fn try_step_forward_by(&self, amount: &Self) -> Option<Self> {
359        self.checked_add(*amount)
360    }
361
362    /// Retreat a timestamp by the least amount possible such that
363    /// `ts.step_back().unwrap().less_than(ts)` is true. Return `None` if unable,
364    /// which must only happen if the timestamp is `Timestamp::minimum()`.
365    pub fn step_back(&self) -> Option<Self> {
366        self.checked_sub(1)
367    }
368}
369
370impl From<u64> for Timestamp {
371    fn from(internal: u64) -> Self {
372        Self { internal }
373    }
374}
375
376impl From<Timestamp> for u64 {
377    fn from(ts: Timestamp) -> Self {
378        ts.internal
379    }
380}
381
382// NOTE: Only timestamps of the `EpochMilliseconds` timeline are wall-clock times.
383// For other timelines the result is not meaningful.
384impl mz_ore::now::AsEpochMillis for Timestamp {
385    fn as_epoch_millis(&self) -> mz_ore::now::EpochMillis {
386        self.internal
387    }
388}
389
390impl From<Timestamp> for u128 {
391    fn from(ts: Timestamp) -> Self {
392        u128::from(ts.internal)
393    }
394}
395
396impl TryFrom<Timestamp> for i64 {
397    type Error = TryFromIntError;
398
399    fn try_from(value: Timestamp) -> Result<Self, Self::Error> {
400        value.internal.try_into()
401    }
402}
403
404impl From<&Timestamp> for u64 {
405    fn from(ts: &Timestamp) -> Self {
406        ts.internal
407    }
408}
409
410impl From<Timestamp> for Numeric {
411    fn from(ts: Timestamp) -> Self {
412        ts.internal.into()
413    }
414}
415
416impl From<Timestamp> for Duration {
417    fn from(ts: Timestamp) -> Self {
418        Duration::from_millis(ts.internal)
419    }
420}
421
422impl std::ops::Rem<Timestamp> for Timestamp {
423    type Output = Timestamp;
424
425    fn rem(self, rhs: Timestamp) -> Self::Output {
426        Self {
427            internal: self.internal % rhs.internal,
428        }
429    }
430}
431
432impl Serialize for Timestamp {
433    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
434    where
435        S: Serializer,
436    {
437        self.internal.serialize(serializer)
438    }
439}
440
441impl<'de> Deserialize<'de> for Timestamp {
442    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
443    where
444        D: serde::Deserializer<'de>,
445    {
446        Ok(Self {
447            internal: u64::deserialize(deserializer)?,
448        })
449    }
450}
451
452impl timely::order::PartialOrder for Timestamp {
453    fn less_equal(&self, other: &Self) -> bool {
454        self.internal.less_equal(&other.internal)
455    }
456}
457
458impl timely::order::PartialOrder<&Timestamp> for Timestamp {
459    fn less_equal(&self, other: &&Self) -> bool {
460        self.internal.less_equal(&other.internal)
461    }
462}
463
464impl timely::order::PartialOrder<Timestamp> for &Timestamp {
465    fn less_equal(&self, other: &Timestamp) -> bool {
466        self.internal.less_equal(&other.internal)
467    }
468}
469
470impl timely::order::TotalOrder for Timestamp {}
471
472impl timely::progress::Timestamp for Timestamp {
473    type Summary = Timestamp;
474
475    fn minimum() -> Self {
476        Self::MIN
477    }
478}
479
480impl timely::progress::PathSummary<Timestamp> for Timestamp {
481    #[inline]
482    fn results_in(&self, src: &Timestamp) -> Option<Timestamp> {
483        self.internal
484            .checked_add(src.internal)
485            .map(|internal| Self { internal })
486    }
487    #[inline]
488    fn followed_by(&self, other: &Timestamp) -> Option<Timestamp> {
489        self.internal
490            .checked_add(other.internal)
491            .map(|internal| Self { internal })
492    }
493}
494
495impl timely::progress::timestamp::Refines<()> for Timestamp {
496    fn to_inner(_: ()) -> Timestamp {
497        Default::default()
498    }
499    fn to_outer(self) -> () {
500        ()
501    }
502    fn summarize(_: <Timestamp as timely::progress::timestamp::Timestamp>::Summary) -> () {
503        ()
504    }
505}
506
507impl differential_dataflow::lattice::Lattice for Timestamp {
508    #[inline]
509    fn join(&self, other: &Self) -> Self {
510        ::std::cmp::max(*self, *other)
511    }
512    #[inline]
513    fn meet(&self, other: &Self) -> Self {
514        ::std::cmp::min(*self, *other)
515    }
516}
517
518impl mz_persist_types::Codec64 for Timestamp {
519    fn codec_name() -> String {
520        u64::codec_name()
521    }
522
523    fn encode(&self) -> [u8; 8] {
524        self.internal.encode()
525    }
526
527    fn decode(buf: [u8; 8]) -> Self {
528        Self {
529            internal: u64::decode(buf),
530        }
531    }
532}
533
534impl std::fmt::Display for Timestamp {
535    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
536        std::fmt::Display::fmt(&self.internal, f)
537    }
538}
539
540impl std::fmt::Debug for Timestamp {
541    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
542        std::fmt::Debug::fmt(&self.internal, f)
543    }
544}
545
546impl std::str::FromStr for Timestamp {
547    type Err = String;
548
549    fn from_str(s: &str) -> Result<Self, Self::Err> {
550        Self::parse_with(s, parse_timestamptz)
551    }
552}
553
554impl Timestamp {
555    /// Like the [`std::str::FromStr`] impl, but the timestamp fallback uses
556    /// the frozen legacy date parsing. Solely for the storage source cast
557    /// `CastStringToMzTimestamp` (see the stability contract in
558    /// `mz_storage_types::sources::casts`).
559    pub fn from_str_legacy(s: &str) -> Result<Self, String> {
560        Self::parse_with(s, parse_timestamptz_legacy)
561    }
562
563    fn parse_with(
564        s: &str,
565        parse_fallback: fn(&str) -> Result<CheckedTimestamp<DateTime<Utc>>, ParseError>,
566    ) -> Result<Self, String> {
567        Ok(Self {
568            internal: s
569                .parse::<u64>()
570                .map_err(|_| "could not parse as number of milliseconds since epoch".to_string())
571                .or_else(|err_num_of_millis| {
572                    parse_fallback(s)
573                        .map_err(|parse_error| {
574                            format!(
575                                "{}; could not parse as date and time: {}",
576                                err_num_of_millis, parse_error
577                            )
578                        })?
579                        .timestamp_millis()
580                        .try_into()
581                        .map_err(|_| "out of range for mz_timestamp".to_string())
582                })
583                .map_err(|e: String| format!("could not parse mz_timestamp: {}", e))?,
584        })
585    }
586}
587
588impl TryFrom<Duration> for Timestamp {
589    type Error = TryFromIntError;
590
591    fn try_from(value: Duration) -> Result<Self, Self::Error> {
592        Ok(Self {
593            internal: value.as_millis().try_into()?,
594        })
595    }
596}
597
598impl TryFrom<u128> for Timestamp {
599    type Error = TryFromIntError;
600
601    fn try_from(value: u128) -> Result<Self, Self::Error> {
602        Ok(Self {
603            internal: value.try_into()?,
604        })
605    }
606}
607
608impl TryFrom<i64> for Timestamp {
609    type Error = TryFromIntError;
610
611    fn try_from(value: i64) -> Result<Self, Self::Error> {
612        Ok(Self {
613            internal: value.try_into()?,
614        })
615    }
616}
617
618impl TryFrom<Numeric> for Timestamp {
619    type Error = TryFromDecimalError;
620
621    fn try_from(value: Numeric) -> Result<Self, Self::Error> {
622        Ok(Self {
623            internal: value.try_into()?,
624        })
625    }
626}
627
628impl columnation::Columnation for Timestamp {
629    type InnerRegion = columnation::CopyRegion<Timestamp>;
630}
631
632/// Returns whether `frontier` is at most `allowed_lag` behind `reference`.
633///
634/// The predicate is `reference <= frontier + allowed_lag`, evaluated on
635/// antichains. Timestamps that would overflow when advanced by `allowed_lag`
636/// saturate at [`Timestamp::MAX`] rather than panicking, so a frontier close to
637/// the end of the timestamp domain reads as far ahead rather than aborting the
638/// caller.
639///
640/// Note the two degenerate cases, which callers usually want to handle
641/// themselves rather than inherit:
642///
643///  * An empty `reference` is the maximum antichain, so the result is `true`
644///    only when `frontier` is empty too. It denotes completion, which a finite
645///    `frontier` cannot satisfy, regardless of the allowance.
646///  * An empty `frontier` is likewise the maximum, so the result is `true` for
647///    every `reference`.
648pub fn frontier_within_lag(
649    frontier: &timely::progress::Antichain<Timestamp>,
650    reference: &timely::progress::Antichain<Timestamp>,
651    allowed_lag: Timestamp,
652) -> bool {
653    // We cannot subtract frontiers, so bump `frontier` forward by the allowance
654    // and compare that against `reference`.
655    let bumped = timely::progress::Antichain::from_iter(frontier.iter().map(|t| {
656        t.try_step_forward_by(&allowed_lag)
657            .unwrap_or(Timestamp::MAX)
658    }));
659    timely::order::PartialOrder::less_equal(reference, &bumped)
660}
661
662#[cfg(test)]
663mod frontier_within_lag_tests {
664    use timely::progress::Antichain;
665
666    use super::{Timestamp, frontier_within_lag};
667
668    fn ac(ts: u64) -> Antichain<Timestamp> {
669        Antichain::from_elem(Timestamp::new(ts))
670    }
671
672    #[mz_ore::test]
673    fn within_and_beyond_the_allowance() {
674        // Exactly at the allowance is within it; one past it is not.
675        assert!(frontier_within_lag(&ac(100), &ac(110), Timestamp::new(10)));
676        assert!(!frontier_within_lag(&ac(100), &ac(111), Timestamp::new(10)));
677        // A frontier ahead of the reference is trivially within any allowance.
678        assert!(frontier_within_lag(&ac(200), &ac(110), Timestamp::new(0)));
679    }
680
681    #[mz_ore::test]
682    fn monotone_in_the_allowance() {
683        // Widening the allowance never turns a `true` into a `false`.
684        for lag in 0..40u64 {
685            let narrow = frontier_within_lag(&ac(100), &ac(120), Timestamp::new(lag));
686            let wide = frontier_within_lag(&ac(100), &ac(120), Timestamp::new(lag + 1));
687            assert!(wide || !narrow, "narrow={narrow} wide={wide} lag={lag}");
688        }
689    }
690
691    #[mz_ore::test]
692    fn empty_frontiers_are_the_maximum() {
693        let empty = Antichain::new();
694        // An empty frontier is ahead of everything.
695        assert!(frontier_within_lag(
696            &empty,
697            &ac(u64::MAX),
698            Timestamp::new(0)
699        ));
700        // An empty reference is only matched by an empty frontier.
701        assert!(!frontier_within_lag(&ac(100), &empty, Timestamp::new(10)));
702        assert!(frontier_within_lag(&empty, &empty, Timestamp::new(10)));
703    }
704
705    #[mz_ore::test]
706    fn saturates_instead_of_panicking_on_overflow() {
707        // `step_forward_by` would panic here; the saturating form reads as
708        // "arbitrarily far ahead" instead.
709        assert!(frontier_within_lag(
710            &ac(u64::MAX - 1),
711            &ac(u64::MAX),
712            Timestamp::new(u64::MAX),
713        ));
714    }
715}
mz_repr/timestamp.rs

mz_repr/
timestamp.rs
