feat(nodedb-sql): extend bitemporal support to columnar and timeseries engines

Add `bitemporal` and `temporal` fields to `AggregatePlanParams` so that
temporal scope is preserved from the outer SELECT into the inner scan
when planning aggregate queries over bitemporal collections.

Update `ColumnarRules` and `TimeseriesRules` to allow `FOR SYSTEM_TIME /
FOR VALID_TIME` clauses when the collection was created `WITH
BITEMPORAL`; non-bitemporal collections still receive a clear error
message. The aggregate path receives the same guard.

Propagate `temporal` into `SqlPlan::TimeseriesScan` so the conversion
layer has the scope available when building the physical plan.

Fix aggregate planning in the document engines to forward `temporal`
into the inner scan instead of resetting it to the default scope.

Author: farhan-syah

nodedb-sql/src/engine_rules/columnar.rs

@@ -32,11 +32,11 @@ impl EngineRules for ColumnarRules {
     }
 
     fn plan_scan(&self, p: ScanParams) -> Result<SqlPlan> {
-        if p.temporal.is_temporal() {
+        if p.temporal.is_temporal() && !p.bitemporal {
             return Err(SqlError::Unsupported {
                 detail: format!(
-                    "FOR SYSTEM_TIME / FOR VALID_TIME is not supported on columnar \
-                     collection '{}'",
+                    "FOR SYSTEM_TIME / FOR VALID_TIME requires a bitemporal \
+                     collection; '{}' was not created WITH bitemporal = true",
                     p.collection
                 ),
             });
@@ -87,6 +87,15 @@ impl EngineRules for ColumnarRules {
     }
 
     fn plan_aggregate(&self, p: AggregateParams) -> Result<SqlPlan> {
+        if p.temporal.is_temporal() && !p.bitemporal {
+            return Err(SqlError::Unsupported {
+                detail: format!(
+                    "FOR SYSTEM_TIME / FOR VALID_TIME requires a bitemporal \
+                     collection; '{}' was not created WITH bitemporal = true",
+                    p.collection
+                ),
+            });
+        }
         let base_scan = SqlPlan::Scan {
             collection: p.collection,
             alias: p.alias,
@@ -98,7 +107,7 @@ impl EngineRules for ColumnarRules {
             offset: 0,
             distinct: false,
             window_functions: Vec::new(),
-            temporal: crate::temporal::TemporalScope::default(),
+            temporal: p.temporal,
         };
         Ok(SqlPlan::Aggregate {
             input: Box::new(base_scan),

nodedb-sql/src/engine_rules/document_schemaless.rs

@@ -99,7 +99,7 @@ impl EngineRules for SchemalessRules {
             offset: 0,
             distinct: false,
             window_functions: Vec::new(),
-            temporal: crate::temporal::TemporalScope::default(),
+            temporal: p.temporal,
         };
         Ok(SqlPlan::Aggregate {
             input: Box::new(base_scan),

nodedb-sql/src/engine_rules/document_strict.rs

@@ -99,7 +99,7 @@ impl EngineRules for StrictRules {
             offset: 0,
             distinct: false,
             window_functions: Vec::new(),
-            temporal: crate::temporal::TemporalScope::default(),
+            temporal: p.temporal,
         };
         Ok(SqlPlan::Aggregate {
             input: Box::new(base_scan),

nodedb-sql/src/engine_rules/mod.rs

@@ -97,6 +97,14 @@ pub struct AggregateParams {
     pub group_columns: Vec<String>,
     /// Whether the collection has auto-tiering enabled.
     pub has_auto_tier: bool,
+    /// Whether this collection was created with bitemporal storage.
+    /// When `true`, the base scan inside the aggregate is allowed to
+    /// carry a non-default temporal scope.
+    pub bitemporal: bool,
+    /// System-time / valid-time scope to propagate into the underlying
+    /// scan so bitemporal aggregate queries project an as-of snapshot
+    /// before grouping.
+    pub temporal: crate::temporal::TemporalScope,
 }
 
 /// Engine-specific planning rules.

nodedb-sql/src/engine_rules/timeseries.rs

@@ -26,11 +26,11 @@ impl EngineRules for TimeseriesRules {
     }
 
     fn plan_scan(&self, p: ScanParams) -> Result<SqlPlan> {
-        if p.temporal.is_temporal() {
+        if p.temporal.is_temporal() && !p.bitemporal {
             return Err(SqlError::Unsupported {
                 detail: format!(
-                    "FOR SYSTEM_TIME / FOR VALID_TIME is not supported on timeseries \
-                     collection '{}' — filter by the existing time column instead",
+                    "FOR SYSTEM_TIME / FOR VALID_TIME requires a bitemporal \
+                     timeseries collection; '{}' was not created WITH bitemporal = true",
                     p.collection
                 ),
             });
@@ -48,6 +48,7 @@ impl EngineRules for TimeseriesRules {
             gap_fill: String::new(),
             limit: p.limit.unwrap_or(10000),
             tiered: false,
+            temporal: p.temporal,
         })
     }
 
@@ -78,6 +79,15 @@ impl EngineRules for TimeseriesRules {
     }
 
     fn plan_aggregate(&self, p: AggregateParams) -> Result<SqlPlan> {
+        if p.temporal.is_temporal() && !p.bitemporal {
+            return Err(SqlError::Unsupported {
+                detail: format!(
+                    "FOR SYSTEM_TIME / FOR VALID_TIME requires a bitemporal \
+                     timeseries collection; '{}' was not created WITH bitemporal = true",
+                    p.collection
+                ),
+            });
+        }
         Ok(SqlPlan::TimeseriesScan {
             collection: p.collection,
             time_range: default_time_range(),
@@ -89,6 +99,7 @@ impl EngineRules for TimeseriesRules {
             gap_fill: String::new(),
             limit: p.limit,
             tiered: p.has_auto_tier,
+            temporal: p.temporal,
         })
     }
 }

nodedb-sql/src/planner/aggregate.rs

@@ -8,6 +8,7 @@ use crate::functions::registry::{FunctionRegistry, SearchTrigger};
 use crate::parser::normalize::normalize_ident;
 use crate::resolver::columns::ResolvedTable;
 use crate::resolver::expr::convert_expr;
+use crate::temporal::TemporalScope;
 use crate::types::*;
 
 /// Plan an aggregate query (GROUP BY + aggregate functions).
@@ -17,6 +18,7 @@ pub fn plan_aggregate(
     filters: &[Filter],
     _scope: &crate::resolver::columns::TableScope,
     functions: &FunctionRegistry,
+    temporal: &TemporalScope,
 ) -> Result<SqlPlan> {
     let group_by_exprs = convert_group_by(&select.group_by)?;
     let aggregates = extract_aggregates_from_projection(&select.projection, functions)?;
@@ -41,6 +43,8 @@ pub fn plan_aggregate(
         bucket_interval_ms,
         group_columns,
         has_auto_tier: table.info.has_auto_tier,
+        bitemporal: table.info.bitemporal,
+        temporal: *temporal,
     })
 }
 

nodedb-sql/src/planner/select.rs

@@ -164,7 +164,7 @@ fn plan_select(
     // 6. Check for GROUP BY / aggregation.
     if has_aggregation(select, functions) {
         let mut plan =
-            super::aggregate::plan_aggregate(select, table, &filters, &scope, functions)?;
+            super::aggregate::plan_aggregate(select, table, &filters, &scope, functions, &temporal)?;
 
         // Semi/anti subquery joins belong below the aggregate so they filter
         // the input rows before grouping. Scalar subqueries remain above the

nodedb-sql/src/types.rs

@@ -181,6 +181,10 @@ pub enum SqlPlan {
         gap_fill: String,
         limit: usize,
         tiered: bool,
+        /// Bitemporal system-time / valid-time scope. Only non-default
+        /// on collections created `WITH BITEMPORAL`; `TimeseriesRules::plan_scan`
+        /// rejects temporal scopes otherwise.
+        temporal: crate::temporal::TemporalScope,
     },
     TimeseriesIngest {
         collection: String,