From d23eb0979b0a467f2cb2a09652d1745c942f4b6a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 19:25:47 -0400 Subject: [PATCH 1/5] feat(sql-parser): match two-layer nested aggregation queries Add SQLPatternMatcher::nested_query_to_pattern, which accepts an outer SUM/COUNT/AVG/MIN/MAX/QUANTILE over the inner aggregate's alias, grouped by a subset of the inner GROUP BY, and returns the inner layer's pattern. query_info_to_pattern still rejects nested queries. The parser now rejects clauses on either layer it cannot represent (outer WHERE, HAVING, inner ORDER BY / LIMIT) instead of dropping them. Part of #819. Co-Authored-By: Claude Opus 5.5 --- .../src/ast_matching/sqlparser_test.rs | 142 ++++++++++++++++++ .../src/ast_matching/sqlpattern_matcher.rs | 45 ++++++ .../src/ast_matching/sqlpattern_parser.rs | 58 ++++--- 3 files changed, 228 insertions(+), 17 deletions(-) diff --git a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs index 5c332376..99ff40e8 100644 --- a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs +++ b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs @@ -1403,4 +1403,146 @@ mod tests { ) .is_none()); } + + // ── Two-layer nested aggregation ───────────────────────────────────────── + + const NESTED_INNER: &str = "SELECT L1, L2, SUM(value) AS v FROM cpu_usage \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1, L2"; + + fn nested_pattern(sql: &str) -> crate::sqlpattern_matcher::SQLQuery { + let query_data = + parse_sql_query(sql).unwrap_or_else(|| panic!("Failed to parse query: {}", sql)); + SQLPatternMatcher::new(create_test_schema(), 1.0).nested_query_to_pattern(&query_data) + } + + fn assert_invalid_nested(sql: &str) { + let result = nested_pattern(sql); + assert_eq!( + result.error, + Some(QueryError::InvalidNestedQuery), + "{sql}: {:?}", + result.msg + ); + } + + #[test] + fn test_nested_query_matches_its_inner_layer() { + let result = nested_pattern(&format!( + "SELECT L1, SUM(v) FROM ({NESTED_INNER}) GROUP BY L1" + )); + + assert!(result.is_valid(), "{:?}", result.msg); + assert_eq!(result.query_type, vec![QueryType::SpatioTemporal]); + let inner = result.outer_data().expect("inner layer"); + assert_eq!(inner.aggregation_info.get_name(), "SUM"); + assert_eq!(inner.aggregation_info.get_value_column_name(), "value"); + assert_eq!( + inner.labels, + HashSet::from(["L1".to_string(), "L2".to_string()]) + ); + } + + #[test] + fn test_nested_query_accepts_every_query_time_aggregation() { + for outer in [ + "SUM(v)", + "COUNT(v)", + "AVG(v)", + "MIN(v)", + "MAX(v)", + "QUANTILE(0.9, v)", + ] { + let sql = format!("SELECT L1, {outer} FROM ({NESTED_INNER}) GROUP BY L1"); + let result = nested_pattern(&sql); + assert!(result.is_valid(), "{sql}: {:?}", result.msg); + } + } + + #[test] + fn test_nested_query_accepts_a_global_outer_aggregation() { + let result = nested_pattern(&format!("SELECT SUM(v) FROM ({NESTED_INNER})")); + assert!(result.is_valid(), "{:?}", result.msg); + } + + #[test] + fn test_nested_query_rejects_an_outer_aggregate_not_over_the_inner_alias() { + assert_invalid_nested(&format!( + "SELECT L1, SUM(value) FROM ({NESTED_INNER}) GROUP BY L1" + )); + // Without an inner alias there is no column for the outer layer to aggregate. + assert_invalid_nested( + "SELECT L1, SUM(value) FROM (SELECT L1, SUM(value) FROM cpu_usage \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1) GROUP BY L1", + ); + } + + #[test] + fn test_nested_query_rejects_an_outer_label_missing_from_the_inner_layer() { + assert_invalid_nested(&format!( + "SELECT L3, SUM(v) FROM ({NESTED_INNER}) GROUP BY L3" + )); + } + + #[test] + fn test_nested_query_rejects_count_distinct_outer_aggregation() { + assert_invalid_nested(&format!( + "SELECT L1, COUNT(DISTINCT v) FROM ({NESTED_INNER}) GROUP BY L1" + )); + } + + #[test] + fn test_nested_query_rejects_outer_order_by_and_limit() { + assert_invalid_nested(&format!( + "SELECT L1, SUM(v) AS s FROM ({NESTED_INNER}) GROUP BY L1 ORDER BY s DESC LIMIT 3" + )); + assert_invalid_nested(&format!( + "SELECT L1, SUM(v) FROM ({NESTED_INNER}) GROUP BY L1 LIMIT 3" + )); + } + + #[test] + fn test_nested_query_rejects_a_flat_query() { + assert_invalid_nested(NESTED_INNER); + } + + #[test] + fn test_nested_query_propagates_inner_layer_errors() { + let result = nested_pattern( + "SELECT L1, SUM(v) FROM (SELECT L1, SUM(mb) AS v FROM cpu_usage \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1) GROUP BY L1", + ); + assert_eq!(result.error, Some(QueryError::InvalidValueCol)); + } + + /// Each of these clauses would otherwise be silently dropped from the + /// nested query's data, so it must fail to parse instead. + #[test] + fn test_nested_query_rejects_clauses_it_cannot_represent() { + let rejected = [ + // Outer WHERE (a filter, or a second time range). + format!("SELECT L1, SUM(v) FROM ({NESTED_INNER}) WHERE L1 = 'a' GROUP BY L1"), + format!( + "SELECT L1, SUM(v) FROM ({NESTED_INNER}) \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1" + ), + // Outer HAVING. + format!("SELECT L1, SUM(v) FROM ({NESTED_INNER}) GROUP BY L1 HAVING SUM(v) > 1"), + // Inner ORDER BY / LIMIT (per-group or plain). + format!("SELECT L1, SUM(v) FROM ({NESTED_INNER} ORDER BY v DESC LIMIT 3 BY L1) GROUP BY L1"), + format!("SELECT L1, SUM(v) FROM ({NESTED_INNER} ORDER BY v DESC LIMIT 3) GROUP BY L1"), + format!("SELECT L1, SUM(v) FROM ({NESTED_INNER} LIMIT 3) GROUP BY L1"), + // Inner HAVING. + "SELECT L1, SUM(v) FROM (SELECT L1, L2, SUM(value) AS v FROM cpu_usage \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1, L2 HAVING SUM(value) > 1) \ + GROUP BY L1" + .to_string(), + // Three layers. + format!( + "SELECT L1, SUM(m) FROM (SELECT L1, SUM(v) AS m FROM ({NESTED_INNER}) GROUP BY L1) GROUP BY L1" + ), + ]; + for sql in rejected { + assert!(parse_sql_query(&sql).is_none(), "should not parse: {sql}"); + } + } } diff --git a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs index d24a22b9..d9ca6075 100644 --- a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs +++ b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs @@ -23,9 +23,15 @@ pub enum QueryError { IllegalAggregationFn, SpatialDurationSmall, NestedQueryUnsupported, + /// A two-layer query whose outer layer is not a query-time aggregation + /// over its inner layer's output. + InvalidNestedQuery, MissingTimeRange, } +/// Aggregations an outer layer can apply to its inner layer's rows. +const QUERY_TIME_AGGREGATIONS: [&str; 6] = ["SUM", "COUNT", "AVG", "MIN", "MAX", "QUANTILE"]; + #[derive(Debug)] pub struct SQLQuery { pub query_type: Vec, @@ -294,4 +300,43 @@ impl SQLPatternMatcher { sql_query } + + /// Matches a two-layer query whose outer layer aggregates its inner + /// layer's rows, e.g. `SELECT job, SUM(v) FROM () GROUP BY job`. + /// Returns the inner layer's pattern; the outer layer runs at query time. + pub fn nested_query_to_pattern(&self, query: &SQLQueryData) -> SQLQuery { + let invalid = |msg: String| { + SQLQuery::new(Vec::new(), Some(QueryError::InvalidNestedQuery), Some(msg)) + }; + let Some(inner) = query.subquery.as_deref() else { + return invalid("query has no inner layer".to_string()); + }; + + if !query.order_by.is_empty() || query.limit.is_some() { + return invalid("ORDER BY / LIMIT on the outer layer is not supported".to_string()); + } + + let outer_aggregation = query.aggregation_info.get_name(); + if !QUERY_TIME_AGGREGATIONS.contains(&outer_aggregation) { + return invalid(format!( + "outer aggregation {outer_aggregation} cannot run over the inner layer" + )); + } + + let outer_column = query.aggregation_info.get_value_column_name(); + if inner.aggregation_alias.as_deref() != Some(outer_column) { + return invalid(format!( + "outer aggregation reads {outer_column}, which is not the inner aggregate's alias" + )); + } + + if !query.labels.is_subset(&inner.labels) { + let missing: HashSet<_> = query.labels.difference(&inner.labels).collect(); + return invalid(format!( + "outer layer groups by {missing:?}, which the inner layer does not output" + )); + } + + self.query_info_to_pattern(inner) + } } diff --git a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs index 1b22f105..a6e572ca 100644 --- a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs +++ b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs @@ -193,6 +193,21 @@ impl SQLPatternParser { query } + /// SELECT clauses that SQLQueryData cannot represent. + fn has_unsupported_clauses(select: &Select) -> bool { + select.distinct.is_some() + || select.top.is_some() + || select.into.is_some() + || !select.lateral_views.is_empty() + || select.prewhere.is_some() + || !select.cluster_by.is_empty() + || !select.distribute_by.is_empty() + || !select.sort_by.is_empty() + || select.having.is_some() + || !select.named_window.is_empty() + || select.window_before_qualify + } + fn parse_select(&self, select: &Select) -> Option { let (metric, has_subquery) = self.get_metric(select)?; @@ -204,26 +219,14 @@ impl SQLPatternParser { return None; } + if Self::has_unsupported_clauses(select) { + println!("Unexpected SELECT fields present"); + return None; + } + if !has_subquery { let time_info = self.get_time_info(select, &metric)?; - // Check for unexpected fields - if select.distinct.is_some() - || select.top.is_some() - || select.into.is_some() - || !select.lateral_views.is_empty() - || select.prewhere.is_some() - || !select.cluster_by.is_empty() - || !select.distribute_by.is_empty() - || !select.sort_by.is_empty() - || select.having.is_some() - || !select.named_window.is_empty() - || select.window_before_qualify - { - println!("Unexpected SELECT fields present"); - return None; - } - Some(SQLQueryData { aggregation_info: aggregation, aggregation_alias, @@ -235,10 +238,31 @@ impl SQLPatternParser { limit: None, }) } else { + // The outer layer filters nothing: its rows are the inner layer's + // output, and the time range belongs to the inner layer. + if select.selection.is_some() { + println!("WHERE on the outer layer of a nested query is not supported"); + return None; + } + // Parse subquery let subquery = match &select.from[0].relation { TableFactor::Derived { subquery, .. } => match subquery.body.as_ref() { SetExpr::Select(inner_select) => { + // Query-level clauses of the inner layer have no place + // in its SQLQueryData, so they would be dropped. + if subquery.with.is_some() + || subquery.order_by.is_some() + || subquery.limit_clause.is_some() + || subquery.fetch.is_some() + { + println!("ORDER BY / LIMIT inside a subquery is not supported"); + return None; + } + if Self::has_unsupported_clauses(inner_select) { + println!("Unexpected SELECT fields present in subquery"); + return None; + } let (inner_aggregation, inner_alias) = self.get_aggregation(inner_select)?; let inner_group_bys = self.get_groupbys(inner_select)?; From d8a5dc64f27e69eb83d8d3ef8bf6105095732f16 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 19:30:36 -0400 Subject: [PATCH 2/5] feat(planner): plan two-layer SQL queries as inner query plus query-time aggregation A nested SQL query is planned through its inner layer, like any single-level query, and its outer layer is emitted as a query-time aggregation stage, mirroring PromQL's nested_aggregation_plan. Part of #819. Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/planner/sql.rs | 157 +++++++++++++++++++++-- asap-planner-rs/src/sql/generator.rs | 20 ++- asap-planner-rs/tests/sql_integration.rs | 105 +++++++++++++++ 3 files changed, 268 insertions(+), 14 deletions(-) diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs index bbf2e67d..b4be38fe 100644 --- a/asap-planner-rs/src/planner/sql.rs +++ b/asap-planner-rs/src/planner/sql.rs @@ -1,6 +1,10 @@ use std::collections::HashSet; use asap_types::enums::{CleanupPolicy, WindowType}; +use asap_types::query_config::{ + QueryTimeAggregation, QueryTimeAggregationOperator, QueryTimeAggregationParameter, + QueryTimeGrouping, QueryTimeGroupingMode, +}; use promql_utilities::data_model::KeyByLabelNames; use promql_utilities::query_logics::enums::{AggregationType, QueryTreatmentType, Statistic}; use sql_utilities::ast_matching::sqlhelper::{ @@ -9,6 +13,7 @@ use sql_utilities::ast_matching::sqlhelper::{ use sql_utilities::ast_matching::sqlpattern_matcher::SQLPatternMatcher; use sql_utilities::ast_matching::sqlpattern_parser::SQLPatternParser; use sql_utilities::ast_matching::SQLSchema; +use sqlparser::ast::{SetExpr, Statement, TableFactor}; use sqlparser::dialect::ClickHouseDialect; use sqlparser::parser::Parser as SqlParser; @@ -16,6 +21,7 @@ use crate::config::input::{SketchParameterOverrides, TableDefinition, WindowingC use crate::error::ControllerError; use crate::planner::agg_config::{build_agg_configs_for_statistics, IntermediateAggConfig}; use crate::planner::cleanup::get_sql_cleanup_param; +use crate::planner::promql::NestedAggregationPlan; use crate::planner::sketch::build_sketch_parameters; use crate::planner::window::IntermediateWindowConfig; use crate::StreamingEngine; @@ -157,6 +163,145 @@ impl SQLSingleQueryProcessor { Ok((configs, cleanup_param)) } + + /// For a two-layer query, its inner layer (planned like any single-level + /// query) and the outer layer as a query-time aggregation over it. `None` + /// for a single-level query. + pub fn nested_aggregation_plan( + &self, + query_evaluation_time: f64, + ) -> Result, ControllerError> { + let schema = build_sql_schema(&self.table_definitions); + let (stmts, qdata) = + parse_sql_statements(&self.query_string, &schema, query_evaluation_time)?; + if qdata.subquery.is_none() { + return Ok(None); + } + + let inner_pattern = sql_pattern_matcher(schema, self.data_ingestion_interval_ms) + .nested_query_to_pattern(&qdata); + if !inner_pattern.is_valid() { + return Err(ControllerError::SqlParse( + inner_pattern.msg.unwrap_or_default(), + )); + } + + let planned_subquery = inner_query_sql(&stmts).ok_or_else(|| { + ControllerError::SqlParse(format!( + "nested query has no subquery in FROM: {}", + self.query_string + )) + })?; + Ok(Some(NestedAggregationPlan { + planned_subquery, + query_time_aggregations: vec![sql_query_time_aggregation(&qdata)?], + })) + } + + /// A processor for `query` with this one's settings. + pub fn make_subquery_processor(&self, query: String) -> Self { + Self::new( + query, + self.t_repeat_ms, + self.data_ingestion_interval_ms, + self.table_definitions.clone(), + self.streaming_engine, + self.sketch_parameters.clone(), + self.cleanup_policy, + self.windowing.clone(), + ) + } +} + +/// Parses `query` into its statements and pattern-parser data. +fn parse_sql_statements( + query: &str, + schema: &SQLSchema, + query_evaluation_time: f64, +) -> Result<(Vec, SQLQueryData), ControllerError> { + let stmts = SqlParser::parse_sql(&ClickHouseDialect {}, query) + .map_err(|e| ControllerError::SqlParse(e.to_string()))?; + let qdata = SQLPatternParser::new(schema, query_evaluation_time) + .parse_query(&stmts) + .ok_or_else(|| ControllerError::SqlParse(format!("Failed to parse SQL query: {query}")))?; + Ok((stmts, qdata)) +} + +/// SQLPatternMatcher.scrape_interval is in seconds (SQL timestamps are seconds-based). +fn sql_pattern_matcher(schema: SQLSchema, data_ingestion_interval_ms: u64) -> SQLPatternMatcher { + SQLPatternMatcher::new(schema, data_ingestion_interval_ms as f64 / 1000.0) +} + +/// The SQL text of the subquery in the outer layer's FROM. +fn inner_query_sql(stmts: &[Statement]) -> Option { + let [Statement::Query(query)] = stmts else { + return None; + }; + let SetExpr::Select(select) = query.body.as_ref() else { + return None; + }; + match &select.from.first()?.relation { + TableFactor::Derived { subquery, .. } => Some(subquery.to_string()), + _ => None, + } +} + +/// The query-time aggregation an outer SQL layer applies to its inner +/// layer's rows: its aggregate, grouped by its GROUP BY (all rows without one). +fn sql_query_time_aggregation( + outer: &SQLQueryData, +) -> Result { + let info = &outer.aggregation_info; + let operator = match info.get_name() { + "SUM" => QueryTimeAggregationOperator::Sum, + "COUNT" => QueryTimeAggregationOperator::Count, + "AVG" => QueryTimeAggregationOperator::Avg, + "MIN" => QueryTimeAggregationOperator::Min, + "MAX" => QueryTimeAggregationOperator::Max, + "QUANTILE" => QueryTimeAggregationOperator::Quantile, + other => { + return Err(ControllerError::SqlParse(format!( + "unsupported outer aggregation: {other}" + ))) + } + }; + let parameter = match operator { + QueryTimeAggregationOperator::Quantile => { + let phi = info + .get_args() + .first() + .and_then(|arg| arg.parse::().ok()) + .ok_or_else(|| { + ControllerError::SqlParse(format!( + "QUANTILE needs a numeric level, got {:?}", + info.get_args() + )) + })?; + Some(QueryTimeAggregationParameter::Float(phi)) + } + _ => None, + }; + let grouping = if outer.labels.is_empty() { + QueryTimeGrouping { + mode: QueryTimeGroupingMode::All, + labels: Vec::new(), + } + } else { + let mut labels: Vec = outer.labels.iter().cloned().collect(); + labels.sort(); + QueryTimeGrouping { + mode: QueryTimeGroupingMode::By, + labels, + } + }; + + let aggregation = QueryTimeAggregation { + operator, + grouping, + parameter, + }; + aggregation.validate().map_err(ControllerError::SqlParse)?; + Ok(aggregation) } /// Parses `query` and matches it against `tables`: a single-level query's @@ -168,15 +313,9 @@ pub(crate) fn parse_sql_query( query_evaluation_time: f64, ) -> Result<(SQLQueryData, Option), ControllerError> { let schema = build_sql_schema(tables); - let stmts = SqlParser::parse_sql(&ClickHouseDialect {}, query) - .map_err(|e| ControllerError::SqlParse(e.to_string()))?; - let qdata = SQLPatternParser::new(&schema, query_evaluation_time) - .parse_query(&stmts) - .ok_or_else(|| ControllerError::SqlParse(format!("Failed to parse SQL query: {query}")))?; - - // SQLPatternMatcher.scrape_interval is in seconds (SQL timestamps are seconds-based). - let sql_query = SQLPatternMatcher::new(schema, data_ingestion_interval_ms as f64 / 1000.0) - .query_info_to_pattern(&qdata); + let (_, qdata) = parse_sql_statements(query, &schema, query_evaluation_time)?; + let sql_query = + sql_pattern_matcher(schema, data_ingestion_interval_ms).query_info_to_pattern(&qdata); if !sql_query.is_valid() { return Err(ControllerError::SqlParse(sql_query.msg.unwrap_or_default())); } diff --git a/asap-planner-rs/src/sql/generator.rs b/asap-planner-rs/src/sql/generator.rs index aa6df271..80b95571 100644 --- a/asap-planner-rs/src/sql/generator.rs +++ b/asap-planner-rs/src/sql/generator.rs @@ -96,8 +96,13 @@ pub fn generate_sql_plan( config.windowing.clone(), ); + let nested_plan = processor.nested_aggregation_plan(eval_time)?; + let planned_processor = match &nested_plan { + Some(plan) => processor.make_subquery_processor(plan.planned_subquery.clone()), + None => processor, + }; let (configs, cleanup_param) = - match processor.get_streaming_aggregation_configs(eval_time) { + match planned_processor.get_streaming_aggregation_configs(eval_time) { Ok(result) => result, Err(ControllerError::Windowing(error)) => { windowing_errors.push(format!("query '{query_string}': {error}")); @@ -112,10 +117,15 @@ pub fn generate_sql_plan( keys_for_query.push((key.clone(), cleanup_param)); dedup_map.entry(key).or_insert(config_item); } - query_plan_map.insert( - query_string.clone(), - QueryPlanEntry::fully_planned(query_string.clone(), keys_for_query), - ); + let plan_entry = match nested_plan { + Some(plan) => QueryPlanEntry { + aggregation_keys: keys_for_query, + planned_subquery: plan.planned_subquery, + query_time_aggregations: plan.query_time_aggregations, + }, + None => QueryPlanEntry::fully_planned(query_string.clone(), keys_for_query), + }; + query_plan_map.insert(query_string.clone(), plan_entry); } } diff --git a/asap-planner-rs/tests/sql_integration.rs b/asap-planner-rs/tests/sql_integration.rs index ee77222b..c28636ec 100644 --- a/asap-planner-rs/tests/sql_integration.rs +++ b/asap-planner-rs/tests/sql_integration.rs @@ -1215,3 +1215,108 @@ fn spatial_count_limit_zero_is_not_topk() { assert!(out.has_aggregation_type("DeltaSetAggregator")); assert!(!out.has_aggregation_type("CountMinSketchWithHeap")); } + +// ── two-layer nested aggregation ────────────────────────────────────────────── + +const NESTED_INNER: &str = "SELECT datacenter, hostname, SUM(cpu_usage) AS v FROM metrics_table WHERE time BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY datacenter, hostname"; + +/// The single inference query entry the planner wrote for `query`. +fn planned_query_entry(query: &str) -> serde_yaml::Value { + let out = SQLController::from_yaml(&one_query_config(query, 300_000), sql_opts()) + .unwrap() + .generate() + .unwrap(); + let inference: serde_yaml::Value = + serde_yaml::from_str(&out.to_inference_yaml_string().unwrap()).unwrap(); + let queries = inference["queries"].as_sequence().unwrap(); + assert_eq!(queries.len(), 1); + queries[0].clone() +} + +#[test] +fn nested_query_plans_its_inner_layer_and_emits_the_outer_as_a_query_time_stage() { + let q = format!("SELECT datacenter, SUM(v) FROM ({NESTED_INNER}) GROUP BY datacenter"); + let entry = planned_query_entry(&q); + + assert_eq!(entry["query"].as_str(), Some(q.as_str())); + assert_eq!(entry["planned_subquery"].as_str(), Some(NESTED_INNER)); + let stages = entry["query_time_aggregations"].as_sequence().unwrap(); + assert_eq!(stages.len(), 1); + assert_eq!(stages[0]["operator"].as_str(), Some("sum")); + assert_eq!(stages[0]["grouping"]["mode"].as_str(), Some("by")); + assert_eq!( + stages[0]["grouping"]["labels"].as_sequence().unwrap(), + &vec![serde_yaml::Value::from("datacenter")] + ); + assert!(stages[0]["parameter"].is_null()); +} + +/// The streaming aggregations of a nested query are exactly those of its inner +/// layer planned on its own. +#[test] +fn nested_query_streaming_plan_equals_its_inner_layers() { + let nested = format!("SELECT datacenter, MAX(v) FROM ({NESTED_INNER}) GROUP BY datacenter"); + let plan = |q: &str| { + SQLController::from_yaml(&one_query_config(q, 300_000), sql_opts()) + .unwrap() + .generate() + .unwrap() + .to_streaming_yaml_string() + .unwrap() + }; + assert_eq!(plan(&nested), plan(NESTED_INNER)); +} + +#[test] +fn nested_query_maps_each_outer_aggregation_to_its_operator() { + let cases = [ + ("SUM(v)", "sum"), + ("COUNT(v)", "count"), + ("AVG(v)", "avg"), + ("MIN(v)", "min"), + ("MAX(v)", "max"), + ("QUANTILE(0.9, v)", "quantile"), + ]; + for (outer, operator) in cases { + let q = format!("SELECT datacenter, {outer} FROM ({NESTED_INNER}) GROUP BY datacenter"); + let entry = planned_query_entry(&q); + let stage = &entry["query_time_aggregations"][0]; + assert_eq!(stage["operator"].as_str(), Some(operator), "{q}"); + if operator == "quantile" { + assert_eq!(stage["parameter"].as_f64(), Some(0.9), "{q}"); + } else { + assert!(stage["parameter"].is_null(), "{q}"); + } + } +} + +#[test] +fn nested_query_without_outer_group_by_aggregates_all_series() { + let entry = planned_query_entry(&format!("SELECT SUM(v) FROM ({NESTED_INNER})")); + let stage = &entry["query_time_aggregations"][0]; + assert_eq!(stage["grouping"]["mode"].as_str(), Some("all")); + assert!(stage["grouping"]["labels"] + .as_sequence() + .unwrap() + .is_empty()); +} + +#[test] +fn flat_query_has_no_query_time_stages() { + let entry = planned_query_entry(NESTED_INNER); + assert_eq!(entry["planned_subquery"].as_str(), Some(NESTED_INNER)); + assert!(entry["query_time_aggregations"] + .as_sequence() + .unwrap() + .is_empty()); +} + +#[test] +fn invalid_nested_query_returns_sql_parse_error() { + // The outer layer groups by a column the inner layer does not output. + let q = format!("SELECT region, SUM(v) FROM ({NESTED_INNER}) GROUP BY region"); + let result = SQLController::from_yaml(&one_query_config(&q, 300_000), sql_opts()) + .unwrap() + .generate(); + assert!(matches!(result, Err(ControllerError::SqlParse(_)))); +} From b9719c4a36202a6493e410b25751357a0ec0a6b4 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 19:34:16 -0400 Subject: [PATCH 3/5] feat(query-engine): serve two-layer SQL queries via query-time aggregation A nested SQL query is executed through its inner layer, using the nested query's config, and the config's query-time aggregations are applied to the result as its outer layer. A nested query whose config has no query-time aggregations is not served, since running the inner layer alone would drop the outer aggregation. The 8 non-topk one-stage nested-aggregations cases and the two nested cases in basic.yaml are now supported. Closes #819. Co-Authored-By: Claude Opus 5.5 --- .../src/engines/simple_engine/sql.rs | 100 +++++++-- asap-query-engine/src/tests/mod.rs | 1 + .../src/tests/query_equivalence_tests.rs | 6 +- .../src/tests/sql_nested_aggregation_tests.rs | 199 ++++++++++++++++++ .../src/tests/sql_pattern_matching_tests.rs | 9 +- .../tests/sql_query_support/basic.yaml | 4 +- .../nested-aggregations.yaml | 16 +- 7 files changed, 304 insertions(+), 31 deletions(-) create mode 100644 asap-query-engine/src/tests/sql_nested_aggregation_tests.rs diff --git a/asap-query-engine/src/engines/simple_engine/sql.rs b/asap-query-engine/src/engines/simple_engine/sql.rs index 11b71320..55616788 100644 --- a/asap-query-engine/src/engines/simple_engine/sql.rs +++ b/asap-query-engine/src/engines/simple_engine/sql.rs @@ -6,6 +6,8 @@ use super::SimpleEngine; use super::{QueryExecutionContext, QueryMetadata, QueryTimestamps}; use crate::data_model::{AggregationIdInfo, QueryConfig, SchemaConfig}; use crate::engines::query_result::{InstantVector, InstantVectorElement, QueryResult}; +use crate::engines::query_time_aggregation::apply_instant_pipeline; +use asap_types::query_config::QueryTimeAggregation; use asap_types::query_requirements::QueryRequirements; use asap_types::utils::normalize_spatial_filter; use promql_utilities::data_model::KeyByLabelNames; @@ -17,7 +19,7 @@ use sql_utilities::sqlhelper::{OrderByItem, SQLQueryData}; use sqlparser::dialect::*; use sqlparser::parser::Parser as parser; use std::collections::HashMap; -use tracing::{debug, warn}; +use tracing::{debug, error, warn}; /// SQL-only post-processing produced alongside a `QueryExecutionContext`: /// rules for ordering and truncating the final result vector. @@ -279,7 +281,7 @@ impl SimpleEngine { query: String, time: f64, ) -> Option<(KeyByLabelNames, QueryResult)> { - let (context, post) = + let (context, post, query_time_aggregations) = self.build_query_execution_context_sql_with_post_processing(query, time)?; let is_topk = context.metadata.statistic_to_compute == Statistic::Topk; // Top-k: enable heap-based limiting (truncate to k) but NOT PromQL-style @@ -287,6 +289,21 @@ impl SimpleEngine { // `(group-by key, count)` rows, so SQL ORDER BY / LIMIT post-processing // would be redundant and is skipped. let (output_labels, result) = self.execute_context(context, is_topk, false)?; + let (output_labels, result) = if query_time_aggregations.is_empty() { + (output_labels, result) + } else { + let QueryResult::Vector(InstantVector { values, timestamp }) = result else { + error!("SQL query-time aggregations need an instant vector"); + return None; + }; + match apply_instant_pipeline(output_labels, values, &query_time_aggregations) { + Ok((labels, values)) => (labels, QueryResult::vector(values, timestamp)), + Err(e) => { + error!("SQL query-time aggregation failed: {e}"); + return None; + } + } + }; let result = if is_topk { result } else { @@ -305,17 +322,24 @@ impl SimpleEngine { time: f64, ) -> Option { self.build_query_execution_context_sql_with_post_processing(query, time) - .map(|(ctx, _)| ctx) + .map(|(ctx, _, _)| ctx) } - /// Internal: parses + plans a SQL query and returns both the execution - /// context (shared with PromQL/Elastic engines) and the SQL-only - /// post-processing rules (ORDER BY / LIMIT / alias resolution). + /// Internal: parses + plans a SQL query and returns the execution + /// context (shared with PromQL/Elastic engines), the SQL-only + /// post-processing rules (ORDER BY / LIMIT / alias resolution), and the + /// query-time aggregations to apply to the context's result. For a + /// nested query the context serves its inner layer and the aggregations + /// are its outer layer. fn build_query_execution_context_sql_with_post_processing( &self, query: String, time: f64, - ) -> Option<(QueryExecutionContext, SqlPostProcessing)> { + ) -> Option<( + QueryExecutionContext, + SqlPostProcessing, + Vec, + )> { // Get SQL schema from inference config let schema = match &self.inference_config.read().unwrap().schema { SchemaConfig::SQL(sql_schema) => sql_schema.clone(), @@ -342,6 +366,9 @@ impl SimpleEngine { // seconds-denominated SQL query duration by this value — convert back to seconds. let matcher = SQLPatternMatcher::new(schema, self.data_ingestion_interval_ms as f64 / 1000.0); + if query_data.subquery.is_some() { + return self.build_nested_query_execution_context_sql(&matcher, &query_data); + } let match_result = matcher.query_info_to_pattern(&query_data); debug!("Match result: {:?}", match_result); @@ -365,8 +392,52 @@ impl SimpleEngine { let query_time = Self::convert_query_time_to_data_time( query_data.time_info.get_start() + query_data.time_info.get_duration(), ); - let ctx = self.build_spatiotemporal_context(&match_result, query_time, &query_data)?; - Some((ctx, post)) + let config = self.find_query_config_sql(&query_data); + let ctx = + self.build_spatiotemporal_context(&match_result, query_time, &query_data, config)?; + Some((ctx, post, Vec::new())) + } + + /// A nested query is served from its inner layer, with its outer layer + /// applied as the planned query-time aggregations. Only a config carries + /// those, so a nested query without one is not served: running its inner + /// layer alone would drop the outer aggregation. + fn build_nested_query_execution_context_sql( + &self, + matcher: &SQLPatternMatcher, + query_data: &SQLQueryData, + ) -> Option<( + QueryExecutionContext, + SqlPostProcessing, + Vec, + )> { + let inner_match = matcher.nested_query_to_pattern(query_data); + if !inner_match.is_valid() { + debug!("Nested query not matched: {:?}", inner_match.msg); + return None; + } + let Some(config) = self.find_query_config_sql(query_data) else { + warn!("No query_config entry for nested SQL query; not serving it"); + return None; + }; + if config.query_time_aggregations.is_empty() { + warn!( + "query_config for nested SQL query has no query-time aggregations; not serving it" + ); + return None; + } + let inner = query_data.subquery.as_deref()?; + let query_time = Self::convert_query_time_to_data_time( + inner.time_info.get_start() + inner.time_info.get_duration(), + ); + let query_time_aggregations = config.query_time_aggregations.clone(); + let ctx = + self.build_spatiotemporal_context(&inner_match, query_time, inner, Some(config))?; + Some(( + ctx, + SqlPostProcessing::from_query_data(query_data), + query_time_aggregations, + )) } /// Shared context-building tail for both SQL context builders. @@ -420,11 +491,14 @@ impl SimpleEngine { /// Build execution context for SpatioTemporal queries. /// These queries span multiple scrape intervals but GROUP BY a subset of labels. + /// `config` is the query's planned config; without one the aggregation + /// is found by capability matching. fn build_spatiotemporal_context( &self, match_result: &SQLQuery, query_time: u64, query_data: &SQLQueryData, + config: Option, ) -> Option { // Output labels are the GROUP BY columns (subset of all labels) let query_output_labels = KeyByLabelNames::new( @@ -489,9 +563,7 @@ impl SimpleEngine { }; // Resolve aggregation: try pre-configured query_configs first, fall back to capability matching. - let agg_info: AggregationIdInfo = if let Some(config) = - self.find_query_config_sql(query_data) - { + let agg_info: AggregationIdInfo = if let Some(config) = config { self.get_aggregation_id_info(&config) .map_err(|e| { warn!("{}", e); @@ -711,8 +783,8 @@ mod detect_topk_tests { let qd = parse(&sql).expect("nested query should parse"); assert!( detect_sql_topk(&qd).is_some(), - "outer SELECT alone matches the top-k shape — this is why nested queries must be \ - rejected (NestedQueryUnsupported) before topk detection ever runs on them", + "outer SELECT alone matches the top-k shape — this is why a nested query's outer \ + ORDER BY / LIMIT is rejected and top-k detection only ever sees its inner layer", ); } } diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs index 47cab84a..9bb4a15d 100644 --- a/asap-query-engine/src/tests/mod.rs +++ b/asap-query-engine/src/tests/mod.rs @@ -13,6 +13,7 @@ pub mod query_equivalence_tests; pub mod query_time_aggregation_public_tests; pub mod range_query_arithmetic_tests; pub mod rollup_tests; +pub mod sql_nested_aggregation_tests; pub mod sql_pattern_matching_tests; pub mod stage_e_instant_range_equivalence_tests; pub mod store_correctness_tests; diff --git a/asap-query-engine/src/tests/query_equivalence_tests.rs b/asap-query-engine/src/tests/query_equivalence_tests.rs index 60f115c3..138f1117 100644 --- a/asap-query-engine/src/tests/query_equivalence_tests.rs +++ b/asap-query-engine/src/tests/query_equivalence_tests.rs @@ -482,9 +482,9 @@ mod tests { #[test] fn test_spatial_of_temporal_sql_is_rejected() { // PromQL still supports (and collapses) sum(sum_over_time(...)) by (label); - // the SQL equivalent is a nested subquery, which the matcher now rejects - // outright with QueryError::NestedQueryUnsupported (#499) rather than - // mis-executing it, so there's no longer a SQL context to compare. + // the SQL equivalent is a nested subquery. Its config here has no + // query-time aggregations for the outer layer, so the engine rejects it + // rather than mis-executing it as the inner layer alone. let scrape_interval_ms = 1000; let promql_query = "sum(sum_over_time(cpu_usage[10s])) by (L1)"; let sql_query = "SELECT SUM(result) FROM (SELECT SUM(value) AS result FROM cpu_usage WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY L1, L2, L3, L4) GROUP BY L1"; diff --git a/asap-query-engine/src/tests/sql_nested_aggregation_tests.rs b/asap-query-engine/src/tests/sql_nested_aggregation_tests.rs new file mode 100644 index 00000000..f2e04861 --- /dev/null +++ b/asap-query-engine/src/tests/sql_nested_aggregation_tests.rs @@ -0,0 +1,199 @@ +//! Two-layer SQL queries: the inner layer is served from a precomputed +//! aggregation and the outer layer runs as a query-time aggregation over it. + +#[cfg(test)] +mod tests { + use crate::data_model::{ + AggregationConfig, AggregationReference, AggregationType, CleanupPolicy, InferenceConfig, + KeyByLabelValues, PrecomputedOutput, QueryConfig, QueryLanguage, SchemaConfig, + StreamingConfig, WindowType, + }; + use crate::engines::query_result::QueryResult; + use crate::engines::simple_engine::SimpleEngine; + use crate::precompute_operators::sum_accumulator::SumAccumulator; + use crate::stores::simple_map_store::SimpleMapStore; + use crate::stores::Store; + use crate::tests::test_utilities::config_builders::create_test_schema; + use asap_types::query_config::{ + QueryTimeAggregation, QueryTimeAggregationOperator, QueryTimeAggregationParameter, + QueryTimeGrouping, QueryTimeGroupingMode, + }; + use promql_utilities::data_model::KeyByLabelNames; + use std::collections::HashMap; + use std::sync::Arc; + + const QUERY_TIME_S: f64 = 1_000.0; + const WINDOW_MS: u64 = 1_000; + const INNER: &str = "SELECT job, instance, SUM(value) AS v FROM data \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance"; + + fn by_job() -> QueryTimeGrouping { + QueryTimeGrouping { + mode: QueryTimeGroupingMode::By, + labels: vec!["job".to_string()], + } + } + + fn stage( + operator: QueryTimeAggregationOperator, + grouping: QueryTimeGrouping, + parameter: Option, + ) -> QueryTimeAggregation { + QueryTimeAggregation { + operator, + grouping, + parameter, + } + } + + /// An engine whose only aggregation is a per-(instance, job) sum holding + /// (i1, a) = 1, (i2, a) = 2, (i1, b) = 4, planned for `config`. + fn engine(config: QueryConfig) -> SimpleEngine { + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs: HashMap::from([( + 1, + AggregationConfig { + aggregation_id: 1, + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(vec![ + "instance".to_string(), + "job".to_string(), + ]), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: WINDOW_MS, + slide_interval_ms: WINDOW_MS, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: "data".to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: Some("data".to_string()), + value_column: Some("value".to_string()), + }, + )]), + }); + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + let query_time_ms = (QUERY_TIME_S * 1_000.0) as u64; + for (instance, job, value) in [("i1", "a", 1.0), ("i2", "a", 2.0), ("i1", "b", 4.0)] { + store + .insert_precomputed_output( + PrecomputedOutput::new( + query_time_ms - WINDOW_MS, + query_time_ms, + Some(KeyByLabelValues { + labels: vec![instance.to_string(), job.to_string()], + }), + 1, + ), + Box::new(SumAccumulator::with_sum(value)), + ) + .unwrap(); + } + let inference_config = InferenceConfig { + schema: SchemaConfig::SQL(create_test_schema( + "data", + "time", + "value", + vec!["job", "instance"], + )), + query_configs: vec![config.add_aggregation(AggregationReference::new(1, None))], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + SimpleEngine::new( + store, + inference_config, + streaming_config, + WINDOW_MS, + QueryLanguage::sql, + ) + } + + /// `query` planned with `stages` over `INNER`, executed; rows as + /// (labels, value) sorted by labels. + fn rows(query: &str, stages: Vec) -> Vec<(Vec, f64)> { + let engine = engine(QueryConfig::with_plan( + query.to_string(), + INNER.to_string(), + stages, + )); + let (_, result) = engine + .handle_query_sql(query.to_string(), QUERY_TIME_S) + .expect("nested query should be served locally"); + let QueryResult::Vector(vector) = result else { + panic!("expected an instant vector"); + }; + let mut rows: Vec<_> = vector + .values + .into_iter() + .map(|element| (element.labels.labels, element.value)) + .collect(); + rows.sort_by(|left, right| left.0.cmp(&right.0)); + rows + } + + fn job_rows(a: f64, b: f64) -> Vec<(Vec, f64)> { + vec![(vec!["a".to_string()], a), (vec!["b".to_string()], b)] + } + + #[test] + fn outer_sum_aggregates_the_inner_rows_by_job() { + let query = format!("SELECT job, SUM(v) FROM ({INNER}) GROUP BY job"); + assert_eq!( + rows( + &query, + vec![stage(QueryTimeAggregationOperator::Sum, by_job(), None)] + ), + job_rows(3.0, 4.0) + ); + } + + #[test] + fn outer_quantile_aggregates_the_inner_rows_by_job() { + let query = format!("SELECT job, QUANTILE(0.5, v) FROM ({INNER}) GROUP BY job"); + assert_eq!( + rows( + &query, + vec![stage( + QueryTimeAggregationOperator::Quantile, + by_job(), + Some(QueryTimeAggregationParameter::Float(0.5)), + )] + ), + job_rows(1.5, 4.0) + ); + } + + #[test] + fn outer_aggregation_without_group_by_aggregates_every_inner_row() { + let query = format!("SELECT SUM(v) FROM ({INNER})"); + let all = QueryTimeGrouping { + mode: QueryTimeGroupingMode::All, + labels: Vec::new(), + }; + assert_eq!( + rows( + &query, + vec![stage(QueryTimeAggregationOperator::Sum, all, None)] + ), + vec![(Vec::new(), 7.0)] + ); + } + + /// A nested query whose config carries no query-time stages must not be + /// served from the inner layer alone: that would silently drop the outer + /// aggregation. + #[test] + fn nested_query_planned_without_stages_is_not_served() { + let query = format!("SELECT job, SUM(v) FROM ({INNER}) GROUP BY job"); + let engine = engine(QueryConfig::new(query.clone())); + assert!(engine.handle_query_sql(query, QUERY_TIME_S).is_none()); + } +} diff --git a/asap-query-engine/src/tests/sql_pattern_matching_tests.rs b/asap-query-engine/src/tests/sql_pattern_matching_tests.rs index 8204cefc..8f8d135c 100644 --- a/asap-query-engine/src/tests/sql_pattern_matching_tests.rs +++ b/asap-query-engine/src/tests/sql_pattern_matching_tests.rs @@ -267,8 +267,9 @@ mod tests { #[test] fn test_spatial_of_temporal_subquery_is_rejected() { - // Nested SQL (spatial-of-temporal subquery) is unsupported: the matcher rejects - // it with QueryError::NestedQueryUnsupported instead of mis-executing it (#499). + // Nested SQL (spatial-of-temporal subquery) planned without query-time + // aggregations for its outer layer: the engine rejects it instead of + // mis-executing it as the inner layer alone. let template = "SELECT SUM(result) FROM (SELECT SUM(value) AS result FROM cpu_usage WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY L1, L2, L3, L4) GROUP BY L1"; let (engine, _) = build_sql_engine(template, 1, 10_000, AggregationType::Sum, &["L1"], &[]); @@ -284,8 +285,8 @@ mod tests { #[test] fn nested_order_by_limit_query_is_rejected() { - // Nested SQL with an outer ORDER BY + LIMIT is unsupported for the same reason: - // the matcher rejects it before topk detection ever runs. + // Nested SQL with an outer ORDER BY + LIMIT is unsupported: the matcher + // rejects it before topk detection ever runs. let template = "SELECT L1, SUM(result) FROM (SELECT SUM(value) AS result FROM cpu_usage WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY L1, L2, L3, L4) sub GROUP BY L1"; let (engine, _) = build_sql_engine(template, 1, 10_000, AggregationType::Sum, &["L1"], &[]); diff --git a/asap-query-engine/tests/sql_query_support/basic.yaml b/asap-query-engine/tests/sql_query_support/basic.yaml index e88238c0..54377498 100644 --- a/asap-query-engine/tests/sql_query_support/basic.yaml +++ b/asap-query-engine/tests/sql_query_support/basic.yaml @@ -108,7 +108,7 @@ cases: # Nested aggregations - name: nested_avg_of_quantile query: SELECT L1, AVG(result) FROM (SELECT L1, L2, QUANTILE(0.95, value) AS result FROM cpu_usage WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY L1, L2) GROUP BY L1 - expected: unsupported + expected: supported - name: nested_sum_of_sum query: SELECT L1, SUM(total) FROM (SELECT L1, L2, SUM(value) AS total FROM cpu_usage WHERE time BETWEEN DATEADD(s, -10, NOW()) AND NOW() GROUP BY L1, L2) GROUP BY L1 - expected: unsupported + expected: supported diff --git a/asap-query-engine/tests/sql_query_support/nested-aggregations.yaml b/asap-query-engine/tests/sql_query_support/nested-aggregations.yaml index fdab2dbf..7e1bad89 100644 --- a/asap-query-engine/tests/sql_query_support/nested-aggregations.yaml +++ b/asap-query-engine/tests/sql_query_support/nested-aggregations.yaml @@ -16,19 +16,19 @@ repetition_delay_ms: 60000 cases: - name: one-stage-sum query: SELECT job, SUM(v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported - name: one-stage-count query: SELECT job, COUNT(v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported - name: one-stage-avg query: SELECT job, AVG(v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported - name: one-stage-min query: SELECT job, MIN(v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported - name: one-stage-max query: SELECT job, MAX(v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported - name: one-stage-topk-3 query: SELECT job, instance, v FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) ORDER BY v DESC LIMIT 3 BY job expected: unsupported @@ -37,13 +37,13 @@ cases: expected: unsupported - name: one-stage-quantile-0-5 query: SELECT job, QUANTILE(0.5, v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported - name: one-stage-quantile-0-75 query: SELECT job, QUANTILE(0.75, v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported - name: one-stage-quantile-0-9 query: SELECT job, QUANTILE(0.9, v) FROM (SELECT job, instance, SUM(value) AS v FROM data WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY job, instance) GROUP BY job - expected: unsupported + expected: supported # Middle stage: sum by (job) - name: two-stage-sum-then-sum From d2d09b28822405b238f4653c5ebe38be18508ec8 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 20:10:53 -0400 Subject: [PATCH 4/5] fix(sql-parser): reject CTEs, joins and QUALIFY instead of dropping them cte_to_subquery bound FROM to the first CTE whatever name it referenced, and replaced FROM wholesale, dropping its joins. Every CTE query was nested and so rejected until nested queries became supported; reject WITH outright instead. Joins, extra FROM entries, QUALIFY, CONNECT BY and value-table modes are now rejected on every layer, flat queries included, rather than silently ignored. Co-Authored-By: Claude Opus 5.5 --- .../src/ast_matching/sqlparser_test.rs | 53 ++++++++++++++++++ .../src/ast_matching/sqlpattern_parser.rs | 55 ++++++------------- 2 files changed, 70 insertions(+), 38 deletions(-) diff --git a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs index 99ff40e8..dc87d2fc 100644 --- a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs +++ b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs @@ -1540,6 +1540,59 @@ mod tests { format!( "SELECT L1, SUM(m) FROM (SELECT L1, SUM(v) AS m FROM ({NESTED_INNER}) GROUP BY L1) GROUP BY L1" ), + // Joins and extra FROM entries on the outer layer. + format!( + "SELECT L1, SUM(v) FROM ({NESTED_INNER}) t JOIN mem_usage m ON t.L1 = m.L1 GROUP BY L1" + ), + format!("SELECT L1, SUM(v) FROM ({NESTED_INNER}) t, mem_usage GROUP BY L1"), + // A join on the inner layer. + "SELECT L1, SUM(v) FROM (SELECT L1, L2, SUM(value) AS v FROM cpu_usage \ + JOIN mem_usage ON cpu_usage.L1 = mem_usage.L1 \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1, L2) GROUP BY L1" + .to_string(), + // Inner QUALIFY. + "SELECT L1, SUM(v) FROM (SELECT L1, L2, SUM(value) AS v FROM cpu_usage \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1, L2 QUALIFY v > 1) \ + GROUP BY L1" + .to_string(), + ]; + for sql in rejected { + assert!(parse_sql_query(&sql).is_none(), "should not parse: {sql}"); + } + } + + /// A CTE would be served from whichever CTE comes first, whatever name + /// FROM references, so CTEs are rejected outright. + #[test] + fn test_ctes_are_rejected() { + let rejected = [ + format!("WITH t AS ({NESTED_INNER}) SELECT L1, SUM(v) FROM t GROUP BY L1"), + format!( + "WITH a AS ({NESTED_INNER}), b AS (SELECT L1, L2, MAX(value) AS v FROM cpu_usage \ + WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW() GROUP BY L1, L2) \ + SELECT L1, SUM(v) FROM b GROUP BY L1" + ), + format!( + "SELECT L1, SUM(v) FROM (WITH t AS ({NESTED_INNER}) SELECT * FROM t) GROUP BY L1" + ), + ]; + for sql in rejected { + assert!(parse_sql_query(&sql).is_none(), "should not parse: {sql}"); + } + } + + /// Single-level queries reject the same unrepresentable clauses: each + /// would otherwise be dropped from the query's data. + #[test] + fn test_flat_query_rejects_joins_and_qualify() { + let window = "WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW()"; + let rejected = [ + format!( + "SELECT L1, SUM(value) FROM cpu_usage JOIN mem_usage ON cpu_usage.L1 = mem_usage.L1 \ + {window} GROUP BY L1" + ), + format!("SELECT L1, SUM(value) FROM cpu_usage, mem_usage {window} GROUP BY L1"), + format!("SELECT L1, SUM(value) AS s FROM cpu_usage {window} GROUP BY L1 QUALIFY s > 1"), ]; for sql in rejected { assert!(parse_sql_query(&sql).is_none(), "should not parse: {sql}"); diff --git a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs index a6e572ca..64e24ad5 100644 --- a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs +++ b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs @@ -65,8 +65,10 @@ impl SQLPatternParser { let order_by_items = self.parse_order_by_items(query)?; let limit = self.parse_limit_value(query)?; - // Convert CTE to subquery if present - let query = self.cte_to_subquery(query); + if query.with.is_some() { + println!("CTEs (WITH) are not supported"); + return None; + } let mut data = match &query.body.as_ref() { SetExpr::Select(select) => self.parse_select(select)?, @@ -161,41 +163,12 @@ impl SQLPatternParser { } } - fn cte_to_subquery(&self, query: &Query) -> Query { - let mut query = query.clone(); - - if let Some(with) = &query.with { - if !with.cte_tables.is_empty() { - let cte = &with.cte_tables[0]; - - // Create a subquery from the CTE - if let Some(new_body) = match &query.body.as_ref() { - SetExpr::Select(select) => { - let mut new_select = select.clone(); - new_select.from = vec![TableWithJoins { - relation: TableFactor::Derived { - lateral: false, - subquery: Box::new(*(cte.query).clone()), - alias: None, - }, - joins: vec![], - }]; - Some(SetExpr::Select(Box::new(*new_select))) - } - _ => None, - } { - query.body = Box::new(new_body); - query.with = None; - } - } - } - - query - } - - /// SELECT clauses that SQLQueryData cannot represent. + /// SELECT clauses that SQLQueryData cannot represent. FROM must be a + /// single table or subquery: joins and extra entries would be dropped. fn has_unsupported_clauses(select: &Select) -> bool { - select.distinct.is_some() + select.from.len() != 1 + || !select.from[0].joins.is_empty() + || select.distinct.is_some() || select.top.is_some() || select.into.is_some() || !select.lateral_views.is_empty() @@ -205,7 +178,10 @@ impl SQLPatternParser { || !select.sort_by.is_empty() || select.having.is_some() || !select.named_window.is_empty() + || select.qualify.is_some() || select.window_before_qualify + || select.connect_by.is_some() + || select.value_table_mode.is_some() } fn parse_select(&self, select: &Select) -> Option { @@ -249,10 +225,13 @@ impl SQLPatternParser { let subquery = match &select.from[0].relation { TableFactor::Derived { subquery, .. } => match subquery.body.as_ref() { SetExpr::Select(inner_select) => { + if subquery.with.is_some() { + println!("CTEs (WITH) are not supported"); + return None; + } // Query-level clauses of the inner layer have no place // in its SQLQueryData, so they would be dropped. - if subquery.with.is_some() - || subquery.order_by.is_some() + if subquery.order_by.is_some() || subquery.limit_clause.is_some() || subquery.fetch.is_some() { From 7791d233c0d22f3bbaa7c56f97a45e60198c985d Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 20:12:54 -0400 Subject: [PATCH 5/5] refactor(planner): share the aggregation name-to-operator mapping Add QueryTimeAggregationOperator::from_name and use it in both the PromQL and SQL planners. A test checks that every outer aggregation the SQL matcher accepts maps to an operator, since sql_utilities cannot depend on asap_types to share the list itself. Co-Authored-By: Claude Opus 5.5 --- .../rs/asap_types/src/query_config.rs | 42 +++++++++++++++++++ .../src/ast_matching/sqlpattern_matcher.rs | 2 +- asap-planner-rs/src/planner/promql.rs | 11 +---- asap-planner-rs/src/planner/sql.rs | 19 +++------ 4 files changed, 50 insertions(+), 24 deletions(-) diff --git a/asap-common/dependencies/rs/asap_types/src/query_config.rs b/asap-common/dependencies/rs/asap_types/src/query_config.rs index 644b32b4..dd109567 100644 --- a/asap-common/dependencies/rs/asap_types/src/query_config.rs +++ b/asap-common/dependencies/rs/asap_types/src/query_config.rs @@ -22,6 +22,23 @@ pub enum QueryTimeAggregationOperator { Topk, } +impl QueryTimeAggregationOperator { + /// The operator for an aggregation function name in any case, e.g. + /// PromQL's `sum` or SQL's `SUM`. + pub fn from_name(name: &str) -> Option { + match name.to_ascii_lowercase().as_str() { + "sum" => Some(Self::Sum), + "count" => Some(Self::Count), + "avg" => Some(Self::Avg), + "min" => Some(Self::Min), + "max" => Some(Self::Max), + "quantile" => Some(Self::Quantile), + "topk" => Some(Self::Topk), + _ => None, + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum QueryTimeGroupingMode { @@ -193,6 +210,31 @@ mod tests { .is_err()); } + #[test] + fn operator_names_parse_case_insensitively() { + assert!(matches!( + QueryTimeAggregationOperator::from_name("quantile"), + Some(QueryTimeAggregationOperator::Quantile) + )); + assert!(matches!( + QueryTimeAggregationOperator::from_name("QUANTILE"), + Some(QueryTimeAggregationOperator::Quantile) + )); + assert!(QueryTimeAggregationOperator::from_name("stddev").is_none()); + } + + /// The SQL matcher accepts these outer aggregations on its own list; the + /// planner turns them into operators, so every one must map. + #[test] + fn every_sql_outer_aggregation_has_an_operator() { + for name in sql_utilities::ast_matching::QUERY_TIME_AGGREGATIONS { + assert!( + QueryTimeAggregationOperator::from_name(name).is_some(), + "{name}" + ); + } + } + #[test] fn without_empty_labels_is_a_valid_promql_grouping() { assert!(QueryTimeAggregation { diff --git a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs index d9ca6075..2c28c0e4 100644 --- a/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs +++ b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs @@ -30,7 +30,7 @@ pub enum QueryError { } /// Aggregations an outer layer can apply to its inner layer's rows. -const QUERY_TIME_AGGREGATIONS: [&str; 6] = ["SUM", "COUNT", "AVG", "MIN", "MAX", "QUANTILE"]; +pub const QUERY_TIME_AGGREGATIONS: [&str; 6] = ["SUM", "COUNT", "AVG", "MIN", "MAX", "QUANTILE"]; #[derive(Debug)] pub struct SQLQuery { diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs index 38ebb451..e9588368 100644 --- a/asap-planner-rs/src/planner/promql.rs +++ b/asap-planner-rs/src/planner/promql.rs @@ -42,16 +42,7 @@ pub struct NestedAggregationPlan { pub(crate) fn aggregation_stage( aggregate: &promql_parser::parser::AggregateExpr, ) -> Option { - let operator = match aggregate.op.to_string().as_str() { - "sum" => QueryTimeAggregationOperator::Sum, - "count" => QueryTimeAggregationOperator::Count, - "avg" => QueryTimeAggregationOperator::Avg, - "min" => QueryTimeAggregationOperator::Min, - "max" => QueryTimeAggregationOperator::Max, - "quantile" => QueryTimeAggregationOperator::Quantile, - "topk" => QueryTimeAggregationOperator::Topk, - _ => return None, - }; + let operator = QueryTimeAggregationOperator::from_name(&aggregate.op.to_string())?; let grouping = match &aggregate.modifier { Some(promql_parser::parser::LabelModifier::Include(labels)) if !labels.is_empty() => { diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs index b4be38fe..b06926d3 100644 --- a/asap-planner-rs/src/planner/sql.rs +++ b/asap-planner-rs/src/planner/sql.rs @@ -252,19 +252,12 @@ fn sql_query_time_aggregation( outer: &SQLQueryData, ) -> Result { let info = &outer.aggregation_info; - let operator = match info.get_name() { - "SUM" => QueryTimeAggregationOperator::Sum, - "COUNT" => QueryTimeAggregationOperator::Count, - "AVG" => QueryTimeAggregationOperator::Avg, - "MIN" => QueryTimeAggregationOperator::Min, - "MAX" => QueryTimeAggregationOperator::Max, - "QUANTILE" => QueryTimeAggregationOperator::Quantile, - other => { - return Err(ControllerError::SqlParse(format!( - "unsupported outer aggregation: {other}" - ))) - } - }; + let operator = QueryTimeAggregationOperator::from_name(info.get_name()).ok_or_else(|| { + ControllerError::SqlParse(format!( + "unsupported outer aggregation: {}", + info.get_name() + )) + })?; let parameter = match operator { QueryTimeAggregationOperator::Quantile => { let phi = info