From fa7056eae70c5de667298fbcbd82ee6478efa09a Mon Sep 17 00:00:00 2001 From: discord9 Date: Fri, 6 Nov 2026 18:10:58 +0800 Subject: [PATCH] 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 --- src/promql/src/native/plan.rs | 12 +- src/query/src/promql/planner/native.rs | 85 ++++++------- src/query/src/promql/planner/test.rs | 18 +++ tests-integration/src/tests/promql_test.rs | 133 +++++++++++++++++---- 4 files changed, 179 insertions(+), 69 deletions(-) diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index 44b7180623f..6ee8fe512f1 100644 --- a/src/promql/src/native/plan.rs +++ b/src/promql/src/native/plan.rs @@ -239,7 +239,7 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Opti _ => { let kept = |tags: Vec| -> Vec { 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]) -> 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(), diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index 67aaa6e9188..f9db2da6c64 100644 --- a/src/query/src/promql/planner/native.rs +++ b/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]) -> Option< _ => { let kept = |tags: Vec| -> Vec { 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]) -> 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()], diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index b8158c28d34..d40b1c8236e 100644 --- a/src/query/src/promql/planner/test.rs +++ b/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"] }; diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index 21a57a9dfc2..779a172c975 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/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::().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::>() }) .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::>() + }; + 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();