feat(nodedb): wire bitemporal temporal scope through columnar and timeseries physical plans

Add `system_as_of_ms` and `valid_at_ms` fields to `ColumnarOp::Scan` and
`TimeseriesOp::Scan` in the physical plan. All non-bitemporal construction
sites (auto-tier, aggregate convert, native plan builders, wire helpers,
alert executor) initialise both fields to `None`.

In `sql_plan_convert`, propagate `temporal.system_as_of_ms` and the
valid-time point derived from `temporal` into the physical scan for
plain-columnar and timeseries paths; spatial and KV paths continue to
pass `None`. Thread `temporal` through `TimeseriesScanParams` so the
timeseries conversion function has access to the scope.

Author: farhan-syah

nodedb/src/bridge/physical_plan/columnar.rs

@@ -61,6 +61,18 @@ pub enum ColumnarOp {
         /// against matching rows before the `limit` is enforced. Empty
         /// for scans with no ORDER BY.
         sort_keys: Vec<(String, bool)>,
+        /// Bitemporal system-time cutoff: read rows whose `_ts_system`
+        /// is ≤ this value. `None` = current-state read. Only populated
+        /// for collections created `WITH BITEMPORAL`.
+        #[serde(default)]
+        #[msgpack(default)]
+        system_as_of_ms: Option<i64>,
+        /// Bitemporal valid-time predicate: keep rows whose
+        /// `[_ts_valid_from, _ts_valid_until)` interval contains this
+        /// point. `None` = no valid-time filter.
+        #[serde(default)]
+        #[msgpack(default)]
+        valid_at_ms: Option<i64>,
     },
 
     /// Insert rows into a columnar memtable.

nodedb/src/bridge/physical_plan/timeseries.rs

@@ -39,6 +39,19 @@ pub enum TimeseriesOp {
         computed_columns: Vec<u8>,
         /// RLS post-scan filters (applied after time-range pruning).
         rls_filters: Vec<u8>,
+        /// Bitemporal system-time cutoff. When `Some`, block-skip on
+        /// `_ts_system` min/max and a post-fetch filter exclude rows
+        /// written after the given epoch-ms. Only populated for
+        /// timeseries collections created `WITH BITEMPORAL`.
+        #[serde(default)]
+        #[msgpack(default)]
+        system_as_of_ms: Option<i64>,
+        /// Bitemporal valid-time point. When `Some`, only rows whose
+        /// `[_ts_valid_from, _ts_valid_until)` interval contains this
+        /// point are returned.
+        #[serde(default)]
+        #[msgpack(default)]
+        valid_at_ms: Option<i64>,
     },
 
     /// Write a batch of samples to the columnar memtable.

nodedb/src/bridge/physical_plan/wire.rs

@@ -150,6 +150,8 @@ mod tests {
             filters: vec![],
             rls_filters: vec![],
             sort_keys: vec![],
+            system_as_of_ms: None,
+            valid_at_ms: None,
         }));
     }
 
@@ -167,6 +169,8 @@ mod tests {
             gap_fill: "null".into(),
             computed_columns: vec![],
             rls_filters: vec![],
+                system_as_of_ms: None,
+                valid_at_ms: None,
         }));
     }
 

nodedb/src/control/planner/auto_tier.rs

@@ -225,6 +225,8 @@ fn build_scan_task(
             gap_fill: gap_fill.to_string(),
             computed_columns: Vec::new(),
             rls_filters: Vec::new(),
+                system_as_of_ms: None,
+                valid_at_ms: None,
         }),
         post_set_op: PostSetOp::None,
     }

nodedb/src/control/planner/sql_plan_convert/aggregate.rs

@@ -112,6 +112,8 @@ pub(super) fn convert_aggregate(
                 gap_fill: String::new(),
                 computed_columns: Vec::new(),
                 rls_filters: Vec::new(),
+                system_as_of_ms: None,
+                valid_at_ms: None,
             }),
             post_set_op: PostSetOp::None,
         }]);

nodedb/src/control/planner/sql_plan_convert/convert.rs

