Add projectionHint to SG Parquet mode

CHANGELOG.md

@@ -4,6 +4,7 @@
 * [BUGFIX] Compactor: Fix the final cleanup of a tenant marked for deletion being reported as failed on object stores that return an error when deleting a missing object (GCS, Azure, Swift, OCI). #7861
 
 * [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
 
 ## 1.22.0 in progress
 * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792

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.Fail(t, "series should not have unexpected label: %s", lbl)
+							}
+						}
+					}
+				})
+			}
+		})
+	}
+}

pkg/querier/querier.go

@@ -529,13 +529,11 @@ func (q querier) Select(ctx context.Context, sortSeries bool, sp *storage.Select
 		}
 	}
 
-	// Reset projection hints if querying ingesters or projection is not included.
-	// Projection can only be applied when not querying mixed sources (ingester + store).
-	if q.honorProjectionHints {
-		if !sp.ProjectionInclude || q.distributor.UseQueryable(q.now, userID, mint, maxt) {
-			sp.ProjectionLabels = nil
-			sp.ProjectionInclude = false
-		}
+	// Reset projection hints unless the querier honors them, projection is included and ingesters are not queried.
+	// Ingesters always return full label sets, so projected store series could not be merged with them.
+	if !q.honorProjectionHints || !sp.ProjectionInclude || q.distributor.UseQueryable(q.now, userID, mint, maxt) {
+		sp.ProjectionLabels = nil
+		sp.ProjectionInclude = false
 	}
 
 	if len(queriers) == 1 {