From 174bfe4e098311d3d9dc572e3818ff3bd45c146a Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 12 May 2026 21:12:34 -0600 Subject: [PATCH] feat(data_plane): wire DualWriteSink into production main MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 5 M2.3.4b — replaces the production `StoreOutputSink` in `main.rs` with `DualWriteSink` (introduced in PR #156). Every precompute the worker emits now lands in BOTH the legacy `SketchStore` (which still serves queries today) AND the new sid-keyed `SketchIndex` via `append_precompute`. The legacy query path is unchanged — SketchIndex is being populated but no consumer reads precompute payloads from it yet. That's M2.3.5. Per-row overhead is one snapshot read + per-row agg-id lookup + sid hash + Box clone, all paid on the worker thread. Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/lib.rs | 2 +- data_plane/src/main.rs | 14 +++++++++++--- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 590beedc3..f0015984f 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -26,7 +26,7 @@ pub use query_engines::{ASAPQueryEngine, InstantVector, QueryResult}; pub use drivers::{HttpServer, HttpServerConfig, OtlpReceiver, OtlpReceiverConfig}; pub use precompute_engine::config::{LateDataPolicy, PrecomputeEngineConfig}; -pub use precompute_engine::output_sink::StoreOutputSink; +pub use precompute_engine::output_sink::{DualWriteSink, StoreOutputSink}; pub use precompute_engine::PrecomputeEngine; pub use utils::read_streaming_config; diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 6abbba6ae..50ffdcec0 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -22,8 +22,8 @@ use data_plane::precompute_engine::config::LateDataPolicy; use data_plane::precompute_engine::PrecomputeWorkerDiagnostics; use data_plane::utils::file_io::read_streaming_config; use data_plane::{ - HttpServer, HttpServerConfig, OtlpReceiver, OtlpReceiverConfig, PrecomputeEngine, - PrecomputeEngineConfig, Result, ASAPQueryEngine, SketchStore, StoreOutputSink, + DualWriteSink, HttpServer, HttpServerConfig, OtlpReceiver, OtlpReceiverConfig, + PrecomputeEngine, PrecomputeEngineConfig, Result, ASAPQueryEngine, SketchStore, }; #[derive(Parser, Debug)] @@ -435,7 +435,15 @@ async fn main() -> Result<()> { wall_clock_grace_period_ms: 5_000, schema_persist_path: args.schema_persist_path.clone(), }; - let output_sink = Arc::new(StoreOutputSink::new(store.clone())); + // M2.3.4b — DualWriteSink mirrors precompute writes into the + // legacy SketchStore (still serves queries) AND the new + // SketchIndex (sid-keyed precompute path; query path migration + // is M2.3.5). Replaces StoreOutputSink. + let output_sink = Arc::new(DualWriteSink::new( + store.clone(), + sketch_index.clone(), + hot_reload_config.clone(), + )); let engine = PrecomputeEngine::new( precompute_config, hot_reload_config.clone(),