Compactor: add block merge fan-in to parallelise the compaction merge

CHANGELOG.md

@@ -12,6 +12,7 @@
 * [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
 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889
 * [ENHANCEMENT] Ruler: Retry a notification the Alertmanager failed to accept with a transport error or a 5xx response, instead of dropping it and leaving its alert undelivered until the next evaluation of the rule group. Enable the retries with `-ruler.notification.backoff-retries` (`0`, no retry, by default), and tune the delay between them with `-ruler.notification.backoff-min-period` and `-ruler.notification.backoff-max-period`. The retries are bounded by `-ruler.notification-timeout` and counted by the new `cortex_ruler_notification_retries_total` metric. #7909
+* [ENHANCEMENT] Compactor: Add `-compactor.block-merge-fan-in` to group a partitioning compaction's per-block series sets into subtrees, each advanced by its own goroutine, before the final merge. A flat merge issues every index read from the single goroutine that advances the merge heap, so a compaction over a large number of blocks drives the disk at a queue depth near one however much IOPS it has provisioned, and its wall clock is bound by read latency rather than by disk capacity. Disabled by default (`0`), which keeps the single flat merge. #7920
 * [ENHANCEMENT] Update Go version to 1.27.3, which includes security fixes to the `crypto/x509` package (CVE-2026-111943). #7906 #7923
 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877
 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869

docs/blocks-storage/compactor.md

@@ -113,6 +113,15 @@ compactor:
   # CLI flag: -compactor.meta-sync-concurrency
   [meta_sync_concurrency: <int> | default = 20]
 
+  # Number of per-block series sets merged per subtree during compaction. Each
+  # subtree advances in its own goroutine, so a group of blocks larger than this
+  # value no longer issues every index read from a single goroutine. Only
+  # affects partitioning compaction. 0 disables it and keeps one flat merge over
+  # all blocks. Useful range is 32 to 128; lower values create more goroutines
+  # without adding disk concurrency.
+  # CLI flag: -compactor.block-merge-fan-in
+  [block_merge_fan_in: <int> | default = 0]
+
   # Minimum age of fresh (non-compacted) blocks before they are being processed.
   # Malformed blocks older than the maximum of consistency-delay and 48h0m0s
   # will be removed.

docs/configuration/config-file-reference.md

@@ -3124,6 +3124,15 @@ The `compactor_config` configures the compactor for the blocks storage.
 # CLI flag: -compactor.meta-sync-concurrency
 [meta_sync_concurrency: <int> | default = 20]
 
+# Number of per-block series sets merged per subtree during compaction. Each
+# subtree advances in its own goroutine, so a group of blocks larger than this
+# value no longer issues every index read from a single goroutine. Only affects
+# partitioning compaction. 0 disables it and keeps one flat merge over all
+# blocks. Useful range is 32 to 128; lower values create more goroutines without
+# adding disk concurrency.
+# CLI flag: -compactor.block-merge-fan-in
+[block_merge_fan_in: <int> | default = 0]
+
 # Minimum age of fresh (non-compacted) blocks before they are being processed.
 # Malformed blocks older than the maximum of consistency-delay and 48h0m0s will
 # be removed.

pkg/compactor/background_chunks_series_set.go

@@ -46,10 +46,22 @@ func (b *backgrounChunkSeriesSet) run(ctx context.Context) {
 	}
 }
 
