Add projectionHint to SG Parquet mode

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## master / unreleased
 
 * [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `source_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828
+* [FEATURE] StoreGateway: Add a flag `-blocks-storage.bucket-store.honor-projection-hints`. If enabled, Store Gateway in Parquet mode will honor projection hints and only materialize requested labels. #7206
 * [FEATURE] Distributor: Add experimental OTLP metrics ingestion over gRPC (`opentelemetry.proto.collector.metrics.v1.MetricsService/Export`) on the gRPC server port. Enable it with `-distributor.otlp.grpc-enabled`. The tenant is read from the `X-Scope-OrgID` gRPC metadata, and the request size is limited by `-server.grpc-max-recv-msg-size-bytes`. #7873
 * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539
 * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873

docs/blocks-storage/querier.md

@@ -2166,6 +2166,15 @@ blocks_storage:
     # CLI flag: -blocks-storage.bucket-store.parquet-shard-cache-ttl
     [parquet_shard_cache_ttl: <duration> | default = 24h]
 
+    # [Experimental] If enabled, Store Gateway will honor projection hints and
+    # only materialize requested labels. It only takes effect when
+    # `-blocks-storage.bucket-store.bucket-store-type` is parquet,
+    # `-querier.honor-projection-hints` is enabled and the querier uses the
+    # Thanos engine (`-querier.thanos-engine`) with the `projection` optimizer
+    # (`-querier.optimizers`).
+    # CLI flag: -blocks-storage.bucket-store.honor-projection-hints
+    [honor_projection_hints: <boolean> | default = false]
+
     # Maximum number of concurrent goroutines per query applied at each level of
     # parquet processing: shard querying, row group processing, and column
     # materialization. Note: this limit is applied independently at each level,

docs/blocks-storage/store-gateway.md

@@ -2219,6 +2219,15 @@ blocks_storage:
     # CLI flag: -blocks-storage.bucket-store.parquet-shard-cache-ttl
     [parquet_shard_cache_ttl: <duration> | default = 24h]
 
+    # [Experimental] If enabled, Store Gateway will honor projection hints and
+    # only materialize requested labels. It only takes effect when
+    # `-blocks-storage.bucket-store.bucket-store-type` is parquet,
+    # `-querier.honor-projection-hints` is enabled and the querier uses the
+    # Thanos engine (`-querier.thanos-engine`) with the `projection` optimizer
+    # (`-querier.optimizers`).
+    # CLI flag: -blocks-storage.bucket-store.honor-projection-hints
+    [honor_projection_hints: <boolean> | default = false]
+
     # Maximum number of concurrent goroutines per query applied at each level of
     # parquet processing: shard querying, row group processing, and column
     # materialization. Note: this limit is applied independently at each level,

docs/configuration/config-file-reference.md

@@ -2863,6 +2863,15 @@ bucket_store:
   # CLI flag: -blocks-storage.bucket-store.parquet-shard-cache-ttl
   [parquet_shard_cache_ttl: <duration> | default = 24h]
 
+  # [Experimental] If enabled, Store Gateway will honor projection hints and
+  # only materialize requested labels. It only takes effect when
+  # `-blocks-storage.bucket-store.bucket-store-type` is parquet,
+  # `-querier.honor-projection-hints` is enabled and the querier uses the Thanos
+  # engine (`-querier.thanos-engine`) with the `projection` optimizer
+  # (`-querier.optimizers`).
+  # CLI flag: -blocks-storage.bucket-store.honor-projection-hints
+  [honor_projection_hints: <boolean> | default = false]
+
   # Maximum number of concurrent goroutines per query applied at each level of
   # parquet processing: shard querying, row group processing, and column
   # materialization. Note: this limit is applied independently at each level, so

docs/configuration/v1-guarantees.md

@@ -97,6 +97,8 @@ Currently experimental features are:
 - Store Gateway Zone Stable Shuffle Sharding
   - `-store-gateway.sharding-ring.zone-stable-shuffle-sharding` CLI flag
   - `zone_stable_shuffle_sharding` (boolean) field in config file
+- Store Gateway HonorProjectionHints in Parquet Mode
+  - `-blocks-storage.bucket-store.honor-projection-hints` CLI flag
 - Basic Lifecycler (Storegateway, Alertmanager, Ruler) Final Sleep on shutdown, which tells the pod wait before shutdown, allowing a delay to propagate ring changes.
   - `-ruler.ring.final-sleep` (duration) CLI flag
   - `store-gateway.sharding-ring.final-sleep` (duration) CLI flag

integration/parquet_store_gateway_test.go

@@ -0,0 +1,240 @@
+//go:build integration_querier
+
+package integration
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"path/filepath"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/thanos/pkg/block"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
+
+	"github.com/cortexproject/cortex/integration/e2e"
+	e2ecache "github.com/cortexproject/cortex/integration/e2e/cache"
+	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+	"github.com/cortexproject/cortex/integration/e2ecortex"
+	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/util/log"
+	cortex_testutil "github.com/cortexproject/cortex/pkg/util/test"
+)
+
+func TestParquetBucketStore_ProjectionHint(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	consul := e2edb.NewConsulWithName("consul")
+	minio := e2edb.NewMinio(9000, bucketName)
+	memcached := e2ecache.NewMemcached()
+	require.NoError(t, s.StartAndWaitReady(consul, minio, memcached))
+
+	// Define configuration flags.
+	flags := BlocksStorageFlags()
+	flags = mergeFlags(flags, map[string]string{
+		// Enable Thanos engine and projection optimization.
+		"-querier.thanos-engine": "true",
+		"-querier.optimizers":    "projection",
+
+		// enable honor-projection-hints querier and store gateway
+		"-querier.honor-projection-hints":                     "true",
+		"-blocks-storage.bucket-store.honor-projection-hints": "true",
+		// enable Store Gateway Parquet mode
+		"-blocks-storage.bucket-store.bucket-store-type": "parquet",
+
+		// Set query-ingesters-within to 1h so queries older than 1h don't hit ingesters
+		"-limits.query-ingesters-within": "1h",
+
+		// Configure Parquet Converter
+		"-parquet-converter.enabled":              "true",
+		"-parquet-converter.conversion-interval":  "1s",
+		"-parquet-converter.ring.consul.hostname": consul.NetworkHTTPEndpoint(),
+		"-compactor.block-ranges":                 "1ms,12h",
+		// Enable cache
+		"-blocks-storage.bucket-store.parquet-labels-cache.backend":             "inmemory,memcached",
+		"-blocks-storage.bucket-store.parquet-labels-cache.memcached.addresses": "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort),
+		"-blocks-storage.bucket-store.sync-interval":                            "1s",
+
+		// Compactor
+		"-compactor.cleanup-interval": "1s", // to update bucket index quickly
+	})
+
+	// Store Gateway
+	storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(storeGateway))
+
+	// Parquet Converter
+	parquetConverter := e2ecortex.NewParquetConverter("parquet-converter", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(parquetConverter))
+
+	// Querier (honor projection hints enabled on the querier side)
+	querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
+		"-querier.store-gateway-addresses": storeGateway.NetworkGRPCEndpoint(),
+	}), "")
+	require.NoError(t, s.StartAndWaitReady(querier))
+
+	// Querier with honor-projection-hints disabled on the querier side. The querier drops the
+	// projection hints, so the Store Gateway materializes all labels.
+	querierHonorOff := e2ecortex.NewQuerier("querier-honor-off", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
+		"-querier.store-gateway-addresses": storeGateway.NetworkGRPCEndpoint(),
+		"-querier.honor-projection-hints":  "false",
+	}), "")
+	require.NoError(t, s.StartAndWaitReady(querierHonorOff))
+
+	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+	require.NoError(t, querierHonorOff.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+	require.NoError(t, storeGateway.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+	// Create block
+	now := time.Now()
+	// Time range: [Now - 24h] to [Now - 20h]
+	start := now.Add(-24 * time.Hour)
+	end := now.Add(-20 * time.Hour)
+
+	ctx := context.Background()
+
+	rnd := rand.New(rand.NewSource(time.Now().Unix()))
+	dir := filepath.Join(s.SharedDir(), "data")
+	scrapeInterval := time.Minute
+	statusCodes := []string{"200", "400", "404", "500", "502"}
+	methods := []string{"GET", "POST", "PUT", "DELETE"}
+
+	numSeries := 10
+	numSamples := 100
+
+	lbls := make([]labels.Labels, 0, numSeries)
+	for i := range numSeries {
+		lbls = append(lbls, labels.FromStrings(
+			labels.MetricName, "http_requests_total",
+			"job", "api-server",
+			"instance", fmt.Sprintf("instance-%d", i),
+			"status_code", statusCodes[i%len(statusCodes)],
+			"method", methods[i%len(methods)],
+			"path", fmt.Sprintf("/api/v1/endpoint%d", i%3),
+			"cluster", "test-cluster",
+		))
+	}
+
+	id, err := e2e.CreateBlock(ctx, rnd, dir, lbls, numSamples, start.UnixMilli(), end.UnixMilli(), scrapeInterval.Milliseconds(), 10)
+	require.NoError(t, err)
+
+	storage, err := e2ecortex.NewS3ClientForMinio(minio, bucketName)
+	require.NoError(t, err)
+	bkt := bucket.NewUserBucketClient("user-1", storage.GetBucket(), nil)
+
+	// Upload TSDB Block
+	require.NoError(t, block.Upload(ctx, log.Logger, bkt, filepath.Join(dir, id.String()), metadata.NoneFunc))
+
+	// Start compactor to create the bucket index.
+	compactor := e2ecortex.NewCompactor("compactor", consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(compactor))
+
+	// Wait until parquet converter convert block
+	require.NoError(t, parquetConverter.WaitSumMetricsWithOptions(e2e.Equals(float64(1)), []string{"cortex_parquet_converter_blocks_converted_total"}, e2e.WaitMissingMetrics))
+
+	// Create clients for both queriers (honor enabled / disabled on the querier side).
+	c, err := e2ecortex.NewClient("", querier.HTTPEndpoint(), "", "", "user-1")
+	require.NoError(t, err)
+	cHonorOff, err := e2ecortex.NewClient("", querierHonorOff.HTTPEndpoint(), "", "", "user-1")
+	require.NoError(t, err)
+
+	cortex_testutil.Poll(t, 60*time.Second, true, func() any {
+		labelSets, err := c.Series([]string{`{job="api-server"}`}, start, end)
+		if err != nil {
+			t.Logf("Series query failed: %v", err)
+			return false
+		}
+		return len(labelSets) > 0
+	})
+
+	testCases := []struct {
+		name           string
+		query          string
+		expectedLabels []string // query result should contain these labels
+	}{
+		{
+			name:  "vector selector query",
+			query: `http_requests_total`,
+			expectedLabels: []string{
+				"__name__", "job", "instance", "status_code", "method", "path", "cluster",
+			},
+		},
+		{
+			name:           "simple_sum_by_job",
+			query:          `sum by (job) (http_requests_total)`,
+			expectedLabels: []string{"job"},
+		},
+		{
+			name:           "rate_with_aggregation",
+			query:          `sum by (method) (rate(http_requests_total[5m]))`,
+			expectedLabels: []string{"method"},
+		},
+		{
+			name:           "multiple_grouping_labels",
+			query:          `sum by (job, status_code) (http_requests_total)`,
+			expectedLabels: []string{"job", "status_code"},
+		},
+		{
+			name:           "aggregation without query",
+			query:          `sum without (instance, method) (http_requests_total)`,
+			expectedLabels: []string{"job", "status_code", "path", "cluster"},
+		},
+	}
+	// Run the same assertions against both queriers. The querier with honor-projection-hints
+	// disabled must produce identical results even though the Store Gateway does not project.
+	clients := []struct {
+		name   string
+		client *e2ecortex.Client
+	}{
+		{name: "querier_honor_on", client: c},
+		{name: "querier_honor_off", client: cHonorOff},
+	}
+
+	for _, qc := range clients {
+		t.Run(qc.name, func(t *testing.T) {
+			for _, tc := range testCases {
+				t.Run(tc.name, func(t *testing.T) {
+					t.Logf("Testing: %s", tc.query)
+
+					// Execute instant query
+					result, err := qc.client.Query(tc.query, end)
+					require.NoError(t, err)
+					require.NotNil(t, result)
+
+					// Verify we got results
+					vector, ok := result.(model.Vector)
+					require.True(t, ok, "result should be a vector")
+					require.NotEmpty(t, vector, "query should return results")
+
+					for _, sample := range vector {
+						actualLabels := make(map[string]struct{})
+						for label := range sample.Metric {
+							actualLabels[string(label)] = struct{}{}
+						}
+
+						// Check that all expected labels are present
+						for _, expectedLabel := range tc.expectedLabels {
+							_, ok := actualLabels[expectedLabel]
+							require.True(t, ok,
+								"series should have %s label", expectedLabel)
+						}
+
+						// Check that no unexpected labels are present
+						for lbl := range actualLabels {
+							if !slices.Contains(tc.expectedLabels, lbl) {
+								require.Failf(t, "unexpected label", "series should not have unexpected label: %s", lbl)
+							}
+						}
+					}
+				})
+			}
+		})
+	}
+}

pkg/querier/blocks_store_queryable.go

@@ -368,7 +368,7 @@ func (q *blocksStoreQuerier) LabelNames(ctx context.Context, hints *storage.Labe
 		convertedMatchers = convertMatchersToLabelMatcher(matchers)
 	)
 
-	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64) ([]ulid.ULID, error, error) {
+	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, _ bucketindex.Blocks, minT, maxT int64) ([]ulid.ULID, error, error) {
 		nameSets, warnings, queriedBlocks, err, retryableError := q.fetchLabelNamesFromStore(spanCtx, userID, clients, minT, maxT, limit, convertedMatchers)
 		if err != nil {
 			return nil, err, retryableError
@@ -411,7 +411,7 @@ func (q *blocksStoreQuerier) LabelValues(ctx context.Context, name string, hints
 		resultMtx sync.Mutex
 	)
 
-	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64) ([]ulid.ULID, error, error) {
+	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, _ bucketindex.Blocks, minT, maxT int64) ([]ulid.ULID, error, error) {
 		valueSets, warnings, queriedBlocks, err, retryableError := q.fetchLabelValuesFromStore(spanCtx, userID, name, clients, minT, maxT, limit, matchers...)
 		if err != nil {
 			return nil, err, retryableError
@@ -460,8 +460,9 @@ func (q *blocksStoreQuerier) selectSorted(ctx context.Context, sp *storage.Selec
 		resultMtx sync.Mutex
 	)
 
-	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64) ([]ulid.ULID, error, error) {
-		seriesSets, queriedBlocks, warnings, numChunks, err, retryableError := q.fetchSeriesFromStores(spanCtx, sp, userID, clients, minT, maxT, limit, matchers, maxChunksLimit, leftChunksLimit)
+	queryFunc := func(clients map[BlocksStoreClient][]ulid.ULID, knownBlocks bucketindex.Blocks, minT, maxT int64) ([]ulid.ULID, error, error) {
+		storeHints := storeGatewaySelectHints(sp, knownBlocks)
+		seriesSets, queriedBlocks, warnings, numChunks, err, retryableError := q.fetchSeriesFromStores(spanCtx, storeHints, userID, clients, minT, maxT, limit, matchers, maxChunksLimit, leftChunksLimit)
 		if err != nil {
 			return nil, err, retryableError
 		}
@@ -494,8 +495,24 @@ func (q *blocksStoreQuerier) selectSorted(ctx context.Context, sp *storage.Selec
 		resWarnings)
 }
 
+// storeGatewaySelectHints returns the select hints to send to store-gateways for the given blocks.
+// Projection hints are reset unless all blocks have the series hash column (parquet version >= 2).
+// The decision is taken once for all blocks before fanning out, because blocks are spread across
+// store-gateways and a series must have the same labels in every response to be merged.
+func storeGatewaySelectHints(sp *storage.SelectHints, blocks bucketindex.Blocks) *storage.SelectHints {
+	if sp == nil || !sp.ProjectionInclude || allParquetBlocksHaveHashColumn(blocks) {
+		return sp
+	}
+
+	// Copy so the caller's hints are not modified.
+	hints := *sp
+	hints.ProjectionInclude = false
+	hints.ProjectionLabels = nil
+	return &hints
+}
+
 func (q *blocksStoreQuerier) queryWithConsistencyCheck(ctx context.Context, logger log.Logger, minT, maxT int64, matchers []*labels.Matcher,
-	userID string, queryFunc func(clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64) ([]ulid.ULID, error, error)) error {
+	userID string, queryFunc func(clients map[BlocksStoreClient][]ulid.ULID, knownBlocks bucketindex.Blocks, minT, maxT int64) ([]ulid.ULID, error, error)) error {
 	queryStoreAfter := q.limits.QueryStoreAfter(userID)
 	// If queryStoreAfter is enabled, we do manipulate the query maxt to query samples up until
 	// now - queryStoreAfter, because the most recent time range is covered by ingesters. This
@@ -567,7 +584,7 @@ func (q *blocksStoreQuerier) queryWithConsistencyCheck(ctx context.Context, logg
 
 		// Fetch series from stores. If an error occur we do not retry because retries
 		// are only meant to cover missing blocks.
-		queriedBlocks, err, retryableError = queryFunc(clients, minT, maxT)
+		queriedBlocks, err, retryableError = queryFunc(clients, knownBlocks, minT, maxT)
 		if err != nil {
 			return err
 		}