Add parquet shard concurrency configuration for store-gateway

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

Author: SungJin1212

CHANGELOG.md

@@ -16,6 +16,7 @@
 * [ENHANCEMENT] Tenant Federation: Avoid purging the regex resolver LRU cache on user-sync ticks when the set of known users has not changed. #7489
 * [ENHANCEMENT] Memberlist: Add `-memberlist.packet-read-timeout`, `-memberlist.max-packet-size`, and `-memberlist.max-concurrent-connections` flags to bound inbound gossip TCP connections, preventing slow-read, OOM, and connection-flood attacks on the gossip port. #7518
 * [ENHANCEMENT] Parquet Converter: Add a ring status page to expose the ring status. #7455
+* [ENHANCEMENT] Parquet: Add `-blocks-storage.bucket-store.parquet-shard-concurrency` flag to configure the maximum number of concurrent goroutines applied at each level of parquet query processing in store-gateway: shard querying, row group filtering, and column materialization. #7613
 * [ENHANCEMENT] Parquet: Add a row ranges cache for parquet query filtering in querier and store-gateway. #7478
 * [ENHANCEMENT] Ingester: Add WAL record metrics to help evaluate the effectiveness of WAL compression type (e.g. snappy, zstd): `cortex_ingester_tsdb_wal_record_part_writes_total`, `cortex_ingester_tsdb_wal_record_parts_bytes_written_total`, and `cortex_ingester_tsdb_wal_record_bytes_saved_total`. #7420
 * [ENHANCEMENT] Distributor: Introduce dynamic `Symbols` slice capacity pooling. #7398 #7401

docs/blocks-storage/querier.md

@@ -2098,6 +2098,12 @@ blocks_storage:
     # CLI flag: -blocks-storage.bucket-store.parquet-shard-cache-ttl
     [parquet_shard_cache_ttl: <duration> | default = 24h]
 
+    # Maximum number of concurrent goroutines per query applied at each level of
+    # parquet processing in store-gateway: shard querying, row group filtering,
+    # and column materialization.
+    # CLI flag: -blocks-storage.bucket-store.parquet-shard-concurrency
+    [parquet_shard_concurrency: <int> | default = 4]
+
   tsdb:
     # Local directory to store TSDBs in the ingesters.
     # CLI flag: -blocks-storage.tsdb.dir

docs/blocks-storage/store-gateway.md

@@ -2156,6 +2156,12 @@ blocks_storage:
     # CLI flag: -blocks-storage.bucket-store.parquet-shard-cache-ttl
     [parquet_shard_cache_ttl: <duration> | default = 24h]
 
+    # Maximum number of concurrent goroutines per query applied at each level of
+    # parquet processing in store-gateway: shard querying, row group filtering,
+    # and column materialization.
+    # CLI flag: -blocks-storage.bucket-store.parquet-shard-concurrency
+    [parquet_shard_concurrency: <int> | default = 4]
+
   tsdb:
     # Local directory to store TSDBs in the ingesters.
     # CLI flag: -blocks-storage.tsdb.dir

docs/configuration/config-file-reference.md

@@ -2735,6 +2735,12 @@ bucket_store:
   # CLI flag: -blocks-storage.bucket-store.parquet-shard-cache-ttl
   [parquet_shard_cache_ttl: <duration> | default = 24h]
 
+  # Maximum number of concurrent goroutines per query applied at each level of
+  # parquet processing in store-gateway: shard querying, row group filtering,
+  # and column materialization.
+  # CLI flag: -blocks-storage.bucket-store.parquet-shard-concurrency
+  [parquet_shard_concurrency: <int> | default = 4]
+
 tsdb:
   # Local directory to store TSDBs in the ingesters.
   # CLI flag: -blocks-storage.tsdb.dir

pkg/storage/tsdb/config.go

