feat(nodedb): add bitemporal stamping to timeseries ILP ingest

Extend the ILP ingest path to stamp rows with bitemporal columns when
the target collection is configured as bitemporal. The executor handler
now resolves the bitemporal flag before memtable creation and passes
BitempStamps through ingest_batch_with_lvc.

On the engine side, ingest_batch_with_lvc gains an optional BitempStamps
parameter. When present, _ts_system is set to the server ingest time,
while _ts_valid_from and _ts_valid_until are sourced from the line's
field set, defaulting to the open interval [i64::MIN, i64::MAX).

Schema inference and evolution logic is delegated to the ilp_schema
sibling module (via re-exports), and the msgpack ingest handler is moved
to the ingest_formats sibling module, keeping both files within the
500-line limit.

Adds a unit test covering late-arriving IoT backfill to verify the three
reserved column slots carry the expected stamps.

Author: farhan-syah

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

@@ -1,10 +1,10 @@
 //! Timeseries ILP ingest handler.
+//!
+//! msgpack / JSON row ingests that normalize into ILP text live in the
+//! sibling `ingest_formats` module.
 
 use std::collections::HashMap;
 
-use sonic_rs::{JsonContainerTrait, JsonValueTrait};
-
-use super::msgpack_decode::{self, MsgpackValue};
 use crate::bridge::envelope::{ErrorCode, Payload, Response, Status};
 use crate::data::executor::core_loop::CoreLoop;
 use crate::data::executor::response_codec;
@@ -91,7 +91,7 @@ impl CoreLoop {
         }
     }
 
