Repository: cockroachdb/pebble
Pull request #3983 (2024-09-29 13:13:12)

db: fill missing parameters
The format log is missing a parameter

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/3983)
<!-- Reviewable:end -->

caicancai: @RaduBerinde If you have time, could you help me take a look? Thanks

Patch:
diff --git a/compaction.go b/compaction.go
index 814f7949..ead7ee49 100644
--- a/compaction.go
+++ b/compaction.go
@@ -2667,9 +2667,7 @@ func (c *compaction) makeVersionEdit(result compact.Result) (*versionEdit, error
 func (d *DB) newCompactionOutput(
 	jobID JobID, c *compaction, writerOpts sstable.WriterOptions,
 ) (objstorage.ObjectMetadata, sstable.RawWriter, CPUWorkHandle, error) {
-	d.mu.Lock()
 	diskFileNum := d.mu.versions.getNextDiskFileNum()
-	d.mu.Unlock()
 
 	var writeCategory vfs.DiskWriteCategory
 	if d.opts.EnableSQLRowSpillMetrics {
diff --git a/ingest.go b/ingest.go
index 0fd20886..acf234fb 100644
--- a/ingest.go
+++ b/ingest.go
@@ -1355,14 +1355,12 @@ func (d *DB) ingest(
 	// the file number ordering to be out of alignment with sequence number
 	// ordering. The sorting of L0 tables by sequence number avoids relying on
 	// that (busted) invariant.
-	d.mu.Lock()
 	pendingOutputs := make([]base.FileNum, len(paths)+len(shared)+len(external))
 	for i := 0; i < len(paths)+len(shared)+len(external); i++ {
 		pendingOutputs[i] = d.mu.versions.getNextFileNum()
 	}
 
-	jobID := d.newJobIDLocked()
-	d.mu.Unlock()
+	jobID := d.newJobID()
 
 	// Load the metadata for all the files being ingested. This step detects
 	// and elides empty sstables.
diff --git a/internal/base/internal.go b/internal/base/internal.go
index 0fc719f4..0c123eb4 100644
--- a/internal/base/internal.go
+++ b/internal/base/internal.go
@@ -230,6 +230,11 @@ func MakeTrailer(seqNum SeqNum, kind InternalKeyKind) InternalKeyTrailer {
 	return (InternalKeyTrailer(seqNum) << 8) | InternalKeyTrailer(kind)
 }
 
+// String imlements the fmt.Stringer interface.
+func (t InternalKeyTrailer) String() string {
+	return fmt.Sprintf("%s,%s", SeqNum(t>>8), InternalKeyKind(t&0xff))
+}
+
 // SeqNum returns the sequence number component of the trailer.
 func (t InternalKeyTrailer) SeqNum() SeqNum {
 	return SeqNum(t >> 8)
diff --git a/open.go b/open.go
index f274b4d5..2610e345 100644
--- a/open.go
+++ b/open.go
@@ -177,7 +177,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		if opts.Experimental.CreateOnShared != remote.CreateOnSharedNone && formatVersion < FormatMinForSharedObjects {
 			return nil, errors.Newf(
 				"pebble: database %q configured with shared objects but written in too old format major version %d",
-				formatVersion)
+				dirname, formatVersion)
 		}
 	}
 
diff --git a/options.go b/options.go
index 21a7b925..db4c9b9a 100644
--- a/options.go
+++ b/options.go
@@ -1984,7 +1984,6 @@ func (o *Options) Validate() error {
 	if o.Experimental.CreateOnShared != remote.CreateOnSharedNone && o.FormatMajorVersion < FormatMinForSharedObjects {
 		fmt.Fprintf(&buf, "FormatMajorVersion (%d) when CreateOnShared is set must be at least %d\n",
 			o.FormatMajorVersion, FormatMinForSharedObjects)
-
 	}
 	if o.TableCache != nil && o.Cache != o.TableCache.cache {
 		fmt.Fprintf(&buf, "underlying cache in the TableCache and the Cache dont match\n")
diff --git a/sstable/block/block.go b/sstable/block/block.go
index e8a3ab16..82085b4f 100644
--- a/sstable/block/block.go
+++ b/sstable/block/block.go
@@ -162,8 +162,10 @@ type DataBlockIterator interface {
 	// contained within the data block. It's equivalent to performing
 	//   Compare(firstUserKey, k)
 	CompareFirstUserKey(k []byte) int
-	// Invalidate invalidates the block iterator, removing references to the block
-	// it was initialized with.
+	// Invalidate invalidates the block iterator, removing references to the
+	// block it was initialized with. The iterator may continue to be used after
+	// a call to Invalidate, but all positioning methods should return false.
+	// Valid() must also return false.
 	Invalidate()
 	// IsDataInvalidated returns true when the iterator has been invalidated
 	// using an Invalidate call.
@@ -191,7 +193,9 @@ type IndexBlockIterator interface {
 	// currently positioned over a valid block entry.
 	IsDataInvalidated() bool
 	// Invalidate invalidates the block iterator, removing references to the
-	// block it was initialized with.
+	// block it was initialized with. The iterator may continue to be used after
+	// a call to Invalidate, but all positioning methods should return false.
+	// Valid() must also return false.
 	Invalidate()
 	// Handle returns the underlying block buffer handle, if the iterator was
 	// initialized with one.
@@ -217,10 +221,12 @@ type IndexBlockIterator interface {
 	// the index block is empty.
 	Last() bool
 	// Next steps the index iterator to the next block entry. It returns false
-	// if the index block is exhausted.
+	// if the index block is exhausted in the forward direction. A call to Next
+	// while already exhausted in the forward direction is a no-op.
 	Next() bool
 	// Prev steps the index iterator to the previous block entry. It returns
-	// false if the index block is exhausted.
+	// false if the index block is exhausted in the reverse direction. A call to
+	// Prev while already exhausted in the reverse direction is a no-op.
 	Prev() bool
 	// Close closes the iterator, releasing any resources it holds.
 	Close() error
@@ -233,8 +239,11 @@ type IndexBlockIterator interface {
 // preferable.
 type IterTransforms struct {
 	// SyntheticSeqNum, if set, overrides the sequence number in all keys. It is
-	// set if the sstable was ingested or it is foregin.
-	SyntheticSeqNum    SyntheticSeqNum
+	// set if the sstable was ingested or it is foreign.
+	SyntheticSeqNum SyntheticSeqNum
+	// HideObsoletePoints, if true, skips over obsolete points during iteration.
+	// This is the norm when the sstable is foreign or the largest sequence number
+	// of the sstable is below the one we are reading.
 	HideObsoletePoints bool
 	SyntheticPrefix    SyntheticPrefix
 	SyntheticSuffix    SyntheticSuffix
diff --git a/sstable/colblk/bitmap.go b/sstable/colblk/bitmap.go
index fb4b765a..3c1b8c64 100644
--- a/sstable/colblk/bitmap.go
+++ b/sstable/colblk/bitmap.go
@@ -77,10 +77,10 @@ func (b Bitmap) At(i int) bool {
 }
 
 // SeekSetBitGE 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.
+// The i parameter must be ≥ 0. Returns the number of bits
+// represented by the bitmap if no next bit is set (or if i >= bitCount).
 func (b Bitmap) SeekSetBitGE(i int) int {
-	if b.data.ptr == nil {
+	if b.data.ptr == nil || i >= b.bitCount {
 		// Zero bitmap case.
 		return b.bitCount
 	}
@@ -243,6 +243,10 @@ func (b Bitmap) String() string {
 // BitmapBuilder constructs a Bitmap. Bits default to false.
 type BitmapBuilder struct {
 	words []uint64
+	// minNonZeroRowCount is the row count at which the bitmap should begin to
+	// use the defaultBitmapEncoding (as opposed to the zeroBitmapEncoding).
+	// It's updated on the first call to Set and defaults to zero.
+	minNonZeroRowCount int
 }
 
 type bitmapEncoding uint8
@@ -268,6 +272,11 @@ func bitmapRequiredSize(total int) int {
 
 // Set sets the bit at position i to true.
 func (b *BitmapBuilder) Set(i int) {
+	// Update minNonZeroRowCount if necessary. This is used to determine whether
+	// the bitmap should be encoded using the all-zeros encoding.
+	if b.isZero(i + 1) {
+		b.minNonZeroRowCount = i + 1
+	}
 	w := i >> 6 // divide by 64
 	for len(b.words) <= w {
 		b.words = append(b.words, 0)
@@ -276,14 +285,17 @@ func (b *BitmapBuilder) Set(i int) {
 }
 
 // isZero returns true if no bits are set and Invert was not called.
-func (b *BitmapBuilder) isZero() bool {
-	return len(b.words) == 0
+//
+//gcassert:inline
+func (b *BitmapBuilder) isZero(rows int) bool {
+	return b.minNonZeroRowCount == 0 || rows < b.minNonZeroRowCount
 }
 
 // Reset resets the bitmap to the empty state.
 func (b *BitmapBuilder) Reset() {
 	clear(b.words)
 	b.words = b.words[:0]
+	b.minNonZeroRowCount = 0
 }
 
 // NumColumns implements the ColumnWriter interface.
@@ -296,7 +308,7 @@ func (b *BitmapBuilder) DataType(int) DataType { return DataTypeBool }
 func (b *BitmapBuilder) Size(rows int, offset uint32) uint32 {
 	// First byte will be the encoding type.
 	offset++
-	if b.isZero() {
+	if b.isZero(rows) {
 		return offset
 	}
 	offset = align(offset, align64)
@@ -320,9 +332,16 @@ func (b *BitmapBuilder) InvertedSize(rows int, offset uint32) uint32 {
 // 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) {
+	// Inverted bitmaps never use the all-zero encoding, so we set
+	// rowCountIncludingFirstSetBit to 1 so that as long as the bitmap is
+	// finished encoding any rows at all, it uses the default encoding.
+	b.minNonZeroRowCount = 1
 	// If the tail of b is sparse, fill in zeroes before inverting.
 	nBitmapWords := (nRows + 63) >> 6
-	b.words = slices.Grow(b.words, nBitmapWords-len(b.words))[:nBitmapWords]
+	if len(b.words) < nBitmapWords {
+		b.words = slices.Grow(b.words, nBitmapWords-len(b.words))
+	}
+	b.words = b.words[:nBitmapWords]
 	for i := range b.words {
 		b.words[i] = ^b.words[i]
 	}
@@ -331,7 +350,7 @@ 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() {
+	if b.isZero(nRows) {
 		buf[offset] = byte(zeroBitmapEncoding)
 		return offset + 1
 	}
diff --git a/sstable/colblk/data_block.go b/sstable/colblk/data_block.go
index c95b3753..f1cb75a4 100644
--- a/sstable/colblk/data_block.go
+++ b/sstable/colblk/data_block.go
@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/binfmt"
+	"github.com/cockroachdb/pebble/internal/bytealloc"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -182,9 +183,6 @@ func (w *defaultKeyWriter) ComparePrev(key []byte) KeyComparison {
 	lp := w.prefixes.UnsafeGet(w.prefixes.nKeys - 1)
 	cmpv.CommonPrefixLen = int32(crbytes.CommonPrefix(lp, key[:cmpv.PrefixLen]))
 
-	if invariants.Enabled && bytes.Compare(lp, key[:cmpv.PrefixLen]) > 0 {
-		panic(errors.AssertionFailedf("keys are not in order: %q > %q", lp, key[:cmpv.PrefixLen]))
-	}
 	// Keys are written in order and prefixes must be sorted lexicograpgically,
 	// so CommonPrefixLen == PrefixLen implies that the keys share the same
 	// logical prefix. (If the previous key had a prefix longer than
@@ -580,6 +578,105 @@ func (w *DataBlockWriter) Finish(rows, size int) (finished []byte, lastKey base.
 	return finished, lastKey
 }
 
+// DataBlockRewriter rewrites data blocks. See RewriteSuffixes.
+type DataBlockRewriter struct {
+	KeySchema KeySchema
+
+	writer    DataBlockWriter
+	reader    DataBlockReader
+	iter      DataBlockIter
+	keySeeker KeySeeker
+	keyBuf    []byte
+	// keyAlloc grown throughout the lifetime of the rewriter.
+	keyAlloc        bytealloc.A
+	prefixBytesIter PrefixBytesIter
+	initialized     bool
+}
+
+// RewriteSuffixes rewrites the input block. It expects the input block to only
+// contain keys with the suffix `from`. It rewrites the block to contain the
+// same keys with the suffix `to`.
+//
+// RewriteSuffixes returns the start and end keys of the rewritten block, and
+// the finished rewritten block. The returned start and end keys have indefinite
+// lifetimes. The returned rewritten block is owned by the DataBlockRewriter. If
+// it must be retained beyond the next call to RewriteSuffixes, the caller
+// should make a copy.
+func (rw *DataBlockRewriter) RewriteSuffixes(
+	input []byte, from []byte, to []byte,
+) (start, end base.InternalKey, rewritten []byte, err error) {
+	if !rw.initialized {
+		rw.keySeeker = rw.KeySchema.NewKeySeeker()
+		rw.writer.Init(rw.KeySchema)
+		rw.initialized = true
+	}
+
+	// TODO(jackson): RewriteSuffixes performs a naïve rewrite of the block,
+	// iterating over the input block while building a new block, KV-by-KV.
+	// Since key columns are stored separately from other data, we could copy
+	// columns that are unchanged (at least all the non-key columns, and in
+	// practice a PrefixBytes column) wholesale without retrieving rows
+	// one-by-one. In practice, there a few obstacles to making this work:
+	//
+	// - Only the beginning of a data block is assumed to be aligned. Columns
+	//   then add padding as necessary to align data that needs to be aligned.
+	//   If we copy a column, we have no guarantee that the alignment of the
+	//   column start in the old block matches the alignment in the new block.
+	//   We'd have to add padding to between columns to match the original
+	//   alignment. It's a bit subtle.
+	// - We still need to read all the key columns in order to synthesize
+	//   [start] and [end].
+	//
+	// The columnar format is designed to support fast IterTransforms at read
+	// time, including IterTransforms.SyntheticSuffix. Our effort might be
+	// better spent dropping support for the physical rewriting of data blocks
+	// we're performing here and instead use a read-time IterTransform.
+
+	rw.reader.Init(rw.KeySchema, input)
+	rw.keySeeker.Init(&rw.reader)
+	rw.writer.Reset()
+	if err = rw.iter.Init(&rw.reader, rw.keySeeker, nil, block.IterTransforms{}); err != nil {
+		return base.InternalKey{}, base.InternalKey{}, nil, err
+	}
+
+	// Allocate a keyIter buffer that's large enough to hold the largest user
+	// key in the block with 1 byte to spare (so that pointer arithmetic is
+	// never pointing beyond the allocation, which would violate Go rules).
+	if cap(rw.prefixBytesIter.buf) < int(rw.reader.maximumKeyLength)+1 {
+		rw.prefixBytesIter.buf = make([]byte, rw.reader.maximumKeyLength+1)
+	}
+	if newMax := int(rw.reader.maximumKeyLength) - len(from) + len(to) + 1; cap(rw.keyBuf) < newMax {
+		rw.keyBuf = make([]byte, newMax)
+	}
+
+	// Rewrite each key-value pair one-by-one.
+	for i, kv := 0, rw.iter.First(); kv != nil; i, kv = i+1, rw.iter.Next() {
+		value := kv.V.ValueOrHandle
+		valuePrefix := block.InPlaceValuePrefix(false /* setHasSamePrefix (unused) */)
+		isValueExternal := rw.reader.isValueExternal.At(i)
+		if isValueExternal {
+			valuePrefix = block.ValuePrefix(kv.V.ValueOrHandle[0])
+			value = kv.V.ValueOrHandle[1:]
+		}
+		kcmp := rw.writer.KeyWriter.ComparePrev(kv.K.UserKey)
+		if !bytes.Equal(kv.K.UserKey[kcmp.PrefixLen:], from) {
+			return base.InternalKey{}, base.InternalKey{}, nil,
+				errors.Newf("key %s has suffix 0x%x; require 0x%x", kv.K, kv.K.UserKey[kcmp.PrefixLen:], from)
+		}
+		rw.keyBuf = append(rw.keyBuf[:0], kv.K.UserKey[:kcmp.PrefixLen]...)
+		rw.keyBuf = append(rw.keyBuf, to...)
+		if i == 0 {
+			start.UserKey, rw.keyAlloc = rw.keyAlloc.Copy(rw.keyBuf)
+			start.Trailer = kv.K.Trailer
+		}
+		k := base.InternalKey{UserKey: rw.keyBuf, Trailer: kv.K.Trailer}
+		rw.writer.Add(k, value, valuePrefix, kcmp, rw.reader.isObsolete.At(i))
+	}
+	rewritten, end = rw.writer.Finish(int(rw.reader.r.header.Rows), rw.writer.Size())
+	end.UserKey, rw.keyAlloc = rw.keyAlloc.Copy(end.UserKey)
+	return start, end, rewritten, nil
+}
+
 // DataBlockReaderSize is the size of a DataBlockReader struct. If allocating
 // memory for a data block, the caller may want to additionally allocate memory
 // for the corresponding DataBlockReader.
@@ -610,7 +707,10 @@ type DataBlockReader struct {
 	isObsolete Bitmap
 	// maximumKeyLength is the maximum length of a user key in the block.
 	// Iterators may use it to allocate a sufficiently large buffer up front,
-	// and elide size checks during iteration.
+	// and elide size checks during iteration. Note that iterators should add +1
+	// to the key length to ensure pointer arithmetric that computes a pointer
+	// to the tail of the key does not point to memory beyond the allocation
+	// (prohibited by Go pointer rules).
 	maximumKeyLength uint32
 }
 
@@ -690,8 +790,21 @@ func (c *MultiDataBlockIter) Handle() block.BufferHandle {
 	return c.h
 }
 
+// Valid returns true if the iterator is currently positioned at a valid KV.
+func (i *MultiDataBlockIter) Valid() bool {
+	return i.row >= 0 && i.row <= i.maxRow
+}
+
+// KV returns the key-value pair at the current iterator position. The
+// iterator must be positioned over a valid KV.
+func (i *MultiDataBlockIter) KV() *base.InternalKV {
+	return &i.kv
+}
+
 // Invalidate invalidates the block iterator, removing references to the block
-// it was initialized with.
+// it was initialized with. The iterator may continue to be used after
+// a call to Invalidate, but all positioning methods should return false.
+// Valid() must also return false.
 func (c *MultiDataBlockIter) Invalidate() {
 	c.DataBlockIter = DataBlockIter{}
 }
@@ -740,6 +853,11 @@ type DataBlockIter struct {
 	row     int
 	kv      base.InternalKV
 	kvRow   int // the row currently held in kv
+
+	// nextObsoletePoint is the row index of the first obsolete point after i.row.
+	// It is used to optimize skipping of obsolete points during forward
+	// iteration.
+	nextObsoletePoint int
 }
 
 // Init initializes the data block iterator, configuring it to read from the
@@ -750,9 +868,10 @@ func (i *DataBlockIter) Init(
 	getLazyValuer block.GetLazyValueForPrefixAndValueHandler,
 	transforms block.IterTransforms,
 ) error {
+	numRows := int(r.r.header.Rows)
 	*i = DataBlockIter{
 		r:             r,
-		maxRow:        int(r.r.header.Rows) - 1,
+		maxRow:        numRows - 1,
 		keySeeker:     keyIterator,
 		getLazyValuer: getLazyValuer,
 		transforms:    transforms,
@@ -761,6 +880,10 @@ func (i *DataBlockIter) Init(
 		kv:            base.InternalKV{},
 		keyIter:       PrefixBytesIter{},
 	}
+	if i.transforms.HideObsoletePoints && r.isObsolete.SeekSetBitGE(0) == numRows {
+		// There are no obsolete points in the block; don't bother checking.
+		i.transforms.HideObsoletePoints = false
+	}
 	// Allocate a keyIter buffer that's large enough to hold the largest user
 	// key in the block with 1 byte to spare (so that pointer arithmetic is
 	// never pointing beyond the allocation, which would violate Go rules).
@@ -783,11 +906,24 @@ func (i *DataBlockIter) CompareFirstUserKey(k []byte) int {
 
 // SeekGE implements the base.InternalIterator interface.
 func (i *DataBlockIter) SeekGE(key []byte, flags base.SeekGEFlags) *base.InternalKV {
+	if i.r == nil {
+		return nil
+	}
 	searchDir := int8(0)
 	if flags.TrySeekUsingNext() {
 		searchDir = +1
 	}
-	return i.decodeRow(i.keySeeker.SeekGE(key, i.row, searchDir))
+	i.row = i.keySeeker.SeekGE(key, i.row, searchDir)
+	if i.transforms.HideObsoletePoints {
+		i.nextObsoletePoint = i.r.isObsolete.SeekSetBitGE(i.row)
+		if i.atObsoletePointForward() {
+			i.skipObsoletePointsForward()
+			if i.row > i.maxRow {
+				return nil
+			}
+		}
+	}
+	return i.decodeRow()
 }
 
 // SeekPrefixGE implements the base.InternalIterator interface.
@@ -806,28 +942,75 @@ func (i *DataBlockIter) SeekPrefixGE(prefix, key []byte, flags base.SeekGEFlags)
 
 // SeekLT implements the base.InternalIterator interface.
 func (i *DataBlockIter) SeekLT(key []byte, _ base.SeekLTFlags) *base.InternalKV {
-	row := i.keySeeker.SeekGE(key, i.row, 0 /* searchDir */)
-	return i.decodeRow(row - 1)
+	if i.r == nil {
+		return nil
+	}
+	i.row = i.keySeeker.SeekGE(key, i.row, 0 /* searchDir */) - 1
+	if i.transforms.HideObsoletePoints {
+		i.nextObsoletePoint = i.r.isObsolete.SeekSetBitGE(max(i.row, 0))
+		if i.atObsoletePointBackward() {
+			i.skipObsoletePointsBackward()
+			if i.row < 0 {
+				return nil
+			}
+		}
+	}
+	return i.decodeRow()
 }
 
 // First implements the base.InternalIterator interface.
 func (i *DataBlockIter) First() *base.InternalKV {
-	return i.decodeRow(0)
+	if i.r == nil {
+		return nil
+	}
+	i.row = 0
+	if i.transforms.HideObsoletePoints {
+		i.nextObsoletePoint = i.r.isObsolete.SeekSetBitGE(0)
+		if i.atObsoletePointForward() {
+			i.skipObsoletePointsForward()
+			if i.row > i.maxRow {
+				return nil
+			}
+		}
+	}
+	return i.decodeRow()
 }
 
 // Last implements the base.InternalIterator interface.
 func (i *DataBlockIter) Last() *base.InternalKV {
-	return i.decodeRow(i.r.BlockReader().Rows() - 1)
+	if i.r == nil {
+		return nil
+	}
+	i.row = i.maxRow
+	if i.transforms.HideObsoletePoints {
+		i.nextObsoletePoint = i.maxRow + 1
+		if i.atObsoletePointBackward() {
+			i.skipObsoletePointsBackward()
+			if i.row < 0 {
+				return nil
+			}
+		}
+	}
+	return i.decodeRow()
 }
 
 // Next advances to the next KV pair in the block.
 func (i *DataBlockIter) Next() *base.InternalKV {
+	if i.r == nil {
+		return nil
+	}
 	// Inline decodeRow, but avoiding unnecessary checks against i.row.
 	if i.row >= i.maxRow {
 		i.row = i.maxRow + 1
 		return nil
 	}
 	i.row++
+	if i.transforms.HideObsoletePoints && i.atObsoletePointForward() {
+		i.skipObsoletePointsForward()
+		if i.row > i.maxRow {
+			return nil
+		}
+	}
 	// Inline decodeKey().
 	i.kv.K = base.InternalKey{
 		UserKey: i.keySeeker.MaterializeUserKey(&i.keyIter, i.kvRow, i.row),
@@ -856,7 +1039,7 @@ func (i *DataBlockIter) Next() *base.InternalKV {
 // organized as a slice of 64-bit words. If a 64-bit word in the bitmap is zero
 // then all of the rows corresponding to the bits in that word have the same
 // prefix and we can skip ahead. If a row is non-zero a small bit of bit
-// shifting and masking combined with bits.TrailingZeros64 can identify the the
+// shifting and masking combined with bits.TrailingZeros64 can identify the
 // next bit that is set after the current row. The bitmap uses 1 bit/row (0.125
 // bytes/row). A 32KB block containing 1300 rows (25 bytes/row) would need a
 // bitmap of 21 64-bit words. Even in the worst case where every word is 0 this
@@ -864,19 +1047,88 @@ func (i *DataBlockIter) Next() *base.InternalKV {
 // time of ~30 ns if a row is found and decodeRow are called. In more normal
 // cases, NextPrefix takes ~15% longer that a single Next call.
 //
-// For comparision, the rowblk nextPrefixV3 optimizations work by setting a bit
+// For comparison, the rowblk nextPrefixV3 optimizations work by setting a bit
 // in the value prefix byte that indicates that the current key has the same
 // prefix as the previous key. Additionally, a bit is stolen from the restart
 // table entries indicating whether a restart table entry has the same key
 // prefix as the previous entry. Checking the value prefix byte bit requires
 // locating that byte which requires decoding 3 varints per key/value pair.
 func (i *DataBlockIter) NextPrefix(_ []byte) *base.InternalKV {
-	return i.decodeRow(i.r.prefixChanged.SeekSetBitGE(i.row + 1))
+	if i.r == nil {
+		return nil
+	}
+	i.row = i.r.prefixChanged.SeekSetBitGE(i.row + 1)
+	if i.transforms.HideObsoletePoints {
+		i.nextObsoletePoint = i.r.isObsolete.SeekSetBitGE(i.row)
+		if i.atObsoletePointForward() {
+			i.skipObsoletePointsForward()
+		}
+	}
+
+	return i.decodeRow()
 }
 
 // Prev moves the iterator to the previous KV pair in the block.
 func (i *DataBlockIter) Prev() *base.InternalKV {
-	return i.decodeRow(i.row - 1)
+	if i.r == nil {
+		return nil
+	}
+	i.row--
+	if i.transforms.HideObsoletePoints && i.atObsoletePointBackward() {
+		i.skipObsoletePointsBackward()
+		if i.row < 0 {
+			return nil
+		}
+	}
+	return i.decodeRow()
+}
+
+// atObsoletePointForward returns true if i.row is an obsolete point. It is
+// separate from skipObsoletePointsForward() because that method does not
+// inline. It can only be used during forward iteration (i.e. i.row was
+// incremented).
+//
+//gcassert:inline
+func (i *DataBlockIter) atObsoletePointForward() bool {
+	if invariants.Enabled && i.row > i.nextObsoletePoint {
+		panic("invalid nextObsoletePoint")
+	}
+	return i.row == i.nextObsoletePoint && i.row <= i.maxRow
+}
+
+func (i *DataBlockIter) skipObsoletePointsForward() {
+	if invariants.Enabled {
+		i.atObsoletePointCheck()
+	}
+	i.row = i.r.isObsolete.SeekUnsetBitGE(i.row)
+	i.nextObsoletePoint = i.r.isObsolete.SeekSetBitGE(i.row)
+}
+
+// atObsoletePointBackward returns true if i.row is an obsolete point. It is
+// separate from skipObsoletePointsBackward() because that method does not
+// inline. It can only be used during reverse iteration (i.e. i.row was
+// decremented).
+//
+//gcassert:inline
+func (i *DataBlockIter) atObsoletePointBackward() bool {
+	return i.row >= 0 && i.r.isObsolete.At(i.row)
+}
+
+func (i *DataBlockIter) skipObsoletePointsBackward() {
+	if invariants.Enabled {
+		i.atObsoletePointCheck()
+	}
+	i.row = i.r.isObsolete.SeekUnsetBitLE(i.row)
+	i.nextObsoletePoint = i.row + 1
+}
+
+func (i *DataBlockIter) atObsoletePointCheck() {
+	// We extract this code into a separate function to avoid getting a spurious
+	// error from GCAssert about At not being inlined because it is compiled out
+	// altogether in non-invariant builds.
+	if !i.transforms.HideObsoletePoints || !i.r.isObsolete.At(i.row) {
+		panic("expected obsolete point")
+	}
 }
 
 // Error implements the base.InternalIterator interface. A DataBlockIter is
@@ -906,20 +1158,15 @@ func (i *DataBlockIter) DebugTree(tp treeprinter.Node) {
 	tp.Childf("%T(%p)", i, i)
 }
 
-func (i *DataBlockIter) decodeRow(row int) *base.InternalKV {
+// decodeRow decodes i.row into i.kv. If i.row is invalid, it returns nil.
+func (i *DataBlockIter) decodeRow() *base.InternalKV {
 	switch {
-	case row < 0:
-		i.row = -1
-		return nil
-	case row >= i.r.BlockReader().Rows():
-		i.row = i.r.BlockReader().Rows()
+	case i.row < 0 || i.row > i.maxRow:
 		return nil
-	case i.kvRow == row:
-		i.row = row
+	case i.kvRow == i.row:
 		// Already synthesized the kv at row.
 		return &i.kv
 	default:
-		i.row = row
 		// Inline decodeKey().
 		i.kv.K = base.InternalKey{
 			UserKey: i.keySeeker.MaterializeUserKey(&i.keyIter, i.kvRow, i.row),
@@ -929,14 +1176,14 @@ func (i *DataBlockIter) decodeRow(row int) *base.InternalKV {
 			i.kv.K.SetSeqNum(base.SeqNum(n))
 		}
 		// Inline i.r.values.At(row).
-		startOffset := i.r.values.offsets.At(row)
-		v := unsafe.Slice((*byte)(i.r.values.ptr(startOffset)), i.r.values.offsets.At(row+1)-startOffset)
-		if i.r.isValueExternal.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)
+		if i.r.isValueExternal.At(i.row) {
 			i.kv.V = i.getLazyValuer.GetLazyValueForPrefixAndValueHandle(v)
 		} else {
 			i.kv.V = base.MakeInPlaceValue(v)
 		}
-		i.kvRow = row
+		i.kvRow = i.row
 		return &i.kv
 	}
 }
@@ -956,17 +1203,6 @@ func (i *DataBlockIter) decodeKey() {
 
 var _ = (*DataBlockIter).decodeKey
 
-// Valid returns true if the iterator is currently positioned at a valid KV.
-func (i *DataBlockIter) Valid() bool {
-	return i.row < 0 || i.row > i.maxRow
-}
-
-// KV returns the key-value pair at the current iterator position. The
-// iterator must be positioned over a valid KV.
-func (i *DataBlockIter) KV() *base.InternalKV {
-	return &i.kv
-}
-
 // Close implements the base.InternalIterator interface.
 func (i *DataBlockIter) Close() error {
 	i.keySeeker.Release()
diff --git a/sstable/colblk/index_block.go b/sstable/colblk/index_block.go
index dc7747c2..e840c8d6 100644
--- a/sstable/colblk/index_block.go
+++ b/sstable/colblk/index_block.go
@@ -53,7 +53,7 @@ const (
 )
 
 // Init initializes the index block writer.
-func (w *IndexBlockWriter) Init(blockProperties ...ColumnWriter) {
+func (w *IndexBlockWriter) Init() {
 	w.separators.Init()
 	w.offsets.Init()
 	w.lengths.InitWithDefault()
@@ -191,13 +191,15 @@ var _ block.IndexBlockIterator = (*IndexIter)(nil)
 
 // InitReader initializes an index iterator from the provided reader.
 func (i *IndexIter) InitReader(r *IndexReader) {
-	*i = IndexIter{r: r, n: int(r.br.header.Rows), allocReader: i.allocReader}
+	*i = IndexIter{r: r, n: int(r.br.header.Rows), h: i.h, allocReader: i.allocReader}
 }
 
 // Init initializes an iterator from the provided block data slice.
 func (i *IndexIter) Init(
 	cmp base.Compare, split base.Split, blk []byte, transforms block.IterTransforms,
 ) error {
+	i.h.Release()
+	i.h = block.BufferHandle{}
 	// TODO(jackson): Handle the transforms.
 	i.allocReader.Init(blk)
 	i.InitReader(&i.allocReader)
@@ -206,7 +208,7 @@ func (i *IndexIter) Init(
 
 // 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,
+	cmp base.Compare, split base.Split, blk block.BufferHandle, transforms block.IterTransforms,
 ) error {
 	// TODO(jackson): Handle the transforms.
 
@@ -216,8 +218,10 @@ func (i *IndexIter) InitHandle(
 	// common to open an iterator and perform just a few seeks, so avoiding the
 	// overhead can be material.)
 	i.h.Release()
-	i.h = block
-	return i.Init(cmp, split, i.h.Get(), transforms)
+	i.h = blk
+	i.allocReader.Init(i.h.Get())
+	i.InitReader(&i.allocReader)
+	return nil
 }
 
 // RowIndex returns the index of the block entry at the iterator's current
@@ -229,20 +233,23 @@ func (i *IndexIter) RowIndex() int {
 // ResetForReuse resets the IndexIter for reuse, retaining buffers to avoid
 // future allocations.
 func (i *IndexIter) ResetForReuse() IndexIter {
+	if invariants.Enabled && i.h != (block.BufferHandle{}) {
+		panic(errors.AssertionFailedf("IndexIter reset for reuse with non-empty handle"))
+	}
 	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)
+	return 0 <= i.row && i.row < i.n
 }
 
 // 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{}
+	i.n = 0
 }
 
 // IsDataInvalidated returns true when the iterator has been invalidated
@@ -315,17 +322,19 @@ func (i *IndexIter) Last() bool {
 }
 
 // Next steps the index iterator to the next block entry. It returns false if
-// the index block is exhausted.
+// the index block is exhausted in the forward direction. A call to Next while
+// already exhausted in the forward direction is a no-op.
 func (i *IndexIter) Next() bool {
-	i.row++
+	i.row = min(i.n, i.row+1)
 	return i.row < i.n
 }
 
 // Prev steps the index iterator to the previous block entry. It returns false
-// if the index block is exhausted.
+// if the index block is exhausted in the reverse direction. A call to Prev
+// while already exhausted in the reverse direction is a no-op.
 func (i *IndexIter) Prev() bool {
-	i.row--
-	return i.row >= 0
+	i.row = max(-1, i.row-1)
+	return i.row >= 0 && i.row < i.n
 }
 
 // Close closes the iterator, releasing any resources it holds.
diff --git a/sstable/colblk/keyspan.go b/sstable/colblk/keyspan.go
index dac1cad1..d8e50f0d 100644
--- a/sstable/colblk/keyspan.go
+++ b/sstable/colblk/keyspan.go
@@ -132,6 +132,21 @@ func (w *KeyspanBlockWriter) UnsafeBoundaryKeys() (smallest, largest base.Intern
 	return smallest, largest
 }
 
+// UnsafeLastSpan returns the start and end user keys of the last span written
+// to the block and the trailer of its largest key. The returned keys point
+// directly into the block writer's memory and must not be mutated.
+func (w *KeyspanBlockWriter) UnsafeLastSpan() (
+	start, end []byte,
+	largestTrailer base.InternalKeyTrailer,
+) {
+	if w.keyCount == 0 {
+		return nil, nil, 0
+	}
+	return w.boundaryUserKeys.UnsafeGet(w.boundaryUserKeys.rows - 2),
+		w.boundaryUserKeys.UnsafeGet(w.boundaryUserKeys.rows - 1),
+		base.InternalKeyTrailer(w.trailers.Get(w.keyCount - 1))
+}
+
 // Size returns the size of the pending block.
 func (w *KeyspanBlockWriter) Size() int {
 	off := blockHeaderSize(keyspanColumnCount, keyspanHeaderSize)
diff --git a/sstable/colblk_writer.go b/sstable/colblk_writer.go
index 55bd3497..f9f517a8 100644
--- a/sstable/colblk_writer.go
+++ b/sstable/colblk_writer.go
@@ -254,11 +254,31 @@ func (w *RawColumnWriter) EncodeSpan(span keyspan.Span) error {
 	for _, k := range span.Keys {
 		w.meta.updateSeqNum(k.SeqNum())
 	}
+
+	blockWriter := &w.rangeKeyBlock
 	if span.Keys[0].Kind() == base.InternalKeyKindRangeDelete {
-		w.rangeDelBlock.AddSpan(span)
-		return nil
+		blockWriter = &w.rangeDelBlock
+	}
+	if !w.disableKeyOrderChecks && blockWriter.KeyCount() > 0 {
+		// Check that spans are being added in fragmented order. If the two
+		// tombstones overlap, their start and end keys must be identical.
+		prevStart, prevEnd, prevTrailer := blockWriter.UnsafeLastSpan()
+		if w.opts.Comparer.Equal(prevStart, span.Start) && w.opts.Comparer.Equal(prevEnd, span.End) {
+			if prevTrailer < span.Keys[0].Trailer {
+				w.err = errors.Errorf("pebble: keys must be added in order: %s-%s:{(#%s)}, %s",
+					w.opts.Comparer.FormatKey(prevStart),
+					w.opts.Comparer.FormatKey(prevEnd),
+					prevTrailer, span.Pretty(w.opts.Comparer.FormatKey))
+			}
+		} else if c := w.opts.Comparer.Compare(prevEnd, span.Start); c > 0 {
+			w.err = errors.Errorf("pebble: keys must be added in order: %s-%s:{(#%s)}, %s",
+				w.opts.Comparer.FormatKey(prevStart),
+				w.opts.Comparer.FormatKey(prevEnd),
+				prevTrailer, span.Pretty(w.opts.Comparer.FormatKey))
+			return w.err
+		}
 	}
-	w.rangeKeyBlock.AddSpan(span)
+	blockWriter.AddSpan(span)
 	return nil
 }
 
@@ -589,6 +609,10 @@ func (w *RawColumnWriter) enqueueDataBlock(
 		// lifetime of the DataBlockWriters?
 		cb.physical = cb.physical.Clone()
 	}
+	return w.enqueuePhysicalBlock(cb, separator)
+}
+
+func (w *RawColumnWriter) enqueuePhysicalBlock(cb *compressedBlock, separator []byte) error {
 	dataBlockHandle := block.Handle{
 		Offset: w.queuedDataSize,
 		Length: uint64(cb.physical.LengthWithoutTrailer()),
@@ -755,6 +779,9 @@ func (w *RawColumnWriter) Close() (err error) {
 			w.err = err
 		}
 	}()
+	if w.layout.writable == nil {
+		return w.err
+	}
 
 	// Finish the last data block and send it to the write queue if it contains
 	// any pending KVs.
@@ -891,7 +918,92 @@ func (w *RawColumnWriter) Close() (err error) {
 func (w *RawColumnWriter) rewriteSuffixes(
 	r *Reader, wo WriterOptions, from, to []byte, concurrency int,
 ) error {
-	panic("unimplemented")
+	for _, c := range w.blockPropCollectors {
+		if !c.SupportsSuffixReplacement() {
+			return errors.Errorf("block property collector %s does not support suffix replacement", c.Name())
+		}
+	}
+	l, err := r.Layout()
+	if err != nil {
+		return errors.Wrap(err, "reading layout")
+	}
+	// Copy data blocks in parallel, rewriting suffixes as we go.
+	blocks, err := rewriteDataBlocksInParallel(r, wo, l.Data, from, to, concurrency, func() blockRewriter {
+		return &colblk.DataBlockRewriter{KeySchema: wo.KeySchema}
+	})
+	if err != nil {
+		return errors.Wrap(err, "rewriting data blocks")
+	}
+
+	// oldShortIDs maps the shortID for the block property collector in the old
+	// blocks to the shortID in the new blocks. Initialized once for the sstable.
+	oldShortIDs, err := getShortIDs(r, w.blockPropCollectors)
+	if err != nil {
+		return errors.Wrap(err, "getting short IDs")
+	}
+	oldProps := make([][]byte, len(w.blockPropCollectors))
+	for i := range blocks {
+		cb := compressedBlockPool.Get().(*compressedBlock)
+		cb.physical = blocks[i].physical
+
+		// Load any previous values for our prop collectors into oldProps.
+		for i := range oldProps {
+			oldProps[i] = nil
+		}
+		decoder := makeBlockPropertiesDecoder(len(oldProps), l.Data[i].Props)
+		for !decoder.Done() {
+			id, val, err := decoder.Next()
+			if err != nil {
+				return err
+			}
+			if oldShortIDs[id].IsValid() {
+				oldProps[oldShortIDs[id]] = val
+			}
+		}
+		for i, p := range w.blockPropCollectors {
+			if err := p.AddCollectedWithSuffixReplacement(oldProps[i], from, to); err != nil {
+				return err
+			}
+		}
+		var separator []byte
+		if i+1 < len(blocks) {
+			w.separatorBuf = w.comparer.Separator(w.separatorBuf[:0], blocks[i].end.UserKey, blocks[i+1].start.UserKey)
+			separator = w.separatorBuf
+		} else {
+			w.separatorBuf = w.comparer.Successor(w.separatorBuf[:0], blocks[i].end.UserKey)
+			separator = w.separatorBuf
+		}
+		w.enqueuePhysicalBlock(cb, separator)
+	}
+
+	if len(blocks) > 0 {
+		w.meta.updateSeqNum(blocks[0].start.SeqNum())
+		w.props.NumEntries = r.Properties.NumEntries
+		w.props.RawKeySize = r.Properties.RawKeySize
+		w.props.RawValueSize = r.Properties.RawValueSize
+		w.meta.SetSmallestPointKey(blocks[0].start)
+		w.meta.SetLargestPointKey(blocks[len(blocks)-1].end)
+	}
+
+	// Copy range key block, replacing suffixes if it exists.
+	if err := rewriteRangeKeyBlockToWriter(r, w, from, to); err != nil {
+		return errors.Wrap(err, "rewriting range key blocks")
+	}
+	// Copy over the filter block if it exists.
+	if w.filterBlock != nil {
+		if filterBlockBH, ok := l.FilterByName(w.filterBlock.metaName()); ok {
+			filterBlock, _, err := readBlockBuf(r, filterBlockBH, nil)
+			if err != nil {
+				return errors.Wrap(err, "reading filter")
+			}
+			w.filterBlock = copyFilterWriter{
+				origPolicyName: w.filterBlock.policyName(),
+				origMetaName:   w.filterBlock.metaName(),
+				data:           filterBlock,
+			}
+		}
+	}
+	return nil
 }
 
 func shouldFlushWithoutLatestKV(
diff --git a/sstable/reader.go b/sstable/reader.go
index 36ae62ea..7cc5f1dc 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -232,21 +232,40 @@ func (r *Reader) newCompactionIter(
 		transforms.HideObsoletePoints = true
 	}
 	if r.Properties.IndexType == twoLevelIndex {
-		i, err := newRowBlockTwoLevelIterator(
+		if !r.tableFormat.BlockColumnar() {
+			i, err := newRowBlockTwoLevelIterator(
+				context.Background(),
+				r, vState, transforms, nil /* lower */, nil /* upper */, nil,
+				NeverUseFilterBlock, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool)
+			if err != nil {
+				return nil, err
+			}
+			i.SetupForCompaction()
+			return i, nil
+		}
+		i, err := newColumnBlockTwoLevelIterator(
 			context.Background(),
 			r, vState, transforms, nil /* lower */, nil /* upper */, nil,
-			NeverUseFilterBlock, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool,
-		)
+			NeverUseFilterBlock, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool)
 		if err != nil {
 			return nil, err
 		}
 		i.SetupForCompaction()
 		return i, nil
 	}
-	i, err := newRowBlockSingleLevelIterator(
+	if !r.tableFormat.BlockColumnar() {
+		i, err := newRowBlockSingleLevelIterator(
+			context.Background(), r, vState, transforms, nil /* lower */, nil, /* upper */
+			nil, NeverUseFilterBlock, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool)
+		if err != nil {
+			return nil, err
+		}
+		i.SetupForCompaction()
+		return i, nil
+	}
+	i, err := newColumnBlockSingleLevelIterator(
 		context.Background(), r, vState, transforms, nil /* lower */, nil, /* upper */
-		nil, NeverUseFilterBlock, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool,
-	)
+		nil, NeverUseFilterBlock, nil /* stats */, categoryAndQoS, statsCollector, rp, bufferPool)
 	if err != nil {
 		return nil, err
 	}
@@ -739,7 +758,10 @@ func (r *Reader) CommonProperties() *CommonProperties {
 // data blocks overlapped and add that same fraction of the metadata blocks to the
 // estimate.
 func (r *Reader) EstimateDiskUsage(start, end []byte) (uint64, error) {
-	return estimateDiskUsage[rowblk.IndexIter, *rowblk.IndexIter](r, start, end)
+	if !r.tableFormat.BlockColumnar() {
+		return estimateDiskUsage[rowblk.IndexIter, *rowblk.IndexIter](r, start, end)
+	}
+	return estimateDiskUsage[colblk.IndexIter, *colblk.IndexIter](r, start, end)
 }
 
 func estimateDiskUsage[I any, PI indexBlockIterator[I]](
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index e649a71c..b8085536 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -6,8 +6,10 @@ package sstable
 
 import (
 	"bytes"
+	"cmp"
 	"context"
 	"math"
+	"slices"
 	"sync"
 
 	"github.com/cockroachdb/errors"
@@ -144,7 +146,11 @@ func rewriteDataBlocksInParallel(
 
 	g := &sync.WaitGroup{}
 	g.Add(concurrency)
-	errCh := make(chan error, concurrency)
+	type workerErr struct {
+		worker int
+		err    error
+	}
+	errCh := make(chan workerErr, concurrency)
 	for j := 0; j < concurrency; j++ {
 		worker := j
 		go func() {
@@ -177,14 +183,20 @@ func rewriteDataBlocksInParallel(
 				return nil
 			}()
 			if err != nil {
-				errCh <- err
+				errCh <- workerErr{worker: worker, err: err}
 			}
 		}()
 	}
 	g.Wait()
 	close(errCh)
-	if err, ok := <-errCh; ok {
-		return nil, err
+	if werr, ok := <-errCh; ok {
+		// Collect errors from all workers and sort them by worker for determinism.
+		werrs := []workerErr{werr}
+		for werr := range errCh {
+			werrs = append(werrs, werr)
+		}
+		slices.SortFunc(werrs, func(a, b workerErr) int { return cmp.Compare(a.worker, b.worker) })
+		return nil, werrs[0].err
 	}
 	return output, nil
 }
diff --git a/version_set.go b/version_set.go
index 7ed72611..f8bea8f3 100644
--- a/version_set.go
+++ b/version_set.go
@@ -113,7 +113,7 @@ type versionSet struct {
 
 	// The next file number. A single counter is used to assign file
 	// numbers for the WAL, MANIFEST, sstable, and OPTIONS files.
-	nextFileNum uint64
+	nextFileNum atomic.Uint64
 
 	// The current manifest file number.
 	manifestFileNum base.DiskFileNum
@@ -154,7 +154,7 @@ func (vs *versionSet) init(
 	vs.obsoleteFn = vs.addObsoleteLocked
 	vs.zombieTables = make(map[base.DiskFileNum]tableInfo)
 	vs.virtualBackings = manifest.MakeVirtualBackings()
-	vs.nextFileNum = 1
+	vs.nextFileNum.Store(1)
 	vs.manifestMarker = marker
 	vs.getFormatMajorVersion = getFMV
 }
@@ -181,7 +181,7 @@ func (vs *versionSet) create(
 	// Note that a "snapshot" version edit is written to the manifest when it is
 	// created.
 	vs.manifestFileNum = vs.getNextDiskFileNum()
-	err = vs.createManifest(vs.dirname, vs.manifestFileNum, vs.minUnflushedLogNum, vs.nextFileNum, nil /* virtualBackings */)
+	err = vs.createManifest(vs.dirname, vs.manifestFileNum, vs.minUnflushedLogNum, vs.nextFileNum.Load(), nil /* virtualBackings */)
 	if err == nil {
 		if err = vs.manifest.Flush(); err != nil {
 			vs.opts.Logger.Fatalf("MANIFEST flush failed: %v", err)
@@ -270,7 +270,7 @@ func (vs *versionSet) load(
 			vs.minUnflushedLogNum = ve.MinUnflushedLogNum
 		}
 		if ve.NextFileNum != 0 {
-			vs.nextFileNum = ve.NextFileNum
+			vs.nextFileNum.Store(ve.NextFileNum)
 		}
 		if ve.LastSeqNum != 0 {
 			// logSeqNum is the _next_ sequence number that will be assigned,
@@ -294,7 +294,7 @@ func (vs *versionSet) load(
 	// function and could have only updated it to some other non-zero value,
 	// so it cannot be 0 here.
 	if vs.minUnflushedLogNum == 0 {
-		if vs.nextFileNum >= 2 {
+		if vs.nextFileNum.Load() >= 2 {
 			// We either have a freshly created DB, or a DB created by RocksDB
 			// that has not had a single flushed SSTable yet. This is because
 			// RocksDB bumps up nextFileNum in this case without bumping up
@@ -441,7 +441,7 @@ func (vs *versionSet) logAndApply(
 
 	if ve.MinUnflushedLogNum != 0 {
 		if ve.MinUnflushedLogNum < vs.minUnflushedLogNum ||
-			vs.nextFileNum <= uint64(ve.MinUnflushedLogNum) {
+			vs.nextFileNum.Load() <= uint64(ve.MinUnflushedLogNum) {
 			panic(fmt.Sprintf("pebble: inconsistent versionEdit minUnflushedLogNum %d",
 				ve.MinUnflushedLogNum))
 		}
@@ -452,7 +452,7 @@ func (vs *versionSet) logAndApply(
 	// current filenum and not the next one.
 	//
 	// TODO(sbhola): figure out why this is correct and update comment.
-	ve.NextFileNum = vs.nextFileNum
+	ve.NextFileNum = vs.nextFileNum.Load()
 
 	// LastSeqNum is set to the current upper bound on the assigned sequence
 	// numbers. Note that this is exactly the behavior of RocksDB. LastSeqNum is
@@ -542,7 +542,7 @@ func (vs *versionSet) logAndApply(
 	// Grab certain values before releasing vs.mu, in case createManifest() needs
 	// to be called.
 	minUnflushedLogNum := vs.minUnflushedLogNum
-	nextFileNum := vs.nextFileNum
+	nextFileNum := vs.nextFileNum.Load()
 
 	// Note: this call populates ve.RemovedBackingTables.
 	zombieBackings, removedVirtualBackings, localLiveSizeDelta :=
@@ -949,21 +949,21 @@ func (vs *versionSet) createManifest(
 	return nil
 }
 
+// NB: This method is not safe for concurrent use. It is only safe
+// to be called when concurrent changes to nextFileNum are not expected.
 func (vs *versionSet) markFileNumUsed(fileNum base.DiskFileNum) {
-	if vs.nextFileNum <= uint64(fileNum) {
-		vs.nextFileNum = uint64(fileNum + 1)
+	if vs.nextFileNum.Load() <= uint64(fileNum) {
+		vs.nextFileNum.Store(uint64(fileNum + 1))
 	}
 }
 
 func (vs *versionSet) getNextFileNum() base.FileNum {
-	x := vs.nextFileNum
-	vs.nextFileNum++
+	x := vs.nextFileNum.Add(1) - 1
 	return base.FileNum(x)
 }
 
 func (vs *versionSet) getNextDiskFileNum() base.DiskFileNum {
-	x := vs.nextFileNum
-	vs.nextFileNum++
+	x := vs.nextFileNum.Add(1) - 1
 	return base.DiskFileNum(x)
 }
 


Test patch:
diff --git a/flushable_test.go b/flushable_test.go
index 3be5fd90..982757a5 100644
--- a/flushable_test.go
+++ b/flushable_test.go
@@ -49,13 +49,11 @@ func TestIngestedSSTFlushableAPI(t *testing.T) {
 	reset()
 
 	loadFileMeta := func(paths []string, exciseSpan KeyRange, seqNum base.SeqNum) []*fileMetadata {
-		d.mu.Lock()
 		pendingOutputs := make([]base.FileNum, len(paths))
 		for i := range paths {
 			pendingOutputs[i] = d.mu.versions.getNextFileNum()
 		}
-		jobID := d.newJobIDLocked()
-		d.mu.Unlock()
+		jobID := d.newJobID()
 
 		// We can reuse the ingestLoad function for this test even if we're
 		// not actually ingesting a file.
diff --git a/internal/crdbtest/crdbtest.go b/internal/crdbtest/crdbtest.go
index ffa45bff..5a65f3ef 100644
--- a/internal/crdbtest/crdbtest.go
+++ b/internal/crdbtest/crdbtest.go
@@ -14,6 +14,7 @@ import (
 	"slices"
 	"time"
 
+	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"golang.org/x/exp/rand"
@@ -348,8 +349,8 @@ type KeyConfig struct {
 }
 
 func (cfg KeyConfig) String() string {
-	return fmt.Sprintf("AlphaLen=%d,Shared=%d,PrefixLen=%d,Logical=%d",
-		cfg.PrefixAlphabetLen, cfg.PrefixLenShared, cfg.PrefixLen, cfg.Logical)
+	return fmt.Sprintf("AlphaLen=%d,Prefix=%d,Shared=%d%s",
+		cfg.PrefixAlphabetLen, cfg.PrefixLen, cfg.PrefixLenShared, crstrings.If(cfg.Logical != 0, ",Logical"))
 }
 
 // RandomKVs constructs count random KVs with the provided parameters.
diff --git a/internal/itertest/datadriven.go b/internal/itertest/datadriven.go
index 35d5d2d1..abfc03c4 100644
--- a/internal/itertest/datadriven.go
+++ b/internal/itertest/datadriven.go
@@ -13,6 +13,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/keyspan"
@@ -22,6 +23,7 @@ import (
 
 type iterCmdOpts struct {
 	fmtKV           formatKV
+	showCommands    bool
 	withoutNewlines bool
 	stats           *base.InternalIteratorStats
 }
@@ -39,6 +41,12 @@ func Condensed(opts *iterCmdOpts) {
 	opts.withoutNewlines = true
 }
 
+// ShowCommands configures RunInternalIterCmd to show the command in each output
+// line (so you don't have to visually match the line to the command).
+func ShowCommands(opts *iterCmdOpts) {
+	opts.showCommands = true
+}
+
 // Verbose configures RunInternalIterCmd to output verbose results.
 func Verbose(opts *iterCmdOpts) { opts.fmtKV = verboseFormatKV }
 
@@ -125,11 +133,13 @@ func RunInternalIterCmdWriter(
 		require.NoError(t, err)
 		return &kv.K, v
 	}
-	for _, line := range strings.Split(d.Input, "\n") {
+	lines := crstrings.Lines(d.Input)
+	maxCmdLen := 1
+	for _, line := range lines {
+		maxCmdLen = max(maxCmdLen, len(line))
+	}
+	for _, line := range lines {
 		parts := strings.Fields(line)
-		if len(parts) == 0 {
-			continue
-		}
 		var key *base.InternalKey
 		var value []byte
 		switch parts[0] {
@@ -222,6 +232,9 @@ func RunInternalIterCmdWriter(
 			fmt.Fprintf(w, "unknown op: %s", parts[0])
 			return
 		}
+		if o.showCommands {
+			fmt.Fprintf(w, "%*s: ", min(maxCmdLen, 40), line)
+		}
 		o.fmtKV(w, key, value, iter)
 		if !o.withoutNewlines {
 			fmt.Fprintln(w)
diff --git a/sstable/block_property_test.go b/sstable/block_property_test.go
index bc5af2da..75027abc 100644
--- a/sstable/block_property_test.go
+++ b/sstable/block_property_test.go
@@ -24,8 +24,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/rangekey"
 	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/cockroachdb/pebble/sstable/block"
-	"github.com/cockroachdb/pebble/sstable/rowblk"
 	"github.com/stretchr/testify/require"
 )
 
@@ -859,7 +857,7 @@ func TestBlockProperties(t *testing.T) {
 			return runTablePropsCmd(r, td)
 
 		case "block-props":
-			return runBlockPropsCmd(r, td)
+			return runBlockPropsCmd(r)
 
 		case "filter":
 			syntheticSuffix := make([]byte, 0, binary.MaxVarintLen64)
@@ -912,9 +910,12 @@ func TestBlockProperties(t *testing.T) {
 
 					var blocks []int
 					var i int
-					iter, _ := rowblk.NewIter(r.Compare, r.Split, indexH.Get(), NoTransforms)
-					for kv := iter.First(); kv != nil; kv = iter.Next() {
-						bh, err := block.DecodeHandleWithProperties(kv.InPlaceValue())
+					iter := r.tableFormat.newIndexIter()
+					if err := iter.Init(r.Compare, r.Split, indexH.Get(), NoTransforms); err != nil {
+						return err.Error()
+					}
+					for valid := iter.First(); valid; valid = iter.Next() {
+						bh, err := iter.BlockHandleWithProperties()
 						if err != nil {
 							return err.Error()
 						}
@@ -1022,7 +1023,7 @@ func TestBlockProperties_BoundLimited(t *testing.T) {
 		case "table-props":
 			return runTablePropsCmd(r, td)
 		case "block-props":
-			return runBlockPropsCmd(r, td)
+			return runBlockPropsCmd(r)
 		case "iter":
 			var buf bytes.Buffer
 			var lower, upper []byte
@@ -1267,17 +1268,17 @@ func runBlockPropertiesBuildCmd(td *datadriven.TestData) (r *Reader, out string)
 	return r, formatWriterMetadata(td, meta)
 }
 
-func runBlockPropsCmd(r *Reader, td *datadriven.TestData) string {
+func runBlockPropsCmd(r *Reader) string {
 	bh, err := r.readIndex(context.Background(), nil, nil, nil)
 	if err != nil {
 		return err.Error()
 	}
+	defer bh.Release()
 	twoLevelIndex := r.Properties.IndexPartitions > 0
-	i, err := rowblk.NewIter(r.Compare, r.Split, bh.Get(), NoTransforms)
-	if err != nil {
+	i := r.tableFormat.newIndexIter()
+	if err := i.Init(r.Compare, r.Split, bh.Get(), NoTransforms); err != nil {
 		return err.Error()
 	}
-	defer bh.Release()
 	var sb strings.Builder
 	decodeProps := func(props []byte, indent string) error {
 		d := makeBlockPropertiesDecoder(11, props)
@@ -1304,9 +1305,9 @@ func runBlockPropsCmd(r *Reader, td *datadriven.TestData) string {
 		return nil
 	}
 
-	for kv := i.First(); kv != nil; kv = i.Next() {
-		sb.WriteString(fmt.Sprintf("%s:\n", kv.K))
-		bhp, err := block.DecodeHandleWithProperties(kv.InPlaceValue())
+	for valid := i.First(); valid; valid = i.Next() {
+		sb.WriteString(fmt.Sprintf("%s:\n", i.Separator()))
+		bhp, err := i.BlockHandleWithProperties()
 		if err != nil {
 			return err.Error()
 		}
@@ -1317,26 +1318,32 @@ func runBlockPropsCmd(r *Reader, td *datadriven.TestData) string {
 		// If the table has a two-level index, also decode the index
 		// block that bhp points to, along with its block properties.
 		if twoLevelIndex {
-			subiter := &rowblk.Iter{}
 			subIndex, err := r.readBlock(
 				context.Background(), bhp.Handle, nil, nil, nil, nil, nil)
 			if err != nil {
 				return err.Error()
 			}
-			if err := subiter.Init(r.Compare, r.Split, subIndex.Get(), NoTransforms); err != nil {
-				return err.Error()
-			}
-			for kv := subiter.First(); kv != nil; kv = subiter.Next() {
-				sb.WriteString(fmt.Sprintf("  %s:\n", kv.K))
-				dataBH, err := block.DecodeHandleWithProperties(kv.InPlaceValue())
-				if err != nil {
-					return err.Error()
+			err = func() error {
+				defer subIndex.Release()
+				subiter := r.tableFormat.newIndexIter()
+				if err := subiter.Init(r.Compare, r.Split, subIndex.Get(), NoTransforms); err != nil {
+					return err
 				}
-				if err := decodeProps(dataBH.Props, "    "); err != nil {
-					return err.Error()
+				for valid := subiter.First(); valid; valid = subiter.Next() {
+					sb.WriteString(fmt.Sprintf("  %s:\n", subiter.Separator()))
+					dataBH, err := subiter.BlockHandleWithProperties()
+					if err != nil {
+						return err
+					}
+					if err := decodeProps(dataBH.Props, "    "); err != nil {
+						return err
+					}
 				}
+				return nil
+			}()
+			if err != nil {
+				return err.Error()
 			}
-			subIndex.Release()
 		}
 	}
 	return sb.String()
diff --git a/sstable/colblk/cockroach_test.go b/sstable/colblk/cockroach_test.go
index 5cd1f209..ba9c10b5 100644
--- a/sstable/colblk/cockroach_test.go
+++ b/sstable/colblk/cockroach_test.go
@@ -16,6 +16,7 @@ import (
 	"unsafe"
 
 	"github.com/cockroachdb/crlib/crbytes"
+	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/crdbtest"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -444,24 +445,25 @@ func benchmarkCockroachDataBlockWriter(b *testing.B, keyConfig crdbtest.KeyConfi
 	}
 }
 
-func BenchmarkCockroachDataBlockIter(b *testing.B) {
+func BenchmarkCockroachDataBlockIterFull(b *testing.B) {
 	for _, alphaLen := range []int{4, 8, 26} {
 		for _, lenSharedPct := range []float64{0.25, 0.5} {
 			for _, prefixLen := range []int{8, 32, 128} {
 				lenShared := int(float64(prefixLen) * lenSharedPct)
 				for _, logical := range []uint32{0, 1} {
 					for _, valueLen := range []int{8, 128, 1024} {
-						keyConfig := crdbtest.KeyConfig{
-							PrefixAlphabetLen: alphaLen,
-							PrefixLen:         prefixLen,
-							PrefixLenShared:   lenShared,
-							Logical:           logical,
-							BaseWallTime:      uint64(time.Now().UnixNano()),
+						cfg := benchConfig{
+							KeyConfig: crdbtest.KeyConfig{
+								PrefixAlphabetLen: alphaLen,
+								PrefixLen:         prefixLen,
+								PrefixLenShared:   lenShared,
+								Logical:           logical,
+							},
+							ValueLen: valueLen,
 						}
-						b.Run(fmt.Sprintf("%s,value=%d", keyConfig, valueLen),
-							func(b *testing.B) {
-								benchmarkCockroachDataBlockIter(b, keyConfig, valueLen)
-							})
+						b.Run(cfg.String(), func(b *testing.B) {
+							benchmarkCockroachDataBlockIter(b, cfg, block.IterTransforms{})
+						})
 					}
 				}
 			}
@@ -469,11 +471,79 @@ func BenchmarkCockroachDataBlockIter(b *testing.B) {
 	}
 }
 
-func benchmarkCockroachDataBlockIter(b *testing.B, keyConfig crdbtest.KeyConfig, valueLen int) {
+var shortBenchConfigs = []benchConfig{
+	{
+		KeyConfig: crdbtest.KeyConfig{
+			PrefixAlphabetLen: 8,
+			PrefixLen:         8,
+			PrefixLenShared:   4,
+		},
+		ValueLen: 8,
+	},
+	{
+		KeyConfig: crdbtest.KeyConfig{
+			PrefixAlphabetLen: 8,
+			PrefixLen:         128,
+			PrefixLenShared:   64,
+		},
+		ValueLen: 128,
+	},
+}
+
+func BenchmarkCockroachDataBlockIterShort(b *testing.B) {
+	for _, cfg := range shortBenchConfigs {
+		b.Run(cfg.String(), func(b *testing.B) {
+			benchmarkCockroachDataBlockIter(b, cfg, block.IterTransforms{})
+		})
+	}
+}
+
+func BenchmarkCockroachDataBlockIterTransforms(b *testing.B) {
+	transforms := []struct {
+		description string
+		transforms  block.IterTransforms
+	}{
+		{},
+		{
+			description: "SynthSeqNum",
+			transforms: block.IterTransforms{
+				SyntheticSeqNum: 1234,
+			},
+		},
+		{
+			description: "HideObsolete",
+			transforms: block.IterTransforms{
+				HideObsoletePoints: true,
+			},
+		},
+	}
+	for _, cfg := range shortBenchConfigs {
+		for _, t := range transforms {
+			name := cfg.String() + crstrings.If(t.description != "", ","+t.description)
+			b.Run(name, func(b *testing.B) {
+				benchmarkCockroachDataBlockIter(b, cfg, t.transforms)
+			})
+		}
+	}
+}
+
+type benchConfig struct {
+	crdbtest.KeyConfig
+	ValueLen int
+}
+
+func (cfg benchConfig) String() string {
+	return fmt.Sprintf("%s,ValueLen=%d", cfg.KeyConfig, cfg.ValueLen)
+}
+
+func benchmarkCockroachDataBlockIter(
+	b *testing.B, cfg benchConfig, transforms block.IterTransforms,
+) {
 	const targetBlockSize = 32 << 10
 	seed := uint64(time.Now().UnixNano())
 	rng := rand.New(rand.NewSource(seed))
-	keys, values := crdbtest.RandomKVs(rng, targetBlockSize/valueLen, keyConfig, valueLen)
+	cfg.BaseWallTime = seed
+	keys, values := crdbtest.RandomKVs(rng, targetBlockSize/cfg.ValueLen, cfg.KeyConfig, cfg.ValueLen)
 
 	var w DataBlockWriter
 	w.Init(cockroachKeySchema)
@@ -481,7 +551,8 @@ func benchmarkCockroachDataBlockIter(b *testing.B, keyConfig crdbtest.KeyConfig,
 	for w.Size() < targetBlockSize {
 		ik := base.MakeInternalKey(keys[count], base.SeqNum(rng.Uint64n(uint64(base.SeqNumMax))), base.InternalKeyKindSet)
 		kcmp := w.KeyWriter.ComparePrev(ik.UserKey)
-		w.Add(ik, values[count], block.InPlaceValuePrefix(kcmp.PrefixEqual()), kcmp, false /* isObsolete */)
+		isObsolete := rng.Intn(20) == 0
+		w.Add(ik, values[count], block.InPlaceValuePrefix(kcmp.PrefixEqual()), kcmp, isObsolete)
 		count++
 	}
 	serializedBlock, _ := w.Finish(w.Rows(), w.Size())
@@ -490,7 +561,7 @@ func benchmarkCockroachDataBlockIter(b *testing.B, keyConfig crdbtest.KeyConfig,
 	reader.Init(cockroachKeySchema, serializedBlock)
 	it.Init(&reader, cockroachKeySchema.NewKeySeeker(), getLazyValuer(func([]byte) base.LazyValue {
 		return base.LazyValue{ValueOrHandle: []byte("mock external value")}
-	}), block.IterTransforms{})
+	}), transforms)
 	avgRowSize := float64(len(serializedBlock)) / float64(count)
 
 	b.Run("Next", func(b *testing.B) {
diff --git a/sstable/colblk/data_block_test.go b/sstable/colblk/data_block_test.go
index 0dcd3c3f..a25a215a 100644
--- a/sstable/colblk/data_block_test.go
+++ b/sstable/colblk/data_block_test.go
@@ -28,6 +28,8 @@ func TestDataBlock(t *testing.T) {
 	var w DataBlockWriter
 	var r DataBlockReader
 	var it DataBlockIter
+	var rw DataBlockRewriter
+	rw.KeySchema = testKeysSchema
 	var sizes []int
 	datadriven.Walk(t, "testdata/data_block", func(t *testing.T, path string) {
 		datadriven.RunTest(t, path, func(t *testing.T, td *datadriven.TestData) string {
@@ -43,12 +45,14 @@ func TestDataBlock(t *testing.T) {
 				fmt.Fprint(&buf, &w)
 				sizes = sizes[:0]
 				return buf.String()
-			case "write":
+			case "write", "write-block":
+				// write-block does init/write/finish in a single command, and doesn't
+				// print anything.
+				if td.Cmd == "write-block" {
+					w.Init(testKeysSchema)
+				}
 				for _, line := range strings.Split(td.Input, "\n") {
-					isObsolete := strings.HasSuffix(line, "obsolete")
-					if isObsolete {
-						line = strings.TrimSuffix(line, "obsolete")
-					}
+					line, isObsolete := strings.CutSuffix(line, "obsolete")
 
 					j := strings.IndexRune(line, ':')
 					ik := base.ParseInternalKey(line[:j])
@@ -63,8 +67,29 @@ func TestDataBlock(t *testing.T) {
 					w.Add(ik, v, vp, kcmp, isObsolete)
 					sizes = append(sizes, w.Size())
 				}
+				if td.Cmd == "write-block" {
+					block, _ := w.Finish(w.Rows(), w.Size())
+					r.Init(testKeysSchema, block)
+					return ""
+				}
 				fmt.Fprint(&buf, &w)
 				return buf.String()
+			case "rewrite":
+				var from, to string
+				td.ScanArgs(t, "from", &from)
+				td.ScanArgs(t, "to", &to)
+				start, end, rewrittenBlock, err := rw.RewriteSuffixes(r.r.data, []byte(from), []byte(to))
+				if err != nil {
+					return fmt.Sprintf("error: %s", err)
+				}
+				r.Init(testKeysSchema, rewrittenBlock)
+				f := binfmt.New(r.r.data).LineWidth(20)
+				r.Describe(f)
+				fmt.Fprintf(&buf, "Start: %s\nEnd: %s\n%s",
+					start.Pretty(testkeys.Comparer.FormatKey),
+					end.Pretty(testkeys.Comparer.FormatKey),
+					f.String())
+				return buf.String()
 			case "finish":
 				rows := w.Rows()
 				td.MaybeScanArgs(t, "rows", &rows)
@@ -75,17 +100,27 @@ func TestDataBlock(t *testing.T) {
 				fmt.Fprintf(&buf, "LastKey: %s\n%s", lastKey.Pretty(testkeys.Comparer.FormatKey), f.String())
 				return buf.String()
 			case "iter":
-				var transforms block.IterTransforms
 				var seqNum uint64
+				var syntheticPrefix, syntheticSuffix string
 				td.MaybeScanArgs(t, "synthetic-seq-num", &seqNum)
-				transforms.SyntheticSeqNum = block.SyntheticSeqNum(seqNum)
+				td.MaybeScanArgs(t, "synthetic-prefix", &syntheticPrefix)
+				td.MaybeScanArgs(t, "synthetic-suffix", &syntheticSuffix)
+				transforms := block.IterTransforms{
+					SyntheticSeqNum:    block.SyntheticSeqNum(seqNum),
+					HideObsoletePoints: td.HasArg("hide-obsolete-points"),
+					SyntheticPrefix:    []byte(syntheticPrefix),
+					SyntheticSuffix:    []byte(syntheticSuffix),
+				}
 				it.Init(&r, testKeysSchema.NewKeySeeker(), getLazyValuer(func([]byte) base.LazyValue {
 					return base.LazyValue{ValueOrHandle: []byte("mock external value")}
 				}), transforms)
-				var o []itertest.IterOpt
+				o := []itertest.IterOpt{itertest.ShowCommands}
 				if td.HasArg("verbose") {
 					o = append(o, itertest.Verbose)
 				}
+				if td.HasArg("invalidated") {
+					it = DataBlockIter{}
+				}
 				return itertest.RunInternalIterCmd(t, td, &it, o...)
 			default:
 				return fmt.Sprintf("unknown command: %s", td.Cmd)
diff --git a/sstable/colblk/index_block_test.go b/sstable/colblk/index_block_test.go
index b7667cbb..da7ce070 100644
--- a/sstable/colblk/index_block_test.go
+++ b/sstable/colblk/index_block_test.go
@@ -9,9 +9,13 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/stretchr/testify/require"
 )
@@ -64,6 +68,11 @@ func TestIndexBlock(t *testing.T) {
 					valid = it.Next()
 				case "prev":
 					valid = it.Prev()
+				case "is-valid":
+					fmt.Fprintf(&buf, "Valid()=%t\n", it.Valid())
+					continue
+				case "invalidate":
+					it.Invalidate()
 				default:
 					panic(fmt.Sprintf("unknown command: %s", fields[0]))
 				}
@@ -88,3 +97,58 @@ func TestIndexBlock(t *testing.T) {
 		}
 	})
 }
+
+// TestIndexIterInitHandle exercises initializing an IndexIter through
+// InitHandle.
+func TestIndexIterInitHandle(t *testing.T) {
+	var w IndexBlockWriter
+	w.Init()
+	bh1 := block.Handle{Offset: 0, Length: 2000}
+	bh2 := block.Handle{Offset: 2008, Length: 1000}
+	w.AddBlockHandle([]byte("a"), bh1, nil)
+	w.AddBlockHandle([]byte("b"), bh2, nil)
+	b := w.Finish(w.Rows())
+
+	c := cache.New(10 << 10)
+	defer c.Unref()
+	v := block.Alloc(len(b), nil)
+	copy(v.Get(), b)
+	v.MakeHandle(c, cache.ID(1), base.DiskFileNum(1), 0).Release()
+
+	getBlockAndIterate := func(it *IndexIter) {
+		h := c.Get(cache.ID(1), base.DiskFileNum(1), 0)
+		require.NotNil(t, h.Get())
+		require.NoError(t, it.InitHandle(
+			testkeys.Comparer.Compare,
+			testkeys.Comparer.Split,
+			block.CacheBufferHandle(h),
+			block.NoTransforms))
+		defer it.Close()
+		require.True(t, it.First())
+		bh, err := it.BlockHandleWithProperties()
+		require.NoError(t, err)
+		require.Equal(t, bh1, bh.Handle)
+		require.True(t, it.Next())
+		bh, err = it.BlockHandleWithProperties()
+		require.NoError(t, err)
+		require.Equal(t, bh2, bh.Handle)
+		require.False(t, it.Next())
+		require.False(t, it.IsDataInvalidated())
+		it.Invalidate()
+		require.True(t, it.IsDataInvalidated())
+	}
+
+	const workers = 8
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for w := 0; w < workers; w++ {
+		go func() {
+			var iter IndexIter
+			defer wg.Done()
+			for i := 0; i < 10; i++ {
+				getBlockAndIterate(&iter)
+			}
+		}()
+	}
+	wg.Wait()
+}
diff --git a/sstable/colblk/testdata/bitmap b/sstable/colblk/testdata/bitmap
index f0979c9f..ebe7dff1 100644
--- a/sstable/colblk/testdata/bitmap
+++ b/sstable/colblk/testdata/bitmap
@@ -372,22 +372,60 @@ Binary representation:
 40-48: b 0000000000000000000000000000000000000000000000000000000010000000 # bitmap word 4
 48-56: b 0001000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
 
+# Write out one fewer row than is set, which should result in the all-zeroes
+# encoding.
+
+build rows=319
+0000000000000000000000000000000000000000000000000000000000000000
+0000000000000000000000000000000000000000000000000000000000000000
+0000000000000000000000000000000000000000000000000000000000000000
+0000000000000000000000000000000000000000000000000000000000000000
+0000000000000000000000000000000000000000000000000000000000000001
+----
+0000000000000000000000000000000000000000000000000000000000000000
+0000000000000000000000000000000000000000000000000000000000000000
+0000000000000000000000000000000000000000000000000000000000000000
+0000000000000000000000000000000000000000000000000000000000000000
+000000000000000000000000000000000000000000000000000000000000000
+Binary representation:
+0-1: x 01 # bitmap encoding
+
 # Write out fewer rows than we set. The bitmap summary should reflect the
 # truncated view of the bitmap.
 
-build rows=192
+build rows=260
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000111000000000000000000000000000000000000000000000000000
+0000000000111000000000000000000000000000000000000000000000000000
 ----
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000
 0000000000000000000000000000000000000000000000000000000000000000
+0000000000111000000000000000000000000000000000000000000000000000
+0000
 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 word 1
 24-32: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 2
-32-40: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+32-40: b 0000000000011100000000000000000000000000000000000000000000000000 # bitmap word 3
+40-48: b 0000000000000000000000000000000000000000000000000000000000000000 # bitmap word 4
+48-56: b 0000100000000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+
+# Test a case where we invert a bitmap that has more words than we'll need in
+# the finished serialized bitmap. Regression test for a bug that caused us to
+# pass a negative number into slices.Grow.
+
+build rows=64 invert
+0000000000000000000000000000000000000000000000000000000000000000
+1
+----
+1111111111111111111111111111111111111111111111111111111111111111
+Binary representation:
+00-01: x 00                                                               # bitmap encoding
+01-08: x 00000000000000                                                   # padding to align to 64-bit boundary
+08-16: b 1111111111111111111111111111111111111111111111111111111111111111 # bitmap word 0
+16-24: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
diff --git a/sstable/colblk/testdata/data_block/bundle_search b/sstable/colblk/testdata/data_block/bundle_search
index 61f863aa..433c1db9 100644
--- a/sstable/colblk/testdata/data_block/bundle_search
+++ b/sstable/colblk/testdata/data_block/bundle_search
@@ -528,11 +528,11 @@ seek-ge bac
 seek-ge bacl
 seek-ge bacu
 ----
-backache:v
-backslide:v
-backslide:v
-backhanded:v
-backhanded:v
-bacchanal:v
-bacteria:v
-.
+  seek-ge backache: backache:v
+ seek-ge backslide: backslide:v
+  seek-ge backslid: backslide:v
+seek-ge backhanded: backhanded:v
+ seek-ge backhande: backhanded:v
+       seek-ge bac: bacchanal:v
+      seek-ge bacl: bacteria:v
+      seek-ge bacu: .
diff --git a/sstable/colblk/testdata/data_block/external_value b/sstable/colblk/testdata/data_block/external_value
index 131586c7..e9ab8369 100644
--- a/sstable/colblk/testdata/data_block/external_value
+++ b/sstable/colblk/testdata/data_block/external_value
@@ -290,27 +290,27 @@ next
 next
 next
 ----
-blockprefix_apple@98:apple98
-blockprefix_apple@52:mock external value
-blockprefix_apple@23:mock external value
-blockprefix_apple@11:mock external value
-blockprefix_banana@94:banana94
-blockprefix_banana@93:
-blockprefix_banana@93:mock external value
-blockprefix_banana@72:mock external value
-blockprefix_banana@9:mock external value
-blockprefix_banana@1:mock external value
-blockprefix_coconut:coconut
-blockprefix_coconut@92:mock external value
-blockprefix_coconut@35:mock external value
-blockprefix_coconut@22:mock external value
-blockprefix_coconut@21:mock external value
-blockprefix_coconut@1:mock external value
-blockprefix_guava@99:mock external value
-blockprefix_kiwi@99:mock external value
-blockprefix_kiwi@98:mock external value
-blockprefix_lemon@92:
-.
+first: blockprefix_apple@98:apple98
+ next: blockprefix_apple@52:mock external value
+ next: blockprefix_apple@23:mock external value
+ next: blockprefix_apple@11:mock external value
+ next: blockprefix_banana@94:banana94
+ next: blockprefix_banana@93:
+ next: blockprefix_banana@93:mock external value
+ next: blockprefix_banana@72:mock external value
+ next: blockprefix_banana@9:mock external value
+ next: blockprefix_banana@1:mock external value
+ next: blockprefix_coconut:coconut
+ next: blockprefix_coconut@92:mock external value
+ next: blockprefix_coconut@35:mock external value
+ next: blockprefix_coconut@22:mock external value
+ next: blockprefix_coconut@21:mock external value
+ next: blockprefix_coconut@1:mock external value
+ next: blockprefix_guava@99:mock external value
+ next: blockprefix_kiwi@99:mock external value
+ next: blockprefix_kiwi@98:mock external value
+ next: blockprefix_lemon@92:
+ next: .
 
 # Scan across the block using next prefix.
 
@@ -323,17 +323,17 @@ next-prefix
 next-prefix
 next-prefix
 ----
-blockprefix_apple@98:apple98
-blockprefix_banana@94:banana94
-blockprefix_coconut:coconut
-blockprefix_guava@99:mock external value
-blockprefix_kiwi@99:mock external value
-blockprefix_lemon@92:
-.
+      first: blockprefix_apple@98:apple98
+next-prefix: blockprefix_banana@94:banana94
+next-prefix: blockprefix_coconut:coconut
+next-prefix: blockprefix_guava@99:mock external value
+next-prefix: blockprefix_kiwi@99:mock external value
+next-prefix: blockprefix_lemon@92:
+next-prefix: .
 
 iter
 seek-ge blockprefix_banana@73
 next-prefix
 ----
-blockprefix_banana@72:mock external value
-blockprefix_coconut:coconut
+seek-ge blockprefix_banana@73: blockprefix_banana@72:mock external value
+                  next-prefix: blockprefix_coconut:coconut
diff --git a/sstable/colblk/testdata/data_block/finish_without_final_row b/sstable/colblk/testdata/data_block/finish_without_final_row
index 26f068bf..3bc77ece 100644
--- a/sstable/colblk/testdata/data_block/finish_without_final_row
+++ b/sstable/colblk/testdata/data_block/finish_without_final_row
@@ -127,12 +127,12 @@ next
 next
 next
 ----
-a@10:apple
-b@5:banana
-b@2:blueberry
-c@9:coconut
-c@6:cantelope
-.
+first: a@10:apple
+ next: b@5:banana
+ next: b@2:blueberry
+ next: c@9:coconut
+ next: c@6:cantelope
+ next: .
 
 init
 ----
diff --git a/sstable/colblk/testdata/data_block/next_prefix b/sstable/colblk/testdata/data_block/next_prefix
index fd88c631..95c94bc4 100644
--- a/sstable/colblk/testdata/data_block/next_prefix
+++ b/sstable/colblk/testdata/data_block/next_prefix
@@ -252,17 +252,17 @@ next-prefix
 next-prefix
 next-prefix
 ----
-blockprefix_apple@98:apple98
-blockprefix_banana@94:banana94
-blockprefix_coconut:coconut
-blockprefix_guava@99:guava99
-blockprefix_kiwi@99:kiwi99
-blockprefix_lemon@92:
-.
+      first: blockprefix_apple@98:apple98
+next-prefix: blockprefix_banana@94:banana94
+next-prefix: blockprefix_coconut:coconut
+next-prefix: blockprefix_guava@99:guava99
+next-prefix: blockprefix_kiwi@99:kiwi99
+next-prefix: blockprefix_lemon@92:
+next-prefix: .
 
 iter
 seek-ge blockprefix_banana@73
 next-prefix
 ----
-blockprefix_banana@72:banana72
-blockprefix_coconut:coconut
+seek-ge blockprefix_banana@73: blockprefix_banana@72:banana72
+                  next-prefix: blockprefix_coconut:coconut
diff --git a/sstable/colblk/testdata/data_block/rewrite_suffixes b/sstable/colblk/testdata/data_block/rewrite_suffixes
new file mode 100644
index 00000000..7b27e1f6
--- /dev/null
+++ b/sstable/colblk/testdata/data_block/rewrite_suffixes
@@ -0,0 +1,278 @@
+init
+----
+size=51:
+0: prefixes:       prefixbytes(16): 0 keys
+1: suffixes:       bytes: 0 rows set; 0 bytes in data
+2: trailers:       uint: 0 rows
+3: prefix changed: bitmap
+4: values:         bytes: 0 rows set; 0 bytes in data
+5: is-value-ext:   bitmap
+6: is-obsolete:    bitmap
+
+write
+poi@12#0,SET:big
+yaya@5#0,SET:mini
+----
+size=110:
+0: prefixes:       prefixbytes(16): 2 keys
+1: suffixes:       bytes: 2 rows set; 5 bytes in data
+2: trailers:       uint: 2 rows
+3: prefix changed: bitmap
+4: values:         bytes: 2 rows set; 7 bytes in data
+5: is-value-ext:   bitmap
+6: is-obsolete:    bitmap
+
+finish
+----
+LastKey: yaya@5#0,SET
+# data block header
+000-004: x 06000000                                                         # maximum key length: 6
+# columnar block header
+004-005: x 01                                                               # version 1
+005-007: x 0700                                                             # 7 columns
+007-011: x 02000000                                                         # 2 rows
+011-012: b 00000100                                                         # col 0: prefixbytes
+012-016: x 2e000000                                                         # col 0: page start 46
+016-017: b 00000011                                                         # col 1: bytes
+017-021: x 3b000000                                                         # col 1: page start 59
+021-022: b 00000010                                                         # col 2: uint
+022-026: x 44000000                                                         # col 2: page start 68
+026-027: b 00000001                                                         # col 3: bool
+027-031: x 4d000000                                                         # col 3: page start 77
+031-032: b 00000011                                                         # col 4: bytes
+032-036: x 60000000                                                         # col 4: page start 96
+036-037: b 00000001                                                         # col 5: bool
+037-041: x 6b000000                                                         # col 5: page start 107
+041-042: b 00000001                                                         # col 6: bool
+042-046: x 6c000000                                                         # col 6: page start 108
+# data for column 0
+# PrefixBytes
+046-047: x 04                                                               # bundleSize: 16
+# Offsets table
+047-048: x 01                                                               # encoding: 1b
+048-049: x 00                                                               # data[0] = 0 [52 overall]
+049-050: x 00                                                               # data[1] = 0 [52 overall]
+050-051: x 03                                                               # data[2] = 3 [55 overall]
+051-052: x 07                                                               # data[3] = 7 [59 overall]
+# Data
+052-052: x                                                                  # data[00]:  (block prefix)
+052-052: x                                                                  # data[01]:  (bundle prefix)
+052-055: x 706f69                                                           # data[02]: poi
+055-059: x 79617961                                                         # data[03]: yaya
+# data for column 1
+# rawbytes
+# offsets table
+059-060: x 01                                                               # encoding: 1b
+060-061: x 00                                                               # data[0] = 0 [63 overall]
+061-062: x 03                                                               # data[1] = 3 [66 overall]
+062-063: x 05                                                               # data[2] = 5 [68 overall]
+# data
+063-066: x 403132                                                           # data[0]: @12
+066-068: x 4035                                                             # data[1]: @5
+# data for column 2
+068-069: x 80                                                               # encoding: const
+069-077: x 0100000000000000                                                 # 64-bit constant: 1
+# data for column 3
+077-078: x 00                                                               # bitmap encoding
+078-080: x 0000                                                             # padding to align to 64-bit boundary
+080-088: b 0000001100000000000000000000000000000000000000000000000000000000 # bitmap word 0
+088-096: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+# data for column 4
+# rawbytes
+# offsets table
+096-097: x 01                                                               # encoding: 1b
+097-098: x 00                                                               # data[0] = 0 [100 overall]
+098-099: x 03                                                               # data[1] = 3 [103 overall]
+099-100: x 07                                                               # data[2] = 7 [107 overall]
+# data
+100-103: x 626967                                                           # data[0]: big
+103-107: x 6d696e69                                                         # data[1]: mini
+# data for column 5
+107-108: x 01                                                               # bitmap encoding
+# data for column 6
+108-109: x 01                                                               # bitmap encoding
+109-110: x 00                                                               # block padding byte
+
+rewrite from=@12 to=@22
+----
+error: key yaya@5#0,SET has suffix 0x4035; require 0x403132
+
+init
+----
+size=51:
+0: prefixes:       prefixbytes(16): 0 keys
+1: suffixes:       bytes: 0 rows set; 0 bytes in data
+2: trailers:       uint: 0 rows
+3: prefix changed: bitmap
+4: values:         bytes: 0 rows set; 0 bytes in data
+5: is-value-ext:   bitmap
+6: is-obsolete:    bitmap
+
+write
+marble@6#0,SET:foo
+poi@6#0,SET:bar
+yaya@6#0,SET:bax
+----
+size=121:
+0: prefixes:       prefixbytes(16): 3 keys
+1: suffixes:       bytes: 3 rows set; 6 bytes in data
+2: trailers:       uint: 3 rows
+3: prefix changed: bitmap
+4: values:         bytes: 3 rows set; 9 bytes in data
+5: is-value-ext:   bitmap
+6: is-obsolete:    bitmap
+
+finish
+----
+LastKey: yaya@6#0,SET
+# data block header
+000-004: x 08000000                                                         # maximum key length: 8
+# columnar block header
+004-005: x 01                                                               # version 1
+005-007: x 0700                                                             # 7 columns
+007-011: x 03000000                                                         # 3 rows
+011-012: b 00000100                                                         # col 0: prefixbytes
+012-016: x 2e000000                                                         # col 0: page start 46
+016-017: b 00000011                                                         # col 1: bytes
+017-021: x 42000000                                                         # col 1: page start 66
+021-022: b 00000010                                                         # col 2: uint
+022-026: x 4d000000                                                         # col 2: page start 77
+026-027: b 00000001                                                         # col 3: bool
+027-031: x 56000000                                                         # col 3: page start 86
+031-032: b 00000011                                                         # col 4: bytes
+032-036: x 68000000                                                         # col 4: page start 104
+036-037: b 00000001                                                         # col 5: bool
+037-041: x 76000000                                                         # col 5: page start 118
+041-042: b 00000001                                                         # col 6: bool
+042-046: x 77000000                                                         # col 6: page start 119
+# data for column 0
+# PrefixBytes
+046-047: x 04                                                               # bundleSize: 16
+# Offsets table
+047-048: x 01                                                               # encoding: 1b
+048-049: x 00                                                               # data[0] = 0 [53 overall]
+049-050: x 00                                                               # data[1] = 0 [53 overall]
+050-051: x 06                                                               # data[2] = 6 [59 overall]
+051-052: x 09                                                               # data[3] = 9 [62 overall]
+052-053: x 0d                                                               # data[4] = 13 [66 overall]
+# Data
+053-053: x                                                                  # data[00]:  (block prefix)
+053-053: x                                                                  # data[01]:  (bundle prefix)
+053-059: x 6d6172626c65                                                     # data[02]: marble
+059-062: x 706f69                                                           # data[03]: poi
+062-066: x 79617961                                                         # data[04]: yaya
+# data for column 1
+# rawbytes
+# offsets table
+066-067: x 01                                                               # encoding: 1b
+067-068: x 00                                                               # data[0] = 0 [71 overall]
+068-069: x 02                                                               # data[1] = 2 [73 overall]
+069-070: x 04                                                               # data[2] = 4 [75 overall]
+070-071: x 06                                                               # data[3] = 6 [77 overall]
+# data
+071-073: x 4036                                                             # data[0]: @6
+073-075: x 4036                                                             # data[1]: @6
+075-077: x 4036                                                             # data[2]: @6
+# data for column 2
+077-078: x 80                                                               # encoding: const
+078-086: x 0100000000000000                                                 # 64-bit constant: 1
+# data for column 3
+086-087: x 00                                                               # bitmap encoding
+087-088: x 00                                                               # padding to align to 64-bit boundary
+088-096: b 0000011100000000000000000000000000000000000000000000000000000000 # bitmap word 0
+096-104: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+# data for column 4
+# rawbytes
+# offsets table
+104-105: x 01                                                               # encoding: 1b
+105-106: x 00                                                               # data[0] = 0 [109 overall]
+106-107: x 03                                                               # data[1] = 3 [112 overall]
+107-108: x 06                                                               # data[2] = 6 [115 overall]
+108-109: x 09                                                               # data[3] = 9 [118 overall]
+# data
+109-112: x 666f6f                                                           # data[0]: foo
+112-115: x 626172                                                           # data[1]: bar
+115-118: x 626178                                                           # data[2]: bax
+# data for column 5
+118-119: x 01                                                               # bitmap encoding
+# data for column 6
+119-120: x 01                                                               # bitmap encoding
+120-121: x 00                                                               # block padding byte
+
+rewrite from=@6 to=@54
+----
+Start: marble@54#0,SET
+End: yaya@54#0,SET
+# data block header
+000-004: x 09000000                                                         # maximum key length: 9
+# columnar block header
+004-005: x 01                                                               # version 1
+005-007: x 0700                                                             # 7 columns
+007-011: x 03000000                                                         # 3 rows
+011-012: b 00000100                                                         # col 0: prefixbytes
+012-016: x 2e000000                                                         # col 0: page start 46
+016-017: b 00000011                                                         # col 1: bytes
+017-021: x 42000000                                                         # col 1: page start 66
+021-022: b 00000010                                                         # col 2: uint
+022-026: x 50000000                                                         # col 2: page start 80
+026-027: b 00000001                                                         # col 3: bool
+027-031: x 59000000                                                         # col 3: page start 89
+031-032: b 00000011                                                         # col 4: bytes
+032-036: x 70000000                                                         # col 4: page start 112
+036-037: b 00000001                                                         # col 5: bool
+037-041: x 7e000000                                                         # col 5: page start 126
+041-042: b 00000001                                                         # col 6: bool
+042-046: x 7f000000                                                         # col 6: page start 127
+# data for column 0
+# PrefixBytes
+046-047: x 04                                                               # bundleSize: 16
+# Offsets table
+047-048: x 01                                                               # encoding: 1b
+048-049: x 00                                                               # data[0] = 0 [53 overall]
+049-050: x 00                                                               # data[1] = 0 [53 overall]
+050-051: x 06                                                               # data[2] = 6 [59 overall]
+051-052: x 09                                                               # data[3] = 9 [62 overall]
+052-053: x 0d                                                               # data[4] = 13 [66 overall]
+# Data
+053-053: x                                                                  # data[00]:  (block prefix)
+053-053: x                                                                  # data[01]:  (bundle prefix)
+053-059: x 6d6172626c65                                                     # data[02]: marble
+059-062: x 706f69                                                           # data[03]: poi
+062-066: x 79617961                                                         # data[04]: yaya
+# data for column 1
+# rawbytes
+# offsets table
+066-067: x 01                                                               # encoding: 1b
+067-068: x 00                                                               # data[0] = 0 [71 overall]
+068-069: x 03                                                               # data[1] = 3 [74 overall]
+069-070: x 06                                                               # data[2] = 6 [77 overall]
+070-071: x 09                                                               # data[3] = 9 [80 overall]
+# data
+071-074: x 403534                                                           # data[0]: @54
+074-077: x 403534                                                           # data[1]: @54
+077-080: x 403534                                                           # data[2]: @54
+# data for column 2
+080-081: x 80                                                               # encoding: const
+081-089: x 0100000000000000                                                 # 64-bit constant: 1
+# data for column 3
+089-090: x 00                                                               # bitmap encoding
+090-096: x 000000000000                                                     # padding to align to 64-bit boundary
+096-104: b 0000011100000000000000000000000000000000000000000000000000000000 # bitmap word 0
+104-112: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
+# data for column 4
+# rawbytes
+# offsets table
+112-113: x 01                                                               # encoding: 1b
+113-114: x 00                                                               # data[0] = 0 [117 overall]
+114-115: x 03                                                               # data[1] = 3 [120 overall]
+115-116: x 06                                                               # data[2] = 6 [123 overall]
+116-117: x 09                                                               # data[3] = 9 [126 overall]
+# data
+117-120: x 666f6f                                                           # data[0]: foo
+120-123: x 626172                                                           # data[1]: bar
+123-126: x 626178                                                           # data[2]: bax
+# data for column 5
+126-127: x 01                                                               # bitmap encoding
+# data for column 6
+127-128: x 01                                                               # bitmap encoding
+128-129: x 00                                                               # block padding byte
diff --git a/sstable/colblk/testdata/data_block/simple b/sstable/colblk/testdata/data_block/simple
index db348622..91472ab0 100644
--- a/sstable/colblk/testdata/data_block/simple
+++ b/sstable/colblk/testdata/data_block/simple
@@ -158,14 +158,14 @@ next
 next
 next
 ----
-a@10:apple
-b@5:banana
-b@2:blueberry
-c@9:coconut
-c@6:cantelope
-c@1:clementine
-d@11: 
-.
+first: a@10:apple
+ next: b@5:banana
+ next: b@2:blueberry
+ next: c@9:coconut
+ next: c@6:cantelope
+ next: c@1:clementine
+ next: d@11: 
+ next: .
 
 iter
 last
@@ -177,14 +177,14 @@ prev
 prev
 prev
 ----
-d@11: 
-c@1:clementine
-c@6:cantelope
-c@9:coconut
-b@2:blueberry
-b@5:banana
-a@10:apple
-.
+last: d@11: 
+prev: c@1:clementine
+prev: c@6:cantelope
+prev: c@9:coconut
+prev: b@2:blueberry
+prev: b@5:banana
+prev: a@10:apple
+prev: .
 
 iter
 seek-ge a
@@ -200,18 +200,18 @@ seek-ge b@2
 seek-ge b@1
 seek-ge c
 ----
-a@10:apple
-a@10:apple
-b@5:banana
-b@5:banana
-b@5:banana
-b@5:banana
-b@5:banana
-b@2:blueberry
-b@2:blueberry
-b@2:blueberry
-c@9:coconut
-c@9:coconut
+       seek-ge a: a@10:apple
+    seek-ge a@10: a@10:apple
+     seek-ge a@3: b@5:banana
+     seek-ge a@1: b@5:banana
+       seek-ge b: b@5:banana
+seek-ge b@999999: b@5:banana
+     seek-ge b@5: b@5:banana
+     seek-ge b@4: b@2:blueberry
+     seek-ge b@3: b@2:blueberry
+     seek-ge b@2: b@2:blueberry
+     seek-ge b@1: c@9:coconut
+       seek-ge c: c@9:coconut
 
 iter
 seek-ge c@10
@@ -229,20 +229,20 @@ seek-ge d
 seek-ge d@11
 seek-ge d@10
 ----
-c@9:coconut
-c@9:coconut
-c@6:cantelope
-c@6:cantelope
-c@6:cantelope
-c@1:clementine
-c@1:clementine
-c@1:clementine
-c@1:clementine
-c@1:clementine
-d@11: 
-d@11: 
-d@11: 
-.
+seek-ge c@10: c@9:coconut
+ seek-ge c@9: c@9:coconut
+ seek-ge c@8: c@6:cantelope
+ seek-ge c@7: c@6:cantelope
+ seek-ge c@6: c@6:cantelope
+ seek-ge c@5: c@1:clementine
+ seek-ge c@4: c@1:clementine
+ seek-ge c@3: c@1:clementine
+ seek-ge c@2: c@1:clementine
+ seek-ge c@1: c@1:clementine
+ seek-ge c@0: d@11: 
+   seek-ge d: d@11: 
+seek-ge d@11: d@11: 
+seek-ge d@10: .
 
 iter
 seek-lt a
@@ -258,18 +258,18 @@ seek-lt b@2
 seek-lt b@1
 seek-lt c
 ----
-.
-.
-a@10:apple
-a@10:apple
-a@10:apple
-a@10:apple
-a@10:apple
-b@5:banana
-b@5:banana
-b@5:banana
-b@2:blueberry
-b@2:blueberry
+       seek-lt a: .
+    seek-lt a@10: .
+     seek-lt a@3: a@10:apple
+     seek-lt a@1: a@10:apple
+       seek-lt b: a@10:apple
+seek-lt b@999999: a@10:apple
+     seek-lt b@5: a@10:apple
+     seek-lt b@4: b@5:banana
+     seek-lt b@3: b@5:banana
+     seek-lt b@2: b@5:banana
+     seek-lt b@1: b@2:blueberry
+       seek-lt c: b@2:blueberry
 
 iter
 seek-lt c@10
@@ -287,20 +287,35 @@ seek-lt d
 seek-lt d@11
 seek-lt d@10
 ----
-b@2:blueberry
-b@2:blueberry
-c@9:coconut
-c@9:coconut
-c@9:coconut
-c@6:cantelope
-c@6:cantelope
-c@6:cantelope
-c@6:cantelope
-c@6:cantelope
-c@1:clementine
-c@1:clementine
-c@1:clementine
-d@11: 
+seek-lt c@10: b@2:blueberry
+ seek-lt c@9: b@2:blueberry
+ seek-lt c@8: c@9:coconut
+ seek-lt c@7: c@9:coconut
+ seek-lt c@6: c@9:coconut
+ seek-lt c@5: c@6:cantelope
+ seek-lt c@4: c@6:cantelope
+ seek-lt c@3: c@6:cantelope
+ seek-lt c@2: c@6:cantelope
+ seek-lt c@1: c@6:cantelope
+ seek-lt c@0: c@1:clementine
+   seek-lt d: c@1:clementine
+seek-lt d@11: c@1:clementine
+seek-lt d@10: d@11: 
+
+iter invalidated
+first
+seek-ge foo
+next
+prev
+seek-lt foo
+last
+----
+      first: .
+seek-ge foo: .
+       next: .
+       prev: .
+seek-lt foo: .
+       last: .
 
 init
 ----
@@ -513,13 +528,13 @@ seek-ge aaaaaaaaaaaaaaapparh
 seek-ge aaaaaaaaaaaaaaarresting@10
 seek-ge aaaaaaaaaaaaaaarrived@10
 ----
-aaaaaaaaaaaaaaappalling@10:a
-aaaaaaaaaaaaaaappalling@10:a
-aaaaaaaaaaaaaaappalling@10:a
-aaaaaaaaaaaaaaapparel@10:a
-aaaaaaaaaaaaaaapparition@10:a
-aaaaaaaaaaaaaaarresting@10:a
-aaaaaaaaaaaaaaarrived@10:a
+                       seek-ge aaa: aaaaaaaaaaaaaaappalling@10:a
+                 seek-ge aaaaaaaaa: aaaaaaaaaaaaaaappalling@10:a
+         seek-ge aaaaaaaaaaaaaaapp: aaaaaaaaaaaaaaappalling@10:a
+  seek-ge aaaaaaaaaaaaaaapparel@10: aaaaaaaaaaaaaaapparel@10:a
+      seek-ge aaaaaaaaaaaaaaapparh: aaaaaaaaaaaaaaapparition@10:a
+seek-ge aaaaaaaaaaaaaaarresting@10: aaaaaaaaaaaaaaarresting@10:a
+  seek-ge aaaaaaaaaaaaaaarrived@10: aaaaaaaaaaaaaaarrived@10:a
 
 iter
 seek-lt aaa
@@ -530,10 +545,10 @@ seek-lt aaaaaaaaaaaaaaapparh
 seek-lt aaaaaaaaaaaaaaarresting@10
 seek-lt aaaaaaaaaaaaaaarrived@10
 ----
-.
-.
-.
-aaaaaaaaaaaaaaappalling@10:a
-aaaaaaaaaaaaaaapparel@10:a
-aaaaaaaaaaaaaaapproves@10:a
-aaaaaaaaaaaaaaarresting@10:a
+                       seek-lt aaa: .
+                 seek-lt aaaaaaaaa: .
+         seek-lt aaaaaaaaaaaaaaapp: .
+  seek-lt aaaaaaaaaaaaaaapparel@10: aaaaaaaaaaaaaaappalling@10:a
+      seek-lt aaaaaaaaaaaaaaapparh: aaaaaaaaaaaaaaapparel@10:a
+seek-lt aaaaaaaaaaaaaaarresting@10: aaaaaaaaaaaaaaapproves@10:a
+  seek-lt aaaaaaaaaaaaaaarrived@10: aaaaaaaaaaaaaaarresting@10:a
diff --git a/sstable/colblk/testdata/data_block/transforms b/sstable/colblk/testdata/data_block/transforms
index eb537342..fce33f9a 100644
--- a/sstable/colblk/testdata/data_block/transforms
+++ b/sstable/colblk/testdata/data_block/transforms
@@ -1,15 +1,4 @@
-init
-----
-size=51:
-0: prefixes:       prefixbytes(16): 0 keys
-1: suffixes:       bytes: 0 rows set; 0 bytes in data
-2: trailers:       uint: 0 rows
-3: prefix changed: bitmap
-4: values:         bytes: 0 rows set; 0 bytes in data
-5: is-value-ext:   bitmap
-6: is-obsolete:    bitmap
-
-write
+write-block
 a@10#1,SET:apple
 b@5#2,SET:banana
 b@2#3,SETWITHDEL:blueberry
@@ -17,115 +6,6 @@ c@9#4,SETWITHDEL:coconut
 c@6#5,SET:cantaloupe
 c@1#6,SET:clementine
 ----
-size=170:
-0: prefixes:       prefixbytes(16): 6 keys
-1: suffixes:       bytes: 6 rows set; 13 bytes in data
-2: trailers:       uint: 6 rows
-3: prefix changed: bitmap
-4: values:         bytes: 6 rows set; 47 bytes in data
-5: is-value-ext:   bitmap
-6: is-obsolete:    bitmap
-
-finish
-----
-LastKey: c@1#6,SET
-# data block header
-000-004: x 04000000                                                         # maximum key length: 4
-# columnar block header
-004-005: x 01                                                               # version 1
-005-007: x 0700                                                             # 7 columns
-007-011: x 06000000                                                         # 6 rows
-011-012: b 00000100                                                         # col 0: prefixbytes
-012-016: x 2e000000                                                         # col 0: page start 46
-016-017: b 00000011                                                         # col 1: bytes
-017-021: x 3b000000                                                         # col 1: page start 59
-021-022: b 00000010                                                         # col 2: uint
-022-026: x 50000000                                                         # col 2: page start 80
-026-027: b 00000001                                                         # col 3: bool
-027-031: x 5e000000                                                         # col 3: page start 94
-031-032: b 00000011                                                         # col 4: bytes
-032-036: x 70000000                                                         # col 4: page start 112
-036-037: b 00000001                                                         # col 5: bool
-037-041: x a7000000                                                         # col 5: page start 167
-041-042: b 00000001                                                         # col 6: bool
-042-046: x a8000000                                                         # col 6: page start 168
-# data for column 0
-# PrefixBytes
-046-047: x 04                                                               # bundleSize: 16
-# Offsets table
-047-048: x 01                                                               # encoding: 1b
-048-049: x 00                                                               # data[0] = 0 [56 overall]
-049-050: x 00                                                               # data[1] = 0 [56 overall]
-050-051: x 01                                                               # data[2] = 1 [57 overall]
-051-052: x 02                                                               # data[3] = 2 [58 overall]
-052-053: x 02                                                               # data[4] = 2 [58 overall]
-053-054: x 03                                                               # data[5] = 3 [59 overall]
-054-055: x 03                                                               # data[6] = 3 [59 overall]
-055-056: x 03                                                               # data[7] = 3 [59 overall]
-# Data
-056-056: x                                                                  # data[00]:  (block prefix)
-056-056: x                                                                  # data[01]:  (bundle prefix)
-056-057: x 61                                                               # data[02]: a
-057-058: x 62                                                               # data[03]: b
-058-058: x                                                                  # data[04]: .
-058-059: x 63                                                               # data[05]: c
-059-059: x                                                                  # data[06]: .
-059-059: x                                                                  # data[07]: .
-# data for column 1
-# rawbytes
-# offsets table
-059-060: x 01                                                               # encoding: 1b
-060-061: x 00                                                               # data[0] = 0 [67 overall]
-061-062: x 03                                                               # data[1] = 3 [70 overall]
-062-063: x 05                                                               # data[2] = 5 [72 overall]
-063-064: x 07                                                               # data[3] = 7 [74 overall]
-064-065: x 09                                                               # data[4] = 9 [76 overall]
-065-066: x 0b                                                               # data[5] = 11 [78 overall]
-066-067: x 0d                                                               # data[6] = 13 [80 overall]
-# data
-067-070: x 403130                                                           # data[0]: @10
-070-072: x 4035                                                             # data[1]: @5
-072-074: x 4032                                                             # data[2]: @2
-074-076: x 4039                                                             # data[3]: @9
-076-078: x 4036                                                             # data[4]: @6
-078-080: x 4031                                                             # data[5]: @1
-# data for column 2
-080-081: x 02                                                               # encoding: 2b
-081-082: x 00                                                               # padding (aligning to 16-bit boundary)
-082-084: x 0101                                                             # data[0] = 257
-084-086: x 0102                                                             # data[1] = 513
-086-088: x 1203                                                             # data[2] = 786
-088-090: x 1204                                                             # data[3] = 1042
-090-092: x 0105                                                             # data[4] = 1281
-092-094: x 0106                                                             # data[5] = 1537
-# data for column 3
-094-095: x 00                                                               # bitmap encoding
-095-096: x 00                                                               # padding to align to 64-bit boundary
-096-104: b 0000101100000000000000000000000000000000000000000000000000000000 # bitmap word 0
-104-112: b 0000000100000000000000000000000000000000000000000000000000000000 # bitmap summary word 0-63
-# data for column 4
-# rawbytes
-# offsets table
-112-113: x 01                                                               # encoding: 1b
-113-114: x 00                                                               # data[0] = 0 [120 overall]
-114-115: x 05                                                               # data[1] = 5 [125 overall]
-115-116: x 0b                                                               # data[2] = 11 [131 overall]
-116-117: x 14                                                               # data[3] = 20 [140 overall]
-117-118: x 1b                                                               # data[4] = 27 [147 overall]
-118-119: x 25                                                               # data[5] = 37 [157 overall]
-119-120: x 2f                                                               # data[6] = 47 [167 overall]
-# data
-120-125: x 6170706c65                                                       # data[0]: apple
-125-131: x 62616e616e61                                                     # data[1]: banana
-131-140: x 626c75656265727279                                               # data[2]: blueberry
-140-147: x 636f636f6e7574                                                   # data[3]: coconut
-147-157: x 63616e74616c6f757065                                             # data[4]: cantaloupe
-157-167: x 636c656d656e74696e65                                             # data[5]: clementine
-# data for column 5
-167-168: x 01                                                               # bitmap encoding
-# data for column 6
-168-169: x 01                                                               # bitmap encoding
-169-170: x 00                                                               # block padding byte
 
 iter verbose synthetic-seq-num=1234
 first
@@ -141,15 +21,163 @@ prev
 prev
 next
 ----
-a@10#1234,SET:apple
-b@5#1234,SET:banana
-b@2#1234,SETWITHDEL:blueberry
-c@9#1234,SETWITHDEL:coconut
-c@6#1234,SET:cantaloupe
-c@9#1234,SETWITHDEL:coconut
-b@2#1234,SETWITHDEL:blueberry
-b@5#1234,SET:banana
-b@2#1234,SETWITHDEL:blueberry
-b@5#1234,SET:banana
-a@10#1234,SET:apple
-b@5#1234,SET:banana
+    first: a@10#1234,SET:apple
+     next: b@5#1234,SET:banana
+     next: b@2#1234,SETWITHDEL:blueberry
+seek-ge c: c@9#1234,SETWITHDEL:coconut
+     next: c@6#1234,SET:cantaloupe
+     prev: c@9#1234,SETWITHDEL:coconut
+     prev: b@2#1234,SETWITHDEL:blueberry
+     prev: b@5#1234,SET:banana
+seek-lt c: b@2#1234,SETWITHDEL:blueberry
+     prev: b@5#1234,SET:banana
+     prev: a@10#1234,SET:apple
+     next: b@5#1234,SET:banana
+
+write-block
+a@10#1,SET:apple obsolete
+b@5#3,SET:banana
+b@5#2,SET:banana-old obsolete
+b@2#3,SET:blueberry-old obsolete
+c@9#4,SET:coconut
+c@6#5,SET:cantaloupe-old obsolete
+c@1#7,SET:clementine
+c@1#6,SET:clementine-old obsolete
+----
+
+iter hide-obsolete-points
+first
+next
+next
+next
+----
+first: b@5:banana
+ next: c@9:coconut
+ next: c@1:clementine
+ next: .
+
+iter hide-obsolete-points
+first
+next
+next
+next
+prev
+prev
+next
+prev
+prev
+prev
+----
+first: b@5:banana
+ next: c@9:coconut
+ next: c@1:clementine
+ next: .
+ prev: c@1:clementine
+ prev: c@9:coconut
+ next: c@1:clementine
+ prev: c@9:coconut
+ prev: b@5:banana
+ prev: .
+
+
+iter hide-obsolete-points
+last
+prev
+next
+prev
+prev
+prev
+next
+----
+last: c@1:clementine
+prev: c@9:coconut
+next: c@1:clementine
+prev: c@9:coconut
+prev: b@5:banana
+prev: .
+next: b@5:banana
+
+iter hide-obsolete-points
+seek-ge a
+prev
+next
+----
+seek-ge a: b@5:banana
+     prev: .
+     next: b@5:banana
+
+iter hide-obsolete-points
+seek-ge d
+prev
+prev
+----
+seek-ge d: .
+     prev: c@1:clementine
+     prev: c@9:coconut
+
+iter hide-obsolete-points
+seek-lt c
+next
+next
+prev
+prev
+prev
+----
+seek-lt c: b@5:banana
+     next: c@9:coconut
+     next: c@1:clementine
+     prev: c@9:coconut
+     prev: b@5:banana
+     prev: .
+
+# Test a block with only obsolete points.
+write-block
+a@1#1,SET:a obsolete
+b@1#1,SET:b obsolete
+c@1#1,SET:c obsolete
+----
+
+iter hide-obsolete-points
+first
+next
+prev
+last
+next
+prev
+seek-ge a
+next
+prev
+seek-ge b
+next
+prev
+seek-ge c
+next
+prev
+seek-lt z
+next
+prev
+seek-lt b
+next
+prev
+----
+    first: .
+     next: .
+     prev: .
+     last: .
+     next: .
+     prev: .
+seek-ge a: .
+     next: .
+     prev: .
+seek-ge b: .
+     next: .
+     prev: .
+seek-ge c: .
+     next: .
+     prev: .
+seek-lt z: .
+     next: .
+     prev: .
+seek-lt b: .
+     next: .
+     prev: .
diff --git a/sstable/colblk/testdata/index_block b/sstable/colblk/testdata/index_block
index c54f9fa3..7428f75e 100644
--- a/sstable/colblk/testdata/index_block
+++ b/sstable/colblk/testdata/index_block
@@ -125,6 +125,60 @@ block 1: 141-253 props="bp2"
 block 0: 24-48 props="bp1"
 .
 
+# Test Next-ing an index iterator that's already exhausted in the forward
+# direction. The sstable iterators do this in some circumstances. The Next
+# should be a no-op and a subsequent Prev should still return to the last block
+# handle in the block.
+
+iter
+last
+next
+next
+prev
+----
+block 5: 963-1289 props="bp6"
+.
+.
+block 5: 963-1289 props="bp6"
+
+# Test Prev-ing an index iterator that's already exhausted in the reverse
+# direction. The sstable iterators do this in some circumstances. The Prev
+# should be a no-op and a subsequent Next should still return to the first block
+# handle in the block.
+
+iter
+first
+prev
+prev
+next
+----
+block 0: 24-48 props="bp1"
+.
+.
+block 0: 24-48 props="bp1"
+
+# Test Invalidate() and ensure that Valid() and all positioning methods return
+# false after it.
+
+iter
+first
+invalidate
+is-valid
+next
+prev
+first
+last
+seek-ge a
+----
+block 0: 24-48 props="bp1"
+.
+Valid()=false
+.
+.
+.
+.
+.
+
 # Rebuild the same index block, but excluding the last row during the final Finish().
 
 build rows=5
diff --git a/sstable/data_test.go b/sstable/data_test.go
index 24d6442d..16524178 100644
--- a/sstable/data_test.go
+++ b/sstable/data_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/cockroachdb/pebble/vfs"
 )
 
@@ -54,17 +55,39 @@ func optsFromArgs(td *datadriven.TestData, writerOpts *WriterOptions) error {
 			}
 		case "filter":
 			writerOpts.FilterPolicy = bloom.FilterPolicy(10)
-		case "comparer-split-4b-suffix":
-			writerOpts.Comparer = test4bSuffixComparer
+		case "comparer":
+			var err error
+			if writerOpts.Comparer, err = comparerFromCmdArg(arg); err != nil {
+				return err
+			}
 		case "writing-to-lowest-level":
 			writerOpts.WritingToLowestLevel = true
 		case "is-strict-obsolete":
 			writerOpts.IsStrictObsolete = true
 		}
 	}
+	if writerOpts.Comparer == nil {
+		writerOpts.Comparer = testkeys.Comparer
+	}
+	if len(writerOpts.KeySchema.ColumnTypes) == 0 {
+		writerOpts.KeySchema = colblk.DefaultKeySchema(writerOpts.Comparer, 16)
+	}
 	return nil
 }
 
+func comparerFromCmdArg(arg datadriven.CmdArg) (*Comparer, error) {
+	switch arg.Vals[0] {
+	case "split-4b-suffix":
+		return test4bSuffixComparer, nil
+	case "testkeys":
+		return testkeys.Comparer, nil
+	case "default":
+		return base.DefaultComparer, nil
+	default:
+		return nil, errors.Errorf("unknown comparer: %s", arg.Vals[0])
+	}
+}
+
 func runBuildMemObjCmd(
 	td *datadriven.TestData, writerOpts *WriterOptions,
 ) (*WriterMetadata, *objstorage.MemObj, error) {
@@ -139,7 +162,10 @@ func runBuildCmd(
 }
 
 func openReader(obj *objstorage.MemObj, writerOpts *WriterOptions, cacheSize int) (*Reader, error) {
-	readerOpts := ReaderOptions{Comparer: writerOpts.Comparer}
+	readerOpts := ReaderOptions{
+		Comparer:  writerOpts.Comparer,
+		KeySchema: writerOpts.KeySchema,
+	}
 	if writerOpts.FilterPolicy != nil {
 		readerOpts.Filters = map[string]FilterPolicy{
 			writerOpts.FilterPolicy.Name(): writerOpts.FilterPolicy,
@@ -149,9 +175,7 @@ func openReader(obj *objstorage.MemObj, writerOpts *WriterOptions, cacheSize int
 		c := cache.New(int64(cacheSize))
 		defer c.Unref()
 		readerOpts.SetInternal(sstableinternal.ReaderOptions{
-			CacheOpts: sstableinternal.CacheOptions{
-				Cache: c,
-			},
+			CacheOpts: sstableinternal.CacheOptions{Cache: c},
 		})
 	}
 	r, err := NewMemReader(obj.Data(), readerOpts)
@@ -164,6 +188,10 @@ func openReader(obj *objstorage.MemObj, writerOpts *WriterOptions, cacheSize int
 func runBuildRawCmd(
 	td *datadriven.TestData, opts *WriterOptions,
 ) (*WriterMetadata, *Reader, error) {
+	if err := optsFromArgs(td, opts); err != nil {
+		return nil, nil, err
+	}
+
 	mem := vfs.NewMem()
 	provider, err := objstorageprovider.Open(objstorageprovider.DefaultSettings(mem, "" /* dirName */))
 	if err != nil {
@@ -211,7 +239,10 @@ func runBuildRawCmd(
 	if err != nil {
 		return nil, nil, err
 	}
-	r, err := NewReader(context.Background(), f1, ReaderOptions{})
+	r, err := NewReader(context.Background(), f1, ReaderOptions{
+		Comparer:  opts.Comparer,
+		KeySchema: opts.KeySchema,
+	})
 	if err != nil {
 		return nil, nil, err
 	}
@@ -470,7 +501,10 @@ func runRewriteCmd(
 	if err != nil {
 		return nil, r, errors.Wrap(err, "rewrite failed")
 	}
-	readerOpts := ReaderOptions{Comparer: opts.Comparer}
+	readerOpts := ReaderOptions{
+		Comparer:  opts.Comparer,
+		KeySchema: opts.KeySchema,
+	}
 	if opts.FilterPolicy != nil {
 		readerOpts.Filters = map[string]FilterPolicy{
 			opts.FilterPolicy.Name(): opts.FilterPolicy,
diff --git a/sstable/random_test.go b/sstable/random_test.go
index 12bac379..c3e51a42 100644
--- a/sstable/random_test.go
+++ b/sstable/random_test.go
@@ -274,8 +274,9 @@ type randomTableConfig struct {
 
 func (cfg *randomTableConfig) readerOpts() ReaderOptions {
 	rOpts := ReaderOptions{
-		Comparer: testkeys.Comparer,
-		Filters:  map[string]FilterPolicy{},
+		Comparer:  testkeys.Comparer,
+		KeySchema: cfg.wopts.KeySchema,
+		Filters:   map[string]FilterPolicy{},
 	}
 	if cfg.wopts.FilterPolicy != nil {
 		rOpts.Filters[cfg.wopts.FilterPolicy.Name()] = cfg.wopts.FilterPolicy
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index 71fe6a4b..94c1732f 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -32,7 +32,6 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable/block"
-	"github.com/cockroachdb/pebble/sstable/rowblk"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/stretchr/testify/require"
@@ -625,28 +624,30 @@ func indexLayoutString(t *testing.T, r *Reader) string {
 	var buf strings.Builder
 	twoLevelIndex := r.Properties.IndexType == twoLevelIndex
 	buf.WriteString("index entries:\n")
-	iter, err := rowblk.NewIter(r.Compare, r.Split, indexH.Get(), NoTransforms)
+	iter := r.tableFormat.newIndexIter()
+	require.NoError(t, iter.Init(r.Compare, r.Split, indexH.Get(), NoTransforms))
 	defer func() {
 		require.NoError(t, iter.Close())
 	}()
 	require.NoError(t, err)
-	for kv := iter.First(); kv != nil; kv = iter.Next() {
-		bh, err := block.DecodeHandleWithProperties(kv.InPlaceValue())
+	for valid := iter.First(); valid; valid = iter.Next() {
+		bh, err := iter.BlockHandleWithProperties()
 		require.NoError(t, err)
-		fmt.Fprintf(&buf, " %s: size %d\n", string(kv.K.UserKey), bh.Length)
+		fmt.Fprintf(&buf, " %s: size %d\n", string(iter.Separator()), bh.Length)
 		if twoLevelIndex {
 			b, err := r.readBlock(context.Background(), bh.Handle, nil, nil, nil, nil, nil)
 			require.NoError(t, err)
 			defer b.Release()
-			iter2, err := rowblk.NewIter(r.Compare, r.Split, b.Get(), NoTransforms)
+			iter2 := r.tableFormat.newIndexIter()
+			require.NoError(t, iter2.Init(r.Compare, r.Split, b.Get(), NoTransforms))
 			defer func() {
 				require.NoError(t, iter2.Close())
 			}()
 			require.NoError(t, err)
-			for kv := iter2.First(); kv != nil; kv = iter2.Next() {
-				bh, err := block.DecodeHandleWithProperties(kv.InPlaceValue())
+			for valid := iter2.First(); valid; valid = iter2.Next() {
+				bh, err := iter2.BlockHandleWithProperties()
 				require.NoError(t, err)
-				fmt.Fprintf(&buf, "   %s: size %d\n", string(kv.K.UserKey), bh.Length)
+				fmt.Fprintf(&buf, "   %s: size %d\n", string(iter2.Separator()), bh.Length)
 			}
 		}
 	}
diff --git a/sstable/testdata/block_properties b/sstable/testdata/block_properties
index 7173df8b..1a139cc3 100644
--- a/sstable/testdata/block_properties
+++ b/sstable/testdata/block_properties
@@ -31,7 +31,7 @@ table-props
 # is printed, along with the properties for the block, keyed by the shortID.
 block-props
 ----
-c#3,SET:
+c:
   0: [1, 4)
 
 # Multiple collectors.
@@ -56,7 +56,7 @@ table-props
 
 block-props
 ----
-c#3,SET:
+c:
   0: [1, 4)
   1: [7, 10)
 
@@ -88,16 +88,16 @@ table-props
 
 block-props
 ----
-b#2,SET:
+b:
   0: [1, 9)
   1: [5, 7)
-d#4,SET:
+d:
   0: [2, 8)
   1: [1, 3)
-f#6,SET:
+f:
   0: [4, 6)
   1: [4, 8)
-h#8,SET:
+h:
   0: [3, 7)
   1: [3, 9)
 
@@ -122,7 +122,7 @@ collectors
 
 block-props
 ----
-d#inf,SEPARATOR:
+d:
   0: [5, 16)
 
 table-props
@@ -149,7 +149,7 @@ collectors
 
 block-props
 ----
-d#inf,SEPARATOR:
+d:
   0: [5, 16)
 
 table-props
@@ -176,7 +176,7 @@ collectors
 
 block-props
 ----
-d#inf,SEPARATOR:
+d:
 
 table-props
 ----
@@ -205,11 +205,11 @@ collectors
 
 block-props
 ----
-a@5#1,SET:
+a@5:
   0: [5, 6)
-b@10#2,SET:
+b@10:
   0: [10, 11)
-d#inf,SEPARATOR:
+d:
   0: [15, 16)
 
 table-props
@@ -307,29 +307,29 @@ table-props
 
 block-props
 ----
-a@1#1,SET:
+a@1:
   0: [1, 2)
-  a@1#1,SET:
+  a@1:
     0: [1, 2)
-b@10#2,SET:
+b@10:
   0: [10, 11)
-  b@10#2,SET:
+  b@10:
     0: [10, 11)
-c@15#3,SET:
+c@15:
   0: [15, 16)
-  c@15#3,SET:
+  c@15:
     0: [15, 16)
-d@25#4,SET:
+d@25:
   0: [25, 26)
-  d@25#4,SET:
+  d@25:
     0: [25, 26)
-e@3#5,SET:
+e@3:
   0: [3, 4)
-  e@3#5,SET:
+  e@3:
     0: [3, 4)
-g#inf,SEPARATOR:
+g:
   0: [5, 6)
-  g#inf,SEPARATOR:
+  g:
     0: [5, 6)
 
 # Test the same sstable, but with a larger index block size that fits multiple
@@ -353,21 +353,21 @@ collectors
 
 block-props
 ----
-c@15#3,SET:
+c@15:
   0: [1, 16)
-  a@1#1,SET:
+  a@1:
     0: [1, 2)
-  b@10#2,SET:
+  b@10:
     0: [10, 11)
-  c@15#3,SET:
+  c@15:
     0: [15, 16)
-g#inf,SEPARATOR:
+g:
   0: [3, 26)
-  d@25#4,SET:
+  d@25:
     0: [25, 26)
-  e@3#5,SET:
+  e@3:
     0: [3, 4)
-  g#inf,SEPARATOR:
+  g:
     0: [5, 6)
 
 iter upper=f point-key-filter=(suffix-point-keys-only,1,2)
@@ -404,23 +404,23 @@ seqnums:  [1-5]
 
 block-props
 ----
-b@10#2,SET:
+b@10:
   0: [1, 11)
-  a@1#1,SET:
+  a@1:
     0: [1, 2)
-  b@10#2,SET:
+  b@10:
     0: [10, 11)
-d@25#4,SET:
+d@25:
   0: [15, 26)
-  c@15#3,SET:
+  c@15:
     0: [15, 16)
-  d@25#4,SET:
+  d@25:
     0: [25, 26)
-g#inf,SEPARATOR:
+g:
   0: [3, 6)
-  e@3#5,SET:
+  e@3:
     0: [3, 4)
-  g#inf,SEPARATOR:
+  g:
     0: [5, 6)
 
 # Regression test for a bug in boundary checking when skipping over irrelevant
@@ -600,13 +600,13 @@ seqnums:  [1-5]
 
 block-props
 ----
-c@15#3,SET:
+c@15:
   0: [1, 16)
-  c@15#3,SET:
+  c@15:
     0: [1, 16)
-g#inf,SEPARATOR:
+g:
   0: [3, 26)
-  g#inf,SEPARATOR:
+  g:
     0: [3, 26)
 
 iter point-key-filter=(suffix-point-keys-only,1,2)
@@ -639,11 +639,11 @@ seqnums:  [1-5]
 
 block-props
 ----
-b@10#2,SET:
+b@10:
   0: [1, 11)
-d@25#4,SET:
+d@25:
   0: [15, 26)
-g#inf,SEPARATOR:
+g:
   0: [3, 6)
 
 iter point-key-filter=(suffix-point-keys-only,1,2)
@@ -674,11 +674,11 @@ collectors
 
 block-props
 ----
-a@5#1,SET:
+a@5:
   0: [5, 6)
-b@8#2,SET:
+b@8:
   0: [8, 9)
-d#inf,SEPARATOR:
+d:
   0: [7, 8)
 
 table-props
@@ -711,27 +711,27 @@ seqnums:  [1-3]
 
 block-props
 ----
-a@9#2,SET:
+a@9:
   0: [9, 11)
-  a@10#1,SET:
+  a@10:
     0: [10, 11)
-  a@9#2,SET:
+  a@9:
     0: [9, 10)
-a@7#2,SET:
+a@7:
   0: [7, 9)
-  a@8#2,SET:
+  a@8:
     0: [8, 9)
-  a@7#2,SET:
+  a@7:
     0: [7, 8)
-b#inf,SEPARATOR:
+b:
   0: [8, 10)
-  aa@9#3,SET:
+  aa@9:
     0: [9, 10)
-  b#inf,SEPARATOR:
+  b:
     0: [8, 9)
-g#inf,SEPARATOR:
+g:
   0: [5, 6)
-  g#inf,SEPARATOR:
+  g:
     0: [5, 6)
 
 # Regression test for a nil-pointer when we find a second-level index block
diff --git a/sstable/testdata/block_properties_boundlimited b/sstable/testdata/block_properties_boundlimited
index 55a23304..ffe75fab 100644
--- a/sstable/testdata/block_properties_boundlimited
+++ b/sstable/testdata/block_properties_boundlimited
@@ -21,13 +21,13 @@ table-props
 
 block-props
 ----
-b@2#2,SET:
+b@2:
   0: [2, 6)
-d@3#4,SET:
+d@3:
   0: [3, 10)
-f@0#6,SET:
+f@0:
   0: [0, 3)
-i#inf,SEPARATOR:
+i:
   0: [3, 9)
 
 # Test an interator with a bound-limited filter that has a filtering criteria
diff --git a/sstable/testdata/rewriter b/sstable/testdata/rewriter
index 8d86c9bb..5724c43a 100644
--- a/sstable/testdata/rewriter
+++ b/sstable/testdata/rewriter
@@ -6,11 +6,11 @@ c_xyz.SET.1:c
 point:    [a_xyz#1,SET-c_xyz#1,SET]
 seqnums:  [1-1]
 
-rewrite from=xyz to=123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=xyz to=123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
-rewrite failed: mismatched Comparer leveldb.BytewiseComparator vs comparer-split-4b-suffix, replacement requires same splitter to copy filters
+rewrite failed: mismatched Comparer pebble.internal.testkeys vs comparer-split-4b-suffix, replacement requires same splitter to copy filters
 
-build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+build block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 aa_xyz.SET.1:a
 ba_xyz.SET.1:b
 ca_xyz.SET.1:c
@@ -18,15 +18,15 @@ ca_xyz.SET.1:c
 point:    [aa_xyz#1,SET-ca_xyz#1,SET]
 seqnums:  [1-1]
 
-rewrite from=yz to=23 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=yz to=23 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 rewrite failed: rewriting data blocks: key has suffix "_xyz", expected "yz"
 
-rewrite from=a_xyz to=a_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=a_xyz to=a_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 rewrite failed: rewriting data blocks: key has suffix "_xyz", expected "a_xyz"
 
-build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+build block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 a_xyz.SET.1:a
 b_xyz.SET.1:b
 c_xyz.SET.1:c
@@ -64,7 +64,7 @@ b
 get f_xyz: pebble: not found
 c
 
-rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 point:    [a_123#1,SET-c_123#1,SET]
 seqnums:  [1-1]
@@ -99,7 +99,7 @@ b
 get f_123: pebble: not found
 c
 
-rewrite from=_123 to=_456 block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=2
+rewrite from=_123 to=_456 block-size=1 index-block-size=1 filter comparer=split-4b-suffix concurrency=2
 ----
 point:    [a_456#1,SET-c_456#1,SET]
 seqnums:  [1-1]
@@ -134,7 +134,7 @@ b
 get f_456: pebble: not found
 c
 
-rewrite from=_456 to=_xyz block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=3
+rewrite from=_456 to=_xyz block-size=1 index-block-size=1 filter comparer=split-4b-suffix concurrency=3
 ----
 point:    [a_xyz#1,SET-c_xyz#1,SET]
 seqnums:  [1-1]
@@ -170,7 +170,7 @@ get f_xyz: pebble: not found
 c
 
 
-rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=4
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix concurrency=4
 ----
 point:    [a_123#1,SET-c_123#1,SET]
 seqnums:  [1-1]
@@ -207,7 +207,7 @@ c
 
 # Rewrite a table that contain only range keys.
 
-build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+build block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 EncodeSpan: a-b:{(#1,RANGEKEYSET,_xyz)}
 EncodeSpan: b-c:{(#1,RANGEKEYSET,_xyz)}
 EncodeSpan: c-d:{(#1,RANGEKEYSET,_xyz)}
@@ -221,7 +221,7 @@ a-b:{(#1,RANGEKEYSET,_xyz)}
 b-c:{(#1,RANGEKEYSET,_xyz)}
 c-d:{(#1,RANGEKEYSET,_xyz)}
 
-rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 rangekey: [a#1,RANGEKEYSET-d#inf,RANGEKEYSET]
 seqnums:  [1-1]
diff --git a/sstable/testdata/rewriter_v3 b/sstable/testdata/rewriter_v3
index a62c326d..3061c8fb 100644
--- a/sstable/testdata/rewriter_v3
+++ b/sstable/testdata/rewriter_v3
@@ -6,11 +6,11 @@ c_xyz.SET.1:c
 point:    [a_xyz#1,SET-c_xyz#1,SET]
 seqnums:  [1-1]
 
-rewrite from=xyz to=123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=xyz to=123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
-rewrite failed: mismatched Comparer leveldb.BytewiseComparator vs comparer-split-4b-suffix, replacement requires same splitter to copy filters
+rewrite failed: mismatched Comparer pebble.internal.testkeys vs comparer-split-4b-suffix, replacement requires same splitter to copy filters
 
-build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+build block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 aa_xyz.SET.1:a
 ba_xyz.SET.1:b
 ca_xyz.SET.1:c
@@ -18,15 +18,15 @@ ca_xyz.SET.1:c
 point:    [aa_xyz#1,SET-ca_xyz#1,SET]
 seqnums:  [1-1]
 
-rewrite from=yz to=23 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=yz to=23 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 rewrite failed: rewriting data blocks: key has suffix "_xyz", expected "yz"
 
-rewrite from=a_xyz to=a_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=a_xyz to=a_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 rewrite failed: rewriting data blocks: key has suffix "_xyz", expected "a_xyz"
 
-build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+build block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 a_xyz.SET.1:a
 b_xyz.SET.1:b
 c_xyz.SET.1:c
@@ -64,7 +64,7 @@ b
 get f_xyz: pebble: not found
 c
 
-rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 point:    [a_123#1,SET-c_123#1,SET]
 seqnums:  [1-1]
@@ -99,7 +99,7 @@ b
 get f_123: pebble: not found
 c
 
-rewrite from=_123 to=_456 block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=2
+rewrite from=_123 to=_456 block-size=1 index-block-size=1 filter comparer=split-4b-suffix concurrency=2
 ----
 point:    [a_456#1,SET-c_456#1,SET]
 seqnums:  [1-1]
@@ -134,7 +134,7 @@ b
 get f_456: pebble: not found
 c
 
-rewrite from=_456 to=_xyz block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=3
+rewrite from=_456 to=_xyz block-size=1 index-block-size=1 filter comparer=split-4b-suffix concurrency=3
 ----
 point:    [a_xyz#1,SET-c_xyz#1,SET]
 seqnums:  [1-1]
@@ -170,7 +170,7 @@ get f_xyz: pebble: not found
 c
 
 
-rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=4
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix concurrency=4
 ----
 point:    [a_123#1,SET-c_123#1,SET]
 seqnums:  [1-1]
@@ -207,7 +207,7 @@ c
 
 # Rewrite a table that contain only range keys.
 
-build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+build block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 EncodeSpan: a-b:{(#1,RANGEKEYSET,_xyz)}
 EncodeSpan: b-c:{(#1,RANGEKEYSET,_xyz)}
 EncodeSpan: c-d:{(#1,RANGEKEYSET,_xyz)}
@@ -221,7 +221,7 @@ a-b:{(#1,RANGEKEYSET,_xyz)}
 b-c:{(#1,RANGEKEYSET,_xyz)}
 c-d:{(#1,RANGEKEYSET,_xyz)}
 
-rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
 ----
 rangekey: [a#1,RANGEKEYSET-d#inf,RANGEKEYSET]
 seqnums:  [1-1]
diff --git a/sstable/testdata/rewriter_v5 b/sstable/testdata/rewriter_v5
new file mode 100644
index 00000000..24a5a1be
--- /dev/null
+++ b/sstable/testdata/rewriter_v5
@@ -0,0 +1,246 @@
+build block-size=1 index-block-size=1 filter
+a@xyz.SET.1:a
+b@xyz.SET.1:b
+c@xyz.SET.1:c
+----
+point:    [a@xyz#1,SET-c@xyz#1,SET]
+seqnums:  [1-1]
+
+rewrite from=@xyz to=@123 block-size=1 index-block-size=1 filter comparer=split-4b-suffix
+----
+rewrite failed: mismatched Comparer pebble.internal.testkeys vs comparer-split-4b-suffix, replacement requires same splitter to copy filters
+
+build block-size=1 index-block-size=1 filter
+aa@xyz.SET.1:a
+ba@xyz.SET.1:b
+ca@xyz.SET.1:c
+----
+point:    [aa@xyz#1,SET-ca@xyz#1,SET]
+seqnums:  [1-1]
+
+rewrite from=yz to=23 block-size=1 index-block-size=1 filter
+----
+rewrite failed: rewriting data blocks: key aa@xyz#1,SET has suffix 0x4078797a; require 0x797a
+
+rewrite from=a@xyz to=a@123 block-size=1 index-block-size=1 filter
+----
+rewrite failed: rewriting data blocks: key aa@xyz#1,SET has suffix 0x4078797a; require 0x614078797a
+
+build block-size=1 index-block-size=1 filter
+a@0.SET.1:a
+b@0.SET.1:b
+c@0.SET.1:c
+----
+point:    [a@0#1,SET-c@0#1,SET]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (78)
+        83  data (78)
+       166  data (78)
+       249  index (38)
+       292  index (46)
+       343  index (44)
+       392  top-index (57)
+       454  fullfilter.rocksdb.BuiltinBloomFilter (69)
+       528  properties (515)
+      1048  meta-index (80)
+      1133  footer (53)
+      1186  EOF
+
+scan
+----
+a@0#1,SET:a
+b@0#1,SET:b
+c@0#1,SET:c
+
+get
+b@0
+f@0
+c@0
+----
+b
+get f@0: pebble: not found
+c
+
+rewrite from=@0 to=@123 block-size=1 index-block-size=1 filter
+----
+point:    [a@123#1,SET-c@123#1,SET]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (80)
+        85  data (80)
+       170  data (80)
+       255  index (40)
+       300  index (48)
+       353  index (44)
+       402  top-index (61)
+       468  fullfilter.rocksdb.BuiltinBloomFilter (69)
+       542  properties (515)
+      1062  meta-index (80)
+      1147  footer (53)
+      1200  EOF
+
+scan
+----
+a@123#1,SET:a
+b@123#1,SET:b
+c@123#1,SET:c
+
+get
+b@123
+f@123
+c@123
+----
+b
+get f@123: pebble: not found
+c
+
+rewrite from=@123 to=@456 block-size=1 index-block-size=1 filter concurrency=2
+----
+point:    [a@456#1,SET-c@456#1,SET]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (80)
+        85  data (80)
+       170  data (80)
+       255  index (40)
+       300  index (48)
+       353  index (44)
+       402  top-index (61)
+       468  fullfilter.rocksdb.BuiltinBloomFilter (69)
+       542  properties (515)
+      1062  meta-index (80)
+      1147  footer (53)
+      1200  EOF
+
+scan
+----
+a@456#1,SET:a
+b@456#1,SET:b
+c@456#1,SET:c
+
+get
+b@456
+f@456
+c@456
+----
+b
+get f@456: pebble: not found
+c
+
+rewrite from=@456 to=@123 block-size=1 index-block-size=1 filter concurrency=3
+----
+point:    [a@123#1,SET-c@123#1,SET]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (80)
+        85  data (80)
+       170  data (80)
+       255  index (40)
+       300  index (48)
+       353  index (44)
+       402  top-index (61)
+       468  fullfilter.rocksdb.BuiltinBloomFilter (69)
+       542  properties (515)
+      1062  meta-index (80)
+      1147  footer (53)
+      1200  EOF
+
+scan
+----
+a@123#1,SET:a
+b@123#1,SET:b
+c@123#1,SET:c
+
+get
+b@123
+f@123
+c@123
+----
+b
+get f@123: pebble: not found
+c
+
+
+rewrite from=@123 to=@0 block-size=1 index-block-size=1 filter concurrency=4
+----
+point:    [a@0#1,SET-c@0#1,SET]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (78)
+        83  data (78)
+       166  data (78)
+       249  index (38)
+       292  index (46)
+       343  index (44)
+       392  top-index (57)
+       454  fullfilter.rocksdb.BuiltinBloomFilter (69)
+       528  properties (515)
+      1048  meta-index (80)
+      1133  footer (53)
+      1186  EOF
+
+scan
+----
+a@0#1,SET:a
+b@0#1,SET:b
+c@0#1,SET:c
+
+get
+b@0
+f@0
+c@0
+----
+b
+get f@0: pebble: not found
+c
+
+# Rewrite a table that contain only range keys.
+
+build block-size=1 index-block-size=1 filter
+EncodeSpan: a-b:{(#1,RANGEKEYSET,@0)}
+EncodeSpan: b-c:{(#1,RANGEKEYSET,@0)}
+EncodeSpan: c-d:{(#1,RANGEKEYSET,@0)}
+----
+rangekey: [a#1,RANGEKEYSET-d#inf,RANGEKEYSET]
+seqnums:  [1-1]
+
+scan-range-key
+----
+a-b:{(#1,RANGEKEYSET,@0)}
+b-c:{(#1,RANGEKEYSET,@0)}
+c-d:{(#1,RANGEKEYSET,@0)}
+
+rewrite from=@0 to=@123 block-size=1 index-block-size=1 filter
+----
+rangekey: [a#1,RANGEKEYSET-d#inf,RANGEKEYSET]
+seqnums:  [1-1]
+
+scan-range-key
+----
+a-b:{(#1,RANGEKEYSET,@123)}
+b-c:{(#1,RANGEKEYSET,@123)}
+c-d:{(#1,RANGEKEYSET,@123)}
+
+build block-size=1 index-block-size=1 filter
+a.SET.1:a
+b.SET.1:b
+c.SET.1:c
+----
+point:    [a#1,SET-c#1,SET]
+seqnums:  [1-1]
+
+rewrite from= to=@123 block-size=1 index-block-size=1 filter
+----
+point:    [a@123#1,SET-c@123#1,SET]
+seqnums:  [1-1]
diff --git a/sstable/testdata/writer b/sstable/testdata/writer
index 93b1099e..22f73a46 100644
--- a/sstable/testdata/writer
+++ b/sstable/testdata/writer
@@ -238,11 +238,11 @@ layout
         78  index (22)
        105  index (22)
        132  index (22)
-       159  top-index (50)
-       214  properties (453)
-       672  meta-index (33)
-       710  footer (53)
-       763  EOF
+       159  top-index (48)
+       212  properties (451)
+       668  meta-index (33)
+       706  footer (53)
+       759  EOF
 
 scan
 ----
@@ -266,11 +266,11 @@ layout
          0  data (21)
         26  data (21)
         52  data (21)
-        78  index (47)
-       130  properties (551)
-       686  meta-index (33)
-       724  leveldb-footer (48)
-       772  EOF
+        78  index (45)
+       128  properties (549)
+       682  meta-index (33)
+       720  leveldb-footer (48)
+       768  EOF
 
 # Range keys, if present, are shown in the layout.
 
@@ -287,7 +287,7 @@ layout
          0  data (8)
         13  index (21)
         39  range-key (79)
-       123  properties (501)
-       629  meta-index (57)
-       691  footer (53)
-       744  EOF
+       123  properties (499)
+       627  meta-index (57)
+       689  footer (53)
+       742  EOF
diff --git a/sstable/testdata/writer_v3 b/sstable/testdata/writer_v3
index ed445059..5fbbd8dd 100644
--- a/sstable/testdata/writer_v3
+++ b/sstable/testdata/writer_v3
@@ -212,11 +212,11 @@ layout
         81  index (22)
        108  index (22)
        135  index (22)
-       162  top-index (51)
-       218  properties (453)
-       676  meta-index (33)
-       714  footer (53)
-       767  EOF
+       162  top-index (49)
+       216  properties (451)
+       672  meta-index (33)
+       710  footer (53)
+       763  EOF
 
 # Exercise the non-Reader layout-decoding codepath.
 
@@ -228,11 +228,11 @@ decode-layout
         81  index (22)
        108  index (22)
        135  index (22)
-       162  top-index (51)
-       218  properties (453)
-       676  meta-index (33)
-       714  footer (53)
-       767  EOF
+       162  top-index (49)
+       216  properties (451)
+       672  meta-index (33)
+       710  footer (53)
+       763  EOF
 
 scan
 ----
@@ -256,11 +256,11 @@ layout
          0  data (21)
         26  data (21)
         52  data (21)
-        78  index (47)
-       130  properties (551)
-       686  meta-index (33)
-       724  leveldb-footer (48)
-       772  EOF
+        78  index (45)
+       128  properties (549)
+       682  meta-index (33)
+       720  leveldb-footer (48)
+       768  EOF
 
 # Range keys, if present, are shown in the layout.
 
@@ -277,7 +277,7 @@ layout
          0  data (8)
         13  index (21)
         39  range-key (79)
-       123  properties (501)
-       629  meta-index (57)
-       691  footer (53)
-       744  EOF
+       123  properties (499)
+       627  meta-index (57)
+       689  footer (53)
+       742  EOF
diff --git a/sstable/testdata/writer_v5 b/sstable/testdata/writer_v5
new file mode 100644
index 00000000..1fdcd552
--- /dev/null
+++ b/sstable/testdata/writer_v5
@@ -0,0 +1,282 @@
+build
+a.SET.1:a
+----
+point:    [a#1,SET-a#1,SET]
+seqnums:  [1-1]
+
+scan
+----
+a#1,SET:a
+
+scan-range-del
+----
+
+scan-range-key
+----
+
+build
+a.SET.1:a
+b.DEL.2:
+c.MERGE.3:c
+EncodeSpan: d-e:{(#4,RANGEDEL)}
+f.SET.5:f
+g.DEL.6:
+h.MERGE.7:h
+EncodeSpan: i-j:{(#8,RANGEDEL)}
+EncodeSpan: j-k:{(#9,RANGEKEYDEL)}
+EncodeSpan: k-l:{(#10,RANGEKEYUNSET,@5)}
+EncodeSpan: l-m:{(#11,RANGEKEYSET,@10,foo)}
+----
+point:    [a#1,SET-h#7,MERGE]
+rangedel: [d#4,RANGEDEL-j#inf,RANGEDEL]
+rangekey: [j#9,RANGEKEYDEL-m#inf,RANGEKEYSET]
+seqnums:  [1-11]
+
+build
+a.SET.1:a
+b.DEL.2:
+c.MERGE.3:c
+EncodeSpan: d-e:{(#4,RANGEDEL)}
+f.SET.5:f
+g.DEL.6:
+h.MERGE.7:h
+EncodeSpan: i-j:{(#8,RANGEDEL)}
+----
+point:    [a#1,SET-h#7,MERGE]
+rangedel: [d#4,RANGEDEL-j#inf,RANGEDEL]
+seqnums:  [1-8]
+
+scan
+----
+a#1,SET:a
+b#2,DEL:
+c#3,MERGE:c
+f#5,SET:f
+g#6,DEL:
+h#7,MERGE:h
+
+scan-range-del
+----
+d-e:{(#4,RANGEDEL)}
+i-j:{(#8,RANGEDEL)}
+
+# 3: a-----------m
+# 2:      f------------s
+# 1:          j---------------z
+
+build
+EncodeSpan: a-f:{(#3,RANGEDEL)}
+EncodeSpan: f-j:{(#3,RANGEDEL) (#2,RANGEDEL)}
+EncodeSpan: j-m:{(#3,RANGEDEL) (#2,RANGEDEL) (#1,RANGEDEL)}
+EncodeSpan: m-s:{(#2,RANGEDEL) (#1,RANGEDEL)}
+EncodeSpan: s-z:{(#1,RANGEDEL)}
+----
+rangedel: [a#3,RANGEDEL-z#inf,RANGEDEL]
+seqnums:  [1-3]
+
+scan
+----
+
+scan-range-del
+----
+a-f:{(#3,RANGEDEL)}
+f-j:{(#3,RANGEDEL) (#2,RANGEDEL)}
+j-m:{(#3,RANGEDEL) (#2,RANGEDEL) (#1,RANGEDEL)}
+m-s:{(#2,RANGEDEL) (#1,RANGEDEL)}
+s-z:{(#1,RANGEDEL)}
+
+scan-range-key
+----
+
+# The range tombstone upper bound is exclusive, so a point operation
+# on that same key will be the actual boundary.
+
+build
+EncodeSpan: a-b:{(#3,RANGEDEL)}
+b.SET.4:c
+----
+point:    [b#4,SET-b#4,SET]
+rangedel: [a#3,RANGEDEL-b#inf,RANGEDEL]
+seqnums:  [3-4]
+
+build
+EncodeSpan: a-b:{(#3,RANGEDEL)}
+b.SET.2:c
+----
+point:    [b#2,SET-b#2,SET]
+rangedel: [a#3,RANGEDEL-b#inf,RANGEDEL]
+seqnums:  [2-3]
+
+build
+EncodeSpan: a-c:{(#3,RANGEDEL)}
+b.SET.2:c
+----
+point:    [b#2,SET-b#2,SET]
+rangedel: [a#3,RANGEDEL-c#inf,RANGEDEL]
+seqnums:  [2-3]
+
+# Keys must be added in order.
+
+build
+a.SET.1:b
+a.SET.2:c
+----
+pebble: keys must be added in strictly increasing order: a#2,SET
+
+build
+b.SET.1:a
+a.SET.2:b
+----
+pebble: keys must be added in strictly increasing order: a#2,SET
+
+build
+b.RANGEDEL.1:c
+a.RANGEDEL.2:b
+----
+RANGEDEL must be added through EncodeSpan
+
+build
+EncodeSpan: b-c:{(#1,RANGEDEL)}
+EncodeSpan: a-b:{(#2,RANGEDEL)}
+----
+pebble: keys must be added in order: b-c:{(#1,RANGEDEL)}, a-b:{(#2,RANGEDEL)}
+
+build-raw
+EncodeSpan: a-c:{(#1,RANGEDEL)}
+EncodeSpan: a-c:{(#2,RANGEDEL)}
+----
+pebble: keys must be added in order: a-c:{(#1,RANGEDEL)}, a-c:{(#2,RANGEDEL)}
+
+build-raw
+EncodeSpan: a-c:{(#1,RANGEDEL)}
+EncodeSpan: b-d:{(#2,RANGEDEL)}
+----
+pebble: keys must be added in order: a-c:{(#1,RANGEDEL)}, b-d:{(#2,RANGEDEL)}
+
+build-raw
+EncodeSpan: a-c:{(#2,RANGEDEL)}
+EncodeSpan: a-d:{(#1,RANGEDEL)}
+----
+pebble: keys must be added in order: a-c:{(#2,RANGEDEL)}, a-d:{(#1,RANGEDEL)}
+
+build-raw
+EncodeSpan: a-c:{(#1,RANGEDEL)}
+EncodeSpan: c-d:{(#2,RANGEDEL)}
+----
+rangedel: [a#1,RANGEDEL-d#inf,RANGEDEL]
+seqnums:  [1-2]
+
+build-raw
+EncodeSpan: a-b:{(#2,RANGEKEYSET,@10,foo) (#1,RANGEKEYSET,@10,foo)}
+----
+rangekey: [a#2,RANGEKEYSET-b#inf,RANGEKEYSET]
+seqnums:  [1-2]
+
+build-raw
+EncodeSpan: b-c:{(#2,RANGEKEYSET,@10,foo)}
+EncodeSpan: a-b:{(#1,RANGEKEYSET,@10,foo)}
+----
+pebble: keys must be added in order: b-c:{(#2,RANGEKEYSET)}, a-b:{(#1,RANGEKEYSET,@10,foo)}
+
+build-raw
+EncodeSpan: a-c:{(#1,RANGEKEYSET,@10,foo)}
+EncodeSpan: c-d:{(#2,RANGEKEYSET,@10,foo)}
+----
+rangekey: [a#1,RANGEKEYSET-d#inf,RANGEKEYSET]
+seqnums:  [1-2]
+
+# Range keys may have perfectly aligned spans (including sequence numbers),
+# though the key kinds must be ordered (descending).
+
+build-raw
+EncodeSpan: a-b:{(#1,RANGEKEYSET,@10,foo) (#1,RANGEKEYUNSET,@10) (#1,RANGEKEYDEL)}
+----
+rangekey: [a#1,RANGEKEYSET-b#inf,RANGEKEYDEL]
+seqnums:  [1-1]
+
+# Setting a very small index-block-size results in a two-level index.
+
+build block-size=1 index-block-size=1
+a.SET.1:a
+b.SET.1:b
+c.SET.1:c
+----
+point:    [a#1,SET-c#1,SET]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (74)
+        79  data (74)
+       158  data (74)
+       237  index (36)
+       278  index (44)
+       327  index (44)
+       376  top-index (53)
+       434  properties (480)
+       919  meta-index (33)
+       957  footer (53)
+      1010  EOF
+
+# Exercise the non-Reader layout-decoding codepath.
+
+decode-layout
+----
+         0  data (74)
+        79  data (74)
+       158  data (74)
+       237  index (36)
+       278  index (44)
+       327  index (44)
+       376  top-index (53)
+       434  properties (480)
+       919  meta-index (33)
+       957  footer (53)
+      1010  EOF
+
+scan
+----
+a#1,SET:a
+b#1,SET:b
+c#1,SET:c
+
+# Enabling leveldb format disables the creation of a two-level index
+# (the input data here mirrors the test case above).
+
+build leveldb block-size=1 index-block-size=1
+a.SET.1:a
+b.SET.1:b
+c.SET.1:c
+----
+point:    [a#1,SET-c#1,SET]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (21)
+        26  data (21)
+        52  data (21)
+        78  index (45)
+       128  properties (549)
+       682  meta-index (33)
+       720  leveldb-footer (48)
+       768  EOF
+
+# Range keys, if present, are shown in the layout.
+
+build
+EncodeSpan: a-b:{(#3,RANGEKEYSET,@3,foo)}
+EncodeSpan: b-c:{(#2,RANGEKEYSET,@2,bar)}
+EncodeSpan: c-d:{(#1,RANGEKEYSET,@1,baz)}
+----
+rangekey: [a#3,RANGEKEYSET-d#inf,RANGEKEYSET]
+seqnums:  [1-3]
+
+layout
+----
+         0  index (28)
+        33  range-key (84)
+       122  properties (441)
+       568  meta-index (57)
+       630  footer (53)
+       683  EOF
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 53c8e187..91b15561 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -33,13 +33,21 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+type tableFormatFile struct {
+	TableFormat TableFormat
+	File        string
+}
+
 func testWriterParallelism(t *testing.T, parallelism bool) {
-	for _, format := range []TableFormat{TableFormatPebblev2, TableFormatPebblev3} {
-		tdFile := "testdata/writer"
-		if format == TableFormatPebblev3 {
-			tdFile = "testdata/writer_v3"
-		}
-		t.Run(format.String(), func(t *testing.T) { runDataDriven(t, tdFile, format, parallelism) })
+	formatFiles := []tableFormatFile{
+		{TableFormat: TableFormatPebblev2, File: "testdata/writer"},
+		{TableFormat: TableFormatPebblev3, File: "testdata/writer_v3"},
+		{TableFormat: TableFormatPebblev5, File: "testdata/writer_v5"},
+	}
+	for _, tff := range formatFiles {
+		t.Run(tff.TableFormat.String(), func(t *testing.T) {
+			runDataDriven(t, tff.File, tff.TableFormat, parallelism)
+		})
 	}
 }
 func TestWriter(t *testing.T) {
@@ -48,12 +56,15 @@ func TestWriter(t *testing.T) {
 }
 
 func testRewriterParallelism(t *testing.T, parallelism bool) {
-	for _, format := range []TableFormat{TableFormatPebblev2, TableFormatPebblev3} {
-		tdFile := "testdata/rewriter"
-		if format == TableFormatPebblev3 {
-			tdFile = "testdata/rewriter_v3"
-		}
-		t.Run(format.String(), func(t *testing.T) { runDataDriven(t, tdFile, format, parallelism) })
+	formatFiles := []tableFormatFile{
+		{TableFormat: TableFormatPebblev2, File: "testdata/rewriter"},
+		{TableFormat: TableFormatPebblev3, File: "testdata/rewriter_v3"},
+		{TableFormat: TableFormatPebblev5, File: "testdata/rewriter_v5"},
+	}
+	for _, tff := range formatFiles {
+		t.Run(tff.TableFormat.String(), func(t *testing.T) {
+			runDataDriven(t, tff.File, tff.TableFormat, parallelism)
+		})
 	}
 }
 
@@ -126,6 +137,7 @@ func runDataDriven(t *testing.T, file string, tableFormat TableFormat, paralleli
 			var meta *WriterMetadata
 			var err error
 			wopts := &WriterOptions{
+				Comparer:    testkeys.Comparer,
 				TableFormat: tableFormat,
 				Parallelism: parallelism,
 			}
@@ -147,6 +159,7 @@ func runDataDriven(t *testing.T, file string, tableFormat TableFormat, paralleli
 			var meta *WriterMetadata
 			var err error
 			meta, r, err = runBuildRawCmd(td, &WriterOptions{
+				Comparer:    testkeys.Comparer,
 				TableFormat: tableFormat,
 			})
 			if err != nil {
diff --git a/table_cache_test.go b/table_cache_test.go
index 8cda9295..12f6314e 100644
--- a/table_cache_test.go
+++ b/table_cache_test.go
@@ -283,9 +283,9 @@ func TestVirtualReadsWiring(t *testing.T) {
 	l6 := currVersion.Levels[6]
 	l6FileIter := l6.Iter()
 	parentFile := l6FileIter.First()
-	f1 := FileNum(d.mu.versions.nextFileNum)
+	f1 := FileNum(d.mu.versions.nextFileNum.Load())
 	f2 := f1 + 1
-	d.mu.versions.nextFileNum += 2
+	d.mu.versions.nextFileNum.Add(2)
 
 	seqNumA := parentFile.Smallest.SeqNum()
 	// See SeqNum comments above.