// Copyright The Prometheus Authors

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

package promql

import (

"context"

"errors"

"fmt"

"math"

"slices"

"sort"

"strings"

"time"

"github.com/facette/natsort"

"github.com/grafana/regexp"

"github.com/prometheus/common/model"

"github.com/prometheus/prometheus/model/histogram"

"github.com/prometheus/prometheus/model/labels"

"github.com/prometheus/prometheus/promql/parser"

"github.com/prometheus/prometheus/promql/parser/posrange"

"github.com/prometheus/prometheus/schema"

"github.com/prometheus/prometheus/util/annotations"

"github.com/prometheus/prometheus/util/kahansum"

)

// FunctionCall is the type of a PromQL function implementation

//

// vals is a list of the evaluated arguments for the function call.

//

// For range vectors it will be a Matrix with one series, instant vectors a

// Vector, scalars a Vector with one series whose value is the scalar

// value,and nil for strings.

//

// args are the original arguments to the function, where you can access

// matrixSelectors, vectorSelectors, and StringLiterals.

//

// enh.Out is a pre-allocated empty vector that you may use to accumulate

// output before returning it. The vectors in vals should not be returned.a

//

// Range vector functions need only return a vector with the right value,

// the metric and timestamp are not needed.

//

// Instant vector functions need only return a vector with the right values and

// metrics, the timestamp are not needed.

//

// Scalar results should be returned as the value of a sample in a Vector.

type FunctionCall func(vectorVals []Vector, matrixVals Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations)

// funcQueryContext is a placeholder for start(), end(), range(), and step() functions.

// These are folded into NumberLiteral nodes by foldQueryContextFunctions during query

// preprocessing and must never reach the evaluator.

func funcQueryContext(_ []Vector, _ Matrix, _ parser.Expressions, _ *EvalNodeHelper) (Vector, annotations.Annotations) {

panic("query context functions must be folded during preprocessing and must never be evaluated")

}

// === time() float64 ===

func funcTime(_ []Vector, _ Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

return Vector{Sample{

F: float64(enh.Ts) / 1000,

}}, nil

}

// pickOrInterpolateLeft returns the value at the left boundary of the range.

// If interpolation is needed (when smoothed is true and the first sample is before the range start),

// it returns the interpolated value at the left boundary; otherwise, it returns the first sample's value.

func pickOrInterpolateLeft(floats []FPoint, first int, rangeStart int64, smoothed, isCounter bool) float64 {

if smoothed && floats[first].T < rangeStart {

return interpolate(floats[first], floats[first+1], rangeStart, isCounter)

}

return floats[first].F

}

// pickOrInterpolateRight returns the value at the right boundary of the range.

// If interpolation is needed (when smoothed is true and the last sample is after the range end),

// it returns the interpolated value at the right boundary; otherwise, it returns the last sample's value.

func pickOrInterpolateRight(floats []FPoint, last int, rangeEnd int64, smoothed, isCounter bool) float64 {

if smoothed && last > 0 && floats[last].T > rangeEnd {

return interpolate(floats[last-1], floats[last], rangeEnd, isCounter)

}

return floats[last].F

}

// interpolate performs linear interpolation between two points.

// If isCounter is true and there is a counter reset, it models the counter

// as starting from 0 (post-reset) by setting y1 to 0.

// It then calculates the interpolated value at the given timestamp.

func interpolate(p1, p2 FPoint, t int64, isCounter bool) float64 {

y1 := p1.F

y2 := p2.F

if isCounter && y2 < y1 {

y1 = 0

}

return y1 + (y2-y1)*float64(t-p1.T)/float64(p2.T-p1.T)

}

// correctForCounterResets calculates the correction for counter resets.

// This function is only used for extendedRate functions with smoothed or anchored rates.

func correctForCounterResets(left, right float64, points []FPoint) float64 {

var correction float64

prev := left

for _, p := range points {

if p.F < prev {

correction += prev

}

prev = p.F

}

if right < prev {

correction += prev

}

return correction

}

// extendedRate is a utility function for anchored/smoothed rate/increase/delta.

// It calculates the rate (allowing for counter resets if isCounter is true),

// extrapolates if the first/last sample if needed, and returns

// the result as either per-second (if isRate is true) or overall.

