refactor(promql): share metadata labels, error mapping and Kahan increments across Narthex operators

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

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: <https://github.com/prometheus/prometheus/blob/90b2f7a540b8a70d8d81372e6692dcbb67ccbaaa/promql/functions.go#L793-L837>

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<str>, Arc<str>)>;
 
 /// The label set Prometheus compares, which has no empty values.
@@ -260,6 +263,6 @@ fn read_row(series: &Series, row: usize, values: &mut [Option<f64>]) -> Result<b
     Ok(present)
 }
 
-fn external(error: datatypes::error::Error) -> DataFusionError {
+pub(crate) fn external(error: datatypes::error::Error) -> DataFusionError {
     DataFusionError::External(Box::new(error))
 }

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)]

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))
-}

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<NativeBucket>) -
     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

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))
-}

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<String>,
 }
 
-/// 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>,

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

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<Option<f64>> {
         )),
     }
 }
-
-fn external(error: datatypes::error::Error) -> DataFusionError {
-    DataFusionError::External(Box::new(error))
-}

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,