feat(sketch-db): Phase 2c — on-disk schema registry persistence

asap-query-engine/src/bin/bench_precompute_sketch.rs

@@ -174,6 +174,7 @@ async fn start_engine(
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        schema_persist_path: None,
     };
     let engine = PrecomputeEngine::new(
         config,

asap-query-engine/src/bin/e2e_quickstart_resource_test.rs

@@ -233,6 +233,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        schema_persist_path: None,
     };
     let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
     let engine = PrecomputeEngine::new(

asap-query-engine/src/bin/precompute_engine.rs

@@ -216,6 +216,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: args.pass_raw_samples,
         raw_mode_aggregation_id: args.raw_mode_aggregation_id,
         late_data_policy: args.late_data_policy,
+        schema_persist_path: None,
     };
 
     // Create the output sink (writes directly to the store)

asap-query-engine/src/bin/test_e2e_precompute.rs

@@ -150,6 +150,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        schema_persist_path: None,
     };
     let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
     let engine = PrecomputeEngine::new(
@@ -294,6 +295,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: true,
         raw_mode_aggregation_id: raw_agg_id,
         late_data_policy: LateDataPolicy::Drop,
+        schema_persist_path: None,
     };
     let raw_sink = Arc::new(RawPassthroughSink::new(store.clone()));
     let raw_engine = PrecomputeEngine::new(
@@ -646,6 +648,7 @@ async fn run_single_bench(
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        schema_persist_path: None,
     };
     let engine = PrecomputeEngine::new(
         engine_config,

asap-query-engine/src/main.rs

@@ -179,6 +179,14 @@ struct Args {
     #[arg(long, default_value = "10000")]
     precompute_channel_buffer_size: usize,
 
+    /// Optional path where the schema registry persists per-`agg_id`
+    /// lifecycle state (created_at / retired_at / expires_at) across
+    /// restarts (sketch DB Phase 2c). When unset, the registry is
+    /// memory-only and the §7 schema timeline loses all pre-restart
+    /// history.
+    #[arg(long)]
+    schema_persist_path: Option<std::path::PathBuf>,
+
     /// Enable automatic query tracking and planning
     #[arg(long)]
     enable_query_tracker: bool,
@@ -482,6 +490,7 @@ async fn main() -> Result<()> {
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
             late_data_policy: LateDataPolicy::Drop,
+            schema_persist_path: args.schema_persist_path.clone(),
         };
         let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
         let engine =

asap-query-engine/src/precompute_engine/config.rs

@@ -32,6 +32,15 @@ pub struct PrecomputeEngineConfig {
     pub raw_mode_aggregation_id: u64,
     /// Policy for handling late samples that arrive after their window has closed.
     pub late_data_policy: LateDataPolicy,
+    /// Optional path where the `SchemaRegistry` persists per-`agg_id`
+    /// lifecycle state across restarts (sketch DB Phase 2c). When
+    /// set, the registry loads prior `created_at_ms` / `retired_at_ms`
+    /// timestamps from the file on startup and atomically rewrites
+    /// the file after every config-driven reconcile. When `None`,
+    /// the registry is memory-only and the §7 timeline reflects only
+    /// post-restart history.
+    #[serde(default)]
+    pub schema_persist_path: Option<std::path::PathBuf>,
 }
 
 impl Default for PrecomputeEngineConfig {
@@ -46,6 +55,7 @@ impl Default for PrecomputeEngineConfig {
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
             late_data_policy: LateDataPolicy::Drop,
+            schema_persist_path: None,
         }
     }
 }

asap-query-engine/src/precompute_engine/engine.rs

@@ -77,11 +77,15 @@ impl PrecomputeEngine {
         // reconciliation onto the HTTP swap handler so it's
         // event-driven instead of per-batch.
         let initial_snapshot = hot_reload_config.snapshot();
-        let schemas = Arc::new(
-            crate::stores::sketch_db::SchemaRegistry::from_streaming_config(
+        let schemas = Arc::new(match config.schema_persist_path.as_ref() {
+            Some(path) => crate::stores::sketch_db::SchemaRegistry::load_or_new_from_config(
+                path.clone(),
                 initial_snapshot.as_ref(),
             ),
-        );
+            None => crate::stores::sketch_db::SchemaRegistry::from_streaming_config(
+                initial_snapshot.as_ref(),
+            ),
+        });
         let ingest_state = Arc::new(IngestState {
             router,
             samples_ingested: std::sync::atomic::AtomicU64::new(0),