func extendedRate(vals Matrix, args parser.Expressions, enh *EvalNodeHelper, isCounter, isRate bool) (Vector, annotations.Annotations) {

var (

ms = args[0].(*parser.MatrixSelector)

vs = ms.VectorSelector.(*parser.VectorSelector)

samples = vals[0]

f = samples.Floats

lastSampleIndex = len(f) - 1

rangeStart = enh.Ts - durationMilliseconds(ms.Range+vs.Offset)

rangeEnd = enh.Ts - durationMilliseconds(vs.Offset)

annos annotations.Annotations

smoothed = vs.Smoothed

)

firstSampleIndex := max(0, sort.Search(lastSampleIndex, func(i int) bool { return f[i].T > rangeStart })-1)

if smoothed {

lastSampleIndex = sort.Search(lastSampleIndex, func(i int) bool { return f[i].T >= rangeEnd })

}

if f[lastSampleIndex].T <= rangeStart {

return enh.Out, annos

}

if smoothed && f[firstSampleIndex].T > rangeEnd {

return enh.Out, annos

}

left := pickOrInterpolateLeft(f, firstSampleIndex, rangeStart, smoothed, isCounter)

right := pickOrInterpolateRight(f, lastSampleIndex, rangeEnd, smoothed, isCounter)

resultFloat := right - left

if isCounter {

// We only need to consider samples exactly within the range

// for counter resets correction, as pickOrInterpolateLeft and

// pickOrInterpolateRight already handle the resets at boundaries.

if f[firstSampleIndex].T <= rangeStart {

firstSampleIndex++

}

if f[lastSampleIndex].T >= rangeEnd {

lastSampleIndex--

}

resultFloat += correctForCounterResets(left, right, f[firstSampleIndex:lastSampleIndex+1])

}

if isRate {

resultFloat /= ms.Range.Seconds()

}

return append(enh.Out, Sample{F: resultFloat}), annos

}

// extrapolatedRate is a utility function for rate/increase/delta.

// It calculates the rate (allowing for counter resets if isCounter is true),

// extrapolates if the first/last sample is close to the boundary, and returns

// the result as either per-second (if isRate is true) or overall.

//

// Note: If the vector selector is smoothed or anchored, it will use the

// extendedRate function instead.

func extrapolatedRate(vals Matrix, args parser.Expressions, enh *EvalNodeHelper, isCounter, isRate bool) (Vector, annotations.Annotations) {

ms := args[0].(*parser.MatrixSelector)

vs := ms.VectorSelector.(*parser.VectorSelector)

if vs.Anchored || vs.Smoothed {

return extendedRate(vals, args, enh, isCounter, isRate)

}

var (

samples = vals[0]

rangeStart = enh.Ts - durationMilliseconds(ms.Range+vs.Offset)

rangeEnd = enh.Ts - durationMilliseconds(vs.Offset)

resultFloat float64

resultHistogram *histogram.FloatHistogram

firstT, lastT int64

numSamplesMinusOne int

annos annotations.Annotations

)

// We need either at least two Histograms and no Floats, or at least two

// Floats and no Histograms to calculate a rate. Otherwise, drop this

// Vector element.

if len(samples.Histograms) > 0 && len(samples.Floats) > 0 {

return enh.Out, annos.Add(annotations.NewMixedFloatsHistogramsWarning(getMetricName(samples.Metric), args[0].PositionRange()))

}

switch {

case len(samples.Histograms) > 1:

numSamplesMinusOne = len(samples.Histograms) - 1

firstT = samples.Histograms[0].T

lastT = samples.Histograms[numSamplesMinusOne].T

var newAnnos annotations.Annotations

resultHistogram, newAnnos = histogramRate(samples.Histograms, isCounter, samples.Metric, args[0].PositionRange())

annos.Merge(newAnnos)

if resultHistogram == nil {

// The histograms are not compatible with each other.

return enh.Out, annos

}

case len(samples.Floats) > 1:

numSamplesMinusOne = len(samples.Floats) - 1

firstT = samples.Floats[0].T

lastT = samples.Floats[numSamplesMinusOne].T

resultFloat = samples.Floats[numSamplesMinusOne].F - samples.Floats[0].F

if !isCounter {

break

}

// Handle counter resets:

prevValue := samples.Floats[0].F

for _, currPoint := range samples.Floats[1:] {

if currPoint.F < prevValue {

resultFloat += prevValue

}

prevValue = currPoint.F

}

default:

// TODO: add RangeTooShortWarning

return enh.Out, annos

}

// Duration between first/last samples and boundary of range.

durationToStart := float64(firstT-rangeStart) / 1000

durationToEnd := float64(rangeEnd-lastT) / 1000

sampledInterval := float64(lastT-firstT) / 1000

averageDurationBetweenSamples := sampledInterval / float64(numSamplesMinusOne)

// If samples are close enough to the (lower or upper) boundary of the

// range, we extrapolate the rate all the way to the boundary in

// question. "Close enough" is defined as "up to 10% more than the

// average duration between samples within the range", see

// extrapolationThreshold below. Essentially, we are assuming a more or

// less regular spacing between samples, and if we don't see a sample

// where we would expect one, we assume the series does not cover the

// whole range, but starts and/or ends within the range. We still

// extrapolate the rate in this case, but not all the way to the

// boundary, but only by half of the average duration between samples

// (which is our guess for where the series actually starts or ends).

extrapolationThreshold := averageDurationBetweenSamples * 1.1

if durationToStart >= extrapolationThreshold {

durationToStart = averageDurationBetweenSamples / 2

}

if isCounter {

// Counters cannot be negative. If we have any slope at all

// (i.e. resultFloat went up), we can extrapolate the zero point

// of the counter. If the duration to the zero point is shorter

// than the durationToStart, we take the zero point as the start

// of the series, thereby avoiding extrapolation to negative

// counter values.

durationToZero := durationToStart

if resultFloat > 0 &&

len(samples.Floats) > 0 &&

samples.Floats[0].F >= 0 {

durationToZero = sampledInterval * (samples.Floats[0].F / resultFloat)

} else if resultHistogram != nil &&

resultHistogram.Count > 0 &&

len(samples.Histograms) > 0 &&

samples.Histograms[0].H.Count >= 0 {

durationToZero = sampledInterval * (samples.Histograms[0].H.Count / resultHistogram.Count)

}

if durationToZero < durationToStart {

durationToStart = durationToZero

}

}

if durationToEnd >= extrapolationThreshold {

durationToEnd = averageDurationBetweenSamples / 2

}

factor := (sampledInterval + durationToStart + durationToEnd) / sampledInterval

if isRate {

factor /= ms.Range.Seconds()

}

if resultHistogram == nil {

resultFloat *= factor

} else {

resultHistogram.Mul(factor)

}

return append(enh.Out, Sample{F: resultFloat, H: resultHistogram}), annos

}

