From e7c60041473e72c24ae6ad0cc1968a962a10bc9d Mon Sep 17 00:00:00 2001 From: Dimitry Linkov Date: Wed, 2 Sep 2026 20:24:28 -0700 Subject: [PATCH 1/3] Stop losing data when split-and-merge compaction stalls mid-way Split-and-merge compaction can leave a group's sharded split output at level 2 forever: the merge stage only runs when a shard has two or more blocks, so a range whose split produced a single block per shard never gets promoted. That is not a rare corner - it happens whenever ingestion is scheduled rather than continuous, because the "don't compact the most recent blocks prematurely" guard rejects the merge job for the last open range and no later block ever arrives to un-gate it. Once the level 1 ancestors are deleted, those level 2 blocks are the only copy of the data, and the querier dropped them unconditionally: sharded blocks below the deduplication level were treated as always superseded, with no check that anything survived to serve the range. An hour of profiles silently returned nothing. Three changes, all pulling in the same direction: - The querier only prefers a block's ancestors over the intermediate block itself when those ancestors are actually still there. If they are gone, it serves the intermediate block and deduplicates, and warns so the stall is visible. Completeness of the sharded set is re-checked over what survives the collapse, so a partial fallback still yields an empty plan rather than half the data. - The deduplication level (2, or 3 when sharding is enabled) becomes a single shared definition in pkg/phlaredb/block instead of a constant hardcoded on both sides, so the querier and the compactor cannot disagree about which blocks are authoritative. - The compactor plans a merge for a lone sharded block that is still below the deduplication level, promoting it instead of leaving it stuck. This terminates because the output is one level higher, and the general "not enough blocks to compact" guard is relaxed only for that deliberate single-block promotion. The premature-compaction guard now takes its reference point from the most recent block of the group being planned, not of the whole tenant. Groups are independent streams written by unrelated services on unrelated schedules; taking the maximum across all of them let whichever group had the freshest data un-gate compaction for every other group, which is how the half-compacted range above came about in the first place. Co-Authored-By: Claude Opus 5 --- pkg/compactor/bucket_compactor.go | 26 ++- pkg/compactor/split_merge_grouper.go | 90 +++++++--- pkg/compactor/split_merge_grouper_test.go | 61 +++++++ pkg/phlaredb/block/dedup.go | 38 ++++ pkg/phlaredb/compact.go | 8 +- pkg/phlaredb/compact_test.go | 51 ++++++ pkg/querier/replication.go | 122 +++++++++---- pkg/querier/replication_test.go | 205 +++++++++++++++++++++- 8 files changed, 537 insertions(+), 64 deletions(-) create mode 100644 pkg/phlaredb/block/dedup.go diff --git a/pkg/compactor/bucket_compactor.go b/pkg/compactor/bucket_compactor.go index dfbc5aad87..b5cb50f339 100644 --- a/pkg/compactor/bucket_compactor.go +++ b/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, }) diff --git a/pkg/compactor/split_merge_grouper.go b/pkg/compactor/split_merge_grouper.go index 3b483b9900..bf630a14da 100644 --- a/pkg/compactor/split_merge_grouper.go +++ b/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 { diff --git a/pkg/compactor/split_merge_grouper_test.go b/pkg/compactor/split_merge_grouper_test.go index b86b23ab6c..0688fc7779 100644 --- a/pkg/compactor/split_merge_grouper_test.go +++ b/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 { diff --git a/pkg/phlaredb/block/dedup.go b/pkg/phlaredb/block/dedup.go new file mode 100644 index 0000000000..ab329e425c --- /dev/null +++ b/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) +} diff --git a/pkg/phlaredb/compact.go b/pkg/phlaredb/compact.go index 99b67492ba..f02b79874b 100644 --- a/pkg/phlaredb/compact.go +++ b/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 { diff --git a/pkg/phlaredb/compact_test.go b/pkg/phlaredb/compact_test.go index 7f99a43f2f..943ec5c9a9 100644 --- a/pkg/phlaredb/compact_test.go +++ b/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() diff --git a/pkg/querier/replication.go b/pkg/querier/replication.go index 41a7689d2d..c08c3881bc 100644 --- a/pkg/querier/replication.go +++ b/pkg/querier/replication.go @@ -17,6 +17,7 @@ import ( ingestv1 "github.com/grafana/pyroscope/api/gen/proto/go/ingester/v1" typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1" + "github.com/grafana/pyroscope/v2/pkg/phlaredb/block" "github.com/grafana/pyroscope/v2/pkg/phlaredb/sharding" "github.com/grafana/pyroscope/v2/pkg/util" "github.com/grafana/pyroscope/v2/pkg/util/spanlogger" @@ -216,10 +217,16 @@ func (r *replicasPerBlockID) hasShardedBlocks() bool { } // pruneIncompleteShardedBlocks drops sharded blocks for any window instant that -// is missing a shard. It must run before pruneSupersededBlocks, so an incomplete -// set's lower-level ancestors survive as a fallback. deduplicationLevel is the -// level at/above which blocks are deduplicated (3 when sharded). -func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int32) error { +// is missing a shard. Only blocks at or above minLevel are counted towards +// completeness. +// +// It runs twice. First with minLevel = deduplicationLevel, before +// pruneSupersededBlocks, so that an incomplete set of deduplicated blocks is +// pruned while its lower-level ancestors are still around to serve as a +// fallback. Then again with minLevel = 0, after pruneSupersededBlocks, to check +// that whatever survived the collapse - which may include intermediate blocks +// kept as a last resort - still covers every shard. +func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(minLevel int32) error { // Completeness is checked per sharding (shard count) and per time instant, not // by compaction level: a window's shards can legitimately sit at different // levels (a partial late re-merge advances only some), so keying on level would @@ -230,9 +237,11 @@ func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int // covering that instant. Coverage (minTime <= t < maxTime), rather than an exact // minTime match, keeps a wider block (a shard merged to a longer span) counted // for the later windows it overlaps, so sibling shards' blocks there are not - // orphaned and silently pruned. Only blocks >= deduplicationLevel count; - // intermediate lower blocks are never served (pruneSupersededBlocks drops them) - // so must not satisfy a shard. + // orphaned and silently pruned. Only blocks >= minLevel count: + // intermediate lower blocks are preferentially dropped by + // pruneSupersededBlocks, so they must not satisfy a shard here. They are also + // never pruned here, which is what lets pruneSupersededBlocks keep them as a + // last resort when their ancestors are gone. type shardedBlock struct { id string shard uint64 @@ -249,7 +258,7 @@ func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int if !ok { continue } - if meta.Compaction == nil || meta.Compaction.Level < deduplicationLevel { + if meta.Compaction == nil || meta.Compaction.Level < minLevel { continue } byShardCount[shardCount] = append(byShardCount[shardCount], @@ -320,37 +329,83 @@ func (r *replicasPerBlockID) pruneIncompleteShardedBlocks(deduplicationLevel int return nil } -// prunes blocks that are contained by a higher compaction level block -func (r *replicasPerBlockID) pruneSupersededBlocks(sharded bool) error { +// pruneSupersededBlocks removes blocks whose data is fully contained in a block +// at or above the deduplication level, and collapses intermediate compaction +// artefacts when a complete lower-level fallback is still available. +// +// deduplicationLevel is the level at/above which a block is authoritative for +// its range (see block.DeduplicationLevel). +func (r *replicasPerBlockID) pruneSupersededBlocks(deduplicationLevel int32) error { + // First pass: authoritative blocks supersede every ancestor they were built + // from, including any intermediate blocks recorded as parents. for blockID := range r.m { meta, ok := r.meta[blockID] if !ok { return fmt.Errorf("meta missing for block id %s", blockID) } - if meta.Compaction == nil { + if meta.Compaction == nil || meta.Compaction.Level < deduplicationLevel { continue } - if meta.Compaction.Level < 2 { + for _, ancestor := range meta.Compaction.Parents { + r.removeBlock(ancestor) + } + for _, ancestor := range meta.Compaction.Sources { + r.removeBlock(ancestor) + } + } + + // Second pass: intermediate blocks (compacted at least once, but still below + // the deduplication level – the split stage of split-and-merge compaction). + // Reading them is more expensive than reading their ancestors: the split + // stage fans a group out into shard_count blocks, which is typically + // _significantly_ more blocks than it consumed, and they are not yet + // deduplicated either way. So we prefer the ancestors – but only if they are + // all still there. If any of them is gone, the intermediate block is the only + // remaining copy of that data and dropping it loses the range entirely; keep + // it instead and let the query deduplicate. + for blockID := range r.m { + meta, ok := r.meta[blockID] + if !ok { + return fmt.Errorf("meta missing for block id %s", blockID) + } + // Level < 2 means the block has not been compacted yet: it has no + // ancestors to fall back to and is served as-is. + if meta.Compaction == nil || meta.Compaction.Level < 2 || meta.Compaction.Level >= deduplicationLevel { continue } - // At split phase of compaction, L2 is an intermediate step where we - // split each group into split_shards parts, thus there will be up to - // groups_num * split_shards blocks, which is typically _significantly_ - // greater that the number of source blocks. Moreover, these blocks are - // not yet deduplicated, therefore we should prefer L1 blocks over them. - // As an optimisation, we drop all L2 blocks. - if sharded && meta.Compaction.Level == 2 { + if r.ancestorsLive(meta) { r.removeBlock(blockID) continue } - for _, blockID := range meta.Compaction.Parents { - r.removeBlock(blockID) + level.Warn(r.logger).Log( + "msg", "querying intermediate compaction level block: its ancestors are no longer available", + "block", blockID, + "compaction_level", meta.Compaction.Level, + "deduplication_level", deduplicationLevel, + ) + } + + return nil +} + +// ancestorsLive reports whether the block's ancestors are all still present and +// can therefore serve its range in its place. A block with no recorded +// ancestors has no known fallback. +func (r *replicasPerBlockID) ancestorsLive(meta *typesv1.BlockInfo) bool { + if len(meta.Compaction.Sources) == 0 && len(meta.Compaction.Parents) == 0 { + return false + } + for _, ancestor := range meta.Compaction.Sources { + if _, ok := r.m[ancestor]; !ok { + return false } - for _, blockID := range meta.Compaction.Sources { - r.removeBlock(blockID) + } + for _, ancestor := range meta.Compaction.Parents { + if _, ok := r.m[ancestor]; !ok { + return false } } - return nil + return true } type blockPlanEntry struct { @@ -391,12 +446,9 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla sharded := r.hasShardedBlocks() // Depending on whether split sharding is used, the compaction level at - // which the data gets deduplicated differs: if split sharding is enabled, - // we deduplicate at level 3, and at level 2 otherwise. - var deduplicationLevel int32 = 2 - if sharded { - deduplicationLevel = 3 - } + // which the data gets deduplicated differs. The compactor uses the same + // threshold, so the two stay in agreement. + deduplicationLevel := block.DeduplicationLevel(sharded) // Prune incomplete sharded sets first, so that any lower-level ancestors // survive as a fallback, then collapse the surviving blocks per shard. @@ -405,11 +457,19 @@ func (r *replicasPerBlockID) blockPlan(ctx context.Context) map[string]*blockPla return nil } - if err := r.pruneSupersededBlocks(sharded); err != nil { + if err := r.pruneSupersededBlocks(deduplicationLevel); err != nil { level.Warn(r.logger).Log("msg", "block planning failed to prune superseded blocks", "err", err) return nil } + // Re-check completeness over what survived: pruneSupersededBlocks may have + // kept intermediate blocks that the first check did not count, and it may + // have removed ancestors that were covering a shard. + if err := r.pruneIncompleteShardedBlocks(0); err != nil { + level.Warn(r.logger).Log("msg", "block planning failed to prune incomplete sharded blocks", "err", err) + return nil + } + // now we go through all blocks and choose the replicas that we want to query for blockID, replicas := range r.m { // skip if we have no replicas, then block is already contained i an higher compaction level one diff --git a/pkg/querier/replication_test.go b/pkg/querier/replication_test.go index 43c256a4e9..d2e0f38cb3 100644 --- a/pkg/querier/replication_test.go +++ b/pkg/querier/replication_test.go @@ -476,10 +476,11 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { }, { // Mid-merge: shard 0 at L3, shard 1 only in an intermediate L2 block, no - // L1. The L2 block gets dropped by superseding, so it must not count shard - // 1 as present - otherwise we'd serve shard 0 alone (half the data). The - // set is incomplete -> everything pruned (transient empty, not undercount). - name: "do not let an intermediate L2 block satisfy shard completeness", + // L1. The L2 block must not count shard 1 as present at the deduplicated + // level - otherwise we'd serve shard 0 alone (half the data). The L3 block + // is therefore pruned, and the intermediate L2 blocks left behind do cover + // both shards, so they are served with deduplication. + name: "an intermediate L2 block does not satisfy shard completeness at the deduplicated level", inputs: func(r *replicasPerBlockID) { t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ @@ -507,6 +508,37 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { }, }, storeGatewayInstance) }, + validators: []validatorFunc{ + validatePlanBlockIDs("s0-l2", "s1-l2"), + validatePlanDeduplication(true), + }, + }, + { + // Same shape, but shard 0 has no intermediate block left behind: pruning + // the incomplete L3 set leaves shard 1 alone, which would be half the + // data, so everything is pruned (transient empty, not undercount). + name: "prune a surviving intermediate block that does not cover every shard", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0-l3"). + withCompactionLevel(3). + withCompactorShard(0, 2). + withMinTime(t1, time.Hour). + info(), + // shard 1 only exists as an intermediate L2 block. + newBlockInfo("s1-l2"). + withCompactionLevel(2). + withCompactorShard(1, 2). + withMinTime(t1, time.Hour). + info(), + }, + }, + }, storeGatewayInstance) + }, validators: []validatorFunc{ validatePlanBlockIDs(), }, @@ -560,6 +592,121 @@ func Test_replicasPerBlockID_blockPlan(t *testing.T) { validatePlanBlocksOnReplica("ingester-0", "b"), }, }, + { + // The split stage of compaction produces intermediate blocks that are + // preferentially dropped in favour of their ancestors. If the ancestors + // are gone, dropping them loses the range entirely, so they must be + // served (with deduplication) instead. + name: "keep intermediate sharded blocks whose ancestors are gone", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("s1"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(1, 2). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs("s0", "s1"), + validatePlanDeduplication(true), + }, + }, + { + name: "drop intermediate sharded blocks whose ancestors are all still available", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("a").withMinTime(t1, 30*time.Minute).info(), + newBlockInfo("b").withMinTime(t1, time.Hour).info(), + + newBlockInfo("s0"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("s1"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(1, 2). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs("a", "b"), + validatePlanDeduplication(true), + }, + }, + { + // A block at or above the deduplication level supersedes everything it + // was built from, including the intermediate blocks recorded as parents. + name: "drop intermediate sharded blocks superseded by a deduplicated block", + inputs: func(r *replicasPerBlockID) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("s1"). + withMinTime(t1, time.Hour). + withCompactionLevel(2). + withCompactionSources("a", "b"). + withCompactorShard(1, 2). + info(), + + newBlockInfo("m0"). + withMinTime(t1, time.Hour). + withCompactionLevel(3). + withCompactionSources("a", "b"). + withCompactionParents("s0"). + withCompactorShard(0, 2). + info(), + + newBlockInfo("m1"). + withMinTime(t1, time.Hour). + withCompactionLevel(3). + withCompactionSources("a", "b"). + withCompactionParents("s1"). + withCompactorShard(1, 2). + info(), + }, + }, + }, storeGatewayInstance) + }, + validators: []validatorFunc{ + validatePlanBlockIDs("m0", "m1"), + validatePlanDeduplication(false), + }, + }, } { t.Run(tc.name, func(t *testing.T) { r := newReplicasPerBlockID(log.NewNopLogger()) @@ -617,3 +764,53 @@ func Test_pruneIncompleteShardedBlocks_logging(t *testing.T) { require.NotContains(t, buf.String(), warnMsg, "must not warn for a complete window") }) } + +// Serving an intermediate compaction level block is a last resort: it means the +// compactor left the data in a state it never promoted out of, so it must be +// visible in the logs. +func Test_pruneSupersededBlocks_logging(t *testing.T) { + t1, _ := time.Parse(time.RFC3339, "2021-01-01T00:00:00Z") + + const warnMsg = "its ancestors are no longer available" + + t.Run("warns when an intermediate block has to be served", func(t *testing.T) { + var buf bytes.Buffer + r := newReplicasPerBlockID(log.NewLogfmtLogger(&buf)) + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("s0").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(0, 2).info(), + newBlockInfo("s1").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(1, 2).info(), + }, + }, + }, storeGatewayInstance) + + plan := r.blockPlan(context.TODO()) + require.NotEmpty(t, plan) + require.Contains(t, buf.String(), warnMsg) + }) + + t.Run("does not warn when the ancestors are still available", func(t *testing.T) { + var buf bytes.Buffer + r := newReplicasPerBlockID(log.NewLogfmtLogger(&buf)) + r.add([]ResponseFromReplica[[]*typesv1.BlockInfo]{ + { + addr: "store-gateway-0", + response: []*typesv1.BlockInfo{ + newBlockInfo("a").withMinTime(t1, time.Hour).info(), + newBlockInfo("s0").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(0, 2).info(), + newBlockInfo("s1").withMinTime(t1, time.Hour).withCompactionLevel(2). + withCompactionSources("a").withCompactorShard(1, 2).info(), + }, + }, + }, storeGatewayInstance) + + plan := r.blockPlan(context.TODO()) + require.NotEmpty(t, plan) + require.NotContains(t, buf.String(), warnMsg) + }) +} From 944f46db5d930d6bba8ba76cfb44e3526ac43bf0 Mon Sep 17 00:00:00 2001 From: Dimitry Linkov Date: Wed, 2 Sep 2026 20:45:52 -0700 Subject: [PATCH 2/3] Regenerate the reference help for the symbol cache flag blocks-storage.bucket-store.symbol-cache-max-bytes was added without rerunning `make reference-help`, so TestHelp/all fails on every branch. Co-Authored-By: Claude Opus 5 --- cmd/pyroscope/help-all.txt.tmpl | 2 ++ 1 file changed, 2 insertions(+) diff --git a/cmd/pyroscope/help-all.txt.tmpl b/cmd/pyroscope/help-all.txt.tmpl index 5324c26455..0b66795479 100644 --- a/cmd/pyroscope/help-all.txt.tmpl +++ b/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 From 0f6d035a71eff3aba3c99f626d024588ea674e4c Mon Sep 17 00:00:00 2001 From: Dimitry Linkov Date: Wed, 2 Sep 2026 21:11:21 -0700 Subject: [PATCH 3/3] Regenerate the config reference for the symbol cache flag Same omission as the reference help: blocks-storage.bucket-store.symbol-cache-max-bytes was added without rerunning `make generate`, so the check-generated CI job reports a dirty tree on every branch. Co-Authored-By: Claude Opus 5 --- .../reference-configuration-parameters/index.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/docs/sources/configure-server/reference-configuration-parameters/index.md b/docs/sources/configure-server/reference-configuration-parameters/index.md index 192d7e4ed3..40a1c1644e 100644 --- a/docs/sources/configure-server/reference-configuration-parameters/index.md +++ b/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: | 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: | default = 0] ``` ### compactor