feat(nodedb): use system-time axis for bitemporal retention

Bitemporal partitions now evaluate retention staleness against
`max_system_ts` (the maximum `_ts_system` value written to the
partition) rather than `max_ts` (event-time). This prevents
late-arriving backfill rows from being prematurely expired when their
event-time predates the retention cutoff but their write-time does not.

Non-bitemporal partitions are unaffected: `max_system_ts` defaults to 0
and the existing `max_ts` path is taken.

Changes:
- Add `max_system_ts` field to `PartitionMeta` (serde default = 0 for
  backward-compatible deserialization)
- Populate `max_system_ts` on memtable drain by scanning the
  `_ts_system` column; propagate through segment writer, O3 merge, and
  partition merge
- Add `is_bitemporal()` / `ts_system_idx()` helpers to `ColumnarSchema`
  and `ColumnarMemtableSchema`
- Thread `bitemporal: bool` into `find_expired` and the compaction /
  retention-scan paths in the executor
- Skip upsert-tombstone in columnar write for bitemporal collections so
  every version of a PK is preserved for `AS OF` queries

Author: farhan-syah

nodedb-columnar/src/mutation/write.rs

@@ -28,10 +28,19 @@ impl MutationEngine {
         let pk_bytes = self.extract_pk_bytes(values)?;
         let mut wal_records = Vec::with_capacity(2);
 
+        // Bitemporal collections preserve every version of a PK: each
+        // write appends a new row with a distinct `_ts_system` stamp and
+        // the prior row stays visible to `AS OF` queries. Skipping the
+        // upsert-tombstone here keeps compaction lossless without
+        // needing a separate "version-aware" delete bitmap. The PK
+        // index is still rebound below so current-state reads see the
+        // latest version.
+        let bitemporal = self.schema.is_bitemporal();
+
         // If a prior row exists for this PK, tombstone it in place so
         // subsequent scans skip the stale row. The PK index is rebound
         // below to the freshly-appended row.
-        if let Some(prior) = self.pk_index.get(&pk_bytes).copied() {
+        if !bitemporal && let Some(prior) = self.pk_index.get(&pk_bytes).copied() {
             let bitmap = self.delete_bitmaps.entry(prior.segment_id).or_default();
             bitmap.mark_deleted(prior.row_index);
             wal_records.push(ColumnarWalRecord::DeleteRows {

nodedb-types/src/columnar/schema.rs

@@ -305,6 +305,21 @@ impl ColumnarSchema {
             version: 1,
         })
     }
+
+    /// Whether this schema has the reserved `_ts_system` bitemporal column.
+    ///
+    /// Detected by column name rather than a separate flag to keep the
+    /// on-disk manifest format unchanged; `_ts_system` is only inserted
+    /// by `prepend_bitemporal_columns` on the write path, so its
+    /// presence is a reliable bitemporal signal.
+    pub fn is_bitemporal(&self) -> bool {
+        self.columns.iter().any(|c| c.name == "_ts_system")
+    }
+
+    /// Position of the `_ts_system` column, or `None` for non-bitemporal.
+    pub fn ts_system_idx(&self) -> Option<usize> {
+        self.columns.iter().position(|c| c.name == "_ts_system")
+    }
 }
 
 #[cfg(test)]

nodedb-types/src/timeseries/mod.rs

@@ -164,6 +164,7 @@ mod tests {
             interval_ms: 86_400_000,
             last_flushed_wal_lsn: 42,
             column_stats: HashMap::new(),
+            max_system_ts: 0,
         };
         assert!(meta.is_queryable());
         assert!(meta.overlaps(&TimeRange::new(1500, 2500)));
@@ -182,6 +183,7 @@ mod tests {
             interval_ms: 0,
             last_flushed_wal_lsn: 0,
             column_stats: HashMap::new(),
+            max_system_ts: 0,
         };
         assert!(!meta.is_queryable());
     }

nodedb-types/src/timeseries/partition.rs

@@ -66,6 +66,12 @@ pub struct PartitionMeta {
     /// Per-column statistics (codec, min/max/sum/count/cardinality).
     #[serde(default, skip_serializing_if = "HashMap::is_empty")]
     pub column_stats: HashMap<String, nodedb_codec::ColumnStatistics>,
+    /// Maximum `_ts_system` value across rows in this partition
+    /// (bitemporal only; 0 for non-bitemporal partitions). Retention on
+    /// bitemporal collections uses this instead of `max_ts` so that
+    /// late-arriving backfill survives an event-time-based TTL.
+    #[serde(default)]
+    pub max_system_ts: i64,
 }
 
 impl PartitionMeta {

nodedb/src/data/executor/dispatch/other.rs

@@ -327,13 +327,25 @@ impl CoreLoop {
                 let mut deleted = 0usize;
                 let ts_base = self.data_dir.join("ts").join(collection.as_str());
 
+                // Bitemporal collections key retention on system-time
+                // (`max_system_ts`) so backfill with old event-time but
+                // current write-time survives an event-time-based TTL.
+                // Non-bitemporal partitions have `max_system_ts == 0` and
+                // fall through to the existing `max_ts` path.
+                let bitemporal = self.is_bitemporal(task.request.tenant_id.as_u32(), collection);
+
                 let ts_key = (task.request.tenant_id, collection.to_string());
                 if let Some(registry) = self.ts_registries.get_mut(&ts_key) {
                     // Find partitions older than cutoff.
                     let expired: Vec<(i64, String)> = registry
                         .iter()
                         .filter(|(_, e)| {
-                            e.meta.max_ts < cutoff
+                            let axis_ts = if bitemporal && e.meta.max_system_ts > 0 {
+                                e.meta.max_system_ts
+                            } else {
+                                e.meta.max_ts
+                            };
+                            axis_ts < cutoff
                                 && e.meta.state != nodedb_types::timeseries::PartitionState::Deleted
                         })
                         .map(|(&start, e)| (start, e.dir_name.clone()))

nodedb/src/data/executor/handlers/compact.rs

@@ -261,7 +261,23 @@ impl CoreLoop {
         let max_per_pass = self.segment_compaction_config.max_segments_per_pass;
         let mut total_merged = 0;
 
-        for ((_, collection), registry) in &mut self.ts_registries {
+        // Snapshot bitemporal flags per collection up front so the
+        // mutable-borrow loop below can query them without re-borrowing `self`.
+        let bitemporal_flags: std::collections::HashMap<(crate::types::TenantId, String), bool> =
+            self.ts_registries
+                .keys()
+                .map(|(tid, col)| {
+                    let flag = self.is_bitemporal(tid.as_u32(), col);
+                    ((*tid, col.clone()), flag)
+                })
+                .collect();
+
+        for ((tid, collection), registry) in &mut self.ts_registries {
+            let bitemporal = bitemporal_flags
+                .get(&(*tid, collection.clone()))
+                .copied()
+                .unwrap_or(false);
+
             // Find mergeable partition groups, limited by config.
             let mut groups = registry.find_mergeable(now_ms);
             if !force {
@@ -275,7 +291,7 @@ impl CoreLoop {
             }
 
             // Retention: find and purge expired partitions.
-            let expired = registry.find_expired(now_ms);
+            let expired = registry.find_expired(now_ms, bitemporal);
             for start_ts in &expired {
                 registry.mark_deleted(*start_ts);
             }

nodedb/src/engine/timeseries/columnar_memtable.rs

@@ -64,6 +64,11 @@ impl ColumnarSchema {
             .copied()
             .unwrap_or(nodedb_codec::ColumnCodec::Auto)
     }
+
+    /// Index of the reserved `_ts_system` column, or `None` for non-bitemporal.
+    pub fn ts_system_idx(&self) -> Option<usize> {
+        self.columns.iter().position(|(n, _)| n == "_ts_system")
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -455,13 +460,29 @@ impl ColumnarMemtable {
             }
         }
 
+        // Scan `_ts_system` column (if present) for retention's system-time axis.
+        // For non-bitemporal partitions this is 0 and retention falls through
+        // to the existing `max_ts` (event-time) path.
+        let max_system_ts = self
+            .schema
+            .ts_system_idx()
+            .and_then(|idx| drained_columns.get(idx))
+            .map(|col| match col {
+                ColumnData::Timestamp(v) | ColumnData::Int64(v) => {
+                    v.iter().copied().max().unwrap_or(0)
+                }
+                _ => 0,
+            })
+            .unwrap_or(0);
+
         let result = ColumnarDrainResult {
             columns: drained_columns,
             schema: self.schema.clone(),
             symbol_dicts: drained_dicts,
             row_count: self.row_count,
             min_ts: self.min_ts,
             max_ts: self.max_ts,
+            max_system_ts,
             series_row_counts: std::mem::take(&mut self.series_row_counts),
         };
 
@@ -584,6 +605,10 @@ pub struct ColumnarDrainResult {
     pub row_count: u64,
     pub min_ts: i64,
     pub max_ts: i64,
+    /// Maximum `_ts_system` value across rows (bitemporal only; 0 otherwise).
+    /// Populated on drain by scanning the `_ts_system` column so retention
+    /// can evaluate system-time staleness without re-reading the segment.
+    pub max_system_ts: i64,
     pub series_row_counts: HashMap<SeriesId, u64>,
 }
 

nodedb/src/engine/timeseries/columnar_segment/writer.rs

@@ -98,6 +98,7 @@ impl ColumnarSegmentWriter {
             interval_ms,
             last_flushed_wal_lsn: flush_wal_lsn,
             column_stats,
+            max_system_ts: drain.max_system_ts,
         };
 
         let meta_json = sonic_rs::to_vec(&meta)

nodedb/src/engine/timeseries/merge/o3.rs

@@ -99,6 +99,9 @@ pub fn merge_o3_into_partition(
                 .max()
                 .unwrap_or(i64::MIN),
         ),
+        // O3 merge preserves the existing partition's system-time max;
+        // O3Row does not carry `_ts_system` (O3 is a value-only buffer).
+        max_system_ts: existing_meta.max_system_ts,
         series_row_counts: std::collections::HashMap::new(),
     };
 

nodedb/src/engine/timeseries/merge/partitions.rs

@@ -102,6 +102,17 @@ pub fn merge_partitions(
     // Write merged partition.
     let writer = ColumnarSegmentWriter::new(base_dir);
 
+    // Preserve the merged partition's system-time max so retention's
+    // bitemporal path sees the true latest write time, not just event-time.
+    let max_system_ts = ref_schema
+        .ts_system_idx()
+        .and_then(|idx| merged_columns.get(idx))
+        .map(|col| match col {
+            ColumnData::Timestamp(v) | ColumnData::Int64(v) => v.iter().copied().max().unwrap_or(0),
+            _ => 0,
+        })
+        .unwrap_or(0);
+
     // Build a drain-like result for the writer.
     let drain = crate::engine::timeseries::columnar_memtable::ColumnarDrainResult {
         columns: merged_columns,
@@ -110,6 +121,7 @@ pub fn merge_partitions(
         row_count: total_rows,
         min_ts: global_min_ts,
         max_ts: global_max_ts,
+        max_system_ts,
         series_row_counts: std::collections::HashMap::new(),
     };