expose projection hints in parquet queryable

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

Author: yeya24

docs/blocks-storage/querier.md

@@ -311,6 +311,18 @@ querier:
   # [Experimental] TTL of the Parquet queryable shard cache. 0 to no TTL.
   # CLI flag: -querier.parquet-queryable-shard-cache-ttl
   [parquet_queryable_shard_cache_ttl: <duration> | default = 24h]
+
+  # [Experimental] If true, parquet queryable will honor projection hints and
+  # only materialize requested labels. Projection is only applied when all
+  # queried blocks are parquet blocks and not querying ingesters.
+  # CLI flag: -querier.parquet-queryable-honor-projection-hints
+  [parquet_queryable_honor_projection_hints: <boolean> | default = false]
+
+  # [Experimental] Time buffer to use when checking if query overlaps with
+  # ingester data. Projection hints are disabled if query time range overlaps
+  # with (now - query-ingesters-within - buffer).
+  # CLI flag: -querier.parquet-queryable-projection-hints-ingester-buffer
+  [parquet_queryable_projection_hints_ingester_buffer: <duration> | default = 1h]
 ```
 
 ### `blocks_storage_config`

docs/configuration/config-file-reference.md

@@ -4828,6 +4828,18 @@ thanos_engine:
 # [Experimental] TTL of the Parquet queryable shard cache. 0 to no TTL.
 # CLI flag: -querier.parquet-queryable-shard-cache-ttl
 [parquet_queryable_shard_cache_ttl: <duration> | default = 24h]
+
+# [Experimental] If true, parquet queryable will honor projection hints and only
+# materialize requested labels. Projection is only applied when all queried
+# blocks are parquet blocks and not querying ingesters.
+# CLI flag: -querier.parquet-queryable-honor-projection-hints
+[parquet_queryable_honor_projection_hints: <boolean> | default = false]
+
+# [Experimental] Time buffer to use when checking if query overlaps with
+# ingester data. Projection hints are disabled if query time range overlaps with
+# (now - query-ingesters-within - buffer).
+# CLI flag: -querier.parquet-queryable-projection-hints-ingester-buffer
+[parquet_queryable_projection_hints_ingester_buffer: <duration> | default = 1h]
 ```
 
 ### `query_frontend_config`

integration/parquet_querier_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/cortexproject/promqlsmith"
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/thanos/pkg/block"
@@ -176,3 +177,190 @@ func TestParquetFuzz(t *testing.T) {
 	require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_parquet_queryable_blocks_queried_total"}, e2e.WithLabelMatchers(
 		labels.MustNewMatcher(labels.MatchEqual, "type", "parquet"))))
 }
