Stop losing data when split-and-merge compaction stalls mid-way

cmd/pyroscope/help-all.txt.tmpl

@@ -51,6 +51,8 @@ Usage of ./pyroscope:
     	Duration after which the blocks marked for deletion will be filtered out while fetching blocks. The idea of ignore-deletion-marks-delay is to ignore blocks that are marked for deletion with some delay. This ensures store can still serve blocks that are meant to be deleted but do not have a replacement yet. (default 30m0s)
   -blocks-storage.bucket-store.meta-sync-concurrency int
     	Number of Go routines to use when syncing block meta files from object storage per tenant. (default 20)
+  -blocks-storage.bucket-store.symbol-cache-max-bytes int
+    	[experimental] Max size - in bytes - of the in-process cache of decoded symbol tables, shared across all blocks and tenants. Avoids re-decoding a block's symbols on every query. 0 disables the cache.
   -blocks-storage.bucket-store.sync-dir string
     	Directory to store synchronized pyroscope block headers. This directory is not required to be persisted between restarts, but it's highly recommended in order to improve the store-gateway startup time. (default "./data/pyroscope-sync/")
   -blocks-storage.bucket-store.sync-interval duration

docs/sources/configure-server/reference-configuration-parameters/index.md

@@ -2056,6 +2056,12 @@ bucket_store:
   # have a replacement yet.
   # CLI flag: -blocks-storage.bucket-store.ignore-deletion-marks-delay
   [ignore_deletion_mark_delay: <duration> | default = 30m]
