feat(timeseries): honor declared TIME_KEY from DDL instead of inferring it

Timeseries collections created through DDL now carry their column list and
designated TIME_KEY from the catalog into the Data Plane's collection
config, so ingest and scan use the name and shape DDL resolved rather than
guessing from the first ingested batch or matching conventional column
names (ts/timestamp/time). Only measurements with no DDL behind them (raw
ILP protocol ingest) still fall back to inference.

Time-range extraction for scan pruning moves from a heuristic WHERE-clause
walk over recognized time-field names into a schema-aware path driven by
the declared TIME_KEY, and lives in the shared scan_filter module so both
planner and executor can use it. TimeseriesScan also gains ORDER BY
pushdown (sort_keys) since the engine's natural scan order does not match
requested sort order.

The `ddl_ast::collection_type` module is split into a directory of
focused submodules (build, designated, kv, strict, type_str) to keep the
TIME_KEY resolution logic isolated from the surrounding DDL parsing.

Author: farhan-syah

Cargo.lock

@@ -6,6 +6,7 @@ pub mod enforcement_types;
 pub mod merge_types;
 pub mod ollp_edge;
 pub mod op;
+pub mod timeseries_schema;
 pub mod types;
 pub mod update_value;
 
@@ -15,6 +16,7 @@ pub use enforcement_types::{
 pub use merge_types::{MergeActionOp, MergeClauseKind as MergeClauseKindOp, MergeClauseOp};
 pub use ollp_edge::OllpPredictedEdge;
 pub use op::DocumentOp;
+pub use timeseries_schema::TimeseriesSchema;
 pub use types::{
     BalancedDef, EnforcementOptions, GeneratedColumnSpec, MaterializedSumBinding, PeriodLockConfig,
     RegisteredIndex, RegisteredIndexState, ReturningColumns, ReturningItem, ReturningSpec,

nodedb-physical/src/physical_plan/document/mod.rs

@@ -4,6 +4,7 @@ use nodedb_types::{Surrogate, SurrogateBitmap, SystemTimeScope};
 
 use super::merge_types::MergeClauseOp;
 use super::ollp_edge::OllpPredictedEdge;
+use super::timeseries_schema::TimeseriesSchema;
 use super::types::{EnforcementOptions, RegisteredIndex, ReturningSpec, StorageMode, UpdateValue};
 
 /// Document engine physical operations (schemaless + strict + DML).
@@ -184,6 +185,12 @@ pub enum DocumentOp {
         /// `None` = no explicit policy persisted; the registry falls back to
         /// the ephemeral default.
         conflict_policy: Option<String>,
+        /// Declared columns + designated `TIME_KEY` for a timeseries
+        /// collection. `Some` for every `engine='timeseries'` collection;
+        /// `None` for every other engine. The Data Plane builds the
+        /// collection's memtable schema from this instead of inferring one
+        /// from the first ingested batch.
+        timeseries: Option<Box<TimeseriesSchema>>,
     },
 
     /// Lookup documents by secondary index value.

nodedb-physical/src/physical_plan/document/op.rs

@@ -0,0 +1,38 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! The declared shape of a timeseries collection, propagated from the
+//! catalog to the Data Plane on collection registration.
+
+/// A timeseries collection's DDL-declared column list plus its designated
+/// `TIME_KEY` column.
+///
+/// This is the authority for the collection's storage layout. Without it the
+/// Data Plane would have to derive a schema from whatever the first ingested
+/// batch happened to contain, which cannot recover the declared column order,
+/// the declared types, or — critically — which column is the time key. Every
+/// path that needs to know "which column carries this collection's time"
+/// reads [`Self::time_key`]; none may guess it from a column name.
+#[derive(
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    serde::Serialize,
+    serde::Deserialize,
+    zerompk::ToMessagePack,
+    zerompk::FromMessagePack,
+)]
+pub struct TimeseriesSchema {
+    /// Name of the designated time-key column, exactly as declared.
+    pub time_key: String,
+    /// Declared `(column_name, type_str)` pairs in declaration order.
+    pub columns: Vec<(String, String)>,
+}
+
+impl TimeseriesSchema {
+    /// Position of the time key in `columns`, or `None` when the declared
+    /// column list does not contain it (an inconsistent catalog record).
+    pub fn time_key_index(&self) -> Option<usize> {
+        self.columns.iter().position(|(n, _)| *n == self.time_key)
+    }
+}

nodedb-physical/src/physical_plan/document/timeseries_schema.rs

