Remove test-only query planning and unused backend APIs and dependencies

Cargo.toml

@@ -40,6 +40,7 @@ chrono = { version = "0.4", features = ["serde"] }
 promql-parser = { git = "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/ProjectASAP/promql-parser" }
 tokio = { version = "1.0", features = ["full"] }
 arc-swap = "1.7"
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 
 # Internal crates
 asap_types = { path = "crates/asap_types" }

control_plane/Cargo.toml

@@ -19,7 +19,7 @@ serde       = { version = "1",    features = ["derive"] }
 serde_json  = "1"
 sha2 = "0.10"
 serde_yaml  = "0.9"
-reqwest     = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+reqwest.workspace = true
 anyhow      = "1"
 thiserror   = "1"
 tracing     = "0.1"
@@ -28,11 +28,9 @@ chrono      = { version = "0.4",  features = ["serde"] }
 promql-parser.workspace = true
 prost          = "0.13"
 bytes          = "1"
-zstd           = "0.13"
 parking_lot    = "0.12"
 prometheus     = { version = "0.13", default-features = false, features = ["process"] }
 tonic          = { version = "0.12", features = ["gzip"] }
-tokio-stream   = { version = "0.1", features = ["net"] }
 asap_types.workspace = true
 
 # Planner's IR, replacement search, and frontends share the workspace pin.

control_plane/examples/offline_frequency_plan.rs

@@ -8,7 +8,7 @@ use asap_aware_mapping::empirical_comparison::{
 use control_plane::{
     physical::post_asap::{bind_query_expr_with_cost_model, cost_model::ControlPlaneCostModel},
     planner_selection::frequency,
-    types_v2::AccuracyTarget,
+    types::AccuracyTarget,
 };
 use planner_types::pre_asap::{AggIntent, Column, DataType, QueryExpr, Reduction, Schema, Source};
 use serde_json::json;

control_plane/examples/offline_planner_replay.rs

@@ -11,7 +11,7 @@ use control_plane::{
         PostAsapPlan,
     },
     query_parser::parse_query_expr_canonical,
-    types_v2::AccuracyTarget,
+    types::AccuracyTarget,
 };
 use planner_types::post_asap::{SummaryExpr, SummaryFamilyType, SummaryNode};
 use serde_json::{json, Value};

control_plane/src/asap_tier_implement.rs

@@ -77,7 +77,7 @@ use asap_aware_mapping::DefaultCostModel;
 use planner_types::post_asap::SummaryNode;
 
 use crate::query_parser::parse_query_expr_canonical;
-use crate::types_v2::AccuracyTarget;
+use crate::types::AccuracyTarget;
 use planner_types::pre_asap::QueryExpr;
 
 /// Fixed accuracy target for this L1 call site (L1 adoption,

control_plane/src/backend_client.rs

@@ -17,7 +17,9 @@ use std::time::Duration;
 
 use anyhow::{Context, Result};
 use reqwest::Client;
-use tracing::{debug, warn};
+use tracing::debug;
+#[cfg(test)]
+use tracing::warn;
 
 /// Classification of an HTTP push failure used by the retry layer in
 /// [`crate::emit::backend_push`]. Transient errors are safe to retry
@@ -167,6 +169,7 @@ impl BackendClient {
         }
     }
 
+    #[cfg(test)]
     /// Post streaming configuration as `application/json`. The backend accepts
     /// both JSON and YAML; a non-2xx response is an error for the caller to log.
     pub async fn post_streaming_config_json(&self, json: String) -> Result<()> {
@@ -197,6 +200,7 @@ impl BackendClient {
         }
     }
 
+    #[cfg(test)]
     /// Typed sibling of [`Self::post_streaming_config_json`] for the
     /// retry layer. Returns the same `Ok(())` on 2xx, but on failure
     /// classifies the underlying cause as [`BackendPostError::Transient`]
@@ -239,41 +243,7 @@ impl BackendClient {
         }
     }
 
