feat(nodedb): wire bitemporal strict collections through DDL and Data Plane

farhan-syah · farhan-syah · commit 01512b3cf4dd · 2026-04-24T16:30:44.000+08:00
DDL: the pgwire `CREATE COLLECTION … BITEMPORAL TYPE strict` path now reads
the BITEMPORAL flag before choosing the schema constructor, routing through
`parse_typed_schema_with_flags` which calls `StrictSchema::new_bitemporal`.
The flag is propagated to non-strict collection paths unchanged.

Data Plane: add `bytes_to_binary_tuple_bitemporal` and
`value_to_binary_tuple_bitemporal` helpers in `strict_format`. The put,
update, and upsert handlers check `schema.bitemporal` and call the
bitemporal encoder when appropriate, capturing `system_from_ms` once per
operation so it is consistent between encoding and the versioned-store entry.

Tests cover the value → binary-tuple → timestamp-extraction roundtrip and
rejection of the bitemporal path on plain schemas.
diff --git a/nodedb/src/control/server/pgwire/ddl/collection/create/handler.rs b/nodedb/src/control/server/pgwire/ddl/collection/create/handler.rs
@@ -78,10 +78,18 @@ pub fn create_collection(
     // DataFusion needs all column names+types to build the Arrow schema for planning.
     let mut columnar_schema_columns: Vec<(String, String)> = Vec::new();
 
+    // `BITEMPORAL` flag is consulted here so strict schemas can reserve
+    // tuple slots 0/1/2 for `__system_from_ms` / `__valid_from_ms` /
+    // `__valid_until_ms`. The flag is re-read below for non-strict
+    // collections which route through the versioned sparse table only.
+    let bitemporal_flag = upper.contains("BITEMPORAL");
+
     let collection_type = if upper.contains("TYPE") && upper.contains("KEY_VALUE") {
         super::super::super::kv::parse_kv_collection(sql, &upper)?
     } else if upper.contains("TYPE") && upper.contains("STRICT") {
-        let schema = parse_typed_schema(sql).map_err(|e| sqlstate_error("42601", &e))?;
+        let schema =
+            super::super::helpers::parse_typed_schema_with_flags(sql, bitemporal_flag)
+                .map_err(|e| sqlstate_error("42601", &e))?;
         nodedb_types::CollectionType::strict(schema)
     } else if upper.contains("TYPE") && upper.contains("COLUMNAR") {
         resolve_columnar(sql, &upper, &mut columnar_schema_columns)
@@ -115,7 +123,7 @@ pub fn create_collection(
     // version keyed by `system_from_ms` and reads use the versioned
     // table + Ceiling resolver. Enables `FOR SYSTEM_TIME AS OF` /
     // `FOR VALID_TIME` queries on this collection.
-    let bitemporal = upper.contains("BITEMPORAL");
+    let bitemporal = bitemporal_flag;
     if hash_chain && !append_only {
         return Err(sqlstate_error("42601", "HASH_CHAIN requires APPEND_ONLY"));
     }
diff --git a/nodedb/src/control/server/pgwire/ddl/collection/helpers.rs b/nodedb/src/control/server/pgwire/ddl/collection/helpers.rs
@@ -38,6 +38,16 @@ pub(super) fn extract_with_value(sql: &str, key: &str) -> Option<String> {
 /// Auto-generates `_rowid` PK if no PK column is declared.
 pub(in crate::control::server::pgwire::ddl) fn parse_typed_schema(
     sql: &str,
+) -> Result<nodedb_types::columnar::StrictSchema, String> {
+    parse_typed_schema_with_flags(sql, false)
+}
+
+/// Variant of `parse_typed_schema` that additionally flags the schema
+/// as bitemporal, prepending the three reserved fixed-Int64 slots
+/// (`__system_from_ms`, `__valid_from_ms`, `__valid_until_ms`).
+pub(in crate::control::server::pgwire::ddl) fn parse_typed_schema_with_flags(
+    sql: &str,
+    bitemporal: bool,
 ) -> Result<nodedb_types::columnar::StrictSchema, String> {
     use nodedb_types::columnar::{ColumnDef, ColumnType, StrictSchema};
 
@@ -100,7 +110,11 @@ pub(in crate::control::server::pgwire::ddl) fn parse_typed_schema(
         );
     }
 
-    StrictSchema::new(columns).map_err(|e| e.to_string())
+    if bitemporal {
+        StrictSchema::new_bitemporal(columns).map_err(|e| e.to_string())
+    } else {
+        StrictSchema::new(columns).map_err(|e| e.to_string())
+    }
 }
 
 /// Parse a single column definition: `name TYPE [NOT NULL] [PRIMARY KEY] [DEFAULT expr]`
diff --git a/nodedb/src/control/server/pgwire/ddl/convert.rs b/nodedb/src/control/server/pgwire/ddl/convert.rs
@@ -93,6 +93,7 @@ pub async fn convert_collection(
                 columns,
                 version: 1,
                 dropped_columns: Vec::new(),
+            bitemporal: false,
             };
             if target_type == "kv" {
                 nodedb_types::CollectionType::kv(schema)
diff --git a/nodedb/src/data/executor/handlers/convert.rs b/nodedb/src/data/executor/handlers/convert.rs
@@ -106,6 +106,7 @@ impl CoreLoop {
             columns,
             version: 1,
             dropped_columns: Vec::new(),
+            bitemporal: false,
         };
 
         // Scan all existing documents.
diff --git a/nodedb/src/data/executor/handlers/document/read.rs b/nodedb/src/data/executor/handlers/document/read.rs
@@ -733,6 +733,7 @@ mod tests {
             ],
             version: 1,
             dropped_columns: Vec::new(),
+            bitemporal: false,
         };
         let mut map = std::collections::HashMap::new();
         map.insert("id".into(), Value::String("u1".into()));
diff --git a/nodedb/src/data/executor/handlers/point/apply_put.rs b/nodedb/src/data/executor/handlers/point/apply_put.rs
@@ -50,17 +50,31 @@ impl CoreLoop {
         };
         let value = &value;
 
+        let bitemporal = self.is_bitemporal(tid, collection);
+        let sys_from_ms = self.bitemporal_now_ms();
+        let valid_from_ms = i64::MIN;
+        let valid_until_ms = i64::MAX;
+
         // Check if this collection uses strict (Binary Tuple) encoding.
         let stored = if let Some(config) = self.doc_configs.get(&config_key)
             && let crate::bridge::physical_plan::StorageMode::Strict { ref schema } =
                 config.storage_mode
         {
-            super::super::super::strict_format::bytes_to_binary_tuple(value, schema).map_err(
-                |e| crate::Error::Serialization {
-                    format: "binary_tuple".into(),
-                    detail: e,
-                },
-            )?
+            if bitemporal && schema.bitemporal {
+                super::super::super::strict_format::bytes_to_binary_tuple_bitemporal(
+                    value,
+                    schema,
+                    sys_from_ms,
+                    valid_from_ms,
+                    valid_until_ms,
+                )
+            } else {
+                super::super::super::strict_format::bytes_to_binary_tuple(value, schema)
+            }
+            .map_err(|e| crate::Error::Serialization {
+                format: "binary_tuple".into(),
+                detail: e,
+            })?
         } else {
             value.to_vec()
         };
@@ -69,7 +83,6 @@ impl CoreLoop {
         // (pre-write) version for the `prior` slot, then append a new
         // version at `sys_from = now()`. Non-bitemporal collections use
         // the legacy overwrite path, returning the old bytes redb replaced.
-        let bitemporal = self.is_bitemporal(tid, collection);
         let prior = if bitemporal {
             let current = self
                 .sparse
@@ -80,9 +93,9 @@ impl CoreLoop {
                     tenant: tid,
                     coll: collection,
                     doc_id: document_id,
-                    sys_from_ms: self.bitemporal_now_ms(),
-                    valid_from_ms: i64::MIN,
-                    valid_until_ms: i64::MAX,
+                    sys_from_ms,
+                    valid_from_ms,
+                    valid_until_ms,
                     body: &stored,
                 },
             )?;
diff --git a/nodedb/src/data/executor/handlers/point/update.rs b/nodedb/src/data/executor/handlers/point/update.rs
@@ -56,6 +56,11 @@ impl CoreLoop {
             .any(|(_, v)| matches!(v, UpdateValue::Expr(_)));
 
         let bitemporal = self.is_bitemporal(tid, collection);
+        let sys_from_for_encode = if bitemporal {
+            self.bitemporal_now_ms()
+        } else {
+            0
+        };
         let get_result = if bitemporal {
             self.sparse
                 .versioned_get_current(tid, collection, document_id)
@@ -181,9 +186,20 @@ impl CoreLoop {
                                 config.storage_mode
                         {
                             let ndb_val: nodedb_types::Value = doc.clone().into();
-                            match super::super::super::strict_format::value_to_binary_tuple(
-                                &ndb_val, schema,
-                            ) {
+                            let result = if bitemporal && schema.bitemporal {
+                                super::super::super::strict_format::value_to_binary_tuple_bitemporal(
+                                    &ndb_val,
+                                    schema,
+                                    sys_from_for_encode,
+                                    i64::MIN,
+                                    i64::MAX,
+                                )
+                            } else {
+                                super::super::super::strict_format::value_to_binary_tuple(
+                                    &ndb_val, schema,
+                                )
+                            };
+                            match result {
                                 Ok(bytes) => bytes,
                                 Err(e) => {
                                     return self.response_error(
@@ -213,7 +229,7 @@ impl CoreLoop {
                             tenant: tid,
                             coll: collection,
                             doc_id: document_id,
-                            sys_from_ms: self.bitemporal_now_ms(),
+                            sys_from_ms: sys_from_for_encode,
                             valid_from_ms: i64::MIN,
                             valid_until_ms: i64::MAX,
                             body: &updated_bytes,
diff --git a/nodedb/src/data/executor/handlers/upsert.rs b/nodedb/src/data/executor/handlers/upsert.rs
@@ -120,10 +120,25 @@ impl CoreLoop {
                     apply_on_conflict_updates(existing_val, &new_val, on_conflict_updates)
                 };
 
+                let sys_from_ms = if bitemporal {
+                    self.bitemporal_now_ms()
+                } else {
+                    0
+                };
                 // Encode merged value for storage.
                 let stored_bytes = if let Some(ref schema) = strict_schema {
-                    // Strict: encode directly to binary tuple.
-                    match super::super::strict_format::value_to_binary_tuple(&merged, schema) {
+                    let result = if bitemporal && schema.bitemporal {
+                        super::super::strict_format::value_to_binary_tuple_bitemporal(
+                            &merged,
+                            schema,
+                            sys_from_ms,
+                            i64::MIN,
+                            i64::MAX,
+                        )
+                    } else {
+                        super::super::strict_format::value_to_binary_tuple(&merged, schema)
+                    };
+                    match result {
                         Ok(bt) => bt,
                         Err(e) => {
                             return self.response_error(
@@ -160,7 +175,7 @@ impl CoreLoop {
                             tenant: tid,
                             coll: collection,
                             doc_id: document_id,
-                            sys_from_ms: self.bitemporal_now_ms(),
+                            sys_from_ms,
                             valid_from_ms: i64::MIN,
                             valid_until_ms: i64::MAX,
                             body: &stored_bytes,
diff --git a/nodedb/src/data/executor/strict_format.rs b/nodedb/src/data/executor/strict_format.rs
@@ -62,6 +62,79 @@ pub(super) fn value_to_binary_tuple(
         .map_err(|e| format!("Binary Tuple encode: {e}"))
 }
 
+/// Bitemporal variant: decode msgpack to `Value`, then encode as a Binary
+/// Tuple with reserved slots 0/1/2 populated from the supplied timestamps.
+pub(super) fn bytes_to_binary_tuple_bitemporal(
+    bytes: &[u8],
+    schema: &StrictSchema,
+    system_from_ms: i64,
+    valid_from_ms: i64,
+    valid_until_ms: i64,
+) -> Result<Vec<u8>, String> {
+    let value =
+        nodedb_types::value_from_msgpack(bytes).map_err(|e| format!("zerompk decode: {e}"))?;
+    value_to_binary_tuple_bitemporal(
+        &value,
+        schema,
+        system_from_ms,
+        valid_from_ms,
+        valid_until_ms,
+    )
+}
+
+/// Bitemporal variant: encode a user-supplied `Value::Object` together
+/// with the three reserved bitemporal timestamps.
+pub(super) fn value_to_binary_tuple_bitemporal(
+    value: &Value,
+    schema: &StrictSchema,
+    system_from_ms: i64,
+    valid_from_ms: i64,
+    valid_until_ms: i64,
+) -> Result<Vec<u8>, String> {
+    if !schema.bitemporal {
+        return Err("schema is not bitemporal".into());
+    }
+    let map = match value {
+        Value::Object(m) => m,
+        _ => return Err("strict value must be an Object".to_string()),
+    };
+
+    let user_columns = &schema.columns[3..];
+    let user_names: std::collections::HashSet<&str> =
+        user_columns.iter().map(|c| c.name.as_str()).collect();
+    if let Some(unknown) = map.keys().find(|k| {
+        !user_names.contains(k.as_str())
+            && !nodedb_types::columnar::BITEMPORAL_RESERVED_COLUMNS.contains(&k.as_str())
+    }) {
+        return Err(format!(
+            "unknown field '{unknown}' not present in strict schema"
+        ));
+    }
+
+    let mut user_values = Vec::with_capacity(user_columns.len());
+    for col in user_columns {
+        let field_val = map.get(&col.name);
+        let typed = match field_val {
+            None | Some(Value::Null) => {
+                if !col.nullable {
+                    return Err(format!(
+                        "column '{}' is NOT NULL but no value provided",
+                        col.name
+                    ));
+                }
+                Value::Null
+            }
+            Some(v) => coerce_value(v, &col.column_type, &col.name)?,
+        };
+        user_values.push(typed);
+    }
+
+    let encoder = nodedb_strict::TupleEncoder::new(schema);
+    encoder
+        .encode_bitemporal(system_from_ms, valid_from_ms, valid_until_ms, &user_values)
+        .map_err(|e| format!("Binary Tuple encode: {e}"))
+}
+
 /// Decode a Binary Tuple to `nodedb_types::Value::Object` using the schema.
 ///
 /// Returns `None` if the bytes are not a valid binary tuple (e.g., if they
@@ -411,6 +484,7 @@ mod tests {
             ],
             version: 1,
             dropped_columns: Vec::new(),
+            bitemporal: false,
         }
     }
 
@@ -453,6 +527,52 @@ mod tests {
         assert!(result.unwrap_err().contains("NOT NULL"));
     }
 
+    #[test]
+    fn bitemporal_value_roundtrip() {
+        let schema = StrictSchema::new_bitemporal(vec![
+            ColumnDef::required("id", ColumnType::String).with_primary_key(),
+            ColumnDef::required("name", ColumnType::String),
+        ])
+        .unwrap();
+        let mut map = std::collections::HashMap::new();
+        map.insert("id".into(), Value::String("u1".into()));
+        map.insert("name".into(), Value::String("Alice".into()));
+
+        let tuple = value_to_binary_tuple_bitemporal(
+            &Value::Object(map),
+            &schema,
+            1_700_000_000_000,
+            0,
+            i64::MAX,
+        )
+        .unwrap();
+
+        let decoder = nodedb_strict::TupleDecoder::new(&schema);
+        let (sys, vf, vu) = decoder.extract_bitemporal_timestamps(&tuple).unwrap();
+        assert_eq!(sys, 1_700_000_000_000);
+        assert_eq!(vf, 0);
+        assert_eq!(vu, i64::MAX);
+        assert_eq!(
+            decoder.extract_by_name(&tuple, "id").unwrap(),
+            Value::String("u1".into())
+        );
+    }
+
+    #[test]
+    fn bitemporal_encode_rejects_non_bitemporal_schema() {
+        let schema = test_schema();
+        let map = std::collections::HashMap::new();
+        let result = value_to_binary_tuple_bitemporal(
+            &Value::Object(map),
+            &schema,
+            0,
+            0,
+            0,
+        );
+        assert!(result.is_err());
+        assert!(result.unwrap_err().contains("not bitemporal"));
+    }
+
     #[test]
     fn unknown_field_errors() {
         let schema = test_schema();
diff --git a/nodedb/tests/executor_tests/test_cross_type_join.rs b/nodedb/tests/executor_tests/test_cross_type_join.rs
@@ -97,6 +97,7 @@ fn document_scan_preserves_kv_rows_when_collection_has_strict_config() {
                     ],
                     version: 1,
                     dropped_columns: Vec::new(),
+            bitemporal: false,
                 },
             },
             enforcement: Box::new(EnforcementOptions::default()),
