db: add high-priority blob file rewrite compactions

This commit introduces a new trigger for blob-file rewrite compactions.
Previously blob-file rewrite compactions were triggered when
  a) no default score-based compactions were available to run
  b) no tombstone density or elision-only compaction were available to run
  c) total unreferenced blob values exceeded the 'garbage ratio' threshold
  d) a blob file meets the criteria for rewrite (including the minimum age)

We've observed that score-based compactions can starve blob-file rewrite
compactions, leaving blob-file—induced space amplification unbounded. This
commit renames the old 'garbage ratio' threshold to GarbageRatioLowPriority and
introduces a new GarbageRatioHighPriority setting. When the ratio of
unreferenced values reaches the high priority setting, Pebble will schedule up
to 1 blob file rewrite compaction at a time as long as eligible blob files
exist. To avoid reducing the concurrency available for default compactions,
compaction concurrency is temporarily inflated for the duration of the blob
file rewrite compaction.

Informs #112.

Author: jbowens

blob_rewrite.go

@@ -7,6 +7,7 @@ package pebble
 import (
 	"container/heap"
 	"context"
+	"fmt"
 	"iter"
 	"runtime/pprof"
 	"slices"
@@ -28,6 +29,12 @@ import (
 // A pickedBlobFileCompaction is a blob file rewrite compaction that has been
 // picked by the compaction picker.
 type pickedBlobFileCompaction struct {
+	// highPriority is set to true if the compaction was picked because the
+	// ValueSeparationPolcy.GarbageThresholdHighPriority heuristic was
+	// triggered. In this case, the resulting compaction will be permitted to
+	// use a burst compaction concurrency slot to avoid starving default
+	// compactions.
+	highPriority      bool
 	vers              *manifest.Version
 	file              manifest.BlobFileMetadata
 	referencingTables []*manifest.TableMetadata
@@ -64,6 +71,7 @@ func (c *pickedBlobFileCompaction) ConstructCompaction(
 			PreferSharedStorage: false,
 			WriteCategory:       getDiskWriteCategoryForCompaction(d.opts, compactionKindBlobFileRewrite),
 		},
+		highPriority: c.highPriority,
 	}
 }
 
@@ -98,13 +106,31 @@ type blobFileRewriteCompaction struct {
 	objCreateOpts         objstorage.CreateOptions
 	internalIteratorStats base.InternalIteratorStats
 	bytesWritten          atomic.Int64 // Total bytes written to the new blob file.
+	// highPriority is set to true if the compaction was picked because the
+	// ValueSeparationPolcy.GarbageThresholdHighPriority heuristic was
+	// triggered. In this case, the resulting compaction will be permitted to
+	// use a burst compaction concurrency slot to avoid starving default
+	// compactions. This field is set when the compaction is created.
+	highPriority bool
+}
+
+func (c *blobFileRewriteCompaction) String() string {
+	s := fmt.Sprintf("blob file (ID: %s) %s (%s) being rewritten",
+		c.input.FileID, c.input.Physical.FileNum, humanizeBytes(uint64(c.input.Physical.Size)))
+	if c.highPriority {
+		s += " (high priority)"
+	}
+	return s
 }
 
 // Assert that *blobFileRewriteCompaction implements the Compaction interface.
 var _ compaction = (*blobFileRewriteCompaction)(nil)
 
 func (c *blobFileRewriteCompaction) AddInProgressLocked(d *DB) {
 	d.mu.compact.inProgress[c] = struct{}{}
+	if c.highPriority {
+		d.mu.compact.burstConcurrency.Add(1)
+	}
 	// TODO(jackson): Currently the compaction picker iterates through all
 	// ongoing compactions in order to limit the number of concurrent blob
 	// rewrite compactions to 1.
@@ -249,6 +275,10 @@ func (c *blobFileRewriteCompaction) Info() compactionInfo {
 	}
 }
 
+func (c *blobFileRewriteCompaction) UsesBurstConcurrency() bool {
+	return c.highPriority
+}
+
 func (c *blobFileRewriteCompaction) RecordError(*problemspans.ByLevel, error) {
 	// TODO(jackson): Track problematic blob files and avoid re-picking the same
 	// blob file compaction.

cmd/pebble/db.go

@@ -81,11 +81,12 @@ func newPebbleDB(dir string) DB {
 	// value separation.
 	opts.Experimental.ValueSeparationPolicy = func() pebble.ValueSeparationPolicy {
 		return pebble.ValueSeparationPolicy{
-			Enabled:               true,
-			MinimumSize:           512,
-			MaxBlobReferenceDepth: 10,
-			RewriteMinimumAge:     5 * time.Minute,
-			TargetGarbageRatio:    0.1,
+			Enabled:                  true,
+			MinimumSize:              512,
+			MaxBlobReferenceDepth:    10,
+			RewriteMinimumAge:        5 * time.Minute,
+			GarbageRatioLowPriority:  0.10, // 10% garbage
+			GarbageRatioHighPriority: 0.30, // 30% garbage
 		}
 	}
 

compaction.go

@@ -207,6 +207,7 @@ type compaction interface {
 	PprofLabels(UserKeyCategories) pprof.LabelSet
 	RecordError(*problemspans.ByLevel, error)
 	Tables() iter.Seq2[int, *manifest.TableMetadata]
+	UsesBurstConcurrency() bool
 	VersionEditApplied() bool
 }
 
@@ -343,6 +344,9 @@ var _ compaction = (*tableCompaction)(nil)
 
 func (c *tableCompaction) AddInProgressLocked(d *DB) {
 	d.mu.compact.inProgress[c] = struct{}{}
+	if c.UsesBurstConcurrency() {
+		d.mu.compact.burstConcurrency.Add(1)
+	}
 	var isBase, isIntraL0 bool
 	for _, cl := range c.inputs {
 		for f := range cl.files.All() {
@@ -457,6 +461,8 @@ func (c *tableCompaction) Tables() iter.Seq2[int, *manifest.TableMetadata] {
 	}
 }
 
+func (c *tableCompaction) UsesBurstConcurrency() bool { return false }
+
 func (c *tableCompaction) VersionEditApplied() bool { return c.versionEditApplied }
 
 // compactionMetrics contians metrics surrounding a compaction.
@@ -1267,10 +1273,9 @@ func (c *tableCompaction) String() string {
 	}
 
 	var buf bytes.Buffer
-	for level := c.startLevel.level; level <= c.outputLevel.level; level++ {
-		i := level - c.startLevel.level
-		fmt.Fprintf(&buf, "%d:", level)
-		for f := range c.inputs[i].files.All() {
+	for _, l := range c.inputs {
+		fmt.Fprintf(&buf, "%d:", l.level)
+		for f := range l.files.All() {
 			fmt.Fprintf(&buf, " %s:%s-%s", f.TableNum, f.Smallest(), f.Largest())
 		}
 		fmt.Fprintf(&buf, "\n")
@@ -1832,6 +1837,11 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	}
 
 	d.clearCompactingState(c, err != nil)
+	if c.UsesBurstConcurrency() {
+		if v := d.mu.compact.burstConcurrency.Add(-1); v < 0 {
+			panic(errors.AssertionFailedf("burst concurrency underflow: %d", v))
+		}
+	}
 	delete(d.mu.compact.inProgress, c)
 	d.mu.versions.incrementCompactions(c.kind, c.extraLevels, c.metrics.bytesWritten.Load(), err)
 
@@ -2052,8 +2062,16 @@ func (d *DB) makeCompactionEnvLocked() *compactionEnv {
 
 // pickAnyCompaction tries to pick a manual or automatic compaction.
 func (d *DB) pickAnyCompaction(env compactionEnv) (pc pickedCompaction) {
-	// Pick a score-based compaction first, since a misshapen LSM is bad.
 	if !d.opts.DisableAutomaticCompactions {
+		// Pick a score-based compaction first, since a misshapen LSM is bad.
+		// We allow an exception for a high-priority disk-space reclamation
+		// compaction. Future work will explore balancing the various competing
+		// compaction priorities more judiciously. For now, we're relying on the
+		// configured heuristic to be set carefully so that we don't starve
+		// score-based compactions.
+		if pc := d.mu.versions.picker.pickHighPrioritySpaceCompaction(env); pc != nil {
+			return pc
+		}
 		if pc = d.mu.versions.picker.pickAutoScore(env); pc != nil {
 			return pc
 		}
@@ -2160,13 +2178,15 @@ func (d *DB) GetWaitingCompaction() (bool, WaitingCompaction) {
 // CompactionScheduler).
 func (d *DB) GetAllowedWithoutPermission() int {
 	allowedBasedOnBacklog := int(d.mu.versions.curCompactionConcurrency.Load())
-	allowedBasedOnManual := 0
-	manualBacklog := int(d.mu.compact.manualLen.Load())
-	if manualBacklog > 0 {
-		_, maxAllowed := d.opts.CompactionConcurrencyRange()
-		allowedBasedOnManual = min(maxAllowed, manualBacklog+allowedBasedOnBacklog)
+	allowedBasedOnManual := int(d.mu.compact.manualLen.Load())
+	allowedBasedOnSpaceHeuristic := int(d.mu.compact.burstConcurrency.Load())
+
+	v := allowedBasedOnBacklog + allowedBasedOnManual + allowedBasedOnSpaceHeuristic
+	if v == allowedBasedOnBacklog {
+		return v
 	}
-	return max(allowedBasedOnBacklog, allowedBasedOnManual)
+	_, maxAllowed := d.opts.CompactionConcurrencyRange()
+	return min(v, maxAllowed)
 }
 
 // tryScheduleDownloadCompactions tries to start download compactions.
@@ -2588,6 +2608,11 @@ func (d *DB) compact(c compaction, errChannel chan error) {
 			} else {
 				d.mu.compact.compactingCount--
 			}
+			if c.UsesBurstConcurrency() {
+				if v := d.mu.compact.burstConcurrency.Add(-1); v < 0 {
+					panic(errors.AssertionFailedf("burst concurrency underflow: %d", v))
+				}
+			}
 			delete(d.mu.compact.inProgress, c)
 			// Add this compaction's duration to the cumulative duration. NB: This
 			// must be atomic with the above removal of c from

compaction_picker.go

@@ -60,6 +60,7 @@ type compactionPicker interface {
 	getMetrics([]compactionInfo) compactionPickerMetrics
 	getBaseLevel() int
 	estimatedCompactionDebt() uint64
+	pickHighPrioritySpaceCompaction(env compactionEnv) pickedCompaction
 	pickAutoScore(env compactionEnv) (pc pickedCompaction)
 	pickAutoNonScore(env compactionEnv) (pc pickedCompaction)
 	forceBaseLevel1()
@@ -1406,6 +1407,22 @@ func (p *compactionPickerByScore) logCompactionForTesting(
 		pc.startLevel.level, pc.outputLevel.level, buf.String())
 }
 
+// pickHighPrioritySpaceCompaction checks for a high-priority space reclamation
+// compaction. Under some circumstances, we want to persue a compaction for the
+// purpose of reclaiming disk space even when there are eligible default
+// compactions.
+func (p *compactionPickerByScore) pickHighPrioritySpaceCompaction(
+	env compactionEnv,
+) pickedCompaction {
+	if pc := p.pickBlobFileRewriteCompactionHighPriority(env); pc != nil {
+		return pc
+	}
+	// NB: We can't just return the above result because the above func returns
+	// a *pickedBlobFileCompaction, not a pickedCompaction. We need to return an
+	// untyped nil.
+	return nil
+}
+
 // pickAutoScore picks the best score-based compaction, if any.
 //
 // On each call, pickAutoScore computes per-level size adjustments based on
@@ -1482,7 +1499,7 @@ func (p *compactionPickerByScore) pickAutoNonScore(env compactionEnv) (pc picked
 
 	// Check for blob file rewrites. These are low-priority compactions because
 	// they don't help us keep up with writes, just reclaim disk space.
-	if pc := p.pickBlobFileRewriteCompaction(env); pc != nil {
+	if pc := p.pickBlobFileRewriteCompactionLowPriority(env); pc != nil {
 		return pc
 	}
 
@@ -1675,9 +1692,40 @@ func (p *compactionPickerByScore) pickRewriteCompaction(
 	return nil
 }
 
-// pickBlobFileRewriteCompaction looks for compactions of blob files that
-// can be rewritten to reclaim disk space.
-func (p *compactionPickerByScore) pickBlobFileRewriteCompaction(
+// pickBlobFileRewriteCompactionHighPriority picks a compaction that rewrites a
+// blob file to reclaim disk space if the heuristics for high-priority blob file
+// rewrites are met.
+func (p *compactionPickerByScore) pickBlobFileRewriteCompactionHighPriority(
+	env compactionEnv,
+) (pc *pickedBlobFileCompaction) {
+	policy := p.opts.Experimental.ValueSeparationPolicy()
+	if policy.GarbageRatioHighPriority >= 1.0 {
+		// High-priority blob file rewrite compactions are disabled.
+		return nil
+	}
+	aggregateStats, heuristicStats := p.latestVersionState.blobFiles.Stats()
+	if heuristicStats.CountFilesEligible == 0 && heuristicStats.CountFilesTooRecent == 0 {
+		// No blob files with any garbage to rewrite.
+		return nil
+	}
+
+	garbagePct := float64(aggregateStats.ValueSize-aggregateStats.ReferencedValueSize) /
+		float64(aggregateStats.ValueSize)
+	if garbagePct <= policy.GarbageRatioHighPriority {
+		// Not enough garbage to warrant a rewrite compaction.
+		return nil
+	}
+	pc = p.pickBlobFileRewriteCandidate(env)
+	if pc != nil {
+		pc.highPriority = true
+	}
+	return pc
+}
+
+// pickBlobFileRewriteCompactionLowPriority picks a compaction that rewrites a
+// blob file to reclaim disk space if the heuristics for low-priority blob file
+// rewrites are met.
+func (p *compactionPickerByScore) pickBlobFileRewriteCompactionLowPriority(
 	env compactionEnv,
 ) (pc *pickedBlobFileCompaction) {
 	aggregateStats, heuristicStats := p.latestVersionState.blobFiles.Stats()
@@ -1686,7 +1734,7 @@ func (p *compactionPickerByScore) pickBlobFileRewriteCompaction(
 		return nil
 	}
 	policy := p.opts.Experimental.ValueSeparationPolicy()
-	if policy.TargetGarbageRatio >= 1.0 {
+	if policy.GarbageRatioLowPriority >= 1.0 {
 		// Blob file rewrite compactions are disabled.
 		return nil
 	}
@@ -1696,19 +1744,23 @@ func (p *compactionPickerByScore) pickBlobFileRewriteCompaction(
 	// are actually any candidates with garbage to reclaim.
 	garbagePct := float64(aggregateStats.ValueSize-aggregateStats.ReferencedValueSize) /
 		float64(aggregateStats.ValueSize)
-	if garbagePct <= policy.TargetGarbageRatio {
+	if garbagePct <= policy.GarbageRatioLowPriority {
 		// Not enough garbage to warrant a rewrite compaction.
 		return nil
 	}
+	return p.pickBlobFileRewriteCandidate(env)
+}
 
+func (p *compactionPickerByScore) pickBlobFileRewriteCandidate(
+	env compactionEnv,
+) (pc *pickedBlobFileCompaction) {
 	// Check if there is an ongoing blob file rewrite compaction. If there is,
 	// don't schedule a new one.
 	for _, c := range env.inProgressCompactions {
 		if c.kind == compactionKindBlobFileRewrite {
 			return nil
 		}
 	}
-
 	candidate, ok := p.latestVersionState.blobFiles.ReplacementCandidate()
 	if !ok {
 		// None meet the heuristic.

compaction_picker_test.go

@@ -6,8 +6,11 @@ package pebble
 
 import (
 	"bytes"
+	"cmp"
 	"fmt"
+	"maps"
 	"math"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -1486,6 +1489,17 @@ func TestCompactionPickerScores(t *testing.T) {
 	var buf bytes.Buffer
 	datadriven.RunTest(t, "testdata/compaction_picker_scores", func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
+		case "batch":
+			b := d.NewBatch()
+			err := runBatchDefineCmd(td, b)
+			if err != nil {
+				return err.Error()
+			}
+			if err = b.Commit(Sync); err != nil {
+				return err.Error()
+			}
+			return ""
+
 		case "define":
 			require.NoError(t, closeAllSnapshots(d))
 			require.NoError(t, d.Close())
@@ -1518,6 +1532,12 @@ func TestCompactionPickerScores(t *testing.T) {
 			d.mu.Unlock()
 			return ""
 
+		case "flush":
+			if err := d.Flush(); err != nil {
+				return err.Error()
+			}
+			return runLSMCmd(td, d)
+
 		case "resume-cleaning":
 			cleaner.resume()
 			return ""
@@ -1542,8 +1562,15 @@ func TestCompactionPickerScores(t *testing.T) {
 			d.mu.Lock()
 			d.opts.DisableAutomaticCompactions = false
 			d.maybeScheduleCompaction()
+			if v := d.mu.compact.burstConcurrency.Load(); v > 0 {
+				fmt.Fprintf(&buf, "%d burst concurrency active\n", v)
+			}
 			fmt.Fprintf(&buf, "%d compactions in progress:", d.mu.compact.compactingCount)
-			for c := range d.mu.compact.inProgress {
+
+			runningCompactions := slices.SortedFunc(maps.Keys(d.mu.compact.inProgress), func(a, b compaction) int {
+				return cmp.Compare(a.Info().String(), b.Info().String())
+			})
+			for _, c := range runningCompactions {
 				fmt.Fprintf(&buf, "\n%s", c)
 			}
 			d.opts.DisableAutomaticCompactions = true
@@ -1596,6 +1623,14 @@ func TestCompactionPickerScores(t *testing.T) {
 			tw.Flush()
 			return buf.String()
 
+		case "wait-compactions":
+			d.mu.Lock()
+			for d.mu.compact.compactingCount > 0 {
+				d.mu.compact.cond.Wait()
+			}
+			d.mu.Unlock()
+			return ""
+
 		case "wait-pending-table-stats":
 			return runWaitForTableStatsCmd(td, d)