diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index 6a4966aef93..e35712c0237 100644 --- a/src/query/src/promql/planner/native.rs +++ b/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() )); } diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index 10b83fdab77..c6b30c24821 100644 --- a/src/query/src/promql/planner/test.rs +++ b/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}")); diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index 2dc49a76064..e24f81115f1 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/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")]