From e4a835144e1199bca8ed4568aca22696764a06df Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 10 Sep 2026 12:01:51 +0800 Subject: [PATCH] perf(promql): reuse sliding min and max candidates Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/promql/benches/bench_range_fn.rs | 148 ++++- src/promql/src/functions/aggr_over_time.rs | 538 +++++++++++++++++- .../standalone/common/tql/aggr_over_time.sql | 24 + .../promql_range_boundary/case.toml | 86 ++- 4 files changed, 776 insertions(+), 20 deletions(-) diff --git a/src/promql/benches/bench_range_fn.rs b/src/promql/benches/bench_range_fn.rs index 551dd0b0bc..1b365cc4c4 100644 --- a/src/promql/benches/bench_range_fn.rs +++ b/src/promql/benches/bench_range_fn.rs @@ -37,7 +37,8 @@ use datatypes::arrow::datatypes::{DataType, Field}; use futures::StreamExt; use promql::extension_plan::RangeManipulate; use promql::functions::{ - Changes, Delta, IDelta, Increase, PredictLinear, QuantileOverTime, Rate, Resets, SumOverTime, + Changes, Delta, IDelta, Increase, MaxOverTime, MinOverTime, PredictLinear, QuantileOverTime, + Rate, Resets, SumOverTime, }; use promql::range_array::RangeArray; @@ -228,6 +229,19 @@ fn make_edge_count_input_with_ranges( ] } +fn make_extrema_input_with_ranges(values: Vec, ranges: Vec<(u32, u32)>) -> Vec { + let timestamps = Arc::new(TimestampMillisecondArray::from_iter_values( + (0..values.len()).map(|index| index as i64 * 1_000), + )); + let values = Arc::new(Float64Array::from(values)); + let timestamp_ranges = RangeArray::from_ranges(timestamps, ranges.clone()).unwrap(); + let value_ranges = RangeArray::from_ranges(values, ranges).unwrap(); + vec![ + ColumnarValue::Array(Arc::new(timestamp_ranges.into_dict())), + ColumnarValue::Array(Arc::new(value_ranges.into_dict())), + ] +} + fn make_quantile_input(num_points: usize, window_size: u32) -> Vec { let (ts_range, val_range, _) = build_sliding_ranges(num_points, window_size, build_default_values(num_points), 0); @@ -338,6 +352,46 @@ fn assert_edge_count_output( assert_eq!(actual, expected); } +fn extrema_oracle( + values: &[f64], + ranges: &[(u32, u32)], + is_better: impl Fn(f64, f64) -> bool, +) -> Vec> { + ranges + .iter() + .map(|(offset, length)| { + let window = &values[*offset as usize..(*offset + *length) as usize]; + let mut extrema = *window.first()?; + for value in &window[1..] { + if is_better(*value, extrema) || extrema.is_nan() { + extrema = *value; + } + } + Some(extrema) + }) + .collect() +} + +fn assert_extrema_output( + udf: &datafusion::logical_expr::ScalarUDF, + prepared: &PreparedUdfCall, + expected: &[Option], +) { + let output = invoke_prepared_output(udf, prepared); + let ColumnarValue::Array(output) = output else { + panic!("extrema range UDF must return an array"); + }; + let output = output.as_any().downcast_ref::().unwrap(); + assert_eq!(output.len(), expected.len()); + for (actual, expected) in output.iter().zip(expected) { + match (actual, expected) { + (Some(actual), Some(expected)) => assert_eq!(actual.to_bits(), expected.to_bits()), + (None, None) => {} + (actual, expected) => panic!("expected {expected:?}, got {actual:?}"), + } + } +} + fn bench_range_functions(c: &mut Criterion) { let mut group = c.benchmark_group("range_fn"); @@ -552,6 +606,97 @@ fn bench_delta_rate_comparison(c: &mut Criterion) { group.finish(); } +fn bench_extrema_functions(c: &mut Criterion) { + let mut group = c.benchmark_group("extrema_fn"); + let num_points = 4_096; + let values = build_gauge_values(num_points); + let min_udf = MinOverTime::scalar_udf(); + let max_udf = MaxOverTime::scalar_udf(); + let mut backwards_ranges = (0..=num_points - 20) + .step_by(5) + .map(|offset| (offset as u32, 20)) + .collect::>(); + backwards_ranges.extend((0..=512).step_by(5).map(|offset| (offset as u32, 20))); + + // The last two controls use explicit ranges instead of a regular window/step sweep. + let cases = vec![ + ( + "w4_step1", + (0..=num_points - 4) + .map(|offset| (offset as u32, 4)) + .collect::>(), + ), + ( + "w20_step1", + (0..=num_points - 20) + .map(|offset| (offset as u32, 20)) + .collect::>(), + ), + ( + "w20_step5", + (0..=num_points - 20) + .step_by(5) + .map(|offset| (offset as u32, 20)) + .collect::>(), + ), + ( + "w240_step1", + (0..=num_points - 240) + .map(|offset| (offset as u32, 240)) + .collect::>(), + ), + ( + "w240_step240", + (0..=num_points - 240) + .step_by(240) + .map(|offset| (offset as u32, 240)) + .collect::>(), + ), + ("backwards_reset_rebuild_w20_step5", backwards_ranges), + ( + "low_coverage_full_backing_w4", + vec![ + (0, 4), + (512, 4), + (1_024, 4), + (1_536, 4), + (2_048, 4), + (2_560, 4), + (3_584, 4), + (4_092, 4), + ], + ), + ]; + let functions: [( + &str, + &datafusion::logical_expr::ScalarUDF, + fn(f64, f64) -> bool, + ); 2] = [ + ("min_over_time", &min_udf, |value, extrema| value < extrema), + ("max_over_time", &max_udf, |value, extrema| value > extrema), + ]; + + for (case_name, ranges) in cases { + let prepared = PreparedUdfCall::new(make_extrema_input_with_ranges( + values.clone(), + ranges.clone(), + )); + for (function_name, udf, is_better) in functions { + assert_extrema_output(udf, &prepared, &extrema_oracle(&values, &ranges, is_better)); + group.bench_with_input( + BenchmarkId::new( + format!("{function_name}_{case_name}"), + format!("N{num_points}"), + ), + &(), + |b, _| b.iter(|| invoke_prepared(udf, &prepared)), + ); + } + } + + group.finish(); +} + fn bench_edge_count_functions(c: &mut Criterion) { let mut group = c.benchmark_group("edge_count_fn"); let num_points = 4_096; @@ -882,5 +1027,6 @@ criterion_group!( bench_range_functions, bench_delta_rate_comparison, bench_edge_count_functions, + bench_extrema_functions, bench_range_manipulate_wall_time ); diff --git a/src/promql/src/functions/aggr_over_time.rs b/src/promql/src/functions/aggr_over_time.rs index 05b2b1b2e2..12d42823f8 100644 --- a/src/promql/src/functions/aggr_over_time.rs +++ b/src/promql/src/functions/aggr_over_time.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::VecDeque; use std::sync::Arc; use common_macro::range_fn; @@ -21,10 +22,10 @@ use datafusion::logical_expr::{ScalarUDF, Volatility}; use datafusion::physical_plan::ColumnarValue; use datatypes::arrow::array::Array; use datatypes::arrow::compute; -use datatypes::arrow::datatypes::DataType; +use datatypes::arrow::datatypes::{DataType, TimeUnit}; -use crate::functions::{compensated_sum_inc, extract_array}; -use crate::range_array::RangeArray; +use crate::functions::{compensated_sum_inc, extract_array, extract_range_dict}; +use crate::range_array::{RangeArray, unpack}; /// The average value of all points in the specified interval. #[range_fn( @@ -37,12 +38,219 @@ pub fn avg_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Op } /// The minimum value of all points in the specified interval. -#[range_fn( - name = MinOverTime, - ret = Float64Array, - display_name = prom_min_over_time -)] -pub fn min_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { +#[derive(Debug)] +pub struct MinOverTime {} + +impl MinOverTime { + pub const fn name() -> &'static str { + "prom_min_over_time" + } + + pub fn scalar_udf() -> ScalarUDF { + datafusion_expr::create_udf( + Self::name(), + Self::input_type(), + Self::return_type(), + Volatility::Volatile, + Arc::new(Self::calc) as _, + ) + } + + fn input_type() -> Vec { + min_max_input_type() + } + + fn return_type() -> DataType { + Float64Array::new_null(0).data_type().clone() + } + + fn calc(input: &[ColumnarValue]) -> Result { + min_max_over_time_batch(input, true, Self::name()) + } +} + +/// The maximum value of all points in the specified interval. +#[derive(Debug)] +pub struct MaxOverTime {} + +impl MaxOverTime { + pub const fn name() -> &'static str { + "prom_max_over_time" + } + + pub fn scalar_udf() -> ScalarUDF { + datafusion_expr::create_udf( + Self::name(), + Self::input_type(), + Self::return_type(), + Volatility::Volatile, + Arc::new(Self::calc) as _, + ) + } + + fn input_type() -> Vec { + min_max_input_type() + } + + fn return_type() -> DataType { + Float64Array::new_null(0).data_type().clone() + } + + fn calc(input: &[ColumnarValue]) -> Result { + min_max_over_time_batch(input, false, Self::name()) + } +} + +fn min_max_input_type() -> Vec { + vec![ + RangeArray::convert_data_type(DataType::Timestamp(TimeUnit::Millisecond, None)), + RangeArray::convert_data_type(DataType::Float64), + ] +} + +fn min_max_over_time_batch( + input: &[ColumnarValue], + is_min: bool, + func_name: &str, +) -> Result { + if input.len() != 2 { + return Err(DataFusionError::Execution(format!( + "{func_name}: expected 2 inputs, found {}", + input.len() + ))); + } + + let timestamps = extract_range_dict( + &input[0], + func_name, + "timestamp range vector", + &DataType::Timestamp(TimeUnit::Millisecond, None), + )?; + let values = extract_range_dict( + &input[1], + func_name, + "value range vector", + &DataType::Float64, + )?; + + let timestamp_keys = timestamps.keys().values(); + let value_keys = values.keys().values(); + if timestamp_keys.len() != value_keys.len() { + return Err(DataFusionError::Execution(format!( + "{func_name}: timestamp and value ranges should have the same number of windows, found {} and {}", + timestamp_keys.len(), + value_keys.len() + ))); + } + + let values = values + .values() + .as_any() + .downcast_ref::() + .ok_or_else(|| { + DataFusionError::Execution(format!( + "{func_name}: expect value range vector values of type Float64" + )) + })?; + let mut extrema = VecDeque::new(); + let mut latest_nan = None; + let mut previous_window = None; + let mut result = Vec::with_capacity(value_keys.len()); + + for index in 0..value_keys.len() { + let (_, timestamp_length) = unpack(timestamp_keys[index]); + let (value_offset, value_length) = unpack(value_keys[index]); + if timestamp_length != value_length { + return Err(DataFusionError::Execution(format!( + "{func_name}: timestamp and value ranges have different lengths at window {index}: {timestamp_length} and {value_length}" + ))); + } + + let start = value_offset as usize; + let end = start + value_length as usize; + let append_start = match previous_window { + Some((previous_start, previous_end)) + if start >= previous_start && end >= previous_end => + { + while extrema + .front() + .is_some_and(|&value_index| value_index < start) + { + extrema.pop_front(); + } + if latest_nan.is_some_and(|value_index| value_index < start) { + latest_nan = None; + } + previous_end.max(start) + } + Some(_) | None => { + extrema.clear(); + latest_nan = None; + start + } + }; + append_min_max_values( + values, + append_start, + end, + is_min, + &mut extrema, + &mut latest_nan, + ); + + result.push( + extrema + .front() + .map(|&value_index| values.value(value_index)) + .or_else(|| latest_nan.map(|value_index| values.value(value_index))), + ); + previous_window = Some((start, end)); + } + + Ok(ColumnarValue::Array(Arc::new(Float64Array::from_iter( + result, + )))) +} + +fn append_min_max_values( + values: &Float64Array, + start: usize, + end: usize, + is_min: bool, + extrema: &mut VecDeque, + latest_nan: &mut Option, +) { + for index in start..end { + if values.is_null(index) { + continue; + } + + let value = values.value(index); + if value.is_nan() { + // Keep the latest payload while no non-NaN value is in the window. + *latest_nan = Some(index); + continue; + } + + while let Some(&tail_index) = extrema.back() { + let tail_value = values.value(tail_index); + // Strict comparison retains the first equal value, including signed zero. + if if is_min { + value < tail_value + } else { + value > tail_value + } { + extrema.pop_back(); + } else { + break; + } + } + extrema.push_back(index); + } +} + +#[cfg(test)] +fn min_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { let mut valid_values = values.iter().flatten(); let mut min = valid_values.next()?; for value in valid_values { @@ -53,13 +261,8 @@ pub fn min_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Op Some(min) } -/// The maximum value of all points in the specified interval. -#[range_fn( - name = MaxOverTime, - ret = Float64Array, - display_name = prom_max_over_time -)] -pub fn max_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { +#[cfg(test)] +fn max_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { let mut valid_values = values.iter().flatten(); let mut max = valid_values.next()?; for value in valid_values { @@ -195,7 +398,308 @@ pub fn stddev_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> #[cfg(test)] mod test { use super::*; - use crate::functions::test_util::simple_range_udf_runner; + use crate::functions::test_util::{invoke_range_udf, simple_range_udf_runner}; + + fn assert_option_bits(actual: &[Option], expected: &[Option]) { + assert_eq!(actual.len(), expected.len()); + for (actual, expected) in actual.iter().zip(expected) { + match (actual, expected) { + (Some(actual), Some(expected)) => assert_eq!(actual.to_bits(), expected.to_bits()), + (None, None) => {} + (actual, expected) => panic!("expected {expected:?}, got {actual:?}"), + } + } + } + + fn old_min_max_over_windows( + values: &Float64Array, + ranges: &[(u32, u32)], + is_min: bool, + ) -> Vec> { + ranges + .iter() + .map(|&(offset, length)| { + let values = values.slice(offset as usize, length as usize); + let values = values.as_any().downcast_ref::().unwrap(); + let timestamps = TimestampMillisecondArray::new_null(values.len()); + if is_min { + min_over_time(×tamps, values) + } else { + max_over_time(×tamps, values) + } + }) + .collect() + } + + fn run_min_max_udf( + udf: ScalarUDF, + timestamps: RangeArray, + values: RangeArray, + ) -> Vec> { + let result = invoke_range_udf(udf, timestamps, values).unwrap(); + extract_array(&result) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect() + } + + fn slice_range_array(range: RangeArray, offset: usize, length: usize) -> RangeArray { + RangeArray::try_new(range.into_dict().slice(offset, length).to_data().into()).unwrap() + } + + fn sliced_float_values(values: Vec>) -> Float64Array { + let mut backing = vec![Some(1234.0)]; + backing.extend(values); + let length = backing.len() - 1; + let backing = Float64Array::from(backing); + backing + .slice(1, length) + .as_any() + .downcast_ref::() + .unwrap() + .clone() + } + + fn min_max_range_inputs( + timestamps: &TimestampMillisecondArray, + values: &Float64Array, + timestamp_ranges: &[(u32, u32)], + value_ranges: &[(u32, u32)], + ) -> (RangeArray, RangeArray) { + let timestamp_prefix = [(0, 1)]; + let value_prefix = [(0, 1)]; + let timestamps = RangeArray::from_ranges( + Arc::new(timestamps.clone()), + timestamp_prefix + .into_iter() + .chain(timestamp_ranges.iter().copied()), + ) + .unwrap(); + let values = RangeArray::from_ranges( + Arc::new(values.clone()), + value_prefix.into_iter().chain(value_ranges.iter().copied()), + ) + .unwrap(); + ( + slice_range_array(timestamps, 1, timestamp_ranges.len()), + slice_range_array(values, 1, value_ranges.len()), + ) + } + + fn assert_min_max_udfs_match_oracle( + timestamp_values: &TimestampMillisecondArray, + all_values: &Float64Array, + timestamp_ranges: &[(u32, u32)], + value_ranges: &[(u32, u32)], + ) { + let expected_min = old_min_max_over_windows(all_values, value_ranges, true); + let expected_max = old_min_max_over_windows(all_values, value_ranges, false); + let (timestamps, values) = + min_max_range_inputs(timestamp_values, all_values, timestamp_ranges, value_ranges); + assert_option_bits( + &run_min_max_udf(MinOverTime::scalar_udf(), timestamps, values), + &expected_min, + ); + let (timestamps, values) = + min_max_range_inputs(timestamp_values, all_values, timestamp_ranges, value_ranges); + assert_option_bits( + &run_min_max_udf(MaxOverTime::scalar_udf(), timestamps, values), + &expected_max, + ); + } + + #[test] + fn min_max_over_time_batch_preserves_bits_across_irregular_windows() { + let first_nan = f64::from_bits(0x7ff8_0000_0000_00a1); + let second_nan = f64::from_bits(0x7ff8_0000_0000_00b2); + let third_nan = f64::from_bits(0x7ff8_0000_0000_00c3); + let values = sliced_float_values(vec![ + None, + Some(first_nan), + Some(-0.0), + Some(0.0), + Some(2.0), + Some(2.0), + Some(f64::INFINITY), + Some(f64::NEG_INFINITY), + Some(second_nan), + None, + Some(3.0), + Some(third_nan), + ]); + let timestamps = TimestampMillisecondArray::from_iter((0..16).map(Some)); + let value_ranges = [ + (0, 0), + (0, 2), + (0, 4), + (1, 4), + (4, 4), + (4, 4), + (8, 2), + (5, 2), + (3, 3), + (11, 1), + ]; + let timestamp_ranges = [ + (8, 0), + (7, 2), + (6, 4), + (5, 4), + (4, 4), + (4, 4), + (3, 2), + (2, 2), + (1, 3), + (0, 1), + ]; + + assert_min_max_udfs_match_oracle(×tamps, &values, ×tamp_ranges, &value_ranges); + } + + #[test] + fn min_max_over_time_batch_preserves_first_signed_zero_in_both_orders() { + let timestamps = TimestampMillisecondArray::from(vec![0, 1, 2, 3]); + for (values, expected) in [ + (Float64Array::from(vec![-0.0, 0.0]), -0.0), + (Float64Array::from(vec![0.0, -0.0]), 0.0), + ] { + let (timestamp_ranges, value_ranges) = + min_max_range_inputs(×tamps, &values, &[(2, 2)], &[(0, 2)]); + assert_option_bits( + &run_min_max_udf(MinOverTime::scalar_udf(), timestamp_ranges, value_ranges), + &[Some(expected)], + ); + let (timestamp_ranges, value_ranges) = + min_max_range_inputs(×tamps, &values, &[(2, 2)], &[(0, 2)]); + assert_option_bits( + &run_min_max_udf(MaxOverTime::scalar_udf(), timestamp_ranges, value_ranges), + &[Some(expected)], + ); + } + } + + #[test] + fn min_max_over_time_batch_rebuilds_after_right_bound_retreat() { + let timestamps = TimestampMillisecondArray::from(vec![0, 1, 2]); + let ranges = [(0, 3), (0, 2)]; + assert_min_max_udfs_match_oracle( + ×tamps, + &Float64Array::from(vec![3.0, 2.0, 1.0]), + &ranges, + &ranges, + ); + assert_min_max_udfs_match_oracle( + ×tamps, + &Float64Array::from(vec![1.0, 2.0, 3.0]), + &ranges, + &ranges, + ); + } + + #[test] + fn min_max_over_time_batch_returns_null_after_expiring_last_nan() { + let nan = f64::from_bits(0x7ff8_0000_0000_00d4); + let timestamps = TimestampMillisecondArray::from(vec![0, 1]); + let values = Float64Array::from(vec![Some(nan), None]); + let ranges = [(0, 1), (1, 1)]; + let (timestamp_ranges, value_ranges) = + min_max_range_inputs(×tamps, &values, &ranges, &ranges); + assert_option_bits( + &run_min_max_udf(MinOverTime::scalar_udf(), timestamp_ranges, value_ranges), + &[Some(nan), None], + ); + let (timestamp_ranges, value_ranges) = + min_max_range_inputs(×tamps, &values, &ranges, &ranges); + assert_option_bits( + &run_min_max_udf(MaxOverTime::scalar_udf(), timestamp_ranges, value_ranges), + &[Some(nan), None], + ); + } + + #[test] + fn min_max_over_time_batch_matches_scalar_oracle_for_random_windows() { + let values = sliced_float_values( + (0..64) + .map(|index| match index % 11 { + 0 => None, + 1 => Some(f64::from_bits(0x7ff8_0000_0000_0100 + index)), + 2 => Some(-0.0), + 3 => Some(0.0), + 4 => Some(f64::INFINITY), + 5 => Some(f64::NEG_INFINITY), + _ => Some((index as f64 * 17.0).sin()), + }) + .collect(), + ); + let timestamps = TimestampMillisecondArray::from_iter((0..128).map(Some)); + let mut seed = 0x5eed_u64; + let mut value_ranges = Vec::new(); + let mut timestamp_ranges = Vec::new(); + for _ in 0..256 { + seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1); + let start = (seed as usize) % values.len(); + let length = ((seed >> 32) as usize) % (values.len() - start + 1); + value_ranges.push((start as u32, length as u32)); + timestamp_ranges.push(((values.len() - start) as u32, length as u32)); + } + + assert_min_max_udfs_match_oracle(×tamps, &values, ×tamp_ranges, &value_ranges); + } + + #[test] + fn min_max_over_time_batch_validates_inputs() { + let error = MinOverTime::calc(&[]).unwrap_err(); + assert!(error.to_string().contains("expected 2 inputs, found 0")); + + let timestamps = + RangeArray::from_ranges(Arc::new(TimestampMillisecondArray::from(vec![0])), [(0, 1)]) + .unwrap(); + let error = MaxOverTime::calc(&[ + ColumnarValue::Array(Arc::new(timestamps.into_dict())), + ColumnarValue::Array(Arc::new(datatypes::arrow::array::Int64Array::from(vec![1]))), + ]) + .unwrap_err(); + assert!( + error + .to_string() + .contains("expect value range vector as DictionaryArray") + ); + + let null_keys = datatypes::arrow::array::Int64Array::from_iter([Some(0), None]); + let null_key_dict = datatypes::arrow::array::DictionaryArray::< + datatypes::arrow::datatypes::Int64Type, + >::try_new( + null_keys, + Arc::new(TimestampMillisecondArray::from(vec![0, 1])), + ) + .unwrap(); + let error = MinOverTime::calc(&[ + ColumnarValue::Array(Arc::new(null_key_dict)), + ColumnarValue::Array(Arc::new(datatypes::arrow::array::Int64Array::from(vec![1]))), + ]) + .unwrap_err(); + assert!(error.to_string().contains("Empty range is not expected")); + } + + #[test] + fn min_max_over_time_batch_rejects_mismatched_windows() { + let timestamps = Arc::new(TimestampMillisecondArray::from(vec![0, 1, 2])); + let values = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0])); + let timestamp_ranges = RangeArray::from_ranges(timestamps.clone(), [(0, 1)]).unwrap(); + let value_ranges = RangeArray::from_ranges(values.clone(), [(0, 1), (1, 1)]).unwrap(); + let error = invoke_range_udf(MinOverTime::scalar_udf(), timestamp_ranges, value_ranges) + .unwrap_err(); + assert!(error.to_string().contains("same number of windows")); + + let timestamp_ranges = RangeArray::from_ranges(timestamps, [(0, 1)]).unwrap(); + let value_ranges = RangeArray::from_ranges(values, [(1, 2)]).unwrap(); + let error = invoke_range_udf(MaxOverTime::scalar_udf(), timestamp_ranges, value_ranges) + .unwrap_err(); + assert!(error.to_string().contains("different lengths at window 0")); + } fn assert_over_time_value(actual: Option, expected: Option) { match (actual, expected) { diff --git a/tests/cases/standalone/common/tql/aggr_over_time.sql b/tests/cases/standalone/common/tql/aggr_over_time.sql index f6068e7225..c99d01a6d2 100644 --- a/tests/cases/standalone/common/tql/aggr_over_time.sql +++ b/tests/cases/standalone/common/tql/aggr_over_time.sql @@ -185,3 +185,27 @@ tql eval (60, 60, '1s') max_over_time(data[2m]); tql eval (60, 60, '1s') last_over_time(data[2m]); drop table data; + +-- Sliding min/max regression: overlapping 30s windows expire extrema at the +-- left boundary, retain equal extrema, and end with a sparse empty window. +create table moving_extrema (ts timestamp(3) time index, val double, series string primary key); + +insert into moving_extrema values + (0, 5::double, 'moving'), + (10000, 1::double, 'moving'), + (20000, 4::double, 'moving'), + (30000, 4::double, 'moving'), + (40000, 2::double, 'moving'), + (60000, 3::double, 'moving'); + +-- eval range from 20s to 100s min_over_time(moving_extrema[30s]); 90s and 100s are empty. +-- {series="moving"} 1 1 2 2 2 3 3 +-- SQLNESS SORT_RESULT 2 1 +tql eval (20, 100, '10s') min_over_time(moving_extrema[30s]); + +-- eval range from 20s to 100s max_over_time(moving_extrema[30s]); 90s and 100s are empty. +-- {series="moving"} 5 4 4 4 4 3 3 +-- SQLNESS SORT_RESULT 2 1 +tql eval (20, 100, '10s') max_over_time(moving_extrema[30s]); + +drop table moving_extrema; diff --git a/tests/perf/query_cases/promql_range_boundary/case.toml b/tests/perf/query_cases/promql_range_boundary/case.toml index c533db8a22..2e62e2cb04 100644 --- a/tests/perf/query_cases/promql_range_boundary/case.toml +++ b/tests/perf/query_cases/promql_range_boundary/case.toml @@ -2,8 +2,8 @@ # # 128 series × 780 timestamps at 15s cadence = 99,840 rows. Data starts one # hour before evaluation; 481 evaluations span two hours. `count_over_time` is -# boundary-sensitive, while `sum_over_time` intentionally includes kernel scan -# work and the plain selector controls unrelated path variance. +# boundary-sensitive. Min/max cover sliding extrema, while `sum_over_time` +# intentionally includes kernel scan work and the plain selector controls unrelated path variance. [case] name = "promql_range_boundary" @@ -99,6 +99,88 @@ iterations = 9 [scenario.queries.thresholds] max_candidate_latency_regression_pct = 20 +# Sliding extrema cover small, representative, and wide overlapping windows. +[[scenario.queries]] +name = "min_over_time_1m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') min_over_time(promql_range_boundary{host=~'host.*'}[1m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + +[[scenario.queries]] +name = "max_over_time_1m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') max_over_time(promql_range_boundary{host=~'host.*'}[1m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + +[[scenario.queries]] +name = "min_over_time_5m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') min_over_time(promql_range_boundary{host=~'host.*'}[5m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + +[[scenario.queries]] +name = "max_over_time_5m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') max_over_time(promql_range_boundary{host=~'host.*'}[5m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + +[[scenario.queries]] +name = "min_over_time_1h" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') min_over_time(promql_range_boundary{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + +[[scenario.queries]] +name = "max_over_time_1h" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') max_over_time(promql_range_boundary{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + +# A bounded control with a one-hour evaluation step avoids overlapping windows. +[[scenario.queries]] +name = "min_over_time_1h_nonoverlap_control" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '1h') min_over_time(promql_range_boundary{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + +[[scenario.queries]] +name = "max_over_time_1h_nonoverlap_control" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '1h') max_over_time(promql_range_boundary{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 20 + # Edge-scan kernel candidates: the Phase 3 implementation replaces repeated # per-window scans with exact edge-prefix counts. [[scenario.queries]]