Add pane-based incremental sliding window computation

Route each sample to exactly 1 pane (sub-window of size slide_interval)
instead of W overlapping window accumulators. When a window closes, merge
its constituent panes via AggregateCore::merge_with(). This reduces
per-sample accumulator updates from W to 1 for sliding windows.

- Add snapshot_accumulator() to AccumulatorUpdater trait (9 implementations)
- Add pane_start_for(), panes_for_window(), slide_interval_ms() to WindowManager
- Replace active_windows HashMap with active_panes BTreeMap in worker
- Rewrite sample routing and window close logic with pane merging
- Extract PrecomputeEngine to engine.rs, ingest handlers to ingest_handler.rs
- Update design doc with pane-based architecture

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: zzylol

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

@@ -20,6 +20,10 @@ pub trait AccumulatorUpdater: Send {
     /// Extract the final accumulator as a boxed `AggregateCore`.
     fn take_accumulator(&mut self) -> Box<dyn AggregateCore>;
 
+    /// Non-destructive read of the current accumulator state (clone without reset).
+    /// Used by pane-based sliding windows to read shared panes.
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore>;
+
     /// Reset internal state for reuse (avoids re-allocation).
     fn reset(&mut self);
 
@@ -61,6 +65,10 @@ impl AccumulatorUpdater for SumAccumulatorUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = SumAccumulator::new();
     }
@@ -107,6 +115,10 @@ impl AccumulatorUpdater for MinMaxAccumulatorUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = MinMaxAccumulator::new(self.sub_type.clone());
     }
@@ -163,6 +175,18 @@ impl AccumulatorUpdater for IncreaseAccumulatorUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        match &self.acc {
+            Some(acc) => Box::new(acc.clone()),
+            None => Box::new(IncreaseAccumulator::new(
+                Measurement::new(0.0),
+                0,
+                Measurement::new(0.0),
+                0,
+            )),
+        }
+    }
+
     fn reset(&mut self) {
         self.acc = None;
     }
@@ -209,6 +233,10 @@ impl AccumulatorUpdater for KllAccumulatorUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = DatasketchesKLLAccumulator::new(self.k);
     }
@@ -254,6 +282,10 @@ impl AccumulatorUpdater for MultipleSumUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = MultipleSumAccumulator::new();
     }
@@ -301,6 +333,10 @@ impl AccumulatorUpdater for MultipleMinMaxUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = MultipleMinMaxAccumulator::new(self.sub_type.clone());
     }
@@ -360,6 +396,10 @@ impl AccumulatorUpdater for MultipleIncreaseUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = MultipleIncreaseAccumulator::new();
     }
@@ -428,6 +468,10 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = CountMinSketchAccumulator::new(self.row_num, self.col_num);
     }
@@ -479,6 +523,10 @@ impl AccumulatorUpdater for HydraKllAccumulatorUpdater {
         result
     }
 
+    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore> {
+        Box::new(self.acc.clone())
+    }
+
     fn reset(&mut self) {
         self.acc = HydraKllSketchAccumulator::new(self.row_num, self.col_num, self.k);
     }

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

@@ -0,0 +1,125 @@
+use crate::data_model::StreamingConfig;
+use crate::precompute_engine::config::PrecomputeEngineConfig;
+use crate::precompute_engine::ingest_handler::{
+    handle_prometheus_ingest, handle_victoriametrics_ingest, IngestState,
+};
+use crate::precompute_engine::output_sink::OutputSink;
+use crate::precompute_engine::series_router::{SeriesRouter, WorkerMessage};
+use crate::precompute_engine::worker::Worker;
+use axum::{routing::post, Router};
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tokio::sync::mpsc;
+use tracing::{info, warn};
+
+/// The top-level precompute engine orchestrator.
+///
+/// Creates worker threads, the series router, and the Axum ingest server.
+pub struct PrecomputeEngine {
+    config: PrecomputeEngineConfig,
+    streaming_config: Arc<StreamingConfig>,
+    output_sink: Arc<dyn OutputSink>,
+}
+
+impl PrecomputeEngine {
+    pub fn new(
+        config: PrecomputeEngineConfig,
+        streaming_config: Arc<StreamingConfig>,
+        output_sink: Arc<dyn OutputSink>,
+    ) -> Self {
+        Self {
+            config,
+            streaming_config,
+            output_sink,
+        }
+    }
+
+    /// Start the precompute engine. This spawns worker tasks and the HTTP
+    /// ingest server, then blocks until shutdown.
+    pub async fn run(self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let num_workers = self.config.num_workers;
+        let channel_size = self.config.channel_buffer_size;
+
+        // Build MPSC channels for each worker
+        let mut senders = Vec::with_capacity(num_workers);
+        let mut receivers = Vec::with_capacity(num_workers);
+        for _ in 0..num_workers {
+            let (tx, rx) = mpsc::channel::<WorkerMessage>(channel_size);
+            senders.push(tx);
+            receivers.push(rx);
+        }
+
+        // Build the router
+        let router = SeriesRouter::new(senders);
+
+        // Build aggregation config map from streaming config
+        let agg_configs: HashMap<u64, _> =
+            self.streaming_config.get_all_aggregation_configs().clone();
+
+        // Spawn workers
+        let mut worker_handles = Vec::with_capacity(num_workers);
+        for (id, rx) in receivers.into_iter().enumerate() {
+            let worker = Worker::new(
+                id,
+                rx,
+                self.output_sink.clone(),
+                agg_configs.clone(),
+                self.config.max_buffer_per_series,
+                self.config.allowed_lateness_ms,
+                self.config.pass_raw_samples,
+                self.config.raw_mode_aggregation_id,
+                self.config.late_data_policy,
+            );
+            let handle = tokio::spawn(async move {
+                worker.run().await;
+            });
+            worker_handles.push(handle);
+        }
+
+        info!(
+            "PrecomputeEngine started with {} workers on port {}",
+            num_workers, self.config.ingest_port
+        );
+
+        // Build the ingest state
+        let ingest_state = Arc::new(IngestState {
+            router,
+            samples_ingested: std::sync::atomic::AtomicU64::new(0),
+        });
+
+        // Start flush timer
+        let flush_state = ingest_state.clone();
+        let flush_interval_ms = self.config.flush_interval_ms;
+        tokio::spawn(async move {
+            let mut interval =
+                tokio::time::interval(tokio::time::Duration::from_millis(flush_interval_ms));
+            loop {
+                interval.tick().await;
+                if let Err(e) = flush_state.router.broadcast_flush().await {
+                    warn!("Flush broadcast error: {}", e);
+                    break;
+                }
+            }
+        });
+
+        // Start the Axum HTTP server for ingest (Prometheus + VictoriaMetrics)
+        let app = Router::new()
+            .route("/api/v1/write", post(handle_prometheus_ingest))
+            .route("/api/v1/import", post(handle_victoriametrics_ingest))
+            .with_state(ingest_state);
+
+        let addr = format!("0.0.0.0:{}", self.config.ingest_port);
+        info!("Ingest server listening on {}", addr);
+
+        let listener = TcpListener::bind(&addr).await?;
+        axum::serve(listener, app).await?;
+
+        // Wait for workers to finish (this only happens on shutdown)
+        for handle in worker_handles {
+            let _ = handle.await;
+        }
+
+        Ok(())
+    }
+}

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

