Repository: cockroachdb/pebble

Issue:
db: copy blob files in checkpoints
Today taking a checkpoint with `DB.Checkpoint` does not copy blob files at all. We should adjust `Checkpoint` to copy all referenced blob files. When the caller specifies checkpoint spans, we exclude some tables from the checkpoint and amend the checkpointed manifest to remove the excluded tables. We should also exclude any blob files that have no references from _included_ tables, and include the removal of those excluded blob files in the version edit we append.

Jira issue: PEBBLE-400

Discussion:

Solution patch:
diff --git a/checkpoint.go b/checkpoint.go
index 5901800a..9c616f08 100644
--- a/checkpoint.go
+++ b/checkpoint.go
@@ -12,6 +12,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/errors/oserror"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/atomicfs"
@@ -196,6 +197,7 @@ func (d *DB) Checkpoint(
 	d.mu.versions.virtualBackings.ForEach(func(backing *fileBacking) {
 		virtualBackingFiles[backing.DiskFileNum] = struct{}{}
 	})
+	versionBlobFiles := d.mu.versions.blobFiles.Metadatas()
 
 	// Acquire the logs while holding mutexes to ensure we don't race with a
 	// flush that might mark a log that's relevant to `current` as obsolete
@@ -263,10 +265,32 @@ func (d *DB) Checkpoint(
 	}
 
 	var excludedTables map[deletedFileEntry]*tableMetadata
+	var includedBlobFiles map[base.DiskFileNum]struct{}
 	var remoteFiles []base.DiskFileNum
 	// Set of FileBacking.DiskFileNum which will be required by virtual sstables
 	// in the checkpoint.
 	requiredVirtualBackingFiles := make(map[base.DiskFileNum]struct{})
+
+	copyFile := func(typ base.FileType, fileNum base.DiskFileNum) error {
+		meta, err := d.objProvider.Lookup(typ, fileNum)
+		if err != nil {
+			return err
+		}
+		if meta.IsRemote() {
+			// We don't copy remote files. This is desirable as checkpointing is
+			// supposed to be a fast operation, and references to remote files can
+			// always be resolved by any checkpoint readers by reading the object
+			// catalog. We don't add this file to excludedFiles either, as that'd
+			// cause it to be deleted in the second manifest entry which is also
+			// inaccurate.
+			remoteFiles = append(remoteFiles, meta.DiskFileNum)
+			return nil
+		}
+		srcPath := base.MakeFilepath(fs, d.dirname, typ, fileNum)
+		destPath := fs.PathJoin(destDir, fs.PathBase(srcPath))
+		return vfs.LinkOrCopy(fs, srcPath, destPath)
+	}
+
 	// Link or copy the sstables.
 	for l := range current.Levels {
 		iter := current.Levels[l].Iter()
@@ -282,6 +306,22 @@ func (d *DB) Checkpoint(
 				continue
 			}
 
+			// Copy any referenced blob files that have not already been copied.
+			if len(f.BlobReferences) > 0 {
+				if includedBlobFiles == nil {
+					includedBlobFiles = make(map[base.DiskFileNum]struct{})
+				}
+				for _, ref := range f.BlobReferences {
+					if _, ok := includedBlobFiles[ref.FileNum]; !ok {
+						includedBlobFiles[ref.FileNum] = struct{}{}
+						ckErr = copyFile(base.FileTypeBlob, ref.FileNum)
+						if ckErr != nil {
+							return ckErr
+						}
+					}
+				}
+			}
+
 			fileBacking := f.FileBacking
 			if f.Virtual {
 				if _, ok := requiredVirtualBackingFiles[fileBacking.DiskFileNum]; ok {
@@ -289,25 +329,7 @@ func (d *DB) Checkpoint(
 				}
 				requiredVirtualBackingFiles[fileBacking.DiskFileNum] = struct{}{}
 			}
-			meta, err := d.objProvider.Lookup(base.FileTypeTable, fileBacking.DiskFileNum)
-			if err != nil {
-				ckErr = err
-				return ckErr
-			}
-			if meta.IsRemote() {
-				// We don't copy remote files. This is desirable as checkpointing is
-				// supposed to be a fast operation, and references to remote files can
-				// always be resolved by any checkpoint readers by reading the object
-				// catalog. We don't add this file to excludedFiles either, as that'd
-				// cause it to be deleted in the second manifest entry which is also
-				// inaccurate.
-				remoteFiles = append(remoteFiles, meta.DiskFileNum)
-				continue
-			}
-
-			srcPath := base.MakeFilepath(fs, d.dirname, base.FileTypeTable, fileBacking.DiskFileNum)
-			destPath := fs.PathJoin(destDir, fs.PathBase(srcPath))
-			ckErr = vfs.LinkOrCopy(fs, srcPath, destPath)
+			ckErr = copyFile(base.FileTypeTable, fileBacking.DiskFileNum)
 			if ckErr != nil {
 				return ckErr
 			}
@@ -322,16 +344,29 @@ func (d *DB) Checkpoint(
 			removeBackingTables = append(removeBackingTables, diskFileNum)
 		}
 	}
+	// Record the blob files that are not referenced by any included sstables.
+	// 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
+	if len(includedBlobFiles) < len(versionBlobFiles) {
+		excludedBlobFiles = make(map[base.DiskFileNum]*manifest.BlobFileMetadata, len(versionBlobFiles)-len(includedBlobFiles))
+		for _, blobFile := range versionBlobFiles {
+			if _, ok := includedBlobFiles[blobFile.FileNum]; !ok {
+				excludedBlobFiles[blobFile.FileNum] = blobFile
+			}
+		}
+	}
 
 	ckErr = d.writeCheckpointManifest(
 		fs, formatVers, destDir, dir, manifestFileNum, manifestSize,
-		excludedTables, removeBackingTables,
+		excludedTables, removeBackingTables, excludedBlobFiles,
 	)
 	if ckErr != nil {
 		return ckErr
 	}
 	if len(remoteFiles) > 0 {
-		ckErr = d.objProvider.CheckpointState(fs, destDir, base.FileTypeTable, remoteFiles)
+		ckErr = d.objProvider.CheckpointState(fs, destDir, remoteFiles)
 		if ckErr != nil {
 			return ckErr
 		}
@@ -429,6 +464,7 @@ func (d *DB) writeCheckpointManifest(
 	manifestSize int64,
 	excludedTables map[deletedFileEntry]*tableMetadata,
 	removeBackingTables []base.DiskFileNum,
+	excludedBlobFiles map[base.DiskFileNum]*manifest.BlobFileMetadata,
 ) error {
 	// Copy the MANIFEST, and create a pointer to it. We copy rather
 	// than link because additional version edits added to the
@@ -477,11 +513,12 @@ func (d *DB) writeCheckpointManifest(
 			}
 		}
 
-		if len(excludedTables) > 0 {
+		if len(excludedTables) > 0 || len(excludedBlobFiles) > 0 {
 			// Write out an additional VersionEdit that deletes the excluded SST files.
 			ve := versionEdit{
 				DeletedTables:        excludedTables,
 				RemovedBackingTables: removeBackingTables,
+				DeletedBlobFiles:     excludedBlobFiles,
 			}
 
 			rw, err := w.Next()
diff --git a/objstorage/objstorage.go b/objstorage/objstorage.go
index 1884c16a..c26095dc 100644
--- a/objstorage/objstorage.go
+++ b/objstorage/objstorage.go
@@ -338,7 +338,7 @@ type Provider interface {
 	// CheckpointState saves any saved state on local disk to the specified
 	// directory on the specified VFS. A new Pebble instance instantiated at that
 	// path should be able to resolve references to the specified files.
-	CheckpointState(fs vfs.FS, dir string, fileType base.FileType, fileNums []base.DiskFileNum) error
+	CheckpointState(fs vfs.FS, dir string, fileNums []base.DiskFileNum) error
 
 	// Metrics returns metrics about objstorage. Currently, it only returns metrics
 	// about the shared cache.
diff --git a/objstorage/objstorageprovider/provider.go b/objstorage/objstorageprovider/provider.go
index 952dda2b..92f44ce9 100644
--- a/objstorage/objstorageprovider/provider.go
+++ b/objstorage/objstorageprovider/provider.go
@@ -541,17 +541,15 @@ func (p *provider) Metrics() sharedcache.Metrics {
 }
 
 // CheckpointState is part of the objstorage.Provider interface.
-func (p *provider) CheckpointState(
-	fs vfs.FS, dir string, fileType base.FileType, fileNums []base.DiskFileNum,
-) error {
+func (p *provider) CheckpointState(fs vfs.FS, dir string, fileNums []base.DiskFileNum) error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	for i := range fileNums {
 		if _, ok := p.mu.knownObjects[fileNums[i]]; !ok {
 			return errors.Wrapf(
 				os.ErrNotExist,
-				"file %s (type %s) unknown to the objstorage provider",
-				fileNums[i], fileType,
+				"file %s unknown to the objstorage provider",
+				fileNums[i],
 			)
 		}
 		// Prevent this object from deletion, at least for the life of this instance.



Test patch:
diff --git a/testdata/checkpoint b/testdata/checkpoint
index cd50b9a6..b952e0ca 100644
--- a/testdata/checkpoint
+++ b/testdata/checkpoint
@@ -844,3 +844,290 @@ L6:
   000015(000011):[i#20,SET-i#20,SET]
   000016(000011):[k#20,SET-k#20,SET]
   000014:[z#22,SET-z#22,SET]
+
+# Create a new database with value separation enabled to test that blob files
+# are included in checkpoints.
+
+open valsepdb value-separation=(true,1,5)
+----
+mkdir-all: valsepdb 0755
+open-dir: .
+sync: .
+close: .
+open-dir: valsepdb
+close: valsepdb
+open-dir: valsepdb
+lock: valsepdb/LOCK
+open-dir: valsepdb
+open-dir: valsepdb
+open-dir: valsepdb
+open-dir: valsepdb
+create: valsepdb/MANIFEST-000001
+sync: valsepdb/MANIFEST-000001
+create: valsepdb/marker.manifest.000001.MANIFEST-000001
+close: valsepdb/marker.manifest.000001.MANIFEST-000001
+sync: valsepdb
+open-dir: valsepdb
+create: valsepdb/000002.log
+sync: valsepdb
+create: valsepdb/marker.format-version.000001.014
+close: valsepdb/marker.format-version.000001.014
+sync: valsepdb
+create: valsepdb/marker.format-version.000002.015
+close: valsepdb/marker.format-version.000002.015
+remove: valsepdb/marker.format-version.000001.014
+sync: valsepdb
+create: valsepdb/marker.format-version.000003.016
+close: valsepdb/marker.format-version.000003.016
+remove: valsepdb/marker.format-version.000002.015
+sync: valsepdb
+create: valsepdb/marker.format-version.000004.017
+close: valsepdb/marker.format-version.000004.017
+remove: valsepdb/marker.format-version.000003.016
+sync: valsepdb
+create: valsepdb/marker.format-version.000005.018
+close: valsepdb/marker.format-version.000005.018
+remove: valsepdb/marker.format-version.000004.017
+sync: valsepdb
+create: valsepdb/marker.format-version.000006.019
+close: valsepdb/marker.format-version.000006.019
+remove: valsepdb/marker.format-version.000005.018
+sync: valsepdb
+create: valsepdb/marker.format-version.000007.020
+close: valsepdb/marker.format-version.000007.020
+remove: valsepdb/marker.format-version.000006.019
+sync: valsepdb
+create: valsepdb/marker.format-version.000008.021
+close: valsepdb/marker.format-version.000008.021
+remove: valsepdb/marker.format-version.000007.020
+sync: valsepdb
+create: valsepdb/marker.format-version.000009.022
+close: valsepdb/marker.format-version.000009.022
+remove: valsepdb/marker.format-version.000008.021
+sync: valsepdb
+create: valsepdb/temporary.000003.dbtmp
+sync: valsepdb/temporary.000003.dbtmp
+close: valsepdb/temporary.000003.dbtmp
+rename: valsepdb/temporary.000003.dbtmp -> valsepdb/OPTIONS-000003
+sync: valsepdb
+
+batch valsepdb
+set a a
+set b b
+set c c
+set d d
+set e e
+set f f
+----
+sync-data: valsepdb/000002.log
+
+# The flush creates a blob file.
+
+flush valsepdb
+----
+sync-data: valsepdb/000002.log
+close: valsepdb/000002.log
+create: valsepdb/000004.log
+sync: valsepdb
+create: valsepdb/000005.sst
+create: valsepdb/000006.blob
+sync-data: valsepdb/000006.blob
+close: valsepdb/000006.blob
+sync-data: valsepdb/000005.sst
+close: valsepdb/000005.sst
+sync: valsepdb
+sync: valsepdb/MANIFEST-000001
+
+# Perform a checkpoint. It should include the blob file.
+
+checkpoint valsepdb checkpoints/checkpoint8
+----
+mkdir-all: checkpoints/checkpoint8 0755
+open-dir: checkpoints
+sync: checkpoints
+close: checkpoints
+open-dir: checkpoints/checkpoint8
+open: valsepdb/OPTIONS-000003
+create: checkpoints/checkpoint8/OPTIONS-000003
+sync-data: checkpoints/checkpoint8/OPTIONS-000003
+close: checkpoints/checkpoint8/OPTIONS-000003
+close: valsepdb/OPTIONS-000003
+open-dir: checkpoints/checkpoint8
+create: checkpoints/checkpoint8/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint8/marker.format-version.000001.022
+close: checkpoints/checkpoint8/marker.format-version.000001.022
+sync: checkpoints/checkpoint8
+close: checkpoints/checkpoint8
+link: valsepdb/000006.blob -> checkpoints/checkpoint8/000006.blob
+link: valsepdb/000005.sst -> checkpoints/checkpoint8/000005.sst
+open: valsepdb/MANIFEST-000001 (options: *vfs.sequentialReadsOption)
+create: checkpoints/checkpoint8/MANIFEST-000001
+sync-data: checkpoints/checkpoint8/MANIFEST-000001
+close: checkpoints/checkpoint8/MANIFEST-000001
+close: valsepdb/MANIFEST-000001
+open-dir: checkpoints/checkpoint8
+create: checkpoints/checkpoint8/marker.manifest.000001.MANIFEST-000001
+sync-data: checkpoints/checkpoint8/marker.manifest.000001.MANIFEST-000001
+close: checkpoints/checkpoint8/marker.manifest.000001.MANIFEST-000001
+sync: checkpoints/checkpoint8
+close: checkpoints/checkpoint8
+open: valsepdb/000004.log (options: *vfs.sequentialReadsOption)
+create: checkpoints/checkpoint8/000004.log
+sync-data: checkpoints/checkpoint8/000004.log
+close: checkpoints/checkpoint8/000004.log
+close: valsepdb/000004.log
+sync: checkpoints/checkpoint8
+close: checkpoints/checkpoint8
+
+# Validate that we can open the checkpoint.
+
+open checkpoints/checkpoint8 readonly
+----
+open-dir: checkpoints/checkpoint8
+lock: checkpoints/checkpoint8/LOCK
+open-dir: checkpoints/checkpoint8
+open-dir: checkpoints/checkpoint8
+open-dir: checkpoints/checkpoint8
+open-dir: checkpoints/checkpoint8
+open: checkpoints/checkpoint8/MANIFEST-000001
+close: checkpoints/checkpoint8/MANIFEST-000001
+open-dir: checkpoints/checkpoint8
+open: checkpoints/checkpoint8/OPTIONS-000003
+close: checkpoints/checkpoint8/OPTIONS-000003
+open: checkpoints/checkpoint8/000004.log
+close: checkpoints/checkpoint8/000004.log
+
+# Ensure that we can read values contained in the blob file.
+
+scan checkpoints/checkpoint8
+----
+open: checkpoints/checkpoint8/000005.sst (options: *vfs.randomReadsOption)
+read-at(750, 57): checkpoints/checkpoint8/000005.sst
+read-at(699, 51): checkpoints/checkpoint8/000005.sst
+read-at(172, 527): checkpoints/checkpoint8/000005.sst
+read-at(131, 41): checkpoints/checkpoint8/000005.sst
+read-at(0, 131): checkpoints/checkpoint8/000005.sst
+open: checkpoints/checkpoint8/000006.blob (options: *vfs.randomReadsOption)
+read-at(19, 33): checkpoints/checkpoint8/000006.blob
+read-at(11, 8): checkpoints/checkpoint8/000006.blob
+read-at(0, 11): checkpoints/checkpoint8/000006.blob
+a a
+b b
+c c
+d d
+e e
+f f
+.
+
+close checkpoints/checkpoint8
+----
+
+# Write and flush new keys.
+
+batch valsepdb
+set m m
+set n n
+set o o
+set p p
+set q q
+set r r
+----
+sync-data: valsepdb/000004.log
+
+flush valsepdb
+----
+sync-data: valsepdb/000004.log
+close: valsepdb/000004.log
+reuseForWrite: valsepdb/000002.log -> valsepdb/000007.log
+sync: valsepdb
+create: valsepdb/000008.sst
+create: valsepdb/000009.blob
+sync-data: valsepdb/000009.blob
+close: valsepdb/000009.blob
+sync-data: valsepdb/000008.sst
+close: valsepdb/000008.sst
+sync: valsepdb
+sync: valsepdb/MANIFEST-000001
+
+
+# Create a checkpoint that omits SSTs that don't overlap with the [a - c) range.
+# Blob file 000009.blob is not included because all the referencing tables are
+# exlcuded.
+
+checkpoint valsepdb checkpoints/checkpoint9 restrict=(a-c)
+----
+mkdir-all: checkpoints/checkpoint9 0755
+open-dir: checkpoints
+sync: checkpoints
+close: checkpoints
+open-dir: checkpoints/checkpoint9
+open: valsepdb/OPTIONS-000003
+create: checkpoints/checkpoint9/OPTIONS-000003
+sync-data: checkpoints/checkpoint9/OPTIONS-000003
+close: checkpoints/checkpoint9/OPTIONS-000003
+close: valsepdb/OPTIONS-000003
+open-dir: checkpoints/checkpoint9
+create: checkpoints/checkpoint9/marker.format-version.000001.022
+sync-data: checkpoints/checkpoint9/marker.format-version.000001.022
+close: checkpoints/checkpoint9/marker.format-version.000001.022
+sync: checkpoints/checkpoint9
+close: checkpoints/checkpoint9
+link: valsepdb/000006.blob -> checkpoints/checkpoint9/000006.blob
+link: valsepdb/000005.sst -> checkpoints/checkpoint9/000005.sst
+open: valsepdb/MANIFEST-000001 (options: *vfs.sequentialReadsOption)
+create: checkpoints/checkpoint9/MANIFEST-000001
+sync-data: checkpoints/checkpoint9/MANIFEST-000001
+close: checkpoints/checkpoint9/MANIFEST-000001
+close: valsepdb/MANIFEST-000001
+open-dir: checkpoints/checkpoint9
+create: checkpoints/checkpoint9/marker.manifest.000001.MANIFEST-000001
+sync-data: checkpoints/checkpoint9/marker.manifest.000001.MANIFEST-000001
+close: checkpoints/checkpoint9/marker.manifest.000001.MANIFEST-000001
+sync: checkpoints/checkpoint9
+close: checkpoints/checkpoint9
+open: valsepdb/000007.log (options: *vfs.sequentialReadsOption)
+create: checkpoints/checkpoint9/000007.log
+sync-data: checkpoints/checkpoint9/000007.log
+close: checkpoints/checkpoint9/000007.log
+close: valsepdb/000007.log
+sync: checkpoints/checkpoint9
+close: checkpoints/checkpoint9
+
+open checkpoints/checkpoint9 readonly
+----
+open-dir: checkpoints/checkpoint9
+lock: checkpoints/checkpoint9/LOCK
+open-dir: checkpoints/checkpoint9
+open-dir: checkpoints/checkpoint9
+open-dir: checkpoints/checkpoint9
+open-dir: checkpoints/checkpoint9
+open: checkpoints/checkpoint9/MANIFEST-000001
+close: checkpoints/checkpoint9/MANIFEST-000001
+open-dir: checkpoints/checkpoint9
+open: checkpoints/checkpoint9/OPTIONS-000003
+close: checkpoints/checkpoint9/OPTIONS-000003
+open: checkpoints/checkpoint9/000007.log
+close: checkpoints/checkpoint9/000007.log
+
+scan checkpoints/checkpoint9
+----
+open: checkpoints/checkpoint9/000005.sst (options: *vfs.randomReadsOption)
+read-at(750, 57): checkpoints/checkpoint9/000005.sst
+read-at(699, 51): checkpoints/checkpoint9/000005.sst
+read-at(172, 527): checkpoints/checkpoint9/000005.sst
+read-at(131, 41): checkpoints/checkpoint9/000005.sst
+read-at(0, 131): checkpoints/checkpoint9/000005.sst
+open: checkpoints/checkpoint9/000006.blob (options: *vfs.randomReadsOption)
+read-at(19, 33): checkpoints/checkpoint9/000006.blob
+read-at(11, 8): checkpoints/checkpoint9/000006.blob
+read-at(0, 11): checkpoints/checkpoint9/000006.blob
+a a
+b b
+c c
+d d
+e e
+f f
+.
+
+close checkpoints/checkpoint9
+----