diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs index 7d462beb..816bd286 100644 --- a/control_plane/src/optimizer/rules/mod.rs +++ b/control_plane/src/optimizer/rules/mod.rs @@ -5,29 +5,6 @@ use crate::types::*; pub const DEFAULT_VALID_FOR: Duration = Duration::from_secs(10 * 60); -/// Env-var that opts the planner into the typed L4 binding path -/// (`sketch_algebra::bind_query_expr`). Additive — when unset, the -/// existing untyped `algebra::directory::sketch_type_for_agg` path runs -/// unchanged. Phase E (stage_split refactor) is the natural migration -/// point at which the typed path becomes the only path. -/// -/// Set `USE_TYPED_SKETCH_ALGEBRA=1` to opt in. -#[allow(dead_code)] -pub const ENV_USE_TYPED_SKETCH_ALGEBRA: &str = "USE_TYPED_SKETCH_ALGEBRA"; - -/// Whether the typed L4 binding path is enabled for this process. -/// Reads the env var once per call (cheap; called per `plan()` invocation -/// at most). Phase C is additive — both code paths produce the same -/// `CollectionPlan` shape; the typed path is a *parallel* binding that -/// the planner can compare against the legacy path during development. -#[allow(dead_code)] -pub fn typed_sketch_algebra_enabled() -> bool { - matches!( - std::env::var(ENV_USE_TYPED_SKETCH_ALGEBRA).as_deref(), - Ok("1") | Ok("true") | Ok("yes") - ) -} - /// Bind a `QueryWorkload` into the typed L4 [`crate::sketch_algebra::PhysicalExpr`] /// IR, when callers want to inspect the typed binding alongside the /// legacy `CollectionPlan` output. diff --git a/control_plane/src/sketch_algebra/cost_model.rs b/control_plane/src/sketch_algebra/cost_model.rs index 8610c062..80e584ef 100644 --- a/control_plane/src/sketch_algebra/cost_model.rs +++ b/control_plane/src/sketch_algebra/cost_model.rs @@ -26,11 +26,12 @@ //! //! - `AggIntent::TopK { accuracy: AccuracyTarget::Exact, .. }` — routes to //! `exact_realization`, which has no accumulator form for `TopK` and -//! returns `PassThrough`, even though control_plane's own -//! `BindCountSketchOnTopK` still binds this shape (Tight recall tier → -//! `CountSketchWithHeap`). Still intercepted in `lower.rs` *before* -//! `implement_tree_in_with` runs — see that module's `bind_recursive` -//! for the pre-pass (ASAPController#151, still open). +//! returns `PassThrough`, so `implement_tree_in_with` falls through to +//! its own `Logical` fallback for this shape unchanged. There is no +//! local pre-pass binding it: the `BindCountSketchOnTopK` rule that +//! once did was deleted (see `lower.rs`'s module doc) — this is a +//! genuine, still-open `asap-plan` coverage gap (ASAPController#151), +//! not something this deployment routes around locally. #![allow(dead_code)] diff --git a/control_plane/src/sketch_algebra/mod.rs b/control_plane/src/sketch_algebra/mod.rs index c58dd754..ce8f655d 100644 --- a/control_plane/src/sketch_algebra/mod.rs +++ b/control_plane/src/sketch_algebra/mod.rs @@ -23,10 +23,11 @@ //! yet — they're gated on rules that haven't landed. Adding them is //! purely additive. //! -//! Wire-up state. The typed path is opt-in via the -//! `USE_TYPED_SKETCH_ALGEBRA` env var consulted by `planner::rules`; -//! existing call sites continue to use the legacy untyped binding path. -//! Phase E (stage_split) is the natural migration point. +//! Wire-up state. `bind_query_expr` runs unconditionally from `main.rs` +//! — there is no env-gate on this L3→L4 binding step itself. The one env +//! var in this area, `USE_TYPED_STAGE_SPLIT` +//! (`physical::stage_split::ENV_USE_TYPED_STAGE_SPLIT`), gates the +//! *downstream* L4→L5 stage-split step, not this module. #![allow(dead_code, unused_imports)] diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs index 0c3ee3d5..0f9b42ce 100644 --- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs +++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs @@ -368,7 +368,11 @@ impl<'a> SketchReducer<'a> { /// `is_cumulative` is the only genuinely function-name-derived /// signal (per-window vs `*_over_time` rollup), so the engine — which /// knows the original outer function — passes it explicitly. The - /// string [`Self::evaluate`] entry is retained for legacy callers. + /// string [`Self::evaluate`] entry has no production callers today — + /// `engine.rs` calls this typed entry exclusively — but is kept + /// because the query-path test suite (`tests.rs`) still exercises it + /// via PromQL function-name strings; it's a live test fixture, not + /// dead code. /// /// Returns `UnsupportedCapability` for `Capability::ExactAgg(_)` /// (served by the exact-agg dispatch path, not the sketch reducer) @@ -746,10 +750,15 @@ impl<'a> SketchReducer<'a> { if w_end_u64 > cov_hi { cov_hi = w_end_u64; } - // Sort descending by summed count, take top-k. + // Sort descending by summed count, take top-k. Tie-break + // on key so equal counts don't depend on `summed`'s + // HashMap iteration order (non-deterministic across runs). let mut items: Vec<(String, f64)> = summed.into_iter().collect(); - items - .sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); + items.sort_by(|a, b| { + b.1.partial_cmp(&a.1) + .unwrap_or(std::cmp::Ordering::Equal) + .then_with(|| a.0.cmp(&b.0)) + }); for (key, value) in items.into_iter().take(k) { let mut lv = ts.series_label_values.clone(); lv.insert("item".to_string(), key); diff --git a/data_plane/src/utils/http.rs b/data_plane/src/utils/http.rs index 77c9a1bf..26db630f 100644 --- a/data_plane/src/utils/http.rs +++ b/data_plane/src/utils/http.rs @@ -4,124 +4,6 @@ use std::collections::HashMap; use crate::query_engines::QueryResult; -// /// Prometheus-compatible response structure -// #[derive(Debug, serde::Serialize, serde::Deserialize)] -// pub struct PrometheusResponse { -// pub status: String, -// pub data: PrometheusData, -// } - -// #[derive(Debug, serde::Serialize, serde::Deserialize)] -// pub struct PrometheusData { -// #[serde(rename = "resultType")] -// pub result_type: String, -// pub result: Vec, -// } - -// #[derive(Debug, serde::Serialize, serde::Deserialize)] -// pub struct PrometheusResult { -// pub metric: HashMap, -// pub value: (f64, String), // [timestamp, value] -// } - -// /// Format results as Prometheus-compatible HTTP response -// pub fn format_results_as_http_response( -// results: &[PrecomputedOutput], -// timestamp: f64, -// ) -> Result { -// let mut prometheus_results = Vec::new(); - -// for result in results { -// if let Some(ref key) = result.key { -// let prometheus_result = PrometheusResult { -// metric: key.labels.clone(), -// value: (timestamp, "0.0".to_string()), // TODO: Extract actual value from accumulator -// }; -// prometheus_results.push(prometheus_result); -// } -// } - -// let response = PrometheusResponse { -// status: "success".to_string(), -// data: PrometheusData { -// result_type: "vector".to_string(), -// result: prometheus_results, -// }, -// }; - -// Ok(response) -// } - -// /// Format error response in Prometheus format -// pub fn format_error_response(error_msg: &str) -> PrometheusResponse { -// tracing::error!("Error: {}", error_msg); -// PrometheusResponse { -// status: "error".to_string(), -// data: PrometheusData { -// result_type: "vector".to_string(), -// result: vec![], -// }, -// } -// } - -// /// Parse query parameters from HTTP request -// pub fn parse_query_params(query_string: &str) -> HashMap> { -// let mut params = HashMap::new(); - -// for pair in query_string.split('&') { -// if let Some((key, value)) = pair.split_once('=') { -// let decoded_key = urlencoding::decode(key).unwrap_or_default().into_owned(); -// let decoded_value = urlencoding::decode(value).unwrap_or_default().into_owned(); - -// params -// .entry(decoded_key) -// .or_insert_with(Vec::new) -// .push(decoded_value); -// } -// } - -// params -// } - -// /// Format results as Prometheus-compatible HTTP response -// pub fn format_results_as_http_response( -// result_type: QueryResultType, -// results: &HashMap, // Simplified - key as string, value as f64 -// grouping_labels: &KeyByLabelNames, -// time: u64, -// ) -> Value { -// match result_type { -// QueryResultType::InstantVector => { -// let mut result = Vec::new(); -// for (k, v) in results.iter() { -// // Parse the key string back to values - this is a simplification -// // In the Python version, k is a Key object with values attribute -// let key_values: Vec<&str> = k.split(',').collect(); - -// let metric: HashMap = grouping_labels -// .keys -// .iter() -// .zip(key_values.iter()) -// .map(|(label, value)| (label.clone(), value.to_string())) -// .collect(); - -// result.push(json!({ -// "metric": metric, -// "value": [time as f64 / 1000.0, v.to_string()] -// })); -// } - -// json!({ -// "status": "success", -// "data": { -// "resultType": "vector", -// "result": result -// } -// }) -// } -// } -// } - /// Convert QueryResult to Prometheus-compatible format (for instant queries only) /// /// Returns an error if passed a Matrix result - use `convert_range_result_to_prometheus` for that.