@@ -210,6 +210,7 @@ pub(super) fn convert_one(
             gap_fill,
             limit,
             tiered,
+            temporal,
         } => super::scan::convert_timeseries_scan(super::scan_params::TimeseriesScanParams {
             collection,
             time_range,
@@ -223,6 +224,7 @@ pub(super) fn convert_one(
             tiered,
             tenant_id,
             ctx,
+            temporal,
         }),
 
         SqlPlan::TimeseriesIngest { collection, rows } => {

nodedb/src/control/planner/sql_plan_convert/scan.rs

@@ -103,6 +103,8 @@ pub(super) fn convert_scan(p: ScanParams<'_>) -> crate::Result<Vec<PhysicalTask>
                 gap_fill: String::new(),
                 computed_columns: computed_bytes,
                 rls_filters: Vec::new(),
+                system_as_of_ms: None,
+                valid_at_ms: None,
             })
         }
         EngineType::Columnar => PhysicalPlan::Columnar(ColumnarOp::Scan {
@@ -112,6 +114,8 @@ pub(super) fn convert_scan(p: ScanParams<'_>) -> crate::Result<Vec<PhysicalTask>
             filters: filter_bytes,
             rls_filters: Vec::new(),
             sort_keys: sort.clone(),
+            system_as_of_ms: temporal.system_as_of_ms,
+            valid_at_ms: valid_at_from_scope(temporal),
         }),
         EngineType::Spatial => PhysicalPlan::Columnar(ColumnarOp::Scan {
             collection: collection.into(),
@@ -120,6 +124,8 @@ pub(super) fn convert_scan(p: ScanParams<'_>) -> crate::Result<Vec<PhysicalTask>
             filters: filter_bytes,
             rls_filters: Vec::new(),
             sort_keys: sort.clone(),
+            system_as_of_ms: None,
+            valid_at_ms: None,
         }),
         EngineType::KeyValue => PhysicalPlan::Kv(KvOp::Scan {
             collection: collection.into(),
@@ -237,6 +243,8 @@ pub(super) fn convert_point_get(
                 filters: filter_bytes,
                 rls_filters: Vec::new(),
                 sort_keys: Vec::new(),
+                system_as_of_ms: None,
+                valid_at_ms: None,
             })
         }
         // Timeseries should never reach here — nodedb-sql rejects point gets.
@@ -341,6 +349,7 @@ pub(super) fn convert_timeseries_scan(
         tiered,
         tenant_id,
         ctx,
+        temporal,
     } = p;
     let filter_bytes = serialize_filters(filters)?;
     let agg_pairs: Vec<(String, String)> = aggregates.iter().map(agg_expr_to_pair).collect();
@@ -379,6 +388,8 @@ pub(super) fn convert_timeseries_scan(
             gap_fill: gap_fill.to_string(),
             computed_columns: Vec::new(),
             rls_filters: Vec::new(),
+            system_as_of_ms: temporal.system_as_of_ms,
+            valid_at_ms: valid_at_from_scope(temporal),
         }),
         post_set_op: PostSetOp::None,
     }])

nodedb/src/control/planner/sql_plan_convert/scan_params.rs

@@ -61,6 +61,7 @@ pub(super) struct TimeseriesScanParams<'a> {
     pub tiered: &'a bool,
     pub tenant_id: TenantId,
     pub ctx: &'a ConvertContext,
+    pub temporal: &'a nodedb_sql::TemporalScope,
 }
 
 /// Parameters for `convert_hybrid_search`.

nodedb/src/control/server/native/dispatch/plan_builder/columnar.rs

@@ -16,6 +16,8 @@ pub(crate) fn build_scan(fields: &TextFields, collection: &str) -> crate::Result
         filters,
         rls_filters: Vec::new(),
         sort_keys: Vec::new(),
+        system_as_of_ms: None,
+        valid_at_ms: None,
     }))
 }
 

nodedb/src/control/server/native/dispatch/plan_builder/timeseries.rs

@@ -31,6 +31,8 @@ pub(crate) fn build_scan(fields: &TextFields, collection: &str) -> crate::Result
         gap_fill: String::new(),
         computed_columns: Vec::new(),
         rls_filters: Vec::new(),
+                system_as_of_ms: None,
+                valid_at_ms: None,
     }))
 }