// histogramRate is a helper function for extrapolatedRate. It requires

// points[0] to be a histogram. It returns nil if any other Point in points is

// not a histogram, and a warning wrapped in an annotation in that case.

// Otherwise, it returns the calculated histogram and an empty annotation.

func histogramRate(points []HPoint, isCounter bool, labels labels.Labels, pos posrange.PositionRange) (*histogram.FloatHistogram, annotations.Annotations) {

var (

prev = points[0].H

usingCustomBuckets = prev.UsesCustomBuckets()

last = points[len(points)-1].H

annos annotations.Annotations

)

if last == nil {

return nil, annos.Add(annotations.NewMixedFloatsHistogramsWarning(getMetricName(labels), pos))

}

// We check for gauge type histograms in the loop below, but the loop

// below does not run on the first and last point, so check the first

// and last point now.

if isCounter && (prev.CounterResetHint == histogram.GaugeType || last.CounterResetHint == histogram.GaugeType) {

annos.Add(annotations.NewNativeHistogramNotCounterWarning(getMetricName(labels), pos))

}

// Null out the 1st sample if there is a counter reset between the 1st

// and 2nd. In this case, we want to ignore any incompatibility in the

// bucket layout of the 1st sample because we do not need to look at it.

if isCounter && len(points) > 1 {

second := points[1].H

if second != nil && second.DetectReset(prev) {

prev = &histogram.FloatHistogram{}

prev.Schema = second.Schema

prev.CustomValues = second.CustomValues

usingCustomBuckets = second.UsesCustomBuckets()

}

}

if last.UsesCustomBuckets() != usingCustomBuckets {

return nil, annos.Add(annotations.NewMixedExponentialCustomHistogramsWarning(getMetricName(labels), pos))

}

// First iteration to find out two things:

// - What's the smallest relevant schema?

// - Are all data points histograms?

minSchema := min(last.Schema, prev.Schema)

for _, currPoint := range points[1 : len(points)-1] {

curr := currPoint.H

if curr == nil {

return nil, annotations.New().Add(annotations.NewMixedFloatsHistogramsWarning(getMetricName(labels), pos))

}

if !isCounter {

continue

}

if curr.CounterResetHint == histogram.GaugeType {

annos.Add(annotations.NewNativeHistogramNotCounterWarning(getMetricName(labels), pos))

}

if curr.Schema < minSchema {

minSchema = curr.Schema

}

if curr.UsesCustomBuckets() != usingCustomBuckets {

return nil, annotations.New().Add(annotations.NewMixedExponentialCustomHistogramsWarning(getMetricName(labels), pos))

}

}

h := last.CopyToSchema(minSchema)

// This subtraction may deliberately include conflicting counter resets.

// Counter resets are treated explicitly in this function, so the

// information about conflicting counter resets is ignored here.

_, _, nhcbBoundsReconciled, err := h.Sub(prev)

if err != nil {

if errors.Is(err, histogram.ErrHistogramsIncompatibleSchema) {

return nil, annotations.New().Add(annotations.NewMixedExponentialCustomHistogramsWarning(getMetricName(labels), pos))

}

}

if nhcbBoundsReconciled {

annos.Add(annotations.NewMismatchedCustomBucketsHistogramsInfo(pos, annotations.HistogramSub))

}

if isCounter {

// Second iteration to deal with counter resets.

for _, currPoint := range points[1:] {

curr := currPoint.H

if curr.DetectReset(prev) {

// Counter reset conflict ignored here for the same reason as above.

_, _, nhcbBoundsReconciled, err := h.Add(prev)

if err != nil {

if errors.Is(err, histogram.ErrHistogramsIncompatibleSchema) {

return nil, annotations.New().Add(annotations.NewMixedExponentialCustomHistogramsWarning(getMetricName(labels), pos))

}

}

if nhcbBoundsReconciled {

annos.Add(annotations.NewMismatchedCustomBucketsHistogramsInfo(pos, annotations.HistogramAdd))

}

}

prev = curr

}

} else if points[0].H.CounterResetHint != histogram.GaugeType || points[len(points)-1].H.CounterResetHint != histogram.GaugeType {

annos.Add(annotations.NewNativeHistogramNotGaugeWarning(getMetricName(labels), pos))

}

h.CounterResetHint = histogram.GaugeType

return h.Compact(0), annos

}

