From b37f47022b509a4564573b048fae3a5331cfab3d Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Sun, 18 Oct 2026 01:22:08 +0800 Subject: [PATCH] refactor(promql): share metadata labels, error mapping and Kahan increments across Narthex operators Signed-off-by: Dennis Zhuang --- src/promql/src/functions.rs | 10 ++++++++++ src/promql/src/grid_series.rs | 5 ++++- src/promql/src/histogram_eval.rs | 6 +++--- src/promql/src/histogram_eval/aggregate.rs | 5 +---- src/promql/src/histogram_eval/native.rs | 15 +-------------- src/promql/src/map_time.rs | 9 +-------- src/promql/src/range_eval.rs | 5 +---- src/promql/src/range_eval/function.rs | 18 +++--------------- src/promql/src/select.rs | 6 +----- src/promql/src/vector_match.rs | 5 +---- 10 files changed, 26 insertions(+), 58 deletions(-) diff --git a/src/promql/src/functions.rs b/src/promql/src/functions.rs index 5662323e137..7b7e84bd967 100644 --- a/src/promql/src/functions.rs +++ b/src/promql/src/functions.rs @@ -149,6 +149,16 @@ pub(crate) fn compensated_sum_inc(inc: f64, sum: f64, mut compensation: f64) -> (new_sum, compensation) } +/// Prometheus `kahansum.Inc`: [`compensated_sum_inc`] that also drops the compensation once the +/// sum is infinite. +pub(crate) fn kahan_inc(inc: f64, sum: f64, c: f64) -> (f64, f64) { + let t = sum + inc; + if t.is_infinite() { + return (t, 0.0); + } + compensated_sum_inc(inc, sum, c) +} + /// linear_regression performs a least-square linear regression analysis on the /// times and values. It return the slope and intercept based on times and values. /// Prometheus's implementation: diff --git a/src/promql/src/grid_series.rs b/src/promql/src/grid_series.rs index 0307c802f65..0a5d1bdedce 100644 --- a/src/promql/src/grid_series.rs +++ b/src/promql/src/grid_series.rs @@ -142,6 +142,9 @@ impl GridSeriesBuilder { } } +/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name. +pub(crate) const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"]; + pub(crate) type LabelSet = Vec<(Arc, Arc)>; /// The label set Prometheus compares, which has no empty values. @@ -260,6 +263,6 @@ fn read_row(series: &Series, row: usize, values: &mut [Option]) -> Result DataFusionError { +pub(crate) fn external(error: datatypes::error::Error) -> DataFusionError { DataFusionError::External(Box::new(error)) } diff --git a/src/promql/src/histogram_eval.rs b/src/promql/src/histogram_eval.rs index 19319110b23..14fe167129c 100644 --- a/src/promql/src/histogram_eval.rs +++ b/src/promql/src/histogram_eval.rs @@ -40,11 +40,11 @@ use datatypes::series::time::TimeAxis; pub use aggregate::{HistogramAggregate, aggregate_histograms}; use crate::go_sort; -use crate::grid_series::{GridSeriesBuilder, LabelSet, label_set, merge_same_labelset}; +use crate::grid_series::{ + GridSeriesBuilder, LabelSet, METADATA_LABELS, label_set, merge_same_labelset, +}; const BUCKET_LABEL: &str = "le"; -/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name. -const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"]; /// `histogram_quantile` or `histogram_fraction` with its constant arguments. #[derive(Debug, Clone, Copy, PartialEq)] diff --git a/src/promql/src/histogram_eval/aggregate.rs b/src/promql/src/histogram_eval/aggregate.rs index 6653c5a40b6..dc6acbf46d4 100644 --- a/src/promql/src/histogram_eval/aggregate.rs +++ b/src/promql/src/histogram_eval/aggregate.rs @@ -26,6 +26,7 @@ use datatypes::series::time::{GridRange, TimeAxis}; use super::float_histogram::FloatHistogram; use super::grid_range; +use crate::grid_series::external; /// A native histogram aggregation. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -199,7 +200,3 @@ fn aggregate_point<'a>( value.compact(); Ok(Some(value)) } - -fn external(error: datatypes::error::Error) -> DataFusionError { - DataFusionError::External(Box::new(error)) -} diff --git a/src/promql/src/histogram_eval/native.rs b/src/promql/src/histogram_eval/native.rs index 95b662b3071..5668eb5d606 100644 --- a/src/promql/src/histogram_eval/native.rs +++ b/src/promql/src/histogram_eval/native.rs @@ -19,6 +19,7 @@ use datafusion::error::{DataFusionError, Result}; use datatypes::series::histogram::{BucketSideView, HistogramView}; use super::bounds::EXPONENTIAL_BOUNDS; +pub(super) use crate::functions::kahan_inc; pub(super) const CUSTOM_BUCKETS_SCHEMA: i32 = -53; @@ -381,20 +382,6 @@ pub(super) fn variance(h: &HistogramView<'_>, buckets: &mut Vec) - Ok((variance + compensation) / h.count) } -/// Prometheus `kahansum.Inc`. -pub(super) fn kahan_inc(inc: f64, sum: f64, c: f64) -> (f64, f64) { - let t = sum + inc; - if t.is_infinite() { - return (t, 0.0); - } - let c = if sum.abs() >= inc.abs() { - c + ((sum - t) + inc) - } else { - c + ((inc - t) + sum) - }; - (t, c) -} - #[cfg(test)] mod tests { /// The last bucket of finite values ends at `f64::MAX`; only the next one, which counts diff --git a/src/promql/src/map_time.rs b/src/promql/src/map_time.rs index 15d959a3bd8..7781be82ca3 100644 --- a/src/promql/src/map_time.rs +++ b/src/promql/src/map_time.rs @@ -25,7 +25,7 @@ use datatypes::series::samples::{ }; use datatypes::series::time::{GridRange, TimeAxis}; -use crate::grid_series::GridSeriesBuilder; +use crate::grid_series::{GridSeriesBuilder, METADATA_LABELS, external}; #[cfg(test)] mod tests; @@ -145,9 +145,6 @@ fn go_int64(value: f64) -> i64 { } } -/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name. -const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"]; - const SECONDS_PER_DAY: u64 = 86_400; const MARCH_THRU_DECEMBER: u32 = 306; const ABSOLUTE_YEARS: u64 = 292_277_022_400; @@ -213,7 +210,3 @@ fn civil(days: u64) -> Civil { - 365 * jan_feb as i64, } } - -fn external(error: datatypes::error::Error) -> DataFusionError { - DataFusionError::External(Box::new(error)) -} diff --git a/src/promql/src/range_eval.rs b/src/promql/src/range_eval.rs index 8f3df3315b6..79fecb8cdce 100644 --- a/src/promql/src/range_eval.rs +++ b/src/promql/src/range_eval.rs @@ -24,7 +24,7 @@ use datatypes::series::raw::RawSeriesChunk; use datatypes::series::samples::{SampleRef, Series, SeriesFields, SeriesSchema}; use datatypes::series::time::{EvalGrid, GridSpec, TimeAxis}; -use crate::grid_series::GridSeriesBuilder; +use crate::grid_series::{GridSeriesBuilder, METADATA_LABELS}; mod function; #[cfg(test)] @@ -84,9 +84,6 @@ pub struct RangeEvaluator { window_error: Option, } -/// 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, diff --git a/src/promql/src/range_eval/function.rs b/src/promql/src/range_eval/function.rs index 90ca3d2b1e3..1038021e059 100644 --- a/src/promql/src/range_eval/function.rs +++ b/src/promql/src/range_eval/function.rs @@ -19,7 +19,9 @@ //! them. use super::{FloatWindow, WindowKernel}; -use crate::functions::{EdgeKind, add_counter_resets, count_edges, extremum, instant_delta}; +use crate::functions::{ + EdgeKind, add_counter_resets, count_edges, extremum, instant_delta, kahan_inc, +}; /// A PromQL range-vector function over float samples. #[derive(Clone, Copy, Debug, PartialEq)] @@ -141,20 +143,6 @@ impl RangeFunction { } } -/// Prometheus `kahansum.Inc`: Kahan–Neumaier summation that drops the compensation once the -/// sum is infinite. -fn kahan_inc(inc: f64, sum: f64, c: f64) -> (f64, f64) { - let t = sum + inc; - let c = if t.is_infinite() { - 0.0 - } else if sum.abs() >= inc.abs() { - c + ((sum - t) + inc) - } else { - c + ((inc - t) + sum) - }; - (t, c) -} - /// Prometheus `funcSumOverTime`. fn sum(values: &[f64]) -> f64 { let (sum, c) = values diff --git a/src/promql/src/select.rs b/src/promql/src/select.rs index 6516ed1e7fb..1254cddaa0c 100644 --- a/src/promql/src/select.rs +++ b/src/promql/src/select.rs @@ -29,7 +29,7 @@ use datatypes::series::time::TimeAxis; use datatypes::series::{Label, MetricName}; use crate::go_sort; -use crate::grid_series::GridSeriesBuilder; +use crate::grid_series::{GridSeriesBuilder, external}; #[cfg(test)] mod tests; @@ -451,7 +451,3 @@ fn value_at(series: &Series, point: usize) -> Result> { )), } } - -fn external(error: datatypes::error::Error) -> DataFusionError { - DataFusionError::External(Box::new(error)) -} diff --git a/src/promql/src/vector_match.rs b/src/promql/src/vector_match.rs index 2027122a802..7a4e5934c97 100644 --- a/src/promql/src/vector_match.rs +++ b/src/promql/src/vector_match.rs @@ -30,7 +30,7 @@ use datatypes::series::MetricName; use datatypes::series::samples::{DenseSamplesRef, SampleRef, Samples, Series, SeriesSchema}; use datatypes::series::time::{EvalGrid, TimeAxis}; -use crate::grid_series::GridSeriesBuilder; +use crate::grid_series::{GridSeriesBuilder, METADATA_LABELS}; #[cfg(test)] mod tests; @@ -529,9 +529,6 @@ fn overlap( Ok(None) } -/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name. -const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"]; - fn result_metric( many: &MetricName, one: &MetricName,