From 4dd015a917ec7c396f80fdc93f931b06de1f208a Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 12:31:45 -0400
Subject: [PATCH 01/10] fix(query-engine): normalize topk labels for instant
 binary arms

---
 .../src/engines/simple_engine/promql.rs       | 38 +++++++------
 .../src/tests/range_query_arithmetic_tests.rs | 54 +++++++++++++++++++
 2 files changed, 72 insertions(+), 20 deletions(-)

diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs
index 73f0094..7875c83 100644
--- a/asap-query-engine/src/engines/simple_engine/promql.rs
+++ b/asap-query-engine/src/engines/simple_engine/promql.rs
@@ -94,6 +94,13 @@ fn combine_scalar(
         .collect()
 }
 
+fn binary_matching_label_names(label_names: Vec<String>) -> Vec<String> {
+    label_names
+        .into_iter()
+        .filter(|label| label != "__name__")
+        .collect()
+}
+
 impl SimpleEngine {
     /// Aligns `end_timestamp` down to the nearest data-ingestion-interval
     /// boundary, unconditionally — mirroring SQL's `align_end_timestamp_sql`.
@@ -460,11 +467,10 @@ impl SimpleEngine {
                 // for just this arm. Accepted behavior change (#567); warn loudly
                 // so it's visible rather than silent.
                 let results = self
-                    // (true, true): safe unconditionally — both flags are
-                    // self-gated on statistic == Topk / a "k" kwarg being
-                    // present, same as the main instant-query path (see
-                    // execute_query_pipeline's doc comment).
-                    .execute_query_pipeline(&ctx, true, true)
+                    // Binary arms need Topk limiting, but must remain in the
+                    // unformatted intermediate label representation until
+                    // after the binary join.
+                    .execute_query_pipeline(&ctx, true, false)
                     .map_err(|e| {
                         warn!(
                             "Binary-expr arm for metric '{}' failed ({}) — \
@@ -475,7 +481,7 @@ impl SimpleEngine {
                         e
                     })
                     .ok()?;
-                Some((results, label_names))
+                Some((results, binary_matching_label_names(label_names)))
             }
         }
     }
@@ -1577,12 +1583,10 @@ mod topk_pipeline_tests {
         }
     }
 
-    /// A topk leaf wrapped in a binary expr (`topk(10, ...) + 0`) must still
-    /// get the same top-10 truncation and metric-name-prefixed formatting as
-    /// the bare `topk(10, ...)` query — evaluate_arm_native's leaf branch
-    /// used to hardcode (false, false) for enable_topk_limiting/formatting,
-    /// which would have returned all 15 unformatted (single-label) rows here
-    /// instead of the top 10 with the metric-name prefix.
+    /// A topk leaf wrapped in an arithmetic binary expr (`topk(10, ...) + 0`)
+    /// must still truncate to the top 10, while arithmetic output drops the
+    /// metric name. Binary-arm evaluation must not apply standalone Topk
+    /// presentation formatting before the join.
     #[test]
     fn topk_wrapped_in_binary_expr_still_truncates_and_formats() {
         let (engine, store) = build_topk_engine();
@@ -1624,16 +1628,10 @@ mod topk_pipeline_tests {
                 "results must stay sorted by count descending"
             );
         }
-        assert_eq!(
-            results[0].labels.labels,
-            vec![METRIC.to_string(), "10.0.0.15".to_string()],
-        );
+        assert_eq!(results[0].labels.labels, vec!["10.0.0.15".to_string()],);
         assert_eq!(results[0].value, 150.0);
         for element in &results {
-            assert_eq!(
-                element.labels.labels[0], METRIC,
-                "binary-expr path must still prepend the metric name (PromQL top-k formatting)",
-            );
+            assert_eq!(element.labels.labels.len(), 1);
         }
     }
 }
diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index e1bc962..6b4ec2b 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -535,6 +535,60 @@ mod tests {
         )
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn instant_vector_vector_topk_lhs_topk_rhs_joins_by_original_labels() {
+        let query = "topk(2, metric_a) + topk(2, metric_b)";
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+            &[("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+        );
+
+        let (_, qr) = engine
+            .handle_query_promql(query.to_string(), 1.0)
+            .expect("Expected result for instant topk/topk query");
+        let elements = match qr {
+            QueryResult::Vector(vector) => vector.values,
+            other => panic!("Expected instant vector result, got {other:?}"),
+        };
+
+        assert_eq!(elements.len(), 1);
+        assert_eq!(elements[0].labels.labels, vec!["host-b".to_string()]);
+        assert!((elements[0].value - 250.0).abs() < 1e-10);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn instant_vector_vector_topk_lhs_plus_plain_rhs_joins_by_original_labels() {
+        let query = "topk(2, metric_a) + sum(metric_b) by (host)";
+        let engine = build_range_topk_plus_plain_engine(
+            "topk(2, metric_a)",
+            "sum(metric_b) by (host)",
+            &[("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+            &[("host-a", 1000.0), ("host-b", 2000.0), ("host-c", 3000.0)],
+        );
+
+        let (_, qr) = engine
+            .handle_query_promql(query.to_string(), 1.0)
+            .expect("Expected result for instant topk/plain query");
+        let elements = match qr {
+            QueryResult::Vector(vector) => vector.values,
+            other => panic!("Expected instant vector result, got {other:?}"),
+        };
+
+        assert_eq!(elements.len(), 2);
+        let mut values: HashMap<String, f64> = elements
+            .into_iter()
+            .map(|element| {
+                assert_eq!(element.labels.labels.len(), 1);
+                (element.labels.labels[0].clone(), element.value)
+            })
+            .collect();
+        assert_eq!(values.remove("host-a"), Some(1100.0));
+        assert_eq!(values.remove("host-b"), Some(2050.0));
+        assert!(values.is_empty());
+    }
+
     // RED test for PR #629 review finding 1 (see
     // .design_docs/pr-629-review-findings-handoff.md): `apply_range_topk`'s
     // formatting step (`enable_topk_formatting=true`) prepends EACH arm's

From 6cf918487394ac82aed39b768704602b280ba8ee Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 12:33:36 -0400
Subject: [PATCH 02/10] fix(query-engine): normalize topk labels for range
 binary arms

---
 .../src/engines/simple_engine/promql.rs       | 16 ++---
 .../src/tests/range_query_arithmetic_tests.rs | 64 +++++++------------
 2 files changed, 30 insertions(+), 50 deletions(-)

diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs
index 7875c83..68135c8 100644
--- a/asap-query-engine/src/engines/simple_engine/promql.rs
+++ b/asap-query-engine/src/engines/simple_engine/promql.rs
@@ -675,7 +675,7 @@ impl SimpleEngine {
         let (base_context, label_names) = self.resolve_arm_leaf_context(arm_ast, end)?;
         let range_context = self.finish_range_context(base_context, start, end, step)?;
 
-        Some((range_context, label_names))
+        Some((range_context, binary_matching_label_names(label_names)))
     }
 
     /// Handles a binary arithmetic PromQL expression for range queries.
@@ -703,10 +703,10 @@ impl SimpleEngine {
 
         if let Some((scalar, vector_arm, scalar_on_left)) = detect_scalar_arm(lhs, rhs) {
             let (ctx, labels) = self.build_arm_range_context(vector_arm, start, end, step)?;
-            // (true, true): self-gated, same as instant's binary-arm call
-            // (evaluate_binary_arm) -- both flags are no-ops unless the arm's
-            // statistic is Topk.
-            let results = self.execute_range_query_pipeline(&ctx, true, true).ok()?;
+            // Binary arms need Topk limiting, but must remain in the
+            // unformatted intermediate label representation until after the
+            // arithmetic operation.
+            let results = self.execute_range_query_pipeline(&ctx, true, false).ok()?;
             let combined: Vec<RangeVectorElement> = results
                 .into_iter()
                 .map(|mut elem| {
@@ -734,12 +734,12 @@ impl SimpleEngine {
         if lhs_labels != rhs_labels {
             return None;
         }
-        // (true, true): self-gated, same rationale as the scalar-arm call above.
+        // Binary arms need Topk limiting, but not final presentation formatting.
         let lhs_results = self
-            .execute_range_query_pipeline(&lhs_ctx, true, true)
+            .execute_range_query_pipeline(&lhs_ctx, true, false)
             .ok()?;
         let rhs_results = self
-            .execute_range_query_pipeline(&rhs_ctx, true, true)
+            .execute_range_query_pipeline(&rhs_ctx, true, false)
             .ok()?;
 
         // Build lookup: label_key -> {timestamp -> value} for rhs
diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index 6b4ec2b..0e37247 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -409,25 +409,10 @@ mod tests {
         )
     }
 
-    // Documents a PRE-EXISTING bug, unrelated to PR #629, NOT one of the 4
-    // review findings being addressed here (see
-    // .design_docs/pr-629-review-findings-handoff.md, Finding 1). Confirmed
-    // via `git log -L` that `build_promql_execution_context_tail` has
-    // unconditionally prepended `"__name__"` to a Topk arm's *label names*
-    // (promql.rs, the `if statistic_to_compute == Statistic::Topk` block)
-    // since commit 9ac794c ("simple engine split by language #284"), long
-    // before #629. Because a plain (non-topk) arm never gets that prepend,
-    // `handle_binary_expr_range_promql`'s `lhs_labels != rhs_labels` guard
-    // (label *names*, checked before any join) rejects EVERY
-    // `topk(...) OP plain_metric` binary expression outright, on both the
-    // range and instant paths identically -- the query never reaches the
-    // join code at all, let alone the `apply_range_topk` formatting-mutates
-    // `elem.labels` *values* bug Finding 1 actually describes. This is
-    // asserted here only so the (surprising) current behavior is pinned;
-    // fixing it is out of scope for PR #629's review comments -- tracked as
-    // its own issue, #631, alongside the topk+topk repro below.
+    // Regression test for issue #631: Topk preserves the original grouping
+    // labels for binary matching, while arithmetic drops the metric name.
     #[tokio::test(flavor = "multi_thread")]
-    async fn test_range_vector_vector_topk_lhs_plus_plain_rhs_returns_none() {
+    async fn test_range_vector_vector_topk_lhs_plus_plain_rhs_joins_by_original_labels() {
         let query = "topk(2, metric_a) + sum(metric_b) by (host)";
         let engine = build_range_topk_plus_plain_engine(
             "topk(2, metric_a)",
@@ -436,12 +421,23 @@ mod tests {
             &[("host-a", 1000.0), ("host-b", 2000.0), ("host-c", 3000.0)],
         );
 
-        let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0);
-        assert!(
-            result.is_none(),
-            "pre-existing __name__ label-name mismatch (predates #629) should reject this \
-             query outright, got {result:?}"
-        );
+        let (_, qr) = engine
+            .handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0)
+            .expect("Expected result for range topk/plain query");
+        let elements = matrix_values(qr);
+
+        assert_eq!(elements.len(), 2);
+        let mut values: HashMap<String, f64> = elements
+            .into_iter()
+            .map(|element| {
+                assert_eq!(element.labels.labels.len(), 1);
+                assert_eq!(element.samples.len(), 1);
+                (element.labels.labels[0].clone(), element.samples[0].value)
+            })
+            .collect();
+        assert_eq!(values.remove("host-a"), Some(1100.0));
+        assert_eq!(values.remove("host-b"), Some(2050.0));
+        assert!(values.is_empty());
     }
 
     /// Builds a SimpleEngine with two independent self-keyed topk-capable
@@ -589,25 +585,9 @@ mod tests {
         assert!(values.is_empty());
     }
 
-    // RED test for PR #629 review finding 1 (see
-    // .design_docs/pr-629-review-findings-handoff.md): `apply_range_topk`'s
-    // formatting step (`enable_topk_formatting=true`) prepends EACH arm's
-    // own metric name to `elem.labels` before the vector-vector join. Two
-    // *different* topk metrics joined by a shared grouping label ("host")
-    // pass the earlier label-*names* guard (both get `"__name__"`
-    // prepended identically), so this reaches the join -- but the join
-    // then compares `["metric_a", host]`-shaped values against
-    // `["metric_b", host]`-shaped values, which never match regardless of
-    // whether the host itself is common to both topk's surviving sets.
-    // Real PromQL vector matching ignores `__name__`/joins by the shared
-    // label ("host") alone, so this should succeed wherever both topks kept
-    // that host -- the premature per-arm metric-name prepend breaks that.
-    //
-    // Tracked in #631, not fixed as part of PR #629 -- ignored so this RED
-    // repro doesn't fail this PR's test suite. Real Prometheus semantics for
-    // this query shape haven't been confirmed yet either.
+    // Regression test for issue #631: different Topk metric names must not
+    // become part of the intermediate vector-matching identity.
     #[tokio::test(flavor = "multi_thread")]
-    #[ignore = "tracked in #631, not part of PR #629's scope"]
     async fn test_range_vector_vector_topk_lhs_topk_rhs() {
         // topk(2, metric_a): host-a=100, host-b=50 survive; host-c=10 dropped.
         // topk(2, metric_b): host-b=200, host-c=300 survive; host-a=5 dropped.

From 59154a0c28260b473539cc2e51f66b6a4ca0a898 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 12:34:40 -0400
Subject: [PATCH 03/10] test(query-engine): cover range topk binary arithmetic
 operators

---
 .../src/tests/range_query_arithmetic_tests.rs | 32 +++++++++++++++++++
 1 file changed, 32 insertions(+)

diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index 0e37247..ba99801 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -613,4 +613,36 @@ mod tests {
         assert_eq!(elements[0].samples.len(), 1);
         assert!((elements[0].samples[0].value - 250.0).abs() < 1e-10);
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn range_topk_binary_join_preserves_matching_for_all_arithmetic_ops() {
+        for (op, expected) in [
+            ("+", 250.0),
+            ("-", -150.0),
+            ("*", 10_000.0),
+            ("/", 0.25),
+            ("%", 50.0),
+        ] {
+            let engine = build_range_two_topk_engine(
+                "topk(2, metric_a)",
+                "topk(2, metric_b)",
+                &[("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+                &[("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+            );
+            let query = format!("topk(2, metric_a) {op} topk(2, metric_b)");
+            let (_, qr) = engine
+                .handle_range_query_promql(query, 1.0, 2.0, 1.0)
+                .unwrap_or_else(|| panic!("Expected result for operator {op}"));
+            let elements = matrix_values(qr);
+
+            assert_eq!(elements.len(), 1, "operator {op}");
+            assert_eq!(elements[0].labels.labels, vec!["host-b".to_string()]);
+            assert_eq!(elements[0].samples.len(), 1, "operator {op}");
+            assert!(
+                (elements[0].samples[0].value - expected).abs() < 1e-10,
+                "operator {op}: expected {expected}, got {}",
+                elements[0].samples[0].value
+            );
+        }
+    }
 }

From 2930e83bbfbf62777d18d7a31b42de75c21205b8 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 12:35:29 -0400
Subject: [PATCH 04/10] test(query-engine): cover empty topk binary joins

---
 .../src/tests/range_query_arithmetic_tests.rs | 38 +++++++++++++++++++
 1 file changed, 38 insertions(+)

diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index ba99801..d061892 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -645,4 +645,42 @@ mod tests {
             );
         }
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn instant_topk_binary_join_with_no_matches_returns_empty_vector() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-c", 300.0), ("host-d", 200.0)],
+        );
+
+        let result =
+            engine.handle_query_promql("topk(2, metric_a) + topk(2, metric_b)".to_string(), 1.0);
+        let (_, qr) = result.expect("valid no-match join should return an empty vector");
+        let elements = match qr {
+            QueryResult::Vector(vector) => vector.values,
+            other => panic!("Expected instant vector result, got {other:?}"),
+        };
+        assert!(elements.is_empty());
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn range_topk_binary_join_with_no_matches_returns_empty_matrix() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-c", 300.0), ("host-d", 200.0)],
+        );
+
+        let result = engine.handle_range_query_promql(
+            "topk(2, metric_a) + topk(2, metric_b)".to_string(),
+            1.0,
+            2.0,
+            1.0,
+        );
+        let (_, qr) = result.expect("valid no-match join should return an empty matrix");
+        assert!(matrix_values(qr).is_empty());
+    }
 }

From 3e20f0352bb2ac1c1b2edd71d71fd405b97aa264 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 12:36:16 -0400
Subject: [PATCH 05/10] test(query-engine): cover range topk scalar arithmetic

---
 .../src/tests/range_query_arithmetic_tests.rs | 21 +++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index d061892..eedfcf2 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -683,4 +683,25 @@ mod tests {
         let (_, qr) = result.expect("valid no-match join should return an empty matrix");
         assert!(matrix_values(qr).is_empty());
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn range_topk_scalar_arithmetic_drops_metric_name() {
+        let engine = build_range_topk_plus_plain_engine(
+            "topk(2, metric_a)",
+            "sum(metric_b) by (host)",
+            &[("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+            &[],
+        );
+
+        let (_, qr) = engine
+            .handle_range_query_promql("topk(2, metric_a) + 0".to_string(), 1.0, 2.0, 1.0)
+            .expect("Expected result for range topk/scalar query");
+        let elements = matrix_values(qr);
+
+        assert_eq!(elements.len(), 2);
+        for element in elements {
+            assert_eq!(element.labels.labels.len(), 1);
+            assert_eq!(element.samples.len(), 1);
+        }
+    }
 }

From c6c8bccc65e7700d583a3b2c98f45655ef89ba85 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 13:33:36 -0400
Subject: [PATCH 06/10] fix(query-engine): tighten topk binary operator
 coverage

---
 .../src/engines/simple_engine/promql.rs       | 16 ++++
 .../src/tests/range_query_arithmetic_tests.rs | 76 +++++++++++++++++--
 2 files changed, 87 insertions(+), 5 deletions(-)

diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs
index 68135c8..6504453 100644
--- a/asap-query-engine/src/engines/simple_engine/promql.rs
+++ b/asap-query-engine/src/engines/simple_engine/promql.rs
@@ -101,6 +101,11 @@ fn binary_matching_label_names(label_names: Vec<String>) -> Vec<String> {
         .collect()
 }
 
+fn is_supported_binary_arithmetic_op(op: &promql_parser::parser::token::TokenType) -> bool {
+    use promql_parser::parser::token::{T_ADD, T_DIV, T_MOD, T_MUL, T_POW, T_SUB};
+    matches!(op.id(), T_ADD | T_SUB | T_MUL | T_DIV | T_MOD | T_POW)
+}
+
 impl SimpleEngine {
     /// Aligns `end_timestamp` down to the nearest data-ingestion-interval
     /// boundary, unconditionally — mirroring SQL's `align_end_timestamp_sql`.
@@ -444,6 +449,9 @@ impl SimpleEngine {
             Expr::NumberLiteral(_) => None, // caller handles scalars
             Expr::Paren(paren) => self.evaluate_binary_arm(&paren.expr, time),
             Expr::Binary(binary) => {
+                if !is_supported_binary_arithmetic_op(&binary.op) {
+                    return None;
+                }
                 // Nested binary expression — recurse on both sides
                 let (lhs_results, lhs_labels) = self.evaluate_binary_arm(&binary.lhs, time)?;
                 let (rhs_results, rhs_labels) = self.evaluate_binary_arm(&binary.rhs, time)?;
@@ -504,6 +512,10 @@ impl SimpleEngine {
             _ => return None,
         };
 
+        if !is_supported_binary_arithmetic_op(&binary.op) {
+            return None;
+        }
+
         let lhs = binary.lhs.as_ref();
         let rhs = binary.rhs.as_ref();
         let op = &binary.op;
@@ -697,6 +709,10 @@ impl SimpleEngine {
             _ => return None,
         };
 
+        if !is_supported_binary_arithmetic_op(&binary.op) {
+            return None;
+        }
+
         let lhs = binary.lhs.as_ref();
         let rhs = binary.rhs.as_ref();
         let op = &binary.op;
diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index eedfcf2..91dcd01 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -646,6 +646,33 @@ mod tests {
         }
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn range_topk_binary_join_preserves_matching_for_power() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 2.0), ("host-b", 1.0)],
+            &[("host-a", 3.0), ("host-b", 1.0)],
+        );
+
+        let (_, qr) = engine
+            .handle_range_query_promql(
+                "topk(2, metric_a) ^ topk(2, metric_b)".to_string(),
+                1.0,
+                2.0,
+                1.0,
+            )
+            .expect("Expected result for power operator");
+        let elements = matrix_values(qr);
+        let values: HashMap<String, f64> = elements
+            .into_iter()
+            .map(|element| (element.labels.labels[0].clone(), element.samples[0].value))
+            .collect();
+
+        assert_eq!(values.get("host-a"), Some(&8.0));
+        assert_eq!(values.get("host-b"), Some(&1.0));
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn instant_topk_binary_join_with_no_matches_returns_empty_vector() {
         let engine = build_range_two_topk_engine(
@@ -694,14 +721,53 @@ mod tests {
         );
 
         let (_, qr) = engine
-            .handle_range_query_promql("topk(2, metric_a) + 0".to_string(), 1.0, 2.0, 1.0)
+            .handle_range_query_promql("topk(2, metric_a) + 2".to_string(), 1.0, 2.0, 1.0)
             .expect("Expected result for range topk/scalar query");
         let elements = matrix_values(qr);
 
         assert_eq!(elements.len(), 2);
-        for element in elements {
-            assert_eq!(element.labels.labels.len(), 1);
-            assert_eq!(element.samples.len(), 1);
-        }
+        let values: HashMap<String, f64> = elements
+            .into_iter()
+            .map(|element| {
+                assert_eq!(element.labels.labels.len(), 1);
+                assert_eq!(element.samples.len(), 1);
+                (element.labels.labels[0].clone(), element.samples[0].value)
+            })
+            .collect();
+        assert_eq!(values.get("host-a"), Some(&102.0));
+        assert_eq!(values.get("host-b"), Some(&52.0));
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn instant_topk_binary_comparison_falls_back_to_prometheus() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-a", 5.0), ("host-b", 200.0)],
+        );
+
+        assert!(engine
+            .handle_query_promql("topk(2, metric_a) == topk(2, metric_b)".to_string(), 1.0,)
+            .is_none());
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn range_topk_binary_comparison_falls_back_to_prometheus() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-a", 5.0), ("host-b", 200.0)],
+        );
+
+        assert!(engine
+            .handle_range_query_promql(
+                "topk(2, metric_a) == topk(2, metric_b)".to_string(),
+                1.0,
+                2.0,
+                1.0,
+            )
+            .is_none());
     }
 }

From 32d8ecf3d406f8e99213a7130696c8d58d585d94 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 13:37:55 -0400
Subject: [PATCH 07/10] fix(query-engine): address roborev topk findings

---
 .../src/engines/simple_engine/promql.rs       | 17 ++++---
 .../src/tests/range_query_arithmetic_tests.rs | 47 +++++++++++++++----
 2 files changed, 49 insertions(+), 15 deletions(-)

diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs
index 6504453..7f3237e 100644
--- a/asap-query-engine/src/engines/simple_engine/promql.rs
+++ b/asap-query-engine/src/engines/simple_engine/promql.rs
@@ -17,6 +17,8 @@ use std::collections::HashMap;
 use std::time::Instant;
 use tracing::{debug, warn};
 
+const METRIC_NAME_LABEL: &str = "__name__";
+
 /// Detects whether either side of a PromQL binary expression is a scalar
 /// (numeric literal), returning the scalar value, the other (vector) arm,
 /// and whether the scalar was on the left. Shared by instant and range
@@ -97,7 +99,7 @@ fn combine_scalar(
 fn binary_matching_label_names(label_names: Vec<String>) -> Vec<String> {
     label_names
         .into_iter()
-        .filter(|label| label != "__name__")
+        .filter(|label| label != METRIC_NAME_LABEL)
         .collect()
 }
 
@@ -350,7 +352,7 @@ impl SimpleEngine {
         let statistic_to_compute = requirements.statistics[0];
 
         if statistic_to_compute == Statistic::Topk {
-            let mut new_labels = vec!["__name__".to_string()];
+            let mut new_labels = vec![METRIC_NAME_LABEL.to_string()];
             new_labels.extend(query_output_labels.labels);
             query_output_labels = KeyByLabelNames::new(new_labels);
         }
@@ -422,7 +424,8 @@ impl SimpleEngine {
             other => {
                 let config = self.find_query_config_promql_structural(other)?;
                 let ctx = self.build_query_execution_context_from_ast(other, &config, time)?;
-                let label_names = ctx.metadata.query_output_labels.labels.clone();
+                let label_names =
+                    binary_matching_label_names(ctx.metadata.query_output_labels.labels.clone());
                 Some((ctx, label_names))
             }
         }
@@ -489,7 +492,7 @@ impl SimpleEngine {
                         e
                     })
                     .ok()?;
-                Some((results, binary_matching_label_names(label_names)))
+                Some((results, label_names))
             }
         }
     }
@@ -687,7 +690,7 @@ impl SimpleEngine {
         let (base_context, label_names) = self.resolve_arm_leaf_context(arm_ast, end)?;
         let range_context = self.finish_range_context(base_context, start, end, step)?;
 
-        Some((range_context, binary_matching_label_names(label_names)))
+        Some((range_context, label_names))
     }
 
     /// Handles a binary arithmetic PromQL expression for range queries.
@@ -1602,9 +1605,9 @@ mod topk_pipeline_tests {
     /// A topk leaf wrapped in an arithmetic binary expr (`topk(10, ...) + 0`)
     /// must still truncate to the top 10, while arithmetic output drops the
     /// metric name. Binary-arm evaluation must not apply standalone Topk
-    /// presentation formatting before the join.
+    /// presentation formatting before the arithmetic operation.
     #[test]
-    fn topk_wrapped_in_binary_expr_still_truncates_and_formats() {
+    fn topk_wrapped_in_binary_expr_truncates_without_metric_name() {
         let (engine, store) = build_topk_engine();
 
         let context = engine
diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index 91dcd01..b162151 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -616,18 +616,49 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn range_topk_binary_join_preserves_matching_for_all_arithmetic_ops() {
-        for (op, expected) in [
-            ("+", 250.0),
-            ("-", -150.0),
-            ("*", 10_000.0),
-            ("/", 0.25),
-            ("%", 50.0),
+        for (op, expected, candidates_a, candidates_b) in [
+            (
+                "+",
+                250.0,
+                vec![("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+                vec![("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+            ),
+            (
+                "-",
+                -150.0,
+                vec![("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+                vec![("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+            ),
+            (
+                "*",
+                10_000.0,
+                vec![("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+                vec![("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+            ),
+            (
+                "/",
+                0.25,
+                vec![("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+                vec![("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+            ),
+            (
+                "%",
+                50.0,
+                vec![("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
+                vec![("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+            ),
+            (
+                "^",
+                8.0,
+                vec![("host-a", 1.0), ("host-b", 2.0), ("host-c", 0.0)],
+                vec![("host-a", 1.0), ("host-b", 3.0), ("host-c", 2.0)],
+            ),
         ] {
             let engine = build_range_two_topk_engine(
                 "topk(2, metric_a)",
                 "topk(2, metric_b)",
-                &[("host-a", 100.0), ("host-b", 50.0), ("host-c", 10.0)],
-                &[("host-a", 5.0), ("host-b", 200.0), ("host-c", 300.0)],
+                &candidates_a,
+                &candidates_b,
             );
             let query = format!("topk(2, metric_a) {op} topk(2, metric_b)");
             let (_, qr) = engine

From 3c90a4784ffbb89d252d5c263e66b06da425236c Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 13:41:05 -0400
Subject: [PATCH 08/10] test(query-engine): cover nested topk comparison
 fallback

---
 .../src/tests/range_query_arithmetic_tests.rs   | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index b162151..59196f0 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -801,4 +801,21 @@ mod tests {
             )
             .is_none());
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn nested_topk_binary_comparison_falls_back_to_prometheus() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-a", 5.0), ("host-b", 200.0)],
+        );
+
+        assert!(engine
+            .handle_query_promql(
+                "(topk(2, metric_a) == topk(2, metric_b)) + topk(2, metric_a)".to_string(),
+                1.0,
+            )
+            .is_none());
+    }
 }

From c82f9c7a8042ae240c52bfa59c4282deb898c422 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 13:46:43 -0400
Subject: [PATCH 09/10] fix(query-engine): fall back for topk matching
 modifiers

---
 .../src/engines/simple_engine/promql.rs       |  9 +++++
 .../src/tests/range_query_arithmetic_tests.rs | 36 +++++++++++++++++++
 2 files changed, 45 insertions(+)

diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs
index 7f3237e..897d352 100644
--- a/asap-query-engine/src/engines/simple_engine/promql.rs
+++ b/asap-query-engine/src/engines/simple_engine/promql.rs
@@ -452,6 +452,9 @@ impl SimpleEngine {
             Expr::NumberLiteral(_) => None, // caller handles scalars
             Expr::Paren(paren) => self.evaluate_binary_arm(&paren.expr, time),
             Expr::Binary(binary) => {
+                if binary.modifier.is_some() {
+                    return None;
+                }
                 if !is_supported_binary_arithmetic_op(&binary.op) {
                     return None;
                 }
@@ -518,6 +521,9 @@ impl SimpleEngine {
         if !is_supported_binary_arithmetic_op(&binary.op) {
             return None;
         }
+        if binary.modifier.is_some() {
+            return None;
+        }
 
         let lhs = binary.lhs.as_ref();
         let rhs = binary.rhs.as_ref();
@@ -715,6 +721,9 @@ impl SimpleEngine {
         if !is_supported_binary_arithmetic_op(&binary.op) {
             return None;
         }
+        if binary.modifier.is_some() {
+            return None;
+        }
 
         let lhs = binary.lhs.as_ref();
         let rhs = binary.rhs.as_ref();
diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index 59196f0..59bfae0 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -818,4 +818,40 @@ mod tests {
             )
             .is_none());
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn instant_topk_binary_matching_modifier_falls_back_to_prometheus() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-a", 5.0), ("host-b", 200.0)],
+        );
+
+        assert!(engine
+            .handle_query_promql(
+                "topk(2, metric_a) + on(__name__) topk(2, metric_b)".to_string(),
+                1.0,
+            )
+            .is_none());
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn range_topk_binary_matching_modifier_falls_back_to_prometheus() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-a", 5.0), ("host-b", 200.0)],
+        );
+
+        assert!(engine
+            .handle_range_query_promql(
+                "topk(2, metric_a) + on(__name__) topk(2, metric_b)".to_string(),
+                1.0,
+                2.0,
+                1.0,
+            )
+            .is_none());
+    }
 }

From 701e3a9b3dbfcbf26dab4af3b866d5f22e6d18e9 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Thu, 27 Aug 2026 13:49:49 -0400
Subject: [PATCH 10/10] test(query-engine): cover nested topk matching fallback

---
 .../src/tests/range_query_arithmetic_tests.rs  | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
index 59bfae0..1a91294 100644
--- a/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
+++ b/asap-query-engine/src/tests/range_query_arithmetic_tests.rs
@@ -854,4 +854,22 @@ mod tests {
             )
             .is_none());
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn nested_topk_binary_matching_modifier_falls_back_to_prometheus() {
+        let engine = build_range_two_topk_engine(
+            "topk(2, metric_a)",
+            "topk(2, metric_b)",
+            &[("host-a", 100.0), ("host-b", 50.0)],
+            &[("host-a", 5.0), ("host-b", 200.0)],
+        );
+
+        assert!(engine
+            .handle_query_promql(
+                "(topk(2, metric_a) + on(__name__) topk(2, metric_b)) + topk(2, metric_a)"
+                    .to_string(),
+                1.0,
+            )
+            .is_none());
+    }
 }