From 35f592a02c86932ea8f756b0e0a6d88391d341c1 Mon Sep 17 00:00:00 2001 From: Aliaksandr Valialkin Date: Sun, 25 Feb 2024 01:47:09 +0200 Subject: [PATCH] app/vmselect/promql: properly handle args in count_values_over_time() function Prevsiously they were swapped - the first arg should be the label name and the second arg should be label filters This is a follow-up for e389b7b959e8144fdff5075bf7a5a39b2b0c6dd3 Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5847 --- app/vmselect/promql/eval.go | 4 +- app/vmselect/promql/exec_test.go | 5 +- app/vmselect/promql/rollup.go | 6 +- go.mod | 2 +- go.sum | 4 +- .../VictoriaMetrics/metricsql/optimizer.go | 161 ++++++++++++------ .../VictoriaMetrics/metricsql/rollup.go | 2 +- vendor/modules.txt | 2 +- 8 files changed, 121 insertions(+), 65 deletions(-) diff --git a/app/vmselect/promql/eval.go b/app/vmselect/promql/eval.go index da55ad86d..e4ab2f42e 100644 --- a/app/vmselect/promql/eval.go +++ b/app/vmselect/promql/eval.go @@ -314,7 +314,7 @@ func evalExprInternal(qt *querytracer.Tracer, ec *EvalConfig, e metricsql.Expr) } rf, err := nrf(args) if err != nil { - return nil, err + return nil, fmt.Errorf("cannot evaluate args for %q: %w", fe.AppendString(nil), err) } rv, err := evalRollupFunc(qt, ec, fe.Name, rf, e, re, nil) if err != nil { @@ -395,7 +395,7 @@ func evalAggrFunc(qt *querytracer.Tracer, ec *EvalConfig, ae *metricsql.AggrFunc } rf, err := nrf(args) if err != nil { - return nil, err + return nil, fmt.Errorf("cannot evaluate args for aggregate func %q: %w", ae.AppendString(nil), err) } iafc := newIncrementalAggrFuncContext(ae, callbacks) return evalRollupFunc(qt, ec, fe.Name, rf, ae, re, iafc) diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go index 472f7d210..c1f55fb94 100644 --- a/app/vmselect/promql/exec_test.go +++ b/app/vmselect/promql/exec_test.go @@ -5824,7 +5824,10 @@ func TestExecSuccess(t *testing.T) { }) t.Run(`count_values_over_time`, func(t *testing.T) { t.Parallel() - q := `sort_by_label(count_values_over_time(round(label_set(rand(0), "x", "y"), 0.4)[200s:5s], "foo"), "foo")` + q := `sort_by_label( + count_values_over_time("foo", round(label_set(rand(0), "x", "y"), 0.4)[200s:5s]), + "foo", + )` r1 := netstorage.Result{ MetricName: metricNameExpected, Values: []float64{4, 8, 7, 6, 10, 9}, diff --git a/app/vmselect/promql/rollup.go b/app/vmselect/promql/rollup.go index 6e7c04a8a..4700ba643 100644 --- a/app/vmselect/promql/rollup.go +++ b/app/vmselect/promql/rollup.go @@ -1449,11 +1449,11 @@ func newRollupCountValues(args []interface{}) (rollupFunc, error) { if err := expectRollupArgsNum(args, 2); err != nil { return nil, err } - tssLabelNum, ok := args[1].([]*timeseries) + tssLabelNum, ok := args[0].([]*timeseries) if !ok { - return nil, fmt.Errorf(`unexpected type for labelName arg; got %T; want %T`, args[1], tssLabelNum) + return nil, fmt.Errorf(`unexpected type for labelName arg; got %T; want %T`, args[0], tssLabelNum) } - labelName, err := getString(tssLabelNum, 1) + labelName, err := getString(tssLabelNum, 0) if err != nil { return nil, fmt.Errorf("cannot get labelName: %w", err) } diff --git a/go.mod b/go.mod index dd436fb3b..98c4dfb0a 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/VictoriaMetrics/easyproto v0.1.4 github.com/VictoriaMetrics/fastcache v1.12.2 github.com/VictoriaMetrics/metrics v1.33.0 - github.com/VictoriaMetrics/metricsql v0.74.0 + github.com/VictoriaMetrics/metricsql v0.75.0 github.com/aws/aws-sdk-go-v2 v1.25.2 github.com/aws/aws-sdk-go-v2/config v1.27.4 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.16.6 diff --git a/go.sum b/go.sum index 21b607f8c..477a8b9ec 100644 --- a/go.sum +++ b/go.sum @@ -70,8 +70,8 @@ github.com/VictoriaMetrics/fastcache v1.12.2/go.mod h1:AmC+Nzz1+3G2eCPapF6UcsnkT github.com/VictoriaMetrics/metrics v1.24.0/go.mod h1:eFT25kvsTidQFHb6U0oa0rTrDRdz4xTYjpL8+UPohys= github.com/VictoriaMetrics/metrics v1.33.0 h1:EnkDEaGiL2u95t+W76GfecC/LMYpy+tFrexYzBWQIAc= github.com/VictoriaMetrics/metrics v1.33.0/go.mod h1:r7hveu6xMdUACXvB8TYdAj8WEsKzWB0EkpJN+RDtOf8= -github.com/VictoriaMetrics/metricsql v0.74.0 h1:bVO7USXBBYEuEHQ3PZg/6216j0DvblZM+Q8sTRECkv0= -github.com/VictoriaMetrics/metricsql v0.74.0/go.mod h1:k4UaP/+CjuZslIjd+kCigNG9TQmUqh5v0TP/nMEy90I= +github.com/VictoriaMetrics/metricsql v0.75.0 h1:ZRt2tE+GwLqa7fhY7A3wxIZe5Bdb6KiCm9MRHilv4eo= +github.com/VictoriaMetrics/metricsql v0.75.0/go.mod h1:k4UaP/+CjuZslIjd+kCigNG9TQmUqh5v0TP/nMEy90I= github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1ow= github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= diff --git a/vendor/github.com/VictoriaMetrics/metricsql/optimizer.go b/vendor/github.com/VictoriaMetrics/metricsql/optimizer.go index 54302a2df..e0deae7b1 100644 --- a/vendor/github.com/VictoriaMetrics/metricsql/optimizer.go +++ b/vendor/github.com/VictoriaMetrics/metricsql/optimizer.go @@ -60,18 +60,20 @@ func optimizeInplace(e Expr) { optimizeInplace(t.Expr) optimizeInplace(t.At) case *FuncExpr: - for _, arg := range t.Args { - optimizeInplace(arg) - } + optimizeArgsInplace(t.Args) case *AggrFuncExpr: - for _, arg := range t.Args { - optimizeInplace(arg) - } + optimizeArgsInplace(t.Args) case *BinaryOpExpr: optimizeInplace(t.Left) optimizeInplace(t.Right) lfs := getCommonLabelFilters(t) - pushdownBinaryOpFiltersInplace(t, lfs) + pushdownBinaryOpFiltersInplace(lfs, t) + } +} + +func optimizeArgsInplace(args []Expr) { + for _, arg := range args { + optimizeInplace(arg) } } @@ -82,19 +84,24 @@ func getCommonLabelFilters(e Expr) []LabelFilter { case *RollupExpr: return getCommonLabelFilters(t.Expr) case *FuncExpr: + args := t.Args switch strings.ToLower(t.Name) { case "label_set": - return getCommonLabelFiltersForLabelSet(t.Args) + return getCommonLabelFiltersForLabelSet(args) case "label_replace", "label_join", "label_map", "label_match", "label_mismatch", "label_transform": - return getCommonLabelFiltersForLabelReplace(t.Args) + return getCommonLabelFiltersForLabelReplace(args) case "label_copy", "label_move": - return getCommonLabelFiltersForLabelCopy(t.Args) + return getCommonLabelFiltersForLabelCopy(args) case "label_del", "label_uppercase", "label_lowercase", "labels_equal": - return getCommonLabelFiltersForLabelDel(t.Args) + return getCommonLabelFiltersForLabelDel(args) case "label_keep": - return getCommonLabelFiltersForLabelKeep(t.Args) + return getCommonLabelFiltersForLabelKeep(args) + case "count_values_over_time": + return getCommonLabelFiltersForCountValuesOverTime(args) + case "range_normalize", "union", "": + return intersectLabelFiltersForAllArgs(args) default: - arg := getFuncArgForOptimization(t.Name, t.Args) + arg := getFuncArgForOptimization(t.Name, args) if arg == nil { return nil } @@ -102,12 +109,16 @@ func getCommonLabelFilters(e Expr) []LabelFilter { } case *AggrFuncExpr: args := t.Args - if len(args) > 0 && canAcceptMultipleArgsForAggrFunc(t.Name) { - lfs := getCommonLabelFilters(args[0]) - for _, arg := range args[1:] { - lfsNext := getCommonLabelFilters(arg) - lfs = intersectLabelFilters(lfs, lfsNext) + if strings.ToLower(t.Name) == "count_values" { + if len(args) != 2 { + return nil } + lfs := getCommonLabelFilters(args[1]) + lfs = dropLabelFiltersForLabelName(lfs, args[0]) + return trimFiltersByAggrModifier(lfs, t) + } + if canAcceptMultipleArgsForAggrFunc(t.Name) { + lfs := intersectLabelFiltersForAllArgs(args) return trimFiltersByAggrModifier(lfs, t) } arg := getFuncArgForOptimization(t.Name, args) @@ -174,6 +185,26 @@ func getCommonLabelFilters(e Expr) []LabelFilter { } } +func intersectLabelFiltersForAllArgs(args []Expr) []LabelFilter { + if len(args) == 0 { + return nil + } + lfs := getCommonLabelFilters(args[0]) + for _, arg := range args[1:] { + lfsNext := getCommonLabelFilters(arg) + lfs = intersectLabelFilters(lfs, lfsNext) + } + return lfs +} + +func getCommonLabelFiltersForCountValuesOverTime(args []Expr) []LabelFilter { + if len(args) != 2 { + return nil + } + lfs := getCommonLabelFilters(args[1]) + return dropLabelFiltersForLabelName(lfs, args[0]) +} + func getCommonLabelFiltersForLabelKeep(args []Expr) []LabelFilter { if len(args) == 0 { return nil @@ -318,11 +349,11 @@ func PushdownBinaryOpFilters(e Expr, commonFilters []LabelFilter) Expr { return e } eCopy := Clone(e) - pushdownBinaryOpFiltersInplace(eCopy, commonFilters) + pushdownBinaryOpFiltersInplace(commonFilters, eCopy) return eCopy } -func pushdownBinaryOpFiltersInplace(e Expr, lfs []LabelFilter) { +func pushdownBinaryOpFiltersInplace(lfs []LabelFilter, e Expr) { if len(lfs) == 0 { return } @@ -334,62 +365,84 @@ func pushdownBinaryOpFiltersInplace(e Expr, lfs []LabelFilter) { t.LabelFilterss[i] = lfsLocal } case *RollupExpr: - pushdownBinaryOpFiltersInplace(t.Expr, lfs) + pushdownBinaryOpFiltersInplace(lfs, t.Expr) case *FuncExpr: + args := t.Args switch strings.ToLower(t.Name) { case "label_set": - pushdownLabelFiltersForLabelSet(t.Args, lfs) + pushdownLabelFiltersForLabelSet(lfs, args) case "label_replace", "label_join", "label_map", "label_match", "label_mismatch", "label_transform": - pushdownLabelFiltersForLabelReplace(t.Args, lfs) + pushdownLabelFiltersForLabelReplace(lfs, args) case "label_copy", "label_move": - pushdownLabelFiltersForLabelCopy(t.Args, lfs) + pushdownLabelFiltersForLabelCopy(lfs, args) case "label_del", "label_uppercase", "label_lowercase", "labels_equal": - pushdownLabelFiltersForLabelDel(t.Args, lfs) + pushdownLabelFiltersForLabelDel(lfs, args) case "label_keep": - pushdownLabelFiltersForLabelKeep(t.Args, lfs) + pushdownLabelFiltersForLabelKeep(lfs, args) + case "count_values_over_time": + pushdownLabelFiltersForCountValuesOverTime(lfs, args) + case "range_normalize", "union", "": + pushdownLabelFiltersForAllArgs(lfs, args) default: - arg := getFuncArgForOptimization(t.Name, t.Args) + arg := getFuncArgForOptimization(t.Name, args) if arg != nil { - pushdownBinaryOpFiltersInplace(arg, lfs) + pushdownBinaryOpFiltersInplace(lfs, arg) } } case *AggrFuncExpr: lfs = trimFiltersByAggrModifier(lfs, t) args := t.Args - if len(args) > 0 && canAcceptMultipleArgsForAggrFunc(t.Name) { - for _, arg := range args { - pushdownBinaryOpFiltersInplace(arg, lfs) + if strings.ToLower(t.Name) == "count_values" { + if len(args) == 2 { + lfs = dropLabelFiltersForLabelName(lfs, args[0]) + pushdownBinaryOpFiltersInplace(lfs, args[1]) } + } else if canAcceptMultipleArgsForAggrFunc(t.Name) { + pushdownLabelFiltersForAllArgs(lfs, args) } else { arg := getFuncArgForOptimization(t.Name, args) if arg != nil { - pushdownBinaryOpFiltersInplace(arg, lfs) + pushdownBinaryOpFiltersInplace(lfs, arg) } } case *BinaryOpExpr: lfs = TrimFiltersByGroupModifier(lfs, t) - pushdownBinaryOpFiltersInplace(t.Left, lfs) - pushdownBinaryOpFiltersInplace(t.Right, lfs) + pushdownBinaryOpFiltersInplace(lfs, t.Left) + pushdownBinaryOpFiltersInplace(lfs, t.Right) } } -func pushdownLabelFiltersForLabelKeep(args []Expr, lfs []LabelFilter) { +func pushdownLabelFiltersForAllArgs(lfs []LabelFilter, args []Expr) { + for _, arg := range args { + pushdownBinaryOpFiltersInplace(lfs, arg) + } +} + +func pushdownLabelFiltersForCountValuesOverTime(lfs []LabelFilter, args []Expr) { + if len(args) != 2 { + return + } + lfs = dropLabelFiltersForLabelName(lfs, args[0]) + pushdownBinaryOpFiltersInplace(lfs, args[1]) +} + +func pushdownLabelFiltersForLabelKeep(lfs []LabelFilter, args []Expr) { if len(args) == 0 { return } lfs = keepLabelFiltersForLabelNames(lfs, args[1:]) - pushdownBinaryOpFiltersInplace(args[0], lfs) + pushdownBinaryOpFiltersInplace(lfs, args[0]) } -func pushdownLabelFiltersForLabelDel(args []Expr, lfs []LabelFilter) { +func pushdownLabelFiltersForLabelDel(lfs []LabelFilter, args []Expr) { if len(args) == 0 { return } lfs = dropLabelFiltersForLabelNames(lfs, args[1:]) - pushdownBinaryOpFiltersInplace(args[0], lfs) + pushdownBinaryOpFiltersInplace(lfs, args[0]) } -func pushdownLabelFiltersForLabelCopy(args []Expr, lfs []LabelFilter) { +func pushdownLabelFiltersForLabelCopy(lfs []LabelFilter, args []Expr) { if len(args) == 0 { return } @@ -403,18 +456,18 @@ func pushdownLabelFiltersForLabelCopy(args []Expr, lfs []LabelFilter) { labelNames = append(labelNames, args[i+1]) } lfs = dropLabelFiltersForLabelNames(lfs, labelNames) - pushdownBinaryOpFiltersInplace(arg, lfs) + pushdownBinaryOpFiltersInplace(lfs, arg) } -func pushdownLabelFiltersForLabelReplace(args []Expr, lfs []LabelFilter) { +func pushdownLabelFiltersForLabelReplace(lfs []LabelFilter, args []Expr) { if len(args) < 2 { return } lfs = dropLabelFiltersForLabelName(lfs, args[1]) - pushdownBinaryOpFiltersInplace(args[0], lfs) + pushdownBinaryOpFiltersInplace(lfs, args[0]) } -func pushdownLabelFiltersForLabelSet(args []Expr, lfs []LabelFilter) { +func pushdownLabelFiltersForLabelSet(lfs []LabelFilter, args []Expr) { if len(args) == 0 { return } @@ -425,7 +478,7 @@ func pushdownLabelFiltersForLabelSet(args []Expr, lfs []LabelFilter) { labelNames = append(labelNames, args[i]) } lfs = dropLabelFiltersForLabelNames(lfs, labelNames) - pushdownBinaryOpFiltersInplace(arg, lfs) + pushdownBinaryOpFiltersInplace(lfs, arg) } func intersectLabelFilters(lfsA, lfsB []LabelFilter) []LabelFilter { @@ -591,13 +644,13 @@ func getAggrArgIdxForOptimization(funcName string, args []Expr) int { "limitk", "outliers_mad", "outliersk", "quantile", "topk", "topk_avg", "topk_max", "topk_median", "topk_last", "topk_min": return 1 - case "count_values": - return -1 case "quantiles": return len(args) - 1 + case "count_values": + panic(fmt.Errorf("BUG: count_values must be already handled")) default: - if len(args) > 1 && canAcceptMultipleArgsForAggrFunc(funcName) { - panic(fmt.Errorf("BUG: %d > 1 args passed to aggregate function %q; this case must be already handled", len(args), funcName)) + if canAcceptMultipleArgsForAggrFunc(funcName) { + panic(fmt.Errorf("BUG: %s must be already handled", funcName)) } return 0 } @@ -616,6 +669,8 @@ func canAcceptMultipleArgsForAggrFunc(funcName string) bool { func getRollupArgIdxForOptimization(funcName string, args []Expr) int { // This must be kept in sync with GetRollupArgIdx() switch strings.ToLower(funcName) { + case "count_values_over_time": + panic(fmt.Errorf("BUG: count_values_over_time must be already handled")) case "absent_over_time": return -1 case "quantile_over_time", "aggr_over_time", @@ -632,11 +687,11 @@ func getTransformArgIdxForOptimization(funcName string, args []Expr) int { switch strings.ToLower(funcName) { case "label_copy", "label_del", "label_join", "label_keep", "label_lowercase", "label_map", "label_match", "label_mismatch", "label_move", "label_replace", "label_set", "label_transform", - "label_uppercase", "labels_equal": - panic(fmt.Errorf("BUG: unexpected funcName passed to getTransformArgIdxForOptimization: %s", funcName)) - case "drop_common_labels", "range_normalize": + "label_uppercase", "labels_equal", "range_normalize", "", "union": + panic(fmt.Errorf("BUG: %s must be already handled", funcName)) + case "drop_common_labels": return -1 - case "", "absent", "scalar", "union", "vector": + case "absent", "scalar": return -1 case "end", "now", "pi", "ru", "start", "step", "time": return -1 @@ -647,8 +702,6 @@ func getTransformArgIdxForOptimization(funcName string, args []Expr) int { return 1 case "histogram_quantiles": return len(args) - 1 - case "label_graphite_group": - return 0 default: return 0 } diff --git a/vendor/github.com/VictoriaMetrics/metricsql/rollup.go b/vendor/github.com/VictoriaMetrics/metricsql/rollup.go index a7b5b9f57..d7cbd702a 100644 --- a/vendor/github.com/VictoriaMetrics/metricsql/rollup.go +++ b/vendor/github.com/VictoriaMetrics/metricsql/rollup.go @@ -104,7 +104,7 @@ func GetRollupArgIdx(fe *FuncExpr) int { return -1 } switch funcName { - case "quantile_over_time", "aggr_over_time", + case "quantile_over_time", "aggr_over_time", "count_values_over_time", "hoeffding_bound_lower", "hoeffding_bound_upper": return 1 case "quantiles_over_time": diff --git a/vendor/modules.txt b/vendor/modules.txt index 9f1ef0bcd..d9c8eb7cd 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -102,7 +102,7 @@ github.com/VictoriaMetrics/fastcache # github.com/VictoriaMetrics/metrics v1.33.0 ## explicit; go 1.17 github.com/VictoriaMetrics/metrics -# github.com/VictoriaMetrics/metricsql v0.74.0 +# github.com/VictoriaMetrics/metricsql v0.75.0 ## explicit; go 1.13 github.com/VictoriaMetrics/metricsql github.com/VictoriaMetrics/metricsql/binaryop