nodedb_physical/physical_plan/
tiering.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! AUTO_TIER routing of one timeseries scan across retention tiers.
4//!
5//! A tiered retention policy keeps raw rows for `raw_retain_ms`, then
6//! downsampled buckets in one continuous aggregate per tier. A scan carries
7//! the policy's tier layout; the core that owns the collection splits the
8//! query range into segments and reads each segment from its tier.
9
10/// The tier layout of an AUTO_TIER retention policy.
11#[derive(
12    Debug,
13    Clone,
14    PartialEq,
15    serde::Serialize,
16    serde::Deserialize,
17    zerompk::ToMessagePack,
18    zerompk::FromMessagePack,
19)]
20pub struct TieringPlan {
21    /// How long raw rows are kept, in milliseconds. 0 = forever.
22    pub raw_retain_ms: u64,
23    /// Downsample tiers, finest first.
24    pub tiers: Vec<TierRead>,
25}
26
27/// One downsample tier of a [`TieringPlan`].
28#[derive(
29    Debug,
30    Clone,
31    PartialEq,
32    serde::Serialize,
33    serde::Deserialize,
34    zerompk::ToMessagePack,
35    zerompk::FromMessagePack,
36)]
37pub struct TierRead {
38    /// The policy's tier number (1 = first downsample tier).
39    pub tier_index: u32,
40    /// The continuous aggregate that holds this tier's buckets.
41    pub aggregate: String,
42    /// Bucket width in milliseconds.
43    pub resolution_ms: i64,
44    /// How long this tier keeps buckets, in milliseconds. 0 = forever.
45    pub retain_ms: u64,
46}
47
48/// Which source serves one segment of a tiered scan.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum TierSource {
51    /// Raw rows of the collection.
52    Raw,
53    /// Buckets of `TieringPlan::tiers[index]`.
54    Tier(usize),
55}
56
57/// One time segment of a tiered scan: `[start, end)` read from `source`.
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub struct TierSegment {
60    pub source: TierSource,
61    pub start: i64,
62    pub end: i64,
63}
64
65/// Why a tier layout cannot split a range.
66#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
67pub enum TieringError {
68    #[error(
69        "tier {tier_index} resolution {resolution_ms}ms is not a positive multiple of the \
70         finer tier's {finer_ms}ms; a coarser bucket would straddle the finer tier's edge"
71    )]
72    UnalignedResolution {
73        tier_index: u32,
74        resolution_ms: i64,
75        finer_ms: i64,
76    },
77}
78
79impl TieringPlan {
80    /// Split `query` (an inclusive `(min, max)` millisecond range) into
81    /// segments, newest first.
82    ///
83    /// The edge between two sources is the retention boundary rounded up to
84    /// the coarser tier's resolution, so every bucket a tier serves lies
85    /// wholly before the edge and no row is read from two sources.
86    pub fn segments(
87        &self,
88        query: (i64, i64),
89        now_ms: i64,
90    ) -> Result<Vec<TierSegment>, TieringError> {
91        let (query_start, query_end) = query;
92        let mut segments = Vec::new();
93        // `upper` is the exclusive end of the source being placed.
94        let mut upper = query_end.saturating_add(1);
95        let mut source = TierSource::Raw;
96        let mut boundary = now_ms;
97        let mut retain = self.raw_retain_ms;
98        let mut finer_ms = 1i64;
99        for (index, tier) in self.tiers.iter().enumerate() {
100            if retain == 0 {
101                break;
102            }
103            if tier.resolution_ms <= 0 || tier.resolution_ms % finer_ms != 0 {
104                return Err(TieringError::UnalignedResolution {
105                    tier_index: tier.tier_index,
106                    resolution_ms: tier.resolution_ms,
107                    finer_ms,
108                });
109            }
110            boundary = boundary.saturating_sub(retain as i64);
111            let edge = align_up(boundary, tier.resolution_ms);
112            push_segment(&mut segments, source, edge.max(query_start), upper);
113            upper = upper.min(edge);
114            source = TierSource::Tier(index);
115            retain = tier.retain_ms;
116            finer_ms = tier.resolution_ms;
117        }
118        push_segment(&mut segments, source, query_start, upper);
119        Ok(segments)
120    }
121}
122
123/// Append `[start, end)` when it is non-empty.
124fn push_segment(out: &mut Vec<TierSegment>, source: TierSource, start: i64, end: i64) {
125    if start < end {
126        out.push(TierSegment { source, start, end });
127    }
128}
129
130/// `ms` rounded up to a multiple of `step` (`step > 0`).
131fn align_up(ms: i64, step: i64) -> i64 {
132    let floor = ms.div_euclid(step).saturating_mul(step);
133    if floor == ms {
134        floor
135    } else {
136        floor.saturating_add(step)
137    }
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143
144    const DAY: i64 = 86_400_000;
145
146    fn plan() -> TieringPlan {
147        TieringPlan {
148            raw_retain_ms: 7 * DAY as u64,
149            tiers: vec![
150                TierRead {
151                    tier_index: 1,
152                    aggregate: "_policy_p_tier1".into(),
153                    resolution_ms: 60_000,
154                    retain_ms: 90 * DAY as u64,
155                },
156                TierRead {
157                    tier_index: 2,
158                    aggregate: "_policy_p_tier2".into(),
159                    resolution_ms: 3_600_000,
160                    retain_ms: 0,
161                },
162            ],
163        }
164    }
165
166    #[test]
167    fn recent_range_reads_raw_only() {
168        let now = 100_000 * DAY;
169        let segments = plan().segments((now - 3_600_000, now), now).expect("split");
170        assert_eq!(segments.len(), 1);
171        assert_eq!(segments[0].source, TierSource::Raw);
172        assert_eq!(segments[0].end, now + 1);
173    }
174
175    #[test]
176    fn range_spanning_tiers_splits_on_aligned_edges() {
177        let now = 100_000 * DAY + 12_345;
178        let segments = plan().segments((now - 200 * DAY, now), now).expect("split");
179        assert_eq!(segments.len(), 3);
180        assert_eq!(segments[0].source, TierSource::Raw);
181        assert_eq!(segments[1].source, TierSource::Tier(0));
182        assert_eq!(segments[2].source, TierSource::Tier(1));
183        // Edges are aligned to the coarser source's resolution and contiguous.
184        assert_eq!(segments[0].start, segments[1].end);
185        assert_eq!(segments[1].start, segments[2].end);
186        assert_eq!(segments[0].start % 60_000, 0);
187        assert_eq!(segments[1].start % 3_600_000, 0);
188    }
189
190    #[test]
191    fn forever_raw_retention_never_reads_a_tier() {
192        let mut plan = plan();
193        plan.raw_retain_ms = 0;
194        let now = 100_000 * DAY;
195        let segments = plan.segments((0, now), now).expect("split");
196        assert_eq!(segments.len(), 1);
197        assert_eq!(segments[0].source, TierSource::Raw);
198    }
199
200    #[test]
201    fn unaligned_resolution_is_refused() {
202        let mut plan = plan();
203        plan.tiers[1].resolution_ms = 90_000;
204        let now = 100_000 * DAY;
205        assert!(matches!(
206            plan.segments((0, now), now),
207            Err(TieringError::UnalignedResolution { tier_index: 2, .. })
208        ));
209    }
210}
nodedb_physical/physical_plan/tiering.rs

nodedb_physical/physical_plan/
tiering.rs