// === delta(Matrix parser.ValueTypeMatrix) (Vector, Annotations) ===

func funcDelta(_ []Vector, matrixVals Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

return extrapolatedRate(matrixVals, args, enh, false, false)

}

// === rate(node parser.ValueTypeMatrix) (Vector, Annotations) ===

func funcRate(_ []Vector, matrixVals Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

return extrapolatedRate(matrixVals, args, enh, true, true)

}

// === increase(node parser.ValueTypeMatrix) (Vector, Annotations) ===

func funcIncrease(_ []Vector, matrixVals Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

return extrapolatedRate(matrixVals, args, enh, true, false)

}

// === irate(node parser.ValueTypeMatrix) (Vector, Annotations) ===

func funcIrate(_ []Vector, matrixVals Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

return instantValue(matrixVals, args, enh.Out, true)

}

// === idelta(node model.ValMatrix) (Vector, Annotations) ===

func funcIdelta(_ []Vector, matrixVals Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

return instantValue(matrixVals, args, enh.Out, false)

}

func instantValue(vals Matrix, args parser.Expressions, out Vector, isRate bool) (Vector, annotations.Annotations) {

var (

samples = vals[0]

ss = make([]Sample, 0, 2)

annos annotations.Annotations

)

// No sense in trying to compute a rate without at least two points. Drop

// this Vector element.

// TODO: add RangeTooShortWarning

if len(samples.Floats)+len(samples.Histograms) < 2 {

return out, nil

}

// Add the last 2 float samples if they exist.

for i := max(0, len(samples.Floats)-2); i < len(samples.Floats); i++ {

ss = append(ss, Sample{

F: samples.Floats[i].F,

T: samples.Floats[i].T,

})

}

// Add the last 2 histogram samples into their correct position if they exist.

for i := max(0, len(samples.Histograms)-2); i < len(samples.Histograms); i++ {

s := Sample{

H: samples.Histograms[i].H,

T: samples.Histograms[i].T,

}

switch {

case len(ss) == 0:

ss = append(ss, s)

case len(ss) == 1:

if s.T < ss[0].T {

ss = append([]Sample{s}, ss...)

} else {

ss = append(ss, s)

}

case s.T < ss[0].T:

// s is older than 1st, so discard it.

case s.T > ss[1].T:

// s is newest, so add it as 2nd and make the old 2nd the new 1st.

ss[0] = ss[1]

ss[1] = s

default:

// In all other cases, we just make s the new 1st.

// This establishes a correct order, even in the (irregular)

// case of equal timestamps.

ss[0] = s

}

}

resultSample := ss[1]

sampledInterval := ss[1].T - ss[0].T

if sampledInterval == 0 {

// Avoid dividing by 0.

return out, nil

}

switch {

case ss[1].H == nil && ss[0].H == nil:

if !isRate || !(ss[1].F < ss[0].F) {

// Gauge, or counter without reset, or counter with NaN value.

resultSample.F = ss[1].F - ss[0].F

}

// In case of a counter reset, we leave resultSample at

// its current value, which is already ss[1].

case ss[1].H != nil && ss[0].H != nil:

resultSample.H = ss[1].H.Copy()

// irate should only be applied to counters.

if isRate && (ss[1].H.CounterResetHint == histogram.GaugeType || ss[0].H.CounterResetHint == histogram.GaugeType) {

annos.Add(annotations.NewNativeHistogramNotCounterWarning(getMetricName(samples.Metric), args.PositionRange()))

}

// idelta should only be applied to gauges.

if !isRate && (ss[1].H.CounterResetHint != histogram.GaugeType || ss[0].H.CounterResetHint != histogram.GaugeType) {

annos.Add(annotations.NewNativeHistogramNotGaugeWarning(getMetricName(samples.Metric), args.PositionRange()))

}

if !isRate || !ss[1].H.DetectReset(ss[0].H) {

// This subtraction may deliberately include conflicting

// counter resets. Counter resets are treated explicitly

// in this function, so the information about

// conflicting counter resets is ignored here.

_, _, nhcbBoundsReconciled, err := resultSample.H.Sub(ss[0].H)

if errors.Is(err, histogram.ErrHistogramsIncompatibleSchema) {

return out, annos.Add(annotations.NewMixedExponentialCustomHistogramsWarning(getMetricName(samples.Metric), args.PositionRange()))

}

if nhcbBoundsReconciled {

annos.Add(annotations.NewMismatchedCustomBucketsHistogramsInfo(args.PositionRange(), annotations.HistogramSub))

}

}

resultSample.H.CounterResetHint = histogram.GaugeType

resultSample.H.Compact(0)

default:

// Mix of a float and a histogram.

return out, annos.Add(annotations.NewMixedFloatsHistogramsWarning(getMetricName(samples.Metric), args.PositionRange()))

}

if isRate {

// Convert to per-second.

if resultSample.H == nil {

resultSample.F /= float64(sampledInterval) / 1000

} else {

resultSample.H.Div(float64(sampledInterval) / 1000)

}

}

return append(out, resultSample), annos

}