+// defaultBackgroundChunkSeriesSetBuffer is the lookahead used for a single
+// background set. A merge tree holds one buffer per subtree, so inner levels
+// pass a smaller value to bound total series held in memory.
+const defaultBackgroundChunkSeriesSetBuffer = 1000
+
 func NewBackgroundChunkSeriesSet(ctx context.Context, cs storage.ChunkSeriesSet) storage.ChunkSeriesSet {
+	return NewBackgroundChunkSeriesSetWithBuffer(ctx, cs, defaultBackgroundChunkSeriesSetBuffer)
+}
+
+func NewBackgroundChunkSeriesSetWithBuffer(ctx context.Context, cs storage.ChunkSeriesSet, buffer int) storage.ChunkSeriesSet {
+	if buffer < 1 {
+		buffer = 1
+	}
 	r := &backgrounChunkSeriesSet{
 		cs:      cs,
-		nextSet: make(chan storage.ChunkSeries, 1000),
+		nextSet: make(chan storage.ChunkSeries, buffer),
 	}
 
 	go func() {

pkg/compactor/compactor.go

@@ -59,6 +59,7 @@ var (
 
 	supportedShardingStrategies              = []string{util.ShardingStrategyDefault, util.ShardingStrategyShuffle}
 	errInvalidShardingStrategy               = errors.New("invalid sharding strategy")
+	errInvalidBlockMergeFanIn                = errors.New("invalid block merge fan-in, the value must not be negative")
 	errInvalidTenantShardSize                = errors.New("invalid tenant shard size, the value must be greater than 0")
 	supportedCompactionStrategies            = []string{util.CompactionStrategyDefault, util.CompactionStrategyPartitioning}
 	errInvalidCompactionStrategy             = errors.New("invalid compaction strategy")
@@ -173,16 +174,17 @@ var (
 		return NewPartitionCompactionBlockDeletableChecker()
 	}
 
-	DefaultCompactionLifecycleCallbackFactory = func(_ context.Context, _ objstore.InstrumentedBucket, _ log.Logger, _ int, _ string, _ string, _ *compactorMetrics) compact.CompactionLifecycleCallback {
+	DefaultCompactionLifecycleCallbackFactory = func(_ context.Context, _ objstore.InstrumentedBucket, _ log.Logger, _ int, _ int, _ string, _ string, _ *compactorMetrics) compact.CompactionLifecycleCallback {
 		return compact.DefaultCompactionLifecycleCallback{}
 	}
 
-	ShardedCompactionLifecycleCallbackFactory = func(ctx context.Context, userBucket objstore.InstrumentedBucket, logger log.Logger, metaSyncConcurrency int, compactDir string, userID string, compactorMetrics *compactorMetrics) compact.CompactionLifecycleCallback {
+	ShardedCompactionLifecycleCallbackFactory = func(ctx context.Context, userBucket objstore.InstrumentedBucket, logger log.Logger, metaSyncConcurrency int, blockMergeFanIn int, compactDir string, userID string, compactorMetrics *compactorMetrics) compact.CompactionLifecycleCallback {
 		return NewShardedCompactionLifecycleCallback(
 			ctx,
 			userBucket,
 			logger,
 			metaSyncConcurrency,
+			blockMergeFanIn,
 			compactDir,
 			userID,
 			compactorMetrics,
@@ -236,6 +238,7 @@ type CompactionLifecycleCallbackFactory func(
 	userBucket objstore.InstrumentedBucket,
 	logger log.Logger,
 	metaSyncConcurrency int,
+	blockMergeFanIn int,
 	compactDir string,
 	userID string,
 	compactorMetrics *compactorMetrics,
@@ -260,6 +263,7 @@ type Config struct {
 	BlockRangesCompactionDelay            cortex_tsdb.DurationList `yaml:"block_ranges_compaction_delay"`
 	BlockSyncConcurrency                  int                      `yaml:"block_sync_concurrency"`
 	MetaSyncConcurrency                   int                      `yaml:"meta_sync_concurrency"`
+	BlockMergeFanIn                       int                      `yaml:"block_merge_fan_in"`
 	ConsistencyDelay                      time.Duration            `yaml:"consistency_delay"`
 	DataDir                               string                   `yaml:"data_dir"`
 	CompactionInterval                    time.Duration            `yaml:"compaction_interval"`
@@ -324,6 +328,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.ConsistencyDelay, "compactor.consistency-delay", 0, fmt.Sprintf("Minimum age of fresh (non-compacted) blocks before they are being processed. Malformed blocks older than the maximum of consistency-delay and %s will be removed.", compact.PartialUploadThresholdAge))
 	f.IntVar(&cfg.BlockSyncConcurrency, "compactor.block-sync-concurrency", 20, "Number of Go routines to use when syncing block index and chunks files from the long term storage.")
 	f.IntVar(&cfg.MetaSyncConcurrency, "compactor.meta-sync-concurrency", 20, "Number of Go routines to use when syncing block meta files from the long term storage.")
+	f.IntVar(&cfg.BlockMergeFanIn, "compactor.block-merge-fan-in", 0, "Number of per-block series sets merged per subtree during compaction. Each subtree advances in its own goroutine, so a group of blocks larger than this value no longer issues every index read from a single goroutine. Only affects partitioning compaction. 0 disables it and keeps one flat merge over all blocks. Useful range is 32 to 128; lower values create more goroutines without adding disk concurrency.")
 	f.StringVar(&cfg.DataDir, "compactor.data-dir", "./data", "Data directory in which to cache blocks and process compactions")
 	f.DurationVar(&cfg.CompactionInterval, "compactor.compaction-interval", time.Hour, "The frequency at which the compaction runs")
 	f.IntVar(&cfg.CompactionRetries, "compactor.compaction-retries", 3, "How many times to retry a failed compaction within a single compaction run.")
@@ -382,6 +387,10 @@ func (cfg *Config) Validate(limits validation.Limits) error {
 		}
 	}
 
+	if cfg.BlockMergeFanIn < 0 {
+		return errInvalidBlockMergeFanIn
+	}
+
 	// Make sure a valid sharding strategy is being used
 	if !slices.Contains(supportedShardingStrategies, cfg.ShardingStrategy) {
 		return errInvalidShardingStrategy
@@ -1114,7 +1123,7 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 		c.blocksPlannerFactory(currentCtx, bucket, ulogger, c.compactorCfg, noCompactMarkerFilter, c.ringLifecycler, userID, c.blockVisitMarkerReadFailed, c.blockVisitMarkerWriteFailed, c.compactorMetrics, ignoreDeletionMarkFilter),
 		c.blocksCompactor,
 		c.blockDeletableCheckerFactory(currentCtx, bucket, ulogger),
-		c.compactionLifecycleCallbackFactory(currentCtx, bucket, ulogger, c.compactorCfg.MetaSyncConcurrency, c.compactDirForUser(userID), userID, c.compactorMetrics),
+		c.compactionLifecycleCallbackFactory(currentCtx, bucket, ulogger, c.compactorCfg.MetaSyncConcurrency, c.compactorCfg.BlockMergeFanIn, c.compactDirForUser(userID), userID, c.compactorMetrics),
 		c.compactDirForUser(userID),
 		bucket,
 		c.compactorCfg.CompactionConcurrency,

pkg/compactor/sharded_block_populator.go

@@ -23,7 +23,10 @@ import (
 type ShardedBlockPopulator struct {
 	partitionCount int
 	partitionID    int
-	logger         log.Logger
+	// blockMergeFanIn groups the per-block series sets into subtrees before the
+	// final merge. 0 or 1 keeps the single flat merge.
+	blockMergeFanIn int
+	logger          log.Logger
 }
 
 // PopulateBlock fills the index and chunk writers with new data gathered as the union
@@ -131,7 +134,7 @@ func (c ShardedBlockPopulator) PopulateBlock(ctx context.Context, metrics *tsdb.
 		iCtx, cancel := context.WithCancel(ctx)
 		// Merge series using specified chunk series merger.
 		// The default one is the compacting series merger.
-		set = NewBackgroundChunkSeriesSet(iCtx, storage.NewMergeChunkSeriesSet(sets, 0, mergeFunc))
+		set = NewBackgroundChunkSeriesSet(iCtx, storage.NewMergeChunkSeriesSet(c.mergeSubtrees(iCtx, sets, mergeFunc), 0, mergeFunc))
 		defer cancel()
 	}
 
@@ -222,3 +225,30 @@ func (c ShardedBlockPopulator) PopulateBlock(ctx context.Context, metrics *tsdb.
 
 	return nil
 }
+
+// mergeSubtrees groups sets into subtrees, each advanced by its own goroutine,
+// and returns the subtree roots for the caller to merge.
+//
+// A flat merge over every block issues all of its index reads from the single
+// goroutine that advances the heap, so the disk sees a queue depth near one
+// whatever it has provisioned. Grouping gives one in-flight read per subtree.
+// Returns sets unchanged when disabled or when there is nothing to group.
+func (c ShardedBlockPopulator) mergeSubtrees(ctx context.Context, sets []storage.ChunkSeriesSet, mergeFunc storage.VerticalChunkSeriesMergeFunc) []storage.ChunkSeriesSet {
+	fanIn := c.blockMergeFanIn
+	if fanIn < 2 || len(sets) <= fanIn {
+		return sets
+	}
+
+	subtrees := (len(sets) + fanIn - 1) / fanIn
+	// Split the single-set lookahead across the subtrees so total series held
+	// in memory stays comparable to the flat merge.
+	buffer := max(defaultBackgroundChunkSeriesSetBuffer/subtrees, 8)
+
+	roots := make([]storage.ChunkSeriesSet, 0, subtrees)
+	for i := 0; i < len(sets); i += fanIn {
+		group := sets[i:min(i+fanIn, len(sets))]
+		roots = append(roots, NewBackgroundChunkSeriesSetWithBuffer(ctx, storage.NewMergeChunkSeriesSet(group, 0, mergeFunc), buffer))
+	}
+	level.Info(c.logger).Log("msg", "grouped series sets into merge subtrees", "series_sets_count", len(sets), "subtrees", len(roots), "fan_in", fanIn, "subtree_buffer", buffer)
+	return roots
+}