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