blob: handle sparseness in AddVirtualBlockMapping

When rewriting a blob file, we need record the mapping from the original blob
file's block ID to the new blob file's physical block in the 'virtual block'
column. Previously the onus was on the user of the BlobFileWriter to call
AddVirtualBlockMapping for every blockID in the original file, even for
blockIDs that have no extant values.

This commit removes the onus from the caller and handles populating sparseness
within the implementation of BlobFileWriter.AddVirtualBlockMapping. This fixes
a bug in blob-file rewriting when multiple blocks map to the first physical
block. Additionally, this commit updates the virtual column entries for empty
virtual blocks to reflect the sparseness storing 0xFFFFFFFF in the column
(semantically equivalent to an entry mapping to the 0'th physical block with a
valueID offset of 0xFFFFFFFF).

Author: jbowens

blob_rewrite.go

@@ -456,7 +456,12 @@ func (rw *blobFileRewriter) copyBlockValues(ctx context.Context, finishedBlock b
 	if shouldFlush {
 		rw.writer.ForceFlush()
 	}
+
+	// Record the mapping from the virtual block ID to the current physical
+	// block and value ID offset.
+	rw.writer.BeginNewVirtualBlock(finishedBlock.blockID)
 	slices.Sort(finishedBlock.liveValueIDs)
+
 	for i, valueID := range finishedBlock.liveValueIDs {
 		if i > 0 && finishedBlock.liveValueIDs[i-1]+1 != valueID {
 			// There's a gap in the referenced value IDs.
@@ -489,11 +494,6 @@ func (rw *blobFileRewriter) Rewrite(ctx context.Context) (blob.FileWriterStats,
 	// Begin constructing our output blob file. We maintain a map of blockID
 	// to accumulated liveness data across all referencing sstables.
 	firstBlock := heap.Pop(&rw.blkHeap).(*sstable.BlobRefLivenessEncoding)
-
-	// Add virtual block mappings for all blocks from 0 to the first block.
-	for blockID := blob.BlockID(0); blockID < firstBlock.BlockID; blockID++ {
-		rw.writer.BeginNewVirtualBlock(blockID)
-	}
 	pendingBlock := blockValues{
 		blockID:      firstBlock.BlockID,
 		valuesSize:   firstBlock.ValuesSize,
@@ -505,11 +505,6 @@ func (rw *blobFileRewriter) Rewrite(ctx context.Context) (blob.FileWriterStats,
 		// If we are encountering a new block, write the last accumulated block
 		// to the blob file.
 		if pendingBlock.blockID != nextBlock.BlockID {
-			// Add virtual block mappings for all blocks between the last block
-			// we encountered and the current block.
-			for blockID := pendingBlock.blockID; blockID < nextBlock.BlockID; blockID++ {
-				rw.writer.BeginNewVirtualBlock(blockID)
-			}
 			// Write the last accumulated block's values to the blob file.
 			if err := rw.copyBlockValues(ctx, pendingBlock); err != nil {
 				return blob.FileWriterStats{}, err
@@ -522,7 +517,6 @@ func (rw *blobFileRewriter) Rewrite(ctx context.Context) (blob.FileWriterStats,
 	}
 
 	// Copy the last accumulated block.
-	rw.writer.BeginNewVirtualBlock(pendingBlock.blockID)
 	if err := rw.copyBlockValues(ctx, pendingBlock); err != nil {
 		return blob.FileWriterStats{}, err
 	}

sstable/blob/blocks.go

@@ -95,9 +95,16 @@ func (e *indexBlockEncoder) AddBlockHandle(h block.Handle) {
 func (e *indexBlockEncoder) AddVirtualBlockMapping(
 	virtualBlockID BlockID, physicalBlockIndex int, valueIDOffset BlockValueID,
 ) {
-	if virtualBlockID != BlockID(e.countVirtualBlocks) {
+	// Require that virtual blocks are added in order.
+	if virtualBlockID < BlockID(e.countVirtualBlocks) {
 		panic(errors.AssertionFailedf("virtual block ID %d is out of order; expected %d", virtualBlockID, e.countVirtualBlocks))
 	}
+	// If there's a gap within the virtual block IDs, we fill in the gap with
+	// entries that clarify these blocks are empty.
+	for id := BlockID(e.countVirtualBlocks); id < virtualBlockID; id++ {
+		e.virtualBlocks.Set(int(id), virtualBlockIndexMask)
+		e.countVirtualBlocks++
+	}
 	e.virtualBlocks.Set(int(virtualBlockID), uint64(physicalBlockIndex)|(uint64(valueIDOffset)<<32))
 	e.countVirtualBlocks++
 }

sstable/blob/fetcher.go

@@ -261,6 +261,10 @@ func (cr *cachedReader) GetUnsafeValue(
 		var valueIDOffset BlockValueID
 		if cr.indexBlock.dec.virtualBlockCount > 0 {
 			physicalBlockIndex, valueIDOffset = cr.indexBlock.dec.RemapVirtualBlockID(vh.BlockID)
+			if valueIDOffset == virtualBlockIndexMask {
+				return nil, errors.AssertionFailedf("blob file indicates virtual block ID %d in %s should be unreferenced",
+					vh.BlockID, vh.BlobFileID)
+			}
 		}
 		invariants.CheckBounds(physicalBlockIndex, cr.indexBlock.dec.BlockCount())