// Calculate the trend value at the given index i in raw data d.

// This is somewhat analogous to the slope of the trend at the given index.

// The argument "tf" is the trend factor.

// The argument "s0" is the computed smoothed value.

// The argument "s1" is the computed trend factor.

// The argument "b" is the raw input value.

func calcTrendValue(i int, tf, s0, s1, b float64) float64 {

if i == 0 {

return b

}

x := tf * (s1 - s0)

y := (1 - tf) * b

return x + y

}

// Double exponential smoothing is similar to a weighted moving average, where

// historical data has exponentially less influence on the current data. It also

// accounts for trends in data. The smoothing factor (0 < sf < 1) affects how

// historical data will affect the current data. A lower smoothing factor

// increases the influence of historical data. The trend factor (0 < tf < 1)

// affects how trends in historical data will affect the current data. A higher

// trend factor increases the influence. of trends. Algorithm taken from

// https://en.wikipedia.org/wiki/Exponential_smoothing .

func funcDoubleExponentialSmoothing(vectorVals []Vector, matrixVal Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

if len(vectorVals) < 2 || len(vectorVals[0]) == 0 || len(vectorVals[1]) == 0 || len(matrixVal) == 0 {

return enh.Out, nil

}

samples := matrixVal[0]

// The smoothing factor argument.

sf := vectorVals[0][0].F

// The trend factor argument.

tf := vectorVals[1][0].F

// Check that the input parameters are valid.

if sf <= 0 || sf >= 1 {

panic(fmt.Errorf("invalid smoothing factor. Expected: 0 < sf < 1, got: %f", sf))

}

if tf <= 0 || tf >= 1 {

panic(fmt.Errorf("invalid trend factor. Expected: 0 < tf < 1, got: %f", tf))

}

l := len(samples.Floats)

// Can't do the smoothing operation with less than two points.

if l < 2 {

// Annotate mix of float and histogram.

if l == 1 && len(samples.Histograms) > 0 {

return enh.Out, annotations.New().Add(annotations.NewHistogramIgnoredInMixedRangeInfo(getMetricName(samples.Metric), args[0].PositionRange()))

}

return enh.Out, nil

}

var s0, s1, b float64

// Set initial values.

s1 = samples.Floats[0].F

b = samples.Floats[1].F - samples.Floats[0].F

// Run the smoothing operation.

var x, y float64

for i := 1; i < l; i++ {

// Scale the raw value against the smoothing factor.

x = sf * samples.Floats[i].F

// Scale the last smoothed value with the trend at this point.

b = calcTrendValue(i-1, tf, s0, s1, b)

y = (1 - sf) * (s1 + b)

s0, s1 = s1, x+y

}

if len(samples.Histograms) > 0 {

return append(enh.Out, Sample{F: s1}), annotations.New().Add(annotations.NewHistogramIgnoredInMixedRangeInfo(getMetricName(samples.Metric), args[0].PositionRange()))

}

return append(enh.Out, Sample{F: s1}), nil

}