@@ -49,15 +49,16 @@ const (
 
 // Validation errors
 var (
-	errInvalidShipConcurrency        = errors.New("invalid TSDB ship concurrency")
-	errInvalidOpeningConcurrency     = errors.New("invalid TSDB opening concurrency")
-	errInvalidCompactionInterval     = errors.New("invalid TSDB compaction interval")
-	errInvalidCompactionConcurrency  = errors.New("invalid TSDB compaction concurrency")
-	errInvalidWALSegmentSizeBytes    = errors.New("invalid TSDB WAL segment size bytes")
-	errInvalidStripeSize             = errors.New("invalid TSDB stripe size")
-	errInvalidOutOfOrderCapMax       = errors.New("invalid TSDB OOO chunks capacity (in samples)")
-	errEmptyBlockranges              = errors.New("empty block ranges for TSDB")
-	errUnSupportedWALCompressionType = errors.New("unsupported WAL compression type, valid types are (zstd, snappy and '')")
+	errInvalidShipConcurrency         = errors.New("invalid TSDB ship concurrency")
+	errInvalidOpeningConcurrency      = errors.New("invalid TSDB opening concurrency")
+	errInvalidCompactionInterval      = errors.New("invalid TSDB compaction interval")
+	errInvalidCompactionConcurrency   = errors.New("invalid TSDB compaction concurrency")
+	errInvalidWALSegmentSizeBytes     = errors.New("invalid TSDB WAL segment size bytes")
+	errInvalidStripeSize              = errors.New("invalid TSDB stripe size")
+	errInvalidOutOfOrderCapMax        = errors.New("invalid TSDB OOO chunks capacity (in samples)")
+	errEmptyBlockranges               = errors.New("empty block ranges for TSDB")
+	errUnSupportedWALCompressionType  = errors.New("unsupported WAL compression type, valid types are (zstd, snappy and '')")
+	errInvalidParquetShardConcurrency = errors.New("invalid parquet shard concurrency, the value must be greater than 0")
 
 	ErrInvalidBucketIndexBlockDiscoveryStrategy         = errors.New("bucket index block discovery strategy can only be enabled when bucket index is enabled")
 	ErrBlockDiscoveryStrategy                           = errors.New("invalid block discovery strategy")
@@ -336,6 +337,10 @@ type BucketStoreConfig struct {
 	TokenBucketBytesLimiter TokenBucketBytesLimiterConfig `yaml:"token_bucket_bytes_limiter"`
 	// Parquet shard cache config
 	ParquetShardCache parquetutil.CacheConfig `yaml:",inline"`
+
+	// ParquetShardConcurrency controls the maximum number of concurrent goroutines
+	// used when querying parquet shards.
+	ParquetShardConcurrency int `yaml:"parquet_shard_concurrency"`
 }
 
 type TokenBucketBytesLimiterConfig struct {
@@ -398,6 +403,7 @@ func (cfg *BucketStoreConfig) RegisterFlags(f *flag.FlagSet) {
 	f.Float64Var(&cfg.TokenBucketBytesLimiter.FetchedChunksTokenFactor, "blocks-storage.bucket-store.token-bucket-bytes-limiter.fetched-chunks-token-factor", 0, "Multiplication factor used for fetched chunks token")
 	f.Float64Var(&cfg.TokenBucketBytesLimiter.TouchedChunksTokenFactor, "blocks-storage.bucket-store.token-bucket-bytes-limiter.touched-chunks-token-factor", 1, "Multiplication factor used for touched chunks token")
 	f.IntVar(&cfg.MatchersCacheMaxItems, "blocks-storage.bucket-store.matchers-cache-max-items", 0, "Maximum number of entries in the regex matchers cache. 0 to disable.")
+	f.IntVar(&cfg.ParquetShardConcurrency, "blocks-storage.bucket-store.parquet-shard-concurrency", 4, "Maximum number of concurrent goroutines per query applied at each level of parquet processing in store-gateway: shard querying, row group filtering, and column materialization.")
 	cfg.ParquetShardCache.RegisterFlagsWithPrefix("blocks-storage.bucket-store.", f)
 }
 
@@ -435,6 +441,9 @@ func (cfg *BucketStoreConfig) Validate() error {
 	if cfg.LazyExpandedPostingGroupMaxKeySeriesRatio < 0 {
 		return ErrInvalidLazyExpandedPostingGroupMaxKeySeriesRatio
 	}
+	if cfg.ParquetShardConcurrency <= 0 {
+		return errInvalidParquetShardConcurrency
+	}
 	return nil
 }
 

pkg/storage/tsdb/config_test.go

@@ -145,6 +145,24 @@ func TestConfig_Validate(t *testing.T) {
 			},
 			expectedErr: errUnSupportedWALCompressionType,
 		},
+		"should fail on parquet shard concurrency set to 0": {
+			setup: func(cfg *BlocksStorageConfig) {
+				cfg.BucketStore.ParquetShardConcurrency = 0
+			},
+			expectedErr: errInvalidParquetShardConcurrency,
+		},
+		"should fail on negative parquet shard concurrency": {
+			setup: func(cfg *BlocksStorageConfig) {
+				cfg.BucketStore.ParquetShardConcurrency = -1
+			},
+			expectedErr: errInvalidParquetShardConcurrency,
+		},
+		"should pass on valid parquet shard concurrency": {
+			setup: func(cfg *BlocksStorageConfig) {
+				cfg.BucketStore.ParquetShardConcurrency = 4
+			},
+			expectedErr: nil,
+		},
 	}
 
 	for testName, testData := range tests {

pkg/storegateway/parquet_bucket_stores.go

@@ -273,7 +273,7 @@ func (u *ParquetBucketStores) createParquetBucketStore(userID string, userLogger
 		logger:            userLogger,
 		bucket:            userBucket,
 		limits:            u.limits,
-		concurrency:       4, // TODO: make this configurable
+		concurrency:       u.cfg.BucketStore.ParquetShardConcurrency,
 		chunksDecoder:     u.chunksDecoder,
 		matcherCache:      u.matcherCache,
 		parquetShardCache: u.parquetShardCache,

schemas/cortex-config-schema.json

@@ -2952,6 +2952,12 @@
               "x-cli-flag": "blocks-storage.bucket-store.parquet-shard-cache-ttl",
               "x-format": "duration"
             },
+            "parquet_shard_concurrency": {
+              "default": 4,
+              "description": "Maximum number of concurrent goroutines per query applied at each level of parquet processing in store-gateway: shard querying, row group filtering, and column materialization.",
+              "type": "number",
+              "x-cli-flag": "blocks-storage.bucket-store.parquet-shard-concurrency"
+            },
             "series_batch_size": {
               "default": 10000,
               "description": "Controls how many series to fetch per batch in Store Gateway. Default value is 10000.",