compression: return Setting from Compress

Author: RaduBerinde

internal/compression/compression.go

@@ -10,7 +10,7 @@ import (
 	"github.com/minio/minlz"
 )
 
-// Algorithm identifies a compression algorithm. Some compression algorithm
+// Algorithm identifies a compression algorithm. Some compression algorithms
 // support multiple compression levels.
 //
 // Decompressing data requires only an Algorithm.
@@ -21,7 +21,8 @@ const (
 	SnappyAlgorithm
 	Zstd
 	MinLZ
-	numAlgorithms
+
+	NumAlgorithms
 )
 
 // String implements fmt.Stringer, returning a human-readable name for the
@@ -72,11 +73,9 @@ var (
 // Compressor is an interface for compressing data. An instance is associated
 // with a specific Setting.
 type Compressor interface {
-	// Algorithm returns the algorithm used by this Compressor.
-	Algorithm() Algorithm
-
 	// Compress a block, appending the compressed data to dst[:0].
-	Compress(dst, src []byte) []byte
+	// Returns setting used.
+	Compress(dst, src []byte) ([]byte, Setting)
 
 	// Close must be called when the Compressor is no longer needed.
 	// After Close is called, the Compressor must not be used again.

internal/compression/compression_test.go

@@ -32,8 +32,8 @@ func TestCompressionRoundtrip(t *testing.T) {
 			compressedBuf := make([]byte, 1+rng.IntN(1<<10 /* 1 KiB */))
 			compressor := GetCompressor(s)
 			defer compressor.Close()
-			compressed := compressor.Compress(compressedBuf, payload)
-			got, err := decompress(s.Algorithm, compressed)
+			compressed, st := compressor.Compress(compressedBuf, payload)
+			got, err := decompress(st.Algorithm, compressed)
 			require.NoError(t, err)
 			require.Equal(t, payload, got)
 		})

internal/compression/minlz.go

@@ -18,9 +18,9 @@ var _ Compressor = (*minlzCompressor)(nil)
 
 func (c *minlzCompressor) Algorithm() Algorithm { return MinLZ }
 
-func (c *minlzCompressor) Compress(dst, src []byte) []byte {
+func (c *minlzCompressor) Compress(dst, src []byte) ([]byte, Setting) {
 	// MinLZ cannot encode blocks greater than 8MB. Fall back to Snappy in those
-	// cases. Note that MinLZ can decode the Snappy compressed block.
+	// cases.
 	if len(src) > minlz.MaxBlockSize {
 		return (snappyCompressor{}).Compress(dst, src)
 	}
@@ -29,7 +29,7 @@ func (c *minlzCompressor) Compress(dst, src []byte) []byte {
 	if err != nil {
 		panic(errors.Wrap(err, "minlz compression"))
 	}
-	return compressed
+	return compressed, Setting{Algorithm: MinLZ, Level: uint8(c.level)}
 }
 
 func (c *minlzCompressor) Close() {}

internal/compression/minlz_test.go

@@ -20,12 +20,20 @@ func TestMinLZLargeBlock(t *testing.T) {
 		}
 		c := GetCompressor(MinLZFastest)
 		defer c.Close()
-		compressed := c.Compress(nil, b)
-		d := GetDecompressor(MinLZ)
+		compressed, st := c.Compress(nil, b)
+
+		d := GetDecompressor(st.Algorithm)
 		decompressed := make([]byte, len(b))
-		defer d.Close()
+		require.NoError(t, d.DecompressInto(decompressed, compressed))
+		require.Equal(t, b, decompressed)
+		d.Close()
 
+		// Verify that a MinLZ decompressor always works (even if Compress returned
+		// Snappy).
+		d = GetDecompressor(MinLZ)
+		clear(decompressed)
 		require.NoError(t, d.DecompressInto(decompressed, compressed))
 		require.Equal(t, b, decompressed)
+		d.Close()
 	}
 }

internal/compression/noop.go

@@ -8,9 +8,8 @@ type noopCompressor struct{}
 
 var _ Compressor = noopCompressor{}
 
-func (noopCompressor) Algorithm() Algorithm { return NoCompression }
-func (noopCompressor) Compress(dst, src []byte) []byte {
-	return append(dst[:0], src...)
+func (noopCompressor) Compress(dst, src []byte) ([]byte, Setting) {
+	return append(dst[:0], src...), None
 }
 func (noopCompressor) Close() {}
 

internal/compression/snappy.go

@@ -16,9 +16,9 @@ var _ Compressor = snappyCompressor{}
 
 func (snappyCompressor) Algorithm() Algorithm { return SnappyAlgorithm }
 
-func (snappyCompressor) Compress(dst, src []byte) []byte {
+func (snappyCompressor) Compress(dst, src []byte) ([]byte, Setting) {
 	dst = dst[:cap(dst):cap(dst)]
-	return snappy.Encode(dst, src)
+	return snappy.Encode(dst, src), Snappy
 }
 
 func (snappyCompressor) Close() {}

internal/compression/zstd_cgo.go

@@ -39,9 +39,7 @@ var zstdCompressorPool = sync.Pool{
 // relies on CGo.
 const UseStandardZstdLib = true
 
-func (z *zstdCompressor) Algorithm() Algorithm { return Zstd }
-
-func (z *zstdCompressor) Compress(compressedBuf []byte, b []byte) []byte {
+func (z *zstdCompressor) Compress(compressedBuf []byte, b []byte) ([]byte, Setting) {
 	if len(compressedBuf) < binary.MaxVarintLen64 {
 		compressedBuf = append(compressedBuf, make([]byte, binary.MaxVarintLen64-len(compressedBuf))...)
 	}
@@ -63,7 +61,7 @@ func (z *zstdCompressor) Compress(compressedBuf []byte, b []byte) []byte {
 		panic("Allocated a new buffer despite checking CompressBound.")
 	}
 
-	return compressedBuf[:varIntLen+len(result)]
+	return compressedBuf[:varIntLen+len(result)], Setting{Algorithm: Zstd, Level: uint8(z.level)}
 }
 
 func (z *zstdCompressor) Close() {

internal/compression/zstd_nocgo.go

@@ -34,14 +34,13 @@ func getZstdCompressor(level int) *zstdCompressor {
 // relies on CGo.
 const UseStandardZstdLib = false
 
-func (z *zstdCompressor) Algorithm() Algorithm { return Zstd }
-
 func (z *zstdCompressor) Compress(compressedBuf, b []byte) []byte {
 	if len(compressedBuf) < binary.MaxVarintLen64 {
 		compressedBuf = append(compressedBuf, make([]byte, binary.MaxVarintLen64-len(compressedBuf))...)
 	}
 	varIntLen := binary.PutUvarint(compressedBuf, uint64(len(b)))
-	return (*zstd.Encoder)(z).EncodeAll(b, compressedBuf[:varIntLen])
+	res := (*zstd.Encoder)(z).EncodeAll(b, compressedBuf[:varIntLen])
+	return res, Setting{Algorithm: Zstd, Level: uint8(z.level)}
 }
 
 func (z *zstdCompressor) Close() {

sstable/block/compressor.go

@@ -70,20 +70,19 @@ func (c *Compressor) Compress(dst, src []byte, kind Kind) (CompressionIndicator,
 		compressor = c.otherBlocksCompressor
 	}
 
-	out := compressor.Compress(dst, src)
+	out, setting := compressor.Compress(dst, src)
 
 	// Return the original data uncompressed if the reduction is less than the
 	// minimum, i.e.:
 	//
 	//   after * 100
 	//   -----------  >  100 - MinReductionPercent
 	//      before
-	algorithm := compressor.Algorithm()
-	if algorithm != compression.NoCompression &&
+	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(algorithm), out
+	return compressionIndicatorFromAlgorithm(setting.Algorithm), out
 }
 
 // NoopCompressor is a Compressor that does not compress data. It does not have

sstable/compressionanalyzer/block_analyzer.go

@@ -18,7 +18,7 @@ import (
 type BlockAnalyzer struct {
 	b             Buckets
 	compressors   [numSettings]compression.Compressor
-	decompressors [numSettings]compression.Decompressor
+	decompressors [compression.NumAlgorithms]compression.Decompressor
 	minLZFastest  compression.Compressor
 	buf1          []byte
 	buf2          []byte
@@ -28,7 +28,9 @@ func NewBlockAnalyzer() *BlockAnalyzer {
 	a := &BlockAnalyzer{}
 	for i, s := range Settings {
 		a.compressors[i] = compression.GetCompressor(s)
-		a.decompressors[i] = compression.GetDecompressor(s.Algorithm)
+	}
+	for i := range a.decompressors {
+		a.decompressors[i] = compression.GetDecompressor(compression.Algorithm(i))
 	}
 	a.minLZFastest = compression.GetCompressor(compression.MinLZFastest)
 	a.buf1 = make([]byte, 256*1024)
@@ -51,11 +53,12 @@ func (a *BlockAnalyzer) Close() {
 // of various compression algorithms on it.
 func (a *BlockAnalyzer) Block(kind block.Kind, block []byte) {
 	size := MakeBlockSize(len(block))
-	compressibility := MakeCompressibility(len(block), len(a.minLZFastest.Compress(a.buf1[:0], block)))
+	compressed, _ := a.minLZFastest.Compress(a.buf1[:0], block)
+	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[i])
+		a.runExperiment(&bucket.Experiments[i], block, a.compressors[i], a.decompressors)
 	}
 }
 
@@ -67,7 +70,7 @@ func (a *BlockAnalyzer) runExperiment(
 	pa *PerSetting,
 	block []byte,
 	compressor compression.Compressor,
-	decompressor compression.Decompressor,
+	decompressors [compression.NumAlgorithms]compression.Decompressor,
 ) {
 	// buf1 will hold the compressed data; it can get a bit larger in the worst
 	// case, add a bit of head
@@ -77,14 +80,14 @@ func (a *BlockAnalyzer) runExperiment(
 	// Compress.
 	runtime.Gosched()
 	t1 := crtime.NowMono()
-	compressed := compressor.Compress(a.buf1[:0], block)
+	compressed, setting := compressor.Compress(a.buf1[:0], block)
 	compressionTime := t1.Elapsed()
 
 	// Yield the processor, reducing the chance that we get preempted during
 	// DecompressInto.
 	runtime.Gosched()
 	t2 := crtime.NowMono()
-	if err := decompressor.DecompressInto(a.buf2, compressed); err != nil {
+	if err := decompressors[setting.Algorithm].DecompressInto(a.buf2, compressed); err != nil {
 		panic(err)
 	}
 	decompressionTime := t2.Elapsed()