From d8939788db326efd316d8b69d449914b96427c78 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 8 Jul 2026 09:20:24 -0400 Subject: [PATCH] refactor(query-engine): dedup instant/range PromQL arm-building and parsing in promql.rs --- .../src/engines/simple_engine/promql.rs | 282 +++++++++++------- asap-query-engine/src/tests/datafusion/mod.rs | 1 + .../range_query_arithmetic_tests.rs | 246 +++++++++++++++ 3 files changed, 425 insertions(+), 104 deletions(-) create mode 100644 asap-query-engine/src/tests/datafusion/range_query_arithmetic_tests.rs diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index ed98b05a..cda3623b 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -25,6 +25,22 @@ use std::collections::HashMap; use std::time::Instant; use tracing::{debug, warn}; +/// Detects whether either side of a PromQL binary expression is a scalar +/// (numeric literal), returning the scalar value, the other (vector) arm, +/// and whether the scalar was on the left. Shared by instant and range +/// binary-expression handling. +fn detect_scalar_arm<'a>( + lhs: &'a promql_parser::parser::Expr, + rhs: &'a promql_parser::parser::Expr, +) -> Option<(f64, &'a promql_parser::parser::Expr, bool)> { + use promql_parser::parser::Expr; + match (lhs, rhs) { + (_, Expr::NumberLiteral(nl)) => Some((nl.val, lhs, false)), + (Expr::NumberLiteral(nl), _) => Some((nl.val, rhs, true)), + _ => None, + } +} + impl SimpleEngine { /// Calculates start timestamp for PromQL queries fn calculate_start_timestamp_promql( @@ -346,11 +362,37 @@ 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). + /// + /// Returns `None` for `Binary` arms (caller handles recursion) and + /// `NumberLiteral` arms (caller handles scalars). + fn resolve_arm_leaf_context( + &self, + arm_ast: &promql_parser::parser::Expr, + time: f64, + ) -> Option<(QueryExecutionContext, Vec)> { + use promql_parser::parser::Expr; + + match arm_ast { + Expr::NumberLiteral(_) | Expr::Binary(_) => None, + Expr::Paren(paren) => self.resolve_arm_leaf_context(&paren.expr, time), + other => { + let config = self.find_query_config_promql_structural(other)?; + let ctx = self.build_query_execution_context_from_ast(other, &config, time)?; + let label_names = ctx.metadata.query_output_labels.labels.clone(); + Some((ctx, label_names)) + } + } + } + /// Recursively builds a DataFusion logical plan for one arm of a binary /// arithmetic expression. /// - /// - Leaf arm (supported PromQL pattern): look up config structurally, build - /// context, return its `to_logical_plan()` together with the output label names. + /// - 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). @@ -374,11 +416,8 @@ impl SimpleEngine { .ok()?; Some((combined, lhs_labels)) } - other => { - // Leaf pattern: structural config lookup + context + plan - let config = self.find_query_config_promql_structural(other)?; - let ctx = self.build_query_execution_context_from_ast(other, &config, time)?; - let label_names = ctx.metadata.query_output_labels.labels.clone(); + _ => { + let (ctx, label_names) = self.resolve_arm_leaf_context(arm_ast, time)?; let plan = ctx.to_logical_plan().ok()?; Some((plan, label_names)) } @@ -408,13 +447,7 @@ impl SimpleEngine { let rhs = binary.rhs.as_ref(); let op = &binary.op; - // Scalar case: either side may be a numeric literal - let scalar_case: Option<(f64, &Expr, bool)> = match (lhs, rhs) { - (_, Expr::NumberLiteral(nl)) => Some((nl.val, lhs, false)), - (Expr::NumberLiteral(nl), _) => Some((nl.val, rhs, true)), - _ => None, - }; - if let Some((scalar, vector_arm, scalar_on_left)) = scalar_case { + 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()) @@ -469,7 +502,13 @@ impl SimpleEngine { } } - /// Recursively builds a range execution context for one arm of a binary arithmetic expression. + /// Recursively builds a range execution context for one arm of a binary + /// arithmetic expression. + /// + /// Leaf resolution (Paren-unwrap + structural config lookup) is shared + /// with `build_arm_logical_plan` 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( &self, arm_ast: &promql_parser::parser::Expr, @@ -479,63 +518,57 @@ impl SimpleEngine { ) -> Option<(RangeQueryExecutionContext, Vec)> { use promql_parser::parser::Expr; - match arm_ast { - Expr::NumberLiteral(_) => None, // caller handles scalars - Expr::Paren(paren) => self.build_arm_range_context(&paren.expr, start, end, step), - other => { - let config = self.find_query_config_promql_structural(other)?; - let base_context = - self.build_query_execution_context_from_ast(other, &config, end)?; - let label_names = base_context.metadata.query_output_labels.labels.clone(); + if matches!(arm_ast, Expr::NumberLiteral(_)) { + return None; // caller handles scalars + } - let start_ms = Self::convert_query_time_to_data_time(start); - let end_ms = Self::convert_query_time_to_data_time(end); - let step_ms = (step * 1000.0) as u64; + let (base_context, label_names) = self.resolve_arm_leaf_context(arm_ast, end)?; - let tumbling_window_ms = self - .streaming_config - .read() - .unwrap() - .get_aggregation_config(base_context.agg_info.aggregation_id_for_value) - .map(|c| c.window_size_ms)?; - - self.validate_range_query_params(start_ms, end_ms, step_ms, tumbling_window_ms) - .map_err(|e| { - warn!("Range arm query validation failed: {}", e); - e - }) - .ok()?; + let start_ms = Self::convert_query_time_to_data_time(start); + let end_ms = Self::convert_query_time_to_data_time(end); + let step_ms = (step * 1000.0) as u64; - let lookback_ms = base_context.store_plan.values_query.end_timestamp - - base_context.store_plan.values_query.start_timestamp; - - let buckets_per_step = (step_ms / tumbling_window_ms) as usize; - let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; - - let mut extended_store_plan = base_context.store_plan.clone(); - extended_store_plan.values_query.start_timestamp = - start_ms.saturating_sub(lookback_ms); - extended_store_plan.values_query.end_timestamp = end_ms; - extended_store_plan.values_query.is_exact_query = false; - - let range_context = RangeQueryExecutionContext { - base: QueryExecutionContext { - store_plan: extended_store_plan, - ..base_context - }, - range_params: RangeQueryParams { - start: start_ms, - end: end_ms, - step: step_ms, - }, - buckets_per_step, - lookback_bucket_count, - tumbling_window_ms, - }; - - Some((range_context, label_names)) - } - } + let tumbling_window_ms = self + .streaming_config + .read() + .unwrap() + .get_aggregation_config(base_context.agg_info.aggregation_id_for_value) + .map(|c| c.window_size_ms)?; + + self.validate_range_query_params(start_ms, end_ms, step_ms, tumbling_window_ms) + .map_err(|e| { + warn!("Range arm query validation failed: {}", e); + e + }) + .ok()?; + + let lookback_ms = base_context.store_plan.values_query.end_timestamp + - base_context.store_plan.values_query.start_timestamp; + + let buckets_per_step = (step_ms / tumbling_window_ms) as usize; + let lookback_bucket_count = (lookback_ms / tumbling_window_ms) as usize; + + let mut extended_store_plan = base_context.store_plan.clone(); + extended_store_plan.values_query.start_timestamp = start_ms.saturating_sub(lookback_ms); + extended_store_plan.values_query.end_timestamp = end_ms; + extended_store_plan.values_query.is_exact_query = false; + + let range_context = RangeQueryExecutionContext { + base: QueryExecutionContext { + store_plan: extended_store_plan, + ..base_context + }, + range_params: RangeQueryParams { + start: start_ms, + end: end_ms, + step: step_ms, + }, + buckets_per_step, + lookback_bucket_count, + tumbling_window_ms, + }; + + Some((range_context, label_names)) } /// Handles a binary arithmetic PromQL expression for range queries. @@ -561,13 +594,7 @@ impl SimpleEngine { let rhs = binary.rhs.as_ref(); let op = &binary.op; - // Scalar case: either side may be a numeric literal - let scalar_case: Option<(f64, &Expr, bool)> = match (lhs, rhs) { - (_, Expr::NumberLiteral(nl)) => Some((nl.val, lhs, false)), - (Expr::NumberLiteral(nl), _) => Some((nl.val, rhs, true)), - _ => None, - }; - if let Some((scalar, vector_arm, scalar_on_left)) = scalar_case { + if let Some((scalar, vector_arm, scalar_on_left)) = detect_scalar_arm(lhs, rhs) { let (ctx, labels) = self.build_arm_range_context(vector_arm, start, end, step)?; let results = self.execute_range_query_pipeline(&ctx).ok()?; let combined: Vec = results @@ -869,21 +896,27 @@ impl SimpleEngine { let query_start_time = Instant::now(); debug!("Handling query: {} at time {}", query, time); + let ast = match promql_parser::parser::parse(&query) { + Ok(ast) => ast, + Err(e) => { + warn!("Failed to parse PromQL query '{}': {}", query, e); + return None; + } + }; + // Check for binary arithmetic before attempting single-query dispatch. // Binary expressions won't have a matching query_config, so we handle them here. - if let Ok(ast) = promql_parser::parser::parse(&query) { - if matches!(&ast, promql_parser::parser::Expr::Binary(_)) { - let result = self.handle_binary_expr_promql(&ast, time); - let total_query_duration = query_start_time.elapsed(); - debug!( - "Binary arithmetic query handling took: {:.2}ms", - total_query_duration.as_secs_f64() * 1000.0 - ); - return result; - } + if matches!(&ast, promql_parser::parser::Expr::Binary(_)) { + let result = self.handle_binary_expr_promql(&ast, time); + let total_query_duration = query_start_time.elapsed(); + debug!( + "Binary arithmetic query handling took: {:.2}ms", + total_query_duration.as_secs_f64() * 1000.0 + ); + return result; } - let context = self.build_query_execution_context_promql(query, time)?; + let context = self.build_query_execution_context_from_parsed(&ast, &query, time)?; debug!( "Querying store for metric: {}, aggregation_id: {}, range: [{}, {}]", @@ -974,8 +1007,6 @@ impl SimpleEngine { query: String, time: f64, ) -> Option { - let query_time = Self::convert_query_time_to_data_time(time); - // Parse PromQL AST using promql-parser crate let parse_start_time = Instant::now(); let ast = match promql_parser::parser::parse(&query) { @@ -993,9 +1024,23 @@ impl SimpleEngine { } }; + self.build_query_execution_context_from_parsed(&ast, &query, time) + } + + /// Variant of `build_query_execution_context_promql` that accepts an + /// already-parsed AST, avoiding a redundant re-parse when the caller + /// (e.g. `handle_query_promql`) has already parsed `query` once. + fn build_query_execution_context_from_parsed( + &self, + ast: &promql_parser::parser::Expr, + query: &str, + time: f64, + ) -> Option { + let query_time = Self::convert_query_time_to_data_time(time); + let pattern_match_start_time = Instant::now(); - let found_match = self.find_matching_controller_pattern(&ast, &query); + let found_match = self.find_matching_controller_pattern(ast, query); let (query_pattern_type, match_result) = match found_match { Some((pt, result)) => { @@ -1017,7 +1062,7 @@ impl SimpleEngine { let query_context_start_time = Instant::now(); // Resolve aggregation: try pre-configured query_configs first, fall back to capability matching. - let agg_info: AggregationIdInfo = if let Some(config) = self.find_query_config(&query) { + let agg_info: AggregationIdInfo = if let Some(config) = self.find_query_config(query) { self.get_aggregation_id_info(&config) .map_err(|e| { warn!("{}", e); @@ -1036,7 +1081,7 @@ impl SimpleEngine { _ => &empty_schema, }; let requirements = build_query_requirements_promql( - &query, + query, &match_result, query_pattern_type, metric_schema, @@ -1072,10 +1117,32 @@ impl SimpleEngine { start: f64, end: f64, step: f64, + ) -> Option { + let ast = match promql_parser::parser::parse(&query) { + Ok(ast) => ast, + Err(e) => { + warn!("Failed to parse PromQL query '{}': {}", query, e); + return None; + } + }; + + self.build_range_query_execution_context_from_parsed(&ast, &query, start, end, step) + } + + /// Variant of `build_range_query_execution_context_promql` that accepts an + /// already-parsed AST, avoiding a redundant re-parse when the caller + /// (e.g. `handle_range_query_promql`) has already parsed `query` once. + fn build_range_query_execution_context_from_parsed( + &self, + ast: &promql_parser::parser::Expr, + query: &str, + start: f64, + end: f64, + step: f64, ) -> Option { // First, build the base instant query context (reuse existing logic) // Use 'end' as the reference time for parsing - let base_context = self.build_query_execution_context_promql(query, end)?; + let base_context = self.build_query_execution_context_from_parsed(ast, query, end)?; // Convert to milliseconds let start_ms = Self::convert_query_time_to_data_time(start); @@ -1142,20 +1209,27 @@ impl SimpleEngine { query, start, end, step ); - // Check for binary arithmetic before attempting single-query dispatch. - if let Ok(ast) = promql_parser::parser::parse(&query) { - if matches!(&ast, promql_parser::parser::Expr::Binary(_)) { - let result = self.handle_binary_expr_range_promql(&ast, start, end, step); - let total_duration = query_start_time.elapsed(); - debug!( - "Binary arithmetic range query handling took: {:.2}ms", - total_duration.as_secs_f64() * 1000.0 - ); - return result; + let ast = match promql_parser::parser::parse(&query) { + Ok(ast) => ast, + Err(e) => { + warn!("Failed to parse PromQL query '{}': {}", query, e); + return None; } + }; + + // Check for binary arithmetic before attempting single-query dispatch. + if matches!(&ast, promql_parser::parser::Expr::Binary(_)) { + let result = self.handle_binary_expr_range_promql(&ast, start, end, step); + let total_duration = query_start_time.elapsed(); + debug!( + "Binary arithmetic range query handling took: {:.2}ms", + total_duration.as_secs_f64() * 1000.0 + ); + return result; } - let context = self.build_range_query_execution_context_promql(query, start, end, step)?; + let context = + self.build_range_query_execution_context_from_parsed(&ast, &query, start, end, step)?; // Execute range query pipeline let results: Vec = self diff --git a/asap-query-engine/src/tests/datafusion/mod.rs b/asap-query-engine/src/tests/datafusion/mod.rs index 0d1df2fa..143cc0a4 100644 --- a/asap-query-engine/src/tests/datafusion/mod.rs +++ b/asap-query-engine/src/tests/datafusion/mod.rs @@ -11,4 +11,5 @@ pub mod plan_execution_arithmetic_tests; pub mod plan_execution_dual_input_tests; pub mod plan_execution_temporal_tests; pub mod plan_execution_tests; +pub mod range_query_arithmetic_tests; pub mod structural_matching_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 new file mode 100644 index 00000000..a2fca523 --- /dev/null +++ b/asap-query-engine/src/tests/datafusion/range_query_arithmetic_tests.rs @@ -0,0 +1,246 @@ +//! Range-query binary arithmetic integration tests. +//! +//! Verify that binary arithmetic range queries (vector/vector and +//! scalar/vector) produce numerically correct results end-to-end through +//! `handle_range_query_promql`, which joins each arm's samples by timestamp. +//! +//! Added alongside #507 (promql.rs instant vs range dedup): no test +//! previously exercised `handle_range_query_promql`'s binary path at all. +//! Nested binary over a range (e.g. `(a+b)*c`) is intentionally not covered +//! here — that gap is tracked separately in #516. + +#[cfg(test)] +mod tests { + use crate::data_model::{ + AggregationConfig, AggregationReference, AggregationType, CleanupPolicy, InferenceConfig, + KeyByLabelValues, PrecomputedOutput, PromQLSchema, QueryConfig, QueryLanguage, + SchemaConfig, StreamingConfig, WindowType, + }; + use crate::engines::query_result::{QueryResult, RangeVectorElement}; + use crate::engines::simple_engine::SimpleEngine; + use crate::precompute_operators::sum_accumulator::SumAccumulator; + use crate::stores::simple_map_store::SimpleMapStore; + use crate::stores::Store; + use crate::AggregateCore; + use promql_utilities::data_model::KeyByLabelNames; + use std::collections::HashMap; + use std::sync::Arc; + + /// One tumbling-window bucket: (bucket end timestamp ms, label values, accumulator). + type TimeSeriesData = Vec<(u64, Option>, Box)>; + + const WINDOW_MS: u64 = 1000; + + /// Builds a SimpleEngine with two independent metrics, each carrying data + /// across multiple 1s tumbling-window buckets. Unlike + /// `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. + fn create_range_engine_two_metrics( + metric_a: &str, + data_a: TimeSeriesData, + query_a: &str, + metric_b: &str, + data_b: TimeSeriesData, + query_b: &str, + ) -> SimpleEngine { + let labels = vec!["host".to_string()]; + + let mut aggregation_configs = HashMap::new(); + for (id, metric) in [(1u64, metric_a), (2u64, metric_b)] { + aggregation_configs.insert( + id, + AggregationConfig { + aggregation_id: id, + aggregation_type: AggregationType::Sum, + aggregation_sub_type: String::new(), + parameters: HashMap::new(), + grouping_labels: KeyByLabelNames::new(labels.clone()), + aggregated_labels: KeyByLabelNames::empty(), + rollup_labels: KeyByLabelNames::empty(), + original_yaml: String::new(), + window_size_ms: WINDOW_MS, + slide_interval_ms: WINDOW_MS, + window_type: WindowType::Tumbling, + spatial_filter: String::new(), + spatial_filter_normalized: String::new(), + metric: metric.to_string(), + num_aggregates_to_retain: None, + read_count_threshold: None, + table_name: None, + value_column: None, + }, + ); + } + + let streaming_config = Arc::new(StreamingConfig { + aggregation_configs, + }); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + + for (agg_id, data) in [(1u64, data_a), (2u64, data_b)] { + for (timestamp, label_values_opt, acc) in data { + let key = label_values_opt.map(|labels| KeyByLabelValues { labels }); + let output = PrecomputedOutput::new(timestamp - WINDOW_MS, timestamp, key, agg_id); + store.insert_precomputed_output(output, acc).unwrap(); + } + } + + let promql_schema = PromQLSchema::new() + .add_metric(metric_a.to_string(), KeyByLabelNames::new(labels.clone())) + .add_metric(metric_b.to_string(), KeyByLabelNames::new(labels)); + + let inference_config = InferenceConfig { + schema: SchemaConfig::PromQL(promql_schema), + query_configs: vec![ + QueryConfig::new(query_a.to_string()) + .add_aggregation(AggregationReference::new(1, None)), + QueryConfig::new(query_b.to_string()) + .add_aggregation(AggregationReference::new(2, None)), + ], + cleanup_policy: CleanupPolicy::NoCleanup, + }; + + SimpleEngine::new( + store, + inference_config, + streaming_config, + WINDOW_MS, + QueryLanguage::promql, + ) + } + + /// Builds two 1s-window buckets for host-a with the given `(timestamp, value)` pairs. + fn host_a_series(values: [(u64, f64); 2]) -> TimeSeriesData { + values + .into_iter() + .map(|(ts, val)| { + ( + ts, + Some(vec!["host-a".to_string()]), + Box::new(SumAccumulator::with_sum(val)) as Box, + ) + }) + .collect() + } + + fn matrix_values(qr: QueryResult) -> Vec { + match qr { + QueryResult::Matrix(m) => m.values, + _ => panic!("Expected matrix (range vector) result"), + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_range_vector_vector_divide_produces_ratio_per_timestamp() { + // t=1000: errors=100, requests=200 -> 0.5 + // t=2000: errors=150, requests=300 -> 0.5 + let data_errors = host_a_series([(1000, 100.0), (2000, 150.0)]); + let data_requests = host_a_series([(1000, 200.0), (2000, 300.0)]); + let engine = create_range_engine_two_metrics( + "errors_total", + data_errors, + "sum(errors_total) by (host)", + "requests_total", + data_requests, + "sum(requests_total) by (host)", + ); + + let query = "sum(errors_total) by (host) / sum(requests_total) by (host)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("Expected result for range vector-vector query"); + let elements = matrix_values(qr); + assert_eq!(elements.len(), 1, "Expected 1 series (host-a)"); + assert_eq!(elements[0].samples.len(), 2, "Expected 2 timestamps"); + for sample in &elements[0].samples { + assert!( + (sample.value - 0.5).abs() < 1e-10, + "Expected ratio 0.5 at t={}, got {}", + sample.timestamp, + sample.value + ); + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_range_vector_vector_add() { + // t=1000: a=10, b=20 -> 30 + // t=2000: a=15, b=25 -> 40 + let data_a = host_a_series([(1000, 10.0), (2000, 15.0)]); + let data_b = host_a_series([(1000, 20.0), (2000, 25.0)]); + let engine = create_range_engine_two_metrics( + "metric_a", + data_a, + "sum(metric_a) by (host)", + "metric_b", + data_b, + "sum(metric_b) by (host)", + ); + + let query = "sum(metric_a) by (host) + sum(metric_b) by (host)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("Expected result"); + let elements = matrix_values(qr); + assert_eq!(elements.len(), 1); + let samples = &elements[0].samples; + assert_eq!(samples.len(), 2); + let by_ts: HashMap = samples.iter().map(|s| (s.timestamp, s.value)).collect(); + assert!((by_ts[&1000] - 30.0).abs() < 1e-10); + assert!((by_ts[&2000] - 40.0).abs() < 1e-10); + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_range_scalar_right_multiply() { + // metric_a * 100: t=1000: 5 -> 500, t=2000: 6 -> 600 + let data_a = host_a_series([(1000, 5.0), (2000, 6.0)]); + let engine = create_range_engine_two_metrics( + "metric_a", + data_a, + "sum(metric_a) by (host)", + // second metric not used but the helper requires it; empty data. + "dummy", + vec![], + "sum(dummy) by (host)", + ); + + let query = "sum(metric_a) by (host) * 100"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("Expected result for scalar range query"); + let elements = matrix_values(qr); + assert_eq!(elements.len(), 1); + let samples = &elements[0].samples; + assert_eq!(samples.len(), 2); + let by_ts: HashMap = samples.iter().map(|s| (s.timestamp, s.value)).collect(); + assert!((by_ts[&1000] - 500.0).abs() < 1e-10); + assert!((by_ts[&2000] - 600.0).abs() < 1e-10); + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_range_scalar_left_subtract() { + // 1 - metric_a: t=1000: 0.9 -> 0.1, t=2000: 0.75 -> 0.25 + let data_a = host_a_series([(1000, 0.9), (2000, 0.75)]); + let engine = create_range_engine_two_metrics( + "metric_a", + data_a, + "sum(metric_a) by (host)", + "dummy", + vec![], + "sum(dummy) by (host)", + ); + + let query = "1 - sum(metric_a) by (host)"; + let result = engine.handle_range_query_promql(query.to_string(), 1.0, 2.0, 1.0); + let (_, qr) = result.expect("Expected result for scalar-left range query"); + let elements = matrix_values(qr); + assert_eq!(elements.len(), 1); + let samples = &elements[0].samples; + assert_eq!(samples.len(), 2); + let by_ts: HashMap = samples.iter().map(|s| (s.timestamp, s.value)).collect(); + assert!((by_ts[&1000] - 0.1).abs() < 1e-10); + assert!((by_ts[&2000] - 0.25).abs() < 1e-10); + } +}