metrics: add compression counters

Add running counters that keep track of how many bytes were compressed
and decompressed. The counters are segregated along the same lines
where compression settings can differ: L5 vs L6 vs other levels, and
data vs value vs other blocks.

The intention is to estimate the CPU usage change for a different
compression profile (in conjunction with data about each algorithm's
performance, as obtained by the compression analyzer).

Author: RaduBerinde

compaction.go

@@ -2959,6 +2959,8 @@ func (d *DB) runCopyCompaction(
 		var wrote uint64
 		err = d.fileCache.withReader(ctx, block.NoReadEnv, inputMeta.VirtualMeta(), func(r *sstable.Reader, env sstable.ReadEnv) error {
 			var err error
+			writerOpts := d.opts.MakeWriterOptions(c.outputLevel.level, d.TableFormat())
+			writerOpts.CompressionCounters = d.compressionCounters.Compressed.ForLevel(base.MakeLevel(c.outputLevel.level))
 			// TODO(radu): plumb a ReadEnv to CopySpan (it could use the buffer pool
 			// or update category stats).
 			wrote, err = sstable.CopySpan(ctx,

db.go

@@ -552,6 +552,8 @@ type DB struct {
 	// compaction concurrency
 	openedAt time.Time
 
+	compressionCounters block.CompressionCounters
+
 	iterTracker *inflight.Tracker
 }
 
@@ -2014,6 +2016,9 @@ func (d *DB) Metrics() *Metrics {
 	blobCompressionMetrics := blobCompressionStatsAnnotator.Annotation(&vers.BlobFiles)
 	metrics.BlobFiles.Compression.MergeWith(&blobCompressionMetrics)
 
+	metrics.CompressionCounters.LogicalBytesCompressed = d.compressionCounters.LoadCompressed()
+	metrics.CompressionCounters.LogicalBytesDecompressed = d.compressionCounters.LoadDecompressed()
+
 	metrics.BlockCache = d.opts.Cache.Metrics()
 	metrics.FileCache, metrics.Filter = d.fileCache.Metrics()
 	metrics.TableIters = d.fileCache.IterCount()

flushable_test.go

@@ -58,7 +58,7 @@ func TestIngestedSSTFlushableAPI(t *testing.T) {
 
 		// We can reuse the ingestLoad function for this test even if we're
 		// not actually ingesting a file.
-		lr, err := ingestLoad(context.Background(), d.opts, d.FormatMajorVersion(), paths, nil, nil, d.cacheHandle, pendingOutputs)
+		lr, err := ingestLoad(context.Background(), d.opts, d.FormatMajorVersion(), paths, nil, nil, d.cacheHandle, &d.compressionCounters, pendingOutputs)
 		if err != nil {
 			t.Fatal(err)
 		}

ingest.go

@@ -257,6 +257,7 @@ func ingestLoad1(
 	fmv FormatMajorVersion,
 	readable objstorage.Readable,
 	cacheHandle *cache.Handle,
+	compressionCounters *block.CompressionCounters,
 	tableNum base.TableNum,
 	rangeKeyValidator rangeKeyIngestValidator,
 ) (
@@ -270,6 +271,9 @@ func ingestLoad1(
 		CacheHandle: cacheHandle,
 		FileNum:     base.PhysicalTableDiskFileNum(tableNum),
 	}
+	if compressionCounters != nil {
+		o.CompressionCounters = &compressionCounters.Decompressed
+	}
 	r, err := sstable.NewReader(ctx, readable, o)
 	if err != nil {
 		return nil, keyspan.Span{}, base.BlockReadStats{}, errors.CombineErrors(err, readable.Close())
@@ -498,6 +502,7 @@ func ingestLoad(
 	shared []SharedSSTMeta,
 	external []ExternalFile,
 	cacheHandle *cache.Handle,
+	compressionCounters *block.CompressionCounters,
 	pending []base.TableNum,
 ) (ingestLoadResult, error) {
 	localFileNums := pending[:len(paths)]
@@ -531,7 +536,7 @@ func ingestLoad(
 		if !shouldDisableRangeKeyChecks {
 			rangeKeyValidator = validateSuffixedBoundaries(opts.Comparer, lastRangeKey)
 		}
-		m, lastRangeKey, blockReadStats, err = ingestLoad1(ctx, opts, fmv, readable, cacheHandle, localFileNums[i], rangeKeyValidator)
+		m, lastRangeKey, blockReadStats, err = ingestLoad1(ctx, opts, fmv, readable, cacheHandle, compressionCounters, localFileNums[i], rangeKeyValidator)
 		if err != nil {
 			return ingestLoadResult{}, err
 		}
@@ -1480,7 +1485,7 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 	// Load the metadata for all the files being ingested. This step detects
 	// and elides empty sstables.
 	loadResult, err := ingestLoad(ctx, d.opts, d.FormatMajorVersion(), paths, shared, external,
-		d.cacheHandle, pendingOutputs)
+		d.cacheHandle, &d.compressionCounters, pendingOutputs)
 	if err != nil {
 		return IngestOperationStats{}, err
 	}

ingest_test.go

@@ -148,7 +148,7 @@ func TestIngestLoad(t *testing.T) {
 				FS:         mem,
 			}
 			opts.WithFSDefaults()
-			lr, err := ingestLoad(context.Background(), opts, dbVersion, []string{"ext"}, nil, nil, nil, []base.TableNum{1})
+			lr, err := ingestLoad(context.Background(), opts, dbVersion, []string{"ext"}, nil, nil, nil, nil, []base.TableNum{1})
 			if err != nil {
 				return err.Error()
 			}
@@ -247,7 +247,7 @@ func TestIngestLoadRand(t *testing.T) {
 	}
 	opts.WithFSDefaults()
 	opts.EnsureDefaults()
-	lr, err := ingestLoad(context.Background(), opts, version, paths, nil, nil, nil, pending)
+	lr, err := ingestLoad(context.Background(), opts, version, paths, nil, nil, nil, nil, pending)
 	require.NoError(t, err)
 
 	// Reset flaky stats.
@@ -272,7 +272,7 @@ func TestIngestLoadInvalid(t *testing.T) {
 		FS:       mem,
 	}
 	opts.WithFSDefaults()
-	if _, err := ingestLoad(context.Background(), opts, internalFormatNewest, []string{"invalid"}, nil, nil, nil, []base.TableNum{1}); err == nil {
+	if _, err := ingestLoad(context.Background(), opts, internalFormatNewest, []string{"invalid"}, nil, nil, nil, nil, []base.TableNum{1}); err == nil {
 		t.Fatalf("expected error, but found success")
 	}
 }

internal.go

@@ -4,7 +4,10 @@
 
 package pebble
 
-import "github.com/cockroachdb/pebble/internal/base"
+import (
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/sstable/block"
+)
 
 // SeqNum exports the base.SeqNum type.
 type SeqNum = base.SeqNum
@@ -80,3 +83,5 @@ type ShortAttribute = base.ShortAttribute
 // LazyValue.Clone requires a pointer to a LazyFetcher struct to avoid
 // allocations. No code outside Pebble needs to peer into a LazyFetcher.
 type LazyFetcher = base.LazyFetcher
+
+type CompressionCounters = block.CompressionCounters

metrics.go

@@ -256,7 +256,7 @@ type Metrics struct {
 			Zombie metrics.CountAndSizeByPlacement
 		}
 
-		// Compression statistics for sstable data (does not include blob files).
+		// Compression statistics for the live sstables.
 		Compression CompressionMetrics
 
 		// Garbage bytes.
@@ -322,9 +322,17 @@ type Metrics struct {
 		// value sizes.
 		ReferencedBackingValueSize uint64
 
+		// Compression statistics for the live blob files.
 		Compression CompressionMetrics
 	}
 
+	// CompressionCounters are cumulative counters for the number of logical
+	// (uncompressed) bytes that went through compression and decompression.
+	CompressionCounters struct {
+		LogicalBytesCompressed   block.ByLevel[block.ByKind[uint64]]
+		LogicalBytesDecompressed block.ByLevel[block.ByKind[uint64]]
+	}
+
 	FileCache FileCacheMetrics
 
 	// Count of the number of open sstable iterators.
@@ -465,7 +473,7 @@ type KeysMetrics struct {
 
 // CompressionMetrics contains compression metrics for sstables or blob files.
 type CompressionMetrics struct {
-	// NoCompressionBytes is the total number of bytes in files that do are not
+	// NoCompressionBytes is the total number of bytes in files that are not
 	// compressed. Data can be uncompressed when 1) compression is disabled; 2)
 	// for certain special types of blocks; and 3) for blocks that are not
 	// compressible.
@@ -790,6 +798,17 @@ var (
 		table.Div(),
 		table.String("blob files", 13, table.AlignRight, func(i compressionInfo) string { return i.blobFiles }),
 	)
+	compressionCountersTableHeader = `        Logical bytes compressed / decompressed`
+
+	compressionCountersTable = table.Define[compressionCountersInfo](
+		table.String("level", 5, table.AlignRight, func(i compressionCountersInfo) string { return i.level }),
+		table.Div(),
+		table.String("data blocks", 14, table.AlignCenter, func(i compressionCountersInfo) string { return i.DataBlocks }),
+		table.Div(),
+		table.String("value blocks", 14, table.AlignCenter, func(i compressionCountersInfo) string { return i.ValueBlocks }),
+		table.Div(),
+		table.String("other blocks", 14, table.AlignCenter, func(i compressionCountersInfo) string { return i.OtherBlocks }),
+	)
 	deletePacerTableHeader = `DELETE PACER`
 	deletePacerTable       = table.Define[deletePacerInfo](
 		table.String("", 14, table.AlignRight, func(i deletePacerInfo) string { return i.label }),
@@ -921,6 +940,34 @@ func makeCompressionInfo(algorithm string, table, blob CompressionStatsForSettin
 	return i
 }
 
+type compressionCountersInfo struct {
+	level string
+	block.ByKind[string]
+}
+
+func makeCompressionCountersInfo(m *Metrics) []compressionCountersInfo {
+	var result []compressionCountersInfo
+	isZero := func(c *block.ByKind[uint64]) bool {
+		return c.DataBlocks == 0 && c.ValueBlocks == 0 && c.OtherBlocks == 0
+	}
+	addLevel := func(level string, compressed, decompressed *block.ByKind[uint64]) {
+		if isZero(compressed) && isZero(decompressed) {
+			return
+		}
+		result = append(result, compressionCountersInfo{
+			level: level,
+			ByKind: block.ByKind[string]{
+				DataBlocks:  humanizeBytes(compressed.DataBlocks) + " / " + humanizeBytes(decompressed.DataBlocks),
+				ValueBlocks: humanizeBytes(compressed.ValueBlocks) + " / " + humanizeBytes(decompressed.ValueBlocks),
+				OtherBlocks: humanizeBytes(compressed.OtherBlocks) + " / " + humanizeBytes(decompressed.OtherBlocks)},
+		})
+	}
+	addLevel("L0-L4", &m.CompressionCounters.LogicalBytesCompressed.OtherLevels, &m.CompressionCounters.LogicalBytesDecompressed.OtherLevels)
+	addLevel("L5", &m.CompressionCounters.LogicalBytesCompressed.L5, &m.CompressionCounters.LogicalBytesDecompressed.L5)
+	addLevel("L6", &m.CompressionCounters.LogicalBytesCompressed.L6, &m.CompressionCounters.LogicalBytesDecompressed.L6)
+	return result
+}
+
 // String pretty-prints the metrics.
 //
 // See testdata/metrics for an example.
@@ -1112,6 +1159,10 @@ func (m *Metrics) String() string {
 	})
 	cur = compressionTable.Render(cur, table.RenderOptions{}, compressionContents...)
 
+	cur = cur.NewlineReturn()
+	cur = cur.WriteString(compressionCountersTableHeader).NewlineReturn()
+	cur = compressionCountersTable.Render(cur, table.RenderOptions{}, makeCompressionCountersInfo(m)...)
+
 	cur = cur.NewlineReturn()
 	cur.WriteString(deletePacerTableHeader)
 	deletePacerContents := []deletePacerInfo{
@@ -1162,8 +1213,8 @@ func (m *Metrics) StringForTests() string {
 
 	// We recalculate the file cache size using the 64-bit sizes, and we ignore
 	// the genericcache metadata size which is harder to adjust.
-	const sstableReaderSize64bit = 280
-	const blobFileReaderSize64bit = 112
+	const sstableReaderSize64bit = 288
+	const blobFileReaderSize64bit = 120
 	mCopy.FileCache.Size = mCopy.FileCache.TableCount*sstableReaderSize64bit + mCopy.FileCache.BlobFileCount*blobFileReaderSize64bit
 	if math.MaxInt == math.MaxInt64 {
 		// Verify the 64-bit sizes, so they are kept updated.

metrics_test.go

@@ -174,6 +174,24 @@ func exampleMetrics() Metrics {
 	m.BlobFiles.Compression.Zstd.CompressedBytes = 100 * GB
 	m.BlobFiles.Compression.Zstd.UncompressedBytes = 500 * GB
 
+	byKind := func(n uint64) block.ByKind[uint64] {
+		return block.ByKind[uint64]{
+			DataBlocks:  n * 10 * GB,
+			ValueBlocks: n * 100 * GB,
+			OtherBlocks: n * GB,
+		}
+	}
+	m.CompressionCounters.LogicalBytesCompressed = block.ByLevel[block.ByKind[uint64]]{
+		L5:          byKind(5),
+		L6:          byKind(6),
+		OtherLevels: byKind(1),
+	}
+	m.CompressionCounters.LogicalBytesDecompressed = block.ByLevel[block.ByKind[uint64]]{
+		L5:          byKind(50),
+		L6:          byKind(60),
+		OtherLevels: byKind(10),
+	}
+
 	m.FileCache.Size = 1 * MB
 	m.FileCache.TableCount = 180
 	m.FileCache.BlobFileCount = 181

open.go

@@ -319,7 +319,9 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		opts.FileCache = NewFileCache(opts.Experimental.FileCacheShards, fileCacheSize)
 		defer opts.FileCache.Unref()
 	}
-	d.fileCache = opts.FileCache.newHandle(d.cacheHandle, d.objProvider, d.opts.LoggerAndTracer, d.opts.MakeReaderOptions(), d.reportCorruption)
+	fileCacheReaderOpts := d.opts.MakeReaderOptions()
+	fileCacheReaderOpts.CompressionCounters = &d.compressionCounters.Decompressed
+	d.fileCache = opts.FileCache.newHandle(d.cacheHandle, d.objProvider, d.opts.LoggerAndTracer, fileCacheReaderOpts, d.reportCorruption)
 	d.newIters = d.fileCache.newIters
 	d.tableNewRangeKeyIter = tableNewRangeKeyIter(d.newIters)
 

options.go

@@ -2692,17 +2692,20 @@ func (o *Options) MakeWriterOptions(level int, format sstable.TableFormat) sstab
 // makeWriterOptions constructs sstable.WriterOptions for the specified level
 // using the current DB options and format.
 func (d *DB) makeWriterOptions(level int) sstable.WriterOptions {
-	return d.opts.MakeWriterOptions(level, d.TableFormat())
+	o := d.opts.MakeWriterOptions(level, d.TableFormat())
+	o.CompressionCounters = d.compressionCounters.Compressed.ForLevel(base.MakeLevel(level))
+	return o
 }
 
 // makeBlobWriterOptions constructs blob.FileWriterOptions using the current DB
 // options and format.
 func (d *DB) makeBlobWriterOptions(level int) blob.FileWriterOptions {
 	lo := &d.opts.Levels[level]
 	return blob.FileWriterOptions{
-		Format:       d.BlobFileFormat(),
-		Compression:  lo.Compression(),
-		ChecksumType: block.ChecksumTypeCRC32c,
+		Format:              d.BlobFileFormat(),
+		Compression:         lo.Compression(),
+		CompressionCounters: d.compressionCounters.Compressed.ForLevel(base.MakeLevel(level)),
+		ChecksumType:        block.ChecksumTypeCRC32c,
 		FlushGovernor: block.MakeFlushGovernor(
 			lo.BlockSize,
 			lo.BlockSizeThreshold,