diff --git a/src/promql/benches/bench_range_fn.rs b/src/promql/benches/bench_range_fn.rs index 840956b942..80eefde2cd 100644 --- a/src/promql/benches/bench_range_fn.rs +++ b/src/promql/benches/bench_range_fn.rs @@ -26,9 +26,12 @@ use datatypes::arrow::datatypes::{DataType, Field}; use promql::functions::{Delta, IDelta, Increase, PredictLinear, QuantileOverTime, Rate}; use promql::range_array::RangeArray; +/// A `window_step` below `window_size` makes consecutive windows overlap, which is the normal +/// PromQL range query shape. fn build_sliding_ranges( num_points: usize, window_size: u32, + window_step: usize, values: Vec, eval_offset_ms: i64, ) -> (RangeArray, RangeArray, Arc) { @@ -44,10 +47,12 @@ fn build_sliding_ranges( 0 }; - let ranges: Vec<(u32, u32)> = (0..num_windows).map(|i| (i as u32, window_size)).collect(); + let offsets: Vec = (0..num_windows).step_by(window_step).collect(); + let ranges: Vec<(u32, u32)> = offsets.iter().map(|&i| (i as u32, window_size)).collect(); - let eval_ts: Vec = (0..num_windows) - .map(|i| timestamps[i + window_size as usize - 1] + eval_offset_ms) + let eval_ts: Vec = offsets + .iter() + .map(|&i| timestamps[i + window_size as usize - 1] + eval_offset_ms) .collect(); let eval_ts_array = Arc::new(TimestampMillisecondArray::from(eval_ts)); @@ -94,11 +99,12 @@ fn build_default_values(num_points: usize) -> Vec { fn make_extrapolated_rate_input( num_points: usize, window_size: u32, + window_step: usize, values: Vec, eval_offset_ms: i64, ) -> Vec { let (ts_range, val_range, eval_ts) = - build_sliding_ranges(num_points, window_size, values, eval_offset_ms); + build_sliding_ranges(num_points, window_size, window_step, values, eval_offset_ms); let range_length = window_size as i64 * 1000; vec![ ColumnarValue::Array(Arc::new(ts_range.into_dict())), @@ -109,8 +115,13 @@ fn make_extrapolated_rate_input( } fn make_idelta_input(num_points: usize, window_size: u32) -> Vec { - let (ts_range, val_range, _) = - build_sliding_ranges(num_points, window_size, build_default_values(num_points), 0); + let (ts_range, val_range, _) = build_sliding_ranges( + num_points, + window_size, + 1, + build_default_values(num_points), + 0, + ); vec![ ColumnarValue::Array(Arc::new(ts_range.into_dict())), ColumnarValue::Array(Arc::new(val_range.into_dict())), @@ -118,8 +129,13 @@ fn make_idelta_input(num_points: usize, window_size: u32) -> Vec } fn make_quantile_input(num_points: usize, window_size: u32) -> Vec { - let (ts_range, val_range, _) = - build_sliding_ranges(num_points, window_size, build_default_values(num_points), 0); + let (ts_range, val_range, _) = build_sliding_ranges( + num_points, + window_size, + 1, + build_default_values(num_points), + 0, + ); vec![ ColumnarValue::Array(Arc::new(ts_range.into_dict())), ColumnarValue::Array(Arc::new(val_range.into_dict())), @@ -128,8 +144,13 @@ fn make_quantile_input(num_points: usize, window_size: u32) -> Vec Vec { - let (ts_range, val_range, _) = - build_sliding_ranges(num_points, window_size, build_default_values(num_points), 0); + let (ts_range, val_range, _) = build_sliding_ranges( + num_points, + window_size, + 1, + build_default_values(num_points), + 0, + ); vec![ ColumnarValue::Array(Arc::new(ts_range.into_dict())), ColumnarValue::Array(Arc::new(val_range.into_dict())), @@ -198,6 +219,7 @@ fn bench_range_functions(c: &mut Criterion) { let prepared = PreparedUdfCall::new(make_extrapolated_rate_input( n, w, + 1, build_monotonic_counter_values(n), 500, )); @@ -213,6 +235,7 @@ fn bench_range_functions(c: &mut Criterion) { let prepared = PreparedUdfCall::new(make_extrapolated_rate_input( n, w, + 1, build_resetting_counter_values(n), 500, )); @@ -229,6 +252,7 @@ fn bench_range_functions(c: &mut Criterion) { let prepared = PreparedUdfCall::new(make_extrapolated_rate_input( n, w, + 1, build_monotonic_counter_values(n), 500, )); @@ -244,6 +268,7 @@ fn bench_range_functions(c: &mut Criterion) { let prepared = PreparedUdfCall::new(make_extrapolated_rate_input( n, w, + 1, build_resetting_counter_values(n), 500, )); @@ -260,6 +285,7 @@ fn bench_range_functions(c: &mut Criterion) { let prepared = PreparedUdfCall::new(make_extrapolated_rate_input( n, w, + 1, build_gauge_values(n), 500, )); @@ -352,4 +378,38 @@ fn bench_range_functions(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, bench_range_functions); +fn bench_rate_window_steps(c: &mut Criterion) { + let mut group = c.benchmark_group("rate_window_steps"); + let rate_udf = Rate::scalar_udf(); + let num_points = 20_280; + let window_size = 120u32; + + // No resets, one reset per 24 window widths, and roughly three resets per window. + for reset_period in [0usize, 2_880, 37] { + let values: Vec = match reset_period { + 0 => (0..num_points).map(|i| i as f64).collect(), + period => (0..num_points).map(|i| (i % period) as f64).collect(), + }; + for window_step in [1usize, 10, 120] { + let prepared = PreparedUdfCall::new(make_extrapolated_rate_input( + num_points, + window_size, + window_step, + values.clone(), + 500, + )); + group.bench_with_input( + BenchmarkId::new( + "rate_counter", + format!("reset{reset_period}_step{window_step}"), + ), + &(), + |b, _| b.iter(|| invoke_prepared(&rate_udf, &prepared)), + ); + } + } + + group.finish(); +} + +criterion_group!(benches, bench_range_functions, bench_rate_window_steps); diff --git a/src/promql/src/functions/extrapolate_rate.rs b/src/promql/src/functions/extrapolate_rate.rs index af0714a8f7..f072400d3e 100644 --- a/src/promql/src/functions/extrapolate_rate.rs +++ b/src/promql/src/functions/extrapolate_rate.rs @@ -30,6 +30,7 @@ //! Implementations of `rate`, `increase` and `delta` functions in PromQL. use std::fmt::Display; +use std::ops::Range; use std::sync::Arc; use datafusion::arrow::array::{Float64Array, Float64Builder, TimestampMillisecondArray}; @@ -192,9 +193,24 @@ impl ExtrapolatedRate budget + }) + .then(|| CounterResetIndex::new(all_values)) + } else { + None + }; for index in 0..num_windows { let (raw_offset, raw_length) = unpack(keys[index]); @@ -203,7 +219,6 @@ impl ExtrapolatedRate ExtrapolatedRate reset_index.add_resets(result_value, offset, end), + None => add_counter_resets(result_value, &all_values[offset..end]), + }; + } let first_ts = all_timestamps[offset]; let last_ts = all_timestamps[end - 1]; @@ -286,6 +282,105 @@ impl ExtrapolatedRate f64 { + values + .windows(2) + .filter(|pair| pair[1] < pair[0]) + .fold(result, |result, pair| result + pair[0]) +} + +/// Positions of the counter resets in a value array, so that a window can accumulate the +/// resets it contains instead of scanning all of its samples. +struct CounterResetIndex<'a> { + values: &'a [f64], + /// Ascending indices `i` where `values[i] < values[i - 1]`. + positions: Vec, + /// Slice of `positions` covered by the last window. + active: Range, + /// That window, so the next one can tell whether it advanced. + previous: Range, + /// `positions[active.start]`: the reset a later `start` would drop. `usize::MAX` when the + /// active slice reaches the end of `positions`. + drops_at: usize, + /// `positions[active.end]`: the reset a later `end` would gain, saturated the same way. + gains_at: usize, +} + +impl<'a> CounterResetIndex<'a> { + fn new(values: &'a [f64]) -> Self { + let positions: Vec = (1..values.len()) + .filter(|&i| values[i] < values[i - 1]) + .collect(); + let first = positions.first().copied().unwrap_or(usize::MAX); + Self { + values, + positions, + active: 0..0, + previous: 0..0, + drops_at: first, + gains_at: first, + } + } + + /// Same additions [`add_counter_resets`] performs over `values[start..end]`, in the same + /// order, reached through the index instead of by scanning the window. + #[inline] + fn add_resets(&mut self, result: f64, start: usize, end: usize) -> f64 { + // The active slice only stays put if the window advanced without reaching either of + // the resets that bound it. + if start < self.previous.start + || end < self.previous.end + || start >= self.drops_at + || end > self.gains_at + { + self.locate(start, end); + } + self.previous = start..end; + + if self.active.start == self.active.end { + // A counter that has not reset inside this window, which is the normal case, would + // otherwise pay a range bounds check and an empty iterator for nothing. + return result; + } + + let values = self.values; + self.positions[self.active.start..self.active.end] + .iter() + .fold(result, |result, &i| result + values[i - 1]) + } + + fn locate(&mut self, start: usize, end: usize) { + // Walk the bounds forward from the previous window and only search when they move + // back. On a series that resets often the searches cost more than the additions they + // locate, because they run deep and a window holds a handful of resets. + let (left, right) = if start < self.previous.start || end < self.previous.end { + ( + self.positions.partition_point(|&i| i <= start), + self.positions.partition_point(|&i| i < end), + ) + } else { + let mut left = self.active.start; + while left < self.positions.len() && self.positions[left] <= start { + left += 1; + } + let mut right = self.active.end.max(left); + while right < self.positions.len() && self.positions[right] < end { + right += 1; + } + (left, right) + }; + self.active = left..right; + self.drops_at = self.positions.get(left).copied().unwrap_or(usize::MAX); + self.gains_at = self.positions.get(right).copied().unwrap_or(usize::MAX); + } +} + fn extract_eval_timestamps( columnar_value: &ColumnarValue, func_name: &str, @@ -409,6 +504,109 @@ mod test { ) } + /// Evaluates `ranges` as one batch and asserts every window is bit-identical to evaluating + /// that window on its own, which always takes the direct per-window reduction. + fn assert_counter_windows_match_single(values: &[f64], ranges: &[(u32, u32)]) { + let timestamps = Arc::new(TimestampMillisecondArray::from_iter_values( + (0..values.len()).map(|i| i as i64 * 30_000 + (i % 5) as i64 * 1_000), + )); + let values = Arc::new(Float64Array::from(values.to_vec())); + let evaluate = |ranges: &[(u32, u32)]| { + let eval_ts = Arc::new(TimestampMillisecondArray::from_iter_values( + ranges.iter().map(|&(offset, length)| { + timestamps.value((offset + length.saturating_sub(1)) as usize) + 5_000 + }), + )); + let input = [ + ColumnarValue::Array(Arc::new( + RangeArray::from_ranges(timestamps.clone(), ranges.iter().copied()) + .unwrap() + .into_dict(), + )), + ColumnarValue::Array(Arc::new( + RangeArray::from_ranges(values.clone(), ranges.iter().copied()) + .unwrap() + .into_dict(), + )), + ColumnarValue::Array(eval_ts), + ColumnarValue::Scalar(ScalarValue::Int64(Some(3_600_000))), + ]; + extract_array(&Rate::new(3_600_000).calc(&input).unwrap()) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect::>() + }; + + for (range, batched) in ranges.iter().zip(evaluate(ranges)) { + let single = evaluate(std::slice::from_ref(range))[0]; + match (batched, single) { + (None, None) => {} + (Some(batched), Some(single)) => assert!( + batched.to_bits() == single.to_bits() || (batched.is_nan() && single.is_nan()), + "range {range:?}: batched {batched} != single {single}" + ), + _ => panic!("range {range:?}: batched {batched:?} != single {single:?}"), + } + } + } + + #[test] + fn counter_resets_accumulate_into_the_running_result() { + // Summed on their own, 1e16 and 1.0 round to 1e16, which then cancels against the + // first sample and reports no increase at all. Folding each reset into `last - first` + // as Prometheus does keeps the 1.0. Both paths detect the same two resets. + let ts_array = Arc::new(TimestampMillisecondArray::from_iter( + [1, 2, 3, 4].into_iter().map(Some), + )); + let values_array = Arc::new(Float64Array::from_iter([1e16, 1.0, 0.0, 1.0])); + let ranges = [(0, 4)]; + let ts_range = RangeArray::from_ranges(ts_array, ranges).unwrap(); + let value_range = RangeArray::from_ranges(values_array, ranges).unwrap(); + let timestamps = Arc::new(TimestampMillisecondArray::from_iter([Some(4)])) as _; + + extrapolated_rate_runner::( + ts_range, + value_range, + timestamps, + vec![1.1666666666666667], + ); + } + + #[test] + fn counter_correction_survives_huge_and_infinite_resets() { + // Both series reset twice inside the first window and once inside the second, but the + // first reset is large enough to swallow the second one when they are summed together. + for values in [ + vec![1e16, 1.0, 0.0, 1.0, 2.0], + vec![f64::INFINITY, 1.0, 0.0, 1.0, 2.0], + ] { + assert_counter_windows_match_single(&values, &[(0, 4), (1, 4)]); + } + } + + #[test] + fn counter_correction_matches_single_window_on_irregular_layouts() { + let mut values: Vec = (0..512).map(|i| (i % 37) as f64 * 0.25).collect(); + values[20] = f64::NAN; + values[70] = f64::INFINITY; + values[140] = f64::NEG_INFINITY; + values[220] = 1e300; + values[221] = 1e-200; + + // A stride of one walks both bounds across the reset positions, which sit every 37 + // samples: `start` reaches one when that reset has to leave the window, `end` when the + // next one has to stay out of it, both while the cached bounds are live. + let mut ranges: Vec<(u32, u32)> = (0..390).map(|i| (i, 120)).collect(); + // Empty, too-short, backward and disjoint windows all break a forward-only slide. + ranges.extend([(400, 0), (400, 1), (2, 20), (450, 30), (0, 120)]); + ranges.extend((0..390).rev().step_by(10).map(|i| (i, 120))); + + assert_counter_windows_match_single(&values, &ranges); + } + #[test] fn rate_rejects_wrong_input_arity() { let err = ExtrapolatedRate::::new(5) diff --git a/tests/cases/standalone/common/promql/counter_reset_precision.result b/tests/cases/standalone/common/promql/counter_reset_precision.result new file mode 100644 index 0000000000..3482f7ca9b --- /dev/null +++ b/tests/cases/standalone/common/promql/counter_reset_precision.result @@ -0,0 +1,43 @@ +CREATE TABLE counter_reset_precision ( + ts TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + series STRING PRIMARY KEY +); + +Affected Rows: 0 + +INSERT INTO counter_reset_precision VALUES + (0, 10000000000000000.0, 'a'), + (30000, 1.0, 'a'), + (60000, 0.0, 'a'), + (90000, 1.0, 'a'), + (120000, 2.0, 'a'), + (150000, 3.0, 'a'), + (180000, 4.0, 'a'); + +Affected Rows: 7 + +-- The counter resets twice, at 30s from 1e16 and at 60s from 1.0, and 1e16 + 1.0 is 1e16 in +-- f64. Windows are two minutes wide and step by one sample, so every row here depends on the +-- 1.0 reset surviving. +-- +-- Summing the two corrections on their own drops the 1.0 and then cancels against the first +-- sample, so the 90s window reports no increase. Carrying that sum into the next window and +-- subtracting the 1e16 that left it reports half the increase, and the windows after that +-- subtract a reset the sum never held, so the correction turns negative and stays there for +-- the rest of the batch. +TQL EVAL (90, 180, '30s') increase(counter_reset_precision[2m]); + ++---------------------+---------------------------------------------------------+--------+ +| ts | prom_increase(ts_range,greptime_value,ts,Int64(120000)) | series | ++---------------------+---------------------------------------------------------+--------+ +| 1970-01-01T00:01:30 | 1.3333333333333333 | a | +| 1970-01-01T00:02:00 | 2.6666666666666665 | a | +| 1970-01-01T00:02:30 | 3.0 | a | +| 1970-01-01T00:03:00 | 4.0 | a | ++---------------------+---------------------------------------------------------+--------+ + +DROP TABLE counter_reset_precision; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/promql/counter_reset_precision.sql b/tests/cases/standalone/common/promql/counter_reset_precision.sql new file mode 100644 index 0000000000..867260dc3d --- /dev/null +++ b/tests/cases/standalone/common/promql/counter_reset_precision.sql @@ -0,0 +1,27 @@ +CREATE TABLE counter_reset_precision ( + ts TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + series STRING PRIMARY KEY +); + +INSERT INTO counter_reset_precision VALUES + (0, 10000000000000000.0, 'a'), + (30000, 1.0, 'a'), + (60000, 0.0, 'a'), + (90000, 1.0, 'a'), + (120000, 2.0, 'a'), + (150000, 3.0, 'a'), + (180000, 4.0, 'a'); + +-- The counter resets twice, at 30s from 1e16 and at 60s from 1.0, and 1e16 + 1.0 is 1e16 in +-- f64. Windows are two minutes wide and step by one sample, so every row here depends on the +-- 1.0 reset surviving. +-- +-- Summing the two corrections on their own drops the 1.0 and then cancels against the first +-- sample, so the 90s window reports no increase. Carrying that sum into the next window and +-- subtracting the 1e16 that left it reports half the increase, and the windows after that +-- subtract a reset the sum never held, so the correction turns negative and stays there for +-- the rest of the batch. +TQL EVAL (90, 180, '30s') increase(counter_reset_precision[2m]); + +DROP TABLE counter_reset_precision;