diff --git a/CHANGELOG.md b/CHANGELOG.md index 16148eeae63..a81d8a57682 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -92,6 +92,7 @@ * [BUGFIX] Parquet Converter: Fix `auto_forget_delay` having no effect. The ring lifecycler was created without the auto-forget delegate, so unhealthy instances were never automatically removed from the ring. #7752 * [BUGFIX] Alertmanager: Reject the global `mattermost_webhook_url_file` setting in per-tenant configs, consistent with every other global `*_file` setting. #7768 * [BUGFIX] Alertmanager: Tighten per-tenant config validation to reject additional file-based settings. #7767 +* [BUGFIX] Query Frontend: Disable vertical sharding for queries using `vector()`. `vector()` is not backed by any series selector, so every shard produced its sample and a query such as ` or vector(0)` could return the `vector()` fallback instead of the left-hand side. #7806 ## 1.21.1 2026-06-04 diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go index 1106a458e38..44f1855be1f 100644 --- a/pkg/cortex/modules.go +++ b/pkg/cortex/modules.go @@ -548,6 +548,10 @@ func (t *Cortex) initFlusher() (serv services.Service, err error) { func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err error) { var queryAnalyzer querysharding.Analyzer queryAnalyzer = querysharding.NewQueryAnalyzer() + // `vector()` synthesises a series that is not backed by any selector, so every + // vertical shard produces it. Sharding such a query can return the `vector()` + // fallback instead of the real data, so disable sharding for those queries. + queryAnalyzer = cortexquerysharding.NewDisableVectorFunctionAnalyzer(queryAnalyzer) if t.Cfg.Querier.EnableParquetQueryable { // Disable vertical sharding for binary expression with ignore for parquet queryable. queryAnalyzer = cortexquerysharding.NewDisableBinaryExpressionAnalyzer(queryAnalyzer) diff --git a/pkg/querysharding/util.go b/pkg/querysharding/util.go index 05a8552cc32..da2b5816eb3 100644 --- a/pkg/querysharding/util.go +++ b/pkg/querysharding/util.go @@ -82,6 +82,61 @@ func ExtractShardingMatchers(matchers []*labels.Matcher) ([]*labels.Matcher, *st return r, shardInfo.Matcher(&Buffers), nil } +type disableVectorFunctionAnalyzer struct { + analyzer querysharding.Analyzer +} + +// NewDisableVectorFunctionAnalyzer is a wrapper around the analyzer that marks queries +// using the `vector()` function as not shardable. +// +// Vertical sharding is implemented by appending a shard matcher to every vector +// selector, so each shard only evaluates the query over the series it owns. That is +// only correct as long as every output series of the query is derived from a selector, +// because then the shard that owns the series is the only shard that can produce it. +// +// `vector(s)` breaks that invariant: it synthesises a series with an empty labelset out +// of a scalar, with no selector behind it, so *every* shard produces it. When that +// sample can reach the query result - most notably as the right hand side of `or`, the +// idiomatic "default value" pattern - the shards where the left hand side is empty emit +// the `vector()` sample, and merging the shards can pick it over the real sample +// produced by the shard that does own the data. The result is that a sharded query +// returns the `vector()` fallback where an unsharded query correctly returns the left +// hand side. +// +// This mirrors the way the upstream Thanos analyzer already refuses to shard `absent`, +// `absent_over_time` and `scalar`, whose results likewise depend on data a single shard +// cannot see. See https://github.com/cortexproject/cortex/issues/7804. +func NewDisableVectorFunctionAnalyzer(analyzer querysharding.Analyzer) *disableVectorFunctionAnalyzer { + return &disableVectorFunctionAnalyzer{analyzer: analyzer} +} + +func (d *disableVectorFunctionAnalyzer) Analyze(query string) (querysharding.QueryAnalysis, error) { + analysis, err := d.analyzer.Analyze(query) + if err != nil || !analysis.IsShardable() { + return analysis, err + } + + expr, err := cortexparser.ParseExpr(query) + if err != nil { + // The wrapped analyzer already parsed the query successfully, so this should not + // happen. Be conservative and keep the wrapped analyzer's answer. + return analysis, nil + } + isShardable := true + parser.Inspect(expr, func(node parser.Node, nodes []parser.Node) error { + if n, ok := node.(*parser.Call); ok && n.Func != nil && n.Func.Name == "vector" { + isShardable = false + return stop + } + return nil + }) + if !isShardable { + // Mark as not shardable. + return querysharding.QueryAnalysis{}, nil + } + return analysis, nil +} + type disableBinaryExpressionAnalyzer struct { analyzer querysharding.Analyzer } diff --git a/pkg/querysharding/util_test.go b/pkg/querysharding/util_test.go index cba23190723..54235484c05 100644 --- a/pkg/querysharding/util_test.go +++ b/pkg/querysharding/util_test.go @@ -8,6 +8,75 @@ import ( "github.com/thanos-io/thanos/pkg/querysharding" ) +func TestDisableVectorFunctionAnalyzer_Analyze(t *testing.T) { + tests := []struct { + name string + query string + expectShardable bool + expectError bool + description string + }{ + { + name: "aggregation without vector()", + query: `sum(rate(http_requests_total[5m])) by (job)`, + expectShardable: true, + expectError: false, + description: "Queries not using vector() are unaffected", + }, + { + name: "or vector() fallback", + query: `sum(rate(http_requests_total[5m])) by (job) or vector(0)`, + expectShardable: false, + expectError: false, + description: "vector() is produced by every shard, so the query is not shardable", + }, + { + name: "vector() nested in an or operand", + query: `max without (__name__, job, series) (test_series_a) or (-test_series_b or vector(-0.35))`, + expectShardable: false, + expectError: false, + description: "vector() anywhere in the query makes it not shardable", + }, + { + name: "vector() as an argument of another function", + query: `sum(clamp_min(vector(1), 0)) by (job)`, + expectShardable: false, + expectError: false, + description: "vector() nested inside a call still makes the query not shardable", + }, + { + name: "series named vector", + query: `sum(vector_series) by (job)`, + expectShardable: true, + expectError: false, + description: "A selector whose name merely contains vector must stay shardable", + }, + { + name: "invalid query", + query: "invalid{query", + expectShardable: false, + expectError: true, + description: "Invalid queries should return error", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + analyzer := NewDisableVectorFunctionAnalyzer(querysharding.NewQueryAnalyzer()) + + result, err := analyzer.Analyze(tt.query) + + if tt.expectError { + require.Error(t, err, tt.description) + return + } + + require.NoError(t, err, tt.description) + assert.Equal(t, tt.expectShardable, result.IsShardable(), tt.description) + }) + } +} + func TestDisableBinaryExpressionAnalyzer_Analyze(t *testing.T) { tests := []struct { name string