valsep: wrap new blob files in valsep output metadata in NewBlobFileInfo slice

Author: xinhaoz

blob_rewrite_test.go

@@ -131,11 +131,13 @@ func TestBlobRewrite(t *testing.T) {
 
 				meta, err := vs.FinishOutput()
 				require.NoError(t, err)
-				if meta.BlobFileObject.DiskFileNum == 0 {
+				if len(meta.NewBlobFiles) == 0 {
 					fmt.Fprintln(&buf, "no blob file created")
 				} else {
-					fmt.Fprintf(&buf, "Blob file created: %s\n", meta.BlobFileMetadata)
-					fmt.Fprintln(&buf, meta.BlobFileStats)
+					for _, bf := range meta.NewBlobFiles {
+						fmt.Fprintf(&buf, "Blob file created: %s\n", bf.FileMetadata)
+						fmt.Fprintln(&buf, bf.FileStats)
+					}
 				}
 				if len(meta.BlobReferences) == 0 {
 					fmt.Fprintln(&buf, "blobrefs:[]")

compaction_value_separation.go

@@ -398,9 +398,11 @@ func (vs *writeNewBlobFiles) FinishOutput() (valsep.ValueSeparationMetadata, err
 		},
 		BlobReferenceSize:  stats.UncompressedValueBytes,
 		BlobReferenceDepth: 1,
-		BlobFileStats:      stats,
-		BlobFileObject:     vs.objMeta,
-		BlobFileMetadata:   meta,
+		NewBlobFiles: []valsep.NewBlobFileInfo{{
+			FileStats:    stats,
+			FileObject:   vs.objMeta,
+			FileMetadata: meta,
+		}},
 	}, nil
 }
 

compaction_value_separation_test.go

@@ -143,11 +143,13 @@ func TestValueSeparationPolicy(t *testing.T) {
 
 				meta, err := vs.FinishOutput()
 				require.NoError(t, err)
-				if meta.BlobFileObject.DiskFileNum == 0 {
+				if len(meta.NewBlobFiles) == 0 {
 					fmt.Fprintln(&buf, "no blob file created")
 				} else {
-					fmt.Fprintf(&buf, "Blob file created: %s\n", meta.BlobFileMetadata)
-					fmt.Fprintln(&buf, meta.BlobFileStats)
+					for _, bf := range meta.NewBlobFiles {
+						fmt.Fprintf(&buf, "Blob file created: %s\n", bf.FileMetadata)
+						fmt.Fprintln(&buf, bf.FileStats)
+					}
 				}
 				if len(meta.BlobReferences) == 0 {
 					fmt.Fprintln(&buf, "blobrefs:[]")

internal/compact/run.go

@@ -202,16 +202,18 @@ func (r *Runner) WriteTable(
 	// Inform the value separation policy that the table is finished.
 	valSepMeta, valSepErr := valueSeparation.FinishOutput()
 	if valSepErr != nil {
-		r.err = errors.CombineErrors(r.err, valSepErr)
+		err = errors.CombineErrors(err, valSepErr)
 	} else {
 		r.tables[len(r.tables)-1].BlobReferences = valSepMeta.BlobReferences
 		r.tables[len(r.tables)-1].BlobReferenceDepth = valSepMeta.BlobReferenceDepth
-		if valSepMeta.BlobFileObject.DiskFileNum != 0 {
-			r.blobs = append(r.blobs, OutputBlob{
-				Stats:    valSepMeta.BlobFileStats,
-				ObjMeta:  valSepMeta.BlobFileObject,
-				Metadata: valSepMeta.BlobFileMetadata,
-			})
+		if len(valSepMeta.NewBlobFiles) != 0 {
+			for _, bf := range valSepMeta.NewBlobFiles {
+				r.blobs = append(r.blobs, OutputBlob{
+					Stats:    bf.FileStats,
+					ObjMeta:  bf.FileObject,
+					Metadata: bf.FileMetadata,
+				})
+			}
 		}
 	}
 
@@ -231,9 +233,12 @@ func (r *Runner) WriteTable(
 		return
 	}
 	r.tables[len(r.tables)-1].WriterMeta = *writerMeta
-	r.stats.CumulativeWrittenSize += writerMeta.Size + valSepMeta.BlobFileStats.FileLen
+	r.stats.CumulativeWrittenSize += writerMeta.Size
 	r.stats.CumulativeBlobReferenceSize += valSepMeta.BlobReferenceSize
-	r.stats.CumulativeBlobFileSize += valSepMeta.BlobFileStats.FileLen
+	for _, bf := range valSepMeta.NewBlobFiles {
+		r.stats.CumulativeWrittenSize += bf.FileStats.FileLen
+		r.stats.CumulativeBlobFileSize += bf.FileStats.FileLen
+	}
 }
 
 func (r *Runner) writeKeysToTable(

valsep/value_separation.go

@@ -49,6 +49,15 @@ type ValueSeparation interface {
 	FinishOutput() (ValueSeparationMetadata, error)
 }
 
+// NewBlobFileInfo describes a newly written blob file
+// created by value separation. This is returned by
+// ValueSeparation.FinishOutput.
+type NewBlobFileInfo struct {
+	FileStats    blob.FileWriterStats
+	FileObject   objstorage.ObjectMetadata
+	FileMetadata *manifest.PhysicalBlobFile
+}
+
 // ValueSeparationMetadata describes metadata about a table's blob references,
 // and optionally a newly constructed blob file.
 type ValueSeparationMetadata struct {
@@ -57,9 +66,7 @@ type ValueSeparationMetadata struct {
 	BlobReferenceDepth manifest.BlobReferenceDepth
 
 	// The below fields are only populated if a new blob file was created.
-	BlobFileStats    blob.FileWriterStats
-	BlobFileObject   objstorage.ObjectMetadata
-	BlobFileMetadata *manifest.PhysicalBlobFile
+	NewBlobFiles []NewBlobFileInfo
 }
 
 // NeverSeparateValues is a ValueSeparation implementation that never separates