block: introduce CompressionProfile

CompressionProfile is intended as a replacement for Compression; it
allows using different compressors for data vs other blocks, and it
allows tuning of the minimum reduction percentage (previously
hardcoded to 12.5%).

Author: RaduBerinde

sstable/blob/blob.go

@@ -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)
+	fw.compressor = block.MakeCompressor(opts.Compression.ToProfile())
 	fw.cpuMeasurer = opts.CpuMeasurer
 	fw.writeQueue.ch = make(chan compressedBlock)
 	fw.writeQueue.wg.Add(1)
@@ -213,7 +213,7 @@ func (w *FileWriter) flush() {
 	if w.valuesEncoder.Count() == 0 {
 		panic(errors.AssertionFailedf("no values to flush"))
 	}
-	pb, bh := block.CompressAndChecksumToTempBuffer(w.valuesEncoder.Finish(), &w.compressor, &w.checksummer)
+	pb, bh := block.CompressAndChecksumToTempBuffer(w.valuesEncoder.Finish(), blockkind.BlobValue, &w.compressor, &w.checksummer)
 	compressedLen := uint64(pb.LengthWithoutTrailer())
 	w.stats.BlockCount++
 	off := w.stats.FileLen
@@ -283,7 +283,7 @@ func (w *FileWriter) Close() (FileWriterStats, error) {
 	{
 		indexBlock := w.indexEncoder.Finish()
 		var compressedBuf []byte
-		pb := block.CompressAndChecksum(&compressedBuf, indexBlock, block.NoopCompressor, &w.checksummer)
+		pb := block.CompressAndChecksum(&compressedBuf, indexBlock, blockkind.Metadata, block.NoopCompressor, &w.checksummer)
 		if _, w.err = pb.WriteTo(w.w); w.err != nil {
 			err = w.err
 			if w.w != nil {

sstable/block/compression.go

@@ -31,16 +31,16 @@ const (
 	NCompression
 )
 
-var setting = [...]compression.Setting{
-	DefaultCompression: compression.Snappy,
-	NoCompression:      compression.None,
-	SnappyCompression:  compression.Snappy,
-	ZstdCompression:    compression.ZstdLevel3,
-	MinLZCompression:   compression.MinLZFastest,
+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) setting() compression.Setting {
-	return setting[c]
+func (c Compression) ToProfile() *CompressionProfile {
+	return &profiles[c]
 }
 
 // String implements fmt.Stringer, returning a human-readable name for the
@@ -81,6 +81,45 @@ func CompressionFromString(s string) Compression {
 	}
 }
 
+// CompressionProfile contains the parameters for compressing blocks in an
+// sstable or blob file.
+//
+// CompressionProfile is a more advanced successor to Compression.
+type CompressionProfile struct {
+	Name string
+
+	// DataBlocks applies to sstable data and value blocks, as well as blob file
+	// value blocks. OtherBlocks applies to all other blocks (such as index,
+	// filter, metadata blocks).
+	//
+	// Some blocks (like rangedel) never use compression; this is at the
+	// discretion of the sstable or blob file writer.
+	//
+	// Note that MinLZ is only supported with table formats v6+. Older formats
+	// fall back to Snappy.
+	DataBlocks  compression.Setting
+	OtherBlocks compression.Setting
+
+	// Blocks that are reduced by less than this percentage are stored
+	// uncompressed.
+	MinReductionPercent uint8
+
+	// TODO(radu): knobs for adaptive compression go here.
+}
+
+// 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{
+		Name:                name,
+		DataBlocks:          setting,
+		OtherBlocks:         setting,
+		MinReductionPercent: 12,
+	}
+}
+
 // CompressionIndicator is the byte stored physically within the block.Trailer
 // to indicate the compression type.
 //
@@ -260,17 +299,10 @@ func (b *PhysicalBlock) WriteTo(w objstorage.Writable) (n int, err error) {
 // the compressed payload is discarded and the original, uncompressed block data
 // is used to avoid unnecessary decompression overhead at read time.
 func CompressAndChecksum(
-	dst *[]byte, blockData []byte, compressor *Compressor, checksummer *Checksummer,
+	dst *[]byte, blockData []byte, blockKind Kind, compressor *Compressor, checksummer *Checksummer,
 ) PhysicalBlock {
 	buf := (*dst)[:0]
-	// Compress the buffer, discarding the result if the improvement isn't at
-	// least 12.5%.
-	ci, buf := compressor.Compress(buf, blockData)
-	if len(buf) >= len(blockData)-len(blockData)/8 && ci != NoCompressionIndicator {
-		ci = NoCompressionIndicator
-		buf = append(buf[:0], blockData...)
-	}
-
+	ci, buf := compressor.Compress(buf, blockData, blockKind)
 	*dst = buf
 
 	// Calculate the checksum.
@@ -284,11 +316,11 @@ func CompressAndChecksum(
 // into a TempBuffer. The caller should Release() the TempBuffer once it is no
 // longer necessary.
 func CompressAndChecksumToTempBuffer(
-	blockData []byte, compressor *Compressor, checksummer *Checksummer,
+	blockData []byte, blockKind Kind, compressor *Compressor, checksummer *Checksummer,
 ) (PhysicalBlock, *TempBuffer) {
 	// Grab a buffer to use as the destination for compression.
 	compressedBuf := NewTempBuffer()
-	pb := CompressAndChecksum(&compressedBuf.b, blockData, compressor, checksummer)
+	pb := CompressAndChecksum(&compressedBuf.b, blockData, blockKind, compressor, checksummer)
 	return pb, compressedBuf
 }
 

sstable/block/compression_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/pebble/sstable/block/blockkind"
 	"github.com/stretchr/testify/require"
 )
 
@@ -18,7 +19,7 @@ func TestBufferRandomized(t *testing.T) {
 	t.Logf("seed %d", seed)
 	rng := rand.New(rand.NewPCG(0, seed))
 
-	compressor := MakeCompressor(SnappyCompression)
+	compressor := MakeCompressor(SnappyCompression.ToProfile())
 	defer compressor.Close()
 	var checksummer Checksummer
 	checksummer.Init(ChecksumTypeCRC32c)
@@ -52,7 +53,7 @@ func TestBufferRandomized(t *testing.T) {
 				s := b.Data()
 				require.Equal(t, vbuf, s[len(s)-len(vbuf):])
 			}
-			_, bh := CompressAndChecksumToTempBuffer(b.Data(), &compressor, &checksummer)
+			_, bh := CompressAndChecksumToTempBuffer(b.Data(), blockkind.SSTableData, &compressor, &checksummer)
 			b.Reset()
 			bh.Release()
 		})

sstable/block/compressor.go

@@ -1,49 +1,79 @@
 package block
 
-import "github.com/cockroachdb/pebble/internal/compression"
+import (
+	"github.com/cockroachdb/pebble/internal/compression"
+	"github.com/cockroachdb/pebble/sstable/block/blockkind"
+)
 
 // Compressor is used to compress blocks. Typical usage:
 //
-//	c := GetCompressor(compression)
+//	c := MakeCompressor(profile)
 //	.. = c.Compress(..)
 //	.. = c.Compress(..)
 //	c.Close()
 type Compressor struct {
-	algorithm  compression.Algorithm
-	compressor compression.Compressor
+	profile              CompressionProfile
+	dataBlocksCompressor compression.Compressor
+	// otherBlocksCompressor is used for blocks that are not data blocks, such as
+	// index blocks or metadata blocks. It can be the same object as
+	// dataBlocksCompressor.
+	otherBlocksCompressor compression.Compressor
 }
 
-// MakeCompressor returns a Compressor that applies the given compression. Close
-// must be called when it is no longer needed.
-func MakeCompressor(c Compression) Compressor {
-	s := c.setting()
-	return Compressor{
-		algorithm:  s.Algorithm,
-		compressor: compression.GetCompressor(s),
+// MakeCompressor returns a Compressor that applies the given compression
+// profile. Close must be called when the compressor is no longer needed.
+func MakeCompressor(profile *CompressionProfile) Compressor {
+	c := Compressor{
+		profile: *profile,
 	}
-}
-
-// Compress a block, appending the compressed data to dst[:0].
-//
-// In addition to the buffer, returns the algorithm that was used.
-func (c *Compressor) Compress(dst, src []byte) (CompressionIndicator, []byte) {
-	ci := compressionIndicatorFromAlgorithm(c.algorithm)
-	return ci, c.compressor.Compress(dst, src)
+	c.dataBlocksCompressor = compression.GetCompressor(profile.DataBlocks)
+	c.otherBlocksCompressor = c.dataBlocksCompressor
+	if profile.OtherBlocks != profile.DataBlocks {
+		c.otherBlocksCompressor = compression.GetCompressor(profile.OtherBlocks)
+	}
+	return c
 }
 
 // Close must be called when the Compressor is no longer needed.
 // After Close is called, the Compressor must not be used again.
 func (c *Compressor) Close() {
-	c.compressor.Close()
+	if c.otherBlocksCompressor != c.dataBlocksCompressor {
+		c.otherBlocksCompressor.Close()
+	}
+	c.dataBlocksCompressor.Close()
 	*c = Compressor{}
 }
 
+// Compress a block, appending the compressed data to dst[:0].
+//
+// In addition to the buffer, returns the algorithm that was used.
+func (c *Compressor) Compress(dst, src []byte, kind Kind) (CompressionIndicator, []byte) {
+	setting := c.profile.DataBlocks
+	compressor := c.dataBlocksCompressor
+	if kind != blockkind.SSTableData && kind != blockkind.SSTableValue && kind != blockkind.BlobValue {
+		setting = c.profile.OtherBlocks
+		compressor = c.otherBlocksCompressor
+	}
+	out := compressor.Compress(dst, src)
+
+	// Return the original data uncompressed if the reduction is less than the
+	// minimum, i.e.:
+	//
+	//   after * 100
+	//   -----------  >  100 - MinReductionPercent
+	//      before
+	if setting.Algorithm != compression.NoCompression &&
+		int64(len(out))*100 > int64(len(src))*int64(100-c.profile.MinReductionPercent) {
+		return NoCompressionIndicator, append(out[:0], src...)
+	}
+	return compressionIndicatorFromAlgorithm(setting.Algorithm), out
+}
+
 // NoopCompressor is a Compressor that does not compress data. It does not have
 // any state and can be used in parallel.
-var NoopCompressor = &Compressor{
-	algorithm:  compression.NoCompression,
-	compressor: compression.GetCompressor(compression.None),
-}
+var NoopCompressor = &noopCompressor
+
+var noopCompressor = MakeCompressor(NoCompression.ToProfile())
 
 type Decompressor = compression.Decompressor
 

sstable/colblk_writer.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/sstable/block/blockkind"
 	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/cockroachdb/pebble/sstable/rowblk"
 	"github.com/cockroachdb/pebble/sstable/valblk"
@@ -148,7 +149,7 @@ func newColumnarWriter(
 	if !o.DisableValueBlocks {
 		w.valueBlock = valblk.NewWriter(
 			block.MakeFlushGovernor(o.BlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses),
-			w.opts.Compression, w.opts.Checksum, func(compressedSize int) {})
+			&w.compressor, w.opts.Checksum, func(compressedSize int) {})
 	}
 	if o.FilterPolicy != base.NoFilterPolicy {
 		switch o.FilterType {
@@ -194,7 +195,7 @@ func newColumnarWriter(
 	w.cpuMeasurer = cpuMeasurer
 	go w.drainWriteQueue()
 
-	w.compressor = block.MakeCompressor(w.opts.Compression)
+	w.compressor = block.MakeCompressor(w.opts.Compression.ToProfile())
 	return w
 }
 
@@ -723,6 +724,7 @@ func (w *RawColumnWriter) enqueueDataBlock(
 	cb.physical = block.CompressAndChecksum(
 		&cb.blockBuf.dataBuf,
 		serializedBlock,
+		blockkind.SSTableData,
 		&w.compressor,
 		&cb.blockBuf.checksummer,
 	)
@@ -1250,6 +1252,7 @@ func (w *RawColumnWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProper
 	cb.physical = block.CompressAndChecksum(
 		&cb.blockBuf.dataBuf,
 		b,
+		blockkind.SSTableData,
 		&w.compressor,
 		&cb.blockBuf.checksummer,
 	)

sstable/layout.go

@@ -796,7 +796,7 @@ func makeLayoutWriter(w objstorage.Writable, opts WriterOptions) layoutWriter {
 		writable:     w,
 		cacheOpts:    opts.internal.CacheOpts,
 		tableFormat:  opts.TableFormat,
-		compressor:   block.MakeCompressor(opts.Compression),
+		compressor:   block.MakeCompressor(opts.Compression.ToProfile()),
 		checksumType: opts.Checksum,
 		buf: blockBuf{
 			checksummer: block.Checksummer{Type: opts.Checksum},
@@ -822,7 +822,7 @@ func (w *layoutWriter) Abort() {
 // WriteDataBlock constructs a trailer for the provided data block and writes
 // the block and trailer to the writer. It returns the block's handle.
 func (w *layoutWriter) WriteDataBlock(b []byte, buf *blockBuf) (block.Handle, error) {
-	return w.writeBlock(b, &w.compressor, buf)
+	return w.writeBlock(b, blockkind.SSTableData, &w.compressor, buf)
 }
 
 // WritePrecompressedDataBlock writes a pre-compressed data block and its
@@ -837,7 +837,7 @@ func (w *layoutWriter) WritePrecompressedDataBlock(blk block.PhysicalBlock) (blo
 // the last-written index block's handle and adds it to the file's meta index
 // when the writer is finished.
 func (w *layoutWriter) WriteIndexBlock(b []byte) (block.Handle, error) {
-	h, err := w.writeBlock(b, &w.compressor, &w.buf)
+	h, err := w.writeBlock(b, blockkind.SSTableIndex, &w.compressor, &w.buf)
 	if err == nil {
 		w.lastIndexBlockHandle = h
 	}
@@ -852,7 +852,7 @@ func (w *layoutWriter) WriteFilterBlock(f filterWriter) (bh block.Handle, err er
 	if err != nil {
 		return block.Handle{}, err
 	}
-	return w.writeNamedBlock(b, block.NoopCompressor, f.metaName())
+	return w.writeNamedBlock(b, blockkind.Filter, block.NoopCompressor, f.metaName())
 }
 
 // WritePropertiesBlock constructs a trailer for the provided properties block
@@ -867,28 +867,28 @@ func (w *layoutWriter) WritePropertiesBlock(b []byte) (block.Handle, error) {
 	if w.tableFormat < TableFormatPebblev7 {
 		compressor = block.NoopCompressor
 	}
-	return w.writeNamedBlock(b, compressor, metaPropertiesName)
+	return w.writeNamedBlock(b, blockkind.Metadata, compressor, metaPropertiesName)
 }
 
 // WriteRangeKeyBlock constructs a trailer for the provided range key block and
 // writes the block and trailer to the writer. It automatically adds the range
 // key block to the file's meta index when the writer is finished.
 func (w *layoutWriter) WriteRangeKeyBlock(b []byte) (block.Handle, error) {
-	return w.writeNamedBlock(b, block.NoopCompressor, metaRangeKeyName)
+	return w.writeNamedBlock(b, blockkind.RangeKey, block.NoopCompressor, metaRangeKeyName)
 }
 
 // WriteRangeDeletionBlock constructs a trailer for the provided range deletion
 // block and writes the block and trailer to the writer. It automatically adds
 // the range deletion block to the file's meta index when the writer is
 // finished.
 func (w *layoutWriter) WriteRangeDeletionBlock(b []byte) (block.Handle, error) {
-	return w.writeNamedBlock(b, block.NoopCompressor, metaRangeDelV2Name)
+	return w.writeNamedBlock(b, blockkind.RangeDel, block.NoopCompressor, metaRangeDelV2Name)
 }
 
 func (w *layoutWriter) writeNamedBlock(
-	b []byte, compressor *block.Compressor, name string,
+	b []byte, kind block.Kind, compressor *block.Compressor, name string,
 ) (bh block.Handle, err error) {
-	bh, err = w.writeBlock(b, compressor, &w.buf)
+	bh, err = w.writeBlock(b, kind, compressor, &w.buf)
 	if err == nil {
 		w.recordToMetaindex(name, bh)
 	}
@@ -917,9 +917,9 @@ func (w *layoutWriter) WriteValueIndexBlock(
 
 // writeBlock checksums, compresses, and writes out a block.
 func (w *layoutWriter) writeBlock(
-	b []byte, compressor *block.Compressor, buf *blockBuf,
+	b []byte, kind block.Kind, compressor *block.Compressor, buf *blockBuf,
 ) (block.Handle, error) {
-	pb := block.CompressAndChecksum(&buf.dataBuf, b, compressor, &buf.checksummer)
+	pb := block.CompressAndChecksum(&buf.dataBuf, b, kind, compressor, &buf.checksummer)
 	h, err := w.writePrecompressedBlock(pb)
 	return h, err
 }
@@ -1003,7 +1003,7 @@ func (w *layoutWriter) Finish() (size uint64, err error) {
 		}
 		b = bw.Finish()
 	}
-	metaIndexHandle, err := w.writeBlock(b, block.NoopCompressor, &w.buf)
+	metaIndexHandle, err := w.writeBlock(b, blockkind.Metadata, block.NoopCompressor, &w.buf)
 	if err != nil {
 		return 0, err
 	}