From 528ceb77337e9dae1ba055b8340dd74bd10e57ea Mon Sep 17 00:00:00 2001 From: discord9 Date: Thu, 17 Sep 2026 03:18:21 +0000 Subject: [PATCH] perf(promql): reuse sliding min and max candidates (#9099) * perf(promql): reuse sliding min and max candidates Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): simplify extrema benchmark parameters Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): record baseline sliding extrema SQL results Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * perf(promql): rescan windows that barely overlap Reusing candidates loses to a plain scan when consecutive windows overlap little: the deque bookkeeping then costs more than the rescan it replaces. A local Criterion run on 4096 samples at width 240 / step 240 measured 10.79 -> 22.14 us for min and 12.83 -> 19.85 us for max. Pick the evaluator once per batch from the first two windows. RangeManipulate emits one window length and one step per batch, so that sample decides for all of them, and both evaluators return identical bits, so a wrong pick costs time only. Batches that do not qualify fold each window on its own. Move the incremental state into SlidingExtrema so tests can drive it directly: the exhaustive four-sample differential test cannot reach it through a UDF call, because such a batch never qualifies for reuse. Signed-off-by: Dennis Zhuang * fix(promql): select the extrema evaluator from batch averages Reading the window shape off the first two windows misreads the batch. RangeManipulate starts a series at max(query start, first aligned sample), so a series that begins inside the query range gets a first window covering roughly one step, and a window covering no sample at all is emitted as (0, 0). Either one closed the gate for the whole batch, including the one-hour window at a 15s step that candidate reuse was written for. Compare the batch averages instead: at least 32 samples per window, and a step advancing at most a quarter of that. Uniform batches select exactly as before, so the thresholds keep the meaning they were measured with. The 32-sample rule had also moved most of the benchmark and query-regression shapes onto the rescan, including the case built to measure reset and rebuild. Widen those windows to 40 samples, add a step at the selection boundary, and add an end-to-end case with 40-sample windows advancing 5. Signed-off-by: Dennis Zhuang * fix(promql): ignore empty windows when measuring batch advance The advance was read from the first and last window offsets, but a window covering no sample is emitted as (0, 0). A query whose last evaluation lands exactly one window past the last sample ends on such a window, and its zero offset made a batch of disjoint windows look like one that never moved, which selected the evaluator built for overlap. Results stayed correct; the cost was deque bookkeeping on the shape the scan fallback exists for. Take the offset span over the windows that cover a sample. Empty windows stay in the window count, where they only make both conditions stricter. Signed-off-by: Dennis Zhuang --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: Dennis Zhuang Co-authored-by: Dennis Zhuang --- src/promql/benches/bench_range_fn.rs | 170 +++- src/promql/src/functions/aggr_over_time.rs | 803 +++++++++++++++++- .../common/tql/aggr_over_time.result | 155 ++++ .../standalone/common/tql/aggr_over_time.sql | 61 ++ .../promql_range_boundary/case.toml | 110 ++- 5 files changed, 1279 insertions(+), 20 deletions(-) diff --git a/src/promql/benches/bench_range_fn.rs b/src/promql/benches/bench_range_fn.rs index b42b4f60b0..0f0a486766 100644 --- a/src/promql/benches/bench_range_fn.rs +++ b/src/promql/benches/bench_range_fn.rs @@ -38,7 +38,8 @@ use futures::StreamExt; use promql::extension_plan::RangeManipulate; use promql::functions::{ AbsentOverTime, Changes, CountOverTime, Delta, DoubleExponentialSmoothing, IDelta, Increase, - LastOverTime, PredictLinear, PresentOverTime, QuantileOverTime, Rate, Resets, SumOverTime, + LastOverTime, MaxOverTime, MinOverTime, PredictLinear, PresentOverTime, QuantileOverTime, Rate, + Resets, SumOverTime, }; use promql::range_array::RangeArray; @@ -240,6 +241,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, @@ -417,6 +431,46 @@ fn bench_presence_range_functions(c: &mut Criterion) { group.finish(); } +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"); @@ -693,6 +747,119 @@ fn bench_rate_window_steps(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(); + // Cases meant to reuse candidates use 40-sample windows: the UDF rescans batches + // whose windows average fewer than 32 samples. + let mut backwards_ranges = (0..=num_points - 40) + .step_by(5) + .map(|offset| (offset as u32, 40)) + .collect::>(); + backwards_ranges.extend((0..=512).step_by(5).map(|offset| (offset as u32, 40))); + + // 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::>(), + ), + ( + "w40_step1", + (0..=num_points - 40) + .map(|offset| (offset as u32, 40)) + .collect::>(), + ), + ( + "w40_step5", + (0..=num_points - 40) + .step_by(5) + .map(|offset| (offset as u32, 40)) + .collect::>(), + ), + ( + "w240_step1", + (0..=num_points - 240) + .map(|offset| (offset as u32, 240)) + .collect::>(), + ), + // A quarter of the window is the widest step that still reuses candidates. + ( + "w240_step60", + (0..=num_points - 240) + .step_by(60) + .map(|offset| (offset as u32, 240)) + .collect::>(), + ), + ( + "w240_step240", + (0..=num_points - 240) + .step_by(240) + .map(|offset| (offset as u32, 240)) + .collect::>(), + ), + // A query ending one window past the last sample closes on an empty window. + ( + "w240_step240_trailing_empty", + (0..=num_points - 240) + .step_by(240) + .map(|offset| (offset as u32, 240)) + .chain([(0, 0)]) + .collect::>(), + ), + ("backwards_reset_rebuild_w40_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 = [ + ("min_over_time", &min_udf, true), + ("max_over_time", &max_udf, false), + ]; + + for (case_name, ranges) in cases { + let prepared = PreparedUdfCall::new(make_extrema_input_with_ranges( + values.clone(), + ranges.clone(), + )); + for (function_name, udf, is_min) in functions { + let expected = extrema_oracle(&values, &ranges, |value, extrema| { + if is_min { + value < extrema + } else { + value > extrema + } + }); + assert_extrema_output(udf, &prepared, &expected); + 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; @@ -1026,5 +1193,6 @@ criterion_group!( bench_delta_rate_comparison, bench_rate_window_steps, 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 3eeae47e61..2e7be0ab3b 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}; #[derive(Clone, Copy)] enum PresenceEvaluator { @@ -163,12 +164,306 @@ 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), + ] +} + +/// Batches with fewer windows than this have too little repeated work to reclaim. +const MIN_SLIDING_WINDOWS: usize = 4; +/// Below this average window length the per-sample bookkeeping is comparable to the +/// scan it would replace. +const MIN_SLIDING_WINDOW_LENGTH: u64 = 32; +/// Reuse is taken only when a step advances at most this fraction of the window. +const MAX_SLIDING_STEP_FRACTION: u64 = 4; + +/// Whether reusing candidates across windows is expected to beat rescanning each one. +/// +/// Reuse pays off in proportion to how much consecutive windows overlap, and loses to +/// a plain scan on wide windows that barely overlap: maintaining the deque then costs +/// more than the rescan it replaces. +/// +/// The shape is read from the whole batch rather than from its leading windows. +/// `RangeManipulate` holds the window duration and the evaluation step fixed, but the +/// sample counts still vary: a series that starts inside the query range gets a first +/// window covering roughly one step. Averages survive that; the first two windows do not. +/// +/// A wrong answer costs time, not correctness — both evaluators return the same bits. +fn reuses_candidates(window_keys: &[i64]) -> bool { + if window_keys.len() < MIN_SLIDING_WINDOWS { + return false; + } + + let windows = window_keys.len() as u64; + let mut total_length = 0u64; + let mut lowest_offset = u32::MAX; + let mut highest_offset = 0u32; + for &key in window_keys { + let (offset, length) = unpack(key); + // `RangeManipulate` emits a window covering no sample as `(0, 0)`, including + // the trailing one a query gets when its last evaluation lands exactly one + // window past the last sample. Its offset says nothing about the batch. + if length == 0 { + continue; + } + total_length += u64::from(length); + lowest_offset = lowest_offset.min(offset); + highest_offset = highest_offset.max(offset); + } + // What reuse skips re-reading: the distance the left bound travels over the batch. + let total_advance = u64::from(highest_offset.saturating_sub(lowest_offset)); + + // `total_length / windows >= MIN_SLIDING_WINDOW_LENGTH` and + // `total_advance / (windows - 1) <= (total_length / windows) / MAX_SLIDING_STEP_FRACTION`, + // cross-multiplied to keep the averages exact. Empty windows stay in the window + // count, which only makes both conditions stricter. + total_length >= windows * MIN_SLIDING_WINDOW_LENGTH + && total_advance * MAX_SLIDING_STEP_FRACTION * windows <= total_length * (windows - 1) +} + +fn is_better(value: f64, current: f64, is_min: bool) -> bool { + if is_min { + value < current + } else { + value > current + } +} + +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 sliding = reuses_candidates(value_keys).then(|| SlidingExtrema::new(is_min)); + 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; + result.push(match sliding.as_mut() { + Some(sliding) => sliding.evaluate(values, start, end), + None => scan_extremum(values, start, end, is_min), + }); + } + + Ok(ColumnarValue::Array(Arc::new(Float64Array::from_iter( + result, + )))) +} + +/// Reuses extrema candidates across windows whose bounds do not retreat. +struct SlidingExtrema { + is_min: bool, + /// Indices of the samples that can still become the extremum, in arrival order. + /// A strictly better sample evicts the ones queued before it, so the front is the + /// extremum of the current window and tied values keep their arrival order — and + /// with it the sign of tied zeros. + candidates: VecDeque, + /// Backs the all-NaN window, which keeps the last NaN of the window. + latest_nan: Option, + previous_window: Option<(usize, usize)>, +} + +impl SlidingExtrema { + fn new(is_min: bool) -> Self { + Self { + is_min, + candidates: VecDeque::new(), + latest_nan: None, + previous_window: None, + } + } + + fn evaluate(&mut self, values: &Float64Array, start: usize, end: usize) -> Option { + let append_start = match self.previous_window { + Some((previous_start, previous_end)) + if start >= previous_start && end >= previous_end => + { + while self.candidates.front().is_some_and(|&index| index < start) { + self.candidates.pop_front(); + } + if self.latest_nan.is_some_and(|index| index < start) { + self.latest_nan = None; + } + // Samples between two disjoint windows belong to neither, and later + // windows only move right, so skipping them keeps the state exact. + previous_end.max(start) + } + // A retreating bound can bring back samples that are no longer tracked. + Some(_) | None => { + self.candidates.clear(); + self.latest_nan = None; + start + } + }; + self.append(values, append_start, end); + self.previous_window = Some((start, end)); + + self.candidates + .front() + .or(self.latest_nan.as_ref()) + .map(|&index| values.value(index)) + } + + fn append(&mut self, values: &Float64Array, start: usize, end: usize) { + 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. + self.latest_nan = Some(index); + continue; + } + + while let Some(&tail_index) = self.candidates.back() { + if !is_better(value, values.value(tail_index), self.is_min) { + break; + } + self.candidates.pop_back(); + } + self.candidates.push_back(index); + } + } +} + +/// Folds one window on its own, the way the per-window scan used to. +fn scan_extremum(values: &Float64Array, start: usize, end: usize, is_min: bool) -> Option { + let mut extremum: Option = None; + for index in start..end { + if values.is_null(index) { + continue; + } + + let value = values.value(index); + let replace = match extremum { + None => true, + // A NaN only holds the slot until any other sample arrives. + Some(current) => current.is_nan() || is_better(value, current, is_min), + }; + if replace { + extremum = Some(value); + } + } + extremum +} + +#[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 { @@ -179,13 +474,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 { @@ -314,7 +604,486 @@ mod test { use datafusion::arrow::buffer::NullBuffer; 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 window_keys(ranges: &[(u32, u32)]) -> Vec { + let length = ranges + .iter() + .map(|&(offset, length)| (offset + length) as usize) + .max() + .unwrap_or_default(); + RangeArray::from_ranges( + Arc::new(Float64Array::from(vec![0.0; length])), + ranges.iter().copied(), + ) + .unwrap() + .into_dict() + .keys() + .values() + .to_vec() + } + + 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")); + } + + #[test] + fn sliding_evaluator_is_selected_only_for_overlapping_window_batches() { + // Enough windows, long enough, advancing by at most a quarter of the window. + assert!(reuses_candidates(&window_keys(&[ + (0, 32), + (8, 32), + (16, 32), + (24, 32) + ]))); + + // One window too few. + assert!(!reuses_candidates(&window_keys(&[ + (0, 32), + (8, 32), + (16, 32) + ]))); + // One sample per window too short. + assert!(!reuses_candidates(&window_keys(&[ + (0, 31), + (7, 31), + (14, 31), + (21, 31) + ]))); + // One sample per step too far apart. + assert!(!reuses_candidates(&window_keys(&[ + (0, 32), + (9, 32), + (18, 32), + (27, 32) + ]))); + } + + #[test] + fn sliding_evaluator_survives_an_unrepresentative_leading_window() { + // `RangeManipulate` gives a series that starts inside the query range a first + // window covering roughly one step, and emits a window covering no sample at + // all as (0, 0). Reading either one as the batch shape hides the overlap that + // the twenty windows behind it do have. + let overlapping = (0..20u32).map(|index| (index * 8, 40)); + for leading in [(0, 1), (0, 0)] { + let ranges = std::iter::once(leading) + .chain(overlapping.clone()) + .collect::>(); + assert!(reuses_candidates(&window_keys(&ranges))); + } + } + + #[test] + fn empty_windows_do_not_shorten_the_measured_advance() { + // A query whose last evaluation lands one window past the last sample ends on + // a (0, 0) window. Its zero offset must not read as a batch that never moved, + // which would put disjoint windows on the evaluator built for overlap. + let disjoint = (0..4u32) + .map(|index| (index * 240, 240)) + .collect::>(); + assert!(!reuses_candidates(&window_keys(&disjoint))); + let with_trailing_empty = [disjoint.as_slice(), &[(0, 0)]].concat(); + assert!(!reuses_candidates(&window_keys(&with_trailing_empty))); + } + + #[test] + fn min_max_over_time_batch_matches_oracle_on_overlapping_window_batches() { + let values = sliced_float_values( + (0..96) + .map(|index| match index % 13 { + 0 => None, + 1 => Some(f64::from_bits(0x7ff8_0000_0000_0200 + index as u64)), + 2 => Some(-0.0), + 3 => Some(0.0), + 4 => Some(f64::INFINITY), + 5 => Some(f64::NEG_INFINITY), + 6 | 7 => Some(5.0), + _ => Some(((index * 37) % 23) as f64), + }) + .collect(), + ); + let timestamps = TimestampMillisecondArray::from_iter((0..96).map(Some)); + // The first two windows open the sliding path; the rest then grow, repeat, + // jump over a gap, collapse to empty, retreat, and rebuild from scratch. + let ranges = [ + (0, 32), + (4, 32), + (8, 32), + (12, 32), + (16, 40), + (20, 36), + (20, 36), + (64, 32), + (64, 0), + (60, 20), + (0, 96), + ]; + assert!(reuses_candidates(&window_keys(&ranges))); + + assert_min_max_udfs_match_oracle(×tamps, &values, &ranges, &ranges); + } + + #[test] + fn both_extrema_paths_match_the_scalar_oracle_on_every_four_sample_window() { + let alphabet = [ + None, + Some(-0.0), + Some(0.0), + Some(-3.0), + Some(2.0), + Some(f64::INFINITY), + Some(f64::NEG_INFINITY), + Some(f64::from_bits(0x7ff8_0000_0000_0042)), + Some(f64::from_bits(0xfff8_0000_0000_0066)), + ]; + // Every window of a four-sample array, walked forwards and then backwards, so + // the evaluator sees growing, shrinking, disjoint, empty and retreating bounds. + let windows = (0..=4) + .flat_map(|start| (0..=4 - start).map(move |length| (start, start + length))) + .collect::>(); + + for encoded in 0..alphabet.len().pow(4) { + let mut remaining = encoded; + let values = Float64Array::from( + (0..4) + .map(|_| { + let value = alphabet[remaining % alphabet.len()]; + remaining /= alphabet.len(); + value + }) + .collect::>(), + ); + let mut sliding_min = SlidingExtrema::new(true); + let mut sliding_max = SlidingExtrema::new(false); + + for &(start, end) in windows.iter().chain(windows.iter().rev()) { + let window = values.slice(start, end - start); + let timestamps = TimestampMillisecondArray::new_null(window.len()); + + let expected = min_over_time(×tamps, &window).map(f64::to_bits); + assert_eq!( + sliding_min.evaluate(&values, start, end).map(f64::to_bits), + expected, + "sliding min, values {encoded}, window {start}..{end}" + ); + assert_eq!( + scan_extremum(&values, start, end, true).map(f64::to_bits), + expected, + "scanned min, values {encoded}, window {start}..{end}" + ); + + let expected = max_over_time(×tamps, &window).map(f64::to_bits); + assert_eq!( + sliding_max.evaluate(&values, start, end).map(f64::to_bits), + expected, + "sliding max, values {encoded}, window {start}..{end}" + ); + assert_eq!( + scan_extremum(&values, start, end, false).map(f64::to_bits), + expected, + "scanned max, values {encoded}, window {start}..{end}" + ); + } + } + } fn assert_over_time_value(actual: Option, expected: Option) { match (actual, expected) { diff --git a/tests/cases/standalone/common/tql/aggr_over_time.result b/tests/cases/standalone/common/tql/aggr_over_time.result index e39c9d5238..c2b38febd4 100644 --- a/tests/cases/standalone/common/tql/aggr_over_time.result +++ b/tests/cases/standalone/common/tql/aggr_over_time.result @@ -340,3 +340,158 @@ drop table data; Affected Rows: 0 +-- 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); + +Affected Rows: 0 + +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'); + +Affected Rows: 6 + +-- 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]); + ++---------------------+----------------------------------+--------+ +| ts | prom_min_over_time(ts_range,val) | series | ++---------------------+----------------------------------+--------+ +| 1970-01-01T00:00:20 | 1.0 | moving | +| 1970-01-01T00:00:30 | 1.0 | moving | +| 1970-01-01T00:00:40 | 2.0 | moving | +| 1970-01-01T00:00:50 | 2.0 | moving | +| 1970-01-01T00:01:00 | 2.0 | moving | +| 1970-01-01T00:01:10 | 3.0 | moving | +| 1970-01-01T00:01:20 | 3.0 | moving | ++---------------------+----------------------------------+--------+ + +-- eval range from 20s to 100s max_over_time(moving_extrema[30s]); 90s and 100s are empty. +-- {series="moving"} 5 4 4 4 3 3 3 +-- SQLNESS SORT_RESULT 2 1 +tql eval (20, 100, '10s') max_over_time(moving_extrema[30s]); + ++---------------------+----------------------------------+--------+ +| ts | prom_max_over_time(ts_range,val) | series | ++---------------------+----------------------------------+--------+ +| 1970-01-01T00:00:20 | 5.0 | moving | +| 1970-01-01T00:00:30 | 4.0 | moving | +| 1970-01-01T00:00:40 | 4.0 | moving | +| 1970-01-01T00:00:50 | 4.0 | moving | +| 1970-01-01T00:01:00 | 3.0 | moving | +| 1970-01-01T00:01:10 | 3.0 | moving | +| 1970-01-01T00:01:20 | 3.0 | moving | ++---------------------+----------------------------------+--------+ + +drop table moving_extrema; + +Affected Rows: 0 + +-- Dense sliding min/max: 40-sample windows advancing 5 samples, the shape the UDF +-- reuses candidates for rather than rescanning. Values follow the timestamp, so each +-- window's minimum is its oldest sample and its maximum is its newest, and both +-- extrema expire on every step. +create table dense_extrema (ts timestamp_s time index, val double, series string primary key); + +Affected Rows: 0 + +insert into dense_extrema values (0, 0::double, 'dense'); + +Affected Rows: 1 + +-- Doubling eight times gives 256 rows at a one-second cadence, with val = ts. +insert into dense_extrema select to_unixtime(ts) + 1, val + 1, series from dense_extrema; + +Affected Rows: 1 + +insert into dense_extrema select to_unixtime(ts) + 2, val + 2, series from dense_extrema; + +Affected Rows: 2 + +insert into dense_extrema select to_unixtime(ts) + 4, val + 4, series from dense_extrema; + +Affected Rows: 4 + +insert into dense_extrema select to_unixtime(ts) + 8, val + 8, series from dense_extrema; + +Affected Rows: 8 + +insert into dense_extrema select to_unixtime(ts) + 16, val + 16, series from dense_extrema; + +Affected Rows: 16 + +insert into dense_extrema select to_unixtime(ts) + 32, val + 32, series from dense_extrema; + +Affected Rows: 32 + +insert into dense_extrema select to_unixtime(ts) + 64, val + 64, series from dense_extrema; + +Affected Rows: 64 + +insert into dense_extrema select to_unixtime(ts) + 128, val + 128, series from dense_extrema; + +Affected Rows: 128 + +select count(*), min(val), max(val) from dense_extrema; + ++----------+------------------------+------------------------+ +| count(*) | min(dense_extrema.val) | max(dense_extrema.val) | ++----------+------------------------+------------------------+ +| 256 | 0.0 | 255.0 | ++----------+------------------------+------------------------+ + +-- {series="dense"} 61 66 71 76 81 86 91 96 101 106 111 116 121 +-- SQLNESS SORT_RESULT 2 1 +tql eval (100, 160, '5s') min_over_time(dense_extrema[40s]); + ++---------------------+----------------------------------+--------+ +| ts | prom_min_over_time(ts_range,val) | series | ++---------------------+----------------------------------+--------+ +| 1970-01-01T00:01:40 | 61.0 | dense | +| 1970-01-01T00:01:45 | 66.0 | dense | +| 1970-01-01T00:01:50 | 71.0 | dense | +| 1970-01-01T00:01:55 | 76.0 | dense | +| 1970-01-01T00:02:00 | 81.0 | dense | +| 1970-01-01T00:02:05 | 86.0 | dense | +| 1970-01-01T00:02:10 | 91.0 | dense | +| 1970-01-01T00:02:15 | 96.0 | dense | +| 1970-01-01T00:02:20 | 101.0 | dense | +| 1970-01-01T00:02:25 | 106.0 | dense | +| 1970-01-01T00:02:30 | 111.0 | dense | +| 1970-01-01T00:02:35 | 116.0 | dense | +| 1970-01-01T00:02:40 | 121.0 | dense | ++---------------------+----------------------------------+--------+ + +-- {series="dense"} 100 105 110 115 120 125 130 135 140 145 150 155 160 +-- SQLNESS SORT_RESULT 2 1 +tql eval (100, 160, '5s') max_over_time(dense_extrema[40s]); + ++---------------------+----------------------------------+--------+ +| ts | prom_max_over_time(ts_range,val) | series | ++---------------------+----------------------------------+--------+ +| 1970-01-01T00:01:40 | 100.0 | dense | +| 1970-01-01T00:01:45 | 105.0 | dense | +| 1970-01-01T00:01:50 | 110.0 | dense | +| 1970-01-01T00:01:55 | 115.0 | dense | +| 1970-01-01T00:02:00 | 120.0 | dense | +| 1970-01-01T00:02:05 | 125.0 | dense | +| 1970-01-01T00:02:10 | 130.0 | dense | +| 1970-01-01T00:02:15 | 135.0 | dense | +| 1970-01-01T00:02:20 | 140.0 | dense | +| 1970-01-01T00:02:25 | 145.0 | dense | +| 1970-01-01T00:02:30 | 150.0 | dense | +| 1970-01-01T00:02:35 | 155.0 | dense | +| 1970-01-01T00:02:40 | 160.0 | dense | ++---------------------+----------------------------------+--------+ + +drop table dense_extrema; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/tql/aggr_over_time.sql b/tests/cases/standalone/common/tql/aggr_over_time.sql index f6068e7225..f8e2fe26bd 100644 --- a/tests/cases/standalone/common/tql/aggr_over_time.sql +++ b/tests/cases/standalone/common/tql/aggr_over_time.sql @@ -185,3 +185,64 @@ 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 3 3 3 +-- SQLNESS SORT_RESULT 2 1 +tql eval (20, 100, '10s') max_over_time(moving_extrema[30s]); + +drop table moving_extrema; + +-- Dense sliding min/max: 40-sample windows advancing 5 samples, the shape the UDF +-- reuses candidates for rather than rescanning. Values follow the timestamp, so each +-- window's minimum is its oldest sample and its maximum is its newest, and both +-- extrema expire on every step. +create table dense_extrema (ts timestamp_s time index, val double, series string primary key); + +insert into dense_extrema values (0, 0::double, 'dense'); + +-- Doubling eight times gives 256 rows at a one-second cadence, with val = ts. +insert into dense_extrema select to_unixtime(ts) + 1, val + 1, series from dense_extrema; + +insert into dense_extrema select to_unixtime(ts) + 2, val + 2, series from dense_extrema; + +insert into dense_extrema select to_unixtime(ts) + 4, val + 4, series from dense_extrema; + +insert into dense_extrema select to_unixtime(ts) + 8, val + 8, series from dense_extrema; + +insert into dense_extrema select to_unixtime(ts) + 16, val + 16, series from dense_extrema; + +insert into dense_extrema select to_unixtime(ts) + 32, val + 32, series from dense_extrema; + +insert into dense_extrema select to_unixtime(ts) + 64, val + 64, series from dense_extrema; + +insert into dense_extrema select to_unixtime(ts) + 128, val + 128, series from dense_extrema; + +select count(*), min(val), max(val) from dense_extrema; + +-- {series="dense"} 61 66 71 76 81 86 91 96 101 106 111 116 121 +-- SQLNESS SORT_RESULT 2 1 +tql eval (100, 160, '5s') min_over_time(dense_extrema[40s]); + +-- {series="dense"} 100 105 110 115 120 125 130 135 140 145 150 155 160 +-- SQLNESS SORT_RESULT 2 1 +tql eval (100, 160, '5s') max_over_time(dense_extrema[40s]); + +drop table dense_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..4a25476c29 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,112 @@ iterations = 9 [scenario.queries.thresholds] max_candidate_latency_regression_pct = 20 +# Sliding extrema cover small, representative, and wide overlapping windows. At this +# cadence the 1m and 5m windows hold about 5 and 21 samples, below the 32-sample +# average the UDF requires before it reuses candidates, so they measure the rescan. +[[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 ten-minute step keeps the one-hour window on the reuse path without the +# single-sample step the two cases above use. +[[scenario.queries]] +name = "min_over_time_1h_step10m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '10m') 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_step10m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '10m') 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]]