chore: merge feat/narthex-window-eval-at into feat/narthex-new-eval

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

# Conflicts:
#	src/promql/src/range_eval/tests.rs
This commit is contained in:
Dennis Zhuang
2026-10-17 22:34:09 +08:00
3 changed files with 82 additions and 7 deletions
+13 -3
View File
@@ -79,8 +79,12 @@ pub struct RangeEvaluator {
series: Option<SeriesState>,
fields: Vec<FieldSamples>,
row: Vec<Option<f64>>,
drops_metric_name: bool,
}
/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name.
const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"];
struct SeriesState {
/// `None` for evaluated input, which arrives as one complete series.
primary_key: Option<Buffer>,
@@ -144,13 +148,15 @@ impl RangeEvaluator {
}
}
}
Self::with_kernel(
let mut evaluator = Self::with_kernel(
input,
grid,
window,
function.kernel(fold_offset),
step_invariant,
)
)?;
evaluator.drops_metric_name = function.drops_metric_name();
Ok(evaluator)
}
/// `step_invariant` tells whether an `@`-anchored result holds at every grid point.
@@ -192,6 +198,7 @@ impl RangeEvaluator {
.map(|_| FieldSamples::default())
.collect(),
row: Vec::with_capacity(input.fields.len()),
drops_metric_name: false,
})
}
@@ -259,12 +266,15 @@ impl RangeEvaluator {
fn finish_series(&mut self) -> Result<Option<Series>> {
self.evaluate(None)?;
let state = self.series.take().expect("set by the caller");
let mut state = self.series.take().expect("set by the caller");
for field in &mut self.fields {
field.timestamps.clear();
field.values.clear();
field.head = 0;
}
if self.drops_metric_name {
state.metric.remove_all(&METADATA_LABELS);
}
self.output.finish(state.metric)
}
+6
View File
@@ -60,6 +60,12 @@ pub enum RangeFunction {
}
impl RangeFunction {
/// Whether the result drops the metric name, as Prometheus does for every range function
/// except `last_over_time` and `first_over_time`.
pub(crate) fn drops_metric_name(self) -> bool {
!matches!(self, Self::LastOverTime)
}
/// Returns the kernel computing this function over windows read with `offset_ms`.
pub(crate) fn kernel(self, offset_ms: i64) -> WindowKernel {
match self {
+63 -4
View File
@@ -110,6 +110,15 @@ fn per_point_probe_evaluator(
}
/// Builds a chunk whose timestamp and value arrays are slices at a nonzero offset.
/// The metric of a result of `function` over the series of `metric(host)`.
fn output_metric(function: RangeFunction, host: u64) -> MetricName {
let mut metric = metric(host);
if function.drops_metric_name() {
metric.remove("__name__");
}
metric
}
fn chunk(rows: &[Row], fields: usize, host: u64, is_last: bool) -> RawSeriesChunk {
let padding = 3;
let timestamps: Vec<_> = (0..padding)
@@ -467,7 +476,7 @@ fn functions_match_range_udfs() {
for piece in random_chunks(&mut rng, &rows, fields, host) {
result = evaluator.push(&piece).unwrap();
}
assert_series_eq(result.as_ref(), &expected, &metric(host));
assert_series_eq(result.as_ref(), &expected, &output_metric(function, host));
}
}
}
@@ -744,7 +753,7 @@ fn subquery_series_match_references() {
assert_series_eq(
evaluator.push_series(&series).unwrap().as_ref(),
&expected,
&metric(seed),
&output_metric(function, seed),
);
}
}
@@ -891,7 +900,12 @@ fn range_functions_match_prometheus() {
.push(&chunk(&rows[split..], 1, 0, true))
.unwrap()
.or(result);
assert_series_eq_any_nan(result.as_ref(), &expected, &metric(0), line);
assert_series_eq_any_nan(
result.as_ref(),
&expected,
&output_metric(function, 0),
line,
);
checked += expected_points.len();
}
assert!(checked > 2_000, "{checked} points checked");
@@ -917,6 +931,47 @@ fn rejects_double_exponential_smoothing_factors_outside_the_unit_interval() {
}
}
/// Range functions drop the metric name and the other metadata labels, except
/// `last_over_time`, as Prometheus does.
#[test]
fn range_functions_drop_metadata_labels() {
let labels = |names: &[&str]| -> Vec<Label> {
names
.iter()
.map(|name| Label {
name: Arc::from(*name),
value: Arc::from("v"),
})
.collect()
};
let input = MetricName::try_new(
Some(Arc::from("m")),
labels(&["__type__", "__unit__", "host"]),
)
.unwrap();
let window = RangeWindow {
range_ms: 10_000,
offset_ms: 0,
anchor_ms: None,
};
let rows: Vec<Row> = vec![(1_000, vec![Some(1.0)]), (2_000, vec![Some(2.0)])];
for (function, expected) in [
(
RangeFunction::Rate,
MetricName::try_new(None, labels(&["host"])).unwrap(),
),
(RangeFunction::LastOverTime, input.clone()),
] {
let mut chunk = chunk(&rows, 1, 0, true);
chunk.metric = Some(input.clone());
let mut evaluator =
RangeEvaluator::try_new(&raw_schema(1), grid(2_000, 1_000, 1), window, function)
.unwrap();
let series = evaluator.push(&chunk).unwrap().unwrap();
assert_eq!(series.metric, expected, "{function:?}");
}
}
/// A leading NaN is replaced by the next sample in `ts_of_min/max_over_time`, so the timestamp
/// moves past it even though no comparison with NaN holds.
#[test]
@@ -937,6 +992,10 @@ fn ts_of_extremum_moves_past_a_leading_nan() {
RangeEvaluator::try_new(&raw_schema(1), grid(3_000, 1_000, 1), window, function)
.unwrap();
let series = evaluator.push(&chunk(&rows, 1, 0, true)).unwrap().unwrap();
assert_series_eq(Some(&series), &vec![vec![Some(expected)]], &metric(0));
assert_series_eq(
Some(&series),
&vec![vec![Some(expected)]],
&output_metric(function, 0),
);
}
}