pebble: replace Compression with CompressionProfile

Author: RaduBerinde

checkpoint_test.go

@@ -434,7 +434,7 @@ func TestCheckpointManyFiles(t *testing.T) {
 	// Disable compression to speed up the test.
 	opts.EnsureDefaults()
 	for i := range opts.Levels {
-		opts.Levels[i].Compression = func() Compression { return NoCompression }
+		opts.Levels[i].Compression = func() *CompressionProfile { return NoCompression }
 	}
 
 	d, err := Open("", opts)

internal/manifest/table_metadata.go

@@ -1168,8 +1168,9 @@ type TableStats struct {
 	RangeDeletionsBytesEstimate uint64
 	// Total size of value blocks and value index block.
 	ValueBlocksSize uint64
-	// CompressionType is the compression type of the table.
-	CompressionType block.Compression
+	// CompressionType is the compression profile used for the table (or nil if
+	// the profile name is not recognized).
+	CompressionType *block.CompressionProfile
 	// 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

metamorphic/options.go

@@ -797,14 +797,15 @@ func RandomOptions(
 		lopts.FilterPolicy = newTestingFilterPolicy(1 << rng.IntN(5))
 	}
 
-	// We use either no compression, snappy compression or zstd compression.
-	switch rng.IntN(3) {
+	switch rng.IntN(4) {
 	case 0:
-		lopts.Compression = func() block.Compression { return pebble.NoCompression }
+		lopts.Compression = func() *block.CompressionProfile { return pebble.NoCompression }
 	case 1:
-		lopts.Compression = func() block.Compression { return pebble.ZstdCompression }
+		lopts.Compression = func() *block.CompressionProfile { return pebble.ZstdCompression }
+	case 2:
+		lopts.Compression = func() *block.CompressionProfile { return pebble.SnappyCompression }
 	default:
-		lopts.Compression = func() block.Compression { return pebble.SnappyCompression }
+		lopts.Compression = func() *block.CompressionProfile { return pebble.MinLZCompression }
 	}
 	opts.Levels[0] = lopts
 

options.go

@@ -41,11 +41,10 @@ const (
 	defaultLevelMultiplier = 10
 )
 
-// Compression exports the base.Compression type.
-type Compression = block.Compression
+type CompressionProfile = block.CompressionProfile
 
 // Exported Compression constants.
-const (
+var (
 	DefaultCompression = block.DefaultCompression
 	NoCompression      = block.NoCompression
 	SnappyCompression  = block.SnappyCompression
@@ -417,9 +416,9 @@ type LevelOptions struct {
 
 	// Compression defines the per-block compression to use.
 	//
-	// The default value is DefaultCompression (which uses Snappy) for L0, or the
-	// function from the previous level for all other levels.
-	Compression func() Compression
+	// The default value is Snappy for L0, or the function from the previous level
+	// for all other levels.
+	Compression func() *CompressionProfile
 
 	// FilterPolicy defines a filter algorithm (such as a Bloom filter) that can
 	// reduce disk reads for Get calls.
@@ -467,7 +466,7 @@ func (o *LevelOptions) EnsureL0Defaults() {
 		o.BlockSizeThreshold = base.DefaultBlockSizeThreshold
 	}
 	if o.Compression == nil {
-		o.Compression = func() Compression { return DefaultCompression }
+		o.Compression = func() *CompressionProfile { return SnappyCompression }
 	}
 	if o.FilterPolicy == nil {
 		o.FilterPolicy = NoFilterPolicy
@@ -1614,7 +1613,7 @@ func (o *Options) String() string {
 		fmt.Fprintf(&buf, "  block_restart_interval=%d\n", l.BlockRestartInterval)
 		fmt.Fprintf(&buf, "  block_size=%d\n", l.BlockSize)
 		fmt.Fprintf(&buf, "  block_size_threshold=%d\n", l.BlockSizeThreshold)
-		fmt.Fprintf(&buf, "  compression=%s\n", resolveDefaultCompression(l.Compression()))
+		fmt.Fprintf(&buf, "  compression=%s\n", l.Compression().Name)
 		fmt.Fprintf(&buf, "  filter_policy=%s\n", l.FilterPolicy.Name())
 		fmt.Fprintf(&buf, "  filter_type=%s\n", l.FilterType)
 		fmt.Fprintf(&buf, "  index_block_size=%d\n", l.IndexBlockSize)
@@ -2084,20 +2083,11 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 			case "block_size_threshold":
 				l.BlockSizeThreshold, err = strconv.Atoi(value)
 			case "compression":
-				switch value {
-				case "Default":
-					l.Compression = func() Compression { return DefaultCompression }
-				case "NoCompression":
-					l.Compression = func() Compression { return NoCompression }
-				case "Snappy":
-					l.Compression = func() Compression { return SnappyCompression }
-				case "ZSTD":
-					l.Compression = func() Compression { return ZstdCompression }
-				case "MinLZ":
-					l.Compression = func() Compression { return MinLZCompression }
-				default:
+				profile := block.CompressionProfileByName(value)
+				if profile == nil {
 					return errors.Errorf("pebble: unknown compression: %q", errors.Safe(value))
 				}
+				l.Compression = func() *CompressionProfile { return profile }
 			case "filter_policy":
 				if hooks != nil && hooks.NewFilterPolicy != nil {
 					l.FilterPolicy, err = hooks.NewFilterPolicy(value)
@@ -2335,7 +2325,7 @@ func (o *Options) MakeWriterOptions(level int, format sstable.TableFormat) sstab
 	writerOpts.BlockRestartInterval = levelOpts.BlockRestartInterval
 	writerOpts.BlockSize = levelOpts.BlockSize
 	writerOpts.BlockSizeThreshold = levelOpts.BlockSizeThreshold
-	writerOpts.Compression = resolveDefaultCompression(levelOpts.Compression())
+	writerOpts.Compression = levelOpts.Compression()
 	writerOpts.FilterPolicy = levelOpts.FilterPolicy
 	writerOpts.FilterType = levelOpts.FilterType
 	writerOpts.IndexBlockSize = levelOpts.IndexBlockSize
@@ -2357,7 +2347,7 @@ func (o *Options) MakeWriterOptions(level int, format sstable.TableFormat) sstab
 func (o *Options) MakeBlobWriterOptions(level int) blob.FileWriterOptions {
 	lo := o.Levels[level]
 	return blob.FileWriterOptions{
-		Compression:  resolveDefaultCompression(lo.Compression()),
+		Compression:  lo.Compression(),
 		ChecksumType: block.ChecksumTypeCRC32c,
 		FlushGovernor: block.MakeFlushGovernor(
 			lo.BlockSize,
@@ -2368,13 +2358,6 @@ func (o *Options) MakeBlobWriterOptions(level int) blob.FileWriterOptions {
 	}
 }
 
-func resolveDefaultCompression(c Compression) Compression {
-	if c <= DefaultCompression || c >= block.NCompression {
-		c = SnappyCompression
-	}
-	return c
-}
-
 func (o *Options) MakeObjStorageProviderSettings(dirname string) objstorageprovider.Settings {
 	s := objstorageprovider.Settings{
 		Logger:        o.Logger,

sstable/blob/blob.go

@@ -49,15 +49,15 @@ const (
 
 // FileWriterOptions are used to configure the FileWriter.
 type FileWriterOptions struct {
-	Compression   block.Compression
+	Compression   *block.CompressionProfile
 	ChecksumType  block.ChecksumType
 	FlushGovernor block.FlushGovernor
 	// Only CPUMeasurer.MeasureCPUBlobFileSecondary is used.
 	CpuMeasurer base.CPUMeasurer
 }
 
 func (o *FileWriterOptions) ensureDefaults() {
-	if o.Compression <= block.DefaultCompression || o.Compression >= block.NCompression {
+	if o.Compression == nil {
 		o.Compression = block.SnappyCompression
 	}
 	if o.ChecksumType == block.ChecksumTypeNone {
@@ -131,7 +131,7 @@ func NewFileWriter(fn base.DiskFileNum, w objstorage.Writable, opts FileWriterOp
 	fw.flushGov = opts.FlushGovernor
 	fw.indexEncoder.Init()
 	fw.checksummer = block.Checksummer{Type: opts.ChecksumType}
-	fw.compressor = block.MakeCompressor(opts.Compression.ToProfile())
+	fw.compressor = block.MakeCompressor(opts.Compression)
 	fw.cpuMeasurer = opts.CpuMeasurer
 	fw.writeQueue.ch = make(chan compressedBlock)
 	fw.writeQueue.wg.Add(1)

sstable/blob/blob_test.go

@@ -85,12 +85,15 @@ func scanFileWriterOptions(t *testing.T, td *datadriven.TestData) FileWriterOpti
 	var (
 		targetBlockSize    int = 128
 		blockSizeThreshold int = 90
-		compression            = block.NoCompression
 	)
 	td.MaybeScanArgs(t, "target-block-size", &targetBlockSize)
 	td.MaybeScanArgs(t, "block-size-threshold", &blockSizeThreshold)
+	var compression *block.CompressionProfile
 	if cmdArg, ok := td.Arg("compression"); ok {
-		compression = block.CompressionFromString(cmdArg.SingleVal(t))
+		compression = block.CompressionProfileByName(cmdArg.SingleVal(t))
+		if compression == nil {
+			t.Fatalf("unknown compression %q", cmdArg.SingleVal(t))
+		}
 	}
 	return FileWriterOptions{
 		Compression:   compression,

sstable/block/compression.go

@@ -6,6 +6,7 @@ package block
 
 import (
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/cockroachdb/errors"
@@ -16,71 +17,6 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 )
 
-// Compression is the per-block compression algorithm to use.
-type Compression int
-
-// The available compression types.
-const (
-	DefaultCompression Compression = iota
-	NoCompression
-	SnappyCompression
-	ZstdCompression
-	// MinLZCompression is only supported with table formats v6+. Older formats
-	// fall back to snappy.
-	MinLZCompression
-	NCompression
-)
-
-var profiles = [...]CompressionProfile{
-	DefaultCompression: SimpleCompressionProfile(DefaultCompression.String(), compression.Snappy),
-	NoCompression:      SimpleCompressionProfile(NoCompression.String(), compression.None),
-	SnappyCompression:  SimpleCompressionProfile(SnappyCompression.String(), compression.Snappy),
-	ZstdCompression:    SimpleCompressionProfile(ZstdCompression.String(), compression.ZstdLevel3),
-	MinLZCompression:   SimpleCompressionProfile(MinLZCompression.String(), compression.MinLZFastest),
-}
-
-func (c Compression) ToProfile() *CompressionProfile {
-	return &profiles[c]
-}
-
-// String implements fmt.Stringer, returning a human-readable name for the
-// compression algorithm.
-func (c Compression) String() string {
-	switch c {
-	case DefaultCompression:
-		return "Default"
-	case NoCompression:
-		return "NoCompression"
-	case SnappyCompression:
-		return "Snappy"
-	case ZstdCompression:
-		return "ZSTD"
-	case MinLZCompression:
-		return "MinLZ"
-	default:
-		return "Unknown"
-	}
-}
-
-// CompressionFromString returns an sstable.Compression from its
-// string representation. Inverse of c.String() above.
-func CompressionFromString(s string) Compression {
-	switch s {
-	case "Default":
-		return DefaultCompression
-	case "NoCompression":
-		return NoCompression
-	case "Snappy":
-		return SnappyCompression
-	case "ZSTD":
-		return ZstdCompression
-	case "MinLZ":
-		return MinLZCompression
-	default:
-		return DefaultCompression
-	}
-}
-
 // CompressionProfile contains the parameters for compressing blocks in an
 // sstable or blob file.
 //
@@ -107,17 +43,49 @@ type CompressionProfile struct {
 	// TODO(radu): knobs for adaptive compression go here.
 }
 
-// SimpleCompressionProfile returns a CompressionProfile that uses the same
+var (
+	NoCompression     = simpleCompressionProfile("NoCompression", compression.None)
+	SnappyCompression = simpleCompressionProfile("Snappy", compression.Snappy)
+	ZstdCompression   = simpleCompressionProfile("ZSTD", compression.ZstdLevel3)
+	MinLZCompression  = simpleCompressionProfile("MinLZ", compression.MinLZFastest)
+
+	DefaultCompression = SnappyCompression
+)
+
+// simpleCompressionProfile returns a CompressionProfile that uses the same
 // compression setting for all blocks and which uses the uncompressed block if
 // compression reduces it by less than 12%. This is similar to older Pebble
 // versions which used Compression.
-func SimpleCompressionProfile(name string, setting compression.Setting) CompressionProfile {
-	return CompressionProfile{
+//
+// It should only be used during global initialization.
+func simpleCompressionProfile(name string, setting compression.Setting) *CompressionProfile {
+	p := &CompressionProfile{
 		Name:                name,
 		DataBlocks:          setting,
 		OtherBlocks:         setting,
 		MinReductionPercent: 12,
 	}
+	registerCompressionProfile(p)
+	return p
+}
+
+// CompressionProfileByName returns the built-in compression profile with the
+// given name, or nil if there is no such profile. It is case-insensitive.
+//
+// The caller must gracefully handle the nil return case as an unknown
+// (user-defined or deprecated) profile.
+func CompressionProfileByName(name string) *CompressionProfile {
+	return compressionProfileMap[strings.ToLower(name)]
+}
+
+var compressionProfileMap = make(map[string]*CompressionProfile)
+
+func registerCompressionProfile(p *CompressionProfile) {
+	key := strings.ToLower(p.Name)
+	if _, ok := compressionProfileMap[key]; ok {
+		panic(errors.AssertionFailedf("duplicate compression profile: %s", p.Name))
+	}
+	compressionProfileMap[key] = p
 }
 
 // CompressionIndicator is the byte stored physically within the block.Trailer

sstable/block/compression_test.go

@@ -19,7 +19,7 @@ func TestBufferRandomized(t *testing.T) {
 	t.Logf("seed %d", seed)
 	rng := rand.New(rand.NewPCG(0, seed))
 
-	compressor := MakeCompressor(SnappyCompression.ToProfile())
+	compressor := MakeCompressor(SnappyCompression)
 	defer compressor.Close()
 	var checksummer Checksummer
 	checksummer.Init(ChecksumTypeCRC32c)

sstable/block/compressor.go

@@ -73,7 +73,7 @@ func (c *Compressor) Compress(dst, src []byte, kind Kind) (CompressionIndicator,
 // any state and can be used in parallel.
 var NoopCompressor = &noopCompressor
 
-var noopCompressor = MakeCompressor(NoCompression.ToProfile())
+var noopCompressor = MakeCompressor(NoCompression)
 
 type Decompressor = compression.Decompressor
 

sstable/colblk_writer.go

@@ -186,7 +186,7 @@ func newColumnarWriter(
 	w.props.PropertyCollectorNames = buf.String()
 
 	w.props.ComparerName = o.Comparer.Name
-	w.props.CompressionName = o.Compression.String()
+	w.props.CompressionName = o.Compression.Name
 	w.props.KeySchemaName = o.KeySchema.Name
 	w.props.MergerName = o.MergerName
 
@@ -195,7 +195,7 @@ func newColumnarWriter(
 	w.cpuMeasurer = cpuMeasurer
 	go w.drainWriteQueue()
 
-	w.compressor = block.MakeCompressor(w.opts.Compression.ToProfile())
+	w.compressor = block.MakeCompressor(w.opts.Compression)
 	return w
 }