Repository: cockroachdb/pebble
Pull request #3864 (2024-08-19 19:16:44)

tool: fix up wal dump, add wal dump-merged
Adjust the wal dump tool to error on unrecognized key kinds, and surface errors
encounted more prominently in the output.

Additionally, add a new `wal dump-merged` command that dumps the merged view of
a WAL that's composed of multiple physical segment files.

Informs #3865.

cockroach-teamcity: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cockroachdb/pebble/3864)
<!-- Reviewable:end -->


Patch:
diff --git a/compaction.go b/compaction.go
index 7947748a..fe6024bc 100644
--- a/compaction.go
+++ b/compaction.go
@@ -2666,7 +2666,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.RawWriter, CPUWorkHandle, error) {
+) (objstorage.ObjectMetadata, *sstable.RawRowWriter, CPUWorkHandle, error) {
 	d.mu.Lock()
 	diskFileNum := d.mu.versions.getNextDiskFileNum()
 	d.mu.Unlock()
diff --git a/internal/compact/run.go b/internal/compact/run.go
index 02a638da..75439a1e 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.RawWriter) {
+func (r *Runner) WriteTable(objMeta objstorage.ObjectMetadata, tw *sstable.RawRowWriter) {
 	if r.err != nil {
 		panic("error already encountered")
 	}
@@ -159,7 +159,7 @@ func (r *Runner) WriteTable(objMeta objstorage.ObjectMetadata, tw *sstable.RawWr
 	r.tables[len(r.tables)-1].WriterMeta = *writerMeta
 }
 
-func (r *Runner) writeKeysToTable(tw *sstable.RawWriter) (splitKey []byte, _ error) {
+func (r *Runner) writeKeysToTable(tw *sstable.RawRowWriter) (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 d8d5bd47..41206e19 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.RawWriter,
+	cmp base.Compare, span *keyspan.Span, upToKey []byte, tw *sstable.RawRowWriter,
 ) error {
 	if span.Empty() {
 		return nil
diff --git a/metamorphic/build.go b/metamorphic/build.go
index f6909e48..7040a3c2 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -96,7 +96,7 @@ func writeSSTForIngestion(
 		if err != nil {
 			return nil, err
 		}
-		if err := w.Raw().Add(k.K, valBytes); err != nil {
+		if err := w.Raw().AddWithForceObsolete(k.K, valBytes, false); err != nil {
 			return nil, err
 		}
 	}
diff --git a/metamorphic/ops.go b/metamorphic/ops.go
index 84572330..10337389 100644
--- a/metamorphic/ops.go
+++ b/metamorphic/ops.go
@@ -1928,7 +1928,7 @@ func (r *replicateOp) runSharedReplicate(
 			if err != nil {
 				panic(err)
 			}
-			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val)
+			return w.Raw().AddWithForceObsolete(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false)
 		},
 		func(start, end []byte, seqNum base.SeqNum) error {
 			return w.DeleteRange(start, end)
@@ -1991,7 +1991,7 @@ func (r *replicateOp) runExternalReplicate(
 			if err != nil {
 				panic(err)
 			}
-			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val)
+			return w.Raw().AddWithForceObsolete(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false)
 		},
 		func(start, end []byte, seqNum base.SeqNum) error {
 			return w.DeleteRange(start, end)
diff --git a/sstable/colblk/bitmap.go b/sstable/colblk/bitmap.go
index a89d256d..3b5aacd6 100644
--- a/sstable/colblk/bitmap.go
+++ b/sstable/colblk/bitmap.go
@@ -26,6 +26,8 @@ import (
 // non-zero. The summary bitmap accelerates predecessor and successor
 // operations.
 type Bitmap struct {
+	// data contains the bitmap data, according to defaultBitmapEncoding, or it
+	// is nil if the bitmap is all zeros.
 	data     UnsafeRawSlice[uint64]
 	bitCount int
 }
@@ -38,8 +40,13 @@ var _ Array[bool] = Bitmap{}
 // performed, so the caller must guarantee the bitmap is appropriately sized and
 // the provided bitCount correctly identifies the number of bits in the bitmap.
 func DecodeBitmap(b []byte, off uint32, bitCount int) (bitmap Bitmap, endOffset uint32) {
-	sz := bitmapRequiredSize(bitCount)
+	encoding := bitmapEncoding(b[off])
+	off++
+	if encoding == zeroBitmapEncoding {
+		return Bitmap{bitCount: bitCount}, off
+	}
 	off = align(off, align64)
+	sz := bitmapRequiredSize(bitCount)
 	if len(b) < int(off)+sz {
 		panic(errors.AssertionFailedf("bitmap of %d bits requires at least %d bytes; provided with %d-byte slice",
 			bitCount, bitmapRequiredSize(bitCount), len(b[off:])))
@@ -55,13 +62,25 @@ var _ DecodeFunc[Bitmap] = DecodeBitmap
 
 // At returns true if the bit at position i is set and false otherwise.
 func (b Bitmap) At(i int) bool {
-	return (b.data.At(i>>6 /* i/64 */) & (1 << uint(i%64))) != 0
+	if b.data.ptr == nil {
+		// zero bitmap case.
+		return false
+	}
+	// Inline b.data.At(i/64).
+	// The offset of the correct word is i / 64 * 8 = (i >> 3) &^ 0b111
+	const mask = ^uintptr(0b111)
+	val := *(*uint64)(unsafe.Pointer(uintptr(b.data.ptr) + (uintptr(i)>>3)&mask))
+	return val&(1<<(uint(i)&63)) != 0
 }
 
 // Successor returns the next bit greater than or equal to i set in the bitmap.
 // The i parameter must be in [0, bitCount). Returns the number of bits
 // represented by the bitmap if no next bit is set.
 func (b Bitmap) Successor(i int) int {
+	if b.data.ptr == nil {
+		// Zero bitmap case.
+		return b.bitCount
+	}
 	// nextInWord returns the index of the smallest set bit with an index >= bit
 	// within the provided word.  The returned index is an index local to the
 	// word.
@@ -127,6 +146,10 @@ func (b Bitmap) Successor(i int) int {
 // bitmap. The i parameter must be in [0, bitCount). Returns -1 if no previous
 // bit is set.
 func (b Bitmap) Predecessor(i int) int {
+	if b.data.ptr == nil {
+		// Zero bitmap case.
+		return -1
+	}
 	// prevInWord returns the index of the largest set bit ≤ bit within the
 	// provided word. The returned index is an index local to the word. Returns
 	// -1 if no set bit is found.
@@ -204,33 +227,44 @@ func (b Bitmap) String() string {
 	return sb.String()
 }
 
-// BitmapBuilder constructs a Bitmap. Bits are default false.
+// BitmapBuilder constructs a Bitmap. Bits default to false.
 type BitmapBuilder struct {
 	words []uint64
 }
 
+type bitmapEncoding uint8
+
+const (
+	// defaultBitmapEncoding encodes the bitmap using ⌈n/64⌉ words followed by
+	// ⌈⌈n/64⌉/64⌉ summary words.
+	defaultBitmapEncoding bitmapEncoding = iota
+	// zeroBitmapEncoding is used for the special case when the bitmap is empty.
+	zeroBitmapEncoding
+)
+
 // Assert that BitmapBuilder implements ColumnWriter.
 var _ ColumnWriter = (*BitmapBuilder)(nil)
 
+// bitmapRequiredSize returns the size of an encoded bitmap in bytes, using the
+// defaultBitmapEncoding.
 func bitmapRequiredSize(total int) int {
 	nWords := (total + 63) >> 6          // divide by 64
 	nSummaryWords := (nWords + 63) >> 6  // divide by 64
 	return (nWords + nSummaryWords) << 3 // multiply by 8
 }
 
-// Set sets the bit at position i if v is true and clears the bit at position i
-// otherwise. Callers need not call Set if v is false and Set(i, true) has not
-// been called yet.
-func (b *BitmapBuilder) Set(i int, v bool) {
+// Set sets the bit at position i to true.
+func (b *BitmapBuilder) Set(i int) {
 	w := i >> 6 // divide by 64
 	for len(b.words) <= w {
 		b.words = append(b.words, 0)
 	}
-	if v {
-		b.words[w] |= 1 << uint(i%64)
-	} else {
-		b.words[w] &^= 1 << uint(i%64)
-	}
+	b.words[w] |= 1 << uint(i%64)
+}
+
+// isZero returns true if no bits are set and Invert was not called.
+func (b *BitmapBuilder) isZero() bool {
+	return len(b.words) == 0
 }
 
 // Reset resets the bitmap to the empty state.
@@ -247,6 +281,21 @@ func (b *BitmapBuilder) DataType(int) DataType { return DataTypeBool }
 
 // Size implements the ColumnWriter interface.
 func (b *BitmapBuilder) Size(rows int, offset uint32) uint32 {
+	// First byte will be the encoding type.
+	offset++
+	if b.isZero() {
+		return offset
+	}
+	offset = align(offset, align64)
+	return offset + uint32(bitmapRequiredSize(rows))
+}
+
+// InvertedSize returns the size of the encoded bitmap, assuming Invert will be called.
+func (b *BitmapBuilder) InvertedSize(rows int, offset uint32) uint32 {
+	// First byte will be the encoding type.
+	offset++
+	// An inverted bitmap will never use all-zeros encoding (even if it happens to
+	// be all zero).
 	offset = align(offset, align64)
 	return offset + uint32(bitmapRequiredSize(rows))
 }
@@ -254,6 +303,9 @@ func (b *BitmapBuilder) Size(rows int, offset uint32) uint32 {
 // Invert inverts the bitmap, setting all bits that are not set and clearing all
 // bits that are set. If the bitmap's tail is sparse and is not large enough to
 // represent nRows rows, it's first materialized.
+//
+// Note that Invert can affect the Size of the bitmap. Use InvertedSize() if you
+// intend to invert the bitmap before finishing.
 func (b *BitmapBuilder) Invert(nRows int) {
 	// If the tail of b is sparse, fill in zeroes before inverting.
 	nBitmapWords := (nRows + 63) >> 6
@@ -266,6 +318,12 @@ func (b *BitmapBuilder) Invert(nRows int) {
 // Finish finalizes the bitmap, computing the per-word summary bitmap and
 // writing the resulting data to buf at offset.
 func (b *BitmapBuilder) Finish(col, nRows int, offset uint32, buf []byte) uint32 {
+	if b.isZero() {
+		buf[offset] = byte(zeroBitmapEncoding)
+		return offset + 1
+	}
+	buf[offset] = byte(defaultBitmapEncoding)
+	offset++
 	offset = alignWithZeroes(buf, offset, align64)
 	dest := makeUnsafeRawSlice[uint64](unsafe.Pointer(&buf[offset]))
 
@@ -318,6 +376,14 @@ func (b *BitmapBuilder) WriteDebug(w io.Writer, rows int) {
 }
 
 func bitmapToBinFormatter(f *binfmt.Formatter, rows int) {
+	encoding := bitmapEncoding(f.PeekUint(1))
+	f.HexBytesln(1, "bitmap encoding")
+	if encoding == zeroBitmapEncoding {
+		return
+	}
+	if encoding != defaultBitmapEncoding {
+		panic(fmt.Sprintf("unknown bitmap encoding %d", encoding))
+	}
 	if aligned := align(f.Offset(), 8); aligned-f.Offset() != 0 {
 		f.HexBytesln(aligned-f.Offset(), "padding to align to 64-bit boundary")
 	}
diff --git a/sstable/colblk/data_block.go b/sstable/colblk/data_block.go
index dccf0eaf..1d252dbd 100644
--- a/sstable/colblk/data_block.go
+++ b/sstable/colblk/data_block.go
@@ -327,8 +327,13 @@ func (ks *defaultKeySeeker) MaterializeUserKey(keyIter *PrefixBytesIter, prevRow
 		ks.prefixes.SetAt(keyIter, row)
 	}
 	suffix := ks.suffixes.At(row)
-	memmove(unsafe.Pointer(uintptr(keyIter.ptr)+uintptr(keyIter.len)), unsafe.Pointer(unsafe.SliceData(suffix)), uintptr(len(suffix)))
-	return unsafe.Slice((*byte)(keyIter.ptr), keyIter.len+len(suffix))
+	res := keyIter.buf[:len(keyIter.buf)+len(suffix)]
+	memmove(
+		unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(keyIter.buf)))+uintptr(len(keyIter.buf))),
+		unsafe.Pointer(unsafe.SliceData(suffix)),
+		uintptr(len(suffix)),
+	)
+	return res
 }
 
 func (ks *defaultKeySeeker) Release() {
@@ -441,11 +446,11 @@ func (w *DataBlockWriter) Add(
 ) {
 	w.KeyWriter.WriteKey(w.rows, ikey.UserKey, kcmp.PrefixLen, kcmp.CommonPrefixLen)
 	if kcmp.PrefixEqual() {
-		w.prefixSame.Set(w.rows, true)
+		w.prefixSame.Set(w.rows)
 	}
 	w.trailers.Set(w.rows, uint64(ikey.Trailer))
 	if valuePrefix.IsValueHandle() {
-		w.isValueExternal.Set(w.rows, true)
+		w.isValueExternal.Set(w.rows)
 		// Write the value with the value prefix byte preceding the value.
 		w.valuePrefixTmp[0] = byte(valuePrefix)
 		w.values.PutConcat(w.valuePrefixTmp[:], value)
@@ -470,7 +475,7 @@ func (w *DataBlockWriter) Size() int {
 	off := blockHeaderSize(len(w.Schema.ColumnTypes)+dataBlockColumnMax, dataBlockCustomHeaderSize)
 	off = w.KeyWriter.Size(w.rows, off)
 	off = w.trailers.Size(w.rows, off)
-	off = w.prefixSame.Size(w.rows, off)
+	off = w.prefixSame.InvertedSize(w.rows, off)
 	off = w.values.Size(w.rows, off)
 	off = w.isValueExternal.Size(w.rows, off)
 	off++ // trailer padding byte
@@ -485,6 +490,11 @@ func (w *DataBlockWriter) Finish() []byte {
 		Columns: uint16(cols),
 		Rows:    uint32(w.rows),
 	}
+
+	// Invert the prefix-same bitmap before writing it out, because we want it
+	// to represent when the prefix changes.
+	w.prefixSame.Invert(w.rows)
+
 	w.enc.init(w.Size(), h, dataBlockCustomHeaderSize)
 
 	// Write the max key length in the custom header.
@@ -496,9 +506,6 @@ func (w *DataBlockWriter) Finish() []byte {
 	// Write the internal key trailers.
 	w.enc.encode(w.rows, &w.trailers)
 
-	// Invert the prefix-same bitmap before writing it out, because we want it
-	// to represent when the prefix changes.
-	w.prefixSame.Invert(w.rows)
 	w.enc.encode(w.rows, &w.prefixSame)
 
 	// Write the value columns.
@@ -596,7 +603,10 @@ func (i *DataBlockIter) Init(
 	}
 	// Allocate a keyIter buffer that's large enough to hold the largest user
 	// key in the block.
-	i.keyIter.Alloc(int(r.maximumKeyLength))
+
+	n := int(r.maximumKeyLength)
+	ptr := mallocgc(uintptr(n), nil, false)
+	i.keyIter.buf = unsafe.Slice((*byte)(ptr), n)[:0]
 	return i.keySeeker.Init(r)
 }
 
@@ -652,8 +662,8 @@ func (i *DataBlockIter) Next() *base.InternalKV {
 		Trailer: base.InternalKeyTrailer(i.r.trailers.At(i.row)),
 	}
 	// Inline i.r.values.At(row).
-	startOffset := i.r.values.offsets.At(i.row)
-	v := unsafe.Slice((*byte)(i.r.values.ptr(startOffset)), i.r.values.offsets.At(i.row+1)-startOffset)
+	startOffset, endOffset := i.r.values.offsets.At2(i.row)
+	v := unsafe.Slice((*byte)(i.r.values.ptr(startOffset)), endOffset-startOffset)
 	if i.r.isValueExternal.At(i.row) {
 		i.kv.V = i.getLazyValue(v)
 	} else {
diff --git a/sstable/colblk/index_block.go b/sstable/colblk/index_block.go
new file mode 100644
index 00000000..34a483a5
--- /dev/null
+++ b/sstable/colblk/index_block.go
@@ -0,0 +1,297 @@
+// 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 colblk
+
+import (
+	"bytes"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/binfmt"
+	"github.com/cockroachdb/pebble/sstable/block"
+)
+
+const indexBlockCustomHeaderSize = 0
+
+// IndexBlockWriter writes columnar index blocks. The writer is used for both
+// first-level and second-level index blocks. The index block schema consists of
+// three primary columns:
+//   - Separators: a user key that is ≥ the largest user key in the
+//     corresponding entry, and ≤ the smallest user key in the next entry.
+//     Note that this allows consecutive separators to be equal. This is
+//     possible when snapshots required we preserve duplicate user keys at
+//     different sequence numbers.
+//   - Offsets: the offset of the end of the corresponding block.
+//   - Lengths: the length in bytes of the corresponding block.
+//   - Block properties: a slice encoding arbitrary user-defined block
+//     properties.
+//
+// TODO(jackson): Consider splitting separators into prefixes and suffixes (even
+// without user-defined columns). This would allow us to use prefix compression
+// for the prefix. Separators should typically be suffixless unless two KVs with
+// the same prefix straddle a block boundary. We would need to use a buffer to
+// materialize the separator key when we need to use it outside the context of
+// seeking within the block.
+type IndexBlockWriter struct {
+	separators      RawBytesBuilder
+	offsets         UintBuilder
+	lengths         UintBuilder
+	blockProperties RawBytesBuilder
+	numColumns      int
+	rows            int
+	enc             blockEncoder
+}
+
+const (
+	indexBlockColumnSeparator = iota
+	indexBlockColumnOffsets
+	indexBlockColumnLengths
+	indexBlockColumnBlockProperties
+	indexBlockColumnCount
+)
+
+// Init initializes the index block writer.
+func (w *IndexBlockWriter) Init(blockProperties ...ColumnWriter) {
+	w.separators.Init()
+	w.offsets.Init()
+	w.lengths.InitWithDefault()
+	w.numColumns = indexBlockColumnCount
+	w.blockProperties.Init()
+	w.rows = 0
+}
+
+// Reset resets the index block writer to its initial state, retaining buffers.
+func (w *IndexBlockWriter) Reset() {
+	w.separators.Reset()
+	w.offsets.Reset()
+	w.lengths.Reset()
+	w.blockProperties.Reset()
+	w.rows = 0
+	w.enc.reset()
+}
+
+// AddBlockHandle adds a new separator and end offset of a data block to the
+// index block.  Add returns the index of the row. The caller should add rows to
+// the block properties columns too if necessary.
+//
+// AddBlockHandle should only be used for first-level index blocks.
+func (w *IndexBlockWriter) AddBlockHandle(
+	separator []byte, handle block.Handle, blockProperties []byte,
+) int {
+	idx := w.rows
+	w.separators.Put(separator)
+	w.offsets.Set(w.rows, handle.Offset)
+	w.lengths.Set(w.rows, handle.Length)
+	w.blockProperties.Put(blockProperties)
+	w.rows++
+	return idx
+}
+
+// Size returns the size of the pending index block.
+func (w *IndexBlockWriter) Size() int {
+	off := blockHeaderSize(w.numColumns, indexBlockCustomHeaderSize)
+	off = w.separators.Size(w.rows, off)
+	off = w.offsets.Size(w.rows, off)
+	off = w.lengths.Size(w.rows, off)
+	off = w.blockProperties.Size(w.rows, off)
+	off++
+	return int(off)
+}
+
+// Finish serializes the pending index block.
+func (w *IndexBlockWriter) Finish() []byte {
+	w.enc.init(w.Size(), Header{
+		Version: Version1,
+		Columns: uint16(w.numColumns),
+		Rows:    uint32(w.rows),
+	}, indexBlockCustomHeaderSize)
+	w.enc.encode(w.rows, &w.separators)
+	w.enc.encode(w.rows, &w.offsets)
+	w.enc.encode(w.rows, &w.lengths)
+	w.enc.encode(w.rows, &w.blockProperties)
+	return w.enc.finish()
+}
+
+// An IndexReader reads columnar index blocks.
+type IndexReader struct {
+	separators RawBytes
+	offsets    UnsafeUints
+	lengths    UnsafeUints // only used for second-level index blocks
+	br         BlockReader
+}
+
+// Init initializes the index reader with the given serialized index block.
+func (r *IndexReader) Init(data []byte) {
+	r.br.Init(data, indexBlockCustomHeaderSize)
+	r.separators = r.br.RawBytes(indexBlockColumnSeparator)
+	r.offsets = r.br.Uints(indexBlockColumnOffsets)
+	r.lengths = r.br.Uints(indexBlockColumnLengths)
+}
+
+// DebugString prints a human-readable explanation of the keyspan block's binary
+// representation.
+func (r *IndexReader) DebugString() string {
+	f := binfmt.New(r.br.data).LineWidth(20)
+	f.CommentLine("index block header")
+	r.br.headerToBinFormatter(f)
+	for i := 0; i < indexBlockColumnCount; i++ {
+		r.br.columnToBinFormatter(f, i, int(r.br.header.Rows))
+	}
+	return f.String()
+}
+
+// IndexIter is an iterator over the block entries in an index block.
+type IndexIter struct {
+	r   *IndexReader
+	n   int
+	row int
+
+	h           block.BufferHandle
+	allocReader IndexReader
+}
+
+// Assert that IndexIter satisfies the block.IndexBlockIterator constraint.
+var _ = satisfiesIndexBlock[IndexIter, *IndexIter](IndexIter{})
+
+func satisfiesIndexBlock[T any, PT block.IndexBlockIterator[T]](indexBlock T) PT {
+	return &indexBlock
+}
+
+// Init initializes an index iterator from the provided reader.
+func (i *IndexIter) Init(r *IndexReader) {
+	*i = IndexIter{r: r, n: int(r.br.header.Rows)}
+}
+
+// InitHandle initializes an iterator from the provided block handle.
+func (i *IndexIter) InitHandle(
+	cmp base.Compare, split base.Split, block block.BufferHandle, transforms block.IterTransforms,
+) error {
+	// TODO(jackson): Handle the transforms.
+
+	// TODO(jackson): If block.h != nil, use a *IndexReader that's allocated
+	// when the block is loaded into the block cache. On cache hits, this will
+	// reduce the amount of setup necessary to use an iterator. (It's relatively
+	// common to open an iterator and perform just a few seeks, so avoiding the
+	// overhead can be material.)
+	i.h.Release()
+	i.h = block
+	i.allocReader.Init(i.h.Get())
+	i.r = &i.allocReader
+	return nil
+}
+
+// RowIndex returns the index of the block entry at the iterator's current
+// position.
+func (i *IndexIter) RowIndex() int {
+	return i.row
+}
+
+// ResetForReuse resets the IndexIter for reuse, retaining buffers to avoid
+// future allocations.
+func (i *IndexIter) ResetForReuse() IndexIter {
+	return IndexIter{ /* nothing to retain */ }
+}
+
+// Valid returns true if the iterator is currently positioned at a valid block
+// handle.
+func (i *IndexIter) Valid() bool {
+	return 0 <= i.row && i.row < int(i.r.br.header.Rows)
+}
+
+// Invalidate invalidates the block iterator, removing references to the block
+// it was initialized with.
+func (i *IndexIter) Invalidate() {
+	i.r = nil
+	i.h = block.BufferHandle{}
+}
+
+// IsDataInvalidated returns true when the iterator has been invalidated
+// using an Invalidate call. NB: this is different from Valid.
+func (i *IndexIter) IsDataInvalidated() bool {
+	return i.r == nil
+}
+
+// Handle returns the underlying block buffer handle, if the iterator was
+// initialized with one.
+func (i *IndexIter) Handle() block.BufferHandle {
+	return i.h
+}
+
+// Separator returns the separator at the iterator's current position. The
+// iterator must be positioned at a valid row.
+func (i *IndexIter) Separator() []byte {
+	return i.r.separators.At(i.row)
+}
+
+// BlockHandleWithProperties decodes the block handle with any encoded
+// properties at the iterator's current position.
+func (i *IndexIter) BlockHandleWithProperties() (block.HandleWithProperties, error) {
+	return block.HandleWithProperties{
+		Handle: block.Handle{
+			Offset: i.r.offsets.At(i.row),
+			Length: i.r.lengths.At(i.row),
+		},
+		Props: i.r.br.RawBytes(indexBlockColumnBlockProperties).At(i.row),
+	}, nil
+}
+
+// SeekGE seeks the index iterator to the first block entry with a separator key
+// greater or equal to the given key. It returns false if the seek key is
+// greater than all index block separators.
+func (i *IndexIter) SeekGE(key []byte) bool {
+	// Define f(-1) == false and f(upper) == true.
+	// Invariant: f(index-1) == false, f(upper) == true.
+	index, upper := 0, i.n
+	for index < upper {
+		h := int(uint(index+upper) >> 1) // avoid overflow when computing h
+		// index ≤ h < upper
+
+		// TODO(jackson): Is Bytes.At or Bytes.Slice(Bytes.Offset(h),
+		// Bytes.Offset(h+1)) faster in this code?
+		c := bytes.Compare(key, i.r.separators.At(h))
+		if c > 0 {
+			index = h + 1 // preserves f(index-1) == false
+		} else {
+			upper = h // preserves f(upper) == true
+		}
+	}
+	// index == upper, f(index-1) == false, and f(upper) (= f(index)) == true  =>  answer is index.
+	i.row = index
+	return index < i.n
+}
+
+// First seeks index iterator to the first block entry. It returns false if the
+// index block is empty.
+func (i *IndexIter) First() bool {
+	i.row = 0
+	return i.n > 0
+}
+
+// Last seeks index iterator to the last block entry. It returns false if the
+// index block is empty.
+func (i *IndexIter) Last() bool {
+	i.row = i.n - 1
+	return i.n > 0
+}
+
+// Next steps the index iterator to the next block entry. It returns false if
+// the index block is exhausted.
+func (i *IndexIter) Next() bool {
+	i.row++
+	return i.row < i.n
+}
+
+// Prev steps the index iterator to the previous block entry. It returns false
+// if the index block is exhausted.
+func (i *IndexIter) Prev() bool {
+	i.row--
+	return i.row >= 0
+}
+
+// Close closes the iterator, releasing any resources it holds.
+func (i *IndexIter) Close() error {
+	i.h.Release()
+	*i = IndexIter{}
+	return nil
+}
diff --git a/sstable/colblk/prefix_bytes.go b/sstable/colblk/prefix_bytes.go
index 53ea8635..de44d8db 100644
--- a/sstable/colblk/prefix_bytes.go
+++ b/sstable/colblk/prefix_bytes.go
@@ -239,10 +239,12 @@ func (b PrefixBytes) At(i int) []byte {
 // within a PrefixBytes, avoiding unnecessary copying when portions of slices
 // are shared.
 type PrefixBytesIter struct {
-	UnsafeBuf
-	bundlePrefixLen       uint32
-	offsetIndex           int
-	nextBundleOffsetIndex int
+	// buf is used for materializing a user key. It is preallocated to the maximum
+	// key length in the data block.
+	buf                      []byte
+	sharedAndBundlePrefixLen uint32
+	offsetIndex              int
+	nextBundleOffsetIndex    int
 }
 
 // SetAt updates the provided PrefixBytesIter to hold the i'th []byte slice in
@@ -252,8 +254,8 @@ type PrefixBytesIter struct {
 func (b *PrefixBytes) SetAt(it *PrefixBytesIter, i int) {
 	// Determine the offset and length of the bundle prefix.
 	bundleOffsetIndex := b.bundleOffsetIndexForRow(i)
-	bundleOffsetStart := b.rawBytes.offsets.At(bundleOffsetIndex)
-	it.bundlePrefixLen = b.rawBytes.offsets.At(bundleOffsetIndex+1) - bundleOffsetStart
+	bundleOffsetStart, bundleOffsetEnd := b.rawBytes.offsets.At2(bundleOffsetIndex)
+	bundlePrefixLen := bundleOffsetEnd - bundleOffsetStart
 
 	// Determine the offset and length of the row's individual suffix.
 	it.offsetIndex = b.rowSuffixIndex(i)
@@ -262,25 +264,25 @@ func (b *PrefixBytes) SetAt(it *PrefixBytesIter, i int) {
 	// this recomputation? The expected case is non-duplicate keys, so it may
 	// not be worthwhile.
 	rowSuffixStart, rowSuffixEnd := b.rowSuffixOffsets(i, it.offsetIndex)
+	rowSuffixLen := rowSuffixEnd - rowSuffixStart
 
-	// Grow the size of the iterator's buffer if necessary.
-	it.len = b.sharedPrefixLen + int(it.bundlePrefixLen) + int(rowSuffixEnd-rowSuffixStart)
-	if it.len > it.cap {
-		panic(errors.AssertionFailedf("buffer too small: %d > %d", it.len, it.cap))
-	}
+	it.sharedAndBundlePrefixLen = uint32(b.sharedPrefixLen) + bundlePrefixLen
+	it.buf = it.buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
 
+	ptr := unsafe.Pointer(unsafe.SliceData(it.buf))
 	// Copy the shared key prefix.
-	memmove(it.ptr, b.rawBytes.data, uintptr(b.sharedPrefixLen))
+	memmove(ptr, b.rawBytes.data, uintptr(b.sharedPrefixLen))
 	// Copy the bundle prefix.
 	memmove(
-		unsafe.Pointer(uintptr(it.ptr)+uintptr(b.sharedPrefixLen)),
+		unsafe.Pointer(uintptr(ptr)+uintptr(b.sharedPrefixLen)),
 		unsafe.Pointer(uintptr(b.rawBytes.data)+uintptr(bundleOffsetStart)),
-		uintptr(it.bundlePrefixLen))
+		uintptr(bundlePrefixLen))
+
 	// Copy the per-row suffix.
 	memmove(
-		unsafe.Pointer(uintptr(it.ptr)+uintptr(b.sharedPrefixLen)+uintptr(it.bundlePrefixLen)),
+		unsafe.Pointer(uintptr(ptr)+uintptr(it.sharedAndBundlePrefixLen)),
 		unsafe.Pointer(uintptr(b.rawBytes.data)+uintptr(rowSuffixStart)),
-		uintptr(rowSuffixEnd-rowSuffixStart))
+		uintptr(rowSuffixLen))
 	// Set nextBundleOffsetIndex so that a call to SetNext can cheaply determine
 	// whether the next row is in the same bundle.
 	it.nextBundleOffsetIndex = bundleOffsetIndex + (1 << b.bundleShift) + 1
@@ -296,24 +298,21 @@ func (b *PrefixBytes) SetNext(it *PrefixBytesIter) {
 	// If the next row is in the same bundle, we can take a fast path of only
 	// updating the per-row suffix.
 	if it.offsetIndex < it.nextBundleOffsetIndex {
-		rowSuffixStart := b.rawBytes.offsets.At(it.offsetIndex)
-		rowSuffixEnd := b.rawBytes.offsets.At(it.offsetIndex + 1)
-		if rowSuffixStart == rowSuffixEnd {
+		rowSuffixStart, rowSuffixEnd := b.rawBytes.offsets.At2(it.offsetIndex)
+		rowSuffixLen := rowSuffixEnd - rowSuffixStart
+		if rowSuffixLen == 0 {
 			// The start and end offsets are equal, indicating that the key is a
 			// duplicate. Since it's identical to the previous key, there's
 			// nothing left to do, we can leave buf as-is.
 			return
 		}
-		// Grow the buffer if necessary.
-		it.len = b.sharedPrefixLen + int(it.bundlePrefixLen) + int(rowSuffixEnd-rowSuffixStart)
-		if it.len > it.cap {
-			panic(errors.AssertionFailedf("buffer too small: %d > %d", it.len, it.cap))
-		}
+		it.buf = it.buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
 		// Copy in the per-row suffix.
+		ptr := unsafe.Pointer(unsafe.SliceData(it.buf))
 		memmove(
-			unsafe.Pointer(uintptr(it.ptr)+uintptr(b.sharedPrefixLen)+uintptr(it.bundlePrefixLen)),
+			unsafe.Pointer(uintptr(ptr)+uintptr(it.sharedAndBundlePrefixLen)),
 			unsafe.Pointer(uintptr(b.rawBytes.data)+uintptr(rowSuffixStart)),
-			uintptr(rowSuffixEnd-rowSuffixStart))
+			uintptr(rowSuffixLen))
 		return
 	}
 
@@ -321,30 +320,28 @@ func (b *PrefixBytes) SetNext(it *PrefixBytesIter) {
 	// The offsetIndex is currently pointing to the start of the new bundle
 	// prefix. Increment it to point at the start of the new row suffix.
 	it.offsetIndex++
-	rowSuffixStart := b.rawBytes.offsets.At(it.offsetIndex)
-	rowSuffixEnd := b.rawBytes.offsets.At(it.offsetIndex + 1)
+	rowSuffixStart, rowSuffixEnd := b.rawBytes.offsets.At2(it.offsetIndex)
+	rowSuffixLen := rowSuffixEnd - rowSuffixStart
 
 	// Read the offsets of the new bundle prefix and update the index of the
 	// next bundle.
 	bundlePrefixStart := b.rawBytes.offsets.At(it.nextBundleOffsetIndex)
-	it.bundlePrefixLen = rowSuffixStart - bundlePrefixStart
+	bundlePrefixLen := rowSuffixStart - bundlePrefixStart
 	it.nextBundleOffsetIndex = it.offsetIndex + (1 << b.bundleShift)
 
-	// Grow the buffer if necessary.
-	it.len = b.sharedPrefixLen + int(it.bundlePrefixLen) + int(rowSuffixEnd-rowSuffixStart)
-	if it.len > it.cap {
-		panic(errors.AssertionFailedf("buffer too small: %d > %d", it.len, it.cap))
-	}
+	it.sharedAndBundlePrefixLen = uint32(b.sharedPrefixLen) + bundlePrefixLen
+	it.buf = it.buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
 	// Copy in the new bundle suffix.
+	ptr := unsafe.Pointer(unsafe.SliceData(it.buf))
 	memmove(
-		unsafe.Pointer(uintptr(it.ptr)+uintptr(b.sharedPrefixLen)),
+		unsafe.Pointer(uintptr(ptr)+uintptr(b.sharedPrefixLen)),
 		unsafe.Pointer(uintptr(b.rawBytes.data)+uintptr(bundlePrefixStart)),
-		uintptr(it.bundlePrefixLen))
+		uintptr(bundlePrefixLen))
 	// Copy in the per-row suffix.
 	memmove(
-		unsafe.Pointer(uintptr(it.ptr)+uintptr(b.sharedPrefixLen)+uintptr(it.bundlePrefixLen)),
+		unsafe.Pointer(uintptr(ptr)+uintptr(it.sharedAndBundlePrefixLen)),
 		unsafe.Pointer(uintptr(b.rawBytes.data)+uintptr(rowSuffixStart)),
-		uintptr(rowSuffixEnd-rowSuffixStart))
+		uintptr(rowSuffixLen))
 }
 
 // SharedPrefix return a []byte of the shared prefix that was extracted from
@@ -386,8 +383,7 @@ func (b *PrefixBytes) RowSuffix(row int) []byte {
 func (b *PrefixBytes) rowSuffixOffsets(row, i int) (low uint32, high uint32) {
 	// Retrieve the low and high offsets indicating the start and end of the
 	// row's suffix slice.
-	low = b.rawBytes.offsets.At(i)
-	high = b.rawBytes.offsets.At(i + 1)
+	low, high = b.rawBytes.offsets.At2(i)
 	// If there's a non-empty slice for the row, this row is different than its
 	// predecessor.
 	if low != high {
diff --git a/sstable/colblk/unsafe_slice.go b/sstable/colblk/unsafe_slice.go
index bc3b3966..c8c63e8a 100644
--- a/sstable/colblk/unsafe_slice.go
+++ b/sstable/colblk/unsafe_slice.go
@@ -92,23 +92,27 @@ func makeUnsafeUints(base uint64, ptr unsafe.Pointer, width int) UnsafeUints {
 
 // At returns the `i`-th element.
 func (s UnsafeUints) At(i int) uint64 {
-	// TODO(jackson): Experiment with other alternatives that might be faster
-	// and avoid switching on the width.
-	switch s.width {
-	case 0:
-		return s.base
-	case 1:
-		return s.base + uint64(*(*uint8)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i))))
-	case 2:
-		return s.base + uint64(*(*uint16)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align16Shift)))
-	case 4:
-		return s.base + uint64(*(*uint32)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align32Shift)))
-	default:
+	// TODO(radu): this implementation assumes little-endian architecture.
+
+	// One of the most common case is decoding timestamps, which require the full
+	// 8 bytes (2^32 nanoseconds is only ~4 seconds).
+	if s.width == 8 {
 		// NB: The slice encodes 64-bit integers, there is no base (it doesn't save
 		// any bits to compute a delta). We cast directly into a *uint64 pointer and
 		// don't add the base.
 		return *(*uint64)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align64Shift))
 	}
+	// Another common case is 0 width, when all keys have zero logical timestamps.
+	if s.width == 0 {
+		return s.base
+	}
+	if s.width == 4 {
+		return s.base + uint64(*(*uint32)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align32Shift)))
+	}
+	if s.width == 2 {
+		return s.base + uint64(*(*uint16)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align16Shift)))
+	}
+	return s.base + uint64(*(*uint8)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i))))
 }
 
 // UnsafeOffsets is a specialization of UnsafeInts (providing the same
@@ -135,6 +139,8 @@ func DecodeUnsafeOffsets(b []byte, off uint32, rows int) (_ UnsafeOffsets, endOf
 
 // At returns the `i`-th offset.
 func (s UnsafeOffsets) At(i int) uint32 {
+	// TODO(radu): this implementation assumes little-endian architecture.
+
 	// We expect offsets to be encoded as 16-bit integers in most cases.
 	if s.width == 2 {
 		return uint32(*(*uint16)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align16Shift)))
@@ -148,6 +154,26 @@ func (s UnsafeOffsets) At(i int) uint32 {
 	return *(*uint32)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align32Shift))
 }
 
+// At2 returns the `i`-th and `i+1`-th offsets.
+func (s UnsafeOffsets) At2(i int) (uint32, uint32) {
+	// TODO(radu): this implementation assumes little-endian architecture.
+
+	// We expect offsets to be encoded as 16-bit integers in most cases.
+	if s.width == 2 {
+		v := *(*uint32)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align16Shift))
+		return v & 0xFFFF, v >> 16
+	}
+	if s.width <= 1 {
+		if s.width == 0 {
+			return 0, 0
+		}
+		v := *(*uint16)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)))
+		return uint32(v & 0xFF), uint32(v >> 8)
+	}
+	v := *(*uint64)(unsafe.Pointer(uintptr(s.ptr) + uintptr(i)<<align32Shift))
+	return uint32(v), uint32(v >> 32)
+}
+
 // UnsafeBuf provides a buffer without bounds checking. Every buf has a len and
 // capacity.
 type UnsafeBuf struct {
diff --git a/sstable/reader_iter.go b/sstable/reader_iter.go
index 52904fbe..8fb14895 100644
--- a/sstable/reader_iter.go
+++ b/sstable/reader_iter.go
@@ -111,48 +111,52 @@ type Iterator interface {
 // TODO(sumeer): remove the aforementioned defensive code.
 
 var (
-	singleLevelIterRowBlockPool sync.Pool // *singleLevelIterator[rowblk.Iter, *rowblk.Iter]
-	twoLevelIterRowBlockPool    sync.Pool // *twoLevelIterator[rowblk.Iter, *rowblk.Iter]
+	singleLevelIterRowBlockPool sync.Pool // *singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]
+	twoLevelIterRowBlockPool    sync.Pool // *twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]
 )
 
 func init() {
 	singleLevelIterRowBlockPool = sync.Pool{
 		New: func() interface{} {
-			i := &singleLevelIterator[rowblk.Iter, *rowblk.Iter]{pool: &singleLevelIterRowBlockPool}
+			i := &singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]{pool: &singleLevelIterRowBlockPool}
 			// Note: this is a no-op if invariants are disabled or race is enabled.
-			invariants.SetFinalizer(i, checkSingleLevelIterator[rowblk.Iter, *rowblk.Iter])
+			invariants.SetFinalizer(i, checkSingleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])
 			return i
 		},
 	}
 	twoLevelIterRowBlockPool = sync.Pool{
 		New: func() interface{} {
-			i := &twoLevelIterator[rowblk.Iter, *rowblk.Iter]{pool: &twoLevelIterRowBlockPool}
+			i := &twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]{pool: &twoLevelIterRowBlockPool}
 			// Note: this is a no-op if invariants are disabled or race is enabled.
-			invariants.SetFinalizer(i, checkTwoLevelIterator[rowblk.Iter, *rowblk.Iter])
+			invariants.SetFinalizer(i, checkTwoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])
 			return i
 		},
 	}
 }
 
-func checkSingleLevelIterator[D any, PD block.DataBlockIterator[D]](obj interface{}) {
-	i := obj.(*singleLevelIterator[D, PD])
+func checkSingleLevelIterator[I any, PI block.IndexBlockIterator[I], D any, PD block.DataBlockIterator[D]](
+	obj interface{},
+) {
+	i := obj.(*singleLevelIterator[I, PI, D, PD])
 	if p := PD(&i.data).Handle().Get(); p != nil {
 		fmt.Fprintf(os.Stderr, "singleLevelIterator.data.handle is not nil: %p\n", p)
 		os.Exit(1)
 	}
-	if p := i.index.Handle().Get(); p != nil {
+	if p := PI(&i.index).Handle().Get(); p != nil {
 		fmt.Fprintf(os.Stderr, "singleLevelIterator.index.handle is not nil: %p\n", p)
 		os.Exit(1)
 	}
 }
 
-func checkTwoLevelIterator[D any, PD block.DataBlockIterator[D]](obj interface{}) {
-	i := obj.(*twoLevelIterator[D, PD])
+func checkTwoLevelIterator[I any, PI block.IndexBlockIterator[I], D any, PD block.DataBlockIterator[D]](
+	obj interface{},
+) {
+	i := obj.(*twoLevelIterator[I, PI, D, PD])
 	if p := PD(&i.secondLevel.data).Handle().Get(); p != nil {
 		fmt.Fprintf(os.Stderr, "singleLevelIterator.data.handle is not nil: %p\n", p)
 		os.Exit(1)
 	}
-	if p := i.secondLevel.index.Handle().Get(); p != nil {
+	if p := PI(&i.secondLevel.index).Handle().Get(); p != nil {
 		fmt.Fprintf(os.Stderr, "singleLevelIterator.index.handle is not nil: %p\n", p)
 		os.Exit(1)
 	}
diff --git a/sstable/reader_iter_single_lvl.go b/sstable/reader_iter_single_lvl.go
index 284152ff..d1f63ba6 100644
--- a/sstable/reader_iter_single_lvl.go
+++ b/sstable/reader_iter_single_lvl.go
@@ -25,20 +25,21 @@ import (
 // key, it first looks in the index for the block that contains that key, and then
 // looks inside that block.
 //
-// singleLevelIterator is parameterized by the type of the data block iterator.
-// The type parameters are designed to allow the singleLevelIterator to embed
-// the data block iterator struct within itself, avoiding an extra allocation
-// and pointer indirection. The complication comes from the fact that we want to
-// implement the interface on a pointer receiver but embed the non-pointer type
-// within the struct. The D type parameter is the non-pointer data block
-// iterator type, and the PD type parameter is the *D type that actually
-// implements the DataBlockIterator constraint.
+// singleLevelIterator is parameterized by the type of the data block iterator
+// and index block iterator.  The type parameters are designed to allow the
+// singleLevelIterator to embed the data block and index block iterator structs
+// within itself, avoiding an extra allocation and pointer indirection. The
+// complication comes from the fact that we want to implement the interfaces on
+// pointer receivers but embed the non-pointer types within the struct. The D
+// and I type parameters are the non-pointer data and index block iterator
+// types, and the PD and PI type parameters are the *D and *I types that
+// actually implement the DataBlockIterator and IndexBlockIterator constraints.
 //
-// Unfortunately, uses of the [data] field must explicitly cast &data to the PD
-// type in order to access its interface methods. This pattern is taken from the
-// Go generics proposal:
+// Unfortunately, uses of the [data] and [index] fields must explicitly cast
+// &data/&index to the PD/PI type in order to access its interface methods. This
+// pattern is taken from the Go generics proposal:
 // https://go.googlesource.com/proposal/+/refs/heads/master/design/43651-type-parameters.md#pointer-method-example
-type singleLevelIterator[D any, PD block.DataBlockIterator[D]] struct {
+type singleLevelIterator[I any, PI block.IndexBlockIterator[I], D any, PD block.DataBlockIterator[D]] struct {
 	ctx context.Context
 	cmp Compare
 	// Global lower/upper bound for the iterator.
@@ -56,7 +57,7 @@ type singleLevelIterator[D any, PD block.DataBlockIterator[D]] struct {
 	// endKeyInclusive is set to force the iterator to treat the upper field as
 	// inclusive while iterating instead of exclusive.
 	endKeyInclusive       bool
-	index                 rowblk.IndexIter
+	index                 I
 	indexFilterRH         objstorage.ReadHandle
 	indexFilterRHPrealloc objstorageprovider.PreallocatedReadHandle
 	data                  D
@@ -195,7 +196,7 @@ type singleLevelIterator[D any, PD block.DataBlockIterator[D]] struct {
 }
 
 // singleLevelIterator implements the base.InternalIterator interface.
-var _ base.InternalIterator = (*singleLevelIterator[rowblk.Iter, *rowblk.Iter])(nil)
+var _ base.InternalIterator = (*singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])(nil)
 
 // newRowBlockSingleLevelIterator reads the index block and creates and
 // initializes a singleLevelIterator over an sstable with row-oriented data
@@ -217,13 +218,13 @@ func newRowBlockSingleLevelIterator(
 	statsCollector *CategoryStatsCollector,
 	rp ReaderProvider,
 	bufferPool *block.BufferPool,
-) (*singleLevelIterator[rowblk.Iter, *rowblk.Iter], error) {
+) (*singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter], error) {
 	if r.err != nil {
 		return nil, r.err
 	}
 	// TODO(jackson): When we have a columnar-block sstable format, assert that
 	// the table format is row-oriented.
-	i := singleLevelIterRowBlockPool.Get().(*singleLevelIterator[rowblk.Iter, *rowblk.Iter])
+	i := singleLevelIterRowBlockPool.Get().(*singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])
 	useFilterBlock := shouldUseFilterBlock(r, filterBlockSizeLimit)
 	i.init(
 		ctx, r, v, transforms, lower, upper, filterer, useFilterBlock,
@@ -264,7 +265,7 @@ func newRowBlockSingleLevelIterator(
 }
 
 // init initializes the singleLevelIterator struct. It does not read the index.
-func (i *singleLevelIterator[D, PD]) init(
+func (i *singleLevelIterator[I, PI, D, PD]) init(
 	ctx context.Context,
 	r *Reader,
 	v *virtualState,
@@ -302,7 +303,7 @@ func (i *singleLevelIterator[D, PD]) init(
 }
 
 // Helper function to check if keys returned from iterator are within virtual bounds.
-func (i *singleLevelIterator[D, PD]) maybeVerifyKey(kv *base.InternalKV) *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) maybeVerifyKey(kv *base.InternalKV) *base.InternalKV {
 	if invariants.Enabled && kv != nil && i.vState != nil {
 		key := kv.K.UserKey
 		v := i.vState
@@ -317,23 +318,23 @@ func (i *singleLevelIterator[D, PD]) maybeVerifyKey(kv *base.InternalKV) *base.I
 
 // SetupForCompaction sets up the singleLevelIterator for use with compactionIter.
 // Currently, it skips readahead ramp-up. It should be called after init is called.
-func (i *singleLevelIterator[D, PD]) SetupForCompaction() {
+func (i *singleLevelIterator[I, PI, D, PD]) SetupForCompaction() {
 	i.dataRH.SetupForCompaction()
 	if i.vbRH != nil {
 		i.vbRH.SetupForCompaction()
 	}
 }
 
-func (i *singleLevelIterator[D, PD]) resetForReuse() singleLevelIterator[D, PD] {
-	return singleLevelIterator[D, PD]{
-		index:  i.index.ResetForReuse(),
+func (i *singleLevelIterator[I, PI, D, PD]) resetForReuse() singleLevelIterator[I, PI, D, PD] {
+	return singleLevelIterator[I, PI, D, PD]{
+		index:  PI(&i.index).ResetForReuse(),
 		data:   PD(&i.data).ResetForReuse(),
 		pool:   i.pool,
 		inPool: true,
 	}
 }
 
-func (i *singleLevelIterator[D, PD]) initBounds() {
+func (i *singleLevelIterator[I, PI, D, PD]) initBounds() {
 	// Trim the iteration bounds for the current block. We don't have to check
 	// the bounds on each iteration if the block is entirely contained within the
 	// iteration bounds.
@@ -349,7 +350,7 @@ func (i *singleLevelIterator[D, PD]) initBounds() {
 	}
 	i.blockUpper = i.upper
 	// TODO(radu): this should be >= 0 if blockUpper is inclusive.
-	if i.blockUpper != nil && i.cmp(i.blockUpper, i.index.Separator()) > 0 {
+	if i.blockUpper != nil && i.cmp(i.blockUpper, PI(&i.index).Separator()) > 0 {
 		// The upper-bound is greater than the index key which itself is greater
 		// than or equal to every key in the block. No need to check the
 		// upper-bound again for this block. Even if blockUpper is inclusive
@@ -359,7 +360,7 @@ func (i *singleLevelIterator[D, PD]) initBounds() {
 	}
 }
 
-func (i *singleLevelIterator[D, PD]) initBoundsForAlreadyLoadedBlock() {
+func (i *singleLevelIterator[I, PI, D, PD]) initBoundsForAlreadyLoadedBlock() {
 	// TODO(radu): determine automatically if we need to call First or not and
 	// unify this function with initBounds().
 	if PD(&i.data).FirstUserKey() == nil {
@@ -377,7 +378,7 @@ func (i *singleLevelIterator[D, PD]) initBoundsForAlreadyLoadedBlock() {
 	}
 	i.blockUpper = i.upper
 	// TODO(radu): this should be >= 0 if blockUpper is inclusive.
-	if i.blockUpper != nil && i.cmp(i.blockUpper, i.index.Separator()) > 0 {
+	if i.blockUpper != nil && i.cmp(i.blockUpper, PI(&i.index).Separator()) > 0 {
 		// The upper-bound is greater than the index key which itself is greater
 		// than or equal to every key in the block. No need to check the
 		// upper-bound again for this block.
@@ -416,13 +417,13 @@ var ensureBoundsOptDeterminism bool
 // synthetic prefix in order to combine them with the vState bounds. Thus, if
 // this iterator knows bounds will be passed to vState, it can signal that it
 // they should be passed without being rewritten to skip converting to and fro.
-func (i singleLevelIterator[P, PD]) SetBoundsWithSyntheticPrefix() bool {
+func (i singleLevelIterator[I, PI, P, PD]) SetBoundsWithSyntheticPrefix() bool {
 	return i.vState != nil
 }
 
 // SetBounds implements internalIterator.SetBounds, as documented in the pebble
 // package. Note that the upper field is exclusive.
-func (i *singleLevelIterator[P, PD]) SetBounds(lower, upper []byte) {
+func (i *singleLevelIterator[I, PI, P, PD]) SetBounds(lower, upper []byte) {
 	i.boundsCmp = 0
 	if i.vState != nil {
 		// If the reader is constructed for a virtual sstable, then we must
@@ -457,22 +458,22 @@ func (i *singleLevelIterator[P, PD]) SetBounds(lower, upper []byte) {
 	i.blockUpper = nil
 }
 
-func (i *singleLevelIterator[P, PD]) SetContext(ctx context.Context) {
+func (i *singleLevelIterator[I, PI, P, PD]) SetContext(ctx context.Context) {
 	i.ctx = ctx
 }
 
 // loadBlock loads the block at the current index position and leaves i.data
 // unpositioned. If unsuccessful, it sets i.err to any error encountered, which
 // may be nil if we have simply exhausted the entire table.
-func (i *singleLevelIterator[P, PD]) loadBlock(dir int8) loadBlockResult {
-	if !i.index.Valid() {
+func (i *singleLevelIterator[I, PI, P, PD]) loadBlock(dir int8) loadBlockResult {
+	if !PI(&i.index).Valid() {
 		// Ensure the data block iterator is invalidated even if loading of the block
 		// fails.
 		PD(&i.data).Invalidate()
 		return loadBlockFailed
 	}
 	// Load the next block.
-	bhp, err := i.index.BlockHandleWithProperties()
+	bhp, err := PI(&i.index).BlockHandleWithProperties()
 	if i.dataBH == bhp.Handle && PD(&i.data).Valid() {
 		// We're already at the data block we want to load. Reset bounds in case
 		// they changed since the last seek, but don't reload the block from cache
@@ -525,7 +526,7 @@ func (i *singleLevelIterator[P, PD]) loadBlock(dir int8) loadBlockResult {
 
 // readBlockForVBR implements the blockProviderWhenOpen interface for use by
 // the valueBlockReader.
-func (i *singleLevelIterator[D, PD]) readBlockForVBR(
+func (i *singleLevelIterator[I, PI, D, PD]) readBlockForVBR(
 	h block.Handle, stats *base.InternalIteratorStats,
 ) (block.BufferHandle, error) {
 	ctx := objiotracing.WithBlockType(i.ctx, objiotracing.ValueBlock)
@@ -537,7 +538,7 @@ func (i *singleLevelIterator[D, PD]) readBlockForVBR(
 // fall within the filter's current bounds.  This function consults the
 // apprioriate bound, depending on the iteration direction, and returns either
 // `blockIntersects` or `blockExcluded`.
-func (i *singleLevelIterator[D, PD]) resolveMaybeExcluded(dir int8) intersectsResult {
+func (i *singleLevelIterator[I, PI, D, PD]) resolveMaybeExcluded(dir int8) intersectsResult {
 	// TODO(jackson): We could first try comparing to top-level index block's
 	// key, and if within bounds avoid per-data block key comparisons.
 
@@ -559,7 +560,7 @@ func (i *singleLevelIterator[D, PD]) resolveMaybeExcluded(dir int8) intersectsRe
 	// need.
 	if dir > 0 {
 		// Forward iteration.
-		if i.bpfs.boundLimitedFilter.KeyIsWithinUpperBound(i.index.Separator()) {
+		if i.bpfs.boundLimitedFilter.KeyIsWithinUpperBound(PI(&i.index).Separator()) {
 			return blockExcluded
 		}
 		return blockIntersects
@@ -583,19 +584,19 @@ func (i *singleLevelIterator[D, PD]) resolveMaybeExcluded(dir int8) intersectsRe
 	// previous block's separator, which provides an inclusive lower bound on
 	// the original block's keys. Afterwards, we step forward to restore our
 	// index position.
-	if !i.index.Prev() {
+	if !PI(&i.index).Prev() {
 		// The original block points to the first block of this index block. If
 		// there's a two-level index, it could potentially provide a lower
 		// bound, but the code refactoring necessary to read it doesn't seem
 		// worth the payoff. We fall through to loading the block.
-	} else if i.bpfs.boundLimitedFilter.KeyIsWithinLowerBound(i.index.Separator()) {
+	} else if i.bpfs.boundLimitedFilter.KeyIsWithinLowerBound(PI(&i.index).Separator()) {
 		// The lower-bound on the original block falls within the filter's
 		// bounds, and we can skip the block (after restoring our current index
 		// position).
-		_ = i.index.Next()
+		_ = PI(&i.index).Next()
 		return blockExcluded
 	}
-	_ = i.index.Next()
+	_ = PI(&i.index).Next()
 	return blockIntersects
 }
 
@@ -605,7 +606,7 @@ func (i *singleLevelIterator[D, PD]) resolveMaybeExcluded(dir int8) intersectsRe
 // seeks for a particular iterator.
 const numStepsBeforeSeek = 4
 
-func (i *singleLevelIterator[D, PD]) trySeekGEUsingNextWithinBlock(
+func (i *singleLevelIterator[I, PI, D, PD]) trySeekGEUsingNextWithinBlock(
 	key []byte,
 ) (kv *base.InternalKV, done bool) {
 	kv = PD(&i.data).KV()
@@ -629,7 +630,7 @@ func (i *singleLevelIterator[D, PD]) trySeekGEUsingNextWithinBlock(
 	return kv, false
 }
 
-func (i *singleLevelIterator[D, PD]) trySeekLTUsingPrevWithinBlock(
+func (i *singleLevelIterator[I, PI, D, PD]) trySeekLTUsingPrevWithinBlock(
 	key []byte,
 ) (kv *base.InternalKV, done bool) {
 	kv = PD(&i.data).KV()
@@ -653,7 +654,9 @@ func (i *singleLevelIterator[D, PD]) trySeekLTUsingPrevWithinBlock(
 // SeekGE implements internalIterator.SeekGE, as documented in the pebble
 // package. Note that SeekGE only checks the upper bound. It is up to the
 // caller to ensure that key is greater than or equal to the lower bound.
-func (i *singleLevelIterator[D, PD]) SeekGE(key []byte, flags base.SeekGEFlags) *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) SeekGE(
+	key []byte, flags base.SeekGEFlags,
+) *base.InternalKV {
 	if i.vState != nil {
 		// Callers of SeekGE don't know about virtual sstable bounds, so we may
 		// have to internally restrict the bounds.
@@ -694,7 +697,7 @@ func (i *singleLevelIterator[D, PD]) SeekGE(key []byte, flags base.SeekGEFlags)
 }
 
 // seekGEHelper contains the common functionality for SeekGE and SeekPrefixGE.
-func (i *singleLevelIterator[D, PD]) seekGEHelper(
+func (i *singleLevelIterator[I, PI, D, PD]) seekGEHelper(
 	key []byte, boundsCmp int, flags base.SeekGEFlags,
 ) *base.InternalKV {
 	// Invariant: trySeekUsingNext => !i.data.isDataInvalidated() && i.exhaustedBounds != +1
@@ -703,8 +706,8 @@ func (i *singleLevelIterator[D, PD]) seekGEHelper(
 	// by trySeekUsingNext, or by monotonically increasing bounds (i.boundsCmp).
 
 	var dontSeekWithinBlock bool
-	if !PD(&i.data).IsDataInvalidated() && PD(&i.data).Valid() && i.index.Valid() &&
-		boundsCmp > 0 && i.cmp(key, i.index.Separator()) <= 0 {
+	if !PD(&i.data).IsDataInvalidated() && PD(&i.data).Valid() && PI(&i.index).Valid() &&
+		boundsCmp > 0 && i.cmp(key, PI(&i.index).Separator()) <= 0 {
 		// Fast-path: The bounds have moved forward and this SeekGE is
 		// respecting the lower bound (guaranteed by Iterator). We know that
 		// the iterator must already be positioned within or just outside the
@@ -758,7 +761,7 @@ func (i *singleLevelIterator[D, PD]) seekGEHelper(
 
 		// Slow-path.
 
-		if !i.index.SeekGE(key) {
+		if !PI(&i.index).SeekGE(key) {
 			// The target key is greater than any key in the index block.
 			// Invalidate the block iterator so that a subsequent call to Prev()
 			// will return the last key in the table.
@@ -777,7 +780,7 @@ func (i *singleLevelIterator[D, PD]) seekGEHelper(
 			// multiple blocks. If upper is exclusive we use >= below, else
 			// we use >.
 			if i.upper != nil {
-				cmp := i.cmp(i.index.Separator(), i.upper)
+				cmp := i.cmp(PI(&i.index).Separator(), i.upper)
 				if (!i.endKeyInclusive && cmp >= 0) || cmp > 0 {
 					i.exhaustedBounds = +1
 					return nil
@@ -805,7 +808,7 @@ func (i *singleLevelIterator[D, PD]) seekGEHelper(
 // SeekPrefixGE implements internalIterator.SeekPrefixGE, as documented in the
 // pebble package. Note that SeekPrefixGE only checks the upper bound. It is up
 // to the caller to ensure that key is greater than or equal to the lower bound.
-func (i *singleLevelIterator[D, PD]) SeekPrefixGE(
+func (i *singleLevelIterator[I, PI, D, PD]) SeekPrefixGE(
 	prefix, key []byte, flags base.SeekGEFlags,
 ) *base.InternalKV {
 	if i.vState != nil {
@@ -821,7 +824,7 @@ func (i *singleLevelIterator[D, PD]) SeekPrefixGE(
 	return i.seekPrefixGE(prefix, key, flags)
 }
 
-func (i *singleLevelIterator[D, PD]) seekPrefixGE(
+func (i *singleLevelIterator[I, PI, D, PD]) seekPrefixGE(
 	prefix, key []byte, flags base.SeekGEFlags,
 ) (kv *base.InternalKV) {
 	// NOTE: prefix is only used for bloom filter checking and not later work in
@@ -884,7 +887,7 @@ func shouldUseFilterBlock(reader *Reader, filterBlockSizeLimit FilterBlockSizeLi
 	return reader.tableFilter != nil && reader.filterBH.Length <= uint64(filterBlockSizeLimit)
 }
 
-func (i *singleLevelIterator[D, PD]) bloomFilterMayContain(prefix []byte) (bool, error) {
+func (i *singleLevelIterator[I, PI, D, PD]) bloomFilterMayContain(prefix []byte) (bool, error) {
 	// Check prefix bloom filter.
 	prefixToCheck := prefix
 	if i.transforms.SyntheticPrefix.IsSet() {
@@ -906,7 +909,7 @@ func (i *singleLevelIterator[D, PD]) bloomFilterMayContain(prefix []byte) (bool,
 }
 
 // virtualLast should only be called if i.vReader != nil.
-func (i *singleLevelIterator[D, PD]) virtualLast() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) virtualLast() *base.InternalKV {
 	if i.vState == nil {
 		panic("pebble: invalid call to virtualLast")
 	}
@@ -922,7 +925,7 @@ func (i *singleLevelIterator[D, PD]) virtualLast() *base.InternalKV {
 // virtualLast. Consider generalizing this into a SeekLE() if there are other
 // uses of this method in the future. Does a SeekLE on the upper bound of the
 // file/iterator.
-func (i *singleLevelIterator[D, PD]) virtualLastSeekLE() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) virtualLastSeekLE() *base.InternalKV {
 	// Callers of SeekLE don't know about virtual sstable bounds, so we may
 	// have to internally restrict the bounds.
 	//
@@ -940,7 +943,7 @@ func (i *singleLevelIterator[D, PD]) virtualLastSeekLE() *base.InternalKV {
 	i.boundsCmp = 0
 	i.positionedUsingLatestBounds = true
 
-	indexOk := i.index.SeekGE(key)
+	indexOk := PI(&i.index).SeekGE(key)
 	// We can have multiple internal keys with the same user key as the seek
 	// key. In that case, we want the last (greatest) internal key.
 	//
@@ -960,8 +963,8 @@ func (i *singleLevelIterator[D, PD]) virtualLastSeekLE() *base.InternalKV {
 	// for now.
 	// TODO(jackson): Consider implementing SeekLE since it's easier to do in
 	// colblk.
-	for indexOk && bytes.Equal(i.index.Separator(), key) {
-		indexOk = i.index.Next()
+	for indexOk && bytes.Equal(PI(&i.index).Separator(), key) {
+		indexOk = PI(&i.index).Next()
 	}
 	if !indexOk {
 		// Cases A or B1 where B1 exhausted all blocks. In both cases the last block
@@ -1008,7 +1011,9 @@ func (i *singleLevelIterator[D, PD]) virtualLastSeekLE() *base.InternalKV {
 // SeekLT implements internalIterator.SeekLT, as documented in the pebble
 // package. Note that SeekLT only checks the lower bound. It is up to the
 // caller to ensure that key is less than or equal to the upper bound.
-func (i *singleLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags) *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) SeekLT(
+	key []byte, flags base.SeekLTFlags,
+) *base.InternalKV {
 	if i.vState != nil {
 		// Might have to fix upper bound since virtual sstable bounds are not
 		// known to callers of SeekLT.
@@ -1036,7 +1041,7 @@ func (i *singleLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags)
 	i.positionedUsingLatestBounds = true
 
 	var dontSeekWithinBlock bool
-	if !PD(&i.data).IsDataInvalidated() && PD(&i.data).Valid() && i.index.Valid() &&
+	if !PD(&i.data).IsDataInvalidated() && PD(&i.data).Valid() && PI(&i.index).Valid() &&
 		boundsCmp < 0 && i.cmp(PD(&i.data).FirstUserKey(), key) < 0 {
 		// Fast-path: The bounds have moved backward, and this SeekLT is
 		// respecting the upper bound (guaranteed by Iterator). We know that
@@ -1063,8 +1068,8 @@ func (i *singleLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags)
 		// externally ensured that the filter is disabled (through returning
 		// Intersects=false irrespective of the block props provided) during
 		// seeks.
-		if !i.index.SeekGE(key) {
-			if !i.index.Last() {
+		if !PI(&i.index).SeekGE(key) {
+			if !PI(&i.index).Last() {
 				return nil
 			}
 		}
@@ -1079,7 +1084,7 @@ func (i *singleLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags)
 			// that the previous block starts with keys <= ikey.UserKey since
 			// even though this is the current block's separator, the same
 			// user key can span multiple blocks.
-			if i.lower != nil && i.cmp(i.index.Separator(), i.lower) < 0 {
+			if i.lower != nil && i.cmp(PI(&i.index).Separator(), i.lower) < 0 {
 				i.exhaustedBounds = -1
 				return nil
 			}
@@ -1114,7 +1119,7 @@ func (i *singleLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags)
 // package. Note that First only checks the upper bound. It is up to the caller
 // to ensure that key is greater than or equal to the lower bound (e.g. via a
 // call to SeekGE(lower)).
-func (i *singleLevelIterator[D, PD]) First() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) First() *base.InternalKV {
 	// If we have a lower bound, use SeekGE. Note that in general this is not
 	// supported usage, except when the lower bound is there because the table is
 	// virtual.
@@ -1131,13 +1136,13 @@ func (i *singleLevelIterator[D, PD]) First() *base.InternalKV {
 // index file, or for positioning in the second-level index in a two-level
 // index file. For the latter, one cannot make any claims about absolute
 // positioning.
-func (i *singleLevelIterator[D, PD]) firstInternal() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) firstInternal() *base.InternalKV {
 	i.exhaustedBounds = 0
 	i.err = nil // clear cached iteration error
 	// Seek optimization only applies until iterator is first positioned after SetBounds.
 	i.boundsCmp = 0
 
-	if !i.index.First() {
+	if !PI(&i.index).First() {
 		PD(&i.data).Invalidate()
 		return nil
 	}
@@ -1165,7 +1170,7 @@ func (i *singleLevelIterator[D, PD]) firstInternal() *base.InternalKV {
 		// same user key can span multiple blocks. If upper is exclusive we
 		// use >= below, else we use >.
 		if i.upper != nil {
-			cmp := i.cmp(i.index.Separator(), i.upper)
+			cmp := i.cmp(PI(&i.index).Separator(), i.upper)
 			if (!i.endKeyInclusive && cmp >= 0) || cmp > 0 {
 				i.exhaustedBounds = +1
 				return nil
@@ -1181,7 +1186,7 @@ func (i *singleLevelIterator[D, PD]) firstInternal() *base.InternalKV {
 // package. Note that Last only checks the lower bound. It is up to the caller
 // to ensure that key is less than the upper bound (e.g. via a call to
 // SeekLT(upper))
-func (i *singleLevelIterator[D, PD]) Last() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) Last() *base.InternalKV {
 	if i.vState != nil {
 		return i.maybeVerifyKey(i.virtualLast())
 	}
@@ -1197,13 +1202,13 @@ func (i *singleLevelIterator[D, PD]) Last() *base.InternalKV {
 // index file, or for positioning in the second-level index in a two-level
 // index file. For the latter, one cannot make any claims about absolute
 // positioning.
-func (i *singleLevelIterator[D, PD]) lastInternal() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) lastInternal() *base.InternalKV {
 	i.exhaustedBounds = 0
 	i.err = nil // clear cached iteration error
 	// Seek optimization only applies until iterator is first positioned after SetBounds.
 	i.boundsCmp = 0
 
-	if !i.index.Last() {
+	if !PI(&i.index).Last() {
 		PD(&i.data).Invalidate()
 		return nil
 	}
@@ -1226,7 +1231,7 @@ func (i *singleLevelIterator[D, PD]) lastInternal() *base.InternalKV {
 		// already exceeded. Note that the previous block starts with keys <=
 		// key.UserKey since even though this is the current block's
 		// separator, the same user key can span multiple blocks.
-		if i.lower != nil && i.cmp(i.index.Separator(), i.lower) < 0 {
+		if i.lower != nil && i.cmp(PI(&i.index).Separator(), i.lower) < 0 {
 			i.exhaustedBounds = -1
 			return nil
 		}
@@ -1239,7 +1244,7 @@ func (i *singleLevelIterator[D, PD]) lastInternal() *base.InternalKV {
 // package.
 // Note: compactionIterator.Next mirrors the implementation of Iterator.Next
 // due to performance. Keep the two in sync.
-func (i *singleLevelIterator[D, PD]) Next() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) Next() *base.InternalKV {
 	if i.exhaustedBounds == +1 {
 		panic("Next called even though exhausted upper bound")
 	}
@@ -1266,7 +1271,7 @@ func (i *singleLevelIterator[D, PD]) Next() *base.InternalKV {
 }
 
 // NextPrefix implements (base.InternalIterator).NextPrefix.
-func (i *singleLevelIterator[D, PD]) NextPrefix(succKey []byte) *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) NextPrefix(succKey []byte) *base.InternalKV {
 	if i.exhaustedBounds == +1 {
 		panic("NextPrefix called even though exhausted upper bound")
 	}
@@ -1291,16 +1296,16 @@ func (i *singleLevelIterator[D, PD]) NextPrefix(succKey []byte) *base.InternalKV
 	// Did not find prefix in the existing data block. This is the slow-path
 	// where we effectively seek the iterator.
 	// The key is likely to be in the next data block, so try one step.
-	if !i.index.Next() {
+	if !PI(&i.index).Next() {
 		// The target key is greater than any key in the index block.
 		// Invalidate the block iterator so that a subsequent call to Prev()
 		// will return the last key in the table.
 		PD(&i.data).Invalidate()
 		return nil
 	}
-	if i.cmp(succKey, i.index.Separator()) > 0 {
+	if i.cmp(succKey, PI(&i.index).Separator()) > 0 {
 		// Not in the next data block, so seek the index.
-		if !i.index.SeekGE(succKey) {
+		if !PI(&i.index).SeekGE(succKey) {
 			// The target key is greater than any key in the index block.
 			// Invalidate the block iterator so that a subsequent call to Prev()
 			// will return the last key in the table.
@@ -1320,7 +1325,7 @@ func (i *singleLevelIterator[D, PD]) NextPrefix(succKey []byte) *base.InternalKV
 		// multiple blocks. If upper is exclusive we use >= below, else we use
 		// >.
 		if i.upper != nil {
-			cmp := i.cmp(i.index.Separator(), i.upper)
+			cmp := i.cmp(PI(&i.index).Separator(), i.upper)
 			if (!i.endKeyInclusive && cmp >= 0) || cmp > 0 {
 				i.exhaustedBounds = +1
 				return nil
@@ -1342,7 +1347,7 @@ func (i *singleLevelIterator[D, PD]) NextPrefix(succKey []byte) *base.InternalKV
 
 // Prev implements internalIterator.Prev, as documented in the pebble
 // package.
-func (i *singleLevelIterator[D, PD]) Prev() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) Prev() *base.InternalKV {
 	if i.exhaustedBounds == -1 {
 		panic("Prev called even though exhausted lower bound")
 	}
@@ -1363,9 +1368,9 @@ func (i *singleLevelIterator[D, PD]) Prev() *base.InternalKV {
 	return i.skipBackward()
 }
 
-func (i *singleLevelIterator[D, PD]) skipForward() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) skipForward() *base.InternalKV {
 	for {
-		if !i.index.Next() {
+		if !PI(&i.index).Next() {
 			PD(&i.data).Invalidate()
 			break
 		}
@@ -1387,7 +1392,7 @@ func (i *singleLevelIterator[D, PD]) skipForward() *base.InternalKV {
 			// separator, the same user key can span multiple blocks. If upper
 			// is exclusive we use >= below, else we use >.
 			if i.upper != nil {
-				cmp := i.cmp(i.index.Separator(), i.upper)
+				cmp := i.cmp(PI(&i.index).Separator(), i.upper)
 				if (!i.endKeyInclusive && cmp >= 0) || cmp > 0 {
 					i.exhaustedBounds = +1
 					return nil
@@ -1443,9 +1448,9 @@ func (i *singleLevelIterator[D, PD]) skipForward() *base.InternalKV {
 	return nil
 }
 
-func (i *singleLevelIterator[D, PD]) skipBackward() *base.InternalKV {
+func (i *singleLevelIterator[I, PI, D, PD]) skipBackward() *base.InternalKV {
 	for {
-		if !i.index.Prev() {
+		if !PI(&i.index).Prev() {
 			PD(&i.data).Invalidate()
 			break
 		}
@@ -1465,7 +1470,7 @@ func (i *singleLevelIterator[D, PD]) skipBackward() *base.InternalKV {
 			// bound is already exceeded. Note that the previous block starts with
 			// keys <= key.UserKey since even though this is the current block's
 			// separator, the same user key can span multiple blocks.
-			if i.lower != nil && i.cmp(i.index.Separator(), i.lower) < 0 {
+			if i.lower != nil && i.cmp(PI(&i.index).Separator(), i.lower) < 0 {
 				i.exhaustedBounds = -1
 				return nil
 			}
@@ -1477,7 +1482,7 @@ func (i *singleLevelIterator[D, PD]) skipBackward() *base.InternalKV {
 			// safe to assume that there are no keys if we keep skipping backwards.
 			// Check the previous block, but check the lower bound before doing
 			// that.
-			if i.lower != nil && i.cmp(i.index.Separator(), i.lower) < 0 {
+			if i.lower != nil && i.cmp(PI(&i.index).Separator(), i.lower) < 0 {
 				i.exhaustedBounds = -1
 				return nil
 			}
@@ -1494,7 +1499,7 @@ func (i *singleLevelIterator[D, PD]) skipBackward() *base.InternalKV {
 
 // Error implements internalIterator.Error, as documented in the pebble
 // package.
-func (i *singleLevelIterator[D, PD]) Error() error {
+func (i *singleLevelIterator[I, PI, D, PD]) Error() error {
 	if err := PD(&i.data).Error(); err != nil {
 		return err
 	}
@@ -1503,7 +1508,7 @@ func (i *singleLevelIterator[D, PD]) Error() error {
 
 // SetCloseHook sets a function that will be called when the iterator is
 // closed.
-func (i *singleLevelIterator[D, PD]) SetCloseHook(fn func(i Iterator) error) {
+func (i *singleLevelIterator[I, PI, D, PD]) SetCloseHook(fn func(i Iterator) error) {
 	i.closeHook = fn
 }
 
@@ -1516,7 +1521,7 @@ func firstError(err0, err1 error) error {
 
 // Close implements internalIterator.Close, as documented in the pebble
 // package.
-func (i *singleLevelIterator[D, PD]) Close() error {
+func (i *singleLevelIterator[I, PI, D, PD]) Close() error {
 	err := i.closeInternal()
 	pool := i.pool
 	*i = i.resetForReuse()
@@ -1526,7 +1531,7 @@ func (i *singleLevelIterator[D, PD]) Close() error {
 	return err
 }
 
-func (i *singleLevelIterator[D, PD]) closeInternal() error {
+func (i *singleLevelIterator[I, PI, D, PD]) closeInternal() error {
 	if invariants.Enabled && i.inPool {
 		panic("Close called on interator in pool")
 	}
@@ -1536,7 +1541,7 @@ func (i *singleLevelIterator[D, PD]) closeInternal() error {
 		err = firstError(err, i.closeHook(i))
 	}
 	err = firstError(err, PD(&i.data).Close())
-	err = firstError(err, i.index.Close())
+	err = firstError(err, PI(&i.index).Close())
 	if i.indexFilterRH != nil {
 		err = firstError(err, i.indexFilterRH.Close())
 		i.indexFilterRH = nil
@@ -1559,7 +1564,7 @@ func (i *singleLevelIterator[D, PD]) closeInternal() error {
 	return err
 }
 
-func (i *singleLevelIterator[D, PD]) String() string {
+func (i *singleLevelIterator[I, PI, D, PD]) String() string {
 	if i.vState != nil {
 		return i.vState.fileNum.String()
 	}
@@ -1567,6 +1572,6 @@ func (i *singleLevelIterator[D, PD]) String() string {
 }
 
 // DebugTree is part of the InternalIterator interface.
-func (i *singleLevelIterator[D, PD]) DebugTree(tp treeprinter.Node) {
+func (i *singleLevelIterator[I, PI, D, PD]) DebugTree(tp treeprinter.Node) {
 	tp.Childf("%T(%p) fileNum=%s", i, i, i.String())
 }
diff --git a/sstable/reader_iter_two_lvl.go b/sstable/reader_iter_two_lvl.go
index 88a49723..26fdd84e 100644
--- a/sstable/reader_iter_two_lvl.go
+++ b/sstable/reader_iter_two_lvl.go
@@ -20,8 +20,8 @@ import (
 	"github.com/cockroachdb/pebble/sstable/rowblk"
 )
 
-type twoLevelIterator[D any, PD block.DataBlockIterator[D]] struct {
-	secondLevel   singleLevelIterator[D, PD]
+type twoLevelIterator[I any, PI block.IndexBlockIterator[I], D any, PD block.DataBlockIterator[D]] struct {
+	secondLevel   singleLevelIterator[I, PI, D, PD]
 	topLevelIndex rowblk.IndexIter
 	// pool is the pool from which the iterator was allocated and to which the
 	// iterator should be returned on Close. Because the iterator is
@@ -36,17 +36,17 @@ type twoLevelIterator[D any, PD block.DataBlockIterator[D]] struct {
 	lastBloomFilterMatched bool
 }
 
-var _ Iterator = (*twoLevelIterator[rowblk.Iter, *rowblk.Iter])(nil)
+var _ Iterator = (*twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])(nil)
 
 // loadIndex loads the index block at the current top level index position and
 // leaves i.index unpositioned. If unsuccessful, it gets i.secondLevel.err to any error
 // encountered, which may be nil if we have simply exhausted the entire table.
 // This is used for two level indexes.
-func (i *twoLevelIterator[D, PD]) loadIndex(dir int8) loadBlockResult {
+func (i *twoLevelIterator[I, PI, D, PD]) loadIndex(dir int8) loadBlockResult {
 	// Ensure the index data block iterators are invalidated even if loading of
 	// the index fails.
 	PD(&i.secondLevel.data).Invalidate()
-	i.secondLevel.index.Invalidate()
+	PI(&i.secondLevel.index).Invalidate()
 	if !i.topLevelIndex.Valid() {
 		return loadBlockFailed
 	}
@@ -73,7 +73,7 @@ func (i *twoLevelIterator[D, PD]) loadIndex(dir int8) loadBlockResult {
 	indexBlock, err := i.secondLevel.reader.readBlock(
 		ctx, bhp.Handle, nil /* transform */, i.secondLevel.indexFilterRH, i.secondLevel.stats, &i.secondLevel.iterStats, i.secondLevel.bufferPool)
 	if err == nil {
-		err = i.secondLevel.index.InitHandle(i.secondLevel.cmp, i.secondLevel.reader.Split, indexBlock, i.secondLevel.transforms)
+		err = PI(&i.secondLevel.index).InitHandle(i.secondLevel.cmp, i.secondLevel.reader.Split, indexBlock, i.secondLevel.transforms)
 	}
 	if err != nil {
 		i.secondLevel.err = err
@@ -87,7 +87,7 @@ func (i *twoLevelIterator[D, PD]) loadIndex(dir int8) loadBlockResult {
 // bounds fall within the filter's current bounds. This function consults the
 // appropriate bound, depending on the iteration direction, and returns either
 // `blockIntersects` or `blockExcluded`.
-func (i *twoLevelIterator[D, PD]) resolveMaybeExcluded(dir int8) intersectsResult {
+func (i *twoLevelIterator[I, PI, D, PD]) resolveMaybeExcluded(dir int8) intersectsResult {
 	// This iterator is configured with a bound-limited block property filter.
 	// The bpf determined this entire index block could be excluded from
 	// iteration based on the property encoded in the block handle. However, we
@@ -165,13 +165,13 @@ func newRowBlockTwoLevelIterator(
 	statsCollector *CategoryStatsCollector,
 	rp ReaderProvider,
 	bufferPool *block.BufferPool,
-) (*twoLevelIterator[rowblk.Iter, *rowblk.Iter], error) {
+) (*twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter], error) {
 	if r.err != nil {
 		return nil, r.err
 	}
 	// TODO(jackson): When we have a columnar-block sstable format, assert that
 	// the table format is row-oriented.
-	i := twoLevelIterRowBlockPool.Get().(*twoLevelIterator[rowblk.Iter, *rowblk.Iter])
+	i := twoLevelIterRowBlockPool.Get().(*twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])
 	i.secondLevel.init(ctx, r, v, transforms, lower, upper, filterer,
 		false, // Disable the use of the filter block in the second level.
 		stats, categoryAndQoS, statsCollector, bufferPool)
@@ -211,19 +211,21 @@ func newRowBlockTwoLevelIterator(
 	return i, nil
 }
 
-func (i *twoLevelIterator[D, PD]) String() string {
+func (i *twoLevelIterator[I, PI, D, PD]) String() string {
 	return i.secondLevel.String()
 }
 
 // DebugTree is part of the InternalIterator interface.
-func (i *twoLevelIterator[D, PD]) DebugTree(tp treeprinter.Node) {
+func (i *twoLevelIterator[I, PI, D, PD]) DebugTree(tp treeprinter.Node) {
 	tp.Childf("%T(%p) fileNum=%s", i, i, i.String())
 }
 
 // SeekGE implements internalIterator.SeekGE, as documented in the pebble
 // package. Note that SeekGE only checks the upper bound. It is up to the
 // caller to ensure that key is greater than or equal to the lower bound.
-func (i *twoLevelIterator[D, PD]) SeekGE(key []byte, flags base.SeekGEFlags) *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) SeekGE(
+	key []byte, flags base.SeekGEFlags,
+) *base.InternalKV {
 	if i.secondLevel.vState != nil {
 		// Callers of SeekGE don't know about virtual sstable bounds, so we may
 		// have to internally restrict the bounds.
@@ -242,7 +244,7 @@ func (i *twoLevelIterator[D, PD]) SeekGE(key []byte, flags base.SeekGEFlags) *ba
 	// trySeekUsingNext is true. See the comment about data-exhausted, PGDE, and
 	// bounds-exhausted near the top of the file.
 	if flags.TrySeekUsingNext() &&
-		(i.secondLevel.exhaustedBounds == +1 || (PD(&i.secondLevel.data).IsDataInvalidated() && i.secondLevel.index.IsDataInvalidated())) &&
+		(i.secondLevel.exhaustedBounds == +1 || (PD(&i.secondLevel.data).IsDataInvalidated() && PI(&i.secondLevel.index).IsDataInvalidated())) &&
 		err == nil {
 		// Already exhausted, so return nil.
 		return nil
@@ -260,7 +262,7 @@ func (i *twoLevelIterator[D, PD]) SeekGE(key []byte, flags base.SeekGEFlags) *ba
 	// block load.
 
 	var dontSeekWithinSingleLevelIter bool
-	if i.topLevelIndex.IsDataInvalidated() || !i.topLevelIndex.Valid() || i.secondLevel.index.IsDataInvalidated() || err != nil ||
+	if i.topLevelIndex.IsDataInvalidated() || !i.topLevelIndex.Valid() || PI(&i.secondLevel.index).IsDataInvalidated() || err != nil ||
 		(i.secondLevel.boundsCmp <= 0 && !flags.TrySeekUsingNext()) || i.secondLevel.cmp(key, i.topLevelIndex.Separator()) > 0 {
 		// Slow-path: need to position the topLevelIndex.
 
@@ -270,7 +272,7 @@ func (i *twoLevelIterator[D, PD]) SeekGE(key []byte, flags base.SeekGEFlags) *ba
 		flags = flags.DisableTrySeekUsingNext()
 		if !i.topLevelIndex.SeekGE(key) {
 			PD(&i.secondLevel.data).Invalidate()
-			i.secondLevel.index.Invalidate()
+			PI(&i.secondLevel.index).Invalidate()
 			return nil
 		}
 
@@ -360,7 +362,7 @@ func (i *twoLevelIterator[D, PD]) SeekGE(key []byte, flags base.SeekGEFlags) *ba
 // SeekPrefixGE implements internalIterator.SeekPrefixGE, as documented in the
 // pebble package. Note that SeekPrefixGE only checks the upper bound. It is up
 // to the caller to ensure that key is greater than or equal to the lower bound.
-func (i *twoLevelIterator[D, PD]) SeekPrefixGE(
+func (i *twoLevelIterator[I, PI, D, PD]) SeekPrefixGE(
 	prefix, key []byte, flags base.SeekGEFlags,
 ) *base.InternalKV {
 	if i.secondLevel.vState != nil {
@@ -386,7 +388,7 @@ func (i *twoLevelIterator[D, PD]) SeekPrefixGE(
 	// bounds-exhausted near the top of the file.
 	filterUsedAndDidNotMatch := i.useFilterBlock && !i.lastBloomFilterMatched
 	if flags.TrySeekUsingNext() && !filterUsedAndDidNotMatch &&
-		(i.secondLevel.exhaustedBounds == +1 || (PD(&i.secondLevel.data).IsDataInvalidated() && i.secondLevel.index.IsDataInvalidated())) &&
+		(i.secondLevel.exhaustedBounds == +1 || (PD(&i.secondLevel.data).IsDataInvalidated() && PI(&i.secondLevel.index).IsDataInvalidated())) &&
 		err == nil {
 		// Already exhausted, so return nil.
 		return nil
@@ -428,7 +430,7 @@ func (i *twoLevelIterator[D, PD]) SeekPrefixGE(
 	// block load.
 
 	var dontSeekWithinSingleLevelIter bool
-	if i.topLevelIndex.IsDataInvalidated() || !i.topLevelIndex.Valid() || i.secondLevel.index.IsDataInvalidated() || err != nil ||
+	if i.topLevelIndex.IsDataInvalidated() || !i.topLevelIndex.Valid() || PI(&i.secondLevel.index).IsDataInvalidated() || err != nil ||
 		(i.secondLevel.boundsCmp <= 0 && !flags.TrySeekUsingNext()) || i.secondLevel.cmp(key, i.topLevelIndex.Separator()) > 0 {
 		// Slow-path: need to position the topLevelIndex.
 
@@ -438,7 +440,7 @@ func (i *twoLevelIterator[D, PD]) SeekPrefixGE(
 		flags = flags.DisableTrySeekUsingNext()
 		if !i.topLevelIndex.SeekGE(key) {
 			PD(&i.secondLevel.data).Invalidate()
-			i.secondLevel.index.Invalidate()
+			PI(&i.secondLevel.index).Invalidate()
 			return nil
 		}
 
@@ -525,7 +527,7 @@ func (i *twoLevelIterator[D, PD]) SeekPrefixGE(
 }
 
 // virtualLast should only be called if i.vReader != nil.
-func (i *twoLevelIterator[D, PD]) virtualLast() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) virtualLast() *base.InternalKV {
 	if i.secondLevel.vState == nil {
 		panic("pebble: invalid call to virtualLast")
 	}
@@ -539,7 +541,7 @@ func (i *twoLevelIterator[D, PD]) virtualLast() *base.InternalKV {
 // virtualLastSeekLE implements a SeekLE() that can be used as part
 // of reverse-iteration calls such as a Last() on a virtual sstable. Does a
 // SeekLE on the upper bound of the file/iterator.
-func (i *twoLevelIterator[D, PD]) virtualLastSeekLE() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) virtualLastSeekLE() *base.InternalKV {
 	// Callers of SeekLE don't know about virtual sstable bounds, so we may
 	// have to internally restrict the bounds.
 	//
@@ -584,7 +586,9 @@ func (i *twoLevelIterator[D, PD]) virtualLastSeekLE() *base.InternalKV {
 // SeekLT implements internalIterator.SeekLT, as documented in the pebble
 // package. Note that SeekLT only checks the lower bound. It is up to the
 // caller to ensure that key is less than the upper bound.
-func (i *twoLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags) *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) SeekLT(
+	key []byte, flags base.SeekLTFlags,
+) *base.InternalKV {
 	if i.secondLevel.vState != nil {
 		// Might have to fix upper bound since virtual sstable bounds are not
 		// known to callers of SeekLT.
@@ -615,7 +619,7 @@ func (i *twoLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags) *ba
 	if !i.topLevelIndex.SeekGE(key) {
 		if !i.topLevelIndex.Last() {
 			PD(&i.secondLevel.data).Invalidate()
-			i.secondLevel.index.Invalidate()
+			PI(&i.secondLevel.index).Invalidate()
 			return nil
 		}
 
@@ -665,7 +669,7 @@ func (i *twoLevelIterator[D, PD]) SeekLT(key []byte, flags base.SeekLTFlags) *ba
 // package. Note that First only checks the upper bound. It is up to the caller
 // to ensure that key is greater than or equal to the lower bound (e.g. via a
 // call to SeekGE(lower)).
-func (i *twoLevelIterator[D, PD]) First() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) First() *base.InternalKV {
 	// If we have a lower bound, use SeekGE. Note that in general this is not
 	// supported usage, except when the lower bound is there because the table is
 	// virtual.
@@ -711,7 +715,7 @@ func (i *twoLevelIterator[D, PD]) First() *base.InternalKV {
 // package. Note that Last only checks the lower bound. It is up to the caller
 // to ensure that key is less than the upper bound (e.g. via a call to
 // SeekLT(upper))
-func (i *twoLevelIterator[D, PD]) Last() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) Last() *base.InternalKV {
 	if i.secondLevel.vState != nil {
 		if i.secondLevel.endKeyInclusive {
 			return i.virtualLast()
@@ -758,7 +762,7 @@ func (i *twoLevelIterator[D, PD]) Last() *base.InternalKV {
 // package.
 // Note: twoLevelCompactionIterator.Next mirrors the implementation of
 // twoLevelIterator.Next due to performance. Keep the two in sync.
-func (i *twoLevelIterator[D, PD]) Next() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) Next() *base.InternalKV {
 	// Seek optimization only applies until iterator is first positioned after SetBounds.
 	i.secondLevel.boundsCmp = 0
 	if i.secondLevel.err != nil {
@@ -773,7 +777,7 @@ func (i *twoLevelIterator[D, PD]) Next() *base.InternalKV {
 }
 
 // NextPrefix implements (base.InternalIterator).NextPrefix.
-func (i *twoLevelIterator[D, PD]) NextPrefix(succKey []byte) *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) NextPrefix(succKey []byte) *base.InternalKV {
 	if i.secondLevel.exhaustedBounds == +1 {
 		panic("Next called even though exhausted upper bound")
 	}
@@ -796,7 +800,7 @@ func (i *twoLevelIterator[D, PD]) NextPrefix(succKey []byte) *base.InternalKV {
 	// slow-path where we seek the iterator.
 	if !i.topLevelIndex.SeekGE(succKey) {
 		PD(&i.secondLevel.data).Invalidate()
-		i.secondLevel.index.Invalidate()
+		PI(&i.secondLevel.index).Invalidate()
 		return nil
 	}
 	result := i.loadIndex(+1)
@@ -824,7 +828,7 @@ func (i *twoLevelIterator[D, PD]) NextPrefix(succKey []byte) *base.InternalKV {
 
 // Prev implements internalIterator.Prev, as documented in the pebble
 // package.
-func (i *twoLevelIterator[D, PD]) Prev() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) Prev() *base.InternalKV {
 	// Seek optimization only applies until iterator is first positioned after SetBounds.
 	i.secondLevel.boundsCmp = 0
 	if i.secondLevel.err != nil {
@@ -836,7 +840,7 @@ func (i *twoLevelIterator[D, PD]) Prev() *base.InternalKV {
 	return i.skipBackward()
 }
 
-func (i *twoLevelIterator[D, PD]) skipForward() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) skipForward() *base.InternalKV {
 	for {
 		if i.secondLevel.err != nil || i.secondLevel.exhaustedBounds > 0 {
 			return nil
@@ -877,7 +881,7 @@ func (i *twoLevelIterator[D, PD]) skipForward() *base.InternalKV {
 		i.secondLevel.exhaustedBounds = 0
 		if !i.topLevelIndex.Next() {
 			PD(&i.secondLevel.data).Invalidate()
-			i.secondLevel.index.Invalidate()
+			PI(&i.secondLevel.index).Invalidate()
 			return nil
 		}
 		result := i.loadIndex(+1)
@@ -915,7 +919,7 @@ func (i *twoLevelIterator[D, PD]) skipForward() *base.InternalKV {
 	}
 }
 
-func (i *twoLevelIterator[D, PD]) skipBackward() *base.InternalKV {
+func (i *twoLevelIterator[I, PI, D, PD]) skipBackward() *base.InternalKV {
 	for {
 		if i.secondLevel.err != nil || i.secondLevel.exhaustedBounds < 0 {
 			return nil
@@ -923,7 +927,7 @@ func (i *twoLevelIterator[D, PD]) skipBackward() *base.InternalKV {
 		i.secondLevel.exhaustedBounds = 0
 		if !i.topLevelIndex.Prev() {
 			PD(&i.secondLevel.data).Invalidate()
-			i.secondLevel.index.Invalidate()
+			PI(&i.secondLevel.index).Invalidate()
 			return nil
 		}
 		result := i.loadIndex(-1)
@@ -953,36 +957,36 @@ func (i *twoLevelIterator[D, PD]) skipBackward() *base.InternalKV {
 	}
 }
 
-func (i *twoLevelIterator[D, PD]) Error() error {
+func (i *twoLevelIterator[I, PI, D, PD]) Error() error {
 	return i.secondLevel.Error()
 }
 
-func (i *twoLevelIterator[D, PD]) SetBounds(lower, upper []byte) {
+func (i *twoLevelIterator[I, PI, D, PD]) SetBounds(lower, upper []byte) {
 	i.secondLevel.SetBounds(lower, upper)
 }
 
-func (i *twoLevelIterator[D, PD]) SetContext(ctx context.Context) {
+func (i *twoLevelIterator[I, PI, D, PD]) SetContext(ctx context.Context) {
 	i.secondLevel.SetContext(ctx)
 }
 
-func (i *twoLevelIterator[D, PD]) SetCloseHook(fn func(i Iterator) error) {
+func (i *twoLevelIterator[I, PI, D, PD]) SetCloseHook(fn func(i Iterator) error) {
 	i.secondLevel.SetCloseHook(fn)
 }
 
-func (i *twoLevelIterator[D, PD]) SetupForCompaction() {
+func (i *twoLevelIterator[I, PI, D, PD]) SetupForCompaction() {
 	i.secondLevel.SetupForCompaction()
 }
 
 // Close implements internalIterator.Close, as documented in the pebble
 // package.
-func (i *twoLevelIterator[D, PD]) Close() error {
+func (i *twoLevelIterator[I, PI, D, PD]) Close() error {
 	if invariants.Enabled && i.secondLevel.pool != nil {
 		panic("twoLevelIterator's singleLevelIterator has its own non-nil pool")
 	}
 	pool := i.pool
 	err := i.secondLevel.closeInternal()
 	err = firstError(err, i.topLevelIndex.Close())
-	*i = twoLevelIterator[D, PD]{
+	*i = twoLevelIterator[I, PI, D, PD]{
 		secondLevel:   i.secondLevel.resetForReuse(),
 		topLevelIndex: i.topLevelIndex.ResetForReuse(),
 		pool:          pool,
diff --git a/sstable/raw_writer.go b/sstable/rowblk_writer.go
similarity index 94%
rename from sstable/raw_writer.go
rename to sstable/rowblk_writer.go
index 56aec74b..e06e81e3 100644
--- a/sstable/raw_writer.go
+++ b/sstable/rowblk_writer.go
@@ -33,83 +33,6 @@ 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 {
@@ -120,8 +43,10 @@ type flushDecisionOptions struct {
 	sizeClassAwareThreshold int
 }
 
-// RawWriter is a table writer.
-type RawWriter struct {
+// RawRowWriter is a sstable RawWriter that writes sstables with row-oriented
+// blocks. All table formats TableFormatPebblev4 and earlier write row-oriented
+// blocks and use RawRowWriter.
+type RawRowWriter struct {
 	layout layoutWriter
 	meta   WriterMetadata
 	err    error
@@ -233,7 +158,7 @@ type coordinationState struct {
 	sizeEstimate dataBlockEstimates
 }
 
-func (c *coordinationState) init(parallelismEnabled bool, writer *RawWriter) {
+func (c *coordinationState) init(parallelismEnabled bool, writer *RawRowWriter) {
 	c.parallelismEnabled = parallelismEnabled
 	// useMutex is false regardless of parallelismEnabled, because we do not do
 	// parallel compression yet.
@@ -659,7 +584,7 @@ type indexBlockAndBlockProperties struct {
 // 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 {
+func (w *RawRowWriter) Add(key InternalKey, value []byte) error {
 	if w.isStrictObsolete {
 		return errors.Errorf("use AddWithForceObsolete")
 	}
@@ -677,7 +602,9 @@ func (w *RawWriter) Add(key InternalKey, value []byte) error {
 // 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 {
+func (w *RawRowWriter) AddWithForceObsolete(
+	key InternalKey, value []byte, forceObsolete bool,
+) error {
 	if w.err != nil {
 		return w.err
 	}
@@ -695,7 +622,7 @@ func (w *RawWriter) AddWithForceObsolete(key InternalKey, value []byte, forceObs
 	return w.addPoint(key, value, forceObsolete)
 }
 
-func (w *RawWriter) makeAddPointDecisionV2(key InternalKey) error {
+func (w *RawRowWriter) makeAddPointDecisionV2(key InternalKey) error {
 	prevTrailer := w.lastPointKeyInfo.trailer
 	w.lastPointKeyInfo.trailer = key.Trailer
 	if w.dataBlockBuf.dataBlock.EntryCount() == 0 {
@@ -715,7 +642,7 @@ func (w *RawWriter) makeAddPointDecisionV2(key InternalKey) error {
 }
 
 // REQUIRES: at least one point has been written to the Writer.
-func (w *RawWriter) getLastPointUserKey() []byte {
+func (w *RawRowWriter) getLastPointUserKey() []byte {
 	if w.dataBlockBuf.dataBlock.EntryCount() == 0 {
 		panic(errors.AssertionFailedf("no point keys added to writer"))
 	}
@@ -723,7 +650,7 @@ func (w *RawWriter) getLastPointUserKey() []byte {
 }
 
 // REQUIRES: w.tableFormat >= TableFormatPebblev3
-func (w *RawWriter) makeAddPointDecisionV3(
+func (w *RawRowWriter) makeAddPointDecisionV3(
 	key InternalKey, valueLen int,
 ) (setHasSamePrefix bool, writeToValueBlock bool, isObsolete bool, err error) {
 	prevPointKeyInfo := w.lastPointKeyInfo
@@ -839,7 +766,7 @@ func (w *RawWriter) makeAddPointDecisionV3(
 	return setHasSamePrefix, considerWriteToValueBlock, isObsolete, nil
 }
 
-func (w *RawWriter) addPoint(key InternalKey, value []byte, forceObsolete bool) error {
+func (w *RawRowWriter) 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")
 	}
@@ -968,7 +895,7 @@ func (w *RawWriter) addPoint(key InternalKey, value []byte, forceObsolete bool)
 	return nil
 }
 
-func (w *RawWriter) prettyTombstone(k InternalKey, value []byte) fmt.Formatter {
+func (w *RawRowWriter) prettyTombstone(k InternalKey, value []byte) fmt.Formatter {
 	return keyspan.Span{
 		Start: k.UserKey,
 		End:   value,
@@ -976,7 +903,7 @@ func (w *RawWriter) prettyTombstone(k InternalKey, value []byte) fmt.Formatter {
 	}.Pretty(w.formatKey)
 }
 
-func (w *RawWriter) addTombstone(key InternalKey, value []byte) error {
+func (w *RawRowWriter) 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.
@@ -1034,7 +961,7 @@ func (w *RawWriter) addTombstone(key InternalKey, value []byte) error {
 	return nil
 }
 
-func (w *RawWriter) encodeFragmentedRangeKeySpan(span keyspan.Span) {
+func (w *RawRowWriter) 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
@@ -1061,7 +988,7 @@ func (w *RawWriter) encodeFragmentedRangeKeySpan(span keyspan.Span) {
 // 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 {
+func (w *RawRowWriter) 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())
@@ -1143,7 +1070,7 @@ func (w *RawWriter) addRangeKey(key InternalKey, value []byte) error {
 	return nil
 }
 
-func (w *RawWriter) maybeAddToFilter(key []byte) {
+func (w *RawRowWriter) maybeAddToFilter(key []byte) {
 	if w.filter != nil {
 		prefix := key[:w.split(key)]
 		w.filter.addKey(prefix)
@@ -1155,7 +1082,7 @@ func (w *RawWriter) maybeAddToFilter(key []byte) {
 // the deletion size exceeds a threshold. It should be called after the
 // data block has been finished.
 // Invariant: w.dataBlockBuf.uncompressed must already be populated.
-func (w *RawWriter) maybeIncrementTombstoneDenseBlocks() {
+func (w *RawRowWriter) maybeIncrementTombstoneDenseBlocks() {
 	minSize := w.deletionSizeRatioThreshold * float32(len(w.dataBlockBuf.uncompressed))
 	if w.dataBlockBuf.numDeletions > w.numDeletionsThreshold || float32(w.dataBlockBuf.deletionSize) > minSize {
 		w.props.NumTombstoneDenseBlocks++
@@ -1164,7 +1091,7 @@ func (w *RawWriter) maybeIncrementTombstoneDenseBlocks() {
 	w.dataBlockBuf.deletionSize = 0
 }
 
-func (w *RawWriter) flush(key InternalKey) error {
+func (w *RawRowWriter) flush(key InternalKey) error {
 	// We're finishing a data block.
 	err := w.finishDataBlockProps(w.dataBlockBuf)
 	if err != nil {
@@ -1241,7 +1168,7 @@ func (w *RawWriter) flush(key InternalKey) error {
 	return err
 }
 
-func (w *RawWriter) maybeFlush(key InternalKey, valueLen int) error {
+func (w *RawRowWriter) maybeFlush(key InternalKey, valueLen int) error {
 	if !w.dataBlockBuf.shouldFlush(key, valueLen, w.dataBlockOptions, w.allocatorSizeClasses) {
 		return nil
 	}
@@ -1259,7 +1186,7 @@ func (w *RawWriter) maybeFlush(key InternalKey, valueLen int) error {
 // 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 {
+func (w *RawRowWriter) finishDataBlockProps(buf *dataBlockBuf) error {
 	if len(w.blockPropCollectors) == 0 {
 		return nil
 	}
@@ -1281,7 +1208,7 @@ func (w *RawWriter) finishDataBlockProps(buf *dataBlockBuf) error {
 // 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(
+func (w *RawRowWriter) maybeAddBlockPropertiesToBlockHandle(
 	bh block.Handle,
 ) (block.HandleWithProperties, error) {
 	err := w.finishDataBlockProps(w.dataBlockBuf)
@@ -1291,7 +1218,7 @@ func (w *RawWriter) maybeAddBlockPropertiesToBlockHandle(
 	return block.HandleWithProperties{Handle: bh, Props: w.dataBlockBuf.dataBlockProps}, nil
 }
 
-func (w *RawWriter) indexEntrySep(
+func (w *RawRowWriter) indexEntrySep(
 	prevKey, key InternalKey, dataBlockBuf *dataBlockBuf,
 ) InternalKey {
 	// Make a rough guess that we want key-sized scratch to compute the separator.
@@ -1319,7 +1246,7 @@ func (w *RawWriter) indexEntrySep(
 //     encoded.
 //  2. addIndexEntry must not hold references to the flushIndexBuf, and the writeTo
 //     indexBlockBufs.
-func (w *RawWriter) addIndexEntry(
+func (w *RawRowWriter) addIndexEntry(
 	sep InternalKey,
 	bhp block.HandleWithProperties,
 	tmp []byte,
@@ -1350,7 +1277,7 @@ func (w *RawWriter) addIndexEntry(
 	return nil
 }
 
-func (w *RawWriter) addPrevDataBlockToIndexBlockProps() {
+func (w *RawRowWriter) addPrevDataBlockToIndexBlockProps() {
 	for i := range w.blockPropCollectors {
 		w.blockPropCollectors[i].AddPrevDataBlockToIndexBlock()
 	}
@@ -1367,13 +1294,13 @@ func (w *RawWriter) addPrevDataBlockToIndexBlockProps() {
 //
 // TODO: Improve coverage of this method. e.g. tests passed without the line
 // `w.twoLevelIndex = true` previously.
-func (w *RawWriter) addIndexEntrySync(
+func (w *RawRowWriter) addIndexEntrySync(
 	prevKey, key InternalKey, bhp block.HandleWithProperties, tmp []byte,
 ) error {
 	return w.addIndexEntrySep(w.indexEntrySep(prevKey, key, w.dataBlockBuf), bhp, tmp)
 }
 
-func (w *RawWriter) addIndexEntrySep(
+func (w *RawRowWriter) addIndexEntrySep(
 	sep InternalKey, bhp block.HandleWithProperties, tmp []byte,
 ) error {
 	shouldFlush := supportsTwoLevelIndex(
@@ -1530,7 +1457,7 @@ func cloneKeyWithBuf(k InternalKey, a bytealloc.A) (bytealloc.A, InternalKey) {
 //  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) {
+func (w *RawRowWriter) finishIndexBlockProps() ([]byte, error) {
 	w.blockPropsEncoder.resetProps()
 	for i := range w.blockPropCollectors {
 		scratch := w.blockPropsEncoder.getScratchForProp()
@@ -1552,7 +1479,7 @@ func (w *RawWriter) finishIndexBlockProps() ([]byte, error) {
 //     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 {
+func (w *RawRowWriter) finishIndexBlock(indexBuf *indexBlockBuf, props []byte) error {
 	part := indexBlockAndBlockProperties{
 		nEntries: indexBuf.block.EntryCount(), properties: props,
 	}
@@ -1571,7 +1498,7 @@ func (w *RawWriter) finishIndexBlock(indexBuf *indexBlockBuf, props []byte) erro
 	return nil
 }
 
-func (w *RawWriter) writeTwoLevelIndex() (block.Handle, error) {
+func (w *RawRowWriter) writeTwoLevelIndex() (block.Handle, error) {
 	props, err := w.finishIndexBlockProps()
 	if err != nil {
 		return block.Handle{}, err
@@ -1608,7 +1535,7 @@ func (w *RawWriter) writeTwoLevelIndex() (block.Handle, error) {
 
 // assertFormatCompatibility ensures that the features present on the table are
 // compatible with the table format version.
-func (w *RawWriter) assertFormatCompatibility() error {
+func (w *RawRowWriter) assertFormatCompatibility() error {
 	// PebbleDBv1: block properties.
 	if len(w.blockPropCollectors) > 0 && w.tableFormat < TableFormatPebblev1 {
 		return errors.Newf(
@@ -1642,13 +1569,19 @@ func (w *RawWriter) assertFormatCompatibility() error {
 	return nil
 }
 
+// IsStrictObsolete returns true if the writer is configured to write and
+// enforce a 'strict obsolete' sstable.
+func (w *RawRowWriter) IsStrictObsolete() bool {
+	return w.isStrictObsolete
+}
+
 // 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 {
+func (w *RawRowWriter) 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.
@@ -1662,7 +1595,7 @@ func (w *RawWriter) UnsafeLastPointUserKey() []byte {
 //
 // 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 {
+func (w *RawRowWriter) EncodeSpan(span keyspan.Span) error {
 	if span.Empty() {
 		return nil
 	}
@@ -1677,9 +1610,14 @@ func (w *RawWriter) EncodeSpan(span keyspan.Span) error {
 	return w.rangeKeyEncoder.Encode(span)
 }
 
+// Error returns the current accumulated error, if any.
+func (w *RawRowWriter) Error() error {
+	return w.err
+}
+
 // Close finishes writing the table and closes the underlying file that the
 // table was written to.
-func (w *RawWriter) Close() (err error) {
+func (w *RawRowWriter) Close() (err error) {
 	defer func() {
 		if w.valueBlockWriter != nil {
 			releaseValueBlockWriter(w.valueBlockWriter)
@@ -1878,7 +1816,7 @@ func (w *RawWriter) Close() (err error) {
 
 // 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 {
+func (w *RawRowWriter) EstimatedSize() uint64 {
 	if w == nil {
 		return 0
 	}
@@ -1889,7 +1827,7 @@ func (w *RawWriter) EstimatedSize() uint64 {
 
 // Metadata returns the metadata for the finished sstable. Only valid to call
 // after the sstable has been finished.
-func (w *RawWriter) Metadata() (*WriterMetadata, error) {
+func (w *RawRowWriter) Metadata() (*WriterMetadata, error) {
 	if !w.layout.IsFinished() {
 		return nil, errors.New("pebble: writer is not closed")
 	}
@@ -1898,9 +1836,9 @@ func (w *RawWriter) Metadata() (*WriterMetadata, error) {
 
 // NewRawWriter returns a new table writer for the file. Closing the writer will
 // close the file.
-func NewRawWriter(writable objstorage.Writable, o WriterOptions) *RawWriter {
+func NewRawWriter(writable objstorage.Writable, o WriterOptions) *RawRowWriter {
 	o = o.ensureDefaults()
-	w := &RawWriter{
+	w := &RawRowWriter{
 		layout: makeLayoutWriter(writable, o),
 		meta: WriterMetadata{
 			SmallestSeqNum: math.MaxUint64,
@@ -2018,7 +1956,7 @@ func NewRawWriter(writable objstorage.Writable, o WriterOptions) *RawWriter {
 
 // SetSnapshotPinnedProperties sets the properties for pinned keys. Should only
 // be used internally by Pebble.
-func (w *RawWriter) SetSnapshotPinnedProperties(
+func (w *RawRowWriter) SetSnapshotPinnedProperties(
 	pinnedKeyCount, pinnedKeySize, pinnedValueSize uint64,
 ) {
 	w.props.SnapshotPinnedKeys = pinnedKeyCount
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index c8a5d642..8a5a18cf 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -195,7 +195,7 @@ func rewriteBlocks[R blockRewriter](
 	return nil
 }
 
-func checkWriterFilterMatchesReader(r *Reader, w *RawWriter) error {
+func checkWriterFilterMatchesReader(r *Reader, w *RawRowWriter) error {
 	if r.Properties.FilterPolicyName != w.filter.policyName() {
 		return errors.New("mismatched filters")
 	}
@@ -206,7 +206,7 @@ func checkWriterFilterMatchesReader(r *Reader, w *RawWriter) error {
 }
 
 func rewriteDataBlocksToWriter(
-	r *Reader, w *RawWriter, data []block.HandleWithProperties, from, to []byte, concurrency int,
+	r *Reader, w *RawRowWriter, data []block.HandleWithProperties, from, to []byte, concurrency int,
 ) error {
 	if r.Properties.NumEntries == 0 {
 		// No point keys.
@@ -326,7 +326,7 @@ func rewriteDataBlocksToWriter(
 	return nil
 }
 
-func rewriteRangeKeyBlockToWriter(r *Reader, w *RawWriter, from, to []byte) error {
+func rewriteRangeKeyBlockToWriter(r *Reader, w *RawRowWriter, from, to []byte) error {
 	iter, err := r.NewRawRangeKeyIter(context.TODO(), NoFragmentTransforms)
 	if err != nil {
 		return err
diff --git a/sstable/write_queue.go b/sstable/write_queue.go
index ffe5fa52..e00d0b8a 100644
--- a/sstable/write_queue.go
+++ b/sstable/write_queue.go
@@ -40,7 +40,7 @@ func (task *writeTask) clear() {
 type writeQueue struct {
 	tasks  chan *writeTask
 	wg     sync.WaitGroup
-	writer *RawWriter
+	writer *RawRowWriter
 
 	// 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
@@ -49,7 +49,7 @@ type writeQueue struct {
 	closed bool
 }
 
-func newWriteQueue(size int, writer *RawWriter) *writeQueue {
+func newWriteQueue(size int, writer *RawRowWriter) *writeQueue {
 	w := &writeQueue{}
 	w.tasks = make(chan *writeTask, size)
 	w.writer = writer
diff --git a/sstable/writer.go b/sstable/writer.go
index 2c0877eb..2ff142d2 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -13,12 +13,13 @@ import (
 
 // Writer is a table writer.
 type Writer struct {
-	rw *RawWriter
+	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
+	comparer    *base.Comparer
 	rkBuf       []byte
 	keyspanKeys []keyspan.Key
 }
@@ -38,11 +39,12 @@ func NewWriter(writable objstorage.Writable, o WriterOptions) *Writer {
 			Format: o.Comparer.FormatKey,
 			Emit:   rw.encodeFragmentedRangeKeySpan,
 		},
+		comparer: o.Comparer,
 	}
 }
 
 // Raw returns the underlying RawWriter.
-func (w *Writer) Raw() *RawWriter { return w.rw }
+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
@@ -51,15 +53,15 @@ func (w *Writer) Raw() *RawWriter { return w.rw }
 //
 // TODO(peter): untested
 func (w *Writer) Set(key, value []byte) error {
-	if w.rw.err != nil {
-		return w.rw.err
+	if err := w.rw.Error(); err != nil {
+		return err
 	}
-	if w.rw.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.rw.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindSet), value, false)
+	return w.rw.AddWithForceObsolete(base.MakeInternalKey(key, 0, InternalKeyKindSet), value, false)
 }
 
 // Delete deletes the value for the given key. The sequence number is set to
@@ -68,15 +70,15 @@ func (w *Writer) Set(key, value []byte) error {
 //
 // TODO(peter): untested
 func (w *Writer) Delete(key []byte) error {
-	if w.rw.err != nil {
-		return w.rw.err
+	if err := w.rw.Error(); err != nil {
+		return err
 	}
-	if w.rw.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.rw.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindDelete), nil, false)
+	return w.rw.AddWithForceObsolete(base.MakeInternalKey(key, 0, InternalKeyKindDelete), nil, false)
 }
 
 // DeleteRange deletes all of the keys (and values) in the range [start,end)
@@ -89,8 +91,8 @@ func (w *Writer) Delete(key []byte) error {
 //
 // TODO(peter): untested
 func (w *Writer) DeleteRange(start, end []byte) error {
-	if w.rw.err != nil {
-		return w.rw.err
+	if err := w.rw.Error(); err != nil {
+		return err
 	}
 	return w.rw.EncodeSpan(keyspan.Span{
 		Start: start,
@@ -108,16 +110,16 @@ func (w *Writer) DeleteRange(start, end []byte) error {
 //
 // TODO(peter): untested
 func (w *Writer) Merge(key, value []byte) error {
-	if w.rw.err != nil {
-		return w.rw.err
+	if err := w.rw.Error(); err != nil {
+		return err
 	}
-	if w.rw.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.rw.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindMerge), value, false)
+	return w.rw.AddWithForceObsolete(base.MakeInternalKey(key, 0, InternalKeyKindMerge), value, false)
 }
 
 // RangeKeySet sets a range between start (inclusive) and end (exclusive) with
@@ -182,18 +184,18 @@ func (w *Writer) RangeKeyDelete(start, end []byte) error {
 }
 
 func (w *Writer) addRangeKeySpanToFragmenter(span keyspan.Span) error {
-	if w.rw.compare(span.Start, span.End) >= 0 {
+	if w.comparer.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.rw.compare(w.fragmenter.Start(), span.Start) > 0 {
+	if w.fragmenter.Start() != nil && w.comparer.Compare(w.fragmenter.Start(), span.Start) > 0 {
 		return errors.Errorf("pebble: spans must be added in order: %s > %s",
-			w.rw.formatKey(w.fragmenter.Start()), w.rw.formatKey(span.Start))
+			w.comparer.FormatKey(w.fragmenter.Start()), w.comparer.FormatKey(span.Start))
 	}
 	// Add this span to the fragmenter.
 	w.fragmenter.Add(span)
-	return w.rw.err
+	return w.rw.Error()
 }
 
 // tempRangeKeyBuf returns a slice of length n from the Writer's rkBuf byte
@@ -228,10 +230,125 @@ func (w *Writer) tempRangeKeyCopy(k []byte) []byte {
 // Close finishes writing the table and closes the underlying file that the
 // table was written to.
 func (w *Writer) Close() (err error) {
-	if w.rw.err == nil {
+	if w.rw.Error() == nil {
 		// Write the range-key block, flushing any remaining spans from the
 		// fragmenter first.
 		w.fragmenter.Finish()
 	}
 	return w.rw.Close()
 }
+
+// RawWriter defines an interface for sstable writers. Implementations may vary
+// depending on the TableFormat being written.
+type RawWriter interface {
+	// Error returns the current accumulated error if any.
+	Error() error
+	// IsStrictObsolete returns true if the writer is configured to write and
+	// enforce a 'strict obsolete' sstable. This includes prohibiting the
+	// addition of MERGE keys. See the documentation in format.go for more
+	// details.
+	IsStrictObsolete() bool
+	// 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. It's optional for non-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.
+	AddWithForceObsolete(
+		key InternalKey, value []byte, forceObsolete bool,
+	) error
+	// 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.
+	EncodeSpan(span keyspan.Span) error
+	// Close finishes writing the table and closes the underlying file that the
+	// table was written to.
+	Close() error
+	// Metadata returns the metadata for the finished sstable. Only valid to
+	// call after the sstable has been finished.
+	Metadata() (*WriterMetadata, error)
+}
+
+// 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
+	}
+}
diff --git a/tool/wal.go b/tool/wal.go
index b52db349..27a279b8 100644
--- a/tool/wal.go
+++ b/tool/wal.go
@@ -9,8 +9,11 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"os"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/batchrepr"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/rangekey"
 	"github.com/cockroachdb/pebble/record"
@@ -22,8 +25,9 @@ import (
 // walT implements WAL-level tools, including both configuration state and the
 // commands themselves.
 type walT struct {
-	Root *cobra.Command
-	Dump *cobra.Command
+	Root       *cobra.Command
+	Dump       *cobra.Command
+	DumpMerged *cobra.Command
 
 	opts     *pebble.Options
 	fmtKey   keyFormatter
@@ -56,8 +60,19 @@ Print the contents of the WAL files.
 		Args: cobra.MinimumNArgs(1),
 		Run:  w.runDump,
 	}
+	w.DumpMerged = &cobra.Command{
+		Use:   "dump-merged <wal-files>",
+		Short: "print WAL contents",
+		Long: `
+Print the merged contents of multiple WAL segment files that
+together form a single logical WAL.
+`,
+		Args: cobra.MinimumNArgs(1),
+		Run:  w.runDumpMerged,
+	}
 
 	w.Root.AddCommand(w.Dump)
+	w.Root.AddCommand(w.DumpMerged)
 	w.Root.PersistentFlags().BoolVarP(&w.verbose, "verbose", "v", false, "verbose output")
 
 	w.Dump.Flags().Var(
@@ -67,10 +82,24 @@ Print the contents of the WAL files.
 	return w
 }
 
+type errAndArg struct {
+	err error
+	arg string
+}
+
 func (w *walT) runDump(cmd *cobra.Command, args []string) {
 	stdout, stderr := cmd.OutOrStdout(), cmd.OutOrStderr()
 	w.fmtKey.setForComparer(w.defaultComparer, w.comparers)
 	w.fmtValue.setForComparer(w.defaultComparer, w.comparers)
+	var errs []errAndArg
+	logErr := func(arg string, offset int64, err error) {
+		err = errors.Wrapf(err, "%s: offset %d: ", arg, offset)
+		fmt.Fprintf(stderr, "%s\n", err)
+		errs = append(errs, errAndArg{
+			arg: arg,
+			err: err,
+		})
+	}
 
 	for _, arg := range args {
 		func() {
@@ -107,10 +136,10 @@ func (w *walT) runDump(cmd *cobra.Command, args []string) {
 					// preallocation and WAL recycling. We need to distinguish these
 					// errors from EOF in order to recognize that the record was
 					// truncated, but want to otherwise treat them like EOF.
-					switch err {
-					case record.ErrZeroedChunk:
+					switch {
+					case errors.Is(err, record.ErrZeroedChunk):
 						fmt.Fprintf(stdout, "EOF [%s] (may be due to WAL preallocation)\n", err)
-					case record.ErrInvalidChunk:
+					case errors.Is(err, record.ErrInvalidChunk):
 						fmt.Fprintf(stdout, "EOF [%s] (may be due to WAL recycling)\n", err)
 					default:
 						fmt.Fprintf(stdout, "%s\n", err)
@@ -123,50 +152,148 @@ func (w *walT) runDump(cmd *cobra.Command, args []string) {
 					fmt.Fprintf(stdout, "corrupt batch within log file %q: %v", arg, err)
 					return
 				}
-				fmt.Fprintf(stdout, "%d(%d) seq=%d count=%d\n",
-					offset, len(b.Repr()), b.SeqNum(), b.Count())
-				for r, idx := b.Reader(), 0; ; idx++ {
-					kind, ukey, value, ok, err := r.Next()
-					if !ok {
-						if err != nil {
-							fmt.Fprintf(stdout, "corrupt batch within log file %q: %v", arg, err)
-						}
-						break
-					}
-					fmt.Fprintf(stdout, "    %s(", kind)
-					switch kind {
-					case base.InternalKeyKindDelete:
-						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
-					case base.InternalKeyKindSet:
-						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
-					case base.InternalKeyKindMerge:
-						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
-					case base.InternalKeyKindLogData:
-						fmt.Fprintf(stdout, "<%d>", len(value))
-					case base.InternalKeyKindIngestSST:
-						fileNum, _ := binary.Uvarint(ukey)
-						fmt.Fprintf(stdout, "%s", base.FileNum(fileNum))
-					case base.InternalKeyKindSingleDelete:
-						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
-					case base.InternalKeyKindSetWithDelete:
-						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
-					case base.InternalKeyKindRangeDelete:
-						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtKey.fn(value))
-					case base.InternalKeyKindRangeKeySet, base.InternalKeyKindRangeKeyUnset, base.InternalKeyKindRangeKeyDelete:
-						ik := base.MakeInternalKey(ukey, b.SeqNum()+base.SeqNum(idx), kind)
-						s, err := rangekey.Decode(ik, value, nil)
-						if err != nil {
-							fmt.Fprintf(stdout, "%s: error decoding %s", w.fmtKey.fn(ukey), err)
-						} else {
-							fmt.Fprintf(stdout, "%s", s.Pretty(w.fmtKey.fn))
-						}
-					case base.InternalKeyKindDeleteSized:
-						v, _ := binary.Uvarint(value)
-						fmt.Fprintf(stdout, "%s,%d", w.fmtKey.fn(ukey), v)
-					}
-					fmt.Fprintf(stdout, ")\n")
-				}
+				fmt.Fprintf(stdout, "%d(%d) seq=%d count=%d, len=%d\n",
+					offset, len(b.Repr()), b.SeqNum(), b.Count(), buf.Len())
+				w.dumpBatch(stdout, &b, b.Reader(), func(err error) {
+					logErr(arg, offset, err)
+				})
 			}
 		}()
 	}
+	if len(errs) > 0 {
+		fmt.Fprintln(stderr, "Errors: ")
+		for _, ea := range errs {
+			fmt.Fprintf(stderr, "%s: %s\n", ea.arg, ea.err)
+		}
+	}
+}
+
+func (w *walT) runDumpMerged(cmd *cobra.Command, args []string) {
+	stdout, stderr := cmd.OutOrStdout(), cmd.OutOrStderr()
+	w.fmtKey.setForComparer(w.defaultComparer, w.comparers)
+	w.fmtValue.setForComparer(w.defaultComparer, w.comparers)
+	var a wal.FileAccumulator
+	for _, arg := range args {
+		isLog, err := a.MaybeAccumulate(w.opts.FS, arg)
+		if !isLog {
+			fmt.Fprintf(stderr, "%q does not parse as a log file\n", arg)
+			os.Exit(1)
+		} else if err != nil {
+			fmt.Fprintf(stderr, "%s: %s\n", arg, err)
+			os.Exit(1)
+		}
+	}
+	logs := a.Finish()
+	var errs []error
+	for _, log := range logs {
+		fmt.Fprintf(stdout, "log file %s contains %d segment files:\n", log.Num, log.NumSegments())
+		errs = append(errs, w.runDumpMergedOne(cmd, log)...)
+	}
+	if len(errs) > 0 {
+		fmt.Fprintln(stderr, "Errors: ")
+		for _, err := range errs {
+			fmt.Fprintf(stderr, "%s\n", err)
+		}
+	}
+}
+
+func (w *walT) runDumpMergedOne(cmd *cobra.Command, ll wal.LogicalLog) []error {
+	stdout, stderr := cmd.OutOrStdout(), cmd.OutOrStderr()
+	var buf bytes.Buffer
+	var errs []error
+	var b pebble.Batch
+
+	logErr := func(offset wal.Offset, err error) {
+		err = errors.Wrapf(err, "offset %s: ", offset)
+		fmt.Fprintln(stderr, err)
+		errs = append(errs, err)
+	}
+
+	rr := ll.OpenForRead()
+	for {
+		buf.Reset()
+		r, offset, err := rr.NextRecord()
+		if err == nil {
+			_, err = io.Copy(&buf, r)
+		}
+		if err != nil {
+			// It is common to encounter a zeroed or invalid chunk due to WAL
+			// preallocation and WAL recycling. We need to distinguish these
+			// errors from EOF in order to recognize that the record was
+			// truncated and to avoid replaying subsequent WALs, but want
+			// to otherwise treat them like EOF.
+			if err == io.EOF {
+				break
+			} else if record.IsInvalidRecord(err) {
+				break
+			}
+			return append(errs, err)
+		}
+		if buf.Len() < batchrepr.HeaderLen {
+			logErr(offset, errors.Newf("%d-byte batch too short", buf.Len()))
+			continue
+		}
+		b = pebble.Batch{}
+		if err := b.SetRepr(buf.Bytes()); err != nil {
+			logErr(offset, errors.Newf("unable to parse batch: %x", buf.Bytes()))
+			continue
+		}
+		fmt.Fprintf(stdout, "%s(%d) seq=%d count=%d, len=%d\n",
+			offset, len(b.Repr()), b.SeqNum(), b.Count(), buf.Len())
+		w.dumpBatch(stdout, &b, b.Reader(), func(err error) {
+			logErr(offset, err)
+		})
+	}
+	return nil
+}
+
+func (w *walT) dumpBatch(
+	stdout io.Writer, b *pebble.Batch, r batchrepr.Reader, logErr func(error),
+) {
+	for idx := 0; ; idx++ {
+		kind, ukey, value, ok, err := r.Next()
+		if !ok {
+			if err != nil {
+				logErr(errors.Newf("unable to decode %d'th key in batch; %s", idx, err))
+			}
+			break
+		}
+		fmt.Fprintf(stdout, "    %s(", kind)
+		switch kind {
+		case base.InternalKeyKindDelete:
+			fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
+		case base.InternalKeyKindSet:
+			fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
+		case base.InternalKeyKindMerge:
+			fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
+		case base.InternalKeyKindLogData:
+			fmt.Fprintf(stdout, "<%d>", len(value))
+		case base.InternalKeyKindIngestSST:
+			fileNum, _ := binary.Uvarint(ukey)
+			fmt.Fprintf(stdout, "%s", base.FileNum(fileNum))
+		case base.InternalKeyKindSingleDelete:
+			fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
+		case base.InternalKeyKindSetWithDelete:
+			fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
+		case base.InternalKeyKindRangeDelete:
+			fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtKey.fn(value))
+		case base.InternalKeyKindRangeKeySet, base.InternalKeyKindRangeKeyUnset, base.InternalKeyKindRangeKeyDelete:
+			ik := base.MakeInternalKey(ukey, b.SeqNum()+base.SeqNum(idx), kind)
+			s, err := rangekey.Decode(ik, value, nil)
+			if err != nil {
+				logErr(errors.Newf("%s: error decoding %s", w.fmtKey.fn(ukey), err))
+			} else {
+				fmt.Fprintf(stdout, "%s", s.Pretty(w.fmtKey.fn))
+			}
+		case base.InternalKeyKindDeleteSized:
+			v, _ := binary.Uvarint(value)
+			fmt.Fprintf(stdout, "%s,%d", w.fmtKey.fn(ukey), v)
+		default:
+			err := errors.Newf("invalid key kind %d in key at index %d/%d of batch with seqnum %d at offset %d",
+				kind, idx, b.Count(), b.SeqNum())
+			fmt.Fprintf(stdout, "<error: %s>", err)
+			logErr(err)
+		}
+		fmt.Fprintf(stdout, ")\n")
+	}
 }
diff --git a/vfs/disk_health.go b/vfs/disk_health.go
index 45da9ac6..d01218ff 100644
--- a/vfs/disk_health.go
+++ b/vfs/disk_health.go
@@ -581,6 +581,11 @@ func WithDiskHealthChecks(
 	return fs, fs
 }
 
+// Unwrap is part of the FS interface.
+func (d *diskHealthCheckingFS) Unwrap() FS {
+	return d.fs
+}
+
 // timeFilesystemOp executes the provided closure, which should perform a
 // singular filesystem operation of a type matching opType on the named file. It
 // records the provided start time such that the long-lived disk-health checking
diff --git a/vfs/mem_fs.go b/vfs/mem_fs.go
index 72827c0b..1bb5fd8e 100644
--- a/vfs/mem_fs.go
+++ b/vfs/mem_fs.go
@@ -564,6 +564,9 @@ func (*MemFS) GetDiskUsage(string) (DiskUsage, error) {
 	return DiskUsage{}, ErrUnsupported
 }
 
+// Unwrap implements FS.Unwrap.
+func (*MemFS) Unwrap() FS { return nil }
+
 // memNode holds a file's data or a directory's children.
 type memNode struct {
 	isDir bool
@@ -721,6 +724,9 @@ func (f *memFile) Write(p []byte) (int, error) {
 			panic("stuff")
 		}
 	} else {
+		if grow := f.pos - len(f.n.mu.data); grow > 0 {
+			f.n.mu.data = append(f.n.mu.data, make([]byte, grow)...)
+		}
 		f.n.mu.data = append(f.n.mu.data[:f.pos], p...)
 	}
 	f.pos += len(p)
diff --git a/vfs/vfs.go b/vfs/vfs.go
index 36ccc89e..b1255df6 100644
--- a/vfs/vfs.go
+++ b/vfs/vfs.go
@@ -176,6 +176,13 @@ type FS interface {
 	// GetDiskUsage returns disk space statistics for the filesystem where
 	// path is any file or directory within that filesystem.
 	GetDiskUsage(path string) (DiskUsage, error)
+
+	// Unwrap is implemented by "wrapping" filesystems (those that add some
+	// functionality on top of an underlying FS); it returns the wrapped FS.
+	// It is used by vfs.Root.
+	//
+	// Returns nil if this is not a wrapping filesystem.
+	Unwrap() FS
 }
 
 // A DeviceID uniquely identifies a block device on which filesystem data is
@@ -344,6 +351,8 @@ func (defaultFS) PathDir(path string) string {
 	return filepath.Dir(path)
 }
 
+func (defaultFS) Unwrap() FS { return nil }
+
 type randomReadsOption struct{}
 
 // RandomReadsOption is an OpenOption that optimizes opened file handle for
@@ -444,18 +453,13 @@ func LinkOrCopy(fs FS, oldname, newname string) error {
 // Root returns the base FS implementation, unwrapping all nested FSs that
 // expose an Unwrap method.
 func Root(fs FS) FS {
-	type unwrapper interface {
-		Unwrap() FS
-	}
-
 	for {
-		u, ok := fs.(unwrapper)
-		if !ok {
-			break
+		n := fs.Unwrap()
+		if n == nil {
+			return fs
 		}
-		fs = u.Unwrap()
+		fs = n
 	}
-	return fs
 }
 
 // ErrUnsupported may be returned a FS when it does not support an operation.


Test patch:
diff --git a/data_test.go b/data_test.go
index 3e356b2d..3679a39d 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.Raw().Add(tmp, kv.InPlaceValue()); err != nil {
+		if err := w.Raw().AddWithForceObsolete(tmp, kv.InPlaceValue(), false); 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.Raw().Add(k, v)
+				return w.Raw().AddWithForceObsolete(k, v, false)
 			})
 			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.Raw().Add(tmp, kv.InPlaceValue()); err != nil {
+		if err := w.Raw().AddWithForceObsolete(tmp, kv.InPlaceValue(), false); 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.Raw().Add(k, v)
+				return w.Raw().AddWithForceObsolete(k, v, false)
 			})
 			if err != nil {
 				return err
diff --git a/ingest_test.go b/ingest_test.go
index fd703e60..200365ba 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -3224,7 +3224,7 @@ func TestIngestFileNumReuseCrash(t *testing.T) {
 func TestIngest_UpdateSequenceNumber(t *testing.T) {
 	mem := vfs.NewMem()
 	cmp := base.DefaultComparer.Compare
-	parse := func(input string) (*sstable.RawWriter, error) {
+	parse := func(input string) (*sstable.RawRowWriter, error) {
 		f, err := mem.Create("ext", vfs.WriteCategoryUnspecified)
 		if err != nil {
 			return nil, err
diff --git a/level_iter_test.go b/level_iter_test.go
index 9b105e1e..471e8c42 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -508,7 +508,7 @@ func buildLevelIterTables(
 		files[i] = f
 	}
 
-	writers := make([]*sstable.RawWriter, len(files))
+	writers := make([]*sstable.RawRowWriter, len(files))
 	for i := range files {
 		writers[i] = sstable.NewRawWriter(objstorageprovider.NewFileWritable(files[i]), sstable.WriterOptions{
 			BlockRestartInterval: restartInterval,
diff --git a/merging_iter_test.go b/merging_iter_test.go
index 7a0b912f..2e2ed5de 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -340,7 +340,7 @@ func buildMergingIterTables(
 		files[i] = f
 	}
 
-	writers := make([]*sstable.RawWriter, len(files))
+	writers := make([]*sstable.RawRowWriter, len(files))
 	for i := range files {
 		writers[i] = sstable.NewRawWriter(objstorageprovider.NewFileWritable(files[i]), sstable.WriterOptions{
 			BlockRestartInterval: restartInterval,
@@ -543,7 +543,7 @@ func buildLevelsForMergingIterSeqSeek(
 	}
 
 	const targetL6FirstFileSize = 2 << 20
-	writers := make([][]*sstable.RawWriter, levelCount)
+	writers := make([][]*sstable.RawRowWriter, levelCount)
 	// A policy unlikely to have false positives.
 	filterPolicy := bloom.FilterPolicy(100)
 	for i := range files {
diff --git a/metamorphic/test.go b/metamorphic/test.go
index 8a3c4e09..48ab870a 100644
--- a/metamorphic/test.go
+++ b/metamorphic/test.go
@@ -267,15 +267,28 @@ func (t *Test) minFMV() pebble.FormatMajorVersion {
 
 func (t *Test) restartDB(dbID objID) error {
 	db := t.getDB(dbID)
+	// If strictFS is not used, we use pebble.NoSync for writeOpts, so we can't
+	// restart the database (even if we don't revert to synced data).
 	if !t.testOpts.strictFS {
 		return nil
 	}
-	t.opts.Cache.Ref()
-	// The fs isn't necessarily a MemFS.
-	fs, ok := vfs.Root(t.opts.FS).(*vfs.MemFS)
-	if ok {
-		fs.SetIgnoreSyncs(true)
+	if t.testOpts.sharedStorageEnabled {
+		// We simulate a crash by essentially ignoring writes to disk after a
+		// certain point. However, we cannot prevent the process (which didn't
+		// actually crash) from deleting an external object before we call Close().
+		// TODO(radu): perhaps we want all syncs to fail after the "crash" point?
+		return nil
+	}
+	// We can't do this if we have more than one database since they share the
+	// same FS (and we only close/reopen one of them).
+	// TODO(radu): have each database use its own MemFS.
+	if len(t.dbs) > 1 {
+		return nil
 	}
+	t.opts.Cache.Ref()
+
+	memFS := vfs.Root(t.opts.FS).(*vfs.MemFS)
+	memFS.SetIgnoreSyncs(true)
 	if err := db.Close(); err != nil {
 		return err
 	}
@@ -287,10 +300,9 @@ func (t *Test) restartDB(dbID objID) error {
 			return err
 		}
 	}
-	if ok {
-		fs.ResetToSyncedState()
-		fs.SetIgnoreSyncs(false)
-	}
+
+	memFS.ResetToSyncedState()
+	memFS.SetIgnoreSyncs(false)
 
 	// TODO(jackson): Audit errorRate and ensure custom options' hooks semantics
 	// are well defined within the context of retries.
diff --git a/scan_internal_test.go b/scan_internal_test.go
index 9a7fff1d..e8e6255a 100644
--- a/scan_internal_test.go
+++ b/scan_internal_test.go
@@ -441,7 +441,7 @@ func TestScanInternal(t *testing.T) {
 					var err error
 					value, _, err = kv.Value(value)
 					require.NoError(t, err)
-					require.NoError(t, w.Raw().Add(kv.K, value))
+					require.NoError(t, w.Raw().AddWithForceObsolete(kv.K, value, false))
 				}
 				points.Close()
 				require.NoError(t, w.Close())
diff --git a/sstable/colblk/bitmap_test.go b/sstable/colblk/bitmap_test.go
index 58c5e320..e544487f 100644
--- a/sstable/colblk/bitmap_test.go
+++ b/sstable/colblk/bitmap_test.go
@@ -32,18 +32,26 @@ func TestBitmapFixed(t *testing.T) {
 					continue
 				}
 				if r == '1' {
-					builder.Set(n, r == '1')
+					builder.Set(n)
 				}
 				n++
 			}
 			td.MaybeScanArgs(t, "rows", &n)
 
-			data := make([]byte, builder.Size(n, 0))
+			size := builder.Size(n, 0)
 			if td.HasArg("invert") {
+				size = builder.InvertedSize(n, 0)
 				builder.Invert(n)
+				if newSize := builder.Size(n, 0); size != newSize {
+					td.Fatalf(t, "InvertedSize=%d, after invert Size=%d", size, newSize)
+				}
 			}
+			data := make([]byte, builder.Size(n, size))
 
-			_ = builder.Finish(0, n, 0, data)
+			endOffset := builder.Finish(0, n, 0, data)
+			if endOffset != size {
+				td.Fatalf(t, "endOffset=%d size=%d", endOffset, size)
+			}
 			bitmap, _ = DecodeBitmap(data, 0, n)
 			dumpBitmap(&buf, bitmap)
 			fmt.Fprint(&buf, "\nBinary representation:\n")
@@ -96,7 +104,7 @@ func TestBitmapRandom(t *testing.T) {
 		for i := 0; i < size; i++ {
 			v[i] = rng.Float64() < p
 			if v[i] {
-				builder.Set(i, v[i])
+				builder.Set(i)
 			}
 		}
 		data := make([]byte, builder.Size(size, 0))
@@ -163,7 +171,7 @@ func BenchmarkBitmapBuilder(b *testing.B) {
 		var builder BitmapBuilder
 		for i := 0; i < size; i++ {
 			if v[i] {
-				builder.Set(i, v[i])
+				builder.Set(i)
 			}
 		}
 		_ = builder.Finish(0, size, 0, data)
diff --git a/sstable/colblk/block_test.go b/sstable/colblk/block_test.go
index 3f077d3e..a49a8dd2 100644
--- a/sstable/colblk/block_test.go
+++ b/sstable/colblk/block_test.go
@@ -130,7 +130,9 @@ func TestBlockWriter(t *testing.T) {
 					for r := range lineFields {
 						v, err := strconv.ParseBool(lineFields[r][c])
 						panicIfErr(dataType, lineFields[r][c], err)
-						bb.Set(r, v)
+						if v {
+							bb.Set(r)
+						}
 					}
 				case DataTypeUint:
 					b := colWriters[c].(*UintBuilder)
@@ -221,7 +223,9 @@ func buildBlock(schema []testColumnSpec, rows int, data []interface{}) []byte {
 			var bb BitmapBuilder
 			bb.Reset()
 			for row, v := range data[col].([]bool) {
-				bb.Set(row, v)
+				if v {
+					bb.Set(row)
+				}
 			}
 			cw[col] = &bb
 		case DataTypeUint:
diff --git a/sstable/colblk/cockroach_test.go b/sstable/colblk/cockroach_test.go
index f23272df..6451fdce 100644
--- a/sstable/colblk/cockroach_test.go
+++ b/sstable/colblk/cockroach_test.go
@@ -270,25 +270,42 @@ func (ks *cockroachKeySeeker) MaterializeUserKey(ki *PrefixBytesIter, prevRow, r
 		ks.prefixes.SetAt(ki, row)
 	}
 
+	ptr := unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(ki.buf))) + uintptr(len(ki.buf)))
 	mvccWall := ks.mvccWallTimes.At(row)
 	mvccLogical := uint32(ks.mvccLogical.At(row))
 	if mvccWall == 0 && mvccLogical == 0 {
 		// This is not an MVCC key. Use the untyped suffix.
 		untypedSuffixed := ks.untypedSuffixes.At(row)
-		memmove(unsafe.Pointer(uintptr(ki.ptr)+uintptr(ki.len)), unsafe.Pointer(unsafe.SliceData(untypedSuffixed)), uintptr(len(untypedSuffixed)))
-		return unsafe.Slice((*byte)(ki.ptr), ki.len+len(untypedSuffixed))
+		res := ki.buf[:len(ki.buf)+len(untypedSuffixed)]
+		memmove(ptr, unsafe.Pointer(unsafe.SliceData(untypedSuffixed)), uintptr(len(untypedSuffixed)))
+		return res
 	}
 
+	// Inline binary.BigEndian.PutUint64. Note that this code is converted into
+	// word-size instructions by the compiler.
+	*(*byte)(ptr) = byte(mvccWall >> 56)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 1)) = byte(mvccWall >> 48)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 2)) = byte(mvccWall >> 40)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 3)) = byte(mvccWall >> 32)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 4)) = byte(mvccWall >> 24)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 5)) = byte(mvccWall >> 16)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 6)) = byte(mvccWall >> 8)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 7)) = byte(mvccWall)
+
+	ptr = unsafe.Pointer(uintptr(ptr) + 8)
 	// This is an MVCC key.
 	if mvccLogical == 0 {
-		binary.BigEndian.PutUint64(unsafe.Slice((*byte)(unsafe.Pointer(uintptr(ki.ptr)+uintptr(ki.len))), 8), mvccWall)
-		*(*byte)(unsafe.Pointer(uintptr(ki.ptr) + uintptr(ki.len) + 8)) = 9
-		return unsafe.Slice((*byte)(ki.ptr), ki.len+9)
+		*(*byte)(ptr) = 9
+		return ki.buf[:len(ki.buf)+9]
 	}
-	binary.BigEndian.PutUint64(unsafe.Slice((*byte)(unsafe.Pointer(uintptr(ki.ptr)+uintptr(ki.len))), 8), mvccWall)
-	binary.BigEndian.PutUint32(unsafe.Slice((*byte)(unsafe.Pointer(uintptr(ki.ptr)+uintptr(ki.len+8))), 4), mvccLogical)
-	*(*byte)(unsafe.Pointer(uintptr(ki.ptr) + uintptr(ki.len) + 12)) = 13
-	return unsafe.Slice((*byte)(ki.ptr), ki.len+13)
+
+	// Inline binary.BigEndian.PutUint32.
+	*(*byte)(ptr) = byte(mvccWall >> 24)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 1)) = byte(mvccWall >> 16)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 2)) = byte(mvccWall >> 8)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 3)) = byte(mvccWall)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 4)) = 13
+	return ki.buf[:len(ki.buf)+13]
 }
 
 // Release is part of the KeySeeker interface.
diff --git a/sstable/colblk/index_block_test.go b/sstable/colblk/index_block_test.go
new file mode 100644
index 00000000..7b06a143
--- /dev/null
+++ b/sstable/colblk/index_block_test.go
@@ -0,0 +1,85 @@
+// 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 colblk
+
+import (
+	"bytes"
+	"fmt"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIndexBlock(t *testing.T) {
+	var r IndexReader
+	var buf bytes.Buffer
+	datadriven.RunTest(t, "testdata/index_block", func(t *testing.T, d *datadriven.TestData) string {
+		buf.Reset()
+		switch d.Cmd {
+		case "build":
+			var w IndexBlockWriter
+			w.Init()
+			for _, line := range strings.Split(d.Input, "\n") {
+				fields := strings.Fields(line)
+				var err error
+				var h block.Handle
+				h.Offset, err = strconv.ParseUint(fields[1], 10, 64)
+				require.NoError(t, err)
+				h.Length, err = strconv.ParseUint(fields[2], 10, 64)
+				require.NoError(t, err)
+				var bp []byte
+				if len(fields) > 3 {
+					bp = []byte(fields[3])
+				}
+				w.AddBlockHandle([]byte(fields[0]), h, bp)
+			}
+			data := w.Finish()
+			r.Init(data)
+			return r.DebugString()
+		case "iter":
+			var it IndexIter
+			it.Init(&r)
+			for _, line := range strings.Split(d.Input, "\n") {
+				fields := strings.Fields(line)
+				var valid bool
+				switch fields[0] {
+				case "seek-ge":
+					valid = it.SeekGE([]byte(fields[1]))
+				case "first":
+					valid = it.First()
+				case "last":
+					valid = it.Last()
+				case "next":
+					valid = it.Next()
+				case "prev":
+					valid = it.Prev()
+				default:
+					panic(fmt.Sprintf("unknown command: %s", fields[0]))
+				}
+				if valid {
+					var bp string
+					bhp, err := it.BlockHandleWithProperties()
+					if err != nil {
+						fmt.Fprintf(&buf, "<err invalid bh: %s>", err)
+						continue
+					}
+					if len(bhp.Props) > 0 {
+						bp = fmt.Sprintf(" props=%q", bhp.Props)
+					}
+					fmt.Fprintf(&buf, "block %d: %d-%d%s\n", it.row, bhp.Offset, bhp.Offset+bhp.Length, bp)
+				} else {
+					fmt.Fprintln(&buf, ".")
+				}
+			}
+			return buf.String()
+		default:
+			panic(fmt.Sprintf("unknown command: %s", d.Cmd))
+		}
+	})
+}
diff --git a/sstable/colblk/prefix_bytes_test.go b/sstable/colblk/prefix_bytes_test.go
index c41bbbd3..cf747d11 100644
--- a/sstable/colblk/prefix_bytes_test.go
+++ b/sstable/colblk/prefix_bytes_test.go
@@ -276,7 +276,7 @@ func BenchmarkPrefixBytes(b *testing.B) {
 			pb, _ := DecodePrefixBytes(buf, 0, n)
 			b.ResetTimer()
 			var pbi PrefixBytesIter
-			pbi.Alloc(maxLen)
+			pbi.buf = make([]byte, 0, maxLen)
 			for i := 0; i < b.N; i++ {
 				j := i % n
 				if j == 0 {
@@ -284,9 +284,9 @@ func BenchmarkPrefixBytes(b *testing.B) {
 				} else {
 					pb.SetNext(&pbi)
 				}
-				if invariants.Enabled && !bytes.Equal(pbi.UnsafeSlice(), userKeys[j]) {
+				if invariants.Enabled && !bytes.Equal(pbi.buf, userKeys[j]) {
 					b.Fatalf("Constructed key %q (%q, %q, %q) for index %d; expected %q",
-						pbi.UnsafeSlice(), pb.SharedPrefix(), pb.RowBundlePrefix(j), pb.RowSuffix(j), j, userKeys[j])
+						pbi.buf, pb.SharedPrefix(), pb.RowBundlePrefix(j), pb.RowSuffix(j), j, userKeys[j])
 				}
 			}
 		})
diff --git a/sstable/colblk/testdata/bitmap b/sstable/colblk/testdata/bitmap
index 36ccb5fa..74239c79 100644
--- a/sstable/colblk/testdata/bitmap
+++ b/sstable/colblk/testdata/bitmap
@@ -1,10 +1,48 @@
+build
+00000
+----
+00000
+Binary representation:
+0-1: x 01 # bitmap encoding
+
+build invert
+00000
+----
+11111
+Binary representation:
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0001111100000000000000000000000000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+
+# An inverted bitmap never uses the zero encoding.
+build invert
+11111
+----
+00000
+Binary representation:
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+
+build rows=100
+0
+----
+0000000000000000000000000000000000000000000000000000000000000000
+000000000000000000000000000000000000
+Binary representation:
+0-1: x 01 # bitmap encoding
+
 build
 10101011100011100
 ----
 10101011100011100
 Binary representation:
-00-08: b 1101010101110001000000000000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 1101010101110001000000000000000000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 successor indexes=(0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16)
 ----
@@ -53,8 +91,10 @@ build invert
 ----
 01010100011100011
 Binary representation:
-00-08: b 0010101010001110000000010000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0010101010001110000000010000000000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 
 build
@@ -62,8 +102,10 @@ build
 ----
 1
 Binary representation:
-00-08: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # Test sparseness; relying on the tailing bits being implicitly zeroed.
 
@@ -79,23 +121,27 @@ build rows=512
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000
 Binary representation:
-00-08: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
-16-24: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
-24-32: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 3
-32-40: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 4
-40-48: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 5
-48-56: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 6
-56-64: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 7
-64-72: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+000-001: x 00                                                               # bitmap encoding
+001-008: x 00000000000000                                                   # padding to align to 64-bit boundary
+008-016: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap word 0
+016-024: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
+024-032: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
+032-040: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 3
+040-048: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 4
+048-056: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 5
+056-064: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 6
+064-072: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 7
+072-080: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 build invert
 1
 ----
 0
 Binary representation:
-00-08: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # Test sparseness with inversion, relying on the trailing bits being implicitly
 # set to one.
@@ -112,15 +158,17 @@ build invert rows=512
 1111111111111111111111111111111111111111111111111111111111111111
 1111111111111111111111111111111111111111111111111111111111111111
 Binary representation:
-00-08: b 1111111011111111111111111111111111111111111111111111111111111111 # bitmap word 0
-08-16: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 1
-16-24: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 2
-24-32: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 3
-32-40: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 4
-40-48: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 5
-48-56: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 6
-56-64: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 7
-64-72: b 1111111100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+000-001: x 00                                                               # bitmap encoding
+001-008: x 00000000000000                                                   # padding to align to 64-bit boundary
+008-016: b 1111111011111111111111111111111111111111111111111111111111111111 # bitmap word 0
+016-024: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 1
+024-032: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 2
+032-040: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 3
+040-048: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 4
+048-056: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 5
+056-064: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 6
+064-072: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 7
+072-080: b 1111111100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # 32-bits wide
 
@@ -129,8 +177,10 @@ build
 ----
 1010101010111111111110000001110
 Binary representation:
-00-08: b 0101010111111101000111110011100000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0101010111111101000111110011100000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # 33-bits wide
 
@@ -139,8 +189,10 @@ build
 ----
 10101010101111111111100000011101
 Binary representation:
-00-08: b 0101010111111101000111111011100000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0101010111111101000111111011100000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # 64-bits wide
 
@@ -149,8 +201,10 @@ build
 ----
 1010101010111111111110000001110101010101011111111111000000111011
 Binary representation:
-00-08: b 0101010111111101000111111011100010101010111111100000111111011100 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0101010111111101000111111011100010101010111111100000111111011100 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # 63-bits wide
 
@@ -159,8 +213,10 @@ build
 ----
 101010101011111111111000000111010101010101111111111100000011101
 Binary representation:
-00-08: b 0101010111111101000111111011100010101010111111100000111101011100 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0101010111111101000111111011100010101010111111100000111101011100 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # 65-bits wide
 
@@ -171,9 +227,11 @@ build
 1010101010111111111110000001110101010101011111111111000000111011
 1
 Binary representation:
-00-08: b 0101010111111101000111111011100010101010111111100000111111011100 # bitmap word 0
-08-16: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap word 1
-16-24: b 0000001100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0101010111111101000111111011100010101010111111100000111111011100 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap word 1
+24-32: b 0000001100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 build
 1111111111111111111111111111111111111111111111111111111111111111
@@ -188,12 +246,14 @@ build
 1111111111111111111111111111111111111111111111111111111111111111
 1111111111111111111111111111111111111111111111111111111111111111
 Binary representation:
-00-08: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 0
-08-16: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 1
-16-24: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 2
-24-32: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 3
-32-40: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 4
-40-48: b 0001111100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+000-001: x 00                                                               # bitmap encoding
+001-008: x 00000000000000                                                   # padding to align to 64-bit boundary
+008-016: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 0
+016-024: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 1
+024-032: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 2
+032-040: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 3
+040-048: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 4
+048-056: b 0001111100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 build
 0000000000000000000000000000000000000000000000000000000000000000
@@ -208,12 +268,7 @@ build
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000
 Binary representation:
-00-08: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
-16-24: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
-24-32: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 3
-32-40: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 4
-40-48: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+0-1: x 01 # bitmap encoding
 
 build
 0000000000000000000000000000000000000000000000000000000000000000
@@ -228,12 +283,14 @@ build
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000001
 Binary representation:
-00-08: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
-16-24: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
-24-32: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 3
-32-40: b 0000000000000000000000000000000000000000000000000000000010000000 # bitmap word 4
-40-48: b 0001000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+000-001: x 00                                                               # bitmap encoding
+001-008: x 00000000000000                                                   # padding to align to 64-bit boundary
+008-016: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
+016-024: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
+024-032: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
+032-040: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 3
+040-048: b 0000000000000000000000000000000000000000000000000000000010000000 # bitmap word 4
+048-056: b 0001000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
 # Write out fewer rows than we set. The bitmap summary should reflect the
 # truncated view of the bitmap.
@@ -248,7 +305,9 @@ build rows=192
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000
 Binary representation:
-00-08: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
-16-24: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
-24-32: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
+16-24: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
+24-32: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
+32-40: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
diff --git a/sstable/colblk/testdata/block_writer b/sstable/colblk/testdata/block_writer
index f86c5e79..b7b3a7a9 100644
--- a/sstable/colblk/testdata/block_writer
+++ b/sstable/colblk/testdata/block_writer
@@ -273,7 +273,8 @@ finish
 07-08: b 00000001                                                         # bool
 08-12: x 0c000000                                                         # page start 12
 # data for column 0
-12-16: x 00000000                                                         # padding to align to 64-bit boundary
+12-13: x 00                                                               # bitmap encoding
+13-16: x 000000                                                           # padding to align to 64-bit boundary
 16-24: b 1011011101111101110110110000000000000000000000000000000000000000 # bitmap word 0
 24-32: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 32-33: x 00                                                               # block trailer padding
diff --git a/sstable/colblk/testdata/data_block/bundle_search b/sstable/colblk/testdata/data_block/bundle_search
index c23b7b6a..6a10e82f 100644
--- a/sstable/colblk/testdata/data_block/bundle_search
+++ b/sstable/colblk/testdata/data_block/bundle_search
@@ -10,7 +10,7 @@
 
 init bundle-size=4
 ----
-size=49:
+size=50:
 0: prefixes:       prefixbytes(4): 0 keys
 1: suffixes:       bytes: 0 rows set; 0 bytes in data
 2: trailers:       uint: 0 rows
@@ -86,7 +86,7 @@ backwash#1,SET:v
 backwoods#1,SET:v
 bacteria#1,SET:v
 ----
-size=745:
+size=720:
 0: prefixes:       prefixbytes(4): 66 keys
 1: suffixes:       bytes: 66 rows set; 0 bytes in data
 2: trailers:       uint: 66 rows
@@ -371,7 +371,7 @@ finish
 550-551: x 80                                                               # encoding: const
 551-559: x 0101000000000000                                                 # 64-bit constant: 257
 # data for column 3
-559-560: x 00                                                               # padding to align to 64-bit boundary
+559-560: x 00                                                               # bitmap encoding
 560-568: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 0
 568-576: b 0000001100000000000000000000000000000000000000000000000000000000 # bitmap word 1
 576-584: b 0000001100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
@@ -514,10 +514,7 @@ finish
 716-717: x 76                                                               # data[64]: v
 717-718: x 76                                                               # data[65]: v
 # data for column 5
-718-720: x 0000                                                             # padding to align to 64-bit boundary
-720-728: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-728-736: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 1
-736-744: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+718-719: x 01                                                               # bitmap encoding
 
 iter
 seek-ge backache
diff --git a/sstable/colblk/testdata/data_block/external_value b/sstable/colblk/testdata/data_block/external_value
index 301f56c6..9b9a36f7 100644
--- a/sstable/colblk/testdata/data_block/external_value
+++ b/sstable/colblk/testdata/data_block/external_value
@@ -1,6 +1,6 @@
 init
 ----
-size=49:
+size=50:
 0: prefixes:       prefixbytes(16): 0 keys
 1: suffixes:       bytes: 0 rows set; 0 bytes in data
 2: trailers:       uint: 0 rows
@@ -187,7 +187,8 @@ finish
 224-226: x 0100                                                             # data[18] = 1
 226-228: x 0000                                                             # data[19] = 0
 # data for column 3
-228-232: x 00000000                                                         # padding to align to 64-bit boundary
+228-229: x 00                                                               # bitmap encoding
+229-232: x 000000                                                           # padding to align to 64-bit boundary
 232-240: b 0001000100000100000010110000000000000000000000000000000000000000 # bitmap word 0
 240-248: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 # data for column 4
@@ -260,7 +261,7 @@ finish
 614-623: x 6c652d6b6977693938                                               # (continued...)
 623-623: x                                                                  # data[19]:
 # data for column 5
-623-624: x 00                                                               # padding to align to 64-bit boundary
+623-624: x 00                                                               # bitmap encoding
 624-632: b 1100111011111011000001110000000000000000000000000000000000000000 # bitmap word 0
 632-640: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
diff --git a/sstable/colblk/testdata/data_block/next_prefix b/sstable/colblk/testdata/data_block/next_prefix
index f56cd1f2..b1efbcb0 100644
--- a/sstable/colblk/testdata/data_block/next_prefix
+++ b/sstable/colblk/testdata/data_block/next_prefix
@@ -1,6 +1,6 @@
 init
 ----
-size=49:
+size=50:
 0: prefixes:       prefixbytes(16): 0 keys
 1: suffixes:       bytes: 0 rows set; 0 bytes in data
 2: trailers:       uint: 0 rows
@@ -30,7 +30,7 @@ blockprefix_kiwi@99#0,SET:kiwi99
 blockprefix_kiwi@98#0,SET:kiwi98
 blockprefix_lemon@92#0,DEL:
 ----
-size=425:
+size=408:
 0: prefixes:       prefixbytes(16): 20 keys
 1: suffixes:       bytes: 20 rows set; 54 bytes in data
 2: trailers:       uint: 20 rows
@@ -187,7 +187,8 @@ finish
 224-226: x 0100                                                             # data[18] = 1
 226-228: x 0000                                                             # data[19] = 0
 # data for column 3
-228-232: x 00000000                                                         # padding to align to 64-bit boundary
+228-229: x 00                                                               # bitmap encoding
+229-232: x 000000                                                           # padding to align to 64-bit boundary
 232-240: b 0001000100000100000010110000000000000000000000000000000000000000 # bitmap word 0
 240-248: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 # data for column 4
@@ -237,9 +238,7 @@ finish
 400-406: x 6b6977693938                                                     # data[18]: kiwi98
 406-406: x                                                                  # data[19]:
 # data for column 5
-406-408: x 0000                                                             # padding to align to 64-bit boundary
-408-416: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-416-424: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+406-407: x 01                                                               # bitmap encoding
 
 # Scan across the block using next prefix.
 
diff --git a/sstable/colblk/testdata/data_block/simple b/sstable/colblk/testdata/data_block/simple
index b1f304a2..c3594142 100644
--- a/sstable/colblk/testdata/data_block/simple
+++ b/sstable/colblk/testdata/data_block/simple
@@ -1,6 +1,6 @@
 init
 ----
-size=49:
+size=50:
 0: prefixes:       prefixbytes(16): 0 keys
 1: suffixes:       bytes: 0 rows set; 0 bytes in data
 2: trailers:       uint: 0 rows
@@ -16,7 +16,7 @@ c@9#0,SETWITHDEL:coconut
 c@6#0,SET:cantelope
 c@1#0,SET:clementine
 ----
-size=177:
+size=160:
 0: prefixes:       prefixbytes(16): 6 keys
 1: suffixes:       bytes: 6 rows set; 13 bytes in data
 2: trailers:       uint: 6 rows
@@ -27,7 +27,7 @@ size=177:
 write
 d@11#0,DEL:
 ----
-size=185:
+size=169:
 0: prefixes:       prefixbytes(16): 7 keys
 1: suffixes:       bytes: 7 rows set; 16 bytes in data
 2: trailers:       uint: 7 rows
@@ -115,7 +115,8 @@ finish
 087-088: x 01                                                               # data[5] = 1
 088-089: x 00                                                               # data[6] = 0
 # data for column 3
-089-096: x 00000000000000                                                   # padding to align to 64-bit boundary
+089-090: x 00                                                               # bitmap encoding
+090-096: x 000000000000                                                     # padding to align to 64-bit boundary
 096-104: b 0100101100000000000000000000000000000000000000000000000000000000 # bitmap word 0
 104-112: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 # data for column 4
@@ -139,9 +140,7 @@ finish
 157-167: x 636c656d656e74696e65                                             # data[5]: clementine
 167-167: x                                                                  # data[6]:
 # data for column 5
-167-168: x 00                                                               # padding to align to 64-bit boundary
-168-176: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-176-184: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+167-168: x 01                                                               # bitmap encoding
 
 iter
 first
@@ -299,7 +298,7 @@ d@11:
 
 init
 ----
-size=49:
+size=50:
 0: prefixes:       prefixbytes(16): 0 keys
 1: suffixes:       bytes: 0 rows set; 0 bytes in data
 2: trailers:       uint: 0 rows
@@ -326,7 +325,7 @@ aaaaaaaaaaaaaaapproves@10#0,SET:a
 aaaaaaaaaaaaaaarresting@10#0,SET:a
 aaaaaaaaaaaaaaarrived@10#0,SET:a
 ----
-size=353:
+size=334:
 0: prefixes:       prefixbytes(16): 17 keys
 1: suffixes:       bytes: 17 rows set; 51 bytes in data
 2: trailers:       uint: 17 rows
@@ -451,7 +450,8 @@ finish
 265-266: x 80                                                               # encoding: const
 266-274: x 0100000000000000                                                 # 64-bit constant: 1
 # data for column 3
-274-280: x 000000000000                                                     # padding to align to 64-bit boundary
+274-275: x 00                                                               # bitmap encoding
+275-280: x 0000000000                                                       # padding to align to 64-bit boundary
 280-288: b 1111111111111111000000010000000000000000000000000000000000000000 # bitmap word 0
 288-296: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 # data for column 4
@@ -495,9 +495,7 @@ finish
 330-331: x 61                                                               # data[15]: a
 331-332: x 61                                                               # data[16]: a
 # data for column 5
-332-336: x 00000000                                                         # padding to align to 64-bit boundary
-336-344: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 0
-344-352: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+332-333: x 01                                                               # bitmap encoding
 
 iter
 seek-ge aaa
diff --git a/sstable/colblk/testdata/index_block b/sstable/colblk/testdata/index_block
new file mode 100644
index 00000000..c7bf9b0e
--- /dev/null
+++ b/sstable/colblk/testdata/index_block
@@ -0,0 +1,231 @@
+build
+apple       24   24    bp1
+applied    141  112    bp2
+atone      195   49    bp3
+bacitracin 412  212
+banana     632  215    bp5
+bonifide   963  326    bp6
+----
+# index block header
+# columnar block header
+000-001: x 01                   # version 1
+001-003: x 0400                 # 4 columns
+003-007: x 06000000             # 6 rows
+# column 0
+007-008: b 00000011             # bytes
+008-012: x 1b000000             # page start 27
+# column 1
+012-013: b 00000010             # uint
+013-017: x 4c000000             # page start 76
+# column 2
+017-018: b 00000010             # uint
+018-022: x 5a000000             # page start 90
+# column 3
+022-023: b 00000011             # bytes
+023-027: x 68000000             # page start 104
+# data for column 0
+# rawbytes
+# offsets table
+027-028: x 01                   # encoding: 1b
+028-029: x 00                   # data[0] = 0 [35 overall]
+029-030: x 05                   # data[1] = 5 [40 overall]
+030-031: x 0c                   # data[2] = 12 [47 overall]
+031-032: x 11                   # data[3] = 17 [52 overall]
+032-033: x 1b                   # data[4] = 27 [62 overall]
+033-034: x 21                   # data[5] = 33 [68 overall]
+034-035: x 29                   # data[6] = 41 [76 overall]
+# data
+035-040: x 6170706c65           # data[0]: apple
+040-047: x 6170706c696564       # data[1]: applied
+047-052: x 61746f6e65           # data[2]: atone
+052-062: x 6261636974726163696e # data[3]: bacitracin
+062-068: x 62616e616e61         # data[4]: banana
+068-076: x 626f6e6966696465     # data[5]: bonifide
+# data for column 1
+076-077: x 02                   # encoding: 2b
+077-078: x 00                   # padding (aligning to 16-bit boundary)
+078-080: x 1800                 # data[0] = 24
+080-082: x 8d00                 # data[1] = 141
+082-084: x c300                 # data[2] = 195
+084-086: x 9c01                 # data[3] = 412
+086-088: x 7802                 # data[4] = 632
+088-090: x c303                 # data[5] = 963
+# data for column 2
+090-091: x 02                   # encoding: 2b
+091-092: x 00                   # padding (aligning to 16-bit boundary)
+092-094: x 1800                 # data[0] = 24
+094-096: x 7000                 # data[1] = 112
+096-098: x 3100                 # data[2] = 49
+098-100: x d400                 # data[3] = 212
+100-102: x d700                 # data[4] = 215
+102-104: x 4601                 # data[5] = 326
+# data for column 3
+# rawbytes
+# offsets table
+104-105: x 01                   # encoding: 1b
+105-106: x 00                   # data[0] = 0 [112 overall]
+106-107: x 03                   # data[1] = 3 [115 overall]
+107-108: x 06                   # data[2] = 6 [118 overall]
+108-109: x 09                   # data[3] = 9 [121 overall]
+109-110: x 09                   # data[4] = 9 [121 overall]
+110-111: x 0c                   # data[5] = 12 [124 overall]
+111-112: x 0f                   # data[6] = 15 [127 overall]
+# data
+112-115: x 627031               # data[0]: bp1
+115-118: x 627032               # data[1]: bp2
+118-121: x 627033               # data[2]: bp3
+121-121: x                      # data[3]:
+121-124: x 627035               # data[4]: bp5
+124-127: x 627036               # data[5]: bp6
+
+iter
+seek-ge zoo
+seek-ge bitter
+seek-ge abbey
+seek-ge banana
+seek-ge babble
+seek-ge assumption
+----
+.
+block 5: 963-1289 props="bp6"
+block 0: 24-48 props="bp1"
+block 4: 632-847 props="bp5"
+block 3: 412-624
+block 2: 195-244 props="bp3"
+
+iter
+first
+next
+next
+next
+next
+next
+next
+----
+block 0: 24-48 props="bp1"
+block 1: 141-253 props="bp2"
+block 2: 195-244 props="bp3"
+block 3: 412-624
+block 4: 632-847 props="bp5"
+block 5: 963-1289 props="bp6"
+.
+
+iter
+last
+prev
+prev
+prev
+prev
+prev
+prev
+----
+block 5: 963-1289 props="bp6"
+block 4: 632-847 props="bp5"
+block 3: 412-624
+block 2: 195-244 props="bp3"
+block 1: 141-253 props="bp2"
+block 0: 24-48 props="bp1"
+.
+
+build
+cat           3021   2052   bp1
+catastrophe  91251   1899
+catatonic   102422  20442
+cephalopod  122864   9104   bp4
+coat        293128  32104
+----
+# index block header
+# columnar block header
+000-001: x 01                   # version 1
+001-003: x 0400                 # 4 columns
+003-007: x 05000000             # 5 rows
+# column 0
+007-008: b 00000011             # bytes
+008-012: x 1b000000             # page start 27
+# column 1
+012-013: b 00000010             # uint
+013-017: x 47000000             # page start 71
+# column 2
+017-018: b 00000010             # uint
+018-022: x 5c000000             # page start 92
+# column 3
+022-023: b 00000011             # bytes
+023-027: x 68000000             # page start 104
+# data for column 0
+# rawbytes
+# offsets table
+027-028: x 01                   # encoding: 1b
+028-029: x 00                   # data[0] = 0 [34 overall]
+029-030: x 03                   # data[1] = 3 [37 overall]
+030-031: x 0e                   # data[2] = 14 [48 overall]
+031-032: x 17                   # data[3] = 23 [57 overall]
+032-033: x 21                   # data[4] = 33 [67 overall]
+033-034: x 25                   # data[5] = 37 [71 overall]
+# data
+034-037: x 636174               # data[0]: cat
+037-047: x 636174617374726f7068 # data[1]: catastrophe
+047-048: x 65                   # (continued...)
+048-057: x 63617461746f6e6963   # data[2]: catatonic
+057-067: x 63657068616c6f706f64 # data[3]: cephalopod
+067-071: x 636f6174             # data[4]: coat
+# data for column 1
+071-072: x 04                   # encoding: 4b
+072-076: x cd0b0000             # data[0] = 3021
+076-080: x 73640100             # data[1] = 91251
+080-084: x 16900100             # data[2] = 102422
+084-088: x f0df0100             # data[3] = 122864
+088-092: x 08790400             # data[4] = 293128
+# data for column 2
+092-093: x 02                   # encoding: 2b
+093-094: x 00                   # padding (aligning to 16-bit boundary)
+094-096: x 0408                 # data[0] = 2052
+096-098: x 6b07                 # data[1] = 1899
+098-100: x da4f                 # data[2] = 20442
+100-102: x 9023                 # data[3] = 9104
+102-104: x 687d                 # data[4] = 32104
+# data for column 3
+# rawbytes
+# offsets table
+104-105: x 01                   # encoding: 1b
+105-106: x 00                   # data[0] = 0 [111 overall]
+106-107: x 03                   # data[1] = 3 [114 overall]
+107-108: x 03                   # data[2] = 3 [114 overall]
+108-109: x 03                   # data[3] = 3 [114 overall]
+109-110: x 06                   # data[4] = 6 [117 overall]
+110-111: x 06                   # data[5] = 6 [117 overall]
+# data
+111-114: x 627031               # data[0]: bp1
+114-114: x                      # data[1]:
+114-114: x                      # data[2]:
+114-117: x 627034               # data[3]: bp4
+117-117: x                      # data[4]:
+
+iter
+first
+next
+next
+next
+next
+next
+----
+block 0: 3021-5073 props="bp1"
+block 1: 91251-93150
+block 2: 102422-122864
+block 3: 122864-131968 props="bp4"
+block 4: 293128-325232
+.
+
+iter
+last
+prev
+prev
+prev
+prev
+prev
+----
+block 4: 293128-325232
+block 3: 122864-131968 props="bp4"
+block 2: 102422-122864
+block 1: 91251-93150
+block 0: 3021-5073 props="bp1"
+.
diff --git a/sstable/colblk/unsafe_slice_test.go b/sstable/colblk/unsafe_slice_test.go
index c1ab661b..53aa5e5e 100644
--- a/sstable/colblk/unsafe_slice_test.go
+++ b/sstable/colblk/unsafe_slice_test.go
@@ -15,11 +15,56 @@ import (
 	"golang.org/x/exp/rand"
 )
 
+func TestUnsafeUints(t *testing.T) {
+	rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
+	for _, r := range interestingIntRanges {
+		t.Run(r.ExpectedEncoding.String(), func(t *testing.T) {
+			for _, rows := range []int{1, 10, 100, 10000} {
+				t.Run(fmt.Sprint(rows), func(t *testing.T) {
+					vals := make([]uint64, rows)
+					for i := range vals {
+						vals[i] = r.Rand(rng)
+					}
+					var ub UintBuilder
+					ub.Init()
+					for i := 0; i < rows; i++ {
+						ub.Set(i, vals[i])
+					}
+					sz := ub.Size(rows, 0)
+					buf := aligned.ByteSlice(int(sz) + 1 /* trailing padding byte */)
+					_ = ub.Finish(0, rows, 0, buf)
+
+					uints, _ := DecodeUnsafeUints(buf, 0, rows)
+					for i := range rows {
+						if uints.At(i) != vals[i] {
+							t.Fatalf("mismatch at row %d: got %d, expected %d", i, uints.At(i), vals[i])
+						}
+					}
+					if encoding := UintEncoding(buf[0]); encoding.Width() <= 4 && !encoding.IsDelta() {
+						offsets, _ := DecodeUnsafeOffsets(buf, 0, rows)
+						for i := range rows {
+							if uint64(offsets.At(i)) != vals[i] {
+								t.Fatalf("mismatch at row %d: got %d, expected %d", i, uints.At(i), vals[i])
+							}
+						}
+						for i := 0; i < rows-1; i++ {
+							a, b := offsets.At2(i)
+							if uint64(a) != vals[i] || uint64(b) != vals[i+1] {
+								t.Fatalf("mismatch at row %d: got %d,%d, expected %d,%d", i, a, b, vals[i], vals[i+1])
+							}
+						}
+					}
+				})
+			}
+		})
+	}
+}
+
 func BenchmarkUnsafeUints(b *testing.B) {
 	rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
 	intRanges := []intRange{
-		// const
-		{Min: 1, Max: 1, ExpectedEncoding: makeUintEncoding(0, true)},
+		// zero
+		{Min: 0, Max: 0, ExpectedEncoding: makeUintEncoding(0, false)},
 		// 1b
 		{Min: 10, Max: 200, ExpectedEncoding: makeUintEncoding(1, false)},
 		// 1b,delta
@@ -31,10 +76,11 @@ func BenchmarkUnsafeUints(b *testing.B) {
 		// 4b
 		{Min: 0, Max: math.MaxUint32, ExpectedEncoding: makeUintEncoding(4, false)},
 		// 4b,delta
-		{Min: 100_000, Max: math.MaxUint32 + 10, ExpectedEncoding: makeUintEncoding(4, true)},
+		{Min: 1_000_000_100, Max: math.MaxUint32 + 1_000_000_000, ExpectedEncoding: makeUintEncoding(4, true)},
 		// 8b
 		{Min: 0, Max: math.MaxUint64, ExpectedEncoding: makeUintEncoding(8, false)},
 	}
+
 	for _, r := range intRanges {
 		benchmarkUnsafeUints(b, rng, 1000, r)
 	}
diff --git a/sstable/data_test.go b/sstable/data_test.go
index dbaaf9d4..20f57353 100644
--- a/sstable/data_test.go
+++ b/sstable/data_test.go
@@ -152,7 +152,7 @@ func runBuildCmd(
 	for _, v := range rangeDels {
 		for _, k := range v.Keys {
 			ik := base.InternalKey{UserKey: v.Start, Trailer: k.Trailer}
-			if err := w.Raw().Add(ik, v.End); err != nil {
+			if err := w.Raw().AddWithForceObsolete(ik, v.End, false); err != nil {
 				return nil, nil, err
 			}
 		}
@@ -227,20 +227,8 @@ func runBuildRawCmd(
 		j := strings.Index(data, ":")
 		key := base.ParseInternalKey(data[:j])
 		value := []byte(data[j+1:])
-		switch key.Kind() {
-		case base.InternalKeyKindRangeKeyDelete,
-			base.InternalKeyKindRangeKeyUnset,
-			base.InternalKeyKindRangeKeySet:
-			// 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.Raw().addRangeKey(key, value); err != nil {
-				return nil, nil, err
-			}
-		default:
-			if err := w.Raw().Add(key, value); err != nil {
-				return nil, nil, err
-			}
+		if err := w.Raw().AddWithForceObsolete(key, value, false); err != nil {
+			return nil, nil, err
 		}
 	}
 	if err := w.Close(); err != nil {
@@ -428,8 +416,8 @@ func runIterCmd(
 			continue
 		case "internal-iter-state":
 			fmt.Fprintf(&b, "| %T:\n", origIter)
-			si, _ := origIter.(*singleLevelIterator[rowblk.Iter, *rowblk.Iter])
-			if twoLevelIter, ok := origIter.(*twoLevelIterator[rowblk.Iter, *rowblk.Iter]); ok {
+			si, _ := origIter.(*singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])
+			if twoLevelIter, ok := origIter.(*twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]); ok {
 				si = &twoLevelIter.secondLevel
 				if twoLevelIter.topLevelIndex.Valid() {
 					fmt.Fprintf(&b, "|  topLevelIndex.Key() = %q\n", twoLevelIter.topLevelIndex.Separator())
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index 1fe854d8..16400ccf 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -976,12 +976,12 @@ func TestCompactionIteratorSetupForCompaction(t *testing.T) {
 					NoTransforms, CategoryAndQoS{}, nil, MakeTrivialReaderProvider(r), &pool)
 				require.NoError(t, err)
 				switch i := citer.(type) {
-				case *singleLevelIterator[rowblk.Iter, *rowblk.Iter]:
+				case *singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]:
 					require.True(t, objstorageprovider.TestingCheckMaxReadahead(i.dataRH))
 					// Each key has one version, so no value block, regardless of
 					// sstable version.
 					require.Nil(t, i.vbRH)
-				case *twoLevelIterator[rowblk.Iter, *rowblk.Iter]:
+				case *twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]:
 					require.True(t, objstorageprovider.TestingCheckMaxReadahead(i.secondLevel.dataRH))
 					// Each key has one version, so no value block, regardless of
 					// sstable version.
@@ -1033,7 +1033,7 @@ func TestReadaheadSetupForV3TablesWithMultipleVersions(t *testing.T) {
 			NoTransforms, CategoryAndQoS{}, nil, MakeTrivialReaderProvider(r), &pool)
 		require.NoError(t, err)
 		defer citer.Close()
-		i := citer.(*singleLevelIterator[rowblk.Iter, *rowblk.Iter])
+		i := citer.(*singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])
 		require.True(t, objstorageprovider.TestingCheckMaxReadahead(i.dataRH))
 		require.True(t, objstorageprovider.TestingCheckMaxReadahead(i.vbRH))
 	}
@@ -1041,7 +1041,7 @@ func TestReadaheadSetupForV3TablesWithMultipleVersions(t *testing.T) {
 		iter, err := r.NewIter(NoTransforms, nil, nil)
 		require.NoError(t, err)
 		defer iter.Close()
-		i := iter.(*singleLevelIterator[rowblk.Iter, *rowblk.Iter])
+		i := iter.(*singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter])
 		require.False(t, objstorageprovider.TestingCheckMaxReadahead(i.dataRH))
 		require.False(t, objstorageprovider.TestingCheckMaxReadahead(i.vbRH))
 	}
diff --git a/sstable/testdata/reader_bpf/Pebblev3/iter b/sstable/testdata/reader_bpf/Pebblev3/iter
index 3a75aeba..e93281eb 100644
--- a/sstable/testdata/reader_bpf/Pebblev3/iter
+++ b/sstable/testdata/reader_bpf/Pebblev3/iter
@@ -94,7 +94,7 @@ next
 ----
 .
 .
-| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
+| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,*github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
 |  topLevelIndex.Key() = "x"
 |  topLevelIndex.BlockHandleWithProperties() = (Offset: 193, Length: 26, Props: 00020801)
 |  topLevelIndex.isDataInvalidated()=false
@@ -107,7 +107,7 @@ next
 |  (boundsCmp,positionedUsingLatestBounds) = (0,true)
 |  exhaustedBounds = 1
 .
-| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
+| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,*github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
 |  topLevelIndex.Key() = "wc"
 |  topLevelIndex.BlockHandleWithProperties() = (Offset: 161, Length: 27, Props: 00020201)
 |  topLevelIndex.isDataInvalidated()=false
@@ -119,7 +119,7 @@ next
 |  (boundsCmp,positionedUsingLatestBounds) = (0,true)
 |  exhaustedBounds = 1
 .
-| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
+| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,*github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
 |  topLevelIndex.Key() = "wc"
 |  topLevelIndex.BlockHandleWithProperties() = (Offset: 161, Length: 27, Props: 00020201)
 |  topLevelIndex.isDataInvalidated()=false
@@ -131,7 +131,7 @@ next
 |  (boundsCmp,positionedUsingLatestBounds) = (1,false)
 |  exhaustedBounds = 1
 <wz@8:8>
-| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
+| *sstable.twoLevelIterator[github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,*github.com/cockroachdb/pebble/sstable/rowblk.IndexIter,github.com/cockroachdb/pebble/sstable/rowblk.Iter,*github.com/cockroachdb/pebble/sstable/rowblk.Iter]:
 |  topLevelIndex.Key() = "x"
 |  topLevelIndex.BlockHandleWithProperties() = (Offset: 193, Length: 26, Props: 00020801)
 |  topLevelIndex.isDataInvalidated()=false
diff --git a/sstable/testdata/writer b/sstable/testdata/writer
index 6053e7de..24417dd1 100644
--- a/sstable/testdata/writer
+++ b/sstable/testdata/writer
@@ -203,18 +203,6 @@ rangekey: a-b:{(#1,RANGEKEYSET,@10,foo)}
 ----
 pebble: spans must be added in order: b > a
 
-build-raw
-a.RANGEKEYDEL.1:c
-b.RANGEKEYDEL.2:d
-----
-pebble: overlapping range keys must be fragmented: a#1,RANGEKEYDEL, b#2,RANGEKEYDEL
-
-build-raw
-a.RANGEKEYDEL.2:c
-a.RANGEKEYDEL.1:d
-----
-pebble: overlapping range keys must be fragmented: a#2,RANGEKEYDEL, a#1,RANGEKEYDEL
-
 build-raw
 rangekey: a-c:{(#1,RANGEKEYSET,@10,foo)}
 rangekey: c-d:{(#2,RANGEKEYSET,@10,foo)}
@@ -225,12 +213,6 @@ seqnums:  [1-2]
 # Range keys may have perfectly aligned spans (including sequence numbers),
 # though the key kinds must be ordered (descending).
 
-build-raw
-a.RANGEKEYDEL.1:b
-a.RANGEKEYDEL.1:b
-----
-pebble: range keys starts must be added in increasing order: a#1,RANGEKEYDEL, a#1,RANGEKEYDEL
-
 build-raw
 rangekey: a-b:{(#1,RANGEKEYSET,@10,foo) (#1,RANGEKEYUNSET,@t10) (#1,RANGEKEYDEL)}
 ----
diff --git a/sstable/testdata/writer_v3 b/sstable/testdata/writer_v3
index 883a629d..d7216b39 100644
--- a/sstable/testdata/writer_v3
+++ b/sstable/testdata/writer_v3
@@ -177,18 +177,6 @@ rangekey: a-b:{(#1,RANGEKEYSET,@10,foo)}
 ----
 pebble: spans must be added in order: b > a
 
-build-raw
-a.RANGEKEYDEL.1:c
-b.RANGEKEYDEL.2:d
-----
-pebble: overlapping range keys must be fragmented: a#1,RANGEKEYDEL, b#2,RANGEKEYDEL
-
-build-raw
-a.RANGEKEYDEL.2:c
-a.RANGEKEYDEL.1:d
-----
-pebble: overlapping range keys must be fragmented: a#2,RANGEKEYDEL, a#1,RANGEKEYDEL
-
 build-raw
 rangekey: a-c:{(#1,RANGEKEYSET,@10,foo)}
 rangekey: c-d:{(#2,RANGEKEYSET,@10,foo)}
@@ -199,12 +187,6 @@ seqnums:  [1-2]
 # Range keys may have perfectly aligned spans (including sequence numbers),
 # though the key kinds must be ordered (descending).
 
-build-raw
-a.RANGEKEYDEL.1:b
-a.RANGEKEYDEL.1:b
-----
-pebble: range keys starts must be added in increasing order: a#1,RANGEKEYDEL, a#1,RANGEKEYDEL
-
 build-raw
 rangekey: a-b:{(#1,RANGEKEYSET,@10,foo) (#1,RANGEKEYUNSET,@10) (#1,RANGEKEYDEL)}
 ----
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 22e3f861..0662cf61 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -331,15 +331,15 @@ func TestWriterWithValueBlocks(t *testing.T) {
 			if err != nil {
 				return err.Error()
 			}
-			forceIgnoreValueBlocks := func(i *singleLevelIterator[rowblk.Iter, *rowblk.Iter]) {
+			forceIgnoreValueBlocks := func(i *singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]) {
 				i.vbReader = nil
 				i.data.SetGetLazyValuer(nil)
 				i.data.SetHasValuePrefix(false)
 			}
 			switch i := origIter.(type) {
-			case *twoLevelIterator[rowblk.Iter, *rowblk.Iter]:
+			case *twoLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]:
 				forceIgnoreValueBlocks(&i.secondLevel)
-			case *singleLevelIterator[rowblk.Iter, *rowblk.Iter]:
+			case *singleLevelIterator[rowblk.IndexIter, *rowblk.IndexIter, rowblk.Iter, *rowblk.Iter]:
 				forceIgnoreValueBlocks(i)
 			}
 			iter := newIterAdapter(origIter)
@@ -528,7 +528,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.rw.coordination.writeQueue.err = errors.New("write queue write error")
+	w.rw.(*RawRowWriter).coordination.writeQueue.err = errors.New("write queue write error")
 	w.Set(ikey("a").UserKey, nil)
 	w.Set(ikey("b").UserKey, nil)
 	err = w.Close()
diff --git a/table_cache_test.go b/table_cache_test.go
index fb5ccc9e..a7f1c03c 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.Raw().Add(ik, xxx[:i]); err != nil {
+		if err := tw.Raw().AddWithForceObsolete(ik, xxx[:i], false); 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/tool/testdata/wal_dump b/tool/testdata/wal_dump
index 2fa4e30f..95725fca 100644
--- a/tool/testdata/wal_dump
+++ b/tool/testdata/wal_dump
@@ -6,15 +6,15 @@ wal dump
 ../testdata/db-stage-2/000002.log
 ----
 000002.log
-0(21) seq=10 count=1
+0(21) seq=10 count=1, len=21
     SET(test formatter: foo,test value formatter: one)
-32(21) seq=11 count=1
+32(21) seq=11 count=1, len=21
     SET(test formatter: bar,test value formatter: two)
-64(23) seq=12 count=1
+64(23) seq=12 count=1, len=23
     SET(test formatter: baz,test value formatter: three)
-98(22) seq=13 count=1
+98(22) seq=13 count=1, len=22
     SET(test formatter: foo,test value formatter: four)
-131(17) seq=14 count=1
+131(17) seq=14 count=1, len=17
     DEL(test formatter: bar)
 EOF
 
@@ -24,15 +24,15 @@ wal dump
 --value=size
 ----
 000002.log
-0(21) seq=10 count=1
+0(21) seq=10 count=1, len=21
     SET(foo,<3>)
-32(21) seq=11 count=1
+32(21) seq=11 count=1, len=21
     SET(bar,<3>)
-64(23) seq=12 count=1
+64(23) seq=12 count=1, len=23
     SET(baz,<5>)
-98(22) seq=13 count=1
+98(22) seq=13 count=1, len=22
     SET(foo,<4>)
-131(17) seq=14 count=1
+131(17) seq=14 count=1, len=17
     DEL(bar)
 EOF
 
@@ -42,11 +42,11 @@ wal dump
 ../testdata/db-stage-4/000005.log
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(foo,<4>)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(quux,<3>)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(baz)
 EOF
 
@@ -56,11 +56,11 @@ wal dump
 --value=%x
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(666f6f,66697665)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(71757578,736978)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(62617a)
 EOF
 
@@ -70,11 +70,11 @@ wal dump
 --value=pretty:test-comparer
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(foo,test value formatter: five)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(quux,test value formatter: six)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(baz)
 EOF
 
@@ -84,11 +84,11 @@ wal dump
 --value=%x
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(test formatter: foo,66697665)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(test formatter: quux,736978)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(test formatter: baz)
 EOF
 
@@ -98,11 +98,11 @@ wal dump
 --value=quoted
 ----
 000005.log
-0(22) seq=15 count=1
+0(22) seq=15 count=1, len=22
     SET(foo,five)
-33(22) seq=16 count=1
+33(22) seq=16 count=1, len=22
     SET(quux,six)
-66(17) seq=17 count=1
+66(17) seq=17 count=1, len=17
     DEL(baz)
 EOF
 
@@ -110,7 +110,7 @@ wal dump
 ./testdata/mixed/000004.log
 ----
 000004.log
-0(42) seq=39 count=4
+0(42) seq=39 count=4, len=42
     SET(test formatter: a@2,test value formatter: )
     RANGEKEYSET(test formatter: a-test formatter: z:{(#40,RANGEKEYSET,@3)})
     RANGEKEYUNSET(test formatter: a-test formatter: z:{(#41,RANGEKEYUNSET,@4)})
diff --git a/vfs/disk_health_test.go b/vfs/disk_health_test.go
index 6a6ec327..4a0b6db4 100644
--- a/vfs/disk_health_test.go
+++ b/vfs/disk_health_test.go
@@ -215,6 +215,8 @@ func (m mockFS) GetDiskUsage(path string) (DiskUsage, error) {
 	return m.getDiskUsage(path)
 }
 
+func (m mockFS) Unwrap() FS { return nil }
+
 var _ FS = &mockFS{}
 
 func TestDiskHealthChecking_WriteStatsCollector(t *testing.T) {
diff --git a/vfs/mem_fs_test.go b/vfs/mem_fs_test.go
index 26928c5c..61f6573a 100644
--- a/vfs/mem_fs_test.go
+++ b/vfs/mem_fs_test.go
@@ -7,7 +7,6 @@ package vfs
 import (
 	"fmt"
 	"io"
-	"os"
 	"sort"
 	"strconv"
 	"strings"
@@ -17,107 +16,82 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func runTestCases(t *testing.T, testCases []string, fs *MemFS) {
+func runMemFSDataDriven(t *testing.T, path string, fs *MemFS) {
 	var f File
-	for _, tc := range testCases {
-		s := strings.Split(tc, " ")[1:]
-
-		saveF := s[0] == "f" && s[1] == "="
-		if saveF {
-			s = s[2:]
-		}
-
-		fails := s[len(s)-1] == "fails"
-		if fails {
-			s = s[:len(s)-1]
-		}
-
-		var (
-			fi  os.FileInfo
-			g   File
-			err error
-		)
-		switch s[0] {
+	datadriven.RunTest(t, path, func(t *testing.T, td *datadriven.TestData) string {
+		var err error
+		switch td.Cmd {
 		case "create":
-			g, err = fs.Create(s[1], WriteCategoryUnspecified)
+			f, err = fs.Create(td.CmdArgs[0].String(), WriteCategoryUnspecified)
 		case "link":
-			err = fs.Link(s[1], s[2])
+			err = fs.Link(td.CmdArgs[0].String(), td.CmdArgs[1].String())
 		case "open":
-			g, err = fs.Open(s[1])
-		case "openDir":
-			g, err = fs.OpenDir(s[1])
-		case "openReadWrite":
-			g, err = fs.OpenReadWrite(s[1], WriteCategoryUnspecified)
+			f, err = fs.Open(td.CmdArgs[0].String())
+		case "open-dir":
+			f, err = fs.OpenDir(td.CmdArgs[0].String())
+		case "open-read-write":
+			f, err = fs.OpenReadWrite(td.CmdArgs[0].String(), WriteCategoryUnspecified)
 		case "mkdirall":
-			err = fs.MkdirAll(s[1], 0755)
+			err = fs.MkdirAll(td.CmdArgs[0].String(), 0755)
 		case "remove":
-			err = fs.Remove(s[1])
+			err = fs.Remove(td.CmdArgs[0].String())
 		case "rename":
-			err = fs.Rename(s[1], s[2])
-		case "reuseForWrite":
-			g, err = fs.ReuseForWrite(s[1], s[2], WriteCategoryUnspecified)
-		case "resetToSynced":
+			err = fs.Rename(td.CmdArgs[0].String(), td.CmdArgs[1].String())
+		case "reuse-for-write":
+			f, err = fs.ReuseForWrite(td.CmdArgs[0].String(), td.CmdArgs[1].String(), WriteCategoryUnspecified)
+		case "reset-to-synced":
 			fs.ResetToSyncedState()
-		case "ignoreSyncs":
+		case "ignore-syncs":
 			fs.SetIgnoreSyncs(true)
-		case "stopIgnoringSyncs":
+		case "stop-ignoring-syncs":
 			fs.SetIgnoreSyncs(false)
 		case "f.write":
-			_, err = f.Write([]byte(s[1]))
+			_, err = f.Write([]byte(strings.TrimSpace(td.Input)))
 		case "f.sync":
 			err = f.Sync()
 		case "f.read":
-			n, _ := strconv.Atoi(s[1])
+			n, _ := strconv.Atoi(td.CmdArgs[0].String())
 			buf := make([]byte, n)
 			_, err = io.ReadFull(f, buf)
 			if err != nil {
 				break
 			}
-			if got, want := string(buf), s[3]; got != want {
-				t.Fatalf("%q: got %q, want %q", tc, got, want)
-			}
+			return string(buf)
 		case "f.readat":
-			n, _ := strconv.Atoi(s[1])
-			off, _ := strconv.Atoi(s[2])
+			n, _ := strconv.Atoi(td.CmdArgs[0].String())
+			off, _ := strconv.Atoi(td.CmdArgs[0].String())
 			buf := make([]byte, n)
 			_, err = f.ReadAt(buf, int64(off))
 			if err != nil {
 				break
 			}
-			if got, want := string(buf), s[4]; got != want {
-				t.Fatalf("%q: got %q, want %q", tc, got, want)
-			}
+			return string(buf)
 		case "f.close":
 			f, err = nil, f.Close()
 		case "f.stat.name":
+			var fi FileInfo
 			fi, err = f.Stat()
 			if err != nil {
 				break
 			}
-			if got, want := fi.Name(), s[2]; got != want {
-				t.Fatalf("%q: got %q, want %q", tc, got, want)
+			return fi.Name()
+		case "list":
+			list, err := fs.List(td.CmdArgs[0].String())
+			if err != nil {
+				break
 			}
+			sort.Strings(list)
+			return strings.Join(list, "\n")
+		case "fs-string":
+			return fs.String()
 		default:
-			t.Fatalf("bad test case: %q", tc)
+			t.Fatalf("unknown command %q", td.Cmd)
 		}
-
-		if saveF {
-			f, g = g, nil
-		} else if g != nil {
-			g.Close()
-		}
-
-		if fails {
-			if err == nil {
-				t.Fatalf("%q: got nil error, want non-nil", tc)
-			}
-		} else {
-			if err != nil {
-				t.Fatalf("%q: %v", tc, err)
-			}
+		if err != nil {
+			return fmt.Sprintf("error: %v", err)
 		}
-	}
-
+		return ""
+	})
 	// Both "" and "/" are allowed to be used to refer to the root of the FS
 	// for the purposes of cloning.
 	checkClonedIsEquivalent(t, fs, "")
@@ -134,159 +108,16 @@ func checkClonedIsEquivalent(t *testing.T, fs *MemFS, path string) {
 	require.Equal(t, fs.String(), clone.String())
 }
 
-func TestBasics(t *testing.T) {
-	fs := NewMem()
-	testCases := []string{
-		// Create a top-level file.
-		"1a: create /foo",
-		// Create a child of that file. It should fail, since /foo is not a directory.
-		"2a: create /foo/x fails",
-		// Create a third-level file. It should fail, since /bar has not been created.
-		// Similarly, opening that file should fail.
-		"3a: create /bar/baz/y fails",
-		"3b: open /bar/baz/y fails",
-		// Make the /bar/baz directory; create a third-level file. Creation should now succeed.
-		"4a: mkdirall /bar/baz",
-		"4b: f = create /bar/baz/y",
-		"4c: f.stat.name == y",
-		// Write some data; read it back.
-		"5a: f.write abcde",
-		"5b: f.close",
-		"5c: f = open /bar/baz/y",
-		"5d: f.read 5 == abcde",
-		"5e: f.readat 2 1 == bc",
-		"5f: f.close",
-		// Link /bar/baz/y to /bar/baz/z. We should be able to read from both files
-		// and remove them independently.
-		"6a: link /bar/baz/y /bar/baz/z",
-		"6b: f = open /bar/baz/z",
-		"6c: f.read 5 == abcde",
-		"6d: f.close",
-		"6e: remove /bar/baz/z",
-		"6f: f = open /bar/baz/y",
-		"6g: f.read 5 == abcde",
-		"6h: f.close",
-		// Remove the file twice. The first should succeed, the second should fail.
-		"7a: remove /bar/baz/y",
-		"7b: remove /bar/baz/y fails",
-		"7c: open /bar/baz/y fails",
-		// Rename /foo to /goo. Trying to open /foo should succeed before the rename and
-		// fail afterwards, and vice versa for /goo.
-		"8a: open /foo",
-		"8b: open /goo fails",
-		"8c: rename /foo /goo",
-		"8d: open /foo fails",
-		"8e: open /goo",
-		// Create /bar/baz/z and rename /bar/baz to /bar/caz.
-		"9a: create /bar/baz/z",
-		"9b: open /bar/baz/z",
-		"9c: open /bar/caz/z fails",
-		"9d: rename /bar/baz /bar/caz",
-		"9e: open /bar/baz/z fails",
-		"9f: open /bar/caz/z",
-		// ReuseForWrite
-		"10a: reuseForWrite /bar/caz/z /bar/z",
-		"10b: open /bar/caz/z fails",
-		"10c: open /bar/z",
-		// Opening the root directory works.
-		"11a: f = open /",
-		"11b: f.stat.name == /",
-		// Read bytes after writing some bytes.
-		"12a: f = create foo",
-		"12b: f.write foobar",
-		"12c: f.close",
-		"12d: f = openReadWrite foo",
-		"12e: f.write abc",
-		"12f: f.read 3 == bar",
-		"12g: f.close",
-	}
-	runTestCases(t, testCases, fs)
+func TestMemFSBasics(t *testing.T) {
+	runMemFSDataDriven(t, "testdata/memfs_basics", NewMem())
 }
 
-func TestList(t *testing.T) {
-	fs := NewMem()
-
-	dirnames := []string{
-		"/bar",
-		"/foo/2",
-	}
-	for _, dirname := range dirnames {
-		err := fs.MkdirAll(dirname, 0755)
-		if err != nil {
-			t.Fatalf("MkdirAll %q: %v", dirname, err)
-		}
-	}
-
-	filenames := []string{
-		"/a",
-		"/bar/baz",
-		"/foo/0",
-		"/foo/1",
-		"/foo/2/a",
-		"/foo/2/b",
-		"/foo/3",
-		"/foot",
-	}
-	for _, filename := range filenames {
-		f, err := fs.Create(filename, WriteCategoryUnspecified)
-		if err != nil {
-			t.Fatalf("Create %q: %v", filename, err)
-		}
-		if err := f.Close(); err != nil {
-			t.Fatalf("Close %q: %v", filename, err)
-		}
-	}
-
-	require.NoError(t, fs.Link("/foo/3", "/bar/link-to-3"))
-	require.NoError(t, fs.Link("/foo/3", "/bar/another-link-to-3"))
-
-	{
-		got := fs.String()
-		const want = `          /
-       0    a
-            bar/
-       0      another-link-to-3
-       0      baz
-       0      link-to-3
-            foo/
-       0      0
-       0      1
-              2/
-       0        a
-       0        b
-       0      3
-       0    foot
-`
-		if got != want {
-			t.Fatalf("String:\n----got----\n%s----want----\n%s", got, want)
-		}
-	}
+func TestMemFSList(t *testing.T) {
+	runMemFSDataDriven(t, "testdata/memfs_list", NewMem())
+}
 
-	testCases := []string{
-		"/:a bar foo foot",
-		"/bar:another-link-to-3 baz link-to-3",
-		"/bar/:another-link-to-3 baz link-to-3",
-		"/baz:",
-		"/baz/:",
-		"/foo:0 1 2 3",
-		"/foo/:0 1 2 3",
-		"/foo/1:",
-		"/foo/1/:",
-		"/foo/2:a b",
-		"/foo/2/:a b",
-		"/foot:",
-		"/foot/:",
-	}
-	for _, tc := range testCases {
-		s := strings.Split(tc, ":")
-		list, _ := fs.List(s[0])
-		sort.Strings(list)
-		got := strings.Join(list, " ")
-		want := s[1]
-		if got != want {
-			t.Errorf("List %q: got %q, want %q", s[0], got, want)
-		}
-	}
+func TestMemFSStrict(t *testing.T) {
+	runMemFSDataDriven(t, "testdata/memfs_strict", NewStrictMem())
 }
 
 func TestMemFile(t *testing.T) {
@@ -301,116 +132,6 @@ func TestMemFile(t *testing.T) {
 	}
 }
 
-func TestStrictFS(t *testing.T) {
-	fs := NewStrictMem()
-	testCases := []string{
-		// Created file disappears if directory is not synced.
-		"1a: create /foo",
-		"1b: open /foo",
-		"1c: resetToSynced",
-		"1d: open /foo fails",
-
-		// Create directory and a file in it and write and read from it.
-		"2a: mkdirall /bar",
-		"2b: f = create /bar/y",
-		"2c: f.stat.name == y",
-		// Write some data; read it back.
-		"2d: f.write abcde",
-		"2e: f.close",
-		"2f: f = open /bar/y",
-		"2g: f.read 5 == abcde",
-		"2h: f.close",
-		"2i: open /bar",
-
-		// Resetting causes both the directory and file to disappear.
-		"3a: resetToSynced",
-		"3b: openDir /bar fails",
-		"3c: open /bar/y fails",
-
-		// Create the directory and file again. Link the file to another file in the same dir,
-		// and to a file in the root dir. Sync the root dir. After reset, the created dir and the
-		// file in the root dir are the only ones visible.
-		"4a: mkdirall /bar",
-		"4b: create /bar/y",
-		"4c: f = openDir /",
-		"4d: f.sync",
-		"4e: f.close",
-		"4f: link /bar/y /bar/z",
-		"4g: link /bar/y /z",
-		"4h: f = openDir /",
-		"4i: f.sync",
-		"4j: f.close",
-		"4k: resetToSynced",
-		"4l: openDir /bar",
-		"4m: open /bar/y fails",
-		"4n: open /bar/z fails",
-		"4o: open /z",
-
-		// Create the file in the directory again and this time sync /bar directory. The file is
-		// preserved after reset.
-		"5a: create /bar/y",
-		"5b: f = openDir /bar",
-		"5c: f.sync",
-		"5d: f.close",
-		"5e: resetToSynced",
-		"5f: openDir /bar",
-		"5g: open /bar/y",
-
-		// Unsynced data in the file is lost on reset.
-		"5a: f = create /bar/y",
-		"5b: f.write a",
-		"5c: f.sync",
-		"5d: f.write b",
-		"5e: f.close",
-		"5f: f = openDir /bar",
-		"5g: f.sync",
-		"5h: f.close",
-		"5i: resetToSynced",
-		"5j: f = open /bar/y",
-		"5k: f.read 1 = a",
-		"5l: f.read 1 fails",
-		"5m: f.close",
-
-		// reuseForWrite works correctly in strict mode in that unsynced data does not overwrite
-		// previous contents when a reset happens.
-		"6a: f = create /z",
-		"6b: f.write abcdefgh",
-		"6c: f.sync",
-		"6d: f.close",
-		"6e: f = reuseForWrite /z /y",
-		"6f: f.write x",
-		"6g: f.sync",
-		"6h: f.write y", // will be lost
-		"6i: f.close",
-		"6j: f = openDir /",
-		"6k: f.sync",
-		"6l: f.close",
-		"6m: resetToSynced",
-		"6n: f = open /y",
-		"6o: f.read 8 = xbcdefgh",
-		"6p: f.close",
-
-		// Ignore syncs.
-		"7a: f = create /z",
-		"7b: f.write a",
-		"7c: f.sync",
-		"7d: ignoreSyncs",
-		"7e: f.write b",
-		"7f: f.sync",
-		"7g: f.close",
-		"7h: stopIgnoringSyncs",
-		"7e: f = openDir /",
-		"7f: f.sync",
-		"7g: f.close",
-		"7h: resetToSynced",
-		"7i: f = open /z",
-		"7j: f.read 1 = a",
-		"7k: f.read 1 fails",
-		"7l: f.close",
-	}
-	runTestCases(t, testCases, fs)
-}
-
 func TestMemFSLock(t *testing.T) {
 	filesystems := map[string]FS{}
 	fileLocks := map[string]io.Closer{}
diff --git a/vfs/testdata/memfs_basics b/vfs/testdata/memfs_basics
new file mode 100644
index 00000000..ac2ac091
--- /dev/null
+++ b/vfs/testdata/memfs_basics
@@ -0,0 +1,174 @@
+# Create a top-level file.
+create /foo
+----
+
+# Create a child of that file. It should fail, since /foo is not a directory.
+create /foo/x
+----
+error: open foo/x: not a directory
+
+# Create a third-level file. It should fail, since /bar has not been created.
+# Similarly, opening that file should fail.
+create /bar/baz/y
+----
+error: open bar/baz/y: file does not exist
+
+open /bar/baz/y
+----
+error: open bar/baz/y: file does not exist
+
+# Make the /bar/baz directory; create a third-level file. Creation should now succeed.
+mkdirall /bar/baz
+----
+
+create /bar/baz/y
+----
+
+f.stat.name
+----
+y
+
+# Write some data; read it back.
+f.write
+abcde
+----
+
+f.close
+----
+
+open /bar/baz/y
+----
+
+f.read 5
+----
+abcde
+
+f.readat 2 1
+----
+cd
+
+f.close
+----
+
+# Link /bar/baz/y to /bar/baz/z. We should be able to read from both files
+# and remove them independently.
+link /bar/baz/y /bar/baz/z
+----
+
+open /bar/baz/z
+----
+
+f.read 5
+----
+abcde
+
+f.close
+----
+
+remove /bar/baz/z
+----
+
+open /bar/baz/y
+----
+
+f.read 5
+----
+abcde
+
+f.close
+----
+
+# Remove the file twice. The first should succeed, the second should fail.
+remove /bar/baz/y
+----
+
+remove /bar/baz/y
+----
+error: file does not exist
+
+open /bar/baz/y
+----
+error: open /bar/baz/y: file does not exist
+
+# Rename /foo to /goo. Trying to open /foo should succeed before the rename and
+# fail afterwards, and vice versa for /goo.
+open /foo
+----
+
+open /goo
+----
+error: open /goo: file does not exist
+
+rename /foo /goo
+----
+
+open /foo
+----
+error: open /foo: file does not exist
+
+open /goo
+----
+
+# Create /bar/baz/z and rename /bar/baz to /bar/caz.
+create /bar/baz/z
+----
+
+open /bar/baz/z
+----
+
+open /bar/caz/z
+----
+error: open caz/z: file does not exist
+
+rename /bar/baz /bar/caz
+----
+
+open /bar/baz/z
+----
+error: open baz/z: file does not exist
+
+open /bar/caz/z
+----
+
+reuse-for-write /bar/caz/z /bar/z
+----
+
+open /bar/caz/z
+----
+error: open /bar/caz/z: file does not exist
+
+open /bar/z
+----
+
+# Opening the root directory works.
+open /
+----
+
+f.stat.name
+----
+/
+
+# Read bytes after writing some bytes.
+create foo
+----
+
+f.write
+foobar
+----
+
+f.close
+----
+
+open-read-write foo
+----
+
+f.write
+abc
+----
+
+f.read 3
+----
+bar
+
+f.close
+----
diff --git a/vfs/testdata/memfs_list b/vfs/testdata/memfs_list
new file mode 100644
index 00000000..81a55863
--- /dev/null
+++ b/vfs/testdata/memfs_list
@@ -0,0 +1,113 @@
+mkdirall /bar
+----
+
+mkdirall /foo/2
+----
+
+create /a
+----
+
+create /bar/baz
+----
+
+create /foo/0
+----
+
+create /foo/1
+----
+
+create /foo/2/a
+----
+
+create /foo/2/b
+----
+
+create /foo/3
+----
+
+create /foot
+----
+
+link /foo/3 /bar/link-to-3
+----
+
+link /foo/3 /bar/another-link-to-3
+----
+
+fs-string
+----
+          /
+       0    a
+            bar/
+       0      another-link-to-3
+       0      baz
+       0      link-to-3
+            foo/
+       0      0
+       0      1
+              2/
+       0        a
+       0        b
+       0      3
+       0    foot
+
+list /
+----
+a
+bar
+foo
+foot
+
+list /bar
+----
+another-link-to-3
+baz
+link-to-3
+
+list /bar/
+----
+another-link-to-3
+baz
+link-to-3
+
+list /baz
+----
+
+list /baz/
+----
+
+list /foo
+----
+0
+1
+2
+3
+
+list /foo/
+----
+0
+1
+2
+3
+
+list /foo/1
+----
+
+list /foo/1/
+----
+
+list /foo/2
+----
+a
+b
+
+list /foo/2/
+----
+a
+b
+
+list /foot
+----
+
+list /foot/
+----
diff --git a/vfs/testdata/memfs_strict b/vfs/testdata/memfs_strict
new file mode 100644
index 00000000..03f9e885
--- /dev/null
+++ b/vfs/testdata/memfs_strict
@@ -0,0 +1,282 @@
+# Created file disappears if directory is not synced.
+create /foo
+----
+
+open /foo
+----
+
+reset-to-synced
+----
+
+open /foo
+----
+error: open /foo: file does not exist
+
+# Create directory and a file in it and write and read from it.
+mkdirall /bar
+----
+
+create /bar/y
+----
+
+f.stat.name
+----
+y
+
+# Write some data; read it back.
+f.write
+abcde
+----
+
+f.close
+----
+
+open /bar/y
+----
+
+f.read 5
+----
+abcde
+
+f.close
+----
+
+open /bar
+----
+
+# Resetting causes both the directory and file to disappear.
+reset-to-synced
+----
+
+open-dir /bar
+----
+error: open /bar: file does not exist
+
+open /bar/y
+----
+error: open bar/y: file does not exist
+
+# Create the directory and file again. Link the file to another file in the same
+# dir, and to a file in the root dir. Sync the root dir. After reset, the
+# created dir and the file in the root dir are the only ones visible.
+mkdirall /bar
+----
+
+create /bar/y
+----
+
+open-dir /
+----
+
+f.sync
+----
+
+f.close
+----
+
+link /bar/y /bar/z
+----
+
+link /bar/y /z
+----
+
+open-dir /
+----
+
+f.sync
+----
+
+f.close
+----
+
+reset-to-synced
+----
+
+open-dir /bar
+----
+
+open /bar/y
+----
+error: open /bar/y: file does not exist
+
+open /bar/z
+----
+error: open /bar/z: file does not exist
+
+open /z
+----
+
+# Create the file in the directory again and this time sync /bar directory. The
+# file is preserved after reset.
+create /bar/y
+----
+
+open-dir /bar
+----
+
+f.sync
+----
+
+f.close
+----
+
+reset-to-synced
+----
+
+open-dir /bar
+----
+
+open /bar/y
+----
+
+# Unsynced data in the file is lost on reset.
+create /bar/y
+----
+
+f.write
+a
+----
+
+f.sync
+----
+
+f.write
+b
+----
+
+f.close
+----
+
+open-dir /bar
+----
+
+f.sync
+----
+
+f.close
+----
+
+reset-to-synced
+----
+
+open /bar/y
+----
+
+f.read 1
+----
+a
+
+f.read 1
+----
+error: EOF
+
+f.close
+----
+
+# reuse-for-write works correctly in strict mode in that unsynced data does not overwrite
+# previous contents when a reset happens.
+create /z
+----
+
+f.write
+abcdefgh
+----
+
+f.sync
+----
+
+f.close
+----
+
+reuse-for-write /z /y
+----
+
+f.write
+x
+----
+
+f.sync
+----
+
+# Will be lost.
+f.write
+y
+----
+
+f.close
+----
+
+open-dir /
+----
+
+f.sync
+----
+
+f.close
+----
+
+reset-to-synced
+----
+
+open /y
+----
+
+f.read 8
+----
+xbcdefgh
+
+f.close
+----
+
+# Ignore syncs.
+create /z
+----
+
+f.write
+a
+----
+
+f.sync
+----
+
+ignore-syncs
+----
+
+f.write
+b
+----
+
+f.sync
+----
+
+f.close
+----
+
+stop-ignoring-syncs
+----
+
+open-dir /
+----
+
+f.sync
+----
+
+f.close
+----
+
+reset-to-synced
+----
+
+open /z
+----
+
+f.read 1
+----
+a
+
+f.read 1
+----
+error: EOF
+
+f.close
+----
diff --git a/vfs/vfstest/open_files.go b/vfs/vfstest/open_files.go
index e2fb7af6..4d23fe05 100644
--- a/vfs/vfstest/open_files.go
+++ b/vfs/vfstest/open_files.go
@@ -119,6 +119,8 @@ func (fs *openFilesFS) GetDiskUsage(path string) (vfs.DiskUsage, error) {
 	return fs.inner.GetDiskUsage(path)
 }
 
+func (fs *openFilesFS) Unwrap() vfs.FS { return fs.inner }
+
 func (fs *openFilesFS) wrapOpenFile(f vfs.File, err error) (vfs.File, error) {
 	if f == nil || err != nil {
 		return f, err