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]]