fix(promql): propagate histogram aggregation errors other than mixed schemas

Also drop the unobservable 0.0 + count arm in iterate.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
Dennis Zhuang
2026-10-18 00:45:14 +08:00
parent ee1ba3923c
commit 2b01ac6aad
3 changed files with 73 additions and 28 deletions
+22 -26
View File
@@ -128,33 +128,31 @@ fn aggregate_point<'a>(
if incompatible {
continue;
}
// The only addition error Go's `handleAggregationError` absorbs; others propagate.
if current.uses_custom_buckets() != h.uses_custom_buckets() {
incompatible = true;
continue;
}
match op {
HistogramAggregate::Sum => match current.kahan_add(&h, compensation.take()) {
Ok((c, _)) => compensation = Some(c),
Err(_) => incompatible = true,
},
HistogramAggregate::Sum => {
let (c, _) = current.kahan_add(&h, compensation.take())?;
compensation = Some(c);
}
HistogramAggregate::Avg => {
if mean.is_none() {
let mut sum = current.clone();
match sum.kahan_add(&h, compensation.clone()) {
Ok((c, _)) if !sum.has_overflow() => {
*current = sum;
compensation = Some(c);
continue;
}
Ok(_) => {
let mut started = current.clone();
started.div(count - 1.0);
if let Some(c) = compensation.as_mut() {
c.div(count - 1.0);
}
mean = Some(started);
}
Err(_) => {
incompatible = true;
continue;
}
let (c, _) = sum.kahan_add(&h, compensation.clone())?;
if !sum.has_overflow() {
*current = sum;
compensation = Some(c);
continue;
}
let mut started = current.clone();
started.div(count - 1.0);
if let Some(c) = compensation.as_mut() {
c.div(count - 1.0);
}
mean = Some(started);
}
// Go switches to an incremental mean once the plain sum overflows.
let q = (count - 1.0) / count;
@@ -165,10 +163,8 @@ fn aggregate_point<'a>(
to_add.div(count);
let mean = mean.as_mut().expect("set above");
mean.mul(q);
match mean.kahan_add(&to_add, compensation.take()) {
Ok((c, _)) => compensation = Some(c),
Err(_) => incompatible = true,
}
let (c, _) = mean.kahan_add(&to_add, compensation.take())?;
compensation = Some(c);
}
}
}
@@ -679,8 +679,6 @@ fn iterate(side: &Buckets, schema: i32, target: i32, threshold: f64) -> Result<B
let target_idx = target_index(idx, schema, target);
match out.last_mut() {
Some(last) if schema != target && last.0 == target_idx => last.1 += count,
// A merged bucket starts its sum from zero.
_ if schema != target => out.push((target_idx, 0.0 + count)),
_ => out.push((target_idx, count)),
}
}
+51
View File
@@ -691,3 +691,54 @@ fn widened_zero_threshold_uses_go_bounds() {
));
assert!(!current.detect_reset(&previous).unwrap());
}
/// Only mixing exponential and custom buckets ends a group without a result; any other
/// addition error, such as an invalid layout, fails the aggregation.
#[test]
fn aggregation_propagates_errors_other_than_mixed_schemas() {
let grid = Arc::new(
datatypes::series::time::EvalGrid::try_new(
datatypes::series::time::GridSpec::try_new(0, 1_000, 1).unwrap(),
)
.unwrap(),
);
let schema = Arc::new(SeriesSchema {
fields: vec![FieldSpec {
name: Arc::from("value"),
}]
.into(),
eval_grid: Some(grid),
});
let group = |texts: [&str; 2]| -> Vec<Series> {
texts
.iter()
.enumerate()
.map(|(host, text)| {
let labels = Labels::from([("host".to_string(), host.to_string())]);
let histograms = BTreeMap::from([(0, text.to_string())]);
mixed_series(&schema, &labels, &BTreeMap::new(), &histograms).unwrap()
})
.collect()
};
let exponential = "0/0/0/3ff0000000000000/3ff0000000000000/0/1_1/3ff0000000000000///";
let custom =
"-53/0/0/3ff0000000000000/3ff0000000000000/0/0_1/3ff0000000000000///3ff0000000000000";
let invalid =
"9/3ff0000000000000/0/3ff0000000000000/3ff0000000000000/0/1_1/3ff0000000000000///";
for op in [HistogramAggregate::Sum, HistogramAggregate::Avg] {
let mixed = aggregate_histograms(
op,
&schema,
&group([exponential, custom]),
metric_of(&Labels::new()),
);
assert!(mixed.unwrap().is_none());
let invalid = aggregate_histograms(
op,
&schema,
&group([invalid, invalid]),
metric_of(&Labels::new()),
);
assert!(invalid.is_err());
}
}