app/vmselect/promql: add range_mad(q) and range_trim_outliers(k, q) functions

These functions may help trimming outliers during query time
for the use case described at https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3759

app/vmselect/promql/exec_test.go

@@ -6503,6 +6503,17 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r1, r2}
 		f(q, resultExpected)
 	})
+	t.Run(`range_trim_outliers()`, func(t *testing.T) {
+		t.Parallel()
+		q := `range_trim_outliers(0.5, time())`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{nan, nan, 1400, 1600, nan, nan},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`range_trim_spikes()`, func(t *testing.T) {
 		t.Parallel()
 		q := `range_trim_spikes(0.2, time())`
@@ -7059,6 +7070,17 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
+	t.Run(`range_mad(time())`, func(t *testing.T) {
+		t.Parallel()
+		q := `range_mad(time())`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{300, 300, 300, 300, 300, 300},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`range_max(time())`, func(t *testing.T) {
 		t.Parallel()
 		q := `range_max(time())`
@@ -8317,8 +8339,10 @@ func TestExecError(t *testing.T) {
 	f(`end(1)`)
 	f(`step(1)`)
 	f(`running_sum(1, 2)`)
+	f(`range_mad()`)
 	f(`range_sum(1, 2)`)
 	f(`range_trim_spikes()`)
+	f(`range_trim_outliers()`)
 	f(`range_first(1,  2)`)
 	f(`range_last(1, 2)`)
 	f(`range_linear_regression(1, 2)`)

app/vmselect/promql/rollup.go

@@ -1219,18 +1219,21 @@ func rollupMAD(rfa *rollupFuncArg) float64 {
 	// There is no need in handling NaNs here, since they must be cleaned up
 	// before calling rollup funcs.
 
+	return mad(rfa.values)
+}
+
+func mad(values []float64) float64 {
 	// See https://en.wikipedia.org/wiki/Median_absolute_deviation
-	values := rfa.values
 	median := quantile(0.5, values)
 	a := getFloat64s()
 	ds := a.A[:0]
 	for _, v := range values {
 		ds = append(ds, math.Abs(v-median))
 	}
-	mad := quantile(0.5, ds)
+	v := quantile(0.5, ds)
 	a.A = ds
 	putFloat64s(a)
-	return mad
+	return v
 }
 
 func rollupHistogram(rfa *rollupFuncArg) float64 {

app/vmselect/promql/transform.go

@@ -89,6 +89,7 @@ var transformFuncs = map[string]transformFunc{
 	"range_first":                transformRangeFirst,
 	"range_last":                 transformRangeLast,
 	"range_linear_regression":    transformRangeLinearRegression,
+	"range_mad":                  transformRangeMAD,
 	"range_max":                  newTransformFuncRange(runningMax),
 	"range_min":                  newTransformFuncRange(runningMin),
 	"range_normalize":            transformRangeNormalize,
@@ -96,6 +97,7 @@ var transformFuncs = map[string]transformFunc{
 	"range_stddev":               transformRangeStddev,
 	"range_stdvar":               transformRangeStdvar,
 	"range_sum":                  newTransformFuncRange(runningSum),
+	"range_trim_outliers":        transformRangeTrimOutliers,
 	"range_trim_spikes":          transformRangeTrimSpikes,
 	"remove_resets":              transformRemoveResets,
 	"round":                      transformRound,
@@ -1275,6 +1277,34 @@ func transformRangeNormalize(tfa *transformFuncArg) ([]*timeseries, error) {
 	return rvs, nil
 }
 
+func transformRangeTrimOutliers(tfa *transformFuncArg) ([]*timeseries, error) {
+	args := tfa.args
+	if err := expectTransformArgsNum(args, 2); err != nil {
+		return nil, err
+	}
+	ks, err := getScalar(args[0], 0)
+	if err != nil {
+		return nil, err
+	}
+	k := float64(0)
+	if len(ks) > 0 {
+		k = ks[0]
+	}
+	// Trim samples v satisfying the `abs(v - range_median(q)) > k*range_mad(q)`
+	rvs := args[1]
+	for _, ts := range rvs {
+		values := ts.Values
+		dMax := k * mad(values)
+		qMedian := quantile(0.5, values)
+		for i, v := range values {
+			if math.Abs(v-qMedian) > dMax {
+				values[i] = nan
+			}
+		}
+	}
+	return rvs, nil
+}
+
 func transformRangeTrimSpikes(tfa *transformFuncArg) ([]*timeseries, error) {
 	args := tfa.args
 	if err := expectTransformArgsNum(args, 2); err != nil {
@@ -1344,6 +1374,22 @@ func transformRangeLinearRegression(tfa *transformFuncArg) ([]*timeseries, error
 	return rvs, nil
 }
 
+func transformRangeMAD(tfa *transformFuncArg) ([]*timeseries, error) {
+	args := tfa.args
+	if err := expectTransformArgsNum(args, 1); err != nil {
+		return nil, err
+	}
+	rvs := args[0]
+	for _, ts := range rvs {
+		values := ts.Values
+		v := mad(values)
+		for i := range values {
+			values[i] = v
+		}
+	}
+	return rvs, nil
+}
+
 func transformRangeStddev(tfa *transformFuncArg) ([]*timeseries, error) {
 	args := tfa.args
 	if err := expectTransformArgsNum(args, 1); err != nil {

docs/CHANGELOG.md

@@ -23,6 +23,8 @@ The following tip changes can be tested by building VictoriaMetrics components f
 * FEATURE: [vmauth](https://docs.victoriametrics.com/vmauth.html): choose the backend with the minimum number of concurrently executed requests [among the configured backends](https://docs.victoriametrics.com/vmauth.html#load-balancing) in a round-robin manner for serving the incoming requests. This allows spreading the load among backends more evenly, while improving the response time.
 * FEATURE: [vmalert enterprise](https://docs.victoriametrics.com/vmalert.html): add ability to read alerting and recording rules from S3, GCS or S3-compatible object storage. See [these docs](https://docs.victoriametrics.com/vmalert.html#reading-rules-from-object-storage).
 * FEATURE: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): add `mad_over_time(m[d])` function for calculating the [median absolute deviation](https://en.wikipedia.org/wiki/Median_absolute_deviation) over raw samples on the lookbehind window `d`. See [this feature request](https://github.com/prometheus/prometheus/issues/5514).
+* FEATURE: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): add `range_mad(q)` function for calculating the [median absolute deviation](https://en.wikipedia.org/wiki/Median_absolute_deviation) over points per each time series returned by `q`.
+* FEATURE: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): add `range_trim_outliers(k, q)` function for dropping outliers farther than `k*range_mad(q)` from the `range_median(q)`. This should removing outliers at query time at [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3759).
 * FEATURE: [vmui](https://docs.victoriametrics.com/#vmui): show `median` instead of `avg` in graph tooltip and line legend, since `median` is more tolerant against spikes. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3706).
 
 * BUGFIX: prevent from possible data ingestion slowdown and query performance slowdown during [background merges of big parts](https://docs.victoriametrics.com/#storage) on systems with small number of CPU cores (1 or 2 CPU cores). The issue has been introduced in [v1.85.0](https://docs.victoriametrics.com/CHANGELOG.html#v1850) when implementing [this feature](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3337). See also [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3790).

docs/MetricsQL.md

@@ -508,7 +508,7 @@ See also [duration_over_time](#duration_over_time) and [lag](#lag).
 `mad_over_time(series_selector[d])` is a [rollup function](#rollup-functions), which calculates [median absolute deviation](https://en.wikipedia.org/wiki/Median_absolute_deviation)
 over raw samples on the given lookbehind window `d` per each time series returned from the given [series_selector](https://docs.victoriametrics.com/keyConcepts.html#filtering).
 
-See also [mad](#mad).
+See also [mad](#mad) and [range_mad](#range_mad).
 
 #### max_over_time
 
@@ -1221,6 +1221,13 @@ See also [rand](#rand) and [rand_exponential](#rand_exponential).
 `range_linear_regression(q)` is a [transform function](#transform-functions), which calculates [simple linear regression](https://en.wikipedia.org/wiki/Simple_linear_regression)
 over the selected time range per each time series returned by `q`. This function is useful for capacity planning and predictions.
 
+#### range_mad
+
+`range_mad(q)` is a [transform function](#transform-functions), which calculates the [median absolute deviation](https://en.wikipedia.org/wiki/Median_absolute_deviation)
+across points per each time series returned by `q`.
+
+See also [mad](#mad) and [mad_over_time](#mad_over_time).
+
 #### range_max
 
 `range_max(q)` is a [transform function](#transform-functions), which calculates the max value across points per each time series returned by `q`.
@@ -1257,11 +1264,22 @@ per each time series returned by `q` on the selected time range.
 
 `range_sum(q)` is a [transform function](#transform-functions), which calculates the sum of points per each time series returned by `q`.
 
+#### range_trim_outliers
+
+`range_trim_outliers(k, q)` is a [transform function](#transform-functions), which drops points located farther than `k*range_mad(q)`
+from the `range_median(q)`. E.g., it is equivalent to the following query: `q ifnot (abs(q - range_median(q)) > k*range_mad(q))`.
+
+The `phi` must be in the range `[0..1]`, where `0` means `0%` and `1` means `100%`.
+
+See also [range_trim_outliers](#range_trim_outliers).
+
 #### range_trim_spikes
 
 `range_trim_spikes(phi, q)` is a [transform function](#transform-functions), which drops `phi` percent of biggest spikes from time series returned by `q`.
 The `phi` must be in the range `[0..1]`, where `0` means `0%` and `1` means `100%`.
 
+See also [range_trim_outliers](#range_trim_outliers).
+
 #### remove_resets
 
 `remove_resets(q)` is a [transform function](#transform-functions), which removes counter resets from time series returned by `q`.
@@ -1731,7 +1749,7 @@ See also [limit_offset](#limit_offset).
 `mad(q) by (group_labels)` is [aggregate function](#aggregate-functions), which returns the [Median absolute deviation](https://en.wikipedia.org/wiki/Median_absolute_deviation)
 per each `group_labels` for all the time series returned by `q`. The aggregate is calculated individually per each group of points with the same timestamp.
 
-See also [outliers_mad](#outliers_mad) and [stddev](#stddev).
+See also [range_mad](#range_mad), [mad_over_time](#mad_over_time), [outliers_mad](#outliers_mad) and [stddev](#stddev).
 
 #### max
 

go.mod

@@ -12,7 +12,7 @@ require (
 	// like https://github.com/valyala/fasthttp/commit/996610f021ff45fdc98c2ce7884d5fa4e7f9199b
 	github.com/VictoriaMetrics/fasthttp v1.1.0
 	github.com/VictoriaMetrics/metrics v1.23.1
-	github.com/VictoriaMetrics/metricsql v0.53.0
+	github.com/VictoriaMetrics/metricsql v0.54.0
 	github.com/aws/aws-sdk-go-v2 v1.17.4
 	github.com/aws/aws-sdk-go-v2/config v1.18.12
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.11.51

go.sum

@@ -69,8 +69,8 @@ github.com/VictoriaMetrics/fasthttp v1.1.0/go.mod h1:/7DMcogqd+aaD3G3Hg5kFgoFwlR
 github.com/VictoriaMetrics/metrics v1.18.1/go.mod h1:ArjwVz7WpgpegX/JpB0zpNF2h2232kErkEnzH1sxMmA=
 github.com/VictoriaMetrics/metrics v1.23.1 h1:/j8DzeJBxSpL2qSIdqnRFLvQQhbJyJbbEi22yMm7oL0=
 github.com/VictoriaMetrics/metrics v1.23.1/go.mod h1:rAr/llLpEnAdTehiNlUxKgnjcOuROSzpw0GvjpEbvFc=
-github.com/VictoriaMetrics/metricsql v0.53.0 h1:R//oEGo+G0DtmNxF111ClM2e2pjC4sG14geyZzXfbjU=
+github.com/VictoriaMetrics/metricsql v0.54.0 h1:dKAIJtWcSPKnMNhRY5MYpqC77ZyHtA1xuDRr1pJuN5Q=
-github.com/VictoriaMetrics/metricsql v0.53.0/go.mod h1:6pP1ZeLVJHqJrHlF6Ij3gmpQIznSsgktEcZgsAWYel0=
+github.com/VictoriaMetrics/metricsql v0.54.0/go.mod h1:6pP1ZeLVJHqJrHlF6Ij3gmpQIznSsgktEcZgsAWYel0=
 github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=
 github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1ow=
 github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4=

vendor/github.com/VictoriaMetrics/metricsql/optimizer.go

@@ -392,7 +392,8 @@ func getTransformArgIdxForOptimization(funcName string, args []Expr) int {
 		return -1
 	case "limit_offset":
 		return 2
-	case "buckets_limit", "histogram_quantile", "histogram_share", "range_quantile", "range_trim_spikes":
+	case "buckets_limit", "histogram_quantile", "histogram_share", "range_quantile",
+		"range_trim_outliers", "range_trim_spikes":
 		return 1
 	case "histogram_quantiles":
 		return len(args) - 1

vendor/github.com/VictoriaMetrics/metricsql/transform.go

@@ -74,6 +74,7 @@ var transformFuncs = map[string]bool{
 	"range_first":                true,
 	"range_last":                 true,
 	"range_linear_regression":    true,
+	"range_mad":                  true,
 	"range_max":                  true,
 	"range_min":                  true,
 	"range_normalize":            true,
@@ -81,6 +82,7 @@ var transformFuncs = map[string]bool{
 	"range_stddev":               true,
 	"range_stdvar":               true,
 	"range_sum":                  true,
+	"range_trim_outliers":        true,
 	"range_trim_spikes":          true,
 	"remove_resets":              true,
 	"round":                      true,

vendor/modules.txt

@@ -71,7 +71,7 @@ github.com/VictoriaMetrics/fasthttp/stackless
 # github.com/VictoriaMetrics/metrics v1.23.1
 ## explicit; go 1.15
 github.com/VictoriaMetrics/metrics
-# github.com/VictoriaMetrics/metricsql v0.53.0
+# github.com/VictoriaMetrics/metricsql v0.54.0
 ## explicit; go 1.13
 github.com/VictoriaMetrics/metricsql
 github.com/VictoriaMetrics/metricsql/binaryop