rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-12 16:32:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(promql): avoid per-window allocations in simple range functions (#9104)

Browse Source 
* perf(promql): avoid per-window allocations in simple range functions

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* perf(promql): avoid copying smoothing window values

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(perf): cover smoothing copy removal across window layouts

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(promql): skip null samples in simple range functions

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-11 11:59:45 +00:00
committed by GitHub
parent 21aed0371f
commit 555485c40e
9 changed files with 1054 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/macro/src/lib.rs
+4
View File
@@ -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<ColumnarValue, DataFusionError>`. 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)
src/common/macro/src/range_fn.rs
+29 -4
View File
@@ -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<Ident> = None;
let mut display_name: Option<Ident> = None;
let mut ret: Option<Ident> = None;
let mut evaluator: Option<Path> = 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<Type>,
fn_name: Ident,
ret_type: Ident,
evaluator: Option<Path>,
) -> TokenStream {
if let Some(evaluator) = evaluator {
return quote! {
impl #name {
fn calc(input: &[ColumnarValue]) -> Result<ColumnarValue, DataFusionError> {
#evaluator(input, Self::name())
}
}
}
.into();
}
let param_names = param_types
.iter()
.enumerate()
src/promql/benches/bench_range_fn.rs
+80 -1
View File
@@ -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<Columna
]
}
fn make_double_exponential_smoothing_input(
num_points: usize,
window_size: u32,
window_step: usize,
) -> Vec<ColumnarValue> {
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<ColumnarValue>,
arg_fields: Vec<Arc<Field>>,
@@ -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,
src/promql/src/functions/aggr_over_time.rs
+445 -12
View File
@@ -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<ColumnarValue, DataFusionError> {
evaluate_presence(input, name, PresenceEvaluator::Count)
}
fn last_over_time_evaluator(
input: &[ColumnarValue],
name: &str,
) -> Result<ColumnarValue, DataFusionError> {
evaluate_presence(input, name, PresenceEvaluator::Last)
}
fn absent_over_time_evaluator(
input: &[ColumnarValue],
name: &str,
) -> Result<ColumnarValue, DataFusionError> {
evaluate_presence(input, name, PresenceEvaluator::Absent)
}
fn present_over_time_evaluator(
input: &[ColumnarValue],
name: &str,
) -> Result<ColumnarValue, DataFusionError> {
evaluate_presence(input, name, PresenceEvaluator::Present)
}
fn evaluate_presence(
input: &[ColumnarValue],
name: &str,
operation: PresenceEvaluator,
) -> Result<ColumnarValue, DataFusionError> {
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::<Option<f64>>(),
))));
}
// 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::<TimestampMillisecondArray>()
);
let values = value_ranges
.values()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let evaluator: fn(&Float64Array, usize, usize) -> Option<f64> = 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<f64> {
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<f64> {
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<f64> {
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<f64> {
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(&timestamps, &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<f64>,
) {
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::<TimestampMillisecondArray>()
.unwrap(),
values.as_any().downcast_ref::<Float64Array>().unwrap(),
)
})
.collect::<Vec<_>>();
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::<Float64Array>()
.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::<Float64Array>()
.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<Int64Type>,
values: DictionaryArray<Int64Type>| {
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::<i64>::new()),
Arc::new(Float64Array::from_iter_values([1.0])),
),
DictionaryArray::new(
Int64Array::from(Vec::<i64>::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::<Float64Array>()
.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);
src/promql/src/functions/double_exponential_smoothing.rs
+104 -3
View File
@@ -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::<Vec<_>>();
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::<Vec<_>>();
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<f64> {
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<f64> {
input.split(' ').map(parse_promql_series_entry).fold(
tests/cases/standalone/common/promql/range_presence_null.result
+99
View File
@@ -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
tests/cases/standalone/common/promql/range_presence_null.sql
+35
View File
@@ -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;
tests/perf/query_cases/promql_simple_range_allocation/case.toml
+143
View File
@@ -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
tests/perf/query_cases/promql_smoothing_copy/case.toml
+115
View File
@@ -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