db: add option for CompactionGarbageFractionForMaxConcurrency

This increases the compaction concurrency up to the maximum concurrency
based on the fraction of garbage. The default is 0.4, i.e., if 40% of
the DB is garbage, it will allow up to MaxConcurrentCompactions.

The assumption is that compensated level sizes used in scoring have
captured this garbage state, so the levels responsible for garbage
have scores > 1.0, so will eventually get picked for compactions.

Informs #4602

Author: sumeerbhola

cmd/pebble/db.go

@@ -80,6 +80,10 @@ func newPebbleDB(dir string) DB {
 	// writing), columnar blocks are only written if explicitly opted into.
 	opts.Experimental.EnableColumnarBlocks = func() bool { return true }
 
+	// Running the tool should not start compactions due to garbage.
+	opts.Experimental.CompactionGarbageFractionForMaxConcurrency = func() float64 {
+		return -1.0
+	}
 	for i := 0; i < len(opts.Levels); i++ {
 		l := &opts.Levels[i]
 		l.BlockSize = 32 << 10       // 32 KB

compaction_picker.go

@@ -706,6 +706,7 @@ type compactionPickerByScore struct {
 	// levelMaxBytes holds the dynamically adjusted max bytes setting for each
 	// level.
 	levelMaxBytes [numLevels]int64
+	dbSizeBytes   uint64
 }
 
 var _ compactionPicker = &compactionPickerByScore{}
@@ -817,7 +818,10 @@ func (p *compactionPickerByScore) initLevelMaxBytes(inProgressCompactions []comp
 		p.levelMaxBytes[level] = math.MaxInt64
 	}
 
-	if dbSize == 0 {
+	dbSizeBelowL0 := dbSize
+	dbSize += p.vers.Levels[0].Size()
+	p.dbSizeBytes = dbSize
+	if dbSizeBelowL0 == 0 {
 		// No levels for L1 and up contain any data. Target L0 compactions for the
 		// last level or to the level to which there is an ongoing L0 compaction.
 		p.baseLevel = numLevels - 1
@@ -827,7 +831,6 @@ func (p *compactionPickerByScore) initLevelMaxBytes(inProgressCompactions []comp
 		return
 	}
 
-	dbSize += p.vers.Levels[0].Size()
 	bottomLevelSize := dbSize - dbSize/uint64(p.opts.Experimental.LevelMultiplier)
 
 	curLevelSize := bottomLevelSize
@@ -1252,7 +1255,19 @@ func (p *compactionPickerByScore) getCompactionConcurrency() int {
 		compactionDebt := p.estimatedCompactionDebt(0)
 		compactionDebtCompactions = int(compactionDebt/p.opts.Experimental.CompactionDebtConcurrency) + 1
 	}
-	return max(min(maxConcurrentCompactions, max(l0ReadAmpCompactions, compactionDebtCompactions)), 1)
+	compactableGarbageCompactions := 0
+	garbageFractionLimit := p.opts.Experimental.CompactionGarbageFractionForMaxConcurrency()
+	if garbageFractionLimit > 0 && p.dbSizeBytes > 0 {
+		compactableGarbageBytes :=
+			*pointDeletionsBytesEstimateAnnotator.MultiLevelAnnotation(p.vers.Levels[:]) +
+				*rangeDeletionsBytesEstimateAnnotator.MultiLevelAnnotation(p.vers.Levels[:])
+		garbageFraction := float64(compactableGarbageBytes) / float64(p.dbSizeBytes)
+		compactableGarbageCompactions =
+			int((garbageFraction / garbageFractionLimit) * float64(maxConcurrentCompactions))
+	}
+	concurrencyBasedOnSignals :=
+		max(l0ReadAmpCompactions, compactionDebtCompactions, compactableGarbageCompactions)
+	return min(maxConcurrentCompactions, max(concurrencyBasedOnSignals, 1))
 }
 
 // TODO(sumeer): remove unless someone actually finds this useful.

compaction_picker_test.go

@@ -34,16 +34,7 @@ func loadVersion(
 	opts := &Options{}
 	opts.testingRandomized(t)
 	opts.EnsureDefaults()
-
-	if len(d.CmdArgs) != 1 {
-		return nil, nil, nil, fmt.Sprintf("%s expects 1 argument", d.Cmd)
-	}
-	var err error
-	opts.LBaseMaxBytes, err = strconv.ParseInt(d.CmdArgs[0].Key, 10, 64)
-	if err != nil {
-		return nil, nil, nil, err.Error()
-	}
-
+	d.ScanArgs(t, "l-base-max-bytes", &opts.LBaseMaxBytes)
 	var files [numLevels][]*tableMetadata
 	if len(d.Input) > 0 {
 		// Parse each line as
@@ -190,15 +181,16 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 				f.CompactionState = manifest.CompactionStateNotCompacting
 			}
 		}
+		l0Organizer.ResetForTesting(vers)
 	}
 
-	pickAuto := func(env compactionEnv, pickerByScore *compactionPickerByScore) *pickedCompaction {
+	pickAuto := func(env compactionEnv, pickerByScore *compactionPickerByScore) (pc *pickedCompaction, allowedCompactions int) {
 		inProgressCompactions := len(env.inProgressCompactions)
-		allowedCompactions := pickerByScore.getCompactionConcurrency()
+		allowedCompactions = pickerByScore.getCompactionConcurrency()
 		if inProgressCompactions >= allowedCompactions {
-			return nil
+			return nil, allowedCompactions
 		}
-		return pickerByScore.pickAutoScore(env)
+		return pickerByScore.pickAutoScore(env), allowedCompactions
 	}
 
 	datadriven.RunTest(t, "testdata/compaction_picker_target_level",
@@ -211,18 +203,26 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 				// and optionally additional compensation to be added during
 				// compensated file size calculations. Eg:
 				//
-				// init <LBaseMaxBytes>
+				// init l-base-max-bytes=<LBaseMaxBytes>
 				// <level>: <size> [compensation]
 				// <level>: <size> [compensation]
 				var errMsg string
 				vers, l0Organizer, opts, errMsg = loadVersion(t, d)
+				// This test limits the count based on the L0 read amp, compaction
+				// debt, and deleted garbage, so we would like to return math.MaxInt
+				// for most test cases. But we don't since it is also used in
+				// expandedCompactionByteSizeLimit, and causes the expanded bytes to
+				// reduce. The test cases never pick more than 4 compactions, so we
+				// use 4 as the default.
+				maxConcurrentCompactions := 4
+				if d.HasArg("max-concurrent-compactions") {
+					d.ScanArgs(t, "max-concurrent-compactions", &maxConcurrentCompactions)
+				}
 				opts.MaxConcurrentCompactions = func() int {
-					// This test only limits the count based on the L0 read amp and
-					// compaction debt, so we would like to return math.MaxInt. But we
-					// don't since it is also used in expandedCompactionByteSizeLimit,
-					// and causes the expanded bytes to reduce. The test cases never
-					// pick more than 4 compactions, so we use 4.
-					return 4
+					return maxConcurrentCompactions
+				}
+				if d.HasArg("compaction-debt-concurrency") {
+					d.ScanArgs(t, "compaction-debt-concurrency", &opts.Experimental.CompactionDebtConcurrency)
 				}
 				if errMsg != "" {
 					return errMsg
@@ -246,13 +246,20 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 			case "queue":
 				var b strings.Builder
 				var inProgress []compactionInfo
+				printConcurrency := false
+				if d.HasArg("print-compaction-concurrency") {
+					printConcurrency = true
+				}
 				for {
 					env := compactionEnv{
 						diskAvailBytes:          math.MaxUint64,
 						earliestUnflushedSeqNum: base.SeqNumMax,
 						inProgressCompactions:   inProgress,
 					}
-					pc := pickAuto(env, pickerByScore)
+					pc, concurrency := pickAuto(env, pickerByScore)
+					if printConcurrency {
+						fmt.Fprintf(&b, "compaction concurrency: %d\n", concurrency)
+					}
 					if pc == nil {
 						break
 					}
@@ -268,6 +275,10 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 						// because the test isn't marking files as Compacting.
 						break
 					}
+					if pc.startLevel != nil && pc.startLevel.level == 0 {
+						require.NoError(t, l0Organizer.UpdateStateForStartedCompaction(
+							[]manifest.LevelSlice{pc.startLevel.files}, true /* isBase */))
+					}
 					for _, cl := range pc.inputs {
 						for f := range cl.files.All() {
 							f.CompactionState = manifest.CompactionStateCompacting
@@ -336,7 +347,7 @@ func TestCompactionPickerTargetLevel(t *testing.T) {
 
 				var b strings.Builder
 				fmt.Fprintf(&b, "Initial state before pick:\n%s", runVersionFileSizes(vers))
-				pc := pickAuto(compactionEnv{
+				pc, _ := pickAuto(compactionEnv{
 					earliestUnflushedSeqNum: base.SeqNumMax,
 					inProgressCompactions:   inProgress,
 				}, pickerByScore)

metamorphic/options.go

@@ -690,6 +690,11 @@ func RandomOptions(
 	opts.MaxConcurrentCompactions = func() int {
 		return maxConcurrentCompactions
 	}
+	// [-0.2, 0.4], in steps of 0.2.
+	garbageFrac := float64(rng.IntN(5))/5.0 - 0.2
+	opts.Experimental.CompactionGarbageFractionForMaxConcurrency = func() float64 {
+		return garbageFrac
+	}
 	maxConcurrentDownloads := rng.IntN(3) + 1 // 1-3
 	opts.MaxConcurrentDownloads = func() int {
 		return maxConcurrentDownloads

metamorphic/options_test.go

@@ -73,6 +73,7 @@ func TestOptionsRoundtrip(t *testing.T) {
 		"EventListener:",
 		"MaxConcurrentCompactions:",
 		"MaxConcurrentDownloads:",
+		"Experimental.CompactionGarbageFractionForMaxConcurrency:",
 		"Experimental.DisableIngestAsFlushable:",
 		"Experimental.EnableColumnarBlocks:",
 		"Experimental.EnableValueBlocks:",
@@ -117,6 +118,12 @@ 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.Experimental.CompactionGarbageFractionForMaxConcurrency == nil,
+			parsed.Opts.Experimental.CompactionGarbageFractionForMaxConcurrency == nil)
+		if o.Opts.Experimental.CompactionGarbageFractionForMaxConcurrency != nil {
+			require.InDelta(t, o.Opts.Experimental.CompactionGarbageFractionForMaxConcurrency(),
+				parsed.Opts.Experimental.CompactionGarbageFractionForMaxConcurrency(), 1e-5)
+		}
 		require.Equal(t, o.Opts.MaxConcurrentCompactions(), parsed.Opts.MaxConcurrentCompactions())
 		require.Equal(t, o.Opts.MaxConcurrentDownloads(), parsed.Opts.MaxConcurrentDownloads())
 		require.Equal(t, len(o.Opts.BlockPropertyCollectors), len(parsed.Opts.BlockPropertyCollectors))

options.go

@@ -571,6 +571,13 @@ type Options struct {
 		// concurrency slots as determined by the two options is chosen.
 		CompactionDebtConcurrency uint64
 
+		// CompactionGarbageFractionForMaxConcurrency is the fraction of garbage
+		// due to DELs and RANGEDELs that causes MaxConcurrentCompactions to be
+		// allowed. Concurrent compactions are allowed in a linear manner upto
+		// this limit being reached. A value <= 0.0 disables adding concurrency
+		// due to garbage.
+		CompactionGarbageFractionForMaxConcurrency func() float64
+
 		// IngestSplit, if it returns true, allows for ingest-time splitting of
 		// existing sstables into two virtual sstables to allow ingestion sstables to
 		// slot into a lower level than they otherwise would have.
@@ -1208,6 +1215,11 @@ func (o *Options) EnsureDefaults() {
 	if o.Experimental.CompactionDebtConcurrency <= 0 {
 		o.Experimental.CompactionDebtConcurrency = 1 << 30 // 1 GB
 	}
+	if o.Experimental.CompactionGarbageFractionForMaxConcurrency == nil {
+		// When 40% of the DB is garbage, the compaction concurrency is at the
+		// maximum permitted.
+		o.Experimental.CompactionGarbageFractionForMaxConcurrency = func() float64 { return 0.4 }
+	}
 	if o.KeySchema == "" && len(o.KeySchemas) == 0 {
 		ks := colblk.DefaultKeySchema(o.Comparer, 16 /* bundleSize */)
 		o.KeySchema = ks.Name
@@ -1434,6 +1446,8 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  cache_size=%d\n", cacheSize)
 	fmt.Fprintf(&buf, "  cleaner=%s\n", o.Cleaner)
 	fmt.Fprintf(&buf, "  compaction_debt_concurrency=%d\n", o.Experimental.CompactionDebtConcurrency)
+	fmt.Fprintf(&buf, "  compaction_garbage_fraction_for_max_concurrency=%.2f\n",
+		o.Experimental.CompactionGarbageFractionForMaxConcurrency())
 	fmt.Fprintf(&buf, "  comparer=%s\n", o.Comparer.Name)
 	fmt.Fprintf(&buf, "  disable_wal=%t\n", o.DisableWAL)
 	if o.Experimental.DisableIngestAsFlushable != nil && o.Experimental.DisableIngestAsFlushable() {
@@ -1698,6 +1712,13 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 				}
 			case "compaction_debt_concurrency":
 				o.Experimental.CompactionDebtConcurrency, err = strconv.ParseUint(value, 10, 64)
+			case "compaction_garbage_fraction_for_max_concurrency":
+				var frac float64
+				frac, err = strconv.ParseFloat(value, 64)
+				if err == nil {
+					o.Experimental.CompactionGarbageFractionForMaxConcurrency =
+						func() float64 { return frac }
+				}
 			case "delete_range_flush_delay":
 				// NB: This is a deprecated serialization of the
 				// `flush_delay_delete_range`.

options_test.go

@@ -84,6 +84,7 @@ func TestDefaultOptionsString(t *testing.T) {
   cache_size=8388608
   cleaner=delete
   compaction_debt_concurrency=1073741824
+  compaction_garbage_fraction_for_max_concurrency=0.40
   comparer=leveldb.BytewiseComparator
   disable_wal=false
   enable_columnar_blocks=true

replay/testdata/replay

@@ -11,7 +11,7 @@ tree
      614      000007.sst
        0      LOCK
      133      MANIFEST-000001
-    1524      OPTIONS-000003
+    1579      OPTIONS-000003
        0      marker.format-version.000001.013
        0      marker.manifest.000001.MANIFEST-000001
             simple/
@@ -21,7 +21,7 @@ tree
       25        000004.log
      586        000005.sst
       85        MANIFEST-000001
-    1524        OPTIONS-000003
+    1579        OPTIONS-000003
        0        marker.format-version.000001.013
        0        marker.manifest.000001.MANIFEST-000001
 
@@ -36,6 +36,7 @@ cat build/OPTIONS-000003
   cache_size=8388608
   cleaner=replay.WorkloadCollector("delete")
   compaction_debt_concurrency=1073741824
+  compaction_garbage_fraction_for_max_concurrency=0.40
   comparer=pebble.internal.testkeys
   disable_wal=false
   enable_columnar_blocks=true

replay/testdata/replay_paced

@@ -14,7 +14,7 @@ tree
        0      LOCK
      133      MANIFEST-000001
      205      MANIFEST-000010
-    1524      OPTIONS-000003
+    1579      OPTIONS-000003
        0      marker.format-version.000001.013
        0      marker.manifest.000002.MANIFEST-000010
             high_read_amp/
@@ -26,7 +26,7 @@ tree
       39        000008.log
      560        000009.sst
      157        MANIFEST-000010
-    1524        OPTIONS-000003
+    1579        OPTIONS-000003
        0        marker.format-version.000001.013
        0        marker.manifest.000001.MANIFEST-000010