compressionanalyzer: support adaptive compressors

Switch experiments from using `compression.Setting` to
`block.CompressionProfile` and add two adaptive profiles.

Author: RaduBerinde

sstable/block/compression.go

@@ -217,7 +217,7 @@ func (i CompressionIndicator) String() string {
 	}
 }
 
-func (i CompressionIndicator) algorithm() compression.Algorithm {
+func (i CompressionIndicator) Algorithm() compression.Algorithm {
 	switch i {
 	case NoCompressionIndicator:
 		return compression.NoCompression

sstable/block/compressor.go

@@ -97,5 +97,5 @@ var noopCompressor = MakeCompressor(NoCompression)
 type Decompressor = compression.Decompressor
 
 func GetDecompressor(c CompressionIndicator) Decompressor {
-	return compression.GetDecompressor(c.algorithm())
+	return compression.GetDecompressor(c.Algorithm())
 }

sstable/compressionanalyzer/block_analyzer.go

@@ -17,7 +17,7 @@ import (
 // compression algorithms on sstable and blob file blocks.
 type BlockAnalyzer struct {
 	b             Buckets
-	compressors   [numSettings]compression.Compressor
+	compressors   [numProfiles]block.Compressor
 	decompressors [compression.NumAlgorithms]compression.Decompressor
 	minLZFastest  compression.Compressor
 	buf1          []byte
@@ -26,8 +26,8 @@ type BlockAnalyzer struct {
 
 func NewBlockAnalyzer() *BlockAnalyzer {
 	a := &BlockAnalyzer{}
-	for i, s := range Settings {
-		a.compressors[i] = compression.GetCompressor(s)
+	for i, p := range Profiles {
+		a.compressors[i] = block.MakeCompressor(p)
 	}
 	for i := range a.decompressors {
 		a.decompressors[i] = compression.GetDecompressor(compression.Algorithm(i))
@@ -38,6 +38,15 @@ func NewBlockAnalyzer() *BlockAnalyzer {
 	return a
 }
 
+// ResetCompressors the compressors. This is useful for adaptive compressors
+// which keep some state; we want to clear that state for each sstable.
+func (a *BlockAnalyzer) ResetCompressors() {
+	for i, p := range Profiles {
+		a.compressors[i].Close()
+		a.compressors[i] = block.MakeCompressor(p)
+	}
+}
+
 func (a *BlockAnalyzer) Close() {
 	for _, c := range a.compressors {
 		c.Close()
@@ -57,8 +66,8 @@ func (a *BlockAnalyzer) Block(kind block.Kind, block []byte) {
 	compressibility := MakeCompressibility(len(block), len(compressed))
 	bucket := &a.b[kind][size][compressibility]
 	bucket.UncompressedSize.Add(float64(len(block)))
-	for i := range Settings {
-		a.runExperiment(&bucket.Experiments[i], block, a.compressors[i], a.decompressors)
+	for i := range a.compressors {
+		a.runExperiment(&bucket.Experiments[i], block, kind, &a.compressors[i], a.decompressors)
 	}
 }
 
@@ -67,9 +76,10 @@ func (a *BlockAnalyzer) Buckets() *Buckets {
 }
 
 func (a *BlockAnalyzer) runExperiment(
-	pa *PerSetting,
+	pa *PerProfile,
 	block []byte,
-	compressor compression.Compressor,
+	blockKind block.Kind,
+	compressor *block.Compressor,
 	decompressors [compression.NumAlgorithms]compression.Decompressor,
 ) {
 	// buf1 will hold the compressed data; it can get a bit larger in the worst
@@ -80,14 +90,14 @@ func (a *BlockAnalyzer) runExperiment(
 	// Compress.
 	runtime.Gosched()
 	t1 := crtime.NowMono()
-	compressed, setting := compressor.Compress(a.buf1[:0], block)
+	ci, compressed := compressor.Compress(a.buf1[:0], block, blockKind)
 	compressionTime := t1.Elapsed()
 
 	// Yield the processor, reducing the chance that we get preempted during
 	// DecompressInto.
 	runtime.Gosched()
 	t2 := crtime.NowMono()
-	if err := decompressors[setting.Algorithm].DecompressInto(a.buf2, compressed); err != nil {
+	if err := decompressors[ci.Algorithm()].DecompressInto(a.buf2, compressed); err != nil {
 		panic(err)
 	}
 	decompressionTime := t2.Elapsed()

sstable/compressionanalyzer/buckets.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/pebble/internal/compression"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/block/blockkind"
 )
 
@@ -95,19 +96,69 @@ func (c Compressibility) String() string {
 	}
 }
 
-var Settings = [...]compression.Setting{
-	compression.Snappy,
-	compression.MinLZFastest,
-	compression.MinLZBalanced,
-	compression.ZstdLevel1,
-	compression.ZstdLevel3,
+var Profiles = [...]*block.CompressionProfile{
+	{
+		Name:                "Snappy",
+		DataBlocks:          compression.Snappy,
+		ValueBlocks:         compression.Snappy,
+		OtherBlocks:         compression.Snappy,
+		MinReductionPercent: 0,
+	},
+
+	{
+		Name:                "MinLZ1",
+		DataBlocks:          compression.MinLZFastest,
+		ValueBlocks:         compression.MinLZFastest,
+		OtherBlocks:         compression.MinLZFastest,
+		MinReductionPercent: 0,
+	},
+
+	{
+		Name:                "MinLZ2",
+		DataBlocks:          compression.MinLZBalanced,
+		ValueBlocks:         compression.MinLZBalanced,
+		OtherBlocks:         compression.MinLZBalanced,
+		MinReductionPercent: 0,
+	},
+
+	{
+		Name:                "Zstd1",
+		DataBlocks:          compression.ZstdLevel1,
+		ValueBlocks:         compression.ZstdLevel1,
+		OtherBlocks:         compression.ZstdLevel1,
+		MinReductionPercent: 0,
+	},
+
+	{
+		Name:                           "Auto1",
+		DataBlocks:                     compression.ZstdLevel1,
+		ValueBlocks:                    compression.ZstdLevel1,
+		OtherBlocks:                    compression.MinLZFastest,
+		AdaptiveReductionCutoffPercent: 30,
+		MinReductionPercent:            0,
+	},
+
+	{
+		Name:                "Zstd3",
+		DataBlocks:          compression.ZstdLevel3,
+		ValueBlocks:         compression.ZstdLevel3,
+		OtherBlocks:         compression.ZstdLevel3,
+		MinReductionPercent: 0,
+	},
+
+	{
+		Name:                           "Auto3",
+		DataBlocks:                     compression.ZstdLevel3,
+		ValueBlocks:                    compression.ZstdLevel3,
+		OtherBlocks:                    compression.MinLZFastest,
+		AdaptiveReductionCutoffPercent: 30,
+		MinReductionPercent:            0,
+	},
 	// Zstd levels 5+ are too slow (on the order of 15-20MB/s to compress) and
 	// don't usually offer a very large benefit in terms of size vs. level 3.
-	// compression.ZstdLevel5,
-	// compression.ZstdLevel7,
 }
 
-const numSettings = len(Settings)
+const numProfiles = len(Profiles)
 
 // Buckets holds the results of all experiments.
 type Buckets [blockkind.NumKinds][numBlockSizes][numCompressibility]Bucket
@@ -116,12 +167,12 @@ type Buckets [blockkind.NumKinds][numBlockSizes][numCompressibility]Bucket
 // compressibility.
 type Bucket struct {
 	UncompressedSize Welford
-	Experiments      [numSettings]PerSetting
+	Experiments      [numProfiles]PerProfile
 }
 
-// PerSetting holds statistics from experiments on blocks in a bucket with a
+// PerProfile holds statistics from experiments on blocks in a bucket with a
 // specific compression.Setting.
-type PerSetting struct {
+type PerProfile struct {
 	CompressionRatio WeightedWelford
 	// CPU times are in nanoseconds per uncompressed byte.
 	CompressionTime   WeightedWelford
@@ -133,8 +184,8 @@ func (b *Buckets) String(minSamples int) string {
 	tw := tabwriter.NewWriter(&buf, 2, 1, 2, ' ', 0)
 
 	fmt.Fprintf(tw, "Kind\tSize Range\tTest CR\tSamples\tSize\t")
-	for _, s := range Settings {
-		fmt.Fprintf(tw, "\t%s", s.String())
+	for _, p := range Profiles {
+		fmt.Fprintf(tw, "\t%s", p.Name)
 	}
 	fmt.Fprintf(tw, "\n")
 	for k := range blockkind.All() {
@@ -190,13 +241,13 @@ func stdDevStr(mean, stddev float64) string {
 func (b *Buckets) ToCSV(minSamples int) string {
 	var buf strings.Builder
 	fmt.Fprintf(&buf, "Kind,Size Range,Test CR,Samples,Size,Size±")
-	for _, s := range Settings {
-		fmt.Fprintf(&buf, ",%s CR", s.String())
-		fmt.Fprintf(&buf, ",%s CR±", s.String())
-		fmt.Fprintf(&buf, ",%s Comp ns/b", s.String())
-		fmt.Fprintf(&buf, ",%s Comp±", s.String())
-		fmt.Fprintf(&buf, ",%s Decomp ns/b", s.String())
-		fmt.Fprintf(&buf, ",%s Decomp±", s.String())
+	for _, p := range Profiles {
+		fmt.Fprintf(&buf, ",%s CR", p.Name)
+		fmt.Fprintf(&buf, ",%s CR±", p.Name)
+		fmt.Fprintf(&buf, ",%s Comp ns/b", p.Name)
+		fmt.Fprintf(&buf, ",%s Comp±", p.Name)
+		fmt.Fprintf(&buf, ",%s Decomp ns/b", p.Name)
+		fmt.Fprintf(&buf, ",%s Decomp±", p.Name)
 	}
 	fmt.Fprintf(&buf, "\n")
 	for k := range blockkind.All() {

sstable/compressionanalyzer/file_analyzer.go

@@ -20,7 +20,7 @@ import (
 //
 // TODO(radu): add support for blob files.
 type FileAnalyzer struct {
-	*BlockAnalyzer
+	blockAnalyzer *BlockAnalyzer
 
 	readLimiter *tokenbucket.TokenBucket
 	sstReadOpts sstable.ReaderOptions
@@ -35,22 +35,27 @@ func NewFileAnalyzer(
 		panic("sstReadOpts.CacheOpts.CacheHandle must be nil")
 	}
 	return &FileAnalyzer{
-		BlockAnalyzer: NewBlockAnalyzer(),
+		blockAnalyzer: NewBlockAnalyzer(),
 		readLimiter:   readLimiter,
 		sstReadOpts:   sstReadOpts,
 	}
 }
 
+func (fa *FileAnalyzer) Buckets() *Buckets {
+	return fa.blockAnalyzer.Buckets()
+}
+
 func (fa *FileAnalyzer) Close() {
-	if fa.BlockAnalyzer != nil {
-		fa.BlockAnalyzer.Close()
+	if fa.blockAnalyzer != nil {
+		fa.blockAnalyzer.Close()
 	}
 	*fa = FileAnalyzer{}
 }
 
 // SSTable analyzes the blocks in an sstable file and closes the readable (even
 // in error cases).
 func (fa *FileAnalyzer) SSTable(ctx context.Context, readable objstorage.Readable) error {
+	fa.blockAnalyzer.ResetCompressors()
 	r, err := sstable.NewReader(ctx, readable, fa.sstReadOpts)
 	if err != nil {
 		_ = readable.Close()
@@ -124,7 +129,7 @@ func (fa *FileAnalyzer) sstBlock(
 	if err != nil {
 		return err
 	}
-	fa.BlockAnalyzer.Block(kind, h.BlockData())
+	fa.blockAnalyzer.Block(kind, h.BlockData())
 	h.Release()
 	return nil
 }

sstable/compressionanalyzer/file_analyzer_test.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/datadriven"
-	"github.com/cockroachdb/pebble/internal/compression"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 )
@@ -43,7 +42,7 @@ func TestFileAnalyzer(t *testing.T) {
 						for l := range bucket.Experiments {
 							// Snappy always has the same output in all configurations and on
 							// all platforms.
-							if Settings[l].Algorithm != compression.SnappyAlgorithm {
+							if Profiles[l].Name != "Snappy" {
 								bucket.Experiments[l].CompressionRatio = WeightedWelford{}
 							}
 							bucket.Experiments[l].CompressionTime = WeightedWelford{}