options: make TargetByteDeletionRate dynamically configurable

Change this setting to a function. This change will be backported to
older releases, as it provides an important "escape hatch" if delete
pacing goes wrong.

Informs #5424

Author: RaduBerinde

compaction_test.go

@@ -2700,7 +2700,7 @@ func TestSharedObjectDeletePacing(t *testing.T) {
 		"": remote.NewInMem(),
 	})
 	opts.Experimental.CreateOnShared = remote.CreateOnSharedAll
-	opts.TargetByteDeletionRate = 1
+	opts.TargetByteDeletionRate = func() int { return 1 }
 	opts.Logger = testLogger{t}
 
 	d, err := Open("", &opts)

metamorphic/options.go

@@ -663,7 +663,8 @@ func RandomOptions(
 	opts.FormatMajorVersion += pebble.FormatMajorVersion(rng.IntN(n + 1))
 	opts.Experimental.L0CompactionConcurrency = 1 + rng.IntN(4) // 1-4
 	opts.Experimental.LevelMultiplier = 5 << rng.IntN(7)        // 5 - 320
-	opts.TargetByteDeletionRate = 1 << uint(20+rng.IntN(10))    // 1MB - 1GB
+	targetByteDeletionRate := 1 << uint(20+rng.IntN(10))        // 1MB - 1GB
+	opts.TargetByteDeletionRate = func() int { return targetByteDeletionRate }
 	opts.Experimental.ValidateOnIngest = rng.IntN(2) != 0
 	opts.L0CompactionThreshold = 1 + rng.IntN(100)     // 1 - 100
 	opts.L0CompactionFileThreshold = 1 << rng.IntN(11) // 1 - 1024

metamorphic/options_test.go

@@ -73,6 +73,7 @@ func TestOptionsRoundtrip(t *testing.T) {
 		"EventListener:",
 		"CompactionConcurrencyRange:",
 		"MaxConcurrentDownloads:",
+		"TargetByteDeletionRate:",
 		"Experimental.DisableIngestAsFlushable:",
 		"Experimental.EnableColumnarBlocks:",
 		"Experimental.EnableValueBlocks:",
@@ -117,6 +118,7 @@ func TestOptionsRoundtrip(t *testing.T) {
 		if o.Opts.Experimental.IngestSplit != nil && o.Opts.Experimental.IngestSplit() {
 			require.Equal(t, o.Opts.Experimental.IngestSplit(), parsed.Opts.Experimental.IngestSplit())
 		}
+		require.Equal(t, o.Opts.TargetByteDeletionRate(), parsed.Opts.TargetByteDeletionRate())
 
 		expBaseline, expUpper := o.Opts.CompactionConcurrencyRange()
 		parsedBaseline, parsedUpper := parsed.Opts.CompactionConcurrencyRange()

obsolete_files.go

@@ -91,7 +91,7 @@ func openCleanupManager(
 		deletePacer: newDeletionPacer(
 			crtime.NowMono(),
 			opts.FreeSpaceThresholdBytes,
-			int64(opts.TargetByteDeletionRate),
+			opts.TargetByteDeletionRate,
 			opts.FreeSpaceTimeframe,
 			opts.ObsoleteBytesMaxRatio,
 			opts.ObsoleteBytesTimeframe,

options.go

@@ -1056,8 +1056,8 @@ type Options struct {
 	// This value is only a best-effort target; the effective rate can be
 	// higher if deletions are falling behind or disk space is running low.
 	//
-	// Setting this to 0 disables deletion pacing, which is also the default.
-	TargetByteDeletionRate int
+	// A returned value of 0 disables deletion pacing (this is also the default).
+	TargetByteDeletionRate func() int
 
 	// FreeSpaceThresholdBytes specifies the minimum amount of free disk space that Pebble
 	// attempts to maintain. If free disk space drops below this threshold, deletions
@@ -1166,6 +1166,10 @@ func (o *Options) EnsureDefaults() {
 		o.Cleaner = DeleteCleaner{}
 	}
 
+	if o.TargetByteDeletionRate == nil {
+		o.TargetByteDeletionRate = func() int { return 0 }
+	}
+
 	if o.FreeSpaceThresholdBytes == 0 {
 		o.FreeSpaceThresholdBytes = 16 << 30 // 16 GB
 	}
@@ -1444,7 +1448,7 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  max_open_files=%d\n", o.MaxOpenFiles)
 	fmt.Fprintf(&buf, "  mem_table_size=%d\n", o.MemTableSize)
 	fmt.Fprintf(&buf, "  mem_table_stop_writes_threshold=%d\n", o.MemTableStopWritesThreshold)
-	fmt.Fprintf(&buf, "  min_deletion_rate=%d\n", o.TargetByteDeletionRate)
+	fmt.Fprintf(&buf, "  min_deletion_rate=%d\n", o.TargetByteDeletionRate())
 	fmt.Fprintf(&buf, "  free_space_threshold_bytes=%d\n", o.FreeSpaceThresholdBytes)
 	fmt.Fprintf(&buf, "  free_space_timeframe=%s\n", o.FreeSpaceTimeframe.String())
 	fmt.Fprintf(&buf, "  obsolete_bytes_max_ratio=%f\n", o.ObsoleteBytesMaxRatio)
@@ -1789,7 +1793,11 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 				// Do nothing; option existed in older versions of pebble, and
 				// may be meaningful again eventually.
 			case "min_deletion_rate":
-				o.TargetByteDeletionRate, err = strconv.Atoi(value)
+				var rate int
+				rate, err = strconv.Atoi(value)
+				if err == nil {
+					o.TargetByteDeletionRate = func() int { return rate }
+				}
 			case "free_space_threshold_bytes":
 				o.FreeSpaceThresholdBytes, err = strconv.ParseUint(value, 10, 64)
 			case "free_space_timeframe":

options_test.go

@@ -322,7 +322,7 @@ func TestOptionsParse(t *testing.T) {
 			opts.FlushDelayDeleteRange = 10 * time.Second
 			opts.FlushDelayRangeKey = 11 * time.Second
 			opts.Experimental.LevelMultiplier = 5
-			opts.TargetByteDeletionRate = 200
+			opts.TargetByteDeletionRate = func() int { return 200 }
 			opts.WALFailover = &WALFailoverOptions{
 				Secondary: wal.Dir{Dirname: "wal_secondary", FS: vfs.Default},
 			}

pacer.go

@@ -46,7 +46,7 @@ type deletionPacer struct {
 		history history
 	}
 
-	targetByteDeletionRate int64
+	targetByteDeletionRate func() int
 
 	getInfo func() deletionPacerInfo
 }
@@ -62,7 +62,7 @@ const deletePacerHistory = 5 * time.Minute
 func newDeletionPacer(
 	now crtime.Mono,
 	freeSpaceThreshold uint64,
-	targetByteDeletionRate int64,
+	targetByteDeletionRate func() int,
 	freeSpaceTimeframe time.Duration,
 	obsoleteBytesMaxRatio float64,
 	obsoleteBytesTimeframe time.Duration,
@@ -99,12 +99,13 @@ func (p *deletionPacer) ReportDeletion(now crtime.Mono, bytesToDelete uint64) {
 //
 // PacingDelay is thread-safe.
 func (p *deletionPacer) PacingDelay(now crtime.Mono, bytesToDelete uint64) (waitSeconds float64) {
-	if p.targetByteDeletionRate == 0 {
+	targetByteDeletionRate := p.targetByteDeletionRate()
+	if targetByteDeletionRate == 0 {
 		// Pacing disabled.
 		return 0.0
 	}
 
-	baseRate := float64(p.targetByteDeletionRate)
+	baseRate := float64(targetByteDeletionRate)
 	// If recent deletion rate is more than our target, use that so that we don't
 	// fall behind.
 	historicRate := func() float64 {

pacer_test.go

@@ -10,16 +10,18 @@ import (
 	"math"
 	"math/rand/v2"
 	"slices"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/cockroachdb/crlib/crtime"
 	"github.com/stretchr/testify/require"
 )
 
+const MB = 1 << 20
+const GB = 1 << 30
+
 func TestDeletionPacer(t *testing.T) {
-	const MB = 1 << 20
-	const GB = 1 << 30
 	testCases := []struct {
 		freeBytes     uint64
 		obsoleteBytes uint64
@@ -126,7 +128,7 @@ func TestDeletionPacer(t *testing.T) {
 			pacer := newDeletionPacer(
 				start,
 				opts.FreeSpaceThresholdBytes,
-				100*MB,
+				func() int { return 100 * MB },
 				opts.FreeSpaceTimeframe,
 				opts.ObsoleteBytesMaxRatio,
 				opts.ObsoleteBytesTimeframe,
@@ -142,6 +144,37 @@ func TestDeletionPacer(t *testing.T) {
 	}
 }
 
+func TestDeletionPacerCfgChange(t *testing.T) {
+	getInfo := func() deletionPacerInfo {
+		return deletionPacerInfo{
+			freeBytes:     100 * GB,
+			liveBytes:     10 * GB,
+			obsoleteBytes: 10 * MB,
+		}
+	}
+
+	var targetRate atomic.Int32
+	targetRate.Store(100 * MB)
+
+	start := crtime.NowMono()
+	var opts Options
+	opts.EnsureDefaults()
+	pacer := newDeletionPacer(
+		start,
+		opts.FreeSpaceThresholdBytes,
+		func() int { return int(targetRate.Load()) },
+		opts.FreeSpaceTimeframe,
+		opts.ObsoleteBytesMaxRatio,
+		opts.ObsoleteBytesTimeframe,
+		getInfo,
+	)
+	require.InDelta(t, 1.0/100, pacer.PacingDelay(start, 1*MB), 1e-4)
+	targetRate.Store(200 * MB)
+	require.InDelta(t, 1.0/200, pacer.PacingDelay(start, 1*MB), 1e-4)
+	targetRate.Store(0)
+	require.Equal(t, 0.0, pacer.PacingDelay(start, 1*MB), 1e-4)
+}
+
 // TestDeletionPacerHistory tests the history helper by crosschecking Sum()
 // against a naive implementation.
 func TestDeletionPacerHistory(t *testing.T) {