metrics: expose new compression stats

Update the compression metrics to use the new compression statistics.

Author: RaduBerinde

compaction.go

@@ -3512,7 +3512,7 @@ func (c *tableCompaction) makeVersionEdit(result compact.Result) (*manifest.Vers
 		// If the file didn't contain any range deletions, we can fill its
 		// table stats now, avoiding unnecessarily loading the table later.
 		maybeSetStatsFromProperties(
-			fileMeta.PhysicalMeta(), &t.WriterMeta.Properties.CommonProperties, c.logger,
+			fileMeta.PhysicalMeta(), &t.WriterMeta.Properties, c.logger,
 		)
 
 		if t.WriterMeta.HasPointKeys {

db.go

@@ -2185,12 +2185,8 @@ func (d *DB) Metrics() *Metrics {
 	}
 	for i := 0; i < numLevels; i++ {
 		metrics.Levels[i].Additional.ValueBlocksSize = *valueBlockSizeAnnotator.LevelAnnotation(vers.Levels[i])
-		compressionTypes := compressionTypeAnnotator.LevelAnnotation(vers.Levels[i])
-		metrics.Table.CompressedCountUnknown += int64(compressionTypes.unknown)
-		metrics.Table.CompressedCountSnappy += int64(compressionTypes.snappy)
-		metrics.Table.CompressedCountZstd += int64(compressionTypes.zstd)
-		metrics.Table.CompressedCountMinLZ += int64(compressionTypes.minlz)
-		metrics.Table.CompressedCountNone += int64(compressionTypes.none)
+		compressionMetrics := compressionStatsAnnotator.LevelAnnotation(vers.Levels[i])
+		metrics.Compression.MergeWith(compressionMetrics)
 	}
 
 	metrics.Table.PendingStatsCollectionCount = int64(len(d.mu.tableStats.pending))

ingest.go

@@ -316,7 +316,7 @@ func ingestLoad1(
 	// disallowing removal of an open file. Under MemFS, if we don't populate
 	// meta.Stats here, the file will be loaded into the file cache for
 	// calculating stats before we can remove the original link.
-	maybeSetStatsFromProperties(meta.PhysicalMeta(), &props.CommonProperties, opts.Logger)
+	maybeSetStatsFromProperties(meta.PhysicalMeta(), &props, opts.Logger)
 
 	{
 		iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)

ingest_test.go

@@ -249,6 +249,7 @@ func TestIngestLoadRand(t *testing.T) {
 
 	for _, m := range lr.local {
 		m.CreationTime = 0
+		m.Stats.CompressionStats = block.CompressionStats{}
 	}
 	t.Log(strings.Join(pretty.Diff(expected, lr.local), "\n"))
 	require.Equal(t, expected, lr.local)

internal/manifest/table_metadata.go

@@ -1167,6 +1167,9 @@ type TableStats struct {
 	// CompressionType is the compression profile used for the table (or nil if
 	// the profile name is not recognized).
 	CompressionType *block.CompressionProfile
+	// CompressionStats contains compression statistics; not available for older
+	// table versions.
+	CompressionStats block.CompressionStats
 	// TombstoneDenseBlocksRatio is the ratio of data blocks in this table that
 	// fulfills at least one of the following:
 	// 1. The block contains at least options.Experimental.NumDeletionsThreshold

internal/manifest/table_metadata_test.go

@@ -148,7 +148,7 @@ func TestTableMetadataSize(t *testing.T) {
 	}
 	structSize := unsafe.Sizeof(TableMetadata{})
 
-	const tableMetadataSize = 320
+	const tableMetadataSize = 352
 	if structSize != tableMetadataSize {
 		t.Errorf("TableMetadata struct size (%d bytes) is not expected size (%d bytes)",
 			structSize, tableMetadataSize)

metrics.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/ascii/table"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/compression"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/manual"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/sharedcache"
@@ -347,18 +348,6 @@ type Metrics struct {
 		BackingTableCount uint64
 		// The sum of the sizes of the BackingTableCount sstables that are backing virtual tables.
 		BackingTableSize uint64
-		// The number of sstables that are compressed with an unknown compression
-		// algorithm.
-		CompressedCountUnknown int64
-		// The number of sstables that are compressed with the default compression
-		// algorithm, snappy.
-		CompressedCountSnappy int64
-		// The number of sstables that are compressed with zstd.
-		CompressedCountZstd int64
-		// The number of sstables that are compressed with minlz.
-		CompressedCountMinLZ int64
-		// The number of sstables that are uncompressed.
-		CompressedCountNone int64
 
 		// Local file sizes.
 		Local struct {
@@ -450,6 +439,8 @@ type Metrics struct {
 		}
 	}
 
+	Compression CompressionMetrics
+
 	FileCache FileCacheMetrics
 
 	// Count of the number of open sstable iterators.
@@ -498,6 +489,46 @@ type Metrics struct {
 	manualMemory manual.Metrics
 }
 
+// CompressionMetrics contains compression metrics for sstables and blob files.
+type CompressionMetrics struct {
+	// NoCompressionBytes is the total number of bytes in files that do 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.
+	NoCompressionBytes uint64
+	// CompressedBytesWithoutStats is the total number of bytes in files that do not
+	// encode compression statistics (or for which there are no statistics yet).
+	CompressedBytesWithoutStats uint64
+	Snappy                      CompressionStatsForSetting
+	MinLZ                       CompressionStatsForSetting
+	Zstd                        CompressionStatsForSetting
+}
+
+type CompressionStatsForSetting = block.CompressionStatsForSetting
+
+func (cm *CompressionMetrics) Add(stats *block.CompressionStats) {
+	for s, cs := range stats.All() {
+		switch s.Algorithm {
+		case compression.NoCompression:
+			cm.NoCompressionBytes += cs.UncompressedBytes
+		case compression.SnappyAlgorithm:
+			cm.Snappy.Add(cs)
+		case compression.MinLZ:
+			cm.MinLZ.Add(cs)
+		case compression.Zstd:
+			cm.Zstd.Add(cs)
+		}
+	}
+}
+
+func (cm *CompressionMetrics) MergeWith(o *CompressionMetrics) {
+	cm.NoCompressionBytes += o.NoCompressionBytes
+	cm.CompressedBytesWithoutStats += o.CompressedBytesWithoutStats
+	cm.Snappy.Add(o.Snappy)
+	cm.MinLZ.Add(o.MinLZ)
+	cm.Zstd.Add(o.Zstd)
+}
+
 var (
 	// FsyncLatencyBuckets are prometheus histogram buckets suitable for a histogram
 	// that records latencies for fsyncs.
@@ -751,9 +782,18 @@ var (
 		table.String("range dels", 15, table.AlignRight, func(i keysInfo) string { return i.rangeDels }),
 	)
 	compressionTableHeader = `COMPRESSION`
-	compressionTable       = table.Define[pair[string, int64]](
-		table.String("algorithm", 10, table.AlignRight, func(p pair[string, int64]) string { return p.k }),
-		table.Count("tables", 10, table.AlignRight, func(p pair[string, int64]) int64 { return p.v }),
+	compressionTable       = table.Define[pair[string, CompressionStatsForSetting]](
+		table.String("algorithm", 13, table.AlignRight, func(p pair[string, CompressionStatsForSetting]) string { return p.k }),
+		table.Bytes("on disk bytes", 13, table.AlignRight, func(p pair[string, CompressionStatsForSetting]) uint64 { return p.v.CompressedBytes }),
+		table.String("CR", 13, table.AlignRight, func(p pair[string, CompressionStatsForSetting]) string {
+			if p.v.UncompressedBytes == p.v.CompressedBytes {
+				return ""
+			}
+			if p.v.UncompressedBytes == 0 {
+				return "?"
+			}
+			return crhumanize.Float(p.v.CompressionRatio(), 2 /* precision */).String()
+		}),
 	)
 )
 
@@ -974,15 +1014,19 @@ func (m *Metrics) String() string {
 	cur = cur.NewlineReturn()
 
 	cur = cur.WriteString(compressionTableHeader).NewlineReturn()
-	compressionContents := []pair[string, int64]{
-		{k: "none", v: m.Table.CompressedCountNone},
-		{k: "snappy", v: m.Table.CompressedCountSnappy},
-		{k: "minlz", v: m.Table.CompressedCountMinLZ},
-		{k: "zstd", v: m.Table.CompressedCountZstd},
-	}
-	if m.Table.CompressedCountUnknown > 0 {
-		compressionContents = append(compressionContents, pair[string, int64]{k: "???", v: m.Table.CompressedCountUnknown})
+	compressionContents := []pair[string, CompressionStatsForSetting]{
+		{k: "none", v: CompressionStatsForSetting{
+			CompressedBytes:   m.Compression.NoCompressionBytes,
+			UncompressedBytes: m.Compression.NoCompressionBytes,
+		}},
+		{k: "snappy", v: m.Compression.Snappy},
+		{k: "minlz", v: m.Compression.MinLZ},
+		{k: "zstd", v: m.Compression.Zstd},
+		{k: "unknown", v: CompressionStatsForSetting{CompressedBytes: m.Compression.CompressedBytesWithoutStats}},
 	}
+	compressionContents = slices.DeleteFunc(compressionContents, func(p pair[string, CompressionStatsForSetting]) bool {
+		return p.v.CompressedBytes == 0
+	})
 	compressionTable.Render(cur, table.RenderOptions{Orientation: table.Horizontally}, slices.Values(compressionContents))
 
 	return wb.String()

metrics_test.go

@@ -22,7 +22,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/manifest"
-	"github.com/cockroachdb/pebble/internal/manual"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -130,10 +129,6 @@ func exampleMetrics() Metrics {
 	m.Table.ZombieCount = 18
 	m.Table.BackingTableCount = 1
 	m.Table.BackingTableSize = 2 * MB
-	m.Table.CompressedCountMinLZ = 32
-	m.Table.CompressedCountSnappy = 33
-	m.Table.CompressedCountZstd = 34
-	m.Table.CompressedCountNone = 35
 	m.Table.Local.LiveSize = 28 * GB
 	m.Table.Local.LiveCount = 10_000
 	m.Table.Local.ObsoleteSize = 29 * MB
@@ -160,6 +155,15 @@ func exampleMetrics() Metrics {
 	m.BlobFiles.Local.ZombieSize = 30 * MB
 	m.BlobFiles.Local.ZombieCount = 14
 
+	m.Compression.NoCompressionBytes = 100 * MB
+	m.Compression.CompressedBytesWithoutStats = 500 * MB
+	m.Compression.Snappy.CompressedBytes = 1 * GB
+	m.Compression.Snappy.UncompressedBytes = 2 * GB
+	m.Compression.MinLZ.CompressedBytes = 1 * GB
+	m.Compression.MinLZ.UncompressedBytes = 3 * GB
+	m.Compression.Zstd.CompressedBytes = 10 * GB
+	m.Compression.Zstd.UncompressedBytes = 50 * GB
+
 	m.FileCache.Size = 1 * MB
 	m.FileCache.TableCount = 180
 	m.FileCache.BlobFileCount = 181
@@ -418,9 +422,6 @@ func TestMetrics(t *testing.T) {
 			d.mu.Unlock()
 
 			m := d.Metrics()
-			// Don't show memory usage as that can depend on architecture, invariants
-			// tag, etc.
-			m.manualMemory = manual.Metrics{}
 			// Some subset of cases show non-determinism in cache hits/misses.
 			if td.HasArg("zero-cache-hits-misses") {
 				// Avoid non-determinism.
@@ -432,7 +433,7 @@ func TestMetrics(t *testing.T) {
 				}
 			}
 			var buf strings.Builder
-			fmt.Fprintf(&buf, "%s", m.StringForTests())
+			fmt.Fprintf(&buf, "%s\n", m.StringForTests())
 			if len(m.CategoryStats) > 0 {
 				fmt.Fprintf(&buf, "\nIter category stats:\n")
 				for _, stats := range m.CategoryStats {

sstable/block/compression_stats.go

@@ -11,6 +11,7 @@ import (
 	"slices"
 	"strings"
 
+	"github.com/cockroachdb/crlib/crmath"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/compression"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -37,13 +38,28 @@ type CompressionStatsForSetting struct {
 	UncompressedBytes uint64
 }
 
+// CompressionRatio returns the compression ratio for the setting.
+func (cs CompressionStatsForSetting) CompressionRatio() float64 {
+	return float64(cs.UncompressedBytes) / float64(cs.CompressedBytes)
+}
+
 func (cs *CompressionStatsForSetting) Add(other CompressionStatsForSetting) {
 	cs.CompressedBytes += other.CompressedBytes
 	cs.UncompressedBytes += other.UncompressedBytes
 }
 
-// add updates the stats to reflect a block that was compressed with the given setting.
-func (c *CompressionStats) add(setting compression.Setting, stats CompressionStatsForSetting) {
+func (c *CompressionStats) IsEmpty() bool {
+	return c.noCompressionBytes == 0 && c.fastest.CompressedBytes == 0 && len(c.others) == 0
+}
+
+func (c *CompressionStats) Reset() {
+	c.noCompressionBytes = 0
+	c.fastest = CompressionStatsForSetting{}
+	clear(c.others)
+}
+
+// addOne updates the stats to reflect a block that was compressed with the given setting.
+func (c *CompressionStats) addOne(setting compression.Setting, stats CompressionStatsForSetting) {
 	switch setting {
 	case compression.None:
 		c.noCompressionBytes += stats.UncompressedBytes
@@ -62,10 +78,10 @@ func (c *CompressionStats) add(setting compression.Setting, stats CompressionSta
 	}
 }
 
-// MergeWith updates the receiver stats to include the other stats.
-func (c *CompressionStats) MergeWith(other *CompressionStats) {
+// Add updates the receiver stats to include the other stats.
+func (c *CompressionStats) Add(other *CompressionStats) {
 	for s, cs := range other.All() {
-		c.add(s, cs)
+		c.addOne(s, cs)
 	}
 }
 
@@ -128,11 +144,32 @@ func (c CompressionStats) String() string {
 	return buf.String()
 }
 
+// Scale the stats by (size/backingSize). Used to obtain an approximation of the
+// stats for a virtual table.
+func (c *CompressionStats) Scale(size uint64, backingSize uint64) {
+	// Make sure the sizes are sane, just in case.
+	size = max(size, 1)
+	backingSize = max(backingSize, size)
+
+	c.noCompressionBytes = crmath.ScaleUint64(c.noCompressionBytes, size, backingSize)
+	c.fastest.CompressedBytes = crmath.ScaleUint64(c.fastest.CompressedBytes, size, backingSize)
+	c.fastest.UncompressedBytes = crmath.ScaleUint64(c.fastest.UncompressedBytes, size, backingSize)
+
+	for s, cs := range c.others {
+		cs.CompressedBytes = crmath.ScaleUint64(cs.CompressedBytes, size, backingSize)
+		cs.UncompressedBytes = crmath.ScaleUint64(cs.UncompressedBytes, size, backingSize)
+		c.others[s] = cs
+	}
+}
+
 // ParseCompressionStats parses the output of CompressionStats.String back into CompressionStats.
 //
 // If the string contains statistics for unknown compression settings, these are
 // accumulated under a special "unknown" setting.
 func ParseCompressionStats(s string) (CompressionStats, error) {
+	if s == "" {
+		return CompressionStats{}, nil
+	}
 	var stats CompressionStats
 	for _, a := range strings.Split(s, ",") {
 		b := strings.Split(a, ":")
@@ -147,7 +184,7 @@ func ParseCompressionStats(s string) (CompressionStats, error) {
 		if _, err := fmt.Sscanf(b[1], "%d/%d", &cs.CompressedBytes, &cs.UncompressedBytes); err != nil {
 			return CompressionStats{}, errors.Errorf("cannot parse compression stats %q", s)
 		}
-		stats.add(setting, cs)
+		stats.addOne(setting, cs)
 	}
 	return stats, nil
 }

sstable/block/compression_stats_test.go

@@ -15,27 +15,27 @@ import (
 func TestCompressionStatsString(t *testing.T) {
 	var stats CompressionStats
 
-	stats.add(compression.None, CompressionStatsForSetting{CompressedBytes: 100, UncompressedBytes: 100})
+	stats.addOne(compression.None, CompressionStatsForSetting{CompressedBytes: 100, UncompressedBytes: 100})
 	require.Equal(t, "NoCompression:100/100", stats.String())
 
-	stats.add(compression.Snappy, CompressionStatsForSetting{CompressedBytes: 100, UncompressedBytes: 200})
+	stats.addOne(compression.Snappy, CompressionStatsForSetting{CompressedBytes: 100, UncompressedBytes: 200})
 	require.Equal(t, "NoCompression:100/100,Snappy:100/200", stats.String())
 
-	stats.add(compression.Snappy, CompressionStatsForSetting{CompressedBytes: 100, UncompressedBytes: 200})
+	stats.addOne(compression.Snappy, CompressionStatsForSetting{CompressedBytes: 100, UncompressedBytes: 200})
 	require.Equal(t, "NoCompression:100/100,Snappy:200/400", stats.String())
 
-	stats.add(compression.MinLZFastest, CompressionStatsForSetting{CompressedBytes: 1000, UncompressedBytes: 4000})
+	stats.addOne(compression.MinLZFastest, CompressionStatsForSetting{CompressedBytes: 1000, UncompressedBytes: 4000})
 	require.Equal(t, "MinLZ1:1000/4000,NoCompression:100/100,Snappy:200/400", stats.String())
 
-	stats.add(compression.ZstdLevel1, CompressionStatsForSetting{CompressedBytes: 10000, UncompressedBytes: 80000})
+	stats.addOne(compression.ZstdLevel1, CompressionStatsForSetting{CompressedBytes: 10000, UncompressedBytes: 80000})
 	require.Equal(t, "MinLZ1:1000/4000,NoCompression:100/100,Snappy:200/400,ZSTD1:10000/80000", stats.String())
 
 	stats = CompressionStats{}
-	stats.add(compression.MinLZFastest, CompressionStatsForSetting{CompressedBytes: 1000, UncompressedBytes: 4000})
+	stats.addOne(compression.MinLZFastest, CompressionStatsForSetting{CompressedBytes: 1000, UncompressedBytes: 4000})
 	require.Equal(t, "MinLZ1:1000/4000", stats.String())
 
 	stats = CompressionStats{}
-	stats.add(compression.Snappy, CompressionStatsForSetting{CompressedBytes: 1000, UncompressedBytes: 4000})
+	stats.addOne(compression.Snappy, CompressionStatsForSetting{CompressedBytes: 1000, UncompressedBytes: 4000})
 	require.Equal(t, "Snappy:1000/4000", stats.String())
 }
 
@@ -49,7 +49,7 @@ func TestCompressionStatsRoundtrip(t *testing.T) {
 			if settings[i] != compression.None {
 				uncompressed += compressed * rand.Uint64N(20) / 10
 			}
-			stats.add(settings[i], CompressionStatsForSetting{CompressedBytes: compressed, UncompressedBytes: uncompressed})
+			stats.addOne(settings[i], CompressionStatsForSetting{CompressedBytes: compressed, UncompressedBytes: uncompressed})
 			str := stats.String()
 			stats2, err := ParseCompressionStats(str)
 			require.NoError(t, err)