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

Code before:
// 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])
		}
	}

	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 {

Review comment (suggestion):
As I mentioned in https://cloud-native.slack.com/archives/C01AUBA4PFE/p1749800759431499?thread_ts=1749797556.935899&cid=C01AUBA4PFE, I think we want to align their behaviors' with quantile's, as quantile still does some pre-checks here https://github.com/prometheus/prometheus/blob/4aee718013c52bef1ccb336bcc21886336e6252e/promql/engine.go#L1452 after https://github.com/prometheus/prometheus/pull/16404

Still, there is a difference; quantile uses annotations for that, but here we error, which still isn't consistent IMHO.

Maybe switching to annotations will make the operators more resilient to missing data/gaps in fParam as was discussed here https://github.com/prometheus/prometheus/pull/16404#discussion_r2058346491

WDYT @beorn7 @NeerajGartia21 about this

(this is not a blocker of the PR, we can discuss this somewhere else, I'm ok to just restore the old behavior in this PR (if others think we should of course))

Code after:
		addToSeries(ss, enh.Ts, aggr.floatValue, aggr.histogramValue, numSteps)
		ss.DropName = inputMatrix[ri].DropName
	}

	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