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

Code before:
// For a range query, aggregates output in the seriess map.
func (ev *evaluator) aggregationK(e *parser.AggregateExpr, fParam float64, inputMatrix Matrix, seriesToResult []int, groups []groupedAggregation, enh *EvalNodeHelper, seriess map[uint64]Series) (Matrix, annotations.Annotations) {
	op := e.Op
	var s Sample
	var annos annotations.Annotations
	// Used to short-cut the loop for LIMITK if we already collected k elements for every group
	groupsRemaining := len(groups)
	for i := range groups {
		groups[i].seen = false
	}
	// advanceRemainingSeries discards any values at the current timestamp `ts`
	// for the remaining input series. In range queries, if these values are not
	// consumed now, they will no longer be accessible in the next evaluation step.
	advanceRemainingSeries := func(ts int64, startIdx int) {
		for i := startIdx; i < len(inputMatrix); i++ {
			_, _, _ = ev.nextValues(ts, &inputMatrix[i])
		}
	}

	switch op {
	case parser.TOPK, parser.BOTTOMK, parser.LIMITK:
		if !convertibleToInt64(fParam) {
			ev.errorf("Scalar value %v overflows int64", fParam)
		}
	case parser.LIMIT_RATIO:
		if math.IsNaN(fParam) {
			ev.errorf("Ratio value %v is NaN", fParam)
		}
	}

seriesLoop:
	for si := range inputMatrix {
		f, h, ok := ev.nextValues(enh.Ts, &inputMatrix[si])
		if !ok {
			continue
		}
		s = Sample{Metric: inputMatrix[si].Metric, F: f, H: h, DropName: inputMatrix[si].DropName}

		var k int64
		var r float64
		switch op {
		case parser.TOPK, parser.BOTTOMK, parser.LIMITK:
			k = int64(fParam)
			if k > int64(len(inputMatrix)) {
				k = int64(len(inputMatrix))
			}
			if k < 1 {
				if enh.Ts != ev.endTimestamp {
					advanceRemainingSeries(enh.Ts, si+1)
				}
				return nil, annos

Review comment (suggestion):
I think we can just check for a `if params.HasAnyNaN() {` here https://github.com/prometheus/prometheus/blob/4aee718013c52bef1ccb336bcc21886336e6252e/promql/engine.go#L1438 



Code after:
	return annos
}

// aggregationK evaluates topk, bottomk, limitk, or limit_ratio at one timestep on inputMatrix.
// Output that has the same labels as the input, but just k of them per group.
// seriesToResult maps inputMatrix indexes to groups indexes.
// For an instant query, returns a Matrix in descending order for topk or ascending for bottomk, or without any order for limitk / limit_ratio.
// For a range query, aggregates output in the seriess map.
func (ev *evaluator) aggregationK(e *parser.AggregateExpr, fParam float64, inputMatrix Matrix, seriesToResult []int, groups []groupedAggregation, enh *EvalNodeHelper, seriess map[uint64]Series) (Matrix, annotations.Annotations) {
	op := e.Op
	var s Sample
	var annos annotations.Annotations
	// Used to short-cut the loop for LIMITK if we already collected k elements for every group
	groupsRemaining := len(groups)
	for i := range groups {
		groups[i].seen = false
	}
	// advanceRemainingSeries discards any values at the current timestamp `ts`
	// for the remaining input series. In range queries, if these values are not
	// consumed now, they will no longer be accessible in the next evaluation step.
	advanceRemainingSeries := func(ts int64, startIdx int) {
		for i := startIdx; i < len(inputMatrix); i++ {
			_, _, _ = ev.nextValues(ts, &inputMatrix[i])
		}
	}

seriesLoop:
	for si := range inputMatrix {
		f, h, ok := ev.nextValues(enh.Ts, &inputMatrix[si])
		if !ok {
			continue
		}
		s = Sample{Metric: inputMatrix[si].Metric, F: f, H: h, DropName: inputMatrix[si].DropName}

		var k int64
		var r float64
		switch op {
		case parser.TOPK, parser.BOTTOMK, parser.LIMITK:
			k = int64(fParam)
			if k > int64(len(inputMatrix)) {
				k = int64(len(inputMatrix))
			}
			if k < 1 {
				if enh.Ts != ev.endTimestamp {
					advanceRemainingSeries(enh.Ts, si+1)
				}
				return nil, annos
			}
		case parser.LIMIT_RATIO:
			switch {
			case fParam == 0: