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/sqlparser_test.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs index 5c332376..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 @@ -1403,4 +1403,199 @@ 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" + ), + // 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_matcher.rs b/asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_matcher.rs index d24a22b9..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 @@ -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. +pub 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..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,36 +163,25 @@ 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. FROM must be a + /// single table or subquery: joins and extra entries would be dropped. + fn has_unsupported_clauses(select: &Select) -> bool { + 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() + || 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.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 { @@ -204,26 +195,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 +214,34 @@ 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) => { + 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.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)?; 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 bbf2e67d..b06926d3 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,138 @@ 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 = 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 + .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 +306,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(_)))); +} 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