parquet bucket store: support sharded parquet file querying

Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>

Author: SungJin1212

CHANGELOG.md

@@ -4,6 +4,7 @@
 * [CHANGE] Querier: Make query time range configurations per-tenant: `query_ingesters_within`, `query_store_after`, and `shuffle_sharding_ingesters_lookback_period`. Uses `model.Duration` instead of `time.Duration` to support serialization but has minimum unit of 1ms (nanoseconds/microseconds not supported). #7160
 * [CHANGE] Cache: Setting `-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl` to 0 will disable the bucket-index cache. #7446
 * [CHANGE] HA Tracker: Move `-distributor.ha-tracker.failover-timeout` from a global config to a per-tenant runtime config. The flag name and default value (30s) remain the same. #7481
+* [FEATURE] Parquet: Support sharded parquet file conversion and querying. #7610
 * [FEATURE] Ingester: Add experimental active series tracker that counts active series by configurable label matchers (including regex) per tenant and exposes `cortex_ingester_active_series_per_tracker` metric. Configured via `active_series_trackers` in runtime config overrides. #7476
 * [FEATURE] Ingester: Add experimental head-only queried series metric. `cortex_ingester_queried_head_series` tracks unique series queried from head via HLL. Enabled via `-ingester.head-queried-series-metrics-enabled`. #7500
 * [FEATURE] Ruler: Add per-tenant `ruler_alert_generator_url_template` runtime config option to customize alert generator URLs using Go templates. Includes a `jsonEscape` template function for safely embedding expressions in JSON-encoded URL parameters (e.g., Grafana Explore panes). Supports Grafana Explore, Perses, and other UIs. #7302

pkg/storegateway/parquet_bucket_store.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/gogo/protobuf/types"
+	"github.com/oklog/ulid/v2"
 	"github.com/pkg/errors"
 	"github.com/prometheus-community/parquet-common/convert"
 	"github.com/prometheus-community/parquet-common/schema"
@@ -25,14 +26,15 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	cortex_parquet "github.com/cortexproject/cortex/pkg/storage/parquet"
 	"github.com/cortexproject/cortex/pkg/util/parquetutil"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
 type parquetBucketStore struct {
 	logger      log.Logger
-	bucket      objstore.Bucket
+	bucket      objstore.InstrumentedBucket
 	limits      *validation.Overrides
 	concurrency int
 
@@ -62,19 +64,51 @@ func (p *parquetBucketStore) findParquetBlocks(ctx context.Context, blockMatcher
 	}
 
 	blockIDs := strings.Split(blockMatchers[0].Value, "|")
-	blocks := make([]*parquetBlock, 0, len(blockIDs))
 	bucketOpener := parquet_storage.NewParquetBucketOpener(p.bucket)
 	noopQuota := search.NewQuota(search.NoopQuotaLimitFunc(ctx))
+
+	// Read converter marks and expand to per-shard (blockID, shardID) lists.
+	var shardBlockIDs []string
+	var shardIDs []int
 	for _, blockID := range blockIDs {
-		// TODO: support shard ID > 0 later.
-		block, err := p.newParquetBlock(ctx, blockID, 0, bucketOpener, bucketOpener, p.chunksDecoder, p.rowRangesCache, noopQuota, noopQuota, noopQuota)
+		uid, err := ulid.Parse(blockID)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to parse block ID %s", blockID)
+		}
+		marker, err := cortex_parquet.ReadConverterMark(ctx, uid, p.bucket, p.logger)
 		if err != nil {
-			return nil, err
+			return nil, errors.Wrapf(err, "failed to read converter mark for block %s", blockID)
 		}
-		blocks = append(blocks, block)
+		numShards := marker.Shards
+		if numShards <= 0 {
+			// backward compatibility: blocks without a shard count have one shard
+			numShards = 1
+		}
+		for shardID := 0; shardID < numShards; shardID++ {
+			shardBlockIDs = append(shardBlockIDs, blockID)
+			shardIDs = append(shardIDs, shardID)
+		}
+	}
+
+	// Open all shards in parallel.
+	parquetBlocks := make([]*parquetBlock, len(shardBlockIDs))
+	errGroup, egCtx := errgroup.WithContext(ctx)
+	errGroup.SetLimit(p.concurrency)
+	for i := range shardBlockIDs {
+		errGroup.Go(func() error {
+			blk, err := p.newParquetBlock(egCtx, shardBlockIDs[i], shardIDs[i], bucketOpener, bucketOpener, p.chunksDecoder, p.rowRangesCache, noopQuota, noopQuota, noopQuota)
+			if err != nil {
+				return err
+			}
+			parquetBlocks[i] = blk
+			return nil
+		})
+	}
+	if err := errGroup.Wait(); err != nil {
+		return nil, err
 	}
 
