Harden SQL top-k detection (LIMIT 0, alias case) and expand planner/engine tests

Author: akanksha-akkihal

asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlhelper.rs

@@ -280,6 +280,10 @@ impl SqlTopk {
 /// query pattern type; the planner rejects nested queries up front).
 pub fn detect_sql_topk(query_data: &SQLQueryData) -> Option<SqlTopk> {
     let k = query_data.limit?;
+    // LIMIT 0 is an empty-result query, not a top-k heavy-hitter request.
+    if k == 0 {
+        return None;
+    }
     // Need a GROUP BY key to rank and an ORDER BY to define "top".
     if query_data.labels.is_empty() || query_data.order_by.is_empty() {
         return None;
@@ -300,8 +304,11 @@ pub fn detect_sql_topk(query_data: &SQLQueryData) -> Option<SqlTopk> {
     if primary.ascending {
         return None;
     }
-    if query_data.aggregation_alias.as_deref() != Some(primary.column.as_str()) {
+    // ORDER BY may differ only by identifier case for unquoted aliases.
+    let alias = query_data.aggregation_alias.as_deref()?;
+    if !alias.eq_ignore_ascii_case(primary.column.as_str()) {
         return None;
     }
     Some(SqlTopk { k, weighting })
 }
+

asap-planner-rs/src/planner/sql.rs

@@ -105,6 +105,8 @@ impl SQLSingleQueryProcessor {
 
         // Label routing
         let spatial_output = KeyByLabelNames::new(labels.iter().cloned().collect::<Vec<_>>());
+        // Top-k needs ORDER BY / LIMIT from the parser; SQLPatternMatcher drops them
+        // when building `sql_query.query_data[0]`, so use `qdata` not query_data[0].
         let sql_topk = detect_sql_topk(&qdata);
         let treatment_type = get_sql_treatment_type(agg_info.get_name());
         let statistics = if sql_topk.is_some() {

asap-planner-rs/tests/sql_integration.rs

@@ -890,8 +890,28 @@ fn spatial_sum_topk_heap() {
         .unwrap();
 
     assert_eq!(out.streaming_aggregation_count(), 1);
+    assert_eq!(out.inference_query_count(), 1);
     assert!(out.has_aggregation_type("CountMinSketchWithHeap"));
+    assert!(out.has_aggregation_type_and_sub_type("CountMinSketchWithHeap", "topk"));
     assert!(!out.has_aggregation_type("DeltaSetAggregator"));
+    assert!(!out.has_aggregation_type("CountMinSketch"));
+    assert!(out.all_tumbling_window_sizes_eq(1));
+    assert_eq!(
+        out.aggregation_labels("CountMinSketchWithHeap", "grouping"),
+        Vec::<String>::new()
+    );
+    assert_eq!(
+        out.aggregation_labels("CountMinSketchWithHeap", "aggregated"),
+        vec!["srcip".to_string()]
+    );
+    let mut rollup = out.aggregation_labels("CountMinSketchWithHeap", "rollup");
+    rollup.sort();
+    assert_eq!(rollup, vec!["dstip".to_string(), "proto".to_string()]);
+    assert_eq!(
+        out.aggregation_parameter("CountMinSketchWithHeap", "heapsize")
+            .and_then(|v| v.as_u64()),
+        Some(40)
+    );
     assert_eq!(
         out.aggregation_parameter("CountMinSketchWithHeap", "count_events")
             .and_then(|v| v.as_bool()),
@@ -915,3 +935,37 @@ fn spatial_count_without_order_by_is_not_topk() {
     assert!(out.has_aggregation_type("DeltaSetAggregator"));
     assert!(!out.has_aggregation_type("CountMinSketchWithHeap"));
 }
+
+/// ORDER BY aggregate alias ASC (bottom-k) stays on the CMS + DeltaSet path.
+#[test]
+fn spatial_count_order_by_asc_limit_is_not_topk() {
+    let q = "SELECT srcip, COUNT(pkt_len) AS transfer_events FROM netflow_table \
+             WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) \
+             GROUP BY srcip ORDER BY transfer_events ASC LIMIT 10";
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+        .unwrap()
+        .generate()
+        .unwrap();
+
+    assert_eq!(out.streaming_aggregation_count(), 2);
+    assert!(out.has_aggregation_type("CountMinSketch"));
+    assert!(out.has_aggregation_type("DeltaSetAggregator"));
+    assert!(!out.has_aggregation_type("CountMinSketchWithHeap"));
+}
+
+/// LIMIT 0 is treated as non-top-k and uses the normal CMS + DeltaSet path.
+#[test]
+fn spatial_count_limit_zero_is_not_topk() {
+    let q = "SELECT srcip, COUNT(pkt_len) AS transfer_events FROM netflow_table \
+             WHERE time BETWEEN DATEADD(s, -11, NOW()) AND DATEADD(s, -10, NOW()) \
+             GROUP BY srcip ORDER BY transfer_events DESC LIMIT 0";
+    let out = SQLController::from_yaml(&netflow_one_query_config(q, 1), sql_opts_1s_ingest())
+        .unwrap()
+        .generate()
+        .unwrap();
+
+    assert_eq!(out.streaming_aggregation_count(), 2);
+    assert!(out.has_aggregation_type("CountMinSketch"));
+    assert!(out.has_aggregation_type("DeltaSetAggregator"));
+    assert!(!out.has_aggregation_type("CountMinSketchWithHeap"));
+}

asap-query-engine/src/engines/simple_engine/sql.rs

@@ -772,7 +772,7 @@ impl SimpleEngine {
 #[cfg(test)]
 mod detect_topk_tests {
     use sql_utilities::ast_matching::{detect_sql_topk, SQLPatternParser, SqlTopk, TopkWeighting};
-    use sql_utilities::sqlhelper::{SQLSchema, Table};
+    use sql_utilities::sqlhelper::{AggregationInfo, OrderByItem, SQLQueryData, SQLSchema, Table, TimeInfo};
     use sqlparser::dialect::GenericDialect;
     use sqlparser::parser::Parser;
     use std::collections::HashSet;
@@ -836,6 +836,42 @@ mod detect_topk_tests {
         );
     }
 
+    #[test]
+    fn alias_case_mismatch_still_detects_topk() {
+        // The parser path can normalize/canonicalize identifiers; verify directly on
+        // SQLQueryData that alias matching in detect_sql_topk is case-insensitive.
+        let qd = SQLQueryData {
+            aggregation_info: AggregationInfo::new("COUNT".to_string(), "pkt_len".to_string(), vec![]),
+            aggregation_alias: Some("transfer_events".to_string()),
+            metric: "netflow_table".to_string(),
+            labels: HashSet::from(["srcip".to_string()]),
+            time_info: TimeInfo::new("time".to_string(), 0.0, 1.0),
+            subquery: None,
+            order_by: vec![OrderByItem {
+                column: "TRANSFER_EVENTS".to_string(),
+                ascending: false,
+            }],
+            limit: Some(10),
+        };
+        assert_eq!(
+            detect_sql_topk(&qd),
+            Some(SqlTopk {
+                k: 10,
+                weighting: TopkWeighting::Count,
+            }),
+        );
+    }
+
+    #[test]
+    fn zero_limit_is_not_topk() {
+        let sql = format!(
+            "SELECT srcip, COUNT(pkt_len) AS transfer_events FROM netflow_table {WINDOW} \
+             GROUP BY srcip ORDER BY transfer_events DESC LIMIT 0"
+        );
+        let qd = parse(&sql).expect("query should parse");
+        assert_eq!(detect_sql_topk(&qd), None, "LIMIT 0 is not top-k");
+    }
+
     #[test]
     fn missing_limit_is_not_topk() {
         let sql = format!(