nodedb_types/timeseries/
ingest.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Ingest types, time range, and symbol dictionary.
4
5use std::collections::HashMap;
6
7use serde::{Deserialize, Serialize};
8
9/// A single metric sample (timestamp + scalar value).
10#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
11pub struct MetricSample {
12    pub timestamp_ms: i64,
13    pub value: f64,
14}
15
16/// A single log entry (timestamp + arbitrary bytes).
17#[derive(Debug, Clone, Serialize, Deserialize)]
18pub struct LogEntry {
19    pub timestamp_ms: i64,
20    pub data: Vec<u8>,
21}
22
23/// Result of an ingest operation.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25#[non_exhaustive]
26pub enum IngestResult {
27    /// Write accepted, memtable healthy.
28    Ok,
29    /// Write accepted, but memtable should be flushed (memory pressure).
30    FlushNeeded,
31    /// Write rejected — memory budget exhausted and cannot evict further.
32    Rejected,
33}
34
35impl IngestResult {
36    pub fn is_flush_needed(&self) -> bool {
37        matches!(self, Self::FlushNeeded)
38    }
39
40    pub fn is_rejected(&self) -> bool {
41        matches!(self, Self::Rejected)
42    }
43}
44
45/// Time range for queries (inclusive on both ends).
46#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
47pub struct TimeRange {
48    pub start_ms: i64,
49    pub end_ms: i64,
50}
51
52impl TimeRange {
53    pub fn new(start_ms: i64, end_ms: i64) -> Self {
54        Self { start_ms, end_ms }
55    }
56
57    pub fn contains(&self, ts: i64) -> bool {
58        ts >= self.start_ms && ts <= self.end_ms
59    }
60
61    /// Whether two ranges overlap.
62    pub fn overlaps(&self, other: &TimeRange) -> bool {
63        self.start_ms <= other.end_ms && other.start_ms <= self.end_ms
64    }
65}
66
67/// Bidirectional symbol dictionary for tag value interning.
68///
69/// Tag columns store 4-byte u32 IDs instead of full strings. Shared by
70/// Origin columnar segments, Lite native segments, and WASM segments.
71#[derive(
72    Debug,
73    Clone,
74    Default,
75    PartialEq,
76    Serialize,
77    Deserialize,
78    zerompk::ToMessagePack,
79    zerompk::FromMessagePack,
80)]
81pub struct SymbolDictionary {
82    /// String → symbol ID.
83    forward: HashMap<String, u32>,
84    /// Symbol ID → string (index = id).
85    reverse: Vec<String>,
86}
87
88impl SymbolDictionary {
89    pub fn new() -> Self {
90        Self::default()
91    }
92
93    /// Resolve a string to its symbol ID, inserting if new.
94    ///
95    /// Returns `None` if the dictionary has reached `max_cardinality`.
96    pub fn resolve(&mut self, value: &str, max_cardinality: u32) -> Option<u32> {
97        if let Some(&id) = self.forward.get(value) {
98            return Some(id);
99        }
100        if self.reverse.len() as u32 >= max_cardinality {
101            return None;
102        }
103        let id = self.reverse.len() as u32;
104        self.forward.insert(value.to_string(), id);
105        self.reverse.push(value.to_string());
106        Some(id)
107    }
108
109    /// Look up a string by symbol ID.
110    pub fn get(&self, id: u32) -> Option<&str> {
111        self.reverse.get(id as usize).map(|s| s.as_str())
112    }
113
114    /// Look up a symbol ID by string.
115    pub fn get_id(&self, value: &str) -> Option<u32> {
116        self.forward.get(value).copied()
117    }
118
119    /// Number of symbols.
120    pub fn len(&self) -> usize {
121        self.reverse.len()
122    }
123
124    pub fn is_empty(&self) -> bool {
125        self.reverse.is_empty()
126    }
127
128    /// Merge another dictionary into this one.
129    ///
130    /// Returns a remap table: `old_id → new_id` for the source dictionary.
131    pub fn merge(&mut self, other: &SymbolDictionary, max_cardinality: u32) -> Vec<u32> {
132        let mut remap = Vec::with_capacity(other.reverse.len());
133        for symbol in &other.reverse {
134            match self.resolve(symbol, max_cardinality) {
135                Some(new_id) => remap.push(new_id),
136                None => remap.push(u32::MAX),
137            }
138        }
139        remap
140    }
141}
142
143#[cfg(test)]
144mod tests {
145    use super::*;
146
147    #[test]
148    fn symbol_dictionary_basic() {
149        let mut dict = SymbolDictionary::new();
150        assert_eq!(dict.resolve("us-east-1", 100_000), Some(0));
151        assert_eq!(dict.resolve("us-west-2", 100_000), Some(1));
152        assert_eq!(dict.resolve("us-east-1", 100_000), Some(0));
153        assert_eq!(dict.len(), 2);
154        assert_eq!(dict.get(0), Some("us-east-1"));
155        assert_eq!(dict.get(1), Some("us-west-2"));
156        assert_eq!(dict.get_id("us-east-1"), Some(0));
157    }
158
159    #[test]
160    fn symbol_dictionary_cardinality_breaker() {
161        let mut dict = SymbolDictionary::new();
162        for i in 0..100 {
163            assert!(dict.resolve(&format!("val-{i}"), 100).is_some());
164        }
165        assert!(dict.resolve("one-too-many", 100).is_none());
166        assert_eq!(dict.len(), 100);
167    }
168
169    #[test]
170    fn symbol_dictionary_merge() {
171        let mut d1 = SymbolDictionary::new();
172        d1.resolve("a", 1000);
173        d1.resolve("b", 1000);
174
175        let mut d2 = SymbolDictionary::new();
176        d2.resolve("b", 1000);
177        d2.resolve("c", 1000);
178
179        let remap = d1.merge(&d2, 1000);
180        assert_eq!(d1.len(), 3);
181        assert_eq!(remap[0], d1.get_id("b").unwrap());
182        assert_eq!(remap[1], d1.get_id("c").unwrap());
183    }
184
185    #[test]
186    fn time_range_overlap() {
187        let r1 = TimeRange::new(100, 200);
188        let r2 = TimeRange::new(150, 250);
189        let r3 = TimeRange::new(300, 400);
190        assert!(r1.overlaps(&r2));
191        assert!(!r1.overlaps(&r3));
192    }
193}
nodedb_types/timeseries/ingest.rs

nodedb_types/timeseries/
ingest.rs
