Repository: cockroachdb/pebble
Pull request #4565 (2025-04-16 21:59:57)

sstable: fix IndexSize, TopLevelIndexSize and NumDataBlocks
The RawColumnWriter has several bugs related to setting properties:

IndexSize: In previous table formats, IndexSize recorded the _uncompressed_ size of the index block.  The RawColumnWriter incorrectly totalled the compressed sizes of all index blocks.

TopLevelIndexSize: The RawColumnWriter did not set this property at all.

NumDataBlocks: When a table had a two-level index, the RawColumnWriter miscalculated NumDataBlocks. It improperly set the value to the number of entries in the first index block multipled by the count of bottom-level index blocks.

cockroach-teamcity: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cockroachdb/pebble/4565)
<!-- Reviewable:end -->


Patch:
diff --git a/compaction.go b/compaction.go
index 42b1cb23..c4800dd1 100644
--- a/compaction.go
+++ b/compaction.go
@@ -174,6 +174,15 @@ func (k compactionKind) String() string {
 	return "?"
 }
 
+// compactingOrFlushing returns "flushing" if the compaction kind is a flush,
+// otherwise it returns "compacting".
+func (k compactionKind) compactingOrFlushing() string {
+	if k == compactionKindFlush {
+		return "flushing"
+	}
+	return "compacting"
+}
+
 // compaction is a table compaction from one level to the next, starting from a
 // given version.
 type compaction struct {
@@ -354,32 +363,33 @@ func (c *compaction) userKeyBounds() base.UserKeyBounds {
 	return base.UserKeyBoundsFromInternal(c.smallest, c.largest)
 }
 
+type getValueSeparation func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation
+
 func newCompaction(
 	pc *pickedCompaction,
 	opts *Options,
 	beganAt time.Time,
 	provider objstorage.Provider,
 	grantHandle CompactionGrantHandle,
+	getValueSeparation getValueSeparation,
 ) *compaction {
 	c := &compaction{
-		kind:      compactionKindDefault,
-		cmp:       pc.cmp,
-		equal:     opts.Comparer.Equal,
-		comparer:  opts.Comparer,
-		formatKey: opts.Comparer.FormatKey,
-		inputs:    pc.inputs,
-		smallest:  pc.smallest,
-		largest:   pc.largest,
-		logger:    opts.Logger,
-		version:   pc.version,
-		beganAt:   beganAt,
-		getValueSeparation: func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation {
-			return compact.NeverSeparateValues{}
-		},
-		maxOutputFileSize: pc.maxOutputFileSize,
-		maxOverlapBytes:   pc.maxOverlapBytes,
-		pickerMetrics:     pc.pickerMetrics,
-		grantHandle:       grantHandle,
+		kind:               compactionKindDefault,
+		cmp:                pc.cmp,
+		equal:              opts.Comparer.Equal,
+		comparer:           opts.Comparer,
+		formatKey:          opts.Comparer.FormatKey,
+		inputs:             pc.inputs,
+		smallest:           pc.smallest,
+		largest:            pc.largest,
+		logger:             opts.Logger,
+		version:            pc.version,
+		beganAt:            beganAt,
+		getValueSeparation: getValueSeparation,
+		maxOutputFileSize:  pc.maxOutputFileSize,
+		maxOverlapBytes:    pc.maxOverlapBytes,
+		pickerMetrics:      pc.pickerMetrics,
+		grantHandle:        grantHandle,
 	}
 	c.startLevel = &c.inputs[0]
 	if pc.startLevel.l0SublevelInfo != nil {
@@ -540,24 +550,23 @@ func newFlush(
 	baseLevel int,
 	flushing flushableList,
 	beganAt time.Time,
+	getValueSeparation getValueSeparation,
 ) (*compaction, error) {
 	c := &compaction{
-		kind:      compactionKindFlush,
-		cmp:       opts.Comparer.Compare,
-		equal:     opts.Comparer.Equal,
-		comparer:  opts.Comparer,
-		formatKey: opts.Comparer.FormatKey,
-		logger:    opts.Logger,
-		version:   cur,
-		beganAt:   beganAt,
-		inputs:    []compactionLevel{{level: -1}, {level: 0}},
-		getValueSeparation: func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation {
-			return compact.NeverSeparateValues{}
-		},
-		maxOutputFileSize: math.MaxUint64,
-		maxOverlapBytes:   math.MaxUint64,
-		flushing:          flushing,
-		grantHandle:       noopGrantHandle{},
+		kind:               compactionKindFlush,
+		cmp:                opts.Comparer.Compare,
+		equal:              opts.Comparer.Equal,
+		comparer:           opts.Comparer,
+		formatKey:          opts.Comparer.FormatKey,
+		logger:             opts.Logger,
+		version:            cur,
+		beganAt:            beganAt,
+		inputs:             []compactionLevel{{level: -1}, {level: 0}},
+		getValueSeparation: getValueSeparation,
+		maxOutputFileSize:  math.MaxUint64,
+		maxOverlapBytes:    math.MaxUint64,
+		flushing:           flushing,
+		grantHandle:        noopGrantHandle{},
 	}
 	c.startLevel = &c.inputs[0]
 	c.outputLevel = &c.inputs[1]
@@ -1461,7 +1470,7 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	}
 
 	c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.l0Organizer,
-		d.mu.versions.picker.getBaseLevel(), d.mu.mem.queue[:n], d.timeNow())
+		d.mu.versions.picker.getBaseLevel(), d.mu.mem.queue[:n], d.timeNow(), d.determineCompactionValueSeparation)
 	if err != nil {
 		return 0, err
 	}
@@ -1827,7 +1836,7 @@ func (d *DB) runPickedCompaction(pc *pickedCompaction, grantHandle CompactionGra
 	}
 
 	d.mu.compact.compactingCount++
-	compaction := newCompaction(pc, d.opts, d.timeNow(), d.ObjProvider(), grantHandle)
+	compaction := newCompaction(pc, d.opts, d.timeNow(), d.ObjProvider(), grantHandle, d.determineCompactionValueSeparation)
 	d.addInProgressCompaction(compaction)
 	go func() {
 		d.compact(compaction, doneChannel)
@@ -2396,11 +2405,22 @@ func (d *DB) handleCompactFailure(c *compaction, err error) {
 func (d *DB) cleanupVersionEdit(ve *versionEdit) {
 	obsoleteFiles := manifest.ObsoleteFiles{
 		FileBackings: make([]*fileBacking, 0, len(ve.NewTables)),
+		BlobFiles:    make([]*manifest.BlobFileMetadata, 0, len(ve.NewBlobFiles)),
 	}
 	deletedFiles := make(map[base.FileNum]struct{})
 	for key := range ve.DeletedTables {
 		deletedFiles[key.FileNum] = struct{}{}
 	}
+	for i := range ve.NewBlobFiles {
+		obsoleteFiles.AddBlob(ve.NewBlobFiles[i])
+		d.mu.versions.zombieBlobs.Add(objectInfo{
+			fileInfo: fileInfo{
+				FileNum:  ve.NewBlobFiles[i].FileNum,
+				FileSize: ve.NewBlobFiles[i].Size,
+			},
+			isLocal: objstorage.IsLocalBlobFile(d.objProvider, ve.NewBlobFiles[i].FileNum),
+		})
+	}
 	for i := range ve.NewTables {
 		if ve.NewTables[i].Meta.Virtual {
 			// We handle backing files separately.
@@ -2427,12 +2447,7 @@ func (d *DB) cleanupVersionEdit(ve *versionEdit) {
 				FileNum:  of.DiskFileNum,
 				FileSize: of.Size,
 			},
-			// TODO(bilal): This is harmless if it's wrong, as it only causes
-			// incorrect accounting for the size of it in metrics. Currently
-			// all compactions only write to local files anyway except with
-			// disaggregated storage; if this becomes the norm, we should do
-			// an objprovider lookup here.
-			isLocal: true,
+			isLocal: objstorage.IsLocalTable(d.objProvider, of.DiskFileNum),
 		})
 	}
 	d.mu.versions.addObsoleteLocked(obsoleteFiles)
@@ -3021,9 +3036,10 @@ func (d *DB) runCompaction(
 		ve, result.Err = c.makeVersionEdit(result)
 	}
 	if result.Err != nil {
-		// Delete any created tables.
+		// Delete any created tables or blob files.
 		obsoleteFiles := manifest.ObsoleteFiles{
 			FileBackings: make([]*fileBacking, 0, len(result.Tables)),
+			BlobFiles:    make([]*manifest.BlobFileMetadata, 0, len(result.Blobs)),
 		}
 		d.mu.Lock()
 		for i := range result.Tables {
@@ -3037,6 +3053,13 @@ func (d *DB) runCompaction(
 			// marked zombie.
 			d.mu.versions.zombieTables.AddMetadata(&result.Tables[i].ObjMeta, backing.Size)
 		}
+		for i := range result.Blobs {
+			obsoleteFiles.AddBlob(result.Blobs[i].Metadata)
+			// Add this file to zombie blobs as well, as the versionSet
+			// asserts on whether every obsolete file was at one point
+			// marked zombie.
+			d.mu.versions.zombieBlobs.AddMetadata(&result.Blobs[i].ObjMeta, result.Blobs[i].Metadata.Size)
+		}
 		d.mu.versions.addObsoleteLocked(obsoleteFiles)
 		d.mu.Unlock()
 	}
@@ -3094,6 +3117,7 @@ func (d *DB) compactAndWrite(
 		readEnv:          sstable.ReadEnv{Block: blockReadEnv},
 		blobValueFetcher: &c.valueFetcher,
 	}
+	defer func() { _ = c.valueFetcher.Close() }()
 
 	pointIter, rangeDelIter, rangeKeyIter, err := c.newInputIters(d.newIters, d.tableNewRangeKeyIter, iiopts)
 	defer func() {
@@ -3143,7 +3167,7 @@ func (d *DB) compactAndWrite(
 		}
 		// Create a new table.
 		writerOpts := d.opts.MakeWriterOptions(c.outputLevel.level, tableFormat)
-		objMeta, tw, err := d.newCompactionOutput(jobID, c, writerOpts)
+		objMeta, tw, err := d.newCompactionOutputTable(jobID, c, writerOpts)
 		if err != nil {
 			return runner.Finish().WithError(err)
 		}
@@ -3271,43 +3295,52 @@ func (c *compaction) makeVersionEdit(result compact.Result) (*versionEdit, error
 	return ve, nil
 }
 
-// newCompactionOutput creates an object for a new table produced by a
+// newCompactionOutputTable creates an object for a new table produced by a
 // compaction or flush.
-func (d *DB) newCompactionOutput(
+func (d *DB) newCompactionOutputTable(
 	jobID JobID, c *compaction, writerOpts sstable.WriterOptions,
 ) (objstorage.ObjectMetadata, sstable.RawWriter, error) {
-	writable, objMeta, err := d.newCompactionOutputObj(jobID, c, base.FileTypeTable)
+	writable, objMeta, err := d.newCompactionOutputObj(c, base.FileTypeTable)
 	if err != nil {
 		return objstorage.ObjectMetadata{}, nil, err
 	}
-
-	var reason string
-	if c.kind == compactionKindFlush {
-		reason = "flushing"
-	} else {
-		reason = "compacting"
-	}
 	d.opts.EventListener.TableCreated(TableCreateInfo{
 		JobID:   int(jobID),
-		Reason:  reason,
+		Reason:  c.kind.compactingOrFlushing(),
 		Path:    d.objProvider.Path(objMeta),
 		FileNum: objMeta.DiskFileNum,
 	})
-
 	writerOpts.SetInternal(sstableinternal.WriterOptions{
 		CacheOpts: sstableinternal.CacheOptions{
 			CacheHandle: d.cacheHandle,
 			FileNum:     objMeta.DiskFileNum,
 		},
 	})
-
 	tw := sstable.NewRawWriterWithCPUMeasurer(writable, writerOpts, c.grantHandle)
 	return objMeta, tw, nil
 }
 
+// newCompactionOutputBlob creates an object for a new blob produced by a
+// compaction or flush.
+func (d *DB) newCompactionOutputBlob(
+	jobID JobID, c *compaction,
+) (objstorage.Writable, objstorage.ObjectMetadata, error) {
+	writable, objMeta, err := d.newCompactionOutputObj(c, base.FileTypeBlob)
+	if err != nil {
+		return nil, objstorage.ObjectMetadata{}, err
+	}
+	d.opts.EventListener.BlobFileCreated(BlobFileCreateInfo{
+		JobID:   int(jobID),
+		Reason:  c.kind.compactingOrFlushing(),
+		Path:    d.objProvider.Path(objMeta),
+		FileNum: objMeta.DiskFileNum,
+	})
+	return writable, objMeta, nil
+}
+
 // newCompactionOutputObj creates an object produced by a compaction or flush.
 func (d *DB) newCompactionOutputObj(
-	jobID JobID, c *compaction, typ base.FileType,
+	c *compaction, typ base.FileType,
 ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
 	diskFileNum := d.mu.versions.getNextDiskFileNum()
 
diff --git a/db.go b/db.go
index d13b1808..c5c3fbf0 100644
--- a/db.go
+++ b/db.go
@@ -612,6 +612,9 @@ func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, io.Closer,
 		readState:    readState,
 		keyBuf:       buf.keyBuf,
 	}
+	// Set up a blob value fetcher to use for retrieving values from blob files.
+	i.blobValueFetcher.Init(d.fileCache, block.NoReadEnv)
+	get.iiopts.blobValueFetcher = &i.blobValueFetcher
 
 	if !i.First() {
 		err := i.Close()
@@ -1755,7 +1758,7 @@ func (d *DB) Close() error {
 
 	// Since we called d.readState.val.unrefLocked() above, we are expected to
 	// manually schedule deletion of obsolete files.
-	if len(d.mu.versions.obsoleteTables) > 0 {
+	if len(d.mu.versions.obsoleteTables) > 0 || len(d.mu.versions.obsoleteBlobs) > 0 {
 		d.deleteObsoleteFiles(d.newJobIDLocked())
 	}
 
@@ -1775,11 +1778,14 @@ func (d *DB) Close() error {
 
 	d.mu.Lock()
 
-	// As a sanity check, ensure that there are no zombie tables. A non-zero count
-	// hints at a reference count leak.
+	// As a sanity check, ensure that there are no zombie tables or blob files.
+	// A non-zero count hints at a reference count leak.
 	if ztbls := d.mu.versions.zombieTables.Count(); ztbls > 0 {
 		err = firstError(err, errors.Errorf("non-zero zombie file count: %d", ztbls))
 	}
+	if zblobs := d.mu.versions.zombieBlobs.Count(); zblobs > 0 {
+		err = firstError(err, errors.Errorf("non-zero zombie blob count: %d", zblobs))
+	}
 
 	err = firstError(err, d.objProvider.Close())
 
@@ -2061,7 +2067,12 @@ func (d *DB) Metrics() *Metrics {
 	backingCount, backingTotalSize := d.mu.versions.virtualBackings.Stats()
 	metrics.Table.BackingTableCount = uint64(backingCount)
 	metrics.Table.BackingTableSize = backingTotalSize
+	blobStats := d.mu.versions.blobFiles.Stats()
 	d.mu.versions.logUnlock()
+	metrics.BlobFiles.LiveCount = blobStats.Count
+	metrics.BlobFiles.LiveSize = blobStats.PhysicalSize
+	metrics.BlobFiles.ValueSize = blobStats.ValueSize
+	metrics.BlobFiles.ReferencedValueSize = blobStats.ReferencedValueSize
 
 	metrics.LogWriter.FsyncLatency = d.mu.log.metrics.fsyncLatency
 	if err := metrics.LogWriter.Merge(&d.mu.log.metrics.LogWriterMetrics); err != nil {
diff --git a/download.go b/download.go
index 9af69d05..96752368 100644
--- a/download.go
+++ b/download.go
@@ -439,7 +439,7 @@ func (d *DB) tryLaunchDownloadForFile(
 
 	download.numLaunchedDownloads++
 	doneCh = make(chan error, 1)
-	c := newCompaction(pc, d.opts, d.timeNow(), d.objProvider, noopGrantHandle{})
+	c := newCompaction(pc, d.opts, d.timeNow(), d.objProvider, noopGrantHandle{}, d.determineCompactionValueSeparation)
 	c.isDownload = true
 	d.mu.compact.downloadingCount++
 	d.addInProgressCompaction(c)
diff --git a/event.go b/event.go
index 68f247c5..ea5fdc42 100644
--- a/event.go
+++ b/event.go
@@ -97,6 +97,48 @@ func (i LevelInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 		redact.Safe(i.Score))
 }
 
+// BlobFileCreateInfo contains the info for a blob file creation event.
+type BlobFileCreateInfo struct {
+	JobID int
+	// Reason is the reason for the table creation: "compacting", "flushing", or
+	// "ingesting".
+	Reason  string
+	Path    string
+	FileNum base.DiskFileNum
+}
+
+func (i BlobFileCreateInfo) String() string {
+	return redact.StringWithoutMarkers(i)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (i BlobFileCreateInfo) SafeFormat(w redact.SafePrinter, _ rune) {
+	w.Printf("[JOB %d] %s: blob file created %s",
+		redact.Safe(i.JobID), redact.Safe(i.Reason), i.FileNum)
+}
+
+// BlobFileDeleteInfo contains the info for a blob file deletion event.
+type BlobFileDeleteInfo struct {
+	JobID   int
+	Path    string
+	FileNum base.DiskFileNum
+	Err     error
+}
+
+func (i BlobFileDeleteInfo) String() string {
+	return redact.StringWithoutMarkers(i)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (i BlobFileDeleteInfo) SafeFormat(w redact.SafePrinter, _ rune) {
+	if i.Err != nil {
+		w.Printf("[JOB %d] blob file delete error %s: %s",
+			redact.Safe(i.JobID), i.FileNum, i.Err)
+		return
+	}
+	w.Printf("[JOB %d] blob file deleted %s", redact.Safe(i.JobID), i.FileNum)
+}
+
 // CompactionInfo contains the info for a compaction event.
 type CompactionInfo struct {
 	// JobID is the ID of the compaction job.
@@ -728,6 +770,12 @@ type EventListener struct {
 	// operation such as flush or compaction.
 	BackgroundError func(error)
 
+	// BlobFileCreated is invoked after a blob file has been created.
+	BlobFileCreated func(BlobFileCreateInfo)
+
+	// BlobFileDeleted is invoked after a blob file has been deleted.
+	BlobFileDeleted func(BlobFileDeleteInfo)
+
 	// DataCorruption is invoked when an on-disk corruption is detected. It should
 	// not block, as it is called synchronously in read paths.
 	DataCorruption func(DataCorruptionInfo)
@@ -827,6 +875,12 @@ func (l *EventListener) EnsureDefaults(logger Logger) {
 			l.BackgroundError = func(error) {}
 		}
 	}
+	if l.BlobFileCreated == nil {
+		l.BlobFileCreated = func(info BlobFileCreateInfo) {}
+	}
+	if l.BlobFileDeleted == nil {
+		l.BlobFileDeleted = func(info BlobFileDeleteInfo) {}
+	}
 	if l.DataCorruption == nil {
 		if logger != nil {
 			l.DataCorruption = func(info DataCorruptionInfo) {
@@ -912,6 +966,12 @@ func MakeLoggingEventListener(logger Logger) EventListener {
 		BackgroundError: func(err error) {
 			logger.Errorf("background error: %s", err)
 		},
+		BlobFileCreated: func(info BlobFileCreateInfo) {
+			logger.Infof("%s", info)
+		},
+		BlobFileDeleted: func(info BlobFileDeleteInfo) {
+			logger.Infof("%s", info)
+		},
 		DataCorruption: func(info DataCorruptionInfo) {
 			logger.Errorf("%s", info)
 		},
@@ -990,6 +1050,14 @@ func TeeEventListener(a, b EventListener) EventListener {
 			a.BackgroundError(err)
 			b.BackgroundError(err)
 		},
+		BlobFileCreated: func(info BlobFileCreateInfo) {
+			a.BlobFileCreated(info)
+			b.BlobFileCreated(info)
+		},
+		BlobFileDeleted: func(info BlobFileDeleteInfo) {
+			a.BlobFileDeleted(info)
+			b.BlobFileDeleted(info)
+		},
 		DataCorruption: func(info DataCorruptionInfo) {
 			a.DataCorruption(info)
 			b.DataCorruption(info)
diff --git a/format_major_version.go b/format_major_version.go
index 3f5f6323..c8af819d 100644
--- a/format_major_version.go
+++ b/format_major_version.go
@@ -215,10 +215,17 @@ const (
 	// This format major version does not yet enable use of value separation.
 	FormatTableFormatV6
 
+	// formatValueSeparation enables the use of value separation, separating
+	// values into external blob files that do not participate in every
+	// compaction.
+	//
+	// TODO(jackson): Export this format major version once stable.
+	formatValueSeparation
+
 	// -- Add new versions here --
 
 	// FormatNewest is the most recent format major version.
-	FormatNewest FormatMajorVersion = iota - 1
+	FormatNewest FormatMajorVersion = FormatTableFormatV6
 
 	// Experimental versions, which are excluded by FormatNewest (but can be used
 	// in tests) can be defined here.
@@ -255,7 +262,7 @@ func (v FormatMajorVersion) MaxTableFormat() sstable.TableFormat {
 		return sstable.TableFormatPebblev4
 	case FormatColumnarBlocks, FormatWALSyncChunks:
 		return sstable.TableFormatPebblev5
-	case FormatTableFormatV6:
+	case FormatTableFormatV6, formatValueSeparation:
 		return sstable.TableFormatPebblev6
 	default:
 		panic(fmt.Sprintf("pebble: unsupported format major version: %s", v))
@@ -269,7 +276,7 @@ func (v FormatMajorVersion) MinTableFormat() sstable.TableFormat {
 	case FormatDefault, FormatFlushableIngest, FormatPrePebblev1MarkedCompacted,
 		FormatDeleteSizedAndObsolete, FormatVirtualSSTables, FormatSyntheticPrefixSuffix,
 		FormatFlushableIngestExcises, FormatColumnarBlocks, FormatWALSyncChunks,
-		FormatTableFormatV6:
+		FormatTableFormatV6, formatValueSeparation:
 		return sstable.TableFormatPebblev1
 	default:
 		panic(fmt.Sprintf("pebble: unsupported format major version: %s", v))
@@ -318,6 +325,9 @@ var formatMajorVersionMigrations = map[FormatMajorVersion]func(*DB) error{
 	FormatTableFormatV6: func(d *DB) error {
 		return d.finalizeFormatVersUpgrade(FormatTableFormatV6)
 	},
+	formatValueSeparation: func(d *DB) error {
+		return d.finalizeFormatVersUpgrade(formatValueSeparation)
+	},
 }
 
 const formatVersionMarkerName = `format-version`
diff --git a/get_iter.go b/get_iter.go
index 0410feb0..81675611 100644
--- a/get_iter.go
+++ b/get_iter.go
@@ -23,6 +23,7 @@ type getIter struct {
 	newIters tableNewIters
 	snapshot base.SeqNum
 	iterOpts IterOptions
+	iiopts   internalIterOpts
 	key      []byte
 	prefix   []byte
 	iter     internalIterator
@@ -271,7 +272,7 @@ func (g *getIter) getSSTableIterators(
 	}
 	// m may possibly contain point (or range deletion) keys relevant to g.key.
 	g.iterOpts.layer = level
-	iters, err := g.newIters(context.Background(), m, &g.iterOpts, internalIterOpts{}, iterPointKeys|iterRangeDeletions)
+	iters, err := g.newIters(context.Background(), m, &g.iterOpts, g.iiopts, iterPointKeys|iterRangeDeletions)
 	if err != nil {
 		return emptyIter, nil, err
 	}
diff --git a/internal/manifest/blob_metadata.go b/internal/manifest/blob_metadata.go
index 481485af..97424a07 100644
--- a/internal/manifest/blob_metadata.go
+++ b/internal/manifest/blob_metadata.go
@@ -5,6 +5,9 @@
 package manifest
 
 import (
+	stdcmp "cmp"
+	"fmt"
+	"slices"
 	"sync/atomic"
 
 	"github.com/cockroachdb/errors"
@@ -57,55 +60,6 @@ type BlobFileMetadata struct {
 	// referencing TableMetadata is installed in a Version and decremented when
 	// that TableMetadata becomes obsolete.
 	refs atomic.Int32
-
-	// ActiveRefs holds state that describes the latest (the 'live') Version
-	// containing this blob file and the references to the file within that
-	// version.
-	ActiveRefs BlobFileActiveRefs
-}
-
-// BlobFileActiveRefs describes the state of a blob file within the latest
-// Version and the references to the file within that version.
-type BlobFileActiveRefs struct {
-	// count holds the number of tables in the latest version that reference
-	// this blob file. When this reference count falls to zero, the blob file is
-	// either a zombie file (if BlobFileMetadata.refs > 0), or obsolete and
-	// ready to be deleted.
-	//
-	// INVARIANT: BlobFileMetadata.refs > BlobFileMetadata.ActiveRefs.count
-	count int32
-	// valueSize is the sum of the length of uncompressed values in this blob
-	// file that are still live (i.e., referred to by sstables in the latest
-	// version).
-	valueSize uint64
-}
-
-// AddRef records a new reference to the blob file from a sstable in the latest
-// Version. The provided valueSize should be the value of the sstable's
-// BlobReference.ValueSize.
-//
-// Requires the manifest logLock be held.
-func (r *BlobFileActiveRefs) AddRef(valueSize uint64) {
-	r.count++
-	r.valueSize += valueSize
-}
-
-// RemoveRef removes a reference that no longer exists in the latest Version.
-// The provided valueSize should be the value of the removed sstable's
-// BlobReference.ValueSize.
-//
-// Requires the manifest logLock be held.
-func (r *BlobFileActiveRefs) RemoveRef(valueSize uint64) {
-	if invariants.Enabled {
-		if r.count <= 0 {
-			panic(errors.AssertionFailedf("pebble: negative active ref count"))
-		}
-		if valueSize > r.valueSize {
-			panic(errors.AssertionFailedf("pebble: negative active value size"))
-		}
-	}
-	r.count--
-	r.valueSize -= valueSize
 }
 
 // SafeFormat implements redact.SafeFormatter.
@@ -244,3 +198,207 @@ func (br BlobReferences) IDByFileNum(fileNum base.DiskFileNum) (blob.ReferenceID
 	}
 	return blob.ReferenceID(len(br)), false
 }
+
+// AggregateBlobFileStats records cumulative stats across blob files.
+type AggregateBlobFileStats struct {
+	// Count is the number of blob files in the set.
+	Count uint64
+	// PhysicalSize is the sum of the size of all blob files in the set.  This
+	// is the size of the blob files on physical storage. Data within blob files
+	// is compressed, so this value may be less than ValueSize.
+	PhysicalSize uint64
+	// ValueSize is the sum of the length of the uncompressed values in all blob
+	// files in the set.
+	ValueSize uint64
+	// ReferencedValueSize is the sum of the length of the uncompressed values
+	// in all blob files in the set that are still referenced by live tables
+	// (i.e., in the latest version).
+	ReferencedValueSize uint64
+	// ReferencesCount is the total number of tracked references in live tables
+	// (i.e., in the latest version). When virtual sstables are present, this
+	// count is per-virtual sstable (not per backing physical sstable).
+	ReferencesCount uint64
+}
+
+// String implements fmt.Stringer.
+func (s AggregateBlobFileStats) String() string {
+	return fmt.Sprintf("Files:{Count: %d, Size: %d, ValueSize: %d}, References:{ValueSize: %d, Count: %d}",
+		s.Count, s.PhysicalSize, s.ValueSize, s.ReferencedValueSize, s.ReferencesCount)
+}
+
+// CurrentBlobFileSet describes the set of blob files that are currently live in
+// the latest Version. CurrentBlobFileSet is not thread-safe. In practice its
+// use is protected by the versionSet logLock.
+type CurrentBlobFileSet struct {
+	// files is a map of blob file numbers to a *currentBlobFile, recording
+	// metadata about the blob file's active references in the latest Version.
+	files map[base.DiskFileNum]*currentBlobFile
+	// stats records cumulative stats across all blob files in the set.
+	stats AggregateBlobFileStats
+}
+
+type currentBlobFile struct {
+	metadata *BlobFileMetadata
+	// references holds pointers to TableMetadatas that exist in the latest
+	// version and reference this blob file. When the length of references falls
+	// to zero, the blob file is either a zombie file (if BlobFileMetadata.refs
+	// > 0), or obsolete and ready to be deleted.
+	//
+	// INVARIANT: BlobFileMetadata.refs >= len(references)
+	//
+	// TODO(jackson): Rather than using 1 map per blob file which needs to grow
+	// and shrink over the lifetime of the blob file, we could use a single
+	// B-Tree that holds all blob references, sorted by blob file then by
+	// referencing table number. This would likely be more memory efficient,
+	// reduce overall number of pointers to chase and suffer fewer allocations
+	// (and we can pool the B-Tree nodes to further reduce allocs)
+	references map[*TableMetadata]struct{}
+	// referencedValueSize is the sum of the length of uncompressed values in
+	// this blob file that are still live.
+	referencedValueSize uint64
+}
+
+// Init initializes the CurrentBlobFileSet with the state of the provided
+// BulkVersionEdit. This is used after replaying a manifest.
+func (s *CurrentBlobFileSet) Init(bve *BulkVersionEdit) {
+	*s = CurrentBlobFileSet{files: make(map[base.DiskFileNum]*currentBlobFile)}
+	if bve == nil {
+		return
+	}
+	for _, m := range bve.BlobFiles.Added {
+		s.files[m.FileNum] = &currentBlobFile{
+			metadata:   m,
+			references: make(map[*TableMetadata]struct{}),
+		}
+		s.stats.Count++
+		s.stats.PhysicalSize += m.Size
+		s.stats.ValueSize += m.ValueSize
+	}
+	// Record references to blob files from extant tables. Any referenced blob
+	// files should already exist in s.files.
+	for _, levelFiles := range bve.AddedTables {
+		for _, m := range levelFiles {
+			for _, ref := range m.BlobReferences {
+				cbf, ok := s.files[ref.FileNum]
+				if !ok {
+					panic(errors.AssertionFailedf("pebble: referenced blob file %d not found", ref.FileNum))
+				}
+				cbf.references[m] = struct{}{}
+				cbf.referencedValueSize += ref.ValueSize
+				s.stats.ReferencedValueSize += ref.ValueSize
+				s.stats.ReferencesCount++
+			}
+		}
+	}
+}
+
+// Stats returns the cumulative stats across all blob files in the set.
+func (s *CurrentBlobFileSet) Stats() AggregateBlobFileStats {
+	return s.stats
+}
+
+// Metadatas returns a slice of all blob file metadata in the set, sorted by
+// file number for determinism.
+func (s *CurrentBlobFileSet) Metadatas() []*BlobFileMetadata {
+	m := make([]*BlobFileMetadata, 0, len(s.files))
+	for _, cbf := range s.files {
+		m = append(m, cbf.metadata)
+	}
+	slices.SortFunc(m, func(a, b *BlobFileMetadata) int {
+		return stdcmp.Compare(a.FileNum, b.FileNum)
+	})
+	return m
+}
+
+// ApplyAndUpdateVersionEdit applies a version edit to the current blob file
+// set, updating its internal tracking of extant blob file references. If after
+// applying the version edit a blob file has no more references, the version
+// edit is modified to record the blob file removal.
+func (s *CurrentBlobFileSet) ApplyAndUpdateVersionEdit(ve *VersionEdit) error {
+	// Insert new blob files into the set.
+	for _, nf := range ve.NewBlobFiles {
+		if _, ok := s.files[nf.FileNum]; ok {
+			return errors.AssertionFailedf("pebble: new blob file %d already exists", nf.FileNum)
+		}
+		cbf := &currentBlobFile{references: make(map[*TableMetadata]struct{})}
+		cbf.metadata = nf
+		s.files[nf.FileNum] = cbf
+		s.stats.Count++
+		s.stats.PhysicalSize += nf.Size
+		s.stats.ValueSize += nf.ValueSize
+	}
+
+	// Update references to blob files from new tables. Any referenced blob
+	// files should already exist in s.files.
+	newTables := make(map[base.FileNum]struct{})
+	for _, e := range ve.NewTables {
+		newTables[e.Meta.FileNum] = struct{}{}
+		for _, ref := range e.Meta.BlobReferences {
+			cbf, ok := s.files[ref.FileNum]
+			if !ok {
+				return errors.AssertionFailedf("pebble: referenced blob file %d not found", ref.FileNum)
+			}
+			cbf.references[e.Meta] = struct{}{}
+			cbf.referencedValueSize += ref.ValueSize
+			s.stats.ReferencedValueSize += ref.ValueSize
+			s.stats.ReferencesCount++
+		}
+	}
+
+	// Remove references to blob files from deleted tables. Any referenced blob
+	// files should already exist in s.files. If the removal of a reference
+	// causes the blob file's ref count to drop to zero, the blob file is a
+	// zombie. We update the version edit to record the blob file removal and
+	// remove it from the set.
+	for _, meta := range ve.DeletedTables {
+		for _, ref := range meta.BlobReferences {
+			cbf, ok := s.files[ref.FileNum]
+			if !ok {
+				return errors.AssertionFailedf("pebble: referenced blob file %d not found", ref.FileNum)
+			}
+			if invariants.Enabled {
+				if ref.ValueSize > cbf.referencedValueSize {
+					return errors.AssertionFailedf("pebble: referenced value size %d for blob file %d is greater than the referenced value size %d",
+						ref.ValueSize, cbf.metadata.FileNum, cbf.referencedValueSize)
+				}
+				if _, ok := cbf.references[meta]; !ok {
+					return errors.AssertionFailedf("pebble: deleted table %s's reference to blob file %d not known",
+						meta.FileNum, ref.FileNum)
+				}
+			}
+
+			// Decrement the stats for this reference.
+			cbf.referencedValueSize -= ref.ValueSize
+			s.stats.ReferencedValueSize -= ref.ValueSize
+			s.stats.ReferencesCount--
+			if _, ok := newTables[meta.FileNum]; ok {
+				// This table was added to a different level of the LSM in the
+				// same version edit. It's being moved. We can preserve the
+				// existing reference.  We still needed to reduce the counts
+				// above because we doubled it when we incremented stats on
+				// account of files in NewTables.
+				continue
+			}
+			// Remove the reference of this table to this blob file.
+			delete(cbf.references, meta)
+
+			// If there are no more references to the blob file, remove it from
+			// the set and add the removal of the blob file to the version edit.
+			if len(cbf.references) == 0 {
+				if cbf.referencedValueSize != 0 {
+					return errors.AssertionFailedf("pebble: referenced value size %d is non-zero for blob file %s with no refs",
+						cbf.referencedValueSize, cbf.metadata.FileNum)
+				}
+				if ve.DeletedBlobFiles == nil {
+					ve.DeletedBlobFiles = make(map[base.DiskFileNum]*BlobFileMetadata)
+				}
+				ve.DeletedBlobFiles[cbf.metadata.FileNum] = cbf.metadata
+				s.stats.Count--
+				s.stats.PhysicalSize -= cbf.metadata.Size
+				s.stats.ValueSize -= cbf.metadata.ValueSize
+				delete(s.files, cbf.metadata.FileNum)
+			}
+		}
+	}
+	return nil
+}
diff --git a/internal/manifest/version_edit.go b/internal/manifest/version_edit.go
index 1607a5c8..0a26c633 100644
--- a/internal/manifest/version_edit.go
+++ b/internal/manifest/version_edit.go
@@ -18,7 +18,6 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
-	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/strparse"
 	"github.com/cockroachdb/pebble/sstable"
 )
@@ -1068,7 +1067,7 @@ func (b *BulkVersionEdit) Accumulate(ve *VersionEdit) error {
 		// referenced blob file to BlobFiles.DeletedReferences. If the
 		// references are carried forward to new files (eg, during a compaction
 		// that decides not to rewrite the blob file), then we'll have the
-		// *BlobFileMetadata availabile when we process the NewTableEntry.
+		// *BlobFileMetadata available when we process the NewTableEntry.
 		for _, blobRef := range m.BlobReferences {
 			if b.BlobFiles.DeletedReferences == nil {
 				b.BlobFiles.DeletedReferences = make(map[base.DiskFileNum]*BlobFileMetadata)
@@ -1208,17 +1207,6 @@ func (b *BulkVersionEdit) Apply(curr *Version, readCompactionRate int64) (*Versi
 			// zero. The remove call will panic if this happens.
 			v.Levels[level].remove(f)
 			v.RangeKeyLevels[level].remove(f)
-
-			// The BlobFileMetadata maintains state about the file's references
-			// in the current Version of the LSM. Update this metadata for every
-			// deleted reference. The metadata pointer in BlobReference must
-			// already be populated when the version edit was accumulated.
-			for _, ref := range f.BlobReferences {
-				if ref.Metadata == nil {
-					return nil, errors.Newf("%s has a BlobFileReference with no metadata", ref.FileNum)
-				}
-				ref.Metadata.ActiveRefs.RemoveRef(ref.ValueSize)
-			}
 		}
 
 		addedTables := make([]*TableMetadata, 0, len(addedTablesMap))
@@ -1265,17 +1253,6 @@ func (b *BulkVersionEdit) Apply(curr *Version, readCompactionRate int64) (*Versi
 			if la == nil || base.InternalCompare(comparer.Compare, la.Largest, f.Largest) < 0 {
 				la = f
 			}
-
-			// The BlobFileMetadata maintains state about the file's references
-			// in the current Version of the LSM. Update this metadata for every
-			// new reference. The metadata pointer in BlobReference must already
-			// be populated when the version edit was accumulated.
-			for _, ref := range f.BlobReferences {
-				if ref.Metadata == nil {
-					return nil, errors.Newf("%s has a BlobFileReference with no metadata", ref.FileNum)
-				}
-				ref.Metadata.ActiveRefs.AddRef(ref.ValueSize)
-			}
 		}
 
 		if level == 0 {
@@ -1304,28 +1281,5 @@ func (b *BulkVersionEdit) Apply(curr *Version, readCompactionRate int64) (*Versi
 			}
 		}
 	}
-
-	// We maintain stats about active references in blob files and can infer
-	// when a blob file has become a 'zombie,' and is no longer referenced in
-	// the resulting version. However, we expect the caller to explicitly list
-	// all deleted blob files in the VersionEdit. In invariant builds, we loop
-	// back over all the deleted tables and their blob references. If any of the
-	// blob files now have zero active refs but weren't listed in the deleted
-	// blob files, the caller has broken the contract.
-	if invariants.Enabled {
-		for _, deletedLevel := range b.DeletedTables {
-			for _, dt := range deletedLevel {
-				for _, ref := range dt.BlobReferences {
-					if ref.Metadata.ActiveRefs.count != 0 {
-						continue
-					}
-					if _, ok := b.BlobFiles.Deleted[ref.FileNum]; !ok {
-						panic(errors.AssertionFailedf("blob file %s has no active refs, but was not deleted in the version edit", ref.FileNum))
-					}
-				}
-			}
-		}
-	}
-
 	return v, nil
 }
diff --git a/metrics.go b/metrics.go
index b1614043..d25d557a 100644
--- a/metrics.go
+++ b/metrics.go
@@ -315,29 +315,50 @@ type Metrics struct {
 	BlobFiles struct {
 		// The count of all live blob files.
 		LiveCount uint64
-		// The size of all live blob files.
+		// The physical file size of all live blob files.
 		LiveSize uint64
+		// ValueSize is the sum of the length of the uncompressed values in all
+		// live (referenced by some sstable(s) within the current version) blob
+		// files. ValueSize may be greater than LiveSize when compression is
+		// effective. ValueSize includes bytes in live blob files that are not
+		// actually reachable by any sstable key. If any value within the blob
+		// file is reachable by a key in a live sstable, then the entirety of
+		// the blob file's values are included within ValueSize.
+		ValueSize uint64
+		// ReferencedValueSize is the sum of the length of the uncompressed
+		// values (in all live blob files) that are still referenced by keys
+		// within live tables. Over the lifetime of a blob file, a blob file's
+		// references are removed as some compactions choose to write new blob
+		// files containing the same values or keys referencing the file's
+		// values are deleted. ReferencedValueSize accounts the volume of bytes
+		// that are actually reachable by some key in a live table.
+		//
+		// The difference between ValueSize and ReferencedValueSize is
+		// (uncompressed) space amplification that could be reclaimed if all
+		// blob files were rewritten, discarding values that are no longer
+		// referenced by any keys in any sstables within the current version.
+		ReferencedValueSize uint64
 		// The count of all obsolete blob files.
 		ObsoleteCount uint64
-		// The size of all obsolete blob files.
+		// The physical size of all obsolete blob files.
 		ObsoleteSize uint64
 		// The count of all zombie blob files.
 		ZombieCount uint64
-		// The size of all zombie blob files.
+		// The physical size of all zombie blob files.
 		ZombieSize uint64
 		// Local file sizes.
 		Local struct {
-			// LiveSize is the number of bytes in live blob files.
+			// LiveSize is the physical size of local live blob files.
 			LiveSize uint64
-			// LiveCount is the number of live blob files.
+			// LiveCount is the number of local live blob files.
 			LiveCount uint64
-			// ObsoleteSize is the number of bytes in obsolete blob files.
+			// ObsoleteSize is the physical size of local obsolete blob files.
 			ObsoleteSize uint64
-			// ObsoleteCount is the number of obsolete blob files.
+			// ObsoleteCount is the number of local obsolete blob files.
 			ObsoleteCount uint64
-			// ZombieSize is the number of bytes in zombie blob files.
+			// ZombieSize is the physical size of local zombie blob files.
 			ZombieSize uint64
-			// ZombieCount is the number of zombie blob files.
+			// ZombieCount is the number of local zombie blob files.
 			ZombieCount uint64
 		}
 	}
diff --git a/obsolete_files.go b/obsolete_files.go
index ed392531..b7b73fd3 100644
--- a/obsolete_files.go
+++ b/obsolete_files.go
@@ -286,7 +286,13 @@ func (cm *cleanupManager) deleteObsoleteObject(
 			FileNum: fileNum,
 			Err:     err,
 		})
-		// TODO(jackson): Add BlobFileDeleted event.
+	case base.FileTypeBlob:
+		cm.opts.EventListener.BlobFileDeleted(BlobFileDeleteInfo{
+			JobID:   int(jobID),
+			Path:    path,
+			FileNum: fileNum,
+			Err:     err,
+		})
 	}
 }
 
@@ -564,14 +570,10 @@ func (d *DB) deleteObsoleteFiles(jobID JobID) {
 	}
 }
 
-func (d *DB) maybeScheduleObsoleteTableDeletion() {
+func (d *DB) maybeScheduleObsoleteObjectDeletion() {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	d.maybeScheduleObsoleteTableDeletionLocked()
-}
-
-func (d *DB) maybeScheduleObsoleteTableDeletionLocked() {
-	if len(d.mu.versions.obsoleteTables) > 0 {
+	if len(d.mu.versions.obsoleteTables) > 0 || len(d.mu.versions.obsoleteBlobs) > 0 {
 		d.deleteObsoleteFiles(d.newJobIDLocked())
 	}
 }
diff --git a/options.go b/options.go
index fa24878e..02dfc595 100644
--- a/options.go
+++ b/options.go
@@ -742,6 +742,11 @@ type Options struct {
 		CompactionScheduler CompactionScheduler
 
 		UserKeyCategories UserKeyCategories
+
+		// ValueSeparationPolicy controls the policy for separating values into
+		// external blob files. If nil, value separation is disabled. The value
+		// separation policy is ignored if EnableColumnarBlocks() is false.
+		ValueSeparationPolicy func() ValueSeparationPolicy
 	}
 
 	// Filters is a map from filter policy name to filter policy. It is used for
@@ -1120,6 +1125,24 @@ type Options struct {
 	}
 }
 
+// ValueSeparationPolicy controls the policy for separating values into
+// external blob files.
+type ValueSeparationPolicy struct {
+	// Enabled controls whether value separation is enabled.
+	Enabled bool
+	// MinimumSize imposes a lower bound on the size of values that can be
+	// separated into a blob file. Values smaller than this are always written
+	// to the sstable (but may still be written to a value block within the
+	// sstable).
+	MinimumSize int
+	// MaxBlobReferenceDepth limits the number of potentially overlapping (in
+	// the keyspace) blob files that can be referenced by a single sstable. If a
+	// compaction may produce an output sstable referencing more than this many
+	// overlapping blob files, the compaction will instead rewrite referenced
+	// values into new blob files.
+	MaxBlobReferenceDepth manifest.BlobReferenceDepth
+}
+
 // WALFailoverOptions configures the WAL failover mechanics to use during
 // transient write unavailability on the primary WAL volume.
 type WALFailoverOptions struct {
@@ -1316,6 +1339,8 @@ func (o *Options) EnsureDefaults() {
 	if o.Experimental.MultiLevelCompactionHeuristic == nil {
 		o.Experimental.MultiLevelCompactionHeuristic = WriteAmpHeuristic{}
 	}
+	// TODO(jackson): Enable value separation by default once we have confidence
+	// in a default policy.
 
 	o.initMaps()
 }
@@ -1476,6 +1501,15 @@ func (o *Options) String() string {
 		fmt.Fprintln(&buf, "  disable_lazy_combined_iteration=true")
 	}
 
+	if o.Experimental.ValueSeparationPolicy != nil {
+		policy := o.Experimental.ValueSeparationPolicy()
+		fmt.Fprintln(&buf)
+		fmt.Fprintln(&buf, "[Value Separation]")
+		fmt.Fprintf(&buf, "  enabled=%t\n", policy.Enabled)
+		fmt.Fprintf(&buf, "  minimum_size=%d\n", policy.MinimumSize)
+		fmt.Fprintf(&buf, "  max_blob_reference_depth=%d\n", policy.MaxBlobReferenceDepth)
+	}
+
 	if o.WALFailover != nil {
 		unhealthyThreshold, _ := o.WALFailover.FailoverOptions.UnhealthyOperationLatencyThreshold()
 		fmt.Fprintf(&buf, "\n")
@@ -1603,6 +1637,8 @@ type ParseHooks struct {
 // options cannot be parsed into populated fields. For example, comparer and
 // merger.
 func (o *Options) Parse(s string, hooks *ParseHooks) error {
+	var valSepPolicy ValueSeparationPolicy
+	var valSepPolicyOk bool
 	visitKeyValue := func(i, j int, section, key, value string) error {
 		// WARNING: DO NOT remove entries from the switches below because doing so
 		// causes a key previously written to the OPTIONS file to be considered unknown,
@@ -1881,6 +1917,28 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 			}
 			return err
 
+		case section == "Value Separation":
+			valSepPolicyOk = true
+			var err error
+			switch key {
+			case "enabled":
+				valSepPolicy.Enabled, err = strconv.ParseBool(value)
+			case "minimum_size":
+				var minimumSize int
+				minimumSize, err = strconv.Atoi(value)
+				valSepPolicy.MinimumSize = minimumSize
+			case "max_blob_reference_depth":
+				var maxBlobReferenceDepth int
+				maxBlobReferenceDepth, err = strconv.Atoi(value)
+				valSepPolicy.MaxBlobReferenceDepth = manifest.BlobReferenceDepth(maxBlobReferenceDepth)
+			default:
+				if hooks != nil && hooks.SkipUnknown != nil && hooks.SkipUnknown(section+"."+key, value) {
+					return nil
+				}
+				return errors.Errorf("pebble: unknown option: %s.%s", errors.Safe(section), errors.Safe(key))
+			}
+			return err
+
 		case section == "WAL Failover":
 			if o.WALFailover == nil {
 				o.WALFailover = new(WALFailoverOptions)
@@ -1983,9 +2041,16 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 		}
 		return errors.Errorf("pebble: unknown section: %q", errors.Safe(section))
 	}
-	return parseOptions(s, parseOptionsFuncs{
+	err := parseOptions(s, parseOptionsFuncs{
 		visitKeyValue: visitKeyValue,
 	})
+	if err != nil {
+		return err
+	}
+	if valSepPolicyOk {
+		o.Experimental.ValueSeparationPolicy = func() ValueSeparationPolicy { return valSepPolicy }
+	}
+	return nil
 }
 
 // ErrMissingWALRecoveryDir is an error returned when a database is attempted to be
diff --git a/read_state.go b/read_state.go
index d3a78baa..d622bda8 100644
--- a/read_state.go
+++ b/read_state.go
@@ -44,8 +44,8 @@ func (s *readState) unref() {
 	}
 
 	// The last reference to the readState was released. Check to see if there
-	// are new obsolete tables to delete.
-	s.db.maybeScheduleObsoleteTableDeletion()
+	// are new obsolete objects to delete.
+	s.db.maybeScheduleObsoleteObjectDeletion()
 }
 
 // unrefLocked removes a reference to the readState. If this was the last
@@ -62,8 +62,8 @@ func (s *readState) unrefLocked() {
 		mem.readerUnrefLocked(true)
 	}
 
-	// In this code path, the caller is responsible for scheduling obsolete table
-	// deletion as necessary.
+	// In this code path, the caller is responsible for scheduling obsolete
+	// object deletions as necessary.
 }
 
 // loadReadState returns the current readState. The returned readState must be
diff --git a/sstable/colblk_writer.go b/sstable/colblk_writer.go
index 27084615..81b47af7 100644
--- a/sstable/colblk_writer.go
+++ b/sstable/colblk_writer.go
@@ -845,7 +845,7 @@ func (w *RawColumnWriter) flushBufferedIndexBlocks() (rootIndex block.Handle, er
 		if err != nil {
 			return rootIndex, err
 		}
-		w.props.IndexSize = rootIndex.Length + block.TrailerLen
+		w.props.IndexSize = uint64(len(w.indexBuffering.partitions[0].block))
 		w.props.NumDataBlocks = uint64(w.indexBuffering.partitions[0].nEntries)
 		w.props.IndexType = binarySearchIndex
 	default:
@@ -855,15 +855,17 @@ func (w *RawColumnWriter) flushBufferedIndexBlocks() (rootIndex block.Handle, er
 			if err != nil {
 				return block.Handle{}, err
 			}
-			w.props.IndexSize += bh.Length + block.TrailerLen
-			w.props.NumDataBlocks += uint64(w.indexBuffering.partitions[0].nEntries)
+			w.props.IndexSize += uint64(len(part.block))
+			w.props.NumDataBlocks += uint64(part.nEntries)
 			w.topLevelIndexBlock.AddBlockHandle(part.sep.UserKey, bh, part.properties)
 		}
-		rootIndex, err = w.layout.WriteIndexBlock(w.topLevelIndexBlock.Finish(w.topLevelIndexBlock.Rows()))
+		topLevelIndex := w.topLevelIndexBlock.Finish(w.topLevelIndexBlock.Rows())
+		rootIndex, err = w.layout.WriteIndexBlock(topLevelIndex)
 		if err != nil {
 			return block.Handle{}, err
 		}
-		w.props.IndexSize += rootIndex.Length + block.TrailerLen
+		w.props.TopLevelIndexSize = uint64(len(topLevelIndex))
+		w.props.IndexSize += uint64(len(topLevelIndex))
 		w.props.IndexType = twoLevelIndex
 		w.props.IndexPartitions = uint64(len(w.indexBuffering.partitions))
 	}
diff --git a/sstable/properties.go b/sstable/properties.go
index c243764d..7c9f94b7 100644
--- a/sstable/properties.go
+++ b/sstable/properties.go
@@ -153,7 +153,7 @@ type Properties struct {
 	FilterSize uint64 `prop:"rocksdb.filter.size"`
 	// Total number of index partitions if kTwoLevelIndexSearch is used.
 	IndexPartitions uint64 `prop:"rocksdb.index.partitions"`
-	// The size of index block.
+	// The size (uncompressed) of index block.
 	IndexSize uint64 `prop:"rocksdb.index.size"`
 	// The index type. TODO(peter): add a more detailed description.
 	IndexType uint32 `prop:"rocksdb.block.based.table.index.type"`
@@ -190,7 +190,7 @@ type Properties struct {
 	// The cumulative bytes of values in this table that were pinned by
 	// open snapshots. This value is comparable to RawValueSize.
 	SnapshotPinnedValueSize uint64 `prop:"pebble.raw.snapshot-pinned-values.size"`
-	// Size of the top-level index if kTwoLevelIndexSearch is used.
+	// Size (uncompressed) of the top-level index if kTwoLevelIndexSearch is used.
 	TopLevelIndexSize uint64 `prop:"rocksdb.top-level.index.size"`
 	// User collected properties. Currently, we only use them to store block
 	// properties aggregated at the table level.
diff --git a/value_separation.go b/value_separation.go
index da4e0242..5e10a9fc 100644
--- a/value_separation.go
+++ b/value_separation.go
@@ -6,6 +6,7 @@ package pebble
 
 import (
 	"cmp"
+	"maps"
 	"slices"
 	"time"
 
@@ -19,6 +20,120 @@ import (
 	"github.com/cockroachdb/pebble/sstable/blob"
 )
 
+var neverSeparateValues getValueSeparation = func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation {
+	return compact.NeverSeparateValues{}
+}
+
+// determineCompactionValueSeparation determines whether a compaction should
+// separate values into blob files. It returns a compact.ValueSeparation
+// implementation that should be used for the compaction.
+func (d *DB) determineCompactionValueSeparation(
+	jobID JobID, c *compaction, tableFormat sstable.TableFormat,
+) compact.ValueSeparation {
+	if tableFormat < sstable.TableFormatPebblev6 || d.FormatMajorVersion() < formatValueSeparation ||
+		d.opts.Experimental.ValueSeparationPolicy == nil {
+		return compact.NeverSeparateValues{}
+	}
+	policy := d.opts.Experimental.ValueSeparationPolicy()
+	if !policy.Enabled {
+		return compact.NeverSeparateValues{}
+	}
+
+	// We're allowed to write blob references. Determine whether we should carry
+	// forward existing blob references, or write new ones.
+	if writeBlobs, outputBlobReferenceDepth := shouldWriteBlobFiles(c, policy); !writeBlobs {
+		// This compaction should preserve existing blob references.
+		return &preserveBlobReferences{
+			inputBlobMetadatas:       uniqueInputBlobMetadatas(c.inputs),
+			outputBlobReferenceDepth: outputBlobReferenceDepth,
+		}
+	}
+
+	// This compaction should write values to new blob files.
+	return &writeNewBlobFiles{
+		comparer: d.opts.Comparer,
+		newBlobObject: func() (objstorage.Writable, objstorage.ObjectMetadata, error) {
+			return d.newCompactionOutputBlob(jobID, c)
+		},
+		shortAttrExtractor: d.opts.Experimental.ShortAttributeExtractor,
+		writerOpts:         d.opts.MakeBlobWriterOptions(c.outputLevel.level),
+		minimumSize:        policy.MinimumSize,
+		// TODO(jackson): Don't propagate these bounds if they don't
+		// overlap with the compaction's bounds.
+		requiredInPlaceValueBound: d.opts.Experimental.RequiredInPlaceValueBound,
+	}
+}
+
+// shouldWriteBlobFiles returns true if the compaction should write new blob
+// files. It also returns the maximum blob reference depth to assign to output
+// sstables (the actual value may be lower iff the output table references fewer
+// distinct blob files).
+func shouldWriteBlobFiles(
+	c *compaction, policy ValueSeparationPolicy,
+) (writeBlobs bool, referenceDepth manifest.BlobReferenceDepth) {
+	// Flushes will have no existing references to blob files and should write
+	// their values to new blob files.
+	if c.kind == compactionKindFlush {
+		return true, 1
+	}
+	inputReferenceDepth := compactionBlobReferenceDepth(c.inputs)
+	if inputReferenceDepth == 0 {
+		// None of the input sstables reference blob files. It may be the case
+		// that these sstables were created before value separation was enabled.
+		// We should try to write to new blob files.
+		return true, 1
+	}
+	// If the compaction's output blob reference depth would be greater than the
+	// configured max, we should rewrite the values into new blob files to
+	// restore locality.
+	if inputReferenceDepth > policy.MaxBlobReferenceDepth {
+		return true, 1
+	}
+	// Otherwise, we won't write any new blob files but will carry forward
+	// existing references.
+	return false, inputReferenceDepth
+}
+
+// compactionBlobReferenceDepth computes the blob reference depth for a
+// compaction. It's computed by finding the maximum blob reference depth of
+// input sstables in each level. These per-level depths are then summed to
+// produce a worst-case approximation of the blob reference locality of the
+// compaction's output sstables.
+//
+// The intuition is that as compactions combine files referencing distinct blob
+// files, outputted sstables begin to reference more and more distinct blob
+// files. In the worst case, these references are evenly distributed across the
+// keyspace and there is very little locality.
+func compactionBlobReferenceDepth(levels []compactionLevel) manifest.BlobReferenceDepth {
+	var depth manifest.BlobReferenceDepth
+	for _, level := range levels {
+		var levelDepth manifest.BlobReferenceDepth
+		for t := range level.files.All() {
+			levelDepth = max(levelDepth, t.BlobReferenceDepth)
+		}
+		depth += levelDepth
+	}
+	return depth
+}
+
+// uniqueInputBlobMetadatas returns a slice of all unique blob file metadata
+// objects referenced by tables in levels.
+func uniqueInputBlobMetadatas(levels []compactionLevel) []*manifest.BlobFileMetadata {
+	m := make(map[*manifest.BlobFileMetadata]struct{})
+	for _, level := range levels {
+		for t := range level.files.All() {
+			for _, ref := range t.BlobReferences {
+				m[ref.Metadata] = struct{}{}
+			}
+		}
+	}
+	metadatas := slices.Collect(maps.Keys(m))
+	slices.SortFunc(metadatas, func(a, b *manifest.BlobFileMetadata) int {
+		return cmp.Compare(a.FileNum, b.FileNum)
+	})
+	return metadatas
+}
+
 // writeNewBlobFiles implements the strategy and mechanics for separating values
 // into external blob files.
 type writeNewBlobFiles struct {
@@ -301,6 +416,7 @@ func (vs *preserveBlobReferences) findInputBlobMetadata(fn base.DiskFileNum) (in
 func (vs *preserveBlobReferences) FinishOutput() (compact.ValueSeparationMetadata, error) {
 	references := slices.Clone(vs.currReferences)
 	vs.currReferences = vs.currReferences[:0]
+	vs.totalValueSize = 0
 
 	referenceSize := uint64(0)
 	for _, ref := range references {
diff --git a/version_set.go b/version_set.go
index 80b0c5c8..25826b7d 100644
--- a/version_set.go
+++ b/version_set.go
@@ -69,7 +69,10 @@ type versionSet struct {
 	// Mutable fields.
 	versions    versionList
 	l0Organizer *manifest.L0Organizer
-	picker      compactionPicker
+	// blobFiles is the set of blob files referenced by the current version.
+	// blobFiles is protected by the manifest logLock (not vs.mu).
+	blobFiles manifest.CurrentBlobFileSet
+	picker    compactionPicker
 	// curCompactionConcurrency is updated whenever picker is updated.
 	// INVARIANT: >= 1.
 	curCompactionConcurrency atomic.Int32
@@ -177,6 +180,7 @@ func (vs *versionSet) create(
 	vs.init(dirname, provider, opts, marker, getFormatMajorVersion, mu)
 	emptyVersion := manifest.NewInitialVersion(opts.Comparer)
 	vs.append(emptyVersion)
+	vs.blobFiles.Init(nil)
 
 	vs.setCompactionPicker(
 		newCompactionPickerByScore(emptyVersion, vs.l0Organizer, &vs.virtualBackings, vs.opts, nil))
@@ -343,6 +347,7 @@ func (vs *versionSet) load(
 	}
 	vs.l0Organizer.PerformUpdate(vs.l0Organizer.PrepareUpdate(&bve, newVersion), newVersion)
 	vs.l0Organizer.InitCompactingFileInfo(nil /* in-progress compactions */)
+	vs.blobFiles.Init(&bve)
 	vs.append(newVersion)
 
 	for i := range vs.metrics.Levels {
@@ -578,7 +583,6 @@ func (vs *versionSet) UpdateVersionLocked(updateFn func() (versionUpdate, error)
 	// Note: this call populates ve.RemovedBackingTables.
 	zombieBackings, removedVirtualBackings, localTablesLiveDelta :=
 		getZombieTablesAndUpdateVirtualBackings(ve, &vs.virtualBackings, vs.provider)
-	zombieBlobs, blobLiveDelta, localBlobLiveDelta := getZombieBlobFilesAndComputeLocalMetrics(ve, vs.provider)
 
 	var l0Update manifest.L0PreparedUpdate
 	if err := func() error {
@@ -588,17 +592,13 @@ func (vs *versionSet) UpdateVersionLocked(updateFn func() (versionUpdate, error)
 		if vs.getFormatMajorVersion() < FormatVirtualSSTables && len(ve.CreatedBackingTables) > 0 {
 			return base.AssertionFailedf("MANIFEST cannot contain virtual sstable records due to format major version")
 		}
-		var bulkEdit bulkVersionEdit
-		err := bulkEdit.Accumulate(ve)
-		if err != nil {
-			return errors.Wrap(err, "MANIFEST accumulate failed")
-		}
-		newVersion, err = bulkEdit.Apply(currentVersion, vs.opts.Experimental.ReadCompactionRate)
-		if err != nil {
-			return errors.Wrap(err, "MANIFEST apply failed")
-		}
-		l0Update = vs.l0Organizer.PrepareUpdate(&bulkEdit, newVersion)
 
+		// Rotate the manifest if necessary. Rotating the manifest involves
+		// creating a new file and writing an initial version edit containing a
+		// snapshot of the current version. This initial version edit will
+		// reflect the Version prior to the pending version edit (`ve`). Once
+		// we've created the new manifest with the previous version state, we'll
+		// append the version edit `ve` to the tail of the new manifest.
 		if newManifestFileNum != 0 {
 			if err := vs.createManifest(vs.dirname, newManifestFileNum, minUnflushedLogNum, nextFileNum, newManifestVirtualBackings); err != nil {
 				vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
@@ -611,6 +611,27 @@ func (vs *versionSet) UpdateVersionLocked(updateFn func() (versionUpdate, error)
 			}
 		}
 
+		// Call ApplyAndUpdateVersionEdit before accumulating the version edit.
+		// If any blob files are no longer referenced, the version edit will be
+		// updated to explicitly record the deletion of the blob files. This can
+		// happen here because vs.blobFiles is protected by the manifest logLock
+		// (NOT vs.mu). We only read or write vs.blobFiles while holding the
+		// manifest lock.
+		if err := vs.blobFiles.ApplyAndUpdateVersionEdit(ve); err != nil {
+			return errors.Wrap(err, "MANIFEST blob files apply and update failed")
+		}
+
+		var bulkEdit bulkVersionEdit
+		err := bulkEdit.Accumulate(ve)
+		if err != nil {
+			return errors.Wrap(err, "MANIFEST accumulate failed")
+		}
+		newVersion, err = bulkEdit.Apply(currentVersion, vs.opts.Experimental.ReadCompactionRate)
+		if err != nil {
+			return errors.Wrap(err, "MANIFEST apply failed")
+		}
+		l0Update = vs.l0Organizer.PrepareUpdate(&bulkEdit, newVersion)
+
 		w, err := vs.manifest.Next()
 		if err != nil {
 			return errors.Wrap(err, "MANIFEST next record write failed")
@@ -661,6 +682,7 @@ func (vs *versionSet) UpdateVersionLocked(updateFn func() (versionUpdate, error)
 	// L0Sublevels.
 	inProgress := vu.InProgressCompactionsFn()
 
+	zombieBlobs, localBlobLiveDelta := getZombieBlobFilesAndComputeLocalMetrics(ve, vs.provider)
 	vs.l0Organizer.PerformUpdate(l0Update, newVersion)
 	vs.l0Organizer.InitCompactingFileInfo(inProgressL0Compactions(inProgress))
 
@@ -745,8 +767,6 @@ func (vs *versionSet) UpdateVersionLocked(updateFn func() (versionUpdate, error)
 	vs.metrics.Levels[0].Sublevels = int32(len(newVersion.L0SublevelFiles))
 	vs.metrics.Table.Local.LiveSize = uint64(int64(vs.metrics.Table.Local.LiveSize) + localTablesLiveDelta.size)
 	vs.metrics.Table.Local.LiveCount = uint64(int64(vs.metrics.Table.Local.LiveCount) + localTablesLiveDelta.count)
-	vs.metrics.BlobFiles.LiveSize = uint64(int64(vs.metrics.BlobFiles.LiveSize) + blobLiveDelta.size)
-	vs.metrics.BlobFiles.LiveCount = uint64(int64(vs.metrics.BlobFiles.LiveCount) + blobLiveDelta.count)
 	vs.metrics.BlobFiles.Local.LiveSize = uint64(int64(vs.metrics.BlobFiles.Local.LiveSize) + localBlobLiveDelta.size)
 	vs.metrics.BlobFiles.Local.LiveCount = uint64(int64(vs.metrics.BlobFiles.Local.LiveCount) + localBlobLiveDelta.count)
 
@@ -877,10 +897,8 @@ func getZombieTablesAndUpdateVirtualBackings(
 // locally.
 func getZombieBlobFilesAndComputeLocalMetrics(
 	ve *versionEdit, provider objstorage.Provider,
-) (zombieBlobFiles []objectInfo, liveDelta, localLiveDelta fileMetricDelta) {
+) (zombieBlobFiles []objectInfo, localLiveDelta fileMetricDelta) {
 	for _, b := range ve.NewBlobFiles {
-		liveDelta.count++
-		liveDelta.size += int64(b.Size)
 		if objstorage.IsLocalBlobFile(provider, b.FileNum) {
 			localLiveDelta.count++
 			localLiveDelta.size += int64(b.Size)
@@ -893,8 +911,6 @@ func getZombieBlobFilesAndComputeLocalMetrics(
 			localLiveDelta.count--
 			localLiveDelta.size -= int64(b.Size)
 		}
-		liveDelta.count--
-		liveDelta.size -= int64(b.Size)
 		zombieBlobFiles = append(zombieBlobFiles, objectInfo{
 			fileInfo: fileInfo{
 				FileNum:  b.FileNum,
@@ -903,7 +919,7 @@ func getZombieBlobFilesAndComputeLocalMetrics(
 			isLocal: isLocal,
 		})
 	}
-	return zombieBlobFiles, liveDelta, localLiveDelta
+	return zombieBlobFiles, localLiveDelta
 }
 
 // sizeIfLocal returns backing.Size if the backing is a local file, else 0.
@@ -999,41 +1015,31 @@ func (vs *versionSet) createManifest(
 	}
 	manifestWriter = record.NewWriter(manifestFile)
 
-	snapshot := versionEdit{
+	snapshot := manifest.VersionEdit{
 		ComparerName: vs.cmp.Name,
-	}
-	// TODO(jackson): Consider maintaining the set of all live blob files on the
-	// Version, rather than inferring it from the extant references on sstables.
-	// We may need this for maintaining the re-mapping of rewritten blob files.
-	blobFiles := make(map[base.DiskFileNum]*manifest.BlobFileMetadata)
+		// When creating a version snapshot for an existing DB, this snapshot
+		// VersionEdit will be immediately followed by another VersionEdit
+		// (being written in UpdateVersionLocked()). That VersionEdit always
+		// contains a LastSeqNum, so we don't need to include that in the
+		// snapshot.  But it does not necessarily include MinUnflushedLogNum,
+		// NextFileNum, so we initialize those using the corresponding fields in
+		// the versionSet (which came from the latest preceding VersionEdit that
+		// had those fields).
+		MinUnflushedLogNum:   minUnflushedLogNum,
+		NextFileNum:          nextFileNum,
+		CreatedBackingTables: virtualBackings,
+		NewBlobFiles:         vs.blobFiles.Metadatas(),
+	}
+	// Add all extant sstables in the current version.
 	for level, levelMetadata := range vs.currentVersion().Levels {
 		for meta := range levelMetadata.All() {
 			snapshot.NewTables = append(snapshot.NewTables, newTableEntry{
 				Level: level,
 				Meta:  meta,
 			})
-			if meta.BlobReferences != nil {
-				for _, blobRef := range meta.BlobReferences {
-					if _, ok := blobFiles[blobRef.FileNum]; !ok {
-						blobFiles[blobRef.FileNum] = blobRef.Metadata
-						snapshot.NewBlobFiles = append(snapshot.NewBlobFiles, blobRef.Metadata)
-					}
-				}
-			}
 		}
 	}
 
-	snapshot.CreatedBackingTables = virtualBackings
-
-	// When creating a version snapshot for an existing DB, this snapshot VersionEdit will be
-	// immediately followed by another VersionEdit (being written in UpdateVersionLocked()). That
-	// VersionEdit always contains a LastSeqNum, so we don't need to include that in the snapshot.
-	// But it does not necessarily include MinUnflushedLogNum, NextFileNum, so we initialize those
-	// using the corresponding fields in the versionSet (which came from the latest preceding
-	// VersionEdit that had those fields).
-	snapshot.MinUnflushedLogNum = minUnflushedLogNum
-	snapshot.NextFileNum = nextFileNum
-
 	w, err1 := manifestWriter.Next()
 	if err1 != nil {
 		return err1


Test patch:
diff --git a/compaction_picker_test.go b/compaction_picker_test.go
index 27d8a88c..d4fba3c5 100644
--- a/compaction_picker_test.go
+++ b/compaction_picker_test.go
@@ -580,7 +580,7 @@ func TestCompactionPickerL0(t *testing.T) {
 			var result strings.Builder
 			if pc != nil {
 				checkClone(t, pc)
-				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{})
+				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, neverSeparateValues)
 				fmt.Fprintf(&result, "L%d -> L%d\n", pc.startLevel.level, pc.outputLevel.level)
 				fmt.Fprintf(&result, "L%d: %s\n", pc.startLevel.level, fileNums(pc.startLevel.files))
 				if !pc.outputLevel.files.Empty() {
@@ -809,7 +809,7 @@ func TestCompactionPickerConcurrency(t *testing.T) {
 			var result strings.Builder
 			fmt.Fprintf(&result, "picker.getCompactionConcurrency: %d\n", allowedCompactions)
 			if pc != nil {
-				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{})
+				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, neverSeparateValues)
 				fmt.Fprintf(&result, "L%d -> L%d\n", pc.startLevel.level, pc.outputLevel.level)
 				fmt.Fprintf(&result, "L%d: %s\n", pc.startLevel.level, fileNums(pc.startLevel.files))
 				if !pc.outputLevel.files.Empty() {
diff --git a/compaction_test.go b/compaction_test.go
index c9b44c55..5a07e90f 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -530,7 +530,7 @@ func TestPickCompaction(t *testing.T) {
 		vs.picker = &tc.picker
 		pc, got := vs.picker.pickAuto(compactionEnv{diskAvailBytes: math.MaxUint64}), ""
 		if pc != nil {
-			c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{})
+			c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, neverSeparateValues)
 
 			gotStart := fileNums(c.startLevel.files)
 			gotML := ""
@@ -970,12 +970,7 @@ func TestCompaction(t *testing.T) {
 				if err := runCompactCmd(td, d); err != nil {
 					return err.Error()
 				}
-				d.mu.Lock()
-				s := d.mu.versions.currentVersion().String()
-				if verbose {
-					s = d.mu.versions.currentVersion().DebugString()
-				}
-				d.mu.Unlock()
+				s := describeLSM(d, verbose)
 				if td.HasArg("hide-file-num") {
 					re := regexp.MustCompile(`([0-9]*):\[`)
 					s = re.ReplaceAllString(s, "[")
@@ -1030,13 +1025,10 @@ func TestCompaction(t *testing.T) {
 				if err := d.Flush(); err != nil {
 					return err.Error()
 				}
-				d.mu.Lock()
-				s := d.mu.versions.currentVersion().String()
-				if verbose {
-					s = d.mu.versions.currentVersion().DebugString()
-				}
-				d.mu.Unlock()
-				return s
+				return describeLSM(d, verbose)
+
+			case "get":
+				return runGetCmd(t, td, d)
 
 			case "ingest":
 				if err := runIngestCmd(td, d, mem); err != nil {
@@ -1071,6 +1063,18 @@ func TestCompaction(t *testing.T) {
 				require.NoError(t, b.Commit(nil))
 				return fmt.Sprintf("wrote %d keys\n", count)
 
+			case "auto-compact":
+				d.mu.Lock()
+				prev := d.opts.DisableAutomaticCompactions
+				d.opts.DisableAutomaticCompactions = false
+				d.maybeScheduleCompaction()
+				for d.mu.compact.compactingCount > 0 {
+					d.mu.compact.cond.Wait()
+				}
+				d.opts.DisableAutomaticCompactions = prev
+				d.mu.Unlock()
+				return describeLSM(d, verbose)
+
 			case "async-compact":
 				var s string
 				ch := make(chan error, 1)
@@ -1225,7 +1229,12 @@ func TestCompaction(t *testing.T) {
 		},
 		"set_with_del_sstable_Pebblev6": {
 			minVersion: FormatTableFormatV6,
-			maxVersion: FormatTableFormatV6,
+			maxVersion: internalFormatNewest,
+		},
+		"value_separation": {
+			minVersion: formatValueSeparation,
+			maxVersion: formatValueSeparation,
+			verbose:    true,
 		},
 	}
 	datadriven.Walk(t, "testdata/compaction", func(t *testing.T, path string) {
@@ -1266,7 +1275,7 @@ func TestCompactionOutputLevel(t *testing.T) {
 				d.ScanArgs(t, "start", &start)
 				d.ScanArgs(t, "base", &base)
 				pc := newPickedCompaction(opts, version, l0Organizer, start, defaultOutputLevel(start, base), base)
-				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{})
+				c := newCompaction(pc, opts, time.Now(), nil /* provider */, noopGrantHandle{}, neverSeparateValues)
 				return fmt.Sprintf("output=%d\nmax-output-file-size=%d\n",
 					c.outputLevel.level, c.maxOutputFileSize)
 
diff --git a/data_test.go b/data_test.go
index d9a37f73..04452a08 100644
--- a/data_test.go
+++ b/data_test.go
@@ -906,6 +906,7 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 
 	var mem *memTable
 	var start, end *base.InternalKey
+	var blobDepth manifest.BlobReferenceDepth
 	ve := &versionEdit{}
 	level := -1
 
@@ -919,7 +920,7 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 			flushed:   make(chan struct{}),
 		}}
 		c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.l0Organizer,
-			d.mu.versions.picker.getBaseLevel(), toFlush, time.Now())
+			d.mu.versions.picker.getBaseLevel(), toFlush, time.Now(), d.determineCompactionValueSeparation)
 		if err != nil {
 			return err
 		}
@@ -949,6 +950,9 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 				f.Meta.LargestPointKey = *end
 				f.Meta.Largest = *end
 			}
+			if blobDepth > 0 {
+				f.Meta.BlobReferenceDepth = blobDepth
+			}
 			if largestSeqNum <= f.Meta.LargestSeqNum {
 				largestSeqNum = f.Meta.LargestSeqNum + 1
 			}
@@ -1035,6 +1039,7 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 				}
 				fields = fields[1:]
 				start, end = nil, nil
+				blobDepth = 0
 				boundFields := 0
 				for _, field := range fields {
 					toBreak := false
@@ -1047,6 +1052,13 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 						ikey := base.ParseInternalKey(strings.TrimPrefix(field, "end="))
 						end = &ikey
 						boundFields++
+					case strings.HasPrefix(field, "blob-depth="):
+						v, err := strconv.Atoi(strings.TrimPrefix(field, "blob-depth="))
+						if err != nil {
+							return nil, err
+						}
+						blobDepth = manifest.BlobReferenceDepth(v)
+						boundFields++
 					default:
 						toBreak = true
 					}
@@ -1486,12 +1498,25 @@ func runIngestExternalCmd(
 }
 
 func runLSMCmd(td *datadriven.TestData, d *DB) string {
+	return describeLSM(d, td.HasArg("verbose"))
+}
+
+func describeLSM(d *DB, verbose bool) string {
+	var buf bytes.Buffer
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	if td.HasArg("verbose") {
-		return d.mu.versions.currentVersion().DebugString()
+	if verbose {
+		buf.WriteString(d.mu.versions.currentVersion().DebugString())
+	} else {
+		buf.WriteString(d.mu.versions.currentVersion().String())
 	}
-	return d.mu.versions.currentVersion().String()
+	if blobFileMetas := d.mu.versions.blobFiles.Metadatas(); len(blobFileMetas) > 0 {
+		buf.WriteString("Blob files:\n")
+		for _, meta := range blobFileMetas {
+			fmt.Fprintf(&buf, "  %s: %d physical bytes, %d value bytes\n", meta.FileNum, meta.Size, meta.ValueSize)
+		}
+	}
+	return buf.String()
 }
 
 func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
@@ -1566,6 +1591,12 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 				injs[i] = inj
 			}
 			opts.FS = errorfs.Wrap(opts.FS, errorfs.Any(injs...))
+		case "l0-compaction-threshold":
+			v, err := strconv.Atoi(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			opts.L0CompactionThreshold = v
 		case "lbase-max-bytes":
 			lbaseMaxBytes, err := strconv.ParseInt(cmdArg.Vals[0], 10, 64)
 			if err != nil {
@@ -1587,6 +1618,15 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 			}
 		case "readonly":
 			opts.ReadOnly = true
+		case "required-in-place":
+			if len(cmdArg.Vals) != 2 {
+				return errors.New("required-in-place expects 2 arguments: <start-key> <end-key>")
+			}
+			start, end := cmdArg.Vals[0], cmdArg.Vals[1]
+			opts.Experimental.RequiredInPlaceValueBound = UserKeyPrefixBound{
+				Lower: []byte(start),
+				Upper: []byte(end),
+			}
 		case "target-file-sizes":
 			if len(opts.Levels) < len(cmdArg.Vals) {
 				opts.Levels = slices.Grow(opts.Levels, len(cmdArg.Vals)-len(opts.Levels))[0:len(cmdArg.Vals)]
@@ -1598,6 +1638,29 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 				}
 				opts.Levels[i].TargetFileSize = size
 			}
+		case "value-separation":
+			if len(cmdArg.Vals) != 3 {
+				return errors.New("value-separation-policy expects 3 arguments: (enabled, minimum-size, max-blob-reference-depth)")
+			}
+			var policy ValueSeparationPolicy
+			var err error
+			policy.Enabled, err = strconv.ParseBool(cmdArg.Vals[0])
+			if err != nil {
+				return err
+			}
+			policy.MinimumSize, err = strconv.Atoi(cmdArg.Vals[1])
+			if err != nil {
+				return err
+			}
+			var maxDepth int
+			maxDepth, err = strconv.Atoi(cmdArg.Vals[2])
+			if err != nil {
+				return err
+			}
+			policy.MaxBlobReferenceDepth = manifest.BlobReferenceDepth(maxDepth)
+			opts.Experimental.ValueSeparationPolicy = func() ValueSeparationPolicy {
+				return policy
+			}
 		case "wal-failover":
 			if v := cmdArg.Vals[0]; v == "off" || v == "disabled" {
 				opts.WALFailover = nil
diff --git a/format_major_version_test.go b/format_major_version_test.go
index ae72db59..43f4cf94 100644
--- a/format_major_version_test.go
+++ b/format_major_version_test.go
@@ -27,11 +27,13 @@ func TestFormatMajorVersionStableValues(t *testing.T) {
 	require.Equal(t, FormatFlushableIngestExcises, FormatMajorVersion(18))
 	require.Equal(t, FormatColumnarBlocks, FormatMajorVersion(19))
 	require.Equal(t, FormatWALSyncChunks, FormatMajorVersion(20))
+	require.Equal(t, FormatTableFormatV6, FormatMajorVersion(21))
+	require.Equal(t, formatValueSeparation, FormatMajorVersion(22))
 
 	// When we add a new version, we should add a check for the new version in
 	// addition to updating these expected values.
 	require.Equal(t, FormatNewest, FormatMajorVersion(21))
-	require.Equal(t, internalFormatNewest, FormatMajorVersion(21))
+	require.Equal(t, internalFormatNewest, FormatMajorVersion(22))
 }
 
 func TestFormatMajorVersion_MigrationDefined(t *testing.T) {
@@ -66,6 +68,8 @@ func TestRatchetFormat(t *testing.T) {
 	require.Equal(t, FormatWALSyncChunks, d.FormatMajorVersion())
 	require.NoError(t, d.RatchetFormatMajorVersion(FormatTableFormatV6))
 	require.Equal(t, FormatTableFormatV6, d.FormatMajorVersion())
+	require.NoError(t, d.RatchetFormatMajorVersion(formatValueSeparation))
+	require.Equal(t, formatValueSeparation, d.FormatMajorVersion())
 
 	require.NoError(t, d.Close())
 
@@ -225,6 +229,7 @@ func TestFormatMajorVersions_TableFormat(t *testing.T) {
 		FormatColumnarBlocks:             {sstable.TableFormatPebblev1, sstable.TableFormatPebblev5},
 		FormatWALSyncChunks:              {sstable.TableFormatPebblev1, sstable.TableFormatPebblev5},
 		FormatTableFormatV6:              {sstable.TableFormatPebblev1, sstable.TableFormatPebblev6},
+		formatValueSeparation:            {sstable.TableFormatPebblev1, sstable.TableFormatPebblev6},
 	}
 
 	// Valid versions.
@@ -280,6 +285,16 @@ func TestFormatMajorVersions_ColumnarBlocks(t *testing.T) {
 			colBlks: false,
 			want:    sstable.TableFormatPebblev4,
 		},
+		{
+			fmv:     formatValueSeparation,
+			colBlks: true,
+			want:    sstable.TableFormatPebblev6,
+		},
+		{
+			fmv:     formatValueSeparation,
+			colBlks: false,
+			want:    sstable.TableFormatPebblev4,
+		},
 		{
 			fmv:     FormatNewest,
 			colBlks: true,
diff --git a/internal/manifest/blob_metadata_test.go b/internal/manifest/blob_metadata_test.go
index 1f35dce6..efe8914b 100644
--- a/internal/manifest/blob_metadata_test.go
+++ b/internal/manifest/blob_metadata_test.go
@@ -5,8 +5,12 @@
 package manifest
 
 import (
+	"bytes"
+	"fmt"
 	"testing"
 
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/stretchr/testify/require"
 )
 
@@ -44,3 +48,61 @@ func TestBlobFileMetadata_ParseRoundTrip(t *testing.T) {
 		})
 	}
 }
+
+func TestCurrentBlobFileSet(t *testing.T) {
+	var (
+		buf        bytes.Buffer
+		set        CurrentBlobFileSet
+		tableMetas = make(map[base.FileNum]*TableMetadata)
+	)
+	parseAndFillVersionEdit := func(s string) *VersionEdit {
+		ve, err := ParseVersionEditDebug(s)
+		require.NoError(t, err)
+		for i, m := range ve.NewTables {
+			if existingMeta, ok := tableMetas[m.Meta.FileNum]; ok {
+				// Ensure pointer equality of the *TableMetadata.
+				// ParseVersionEditDebug will return a new *TableMetadata every
+				// time it decodes it.
+				ve.NewTables[i].Meta = existingMeta
+			} else {
+				tableMetas[m.Meta.FileNum] = m.Meta
+			}
+		}
+		for dte := range ve.DeletedTables {
+			ve.DeletedTables[dte] = tableMetas[dte.FileNum]
+		}
+		return ve
+	}
+
+	datadriven.RunTest(t, "testdata/current_blob_file_set", func(t *testing.T, d *datadriven.TestData) string {
+		buf.Reset()
+		switch d.Cmd {
+		case "init":
+			ve := parseAndFillVersionEdit(d.Input)
+			bve := &BulkVersionEdit{}
+			if err := bve.Accumulate(ve); err != nil {
+				return fmt.Sprintf("error accumulating version edit: %s", err)
+			}
+			set.Init(bve)
+			return set.Stats().String()
+		case "applyAndUpdateVersionEdit":
+			ve := parseAndFillVersionEdit(d.Input)
+			if err := set.ApplyAndUpdateVersionEdit(ve); err != nil {
+				return fmt.Sprintf("error applying and updating version edit: %s", err)
+			}
+			fmt.Fprintf(&buf, "modified version edit:\n%s", ve.DebugString(base.DefaultFormatter))
+			fmt.Fprintf(&buf, "current blob file set:\n%s", set.Stats().String())
+			return buf.String()
+		case "metadatas":
+			for _, m := range set.Metadatas() {
+				fmt.Fprintf(&buf, "%s\n", m)
+			}
+			return buf.String()
+		case "stats":
+			return set.Stats().String()
+		default:
+			t.Fatalf("unknown command: %s", d.Cmd)
+		}
+		return ""
+	})
+}
diff --git a/internal/manifest/testdata/current_blob_file_set b/internal/manifest/testdata/current_blob_file_set
new file mode 100644
index 00000000..8740b86c
--- /dev/null
+++ b/internal/manifest/testdata/current_blob_file_set
@@ -0,0 +1,75 @@
+init
+----
+Files:{Count: 0, Size: 0, ValueSize: 0}, References:{ValueSize: 0, Count: 0}
+
+# A version edit that does not contain blob files leaves the set unchanged.
+
+applyAndUpdateVersionEdit
+  add-table: L3 000010:[d#1,SET-e#1,SET]
+----
+modified version edit:
+  add-table:     L3 000010:[d#1,SET-e#1,SET] seqnums:[0-0] points:[d#1,SET-e#1,SET]
+current blob file set:
+Files:{Count: 0, Size: 0, ValueSize: 0}, References:{ValueSize: 0, Count: 0}
+
+# A version edit that adds a new blob file records the new file and reference.
+
+applyAndUpdateVersionEdit
+  add-table: L3 000011:[d#1,SET-e#1,SET] blobrefs:[(000012: 25935); depth:1]
+  add-blob-file: 000012 size:[20535 (20KB)] vals:[25935 (25KB)]
+----
+modified version edit:
+  add-table:     L3 000011:[d#1,SET-e#1,SET] seqnums:[0-0] points:[d#1,SET-e#1,SET] blobrefs:[(000012: 25935); depth:1]
+  add-blob-file: 000012 size:[20535 (20KB)] vals:[25935 (25KB)]
+current blob file set:
+Files:{Count: 1, Size: 20535, ValueSize: 25935}, References:{ValueSize: 25935, Count: 1}
+
+# A version edit that moves a referencing table from one level to another should
+# leave the blob set unchanged.
+
+applyAndUpdateVersionEdit
+  del-table: L3 000011
+  add-table: L4 000011:[d#1,SET-e#1,SET] blobrefs:[(000012: 25935); depth:1]
+----
+modified version edit:
+  del-table:     L3 000011
+  add-table:     L4 000011:[d#1,SET-e#1,SET] seqnums:[0-0] points:[d#1,SET-e#1,SET] blobrefs:[(000012: 25935); depth:1]
+current blob file set:
+Files:{Count: 1, Size: 20535, ValueSize: 25935}, References:{ValueSize: 25935, Count: 1}
+
+# A version edit that moves references from deleted tables to created tables
+# preseves the referenced blob file in the set, but updates the reference data.
+
+applyAndUpdateVersionEdit
+  del-table: L4 000011
+  add-table: L5 000013:[d#1,SET-e#1,SET] blobrefs:[(000012: 10); depth:2]
+  add-table: L5 000014:[f#1,SET-g#1,SET] blobrefs:[(000012: 15935); depth:2]
+----
+modified version edit:
+  del-table:     L4 000011
+  add-table:     L5 000013:[d#1,SET-e#1,SET] seqnums:[0-0] points:[d#1,SET-e#1,SET] blobrefs:[(000012: 10); depth:2]
+  add-table:     L5 000014:[f#1,SET-g#1,SET] seqnums:[0-0] points:[f#1,SET-g#1,SET] blobrefs:[(000012: 15935); depth:2]
+current blob file set:
+Files:{Count: 1, Size: 20535, ValueSize: 25935}, References:{ValueSize: 15945, Count: 2}
+
+# Remove one of the two references.
+
+applyAndUpdateVersionEdit
+  del-table: L5 000014
+----
+modified version edit:
+  del-table:     L5 000014
+current blob file set:
+Files:{Count: 1, Size: 20535, ValueSize: 25935}, References:{ValueSize: 10, Count: 1}
+
+# Remove the last reference. The version edit should be modified to include the
+# removal of the blob file.
+
+applyAndUpdateVersionEdit
+  del-table: L5 000013
+----
+modified version edit:
+  del-table:     L5 000013
+  del-blob-file: 000012
+current blob file set:
+Files:{Count: 0, Size: 0, ValueSize: 0}, References:{ValueSize: 0, Count: 0}
diff --git a/open_test.go b/open_test.go
index 6e0a8a03..5e020d48 100644
--- a/open_test.go
+++ b/open_test.go
@@ -333,7 +333,7 @@ func TestNewDBFilenames(t *testing.T) {
 			"LOCK",
 			"MANIFEST-000001",
 			"OPTIONS-000003",
-			"marker.format-version.000008.021",
+			"marker.format-version.000009.022",
 			"marker.manifest.000001.MANIFEST-000001",
 		},
 	}
diff --git a/options_test.go b/options_test.go
index e46513dc..cf17f655 100644
--- a/options_test.go
+++ b/options_test.go
@@ -300,6 +300,13 @@ func TestOptionsParse(t *testing.T) {
 			opts.Experimental.TombstoneDenseCompactionThreshold = 0.2
 			opts.Experimental.FileCacheShards = 500
 			opts.Experimental.SecondaryCacheSizeBytes = 1024
+			opts.Experimental.ValueSeparationPolicy = func() ValueSeparationPolicy {
+				return ValueSeparationPolicy{
+					Enabled:               true,
+					MinimumSize:           1024,
+					MaxBlobReferenceDepth: 10,
+				}
+			}
 			opts.EnsureDefaults()
 			str := opts.String()
 
diff --git a/sstable/testdata/columnar_writer/simple_binary b/sstable/testdata/columnar_writer/simple_binary
index 94043b24..5b1ccef0 100644
--- a/sstable/testdata/columnar_writer/simple_binary
+++ b/sstable/testdata/columnar_writer/simple_binary
@@ -103,7 +103,7 @@ sstable
  │    ├── 00600    rocksdb.raw.value.size (14)
  │    ├── restart points
  │    │    └── 00614 [restart 0]
- │    └── trailer [compression=none checksum=0xf75fa767]
+ │    └── trailer [compression=none checksum=0x38193e35]
  ├── meta-index  offset: 776  length: 33
  │    ├── 0000    rocksdb.properties block:149/622 [restart]
  │    ├── restart points
@@ -130,7 +130,7 @@ rocksdb.compression_options: window_bits=-14; level=32767; strategy=0; max_dict_
 rocksdb.comparator: pebble.internal.testkeys
 rocksdb.data.size: 93
 rocksdb.filter.size: 0
-rocksdb.index.size: 56
+rocksdb.index.size: 51
 rocksdb.block.based.table.index.type: 0
 pebble.colblk.schema: DefaultKeySchema(pebble.internal.testkeys,16)
 rocksdb.merge.operator: pebble.concatenate
@@ -250,7 +250,7 @@ sstable
  │    ├── 00568    rocksdb.raw.value.size (14)
  │    ├── restart points
  │    │    └── 00582 [restart 0]
- │    └── trailer [compression=none checksum=0x54772987]
+ │    └── trailer [compression=none checksum=0x1341bc55]
  ├── meta-index  offset: 772  length: 33
  │    ├── 0000    rocksdb.properties block:177/590 [restart]
  │    ├── restart points
@@ -645,7 +645,7 @@ sstable
  │    ├── 00570    rocksdb.raw.value.size (15)
  │    ├── restart points
  │    │    └── 00585 [restart 0]
- │    └── trailer [compression=none checksum=0x98e2c466]
+ │    └── trailer [compression=none checksum=0xa3e92c5a]
  ├── meta-index  offset: 1295  length: 33
  │    ├── 0000    rocksdb.properties block:697/593 [restart]
  │    ├── restart points
@@ -675,7 +675,7 @@ rocksdb.compression_options: window_bits=-14; level=32767; strategy=0; max_dict_
 rocksdb.comparator: pebble.internal.testkeys
 rocksdb.data.size: 582
 rocksdb.filter.size: 0
-rocksdb.index.size: 115
+rocksdb.index.size: 110
 rocksdb.block.based.table.index.type: 0
 pebble.colblk.schema: DefaultKeySchema(pebble.internal.testkeys,16)
 rocksdb.merge.operator: pebble.concatenate
@@ -1036,7 +1036,7 @@ sstable
  │    ├── 00570    rocksdb.raw.value.size (15)
  │    ├── restart points
  │    │    └── 00585 [restart 0]
- │    └── trailer [compression=none checksum=0x98e2c466]
+ │    └── trailer [compression=none checksum=0xa3e92c5a]
  ├── meta-index  offset: 1295  length: 33
  │    ├── 0000    rocksdb.properties block:697/593 [restart]
  │    ├── restart points
@@ -1133,7 +1133,7 @@ sstable
  │    ├── 00558    rocksdb.raw.value.size (14)
  │    ├── restart points
  │    │    └── 00572 [restart 0]
- │    └── trailer [compression=none checksum=0xd4a2d9f6]
+ │    └── trailer [compression=none checksum=0x50f07e7e]
  ├── meta-index  offset: 674  length: 59
  │    ├── 0000    rocksdb.properties block:89/580 [restart]
  │    ├── 0024    rocksdb.range_del2 block:33/51 [restart]
@@ -1251,7 +1251,7 @@ sstable
  │    ├── 00640    rocksdb.raw.value.size (14)
  │    ├── restart points
  │    │    └── 00654 [restart 0]
- │    └── trailer [compression=none checksum=0xbdc87030]
+ │    └── trailer [compression=none checksum=0x29bae4b8]
  ├── meta-index  offset: 773  length: 57
  │    ├── 0000    pebble.range_key block:33/68 [restart]
  │    ├── 0021    rocksdb.properties block:106/662 [restart]
diff --git a/sstable/testdata/writer_blob_value_handles b/sstable/testdata/writer_blob_value_handles
index d8d57a3c..f250d577 100644
--- a/sstable/testdata/writer_blob_value_handles
+++ b/sstable/testdata/writer_blob_value_handles
@@ -132,7 +132,7 @@ sstable
  │    ├── 00589    rocksdb.property.collectors (41)
  │    ├── 00630    rocksdb.raw.key.size (21)
  │    ├── 00651    rocksdb.raw.value.size (24)
- │    └── trailer [compression=snappy checksum=0xf0c5768d]
+ │    └── trailer [compression=snappy checksum=0x277c72db]
  ├── meta-index  offset: 767  length: 46
  │    ├── 0000    rocksdb.properties block:200/562
  │    │   
diff --git a/sstable/testdata/writer_v5 b/sstable/testdata/writer_v5
index 2df7ce34..b202f283 100644
--- a/sstable/testdata/writer_v5
+++ b/sstable/testdata/writer_v5
@@ -101,7 +101,7 @@ rocksdb.compression_options: window_bits=-14; level=32767; strategy=0; max_dict_
 rocksdb.comparator: pebble.internal.testkeys
 rocksdb.data.size: 0
 rocksdb.filter.size: 0
-rocksdb.index.size: 33
+rocksdb.index.size: 28
 rocksdb.block.based.table.index.type: 0
 pebble.colblk.schema: DefaultKeySchema(pebble.internal.testkeys,16)
 rocksdb.merge.operator: pebble.concatenate
@@ -317,7 +317,7 @@ rocksdb.compression_options: window_bits=-14; level=32767; strategy=0; max_dict_
 rocksdb.comparator: pebble.internal.testkeys
 rocksdb.data.size: 0
 rocksdb.filter.size: 0
-rocksdb.index.size: 33
+rocksdb.index.size: 28
 rocksdb.block.based.table.index.type: 0
 pebble.colblk.schema: DefaultKeySchema(pebble.internal.testkeys,16)
 rocksdb.merge.operator: pebble.concatenate
diff --git a/sstable/testdata/writer_v6 b/sstable/testdata/writer_v6
index 356cff99..bd914124 100644
--- a/sstable/testdata/writer_v6
+++ b/sstable/testdata/writer_v6
@@ -101,7 +101,7 @@ rocksdb.compression_options: window_bits=-14; level=32767; strategy=0; max_dict_
 rocksdb.comparator: pebble.internal.testkeys
 rocksdb.data.size: 0
 rocksdb.filter.size: 0
-rocksdb.index.size: 33
+rocksdb.index.size: 28
 rocksdb.block.based.table.index.type: 0
 pebble.colblk.schema: DefaultKeySchema(pebble.internal.testkeys,16)
 rocksdb.merge.operator: pebble.concatenate
@@ -239,6 +239,29 @@ sstable
  ├── meta-index  offset: 974  length: 46
  └── footer  offset: 1025  length: 57
 
+props
+----
+rocksdb.num.entries: 3
+rocksdb.raw.key.size: 27
+rocksdb.raw.value.size: 3
+rocksdb.deleted.keys: 0
+rocksdb.num.range-deletions: 0
+rocksdb.num.data.blocks: 3
+rocksdb.compression: Snappy
+rocksdb.compression_options: window_bits=-14; level=32767; strategy=0; max_dict_bytes=0; zstd_max_train_bytes=0; enabled=0; 
+rocksdb.comparator: pebble.internal.testkeys
+rocksdb.data.size: 237
+rocksdb.filter.size: 0
+rocksdb.index.partitions: 3
+rocksdb.index.size: 158
+rocksdb.block.based.table.index.type: 2
+pebble.colblk.schema: DefaultKeySchema(pebble.internal.testkeys,16)
+rocksdb.merge.operator: pebble.concatenate
+rocksdb.merge.operands: 0
+rocksdb.property.collectors: [obsolete-key]
+rocksdb.top-level.index.size: 48
+obsolete-key: hex:00
+
 # Exercise the non-Reader layout-decoding codepath.
 
 decode-layout
@@ -317,7 +340,7 @@ rocksdb.compression_options: window_bits=-14; level=32767; strategy=0; max_dict_
 rocksdb.comparator: pebble.internal.testkeys
 rocksdb.data.size: 0
 rocksdb.filter.size: 0
-rocksdb.index.size: 33
+rocksdb.index.size: 28
 rocksdb.block.based.table.index.type: 0
 pebble.colblk.schema: DefaultKeySchema(pebble.internal.testkeys,16)
 rocksdb.merge.operator: pebble.concatenate
diff --git a/sstable/testdata/writer_value_blocks b/sstable/testdata/writer_value_blocks
index 646b1fff..2c13e30f 100644
--- a/sstable/testdata/writer_value_blocks
+++ b/sstable/testdata/writer_value_blocks
@@ -669,7 +669,7 @@ sstable
  │    ├── 00578    rocksdb.top-level.index.size (24)
  │    ├── restart points
  │    │    └── 00602 [restart 0]
- │    └── trailer [compression=none checksum=0x307b671a]
+ │    └── trailer [compression=none checksum=0xc70b061b]
  ├── meta-index  offset: 1308  length: 64
  │    ├── 0000    pebble.value_index block:680/8 value-blocks-index-lengths: 1(num), 2(offset), 1(length) [restart]
  │    ├── 0027    rocksdb.properties block:693/610 [restart]
@@ -961,7 +961,7 @@ sstable
  │    ├── 00518    rocksdb.raw.value.size (14)
  │    ├── restart points
  │    │    └── 00532 [restart 0]
- │    └── trailer [compression=none checksum=0x62079e3b]
+ │    └── trailer [compression=none checksum=0x9b925786]
  ├── meta-index  offset: 691  length: 33
  │    ├── 0000    rocksdb.properties block:146/540 [restart]
  │    ├── restart points
@@ -1103,7 +1103,7 @@ sstable
  │    ├── 00556    rocksdb.property.collectors (41)
  │    ├── 00597    rocksdb.raw.key.size (21)
  │    ├── 00618    rocksdb.raw.value.size (23)
- │    └── trailer [compression=snappy checksum=0x6a5dbad6]
+ │    └── trailer [compression=snappy checksum=0x5d7f7403]
  ├── meta-index  offset: 676  length: 46
  │    ├── 0000    rocksdb.properties block:146/525
  │    │   
diff --git a/testdata/checkpoint b/testdata/checkpoint
index bf236fbb..cd50b9a6 100644
--- a/testdata/checkpoint
+++ b/testdata/checkpoint
@@ -51,6 +51,10 @@ create: db/marker.format-version.000008.021
 close: db/marker.format-version.000008.021
 remove: db/marker.format-version.000007.020
 sync: db
+create: db/marker.format-version.000009.022
+close: db/marker.format-version.000009.022
+remove: db/marker.format-version.000008.021
+sync: db
 create: db/temporary.000003.dbtmp
 sync: db/temporary.000003.dbtmp
 close: db/temporary.000003.dbtmp
@@ -117,9 +121,9 @@ sync-data: checkpoints/checkpoint1/OPTIONS-000003
 close: checkpoints/checkpoint1/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint1
-create: checkpoints/checkpoint1/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint1/marker.format-version.000001.021
-close: checkpoints/checkpoint1/marker.format-version.000001.021
+create: checkpoints/checkpoint1/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint1/marker.format-version.000001.022
+close: checkpoints/checkpoint1/marker.format-version.000001.022
 sync: checkpoints/checkpoint1
 close: checkpoints/checkpoint1
 link: db/000005.sst -> checkpoints/checkpoint1/000005.sst
@@ -161,9 +165,9 @@ sync-data: checkpoints/checkpoint2/OPTIONS-000003
 close: checkpoints/checkpoint2/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint2
-create: checkpoints/checkpoint2/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint2/marker.format-version.000001.021
-close: checkpoints/checkpoint2/marker.format-version.000001.021
+create: checkpoints/checkpoint2/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint2/marker.format-version.000001.022
+close: checkpoints/checkpoint2/marker.format-version.000001.022
 sync: checkpoints/checkpoint2
 close: checkpoints/checkpoint2
 link: db/000007.sst -> checkpoints/checkpoint2/000007.sst
@@ -200,9 +204,9 @@ sync-data: checkpoints/checkpoint3/OPTIONS-000003
 close: checkpoints/checkpoint3/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint3
-create: checkpoints/checkpoint3/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint3/marker.format-version.000001.021
-close: checkpoints/checkpoint3/marker.format-version.000001.021
+create: checkpoints/checkpoint3/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint3/marker.format-version.000001.022
+close: checkpoints/checkpoint3/marker.format-version.000001.022
 sync: checkpoints/checkpoint3
 close: checkpoints/checkpoint3
 link: db/000005.sst -> checkpoints/checkpoint3/000005.sst
@@ -286,7 +290,7 @@ list db
 LOCK
 MANIFEST-000001
 OPTIONS-000003
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 list checkpoints/checkpoint1
@@ -296,7 +300,7 @@ list checkpoints/checkpoint1
 000007.sst
 MANIFEST-000001
 OPTIONS-000003
-marker.format-version.000001.021
+marker.format-version.000001.022
 marker.manifest.000001.MANIFEST-000001
 
 open checkpoints/checkpoint1 readonly
@@ -363,7 +367,7 @@ list checkpoints/checkpoint2
 000007.sst
 MANIFEST-000001
 OPTIONS-000003
-marker.format-version.000001.021
+marker.format-version.000001.022
 marker.manifest.000001.MANIFEST-000001
 
 open checkpoints/checkpoint2 readonly
@@ -405,7 +409,7 @@ list checkpoints/checkpoint3
 000007.sst
 MANIFEST-000001
 OPTIONS-000003
-marker.format-version.000001.021
+marker.format-version.000001.022
 marker.manifest.000001.MANIFEST-000001
 
 open checkpoints/checkpoint3 readonly
@@ -524,9 +528,9 @@ sync-data: checkpoints/checkpoint4/OPTIONS-000003
 close: checkpoints/checkpoint4/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint4
-create: checkpoints/checkpoint4/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint4/marker.format-version.000001.021
-close: checkpoints/checkpoint4/marker.format-version.000001.021
+create: checkpoints/checkpoint4/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint4/marker.format-version.000001.022
+close: checkpoints/checkpoint4/marker.format-version.000001.022
 sync: checkpoints/checkpoint4
 close: checkpoints/checkpoint4
 link: db/000010.sst -> checkpoints/checkpoint4/000010.sst
@@ -614,7 +618,7 @@ list db
 LOCK
 MANIFEST-000001
 OPTIONS-000003
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 
@@ -633,9 +637,9 @@ sync-data: checkpoints/checkpoint5/OPTIONS-000003
 close: checkpoints/checkpoint5/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint5
-create: checkpoints/checkpoint5/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint5/marker.format-version.000001.021
-close: checkpoints/checkpoint5/marker.format-version.000001.021
+create: checkpoints/checkpoint5/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint5/marker.format-version.000001.022
+close: checkpoints/checkpoint5/marker.format-version.000001.022
 sync: checkpoints/checkpoint5
 close: checkpoints/checkpoint5
 link: db/000010.sst -> checkpoints/checkpoint5/000010.sst
@@ -735,9 +739,9 @@ sync-data: checkpoints/checkpoint6/OPTIONS-000003
 close: checkpoints/checkpoint6/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint6
-create: checkpoints/checkpoint6/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint6/marker.format-version.000001.021
-close: checkpoints/checkpoint6/marker.format-version.000001.021
+create: checkpoints/checkpoint6/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint6/marker.format-version.000001.022
+close: checkpoints/checkpoint6/marker.format-version.000001.022
 sync: checkpoints/checkpoint6
 close: checkpoints/checkpoint6
 link: db/000011.sst -> checkpoints/checkpoint6/000011.sst
diff --git a/testdata/checkpoint_shared b/testdata/checkpoint_shared
index 0685beac..2d2d9004 100644
--- a/testdata/checkpoint_shared
+++ b/testdata/checkpoint_shared
@@ -39,6 +39,10 @@ create: db/marker.format-version.000005.021
 close: db/marker.format-version.000005.021
 remove: db/marker.format-version.000004.020
 sync: db
+create: db/marker.format-version.000006.022
+close: db/marker.format-version.000006.022
+remove: db/marker.format-version.000005.021
+sync: db
 create: db/temporary.000003.dbtmp
 sync: db/temporary.000003.dbtmp
 close: db/temporary.000003.dbtmp
@@ -105,9 +109,9 @@ sync-data: checkpoints/checkpoint1/OPTIONS-000003
 close: checkpoints/checkpoint1/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint1
-create: checkpoints/checkpoint1/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint1/marker.format-version.000001.021
-close: checkpoints/checkpoint1/marker.format-version.000001.021
+create: checkpoints/checkpoint1/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint1/marker.format-version.000001.022
+close: checkpoints/checkpoint1/marker.format-version.000001.022
 sync: checkpoints/checkpoint1
 close: checkpoints/checkpoint1
 open: db/MANIFEST-000001 (options: *vfs.sequentialReadsOption)
@@ -158,9 +162,9 @@ sync-data: checkpoints/checkpoint2/OPTIONS-000003
 close: checkpoints/checkpoint2/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint2
-create: checkpoints/checkpoint2/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint2/marker.format-version.000001.021
-close: checkpoints/checkpoint2/marker.format-version.000001.021
+create: checkpoints/checkpoint2/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint2/marker.format-version.000001.022
+close: checkpoints/checkpoint2/marker.format-version.000001.022
 sync: checkpoints/checkpoint2
 close: checkpoints/checkpoint2
 open: db/MANIFEST-000001 (options: *vfs.sequentialReadsOption)
@@ -207,9 +211,9 @@ sync-data: checkpoints/checkpoint3/OPTIONS-000003
 close: checkpoints/checkpoint3/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoints/checkpoint3
-create: checkpoints/checkpoint3/marker.format-version.000001.021
-sync-data: checkpoints/checkpoint3/marker.format-version.000001.021
-close: checkpoints/checkpoint3/marker.format-version.000001.021
+create: checkpoints/checkpoint3/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint3/marker.format-version.000001.022
+close: checkpoints/checkpoint3/marker.format-version.000001.022
 sync: checkpoints/checkpoint3
 close: checkpoints/checkpoint3
 open: db/MANIFEST-000001 (options: *vfs.sequentialReadsOption)
@@ -266,7 +270,7 @@ LOCK
 MANIFEST-000001
 OPTIONS-000003
 REMOTE-OBJ-CATALOG-000001
-marker.format-version.000005.021
+marker.format-version.000006.022
 marker.manifest.000001.MANIFEST-000001
 marker.remote-obj-catalog.000001.REMOTE-OBJ-CATALOG-000001
 
@@ -276,7 +280,7 @@ list checkpoints/checkpoint1
 MANIFEST-000001
 OPTIONS-000003
 REMOTE-OBJ-CATALOG-000001
-marker.format-version.000001.021
+marker.format-version.000001.022
 marker.manifest.000001.MANIFEST-000001
 marker.remote-obj-catalog.000001.REMOTE-OBJ-CATALOG-000001
 
@@ -328,7 +332,7 @@ list checkpoints/checkpoint2
 MANIFEST-000001
 OPTIONS-000003
 REMOTE-OBJ-CATALOG-000001
-marker.format-version.000001.021
+marker.format-version.000001.022
 marker.manifest.000001.MANIFEST-000001
 marker.remote-obj-catalog.000001.REMOTE-OBJ-CATALOG-000001
 
diff --git a/testdata/compaction/value_separation b/testdata/compaction/value_separation
new file mode 100644
index 00000000..5f1ac7aa
--- /dev/null
+++ b/testdata/compaction/value_separation
@@ -0,0 +1,285 @@
+# Test a simple sequence of flushes and compactions where all values are
+# separated.
+
+define value-separation=(true, 0, 3)
+----
+
+batch
+set a 1
+set b 2
+----
+
+compact a-b
+----
+L6:
+  000005:[a#10,SET-b#11,SET] seqnums:[10-11] points:[a#10,SET-b#11,SET] size:766 blobrefs:[(000006: 2); depth:1]
+Blob files:
+  000006: 48 physical bytes, 2 value bytes
+
+batch
+set c 3
+set d 4
+----
+
+compact c-d
+----
+L6:
+  000005:[a#10,SET-b#11,SET] seqnums:[10-11] points:[a#10,SET-b#11,SET] size:766 blobrefs:[(000006: 2); depth:1]
+  000008:[c#12,SET-d#13,SET] seqnums:[12-13] points:[c#12,SET-d#13,SET] size:766 blobrefs:[(000009: 2); depth:1]
+Blob files:
+  000006: 48 physical bytes, 2 value bytes
+  000009: 48 physical bytes, 2 value bytes
+
+batch
+set b 5
+set c 6
+----
+
+compact a-d
+----
+L6:
+  000013:[a#0,SET-d#0,SET] seqnums:[0-0] points:[a#0,SET-d#0,SET] size:793 blobrefs:[(000006: 1), (000012: 2), (000009: 1); depth:2]
+Blob files:
+  000006: 48 physical bytes, 2 value bytes
+  000009: 48 physical bytes, 2 value bytes
+  000012: 48 physical bytes, 2 value bytes
+
+batch
+del-range a e
+----
+
+compact a-d
+----
+
+# Set up a scenario where there's a L6 sstable with a blob reference depth of 3,
+# and the value separation policy is configured to limit the blob reference
+# depth to 3.
+
+define verbose value-separation=(true, 3, 3)
+L6 blob-depth=3
+  a.SET.0:a
+  b.SET.0:blob{fileNum=100002 value=bar}
+  f.SET.0:blob{fileNum=100003 value=foo}
+  k.SET.0:k
+  z.SET.0:blob{fileNum=100004 value=zoo}
+----
+L6:
+  000004:[a#0,SET-z#0,SET] seqnums:[0-0] points:[a#0,SET-z#0,SET] size:795 blobrefs:[(100002: 3), (100003: 3), (100004: 3); depth:3]
+
+batch
+set d hello
+set e world
+----
+
+# Flush should write flushed values to a new blob file.
+
+flush
+----
+L0.0:
+  000006:[d#10,SET-e#11,SET] seqnums:[10-11] points:[d#10,SET-e#11,SET] size:765 blobrefs:[(000007: 10); depth:1]
+L6:
+  000004:[a#0,SET-z#0,SET] seqnums:[0-0] points:[a#0,SET-z#0,SET] size:795 blobrefs:[(100002: 3), (100003: 3), (100004: 3); depth:3]
+Blob files:
+  000007: 56 physical bytes, 10 value bytes
+  100002: 49 physical bytes, 3 value bytes
+  100003: 49 physical bytes, 3 value bytes
+  100004: 49 physical bytes, 3 value bytes
+
+# Compacting these two sstables should result in writing the values to a new
+# blob file and the removal of the no longer referenced blob files.
+
+compact a-z
+----
+L6:
+  000008:[a#0,SET-z#0,SET] seqnums:[0-0] points:[a#0,SET-z#0,SET] size:815 blobrefs:[(000009: 19); depth:1]
+Blob files:
+  000009: 65 physical bytes, 19 value bytes
+
+# Ensure we can read the separated values by iterating over the database.
+
+iter
+first
+next
+next
+next
+next
+next
+next
+----
+a: (a, .)
+b: (bar, .)
+d: (hello, .)
+e: (world, .)
+f: (foo, .)
+k: (k, .)
+z: (zoo, .)
+
+# Set the minimum size for a separated value to 5.
+
+define value-separation=(true, 5, 3)
+----
+
+batch
+set bar bar
+set foo foo
+set fuzz fuzz
+set yaya yaya
+----
+
+# The flush should not write a blob file because none of the keys have a
+# sufficiently long value to be separated.
+
+flush
+----
+L0.0:
+  000005:[bar#10,SET-yaya#13,SET] seqnums:[10-13] points:[bar#10,SET-yaya#13,SET] size:768
+
+batch
+set a a
+set b b
+set h hello
+set w world
+----
+
+# This flush *should* write a blob file, containing 2 values: "hello" and
+# "world" totalling 10 bytes of logical values.
+
+flush
+----
+L0.1:
+  000007:[a#14,SET-w#17,SET] seqnums:[14-17] points:[a#14,SET-w#17,SET] size:811 blobrefs:[(000008: 10); depth:1]
+L0.0:
+  000005:[bar#10,SET-yaya#13,SET] seqnums:[10-13] points:[bar#10,SET-yaya#13,SET] size:768
+Blob files:
+  000008: 56 physical bytes, 10 value bytes
+
+get
+a
+b
+h
+w
+----
+a:a
+b:b
+h:hello
+w:world
+
+# Configure the database to require keys in the range [a,m) to be in-place.
+
+define required-in-place=(a,m) value-separation=(true,1,3)
+----
+
+batch
+set a apple
+set b banana
+set c coconut
+set d dragonfruit
+set m mango
+----
+
+# The flush should write a blob file, but only "mango" should be separated. This
+# should be reflected in the 5-byte value bytes of the blob file and the table's
+# blob reference value size.
+
+flush
+----
+L0.0:
+  000005:[a#10,SET-m#14,SET] seqnums:[10-14] points:[a#10,SET-m#14,SET] size:820 blobrefs:[(000006: 5); depth:1]
+Blob files:
+  000006: 51 physical bytes, 5 value bytes
+
+
+define value-separation=(true,5,5) l0-compaction-threshold=1
+----
+
+# Test writing a non-trivial amount of data. With a key length of 4, we'll write
+# 475254 keys each with a 64-byte value, totalling ~30MB of value data.
+
+populate keylen=4 timestamps=(1) vallen=64
+----
+wrote 475254 keys
+
+# Flush the memtable. The resulting L0 sstables should be relatively small, but
+# when their sizes are summed with their corresponding blob files, the sum
+# should be around the target file size of 2MB.
+
+flush
+----
+L0.0:
+  000005:[a@1#10,SET-blof@1#26408,SET] seqnums:[10-26408] points:[a@1#10,SET-blof@1#26408,SET] size:402673 blobrefs:[(000006: 1689536); depth:1]
+  000007:[blog@1#26409,SET-cxcf@1#52799,SET] seqnums:[26409-52799] points:[blog@1#26409,SET-cxcf@1#52799,SET] size:405756 blobrefs:[(000008: 1689024); depth:1]
+  000009:[cxcg@1#52800,SET-einq@1#79120,SET] seqnums:[52800-79120] points:[cxcg@1#52800,SET-einq@1#79120,SET] size:410370 blobrefs:[(000010: 1684544); depth:1]
+  000011:[einr@1#79121,SET-fuau@1#105488,SET] seqnums:[79121-105488] points:[einr@1#79121,SET-fuau@1#105488,SET] size:407440 blobrefs:[(000012: 1687552); depth:1]
+  000013:[fuav@1#105489,SET-hfno@1#131846,SET] seqnums:[105489-131846] points:[fuav@1#105489,SET-hfno@1#131846,SET] size:408549 blobrefs:[(000014: 1686912); depth:1]
+  000015:[hfnp@1#131847,SET-iqzq@1#158184,SET] seqnums:[131847-158184] points:[hfnp@1#131847,SET-iqzq@1#158184,SET] size:408995 blobrefs:[(000016: 1685632); depth:1]
+  000017:[iqzr@1#158185,SET-kchm@1#184410,SET] seqnums:[158185-184410] points:[iqzr@1#158185,SET-kchm@1#184410,SET] size:414699 blobrefs:[(000018: 1678464); depth:1]
+  000019:[kchn@1#184411,SET-lnt@1#210733,SET] seqnums:[184411-210733] points:[kchn@1#184411,SET-lnt@1#210733,SET] size:409800 blobrefs:[(000020: 1684672); depth:1]
+  000021:[lnta@1#210734,SET-mzgo@1#237112,SET] seqnums:[210734-237112] points:[lnta@1#210734,SET-mzgo@1#237112,SET] size:407724 blobrefs:[(000022: 1688256); depth:1]
+  000023:[mzgp@1#237113,SET-okst@1#263454,SET] seqnums:[237113-263454] points:[mzgp@1#237113,SET-okst@1#263454,SET] size:408979 blobrefs:[(000024: 1685888); depth:1]
+  000025:[oksu@1#263455,SET-pwgo@1#289840,SET] seqnums:[263455-289840] points:[oksu@1#263455,SET-pwgo@1#289840,SET] size:403438 blobrefs:[(000026: 1688704); depth:1]
+  000027:[pwgp@1#289841,SET-rhth@1#316197,SET] seqnums:[289841-316197] points:[pwgp@1#289841,SET-rhth@1#316197,SET] size:408839 blobrefs:[(000028: 1686848); depth:1]
+  000029:[rhti@1#316198,SET-stec@1#342502,SET] seqnums:[316198-342502] points:[rhti@1#316198,SET-stec@1#342502,SET] size:411207 blobrefs:[(000030: 1683520); depth:1]
+  000031:[sted@1#342503,SET-uery@1#368888,SET] seqnums:[342503-368888] points:[sted@1#342503,SET-uery@1#368888,SET] size:406286 blobrefs:[(000032: 1688704); depth:1]
+  000033:[uerz@1#368889,SET-vqfq@1#395271,SET] seqnums:[368889-395271] points:[uerz@1#368889,SET-vqfq@1#395271,SET] size:406985 blobrefs:[(000034: 1688512); depth:1]
+  000035:[vqfr@1#395272,SET-xbqj@1#421574,SET] seqnums:[395272-421574] points:[vqfr@1#395272,SET-xbqj@1#421574,SET] size:410632 blobrefs:[(000036: 1683392); depth:1]
+  000037:[xbqk@1#421575,SET-ymzw@1#447842,SET] seqnums:[421575-447842] points:[xbqk@1#421575,SET-ymzw@1#447842,SET] size:412587 blobrefs:[(000038: 1681152); depth:1]
+  000039:[ymzx@1#447843,SET-zyni@1#474219,SET] seqnums:[447843-474219] points:[ymzx@1#447843,SET-zyni@1#474219,SET] size:407748 blobrefs:[(000040: 1688128); depth:1]
+  000041:[zynj@1#474220,SET-zzzz@1#475263,SET] seqnums:[474220-475263] points:[zynj@1#474220,SET-zzzz@1#475263,SET] size:16569 blobrefs:[(000042: 66816); depth:1]
+Blob files:
+  000006: 1694530 physical bytes, 1689536 value bytes
+  000008: 1694018 physical bytes, 1689024 value bytes
+  000010: 1689526 physical bytes, 1684544 value bytes
+  000012: 1692534 physical bytes, 1687552 value bytes
+  000014: 1691894 physical bytes, 1686912 value bytes
+  000016: 1690614 physical bytes, 1685632 value bytes
+  000018: 1683422 physical bytes, 1678464 value bytes
+  000020: 1689654 physical bytes, 1684672 value bytes
+  000022: 1693250 physical bytes, 1688256 value bytes
+  000024: 1690870 physical bytes, 1685888 value bytes
+  000026: 1693698 physical bytes, 1688704 value bytes
+  000028: 1691830 physical bytes, 1686848 value bytes
+  000030: 1688502 physical bytes, 1683520 value bytes
+  000032: 1693698 physical bytes, 1688704 value bytes
+  000034: 1693506 physical bytes, 1688512 value bytes
+  000036: 1688362 physical bytes, 1683392 value bytes
+  000038: 1686122 physical bytes, 1681152 value bytes
+  000040: 1693122 physical bytes, 1688128 value bytes
+  000042: 67041 physical bytes, 66816 value bytes
+
+# Schedule automatic compactions. These compactions should write data to L6. The
+# resulting sstables will reference multiple blob files but maintain a blob
+# reference depth of 1 because L6 has no referenced blob files and all the L0
+# input tables have a reference depth of 1.
+
+auto-compact
+----
+L6:
+  000044:[a@1#0,SET-czma@1#0,SET] seqnums:[0-0] points:[a@1#0,SET-czma@1#0,SET] size:713597 blobrefs:[(000006: 1689536), (000008: 1689024), (000010: 106944); depth:1]
+  000045:[czmb@1#0,SET-fzac@1#0,SET] seqnums:[0-0] points:[czmb@1#0,SET-fzac@1#0,SET] size:710727 blobrefs:[(000010: 1577600), (000012: 1687552), (000014: 223808); depth:1]
+  000046:[fzad@1#0,SET-iyoz@1#0,SET] seqnums:[0-0] points:[fzad@1#0,SET-iyoz@1#0,SET] size:710124 blobrefs:[(000014: 1463104), (000016: 1685632), (000018: 341504); depth:1]
+  000047:[iyp@1#0,SET-lxxp@1#0,SET] seqnums:[0-0] points:[iyp@1#0,SET-lxxp@1#0,SET] size:718439 blobrefs:[(000018: 1336960), (000020: 1684672), (000022: 457856); depth:1]
+  000048:[lxxq@1#0,SET-oxlo@1#0,SET] seqnums:[0-0] points:[lxxq@1#0,SET-oxlo@1#0,SET] size:711593 blobrefs:[(000022: 1230400), (000024: 1685888), (000026: 572480); depth:1]
+  000049:[oxlp@1#0,SET-rwyj@1#0,SET] seqnums:[0-0] points:[oxlp@1#0,SET-rwyj@1#0,SET] size:713709 blobrefs:[(000026: 1116224), (000028: 1686848), (000030: 683648); depth:1]
+  000050:[rwyk@1#0,SET-uwic@1#0,SET] seqnums:[0-0] points:[rwyk@1#0,SET-uwic@1#0,SET] size:718480 blobrefs:[(000030: 999872), (000032: 1688704), (000034: 792896); depth:1]
+  000051:[uwid@1#0,SET-xvqq@1#0,SET] seqnums:[0-0] points:[uwid@1#0,SET-xvqq@1#0,SET] size:720205 blobrefs:[(000034: 895616), (000036: 1683392), (000038: 900288); depth:1]
+  000052:[xvqr@1#0,SET-zzzz@1#0,SET] seqnums:[0-0] points:[xvqr@1#0,SET-zzzz@1#0,SET] size:521418 blobrefs:[(000038: 780864), (000040: 1688128), (000042: 66816); depth:1]
+Blob files:
+  000006: 1694530 physical bytes, 1689536 value bytes
+  000008: 1694018 physical bytes, 1689024 value bytes
+  000010: 1689526 physical bytes, 1684544 value bytes
+  000012: 1692534 physical bytes, 1687552 value bytes
+  000014: 1691894 physical bytes, 1686912 value bytes
+  000016: 1690614 physical bytes, 1685632 value bytes
+  000018: 1683422 physical bytes, 1678464 value bytes
+  000020: 1689654 physical bytes, 1684672 value bytes
+  000022: 1693250 physical bytes, 1688256 value bytes
+  000024: 1690870 physical bytes, 1685888 value bytes
+  000026: 1693698 physical bytes, 1688704 value bytes
+  000028: 1691830 physical bytes, 1686848 value bytes
+  000030: 1688502 physical bytes, 1683520 value bytes
+  000032: 1693698 physical bytes, 1688704 value bytes
+  000034: 1693506 physical bytes, 1688512 value bytes
+  000036: 1688362 physical bytes, 1683392 value bytes
+  000038: 1686122 physical bytes, 1681152 value bytes
+  000040: 1693122 physical bytes, 1688128 value bytes
+  000042: 67041 physical bytes, 66816 value bytes
diff --git a/testdata/event_listener b/testdata/event_listener
index 4b09d8ba..bcffd72d 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -66,6 +66,11 @@ close: db/marker.format-version.000008.021
 remove: db/marker.format-version.000007.020
 sync: db
 upgraded to format version: 021
+create: db/marker.format-version.000009.022
+close: db/marker.format-version.000009.022
+remove: db/marker.format-version.000008.021
+sync: db
+upgraded to format version: 022
 create: db/temporary.000003.dbtmp
 sync: db/temporary.000003.dbtmp
 close: db/temporary.000003.dbtmp
@@ -375,9 +380,9 @@ sync-data: checkpoint/OPTIONS-000003
 close: checkpoint/OPTIONS-000003
 close: db/OPTIONS-000003
 open-dir: checkpoint
-create: checkpoint/marker.format-version.000001.021
-sync-data: checkpoint/marker.format-version.000001.021
-close: checkpoint/marker.format-version.000001.021
+create: checkpoint/marker.format-version.000001.022
+sync-data: checkpoint/marker.format-version.000001.022
+close: checkpoint/marker.format-version.000001.022
 sync: checkpoint
 close: checkpoint
 link: db/000013.sst -> checkpoint/000013.sst
diff --git a/testdata/flushable_ingest b/testdata/flushable_ingest
index 8c9c7437..386df7ca 100644
--- a/testdata/flushable_ingest
+++ b/testdata/flushable_ingest
@@ -60,7 +60,7 @@ LOCK
 MANIFEST-000001
 OPTIONS-000003
 ext
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 # Test basic WAL replay
@@ -81,7 +81,7 @@ LOCK
 MANIFEST-000001
 OPTIONS-000003
 ext
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 open
@@ -390,7 +390,7 @@ LOCK
 MANIFEST-000001
 OPTIONS-000003
 ext
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 close
@@ -410,7 +410,7 @@ LOCK
 MANIFEST-000001
 OPTIONS-000003
 ext
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 open
@@ -443,7 +443,7 @@ MANIFEST-000001
 MANIFEST-000011
 OPTIONS-000014
 ext
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000002.MANIFEST-000011
 
 # Make sure that the new mutable memtable can accept writes.
@@ -586,7 +586,7 @@ LOCK
 MANIFEST-000001
 OPTIONS-000003
 ext
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 close
@@ -605,7 +605,7 @@ MANIFEST-000001
 OPTIONS-000003
 ext
 ext1
-marker.format-version.000008.021
+marker.format-version.000009.022
 marker.manifest.000001.MANIFEST-000001
 
 open
diff --git a/testdata/iter_histories/blob_references b/testdata/iter_histories/blob_references
index f643383b..a62915ff 100644
--- a/testdata/iter_histories/blob_references
+++ b/testdata/iter_histories/blob_references
@@ -15,6 +15,8 @@ L5:
   000004:[b@9#9,SET-d@9#9,SET] seqnums:[9-9] points:[b@9#9,SET-d@9#9,SET] size:893 blobrefs:[(000921: 10); depth:1]
 L6:
   000005:[b@2#2,SET-d@2#2,SET] seqnums:[2-2] points:[b@2#2,SET-d@2#2,SET] size:849 blobrefs:[(000921: 6); depth:1]
+Blob files:
+  000921: 62 physical bytes, 16 value bytes
 
 combined-iter
 first
@@ -73,6 +75,9 @@ L5:
   000004:[b@9#9,SETWITHDEL-e@1#9,SETWITHDEL] seqnums:[9-9] points:[b@9#9,SETWITHDEL-e@1#9,SETWITHDEL] size:852 blobrefs:[(000039: 14), (000921: 20); depth:2]
 L6:
   000005:[b@2#9,SETWITHDEL-f@2#3,SETWITHDEL] seqnums:[2-9] points:[b@2#9,SETWITHDEL-f@2#3,SETWITHDEL] size:861 blobrefs:[(000039: 11), (000921: 13); depth:2]
+Blob files:
+  000039: 71 physical bytes, 25 value bytes
+  000921: 79 physical bytes, 33 value bytes
 
 # The iterator stats should indicate that only the first value from each file
 # should trigger a read of the blob file. Once loaded, subsequent reads of the
@@ -133,6 +138,8 @@ L6
 ----
 L6:
   000004:[a#2,SETWITHDEL-l#2,SETWITHDEL] seqnums:[2-2] points:[a#2,SETWITHDEL-l#2,SETWITHDEL] size:956 blobrefs:[(000009: 96); depth:1]
+Blob files:
+  000009: 190 physical bytes, 96 value bytes
 
 combined-iter
 first
diff --git a/testdata/value_separation_policy b/testdata/value_separation_policy
index 310ff2bd..539bf3b8 100644
--- a/testdata/value_separation_policy
+++ b/testdata/value_separation_policy
@@ -134,7 +134,7 @@ RawWriter.AddWithBlobHandle("e#9,SET", "(f0,blk0[4:9])", 5, false)
 
 estimated-sizes
 ----
-file: 0, references: 14
+file: 0, references: 5
 
 close-output
 ----