diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index fc8330cb..4197b97d 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -392,8 +392,22 @@ pub fn collect_metric_to_family( // samples fan into each per-family pipeline at the agent and the // backend serves every (metric, capability) the workload needs. for (_, workload, _wc) in workload_store.get_all_for_metric(&entry.metric_name) { - let Some(physical_expr) = crate::optimizer::rules::bind_workload_typed(&workload) - else { + // If this metric declares an `item_label` (its inner + // high-cardinality dimension, e.g. "endpoint") and the + // parsed query's own label filters name a value for it (e.g. + // `{endpoint="checkout"}`), thread that through as the + // `Frequency` intent's actual per-item filter -- see + // `bind_workload_typed_with_item_filter`'s doc. + let item_filter = entry.item_label.as_deref().and_then(|label| { + workload + .label_filters + .get(label) + .map(|v| (label, v.as_str())) + }); + let Some(physical_expr) = crate::optimizer::rules::bind_workload_typed_with_item_filter( + &workload, + item_filter, + ) else { continue; }; if let Some(kind) = extract_root_sketch_kind(&physical_expr) { diff --git a/control_plane/src/intent_algebra/agg_intent.rs b/control_plane/src/intent_algebra/agg_intent.rs index b40f9989..c2e2081f 100644 --- a/control_plane/src/intent_algebra/agg_intent.rs +++ b/control_plane/src/intent_algebra/agg_intent.rs @@ -67,17 +67,30 @@ pub(crate) const FREQUENCY_EXT_KIND: &str = "frequency"; /// Construct control_plane's point-frequency-via-CMS intent. See module /// docs for why this is an `Extension`, not a shared first-class variant. -pub fn frequency(accuracy: AccuracyTarget) -> AggIntent { +/// +/// `item` is `Some((label, value))` for a per-item point lookup (e.g. +/// `count(cms_metric{item="checkout"})` -> `("item", "checkout")`) -- +/// `sketch_algebra::cost_model::ControlPlaneCostModel::readout_extension` +/// reads these same `item_label`/`item_value` payload keys back out to +/// build `SketchQuery::PointCount{key: Named(label), value: Some(value)}`. +/// `None` for a bare frequency total (no specific item), which reads out +/// as `PointCount{key: SampleValue, value: None}`. +pub fn frequency(accuracy: AccuracyTarget, item: Option<(String, String)>) -> AggIntent { + let mut payload = serde_json::json!({ "accuracy": accuracy }); + if let Some((label, value)) = item { + payload["item_label"] = serde_json::Value::String(label); + payload["item_value"] = serde_json::Value::String(value); + } AggIntent::Extension { ext_kind: FREQUENCY_EXT_KIND.to_string(), - payload: serde_json::json!({ "accuracy": accuracy }), + payload, } } -/// Default `Frequency` intent — `accuracy = e / 2000`. Unchanged default -/// from before the merge. +/// Default `Frequency` intent — `accuracy = e / 2000`, no item filter. +/// Unchanged default from before the merge. pub fn default_frequency() -> AggIntent { - frequency(AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0)) + frequency(AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0), None) } /// If `intent` is control_plane's `Frequency` extension, extract its diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs index 4b747598..7d462beb 100644 --- a/control_plane/src/optimizer/rules/mod.rs +++ b/control_plane/src/optimizer/rules/mod.rs @@ -63,6 +63,24 @@ pub fn typed_sketch_algebra_enabled() -> bool { /// `PhysicalExpr` is then fed into `planner::stage_split::split_typed_three_stage` /// + the per-stage emitters in `config::stage_config`. pub fn bind_workload_typed(w: &QueryWorkload) -> Option { + bind_workload_typed_with_item_filter(w, None) +} + +/// Like [`bind_workload_typed`], but for a `Frequency` statistic, `item_filter` +/// (a `(label, value)` pair, e.g. `("item", "checkout")`) threads the +/// query's actual per-item filter value through to the bound +/// `SketchQuery::PointCount` -- `None` (what `bind_workload_typed` itself +/// passes) gives the bare bucket total, same as before this parameter +/// existed. `QueryWorkload` itself carries no `item_label` field (adding +/// one would break its 30+ struct-literal construction sites across the +/// crate), so callers that know a metric's item_label -- e.g. +/// `emit::collect_metric_to_family`'s loop, which already has `entry: +/// &WorkloadEntry` and `workload.label_filters` in scope -- pass it in +/// directly instead. +pub fn bind_workload_typed_with_item_filter( + w: &QueryWorkload, + item_filter: Option<(&str, &str)>, +) -> Option { use crate::intent_algebra::schema::{Column, DataType}; use crate::intent_algebra::{AggIntent as L3AggIntent, QueryExpr, Schema, Source, WindowKind}; use crate::sketch_algebra::capability_matching::{ @@ -187,7 +205,10 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option crate::intent_algebra::frequency(intent_accuracy), + StatisticClass::Frequency => crate::intent_algebra::frequency( + intent_accuracy, + item_filter.map(|(label, value)| (label.to_string(), value.to_string())), + ), StatisticClass::TopK => L3AggIntent::TopK { k: 10, accuracy: intent_accuracy, @@ -643,6 +664,7 @@ mod tests { // | `endpoint_request_freq` | CMS | use crate::sketch_algebra::physical_expr::PhysicalExpr; + use asap_ir::intent_algebra::expr_ir::ColumnRef; use asap_sketch::SummaryKind; /// Walk the L4 binding output and pull out the approximate sketch @@ -653,6 +675,22 @@ mod tests { crate::emit::extract_root_sketch_kind(expr) } + /// Pull the `SketchQuery` out of a bound `PhysicalExpr`'s top-level + /// `SummaryEstimate` -- unlike `extract_family`, this needs the + /// readout itself (to check `PointCount`'s `key`/`value`), not just + /// the sketch family underneath it. + fn extract_query(expr: &PhysicalExpr) -> Option { + let PhysicalExpr::Committed(crate::sketch_algebra::physical_expr::L4Plan::Summary(node)) = + expr + else { + return None; + }; + match &node.expr { + asap_sketch::SummaryExpr::SummaryEstimate { query, .. } => Some(query.clone()), + _ => None, + } + } + /// Build a workload with the given metric name + reasonable /// AggType-driven default for the contract row. The metric-name match /// in `classify_demo_metric` overrides the AggType for the @@ -761,6 +799,41 @@ mod tests { ); } + #[test] + fn bind_workload_typed_with_item_filter_threads_the_actual_value() { + // `bind_workload_typed` itself (no item filter) must still read + // out as the bare bucket total -- unchanged behavior. + let w = workload_for("endpoint_request_freq", AggType::Frequency); + let bound = bind_workload_typed(&w).expect("must bind"); + assert!( + matches!( + extract_query(&bound), + Some(asap_sketch::SketchQuery::PointCount { + key: ColumnRef::SampleValue, + value: None + }) + ), + "no item filter given -> bare bucket total, got {:?}", + extract_query(&bound) + ); + + // With an item filter, the SAME workload must read out as a + // per-item point lookup carrying the actual value. + let bound_filtered = + bind_workload_typed_with_item_filter(&w, Some(("endpoint", "checkout"))) + .expect("must bind"); + match extract_query(&bound_filtered) { + Some(asap_sketch::SketchQuery::PointCount { + key: ColumnRef::Named(label), + value: Some(value), + }) => { + assert_eq!(label, "endpoint"); + assert_eq!(value, "checkout"); + } + other => panic!("expected PointCount{{key: Named(\"endpoint\"), value: Some(\"checkout\")}}, got {other:?}"), + } + } + // ── sketch_type_override (= sketch_family_override) wins ────────────────── #[test] diff --git a/control_plane/src/sketch_algebra/capability.rs b/control_plane/src/sketch_algebra/capability.rs index 68f5e101..9fe06fd4 100644 --- a/control_plane/src/sketch_algebra/capability.rs +++ b/control_plane/src/sketch_algebra/capability.rs @@ -897,7 +897,7 @@ mod tests { #[test] fn frequency_estimate_with_epsilon_returns_frequency_estimate_approx() { - let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01)); + let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01), None); assert_eq!( capability_for(&intent), Some(Capability::FrequencyEstimate(SketchKindHandle::Any)) @@ -906,10 +906,13 @@ mod tests { #[test] fn frequency_estimate_with_epsilon_delta_returns_frequency_estimate_approx() { - let intent = crate::intent_algebra::frequency(AccuracyTarget::EpsilonDelta { - epsilon: 0.01, - delta: 0.001, - }); + let intent = crate::intent_algebra::frequency( + AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.001, + }, + None, + ); assert_eq!( capability_for(&intent), Some(Capability::FrequencyEstimate(SketchKindHandle::Any)) @@ -920,7 +923,7 @@ mod tests { fn frequency_estimate_with_exact_returns_none() { // Exact aggregation routes to archive (sketch fallback only // meaningful when raw counters aren't kept). - let intent = crate::intent_algebra::frequency(AccuracyTarget::Exact); + let intent = crate::intent_algebra::frequency(AccuracyTarget::Exact, None); assert_eq!(capability_for(&intent), None); } diff --git a/control_plane/src/sketch_algebra/tests.rs b/control_plane/src/sketch_algebra/tests.rs index e0b7a851..1fd9d55f 100644 --- a/control_plane/src/sketch_algebra/tests.rs +++ b/control_plane/src/sketch_algebra/tests.rs @@ -860,7 +860,7 @@ fn phase_b_archive_only_intents_round_trip_through_binder() { // ── Deliberate behavior changes (ASAPController#150 / #151) ────────────────── // // `AggIntent::Extension` (this deployment's `Frequency` point-query, -// built via `crate::intent_algebra::frequency(accuracy)`) now binds to a +// built via `crate::intent_algebra::frequency(accuracy, item)`) now binds to a // real `Cms` sketch via `ControlPlaneCostModel::realize_extension`/ // `readout_extension` (ASAPController#150) — see `frequency_extension_binds_cms` // below and `optimizer::rules::mod::tests::typed_binding_endpoint_request_freq_binds_cms`. @@ -876,7 +876,7 @@ fn frequency_extension_binds_cms() { // `ControlPlaneCostModel::realize_extension`/`readout_extension` // (ASAPController#150) now realize `AggIntent::Extension{"frequency"}` // as a real `Cms` sketch instead of declining to `Logical`. - let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01)); + let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01), None); let expr = QueryExpr::Aggregate { by: vec![].into(), aggs: vec![intent],