diff --git a/src/common/macro/src/lib.rs b/src/common/macro/src/lib.rs index 5a58c4d7fb..a397292787 100644 --- a/src/common/macro/src/lib.rs +++ b/src/common/macro/src/lib.rs @@ -75,6 +75,10 @@ pub fn as_aggr_func_creator(args: TokenStream, input: TokenStream) -> TokenStrea /// - `name`: The name of the generated `ScalarUDF` struct. /// - `ret`: The return type of the generated UDF function. /// - `display_name`: The display name of the generated UDF function. +/// - `evaluator`: Optional path to a specialized evaluator with the calling convention +/// `fn(&[ColumnarValue], &str) -> Result`. When supplied, +/// the generated UDF `calc` delegates directly to it; without it, the default expansion is +/// unchanged. #[proc_macro_attribute] pub fn range_fn(args: TokenStream, input: TokenStream) -> TokenStream { process_range_fn(args, input) diff --git a/src/common/macro/src/range_fn.rs b/src/common/macro/src/range_fn.rs index 5ee21ca4b0..7d282db0fa 100644 --- a/src/common/macro/src/range_fn.rs +++ b/src/common/macro/src/range_fn.rs @@ -16,7 +16,7 @@ use proc_macro::TokenStream; use quote::quote; use syn::spanned::Spanned; use syn::{ - Attribute, Ident, ItemFn, Signature, Type, TypeReference, Visibility, parse_macro_input, + Attribute, Ident, ItemFn, Path, Signature, Type, TypeReference, Visibility, parse_macro_input, }; use crate::utils::extract_input_types; @@ -34,6 +34,7 @@ pub(crate) fn process_range_fn(args: TokenStream, input: TokenStream) -> TokenSt let mut name: Option = None; let mut display_name: Option = None; let mut ret: Option = None; + let mut evaluator: Option = None; let parser = syn::meta::parser(|meta| { if meta.path.is_ident("name") { @@ -45,6 +46,9 @@ pub(crate) fn process_range_fn(args: TokenStream, input: TokenStream) -> TokenSt } else if meta.path.is_ident("ret") { ret = Some(meta.value()?.parse()?); Ok(()) + } else if meta.path.is_ident("evaluator") { + evaluator = Some(meta.value()?.parse()?); + Ok(()) } else { Err(meta.error("unsupported property")) } @@ -103,10 +107,19 @@ pub(crate) fn process_range_fn(args: TokenStream, input: TokenStream) -> TokenSt arg_types, fn_name.clone(), ret.expect("ret required"), + evaluator.clone(), ); - // preserve this fn, but remove its `pub` modifier - let input_fn_code: TokenStream = quote! { - #sig { #block } + // Preserve this fn, but remove its `pub` modifier. Specialized evaluators + // do not call it in production, while tests keep it as the slice oracle. + let input_fn_code: TokenStream = if evaluator.is_some() { + quote! { + #[cfg(test)] + #sig { #block } + } + } else { + quote! { + #sig { #block } + } } .into(); @@ -161,7 +174,19 @@ fn build_calc_fn( param_types: Vec, fn_name: Ident, ret_type: Ident, + evaluator: Option, ) -> TokenStream { + if let Some(evaluator) = evaluator { + return quote! { + impl #name { + fn calc(input: &[ColumnarValue]) -> Result { + #evaluator(input, Self::name()) + } + } + } + .into(); + } + let param_names = param_types .iter() .enumerate() diff --git a/src/promql/benches/bench_range_fn.rs b/src/promql/benches/bench_range_fn.rs index ccc05ea849..16fdfe3a4c 100644 --- a/src/promql/benches/bench_range_fn.rs +++ b/src/promql/benches/bench_range_fn.rs @@ -37,7 +37,8 @@ use datatypes::arrow::datatypes::{DataType, Field}; use futures::StreamExt; use promql::extension_plan::RangeManipulate; use promql::functions::{ - Changes, Delta, IDelta, Increase, PredictLinear, QuantileOverTime, Rate, Resets, SumOverTime, + AbsentOverTime, Changes, CountOverTime, Delta, DoubleExponentialSmoothing, IDelta, Increase, + LastOverTime, PredictLinear, PresentOverTime, QuantileOverTime, Rate, Resets, SumOverTime, }; use promql::range_array::RangeArray; @@ -270,6 +271,26 @@ fn make_predict_linear_input(num_points: usize, window_size: u32) -> Vec Vec { + let (ts_range, val_range, _) = build_sliding_ranges( + num_points, + window_size, + window_step, + build_gauge_values(num_points), + 0, + ); + vec![ + ColumnarValue::Array(Arc::new(ts_range.into_dict())), + ColumnarValue::Array(Arc::new(val_range.into_dict())), + ColumnarValue::Scalar(ScalarValue::Float64(Some(0.5))), + ColumnarValue::Scalar(ScalarValue::Float64(Some(0.1))), + ] +} + struct PreparedUdfCall { args: Vec, arg_fields: Vec>, @@ -359,6 +380,43 @@ fn assert_edge_count_output( assert_eq!(actual, expected); } +fn bench_presence_range_functions(c: &mut Criterion) { + let mut group = c.benchmark_group("presence_range_fn"); + let values = build_default_values(4_096); + let overlapping = PreparedUdfCall::new(make_edge_count_input(4_096, 20, values.clone())); + let low_coverage_ranges = vec![ + (0, 4), + (512, 4), + (1_024, 4), + (1_536, 4), + (2_048, 4), + (2_560, 4), + (3_584, 4), + (4_092, 4), + ]; + let low_coverage = PreparedUdfCall::new(make_edge_count_input_with_ranges( + values, + low_coverage_ranges, + )); + let udfs = [ + ("count_over_time", CountOverTime::scalar_udf()), + ("last_over_time", LastOverTime::scalar_udf()), + ("present_over_time", PresentOverTime::scalar_udf()), + ("absent_over_time", AbsentOverTime::scalar_udf()), + ]; + + for (name, udf) in &udfs { + group.bench_with_input(BenchmarkId::new(*name, "N4096_overlap_w20"), &(), |b, _| { + b.iter(|| invoke_prepared(udf, &overlapping)) + }); + group.bench_with_input(BenchmarkId::new(*name, "N4096_windows8_w4"), &(), |b, _| { + b.iter(|| invoke_prepared(udf, &low_coverage)) + }); + } + + group.finish(); +} + fn bench_range_functions(c: &mut Criterion) { let mut group = c.benchmark_group("range_fn"); @@ -497,6 +555,26 @@ fn bench_range_functions(c: &mut Criterion) { ); } + // --- double_exponential_smoothing --- + let smoothing_udf = DoubleExponentialSmoothing::scalar_udf(); + for (window_size, window_step, case) in [ + (4, 1, "N4096_w4_overlap"), + (20, 1, "N4096_w20_overlap"), + (240, 1, "N4096_w240_overlap"), + (240, 240, "N4096_w240_nonoverlap"), + ] { + let prepared = PreparedUdfCall::new(make_double_exponential_smoothing_input( + 4_096, + window_size, + window_step, + )); + group.bench_with_input( + BenchmarkId::new("double_exponential_smoothing", case), + &(), + |b, _| b.iter(|| invoke_prepared(&smoothing_udf, &prepared)), + ); + } + // --- RangeArray: get vs get_offset_length micro-benchmark --- // Isolates the overhead of array slicing vs offset/length lookup for &(n, w) in params { @@ -943,6 +1021,7 @@ fn bench_range_manipulate_wall_time(c: &mut Criterion) { criterion_group!( benches, bench_range_functions, + bench_presence_range_functions, bench_delta_rate_comparison, bench_rate_window_steps, bench_edge_count_functions, diff --git a/src/promql/src/functions/aggr_over_time.rs b/src/promql/src/functions/aggr_over_time.rs index 05b2b1b2e2..63f045f134 100644 --- a/src/promql/src/functions/aggr_over_time.rs +++ b/src/promql/src/functions/aggr_over_time.rs @@ -26,6 +26,129 @@ use datatypes::arrow::datatypes::DataType; use crate::functions::{compensated_sum_inc, extract_array}; use crate::range_array::RangeArray; +#[derive(Clone, Copy)] +enum PresenceEvaluator { + Count, + Last, + Absent, + Present, +} + +fn count_over_time_evaluator( + input: &[ColumnarValue], + name: &str, +) -> Result { + evaluate_presence(input, name, PresenceEvaluator::Count) +} + +fn last_over_time_evaluator( + input: &[ColumnarValue], + name: &str, +) -> Result { + evaluate_presence(input, name, PresenceEvaluator::Last) +} + +fn absent_over_time_evaluator( + input: &[ColumnarValue], + name: &str, +) -> Result { + evaluate_presence(input, name, PresenceEvaluator::Absent) +} + +fn present_over_time_evaluator( + input: &[ColumnarValue], + name: &str, +) -> Result { + evaluate_presence(input, name, PresenceEvaluator::Present) +} + +fn evaluate_presence( + input: &[ColumnarValue], + name: &str, + operation: PresenceEvaluator, +) -> Result { + assert_eq!(input.len(), 2); + + let timestamp_ranges = RangeArray::try_new(extract_array(&input[0])?.to_data().into())?; + let value_ranges = RangeArray::try_new(extract_array(&input[1])?.to_data().into())?; + let len = timestamp_ranges.len(); + if len != value_ranges.len() { + return Err(DataFusionError::Execution(format!( + "RangeArray have different lengths in PromQL function {name}: array1={len}, array2={}", + value_ranges.len() + ))); + } + + if timestamp_ranges.is_empty() { + return Ok(ColumnarValue::Array(Arc::new(Float64Array::from_iter( + std::iter::empty::>(), + )))); + } + + // The generic range-function wrapper downcasts both arrays for each window. + // Do it once after retaining its zero-window behavior above. + assert!( + timestamp_ranges + .values() + .as_any() + .is::() + ); + let values = value_ranges + .values() + .as_any() + .downcast_ref::() + .unwrap(); + let evaluator: fn(&Float64Array, usize, usize) -> Option = if values.null_count() == 0 { + match operation { + PresenceEvaluator::Count => |_, _, length| (length != 0).then_some(length as f64), + PresenceEvaluator::Last => { + |values, offset, length| (length != 0).then(|| values.value(offset + length - 1)) + } + PresenceEvaluator::Absent => |_, _, length| (length == 0).then_some(1.0), + PresenceEvaluator::Present => |_, _, length| (length != 0).then_some(1.0), + } + } else { + match operation { + PresenceEvaluator::Count => |values, offset, length| { + let count = (offset..offset + length) + .filter(|&index| values.is_valid(index)) + .count(); + (count != 0).then_some(count as f64) + }, + PresenceEvaluator::Last => |values, offset, length| { + (offset..offset + length) + .rev() + .find(|&index| values.is_valid(index)) + .map(|index| values.value(index)) + }, + PresenceEvaluator::Absent => |values, offset, length| { + (!(offset..offset + length).any(|index| values.is_valid(index))).then_some(1.0) + }, + PresenceEvaluator::Present => |values, offset, length| { + (offset..offset + length) + .any(|index| values.is_valid(index)) + .then_some(1.0) + }, + } + }; + + let mut result = Vec::with_capacity(len); + for index in 0..len { + let (_, timestamp_length) = timestamp_ranges.get_offset_length(index).unwrap(); + let (value_offset, value_length) = value_ranges.get_offset_length(index).unwrap(); + if timestamp_length != value_length { + return Err(DataFusionError::Execution(format!( + "RangeArray's element {index} have different lengths in PromQL function {name}: array1={timestamp_length}, array2={value_length}" + ))); + } + result.push(evaluator(values, value_offset, value_length)); + } + + Ok(ColumnarValue::Array(Arc::new(Float64Array::from_iter( + result, + )))) +} + /// The average value of all points in the specified interval. #[range_fn( name = AvgOverTime, @@ -84,24 +207,23 @@ pub fn sum_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Op #[range_fn( name = CountOverTime, ret = Float64Array, - display_name = prom_count_over_time + display_name = prom_count_over_time, + evaluator = count_over_time_evaluator )] pub fn count_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { - if values.is_empty() { - None - } else { - Some(values.len() as f64) - } + let count = values.iter().flatten().count(); + (count != 0).then_some(count as f64) } /// The most recent point value in specified interval. #[range_fn( name = LastOverTime, ret = Float64Array, - display_name = prom_last_over_time + display_name = prom_last_over_time, + evaluator = last_over_time_evaluator )] pub fn last_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { - values.values().last().copied() + values.iter().flatten().last() } /// absent_over_time returns an empty vector if the range vector passed to it has any @@ -110,20 +232,22 @@ pub fn last_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> O #[range_fn( name = AbsentOverTime, ret = Float64Array, - display_name = prom_absent_over_time + display_name = prom_absent_over_time, + evaluator = absent_over_time_evaluator )] pub fn absent_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { - if values.is_empty() { Some(1.0) } else { None } + values.iter().flatten().next().is_none().then_some(1.0) } /// the value 1 for any series in the specified interval. #[range_fn( name = PresentOverTime, ret = Float64Array, - display_name = prom_present_over_time + display_name = prom_present_over_time, + evaluator = present_over_time_evaluator )] pub fn present_over_time(_: &TimestampMillisecondArray, values: &Float64Array) -> Option { - if values.is_empty() { None } else { Some(1.0) } + values.iter().flatten().next().is_some().then_some(1.0) } /// the population standard variance of the values in the specified interval. @@ -218,6 +342,315 @@ mod test { assert_over_time_value(max_over_time(×tamps, &values), expected_max); } + fn special_ranges() -> (RangeArray, RangeArray) { + use datafusion::arrow::buffer::NullBuffer; + + let timestamps = Arc::new(TimestampMillisecondArray::from_iter_values(0..10)).slice(1, 8); + let values = Arc::new(Float64Array::new( + vec![ + 99.0, + -0.0, + 0.0, + f64::INFINITY, + f64::NEG_INFINITY, + f64::from_bits(0x7ff8_0000_0000_0042), + f64::from_bits(0x7ff8_0000_0000_0066), + 3.0, + -7.0, + 42.0, + ] + .into(), + Some(NullBuffer::from(vec![ + true, true, true, true, true, false, true, true, true, true, + ])), + )) + .slice(1, 8); + // Empty, singleton, overlapping, disjoint, repeated, and backward windows use + // independent timestamp and value offsets. + let timestamp_ranges = [ + (0, 0), + (1, 1), + (2, 1), + (3, 1), + (4, 1), + (5, 1), + (6, 1), + (0, 3), + (3, 2), + (4, 2), + (6, 2), + (5, 1), + (2, 1), + ]; + let value_ranges = [ + (7, 0), + (0, 1), + (1, 1), + (2, 1), + (3, 1), + (4, 1), + (5, 1), + (0, 3), + (3, 2), + (4, 2), + (1, 2), + (5, 1), + (1, 1), + ]; + + let timestamp_ranges = RangeArray::from_ranges( + Arc::new(timestamps), + std::iter::once((0, 0)).chain(timestamp_ranges), + ) + .unwrap() + .into_dict() + .slice(1, timestamp_ranges.len()); + let value_ranges = RangeArray::from_ranges( + Arc::new(values), + std::iter::once((0, 0)).chain(value_ranges), + ) + .unwrap() + .into_dict() + .slice(1, value_ranges.len()); + + ( + RangeArray::try_new(timestamp_ranges).unwrap(), + RangeArray::try_new(value_ranges).unwrap(), + ) + } + + fn assert_specialized_matches_oracle( + udf: ScalarUDF, + kernel: fn(&TimestampMillisecondArray, &Float64Array) -> Option, + ) { + use crate::functions::test_util::invoke_range_udf; + + let (timestamps, values) = special_ranges(); + let expected = (0..timestamps.len()) + .map(|index| { + let timestamps = timestamps.get(index).unwrap(); + let values = values.get(index).unwrap(); + kernel( + timestamps + .as_any() + .downcast_ref::() + .unwrap(), + values.as_any().downcast_ref::().unwrap(), + ) + }) + .collect::>(); + let (timestamps, values) = special_ranges(); + let output = invoke_range_udf(udf, timestamps, values).unwrap(); + let output_array = extract_array(&output).unwrap(); + let output = output_array + .as_any() + .downcast_ref::() + .unwrap(); + + assert_eq!(output.len(), expected.len()); + for (index, expected) in expected.into_iter().enumerate() { + assert_eq!(output.is_null(index), expected.is_none()); + if let Some(expected) = expected { + assert_eq!(output.value(index).to_bits(), expected.to_bits()); + } + } + } + + #[test] + fn specialized_presence_range_udfs_match_slice_oracles() { + assert_specialized_matches_oracle(CountOverTime::scalar_udf(), count_over_time); + assert_specialized_matches_oracle(LastOverTime::scalar_udf(), last_over_time); + assert_specialized_matches_oracle(AbsentOverTime::scalar_udf(), absent_over_time); + assert_specialized_matches_oracle(PresentOverTime::scalar_udf(), present_over_time); + } + + #[test] + fn specialized_presence_range_udfs_ignore_null_samples() { + use datafusion::arrow::buffer::NullBuffer; + + let make_ranges = || { + let timestamps = + Arc::new(TimestampMillisecondArray::from_iter_values(0..9)).slice(1, 7); + let values = Arc::new(Float64Array::new( + vec![99.0, 1.0, 0.0, 0.0, 4.0, f64::NAN, 0.0, -0.0, 42.0].into(), + Some(NullBuffer::from(vec![ + true, true, false, false, true, true, true, true, true, + ])), + )) + .slice(1, 7); + let ranges = [(0, 4), (0, 3), (1, 2), (4, 1), (5, 2), (7, 0)]; + + ( + RangeArray::from_ranges(Arc::new(timestamps), ranges).unwrap(), + RangeArray::from_ranges(Arc::new(values), ranges).unwrap(), + ) + }; + // The first window is [1, NULL, NULL, 4]; the old physical-length evaluator + // incorrectly returned 4 for count_over_time. + let cases = [ + ( + CountOverTime::scalar_udf(), + [Some(2.0), Some(1.0), None, Some(1.0), Some(2.0), None], + ), + ( + LastOverTime::scalar_udf(), + [Some(4.0), Some(1.0), None, Some(f64::NAN), Some(-0.0), None], + ), + ( + AbsentOverTime::scalar_udf(), + [None, None, Some(1.0), None, None, Some(1.0)], + ), + ( + PresentOverTime::scalar_udf(), + [Some(1.0), Some(1.0), None, Some(1.0), Some(1.0), None], + ), + ]; + + for (udf, expected) in cases { + let (timestamps, values) = make_ranges(); + let output = + crate::functions::test_util::invoke_range_udf(udf, timestamps, values).unwrap(); + let output_array = extract_array(&output).unwrap(); + let output = output_array + .as_any() + .downcast_ref::() + .unwrap(); + + for (index, expected) in expected.into_iter().enumerate() { + assert_eq!(output.is_valid(index), expected.is_some()); + if let Some(expected) = expected { + assert_eq!(output.value(index).to_bits(), expected.to_bits()); + } + } + } + } + + #[test] + fn specialized_presence_range_udfs_preserve_errors_and_metadata() { + use datafusion::arrow::array::{DictionaryArray, Int64Array}; + use datafusion::arrow::datatypes::{Field, Int64Type}; + use datafusion_common::config::ConfigOptions; + use datafusion_expr::ScalarFunctionArgs; + + use crate::functions::test_util::{assert_execution_error, invoke_range_udf}; + + for (udf, name) in [ + (CountOverTime::scalar_udf(), "prom_count_over_time"), + (LastOverTime::scalar_udf(), "prom_last_over_time"), + (AbsentOverTime::scalar_udf(), "prom_absent_over_time"), + (PresentOverTime::scalar_udf(), "prom_present_over_time"), + ] { + assert_eq!(udf.name(), name); + assert_eq!(udf.signature().volatility, Volatility::Volatile); + } + + let timestamps = Arc::new(TimestampMillisecondArray::from_iter_values(0..3)); + let values = Arc::new(Float64Array::from_iter_values([1.0, 2.0, 3.0])); + let error = invoke_range_udf( + CountOverTime::scalar_udf(), + RangeArray::from_ranges(timestamps.clone(), [(0, 1), (1, 1)]).unwrap(), + RangeArray::from_ranges(values.clone(), [(0, 1)]).unwrap(), + ) + .unwrap_err(); + assert_execution_error( + error, + "RangeArray have different lengths in PromQL function prom_count_over_time: array1=2, array2=1", + ); + + let error = invoke_range_udf( + CountOverTime::scalar_udf(), + RangeArray::from_ranges(timestamps.clone(), [(0, 1), (1, 2)]).unwrap(), + RangeArray::from_ranges(values.clone(), [(0, 1), (1, 1)]).unwrap(), + ) + .unwrap_err(); + assert_execution_error( + error, + "RangeArray's element 1 have different lengths in PromQL function prom_count_over_time: array1=2, array2=1", + ); + + let invoke_dict = |timestamps: DictionaryArray, + values: DictionaryArray| { + let args = vec![ + ColumnarValue::Array(Arc::new(timestamps)), + ColumnarValue::Array(Arc::new(values)), + ]; + CountOverTime::scalar_udf().invoke_with_args(ScalarFunctionArgs { + arg_fields: args + .iter() + .enumerate() + .map(|(index, value)| { + Arc::new(Field::new( + format!("c{index}"), + value.data_type().clone(), + true, + )) + }) + .collect(), + args, + number_rows: 1, + return_field: Arc::new(Field::new("out", DataType::Float64, true)), + config_options: Arc::new(ConfigOptions::default()), + }) + }; + let empty = invoke_dict( + DictionaryArray::new( + Int64Array::from(Vec::::new()), + Arc::new(Float64Array::from_iter_values([1.0])), + ), + DictionaryArray::new( + Int64Array::from(Vec::::new()), + Arc::new(TimestampMillisecondArray::from_iter_values([0])), + ), + ) + .unwrap(); + assert!(extract_array(&empty).unwrap().is_empty()); + + let null_keys = DictionaryArray::new( + Int64Array::from(vec![None]), + Arc::new(TimestampMillisecondArray::from_iter_values([0])), + ); + let values_range = RangeArray::from_ranges(values.clone(), [(0, 1)]).unwrap(); + assert_eq!( + invoke_dict(null_keys, values_range.into_dict()) + .unwrap_err() + .to_string(), + "External error: Empty range is not expected" + ); + + let timestamps_range = RangeArray::from_ranges(timestamps, [(0, 1)]).unwrap(); + let invalid_values = unsafe { RangeArray::from_ranges_unchecked(values, [(2, 2)]) }; + assert_eq!( + invoke_dict(timestamps_range.into_dict(), invalid_values.into_dict()) + .unwrap_err() + .to_string(), + "External error: Illegal range: offset 2, length 2, array len 3" + ); + + let output = invoke_range_udf( + SumOverTime::scalar_udf(), + RangeArray::from_ranges( + Arc::new(TimestampMillisecondArray::from_iter_values([0, 1])), + [(0, 2)], + ) + .unwrap(), + RangeArray::from_ranges( + Arc::new(Float64Array::from_iter_values([1.0, 2.0])), + [(0, 2)], + ) + .unwrap(), + ) + .unwrap(); + assert_eq!( + extract_array(&output) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 3.0 + ); + } + #[test] fn min_max_over_time_ignore_ordinary_nan_when_finite_values_exist() { let ordinary_nan = f64::from_bits(0x7ff8_0000_0000_0000); diff --git a/src/promql/src/functions/double_exponential_smoothing.rs b/src/promql/src/functions/double_exponential_smoothing.rs index b6768d47a1..2f8e0235d5 100644 --- a/src/promql/src/functions/double_exponential_smoothing.rs +++ b/src/promql/src/functions/double_exponential_smoothing.rs @@ -240,8 +240,6 @@ fn double_exponential_smoothing_impl(values: &[f64], sf: f64, tf: f64) -> Option return Some(f64::NAN); } - let values = values.to_vec(); - let mut s0 = 0.0; let mut s1 = values[0]; let mut b = values[1] - values[0]; @@ -353,6 +351,109 @@ mod tests { ); } + #[test] + fn test_double_exponential_smoothing_impl_copy_oracle() { + let normal_values = (0..240) + .map(|i| (i as f64 - 120.0) * 0.25) + .collect::>(); + let special_values = (0..240) + .map(|i| match i % 8 { + 0 => 0.0, + 1 => -0.0, + 2 => f64::INFINITY, + 3 => f64::NEG_INFINITY, + 4 => f64::NAN, + 5 => f64::from_bits(0x7ff8_0000_0000_0001), + 6 => 42.5, + _ => -42.5, + }) + .collect::>(); + let factors = [ + (0.0, 0.0), + (-0.0, 1.0), + (0.5, 0.1), + (1.0, 1.0), + (-0.5, 0.5), + (0.5, -0.5), + (1.5, 0.5), + (0.5, 1.5), + (f64::NAN, 0.5), + (0.5, f64::NAN), + (f64::INFINITY, 0.5), + (0.5, f64::INFINITY), + (f64::NEG_INFINITY, 0.5), + (0.5, f64::NEG_INFINITY), + ]; + + for (values_name, values) in [ + ("normal", normal_values.as_slice()), + ("special", special_values.as_slice()), + ] { + for len in [0, 1, 2, 3, 20, 240] { + let values = &values[..len]; + for (sf, tf) in factors { + let old = double_exponential_smoothing_impl_with_copy(values, sf, tf).unwrap(); + let new = double_exponential_smoothing_impl(values, sf, tf).unwrap(); + let case = format!("values={values_name}, len={len}, sf={sf:?}, tf={tf:?}"); + + if old.is_nan() || new.is_nan() { + assert!( + old.is_nan() && new.is_nan(), + "NaN mismatch for {case}: old={old:?}, new={new:?}" + ); + assert_eq!( + old.to_bits(), + new.to_bits(), + "NaN bit difference for {case}: old={:#018x}, new={:#018x}", + old.to_bits(), + new.to_bits(), + ); + } else { + assert_eq!( + old.to_bits(), + new.to_bits(), + "non-NaN bit difference for {case}: old={old:?}, new={new:?}" + ); + } + } + } + } + } + + fn double_exponential_smoothing_impl_with_copy( + values: &[f64], + sf: f64, + tf: f64, + ) -> Option { + if sf.is_nan() || tf.is_nan() || values.is_empty() { + return Some(f64::NAN); + } + if sf < 0.0 || tf < 0.0 { + return Some(f64::NEG_INFINITY); + } + if sf > 1.0 || tf > 1.0 { + return Some(f64::INFINITY); + } + + if values.len() <= 2 { + return Some(f64::NAN); + } + + let values = values.to_vec(); + let mut s0 = 0.0; + let mut s1 = values[0]; + let mut b = values[1] - values[0]; + + for (i, value) in values.iter().enumerate().skip(1) { + let x = sf * value; + b = calc_trend_value(i - 1, tf, s0, s1, b); + let y = (1.0 - sf) * (s1 + b); + s0 = s1; + s1 = x + y; + } + Some(s1) + } + #[test] fn test_prom_double_exponential_smoothing_monotonic() { let ranges = [(0, 5)]; @@ -450,7 +551,7 @@ mod tests { (ts_range_array, value_range_array) } - /// Converts a prometheus functions test series into a vector of f64 element with respect to resets and trend direction + /// Converts a prometheus functions test series into a vector of f64 element with respect to resets and trend direction /// The input example: "0+10x1000 100+30x1000" fn create_test_range_from_promql_series(input: &str) -> Vec { input.split(' ').map(parse_promql_series_entry).fold( diff --git a/tests/cases/standalone/common/promql/range_presence_null.result b/tests/cases/standalone/common/promql/range_presence_null.result new file mode 100644 index 0000000000..d0f28ad9fb --- /dev/null +++ b/tests/cases/standalone/common/promql/range_presence_null.result @@ -0,0 +1,99 @@ +-- NULL fields are missing samples, not zero-valued samples. +CREATE TABLE range_presence_null ( + ts TIMESTAMP(3) TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE, +); + +Affected Rows: 0 + +INSERT INTO range_presence_null VALUES + (0, 'a', 1.0), + (1000, 'a', NULL), + (2000, 'a', NULL), + (3000, 'a', 4.0), + (0, 'b', NULL), + (1000, 'b', NULL), + (2000, 'b', NULL), + (3000, 'b', NULL); + +Affected Rows: 8 + +-- At t=2 the trailing NULLs must not hide 1; at t=3 count must be 2. +-- At t=7 the left-open window is empty. Valid results must disappear. +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') count_over_time(range_presence_null{host="a"}[4s]); + ++---------------------+------------------------------------+------+ +| ts | prom_count_over_time(ts_range,val) | host | ++---------------------+------------------------------------+------+ +| 1970-01-01T00:00:02 | 1.0 | a | +| 1970-01-01T00:00:03 | 2.0 | a | +| 1970-01-01T00:00:04 | 1.0 | a | +| 1970-01-01T00:00:05 | 1.0 | a | +| 1970-01-01T00:00:06 | 1.0 | a | ++---------------------+------------------------------------+------+ + +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') last_over_time(range_presence_null{host="a"}[4s]); + ++---------------------+-----------------------------------+------+ +| ts | prom_last_over_time(ts_range,val) | host | ++---------------------+-----------------------------------+------+ +| 1970-01-01T00:00:02 | 1.0 | a | +| 1970-01-01T00:00:03 | 4.0 | a | +| 1970-01-01T00:00:04 | 4.0 | a | +| 1970-01-01T00:00:05 | 4.0 | a | +| 1970-01-01T00:00:06 | 4.0 | a | ++---------------------+-----------------------------------+------+ + +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') present_over_time(range_presence_null{host="a"}[4s]); + ++---------------------+--------------------------------------+------+ +| ts | prom_present_over_time(ts_range,val) | host | ++---------------------+--------------------------------------+------+ +| 1970-01-01T00:00:02 | 1.0 | a | +| 1970-01-01T00:00:03 | 1.0 | a | +| 1970-01-01T00:00:04 | 1.0 | a | +| 1970-01-01T00:00:05 | 1.0 | a | +| 1970-01-01T00:00:06 | 1.0 | a | ++---------------------+--------------------------------------+------+ + +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') absent_over_time(range_presence_null{host="a"}[4s]); + ++---------------------+-------------------------------------+------+ +| ts | prom_absent_over_time(ts_range,val) | host | ++---------------------+-------------------------------------+------+ +| 1970-01-01T00:00:07 | 1.0 | a | ++---------------------+-------------------------------------+------+ + +-- All-NULL windows have no samples: only absent_over_time returns 1. +TQL EVAL (3, 3, '1s') count_over_time(range_presence_null{host="b"}[4s]); + +++ +++ + +TQL EVAL (3, 3, '1s') last_over_time(range_presence_null{host="b"}[4s]); + +++ +++ + +TQL EVAL (3, 3, '1s') present_over_time(range_presence_null{host="b"}[4s]); + +++ +++ + +TQL EVAL (3, 3, '1s') absent_over_time(range_presence_null{host="b"}[4s]); + ++---------------------+-------------------------------------+------+ +| ts | prom_absent_over_time(ts_range,val) | host | ++---------------------+-------------------------------------+------+ +| 1970-01-01T00:00:03 | 1.0 | b | ++---------------------+-------------------------------------+------+ + +DROP TABLE range_presence_null; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/promql/range_presence_null.sql b/tests/cases/standalone/common/promql/range_presence_null.sql new file mode 100644 index 0000000000..e442679b2b --- /dev/null +++ b/tests/cases/standalone/common/promql/range_presence_null.sql @@ -0,0 +1,35 @@ +-- NULL fields are missing samples, not zero-valued samples. +CREATE TABLE range_presence_null ( + ts TIMESTAMP(3) TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE, +); + +INSERT INTO range_presence_null VALUES + (0, 'a', 1.0), + (1000, 'a', NULL), + (2000, 'a', NULL), + (3000, 'a', 4.0), + (0, 'b', NULL), + (1000, 'b', NULL), + (2000, 'b', NULL), + (3000, 'b', NULL); + +-- At t=2 the trailing NULLs must not hide 1; at t=3 count must be 2. +-- At t=7 the left-open window is empty. Valid results must disappear. +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') count_over_time(range_presence_null{host="a"}[4s]); +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') last_over_time(range_presence_null{host="a"}[4s]); +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') present_over_time(range_presence_null{host="a"}[4s]); +-- SQLNESS SORT_RESULT 3 1 +TQL EVAL (2, 7, '1s') absent_over_time(range_presence_null{host="a"}[4s]); + +-- All-NULL windows have no samples: only absent_over_time returns 1. +TQL EVAL (3, 3, '1s') count_over_time(range_presence_null{host="b"}[4s]); +TQL EVAL (3, 3, '1s') last_over_time(range_presence_null{host="b"}[4s]); +TQL EVAL (3, 3, '1s') present_over_time(range_presence_null{host="b"}[4s]); +TQL EVAL (3, 3, '1s') absent_over_time(range_presence_null{host="b"}[4s]); + +DROP TABLE range_presence_null; diff --git a/tests/perf/query_cases/promql_simple_range_allocation/case.toml b/tests/perf/query_cases/promql_simple_range_allocation/case.toml new file mode 100644 index 0000000000..95d00ad069 --- /dev/null +++ b/tests/perf/query_cases/promql_simple_range_allocation/case.toml @@ -0,0 +1,143 @@ +# Bounded PromQL simple-range allocation regression coverage. +# +# 128 series × 780 samples at a 15-second cadence = 99,840 rows. Linear +# values are `base + series_idx % 97 + local_ordinal`; the host0000 HTTP +# control therefore has last values 240, 480, and 720 at its hourly evaluations. + +[case] +name = "promql_simple_range_allocation" +description = "Bounded PromQL simple-range UDF allocation regression coverage" + +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +database = "public" +metric = "promql_simple_range_allocation" +physical_table = "greptime_physical_table" +series_count = 128 +samples_per_series = 780 +sample_chunk_size = 195 +flush_every_sample_chunks = 1 +start_unix_millis = 1704067200000 +step_millis = 15000 +chunk_series_count = 128 +timeout_seconds = 600 +visibility_timeout_seconds = 300 + +[scenario.remote_write.value] +pattern = "linear" +base = 0 +step = 1 + +[scenario.remote_write.prom_store] +pending_rows_flush_interval = "1s" +max_batch_rows = 100000 + +[[scenario.queries]] +name = "selector_2h15s_control" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') promql_simple_range_allocation{host=~'host.*'}" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "count_over_time_1m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') count_over_time(promql_simple_range_allocation{host=~'host.*'}[1m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "count_over_time_1h" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') count_over_time(promql_simple_range_allocation{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "last_over_time_1m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') last_over_time(promql_simple_range_allocation{host=~'host.*'}[1m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "last_over_time_1h" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') last_over_time(promql_simple_range_allocation{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "present_over_time_1m" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') present_over_time(promql_simple_range_allocation{host=~'host.*'}[1m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "present_over_time_1h" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') present_over_time(promql_simple_range_allocation{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# This matcher is intentionally nonempty in the input but produces no output. +# ANALYZE remains valid while exercising the absent-over-time allocation path. +[[scenario.queries]] +name = "absent_over_time_1m_empty" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') absent_over_time(promql_simple_range_allocation{host=~'host.*'}[1m])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "absent_over_time_1h_empty" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') absent_over_time(promql_simple_range_allocation{host=~'host.*'}[1h])" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# Inspect the retained response manually: the runner does not assert numeric values. +# For host0000, last_over_time at the hourly evaluations should return 240, 480, +# and 720 respectively. +[[scenario.queries]] +name = "prom_http_last_over_time_1h_numeric_control" +kind = "prom_http" +query = "last_over_time(promql_simple_range_allocation{host=\"host0000\"}[1h])" +start = "1704070800" +end = "1704078000" +step = "1h" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 diff --git a/tests/perf/query_cases/promql_smoothing_copy/case.toml b/tests/perf/query_cases/promql_smoothing_copy/case.toml new file mode 100644 index 0000000000..187794dd1c --- /dev/null +++ b/tests/perf/query_cases/promql_smoothing_copy/case.toml @@ -0,0 +1,115 @@ +# Bounded nightly PromQL double-exponential-smoothing regression coverage. +# +# 128 series × 780 samples at 15-second cadence = 99,840 rows. Remote write +# ingests four 195-sample chunks and flushes each chunk before querying. + +[case] +name = "promql_smoothing_copy" +description = "Bounded nightly PromQL double-exponential-smoothing copy-removal regression" + +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +database = "public" +metric = "promql_smoothing_copy" +physical_table = "greptime_physical_table" +series_count = 128 +samples_per_series = 780 +sample_chunk_size = 195 +flush_every_sample_chunks = 1 +start_unix_millis = 1_704_067_200_000 +step_millis = 15_000 +chunk_series_count = 128 +timeout_seconds = 180 +visibility_timeout_seconds = 120 + +[scenario.remote_write.value] +pattern = "linear" +base = 0 +step = 1 + +[scenario.remote_write.prom_store] +pending_rows_flush_interval = "1s" +max_batch_rows = 100000 + +# Selector-only scan control for unrelated PromQL/metric-engine variance. +[[scenario.queries]] +name = "selector_2h_15s_control" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') promql_smoothing_copy{host=~'host.*'}" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "double_exponential_smoothing_1m_15s" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') double_exponential_smoothing(promql_smoothing_copy{host=~'host.*'}[1m], 0.5, 0.1)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "double_exponential_smoothing_5m_15s" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') double_exponential_smoothing(promql_smoothing_copy{host=~'host.*'}[5m], 0.5, 0.1)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# Primary copy-removal stress case: 481 highly-overlapping 240-sample windows. +[[scenario.queries]] +name = "double_exponential_smoothing_1h_15s" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '15s') double_exponential_smoothing(promql_smoothing_copy{host=~'host.*'}[1h], 0.5, 0.1)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "double_exponential_smoothing_1h_5m_overlap" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '5m') double_exponential_smoothing(promql_smoothing_copy{host=~'host.*'}[1h], 0.5, 0.1)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# Three disjoint 240-sample windows provide a non-overlapping control. +[[scenario.queries]] +name = "double_exponential_smoothing_1h_1h_nonoverlap" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '1h') double_exponential_smoothing(promql_smoothing_copy{host=~'host.*'}[1h], 0.5, 0.1)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# Inspect retained responses manually; the runner does not assert numeric values. +# The three left-open hourly windows contain host0000 values 1..240, 241..480, +# and 481..720. Initial trend is 1 and remains unchanged on the first iteration; +# subsequent trend updates preserve it. With sf=0.5 and tf=0.1, the expected +# outputs are 240, 480, and 720. +[[scenario.queries]] +name = "double_exponential_smoothing_1h_host0000_numeric_control" +kind = "prom_http" +query = "double_exponential_smoothing(promql_smoothing_copy{host=\"host0000\"}[1h], 0.5, 0.1)" +start = "1704070800" +end = "1704078000" +step = "1h" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10