-    /// Typed sibling of [`Self::post_storage_routing_json`] for the
-    /// retry layer. Identical contract to
-    /// [`Self::post_streaming_config_json_typed`].
-    pub async fn post_storage_routing_json_typed(
-        &self,
-        json: String,
-    ) -> std::result::Result<(), BackendPostError> {
-        let url = derive_storage_routing_url(&self.endpoint);
-        debug!(
-            endpoint = %url,
-            json_bytes = json.len(),
-            "posting storage-routing JSON to ASAPQuery-backend (typed)"
-        );
-        let resp = self
-            .http
-            .post(&url)
-            .header("content-type", "application/json")
-            .body(json)
-            .send()
-            .await
-            .map_err(classify_reqwest_error)?;
-
-        let status = resp.status();
-        if status.is_success() {
-            Ok(())
-        } else {
-            let body = resp.text().await.unwrap_or_default();
-            Err(classify_http_status(
-                status,
-                body,
-                "storage-routing JSON POST",
-            ))
-        }
-    }
-
+    #[cfg(test)]
     /// Post backend storage-routing JSON. Derive the URL by replacing the
     /// `/api/v1/streaming-config` suffix with `/api/v1/storage_routing`; URLs
     /// without that suffix are used verbatim.
@@ -409,6 +379,7 @@ fn derive_physical_plan_url(endpoint: &str) -> String {
 /// end with `/api/v1/streaming-config` (or `/api/v1/streaming_config` —
 /// either spelling is supported) pass through unchanged so tests can
 /// inject a mock-server URL directly.
+#[cfg(test)]
 fn derive_storage_routing_url(endpoint: &str) -> String {
     const STREAMING_PATH_DASH: &str = "/api/v1/streaming-config";
     const STREAMING_PATH_UNDERSCORE: &str = "/api/v1/streaming_config";
@@ -422,6 +393,7 @@ fn derive_storage_routing_url(endpoint: &str) -> String {
     endpoint.to_string()
 }
 
+#[cfg(test)]
 /// Fire-and-forget convenience helper used by the replanner. Logs
 /// errors at WARN and never propagates them — the replanner should
 /// never fail an entire replan because the backend was temporarily

control_plane/src/clickhouse.rs

@@ -72,6 +72,7 @@ pub async fn plan_clickhouse_sql(
     })
 }
 
