sstable: use single physicalBlockMaker in columnar writer

The columnar writer has its own `physicalBlockMaker` and the
`layoutWriter` has one as well. Using two for different purposes
results in incomplete statistics. Consolidate and only use the one in
`layoutWriter`.

Author: RaduBerinde

replay/testdata/corpus/simple_ingest

@@ -101,6 +101,6 @@ simple_ingest/MANIFEST-000008:
   size: 209
 simple_ingest/MANIFEST-000010: stat simple_ingest/MANIFEST-000010: file does not exist
 simple_ingest/000007.sst:
-  size: 757
+  size: 770
 simple_ingest/000009.sst:
-  size: 763
+  size: 765

replay/testdata/corpus/simple_val_sep

@@ -103,6 +103,6 @@ stat simple_val_sep/MANIFEST-000013 simple_val_sep/000015.sst simple_val_sep/000
 simple_val_sep/MANIFEST-000013:
   size: 250
 simple_val_sep/000015.sst:
-  size: 865
+  size: 878
 simple_val_sep/000016.blob:
   size: 97

replay/testdata/replay_ingest

@@ -6,24 +6,24 @@ tree
           /
             build/
       36      000002.log
-     757      000005.sst
-     757      000007.sst
-     763      000009.sst
+     770      000005.sst
+     770      000007.sst
+     765      000009.sst
        0      LOCK
      172      MANIFEST-000001
      209      MANIFEST-000008
     2800      OPTIONS-000003
        0      marker.format-version.000012.025
        0      marker.manifest.000002.MANIFEST-000008
             simple_ingest/
-     757      000007.sst
-     763      000009.sst
+     770      000007.sst
+     765      000009.sst
      172      MANIFEST-000001
      209      MANIFEST-000008
               checkpoint/
       25        000002.log
-     774        000004.sst
-     757        000005.sst
+     787        000004.sst
+     770        000005.sst
      172        MANIFEST-000001
     2800        OPTIONS-000003
        0        marker.format-version.000001.025

replay/testdata/replay_val_sep

@@ -7,12 +7,12 @@ tree
             build/
      906      000005.sst
      101      000006.blob
-     871      000008.sst
+     884      000008.sst
       97      000009.blob
       92      000011.log
-     756      000012.sst
+     769      000012.sst
       63      000014.log
-     865      000015.sst
+     878      000015.sst
       97      000016.blob
        0      LOCK
      152      MANIFEST-000010
@@ -21,16 +21,16 @@ tree
        0      marker.format-version.000011.024
        0      marker.manifest.000003.MANIFEST-000013
             simple_val_sep/
-     865      000015.sst
+     878      000015.sst
       97      000016.blob
      250      MANIFEST-000013
               checkpoint/
      906        000005.sst
      101        000006.blob
-     871        000008.sst
+     884        000008.sst
       97        000009.blob
       64        000011.log
-     756        000012.sst
+     769        000012.sst
      187        MANIFEST-000013
     2936        OPTIONS-000003
        0        marker.format-version.000001.024

sstable/colblk_writer.go

@@ -111,10 +111,6 @@ type RawColumnWriter struct {
 	validator             invariants.Value[*colblk.DataBlockValidator]
 	disableKeyOrderChecks bool
 	cpuMeasurer           base.CPUMeasurer
-
-	// RawColumnWriter writes data sequentially so each writer can have a
-	// physical block maker.
-	physBlockMaker block.PhysicalBlockMaker
 }
 
 // Assert that *RawColumnWriter implements RawWriter.
@@ -148,12 +144,11 @@ func newColumnarWriter(
 	w.topLevelIndexBlock.Init()
 	w.rangeDelBlock.Init(w.comparer.Equal)
 	w.rangeKeyBlock.Init(w.comparer.Equal)
-	w.physBlockMaker.Init(w.opts.Compression, w.opts.Checksum)
 	if !o.DisableValueBlocks {
 		flushGovernor := block.MakeFlushGovernor(o.BlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses)
 		// We use the value block writer in the same goroutine so it's safe to share
 		// the physBlockMaker.
-		w.valueBlock = valblk.NewWriter(flushGovernor, &w.physBlockMaker, func(compressedSize int) {})
+		w.valueBlock = valblk.NewWriter(flushGovernor, &w.layout.physBlockMaker, func(compressedSize int) {})
 	}
 	if o.FilterPolicy != base.NoFilterPolicy {
 		switch o.FilterType {
@@ -714,7 +709,7 @@ func (w *RawColumnWriter) enqueueDataBlock(
 	}
 
 	// Compress and checksum the data block and send it to the write queue.
-	pb := w.physBlockMaker.Make(serializedBlock, blockkind.SSTableData, block.NoFlags)
+	pb := w.layout.physBlockMaker.Make(serializedBlock, blockkind.SSTableData, block.NoFlags)
 	return w.enqueuePhysicalBlock(pb, separator)
 }
 
@@ -1035,7 +1030,7 @@ func (w *RawColumnWriter) Close() (err error) {
 			}
 		}
 
-		w.props.CompressionStats = w.physBlockMaker.Compressor.Stats().String()
+		w.props.CompressionStats = w.layout.physBlockMaker.Compressor.Stats().String()
 		var toWrite []byte
 		w.props.CompressionOptions = rocksDBCompressionOptions
 		if w.opts.TableFormat >= TableFormatPebblev7 {
@@ -1069,7 +1064,6 @@ func (w *RawColumnWriter) Close() (err error) {
 		return err
 	}
 	w.meta.Properties = w.props
-	w.physBlockMaker.Close()
 	// Release any held memory and make any future calls error.
 	*w = RawColumnWriter{meta: w.meta, err: errWriterClosed}
 	return nil
@@ -1089,7 +1083,7 @@ func (w *RawColumnWriter) rewriteSuffixes(
 		return errors.Wrap(err, "reading layout")
 	}
 	// Copy data blocks in parallel, rewriting suffixes as we go.
-	blocks, err := rewriteDataBlocksInParallel(r, sstBytes, wo, l.Data, from, to, concurrency, w.physBlockMaker.Compressor.Stats(), func() blockRewriter {
+	blocks, err := rewriteDataBlocksInParallel(r, sstBytes, wo, l.Data, from, to, concurrency, w.layout.physBlockMaker.Compressor.Stats(), func() blockRewriter {
 		return colblk.NewDataBlockRewriter(wo.KeySchema, w.comparer)
 	})
 	if err != nil {
@@ -1247,7 +1241,7 @@ func (w *RawColumnWriter) copyDataBlocks(
 // using CopySpan().
 func (w *RawColumnWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProperties) error {
 	// Compress and checksum the data block and send it to the write queue.
-	pb := w.physBlockMaker.Make(b, blockkind.SSTableData, block.NoFlags)
+	pb := w.layout.physBlockMaker.Make(b, blockkind.SSTableData, block.NoFlags)
 	if err := w.enqueuePhysicalBlock(pb, sep); err != nil {
 		return err
 	}