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

fix(promql): correct counter reset accumulation in rate windows (#9089)

Browse Source 
* 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 <xzhuang@greptime.com>

* 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 <xzhuang@greptime.com>

* 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 <xzhuang@greptime.com>

* 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 <xzhuang@greptime.com>

* 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 <xzhuang@greptime.com>

* test(promql): note which reset boundaries the stride of one walks

Signed-off-by: Dennis Zhuang <xzhuang@greptime.com>

---------

Signed-off-by: Dennis Zhuang <xzhuang@greptime.com>
(cherry picked from commit 0f625a7e92)
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
dennis zhuang
2026-09-11 03:06:27 +08:00
committed by discord9
parent 9c301fe42c
commit 8174697c96
4 changed files with 369 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/benches/bench_range_fn.rs
+71 -11
View File
@@ -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<f64>,
eval_offset_ms: i64,
) -> (RangeArray, RangeArray, Arc<TimestampMillisecondArray>) {
@@ -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<usize> = (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<i64> = (0..num_windows)
.map(|i| timestamps[i + window_size as usize - 1] + eval_offset_ms)
let eval_ts: Vec<i64> = 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<f64> {
fn make_extrapolated_rate_input(
num_points: usize,
window_size: u32,
window_step: usize,
values: Vec<f64>,
eval_offset_ms: i64,
) -> Vec<ColumnarValue> {
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<ColumnarValue> {
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<ColumnarValue>
}
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);
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<ColumnarValue
}
fn make_predict_linear_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);
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<f64> = 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);
src/promql/src/functions/extrapolate_rate.rs
+228 -30
View File
@@ -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<const IS_COUNTER: bool, const IS_RATE: bool> ExtrapolatedRate<IS_COUNTER, I
let range_length = self.range_length;
let range_length_secs = range_length as f64 / 1000.0;
let mut counter_correction = 0.0;
let mut prev_offset = usize::MAX;
let mut prev_length = 0usize;
// Range windows normally overlap heavily, so scanning every one for resets costs far
// more than a single pass over the values. Index the reset positions once that is the
// cheaper side, and stop counting as soon as the requested pairs pass that budget,
// which heavy overlap does within the first few windows. A short lookback with a long
// step is the shape that never reaches it, and there the per-window scans do win.
let mut reset_index = if IS_COUNTER {
let budget = all_values.len().saturating_sub(1);
let mut scanned_pairs = 0usize;
keys.iter()
.any(|&key| {
scanned_pairs =
scanned_pairs.saturating_add(unpack(key).1.saturating_sub(1) as usize);
scanned_pairs > 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<const IS_COUNTER: bool, const IS_RATE: bool> ExtrapolatedRate<IS_COUNTER, I
if length < 2 {
result_builder.append_null();
prev_offset = usize::MAX;
continue;
}
@@ -211,32 +226,13 @@ impl<const IS_COUNTER: bool, const IS_RATE: bool> ExtrapolatedRate<IS_COUNTER, I
let first_value = all_values[offset];
let last_value = all_values[end - 1];
let result_value = if IS_COUNTER {
// Adjacent normalized windows usually slide forward by one sample. Reuse the
// previous window's accumulated reset correction and adjust only the dropped and
// newly added edges, falling back to a full scan when the layout changes.
if prev_offset != usize::MAX && offset == prev_offset + 1 && length == prev_length {
if all_values[prev_offset + 1] < all_values[prev_offset] {
counter_correction -= all_values[prev_offset];
}
if all_values[end - 1] < all_values[end - 2] {
counter_correction += all_values[end - 2];
}
} else {
counter_correction = 0.0;
for pair in all_values[offset..end].windows(2) {
if pair[1] < pair[0] {
counter_correction += pair[0];
}
}
}
last_value - first_value + counter_correction
} else {
last_value - first_value
};
prev_offset = offset;
prev_length = length;
let mut result_value = last_value - first_value;
if IS_COUNTER {
result_value = match &mut reset_index {
Some(reset_index) => 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<const IS_COUNTER: bool, const IS_RATE: bool> ExtrapolatedRate<IS_COUNTER, I
}
}
/// Adds the value preceding every counter reset in `values` to `result`, in sample order.
///
/// Prometheus accumulates the resets into the running result rather than summing them on
/// their own, and the two are not interchangeable in f64: a reset large enough to swallow a
/// later one in an isolated sum still leaves it visible once the first difference is folded
/// in first.
fn add_counter_resets(result: f64, values: &[f64]) -> 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<usize>,
/// Slice of `positions` covered by the last window.
active: Range<usize>,
/// That window, so the next one can tell whether it advanced.
previous: Range<usize>,
/// `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<usize> = (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::<Float64Array>()
.unwrap()
.iter()
.collect::<Vec<_>>()
};
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::<true, false>(
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<f64> = (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::<true, true>::new(5)
tests/cases/standalone/common/promql/counter_reset_precision.result
+43
View File
@@ -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
tests/cases/standalone/common/promql/counter_reset_precision.sql
+27
View File
@@ -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;