// filterFloats filters out histogram samples from the vector in-place.

func filterFloats(v Vector) Vector {

floats := v[:0]

for _, s := range v {

if s.H == nil {

floats = append(floats, s)

}

}

return floats

}

// === sort(node parser.ValueTypeVector) (Vector, Annotations) ===

func funcSort(vectorVals []Vector, _ Matrix, _ parser.Expressions, _ *EvalNodeHelper) (Vector, annotations.Annotations) {

// NaN should sort to the bottom, so take descending sort with NaN first and

// reverse it.

byValueSorter := vectorByReverseValueHeap(filterFloats(vectorVals[0]))

sort.Sort(sort.Reverse(byValueSorter))

return Vector(byValueSorter), nil

}

// === sortDesc(node parser.ValueTypeVector) (Vector, Annotations) ===

func funcSortDesc(vectorVals []Vector, _ Matrix, _ parser.Expressions, _ *EvalNodeHelper) (Vector, annotations.Annotations) {

// NaN should sort to the bottom, so take ascending sort with NaN first and

// reverse it.

byValueSorter := vectorByValueHeap(filterFloats(vectorVals[0]))

sort.Sort(sort.Reverse(byValueSorter))

return Vector(byValueSorter), nil

}

// === sort_by_label(vector parser.ValueTypeVector, label parser.ValueTypeString...) (Vector, Annotations) ===

func funcSortByLabel(vectorVals []Vector, _ Matrix, args parser.Expressions, _ *EvalNodeHelper) (Vector, annotations.Annotations) {

lbls := stringSliceFromArgs(args[1:])

slices.SortFunc(vectorVals[0], func(a, b Sample) int {

for _, label := range lbls {

lv1 := a.Metric.Get(label)

lv2 := b.Metric.Get(label)

if lv1 == lv2 {

continue

}

if natsort.Compare(lv1, lv2) {

return -1

}

return +1

}

// If all labels provided as arguments were equal, sort by the full label set. This ensures a consistent ordering.

return labels.Compare(a.Metric, b.Metric)

})

return vectorVals[0], nil

}

// === sort_by_label_desc(vector parser.ValueTypeVector, label parser.ValueTypeString...) (Vector, Annotations) ===

func funcSortByLabelDesc(vectorVals []Vector, _ Matrix, args parser.Expressions, _ *EvalNodeHelper) (Vector, annotations.Annotations) {

lbls := stringSliceFromArgs(args[1:])

slices.SortFunc(vectorVals[0], func(a, b Sample) int {

for _, label := range lbls {

lv1 := a.Metric.Get(label)

lv2 := b.Metric.Get(label)

if lv1 == lv2 {

continue

}

if natsort.Compare(lv1, lv2) {

return +1

}

return -1

}

// If all labels provided as arguments were equal, sort by the full label set. This ensures a consistent ordering.

return -labels.Compare(a.Metric, b.Metric)

})

return vectorVals[0], nil

}

func clamp(vec Vector, minVal, maxVal float64, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

if maxVal < minVal {

return enh.Out, nil

}

for _, el := range vec {

if el.H != nil {

// Process only float samples.

continue

}

if !enh.enableDelayedNameRemoval {

el.Metric = el.Metric.DropReserved(schema.IsMetadataLabel)

}

enh.Out = append(enh.Out, Sample{

Metric: el.Metric,

F: math.Max(minVal, math.Min(maxVal, el.F)),

DropName: true,

})

}

return enh.Out, nil

}

// === clamp(Vector parser.ValueTypeVector, min, max Scalar) (Vector, Annotations) ===

func funcClamp(vectorVals []Vector, _ Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

vec := vectorVals[0]

minVal := vectorVals[1][0].F

maxVal := vectorVals[2][0].F

return clamp(vec, minVal, maxVal, enh)

}

// === clamp_max(Vector parser.ValueTypeVector, max Scalar) (Vector, Annotations) ===

func funcClampMax(vectorVals []Vector, _ Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

vec := vectorVals[0]

maxVal := vectorVals[1][0].F

return clamp(vec, math.Inf(-1), maxVal, enh)

}

