rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-13 22:36:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(promql): select the extrema evaluator from batch averages

Browse Source 
Reading the window shape off the first two windows misreads the batch.
RangeManipulate starts a series at max(query start, first aligned sample),
so a series that begins inside the query range gets a first window covering
roughly one step, and a window covering no sample at all is emitted as
(0, 0). Either one closed the gate for the whole batch, including the
one-hour window at a 15s step that candidate reuse was written for.

Compare the batch averages instead: at least 32 samples per window, and a
step advancing at most a quarter of that. Uniform batches select exactly as
before, so the thresholds keep the meaning they were measured with.

The 32-sample rule had also moved most of the benchmark and query-regression
shapes onto the rescan, including the case built to measure reset and
rebuild. Widen those windows to 40 samples, add a step at the selection
boundary, and add an end-to-end case with 40-sample windows advancing 5.

Signed-off-by: Dennis Zhuang <xzhuang@greptime.com>
This commit is contained in:
Dennis Zhuang
2026-09-16 20:06:20 +08:00
parent cacc7f4e54
commit 86aa8b7204
5 changed files with 225 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/benches/bench_range_fn.rs
+20 -10
View File
@@ -753,11 +753,13 @@ fn bench_extrema_functions(c: &mut Criterion) {
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)
// Cases meant to reuse candidates use 40-sample windows: the UDF rescans batches
// whose windows average fewer than 32 samples.
let mut backwards_ranges = (0..=num_points - 40)
.step_by(5)
.map(|offset| (offset as u32, 20))
.map(|offset| (offset as u32, 40))
.collect::<Vec<_>>();
backwards_ranges.extend((0..=512).step_by(5).map(|offset| (offset as u32, 20)));
backwards_ranges.extend((0..=512).step_by(5).map(|offset| (offset as u32, 40)));
// The last two controls use explicit ranges instead of a regular window/step sweep.
let cases = vec![
@@ -768,16 +770,16 @@ fn bench_extrema_functions(c: &mut Criterion) {
.collect::<Vec<_>>(),
),
(
"w20_step1",
(0..=num_points - 20)
.map(|offset| (offset as u32, 20))
"w40_step1",
(0..=num_points - 40)
.map(|offset| (offset as u32, 40))
.collect::<Vec<_>>(),
),
(
"w20_step5",
(0..=num_points - 20)
"w40_step5",
(0..=num_points - 40)
.step_by(5)
.map(|offset| (offset as u32, 20))
.map(|offset| (offset as u32, 40))
.collect::<Vec<_>>(),
),
(
@@ -786,6 +788,14 @@ fn bench_extrema_functions(c: &mut Criterion) {
.map(|offset| (offset as u32, 240))
.collect::<Vec<_>>(),
),
// A quarter of the window is the widest step that still reuses candidates.
(
"w240_step60",
(0..=num_points - 240)
.step_by(60)
.map(|offset| (offset as u32, 240))
.collect::<Vec<_>>(),
),
(
"w240_step240",
(0..=num_points - 240)
@@ -793,7 +803,7 @@ fn bench_extrema_functions(c: &mut Criterion) {
.map(|offset| (offset as u32, 240))
.collect::<Vec<_>>(),
),
("backwards_reset_rebuild_w20_step5", backwards_ranges),
("backwards_reset_rebuild_w40_step5", backwards_ranges),
(
"low_coverage_full_backing_w4",
vec![
src/promql/src/functions/aggr_over_time.rs
+42 -21
View File
@@ -236,29 +236,42 @@ fn min_max_input_type() -> Vec<DataType> {
/// Batches with fewer windows than this have too little repeated work to reclaim.
const MIN_SLIDING_WINDOWS: usize = 4;
/// Below this window length the per-sample bookkeeping is comparable to the scan
/// it would replace.
const MIN_SLIDING_WINDOW_LENGTH: u32 = 32;
/// Below this average window length the per-sample bookkeeping is comparable to the
/// scan it would replace.
const MIN_SLIDING_WINDOW_LENGTH: u64 = 32;
/// Reuse is taken only when a step advances at most this fraction of the window.
const MAX_SLIDING_STEP_FRACTION: u32 = 4;
const MAX_SLIDING_STEP_FRACTION: u64 = 4;
/// Whether reusing candidates across windows is expected to beat rescanning each one.
///
/// Reuse pays off in proportion to how much consecutive windows overlap, and loses to
/// a plain scan on wide windows that barely overlap: maintaining the deque then costs
/// more than the rescan it replaces. `RangeManipulate` emits one window length and one
/// step for a whole batch, so the first two windows decide for all of them. A wrong
/// guess costs time, not correctness — both paths return the same bits.
/// more than the rescan it replaces.
///
/// The shape is read from the whole batch rather than from its leading windows.
/// `RangeManipulate` holds the window duration and the evaluation step fixed, but the
/// sample counts still vary: a series that starts inside the query range gets a first
/// window covering roughly one step, and a window covering no sample at all is emitted
/// as `(0, 0)`. Averages survive both; the first two windows do not.
///
/// A wrong answer costs time, not correctness — both evaluators return the same bits.
fn reuses_candidates(window_keys: &[i64]) -> bool {
if window_keys.len() < MIN_SLIDING_WINDOWS {
return false;
}
let (first_offset, first_length) = unpack(window_keys[0]);
let (second_offset, _) = unpack(window_keys[1]);
first_length >= MIN_SLIDING_WINDOW_LENGTH
&& second_offset >= first_offset
&& second_offset - first_offset <= first_length / MAX_SLIDING_STEP_FRACTION
let windows = window_keys.len() as u64;
let total_length: u64 = window_keys.iter().map(|&key| unpack(key).1 as u64).sum();
// What reuse skips re-reading: the distance the left bound travels over the batch.
let first_offset = unpack(window_keys[0]).0;
let last_offset = unpack(window_keys[window_keys.len() - 1]).0;
let total_advance = u64::from(last_offset.saturating_sub(first_offset));
// `total_length / windows >= MIN_SLIDING_WINDOW_LENGTH` and
// `total_advance / (windows - 1) <= (total_length / windows) / MAX_SLIDING_STEP_FRACTION`,
// cross-multiplied to keep the averages exact.
total_length >= windows * MIN_SLIDING_WINDOW_LENGTH
&& total_advance * MAX_SLIDING_STEP_FRACTION * windows <= total_length * (windows - 1)
}
fn is_better(value: f64, current: f64, is_min: bool) -> bool {
@@ -915,27 +928,35 @@ mod test {
(8, 32),
(16, 32)
])));
// One sample too short.
// One sample per window too short.
assert!(!reuses_candidates(&window_keys(&[
(0, 31),
(7, 31),
(14, 31),
(21, 31)
])));
// One sample too far apart.
// One sample per step too far apart.
assert!(!reuses_candidates(&window_keys(&[
(0, 32),
(9, 32),
(18, 32),
(27, 32)
])));
// Windows that start by retreating.
assert!(!reuses_candidates(&window_keys(&[
(8, 32),
(0, 32),
(16, 32),
(24, 32)
])));
}
#[test]
fn sliding_evaluator_survives_an_unrepresentative_leading_window() {
// `RangeManipulate` gives a series that starts inside the query range a first
// window covering roughly one step, and emits a window covering no sample at
// all as (0, 0). Reading either one as the batch shape hides the overlap that
// the twenty windows behind it do have.
let overlapping = (0..20u32).map(|index| (index * 8, 40));
for leading in [(0, 1), (0, 0)] {
let ranges = std::iter::once(leading)
.chain(overlapping.clone())
.collect::<Vec<_>>();
assert!(reuses_candidates(&window_keys(&ranges)));
}
}
#[test]
tests/cases/standalone/common/tql/aggr_over_time.result
+101
View File
@@ -394,3 +394,104 @@ drop table moving_extrema;
Affected Rows: 0
-- Dense sliding min/max: 40-sample windows advancing 5 samples, the shape the UDF
-- reuses candidates for rather than rescanning. Values follow the timestamp, so each
-- window's minimum is its oldest sample and its maximum is its newest, and both
-- extrema expire on every step.
create table dense_extrema (ts timestamp_s time index, val double, series string primary key);
Affected Rows: 0
insert into dense_extrema values (0, 0::double, 'dense');
Affected Rows: 1
-- Doubling eight times gives 256 rows at a one-second cadence, with val = ts.
insert into dense_extrema select to_unixtime(ts) + 1, val + 1, series from dense_extrema;
Affected Rows: 1
insert into dense_extrema select to_unixtime(ts) + 2, val + 2, series from dense_extrema;
Affected Rows: 2
insert into dense_extrema select to_unixtime(ts) + 4, val + 4, series from dense_extrema;
Affected Rows: 4
insert into dense_extrema select to_unixtime(ts) + 8, val + 8, series from dense_extrema;
Affected Rows: 8
insert into dense_extrema select to_unixtime(ts) + 16, val + 16, series from dense_extrema;
Affected Rows: 16
insert into dense_extrema select to_unixtime(ts) + 32, val + 32, series from dense_extrema;
Affected Rows: 32
insert into dense_extrema select to_unixtime(ts) + 64, val + 64, series from dense_extrema;
Affected Rows: 64
insert into dense_extrema select to_unixtime(ts) + 128, val + 128, series from dense_extrema;
Affected Rows: 128
select count(*), min(val), max(val) from dense_extrema;
+----------+------------------------+------------------------+
| count(*) | min(dense_extrema.val) | max(dense_extrema.val) |
+----------+------------------------+------------------------+
| 256 | 0.0 | 255.0 |
+----------+------------------------+------------------------+
-- {series="dense"} 61 66 71 76 81 86 91 96 101 106 111 116 121
-- SQLNESS SORT_RESULT 2 1
tql eval (100, 160, '5s') min_over_time(dense_extrema[40s]);
+---------------------+----------------------------------+--------+
| ts | prom_min_over_time(ts_range,val) | series |
+---------------------+----------------------------------+--------+
| 1970-01-01T00:01:40 | 61.0 | dense |
| 1970-01-01T00:01:45 | 66.0 | dense |
| 1970-01-01T00:01:50 | 71.0 | dense |
| 1970-01-01T00:01:55 | 76.0 | dense |
| 1970-01-01T00:02:00 | 81.0 | dense |
| 1970-01-01T00:02:05 | 86.0 | dense |
| 1970-01-01T00:02:10 | 91.0 | dense |
| 1970-01-01T00:02:15 | 96.0 | dense |
| 1970-01-01T00:02:20 | 101.0 | dense |
| 1970-01-01T00:02:25 | 106.0 | dense |
| 1970-01-01T00:02:30 | 111.0 | dense |
| 1970-01-01T00:02:35 | 116.0 | dense |
| 1970-01-01T00:02:40 | 121.0 | dense |
+---------------------+----------------------------------+--------+
-- {series="dense"} 100 105 110 115 120 125 130 135 140 145 150 155 160
-- SQLNESS SORT_RESULT 2 1
tql eval (100, 160, '5s') max_over_time(dense_extrema[40s]);
+---------------------+----------------------------------+--------+
| ts | prom_max_over_time(ts_range,val) | series |
+---------------------+----------------------------------+--------+
| 1970-01-01T00:01:40 | 100.0 | dense |
| 1970-01-01T00:01:45 | 105.0 | dense |
| 1970-01-01T00:01:50 | 110.0 | dense |
| 1970-01-01T00:01:55 | 115.0 | dense |
| 1970-01-01T00:02:00 | 120.0 | dense |
| 1970-01-01T00:02:05 | 125.0 | dense |
| 1970-01-01T00:02:10 | 130.0 | dense |
| 1970-01-01T00:02:15 | 135.0 | dense |
| 1970-01-01T00:02:20 | 140.0 | dense |
| 1970-01-01T00:02:25 | 145.0 | dense |
| 1970-01-01T00:02:30 | 150.0 | dense |
| 1970-01-01T00:02:35 | 155.0 | dense |
| 1970-01-01T00:02:40 | 160.0 | dense |
+---------------------+----------------------------------+--------+
drop table dense_extrema;
Affected Rows: 0
tests/cases/standalone/common/tql/aggr_over_time.sql
+37
View File
@@ -209,3 +209,40 @@ tql eval (20, 100, '10s') min_over_time(moving_extrema[30s]);
tql eval (20, 100, '10s') max_over_time(moving_extrema[30s]);
drop table moving_extrema;
-- Dense sliding min/max: 40-sample windows advancing 5 samples, the shape the UDF
-- reuses candidates for rather than rescanning. Values follow the timestamp, so each
-- window's minimum is its oldest sample and its maximum is its newest, and both
-- extrema expire on every step.
create table dense_extrema (ts timestamp_s time index, val double, series string primary key);
insert into dense_extrema values (0, 0::double, 'dense');
-- Doubling eight times gives 256 rows at a one-second cadence, with val = ts.
insert into dense_extrema select to_unixtime(ts) + 1, val + 1, series from dense_extrema;
insert into dense_extrema select to_unixtime(ts) + 2, val + 2, series from dense_extrema;
insert into dense_extrema select to_unixtime(ts) + 4, val + 4, series from dense_extrema;
insert into dense_extrema select to_unixtime(ts) + 8, val + 8, series from dense_extrema;
insert into dense_extrema select to_unixtime(ts) + 16, val + 16, series from dense_extrema;
insert into dense_extrema select to_unixtime(ts) + 32, val + 32, series from dense_extrema;
insert into dense_extrema select to_unixtime(ts) + 64, val + 64, series from dense_extrema;
insert into dense_extrema select to_unixtime(ts) + 128, val + 128, series from dense_extrema;
select count(*), min(val), max(val) from dense_extrema;
-- {series="dense"} 61 66 71 76 81 86 91 96 101 106 111 116 121
-- SQLNESS SORT_RESULT 2 1
tql eval (100, 160, '5s') min_over_time(dense_extrema[40s]);
-- {series="dense"} 100 105 110 115 120 125 130 135 140 145 150 155 160
-- SQLNESS SORT_RESULT 2 1
tql eval (100, 160, '5s') max_over_time(dense_extrema[40s]);
drop table dense_extrema;
tests/perf/query_cases/promql_range_boundary/case.toml
+25 -1
View File
@@ -99,7 +99,9 @@ iterations = 9
[scenario.queries.thresholds]
max_candidate_latency_regression_pct = 20
# Sliding extrema cover small, representative, and wide overlapping windows.
# Sliding extrema cover small, representative, and wide overlapping windows. At this
# cadence the 1m and 5m windows hold about 5 and 21 samples, below the 32-sample
# average the UDF requires before it reuses candidates, so they measure the rescan.
[[scenario.queries]]
name = "min_over_time_1m"
kind = "tql"
@@ -160,6 +162,28 @@ iterations = 9
[scenario.queries.thresholds]
max_candidate_latency_regression_pct = 20
# A ten-minute step keeps the one-hour window on the reuse path without the
# single-sample step the two cases above use.
[[scenario.queries]]
name = "min_over_time_1h_step10m"
kind = "tql"
query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '10m') 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_step10m"
kind = "tql"
query = "TQL ANALYZE VERBOSE (1704070800, 1704078000, '10m') 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"