diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 4cf0fbd2..f4e4c514 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -823,8 +823,13 @@ impl SimpleEngine { /// Executes a pre-built DataFusion logical plan and returns results. /// - /// This is the shared execution kernel used by both `execute_plan` (for single-metric - /// queries) and the binary arithmetic dispatch path. + /// This was the entry point for the DataFusion-based binary arithmetic + /// dispatch path, cut over to a native implementation in #567. Unlike its + /// sibling `execute_plan` (still called by DataFusion-path tests), this + /// function has zero callers anywhere in the repo, including tests — it + /// is genuinely dead code, kept only in case the native cutover needs to + /// be reverted. + #[allow(dead_code)] pub async fn execute_logical_plan( &self, logical_plan: datafusion::logical_expr::LogicalPlan, diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index bdf615e4..8b37f843 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -36,8 +36,8 @@ fn detect_scalar_arm<'a>( } } -/// Native vector-vector combiner for one binary-expr level (instant query): -/// joins two arms' results by label key and applies `op` per matching key, +/// Vector-vector combiner for one binary-expr level (instant query): joins +/// two arms' results by label key and applies `op` per matching key, /// dropping non-matches (inner-join semantics, matching DataFusion's /// `build_binary_vector_plan`). Positional `KeyByLabelValues` equality is /// safe *once `lhs_labels == rhs_labels` is confirmed*: label names are @@ -46,7 +46,7 @@ fn detect_scalar_arm<'a>( /// two arms don't share the same label set — mirroring DataFusion's /// `build_binary_vector_plan`, which fails to resolve a join column that /// only exists on one side. -fn combine_vector_vector_native( +fn combine_vector_vector( lhs_results: Vec, lhs_labels: &[String], rhs_results: Vec, @@ -66,19 +66,19 @@ fn combine_vector_vector_native( lhs_results .into_iter() .filter_map(|lhs_elem| { - rhs_map.get(&lhs_elem.labels).map(|&rhs_val| { + rhs_map.get(&lhs_elem.labels).map(move |&rhs_val| { let value = SimpleEngine::apply_range_binary_op(op, lhs_elem.value, rhs_val); - InstantVectorElement::new(lhs_elem.labels.clone(), value) + InstantVectorElement::new(lhs_elem.labels, value) }) }) .collect(), ) } -/// Native scalar combiner for one binary-expr level (instant query): -/// applies `op(scalar, value)` or `op(value, scalar)` per `scalar_on_left` -/// to every element of the vector arm's results. -fn combine_scalar_native( +/// Scalar combiner for one binary-expr level (instant query): applies +/// `op(scalar, value)` or `op(value, scalar)` per `scalar_on_left` to every +/// element of the vector arm's results. +fn combine_scalar( vector_results: Vec, scalar: f64, op: &promql_parser::parser::token::TokenType, @@ -395,7 +395,7 @@ impl SimpleEngine { /// Recursively unwraps `Paren`, then structurally resolves a leaf PromQL /// arm (i.e. not `Binary` or `NumberLiteral`) to its `QueryConfig` and /// base `QueryExecutionContext`. Shared leaf-resolution step for both - /// `build_arm_logical_plan` (instant) and `build_arm_range_context` (range). + /// `evaluate_binary_arm` (instant) and `build_arm_range_context` (range). /// /// Returns `None` for `Binary` arms (caller handles recursion) and /// `NumberLiteral` arms (caller handles scalars). @@ -418,115 +418,17 @@ impl SimpleEngine { } } - /// Recursively builds a DataFusion logical plan for one arm of a binary - /// arithmetic expression. - /// - /// - Leaf arm (supported PromQL pattern): resolved via `resolve_arm_leaf_context`, - /// returning its `to_logical_plan()` together with the output label names. - /// - Binary arm: recursively build both sub-arms and combine with - /// `build_binary_vector_plan`. - /// - Scalar literal: returns `None` (handled by the caller separately). - fn build_arm_logical_plan( - &self, - arm_ast: &promql_parser::parser::Expr, - time: f64, - ) -> Option<(datafusion::logical_expr::LogicalPlan, Vec)> { - use crate::engines::logical::plan_builder::build_binary_vector_plan; - use promql_parser::parser::Expr; - - match arm_ast { - Expr::NumberLiteral(_) => None, // caller handles scalars - Expr::Paren(paren) => self.build_arm_logical_plan(&paren.expr, time), - Expr::Binary(binary) => { - // Nested binary expression — recurse on both sides - let (lhs_plan, lhs_labels) = self.build_arm_logical_plan(&binary.lhs, time)?; - let (rhs_plan, _) = self.build_arm_logical_plan(&binary.rhs, time)?; - let combined = - build_binary_vector_plan(lhs_plan, rhs_plan, &binary.op, lhs_labels.clone()) - .ok()?; - Some((combined, lhs_labels)) - } - _ => { - let (ctx, label_names) = self.resolve_arm_leaf_context(arm_ast, time)?; - let plan = ctx.to_logical_plan().ok()?; - Some((plan, label_names)) - } - } - } - - /// Handles a binary arithmetic PromQL expression by building a combined - /// DataFusion plan (vector–vector join or scalar projection) and executing it. - /// - /// Returns `None` if any arm is not acceleratable (caller falls back to Prometheus). - fn handle_binary_expr_promql( - &self, - ast: &promql_parser::parser::Expr, - time: f64, - ) -> Option<(KeyByLabelNames, QueryResult)> { - use crate::engines::logical::plan_builder::{build_binary_vector_plan, build_scalar_plan}; - use promql_parser::parser::Expr; - - let query_time = Self::convert_query_time_to_data_time(time); - - let binary = match ast { - Expr::Binary(b) => b, - _ => return None, - }; - - let lhs = binary.lhs.as_ref(); - let rhs = binary.rhs.as_ref(); - let op = &binary.op; - - if let Some((scalar, vector_arm, scalar_on_left)) = detect_scalar_arm(lhs, rhs) { - let (vector_plan, label_names) = self.build_arm_logical_plan(vector_arm, time)?; - let combined = - build_scalar_plan(vector_plan, scalar, op, scalar_on_left, label_names.clone()) - .ok()?; - let results = tokio::task::block_in_place(|| { - tokio::runtime::Handle::current().block_on(self.execute_logical_plan( - combined, - label_names.clone(), - "", - &Statistic::Sum, - )) - }) - .ok()?; - return Some(( - KeyByLabelNames::new(label_names), - QueryResult::vector(results, query_time), - )); - } - - // Vector–vector - let (lhs_plan, lhs_labels) = self.build_arm_logical_plan(lhs, time)?; - let (rhs_plan, _) = self.build_arm_logical_plan(rhs, time)?; - let combined = build_binary_vector_plan(lhs_plan, rhs_plan, op, lhs_labels.clone()).ok()?; - let results = tokio::task::block_in_place(|| { - tokio::runtime::Handle::current().block_on(self.execute_logical_plan( - combined, - lhs_labels.clone(), - "", - &Statistic::Sum, - )) - }) - .ok()?; - let output_labels = KeyByLabelNames::new(lhs_labels); - Some((output_labels, QueryResult::vector(results, query_time))) - } - /// Recursively evaluates one arm of a binary arithmetic expression via - /// the native pipeline (`execute_query_pipeline`), instead of building a - /// DataFusion plan. Mirrors `build_arm_logical_plan`'s shape exactly, - /// including its limitation: nested `Binary` arms are combined as - /// vector-vector only — a scalar inside a nested arm (e.g. `(a+5)*b`) - /// is not supported, matching today's DataFusion path. + /// the native pipeline (`execute_query_pipeline`). /// - /// - Leaf arm: resolved via `resolve_arm_leaf_context`, executed through - /// `execute_query_pipeline`. - /// - Binary arm: recursively evaluate both sides then combine with - /// `combine_vector_vector_native`. + /// - Leaf arm (supported PromQL pattern): resolved via `resolve_arm_leaf_context`, + /// executed through `execute_query_pipeline`. + /// - Binary arm: recursively evaluate both sub-arms and combine with + /// `combine_vector_vector`. Nested `Binary` arms are combined as + /// vector-vector only — a scalar inside a nested arm (e.g. `(a+5)*b`) + /// is not supported (tracked separately, same as before this cutover). /// - Scalar literal: returns `None` (handled by the caller separately). - fn evaluate_arm_native( + fn evaluate_binary_arm( &self, arm_ast: &promql_parser::parser::Expr, time: f64, @@ -535,12 +437,12 @@ impl SimpleEngine { match arm_ast { Expr::NumberLiteral(_) => None, // caller handles scalars - Expr::Paren(paren) => self.evaluate_arm_native(&paren.expr, time), + Expr::Paren(paren) => self.evaluate_binary_arm(&paren.expr, time), Expr::Binary(binary) => { // Nested binary expression — recurse on both sides - let (lhs_results, lhs_labels) = self.evaluate_arm_native(&binary.lhs, time)?; - let (rhs_results, rhs_labels) = self.evaluate_arm_native(&binary.rhs, time)?; - let combined = combine_vector_vector_native( + let (lhs_results, lhs_labels) = self.evaluate_binary_arm(&binary.lhs, time)?; + let (rhs_results, rhs_labels) = self.evaluate_binary_arm(&binary.rhs, time)?; + let combined = combine_vector_vector( lhs_results, &lhs_labels, rhs_results, @@ -551,9 +453,9 @@ impl SimpleEngine { } _ => { let (ctx, label_names) = self.resolve_arm_leaf_context(arm_ast, time)?; - // Unlike DataFusion's PrecomputedSummaryReadExec (which streams - // whatever rows exist, including zero, so a currently-empty arm - // still returns Some(empty vector)), execute_query_pipeline errors + // Unlike DataFusion's PrecomputedSummaryReadExec (which streamed + // whatever rows existed, including zero, so a currently-empty arm + // used to return Some(empty vector)), execute_query_pipeline errors // when the store has no precomputed outputs at all for this arm — // that propagates to None here, triggering a full Prometheus // fallback for the whole expression instead of an empty result @@ -567,10 +469,9 @@ impl SimpleEngine { .execute_query_pipeline(&ctx, true, true) .map_err(|e| { warn!( - "Native binary-expr arm for metric '{}' produced no results \ - ({}) — unlike DataFusion, this falls back to Prometheus for \ - the whole expression rather than returning an empty result \ - for just this arm", + "Binary-expr arm for metric '{}' failed ({}) — \ + falls back to Prometheus for the whole expression rather than \ + returning an empty result for just this arm", ctx.metric, e ); e @@ -581,13 +482,11 @@ impl SimpleEngine { } } - /// Native (non-DataFusion) implementation of binary-expr handling, - /// staged ahead of #567's cutover so it can be equivalence-tested - /// against the DataFusion path (`handle_binary_expr_promql`) before - /// that path is rewired to call this one. Not yet reachable from - /// `handle_query_promql` — see `handle_query_promql_native` for a - /// test-facing entrypoint. - pub fn handle_binary_expr_promql_native( + /// Handles a binary arithmetic PromQL expression via the native pipeline + /// (vector–vector join or scalar combine). + /// + /// Returns `None` if any arm is not acceleratable (caller falls back to Prometheus). + fn handle_binary_expr_promql( &self, ast: &promql_parser::parser::Expr, time: f64, @@ -606,8 +505,8 @@ impl SimpleEngine { let op = &binary.op; if let Some((scalar, vector_arm, scalar_on_left)) = detect_scalar_arm(lhs, rhs) { - let (vector_results, label_names) = self.evaluate_arm_native(vector_arm, time)?; - let combined = combine_scalar_native(vector_results, scalar, op, scalar_on_left); + let (vector_results, label_names) = self.evaluate_binary_arm(vector_arm, time)?; + let combined = combine_scalar(vector_results, scalar, op, scalar_on_left); return Some(( KeyByLabelNames::new(label_names), QueryResult::vector(combined, query_time), @@ -615,26 +514,14 @@ impl SimpleEngine { } // Vector–vector - let (lhs_results, lhs_labels) = self.evaluate_arm_native(lhs, time)?; - let (rhs_results, rhs_labels) = self.evaluate_arm_native(rhs, time)?; + let (lhs_results, lhs_labels) = self.evaluate_binary_arm(lhs, time)?; + let (rhs_results, rhs_labels) = self.evaluate_binary_arm(rhs, time)?; let combined = - combine_vector_vector_native(lhs_results, &lhs_labels, rhs_results, &rhs_labels, op)?; + combine_vector_vector(lhs_results, &lhs_labels, rhs_results, &rhs_labels, op)?; let output_labels = KeyByLabelNames::new(lhs_labels); Some((output_labels, QueryResult::vector(combined, query_time))) } - /// Parses `query` and dispatches to `handle_binary_expr_promql_native`. - /// Test-facing convenience wrapper — mirrors the binary-expr branch of - /// `handle_query_promql`, but for the native path. - pub fn handle_query_promql_native( - &self, - query: String, - time: f64, - ) -> Option<(KeyByLabelNames, QueryResult)> { - let ast = promql_parser::parser::parse(&query).ok()?; - self.handle_binary_expr_promql_native(&ast, time) - } - /// Applies a PromQL binary arithmetic operator to two f64 values. fn apply_range_binary_op( op: &promql_parser::parser::token::TokenType, @@ -657,7 +544,7 @@ impl SimpleEngine { /// arithmetic expression. /// /// Leaf resolution (Paren-unwrap + structural config lookup) is shared - /// with `build_arm_logical_plan` via `resolve_arm_leaf_context`. Note this + /// with `evaluate_binary_arm` via `resolve_arm_leaf_context`. Note this /// does not support nested `Binary` arms (e.g. `(a+b)*c` over a range) — /// tracked separately in #516. fn build_arm_range_context( @@ -764,9 +651,17 @@ impl SimpleEngine { return Some((KeyByLabelNames::new(labels), QueryResult::matrix(combined))); } - // Vector-vector: evaluate both arms, join by label key, apply op per matching timestamp + // Vector-vector: evaluate both arms, join by label key, apply op per matching timestamp. + // Reject mismatched label sets up front — same guard as the instant-query + // combine_vector_vector, and for the same reason: positional + // KeyByLabelValues equality below is only safe once the label *names* + // match (they're canonically sorted by KeyByLabelNames::new(), so two + // arms with the same label set always order their values the same way). let (lhs_ctx, lhs_labels) = self.build_arm_range_context(lhs, start, end, step)?; - let (rhs_ctx, _) = self.build_arm_range_context(rhs, start, end, step)?; + let (rhs_ctx, rhs_labels) = self.build_arm_range_context(rhs, start, end, step)?; + if lhs_labels != rhs_labels { + return None; + } let lhs_results = self.execute_range_query_pipeline(&lhs_ctx).ok()?; let rhs_results = self.execute_range_query_pipeline(&rhs_ctx).ok()?; @@ -1685,8 +1580,8 @@ mod topk_pipeline_tests { .expect("insert should succeed"); let (_, query_result) = engine - .handle_query_promql_native(format!("{TOPK_QUERY} + 0"), QUERY_TIME) - .expect("binary-expr-wrapped topk should still resolve via the native path"); + .handle_query_promql(format!("{TOPK_QUERY} + 0"), QUERY_TIME) + .expect("binary-expr-wrapped topk should still resolve"); let results = match query_result { QueryResult::Vector(iv) => iv.values, diff --git a/asap-query-engine/src/tests/datafusion/mod.rs b/asap-query-engine/src/tests/datafusion/mod.rs index 143cc0a4..fd946643 100644 --- a/asap-query-engine/src/tests/datafusion/mod.rs +++ b/asap-query-engine/src/tests/datafusion/mod.rs @@ -7,7 +7,6 @@ pub mod accumulator_serde_tests; pub mod dispatch_arithmetic_tests; pub mod plan_builder_binary_tests; pub mod plan_builder_regression_tests; -pub mod plan_execution_arithmetic_tests; pub mod plan_execution_dual_input_tests; pub mod plan_execution_temporal_tests; pub mod plan_execution_tests; diff --git a/asap-query-engine/src/tests/datafusion/range_query_arithmetic_tests.rs b/asap-query-engine/src/tests/datafusion/range_query_arithmetic_tests.rs index a2fca523..cdf67201 100644 --- a/asap-query-engine/src/tests/datafusion/range_query_arithmetic_tests.rs +++ b/asap-query-engine/src/tests/datafusion/range_query_arithmetic_tests.rs @@ -36,18 +36,22 @@ mod tests { /// `engine_factories::create_engine_two_metrics` (single timestamp, instant /// queries only), this inserts one bucket per `(timestamp, value)` pair so /// range queries have more than one output sample to join across. + #[allow(clippy::too_many_arguments)] fn create_range_engine_two_metrics( metric_a: &str, + labels_a: Vec<&str>, data_a: TimeSeriesData, query_a: &str, metric_b: &str, + labels_b: Vec<&str>, data_b: TimeSeriesData, query_b: &str, ) -> SimpleEngine { - let labels = vec!["host".to_string()]; + let labels_a: Vec = labels_a.iter().map(|s| s.to_string()).collect(); + let labels_b: Vec = labels_b.iter().map(|s| s.to_string()).collect(); let mut aggregation_configs = HashMap::new(); - for (id, metric) in [(1u64, metric_a), (2u64, metric_b)] { + for (id, metric, labels) in [(1u64, metric_a, &labels_a), (2u64, metric_b, &labels_b)] { aggregation_configs.insert( id, AggregationConfig { @@ -91,8 +95,8 @@ mod tests { } let promql_schema = PromQLSchema::new() - .add_metric(metric_a.to_string(), KeyByLabelNames::new(labels.clone())) - .add_metric(metric_b.to_string(), KeyByLabelNames::new(labels)); + .add_metric(metric_a.to_string(), KeyByLabelNames::new(labels_a)) + .add_metric(metric_b.to_string(), KeyByLabelNames::new(labels_b)); let inference_config = InferenceConfig { schema: SchemaConfig::PromQL(promql_schema), @@ -143,9 +147,11 @@ mod tests { let data_requests = host_a_series([(1000, 200.0), (2000, 300.0)]); let engine = create_range_engine_two_metrics( "errors_total", + vec!["host"], data_errors, "sum(errors_total) by (host)", "requests_total", + vec!["host"], data_requests, "sum(requests_total) by (host)", ); @@ -174,9 +180,11 @@ mod tests { let data_b = host_a_series([(1000, 20.0), (2000, 25.0)]); let engine = create_range_engine_two_metrics( "metric_a", + vec!["host"], data_a, "sum(metric_a) by (host)", "metric_b", + vec!["host"], data_b, "sum(metric_b) by (host)", ); @@ -199,10 +207,12 @@ mod tests { let data_a = host_a_series([(1000, 5.0), (2000, 6.0)]); let engine = create_range_engine_two_metrics( "metric_a", + vec!["host"], data_a, "sum(metric_a) by (host)", // second metric not used but the helper requires it; empty data. "dummy", + vec!["host"], vec![], "sum(dummy) by (host)", ); @@ -225,9 +235,11 @@ mod tests { let data_a = host_a_series([(1000, 0.9), (2000, 0.75)]); let engine = create_range_engine_two_metrics( "metric_a", + vec!["host"], data_a, "sum(metric_a) by (host)", "dummy", + vec!["host"], vec![], "sum(dummy) by (host)", ); @@ -243,4 +255,35 @@ mod tests { assert!((by_ts[&1000] - 0.1).abs() < 1e-10); assert!((by_ts[&2000] - 0.25).abs() < 1e-10); } + + // Regression test: handle_binary_expr_range_promql's vector-vector join + // used to match purely on positional KeyByLabelValues equality (rhs + // labels discarded), unlike the instant-query combine_vector_vector, + // which rejects a join between arms grouped by different label sets. Two + // arms grouped by disjoint labels ((host) vs (region)) that happen to + // produce the same value could silently join into a wrong-but-plausible + // result across the whole range. + #[tokio::test(flavor = "multi_thread")] + async fn test_range_vector_vector_mismatched_label_sets_return_none() { + let data_a = host_a_series([(1000, 10.0), (2000, 15.0)]); + let data_b = host_a_series([(1000, 10.0), (2000, 15.0)]); + let engine = create_range_engine_two_metrics( + "metric_a", + vec!["host"], + data_a, + "sum(metric_a) by (host)", + "metric_b", + vec!["region"], + data_b, + "sum(metric_b) by (region)", + ); + + let query = "sum(metric_a) by (host) + sum(metric_b) by (region)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + assert!( + result.is_none(), + "BUG: arms grouped by different label sets must not join, even when their \ + values coincide, got {result:?}" + ); + } } diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs index d5b31eec..6ac84281 100644 --- a/asap-query-engine/src/tests/mod.rs +++ b/asap-query-engine/src/tests/mod.rs @@ -3,6 +3,7 @@ pub mod clickhouse_forwarding_tests; pub mod datafusion; pub mod elastic_dsl_query_tests; pub mod elastic_forwarding_tests; +pub mod native_binary_arithmetic_plan_tests; pub mod native_binary_instant_tests; pub mod native_pipeline_merge_tests; pub mod prometheus_forwarding_tests; diff --git a/asap-query-engine/src/tests/datafusion/plan_execution_arithmetic_tests.rs b/asap-query-engine/src/tests/native_binary_arithmetic_plan_tests.rs similarity index 99% rename from asap-query-engine/src/tests/datafusion/plan_execution_arithmetic_tests.rs rename to asap-query-engine/src/tests/native_binary_arithmetic_plan_tests.rs index 91d1356d..69a119ea 100644 --- a/asap-query-engine/src/tests/datafusion/plan_execution_arithmetic_tests.rs +++ b/asap-query-engine/src/tests/native_binary_arithmetic_plan_tests.rs @@ -2,7 +2,8 @@ //! //! Verify that binary arithmetic queries (vector/vector and scalar/vector) //! produce numerically correct results when executed end-to-end through -//! `handle_binary_expr_promql` via DataFusion. +//! `handle_binary_expr_promql`, natively as of #567's Stage 3 cutover +//! (previously via DataFusion). #[cfg(test)] mod tests { diff --git a/asap-query-engine/src/tests/native_binary_instant_tests.rs b/asap-query-engine/src/tests/native_binary_instant_tests.rs index 1599d8c4..4f0bd30c 100644 --- a/asap-query-engine/src/tests/native_binary_instant_tests.rs +++ b/asap-query-engine/src/tests/native_binary_instant_tests.rs @@ -1,22 +1,21 @@ -//! Native instant binary-expr evaluator tests (issue #567, Stage 2). +//! PromQL binary-expr instant query tests, native execution (issue #567). //! -//! `handle_binary_expr_promql_native`/`evaluate_arm_native` are a new, -//! parallel implementation of PromQL binary-arithmetic instant queries, -//! built to eventually replace the DataFusion-backed -//! `handle_binary_expr_promql` (#567's Stage 3 cutover). These tests compare -//! the two paths directly — both are reachable today (`handle_query_promql` -//! for DataFusion, `handle_query_promql_native` for the new path) so -//! equivalence can be proven before anything is rewired. +//! `handle_query_promql`'s binary-arithmetic path (`handle_binary_expr_promql` +//! → `evaluate_binary_arm` → `combine_vector_vector`/`combine_scalar`) runs +//! natively as of #567's Stage 3 cutover — no more DataFusion involved. +//! These tests were originally written to compare the native path against +//! the (now-removed) DataFusion path before the cutover landed; they now +//! assert the native path's results directly. #[cfg(test)] mod tests { - use crate::data_model::{AggregationType, KeyByLabelValues}; + use crate::data_model::{AggregationType, KeyByLabelValues, WindowType}; use crate::engines::query_result::QueryResult; use crate::precompute_operators::sum_accumulator::SumAccumulator; use crate::precompute_operators::{CountMinSketchAccumulator, DeltaSetAggregatorAccumulator}; use crate::tests::test_utilities::engine_factories::{ - create_engine_dual_input, create_engine_single_pop, create_engine_three_metrics, - create_engine_two_metrics, + create_engine_dual_input, create_engine_multi_timestamp_with_window, + create_engine_single_pop, create_engine_three_metrics, create_engine_two_metrics, }; use crate::AggregateCore; @@ -39,11 +38,9 @@ mod tests { } #[tokio::test(flavor = "multi_thread")] - async fn native_vector_vector_all_ops_match_datafusion() { - // `^` excluded: DataFusion's build_binary_vector_plan maps T_POW to - // Operator::BitwiseXor "as a proxy" (plan_builder.rs comment) and - // fails to even produce a result for it today — nothing to compare - // native against for that operator. See the dedicated `^` test below. + async fn binary_expr_vector_vector_all_ops() { + // `^` excluded: needs its own accumulator setup (see the dedicated + // power-operator test below). for (op, expected) in [ ("+", 30.0), ("-", -10.0), @@ -71,33 +68,22 @@ mod tests { ); let query = format!("sum(metric_a) by (host) {op} sum(metric_b) by (host)"); - let (_, old_qr) = engine - .handle_query_promql(query.clone(), QUERY_TIME) - .unwrap_or_else(|| panic!("old path failed for op {op}")); - let (_, new_qr) = engine - .handle_query_promql_native(query.clone(), QUERY_TIME) - .unwrap_or_else(|| panic!("new path failed for op {op}")); - - let old_values = sorted(vector_values(old_qr)); - let new_values = sorted(vector_values(new_qr)); - assert_eq!(old_values.len(), 1, "op {op}"); - assert_eq!( - new_values, old_values, - "op {op}: native must match DataFusion" - ); + let (_, qr) = engine + .handle_query_promql(query, QUERY_TIME) + .unwrap_or_else(|| panic!("query failed for op {op}")); + + let values = vector_values(qr); + assert_eq!(values.len(), 1, "op {op}"); assert!( - (new_values[0].1 - expected).abs() < 1e-10, + (values[0].1 - expected).abs() < 1e-10, "op {op}: expected {expected}, got {}", - new_values[0].1 + values[0].1 ); } } #[tokio::test(flavor = "multi_thread")] - async fn native_power_operator_computes_correctly_datafusion_unsupported() { - // DataFusion's `^` support is broken today (see comment above); this - // documents that native computes it correctly via f64::powf, - // independent of a DataFusion comparison. + async fn binary_expr_power_operator_computes_correctly() { let engine = create_engine_two_metrics( "metric_a", AggregationType::Sum, @@ -118,15 +104,15 @@ mod tests { ); let query = "sum(metric_a) by (host) ^ sum(metric_b) by (host)"; - let (_, new_qr) = engine - .handle_query_promql_native(query.to_string(), QUERY_TIME) - .expect("new path failed"); - let new_values = vector_values(new_qr); - assert!((new_values[0].1 - 1024.0).abs() < 1e-6, "2^10 = 1024"); + let (_, qr) = engine + .handle_query_promql(query.to_string(), QUERY_TIME) + .expect("query failed"); + let values = vector_values(qr); + assert!((values[0].1 - 1024.0).abs() < 1e-6, "2^10 = 1024"); } #[tokio::test(flavor = "multi_thread")] - async fn native_vector_scalar_both_orderings_match_datafusion() { + async fn binary_expr_scalar_both_orderings() { let engine = create_engine_single_pop( "errors_total", AggregationType::Sum, @@ -142,22 +128,16 @@ mod tests { "sum(errors_total) by (host) * 100", "100 * sum(errors_total) by (host)", ] { - let (_, old_qr) = engine + let (_, qr) = engine .handle_query_promql(query.to_string(), QUERY_TIME) - .unwrap_or_else(|| panic!("old path failed for {query}")); - let (_, new_qr) = engine - .handle_query_promql_native(query.to_string(), QUERY_TIME) - .unwrap_or_else(|| panic!("new path failed for {query}")); - - let old_values = vector_values(old_qr); - let new_values = vector_values(new_qr); - assert_eq!(new_values, old_values, "{query}"); - assert!((new_values[0].1 - 700.0).abs() < 1e-10, "{query}"); + .unwrap_or_else(|| panic!("query failed for {query}")); + let values = vector_values(qr); + assert!((values[0].1 - 700.0).abs() < 1e-10, "{query}"); } } #[tokio::test(flavor = "multi_thread")] - async fn native_nested_binary_matches_datafusion() { + async fn binary_expr_nested_binary() { // (metric_a + metric_b) * metric_c, all Sum, host-a: (10+20)*3 = 90 let engine = create_engine_three_metrics( "metric_a", @@ -187,26 +167,21 @@ mod tests { ); let query = "(sum(metric_a) by (host) + sum(metric_b) by (host)) * sum(metric_c) by (host)"; - let (_, old_qr) = engine + let (_, qr) = engine .handle_query_promql(query.to_string(), QUERY_TIME) - .expect("old path failed"); - let (_, new_qr) = engine - .handle_query_promql_native(query.to_string(), QUERY_TIME) - .expect("new path failed"); - - let old_values = vector_values(old_qr); - let new_values = vector_values(new_qr); - assert_eq!(new_values, old_values); - assert!((new_values[0].1 - 90.0).abs() < 1e-10, "(10+20)*3 = 90"); + .expect("query failed"); + let values = vector_values(qr); + assert!((values[0].1 - 90.0).abs() < 1e-10, "(10+20)*3 = 90"); } #[tokio::test(flavor = "multi_thread")] - async fn native_binary_expr_no_data_falls_back_to_none() { + async fn binary_expr_no_data_falls_back_to_none() { // metric_a is configured (schema + pattern match) but has zero - // precomputed data. Accepted behavior change (#567): unlike - // DataFusion (which would return an empty-but-present result), - // native falls back to Prometheus (returns None) for the whole - // expression — see evaluate_arm_native's leaf branch. + // precomputed data. Accepted behavior change (#567, kept from + // DataFusion's now-removed empty-result behavior): falls back to + // Prometheus (returns None) for the whole expression — see + // evaluate_binary_arm's leaf branch, which warns loudly when this + // happens. let engine = create_engine_two_metrics( "metric_a", AggregationType::Sum, @@ -224,23 +199,17 @@ mod tests { ); let query = "sum(metric_a) by (host) + sum(metric_b) by (host)"; - let old_result = engine.handle_query_promql(query.to_string(), QUERY_TIME); + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME); assert!( - old_result.is_some(), - "DataFusion returns an empty-but-present result, not None, for a currently-empty arm" - ); - - let new_result = engine.handle_query_promql_native(query.to_string(), QUERY_TIME); - assert!( - new_result.is_none(), - "native falls back to Prometheus (None) for a currently-empty arm" + result.is_none(), + "arm with no current precomputed data falls back to Prometheus" ); } #[tokio::test(flavor = "multi_thread")] - async fn native_binary_expr_unsupported_arm_returns_none() { - // foo() is not a supported PromQL function -> arm lookup fails -> None, - // same as today's DataFusion path (dispatch_arithmetic_tests.rs). + async fn binary_expr_unsupported_arm_returns_none() { + // foo() is not a supported PromQL function -> arm lookup fails -> None + // (graceful fallback to Prometheus). let engine = create_engine_single_pop( "requests_total", AggregationType::Sum, @@ -256,18 +225,14 @@ mod tests { assert!(engine .handle_query_promql(query.to_string(), QUERY_TIME) .is_none()); - assert!(engine - .handle_query_promql_native(query.to_string(), QUERY_TIME) - .is_none()); } #[tokio::test(flavor = "multi_thread")] - async fn native_vector_vector_dual_population_matches_datafusion() { + async fn binary_expr_vector_vector_dual_population() { // event_frequency is dual-population (CountMinSketch values + - // DeltaSetAggregator keys) -- confirms the leaf swap - // (ctx.to_logical_plan() -> execute_query_pipeline) still resolves - // keys_query correctly. Wrapped in `+ 0` to route through the - // binary-expr handler at all. + // DeltaSetAggregator keys) -- confirms the leaf's keys_query + // resolution works through the binary-expr path. Wrapped in `+ 0` + // to route through the binary-expr handler at all. let cms = CountMinSketchAccumulator::new(2, 3); let mut keys = DeltaSetAggregatorAccumulator::new(); keys.add_key(KeyByLabelValues { @@ -286,34 +251,95 @@ mod tests { ); let query = "count(event_frequency) by (host, event) + 0"; - let (_, old_qr) = engine + let (_, qr) = engine .handle_query_promql(query.to_string(), QUERY_TIME) - .expect("old path failed"); - let (_, new_qr) = engine - .handle_query_promql_native(query.to_string(), QUERY_TIME) - .expect("new path failed"); + .expect("query failed"); + assert!(!sorted(vector_values(qr)).is_empty()); + } - assert_eq!(sorted(vector_values(new_qr)), sorted(vector_values(old_qr))); + #[tokio::test(flavor = "multi_thread")] + async fn binary_expr_sliding_window_end_to_end_merges_correctly() { + // Ties Stage 1's sliding-bucket merge fix (#570) to the actual + // production entrypoint this issue changes: 2 buckets for the same + // key under one Sliding exact window must both be merged, not just + // the first, when reached through a real binary-expr query. + let data = vec![ + ( + 1_000_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(10.0)) as Box, + ), + ( + 1_000_000, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(5.0)) as Box, + ), + ]; + let leaf_query = "sum_over_time(http_requests[1s])"; + let engine = create_engine_multi_timestamp_with_window( + "http_requests", + AggregationType::Sum, + vec!["host"], + data, + leaf_query, + 1_000, // window_size_ms, matches the fixed 1000ms bucket width + WindowType::Sliding, + ); + + let query = format!("{leaf_query} + 0"); + let (_, qr) = engine + .handle_query_promql(query, QUERY_TIME) + .expect("query failed"); + let values = vector_values(qr); + assert_eq!(values.len(), 1); + assert!( + (values[0].1 - 15.0).abs() < 1e-10, + "expected both sliding-window buckets merged into 15.0, got {}", + values[0].1 + ); } - // --- Regression tests: combine_vector_vector_native joins purely on - // positional KeyByLabelValues equality (raw Vec of values, no - // label names attached at all -- see key_by_label_values.rs). DataFusion's - // build_binary_vector_plan joins on named columns (lhs.{label} = - // rhs.{label}) instead, so it fails to resolve (-> None) whenever the two - // arms don't share the same label set. Native has no equivalent check, so - // it can silently join two arms grouped by *different* labels whenever - // their values happen to coincide. These tests are expected to be RED - // against the current implementation -- they exist to prove the - // divergence before combine_vector_vector_native is fixed. + #[tokio::test] + async fn binary_expr_works_on_current_thread_runtime() { + // Default (single-threaded) tokio runtime, not `flavor = "multi_thread"` + // like every other test in this file. The old DataFusion path's + // tokio::task::block_in_place(...block_on(...)) wrapper panics on a + // current-thread runtime — this test only passes because that + // wrapper is actually gone (#567 Stage 3 cutover), not because of + // any value it computes. + let engine = create_engine_single_pop( + "errors_total", + AggregationType::Sum, + vec!["host"], + vec![( + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(7.0)) as Box, + )], + "sum(errors_total) by (host)", + ); + + let result = + engine.handle_query_promql("sum(errors_total) by (host) * 2".to_string(), QUERY_TIME); + assert!(result.is_some()); + } + + // --- Regression tests: combine_vector_vector must reject a join between + // arms with different label sets rather than silently matching on + // positional KeyByLabelValues equality alone (raw Vec of values, + // no label names attached -- see key_by_label_values.rs). This mirrors + // DataFusion's build_binary_vector_plan, which fails to resolve a join + // column that only exists on one side. These tests originally compared + // against the DataFusion path to prove the divergence before the fix; + // now that combine_vector_vector checks label-set equality directly and + // DataFusion is no longer in the production path (#567 Stage 3), they + // assert the fixed behavior directly. #[tokio::test(flavor = "multi_thread")] - async fn native_mismatched_label_names_with_colliding_values_diverges_from_datafusion() { + async fn binary_expr_mismatched_label_sets_with_colliding_values_returns_none() { // metric_a grouped by (host), metric_b grouped by (region) -- disjoint - // label sets -- but both happen to produce the value "us-east". - // DataFusion can't join lhs.host against a plan with no "host" column - // -> None. Native joins on the bare value vector ["us-east"] == - // ["us-east"] -> a spurious combined result. + // label sets -- but both happen to produce the value "us-east". A + // value-only join would spuriously match ["us-east"] == ["us-east"]; + // must return None instead. let engine = create_engine_two_metrics( "metric_a", AggregationType::Sum, @@ -334,28 +360,19 @@ mod tests { ); let query = "sum(metric_a) by (host) + sum(metric_b) by (region)"; - let old_result = engine.handle_query_promql(query.to_string(), QUERY_TIME); - let new_result = engine.handle_query_promql_native(query.to_string(), QUERY_TIME); - - match old_result { - None => assert!( - new_result.is_none(), - "BUG: DataFusion refuses to join arms with different label sets (None), \ - but native silently produced {new_result:?} by matching on value alone" - ), - Some((_, old_qr)) => assert_eq!( - new_result.map(|(_, qr)| sorted(vector_values(qr))), - Some(sorted(vector_values(old_qr))), - "native and DataFusion must agree when both produce a result" - ), - } + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME); + + assert!( + result.is_none(), + "mismatched label sets must return None, not a spurious value-matched join: {result:?}" + ); } #[tokio::test(flavor = "multi_thread")] - async fn native_mismatched_label_names_multi_label_full_collision_diverges_from_datafusion() { - // Same bug with a 2-label grouping: (host, dc) vs (region, zone), but + async fn binary_expr_mismatched_label_sets_multi_label_full_collision_returns_none() { + // Same case with a 2-label grouping: (host, dc) vs (region, zone), but // the *entire* ordered value vector coincides ("us-east", "az1" on - // both sides) -- confirms the collision isn't a single-label fluke. + // both sides) -- confirms the check isn't a single-label fluke. let engine = create_engine_two_metrics( "metric_a", AggregationType::Sum, @@ -376,31 +393,20 @@ mod tests { ); let query = "sum(metric_a) by (host, dc) + sum(metric_b) by (region, zone)"; - let old_result = engine.handle_query_promql(query.to_string(), QUERY_TIME); - let new_result = engine.handle_query_promql_native(query.to_string(), QUERY_TIME); - - match old_result { - None => assert!( - new_result.is_none(), - "BUG: DataFusion refuses to join arms with different label sets (None), \ - but native silently produced {new_result:?} by matching on value alone" - ), - Some((_, old_qr)) => assert_eq!( - new_result.map(|(_, qr)| sorted(vector_values(qr))), - Some(sorted(vector_values(old_qr))), - "native and DataFusion must agree when both produce a result" - ), - } + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME); + + assert!( + result.is_none(), + "mismatched label sets must return None, not a spurious value-matched join: {result:?}" + ); } #[tokio::test(flavor = "multi_thread")] - async fn native_mismatched_label_names_non_colliding_values_still_diverges_from_datafusion() { + async fn binary_expr_mismatched_label_sets_non_colliding_values_returns_none() { // Same disjoint label sets (host vs region), but this time the values - // don't collide ("us-east" vs "eu-west"). DataFusion still can't - // resolve the join (column doesn't exist, independent of the data) - // -> None. Native's join correctly finds no match on the *value*, but - // that's incidental -- it still returns Some(empty) instead of None, - // because it never checked whether the label sets matched at all. + // don't collide ("us-east" vs "eu-west") either -- must still return + // None because the label sets themselves don't match, not because no + // values happened to match. let engine = create_engine_two_metrics( "metric_a", AggregationType::Sum, @@ -421,23 +427,19 @@ mod tests { ); let query = "sum(metric_a) by (host) + sum(metric_b) by (region)"; - let old_result = engine.handle_query_promql(query.to_string(), QUERY_TIME); - let new_result = engine.handle_query_promql_native(query.to_string(), QUERY_TIME); - - assert_eq!( - old_result.is_none(), - new_result.is_none(), - "BUG: DataFusion returns {old_result:?} (no such column to join on), native \ - returns {new_result:?} -- native never checks that the two arms' label sets match" + let result = engine.handle_query_promql(query.to_string(), QUERY_TIME); + + assert!( + result.is_none(), + "mismatched label sets must return None even when no values happen to match: {result:?}" ); } #[tokio::test(flavor = "multi_thread")] - async fn native_same_label_set_no_matching_values_both_agree_empty() { + async fn binary_expr_same_label_set_no_matching_values_returns_empty() { // Control case: identical label sets (host on both sides) but - // disjoint values -- both paths should agree there's no match. This - // isolates the bug to *differing* label sets, not "no match" in - // general. + // disjoint values -- should resolve to an empty (not None) result. + // This isolates the label-set check from ordinary "no match" cases. let engine = create_engine_two_metrics( "metric_a", AggregationType::Sum, @@ -458,15 +460,10 @@ mod tests { ); let query = "sum(metric_a) by (host) + sum(metric_b) by (host)"; - let (_, old_qr) = engine + let (_, qr) = engine .handle_query_promql(query.to_string(), QUERY_TIME) - .expect("old path failed"); - let (_, new_qr) = engine - .handle_query_promql_native(query.to_string(), QUERY_TIME) - .expect("new path failed"); - - let old_values = vector_values(old_qr); - assert_eq!(old_values, Vec::new()); - assert_eq!(sorted(vector_values(new_qr)), sorted(old_values)); + .expect("query failed"); + + assert_eq!(vector_values(qr), Vec::new()); } } diff --git a/docs/design-252-arithmetic-operators.md b/docs/design-252-arithmetic-operators.md index 541f2fc2..fc5f22d5 100644 --- a/docs/design-252-arithmetic-operators.md +++ b/docs/design-252-arithmetic-operators.md @@ -1,5 +1,14 @@ # Design: PromQL Arithmetic Operator Acceleration (Issue #252) +> **Superseded (#567):** the DataFusion execution path described below +> (`execute_plan`/`execute_logical_plan`, the `Join + Projection` plan) was +> replaced by a purely native execution path in `handle_binary_expr_promql` / +> `evaluate_binary_arm` / `combine_vector_vector` / `combine_scalar` +> (`asap-query-engine/src/engines/simple_engine/promql.rs`). The DataFusion +> code is kept only for its own dedicated tests and is no longer reachable +> from production. This doc is retained as the historical record of the +> original design decision. + ## Problem ASAPQuery accelerates PromQL queries by pre-computing sketches over streaming data and serving answers from those sketches at query time, bypassing the underlying TSDB for supported query patterns.