From deeb8ebc7a806919da16da8679b496e1c71c2ffd Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 22:21:36 -0600 Subject: [PATCH 1/3] feat(promql): support weighted temporal top-k --- Cargo.lock | 6 +- control_plane/Cargo.toml | 6 +- .../src/backend_plan/from_stage_config.rs | 7 ++- control_plane/src/emit/backend_push.rs | 1 + control_plane/src/emit/stage_config.rs | 31 ++++++++- .../src/physical/colored_dag/emitter.rs | 7 ++- control_plane/src/physical/compiler.rs | 63 +++++++++++++++++-- control_plane/src/physical/post_asap/tests.rs | 2 +- control_plane/src/query_plan.rs | 24 ++++++- control_plane/src/replan.rs | 2 + crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 2 +- .../asap_query_engine/summary_executor.rs | 44 ++++++++++--- .../asapquery_compatibility_process_e2e.rs | 48 +++++++++++--- ...asapquery-compatibility-demo-snapshot.json | 36 +++++++++++ 15 files changed, 245 insertions(+), 36 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 943595fa..4d1e7590 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -343,7 +343,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cb50219c582d43f53ab77d3a595bd1ea4a9aa119#cb50219c582d43f53ab77d3a595bd1ea4a9aa119" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5daccfe#5daccfede6fe75dbe638be8e5eed5382b5b91693" dependencies = [ "asap-types", "serde", @@ -354,7 +354,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cb50219c582d43f53ab77d3a595bd1ea4a9aa119#cb50219c582d43f53ab77d3a595bd1ea4a9aa119" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5daccfe#5daccfede6fe75dbe638be8e5eed5382b5b91693" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cb50219c582d43f53ab77d3a595bd1ea4a9aa119#cb50219c582d43f53ab77d3a595bd1ea4a9aa119" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5daccfe#5daccfede6fe75dbe638be8e5eed5382b5b91693" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 033c0d2a..becb8600 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -93,8 +93,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "5daccfe" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "5daccfe" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -102,7 +102,7 @@ asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "5daccfe" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/backend_plan/from_stage_config.rs b/control_plane/src/backend_plan/from_stage_config.rs index a840420a..6690a500 100644 --- a/control_plane/src/backend_plan/from_stage_config.rs +++ b/control_plane/src/backend_plan/from_stage_config.rs @@ -264,6 +264,7 @@ mod tests { spatial_filter: String::new(), grouping, item_label: None, + topk_weight: None, aggregation_input: AggregationInput::SketchEnvelope, } } @@ -469,7 +470,10 @@ mod tests { readouts: vec![ BackendReadout { aggregation_id: "agg0".into(), - op: SketchQuery::TopK { k: 10 }, + op: SketchQuery::TopK { + k: 10, + weight: planner_types::post_asap::TopKWeight::Value, + }, }, BackendReadout { aggregation_id: "agg1".into(), @@ -514,6 +518,7 @@ mod tests { spatial_filter: String::new(), grouping: vec!["zone".to_string()], item_label: None, + topk_weight: None, aggregation_input: AggregationInput::Raw, }; let cfg = BackendStageConfig { diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index a748377d..7bc963e0 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -746,6 +746,7 @@ mod tests { ), grouping: vec![], item_label: None, + topk_weight: None, spatial_filter: String::new(), window_secs: 60, aggregation_input: AggregationInput::SketchEnvelope, diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 8155a20e..dbcdd796 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -2986,6 +2986,20 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa obj.insert("item_label".to_string(), JsonValue::String(label.clone())); } } + if let Some(weight) = agg.topk_weight { + if let Some(obj) = parameters.as_object_mut() { + obj.insert( + "weight_mode".to_string(), + JsonValue::String( + match weight { + planner_types::post_asap::TopKWeight::Count => "count", + planner_types::post_asap::TopKWeight::Value => "value", + } + .to_string(), + ), + ); + } + } let aggregation_input = match agg.aggregation_input { AggregationInput::SketchEnvelope => "sketch_envelope", AggregationInput::Raw => "raw", @@ -3042,9 +3056,13 @@ fn build_backend_readout_json(r: &BackendReadout) -> JsonValue { "key": column_ref_to_wire_key(key), "value": value, }), - SketchQuery::TopK { k } => json!({ + SketchQuery::TopK { k, weight } => json!({ "op": "topk", "k": k, + "weight_mode": match weight { + planner_types::post_asap::TopKWeight::Count => "count", + planner_types::post_asap::TopKWeight::Value => "value", + }, }), } } @@ -3206,6 +3224,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), item_label: None, + topk_weight: None, aggregation_input, } } @@ -3603,7 +3622,10 @@ mod tests { readouts: vec![ BackendReadout { aggregation_id: "agg0".into(), - op: SketchQuery::TopK { k: 10 }, + op: SketchQuery::TopK { + k: 10, + weight: planner_types::post_asap::TopKWeight::Value, + }, }, BackendReadout { aggregation_id: "agg1".into(), @@ -3691,7 +3713,10 @@ mod tests { SketchQuery::Quantile { q: 0.99 } } SketchAlgorithm::Hll => SketchQuery::Cardinality, - SketchAlgorithm::CountSketch => SketchQuery::TopK { k: 10 }, + SketchAlgorithm::CountSketch => SketchQuery::TopK { + k: 10, + weight: planner_types::post_asap::TopKWeight::Value, + }, SketchAlgorithm::Cms => SketchQuery::PointCount { key: ColumnRef::Named("user_42".into()), value: None, diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 026e1c7a..e0e218f2 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -37,7 +37,7 @@ 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, + SketchQuery, SummaryExpr, SummaryFamilyType, TopKWeight, }; /// Flattened view of one [`ColoredNode`](crate::physical::colored_dag::dag::ColoredNode)'s @@ -701,6 +701,9 @@ pub struct BackendAggregation { /// `parameters["item_label"]` so the data-plane ingest records it on the /// CMS sid and can answer per-item `estimate(key)` (FrequencyEstimate). pub item_label: Option, + /// Heap update mode selected by the TopK readout. `Count` uses unit + /// updates; `Value` ranks by the summed sample value. + pub topk_weight: Option, /// Phase ε.1 — what shape the backend ingests for this /// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the /// default (the wire payload is a sketch state already). Mode 2 @@ -910,6 +913,7 @@ impl Emitter for ThreeStageEmitter { // from workload.group_by_labels — see the // struct doc-comment for the rationale. grouping: Vec::new(), + topk_weight: None, // Mode 1 — sketch built at edge, ships envelope. aggregation_input: AggregationInput::SketchEnvelope, }); @@ -991,6 +995,7 @@ impl Emitter for ThreeStageEmitter { window_secs: edge.window_secs.unwrap_or(0), spatial_filter: spatial_filter_from_label_filters(&edge.label_filters), grouping: Vec::new(), + topk_weight: None, // Mode 2 — backend builds sketch from raw OTLP. aggregation_input: AggregationInput::Raw, }); diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index dce55179..9c8fc129 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -42,7 +42,7 @@ use crate::query_plan::{ use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; -pub const PLANNER_REVISION: &str = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119"; +pub const PLANNER_REVISION: &str = "5daccfede6fe75dbe638be8e5eed5382b5b91693"; #[derive(Debug, Clone)] pub struct PlanningQuery { @@ -185,6 +185,10 @@ pub struct BackendLocalImplementation { pub window_implementation_id: String, pub state_layout: String, pub implementation_cost: ImplementationCostEvidence, + /// Optional per-query membership certificates for approximate TopK. + /// Keys are the exact PromQL strings in `query_workload`. + #[serde(default)] + pub topk_evidence: HashMap, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -1544,6 +1548,7 @@ impl BackendLocalPlanningSnapshot { )); } let mut queries = Vec::with_capacity(entries.len()); + let mut topk_evidence_by_id = HashMap::new(); for (index, entry) in entries.into_iter().enumerate() { let evaluation_interval_ms = match entry.recurrence { QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(interval)) => interval.0, @@ -1589,14 +1594,19 @@ impl BackendLocalPlanningSnapshot { horizon_seconds: self.implementation.horizon_seconds, costs: self.implementation.lifecycle_costs.clone(), }; - let post_asap = select_post_asap(&parsed, accuracy.clone(), &lifecycle, None) + let topk_evidence = self.implementation.topk_evidence.get(&query_string); + let post_asap = select_post_asap(&parsed, accuracy.clone(), &lifecycle, topk_evidence) .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?; let mut cost = self.implementation.implementation_cost.clone(); cost.workload_fingerprint = canonical_promql(&query_string).map_err(CompileError::QueryPlan)?; cost.horizon_seconds = self.implementation.horizon_seconds; + let query_id = format!("compat-query-{index}"); + if let Some(evidence) = topk_evidence { + topk_evidence_by_id.insert(query_id.clone(), evidence.clone()); + } queries.push(PlanningQuery { - query_id: format!("compat-query-{index}"), + query_id, query_string, post_asap, source: Source::TimeSeries { @@ -1620,7 +1630,7 @@ impl BackendLocalPlanningSnapshot { PhysicalCompiler.compile( PlanningRequest { queries, - evidence: HashMap::new(), + evidence: topk_evidence_by_id, planner_revision: PLANNER_REVISION.into(), }, self.environment, @@ -1725,6 +1735,10 @@ impl PhysicalCompiler { spatial_filter: String::new(), grouping: query.group_by.clone(), item_label: None, + topk_weight: selected.readout.as_ref().and_then(|readout| match readout { + SketchQuery::TopK { weight, .. } => Some(*weight), + _ => None, + }), aggregation_input: match environment.target { PhysicalDeploymentTarget::DistributedCollectors => { AggregationInput::SketchEnvelope @@ -2688,6 +2702,7 @@ mod tests { window_implementation_id: "backend-tumbling-v1".into(), state_layout: "anchored-pane-v1".into(), implementation_cost: template.window_implementations[0].cost.clone(), + topk_evidence: HashMap::new(), }, environment, }; @@ -2800,13 +2815,15 @@ mod tests { assert!(plan.collector_plans.is_empty()); assert!(plan.transmission_plan.rules.is_empty()); - assert_eq!(plan.query_plan.entries.len(), 4); - assert_eq!(plan.precompute_plan.materializations.len(), 3); + assert_eq!(plan.query_plan.entries.len(), 6); + assert_eq!(plan.precompute_plan.materializations.len(), 5); for query in [ "rate(asap_demo_counter_total[5s])", "increase(asap_demo_counter_total[5s])", "sum_over_time(asap_demo_gauge[5s])", "quantile_over_time(0.5, asap_demo_latency_ms[5s])", + "topk(5, sum_over_time(asap_demo_gauge[5s]))", + "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))", ] { assert!(plan.query_plan.lookup(query).is_ok(), "missing {query}"); } @@ -3054,6 +3071,40 @@ mod tests { ); } + #[test] + fn asapquery_706_topk_over_temporal_shapes_compile_with_explicit_weights() { + let evidence = || TopKMembershipEvidence { + selected_lower_bound: 101.0, + excluded_upper_bound: 100.0, + interval_failure_probability: 0.005, + observed_at_unix_ms: 9_500, + source: "runtime-margin-monitor".into(), + }; + for (query, expected_weight) in [ + ("topk(5, sum_over_time(m[1m]))", "value"), + ("topk by (job) (5, count_over_time(m[1m]))", "count"), + ] { + let request = request_with_evidence("q-topk-temporal", query, Some(evidence())) + .unwrap_or_else(|error| panic!("{query} must select: {error}")); + let bundle = PhysicalCompiler + .compile(request, environment(10_000)) + .unwrap_or_else(|error| panic!("{query} must compile: {error}")); + let materialization = bundle + .precompute_plan + .materializations + .first() + .expect("TopK materialization"); + assert_eq!( + materialization + .parameters + .get("weight_mode") + .and_then(Value::as_str), + Some(expected_weight), + "{query} must configure the matching heap update mode" + ); + } + } + #[test] fn planner_revision_is_part_of_the_compile_contract() { let mut request = request("q", "quantile_over_time(0.9, m[1m])"); diff --git a/control_plane/src/physical/post_asap/tests.rs b/control_plane/src/physical/post_asap/tests.rs index 324f710c..b8b5850c 100644 --- a/control_plane/src/physical/post_asap/tests.rs +++ b/control_plane/src/physical/post_asap/tests.rs @@ -243,7 +243,7 @@ fn topk_binding_family(bound: &PhysicalExpr) -> (SketchAlgorithm, u32, u32) { query, summary_input, } => { - assert!(matches!(query, SketchQuery::TopK { k } if *k == 10)); + assert!(matches!(query, SketchQuery::TopK { k, .. } if *k == 10)); match &summary_input.expr { SummaryExpr::SummaryAgg { family, .. } => match family { SummaryFamilyType::Sketch(kind, _) diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 61e5796e..d66546d5 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -286,16 +286,30 @@ pub enum QueryReadout { Cardinality, TopK { k: usize, + weight: QueryTopKWeight, }, } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum QueryTopKWeight { + Count, + Value, +} + impl From for QueryReadout { fn from(query: SketchQuery) -> Self { match query { SketchQuery::Quantile { q } => Self::Quantile { q }, SketchQuery::PointCount { key, value } => Self::PointCount { key, value }, SketchQuery::Cardinality => Self::Cardinality, - SketchQuery::TopK { k } => Self::TopK { k }, + SketchQuery::TopK { k, weight } => Self::TopK { + k, + weight: match weight { + planner_types::post_asap::TopKWeight::Count => QueryTopKWeight::Count, + planner_types::post_asap::TopKWeight::Value => QueryTopKWeight::Value, + }, + }, } } } @@ -306,7 +320,13 @@ impl From for SketchQuery { QueryReadout::Quantile { q } => Self::Quantile { q }, QueryReadout::PointCount { key, value } => Self::PointCount { key, value }, QueryReadout::Cardinality => Self::Cardinality, - QueryReadout::TopK { k } => Self::TopK { k }, + QueryReadout::TopK { k, weight } => Self::TopK { + k, + weight: match weight { + QueryTopKWeight::Count => planner_types::post_asap::TopKWeight::Count, + QueryTopKWeight::Value => planner_types::post_asap::TopKWeight::Value, + }, + }, } } } diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 19a86813..2a1bf463 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -686,6 +686,7 @@ impl Replanner { window_secs, spatial_filter: String::new(), grouping: workload.group_by_labels.clone(), + topk_weight: None, // ExactAgg consumes raw values at the backend (the agent // ships counter samples; the backend's // SumAccumulator integrates them). @@ -1319,6 +1320,7 @@ mod tests { grouping: vec!["zone".to_string()], spatial_filter: String::new(), window_secs: 60, + topk_weight: None, aggregation_input: AggregationInput::Raw, }], readouts: Vec::new(), diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index fe3a0270..074cc632 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -30,4 +30,4 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] } # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "5daccfe" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index a424258b..ebdfeb43 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,7 +38,7 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "5daccfe" } # Shared external (workspace) serde.workspace = true diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index 025005cf..7e7083ca 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -739,7 +739,7 @@ fn readout_cumulative( return Err(SummaryExecutorError::NoCandidates); }; let w_end = latest_window_end.unwrap_or(t1_ms); - if let SketchQuery::TopK { k } = query { + if let SketchQuery::TopK { k, .. } = query { Ok(SummaryValue::TopK( vec![(w_end, topk_ranked(&merged, *k)?)], coverage, @@ -801,7 +801,7 @@ fn readout_per_window( if by_window.is_empty() { return Err(SummaryExecutorError::NoCandidates); } - if let SketchQuery::TopK { k } = query { + if let SketchQuery::TopK { k, .. } = query { let points = by_window .into_iter() .map(|(w_end, rs)| topk_ranked(&rs, *k).map(|items| (w_end, items))) @@ -2133,7 +2133,13 @@ mod tests { }, ); let child = scan_node("requests_total", None); - let tree = estimate_node(cms_agg_node(child), SketchQuery::TopK { k: 5 }); + let tree = estimate_node( + cms_agg_node(child), + SketchQuery::TopK { + k: 5, + weight: planner_types::post_asap::TopKWeight::Value, + }, + ); let exec = ctx(&idx); match execute(&tree, &exec) { Err(crate::query_engines::asap_query_engine::summary_exec::ExecError::Executor( @@ -2170,7 +2176,13 @@ mod tests { ); let child = scan_node("requests_by_route", None); - let tree = estimate_node(cms_with_heap_agg_node(child), SketchQuery::TopK { k: 3 }); + let tree = estimate_node( + cms_with_heap_agg_node(child), + SketchQuery::TopK { + k: 3, + weight: planner_types::post_asap::TopKWeight::Value, + }, + ); let exec = ctx(&idx); let ExecOutcome::Value(v) = execute(&tree, &exec).expect("execute should succeed") else { @@ -2220,7 +2232,13 @@ mod tests { ); let child = scan_node("requests_by_route", None); - let tree = estimate_node(cms_with_heap_agg_node(child), SketchQuery::TopK { k: 5 }); + let tree = estimate_node( + cms_with_heap_agg_node(child), + SketchQuery::TopK { + k: 5, + weight: planner_types::post_asap::TopKWeight::Value, + }, + ); let exec = ctx(&idx); let ExecOutcome::Value(v) = execute(&tree, &exec).expect("execute should succeed") else { @@ -2266,7 +2284,13 @@ mod tests { ); let child = scan_node("requests_by_route", None); - let tree = estimate_node(cms_with_heap_agg_node(child), SketchQuery::TopK { k: 2 }); + let tree = estimate_node( + cms_with_heap_agg_node(child), + SketchQuery::TopK { + k: 2, + weight: planner_types::post_asap::TopKWeight::Value, + }, + ); let exec = matrix_ctx(&idx); let ExecOutcome::Value(v) = execute(&tree, &exec).expect("execute should succeed") else { @@ -2586,7 +2610,13 @@ mod tests { ); let child = scan_node("requests_by_route", None); - let tree = estimate_node(cms_with_heap_agg_node(child), SketchQuery::TopK { k: 5 }); + let tree = estimate_node( + cms_with_heap_agg_node(child), + SketchQuery::TopK { + k: 5, + weight: planner_types::post_asap::TopKWeight::Value, + }, + ); let exec = matrix_ctx(&idx); let ExecOutcome::Value(v) = execute(&tree, &exec).expect("execute should succeed") else { diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index eefc1b34..3fc29c3d 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -51,11 +51,20 @@ async fn wait_until_ready(client: &reqwest::Client, url: &str, child: &mut Child } fn series(metric: &str, samples: &[(i64, f64)]) -> TimeSeries { + series_with_labels(metric, &[], samples) +} + +fn series_with_labels(metric: &str, labels: &[(&str, &str)], samples: &[(i64, f64)]) -> TimeSeries { + let mut wire_labels = vec![Label { + name: "__name__".into(), + value: metric.into(), + }]; + wire_labels.extend(labels.iter().map(|(name, value)| Label { + name: (*name).into(), + value: (*value).into(), + })); TimeSeries { - labels: vec![Label { - name: "__name__".into(), - value: metric.into(), - }], + labels: wire_labels, samples: samples .iter() .map(|(timestamp, value)| Sample { @@ -265,8 +274,9 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() (base + 9_400, 12.0), ], ), - series( + series_with_labels( "asap_demo_gauge", + &[("job", "api")], &[ (base + 500, 1.0), (base + 1_700, 2.0), @@ -297,7 +307,11 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() let watermark_advance = WriteRequest { timeseries: vec![ series("asap_demo_counter_total", &[(base + 10_500, 15.0)]), - series("asap_demo_gauge", &[(base + 10_500, 9.0)]), + series_with_labels( + "asap_demo_gauge", + &[("job", "api")], + &[(base + 10_500, 9.0)], + ), series("asap_demo_latency_ms", &[(base + 10_500, 55.0)]), ], }; @@ -352,6 +366,24 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() &backend_log, ) .await; + let topk_sum = wait_for_warm_instant( + &client, + &backend, + "topk(5, sum_over_time(asap_demo_gauge[5s]))", + first_eval, + &backend_log, + ) + .await; + let topk_count = wait_for_warm_instant( + &client, + &backend, + "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))", + first_eval, + &backend_log, + ) + .await; + assert!(first_value(&topk_sum, "value").is_some()); + assert!(first_value(&topk_count, "value").is_some()); let rate_value = first_value(&rate, "value").expect("rate value"); let increase_value = first_value(&increase, "value").expect("increase value"); assert!((rate_value * 5.0 - increase_value).abs() < 1e-9); @@ -367,6 +399,8 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() "increase(asap_demo_counter_total[5s])", "sum_over_time(asap_demo_gauge[5s])", "quantile_over_time(0.5, asap_demo_latency_ms[5s])", + "topk(5, sum_over_time(asap_demo_gauge[5s]))", + "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))", ] { let response: Value = client .get(format!("{backend}/api/v1/query_range")) @@ -467,7 +501,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() let materializations = status["materializations"] .as_array() .expect("materialization statuses"); - assert_eq!(materializations.len(), 3); + assert_eq!(materializations.len(), 5); assert!(materializations .iter() .all(|entry| entry["phase"] == "serving")); diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index 2870e543..007a486f 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -34,6 +34,26 @@ }, "predictability": { "predictable": { "known_at": null } }, "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + }, + { + "query": "topk(5, sum_over_time(asap_demo_gauge[5s]))", + "demand": { "fixed_interval": 1000 }, + "requirements": { + "accuracy": { "explicit": { "EpsilonDelta": { "epsilon": 0.01, "delta": 0.01 } } }, + "response_latency": "unspecified" + }, + "predictability": { "predictable": { "known_at": null } }, + "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + }, + { + "query": "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))", + "demand": { "fixed_interval": 1000 }, + "requirements": { + "accuracy": { "explicit": { "EpsilonDelta": { "epsilon": 0.01, "delta": 0.01 } } }, + "response_latency": "unspecified" + }, + "predictability": { "predictable": { "known_at": null } }, + "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } } ], "data_workload": { @@ -76,6 +96,22 @@ "storage_bytes": 2048, "source_scan_bytes": 0, "weighted_cost": 1.0 + }, + "topk_evidence": { + "topk(5, sum_over_time(asap_demo_gauge[5s]))": { + "selected_lower_bound": 101.0, + "excluded_upper_bound": 100.0, + "interval_failure_probability": 0.005, + "observed_at_unix_ms": 9500, + "source": "compatibility-fixture" + }, + "topk by (job) (5, count_over_time(asap_demo_gauge[5s]))": { + "selected_lower_bound": 101.0, + "excluded_upper_bound": 100.0, + "interval_failure_probability": 0.005, + "observed_at_unix_ms": 9500, + "source": "compatibility-fixture" + } } }, "environment": { From 8614b7c12c49f1bca13bdf6dedfe630cffa1b217 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 4 Sep 2026 05:47:59 -0600 Subject: [PATCH 2/3] fix(promql): preserve canonical and legacy topk plans --- control_plane/src/physical/compiler.rs | 2 +- control_plane/src/query_plan.rs | 17 ++++++++++++++++- 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 9c8fc129..84f45d2d 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -187,7 +187,7 @@ pub struct BackendLocalImplementation { pub implementation_cost: ImplementationCostEvidence, /// Optional per-query membership certificates for approximate TopK. /// Keys are the exact PromQL strings in `query_workload`. - #[serde(default)] + #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub topk_evidence: HashMap, } diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index d66546d5..373864db 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -286,14 +286,16 @@ pub enum QueryReadout { Cardinality, TopK { k: usize, + #[serde(default)] weight: QueryTopKWeight, }, } -#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum QueryTopKWeight { Count, + #[default] Value, } @@ -475,6 +477,19 @@ mod tests { ); } + #[test] + fn legacy_topk_readout_defaults_to_value_weighting() { + let readout: QueryReadout = + serde_json::from_str(r#"{"kind":"top_k","k":5}"#).expect("legacy TopK readout"); + assert_eq!( + readout, + QueryReadout::TopK { + k: 5, + weight: QueryTopKWeight::Value, + } + ); + } + #[test] fn graph_validation_rejects_cycles() { let mut nodes = BTreeMap::new(); From ca7024eb0e1e4d5bd68b93711f29b59fb6bcdd77 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 12:28:13 -0600 Subject: [PATCH 3/3] fix(backfill): share materialization identity and TopK update semantics --- data_plane/src/drivers/ingest/otel.rs | 30 ++--- .../drivers/ingest/prometheus_remote_write.rs | 4 +- data_plane/src/precompute_engine/worker.rs | 2 +- .../sketch_db/backfill/processor.rs | 63 +++++------ .../sketch_db/backfill/window_builder.rs | 106 +++++++++++------- .../src/storage_engines/sketch_db/data/mod.rs | 12 ++ .../storage_engines/sketch_db/index/mod.rs | 14 +-- 7 files changed, 116 insertions(+), 115 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 1af9c00f..12866f28 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -597,13 +597,9 @@ fn flush_barrier_drops(_state: &IngestState, drops: &HashMap, driver_t /// distinct `(rack, node, pod)` tuple under a `grouping_labels=[zone]` /// policy would mint its own sid and never roll up. /// -/// `agg_kind` is `ExactAgg { ... }` for both raw-sample and opaque- -/// envelope sketch paths so the resolver key matches the signature -/// `reconcile_from_streaming_config` derives from the same config; the -/// modified-OTLP first-class sketch path takes a different sid- -/// resolution route inside `route_modified_otlp_sketches_to_precompute` -/// because it carries per-DP `(SketchAlgorithm, SketchConfig)` and -/// must distinguish (e.g.) DDSketch vs Kll over the same series. +/// Configured ingest shares the policy-aware physical identity used by the +/// live storage sink and backfill. Unbound modified-OTLP sketches retain +/// their separate wire-level identity protocol. fn resolve_bucket_sid_for_agg_config( ingest_state: &Arc, config: &asap_types::aggregation_config::AggregationConfig, @@ -619,14 +615,8 @@ fn resolve_bucket_sid_for_agg_config( }) .collect(); let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&grouping_pairs); - let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg { - agg_type: config.aggregation_type, - parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters( - &config.parameters, - ), - spatial_filter_canonical: config.spatial_filter_normalized.clone(), - }; - let agg_kind_canonical = agg_kind.canonical_string(); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(config); let sid = ingest_state .series_resolver .resolve(&config.metric, &fp, &agg_kind_canonical); @@ -4656,14 +4646,8 @@ mod sid_bucketing_tests { // regardless of group_key shape — the test pin is on sid // assignment, not on group_key content), then verify the // sid matches the resolver mint for THAT zone. - let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg { - agg_type: cfg.aggregation_type, - parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters( - &cfg.parameters, - ), - spatial_filter_canonical: cfg.spatial_filter_normalized.clone(), - }; - let agg_kind_canonical = agg_kind.canonical_string(); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(&cfg); for (sid, _, _, samples) in &groups { let mut vals: Vec = samples.iter().map(|(_, _, v)| *v).collect(); vals.sort_by(|a, b| a.partial_cmp(b).unwrap()); diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index a49eb865..b90cd013 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -461,13 +461,13 @@ fn route_messages( }) .collect(); let attrs_fp = super::canonical_attrs_fingerprint(&grouping_pairs); - let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(config); let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64()); // A sketch family is not a complete physical identity. Two // materializations may use the same family and grouping while // differing in update semantics (for example count- versus // value-weighted Top-K). Keep those states on distinct SIDs. - let materialization_kind = format!("{}|{}", agg_kind.canonical_string(), policy_fp); + let materialization_kind = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(config); let sid = ingest .series_resolver diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index b64b2ad7..34ce3c56 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -1191,7 +1191,7 @@ pub fn decode_label_value(s: &str) -> std::borrow::Cow<'_, str> { /// the key dimension *inside* the sketch (e.g., which bucket in a CMS, which /// entry in a MultipleSumAccumulator's HashMap). This matches the Arroyo SQL /// pattern: `udf(concat_ws(';', aggregated_labels), value)`. -fn apply_sample( +pub(crate) fn apply_sample( updater: &mut dyn AccumulatorUpdater, series_key: &str, val: f64, diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs index d9ca4b7c..0d843996 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs @@ -72,7 +72,6 @@ use tracing::debug; use crate::drivers::ingest::canonical_attrs_fingerprint; use crate::drivers::ingest::series_resolver::SeriesIdResolver; use crate::precompute_engine::worker::parse_labels_from_series_key; -use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; use crate::storage_engines::types::{AggregateCore, HotReloadStreamingConfig, KeyByLabelValues}; use asap_types::aggregation_config::AggregationConfig; use asap_types::PolicyFingerprint; @@ -119,12 +118,8 @@ fn build_group_key_label_values(group_key: &str) -> KeyByLabelValues { /// embedded in `series_key` (the `metric{k1="v1",k2="v2"}` text shape /// `RawSample::labels` holds), so we parse them out first. /// -/// The sid identity tuple is `(metric, attrs_fp, agg_kind_canonical)` -/// — identical to what the live ingest path computes, so the same -/// `(metric, grouping-values, agg_kind)` produces the SAME sid no -/// matter which path (live or backfill) saw the sample first. That -/// invariant is what lets backfill writes land in the same store -/// row the live ingest already populated for `[created_at, ∞)`. +/// Policy and grouping identity must match live ingestion so historical and +/// live windows occupy the same storage row. fn resolve_backfill_bucket_sid( resolver: &SeriesIdResolver, config: &AggregationConfig, @@ -138,12 +133,8 @@ fn resolve_backfill_bucket_sid( .map(|name| (name.as_str(), *labels.get(name.as_str()).unwrap_or(&""))) .collect(); let attrs_fp = canonical_attrs_fingerprint(&grouping_pairs); - let agg_kind = AggKind::ExactAgg { - agg_type: config.aggregation_type, - parameters_canonical: canonical_parameters(&config.parameters), - spatial_filter_canonical: config.spatial_filter_normalized.clone(), - }; - let agg_kind_canonical = agg_kind.canonical_string(); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(config); resolver.resolve(&config.metric, &attrs_fp, &agg_kind_canonical) } @@ -912,19 +903,10 @@ mod tests { assert_eq!(written, vec![(fp, (0u64, 100u64))]); } - /// B7.7 invariant: the sid the backfill processor mints for a - /// `(config, grouping-values)` tuple is bit-equal to the sid the - /// live ingest path's `resolve_bucket_sid_for_agg_config` would - /// mint via the SAME `SeriesIdResolver`. Locks the "live and - /// backfill share one sid namespace" contract — without it, the - /// `[created_at, ∞)` and `[0, created_at)` halves of the agg's - /// timeline would live under DIFFERENT sids and the query path - /// would only see half the history. + /// Replay and the actual live storage sink must resolve the same row. #[test] fn backfill_sid_matches_live_ingest_sid_for_same_grouping_values() { - use crate::drivers::ingest::canonical_attrs_fingerprint; use crate::drivers::ingest::series_resolver::SeriesIdResolver; - use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; let cfg = sum_config(1, "latency", vec!["svc", "zone"]); let resolver = SeriesIdResolver::new(); @@ -933,20 +915,29 @@ mod tests { let backfill_sid = resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"a\",zone=\"z0\"}"); - // Live side: mirror what `resolve_bucket_sid_for_agg_config` - // in drivers/ingest/otel.rs does, manually here so the test - // doesn't need to drive the OTLP pipeline. - let live_attrs_fp = canonical_attrs_fingerprint(&[("svc", "a"), ("zone", "z0")]); - let live_agg_kind = AggKind::ExactAgg { - agg_type: cfg.aggregation_type, - parameters_canonical: canonical_parameters(&cfg.parameters), - spatial_filter_canonical: cfg.spatial_filter_normalized.clone(), - }; - let live_sid = resolver.resolve( - &cfg.metric, - &live_attrs_fp, - &live_agg_kind.canonical_string(), + // Exercise the actual live sink instead of duplicating its SID formula. + let store = crate::storage_engines::sketch_db::index::SketchStore::new(); + let output = crate::storage_engines::types::PrecomputedOutput::new( + 100, + 200, + Some( + crate::storage_engines::types::KeyByLabelValues::new_with_labels(vec![ + "a".into(), + "z0".into(), + ]), + ), + cfg.policy_fingerprint(), ); + let acc = + crate::precompute_engine::operators::sum_accumulator::SumAccumulator::with_sum(1.0); + let live_sid = store + .ingest_precompute_for_agg_config( + |metric, attrs, kind| resolver.resolve(metric, attrs, kind), + &cfg, + &output, + &acc, + ) + .expect("live sink write"); assert_eq!( backfill_sid, live_sid, diff --git a/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs b/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs index 0d00e57c..55ba2f77 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs @@ -50,44 +50,16 @@ use crate::precompute_engine::accumulator_factory::{ create_accumulator_updater, AccumulatorUpdater, }; -use crate::precompute_engine::worker::parse_labels_from_series_key; +use crate::precompute_engine::worker::apply_sample; use crate::storage_engines::sketch_db::backfill::raw_sample_reader::RawSample; -use crate::storage_engines::types::{AggregateCore, KeyByLabelValues}; +use crate::storage_engines::types::AggregateCore; use asap_types::aggregation_config::AggregationConfig; -/// Extract the MultipleSubpopulation aggregated-label key from a -/// Prometheus-style series key. Duplicated from -/// `precompute_engine::worker::extract_aggregated_key_from_series` -/// (which is file-private). Kept here so the backfill module -/// doesn't force a `pub(crate)` on a live-path helper — the -/// dependency is one-way: worker does NOT import anything from -/// backfill. -/// -/// The implementation must track the live one exactly; the -/// end-to-end determinism test in `backfill_processor.rs` will -/// fail if they drift. -fn extract_aggregated_key(series_key: &str, config: &AggregationConfig) -> KeyByLabelValues { - let labels = parse_labels_from_series_key(series_key); - let mut values = Vec::new(); - for label_name in &config.aggregated_labels.labels { - if let Some(val) = labels.get(label_name.as_str()) { - values.push(val.to_string()); - } else { - values.push(String::new()); - } - } - KeyByLabelValues::new_with_labels(values) -} - /// Construct the accumulator for one `(agg_id, window)` pair by /// feeding `samples` in order into a fresh `AccumulatorUpdater`. /// -/// Sample format: `samples[i].labels` is the full series key -/// (Prometheus-style `metric{k="v",…}`); the function extracts -/// the MultipleSubpopulation key from the series key using the -/// same helper the live worker uses -/// (`extract_aggregated_key_from_series`), so the keyed dispatch -/// is bit-identical. +/// Samples carry full series keys. Replay uses the live worker's sample +/// dispatch so keyed identity and update semantics remain identical. /// /// Ordering contract: samples are consumed in the iteration order /// of the input `Vec`. §10.5 requires that the caller preserve @@ -101,15 +73,14 @@ pub fn build_backfilled_accumulator( samples: &[RawSample], ) -> Box { let mut updater: Box = create_accumulator_updater(config); - if updater.is_keyed() { - for s in samples { - let key = extract_aggregated_key(&s.labels, config); - updater.update_keyed(&key, s.value, s.timestamp_ms); - } - } else { - for s in samples { - updater.update_single(s.value, s.timestamp_ms); - } + for sample in samples { + apply_sample( + &mut *updater, + &sample.labels, + sample.value, + sample.timestamp_ms, + config, + ); } updater.take_accumulator() } @@ -151,6 +122,59 @@ mod tests { } } + // Replay must preserve each series and rank by the selected update mode. + #[test] + fn backfilled_topk_preserves_series_and_weight_mode() { + use crate::precompute_engine::operators::{ + CountMinSketchWithHeapAccumulator, CountSketchWithHeapAccumulator, + }; + for kind in [ + AggregationType::CountMinSketchWithHeap, + AggregationType::CountSketchWithHeap, + ] { + for mode in ["count", "value"] { + let mut config = sum_config(); + config.aggregation_type = kind; + config.parameters = serde_json::from_value(serde_json::json!({ + "d": 4, "w": 1024, "heap_size": 10, "weight_mode": mode + })) + .unwrap(); + let samples = vec![ + raw("m{svc=\"a\"}", 10, 100.0), + raw("m{svc=\"b\"}", 20, 2.0), + raw("m{svc=\"b\"}", 30, 3.0), + ]; + let acc = build_backfilled_accumulator(&config, &samples); + let mut ranked: Vec<(String, f64)> = if let Some(heap) = + acc.as_any() + .downcast_ref::() + { + heap.inner + .topk_heap_items() + .into_iter() + .map(|i| (i.key, i.value)) + .collect() + } else { + acc.as_any() + .downcast_ref::() + .unwrap() + .inner + .topk_heap_items() + .into_iter() + .map(|i| (i.key, i.value)) + .collect() + }; + ranked.sort_by(|a, b| b.1.total_cmp(&a.1)); + let expected = if mode == "count" { + vec![("m{svc=\"b\"}".into(), 2.0), ("m{svc=\"a\"}".into(), 1.0)] + } else { + vec![("m{svc=\"a\"}".into(), 100.0), ("m{svc=\"b\"}".into(), 5.0)] + }; + assert_eq!(ranked, expected, "{kind:?} {mode}"); + } + } + } + #[test] fn sum_accumulator_sums_all_samples_in_order() { let config = sum_config(); diff --git a/data_plane/src/storage_engines/sketch_db/data/mod.rs b/data_plane/src/storage_engines/sketch_db/data/mod.rs index 1d8aec4d..b150d2c3 100644 --- a/data_plane/src/storage_engines/sketch_db/data/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/data/mod.rs @@ -128,6 +128,18 @@ pub enum AggKind { }, } +/// Complete resolver identity for a configured materialization. All live and +/// replay paths must include policy semantics, not just the sketch family. +pub(crate) fn materialization_kind_for_config( + config: &asap_types::aggregation_config::AggregationConfig, +) -> String { + format!( + "{}|{}", + agg_kind_for_config(config).canonical_string(), + config.policy_fingerprint() + ) +} + /// Resolve the physical state family produced by a precompute policy. This is /// shared by SID minting and store registration so a sketch policy can never /// be minted as `ExactAgg` and later registered as `Sketch` (or vice versa). diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index b85c0524..2eea3068 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -1855,18 +1855,8 @@ impl SketchStore { // samples by sid up-front) skip the resolver round-trip by // invoking the sid-direct sibling. let (attrs_fp, _label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output); - let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg); - // Sid mint delegated to the caller's closure — typically - // `|m, fp, ak| series_resolver.resolve(m, fp, ak)`. Keeps the - // SketchStore free of any layer-inverted dependency on the - // resolver type (which lives in `drivers::ingest`). Tests - // pass either a real local resolver or a counter-mock - // closure. - let agg_kind_canonical = format!( - "{}|{}", - agg_kind.canonical_string(), - agg_cfg.policy_fingerprint() - ); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(agg_cfg); let sid = mint_sid(&agg_cfg.metric, &attrs_fp, &agg_kind_canonical); self.ingest_precompute_with_sid(sid, agg_cfg, output, accumulator) }