nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

// SPDX-License-Identifier: BUSL-1.1

//! Bitemporal row-visibility predicate for system-time / valid-time queries.

/// Extract the system-time (`_ts_system`) value of a columnar row for
/// audit-log ordering. Rows missing the column (non-bitemporal, or the
/// column index is `None`) sort first via `i64::MIN`.
pub(super) fn row_system_time(
    row: &[nodedb_types::value::Value],
    ts_system_idx: Option<usize>,
) -> i64 {
    use nodedb_types::value::Value;
    match ts_system_idx.and_then(|i| row.get(i)) {
        Some(Value::Integer(i)) => *i,
        Some(Value::DateTime(dt)) | Some(Value::NaiveDateTime(dt)) => dt.micros / 1000,
        _ => i64::MIN,
    }
}

/// Bitemporal row-level visibility predicate.
///
/// - `system_as_of_ms`: when `Some`, any row whose `_ts_system` value
///   exceeds the cutoff is hidden (write happened after the query's
///   system-time horizon).
/// - `valid_at_ms`: when `Some`, the row's
///   `[_ts_valid_from, _ts_valid_until)` interval must contain this
///   point.
///
/// Rows from non-bitemporal collections have no `_ts_system` column and
/// all three indices are `None`; the function returns `true` unconditionally.
pub(super) fn bitemporal_row_visible(
    row: &[nodedb_types::value::Value],
    ts_system_idx: Option<usize>,
    ts_valid_from_idx: Option<usize>,
    ts_valid_until_idx: Option<usize>,
    system_as_of_ms: Option<i64>,
    valid_at_ms: Option<i64>,
) -> bool {
    use nodedb_types::value::Value;
    if let Some(cutoff) = system_as_of_ms
        && let Some(idx) = ts_system_idx
        && let Some(v) = row.get(idx)
    {
        let ts = match v {
            Value::Integer(i) => *i,
            Value::DateTime(dt) | Value::NaiveDateTime(dt) => dt.micros / 1000,
            _ => return false,
        };
        if ts > cutoff {
            return false;
        }
    }
    if let Some(point) = valid_at_ms {
        let vf = ts_valid_from_idx
            .and_then(|i| row.get(i))
            .and_then(|v| match v {
                Value::Integer(i) => Some(*i),
                Value::DateTime(dt) => Some(dt.micros / 1000),
                _ => None,
            })
            .unwrap_or(i64::MIN);
        let vu = ts_valid_until_idx
            .and_then(|i| row.get(i))
            .and_then(|v| match v {
                Value::Integer(i) => Some(*i),
                Value::DateTime(dt) => Some(dt.micros / 1000),
                _ => None,
            })
            .unwrap_or(i64::MAX);
        if point < vf || point >= vu {
            return false;
        }
    }
    true
}

#[cfg(test)]
mod tests {
    use super::*;
    use nodedb_types::value::Value;

    fn row(ts_system: i64, value: i64) -> Vec<Value> {
        // [_ts_system, _ts_valid_from, _ts_valid_until, value]
        vec![
            Value::Integer(ts_system),
            Value::Integer(0),
            Value::Integer(i64::MAX),
            Value::Integer(value),
        ]
    }

    #[test]
    fn row_system_time_reads_column() {
        assert_eq!(row_system_time(&row(123, 9), Some(0)), 123);
        // No bitemporal column resolved → sorts first.
        assert_eq!(row_system_time(&row(123, 9), None), i64::MIN);
    }

    #[test]
    fn all_versions_sort_orders_ascending_by_system_time() {
        // A single logical key updated three times: three distinct versions
        // distinguished by `_ts_system`. Under AS OF SYSTEM TIME NULL every
        // version is emitted ordered ascending by system time.
        let mut rows = [row(300, 3), row(100, 1), row(200, 2)];
        let ts_idx = Some(0);
        rows.sort_by_key(|r| row_system_time(r, ts_idx));
        let times: Vec<i64> = rows.iter().map(|r| row_system_time(r, ts_idx)).collect();
        assert_eq!(times, vec![100, 200, 300]);
        // The payload column tracks the version it belongs to.
        let values: Vec<i64> = rows
            .iter()
            .map(|r| match r.get(3) {
                Some(Value::Integer(v)) => *v,
                _ => -1,
            })
            .collect();
        assert_eq!(values, vec![1, 2, 3]);
    }

    #[test]
    fn all_versions_does_not_apply_system_cutoff() {
        // With no system-time cutoff (Current/AllVersions both pass None to the
        // predicate) every row is visible regardless of `_ts_system`.
        assert!(bitemporal_row_visible(
            &row(999, 1),
            Some(0),
            Some(1),
            Some(2),
            None,
            None
        ));
    }
}