manifest: maintain pointers to *TableMetadata for blob references

Previously the CurrentBlobFileSet maintained just the TableNum of referencing
tables. Blob file rewrites will require reading the referencing tables, which
requires the *TableMetadata.

Author: jbowens

internal/manifest/blob_metadata.go

@@ -490,7 +490,7 @@ type currentBlobFile struct {
 	// referencing table number. This would likely be more memory efficient,
 	// reduce overall number of pointers to chase and suffer fewer allocations
 	// (and we can pool the B-Tree nodes to further reduce allocs)
-	references map[base.TableNum]struct{}
+	references map[*TableMetadata]struct{}
 	// referencedValueSize is the sum of the length of uncompressed values in
 	// this blob file that are still live.
 	referencedValueSize uint64
@@ -576,7 +576,7 @@ func (s *CurrentBlobFileSet) Init(bve *BulkVersionEdit, h BlobRewriteHeuristic)
 	for blobFileID, pbf := range bve.BlobFiles.Added {
 		s.files[blobFileID] = &currentBlobFile{
 			metadata:   BlobFileMetadata{FileID: blobFileID, Physical: pbf},
-			references: make(map[base.TableNum]struct{}),
+			references: make(map[*TableMetadata]struct{}),
 		}
 		s.stats.Count++
 		s.stats.PhysicalSize += pbf.Size
@@ -591,7 +591,7 @@ func (s *CurrentBlobFileSet) Init(bve *BulkVersionEdit, h BlobRewriteHeuristic)
 				if !ok {
 					panic(errors.AssertionFailedf("pebble: referenced blob file %d not found", ref.FileID))
 				}
-				cbf.references[table.TableNum] = struct{}{}
+				cbf.references[table] = struct{}{}
 				cbf.referencedValueSize += ref.ValueSize
 				s.stats.ReferencedValueSize += ref.ValueSize
 				s.stats.ReferencesCount++
@@ -724,7 +724,7 @@ func (s *CurrentBlobFileSet) ApplyAndUpdateVersionEdit(ve *VersionEdit) error {
 		}
 
 		blobFileID := m.FileID
-		cbf := &currentBlobFile{references: make(map[base.TableNum]struct{})}
+		cbf := &currentBlobFile{references: make(map[*TableMetadata]struct{})}
 		cbf.metadata = BlobFileMetadata{FileID: blobFileID, Physical: m.Physical}
 		s.files[blobFileID] = cbf
 		s.stats.Count++
@@ -742,7 +742,7 @@ func (s *CurrentBlobFileSet) ApplyAndUpdateVersionEdit(ve *VersionEdit) error {
 			if !ok {
 				return errors.AssertionFailedf("pebble: referenced blob file %d not found", ref.FileID)
 			}
-			cbf.references[e.Meta.TableNum] = struct{}{}
+			cbf.references[e.Meta] = struct{}{}
 			cbf.referencedValueSize += ref.ValueSize
 			s.stats.ReferencedValueSize += ref.ValueSize
 			s.stats.ReferencesCount++
@@ -765,7 +765,7 @@ func (s *CurrentBlobFileSet) ApplyAndUpdateVersionEdit(ve *VersionEdit) error {
 					return errors.AssertionFailedf("pebble: referenced value size %d for blob file %s is greater than the referenced value size %d",
 						ref.ValueSize, cbf.metadata.FileID, cbf.referencedValueSize)
 				}
-				if _, ok := cbf.references[meta.TableNum]; !ok {
+				if _, ok := cbf.references[meta]; !ok {
 					return errors.AssertionFailedf("pebble: deleted table %s's reference to blob file %s not known",
 						meta.TableNum, ref.FileID)
 				}
@@ -786,7 +786,7 @@ func (s *CurrentBlobFileSet) ApplyAndUpdateVersionEdit(ve *VersionEdit) error {
 				continue
 			}
 			// Remove the reference of this table to this blob file.
-			delete(cbf.references, meta.TableNum)
+			delete(cbf.references, meta)
 
 			// If there are no more references to the blob file, remove it from
 			// the set and add the removal of the blob file to the version edit.