Skip to content

Commit 767b7b6

Browse files
fix(sql-parser): accept global aggregates without GROUP BY (#826)
* fix(sql-parser): accept global aggregates without group by * fix(query-engine): execute global SQL aggregates * fix(query-engine): stub benchmark binary in Docker build * fix(query-engine): reject partitioned scalar fallbacks
1 parent cc1d7ce commit 767b7b6

11 files changed

Lines changed: 284 additions & 60 deletions

File tree

‎asap-common/dependencies/rs/asap_types/src/capability_matching.rs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,15 @@ pub fn labels_compatible(config_labels: &KeyByLabelNames, req_labels: &KeyByLabe
157157
config_labels == req_labels
158158
}
159159

160+
fn scalar_requirement_compatible(
161+
config: &AggregationConfig,
162+
requirements: &QueryRequirements,
163+
) -> bool {
164+
!requirements.grouping_labels.labels.is_empty()
165+
|| requirements.statistics.contains(&Statistic::Topk)
166+
|| config.aggregated_labels.labels.is_empty()
167+
}
168+
160169
/// Spatial filter compatibility.
161170
/// - Both empty → compatible.
162171
/// - Config non-empty and matches query → compatible.
@@ -290,6 +299,7 @@ pub fn find_compatible_aggregation(
290299
&& sub_type.is_none_or(|st| c.aggregation_sub_type.eq_ignore_ascii_case(st))
291300
&& window_compatible(c, requirements.data_range_ms)
292301
&& labels_compatible(&c.grouping_labels, &requirements.grouping_labels)
302+
&& scalar_requirement_compatible(c, requirements)
293303
&& spatial_filter_compatible(
294304
&c.spatial_filter_normalized,
295305
&requirements.spatial_filter_normalized,
@@ -552,6 +562,20 @@ mod tests {
552562
assert_eq!(result.unwrap().aggregation_id_for_value, 1);
553563
}
554564

565+
#[test]
566+
fn scalar_sum_does_not_match_an_internally_partitioned_config() {
567+
let mut value = make_config(1, "cpu", "MultipleSum", "sum", 300_000, "tumbling", &[], "");
568+
value.aggregated_labels = KeyByLabelNames::new(vec!["host".to_string()]);
569+
let key = make_config(2, "cpu", "SetAggregator", "", 300_000, "tumbling", &[], "");
570+
let configs = HashMap::from([(1, value), (2, key)]);
571+
572+
assert!(
573+
find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], ""))
574+
.is_none(),
575+
"a scalar query cannot be served by an internally partitioned aggregation"
576+
);
577+
}
578+
555579
#[test]
556580
fn plain_cms_matching_respects_sum_and_count_subtypes() {
557581
let mut configs = HashMap::new();

‎asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlparser_test.rs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,37 @@ mod tests {
8282
assert!(query.labels.contains("L1"));
8383
}
8484

85+
#[test]
86+
fn test_global_aggregate_parses_with_empty_grouping_labels() {
87+
let query = parse_sql_query(
88+
"SELECT SUM(value) FROM cpu_usage WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW()",
89+
)
90+
.expect("global aggregate parses");
91+
92+
assert!(query.labels.is_empty());
93+
}
94+
95+
#[test]
96+
fn test_nested_global_aggregates_parse_with_empty_grouping_labels() {
97+
let query = parse_sql_query(
98+
"SELECT SUM(total) FROM (SELECT SUM(value) AS total FROM cpu_usage WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW())",
99+
)
100+
.expect("nested global aggregates parse");
101+
102+
// Matching and execution may reject nested queries, but parsing retains both layers.
103+
assert!(query.labels.is_empty());
104+
assert!(query.subquery.expect("inner query").labels.is_empty());
105+
}
106+
107+
#[test]
108+
fn test_nested_global_aggregates_remain_unsupported_by_matcher() {
109+
check_query(
110+
"SELECT SUM(total) FROM (SELECT SUM(value) AS total FROM cpu_usage WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW())",
111+
vec![],
112+
Some(QueryError::NestedQueryUnsupported),
113+
);
114+
}
115+
85116
#[test]
86117
fn test_pattern_matching() {
87118
let schema = create_test_schema();

‎asap-common/dependencies/rs/sql_utilities/src/ast_matching/sqlpattern_parser.rs‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -847,11 +847,7 @@ impl SQLPatternParser {
847847
}
848848
}
849849

850-
if group_bys.is_empty() {
851-
None
852-
} else {
853-
Some(group_bys)
854-
}
850+
Some(group_bys)
855851
}
856852
_ => None,
857853
}

