nodedb_types/columnar/
schema.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Strict document and columnar schemas with shared operations trait.
4
5use serde::{Deserialize, Serialize};
6
7use super::column_def::ColumnDef;
8use crate::columnar::ColumnType;
9
10/// Shared schema operations (eliminates duplication between Strict and Columnar).
11pub trait SchemaOps {
12    fn columns(&self) -> &[ColumnDef];
13
14    fn column_index(&self, name: &str) -> Option<usize> {
15        self.columns().iter().position(|c| c.name == name)
16    }
17
18    fn column(&self, name: &str) -> Option<&ColumnDef> {
19        self.columns().iter().find(|c| c.name == name)
20    }
21
22    fn primary_key_columns(&self) -> Vec<&ColumnDef> {
23        self.columns().iter().filter(|c| c.primary_key).collect()
24    }
25
26    fn len(&self) -> usize {
27        self.columns().len()
28    }
29
30    fn is_empty(&self) -> bool {
31        self.columns().is_empty()
32    }
33}
34
35/// Schema for a strict document collection (Binary Tuple serialization).
36#[derive(
37    Debug,
38    Clone,
39    PartialEq,
40    Eq,
41    Serialize,
42    Deserialize,
43    zerompk::ToMessagePack,
44    zerompk::FromMessagePack,
45)]
46#[msgpack(map)]
47pub struct StrictSchema {
48    pub columns: Vec<ColumnDef>,
49    pub version: u32,
50    /// Columns that were removed via `ALTER DROP COLUMN`. Retained so the
51    /// reader can reconstruct the physical layout of tuples written before
52    /// the drop.
53    #[serde(default, skip_serializing_if = "Vec::is_empty")]
54    pub dropped_columns: Vec<DroppedColumn>,
55    /// When true, the tuple reserves fixed-Int64 slots 0/1/2 for
56    /// `__system_from_ms`, `__valid_from_ms`, `__valid_until_ms`. These
57    /// columns are prepended by `StrictSchema::new_bitemporal` and appear
58    /// in `columns` like any other field; the flag preserves the intent
59    /// across catalog round-trips.
60    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
61    #[msgpack(default)]
62    pub bitemporal: bool,
63}
64
65/// Tombstone for a column removed by `ALTER DROP COLUMN`.
66#[derive(
67    Debug,
68    Clone,
69    PartialEq,
70    Eq,
71    Serialize,
72    Deserialize,
73    zerompk::ToMessagePack,
74    zerompk::FromMessagePack,
75)]
76pub struct DroppedColumn {
77    /// The full column definition at time of drop.
78    pub def: ColumnDef,
79    /// The column's position in the column list before it was removed.
80    pub position: usize,
81    /// The schema version at which the column was dropped.
82    pub dropped_at_version: u32,
83}
84
85/// Schema for a columnar collection (compressed segment files).
86#[derive(
87    Debug,
88    Clone,
89    PartialEq,
90    Eq,
91    Serialize,
92    Deserialize,
93    zerompk::ToMessagePack,
94    zerompk::FromMessagePack,
95)]
96pub struct ColumnarSchema {
97    pub columns: Vec<ColumnDef>,
98    pub version: u32,
99}
100
101/// Reserved strict-tuple column names for bitemporal collections. Stored
102/// in fixed Int64 slots 0/1/2 so the decoder can extract them via a
103/// constant-offset jump.
104pub const BITEMPORAL_SYSTEM_FROM: &str = "__system_from_ms";
105pub const BITEMPORAL_VALID_FROM: &str = "__valid_from_ms";
106pub const BITEMPORAL_VALID_UNTIL: &str = "__valid_until_ms";
107
108/// All reserved bitemporal column names, in slot order (0, 1, 2).
109pub const BITEMPORAL_RESERVED_COLUMNS: [&str; 3] = [
110    BITEMPORAL_SYSTEM_FROM,
111    BITEMPORAL_VALID_FROM,
112    BITEMPORAL_VALID_UNTIL,
113];
114
115/// Canonical user-facing audit temporal column names, uniform across every
116/// engine. An `AS OF SYSTEM TIME NULL` (all-versions / audit-log) scan surfaces
117/// each row's real stored temporal data under these three synthetic columns —
118/// `_ts_system` (system-time of the version), `_ts_valid_from` /
119/// `_ts_valid_until` (the row's valid-time interval, raw i64 sentinels
120/// `i64::MIN` / `i64::MAX` when unbounded). Columnar/timeseries and both
121/// Document engines emit the identical triple.
122pub const TS_SYSTEM: &str = "_ts_system";
123pub const TS_VALID_FROM: &str = "_ts_valid_from";
124pub const TS_VALID_UNTIL: &str = "_ts_valid_until";
125
126/// Single source of truth for "is this a reserved internal strict-tuple
127/// bitemporal column that must be hidden from user-facing projections
128/// (`SELECT *`, JSON/msgpack decode boundaries, etc.)". Covers the
129/// strict-tuple triple (`__system_from_ms`, `__valid_from_ms`,
130/// `__valid_until_ms`) that `StrictSchema::new_bitemporal` prepends into the
131/// physical schema. Callers should replace ad-hoc per-site string checks with
132/// this predicate. (The columnar/timeseries `_ts_*` columns are handled by
133/// those engines' own read paths and, for audit queries, are the intended
134/// output — they are deliberately NOT hidden here.)
135pub fn is_reserved_bitemporal_column(name: &str) -> bool {
136    BITEMPORAL_RESERVED_COLUMNS.contains(&name)
137}
138
139/// Schema validation errors.
140#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
141#[non_exhaustive]
142pub enum SchemaError {
143    #[error("schema must have at least one column")]
144    Empty,
145    #[error("duplicate column name: '{0}'")]
146    DuplicateColumn(String),
147    #[error("VECTOR dimension must be positive, got 0 for column '{0}'")]
148    ZeroVectorDim(String),
149    #[error("primary key column '{0}' must be NOT NULL")]
150    NullablePrimaryKey(String),
151    #[error("column name '{0}' is reserved for bitemporal collections")]
152    ReservedColumnName(String),
153}
154
155fn validate_columns(columns: &[ColumnDef]) -> Result<(), SchemaError> {
156    if columns.is_empty() {
157        return Err(SchemaError::Empty);
158    }
159    let mut seen = std::collections::HashSet::with_capacity(columns.len());
160    for col in columns {
161        if !seen.insert(&col.name) {
162            return Err(SchemaError::DuplicateColumn(col.name.clone()));
163        }
164        if col.primary_key && col.nullable {
165            return Err(SchemaError::NullablePrimaryKey(col.name.clone()));
166        }
167        if let ColumnType::Vector(0) = col.column_type {
168            return Err(SchemaError::ZeroVectorDim(col.name.clone()));
169        }
170    }
171    Ok(())
172}
173
174impl SchemaOps for StrictSchema {
175    fn columns(&self) -> &[ColumnDef] {
176        &self.columns
177    }
178}
179
180impl SchemaOps for ColumnarSchema {
181    fn columns(&self) -> &[ColumnDef] {
182        &self.columns
183    }
184}
185
186impl StrictSchema {
187    pub fn new(columns: Vec<ColumnDef>) -> Result<Self, SchemaError> {
188        for col in &columns {
189            if BITEMPORAL_RESERVED_COLUMNS.contains(&col.name.as_str()) {
190                return Err(SchemaError::ReservedColumnName(col.name.clone()));
191            }
192        }
193        validate_columns(&columns)?;
194        Ok(Self {
195            columns,
196            version: 1,
197            dropped_columns: Vec::new(),
198            bitemporal: false,
199        })
200    }
201
202    /// Build a schema for a bitemporal strict collection. Prepends three
203    /// reserved Int64 columns (`__system_from_ms`, `__valid_from_ms`,
204    /// `__valid_until_ms`) at positions 0/1/2 so the tuple decoder can
205    /// extract them via fixed-offset jump. User columns are rejected if
206    /// any collides with a reserved name.
207    pub fn new_bitemporal(user_columns: Vec<ColumnDef>) -> Result<Self, SchemaError> {
208        for col in &user_columns {
209            if BITEMPORAL_RESERVED_COLUMNS.contains(&col.name.as_str()) {
210                return Err(SchemaError::ReservedColumnName(col.name.clone()));
211            }
212        }
213        let mut columns = Vec::with_capacity(3 + user_columns.len());
214        columns.push(ColumnDef::required(
215            BITEMPORAL_SYSTEM_FROM,
216            ColumnType::Int64,
217        ));
218        columns.push(ColumnDef::required(
219            BITEMPORAL_VALID_FROM,
220            ColumnType::Int64,
221        ));
222        columns.push(ColumnDef::required(
223            BITEMPORAL_VALID_UNTIL,
224            ColumnType::Int64,
225        ));
226        columns.extend(user_columns);
227        validate_columns(&columns)?;
228        Ok(Self {
229            columns,
230            version: 1,
231            dropped_columns: Vec::new(),
232            bitemporal: true,
233        })
234    }
235
236    /// Count of variable-length columns (determines offset table size).
237    pub fn variable_column_count(&self) -> usize {
238        self.columns
239            .iter()
240            .filter(|c| c.column_type.is_variable_length())
241            .count()
242    }
243
244    /// Total fixed-field byte size (for Binary Tuple layout computation).
245    pub fn fixed_fields_size(&self) -> usize {
246        self.columns
247            .iter()
248            .filter_map(|c| c.column_type.fixed_size())
249            .sum()
250    }
251
252    /// Null bitmap size in bytes.
253    pub fn null_bitmap_size(&self) -> usize {
254        self.columns.len().div_ceil(8)
255    }
256
257    /// Build a sub-schema matching the physical layout of tuples written at
258    /// the given version. Columns added after `version` are excluded;
259    /// columns dropped after `version` are re-inserted at their original
260    /// positions.
261    pub fn schema_for_version(&self, version: u32) -> StrictSchema {
262        // Start with live columns that existed at this version.
263        let mut cols: Vec<ColumnDef> = self
264            .columns
265            .iter()
266            .filter(|c| c.added_at_version <= version)
267            .cloned()
268            .collect();
269
270        // Re-insert dropped columns that were still alive at this version,
271        // sorted by position (ascending) so inserts don't shift later indices.
272        let mut to_reinsert: Vec<&DroppedColumn> = self
273            .dropped_columns
274            .iter()
275            .filter(|dc| dc.def.added_at_version <= version && dc.dropped_at_version > version)
276            .collect();
277        to_reinsert.sort_by_key(|dc| dc.position);
278        for dc in to_reinsert {
279            let pos = dc.position.min(cols.len());
280            cols.insert(pos, dc.def.clone());
281        }
282
283        StrictSchema {
284            version,
285            columns: cols,
286            dropped_columns: Vec::new(),
287            bitemporal: self.bitemporal,
288        }
289    }
290
291    /// Parse a SQL default literal (e.g. `'n/a'`, `0`, `true`) into a `Value`.
292    ///
293    /// Covers the common cases produced by `ALTER ADD COLUMN ... DEFAULT ...`.
294    /// Returns `Value::Null` for expressions that cannot be trivially evaluated
295    /// at read time (functions, sub-queries, etc.).
296    pub fn parse_default_literal(expr: &str) -> crate::value::Value {
297        use crate::value::Value;
298
299        let trimmed = expr.trim();
300
301        // String literals: 'foo'
302        if trimmed.starts_with('\'') && trimmed.ends_with('\'') && trimmed.len() >= 2 {
303            return Value::String(trimmed[1..trimmed.len() - 1].replace("''", "'"));
304        }
305
306        // Boolean
307        match trimmed.to_uppercase().as_str() {
308            "TRUE" => return Value::Bool(true),
309            "FALSE" => return Value::Bool(false),
310            "NULL" => return Value::Null,
311            _ => {}
312        }
313
314        // Integer
315        if let Ok(i) = trimmed.parse::<i64>() {
316            return Value::Integer(i);
317        }
318
319        // Float
320        if let Ok(f) = trimmed.parse::<f64>() {
321            return Value::Float(f);
322        }
323
324        Value::Null
325    }
326}
327
328impl ColumnarSchema {
329    pub fn new(columns: Vec<ColumnDef>) -> Result<Self, SchemaError> {
330        validate_columns(&columns)?;
331        Ok(Self {
332            columns,
333            version: 1,
334        })
335    }
336
337    /// Whether this schema has the reserved `_ts_system` bitemporal column.
338    ///
339    /// Detected by column name rather than a separate flag to keep the
340    /// on-disk manifest format unchanged; `_ts_system` is only inserted
341    /// by `prepend_bitemporal_columns` on the write path, so its
342    /// presence is a reliable bitemporal signal.
343    pub fn is_bitemporal(&self) -> bool {
344        self.columns.iter().any(|c| c.name == "_ts_system")
345    }
346
347    /// Position of the `_ts_system` column, or `None` for non-bitemporal.
348    pub fn ts_system_idx(&self) -> Option<usize> {
349        self.columns.iter().position(|c| c.name == "_ts_system")
350    }
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356    use crate::columnar::ColumnType;
357
358    #[test]
359    fn strict_schema_validation() {
360        let schema = StrictSchema::new(vec![
361            ColumnDef::required("id", ColumnType::Int64).with_primary_key(),
362            ColumnDef::nullable("name", ColumnType::String),
363        ]);
364        assert!(schema.is_ok());
365        assert!(StrictSchema::new(vec![]).is_err());
366    }
367
368    #[test]
369    fn schema_ops_trait() {
370        let schema = StrictSchema::new(vec![
371            ColumnDef::required("id", ColumnType::Int64).with_primary_key(),
372            ColumnDef::nullable("name", ColumnType::String),
373            ColumnDef::nullable(
374                "balance",
375                ColumnType::Decimal {
376                    precision: 18,
377                    scale: 4,
378                },
379            ),
380        ])
381        .unwrap();
382        assert_eq!(schema.len(), 3);
383        assert_eq!(schema.column_index("balance"), Some(2));
384        assert!(schema.column("nonexistent").is_none());
385        assert_eq!(schema.primary_key_columns().len(), 1);
386    }
387
388    #[test]
389    fn strict_layout_helpers() {
390        let schema = StrictSchema::new(vec![
391            ColumnDef::required("id", ColumnType::Int64).with_primary_key(),
392            ColumnDef::nullable("name", ColumnType::String),
393            ColumnDef::nullable(
394                "balance",
395                ColumnType::Decimal {
396                    precision: 18,
397                    scale: 4,
398                },
399            ),
400            ColumnDef::nullable("bio", ColumnType::String),
401        ])
402        .unwrap();
403        assert_eq!(schema.null_bitmap_size(), 1);
404        assert_eq!(schema.fixed_fields_size(), 8 + 16);
405        assert_eq!(schema.variable_column_count(), 2);
406    }
407
408    #[test]
409    fn columnar_schema_validation() {
410        let schema = ColumnarSchema::new(vec![
411            ColumnDef::required("time", ColumnType::Timestamp),
412            ColumnDef::nullable("cpu", ColumnType::Float64),
413        ]);
414        assert!(schema.is_ok());
415        assert_eq!(schema.unwrap().len(), 2);
416    }
417
418    #[test]
419    fn nullable_pk_rejected() {
420        let cols = vec![ColumnDef {
421            name: "id".into(),
422            column_type: ColumnType::Int64,
423            nullable: true,
424            default: None,
425            primary_key: true,
426            modifiers: Vec::new(),
427            generated_expr: None,
428            generated_deps: Vec::new(),
429            added_at_version: 1,
430        }];
431        assert!(matches!(
432            StrictSchema::new(cols),
433            Err(SchemaError::NullablePrimaryKey(_))
434        ));
435    }
436
437    #[test]
438    fn zero_vector_dim_rejected() {
439        let cols = vec![ColumnDef::required("emb", ColumnType::Vector(0))];
440        assert!(matches!(
441            StrictSchema::new(cols),
442            Err(SchemaError::ZeroVectorDim(_))
443        ));
444    }
445}
nodedb_types/columnar/schema.rs

nodedb_types/columnar/
schema.rs
