laminar_core/checkpoint/checkpoint_manifest/
view_log.rs

1//! Append-only materialized-view logs. Each checkpoint packs every view's new segments into one
2//! view-log object per participant; later checkpoints reference a segment while its view's
3//! retention keeps it. The log is not owner state and never enters a state-frame inventory.
4
5use std::collections::HashMap;
6
7use super::{
8    is_sha256, validate_range, ByteRange, CheckpointManifest, ReferencedStateChunk, StateChunkId,
9};
10
11/// Most segments one manifest may retain across every view log. The bound keeps the manifest
12/// within its encoded-size limit; compaction merges a log's segments, so their number follows
13/// the log's bytes rather than the checkpoints that wrote it.
14pub const MAX_VIEW_LOG_SEGMENTS: usize = 32_768;
15
16/// Most segments one cluster cut retains across all of its participants. Cleanup and topology
17/// migration load a cut's manifests, and a migration root beside its first target cut, within
18/// 16 MiB together; at about 285 encoded bytes a segment, two cuts at this budget stay near
19/// 13 MiB and leave room for state frames.
20pub const MAX_CLUSTER_VIEW_LOG_SEGMENTS: usize = 24_576;
21
22/// Most segments one manifest may retain when its cut has `participants` participants: in a
23/// cluster an even share of [`MAX_CLUSTER_VIEW_LOG_SEGMENTS`], and never more than
24/// [`MAX_VIEW_LOG_SEGMENTS`].
25#[must_use]
26pub fn view_log_segment_limit(participants: Option<usize>) -> usize {
27    participants.map_or(MAX_VIEW_LOG_SEGMENTS, |participants| {
28        MAX_VIEW_LOG_SEGMENTS.min(MAX_CLUSTER_VIEW_LOG_SEGMENTS / participants.max(1))
29    })
30}
31
32/// Inclusive event-time range of one segment's rows.
33#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
34#[serde(deny_unknown_fields)]
35pub struct EventTimeBounds {
36    /// Earliest event time in the segment.
37    pub min: i64,
38    /// Latest event time in the segment.
39    pub max: i64,
40}
41
42/// One immutable run of rows in a view-log partition.
43#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
44#[serde(deny_unknown_fields)]
45pub struct ViewLogSegment {
46    /// Log partition within the manifest's vnode domain.
47    pub partition: u16,
48    /// Position of the segment in its view's log; later segments hold later rows.
49    pub sequence: u64,
50    /// View-log object holding the payload.
51    pub chunk: StateChunkId,
52    /// Exact payload range within that object.
53    pub range: ByteRange,
54    /// SHA-256 of only this payload.
55    pub sha256: String,
56    /// Exact number of rows.
57    pub rows: u64,
58    /// Event-time range of the rows; `None` when the view has no event-time column.
59    pub event_time: Option<EventTimeBounds>,
60}
61
62/// One append-only view's retained log at this cut.
63#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
64#[serde(deny_unknown_fields)]
65pub struct ViewLog {
66    /// Catalog name of the view.
67    pub view: String,
68    /// Sequence of the view's next segment; every retained segment is below it.
69    pub next_sequence: u64,
70    /// Greatest event-time frontier the view had reached on this participant at this cut. In a
71    /// local cut an event-time retention keeps exactly the rows at or after this frontier minus
72    /// its interval; a cluster cut uses the committed index's view frontier instead.
73    pub frontier: Option<i64>,
74    /// Whether every input of the view was idle at this cut. An idle participant's frontier
75    /// does not hold back the view's committed cluster frontier.
76    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
77    pub idle: bool,
78    /// Retained segments ordered by partition, then sequence.
79    pub segments: Vec<ViewLogSegment>,
80}
81
82pub(super) fn validate_view_logs(manifest: &CheckpointManifest, error: &mut impl FnMut(String)) {
83    let current = manifest.view_log.as_ref().map(|object| object.chunk);
84    if let Some(object) = &manifest.view_log {
85        if object.chunk != manifest.node_data.chunk {
86            error("view_log chunk must match the manifest participant and checkpoint".into());
87        }
88        if object.object_length == 0 || !is_sha256(&object.sha256) {
89            error("view_log must be a non-empty object with a lowercase SHA-256 digest".into());
90        }
91    }
92    let referenced = referenced_view_logs(manifest, error);
93    if !manifest
94        .view_logs
95        .windows(2)
96        .all(|pair| pair[0].view < pair[1].view)
97    {
98        error("view_logs must be strictly ordered and unique by view".into());
99    }
100
101    let mut segment_count = 0_usize;
102    let mut prior_counts = HashMap::<StateChunkId, u32>::new();
103    let mut current_ranges = Vec::new();
104    for log in &manifest.view_logs {
105        if log.view.is_empty() {
106            error("view log names an empty view".into());
107        }
108        if !log.segments.windows(2).all(|pair| {
109            (pair[0].partition, pair[0].sequence) < (pair[1].partition, pair[1].sequence)
110        }) {
111            error(format!(
112                "view log '{}' segments must be strictly ordered by partition and sequence",
113                log.view
114            ));
115        }
116        segment_count = segment_count.saturating_add(log.segments.len());
117        for segment in &log.segments {
118            let owner = format!(
119                "view log '{}' partition {} segment {}",
120                log.view, segment.partition, segment.sequence
121            );
122            validate_segment(manifest, log, segment, &owner, error);
123            let object_length = if Some(segment.chunk) == current {
124                current_ranges.push((segment.range, owner.clone()));
125                manifest
126                    .view_log
127                    .as_ref()
128                    .map(|object| object.object_length)
129            } else if let Some(reference) = referenced.get(&segment.chunk) {
130                let count = prior_counts.entry(segment.chunk).or_default();
131                *count = count.saturating_add(1);
132                Some(reference.object_length)
133            } else {
134                error(format!(
135                    "{owner} references an untracked view-log object {:?}",
136                    segment.chunk
137                ));
138                None
139            };
140            validate_range(segment.range, object_length, &owner, error);
141        }
142    }
143    let limit = view_log_segment_limit(
144        manifest
145            .assignment_fence
146            .as_ref()
147            .map(|fence| fence.participants.len()),
148    );
149    if segment_count > limit {
150        error(format!(
151            "view logs retain {segment_count} segments; this manifest admits at most {limit}"
152        ));
153    }
154    for reference in &manifest.referenced_view_logs {
155        let actual = prior_counts.get(&reference.chunk).copied().unwrap_or(0);
156        if actual != reference.ref_count.get() {
157            error(format!(
158                "referenced view log {:?} declares ref_count {}, but {actual} segments reference it",
159                reference.chunk, reference.ref_count
160            ));
161        }
162    }
163    validate_current_layout(manifest, current_ranges, error);
164}
165
166fn referenced_view_logs<'a>(
167    manifest: &'a CheckpointManifest,
168    error: &mut impl FnMut(String),
169) -> HashMap<StateChunkId, &'a ReferencedStateChunk> {
170    let references = &manifest.referenced_view_logs;
171    if !references
172        .windows(2)
173        .all(|pair| pair[0].chunk < pair[1].chunk)
174    {
175        error("referenced_view_logs must be strictly ordered and unique".into());
176    }
177    // A cluster participant may list another participant's segments after adopting the log of
178    // one that left; a local cut has one participant.
179    let cluster = manifest.assignment_fence.is_some();
180    for reference in references {
181        if (!cluster && reference.chunk.participant_id != manifest.participant_id)
182            || reference.chunk.checkpoint_id == 0
183            || reference.chunk.checkpoint_id >= manifest.checkpoint_id
184        {
185            error(format!(
186                "referenced view log {:?} is not an older object this participant may reference",
187                reference.chunk
188            ));
189        }
190        if !is_sha256(&reference.sha256) {
191            error(format!(
192                "referenced view log {:?} digest must be lowercase SHA-256",
193                reference.chunk
194            ));
195        }
196    }
197    references
198        .iter()
199        .map(|reference| (reference.chunk, reference))
200        .collect()
201}
202
203fn validate_segment(
204    manifest: &CheckpointManifest,
205    log: &ViewLog,
206    segment: &ViewLogSegment,
207    owner: &str,
208    error: &mut impl FnMut(String),
209) {
210    if segment.partition >= manifest.vnode_count {
211        error(format!("{owner} is outside the vnode domain"));
212    }
213    if segment.sequence >= log.next_sequence {
214        error(format!(
215            "{owner} is not below the log's next sequence {}",
216            log.next_sequence
217        ));
218    }
219    if segment.rows == 0 || segment.range.length == 0 {
220        error(format!("{owner} is empty"));
221    }
222    if !is_sha256(&segment.sha256) {
223        error(format!("{owner} digest must be lowercase SHA-256"));
224    }
225    if segment
226        .event_time
227        .is_some_and(|bounds| bounds.min > bounds.max)
228    {
229        error(format!("{owner} event-time bounds are reversed"));
230    }
231}
232
233fn validate_current_layout(
234    manifest: &CheckpointManifest,
235    current_ranges: Vec<(ByteRange, String)>,
236    error: &mut impl FnMut(String),
237) {
238    let Some(object) = &manifest.view_log else {
239        return;
240    };
241    let mut expected_offset = 0_u64;
242    for (range, owner) in current_ranges {
243        if range.offset != expected_offset {
244            error(format!(
245                "{owner} starts at {}, expected {expected_offset} in the view-log object",
246                range.offset
247            ));
248        }
249        if let Some(end) = range.end() {
250            expected_offset = end;
251        }
252    }
253    if expected_offset != object.object_length {
254        error(format!(
255            "current view-log segments cover {expected_offset} bytes, but view_log declares {}",
256            object.object_length
257        ));
258    }
259}
laminar_core/checkpoint/checkpoint_manifest/view_log.rs

laminar_core/checkpoint/checkpoint_manifest/
view_log.rs
