feat(nodedb): implement bitemporal read and write in columnar and timeseries handlers

Columnar write: detect bitemporal collections and prepend the three
reserved columns (`_ts_system`, `_ts_valid_from`, `_ts_valid_until`) to
the schema on first write. `_ts_system` is always stamped with the
current wall-clock time; `_ts_valid_from` / `_ts_valid_until` are taken
from the client payload or default to the open interval `[i64::MIN,
i64::MAX)`. Clients cannot supply `_ts_system`.

Columnar read: resolve reserved column positions once per scan, then
apply `bitemporal_row_visible` to each memtable row. The predicate
enforces the system-time cutoff and valid-time point filter before any
user-supplied predicates run.

Timeseries scan: translate `system_as_of_ms` and `valid_at_ms` into
`ScanFilter` entries (`_ts_system <= cutoff`, `_ts_valid_from <= point`,
`_ts_valid_until > point`) so the segment reader's block-skip
infrastructure automatically prunes blocks that cannot satisfy the
bitemporal predicates.

Dispatch: thread `system_as_of_ms` / `valid_at_ms` from the physical
plan into both handler param structs.

Author: farhan-syah

nodedb/src/data/executor/dispatch/other.rs

@@ -462,6 +462,8 @@ impl CoreLoop {
                 filters,
                 rls_filters,
                 sort_keys,
+                system_as_of_ms,
+                valid_at_ms,
             }) => self.execute_columnar_scan(
                 task,
                 super::super::handlers::columnar_read::ColumnarScanParams {
@@ -471,6 +473,8 @@ impl CoreLoop {
                     filters,
                     rls_filters,
                     sort_keys,
+                    system_as_of_ms: *system_as_of_ms,
+                    valid_at_ms: *valid_at_ms,
                 },
             ),
 
@@ -510,6 +514,8 @@ impl CoreLoop {
                 aggregates,
                 gap_fill,
                 computed_columns,
+                system_as_of_ms,
+                valid_at_ms,
                 ..
             }) => self.execute_timeseries_scan(
                 super::super::handlers::timeseries::TimeseriesScanParams {
@@ -524,6 +530,8 @@ impl CoreLoop {
                     aggregates,
                     gap_fill,
                     computed_columns,
+                    system_as_of_ms: *system_as_of_ms,
+                    valid_at_ms: *valid_at_ms,
                 },
             ),
 

nodedb/src/data/executor/handlers/columnar_read.rs

@@ -23,6 +23,13 @@ pub(in crate::data::executor) struct ColumnarScanParams<'a> {
     #[allow(dead_code)]
     pub rls_filters: &'a [u8],
     pub sort_keys: &'a [(String, bool)],
+    /// Bitemporal system-time cutoff: drop rows with `_ts_system > cutoff`.
+    /// `None` is a current-state read (no filter applied).
+    pub system_as_of_ms: Option<i64>,
+    /// Bitemporal valid-time point: drop rows whose
+    /// `[_ts_valid_from, _ts_valid_until)` interval does not contain this
+    /// point. `None` skips valid-time filtering entirely.
+    pub valid_at_ms: Option<i64>,
 }
 
 impl CoreLoop {
@@ -39,6 +46,8 @@ impl CoreLoop {
             filters,
             rls_filters: _,
             sort_keys,
+            system_as_of_ms,
+            valid_at_ms,
         } = params;
         let limit = if limit == 0 { 1000 } else { limit };
 
@@ -86,7 +95,33 @@ impl CoreLoop {
         } else {
             usize::MAX
         };
+        // Resolve hidden bitemporal column positions once; `None` means
+        // the collection is not bitemporal, so the per-row filter is a
+        // no-op regardless of `system_as_of_ms` / `valid_at_ms` values.
+        let ts_system_idx = schema
+            .columns
+            .iter()
+            .position(|c| c.name == "_ts_system");
+        let ts_valid_from_idx = schema
+            .columns
+            .iter()
+            .position(|c| c.name == "_ts_valid_from");
+        let ts_valid_until_idx = schema
+            .columns
+            .iter()
+            .position(|c| c.name == "_ts_valid_until");
+
         for row in engine.scan_memtable_rows().take(scan_budget) {
+            if !bitemporal_row_visible(
+                &row,
+                ts_system_idx,
+                ts_valid_from_idx,
+                ts_valid_until_idx,
+                system_as_of_ms,
+                valid_at_ms,
+            ) {
+                continue;
+            }
             if !filter_predicates.is_empty()
                 && !row_matches_filters(&row, schema, &filter_predicates)
             {
@@ -345,3 +380,60 @@ pub(in crate::data::executor) fn emit_column_value(
         }
     }
 }
+
+/// 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.
+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) => 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
+}

