block: pass compressor as pointer

Author: RaduBerinde

sstable/blob/blob.go

@@ -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(), &w.compressor, &w.checksummer)
 	compressedLen := uint64(pb.LengthWithoutTrailer())
 	w.stats.BlockCount++
 	off := w.stats.FileLen

sstable/block/compression.go

@@ -260,7 +260,7 @@ 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, compressor *Compressor, checksummer *Checksummer,
 ) PhysicalBlock {
 	buf := (*dst)[:0]
 	// Compress the buffer, discarding the result if the improvement isn't at
@@ -284,7 +284,7 @@ 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, compressor *Compressor, checksummer *Checksummer,
 ) (PhysicalBlock, *TempBuffer) {
 	// Grab a buffer to use as the destination for compression.
 	compressedBuf := NewTempBuffer()

sstable/block/compression_test.go

@@ -52,7 +52,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(), &compressor, &checksummer)
 			b.Reset()
 			bh.Release()
 		})

sstable/block/compressor.go

@@ -40,7 +40,7 @@ func (c *Compressor) Close() {
 
 // NoopCompressor is a Compressor that does not compress data. It does not have
 // any state and can be used in parallel.
-var NoopCompressor = Compressor{
+var NoopCompressor = &Compressor{
 	algorithm:  compression.NoCompression,
 	compressor: compression.GetCompressor(compression.None),
 }

sstable/colblk_writer.go

@@ -723,7 +723,7 @@ func (w *RawColumnWriter) enqueueDataBlock(
 	cb.physical = block.CompressAndChecksum(
 		&cb.blockBuf.dataBuf,
 		serializedBlock,
-		w.compressor,
+		&w.compressor,
 		&cb.blockBuf.checksummer,
 	)
 	return w.enqueuePhysicalBlock(cb, separator)
@@ -1250,7 +1250,7 @@ func (w *RawColumnWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProper
 	cb.physical = block.CompressAndChecksum(
 		&cb.blockBuf.dataBuf,
 		b,
-		w.compressor,
+		&w.compressor,
 		&cb.blockBuf.checksummer,
 	)
 	if err := w.enqueuePhysicalBlock(cb, sep); err != nil {

sstable/layout.go

@@ -920,7 +920,7 @@ func (w *layoutWriter) writeBlock(
 	// TODO(radu): store a compressor in the layoutWriter.
 	compressor := block.MakeCompressor(compression)
 	defer compressor.Close()
-	pb := block.CompressAndChecksum(&buf.dataBuf, b, compressor, &buf.checksummer)
+	pb := block.CompressAndChecksum(&buf.dataBuf, b, &compressor, &buf.checksummer)
 	h, err := w.writePrecompressedBlock(pb)
 	return h, err
 }

sstable/rowblk_writer.go

@@ -492,7 +492,7 @@ func (d *dataBlockBuf) compressAndChecksum(c block.Compression) {
 	// TODO(radu): pass a Compressor here.
 	compressor := block.MakeCompressor(c)
 	defer compressor.Close()
-	d.physical = block.CompressAndChecksum(&d.dataBuf, d.uncompressed, compressor, &d.checksummer)
+	d.physical = block.CompressAndChecksum(&d.dataBuf, d.uncompressed, &compressor, &d.checksummer)
 }
 
 func (d *dataBlockBuf) shouldFlush(
@@ -1938,7 +1938,7 @@ func (w *RawRowWriter) addDataBlock(b, sep []byte, bhp block.HandleWithPropertie
 	pb := block.CompressAndChecksum(
 		&blockBuf.dataBuf,
 		b,
-		compressor,
+		&compressor,
 		&blockBuf.checksummer,
 	)
 

sstable/suffix_rewriter.go

@@ -192,7 +192,7 @@ func rewriteDataBlocksInParallel(
 						return err
 					}
 					compressedBuf = compressedBuf[:cap(compressedBuf)]
-					finished := block.CompressAndChecksum(&compressedBuf, outputBlock, compressor, &checksummer)
+					finished := block.CompressAndChecksum(&compressedBuf, outputBlock, &compressor, &checksummer)
 					output[i].physical = finished.CloneWithByteAlloc(&blockAlloc)
 				}
 				return nil

sstable/valblk/writer.go

@@ -91,7 +91,7 @@ func (w *Writer) Size() uint64 {
 }
 
 func (w *Writer) compressAndFlush() {
-	physicalBlock, bufHandle := block.CompressAndChecksumToTempBuffer(w.buf.Data(), w.compressor, &w.checksummer)
+	physicalBlock, bufHandle := block.CompressAndChecksumToTempBuffer(w.buf.Data(), &w.compressor, &w.checksummer)
 	w.buf.Reset()
 	bh := block.Handle{Offset: w.totalBlockBytes, Length: uint64(physicalBlock.LengthWithoutTrailer())}
 	w.totalBlockBytes += uint64(physicalBlock.LengthWithTrailer())