Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions asap-common/dependencies/rs/asap_types/src/capability_matching.rs
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,15 @@ pub fn labels_compatible(config_labels: &KeyByLabelNames, req_labels: &KeyByLabe
config_labels == req_labels
}

fn scalar_requirement_compatible(
config: &AggregationConfig,
requirements: &QueryRequirements,
) -> bool {
!requirements.grouping_labels.labels.is_empty()
|| requirements.statistics.contains(&Statistic::Topk)
|| config.aggregated_labels.labels.is_empty()
}

/// Spatial filter compatibility.
/// - Both empty → compatible.
/// - Config non-empty and matches query → compatible.
Expand Down Expand Up @@ -290,6 +299,7 @@ pub fn find_compatible_aggregation(
&& sub_type.is_none_or(|st| c.aggregation_sub_type.eq_ignore_ascii_case(st))
&& window_compatible(c, requirements.data_range_ms)
&& labels_compatible(&c.grouping_labels, &requirements.grouping_labels)
&& scalar_requirement_compatible(c, requirements)
&& spatial_filter_compatible(
&c.spatial_filter_normalized,
&requirements.spatial_filter_normalized,
Expand Down Expand Up @@ -552,6 +562,20 @@ mod tests {
assert_eq!(result.unwrap().aggregation_id_for_value, 1);
}

#[test]
fn scalar_sum_does_not_match_an_internally_partitioned_config() {
let mut value = make_config(1, "cpu", "MultipleSum", "sum", 300_000, "tumbling", &[], "");
value.aggregated_labels = KeyByLabelNames::new(vec!["host".to_string()]);
let key = make_config(2, "cpu", "SetAggregator", "", 300_000, "tumbling", &[], "");
let configs = HashMap::from([(1, value), (2, key)]);

assert!(
find_compatible_aggregation(&configs, &req("cpu", &[Statistic::Sum], 300_000, &[], ""))
.is_none(),
"a scalar query cannot be served by an internally partitioned aggregation"
);
}

#[test]
fn plain_cms_matching_respects_sum_and_count_subtypes() {
let mut configs = HashMap::new();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,37 @@ mod tests {
assert!(query.labels.contains("L1"));
}

#[test]
fn test_global_aggregate_parses_with_empty_grouping_labels() {
let query = parse_sql_query(
"SELECT SUM(value) FROM cpu_usage WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW()",
)
.expect("global aggregate parses");

assert!(query.labels.is_empty());
}

#[test]
fn test_nested_global_aggregates_parse_with_empty_grouping_labels() {
let query = parse_sql_query(
"SELECT SUM(total) FROM (SELECT SUM(value) AS total FROM cpu_usage WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW())",
)
.expect("nested global aggregates parse");

// Matching and execution may reject nested queries, but parsing retains both layers.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[test-coverage] No negative-path test for nested global aggregates

This test only shows that parsing keeps both layers. Nothing asserts that the matcher/planner then rejects the shape with NestedQueryUnsupported, so loosening the n > 1 check in query_info_to_pattern would let these flow into execution with time_info UNUSED and no test would fail. The "may reject" comment also leaves the expected outcome unstated.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was generated by AI during triage.

Fixed in c9bc666. test_nested_global_aggregates_remain_unsupported_by_matcher now parses the nested global aggregate and asserts QueryError::NestedQueryUnsupported, making the accepted parser shape and rejected matcher shape explicit.

assert!(query.labels.is_empty());
assert!(query.subquery.expect("inner query").labels.is_empty());
}

#[test]
fn test_nested_global_aggregates_remain_unsupported_by_matcher() {
check_query(
"SELECT SUM(total) FROM (SELECT SUM(value) AS total FROM cpu_usage WHERE time BETWEEN DATEADD(s, -1, NOW()) AND NOW())",
vec![],
Some(QueryError::NestedQueryUnsupported),
);
}

#[test]
fn test_pattern_matching() {
let schema = create_test_schema();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -847,11 +847,7 @@ impl SQLPatternParser {
}
}

if group_bys.is_empty() {
None
} else {
Some(group_bys)
}
Some(group_bys)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[correctness] Global SUM/COUNT can match a label-keyed CMS via capability fallback (asap-query-engine/src/engines/simple_engine/sql.rs:258)

Global aggregates now reach find_compatible_aggregation with empty grouping_labels, and labels_compatible (capability_matching.rs:157) compares only grouping_labels, never aggregated_labels.

Scenario: the planner plans SELECT job, SUM(value) ... GROUP BY job, giving CountMinSketch(sum) with grouping=[] and aggregated=[job]. An unplanned SELECT SUM(value) FROM data ... has grouping=[], matches that config, and point-queries the empty key, returning ~0 instead of the total. Before this PR these queries were rejected at parse time.

Suggested fix: require aggregated_labels to be empty when the requirement has no grouping labels, or exclude configs with aggregated labels.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was generated by AI during triage.

Fixed in c9bc666. Non-top-k global aggregates now require an exact query configuration instead of capability fallback, so a grouped CMS cannot be selected for an unlabeled scalar query. Added global_aggregate_does_not_fallback_to_grouped_config.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[correctness] Global COUNT also matches top-k CountMinSketchWithHeap configs (asap-query-engine/src/engines/simple_engine/sql.rs:258)

Statistic::Count lists CountMinSketchWithHeap as compatible, and cms_family_sub_type_compatible only checks count vs sum mode. Heap configs always have grouping=[].

Scenario: SELECT srcip, COUNT(value) AS c ... GROUP BY srcip ORDER BY c DESC LIMIT 5 (heap: grouping=[], aggregated=[srcip], count) plus an unplanned SELECT COUNT(value) FROM t .... The global COUNT is served by the srcip-keyed heap sketch and returns a bogus empty-key estimate instead of the event count.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was generated by AI during triage.

Fixed by the same c9bc666 guard: a non-top-k global COUNT cannot capability-match a keyed heap configuration. Exact configurations remain eligible, and existing top-k capability-fallback tests still pass.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[correctness] Subquery path skips the unexpected-SELECT-fields check (sqlpattern_parser.rs:244)

Inner subqueries without GROUP BY now parse, but the subquery branch never runs the DISTINCT/HAVING/QUALIFY check that the flat path applies.

Scenario: SELECT SUM(t) FROM (SELECT SUM(value) AS t FROM x WHERE time BETWEEN ... HAVING SUM(value) > 5) used to be rejected by get_groupbys and now parses with the HAVING dropped. Only the matcher's NestedQueryUnsupported stops it today.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was generated by AI during triage.

Deferred intentionally: this PR changes parsing only, while nested-query matching and execution remain unsupported. c9bc666 adds a matcher regression asserting NestedQueryUnsupported, so the parsed shape cannot enter execution. Tightening unsupported-clause validation inside nested parsing belongs with the nested-execution work.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[efficiency] Global SUM/COUNT plans a full CMS + DeltaSetAggregator for one scalar (asap-planner-rs/src/planner/sql.rs:124)

With zero aggregated labels, the planner allocates a multi-population CountMinSketch (width x depth) plus a key tracker for a population that is always one empty key. When spatial_output is empty, mapping to a single-population operator (Sum, or MinMax for exact) would be much cheaper.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was generated by AI during triage.

Acknowledged and deferred. This PR is scoped to parser acceptance and safe execution of exact global configurations; choosing a cheaper single-population physical plan is a planner optimization and should be addressed separately.

}
_ => None,
}
Expand Down
1 change: 1 addition & 0 deletions asap-query-engine/Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ RUN mkdir -p asap-query-engine/src/bin \
&& echo "fn main() {}" > asap-query-engine/src/bin/precompute_engine.rs \
&& echo "fn main() {}" > asap-query-engine/src/bin/test_e2e_precompute.rs \
&& echo "fn main() {}" > asap-query-engine/src/bin/bench_precompute_sketch.rs \
&& echo "fn main() {}" > asap-query-engine/src/bin/bench_ingest_sender.rs \
&& echo "fn main() {}" > asap-query-engine/src/bin/e2e_quickstart_resource_test.rs \
&& mkdir -p asap-planner-rs/src && echo "fn main() {}" > asap-planner-rs/src/main.rs \
&& echo "pub fn placeholder() {}" >> asap-planner-rs/src/lib.rs
Expand Down
8 changes: 3 additions & 5 deletions asap-query-engine/src/engines/simple_engine/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3116,11 +3116,9 @@ impl SimpleEngine {
Some(&query_bounds),
)
.into_iter()
// A fully unlabeled result (fallback_key was None and
// the value accumulator has no self-keys) has no
// RangeVectorElement representation -- matches today's
// behavior of producing no sample for this combination.
.filter_map(|(key, value)| key.map(|key| (key, value)))
// A missing store key denotes the single scalar row of
// a global aggregate; represent it with empty labels.
.map(|(key, value)| (key.unwrap_or_else(KeyByLabelValues::new), value))
.collect();

// Rank and truncate to k within this group only -- a
Expand Down
73 changes: 68 additions & 5 deletions asap-query-engine/src/tests/native_range_query_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2202,7 +2202,11 @@ mod tests {
fn create_range_engine_self_keyed(
metric: &str,
aggregated_label: &str,
data: Vec<(u64, CountMinSketchWithHeapAccumulator)>,
data: Vec<(
u64,
Option<KeyByLabelValues>,
CountMinSketchWithHeapAccumulator,
)>,
promql_query: &str,
) -> SimpleEngine {
let mut aggregation_configs = HashMap::new();
Expand Down Expand Up @@ -2239,8 +2243,8 @@ mod tests {
CleanupPolicy::NoCleanup,
));

for (timestamp, acc) in data {
let output = PrecomputedOutput::new(timestamp - WINDOW_MS, timestamp, None, 1);
for (timestamp, key, acc) in data {
let output = PrecomputedOutput::new(timestamp - WINDOW_MS, timestamp, key, 1);
store
.insert_precomputed_output(output, Box::new(acc))
.unwrap();
Expand Down Expand Up @@ -2287,8 +2291,12 @@ mod tests {
sketch.inner.update("host-c", 10.0);

let query = "topk(5, transfer_events)";
let engine =
create_range_engine_self_keyed("transfer_events", "srcip", vec![(1000, sketch)], query);
let engine = create_range_engine_self_keyed(
"transfer_events",
"srcip",
vec![(1000, None, sketch)],
query,
);

let result = engine
.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0)
Expand Down Expand Up @@ -2333,6 +2341,61 @@ mod tests {
);
}

#[tokio::test(flavor = "multi_thread")]
async fn range_query_empty_self_keyed_topk_does_not_emit_outer_key() {
let query = "topk(5, transfer_events)";
let engine = create_range_engine_self_keyed(
"transfer_events",
"srcip",
vec![(
1000,
Some(KeyByLabelValues::new()),
CountMinSketchWithHeapAccumulator::new(3, 1024, 32),
)],
query,
);

let result = engine
.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0)
.expect("native query execution should not fail");
let (_, query_result) = result.expect("range query failed");

assert!(
matrix_values(query_result).is_empty(),
"an empty self-keyed heap must not become an unlabeled result"
);
}

#[tokio::test(flavor = "multi_thread")]
async fn range_query_global_sum_returns_one_unlabeled_series() {
let query = "sum(requests)";
let engine = create_engine_multi_timestamp_with_window(
"requests",
AggregationType::Sum,
vec![],
vec![(
1000,
None,
Box::new(SumAccumulator::with_sum(42.0)) as Box<dyn AggregateCore>,
)],
query,
1000,
1000,
WindowType::Tumbling,
);

let result = engine
.handle_range_query_promql(query.to_string(), 1.0, 1.5, 1.0)
.expect("native query execution should not fail");
let (_, query_result) = result.expect("range query failed");
let elements = matrix_values(query_result);

assert_eq!(elements.len(), 1);
assert_eq!(elements[0].labels, KeyByLabelValues::new());
assert_eq!(elements[0].samples.len(), 1);
assert_eq!(elements[0].samples[0].value, 42.0);
}

#[tokio::test(flavor = "multi_thread")]
async fn range_query_set_aggregator_merges_multiple_buckets_within_one_window() {
// Every other SetAggregator test uses window_size_ms == bucket
Expand Down
Loading
Loading