+
+func TestParquetProjectionPushdown(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	consul := e2edb.NewConsulWithName("consul")
+	memcached := e2ecache.NewMemcached()
+	require.NoError(t, s.StartAndWaitReady(consul, memcached))
+
+	baseFlags := mergeFlags(AlertmanagerLocalFlags(), BlocksStorageFlags())
+	flags := mergeFlags(
+		baseFlags,
+		map[string]string{
+			"-target": "all,parquet-converter",
+			"-blocks-storage.tsdb.block-ranges-period":                             "1m,24h",
+			"-blocks-storage.tsdb.ship-interval":                                   "1s",
+			"-blocks-storage.bucket-store.sync-interval":                           "1s",
+			"-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl": "1s",
+			"-blocks-storage.bucket-store.bucket-index.idle-timeout":               "1s",
+			"-blocks-storage.bucket-store.bucket-index.enabled":                    "true",
+			"-blocks-storage.bucket-store.index-cache.backend":                     tsdb.IndexCacheBackendInMemory,
+			"-querier.query-store-for-labels-enabled":                              "true",
+			// compactor
+			"-compactor.cleanup-interval": "1s",
+			// Ingester.
+			"-ring.store":      "consul",
+			"-consul.hostname": consul.NetworkHTTPEndpoint(),
+			// Distributor.
+			"-distributor.replication-factor": "1",
+			// Store-gateway.
+			"-store-gateway.sharding-enabled":   "false",
+			"--querier.store-gateway-addresses": "nonExistent", // Make sure we do not call Store gateways
+			// alert manager
+			"-alertmanager.web.external-url": "http://localhost/alertmanager",
+			// parquet-converter
+			"-parquet-converter.ring.consul.hostname": consul.NetworkHTTPEndpoint(),
+			"-parquet-converter.conversion-interval":  "1s",
+			"-parquet-converter.enabled":              "true",
+			// Querier - Enable Thanos engine with projection optimizer
+			"-querier.thanos-engine":                            "true",
+			"-querier.optimizers":                               "default,projection", // Enable projection optimizer
+			"-querier.enable-parquet-queryable":                 "true",
+			"-querier.parquet-queryable-honor-projection-hints": "true", // Honor projection hints
+			// Set query-ingesters-within to 2h so queries older than 2h don't hit ingesters
+			// Since test queries are 24-48h old, they won't query ingesters and projection will be enabled
+			"-querier.query-ingesters-within": "2h"
+			// Enable cache for parquet labels and chunks
+			"-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.chunks-cache.backend":                     "inmemory,memcached",
+			"-blocks-storage.bucket-store.chunks-cache.memcached.addresses":         "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort),
+		},
+	)
+
+	// make alert manager config dir
+	require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{}))
+
+	ctx := context.Background()
+	rnd := rand.New(rand.NewSource(time.Now().Unix()))
+	dir := filepath.Join(s.SharedDir(), "data")
+	numSeries := 20
+	numSamples := 100
+	lbls := make([]labels.Labels, 0, numSeries)
+	scrapeInterval := time.Minute
+	statusCodes := []string{"200", "400", "404", "500", "502"}
+	methods := []string{"GET", "POST", "PUT", "DELETE"}
+	now := time.Now()
+	// Make sure query time is old enough to not overlap with ingesters
+	// With query-ingesters-within=2h, queries with maxT < now-2h won't hit ingesters
+	// Using 24h-48h ago ensures no ingester overlap, allowing projection to be enabled
+	start := now.Add(-time.Hour * 48)
+	end := now.Add(-time.Hour * 24)
+
+	// Create series with multiple labels
+	for i := 0; i < numSeries; i++ {
+		lbls = append(lbls, labels.FromStrings(
+			labels.MetricName, "http_requests_total",
+			"job", "api-server",
+			"instance", fmt.Sprintf("instance-%d", i%5),
+			"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)
+	minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(minio))
+
+	cortex := e2ecortex.NewSingleBinary("cortex", flags, "")
+	require.NoError(t, s.StartAndWaitReady(cortex))
+
+	storage, err := e2ecortex.NewS3ClientForMinio(minio, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, err)
+	bkt := bucket.NewUserBucketClient("user-1", storage.GetBucket(), nil)
+
+	err = block.Upload(ctx, log.Logger, bkt, filepath.Join(dir, id.String()), metadata.NoneFunc)
+	require.NoError(t, err)
+
+	// Wait until we convert the blocks to parquet
+	cortex_testutil.Poll(t, 30*time.Second, true, func() interface{} {
+		found := false
+		foundBucketIndex := false
+
+		err := bkt.Iter(context.Background(), "", func(name string) error {
+			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 && foundBucketIndex
+	})
+
+	c, err := e2ecortex.NewClient("", cortex.HTTPEndpoint(), "", "", "user-1")
+	require.NoError(t, err)
+
+	// Test queries that should use projection hints
+	testCases := []struct {
+		name           string
+		query          string
+		expectedLabels []string // Labels that should be present in result (besides __name__)
+	}{
+		{
+			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"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Logf("Testing: %s", tc.query)
+
+			// Execute instant query
+			result, err := c.Query(tc.query, end)
+			require.NoError(t, err)
+			require.NotNil(t, result)
+
+			// Verify we got results
+			matrix := result.(promql.Matrix)
+			require.NotEmpty(t, matrix, "query should return results")
+
+			t.Logf("Query returned %d series", len(matrix))
+
+			// Verify projection worked: series should only have the expected labels
+			for i, series := range matrix {
+				actualLabels := make(map[string]struct{})
+				for _, label := range series.Metric {
+					actualLabels[label.Name] = struct{}{}
+				}
+
+				// Check that no unexpected labels are present
+				for lbl := range actualLabels {
+					if !slices.Contains(tc.expectedLabels, lbl) {
+						require.Fail(t, "series should not have %s label", lbl)
+					}
+				}
+				// Check that all expected labels are present
+				for _, expectedLabel := range tc.expectedLabels {
+					require.True(t, actualLabels[expectedLabel], 
+						"series should have %s label", expectedLabel)
+				}
+			}
+		})
+	}
+
+	// Verify that parquet blocks were queried
+	require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_parquet_queryable_blocks_queried_total"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "type", "parquet"))))
+}

pkg/querier/parquet_queryable.go

@@ -96,11 +96,13 @@ func newParquetQueryableFallbackMetrics(reg prometheus.Registerer) *parquetQuery
 type parquetQueryableWithFallback struct {
 	services.Service
 
-	fallbackDisabled      bool
-	queryStoreAfter       time.Duration
-	parquetQueryable      storage.Queryable
-	cache                 cacheInterface[parquet_storage.ParquetShard]
-	blockStorageQueryable *BlocksStoreQueryable
+	fallbackDisabled              bool
+	queryStoreAfter               time.Duration
+	queryIngestersWithin          time.Duration
+	projectionHintsIngesterBuffer time.Duration
+	parquetQueryable              storage.Queryable
+	cache                         cacheInterface[parquet_storage.ParquetShard]
+	blockStorageQueryable         *BlocksStoreQueryable
 
 	finder BlocksFinder
 
@@ -148,6 +150,7 @@ func NewParquetQueryable(
 	}
 
 	parquetQueryableOpts := []queryable.QueryableOpts{
+		queryable.WithHonorProjectionHints(config.ParquetQueryableHonorProjectionHints),
 		queryable.WithRowCountLimitFunc(func(ctx context.Context) int64 {
 			// Ignore error as this shouldn't happen.
 			// If failed to resolve tenant we will just use the default limit value.
@@ -253,18 +256,20 @@ func NewParquetQueryable(
 	}, constraintCacheFunc, cDecoder, parquetQueryableOpts...)
 
 	p := &parquetQueryableWithFallback{
-		subservices:           manager,
-		blockStorageQueryable: blockStorageQueryable,
-		parquetQueryable:      parquetQueryable,
-		cache:                 cache,
-		queryStoreAfter:       config.QueryStoreAfter,
-		subservicesWatcher:    services.NewFailureWatcher(),
-		finder:                blockStorageQueryable.finder,
-		metrics:               newParquetQueryableFallbackMetrics(reg),
-		limits:                limits,
-		logger:                logger,
-		defaultBlockStoreType: blockStoreType(config.ParquetQueryableDefaultBlockStore),
-		fallbackDisabled:      config.ParquetQueryableFallbackDisabled,
+		subservices:                   manager,
+		blockStorageQueryable:         blockStorageQueryable,
+		parquetQueryable:              parquetQueryable,
+		cache:                         cache,
+		queryStoreAfter:               config.QueryStoreAfter,
+		queryIngestersWithin:          config.QueryIngestersWithin,
+		projectionHintsIngesterBuffer: config.ParquetQueryableProjectionHintsIngesterBuffer,
+		subservicesWatcher:            services.NewFailureWatcher(),
+		finder:                        blockStorageQueryable.finder,
+		metrics:                       newParquetQueryableFallbackMetrics(reg),
+		limits:                        limits,
+		logger:                        logger,
+		defaultBlockStoreType:         blockStoreType(config.ParquetQueryableDefaultBlockStore),
+		fallbackDisabled:              config.ParquetQueryableFallbackDisabled,
 	}
 
 	p.Service = services.NewBasicService(p.starting, p.running, p.stopping)
@@ -311,17 +316,19 @@ func (p *parquetQueryableWithFallback) Querier(mint, maxt int64) (storage.Querie
 	}
 
 	return &parquetQuerierWithFallback{
-		minT:                  mint,
-		maxT:                  maxt,
-		parquetQuerier:        pq,
-		queryStoreAfter:       p.queryStoreAfter,
-		blocksStoreQuerier:    bsq,
-		finder:                p.finder,
-		metrics:               p.metrics,
-		limits:                p.limits,
-		logger:                p.logger,
-		defaultBlockStoreType: p.defaultBlockStoreType,
-		fallbackDisabled:      p.fallbackDisabled,
+		minT:                          mint,
+		maxT:                          maxt,
+		parquetQuerier:                pq,
+		queryStoreAfter:               p.queryStoreAfter,
+		queryIngestersWithin:          p.queryIngestersWithin,
+		projectionHintsIngesterBuffer: p.projectionHintsIngesterBuffer,
+		blocksStoreQuerier:            bsq,
+		finder:                        p.finder,
+		metrics:                       p.metrics,
+		limits:                        p.limits,
+		logger:                        p.logger,
+		defaultBlockStoreType:         p.defaultBlockStoreType,
+		fallbackDisabled:              p.fallbackDisabled,
 	}, nil
 }
 
@@ -335,7 +342,9 @@ type parquetQuerierWithFallback struct {
 
 	// If set, the querier manipulates the max time to not be greater than
 	// "now - queryStoreAfter" so that most recent blocks are not queried.
-	queryStoreAfter time.Duration
+	queryStoreAfter               time.Duration
+	queryIngestersWithin          time.Duration
+	projectionHintsIngesterBuffer time.Duration
 
 	// metrics
 	metrics *parquetQueryableFallbackMetrics
@@ -500,6 +509,14 @@ func (q *parquetQuerierWithFallback) Select(ctx context.Context, sortSeries bool
 		sortSeries = true
 	}
 
+	queryIngesters := q.queryIngestersWithin == 0 || maxt >= util.TimeToMillis(time.Now().Add(-q.queryIngestersWithin).Add(-q.projectionHintsIngesterBuffer))
+	disableProjection := len(remaining) > 0 || queryIngesters
+	// Reset projection hints if there are mixed blocks (both parquet and non-parquet) or the query needs to merge results between ingester and parquet blocks
+	if disableProjection {
+		hints.ProjectionLabels = nil
+		hints.ProjectionInclude = false
+	}
+
 	promises := make([]chan storage.SeriesSet, 0, 2)
 
 	if len(parquet) > 0 {