Skip to content

Commit aaad01c

Browse files
fix(query-engine): normalize topk labels in binary expressions (#644)
* fix(query-engine): normalize topk labels for instant binary arms * fix(query-engine): normalize topk labels for range binary arms * test(query-engine): cover range topk binary arithmetic operators * test(query-engine): cover empty topk binary joins * test(query-engine): cover range topk scalar arithmetic * fix(query-engine): tighten topk binary operator coverage * fix(query-engine): address roborev topk findings * test(query-engine): cover nested topk comparison fallback * fix(query-engine): fall back for topk matching modifiers * test(query-engine): cover nested topk matching fallback
1 parent 4ce87d5 commit aaad01c

2 files changed

Lines changed: 390 additions & 71 deletions

File tree

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

Lines changed: 55 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ use std::collections::HashMap;
1717
use std::time::Instant;
1818
use tracing::{debug, warn};
1919

20+
const METRIC_NAME_LABEL: &str = "__name__";
21+
2022
/// Detects whether either side of a PromQL binary expression is a scalar
2123
/// (numeric literal), returning the scalar value, the other (vector) arm,
2224
/// and whether the scalar was on the left. Shared by instant and range
@@ -94,6 +96,18 @@ fn combine_scalar(
9496
.collect()
9597
}
9698

99+
fn binary_matching_label_names(label_names: Vec<String>) -> Vec<String> {
100+
label_names
101+
.into_iter()
102+
.filter(|label| label != METRIC_NAME_LABEL)
103+
.collect()
104+
}
105+
106+
fn is_supported_binary_arithmetic_op(op: &promql_parser::parser::token::TokenType) -> bool {
107+
use promql_parser::parser::token::{T_ADD, T_DIV, T_MOD, T_MUL, T_POW, T_SUB};
108+
matches!(op.id(), T_ADD | T_SUB | T_MUL | T_DIV | T_MOD | T_POW)
109+
}
110+
97111
impl SimpleEngine {
98112
/// Aligns `end_timestamp` down to the nearest data-ingestion-interval
99113
/// boundary, unconditionally — mirroring SQL's `align_end_timestamp_sql`.
@@ -338,7 +352,7 @@ impl SimpleEngine {
338352
let statistic_to_compute = requirements.statistics[0];
339353

340354
if statistic_to_compute == Statistic::Topk {
341-
let mut new_labels = vec!["__name__".to_string()];
355+
let mut new_labels = vec![METRIC_NAME_LABEL.to_string()];
342356
new_labels.extend(query_output_labels.labels);
343357
query_output_labels = KeyByLabelNames::new(new_labels);
344358
}
@@ -410,7 +424,8 @@ impl SimpleEngine {
410424
other => {
411425
let config = self.find_query_config_promql_structural(other)?;
412426
let ctx = self.build_query_execution_context_from_ast(other, &config, time)?;
413-
let label_names = ctx.metadata.query_output_labels.labels.clone();
427+
let label_names =
428+
binary_matching_label_names(ctx.metadata.query_output_labels.labels.clone());
414429
Some((ctx, label_names))
415430
}
416431
}
@@ -437,6 +452,12 @@ impl SimpleEngine {
437452
Expr::NumberLiteral(_) => None, // caller handles scalars
438453
Expr::Paren(paren) => self.evaluate_binary_arm(&paren.expr, time),
439454
Expr::Binary(binary) => {
455+
if binary.modifier.is_some() {
456+
return None;
457+
}
458+
if !is_supported_binary_arithmetic_op(&binary.op) {
459+
return None;
460+
}
440461
// Nested binary expression — recurse on both sides
441462
let (lhs_results, lhs_labels) = self.evaluate_binary_arm(&binary.lhs, time)?;
442463
let (rhs_results, rhs_labels) = self.evaluate_binary_arm(&binary.rhs, time)?;
@@ -460,11 +481,10 @@ impl SimpleEngine {
460481
// for just this arm. Accepted behavior change (#567); warn loudly
461482
// so it's visible rather than silent.
462483
let results = self
463-
// (true, true): safe unconditionally — both flags are
464-
// self-gated on statistic == Topk / a "k" kwarg being
465-
// present, same as the main instant-query path (see
466-
// execute_query_pipeline's doc comment).
467-
.execute_query_pipeline(&ctx, true, true)
484+
// Binary arms need Topk limiting, but must remain in the
485+
// unformatted intermediate label representation until
486+
// after the binary join.
487+
.execute_query_pipeline(&ctx, true, false)
468488
.map_err(|e| {
469489
warn!(
470490
"Binary-expr arm for metric '{}' failed ({}) — \
@@ -498,6 +518,13 @@ impl SimpleEngine {
498518
_ => return None,
499519
};
500520

521+
if !is_supported_binary_arithmetic_op(&binary.op) {
522+
return None;
523+
}
524+
if binary.modifier.is_some() {
525+
return None;
526+
}
527+
501528
let lhs = binary.lhs.as_ref();
502529
let rhs = binary.rhs.as_ref();
503530
let op = &binary.op;
@@ -691,16 +718,23 @@ impl SimpleEngine {
691718
_ => return None,
692719
};
693720

721+
if !is_supported_binary_arithmetic_op(&binary.op) {
722+
return None;
723+
}
724+
if binary.modifier.is_some() {
725+
return None;
726+
}
727+
694728
let lhs = binary.lhs.as_ref();
695729
let rhs = binary.rhs.as_ref();
696730
let op = &binary.op;
697731

698732
if let Some((scalar, vector_arm, scalar_on_left)) = detect_scalar_arm(lhs, rhs) {
699733
let (ctx, labels) = self.build_arm_range_context(vector_arm, start, end, step)?;
700-
// (true, true): self-gated, same as instant's binary-arm call
701-
// (evaluate_binary_arm) -- both flags are no-ops unless the arm's
702-
// statistic is Topk.
703-
let results = self.execute_range_query_pipeline(&ctx, true, true).ok()?;
734+
// Binary arms need Topk limiting, but must remain in the
735+
// unformatted intermediate label representation until after the
736+
// arithmetic operation.
737+
let results = self.execute_range_query_pipeline(&ctx, true, false).ok()?;
704738
let combined: Vec<RangeVectorElement> = results
705739
.into_iter()
706740
.map(|mut elem| {
@@ -728,12 +762,12 @@ impl SimpleEngine {
728762
if lhs_labels != rhs_labels {
729763
return None;
730764
}
731-
// (true, true): self-gated, same rationale as the scalar-arm call above.
765+
// Binary arms need Topk limiting, but not final presentation formatting.
732766
let lhs_results = self
733-
.execute_range_query_pipeline(&lhs_ctx, true, true)
767+
.execute_range_query_pipeline(&lhs_ctx, true, false)
734768
.ok()?;
735769
let rhs_results = self
736-
.execute_range_query_pipeline(&rhs_ctx, true, true)
770+
.execute_range_query_pipeline(&rhs_ctx, true, false)
737771
.ok()?;
738772

739773
// Build lookup: label_key -> {timestamp -> value} for rhs
@@ -1577,14 +1611,12 @@ mod topk_pipeline_tests {
15771611
}
15781612
}
15791613

1580-
/// A topk leaf wrapped in a binary expr (`topk(10, ...) + 0`) must still
1581-
/// get the same top-10 truncation and metric-name-prefixed formatting as
1582-
/// the bare `topk(10, ...)` query — evaluate_arm_native's leaf branch
1583-
/// used to hardcode (false, false) for enable_topk_limiting/formatting,
1584-
/// which would have returned all 15 unformatted (single-label) rows here
1585-
/// instead of the top 10 with the metric-name prefix.
1614+
/// A topk leaf wrapped in an arithmetic binary expr (`topk(10, ...) + 0`)
1615+
/// must still truncate to the top 10, while arithmetic output drops the
1616+
/// metric name. Binary-arm evaluation must not apply standalone Topk
1617+
/// presentation formatting before the arithmetic operation.
15861618
#[test]
1587-
fn topk_wrapped_in_binary_expr_still_truncates_and_formats() {
1619+
fn topk_wrapped_in_binary_expr_truncates_without_metric_name() {
15881620
let (engine, store) = build_topk_engine();
15891621

15901622
let context = engine
@@ -1624,16 +1656,10 @@ mod topk_pipeline_tests {
16241656
"results must stay sorted by count descending"
16251657
);
16261658
}
1627-
assert_eq!(
1628-
results[0].labels.labels,
1629-
vec![METRIC.to_string(), "10.0.0.15".to_string()],
1630-
);
1659+
assert_eq!(results[0].labels.labels, vec!["10.0.0.15".to_string()],);
16311660
assert_eq!(results[0].value, 150.0);
16321661
for element in &results {
1633-
assert_eq!(
1634-
element.labels.labels[0], METRIC,
1635-
"binary-expr path must still prepend the metric name (PromQL top-k formatting)",
1636-
);
1662+
assert_eq!(element.labels.labels.len(), 1);
16371663
}
16381664
}
16391665
}

0 commit comments

Comments
 (0)