Repository: prometheus/prometheus
File: promql/engine.go (Go)

Code before:
	floatKahanC    float64 // "Compensating value" for Kahan summation.
	groupCount     float64
	heap           vectorByValueHeap

	// All bools together for better packing within the struct.
	seen                   bool // Was this output groups seen in the input at this timestamp.
	hasFloat               bool // Has at least 1 float64 sample aggregated.
	hasHistogram           bool // Has at least 1 histogram sample aggregated.
	incompatibleHistograms bool // If true, group has seen mixed exponential and custom buckets, or incompatible custom buckets.
	groupAggrComplete      bool // Used by LIMITK to short-cut series loop when we've reached K elem on every group.
	incrementalMean        bool // True after reverting to incremental calculation of the mean value.
}

// aggregation evaluates sum, avg, count, stdvar, stddev or quantile at one timestep on inputMatrix.
// These functions produce one output series for each group specified in the expression, with just the labels from `by(...)`.
// outputMatrix should be already populated with grouping labels; groups is one-to-one with outputMatrix.
// seriesToResult maps inputMatrix indexes to outputMatrix indexes.
func (ev *evaluator) aggregation(e *parser.AggregateExpr, q float64, inputMatrix, outputMatrix Matrix, seriesToResult []int, groups []groupedAggregation, enh *EvalNodeHelper) annotations.Annotations {
	op := e.Op
	var annos annotations.Annotations
	for i := range groups {
		groups[i].seen = false
	}

	if e.Op == parser.QUANTILE && (math.IsNaN(q) || q < 0 || q > 1) {
		annos.Add(annotations.NewInvalidQuantileWarning(q, e.Param.PositionRange()))
	}

	for si := range inputMatrix {
		f, h, ok := ev.nextValues(enh.Ts, &inputMatrix[si])
		if !ok {
			continue
		}

		group := &groups[seriesToResult[si]]
		// Initialize this group if it's the first time we've seen it.
		if !group.seen {
			*group = groupedAggregation{
				seen:                   true,
				floatValue:             f,
				floatMean:              f,
				incompatibleHistograms: false,
				groupCount:             1,
			}
			switch op {
			case parser.AVG, parser.SUM:
				if h == nil {
					group.hasFloat = true
				} else {
					group.histogramValue = h.Copy()
					group.hasHistogram = true

Review comment (refactor):
Since this annotation is not specific to the affected series, we need at at most once. I think we can do this once ahead of the iteration by checking the max and min and if any is NaN. (See my idea about the interface above.)

If we have many steps, adding all those annotations, that then are promptly removed as duplicates again, could be quite some churn.

Code after:
	hasFloat               bool // Has at least 1 float64 sample aggregated.
	hasHistogram           bool // Has at least 1 histogram sample aggregated.
	incompatibleHistograms bool // If true, group has seen mixed exponential and custom buckets, or incompatible custom buckets.
	groupAggrComplete      bool // Used by LIMITK to short-cut series loop when we've reached K elem on every group.
	incrementalMean        bool // True after reverting to incremental calculation of the mean value.
}

// aggregation evaluates sum, avg, count, stdvar, stddev or quantile at one timestep on inputMatrix.
// These functions produce one output series for each group specified in the expression, with just the labels from `by(...)`.
// outputMatrix should be already populated with grouping labels; groups is one-to-one with outputMatrix.
// seriesToResult maps inputMatrix indexes to outputMatrix indexes.
func (ev *evaluator) aggregation(e *parser.AggregateExpr, q float64, inputMatrix, outputMatrix Matrix, seriesToResult []int, groups []groupedAggregation, enh *EvalNodeHelper) annotations.Annotations {
	op := e.Op
	var annos annotations.Annotations
	for i := range groups {
		groups[i].seen = false
	}

	for si := range inputMatrix {
		f, h, ok := ev.nextValues(enh.Ts, &inputMatrix[si])
		if !ok {
			continue
		}

		group := &groups[seriesToResult[si]]
		// Initialize this group if it's the first time we've seen it.
		if !group.seen {
			*group = groupedAggregation{
				seen:                   true,
				floatValue:             f,
				floatMean:              f,
				incompatibleHistograms: false,
				groupCount:             1,
			}
			switch op {
			case parser.AVG, parser.SUM:
				if h == nil {
					group.hasFloat = true
				} else {
					group.histogramValue = h.Copy()
					group.hasHistogram = true
				}
			case parser.STDVAR, parser.STDDEV:
				switch {
				case h != nil:
					// Ignore histograms for STDVAR and STDDEV.
					group.seen = false
					if op == parser.STDVAR {
						annos.Add(annotations.NewHistogramIgnoredInAggregationInfo("stdvar", e.Expr.PositionRange()))
					} else {
						annos.Add(annotations.NewHistogramIgnoredInAggregationInfo("stddev", e.Expr.PositionRange()))