Repository: cockroachdb/pebble
Pull request #3786 (2024-07-23 17:15:14)

sstable: split Writer into Writer and RawWriter
The sstable.Writer had some ergonomic methods for writing to the Writer from non-internal client code (implementing the pebble.Writer interface). These methods are internally unused, but may be used by external clients writing to sstables for ingestion, storage outside the LSM, tests, etc. This commit splits the Writer into two types, with the sugaring methods on the Writer type and the lower-level internal methods on the RawWriter type.

In future work, sstable.RawWriter may become an interface to support two implementations: one for rowblk-based sstables and one for colblk-based sstables.

jbowens: TFTR!

Patch:
diff --git a/compaction.go b/compaction.go
index 44cd103d..5b84951b 100644
--- a/compaction.go
+++ b/compaction.go
@@ -2656,7 +2656,7 @@ func (c *compaction) makeVersionEdit(result compact.Result) (*versionEdit, error
 // compaction or flush.
 func (d *DB) newCompactionOutput(
 	jobID JobID, c *compaction, writerOpts sstable.WriterOptions,
-) (objstorage.ObjectMetadata, *sstable.Writer, CPUWorkHandle, error) {
+) (objstorage.ObjectMetadata, *sstable.RawWriter, CPUWorkHandle, error) {
 	d.mu.Lock()
 	diskFileNum := d.mu.versions.getNextDiskFileNum()
 	d.mu.Unlock()
@@ -2730,7 +2730,7 @@ func (d *DB) newCompactionOutput(
 		d.opts.Experimental.MaxWriterConcurrency > 0 &&
 			(cpuWorkHandle.Permitted() || d.opts.Experimental.ForceWriterParallelism)
 
-	tw := sstable.NewWriter(writable, writerOpts)
+	tw := sstable.NewRawWriter(writable, writerOpts)
 	return objMeta, tw, cpuWorkHandle, nil
 }
 
diff --git a/internal/compact/run.go b/internal/compact/run.go
index a15e380a..f1db931c 100644
--- a/internal/compact/run.go
+++ b/internal/compact/run.go
@@ -132,7 +132,7 @@ func (r *Runner) MoreDataToWrite() bool {
 // Result.Tables. Should only be called if MoreDataToWrite() returned true.
 //
 // WriteTable always closes the Writer.
-func (r *Runner) WriteTable(objMeta objstorage.ObjectMetadata, tw *sstable.Writer) {
+func (r *Runner) WriteTable(objMeta objstorage.ObjectMetadata, tw *sstable.RawWriter) {
 	if r.err != nil {
 		panic("error already encountered")
 	}
@@ -159,7 +159,7 @@ func (r *Runner) WriteTable(objMeta objstorage.ObjectMetadata, tw *sstable.Write
 	r.tables[len(r.tables)-1].WriterMeta = *writerMeta
 }
 
-func (r *Runner) writeKeysToTable(tw *sstable.Writer) (splitKey []byte, _ error) {
+func (r *Runner) writeKeysToTable(tw *sstable.RawWriter) (splitKey []byte, _ error) {
 	firstKey := base.MinUserKey(r.cmp, spanStartOrNil(&r.lastRangeDelSpan), spanStartOrNil(&r.lastRangeKeySpan))
 	if r.key != nil && firstKey == nil {
 		firstKey = r.key.UserKey
diff --git a/internal/compact/spans.go b/internal/compact/spans.go
index afe81062..37cfe486 100644
--- a/internal/compact/spans.go
+++ b/internal/compact/spans.go
@@ -182,7 +182,7 @@ func (c *RangeKeySpanCompactor) elideInLastStripe(
 //
 // The span can contain either only RANGEDEL keys or only range keys.
 func SplitAndEncodeSpan(
-	cmp base.Compare, span *keyspan.Span, upToKey []byte, tw *sstable.Writer,
+	cmp base.Compare, span *keyspan.Span, upToKey []byte, tw *sstable.RawWriter,
 ) error {
 	if span.Empty() {
 		return nil
diff --git a/metamorphic/build.go b/metamorphic/build.go
index 9a15c97a..5555e436 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -96,7 +96,7 @@ func writeSSTForIngestion(
 		if err != nil {
 			return nil, err
 		}
-		if err := w.Add(k.K, valBytes); err != nil {
+		if err := w.Raw().Add(k.K, valBytes); err != nil {
 			return nil, err
 		}
 	}
@@ -142,7 +142,7 @@ func writeSSTForIngestion(
 				collapsed.Keys[i].Trailer = base.MakeTrailer(0, collapsed.Keys[i].Kind())
 			}
 			keyspan.SortKeysByTrailer(&collapsed.Keys)
-			if err := w.EncodeSpan(&collapsed); err != nil {
+			if err := w.Raw().EncodeSpan(&collapsed); err != nil {
 				return nil, err
 			}
 		}
@@ -156,7 +156,7 @@ func writeSSTForIngestion(
 	if err := w.Close(); err != nil {
 		return nil, err
 	}
-	sstMeta, err := w.Metadata()
+	sstMeta, err := w.Raw().Metadata()
 	if err != nil {
 		return nil, err
 	}
diff --git a/metamorphic/ops.go b/metamorphic/ops.go
index e383155b..6b771097 100644
--- a/metamorphic/ops.go
+++ b/metamorphic/ops.go
@@ -1928,7 +1928,7 @@ func (r *replicateOp) runSharedReplicate(
 			if err != nil {
 				panic(err)
 			}
-			return w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val)
+			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val)
 		},
 		func(start, end []byte, seqNum base.SeqNum) error {
 			return w.DeleteRange(start, end)
@@ -1939,7 +1939,7 @@ func (r *replicateOp) runSharedReplicate(
 				End:   end,
 				Keys:  keys,
 			}
-			return w.EncodeSpan(&s)
+			return w.Raw().EncodeSpan(&s)
 		},
 		func(sst *pebble.SharedSSTMeta) error {
 			sharedSSTs = append(sharedSSTs, *sst)
@@ -1957,7 +1957,7 @@ func (r *replicateOp) runSharedReplicate(
 		h.Recordf("%s // %v", r, err)
 		return
 	}
-	meta, err := w.Metadata()
+	meta, err := w.Raw().Metadata()
 	if err != nil {
 		h.Recordf("%s // %v", r, err)
 		return
@@ -1992,7 +1992,7 @@ func (r *replicateOp) runExternalReplicate(
 			if err != nil {
 				panic(err)
 			}
-			return w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val)
+			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val)
 		},
 		func(start, end []byte, seqNum base.SeqNum) error {
 			return w.DeleteRange(start, end)
@@ -2003,7 +2003,7 @@ func (r *replicateOp) runExternalReplicate(
 				End:   end,
 				Keys:  keys,
 			}
-			return w.EncodeSpan(&s)
+			return w.Raw().EncodeSpan(&s)
 		},
 		nil,
 		func(sst *pebble.ExternalFile) error {
@@ -2021,7 +2021,7 @@ func (r *replicateOp) runExternalReplicate(
 		h.Recordf("%s // %v", r, err)
 		return
 	}
-	meta, err := w.Metadata()
+	meta, err := w.Raw().Metadata()
 	if err != nil {
 		h.Recordf("%s // %v", r, err)
 		return
@@ -2116,7 +2116,7 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 				}
 			}
 			keyspan.SortKeysByTrailer(&span.Keys)
-			if err := w.EncodeSpan(span); err != nil {
+			if err := w.Raw().EncodeSpan(span); err != nil {
 				panic(err)
 			}
 		}
diff --git a/sstable/copier.go b/sstable/copier.go
index 9a58ae18..113b2f5c 100644
--- a/sstable/copier.go
+++ b/sstable/copier.go
@@ -66,7 +66,7 @@ func CopySpan(
 		o.FilterPolicy = nil
 	}
 	o.TableFormat = r.tableFormat
-	w := NewWriter(output, o)
+	w := NewRawWriter(output, o)
 
 	// We don't want the writer to attempt to write out block property data in
 	// index blocks. This data won't be valid since we're not passing the actual
diff --git a/sstable/raw_writer.go b/sstable/raw_writer.go
new file mode 100644
index 00000000..75bc4804
--- /dev/null
+++ b/sstable/raw_writer.go
@@ -0,0 +1,2041 @@
+// Copyright 2011 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package sstable
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"math"
+	"runtime"
+	"slices"
+	"sort"
+	"sync"
+
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/bytealloc"
+	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/pebble/internal/keyspan"
+	"github.com/cockroachdb/pebble/internal/rangedel"
+	"github.com/cockroachdb/pebble/internal/rangekey"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/sstable/rowblk"
+)
+
+// encodedBHPEstimatedSize estimates the size of the encoded BlockHandleWithProperties.
+// It would also be nice to account for the length of the data block properties here,
+// but isn't necessary since this is an estimate.
+const encodedBHPEstimatedSize = binary.MaxVarintLen64 * 2
+
+var errWriterClosed = errors.New("pebble: writer is closed")
+
+// WriterMetadata holds info about a finished sstable.
+type WriterMetadata struct {
+	Size          uint64
+	SmallestPoint InternalKey
+	// LargestPoint, LargestRangeKey, LargestRangeDel should not be accessed
+	// before Writer.Close is called, because they may only be set on
+	// Writer.Close.
+	LargestPoint     InternalKey
+	SmallestRangeDel InternalKey
+	LargestRangeDel  InternalKey
+	SmallestRangeKey InternalKey
+	LargestRangeKey  InternalKey
+	HasPointKeys     bool
+	HasRangeDelKeys  bool
+	HasRangeKeys     bool
+	SmallestSeqNum   base.SeqNum
+	LargestSeqNum    base.SeqNum
+	Properties       Properties
+}
+
+// SetSmallestPointKey sets the smallest point key to the given key.
+// NB: this method set the "absolute" smallest point key. Any existing key is
+// overridden.
+func (m *WriterMetadata) SetSmallestPointKey(k InternalKey) {
+	m.SmallestPoint = k
+	m.HasPointKeys = true
+}
+
+// SetSmallestRangeDelKey sets the smallest rangedel key to the given key.
+// NB: this method set the "absolute" smallest rangedel key. Any existing key is
+// overridden.
+func (m *WriterMetadata) SetSmallestRangeDelKey(k InternalKey) {
+	m.SmallestRangeDel = k
+	m.HasRangeDelKeys = true
+}
+
+// SetSmallestRangeKey sets the smallest range key to the given key.
+// NB: this method set the "absolute" smallest range key. Any existing key is
+// overridden.
+func (m *WriterMetadata) SetSmallestRangeKey(k InternalKey) {
+	m.SmallestRangeKey = k
+	m.HasRangeKeys = true
+}
+
+// SetLargestPointKey sets the largest point key to the given key.
+// NB: this method set the "absolute" largest point key. Any existing key is
+// overridden.
+func (m *WriterMetadata) SetLargestPointKey(k InternalKey) {
+	m.LargestPoint = k
+	m.HasPointKeys = true
+}
+
+// SetLargestRangeDelKey sets the largest rangedel key to the given key.
+// NB: this method set the "absolute" largest rangedel key. Any existing key is
+// overridden.
+func (m *WriterMetadata) SetLargestRangeDelKey(k InternalKey) {
+	m.LargestRangeDel = k
+	m.HasRangeDelKeys = true
+}
+
+// SetLargestRangeKey sets the largest range key to the given key.
+// NB: this method set the "absolute" largest range key. Any existing key is
+// overridden.
+func (m *WriterMetadata) SetLargestRangeKey(k InternalKey) {
+	m.LargestRangeKey = k
+	m.HasRangeKeys = true
+}
+
+func (m *WriterMetadata) updateSeqNum(seqNum base.SeqNum) {
+	if m.SmallestSeqNum > seqNum {
+		m.SmallestSeqNum = seqNum
+	}
+	if m.LargestSeqNum < seqNum {
+		m.LargestSeqNum = seqNum
+	}
+}
+
+// flushDecisionOptions holds parameters to inform the sstable block flushing
+// heuristics.
+type flushDecisionOptions struct {
+	blockSize          int
+	blockSizeThreshold int
+	// sizeClassAwareThreshold takes precedence over blockSizeThreshold when the
+	// Writer is aware of the allocator's size classes.
+	sizeClassAwareThreshold int
+}
+
+// RawWriter is a table writer.
+type RawWriter struct {
+	layout layoutWriter
+	meta   WriterMetadata
+	err    error
+	// dataBlockOptions and indexBlockOptions are used to configure the sstable
+	// block flush heuristics.
+	dataBlockOptions  flushDecisionOptions
+	indexBlockOptions flushDecisionOptions
+	// The following fields are copied from Options.
+	compare              Compare
+	split                Split
+	formatKey            base.FormatKey
+	compression          Compression
+	separator            Separator
+	successor            Successor
+	tableFormat          TableFormat
+	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.
+	disableKeyOrderChecks bool
+	// With two level indexes, the index/filter of a SST file is partitioned into
+	// smaller blocks with an additional top-level index on them. When reading an
+	// index/filter, only the top-level index is loaded into memory. The two level
+	// index/filter then uses the top-level index to load on demand into the block
+	// cache the partitions that are required to perform the index/filter query.
+	//
+	// Two level indexes are enabled automatically when there is more than one
+	// index block.
+	//
+	// This is useful when there are very large index blocks, which generally occurs
+	// with the usage of large keys. With large index blocks, the index blocks fight
+	// the data blocks for block cache space and the index blocks are likely to be
+	// re-read many times from the disk. The top level index, which has a much
+	// smaller memory footprint, can be used to prevent the entire index block from
+	// being loaded into the block cache.
+	twoLevelIndex       bool
+	indexBlock          *indexBlockBuf
+	rangeDelBlock       rowblk.Writer
+	rangeKeyBlock       rowblk.Writer
+	topLevelIndexBlock  rowblk.Writer
+	props               Properties
+	blockPropCollectors []BlockPropertyCollector
+	obsoleteCollector   obsoleteKeyBlockPropertyCollector
+	blockPropsEncoder   blockPropertiesEncoder
+	// filter accumulates the filter block. If populated, the filter ingests
+	// either the output of w.split (i.e. a prefix extractor) if w.split is not
+	// nil, or the full keys otherwise.
+	filter          filterWriter
+	indexPartitions []indexBlockAndBlockProperties
+
+	// indexBlockAlloc is used to bulk-allocate byte slices used to store index
+	// blocks in indexPartitions. These live until the index finishes.
+	indexBlockAlloc []byte
+	// indexSepAlloc is used to bulk-allocate index block separator slices stored
+	// in indexPartitions. These live until the index finishes.
+	indexSepAlloc bytealloc.A
+
+	rangeKeyEncoder   rangekey.Encoder
+	rangeKeysBySuffix keyspan.KeysBySuffix
+	rangeKeySpan      keyspan.Span
+	// dataBlockBuf consists of the state which is currently owned by and used by
+	// the Writer client goroutine. This state can be handed off to other goroutines.
+	dataBlockBuf *dataBlockBuf
+	// blockBuf consists of the state which is owned by and used by the Writer client
+	// goroutine.
+	blockBuf blockBuf
+
+	coordination coordinationState
+
+	// Information (other than the byte slice) about the last point key, to
+	// avoid extracting it again.
+	lastPointKeyInfo pointKeyInfo
+
+	// For value blocks.
+	shortAttributeExtractor   base.ShortAttributeExtractor
+	requiredInPlaceValueBound UserKeyPrefixBound
+	// When w.tableFormat >= TableFormatPebblev3, valueBlockWriter is nil iff
+	// WriterOptions.DisableValueBlocks was true.
+	valueBlockWriter *valueBlockWriter
+
+	allocatorSizeClasses []int
+}
+
+type pointKeyInfo struct {
+	trailer base.InternalKeyTrailer
+	// Only computed when w.valueBlockWriter is not nil.
+	userKeyLen int
+	// prefixLen uses w.split, if not nil. Only computed when w.valueBlockWriter
+	// is not nil.
+	prefixLen int
+	// True iff the point was marked obsolete.
+	isObsolete bool
+}
+
+type coordinationState struct {
+	parallelismEnabled bool
+
+	// writeQueue is used to write data blocks to disk. The writeQueue is primarily
+	// used to maintain the order in which data blocks must be written to disk. For
+	// this reason, every single data block write must be done through the writeQueue.
+	writeQueue *writeQueue
+
+	sizeEstimate dataBlockEstimates
+}
+
+func (c *coordinationState) init(parallelismEnabled bool, writer *RawWriter) {
+	c.parallelismEnabled = parallelismEnabled
+	// useMutex is false regardless of parallelismEnabled, because we do not do
+	// parallel compression yet.
+	c.sizeEstimate.useMutex = false
+
+	// writeQueueSize determines the size of the write queue, or the number
+	// of items which can be added to the queue without blocking. By default, we
+	// use a writeQueue size of 0, since we won't be doing any block writes in
+	// parallel.
+	writeQueueSize := 0
+	if parallelismEnabled {
+		writeQueueSize = runtime.GOMAXPROCS(0)
+	}
+	c.writeQueue = newWriteQueue(writeQueueSize, writer)
+}
+
+// sizeEstimate is a general purpose helper for estimating two kinds of sizes:
+// A. The compressed sstable size, which is useful for deciding when to start
+//
+//	a new sstable during flushes or compactions. In practice, we use this in
+//	estimating the data size (excluding the index).
+//
+// B. The size of index blocks to decide when to start a new index block.
+//
+// There are some terminology peculiarities which are due to the origin of
+// sizeEstimate for use case A with parallel compression enabled (for which
+// the code has not been merged). Specifically this relates to the terms
+// "written" and "compressed".
+//   - The notion of "written" for case A is sufficiently defined by saying that
+//     the data block is compressed. Waiting for the actual data block write to
+//     happen can result in unnecessary estimation, when we already know how big
+//     it will be in compressed form. Additionally, with the forthcoming value
+//     blocks containing older MVCC values, these compressed block will be held
+//     in-memory until late in the sstable writing, and we do want to accurately
+//     account for them without waiting for the actual write.
+//     For case B, "written" means that the index entry has been fully
+//     generated, and has been added to the uncompressed block buffer for that
+//     index block. It does not include actually writing a potentially
+//     compressed index block.
+//   - The notion of "compressed" is to differentiate between a "inflight" size
+//     and the actual size, and is handled via computing a compression ratio
+//     observed so far (defaults to 1).
+//     For case A, this is actual data block compression, so the "inflight" size
+//     is uncompressed blocks (that are no longer being written to) and the
+//     "compressed" size is after they have been compressed.
+//     For case B the inflight size is for a key-value pair in the index for
+//     which the value size (the encoded size of the BlockHandleWithProperties)
+//     is not accurately known, while the compressed size is the size of that
+//     entry when it has been added to the (in-progress) index ssblock.
+//
+// Usage: To update state, one can optionally provide an inflight write value
+// using addInflight (used for case B). When something is "written" the state
+// can be updated using either writtenWithDelta or writtenWithTotal, which
+// provide the actual delta size or the total size (latter must be
+// monotonically non-decreasing). If there were no calls to addInflight, there
+// isn't any real estimation happening here. So case A does not do any real
+// estimation. However, when we introduce parallel compression, there will be
+// estimation in that the client goroutine will call addInFlight and the
+// compression goroutines will call writtenWithDelta.
+type sizeEstimate struct {
+	// emptySize is the size when there is no inflight data, and numEntries is 0.
+	// emptySize is constant once set.
+	emptySize uint64
+
+	// inflightSize is the estimated size of some inflight data which hasn't
+	// been written yet.
+	inflightSize uint64
+
+	// totalSize is the total size of the data which has already been written.
+	totalSize uint64
+
+	// numWrittenEntries is the total number of entries which have already been
+	// written.
+	numWrittenEntries uint64
+	// numInflightEntries is the total number of entries which are inflight, and
+	// haven't been written.
+	numInflightEntries uint64
+
+	// maxEstimatedSize stores the maximum result returned from sizeEstimate.size.
+	// It ensures that values returned from subsequent calls to Writer.EstimatedSize
+	// never decrease.
+	maxEstimatedSize uint64
+
+	// We assume that the entries added to the sizeEstimate can be compressed.
+	// For this reason, we keep track of a compressedSize and an uncompressedSize
+	// to compute a compression ratio for the inflight entries. If the entries
+	// aren't being compressed, then compressedSize and uncompressedSize must be
+	// equal.
+	compressedSize   uint64
+	uncompressedSize uint64
+}
+
+func (s *sizeEstimate) init(emptySize uint64) {
+	s.emptySize = emptySize
+}
+
+func (s *sizeEstimate) size() uint64 {
+	ratio := float64(1)
+	if s.uncompressedSize > 0 {
+		ratio = float64(s.compressedSize) / float64(s.uncompressedSize)
+	}
+	estimatedInflightSize := uint64(float64(s.inflightSize) * ratio)
+	total := s.totalSize + estimatedInflightSize
+	if total > s.maxEstimatedSize {
+		s.maxEstimatedSize = total
+	} else {
+		total = s.maxEstimatedSize
+	}
+
+	if total == 0 {
+		return s.emptySize
+	}
+
+	return total
+}
+
+func (s *sizeEstimate) numTotalEntries() uint64 {
+	return s.numWrittenEntries + s.numInflightEntries
+}
+
+func (s *sizeEstimate) addInflight(size int) {
+	s.numInflightEntries++
+	s.inflightSize += uint64(size)
+}
+
+func (s *sizeEstimate) writtenWithTotal(newTotalSize uint64, inflightSize int) {
+	finalEntrySize := int(newTotalSize - s.totalSize)
+	s.writtenWithDelta(finalEntrySize, inflightSize)
+}
+
+func (s *sizeEstimate) writtenWithDelta(finalEntrySize int, inflightSize int) {
+	if inflightSize > 0 {
+		// This entry was previously inflight, so we should decrement inflight
+		// entries and update the "compression" stats for future estimation.
+		s.numInflightEntries--
+		s.inflightSize -= uint64(inflightSize)
+		s.uncompressedSize += uint64(inflightSize)
+		s.compressedSize += uint64(finalEntrySize)
+	}
+	s.numWrittenEntries++
+	s.totalSize += uint64(finalEntrySize)
+}
+
+func (s *sizeEstimate) clear() {
+	*s = sizeEstimate{emptySize: s.emptySize}
+}
+
+type indexBlockBuf struct {
+	// block will only be accessed from the writeQueue.
+	block rowblk.Writer
+
+	size struct {
+		useMutex bool
+		mu       sync.Mutex
+		estimate sizeEstimate
+	}
+
+	// restartInterval matches indexBlockBuf.block.restartInterval. We store it twice, because the `block`
+	// must only be accessed from the writeQueue goroutine.
+	restartInterval int
+}
+
+func (i *indexBlockBuf) clear() {
+	i.block.Reset()
+	if i.size.useMutex {
+		i.size.mu.Lock()
+		defer i.size.mu.Unlock()
+	}
+	i.size.estimate.clear()
+	i.restartInterval = 0
+}
+
+var indexBlockBufPool = sync.Pool{
+	New: func() interface{} {
+		return &indexBlockBuf{}
+	},
+}
+
+const indexBlockRestartInterval = 1
+
+func newIndexBlockBuf(useMutex bool) *indexBlockBuf {
+	i := indexBlockBufPool.Get().(*indexBlockBuf)
+	i.size.useMutex = useMutex
+	i.restartInterval = indexBlockRestartInterval
+	i.block.RestartInterval = indexBlockRestartInterval
+	i.size.estimate.init(rowblk.EmptySize)
+	return i
+}
+
+func (i *indexBlockBuf) shouldFlush(
+	sep InternalKey, valueLen int, flushOptions flushDecisionOptions, sizeClassHints []int,
+) bool {
+	if i.size.useMutex {
+		i.size.mu.Lock()
+		defer i.size.mu.Unlock()
+	}
+
+	nEntries := i.size.estimate.numTotalEntries()
+	return shouldFlushWithHints(
+		sep.Size(), valueLen, i.restartInterval, int(i.size.estimate.size()),
+		int(nEntries), flushOptions, sizeClassHints)
+}
+
+func (i *indexBlockBuf) add(key InternalKey, value []byte, inflightSize int) {
+	i.block.Add(key, value)
+	size := i.block.EstimatedSize()
+	if i.size.useMutex {
+		i.size.mu.Lock()
+		defer i.size.mu.Unlock()
+	}
+	i.size.estimate.writtenWithTotal(uint64(size), inflightSize)
+}
+
+func (i *indexBlockBuf) finish() []byte {
+	b := i.block.Finish()
+	return b
+}
+
+func (i *indexBlockBuf) addInflight(inflightSize int) {
+	if i.size.useMutex {
+		i.size.mu.Lock()
+		defer i.size.mu.Unlock()
+	}
+	i.size.estimate.addInflight(inflightSize)
+}
+
+func (i *indexBlockBuf) estimatedSize() uint64 {
+	if i.size.useMutex {
+		i.size.mu.Lock()
+		defer i.size.mu.Unlock()
+	}
+
+	// Make sure that the size estimation works as expected when parallelism
+	// is disabled.
+	if invariants.Enabled && !i.size.useMutex {
+		if i.size.estimate.inflightSize != 0 {
+			panic("unexpected inflight entry in index block size estimation")
+		}
+
+		// NB: The i.block should only be accessed from the writeQueue goroutine,
+		// when parallelism is enabled. We break that invariant here, but that's
+		// okay since parallelism is disabled.
+		if i.size.estimate.size() != uint64(i.block.EstimatedSize()) {
+			panic("index block size estimation sans parallelism is incorrect")
+		}
+	}
+	return i.size.estimate.size()
+}
+
+// sizeEstimate is used for sstable size estimation. sizeEstimate can be
+// accessed by the Writer client and compressionQueue goroutines. Fields
+// should only be read/updated through the functions defined on the
+// *sizeEstimate type.
+type dataBlockEstimates struct {
+	// If we don't do block compression in parallel, then we don't need to take
+	// the performance hit of synchronizing using this mutex.
+	useMutex bool
+	mu       sync.Mutex
+
+	estimate sizeEstimate
+}
+
+// inflightSize is the uncompressed block size estimate which has been
+// previously provided to addInflightDataBlock(). If addInflightDataBlock()
+// has not been called, this must be set to 0. compressedSize is the
+// compressed size of the block.
+func (d *dataBlockEstimates) dataBlockCompressed(compressedSize int, inflightSize int) {
+	if d.useMutex {
+		d.mu.Lock()
+		defer d.mu.Unlock()
+	}
+	d.estimate.writtenWithDelta(compressedSize+block.TrailerLen, inflightSize)
+}
+
+// size is an estimated size of datablock data which has been written to disk.
+func (d *dataBlockEstimates) size() uint64 {
+	if d.useMutex {
+		d.mu.Lock()
+		defer d.mu.Unlock()
+	}
+	// If there is no parallel compression, there should not be any inflight bytes.
+	if invariants.Enabled && !d.useMutex {
+		if d.estimate.inflightSize != 0 {
+			panic("unexpected inflight entry in data block size estimation")
+		}
+	}
+	return d.estimate.size()
+}
+
+// Avoid linter unused error.
+var _ = (&dataBlockEstimates{}).addInflightDataBlock
+
+// NB: unused since no parallel compression.
+func (d *dataBlockEstimates) addInflightDataBlock(size int) {
+	if d.useMutex {
+		d.mu.Lock()
+		defer d.mu.Unlock()
+	}
+
+	d.estimate.addInflight(size)
+}
+
+var writeTaskPool = sync.Pool{
+	New: func() interface{} {
+		t := &writeTask{}
+		t.compressionDone = make(chan bool, 1)
+		return t
+	},
+}
+
+type blockBuf struct {
+	// tmp is a scratch buffer, large enough to hold either footerLen bytes,
+	// blockTrailerLen bytes, (5 * binary.MaxVarintLen64) bytes, and most
+	// likely large enough for a block handle with properties.
+	tmp [blockHandleLikelyMaxLen]byte
+	// compressedBuf is the destination buffer for compression. It is re-used over the
+	// lifetime of the blockBuf, avoiding the allocation of a temporary buffer for each block.
+	compressedBuf []byte
+	checksummer   block.Checksummer
+}
+
+func (b *blockBuf) clear() {
+	// We can't assign b.compressedBuf[:0] to compressedBuf because snappy relies
+	// on the length of the buffer, and not the capacity to determine if it needs
+	// to make an allocation.
+	*b = blockBuf{
+		compressedBuf: b.compressedBuf, checksummer: b.checksummer,
+	}
+}
+
+// A dataBlockBuf holds all the state required to compress and write a data block to disk.
+// A dataBlockBuf begins its lifecycle owned by the Writer client goroutine. The Writer
+// client goroutine adds keys to the sstable, writing directly into a dataBlockBuf's blockWriter
+// until the block is full. Once a dataBlockBuf's block is full, the dataBlockBuf may be passed
+// to other goroutines for compression and file I/O.
+type dataBlockBuf struct {
+	blockBuf
+	dataBlock rowblk.Writer
+
+	// uncompressed is a reference to a byte slice which is owned by the dataBlockBuf. It is the
+	// next byte slice to be compressed. The uncompressed byte slice will be backed by the
+	// dataBlock.buf.
+	uncompressed []byte
+	// compressed is a reference to a byte slice which is owned by the dataBlockBuf. It is the
+	// compressed byte slice which must be written to disk. The compressed byte slice may be
+	// backed by the dataBlock.buf, or the dataBlockBuf.compressedBuf, depending on whether
+	// we use the result of the compression.
+	compressed []byte
+	// trailer is the block trailer encoding the compression type and checksum.
+	trailer block.Trailer
+
+	// We're making calls to BlockPropertyCollectors from the Writer client goroutine. We need to
+	// pass the encoded block properties over to the write queue. To prevent copies, and allocations,
+	// we give each dataBlockBuf, a blockPropertiesEncoder.
+	blockPropsEncoder blockPropertiesEncoder
+	// dataBlockProps is set when Writer.finishDataBlockProps is called. The dataBlockProps slice is
+	// a shallow copy of the internal buffer of the dataBlockBuf.blockPropsEncoder.
+	dataBlockProps []byte
+
+	// sepScratch is reusable scratch space for computing separator keys.
+	sepScratch []byte
+}
+
+func (d *dataBlockBuf) clear() {
+	d.blockBuf.clear()
+	d.dataBlock.Reset()
+
+	d.uncompressed = nil
+	d.compressed = nil
+	d.dataBlockProps = nil
+	d.sepScratch = d.sepScratch[:0]
+}
+
+var dataBlockBufPool = sync.Pool{
+	New: func() interface{} {
+		return &dataBlockBuf{}
+	},
+}
+
+func newDataBlockBuf(restartInterval int, checksumType block.ChecksumType) *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(c Compression) {
+	d.compressed, d.trailer = compressAndChecksum(d.uncompressed, c, &d.blockBuf)
+}
+
+func (d *dataBlockBuf) shouldFlush(
+	key InternalKey, valueLen int, flushOptions flushDecisionOptions, sizeClassHints []int,
+) bool {
+	return shouldFlushWithHints(
+		key.Size(), valueLen, d.dataBlock.RestartInterval, d.dataBlock.EstimatedSize(),
+		d.dataBlock.EntryCount(), flushOptions, sizeClassHints)
+}
+
+type indexBlockAndBlockProperties struct {
+	nEntries int
+	// sep is the last key added to this block, for computing a separator later.
+	sep        InternalKey
+	properties []byte
+	// block is the encoded block produced by blockWriter.finish.
+	block []byte
+}
+
+// Add adds a key/value pair to the table being written. For a given Writer,
+// the keys passed to Add must be in increasing order. The exception to this
+// rule is range deletion tombstones. Range deletion tombstones need to be
+// added ordered by their start key, but they can be added out of order from
+// point entries. Additionally, range deletion tombstones must be fragmented
+// (i.e. by keyspan.Fragmenter).
+func (w *RawWriter) Add(key InternalKey, value []byte) error {
+	if w.isStrictObsolete {
+		return errors.Errorf("use AddWithForceObsolete")
+	}
+	return w.AddWithForceObsolete(key, value, false)
+}
+
+// AddWithForceObsolete must be used when writing a strict-obsolete sstable.
+//
+// forceObsolete indicates whether the caller has determined that this key is
+// obsolete even though it may be the latest point key for this userkey. This
+// should be set to true for keys obsoleted by RANGEDELs, and is required for
+// strict-obsolete sstables.
+//
+// Note that there are two properties, S1 and S2 (see comment in format.go)
+// that strict-obsolete ssts must satisfy. S2, due to RANGEDELs, is solely the
+// responsibility of the caller. S1 is solely the responsibility of the
+// callee.
+func (w *RawWriter) AddWithForceObsolete(key InternalKey, value []byte, forceObsolete bool) error {
+	if w.err != nil {
+		return w.err
+	}
+
+	switch key.Kind() {
+	case InternalKeyKindRangeDelete:
+		return w.addTombstone(key, value)
+	case base.InternalKeyKindRangeKeyDelete,
+		base.InternalKeyKindRangeKeySet,
+		base.InternalKeyKindRangeKeyUnset:
+		w.err = errors.Errorf(
+			"pebble: range keys must be added via one of the RangeKey* functions")
+		return w.err
+	}
+	return w.addPoint(key, value, forceObsolete)
+}
+
+func (w *RawWriter) makeAddPointDecisionV2(key InternalKey) error {
+	prevTrailer := w.lastPointKeyInfo.trailer
+	w.lastPointKeyInfo.trailer = key.Trailer
+	if w.dataBlockBuf.dataBlock.EntryCount() == 0 {
+		return nil
+	}
+	if !w.disableKeyOrderChecks {
+		prevPointUserKey := w.dataBlockBuf.dataBlock.CurUserKey()
+		cmpUser := w.compare(prevPointUserKey, key.UserKey)
+		if cmpUser > 0 || (cmpUser == 0 && prevTrailer <= key.Trailer) {
+			return errors.Errorf(
+				"pebble: keys must be added in strictly increasing order: %s, %s",
+				InternalKey{UserKey: prevPointUserKey, Trailer: prevTrailer}.Pretty(w.formatKey),
+				key.Pretty(w.formatKey))
+		}
+	}
+	return nil
+}
+
+// REQUIRES: at least one point has been written to the Writer.
+func (w *RawWriter) getLastPointUserKey() []byte {
+	if w.dataBlockBuf.dataBlock.EntryCount() == 0 {
+		panic(errors.AssertionFailedf("no point keys added to writer"))
+	}
+	return w.dataBlockBuf.dataBlock.CurUserKey()
+}
+
+// REQUIRES: w.tableFormat >= TableFormatPebblev3
+func (w *RawWriter) makeAddPointDecisionV3(
+	key InternalKey, valueLen int,
+) (setHasSamePrefix bool, writeToValueBlock bool, isObsolete bool, err error) {
+	prevPointKeyInfo := w.lastPointKeyInfo
+	w.lastPointKeyInfo.userKeyLen = len(key.UserKey)
+	w.lastPointKeyInfo.prefixLen = w.split(key.UserKey)
+	w.lastPointKeyInfo.trailer = key.Trailer
+	w.lastPointKeyInfo.isObsolete = false
+	if !w.meta.HasPointKeys {
+		return false, false, false, nil
+	}
+	keyKind := key.Trailer.Kind()
+	prevPointUserKey := w.getLastPointUserKey()
+	prevPointKey := InternalKey{UserKey: prevPointUserKey, Trailer: prevPointKeyInfo.trailer}
+	prevKeyKind := prevPointKeyInfo.trailer.Kind()
+	considerWriteToValueBlock := prevKeyKind == InternalKeyKindSet &&
+		keyKind == InternalKeyKindSet
+	if considerWriteToValueBlock && !w.requiredInPlaceValueBound.IsEmpty() {
+		keyPrefix := key.UserKey[:w.lastPointKeyInfo.prefixLen]
+		cmpUpper := w.compare(
+			w.requiredInPlaceValueBound.Upper, keyPrefix)
+		if cmpUpper <= 0 {
+			// Common case for CockroachDB. Make it empty since all future keys in
+			// this sstable will also have cmpUpper <= 0.
+			w.requiredInPlaceValueBound = UserKeyPrefixBound{}
+		} else if w.compare(keyPrefix, w.requiredInPlaceValueBound.Lower) >= 0 {
+			considerWriteToValueBlock = false
+		}
+	}
+	// cmpPrefix is initialized iff considerWriteToValueBlock.
+	var cmpPrefix int
+	var cmpUser int
+	if considerWriteToValueBlock {
+		// Compare the prefixes.
+		cmpPrefix = w.compare(prevPointUserKey[:prevPointKeyInfo.prefixLen],
+			key.UserKey[:w.lastPointKeyInfo.prefixLen])
+		cmpUser = cmpPrefix
+		if cmpPrefix == 0 {
+			// Need to compare suffixes to compute cmpUser.
+			cmpUser = w.compare(prevPointUserKey[prevPointKeyInfo.prefixLen:],
+				key.UserKey[w.lastPointKeyInfo.prefixLen:])
+		}
+	} else {
+		cmpUser = w.compare(prevPointUserKey, key.UserKey)
+	}
+	// Ensure that no one adds a point key kind without considering the obsolete
+	// handling for that kind.
+	switch keyKind {
+	case InternalKeyKindSet, InternalKeyKindSetWithDelete, InternalKeyKindMerge,
+		InternalKeyKindDelete, InternalKeyKindSingleDelete, InternalKeyKindDeleteSized:
+	default:
+		panic(errors.AssertionFailedf("unexpected key kind %s", keyKind.String()))
+	}
+	// If same user key, then the current key is obsolete if any of the
+	// following is true:
+	// C1 The prev key was obsolete.
+	// C2 The prev key was not a MERGE. When the previous key is a MERGE we must
+	//    preserve SET* and MERGE since their values will be merged into the
+	//    previous key. We also must preserve DEL* since there may be an older
+	//    SET*/MERGE in a lower level that must not be merged with the MERGE --
+	//    if we omit the DEL* that lower SET*/MERGE will become visible.
+	//
+	// Regardless of whether it is the same user key or not
+	// C3 The current key is some kind of point delete, and we are writing to
+	//    the lowest level, then it is also obsolete. The correctness of this
+	//    relies on the same user key not spanning multiple sstables in a level.
+	//
+	// C1 ensures that for a user key there is at most one transition from
+	// !obsolete to obsolete. Consider a user key k, for which the first n keys
+	// are not obsolete. We consider the various value of n:
+	//
+	// n = 0: This happens due to forceObsolete being set by the caller, or due
+	// to C3. forceObsolete must only be set due a RANGEDEL, and that RANGEDEL
+	// must also delete all the lower seqnums for the same user key. C3 triggers
+	// due to a point delete and that deletes all the lower seqnums for the same
+	// user key.
+	//
+	// n = 1: This is the common case. It happens when the first key is not a
+	// MERGE, or the current key is some kind of point delete.
+	//
+	// n > 1: This is due to a sequence of MERGE keys, potentially followed by a
+	// single non-MERGE key.
+	isObsoleteC1AndC2 := cmpUser == 0 &&
+		(prevPointKeyInfo.isObsolete || prevKeyKind != InternalKeyKindMerge)
+	isObsoleteC3 := w.writingToLowestLevel &&
+		(keyKind == InternalKeyKindDelete || keyKind == InternalKeyKindSingleDelete ||
+			keyKind == InternalKeyKindDeleteSized)
+	isObsolete = isObsoleteC1AndC2 || isObsoleteC3
+	// TODO(sumeer): storing isObsolete SET and SETWITHDEL in value blocks is
+	// possible, but requires some care in documenting and checking invariants.
+	// There is code that assumes nothing in value blocks because of single MVCC
+	// version (those should be ok). We have to ensure setHasSamePrefix is
+	// correctly initialized here etc.
+
+	if !w.disableKeyOrderChecks &&
+		(cmpUser > 0 || (cmpUser == 0 && prevPointKeyInfo.trailer <= key.Trailer)) {
+		return false, false, false, errors.Errorf(
+			"pebble: keys must be added in strictly increasing order: %s, %s",
+			prevPointKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
+	}
+	if !considerWriteToValueBlock {
+		return false, false, isObsolete, nil
+	}
+	// NB: it is possible that cmpUser == 0, i.e., these two SETs have identical
+	// user keys (because of an open snapshot). This should be the rare case.
+	setHasSamePrefix = cmpPrefix == 0
+	// Use of 0 here is somewhat arbitrary. Given the minimum 3 byte encoding of
+	// valueHandle, this should be > 3. But tiny values are common in test and
+	// unlikely in production, so we use 0 here for better test coverage.
+	const tinyValueThreshold = 0
+	// NB: setting WriterOptions.DisableValueBlocks does not disable the
+	// setHasSamePrefix optimization.
+	considerWriteToValueBlock = setHasSamePrefix && valueLen > tinyValueThreshold && w.valueBlockWriter != nil
+	return setHasSamePrefix, considerWriteToValueBlock, isObsolete, nil
+}
+
+func (w *RawWriter) addPoint(key InternalKey, value []byte, forceObsolete bool) error {
+	if w.isStrictObsolete && key.Kind() == InternalKeyKindMerge {
+		return errors.Errorf("MERGE not supported in a strict-obsolete sstable")
+	}
+	var err error
+	var setHasSameKeyPrefix, writeToValueBlock, addPrefixToValueStoredWithKey bool
+	var isObsolete bool
+	maxSharedKeyLen := len(key.UserKey)
+	if w.tableFormat >= TableFormatPebblev3 {
+		// maxSharedKeyLen is limited to the prefix of the preceding key. If the
+		// preceding key was in a different block, then the blockWriter will
+		// ignore this maxSharedKeyLen.
+		maxSharedKeyLen = w.lastPointKeyInfo.prefixLen
+		setHasSameKeyPrefix, writeToValueBlock, isObsolete, err =
+			w.makeAddPointDecisionV3(key, len(value))
+		addPrefixToValueStoredWithKey = key.Kind() == InternalKeyKindSet
+	} else {
+		err = w.makeAddPointDecisionV2(key)
+	}
+	if err != nil {
+		return err
+	}
+	isObsolete = w.tableFormat >= TableFormatPebblev4 && (isObsolete || forceObsolete)
+	w.lastPointKeyInfo.isObsolete = isObsolete
+	var valueStoredWithKey []byte
+	var prefix block.ValuePrefix
+	var valueStoredWithKeyLen int
+	if writeToValueBlock {
+		vh, err := w.valueBlockWriter.addValue(value)
+		if err != nil {
+			return err
+		}
+		n := encodeValueHandle(w.blockBuf.tmp[:], vh)
+		valueStoredWithKey = w.blockBuf.tmp[:n]
+		valueStoredWithKeyLen = len(valueStoredWithKey) + 1
+		var attribute base.ShortAttribute
+		if w.shortAttributeExtractor != nil {
+			// TODO(sumeer): for compactions, it is possible that the input sstable
+			// already has this value in the value section and so we have already
+			// extracted the ShortAttribute. Avoid extracting it again. This will
+			// require changing the Writer.Add interface.
+			if attribute, err = w.shortAttributeExtractor(
+				key.UserKey, w.lastPointKeyInfo.prefixLen, value); err != nil {
+				return err
+			}
+		}
+		prefix = block.ValueHandlePrefix(setHasSameKeyPrefix, attribute)
+	} else {
+		valueStoredWithKey = value
+		valueStoredWithKeyLen = len(value)
+		if addPrefixToValueStoredWithKey {
+			valueStoredWithKeyLen++
+		}
+		prefix = block.InPlaceValuePrefix(setHasSameKeyPrefix)
+	}
+
+	if err := w.maybeFlush(key, valueStoredWithKeyLen); err != nil {
+		return err
+	}
+
+	for i := range w.blockPropCollectors {
+		v := value
+		if addPrefixToValueStoredWithKey {
+			// Values for SET are not required to be in-place, and in the future may
+			// not even be read by the compaction, so pass nil values. Block
+			// property collectors in such Pebble DB's must not look at the value.
+			v = nil
+		}
+		if err := w.blockPropCollectors[i].AddPointKey(key, v); err != nil {
+			w.err = err
+			return err
+		}
+	}
+	if w.tableFormat >= TableFormatPebblev4 {
+		w.obsoleteCollector.AddPoint(isObsolete)
+	}
+
+	w.maybeAddToFilter(key.UserKey)
+	w.dataBlockBuf.dataBlock.AddWithOptionalValuePrefix(
+		key, isObsolete, valueStoredWithKey, maxSharedKeyLen, addPrefixToValueStoredWithKey, prefix,
+		setHasSameKeyPrefix)
+
+	w.meta.updateSeqNum(key.SeqNum())
+
+	if !w.meta.HasPointKeys {
+		k := w.dataBlockBuf.dataBlock.CurKey()
+		// NB: We need to ensure that SmallestPoint.UserKey is set, so we create
+		// an InternalKey which is semantically identical to the key, but won't
+		// have a nil UserKey. We do this, because key.UserKey could be nil, and
+		// we don't want SmallestPoint.UserKey to be nil.
+		//
+		// todo(bananabrick): Determine if it's okay to have a nil SmallestPoint
+		// .UserKey now that we don't rely on a nil UserKey to determine if the
+		// key has been set or not.
+		w.meta.SetSmallestPointKey(k.Clone())
+	}
+
+	w.props.NumEntries++
+	switch key.Kind() {
+	case InternalKeyKindDelete, InternalKeyKindSingleDelete:
+		w.props.NumDeletions++
+		w.props.RawPointTombstoneKeySize += uint64(len(key.UserKey))
+	case InternalKeyKindDeleteSized:
+		var size uint64
+		if len(value) > 0 {
+			var n int
+			size, n = binary.Uvarint(value)
+			if n <= 0 {
+				w.err = errors.Newf("%s key's value (%x) does not parse as uvarint",
+					errors.Safe(key.Kind().String()), value)
+				return w.err
+			}
+		}
+		w.props.NumDeletions++
+		w.props.NumSizedDeletions++
+		w.props.RawPointTombstoneKeySize += uint64(len(key.UserKey))
+		w.props.RawPointTombstoneValueSize += size
+	case InternalKeyKindMerge:
+		w.props.NumMergeOperands++
+	}
+	w.props.RawKeySize += uint64(key.Size())
+	w.props.RawValueSize += uint64(len(value))
+	return nil
+}
+
+func (w *RawWriter) prettyTombstone(k InternalKey, value []byte) fmt.Formatter {
+	return keyspan.Span{
+		Start: k.UserKey,
+		End:   value,
+		Keys:  []keyspan.Key{{Trailer: k.Trailer}},
+	}.Pretty(w.formatKey)
+}
+
+func (w *RawWriter) addTombstone(key InternalKey, value []byte) error {
+	if !w.disableKeyOrderChecks && w.rangeDelBlock.EntryCount() > 0 {
+		// Check that tombstones are being added in fragmented order. If the two
+		// tombstones overlap, their start and end keys must be identical.
+		prevKey := w.rangeDelBlock.CurKey()
+		switch c := w.compare(prevKey.UserKey, key.UserKey); {
+		case c > 0:
+			w.err = errors.Errorf("pebble: keys must be added in order: %s, %s",
+				prevKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
+			return w.err
+		case c == 0:
+			prevValue := w.rangeDelBlock.CurValue()
+			if w.compare(prevValue, value) != 0 {
+				w.err = errors.Errorf("pebble: overlapping tombstones must be fragmented: %s vs %s",
+					w.prettyTombstone(prevKey, prevValue),
+					w.prettyTombstone(key, value))
+				return w.err
+			}
+			if prevKey.SeqNum() <= key.SeqNum() {
+				w.err = errors.Errorf("pebble: keys must be added in strictly increasing order: %s, %s",
+					prevKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
+				return w.err
+			}
+		default:
+			prevValue := w.rangeDelBlock.CurValue()
+			if w.compare(prevValue, key.UserKey) > 0 {
+				w.err = errors.Errorf("pebble: overlapping tombstones must be fragmented: %s vs %s",
+					w.prettyTombstone(prevKey, prevValue),
+					w.prettyTombstone(key, value))
+				return w.err
+			}
+		}
+	}
+
+	if key.Trailer == base.InternalKeyRangeDeleteSentinel {
+		w.err = errors.Errorf("pebble: cannot add range delete sentinel: %s", key.Pretty(w.formatKey))
+		return w.err
+	}
+
+	w.meta.updateSeqNum(key.SeqNum())
+
+	// Range tombstones are fragmented in the v2 range deletion block format,
+	// so the start key of the first range tombstone added will be the smallest
+	// range tombstone key. The largest range tombstone key will be determined
+	// in Writer.Close() as the end key of the last range tombstone added.
+	if w.props.NumRangeDeletions == 0 {
+		w.meta.SetSmallestRangeDelKey(key.Clone())
+	}
+
+	w.props.NumEntries++
+	w.props.NumDeletions++
+	w.props.NumRangeDeletions++
+	w.props.RawKeySize += uint64(key.Size())
+	w.props.RawValueSize += uint64(len(value))
+	w.rangeDelBlock.Add(key, value)
+	return nil
+}
+
+func (w *RawWriter) encodeFragmentedRangeKeySpan(span keyspan.Span) {
+	// This method is the emit function of the Fragmenter.
+	//
+	// NB: The span should only contain range keys and be internally consistent
+	// (eg, no duplicate suffixes, no additional keys after a RANGEKEYDEL).
+	//
+	// We use w.rangeKeysBySuffix and w.rangeKeySpan to avoid allocations.
+
+	// Sort the keys by suffix. Iteration doesn't *currently* depend on it, but
+	// we may want to in the future.
+	w.rangeKeysBySuffix.Cmp = w.compare
+	w.rangeKeysBySuffix.Keys = span.Keys
+	sort.Sort(&w.rangeKeysBySuffix)
+
+	w.rangeKeySpan = span
+	w.rangeKeySpan.Keys = w.rangeKeysBySuffix.Keys
+	if w.err == nil {
+		w.err = w.EncodeSpan(&w.rangeKeySpan)
+	}
+}
+
+// addRangeKey adds a range key set, unset, or delete key/value pair to the
+// table being written.
+//
+// Range keys must be supplied in strictly ascending order of start key (i.e.
+// user key ascending, sequence number descending, and key type descending).
+// Ranges added must also be supplied in fragmented span order - i.e. other than
+// spans that are perfectly aligned (same start and end keys), spans may not
+// overlap. Range keys may be added out of order relative to point keys and
+// range deletions.
+func (w *RawWriter) addRangeKey(key InternalKey, value []byte) error {
+	if !w.disableKeyOrderChecks && w.rangeKeyBlock.EntryCount() > 0 {
+		prevStartKey := w.rangeKeyBlock.CurKey()
+		prevEndKey, _, err := rangekey.DecodeEndKey(prevStartKey.Kind(), w.rangeKeyBlock.CurValue())
+		if err != nil {
+			// We panic here as we should have previously decoded and validated this
+			// key and value when it was first added to the range key block.
+			panic(err)
+		}
+
+		curStartKey := key
+		curEndKey, _, err := rangekey.DecodeEndKey(curStartKey.Kind(), value)
+		if err != nil {
+			w.err = err
+			return w.err
+		}
+
+		// Start keys must be strictly increasing.
+		if base.InternalCompare(w.compare, prevStartKey, curStartKey) >= 0 {
+			w.err = errors.Errorf(
+				"pebble: range keys starts must be added in increasing order: %s, %s",
+				prevStartKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
+			return w.err
+		}
+
+		// Start keys are increasing. If the start user keys are equal, the
+		// end keys must be equal (i.e. aligned spans).
+		if w.compare(prevStartKey.UserKey, curStartKey.UserKey) == 0 {
+			if w.compare(prevEndKey, curEndKey) != 0 {
+				w.err = errors.Errorf("pebble: overlapping range keys must be fragmented: %s, %s",
+					prevStartKey.Pretty(w.formatKey),
+					curStartKey.Pretty(w.formatKey))
+				return w.err
+			}
+		} else if w.compare(prevEndKey, curStartKey.UserKey) > 0 {
+			// If the start user keys are NOT equal, the spans must be disjoint (i.e.
+			// no overlap).
+			// NOTE: the inequality excludes zero, as we allow the end key of the
+			// lower span be the same as the start key of the upper span, because
+			// the range end key is considered an exclusive bound.
+			w.err = errors.Errorf("pebble: overlapping range keys must be fragmented: %s, %s",
+				prevStartKey.Pretty(w.formatKey),
+				curStartKey.Pretty(w.formatKey))
+			return w.err
+		}
+	}
+
+	// TODO(travers): Add an invariant-gated check to ensure that suffix-values
+	// are sorted within coalesced spans.
+
+	// Range-keys and point-keys are intended to live in "parallel" keyspaces.
+	// However, we track a single seqnum in the table metadata that spans both of
+	// these keyspaces.
+	// TODO(travers): Consider tracking range key seqnums separately.
+	w.meta.updateSeqNum(key.SeqNum())
+
+	// Range tombstones are fragmented, so the start key of the first range key
+	// added will be the smallest. The largest range key is determined in
+	// Writer.Close() as the end key of the last range key added to the block.
+	if w.props.NumRangeKeys() == 0 {
+		w.meta.SetSmallestRangeKey(key.Clone())
+	}
+
+	// Update table properties.
+	w.props.RawRangeKeyKeySize += uint64(key.Size())
+	w.props.RawRangeKeyValueSize += uint64(len(value))
+	switch key.Kind() {
+	case base.InternalKeyKindRangeKeyDelete:
+		w.props.NumRangeKeyDels++
+	case base.InternalKeyKindRangeKeySet:
+		w.props.NumRangeKeySets++
+	case base.InternalKeyKindRangeKeyUnset:
+		w.props.NumRangeKeyUnsets++
+	default:
+		panic(errors.Errorf("pebble: invalid range key type: %s", key.Kind()))
+	}
+
+	// Add the key to the block.
+	w.rangeKeyBlock.Add(key, value)
+	return nil
+}
+
+func (w *RawWriter) maybeAddToFilter(key []byte) {
+	if w.filter != nil {
+		prefix := key[:w.split(key)]
+		w.filter.addKey(prefix)
+	}
+}
+
+func (w *RawWriter) flush(key InternalKey) error {
+	// We're finishing a data block.
+	err := w.finishDataBlockProps(w.dataBlockBuf)
+	if err != nil {
+		return err
+	}
+	w.dataBlockBuf.finish()
+	w.dataBlockBuf.compressAndChecksum(w.compression)
+	// Since dataBlockEstimates.addInflightDataBlock was never called, the
+	// inflightSize is set to 0.
+	w.coordination.sizeEstimate.dataBlockCompressed(len(w.dataBlockBuf.compressed), 0)
+
+	// Determine if the index block should be flushed. Since we're accessing the
+	// dataBlockBuf.dataBlock.curKey here, we have to make sure that once we start
+	// to pool the dataBlockBufs, the curKey isn't used by the Writer once the
+	// dataBlockBuf is added back to a sync.Pool. In this particular case, the
+	// byte slice which supports "sep" will eventually be copied when "sep" is
+	// added to the index block.
+	prevKey := w.dataBlockBuf.dataBlock.CurKey()
+	sep := w.indexEntrySep(prevKey, key, w.dataBlockBuf)
+	// We determine that we should flush an index block from the Writer client
+	// goroutine, but we actually finish the index block from the writeQueue.
+	// When we determine that an index block should be flushed, we need to call
+	// BlockPropertyCollector.FinishIndexBlock. But block property collector
+	// calls must happen sequentially from the Writer client. Therefore, we need
+	// to determine that we are going to flush the index block from the Writer
+	// client.
+	shouldFlushIndexBlock := supportsTwoLevelIndex(w.tableFormat) && w.indexBlock.shouldFlush(
+		sep, encodedBHPEstimatedSize, w.indexBlockOptions, w.allocatorSizeClasses,
+	)
+
+	var indexProps []byte
+	var flushableIndexBlock *indexBlockBuf
+	if shouldFlushIndexBlock {
+		flushableIndexBlock = w.indexBlock
+		w.indexBlock = newIndexBlockBuf(w.coordination.parallelismEnabled)
+		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to
+		// flush the index block.
+		indexProps, err = w.finishIndexBlockProps()
+		if err != nil {
+			return err
+		}
+	}
+
+	// We've called BlockPropertyCollector.FinishDataBlock, and, if necessary,
+	// BlockPropertyCollector.FinishIndexBlock. Since we've decided to finish
+	// the data block, we can call
+	// BlockPropertyCollector.AddPrevDataBlockToIndexBlock.
+	w.addPrevDataBlockToIndexBlockProps()
+
+	// Schedule a write.
+	writeTask := writeTaskPool.Get().(*writeTask)
+	// We're setting compressionDone to indicate that compression of this block
+	// has already been completed.
+	writeTask.compressionDone <- true
+	writeTask.buf = w.dataBlockBuf
+	writeTask.indexEntrySep = sep
+	writeTask.currIndexBlock = w.indexBlock
+	writeTask.indexInflightSize = sep.Size() + encodedBHPEstimatedSize
+	writeTask.finishedIndexProps = indexProps
+	writeTask.flushableIndexBlock = flushableIndexBlock
+
+	// The writeTask corresponds to an unwritten index entry.
+	w.indexBlock.addInflight(writeTask.indexInflightSize)
+
+	w.dataBlockBuf = nil
+	if w.coordination.parallelismEnabled {
+		w.coordination.writeQueue.add(writeTask)
+	} else {
+		err = w.coordination.writeQueue.addSync(writeTask)
+	}
+	w.dataBlockBuf = newDataBlockBuf(w.restartInterval, w.checksumType)
+
+	return err
+}
+
+func (w *RawWriter) maybeFlush(key InternalKey, valueLen int) error {
+	if !w.dataBlockBuf.shouldFlush(key, valueLen, w.dataBlockOptions, w.allocatorSizeClasses) {
+		return nil
+	}
+
+	err := w.flush(key)
+
+	if err != nil {
+		w.err = err
+		return err
+	}
+
+	return nil
+}
+
+// dataBlockBuf.dataBlockProps set by this method must be encoded before any future use of the
+// dataBlockBuf.blockPropsEncoder, since the properties slice will get reused by the
+// blockPropsEncoder.
+func (w *RawWriter) finishDataBlockProps(buf *dataBlockBuf) error {
+	if len(w.blockPropCollectors) == 0 {
+		return nil
+	}
+	var err error
+	buf.blockPropsEncoder.resetProps()
+	for i := range w.blockPropCollectors {
+		scratch := buf.blockPropsEncoder.getScratchForProp()
+		if scratch, err = w.blockPropCollectors[i].FinishDataBlock(scratch); err != nil {
+			return err
+		}
+		buf.blockPropsEncoder.addProp(shortID(i), scratch)
+	}
+
+	buf.dataBlockProps = buf.blockPropsEncoder.unsafeProps()
+	return nil
+}
+
+// The BlockHandleWithProperties returned by this method must be encoded before any future use of
+// the Writer.blockPropsEncoder, since the properties slice will get reused by the blockPropsEncoder.
+// maybeAddBlockPropertiesToBlockHandle should only be called if block is being written synchronously
+// with the Writer client.
+func (w *RawWriter) maybeAddBlockPropertiesToBlockHandle(
+	bh block.Handle,
+) (BlockHandleWithProperties, error) {
+	err := w.finishDataBlockProps(w.dataBlockBuf)
+	if err != nil {
+		return BlockHandleWithProperties{}, err
+	}
+	return BlockHandleWithProperties{Handle: bh, Props: w.dataBlockBuf.dataBlockProps}, nil
+}
+
+func (w *RawWriter) indexEntrySep(
+	prevKey, key InternalKey, dataBlockBuf *dataBlockBuf,
+) InternalKey {
+	// Make a rough guess that we want key-sized scratch to compute the separator.
+	if cap(dataBlockBuf.sepScratch) < key.Size() {
+		dataBlockBuf.sepScratch = make([]byte, 0, key.Size()*2)
+	}
+
+	var sep InternalKey
+	if key.UserKey == nil && key.Trailer == 0 {
+		sep = prevKey.Successor(w.compare, w.successor, dataBlockBuf.sepScratch[:0])
+	} else {
+		sep = prevKey.Separator(w.compare, w.separator, dataBlockBuf.sepScratch[:0], key)
+	}
+	return sep
+}
+
+// addIndexEntry adds an index entry for the specified key and block handle.
+// addIndexEntry can be called from both the Writer client goroutine, and the
+// writeQueue goroutine. If the flushIndexBuf != nil, then the indexProps, as
+// they're used when the index block is finished.
+//
+// Invariant:
+//  1. addIndexEntry must not store references to the sep InternalKey, the tmp
+//     byte slice, bhp.Props. That is, these must be either deep copied or
+//     encoded.
+//  2. addIndexEntry must not hold references to the flushIndexBuf, and the writeTo
+//     indexBlockBufs.
+func (w *RawWriter) addIndexEntry(
+	sep InternalKey,
+	bhp BlockHandleWithProperties,
+	tmp []byte,
+	flushIndexBuf *indexBlockBuf,
+	writeTo *indexBlockBuf,
+	inflightSize int,
+	indexProps []byte,
+) error {
+	if bhp.Length == 0 {
+		// A valid blockHandle must be non-zero.
+		// In particular, it must have a non-zero length.
+		return nil
+	}
+
+	encoded := encodeBlockHandleWithProperties(tmp, bhp)
+
+	if flushIndexBuf != nil {
+		if cap(w.indexPartitions) == 0 {
+			w.indexPartitions = make([]indexBlockAndBlockProperties, 0, 32)
+		}
+		// Enable two level indexes if there is more than one index block.
+		w.twoLevelIndex = true
+		if err := w.finishIndexBlock(flushIndexBuf, indexProps); err != nil {
+			return err
+		}
+	}
+
+	writeTo.add(sep, encoded, inflightSize)
+	return nil
+}
+
+func (w *RawWriter) addPrevDataBlockToIndexBlockProps() {
+	for i := range w.blockPropCollectors {
+		w.blockPropCollectors[i].AddPrevDataBlockToIndexBlock()
+	}
+}
+
+// addIndexEntrySync adds an index entry for the specified key and block handle.
+// Writer.addIndexEntry is only called synchronously once Writer.Close is called.
+// addIndexEntrySync should only be called if we're sure that index entries
+// aren't being written asynchronously.
+//
+// Invariant:
+//  1. addIndexEntrySync must not store references to the prevKey, key InternalKey's,
+//     the tmp byte slice. That is, these must be either deep copied or encoded.
+//
+// TODO: Improve coverage of this method. e.g. tests passed without the line
+// `w.twoLevelIndex = true` previously.
+func (w *RawWriter) addIndexEntrySync(
+	prevKey, key InternalKey, bhp BlockHandleWithProperties, tmp []byte,
+) error {
+	return w.addIndexEntrySep(w.indexEntrySep(prevKey, key, w.dataBlockBuf), bhp, tmp)
+}
+
+func (w *RawWriter) addIndexEntrySep(
+	sep InternalKey, bhp BlockHandleWithProperties, tmp []byte,
+) error {
+	shouldFlush := supportsTwoLevelIndex(
+		w.tableFormat) && w.indexBlock.shouldFlush(
+		sep, encodedBHPEstimatedSize, w.indexBlockOptions, w.allocatorSizeClasses,
+	)
+	var flushableIndexBlock *indexBlockBuf
+	var props []byte
+	var err error
+	if shouldFlush {
+		flushableIndexBlock = w.indexBlock
+		w.indexBlock = newIndexBlockBuf(w.coordination.parallelismEnabled)
+		w.twoLevelIndex = true
+		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to
+		// flush the index block.
+		props, err = w.finishIndexBlockProps()
+		if err != nil {
+			return err
+		}
+	}
+
+	err = w.addIndexEntry(sep, bhp, tmp, flushableIndexBlock, w.indexBlock, 0, props)
+	if flushableIndexBlock != nil {
+		flushableIndexBlock.clear()
+		indexBlockBufPool.Put(flushableIndexBlock)
+	}
+	w.addPrevDataBlockToIndexBlockProps()
+	return err
+}
+
+func shouldFlushWithHints(
+	keyLen, valueLen int,
+	restartInterval, estimatedBlockSize, numEntries int,
+	flushOptions flushDecisionOptions,
+	sizeClassHints []int,
+) bool {
+	if numEntries == 0 {
+		return false
+	}
+
+	// If we are not informed about the memory allocator's size classes we fall
+	// back to a simple set of flush heuristics that are unaware of internal
+	// fragmentation in block cache allocations.
+	if len(sizeClassHints) == 0 {
+		return shouldFlushWithoutHints(
+			keyLen, valueLen, restartInterval, estimatedBlockSize, numEntries, flushOptions)
+	}
+
+	// For size-class aware flushing we need to account for the metadata that is
+	// allocated when this block is loaded into the block cache. For instance, if
+	// a block has size 1020B it may fit within a 1024B class. However, when
+	// loaded into the block cache we also allocate space for the cache entry
+	// metadata. The new allocation of size ~1052B may now only fit within a
+	// 2048B class, which increases internal fragmentation.
+	blockSizeWithMetadata := estimatedBlockSize + cache.ValueMetadataSize
+
+	// For the fast path we can avoid computing the exact varint encoded
+	// key-value pair size. Instead, we combine the key-value pair size with an
+	// upper-bound estimate of the associated metadata (4B restart point, 4B
+	// shared prefix length, 5B varint unshared key size, 5B varint value size).
+	newEstimatedSize := blockSizeWithMetadata + keyLen + valueLen + 18
+	// Our new block size estimate disregards key prefix compression. This puts
+	// us at risk of overestimating the size and flushing small blocks. We
+	// mitigate this by imposing a minimum size restriction.
+	if blockSizeWithMetadata <= flushOptions.sizeClassAwareThreshold || newEstimatedSize <= flushOptions.blockSize {
+		return false
+	}
+
+	sizeClass, ok := blockSizeClass(blockSizeWithMetadata, sizeClassHints)
+	// If the block size could not be mapped to a size class we fall back to
+	// using a simpler set of flush heuristics.
+	if !ok {
+		return shouldFlushWithoutHints(
+			keyLen, valueLen, restartInterval, estimatedBlockSize, numEntries, flushOptions)
+	}
+
+	// Tighter upper-bound estimate of the metadata stored with the next
+	// key-value pair.
+	newSize := blockSizeWithMetadata + keyLen + valueLen
+	if numEntries%restartInterval == 0 {
+		newSize += 4
+	}
+	newSize += 4                            // varint for shared prefix length
+	newSize += uvarintLen(uint32(keyLen))   // varint for unshared key bytes
+	newSize += uvarintLen(uint32(valueLen)) // varint for value size
+
+	if blockSizeWithMetadata < flushOptions.blockSize {
+		newSizeClass, ok := blockSizeClass(newSize, sizeClassHints)
+		if ok && newSizeClass-newSize >= sizeClass-blockSizeWithMetadata {
+			// Although the block hasn't reached the target size, waiting to insert the
+			// next entry would exceed the target and increase memory fragmentation.
+			return true
+		}
+		return false
+	}
+
+	// Flush if inserting the next entry bumps the block size to the memory
+	// allocator's next size class.
+	return newSize > sizeClass
+}
+
+func shouldFlushWithoutHints(
+	keyLen, valueLen int,
+	restartInterval, estimatedBlockSize, numEntries int,
+	flushOptions flushDecisionOptions,
+) bool {
+	if estimatedBlockSize >= flushOptions.blockSize {
+		return true
+	}
+
+	// The block is currently smaller than the target size.
+	if estimatedBlockSize <= flushOptions.blockSizeThreshold {
+		// The block is smaller than the threshold size at which we'll consider
+		// flushing it.
+		return false
+	}
+
+	newSize := estimatedBlockSize + keyLen + valueLen
+	if numEntries%restartInterval == 0 {
+		newSize += 4
+	}
+	newSize += 4                            // varint for shared prefix length
+	newSize += uvarintLen(uint32(keyLen))   // varint for unshared key bytes
+	newSize += uvarintLen(uint32(valueLen)) // varint for value size
+	// Flush if the block plus the new entry is larger than the target size.
+	return newSize > flushOptions.blockSize
+}
+
+// blockSizeClass returns the smallest memory allocator size class that could
+// hold a block of a given size and returns a boolean indicating whether an
+// appropriate size class was found. It is useful for computing the potential
+// space wasted by an allocation.
+func blockSizeClass(blockSize int, sizeClassHints []int) (int, bool) {
+	sizeClassIdx, _ := slices.BinarySearch(sizeClassHints, blockSize)
+	if sizeClassIdx == len(sizeClassHints) {
+		return -1, false
+	}
+	return sizeClassHints[sizeClassIdx], true
+}
+
+func cloneKeyWithBuf(k InternalKey, a bytealloc.A) (bytealloc.A, InternalKey) {
+	if len(k.UserKey) == 0 {
+		return a, k
+	}
+	a, keyCopy := a.Copy(k.UserKey)
+	return a, InternalKey{UserKey: keyCopy, Trailer: k.Trailer}
+}
+
+// Invariants: The byte slice returned by finishIndexBlockProps is heap-allocated
+//
+//	and has its own lifetime, independent of the Writer and the blockPropsEncoder,
+//
+// and it is safe to:
+//  1. Reuse w.blockPropsEncoder without first encoding the byte slice returned.
+//  2. Store the byte slice in the Writer since it is a copy and not supported by
+//     an underlying buffer.
+func (w *RawWriter) finishIndexBlockProps() ([]byte, error) {
+	w.blockPropsEncoder.resetProps()
+	for i := range w.blockPropCollectors {
+		scratch := w.blockPropsEncoder.getScratchForProp()
+		var err error
+		if scratch, err = w.blockPropCollectors[i].FinishIndexBlock(scratch); err != nil {
+			return nil, err
+		}
+		w.blockPropsEncoder.addProp(shortID(i), scratch)
+	}
+	return w.blockPropsEncoder.props(), nil
+}
+
+// finishIndexBlock finishes the current index block and adds it to the top
+// level index block. This is only used when two level indexes are enabled.
+//
+// Invariants:
+//  1. The props slice passed into finishedIndexBlock must not be a
+//     owned by any other struct, since it will be stored in the Writer.indexPartitions
+//     slice.
+//  2. None of the buffers owned by indexBuf will be shallow copied and stored elsewhere.
+//     That is, it must be safe to reuse indexBuf after finishIndexBlock has been called.
+func (w *RawWriter) finishIndexBlock(indexBuf *indexBlockBuf, props []byte) error {
+	part := indexBlockAndBlockProperties{
+		nEntries: indexBuf.block.EntryCount(), properties: props,
+	}
+	w.indexSepAlloc, part.sep = cloneKeyWithBuf(
+		indexBuf.block.CurKey(), w.indexSepAlloc,
+	)
+	bk := indexBuf.finish()
+	if len(w.indexBlockAlloc) < len(bk) {
+		// Allocate enough bytes for approximately 16 index blocks.
+		w.indexBlockAlloc = make([]byte, len(bk)*16)
+	}
+	n := copy(w.indexBlockAlloc, bk)
+	part.block = w.indexBlockAlloc[:n:n]
+	w.indexBlockAlloc = w.indexBlockAlloc[n:]
+	w.indexPartitions = append(w.indexPartitions, part)
+	return nil
+}
+
+func (w *RawWriter) writeTwoLevelIndex() (block.Handle, error) {
+	props, err := w.finishIndexBlockProps()
+	if err != nil {
+		return block.Handle{}, err
+	}
+	// Add the final unfinished index.
+	if err = w.finishIndexBlock(w.indexBlock, props); err != nil {
+		return block.Handle{}, err
+	}
+
+	for i := range w.indexPartitions {
+		b := &w.indexPartitions[i]
+		w.props.NumDataBlocks += uint64(b.nEntries)
+
+		data := b.block
+		w.props.IndexSize += uint64(len(data))
+		bh, err := w.layout.WriteIndexBlock(data)
+		if err != nil {
+			return block.Handle{}, err
+		}
+		bhp := BlockHandleWithProperties{
+			Handle: bh,
+			Props:  b.properties,
+		}
+		encoded := encodeBlockHandleWithProperties(w.blockBuf.tmp[:], bhp)
+		w.topLevelIndexBlock.Add(b.sep, encoded)
+	}
+
+	// NB: RocksDB includes the block trailer length in the index size
+	// property, though it doesn't include the trailer in the top level
+	// index size property.
+	w.props.IndexPartitions = uint64(len(w.indexPartitions))
+	w.props.TopLevelIndexSize = uint64(w.topLevelIndexBlock.EstimatedSize())
+	w.props.IndexSize += w.props.TopLevelIndexSize + block.TrailerLen
+	return w.layout.WriteIndexBlock(w.topLevelIndexBlock.Finish())
+}
+
+func compressAndChecksum(
+	b []byte, compression Compression, blockBuf *blockBuf,
+) (compressed []byte, trailer block.Trailer) {
+	// Compress the buffer, discarding the result if the improvement isn't at
+	// least 12.5%.
+	blockType, compressed := compressBlock(compression, b, blockBuf.compressedBuf)
+	if blockType != noCompressionBlockType && cap(compressed) > cap(blockBuf.compressedBuf) {
+		blockBuf.compressedBuf = compressed[:cap(compressed)]
+	}
+	if len(compressed) < len(b)-len(b)/8 {
+		b = compressed
+	} else {
+		blockType = noCompressionBlockType
+	}
+
+	// Calculate the checksum.
+	trailer[0] = byte(blockType)
+	checksum := blockBuf.checksummer.Checksum(b, trailer[:1])
+	return b, block.MakeTrailer(byte(blockType), checksum)
+}
+
+// assertFormatCompatibility ensures that the features present on the table are
+// compatible with the table format version.
+func (w *RawWriter) assertFormatCompatibility() error {
+	// PebbleDBv1: block properties.
+	if len(w.blockPropCollectors) > 0 && w.tableFormat < TableFormatPebblev1 {
+		return errors.Newf(
+			"table format version %s is less than the minimum required version %s for block properties",
+			w.tableFormat, TableFormatPebblev1,
+		)
+	}
+
+	// PebbleDBv2: range keys.
+	if w.props.NumRangeKeys() > 0 && w.tableFormat < TableFormatPebblev2 {
+		return errors.Newf(
+			"table format version %s is less than the minimum required version %s for range keys",
+			w.tableFormat, TableFormatPebblev2,
+		)
+	}
+
+	// PebbleDBv3: value blocks.
+	if (w.props.NumValueBlocks > 0 || w.props.NumValuesInValueBlocks > 0 ||
+		w.props.ValueBlocksSize > 0) && w.tableFormat < TableFormatPebblev3 {
+		return errors.Newf(
+			"table format version %s is less than the minimum required version %s for value blocks",
+			w.tableFormat, TableFormatPebblev3)
+	}
+
+	// PebbleDBv4: DELSIZED tombstones.
+	if w.props.NumSizedDeletions > 0 && w.tableFormat < TableFormatPebblev4 {
+		return errors.Newf(
+			"table format version %s is less than the minimum required version %s for sized deletion tombstones",
+			w.tableFormat, TableFormatPebblev4)
+	}
+	return nil
+}
+
+// UnsafeLastPointUserKey returns the last point key written to the writer to
+// which this option was passed during creation. The returned key points
+// directly into a buffer belonging to the Writer. The value's lifetime ends the
+// next time a point key is added to the Writer.
+//
+// Must not be called after Writer is closed.
+func (w *RawWriter) UnsafeLastPointUserKey() []byte {
+	if w != nil && w.dataBlockBuf.dataBlock.EntryCount() >= 1 {
+		// w.dataBlockBuf.dataBlock.curKey is guaranteed to point to the last point key
+		// which was added to the Writer.
+		return w.dataBlockBuf.dataBlock.CurUserKey()
+	}
+	return nil
+}
+
+// EncodeSpan encodes the keys in the given span. The span can contain either
+// only RANGEDEL keys or only range keys.
+//
+// This is a low-level API that bypasses the fragmenter. The spans passed to
+// this function must be fragmented and ordered.
+func (w *RawWriter) EncodeSpan(span *keyspan.Span) error {
+	if span.Empty() {
+		return nil
+	}
+	if span.Keys[0].Kind() == base.InternalKeyKindRangeDelete {
+		return rangedel.Encode(span, w.Add)
+	}
+	for i := range w.blockPropCollectors {
+		if err := w.blockPropCollectors[i].AddRangeKeys(*span); err != nil {
+			return err
+		}
+	}
+	return w.rangeKeyEncoder.Encode(span)
+}
+
+// Close finishes writing the table and closes the underlying file that the
+// table was written to.
+func (w *RawWriter) Close() (err error) {
+	defer func() {
+		if w.valueBlockWriter != nil {
+			releaseValueBlockWriter(w.valueBlockWriter)
+			// Defensive code in case Close gets called again. We don't want to put
+			// the same object to a sync.Pool.
+			w.valueBlockWriter = nil
+		}
+		w.layout.Abort()
+		// Record any error in the writer (so we can exit early if Close is called
+		// again).
+		if err != nil {
+			w.err = err
+		}
+	}()
+
+	// finish must be called before we check for an error, because finish will
+	// block until every single task added to the writeQueue has been processed,
+	// and an error could be encountered while any of those tasks are processed.
+	if err := w.coordination.writeQueue.finish(); err != nil {
+		return err
+	}
+
+	if w.err != nil {
+		return w.err
+	}
+
+	// The w.meta.LargestPointKey is only used once the Writer is closed, so it is safe to set it
+	// when the Writer is closed.
+	//
+	// The following invariants ensure that setting the largest key at this point of a Writer close
+	// is correct:
+	// 1. Keys must only be added to the Writer in an increasing order.
+	// 2. The current w.dataBlockBuf is guaranteed to have the latest key added to the Writer. This
+	//    must be true, because a w.dataBlockBuf is only switched out when a dataBlock is flushed,
+	//    however, if a dataBlock is flushed, then we add a key to the new w.dataBlockBuf in the
+	//    addPoint function after the flush occurs.
+	if w.dataBlockBuf.dataBlock.EntryCount() >= 1 {
+		w.meta.SetLargestPointKey(w.dataBlockBuf.dataBlock.CurKey().Clone())
+	}
+
+	// Finish the last data block, or force an empty data block if there
+	// aren't any data blocks at all.
+	if w.dataBlockBuf.dataBlock.EntryCount() > 0 || w.indexBlock.block.EntryCount() == 0 {
+		bh, err := w.layout.WriteDataBlock(w.dataBlockBuf.dataBlock.Finish(), &w.dataBlockBuf.blockBuf)
+		if err != nil {
+			return err
+		}
+		bhp, err := w.maybeAddBlockPropertiesToBlockHandle(bh)
+		if err != nil {
+			return err
+		}
+		prevKey := w.dataBlockBuf.dataBlock.CurKey()
+		if err := w.addIndexEntrySync(prevKey, InternalKey{}, bhp, w.dataBlockBuf.tmp[:]); err != nil {
+			return err
+		}
+	}
+	w.props.DataSize = w.layout.offset
+
+	// Write the filter block.
+	if w.filter != nil {
+		bh, err := w.layout.WriteFilterBlock(w.filter)
+		if err != nil {
+			return err
+		}
+		w.props.FilterPolicyName = w.filter.policyName()
+		w.props.FilterSize = bh.Length
+	}
+
+	if w.twoLevelIndex {
+		w.props.IndexType = twoLevelIndex
+		// Write the two level index block.
+		if _, err = w.writeTwoLevelIndex(); err != nil {
+			return err
+		}
+	} else {
+		w.props.IndexType = binarySearchIndex
+		// NB: RocksDB includes the block trailer length in the index size
+		// property, though it doesn't include the trailer in the filter size
+		// property.
+		w.props.IndexSize = uint64(w.indexBlock.estimatedSize()) + block.TrailerLen
+		w.props.NumDataBlocks = uint64(w.indexBlock.block.EntryCount())
+		// Write the single level index block.
+		if _, err = w.layout.WriteIndexBlock(w.indexBlock.finish()); err != nil {
+			return err
+		}
+	}
+
+	// Write the range-del block.
+	if w.props.NumRangeDeletions > 0 {
+		// Because the range tombstones are fragmented, the end key of the last
+		// added range tombstone will be the largest range tombstone key. Note
+		// that we need to make this into a range deletion sentinel because
+		// sstable boundaries are inclusive while the end key of a range
+		// deletion tombstone is exclusive. A Clone() is necessary as
+		// rangeDelBlock.curValue is the same slice that will get passed into
+		// w.writer, and some implementations of vfs.File mutate the slice
+		// passed into Write(). Also, w.meta will often outlive the blockWriter,
+		// and so cloning curValue allows the rangeDelBlock's internal buffer to
+		// get gc'd.
+		k := base.MakeRangeDeleteSentinelKey(w.rangeDelBlock.CurValue()).Clone()
+		w.meta.SetLargestRangeDelKey(k)
+		if _, err := w.layout.WriteRangeDeletionBlock(w.rangeDelBlock.Finish()); err != nil {
+			return err
+		}
+	}
+
+	if w.props.NumRangeKeys() > 0 {
+		key := w.rangeKeyBlock.CurKey()
+		kind := key.Kind()
+		endKey, _, err := rangekey.DecodeEndKey(kind, w.rangeKeyBlock.CurValue())
+		if err != nil {
+			return err
+		}
+		k := base.MakeExclusiveSentinelKey(kind, endKey).Clone()
+		w.meta.SetLargestRangeKey(k)
+		if _, err := w.layout.WriteRangeKeyBlock(w.rangeKeyBlock.Finish()); err != nil {
+			return err
+		}
+	}
+
+	if w.valueBlockWriter != nil {
+		_, vbStats, err := w.valueBlockWriter.finish(&w.layout, w.layout.offset)
+		if err != nil {
+			return err
+		}
+		w.props.NumValueBlocks = vbStats.numValueBlocks
+		w.props.NumValuesInValueBlocks = vbStats.numValuesInValueBlocks
+		w.props.ValueBlocksSize = vbStats.valueBlocksAndIndexSize
+	}
+
+	{
+		// Finish and record the prop collectors if props are not yet recorded.
+		// Pre-computed props might have been copied by specialized sst creators
+		// like suffix replacer.
+		if len(w.props.UserProperties) == 0 {
+			userProps := make(map[string]string)
+			for i := range w.blockPropCollectors {
+				scratch := w.blockPropsEncoder.getScratchForProp()
+				// Place the shortID in the first byte.
+				scratch = append(scratch, byte(i))
+				buf, err := w.blockPropCollectors[i].FinishTable(scratch)
+				if err != nil {
+					return err
+				}
+				var prop string
+				if len(buf) > 0 {
+					prop = string(buf)
+				}
+				// NB: The property is populated in the map even if it is the
+				// empty string, since the presence in the map is what indicates
+				// that the block property collector was used when writing.
+				userProps[w.blockPropCollectors[i].Name()] = prop
+			}
+			if len(userProps) > 0 {
+				w.props.UserProperties = userProps
+			}
+		}
+
+		// Write the properties block.
+		var raw rowblk.Writer
+		// The restart interval is set to infinity because the properties block
+		// is always read sequentially and cached in a heap located object. This
+		// reduces table size without a significant impact on performance.
+		raw.RestartInterval = propertiesBlockRestartInterval
+		w.props.CompressionOptions = rocksDBCompressionOptions
+		w.props.save(w.tableFormat, &raw)
+		w.layout.WritePropertiesBlock(raw.Finish())
+	}
+
+	// Write the table footer.
+	w.meta.Size, err = w.layout.Finish()
+	if err != nil {
+		return err
+	}
+	w.meta.Properties = w.props
+
+	// Check that the features present in the table are compatible with the format
+	// configured for the table.
+	if err = w.assertFormatCompatibility(); err != nil {
+		return err
+	}
+
+	w.dataBlockBuf.clear()
+	dataBlockBufPool.Put(w.dataBlockBuf)
+	w.dataBlockBuf = nil
+	w.indexBlock.clear()
+	indexBlockBufPool.Put(w.indexBlock)
+	w.indexBlock = nil
+
+	// Make any future calls to Set or Close return an error.
+	w.err = errWriterClosed
+	return nil
+}
+
+// EstimatedSize returns the estimated size of the sstable being written if a
+// call to Finish() was made without adding additional keys.
+func (w *RawWriter) EstimatedSize() uint64 {
+	if w == nil {
+		return 0
+	}
+	return w.coordination.sizeEstimate.size() +
+		uint64(w.dataBlockBuf.dataBlock.EstimatedSize()) +
+		w.indexBlock.estimatedSize()
+}
+
+// Metadata returns the metadata for the finished sstable. Only valid to call
+// after the sstable has been finished.
+func (w *RawWriter) Metadata() (*WriterMetadata, error) {
+	if !w.layout.IsFinished() {
+		return nil, errors.New("pebble: writer is not closed")
+	}
+	return &w.meta, nil
+}
+
+// WriterOption provide an interface to do work on Writer while it is being
+// opened.
+type WriterOption interface {
+	// writerApply is called on the writer during opening in order to set
+	// internal parameters.
+	writerApply(*RawWriter)
+}
+
+// NewRawWriter returns a new table writer for the file. Closing the writer will
+// close the file.
+func NewRawWriter(
+	writable objstorage.Writable, o WriterOptions, extraOpts ...WriterOption,
+) *RawWriter {
+	o = o.ensureDefaults()
+	w := &RawWriter{
+		layout: makeLayoutWriter(writable, o),
+		meta: WriterMetadata{
+			SmallestSeqNum: math.MaxUint64,
+		},
+		dataBlockOptions: flushDecisionOptions{
+			blockSize:               o.BlockSize,
+			blockSizeThreshold:      (o.BlockSize*o.BlockSizeThreshold + 99) / 100,
+			sizeClassAwareThreshold: (o.BlockSize*o.SizeClassAwareThreshold + 99) / 100,
+		},
+		indexBlockOptions: flushDecisionOptions{
+			blockSize:               o.IndexBlockSize,
+			blockSizeThreshold:      (o.IndexBlockSize*o.BlockSizeThreshold + 99) / 100,
+			sizeClassAwareThreshold: (o.IndexBlockSize*o.SizeClassAwareThreshold + 99) / 100,
+		},
+		compare:               o.Comparer.Compare,
+		split:                 o.Comparer.Split,
+		formatKey:             o.Comparer.FormatKey,
+		compression:           o.Compression,
+		separator:             o.Comparer.Separator,
+		successor:             o.Comparer.Successor,
+		tableFormat:           o.TableFormat,
+		isStrictObsolete:      o.IsStrictObsolete,
+		writingToLowestLevel:  o.WritingToLowestLevel,
+		restartInterval:       o.BlockRestartInterval,
+		checksumType:          o.Checksum,
+		disableKeyOrderChecks: o.internal.DisableKeyOrderChecks,
+		indexBlock:            newIndexBlockBuf(o.Parallelism),
+		rangeDelBlock:         rowblk.Writer{RestartInterval: 1},
+		rangeKeyBlock:         rowblk.Writer{RestartInterval: 1},
+		topLevelIndexBlock:    rowblk.Writer{RestartInterval: 1},
+		allocatorSizeClasses:  o.AllocatorSizeClasses,
+	}
+	if w.tableFormat >= TableFormatPebblev3 {
+		w.shortAttributeExtractor = o.ShortAttributeExtractor
+		w.requiredInPlaceValueBound = o.RequiredInPlaceValueBound
+		if !o.DisableValueBlocks {
+			w.valueBlockWriter = newValueBlockWriter(
+				w.dataBlockOptions.blockSize, w.dataBlockOptions.blockSizeThreshold, w.compression, w.checksumType, 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.coordination.init(o.Parallelism, w)
+
+	if writable == nil {
+		w.err = errors.New("pebble: nil writable")
+		return w
+	}
+
+	// Note that WriterOptions are applied in two places; the ones with a
+	// preApply() method are applied here. The rest are applied down below after
+	// default properties are set.
+	type preApply interface{ preApply() }
+	for _, opt := range extraOpts {
+		if _, ok := opt.(preApply); ok {
+			opt.writerApply(w)
+		}
+	}
+
+	if o.FilterPolicy != nil {
+		switch o.FilterType {
+		case TableFilter:
+			w.filter = newTableFilterWriter(o.FilterPolicy)
+		default:
+			panic(fmt.Sprintf("unknown filter type: %v", o.FilterType))
+		}
+	}
+
+	w.props.ComparerName = o.Comparer.Name
+	w.props.CompressionName = o.Compression.String()
+	w.props.MergerName = o.MergerName
+	w.props.PropertyCollectorNames = "[]"
+
+	numBlockPropertyCollectors := len(o.BlockPropertyCollectors)
+	if w.tableFormat >= TableFormatPebblev4 {
+		numBlockPropertyCollectors++
+	}
+
+	if numBlockPropertyCollectors > 0 {
+		if numBlockPropertyCollectors > maxPropertyCollectors {
+			w.err = errors.New("pebble: too many block property collectors")
+			return w
+		}
+		w.blockPropCollectors = make([]BlockPropertyCollector, 0, numBlockPropertyCollectors)
+		for _, constructFn := range o.BlockPropertyCollectors {
+			w.blockPropCollectors = append(w.blockPropCollectors, constructFn())
+		}
+		if w.tableFormat >= TableFormatPebblev4 {
+			w.blockPropCollectors = append(w.blockPropCollectors, &w.obsoleteCollector)
+		}
+
+		var buf bytes.Buffer
+		buf.WriteString("[")
+		for i := range w.blockPropCollectors {
+			if i > 0 {
+				buf.WriteString(",")
+			}
+			buf.WriteString(w.blockPropCollectors[i].Name())
+		}
+		buf.WriteString("]")
+		w.props.PropertyCollectorNames = buf.String()
+	}
+
+	// Apply the remaining WriterOptions that do not have a preApply() method.
+	for _, opt := range extraOpts {
+		if _, ok := opt.(preApply); ok {
+			continue
+		}
+		opt.writerApply(w)
+	}
+
+	// Initialize the range key fragmenter and encoder.
+	w.rangeKeyEncoder.Emit = w.addRangeKey
+	return w
+}
+
+// SetSnapshotPinnedProperties sets the properties for pinned keys. Should only
+// be used internally by Pebble.
+func (w *RawWriter) SetSnapshotPinnedProperties(
+	pinnedKeyCount, pinnedKeySize, pinnedValueSize uint64,
+) {
+	w.props.SnapshotPinnedKeys = pinnedKeyCount
+	w.props.SnapshotPinnedKeySize = pinnedKeySize
+	w.props.SnapshotPinnedValueSize = pinnedValueSize
+}
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index 916c1506..53efd4c8 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -93,7 +93,7 @@ func rewriteKeySuffixesInBlocks(
 
 	tableFormat := r.tableFormat
 	o.TableFormat = tableFormat
-	w := NewWriter(out, o)
+	w := NewRawWriter(out, o)
 	defer func() {
 		if w != nil {
 			w.Close()
@@ -256,7 +256,7 @@ func rewriteBlocks(
 	return nil
 }
 
-func checkWriterFilterMatchesReader(r *Reader, w *Writer) error {
+func checkWriterFilterMatchesReader(r *Reader, w *RawWriter) error {
 	if r.Properties.FilterPolicyName != w.filter.policyName() {
 		return errors.New("mismatched filters")
 	}
@@ -268,7 +268,7 @@ func checkWriterFilterMatchesReader(r *Reader, w *Writer) error {
 
 func rewriteDataBlocksToWriter(
 	r *Reader,
-	w *Writer,
+	w *RawWriter,
 	data []BlockHandleWithProperties,
 	from, to []byte,
 	split Split,
@@ -389,7 +389,7 @@ func rewriteDataBlocksToWriter(
 	return nil
 }
 
-func rewriteRangeKeyBlockToWriter(r *Reader, w *Writer, from, to []byte) error {
+func rewriteRangeKeyBlockToWriter(r *Reader, w *RawWriter, from, to []byte) error {
 	iter, err := r.NewRawRangeKeyIter(context.TODO(), NoFragmentTransforms)
 	if err != nil {
 		return err
@@ -451,7 +451,7 @@ func RewriteKeySuffixesViaWriter(
 	}
 
 	o.IsStrictObsolete = false
-	w := NewWriter(out, o)
+	w := NewRawWriter(out, o)
 	defer func() {
 		if w != nil {
 			w.Close()
diff --git a/sstable/write_queue.go b/sstable/write_queue.go
index bb89bda5..70f8b6b1 100644
--- a/sstable/write_queue.go
+++ b/sstable/write_queue.go
@@ -39,7 +39,7 @@ func (task *writeTask) clear() {
 type writeQueue struct {
 	tasks  chan *writeTask
 	wg     sync.WaitGroup
-	writer *Writer
+	writer *RawWriter
 
 	// err represents an error which is encountered when the write queue attempts
 	// to write a block to disk. The error is stored here to skip unnecessary block
@@ -48,7 +48,7 @@ type writeQueue struct {
 	closed bool
 }
 
-func newWriteQueue(size int, writer *Writer) *writeQueue {
+func newWriteQueue(size int, writer *RawWriter) *writeQueue {
 	w := &writeQueue{}
 	w.tasks = make(chan *writeTask, size)
 	w.writer = writer
diff --git a/sstable/writer.go b/sstable/writer.go
index 2fe05444..549daaa3 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -1,653 +1,47 @@
-// Copyright 2011 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
 // of this source code is governed by a BSD-style license that can be found in
 // the LICENSE file.
 
 package sstable
 
 import (
-	"bytes"
-	"encoding/binary"
-	"fmt"
-	"math"
-	"runtime"
-	"slices"
-	"sort"
-	"sync"
-
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
-	"github.com/cockroachdb/pebble/internal/bytealloc"
-	"github.com/cockroachdb/pebble/internal/cache"
-	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
-	"github.com/cockroachdb/pebble/internal/rangedel"
-	"github.com/cockroachdb/pebble/internal/rangekey"
 	"github.com/cockroachdb/pebble/objstorage"
-	"github.com/cockroachdb/pebble/sstable/block"
-	"github.com/cockroachdb/pebble/sstable/rowblk"
 )
 
-// encodedBHPEstimatedSize estimates the size of the encoded BlockHandleWithProperties.
-// It would also be nice to account for the length of the data block properties here,
-// but isn't necessary since this is an estimate.
-const encodedBHPEstimatedSize = binary.MaxVarintLen64 * 2
-
-var errWriterClosed = errors.New("pebble: writer is closed")
-
-// WriterMetadata holds info about a finished sstable.
-type WriterMetadata struct {
-	Size          uint64
-	SmallestPoint InternalKey
-	// LargestPoint, LargestRangeKey, LargestRangeDel should not be accessed
-	// before Writer.Close is called, because they may only be set on
-	// Writer.Close.
-	LargestPoint     InternalKey
-	SmallestRangeDel InternalKey
-	LargestRangeDel  InternalKey
-	SmallestRangeKey InternalKey
-	LargestRangeKey  InternalKey
-	HasPointKeys     bool
-	HasRangeDelKeys  bool
-	HasRangeKeys     bool
-	SmallestSeqNum   base.SeqNum
-	LargestSeqNum    base.SeqNum
-	Properties       Properties
-}
-
-// SetSmallestPointKey sets the smallest point key to the given key.
-// NB: this method set the "absolute" smallest point key. Any existing key is
-// overridden.
-func (m *WriterMetadata) SetSmallestPointKey(k InternalKey) {
-	m.SmallestPoint = k
-	m.HasPointKeys = true
-}
-
-// SetSmallestRangeDelKey sets the smallest rangedel key to the given key.
-// NB: this method set the "absolute" smallest rangedel key. Any existing key is
-// overridden.
-func (m *WriterMetadata) SetSmallestRangeDelKey(k InternalKey) {
-	m.SmallestRangeDel = k
-	m.HasRangeDelKeys = true
-}
-
-// SetSmallestRangeKey sets the smallest range key to the given key.
-// NB: this method set the "absolute" smallest range key. Any existing key is
-// overridden.
-func (m *WriterMetadata) SetSmallestRangeKey(k InternalKey) {
-	m.SmallestRangeKey = k
-	m.HasRangeKeys = true
-}
-
-// SetLargestPointKey sets the largest point key to the given key.
-// NB: this method set the "absolute" largest point key. Any existing key is
-// overridden.
-func (m *WriterMetadata) SetLargestPointKey(k InternalKey) {
-	m.LargestPoint = k
-	m.HasPointKeys = true
-}
-
-// SetLargestRangeDelKey sets the largest rangedel key to the given key.
-// NB: this method set the "absolute" largest rangedel key. Any existing key is
-// overridden.
-func (m *WriterMetadata) SetLargestRangeDelKey(k InternalKey) {
-	m.LargestRangeDel = k
-	m.HasRangeDelKeys = true
-}
-
-// SetLargestRangeKey sets the largest range key to the given key.
-// NB: this method set the "absolute" largest range key. Any existing key is
-// overridden.
-func (m *WriterMetadata) SetLargestRangeKey(k InternalKey) {
-	m.LargestRangeKey = k
-	m.HasRangeKeys = true
-}
-
-func (m *WriterMetadata) updateSeqNum(seqNum base.SeqNum) {
-	if m.SmallestSeqNum > seqNum {
-		m.SmallestSeqNum = seqNum
-	}
-	if m.LargestSeqNum < seqNum {
-		m.LargestSeqNum = seqNum
-	}
-}
-
-// flushDecisionOptions holds parameters to inform the sstable block flushing
-// heuristics.
-type flushDecisionOptions struct {
-	blockSize          int
-	blockSizeThreshold int
-	// sizeClassAwareThreshold takes precedence over blockSizeThreshold when the
-	// Writer is aware of the allocator's size classes.
-	sizeClassAwareThreshold int
-}
-
 // Writer is a table writer.
 type Writer struct {
-	layout layoutWriter
-	meta   WriterMetadata
-	err    error
-	// dataBlockOptions and indexBlockOptions are used to configure the sstable
-	// block flush heuristics.
-	dataBlockOptions  flushDecisionOptions
-	indexBlockOptions flushDecisionOptions
-	// The following fields are copied from Options.
-	compare              Compare
-	split                Split
-	formatKey            base.FormatKey
-	compression          Compression
-	separator            Separator
-	successor            Successor
-	tableFormat          TableFormat
-	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.
-	disableKeyOrderChecks bool
-	// With two level indexes, the index/filter of a SST file is partitioned into
-	// smaller blocks with an additional top-level index on them. When reading an
-	// index/filter, only the top-level index is loaded into memory. The two level
-	// index/filter then uses the top-level index to load on demand into the block
-	// cache the partitions that are required to perform the index/filter query.
-	//
-	// Two level indexes are enabled automatically when there is more than one
-	// index block.
-	//
-	// This is useful when there are very large index blocks, which generally occurs
-	// with the usage of large keys. With large index blocks, the index blocks fight
-	// the data blocks for block cache space and the index blocks are likely to be
-	// re-read many times from the disk. The top level index, which has a much
-	// smaller memory footprint, can be used to prevent the entire index block from
-	// being loaded into the block cache.
-	twoLevelIndex       bool
-	indexBlock          *indexBlockBuf
-	rangeDelBlock       rowblk.Writer
-	rangeKeyBlock       rowblk.Writer
-	topLevelIndexBlock  rowblk.Writer
-	props               Properties
-	blockPropCollectors []BlockPropertyCollector
-	obsoleteCollector   obsoleteKeyBlockPropertyCollector
-	blockPropsEncoder   blockPropertiesEncoder
-	// filter accumulates the filter block. If populated, the filter ingests
-	// either the output of w.split (i.e. a prefix extractor) if w.split is not
-	// nil, or the full keys otherwise.
-	filter          filterWriter
-	indexPartitions []indexBlockAndBlockProperties
-
-	// indexBlockAlloc is used to bulk-allocate byte slices used to store index
-	// blocks in indexPartitions. These live until the index finishes.
-	indexBlockAlloc []byte
-	// indexSepAlloc is used to bulk-allocate index block separator slices stored
-	// in indexPartitions. These live until the index finishes.
-	indexSepAlloc bytealloc.A
-
+	rw *RawWriter
 	// To allow potentially overlapping (i.e. un-fragmented) range keys spans to
 	// be added to the Writer, a keyspan.Fragmenter is used to retain the keys
 	// and values, emitting fragmented, coalesced spans as appropriate. Range
 	// keys must be added in order of their start user-key.
-	fragmenter        keyspan.Fragmenter
-	rangeKeyEncoder   rangekey.Encoder
-	rangeKeysBySuffix keyspan.KeysBySuffix
-	rangeKeySpan      keyspan.Span
-	rkBuf             []byte
-	// dataBlockBuf consists of the state which is currently owned by and used by
-	// the Writer client goroutine. This state can be handed off to other goroutines.
-	dataBlockBuf *dataBlockBuf
-	// blockBuf consists of the state which is owned by and used by the Writer client
-	// goroutine.
-	blockBuf blockBuf
-
-	coordination coordinationState
-
-	// Information (other than the byte slice) about the last point key, to
-	// avoid extracting it again.
-	lastPointKeyInfo pointKeyInfo
-
-	// For value blocks.
-	shortAttributeExtractor   base.ShortAttributeExtractor
-	requiredInPlaceValueBound UserKeyPrefixBound
-	// When w.tableFormat >= TableFormatPebblev3, valueBlockWriter is nil iff
-	// WriterOptions.DisableValueBlocks was true.
-	valueBlockWriter *valueBlockWriter
-
-	allocatorSizeClasses []int
+	fragmenter keyspan.Fragmenter
+	rkBuf      []byte
 }
 
-type pointKeyInfo struct {
-	trailer base.InternalKeyTrailer
-	// Only computed when w.valueBlockWriter is not nil.
-	userKeyLen int
-	// prefixLen uses w.split, if not nil. Only computed when w.valueBlockWriter
-	// is not nil.
-	prefixLen int
-	// True iff the point was marked obsolete.
-	isObsolete bool
-}
-
-type coordinationState struct {
-	parallelismEnabled bool
-
-	// writeQueue is used to write data blocks to disk. The writeQueue is primarily
-	// used to maintain the order in which data blocks must be written to disk. For
-	// this reason, every single data block write must be done through the writeQueue.
-	writeQueue *writeQueue
-
-	sizeEstimate dataBlockEstimates
-}
-
-func (c *coordinationState) init(parallelismEnabled bool, writer *Writer) {
-	c.parallelismEnabled = parallelismEnabled
-	// useMutex is false regardless of parallelismEnabled, because we do not do
-	// parallel compression yet.
-	c.sizeEstimate.useMutex = false
-
-	// writeQueueSize determines the size of the write queue, or the number
-	// of items which can be added to the queue without blocking. By default, we
-	// use a writeQueue size of 0, since we won't be doing any block writes in
-	// parallel.
-	writeQueueSize := 0
-	if parallelismEnabled {
-		writeQueueSize = runtime.GOMAXPROCS(0)
-	}
-	c.writeQueue = newWriteQueue(writeQueueSize, writer)
-}
-
-// sizeEstimate is a general purpose helper for estimating two kinds of sizes:
-// A. The compressed sstable size, which is useful for deciding when to start
-//
-//	a new sstable during flushes or compactions. In practice, we use this in
-//	estimating the data size (excluding the index).
-//
-// B. The size of index blocks to decide when to start a new index block.
-//
-// There are some terminology peculiarities which are due to the origin of
-// sizeEstimate for use case A with parallel compression enabled (for which
-// the code has not been merged). Specifically this relates to the terms
-// "written" and "compressed".
-//   - The notion of "written" for case A is sufficiently defined by saying that
-//     the data block is compressed. Waiting for the actual data block write to
-//     happen can result in unnecessary estimation, when we already know how big
-//     it will be in compressed form. Additionally, with the forthcoming value
-//     blocks containing older MVCC values, these compressed block will be held
-//     in-memory until late in the sstable writing, and we do want to accurately
-//     account for them without waiting for the actual write.
-//     For case B, "written" means that the index entry has been fully
-//     generated, and has been added to the uncompressed block buffer for that
-//     index block. It does not include actually writing a potentially
-//     compressed index block.
-//   - The notion of "compressed" is to differentiate between a "inflight" size
-//     and the actual size, and is handled via computing a compression ratio
-//     observed so far (defaults to 1).
-//     For case A, this is actual data block compression, so the "inflight" size
-//     is uncompressed blocks (that are no longer being written to) and the
-//     "compressed" size is after they have been compressed.
-//     For case B the inflight size is for a key-value pair in the index for
-//     which the value size (the encoded size of the BlockHandleWithProperties)
-//     is not accurately known, while the compressed size is the size of that
-//     entry when it has been added to the (in-progress) index ssblock.
+// NewWriter returns a new table writer intended for building external sstables
+// (eg, for ingestion or storage outside the LSM) for the file. Closing the
+// writer will close the file.
 //
-// Usage: To update state, one can optionally provide an inflight write value
-// using addInflight (used for case B). When something is "written" the state
-// can be updated using either writtenWithDelta or writtenWithTotal, which
-// provide the actual delta size or the total size (latter must be
-// monotonically non-decreasing). If there were no calls to addInflight, there
-// isn't any real estimation happening here. So case A does not do any real
-// estimation. However, when we introduce parallel compression, there will be
-// estimation in that the client goroutine will call addInFlight and the
-// compression goroutines will call writtenWithDelta.
-type sizeEstimate struct {
-	// emptySize is the size when there is no inflight data, and numEntries is 0.
-	// emptySize is constant once set.
-	emptySize uint64
-
-	// inflightSize is the estimated size of some inflight data which hasn't
-	// been written yet.
-	inflightSize uint64
-
-	// totalSize is the total size of the data which has already been written.
-	totalSize uint64
-
-	// numWrittenEntries is the total number of entries which have already been
-	// written.
-	numWrittenEntries uint64
-	// numInflightEntries is the total number of entries which are inflight, and
-	// haven't been written.
-	numInflightEntries uint64
-
-	// maxEstimatedSize stores the maximum result returned from sizeEstimate.size.
-	// It ensures that values returned from subsequent calls to Writer.EstimatedSize
-	// never decrease.
-	maxEstimatedSize uint64
-
-	// We assume that the entries added to the sizeEstimate can be compressed.
-	// For this reason, we keep track of a compressedSize and an uncompressedSize
-	// to compute a compression ratio for the inflight entries. If the entries
-	// aren't being compressed, then compressedSize and uncompressedSize must be
-	// equal.
-	compressedSize   uint64
-	uncompressedSize uint64
-}
-
-func (s *sizeEstimate) init(emptySize uint64) {
-	s.emptySize = emptySize
-}
-
-func (s *sizeEstimate) size() uint64 {
-	ratio := float64(1)
-	if s.uncompressedSize > 0 {
-		ratio = float64(s.compressedSize) / float64(s.uncompressedSize)
-	}
-	estimatedInflightSize := uint64(float64(s.inflightSize) * ratio)
-	total := s.totalSize + estimatedInflightSize
-	if total > s.maxEstimatedSize {
-		s.maxEstimatedSize = total
-	} else {
-		total = s.maxEstimatedSize
-	}
-
-	if total == 0 {
-		return s.emptySize
-	}
-
-	return total
-}
-
-func (s *sizeEstimate) numTotalEntries() uint64 {
-	return s.numWrittenEntries + s.numInflightEntries
-}
-
-func (s *sizeEstimate) addInflight(size int) {
-	s.numInflightEntries++
-	s.inflightSize += uint64(size)
-}
-
-func (s *sizeEstimate) writtenWithTotal(newTotalSize uint64, inflightSize int) {
-	finalEntrySize := int(newTotalSize - s.totalSize)
-	s.writtenWithDelta(finalEntrySize, inflightSize)
-}
-
-func (s *sizeEstimate) writtenWithDelta(finalEntrySize int, inflightSize int) {
-	if inflightSize > 0 {
-		// This entry was previously inflight, so we should decrement inflight
-		// entries and update the "compression" stats for future estimation.
-		s.numInflightEntries--
-		s.inflightSize -= uint64(inflightSize)
-		s.uncompressedSize += uint64(inflightSize)
-		s.compressedSize += uint64(finalEntrySize)
-	}
-	s.numWrittenEntries++
-	s.totalSize += uint64(finalEntrySize)
-}
-
-func (s *sizeEstimate) clear() {
-	*s = sizeEstimate{emptySize: s.emptySize}
-}
-
-type indexBlockBuf struct {
-	// block will only be accessed from the writeQueue.
-	block rowblk.Writer
-
-	size struct {
-		useMutex bool
-		mu       sync.Mutex
-		estimate sizeEstimate
-	}
-
-	// restartInterval matches indexBlockBuf.block.restartInterval. We store it twice, because the `block`
-	// must only be accessed from the writeQueue goroutine.
-	restartInterval int
-}
-
-func (i *indexBlockBuf) clear() {
-	i.block.Reset()
-	if i.size.useMutex {
-		i.size.mu.Lock()
-		defer i.size.mu.Unlock()
-	}
-	i.size.estimate.clear()
-	i.restartInterval = 0
-}
-
-var indexBlockBufPool = sync.Pool{
-	New: func() interface{} {
-		return &indexBlockBuf{}
-	},
-}
-
-const indexBlockRestartInterval = 1
-
-func newIndexBlockBuf(useMutex bool) *indexBlockBuf {
-	i := indexBlockBufPool.Get().(*indexBlockBuf)
-	i.size.useMutex = useMutex
-	i.restartInterval = indexBlockRestartInterval
-	i.block.RestartInterval = indexBlockRestartInterval
-	i.size.estimate.init(rowblk.EmptySize)
-	return i
-}
-
-func (i *indexBlockBuf) shouldFlush(
-	sep InternalKey, valueLen int, flushOptions flushDecisionOptions, sizeClassHints []int,
-) bool {
-	if i.size.useMutex {
-		i.size.mu.Lock()
-		defer i.size.mu.Unlock()
-	}
-
-	nEntries := i.size.estimate.numTotalEntries()
-	return shouldFlushWithHints(
-		sep.Size(), valueLen, i.restartInterval, int(i.size.estimate.size()),
-		int(nEntries), flushOptions, sizeClassHints)
-}
-
-func (i *indexBlockBuf) add(key InternalKey, value []byte, inflightSize int) {
-	i.block.Add(key, value)
-	size := i.block.EstimatedSize()
-	if i.size.useMutex {
-		i.size.mu.Lock()
-		defer i.size.mu.Unlock()
-	}
-	i.size.estimate.writtenWithTotal(uint64(size), inflightSize)
-}
-
-func (i *indexBlockBuf) finish() []byte {
-	b := i.block.Finish()
-	return b
-}
-
-func (i *indexBlockBuf) addInflight(inflightSize int) {
-	if i.size.useMutex {
-		i.size.mu.Lock()
-		defer i.size.mu.Unlock()
-	}
-	i.size.estimate.addInflight(inflightSize)
-}
-
-func (i *indexBlockBuf) estimatedSize() uint64 {
-	if i.size.useMutex {
-		i.size.mu.Lock()
-		defer i.size.mu.Unlock()
-	}
-
-	// Make sure that the size estimation works as expected when parallelism
-	// is disabled.
-	if invariants.Enabled && !i.size.useMutex {
-		if i.size.estimate.inflightSize != 0 {
-			panic("unexpected inflight entry in index block size estimation")
-		}
-
-		// NB: The i.block should only be accessed from the writeQueue goroutine,
-		// when parallelism is enabled. We break that invariant here, but that's
-		// okay since parallelism is disabled.
-		if i.size.estimate.size() != uint64(i.block.EstimatedSize()) {
-			panic("index block size estimation sans parallelism is incorrect")
-		}
-	}
-	return i.size.estimate.size()
-}
-
-// sizeEstimate is used for sstable size estimation. sizeEstimate can be
-// accessed by the Writer client and compressionQueue goroutines. Fields
-// should only be read/updated through the functions defined on the
-// *sizeEstimate type.
-type dataBlockEstimates struct {
-	// If we don't do block compression in parallel, then we don't need to take
-	// the performance hit of synchronizing using this mutex.
-	useMutex bool
-	mu       sync.Mutex
-
-	estimate sizeEstimate
-}
-
-// inflightSize is the uncompressed block size estimate which has been
-// previously provided to addInflightDataBlock(). If addInflightDataBlock()
-// has not been called, this must be set to 0. compressedSize is the
-// compressed size of the block.
-func (d *dataBlockEstimates) dataBlockCompressed(compressedSize int, inflightSize int) {
-	if d.useMutex {
-		d.mu.Lock()
-		defer d.mu.Unlock()
-	}
-	d.estimate.writtenWithDelta(compressedSize+block.TrailerLen, inflightSize)
-}
-
-// size is an estimated size of datablock data which has been written to disk.
-func (d *dataBlockEstimates) size() uint64 {
-	if d.useMutex {
-		d.mu.Lock()
-		defer d.mu.Unlock()
-	}
-	// If there is no parallel compression, there should not be any inflight bytes.
-	if invariants.Enabled && !d.useMutex {
-		if d.estimate.inflightSize != 0 {
-			panic("unexpected inflight entry in data block size estimation")
-		}
-	}
-	return d.estimate.size()
-}
-
-// Avoid linter unused error.
-var _ = (&dataBlockEstimates{}).addInflightDataBlock
-
-// NB: unused since no parallel compression.
-func (d *dataBlockEstimates) addInflightDataBlock(size int) {
-	if d.useMutex {
-		d.mu.Lock()
-		defer d.mu.Unlock()
-	}
-
-	d.estimate.addInflight(size)
-}
-
-var writeTaskPool = sync.Pool{
-	New: func() interface{} {
-		t := &writeTask{}
-		t.compressionDone = make(chan bool, 1)
-		return t
-	},
-}
-
-type blockBuf struct {
-	// tmp is a scratch buffer, large enough to hold either footerLen bytes,
-	// blockTrailerLen bytes, (5 * binary.MaxVarintLen64) bytes, and most
-	// likely large enough for a block handle with properties.
-	tmp [blockHandleLikelyMaxLen]byte
-	// compressedBuf is the destination buffer for compression. It is re-used over the
-	// lifetime of the blockBuf, avoiding the allocation of a temporary buffer for each block.
-	compressedBuf []byte
-	checksummer   block.Checksummer
-}
-
-func (b *blockBuf) clear() {
-	// We can't assign b.compressedBuf[:0] to compressedBuf because snappy relies
-	// on the length of the buffer, and not the capacity to determine if it needs
-	// to make an allocation.
-	*b = blockBuf{
-		compressedBuf: b.compressedBuf, checksummer: b.checksummer,
+// Internal clients should generally prefer NewRawWriter.
+func NewWriter(writable objstorage.Writable, o WriterOptions, extraOpts ...WriterOption) *Writer {
+	o = o.ensureDefaults()
+	rw := NewRawWriter(writable, o, extraOpts...)
+	return &Writer{
+		rw: rw,
+		fragmenter: keyspan.Fragmenter{
+			Cmp:    o.Comparer.Compare,
+			Format: o.Comparer.FormatKey,
+			Emit:   rw.encodeFragmentedRangeKeySpan,
+		},
 	}
 }
 
-// A dataBlockBuf holds all the state required to compress and write a data block to disk.
-// A dataBlockBuf begins its lifecycle owned by the Writer client goroutine. The Writer
-// client goroutine adds keys to the sstable, writing directly into a dataBlockBuf's blockWriter
-// until the block is full. Once a dataBlockBuf's block is full, the dataBlockBuf may be passed
-// to other goroutines for compression and file I/O.
-type dataBlockBuf struct {
-	blockBuf
-	dataBlock rowblk.Writer
-
-	// uncompressed is a reference to a byte slice which is owned by the dataBlockBuf. It is the
-	// next byte slice to be compressed. The uncompressed byte slice will be backed by the
-	// dataBlock.buf.
-	uncompressed []byte
-	// compressed is a reference to a byte slice which is owned by the dataBlockBuf. It is the
-	// compressed byte slice which must be written to disk. The compressed byte slice may be
-	// backed by the dataBlock.buf, or the dataBlockBuf.compressedBuf, depending on whether
-	// we use the result of the compression.
-	compressed []byte
-	// trailer is the block trailer encoding the compression type and checksum.
-	trailer block.Trailer
-
-	// We're making calls to BlockPropertyCollectors from the Writer client goroutine. We need to
-	// pass the encoded block properties over to the write queue. To prevent copies, and allocations,
-	// we give each dataBlockBuf, a blockPropertiesEncoder.
-	blockPropsEncoder blockPropertiesEncoder
-	// dataBlockProps is set when Writer.finishDataBlockProps is called. The dataBlockProps slice is
-	// a shallow copy of the internal buffer of the dataBlockBuf.blockPropsEncoder.
-	dataBlockProps []byte
-
-	// sepScratch is reusable scratch space for computing separator keys.
-	sepScratch []byte
-}
-
-func (d *dataBlockBuf) clear() {
-	d.blockBuf.clear()
-	d.dataBlock.Reset()
-
-	d.uncompressed = nil
-	d.compressed = nil
-	d.dataBlockProps = nil
-	d.sepScratch = d.sepScratch[:0]
-}
-
-var dataBlockBufPool = sync.Pool{
-	New: func() interface{} {
-		return &dataBlockBuf{}
-	},
-}
-
-func newDataBlockBuf(restartInterval int, checksumType block.ChecksumType) *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(c Compression) {
-	d.compressed, d.trailer = compressAndChecksum(d.uncompressed, c, &d.blockBuf)
-}
-
-func (d *dataBlockBuf) shouldFlush(
-	key InternalKey, valueLen int, flushOptions flushDecisionOptions, sizeClassHints []int,
-) bool {
-	return shouldFlushWithHints(
-		key.Size(), valueLen, d.dataBlock.RestartInterval, d.dataBlock.EstimatedSize(),
-		d.dataBlock.EntryCount(), flushOptions, sizeClassHints)
-}
-
-type indexBlockAndBlockProperties struct {
-	nEntries int
-	// sep is the last key added to this block, for computing a separator later.
-	sep        InternalKey
-	properties []byte
-	// block is the encoded block produced by blockWriter.finish.
-	block []byte
-}
+// Raw returns the underlying RawWriter.
+func (w *Writer) Raw() *RawWriter { return w.rw }
 
 // Set sets the value for the given key. The sequence number is set to 0.
 // Intended for use to externally construct an sstable before ingestion into a
@@ -656,15 +50,15 @@ type indexBlockAndBlockProperties struct {
 //
 // TODO(peter): untested
 func (w *Writer) Set(key, value []byte) error {
-	if w.err != nil {
-		return w.err
+	if w.rw.err != nil {
+		return w.rw.err
 	}
-	if w.isStrictObsolete {
+	if w.rw.isStrictObsolete {
 		return errors.Errorf("use AddWithForceObsolete")
 	}
 	// forceObsolete is false based on the assumption that no RANGEDELs in the
 	// sstable delete the added points.
-	return w.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindSet), value, false)
+	return w.rw.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindSet), value, false)
 }
 
 // Delete deletes the value for the given key. The sequence number is set to
@@ -673,15 +67,15 @@ func (w *Writer) Set(key, value []byte) error {
 //
 // TODO(peter): untested
 func (w *Writer) Delete(key []byte) error {
-	if w.err != nil {
-		return w.err
+	if w.rw.err != nil {
+		return w.rw.err
 	}
-	if w.isStrictObsolete {
+	if w.rw.isStrictObsolete {
 		return errors.Errorf("use AddWithForceObsolete")
 	}
 	// forceObsolete is false based on the assumption that no RANGEDELs in the
 	// sstable delete the added points.
-	return w.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindDelete), nil, false)
+	return w.rw.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindDelete), nil, false)
 }
 
 // DeleteRange deletes all of the keys (and values) in the range [start,end)
@@ -691,10 +85,10 @@ func (w *Writer) Delete(key []byte) error {
 //
 // TODO(peter): untested
 func (w *Writer) DeleteRange(start, end []byte) error {
-	if w.err != nil {
-		return w.err
+	if w.rw.err != nil {
+		return w.rw.err
 	}
-	return w.addTombstone(base.MakeInternalKey(start, 0, InternalKeyKindRangeDelete), end)
+	return w.rw.addTombstone(base.MakeInternalKey(start, 0, InternalKeyKindRangeDelete), end)
 }
 
 // Merge adds an action to the DB that merges the value at key with the new
@@ -704,393 +98,16 @@ func (w *Writer) DeleteRange(start, end []byte) error {
 //
 // TODO(peter): untested
 func (w *Writer) Merge(key, value []byte) error {
-	if w.err != nil {
-		return w.err
+	if w.rw.err != nil {
+		return w.rw.err
 	}
-	if w.isStrictObsolete {
+	if w.rw.isStrictObsolete {
 		return errors.Errorf("use AddWithForceObsolete")
 	}
 	// forceObsolete is false based on the assumption that no RANGEDELs in the
 	// sstable that delete the added points. If the user configured this writer
 	// to be strict-obsolete, addPoint will reject the addition of this MERGE.
-	return w.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindMerge), value, false)
-}
-
-// Add adds a key/value pair to the table being written. For a given Writer,
-// the keys passed to Add must be in increasing order. The exception to this
-// rule is range deletion tombstones. Range deletion tombstones need to be
-// added ordered by their start key, but they can be added out of order from
-// point entries. Additionally, range deletion tombstones must be fragmented
-// (i.e. by keyspan.Fragmenter).
-func (w *Writer) Add(key InternalKey, value []byte) error {
-	if w.isStrictObsolete {
-		return errors.Errorf("use AddWithForceObsolete")
-	}
-	return w.AddWithForceObsolete(key, value, false)
-}
-
-// AddWithForceObsolete must be used when writing a strict-obsolete sstable.
-//
-// forceObsolete indicates whether the caller has determined that this key is
-// obsolete even though it may be the latest point key for this userkey. This
-// should be set to true for keys obsoleted by RANGEDELs, and is required for
-// strict-obsolete sstables.
-//
-// Note that there are two properties, S1 and S2 (see comment in format.go)
-// that strict-obsolete ssts must satisfy. S2, due to RANGEDELs, is solely the
-// responsibility of the caller. S1 is solely the responsibility of the
-// callee.
-func (w *Writer) AddWithForceObsolete(key InternalKey, value []byte, forceObsolete bool) error {
-	if w.err != nil {
-		return w.err
-	}
-
-	switch key.Kind() {
-	case InternalKeyKindRangeDelete:
-		return w.addTombstone(key, value)
-	case base.InternalKeyKindRangeKeyDelete,
-		base.InternalKeyKindRangeKeySet,
-		base.InternalKeyKindRangeKeyUnset:
-		w.err = errors.Errorf(
-			"pebble: range keys must be added via one of the RangeKey* functions")
-		return w.err
-	}
-	return w.addPoint(key, value, forceObsolete)
-}
-
-func (w *Writer) makeAddPointDecisionV2(key InternalKey) error {
-	prevTrailer := w.lastPointKeyInfo.trailer
-	w.lastPointKeyInfo.trailer = key.Trailer
-	if w.dataBlockBuf.dataBlock.EntryCount() == 0 {
-		return nil
-	}
-	if !w.disableKeyOrderChecks {
-		prevPointUserKey := w.dataBlockBuf.dataBlock.CurUserKey()
-		cmpUser := w.compare(prevPointUserKey, key.UserKey)
-		if cmpUser > 0 || (cmpUser == 0 && prevTrailer <= key.Trailer) {
-			return errors.Errorf(
-				"pebble: keys must be added in strictly increasing order: %s, %s",
-				InternalKey{UserKey: prevPointUserKey, Trailer: prevTrailer}.Pretty(w.formatKey),
-				key.Pretty(w.formatKey))
-		}
-	}
-	return nil
-}
-
-// REQUIRES: at least one point has been written to the Writer.
-func (w *Writer) getLastPointUserKey() []byte {
-	if w.dataBlockBuf.dataBlock.EntryCount() == 0 {
-		panic(errors.AssertionFailedf("no point keys added to writer"))
-	}
-	return w.dataBlockBuf.dataBlock.CurUserKey()
-}
-
-// REQUIRES: w.tableFormat >= TableFormatPebblev3
-func (w *Writer) makeAddPointDecisionV3(
-	key InternalKey, valueLen int,
-) (setHasSamePrefix bool, writeToValueBlock bool, isObsolete bool, err error) {
-	prevPointKeyInfo := w.lastPointKeyInfo
-	w.lastPointKeyInfo.userKeyLen = len(key.UserKey)
-	w.lastPointKeyInfo.prefixLen = w.split(key.UserKey)
-	w.lastPointKeyInfo.trailer = key.Trailer
-	w.lastPointKeyInfo.isObsolete = false
-	if !w.meta.HasPointKeys {
-		return false, false, false, nil
-	}
-	keyKind := key.Trailer.Kind()
-	prevPointUserKey := w.getLastPointUserKey()
-	prevPointKey := InternalKey{UserKey: prevPointUserKey, Trailer: prevPointKeyInfo.trailer}
-	prevKeyKind := prevPointKeyInfo.trailer.Kind()
-	considerWriteToValueBlock := prevKeyKind == InternalKeyKindSet &&
-		keyKind == InternalKeyKindSet
-	if considerWriteToValueBlock && !w.requiredInPlaceValueBound.IsEmpty() {
-		keyPrefix := key.UserKey[:w.lastPointKeyInfo.prefixLen]
-		cmpUpper := w.compare(
-			w.requiredInPlaceValueBound.Upper, keyPrefix)
-		if cmpUpper <= 0 {
-			// Common case for CockroachDB. Make it empty since all future keys in
-			// this sstable will also have cmpUpper <= 0.
-			w.requiredInPlaceValueBound = UserKeyPrefixBound{}
-		} else if w.compare(keyPrefix, w.requiredInPlaceValueBound.Lower) >= 0 {
-			considerWriteToValueBlock = false
-		}
-	}
-	// cmpPrefix is initialized iff considerWriteToValueBlock.
-	var cmpPrefix int
-	var cmpUser int
-	if considerWriteToValueBlock {
-		// Compare the prefixes.
-		cmpPrefix = w.compare(prevPointUserKey[:prevPointKeyInfo.prefixLen],
-			key.UserKey[:w.lastPointKeyInfo.prefixLen])
-		cmpUser = cmpPrefix
-		if cmpPrefix == 0 {
-			// Need to compare suffixes to compute cmpUser.
-			cmpUser = w.compare(prevPointUserKey[prevPointKeyInfo.prefixLen:],
-				key.UserKey[w.lastPointKeyInfo.prefixLen:])
-		}
-	} else {
-		cmpUser = w.compare(prevPointUserKey, key.UserKey)
-	}
-	// Ensure that no one adds a point key kind without considering the obsolete
-	// handling for that kind.
-	switch keyKind {
-	case InternalKeyKindSet, InternalKeyKindSetWithDelete, InternalKeyKindMerge,
-		InternalKeyKindDelete, InternalKeyKindSingleDelete, InternalKeyKindDeleteSized:
-	default:
-		panic(errors.AssertionFailedf("unexpected key kind %s", keyKind.String()))
-	}
-	// If same user key, then the current key is obsolete if any of the
-	// following is true:
-	// C1 The prev key was obsolete.
-	// C2 The prev key was not a MERGE. When the previous key is a MERGE we must
-	//    preserve SET* and MERGE since their values will be merged into the
-	//    previous key. We also must preserve DEL* since there may be an older
-	//    SET*/MERGE in a lower level that must not be merged with the MERGE --
-	//    if we omit the DEL* that lower SET*/MERGE will become visible.
-	//
-	// Regardless of whether it is the same user key or not
-	// C3 The current key is some kind of point delete, and we are writing to
-	//    the lowest level, then it is also obsolete. The correctness of this
-	//    relies on the same user key not spanning multiple sstables in a level.
-	//
-	// C1 ensures that for a user key there is at most one transition from
-	// !obsolete to obsolete. Consider a user key k, for which the first n keys
-	// are not obsolete. We consider the various value of n:
-	//
-	// n = 0: This happens due to forceObsolete being set by the caller, or due
-	// to C3. forceObsolete must only be set due a RANGEDEL, and that RANGEDEL
-	// must also delete all the lower seqnums for the same user key. C3 triggers
-	// due to a point delete and that deletes all the lower seqnums for the same
-	// user key.
-	//
-	// n = 1: This is the common case. It happens when the first key is not a
-	// MERGE, or the current key is some kind of point delete.
-	//
-	// n > 1: This is due to a sequence of MERGE keys, potentially followed by a
-	// single non-MERGE key.
-	isObsoleteC1AndC2 := cmpUser == 0 &&
-		(prevPointKeyInfo.isObsolete || prevKeyKind != InternalKeyKindMerge)
-	isObsoleteC3 := w.writingToLowestLevel &&
-		(keyKind == InternalKeyKindDelete || keyKind == InternalKeyKindSingleDelete ||
-			keyKind == InternalKeyKindDeleteSized)
-	isObsolete = isObsoleteC1AndC2 || isObsoleteC3
-	// TODO(sumeer): storing isObsolete SET and SETWITHDEL in value blocks is
-	// possible, but requires some care in documenting and checking invariants.
-	// There is code that assumes nothing in value blocks because of single MVCC
-	// version (those should be ok). We have to ensure setHasSamePrefix is
-	// correctly initialized here etc.
-
-	if !w.disableKeyOrderChecks &&
-		(cmpUser > 0 || (cmpUser == 0 && prevPointKeyInfo.trailer <= key.Trailer)) {
-		return false, false, false, errors.Errorf(
-			"pebble: keys must be added in strictly increasing order: %s, %s",
-			prevPointKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
-	}
-	if !considerWriteToValueBlock {
-		return false, false, isObsolete, nil
-	}
-	// NB: it is possible that cmpUser == 0, i.e., these two SETs have identical
-	// user keys (because of an open snapshot). This should be the rare case.
-	setHasSamePrefix = cmpPrefix == 0
-	// Use of 0 here is somewhat arbitrary. Given the minimum 3 byte encoding of
-	// valueHandle, this should be > 3. But tiny values are common in test and
-	// unlikely in production, so we use 0 here for better test coverage.
-	const tinyValueThreshold = 0
-	// NB: setting WriterOptions.DisableValueBlocks does not disable the
-	// setHasSamePrefix optimization.
-	considerWriteToValueBlock = setHasSamePrefix && valueLen > tinyValueThreshold && w.valueBlockWriter != nil
-	return setHasSamePrefix, considerWriteToValueBlock, isObsolete, nil
-}
-
-func (w *Writer) addPoint(key InternalKey, value []byte, forceObsolete bool) error {
-	if w.isStrictObsolete && key.Kind() == InternalKeyKindMerge {
-		return errors.Errorf("MERGE not supported in a strict-obsolete sstable")
-	}
-	var err error
-	var setHasSameKeyPrefix, writeToValueBlock, addPrefixToValueStoredWithKey bool
-	var isObsolete bool
-	maxSharedKeyLen := len(key.UserKey)
-	if w.tableFormat >= TableFormatPebblev3 {
-		// maxSharedKeyLen is limited to the prefix of the preceding key. If the
-		// preceding key was in a different block, then the blockWriter will
-		// ignore this maxSharedKeyLen.
-		maxSharedKeyLen = w.lastPointKeyInfo.prefixLen
-		setHasSameKeyPrefix, writeToValueBlock, isObsolete, err =
-			w.makeAddPointDecisionV3(key, len(value))
-		addPrefixToValueStoredWithKey = key.Kind() == InternalKeyKindSet
-	} else {
-		err = w.makeAddPointDecisionV2(key)
-	}
-	if err != nil {
-		return err
-	}
-	isObsolete = w.tableFormat >= TableFormatPebblev4 && (isObsolete || forceObsolete)
-	w.lastPointKeyInfo.isObsolete = isObsolete
-	var valueStoredWithKey []byte
-	var prefix block.ValuePrefix
-	var valueStoredWithKeyLen int
-	if writeToValueBlock {
-		vh, err := w.valueBlockWriter.addValue(value)
-		if err != nil {
-			return err
-		}
-		n := encodeValueHandle(w.blockBuf.tmp[:], vh)
-		valueStoredWithKey = w.blockBuf.tmp[:n]
-		valueStoredWithKeyLen = len(valueStoredWithKey) + 1
-		var attribute base.ShortAttribute
-		if w.shortAttributeExtractor != nil {
-			// TODO(sumeer): for compactions, it is possible that the input sstable
-			// already has this value in the value section and so we have already
-			// extracted the ShortAttribute. Avoid extracting it again. This will
-			// require changing the Writer.Add interface.
-			if attribute, err = w.shortAttributeExtractor(
-				key.UserKey, w.lastPointKeyInfo.prefixLen, value); err != nil {
-				return err
-			}
-		}
-		prefix = block.ValueHandlePrefix(setHasSameKeyPrefix, attribute)
-	} else {
-		valueStoredWithKey = value
-		valueStoredWithKeyLen = len(value)
-		if addPrefixToValueStoredWithKey {
-			valueStoredWithKeyLen++
-		}
-		prefix = block.InPlaceValuePrefix(setHasSameKeyPrefix)
-	}
-
-	if err := w.maybeFlush(key, valueStoredWithKeyLen); err != nil {
-		return err
-	}
-
-	for i := range w.blockPropCollectors {
-		v := value
-		if addPrefixToValueStoredWithKey {
-			// Values for SET are not required to be in-place, and in the future may
-			// not even be read by the compaction, so pass nil values. Block
-			// property collectors in such Pebble DB's must not look at the value.
-			v = nil
-		}
-		if err := w.blockPropCollectors[i].AddPointKey(key, v); err != nil {
-			w.err = err
-			return err
-		}
-	}
-	if w.tableFormat >= TableFormatPebblev4 {
-		w.obsoleteCollector.AddPoint(isObsolete)
-	}
-
-	w.maybeAddToFilter(key.UserKey)
-	w.dataBlockBuf.dataBlock.AddWithOptionalValuePrefix(
-		key, isObsolete, valueStoredWithKey, maxSharedKeyLen, addPrefixToValueStoredWithKey, prefix,
-		setHasSameKeyPrefix)
-
-	w.meta.updateSeqNum(key.SeqNum())
-
-	if !w.meta.HasPointKeys {
-		k := w.dataBlockBuf.dataBlock.CurKey()
-		// NB: We need to ensure that SmallestPoint.UserKey is set, so we create
-		// an InternalKey which is semantically identical to the key, but won't
-		// have a nil UserKey. We do this, because key.UserKey could be nil, and
-		// we don't want SmallestPoint.UserKey to be nil.
-		//
-		// todo(bananabrick): Determine if it's okay to have a nil SmallestPoint
-		// .UserKey now that we don't rely on a nil UserKey to determine if the
-		// key has been set or not.
-		w.meta.SetSmallestPointKey(k.Clone())
-	}
-
-	w.props.NumEntries++
-	switch key.Kind() {
-	case InternalKeyKindDelete, InternalKeyKindSingleDelete:
-		w.props.NumDeletions++
-		w.props.RawPointTombstoneKeySize += uint64(len(key.UserKey))
-	case InternalKeyKindDeleteSized:
-		var size uint64
-		if len(value) > 0 {
-			var n int
-			size, n = binary.Uvarint(value)
-			if n <= 0 {
-				w.err = errors.Newf("%s key's value (%x) does not parse as uvarint",
-					errors.Safe(key.Kind().String()), value)
-				return w.err
-			}
-		}
-		w.props.NumDeletions++
-		w.props.NumSizedDeletions++
-		w.props.RawPointTombstoneKeySize += uint64(len(key.UserKey))
-		w.props.RawPointTombstoneValueSize += size
-	case InternalKeyKindMerge:
-		w.props.NumMergeOperands++
-	}
-	w.props.RawKeySize += uint64(key.Size())
-	w.props.RawValueSize += uint64(len(value))
-	return nil
-}
-
-func (w *Writer) prettyTombstone(k InternalKey, value []byte) fmt.Formatter {
-	return keyspan.Span{
-		Start: k.UserKey,
-		End:   value,
-		Keys:  []keyspan.Key{{Trailer: k.Trailer}},
-	}.Pretty(w.formatKey)
-}
-
-func (w *Writer) addTombstone(key InternalKey, value []byte) error {
-	if !w.disableKeyOrderChecks && w.rangeDelBlock.EntryCount() > 0 {
-		// Check that tombstones are being added in fragmented order. If the two
-		// tombstones overlap, their start and end keys must be identical.
-		prevKey := w.rangeDelBlock.CurKey()
-		switch c := w.compare(prevKey.UserKey, key.UserKey); {
-		case c > 0:
-			w.err = errors.Errorf("pebble: keys must be added in order: %s, %s",
-				prevKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
-			return w.err
-		case c == 0:
-			prevValue := w.rangeDelBlock.CurValue()
-			if w.compare(prevValue, value) != 0 {
-				w.err = errors.Errorf("pebble: overlapping tombstones must be fragmented: %s vs %s",
-					w.prettyTombstone(prevKey, prevValue),
-					w.prettyTombstone(key, value))
-				return w.err
-			}
-			if prevKey.SeqNum() <= key.SeqNum() {
-				w.err = errors.Errorf("pebble: keys must be added in strictly increasing order: %s, %s",
-					prevKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
-				return w.err
-			}
-		default:
-			prevValue := w.rangeDelBlock.CurValue()
-			if w.compare(prevValue, key.UserKey) > 0 {
-				w.err = errors.Errorf("pebble: overlapping tombstones must be fragmented: %s vs %s",
-					w.prettyTombstone(prevKey, prevValue),
-					w.prettyTombstone(key, value))
-				return w.err
-			}
-		}
-	}
-
-	if key.Trailer == base.InternalKeyRangeDeleteSentinel {
-		w.err = errors.Errorf("pebble: cannot add range delete sentinel: %s", key.Pretty(w.formatKey))
-		return w.err
-	}
-
-	w.meta.updateSeqNum(key.SeqNum())
-
-	// Range tombstones are fragmented in the v2 range deletion block format,
-	// so the start key of the first range tombstone added will be the smallest
-	// range tombstone key. The largest range tombstone key will be determined
-	// in Writer.Close() as the end key of the last range tombstone added.
-	if w.props.NumRangeDeletions == 0 {
-		w.meta.SetSmallestRangeDelKey(key.Clone())
-	}
-
-	w.props.NumEntries++
-	w.props.NumDeletions++
-	w.props.NumRangeDeletions++
-	w.props.RawKeySize += uint64(key.Size())
-	w.props.RawValueSize += uint64(len(value))
-	w.rangeDelBlock.Add(key, value)
-	return nil
+	return w.rw.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindMerge), value, false)
 }
 
 // RangeKeySet sets a range between start (inclusive) and end (exclusive) with
@@ -1155,130 +172,18 @@ func (w *Writer) RangeKeyDelete(start, end []byte) error {
 }
 
 func (w *Writer) addRangeKeySpanToFragmenter(span keyspan.Span) error {
-	if w.compare(span.Start, span.End) >= 0 {
+	if w.rw.compare(span.Start, span.End) >= 0 {
 		return errors.Errorf(
 			"pebble: start key must be strictly less than end key",
 		)
 	}
-	if w.fragmenter.Start() != nil && w.compare(w.fragmenter.Start(), span.Start) > 0 {
+	if w.fragmenter.Start() != nil && w.rw.compare(w.fragmenter.Start(), span.Start) > 0 {
 		return errors.Errorf("pebble: spans must be added in order: %s > %s",
-			w.formatKey(w.fragmenter.Start()), w.formatKey(span.Start))
+			w.rw.formatKey(w.fragmenter.Start()), w.rw.formatKey(span.Start))
 	}
 	// Add this span to the fragmenter.
 	w.fragmenter.Add(span)
-	return w.err
-}
-
-func (w *Writer) encodeFragmentedRangeKeySpan(span keyspan.Span) {
-	// This method is the emit function of the Fragmenter.
-	//
-	// NB: The span should only contain range keys and be internally consistent
-	// (eg, no duplicate suffixes, no additional keys after a RANGEKEYDEL).
-	//
-	// We use w.rangeKeysBySuffix and w.rangeKeySpan to avoid allocations.
-
-	// Sort the keys by suffix. Iteration doesn't *currently* depend on it, but
-	// we may want to in the future.
-	w.rangeKeysBySuffix.Cmp = w.compare
-	w.rangeKeysBySuffix.Keys = span.Keys
-	sort.Sort(&w.rangeKeysBySuffix)
-
-	w.rangeKeySpan = span
-	w.rangeKeySpan.Keys = w.rangeKeysBySuffix.Keys
-	if w.err == nil {
-		w.err = w.EncodeSpan(&w.rangeKeySpan)
-	}
-}
-
-// addRangeKey adds a range key set, unset, or delete key/value pair to the
-// table being written.
-//
-// Range keys must be supplied in strictly ascending order of start key (i.e.
-// user key ascending, sequence number descending, and key type descending).
-// Ranges added must also be supplied in fragmented span order - i.e. other than
-// spans that are perfectly aligned (same start and end keys), spans may not
-// overlap. Range keys may be added out of order relative to point keys and
-// range deletions.
-func (w *Writer) addRangeKey(key InternalKey, value []byte) error {
-	if !w.disableKeyOrderChecks && w.rangeKeyBlock.EntryCount() > 0 {
-		prevStartKey := w.rangeKeyBlock.CurKey()
-		prevEndKey, _, err := rangekey.DecodeEndKey(prevStartKey.Kind(), w.rangeKeyBlock.CurValue())
-		if err != nil {
-			// We panic here as we should have previously decoded and validated this
-			// key and value when it was first added to the range key block.
-			panic(err)
-		}
-
-		curStartKey := key
-		curEndKey, _, err := rangekey.DecodeEndKey(curStartKey.Kind(), value)
-		if err != nil {
-			w.err = err
-			return w.err
-		}
-
-		// Start keys must be strictly increasing.
-		if base.InternalCompare(w.compare, prevStartKey, curStartKey) >= 0 {
-			w.err = errors.Errorf(
-				"pebble: range keys starts must be added in increasing order: %s, %s",
-				prevStartKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
-			return w.err
-		}
-
-		// Start keys are increasing. If the start user keys are equal, the
-		// end keys must be equal (i.e. aligned spans).
-		if w.compare(prevStartKey.UserKey, curStartKey.UserKey) == 0 {
-			if w.compare(prevEndKey, curEndKey) != 0 {
-				w.err = errors.Errorf("pebble: overlapping range keys must be fragmented: %s, %s",
-					prevStartKey.Pretty(w.formatKey),
-					curStartKey.Pretty(w.formatKey))
-				return w.err
-			}
-		} else if w.compare(prevEndKey, curStartKey.UserKey) > 0 {
-			// If the start user keys are NOT equal, the spans must be disjoint (i.e.
-			// no overlap).
-			// NOTE: the inequality excludes zero, as we allow the end key of the
-			// lower span be the same as the start key of the upper span, because
-			// the range end key is considered an exclusive bound.
-			w.err = errors.Errorf("pebble: overlapping range keys must be fragmented: %s, %s",
-				prevStartKey.Pretty(w.formatKey),
-				curStartKey.Pretty(w.formatKey))
-			return w.err
-		}
-	}
-
-	// TODO(travers): Add an invariant-gated check to ensure that suffix-values
-	// are sorted within coalesced spans.
-
-	// Range-keys and point-keys are intended to live in "parallel" keyspaces.
-	// However, we track a single seqnum in the table metadata that spans both of
-	// these keyspaces.
-	// TODO(travers): Consider tracking range key seqnums separately.
-	w.meta.updateSeqNum(key.SeqNum())
-
-	// Range tombstones are fragmented, so the start key of the first range key
-	// added will be the smallest. The largest range key is determined in
-	// Writer.Close() as the end key of the last range key added to the block.
-	if w.props.NumRangeKeys() == 0 {
-		w.meta.SetSmallestRangeKey(key.Clone())
-	}
-
-	// Update table properties.
-	w.props.RawRangeKeyKeySize += uint64(key.Size())
-	w.props.RawRangeKeyValueSize += uint64(len(value))
-	switch key.Kind() {
-	case base.InternalKeyKindRangeKeyDelete:
-		w.props.NumRangeKeyDels++
-	case base.InternalKeyKindRangeKeySet:
-		w.props.NumRangeKeySets++
-	case base.InternalKeyKindRangeKeyUnset:
-		w.props.NumRangeKeyUnsets++
-	default:
-		panic(errors.Errorf("pebble: invalid range key type: %s", key.Kind()))
-	}
-
-	// Add the key to the block.
-	w.rangeKeyBlock.Add(key, value)
-	return nil
+	return w.rw.err
 }
 
 // tempRangeKeyBuf returns a slice of length n from the Writer's rkBuf byte
@@ -1310,915 +215,14 @@ func (w *Writer) tempRangeKeyCopy(k []byte) []byte {
 	return buf
 }
 
-func (w *Writer) maybeAddToFilter(key []byte) {
-	if w.filter != nil {
-		prefix := key[:w.split(key)]
-		w.filter.addKey(prefix)
-	}
-}
-
-func (w *Writer) flush(key InternalKey) error {
-	// We're finishing a data block.
-	err := w.finishDataBlockProps(w.dataBlockBuf)
-	if err != nil {
-		return err
-	}
-	w.dataBlockBuf.finish()
-	w.dataBlockBuf.compressAndChecksum(w.compression)
-	// Since dataBlockEstimates.addInflightDataBlock was never called, the
-	// inflightSize is set to 0.
-	w.coordination.sizeEstimate.dataBlockCompressed(len(w.dataBlockBuf.compressed), 0)
-
-	// Determine if the index block should be flushed. Since we're accessing the
-	// dataBlockBuf.dataBlock.curKey here, we have to make sure that once we start
-	// to pool the dataBlockBufs, the curKey isn't used by the Writer once the
-	// dataBlockBuf is added back to a sync.Pool. In this particular case, the
-	// byte slice which supports "sep" will eventually be copied when "sep" is
-	// added to the index block.
-	prevKey := w.dataBlockBuf.dataBlock.CurKey()
-	sep := w.indexEntrySep(prevKey, key, w.dataBlockBuf)
-	// We determine that we should flush an index block from the Writer client
-	// goroutine, but we actually finish the index block from the writeQueue.
-	// When we determine that an index block should be flushed, we need to call
-	// BlockPropertyCollector.FinishIndexBlock. But block property collector
-	// calls must happen sequentially from the Writer client. Therefore, we need
-	// to determine that we are going to flush the index block from the Writer
-	// client.
-	shouldFlushIndexBlock := supportsTwoLevelIndex(w.tableFormat) && w.indexBlock.shouldFlush(
-		sep, encodedBHPEstimatedSize, w.indexBlockOptions, w.allocatorSizeClasses,
-	)
-
-	var indexProps []byte
-	var flushableIndexBlock *indexBlockBuf
-	if shouldFlushIndexBlock {
-		flushableIndexBlock = w.indexBlock
-		w.indexBlock = newIndexBlockBuf(w.coordination.parallelismEnabled)
-		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to
-		// flush the index block.
-		indexProps, err = w.finishIndexBlockProps()
-		if err != nil {
-			return err
-		}
-	}
-
-	// We've called BlockPropertyCollector.FinishDataBlock, and, if necessary,
-	// BlockPropertyCollector.FinishIndexBlock. Since we've decided to finish
-	// the data block, we can call
-	// BlockPropertyCollector.AddPrevDataBlockToIndexBlock.
-	w.addPrevDataBlockToIndexBlockProps()
-
-	// Schedule a write.
-	writeTask := writeTaskPool.Get().(*writeTask)
-	// We're setting compressionDone to indicate that compression of this block
-	// has already been completed.
-	writeTask.compressionDone <- true
-	writeTask.buf = w.dataBlockBuf
-	writeTask.indexEntrySep = sep
-	writeTask.currIndexBlock = w.indexBlock
-	writeTask.indexInflightSize = sep.Size() + encodedBHPEstimatedSize
-	writeTask.finishedIndexProps = indexProps
-	writeTask.flushableIndexBlock = flushableIndexBlock
-
-	// The writeTask corresponds to an unwritten index entry.
-	w.indexBlock.addInflight(writeTask.indexInflightSize)
-
-	w.dataBlockBuf = nil
-	if w.coordination.parallelismEnabled {
-		w.coordination.writeQueue.add(writeTask)
-	} else {
-		err = w.coordination.writeQueue.addSync(writeTask)
-	}
-	w.dataBlockBuf = newDataBlockBuf(w.restartInterval, w.checksumType)
-
-	return err
-}
-
-func (w *Writer) maybeFlush(key InternalKey, valueLen int) error {
-	if !w.dataBlockBuf.shouldFlush(key, valueLen, w.dataBlockOptions, w.allocatorSizeClasses) {
-		return nil
-	}
-
-	err := w.flush(key)
-
-	if err != nil {
-		w.err = err
-		return err
-	}
-
-	return nil
-}
-
-// dataBlockBuf.dataBlockProps set by this method must be encoded before any future use of the
-// dataBlockBuf.blockPropsEncoder, since the properties slice will get reused by the
-// blockPropsEncoder.
-func (w *Writer) finishDataBlockProps(buf *dataBlockBuf) error {
-	if len(w.blockPropCollectors) == 0 {
-		return nil
-	}
-	var err error
-	buf.blockPropsEncoder.resetProps()
-	for i := range w.blockPropCollectors {
-		scratch := buf.blockPropsEncoder.getScratchForProp()
-		if scratch, err = w.blockPropCollectors[i].FinishDataBlock(scratch); err != nil {
-			return err
-		}
-		buf.blockPropsEncoder.addProp(shortID(i), scratch)
-	}
-
-	buf.dataBlockProps = buf.blockPropsEncoder.unsafeProps()
-	return nil
-}
-
-// The BlockHandleWithProperties returned by this method must be encoded before any future use of
-// the Writer.blockPropsEncoder, since the properties slice will get reused by the blockPropsEncoder.
-// maybeAddBlockPropertiesToBlockHandle should only be called if block is being written synchronously
-// with the Writer client.
-func (w *Writer) maybeAddBlockPropertiesToBlockHandle(
-	bh block.Handle,
-) (BlockHandleWithProperties, error) {
-	err := w.finishDataBlockProps(w.dataBlockBuf)
-	if err != nil {
-		return BlockHandleWithProperties{}, err
-	}
-	return BlockHandleWithProperties{Handle: bh, Props: w.dataBlockBuf.dataBlockProps}, nil
-}
-
-func (w *Writer) indexEntrySep(prevKey, key InternalKey, dataBlockBuf *dataBlockBuf) InternalKey {
-	// Make a rough guess that we want key-sized scratch to compute the separator.
-	if cap(dataBlockBuf.sepScratch) < key.Size() {
-		dataBlockBuf.sepScratch = make([]byte, 0, key.Size()*2)
-	}
-
-	var sep InternalKey
-	if key.UserKey == nil && key.Trailer == 0 {
-		sep = prevKey.Successor(w.compare, w.successor, dataBlockBuf.sepScratch[:0])
-	} else {
-		sep = prevKey.Separator(w.compare, w.separator, dataBlockBuf.sepScratch[:0], key)
-	}
-	return sep
-}
-
-// addIndexEntry adds an index entry for the specified key and block handle.
-// addIndexEntry can be called from both the Writer client goroutine, and the
-// writeQueue goroutine. If the flushIndexBuf != nil, then the indexProps, as
-// they're used when the index block is finished.
-//
-// Invariant:
-//  1. addIndexEntry must not store references to the sep InternalKey, the tmp
-//     byte slice, bhp.Props. That is, these must be either deep copied or
-//     encoded.
-//  2. addIndexEntry must not hold references to the flushIndexBuf, and the writeTo
-//     indexBlockBufs.
-func (w *Writer) addIndexEntry(
-	sep InternalKey,
-	bhp BlockHandleWithProperties,
-	tmp []byte,
-	flushIndexBuf *indexBlockBuf,
-	writeTo *indexBlockBuf,
-	inflightSize int,
-	indexProps []byte,
-) error {
-	if bhp.Length == 0 {
-		// A valid blockHandle must be non-zero.
-		// In particular, it must have a non-zero length.
-		return nil
-	}
-
-	encoded := encodeBlockHandleWithProperties(tmp, bhp)
-
-	if flushIndexBuf != nil {
-		if cap(w.indexPartitions) == 0 {
-			w.indexPartitions = make([]indexBlockAndBlockProperties, 0, 32)
-		}
-		// Enable two level indexes if there is more than one index block.
-		w.twoLevelIndex = true
-		if err := w.finishIndexBlock(flushIndexBuf, indexProps); err != nil {
-			return err
-		}
-	}
-
-	writeTo.add(sep, encoded, inflightSize)
-	return nil
-}
-
-func (w *Writer) addPrevDataBlockToIndexBlockProps() {
-	for i := range w.blockPropCollectors {
-		w.blockPropCollectors[i].AddPrevDataBlockToIndexBlock()
-	}
-}
-
-// addIndexEntrySync adds an index entry for the specified key and block handle.
-// Writer.addIndexEntry is only called synchronously once Writer.Close is called.
-// addIndexEntrySync should only be called if we're sure that index entries
-// aren't being written asynchronously.
-//
-// Invariant:
-//  1. addIndexEntrySync must not store references to the prevKey, key InternalKey's,
-//     the tmp byte slice. That is, these must be either deep copied or encoded.
-//
-// TODO: Improve coverage of this method. e.g. tests passed without the line
-// `w.twoLevelIndex = true` previously.
-func (w *Writer) addIndexEntrySync(
-	prevKey, key InternalKey, bhp BlockHandleWithProperties, tmp []byte,
-) error {
-	return w.addIndexEntrySep(w.indexEntrySep(prevKey, key, w.dataBlockBuf), bhp, tmp)
-}
-
-func (w *Writer) addIndexEntrySep(
-	sep InternalKey, bhp BlockHandleWithProperties, tmp []byte,
-) error {
-	shouldFlush := supportsTwoLevelIndex(
-		w.tableFormat) && w.indexBlock.shouldFlush(
-		sep, encodedBHPEstimatedSize, w.indexBlockOptions, w.allocatorSizeClasses,
-	)
-	var flushableIndexBlock *indexBlockBuf
-	var props []byte
-	var err error
-	if shouldFlush {
-		flushableIndexBlock = w.indexBlock
-		w.indexBlock = newIndexBlockBuf(w.coordination.parallelismEnabled)
-		w.twoLevelIndex = true
-		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to
-		// flush the index block.
-		props, err = w.finishIndexBlockProps()
-		if err != nil {
-			return err
-		}
-	}
-
-	err = w.addIndexEntry(sep, bhp, tmp, flushableIndexBlock, w.indexBlock, 0, props)
-	if flushableIndexBlock != nil {
-		flushableIndexBlock.clear()
-		indexBlockBufPool.Put(flushableIndexBlock)
-	}
-	w.addPrevDataBlockToIndexBlockProps()
-	return err
-}
-
-func shouldFlushWithHints(
-	keyLen, valueLen int,
-	restartInterval, estimatedBlockSize, numEntries int,
-	flushOptions flushDecisionOptions,
-	sizeClassHints []int,
-) bool {
-	if numEntries == 0 {
-		return false
-	}
-
-	// If we are not informed about the memory allocator's size classes we fall
-	// back to a simple set of flush heuristics that are unaware of internal
-	// fragmentation in block cache allocations.
-	if len(sizeClassHints) == 0 {
-		return shouldFlushWithoutHints(
-			keyLen, valueLen, restartInterval, estimatedBlockSize, numEntries, flushOptions)
-	}
-
-	// For size-class aware flushing we need to account for the metadata that is
-	// allocated when this block is loaded into the block cache. For instance, if
-	// a block has size 1020B it may fit within a 1024B class. However, when
-	// loaded into the block cache we also allocate space for the cache entry
-	// metadata. The new allocation of size ~1052B may now only fit within a
-	// 2048B class, which increases internal fragmentation.
-	blockSizeWithMetadata := estimatedBlockSize + cache.ValueMetadataSize
-
-	// For the fast path we can avoid computing the exact varint encoded
-	// key-value pair size. Instead, we combine the key-value pair size with an
-	// upper-bound estimate of the associated metadata (4B restart point, 4B
-	// shared prefix length, 5B varint unshared key size, 5B varint value size).
-	newEstimatedSize := blockSizeWithMetadata + keyLen + valueLen + 18
-	// Our new block size estimate disregards key prefix compression. This puts
-	// us at risk of overestimating the size and flushing small blocks. We
-	// mitigate this by imposing a minimum size restriction.
-	if blockSizeWithMetadata <= flushOptions.sizeClassAwareThreshold || newEstimatedSize <= flushOptions.blockSize {
-		return false
-	}
-
-	sizeClass, ok := blockSizeClass(blockSizeWithMetadata, sizeClassHints)
-	// If the block size could not be mapped to a size class we fall back to
-	// using a simpler set of flush heuristics.
-	if !ok {
-		return shouldFlushWithoutHints(
-			keyLen, valueLen, restartInterval, estimatedBlockSize, numEntries, flushOptions)
-	}
-
-	// Tighter upper-bound estimate of the metadata stored with the next
-	// key-value pair.
-	newSize := blockSizeWithMetadata + keyLen + valueLen
-	if numEntries%restartInterval == 0 {
-		newSize += 4
-	}
-	newSize += 4                            // varint for shared prefix length
-	newSize += uvarintLen(uint32(keyLen))   // varint for unshared key bytes
-	newSize += uvarintLen(uint32(valueLen)) // varint for value size
-
-	if blockSizeWithMetadata < flushOptions.blockSize {
-		newSizeClass, ok := blockSizeClass(newSize, sizeClassHints)
-		if ok && newSizeClass-newSize >= sizeClass-blockSizeWithMetadata {
-			// Although the block hasn't reached the target size, waiting to insert the
-			// next entry would exceed the target and increase memory fragmentation.
-			return true
-		}
-		return false
-	}
-
-	// Flush if inserting the next entry bumps the block size to the memory
-	// allocator's next size class.
-	return newSize > sizeClass
-}
-
-func shouldFlushWithoutHints(
-	keyLen, valueLen int,
-	restartInterval, estimatedBlockSize, numEntries int,
-	flushOptions flushDecisionOptions,
-) bool {
-	if estimatedBlockSize >= flushOptions.blockSize {
-		return true
-	}
-
-	// The block is currently smaller than the target size.
-	if estimatedBlockSize <= flushOptions.blockSizeThreshold {
-		// The block is smaller than the threshold size at which we'll consider
-		// flushing it.
-		return false
-	}
-
-	newSize := estimatedBlockSize + keyLen + valueLen
-	if numEntries%restartInterval == 0 {
-		newSize += 4
-	}
-	newSize += 4                            // varint for shared prefix length
-	newSize += uvarintLen(uint32(keyLen))   // varint for unshared key bytes
-	newSize += uvarintLen(uint32(valueLen)) // varint for value size
-	// Flush if the block plus the new entry is larger than the target size.
-	return newSize > flushOptions.blockSize
-}
-
-// blockSizeClass returns the smallest memory allocator size class that could
-// hold a block of a given size and returns a boolean indicating whether an
-// appropriate size class was found. It is useful for computing the potential
-// space wasted by an allocation.
-func blockSizeClass(blockSize int, sizeClassHints []int) (int, bool) {
-	sizeClassIdx, _ := slices.BinarySearch(sizeClassHints, blockSize)
-	if sizeClassIdx == len(sizeClassHints) {
-		return -1, false
-	}
-	return sizeClassHints[sizeClassIdx], true
-}
-
-func cloneKeyWithBuf(k InternalKey, a bytealloc.A) (bytealloc.A, InternalKey) {
-	if len(k.UserKey) == 0 {
-		return a, k
-	}
-	a, keyCopy := a.Copy(k.UserKey)
-	return a, InternalKey{UserKey: keyCopy, Trailer: k.Trailer}
-}
-
-// Invariants: The byte slice returned by finishIndexBlockProps is heap-allocated
-//
-//	and has its own lifetime, independent of the Writer and the blockPropsEncoder,
-//
-// and it is safe to:
-//  1. Reuse w.blockPropsEncoder without first encoding the byte slice returned.
-//  2. Store the byte slice in the Writer since it is a copy and not supported by
-//     an underlying buffer.
-func (w *Writer) finishIndexBlockProps() ([]byte, error) {
-	w.blockPropsEncoder.resetProps()
-	for i := range w.blockPropCollectors {
-		scratch := w.blockPropsEncoder.getScratchForProp()
-		var err error
-		if scratch, err = w.blockPropCollectors[i].FinishIndexBlock(scratch); err != nil {
-			return nil, err
-		}
-		w.blockPropsEncoder.addProp(shortID(i), scratch)
-	}
-	return w.blockPropsEncoder.props(), nil
-}
-
-// finishIndexBlock finishes the current index block and adds it to the top
-// level index block. This is only used when two level indexes are enabled.
-//
-// Invariants:
-//  1. The props slice passed into finishedIndexBlock must not be a
-//     owned by any other struct, since it will be stored in the Writer.indexPartitions
-//     slice.
-//  2. None of the buffers owned by indexBuf will be shallow copied and stored elsewhere.
-//     That is, it must be safe to reuse indexBuf after finishIndexBlock has been called.
-func (w *Writer) finishIndexBlock(indexBuf *indexBlockBuf, props []byte) error {
-	part := indexBlockAndBlockProperties{
-		nEntries: indexBuf.block.EntryCount(), properties: props,
-	}
-	w.indexSepAlloc, part.sep = cloneKeyWithBuf(
-		indexBuf.block.CurKey(), w.indexSepAlloc,
-	)
-	bk := indexBuf.finish()
-	if len(w.indexBlockAlloc) < len(bk) {
-		// Allocate enough bytes for approximately 16 index blocks.
-		w.indexBlockAlloc = make([]byte, len(bk)*16)
-	}
-	n := copy(w.indexBlockAlloc, bk)
-	part.block = w.indexBlockAlloc[:n:n]
-	w.indexBlockAlloc = w.indexBlockAlloc[n:]
-	w.indexPartitions = append(w.indexPartitions, part)
-	return nil
-}
-
-func (w *Writer) writeTwoLevelIndex() (block.Handle, error) {
-	props, err := w.finishIndexBlockProps()
-	if err != nil {
-		return block.Handle{}, err
-	}
-	// Add the final unfinished index.
-	if err = w.finishIndexBlock(w.indexBlock, props); err != nil {
-		return block.Handle{}, err
-	}
-
-	for i := range w.indexPartitions {
-		b := &w.indexPartitions[i]
-		w.props.NumDataBlocks += uint64(b.nEntries)
-
-		data := b.block
-		w.props.IndexSize += uint64(len(data))
-		bh, err := w.layout.WriteIndexBlock(data)
-		if err != nil {
-			return block.Handle{}, err
-		}
-		bhp := BlockHandleWithProperties{
-			Handle: bh,
-			Props:  b.properties,
-		}
-		encoded := encodeBlockHandleWithProperties(w.blockBuf.tmp[:], bhp)
-		w.topLevelIndexBlock.Add(b.sep, encoded)
-	}
-
-	// NB: RocksDB includes the block trailer length in the index size
-	// property, though it doesn't include the trailer in the top level
-	// index size property.
-	w.props.IndexPartitions = uint64(len(w.indexPartitions))
-	w.props.TopLevelIndexSize = uint64(w.topLevelIndexBlock.EstimatedSize())
-	w.props.IndexSize += w.props.TopLevelIndexSize + block.TrailerLen
-	return w.layout.WriteIndexBlock(w.topLevelIndexBlock.Finish())
-}
-
-func compressAndChecksum(
-	b []byte, compression Compression, blockBuf *blockBuf,
-) (compressed []byte, trailer block.Trailer) {
-	// Compress the buffer, discarding the result if the improvement isn't at
-	// least 12.5%.
-	blockType, compressed := compressBlock(compression, b, blockBuf.compressedBuf)
-	if blockType != noCompressionBlockType && cap(compressed) > cap(blockBuf.compressedBuf) {
-		blockBuf.compressedBuf = compressed[:cap(compressed)]
-	}
-	if len(compressed) < len(b)-len(b)/8 {
-		b = compressed
-	} else {
-		blockType = noCompressionBlockType
-	}
-
-	// Calculate the checksum.
-	trailer[0] = byte(blockType)
-	checksum := blockBuf.checksummer.Checksum(b, trailer[:1])
-	return b, block.MakeTrailer(byte(blockType), checksum)
-}
-
-// assertFormatCompatibility ensures that the features present on the table are
-// compatible with the table format version.
-func (w *Writer) assertFormatCompatibility() error {
-	// PebbleDBv1: block properties.
-	if len(w.blockPropCollectors) > 0 && w.tableFormat < TableFormatPebblev1 {
-		return errors.Newf(
-			"table format version %s is less than the minimum required version %s for block properties",
-			w.tableFormat, TableFormatPebblev1,
-		)
-	}
-
-	// PebbleDBv2: range keys.
-	if w.props.NumRangeKeys() > 0 && w.tableFormat < TableFormatPebblev2 {
-		return errors.Newf(
-			"table format version %s is less than the minimum required version %s for range keys",
-			w.tableFormat, TableFormatPebblev2,
-		)
-	}
-
-	// PebbleDBv3: value blocks.
-	if (w.props.NumValueBlocks > 0 || w.props.NumValuesInValueBlocks > 0 ||
-		w.props.ValueBlocksSize > 0) && w.tableFormat < TableFormatPebblev3 {
-		return errors.Newf(
-			"table format version %s is less than the minimum required version %s for value blocks",
-			w.tableFormat, TableFormatPebblev3)
-	}
-
-	// PebbleDBv4: DELSIZED tombstones.
-	if w.props.NumSizedDeletions > 0 && w.tableFormat < TableFormatPebblev4 {
-		return errors.Newf(
-			"table format version %s is less than the minimum required version %s for sized deletion tombstones",
-			w.tableFormat, TableFormatPebblev4)
-	}
-	return nil
-}
-
-// UnsafeLastPointUserKey returns the last point key written to the writer to
-// which this option was passed during creation. The returned key points
-// directly into a buffer belonging to the Writer. The value's lifetime ends the
-// next time a point key is added to the Writer.
-//
-// Must not be called after Writer is closed.
-func (w *Writer) UnsafeLastPointUserKey() []byte {
-	if w != nil && w.dataBlockBuf.dataBlock.EntryCount() >= 1 {
-		// w.dataBlockBuf.dataBlock.curKey is guaranteed to point to the last point key
-		// which was added to the Writer.
-		return w.dataBlockBuf.dataBlock.CurUserKey()
-	}
-	return nil
-}
-
-// EncodeSpan encodes the keys in the given span. The span can contain either
-// only RANGEDEL keys or only range keys.
-//
-// This is a low-level API that bypasses the fragmenter. The spans passed to
-// this function must be fragmented and ordered.
-func (w *Writer) EncodeSpan(span *keyspan.Span) error {
-	if span.Empty() {
-		return nil
-	}
-	if span.Keys[0].Kind() == base.InternalKeyKindRangeDelete {
-		return rangedel.Encode(span, w.Add)
-	}
-	for i := range w.blockPropCollectors {
-		if err := w.blockPropCollectors[i].AddRangeKeys(*span); err != nil {
-			return err
-		}
-	}
-	return w.rangeKeyEncoder.Encode(span)
-}
-
 // Close finishes writing the table and closes the underlying file that the
 // table was written to.
 func (w *Writer) Close() (err error) {
-	defer func() {
-		if w.valueBlockWriter != nil {
-			releaseValueBlockWriter(w.valueBlockWriter)
-			// Defensive code in case Close gets called again. We don't want to put
-			// the same object to a sync.Pool.
-			w.valueBlockWriter = nil
-		}
-		w.layout.Abort()
-		// Record any error in the writer (so we can exit early if Close is called
-		// again).
-		if err != nil {
-			w.err = err
-		}
-	}()
-
-	// finish must be called before we check for an error, because finish will
-	// block until every single task added to the writeQueue has been processed,
-	// and an error could be encountered while any of those tasks are processed.
-	if err := w.coordination.writeQueue.finish(); err != nil {
-		return err
+	if w.rw.err != nil {
+		return w.rw.err
 	}
-
-	if w.err != nil {
-		return w.err
-	}
-
-	// The w.meta.LargestPointKey is only used once the Writer is closed, so it is safe to set it
-	// when the Writer is closed.
-	//
-	// The following invariants ensure that setting the largest key at this point of a Writer close
-	// is correct:
-	// 1. Keys must only be added to the Writer in an increasing order.
-	// 2. The current w.dataBlockBuf is guaranteed to have the latest key added to the Writer. This
-	//    must be true, because a w.dataBlockBuf is only switched out when a dataBlock is flushed,
-	//    however, if a dataBlock is flushed, then we add a key to the new w.dataBlockBuf in the
-	//    addPoint function after the flush occurs.
-	if w.dataBlockBuf.dataBlock.EntryCount() >= 1 {
-		w.meta.SetLargestPointKey(w.dataBlockBuf.dataBlock.CurKey().Clone())
-	}
-
-	// Finish the last data block, or force an empty data block if there
-	// aren't any data blocks at all.
-	if w.dataBlockBuf.dataBlock.EntryCount() > 0 || w.indexBlock.block.EntryCount() == 0 {
-		bh, err := w.layout.WriteDataBlock(w.dataBlockBuf.dataBlock.Finish(), &w.dataBlockBuf.blockBuf)
-		if err != nil {
-			return err
-		}
-		bhp, err := w.maybeAddBlockPropertiesToBlockHandle(bh)
-		if err != nil {
-			return err
-		}
-		prevKey := w.dataBlockBuf.dataBlock.CurKey()
-		if err := w.addIndexEntrySync(prevKey, InternalKey{}, bhp, w.dataBlockBuf.tmp[:]); err != nil {
-			return err
-		}
-	}
-	w.props.DataSize = w.layout.offset
-
-	// Write the filter block.
-	if w.filter != nil {
-		bh, err := w.layout.WriteFilterBlock(w.filter)
-		if err != nil {
-			return err
-		}
-		w.props.FilterPolicyName = w.filter.policyName()
-		w.props.FilterSize = bh.Length
-	}
-
-	if w.twoLevelIndex {
-		w.props.IndexType = twoLevelIndex
-		// Write the two level index block.
-		if _, err = w.writeTwoLevelIndex(); err != nil {
-			return err
-		}
-	} else {
-		w.props.IndexType = binarySearchIndex
-		// NB: RocksDB includes the block trailer length in the index size
-		// property, though it doesn't include the trailer in the filter size
-		// property.
-		w.props.IndexSize = uint64(w.indexBlock.estimatedSize()) + block.TrailerLen
-		w.props.NumDataBlocks = uint64(w.indexBlock.block.EntryCount())
-		// Write the single level index block.
-		if _, err = w.layout.WriteIndexBlock(w.indexBlock.finish()); err != nil {
-			return err
-		}
-	}
-
-	// Write the range-del block.
-	if w.props.NumRangeDeletions > 0 {
-		// Because the range tombstones are fragmented, the end key of the last
-		// added range tombstone will be the largest range tombstone key. Note
-		// that we need to make this into a range deletion sentinel because
-		// sstable boundaries are inclusive while the end key of a range
-		// deletion tombstone is exclusive. A Clone() is necessary as
-		// rangeDelBlock.curValue is the same slice that will get passed into
-		// w.writer, and some implementations of vfs.File mutate the slice
-		// passed into Write(). Also, w.meta will often outlive the blockWriter,
-		// and so cloning curValue allows the rangeDelBlock's internal buffer to
-		// get gc'd.
-		k := base.MakeRangeDeleteSentinelKey(w.rangeDelBlock.CurValue()).Clone()
-		w.meta.SetLargestRangeDelKey(k)
-		if _, err := w.layout.WriteRangeDeletionBlock(w.rangeDelBlock.Finish()); err != nil {
-			return err
-		}
-	}
-
 	// Write the range-key block, flushing any remaining spans from the
 	// fragmenter first.
 	w.fragmenter.Finish()
-
-	if w.props.NumRangeKeys() > 0 {
-		key := w.rangeKeyBlock.CurKey()
-		kind := key.Kind()
-		endKey, _, err := rangekey.DecodeEndKey(kind, w.rangeKeyBlock.CurValue())
-		if err != nil {
-			return err
-		}
-		k := base.MakeExclusiveSentinelKey(kind, endKey).Clone()
-		w.meta.SetLargestRangeKey(k)
-		if _, err := w.layout.WriteRangeKeyBlock(w.rangeKeyBlock.Finish()); err != nil {
-			return err
-		}
-	}
-
-	if w.valueBlockWriter != nil {
-		_, vbStats, err := w.valueBlockWriter.finish(&w.layout, w.layout.offset)
-		if err != nil {
-			return err
-		}
-		w.props.NumValueBlocks = vbStats.numValueBlocks
-		w.props.NumValuesInValueBlocks = vbStats.numValuesInValueBlocks
-		w.props.ValueBlocksSize = vbStats.valueBlocksAndIndexSize
-	}
-
-	{
-		// Finish and record the prop collectors if props are not yet recorded.
-		// Pre-computed props might have been copied by specialized sst creators
-		// like suffix replacer.
-		if len(w.props.UserProperties) == 0 {
-			userProps := make(map[string]string)
-			for i := range w.blockPropCollectors {
-				scratch := w.blockPropsEncoder.getScratchForProp()
-				// Place the shortID in the first byte.
-				scratch = append(scratch, byte(i))
-				buf, err := w.blockPropCollectors[i].FinishTable(scratch)
-				if err != nil {
-					return err
-				}
-				var prop string
-				if len(buf) > 0 {
-					prop = string(buf)
-				}
-				// NB: The property is populated in the map even if it is the
-				// empty string, since the presence in the map is what indicates
-				// that the block property collector was used when writing.
-				userProps[w.blockPropCollectors[i].Name()] = prop
-			}
-			if len(userProps) > 0 {
-				w.props.UserProperties = userProps
-			}
-		}
-
-		// Write the properties block.
-		var raw rowblk.Writer
-		// The restart interval is set to infinity because the properties block
-		// is always read sequentially and cached in a heap located object. This
-		// reduces table size without a significant impact on performance.
-		raw.RestartInterval = propertiesBlockRestartInterval
-		w.props.CompressionOptions = rocksDBCompressionOptions
-		w.props.save(w.tableFormat, &raw)
-		w.layout.WritePropertiesBlock(raw.Finish())
-	}
-
-	// Write the table footer.
-	w.meta.Size, err = w.layout.Finish()
-	if err != nil {
-		return err
-	}
-	w.meta.Properties = w.props
-
-	// Check that the features present in the table are compatible with the format
-	// configured for the table.
-	if err = w.assertFormatCompatibility(); err != nil {
-		return err
-	}
-
-	w.dataBlockBuf.clear()
-	dataBlockBufPool.Put(w.dataBlockBuf)
-	w.dataBlockBuf = nil
-	w.indexBlock.clear()
-	indexBlockBufPool.Put(w.indexBlock)
-	w.indexBlock = nil
-
-	// Make any future calls to Set or Close return an error.
-	w.err = errWriterClosed
-	return nil
-}
-
-// EstimatedSize returns the estimated size of the sstable being written if a
-// call to Finish() was made without adding additional keys.
-func (w *Writer) EstimatedSize() uint64 {
-	if w == nil {
-		return 0
-	}
-	return w.coordination.sizeEstimate.size() +
-		uint64(w.dataBlockBuf.dataBlock.EstimatedSize()) +
-		w.indexBlock.estimatedSize()
-}
-
-// Metadata returns the metadata for the finished sstable. Only valid to call
-// after the sstable has been finished.
-func (w *Writer) Metadata() (*WriterMetadata, error) {
-	if !w.layout.IsFinished() {
-		return nil, errors.New("pebble: writer is not closed")
-	}
-	return &w.meta, nil
-}
-
-// WriterOption provide an interface to do work on Writer while it is being
-// opened.
-type WriterOption interface {
-	// writerApply is called on the writer during opening in order to set
-	// internal parameters.
-	writerApply(*Writer)
-}
-
-// NewWriter returns a new table writer for the file. Closing the writer will
-// close the file.
-func NewWriter(writable objstorage.Writable, o WriterOptions, extraOpts ...WriterOption) *Writer {
-	o = o.ensureDefaults()
-	w := &Writer{
-		layout: makeLayoutWriter(writable, o),
-		meta: WriterMetadata{
-			SmallestSeqNum: math.MaxUint64,
-		},
-		dataBlockOptions: flushDecisionOptions{
-			blockSize:               o.BlockSize,
-			blockSizeThreshold:      (o.BlockSize*o.BlockSizeThreshold + 99) / 100,
-			sizeClassAwareThreshold: (o.BlockSize*o.SizeClassAwareThreshold + 99) / 100,
-		},
-		indexBlockOptions: flushDecisionOptions{
-			blockSize:               o.IndexBlockSize,
-			blockSizeThreshold:      (o.IndexBlockSize*o.BlockSizeThreshold + 99) / 100,
-			sizeClassAwareThreshold: (o.IndexBlockSize*o.SizeClassAwareThreshold + 99) / 100,
-		},
-		compare:               o.Comparer.Compare,
-		split:                 o.Comparer.Split,
-		formatKey:             o.Comparer.FormatKey,
-		compression:           o.Compression,
-		separator:             o.Comparer.Separator,
-		successor:             o.Comparer.Successor,
-		tableFormat:           o.TableFormat,
-		isStrictObsolete:      o.IsStrictObsolete,
-		writingToLowestLevel:  o.WritingToLowestLevel,
-		restartInterval:       o.BlockRestartInterval,
-		checksumType:          o.Checksum,
-		disableKeyOrderChecks: o.internal.DisableKeyOrderChecks,
-		indexBlock:            newIndexBlockBuf(o.Parallelism),
-		rangeDelBlock:         rowblk.Writer{RestartInterval: 1},
-		rangeKeyBlock:         rowblk.Writer{RestartInterval: 1},
-		topLevelIndexBlock:    rowblk.Writer{RestartInterval: 1},
-		fragmenter: keyspan.Fragmenter{
-			Cmp:    o.Comparer.Compare,
-			Format: o.Comparer.FormatKey,
-		},
-		allocatorSizeClasses: o.AllocatorSizeClasses,
-	}
-	if w.tableFormat >= TableFormatPebblev3 {
-		w.shortAttributeExtractor = o.ShortAttributeExtractor
-		w.requiredInPlaceValueBound = o.RequiredInPlaceValueBound
-		if !o.DisableValueBlocks {
-			w.valueBlockWriter = newValueBlockWriter(
-				w.dataBlockOptions.blockSize, w.dataBlockOptions.blockSizeThreshold, w.compression, w.checksumType, 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.coordination.init(o.Parallelism, w)
-
-	if writable == nil {
-		w.err = errors.New("pebble: nil writable")
-		return w
-	}
-
-	// Note that WriterOptions are applied in two places; the ones with a
-	// preApply() method are applied here. The rest are applied down below after
-	// default properties are set.
-	type preApply interface{ preApply() }
-	for _, opt := range extraOpts {
-		if _, ok := opt.(preApply); ok {
-			opt.writerApply(w)
-		}
-	}
-
-	if o.FilterPolicy != nil {
-		switch o.FilterType {
-		case TableFilter:
-			w.filter = newTableFilterWriter(o.FilterPolicy)
-		default:
-			panic(fmt.Sprintf("unknown filter type: %v", o.FilterType))
-		}
-	}
-
-	w.props.ComparerName = o.Comparer.Name
-	w.props.CompressionName = o.Compression.String()
-	w.props.MergerName = o.MergerName
-	w.props.PropertyCollectorNames = "[]"
-
-	numBlockPropertyCollectors := len(o.BlockPropertyCollectors)
-	if w.tableFormat >= TableFormatPebblev4 {
-		numBlockPropertyCollectors++
-	}
-
-	if numBlockPropertyCollectors > 0 {
-		if numBlockPropertyCollectors > maxPropertyCollectors {
-			w.err = errors.New("pebble: too many block property collectors")
-			return w
-		}
-		w.blockPropCollectors = make([]BlockPropertyCollector, 0, numBlockPropertyCollectors)
-		for _, constructFn := range o.BlockPropertyCollectors {
-			w.blockPropCollectors = append(w.blockPropCollectors, constructFn())
-		}
-		if w.tableFormat >= TableFormatPebblev4 {
-			w.blockPropCollectors = append(w.blockPropCollectors, &w.obsoleteCollector)
-		}
-
-		var buf bytes.Buffer
-		buf.WriteString("[")
-		for i := range w.blockPropCollectors {
-			if i > 0 {
-				buf.WriteString(",")
-			}
-			buf.WriteString(w.blockPropCollectors[i].Name())
-		}
-		buf.WriteString("]")
-		w.props.PropertyCollectorNames = buf.String()
-	}
-
-	// Apply the remaining WriterOptions that do not have a preApply() method.
-	for _, opt := range extraOpts {
-		if _, ok := opt.(preApply); ok {
-			continue
-		}
-		opt.writerApply(w)
-	}
-
-	// Initialize the range key fragmenter and encoder.
-	w.fragmenter.Emit = w.encodeFragmentedRangeKeySpan
-	w.rangeKeyEncoder.Emit = w.addRangeKey
-	return w
-}
-
-// SetSnapshotPinnedProperties sets the properties for pinned keys. Should only
-// be used internally by Pebble.
-func (w *Writer) SetSnapshotPinnedProperties(
-	pinnedKeyCount, pinnedKeySize, pinnedValueSize uint64,
-) {
-	w.props.SnapshotPinnedKeys = pinnedKeyCount
-	w.props.SnapshotPinnedKeySize = pinnedKeySize
-	w.props.SnapshotPinnedValueSize = pinnedValueSize
+	return w.rw.Close()
 }


Test patch:
diff --git a/data_test.go b/data_test.go
index 89788020..ebcbe154 100644
--- a/data_test.go
+++ b/data_test.go
@@ -586,7 +586,7 @@ func runBuildRemoteCmd(td *datadriven.TestData, d *DB, storage remote.Storage) e
 	for kv := iter.First(); kv != nil; kv = iter.Next() {
 		tmp := kv.K
 		tmp.SetSeqNum(0)
-		if err := w.Add(tmp, kv.InPlaceValue()); err != nil {
+		if err := w.Raw().Add(tmp, kv.InPlaceValue()); err != nil {
 			return err
 		}
 	}
@@ -599,7 +599,7 @@ func runBuildRemoteCmd(td *datadriven.TestData, d *DB, storage remote.Storage) e
 		for ; s != nil && err == nil; s, err = rdi.Next() {
 			err = rangedel.Encode(s, func(k base.InternalKey, v []byte) error {
 				k.SetSeqNum(0)
-				return w.Add(k, v)
+				return w.Raw().Add(k, v)
 			})
 			if err != nil {
 				return err
@@ -680,7 +680,7 @@ func runBuildCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
 	for kv := iter.First(); kv != nil; kv = iter.Next() {
 		tmp := kv.K
 		tmp.SetSeqNum(0)
-		if err := w.Add(tmp, kv.InPlaceValue()); err != nil {
+		if err := w.Raw().Add(tmp, kv.InPlaceValue()); err != nil {
 			return err
 		}
 	}
@@ -693,7 +693,7 @@ func runBuildCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
 		for ; s != nil && err == nil; s, err = rdi.Next() {
 			err = rangedel.Encode(s, func(k base.InternalKey, v []byte) error {
 				k.SetSeqNum(0)
-				return w.Add(k, v)
+				return w.Raw().Add(k, v)
 			})
 			if err != nil {
 				return err
diff --git a/event_listener_test.go b/event_listener_test.go
index 4942f823..42b1d97d 100644
--- a/event_listener_test.go
+++ b/event_listener_test.go
@@ -138,7 +138,7 @@ func TestEventListener(t *testing.T) {
 			if err != nil {
 				return err.Error()
 			}
-			w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
+			w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
 				TableFormat: d.FormatMajorVersion().MaxTableFormat(),
 			})
 			if err := w.Add(base.MakeInternalKey([]byte("a"), 0, InternalKeyKindSet), nil); err != nil {
@@ -172,7 +172,7 @@ func TestEventListener(t *testing.T) {
 				if err != nil {
 					return err
 				}
-				w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
+				w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
 					TableFormat: d.FormatMajorVersion().MaxTableFormat(),
 				})
 				if err := w.Add(base.MakeInternalKey([]byte{key}, 0, InternalKeyKindSet), nil); err != nil {
diff --git a/ingest_test.go b/ingest_test.go
index 94c49e22..e37896b3 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -99,7 +99,7 @@ func TestIngestLoad(t *testing.T) {
 			if err != nil {
 				return err.Error()
 			}
-			w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), writerOpts)
+			w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), writerOpts)
 			for _, data := range strings.Split(td.Input, "\n") {
 				if strings.HasPrefix(data, "rangekey: ") {
 					data = strings.TrimPrefix(data, "rangekey: ")
@@ -193,7 +193,7 @@ func TestIngestLoadRand(t *testing.T) {
 
 			expected[i].ExtendPointKeyBounds(cmp, keys[0], keys[len(keys)-1])
 
-			w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
+			w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
 				TableFormat: version.MaxTableFormat(),
 			})
 			var count uint64
@@ -1092,7 +1092,7 @@ func testIngestSharedImpl(
 			f, err := to.opts.FS.Create(sstPath, vfs.WriteCategoryUnspecified)
 			require.NoError(t, err)
 			replicateCounter++
-			w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), writeOpts)
+			w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), writeOpts)
 
 			var sharedSSTs []SharedSSTMeta
 			err = from.ScanInternal(context.TODO(), sstable.CategoryAndQoS{}, startKey, endKey,
@@ -1103,7 +1103,11 @@ func testIngestSharedImpl(
 					return nil
 				},
 				func(start, end []byte, seqNum base.SeqNum) error {
-					require.NoError(t, w.DeleteRange(start, end))
+					require.NoError(t, w.EncodeSpan(&keyspan.Span{
+						Start: start,
+						End:   end,
+						Keys:  []keyspan.Key{{Trailer: base.MakeTrailer(0, base.InternalKeyKindRangeDelete)}},
+					}))
 					return nil
 				},
 				func(start, end []byte, keys []keyspan.Key) error {
@@ -1588,7 +1592,7 @@ func TestConcurrentExcise(t *testing.T) {
 			f, err := to.opts.FS.Create(sstPath, vfs.WriteCategoryUnspecified)
 			require.NoError(t, err)
 			replicateCounter++
-			w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), writeOpts)
+			w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), writeOpts)
 
 			var sharedSSTs []SharedSSTMeta
 			err = from.ScanInternal(context.TODO(), sstable.CategoryAndQoS{}, startKey, endKey,
@@ -1599,7 +1603,11 @@ func TestConcurrentExcise(t *testing.T) {
 					return nil
 				},
 				func(start, end []byte, seqNum base.SeqNum) error {
-					require.NoError(t, w.DeleteRange(start, end))
+					require.NoError(t, w.EncodeSpan(&keyspan.Span{
+						Start: start,
+						End:   end,
+						Keys:  []keyspan.Key{{Trailer: base.MakeTrailer(0, base.InternalKeyKindRangeDelete)}},
+					}))
 					return nil
 				},
 				func(start, end []byte, keys []keyspan.Key) error {
@@ -2021,7 +2029,7 @@ func TestIngestExternal(t *testing.T) {
 			f, err := to.opts.FS.Create(sstPath, vfs.WriteCategoryUnspecified)
 			require.NoError(t, err)
 			replicateCounter++
-			w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), writeOpts)
+			w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), writeOpts)
 
 			var externalFiles []ExternalFile
 			err = from.ScanInternal(context.TODO(), sstable.CategoryAndQoS{}, startKey, endKey,
@@ -2032,7 +2040,11 @@ func TestIngestExternal(t *testing.T) {
 					return nil
 				},
 				func(start, end []byte, seqNum base.SeqNum) error {
-					require.NoError(t, w.DeleteRange(start, end))
+					require.NoError(t, w.EncodeSpan(&keyspan.Span{
+						Start: start,
+						End:   end,
+						Keys:  []keyspan.Key{{Trailer: base.MakeTrailer(0, base.InternalKeyKindRangeDelete)}},
+					}))
 					return nil
 				},
 				func(start, end []byte, keys []keyspan.Key) error {
@@ -2574,7 +2586,7 @@ func TestIngestCompact(t *testing.T) {
 	f, err := mem.Create(src(0), vfs.WriteCategoryUnspecified)
 	require.NoError(t, err)
 
-	w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
+	w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
 	key := []byte("a")
 	require.NoError(t, w.Add(base.MakeInternalKey(key, 0, InternalKeyKindSet), nil))
 	require.NoError(t, w.Close())
@@ -3212,12 +3224,12 @@ func TestIngestFileNumReuseCrash(t *testing.T) {
 func TestIngest_UpdateSequenceNumber(t *testing.T) {
 	mem := vfs.NewMem()
 	cmp := base.DefaultComparer.Compare
-	parse := func(input string) (*sstable.Writer, error) {
+	parse := func(input string) (*sstable.RawWriter, error) {
 		f, err := mem.Create("ext", vfs.WriteCategoryUnspecified)
 		if err != nil {
 			return nil, err
 		}
-		w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
+		w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
 			TableFormat: sstable.TableFormatMax,
 		})
 		for _, data := range strings.Split(input, "\n") {
diff --git a/internal/compact/spans_test.go b/internal/compact/spans_test.go
index 622c77e8..a42fe65d 100644
--- a/internal/compact/spans_test.go
+++ b/internal/compact/spans_test.go
@@ -121,7 +121,7 @@ func TestSplitAndEncodeSpan(t *testing.T) {
 			}
 
 			obj := &objstorage.MemObj{}
-			tw := sstable.NewWriter(obj, sstable.WriterOptions{TableFormat: sstable.TableFormatMax})
+			tw := sstable.NewRawWriter(obj, sstable.WriterOptions{TableFormat: sstable.TableFormatMax})
 			require.NoError(t, SplitAndEncodeSpan(base.DefaultComparer.Compare, &span, upToKey, tw))
 			require.NoError(t, tw.Close())
 			_, rangeDels, rangeKeys := sstable.ReadAll(obj)
diff --git a/level_checker_test.go b/level_checker_test.go
index 8f52c808..8a8218fd 100644
--- a/level_checker_test.go
+++ b/level_checker_test.go
@@ -183,7 +183,7 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 				writerOpts.SetInternal(sstableinternal.WriterOptions{
 					DisableKeyOrderChecks: disableKeyOrderChecks,
 				})
-				w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), writerOpts)
+				w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), writerOpts)
 				var tombstones []keyspan.Span
 				frag := keyspan.Fragmenter{
 					Cmp:    testkeys.Comparer.Compare,
diff --git a/level_iter_test.go b/level_iter_test.go
index f42832e5..27436e5c 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -221,7 +221,7 @@ func (lt *levelIterTest) runBuild(d *datadriven.TestData) string {
 		}
 	}
 	fp := bloom.FilterPolicy(10)
-	w := sstable.NewWriter(objstorageprovider.NewFileWritable(f0), sstable.WriterOptions{
+	w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f0), sstable.WriterOptions{
 		Comparer:     &lt.cmp,
 		FilterPolicy: fp,
 		TableFormat:  tableFormat,
@@ -508,9 +508,9 @@ func buildLevelIterTables(
 		files[i] = f
 	}
 
-	writers := make([]*sstable.Writer, len(files))
+	writers := make([]*sstable.RawWriter, len(files))
 	for i := range files {
-		writers[i] = sstable.NewWriter(objstorageprovider.NewFileWritable(files[i]), sstable.WriterOptions{
+		writers[i] = sstable.NewRawWriter(objstorageprovider.NewFileWritable(files[i]), sstable.WriterOptions{
 			BlockRestartInterval: restartInterval,
 			BlockSize:            blockSize,
 			Compression:          NoCompression,
diff --git a/merging_iter_test.go b/merging_iter_test.go
index 21e08253..9cabc5e8 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -221,7 +221,7 @@ func TestMergingIterCornerCases(t *testing.T) {
 				if err != nil {
 					return err.Error()
 				}
-				w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
+				w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
 				var tombstones []keyspan.Span
 				frag := keyspan.Fragmenter{
 					Cmp:    cmp,
@@ -339,9 +339,9 @@ func buildMergingIterTables(
 		files[i] = f
 	}
 
-	writers := make([]*sstable.Writer, len(files))
+	writers := make([]*sstable.RawWriter, len(files))
 	for i := range files {
-		writers[i] = sstable.NewWriter(objstorageprovider.NewFileWritable(files[i]), sstable.WriterOptions{
+		writers[i] = sstable.NewRawWriter(objstorageprovider.NewFileWritable(files[i]), sstable.WriterOptions{
 			BlockRestartInterval: restartInterval,
 			BlockSize:            blockSize,
 			Compression:          NoCompression,
@@ -542,7 +542,7 @@ func buildLevelsForMergingIterSeqSeek(
 	}
 
 	const targetL6FirstFileSize = 2 << 20
-	writers := make([][]*sstable.Writer, levelCount)
+	writers := make([][]*sstable.RawWriter, levelCount)
 	// A policy unlikely to have false positives.
 	filterPolicy := bloom.FilterPolicy(100)
 	for i := range files {
@@ -572,7 +572,7 @@ func buildLevelsForMergingIterSeqSeek(
 					writerOptions.IndexBlockSize = 1
 				}
 			}
-			writers[i] = append(writers[i], sstable.NewWriter(objstorageprovider.NewFileWritable(files[i][j]), writerOptions))
+			writers[i] = append(writers[i], sstable.NewRawWriter(objstorageprovider.NewFileWritable(files[i][j]), writerOptions))
 		}
 	}
 
diff --git a/range_del_test.go b/range_del_test.go
index 69026ef8..dd3b9d01 100644
--- a/range_del_test.go
+++ b/range_del_test.go
@@ -597,7 +597,7 @@ func benchmarkRangeDelIterate(b *testing.B, entries, deleted int, snapshotCompac
 	if err != nil {
 		b.Fatal(err)
 	}
-	w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
+	w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
 		BlockSize: 32 << 10, // 32 KB
 	})
 	for i := 0; i < entries; i++ {
diff --git a/scan_internal_test.go b/scan_internal_test.go
index e6903ba6..8b355250 100644
--- a/scan_internal_test.go
+++ b/scan_internal_test.go
@@ -427,7 +427,7 @@ func TestScanInternal(t *testing.T) {
 						}
 						keyspan.SortKeysByTrailer(&keys)
 						newSpan := &keyspan.Span{Start: span.Start, End: span.End, Keys: keys}
-						require.NoError(t, w.EncodeSpan(newSpan))
+						require.NoError(t, w.Raw().EncodeSpan(newSpan))
 					}
 					require.NoError(t, err)
 				}
@@ -438,7 +438,7 @@ func TestScanInternal(t *testing.T) {
 					var err error
 					value, _, err = kv.Value(value)
 					require.NoError(t, err)
-					require.NoError(t, w.Add(kv.K, value))
+					require.NoError(t, w.Raw().Add(kv.K, value))
 				}
 				points.Close()
 				require.NoError(t, w.Close())
diff --git a/sstable/data_test.go b/sstable/data_test.go
index 31f9cb42..e25e73ea 100644
--- a/sstable/data_test.go
+++ b/sstable/data_test.go
@@ -137,7 +137,7 @@ func runBuildCmd(
 				return nil, nil, err
 			}
 		default:
-			if err := w.AddWithForceObsolete(key, value, forceObsolete); err != nil {
+			if err := w.Raw().AddWithForceObsolete(key, value, forceObsolete); err != nil {
 				return nil, nil, err
 			}
 		}
@@ -146,7 +146,7 @@ func runBuildCmd(
 	for _, v := range rangeDels {
 		for _, k := range v.Keys {
 			ik := base.InternalKey{UserKey: v.Start, Trailer: k.Trailer}
-			if err := w.Add(ik, v.End); err != nil {
+			if err := w.Raw().Add(ik, v.End); err != nil {
 				return nil, nil, err
 			}
 		}
@@ -160,7 +160,7 @@ func runBuildCmd(
 	if err := w.Close(); err != nil {
 		return nil, nil, err
 	}
-	meta, err := w.Metadata()
+	meta, err := w.Raw().Metadata()
 	if err != nil {
 		return nil, nil, err
 	}
@@ -222,11 +222,11 @@ func runBuildRawCmd(
 			// Note: specifying range keys directly (instead of spans) should only be
 			// done to check for error handling; they will not contribute to block
 			// properties.
-			if err := w.addRangeKey(key, value); err != nil {
+			if err := w.Raw().addRangeKey(key, value); err != nil {
 				return nil, nil, err
 			}
 		default:
-			if err := w.Add(key, value); err != nil {
+			if err := w.Raw().Add(key, value); err != nil {
 				return nil, nil, err
 			}
 		}
@@ -234,7 +234,7 @@ func runBuildRawCmd(
 	if err := w.Close(); err != nil {
 		return nil, nil, err
 	}
-	meta, err := w.Metadata()
+	meta, err := w.Raw().Metadata()
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/sstable/random_test.go b/sstable/random_test.go
index 88806c0d..387107dc 100644
--- a/sstable/random_test.go
+++ b/sstable/random_test.go
@@ -380,7 +380,7 @@ func buildRandomSSTable(f vfs.File, cfg randomTableConfig) (*WriterMetadata, err
 	alloc = nil
 
 	valueBuf := make([]byte, cfg.maxValLen)
-	w := NewWriter(objstorageprovider.NewFileWritable(f), *cfg.wopts)
+	w := NewRawWriter(objstorageprovider.NewFileWritable(f), *cfg.wopts)
 	for i := 0; i < len(keys); i++ {
 		var value []byte
 		switch keys[i].Kind() {
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index dc0c2746..33702e61 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -1756,7 +1756,7 @@ func TestReader_TableFormat(t *testing.T) {
 		require.NoError(t, err)
 
 		opts := WriterOptions{TableFormat: want}
-		w := NewWriter(objstorageprovider.NewFileWritable(f), opts)
+		w := NewRawWriter(objstorageprovider.NewFileWritable(f), opts)
 		err = w.Close()
 		require.NoError(t, err)
 
@@ -1789,7 +1789,7 @@ func buildTestTableWithProvider(
 	f0, _, err := provider.Create(context.Background(), base.FileTypeTable, base.DiskFileNum(0), objstorage.CreateOptions{})
 	require.NoError(t, err)
 
-	w := NewWriter(f0, WriterOptions{
+	w := NewRawWriter(f0, WriterOptions{
 		BlockSize:      blockSize,
 		IndexBlockSize: indexBlockSize,
 		Compression:    compression,
@@ -1834,7 +1834,7 @@ func buildBenchmarkTable(
 		b.Fatal(err)
 	}
 
-	w := NewWriter(objstorageprovider.NewFileWritable(f0), options)
+	w := NewRawWriter(objstorageprovider.NewFileWritable(f0), options)
 
 	var keys [][]byte
 	var ikey InternalKey
@@ -2391,7 +2391,7 @@ func BenchmarkIteratorScanObsolete(b *testing.B) {
 		f0, err := mem.Create("bench", vfs.WriteCategoryUnspecified)
 		require.NoError(b, err)
 		options.TableFormat = tableFormat
-		w := NewWriter(objstorageprovider.NewFileWritable(f0), options)
+		w := NewRawWriter(objstorageprovider.NewFileWritable(f0), options)
 		val := make([]byte, 100)
 		rng := rand.New(rand.NewSource(100))
 		for i := int64(0); i < keys.Count(); i++ {
diff --git a/sstable/suffix_rewriter_test.go b/sstable/suffix_rewriter_test.go
index 7d1fe258..c1f5582a 100644
--- a/sstable/suffix_rewriter_test.go
+++ b/sstable/suffix_rewriter_test.go
@@ -154,7 +154,7 @@ func make4bSuffixTestSST(
 		binary.BigEndian.PutUint64(key[:8], 123) // 16-byte shared prefix
 		binary.BigEndian.PutUint64(key[8:16], 456)
 		binary.BigEndian.PutUint64(key[16:], uint64(i))
-		err := w.AddWithForceObsolete(
+		err := w.Raw().AddWithForceObsolete(
 			base.MakeInternalKey(key, 0, InternalKeyKindSet), key, false)
 		if err != nil {
 			t.Fatal(err)
diff --git a/sstable/table_test.go b/sstable/table_test.go
index db7fd531..24436f15 100644
--- a/sstable/table_test.go
+++ b/sstable/table_test.go
@@ -447,7 +447,7 @@ func TestFinalBlockIsWritten(t *testing.T) {
 						t.Errorf("nk=%d, vLen=%d: memFS create: %v", nk, vLen, err)
 						continue
 					}
-					w := NewWriter(objstorageprovider.NewFileWritable(wf), WriterOptions{
+					w := NewRawWriter(objstorageprovider.NewFileWritable(wf), WriterOptions{
 						BlockSize:      blockSize,
 						IndexBlockSize: indexBlockSize,
 					})
diff --git a/sstable/test_fixtures.go b/sstable/test_fixtures.go
index 2d6c863c..1dcafdb0 100644
--- a/sstable/test_fixtures.go
+++ b/sstable/test_fixtures.go
@@ -150,24 +150,23 @@ func buildHamletTestSST(
 	// NB: We don't have byte equality with RocksDB any longer.
 	var rangeDelLength int
 	var rangeDelCounter int
-	var rangeDelStart InternalKey
+	var rangeDelStart []byte
 	for i, k := range keys {
 		v := wordCount[k]
-		ikey := base.MakeInternalKey([]byte(k), 0, InternalKeyKindSet)
-		if err := w.Add(ikey, []byte(v)); err != nil {
+		if err := w.Set([]byte(k), []byte(v)); err != nil {
 			return err
 		}
 		// This mirrors the logic in `make-table.cc`. It adds range deletions of
 		// increasing length for every 100 keys added.
 		if i%100 == 0 {
-			rangeDelStart = ikey.Clone()
+			rangeDelStart = []byte(k)
 			rangeDelCounter = 0
 			rangeDelLength++
 		}
 		rangeDelCounter++
 
 		if rangeDelCounter == rangeDelLength {
-			if err := w.DeleteRange(rangeDelStart.UserKey, ikey.UserKey); err != nil {
+			if err := w.DeleteRange(rangeDelStart, []byte(k)); err != nil {
 				return err
 			}
 		}
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index ef306332..f618e298 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -517,7 +517,7 @@ func TestParallelWriterErrorProp(t *testing.T) {
 	w := NewWriter(objstorageprovider.NewFileWritable(f), opts)
 	// Directly testing this, because it's difficult to get the Writer to
 	// encounter an error, precisely when the writeQueue is doing block writes.
-	w.coordination.writeQueue.err = errors.New("write queue write error")
+	w.rw.coordination.writeQueue.err = errors.New("write queue write error")
 	w.Set(ikey("a").UserKey, nil)
 	w.Set(ikey("b").UserKey, nil)
 	err = w.Close()
@@ -831,7 +831,7 @@ func TestWriterBlockPropertiesErrors(t *testing.T) {
 			f, err := fs.Create("test", vfs.WriteCategoryUnspecified)
 			require.NoError(t, err)
 
-			w := NewWriter(objstorageprovider.NewFileWritable(f), WriterOptions{
+			w := NewRawWriter(objstorageprovider.NewFileWritable(f), WriterOptions{
 				BlockSize: 1,
 				BlockPropertyCollectors: []func() BlockPropertyCollector{
 					func() BlockPropertyCollector {
@@ -967,7 +967,7 @@ func TestWriterRace(t *testing.T) {
 			}
 			defer wg.Done()
 			f := &objstorage.MemObj{}
-			w := NewWriter(f, opts)
+			w := NewRawWriter(f, opts)
 			for ki := 0; ki < len(keys); ki++ {
 				require.NoError(
 					t,
diff --git a/table_cache_test.go b/table_cache_test.go
index c88903b6..81b02bee 100644
--- a/table_cache_test.go
+++ b/table_cache_test.go
@@ -176,7 +176,7 @@ func newTableCacheContainerTest(
 		}
 		tw := sstable.NewWriter(w, sstable.WriterOptions{TableFormat: sstable.TableFormatPebblev2})
 		ik := base.ParseInternalKey(fmt.Sprintf("k.SET.%d", i))
-		if err := tw.Add(ik, xxx[:i]); err != nil {
+		if err := tw.Raw().Add(ik, xxx[:i]); err != nil {
 			return nil, nil, errors.Wrap(err, "tw.Set")
 		}
 		if err := tw.RangeKeySet([]byte("k"), []byte("l"), nil, xxx[:i]); err != nil {
diff --git a/table_stats_test.go b/table_stats_test.go
index 766797f4..16b9e06a 100644
--- a/table_stats_test.go
+++ b/table_stats_test.go
@@ -202,7 +202,7 @@ func TestTableRangeDeletionIter(t *testing.T) {
 			if err != nil {
 				return err.Error()
 			}
-			w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
+			w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
 				TableFormat: sstable.TableFormatMax,
 			})
 			m = &fileMetadata{}
diff --git a/version_set_test.go b/version_set_test.go
index fda6eb47..b280ee15 100644
--- a/version_set_test.go
+++ b/version_set_test.go
@@ -31,7 +31,7 @@ func writeAndIngest(t *testing.T, mem vfs.FS, d *DB, k InternalKey, v []byte, fi
 	path := mem.PathJoin("ext", filename)
 	f, err := mem.Create(path, vfs.WriteCategoryUnspecified)
 	require.NoError(t, err)
-	w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
+	w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
 	require.NoError(t, w.Add(k, v))
 	require.NoError(t, w.Close())
 	require.NoError(t, d.Ingest([]string{path}))