db: aggregate blob file compression stats

We rename `manifest.Annotator` to `TableAnnotator` and add a
`BlobFileAnnotator` type. Internally, we use a generic `annotator`
type to share code.

We use a `BlobFileAnnotator` to report compression statistics for blob
files.

Author: RaduBerinde

compaction_picker.go

@@ -1557,62 +1557,58 @@ func (p *compactionPickerByScore) addScoresToPickedCompactionMetrics(
 	}
 }
 
-// elisionOnlyAnnotator is a manifest.Annotator that annotates B-Tree
+// elisionOnlyAnnotator is a manifest.TableAnnotator that annotates B-Tree
 // nodes with the *fileMetadata of a file meeting the obsolete keys criteria
 // for an elision-only compaction within the subtree. If multiple files meet
 // the criteria, it chooses whichever file has the lowest LargestSeqNum. The
 // lowest LargestSeqNum file will be the first eligible for an elision-only
 // compaction once snapshots less than or equal to its LargestSeqNum are closed.
-var elisionOnlyAnnotator = &manifest.Annotator[manifest.TableMetadata]{
-	Aggregator: manifest.PickFileAggregator{
-		Filter: func(f *manifest.TableMetadata) (eligible bool, cacheOK bool) {
-			if f.IsCompacting() {
-				return false, true
-			}
+var elisionOnlyAnnotator = manifest.NewTableAnnotator[manifest.TableMetadata](manifest.PickFileAggregator{
+	Filter: func(f *manifest.TableMetadata) (eligible bool, cacheOK bool) {
+		if f.IsCompacting() {
+			return false, true
+		}
 
-			backingProps, backingPropsValid := f.TableBacking.Properties()
-			stats, statsValid := f.Stats()
-			if !backingPropsValid || !statsValid {
-				return false, false
-			}
+		backingProps, backingPropsValid := f.TableBacking.Properties()
+		stats, statsValid := f.Stats()
+		if !backingPropsValid || !statsValid {
+			return false, false
+		}
 
-			// Bottommost files are large and not worthwhile to compact just
-			// to remove a few tombstones. Consider a file eligible only if
-			// either its own range deletions delete at least 10% of its data or
-			// its deletion tombstones make at least 10% of its entries.
-			//
-			// TODO(jackson): This does not account for duplicate user keys
-			// which may be collapsed. Ideally, we would have 'obsolete keys'
-			// statistics that would include tombstones, the keys that are
-			// dropped by tombstones and duplicated user keys. See #847.
-			//
-			// Note that tables that contain exclusively range keys (i.e. no point keys,
-			// `NumEntries` and `RangeDeletionsBytesEstimate` are both zero) are excluded
-			// from elision-only compactions.
-			// TODO(travers): Consider an alternative heuristic for elision of range-keys.
-			eligible = stats.RangeDeletionsBytesEstimate*10 >= f.Size || backingProps.NumDeletions*10 > backingProps.NumEntries
-			return eligible, true
-		},
-		Compare: func(f1 *manifest.TableMetadata, f2 *manifest.TableMetadata) bool {
-			return f1.LargestSeqNum < f2.LargestSeqNum
-		},
+		// Bottommost files are large and not worthwhile to compact just
+		// to remove a few tombstones. Consider a file eligible only if
+		// either its own range deletions delete at least 10% of its data or
+		// its deletion tombstones make at least 10% of its entries.
+		//
+		// TODO(jackson): This does not account for duplicate user keys
+		// which may be collapsed. Ideally, we would have 'obsolete keys'
+		// statistics that would include tombstones, the keys that are
+		// dropped by tombstones and duplicated user keys. See #847.
+		//
+		// Note that tables that contain exclusively range keys (i.e. no point keys,
+		// `NumEntries` and `RangeDeletionsBytesEstimate` are both zero) are excluded
+		// from elision-only compactions.
+		// TODO(travers): Consider an alternative heuristic for elision of range-keys.
+		eligible = stats.RangeDeletionsBytesEstimate*10 >= f.Size || backingProps.NumDeletions*10 > backingProps.NumEntries
+		return eligible, true
 	},
-}
+	Compare: func(f1 *manifest.TableMetadata, f2 *manifest.TableMetadata) bool {
+		return f1.LargestSeqNum < f2.LargestSeqNum
+	},
+})
 
-// markedForCompactionAnnotator is a manifest.Annotator that annotates B-Tree
+// markedForCompactionAnnotator is a manifest.TableAnnotator that annotates B-Tree
 // nodes with the *fileMetadata of a file that is marked for compaction
 // within the subtree. If multiple files meet the criteria, it chooses
 // whichever file has the lowest LargestSeqNum.
-var markedForCompactionAnnotator = &manifest.Annotator[manifest.TableMetadata]{
-	Aggregator: manifest.PickFileAggregator{
-		Filter: func(f *manifest.TableMetadata) (eligible bool, cacheOK bool) {
-			return f.MarkedForCompaction, true
-		},
-		Compare: func(f1 *manifest.TableMetadata, f2 *manifest.TableMetadata) bool {
-			return f1.LargestSeqNum < f2.LargestSeqNum
-		},
+var markedForCompactionAnnotator = manifest.NewTableAnnotator[manifest.TableMetadata](manifest.PickFileAggregator{
+	Filter: func(f *manifest.TableMetadata) (eligible bool, cacheOK bool) {
+		return f.MarkedForCompaction, true
 	},
-}
+	Compare: func(f1 *manifest.TableMetadata, f2 *manifest.TableMetadata) bool {
+		return f1.LargestSeqNum < f2.LargestSeqNum
+	},
+})
 
 // pickedCompactionFromCandidateFile creates a pickedCompaction from a *fileMetadata
 // with various checks to ensure that the file still exists in the expected level

db.go

@@ -504,15 +504,15 @@ type DB struct {
 			validating bool
 		}
 
-		// annotators contains various instances of manifest.Annotator which
+		// annotators contains various instances of manifest.TableAnnotator which
 		// should be protected from concurrent access.
 		annotators struct {
 			// totalFileSize is the sum of the size of all files in the
 			// database. This includes local, remote, and external sstables --
 			// along with blob files.
-			totalFileSize *manifest.Annotator[uint64]
-			remoteSize    *manifest.Annotator[uint64]
-			externalSize  *manifest.Annotator[uint64]
+			totalFileSize *manifest.TableAnnotator[uint64]
+			remoteSize    *manifest.TableAnnotator[uint64]
+			externalSize  *manifest.TableAnnotator[uint64]
 		}
 	}
 
@@ -2165,14 +2165,18 @@ func (d *DB) Metrics() *Metrics {
 	if d.mu.compact.flushing {
 		metrics.Flush.NumInProgress = 1
 	}
+
+	metrics.Table.PendingStatsCollectionCount = int64(len(d.mu.tableStats.pending))
+	metrics.Table.InitialStatsCollectionComplete = d.mu.tableStats.loadedInitial
+
 	for i := 0; i < numLevels; i++ {
 		metrics.Levels[i].Additional.ValueBlocksSize = *valueBlockSizeAnnotator.LevelAnnotation(vers.Levels[i])
 		compressionMetrics := compressionStatsAnnotator.LevelAnnotation(vers.Levels[i])
 		metrics.Table.Compression.MergeWith(compressionMetrics)
 	}
 
-	metrics.Table.PendingStatsCollectionCount = int64(len(d.mu.tableStats.pending))
-	metrics.Table.InitialStatsCollectionComplete = d.mu.tableStats.loadedInitial
+	blobCompressionMetrics := blobCompressionStatsAnnotator.Annotation(&vers.BlobFiles)
+	metrics.BlobFiles.Compression.MergeWith(blobCompressionMetrics)
 
 	d.mu.Unlock()
 
@@ -2398,31 +2402,29 @@ func (d *DB) SSTables(opts ...SSTablesOption) ([][]SSTableInfo, error) {
 // referenced blob files.
 func (d *DB) makeFileSizeAnnotator(
 	filter func(f *manifest.TableMetadata) bool,
-) *manifest.Annotator[uint64] {
-	return &manifest.Annotator[uint64]{
-		Aggregator: manifest.SumAggregator{
-			AccumulateFunc: func(f *manifest.TableMetadata) (uint64, bool) {
-				if filter(f) {
-					return f.Size + f.EstimatedReferenceSize(), true
-				}
-				return 0, true
-			},
-			AccumulatePartialOverlapFunc: func(f *manifest.TableMetadata, bounds base.UserKeyBounds) uint64 {
-				if filter(f) {
-					overlappingFileSize, err := d.fileCache.estimateSize(f, bounds.Start, bounds.End.Key)
-					if err != nil {
-						return 0
-					}
-					overlapFraction := float64(overlappingFileSize) / float64(f.Size)
-					// Scale the blob reference size proportionally to the file
-					// overlap from the bounds to approximate only the blob
-					// references that overlap with the requested bounds.
-					return overlappingFileSize + uint64(float64(f.EstimatedReferenceSize())*overlapFraction)
+) *manifest.TableAnnotator[uint64] {
+	return manifest.NewTableAnnotator[uint64](manifest.SumAggregator{
+		AccumulateFunc: func(f *manifest.TableMetadata) (uint64, bool) {
+			if filter(f) {
+				return f.Size + f.EstimatedReferenceSize(), true
+			}
+			return 0, true
+		},
+		AccumulatePartialOverlapFunc: func(f *manifest.TableMetadata, bounds base.UserKeyBounds) uint64 {
+			if filter(f) {
+				overlappingFileSize, err := d.fileCache.estimateSize(f, bounds.Start, bounds.End.Key)
+				if err != nil {
+					return 0
 				}
-				return 0
-			},
+				overlapFraction := float64(overlappingFileSize) / float64(f.Size)
+				// Scale the blob reference size proportionally to the file
+				// overlap from the bounds to approximate only the blob
+				// references that overlap with the requested bounds.
+				return overlappingFileSize + uint64(float64(f.EstimatedReferenceSize())*overlapFraction)
+			}
+			return 0
 		},
-	}
+	})
 }
 
 // EstimateDiskUsage returns the estimated filesystem space used in bytes for