disable vertical sharding for query containing without metric name

Signed-off-by: yeya24 <[email protected]>

Author: yeya24

integration/parquet_querier_test.go

@@ -131,16 +131,20 @@ func TestParquetFuzz(t *testing.T) {
 	// Wait until we convert the blocks
 	cortex_testutil.Poll(t, 30*time.Second, true, func() interface{} {
 		found := false
+		foundBucketIndex := false
 
 		err := bkt.Iter(context.Background(), "", func(name string) error {
 			fmt.Println(name)
 			if name == fmt.Sprintf("parquet-markers/%v-parquet-converter-mark.json", id.String()) {
 				found = true
 			}
+			if name == "bucket-index.json.gz" {
+				foundBucketIndex = true
+			}
 			return nil
 		}, objstore.WithRecursiveIter())
 		require.NoError(t, err)
-		return found
+		return found && foundBucketIndex
 	})
 
 	att, err := bkt.Attributes(context.Background(), "bucket-index.json.gz")
@@ -179,7 +183,7 @@ func TestParquetFuzz(t *testing.T) {
 	}
 	ps := promqlsmith.New(rnd, lbls, opts...)
 
-	runQueryFuzzTestCases(t, ps, c1, c2, end, start, end, scrapeInterval, 500, false)
+	runQueryFuzzTestCases(t, ps, c1, c2, end, start, end, scrapeInterval, 1000, false)
 
 	require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_parquet_queryable_blocks_queried_total"}, e2e.WithLabelMatchers(
 		labels.MustNewMatcher(labels.MatchEqual, "type", "parquet"))))

integration/query_fuzz_test.go

@@ -799,7 +799,7 @@ func TestVerticalShardingFuzz(t *testing.T) {
 	}
 	ps := promqlsmith.New(rnd, lbls, opts...)
 
-	runQueryFuzzTestCases(t, ps, c1, c2, now, start, end, scrapeInterval, 1000, false)
+	runQueryFuzzTestCases(t, ps, c1, c2, end, start, end, scrapeInterval, 1000, false)
 }
 
 func TestProtobufCodecFuzz(t *testing.T) {
@@ -1838,7 +1838,7 @@ func runQueryFuzzTestCases(t *testing.T, ps *promqlsmith.PromQLSmith, c1, c2 *e2
 				failures++
 			}
 		} else if !cmp.Equal(tc.res1, tc.res2, comparer) {
-			t.Logf("case %d results mismatch.\n%s: %s\nres1: %s\nres2: %s\n", i, qt, tc.query, tc.res1.String(), tc.res2.String())
+			t.Logf("case %d results mismatch.\n%s: %s\nres1 len: %d data: %s\nres2 len: %d data: %s\n", i, qt, tc.query, resultLength(tc.res1), tc.res1.String(), resultLength(tc.res2), tc.res2.String())
 			failures++
 		}
 	}
@@ -1872,3 +1872,17 @@ func isValidQuery(generatedQuery parser.Expr, skipStdAggregations bool) bool {
 	}
 	return isValid
 }
+
+func resultLength(x model.Value) int {
+	vx, xvec := x.(model.Vector)
+	if xvec {
+		return vx.Len()
+	}
+
+	mx, xMatrix := x.(model.Matrix)
+	if xMatrix {
+		return mx.Len()
+	}
+	// Other type, return 0
+	return 0
+}

pkg/cortex/modules.go

@@ -511,7 +511,7 @@ func (t *Cortex) initFlusher() (serv services.Service, err error) {
 // initQueryFrontendTripperware instantiates the tripperware used by the query frontend
 // to optimize Prometheus query requests.
 func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err error) {
-	queryAnalyzer := querysharding.NewQueryAnalyzer()
+	queryAnalyzer := tripperware.NewDisableWithoutNameAnalyzer(querysharding.NewQueryAnalyzer())
 	// PrometheusCodec is a codec to encode and decode Prometheus query range requests and responses.
 	prometheusCodec := queryrange.NewPrometheusCodec(false, t.Cfg.Querier.ResponseCompression, t.Cfg.API.QuerierDefaultCodec)
 	// ShardedPrometheusCodec is same as PrometheusCodec but to be used on the sharded queries (it sum up the stats)

pkg/querier/tripperware/shard_by.go

@@ -3,9 +3,11 @@ package tripperware
 import (
 	"context"
 	"net/http"
+	"slices"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/prometheus/prometheus/model/labels"
 	"github.com/thanos-io/thanos/pkg/querysharding"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"github.com/weaveworks/common/httpgrpc"
@@ -124,3 +126,25 @@ func VerticalShardSizeFromContext(ctx context.Context) (int, bool) {
 func InjectVerticalShardSizeToContext(ctx context.Context, verticalShardSize int) context.Context {
 	return context.WithValue(ctx, verticalShardsKey{}, verticalShardSize)
 }
+
+type disableWithoutNameAnalyzer struct {
+	analyzer querysharding.Analyzer
+}
+
+func NewDisableWithoutNameAnalyzer(analyzer querysharding.Analyzer) *disableWithoutNameAnalyzer {
+	return &disableWithoutNameAnalyzer{analyzer: analyzer}
+}
+
+func (d *disableWithoutNameAnalyzer) Analyze(query string) (querysharding.QueryAnalysis, error) {
+	analysis, err := d.analyzer.Analyze(query)
+	if err != nil || !analysis.IsShardable() || analysis.ShardBy() {
+		return analysis, err
+	}
+
+	// We are only interested in not shard by case.
+	if slices.Contains(analysis.ShardingLabels(), labels.MetricName) {
+		// Mark as not shardable.
+		return querysharding.QueryAnalysis{}, nil
+	}
+	return analysis, nil
+}