Streaming PromQL engine: binary arithmetic operations with one-to-one matching

CHANGELOG.md

@@ -14,7 +14,7 @@
 * [FEATURE] Continuous-test: now runable as a module with `mimir -target=continuous-test`. #7747
 * [FEATURE] Store-gateway: Allow specific tenants to be enabled or disabled via `-store-gateway.enabled-tenants` or `-store-gateway.disabled-tenants` CLI flags or their corresponding YAML settings. #7653
 * [FEATURE] New `-<prefix>.s3.bucket-lookup-type` flag configures lookup style type, used to access bucket in s3 compatible providers. #7684
-* [FEATURE] Querier: add experimental streaming PromQL engine, enabled with `-querier.promql-engine=streaming`. #7693 #7898 #7899 #8023 #8058
+* [FEATURE] Querier: add experimental streaming PromQL engine, enabled with `-querier.promql-engine=streaming`. #7693 #7898 #7899 #8023 #8058 #8096
 * [FEATURE] New `/ingester/unregister-on-shutdown` HTTP endpoint allows dynamic access to ingesters' `-ingester.ring.unregister-on-shutdown` configuration. #7739
 * [FEATURE] Server: added experimental [PROXY protocol support](https://www.haproxy.org/download/2.3/doc/proxy-protocol.txt). The PROXY protocol support can be enabled via `-server.proxy-protocol-enabled=true`. When enabled, the support is added both to HTTP and gRPC listening ports. #7698
 * [FEATURE] mimirtool: Add `runtime-config verify` sub-command, for verifying Mimir runtime config files. #8123

pkg/querier/querier.go

@@ -30,6 +30,7 @@ import (
 	"github.com/grafana/mimir/pkg/storage/chunk"
 	"github.com/grafana/mimir/pkg/storage/lazyquery"
 	"github.com/grafana/mimir/pkg/streamingpromql"
+	"github.com/grafana/mimir/pkg/streamingpromql/compat"
 	"github.com/grafana/mimir/pkg/util"
 	"github.com/grafana/mimir/pkg/util/activitytracker"
 	"github.com/grafana/mimir/pkg/util/limiter"
@@ -170,7 +171,7 @@ func New(cfg Config, limits *validation.Overrides, distributor Distributor, stor
 
 		if cfg.EnablePromQLEngineFallback {
 			prometheusEngine := promql.NewEngine(opts)
-			eng = streamingpromql.NewEngineWithFallback(streamingEngine, prometheusEngine, reg, logger)
+			eng = compat.NewEngineWithFallback(streamingEngine, prometheusEngine, reg, logger)
 		} else {
 			eng = streamingEngine
 		}

pkg/streamingpromql/benchmarks/benchmarks.go

@@ -107,13 +107,20 @@ func TestCases(metricSizes []int) []BenchCase {
 		//	Expr: "-a_X",
 		//},
 		//// Binary operators.
-		//{
-		//	Expr: "a_X - b_X",
-		//},
-		//{
-		//	Expr:  "a_X - b_X",
-		//	Steps: 10000,
-		//},
+		{
+			Expr: "a_X - b_X",
+		},
+		{
+			Expr:  "a_X - b_X",
+			Steps: 10000,
+		},
+		// Test the case where one side of a binary operation has many more series than the other.
+		{
+			Expr: `a_100{l=~"[13579]."} - b_100`,
+		},
+		{
+			Expr: `a_2000{l=~"1..."} - b_2000`,
+		},
 		//{
 		//	Expr: "a_X and b_X{l=~'.*[0-4]$'}",
 		//},
@@ -163,9 +170,12 @@ func TestCases(metricSizes []int) []BenchCase {
 		//	Expr: "topk(5, a_X)",
 		//},
 		//// Combinations.
-		//{
-		//	Expr: "rate(a_X[1m]) + rate(b_X[1m])",
-		//},
+		{
+			Expr: "rate(a_X[1m]) + rate(b_X[1m])",
+		},
+		{
+			Expr: "sum(a_X + b_X)",
+		},
 		{
 			Expr: "sum by (le)(rate(h_X[1m]))",
 		},

pkg/streamingpromql/errors.go → pkg/streamingpromql/compat/errors.go

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-only
 
-package streamingpromql
+package compat
 
 import (
 	"errors"

pkg/streamingpromql/fallback_engine.go → pkg/streamingpromql/compat/fallback_engine.go

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-only
 
-package streamingpromql
+package compat
 
 import (
 	"context"

pkg/streamingpromql/fallback_engine_test.go → pkg/streamingpromql/compat/fallback_engine_test.go

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-only
 
-package streamingpromql
+package compat
 
 import (
 	"context"

pkg/streamingpromql/engine_test.go

@@ -15,6 +15,8 @@ import (
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/promql/promqltest"
 	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/mimir/pkg/streamingpromql/compat"
 )
 
 func TestUnsupportedPromQLFeatures(t *testing.T) {
@@ -26,29 +28,34 @@ func TestUnsupportedPromQLFeatures(t *testing.T) {
 	// The goal of this is not to list every conceivable expression that is unsupported, but to cover all the
 	// different cases and make sure we produce a reasonable error message when these cases are encountered.
 	unsupportedExpressions := map[string]string{
-		"a + b":                       "PromQL expression type *parser.BinaryExpr",
-		"1 + 2":                       "scalar value as top-level expression",
-		"metric{} + other_metric{}":   "PromQL expression type *parser.BinaryExpr",
-		"1":                           "scalar value as top-level expression",
-		"metric{} offset 2h":          "instant vector selector with 'offset'",
-		"avg(metric{})":               "'avg' aggregation",
-		"sum without(l) (metric{})":   "grouping with 'without'",
-		"rate(metric{}[5m:1m])":       "PromQL expression type *parser.SubqueryExpr",
-		"avg_over_time(metric{}[5m])": "'avg_over_time' function",
-		"-sum(metric{})":              "PromQL expression type *parser.UnaryExpr",
+		"1 + 2":                      "scalar value as top-level expression",
+		"1 + metric{}":               "binary expression with scalars",
+		"metric{} + 1":               "binary expression with scalars",
+		"metric{} < other_metric{}":  "binary expression with '<'",
+		"metric{} or other_metric{}": "binary expression with many-to-many matching",
+		"metric{} + on() group_left() other_metric{}":  "binary expression with many-to-one matching",
+		"metric{} + on() group_right() other_metric{}": "binary expression with one-to-many matching",
+		"1":                            "scalar value as top-level expression",
+		"metric{} offset 2h":           "instant vector selector with 'offset'",
+		"avg(metric{})":                "'avg' aggregation",
+		"sum without(l) (metric{})":    "grouping with 'without'",
+		"rate(metric{}[5m] offset 2h)": "range vector selector with 'offset'",
+		"rate(metric{}[5m:1m])":        "PromQL expression type *parser.SubqueryExpr",
+		"avg_over_time(metric{}[5m])":  "'avg_over_time' function",
+		"-sum(metric{})":               "PromQL expression type *parser.UnaryExpr",
 	}
 
 	for expression, expectedError := range unsupportedExpressions {
 		t.Run(expression, func(t *testing.T) {
 			qry, err := engine.NewRangeQuery(ctx, nil, nil, expression, time.Now().Add(-time.Hour), time.Now(), time.Minute)
 			require.Error(t, err)
-			require.ErrorIs(t, err, NotSupportedError{})
+			require.ErrorIs(t, err, compat.NotSupportedError{})
 			require.EqualError(t, err, "not supported by streaming engine: "+expectedError)
 			require.Nil(t, qry)
 
 			qry, err = engine.NewInstantQuery(ctx, nil, nil, expression, time.Now())
 			require.Error(t, err)
-			require.ErrorIs(t, err, NotSupportedError{})
+			require.ErrorIs(t, err, compat.NotSupportedError{})
 			require.EqualError(t, err, "not supported by streaming engine: "+expectedError)
 			require.Nil(t, qry)
 		})
@@ -65,7 +72,7 @@ func TestUnsupportedPromQLFeatures(t *testing.T) {
 		t.Run(expression, func(t *testing.T) {
 			qry, err := engine.NewInstantQuery(ctx, nil, nil, expression, time.Now())
 			require.Error(t, err)
-			require.ErrorIs(t, err, NotSupportedError{})
+			require.ErrorIs(t, err, compat.NotSupportedError{})
 			require.EqualError(t, err, "not supported by streaming engine: "+expectedError)
 			require.Nil(t, qry)
 		})

pkg/streamingpromql/operator/aggregation_test.go

@@ -97,19 +97,3 @@ func labelsToSeriesMetadata(lbls []labels.Labels) []SeriesMetadata {
 
 	return m
 }
-
-type testOperator struct {
-	series []labels.Labels
-}
-
-func (t *testOperator) SeriesMetadata(_ context.Context) ([]SeriesMetadata, error) {
-	return labelsToSeriesMetadata(t.series), nil
-}
-
-func (t *testOperator) NextSeries(_ context.Context) (InstantVectorSeriesData, error) {
-	panic("NextSeries() not supported")
-}
-
-func (t *testOperator) Close() {
-	panic("Close() not supported")
-}