laminar_core/checkpoint/committed_checkpoint/
view_frontiers.rs

1//! Committed event-time frontiers of append-only views in a cluster cut.
2
3use super::{
4    validate_name, BTreeMap, CheckpointManifest, CheckpointScope, CheckpointWatermark,
5    CommittedCheckpointIndex, SOURCE_WATERMARK_INDEX_VERSION,
6};
7
8/// The committed frontier of each append-only view at a cluster cut: the minimum of the
9/// participants' frontiers, never below the predecessor's. A participant whose view has no
10/// frontier yet holds the predecessor's, so rows it may still replay stay retained; an idle
11/// participant takes no part. A view no participant logs is dropped.
12#[must_use]
13pub fn fold_view_frontiers<'a>(
14    predecessor: &BTreeMap<String, i64>,
15    manifests: impl IntoIterator<Item = &'a CheckpointManifest>,
16) -> BTreeMap<String, i64> {
17    let mut reported = BTreeMap::<&str, CheckpointWatermark>::new();
18    for log in manifests
19        .into_iter()
20        .flat_map(|manifest| &manifest.view_logs)
21    {
22        let participant = match (log.idle, log.frontier) {
23            (true, _) => CheckpointWatermark::Idle,
24            (false, None) => CheckpointWatermark::Uninitialized,
25            (false, Some(frontier)) => CheckpointWatermark::Active(frontier),
26        };
27        reported
28            .entry(log.view.as_str())
29            .and_modify(|cut| *cut = cut.cluster_min(participant))
30            .or_insert(participant);
31    }
32    reported
33        .into_iter()
34        .filter_map(|(view, cut)| {
35            let held = predecessor.get(view).copied();
36            let frontier = match cut.active_value() {
37                Some(minimum) => Some(held.map_or(minimum, |held| held.max(minimum))),
38                None => held,
39            };
40            frontier.map(|frontier| (view.to_owned(), frontier))
41        })
42        .collect()
43}
44
45impl CommittedCheckpointIndex {
46    pub(super) fn validate_view_frontiers(&self) -> Result<(), String> {
47        if self.view_frontiers.is_empty() {
48            return Ok(());
49        }
50        if self.version <= SOURCE_WATERMARK_INDEX_VERSION || self.scope != CheckpointScope::Cluster
51        {
52            return Err("only a version 5 cluster checkpoint carries view frontiers".into());
53        }
54        for (view, frontier) in &self.view_frontiers {
55            validate_name("view frontier", view)?;
56            if *frontier == i64::MIN {
57                return Err(format!(
58                    "view frontier '{view}' uses the reserved uninitialized value"
59                ));
60            }
61        }
62        Ok(())
63    }
64
65    /// No view's committed frontier regresses across `predecessor`.
66    pub(crate) fn validate_view_frontier_continuity(
67        &self,
68        predecessor: &CommittedCheckpointIndex,
69    ) -> Result<(), String> {
70        for (view, frontier) in &self.view_frontiers {
71            if predecessor
72                .view_frontiers
73                .get(view)
74                .is_some_and(|predecessor| predecessor > frontier)
75            {
76                return Err(format!(
77                    "committed frontier of view '{view}' regresses across its predecessor"
78                ));
79            }
80        }
81        Ok(())
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use super::*;
88    use crate::checkpoint::ViewLog;
89
90    fn manifest(logs: &[(&str, Option<i64>, bool)]) -> CheckpointManifest {
91        let mut manifest = CheckpointManifest::new(1, 1);
92        manifest.view_logs = logs
93            .iter()
94            .map(|&(view, frontier, idle)| ViewLog {
95                view: view.into(),
96                next_sequence: 0,
97                frontier,
98                idle,
99                segments: Vec::new(),
100            })
101            .collect();
102        manifest
103    }
104
105    fn frontiers(entries: &[(&str, i64)]) -> BTreeMap<String, i64> {
106        entries
107            .iter()
108            .map(|&(view, frontier)| (view.to_owned(), frontier))
109            .collect()
110    }
111
112    #[test]
113    fn active_participants_set_the_minimum_and_idle_ones_take_no_part() {
114        let cut = [
115            manifest(&[("v", Some(300), false)]),
116            manifest(&[("v", Some(200), false)]),
117            manifest(&[("v", Some(50), true)]),
118        ];
119        assert_eq!(
120            fold_view_frontiers(&frontiers(&[("v", 100)]), &cut),
121            frontiers(&[("v", 200)])
122        );
123    }
124
125    #[test]
126    fn an_uninitialized_participant_holds_the_predecessor_frontier() {
127        let cut = [
128            manifest(&[("v", Some(300), false)]),
129            manifest(&[("v", None, false)]),
130        ];
131        assert_eq!(
132            fold_view_frontiers(&frontiers(&[("v", 100)]), &cut),
133            frontiers(&[("v", 100)])
134        );
135        assert!(fold_view_frontiers(&BTreeMap::new(), &cut).is_empty());
136    }
137
138    #[test]
139    fn the_frontier_never_regresses_and_an_all_idle_view_holds_it() {
140        let lower = [manifest(&[("v", Some(80), false)])];
141        assert_eq!(
142            fold_view_frontiers(&frontiers(&[("v", 100)]), &lower),
143            frontiers(&[("v", 100)])
144        );
145        let idle = [manifest(&[("v", Some(500), true)])];
146        assert_eq!(
147            fold_view_frontiers(&frontiers(&[("v", 100)]), &idle),
148            frontiers(&[("v", 100)])
149        );
150    }
151
152    #[test]
153    fn views_no_participant_logs_are_dropped() {
154        let cut = [manifest(&[("kept", Some(10), false)])];
155        assert_eq!(
156            fold_view_frontiers(&frontiers(&[("dropped", 5), ("kept", 1)]), &cut),
157            frontiers(&[("kept", 10)])
158        );
159    }
160}
laminar_core/checkpoint/committed_checkpoint/view_frontiers.rs

laminar_core/checkpoint/committed_checkpoint/
view_frontiers.rs