-    fn execute_ilp_ingest(
+    pub(super) fn execute_ilp_ingest(
         &mut self,
         task: &ExecutionTask,
         tid: crate::types::TenantId,
@@ -127,10 +127,14 @@ impl CoreLoop {
             );
         }
 
+        let bitemporal = self.is_bitemporal(tid.as_u32(), collection);
         // Ensure memtable exists (auto-create on first write).
         let is_new_memtable = !self.columnar_memtables.contains_key(&key);
         if is_new_memtable {
-            let schema = ilp_ingest::infer_schema(&lines);
+            let mut schema = ilp_ingest::infer_schema(&lines);
+            if bitemporal {
+                ilp_ingest::ensure_bitemporal_columns(&mut schema);
+            }
             let config = ColumnarMemtableConfig {
                 max_memory_bytes: 64 * 1024 * 1024,
                 hard_memory_limit: 80 * 1024 * 1024,
@@ -184,10 +188,15 @@ impl CoreLoop {
             let _ = gov.try_reserve(nodedb_mem::EngineId::Timeseries, batch_estimate);
         }
 
+        let stamps = if bitemporal {
+            Some(ilp_ingest::BitempStamps { system_ms: now_ms })
+        } else {
+            None
+        };
         let lvc = self.ts_last_value_caches.get_mut(&key);
         let mut series_keys = HashMap::new();
         let (mut accepted, rejected) =
-            ilp_ingest::ingest_batch_with_lvc(mt, &lines, &mut series_keys, now_ms, lvc);
+            ilp_ingest::ingest_batch_with_lvc(mt, &lines, &mut series_keys, now_ms, lvc, stamps);
 
         // If rows were rejected (memtable hit hard limit), flush and re-ingest.
         if rejected > 0 {
@@ -208,6 +217,7 @@ impl CoreLoop {
                     &mut retry_keys,
                     now_ms,
                     retry_lvc,
+                    stamps,
                 );
                 accepted += retry_accepted;
             }
@@ -289,264 +299,4 @@ impl CoreLoop {
             error_code: None,
         }
     }
-
-    /// Ingest JSON row objects from SQL INSERT path.
-    ///
-    /// Payload is a msgpack array of maps (same schema as JSON ingest but in msgpack).
-    /// Converts each row to an ILP line and delegates to the ILP ingest path.
-    fn execute_msgpack_ingest(
-        &mut self,
-        task: &ExecutionTask,
-        tid: crate::types::TenantId,
-        collection: &str,
-        payload: &[u8],
-        wal_lsn: Option<u64>,
-        now_ms: i64,
-    ) -> Response {
-        let measurement = collection
-            .split_once(':')
-            .map(|(_, name)| name)
-            .unwrap_or(collection);
-
-        if !measurement
-            .chars()
-            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
-        {
-            return self.response_error(
-                task,
-                ErrorCode::Internal {
-                    detail: format!(
-                        "invalid measurement name '{measurement}': only [a-zA-Z0-9_-] allowed"
-                    ),
-                },
-            );
-        }
-
-        let rows = match msgpack_decode::decode_msgpack_rows(payload) {
-            Ok(r) => r,
-            Err(e) => {
-                return self.response_error(
-                    task,
-                    ErrorCode::Internal {
-                        detail: format!("msgpack decode error: {e}"),
-                    },
-                );
-            }
-        };
-
-        if rows.is_empty() {
-            return self.response_error(
-                task,
-                ErrorCode::Internal {
-                    detail: "empty msgpack rows array".into(),
-                },
-            );
-        }
-
-        let mut ilp_buf = String::new();
-        for row in &rows {
-            let mut fields = Vec::new();
-            let mut timestamp_ns: Option<i64> = None;
-
-            for (key, val) in row {
-                let lower = key.to_lowercase();
-                if lower == "ts" || lower == "timestamp" || lower == "time" {
-                    match val {
-                        MsgpackValue::Str(s) => {
-                            timestamp_ns = parse_ts_string_to_nanos(s);
-                        }
-                        MsgpackValue::Int(n) => {
-                            timestamp_ns = Some(*n * 1_000_000);
-                        }
-                        MsgpackValue::Float(f) => {
-                            timestamp_ns = Some(*f as i64 * 1_000_000);
-                        }
-                        _ => {}
-                    }
-                    continue;
-                }
-
-                match val {
-                    MsgpackValue::Float(f) => fields.push(format!("{key}={f}")),
-                    MsgpackValue::Int(n) => fields.push(format!("{key}={n}i")),
-                    MsgpackValue::Str(s) => {
-                        fields.push(format!("{key}=\"{}\"", s.replace('\"', "\\\"")));
-                    }
-                    MsgpackValue::Bool(b) => fields.push(format!("{key}={b}")),
-                    _ => {}
-                }
-            }
-
-            if fields.is_empty() {
-                continue;
-            }
-
-            ilp_buf.push_str(measurement);
-            ilp_buf.push(' ');
-            ilp_buf.push_str(&fields.join(","));
-            if let Some(ts) = timestamp_ns {
-                ilp_buf.push(' ');
-                ilp_buf.push_str(&ts.to_string());
-            }
-            ilp_buf.push('\n');
-        }
-
-        if ilp_buf.is_empty() {
-            return self.response_error(
-                task,
-                ErrorCode::Internal {
-                    detail: "no valid rows in msgpack payload".into(),
-                },
-            );
-        }
-
-        self.execute_ilp_ingest(task, tid, collection, ilp_buf.as_bytes(), wal_lsn, now_ms)
-    }
-
-    /// Payload is a JSON array like: `[{"id":"e1","ts":"2024-01-01T00:00:00Z","value":42.0}]`.
-    /// Converts each row to an ILP line and delegates to the ILP ingest path.
-    fn execute_json_ingest(
-        &mut self,
-        task: &ExecutionTask,
-        tid: crate::types::TenantId,
-        collection: &str,
-        payload: &[u8],
-        wal_lsn: Option<u64>,
-        now_ms: i64,
-    ) -> Response {
-        let rows: sonic_rs::Array = match sonic_rs::from_slice(payload) {
-            Ok(r) => r,
-            Err(e) => {
-                return self.response_error(
-                    task,
-                    ErrorCode::Internal {
-                        detail: format!("JSON parse error: {e}"),
-                    },
-                );
-            }
-        };
-
-        if rows.is_empty() {
-            return self.response_error(
-                task,
-                ErrorCode::Internal {
-                    detail: "empty JSON rows array".into(),
-                },
-            );
-        }
-
-        // Convert JSON rows to ILP text.
-        // The collection name serves as the ILP measurement.
-        // Strip tenant scope prefix if present (e.g., "1:events" → "events").
-        let measurement = collection
-            .split_once(':')
-            .map(|(_, name)| name)
-            .unwrap_or(collection);
-
-        // Validate measurement name: only [a-zA-Z0-9_-] allowed.
-        // ILP special characters (space, comma, =, newline) in measurement names corrupt the line.
-        if !measurement
-            .chars()
-            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
-        {
-            return self.response_error(
-                task,
-                ErrorCode::Internal {
-                    detail: format!(
-                        "invalid measurement name '{measurement}': only [a-zA-Z0-9_-] allowed"
-                    ),
-                },
-            );
-        }
-
-        let mut ilp_buf = String::new();
-        for row_val in rows.iter() {
-            let obj = match row_val.as_object() {
-                Some(o) => o,
-                None => continue,
-            };
-
-            let mut fields = Vec::new();
-            let mut timestamp_ns: Option<i64> = None;
-
-            for (key, val) in obj.iter() {
-                let lower = key.to_lowercase();
-                if lower == "ts" || lower == "timestamp" || lower == "time" {
-                    if let Some(s) = val.as_str() {
-                        timestamp_ns = parse_ts_string_to_nanos(s);
-                    } else if let Some(n) = val.as_i64() {
-                        // Treat integer timestamps as milliseconds → convert to ns.
-                        timestamp_ns = Some(n * 1_000_000);
-                    } else if let Some(f) = val.as_f64() {
-                        timestamp_ns = Some(f as i64 * 1_000_000);
-                    }
-                    continue;
-                }
-
-                if let Some(f) = val.as_f64() {
-                    fields.push(format!("{key}={f}"));
-                } else if let Some(n) = val.as_i64() {
-                    fields.push(format!("{key}={n}i"));
-                } else if let Some(s) = val.as_str() {
-                    fields.push(format!("{key}=\"{}\"", s.replace('\"', "\\\"")));
-                } else if let Some(b) = val.as_bool() {
-                    fields.push(format!("{key}={b}"));
-                }
-            }
-
-            if fields.is_empty() {
-                continue;
-            }
-
-            ilp_buf.push_str(measurement);
-            ilp_buf.push(' ');
-            ilp_buf.push_str(&fields.join(","));
-            if let Some(ts) = timestamp_ns {
-                ilp_buf.push(' ');
-                ilp_buf.push_str(&ts.to_string());
-            }
-            ilp_buf.push('\n');
-        }
-
-        if ilp_buf.is_empty() {
-            return self.response_error(
-                task,
-                ErrorCode::Internal {
-                    detail: "no valid rows in JSON payload".into(),
-                },
-            );
-        }
-
-        // Delegate to the ILP ingest path.
-        self.execute_ilp_ingest(task, tid, collection, ilp_buf.as_bytes(), wal_lsn, now_ms)
-    }
-}
-
-/// Parse a datetime string to nanoseconds since Unix epoch.
-///
-/// Accepts RFC3339 / ISO8601 with timezone (e.g., "2024-01-01T00:00:00Z"),
-/// and common datetime formats without timezone (treated as UTC).
-/// Returns nanoseconds since Unix epoch, or `None` if the string cannot be parsed.
-fn parse_ts_string_to_nanos(s: &str) -> Option<i64> {
-    use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
-
-    // Try RFC3339 first (includes timezone info).
-    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
-        return dt.timestamp_nanos_opt();
-    }
-
-    // Try common space-separated format "YYYY-MM-DD HH:MM:SS[.frac]".
-    let formats = [
-        "%Y-%m-%d %H:%M:%S%.f",
-        "%Y-%m-%d %H:%M:%S",
-        "%Y-%m-%dT%H:%M:%S%.f",
-        "%Y-%m-%dT%H:%M:%S",
-    ];
-    for fmt in &formats {
-        if let Ok(ndt) = NaiveDateTime::parse_from_str(s, fmt) {
-            return Utc.from_utc_datetime(&ndt).timestamp_nanos_opt();
-        }
-    }
-
-    None
 }