diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 91cc83cd..5dc7d406 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -3655,7 +3655,12 @@ fn collect_selected_materializations( candidates, values, .. } => { walk(candidates, readout, composable, grouping.clone(), selected)?; - walk(values, readout, composable, grouping.clone(), selected)?; + // In a hybrid TopK, the sketch is only a candidate-membership + // sidecar. Prometheus owns the authoritative value subtree; + // provisioning local exact state here duplicates that work. + if !composable { + walk(values, readout, composable, grouping.clone(), selected)?; + } } SummaryExpr::ValueOperation { child, .. } => { walk(child, readout, composable, grouping.clone(), selected)?; @@ -4125,6 +4130,53 @@ mod tests { assert_eq!(retained_partition_count(counter, Some(5)), 5); } + #[test] + fn hybrid_weighted_topk_installs_only_candidates_and_delegates_filtered_exact_values() { + use crate::query_plan::{logical::LogicalOperator, QueryPlanNode}; + let query = "topk(2, sum by (job) (rate(m[1m])))"; + let evidence = TopKMembershipEvidence { + selected_lower_bound: 101.0, + excluded_upper_bound: 100.0, + interval_failure_probability: 0.001, + observed_at_unix_ms: 9_500, + source: "unit-fixture".into(), + }; + let mut request = request_with_evidence("topk-rate", query, Some(evidence)).unwrap(); + request.hybrid_execution = true; + let mut environment = environment(10_000); + environment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite; + environment.collector_ids.clear(); + + let plan = PhysicalCompiler.compile(request, environment).unwrap(); + + assert_eq!(plan.precompute_plan.materializations.len(), 1); + assert_eq!( + plan.precompute_plan.materializations[0].aggregation_type, + asap_types::AggregationType::CountMinSketchWithHeap + ); + let entry = plan.query_plan.lookup(query).unwrap(); + let QueryPlanNode::CandidateTopK { inputs, .. } = &entry.nodes[&entry.root] else { + panic!("expected candidate TopK: {entry:#?}"); + }; + assert!(matches!( + &entry.nodes[&inputs[1]], + QueryPlanNode::Logical { + operator: LogicalOperator::CandidateExactSubquery { query, item_label }, + inputs: exact_inputs, + } if query == "sum by (job) (rate(m[1m]))" + && item_label == "job" + && exact_inputs == &vec![inputs[0]] + )); + assert!(entry.nodes.values().all(|node| !matches!( + node, + QueryPlanNode::ExactReadout { .. } + | QueryPlanNode::Logical { + operator: LogicalOperator::Scan { .. }, + .. + } + ))); + } + #[test] fn retained_footprint_rejects_eval_sized_cms_across_all_panes() { let evidence = TopKMembershipEvidence { diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index f280452a..1178077e 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -234,7 +234,11 @@ pub fn manifest( )); } if let crate::query_plan::QueryPlanNode::Logical { - operator: crate::query_plan::logical::LogicalOperator::ExactSubquery { query }, + operator: + crate::query_plan::logical::LogicalOperator::ExactSubquery { query } + | crate::query_plan::logical::LogicalOperator::CandidateExactSubquery { + query, .. + }, .. } = node { diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index c60d3a33..6642c07b 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -756,8 +756,60 @@ where }) }) .collect::, _>>()?; + let candidate_input = self.lower(candidates)?; + let value_input = if let Some(original) = &self.logical_source { + let parsed = promql_parser::parser::parse(original) + .map_err(|error| QueryPlanError::Invalid(error.to_string()))?; + let promql_parser::parser::Expr::Aggregate(aggregate) = parsed else { + return Err(QueryPlanError::Invalid( + "CandidateTopK requires a top-level PromQL aggregate".into(), + )); + }; + if aggregate.op.to_string() != "topk" { + return Err(QueryPlanError::Invalid( + "CandidateTopK requires a topk source expression".into(), + )); + } + fn item_label(node: &SummaryNode) -> Option { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => { + item_label(summary_input) + } + SummaryExpr::SummaryAgg { input, .. } => match &input.item { + Some(planner_types::post_asap::SummaryInputExpr::Column( + planner_types::pre_asap::ColumnRef::Named(label), + )) => Some(label.clone()), + Some(planner_types::post_asap::SummaryInputExpr::Column( + planner_types::pre_asap::ColumnRef::Qualified { name, .. }, + )) => Some(name.clone()), + _ => None, + }, + _ => None, + } + } + let item_label = item_label(candidates).ok_or_else(|| { + QueryPlanError::Invalid( + "CandidateTopK membership has no named item label".into(), + ) + })?; + let value_id = QueryNodeId(self.next_id); + self.next_id += 1; + self.nodes.insert( + value_id, + QueryPlanNode::Logical { + operator: logical::LogicalOperator::CandidateExactSubquery { + query: aggregate.expr.to_string(), + item_label, + }, + inputs: vec![candidate_input], + }, + ); + value_id + } else { + self.lower(values)? + }; QueryPlanNode::CandidateTopK { - inputs: [self.lower(candidates)?, self.lower(values)?], + inputs: [candidate_input, value_input], k: u64::try_from(*k).map_err(|_| { QueryPlanError::Invalid("CandidateTopK k exceeds u64".into()) })?, diff --git a/control_plane/src/query_plan/logical.rs b/control_plane/src/query_plan/logical.rs index 4c488a1b..4cd2c11f 100644 --- a/control_plane/src/query_plan/logical.rs +++ b/control_plane/src/query_plan/logical.rs @@ -16,6 +16,12 @@ pub enum LogicalOperator { ExactSubquery { query: String, }, + /// Prometheus exact subtree whose selectors are restricted at runtime by + /// the candidate vector produced by its single input. + CandidateExactSubquery { + query: String, + item_label: String, + }, Scan { metric: Option, matchers: Vec, @@ -141,6 +147,7 @@ impl LogicalOperator { pub fn validate(&self, inputs: usize) -> Result<(), QueryPlanError> { let expected = match self { Self::Scan { .. } | Self::ExactSubquery { .. } => 0, + Self::CandidateExactSubquery { .. } => 1, Self::Binary { .. } | Self::HistogramQuantile => 2, _ => 1, }; @@ -156,7 +163,7 @@ impl LogicalOperator { ) { return Err(invalid("zero range")); } - if let Self::ExactSubquery { query } = self { + if let Self::ExactSubquery { query } | Self::CandidateExactSubquery { query, .. } = self { let parsed = parser::parse(query).map_err(|e| invalid(e.to_string()))?; if matches!(parsed, Expr::MatrixSelector(_) | Expr::Subquery(_)) { return Err(invalid( @@ -1407,7 +1414,9 @@ pub fn externalize_residuals(entry: &mut QueryPlanEntry) -> Result<(), QueryPlan if let QueryPlanNode::Logical { operator, .. } = node { exact = matches!( operator, - LogicalOperator::Scan { .. } | LogicalOperator::ExactSubquery { .. } + LogicalOperator::Scan { .. } + | LogicalOperator::ExactSubquery { .. } + | LogicalOperator::CandidateExactSubquery { .. } ); } for child in node.inputs() { @@ -1422,7 +1431,8 @@ pub fn externalize_residuals(entry: &mut QueryPlanEntry) -> Result<(), QueryPlan if matches!( entry.nodes.get(&id), Some(QueryPlanNode::Logical { - operator: LogicalOperator::ExactSubquery { .. }, + operator: LogicalOperator::ExactSubquery { .. } + | LogicalOperator::CandidateExactSubquery { .. }, .. }) ) { diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 80109e2c..0ee2920d 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -1642,12 +1642,12 @@ async fn annotate_data_source(response: Response, data_source_id: &'static str) } if raw > 0 { ("failed", "invalid_provenance") + } else if remote > 0 && summary > 0 { + ("hybrid", "hybrid") } else if remote > 0 || summary == 0 { ( "exact_fallback", - if summary > 0 { - "hybrid" - } else if remote > 0 { + if remote > 0 { "external_exact" } else { "invalid_provenance" @@ -6795,14 +6795,14 @@ mod logical_provenance_tests { use super::*; #[tokio::test] - async fn hybrid_execution_is_fallback_with_measured_branch_counts() { + async fn hybrid_execution_has_its_own_route_with_measured_branch_counts() { // A successful mixed graph must never inherit the pure-ASAP route from its engine name. let response = Json(serde_json::json!({"status":"success", "warnings":[ "asap_execution:hybrid", "asap_logical_stats:raw=0,summary=1,memo_hits=3,remote=2,remote_rpcs=1,remote_branches=2" ], "data":{"resultType":"vector", "result":[]}})) .into_response(); let response = annotate_data_source(response, "asap_query").await; - assert_eq!(response.headers()["x-asap-execution"], "exact_fallback"); + assert_eq!(response.headers()["x-asap-execution"], "hybrid"); assert_eq!(response.headers()["x-asap-execution-detail"], "hybrid"); assert_eq!( response.headers()["x-asap-summary-readout-evaluations"], diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 54f67584..ba9dea9d 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -176,11 +176,54 @@ impl ASAPQueryEngine { physical.query_plan.plan_id, physical.query_plan.plan_version, )?; - super::exact_subqueries::prepare( + // Candidate-filtered exact cuts have a data dependency: read the + // installed membership subtree once, then use that vector to build the + // Prometheus selector. Keeping the result as a prepared leaf also means + // CandidateTopK reuses the same membership readout during composition. + let dependencies = super::exact_subqueries::candidate_dependencies(entry, times)?; + let mut prepared = super::logical_dag::PreparedLeaves::new(); + let unique_inputs = dependencies + .into_iter() + .map(|(_, input, at, _)| (input, at)) + .collect::>(); + for (input, at) in unique_inputs { + let evaluation_ms = u64::try_from(at).map_err(|_| { + crate::query_engines::EngineError::capability_miss( + "exact_subquery", + "candidate evaluation predates epoch", + ) + })?; + let mut subtree = entry.clone(); + subtree.root = input; + let reachable = subtree.topological_order().map_err(|error| { + crate::query_engines::EngineError::capability_miss( + "installed_logical_dag", + error.to_string(), + ) + })?; + subtree.nodes.retain(|id, _| reachable.contains(id)); + let (result, _) = self.execute_logical_entry( + physical, + &subtree, + &super::logical_dag::PreparedLeaves::new(), + evaluation_ms, + )?; + prepared.insert( + (input, at), + super::logical_dag::PreparedLeaf { + value: super::logical_dag::from_result(result)?, + remote: false, + remote_evaluations: 0, + remote_rpcs: 0, + }, + ); + } + super::exact_subqueries::prepare_with_candidates( entry, times, self.exact_subquery_endpoint.as_deref(), &self.exact_subquery_client, + prepared, ) .await } diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs index 749aecab..a3769e1f 100644 --- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs +++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs @@ -6,6 +6,9 @@ use control_plane::query_plan::{ }; use std::collections::{BTreeMap, BTreeSet}; +const MAX_CANDIDATE_VALUES: usize = 10_000; +const MAX_CANDIDATE_QUERY_BYTES: usize = 1_048_576; + fn miss(message: impl Into) -> EngineError { EngineError::capability_miss("exact_subquery", message.into()) } @@ -40,7 +43,8 @@ fn leaves( LogicalOperator::Scan { .. } => { return Err(miss("local raw Scan is forbidden in deployed plans")) } - LogicalOperator::ExactSubquery { .. } => { + LogicalOperator::ExactSubquery { .. } + | LogicalOperator::CandidateExactSubquery { .. } => { result.insert((id, at), operator.clone()); } LogicalOperator::Subquery { @@ -74,6 +78,11 @@ fn leaves( } _ => pending.extend(inputs.iter().map(|input| (*input, at))), }, + // CandidateTopK is a typed composition node rather than a Logical + // wrapper, but its value input can still be a Prometheus leaf. + QueryPlanNode::CandidateTopK { inputs, .. } => { + pending.extend(inputs.iter().map(|input| (*input, at))); + } // Existing bound summary subtrees are read by the synchronous callback. _ => {} } @@ -81,6 +90,107 @@ fn leaves( Ok(result) } +pub(super) fn candidate_dependencies( + entry: &QueryPlanEntry, + times: &[u64], +) -> Result, EngineError> { + let mut result = Vec::new(); + for ((id, at), operator) in leaves(entry, times)? { + if let LogicalOperator::CandidateExactSubquery { item_label, .. } = operator { + let input = *entry.nodes[&id] + .inputs() + .first() + .ok_or_else(|| miss("candidate exact subtree has no membership input"))?; + result.push((id, input, at, item_label)); + } + } + Ok(result) +} + +fn inject_candidate_matcher( + query: &str, + item_label: &str, + candidates: &[String], +) -> Result { + use promql_parser::{ + label::{MatchOp, Matcher}, + parser::Expr, + }; + // Go's regexp.QuoteMeta (used by Prometheus) escapes a smaller set than + // Rust's regex::escape; in particular, `\-` is not valid RE2 syntax. + fn re2_quote_meta(value: &str) -> String { + let mut quoted = String::with_capacity(value.len()); + for character in value.chars() { + match character { + '\\' | '.' | '+' | '*' | '?' | '(' | ')' | '|' | '[' | ']' | '{' | '}' | '^' + | '$' => { + quoted.push('\\'); + quoted.push(character); + } + '\n' => quoted.push_str("\\n"), + '\r' => quoted.push_str("\\r"), + '\t' => quoted.push_str("\\t"), + character if character.is_control() => { + quoted.push_str(&format!("\\x{{{:x}}}", character as u32)); + } + _ => quoted.push(character), + } + } + quoted + } + let pattern = format!( + "^(?:{})$", + candidates + .iter() + .map(|value| re2_quote_meta(value)) + .collect::>() + .join("|") + ); + // promql-parser's AST formatter writes programmatically constructed + // matcher values verbatim between double quotes. Store the PromQL string + // escapes separately from the compiled regex so its formatted AST remains + // valid for dots, quotes, backslashes, and other literal label bytes. + let promql_pattern = pattern.replace('\\', "\\\\").replace('"', "\\\""); + let matcher = Matcher::new( + MatchOp::Re(regex::Regex::new(&pattern).map_err(|error| miss(error.to_string()))?), + item_label, + &promql_pattern, + ); + fn visit(expr: &mut Expr, matcher: &Matcher) { + let append = |matchers: &mut promql_parser::label::Matchers| { + if matchers.or_matchers.is_empty() { + matchers.matchers.push(matcher.clone()); + } else { + for branch in &mut matchers.or_matchers { + branch.push(matcher.clone()); + } + } + }; + match expr { + Expr::VectorSelector(selector) => append(&mut selector.matchers), + Expr::MatrixSelector(selector) => append(&mut selector.vs.matchers), + Expr::Aggregate(node) => visit(&mut node.expr, matcher), + Expr::Unary(node) => visit(&mut node.expr, matcher), + Expr::Binary(node) => { + visit(&mut node.lhs, matcher); + visit(&mut node.rhs, matcher); + } + Expr::Paren(node) => visit(&mut node.expr, matcher), + Expr::Subquery(node) => visit(&mut node.expr, matcher), + Expr::Call(node) => { + for input in &mut node.args.args { + visit(input, matcher); + } + } + Expr::NumberLiteral(_) | Expr::StringLiteral(_) | Expr::Extension(_) => {} + } + } + let mut expression = promql_parser::parser::parse(query) + .map_err(|error| miss(format!("invalid candidate exact query: {error}")))?; + visit(&mut expression, &matcher); + Ok(expression.to_string()) +} + fn parse_result(body: &serde_json::Value, at: i64) -> Result { if body["status"] != "success" || body @@ -147,19 +257,76 @@ fn parse_result(body: &serde_json::Value, at: i64) -> Result } } +#[cfg(test)] pub(super) async fn prepare( entry: &QueryPlanEntry, times: &[u64], endpoint: Option<&str>, client: &reqwest::Client, ) -> Result { - let mut prepared = PreparedLeaves::new(); + prepare_with_candidates(entry, times, endpoint, client, PreparedLeaves::new()).await +} + +pub(super) async fn prepare_with_candidates( + entry: &QueryPlanEntry, + times: &[u64], + endpoint: Option<&str>, + client: &reqwest::Client, + mut prepared: PreparedLeaves, +) -> Result { // Equivalent exact cuts at the same time share one actual remote request. let mut remote_cache = BTreeMap::<(String, i64), Value>::new(); for ((id, at), operator) in leaves(entry, times)? { u64::try_from(at).map_err(|_| miss("subquery predates epoch"))?; - let query = match &operator { - LogicalOperator::ExactSubquery { query } => query.clone(), + let (query, candidate_filtered) = match &operator { + LogicalOperator::ExactSubquery { query } => (query.clone(), false), + LogicalOperator::CandidateExactSubquery { query, item_label } => { + let candidate_input = entry.nodes[&id].inputs()[0]; + let candidate = prepared + .get(&(candidate_input, at)) + .ok_or_else(|| miss("candidate membership was not prepared"))?; + let Value::Vector(rows) = &candidate.value else { + return Err(miss("candidate membership is not an instant vector")); + }; + let mut values = rows + .iter() + .map(|(labels, _)| { + labels.get(item_label).cloned().ok_or_else(|| { + miss(format!( + "candidate membership is missing item label {item_label}" + )) + }) + }) + .collect::, _>>()?; + values.sort(); + values.dedup(); + if values.is_empty() { + prepared.insert( + (id, at), + PreparedLeaf { + value: Value::Vector(Vec::new()), + remote: true, + remote_evaluations: 0, + remote_rpcs: 0, + }, + ); + continue; + } + if values.len() > MAX_CANDIDATE_VALUES { + return Err(miss(format!( + "candidate set has {} values, exceeding limit {MAX_CANDIDATE_VALUES}", + values.len() + ))); + } + let restricted = inject_candidate_matcher(query, item_label, &values)?; + if restricted.len() > MAX_CANDIDATE_QUERY_BYTES { + return Err(miss(format!( + "candidate-filtered exact query has {} bytes, exceeding limit {MAX_CANDIDATE_QUERY_BYTES}", + restricted.len() + ))); + } + (restricted, true) + } _ => return Err(miss("prepared leaf is not an exact subtree")), }; let key = (query.clone(), at); @@ -168,12 +335,21 @@ pub(super) async fn prepare( value.clone() } else { let endpoint = endpoint.ok_or_else(|| miss("Prometheus exact endpoint unavailable"))?; - let response = client - .get(format!("{}/api/v1/query", endpoint.trim_end_matches('/'))) - .query(&[ - ("query", query.as_str()), - ("time", &format!("{:.3}", at as f64 / 1000.0)), - ]) + let url = format!("{}/api/v1/query", endpoint.trim_end_matches('/')); + let time = format!("{:.3}", at as f64 / 1000.0); + // Candidate sets can be large enough to exceed proxy URL limits; + // Prometheus accepts the instant-query parameters as an encoded + // form body. Static exact cuts keep their existing GET contract. + let request = if candidate_filtered { + client + .post(url) + .form(&[("query", query.as_str()), ("time", time.as_str())]) + } else { + client + .get(url) + .query(&[("query", query.as_str()), ("time", time.as_str())]) + }; + let response = request .send() .await .map_err(|e| miss(format!("exact request failed: {e}")))?; @@ -225,6 +401,232 @@ mod tests { assert!(parse_result(&serde_json::json!({"status":"success","warnings":["partial"],"data":{"resultType":"scalar","result":[1,"2"]}}),1000).is_err()); assert!(parse_result(&serde_json::json!({"status":"success","data":{"resultType":"scalar","result":[2,"2"]}}),1000).is_err()); } + + fn candidate_entry(query: &str) -> QueryPlanEntry { + entry(BTreeMap::from([ + ( + QueryNodeId(0), + QueryPlanNode::Logical { + operator: LogicalOperator::CandidateExactSubquery { + query: query.into(), + item_label: "job".into(), + }, + inputs: vec![QueryNodeId(1)], + }, + ), + ( + QueryNodeId(1), + QueryPlanNode::SummaryMerge { inputs: vec![] }, + ), + ])) + } + + fn candidate_rows(values: impl IntoIterator) -> PreparedLeaves { + [( + (QueryNodeId(1), 1_000), + PreparedLeaf { + value: Value::Vector( + values + .into_iter() + .map(|value| (BTreeMap::from([("job".into(), value)]), 1.0)) + .collect(), + ), + remote: false, + remote_evaluations: 0, + remote_rpcs: 0, + }, + )] + .into_iter() + .collect() + } + + #[test] + fn candidate_matcher_is_ast_conjoined_and_regex_escaped() { + use promql_parser::parser::Expr; + let values = vec!["api.v1".into(), "quote\"slash\\".into(), "a|b".into()]; + let restricted = inject_candidate_matcher( + "sum by (job) (rate(m{cluster=\"prod\",job!=\"blocked\"}[5m]))", + "job", + &values, + ) + .unwrap(); + let expression = promql_parser::parser::parse(&restricted).unwrap(); + let Expr::Aggregate(aggregate) = expression else { + panic!("aggregate expected: {restricted}") + }; + let Expr::Call(call) = aggregate.expr.as_ref() else { + panic!("rate expected: {restricted}") + }; + let Expr::MatrixSelector(selector) = &*call.args.args[0] else { + panic!("matrix selector expected: {restricted}") + }; + assert!(selector + .vs + .matchers + .matchers + .iter() + .any(|matcher| matcher.name == "cluster" && matcher.is_match("prod"))); + assert!(selector + .vs + .matchers + .matchers + .iter() + .any(|matcher| matcher.name == "job" && !matcher.is_match("blocked"))); + let candidate_matcher = selector + .vs + .matchers + .matchers + .iter() + .find(|matcher| matcher.name == "job" && matcher.value.starts_with("^(?:")) + .unwrap(); + assert!(candidate_matcher.value.contains(r"api\\.v1")); + assert!(candidate_matcher.value.contains(r"a\\|b")); + assert!(restricted.contains(r#"quote\"slash\\\\"#)); + let semantic = regex::Regex::new(r#"^(?:api\.v1|quote"slash\\|a\|b)$"#).unwrap(); + for value in &values { + assert!(semantic.is_match(value), "{value:?}: {restricted}"); + } + assert!(!semantic.is_match("apiXv1")); + assert!(!semantic.is_match("a")); + } + + #[tokio::test] + async fn empty_candidate_set_returns_empty_exact_leaf_without_rpc() { + let entry = candidate_entry("sum by (job) (rate(m[5m]))"); + let prepared = prepare_with_candidates( + &entry, + &[1_000], + None, + &reqwest::Client::new(), + candidate_rows(Vec::new()), + ) + .await + .unwrap(); + let exact = &prepared[&(QueryNodeId(0), 1_000)]; + assert!(matches!(&exact.value, Value::Vector(rows) if rows.is_empty())); + assert_eq!((exact.remote_evaluations, exact.remote_rpcs), (0, 0)); + } + + #[tokio::test] + async fn candidate_exact_is_discovered_and_prepared_behind_candidate_topk_root() { + use control_plane::query_plan::{logical::Grouping, CandidateCompleteness}; + let mut entry = candidate_entry("sum by (job) (rate(m[5m]))"); + entry.nodes.insert( + QueryNodeId(2), + QueryPlanNode::CandidateTopK { + inputs: [QueryNodeId(1), QueryNodeId(0)], + k: 2, + grouping: Grouping { + labels: vec![], + without: false, + }, + completeness: CandidateCompleteness::BestEffort { guarantee: None }, + }, + ); + entry.root = QueryNodeId(2); + let dependencies = candidate_dependencies(&entry, &[1_000]).unwrap(); + assert_eq!( + dependencies, + vec![(QueryNodeId(0), QueryNodeId(1), 1_000, "job".into())] + ); + let prepared = prepare_with_candidates( + &entry, + &[1_000], + None, + &reqwest::Client::new(), + candidate_rows(Vec::new()), + ) + .await + .unwrap(); + assert!(prepared.contains_key(&(QueryNodeId(0), 1_000))); + } + + #[tokio::test] + async fn high_cardinality_candidates_use_one_post_and_preserve_exact_labels() { + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + let calls = Arc::new(AtomicUsize::new(0)); + let count = calls.clone(); + let app = axum::Router::new().route( + "/api/v1/query", + axum::routing::post( + move |axum::Form(params): axum::Form>| { + let count = count.clone(); + async move { + count.fetch_add(1, Ordering::SeqCst); + assert_eq!(params["time"], "1.000"); + let parsed = promql_parser::parser::parse(¶ms["query"]).unwrap(); + assert!(matches!(parsed, promql_parser::parser::Expr::Aggregate(_))); + assert!(params["query"].contains("job-999")); + axum::Json(serde_json::json!({ + "status":"success", + "data":{"resultType":"vector","result":[{ + "metric":{"__name__":"m","job":"job-999"}, + "value":[1,"42"] + }]} + })) + } + }, + ), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let entry = candidate_entry("sum by (job) (rate(m{job!=\"blocked\"}[5m]))"); + let candidates = (0..1_000).map(|index| format!("job-{index}")); + + let prepared = prepare_with_candidates( + &entry, + &[1_000], + Some(&format!("http://{address}")), + &reqwest::Client::new(), + candidate_rows(candidates), + ) + .await + .unwrap(); + + assert_eq!(calls.load(Ordering::SeqCst), 1); + let exact = &prepared[&(QueryNodeId(0), 1_000)]; + assert_eq!((exact.remote_evaluations, exact.remote_rpcs), (1, 1)); + assert!(matches!( + &exact.value, + Value::Vector(rows) + if rows.len() == 1 + && rows[0].0.get("job").map(String::as_str) == Some("job-999") + && rows[0].0.get("__name__").map(String::as_str) == Some("m") + )); + server.abort(); + } + + #[tokio::test] + async fn candidate_exact_rpc_failure_is_a_capability_miss_for_whole_query_fallback() { + let app = axum::Router::new().route( + "/api/v1/query", + axum::routing::post(|| async { axum::http::StatusCode::SERVICE_UNAVAILABLE }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let result = prepare_with_candidates( + &candidate_entry("sum by (job) (rate(m[5m]))"), + &[1_000], + Some(&format!("http://{address}")), + &reqwest::Client::new(), + candidate_rows(["api".into()]), + ) + .await; + let Err(error) = result else { + panic!("HTTP failure must reject the hybrid branch") + }; + assert!(matches!( + error, + EngineError::CapabilityMiss { engine_id: "exact_subquery", ref detail } + if detail.contains("HTTP 503") + )); + server.abort(); + } #[tokio::test] async fn exact_leaf_calls_prometheus_and_combines_with_prepared_summary() { // A successful exact branch remains an intermediate, not a whole-root fallback. 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 fbf5f19b..7c7f8196 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 @@ -187,7 +187,9 @@ impl Result> Evaluator<' QueryPlanNode::Logical { operator, inputs } => { if matches!( operator, - LogicalOperator::Scan { .. } | LogicalOperator::ExactSubquery { .. } + LogicalOperator::Scan { .. } + | LogicalOperator::ExactSubquery { .. } + | LogicalOperator::CandidateExactSubquery { .. } ) { return Err(miss( "installed Prometheus leaf was not prepared; backend raw execution is forbidden", @@ -235,7 +237,8 @@ impl Result> Evaluator<' .ok_or_else(|| miss("missing logical input")) }; match operator { - LogicalOperator::ExactSubquery { .. } => { + LogicalOperator::ExactSubquery { .. } + | LogicalOperator::CandidateExactSubquery { .. } => { Err(miss("Prometheus exact leaf was not prepared")) } LogicalOperator::Scan { .. } => { diff --git a/tools/o11y-execution/CALIBRATION.md b/tools/o11y-execution/CALIBRATION.md index 4d45f582..5217d19d 100644 --- a/tools/o11y-execution/CALIBRATION.md +++ b/tools/o11y-execution/CALIBRATION.md @@ -22,7 +22,7 @@ compares its complete quotes and installs the selected artifact. Record installation, ingestion/build/update, residency, and retirement CPU, and each query's inclusive process CPU over enough repetitions to exceed the operating system's accounting resolution. Validate correctness and record - `warm`, `exact_fallback`, or failure. Record memory and storage separately. + `warm`, `hybrid`, `exact_fallback`, or failure. Record memory and storage separately. Retain all raw artifacts. Do not reuse calibration timings as the independent held-out performance evaluation. 4. Run `update_global_profile.py --snapshot DISCOVERY --measurements MEASUREMENTS diff --git a/tools/o11y-execution/README.md b/tools/o11y-execution/README.md index c5c767f0..b95e487b 100644 --- a/tools/o11y-execution/README.md +++ b/tools/o11y-execution/README.md @@ -53,15 +53,15 @@ new output directory after investigating a failure. `planning.json` contains the envelope, cost comparison, lifecycle estimates and install request. `installed.json` records runtime status. `queries.json` preserves -every response, occurrence ID, timing and `warm`, `exact_fallback` or `failed` +every response, occurrence ID, timing and `warm`, `hybrid`, `exact_fallback` or `failed` classification. Forwarded responses carry a backend-owned `x-asap-execution` header; an unmarked success is not counted as warm or fallback. `ingestion.json` records accepted batches; acceptance does not prove worker completion. `execution_provenance` distinguishes summary-only `asap`, `hybrid`, and `external_exact`, and records summary readouts and Prometheus exact-subquery -requests. Hybrid execution remains under `exact_fallback`; it is never reported -as summary-only warm execution. A typed residual DAG can retain selected summary +requests. Hybrid execution has its own top-level classification; `exact_fallback` +means the whole query ran externally. A typed residual DAG can retain selected summary siblings, but a particular corpus may still produce no materializations. The runner waits for the finite-input completion barrier. The first traversal is diff --git a/tools/o11y-execution/calibrate.py b/tools/o11y-execution/calibrate.py index 1fba6183..dc5e7d1a 100644 --- a/tools/o11y-execution/calibrate.py +++ b/tools/o11y-execution/calibrate.py @@ -73,7 +73,7 @@ def calibrate(candidates, measurements, data_snapshot_id, observed_at_unix_ms, v count = row["evaluations"] if isinstance(count, bool) or not isinstance(count, int) or count <= 0: raise ValueError("query evaluation count must be positive") - if row.get("classification") not in ("warm", "exact_fallback") or not row.get("correct"): + if row.get("classification") not in ("warm", "hybrid", "exact_fallback") or not row.get("correct"): raise ValueError(f"candidate {pid}: {qid} failed execution/correctness validation") if not row.get("raw_measurement_file"): raise ValueError("each query requires a raw measurement artifact") diff --git a/tools/o11y-execution/calibrate_runtime.py b/tools/o11y-execution/calibrate_runtime.py index 17c1cc11..905053fb 100644 --- a/tools/o11y-execution/calibrate_runtime.py +++ b/tools/o11y-execution/calibrate_runtime.py @@ -156,7 +156,7 @@ def launch(name, command): runner.write_json(retirement, {"wait4_children_cpu_ns": total_cpu, "proc_before_retirement_cpu_ns": before_total, "note": "difference includes /proc tick rounding; nonnegative clamp below tick resolution"}) row["horizon_phases"]["retirement"] = {"cpu_ns": max(0, total_cpu - before_total), "raw_measurement_file": str(retirement.resolve())} - invalid = [qid for qid, q in row["queries"].items() if not q["correct"] or q["classification"] not in ("warm", "exact_fallback") or q["cpu_resolution_censored"]] + invalid = [qid for qid, q in row["queries"].items() if not q["correct"] or q["classification"] not in ("warm", "hybrid", "exact_fallback") or q["cpu_resolution_censored"]] if invalid: row["unavailable_reason"] = "failed/mixed/incorrect or CPU below measurement resolution: " + ",".join(invalid) else: @@ -207,6 +207,7 @@ def validate_candidate_topk_artifact(artifact): """Reject CandidateTopK plans whose membership sidecar is not locally installed.""" request = artifact.get("install_request", {}) schemas = {str(row["materialization"]): row for row in request.get("precompute_plan", {}).get("schemas", [])} + modes = set() for entry in request.get("query_plan", {}).get("entries", {}).values(): nodes = entry.get("nodes", {}) for node in nodes.values(): @@ -225,28 +226,59 @@ def validate_candidate_topk_artifact(artifact): heap_bindings.append(materialization) if not heap_bindings: raise ValueError("CandidateTopK membership input has no installed heap materialization") - if not any("exact" in json.dumps((schemas.get(mid) or {}).get("family", {})).lower() - and any(kind in json.dumps((schemas.get(mid) or {}).get("family", {})).lower() - for kind in ("counter", "rate", "increase")) - for mid in value_bindings): - raise ValueError("CandidateTopK value input has no installed ExactCounter materialization") + value_node = nodes.get(str(inputs[1]), {}) + operator = value_node.get("operator", {}) if value_node.get("op") == "logical" else {} + if operator.get("kind") == "candidate_exact_subquery": + modes.add("candidate_filtered_exact") + if value_node.get("inputs") != [inputs[0]]: + raise ValueError("candidate exact input must be the shared membership node") + if not operator.get("query") or not operator.get("item_label"): + raise ValueError("candidate exact operator lacks query or item label") + heap_families = [json.dumps(row.get("family", {}), sort_keys=True) + for row in schemas.values()] + if sum("CmsWithHeap" in family or "CountSketchWithHeap" in family + for family in heap_families) != 1 or len(heap_bindings) != 1: + raise ValueError("candidate-filtered TopK requires exactly one installed heap materialization") + if any("exact" in family.lower() and any(kind in family.lower() + for kind in ("counter", "rate", "increase")) for family in heap_families): + raise ValueError("candidate-filtered TopK must not install an ExactCounter materialization") + else: + modes.add("local_exact") + if not any("exact" in json.dumps((schemas.get(mid) or {}).get("family", {})).lower() + and any(kind in json.dumps((schemas.get(mid) or {}).get("family", {})).lower() + for kind in ("counter", "rate", "increase")) + for mid in value_bindings): + raise ValueError("CandidateTopK value input has no installed ExactCounter materialization") + return modes def validate_candidate_topk_execution(artifact, records): - has_candidate_topk = any(node.get("op") == "candidate_top_k" - for entry in artifact.get("install_request", {}).get("query_plan", {}).get("entries", {}).values() - for node in entry.get("nodes", {}).values()) - if not has_candidate_topk: + modes = validate_candidate_topk_artifact(artifact) + if not modes: return + if len(modes) != 1: + raise ValueError("mixed CandidateTopK execution contracts are not calibratable together") + mode = next(iter(modes)) for record in records: provenance = record.get("execution_provenance", {}) - if record.get("execution") != "warm": - raise ValueError("CandidateTopK execution was not warm") - for key in ("exact_subquery_rpcs", "exact_subquery_evaluations", "exact_branch_evaluations"): - if provenance.get(key, 0) != 0: - raise ValueError(f"CandidateTopK execution used exact path: {key}") - if provenance.get("summary_readout_evaluations", 0) < 2: - raise ValueError("CandidateTopK execution did not read both summary branches") + if provenance.get("raw_scan_evaluations", 0) not in (0, None): + raise ValueError("CandidateTopK execution used a forbidden local raw scan") + if mode == "candidate_filtered_exact": + if record.get("execution") != "hybrid" or provenance.get("detail") != "hybrid": + raise ValueError("candidate-filtered TopK did not report hybrid execution") + expected = {"summary_readout_evaluations": 1, "exact_subquery_rpcs": 1, + "exact_subquery_evaluations": 1, "exact_branch_evaluations": 1} + for key, value in expected.items(): + if provenance.get(key) != value: + raise ValueError(f"candidate-filtered TopK has invalid provenance: {key}") + else: + if record.get("execution") != "warm" or provenance.get("detail") not in (None, "asap"): + raise ValueError("local CandidateTopK execution was not warm") + for key in ("exact_subquery_rpcs", "exact_subquery_evaluations", "exact_branch_evaluations"): + if provenance.get(key, 0) != 0: + raise ValueError(f"CandidateTopK execution used exact path: {key}") + if provenance.get("summary_readout_evaluations", 0) < 2: + raise ValueError("CandidateTopK execution did not read both summary branches") def main(): diff --git a/tools/o11y-execution/compare.py b/tools/o11y-execution/compare.py index 497d0293..f8718a65 100644 --- a/tools/o11y-execution/compare.py +++ b/tools/o11y-execution/compare.py @@ -91,7 +91,7 @@ def summarize(rows): row.get("comparison", {}).get("relative_tolerance", 0.0), row.get("comparison", {}).get("absolute_tolerance", 0.0)) for row in rows] eligible = bool(rows) and all( - c["equal"] and r["execution"] in ("warm", "exact_fallback") and r["exact"].get("http_status") == 200 + c["equal"] and r["execution"] in ("warm", "hybrid", "exact_fallback") and r["exact"].get("http_status") == 200 for r, c in zip(rows, comparisons)) actual = sum(r["elapsed_ns"] for r in rows) exact = sum(r.get("exact", {}).get("elapsed_ns", 0) for r in rows) @@ -107,9 +107,9 @@ def cpu_total(requests, names): "baseline_over_backend_cpu_ratio": exact_cpu / backend_cpu if eligible and backend_cpu and exact_cpu is not None else None, "cpu_scope": "request intervals, whole processes including background work; fallback CPU charged to backend; /proc tick granularity", "occurrences": len(rows), - "execution_counts": {k: sum(r["execution"] == k for r in rows) for k in ("warm", "exact_fallback", "failed")}, + "execution_counts": {k: sum(r["execution"] == k for r in rows) for k in ("warm", "hybrid", "exact_fallback", "failed")}, "execution_detail_counts": {k: sum(r.get("execution_provenance", {}).get("detail") == k for r in rows) for k in ("asap", "hybrid", "external_exact")}, - "summary_acceleration_scope": "warm contains summary-only execution; hybrid is reported separately under exact_fallback", + "summary_acceleration_scope": "warm is summary-only; hybrid combines summary candidates with a filtered exact subtree", "equal_results": sum(c["equal"] for c in comparisons), "uncomparable_results": sum(not c["comparable"] for c in comparisons), "comparisons": comparisons, diff --git a/tools/o11y-execution/replay.py b/tools/o11y-execution/replay.py index 02e7baad..a28bd39b 100644 --- a/tools/o11y-execution/replay.py +++ b/tools/o11y-execution/replay.py @@ -39,11 +39,11 @@ def classify(response, headers=None): if response.get("status") != "success": return "failed" declared = (headers or {}).get("x-asap-execution") - if declared in ("exact_fallback", "failed"): + if declared in ("hybrid", "exact_fallback", "failed"): return declared detail = (headers or {}).get("x-asap-execution-detail") if detail == "hybrid": - return "exact_fallback" + return "hybrid" if detail == "invalid_provenance": return "failed" sources = {x for x in response.get("infos", []) if isinstance(x, str) and x.startswith("data_source:")} @@ -59,7 +59,7 @@ def execution_provenance(response, headers=None): route = classify(response, headers) detail = headers.get("x-asap-execution-detail") if detail is None: - detail = "asap" if route == "warm" else "external_exact" if route == "exact_fallback" else "failed" + detail = "asap" if route == "warm" else "hybrid" if route == "hybrid" else "external_exact" if route == "exact_fallback" else "failed" counts = {} for name, header in (("raw_scan_evaluations", "x-asap-raw-scan-evaluations"), ("summary_readout_evaluations", "x-asap-summary-readout-evaluations"), @@ -454,7 +454,7 @@ def disk_bytes(path): "scope": "logical file bytes including WAL; backend output also contains logs; concurrent snapshots are approximate"} write_json(args.output / "storage.json", storage) write_json(args.output / "completion.json", {"complete": True, - "execution_counts": {k: sum(r["execution"] == k for r in results) for k in ["warm", "exact_fallback", "failed"]}, + "execution_counts": {k: sum(r["execution"] == k for r in results) for k in ["warm", "hybrid", "exact_fallback", "failed"]}, "execution_detail_counts": {k: sum(r["execution_provenance"]["detail"] == k for r in results) for k in ["asap", "hybrid", "external_exact", "failed", "invalid_provenance"]}, "benefit_claim": None}) if args.compare: @@ -464,7 +464,7 @@ def disk_bytes(path): "by_query_occurrence": {query["id"]: summarize([r for r in results if r["id"] == query["id"]]) for query in queries}, "by_execution": {route: summarize([r for r in results if r["execution"] == route]) - for route in ["warm", "exact_fallback", "failed"]}, + for route in ["warm", "hybrid", "exact_fallback", "failed"]}, "by_execution_detail": {detail: summarize([r for r in results if r["execution_provenance"]["detail"] == detail]) for detail in ["asap", "hybrid", "external_exact"]}, "estimated_cost": plan["cost_comparison"], diff --git a/tools/o11y-execution/test_calibrate_runtime.py b/tools/o11y-execution/test_calibrate_runtime.py index 24415f43..f8d3533d 100644 --- a/tools/o11y-execution/test_calibrate_runtime.py +++ b/tools/o11y-execution/test_calibrate_runtime.py @@ -21,6 +21,18 @@ def artifact(self, membership): ]}, }} + def candidate_filtered_artifact(self): + artifact = self.artifact({"op": "read_materialization", "binding": {"materialization": 7}}) + request = artifact["install_request"] + request["query_plan"]["entries"]["q"]["nodes"]["3"] = { + "op": "logical", + "operator": {"kind": "candidate_exact_subquery", "query": "sum by (job) (rate(m[5m]))", + "item_label": "job"}, + "inputs": [1], + } + request["precompute_plan"]["schemas"] = request["precompute_plan"]["schemas"][:1] + return artifact + def test_rejects_exact_membership_fallback(self): with self.assertRaisesRegex(ValueError, "contains ExactFallback"): validate_candidate_topk_artifact(self.artifact({"op": "exact_fallback", "reason": "unsupported"})) @@ -67,6 +79,37 @@ def test_requires_two_local_summary_readouts_at_runtime(self): validate_candidate_topk_execution(artifact, [{"execution": "warm", "execution_provenance": { **provenance, "exact_subquery_rpcs": 1}}]) + def test_accepts_one_heap_and_candidate_filtered_external_exact(self): + validate_candidate_topk_artifact(self.candidate_filtered_artifact()) + + def test_candidate_filtered_contract_rejects_local_exact_state_or_unshared_input(self): + artifact = self.candidate_filtered_artifact() + artifact["install_request"]["precompute_plan"]["schemas"].append( + {"materialization": 8, "family": {"family": "exact", "kind": "increase"}}) + with self.assertRaisesRegex(ValueError, "must not install"): + validate_candidate_topk_artifact(artifact) + artifact = self.candidate_filtered_artifact() + artifact["install_request"]["query_plan"]["entries"]["q"]["nodes"]["3"]["inputs"] = [2] + with self.assertRaisesRegex(ValueError, "shared membership"): + validate_candidate_topk_artifact(artifact) + + def test_candidate_filtered_execution_requires_hybrid_one_rpc_and_one_summary_read(self): + artifact = self.candidate_filtered_artifact() + provenance = {"detail": "hybrid", "raw_scan_evaluations": 0, + "summary_readout_evaluations": 1, "exact_subquery_rpcs": 1, + "exact_subquery_evaluations": 1, "exact_branch_evaluations": 1} + validate_candidate_topk_execution( + artifact, [{"execution": "hybrid", "execution_provenance": provenance}]) + for key in ("summary_readout_evaluations", "exact_subquery_rpcs", + "exact_subquery_evaluations", "exact_branch_evaluations"): + with self.subTest(key=key), self.assertRaisesRegex(ValueError, "invalid provenance"): + validate_candidate_topk_execution(artifact, [{"execution": "hybrid", + "execution_provenance": {**provenance, key: 0}}]) + with self.assertRaisesRegex(ValueError, "hybrid execution"): + validate_candidate_topk_execution( + artifact, [{"execution": "hybrid", "execution_provenance": { + **provenance, "detail": "external_exact"}}]) + if __name__ == "__main__": unittest.main() diff --git a/tools/o11y-execution/test_replay.py b/tools/o11y-execution/test_replay.py index 7e79a462..78f33a01 100644 --- a/tools/o11y-execution/test_replay.py +++ b/tools/o11y-execution/test_replay.py @@ -22,9 +22,9 @@ def test_binding_or_unattributed_success_is_not_execution(self): def test_hybrid_is_not_counted_as_pure_summary_acceleration(self): response = {"status": "success", "infos": ["data_source: asap_query"]} - headers = {"x-asap-execution": "exact_fallback", "x-asap-execution-detail": "hybrid", + headers = {"x-asap-execution": "hybrid", "x-asap-execution-detail": "hybrid", "x-asap-raw-scan-evaluations": "2", "x-asap-summary-readout-evaluations": "1"} - self.assertEqual(classify(response, headers), "exact_fallback") + self.assertEqual(classify(response, headers), "hybrid") self.assertEqual(execution_provenance(response, headers)["summary_readout_evaluations"], 1) self.assertEqual(classify(response, {"x-asap-execution": "failed"}), "failed") diff --git a/tools/o11y-execution/update_global_profile.py b/tools/o11y-execution/update_global_profile.py index e7e025c7..547b1ae4 100644 --- a/tools/o11y-execution/update_global_profile.py +++ b/tools/o11y-execution/update_global_profile.py @@ -25,7 +25,7 @@ def phase(name): reads = [] for row in rows: for query in row["queries"].values(): - if query["evaluations"] <= 0 or not query.get("correct") or query.get("classification") not in ("warm", "exact_fallback"): + if query["evaluations"] <= 0 or not query.get("correct") or query.get("classification") not in ("warm", "hybrid", "exact_fallback"): raise ValueError("profile requires successful correct query measurements") reads.append(nonnegative(query["cpu_ns"], "query CPU") / query["evaluations"]) costs = {"build": phase("install"), "maintenance_per_update": phase("ingest_and_build") / sample_count,