From d855e342a3309e4f079174322460ac27d6368fa3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 19:37:31 -0600 Subject: [PATCH 1/2] fix(metricsql): preserve metric identity through value rollups --- .../asap_query_engine/logical_dag.rs | 106 ++++++++++++- .../asap_query_engine/post_asap_readout.rs | 141 +++++++++++++++++- .../storage_engines/sketch_db/index/mod.rs | 7 + 3 files changed, 250 insertions(+), 4 deletions(-) diff --git a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs index b07ecf64a..f2c392724 100644 --- a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs +++ b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs @@ -320,7 +320,24 @@ impl Result> Evaluator<' })) } }; - value.map(|v| (no_name(labels), v)) + value.map(|v| { + let preserve_name = self.entry.language + == control_plane::query_plan::QueryLanguage::MetricsQl + && matches!( + operation, + TemporalOperation::Min + | TemporalOperation::Max + | TemporalOperation::Avg + ); + ( + if preserve_name { + labels + } else { + no_name(labels) + }, + v, + ) + }) }) .collect(), )) @@ -851,6 +868,93 @@ mod topk_tests { assert_eq!(stats.raw_scan_evaluations, 0); } + // A temporal operator over an external subquery follows the same language + // policy as a summary readout; changing execution placement cannot drop names. + #[test] + fn metricsql_temporal_subdag_preserves_names_only_for_value_rollups() { + use control_plane::query_plan::QueryLanguage; + for language in [QueryLanguage::PromQl, QueryLanguage::MetricsQl] { + for operation in [ + TemporalOperation::Max, + TemporalOperation::Min, + TemporalOperation::Avg, + TemporalOperation::Sum, + TemporalOperation::Count, + TemporalOperation::Rate, + ] { + let entry = QueryPlanEntry { + language, + query_id: "labels".into(), + canonical_query: "test".into(), + fixed_evaluation: None, + root: QueryNodeId(1), + nodes: BTreeMap::from([ + ( + QueryNodeId(0), + QueryPlanNode::Logical { + operator: LogicalOperator::ExactSubquery { + query: "m[1s]".into(), + }, + inputs: vec![], + }, + ), + ( + QueryNodeId(1), + QueryPlanNode::Logical { + operator: LogicalOperator::Temporal { operation }, + inputs: vec![QueryNodeId(0)], + }, + ), + ]), + instant: InstantExecution { + lookback_ms: 1000, + full_history: false, + cumulative_readout: true, + }, + fallback: FallbackPolicy::ExactBackend, + }; + let leaves = BTreeMap::from([( + (QueryNodeId(0), 1000), + PreparedLeaf { + value: Value::Matrix( + vec![( + labels(&[("__name__", "m"), ("job", "api")]), + vec![(100, 1.), (900, 3.)], + )], + 0, + 1000, + ), + remote: true, + remote_evaluations: 1, + remote_rpcs: 1, + }, + )]); + let (result, _) = execute_installed(&entry, &leaves, 1000, |_, _| { + panic!("external child supplied") + }) + .unwrap(); + let QueryResult::Vector(result) = result else { + panic!("vector required") + }; + let expected = language == QueryLanguage::MetricsQl + && matches!( + operation, + TemporalOperation::Max | TemporalOperation::Min | TemporalOperation::Avg + ); + assert_eq!( + result.values[0] + .label_keys_override + .as_ref() + .unwrap() + .iter() + .any(|name| name == "__name__"), + expected, + "{language:?} {operation:?}" + ); + } + } + } + #[test] fn installed_topk_ranks_exact_rate_summary_values() { let summary = QueryNodeId(0); diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index a908595dd..4b0808e2a 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -154,6 +154,8 @@ enum PhysicalNodeError { } struct PhysicalQueryRuntime<'a> { + language: control_plane::query_plan::QueryLanguage, + catalog: Option>, context: QueryExecutionContext<'a>, } @@ -182,10 +184,37 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { reduce_sum_values(grouping, values, *coverage) } QueryPlanNode::ReadMaterialization { binding } => { - let groups = self + let mut groups = self .context .read_bound_materialization(binding) .map_err(PhysicalNodeError::Store)?; + // Metric identity belongs to the shared DataDescriptor, not to + // the population labels or a reconstructed query string. + if self.language == control_plane::query_plan::QueryLanguage::MetricsQl + && binding.output_grouping + == control_plane::query_plan::PhysicalGrouping::PerEntity + { + let metric = self + .catalog + .as_ref() + .and_then(|catalog| { + let definition = + catalog.materializations.get(&binding.materialization)?; + catalog + .data_descriptors + .get(&definition.data_descriptor_id)? + .time_series_metric() + }) + .ok_or_else(|| { + PhysicalNodeError::Fallback( + "MetricsQL per-series readout requires catalog metric identity" + .into(), + ) + })?; + for (labels, _) in &mut groups { + labels.insert("__name__".into(), metric.into()); + } + } Ok(PhysicalQueryOutput::State { groups, item_labels: binding.item_labels.clone(), @@ -207,7 +236,17 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { .context .readout_bound(state, &query) .map_err(PhysicalNodeError::Store)?; - let (rows, row_coverage) = expand_item_readout(key, value, item_labels)?; + let (mut rows, row_coverage) = expand_item_readout(key, value, item_labels)?; + if self.language == control_plane::query_plan::QueryLanguage::MetricsQl + && !matches!( + query, + planner_types::post_asap::SketchQuery::Quantile { .. } + ) + { + for (labels, _) in &mut rows { + labels.remove("__name__"); + } + } fold_coverage(&mut coverage, row_coverage); values.extend(rows); } @@ -231,7 +270,17 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { ) .map(|value| { ( - key.clone(), + { + let mut labels = key.clone(); + if self.language + == control_plane::query_plan::QueryLanguage::MetricsQl + && *readout + != control_plane::query_plan::ExactReadout::Max + { + labels.remove("__name__"); + } + labels + }, SummaryValue::Points( vec![(self.context.t1_ms as i64, value)], state.exact_coverage(), @@ -536,6 +585,8 @@ fn execute_physical_query_payload( is_cumulative: bool, ) -> Result { let runtime = PhysicalQueryRuntime { + language: entry.language, + catalog: index.summary_catalog_snapshot(), context: QueryExecutionContext { index, t0_ms, @@ -891,6 +942,90 @@ mod tests { idx } + // The same installed summary follows each language's metric-name semantics; + // spatial reduction must not invent a source metric on the aggregate. + #[test] + fn metricsql_quantile_preserves_catalog_metric_name_only_per_entity() { + use control_plane::query_plan::*; + let config: asap_types::PrecomputeMaterialization = + serde_json::from_value(serde_json::json!({ + "aggregation_type": "DDSketch", "aggregation_sub_type": "", + "metric": "latency_ms", "window_size": 1, "slide_interval": 1, + "window_type": "tumbling", "num_aggregates_to_retain": 3, + "parameters": {"alpha": 0.01}, "pane_origin_ms": 0, + "partitioning": "per_entity", "window_layout": {"kind": "pane", "pane_secs": 1}, + "grouping_labels": {"labels": []}, "aggregated_labels": {"labels": []}, + "rollup_labels": {"labels": []}, "spatial_filter": "", + "spatial_filter_normalized": "", "original_yaml": "" + })) + .unwrap(); + let idx = ddsketch_fixture(); + let mut metadata = (*idx.instance(1).unwrap()).clone(); + metadata.policy_fp = config.policy_fingerprint(); + idx.install_summary_catalog(std::sync::Arc::new( + asap_types::summary_catalog::SummaryCatalog::from_materializations( + 1, + 1, + &[config.clone()], + ) + .unwrap(), + )) + .unwrap(); + idx.register(metadata); + let mut entry = QueryPlanEntry { + language: QueryLanguage::MetricsQl, + query_id: "quantile".into(), + canonical_query: "quantile_over_time(0.9, latency_ms[1s])".into(), + fixed_evaluation: None, + root: QueryNodeId(0), + nodes: BTreeMap::from([ + ( + QueryNodeId(0), + QueryPlanNode::SummaryEstimate { + input: QueryNodeId(1), + query: QueryReadout::Quantile { q: 0.9 }, + }, + ), + ( + QueryNodeId(1), + QueryPlanNode::ReadMaterialization { + binding: MaterializationBinding { + materialization: config.policy_fingerprint().into(), + output_grouping: PhysicalGrouping::PerEntity, + item_labels: vec![], + window_ms: 1000, + pane_origin_ms: Some(0), + readout_lookback_ms: Some(1000), + }, + }, + ), + ]), + instant: InstantExecution { + lookback_ms: 1000, + full_history: false, + cumulative_readout: true, + }, + fallback: FallbackPolicy::ExactBackend, + }; + let result = execute_query_plan_readout(&idx, &entry, 1000, 2000, true).unwrap(); + assert_eq!( + result.series[0].0.get("__name__").map(String::as_str), + Some("latency_ms") + ); + entry.language = QueryLanguage::PromQl; + let result = execute_query_plan_readout(&idx, &entry, 1000, 2000, true).unwrap(); + assert!(!result.series[0].0.contains_key("__name__")); + entry.language = QueryLanguage::MetricsQl; + let QueryPlanNode::ReadMaterialization { binding } = + entry.nodes.get_mut(&QueryNodeId(1)).unwrap() + else { + unreachable!() + }; + binding.output_grouping = PhysicalGrouping::Reduce(vec![]); + let result = execute_query_plan_readout(&idx, &entry, 1000, 2000, true).unwrap(); + assert!(!result.series[0].0.contains_key("__name__")); + } + #[test] fn formal_query_plan_executes_only_its_bound_policy() { let idx = SketchStore::new(); 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 8353d646c..cafb6eb8a 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -752,6 +752,13 @@ impl SketchStore { .map_err(|error| error.to_string()) } + /// Share the installed metadata snapshot without copying descriptors or state. + pub(crate) fn summary_catalog_snapshot( + &self, + ) -> Option> { + self.descriptors.authoritative_catalog() + } + /// Record that `sid` is a per-item (item_label-mode) frequency sketch /// keyed by the data-point attribute `label` (e.g. "service"). The /// query engine consults this to decide whether a keyed selector like From 195dd79a2d7ef4973e223ab9f074eb9bb380b390 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 10:47:24 -0600 Subject: [PATCH 2/2] fix(storage): keep one authoritative catalog snapshot accessor --- data_plane/src/storage_engines/sketch_db/index/mod.rs | 7 ------- 1 file changed, 7 deletions(-) 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 d6662caef..49c109d4b 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -993,13 +993,6 @@ impl SketchStore { self.descriptors.authoritative_catalog() } - /// Share the installed metadata snapshot without copying descriptors or state. - pub(crate) fn summary_catalog_snapshot( - &self, - ) -> Option> { - self.descriptors.authoritative_catalog() - } - /// Record that `sid` is a per-item (item_label-mode) frequency sketch /// keyed by the data-point attribute `label` (e.g. "service"). The /// query engine consults this to decide whether a keyed selector like