nodedb/src/data/executor/handlers/columnar_write.rs

@@ -63,9 +63,16 @@ impl CoreLoop {
         }
 
         let engine_key = (task.request.tenant_id, collection.to_string());
+        let tid = task.request.tenant_id.as_u32();
+        let bitemporal = self.is_bitemporal(tid, collection);
         // Ensure MutationEngine exists (auto-create on first write).
         if !self.columnar_engines.contains_key(&engine_key) {
-            let schema = infer_schema_from_value(&ndb_rows[0]);
+            let base_schema = infer_schema_from_value(&ndb_rows[0]);
+            let schema = if bitemporal {
+                prepend_bitemporal_columns(base_schema)
+            } else {
+                base_schema
+            };
             let engine = MutationEngine::new(collection.to_string(), schema);
             self.columnar_engines.insert(engine_key.clone(), engine);
         }
@@ -89,11 +96,33 @@ impl CoreLoop {
                 _ => continue,
             };
 
-            // Build Value slice in schema order.
+            // Build Value slice in schema order. For bitemporal
+            // collections, the three reserved columns are auto-populated
+            // when absent from the user payload: `_ts_system` is always
+            // clamped to the current wall-clock time (clients cannot
+            // forge system time), `_ts_valid_from` / `_ts_valid_until`
+            // default to the open interval `[i64::MIN, i64::MAX)` if
+            // missing.
+            let sys_now = if bitemporal {
+                self.bitemporal_now_ms()
+            } else {
+                0
+            };
             let values: Vec<Value> = schema
                 .columns
                 .iter()
-                .map(|col| ndb_field_to_value(obj.get(&col.name), &col.column_type))
+                .map(|col| match col.name.as_str() {
+                    "_ts_system" if bitemporal => Value::Integer(sys_now),
+                    "_ts_valid_from" if bitemporal => match obj.get("_ts_valid_from") {
+                        Some(Value::Integer(i)) => Value::Integer(*i),
+                        _ => Value::Integer(i64::MIN),
+                    },
+                    "_ts_valid_until" if bitemporal => match obj.get("_ts_valid_until") {
+                        Some(Value::Integer(i)) => Value::Integer(*i),
+                        _ => Value::Integer(i64::MAX),
+                    },
+                    _ => ndb_field_to_value(obj.get(&col.name), &col.column_type),
+                })
                 .collect();
 
             // Resolve the actual row to write (merged for ON CONFLICT DO
