From 3fca17791cb2ca571fd3dca1407d8116e3ec85c7 Mon Sep 17 00:00:00 2001 From: discord9 Date: Mon, 19 Oct 2026 17:47:10 +0800 Subject: [PATCH] perf(promql): index counter resets per field instead of rescanning the window (#9708) * perf(promql): index counter resets per field instead of rescanning the window The fused evaluator rescanned every window for counter resets, once per grid point. Record the resets of each field as the samples are ingested instead, keyed by an absolute sequence number, and read the ones a window covers with a binary search. Only `rate` and `increase` correct counters, so only they maintain the index. The binary search runs over the field's retained reset index rather than over the window's samples: its input is the remaining slice of resets, so its cost follows the number of retained resets, not the window length. A drop across an intervening native histogram sample is recorded as a reset of two samples that are not neighbours. That is unobservable: a window holding both floats also holds the histogram between them, so it is a mixed window, whose counter functions report nothing; and a window starting after the earlier float excludes a reset on its first sample. Verified with `cargo nextest run -p promql` (453 passed) and `cargo clippy -p promql --all-targets --all-features -D warnings`. Tests assert both that the indexed results match the brute-force scan and that the index is what ran. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor(promql): drop per-field clearing the evaluator never needs Two review findings, both about code that does nothing. `finish_series` cleared every field after `evaluate(None)` returned. That loop is redundant: `evaluate` ends with either every point evaluated -- so `head` moved to the end and `advance` dropped the samples -- or no field holding a sample, and either way the fields are empty. It also did not clear `histograms`, which made it read as though it should. Removing it keeps the suite green, and `push` already documents that an evaluator is not reused after an error. `FieldSamples::clear` existed only for that loop. `overlapping_window_spans_chunks` used a `sum` kernel and never touched resets. Windows spanning chunks are already covered by `evaluator_matches_brute_force_reference`, which chunks randomly, and no mutation was caught only by this test. Its `sum` helper went with it. Signed-off-by: discord9 Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 --- src/promql/src/functions.rs | 1 - src/promql/src/range_eval.rs | 99 +++++++++- src/promql/src/range_eval/function.rs | 21 ++- src/promql/src/range_eval/tests.rs | 257 +++++++++++++++++++++++++- 4 files changed, 360 insertions(+), 18 deletions(-) diff --git a/src/promql/src/functions.rs b/src/promql/src/functions.rs index 7b7e84bd967..925d2fce89c 100644 --- a/src/promql/src/functions.rs +++ b/src/promql/src/functions.rs @@ -49,7 +49,6 @@ use datatypes::arrow::datatypes::{DataType, Int64Type}; pub use deriv::Deriv; pub use double_exponential_smoothing::DoubleExponentialSmoothing; pub(crate) use edge_count::{EdgeKind, count_edges}; -pub(crate) use extrapolate_rate::add_counter_resets; pub use extrapolate_rate::{Delta, Increase, Rate}; pub use idelta::IDelta; pub(crate) use idelta::instant_delta; diff --git a/src/promql/src/range_eval.rs b/src/promql/src/range_eval.rs index a323dc97e2e..b809fe2284d 100644 --- a/src/promql/src/range_eval.rs +++ b/src/promql/src/range_eval.rs @@ -53,6 +53,9 @@ pub(crate) struct FloatWindow<'a> { /// Sample timestamps shifted by the offset onto the evaluation timeline; non-empty. pub timestamps: &'a [i64], pub values: &'a [f64], + /// The counter resets this window contains, ascending by sample. Empty for every function + /// that does not correct counters and for every window of a series that never reset. + pub counter_resets: &'a [CounterReset], /// Evaluation timestamp `T`. `timestamps` lie in `(T - range_ms, T]`, except for an /// `@`-anchored `predict_linear`, which evaluates the first point's window at every `T`. pub eval_ts: i64, @@ -121,6 +124,88 @@ struct FieldSamples { /// Native histogram samples, kept apart from the floats as Prometheus does. histograms: Vec<(i64, FloatHistogram)>, histogram_head: usize, + /// Sequence number of `values[0]`. Absolute and monotonic across the series this evaluator + /// evaluates, so a reset keeps its position when `head` samples are dropped instead of every + /// index being renumbered. An empty field holds the number of samples already dropped, which + /// keeps positions increasing when the next series starts. + base: u64, + /// Resets between consecutive retained samples, ascending by position: exactly the ones a + /// window over these samples can still contain. Maintained only for the counter-correcting + /// functions, and released as the samples they reach back to are dropped. + resets: Vec, + /// Whether `resets` is maintained for this field. + index_resets: bool, +} + +/// One counter reset: a sample that dropped below the sample before it. +#[derive(Clone, Copy, Debug)] +pub(crate) struct CounterReset { + /// Position of the later sample, as an absolute sequence number. Its predecessor is + /// `position - 1`, so a window holding the samples `[start, end)` contains this reset + /// exactly when `start < position < end`. + pub position: u64, + /// The value the reset adds back: the sample at `position - 1`, which is inside the window + /// whenever the reset is. + pub previous_value: f64, +} + +impl FieldSamples { + fn new(index_resets: bool) -> Self { + Self { + index_resets, + ..Self::default() + } + } + + /// Appends one valid sample, recording a counter reset when it dropped below the sample + /// before it. + /// + /// "Before it" means the previous entry of `values`, which leaves native histogram samples + /// out. A drop across an intervening histogram is therefore recorded as a reset of two + /// samples that are not neighbours. That is unobservable: a window holding both floats also + /// holds the histogram between them, so it is a mixed window, whose counter functions report + /// nothing; and a window starting after the earlier float excludes a reset on its first + /// sample. Should the mixed-window rules ever start reading the float resets, this has to + /// compare timestamps instead. + fn push(&mut self, timestamp: i64, value: f64) { + if self.index_resets + && let Some(&previous_value) = self.values.last() + && value < previous_value + { + self.resets.push(CounterReset { + position: self.base + self.values.len() as u64, + previous_value, + }); + } + self.timestamps.push(timestamp); + self.values.push(value); + } + + /// Drops the samples before `head` and the resets whose predecessor is one of them. + fn advance(&mut self) { + let head = std::mem::take(&mut self.head); + if head == 0 { + return; + } + self.base += head as u64; + self.timestamps.drain(..head); + self.values.drain(..head); + // A reset at `base` would add back the last dropped sample, so it goes with it. + let dropped = self + .resets + .partition_point(|reset| reset.position <= self.base); + self.resets.drain(..dropped); + } + + /// The resets inside the window holding `len` samples from `head`. + fn resets_in(&self, len: usize) -> &[CounterReset] { + let start = self.base + self.head as u64; + let end = start + len as u64; + // Left-open: a reset at `start` would add back a sample behind the window. + let first = self.resets.partition_point(|reset| reset.position <= start); + let count = self.resets[first..].partition_point(|reset| reset.position < end); + &self.resets[first..first + count] + } } impl RangeEvaluator { @@ -161,6 +246,7 @@ impl RangeEvaluator { window, function.kernel(fold_offset), step_invariant, + function.indexes_counter_resets(), )?; evaluator.drops_metric_name = function.drops_metric_name(); evaluator.window_error = window_error; @@ -169,12 +255,15 @@ impl RangeEvaluator { } /// `step_invariant` tells whether an `@`-anchored result holds at every grid point. + /// `index_resets` tells whether the kernel corrects counters, which is the only case where + /// the per-field reset index is worth maintaining. pub(crate) fn with_kernel( input: &SeriesSchema, grid: Arc, mut window: RangeWindow, kernel: WindowKernel, step_invariant: bool, + index_resets: bool, ) -> Result { if window.range_ms <= 0 { return Err(DataFusionError::Plan(format!( @@ -205,7 +294,7 @@ impl RangeEvaluator { fields: input .fields .iter() - .map(|_| FieldSamples::default()) + .map(|_| FieldSamples::new(index_resets)) .collect(), row: Vec::with_capacity(input.fields.len()), drops_metric_name: false, @@ -382,8 +471,7 @@ impl RangeEvaluator { match sample { SampleRef::Missing => {} SampleRef::Float(value) => { - field.timestamps.push(shifted); - field.values.push(*value); + field.push(shifted, *value); } SampleRef::Histogram(view) => { let mut h = FloatHistogram::try_from_view(view)?; @@ -482,9 +570,7 @@ impl RangeEvaluator { field.head = field.timestamps.len(); field.histogram_head = field.histograms.len(); } - field.timestamps.drain(..field.head); - field.values.drain(..field.head); - field.head = 0; + field.advance(); field.histograms.drain(..field.histogram_head); field.histogram_head = 0; } @@ -526,6 +612,7 @@ impl RowEvaluator<'_> { let floats = (len > 0).then(|| FloatWindow { timestamps: ×tamps[..len], values: &field.values[field.head..field.head + len], + counter_resets: field.resets_in(len), eval_ts, range_ms: self.range_ms, }); diff --git a/src/promql/src/range_eval/function.rs b/src/promql/src/range_eval/function.rs index 8c126c3654e..29eacebba4c 100644 --- a/src/promql/src/range_eval/function.rs +++ b/src/promql/src/range_eval/function.rs @@ -18,9 +18,7 @@ //! operation order. Arithmetic that GreptimeDB's range UDFs compute the same way is shared with //! them. -use crate::functions::{ - EdgeKind, add_counter_resets, count_edges, extremum, instant_delta, kahan_inc, -}; +use crate::functions::{EdgeKind, count_edges, extremum, instant_delta, kahan_inc}; use crate::go_sort; use crate::range_eval::{FloatWindow, WindowKernel}; @@ -69,6 +67,16 @@ impl RangeFunction { !matches!(self, Self::LastOverTime | Self::FirstOverTime) } + /// Whether the evaluator has to index counter resets for this function. + /// + /// This must hold for exactly the kernels that read [`FloatWindow::counter_resets`], which is + /// the same as saying `IS_COUNTER` in `extrapolated_rate`: `rate` and `increase`. A kernel + /// that corrects counters without the index being maintained would silently drop every + /// correction, so a new counter function has to be added here as well. + pub(crate) fn indexes_counter_resets(self) -> bool { + matches!(self, Self::Rate | Self::Increase) + } + /// Returns the kernel computing this function over windows read with `offset_ms`. pub(crate) fn kernel(self, offset_ms: i64) -> WindowKernel { match self { @@ -225,6 +233,7 @@ fn extrapolated_rate( let FloatWindow { timestamps, values, + counter_resets, eval_ts, range_ms, } = window; @@ -232,7 +241,11 @@ fn extrapolated_rate( let first = values[0]; let mut result = values[samples_minus_one] - first; if IS_COUNTER { - result = add_counter_resets(result, values); + // The same additions in the same order as scanning the window for resets, reached + // through the evaluator's per-field reset index instead of the whole window. + result = counter_resets + .iter() + .fold(result, |result, reset| result + reset.previous_value); } let zero_point = (IS_COUNTER && result > 0.0 && first >= 0.0).then(|| first / result); let factor = extrapolation_factor( diff --git a/src/promql/src/range_eval/tests.rs b/src/promql/src/range_eval/tests.rs index cb8e8b38589..66f38734f98 100644 --- a/src/promql/src/range_eval/tests.rs +++ b/src/promql/src/range_eval/tests.rs @@ -14,7 +14,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::ops::Range; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS; use datafusion::arrow::array::{ArrayRef, Float64Array, TimestampMillisecondArray, UInt32Array}; @@ -36,7 +36,10 @@ use datatypes::series::{Label, MetricName}; use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; -use super::{FloatWindow, RangeEvaluator, RangeFunction, RangeWindow, WindowKernel, subquery_grid}; +use super::{ + CounterReset, FloatWindow, RangeEvaluator, RangeFunction, RangeWindow, WindowKernel, + subquery_grid, +}; use crate::functions::{ Changes, CountOverTime, IDelta, LastOverTime, MaxOverTime, MinOverTime, PresentOverTime, Resets, }; @@ -67,9 +70,25 @@ fn probe(window: FloatWindow<'_>) -> Option { mix(*timestamp as u64); mix(value.to_bits()); } + // The reference builds these by scanning the window, so hashing them turns every random + // differential test into a check of the evaluator's reset index as well. + for reset in window.counter_resets { + mix(reset.previous_value.to_bits()); + } Some((hash >> 11) as f64) } +/// The resets of a window by scanning it, the way the kernels used to before the index. +fn scanned_resets(values: &[f64]) -> Vec { + (1..values.len()) + .filter(|&index| values[index] < values[index - 1]) + .map(|index| CounterReset { + position: index as u64, + previous_value: values[index - 1], + }) + .collect() +} + fn raw_schema(fields: usize) -> SeriesSchema { SeriesSchema { fields: (0..fields) @@ -99,7 +118,7 @@ fn metric(host: u64) -> MetricName { fn probe_evaluator(fields: usize, grid: Arc, window: RangeWindow) -> RangeEvaluator { let kernel: WindowKernel = Box::new(probe); - RangeEvaluator::with_kernel(&raw_schema(fields), grid, window, kernel, true).unwrap() + RangeEvaluator::with_kernel(&raw_schema(fields), grid, window, kernel, true, true).unwrap() } /// A probe evaluator whose anchored result is evaluated at every point, like `predict_linear`. @@ -109,7 +128,7 @@ fn per_point_probe_evaluator( window: RangeWindow, ) -> RangeEvaluator { let kernel: WindowKernel = Box::new(probe); - RangeEvaluator::with_kernel(&raw_schema(fields), grid, window, kernel, false).unwrap() + RangeEvaluator::with_kernel(&raw_schema(fields), grid, window, kernel, false, true).unwrap() } /// The metric of a result of `function` over the series of `metric(host)`. @@ -293,10 +312,12 @@ fn reference( .iter() .filter_map(|(timestamp, row)| Some((*timestamp, row[field]?))) .unzip(); + let counter_resets = scanned_resets(&values); (!timestamps.is_empty()).then(|| { probe(FloatWindow { timestamps: ×tamps, values: &values, + counter_resets: &counter_resets, eval_ts: if step_invariant { window_end(grid, window, eval_ts) } else { @@ -687,6 +708,222 @@ fn retains_only_samples_later_windows_need() { } } +/// A counter whose every window contains resets, so the index holds entries the way a real +/// counter does, still releases them as the samples they reach back to are dropped. +#[test] +fn reset_index_is_released_with_the_samples_it_reaches_back_to() { + let window = RangeWindow { + range_ms: 50, + offset_ms: 0, + anchor_ms: None, + }; + let mut evaluator = probe_evaluator(1, grid(0, 10, 1_000), window); + // A sawtooth that resets every 10 samples. + let rows: Vec = (0..10_000) + .map(|i| (i, vec![Some((i % 10) as f64)])) + .collect(); + for (index, piece) in rows.chunks(100).enumerate() { + evaluator + .push(&chunk(piece, 1, 0, (index + 1) * 100 == rows.len())) + .unwrap(); + // Resets are released with the samples behind the next window, so they stay bounded by + // the same window plus one grid step the samples do, one reset per 10 samples. + assert!( + evaluator.fields[0].resets.len() <= 6, + "{} resets after {} chunks: the index kept history no window can reach", + evaluator.fields[0].resets.len(), + index + 1 + ); + } +} + +/// The reset index, not a rescan of the window, decides which samples a window corrects. The +/// samples are laid out so each rule is visible point by point: a reset is the first sample of a +/// later chunk, a reset's predecessor sits exactly on a window's left edge and must not count, +/// and a reset enters a window whole. The reported resets are compared instead of the function's +/// output so a wrong position cannot hide inside a float's last bits. +#[test] +fn counter_resets_follow_the_window_and_survive_chunking() { + // A counter that drops at index 3 (20 -> 5) and again at index 6 (25 -> 7), one sample per + // 10 s. + let counter = [0.0, 10.0, 20.0, 5.0, 15.0, 25.0, 7.0, 17.0, 27.0, 37.0]; + // A field that only rises, so its windows never correct anything. + let rising = [0.0, 10.0, 20.0, 30.0, 40.0, 50.0, 60.0, 70.0, 80.0, 90.0]; + let rows: Vec = (0..10) + .map(|index| { + ( + index as i64 * 10_000, + vec![Some(counter[index]), Some(rising[index])], + ) + }) + .collect(); + let grid = grid(30_000, 10_000, 4); + let window = RangeWindow { + range_ms: 20_000, + offset_ms: 0, + anchor_ms: None, + }; + // Every chunk starts on a sample an earlier window corrected, so the index has to reach + // across the chunk boundary. + let pieces = [(0, 3), (3, 6), (6, 10)]; + + let recorded = Arc::new(Mutex::new(Vec::new())); + let mut evaluator = { + let sink = recorded.clone(); + let record = move |window: FloatWindow<'_>| { + sink.lock().unwrap().push(( + window.timestamps.to_vec(), + window + .counter_resets + .iter() + .map(|reset| reset.previous_value) + .collect::>(), + )); + Some(window.values[0]) + }; + let mut evaluator = RangeEvaluator::with_kernel( + &raw_schema(2), + grid.clone(), + window, + Box::new(record), + true, + true, + ) + .unwrap(); + for (start, end) in pieces { + let last = end == rows.len(); + let result = evaluator + .push(&chunk(&rows[start..end], 2, 0, last)) + .unwrap(); + assert_eq!( + result.is_some(), + last, + "the series must finish on its last chunk" + ); + } + evaluator + }; + + // Windows (10 s, 30 s], (20 s, 40 s], (30 s, 50 s] and (40 s, 60 s]. + // The first holds 20 and 5, so the reset at index 3 corrects it. The second's left edge is + // the sample before that reset, which a left-open window excludes, so the same reset must not + // count there. The fourth holds 25 and 7, corrected by the reset at index 6. The second field + // never corrects anything. + let expected = [ + (vec![20_000, 30_000], vec![20.0], vec![]), + (vec![30_000, 40_000], vec![], vec![]), + (vec![40_000, 50_000], vec![], vec![]), + (vec![50_000, 60_000], vec![25.0], vec![]), + ]; + let first = recorded.lock().unwrap().clone(); + assert_eq!(first.len(), 8, "four grid points, two fields"); + for (index, (pair, (timestamps, counter_resets, rising_resets))) in + first.chunks(2).zip(expected.iter().cloned()).enumerate() + { + assert_eq!( + pair[0], + (timestamps.clone(), counter_resets), + "field 0 at grid point {index}" + ); + assert_eq!( + pair[1], + (timestamps, rising_resets), + "field 1 at grid point {index}" + ); + } + + // The same samples pushed as a second series through the same evaluator: no buffer, cursor or + // index state of the first series may survive, so the second series reports exactly the same + // windows and resets. + recorded.lock().unwrap().clear(); + let mut result = None; + for (start, end) in pieces { + assert!(result.is_none(), "a result before the last chunk"); + result = evaluator + .push(&chunk(&rows[start..end], 2, 1, end == rows.len())) + .unwrap(); + } + assert_eq!(*recorded.lock().unwrap(), recorded_of(&expected)); + evaluator.finish().unwrap(); +} + +/// The per-call recordings `counter_resets_follow_the_window_and_survive_chunking` expects, in +/// kernel-call order: field 0 then field 1 for each grid point. +fn recorded_of(expected: &[(Vec, Vec, Vec)]) -> Vec<(Vec, Vec)> { + expected + .iter() + .flat_map(|(timestamps, counter, rising)| { + [ + (timestamps.clone(), counter.clone()), + (timestamps.clone(), rising.clone()), + ] + }) + .collect() +} + +/// Missing samples shift which neighbours a reset sits between, and the shift differs per +/// field. The resets each window reports are checked position by position instead of through a +/// function's output, so a field paying for another field's resets shows up directly. +#[test] +fn nulls_change_the_reset_positions_of_each_field_independently() { + let rows: Vec = vec![ + (0, vec![Some(0.0), Some(0.0)]), + (10_000, vec![Some(20.0), Some(20.0)]), + (20_000, vec![None, Some(15.0)]), + (30_000, vec![Some(5.0), Some(5.0)]), + (40_000, vec![Some(5.0), Some(5.0)]), + ]; + let grid = grid(0, 10_000, 5); + // Every window is wider than the series, so each grid point sees every sample up to it. + let window = RangeWindow { + range_ms: 100_000, + offset_ms: 0, + anchor_ms: None, + }; + + // One entry per kernel call, in field order within each grid point. + let reported = Arc::new(Mutex::new(Vec::new())); + { + let sink = reported.clone(); + let record = move |window: FloatWindow<'_>| { + sink.lock().unwrap().push( + window + .counter_resets + .iter() + .map(|reset| reset.previous_value) + .collect::>(), + ); + Some(window.values[0]) + }; + let mut evaluator = + RangeEvaluator::with_kernel(&raw_schema(2), grid, window, Box::new(record), true, true) + .unwrap(); + for (start, end) in [(0, 2), (2, 5)] { + evaluator + .push(&chunk(&rows[start..end], 2, 0, end == rows.len())) + .unwrap(); + } + evaluator.finish().unwrap(); + } + + // Field 0 never sees the missing 15, so 20 is followed directly by 5 and it resets once. + // Field 1 keeps the 15 and resets twice: 20 -> 15 and 15 -> 5. Both fields report their own + // resets even though they share a window and a chunk. + let reported = reported.lock().unwrap().clone(); + assert_eq!(reported.len(), 10, "five grid points, two fields"); + let expected = [ + (vec![], vec![]), + (vec![], vec![]), + (vec![], vec![20.0]), + (vec![20.0], vec![20.0, 15.0]), + (vec![20.0], vec![20.0, 15.0]), + ]; + for (index, (pair, (field0, field1))) in reported.chunks(2).zip(expected).enumerate() { + assert_eq!(pair[0], field0, "field 0 at grid point {index}"); + assert_eq!(pair[1], field1, "field 1 at grid point {index}"); + } +} + /// A random evaluated series on `grid`; values may carry the stale bit pattern, which computed /// input keeps as an ordinary NaN. fn random_evaluated( @@ -737,9 +974,15 @@ fn subquery_series_match_references() { let (Some(series), rows) = random_evaluated(&mut rng, &inner, fields, seed) else { continue; }; - let mut probe = - RangeEvaluator::with_kernel(&schema, outer.clone(), window, Box::new(probe), true) - .unwrap(); + let mut probe = RangeEvaluator::with_kernel( + &schema, + outer.clone(), + window, + Box::new(probe), + true, + true, + ) + .unwrap(); let expected = reference(&rows, &outer, window, fields, false, true); assert_series_eq( probe.push_series(&series).unwrap().as_ref(),