‎asap-query-engine/src/engines/simple_engine/mod.rs‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3141,11 +3141,9 @@ impl SimpleEngine {
31413141
Some(&query_bounds),
31423142
)
31433143
.into_iter()
3144-
// A fully unlabeled result (fallback_key was None and
3145-
// the value accumulator has no self-keys) has no
3146-
// RangeVectorElement representation -- matches today's
3147-
// behavior of producing no sample for this combination.
3148-
.filter_map(|(key, value)| key.map(|key| (key, value)))
3144+
// A missing store key denotes the single scalar row of
3145+
// a global aggregate; represent it with empty labels.
3146+
.map(|(key, value)| (key.unwrap_or_else(KeyByLabelValues::new), value))
31493147
.collect();
31503148

31513149
// Rank and truncate to k within this group only -- a

‎asap-query-engine/src/tests/native_range_query_tests.rs‎

Lines changed: 68 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2202,7 +2202,11 @@ mod tests {
22022202
fn create_range_engine_self_keyed(
22032203
metric: &str,
22042204
aggregated_label: &str,
2205-
data: Vec<(u64, CountMinSketchWithHeapAccumulator)>,
2205+
data: Vec<(
2206+
u64,
2207+
Option<KeyByLabelValues>,
2208+
CountMinSketchWithHeapAccumulator,
2209+
)>,
22062210
promql_query: &str,
22072211
) -> SimpleEngine {
22082212
let mut aggregation_configs = HashMap::new();
@@ -2239,8 +2243,8 @@ mod tests {
22392243
CleanupPolicy::NoCleanup,
22402244
));
22412245

2242-
for (timestamp, acc) in data {
2243-
let output = PrecomputedOutput::new(timestamp - WINDOW_MS, timestamp, None, 1);
2246+
for (timestamp, key, acc) in data {
2247+
let output = PrecomputedOutput::new(timestamp - WINDOW_MS, timestamp, key, 1);
22442248
store
22452249
.insert_precomputed_output(output, Box::new(acc))
22462250
.unwrap();
@@ -2287,8 +2291,12 @@ mod tests {
22872291
sketch.inner.update("host-c", 10.0);
22882292

22892293
let query = "topk(5, transfer_events)";
2290-
let engine =
2291-
create_range_engine_self_keyed("transfer_events", "srcip", vec![(1000, sketch)], query);
2294+
let engine = create_range_engine_self_keyed(
2295+
"transfer_events",
2296+
"srcip",
2297+
vec![(1000, None, sketch)],
2298+
query,
2299+
);
22922300

22932301
let result = engine
22942302
.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0)
@@ -2333,6 +2341,61 @@ mod tests {
23332341
);
23342342
}
23352343

2344+
#[tokio::test(flavor = "multi_thread")]
2345+
async fn range_query_empty_self_keyed_topk_does_not_emit_outer_key() {
2346+
let query = "topk(5, transfer_events)";
2347+
let engine = create_range_engine_self_keyed(
2348+
"transfer_events",
2349+
"srcip",
2350+
vec![(
2351+
1000,
2352+
Some(KeyByLabelValues::new()),
2353+
CountMinSketchWithHeapAccumulator::new(3, 1024, 32),
2354+
)],
2355+
query,
2356+
);
2357+
2358+
let result = engine
2359+
.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0)
2360+
.expect("native query execution should not fail");
2361+
let (_, query_result) = result.expect("range query failed");
2362+
2363+
assert!(
2364+
matrix_values(query_result).is_empty(),
2365+
"an empty self-keyed heap must not become an unlabeled result"
2366+
);
2367+
}
2368+
2369+
#[tokio::test(flavor = "multi_thread")]
2370+
async fn range_query_global_sum_returns_one_unlabeled_series() {
2371+
let query = "sum(requests)";
2372+
let engine = create_engine_multi_timestamp_with_window(
2373+
"requests",
2374+
AggregationType::Sum,
2375+
vec![],
2376+
vec![(
2377+
1000,
2378+
None,
2379+
Box::new(SumAccumulator::with_sum(42.0)) as Box<dyn AggregateCore>,
2380+
)],
2381+
query,
2382+
1000,
2383+
1000,
2384+
WindowType::Tumbling,
2385+
);
2386+
2387+
let result = engine
2388+
.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0)
2389+
.expect("native query execution should not fail");
2390+
let (_, query_result) = result.expect("range query failed");
2391+
let elements = matrix_values(query_result);
2392+
2393+
assert_eq!(elements.len(), 1);
2394+
assert_eq!(elements[0].labels, KeyByLabelValues::new());
2395+
assert_eq!(elements[0].samples.len(), 1);
2396+
assert_eq!(elements[0].samples[0].value, 42.0);
2397+
}
2398+
23362399
#[tokio::test(flavor = "multi_thread")]
23372400
async fn range_query_set_aggregator_merges_multiple_buckets_within_one_window() {
23382401
// Every other SetAggregator test uses window_size_ms == bucket

0 commit comments

Comments
 (0)