From 0256e4089b78428bf9e021fddfae2578dd27440c Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Fri, 18 Sep 2026 16:17:30 +0800 Subject: [PATCH] perf(promql): index valid sample positions for nullable range functions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a value backing array contains any NULL, every range window used to rescan its slots one by one (valid_window_bounds / per-window presence closures). Build an ascending valid_positions index once per batch and locate each window's samples with binary search instead. - extrapolate_rate.rs: nullable branch of rate/increase/delta; counter resets still accumulate in the original sample order over the located slice (add_counter_resets_between_samples now takes &[usize]). - aggr_over_time.rs: evaluate_presence nullable branch for count/last/absent/present_over_time. The null-free fast path (CounterResetIndex) is untouched. No changes to range_array, normalize, range_manipulate, or any existing expectation. Tests: 6 new tests compare every window against a pre-index oracle, including a seeded differential test over random null layouts and window shapes. 281 promql tests pass. Review reference: .e-agent/greptimedb-query-optimization-review-2026-09-17.md ยง4. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/promql/src/functions/aggr_over_time.rs | 352 +++++++++++- src/promql/src/functions/extrapolate_rate.rs | 548 +++++++++++++++++-- 2 files changed, 817 insertions(+), 83 deletions(-) diff --git a/src/promql/src/functions/aggr_over_time.rs b/src/promql/src/functions/aggr_over_time.rs index 2e7be0ab3b8..8f3b0703009 100644 --- a/src/promql/src/functions/aggr_over_time.rs +++ b/src/promql/src/functions/aggr_over_time.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::collections::VecDeque; +use std::ops::Range; use std::sync::Arc; use common_macro::range_fn; @@ -99,37 +100,46 @@ fn evaluate_presence( .as_any() .downcast_ref::() .unwrap(); - let evaluator: fn(&Float64Array, usize, usize) -> Option = if values.null_count() == 0 { + // Windows overlap heavily, so locate every sample once and let each window binary-search + // its own slice of the index instead of rescanning the slots it shares with its neighbours. + // Built only for a nullable batch, which is also the only branch that consults it. + let has_nulls = values.null_count() != 0; + let valid_positions: Vec = if has_nulls { + (0..values.len()) + .filter(|&index| values.is_valid(index)) + .collect() + } else { + Vec::new() + }; + let evaluator: fn(&Float64Array, &[usize], usize, usize) -> Option = if has_nulls { + // `valid_positions` only ever names samples, so the payloads behind them are the ones + // the null-free branch would have read, and no null slot is ever loaded. match operation { - PresenceEvaluator::Count => |_, _, length| (length != 0).then_some(length as f64), - PresenceEvaluator::Last => { - |values, offset, length| (length != 0).then(|| values.value(offset + length - 1)) - } - PresenceEvaluator::Absent => |_, _, length| (length == 0).then_some(1.0), - PresenceEvaluator::Present => |_, _, length| (length != 0).then_some(1.0), + PresenceEvaluator::Count => |_, samples, offset, length| { + let count = valid_slot_bounds(samples, offset, offset + length).len(); + (count != 0).then_some(count as f64) + }, + PresenceEvaluator::Last => |values, samples, offset, length| { + let bounds = valid_slot_bounds(samples, offset, offset + length); + (!bounds.is_empty()).then(|| values.value(samples[bounds.end - 1])) + }, + PresenceEvaluator::Absent => |_, samples, offset, length| { + valid_slot_bounds(samples, offset, offset + length) + .is_empty() + .then_some(1.0) + }, + PresenceEvaluator::Present => |_, samples, offset, length| { + (!valid_slot_bounds(samples, offset, offset + length).is_empty()).then_some(1.0) + }, } } else { match operation { - PresenceEvaluator::Count => |values, offset, length| { - let count = (offset..offset + length) - .filter(|&index| values.is_valid(index)) - .count(); - (count != 0).then_some(count as f64) - }, - PresenceEvaluator::Last => |values, offset, length| { - (offset..offset + length) - .rev() - .find(|&index| values.is_valid(index)) - .map(|index| values.value(index)) - }, - PresenceEvaluator::Absent => |values, offset, length| { - (!(offset..offset + length).any(|index| values.is_valid(index))).then_some(1.0) - }, - PresenceEvaluator::Present => |values, offset, length| { - (offset..offset + length) - .any(|index| values.is_valid(index)) - .then_some(1.0) - }, + PresenceEvaluator::Count => |_, _, _, length| (length != 0).then_some(length as f64), + PresenceEvaluator::Last => { + |values, _, offset, length| (length != 0).then(|| values.value(offset + length - 1)) + } + PresenceEvaluator::Absent => |_, _, _, length| (length == 0).then_some(1.0), + PresenceEvaluator::Present => |_, _, _, length| (length != 0).then_some(1.0), } }; @@ -142,7 +152,12 @@ fn evaluate_presence( "RangeArray's element {index} have different lengths in PromQL function {name}: array1={timestamp_length}, array2={value_length}" ))); } - result.push(evaluator(values, value_offset, value_length)); + result.push(evaluator( + values, + &valid_positions, + value_offset, + value_length, + )); } Ok(ColumnarValue::Array(Arc::new(Float64Array::from_iter( @@ -150,6 +165,15 @@ fn evaluate_presence( )))) } +/// The slice of `valid_positions` holding the samples inside `[offset, end)`: the positions are +/// ascending, so it is the half-open range between the first one at or past `offset` and the +/// first one at or past `end`. Empty when the window holds no sample. +fn valid_slot_bounds(valid_positions: &[usize], offset: usize, end: usize) -> Range { + let lo = valid_positions.partition_point(|&index| index < offset); + let hi = valid_positions.partition_point(|&index| index < end); + lo..hi +} + /// The average value of all points in the specified interval. #[range_fn( name = AvgOverTime, @@ -1289,6 +1313,278 @@ mod test { } } + /// `// oracle: pre-index implementation` + /// + /// The per-window scan the nullable presence evaluators performed before the valid slots + /// were indexed. + fn oracle_presence_output( + values: &Float64Array, + ranges: &[(u32, u32)], + operation: PresenceEvaluator, + ) -> Vec> { + ranges + .iter() + .map(|&(offset, length)| { + let offset = offset as usize; + let end = offset + length as usize; + match operation { + PresenceEvaluator::Count => { + let count = (offset..end) + .filter(|&index| values.is_valid(index)) + .count(); + (count != 0).then_some(count as f64) + } + PresenceEvaluator::Last => (offset..end) + .rev() + .find(|&index| values.is_valid(index)) + .map(|index| values.value(index)), + PresenceEvaluator::Absent => { + (!(offset..end).any(|index| values.is_valid(index))).then_some(1.0) + } + PresenceEvaluator::Present => (offset..end) + .any(|index| values.is_valid(index)) + .then_some(1.0), + } + }) + .collect() + } + + /// Payload kept under the null slots of the layouts below, so an evaluator that read the + /// padding instead of the samples would report a different value for `last_over_time`. + const NULL_PAYLOAD: f64 = -1234.5; + + /// The backing layouts the presence matrix covers: no nulls, a single null, a fixed spread + /// of scattered nulls, a run of missing samples, and valid slots holding the special values + /// that must not be mistaken for missing samples. + fn presence_layouts(len: usize) -> Vec<(&'static str, Vec>)> { + let base = |index: usize| Some((index % 7) as f64 + 1.0); + + let no_nulls = (0..len).map(base).collect::>(); + + let mut one_null = no_nulls.clone(); + one_null[5] = None; + + let mut scattered = no_nulls.clone(); + // Pairs, gaps, and both ends of the backing array. + for index in [0usize, 3, 8, 9, 13, len - 1] { + scattered[index] = None; + } + + let mut missing_run = no_nulls.clone(); + missing_run[11..=15].fill(None); + + let mut special = no_nulls.clone(); + special[0] = Some(f64::NAN); + special[2] = Some(f64::from_bits(0x7ff0_0000_0000_0002)); // stale marker + special[5] = Some(-0.0); + special[7] = Some(f64::INFINITY); + special[11] = Some(f64::NEG_INFINITY); + special[13] = None; + special[23] = None; + + vec![ + ("no nulls", no_nulls), + ("one null", one_null), + ("scattered nulls", scattered), + ("missing run", missing_run), + ("special payloads", special), + ] + } + + /// The window shapes the presence matrix covers, as `(offset, length)` pairs over a 24-slot + /// backing array. + fn presence_window_shapes() -> Vec<(&'static str, Vec<(u32, u32)>)> { + vec![ + // Heavily overlapping: neighbours share eleven of their twelve slots. + ("overlapping", (0..=12).map(|i| (i, 12)).collect()), + // Small windows, each overlapping its neighbour by two slots. + ("small", (0..21).map(|i| (i, 3)).collect()), + // Disjoint: no slot is shared with another window. + ("disjoint", (0..6).map(|i| (i * 4, 4)).collect()), + // Empty windows hold no sample at any offset. + ("empty", vec![(0, 0), (12, 0), (23, 0)]), + // Boundaries on both sides of a null run, and windows sitting inside one. + ( + "null run edges", + vec![(0, 1), (1, 1), (23, 1), (10, 5), (12, 4), (15, 2)], + ), + ] + } + + /// Builds a backing array that keeps `NULL_PAYLOAD` under every null slot. + fn nullable_backing_values(values: &[Option]) -> Float64Array { + Float64Array::new( + values + .iter() + .map(|value| value.unwrap_or(NULL_PAYLOAD)) + .collect::>() + .into(), + Some(NullBuffer::from_iter(values.iter().map(Option::is_some))), + ) + } + + /// Runs one presence range UDF over a backing array and a set of windows. + fn run_presence_udf( + udf: ScalarUDF, + values: &Float64Array, + ranges: &[(u32, u32)], + ) -> Vec> { + let timestamps = Arc::new(TimestampMillisecondArray::from_iter_values( + (0..values.len() as i64).map(|index| index * 1_000), + )); + let output = invoke_range_udf( + udf, + RangeArray::from_ranges(timestamps, ranges.iter().copied()).unwrap(), + RangeArray::from_ranges(Arc::new(values.clone()), ranges.iter().copied()).unwrap(), + ) + .unwrap(); + extract_array(&output) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect() + } + + /// Asserts two `Option` sequences agree bit for bit, naming the layout in failures. + fn assert_presence_matches_oracle(label: &str, actual: &[Option], oracle: &[Option]) { + assert_eq!(actual.len(), oracle.len(), "{label}"); + for (index, (actual, oracle)) in actual.iter().zip(oracle).enumerate() { + let equal = match (actual, oracle) { + (None, None) => true, + (Some(actual), Some(oracle)) => actual.to_bits() == oracle.to_bits(), + _ => false, + }; + assert!( + equal, + "{label}: window {index}: indexed {actual:?} != oracle {oracle:?}" + ); + } + } + + #[test] + fn presence_range_udfs_match_per_window_oracle_on_nullable_layouts() { + // Five backing layouts times five window shapes, each checked against the pre-index + // per-window scan for all four presence functions. + let len = 24; + // Constructors rather than `ScalarUDF` values, so the table is `Copy` and the loops + // below can walk it once per layout. + type PresenceCase = (&'static str, fn() -> ScalarUDF, PresenceEvaluator); + let cases: [PresenceCase; 4] = [ + ("count", CountOverTime::scalar_udf, PresenceEvaluator::Count), + ("last", LastOverTime::scalar_udf, PresenceEvaluator::Last), + ( + "absent", + AbsentOverTime::scalar_udf, + PresenceEvaluator::Absent, + ), + ( + "present", + PresentOverTime::scalar_udf, + PresenceEvaluator::Present, + ), + ]; + + for (layout, nulls) in presence_layouts(len) { + for (shape, ranges) in presence_window_shapes() { + let values = nullable_backing_values(&nulls); + for (name, build_udf, operation) in cases { + let oracle = oracle_presence_output(&values, &ranges, operation); + let actual = run_presence_udf(build_udf(), &values, &ranges); + assert_presence_matches_oracle( + &format!("{name}, {layout}, {shape}"), + &actual, + &oracle, + ); + } + } + } + } + + #[test] + fn presence_range_udfs_measure_samples_across_null_runs() { + // Slots 2..=5 are missing, slots 0, 1 and 6 are single samples, and one window is made + // only of nulls. + let nulls = [Some(1.0), Some(3.0), None, None, None, None, Some(9.0)]; + let ranges = vec![ + (0, 7), + (0, 2), + (2, 4), + (3, 4), + (1, 1), + (2, 1), + (6, 1), + (0, 0), + ]; + let values = nullable_backing_values(&nulls); + let cases = [ + ( + CountOverTime::scalar_udf(), + PresenceEvaluator::Count, + vec![ + Some(3.0), + Some(2.0), + None, + Some(1.0), + Some(1.0), + None, + Some(1.0), + None, + ], + ), + ( + LastOverTime::scalar_udf(), + PresenceEvaluator::Last, + vec![ + Some(9.0), + Some(3.0), + None, + Some(9.0), + Some(3.0), + None, + Some(9.0), + None, + ], + ), + ( + AbsentOverTime::scalar_udf(), + PresenceEvaluator::Absent, + vec![ + None, + None, + Some(1.0), + None, + None, + Some(1.0), + None, + Some(1.0), + ], + ), + ( + PresentOverTime::scalar_udf(), + PresenceEvaluator::Present, + vec![ + Some(1.0), + Some(1.0), + None, + Some(1.0), + Some(1.0), + None, + Some(1.0), + None, + ], + ), + ]; + + for (udf, operation, expected) in cases { + let actual = run_presence_udf(udf, &values, &ranges); + assert_option_bits(&actual, &expected); + let oracle = oracle_presence_output(&values, &ranges, operation); + assert_presence_matches_oracle("null runs", &actual, &oracle); + } + } + #[test] fn specialized_presence_range_udfs_preserve_errors_and_metadata() { use datafusion::arrow::array::{DictionaryArray, Int64Array}; diff --git a/src/promql/src/functions/extrapolate_rate.rs b/src/promql/src/functions/extrapolate_rate.rs index 263ae6337ba..4a375b9eb9e 100644 --- a/src/promql/src/functions/extrapolate_rate.rs +++ b/src/promql/src/functions/extrapolate_rate.rs @@ -190,6 +190,16 @@ impl ExtrapolatedRate 0; + // Windows overlap heavily, so locate every sample once and let each window find its own + // slice of the index, instead of rescanning the slots it shares with its neighbours. + // Built only for a nullable batch, which is also the only branch that consults it. + let valid_positions: Vec = if has_nulls { + (0..value_array.len()) + .filter(|&index| value_array.is_valid(index)) + .collect() + } else { + Vec::new() + }; let all_values = value_array.values(); let eval_ts = eval_ts_array.values(); @@ -222,16 +232,21 @@ impl ExtrapolatedRate bounds, - None => { - result_builder.append_null(); - continue; - } + let (first_index, last_index, sample_count, window_samples) = if has_nulls { + let bounds = valid_slot_bounds(&valid_positions, offset, end); + if bounds.is_empty() { + result_builder.append_null(); + continue; } + let samples = &valid_positions[bounds]; + ( + samples[0], + samples[samples.len() - 1], + samples.len(), + Some(samples), + ) } else { - (offset, end.saturating_sub(1), length) + (offset, end.saturating_sub(1), length, None) }; if sample_count < 2 { @@ -244,18 +259,14 @@ impl ExtrapolatedRate { + add_counter_resets_between_samples(result_value, all_values, samples) + } + None => 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]), - } + }, }; } @@ -407,50 +418,28 @@ impl<'a> CounterResetIndex<'a> { } } -/// Same additions [`add_counter_resets`] performs, over the samples in `[first, last]` instead -/// of over every slot. -fn add_counter_resets_between_samples( - result: f64, - values: &Float64Array, - first: usize, - last: usize, -) -> f64 { - let raw_values = values.values(); +/// Same additions [`add_counter_resets`] performs, over the samples in `samples` instead of over +/// every slot. `samples` holds the ascending slot positions of a window's samples, so the pairs +/// are visited in sample order, exactly as a per-slot scan of the window would visit them, and +/// the additions land on `result` in the same sequence. +fn add_counter_resets_between_samples(result: f64, values: &[f64], samples: &[usize]) -> f64 { let mut result = result; - let mut previous = raw_values[first]; - for index in first + 1..=last { - if values.is_null(index) { - continue; - } - let current = raw_values[index]; - if current < previous { + for pair in samples.windows(2) { + let previous = values[pair[0]]; + if values[pair[1]] < previous { result += previous; } - previous = current; } result } -/// Locates the samples inside `[offset, offset + length)`, returning the first and last -/// non-null index together with the number of non-null slots. Returns `None` when the -/// window holds no sample. -fn valid_window_bounds( - values: &Float64Array, - offset: usize, - length: usize, -) -> Option<(usize, usize, usize)> { - let mut first = None; - let mut last = 0; - let mut count = 0; - for index in offset..offset + length { - if values.is_null(index) { - continue; - } - first.get_or_insert(index); - last = index; - count += 1; - } - first.map(|first| (first, last, count)) +/// The slice of `valid_positions` holding the samples inside `[offset, end)`: the positions are +/// ascending, so it is the half-open range between the first one at or past `offset` and the +/// first one at or past `end`. Empty when the window holds no sample. +fn valid_slot_bounds(valid_positions: &[usize], offset: usize, end: usize) -> Range { + let lo = valid_positions.partition_point(|&index| index < offset); + let hi = valid_positions.partition_point(|&index| index < end); + lo..hi } fn extract_eval_timestamps( @@ -781,6 +770,455 @@ mod test { assert_eq!(output, vec![Some(7.5)]); } + /// `// oracle: pre-index implementation` + /// + /// The window computation as it stood before the valid-slot index existed: every window + /// rescanned its own slots for the samples and for the counter resets. Same math as + /// [`ExtrapolatedRate::calc`], reached the long way. + fn oracle_rate_output( + timestamps: &[i64], + values: &Float64Array, + ranges: &[(u32, u32)], + eval_timestamps: &[i64], + range_length: i64, + ) -> Vec> { + let raw_values = values.values(); + let range_length_secs = range_length as f64 / 1000.0; + let mut output = Vec::with_capacity(ranges.len()); + for (window, &(offset, length)) in ranges.iter().enumerate() { + let offset = offset as usize; + let end = offset + length as usize; + + // oracle: the per-window scan for the first and last sample and the sample count. + let mut first = None; + let mut last = 0; + let mut sample_count = 0; + for slot in offset..end { + if values.is_null(slot) { + continue; + } + first.get_or_insert(slot); + last = slot; + sample_count += 1; + } + let first_index = match first { + Some(first) => first, + None => { + output.push(None); + continue; + } + }; + if sample_count < 2 { + output.push(None); + continue; + } + let last_index = last; + + let first_value = raw_values[first_index]; + let last_value = raw_values[last_index]; + + let mut result_value = last_value - first_value; + if IS_COUNTER { + // oracle: the per-slot walk between the two samples for the resets. + let mut previous = raw_values[first_index]; + for slot in first_index + 1..=last_index { + if values.is_null(slot) { + continue; + } + let current = raw_values[slot]; + if current < previous { + result_value += previous; + } + previous = current; + } + } + + let first_ts = timestamps[first_index]; + let last_ts = timestamps[last_index]; + let range_end = eval_timestamps[window]; + let range_start = range_end - range_length; + let sampled_interval_ms = (last_ts - first_ts) as f64; + let average_interval_ms = sampled_interval_ms / (sample_count - 1) as f64; + let mut duration_to_start_ms = (first_ts - range_start) as f64; + let mut duration_to_end_ms = (range_end - last_ts) as f64; + let extrapolation_threshold = average_interval_ms * 1.1; + if duration_to_start_ms >= extrapolation_threshold { + duration_to_start_ms = average_interval_ms / 2.0; + } + if IS_COUNTER && result_value > 0.0 && first_value >= 0.0 { + let duration_to_zero = sampled_interval_ms * (first_value / result_value); + if duration_to_zero < duration_to_start_ms { + duration_to_start_ms = duration_to_zero; + } + } + if duration_to_end_ms >= extrapolation_threshold { + duration_to_end_ms = average_interval_ms / 2.0; + } + let mut factor = if sampled_interval_ms == 0.0 { + 1.0 + } else { + (sampled_interval_ms + duration_to_start_ms + duration_to_end_ms) + / sampled_interval_ms + }; + if IS_RATE { + factor /= range_length_secs; + } + + output.push(Some(result_value * factor)); + } + output + } + + /// Runs the UDF and the per-window oracle over one layout and asserts they agree window by + /// window. `label` names the layout in failure messages. + fn assert_rate_matches_oracle( + label: &str, + timestamps: Vec, + values: Arc, + ranges: Vec<(u32, u32)>, + eval_timestamps: Vec, + range_length: i64, + ) { + let expected = oracle_rate_output::( + ×tamps, + &values, + &ranges, + &eval_timestamps, + range_length, + ); + let actual = nullable_rate_runner::( + timestamps, + values, + ranges, + eval_timestamps, + range_length, + ); + + assert_eq!(actual.len(), expected.len(), "{label}"); + for (index, (actual, expected)) in actual.iter().zip(&expected).enumerate() { + let equal = match (actual, expected) { + (None, None) => true, + (Some(actual), Some(expected)) => { + actual.to_bits() == expected.to_bits() || (actual.is_nan() && expected.is_nan()) + } + _ => false, + }; + assert!( + equal, + "{label}: window {index}: indexed {actual:?} != oracle {expected:?}" + ); + } + } + + /// [`assert_rate_matches_oracle`] for `0` = rate, `1` = increase and `2` = delta. + fn assert_rate_variant_matches_oracle( + variant: usize, + label: &str, + timestamps: Vec, + values: Arc, + ranges: Vec<(u32, u32)>, + eval_timestamps: Vec, + range_length: i64, + ) { + match variant { + 0 => assert_rate_matches_oracle::( + label, + timestamps, + values, + ranges, + eval_timestamps, + range_length, + ), + 1 => assert_rate_matches_oracle::( + label, + timestamps, + values, + ranges, + eval_timestamps, + range_length, + ), + 2 => assert_rate_matches_oracle::( + label, + timestamps, + values, + ranges, + eval_timestamps, + range_length, + ), + _ => unreachable!("only rate, increase and delta exist"), + } + } + + /// Payload kept under the null slots of the layouts below, so a window that read the + /// padding instead of the samples would report a different value. + const NULL_PAYLOAD: f64 = -1234.5; + + /// The four layouts the index/oracle matrix covers: no nulls, a single interior null, + /// a fixed spread of scattered nulls and a run of missing samples. All of them are the + /// same resetting counter, so only the null layout differs. + fn matrix_inputs(len: usize) -> Vec<(&'static str, Vec>)> { + let no_nulls = (0..len) + .map(|index| Some((index % 7) as f64 + 1.0)) + .collect::>(); + + let mut one_null = no_nulls.clone(); + one_null[5] = None; + + let mut scattered = no_nulls.clone(); + // Pairs, gaps, and both ends of the backing array. + for index in [0usize, 3, 8, 9, 13, len - 1] { + scattered[index] = None; + } + + let mut missing_run = no_nulls.clone(); + missing_run[11..=15].fill(None); + + vec![ + ("no nulls", no_nulls), + ("one null", one_null), + ("scattered nulls", scattered), + ("missing run", missing_run), + ] + } + + /// The three window shapes the matrix covers, as `(offset, length)` pairs over the same + /// backing array. + fn matrix_window_shapes(len: usize) -> Vec<(&'static str, Vec<(u32, u32)>)> { + vec![ + // Heavily overlapping: neighbours share eleven of their twelve slots. + ( + "overlapping", + (0..=(len as u32 - 12)).map(|i| (i, 12)).collect(), + ), + // Small windows, each overlapping its neighbour by two slots. + ("small", (0..(len as u32 - 3)).map(|i| (i, 3)).collect()), + // Disjoint: no slot is shared with another window. + ( + "disjoint", + (0..(len as u32 / 4)).map(|i| (i * 4, 4)).collect(), + ), + ] + } + + /// Evaluation timestamps just past the last slot of every window, which is where a range + /// vector ends in practice. + fn eval_timestamps_after(timestamps: &[i64], ranges: &[(u32, u32)], after_ms: i64) -> Vec { + ranges + .iter() + .map(|&(offset, length)| { + // An empty window has no last slot, so anchor it on its offset and stay inside + // the timestamp vector. + let last = + (offset as usize + length.saturating_sub(1) as usize).min(timestamps.len() - 1); + timestamps[last] + after_ms + }) + .collect() + } + + #[test] + fn nullable_rate_windows_match_per_window_oracle() { + // Four layouts times three window shapes times the three functions, every window of + // every batch checked against the pre-index implementation. + let len = 24; + let timestamps: Vec = (0..len) + .map(|index| index as i64 * 30_000 + (index % 5) as i64 * 1_000) + .collect(); + let range_length = 60_000; + + for (input, nulls) in matrix_inputs(len) { + for (shape, ranges) in matrix_window_shapes(len) { + let eval_timestamps = eval_timestamps_after(×tamps, &ranges, 5_000); + for (variant, name) in ["rate", "increase", "delta"].into_iter().enumerate() { + assert_rate_variant_matches_oracle( + variant, + &format!("{name}, {input}, {shape}"), + timestamps.clone(), + values_with_nulls(nulls.clone(), NULL_PAYLOAD), + ranges.clone(), + eval_timestamps.clone(), + range_length, + ); + } + } + } + } + + #[test] + fn nullable_rate_windows_match_per_window_oracle_on_seeded_layouts() { + // Fixed seeds over random nulls, resets, special values and window layouts, disjoint and + // overlapping, including windows that start, end or sit entirely inside a null run. + for seed in [ + 0x1f2e_3d4c_5b6a_7988, + 0x0bad_c0de_dead_beef, + 0x5eed_cafe_f00d_baad, + ] { + let mut prng = TinyPrng(seed); + let len = 48usize; + let timestamps: Vec = (0..len) + .map(|index| index as i64 * 30_000 + (index % 7) as i64 * 1_000) + .collect(); + + let mut raw = Vec::with_capacity(len); + let mut present = Vec::with_capacity(len); + for index in 0..len { + let value = match prng.next_index(10) { + 0 => f64::NAN, + 1 => f64::INFINITY, + 2 => f64::NEG_INFINITY, + 3 => 0.0, + 4 => -0.0, + 5 => f64::from_bits(0x7ff0_0000_0000_0002), // stale marker + _ => { + (prng.next_index(9) as f64 + 1.0) * if index % 3 == 0 { -1.0 } else { 1.0 } + } + }; + let is_present = prng.next_index(4) != 0; + // Null slots keep a payload that would change the result if it were read. + raw.push(if is_present { + value + } else { + -1.0e6 - index as f64 + }); + present.push(is_present); + } + let values = Arc::new(Float64Array::new( + raw.into(), + Some(NullBuffer::from_iter(present)), + )); + + let ranges: Vec<(u32, u32)> = (0..32) + .map(|_| { + let length = prng.next_index(12) as u32; + let offset = prng.next_index(len - length as usize + 1) as u32; + (offset, length) + }) + .collect(); + let eval_timestamps: Vec = ranges + .iter() + .map(|&(offset, length)| { + // An empty window has no last slot, so anchor it on its offset. + let last = (offset as usize + length.saturating_sub(1) as usize) + .min(timestamps.len() - 1); + timestamps[last] + prng.next_index(3) as i64 * 30_000 + }) + .collect(); + + let label = format!("seed {seed:#x}"); + for variant in 0..3 { + assert_rate_variant_matches_oracle( + variant, + &label, + timestamps.clone(), + values.clone(), + ranges.clone(), + eval_timestamps.clone(), + 120_000, + ); + } + } + } + + #[test] + fn nullable_rate_handles_windows_that_cut_through_nulls() { + // Slots 1..=3 and 6 are missing. The windows cover a single sample, a window made only + // of nulls, a window whose first and last slots are null, and one whose samples are + // entirely inside it. + let timestamps = vec![0, 1000, 2000, 3000, 4000, 5000, 6000, 7000]; + let values = || { + values_with_nulls( + vec![ + Some(1.0), + None, + None, + None, + Some(4.0), + Some(5.0), + None, + Some(8.0), + ], + NULL_PAYLOAD, + ) + }; + let ranges = vec![(0, 1), (1, 3), (3, 3), (3, 5), (2, 6), (6, 1), (7, 1)]; + let eval_timestamps = eval_timestamps_after(×tamps, &ranges, 0); + + // increase: a single sample is not enough for a delta, so (0, 1), (6, 1) and (7, 1) are + // null, and so is the all-null window (1, 3). (3, 3) holds 4.0@4000 and 5.0@5000, whose + // increase of 1.0 over a 1000ms interval is extrapolated by half an interval to 1.5. + // (3, 5) and (2, 6) hold the same three samples, 4.0@4000, 5.0@5000 and 8.0@7000, whose + // increase of 4.0 over two intervals is extrapolated by half an interval to 5.0. Reading + // the null padding as a sample would charge a reset against it and change both values. + let output = nullable_rate_runner::( + timestamps.clone(), + values(), + ranges.clone(), + eval_timestamps.clone(), + 60_000, + ); + assert_eq!( + output, + vec![None, None, Some(1.5), Some(5.0), Some(5.0), None, None] + ); + + assert_rate_variant_matches_oracle( + 1, + "increase, windows cutting through nulls", + timestamps, + values(), + ranges, + eval_timestamps, + 60_000, + ); + } + + #[test] + fn nullable_rate_keeps_special_values_out_of_the_null_branch() { + // NaN, infinities and stale markers are valid samples, not missing ones: the windows + // keep counting them, and only the null slots stay out of the sample set. The result + // is NaN because the samples are, and the oracle agrees. + let timestamps: Vec = (0..6).map(|index| index as i64 * 1000).collect(); + let values = || { + values_with_nulls( + vec![ + None, + Some(f64::from_bits(0x7ff0_0000_0000_0002)), + Some(f64::NAN), + None, + Some(f64::INFINITY), + None, + ], + NULL_PAYLOAD, + ) + }; + let ranges = vec![(0, 6), (1, 3), (3, 3), (2, 1)]; + let eval_timestamps = eval_timestamps_after(×tamps, &ranges, 0); + + let output = nullable_rate_runner::( + timestamps.clone(), + values(), + ranges.clone(), + eval_timestamps.clone(), + 60_000, + ); + // The first two windows hold at least two valid samples, so they report a value, NaN + // because their samples are; the other two hold a single sample and stay null. + // Counting a NaN or a stale marker as missing would flip the first two to null. + assert!(output[0].is_some_and(|value| value.is_nan())); + assert!(output[1].is_some_and(|value| value.is_nan())); + assert_eq!(output[2], None); + assert_eq!(output[3], None); + + assert_rate_variant_matches_oracle( + 2, + "delta, special values", + timestamps, + values(), + ranges, + eval_timestamps, + 60_000, + ); + } + /// Line-by-line port of Prometheus `extrapolatedRate` (promql/functions.go), float path /// without start timestamps. Kept as a second implementation so that the order of the /// threshold clamp and the counter zero-snap stays pinned to the upstream one.