+
+  # (experimental) Max size - in bytes - of the in-process cache of decoded
+  # symbol tables, shared across all blocks and tenants. Avoids re-decoding a
+  # block's symbols on every query. 0 disables the cache.
+  # CLI flag: -blocks-storage.bucket-store.symbol-cache-max-bytes
+  [symbol_cache_max_bytes: <int> | default = 0]
 ```
 
 ### compactor

pkg/compactor/bucket_compactor.go

@@ -340,6 +340,18 @@ func newCompactorMetrics(r prometheus.Registerer) *CompactorMetrics {
 	return m
 }
 
+// isLoneIntermediateShardedBlock reports whether the compaction input is a
+// single sharded block that has not been merged past the deduplication level.
+// The split-and-merge grouper plans such a job on purpose, to promote a block
+// that has no sibling to merge with.
+func isLoneIntermediateShardedBlock(readers []phlaredb.BlockReader) bool {
+	if len(readers) != 1 {
+		return false
+	}
+	meta := readers[0].Meta()
+	return isIntermediateShardedBlock(&meta)
+}
+
 func (c *BlockCompactor) CompactWithSplitting(ctx context.Context, dest string, dirs []string, shardCount, stageSize uint64) ([]ulid.ULID, error) {
 	defer func() {
 		if err := recover(); err != nil {
@@ -410,11 +422,15 @@ func (c *BlockCompactor) CompactWithSplitting(ctx context.Context, dest string,
 	c.metrics.Split.WithLabelValues(fmt.Sprintf("%d", currentLevel)).Observe(float64(shardCount))
 
 	metas, err := phlaredb.CompactWithSplitting(ctx, phlaredb.CompactWithSplittingOpts{
-		Src:                readers,
-		Dst:                dest,
-		SplitCount:         shardCount,
-		StageSize:          stageSize,
-		SplitBy:            c.splitBy,
+		Src:        readers,
+		Dst:        dest,
+		SplitCount: shardCount,
+		StageSize:  stageSize,
+		SplitBy:    c.splitBy,
+		// A merge job with a single input block is planned only to promote a
+		// sharded block that never got merged past the deduplication level, so
+		// the single-block compaction is deliberate here.
+		AllowSingleBlock:   isLoneIntermediateShardedBlock(readers),
 		DownsamplerEnabled: c.downsamplerEnabled,
 		Logger:             c.logger,
 	})

pkg/compactor/split_merge_grouper.go

@@ -124,6 +124,10 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, shardCo
 		// Sort blocks by min time.
 		sortMetasByMinTime(mainBlocks)
 
+		// Remember where this group's jobs start, so the premature-compaction
+		// filter below can be applied to them alone.
+		groupStart := len(jobs)
+
 		for _, tr := range ranges {
 		nextJob:
 			for _, job := range planCompactionByRange(userID, mainBlocks, tr, tr == ranges[0], shardCount, splitGroups) {
@@ -140,31 +144,36 @@ func planCompaction(userID string, blocks []*block.Meta, ranges []int64, shardCo
 				jobs = append(jobs, job)
 			}
 		}
-	}
-
-	// Ensure we don't compact the most recent blocks prematurely when another one of
-	// the same size still fits in the range. To do it, we consider a job valid only
-	// if its range is before the most recent block or if it fully covers the range.
-	highestMaxTime := getMaxTime(blocks)
 
-	for idx := 0; idx < len(jobs); {
-		job := jobs[idx]
+		// Ensure we don't compact the most recent blocks prematurely when another one of
+		// the same size still fits in the range. To do it, we consider a job valid only
+		// if its range is before the most recent block or if it fully covers the range.
+		//
+		// The reference point is the most recent block *of this group*: groups are
+		// independent streams of data, often written by unrelated services on
+		// unrelated schedules, so taking the maximum across all of them lets the
+		// group with the freshest data gate compaction for every other group.
+		highestMaxTime := getMaxTime(mainBlocks)
+
+		for idx := groupStart; idx < len(jobs); {
+			job := jobs[idx]
+
+			// If the job covers a range before the most recent block, it's fine.
+			if job.rangeEnd <= highestMaxTime {
+				idx++
+				continue
+			}
 
-		// If the job covers a range before the most recent block, it's fine.
-		if job.rangeEnd <= highestMaxTime {
-			idx++
-			continue
-		}
+			// If the job covers the full range, it's fine.
+			if job.maxTime()-job.minTime() == job.rangeLength() {
+				idx++
+				continue
+			}
 
-		// If the job covers the full range, it's fine.
-		if job.maxTime()-job.minTime() == job.rangeLength() {
-			idx++
-			continue
+			// We have found a job which would compact recent blocks prematurely,
+			// so we need to filter it out.
+			jobs = append(jobs[:idx], jobs[idx+1:]...)
 		}
-
-		// We have found a job which would compact recent blocks prematurely,
-		// so we need to filter it out.
-		jobs = append(jobs[:idx], jobs[idx+1:]...)
 	}
 
 	// Jobs will be sorted later using configured job sorting algorithm.
@@ -200,8 +209,17 @@ func planCompactionByRange(userID string, blocks []*block.Meta, tr int64, isSmal
 		// (or we're not processing the smallest time range, or splitting is disabled).
 		// Then, we can check if there's any group of blocks to be merged together for each shard.
 		for shardID, shardBlocks := range groupBlocksByShardID(group.blocks) {
-			// No merging to do if there are less than 2 blocks.
-			if len(shardBlocks) < 2 {
+			// Normally there is no merging to do if there are less than 2 blocks:
+			// a block that has no sibling in its shard is already as compacted as
+			// it can get, and re-compacting it on its own would never terminate.
+			//
+			// The exception is a lone block that is still an intermediate
+			// compaction artefact, i.e. below the level at which data is
+			// deduplicated. Such a block is not directly queryable (the querier
+			// prefers its ancestors, which may already have been deleted), so it
+			// must be merged on its own to be promoted past the deduplication
+			// level. This terminates because the merge output is one level higher.
+			if len(shardBlocks) < 2 && !isOrphanedIntermediateBlock(shardCount, shardID, shardBlocks) {
 				continue
 			}
 
@@ -221,6 +239,32 @@ func planCompactionByRange(userID string, blocks []*block.Meta, tr int64, isSmal
 	return jobs
 }
 
+// isOrphanedIntermediateBlock reports whether the given single-block shard group
+// holds a sharded block that is stuck below the deduplication level, and so has
+// to be merged on its own to become queryable.
+func isOrphanedIntermediateBlock(shardCount uint32, shardID string, shardBlocks []*block.Meta) bool {
+	if shardCount == 0 || shardID == "" || len(shardBlocks) != 1 {
+		return false
+	}
+	return isIntermediateShardedBlock(shardBlocks[0])
+}
+
+// isIntermediateShardedBlock reports whether the block is the sharded output of
+// the split stage that has not been merged past the deduplication level yet.
+//
+// Such a block is only queryable at the cost of deduplicating it against its
+// siblings, and only for as long as its ancestors have not been deleted, so the
+// compactor has to promote it even though it has no sibling to merge with.
+func isIntermediateShardedBlock(meta *block.Meta) bool {
+	if meta == nil || meta.Labels[sharding.CompactorShardIDLabel] == "" {
+		return false
+	}
+	// A sharded block is always the output of at least the split stage, so a zero
+	// level means the metadata is missing rather than that the block is new.
+	compactionLevel := int32(meta.Compaction.Level)
+	return compactionLevel > 0 && block.IsIntermediate(compactionLevel, true)
+}
+
 // planSplitting returns a job to split the blocks in the input group or nil if there's nothing to do because
 // all blocks in the group have already been split.
 func planSplitting(userID string, group blocksGroup, splitGroups uint32) []*job {

pkg/compactor/split_merge_grouper_test.go

@@ -509,6 +509,67 @@ func TestPlanCompaction(t *testing.T) {
 				}},
 			},
 		},
+		"should merge a lone sharded block still below the deduplication level, so that it can be promoted": {
+			ranges:     []int64{20, 40},
+			shardCount: 2,
+			blocks: []*block.Meta{
+				{ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}, Compaction: block.BlockMetaCompaction{Level: 2}},
+			},
+			expected: []*job{
+				{userID: userID, stage: stageMerge, shardID: "1_of_2", blocksGroup: blocksGroup{
+					rangeStart: 0,
+					rangeEnd:   20,
+					blocks: []*block.Meta{
+						{ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}, Compaction: block.BlockMetaCompaction{Level: 2}},
+					},
+				}},
+			},
+		},
+		"should not merge a lone sharded block that already reached the deduplication level": {
+			ranges:     []int64{20, 40},
+			shardCount: 2,
+			blocks: []*block.Meta{
+				{ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}, Compaction: block.BlockMetaCompaction{Level: 3}},
+			},
+			expected: nil,
+		},
+		"should not merge a lone sharded block with no compaction metadata": {
+			ranges:     []int64{20, 40},
+			shardCount: 2,
+			blocks: []*block.Meta{
+				{ULID: block1, MinTime: 0, MaxTime: 20, Labels: map[string]string{sharding.CompactorShardIDLabel: "1_of_2"}},
+			},
+			expected: nil,
+		},
+		"should not compact a group's most recent blocks prematurely just because another group has fresher blocks": {
+			ranges: []int64{10, 20},
+			blocks: []*block.Meta{
+				// Group "a" has not reached the end of the [20, 30) range yet.
+				{ULID: block1, MinTime: 20, MaxTime: 25, Labels: map[string]string{"group": "a"}},
+				{ULID: block2, MinTime: 25, MaxTime: 28, Labels: map[string]string{"group": "a"}},
+				// Group "b" is further ahead in time, but that says nothing about group "a".
+				{ULID: block3, MinTime: 40, MaxTime: 50, Labels: map[string]string{"group": "b"}},
+			},
+			expected: nil,
+		},
+		"should compact a group whose range is closed, regardless of other groups": {
+			ranges: []int64{10, 20},
+			blocks: []*block.Meta{
+				{ULID: block1, MinTime: 20, MaxTime: 25, Labels: map[string]string{"group": "a"}},
+				{ULID: block2, MinTime: 25, MaxTime: 30, Labels: map[string]string{"group": "a"}},
+				{ULID: block3, MinTime: 40, MaxTime: 50, Labels: map[string]string{"group": "b"}},
+			},
+			expected: []*job{
+				{userID: userID, stage: stageMerge, blocksGroup: blocksGroup{
+					rangeStart: 20,
+					rangeEnd:   30,
+					blocks: []*block.Meta{
+						{ULID: block1, MinTime: 20, MaxTime: 25, Labels: map[string]string{"group": "a"}},
+						{ULID: block2, MinTime: 25, MaxTime: 30, Labels: map[string]string{"group": "a"}},
+					},
+				}},
+			},
+		},
 	}
 
 	for testName, testData := range tests {

pkg/phlaredb/block/dedup.go

@@ -0,0 +1,38 @@
+package block
+
+// deduplicationLevelSharded and deduplicationLevelUnsharded are the compaction
+// levels at or above which a block's contents are fully deduplicated.
+//
+// With split-and-merge sharding enabled, level 2 is the intermediate split
+// stage: each group is split into split_shards parts which are not yet
+// deduplicated, so the first authoritative level is 3. Without sharding,
+// compaction deduplicates in a single step and level 2 is authoritative.
+const (
+	deduplicationLevelUnsharded int32 = 2
+	deduplicationLevelSharded   int32 = 3
+)
+
+// DeduplicationLevel returns the compaction level at or above which a block is
+// authoritative for its time range: its contents are deduplicated and it fully
+// replaces its ancestors.
+//
+// Both the compactor and the querier must agree on this value. The querier uses
+// it to decide whether a block can be served as-is or has to be merged with
+// deduplication; the compactor uses it to avoid leaving blocks in a state the
+// querier will not serve.
+func DeduplicationLevel(sharded bool) int32 {
+	if sharded {
+		return deduplicationLevelSharded
+	}
+	return deduplicationLevelUnsharded
+}
+
+// IsIntermediate reports whether a block at the given compaction level is an
+// intermediate compaction artefact rather than an authoritative block, i.e. it
+// sits below the deduplication level.
+//
+// Intermediate blocks are still readable, but they may overlap with other
+// blocks, so a query that includes one must deduplicate.
+func IsIntermediate(level int32, sharded bool) bool {
+	return level < DeduplicationLevel(sharded)
+}

pkg/phlaredb/compact.go

@@ -59,6 +59,12 @@ type CompactWithSplittingOpts struct {
 	SplitBy            SplitByFunc
 	DownsamplerEnabled bool
 	Logger             log.Logger
+
+	// AllowSingleBlock permits compacting a single source block without
+	// splitting it. That is normally a no-op and therefore rejected, but the
+	// compactor uses it to promote a block to the next compaction level when it
+	// has no sibling to merge with.
+	AllowSingleBlock bool
 }
 
 func Compact(ctx context.Context, src []BlockReader, dst string) (meta block.Meta, err error) {
@@ -80,7 +86,7 @@ func Compact(ctx context.Context, src []BlockReader, dst string) (meta block.Met
 func CompactWithSplitting(ctx context.Context, opts CompactWithSplittingOpts) (
 	[]block.Meta, error,
 ) {
-	if len(opts.Src) <= 1 && opts.SplitCount == 1 {
+	if len(opts.Src) == 0 || (len(opts.Src) == 1 && opts.SplitCount == 1 && !opts.AllowSingleBlock) {
 		return nil, errors.New("not enough blocks to compact")
 	}
 	if opts.SplitCount == 0 {

pkg/phlaredb/compact_test.go

@@ -191,6 +191,57 @@ func TestCompactWithDownsampling(t *testing.T) {
 	assert.True(t, querier.metrics.profileTableAccess.DeleteLabelValues("profiles.parquet"))
 }
 
+// Compacting a single block on its own is normally a no-op, but the compactor
+// relies on it to promote a block that has no sibling to merge with to the next
+// compaction level.
+func TestCompactWithSplitting_SingleBlock(t *testing.T) {
+	ctx := context.Background()
+
+	b := newBlock(t, func() []*testhelper.ProfileBuilder {
+		return profileSeriesGenerator(t, time.Unix(1, 0), time.Unix(10, 0), time.Second, "job", "a")
+	})
+
+	t.Run("rejected by default", func(t *testing.T) {
+		_, err := CompactWithSplitting(ctx, CompactWithSplittingOpts{
+			Src:        []BlockReader{b},
+			Dst:        t.TempDir(),
+			SplitCount: 1,
+			SplitBy:    SplitByFingerprint,
+			Logger:     log.NewNopLogger(),
+		})
+		require.ErrorContains(t, err, "not enough blocks to compact")
+	})
+
+	t.Run("allowed explicitly", func(t *testing.T) {
+		compacted, err := CompactWithSplitting(ctx, CompactWithSplittingOpts{
+			Src:              []BlockReader{b},
+			Dst:              t.TempDir(),
+			SplitCount:       1,
+			SplitBy:          SplitByFingerprint,
+			AllowSingleBlock: true,
+			Logger:           log.NewNopLogger(),
+		})
+		require.NoError(t, err)
+		require.Len(t, compacted, 1)
+		require.Equal(t, b.Meta().MinTime, compacted[0].MinTime)
+		require.Equal(t, b.Meta().MaxTime, compacted[0].MaxTime)
+		// The point of the promotion: the output sits one level higher.
+		require.Equal(t, b.Meta().Compaction.Level+1, compacted[0].Compaction.Level)
+	})
+
+	t.Run("no source blocks is always rejected", func(t *testing.T) {
+		_, err := CompactWithSplitting(ctx, CompactWithSplittingOpts{
+			Src:              nil,
+			Dst:              t.TempDir(),
+			SplitCount:       1,
+			SplitBy:          SplitByFingerprint,
+			AllowSingleBlock: true,
+			Logger:           log.NewNopLogger(),
+		})
+		require.ErrorContains(t, err, "not enough blocks to compact")
+	})
+}
+
 func TestCompactWithSplitting(t *testing.T) {
 	ctx := context.Background()