execution: backport histogram stdvar and stddev (#521)

Backport histogram_stdvar and histogram_stddev. Eventually we should
import those from prometheus like we do with histogram_quantile and
histogram_fraction.

Signed-off-by: Michael Hoffmann <[email protected]>
Co-authored-by: Michael Hoffmann <[email protected]>

Author: MichaHoffmann

engine/engine_test.go

@@ -92,9 +92,7 @@ func TestPromqlAcceptance(t *testing.T) {
 		skipTests: []string{
 			"testdata/name_label_dropping.test", // feature unsupported
 			"testdata/limit.test",               // limitk, limit_ratio
-			"testdata/native_histograms.test",   // histogram_stddev, histogram_stdvar
 			"testdata/functions.test",           // mad_over_time, predict_linear
-			"testdata/histograms.test",          // histogram_stddev, histogram_stdvar
 		}, // TODO(sungjin1212): change to test whole cases
 		TBRun: t,
 	}

execution/aggregate/accumulator.go

@@ -378,7 +378,7 @@ func (a *avgAcc) Add(ctx context.Context, v float64, h *histogram.FloatHistogram
 	a.hasValue = true
 
 	if !a.incremental {
-		newSum, newC := kahanSumInc(v, a.kahanSum, a.kahanC)
+		newSum, newC := KahanSumInc(v, a.kahanSum, a.kahanC)
 
 		if !math.IsInf(newSum, 0) {
 			// The sum doesn't overflow, so we propagate it to the
@@ -414,7 +414,7 @@ func (a *avgAcc) Add(ctx context.Context, v float64, h *histogram.FloatHistogram
 		}
 	}
 	currentMean := a.avg + a.kahanC
-	a.avg, a.kahanC = kahanSumInc(
+	a.avg, a.kahanC = KahanSumInc(
 		// Divide each side of the `-` by `group.groupCount` to avoid float64 overflows.
 		v/float64(a.count)-currentMean/float64(a.count),
 		a.avg,
@@ -693,7 +693,8 @@ func SumCompensated(s []float64) float64 {
 	return sum + c
 }
 
-func kahanSumInc(inc, sum, c float64) (newSum, newC float64) {
+// KahanSumInc implements kahan summation, see https://en.wikipedia.org/wiki/Kahan_summation_algorithm.
+func KahanSumInc(inc, sum, c float64) (newSum, newC float64) {
 	t := sum + inc
 	switch {
 	case math.IsInf(t, 0):

execution/function/functions.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/promql/parser"
 )
 
@@ -143,7 +144,19 @@ var instantVectorFuncs = map[string]functionCall{
 		if h == nil || len(vargs) != 2 {
 			return 0., false
 		}
-		return histogramFraction(vargs[0], vargs[1], h), true
+		return promql.HistogramFraction(vargs[0], vargs[1], h), true
+	},
+	"histogram_stddev": func(f float64, h *histogram.FloatHistogram, vargs ...float64) (float64, bool) {
+		if h == nil {
+			return 0., false
+		}
+		return histogramStdDev(h), true
+	},
+	"histogram_stdvar": func(f float64, h *histogram.FloatHistogram, vargs ...float64) (float64, bool) {
+		if h == nil {
+			return 0., false
+		}
+		return histogramStdVar(h), true
 	},
 	// variants of date time functions with an argument
 	"days_in_month": func(f float64, h *histogram.FloatHistogram, vargs ...float64) (float64, bool) {

execution/function/histogram.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/cespare/xxhash/v2"
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/promql/parser/posrange"
 	"github.com/prometheus/prometheus/util/annotations"
 
@@ -172,7 +173,7 @@ func (o *histogramOperator) processInputSeries(ctx context.Context, vectors []mo
 			// In that case, we reset the conventional buckets to avoid emitting a sample.
 			// TODO(fpetkovski): Prometheus is looking to solve these conflicts through warnings: https://github.com/prometheus/prometheus/issues/10839.
 			if len(o.seriesBuckets[outputSeriesID]) == 0 {
-				value := histogramQuantile(o.scalarPoints[stepIndex], vector.Histograms[i])
+				value := promql.HistogramQuantile(o.scalarPoints[stepIndex], vector.Histograms[i])
 				step.AppendSample(o.pool, uint64(outputSeriesID), value)
 			} else {
 				o.seriesBuckets[outputSeriesID] = o.seriesBuckets[outputSeriesID][:0]