RepoMirrors
/
prometheus
mirror of https://github.com/prometheus/prometheus
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

promql: simplify k/q parameter to topk/bottomk/quantile 

Pass it as a float64 not as interface{}.
Make k a simple int, since that is the parameter to make().
Pull invalid quantile warning out of the loop.

Signed-off-by: Bryan Boreham <bjboreham@gmail.com>
This commit is contained in:
Bryan Boreham 2024-03-01 16:01:20 +00:00
parent b3bda7df4b
commit cb6c4b3092
1 changed files with 16 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
promql/engine.go
View File

@ -1295,7 +1295,7 @@ func (ev *evaluator) rangeEvalAgg(aggExpr *parser.AggregateExpr, sortedGrouping
originalNumSamples := ev.currentSamples originalNumSamples := ev.currentSamples
var warnings annotations.Annotations var warnings annotations.Annotations
// param is the number k for topk/bottomk. // param is the number k for topk/bottomk, or q for quantile.
var param float64 var param float64
if aggExpr.Param != nil { if aggExpr.Param != nil {
val, ws := ev.eval(aggExpr.Param) val, ws := ev.eval(aggExpr.Param)
@ -2698,26 +2698,29 @@ type groupedAggregation struct {
// aggregation evaluates an aggregation operation on a Vector. The provided grouping labels // aggregation evaluates an aggregation operation on a Vector. The provided grouping labels
// must be sorted. // must be sorted.
func (ev *evaluator) aggregation(e *parser.AggregateExpr, grouping []string, param interface{}, inputMatrix Matrix, seriesHelper []EvalSeriesHelper, enh *EvalNodeHelper, seriess map[uint64]Series) (Matrix, annotations.Annotations) { func (ev *evaluator) aggregation(e *parser.AggregateExpr, grouping []string, q float64, inputMatrix Matrix, seriesHelper []EvalSeriesHelper, enh *EvalNodeHelper, seriess map[uint64]Series) (Matrix, annotations.Annotations) {
op := e.Op op := e.Op
without := e.Without without := e.Without
var annos annotations.Annotations var annos annotations.Annotations
result := map[uint64]*groupedAggregation{} result := map[uint64]*groupedAggregation{}
orderedResult := []*groupedAggregation{} orderedResult := []*groupedAggregation{}
var k int64 k := 1
if op == parser.TOPK || op == parser.BOTTOMK { if op == parser.TOPK || op == parser.BOTTOMK {
f := param.(float64) if !convertibleToInt64(q) {
if !convertibleToInt64(f) { ev.errorf("Scalar value %v overflows int64", q)
ev.errorf("Scalar value %v overflows int64", f) }
k = int(q)
if k > len(inputMatrix) {
k = len(inputMatrix)
} }
k = int64(f)
if k < 1 { if k < 1 {
return nil, annos return nil, annos
} }
} }
var q float64
if op == parser.QUANTILE { if op == parser.QUANTILE {
q = param.(float64) if math.IsNaN(q) || q < 0 || q > 1 {
annos.Add(annotations.NewInvalidQuantileWarning(q, e.Param.PositionRange()))
}
} }
for si, series := range inputMatrix { for si, series := range inputMatrix {
@ -2766,25 +2769,17 @@ func (ev *evaluator) aggregation(e *parser.AggregateExpr, grouping []string, par
newAgg.groupCount = 0 newAgg.groupCount = 0
} }
inputVecLen := int64(len(inputMatrix))
resultSize := k
switch {
case k > inputVecLen:
resultSize = inputVecLen
case k == 0:
resultSize = 1
}
switch op { switch op {
case parser.STDVAR, parser.STDDEV: case parser.STDVAR, parser.STDDEV:
newAgg.floatValue = 0 newAgg.floatValue = 0
case parser.TOPK, parser.QUANTILE: case parser.TOPK, parser.QUANTILE:
newAgg.heap = make(vectorByValueHeap, 1, resultSize) newAgg.heap = make(vectorByValueHeap, 1, k)
newAgg.heap[0] = Sample{ newAgg.heap[0] = Sample{
F: s.F, F: s.F,
Metric: s.Metric, Metric: s.Metric,
} }
case parser.BOTTOMK: case parser.BOTTOMK:
newAgg.reverseHeap = make(vectorByReverseValueHeap, 1, resultSize) newAgg.reverseHeap = make(vectorByReverseValueHeap, 1, k)
newAgg.reverseHeap[0] = Sample{ newAgg.reverseHeap[0] = Sample{
F: s.F, F: s.F,
Metric: s.Metric, Metric: s.Metric,
@ -2876,7 +2871,7 @@ func (ev *evaluator) aggregation(e *parser.AggregateExpr, grouping []string, par
case parser.TOPK: case parser.TOPK:
// We build a heap of up to k elements, with the smallest element at heap[0]. // We build a heap of up to k elements, with the smallest element at heap[0].
switch { switch {
case int64(len(group.heap)) < k: case len(group.heap) < k:
heap.Push(&group.heap, &Sample{ heap.Push(&group.heap, &Sample{
F: s.F, F: s.F,
Metric: s.Metric, Metric: s.Metric,
@ -2895,7 +2890,7 @@ func (ev *evaluator) aggregation(e *parser.AggregateExpr, grouping []string, par
case parser.BOTTOMK: case parser.BOTTOMK:
// We build a heap of up to k elements, with the biggest element at heap[0]. // We build a heap of up to k elements, with the biggest element at heap[0].
switch { switch {
case int64(len(group.reverseHeap)) < k: case len(group.reverseHeap) < k:
heap.Push(&group.reverseHeap, &Sample{ heap.Push(&group.reverseHeap, &Sample{
F: s.F, F: s.F,
Metric: s.Metric, Metric: s.Metric,
@ -2999,9 +2994,6 @@ func (ev *evaluator) aggregation(e *parser.AggregateExpr, grouping []string, par
continue // Bypass default append. continue // Bypass default append.
case parser.QUANTILE: case parser.QUANTILE:
if math.IsNaN(q) || q < 0 || q > 1 {
annos.Add(annotations.NewInvalidQuantileWarning(q, e.Param.PositionRange()))
}
aggr.floatValue = quantile(q, aggr.heap) aggr.floatValue = quantile(q, aggr.heap)
case parser.SUM: case parser.SUM: