block: remove CompressAndChecksum variant without compressor

We now require a compressor for `CompressAndChecksum`.

Author: RaduBerinde

sstable/blob/blob.go

@@ -282,7 +282,7 @@ func (w *FileWriter) Close() (FileWriterStats, error) {
 	{
 		indexBlock := w.indexEncoder.Finish()
 		var compressedBuf []byte
-		pb := block.CompressAndChecksum(&compressedBuf, indexBlock, block.NoCompression, &w.checksummer)
+		pb := block.CompressAndChecksum(&compressedBuf, indexBlock, 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

@@ -260,14 +260,6 @@ 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, c Compression, checksummer *Checksummer,
-) PhysicalBlock {
-	compressor := GetCompressor(c)
-	defer compressor.Close()
-	return CompressAndChecksumWithCompressor(dst, blockData, compressor, checksummer)
-}
-
-func CompressAndChecksumWithCompressor(
 	dst *[]byte, blockData []byte, compressor Compressor, checksummer *Checksummer,
 ) PhysicalBlock {
 	buf := (*dst)[:0]
@@ -296,7 +288,7 @@ func CompressAndChecksumToTempBuffer(
 ) (PhysicalBlock, *TempBuffer) {
 	// Grab a buffer to use as the destination for compression.
 	compressedBuf := NewTempBuffer()
-	pb := CompressAndChecksumWithCompressor(&compressedBuf.b, blockData, compressor, checksummer)
+	pb := CompressAndChecksum(&compressedBuf.b, blockData, compressor, checksummer)
 	return pb, compressedBuf
 }
 

sstable/colblk_writer.go

@@ -720,7 +720,7 @@ func (w *RawColumnWriter) enqueueDataBlock(
 	// Serialize the data block, compress it and send it to the write queue.
 	cb := compressedBlockPool.Get().(*compressedBlock)
 	cb.blockBuf.checksummer.Type = w.opts.Checksum
-	cb.physical = block.CompressAndChecksumWithCompressor(
+	cb.physical = block.CompressAndChecksum(
 		&cb.blockBuf.dataBuf,
 		serializedBlock,
 		w.compressor,
@@ -1247,7 +1247,7 @@ func (w *RawColumnWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProper
 	// Serialize the data block, compress it and send it to the write queue.
 	cb := compressedBlockPool.Get().(*compressedBlock)
 	cb.blockBuf.checksummer.Type = w.opts.Checksum
-	cb.physical = block.CompressAndChecksumWithCompressor(
+	cb.physical = block.CompressAndChecksum(
 		&cb.blockBuf.dataBuf,
 		b,
 		w.compressor,

sstable/layout.go

@@ -916,7 +916,10 @@ func (w *layoutWriter) WriteValueIndexBlock(
 func (w *layoutWriter) writeBlock(
 	b []byte, compression block.Compression, buf *blockBuf,
 ) (block.Handle, error) {
-	pb := block.CompressAndChecksum(&buf.dataBuf, b, compression, &buf.checksummer)
+	// TODO(radu): store a compressor in the layoutWriter.
+	compressor := block.GetCompressor(compression)
+	defer compressor.Close()
+	pb := block.CompressAndChecksum(&buf.dataBuf, b, compressor, &buf.checksummer)
 	h, err := w.writePrecompressedBlock(pb)
 	return h, err
 }

sstable/rowblk_writer.go

@@ -489,7 +489,10 @@ func (d *dataBlockBuf) finish() {
 }
 
 func (d *dataBlockBuf) compressAndChecksum(c block.Compression) {
-	d.physical = block.CompressAndChecksum(&d.dataBuf, d.uncompressed, c, &d.checksummer)
+	// TODO(radu): pass a Compressor here.
+	compressor := block.GetCompressor(c)
+	defer compressor.Close()
+	d.physical = block.CompressAndChecksum(&d.dataBuf, d.uncompressed, compressor, &d.checksummer)
 }
 
 func (d *dataBlockBuf) shouldFlush(
@@ -1929,10 +1932,13 @@ func (w *RawRowWriter) copyDataBlocks(
 // addDataBlock implements RawWriter.
 func (w *RawRowWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProperties) error {
 	blockBuf := &w.dataBlockBuf.blockBuf
+	// TODO(radu): store a compressor in w.layout.
+	compressor := block.GetCompressor(w.layout.compression)
+	defer compressor.Close()
 	pb := block.CompressAndChecksum(
 		&blockBuf.dataBuf,
 		b,
-		w.layout.compression,
+		compressor,
 		&blockBuf.checksummer,
 	)
 

sstable/suffix_rewriter.go

@@ -170,6 +170,8 @@ func rewriteDataBlocksInParallel(
 			// We'll assume all blocks are _roughly_ equal so round-robin static partition
 			// of each worker doing every ith block is probably enough.
 			err := func() error {
+				compressor := block.GetCompressor(opts.Compression)
+				defer compressor.Close()
 				for i := worker; i < len(input); i += concurrency {
 					bh := input[i]
 					var err error
@@ -190,7 +192,7 @@ func rewriteDataBlocksInParallel(
 						return err
 					}
 					compressedBuf = compressedBuf[:cap(compressedBuf)]
-					finished := block.CompressAndChecksum(&compressedBuf, outputBlock, opts.Compression, &checksummer)
+					finished := block.CompressAndChecksum(&compressedBuf, outputBlock, compressor, &checksummer)
 					output[i].physical = finished.CloneWithByteAlloc(&blockAlloc)
 				}
 				return nil