sstable: remove checksummer from blockBuf

The compressor and checksummer are used together but in many code
paths we use a checksummer in a blockBuf. Remove this and always use a
checksummer initialized alongside the compressor.

Author: RaduBerinde

sstable/colblk_writer.go

@@ -112,8 +112,10 @@ type RawColumnWriter struct {
 	disableKeyOrderChecks bool
 	cpuMeasurer           base.CPUMeasurer
 
-	// RawColumnWriter writes data sequentially so each writer can have a compressor
-	compressor block.Compressor
+	// RawColumnWriter writes data sequentially so each writer can have a
+	// compressor and checksummer.
+	compressor  block.Compressor
+	checksummer block.Checksummer
 }
 
 // Assert that *RawColumnWriter implements RawWriter.
@@ -137,9 +139,9 @@ func newColumnarWriter(
 			SmallestSeqNum: math.MaxUint64,
 		},
 		opts:                  o,
-		layout:                makeLayoutWriter(writable, o),
 		disableKeyOrderChecks: o.internal.DisableKeyOrderChecks,
 	}
+	w.layout.Init(writable, o)
 	w.dataFlush = block.MakeFlushGovernor(o.BlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses)
 	w.indexFlush = block.MakeFlushGovernor(o.IndexBlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses)
 	w.dataBlock.Init(o.KeySchema)
@@ -150,7 +152,7 @@ func newColumnarWriter(
 	if !o.DisableValueBlocks {
 		w.valueBlock = valblk.NewWriter(
 			block.MakeFlushGovernor(o.BlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses),
-			&w.compressor, w.opts.Checksum, func(compressedSize int) {})
+			&w.compressor, &w.checksummer, func(compressedSize int) {})
 	}
 	if o.FilterPolicy != base.NoFilterPolicy {
 		switch o.FilterType {
@@ -197,6 +199,7 @@ func newColumnarWriter(
 	go w.drainWriteQueue()
 
 	w.compressor = block.MakeCompressor(w.opts.Compression)
+	w.checksummer.Init(w.opts.Checksum)
 	return w
 }
 
@@ -724,13 +727,12 @@ 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.blockBuf.dataBuf,
 		serializedBlock,
 		blockkind.SSTableData,
 		&w.compressor,
-		&cb.blockBuf.checksummer,
+		&w.checksummer,
 	)
 	return w.enqueuePhysicalBlock(cb, separator)
 }