// === clamp_min(Vector parser.ValueTypeVector, min Scalar) (Vector, Annotations) ===

func funcClampMin(vectorVals []Vector, _ Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

vec := vectorVals[0]

minVal := vectorVals[1][0].F

return clamp(vec, minVal, math.Inf(+1), enh)

}

// === round(Vector parser.ValueTypeVector, toNearest=1 Scalar) (Vector, Annotations) ===

func funcRound(vectorVals []Vector, _ Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

// round returns a number rounded to toNearest.

// Ties are solved by rounding up.

toNearest := float64(1)

if len(args) >= 2 {

toNearest = vectorVals[1][0].F

}

// Invert as it seems to cause fewer floating point accuracy issues.

toNearestInverse := 1.0 / toNearest

return simpleFloatFunc(vectorVals, enh, func(f float64) float64 {

return math.Floor(f*toNearestInverse+0.5) / toNearestInverse

}), nil

}

// === Scalar(node parser.ValueTypeVector) Scalar ===

func funcScalar(vectorVals []Vector, _ Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

var (

v = vectorVals[0]

value float64

found bool

)

for _, s := range v {

if s.H == nil {

if found {

// More than one float found, return NaN.

return append(enh.Out, Sample{F: math.NaN()}), nil

}

found = true

value = s.F

}

}

// Return the single float if found, otherwise return NaN.

if !found {

return append(enh.Out, Sample{F: math.NaN()}), nil

}

return append(enh.Out, Sample{F: value}), nil

}

func aggrOverTime(matrixVal Matrix, enh *EvalNodeHelper, aggrFn func(Series) float64) Vector {

if len(matrixVal) == 0 {

return enh.Out

}

el := matrixVal[0]

return append(enh.Out, Sample{F: aggrFn(el)})

}

func aggrHistOverTime(matrixVal Matrix, enh *EvalNodeHelper, aggrFn func(Series) (*histogram.FloatHistogram, error)) (Vector, error) {

if len(matrixVal) == 0 {

return enh.Out, nil

}

el := matrixVal[0]

res, err := aggrFn(el)

return append(enh.Out, Sample{H: res}), err

}

// === avg_over_time(Matrix parser.ValueTypeMatrix) (Vector, Annotations) ===