+#[cfg(test)]
 pub async fn canonicalize_clickhouse_sql(
     sql: &str,
     catalog: &SqlCatalog,

control_plane/src/emit/mod.rs

@@ -1131,7 +1131,7 @@ mod runtime_tests {
                 time_window: Duration::from_secs(60),
                 repeat_every: None,
                 accuracy_sla: 0.01,
-                accuracy: crate::types_v2::AccuracyTarget::Epsilon(0.01),
+                accuracy: crate::types::AccuracyTarget::Epsilon(0.01),
                 latency_sla: None,
                 sketch_type_override: override_family,
                 exact_required: false,

control_plane/src/emit/monitor.rs

@@ -17,6 +17,7 @@
 //! `ASAPCollector/docs/continuous-monitoring-tumbling-cost-analysis.md`.
 
 pub use asap_types::MonitorFunctional as Functional;
+#[cfg(test)]
 use serde_yaml::{Mapping, Value};
 
 /// One monitored standing-query intent: "alert when the global Σ of `metric`'s
@@ -53,6 +54,7 @@ pub fn agg_id_for_metric(metric: &str) -> u64 {
     h
 }
 
+#[cfg(test)]
 /// Render the edge `threshold:` YAML mapping for the fused asap_edge processor's
 /// per-metric entry. Omits `key`/`coeffs` when not applicable to the functional.
 pub fn edge_threshold_block(intent: &MonitorIntent) -> Value {

control_plane/src/lib.rs

@@ -35,7 +35,6 @@ pub mod runtime_samples;
 pub mod sketch_selection;
 pub mod store;
 pub mod types;
-pub mod types_v2;
 pub mod workload;
 
 /// PromQL → ASAPPlanner's canonical post-ASAP plan via

control_plane/src/main.rs

@@ -14,7 +14,6 @@ use control_plane::replan;
 use control_plane::runtime_samples;
 use control_plane::store;
 use control_plane::types;
-use control_plane::types_v2;
 use control_plane::workload;
 
 use axum::{
@@ -520,7 +519,7 @@ struct PhysicalPlanQueryRequest {
     window_secs: u64,
     #[serde(default)]
     group_by: Vec<String>,
-    accuracy: types_v2::AccuracyTarget,
+    accuracy: types::AccuracyTarget,
     lifecycle: physical::compiler::LifecyclePlanningInput,
     window_implementations: Vec<physical::compiler::WindowImplementationCandidate>,
     #[serde(default)]
@@ -577,7 +576,7 @@ impl PhysicalQueryFrontend {
     fn parse(
         self,
         query: &str,
-        accuracy: types_v2::AccuracyTarget,
+        accuracy: types::AccuracyTarget,
     ) -> Result<planner_types::pre_asap::QueryExpr, String> {
         match self {
             Self::PromQl => parse_query_expr_canonical(query, accuracy)
@@ -2384,7 +2383,7 @@ mod api_tests {
     /// HTTP and stored replan inputs share the resolved typed target, including delta.
     #[tokio::test]
     async fn plan_preserves_typed_accuracy_requirements() {
-        use control_plane::types_v2::AccuracyTarget;
+        use control_plane::types::AccuracyTarget;
         for target in [
             AccuracyTarget::Epsilon(0.05),
             AccuracyTarget::EpsilonDelta {
@@ -2514,7 +2513,7 @@ mod api_tests {
             time_window: std::time::Duration::from_secs(300),
             repeat_every: None,
             accuracy_sla: 0.01,
-            accuracy: crate::types_v2::AccuracyTarget::Epsilon(0.01),
+            accuracy: crate::types::AccuracyTarget::Epsilon(0.01),
             latency_sla: None,
             sketch_type_override: None,
             exact_required: false,
@@ -2777,8 +2776,8 @@ mod api_tests {
             accuracy: None,
             dollars: None,
             deployment_model: None,
-            shape: types_v2::QueryShape::default(),
-            data: types_v2::DataShape::default(),
+            shape: types::QueryShape::default(),
+            data: types::DataShape::default(),
         };
         let wl = analyzer.analyze(spec).unwrap();
         let wc = types::WorkloadCharacteristics::default();
@@ -2922,8 +2921,8 @@ mod api_tests {
             accuracy: None,
             dollars: None,
             deployment_model: None,
-            shape: types_v2::QueryShape::default(),
-            data: types_v2::DataShape::default(),
+            shape: types::QueryShape::default(),
+            data: types::DataShape::default(),
         };
         let wl = analyzer.analyze(spec).unwrap();
         let wc = types::WorkloadCharacteristics::default();
@@ -3187,8 +3186,8 @@ mod api_tests {
             accuracy: None,
             dollars: None,
             deployment_model: None,
-            shape: types_v2::QueryShape::default(),
-            data: types_v2::DataShape::default(),
+            shape: types::QueryShape::default(),
+            data: types::DataShape::default(),
         };
         let wl = analyzer.analyze(spec).expect("analyze");
         let wc = types::WorkloadCharacteristics::default();
@@ -3422,8 +3421,8 @@ mod api_tests {
                 accuracy: None,
                 dollars: None,
                 deployment_model: None,
-                shape: types_v2::QueryShape::default(),
-                data: types_v2::DataShape::default(),
+                shape: types::QueryShape::default(),
+                data: types::DataShape::default(),
             };
             let wl = analyzer.analyze(spec).expect("analyze");
             let wc = types::WorkloadCharacteristics::default();

control_plane/src/opamp/mod.rs

@@ -245,14 +245,6 @@ impl OpampServer {
         }
     }
 
-    /// Broadcasts a config to every connected agent regardless of role.
-    pub async fn push_all(&self, cfg: RemoteConfig) {
-        let ids: Vec<String> = self.agents.read().await.keys().cloned().collect();
-        for id in ids {
-            self.push(&id, cfg.clone()).await;
-        }
-    }
-
     /// Broadcasts a config only to agents matching the given role.
     pub async fn push_to_role(&self, role: AgentRole, cfg: RemoteConfig) {
         let ids: Vec<String> = self

control_plane/src/physical/colored_dag/allocator.rs

@@ -45,7 +45,6 @@ use crate::physical::colored_dag::dag::{ColoredDag, ColoredNode, NodeId};
 use crate::physical::colored_dag::stage_id::{StageId, Topology};
 use crate::physical::post_asap::deployment_expr::PostAsapPlan;
 use crate::physical::post_asap::PhysicalExpr;
-use crate::types_v2::BindingName;
 
 /// Errors surfaced by [`StageAllocator::allocate`].
 #[derive(Debug, thiserror::Error, PartialEq)]
@@ -343,17 +342,6 @@ impl ThreeStageWalker {
     }
 }
 
-// Convenience helper used by tests / external callers that only need a
-// stage lookup keyed by binding name.
-pub(crate) fn binding_stage(dag: &ColoredDag, name: &BindingName) -> Option<StageId> {
-    dag.nodes.iter().find_map(|n| match &n.expr {
-        PhysicalExpr::Committed(PostAsapPlan::LetBinding { name: n2, .. }) if n2 == name => {
-            Some(n.stage)
-        }
-        _ => None,
-    })
-}
-
 // ── Tests ─────────────────────────────────────────────────────────────────────
 
 #[cfg(test)]
@@ -417,7 +405,7 @@ mod tests {
             measures: vec![planner_types::pre_asap::AggIntent::Quantile {
                 col: None,
                 q: 0.99,
-                accuracy: crate::types_v2::AccuracyTarget::Epsilon(0.01),
+                accuracy: crate::types::AccuracyTarget::Epsilon(0.01),
             }],
             output_names: Vec::new(),
             having: None,

control_plane/src/physical/colored_dag/dag.rs

@@ -79,11 +79,6 @@ impl ColoredDag {
         self.nodes.first()
     }
 
-    /// All nodes painted with `stage`.
-    pub fn nodes_in_stage(&self, stage: StageId) -> impl Iterator<Item = &ColoredNode> {
-        self.nodes.iter().filter(move |n| n.stage == stage)
-    }
-
     /// Set of `StageId`s actually present in this colouring (subset of
     /// `topology.stages()`).
     pub fn occupied_stages(&self) -> Vec<StageId> {
@@ -96,6 +91,7 @@ impl ColoredDag {
         seen
     }
 
+    #[cfg(test)]
     /// Edges crossing stage boundaries. They describe the required transport hops.
     pub fn cut_edges(&self) -> Vec<(NodeId, NodeId)> {
         self.edges
@@ -175,7 +171,7 @@ mod tests {
             measures: vec![planner_types::pre_asap::AggIntent::Quantile {
                 col: None,
                 q: 0.99,
-                accuracy: crate::types_v2::AccuracyTarget::Epsilon(0.01),
+                accuracy: crate::types::AccuracyTarget::Epsilon(0.01),
             }],
             output_names: Vec::new(),
             having: None,

control_plane/src/physical/colored_dag/emitter.rs

@@ -13,7 +13,6 @@ use serde::{Deserialize, Serialize};
 use crate::physical::colored_dag::dag::ColoredDag;
 use crate::physical::colored_dag::stage_id::{StageId, Topology};
 use crate::physical::post_asap::deployment_expr::{PhysicalExpr, PostAsapPlan};
-use crate::types_v2::BindingName;
 use planner_types::post_asap::{
     ExactKind, ExactParams, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams,
     SketchQuery, SummaryExpr, SummaryFamilyType,
@@ -45,10 +44,10 @@ enum NodeKind<'a> {
     },
     SketchMerge,
     LetBinding {
-        name: &'a BindingName,
+        name: &'a String,
     },
     Ref {
-        name: &'a BindingName,
+        name: &'a String,
     },
     RawAtEdgeSketchAtBackend {
         family: &'a SketchAlgorithm,
@@ -447,14 +446,7 @@ pub enum ColdFormat {
     Intchunk,
 }
 
-impl ColdFormat {
-    /// `true` for the default ([`ColdFormat::Fragment`]). Drives the
-    /// `skip_serializing_if` on [`EdgeStageConfig::cold_format`] so an unset
-    /// format leaves the serialized config byte-identical to today.
-    pub fn is_default(&self) -> bool {
-        matches!(self, ColdFormat::Fragment)
-    }
-}
+impl ColdFormat {}
 
 /// Named default for [`EdgeStageConfig::cold_ship_endpoint`].
 ///