block: remove Buffer

The `block.Buffer` type does very little and is barely used. Use
`TempBuffer` directly instead.

Author: RaduBerinde

sstable/blob/blob.go

@@ -105,7 +105,7 @@ type FileWriter struct {
 	stats        FileWriterStats
 	flushGov     block.FlushGovernor
 	checksummer  block.Checksummer
-	compression  block.Compression
+	compressor   block.Compressor
 	cpuMeasurer  base.CPUMeasurer
 	writeQueue   struct {
 		wg  sync.WaitGroup
@@ -130,7 +130,7 @@ func NewFileWriter(fn base.DiskFileNum, w objstorage.Writable, opts FileWriterOp
 	fw.flushGov = opts.FlushGovernor
 	fw.indexEncoder.Init()
 	fw.checksummer = block.Checksummer{Type: opts.ChecksumType}
-	fw.compression = opts.Compression
+	fw.compressor = block.GetCompressor(opts.Compression)
 	fw.cpuMeasurer = opts.CpuMeasurer
 	fw.writeQueue.ch = make(chan compressedBlock)
 	fw.writeQueue.wg.Add(1)
@@ -212,7 +212,7 @@ func (w *FileWriter) flush() {
 	if w.valuesEncoder.Count() == 0 {
 		panic(errors.AssertionFailedf("no values to flush"))
 	}
-	pb, bh := block.CompressAndChecksumBufHandle(w.valuesEncoder.Finish(), w.compression, &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/block.go

@@ -144,6 +144,10 @@ type Checksummer struct {
 	blockTypeBuf [1]byte
 }
 
+func (c *Checksummer) Init(typ ChecksumType) {
+	c.Type = typ
+}
+
 // Checksum computes a checksum over the provided block and block type.
 func (c *Checksummer) Checksum(block []byte, blockType byte) (checksum uint32) {
 	// Calculate the checksum.

sstable/block/compression.go

@@ -252,17 +252,6 @@ func (b *PhysicalBlock) WriteTo(w objstorage.Writable) (n int, err error) {
 	return len(b.data) + len(b.trailer), nil
 }
 
-// CompressAndChecksumBufHandle is like CompressAndChecksum, but it will
-// retrieve a buffer for the compressed block from a sync.Pool and return a
-// *TempBuffer that the caller can use to release the buffer back to the pool.
-func CompressAndChecksumBufHandle(
-	blockData []byte, compression Compression, checksummer *Checksummer,
-) (PhysicalBlock, *TempBuffer) {
-	compressedBuf := NewTempBuffer()
-	pb := CompressAndChecksum(&compressedBuf.b, blockData, compression, checksummer)
-	return pb, compressedBuf
-}
-
 // CompressAndChecksum compresses and checksums the provided block, returning
 // the compressed block and its trailer. The result is appended to the dst
 // argument.
@@ -299,108 +288,26 @@ func CompressAndChecksumWithCompressor(
 	return pb
 }
 
-// A Buffer is a buffer for encoding a block. The caller mutates the buffer to
-// construct the uncompressed block, and calls CompressAndChecksum to produce
-// the physical, possibly-compressed PhysicalBlock. A Buffer recycles byte
-// slices used in construction of the uncompressed block and the compressed
-// physical block.
-type Buffer struct {
-	h           *TempBuffer
-	compression Compression
-	checksummer Checksummer
-}
-
-// Init configures the BlockBuffer with the specified compression and checksum
-// type.
-func (b *Buffer) Init(compression Compression, checksumType ChecksumType) {
-	b.h = NewTempBuffer()
-	b.h.b = b.h.b[:0]
-	b.compression = compression
-	b.checksummer.Type = checksumType
-}
-
-// Checksummer returns the Checksummer for the Buffer.
-func (b *Buffer) Checksummer() *Checksummer {
-	return &b.checksummer
-}
-
-// Get returns the byte slice currently backing the Buffer.
-func (b *Buffer) Get() []byte {
-	return b.h.b
-}
-
-// Size returns the current size of the buffer.
-func (b *Buffer) Size() int {
-	return len(b.h.b)
-}
-
-// Append appends the contents of v to the buffer, returning the offset at which
-// it was appended, growing the buffer if necessary.
-func (b *Buffer) Append(v []byte) int {
-	// We may need to grow b.Buffer to accommodate the new value. If necessary,
-	// double the size of the buffer until it's sufficiently large.
-	off := len(b.h.b)
-	newLen := off + len(v)
-	if cap(b.h.b) < newLen {
-		size := max(2*cap(b.h.b), 1024)
-		for size < newLen {
-			size *= 2
-		}
-		b.h.b = slices.Grow(b.h.b, size-len(b.h.b))
-	}
-	b.h.b = b.h.b[:newLen]
-	if n := copy(b.h.b[off:], v); n != len(v) {
-		panic("incorrect length computation")
-	}
-	return off
-}
-
-// Resize resizes the buffer to the specified length, allocating if necessary.
-func (b *Buffer) Resize(length int) {
-	if length > cap(b.h.b) {
-		b.h.b = slices.Grow(b.h.b, length-len(b.h.b))
-	}
-	b.h.b = b.h.b[:length]
-}
-
-// CompressAndChecksum compresses and checksums the block data, returning a
-// PhysicalBlock that is owned by the caller. The returned PhysicalBlock's
-// memory is backed by the returned TempBuffer. If non-nil, the returned
-// TempBuffer may be Released once the caller is done with the physical block to
-// recycle the block's underlying memory.
-//
-// When CompressAndChecksum returns, the callee has been reset and is ready to
-// be reused.
-func (b *Buffer) CompressAndChecksum() (PhysicalBlock, *TempBuffer) {
+// CompressAndChecksumToTempBuffer compresses and checksums the provided block
+// into a TempBuffer. The caller should Release() the TempBuffer once it is no
+// longer necessary.
+func CompressAndChecksumToTempBuffer(
+	blockData []byte, compressor Compressor, checksummer *Checksummer,
+) (PhysicalBlock, *TempBuffer) {
 	// Grab a buffer to use as the destination for compression.
 	compressedBuf := NewTempBuffer()
-	pb := CompressAndChecksum(&compressedBuf.b, b.h.b, b.compression, &b.checksummer)
-	b.h.b = b.h.b[:0]
+	pb := CompressAndChecksumWithCompressor(&compressedBuf.b, blockData, compressor, checksummer)
 	return pb, compressedBuf
 }
 
-// SetCompression changes the compression algorithm used by CompressAndChecksum.
-func (b *Buffer) SetCompression(compression Compression) {
-	b.compression = compression
-}
-
-// Release may be called when a buffer will no longer be used. It releases to
-// pools any memory held by the Buffer so that it may be reused.
-func (b *Buffer) Release() {
-	if b.h != nil {
-		b.h.Release()
-		b.h = nil
-	}
-}
-
 // TempBuffer is a buffer that is used temporarily and is released back to a
 // pool for reuse.
 type TempBuffer struct {
 	b []byte
 }
 
-// NewTempBuffer returns a TempBuffer from the pool. TempBuffer.b will have zero
-// length and arbitrary capacity.
+// NewTempBuffer returns a TempBuffer from the pool. The buffer will have zero
+// size and length and arbitrary capacity.
 func NewTempBuffer() *TempBuffer {
 	tb := tempBufferPool.Get().(*TempBuffer)
 	if invariants.Enabled && len(tb.b) > 0 {
@@ -409,6 +316,39 @@ func NewTempBuffer() *TempBuffer {
 	return tb
 }
 
+// Data returns the byte slice currently backing the Buffer.
+func (tb *TempBuffer) Data() []byte {
+	return tb.b
+}
+
+// Size returns the current size of the buffer.
+func (tb *TempBuffer) Size() int {
+	return len(tb.b)
+}
+
+// Append appends the contents of v to the buffer, growing the buffer if
+// necessary. Returns the offset at which it was appended.
+func (tb *TempBuffer) Append(v []byte) (startOffset int) {
+	startOffset = len(tb.b)
+	tb.b = append(tb.b, v...)
+	return startOffset
+}
+
+// Resize resizes the buffer to the specified length, allocating if necessary.
+// If the length is longer than the current length, the values of the new bytes
+// are arbitrary.
+func (tb *TempBuffer) Resize(length int) {
+	if length > cap(tb.b) {
+		tb.b = slices.Grow(tb.b, length-len(tb.b))
+	}
+	tb.b = tb.b[:length]
+}
+
+// Reset is equivalent to Resize(0).
+func (tb *TempBuffer) Reset() {
+	tb.b = tb.b[:0]
+}
+
 // Release releases the buffer back to the pool for reuse.
 func (tb *TempBuffer) Release() {
 	// Note we avoid releasing buffers that are larger than the configured

sstable/block/compression_test.go

@@ -18,8 +18,11 @@ func TestBufferRandomized(t *testing.T) {
 	t.Logf("seed %d", seed)
 	rng := rand.New(rand.NewPCG(0, seed))
 
-	var b Buffer
-	b.Init(SnappyCompression, ChecksumTypeCRC32c)
+	compressor := GetCompressor(SnappyCompression)
+	defer compressor.Close()
+	var checksummer Checksummer
+	checksummer.Init(ChecksumTypeCRC32c)
+	b := NewTempBuffer()
 	defer b.Release()
 	vbuf := make([]byte, 0, 1<<10) // 1 KiB
 
@@ -28,7 +31,7 @@ func TestBufferRandomized(t *testing.T) {
 			// Randomly release and reinitialize the buffer.
 			if rng.IntN(5) == 1 {
 				b.Release()
-				b.Init(SnappyCompression, ChecksumTypeCRC32c)
+				b = NewTempBuffer()
 			}
 
 			aggregateSizeOfKVs := rng.IntN(4<<20-(1<<10)) + 1<<10 // [1 KiB, 4 MiB)
@@ -46,10 +49,11 @@ func TestBufferRandomized(t *testing.T) {
 				b.Append(vbuf)
 				size += vlen
 				require.Equal(t, size, b.Size())
-				s := b.Get()
+				s := b.Data()
 				require.Equal(t, vbuf, s[len(s)-len(vbuf):])
 			}
-			_, bh := b.CompressAndChecksum()
+			_, bh := CompressAndChecksumToTempBuffer(b.Data(), compressor, &checksummer)
+			b.Reset()
 			bh.Release()
 		})
 	}

sstable/block/compressor.go

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

sstable/layout.go

@@ -758,7 +758,7 @@ func decodeColumnarMetaIndex(
 }
 
 // layoutWriter writes the structure of an sstable to durable storage. It
-// accepts serialized blocks, writes them to storage and returns a block handle
+// accepts serialized blocks, writes them to storage, and returns a block handle
 // describing the offset and length of the block.
 type layoutWriter struct {
 	writable objstorage.Writable
@@ -842,7 +842,7 @@ func (w *layoutWriter) WriteIndexBlock(b []byte) (block.Handle, error) {
 	return h, err
 }
 
-// WriteFilterBlock finishes the provided filter, constructs a trailer and
+// WriteFilterBlock finishes the provided filter, constructs a trailer, and
 // writes the block and trailer to the writer. It automatically adds the filter
 // block to the file's meta index when the writer is finished.
 func (w *layoutWriter) WriteFilterBlock(f filterWriter) (bh block.Handle, err error) {

sstable/valblk/writer.go

@@ -19,8 +19,10 @@ type Writer struct {
 	// Block finished callback.
 	blockFinishedFunc func(compressedSize int)
 
+	compressor  block.Compressor
+	checksummer block.Checksummer
 	// buf is the current block being written to (uncompressed).
-	buf block.Buffer
+	buf *block.TempBuffer
 	// Sequence of blocks that are finished.
 	blocks []bufferedValueBlock
 	// Cumulative value block bytes written so far.
@@ -54,7 +56,9 @@ func NewWriter(
 		blockFinishedFunc: blockFinishedFunc,
 		blocks:            w.blocks[:0],
 	}
-	w.buf.Init(compression, checksumType)
+	w.compressor = block.GetCompressor(compression)
+	w.checksummer.Init(checksumType)
+	w.buf = block.NewTempBuffer()
 	return w
 }
 
@@ -87,7 +91,8 @@ func (w *Writer) Size() uint64 {
 }
 
 func (w *Writer) compressAndFlush() {
-	physicalBlock, bufHandle := w.buf.CompressAndChecksum()
+	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())
 	// blockFinishedFunc length excludes the block trailer.
@@ -147,11 +152,10 @@ func (w *Writer) Finish(layout LayoutWriter, fileOffset uint64) (IndexHandle, Wr
 }
 
 func (w *Writer) writeValueBlocksIndex(layout LayoutWriter, h IndexHandle) (IndexHandle, error) {
-	w.buf.SetCompression(block.NoCompression)
 	blockLen := h.RowWidth() * len(w.blocks)
 	h.Handle.Length = uint64(blockLen)
 	w.buf.Resize(blockLen)
-	b := w.buf.Get()
+	b := w.buf.Data()
 	for i := range w.blocks {
 		littleEndianPut(uint64(i), b, int(h.BlockNumByteLength))
 		b = b[int(h.BlockNumByteLength):]
@@ -163,7 +167,7 @@ func (w *Writer) writeValueBlocksIndex(layout LayoutWriter, h IndexHandle) (Inde
 	if len(b) != 0 {
 		panic("incorrect length calculation")
 	}
-	pb, bufHandle := w.buf.CompressAndChecksum()
+	pb, bufHandle := block.CompressAndChecksumToTempBuffer(w.buf.Data(), block.NoopCompressor, &w.checksummer)
 	if _, err := layout.WriteValueIndexBlock(pb, h); err != nil {
 		return IndexHandle{}, err
 	}
@@ -179,7 +183,8 @@ func (w *Writer) Release() {
 		w.blocks[i] = bufferedValueBlock{}
 	}
 	w.buf.Release()
-	w.buf = block.Buffer{}
+	w.buf = nil
+	w.compressor.Close()
 	*w = Writer{blocks: w.blocks[:0]}
 	valueBlockWriterPool.Put(w)
 }
@@ -202,7 +207,7 @@ type LayoutWriter interface {
 	// WriteValueBlock writes a pre-finished value block (with the trailer) to
 	// the writer.
 	WriteValueBlock(blk block.PhysicalBlock) (block.Handle, error)
-	// WriteValueBlockIndex writes a pre-finished value block index to the
+	// WriteValueIndexBlock writes a pre-finished value block index to the
 	// writer.
 	WriteValueIndexBlock(blk block.PhysicalBlock, vbih IndexHandle) (block.Handle, error)
 }