@@ -33,7 +33,7 @@ pub use crdt::CrdtOp;
 pub use document::{
     BalancedDef, DocumentOp, EnforcementOptions, GeneratedColumnSpec, MaterializedSumBinding,
     OllpPredictedEdge, PeriodLockConfig, RegisteredIndex, RegisteredIndexState, ReturningColumns,
-    ReturningItem, ReturningSpec, StorageMode, UpdateValue,
+    ReturningItem, ReturningSpec, StorageMode, TimeseriesSchema, UpdateValue,
 };
 pub use exchange::{ExchangeMode, ExchangeOp};
 pub use graph::{
@@ -45,7 +45,7 @@ pub use query::{AggregateSpec, GroupKeySpec, JoinProjection, QueryOp};
 pub use routing::plan_contains_cluster_partitioned_leaf;
 pub use spatial::{SpatialOp, SpatialPredicate};
 pub use text::TextOp;
-pub use timeseries::TimeseriesOp;
+pub use timeseries::{TimeseriesOp, UNBOUNDED_TIME_RANGE};
 pub use vector::VectorOp;
 pub use wire::{decode, encode};
 

nodedb-physical/src/physical_plan/mod.rs

@@ -4,6 +4,15 @@
 
 use nodedb_types::{Surrogate, SystemTimeScope};
 
+/// An unconstrained `(min_ts_ms, max_ts_ms)` envelope.
+///
+/// The Control Plane always plans a timeseries scan unbounded: narrowing it
+/// requires knowing which column is the collection's declared `TIME_KEY`, and
+/// that is resolved in the Data Plane where the collection's registered schema
+/// lives. Internal callers that already know an exact window (retention,
+/// continuous-aggregate refresh) pass their own bounds instead.
+pub const UNBOUNDED_TIME_RANGE: (i64, i64) = (i64::MIN, i64::MAX);
+
 /// Timeseries engine physical operations.
 #[derive(
     Debug,
@@ -22,11 +31,18 @@ pub enum TimeseriesOp {
     /// memtable and sealed disk partitions.
     Scan {
         collection: String,
-        /// `(min_ts_ms, max_ts_ms)`. (0, i64::MAX) = no time filter.
+        /// `(min_ts_ms, max_ts_ms)` pruning envelope. The Data Plane narrows
+        /// it further using the query's bounds on the declared `TIME_KEY`;
+        /// see [`UNBOUNDED_TIME_RANGE`].
         time_range: (i64, i64),
         projection: Vec<String>,
         limit: usize,
         filters: Vec<u8>,
+        /// `ORDER BY` keys as `(column, ascending)` in significance order.
+        /// Empty = the engine's natural order. Applied to the materialized
+        /// result before `limit` is enforced, so an ordered query returns the
+        /// first `limit` rows of the ordering the client asked for.
+        sort_keys: Vec<(String, bool)>,
         /// time_bucket interval in milliseconds. 0 = no bucketing.
         bucket_interval_ms: i64,
         /// GROUP BY column names (empty = no grouping or whole-table agg).

nodedb-physical/src/physical_plan/timeseries.rs

@@ -9,9 +9,11 @@
 pub mod like;
 pub mod op;
 pub mod parse;
+pub mod timestamp;
 pub mod types;
 
 pub use like::sql_like_match;
 pub use op::FilterOp;
 pub use parse::parse_simple_predicates;
+pub use timestamp::value_as_timestamp_ms;
 pub use types::ScanFilter;

nodedb-query/src/scan_filter/mod.rs

@@ -0,0 +1,85 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! Coercion of a filter's comparison value to epoch milliseconds.
+//!
+//! A predicate against a timestamp column can arrive carrying any of the
+//! shapes SQL accepts for an instant: a datetime literal (`ts < '2020-03-05
+//! 10:00:00'`), an epoch-millisecond integer (`ts < 1583402400000`), or an
+//! already-typed datetime value. Timestamp columns store epoch milliseconds,
+//! so every one of those has to reduce to the same scalar before comparison —
+//! otherwise a perfectly ordinary predicate silently matches nothing.
+
+use nodedb_types::Value;
+
+/// Reduce a filter value to epoch milliseconds, or `None` when it cannot
+/// denote an instant.
+pub fn value_as_timestamp_ms(value: &Value) -> Option<i64> {
+    match value {
+        Value::Integer(ms) => Some(*ms),
+        Value::Float(ms) => Some(*ms as i64),
+        Value::DateTime(dt) | Value::NaiveDateTime(dt) => Some(dt.unix_millis()),
+        Value::String(text) => nodedb_types::datetime::NdbDateTime::parse(text)
+            .map(|dt| dt.unix_millis())
+            .or_else(|| text.trim().parse::<i64>().ok()),
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// 2020-03-05T10:00:00Z in epoch milliseconds.
+    const MARCH_5_2020: i64 = 1_583_402_400_000;
+
+    #[test]
+    fn epoch_millis_pass_through() {
+        assert_eq!(
+            value_as_timestamp_ms(&Value::Integer(MARCH_5_2020)),
+            Some(MARCH_5_2020)
+        );
+    }
+
+    #[test]
+    fn datetime_literals_are_parsed() {
+        assert_eq!(
+            value_as_timestamp_ms(&Value::String("2020-03-05 10:00:00".into())),
+            Some(MARCH_5_2020)
+        );
+        assert_eq!(
+            value_as_timestamp_ms(&Value::String("2020-03-05T10:00:00Z".into())),
+            Some(MARCH_5_2020)
+        );
+    }
+
+    #[test]
+    fn numeric_strings_are_read_as_epoch_millis() {
+        assert_eq!(
+            value_as_timestamp_ms(&Value::String(MARCH_5_2020.to_string())),
+            Some(MARCH_5_2020)
+        );
+    }
+
+    #[test]
+    fn typed_datetimes_are_accepted() {
+        let dt = nodedb_types::NdbDateTime::from_millis(MARCH_5_2020).unwrap();
+        assert_eq!(
+            value_as_timestamp_ms(&Value::NaiveDateTime(dt)),
+            Some(MARCH_5_2020)
+        );
+        assert_eq!(
+            value_as_timestamp_ms(&Value::DateTime(dt)),
+            Some(MARCH_5_2020)
+        );
+    }
+
+    #[test]
+    fn non_instants_are_rejected() {
+        assert_eq!(value_as_timestamp_ms(&Value::Null), None);
+        assert_eq!(value_as_timestamp_ms(&Value::Bool(true)), None);
+        assert_eq!(
+            value_as_timestamp_ms(&Value::String("not a date".into())),
+            None
+        );
+    }
+}