@@ -0,0 +1,89 @@
+use crate::drivers::ingest::prometheus_remote_write::decode_prometheus_remote_write;
+use crate::drivers::ingest::victoriametrics_remote_write::decode_victoriametrics_remote_write;
+use crate::precompute_engine::series_router::SeriesRouter;
+use axum::{body::Bytes, extract::State, http::StatusCode};
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Instant;
+use tracing::warn;
+
+/// Shared state for the ingest HTTP handler.
+pub(crate) struct IngestState {
+    pub(crate) router: SeriesRouter,
+    pub(crate) samples_ingested: std::sync::atomic::AtomicU64,
+}
+
+/// Shared logic: group decoded samples by series key and route to workers.
+async fn route_decoded_samples(
+    state: &IngestState,
+    samples: Vec<crate::drivers::ingest::prometheus_remote_write::DecodedSample>,
+    ingest_received_at: Instant,
+) -> StatusCode {
+    if samples.is_empty() {
+        return StatusCode::NO_CONTENT;
+    }
+
+    let count = samples.len() as u64;
+    state
+        .samples_ingested
+        .fetch_add(count, std::sync::atomic::Ordering::Relaxed);
+
+    // Group samples by series key for batch routing
+    let mut by_series: HashMap<&str, Vec<(i64, f64)>> = HashMap::new();
+    for s in &samples {
+        by_series
+            .entry(&s.labels)
+            .or_default()
+            .push((s.timestamp_ms, s.value));
+    }
+
+    // Convert to owned keys for batch routing
+    let by_series_owned: HashMap<String, Vec<(i64, f64)>> = by_series
+        .into_iter()
+        .map(|(k, v)| (k.to_string(), v))
+        .collect();
+
+    // Route all series to workers concurrently
+    if let Err(e) = state
+        .router
+        .route_batch(by_series_owned, ingest_received_at)
+        .await
+    {
+        warn!("Batch routing error: {}", e);
+        return StatusCode::INTERNAL_SERVER_ERROR;
+    }
+
+    StatusCode::NO_CONTENT
+}
+
+/// Axum handler for Prometheus remote write (Snappy + Protobuf).
+pub(crate) async fn handle_prometheus_ingest(
+    State(state): State<Arc<IngestState>>,
+    body: Bytes,
+) -> StatusCode {
+    let ingest_received_at = Instant::now();
+    let samples = match decode_prometheus_remote_write(&body) {
+        Ok(s) => s,
+        Err(e) => {
+            warn!("Failed to decode Prometheus remote write: {}", e);
+            return StatusCode::BAD_REQUEST;
+        }
+    };
+    route_decoded_samples(&state, samples, ingest_received_at).await
+}
+
+/// Axum handler for VictoriaMetrics remote write (Zstd + Protobuf).
+pub(crate) async fn handle_victoriametrics_ingest(
+    State(state): State<Arc<IngestState>>,
+    body: Bytes,
+) -> StatusCode {
+    let ingest_received_at = Instant::now();
+    let samples = match decode_victoriametrics_remote_write(&body) {
+        Ok(s) => s,
+        Err(e) => {
+            warn!("Failed to decode VictoriaMetrics remote write: {}", e);
+            return StatusCode::BAD_REQUEST;
+        }
+    };
+    route_decoded_samples(&state, samples, ingest_received_at).await
+}