manifest: rework blob file reference counting

Rework the reference counting of in-use blob files to decouple the reference
counts from the TableMetadatas' blob references. With the introduction of blob
file rewrites and replacement, the physical blob file and the logical blob
file's references have distinct lifetimes. A physical blob file that has been
replaced will need to be removed once all Versions that predate the replacement
have been unreferenced.

This commit renames the struct previously known as BlobFileMetadata to
PhysicalBlobFile, describing metadata particular to a physical file backing a
logical blob file. A new BlobFileMetadata struct is introduced that holds a
BlobFileID and a pointer to the PhysicalBlobFile.

This commit additionally adapts the physical blob file referencing to occur
through a separate B-Tree of BlobFileMetadata structs. Similar to
TableMetadata, references to blob files are maintained by copy-on-write B-Tree
nodes which themselves are reference counted. A PhysicalBlobFile's reference
count is incremented when a new B-Tree node references a containing
BlobFileMetadata, and it's decremented when the B-Tree node's reference count
falls to zero. This indirection ensures that a mutation to the set of blob
files only performs log(n) work. Since addition and removal of blob files
within a version is now modeled directly (as opposed to indirectly via
TableMetadata BlobReferences), a blob file may be removed or replaced within
the set before all referencing TableMetadata are removed.

Informs #4802.

Author: jbowens

checkpoint.go

