Add select merger integration: cachedQueryFunc, executePrefetch, config wiring

- Add cachedQueryFunc that wraps QueryFunc to serve from pre-fetched cache
- Add executePrefetch that evaluates merged selectors and populates cache
- Add buildSelectorExpr helper to construct PromQL selector strings
- Add filterVector to filter cached Vector by additional matchers
- Add mergedSelectIterationFunc in manager.go (Plan → Pre-fetch → Evaluate)
- Add prefetchExpr field to mergedSelect and track broadest rule expression
- Add SelectMergerEnabled/SelectMergerMinRules config fields to ruler.go
- Wire mergedSelectIterationFunc when config is enabled
- Add vendored SetQueryFunc/Opts methods on prometheus rules.Group
- Update prefetchCache to store promql.Vector instead of []storage.Series
- Update tests to match new cache.get signature (returns Vector)

Author: alanprot

pkg/ruler/manager.go

@@ -134,6 +134,9 @@ func NewDefaultMultiTenantManager(cfg Config, limits RulesLimits, managerFactory
 		logger:                 logger,
 		ruleGroupIterationFunc: defaultRuleGroupIterationFunc,
 	}
+	if cfg.SelectMergerEnabled {
+		m.ruleGroupIterationFunc = mergedSelectIterationFunc(cfg.SelectMergerMinRules)
+	}
 	if cfg.RulesBackupEnabled() {
 		m.rulesBackupManager = newRulesBackupManager(cfg, logger, reg)
 	}
@@ -292,6 +295,24 @@ func defaultRuleGroupIterationFunc(ctx context.Context, g *promRules.Group, eval
 	promRules.DefaultEvalIterationFunc(ctx, g, evalTimestamp)
 }
 
