Query: bump promql-engine

Bumping PromQL engine, fixing fallback fallout.

Signed-off-by: Michael Hoffmann <[email protected]>

Author: MichaHoffmann

CHANGELOG.md

@@ -28,6 +28,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 - [#7890](https://github.com/thanos-io/thanos/pull/7890) Query,Ruler: *breaking :warning:* deprecated `--store.sd-file` and `--store.sd-interval` to be replaced with `--endpoint.sd-config` and `--endpoint-sd-config-reload-interval`; removed legacy flags to pass endpoints `--store`, `--metadata`, `--rule`, `--exemplar`.
 - [#7012](https://github.com/thanos-io/thanos/pull/7012) Query: Automatically adjust `max_source_resolution` based on promql query to avoid querying data from higher resolution resulting empty results.
+- [#8118](https://github.com/thanos-io/thanos/pull/8118) Query: Bumped promql-engine
 
 ### Removed
 

cmd/thanos/query.go

@@ -478,7 +478,7 @@ func runQuery(
 		})
 	}
 
-	engineFactory := apiv1.NewQueryEngineFactory(engineOpts, remoteEngineEndpoints)
+	engineFactory := apiv1.NewQueryFactory(engineOpts, remoteEngineEndpoints)
 
 	lookbackDeltaCreator := LookbackDeltaFactory(engineOpts.EngineOpts, dynamicLookbackDelta)
 

go.mod

@@ -69,7 +69,7 @@ require (
 	github.com/sony/gobreaker v0.5.0
 	github.com/stretchr/testify v1.10.0
 	github.com/thanos-io/objstore v0.0.0-20241111205755-d1dd89d41f97
-	github.com/thanos-io/promql-engine v0.0.0-20250211181629-815830ca3e2e
+	github.com/thanos-io/promql-engine v0.0.0-20250221084015-4230034ebb6c
 	github.com/uber/jaeger-client-go v2.30.0+incompatible
 	github.com/vimeo/galaxycache v0.0.0-20210323154928-b7e5d71c067a
 	github.com/weaveworks/common v0.0.0-20230728070032-dd9e68f319d5

go.sum

@@ -2258,8 +2258,8 @@ github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e h1:f1
 github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e/go.mod h1:jXcofnrSln/cLI6/dhlBxPQZEEQHVPCcFaH75M+nSzM=
 github.com/thanos-io/objstore v0.0.0-20241111205755-d1dd89d41f97 h1:VjG0mwhN1DkncwDHFvrpd12/2TLfgYNRmEQA48ikp+0=
 github.com/thanos-io/objstore v0.0.0-20241111205755-d1dd89d41f97/go.mod h1:vyzFrBXgP+fGNG2FopEGWOO/zrIuoy7zt3LpLeezRsw=
-github.com/thanos-io/promql-engine v0.0.0-20250211181629-815830ca3e2e h1:jDKxQzp4JIhpbn6NFHXc8TCsRy8GkfHMZ7XNicY1mx8=
-github.com/thanos-io/promql-engine v0.0.0-20250211181629-815830ca3e2e/go.mod h1:aHSV5hL94fNb7PklN9L0V10j+/RGIlzqbw7OLdNgZFs=
+github.com/thanos-io/promql-engine v0.0.0-20250221084015-4230034ebb6c h1:p/XSjq43vYSorePYsxethwoNpwrJp6HD4QQ7xOH2DGI=
+github.com/thanos-io/promql-engine v0.0.0-20250221084015-4230034ebb6c/go.mod h1:aHSV5hL94fNb7PklN9L0V10j+/RGIlzqbw7OLdNgZFs=
 github.com/themihai/gomemcache v0.0.0-20180902122335-24332e2d58ab h1:7ZR3hmisBWw77ZpO1/o86g+JV3VKlk3d48jopJxzTjU=
 github.com/themihai/gomemcache v0.0.0-20180902122335-24332e2d58ab/go.mod h1:eheTFp954zcWZXCU8d0AT76ftsQOTo4DTqkN/h3k1MY=
 github.com/tinylib/msgp v1.1.5 h1:2gXmtWueD2HefZHQe1QOy9HVzmFrLOVvsXwXBQ0ayy0=

pkg/api/query/engine.go

@@ -0,0 +1,141 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+// Copyright 2016 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// This package is a modified copy from
+// github.com/prometheus/prometheus/web/api/v1@2121b4628baa7d9d9406aa468712a6a332e77aff.
+
+package v1
+
+import (
+	"context"
+	"time"
+
+	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/thanos-io/promql-engine/api"
+	"github.com/thanos-io/promql-engine/engine"
+	"github.com/thanos-io/promql-engine/logicalplan"
+)
+
+type Engine interface {
+	MakeInstantQuery(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, qs string, ts time.Time) (promql.Query, error)
+	MakeRangeQuery(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, qs string, start, end time.Time, step time.Duration) (promql.Query, error)
+	MakeInstantQueryFromPlan(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, plan logicalplan.Node, ts time.Time) (promql.Query, error)
+	MakeRangeQueryFromPlan(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, root logicalplan.Node, start, end time.Time, step time.Duration) (promql.Query, error)
+}
+
+type prometheusEngineAdapter struct {
+	engine promql.QueryEngine
+}
+
+func (a *prometheusEngineAdapter) MakeInstantQuery(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, qs string, ts time.Time) (promql.Query, error) {
+	return a.engine.NewInstantQuery(ctx, q, opts, qs, ts)
+}
+
+func (a *prometheusEngineAdapter) MakeRangeQuery(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, qs string, start, end time.Time, step time.Duration) (promql.Query, error) {
+	return a.engine.NewRangeQuery(ctx, q, opts, qs, start, end, step)
+}
+
+func (a *prometheusEngineAdapter) MakeInstantQueryFromPlan(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, plan logicalplan.Node, ts time.Time) (promql.Query, error) {
+	return a.engine.NewInstantQuery(ctx, q, opts, plan.String(), ts)
+}
+
+func (a *prometheusEngineAdapter) MakeRangeQueryFromPlan(ctx context.Context, q storage.Queryable, opts *engine.QueryOpts, plan logicalplan.Node, start, end time.Time, step time.Duration) (promql.Query, error) {
+	return a.engine.NewRangeQuery(ctx, q, opts, plan.String(), start, end, step)
+}
+
+type QueryFactory struct {
+	prometheus Engine
+	thanos     Engine
+}
+
+func NewQueryFactory(
+	engineOpts engine.Opts,
+	remoteEngineEndpoints api.RemoteEndpoints,
+) *QueryFactory {
+
+	var thanosEngine Engine
+	if remoteEngineEndpoints == nil {
+		thanosEngine = engine.New(engineOpts)
+	} else {
+		thanosEngine = engine.NewDistributedEngine(engineOpts, remoteEngineEndpoints)
+	}
+	promEngine := promql.NewEngine(engineOpts.EngineOpts)
+	return &QueryFactory{
+		prometheus: &prometheusEngineAdapter{promEngine},
+		thanos:     thanosEngine,
+	}
+}
+
+// Always has query, sometimes already has a plan.
+type planOrQuery struct {
+	query string
+	plan  logicalplan.Node
+}
+
+func (f *QueryFactory) makeInstantQuery(
+	ctx context.Context,
+	e PromqlEngineType,
+	q storage.Queryable,
+	qry planOrQuery,
+	opts *engine.QueryOpts,
+	ts time.Time,
+) (res promql.Query, err error) {
+	if e == PromqlEngineThanos {
+		if qry.plan != nil {
+			res, err = f.thanos.MakeInstantQueryFromPlan(ctx, q, opts, qry.plan, ts)
+		} else {
+			res, err = f.thanos.MakeInstantQuery(ctx, q, opts, qry.query, ts)
+		}
+		if err != nil {
+			if engine.IsUnimplemented(err) {
+				goto fallback
+			}
+			return nil, err
+		}
+		return res, nil
+	}
+fallback:
+	return f.prometheus.MakeInstantQuery(ctx, q, opts, qry.query, ts)
+}
+
+func (f *QueryFactory) makeRangeQuery(
+	ctx context.Context,
+	e PromqlEngineType,
+	q storage.Queryable,
+	qry planOrQuery,
+	opts *engine.QueryOpts,
+	start time.Time,
+	end time.Time,
+	step time.Duration,
+) (res promql.Query, err error) {
+	if e == PromqlEngineThanos {
+		if qry.plan != nil {
+			res, err = f.thanos.MakeRangeQueryFromPlan(ctx, q, opts, qry.plan, start, end, step)
+		} else {
+			res, err = f.thanos.MakeRangeQuery(ctx, q, opts, qry.query, start, end, step)
+		}
+		if err != nil {
+			if engine.IsUnimplemented(err) {
+				goto fallback
+			}
+			return nil, err
+		}
+		return res, nil
+	}
+fallback:
+	return f.prometheus.MakeRangeQuery(ctx, q, opts, qry.query, start, end, step)
+}

pkg/api/query/grpc.go

@@ -26,7 +26,7 @@ type GRPCAPI struct {
 	now                         func() time.Time
 	replicaLabels               []string
 	queryableCreate             query.QueryableCreator
-	engineFactory               *QueryEngineFactory
+	queryFactory                *QueryFactory
 	defaultEngine               querypb.EngineType
 	lookbackDeltaCreate         func(int64) time.Duration
 	defaultMaxResolutionSeconds time.Duration
@@ -36,7 +36,7 @@ func NewGRPCAPI(
 	now func() time.Time,
 	replicaLabels []string,
 	creator query.QueryableCreator,
-	engineFactory *QueryEngineFactory,
+	queryFactory *QueryFactory,
 	defaultEngine querypb.EngineType,
 	lookbackDeltaCreate func(int64) time.Duration,
 	defaultMaxResolutionSeconds time.Duration,
@@ -45,7 +45,7 @@ func NewGRPCAPI(
 		now:                         now,
 		replicaLabels:               replicaLabels,
 		queryableCreate:             creator,
-		engineFactory:               engineFactory,
+		queryFactory:                queryFactory,
 		defaultEngine:               defaultEngine,
 		lookbackDeltaCreate:         lookbackDeltaCreate,
 		defaultMaxResolutionSeconds: defaultMaxResolutionSeconds,
@@ -97,7 +97,7 @@ func (g *GRPCAPI) Query(request *querypb.QueryRequest, server querypb.Query_Quer
 	var qry promql.Query
 	if err := tracing.DoInSpanWithErr(ctx, "instant_query_create", func(ctx context.Context) error {
 		var err error
-		qry, err = g.getQueryForEngine(ctx, request, queryable, maxResolution)
+		qry, err = g.getInstantQueryForEngine(ctx, request, queryable, maxResolution)
 		return err
 	}); err != nil {
 		return err
@@ -146,42 +146,6 @@ func (g *GRPCAPI) Query(request *querypb.QueryRequest, server querypb.Query_Quer
 	return nil
 }
 
-func (g *GRPCAPI) getQueryForEngine(ctx context.Context, request *querypb.QueryRequest, queryable storage.Queryable, maxResolution int64) (promql.Query, error) {
-	lookbackDelta := g.lookbackDeltaCreate(maxResolution * 1000)
-	if request.LookbackDeltaSeconds > 0 {
-		lookbackDelta = time.Duration(request.LookbackDeltaSeconds) * time.Second
-	}
-	engineParam := request.Engine
-	if engineParam == querypb.EngineType_default {
-		engineParam = g.defaultEngine
-	}
-
-	var ts time.Time
-	if request.TimeSeconds == 0 {
-		ts = g.now()
-	} else {
-		ts = time.Unix(request.TimeSeconds, 0)
-	}
-	opts := &engine.QueryOpts{
-		LookbackDeltaParam:     lookbackDelta,
-		EnablePartialResponses: request.EnablePartialResponse,
-	}
-	switch engineParam {
-	case querypb.EngineType_prometheus:
-		queryEngine := g.engineFactory.GetPrometheusEngine()
-		return queryEngine.MakeInstantQuery(ctx, queryable, opts, request.Query, ts)
-	case querypb.EngineType_thanos:
-		queryEngine := g.engineFactory.GetThanosEngine()
-		plan, err := logicalplan.Unmarshal(request.QueryPlan.GetJson())
-		if err != nil {
-			return queryEngine.MakeInstantQuery(ctx, queryable, opts, request.Query, ts)
-		}
-		return queryEngine.MakeInstantQueryFromPlan(ctx, queryable, opts, plan, ts)
-	default:
-		return nil, status.Error(codes.InvalidArgument, "invalid engine parameter")
-	}
-}
-
 func (g *GRPCAPI) QueryRange(request *querypb.QueryRangeRequest, srv querypb.Query_QueryRangeServer) error {
 	ctx := srv.Context()
 	if request.TimeoutSeconds != 0 {
@@ -295,14 +259,54 @@ func extractQueryStats(qry promql.Query) *querypb.QueryStats {
 	return stats
 }
 
+func (g *GRPCAPI) getInstantQueryForEngine(ctx context.Context, request *querypb.QueryRequest, queryable storage.Queryable, maxResolution int64) (promql.Query, error) {
+	lookbackDelta := g.lookbackDeltaCreate(maxResolution * 1000)
+	if request.LookbackDeltaSeconds > 0 {
+		lookbackDelta = time.Duration(request.LookbackDeltaSeconds) * time.Second
+	}
+	engineParam := request.Engine
+	if engineParam == querypb.EngineType_default {
+		engineParam = g.defaultEngine
+	}
+
+	var ts time.Time
+	if request.TimeSeconds == 0 {
+		ts = g.now()
+	} else {
+		ts = time.Unix(request.TimeSeconds, 0)
+	}
+	opts := &engine.QueryOpts{
+		LookbackDeltaParam:     lookbackDelta,
+		EnablePartialResponses: request.EnablePartialResponse,
+	}
+
+	var engineType PromqlEngineType
+	switch engineParam {
+	case querypb.EngineType_prometheus:
+		engineType = PromqlEnginePrometheus
+	case querypb.EngineType_thanos:
+		engineType = PromqlEngineThanos
+	default:
+		return nil, status.Error(codes.InvalidArgument, "invalid engine parameter")
+	}
+
+	var qry planOrQuery
+	if plan, err := logicalplan.Unmarshal(request.QueryPlan.GetJson()); err != nil {
+		qry = planOrQuery{plan: plan}
+	} else {
+		qry = planOrQuery{query: request.Query}
+	}
+	return g.queryFactory.makeInstantQuery(ctx, engineType, queryable, qry, opts, ts)
+}
+
 func (g *GRPCAPI) getRangeQueryForEngine(
 	ctx context.Context,
 	request *querypb.QueryRangeRequest,
 	queryable storage.Queryable,
 ) (promql.Query, error) {
-	startTime := time.Unix(request.StartTimeSeconds, 0)
-	endTime := time.Unix(request.EndTimeSeconds, 0)
-	interval := time.Duration(request.IntervalSeconds) * time.Second
+	start := time.Unix(request.StartTimeSeconds, 0)
+	end := time.Unix(request.EndTimeSeconds, 0)
+	step := time.Duration(request.IntervalSeconds) * time.Second
 
 	engineParam := request.Engine
 	if engineParam == querypb.EngineType_default {
@@ -322,18 +326,21 @@ func (g *GRPCAPI) getRangeQueryForEngine(
 		EnablePartialResponses: request.EnablePartialResponse,
 	}
 
+	var engineType PromqlEngineType
 	switch engineParam {
 	case querypb.EngineType_prometheus:
-		queryEngine := g.engineFactory.GetPrometheusEngine()
-		return queryEngine.MakeRangeQuery(ctx, queryable, opts, request.Query, startTime, endTime, interval)
+		engineType = PromqlEnginePrometheus
 	case querypb.EngineType_thanos:
-		thanosEngine := g.engineFactory.GetThanosEngine()
-		plan, err := logicalplan.Unmarshal(request.QueryPlan.GetJson())
-		if err != nil {
-			return thanosEngine.MakeRangeQuery(ctx, queryable, opts, request.Query, startTime, endTime, interval)
-		}
-		return thanosEngine.MakeRangeQueryFromPlan(ctx, queryable, opts, plan, startTime, endTime, interval)
+		engineType = PromqlEngineThanos
 	default:
 		return nil, status.Error(codes.InvalidArgument, "invalid engine parameter")
 	}
+
+	var qry planOrQuery
+	if plan, err := logicalplan.Unmarshal(request.QueryPlan.GetJson()); err != nil {
+		qry = planOrQuery{plan: plan}
+	} else {
+		qry = planOrQuery{query: request.Query}
+	}
+	return g.queryFactory.makeRangeQuery(ctx, engineType, queryable, qry, opts, start, end, step)
 }

pkg/api/query/grpc_test.go

@@ -32,10 +32,10 @@ func TestGRPCQueryAPIWithQueryPlan(t *testing.T) {
 	proxy := store.NewProxyStore(logger, reg, func() []store.Client { return nil }, component.Store, nil, 1*time.Minute, store.LazyRetrieval)
 	queryableCreator := query.NewQueryableCreator(logger, reg, proxy, 1, 1*time.Minute)
 	lookbackDeltaFunc := func(i int64) time.Duration { return 5 * time.Minute }
-	engineFactory := &QueryEngineFactory{
+	queryFactory := &QueryFactory{
 		thanos: &engineStub{},
 	}
-	api := NewGRPCAPI(time.Now, nil, queryableCreator, engineFactory, querypb.EngineType_thanos, lookbackDeltaFunc, 0)
+	api := NewGRPCAPI(time.Now, nil, queryableCreator, queryFactory, querypb.EngineType_thanos, lookbackDeltaFunc, 0)
 
 	expr, err := extpromql.ParseExpr("metric")
 	testutil.Ok(t, err)
@@ -97,10 +97,10 @@ func TestGRPCQueryAPIErrorHandling(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		engineFactory := &QueryEngineFactory{
+		queryFactory := &QueryFactory{
 			prometheus: test.engine,
 		}
-		api := NewGRPCAPI(time.Now, nil, queryableCreator, engineFactory, querypb.EngineType_prometheus, lookbackDeltaFunc, 0)
+		api := NewGRPCAPI(time.Now, nil, queryableCreator, queryFactory, querypb.EngineType_prometheus, lookbackDeltaFunc, 0)
 		t.Run("range_query", func(t *testing.T) {
 			rangeRequest := &querypb.QueryRangeRequest{
 				Query:            "metric",