feat(promql): run vector-vector comparisons natively

A one-to-one comparison between two selectors, such as `a > on(host) b`
or `a == bool b`, kept the existing plan: the pair admission accepted
only arithmetic operators. The kernel already pairs comparisons by the
same matching rules and only differs in the value a pair keeps (the
left sample when the comparison holds, or 1/0 with `bool`), so the
operator check is dropped. This covers one table and two, and the
point-wise wrappers around a pair, such as `(a > b) * 100`.

Adds an end-to-end case over two tables comparing explicit rows, the
legacy answer and the native placement for a filter and a `bool`
comparison.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

src/query/src/promql/planner/native.rs

@@ -686,6 +686,9 @@ fn scan_time_filter(start: i64, end: i64, inclusive: bool, time_column: &str) ->
 /// Whether `root` is a supported one-to-one match or set operation between two independent scan
 /// occurrences, under any number of point-wise wrappers.
 ///
+/// Any arithmetic or comparison operator qualifies: they pair series by the same matching rules,
+/// and a comparison only changes which value a matched pair keeps.
+///
 /// A wrapper, a function of each value or an operation with a constant as in `100 * (1 - a / b)`,
 /// maps each series of the match on its own, so the tree still reads both scans whole exactly once.
 pub(crate) fn independent_scan_pair(root: &NativeNode) -> bool {
@@ -707,22 +710,11 @@ pub(crate) fn independent_scan_pair(root: &NativeNode) -> bool {
         _ => return false,
     };
 
-    let supported_operator = matches!(
-        matching.op,
-        promql::vector_match::BinaryOp::Add
-            | promql::vector_match::BinaryOp::Sub
-            | promql::vector_match::BinaryOp::Mul
-            | promql::vector_match::BinaryOp::Div
-            | promql::vector_match::BinaryOp::Mod
-            | promql::vector_match::BinaryOp::Pow
-            | promql::vector_match::BinaryOp::Atan2
-    );
     let supported_labels = match &matching.labels {
         MatchingLabels::On(names) => !names.is_empty(),
         MatchingLabels::Ignoring(names) => names.is_empty(),
     };
     matching.cardinality == Cardinality::OneToOne
-        && supported_operator
         && supported_labels
         && pointwise_operand(left)
         && pointwise_operand(right)
@@ -1057,6 +1049,9 @@ mod tests {
             "x + on(host) x",
             "x - (x * 2)",
             "rate(x[5s]) + rate(x[10s] offset 2s)",
+            "x == on(host) x",
+            "x > bool x",
+            "2 * (x != on(host) x)",
         ] {
             let tree = compiled_test_tree(query);
             assert_eq!(input_sources(tree.root()), vec![0, 1], "{query}");
@@ -1068,12 +1063,9 @@ mod tests {
         assert!(!independent_scan_pair(
             compiled_test_tree("x + on() x").root()
         ));
-        assert!(!independent_scan_pair(
-            compiled_test_tree("x == on(host) x").root()
-        ));
         // A wrapper does not admit a core the pair itself refuses.
         assert!(!independent_scan_pair(
-            compiled_test_tree("2 * (x == on(host) x)").root()
+            compiled_test_tree("2 * (x > on() x)").root()
         ));
     }
 

src/query/src/promql/planner/test.rs

@@ -2492,6 +2492,9 @@ async fn native_fused_binary_keeps_independent_scan_occurrences() {
         "some_metric + some_metric",
         "some_metric{tag_0=\"a\"} + on(tag_0) some_metric{tag_0=\"b\"}",
         "rate(some_metric[5m]) + on(tag_0) rate(some_metric[10m] offset 2m)",
+        "some_metric != some_metric",
+        "some_metric < on(tag_0) some_metric",
+        "some_metric >= bool on(tag_0) some_metric",
     ] {
         let plan = build_native_fused_plan(query, true).await;
         let fused = find_fused_series(&plan).expect("the plan holds a native node");
@@ -2598,6 +2601,9 @@ async fn native_fused_cross_table_pair_reads_each_table_through_its_own_source()
         "100 * (1 - some_metric{tag_0=\"a\"} / other_metric)",
         "some_metric unless on(tag_0) other_metric",
         "some_metric or other_metric",
+        "some_metric > on(tag_0) other_metric",
+        "some_metric == bool other_metric",
+        "(some_metric <= on(tag_0) other_metric) * 100",
     ] {
         let plan = build_native_fused_plan_with_provider(
             query,
@@ -2769,6 +2775,7 @@ async fn native_fused_pointwise_wrapper_around_a_pair_is_admitted() {
         "1 - some_metric / some_metric",
         "(some_metric / some_metric) * 100",
         "abs(some_metric - on(tag_0) some_metric)",
+        "(some_metric > some_metric) * 100",
     ] {
         let plan = build_native_fused_plan(query, true).await;
         let fused = find_fused_series(&plan).unwrap_or_else(|| panic!("{query}: {plan}"));

tests-integration/src/tests/promql_test.rs

@@ -3388,6 +3388,85 @@ async fn native_promql_cross_table_pairs_over_different_schemas() {
     }
 }
 
+/// A comparison between two tables runs above both regions: without `bool` it keeps the left
+/// sample where the comparison holds, with `bool` it reports 1 or 0 at every matched point, and
+/// `on(host)` narrows the labels to `host` either way.
+#[tokio::test(flavor = "multi_thread")]
+async fn native_promql_cross_table_comparisons() {
+    common_telemetry::init_default_ut_logging();
+    let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new(
+        "native_promql_cross_table_comparisons",
+    ))
+    .await;
+    let instance = standalone.fe_instance().clone();
+    execute_all(
+        &instance,
+        "CREATE TABLE native_xc_a (host STRING, kind STRING, ts TIMESTAMP TIME INDEX, \
+         val DOUBLE, PRIMARY KEY(host, kind))",
+        QueryContext::arc(),
+    )
+    .await;
+    execute_all(
+        &instance,
+        "CREATE TABLE native_xc_b (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, \
+         PRIMARY KEY(host))",
+        QueryContext::arc(),
+    )
+    .await;
+    // `a` is the second, `b` is `2 + host`: `a` passes `b` at second 3 on h0 and 4 on h1.
+    let mut a = Vec::new();
+    let mut b = Vec::new();
+    for second in 0..6 {
+        for host in 0..2 {
+            a.push(format!("('h{host}', 'x', {}, {second})", second * 1_000));
+            b.push(format!("('h{host}', {}, {})", second * 1_000, 2 + host));
+        }
+    }
+    for (table, values) in [("native_xc_a", a), ("native_xc_b", b)] {
+        execute_all(
+            &instance,
+            &format!("INSERT INTO {table} VALUES {}", values.join(", ")),
+            QueryContext::arc(),
+        )
+        .await;
+    }
+
+    let mut filtered = Vec::new();
+    let mut bools = Vec::new();
+    for host in 0..2_i64 {
+        let labels = std::collections::BTreeMap::from([("host".to_string(), format!("h{host}"))]);
+        for second in 0..6_i64 {
+            let holds = second > 2 + host;
+            if holds {
+                filtered.push((second * 1_000, labels.clone(), second as f64));
+            }
+            let equal = second == 2 + host;
+            bools.push((
+                second * 1_000,
+                labels.clone(),
+                if equal { 1.0 } else { 0.0 },
+            ));
+        }
+    }
+    for (query, expected) in [
+        ("native_xc_a > on(host) native_xc_b", filtered),
+        ("native_xc_a == bool on(host) native_xc_b", bools),
+    ] {
+        let sql = format!("TQL EVAL (0, 5, '1s') {query}");
+        observer.clear();
+        let (_, native) =
+            native_labelled_rows(native_tql_output(&instance, &sql, true).await).await;
+        let plan = observer.take();
+        assert_eq!(plan.name(), "FusedSeriesExec", "{query}");
+        assert_eq!(plan.children().len(), 2, "{query}");
+        assert_eq!(observer.take_raw_sources().len(), 2, "{query}");
+        assert_eq!(native, expected, "{query}");
+        let (_, legacy) =
+            native_labelled_rows(native_tql_output(&instance, &sql, false).await).await;
+        assert_eq!(legacy, expected, "legacy: {query}");
+    }
+}
+
 /// A two-selector query over a table split into regions keeps the existing plan: the two-scan
 /// native node has no placement there, and the query used to fail in the dist analyzer.
 #[tokio::test(flavor = "multi_thread")]