+// mergedSelectIterationFunc returns a GroupEvalIterationFunc that pre-fetches
+// merged selectors before evaluating the group, serving individual rule queries
+// from the cached results.
+func mergedSelectIterationFunc(minRules int) promRules.GroupEvalIterationFunc {
+	return func(ctx context.Context, g *promRules.Group, evalTimestamp time.Time) {
+		plan := planMergedSelects(g.Rules(), minRules)
+		if len(plan) > 0 {
+			cache := executePrefetch(ctx, plan, g.Opts().QueryFunc, evalTimestamp)
+			if cache != nil && len(cache.entries) > 0 {
+				orig := g.Opts().QueryFunc
+				g.SetQueryFunc(cachedQueryFunc(orig, cache))
+				defer g.SetQueryFunc(orig)
+			}
+		}
+		defaultRuleGroupIterationFunc(ctx, g, evalTimestamp)
+	}
+}
+
 // newManager creates a prometheus rule manager wrapped with a user id
 // configured storage, appendable, notifier, and instrumentation
 func (r *DefaultMultiTenantManager) newManager(ctx context.Context, userID string) (RulesManager, error) {

pkg/ruler/prefetch_queryable.go

@@ -1,16 +1,22 @@
 package ruler
 
 import (
+	"context"
+	"strings"
+	"time"
+
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/promql/parser"
+	"github.com/prometheus/prometheus/rules"
 	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/util/annotations"
 )
 
-// prefetchEntry holds pre-fetched series for a merged selector.
+// prefetchEntry holds pre-fetched results for a merged selector.
 type prefetchEntry struct {
 	matchers []*labels.Matcher
-	series   []storage.Series
+	vector   promql.Vector
 }
 
 // prefetchCache holds all pre-fetched data for a single group evaluation.
@@ -19,19 +25,81 @@ type prefetchCache struct {
 }
 
 // get checks if any pre-fetched entry is a superset of the requested matchers.
-func (c *prefetchCache) get(queryMatchers []*labels.Matcher) (storage.SeriesSet, bool) {
+// Returns the cached Vector filtered to match the query matchers.
+func (c *prefetchCache) get(queryMatchers []*labels.Matcher) (promql.Vector, bool) {
 	for _, e := range c.entries {
 		if isMatcherSetSuperset(e.matchers, queryMatchers) {
 			extra := extraMatchers(e.matchers, queryMatchers)
 			if len(extra) == 0 {
-				return newSliceSeriesSet(e.series), true
+				return e.vector, true
 			}
-			return newFilterSeriesSet(newSliceSeriesSet(e.series), extra), true
+			return filterVector(e.vector, extra), true
 		}
 	}
 	return nil, false
 }
 
+// filterVector returns only samples whose labels match all filters.
+func filterVector(vec promql.Vector, filters []*labels.Matcher) promql.Vector {
+	var result promql.Vector
+	for _, s := range vec {
+		if matchesAll(s.Metric, filters) {
+			result = append(result, s)
+		}
+	}
+	return result
+}
+
+// cachedQueryFunc wraps a QueryFunc to serve from pre-fetched cache.
+func cachedQueryFunc(inner rules.QueryFunc, cache *prefetchCache) rules.QueryFunc {
+	if cache == nil || len(cache.entries) == 0 {
+		return inner
+	}
+	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
+		selectors := extractSelectorsFromExpr(qs)
+		if len(selectors) == 1 {
+			if vec, ok := cache.get(selectors[0]); ok {
+				return vec, nil
+			}
+		}
+		return inner(ctx, qs, t)
+	}
+}
+
+// executePrefetch runs the merged selectors via QueryFunc and populates a cache.
+func executePrefetch(ctx context.Context, plan []mergedSelect, queryFunc rules.QueryFunc, ts time.Time) *prefetchCache {
+	cache := &prefetchCache{}
+	for _, ms := range plan {
+		expr := ms.prefetchExpr
+		if expr == "" {
+			expr = buildSelectorExpr(ms.mergedMatchers)
+		}
+		vec, err := queryFunc(ctx, expr, ts)
+		if err != nil {
+			continue
+		}
+		cache.entries = append(cache.entries, prefetchEntry{
+			matchers: ms.mergedMatchers,
+			vector:   vec,
+		})
+	}
+	return cache
+}
+
+// buildSelectorExpr constructs a PromQL selector string from matchers.
+func buildSelectorExpr(ms []*labels.Matcher) string {
+	var b strings.Builder
+	b.WriteByte('{')
+	for i, m := range ms {
+		if i > 0 {
+			b.WriteByte(',')
+		}
+		b.WriteString(m.String())
+	}
+	b.WriteByte('}')
+	return b.String()
+}
+
 // isMatcherSetSuperset returns true if superMatchers covers all series that subMatchers would match.
 func isMatcherSetSuperset(superMatchers, subMatchers []*labels.Matcher) bool {
 	for _, sup := range superMatchers {

pkg/ruler/prefetch_queryable_test.go

@@ -1,9 +1,12 @@
 package ruler
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
 	"github.com/stretchr/testify/assert"
@@ -67,9 +70,9 @@ func TestPrefetchCache_FindSuperset(t *testing.T) {
 					labels.MustNewMatcher(labels.MatchEqual, "__name__", "http"),
 					labels.MustNewMatcher(labels.MatchRegexp, "job", ".*"),
 				},
-				series: []storage.Series{
-					&mockSeries{lset: labels.FromStrings("__name__", "http", "job", "api")},
-					&mockSeries{lset: labels.FromStrings("__name__", "http", "job", "web")},
+				vector: promql.Vector{
+					{Metric: labels.FromStrings("__name__", "http", "job", "api"), T: 1000, F: 1.0},
+					{Metric: labels.FromStrings("__name__", "http", "job", "web"), T: 1000, F: 2.0},
 				},
 			},
 		},
@@ -81,16 +84,11 @@ func TestPrefetchCache_FindSuperset(t *testing.T) {
 		labels.MustNewMatcher(labels.MatchEqual, "job", "api"),
 	}
 
-	ss, ok := cache.get(queryMatchers)
+	vec, ok := cache.get(queryMatchers)
 	require.True(t, ok)
-
-	var got []labels.Labels
-	for ss.Next() {
-		got = append(got, ss.At().Labels())
-	}
-	require.NoError(t, ss.Err())
-	assert.Len(t, got, 1)
-	assert.Equal(t, "api", got[0].Get("job"))
+	assert.Len(t, vec, 1)
+	assert.Equal(t, "api", vec[0].Metric.Get("job"))
+	assert.Equal(t, 1.0, vec[0].F)
 }
 
 func TestIsMatcherSetSuperset(t *testing.T) {
@@ -144,3 +142,72 @@ func TestIsMatcherSetSuperset(t *testing.T) {
 		})
 	}
 }
+
+func TestCachedQueryFunc(t *testing.T) {
+	cache := &prefetchCache{
+		entries: []prefetchEntry{
+			{
+				matchers: []*labels.Matcher{
+					labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests"),
+					labels.MustNewMatcher(labels.MatchRegexp, "job", ".*"),
+				},
+				vector: promql.Vector{
+					{Metric: labels.FromStrings("__name__", "http_requests", "job", "api"), T: 1000, F: 10.0},
+					{Metric: labels.FromStrings("__name__", "http_requests", "job", "web"), T: 1000, F: 20.0},
+				},
+			},
+		},
+	}
+
+	innerCalled := false
+	inner := func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
+		innerCalled = true
+		return nil, nil
+	}
+
+	qf := cachedQueryFunc(inner, cache)
+
+	// Query that matches cache — should NOT call inner
+	vec, err := qf(context.Background(), `http_requests{job="api"}`, time.Unix(1, 0))
+	require.NoError(t, err)
+	assert.False(t, innerCalled)
+	assert.Len(t, vec, 1)
+	assert.Equal(t, 10.0, vec[0].F)
+
+	// Query that doesn't match cache — should call inner
+	_, err = qf(context.Background(), `other_metric{job="api"}`, time.Unix(1, 0))
+	require.NoError(t, err)
+	assert.True(t, innerCalled)
+}
+
+func TestExecutePrefetch(t *testing.T) {
+	queryFunc := func(ctx context.Context, qs string, ts time.Time) (promql.Vector, error) {
+		return promql.Vector{
+			{Metric: labels.FromStrings("__name__", "http", "job", "api"), T: ts.UnixMilli(), F: 5.0},
+			{Metric: labels.FromStrings("__name__", "http", "job", "web"), T: ts.UnixMilli(), F: 7.0},
+		}, nil
+	}
+
+	plan := []mergedSelect{
+		{
+			metricName: "http",
+			mergedMatchers: []*labels.Matcher{
+				labels.MustNewMatcher(labels.MatchEqual, "__name__", "http"),
+				labels.MustNewMatcher(labels.MatchRegexp, "job", ".*"),
+			},
+		},
+	}
+
+	cache := executePrefetch(context.Background(), plan, queryFunc, time.Unix(1, 0))
+	require.Len(t, cache.entries, 1)
+	assert.Len(t, cache.entries[0].vector, 2)
+}
+
+func TestBuildSelectorExpr(t *testing.T) {
+	ms := []*labels.Matcher{
+		labels.MustNewMatcher(labels.MatchEqual, "__name__", "http"),
+		labels.MustNewMatcher(labels.MatchRegexp, "job", ".*"),
+	}
+	got := buildSelectorExpr(ms)
+	assert.Equal(t, `{__name__="http",job=~".*"}`, got)
+}

