From 8174697c9691fe668ef36a939c8b34878e112a49 Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Thu, 10 Sep 2026 07:11:39 +0000 Subject: [PATCH] fix(promql): correct counter reset accumulation in rate windows (#9089) * fix(promql): correct counter reset accumulation in rate windows `prom_rate` and `prom_increase` reused the previous window's counter-reset correction when the next window slid forward by exactly one sample, adding the entering reset and subtracting the leaving one. Running a sum through addition and subtraction does not restore the earlier terms in f64: a large reset absorbs the smaller ones that must survive it, and an expired infinity leaves a NaN that no later window can clear. `prom_delta` shares the code but is not a counter function, so it never took that path. Index the reset positions of the value array once instead, and reduce each window over the resets it contains, in sample order. The result is bit-identical to scanning the window directly, so windows keep the direct reduction when they request fewer sample pairs than the input has. Also sweep the query step in the rate benchmarks: the cost of the reset correction depends on how much the windows overlap, which no existing case varied. Signed-off-by: Dennis Zhuang * test(promql): cover counter reset precision over adjacent rate windows The unit tests build the range windows directly, so they do not show that a plain PromQL range query produces the window layout that lost the correction. This case does: with a query step equal to the sample interval, `increase` over the second window returns 1.333 before the fix and 2.667 after it. Signed-off-by: Dennis Zhuang * perf(promql): advance the counter reset bounds instead of searching Locating a window's resets with two binary searches costs more than the reduction it replaces once a series resets often enough for the searches to get deep: on a 20k-sample counter resetting every 37 samples, stepping the windows by one sample was 2.7x slower than the previous code, against 1.2x for a counter that never resets. Windows normally advance, so walk the bounds forward from the previous window and only search when they move back. The cost then no longer depends on the reset density. Signed-off-by: Dennis Zhuang * perf(promql): cut the per-window cost of the counter reset index Two costs the index added showed up on a one-sample query step, where the removed fast path used to answer each window with two comparisons. Cache the two reset positions that bound the active slice. A window that only advanced and reached neither of them covers the same resets as the previous one, so the common case is four integer comparisons and no lookup at all. Stop summing the requested sample pairs once they exceed one pass over the values. The sum only decides which side of that comparison the input falls on, and a query with a short lookback and a long step settles it after a few windows instead of after every key. Together these take the one-sample step from 25-32% slower than the previous code down to 6-11%, measured as before / after / before to bound drift. No other step value regresses, and a ten-sample step stays about 88% faster. Signed-off-by: Dennis Zhuang * fix(promql): accumulate counter resets into the running result `prom_rate` and `prom_increase` summed a window's counter-reset corrections on their own and added that sum to `last - first`. Prometheus folds each reset into the running result instead, and so did this code before #7880. The two are not interchangeable in f64: over samples `[1e16, 1, 0, 1]` the isolated sum rounds `1e16 + 1.0` back to `1e16`, which then cancels against the first sample and reports no increase at all, where folding the resets in one at a time keeps the 1.0. Restore the original order. The reset index accumulates into the result the same way, so it still matches a direct scan of the window bit for bit, but a window's contribution can no longer be cached as a standalone value and is re-added from its own difference each time. The bounds are still cached, so a window that did not cross a reset skips the lookup, and one that holds no resets returns without touching the index at all. Signed-off-by: Dennis Zhuang * test(promql): note which reset boundaries the stride of one walks Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang (cherry picked from commit 0f625a7e922074692767aa9382619deda7aa2b32) Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/promql/benches/bench_range_fn.rs | 82 +++++- src/promql/src/functions/extrapolate_rate.rs | 258 ++++++++++++++++-- .../promql/counter_reset_precision.result | 43 +++ .../common/promql/counter_reset_precision.sql | 27 ++ 4 files changed, 369 insertions(+), 41 deletions(-) create mode 100644 tests/cases/standalone/common/promql/counter_reset_precision.result create mode 100644 tests/cases/standalone/common/promql/counter_reset_precision.sql 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;