func funcAvgOverTime(_ []Vector, matrixVal Matrix, args parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

if len(matrixVal) == 0 {

return enh.Out, nil

}

firstSeries := matrixVal[0]

if len(firstSeries.Floats) > 0 && len(firstSeries.Histograms) > 0 {

return enh.Out, annotations.New().Add(annotations.NewMixedFloatsHistogramsWarning(getMetricName(firstSeries.Metric), args[0].PositionRange()))

}

// We improve the accuracy with the help of Kahan summation.

// For a while, we assumed that incremental mean calculation combined

// with Kahan summation (see

// https://stackoverflow.com/questions/61665473/is-it-beneficial-for-precision-to-calculate-the-incremental-mean-average

// for inspiration) is generally the preferred solution. However, it

// then turned out that direct mean calculation (still in combination

// with Kahan summation) is often more accurate. See discussion in

// https://github.com/prometheus/prometheus/issues/16714 . The problem

// with the direct mean calculation is that it can overflow float64 for

// inputs on which the incremental mean calculation works just fine. Our

// current approach is therefore to use direct mean calculation as long

// as we do not overflow (or underflow) the running sum. Once the latter

// would happen, we switch to incremental mean calculation. This seems

// to work reasonably well, but note that a deeper understanding would

// be needed to find out if maybe an earlier switch to incremental mean

// calculation would be better in terms of accuracy. Also, we could

// apply a number of additional means to improve the accuracy, like

// processing the values in a particular order. For now, we decided that

// the current implementation is accurate enough for practical purposes.

if len(firstSeries.Floats) == 0 {

// The passed values only contain histograms.

var annos annotations.Annotations

vec, err := aggrHistOverTime(matrixVal, enh, func(s Series) (*histogram.FloatHistogram, error) {

var counterResetSeen, notCounterResetSeen, nhcbBoundsReconciledSeen bool

trackCounterReset := func(h *histogram.FloatHistogram) {

switch h.CounterResetHint {

case histogram.CounterReset:

counterResetSeen = true

case histogram.NotCounterReset:

notCounterResetSeen = true

}

}

defer func() {

if counterResetSeen && notCounterResetSeen {

annos.Add(annotations.NewHistogramCounterResetCollisionWarning(args[0].PositionRange(), annotations.HistogramAgg))

}

if nhcbBoundsReconciledSeen {

annos.Add(annotations.NewMismatchedCustomBucketsHistogramsInfo(args[0].PositionRange(), annotations.HistogramAgg))

}

}()

var (

sum = s.Histograms[0].H.Copy()

mean, kahanC *histogram.FloatHistogram

count = 1.

incrementalMean bool

nhcbBoundsReconciled bool

err error

)

trackCounterReset(sum)

for i, h := range s.Histograms[1:] {

trackCounterReset(h.H)

count = float64(i + 2)

if !incrementalMean {

sumCopy := sum.Copy()

var cCopy *histogram.FloatHistogram

if kahanC != nil {

cCopy = kahanC.Copy()

}

cCopy, _, nhcbBoundsReconciled, err = sumCopy.KahanAdd(h.H, cCopy)

if err != nil {

return sumCopy.Div(count), err

}

if nhcbBoundsReconciled {

nhcbBoundsReconciledSeen = true

}

if !sumCopy.HasOverflow() {

sum, kahanC = sumCopy, cCopy

continue

}

incrementalMean = true

mean = sum.Copy().Div(count - 1)

if kahanC != nil {

kahanC.Div(count - 1)

}

}

q := (count - 1) / count

if kahanC != nil {

kahanC.Mul(q)

}

toAdd := h.H.Copy().Div(count)

kahanC, _, nhcbBoundsReconciled, err = mean.Mul(q).KahanAdd(toAdd, kahanC)

if err != nil {

return mean, err

}

if nhcbBoundsReconciled {

nhcbBoundsReconciledSeen = true

}

}

if incrementalMean {

if kahanC != nil {

_, _, _, err := mean.Add(kahanC)

return mean, err

}

return mean, nil

}

if kahanC != nil {

_, _, _, err := sum.Div(count).Add(kahanC.Div(count))

return sum, err

}

return sum.Div(count), nil

})

if err != nil {

if errors.Is(err, histogram.ErrHistogramsIncompatibleSchema) {

return enh.Out, annotations.New().Add(annotations.NewMixedExponentialCustomHistogramsWarning(getMetricName(firstSeries.Metric), args[0].PositionRange()))

}

}

return vec, annos

}

return aggrOverTime(matrixVal, enh, func(s Series) float64 {

var (

// Pre-set the 1st sample to start the loop with the 2nd.

sum, count = s.Floats[0].F, 1.

mean, kahanC float64

incrementalMean bool

)

for i, f := range s.Floats[1:] {

count = float64(i + 2)

if !incrementalMean {

newSum, newC := kahansum.Inc(f.F, sum, kahanC)

// Perform regular mean calculation as long as

// the sum doesn't overflow.

if !math.IsInf(newSum, 0) {

sum, kahanC = newSum, newC

continue

}

// Handle overflow by reverting to incremental

// calculation of the mean value.

incrementalMean = true

mean = sum / (count - 1)

kahanC /= (count - 1)

}

q := (count - 1) / count

mean, kahanC = kahansum.Inc(f.F/count, q*mean, q*kahanC)

}

if incrementalMean {

return mean + kahanC

}

return sum/count + kahanC/count

}), nil

}

// === count_over_time(Matrix parser.ValueTypeMatrix) (Vector, Notes) ===

func funcCountOverTime(_ []Vector, matrixVals Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

return aggrOverTime(matrixVals, enh, func(s Series) float64 {

return float64(len(s.Floats) + len(s.Histograms))

}), nil

}

// === first_over_time(Matrix parser.ValueTypeMatrix) (Vector, Notes) ===

func funcFirstOverTime(_ []Vector, matrixVal Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

if len(matrixVal) == 0 {

return enh.Out, nil

}

el := matrixVal[0]

var f FPoint

if len(el.Floats) > 0 {

f = el.Floats[0]

}

var h HPoint

if len(el.Histograms) > 0 {

h = el.Histograms[0]

}

// If a float data point exists and is older than any histogram data

// points, return it.

if h.H == nil || (len(el.Floats) > 0 && f.T < h.T) {

return append(enh.Out, Sample{

Metric: el.Metric,

F: f.F,

}), nil

}

return append(enh.Out, Sample{

Metric: el.Metric,

H: h.H.Copy(),

}), nil

}

// === last_over_time(Matrix parser.ValueTypeMatrix) (Vector, Notes) ===

func funcLastOverTime(_ []Vector, matrixVal Matrix, _ parser.Expressions, enh *EvalNodeHelper) (Vector, annotations.Annotations) {

if len(matrixVal) == 0 {

return enh.Out, nil