base: add DiskFile interface

Add a DiskFile interface for types with a DiskFileNum() method.

Author: jbowens

blob_rewrite.go

@@ -181,11 +181,12 @@ func (c *blobFileRewriteCompaction) Execute(jobID JobID, d *DB) error {
 			// Now that we have the manifest lock, check if the blob file is
 			// still current. If not, we bubble up ErrCancelledCompaction.
 			v := d.mu.versions.currentVersion()
-			currentDiskFileNum, ok := v.BlobFiles.Lookup(c.input.FileID)
+			currentDiskFile, ok := v.BlobFiles.Lookup(c.input.FileID)
 			if !ok {
 				return versionUpdate{}, errors.Wrapf(ErrCancelledCompaction,
 					"blob file %s became unreferenced", c.input.FileID)
 			}
+			currentDiskFileNum := currentDiskFile.DiskFileNum()
 			// Assert that the current version's disk file number for the blob
 			// matches the one we rewrote. This compaction should be the only
 			// rewrite compaction running for this blob file.

blob_rewrite_test.go

@@ -421,6 +421,6 @@ type constantFileMapping base.DiskFileNum
 // Assert that (*inputFileMapping) implements base.BlobFileMapping.
 var _ base.BlobFileMapping = constantFileMapping(0)
 
-func (m constantFileMapping) Lookup(fileID base.BlobFileID) (base.DiskFileNum, bool) {
+func (m constantFileMapping) Lookup(fileID base.BlobFileID) (base.DiskFile, bool) {
 	return base.DiskFileNum(m), true
 }

checkpoint.go

@@ -326,11 +326,11 @@ func (d *DB) Checkpoint(
 						includedBlobFiles[ref.FileID] = struct{}{}
 
 						// Map the BlobFileID to a DiskFileNum in the current version.
-						diskFileNum, ok := current.BlobFiles.Lookup(ref.FileID)
+						diskFile, ok := current.BlobFiles.Lookup(ref.FileID)
 						if !ok {
 							return errors.Errorf("blob file %s not found", ref.FileID)
 						}
-						ckErr = copyFile(base.FileTypeBlob, diskFileNum)
+						ckErr = copyFile(base.FileTypeBlob, diskFile.DiskFileNum())
 						if ckErr != nil {
 							return ckErr
 						}

internal/base/filenames.go

@@ -64,13 +64,19 @@ func (id BlobFileID) SafeFormat(w redact.SafePrinter, _ rune) {
 // manifest.
 type BlobReferenceID uint32
 
+// DiskFile is an interface that provides access to a disk file number.
+type DiskFile interface {
+	// DiskFileNum returns the disk file number.
+	DiskFileNum() DiskFileNum
+}
+
 // BlobFileMapping defines the mapping between blob file IDs and disk file numbers.
 // It's implemented by *manifest.BlobFileSet.
 type BlobFileMapping interface {
-	// Lookup returns the disk file number for the given blob file ID. It
+	// Lookup returns a DiskFile for the given blob file ID. It
 	// returns false for the second return value if the blob file ID is not
 	// present in the mapping.
-	Lookup(BlobFileID) (DiskFileNum, bool)
+	Lookup(BlobFileID) (DiskFile, bool)
 }
 
 // A DiskFileNum identifies a file or object with exists on disk.
@@ -83,6 +89,11 @@ func (dfn DiskFileNum) SafeFormat(w redact.SafePrinter, verb rune) {
 	w.Printf("%06d", redact.SafeUint(dfn))
 }
 
+// DiskFileNum implements the DiskFile interface.
+func (fdn DiskFileNum) DiskFileNum() DiskFileNum {
+	return fdn
+}
+
 // FileType enumerates the types of files found in a DB.
 type FileType int
 

internal/manifest/blob_metadata.go

@@ -148,6 +148,11 @@ func (m *PhysicalBlobFile) String() string {
 	return redact.StringWithoutMarkers(m)
 }
 
+// DiskFileNum implements base.DiskFile.
+func (m *PhysicalBlobFile) DiskFileNum() base.DiskFileNum {
+	return m.FileNum
+}
+
 // ref increments the reference count for the blob file.
 func (m *PhysicalBlobFile) ref() {
 	m.refs.Add(+1)
@@ -338,15 +343,14 @@ func (s *BlobFileSet) Count() int {
 	return s.tree.Count()
 }
 
-// Lookup returns the file number of the physical blob file backing the given
-// file ID. It returns false for the second return value if the FileID is not
-// present in the set.
-func (s *BlobFileSet) Lookup(fileID base.BlobFileID) (base.DiskFileNum, bool) {
+// Lookup returns the physical blob file backing the given file ID. It returns
+// false for the second return value if the FileID is not present in the set.
+func (s *BlobFileSet) Lookup(fileID base.BlobFileID) (base.DiskFile, bool) {
 	phys, ok := s.LookupPhysical(fileID)
 	if !ok {
-		return 0, false
+		return nil, false
 	}
-	return phys.FileNum, true
+	return phys, true
 }
 
 // LookupPhysical returns the *PhysicalBlobFile backing the given file ID. It

internal/manifest/blob_metadata_test.go

@@ -182,7 +182,7 @@ func TestBlobFileSet_Lookup(t *testing.T) {
 	for i := 0; i < numBlobFiles; i++ {
 		fn, ok := set.Lookup(base.BlobFileID(i))
 		require.True(t, ok)
-		require.Equal(t, files[i].FileNum, fn)
+		require.Equal(t, files[i].FileNum, fn.DiskFileNum())
 	}
 }
 

replay/replay.go

@@ -1025,9 +1025,12 @@ func findManifestStart(
 
 type blobFileMap map[base.BlobFileID]base.DiskFileNum
 
-func (m blobFileMap) Lookup(fileID base.BlobFileID) (base.DiskFileNum, bool) {
+func (m blobFileMap) Lookup(fileID base.BlobFileID) (base.DiskFile, bool) {
 	diskFileNum, ok := m[fileID]
-	return diskFileNum, ok
+	if !ok {
+		return nil, false
+	}
+	return diskFileNum, true
 }
 
 // loadFlushedSSTableKeys copies keys from the sstables specified by `fileNums`

sstable/blob/fetcher.go

@@ -144,10 +144,11 @@ func (r *ValueFetcher) retrieve(ctx context.Context, vh Handle) (val []byte, err
 				return nil, err
 			}
 		}
-		diskFileNum, ok := r.fileMapping.Lookup(vh.BlobFileID)
+		diskFile, ok := r.fileMapping.Lookup(vh.BlobFileID)
 		if !ok {
 			return nil, errors.AssertionFailedf("blob file %s not found", vh.BlobFileID)
 		}
+		diskFileNum := diskFile.DiskFileNum()
 		if cr.r, cr.closeFunc, err = r.readerProvider.GetValueReader(ctx, diskFileNum); err != nil {
 			return nil, err
 		}

sstable/blob/fetcher_test.go

@@ -32,7 +32,7 @@ type identityFileMapping struct{}
 // Assert that (identityFileMapping) implements base.BlobFileMapping.
 var _ base.BlobFileMapping = identityFileMapping{}
 
-func (identityFileMapping) Lookup(blobFileID base.BlobFileID) (base.DiskFileNum, bool) {
+func (identityFileMapping) Lookup(blobFileID base.BlobFileID) (base.DiskFile, bool) {
 	return base.DiskFileNum(blobFileID), true
 }
 

sstable/blob/rewrite.go

@@ -116,6 +116,6 @@ type inputFileMapping base.DiskFileNum
 // Assert that (*inputFileMapping) implements base.BlobFileMapping.
 var _ base.BlobFileMapping = inputFileMapping(0)
 
-func (m inputFileMapping) Lookup(fileID base.BlobFileID) (base.DiskFileNum, bool) {
+func (m inputFileMapping) Lookup(fileID base.BlobFileID) (base.DiskFile, bool) {
 	return base.DiskFileNum(m), true
 }