@@ -1270,13 +1272,12 @@ func (w *RawColumnWriter) copyDataBlocks(
 func (w *RawColumnWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProperties) error {
 	// 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.blockBuf.dataBuf,
 		b,
 		blockkind.SSTableData,
 		&w.compressor,
-		&cb.blockBuf.checksummer,
+		&w.checksummer,
 	)
 	if err := w.enqueuePhysicalBlock(cb, sep); err != nil {
 		return err

sstable/layout.go

@@ -810,9 +810,10 @@ type layoutWriter struct {
 	cacheOpts sstableinternal.CacheOptions
 
 	// options copied from WriterOptions
-	tableFormat  TableFormat
-	compressor   block.Compressor
-	checksumType block.ChecksumType
+	tableFormat TableFormat
+
+	compressor  block.Compressor
+	checksummer block.Checksummer
 
 	// Attribute bitset of the sstable, derived from sstable Properties at the time
 	// of writing.
@@ -832,17 +833,13 @@ type layoutWriter struct {
 	buf                  blockBuf
 }
 
-func makeLayoutWriter(w objstorage.Writable, opts WriterOptions) layoutWriter {
-	return layoutWriter{
-		writable:     w,
-		cacheOpts:    opts.internal.CacheOpts,
-		tableFormat:  opts.TableFormat,
-		compressor:   block.MakeCompressor(opts.Compression),
-		checksumType: opts.Checksum,
-		buf: blockBuf{
-			checksummer: block.Checksummer{Type: opts.Checksum},
-		},
-	}
+func (w *layoutWriter) Init(writable objstorage.Writable, opts WriterOptions) {
+	*w = layoutWriter{}
+	w.writable = writable
+	w.cacheOpts = opts.internal.CacheOpts
+	w.tableFormat = opts.TableFormat
+	w.compressor = block.MakeCompressor(opts.Compression)
+	w.checksummer.Init(opts.Checksum)
 }
 
 type metaIndexHandle struct {
@@ -971,7 +968,7 @@ func (w *layoutWriter) WriteValueIndexBlock(
 
 // writeBlock checksums, compresses, and writes out a block.
 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)
+	pb := block.CompressAndChecksum(&buf.dataBuf, b, kind, &w.compressor, &w.checksummer)
 	h, err := w.writePrecompressedBlock(pb)
 	return h, err
 }
@@ -980,7 +977,7 @@ func (w *layoutWriter) writeBlock(b []byte, kind block.Kind, buf *blockBuf) (blo
 func (w *layoutWriter) writeBlockUncompressed(
 	b []byte, kind block.Kind, buf *blockBuf,
 ) (block.Handle, error) {
-	pb := block.CopyAndChecksum(&buf.dataBuf, b, kind, &w.compressor, &buf.checksummer)
+	pb := block.CopyAndChecksum(&buf.dataBuf, b, kind, &w.compressor, &w.checksummer)
 	h, err := w.writePrecompressedBlock(pb)
 	return h, err
 }
@@ -1072,7 +1069,7 @@ func (w *layoutWriter) Finish() (size uint64, err error) {
 	// Write the table footer.
 	footer := footer{
 		format:      w.tableFormat,
-		checksum:    w.checksumType,
+		checksum:    w.checksummer.Type,
 		metaindexBH: metaIndexHandle,
 		indexBH:     w.lastIndexBlockHandle,
 		attributes:  w.attributes,

sstable/rowblk_writer.go

@@ -56,7 +56,6 @@ type RawRowWriter struct {
 	isStrictObsolete     bool
 	writingToLowestLevel bool
 	restartInterval      int
-	checksumType         block.ChecksumType
 	// disableKeyOrderChecks disables the checks that keys are added to an
 	// sstable in order. It is intended for internal use only in the construction
 	// of invalid sstables for testing. See tool/make_test_sstables.go.
@@ -411,8 +410,7 @@ type blockBuf struct {
 	// compression is not used) for storing a copy of the data. It is re-used over
 	// the lifetime of the blockBuf, avoiding the allocation of a temporary buffer
 	// for each block.
-	dataBuf     []byte
-	checksummer block.Checksummer
+	dataBuf []byte
 }
 
 func (b *blockBuf) clear() {
@@ -477,19 +475,20 @@ var dataBlockBufPool = sync.Pool{
 	},
 }
 
-func newDataBlockBuf(restartInterval int, checksumType block.ChecksumType) *dataBlockBuf {
+func newDataBlockBuf(restartInterval int) *dataBlockBuf {
 	d := dataBlockBufPool.Get().(*dataBlockBuf)
 	d.dataBlock.RestartInterval = restartInterval
-	d.checksummer.Type = checksumType
 	return d
 }
 
 func (d *dataBlockBuf) finish() {
 	d.uncompressed = d.dataBlock.Finish()
 }
 
-func (d *dataBlockBuf) compressAndChecksum(compressor *block.Compressor) {
-	d.physical = block.CompressAndChecksum(&d.dataBuf, d.uncompressed, blockkind.SSTableData, compressor, &d.checksummer)
+func (d *dataBlockBuf) compressAndChecksum(
+	compressor *block.Compressor, checksummer *block.Checksummer,
+) {
+	d.physical = block.CompressAndChecksum(&d.dataBuf, d.uncompressed, blockkind.SSTableData, compressor, checksummer)
 }
 
 func (d *dataBlockBuf) shouldFlush(
@@ -994,7 +993,7 @@ func (w *RawRowWriter) flush(key InternalKey) error {
 	}
 	w.dataBlockBuf.finish()
 	w.maybeIncrementTombstoneDenseBlocks()
-	w.dataBlockBuf.compressAndChecksum(&w.layout.compressor)
+	w.dataBlockBuf.compressAndChecksum(&w.layout.compressor, &w.layout.checksummer)
 	// Since dataBlockEstimates.addInflightDataBlock was never called, the
 	// inflightSize is set to 0.
 	w.coordination.sizeEstimate.dataBlockCompressed(w.dataBlockBuf.physical.LengthWithoutTrailer(), 0)
@@ -1053,7 +1052,7 @@ func (w *RawRowWriter) flush(key InternalKey) error {
 
 	w.dataBlockBuf = nil
 	err = w.coordination.writeQueue.addSync(writeTask)
-	w.dataBlockBuf = newDataBlockBuf(w.restartInterval, w.checksumType)
+	w.dataBlockBuf = newDataBlockBuf(w.restartInterval)
 
 	return err
 }
@@ -1679,7 +1678,6 @@ func newRowWriter(writable objstorage.Writable, o WriterOptions) *RawRowWriter {
 	}
 	o = o.ensureDefaults()
 	w := &RawRowWriter{
-		layout: makeLayoutWriter(writable, o),
 		meta: WriterMetadata{
 			SmallestSeqNum: math.MaxUint64,
 		},
@@ -1694,7 +1692,6 @@ func newRowWriter(writable objstorage.Writable, o WriterOptions) *RawRowWriter {
 		isStrictObsolete:           o.IsStrictObsolete,
 		writingToLowestLevel:       o.WritingToLowestLevel,
 		restartInterval:            o.BlockRestartInterval,
-		checksumType:               o.Checksum,
 		disableKeyOrderChecks:      o.internal.DisableKeyOrderChecks,
 		indexBlock:                 newIndexBlockBuf(),
 		rangeDelBlock:              rowblk.Writer{RestartInterval: 1},
@@ -1704,25 +1701,22 @@ func newRowWriter(writable objstorage.Writable, o WriterOptions) *RawRowWriter {
 		numDeletionsThreshold:      o.NumDeletionsThreshold,
 		deletionSizeRatioThreshold: o.DeletionSizeRatioThreshold,
 	}
+	w.layout.Init(writable, o)
 	w.dataFlush = block.MakeFlushGovernor(o.BlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses)
 	w.indexFlush = block.MakeFlushGovernor(o.IndexBlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses)
 	if w.tableFormat >= TableFormatPebblev3 {
 		w.shortAttributeExtractor = o.ShortAttributeExtractor
 		if !o.DisableValueBlocks {
 			w.valueBlockWriter = valblk.NewWriter(
 				block.MakeFlushGovernor(o.BlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses),
-				&w.layout.compressor, w.checksumType, func(compressedSize int) {
+				&w.layout.compressor, &w.layout.checksummer, func(compressedSize int) {
 					w.coordination.sizeEstimate.dataBlockCompressed(compressedSize, 0)
 				},
 			)
 		}
 	}
 
-	w.dataBlockBuf = newDataBlockBuf(w.restartInterval, w.checksumType)
-
-	w.blockBuf = blockBuf{
-		checksummer: block.Checksummer{Type: o.Checksum},
-	}
+	w.dataBlockBuf = newDataBlockBuf(w.restartInterval)
 
 	w.coordination.init(w)
 	defer func() {
@@ -1929,7 +1923,7 @@ func (w *RawRowWriter) copyDataBlocks(
 func (w *RawRowWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProperties) error {
 	blockBuf := &w.dataBlockBuf.blockBuf
 	pb := block.CompressAndChecksum(
-		&blockBuf.dataBuf, b, blockkind.SSTableData, &w.layout.compressor, &blockBuf.checksummer,
+		&blockBuf.dataBuf, b, blockkind.SSTableData, &w.layout.compressor, &w.layout.checksummer,
 	)
 
 	// layout.WriteDataBlock keeps layout.offset up-to-date for us.

sstable/valblk/writer.go

@@ -21,7 +21,7 @@ type Writer struct {
 	blockFinishedFunc func(compressedSize int)
 
 	compressor  *block.Compressor
-	checksummer block.Checksummer
+	checksummer *block.Checksummer
 	// buf is the current block being written to (uncompressed).
 	buf *block.TempBuffer
 	// Sequence of blocks that are finished.
@@ -47,18 +47,18 @@ var valueBlockWriterPool = sync.Pool{
 func NewWriter(
 	flushGovernor block.FlushGovernor,
 	compressor *block.Compressor,
-	checksumType block.ChecksumType,
+	checksummer *block.Checksummer,
 	// compressedSize should exclude the block trailer.
 	blockFinishedFunc func(compressedSize int),
 ) *Writer {
 	w := valueBlockWriterPool.Get().(*Writer)
 	*w = Writer{
 		flush:             flushGovernor,
 		compressor:        compressor,
+		checksummer:       checksummer,
 		blockFinishedFunc: blockFinishedFunc,
 		blocks:            w.blocks[:0],
 	}
-	w.checksummer.Init(checksumType)
 	w.buf = block.NewTempBuffer()
 	return w
 }
@@ -93,7 +93,7 @@ func (w *Writer) Size() uint64 {
 
 func (w *Writer) compressAndFlush() {
 	physicalBlock, bufHandle := block.CompressAndChecksumToTempBuffer(
-		w.buf.Data(), blockkind.SSTableValue, w.compressor, &w.checksummer,
+		w.buf.Data(), blockkind.SSTableValue, w.compressor, w.checksummer,
 	)
 	w.buf.Reset()
 	bh := block.Handle{Offset: w.totalBlockBytes, Length: uint64(physicalBlock.LengthWithoutTrailer())}
@@ -170,7 +170,7 @@ func (w *Writer) writeValueBlocksIndex(layout LayoutWriter, h IndexHandle) (Inde
 	if len(b) != 0 {
 		panic("incorrect length calculation")
 	}
-	pb, bufHandle := block.CopyAndChecksumToTempBuffer(w.buf.Data(), blockkind.Metadata, w.compressor, &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
 	}

sstable/writer_test.go

@@ -600,7 +600,7 @@ func TestBlockBufClear(t *testing.T) {
 
 func TestClearDataBlockBuf(t *testing.T) {
 	defer leaktest.AfterTest(t)()
-	d := newDataBlockBuf(1, block.ChecksumTypeCRC32c)
+	d := newDataBlockBuf(1)
 	d.blockBuf.dataBuf = make([]byte, 1)
 	require.NoError(t, d.dataBlock.Add(ikey("apple"), nil))
 	require.NoError(t, d.dataBlock.Add(ikey("banana"), nil))