From 6c6511d1b5f95449fde3a9eabc9736bff689c08d Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 10:03:02 -0400 Subject: [PATCH 1/2] feat(planner): --planner saved for SQL (#856) An e2e query is now written in `promql` or `sql`; --query-language picks which ones are planned. In SQL mode each `records` metric is a table (labels as metadata columns, values as value columns, time column `ts`), and a SQL query is one leaf that may read several RQEs, all of which must be served by one deployment. Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/main.rs | 13 +- asap-planner-rs/src/optimizer/saved_plan.rs | 350 +++++++++++++++++--- 2 files changed, 320 insertions(+), 43 deletions(-) diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index 929adfa2..f0056e44 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -59,7 +59,7 @@ struct Args { /// `milp` plans with sketch-bench's rqe-optimizer: PromQL or SQL with /// --input_config only, labels from its `metrics:` hints or tables' - /// `metadata_columns`. `saved` writes a PromQL plan sketch-bench's + /// `metadata_columns`. `saved` writes a PromQL or SQL plan sketch-bench's /// `autosketch_vs_asap` already solved. #[arg(long, value_enum, default_value = "legacy")] planner: PlannerArg, @@ -70,7 +70,8 @@ struct Args { plan: Option, /// Saved only. The eval workload table the plan was solved for; its - /// `e2e` block gives the PromQL queries and metric labels. + /// `e2e` block gives the queries in --query-language and the metrics' + /// labels. #[arg(long = "workload-table", required_if_eq("planner", "saved"))] workload_table: Option, @@ -339,8 +340,11 @@ fn run_milp(args: &Args) -> anyhow::Result<()> { fn run_saved(args: &Args) -> anyhow::Result<()> { anyhow::ensure!( - args.query_language == QueryLanguage::promql, - "--planner saved supports only --query-language promql" + matches!( + args.query_language, + QueryLanguage::promql | QueryLanguage::sql + ), + "--planner saved supports only --query-language promql or sql" ); let scrape_interval_ms = check_optimizer_flags(args, "saved")?; anyhow::ensure!( @@ -359,6 +363,7 @@ fn run_saved(args: &Args) -> anyhow::Result<()> { weights: args.weights.as_deref().expect("clap requires --weights"), sla_ms: args.sla_ms, }, + args.query_language, scrape_interval_ms, )? .write_to_dir(&args.output_dir)?; diff --git a/asap-planner-rs/src/optimizer/saved_plan.rs b/asap-planner-rs/src/optimizer/saved_plan.rs index 4cf402cd..fec276ca 100644 --- a/asap-planner-rs/src/optimizer/saved_plan.rs +++ b/asap-planner-rs/src/optimizer/saved_plan.rs @@ -1,18 +1,21 @@ //! Writes a plan solved elsewhere, sketch-bench `autosketch_vs_asap`'s plan //! file, as the planner's streaming and inference YAML, without re-solving. -//! The workload comes from the eval table's `e2e` block: its PromQL queries, -//! the plan RQEs each reads, and the metrics' labels. +//! The workload comes from the eval table's `e2e` block: its PromQL or SQL +//! queries, the plan RQEs each reads, and the metrics' labels. use std::collections::{BTreeMap, HashMap}; use std::io::Read; use std::path::Path; use anyhow::{bail, ensure, Context}; +use asap_types::enums::QueryLanguage; use rqe_optimizer::{heap_capacity, AtomicCostEntry, Deployment, LabelSet, Raqe}; use serde::Deserialize; use serde_json::{json, Value}; -use crate::config::input::ControllerConfig; +use crate::config::input::{ + ControllerConfig, ControllerOptions, SQLControllerConfig, SQLQueryGroup, TableDefinition, +}; use crate::planner_output::PlannerOutput; use super::aqe_extractor::decompose_to_leaves; @@ -35,17 +38,49 @@ struct E2e { metrics: BTreeMap, } +/// The time column of the `records` tables the e2e SQL queries read. +const SQL_TIME_COLUMN: &str = "ts"; + +/// The metric type whose metrics are SQL tables. +const RECORDS_METRIC_TYPE: &str = "records"; + +/// One query, written in exactly one of PromQL and SQL. #[derive(Deserialize)] struct E2eQuery { - promql: String, + promql: Option, + sql: Option, interval_secs: u64, - /// Plan RQE ids, one per leaf of `promql` in order. + /// Plan RQE ids: for `promql`, one per leaf in order; for `sql`, a single + /// leaf, every RQE it reads. rqes: Vec, } +impl E2eQuery { + /// The query if it is written in `language`. + fn text(&self, language: QueryLanguage) -> anyhow::Result> { + let (promql, sql) = match (&self.promql, &self.sql) { + (Some(_), Some(_)) | (None, None) => bail!( + "e2e query reading {:?} needs exactly one of `promql` and `sql`", + self.rqes + ), + (promql, sql) => (promql.as_deref(), sql.as_deref()), + }; + Ok(match language { + QueryLanguage::promql => promql, + QueryLanguage::sql => sql, + other => bail!("--planner saved supports only promql and sql, not {other:?}"), + }) + } +} + #[derive(Deserialize)] struct E2eMetric { + #[serde(rename = "type")] + kind: String, labels: Vec, + /// A `records` metric's value fields: its table's value columns. + #[serde(default)] + values: BTreeMap, } #[derive(Deserialize)] @@ -89,29 +124,72 @@ pub struct PlanSelector<'a> { pub sla_ms: Option, } -/// The controller config the table's `e2e` block describes: one query group -/// per repeat interval. -fn e2e_config(e2e: &E2e) -> anyhow::Result { - let mut by_interval: BTreeMap> = BTreeMap::new(); - for q in &e2e.queries { - by_interval +/// The selected queries' texts by repeat interval in seconds: one query group +/// each. +fn by_interval(selected: &[(&E2eQuery, &str)]) -> BTreeMap> { + let mut groups: BTreeMap> = BTreeMap::new(); + for (q, text) in selected { + groups .entry(q.interval_secs) .or_default() - .push(&q.promql); + .push(text.to_string()); } + groups +} + +/// The PromQL controller config of the selected queries, with every +/// non-`records` metric's labels as hints. +fn e2e_promql_config( + selected: &[(&E2eQuery, &str)], + metrics: &BTreeMap, +) -> anyhow::Result { let config = json!({ - "query_groups": by_interval.iter().map(|(secs, queries)| json!({ + "query_groups": by_interval(selected).iter().map(|(secs, queries)| json!({ "queries": queries, "repetition_delay_ms": secs * 1000, })).collect::>(), - "metrics": e2e.metrics.iter().map(|(metric, m)| json!({ - "metric": metric, - "labels": m.labels.iter().map(|l| &l.name).collect::>(), - })).collect::>(), + "metrics": metrics.iter().filter(|(_, m)| m.kind != RECORDS_METRIC_TYPE) + .map(|(metric, m)| json!({ + "metric": metric, + "labels": m.labels.iter().map(|l| &l.name).collect::>(), + })).collect::>(), }); Ok(serde_json::from_value(config)?) } +/// The SQL controller config of the selected queries: each `records` metric +/// is a table, its labels the metadata columns and its values the value +/// columns. +fn e2e_sql_config( + selected: &[(&E2eQuery, &str)], + metrics: &BTreeMap, +) -> SQLControllerConfig { + SQLControllerConfig { + query_groups: by_interval(selected) + .into_iter() + .map(|(secs, queries)| SQLQueryGroup { + id: None, + queries, + repetition_delay_ms: secs * 1000, + controller_options: ControllerOptions::default(), + }) + .collect(), + tables: metrics + .iter() + .filter(|(_, m)| m.kind == RECORDS_METRIC_TYPE) + .map(|(name, m)| TableDefinition { + name: name.clone(), + time_column: SQL_TIME_COLUMN.to_string(), + value_columns: m.values.keys().cloned().collect(), + metadata_columns: m.labels.iter().map(|l| l.name.clone()).collect(), + }) + .collect(), + windowing: None, + sketch_parameters: None, + aggregate_cleanup: None, + } +} + fn read_json(path: &Path) -> anyhow::Result { let bytes = std::fs::read(path).with_context(|| format!("reading {}", path.display()))?; let text = if bytes.starts_with(&[0x1f, 0x8b]) { @@ -125,11 +203,13 @@ fn read_json(path: &Path) -> anyhow::Result { } /// Streaming and inference YAML for the selected plan of `plan_path` (JSON, -/// gzipped or not) over the workload of `table_path`. +/// gzipped or not) over the workload of `table_path`: its queries written in +/// `language`. pub fn saved_plan_output( table_path: &Path, plan_path: &Path, selector: &PlanSelector, + language: QueryLanguage, scrape_interval_ms: u64, ) -> anyhow::Result { let table: Table = read_json(table_path)?; @@ -173,28 +253,58 @@ pub fn saved_plan_output( c.rqe ); } - let promql = e2e_config(e2e)?; - let config = MilpConfig::Promql(&promql); + let mut selected: Vec<(&E2eQuery, &str)> = Vec::new(); + for q in &e2e.queries { + if let Some(text) = q.text(language)? { + selected.push((q, text)); + } + } + ensure!( + !selected.is_empty(), + "{}: no e2e query is written in {language:?}", + table_path.display() + ); + let (promql, sql); + let config = match language { + QueryLanguage::sql => { + sql = e2e_sql_config(&selected, &e2e.metrics); + MilpConfig::Sql(&sql) + } + _ => { + promql = e2e_promql_config(&selected, &e2e.metrics)?; + MilpConfig::Promql(&promql) + } + }; // Before the leaf split, which turns an avg into two leaves. reject_unwritable_queries(config)?; - // Leaf query string and repeat interval -> the plan RQE it reads. - let mut rqe_of: HashMap<(String, u64), &str> = HashMap::new(); - for q in &e2e.queries { - let leaves = decompose_to_leaves(&q.promql); - ensure!( - leaves.len() == q.rqes.len(), - "query {:?} has {} leaves but names {} RQEs", - q.promql, - leaves.len(), - q.rqes.len() - ); - for (leaf, rqe) in leaves.into_iter().zip(&q.rqes) { + // Leaf query string and repeat interval -> the plan RQEs it reads. + let mut rqes_of: HashMap<(String, u64), Vec<&str>> = HashMap::new(); + for &(q, text) in &selected { + let rqes = q.rqes.iter().map(String::as_str); + let leaves: Vec<(String, Vec<&str>)> = match config { + // A SQL query is one leaf. + MilpConfig::Sql(_) => { + ensure!(!q.rqes.is_empty(), "query {text:?} names no RQEs"); + vec![(text.to_string(), rqes.collect())] + } + MilpConfig::Promql(_) => { + let leaves = decompose_to_leaves(text); + ensure!( + leaves.len() == q.rqes.len(), + "query {text:?} has {} leaves but names {} RQEs", + leaves.len(), + q.rqes.len() + ); + leaves.into_iter().zip(rqes.map(|r| vec![r])).collect() + } + }; + for (leaf, rqes) in leaves { let key = (leaf, q.interval_secs * 1000); - if let Some(other) = rqe_of.insert(key.clone(), rqe) { + if let Some(other) = rqes_of.insert(key.clone(), rqes.clone()) { ensure!( - other == rqe, - "leaf {:?} every {} ms reads both {other:?} and {rqe:?}", + other == rqes, + "leaf {:?} every {} ms reads both {other:?} and {rqes:?}", key.0, key.1 ); @@ -211,13 +321,18 @@ pub fn saved_plan_output( let mut served_by = Vec::with_capacity(workload.raqes.len()); for (raqe, &item) in workload.raqes.iter().zip(&workload.raqe_items) { let item = &workload.items[item]; - // An item's leaves (say p90 and p50 over one window) read one - // deployment, so their picks must name the same sketch. + // An item's leaves (say p90 and p50 over one window), and every RQE + // a SQL leaf reads, read one deployment, so their picks must name the + // same sketch. let mut index = None; + let mut reads: Vec<(&String, &str)> = Vec::new(); for leaf in &item.query_strings { - let rqe = rqe_of + let rqes = rqes_of .get(&(leaf.clone(), item.t_repeat_ms)) .with_context(|| format!("leaf {leaf:?} has no RQE in the table's e2e block"))?; + reads.extend(rqes.iter().map(|&rqe| (leaf, rqe))); + } + for (leaf, rqe) in reads { let pick = chosen .get(rqe) .with_context(|| format!("the plan doesn't serve RQE {rqe:?}"))?; @@ -358,6 +473,15 @@ mod tests { } fn run_on(table: &str, plan: &[u8], sla_ms: Option) -> anyhow::Result { + run_in(table, plan, sla_ms, QueryLanguage::promql) + } + + fn run_in( + table: &str, + plan: &[u8], + sla_ms: Option, + language: QueryLanguage, + ) -> anyhow::Result { let dir = tempfile::tempdir().unwrap(); let (table_path, plan_path) = (dir.path().join("table.json"), dir.path().join("plan.json")); std::fs::write(&table_path, table).unwrap(); @@ -367,7 +491,7 @@ mod tests { weights: "cpu", sla_ms, }; - saved_plan_output(&table_path, &plan_path, &selector, 1000) + saved_plan_output(&table_path, &plan_path, &selector, language, 1000) } fn refusal(plan: Value, sla_ms: Option) -> String { @@ -550,6 +674,154 @@ mod tests { assert!(err.contains("different sketches"), "{err}"); } + /// An `all`-style table: a classic PromQL query, and SQL queries on the + /// `records` table `http`. The distinct-users query reads two RQEs. + const MIXED_TABLE: &str = r#"{"workloads": [{"e2e": { + "queries": [ + {"promql": "sum by (job) (data)", "interval_secs": 1, "rqes": ["sum"]}, + {"sql": "SELECT service, COUNT(DISTINCT user_id) FROM http WHERE ts BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY service", + "interval_secs": 60, "rqes": ["users_1pct", "users_5pct"]}, + {"sql": "SELECT region, QUANTILE(0.99, latency) FROM http WHERE ts BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY region", + "interval_secs": 60, "rqes": ["p99"]} + ], + "metrics": { + "data": {"type": "gauge", "labels": [{"name": "job"}, {"name": "label_0"}]}, + "http": {"type": "records", "labels": [{"name": "region"}, {"name": "service"}], + "values": {"latency": {}, "user_id": {}}} + } + }}]}"#; + + /// The distinct-users RQEs share an HLL rolled up from (region, service). + fn mixed_plan(users_5pct_deployment: usize) -> Value { + let hll = |rqe: &str, deployment: usize| { + let mut p = pick(rqe, deployment, "hll", json!({"lg_k": 12}), 60); + p["deployment_grouping"] = json!(["region", "service"]); + p + }; + json!({"plans": [{"method": "asap", "weights": "cpu", "kind": "cheapest", "chosen": [ + pick("sum", 3, "exact-sum", json!({}), 1), + hll("users_1pct", 1), + hll("users_5pct", users_5pct_deployment), + pick("p99", 0, "dd", json!({"alpha": 0.01}), 300), + ]}]}) + } + + fn run_mixed(plan: Value, language: QueryLanguage) -> anyhow::Result { + run_in(MIXED_TABLE, plan.to_string().as_bytes(), None, language) + } + + #[test] + fn writes_the_sql_queries_over_records_tables() { + let output = run_mixed(mixed_plan(1), QueryLanguage::sql).unwrap(); + + let inference = output.to_inference_config(QueryLanguage::sql).unwrap(); + let mut queries: Vec<&str> = inference + .query_configs + .iter() + .map(|q| q.query.as_str()) + .collect(); + queries.sort(); + assert_eq!( + queries, + [ + "SELECT region, QUANTILE(0.99, latency) FROM http WHERE ts BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY region", + "SELECT service, COUNT(DISTINCT user_id) FROM http WHERE ts BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY service", + ] + ); + let streaming = output.to_streaming_config(QueryLanguage::sql).unwrap(); + let mut aggregations: Vec<_> = streaming + .aggregation_configs + .values() + .map(|a| { + ( + a.aggregation_type, + a.grouping_labels.labels.clone(), + a.table_name.clone(), + a.value_column.clone(), + ) + }) + .collect(); + aggregations.sort_by_key(|a| a.3.clone()); + assert_eq!( + aggregations, + [ + ( + promql_utilities::query_logics::enums::AggregationType::DDSketch, + vec!["region".to_string()], + Some("http".to_string()), + Some("latency".to_string()), + ), + ( + promql_utilities::query_logics::enums::AggregationType::HLL, + vec!["region".to_string(), "service".to_string()], + Some("http".to_string()), + Some("user_id".to_string()), + ), + ] + ); + } + + /// The PromQL run of the same table skips its SQL queries. + #[test] + fn writes_only_the_promql_queries_of_a_mixed_table() { + let output = run_mixed(mixed_plan(1), QueryLanguage::promql).unwrap(); + assert_eq!(output.inference_query_count(), 1); + assert_eq!(output.streaming_aggregation_count(), 1); + } + + #[test] + fn refuses_a_sql_query_whose_rqes_are_on_two_deployments() { + let err = run_mixed(mixed_plan(2), QueryLanguage::sql) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("served by two deployments"), "{err}"); + } + + #[test] + fn refuses_a_sql_pick_from_another_table() { + let mut plan = mixed_plan(1); + // A 7s window doesn't divide the 5m lookback. + plan["plans"][0]["chosen"][3] = pick("p99", 0, "dd", json!({"alpha": 0.01}), 7); + let err = run_mixed(plan, QueryLanguage::sql) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("does --plan match --workload-table"), "{err}"); + } + + #[test] + fn refuses_a_table_with_no_query_in_the_language() { + let err = run_in( + TABLE, + plans(shared_p50()).to_string().as_bytes(), + None, + QueryLanguage::sql, + ) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("no e2e query is written in sql"), "{err}"); + } + + #[test] + fn refuses_a_query_in_both_languages() { + let table = MIXED_TABLE.replace( + r#"{"promql": "sum by (job) (data)","#, + r#"{"promql": "sum by (job) (data)", "sql": "SELECT 1","#, + ); + let err = run_in( + &table, + mixed_plan(1).to_string().as_bytes(), + None, + QueryLanguage::sql, + ) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("exactly one of `promql` and `sql`"), "{err}"); + } + #[test] fn refuses_a_plan_missing_an_rqe() { let mut plan = plans(shared_p50()); From 7459ec46e814ad6fd01b9a7c67805d2205cf4c63 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 13:19:28 -0400 Subject: [PATCH 2/2] fix(planner): sub-second saved windows; review fixes - A plan's window_secs/slide_secs may be fractional (an instant query's window is one scrape, 0.1 s at 10 samples/s); they are converted to whole ms, and a fraction of a ms is refused. - A SQL query's RQEs are compared as a set. - records metrics go through --planner milp's table checks; empty values are refused. - The query language is checked once, in saved_plan_output. - Test: one SQL query's RQEs needing different sketches is refused. Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/main.rs | 7 - asap-planner-rs/src/optimizer/saved_plan.rs | 207 ++++++++++++++++---- 2 files changed, 167 insertions(+), 47 deletions(-) diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index f0056e44..32dda2b6 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -339,13 +339,6 @@ fn run_milp(args: &Args) -> anyhow::Result<()> { } fn run_saved(args: &Args) -> anyhow::Result<()> { - anyhow::ensure!( - matches!( - args.query_language, - QueryLanguage::promql | QueryLanguage::sql - ), - "--planner saved supports only --query-language promql or sql" - ); let scrape_interval_ms = check_optimizer_flags(args, "saved")?; anyhow::ensure!( args.input_config.is_none(), diff --git a/asap-planner-rs/src/optimizer/saved_plan.rs b/asap-planner-rs/src/optimizer/saved_plan.rs index fec276ca..325274ad 100644 --- a/asap-planner-rs/src/optimizer/saved_plan.rs +++ b/asap-planner-rs/src/optimizer/saved_plan.rs @@ -56,20 +56,15 @@ struct E2eQuery { } impl E2eQuery { - /// The query if it is written in `language`. - fn text(&self, language: QueryLanguage) -> anyhow::Result> { - let (promql, sql) = match (&self.promql, &self.sql) { + /// The query if it is written in SQL (`sql`) or PromQL (otherwise). + fn text(&self, sql: bool) -> anyhow::Result> { + match (&self.promql, &self.sql) { (Some(_), Some(_)) | (None, None) => bail!( "e2e query reading {:?} needs exactly one of `promql` and `sql`", self.rqes ), - (promql, sql) => (promql.as_deref(), sql.as_deref()), - }; - Ok(match language { - QueryLanguage::promql => promql, - QueryLanguage::sql => sql, - other => bail!("--planner saved supports only promql and sql, not {other:?}"), - }) + (promql, sql_text) => Ok(if sql { sql_text } else { promql }.as_deref()), + } } } @@ -107,8 +102,9 @@ struct Chosen { rqe: String, sketch: String, params: Value, - window_secs: u64, - slide_secs: u64, + /// Fractional below a second: an instant query's window is one scrape. + window_secs: f64, + slide_secs: f64, /// Index of the plan's deployment. deployment: usize, /// Present only when the deployment's grouping differs from the RQE's @@ -163,8 +159,24 @@ fn e2e_promql_config( fn e2e_sql_config( selected: &[(&E2eQuery, &str)], metrics: &BTreeMap, -) -> SQLControllerConfig { - SQLControllerConfig { +) -> anyhow::Result { + let mut tables = Vec::new(); + for (name, m) in metrics + .iter() + .filter(|(_, m)| m.kind == RECORDS_METRIC_TYPE) + { + ensure!( + !m.values.is_empty(), + "records metric {name:?} has no values" + ); + tables.push(TableDefinition { + name: name.clone(), + time_column: SQL_TIME_COLUMN.to_string(), + value_columns: m.values.keys().cloned().collect(), + metadata_columns: m.labels.iter().map(|l| l.name.clone()).collect(), + }); + } + Ok(SQLControllerConfig { query_groups: by_interval(selected) .into_iter() .map(|(secs, queries)| SQLQueryGroup { @@ -174,20 +186,11 @@ fn e2e_sql_config( controller_options: ControllerOptions::default(), }) .collect(), - tables: metrics - .iter() - .filter(|(_, m)| m.kind == RECORDS_METRIC_TYPE) - .map(|(name, m)| TableDefinition { - name: name.clone(), - time_column: SQL_TIME_COLUMN.to_string(), - value_columns: m.values.keys().cloned().collect(), - metadata_columns: m.labels.iter().map(|l| l.name.clone()).collect(), - }) - .collect(), + tables, windowing: None, sketch_parameters: None, aggregate_cleanup: None, - } + }) } fn read_json(path: &Path) -> anyhow::Result { @@ -212,6 +215,13 @@ pub fn saved_plan_output( language: QueryLanguage, scrape_interval_ms: u64, ) -> anyhow::Result { + let sql = match language { + QueryLanguage::sql => true, + QueryLanguage::promql => false, + other => { + bail!("--planner saved supports only --query-language promql or sql, not {other:?}") + } + }; let table: Table = read_json(table_path)?; let [TableWorkload { e2e: Some(e2e) }] = table.workloads.as_slice() else { bail!( @@ -255,7 +265,7 @@ pub fn saved_plan_output( } let mut selected: Vec<(&E2eQuery, &str)> = Vec::new(); for q in &e2e.queries { - if let Some(text) = q.text(language)? { + if let Some(text) = q.text(sql)? { selected.push((q, text)); } } @@ -264,17 +274,16 @@ pub fn saved_plan_output( "{}: no e2e query is written in {language:?}", table_path.display() ); - let (promql, sql); - let config = match language { - QueryLanguage::sql => { - sql = e2e_sql_config(&selected, &e2e.metrics); - MilpConfig::Sql(&sql) - } - _ => { - promql = e2e_promql_config(&selected, &e2e.metrics)?; - MilpConfig::Promql(&promql) - } + let (promql_config, sql_config); + let config = if sql { + sql_config = e2e_sql_config(&selected, &e2e.metrics)?; + MilpConfig::Sql(&sql_config) + } else { + promql_config = e2e_promql_config(&selected, &e2e.metrics)?; + MilpConfig::Promql(&promql_config) }; + // The metric and table checks `--planner milp` makes. + config.hints()?; // Before the leaf split, which turns an avg into two leaves. reject_unwritable_queries(config)?; @@ -283,10 +292,12 @@ pub fn saved_plan_output( for &(q, text) in &selected { let rqes = q.rqes.iter().map(String::as_str); let leaves: Vec<(String, Vec<&str>)> = match config { - // A SQL query is one leaf. + // A SQL query is one leaf, and its RQEs a set. MilpConfig::Sql(_) => { ensure!(!q.rqes.is_empty(), "query {text:?} names no RQEs"); - vec![(text.to_string(), rqes.collect())] + let mut rqes: Vec<&str> = rqes.collect(); + rqes.sort_unstable(); + vec![(text.to_string(), rqes)] } MilpConfig::Promql(_) => { let leaves = decompose_to_leaves(text); @@ -345,8 +356,10 @@ pub fn saved_plan_output( .clone() .unwrap_or_else(|| raqe.grouping_labels.clone()), config: cost_entry(pick)?, - window_ms: pick.window_secs * 1000, - slide_ms: pick.slide_secs * 1000, + window_ms: whole_ms(pick.window_secs) + .with_context(|| format!("RQE {rqe:?}'s window"))?, + slide_ms: whole_ms(pick.slide_secs) + .with_context(|| format!("RQE {rqe:?}'s slide"))?, key_tracker: None, }; ensure!( @@ -388,6 +401,20 @@ pub fn saved_plan_output( )?) } +/// Largest gap between `secs` in ms and the nearest whole ms that is still +/// float error, not a fraction of a ms. +const WHOLE_MS_TOLERANCE: f64 = 1e-6; + +/// `secs` in ms, refusing a negative or non-finite value or a fraction of a ms. +fn whole_ms(secs: f64) -> anyhow::Result { + let ms = secs * 1000.0; + ensure!( + ms.is_finite() && ms >= 0.0 && (ms - ms.round()).abs() < WHOLE_MS_TOLERANCE, + "{secs} s is not a whole number of ms" + ); + Ok(ms.round() as u64) +} + /// The rules of sketch-bench's `is_eligible` that need no workload facts or /// accuracy: a pick that breaks them belongs to another workload table. /// Grouping is left to the converter's roll-up check. @@ -778,6 +805,106 @@ mod tests { assert!(err.contains("served by two deployments"), "{err}"); } + /// Every RQE of a SQL query is checked, not just the first. + #[test] + fn refuses_a_sql_query_whose_rqes_need_different_sketches() { + let mut plan = mixed_plan(1); + plan["plans"][0]["chosen"][2]["params"] = json!({"lg_k": 14}); + let err = run_mixed(plan, QueryLanguage::sql) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("different sketches"), "{err}"); + } + + /// A SQL query's RQEs are a set: listing them in another order is the + /// same query. + #[test] + fn reads_a_repeated_sql_query_whose_rqes_are_reordered() { + let entry = r#"{"sql": "SELECT service, COUNT(DISTINCT user_id) FROM http WHERE ts BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY service", + "interval_secs": 60, "rqes": ["users_1pct", "users_5pct"]},"#; + let table = MIXED_TABLE.replace( + entry, + &format!( + "{entry}{}", + entry.replace( + r#"["users_1pct", "users_5pct"]"#, + r#"["users_5pct", "users_1pct"]"# + ) + ), + ); + assert_ne!(table, MIXED_TABLE); + run_in( + &table, + mixed_plan(1).to_string().as_bytes(), + None, + QueryLanguage::sql, + ) + .unwrap(); + } + + #[test] + fn refuses_a_records_metric_without_labels_or_values() { + for (from, to, reason) in [ + ( + r#""labels": [{"name": "region"}, {"name": "service"}],"#, + r#""labels": [],"#, + "metadata", + ), + ( + r#""values": {"latency": {}, "user_id": {}}"#, + r#""values": {}"#, + r#"records metric "http" has no values"#, + ), + ] { + let table = MIXED_TABLE.replace(from, to); + assert_ne!(table, MIXED_TABLE); + let err = run_in( + &table, + mixed_plan(1).to_string().as_bytes(), + None, + QueryLanguage::sql, + ) + .err() + .expect("refused") + .to_string(); + assert!(err.contains(reason), "{err}"); + } + } + + #[test] + fn refuses_a_language_other_than_promql_or_sql() { + let err = run_in( + TABLE, + plans(shared_p50()).to_string().as_bytes(), + None, + QueryLanguage::elastic_sql, + ) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("only --query-language promql or sql"), "{err}"); + } + + /// An instant query's window is one scrape, which sketch-bench writes in + /// fractional seconds (0.1 s at 10 samples/s). + #[test] + fn writes_a_sub_second_window_in_ms() { + let mut plan = plans(shared_p50()); + plan["plans"][0]["chosen"][0]["window_secs"] = json!(0.1); + plan["plans"][0]["chosen"][0]["slide_secs"] = json!(0.1); + let yaml = run(plan, None).unwrap().to_streaming_yaml_string().unwrap(); + assert!(yaml.contains("windowSizeMs: 100\n"), "{yaml}"); + + let mut plan = plans(shared_p50()); + plan["plans"][0]["chosen"][0]["window_secs"] = json!(0.0005); + let err = format!("{:#}", run(plan, None).err().expect("refused")); + assert!( + err.contains("RQE \"sum\"'s window: 0.0005 s is not a whole number of ms"), + "{err}" + ); + } + #[test] fn refuses_a_sql_pick_from_another_table() { let mut plan = mixed_plan(1);