feat(promql): match an empty on() and grouped cardinalities natively (#10223)

The compiler already carries an empty `on()` and both grouped cardinalities,
and the matcher kernel implements them; only the planner's narrow region-local
allowlist refused them, so those queries fell back. Admit them: the pair is
planned as one native binary over the two operands, matching above their
sources.

Admitting them exposed an output-schema bug the allowlist was hiding: both tag
walks treated an empty `on()` as "keep every tag", so the output carried
labels the kernel had dropped. They now keep only the specified labels, and a
grouped match keeps the many side's labels and copies the included ones from
the one side.

Tests: the empty-`on()` subtraction and the `group_left()` form assert native
execution with their original rows and labels, and four cases assert the
kernel's cardinality errors - duplicate right under `on()` and `group_left()`,
an implicit many-to-one under `on()`, and duplicate left under `group_right()`.
The subquery case in the same loop keeps its fallback.

Signed-off-by: discord9 <discord9@163.com>

src/promql/src/native/plan.rs

@@ -239,7 +239,7 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Opti
             _ => {
                 let kept = |tags: Vec<String>| -> Vec<String> {
                     match &matching.labels {
-                        MatchingLabels::On(names) if !names.is_empty() => tags
+                        MatchingLabels::On(names) => tags
                             .into_iter()
                             .filter(|tag| names.iter().any(|name| name.as_ref() == tag.as_str()))
                             .collect(),
@@ -247,16 +247,14 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Opti
                             .into_iter()
                             .filter(|tag| !names.iter().any(|name| name.as_ref() == tag.as_str()))
                             .collect(),
-                        MatchingLabels::On(_) => tags,
                     }
                 };
                 let left_tags = fused_output_tags(left, source_tags)?;
                 let right_tags = fused_output_tags(right, source_tags)?;
-                // `on`/`ignoring` narrow which of the many side's labels survive; a group modifier
-                // then copies the names it lists *from the one side*, with that side's values. The
-                // names are bound in the modifier and the values come from the series, so both
-                // halves are needed to say what the output carries: a `group_left(container)`
-                // keeps the left side's labels and adds `container`, which only the right side has.
+                // One-to-one matching narrows the left side's labels with `on`/`ignoring`.
+                // A group retains the many side's labels and copies the named labels from the one
+                // side: `group_left(container)` keeps the left labels and adds `container` from
+                // the right.
                 let mut tags = match &matching.cardinality {
                     Cardinality::OneToOne => kept(left_tags.clone()),
                     Cardinality::ManyToOne(_) => left_tags.clone(),

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

@@ -2620,16 +2620,12 @@ fn scan_time_filter(start: i64, end: i64, inclusive: bool, time_column: &str) ->
 }
 
 pub(crate) fn independent_scan_pair(root: &NativeNode) -> bool {
-    let (left, right, matching) = match &root.expr {
-        NativeExpr::Binary {
-            left,
-            right,
-            matching,
-        } => match (&left.expr, &right.expr) {
+    let (left, right) = match &root.expr {
+        NativeExpr::Binary { left, right, .. } => match (&left.expr, &right.expr) {
             (NativeExpr::Scalar(_), NativeExpr::Scalar(_)) => return false,
             (NativeExpr::Scalar(_), _) => return independent_scan_pair(right),
             (_, NativeExpr::Scalar(_)) => return independent_scan_pair(left),
-            _ => (left, right, matching),
+            _ => (left, right),
         },
         NativeExpr::Map { input, .. } => return independent_scan_pair(input),
         NativeExpr::Set { left, right, .. } => {
@@ -2641,14 +2637,7 @@ pub(crate) fn independent_scan_pair(root: &NativeNode) -> bool {
         _ => return false,
     };
 
-    let supported_labels = match &matching.labels {
-        MatchingLabels::On(names) => !names.is_empty(),
-        MatchingLabels::Ignoring(names) => names.is_empty(),
-    };
-    matching.cardinality == Cardinality::OneToOne
-        && supported_labels
-        && scan_operand(left)
-        && scan_operand(right)
+    scan_operand(left) && scan_operand(right)
 }
 
 fn scan_operand(node: &NativeNode) -> bool {
@@ -2913,7 +2902,7 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
                 _ => {
                     let kept = |tags: Vec<String>| -> Vec<String> {
                         match &matching.labels {
-                            MatchingLabels::On(names) if !names.is_empty() => tags
+                            MatchingLabels::On(names) => tags
                                 .into_iter()
                                 .filter(|tag| {
                                     names.iter().any(|name| name.as_ref() == tag.as_str())
@@ -2925,7 +2914,6 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
                                     !names.iter().any(|name| name.as_ref() == tag.as_str())
                                 })
                                 .collect(),
-                            MatchingLabels::On(_) => tags,
                         }
                     };
                     let left_tags = native_output_tags(left, source_tags)?;
@@ -3316,8 +3304,8 @@ mod tests {
 
     use super::{
         ArrowDataType, ArrowTimeUnit, FieldKind, METRIC_NAME, NATIVE_PROMQL_EXTENSION_KEY,
-        NativeType, collect_selectors, cut_segments, fused_output_schema, hybrid_binary_admitted,
-        independent_scan_pair, native_output_tags, native_promql_requested,
+        NativeType, collect_selectors, cut_segments, fused_output_schema, fused_output_tags,
+        hybrid_binary_admitted, independent_scan_pair, native_output_tags, native_promql_requested,
         native_raw_plan_bounds_region_reads, scan_time_filter,
     };
 
@@ -3463,6 +3451,11 @@ mod tests {
             "x == on(host) x",
             "x > bool x",
             "2 * (x != on(host) x)",
+            "x + on() x",
+            "2 * (x > on() x)",
+            "x + ignoring(host) x",
+            "x * on(host) group_left x",
+            "x * on(host) group_right x",
         ] {
             let tree = compiled_test_tree(query);
             assert_eq!(input_sources(tree.root()), vec![0, 1], "{query}");
@@ -3471,17 +3464,10 @@ mod tests {
         assert!(!independent_scan_pair(
             compiled_test_tree("sum(x) + x").root()
         ));
-        assert!(!independent_scan_pair(
-            compiled_test_tree("x + on() x").root()
-        ));
-        // A wrapper does not admit a core the pair itself refuses.
-        assert!(!independent_scan_pair(
-            compiled_test_tree("2 * (x > on() x)").root()
-        ));
     }
 
     #[test]
-    fn independent_scan_pair_admits_nested_one_to_one_matches() {
+    fn independent_scan_pair_admits_nested_matches() {
         for query in [
             "x + y + z",
             "(x - y) / on(host) z",
@@ -3491,22 +3477,18 @@ mod tests {
             "(x + y) unless on(host) z",
             "(x == on(host) y) + z",
             "x > bool y + z",
+            "(x * on(host) group_left y) + z",
+            "(x * on(host) group_right y) + z",
+            "(x + on() y) + z",
+            "(x > on() y) + z",
         ] {
             let tree = compiled_test_tree(query);
             assert_eq!(input_sources(tree.root()), vec![0, 1, 2], "{query}");
             assert!(independent_scan_pair(tree.root()), "{query}");
         }
-        for query in [
-            "x + y + sum(z)",
-            "(x * on(host) group_left y) + z",
-            "(x + on() y) + z",
-            "(x > on() y) + z",
-        ] {
-            assert!(
-                !independent_scan_pair(compiled_test_tree(query).root()),
-                "{query}"
-            );
-        }
+        assert!(!independent_scan_pair(
+            compiled_test_tree("x + y + sum(z)").root()
+        ));
     }
 
     #[test]
@@ -3524,10 +3506,8 @@ mod tests {
             native_output_tags(ignoring.root(), &sources)
         );
         let empty_on = compiled_test_tree("x + on() x");
-        assert_eq!(
-            Some(tags.clone()),
-            native_output_tags(empty_on.root(), &sources)
-        );
+        assert_eq!(Some(vec![]), native_output_tags(empty_on.root(), &sources));
+        assert_eq!(Some(vec![]), fused_output_tags(empty_on.root(), &sources));
 
         // Two tables with different tags: the left side's labels pass through a wrapper and a
         // match, and `or` adds the labels only the right side carries.
@@ -3624,6 +3604,27 @@ mod tests {
             ),
             native_output_tags(group_right.root(), &group_sources)
         );
+        for query in [
+            "x * on() group_left(workload) y",
+            "x * on() group_right(container) y",
+        ] {
+            let tree = compiled_test_tree(query);
+            let expected = if query.contains("group_left") {
+                vec!["cluster", "namespace", "pod", "container", "workload"]
+            } else {
+                vec![
+                    "cluster",
+                    "namespace",
+                    "pod",
+                    "workload",
+                    "workload_type",
+                    "container",
+                ]
+            };
+            let expected = Some(expected.into_iter().map(String::from).collect());
+            assert_eq!(expected, native_output_tags(tree.root(), &group_sources));
+            assert_eq!(expected, fused_output_tags(tree.root(), &group_sources));
+        }
         // An outer match narrows the labels the inner one already narrowed.
         let sources = [
             vec!["host".to_string(), "kind".to_string(), "zone".to_string()],

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

@@ -3434,11 +3434,23 @@ async fn native_fused_binary_keeps_independent_scan_occurrences() {
         "some_metric != some_metric",
         "some_metric < on(tag_0) some_metric",
         "some_metric >= bool on(tag_0) some_metric",
+        "some_metric + on() some_metric",
+        "some_metric - ignoring(tag_0) some_metric",
+        "some_metric * on(tag_0) group_left some_metric",
+        "some_metric * on(tag_0) group_right some_metric",
     ] {
         let plan = build_native_fused_plan(query, true).await;
         let fused = find_fused_series(&plan).expect("the plan holds a native node");
         assert_eq!(parser::parse(query).unwrap().to_string(), fused.promql());
         assert_eq!(2, fused.inputs().len(), "{query}");
+        if query.contains("on()") || query.contains("ignoring(tag_0)") {
+            assert_eq!(vec!["field_0", "timestamp"], sorted_column_names(&plan));
+        } else if query.contains("group_left") || query.contains("group_right") {
+            assert_eq!(
+                vec!["field_0", "tag_0", "timestamp"],
+                sorted_column_names(&plan)
+            );
+        }
         let scans = fused
             .inputs()
             .iter()
@@ -3553,6 +3565,10 @@ async fn native_fused_cross_table_pair_reads_each_table_through_its_own_source()
         "some_metric > on(tag_0) other_metric",
         "some_metric == bool other_metric",
         "(some_metric <= on(tag_0) other_metric) * 100",
+        "some_metric + on() other_metric",
+        "some_metric - ignoring(tag_0) other_metric",
+        "some_metric * on(tag_0) group_left other_metric",
+        "some_metric * on(tag_0) group_right other_metric",
     ] {
         let plan = build_native_fused_plan_with_provider(
             query,
@@ -3630,6 +3646,8 @@ async fn native_fused_cross_table_pair_reads_each_table_through_its_own_source()
         // `or` reports the series of both sides, so its rows say each one's own name.
         let expected = if query.contains(" or ") {
             vec!["__name__", "field_0", "tag_0", "timestamp"]
+        } else if query.contains("on()") || query.contains("ignoring(tag_0)") {
+            vec!["field_0", "timestamp"]
         } else {
             vec!["field_0", "tag_0", "timestamp"]
         };

tests-integration/src/tests/promql_test.rs

@@ -6979,16 +6979,20 @@ async fn native_promql_repeated_selector_binary_uses_independent_scans() {
     let on_empty_sql = format!(
         r#"TQL EVAL (1, 1, '1s') {table}{{host="h001"}} - on() ({table}{{host="h001"}} * 2)"#
     );
-    async fn unlabeled_rows(output: Output) -> Vec<(i64, f64, bool)> {
+    async fn unlabeled_rows(output: Output) -> Vec<(i64, f64)> {
         let batches = match output.data {
             OutputData::Stream(stream) => collect_batches(stream).await.unwrap(),
             OutputData::RecordBatches(batches) => batches,
             _ => unreachable!("TQL EVAL returns record batches"),
         };
+        let schema = batches.schema();
+        let schema = schema.arrow_schema();
+        assert_eq!(schema.fields().len(), 2, "only timestamp and value");
+        assert!(schema.field_with_name("host").is_err());
+        assert!(schema.field_with_name("__name__").is_err());
         batches
             .iter()
             .flat_map(|batch| {
-                assert!(batch.column_by_name("host").is_none());
                 let timestamps = batch
                     .column_by_name("ts")
                     .unwrap()
@@ -7003,13 +7007,7 @@ async fn native_promql_repeated_selector_binary_uses_independent_scans() {
                 assert_eq!(values.len(), 1);
                 let values = values[0].as_any().downcast_ref::<Float64Array>().unwrap();
                 (0..batch.num_rows())
-                    .map(|row| {
-                        (
-                            timestamps.value(row),
-                            values.value(row),
-                            batch.num_columns() == 2,
-                        )
-                    })
+                    .map(|row| (timestamps.value(row), values.value(row)))
                     .collect::<Vec<_>>()
             })
             .collect()
@@ -7017,8 +7015,10 @@ async fn native_promql_repeated_selector_binary_uses_independent_scans() {
     observer.clear();
     let on_empty_rows =
         unlabeled_rows(native_tql_output(&instance, &on_empty_sql, true).await).await;
-    assert!(observer.fused().is_none(), "{on_empty_sql}");
-    assert_eq!(on_empty_rows, vec![(1_000, -1_001.0, true)]);
+    let on_empty_plan = observer.two_input_fused();
+    assert_eq!(on_empty_plan.name(), "FusedSeriesExec", "{on_empty_sql}");
+    assert_eq!(on_empty_plan.schema().fields().len(), 2, "{on_empty_sql}");
+    assert_eq!(on_empty_rows, vec![(1_000, -1_001.0)]);
     let on_empty_legacy =
         unlabeled_rows(native_tql_output(&instance, &on_empty_sql, false).await).await;
     assert_eq!(on_empty_rows, on_empty_legacy);
@@ -7036,32 +7036,125 @@ async fn native_promql_repeated_selector_binary_uses_independent_scans() {
         native_result_rows(native_tql_output(&instance, &different_matchers, false).await).await;
     assert_eq!(different_matcher_rows, different_matcher_legacy);
 
-    // Group matching drops the full series identity, and a subquery stays on the ordinary planner path.
-    let fallback_expected = expected[..HOSTS].to_vec();
-    for (sql, expected_rows) in [
+    // Group matching executes natively; the subquery case remains on the ordinary path, unchanged.
+    let grouped_expected = expected[..HOSTS].to_vec();
+    for (index, (sql, expected_rows)) in [
         (
             format!(
                 r#"TQL EVAL (0, 0, '1s') {table}{{host!=""}} - on(host) group_left() ({table}{{host!=""}} * 2)"#
             ),
-            fallback_expected.clone(),
+            grouped_expected,
         ),
         (
             format!(r#"TQL EVAL (1, 1, '1s') rate({table}{{host="h000"}}[2s:1s])"#),
             vec![(1_000, "h000".to_string(), 0.5)],
         ),
-    ] {
+    ]
+    .into_iter()
+    .enumerate()
+    {
         observer.clear();
         let rows = native_result_rows(native_tql_output(&instance, &sql, true).await).await;
-        assert!(
-            observer.fused().is_none(),
-            "a fallback shape must not plan a fused node: {sql}"
-        );
+        if index == 0 {
+            let plan = observer.two_input_fused();
+            assert_two_source_native_plan(&observer, &plan, false);
+        } else {
+            assert!(
+                observer.fused().is_none(),
+                "a fallback shape must not plan a fused node: {sql}"
+            );
+        }
         assert_eq!(rows, expected_rows, "{sql}");
         let legacy_rows = native_result_rows(native_tql_output(&instance, &sql, false).await).await;
         assert_eq!(rows, legacy_rows, "{sql}");
     }
 }
 
+/// Empty-key grouped matches retain the many side's labels and report kernel cardinality errors.
+#[tokio::test(flavor = "multi_thread")]
+async fn native_hybrid_binary_empty_on_grouped_matching_and_errors() {
+    common_telemetry::init_default_ut_logging();
+    let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new(
+        "native_binary_empty_on_matching",
+    ))
+    .await;
+    let instance = standalone.fe_instance().clone();
+    execute_all(
+        &instance,
+        "CREATE TABLE native_empty_match (host STRING, kind STRING, zone_name STRING, \
+         ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(host, kind, zone_name)); \
+         INSERT INTO native_empty_match VALUES \
+         ('h0', 'a', '', 0, 10), ('h0', 'b', '', 0, 20), ('h0', 'c', 'z', 0, 2)",
+        QueryContext::arc(),
+    )
+    .await;
+
+    let labels = |kind: &str| {
+        [
+            ("host".to_string(), "h0".to_string()),
+            ("kind".to_string(), kind.to_string()),
+            ("zone_name".to_string(), "z".to_string()),
+        ]
+        .into_iter()
+        .collect::<std::collections::BTreeMap<_, _>>()
+    };
+    for (query, expected) in [
+        (
+            r#"native_empty_match{kind=~"a|b"} - on() group_left(zone_name) native_empty_match{kind="c"}"#,
+            vec![(0, labels("a"), 8.0), (0, labels("b"), 18.0)],
+        ),
+        (
+            r#"native_empty_match{kind="c"} - on() group_right(zone_name) native_empty_match{kind=~"a|b"}"#,
+            vec![(0, labels("a"), -8.0), (0, labels("b"), -18.0)],
+        ),
+    ] {
+        let sql = format!("TQL EVAL (0, 0, '1s') {query}");
+        observer.clear();
+        let (columns, rows) =
+            native_labelled_rows(native_tql_output(&instance, &sql, true).await).await;
+        let plan = observer.two_input_fused();
+        assert_eq!(plan.name(), "FusedSeriesExec", "{query}");
+        assert_eq!(columns, vec!["host", "kind", "ts", "val", "zone_name"]);
+        assert_eq!(rows, expected, "{query}");
+        let (_, legacy) =
+            native_labelled_rows(native_tql_output(&instance, &sql, false).await).await;
+        assert_eq!(rows, legacy, "legacy: {query}");
+    }
+
+    for (query, expected_error) in [
+        (
+            r#"native_empty_match{kind="c"} - on() native_empty_match{kind=~"a|b"}"#,
+            "found duplicate series for the match group {} on the right hand-side of the operation at 0 ms",
+        ),
+        (
+            r#"native_empty_match{kind=~"a|b"} - on() native_empty_match{kind="c"}"#,
+            "multiple matches for labels {} at 0 ms: many-to-one matching must be explicit",
+        ),
+        (
+            r#"native_empty_match{kind="c"} - on() group_left() native_empty_match{kind=~"a|b"}"#,
+            "found duplicate series for the match group {} on the right hand-side of the operation at 0 ms",
+        ),
+        (
+            r#"native_empty_match{kind=~"a|b"} - on() group_right() native_empty_match{kind="c"}"#,
+            "found duplicate series for the match group {} on the left hand-side of the operation at 0 ms",
+        ),
+    ] {
+        let sql = format!("TQL EVAL (0, 0, '1s') {query}");
+        observer.clear();
+        let result = instance
+            .do_query(&sql, native_query_context(true, 4))
+            .await
+            .remove(0);
+        let error = match result {
+            Ok(output) => collect_error(output).await,
+            Err(error) => error.to_string(),
+        };
+        let plan = observer.two_input_fused();
+        assert_eq!(plan.name(), "FusedSeriesExec", "{query}");
+        assert!(error.contains(expected_error), "{query}: {error}");
+    }
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn native_promql_rate_and_abs_rate_match_legacy_across_storage_encodings() {
     common_telemetry::init_default_ut_logging();