sstable: add Compressor to column writer

For future adaptive compressor changes, the columnar writer should have its own
compressor so that adaptive compressor state can be persisted 1:1 with a
columnar writer.

Author: EdwardX29

sstable/block/compression.go

@@ -217,24 +217,21 @@ func (b *PhysicalBlock) WriteTo(w objstorage.Writable) (n int, err error) {
 // is used to avoid unnecessary decompression overhead at read time.
 func CompressAndChecksum(
 	dst *[]byte, blockData []byte, compression Compression, checksummer *Checksummer,
+) PhysicalBlock {
+	compressor := GetCompressor(compression)
+	defer compressor.Close()
+	return CompressAndChecksumWithCompressor(dst, blockData, compressor, checksummer)
+}
+
+func CompressAndChecksumWithCompressor(
+	dst *[]byte, blockData []byte, compressor Compressor, checksummer *Checksummer,
 ) PhysicalBlock {
 	buf := (*dst)[:0]
 	// Compress the buffer, discarding the result if the improvement isn't at
 	// least 12.5%.
-	algo := NoCompressionIndicator
-	if compression != NoCompression {
-		compressor := GetCompressor(compression)
-		defer compressor.Close()
-		algo, buf = compressor.Compress(buf, blockData)
-		if len(buf) >= len(blockData)-len(blockData)/8 {
-			algo = NoCompressionIndicator
-		}
-	}
-	if algo == NoCompressionIndicator {
-		// We don't want to use the given blockData buffer directly: typically the
-		// result will be written to disk and that can mangle the buffer, leading to
-		// fragile code.
-		buf = append(buf[:0], blockData...)
+	algo, buf := compressor.Compress(buf, blockData)
+	if len(buf) >= len(blockData)-len(blockData)/8 {
+		algo, buf = (noopCompressor{}).Compress(buf, blockData)
 	}
 
 	*dst = buf

sstable/block/compressor.go

@@ -26,7 +26,8 @@ var _ Compressor = snappyCompressor{}
 var _ Compressor = minlzCompressor{}
 
 func (noopCompressor) Compress(dst, src []byte) (CompressionIndicator, []byte) {
-	panic("NoCompressionCompressor.Compress() should not be called.")
+	dst = append(dst[:0], src...)
+	return NoCompressionIndicator, dst
 }
 func (noopCompressor) Close() {}
 

sstable/colblk_writer.go

@@ -103,6 +103,9 @@ type RawColumnWriter struct {
 	validator             invariants.Value[*colblk.DataBlockValidator]
 	disableKeyOrderChecks bool
 	cpuMeasurer           base.CPUMeasurer
+
+	// RawColumnWriter writes data sequentially so each writer can have a compressor
+	compressor block.Compressor
 }
 
 // Assert that *RawColumnWriter implements RawWriter.
@@ -184,6 +187,8 @@ func newColumnarWriter(
 	w.writeQueue.wg.Add(1)
 	w.cpuMeasurer = cpuMeasurer
 	go w.drainWriteQueue()
+
+	w.compressor = block.GetCompressor(w.opts.Compression)
 	return w
 }
 
@@ -711,10 +716,10 @@ 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.CompressAndChecksum(
+	cb.physical = block.CompressAndChecksumWithCompressor(
 		&cb.blockBuf.dataBuf,
 		serializedBlock,
-		w.opts.Compression,
+		w.compressor,
 		&cb.blockBuf.checksummer,
 	)
 	return w.enqueuePhysicalBlock(cb, separator)
@@ -1040,6 +1045,7 @@ func (w *RawColumnWriter) Close() (err error) {
 		return err
 	}
 	w.meta.Properties = w.props
+	w.compressor.Close()
 	// Release any held memory and make any future calls error.
 	*w = RawColumnWriter{meta: w.meta, err: errWriterClosed}
 	return nil
@@ -1220,10 +1226,10 @@ 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.CompressAndChecksum(
+	cb.physical = block.CompressAndChecksumWithCompressor(
 		&cb.blockBuf.dataBuf,
 		b,
-		w.opts.Compression,
+		w.compressor,
 		&cb.blockBuf.checksummer,
 	)
 	if err := w.enqueuePhysicalBlock(cb, sep); err != nil {