pkg/ruler/ruler.go

@@ -181,6 +181,9 @@ type Config struct {
 
 	ThanosEngine engine.ThanosEngineConfig `yaml:"thanos_engine"`
 
+	SelectMergerEnabled  bool `yaml:"select_merger_enabled"`
+	SelectMergerMinRules int  `yaml:"select_merger_min_rules"`
+
 	// NameValidationScheme is the scheme for validating metric and label names (set from root config).
 	NameValidationScheme model.ValidationScheme `yaml:"-"`
 }
@@ -272,6 +275,10 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 
 	f.BoolVar(&cfg.EnableHAEvaluation, "ruler.enable-ha-evaluation", false, "Enable high availability")
 	f.DurationVar(&cfg.LivenessCheckTimeout, "ruler.liveness-check-timeout", 1*time.Second, "Timeout duration for non-primary rulers during liveness checks. If the check times out, the non-primary ruler will evaluate the rule group. Applicable when ruler.enable-ha-evaluation is true.")
+
+	f.BoolVar(&cfg.SelectMergerEnabled, "ruler.select-merger-enabled", false, "Enable merged select pre-fetching to reduce redundant queries for rules sharing the same metric.")
+	f.IntVar(&cfg.SelectMergerMinRules, "ruler.select-merger-min-rules", 3, "Minimum number of rules querying the same metric to trigger merged pre-fetching.")
+
 	cfg.RingCheckPeriod = 5 * time.Second
 }
 

pkg/ruler/select_merger.go

@@ -10,22 +10,25 @@ type mergedSelect struct {
 	metricName      string
 	mergedMatchers  []*labels.Matcher
 	originalEntries [][]*labels.Matcher // per-rule matchers
+	prefetchExpr    string              // expression to evaluate for pre-fetch
 }
 
 func planMergedSelects(rls []rules.Rule, minRules int) []mergedSelect {
 	// Group selectors by metric name.
 	type entry struct {
 		matchers []*labels.Matcher
+		exprStr  string
 	}
 	groups := map[string][]entry{}
 
 	for _, r := range rls {
+		exprStr := r.Query().String()
 		extractSelectors(r.Query(), func(vs *parser.VectorSelector) {
 			name := metricNameFromMatchers(vs.LabelMatchers)
 			if name == "" {
 				return
 			}
-			groups[name] = append(groups[name], entry{matchers: vs.LabelMatchers})
+			groups[name] = append(groups[name], entry{matchers: vs.LabelMatchers, exprStr: exprStr})
 		})
 	}
 
@@ -38,15 +41,47 @@ func planMergedSelects(rls []rules.Rule, minRules int) []mergedSelect {
 		for i, e := range entries {
 			originals[i] = e.matchers
 		}
+		merged := computeMergedMatchers(originals)
+
+		// Find the expression whose matchers equal the merged set (the broadest rule).
+		prefetchExpr := ""
+		for _, e := range entries {
+			if matchersEqual(e.matchers, merged) {
+				prefetchExpr = e.exprStr
+				break
+			}
+		}
+
 		result = append(result, mergedSelect{
 			metricName:      name,
-			mergedMatchers:  computeMergedMatchers(originals),
+			mergedMatchers:  merged,
 			originalEntries: originals,
+			prefetchExpr:    prefetchExpr,
 		})
 	}
 	return result
 }
 
+// matchersEqual returns true if two matcher slices contain the same matchers (order-independent).
+func matchersEqual(a, b []*labels.Matcher) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for _, am := range a {
+		found := false
+		for _, bm := range b {
+			if am.Name == bm.Name && am.Type == bm.Type && am.Value == bm.Value {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return false
+		}
+	}
+	return true
+}
+
 func extractSelectors(expr parser.Expr, fn func(*parser.VectorSelector)) {
 	parser.Inspect(expr, func(node parser.Node, _ []parser.Node) error {
 		if vs, ok := node.(*parser.VectorSelector); ok {