@@ -347,8 +376,19 @@ impl CoreLoop {
         obj: &serde_json::Map<String, serde_json::Value>,
     ) {
         let engine_key = (crate::types::TenantId::new(tid), collection.to_string());
+        let bitemporal = self.is_bitemporal(tid, collection);
+        let sys_now = if bitemporal {
+            self.bitemporal_now_ms()
+        } else {
+            0
+        };
         if !self.columnar_engines.contains_key(&engine_key) {
-            let schema = infer_schema_from_json(&serde_json::Value::Object(obj.clone()));
+            let base_schema = infer_schema_from_json(&serde_json::Value::Object(obj.clone()));
+            let schema = if bitemporal {
+                prepend_bitemporal_columns(base_schema)
+            } else {
+                base_schema
+            };
             let engine = MutationEngine::new(collection.to_string(), schema);
             self.columnar_engines.insert(engine_key.clone(), engine);
         }
@@ -365,7 +405,18 @@ impl CoreLoop {
         let values: Vec<Value> = schema
             .columns
             .iter()
-            .map(|col| ndb_field_to_value(ndb_obj.get(&col.name), &col.column_type))
+            .map(|col| match col.name.as_str() {
+                "_ts_system" if bitemporal => Value::Integer(sys_now),
+                "_ts_valid_from" if bitemporal => match ndb_obj.get("_ts_valid_from") {
+                    Some(Value::Integer(i)) => Value::Integer(*i),
+                    _ => Value::Integer(i64::MIN),
+                },
+                "_ts_valid_until" if bitemporal => match ndb_obj.get("_ts_valid_until") {
+                    Some(Value::Integer(i)) => Value::Integer(*i),
+                    _ => Value::Integer(i64::MAX),
+                },
+                _ => ndb_field_to_value(ndb_obj.get(&col.name), &col.column_type),
+            })
             .collect();
 
         let _ = engine.insert(&values);
@@ -497,6 +548,20 @@ fn infer_schema_from_value(row: &Value) -> ColumnarSchema {
     ColumnarSchema::new(columns).expect("inferred schema must be valid")
 }
 
+/// Prepend the three reserved bitemporal columns (`_ts_system`,
+/// `_ts_valid_from`, `_ts_valid_until`) at positions 0/1/2 of a columnar
+/// schema. All three are required Int64; `_ts_system` is engine-stamped
+/// on every write, the valid-time pair is client-provided (or defaults
+/// to the open interval).
+fn prepend_bitemporal_columns(base: ColumnarSchema) -> ColumnarSchema {
+    let mut cols = Vec::with_capacity(3 + base.columns.len());
+    cols.push(ColumnDef::required("_ts_system", ColumnType::Int64));
+    cols.push(ColumnDef::required("_ts_valid_from", ColumnType::Int64));
+    cols.push(ColumnDef::required("_ts_valid_until", ColumnType::Int64));
+    cols.extend(base.columns);
+    ColumnarSchema::new(cols).expect("bitemporal columnar schema must be valid")
+}
+
 /// Infer a columnar schema from a JSON object — used by the spatial insert path.
 pub(super) fn infer_schema_from_json(row: &serde_json::Value) -> ColumnarSchema {
     let ndb: Value = row.clone().into();

nodedb/src/data/executor/handlers/timeseries/mod.rs

@@ -27,6 +27,11 @@ pub(in crate::data::executor) struct TimeseriesScanParams<'a> {
     pub gap_fill: &'a str,
     /// Serialized computed columns for scalar projection expressions.
     pub computed_columns: &'a [u8],
+    /// Bitemporal system-time cutoff: rows whose `_ts_system` exceeds
+    /// this cutoff are hidden. `None` on non-bitemporal collections.
+    pub system_as_of_ms: Option<i64>,
+    /// Bitemporal valid-time point. `None` skips valid-time filtering.
+    pub valid_at_ms: Option<i64>,
 }
 
 impl CoreLoop {
@@ -52,16 +57,48 @@ impl CoreLoop {
             aggregates,
             gap_fill,
             computed_columns,
+            system_as_of_ms,
+            valid_at_ms,
         } = params;
 
         // Lazy-load partition registry from disk if not yet loaded.
         self.ensure_ts_registry(tid, collection);
 
-        let filter_predicates: Vec<crate::bridge::scan_filter::ScanFilter> = if filters.is_empty() {
-            Vec::new()
-        } else {
-            zerompk::from_msgpack(filters).unwrap_or_default()
-        };
+        let mut filter_predicates: Vec<crate::bridge::scan_filter::ScanFilter> =
+            if filters.is_empty() {
+                Vec::new()
+            } else {
+                zerompk::from_msgpack(filters).unwrap_or_default()
+            };
+        // Bitemporal cutoffs: translate to column-level predicates on
+        // `_ts_system` / `_ts_valid_from` / `_ts_valid_until`. The
+        // segment reader's block-skip infrastructure applies these
+        // against per-block min/max automatically.
+        if let Some(cutoff) = system_as_of_ms {
+            filter_predicates.push(crate::bridge::scan_filter::ScanFilter {
+                field: "_ts_system".into(),
+                op: crate::bridge::scan_filter::FilterOp::Lte,
+                value: nodedb_types::Value::Integer(cutoff),
+                clauses: Vec::new(),
+                expr: None,
+            });
+        }
+        if let Some(point) = valid_at_ms {
+            filter_predicates.push(crate::bridge::scan_filter::ScanFilter {
+                field: "_ts_valid_from".into(),
+                op: crate::bridge::scan_filter::FilterOp::Lte,
+                value: nodedb_types::Value::Integer(point),
+                clauses: Vec::new(),
+                expr: None,
+            });
+            filter_predicates.push(crate::bridge::scan_filter::ScanFilter {
+                field: "_ts_valid_until".into(),
+                op: crate::bridge::scan_filter::FilterOp::Gt,
+                value: nodedb_types::Value::Integer(point),
+                clauses: Vec::new(),
+                expr: None,
+            });
+        }
         let has_filters = !filter_predicates.is_empty();
         let is_aggregate = !aggregates.is_empty();
         let has_time_range = time_range.0 > 0 || time_range.1 < i64::MAX;

nodedb/src/event/alert/executor.rs

@@ -172,6 +172,8 @@ async fn execute_aggregate_scan(
         gap_fill: String::new(),
         computed_columns: Vec::new(),
         rls_filters: Vec::new(),
+                system_as_of_ms: None,
+                valid_at_ms: None,
     });
 
     let payload = sync_dispatch::dispatch_async(