@@ -354,13 +354,12 @@ func (d *DB) Checkpoint(
 	// When we write the MANIFEST of the checkpoint, we'll include a final
 	// VersionEdit that removes these blob files so that the checkpointed
 	// manifest is consistent.
-	var excludedBlobFiles map[base.DiskFileNum]*manifest.BlobFileMetadata
+	var excludedBlobFiles map[base.BlobFileID]*manifest.PhysicalBlobFile
 	if len(includedBlobFiles) < len(versionBlobFiles) {
-		excludedBlobFiles = make(map[base.DiskFileNum]*manifest.BlobFileMetadata, len(versionBlobFiles)-len(includedBlobFiles))
-		for _, blobFile := range versionBlobFiles {
-			if _, ok := includedBlobFiles[blobFile.FileID]; !ok {
-				diskFileNum := blob.DiskFileNumTODO(blobFile.FileID)
-				excludedBlobFiles[diskFileNum] = blobFile
+		excludedBlobFiles = make(map[base.BlobFileID]*manifest.PhysicalBlobFile, len(versionBlobFiles)-len(includedBlobFiles))
+		for _, meta := range versionBlobFiles {
+			if _, ok := includedBlobFiles[meta.FileID]; !ok {
+				excludedBlobFiles[meta.FileID] = meta.Physical
 			}
 		}
 	}
@@ -471,7 +470,7 @@ func (d *DB) writeCheckpointManifest(
 	manifestSize int64,
 	excludedTables map[manifest.DeletedTableEntry]*manifest.TableMetadata,
 	removeBackingTables []base.DiskFileNum,
-	excludedBlobFiles map[base.DiskFileNum]*manifest.BlobFileMetadata,
+	excludedBlobFiles map[base.BlobFileID]*manifest.PhysicalBlobFile,
 ) error {
 	// Copy the MANIFEST, and create a pointer to it. We copy rather
 	// than link because additional version edits added to the

compaction.go

@@ -2478,7 +2478,7 @@ func (d *DB) handleCompactFailure(c *compaction, err error) {
 func (d *DB) cleanupVersionEdit(ve *manifest.VersionEdit) {
 	obsoleteFiles := manifest.ObsoleteFiles{
 		TableBackings: make([]*manifest.TableBacking, 0, len(ve.NewTables)),
-		BlobFiles:     make([]*manifest.BlobFileMetadata, 0, len(ve.NewBlobFiles)),
+		BlobFiles:     make([]*manifest.PhysicalBlobFile, 0, len(ve.NewBlobFiles)),
 	}
 	deletedTables := make(map[base.TableNum]struct{})
 	for key := range ve.DeletedTables {
@@ -2488,10 +2488,10 @@ func (d *DB) cleanupVersionEdit(ve *manifest.VersionEdit) {
 		obsoleteFiles.AddBlob(ve.NewBlobFiles[i])
 		d.mu.versions.zombieBlobs.Add(objectInfo{
 			fileInfo: fileInfo{
-				FileNum:  base.DiskFileNum(ve.NewBlobFiles[i].FileID),
+				FileNum:  base.DiskFileNum(ve.NewBlobFiles[i].FileNum),
 				FileSize: ve.NewBlobFiles[i].Size,
 			},
-			isLocal: objstorage.IsLocalBlobFile(d.objProvider, base.DiskFileNum(ve.NewBlobFiles[i].FileID)),
+			isLocal: objstorage.IsLocalBlobFile(d.objProvider, base.DiskFileNum(ve.NewBlobFiles[i].FileNum)),
 		})
 	}
 	for i := range ve.NewTables {
@@ -3105,7 +3105,7 @@ func (d *DB) runCompaction(
 		// Delete any created tables or blob files.
 		obsoleteFiles := manifest.ObsoleteFiles{
 			TableBackings: make([]*manifest.TableBacking, 0, len(result.Tables)),
-			BlobFiles:     make([]*manifest.BlobFileMetadata, 0, len(result.Blobs)),
+			BlobFiles:     make([]*manifest.PhysicalBlobFile, 0, len(result.Blobs)),
 		}
 		d.mu.Lock()
 		for i := range result.Tables {
@@ -3291,7 +3291,7 @@ func (c *compaction) makeVersionEdit(result compact.Result) (*manifest.VersionEd
 		}
 	}
 	// Add any newly constructed blob files to the version edit.
-	ve.NewBlobFiles = make([]*manifest.BlobFileMetadata, len(result.Blobs))
+	ve.NewBlobFiles = make([]*manifest.PhysicalBlobFile, len(result.Blobs))
 	for i := range result.Blobs {
 		ve.NewBlobFiles[i] = result.Blobs[i].Metadata
 	}

data_test.go

@@ -927,7 +927,7 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 	// them to the final version edit.
 	valueSeparator := &defineDBValueSeparator{
 		pbr:   &preserveBlobReferences{},
-		metas: make(map[base.BlobFileID]*manifest.BlobFileMetadata),
+		metas: make(map[base.BlobFileID]*manifest.PhysicalBlobFile),
 	}
 
 	var mem *memTable
@@ -1640,7 +1640,8 @@ func describeLSM(d *DB, verbose bool) 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.FileID, meta.Size, meta.ValueSize)
+			fmt.Fprintf(&buf, "  %s: [%s] %d physical bytes, %d value bytes\n",
+				meta.FileID, meta.Physical.FileNum, meta.Physical.Size, meta.Physical.ValueSize)
 		}
 	}
 	return buf.String()

event.go

@@ -19,7 +19,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/remote"
-	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/redact"
 )
@@ -1234,10 +1233,9 @@ func (d *DB) reportCorruption(meta any, err error) error {
 	switch meta := meta.(type) {
 	case *manifest.TableMetadata:
 		return d.reportFileCorruption(base.FileTypeTable, meta.TableBacking.DiskFileNum, meta.UserKeyBounds(), err)
-	case *manifest.BlobFileMetadata:
-		diskFileNum := blob.DiskFileNumTODO(meta.FileID)
+	case *manifest.PhysicalBlobFile:
 		// TODO(jackson): Add bounds for blob files.
-		return d.reportFileCorruption(base.FileTypeBlob, diskFileNum, base.UserKeyBounds{}, err)
+		return d.reportFileCorruption(base.FileTypeBlob, meta.FileNum, base.UserKeyBounds{}, err)
 	default:
 		panic(fmt.Sprintf("unknown metadata type: %T", meta))
 	}

internal/compact/run.go

@@ -60,7 +60,7 @@ type OutputBlob struct {
 	// ObjMeta is metadata for the object backing the blob file.
 	ObjMeta objstorage.ObjectMetadata
 	// Metadata is metadata for the blob file.
-	Metadata *manifest.BlobFileMetadata
+	Metadata *manifest.PhysicalBlobFile
 }
 
 // Stats describes stats collected during the compaction.
@@ -140,7 +140,7 @@ type ValueSeparationMetadata struct {
 	// The below fields are only populated if a new blob file was created.
 	BlobFileStats    blob.FileWriterStats
 	BlobFileObject   objstorage.ObjectMetadata
-	BlobFileMetadata *manifest.BlobFileMetadata
+	BlobFileMetadata *manifest.PhysicalBlobFile
 }
 
 // Runner is a helper for running the "data" part of a compaction (where we use