app/vmselect/promql: check for ranged vectors in aggr funcs if implicit conversions are disabled (#6450)

Check for ranged vector arguments in aggregate expressions when
`-search.disableImplicitConversion` or `-search.logImplicitConversion`
are enabled.
 For example, `sum(up[5m])` will fail to execute if these flags are set.

### Describe Your Changes

Please provide a brief description of the changes you made. Be as
specific as possible to help others understand the purpose and impact of
your modifications.

### Checklist

The following checks are **mandatory**:

- [*] My change adheres [VictoriaMetrics contributing
guidelines](https://docs.victoriametrics.com/contributing/).

---------

Signed-off-by: hagen1778 <roman@victoriametrics.com>

app/vmselect/promql/exec.go

@@ -71,12 +71,13 @@ func Exec(qt *querytracer.Tracer, ec *EvalConfig, q string, isFirstPointOnly boo
 	}
 
 	if *disableImplicitConversion || *logImplicitConversion {
-		complete := isSubQueryComplete(e, false)
-		if !complete && *disableImplicitConversion {
-			return nil, fmt.Errorf("query contains subquery that requires implicit conversion and is rejected according to `-search.disableImplicitConversion=true` setting. See https://docs.victoriametrics.com/metricsql/#subqueries for details")
+		noConversion := noImplicitConversionRequired(e, false)
+		if !noConversion && *disableImplicitConversion {
+			// we don't add query=%q to err message as it will be added by the caller
+			return nil, fmt.Errorf("query requires implicit conversion and is rejected according to `-search.disableImplicitConversion=true` setting. See https://docs.victoriametrics.com/metricsql/#implicit-query-conversions for details")
 		}
-		if !complete && *logImplicitConversion {
-			logger.Warnf("query=%q contains subquery that requires implicit conversion, see https://docs.victoriametrics.com/metricsql/#subqueries for details", e.AppendString(nil))
+		if !noConversion && *logImplicitConversion {
+			logger.Warnf("query=%q requires implicit conversion, see https://docs.victoriametrics.com/metricsql/#implicit-query-conversions for details", e.AppendString(nil))
 		}
 	}
 
@@ -422,37 +423,47 @@ func (pc *parseCache) Put(q string, pcv *parseCacheValue) {
 	pc.mu.Unlock()
 }
 
-// isSubQueryComplete checks if expr contains incomplete subquery
-func isSubQueryComplete(e metricsql.Expr, isSubExpr bool) bool {
+// noImplicitConversionRequired checks if expr requires implicit conversion
+func noImplicitConversionRequired(e metricsql.Expr, isSubExpr bool) bool {
 	switch exp := e.(type) {
 	case *metricsql.FuncExpr:
 		if isSubExpr {
 			return false
 		}
 		fe := e.(*metricsql.FuncExpr)
+		isRollupFn := getRollupFunc(fe.Name) != nil
 		for _, arg := range exp.Args {
-			if getRollupFunc(fe.Name) != nil {
+			_, isRollupExpr := arg.(*metricsql.RollupExpr)
+			if (isRollupExpr && !isRollupFn) || (!isRollupExpr && isRollupFn) {
+				return false
+			}
+			if isRollupFn {
 				isSubExpr = true
 			}
-			if !isSubQueryComplete(arg, isSubExpr) {
+			if !noImplicitConversionRequired(arg, isSubExpr) {
 				return false
 			}
 		}
 	case *metricsql.RollupExpr:
 		if _, ok := exp.Expr.(*metricsql.MetricExpr); ok {
-			return true
+			return exp.Step == nil
 		}
 		// exp.Step is optional in subqueries
 		if exp.Window == nil {
 			return false
 		}
-		return isSubQueryComplete(exp.Expr, false)
+		return noImplicitConversionRequired(exp.Expr, false)
 	case *metricsql.AggrFuncExpr:
 		if isSubExpr {
 			return false
 		}
 		for _, arg := range exp.Args {
-			if !isSubQueryComplete(arg, false) {
+			if re, ok := arg.(*metricsql.RollupExpr); ok {
+				if re.Window != nil {
+					return false
+				}
+			}
+			if !noImplicitConversionRequired(arg, false) {
 				return false
 			}
 		}
@@ -460,10 +471,10 @@ func isSubQueryComplete(e metricsql.Expr, isSubExpr bool) bool {
 		if isSubExpr {
 			return false
 		}
-		if !isSubQueryComplete(exp.Left, false) {
+		if !noImplicitConversionRequired(exp.Left, false) {
 			return false
 		}
-		if !isSubQueryComplete(exp.Right, false) {
+		if !noImplicitConversionRequired(exp.Right, false) {
 			return false
 		}
 	case *metricsql.MetricExpr:

app/vmselect/promql/exec_test.go

@@ -9433,50 +9433,51 @@ func testAddLabels(t *testing.T, mn *storage.MetricName, labels ...string) {
 	}
 }
 
-func TestIsSubQueryCompleteTrue(t *testing.T) {
+func TestNoImplicitConversionRequiredTrue(t *testing.T) {
 	f := func(q string) {
 		t.Helper()
 		e, err := metricsql.Parse(q)
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !isSubQueryComplete(e, false) {
-			t.Fatalf("query should be complete: %s", e.AppendString(nil))
+		if !noImplicitConversionRequired(e, false) {
+			t.Fatalf("query should require no implicit conversion: %s", e.AppendString(nil))
 		}
 	}
 
-	f("rate(http_total)")
+	f("http_total[5m]")
 	f("sum(http_total)")
+	f("sum(foo, bar)")
 	f("absent(http_total)")
 	f("rate(http_total[1m])")
 	f("avg_over_time(up[1m])")
-	f("sum(http_total[1m])")
-	f("sum(rate(http_total))")
+	f("sum(rate(http_total[1m]))")
 	f("sum(sum(http_total))")
 	f(`sum(sum_over_time(http_total[1m] )) by (instance)`)
 	f("sum(up{cluster='a'}[1m] or up{cluster='b'}[1m])")
 	f("(avg_over_time(alarm_test1[1m]) - avg_over_time(alarm_test1[1m] offset 5m)) > 0.1")
 	f("http_total[1m] offset 1m")
+	f("sum(http_total offset 1m)")
 
 	// subquery
-	f("rate(http_total)[5m:1m]")
+	f("rate(http_total[5m])[5m:1m]")
 	f("rate(sum(http_total)[5m:1m])")
-	f("rate(rate(http_total)[5m:1m])")
+	f("rate(rate(http_total[5m])[5m:1m])")
 	f("sum(rate(http_total[1m]))")
 	f("sum(rate(sum(http_total)[5m:1m]))")
-	f("rate(sum(rate(http_total))[5m:1m])")
+	f("rate(sum(rate(http_total[5m]))[5m:1m])")
 	f("rate(sum(sum(http_total))[5m:1m])")
-	f("rate(sum(rate(http_total))[5m:1m])")
+	f("rate(sum(rate(http_total[5m]))[5m:1m])")
 	f("rate(sum(sum(http_total))[5m:1m])")
 	f("avg_over_time(rate(http_total[5m])[5m:1m])")
 	f("delta(avg_over_time(up[1m])[5m:1m]) > 0.1")
 	f("avg_over_time(avg by (site) (metric)[2m:1m])")
 
 	f("sum(http_total)[5m:1m] offset 1m")
-	f("round(sum(sum_over_time(http_total[1m])) by (instance)) [5m:1m] offset 1m")
+	f("round(sum(sum_over_time(http_total[1m])) by (instance))[5m:1m] offset 1m")
 
 	f("rate(sum(http_total)[5m:1m]) - rate(sum(http_total)[5m:1m])")
-	f("avg_over_time((rate(http_total)-rate(http_total))[5m:1m])")
+	f("avg_over_time((rate(http_total[5m])-rate(http_total[5m]))[5m:1m])")
 
 	f("sum_over_time((up{cluster='a'} or up{cluster='b'})[5m:1m])")
 	f("sum_over_time((up{cluster='a'} or up{cluster='b'})[5m:1m])")
@@ -9495,17 +9496,25 @@ WITH (
 max_over_time(cpuIdle[1h:])`)
 }
 
-func TestIsSubQueryCompleteFalse(t *testing.T) {
+func TestNoImplicitConversionRequiredFalse(t *testing.T) {
 	f := func(q string) {
 		t.Helper()
 		e, err := metricsql.Parse(q)
 		if err != nil {
 			t.Fatal(err)
 		}
-		if isSubQueryComplete(e, false) {
-			t.Fatalf("expect to detect incomplete subquery: %s", e.AppendString(nil))
+		if noImplicitConversionRequired(e, false) {
+			t.Fatalf("query should have require implicit conversion: %s", e.AppendString(nil))
 		}
 	}
+	f("rate(http_total)[5m:1m]")
+
+	f("up[:5m]")
+	f("sum(up[:5m])")
+	f("absent(foo[5m])")
+	f("sum(up[5m])")
+	f("avg(foo[5m])")
+	f("sort(foo[5m])")
 
 	f("rate(sum(http_total))")
 	f("rate(rate(http_total))")
@@ -9514,6 +9523,13 @@ func TestIsSubQueryCompleteFalse(t *testing.T) {
 	f("rate(sum(sum(http_total)))")
 	f("avg_over_time(rate(http_total[5m]))")
 
+	f("rate(http_total)[5m:1m]")
+	f("rate(rate(http_total)[5m:1m])")
+	f("rate(sum(rate(http_total))[5m:1m])")
+	f("rate(sum(rate(http_total))[5m:1m])")
+	f("count_over_time(http_total)")
+	f("avg_over_time((rate(http_total)-rate(http_total))[5m:1m])")
+
 	// https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3974
 	f("sum(http_total) offset 1m")
 	f(`round(sum(sum_over_time(http_total[1m])) by (instance)) offset 1m`)

docs/CHANGELOG.md

@@ -37,6 +37,8 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * FEATURE: [vmauth](https://docs.victoriametrics.com/vmauth/): add `idleConnTimeout` flag set to 50s by default. It should reduce the probability of `broken pipe` or `connection reset by peer` errors in vmauth logs.
 * FEATURE: [vmauth](https://docs.victoriametrics.com/vmauth/): add auto request retry for trivial network errors, such as `broken pipe` and `connection reset` for requests to the configured backends.
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): increase default value of `-promscrape.maxDroppedTargets` command-line flag to 10_000 from 1000. This makes it easier to track down large number of dropped targets.
+* FEATURE: [vmsingle](https://docs.victoriametrics.com/single-server-victoriametrics/): check for ranged vector arguments in non-rollup expressions when `-search.disableImplicitConversion` or `-search.logImplicitConversion` are enabled. For example, `sum(up[5m])` or `absent(up[5m])` will fail to execute if these flags are set.
+* FEATURE: [vmsingle](https://docs.victoriametrics.com/single-server-victoriametrics/): validate that rollup expressions has ranged vector arguments passed when `-search.disableImplicitConversion` or `-search.logImplicitConversion` are enabled. For example, `rate(metric)` or `count_over_time(metric)` will fail to execute if these flags are set.
 
 * BUGFIX: all VictoriaMetrics components: prioritize `-configAuthKey` and `-reloadAuthKey` over `-httpAuth.*` settings. This change aligns behavior of mentioned flags with other auth flags like `-metricsAuthKey`, `-flagsAuthKey`, `-pprofAuthKey`. Check [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6329).
 * BUGFIX: [vmctl](https://docs.victoriametrics.com/vmctl/): add `--disable-progress-bar` global command-line flag. It can be used for disabling dynamic progress bar for all migration modes. `--vm-disable-progress-bar`  command-line flag is deprecated and will be removed in the future releases. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6367).

lib/streamaggr/streamaggr_test.go

@@ -251,6 +251,32 @@ func TestAggregatorsSuccess(t *testing.T) {
 		}
 	}
 
+	// rate with duplicated events
+	f(`     
+- interval: 1m
+  by: [cde]
+  outputs: [rate_sum, rate_avg]
+`, `
+foo{abc="123", cde="1"} 0  10
+foo{abc="123", cde="1"} 0  10
+`, `foo:1m_by_cde_rate_avg{cde="1"} 0
+foo:1m_by_cde_rate_sum{cde="1"} 0
+`, "11")
+
+	// rate with duplicated events
+	f(`     
+- interval: 1m
+  by: [cde]
+  outputs: [rate_sum, rate_avg]
+`, `
+foo{abc="123", cde="1"} -4  10
+foo{abc="123", cde="1"} -2  20
+`, `foo:1m_by_cde_rate_avg{cde="1"} 0
+foo:1m_by_cde_rate_sum{cde="1"} 0
+`, "11")
+
+	return
+
 	// Empty config
 	f(``, ``, ``, "")
 	f(``, `foo{bar="baz"} 1`, ``, "0")