pebble: add virtual sst rewrite compaction

This commit adds a new low-priority compaction to rewrite virtual tables.
We select the backing table with the least referenced data, and materialize one
of its virtual tables into a physical table, allowing us to delete the backing
sooner and correct overestimations of blob value liveness.

Fixes: #4915

Author: xinhaoz

compaction.go

@@ -155,6 +155,7 @@ const (
 	compactionKindRewrite
 	compactionKindIngestedFlushable
 	compactionKindBlobFileRewrite
+	compactionKindVirtualRewrite
 )
 
 func (k compactionKind) String() string {
@@ -181,6 +182,8 @@ func (k compactionKind) String() string {
 		return "copy"
 	case compactionKindBlobFileRewrite:
 		return "blob-file-rewrite"
+	case compactionKindVirtualRewrite:
+		return "virtual-sst-rewrite"
 	}
 	return "?"
 }
@@ -3281,7 +3284,17 @@ func (d *DB) runCompaction(
 	case compactionKindIngestedFlushable:
 		panic("pebble: runCompaction cannot handle compactionKindIngestedFlushable.")
 	}
+	return d.runDefaultTableCompaction(jobID, c)
+}
 
+func (d *DB) runDefaultTableCompaction(
+	jobID JobID, c *tableCompaction,
+) (
+	ve *manifest.VersionEdit,
+	stats compact.Stats,
+	outputBlobs []compact.OutputBlob,
+	retErr error,
+) {
 	snapshots := d.mu.snapshots.toSlice()
 
 	// Release the d.mu lock while doing I/O.

compaction_picker.go

@@ -1497,6 +1497,12 @@ func (p *compactionPickerByScore) pickAutoNonScore(env compactionEnv) (pc picked
 		return pc
 	}
 
+	// Check for virtual SST rewrites. These compactions materialize virtual tables
+	// to reclaim space in backing files with low utilization.
+	if pc := p.pickVirtualRewriteCompaction(env); pc != nil {
+		return pc
+	}
+
 	// 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.pickBlobFileRewriteCompactionLowPriority(env); pc != nil {
@@ -1629,8 +1635,10 @@ func (p *compactionPickerByScore) pickedCompactionFromCandidateFile(
 	}
 
 	var inputs manifest.LevelSlice
-	if startLevel == 0 {
+	if startLevel == 0 && outputLevel > 0 {
 		// Overlapping L0 files must also be compacted alongside the candidate.
+		// Some compactions attempt to rewrite a file in place (e.g. virtual rewrite)
+		// so we only do this for L0->Lbase compactions.
 		inputs = p.vers.Overlaps(0, candidate.UserKeyBounds())
 	} else {
 		inputs = p.vers.Levels[startLevel].Find(p.opts.Comparer.Compare, candidate)
@@ -1692,6 +1700,39 @@ func (p *compactionPickerByScore) pickRewriteCompaction(
 	return nil
 }
 
+// pickVirtualRewriteCompaction looks for backing tables that have a low percentage
+// of referenced data and materializes their virtual sstables.
+func (p *compactionPickerByScore) pickVirtualRewriteCompaction(
+	env compactionEnv,
+) *pickedTableCompaction {
+
+	for _, c := range env.inProgressCompactions {
+		// Allow only one virtual rewrite compaction at a time.
+		if c.kind == compactionKindVirtualRewrite {
+			return nil
+		}
+	}
+
+	// We'll pick one virtual table at a time to materialize. This works with our
+	// compaction system, which currently doesn't support outputting to multiple levels
+	// or selecting files that aren't contiguous in a level. Successfully materializing
+	// one of the backing's virtual table will also make the backing more likely to be
+	// picked again, since the space amp will increase.
+	_, vtablesByLevel := p.latestVersionState.virtualBackings.ReplacementCandidate()
+	for level, tables := range vtablesByLevel {
+		for _, vt := range tables {
+			if vt.IsCompacting() {
+				continue
+			}
+			if pc := p.pickedCompactionFromCandidateFile(vt, env, level, level, compactionKindVirtualRewrite); pc != nil {
+				return pc
+			}
+		}
+	}
+
+	return nil
+}
+
 // pickBlobFileRewriteCompactionHighPriority picks a compaction that rewrites a
 // blob file to reclaim disk space if the heuristics for high-priority blob file
 // rewrites are met.

compaction_picker_test.go

@@ -1336,9 +1336,10 @@ func TestCompactionPickerPickFile(t *testing.T) {
 			}
 
 			opts := &Options{
-				Comparer:           testkeys.Comparer,
-				FormatMajorVersion: FormatNewest,
-				Logger:             testutils.Logger{T: t},
+				Comparer:                    testkeys.Comparer,
+				FormatMajorVersion:          FormatNewest,
+				Logger:                      testutils.Logger{T: t},
+				DisableAutomaticCompactions: true,
 			}
 			opts.Experimental.CompactionScheduler = func() CompactionScheduler {
 				return NewConcurrencyLimitSchedulerWithNoPeriodicGrantingForTest()

compaction_scheduler.go

@@ -142,12 +142,14 @@ func init() {
 		compactionOptionalAndPriority{optional: true, priority: 60}
 	scheduledCompactionMap[compactionKindElisionOnly] =
 		compactionOptionalAndPriority{optional: true, priority: 50}
-	scheduledCompactionMap[compactionKindBlobFileRewrite] =
+	scheduledCompactionMap[compactionKindVirtualRewrite] =
 		compactionOptionalAndPriority{optional: true, priority: 40}
-	scheduledCompactionMap[compactionKindRead] =
+	scheduledCompactionMap[compactionKindBlobFileRewrite] =
 		compactionOptionalAndPriority{optional: true, priority: 30}
-	scheduledCompactionMap[compactionKindRewrite] =
+	scheduledCompactionMap[compactionKindRead] =
 		compactionOptionalAndPriority{optional: true, priority: 20}
+	scheduledCompactionMap[compactionKindRewrite] =
+		compactionOptionalAndPriority{optional: true, priority: 10}
 }
 
 // noopGrantHandle is used in cases that don't interact with a CompactionScheduler.

compaction_test.go

@@ -1150,6 +1150,37 @@ func runCompactionTest(
 			}
 			return describeLSM(d, verbose)
 
+		case "run-virtual-rewrite-compaction":
+			err := func() error {
+				d.mu.Lock()
+				defer d.mu.Unlock()
+				d.mu.versions.logLock()
+				env := d.makeCompactionEnvLocked()
+				require.NotNil(t, env)
+				picker := d.mu.versions.picker.(*compactionPickerByScore)
+				pc := picker.pickVirtualRewriteCompaction(*env)
+				if pc == nil {
+					d.mu.versions.logUnlock()
+					return errors.New("no virtual rewrite compaction")
+				}
+				d.mu.versions.logUnlock()
+				d.runPickedCompaction(pc, noopGrantHandle{})
+				for d.mu.compact.compactingCount > 0 {
+					d.mu.compact.cond.Wait()
+				}
+				return nil
+			}()
+			if err != nil {
+				return err.Error()
+			}
+			return describeLSM(d, verbose)
+
+		case "virtual-backings":
+			d.mu.Lock()
+			s := d.mu.versions.latest.virtualBackings.String()
+			d.mu.Unlock()
+			return s
+
 		case "validate-blob-reference-index-block":
 			var inputTables []*manifest.TableMetadata
 			for _, line := range crstrings.Lines(td.Input) {
@@ -1550,6 +1581,11 @@ func TestCompaction(t *testing.T) {
 			verbose:    true,
 			cmp:        DefaultComparer,
 		},
+		"virtual_rewrite": {
+			minVersion: FormatNewest,
+			maxVersion: FormatNewest,
+			verbose:    true,
+		},
 	}
 	datadriven.Walk(t, "testdata/compaction", func(t *testing.T, path string) {
 		filename := filepath.Base(path)
@@ -1608,24 +1644,21 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 		}
 	}()
 
-	var compactInfo *CompactionInfo // protected by d.mu
+	var compactInfo []CompactionInfo
 	reset := func() (*Options, error) {
 		if d != nil {
-			compactInfo = nil
 			if err := closeAllSnapshots(d); err != nil {
 				return nil, err
 			}
 			if err := d.Close(); err != nil {
 				return nil, err
 			}
 		}
+		compactInfo = nil
 		el := TeeEventListener(
 			EventListener{
 				CompactionEnd: func(info CompactionInfo) {
-					if compactInfo != nil {
-						return
-					}
-					compactInfo = &info
+					compactInfo = append(compactInfo, info)
 				},
 			},
 			MakeLoggingEventListener(testutils.Logger{T: t}),
@@ -1653,21 +1686,29 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 		return opts, nil
 	}
 
+	compactInfo = nil
 	compactionString := func() string {
 		for d.mu.compact.compactingCount > 0 {
 			d.mu.compact.cond.Wait()
 		}
+		slices.SortFunc(compactInfo, func(a, b CompactionInfo) int {
+			return cmp.Compare(a.String(), b.String())
+		})
 
-		s := "(none)"
-		if compactInfo != nil {
+		var b strings.Builder
+		if len(compactInfo) == 0 {
+			return "(none)\n"
+		}
+
+		for _, c := range compactInfo {
 			// Fix the job ID and durations for determinism.
-			compactInfo.JobID = 100
-			compactInfo.Duration = time.Second
-			compactInfo.TotalDuration = 2 * time.Second
-			s = compactInfo.String()
-			compactInfo = nil
+			c.JobID = 100
+			c.Duration = time.Second
+			c.TotalDuration = 2 * time.Second
+			b.WriteString(fmt.Sprintf("%s\n", c.String()))
 		}
-		return s
+		compactInfo = nil
+		return b.String()
 	}
 
 	var err error
@@ -1768,12 +1809,14 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 
 				// NB: collectTableStats attempts to acquire the lock. Temporarily
 				// unlock here to avoid a deadlock.
-				d.mu.Unlock()
-				if didRun := d.collectTableStats(); !didRun {
-					// If a job was already running, wait for the results.
-					d.waitTableStats()
-				}
-				d.mu.Lock()
+				func() {
+					d.mu.Unlock()
+					defer d.mu.Lock()
+					if didRun := d.collectTableStats(); !didRun {
+						// If a job was already running, wait for the results.
+						d.waitTableStats()
+					}
+				}()
 
 				hints := d.mu.compact.deletionHints
 				if len(hints) == 0 {
@@ -1914,6 +1957,13 @@ func TestCompactionTombstones(t *testing.T) {
 				for i := range c.Output.Tables {
 					c.Output.Tables[i].FileNum = 0
 				}
+				if c.Reason == "virtual-sst-rewrite" {
+					for i := range c.Input {
+						for j := range c.Input[i].Tables {
+							c.Input[i].Tables[j].FileNum = 0
+						}
+					}
+				}
 			}
 		}
 		// Sort for determinism. We use the negative value of cmp.Compare to sort

db_test.go

@@ -1421,6 +1421,9 @@ func TestVirtualSSTables(t *testing.T) {
 	d, err := Open("", testingRandomized(t, &Options{
 		FS:                 vfs.NewMem(),
 		FormatMajorVersion: FormatTableFormatV6,
+		// Disable automatic compactions, as virtual ssts may
+		// be compacted away too quickly.
+		DisableAutomaticCompactions: true,
 	}))
 	require.NoError(t, err)
 	defer func() {

internal/manifest/virtual_backings.go

@@ -310,7 +310,9 @@ func (bv *VirtualBackings) ReplacementCandidate() (*TableBacking, [NumLevels][]*
 	}
 	v := bv.rewriteCandidates.items[0]
 	var tables [NumLevels][]*TableMetadata
-	for _, tl := range v.virtualTables {
+	tableNums := slices.Sorted(maps.Keys(v.virtualTables))
+	for _, t := range tableNums {
+		tl := v.virtualTables[t]
 		tables[tl.level] = append(tables[tl.level], tl.meta)
 	}
 	return v.backing, tables
@@ -330,7 +332,7 @@ func (bv *VirtualBackings) String() string {
 			fmt.Fprintf(&buf, "  %s:  size=%d  refBlobValueSize=%d  useCount=%d  protectionCount=%d  virtualizedSize=%d",
 				n, v.backing.Size, v.backing.ReferencedBlobValueSizeTotal, len(v.virtualTables), v.protectionCount, v.virtualizedSize)
 			if !v.isLocal {
-				fmt.Fprintf(&buf, "  (external)")
+				fmt.Fprintf(&buf, "  (remote)")
 			}
 			tableNums := slices.Sorted(maps.Keys(v.virtualTables))
 			fmt.Fprintf(&buf, "  tables: %v\n", tableNums)

metrics.go

@@ -229,6 +229,7 @@ type Metrics struct {
 		RewriteCount          int64
 		MultiLevelCount       int64
 		BlobFileRewriteCount  int64
+		VirtualRewriteCount   int64
 		// An estimate of the number of bytes that need to be compacted for the LSM
 		// to reach a stable state.
 		EstimatedDebt uint64
@@ -733,6 +734,7 @@ var (
 		table.Int64("copy", 5, table.AlignRight, func(m *Metrics) int64 { return m.Compact.CopyCount }),
 		table.Int64("multi", 6, table.AlignRight, func(m *Metrics) int64 { return m.Compact.MultiLevelCount }),
 		table.Int64("blob", 5, table.AlignRight, func(m *Metrics) int64 { return m.Compact.BlobFileRewriteCount }),
+		table.Int64("virtual", 7, table.AlignRight, func(m *Metrics) int64 { return m.Compact.VirtualRewriteCount }),
 	)
 	commitPipelineInfoTableTopHeader = `COMMIT PIPELINE`
 	commitPipelineInfoTableSubHeader = `               wals                |              memtables              |       ingestions`

metrics_test.go

@@ -53,6 +53,7 @@ func exampleMetrics() Metrics {
 	m.Compact.RewriteCount = 17
 	m.Compact.MultiLevelCount = 18
 	m.Compact.BlobFileRewriteCount = 19
+	m.Compact.VirtualRewriteCount = 20
 	m.Compact.EstimatedDebt = 6 * GB
 	m.Compact.InProgressBytes = 1 * MB
 	m.Compact.NumInProgress = 2

testdata/compaction/l0_to_lbase_compaction

@@ -49,8 +49,8 @@ level | score    ff   cff | tables  size |   top    in  read | tables  blob | ta
 ------+-------------------+--------------+-------------------+--------------+---------------------
 total |     -     -     - |      3   6MB |    0B    0B    0B |     0B    0B |      3   10MB     0B
 
- kind | default  delete  elision  move  read  tomb  rewrite  copy  multi  blob
-count |       0       0        0     3     0     0        0     0      0     0
+ kind | default  delete  elision  move  read  tomb  rewrite  copy  multi  blob virtual
+count |       0       0        0     3     0     0        0     0      0     0       0
 
 COMMIT PIPELINE
                wals                |              memtables              |       ingestions