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(),