block: remove NoopCompressor

Replace the NoopCompressor with new variants for `ChecksumAndCompress`
and inform the `block.Compressor` of any uncompressed blocks.

Author: RaduBerinde

sstable/blob/blob.go

@@ -293,7 +293,7 @@ func (w *FileWriter) Close() (FileWriterStats, error) {
 	{
 		indexBlock := w.indexEncoder.Finish()
 		var compressedBuf []byte
-		pb := block.CompressAndChecksum(&compressedBuf, indexBlock, blockkind.Metadata, block.NoopCompressor, &w.checksummer)
+		pb := block.CopyAndChecksum(&compressedBuf, indexBlock, blockkind.Metadata, &w.compressor, &w.checksummer)
 		if _, w.err = pb.WriteTo(w.w); w.err != nil {
 			err = w.err
 			if w.w != nil {

sstable/block/compression.go

@@ -340,10 +340,6 @@ func (b *PhysicalBlock) WriteTo(w objstorage.Writable) (n int, err error) {
 // CompressAndChecksum compresses and checksums the provided block, returning
 // the compressed block and its trailer. The result is appended to the dst
 // argument.
-//
-// If the compressed block is not sufficiently smaller than the original block,
-// 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, blockKind Kind, compressor *Compressor, checksummer *Checksummer,
 ) PhysicalBlock {
@@ -358,6 +354,27 @@ func CompressAndChecksum(
 	return pb
 }
 
+// CopyAndChecksum copies the provided block (without compressing it) and
+// checksums it, returning the physical block. The result is appended to the dst
+// argument.
+//
+// Note that we still need to provide a Compressor so we can inform it of the
+// uncompressed block (for statistics).
+func CopyAndChecksum(
+	dst *[]byte, blockData []byte, blockKind Kind, compressor *Compressor, checksummer *Checksummer,
+) PhysicalBlock {
+	buf := *dst
+	buf = append(buf[:0], blockData...)
+	*dst = buf
+
+	// Calculate the checksum.
+	pb := PhysicalBlock{data: buf}
+	checksum := checksummer.Checksum(buf, byte(NoCompressionIndicator))
+	pb.trailer = MakeTrailer(byte(NoCompressionIndicator), checksum)
+	compressor.UncompressedBlock(len(blockData), blockKind)
+	return pb
+}
+
 // CompressAndChecksumToTempBuffer compresses and checksums the provided block
 // into a TempBuffer. The caller should Release() the TempBuffer once it is no
 // longer necessary.
@@ -370,6 +387,18 @@ func CompressAndChecksumToTempBuffer(
 	return pb, compressedBuf
 }
 
+// CopyAndChecksumToTempBuffer copies (without compressing) and checksums
+// the provided block into a TempBuffer. The caller should Release() the
+// TempBuffer once it is no longer necessary.
+func CopyAndChecksumToTempBuffer(
+	blockData []byte, blockKind Kind, compressor *Compressor, checksummer *Checksummer,
+) (PhysicalBlock, *TempBuffer) {
+	// Grab a buffer to use as the destination for compression.
+	compressedBuf := NewTempBuffer()
+	pb := CopyAndChecksum(&compressedBuf.b, blockData, blockKind, compressor, checksummer)
+	return pb, compressedBuf
+}
+
 // TempBuffer is a buffer that is used temporarily and is released back to a
 // pool for reuse.
 type TempBuffer struct {

sstable/block/compressor.go

@@ -88,11 +88,11 @@ func (c *Compressor) Compress(dst, src []byte, kind Kind) (CompressionIndicator,
 	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 = &noopCompressor
-
-var noopCompressor = MakeCompressor(NoCompression)
+// UncompressedBlock informs the compressor that a block of the given size and
+// kind was written uncompressed. This is used so that the final statistics are
+// complete.
+func (c *Compressor) UncompressedBlock(size int, kind Kind) {
+}
 
 type Decompressor = compression.Decompressor
 

sstable/layout.go

@@ -858,7 +858,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, blockkind.SSTableData, &w.compressor, buf)
+	return w.writeBlock(b, blockkind.SSTableData, buf)
 }
 
 // WritePrecompressedDataBlock writes a pre-compressed data block and its
@@ -873,7 +873,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, blockkind.SSTableIndex, &w.compressor, &w.buf)
+	h, err := w.writeBlock(b, blockkind.SSTableIndex, &w.buf)
 	if err == nil {
 		w.lastIndexBlockHandle = h
 	}
@@ -888,51 +888,56 @@ func (w *layoutWriter) WriteFilterBlock(f filterWriter) (bh block.Handle, err er
 	if err != nil {
 		return block.Handle{}, err
 	}
-	return w.writeNamedBlock(b, blockkind.Filter, block.NoopCompressor, f.metaName())
+	return w.writeNamedBlockUncompressed(b, blockkind.Filter, f.metaName())
 }
 
 // WritePropertiesBlock constructs a trailer for the provided properties block
 // and writes the block and trailer to the writer. It automatically adds the
 // properties block to the file's meta index when the writer is finished.
-func (w *layoutWriter) WritePropertiesBlock(b []byte) (block.Handle, error) {
-	compressor := &w.compressor
+func (w *layoutWriter) WritePropertiesBlock(b []byte) (bh block.Handle, err error) {
 	// In v6 and earlier, we use a row oriented block with an infinite restart
 	// interval, which provides very good prefix compression. Since v7, we use the
 	// columnar format without prefix compression for this block; we enable block
 	// compression to compensate.
 	if w.tableFormat < TableFormatPebblev7 {
-		compressor = block.NoopCompressor
+		bh, err = w.writeBlockUncompressed(b, blockkind.Metadata, &w.buf)
+	} else {
+		bh, err = w.writeBlock(b, blockkind.Metadata, &w.buf)
+	}
+	if err == nil {
+		w.recordToMetaindex(metaPropertiesName, bh)
 	}
-	return w.writeNamedBlock(b, blockkind.Metadata, compressor, metaPropertiesName)
+	return bh, err
 }
 
 // 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, blockkind.RangeKey, block.NoopCompressor, metaRangeKeyName)
+	return w.writeNamedBlockUncompressed(b, blockkind.RangeKey, metaRangeKeyName)
 }
 
 // WriteBlobRefIndexBlock constructs a trailer for the provided blob reference
 // index block and writes the block and trailer to the writer. It automatically
 // adds the blob reference index block to the file's meta index when the writer
 // is finished.
 func (w *layoutWriter) WriteBlobRefIndexBlock(b []byte) (block.Handle, error) {
-	return w.writeNamedBlock(b, blockkind.BlobReferenceValueLivenessIndex, &w.compressor, metaBlobRefIndexName)
+	return w.writeNamedBlockUncompressed(b, blockkind.BlobReferenceValueLivenessIndex, metaBlobRefIndexName)
 }
 
 // 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, blockkind.RangeDel, block.NoopCompressor, metaRangeDelV2Name)
+	return w.writeNamedBlockUncompressed(b, blockkind.RangeDel, metaRangeDelV2Name)
 }
 
-func (w *layoutWriter) writeNamedBlock(
-	b []byte, kind block.Kind, compressor *block.Compressor, name string,
+// writeNamedBlockUncompressed writes a block without compressing it and adds it to the metaindex.
+func (w *layoutWriter) writeNamedBlockUncompressed(
+	b []byte, kind block.Kind, name string,
 ) (bh block.Handle, err error) {
-	bh, err = w.writeBlock(b, kind, compressor, &w.buf)
+	bh, err = w.writeBlockUncompressed(b, kind, &w.buf)
 	if err == nil {
 		w.recordToMetaindex(name, bh)
 	}
@@ -960,10 +965,17 @@ func (w *layoutWriter) WriteValueIndexBlock(
 }
 
 // writeBlock checksums, compresses, and writes out a block.
-func (w *layoutWriter) writeBlock(
-	b []byte, kind block.Kind, compressor *block.Compressor, buf *blockBuf,
+func (w *layoutWriter) writeBlock(b []byte, kind block.Kind, buf *blockBuf) (block.Handle, error) {
+	pb := block.CompressAndChecksum(&buf.dataBuf, b, kind, &w.compressor, &buf.checksummer)
+	h, err := w.writePrecompressedBlock(pb)
+	return h, err
+}
+
+// writeBlock checksums and writes out a block.
+func (w *layoutWriter) writeBlockUncompressed(
+	b []byte, kind block.Kind, buf *blockBuf,
 ) (block.Handle, error) {
-	pb := block.CompressAndChecksum(&buf.dataBuf, b, kind, compressor, &buf.checksummer)
+	pb := block.CopyAndChecksum(&buf.dataBuf, b, kind, &w.compressor, &buf.checksummer)
 	h, err := w.writePrecompressedBlock(pb)
 	return h, err
 }
@@ -1047,7 +1059,7 @@ func (w *layoutWriter) Finish() (size uint64, err error) {
 		}
 		b = bw.Finish()
 	}
-	metaIndexHandle, err := w.writeBlock(b, blockkind.Metadata, block.NoopCompressor, &w.buf)
+	metaIndexHandle, err := w.writeBlockUncompressed(b, blockkind.Metadata, &w.buf)
 	if err != nil {
 		return 0, err
 	}

sstable/valblk/writer.go

@@ -170,7 +170,7 @@ func (w *Writer) writeValueBlocksIndex(layout LayoutWriter, h IndexHandle) (Inde
 	if len(b) != 0 {
 		panic("incorrect length calculation")
 	}
-	pb, bufHandle := block.CompressAndChecksumToTempBuffer(w.buf.Data(), blockkind.Metadata, block.NoopCompressor, &w.checksummer)
+	pb, bufHandle := block.CopyAndChecksumToTempBuffer(w.buf.Data(), blockkind.Metadata, w.compressor, &w.checksummer)
 	if _, err := layout.WriteValueIndexBlock(pb, h); err != nil {
 		return IndexHandle{}, err
 	}