fix: resolve build errors in precompute engine (E0432, E0609, E0061) and apply fmt

- Restore decode_prometheus_remote_write and related types that were
  fully commented out in prometheus_remote_write.rs
- Fix CountMinSketchAccumulator field access via .inner. in accumulator_factory
- Fix AggregationConfig::new call in worker test (missing slide_interval arg,
  wrong types for spatial_filter/metric, extra trailing None)
- Apply cargo fmt across precompute engine files

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

Author: zzylol

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

@@ -1,4 +1,5 @@
 use clap::Parser;
+use query_engine_rust::data_model::QueryLanguage;
 use query_engine_rust::data_model::{
     CleanupPolicy, InferenceConfig, LockStrategy, StreamingConfig,
 };
@@ -9,7 +10,6 @@ use query_engine_rust::precompute_engine::output_sink::{RawPassthroughSink, Stor
 use query_engine_rust::precompute_engine::PrecomputeEngine;
 use query_engine_rust::stores::SimpleMapStore;
 use query_engine_rust::{HttpServer, HttpServerConfig};
-use query_engine_rust::data_model::QueryLanguage;
 use std::sync::Arc;
 use tracing::info;
 use tracing_subscriber::fmt::format::FmtSpan;
@@ -85,20 +85,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     );
 
     // Create the store
-    let store: Arc<dyn query_engine_rust::stores::Store> = Arc::new(
-        SimpleMapStore::new_with_strategy(
+    let store: Arc<dyn query_engine_rust::stores::Store> =
+        Arc::new(SimpleMapStore::new_with_strategy(
             streaming_config.clone(),
             CleanupPolicy::CircularBuffer,
             args.lock_strategy,
-        ),
-    );
+        ));
 
     // Optionally start the query HTTP server
     if args.query_port > 0 {
-        let inference_config = InferenceConfig::new(
-            QueryLanguage::promql,
-            CleanupPolicy::CircularBuffer,
-        );
+        let inference_config =
+            InferenceConfig::new(QueryLanguage::promql, CleanupPolicy::CircularBuffer);
         let query_engine = Arc::new(SimpleEngine::new(
             store.clone(),
             inference_config,

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

@@ -78,8 +78,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         .init();
 
     // Load configs the same way main.rs does
-    let inference_config =
-        read_inference_config("examples/promql/inference_config.yaml", QueryLanguage::promql)?;
+    let inference_config = read_inference_config(
+        "examples/promql/inference_config.yaml",
+        QueryLanguage::promql,
+    )?;
     println!(
         "Loaded inference config with {} query configs",
         inference_config.query_configs.len()
@@ -101,13 +103,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     println!("\n=== Starting precompute engine (ingest={INGEST_PORT}, query={QUERY_PORT}) ===");
 
     // Create store
-    let store: Arc<dyn query_engine_rust::stores::Store> = Arc::new(
-        SimpleMapStore::new_with_strategy(
+    let store: Arc<dyn query_engine_rust::stores::Store> =
+        Arc::new(SimpleMapStore::new_with_strategy(
             streaming_config.clone(),
             cleanup_policy,
             LockStrategy::PerKey,
-        ),
-    );
+        ));
 
     // Start query server
     let query_engine = Arc::new(SimpleEngine::new(
@@ -207,9 +208,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 
     // Use the exact query pattern from inference_config
     let queries_instant = vec![
-        ("quantile by (label_0) (0.99, fake_metric)", "10", "Configured query at t=10"),
-        ("quantile by (label_0) (0.99, fake_metric)", "15", "Configured query at t=15"),
-        ("sum_over_time(fake_metric[1s])", "10", "Temporal: sum_over_time at t=10"),
+        (
+            "quantile by (label_0) (0.99, fake_metric)",
+            "10",
+            "Configured query at t=10",
+        ),
+        (
+            "quantile by (label_0) (0.99, fake_metric)",
+            "15",
+            "Configured query at t=15",
+        ),
+        (
+            "sum_over_time(fake_metric[1s])",
+            "10",
+            "Temporal: sum_over_time at t=10",
+        ),
         ("sum(fake_metric)", "10", "Spatial: sum at t=10"),
     ];
 
@@ -272,11 +285,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         raw_mode_aggregation_id: raw_agg_id,
     };
     let raw_sink = Arc::new(RawPassthroughSink::new(store.clone()));
-    let raw_engine = PrecomputeEngine::new(
-        raw_engine_config,
-        streaming_config.clone(),
-        raw_sink,
-    );
+    let raw_engine = PrecomputeEngine::new(raw_engine_config, streaming_config.clone(), raw_sink);
     tokio::spawn(async move {
         if let Err(e) = raw_engine.run().await {
             eprintln!("Raw precompute engine error: {e}");
@@ -299,20 +308,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
             .body(body)
             .send()
             .await?;
-        println!("  Sent raw t={ts}ms v={val} -> HTTP {}", resp.status().as_u16());
+        println!(
+            "  Sent raw t={ts}ms v={val} -> HTTP {}",
+            resp.status().as_u16()
+        );
     }
 
     // Short wait for processing (no watermark advancement needed)
     tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
 
     // Verify raw samples appeared in the store
     println!("\n=== Verifying raw samples in store ===");
-    let results = store.query_precomputed_output(
-        "fake_metric",
-        raw_agg_id,
-        100_000,
-        103_000,
-    )?;
+    let results = store.query_precomputed_output("fake_metric", raw_agg_id, 100_000, 103_000)?;
     let total_buckets: usize = results.values().map(|v| v.len()).sum();
     println!("  Found {total_buckets} buckets for aggregation_id={raw_agg_id} in [100000, 103000)");
     assert!(