-	return blocks, nil
+	return parquetBlocks, nil
 }
 
 // Series implements the store interface for a single parquet bucket store
@@ -112,10 +146,14 @@ func (p *parquetBucketStore) Series(req *storepb.SeriesRequest, seriesSrv storep
 	errGroup, ctx := errgroup.WithContext(srv.Context())
 	errGroup.SetLimit(p.concurrency)
 
+	seenBlocks := make(map[string]struct{}, len(shards))
 	for i, shard := range shards {
-		resHints.QueriedBlocks = append(resHints.QueriedBlocks, hintspb.Block{
-			Id: shard.name,
-		})
+		if _, seen := seenBlocks[shard.name]; !seen {
+			seenBlocks[shard.name] = struct{}{}
+			resHints.QueriedBlocks = append(resHints.QueriedBlocks, hintspb.Block{
+				Id: shard.name,
+			})
+		}
 		errGroup.Go(func() error {
 			ss, err := shard.Query(ctx, req.MinTime, req.MaxTime, req.SkipChunks, matchers)
 			seriesSet[i] = ss
@@ -197,10 +235,14 @@ func (p *parquetBucketStore) LabelNames(ctx context.Context, req *storepb.LabelN
 	errGroup, ctx := errgroup.WithContext(ctx)
 	errGroup.SetLimit(p.concurrency)
 
+	seenBlocks := make(map[string]struct{}, len(shards))
 	for i, s := range shards {
-		resHints.QueriedBlocks = append(resHints.QueriedBlocks, hintspb.Block{
-			Id: s.name,
-		})
+		if _, seen := seenBlocks[s.name]; !seen {
+			seenBlocks[s.name] = struct{}{}
+			resHints.QueriedBlocks = append(resHints.QueriedBlocks, hintspb.Block{
+				Id: s.name,
+			})
+		}
 		errGroup.Go(func() error {
 			r, err := s.LabelNames(ctx, req.Limit, matchers)
 			resNameSets[i] = r
@@ -254,10 +296,14 @@ func (p *parquetBucketStore) LabelValues(ctx context.Context, req *storepb.Label
 	errGroup, ctx := errgroup.WithContext(ctx)
 	errGroup.SetLimit(p.concurrency)
 
+	seenBlocks := make(map[string]struct{}, len(shards))
 	for i, s := range shards {
-		resHints.QueriedBlocks = append(resHints.QueriedBlocks, hintspb.Block{
-			Id: s.name,
-		})
+		if _, seen := seenBlocks[s.name]; !seen {
+			seenBlocks[s.name] = struct{}{}
+			resHints.QueriedBlocks = append(resHints.QueriedBlocks, hintspb.Block{
+				Id: s.name,
+			})
+		}
 		errGroup.Go(func() error {
 			r, err := s.LabelValues(ctx, req.Label, req.Limit, matchers)
 			resNameValues[i] = r

pkg/storegateway/parquet_bucket_store_bench_test.go

@@ -120,7 +120,95 @@ func BenchmarkParquetBucketStore_SeriesBatch(b *testing.B) {
 	}
 }
 
+func BenchmarkParquetBucketStore_MultiShard(b *testing.B) {
+	const (
+		totalSeries = 10000
+		numSamples  = 100
+		userID      = "user-1"
+	)
+
+	// shardCases defines configurations to convert a block into different numbers of parquet shards.
+	// totalShards = ceil(NumSeries / (NumRowGroups × MaxRowsPerRowGroup))
+	shardCases := []struct {
+		numShards          int
+		numRowGroups       int
+		maxRowsPerRowGroup int
+	}{
+		{numShards: 1, numRowGroups: math.MaxInt32, maxRowsPerRowGroup: 1_000_000}, // default: single shard (no sharding path)
+		{numShards: 2, numRowGroups: 1, maxRowsPerRowGroup: totalSeries / 2},
+		{numShards: 4, numRowGroups: 1, maxRowsPerRowGroup: totalSeries / 4},
+		{numShards: 8, numRowGroups: 1, maxRowsPerRowGroup: totalSeries / 8},
+	}
+
+	for _, tc := range shardCases {
+		b.Run(fmt.Sprintf("shards=%d", tc.numShards), func(b *testing.B) {
+			ctx := context.Background()
+			tmpDir := b.TempDir()
+			storageDir := filepath.Join(tmpDir, "storage")
+			dataDir := filepath.Join(tmpDir, "data")
+
+			storageCfg := cortex_tsdb.BlocksStorageConfig{
+				UsersScanner: users.UsersScannerConfig{
+					Strategy:       users.UserScanStrategyList,
+					UpdateInterval: time.Second,
+				},
+				Bucket: bucket.Config{
+					Backend: "filesystem",
+					Filesystem: filesystem.Config{
+						Directory: storageDir,
+					},
+				},
+				BucketStore: cortex_tsdb.BucketStoreConfig{
+					SyncDir:                filepath.Join(tmpDir, "sync"),
+					BucketStoreType:        "parquet",
+					BlockDiscoveryStrategy: string(cortex_tsdb.RecursiveDiscovery),
+				},
+			}
+			bucketClient, err := bucket.NewClient(ctx, storageCfg.Bucket, nil, "test", log.NewNopLogger(), prometheus.NewRegistry())
+			require.NoError(b, err)
+
+			blockID := prepareParquetBlockWithShards(
+				b, ctx, storageCfg, bucketClient, dataDir, userID,
+				totalSeries, numSamples, tc.numRowGroups, tc.maxRowsPerRowGroup,
+			)
+
+			stores, err := NewBucketStores(storageCfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bucketClient), defaultLimitsOverrides(nil), mockLoggingLevel(), log.NewNopLogger(), prometheus.NewPedanticRegistry())
+			require.NoError(b, err)
+
+			listener, err := net.Listen("tcp", "localhost:0")
+			require.NoError(b, err)
+			gRPCServer := grpc.NewServer(grpc.StreamInterceptor(middleware.StreamServerUserHeaderInterceptor))
+			storepb.RegisterStoreServer(gRPCServer, stores)
+			go func() {
+				if err := gRPCServer.Serve(listener); err != nil && err != grpc.ErrServerStopped {
+					b.Error(err)
+				}
+			}()
+			defer gRPCServer.Stop()
+
+			conn, err := grpc.NewClient(listener.Addr().String(),
+				grpc.WithTransportCredentials(insecure.NewCredentials()),
+				grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(math.MaxInt32)),
+			)
+			require.NoError(b, err)
+			defer conn.Close()
+
+			gRPCClient := storepb.NewStoreClient(conn)
+
+			b.ResetTimer()
+			b.ReportAllocs()
+			for b.Loop() {
+				benchmarkBatchingForParquetBucketStore(b, gRPCClient, userID, 1000, totalSeries, blockID)
+			}
+		})
+	}
+}
+
 func prepareParquetBlock(b *testing.B, ctx context.Context, storageCfg cortex_tsdb.BlocksStorageConfig, bkt objstore.InstrumentedBucket, dataDir, userID string, numSeries, numSamples int) string {
+	return prepareParquetBlockWithShards(b, ctx, storageCfg, bkt, dataDir, userID, numSeries, numSamples, math.MaxInt32, 1_000_000)
+}
+
+func prepareParquetBlockWithShards(b *testing.B, ctx context.Context, storageCfg cortex_tsdb.BlocksStorageConfig, bkt objstore.InstrumentedBucket, dataDir, userID string, numSeries, numSamples, numRowGroups, maxRowsPerRowGroup int) string {
 	logger := log.NewNopLogger()
 	reg := prometheus.NewRegistry()
 
@@ -170,6 +258,8 @@ func prepareParquetBlock(b *testing.B, ctx context.Context, storageCfg cortex_ts
 	flagext.DefaultValues(&convCfg)
 	convCfg.ConversionInterval = time.Second // to convert quickly
 	convCfg.DataDir = filepath.Join(dataDir, "converter-data")
+	convCfg.NumRowGroups = numRowGroups
+	convCfg.MaxRowsPerRowGroup = maxRowsPerRowGroup
 
 	ringStore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil)
 	b.Cleanup(func() { assert.NoError(b, closer.Close()) })

pkg/storegateway/parquet_bucket_stores.go

@@ -307,7 +307,7 @@ func (p *parquetBucketStore) newParquetBlock(ctx context.Context, name string, s
 			name,
 			labelsFileOpener,
 			chunksFileOpener,
-			0, // we always only have 1 shard - shard 0
+			shardID,
 			parquet_storage.WithFileOptions(
 				parquet.SkipMagicBytes(true),
 				parquet.ReadBufferSize(100*1024),