compressionanalyzer: sample per byte instead of per block

We record compression ratio and block compression/decompression time
per block. This is ok when blocks are about the same size but are less
useful when there is larger variance (as in >128KB buckets).

We want the compression ratio to be the ratio between the total
decompressed size and the total compressed size, not the average
compression ratio among blocks. Similarly, we want to measure
performance in terms of CPU time per byte.

We switch to using frequency-weighted sampling (where frequency is the
size of the block) and we now show compression/decompression
performance in MB/s. Note that the standard deviation for the latter
is derived from the CPU time per second metric, it doesn't apply
directly to the MB/s figure.

Author: RaduBerinde

sstable/compressionanalyzer/block_analyzer.go

@@ -88,10 +88,10 @@ func (a *BlockAnalyzer) runExperiment(
 	}
 	decompressionTime := t2.Elapsed()
 
-	// CPU times are in microseconds.
-	pa.CompressionTime.Add(compressionTime.Seconds() * 1e6)
-	pa.DecompressionTime.Add(decompressionTime.Seconds() * 1e6)
-	pa.CompressionRatio.Add(float64(len(block)) / float64(len(compressed)))
+	// CPU times are in nanoseconds / byte.
+	pa.CompressionTime.Add(float64(compressionTime)/float64(len(block)), uint64(len(block)))
+	pa.DecompressionTime.Add(float64(decompressionTime)/float64(len(block)), uint64(len(block)))
+	pa.CompressionRatio.Add(float64(len(block))/float64(len(compressed)), uint64(len(block)))
 }
 
 func ensureLen(b []byte, n int) []byte {

sstable/compressionanalyzer/buckets.go

@@ -9,6 +9,7 @@ import (
 	"math"
 	"strings"
 	"text/tabwriter"
+	"time"
 
 	"github.com/cockroachdb/pebble/internal/compression"
 )
@@ -145,10 +146,10 @@ type Bucket struct {
 // PerSetting holds statistics from experiments on blocks in a bucket with a
 // specific compression.Setting.
 type PerSetting struct {
-	CompressionRatio Welford
-	// CPU times are in microseconds.
-	CompressionTime   Welford
-	DecompressionTime Welford
+	CompressionRatio WeightedWelford
+	// CPU times are in nanoseconds per byte.
+	CompressionTime   WeightedWelford
+	DecompressionTime WeightedWelford
 }
 
 func (b *Buckets) String(minSamples int) string {
@@ -167,19 +168,22 @@ func (b *Buckets) String(minSamples int) string {
 				if bucket.UncompressedSize.Count() < int64(minSamples) {
 					continue
 				}
-				fmt.Fprintf(tw, "%s\t%s\t%s\t%d\t%s\tCR", k, sz, c, bucket.UncompressedSize.Count(), withStdDev(bucket.UncompressedSize, "KB", 1.0/1024))
+				fmt.Fprintf(tw, "%s\t%s\t%s\t%d\t%.1fKB %s\tCR", k, sz, c, bucket.UncompressedSize.Count(), bucket.UncompressedSize.Mean()/1024, stdDevStr(bucket.UncompressedSize.Mean(), bucket.UncompressedSize.SampleStandardDeviation()))
 				for _, e := range (*b)[k][sz][c].Experiments {
-					fmt.Fprintf(tw, "\t%s", withStdDev(e.CompressionRatio, "", 1.0))
+					mean, stdDev := e.CompressionRatio.Mean(), e.CompressionRatio.SampleStandardDeviation()
+					fmt.Fprintf(tw, "\t%.2f %s", mean, stdDevStr(mean, stdDev))
 				}
 				fmt.Fprintf(tw, "\n")
 				fmt.Fprintf(tw, "\t\t\t\t\tComp")
 				for _, e := range (*b)[k][sz][c].Experiments {
-					fmt.Fprintf(tw, "\t%s", withStdDev(e.CompressionTime, "us", 1.0))
+					mean, stdDev := e.CompressionTime.Mean(), e.CompressionTime.SampleStandardDeviation()
+					fmt.Fprintf(tw, "\t%.0fMBps %s", toMBPS(mean), stdDevStr(mean, stdDev))
 				}
 				fmt.Fprintf(tw, "\n")
 				fmt.Fprintf(tw, "\t\t\t\t\tDecomp")
 				for _, e := range (*b)[k][sz][c].Experiments {
-					fmt.Fprintf(tw, "\t%s", withStdDev(e.DecompressionTime, "us", 1.0))
+					mean, stdDev := e.DecompressionTime.Mean(), e.DecompressionTime.SampleStandardDeviation()
+					fmt.Fprintf(tw, "\t%.0fMBps %s", toMBPS(mean), stdDevStr(mean, stdDev))
 				}
 				fmt.Fprintf(tw, "\n")
 			}
@@ -189,16 +193,22 @@ func (b *Buckets) String(minSamples int) string {
 	return buf.String()
 }
 
-func withStdDev(w Welford, units string, scale float64) string {
-	mean := w.Mean() * scale
-	if math.IsNaN(mean) {
-		mean = 0
+func toMBPS(nsPerByte float64) float64 {
+	if nsPerByte == 0 {
+		return 0
 	}
-	stddev := 0
-	if s := w.SampleStandardDeviation(); !math.IsNaN(s) {
-		stddev = int(100 * s / w.Mean())
+	const oneMB = 1 << 20
+	return float64(time.Second) / (nsPerByte * oneMB)
+}
+
+// stdDevStr formats the standard deviation as a percentage of the mean,
+// for example "± 10%".
+func stdDevStr(mean, stddev float64) string {
+	percent := 0
+	if mean > 0 {
+		percent = int(math.Round(100 * stddev / mean))
 	}
-	return fmt.Sprintf("%.1f%s ± %d%%", mean, units, stddev)
+	return fmt.Sprintf("± %d%%", percent)
 }
 
 func (b *Buckets) ToCSV(minSamples int) string {
@@ -207,9 +217,9 @@ func (b *Buckets) ToCSV(minSamples int) string {
 	for _, s := range Settings {
 		fmt.Fprintf(&buf, ",%s CR", s.String())
 		fmt.Fprintf(&buf, ",%s CR±", s.String())
-		fmt.Fprintf(&buf, ",%s Comp us", s.String())
+		fmt.Fprintf(&buf, ",%s Comp ns/b", s.String())
 		fmt.Fprintf(&buf, ",%s Comp±", s.String())
-		fmt.Fprintf(&buf, ",%s Decomp us", s.String())
+		fmt.Fprintf(&buf, ",%s Decomp ns/b", s.String())
 		fmt.Fprintf(&buf, ",%s Decomp±", s.String())
 	}
 	fmt.Fprintf(&buf, "\n")
@@ -222,9 +232,9 @@ func (b *Buckets) ToCSV(minSamples int) string {
 				}
 				fmt.Fprintf(&buf, "%s,%s,%s,%d,%.0f,%.0f", k, sz, c, bucket.UncompressedSize.Count(), bucket.UncompressedSize.Mean(), bucket.UncompressedSize.SampleStandardDeviation())
 				for _, e := range (*b)[k][sz][c].Experiments {
-					fmt.Fprintf(&buf, ",%.1f,%.1f", e.CompressionRatio.Mean(), e.CompressionRatio.SampleStandardDeviation())
-					fmt.Fprintf(&buf, ",%.1f,%.1f", e.CompressionTime.Mean(), e.CompressionTime.SampleStandardDeviation())
-					fmt.Fprintf(&buf, ",%.1f,%.1f", e.DecompressionTime.Mean(), e.DecompressionTime.SampleStandardDeviation())
+					fmt.Fprintf(&buf, ",%.3f,%.3f", e.CompressionRatio.Mean(), e.CompressionRatio.SampleStandardDeviation())
+					fmt.Fprintf(&buf, ",%.3f,%.3f", e.CompressionTime.Mean(), e.CompressionTime.SampleStandardDeviation())
+					fmt.Fprintf(&buf, ",%.3f,%.3f", e.DecompressionTime.Mean(), e.DecompressionTime.SampleStandardDeviation())
 				}
 				fmt.Fprintf(&buf, "\n")
 			}

sstable/compressionanalyzer/buckets_test.go

@@ -69,9 +69,10 @@ func exampleBuckets() Buckets {
 			b.UncompressedSize.Add(100 + float64(r.IntN(64*1024)))
 			for j := range b.Experiments {
 				e := &b.Experiments[j]
-				e.CompressionRatio.Add(float64(j+1) + 0.1*float64(r.IntN(10)))
-				e.CompressionTime.Add(float64((j+1)*10) + 0.1*float64(r.IntN(10)))
-				e.DecompressionTime.Add(float64((j+1)*100) + 0.1*float64(r.IntN(10)))
+				blockSize := uint64(50 + r.IntN(100))
+				e.CompressionRatio.Add(float64(j+1)+0.1*float64(r.IntN(10)), blockSize)
+				e.CompressionTime.Add(float64((j+1)*10)+0.1*float64(r.IntN(10)), blockSize)
+				e.DecompressionTime.Add(float64((j+1)*100)+0.1*float64(r.IntN(10)), blockSize)
 			}
 		}
 	}

sstable/compressionanalyzer/file_analyzer_test.go

@@ -36,10 +36,10 @@ func TestFileAnalyzer(t *testing.T) {
 							// Snappy always has the same output in all configurations and on
 							// all platforms.
 							if Settings[l].Algorithm != compression.SnappyAlgorithm {
-								bucket.Experiments[l].CompressionRatio = Welford{}
+								bucket.Experiments[l].CompressionRatio = WeightedWelford{}
 							}
-							bucket.Experiments[l].CompressionTime = Welford{}
-							bucket.Experiments[l].DecompressionTime = Welford{}
+							bucket.Experiments[l].CompressionTime = WeightedWelford{}
+							bucket.Experiments[l].DecompressionTime = WeightedWelford{}
 						}
 					}
 				}