Repository: cockroachdb/pebble
Pull request #4196 (2024-12-10 18:21:11)

metamorphic: add KeyFormat type
Refactor the metamorphic test to pull the particulars of the key format and
into a KeyFormat type. This will be used to allow the metamorphic test to use
other key formats, such as cockroachkvs.


Patch:
diff --git a/internal/base/lazy_value.go b/internal/base/lazy_value.go
index ebeb62ba..ff55067e 100644
--- a/internal/base/lazy_value.go
+++ b/internal/base/lazy_value.go
@@ -222,6 +222,12 @@ func (lv *LazyValue) fetchValue(
 	return f.value, f.callerOwned, f.err
 }
 
+// IsInPlaceValue returns true iff the value was stored in-place and does not
+// need to be fetched externally.
+func (lv *LazyValue) IsInPlaceValue() bool {
+	return lv.Fetcher == nil
+}
+
 // InPlaceValue returns the value under the assumption that it is in-place.
 // This is for Pebble-internal code.
 func (lv *LazyValue) InPlaceValue() []byte {
diff --git a/internal/base/merger.go b/internal/base/merger.go
index 757d1509..c4831090 100644
--- a/internal/base/merger.go
+++ b/internal/base/merger.go
@@ -7,7 +7,7 @@ package base
 import "io"
 
 // Merge creates a ValueMerger for the specified key initialized with the value
-// of one merge operand.
+// of one merge operand. The caller retains ownership of key and value.
 type Merge func(key, value []byte) (ValueMerger, error)
 
 // ValueMerger receives merge operands one by one. The operand received is either
diff --git a/internal/cache/clockpro.go b/internal/cache/clockpro.go
index 2459ec7b..53f756cb 100644
--- a/internal/cache/clockpro.go
+++ b/internal/cache/clockpro.go
@@ -18,6 +18,7 @@
 package cache // import "github.com/cockroachdb/pebble/internal/cache"
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"runtime"
@@ -25,6 +26,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -114,9 +116,31 @@ type shard struct {
 	countHot  int64
 	countCold int64
 	countTest int64
+
+	// Some fields in readShard are protected by mu. See comments in declaration
+	// of readShard.
+	readShard readShard
+}
+
+func (c *shard) init(maxSize int64) {
+	*c = shard{
+		maxSize:    maxSize,
+		coldTarget: maxSize,
+	}
+	if entriesGoAllocated {
+		c.entries = make(map[*entry]struct{})
+	}
+	c.blocks.Init(16)
+	c.files.Init(16)
+	c.readShard.Init(c)
 }
 
-func (c *shard) Get(id ID, fileNum base.DiskFileNum, offset uint64) Handle {
+// getWithMaybeReadEntry is the internal helper for implementing
+// Cache.{Get,GetWithReadHandle}. When desireReadEntry is true, and the block
+// is not in the cache (!Handle.Valid()), a non-nil readEntry is returned.
+func (c *shard) getWithMaybeReadEntry(
+	id ID, fileNum base.DiskFileNum, offset uint64, desireReadEntry bool,
+) (Handle, *readEntry) {
 	c.mu.RLock()
 	var value *Value
 	if e, _ := c.blocks.Get(key{fileKey{id, fileNum}, offset}); e != nil {
@@ -126,12 +150,30 @@ func (c *shard) Get(id ID, fileNum base.DiskFileNum, offset uint64) Handle {
 		}
 	}
 	c.mu.RUnlock()
+	var re *readEntry
+	if value == nil && desireReadEntry {
+		c.mu.Lock()
+		// After the c.mu.RUnlock(), someone could have inserted the value in the
+		// cache. We could tolerate the race and do a file read, or do another map
+		// lookup. We choose to do the latter, since the cost of a map lookup is
+		// insignificant compared to the cost of reading a block from a file.
+		if e, _ := c.blocks.Get(key{fileKey{id, fileNum}, offset}); e != nil {
+			value = e.acquireValue()
+			if value != nil {
+				e.referenced.Store(true)
+			}
+		}
+		if value == nil {
+			re = c.readShard.getReadEntryLocked(id, fileNum, offset)
+		}
+		c.mu.Unlock()
+	}
 	if value == nil {
 		c.misses.Add(1)
-		return Handle{}
+	} else {
+		c.hits.Add(1)
 	}
-	c.hits.Add(1)
-	return Handle{value: value}
+	return Handle{value: value}, re
 }
 
 func (c *shard) Set(id ID, fileNum base.DiskFileNum, offset uint64, value *Value) Handle {
@@ -170,6 +212,11 @@ func (c *shard) Set(id ID, fileNum base.DiskFileNum, offset uint64, value *Value
 			value.ref.trace("add-hot")
 			c.sizeHot += delta
 		} else {
+			// TODO(sumeer): unclear why we don't set e.ptype to etHot on this path.
+			// In the default case below, where the state is etTest we set it to
+			// etHot. But etTest is "colder" than etCold, since the only transition
+			// into etTest is etCold => etTest, so since etTest transitions to
+			// etHot, then etCold should also transition.
 			value.ref.trace("add-cold")
 			c.sizeCold += delta
 		}
@@ -737,15 +784,7 @@ func newShards(size int64, shards int) *Cache {
 	c.idAlloc.Store(1)
 	c.trace("alloc", c.refs.Load())
 	for i := range c.shards {
-		c.shards[i] = shard{
-			maxSize:    size / int64(len(c.shards)),
-			coldTarget: size / int64(len(c.shards)),
-		}
-		if entriesGoAllocated {
-			c.shards[i].entries = make(map[*entry]struct{})
-		}
-		c.shards[i].blocks.Init(16)
-		c.shards[i].files.Init(16)
+		c.shards[i].init(size / int64(len(c.shards)))
 	}
 
 	// Note: this is a no-op if invariants are disabled or race is enabled.
@@ -822,7 +861,49 @@ func (c *Cache) Unref() {
 // Get retrieves the cache value for the specified file and offset, returning
 // nil if no value is present.
 func (c *Cache) Get(id ID, fileNum base.DiskFileNum, offset uint64) Handle {
-	return c.getShard(id, fileNum, offset).Get(id, fileNum, offset)
+	h, re := c.getShard(id, fileNum, offset).getWithMaybeReadEntry(
+		id, fileNum, offset, false)
+	if invariants.Enabled && re != nil {
+		panic("readEntry should be nil")
+	}
+	return h
+}
+
+// GetWithReadHandle retrieves the cache value for the specified ID, fileNum
+// and offset. If found, a valid Handle is returned (with cacheHit set to
+// true), else a valid ReadHandle is returned.
+//
+// See the ReadHandle declaration for the contract the caller must satisfy
+// when getting a valid ReadHandle.
+//
+// This method can block before returning since multiple concurrent gets for
+// the same cache value will take turns getting a ReadHandle, which represents
+// permission to do the read. This blocking respects context cancellation, in
+// which case an error is returned (and not a valid ReadHandle).
+//
+// When blocking, the errorDuration return value can be non-zero and is
+// populated with the total duration that other readers that observed an error
+// (see ReadHandle.SetReadError) spent in doing the read. This duration can be
+// greater than the time spent blocked in this method, since some of these
+// errors could have occurred prior to this call. But it serves as a rough
+// indicator of whether turn taking could have caused higher latency due to
+// context cancellation of other readers.
+//
+// While waiting, someone else may successfully read the value, which results
+// in a valid Handle being returned. This is a case where cacheHit=false.
+func (c *Cache) GetWithReadHandle(
+	ctx context.Context, id ID, fileNum base.DiskFileNum, offset uint64,
+) (h Handle, rh ReadHandle, errorDuration time.Duration, cacheHit bool, err error) {
+	h, re := c.getShard(id, fileNum, offset).getWithMaybeReadEntry(
+		id, fileNum, offset, true)
+	if h.Valid() {
+		return h, ReadHandle{}, 0, true, nil
+	}
+	h, errorDuration, err = re.waitForReadPermissionOrHandle(ctx)
+	if err != nil || h.Valid() {
+		return h, ReadHandle{}, errorDuration, false, err
+	}
+	return Handle{}, ReadHandle{entry: re}, errorDuration, false, nil
 }
 
 // Set sets the cache value for the specified file and offset, overwriting an
diff --git a/internal/cache/read_shard.go b/internal/cache/read_shard.go
new file mode 100644
index 00000000..432edc1e
--- /dev/null
+++ b/internal/cache/read_shard.go
@@ -0,0 +1,414 @@
+// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package cache
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/swiss"
+)
+
+// readShard coordinates the read of a block that will be put in the cache. It
+// ensures only one goroutine is reading a block, and other callers block
+// until that goroutine is done (with success or failure). In the case of
+// success, the other goroutines will use the value that was read, even if it
+// is too large to be placed in the cache, or got evicted from the cache
+// before they got scheduled. In the case of a failure (read error or context
+// cancellation), one of the waiters will be given a turn to do the read.
+//
+// This turn-taking ensures that a large number of concurrent attempts to read
+// the same block that is not in the cache does not result in the same number
+// of reads from the filesystem (or remote storage). We have seen large spikes
+// in memory usage (and CPU usage for memory allocation/deallocation) without
+// this turn-taking.
+//
+// It introduces a small risk related to context cancellation -- if many
+// readers assigned a turn exceed their deadline while doing the read and
+// report an error, a reader with a longer deadline can unnecessarily wait. We
+// accept this risk for now since the primary production use in CockroachDB is
+// filesystem reads, where context cancellation is not respected. We do
+// introduce an error duration metric emitted in traces that can be used to
+// quantify such wasteful waiting. Note that this same risk extends to waiting
+// on the Options.LoadBlockSema, so the error duration metric includes the
+// case of an error when waiting on the semaphore (as a side effect of that
+// waiting happening in the caller, sstable.Reader).
+//
+// Design choices and motivation:
+//
+//   - readShard is tightly integrated with a cache shard: At its core,
+//     readShard is a map with synchronization. For the same reason the cache is
+//     sharded (for higher concurrency by sharding the mutex), it is beneficial
+//     to shard synchronization on readShard. By making readShard a member of
+//     shard, this sharding is trivially accomplished. Additionally, the code
+//     feels cleaner when there isn't a race between a cache miss, followed by
+//     creating a readEntry that is no longer needed because someone else has
+//     done the read since the miss and inserted into the cache. By making the
+//     readShard use shard.mu, such a race is avoided. A side benefit is that
+//     the cache interaction can be hidden behind readEntry.SetReadValue. One
+//     disadvantage of this tightly integrated design is that it does not
+//     encompass readers that will put the read value into a block.BufferPool --
+//     we don't worry about those since block.BufferPool is only used for
+//     compactions and there is at most one compaction reader of a block. There
+//     is the possibility that the compaction reader and a user-facing iterator
+//     reader will do duplicate reads, but we accept that deficiency.
+//
+//   - readMap is separate from shard.blocks map: One could have a design which
+//     extends the cache entry and unifies the two maps. However, we never want
+//     to evict a readEntry while there are readers waiting for the block read
+//     (including the case where the corresponding file is being removed from
+//     shard.files). Also, the number of stable cache entries is huge and
+//     therefore is manually allocated, while the number of readEntries is small
+//     (so manual allocation isn't necessary). For these reasons we maintain a
+//     separate map. This separation also results in more modular code, instead
+//     of piling more stuff into shard.
+type readShard struct {
+	// shard is only used for locking, and calling shard.Set.
+	shard *shard
+	// Protected by shard.mu.
+	//
+	// shard.mu is never held when acquiring readEntry.mu. shard.mu is a shared
+	// resource and must be released quickly.
+	shardMu struct {
+		readMap swiss.Map[key, *readEntry]
+	}
+}
+
+func (rs *readShard) Init(shard *shard) *readShard {
+	*rs = readShard{
+		shard: shard,
+	}
+	// Choice of 16 is arbitrary.
+	rs.shardMu.readMap.Init(16)
+	return rs
+}
+
+// getReadEntryLocked gets a *readEntry for (id, fileNum, offset). shard.mu is
+// already write locked.
+func (rs *readShard) getReadEntryLocked(id ID, fileNum base.DiskFileNum, offset uint64) *readEntry {
+	k := key{fileKey{id, fileNum}, offset}
+	e, ok := rs.shardMu.readMap.Get(k)
+	if !ok {
+		e = newReadEntry(rs, id, fileNum, offset)
+		rs.shardMu.readMap.Put(k, e)
+	} else {
+		e.refCount.acquireAllowZero()
+	}
+	return e
+}
+
+func (rs *readShard) lenForTesting() int {
+	rs.shard.mu.Lock()
+	defer rs.shard.mu.Unlock()
+	return rs.shardMu.readMap.Len()
+}
+
+// readEntry is used to coordinate between concurrent attempted readers of the
+// same block.
+type readEntry struct {
+	readShard *readShard
+	id        ID
+	fileNum   base.DiskFileNum
+	offset    uint64
+	mu        struct {
+		sync.RWMutex
+		// v, when non-nil, has a ref from readEntry, which is unreffed when
+		// readEntry is deleted from the readMap.
+		v *Value
+		// isReading and ch together capture the state of whether someone has been
+		// granted a turn to read, and of readers waiting for that read to finish.
+		// ch is lazily allocated since most readEntries will not see concurrent
+		// readers. This lazy allocation results in one transition of ch from nil
+		// to non-nil, so waiters can read this non-nil ch and block on reading
+		// from it without holding mu.
+		//
+		// ch is written to, to signal one waiter to start doing the read. ch is
+		// closed when the value is successfully read and has been stored in v, so
+		// that all waiters wake up and read v. ch is a buffered channel with a
+		// capacity of 1.
+		//
+		// State transitions when trying to wait for turn:
+		// Case !isReading:
+		//   set isReading=true; Drain the ch if non-nil and non-empty; proceed
+		//   with turn to do the read.
+		// Case isReading:
+		//   allocate ch if nil; wait on ch
+		// Finished reading successfully:
+		//   set isReading=false; if ch is non-nil, close ch.
+		// Finished reading with failure:
+		//   set isReading=false; if ch is non-nil, write to ch.
+		//
+		// INVARIANT:
+		// isReading => ch is nil or ch is empty.
+		isReading bool
+		ch        chan struct{}
+		// Total duration of reads and semaphore waiting that resulted in error.
+		errorDuration time.Duration
+		readStart     time.Time
+	}
+	// Count of ReadHandles that refer to this readEntry. Increments always hold
+	// shard.mu. So if this is found to be 0 while holding shard.mu, it is safe
+	// to delete readEntry from readShard.shardMu.readMap.
+	refCount refcnt
+}
+
+var readEntryPool = sync.Pool{
+	New: func() interface{} {
+		return &readEntry{}
+	},
+}
+
+func newReadEntry(rs *readShard, id ID, fileNum base.DiskFileNum, offset uint64) *readEntry {
+	e := readEntryPool.Get().(*readEntry)
+	*e = readEntry{
+		readShard: rs,
+		id:        id,
+		fileNum:   fileNum,
+		offset:    offset,
+	}
+	e.refCount.init(1)
+	return e
+}
+
+// waitForReadPermissionOrHandle returns either an already read value (in
+// Handle), an error (if the context was cancelled), or neither, which is a
+// directive to the caller to do the read. In this last case the caller must
+// call either setReadValue or setReadError.
+//
+// In all cases, errorDuration is populated with the total duration that
+// readers that observed an error (setReadError) spent in doing the read. This
+// duration can be greater than the time spend in waitForReadPermissionHandle,
+// since some of these errors could have occurred prior to this call. But it
+// serves as a rough indicator of whether turn taking could have caused higher
+// latency due to context cancellation.
+func (e *readEntry) waitForReadPermissionOrHandle(
+	ctx context.Context,
+) (h Handle, errorDuration time.Duration, err error) {
+	constructHandleLocked := func() Handle {
+		if e.mu.v == nil {
+			panic("value is nil")
+		}
+		e.mu.v.acquire()
+		return Handle{value: e.mu.v}
+	}
+	becomeReaderLocked := func() {
+		if e.mu.v != nil {
+			panic("value is non-nil")
+		}
+		if e.mu.isReading {
+			panic("isReading is already true")
+		}
+		e.mu.isReading = true
+		if e.mu.ch != nil {
+			// Drain the channel, so that no one else mistakenly believes they
+			// should read.
+			select {
+			case <-e.mu.ch:
+			default:
+			}
+		}
+		e.mu.readStart = time.Now()
+	}
+	unlockAndUnrefAndTryRemoveFromMap := func(readLock bool) (errorDuration time.Duration) {
+		removeState := e.makeTryRemoveStateLocked()
+		errorDuration = e.mu.errorDuration
+		if readLock {
+			e.mu.RUnlock()
+		} else {
+			e.mu.Unlock()
+		}
+		unrefAndTryRemoveFromMap(removeState)
+		return errorDuration
+	}
+
+	for {
+		e.mu.Lock()
+		if e.mu.v != nil {
+			// Value has already been read.
+			h := constructHandleLocked()
+			errorDuration = unlockAndUnrefAndTryRemoveFromMap(false)
+			return h, errorDuration, nil
+		}
+		// Not already read. Wait for turn to do the read or for someone else to do
+		// the read.
+		if !e.mu.isReading {
+			// Have permission to do the read.
+			becomeReaderLocked()
+			errorDuration = e.mu.errorDuration
+			e.mu.Unlock()
+			return Handle{}, errorDuration, nil
+		}
+		if e.mu.ch == nil {
+			// Rare case when multiple readers are concurrently trying to read. If
+			// this turns out to be common enough we could use a sync.Pool.
+			e.mu.ch = make(chan struct{}, 1)
+		}
+		ch := e.mu.ch
+		e.mu.Unlock()
+		select {
+		case <-ctx.Done():
+			e.mu.Lock()
+			errorDuration = unlockAndUnrefAndTryRemoveFromMap(false)
+			return Handle{}, errorDuration, ctx.Err()
+		case _, ok := <-ch:
+			if !ok {
+				// Channel closed, so value was read.
+				e.mu.RLock()
+				if e.mu.v == nil {
+					panic("value is nil")
+				}
+				h := constructHandleLocked()
+				errorDuration = unlockAndUnrefAndTryRemoveFromMap(true)
+				return h, errorDuration, nil
+			}
+			// Else, probably granted permission to do the read. NB: since isReading
+			// is false, someone else can slip through before this thread acquires
+			// e.mu, and take the turn. So try to actually get the turn by trying
+			// again in the loop.
+		}
+	}
+}
+
+// tryRemoveState captures the state needed by tryRemoveFromMap. The caller
+// constructs it before calling unrefAndTryRemoveFromMap since it typically
+// held readEntry.mu, which avoids acquiring it again in
+// unrefAndTryRemoveFromMap.
+type tryRemoveState struct {
+	rs *readShard
+	k  key
+	e  *readEntry
+}
+
+// makeTryRemoveStateLocked initializes tryRemoveState.
+func (e *readEntry) makeTryRemoveStateLocked() tryRemoveState {
+	return tryRemoveState{
+		rs: e.readShard,
+		k:  key{fileKey{e.id, e.fileNum}, e.offset},
+		e:  e,
+	}
+}
+
+// unrefAndTryRemoveFromMap tries to remove s.k => s.e from the map in s.rs.
+// It is possible that after unreffing that s.e has already been removed, and
+// is now back in the sync.Pool, or being reused (for the same or different
+// key). This is because after unreffing, which caused the s.e.refCount to
+// become zero, but before acquiring shard.mu, it could have been incremented
+// and decremented concurrently, and some other goroutine could have observed
+// a different decrement to 0, and raced ahead and deleted s.e from the
+// readMap.
+func unrefAndTryRemoveFromMap(s tryRemoveState) {
+	if !s.e.refCount.release() {
+		return
+	}
+	s.rs.shard.mu.Lock()
+	e2, ok := s.rs.shardMu.readMap.Get(s.k)
+	if !ok || e2 != s.e {
+		// Already removed.
+		s.rs.shard.mu.Unlock()
+		return
+	}
+	if s.e.refCount.value() != 0 {
+		s.rs.shard.mu.Unlock()
+		return
+	}
+	// k => e and e.refCount == 0. And it cannot be incremented since
+	// shard.mu.Lock() is held. So remove from map.
+	s.rs.shardMu.readMap.Delete(s.k)
+	s.rs.shard.mu.Unlock()
+
+	// Free s.e.
+	s.e.mu.Lock()
+	if s.e.mu.v != nil {
+		s.e.mu.v.release()
+		s.e.mu.v = nil
+	}
+	s.e.mu.Unlock()
+	*s.e = readEntry{}
+	readEntryPool.Put(s.e)
+}
+
+func (e *readEntry) setReadValue(v *Value) Handle {
+	// Add to the cache before taking another ref for readEntry, since the cache
+	// expects ref=1 when it is called.
+	//
+	// TODO(sumeer): if e.refCount > 1, we should consider overriding to ensure
+	// that it is added as etHot. The common case will be e.refCount = 1, and we
+	// don't want to acquire e.mu twice, so one way to do this would be relax
+	// the invariant in shard.Set that requires Value.refs() == 1. Then we can
+	// do the work under e.mu before calling shard.Set.
+	h := e.readShard.shard.Set(e.id, e.fileNum, e.offset, v)
+	e.mu.Lock()
+	// Acquire a ref for readEntry, since we are going to remember it in e.mu.v.
+	v.acquire()
+	e.mu.v = v
+	if !e.mu.isReading {
+		panic("isReading is false")
+	}
+	e.mu.isReading = false
+	if e.mu.ch != nil {
+		// Inform all waiters so they can use e.mu.v. Not all readers have called
+		// readEntry.waitForReadPermissionOrHandle, and those will also use
+		// e.mu.v.
+		close(e.mu.ch)
+	}
+	removeState := e.makeTryRemoveStateLocked()
+	e.mu.Unlock()
+	unrefAndTryRemoveFromMap(removeState)
+	return h
+}
+
+func (e *readEntry) setReadError(err error) {
+	e.mu.Lock()
+	if !e.mu.isReading {
+		panic("isReading is false")
+	}
+	e.mu.isReading = false
+	if e.mu.ch != nil {
+		select {
+		case e.mu.ch <- struct{}{}:
+		default:
+			panic("channel is not empty")
+		}
+	}
+	e.mu.errorDuration += time.Since(e.mu.readStart)
+	removeState := e.makeTryRemoveStateLocked()
+	e.mu.Unlock()
+	unrefAndTryRemoveFromMap(removeState)
+}
+
+// ReadHandle represents a contract with a caller that had a miss when doing a
+// cache lookup, and wants to do a read and insert the read block into the
+// cache. The contract applies when ReadHandle.Valid returns true, in which
+// case the caller has been assigned the turn to do the read (and others are
+// potentially waiting for it).
+//
+// Contract:
+//
+// The caller must immediately start doing a read, or can first wait on a
+// shared resource that would also block a different reader if it was assigned
+// the turn instead (specifically, this refers to Options.LoadBlockSema).
+// After the read, it must either call SetReadValue or SetReadError depending
+// on whether the read succeeded or failed.
+type ReadHandle struct {
+	entry *readEntry
+}
+
+// Valid returns true for a valid ReadHandle.
+func (rh ReadHandle) Valid() bool {
+	return rh.entry != nil
+}
+
+// SetReadValue provides the Value that the caller has read. The caller is
+// responsible for releasing the returned Handle when it is no longer needed.
+func (rh ReadHandle) SetReadValue(v *Value) Handle {
+	return rh.entry.setReadValue(v)
+}
+
+// SetReadError specifies that the caller has encountered a read error.
+func (rh ReadHandle) SetReadError(err error) {
+	rh.entry.setReadError(err)
+}
diff --git a/internal/cache/refcnt_normal.go b/internal/cache/refcnt_normal.go
index 9ab33486..bf1a4756 100644
--- a/internal/cache/refcnt_normal.go
+++ b/internal/cache/refcnt_normal.go
@@ -37,6 +37,15 @@ func (v *refcnt) acquire() {
 	}
 }
 
+// acquireAllowZero is the same as acquire, but allows acquireAllowZero to be
+// called with a zero refcnt. This is useful for cases where the entry which
+// is being reference counted is inside a container and the container does not
+// hold a reference. The container uses release() returning true to attempt to
+// do a cleanup from the container.
+func (v *refcnt) acquireAllowZero() {
+	v.val.Add(1)
+}
+
 func (v *refcnt) release() bool {
 	switch v := v.val.Add(-1); {
 	case v < 0:
@@ -48,6 +57,10 @@ func (v *refcnt) release() bool {
 	}
 }
 
+func (v *refcnt) value() int32 {
+	return v.val.Load()
+}
+
 func (v *refcnt) trace(msg string) {
 }
 
diff --git a/internal/cache/refcnt_tracing.go b/internal/cache/refcnt_tracing.go
index 1d5e6c02..dd94088e 100644
--- a/internal/cache/refcnt_tracing.go
+++ b/internal/cache/refcnt_tracing.go
@@ -41,6 +41,16 @@ func (v *refcnt) acquire() {
 	v.trace("acquire")
 }
 
+// acquireAllowZero is the same as acquire, but allows acquireAllowZero to be
+// called with a zero refcnt. This is useful for cases where the entry which
+// is being reference counted is inside a container and the container does not
+// hold a reference. The container uses release() returning true to attempt to
+// do a cleanup from the container.
+func (v *refcnt) acquireAllowZero() {
+	v.val.Add(1)
+	v.trace("acquire")
+}
+
 func (v *refcnt) release() bool {
 	n := v.val.Add(-1)
 	switch {
@@ -51,6 +61,10 @@ func (v *refcnt) release() bool {
 	return n == 0
 }
 
+func (v *refcnt) value() int32 {
+	return v.val.Load()
+}
+
 func (v *refcnt) trace(msg string) {
 	s := fmt.Sprintf("%s: refs=%d\n%s", msg, v.refs(), debug.Stack())
 	v.Lock()
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index d660c938..4f90dc01 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -178,7 +178,7 @@ type Iter struct {
 	// original trailer (eg, before any sequence-number zeroing or changes to
 	// key kind).
 	keyTrailer  base.InternalKeyTrailer
-	value       []byte
+	value       base.LazyValue
 	valueCloser io.Closer
 	// Temporary buffer used for storing the previous user key in order to
 	// determine when iteration has advanced to a new user key and thus a new
@@ -189,7 +189,6 @@ type Iter struct {
 	// advanced.
 	valueBuf         []byte
 	iterKV           *base.InternalKV
-	iterValue        []byte
 	iterStripeChange stripeChangeType
 	// skip indicates whether the remaining entries in the current snapshot
 	// stripe should be skipped or processed. `skip` has no effect when `pos ==
@@ -362,16 +361,12 @@ func (i *Iter) Stats() IterStats {
 }
 
 // First has the same semantics as InternalIterator.First.
-func (i *Iter) First() (*base.InternalKey, []byte) {
+func (i *Iter) First() (*base.InternalKey, base.LazyValue) {
 	if i.err != nil {
-		return nil, nil
+		return nil, base.LazyValue{}
 	}
 	i.iterKV = i.iter.First()
 	if i.iterKV != nil {
-		i.iterValue, _, i.err = i.iterKV.Value(nil)
-		if i.err != nil {
-			return nil, nil
-		}
 		i.curSnapshotIdx, i.curSnapshotSeqNum = i.cfg.Snapshots.IndexAndSeqNum(i.iterKV.SeqNum())
 	}
 	i.pos = iterPosNext
@@ -382,14 +377,14 @@ func (i *Iter) First() (*base.InternalKey, []byte) {
 // Next has the same semantics as InternalIterator.Next. Note that when Next
 // returns a RANGEDEL or a range key, the caller can use Span() to get the
 // corresponding span.
-func (i *Iter) Next() (*base.InternalKey, []byte) {
+func (i *Iter) Next() (*base.InternalKey, base.LazyValue) {
 	if i.err != nil {
-		return nil, nil
+		return nil, base.LazyValue{}
 	}
 
 	// Close the closer for the current value if one was open.
 	if i.closeValueCloser() != nil {
-		return nil, nil
+		return nil, base.LazyValue{}
 	}
 
 	// Prior to this call to `Next()` we are in one of three situations with
@@ -477,12 +472,15 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 			// This goes against the comment on i.key in the struct, and
 			// therefore warrants some investigation.
 			i.saveKey()
+			i.value = i.iterKV.V
+			if invariants.Enabled && !i.value.IsInPlaceValue() {
+				panic(errors.AssertionFailedf("pebble: span key's value is not in-place"))
+			}
 			// TODO(jackson): Handle tracking pinned statistics for range keys
 			// and range deletions. This would require updating
 			// emitRangeDelChunk and rangeKeyCompactionTransform to update
 			// statistics when they apply their own snapshot striping logic.
 			i.snapshotPinned = false
-			i.value = i.iterValue
 			return &i.key, i.value
 		}
 
@@ -545,7 +543,7 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 			switch i.iterKV.Kind() {
 			case base.InternalKeyKindDelete:
 				i.saveKey()
-				i.value = i.iterValue
+				i.value = base.LazyValue{} // DELs are value-less.
 				i.skip = true
 				return &i.key, i.value
 
@@ -559,7 +557,7 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 				if i.singleDeleteNext() {
 					return &i.key, i.value
 				} else if i.err != nil {
-					return nil, nil
+					return nil, base.LazyValue{} // SINGLEDELs are value-less.
 				}
 				continue
 
@@ -576,7 +574,7 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 			// kind.
 			i.setNext()
 			if i.err != nil {
-				return nil, nil
+				return nil, base.LazyValue{}
 			}
 			return &i.key, i.value
 
@@ -585,7 +583,8 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 			// advances the iterator, adjusting curSnapshotIdx.
 			origSnapshotIdx := i.curSnapshotIdx
 			var valueMerger base.ValueMerger
-			valueMerger, i.err = i.cfg.Merge(i.iterKV.K.UserKey, i.iterValue)
+			// MERGE values are always stored in-place.
+			valueMerger, i.err = i.cfg.Merge(i.iterKV.K.UserKey, i.iterKV.V.InPlaceValue())
 			if i.err == nil {
 				i.mergeNext(valueMerger)
 			}
@@ -607,7 +606,7 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 			if i.err == nil {
 				if needDelete {
 					if i.closeValueCloser() != nil {
-						return nil, nil
+						return nil, base.LazyValue{}
 					}
 					continue
 				}
@@ -620,15 +619,15 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 				// MERGE?
 				i.err = base.MarkCorruptionError(i.err)
 			}
-			return nil, nil
+			return nil, base.LazyValue{}
 
 		default:
 			i.err = base.CorruptionErrorf("invalid internal key kind: %d", errors.Safe(i.iterKV.Kind()))
-			return nil, nil
+			return nil, base.LazyValue{}
 		}
 	}
 
-	return nil, nil
+	return nil, base.LazyValue{}
 }
 
 // Span returns the range deletion or range key span corresponding to the
@@ -666,11 +665,8 @@ func (i *Iter) skipInStripe() {
 
 func (i *Iter) iterNext() bool {
 	i.iterKV = i.iter.Next()
-	if i.iterKV != nil {
-		i.iterValue, _, i.err = i.iterKV.Value(nil)
-		if i.err != nil {
-			i.iterKV = nil
-		}
+	if i.iterKV == nil {
+		i.err = i.iter.Error()
 	}
 	return i.iterKV != nil
 }
@@ -779,7 +775,7 @@ func (i *Iter) nextInStripeHelper() stripeChangeType {
 func (i *Iter) setNext() {
 	// Save the current key.
 	i.saveKey()
-	i.value = i.iterValue
+	i.value = i.iterKV.V
 	i.maybeZeroSeqnum(i.curSnapshotIdx)
 
 	// If this key is already a SETWITHDEL we can early return and skip the remaining
@@ -789,9 +785,8 @@ func (i *Iter) setNext() {
 		return
 	}
 
-	// We are iterating forward. Save the current value.
-	i.valueBuf = append(i.valueBuf[:0], i.iterValue...)
-	i.value = i.valueBuf
+	// We need to iterate forward. Save the current value so we don't lose it.
+	i.saveValue()
 
 	// Else, we continue to loop through entries in the stripe looking for a
 	// DEL. Note that we may stop *before* encountering a DEL, if one exists.
@@ -874,7 +869,16 @@ func (i *Iter) mergeNext(valueMerger base.ValueMerger) {
 			// value and return. We change the kind of the resulting key to a
 			// Set so that it shadows keys in lower levels. That is:
 			// MERGE + (SET*) -> SET.
-			i.err = valueMerger.MergeOlder(i.iterValue)
+			var v []byte
+			var callerOwned bool
+			v, callerOwned, i.err = i.iterKV.Value(i.valueBuf[:0])
+			if i.err != nil {
+				return
+			}
+			if callerOwned && cap(v) > cap(i.valueBuf) {
+				i.valueBuf = v
+			}
+			i.err = valueMerger.MergeOlder(v)
 			if i.err != nil {
 				return
 			}
@@ -885,7 +889,9 @@ func (i *Iter) mergeNext(valueMerger base.ValueMerger) {
 		case base.InternalKeyKindMerge:
 			// We've hit another Merge value. Merge with the existing value and
 			// continue looping.
-			i.err = valueMerger.MergeOlder(i.iterValue)
+			//
+			// MERGE values are always stored in-place.
+			i.err = valueMerger.MergeOlder(i.iterKV.InPlaceValue())
 			if i.err != nil {
 				return
 			}
@@ -910,7 +916,10 @@ func (i *Iter) mergeNext(valueMerger base.ValueMerger) {
 func (i *Iter) singleDeleteNext() bool {
 	// Save the current key.
 	i.saveKey()
-	i.value = i.iterValue
+	if invariants.Enabled && (!i.iterKV.V.IsInPlaceValue() || i.iterKV.V.Len() != 0) {
+		panic(errors.AssertionFailedf("pebble: single delete value is not in-place or is non-empty"))
+	}
+	i.value = base.LazyValue{} // SINGLEDELs are value-less.
 
 	// Loop until finds a key to be passed to the next level.
 	for {
@@ -1097,7 +1106,7 @@ func (i *Iter) skipDueToSingleDeleteElision() {
 //
 // When a deleteSizedNext is encountered, we skip ahead to see which keys, if
 // any, are elided as a result of the tombstone.
-func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
+func (i *Iter) deleteSizedNext() (*base.InternalKey, base.LazyValue) {
 	i.saveKey()
 	i.skip = true
 
@@ -1105,12 +1114,9 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
 	// tombstone has already deleted the key that the user originally predicted.
 	// In this case, we still peek forward in case there's another DELSIZED key
 	// with a lower sequence number, in which case we'll adopt its value.
-	if len(i.iterValue) == 0 {
-		i.value = i.valueBuf[:0]
-	} else {
-		i.valueBuf = append(i.valueBuf[:0], i.iterValue...)
-		i.value = i.valueBuf
-	}
+	// If the DELSIZED does have a value, it must be in-place.
+	i.valueBuf = append(i.valueBuf[:0], i.iterKV.V.InPlaceValue()...)
+	i.value = base.MakeInPlaceValue(i.valueBuf)
 
 	// Loop through all the keys within this stripe that are skippable.
 	i.pos = iterPosNext
@@ -1147,13 +1153,15 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
 			// We treat both cases the same functionally, adopting the identity
 			// of the lower-sequence numbered tombstone. However in the second
 			// case, we also increment the stat counting missized tombstones.
-			if len(i.value) > 0 {
+			if i.value.Len() > 0 {
 				// The original DELSIZED key was missized. The key that the user
 				// thought they were deleting does not exist.
 				i.stats.CountMissizedDels++
 			}
-			i.valueBuf = append(i.valueBuf[:0], i.iterValue...)
-			i.value = i.valueBuf
+			// If the tombstone has a value, it must be in-place. To save it, we
+			// can just copy the in-place value directly.
+			i.valueBuf = append(i.valueBuf[:0], i.iterKV.InPlaceValue()...)
+			i.value = base.MakeInPlaceValue(i.valueBuf)
 			if i.iterKV.Kind() != base.InternalKeyKindDeleteSized {
 				// Convert the DELSIZED to a DEL—The DEL/SINGLEDEL we're eliding
 				// may not have deleted the key(s) it was intended to yet. The
@@ -1185,7 +1193,7 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
 			// case has already been elided. We don't count it as a missizing,
 			// instead converting the DELSIZED to a DEL. Skip the remainder of
 			// the snapshot stripe and return.
-			if len(i.value) == 0 {
+			if i.value.Len() == 0 {
 				i.key.SetKind(base.InternalKeyKindDelete)
 				// NB: We skipInStripe now, rather than returning leaving
 				// i.skip=true and returning early, because Next() requires
@@ -1198,13 +1206,15 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
 				return &i.key, i.value
 			}
 			// The deleted key is not a DEL, DELSIZED, and the DELSIZED in i.key
-			// has a positive size.
-			expectedSize, n := binary.Uvarint(i.value)
-			if n != len(i.value) {
-				i.err = base.CorruptionErrorf("DELSIZED holds invalid value: %x", errors.Safe(i.value))
-				return nil, nil
+			// has a positive size. Note that the tombstone's value must be
+			// in-place.
+			v := i.value.InPlaceValue()
+			expectedSize, n := binary.Uvarint(v)
+			if n != len(v) {
+				i.err = base.CorruptionErrorf("DELSIZED holds invalid value: %x", errors.Safe(v))
+				return nil, base.LazyValue{}
 			}
-			elidedSize := uint64(len(i.iterKV.K.UserKey)) + uint64(len(i.iterValue))
+			elidedSize := uint64(len(i.iterKV.K.UserKey)) + uint64(i.iterKV.V.Len())
 			if elidedSize != expectedSize {
 				// The original DELSIZED key was missized. It's unclear what to
 				// do. The user-provided size was wrong, so it's unlikely to be
@@ -1222,7 +1232,7 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
 				// are safer.
 				i.stats.CountMissizedDels++
 				i.key.SetKind(base.InternalKeyKindDelete)
-				i.value = i.valueBuf[:0]
+				i.value = base.LazyValue{}
 				// NB: We skipInStripe now, rather than returning leaving
 				// i.skip=true and returning early, because Next() requires
 				// that i.skip=true only if i.iterPos = iterPosCurForward.
@@ -1236,11 +1246,11 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
 			// NB: We remove the value regardless of whether the key was sized
 			// appropriately. The size encoded is 'consumed' the first time it
 			// meets a key that it deletes.
-			i.value = i.valueBuf[:0]
+			i.value = base.LazyValue{}
 
 		default:
 			i.err = base.CorruptionErrorf("invalid internal key kind: %d", errors.Safe(i.iterKV.Kind()))
-			return nil, nil
+			return nil, base.LazyValue{}
 		}
 	}
 
@@ -1250,11 +1260,12 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, []byte) {
 	// We landed outside the original stripe. Reset skip.
 	i.skip = false
 	if i.err != nil {
-		return nil, nil
+		return nil, base.LazyValue{}
 	}
 	return &i.key, i.value
 }
 
+// saveKey saves the key in iterKV to i.key using i.keyBuf's memory.
 func (i *Iter) saveKey() {
 	i.keyBuf = append(i.keyBuf[:0], i.iterKV.K.UserKey...)
 	i.key = base.InternalKey{
@@ -1264,6 +1275,31 @@ func (i *Iter) saveKey() {
 	i.keyTrailer = i.key.Trailer
 }
 
+// saveValue saves the value in iterKV to i.value. It must be called before
+// stepping the iterator if the value needs to be retained. Unlike keys, values
+// do not need to be copied in all code paths. For example, a SETWITHDEL key may
+// be written to output sstables without needing to read ahead, copying the
+// value directly from the existing input sstable block into the output block
+// builder.
+//
+// If the value is in-place, this copies it into i.valueBuf. If the value is in
+// a value block, it retrieves the value from the block (possibly storing the
+// result into i.valueBuf).
+func (i *Iter) saveValue() {
+	// TODO(jackson): With the introduction of values stored in separate
+	// physical blob files, this should begin to Clone LazyValues that are
+	// stored in blob reference files when non-rewriting blob files.
+	v, callerOwned, err := i.iterKV.V.Value(i.valueBuf[:0])
+	if err != nil {
+		i.err = err
+		i.value = base.LazyValue{}
+	} else if !callerOwned {
+		i.value = base.MakeInPlaceValue(append(i.valueBuf[:0], v...))
+	} else {
+		i.value = base.MakeInPlaceValue(v)
+	}
+}
+
 // Error returns any error encountered.
 //
 // Note that Close will return the error as well.
@@ -1366,11 +1402,12 @@ func (i *Iter) maybeZeroSeqnum(snapshotIdx int) {
 
 func finishValueMerger(
 	valueMerger base.ValueMerger, includesBase bool,
-) (value []byte, needDelete bool, closer io.Closer, err error) {
+) (_ base.LazyValue, needDelete bool, closer io.Closer, err error) {
+	var value []byte
 	if valueMerger2, ok := valueMerger.(base.DeletableValueMerger); ok {
 		value, needDelete, closer, err = valueMerger2.DeletableFinish(includesBase)
 	} else {
 		value, closer, err = valueMerger.Finish(includesBase)
 	}
-	return
+	return base.MakeInPlaceValue(value), needDelete, closer, err
 }
diff --git a/internal/compact/run.go b/internal/compact/run.go
index a617bd29..8eaa3021 100644
--- a/internal/compact/run.go
+++ b/internal/compact/run.go
@@ -109,7 +109,7 @@ type Runner struct {
 	err error
 	// Last key/value returned by the compaction iterator.
 	key   *base.InternalKey
-	value []byte
+	value base.LazyValue
 	// Last RANGEDEL span (or portion of it) that was not yet written to a table.
 	lastRangeDelSpan keyspan.Span
 	// Last range key span (or portion of it) that was not yet written to a table.
@@ -152,7 +152,7 @@ func (r *Runner) WriteTable(objMeta objstorage.ObjectMetadata, tw sstable.RawWri
 	err = errors.CombineErrors(err, tw.Close())
 	if err != nil {
 		r.err = err
-		r.key, r.value = nil, nil
+		r.key, r.value = nil, base.LazyValue{}
 		return
 	}
 	writerMeta, err := tw.Metadata()
@@ -215,7 +215,11 @@ func (r *Runner) writeKeysToTable(tw sstable.RawWriter) (splitKey []byte, _ erro
 			r.lastRangeKeySpan.CopyFrom(r.iter.Span())
 			continue
 		}
-		if err := tw.AddWithForceObsolete(*key, value, r.iter.ForceObsoleteDueToRangeDel()); err != nil {
+		v, _, err := value.Value(nil)
+		if err != nil {
+			return nil, err
+		}
+		if err := tw.AddWithForceObsolete(*key, v, r.iter.ForceObsoleteDueToRangeDel()); err != nil {
 			return nil, err
 		}
 		if r.iter.SnapshotPinned() {
@@ -224,7 +228,7 @@ func (r *Runner) writeKeysToTable(tw sstable.RawWriter) (splitKey []byte, _ erro
 			// its elision. Increment the stats.
 			pinnedCount++
 			pinnedKeySize += uint64(len(key.UserKey)) + base.InternalTrailerLen
-			pinnedValueSize += uint64(len(value))
+			pinnedValueSize += uint64(len(v))
 		}
 	}
 	r.key, r.value = key, value
diff --git a/metamorphic/config.go b/metamorphic/config.go
index 04d941dc..050b9f34 100644
--- a/metamorphic/config.go
+++ b/metamorphic/config.go
@@ -4,7 +4,14 @@
 
 package metamorphic
 
-import "github.com/cockroachdb/pebble/internal/randvar"
+import (
+	"math/rand/v2"
+
+	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/randvar"
+	"github.com/cockroachdb/pebble/sstable"
+)
 
 // OpType is an enum of possible operation types.
 type OpType int
@@ -331,3 +338,68 @@ func mustDynamic(dyn randvar.Dynamic, err error) randvar.Dynamic {
 	}
 	return dyn
 }
+
+var knownKeyFormats = []KeyFormat{
+	TestkeysKeyFormat,
+}
+
+var keyFormatsByName = func() map[string]KeyFormat {
+	m := make(map[string]KeyFormat)
+	for _, kf := range knownKeyFormats {
+		m[kf.Name] = kf
+	}
+	return m
+}()
+
+// A KeyFormat dictates the format of key-value pairs uses by the metamorphic
+// test.
+type KeyFormat struct {
+	Name                         string
+	Comparer                     *base.Comparer
+	FormatKey                    func(UserKey) string
+	FormatKeySuffix              func(UserKeySuffix) string
+	ParseFormattedKey            func(string) (UserKey, error)
+	ParseFormattedKeySuffix      func(string) (UserKeySuffix, error)
+	NewGenerator                 func(*keyManager, *rand.Rand, OpConfig) KeyGenerator
+	NewSuffixFilterMask          func() pebble.BlockPropertyFilterMask
+	NewSuffixBlockPropertyFilter func(min []byte, max []byte) sstable.BlockPropertyFilter
+}
+
+// KeyGenerator is an interface for generating keys, prefixes and suffixes.
+type KeyGenerator interface {
+	// IncMaxSuffix increases the max suffix range and returns the new maximum
+	// suffix (which is guaranteed to be larger than any previously generated
+	// suffix).
+	IncMaxSuffix() []byte
+	// RandKey returns a random key (either a previously known key, or a new
+	// key). The provided probability determines the likelihood of generating a
+	// new key.
+	RandKey(newKeyProbability float64) []byte
+	// RandKeyInRange returns a random key (either a previously known key, or a
+	// new key) in the given key range.
+	RandKeyInRange(newKeyProbability float64, kr pebble.KeyRange) []byte
+	// RandPrefix returns a random prefix key (a key with no suffix). The
+	// provided probability determines the likelihood of generating a new
+	// prefix.
+	RandPrefix(newPrefix float64) []byte
+	// SkewedSuffix generates a random suffix according to the configuration's
+	// suffix distribution. It takes a probability 0 ≤ p ≤ 1.0 indicating the
+	// probability with which the generator should increase the max suffix
+	// generated by the generator.
+	//
+	// May return a nil suffix, with the probability the configuration's suffix
+	// distribution assigns to the zero suffix.
+	SkewedSuffix(incMaxProb float64) []byte
+	// SuffixRange generates a new uniformly random range of suffixes [low,
+	// high) such that high is guaranteed to be strictly greater than low.
+	SuffixRange() (low, high []byte)
+	// UniformSuffix returns a suffix in the same range as SkewedSuffix but with
+	// a uniform distribution. This is used during reads to better exercise
+	// reading a mix of older and newer keys. The suffix can be empty.
+	//
+	// May return a nil suffix.
+	UniformSuffix() []byte
+	// UniqueKeys takes a key-generating function and uses it to generate n
+	// unique keys, returning them in sorted order.
+	UniqueKeys(n int, genFn func() []byte) [][]byte
+}
diff --git a/metamorphic/diagram.go b/metamorphic/diagram.go
index dc05a179..8d5e4821 100644
--- a/metamorphic/diagram.go
+++ b/metamorphic/diagram.go
@@ -11,8 +11,11 @@ import "strings"
 //
 // If the diagram would be too large to be practical, returns the empty string
 // (with no error).
-func TryToGenerateDiagram(opsData []byte) (string, error) {
-	ops, err := parse(opsData, parserOpts{})
+func TryToGenerateDiagram(keyFormat KeyFormat, opsData []byte) (string, error) {
+	ops, err := parse(opsData, parserOpts{
+		parseFormattedUserKey:       keyFormat.ParseFormattedKey,
+		parseFormattedUserKeySuffix: keyFormat.ParseFormattedKeySuffix,
+	})
 	if err != nil {
 		return "", err
 	}
@@ -29,7 +32,7 @@ func TryToGenerateDiagram(opsData []byte) (string, error) {
 	if len(keySet) == 0 {
 		return "", nil
 	}
-	keys := sortedKeys(keySet)
+	keys := sortedKeys(keyFormat.Comparer.Compare, keySet)
 	axis1, axis2, pos := genAxis(keys)
 	if len(axis1) > 200 {
 		return "", nil
@@ -56,7 +59,7 @@ func TryToGenerateDiagram(opsData []byte) (string, error) {
 		for row.Len() <= len(axis1) {
 			row.WriteByte(' ')
 		}
-		row.WriteString(o.String())
+		row.WriteString(o.formattedString(keyFormat))
 
 		rows = append(rows, row.String())
 	}
diff --git a/metamorphic/generator.go b/metamorphic/generator.go
index bc654f12..b31eb711 100644
--- a/metamorphic/generator.go
+++ b/metamorphic/generator.go
@@ -27,7 +27,7 @@ type iterOpts struct {
 	// configure IterOptions.RangeKeyMasking.Suffix.
 	maskSuffix []byte
 
-	// If filterMax is >0, this iterator will filter out any keys that have
+	// If filterMax is != nil, this iterator will filter out any keys that have
 	// suffixes that don't fall within the range [filterMin,filterMax).
 	// Additionally, the iterator will be constructed with a block-property
 	// filter that filters out blocks accordingly. Not all OPTIONS hook up the
@@ -35,8 +35,8 @@ type iterOpts struct {
 	// effectively disabled in some runs. The iterator operations themselves
 	// however will always skip past any points that should be filtered to
 	// ensure determinism.
-	filterMin uint64
-	filterMax uint64
+	filterMin []byte
+	filterMax []byte
 
 	// see IterOptions.UseL6Filters.
 	useL6Filters bool
@@ -46,7 +46,7 @@ type iterOpts struct {
 
 func (o iterOpts) IsZero() bool {
 	return o.lower == nil && o.upper == nil && o.keyTypes == 0 &&
-		o.maskSuffix == nil && o.filterMin == 0 && o.filterMax == 0 && !o.useL6Filters
+		o.maskSuffix == nil && o.filterMin == nil && o.filterMax == nil && !o.useL6Filters
 }
 
 // GenerateOps generates n random operations, drawing randomness from the
@@ -67,7 +67,7 @@ type generator struct {
 
 	// keyManager tracks the state of keys a operation generation time.
 	keyManager   *keyManager
-	keyGenerator *keyGenerator
+	keyGenerator KeyGenerator
 	dbs          objIDSlice
 	// Unordered sets of object IDs for live objects. Used to randomly select on
 	// object when generating an operation. There are 4 concrete objects: the DB
@@ -119,7 +119,7 @@ type generator struct {
 }
 
 func newGenerator(rng *rand.Rand, cfg OpConfig, km *keyManager) *generator {
-	keyGenerator := newKeyGenerator(km, rng, cfg)
+	keyGenerator := km.kf.NewGenerator(km, rng, cfg)
 	g := &generator{
 		cfg:             cfg,
 		rng:             rng,
@@ -207,7 +207,7 @@ func generate(rng *rand.Rand, count uint64, cfg OpConfig, km *keyManager) []op {
 
 	defer func() {
 		if r := recover(); r != nil {
-			fmt.Fprintln(os.Stderr, formatOps(g.ops))
+			fmt.Fprintln(os.Stderr, formatOps(km.kf, g.ops))
 			panic(r)
 		}
 	}()
@@ -531,13 +531,7 @@ func (g *generator) newIter() {
 	// block-property filtering and explicitly within the iterator operations to
 	// ensure determinism.
 	if g.rng.Float64() <= 0.1 {
-		opts.filterMin = uint64(g.keyGenerator.UniformSuffixInt() + 1)
-		opts.filterMax = uint64(g.keyGenerator.UniformSuffixInt() + 1)
-		if opts.filterMin > opts.filterMax {
-			opts.filterMin, opts.filterMax = opts.filterMax, opts.filterMin
-		} else if opts.filterMin == opts.filterMax {
-			opts.filterMax++
-		}
+		opts.filterMin, opts.filterMax = g.keyGenerator.SuffixRange()
 	}
 
 	// Enable L6 filters with a 10% probability.
@@ -839,7 +833,7 @@ func (g *generator) iterSeekPrefixGE(iterID objID) {
 	}
 	// Sometimes limit the key to just the prefix.
 	if g.rng.IntN(3) == 1 {
-		key = g.keyManager.comparer.Split.Prefix(key)
+		key = g.keyManager.kf.Comparer.Split.Prefix(key)
 	}
 	g.add(&iterSeekPrefixGEOp{
 		iterID:          iterID,
@@ -1527,18 +1521,12 @@ func (g *generator) maybeMutateOptions(readerID objID, opts *iterOpts) {
 		}
 
 		// With 1/3 probability, clear existing filter.
-		if opts.filterMax > 0 && g.rng.IntN(3) == 0 {
-			opts.filterMax, opts.filterMin = 0, 0
+		if opts.filterMax != nil && g.rng.IntN(3) == 0 {
+			opts.filterMax, opts.filterMin = nil, nil
 		}
 		// With 10% probability, set a filter range.
 		if g.rng.IntN(10) == 1 {
-			opts.filterMin = uint64(g.keyGenerator.UniformSuffixInt() + 1)
-			opts.filterMax = uint64(g.keyGenerator.UniformSuffixInt() + 1)
-			if opts.filterMin > opts.filterMax {
-				opts.filterMin, opts.filterMax = opts.filterMax, opts.filterMin
-			} else if opts.filterMin == opts.filterMax {
-				opts.filterMax = opts.filterMin + 1
-			}
+			opts.filterMin, opts.filterMax = g.keyGenerator.SuffixRange()
 		}
 		// With 10% probability, flip enablement of L6 filters.
 		if g.rng.Float64() <= 0.1 {
@@ -1548,11 +1536,11 @@ func (g *generator) maybeMutateOptions(readerID objID, opts *iterOpts) {
 }
 
 func (g *generator) cmp(a, b []byte) int {
-	return g.keyManager.comparer.Compare(a, b)
+	return g.keyManager.kf.Comparer.Compare(a, b)
 }
 
 func (g *generator) prefix(a []byte) []byte {
-	n := g.keyManager.comparer.Split(a)
+	n := g.keyManager.kf.Comparer.Split(a)
 	return a[:n:n]
 }
 
diff --git a/metamorphic/key_manager.go b/metamorphic/key_manager.go
index 4373ad35..cbc00219 100644
--- a/metamorphic/key_manager.go
+++ b/metamorphic/key_manager.go
@@ -181,7 +181,7 @@ func (b *bounds) Expand(cmp base.Compare, other bounds) {
 // that we know at generation-time whether or not an ingestion operation will
 // fail and can avoid updating key state.
 type keyManager struct {
-	comparer *base.Comparer
+	kf KeyFormat
 
 	byObj map[objID]*objKeyMeta
 	// globalKeys represents all the keys that have been generated so far. Not
@@ -272,7 +272,7 @@ func (k *keyManager) SortedKeysForObj(o objID) []keyMeta {
 		res = append(res, *m)
 	}
 	slices.SortFunc(res, func(a, b keyMeta) int {
-		cmp := k.comparer.Compare(a.key, b.key)
+		cmp := k.kf.Comparer.Compare(a.key, b.key)
 		if cmp == 0 {
 			panic(fmt.Sprintf("distinct keys %q and %q compared as equal", a.key, b.key))
 		}
@@ -286,8 +286,8 @@ func (k *keyManager) SortedKeysForObj(o objID) []keyMeta {
 func (k *keyManager) InRangeKeysForObj(o objID, lower, upper []byte) []keyMeta {
 	var inRangeKeys []keyMeta
 	for _, km := range k.SortedKeysForObj(o) {
-		if (lower == nil || k.comparer.Compare(km.key, lower) >= 0) &&
-			(upper == nil || k.comparer.Compare(km.key, upper) < 0) {
+		if (lower == nil || k.kf.Comparer.Compare(km.key, lower) >= 0) &&
+			(upper == nil || k.kf.Comparer.Compare(km.key, upper) < 0) {
 			inRangeKeys = append(inRangeKeys, km)
 		}
 	}
@@ -306,22 +306,23 @@ func (k *keyManager) KeysForExternalIngest(obj externalObjWithBounds) []keyMeta
 			km.key = obj.syntheticPrefix.Apply(km.key)
 		}
 		if obj.syntheticSuffix.IsSet() {
-			n := k.comparer.Split(km.key)
+			n := k.kf.Comparer.Split(km.key)
 			km.key = append(km.key[:n:n], obj.syntheticSuffix...)
 		}
-		if lastPrefix != nil && k.comparer.Equal(lastPrefix, km.key[:k.comparer.Split(km.key)]) {
+		if lastPrefix != nil && k.kf.Comparer.Equal(lastPrefix, km.key[:k.kf.Comparer.Split(km.key)]) {
 			// We only keep the first of every unique prefix for external ingests.
 			// See the use of uniquePrefixes in newExternalObjOp.
 			continue
 		}
-		lastPrefix = append(lastPrefix[:0], km.key[:k.comparer.Split(km.key)]...)
-		if k.comparer.Compare(km.key, obj.bounds.Start) >= 0 && k.comparer.Compare(km.key, obj.bounds.End) < 0 {
+		lastPrefix = append(lastPrefix[:0], km.key[:k.kf.Comparer.Split(km.key)]...)
+		if k.kf.Comparer.Compare(km.key, obj.bounds.Start) >= 0 &&
+			k.kf.Comparer.Compare(km.key, obj.bounds.End) < 0 {
 			res = append(res, km)
 		}
 	}
 	// Check for duplicate resulting keys.
 	for i := 1; i < len(res); i++ {
-		if k.comparer.Compare(res[i].key, res[i-1].key) == 0 {
+		if k.kf.Comparer.Compare(res[i].key, res[i-1].key) == 0 {
 			panic(fmt.Sprintf("duplicate external ingest key %q", res[i].key))
 		}
 	}
@@ -336,11 +337,11 @@ func (k *keyManager) ExternalObjectHasOverlappingRangeKeySets(externalObjID objI
 	ranges := meta.rangeKeySets
 	// Sort by start key.
 	slices.SortFunc(ranges, func(a, b pebble.KeyRange) int {
-		return k.comparer.Compare(a.Start, b.Start)
+		return k.kf.Comparer.Compare(a.Start, b.Start)
 	})
 	// Check overlap between adjacent ranges.
 	for i := 0; i < len(ranges)-1; i++ {
-		if ranges[i].OverlapsKeyRange(k.comparer.Compare, ranges[i+1]) {
+		if ranges[i].OverlapsKeyRange(k.kf.Comparer.Compare, ranges[i+1]) {
 			return true
 		}
 	}
@@ -365,7 +366,7 @@ func (k *keyManager) getSetOfVisibleKeys(readerID objID) [][]byte {
 // interact with this using addNewKey, knownKeys, update methods only.
 func newKeyManager(numInstances int) *keyManager {
 	m := &keyManager{
-		comparer:             testkeys.Comparer,
+		kf:                   TestkeysKeyFormat,
 		byObj:                make(map[objID]*objKeyMeta),
 		globalKeysMap:        make(map[string]bool),
 		globalKeyPrefixesMap: make(map[string]struct{}),
@@ -382,15 +383,15 @@ func (k *keyManager) addNewKey(key []byte) bool {
 	if k.globalKeysMap[string(key)] {
 		return false
 	}
-	insertSorted(k.comparer.Compare, &k.globalKeys, key)
+	insertSorted(k.kf.Comparer.Compare, &k.globalKeys, key)
 	k.globalKeysMap[string(key)] = true
 
-	prefixLen := k.comparer.Split(key)
+	prefixLen := k.kf.Comparer.Split(key)
 	if prefixLen == 0 {
 		panic(fmt.Sprintf("key %q has zero length prefix", key))
 	}
 	if _, ok := k.globalKeyPrefixesMap[string(key[:prefixLen])]; !ok {
-		insertSorted(k.comparer.Compare, &k.globalKeyPrefixes, key[:prefixLen])
+		insertSorted(k.kf.Comparer.Compare, &k.globalKeyPrefixes, key[:prefixLen])
 		k.globalKeyPrefixesMap[string(key[:prefixLen])] = struct{}{}
 	}
 	return true
@@ -411,7 +412,7 @@ func (k *keyManager) getOrInit(id objID, key []byte) *keyMeta {
 	// Initialize the key-to-meta index.
 	objKeys.keys[string(key)] = m
 	// Expand the object's bounds to contain this key if they don't already.
-	objKeys.bounds.Expand(k.comparer.Compare, k.makeSingleKeyBounds(key))
+	objKeys.bounds.Expand(k.kf.Comparer.Compare, k.makeSingleKeyBounds(key))
 	return m
 }
 
@@ -419,7 +420,7 @@ func (k *keyManager) getOrInit(id objID, key []byte) *keyMeta {
 // deletes the metadata for the source object (which must not be used again).
 func (k *keyManager) mergeObjectInto(from, to objID) {
 	toMeta := k.objKeyMeta(to)
-	toMeta.MergeFrom(k.objKeyMeta(from), k.comparer.Compare)
+	toMeta.MergeFrom(k.objKeyMeta(from), k.kf.Comparer.Compare)
 
 	delete(k.byObj, from)
 }
@@ -427,7 +428,7 @@ func (k *keyManager) mergeObjectInto(from, to objID) {
 // expandBounds expands the incrementally maintained bounds of o to be at least
 // as wide as `b`.
 func (k *keyManager) expandBounds(o objID, b bounds) {
-	k.objKeyMeta(o).bounds.Expand(k.comparer.Compare, b)
+	k.objKeyMeta(o).bounds.Expand(k.kf.Comparer.Compare, b)
 }
 
 // doObjectBoundsOverlap returns true iff any of the named objects have key
@@ -443,7 +444,7 @@ func (k *keyManager) doObjectBoundsOverlap(objIDs []objID) bool {
 			if !jok {
 				continue
 			}
-			if ib.bounds.Overlaps(k.comparer.Compare, jb.bounds) {
+			if ib.bounds.Overlaps(k.kf.Comparer.Compare, jb.bounds) {
 				return true
 			}
 		}
@@ -699,7 +700,7 @@ func (k *keyManager) update(o op) {
 			for _, keyMeta := range k.KeysForExternalIngest(obj) {
 				dbMeta.MergeKey(&keyMeta)
 			}
-			dbMeta.bounds.Expand(k.comparer.Compare, k.makeEndExclusiveBounds(obj.bounds.Start, obj.bounds.End))
+			dbMeta.bounds.Expand(k.kf.Comparer.Compare, k.makeEndExclusiveBounds(obj.bounds.Start, obj.bounds.End))
 		}
 	case *applyOp:
 		// Merge the keys from this batch into the parent writer.
@@ -718,8 +719,8 @@ func (k *keyManager) knownKeys() (keys [][]byte) {
 // [start,end). The returned slice is owned by the keyManager and must not be
 // retained.
 func (k *keyManager) knownKeysInRange(kr pebble.KeyRange) (keys [][]byte) {
-	s, _ := slices.BinarySearchFunc(k.globalKeys, kr.Start, k.comparer.Compare)
-	e, _ := slices.BinarySearchFunc(k.globalKeys, kr.End, k.comparer.Compare)
+	s, _ := slices.BinarySearchFunc(k.globalKeys, kr.Start, k.kf.Comparer.Compare)
+	e, _ := slices.BinarySearchFunc(k.globalKeys, kr.End, k.kf.Comparer.Compare)
 	if s >= e {
 		return nil
 	}
@@ -775,7 +776,7 @@ func (k *keyManager) makeEndExclusiveBounds(smallest, largest []byte) bounds {
 		largest:     largest,
 		largestExcl: true,
 	}
-	b.checkValid(k.comparer.Compare)
+	b.checkValid(k.kf.Comparer.Compare)
 	return b
 }
 
@@ -919,7 +920,7 @@ func loadPrecedingKeys(t TestingT, ops []op, cfg *OpConfig, m *keyManager) {
 
 			// If the key has a suffix, ratchet up the suffix distribution if
 			// necessary.
-			if s := m.comparer.Split(k); s < len(k) {
+			if s := m.kf.Comparer.Split(k); s < len(k) {
 				suffix, err := testkeys.ParseSuffix(k[s:])
 				require.NoError(t, err)
 				if uint64(suffix) > cfg.writeSuffixDist.Max() {
diff --git a/metamorphic/meta.go b/metamorphic/meta.go
index 72a9a0b9..c968e1ed 100644
--- a/metamorphic/meta.go
+++ b/metamorphic/meta.go
@@ -198,7 +198,7 @@ func RunAndCompare(t *testing.T, rootDir string, rOpts ...RunOption) {
 	}
 	ops := generate(rng, opCount, cfg, km)
 	opsPath := filepath.Join(metaDir, "ops")
-	formattedOps := formatOps(ops)
+	formattedOps := formatOps(km.kf, ops)
 	require.NoError(t, os.WriteFile(opsPath, []byte(formattedOps), 0644))
 
 	// runOptions performs a particular test run with the specified options. The
@@ -470,9 +470,6 @@ func RunOnce(t TestingT, runDir string, seed uint64, historyPath string, rOpts .
 	opsData, err := os.ReadFile(opsPath)
 	require.NoError(t, err)
 
-	ops, err := parse(opsData, parserOpts{})
-	require.NoError(t, err)
-
 	optionsPath := filepath.Join(runDir, "OPTIONS")
 	optionsData, err := os.ReadFile(optionsPath)
 	require.NoError(t, err)
@@ -484,6 +481,12 @@ func RunOnce(t TestingT, runDir string, seed uint64, historyPath string, rOpts .
 	opts := testOpts.Opts
 	require.NoError(t, parseOptions(testOpts, string(optionsData), runOpts.customOptionParsers))
 
+	ops, err := parse(opsData, parserOpts{
+		parseFormattedUserKey:       testOpts.KeyFormat.ParseFormattedKey,
+		parseFormattedUserKeySuffix: testOpts.KeyFormat.ParseFormattedKeySuffix,
+	})
+	require.NoError(t, err)
+
 	// Set up the filesystem to use for the test. Note that by default we use an
 	// in-memory FS.
 	if testOpts.useDisk {
diff --git a/metamorphic/ops.go b/metamorphic/ops.go
index d6455d49..94503121 100644
--- a/metamorphic/ops.go
+++ b/metamorphic/ops.go
@@ -22,7 +22,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/private"
 	"github.com/cockroachdb/pebble/internal/rangekey"
-	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
@@ -35,7 +34,7 @@ type Ops []op
 // op defines the interface for a single operation, such as creating a batch,
 // or advancing an iterator.
 type op interface {
-	String() string
+	formattedString(KeyFormat) string
 
 	run(t *Test, h historyRecorder)
 
@@ -51,17 +50,25 @@ type op interface {
 	// that return these objects from their own syncObjs or receiver methods.
 	syncObjs() objIDSlice
 
-	// keys returns all user keys used by the operation, as pointers to slices.
-	// The caller can then modify these slices to rewrite the keys.
+	// rewriteKeys invokes fn for every user key used by the operation. The key
+	// is updated to be the result of fn(key).
 	//
 	// Used for simplification of operations for easier investigations.
-	keys() []*[]byte
+	rewriteKeys(fn func(UserKey) UserKey)
 
 	// diagramKeyRanges() returns key spans associated with this operation, to be
 	// shown on an ASCII diagram of operations.
 	diagramKeyRanges() []pebble.KeyRange
 }
 
+// A UserKey is a user key used by the metamorphic test. The format is
+// determined by the KeyFormat used by the test.
+type UserKey []byte
+
+// A UserKeySuffix is the suffix of a user key used by the metamorphic test. The
+// format is determined by the KeyFormat used by the test.
+type UserKeySuffix []byte
+
 // initOp performs test initialization
 type initOp struct {
 	dbSlots          uint32
@@ -76,10 +83,10 @@ func (o *initOp) run(t *Test, h historyRecorder) {
 	t.iters = make([]*retryableIter, o.iterSlots)
 	t.snapshots = make([]readerCloser, o.snapshotSlots)
 	t.externalObjs = make([]externalObjMeta, o.externalObjSlots)
-	h.Recordf("%s", o)
+	h.Recordf("%s", o.formattedString(t.testOpts.KeyFormat))
 }
 
-func (o *initOp) String() string {
+func (o *initOp) formattedString(kf KeyFormat) string {
 	return fmt.Sprintf("Init(%d /* dbs */, %d /* batches */, %d /* iters */, %d /* snapshots */, %d /* externalObjs */)",
 		o.dbSlots, o.batchSlots, o.iterSlots, o.snapshotSlots, o.externalObjSlots)
 }
@@ -94,7 +101,7 @@ func (o *initOp) syncObjs() objIDSlice {
 	return syncObjs
 }
 
-func (o *initOp) keys() []*[]byte                     { return nil }
+func (o *initOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *initOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // applyOp models a Writer.Apply operation.
@@ -115,11 +122,13 @@ func (o *applyOp) run(t *Test, h historyRecorder) {
 	} else {
 		err = w.Apply(b, t.writeOpts)
 	}
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 	// batch will be closed by a closeOp which is guaranteed to be generated
 }
 
-func (o *applyOp) String() string  { return fmt.Sprintf("%s.Apply(%s)", o.writerID, o.batchID) }
+func (o *applyOp) formattedString(KeyFormat) string {
+	return fmt.Sprintf("%s.Apply(%s)", o.writerID, o.batchID)
+}
 func (o *applyOp) receiver() objID { return o.writerID }
 func (o *applyOp) syncObjs() objIDSlice {
 	// Apply should not be concurrent with operations that are mutating the
@@ -127,7 +136,7 @@ func (o *applyOp) syncObjs() objIDSlice {
 	return []objID{o.batchID}
 }
 
-func (o *applyOp) keys() []*[]byte                     { return nil }
+func (o *applyOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *applyOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // checkpointOp models a DB.Checkpoint operation.
@@ -145,7 +154,7 @@ func (o *checkpointOp) run(t *Test, h historyRecorder) {
 	// not sure how to do that easily:
 	// https://github.com/cockroachdb/pebble/blob/master/metamorphic/meta.go#L177
 	if t.testOpts.sharedStorageEnabled || t.testOpts.externalStorageEnabled {
-		h.Recordf("%s // %v", o, nil)
+		h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), nil)
 		return
 	}
 	var opts []pebble.CheckpointOption
@@ -156,33 +165,31 @@ func (o *checkpointOp) run(t *Test, h historyRecorder) {
 	err := t.withRetries(func() error {
 		return db.Checkpoint(o.dir(t.dir, h.op), opts...)
 	})
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
 func (o *checkpointOp) dir(dataDir string, idx int) string {
 	return filepath.Join(dataDir, "checkpoints", fmt.Sprintf("op-%06d", idx))
 }
 
-func (o *checkpointOp) String() string {
+func (o *checkpointOp) formattedString(kf KeyFormat) string {
 	var spanStr bytes.Buffer
 	for i, span := range o.spans {
 		if i > 0 {
 			spanStr.WriteString(",")
 		}
-		fmt.Fprintf(&spanStr, "%q,%q", span.Start, span.End)
+		fmt.Fprintf(&spanStr, "%q,%q", kf.FormatKey(span.Start), kf.FormatKey(span.End))
 	}
 	return fmt.Sprintf("%s.Checkpoint(%s)", o.dbID, spanStr.String())
 }
 
 func (o *checkpointOp) receiver() objID      { return o.dbID }
 func (o *checkpointOp) syncObjs() objIDSlice { return nil }
-
-func (o *checkpointOp) keys() []*[]byte {
-	var res []*[]byte
+func (o *checkpointOp) rewriteKeys(fn func(UserKey) UserKey) {
 	for i := range o.spans {
-		res = append(res, &o.spans[i].Start, &o.spans[i].End)
+		o.spans[i].Start = fn(o.spans[i].Start)
+		o.spans[i].End = fn(o.spans[i].End)
 	}
-	return res
 }
 
 func (o *checkpointOp) diagramKeyRanges() []pebble.KeyRange {
@@ -204,24 +211,26 @@ type downloadOp struct {
 
 func (o *downloadOp) run(t *Test, h historyRecorder) {
 	if t.testOpts.disableDownloads {
-		h.Recordf("%s // %v", o, nil)
+		h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), nil)
 		return
 	}
 	db := t.getDB(o.dbID)
 	err := t.withRetries(func() error {
 		return db.Download(context.Background(), o.spans)
 	})
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *downloadOp) String() string {
+func (o *downloadOp) formattedString(kf KeyFormat) string {
 	var spanStr bytes.Buffer
 	for i, span := range o.spans {
 		if i > 0 {
 			spanStr.WriteString(", ")
 		}
 		fmt.Fprintf(&spanStr, "%q /* start */, %q /* end */, %v /* viaBackingFileDownload */",
-			span.StartKey, span.EndKey, span.ViaBackingFileDownload)
+			kf.FormatKey(span.StartKey),
+			kf.FormatKey(span.EndKey),
+			span.ViaBackingFileDownload)
 	}
 	return fmt.Sprintf("%s.Download(%s)", o.dbID, spanStr.String())
 }
@@ -229,12 +238,11 @@ func (o *downloadOp) String() string {
 func (o *downloadOp) receiver() objID     { return o.dbID }
 func (o downloadOp) syncObjs() objIDSlice { return nil }
 
-func (o *downloadOp) keys() []*[]byte {
-	var res []*[]byte
+func (o *downloadOp) rewriteKeys(fn func(UserKey) UserKey) {
 	for i := range o.spans {
-		res = append(res, &o.spans[i].StartKey, &o.spans[i].EndKey)
+		o.spans[i].StartKey = fn(o.spans[i].StartKey)
+		o.spans[i].EndKey = fn(o.spans[i].EndKey)
 	}
-	return res
 }
 
 func (o *downloadOp) diagramKeyRanges() []pebble.KeyRange {
@@ -268,23 +276,23 @@ func (o *closeOp) run(t *Test, h historyRecorder) {
 	}
 	t.clearObj(o.objID)
 	err := c.Close()
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *closeOp) String() string  { return fmt.Sprintf("%s.Close()", o.objID) }
-func (o *closeOp) receiver() objID { return o.objID }
+func (o *closeOp) formattedString(KeyFormat) string { return fmt.Sprintf("%s.Close()", o.objID) }
+func (o *closeOp) receiver() objID                  { return o.objID }
 func (o *closeOp) syncObjs() objIDSlice {
 	return o.affectedObjects
 }
 
-func (o *closeOp) keys() []*[]byte                     { return nil }
+func (o *closeOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *closeOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // compactOp models a DB.Compact operation.
 type compactOp struct {
 	dbID        objID
-	start       []byte
-	end         []byte
+	start       UserKey
+	end         UserKey
 	parallelize bool
 }
 
@@ -292,18 +300,20 @@ func (o *compactOp) run(t *Test, h historyRecorder) {
 	err := t.withRetries(func() error {
 		return t.getDB(o.dbID).Compact(o.start, o.end, o.parallelize)
 	})
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *compactOp) String() string {
-	return fmt.Sprintf("%s.Compact(%q, %q, %t /* parallelize */)", o.dbID, o.start, o.end, o.parallelize)
+func (o *compactOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Compact(%q, %q, %t /* parallelize */)",
+		o.dbID, kf.FormatKey(o.start), kf.FormatKey(o.end), o.parallelize)
 }
 
 func (o *compactOp) receiver() objID      { return o.dbID }
 func (o *compactOp) syncObjs() objIDSlice { return nil }
 
-func (o *compactOp) keys() []*[]byte {
-	return []*[]byte{&o.start, &o.end}
+func (o *compactOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.start = fn(o.start)
+	o.end = fn(o.end)
 }
 
 func (o *compactOp) diagramKeyRanges() []pebble.KeyRange {
@@ -313,7 +323,7 @@ func (o *compactOp) diagramKeyRanges() []pebble.KeyRange {
 // deleteOp models a Write.Delete operation.
 type deleteOp struct {
 	writerID objID
-	key      []byte
+	key      UserKey
 
 	derivedDBID objID
 }
@@ -328,7 +338,7 @@ func (o *deleteOp) run(t *Test, h historyRecorder) {
 	} else {
 		err = w.Delete(o.key, t.writeOpts)
 	}
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
 func hashSize(index int) uint32 {
@@ -336,14 +346,14 @@ func hashSize(index int) uint32 {
 	return uint32((11400714819323198485 * uint64(index)) % maxValueSize)
 }
 
-func (o *deleteOp) String() string {
-	return fmt.Sprintf("%s.Delete(%q)", o.writerID, o.key)
+func (o *deleteOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Delete(%q)", o.writerID, kf.FormatKey(o.key))
 }
 func (o *deleteOp) receiver() objID      { return o.writerID }
 func (o *deleteOp) syncObjs() objIDSlice { return nil }
 
-func (o *deleteOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *deleteOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *deleteOp) diagramKeyRanges() []pebble.KeyRange {
@@ -353,7 +363,7 @@ func (o *deleteOp) diagramKeyRanges() []pebble.KeyRange {
 // singleDeleteOp models a Write.SingleDelete operation.
 type singleDeleteOp struct {
 	writerID           objID
-	key                []byte
+	key                UserKey
 	maybeReplaceDelete bool
 }
 
@@ -370,18 +380,19 @@ func (o *singleDeleteOp) run(t *Test, h historyRecorder) {
 	// or not the delete *could* have been replaced. The OPTIONS file should
 	// also be consulted to determine what happened at runtime (i.e. by taking
 	// the logical AND).
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *singleDeleteOp) String() string {
-	return fmt.Sprintf("%s.SingleDelete(%q, %v /* maybeReplaceDelete */)", o.writerID, o.key, o.maybeReplaceDelete)
+func (o *singleDeleteOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.SingleDelete(%q, %v /* maybeReplaceDelete */)",
+		o.writerID, kf.FormatKey(o.key), o.maybeReplaceDelete)
 }
 
 func (o *singleDeleteOp) receiver() objID      { return o.writerID }
 func (o *singleDeleteOp) syncObjs() objIDSlice { return nil }
 
-func (o *singleDeleteOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *singleDeleteOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *singleDeleteOp) diagramKeyRanges() []pebble.KeyRange {
@@ -391,25 +402,27 @@ func (o *singleDeleteOp) diagramKeyRanges() []pebble.KeyRange {
 // deleteRangeOp models a Write.DeleteRange operation.
 type deleteRangeOp struct {
 	writerID objID
-	start    []byte
-	end      []byte
+	start    UserKey
+	end      UserKey
 }
 
 func (o *deleteRangeOp) run(t *Test, h historyRecorder) {
 	w := t.getWriter(o.writerID)
 	err := w.DeleteRange(o.start, o.end, t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *deleteRangeOp) String() string {
-	return fmt.Sprintf("%s.DeleteRange(%q, %q)", o.writerID, o.start, o.end)
+func (o *deleteRangeOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.DeleteRange(%q, %q)",
+		o.writerID, kf.FormatKey(o.start), kf.FormatKey(o.end))
 }
 
 func (o *deleteRangeOp) receiver() objID      { return o.writerID }
 func (o *deleteRangeOp) syncObjs() objIDSlice { return nil }
 
-func (o *deleteRangeOp) keys() []*[]byte {
-	return []*[]byte{&o.start, &o.end}
+func (o *deleteRangeOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.start = fn(o.start)
+	o.end = fn(o.end)
 }
 
 func (o *deleteRangeOp) diagramKeyRanges() []pebble.KeyRange {
@@ -424,34 +437,36 @@ type flushOp struct {
 func (o *flushOp) run(t *Test, h historyRecorder) {
 	db := t.getDB(o.db)
 	err := db.Flush()
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *flushOp) String() string                      { return fmt.Sprintf("%s.Flush()", o.db) }
+func (o *flushOp) formattedString(KeyFormat) string    { return fmt.Sprintf("%s.Flush()", o.db) }
 func (o *flushOp) receiver() objID                     { return o.db }
 func (o *flushOp) syncObjs() objIDSlice                { return nil }
-func (o *flushOp) keys() []*[]byte                     { return nil }
+func (o *flushOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *flushOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // mergeOp models a Write.Merge operation.
 type mergeOp struct {
 	writerID objID
-	key      []byte
+	key      UserKey
 	value    []byte
 }
 
 func (o *mergeOp) run(t *Test, h historyRecorder) {
 	w := t.getWriter(o.writerID)
 	err := w.Merge(o.key, o.value, t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *mergeOp) String() string       { return fmt.Sprintf("%s.Merge(%q, %q)", o.writerID, o.key, o.value) }
+func (o *mergeOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Merge(%q, %q)", o.writerID, kf.FormatKey(o.key), o.value)
+}
 func (o *mergeOp) receiver() objID      { return o.writerID }
 func (o *mergeOp) syncObjs() objIDSlice { return nil }
 
-func (o *mergeOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *mergeOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *mergeOp) diagramKeyRanges() []pebble.KeyRange {
@@ -461,22 +476,24 @@ func (o *mergeOp) diagramKeyRanges() []pebble.KeyRange {
 // setOp models a Write.Set operation.
 type setOp struct {
 	writerID objID
-	key      []byte
+	key      UserKey
 	value    []byte
 }
 
 func (o *setOp) run(t *Test, h historyRecorder) {
 	w := t.getWriter(o.writerID)
 	err := w.Set(o.key, o.value, t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *setOp) String() string       { return fmt.Sprintf("%s.Set(%q, %q)", o.writerID, o.key, o.value) }
+func (o *setOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Set(%q, %q)", o.writerID, kf.FormatKey(o.key), o.value)
+}
 func (o *setOp) receiver() objID      { return o.writerID }
 func (o *setOp) syncObjs() objIDSlice { return nil }
 
-func (o *setOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *setOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *setOp) diagramKeyRanges() []pebble.KeyRange {
@@ -493,18 +510,20 @@ type rangeKeyDeleteOp struct {
 func (o *rangeKeyDeleteOp) run(t *Test, h historyRecorder) {
 	w := t.getWriter(o.writerID)
 	err := w.RangeKeyDelete(o.start, o.end, t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *rangeKeyDeleteOp) String() string {
-	return fmt.Sprintf("%s.RangeKeyDelete(%q, %q)", o.writerID, o.start, o.end)
+func (o *rangeKeyDeleteOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.RangeKeyDelete(%q, %q)",
+		o.writerID, kf.FormatKey(o.start), kf.FormatKey(o.end))
 }
 
 func (o *rangeKeyDeleteOp) receiver() objID      { return o.writerID }
 func (o *rangeKeyDeleteOp) syncObjs() objIDSlice { return nil }
 
-func (o *rangeKeyDeleteOp) keys() []*[]byte {
-	return []*[]byte{&o.start, &o.end}
+func (o *rangeKeyDeleteOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.start = fn(o.start)
+	o.end = fn(o.end)
 }
 
 func (o *rangeKeyDeleteOp) diagramKeyRanges() []pebble.KeyRange {
@@ -523,19 +542,21 @@ type rangeKeySetOp struct {
 func (o *rangeKeySetOp) run(t *Test, h historyRecorder) {
 	w := t.getWriter(o.writerID)
 	err := w.RangeKeySet(o.start, o.end, o.suffix, o.value, t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *rangeKeySetOp) String() string {
+func (o *rangeKeySetOp) formattedString(kf KeyFormat) string {
 	return fmt.Sprintf("%s.RangeKeySet(%q, %q, %q, %q)",
-		o.writerID, o.start, o.end, o.suffix, o.value)
+		o.writerID, kf.FormatKey(o.start), kf.FormatKey(o.end),
+		kf.FormatKeySuffix(o.suffix), o.value)
 }
 
 func (o *rangeKeySetOp) receiver() objID      { return o.writerID }
 func (o *rangeKeySetOp) syncObjs() objIDSlice { return nil }
 
-func (o *rangeKeySetOp) keys() []*[]byte {
-	return []*[]byte{&o.start, &o.end}
+func (o *rangeKeySetOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.start = fn(o.start)
+	o.end = fn(o.end)
 }
 
 func (o *rangeKeySetOp) diagramKeyRanges() []pebble.KeyRange {
@@ -545,27 +566,29 @@ func (o *rangeKeySetOp) diagramKeyRanges() []pebble.KeyRange {
 // rangeKeyUnsetOp models a Write.RangeKeyUnset operation.
 type rangeKeyUnsetOp struct {
 	writerID objID
-	start    []byte
-	end      []byte
-	suffix   []byte
+	start    UserKey
+	end      UserKey
+	suffix   UserKeySuffix
 }
 
 func (o *rangeKeyUnsetOp) run(t *Test, h historyRecorder) {
 	w := t.getWriter(o.writerID)
 	err := w.RangeKeyUnset(o.start, o.end, o.suffix, t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *rangeKeyUnsetOp) String() string {
+func (o *rangeKeyUnsetOp) formattedString(kf KeyFormat) string {
 	return fmt.Sprintf("%s.RangeKeyUnset(%q, %q, %q)",
-		o.writerID, o.start, o.end, o.suffix)
+		o.writerID, kf.FormatKey(o.start), kf.FormatKey(o.end),
+		kf.FormatKeySuffix(o.suffix))
 }
 
 func (o *rangeKeyUnsetOp) receiver() objID      { return o.writerID }
 func (o *rangeKeyUnsetOp) syncObjs() objIDSlice { return nil }
 
-func (o *rangeKeyUnsetOp) keys() []*[]byte {
-	return []*[]byte{&o.start, &o.end}
+func (o *rangeKeyUnsetOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.start = fn(o.start)
+	o.end = fn(o.end)
 }
 
 func (o *rangeKeyUnsetOp) diagramKeyRanges() []pebble.KeyRange {
@@ -581,16 +604,16 @@ type logDataOp struct {
 func (o *logDataOp) run(t *Test, h historyRecorder) {
 	w := t.getWriter(o.writerID)
 	err := w.LogData(o.data, t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *logDataOp) String() string {
+func (o *logDataOp) formattedString(KeyFormat) string {
 	return fmt.Sprintf("%s.LogData(%q)", o.writerID, o.data)
 }
 
 func (o *logDataOp) receiver() objID                     { return o.writerID }
 func (o *logDataOp) syncObjs() objIDSlice                { return nil }
-func (o *logDataOp) keys() []*[]byte                     { return []*[]byte{} }
+func (o *logDataOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *logDataOp) diagramKeyRanges() []pebble.KeyRange { return []pebble.KeyRange{} }
 
 // newBatchOp models a Write.NewBatch operation.
@@ -602,10 +625,12 @@ type newBatchOp struct {
 func (o *newBatchOp) run(t *Test, h historyRecorder) {
 	b := t.getDB(o.dbID).NewBatch()
 	t.setBatch(o.batchID, b)
-	h.Recordf("%s", o)
+	h.Recordf("%s", o.formattedString(t.testOpts.KeyFormat))
 }
 
-func (o *newBatchOp) String() string  { return fmt.Sprintf("%s = %s.NewBatch()", o.batchID, o.dbID) }
+func (o *newBatchOp) formattedString(KeyFormat) string {
+	return fmt.Sprintf("%s = %s.NewBatch()", o.batchID, o.dbID)
+}
 func (o *newBatchOp) receiver() objID { return o.dbID }
 func (o *newBatchOp) syncObjs() objIDSlice {
 	// NewBatch should not be concurrent with operations that interact with that
@@ -613,8 +638,8 @@ func (o *newBatchOp) syncObjs() objIDSlice {
 	return []objID{o.batchID}
 }
 
-func (o *newBatchOp) keys() []*[]byte                     { return nil }
-func (o *newBatchOp) diagramKeyRanges() []pebble.KeyRange { return nil }
+func (o *newBatchOp) rewriteKeys(fn func(UserKey) UserKey) {}
+func (o *newBatchOp) diagramKeyRanges() []pebble.KeyRange  { return nil }
 
 // newIndexedBatchOp models a Write.NewIndexedBatch operation.
 type newIndexedBatchOp struct {
@@ -625,10 +650,10 @@ type newIndexedBatchOp struct {
 func (o *newIndexedBatchOp) run(t *Test, h historyRecorder) {
 	b := t.getDB(o.dbID).NewIndexedBatch()
 	t.setBatch(o.batchID, b)
-	h.Recordf("%s", o)
+	h.Recordf("%s", o.formattedString(t.testOpts.KeyFormat))
 }
 
-func (o *newIndexedBatchOp) String() string {
+func (o *newIndexedBatchOp) formattedString(KeyFormat) string {
 	return fmt.Sprintf("%s = %s.NewIndexedBatch()", o.batchID, o.dbID)
 }
 func (o *newIndexedBatchOp) receiver() objID { return o.dbID }
@@ -638,7 +663,7 @@ func (o *newIndexedBatchOp) syncObjs() objIDSlice {
 	return []objID{o.batchID}
 }
 
-func (o *newIndexedBatchOp) keys() []*[]byte                     { return nil }
+func (o *newIndexedBatchOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *newIndexedBatchOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // batchCommitOp models a Batch.Commit operation.
@@ -650,17 +675,19 @@ type batchCommitOp struct {
 func (o *batchCommitOp) run(t *Test, h historyRecorder) {
 	b := t.getBatch(o.batchID)
 	err := b.Commit(t.writeOpts)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *batchCommitOp) String() string  { return fmt.Sprintf("%s.Commit()", o.batchID) }
+func (o *batchCommitOp) formattedString(KeyFormat) string {
+	return fmt.Sprintf("%s.Commit()", o.batchID)
+}
 func (o *batchCommitOp) receiver() objID { return o.batchID }
 func (o *batchCommitOp) syncObjs() objIDSlice {
 	// Synchronize on the database so that NewIters wait for the commit.
 	return []objID{o.dbID}
 }
 
-func (o *batchCommitOp) keys() []*[]byte                     { return nil }
+func (o *batchCommitOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *batchCommitOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // ingestOp models a DB.Ingest operation.
@@ -687,7 +714,7 @@ func (o *ingestOp) run(t *Test, h historyRecorder) {
 		_ = b.Close()
 		_ = c.Close()
 		t.clearObj(id)
-		h.Recordf("%s // %v", o, err)
+		h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 
@@ -711,7 +738,7 @@ func (o *ingestOp) run(t *Test, h historyRecorder) {
 		return t.getDB(o.dbID).Ingest(context.Background(), paths)
 	}))
 
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
 func (o *ingestOp) receiver() objID { return o.dbID }
@@ -865,7 +892,7 @@ func (o *ingestOp) collapseBatch(
 	return collapsed, nil
 }
 
-func (o *ingestOp) String() string {
+func (o *ingestOp) formattedString(KeyFormat) string {
 	var buf strings.Builder
 	buf.WriteString(o.dbID.String())
 	buf.WriteString(".Ingest(")
@@ -879,14 +906,14 @@ func (o *ingestOp) String() string {
 	return buf.String()
 }
 
-func (o *ingestOp) keys() []*[]byte                     { return nil }
+func (o *ingestOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *ingestOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 type ingestAndExciseOp struct {
 	dbID                   objID
 	batchID                objID
 	derivedDBID            objID
-	exciseStart, exciseEnd []byte
+	exciseStart, exciseEnd UserKey
 }
 
 func (o *ingestAndExciseOp) run(t *Test, h historyRecorder) {
@@ -898,7 +925,8 @@ func (o *ingestAndExciseOp) run(t *Test, h historyRecorder) {
 	}
 	db := t.getDB(o.dbID)
 	if b.Empty() {
-		h.Recordf("%s // %v", o, o.simulateExcise(db, t))
+		h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat),
+			o.simulateExcise(db, t))
 		return
 	}
 
@@ -910,7 +938,7 @@ func (o *ingestAndExciseOp) run(t *Test, h historyRecorder) {
 	err = firstError(err, b.Close())
 
 	if writerMeta.Properties.NumEntries == 0 && writerMeta.Properties.NumRangeKeys() == 0 {
-		h.Recordf("%s // %v", o, o.simulateExcise(db, t))
+		h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), o.simulateExcise(db, t))
 		return
 	}
 
@@ -929,7 +957,7 @@ func (o *ingestAndExciseOp) run(t *Test, h historyRecorder) {
 		}))
 	}
 
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
 func (o *ingestAndExciseOp) simulateExcise(db *pebble.DB, t *Test) error {
@@ -951,12 +979,14 @@ func (o *ingestAndExciseOp) syncObjs() objIDSlice {
 	return objs
 }
 
-func (o *ingestAndExciseOp) String() string {
-	return fmt.Sprintf("%s.IngestAndExcise(%s, %q, %q)", o.dbID, o.batchID, o.exciseStart, o.exciseEnd)
+func (o *ingestAndExciseOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.IngestAndExcise(%s, %q, %q)",
+		o.dbID, o.batchID, kf.FormatKey(o.exciseStart), kf.FormatKey(o.exciseEnd))
 }
 
-func (o *ingestAndExciseOp) keys() []*[]byte {
-	return []*[]byte{&o.exciseStart, &o.exciseEnd}
+func (o *ingestAndExciseOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.exciseStart = fn(o.exciseStart)
+	o.exciseEnd = fn(o.exciseEnd)
 }
 
 func (o *ingestAndExciseOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1033,7 +1063,7 @@ func (o *ingestExternalFilesOp) run(t *Test, h historyRecorder) {
 		_, err = db.IngestExternalFiles(context.Background(), external)
 	}
 
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
 func (o *ingestExternalFilesOp) receiver() objID { return o.dbID }
@@ -1047,30 +1077,32 @@ func (o *ingestExternalFilesOp) syncObjs() objIDSlice {
 	return slices.Compact(res)
 }
 
-func (o *ingestExternalFilesOp) String() string {
+func (o *ingestExternalFilesOp) formattedString(kf KeyFormat) string {
 	strs := make([]string, len(o.objs))
 	for i, obj := range o.objs {
 		strs[i] = fmt.Sprintf("%s, %q /* start */, %q /* end */, %q /* syntheticSuffix */, %q /* syntheticPrefix */",
-			obj.externalObjID, obj.bounds.Start, obj.bounds.End, obj.syntheticSuffix, obj.syntheticPrefix,
+			obj.externalObjID, kf.FormatKey(obj.bounds.Start), kf.FormatKey(obj.bounds.End),
+			kf.FormatKeySuffix(UserKeySuffix(obj.syntheticSuffix)), kf.FormatKey(UserKey(obj.syntheticPrefix)),
 		)
 	}
 	return fmt.Sprintf("%s.IngestExternalFiles(%s)", o.dbID, strings.Join(strs, ", "))
 }
 
-func (o *ingestExternalFilesOp) keys() []*[]byte {
-	// If any of the objects have synthetic prefixes, we can't allow modification
-	// of external object bounds.
+func (o *ingestExternalFilesOp) rewriteKeys(fn func(UserKey) UserKey) {
+	// If any of the objects have synthetic prefixes, we can't allow
+	// modification of external object bounds.
+	allowModification := true
 	for i := range o.objs {
-		if o.objs[i].syntheticPrefix.IsSet() {
-			return nil
-		}
+		allowModification = allowModification && !o.objs[i].syntheticPrefix.IsSet()
 	}
 
-	var res []*[]byte
 	for i := range o.objs {
-		res = append(res, &o.objs[i].bounds.Start, &o.objs[i].bounds.End)
+		s, e := fn(o.objs[i].bounds.Start), fn(o.objs[i].bounds.End)
+		if allowModification {
+			o.objs[i].bounds.Start = s
+			o.objs[i].bounds.End = e
+		}
 	}
-	return res
 }
 
 func (o *ingestExternalFilesOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1084,7 +1116,7 @@ func (o *ingestExternalFilesOp) diagramKeyRanges() []pebble.KeyRange {
 // getOp models a Reader.Get operation.
 type getOp struct {
 	readerID    objID
-	key         []byte
+	key         UserKey
 	derivedDBID objID
 }
 
@@ -1096,13 +1128,15 @@ func (o *getOp) run(t *Test, h historyRecorder) {
 		val, closer, err = r.Get(o.key)
 		return err
 	})
-	h.Recordf("%s // [%q] %v", o, val, err)
+	h.Recordf("%s // [%q] %v", o.formattedString(t.testOpts.KeyFormat), val, err)
 	if closer != nil {
 		closer.Close()
 	}
 }
 
-func (o *getOp) String() string  { return fmt.Sprintf("%s.Get(%q)", o.readerID, o.key) }
+func (o *getOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Get(%q)", o.readerID, kf.FormatKey(o.key))
+}
 func (o *getOp) receiver() objID { return o.readerID }
 func (o *getOp) syncObjs() objIDSlice {
 	if o.readerID.tag() == dbTag {
@@ -1115,8 +1149,8 @@ func (o *getOp) syncObjs() objIDSlice {
 	return nil
 }
 
-func (o *getOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *getOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *getOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1136,7 +1170,7 @@ const debugIterators = false
 
 func (o *newIterOp) run(t *Test, h historyRecorder) {
 	r := t.getReader(o.readerID)
-	opts := iterOptions(o.iterOpts)
+	opts := iterOptions(t.testOpts.KeyFormat, o.iterOpts)
 	if debugIterators {
 		opts.DebugRangeKeyStack = true
 	}
@@ -1158,12 +1192,14 @@ func (o *newIterOp) run(t *Test, h historyRecorder) {
 		rand.Read(opts.LowerBound[:])
 		rand.Read(opts.UpperBound[:])
 	}
-	h.Recordf("%s // %v", o, i.Error())
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), i.Error())
 }
 
-func (o *newIterOp) String() string {
-	return fmt.Sprintf("%s = %s.NewIter(%q, %q, %d /* key types */, %d, %d, %t /* use L6 filters */, %q /* masking suffix */)",
-		o.iterID, o.readerID, o.lower, o.upper, o.keyTypes, o.filterMin, o.filterMax, o.useL6Filters, o.maskSuffix)
+func (o *newIterOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s = %s.NewIter(%q, %q, %d /* key types */, %q, %q, %t /* use L6 filters */, %q /* masking suffix */)",
+		o.iterID, o.readerID, kf.FormatKey(o.lower), kf.FormatKey(o.upper),
+		o.keyTypes, kf.FormatKeySuffix(o.filterMin), kf.FormatKeySuffix(o.filterMax),
+		o.useL6Filters, kf.FormatKeySuffix(o.maskSuffix))
 }
 
 func (o *newIterOp) receiver() objID { return o.readerID }
@@ -1179,15 +1215,13 @@ func (o *newIterOp) syncObjs() objIDSlice {
 	return objs
 }
 
-func (o *newIterOp) keys() []*[]byte {
-	var res []*[]byte
+func (o *newIterOp) rewriteKeys(fn func(UserKey) UserKey) {
 	if o.lower != nil {
-		res = append(res, &o.lower)
+		o.lower = fn(o.lower)
 	}
 	if o.upper != nil {
-		res = append(res, &o.upper)
+		o.upper = fn(o.upper)
 	}
-	return res
 }
 
 func (o *newIterOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1217,7 +1251,7 @@ type newIterUsingCloneOp struct {
 func (o *newIterUsingCloneOp) run(t *Test, h historyRecorder) {
 	iter := t.getIter(o.existingIterID)
 	cloneOpts := pebble.CloneOptions{
-		IterOptions:      iterOptions(o.iterOpts),
+		IterOptions:      iterOptions(t.testOpts.KeyFormat, o.iterOpts),
 		RefreshBatchView: o.refreshBatch,
 	}
 	i, err := iter.iter.Clone(cloneOpts)
@@ -1225,13 +1259,14 @@ func (o *newIterUsingCloneOp) run(t *Test, h historyRecorder) {
 		panic(err)
 	}
 	t.setIter(o.iterID, i)
-	h.Recordf("%s // %v", o, i.Error())
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), i.Error())
 }
 
-func (o *newIterUsingCloneOp) String() string {
-	return fmt.Sprintf("%s = %s.Clone(%t, %q, %q, %d /* key types */, %d, %d, %t /* use L6 filters */, %q /* masking suffix */)",
-		o.iterID, o.existingIterID, o.refreshBatch, o.lower, o.upper,
-		o.keyTypes, o.filterMin, o.filterMax, o.useL6Filters, o.maskSuffix)
+func (o *newIterUsingCloneOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s = %s.Clone(%t, %q, %q, %d /* key types */, %q, %q, %t /* use L6 filters */, %q /* masking suffix */)",
+		o.iterID, o.existingIterID, o.refreshBatch, kf.FormatKey(o.lower),
+		kf.FormatKey(o.upper), o.keyTypes, kf.FormatKeySuffix(o.filterMin),
+		kf.FormatKeySuffix(o.filterMax), o.useL6Filters, kf.FormatKeySuffix(o.maskSuffix))
 }
 
 func (o *newIterUsingCloneOp) receiver() objID { return o.existingIterID }
@@ -1250,14 +1285,14 @@ func (o *newIterUsingCloneOp) syncObjs() objIDSlice {
 	return objIDs
 }
 
-func (o *newIterUsingCloneOp) keys() []*[]byte                     { return nil }
-func (o *newIterUsingCloneOp) diagramKeyRanges() []pebble.KeyRange { return nil }
+func (o *newIterUsingCloneOp) rewriteKeys(fn func(UserKey) UserKey) {}
+func (o *newIterUsingCloneOp) diagramKeyRanges() []pebble.KeyRange  { return nil }
 
 // iterSetBoundsOp models an Iterator.SetBounds operation.
 type iterSetBoundsOp struct {
 	iterID objID
-	lower  []byte
-	upper  []byte
+	lower  UserKey
+	upper  UserKey
 }
 
 func (o *iterSetBoundsOp) run(t *Test, h historyRecorder) {
@@ -1276,18 +1311,20 @@ func (o *iterSetBoundsOp) run(t *Test, h historyRecorder) {
 	rand.Read(lower[:])
 	rand.Read(upper[:])
 
-	h.Recordf("%s // %v", o, i.Error())
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), i.Error())
 }
 
-func (o *iterSetBoundsOp) String() string {
-	return fmt.Sprintf("%s.SetBounds(%q, %q)", o.iterID, o.lower, o.upper)
+func (o *iterSetBoundsOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.SetBounds(%q, %q)", o.iterID,
+		kf.FormatKey(o.lower), kf.FormatKey(o.upper))
 }
 
 func (o *iterSetBoundsOp) receiver() objID      { return o.iterID }
 func (o *iterSetBoundsOp) syncObjs() objIDSlice { return nil }
 
-func (o *iterSetBoundsOp) keys() []*[]byte {
-	return []*[]byte{&o.lower, &o.upper}
+func (o *iterSetBoundsOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.lower = fn(o.lower)
+	o.upper = fn(o.upper)
 }
 
 func (o *iterSetBoundsOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1308,7 +1345,7 @@ type iterSetOptionsOp struct {
 func (o *iterSetOptionsOp) run(t *Test, h historyRecorder) {
 	i := t.getIter(o.iterID)
 
-	opts := iterOptions(o.iterOpts)
+	opts := iterOptions(t.testOpts.KeyFormat, o.iterOpts)
 	if opts == nil {
 		opts = &pebble.IterOptions{}
 	}
@@ -1319,15 +1356,17 @@ func (o *iterSetOptionsOp) run(t *Test, h historyRecorder) {
 	rand.Read(opts.LowerBound[:])
 	rand.Read(opts.UpperBound[:])
 
-	h.Recordf("%s // %v", o, i.Error())
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), i.Error())
 }
 
-func (o *iterSetOptionsOp) String() string {
-	return fmt.Sprintf("%s.SetOptions(%q, %q, %d /* key types */, %d, %d, %t /* use L6 filters */, %q /* masking suffix */)",
-		o.iterID, o.lower, o.upper, o.keyTypes, o.filterMin, o.filterMax, o.useL6Filters, o.maskSuffix)
+func (o *iterSetOptionsOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.SetOptions(%q, %q, %d /* key types */, %q, %q, %t /* use L6 filters */, %q /* masking suffix */)",
+		o.iterID, kf.FormatKey(o.lower), kf.FormatKey(o.upper),
+		o.keyTypes, kf.FormatKeySuffix(o.filterMin), kf.FormatKeySuffix(o.filterMax),
+		o.useL6Filters, kf.FormatKeySuffix(o.maskSuffix))
 }
 
-func iterOptions(o iterOpts) *pebble.IterOptions {
+func iterOptions(kf KeyFormat, o iterOpts) *pebble.IterOptions {
 	if o.IsZero() && !debugIterators {
 		return nil
 	}
@@ -1349,29 +1388,28 @@ func iterOptions(o iterOpts) *pebble.IterOptions {
 		DebugRangeKeyStack: debugIterators,
 	}
 	if opts.RangeKeyMasking.Suffix != nil {
-		opts.RangeKeyMasking.Filter = func() pebble.BlockPropertyFilterMask {
-			return sstable.NewTestKeysMaskingFilter()
-		}
+		opts.RangeKeyMasking.Filter = kf.NewSuffixFilterMask
 	}
-	if o.filterMax > 0 {
+	if o.filterMax != nil {
 		opts.PointKeyFilters = []pebble.BlockPropertyFilter{
-			sstable.NewTestKeysBlockPropertyFilter(o.filterMin, o.filterMax),
+			kf.NewSuffixBlockPropertyFilter(o.filterMin, o.filterMax),
 		}
 		// Enforce the timestamp bounds in SkipPoint, so that the iterator never
 		// returns a key outside the filterMin, filterMax bounds. This provides
 		// deterministic iteration.
 		opts.SkipPoint = func(k []byte) (skip bool) {
-			n := testkeys.Comparer.Split(k)
+			n := kf.Comparer.Split(k)
 			if n == len(k) {
 				// No suffix, don't skip it.
 				return false
 			}
-			v, err := testkeys.ParseSuffix(k[n:])
-			if err != nil {
-				panic(err)
+			if kf.Comparer.ComparePointSuffixes(k[n:], o.filterMin) < 0 {
+				return true
+			}
+			if kf.Comparer.ComparePointSuffixes(k[n:], o.filterMax) >= 0 {
+				return true
 			}
-			ts := uint64(v)
-			return ts < o.filterMin || ts >= o.filterMax
+			return false
 		}
 	}
 	return opts
@@ -1389,14 +1427,14 @@ func (o *iterSetOptionsOp) syncObjs() objIDSlice {
 	return nil
 }
 
-func (o *iterSetOptionsOp) keys() []*[]byte                     { return nil }
-func (o *iterSetOptionsOp) diagramKeyRanges() []pebble.KeyRange { return nil }
+func (o *iterSetOptionsOp) rewriteKeys(fn func(UserKey) UserKey) {}
+func (o *iterSetOptionsOp) diagramKeyRanges() []pebble.KeyRange  { return nil }
 
 // iterSeekGEOp models an Iterator.SeekGE[WithLimit] operation.
 type iterSeekGEOp struct {
 	iterID objID
-	key    []byte
-	limit  []byte
+	key    UserKey
+	limit  UserKey
 
 	derivedReaderID objID
 }
@@ -1457,20 +1495,23 @@ func (o *iterSeekGEOp) run(t *Test, h historyRecorder) {
 		valid, validStr = validityStateToStr(i.SeekGEWithLimit(o.key, o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
+		h.Recordf("%s // [%s] %v",
+			o.formattedString(t.testOpts.KeyFormat), validStr, i.Error())
 	}
 }
 
-func (o *iterSeekGEOp) String() string {
-	return fmt.Sprintf("%s.SeekGE(%q, %q)", o.iterID, o.key, o.limit)
+func (o *iterSeekGEOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.SeekGE(%q, %q)",
+		o.iterID, kf.FormatKey(o.key), kf.FormatKey(o.limit))
 }
 func (o *iterSeekGEOp) receiver() objID      { return o.iterID }
 func (o *iterSeekGEOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterSeekGEOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *iterSeekGEOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *iterSeekGEOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1490,7 +1531,7 @@ func onlyBatchIDs(ids ...objID) objIDSlice {
 // iterSeekPrefixGEOp models an Iterator.SeekPrefixGE operation.
 type iterSeekPrefixGEOp struct {
 	iterID objID
-	key    []byte
+	key    UserKey
 
 	derivedReaderID objID
 }
@@ -1499,20 +1540,21 @@ func (o *iterSeekPrefixGEOp) run(t *Test, h historyRecorder) {
 	i := t.getIter(o.iterID)
 	valid := i.SeekPrefixGE(o.key)
 	if valid {
-		h.Recordf("%s // [%t,%s] %v", o, valid, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%t,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			valid, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%t] %v", o, valid, i.Error())
+		h.Recordf("%s // [%t] %v", o.formattedString(t.testOpts.KeyFormat), valid, i.Error())
 	}
 }
 
-func (o *iterSeekPrefixGEOp) String() string {
-	return fmt.Sprintf("%s.SeekPrefixGE(%q)", o.iterID, o.key)
+func (o *iterSeekPrefixGEOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.SeekPrefixGE(%q)", o.iterID, kf.FormatKey(o.key))
 }
 func (o *iterSeekPrefixGEOp) receiver() objID      { return o.iterID }
 func (o *iterSeekPrefixGEOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterSeekPrefixGEOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *iterSeekPrefixGEOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *iterSeekPrefixGEOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1522,8 +1564,8 @@ func (o *iterSeekPrefixGEOp) diagramKeyRanges() []pebble.KeyRange {
 // iterSeekLTOp models an Iterator.SeekLT[WithLimit] operation.
 type iterSeekLTOp struct {
 	iterID objID
-	key    []byte
-	limit  []byte
+	key    UserKey
+	limit  UserKey
 
 	derivedReaderID objID
 }
@@ -1539,21 +1581,24 @@ func (o *iterSeekLTOp) run(t *Test, h historyRecorder) {
 		valid, validStr = validityStateToStr(i.SeekLTWithLimit(o.key, o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
+		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, i.Error())
 	}
 }
 
-func (o *iterSeekLTOp) String() string {
-	return fmt.Sprintf("%s.SeekLT(%q, %q)", o.iterID, o.key, o.limit)
+func (o *iterSeekLTOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.SeekLT(%q, %q)", o.iterID,
+		kf.FormatKey(o.key), kf.FormatKey(o.limit))
 }
 
 func (o *iterSeekLTOp) receiver() objID      { return o.iterID }
 func (o *iterSeekLTOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterSeekLTOp) keys() []*[]byte {
-	return []*[]byte{&o.key}
+func (o *iterSeekLTOp) rewriteKeys(fn func(UserKey) UserKey) {
+	o.key = fn(o.key)
 }
 
 func (o *iterSeekLTOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1571,17 +1616,19 @@ func (o *iterFirstOp) run(t *Test, h historyRecorder) {
 	i := t.getIter(o.iterID)
 	valid := i.First()
 	if valid {
-		h.Recordf("%s // [%t,%s] %v", o, valid, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%t,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			valid, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%t] %v", o, valid, i.Error())
+		h.Recordf("%s // [%t] %v", o.formattedString(t.testOpts.KeyFormat),
+			valid, i.Error())
 	}
 }
 
-func (o *iterFirstOp) String() string       { return fmt.Sprintf("%s.First()", o.iterID) }
-func (o *iterFirstOp) receiver() objID      { return o.iterID }
-func (o *iterFirstOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
+func (o *iterFirstOp) formattedString(KeyFormat) string { return fmt.Sprintf("%s.First()", o.iterID) }
+func (o *iterFirstOp) receiver() objID                  { return o.iterID }
+func (o *iterFirstOp) syncObjs() objIDSlice             { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterFirstOp) keys() []*[]byte                     { return nil }
+func (o *iterFirstOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *iterFirstOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // iterLastOp models an Iterator.Last operation.
@@ -1595,23 +1642,25 @@ func (o *iterLastOp) run(t *Test, h historyRecorder) {
 	i := t.getIter(o.iterID)
 	valid := i.Last()
 	if valid {
-		h.Recordf("%s // [%t,%s] %v", o, valid, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%t,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			valid, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%t] %v", o, valid, i.Error())
+		h.Recordf("%s // [%t] %v", o.formattedString(t.testOpts.KeyFormat),
+			valid, i.Error())
 	}
 }
 
-func (o *iterLastOp) String() string       { return fmt.Sprintf("%s.Last()", o.iterID) }
-func (o *iterLastOp) receiver() objID      { return o.iterID }
-func (o *iterLastOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
+func (o *iterLastOp) formattedString(KeyFormat) string { return fmt.Sprintf("%s.Last()", o.iterID) }
+func (o *iterLastOp) receiver() objID                  { return o.iterID }
+func (o *iterLastOp) syncObjs() objIDSlice             { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterLastOp) keys() []*[]byte                     { return nil }
+func (o *iterLastOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *iterLastOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // iterNextOp models an Iterator.Next[WithLimit] operation.
 type iterNextOp struct {
 	iterID objID
-	limit  []byte
+	limit  UserKey
 
 	derivedReaderID objID
 }
@@ -1627,17 +1676,21 @@ func (o *iterNextOp) run(t *Test, h historyRecorder) {
 		valid, validStr = validityStateToStr(i.NextWithLimit(o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
+		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, i.Error())
 	}
 }
 
-func (o *iterNextOp) String() string       { return fmt.Sprintf("%s.Next(%q)", o.iterID, o.limit) }
+func (o *iterNextOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Next(%q)", o.iterID, kf.FormatKey(o.limit))
+}
 func (o *iterNextOp) receiver() objID      { return o.iterID }
 func (o *iterNextOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterNextOp) keys() []*[]byte                     { return nil }
+func (o *iterNextOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *iterNextOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // iterNextPrefixOp models an Iterator.NextPrefix operation.
@@ -1652,17 +1705,21 @@ func (o *iterNextPrefixOp) run(t *Test, h historyRecorder) {
 	valid := i.NextPrefix()
 	validStr := validBoolToStr(valid)
 	if valid {
-		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
+		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, i.Error())
 	}
 }
 
-func (o *iterNextPrefixOp) String() string       { return fmt.Sprintf("%s.NextPrefix()", o.iterID) }
+func (o *iterNextPrefixOp) formattedString(KeyFormat) string {
+	return fmt.Sprintf("%s.NextPrefix()", o.iterID)
+}
 func (o *iterNextPrefixOp) receiver() objID      { return o.iterID }
 func (o *iterNextPrefixOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterNextPrefixOp) keys() []*[]byte                     { return nil }
+func (o *iterNextPrefixOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *iterNextPrefixOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // iterCanSingleDelOp models a call to CanDeterministicallySingleDelete with an
@@ -1688,14 +1745,16 @@ func (o *iterCanSingleDelOp) run(t *Test, h historyRecorder) {
 	//  - The iterator was already in an error state when the operation ran.
 	//  - The operation is deterministically invalid (like using an InternalNext
 	//    to change directions.)
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *iterCanSingleDelOp) String() string       { return fmt.Sprintf("%s.InternalNext()", o.iterID) }
+func (o *iterCanSingleDelOp) formattedString(KeyFormat) string {
+	return fmt.Sprintf("%s.InternalNext()", o.iterID)
+}
 func (o *iterCanSingleDelOp) receiver() objID      { return o.iterID }
 func (o *iterCanSingleDelOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterCanSingleDelOp) keys() []*[]byte                     { return nil }
+func (o *iterCanSingleDelOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *iterCanSingleDelOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // iterPrevOp models an Iterator.Prev[WithLimit] operation.
@@ -1717,17 +1776,21 @@ func (o *iterPrevOp) run(t *Test, h historyRecorder) {
 		valid, validStr = validityStateToStr(i.PrevWithLimit(o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, iteratorPos(i), i.Error())
 	} else {
-		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
+		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
+			validStr, i.Error())
 	}
 }
 
-func (o *iterPrevOp) String() string       { return fmt.Sprintf("%s.Prev(%q)", o.iterID, o.limit) }
+func (o *iterPrevOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Prev(%q)", o.iterID, kf.FormatKey(o.limit))
+}
 func (o *iterPrevOp) receiver() objID      { return o.iterID }
 func (o *iterPrevOp) syncObjs() objIDSlice { return onlyBatchIDs(o.derivedReaderID) }
 
-func (o *iterPrevOp) keys() []*[]byte                     { return nil }
+func (o *iterPrevOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *iterPrevOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 // newSnapshotOp models a DB.NewSnapshot operation.
@@ -1757,17 +1820,17 @@ func (o *newSnapshotOp) run(t *Test, h historyRecorder) {
 		s := t.getDB(o.dbID).NewSnapshot()
 		t.setSnapshot(o.snapID, s)
 	}
-	h.Recordf("%s", o)
+	h.Recordf("%s", o.formattedString(t.testOpts.KeyFormat))
 }
 
-func (o *newSnapshotOp) String() string {
+func (o *newSnapshotOp) formattedString(kf KeyFormat) string {
 	var buf bytes.Buffer
 	fmt.Fprintf(&buf, "%s = %s.NewSnapshot(", o.snapID, o.dbID)
 	for i := range o.bounds {
 		if i > 0 {
 			fmt.Fprint(&buf, ", ")
 		}
-		fmt.Fprintf(&buf, "%q, %q", o.bounds[i].Start, o.bounds[i].End)
+		fmt.Fprintf(&buf, "%q, %q", kf.FormatKey(o.bounds[i].Start), kf.FormatKey(o.bounds[i].End))
 	}
 	fmt.Fprint(&buf, ")")
 	return buf.String()
@@ -1775,12 +1838,11 @@ func (o *newSnapshotOp) String() string {
 func (o *newSnapshotOp) receiver() objID      { return o.dbID }
 func (o *newSnapshotOp) syncObjs() objIDSlice { return []objID{o.snapID} }
 
-func (o *newSnapshotOp) keys() []*[]byte {
-	var res []*[]byte
+func (o *newSnapshotOp) rewriteKeys(fn func(UserKey) UserKey) {
 	for i := range o.bounds {
-		res = append(res, &o.bounds[i].Start, &o.bounds[i].End)
+		o.bounds[i].Start = fn(o.bounds[i].Start)
+		o.bounds[i].End = fn(o.bounds[i].End)
 	}
-	return res
 }
 
 func (o *newSnapshotOp) diagramKeyRanges() []pebble.KeyRange {
@@ -1831,16 +1893,16 @@ func (o *newExternalObjOp) run(t *Test, h historyRecorder) {
 	t.setExternalObj(o.externalObjID, externalObjMeta{
 		sstMeta: sstMeta,
 	})
-	h.Recordf("%s", o)
+	h.Recordf("%s", o.formattedString(t.testOpts.KeyFormat))
 }
 
-func (o *newExternalObjOp) String() string {
+func (o *newExternalObjOp) formattedString(KeyFormat) string {
 	return fmt.Sprintf("%s = %s.NewExternalObj()", o.externalObjID, o.batchID)
 }
 func (o *newExternalObjOp) receiver() objID      { return o.batchID }
 func (o *newExternalObjOp) syncObjs() objIDSlice { return []objID{o.externalObjID} }
 
-func (o *newExternalObjOp) keys() []*[]byte                     { return nil }
+func (o *newExternalObjOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *newExternalObjOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 type dbRatchetFormatMajorVersionOp struct {
@@ -1861,16 +1923,16 @@ func (o *dbRatchetFormatMajorVersionOp) run(t *Test, h historyRecorder) {
 	if t.getDB(o.dbID).FormatMajorVersion() < o.vers {
 		err = t.getDB(o.dbID).RatchetFormatMajorVersion(o.vers)
 	}
-	h.Recordf("%s // %v", o, err)
+	h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (o *dbRatchetFormatMajorVersionOp) String() string {
+func (o *dbRatchetFormatMajorVersionOp) formattedString(KeyFormat) string {
 	return fmt.Sprintf("%s.RatchetFormatMajorVersion(%s)", o.dbID, o.vers)
 }
 func (o *dbRatchetFormatMajorVersionOp) receiver() objID      { return o.dbID }
 func (o *dbRatchetFormatMajorVersionOp) syncObjs() objIDSlice { return nil }
 
-func (o *dbRatchetFormatMajorVersionOp) keys() []*[]byte                     { return nil }
+func (o *dbRatchetFormatMajorVersionOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *dbRatchetFormatMajorVersionOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
 type dbRestartOp struct {
@@ -1884,24 +1946,24 @@ type dbRestartOp struct {
 
 func (o *dbRestartOp) run(t *Test, h historyRecorder) {
 	if err := t.restartDB(o.dbID); err != nil {
-		h.Recordf("%s // %v", o, err)
+		h.Recordf("%s // %v", o.formattedString(t.testOpts.KeyFormat), err)
 		h.history.err.Store(errors.Wrap(err, "dbRestartOp"))
 	} else {
-		h.Recordf("%s", o)
+		h.Recordf("%s", o.formattedString(t.testOpts.KeyFormat))
 	}
 }
 
-func (o *dbRestartOp) String() string       { return fmt.Sprintf("%s.Restart()", o.dbID) }
-func (o *dbRestartOp) receiver() objID      { return o.dbID }
-func (o *dbRestartOp) syncObjs() objIDSlice { return o.affectedObjects }
+func (o *dbRestartOp) formattedString(KeyFormat) string { return fmt.Sprintf("%s.Restart()", o.dbID) }
+func (o *dbRestartOp) receiver() objID                  { return o.dbID }
+func (o *dbRestartOp) syncObjs() objIDSlice             { return o.affectedObjects }
 
-func (o *dbRestartOp) keys() []*[]byte                     { return nil }
+func (o *dbRestartOp) rewriteKeys(func(UserKey) UserKey)   {}
 func (o *dbRestartOp) diagramKeyRanges() []pebble.KeyRange { return nil }
 
-func formatOps(ops []op) string {
+func formatOps(kf KeyFormat, ops []op) string {
 	var buf strings.Builder
 	for _, op := range ops {
-		fmt.Fprintf(&buf, "%s\n", op)
+		fmt.Fprintf(&buf, "%s\n", op.formattedString(kf))
 	}
 	return buf.String()
 }
@@ -1910,7 +1972,7 @@ func formatOps(ops []op) string {
 // another through either an IngestAndExcise, or an Ingest.
 type replicateOp struct {
 	source, dest objID
-	start, end   []byte
+	start, end   UserKey
 }
 
 func (r *replicateOp) runSharedReplicate(
@@ -1943,18 +2005,18 @@ func (r *replicateOp) runSharedReplicate(
 		nil,
 	)
 	if err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 
 	err = w.Close()
 	if err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	meta, err := w.Raw().Metadata()
 	if err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	if len(sharedSSTs) == 0 && meta.Properties.NumEntries == 0 && meta.Properties.NumRangeKeys() == 0 {
@@ -1964,16 +2026,16 @@ func (r *replicateOp) runSharedReplicate(
 		//
 		// TODO(bilal): Remove this when we support excises with no matching ingests.
 		if err := dest.RangeKeyDelete(r.start, r.end, t.writeOpts); err != nil {
-			h.Recordf("%s // %v", r, err)
+			h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 			return
 		}
 		err := dest.DeleteRange(r.start, r.end, t.writeOpts)
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 
 	_, err = dest.IngestAndExcise(context.Background(), []string{sstPath}, sharedSSTs, nil /* external */, pebble.KeyRange{Start: r.start, End: r.end})
-	h.Recordf("%s // %v", r, err)
+	h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 }
 
 func (r *replicateOp) runExternalReplicate(
@@ -2006,18 +2068,18 @@ func (r *replicateOp) runExternalReplicate(
 		},
 	)
 	if err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 
 	err = w.Close()
 	if err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	meta, err := w.Raw().Metadata()
 	if err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	if len(externalSSTs) == 0 && meta.Properties.NumEntries == 0 && meta.Properties.NumRangeKeys() == 0 {
@@ -2027,16 +2089,16 @@ func (r *replicateOp) runExternalReplicate(
 		//
 		// TODO(bilal): Remove this when we support excises with no matching ingests.
 		if err := dest.RangeKeyDelete(r.start, r.end, t.writeOpts); err != nil {
-			h.Recordf("%s // %v", r, err)
+			h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 			return
 		}
 		err := dest.DeleteRange(r.start, r.end, t.writeOpts)
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 
 	_, err = dest.IngestAndExcise(context.Background(), []string{sstPath}, nil, externalSSTs /* external */, pebble.KeyRange{Start: r.start, End: r.end})
-	h.Recordf("%s // %v", r, err)
+	h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 }
 
 func (r *replicateOp) run(t *Test, h historyRecorder) {
@@ -2049,7 +2111,7 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 	sstPath := path.Join(t.tmpDir, fmt.Sprintf("ext-replicate%d.sst", t.idx))
 	f, err := t.opts.FS.Create(sstPath, vfs.WriteCategoryUnspecified)
 	if err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), t.opts.MakeWriterOptions(0, dest.TableFormat()))
@@ -2069,11 +2131,11 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 
 	// First, do a RangeKeyDelete and DeleteRange on the whole span.
 	if err := dest.RangeKeyDelete(r.start, r.end, t.writeOpts); err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	if err := dest.DeleteRange(r.start, r.end, t.writeOpts); err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	iter, err := source.NewIter(&pebble.IterOptions{
@@ -2116,7 +2178,7 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 		}
 	}
 	if err := iter.Error(); err != nil {
-		h.Recordf("%s // %v", r, err)
+		h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 		return
 	}
 	if err := w.Close(); err != nil {
@@ -2124,18 +2186,20 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 	}
 
 	err = dest.Ingest(context.Background(), []string{sstPath})
-	h.Recordf("%s // %v", r, err)
+	h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 }
 
-func (r *replicateOp) String() string {
-	return fmt.Sprintf("%s.Replicate(%s, %q, %q)", r.source, r.dest, r.start, r.end)
+func (r *replicateOp) formattedString(kf KeyFormat) string {
+	return fmt.Sprintf("%s.Replicate(%s, %q, %q)", r.source, r.dest,
+		kf.FormatKey(r.start), kf.FormatKey(r.end))
 }
 
 func (r *replicateOp) receiver() objID      { return r.source }
 func (r *replicateOp) syncObjs() objIDSlice { return objIDSlice{r.dest} }
 
-func (r *replicateOp) keys() []*[]byte {
-	return []*[]byte{&r.start, &r.end}
+func (r *replicateOp) rewriteKeys(fn func(UserKey) UserKey) {
+	r.start = fn(r.start)
+	r.end = fn(r.end)
 }
 
 func (r *replicateOp) diagramKeyRanges() []pebble.KeyRange {
diff --git a/metamorphic/options.go b/metamorphic/options.go
index 8b1dcd08..9bb52b0c 100644
--- a/metamorphic/options.go
+++ b/metamorphic/options.go
@@ -59,6 +59,9 @@ func parseOptions(
 				opts.strictFS = true
 				opts.Opts.FS = vfs.NewCrashableMem()
 				return true
+			case "TestOptions.key_format":
+				opts.KeyFormat = keyFormatsByName[value]
+				return true
 			case "TestOptions.ingest_using_apply":
 				opts.ingestUsingApply = true
 				return true
@@ -203,6 +206,7 @@ func parseOptions(
 
 func optionsToString(opts *TestOptions) string {
 	var buf bytes.Buffer
+	fmt.Fprintf(&buf, "  key_format=%s\n", opts.KeyFormat.Name)
 	if opts.strictFS {
 		fmt.Fprint(&buf, "  strictfs=true\n")
 	}
@@ -296,6 +300,7 @@ func defaultTestOptions() *TestOptions {
 		Opts:        defaultOptions(),
 		Threads:     16,
 		RetryPolicy: NeverRetry,
+		KeyFormat:   TestkeysKeyFormat,
 	}
 }
 
@@ -354,6 +359,8 @@ type TestOptions struct {
 	Threads int
 	// RetryPolicy configures which errors should be retried.
 	RetryPolicy RetryPolicy
+	// KeyFormat defines the format of keys used within the test.
+	KeyFormat KeyFormat
 	// CustomOptions holds custom test options that are defined outside of this
 	// package.
 	CustomOpts []CustomOption
diff --git a/metamorphic/parser.go b/metamorphic/parser.go
index 46924a77..855f788e 100644
--- a/metamorphic/parser.go
+++ b/metamorphic/parser.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 type methodInfo struct {
@@ -185,7 +186,9 @@ type parser struct {
 }
 
 type parserOpts struct {
-	allowUndefinedObjs bool
+	allowUndefinedObjs          bool
+	parseFormattedUserKey       func(string) (UserKey, error)
+	parseFormattedUserKeySuffix func(string) (UserKeySuffix, error)
 }
 
 func parse(src []byte, opts parserOpts) (_ []op, err error) {
@@ -280,6 +283,42 @@ func (p *parser) parseObjID(pos token.Pos, str string) objID {
 	return id
 }
 
+func (p *parser) parseUserKey(pos token.Pos, lit string) UserKey {
+	s, err := strconv.Unquote(lit)
+	if err != nil {
+		panic(p.errorf(pos, "%s", err))
+	}
+	if len(s) == 0 {
+		return nil
+	}
+	if p.opts.parseFormattedUserKey == nil {
+		return []byte(s)
+	}
+	key, err := p.opts.parseFormattedUserKey(s)
+	if err != nil {
+		panic(p.errorf(pos, "%s", err))
+	}
+	return key
+}
+
+func (p *parser) parseUserKeySuffix(pos token.Pos, lit string) UserKeySuffix {
+	s, err := strconv.Unquote(lit)
+	if err != nil {
+		panic(p.errorf(pos, "%s", err))
+	}
+	if len(s) == 0 {
+		return nil
+	}
+	if p.opts.parseFormattedUserKeySuffix == nil {
+		return []byte(s)
+	}
+	suffix, err := p.opts.parseFormattedUserKeySuffix(s)
+	if err != nil {
+		panic(p.errorf(pos, "%s", err))
+	}
+	return suffix
+}
+
 func unquoteBytes(lit string) []byte {
 	s, err := strconv.Unquote(lit)
 	if err != nil {
@@ -331,6 +370,14 @@ func (p *parser) parseArgs(op op, methodName string, args []interface{}) {
 			}
 			*t = val
 
+		case *UserKey:
+			elem.expectToken(p, token.STRING)
+			*t = p.parseUserKey(elem.pos, elem.lit)
+
+		case *UserKeySuffix:
+			elem.expectToken(p, token.STRING)
+			*t = p.parseUserKeySuffix(elem.pos, elem.lit)
+
 		case *[]byte:
 			elem.expectToken(p, token.STRING)
 			*t = unquoteBytes(elem.lit)
@@ -393,8 +440,18 @@ func (e listElem) expectToken(p *parser, expTok token.Token) {
 }
 
 // pop checks that the first element of the list matches the expected token,
-// removes it from the list and returns the literal.
-func (p *parser) pop(list *[]listElem, expTok token.Token) string {
+// removes it from the list and returns the pos and literal.
+func (p *parser) pop(list *[]listElem, expTok token.Token) (token.Pos, string) {
+	(*list)[0].expectToken(p, expTok)
+	pos := (*list)[0].pos
+	lit := (*list)[0].lit
+	(*list) = (*list)[1:]
+	return pos, lit
+}
+
+// popLit checks that the first element of the list matches the expected token,
+// removes it from the list and returns the pos and literal.
+func (p *parser) popLit(list *[]listElem, expTok token.Token) string {
 	(*list)[0].expectToken(p, expTok)
 	lit := (*list)[0].lit
 	(*list) = (*list)[1:]
@@ -439,10 +496,10 @@ func (p *parser) parseObjIDs(list []listElem) []objID {
 	return res
 }
 
-func (p *parser) parseKeys(list []listElem) [][]byte {
-	res := make([][]byte, len(list))
+func (p *parser) parseKeys(list []listElem) []UserKey {
+	res := make([]UserKey, len(list))
 	for i := range res {
-		res[i] = unquoteBytes(p.pop(&list, token.STRING))
+		res[i] = p.parseUserKey(p.pop(&list, token.STRING))
 	}
 	return res
 }
@@ -486,9 +543,9 @@ func (p *parser) parseDownloadSpans(list []listElem) []pebble.DownloadSpan {
 	res := make([]pebble.DownloadSpan, len(list)/3)
 	for i := range res {
 		res[i] = pebble.DownloadSpan{
-			StartKey:               unquoteBytes(p.pop(&list, token.STRING)),
-			EndKey:                 unquoteBytes(p.pop(&list, token.STRING)),
-			ViaBackingFileDownload: p.pop(&list, token.IDENT) == "true",
+			StartKey:               p.parseUserKey(p.pop(&list, token.STRING)),
+			EndKey:                 p.parseUserKey(p.pop(&list, token.STRING)),
+			ViaBackingFileDownload: p.popLit(&list, token.IDENT) == "true",
 		}
 	}
 	return res
@@ -501,19 +558,18 @@ func (p *parser) parseExternalObjsWithBounds(list []listElem) []externalObjWithB
 	}
 	objs := make([]externalObjWithBounds, len(list)/numArgs)
 	for i := range objs {
-		pos := list[0].pos
 		objs[i] = externalObjWithBounds{
-			externalObjID: p.parseObjID(pos, p.pop(&list, token.IDENT)),
+			externalObjID: p.parseObjID(p.pop(&list, token.IDENT)),
 			bounds: pebble.KeyRange{
-				Start: unquoteBytes(p.pop(&list, token.STRING)),
-				End:   unquoteBytes(p.pop(&list, token.STRING)),
+				Start: p.parseUserKey(p.pop(&list, token.STRING)),
+				End:   p.parseUserKey(p.pop(&list, token.STRING)),
 			},
 		}
-		if syntheticSuffix := unquoteBytes(p.pop(&list, token.STRING)); len(syntheticSuffix) > 0 {
-			objs[i].syntheticSuffix = syntheticSuffix
+		if syntheticSuffix := p.parseUserKeySuffix(p.pop(&list, token.STRING)); len(syntheticSuffix) > 0 {
+			objs[i].syntheticSuffix = block.SyntheticSuffix(syntheticSuffix)
 		}
 
-		syntheticPrefix := unquoteBytes(p.pop(&list, token.STRING))
+		syntheticPrefix := p.parseUserKey(p.pop(&list, token.STRING))
 		if len(syntheticPrefix) > 0 {
 			if !bytes.HasPrefix(objs[i].bounds.Start, syntheticPrefix) {
 				panic(fmt.Sprintf("invalid synthetic prefix %q %q", objs[i].bounds.Start, syntheticPrefix))
@@ -521,7 +577,7 @@ func (p *parser) parseExternalObjsWithBounds(list []listElem) []externalObjWithB
 			if !bytes.HasPrefix(objs[i].bounds.End, syntheticPrefix) {
 				panic(fmt.Sprintf("invalid synthetic prefix %q %q", objs[i].bounds.End, syntheticPrefix))
 			}
-			objs[i].syntheticPrefix = syntheticPrefix
+			objs[i].syntheticPrefix = block.SyntheticPrefix(syntheticPrefix)
 		}
 	}
 	return objs
diff --git a/metamorphic/simplify.go b/metamorphic/simplify.go
index dc317a94..bc1b5fce 100644
--- a/metamorphic/simplify.go
+++ b/metamorphic/simplify.go
@@ -7,7 +7,7 @@ package metamorphic
 import (
 	"sort"
 
-	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/internal/base"
 )
 
 // TryToSimplifyKeys parses the operations data and tries to reassign keys to
@@ -17,54 +17,56 @@ import (
 // On success it returns the new operations data.
 //
 // If there are too many distinct keys, returns nil.
-func TryToSimplifyKeys(opsData []byte, retainSuffixes bool) []byte {
-	ops, err := parse(opsData, parserOpts{})
+func TryToSimplifyKeys(keyFormat KeyFormat, opsData []byte, retainSuffixes bool) []byte {
+	ops, err := parse(opsData, parserOpts{
+		parseFormattedUserKey:       keyFormat.ParseFormattedKey,
+		parseFormattedUserKeySuffix: keyFormat.ParseFormattedKeySuffix,
+	})
 	if err != nil {
 		panic(err)
 	}
 	keys := make(map[string]struct{})
 	for i := range ops {
-		for _, k := range ops[i].keys() {
-			key := *k
+		ops[i].rewriteKeys(func(k UserKey) UserKey {
 			if retainSuffixes {
-				key = key[:testkeys.Comparer.Split(key)]
+				keys[string(k[:keyFormat.Comparer.Split(k)])] = struct{}{}
+			} else {
+				keys[string(k)] = struct{}{}
 			}
-			keys[string(key)] = struct{}{}
-		}
+			return k
+		})
 	}
 	if len(keys) > ('z' - 'a' + 1) {
 		return nil
 	}
-	sorted := sortedKeys(keys)
+	sorted := sortedKeys(keyFormat.Comparer.Compare, keys)
 	ordinals := make(map[string]int, len(sorted))
 	for i, k := range sorted {
 		ordinals[k] = i
 	}
 	for i := range ops {
-		for _, k := range ops[i].keys() {
-			key := *k
+		ops[i].rewriteKeys(func(k UserKey) UserKey {
 			var suffix []byte
 			if retainSuffixes {
-				n := testkeys.Comparer.Split(key)
-				suffix = key[n:]
-				key = key[:n]
+				n := keyFormat.Comparer.Split(k)
+				suffix = k[n:]
+				k = k[:n]
 			}
-			idx := ordinals[string(key)]
+			idx := ordinals[string(k)]
 			newKey := []byte{'a' + byte(idx)}
-			newKey = append(newKey, suffix...)
-			*k = newKey
-		}
+			return append(newKey, suffix...)
+		})
 	}
-	return []byte(formatOps(ops))
+	return []byte(formatOps(keyFormat, ops))
 }
 
-func sortedKeys(in map[string]struct{}) []string {
+func sortedKeys(cmp base.Compare, in map[string]struct{}) []string {
 	var sorted []string
 	for k := range in {
 		sorted = append(sorted, k)
 	}
 	sort.Slice(sorted, func(i, j int) bool {
-		return testkeys.Comparer.Compare([]byte(sorted[i]), []byte(sorted[j])) < 0
+		return cmp([]byte(sorted[i]), []byte(sorted[j])) < 0
 	})
 	return sorted
 }
diff --git a/metamorphic/key_generator.go b/metamorphic/testkeys.go
similarity index 73%
rename from metamorphic/key_generator.go
rename to metamorphic/testkeys.go
index f4744499..da6ca444 100644
--- a/metamorphic/key_generator.go
+++ b/metamorphic/testkeys.go
@@ -12,35 +12,57 @@ import (
 
 	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/sstable"
 )
 
-type keyGenerator struct {
+var TestkeysKeyFormat = KeyFormat{
+	Comparer:        testkeys.Comparer,
+	FormatKey:       func(k UserKey) string { return string(k) },
+	FormatKeySuffix: func(s UserKeySuffix) string { return string(s) },
+	NewGenerator: func(km *keyManager, rng *rand.Rand, cfg OpConfig) KeyGenerator {
+		return &testkeyKeyGenerator{
+			keyManager: km,
+			rng:        rng,
+			cfg:        cfg,
+		}
+	},
+	NewSuffixFilterMask: func() pebble.BlockPropertyFilterMask {
+		return sstable.NewTestKeysMaskingFilter()
+	},
+	NewSuffixBlockPropertyFilter: func(filterMin, filterMax []byte) sstable.BlockPropertyFilter {
+		low, err := testkeys.ParseSuffix(filterMin)
+		if err != nil {
+			panic(err)
+		}
+		high, err := testkeys.ParseSuffix(filterMax)
+		if err != nil {
+			panic(err)
+		}
+		return sstable.NewTestKeysBlockPropertyFilter(uint64(low), uint64(high))
+	},
+}
+
+type testkeyKeyGenerator struct {
 	keyManager *keyManager
 	rng        *rand.Rand
 	cfg        OpConfig
 }
 
-func newKeyGenerator(km *keyManager, rng *rand.Rand, cfg OpConfig) *keyGenerator {
-	return &keyGenerator{
-		keyManager: km,
-		rng:        rng,
-		cfg:        cfg,
-	}
-}
-
 // RandKey returns a random key (either a previously known key, or a new key).
-func (kg *keyGenerator) RandKey(newKeyProbability float64) []byte {
+func (kg *testkeyKeyGenerator) RandKey(newKeyProbability float64) []byte {
 	return kg.randKey(newKeyProbability, nil /* bounds */)
 }
 
 // RandKeyInRange returns a random key (either a previously known key, or a new
 // key) in the given key range.
-func (kg *keyGenerator) RandKeyInRange(newKeyProbability float64, kr pebble.KeyRange) []byte {
+func (kg *testkeyKeyGenerator) RandKeyInRange(
+	newKeyProbability float64, kr pebble.KeyRange,
+) []byte {
 	return kg.randKey(newKeyProbability, &kr)
 }
 
 // RandPrefix returns a random prefix key (a key with no suffix).
-func (kg *keyGenerator) RandPrefix(newPrefix float64) []byte {
+func (kg *testkeyKeyGenerator) RandPrefix(newPrefix float64) []byte {
 	prefixes := kg.keyManager.prefixes()
 	if len(prefixes) > 0 && kg.rng.Float64() > newPrefix {
 		return pickOneUniform(kg.rng, prefixes)
@@ -60,7 +82,7 @@ func (kg *keyGenerator) RandPrefix(newPrefix float64) []byte {
 
 // UniqueKeys takes a key-generating function and uses it to generate n unique
 // keys, returning them in sorted order.
-func (kg *keyGenerator) UniqueKeys(n int, genFn func() []byte) [][]byte {
+func (kg *testkeyKeyGenerator) UniqueKeys(n int, genFn func() []byte) [][]byte {
 	keys := make([][]byte, n)
 	used := make(map[string]struct{}, n)
 	for i := range keys {
@@ -86,16 +108,16 @@ func (kg *keyGenerator) UniqueKeys(n int, genFn func() []byte) [][]byte {
 //
 // May return a nil suffix, with the probability the configuration's suffix
 // distribution assigns to the zero suffix.
-func (kg *keyGenerator) SkewedSuffix(incMaxProb float64) []byte {
-	if suffix := kg.SkewedSuffixInt(incMaxProb); suffix != 0 {
+func (kg *testkeyKeyGenerator) SkewedSuffix(incMaxProb float64) []byte {
+	if suffix := kg.skewedSuffixInt(incMaxProb); suffix != 0 {
 		return testkeys.Suffix(suffix)
 	}
 	return nil
 }
 
-// SkewedSuffixInt is a variant of SkewedSuffix which returns the unencoded
+// skewedSuffixInt is a helper of SkewedSuffix which returns the unencoded
 // suffix as an integer.
-func (kg *keyGenerator) SkewedSuffixInt(incMaxProb float64) int64 {
+func (kg *testkeyKeyGenerator) skewedSuffixInt(incMaxProb float64) int64 {
 	if kg.rng.Float64() < incMaxProb {
 		kg.cfg.writeSuffixDist.IncMax(1)
 	}
@@ -105,26 +127,37 @@ func (kg *keyGenerator) SkewedSuffixInt(incMaxProb float64) int64 {
 // IncMaxSuffix increases the max suffix range and returns the new maximum
 // suffix (which is guaranteed to be larger than any previously generated
 // suffix).
-func (kg *keyGenerator) IncMaxSuffix() []byte {
+func (kg *testkeyKeyGenerator) IncMaxSuffix() []byte {
 	kg.cfg.writeSuffixDist.IncMax(1)
 	return testkeys.Suffix(int64(kg.cfg.writeSuffixDist.Max()))
 }
 
+func (kg *testkeyKeyGenerator) SuffixRange() (low, high []byte) {
+	a := kg.uniformSuffixInt()
+	b := kg.uniformSuffixInt()
+	if a > b {
+		a, b = b, a
+	} else if a == b {
+		b++
+	}
+	return testkeys.Suffix(a), testkeys.Suffix(b)
+}
+
 // UniformSuffix returns a suffix in the same range as SkewedSuffix but with a
 // uniform distribution. This is used during reads to better exercise reading a
 // mix of older and newer keys. The suffix can be empty.
 //
 // May return a nil suffix.
-func (kg *keyGenerator) UniformSuffix() []byte {
-	if suffix := kg.UniformSuffixInt(); suffix != 0 {
+func (kg *testkeyKeyGenerator) UniformSuffix() []byte {
+	if suffix := kg.uniformSuffixInt(); suffix != 0 {
 		return testkeys.Suffix(suffix)
 	}
 	return nil
 }
 
-// UniformSuffixInt is a variant of UniformSuffix which returns the unencoded
+// uniformSuffixInt is a helper of UniformSuffix which returns the unencoded
 // suffix as an integer.
-func (kg *keyGenerator) UniformSuffixInt() int64 {
+func (kg *testkeyKeyGenerator) uniformSuffixInt() int64 {
 	maxVal := kg.cfg.writeSuffixDist.Max()
 	return kg.rng.Int64N(int64(maxVal))
 }
@@ -132,7 +165,7 @@ func (kg *keyGenerator) UniformSuffixInt() int64 {
 // randKey returns a random key (either a previously known key or a new key).
 //
 // If bounds is not nil, the key will be inside the bounds.
-func (kg *keyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRange) []byte {
+func (kg *testkeyKeyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRange) []byte {
 	var knownKeys [][]byte
 	if bounds == nil {
 		knownKeys = kg.keyManager.knownKeys()
@@ -192,7 +225,7 @@ func (kg *keyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRan
 	}
 
 	if bounds == nil {
-		suffix := kg.SkewedSuffixInt(0.01)
+		suffix := kg.skewedSuffixInt(0.01)
 		for {
 			key := kg.generateKeyWithSuffix(4, 12, suffix)
 			if !kg.keyManager.prefixExists(kg.prefix(key)) {
@@ -214,7 +247,7 @@ func (kg *keyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRan
 		if cmpSuffix(startSuffix, endSuffix) >= 0 {
 			panic(fmt.Sprintf("invalid bounds [%q, %q)", bounds.Start, bounds.End))
 		}
-		suffix = kg.SkewedSuffixInt(0.01)
+		suffix = kg.skewedSuffixInt(0.01)
 		for i := 0; !(cmpSuffix(startSuffix, suffix) <= 0 && cmpSuffix(suffix, endSuffix) < 0); i++ {
 			if i > 10 {
 				// This value is always >= startSuffix and < endSuffix.
@@ -223,11 +256,11 @@ func (kg *keyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRan
 			}
 			// The suffix we want must exist in the current suffix range, we don't
 			// want to keep increasing it here.
-			suffix = kg.SkewedSuffixInt(0)
+			suffix = kg.skewedSuffixInt(0)
 		}
 	} else {
 		prefix = testkeys.RandomPrefixInRange(startPrefix, endPrefix, kg.rng)
-		suffix = kg.SkewedSuffixInt(0.01)
+		suffix = kg.skewedSuffixInt(0.01)
 		if kg.equal(prefix, startPrefix) {
 			// We can't use a suffix which sorts before startSuffix.
 			for i := 0; cmpSuffix(suffix, startSuffix) < 0; i++ {
@@ -235,7 +268,7 @@ func (kg *keyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRan
 					suffix = startSuffix
 					break
 				}
-				suffix = kg.SkewedSuffixInt(0)
+				suffix = kg.skewedSuffixInt(0)
 			}
 		}
 	}
@@ -244,7 +277,8 @@ func (kg *keyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRan
 		key = append(key, testkeys.Suffix(suffix)...)
 	}
 	if kg.cmp(key, bounds.Start) < 0 || kg.cmp(key, bounds.End) >= 0 {
-		panic(fmt.Sprintf("invalid randKey %q; bounds: [%q, %q) %v %v", key, bounds.Start, bounds.End, kg.cmp(key, bounds.Start), kg.cmp(key, bounds.End)))
+		panic(fmt.Sprintf("invalid randKey %q; bounds: [%q, %q) %v %v",
+			key, bounds.Start, bounds.End, kg.cmp(key, bounds.Start), kg.cmp(key, bounds.End)))
 	}
 	// We might (rarely) produce an existing key here, that's ok.
 	kg.keyManager.addNewKey(key)
@@ -253,7 +287,9 @@ func (kg *keyGenerator) randKey(newKeyProbability float64, bounds *pebble.KeyRan
 
 // generateKeyWithSuffix generates a key with a random prefix and the given
 // suffix. If the given suffix is 0, the key will not have a suffix.
-func (kg *keyGenerator) generateKeyWithSuffix(minPrefixLen, maxPrefixLen int, suffix int64) []byte {
+func (kg *testkeyKeyGenerator) generateKeyWithSuffix(
+	minPrefixLen, maxPrefixLen int, suffix int64,
+) []byte {
 	prefix := randBytes(kg.rng, minPrefixLen, maxPrefixLen)
 	if suffix == 0 {
 		return prefix
@@ -275,24 +311,24 @@ func cmpSuffix(s1, s2 int64) int {
 	}
 }
 
-func (kg *keyGenerator) cmp(a, b []byte) int {
-	return kg.keyManager.comparer.Compare(a, b)
+func (kg *testkeyKeyGenerator) cmp(a, b []byte) int {
+	return kg.keyManager.kf.Comparer.Compare(a, b)
 }
 
-func (kg *keyGenerator) equal(a, b []byte) bool {
-	return kg.keyManager.comparer.Equal(a, b)
+func (kg *testkeyKeyGenerator) equal(a, b []byte) bool {
+	return kg.keyManager.kf.Comparer.Equal(a, b)
 }
 
-func (kg *keyGenerator) split(a []byte) int {
-	return kg.keyManager.comparer.Split(a)
+func (kg *testkeyKeyGenerator) split(a []byte) int {
+	return kg.keyManager.kf.Comparer.Split(a)
 }
 
-func (kg *keyGenerator) prefix(a []byte) []byte {
+func (kg *testkeyKeyGenerator) prefix(a []byte) []byte {
 	n := kg.split(a)
 	return a[:n:n]
 }
 
-func (kg *keyGenerator) parseKey(k []byte) (prefix []byte, suffix int64) {
+func (kg *testkeyKeyGenerator) parseKey(k []byte) (prefix []byte, suffix int64) {
 	n := kg.split(k)
 	if n == len(k) {
 		return k, 0
diff --git a/sstable/block/buffer_pool.go b/sstable/block/buffer_pool.go
index 38809b41..4df870fe 100644
--- a/sstable/block/buffer_pool.go
+++ b/sstable/block/buffer_pool.go
@@ -8,6 +8,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/invariants"
 )
 
 // Alloc allocates a new Value for a block of length n (excluding the block
@@ -52,12 +53,21 @@ func (b Value) BlockMetadata() *Metadata {
 // backed by a buffer pool, MakeHandle inserts the value into the block cache,
 // returning a handle to the now resident value.
 func (b Value) MakeHandle(
-	c *cache.Cache, cacheID cache.ID, fileNum base.DiskFileNum, offset uint64,
+	crh cache.ReadHandle, cacheID cache.ID, fileNum base.DiskFileNum, offset uint64,
 ) BufferHandle {
 	if b.buf.Valid() {
+		if invariants.Enabled && crh.Valid() {
+			panic("cache.ReadHandle was valid")
+		}
 		return BufferHandle{b: b.buf}
 	}
-	return BufferHandle{h: c.Set(cacheID, fileNum, offset, b.v)}
+	return BufferHandle{h: crh.SetReadValue(b.v)}
+}
+
+func (b Value) SetInCacheForTesting(
+	c *cache.Cache, cacheID cache.ID, fileNum base.DiskFileNum, offset uint64,
+) cache.Handle {
+	return c.Set(cacheID, fileNum, offset, b.v)
 }
 
 // Release releases the handle.
diff --git a/sstable/block/compression.go b/sstable/block/compression.go
index f6b91540..751aae26 100644
--- a/sstable/block/compression.go
+++ b/sstable/block/compression.go
@@ -208,6 +208,13 @@ func (b PhysicalBlock) Clone() PhysicalBlock {
 	return PhysicalBlock{data: data, trailer: b.trailer}
 }
 
+// CloneUsingBuf makes a copy of the block data, using the given slice if it has
+// enough capacity.
+func (b PhysicalBlock) CloneUsingBuf(buf []byte) (_ PhysicalBlock, newBuf []byte) {
+	newBuf = append(buf[:0], b.data...)
+	return PhysicalBlock{data: newBuf, trailer: b.trailer}, newBuf
+}
+
 // IsCompressed returns true if the block is compressed.
 func (b *PhysicalBlock) IsCompressed() bool {
 	return CompressionIndicator(b.trailer[0]) != NoCompressionIndicator
@@ -260,14 +267,18 @@ func CompressAndChecksum(
 }
 
 // compress compresses a sstable block, using dstBuf as the desired destination.
+//
+// The result is aliased to dstBuf if that buffer had enough capacity, otherwise
+// it is a newly-allocated buffer.
 func compress(
 	compression Compression, b []byte, dstBuf []byte,
 ) (indicator CompressionIndicator, compressed []byte) {
 	switch compression {
 	case SnappyCompression:
+		// snappy relies on the length of the buffer, and not the capacity to
+		// determine if it needs to make an allocation.
+		dstBuf = dstBuf[:cap(dstBuf):cap(dstBuf)]
 		return SnappyCompressionIndicator, snappy.Encode(dstBuf, b)
-	case NoCompression:
-		return NoCompressionIndicator, b
 	case ZstdCompression:
 		if len(dstBuf) < binary.MaxVarintLen64 {
 			dstBuf = append(dstBuf, make([]byte, binary.MaxVarintLen64-len(dstBuf))...)
diff --git a/sstable/colblk_writer.go b/sstable/colblk_writer.go
index ed32490f..503ce149 100644
--- a/sstable/colblk_writer.go
+++ b/sstable/colblk_writer.go
@@ -95,6 +95,7 @@ type RawColumnWriter struct {
 	}
 	layout layoutWriter
 
+	lastKeyBuf            []byte
 	separatorBuf          []byte
 	tmp                   [blockHandleLikelyMaxLen]byte
 	previousUserKey       invariants.Value[[]byte]
@@ -390,7 +391,10 @@ func (w *RawColumnWriter) AddWithForceObsolete(
 	size := w.dataBlock.Size()
 	if shouldFlushWithoutLatestKV(size, w.pendingDataBlockSize, entriesWithoutKV, &w.dataFlush) {
 		// Flush the data block excluding the key we just added.
-		w.flushDataBlockWithoutNextKey(key.UserKey)
+		if err := w.flushDataBlockWithoutNextKey(key.UserKey); err != nil {
+			w.err = err
+			return err
+		}
 		// flushDataBlockWithoutNextKey reset the data block builder, and we can
 		// add the key to this next block now.
 		w.dataBlock.Add(key, valueStoredWithKey, valuePrefix, eval.kcmp, eval.isObsolete)
@@ -600,16 +604,19 @@ type compressedBlock struct {
 	blockBuf blockBuf
 }
 
-func (w *RawColumnWriter) flushDataBlockWithoutNextKey(nextKey []byte) {
+func (w *RawColumnWriter) flushDataBlockWithoutNextKey(nextKey []byte) error {
 	serializedBlock, lastKey := w.dataBlock.Finish(w.dataBlock.Rows()-1, w.pendingDataBlockSize)
 	w.maybeIncrementTombstoneDenseBlocks(len(serializedBlock))
 	// Compute the separator that will be written to the index block alongside
 	// this data block's end offset. It is the separator between the last key in
 	// the finished block and the [nextKey] that was excluded from the block.
 	w.separatorBuf = w.comparer.Separator(w.separatorBuf[:0], lastKey.UserKey, nextKey)
-	w.enqueueDataBlock(serializedBlock, lastKey, w.separatorBuf)
+	if err := w.enqueueDataBlock(serializedBlock, lastKey, w.separatorBuf); err != nil {
+		return err
+	}
 	w.dataBlock.Reset()
 	w.pendingDataBlockSize = 0
+	return nil
 }
 
 // maybeIncrementTombstoneDenseBlocks increments the number of tombstone dense
@@ -633,10 +640,11 @@ func (w *RawColumnWriter) maybeIncrementTombstoneDenseBlocks(uncompressedLen int
 func (w *RawColumnWriter) enqueueDataBlock(
 	serializedBlock []byte, lastKey base.InternalKey, separator []byte,
 ) error {
-	// TODO(jackson): Avoid allocating the largest point user key every time we
-	// set the largest point key. This is what the rowblk writer does too, but
-	// it's unnecessary.
-	w.meta.SetLargestPointKey(lastKey.Clone())
+	w.lastKeyBuf = append(w.lastKeyBuf[:0], lastKey.UserKey...)
+	w.meta.SetLargestPointKey(base.InternalKey{
+		UserKey: w.lastKeyBuf,
+		Trailer: lastKey.Trailer,
+	})
 
 	if invariants.Enabled {
 		var dec colblk.DataBlockDecoder
@@ -650,7 +658,7 @@ func (w *RawColumnWriter) enqueueDataBlock(
 	cb := compressedBlockPool.Get().(*compressedBlock)
 	cb.blockBuf.checksummer.Type = w.opts.Checksum
 	cb.physical = block.CompressAndChecksum(
-		&cb.blockBuf.compressedBuf,
+		&cb.blockBuf.dataBuf,
 		serializedBlock,
 		w.opts.Compression,
 		&cb.blockBuf.checksummer,
@@ -661,7 +669,7 @@ func (w *RawColumnWriter) enqueueDataBlock(
 		// it to the write queue to be asynchronously written to disk.
 		// TODO(jackson): Should we try to avoid this clone by tracking the
 		// lifetime of the DataBlockWriters?
-		cb.physical = cb.physical.Clone()
+		cb.physical, cb.blockBuf.dataBuf = cb.physical.CloneUsingBuf(cb.blockBuf.dataBuf)
 	}
 	return w.enqueuePhysicalBlock(cb, separator)
 }
@@ -1145,7 +1153,7 @@ func (w *RawColumnWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProper
 	cb := compressedBlockPool.Get().(*compressedBlock)
 	cb.blockBuf.checksummer.Type = w.opts.Checksum
 	cb.physical = block.CompressAndChecksum(
-		&cb.blockBuf.compressedBuf,
+		&cb.blockBuf.dataBuf,
 		b,
 		w.opts.Compression,
 		&cb.blockBuf.checksummer,
@@ -1156,7 +1164,7 @@ func (w *RawColumnWriter) addDataBlock(b, sep []byte, bhp block.HandleWithProper
 		// it to the write queue to be asynchronously written to disk.
 		// TODO(jackson): Should we try to avoid this clone by tracking the
 		// lifetime of the DataBlockWriters?
-		cb.physical = cb.physical.Clone()
+		cb.physical, cb.blockBuf.dataBuf = cb.physical.CloneUsingBuf(cb.blockBuf.dataBuf)
 	}
 	if err := w.enqueuePhysicalBlock(cb, sep); err != nil {
 		return err
diff --git a/sstable/layout.go b/sstable/layout.go
index 09b674ab..e05700fb 100644
--- a/sstable/layout.go
+++ b/sstable/layout.go
@@ -770,7 +770,7 @@ func (w *layoutWriter) writeBlock(
 	b []byte, compression block.Compression, buf *blockBuf,
 ) (block.Handle, error) {
 	return w.writePrecompressedBlock(block.CompressAndChecksum(
-		&buf.compressedBuf, b, compression, &buf.checksummer))
+		&buf.dataBuf, b, compression, &buf.checksummer))
 }
 
 // writePrecompressedBlock writes a pre-compressed block and its
diff --git a/sstable/reader.go b/sstable/reader.go
index 8d4f96df..48853d98 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -507,17 +507,48 @@ func (r *Reader) readBlockInternal(
 	bh block.Handle,
 	initBlockMetadataFn func(*block.Metadata, []byte) error,
 ) (handle block.BufferHandle, _ error) {
-	if h := r.cacheOpts.Cache.Get(r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset); h.Valid() {
-		// Cache hit.
-		if readHandle != nil {
-			readHandle.RecordCacheHit(ctx, int64(bh.Offset), int64(bh.Length+block.TrailerLen))
+	var ch cache.Handle
+	var crh cache.ReadHandle
+	hit := true
+	if env.BufferPool == nil {
+		var errorDuration time.Duration
+		var err error
+		ch, crh, errorDuration, hit, err = r.cacheOpts.Cache.GetWithReadHandle(
+			ctx, r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
+		if errorDuration > 5*time.Millisecond && r.logger.IsTracingEnabled(ctx) {
+			r.logger.Eventf(
+				ctx, "waited for turn when %s time wasted by failed reads", errorDuration.String())
+		}
+		// TODO(sumeer): consider tracing when waited longer than some duration
+		// for turn to do the read.
+		if err != nil {
+			return block.BufferHandle{}, err
+		}
+	} else {
+		// The compaction path uses env.BufferPool, and does not coordinate read
+		// using a cache.ReadHandle. This is ok since only a single compaction is
+		// reading a block.
+		ch = r.cacheOpts.Cache.Get(r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
+		if ch.Valid() {
+			hit = true
 		}
-		env.BlockServedFromCache(bh.Length)
-		return block.CacheBufferHandle(h), nil
+	}
+	// INVARIANT: hit => ch.Valid()
+	if ch.Valid() {
+		if hit {
+			// Cache hit.
+			if readHandle != nil {
+				readHandle.RecordCacheHit(ctx, int64(bh.Offset), int64(bh.Length+block.TrailerLen))
+			}
+			env.BlockServedFromCache(bh.Length)
+		}
+		if invariants.Enabled && crh.Valid() {
+			panic("cache.ReadHandle must not be valid")
+		}
+		return block.CacheBufferHandle(ch), nil
 	}
 
-	// Cache miss.
-
+	// Need to read. First acquire loadBlockSema, if needed.
 	if sema := r.loadBlockSema; sema != nil {
 		if err := sema.Acquire(ctx, 1); err != nil {
 			// An error here can only come from the context.
@@ -525,7 +556,26 @@ func (r *Reader) readBlockInternal(
 		}
 		defer sema.Release(1)
 	}
+	value, err := r.doRead(ctx, env, readHandle, bh, initBlockMetadataFn)
+	if err != nil {
+		if crh.Valid() {
+			crh.SetReadError(err)
+		}
+		return block.BufferHandle{}, err
+	}
+	h := value.MakeHandle(crh, r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
+	return h, nil
+}
 
+// doRead is a helper for readBlockInternal that does the read, checksum
+// check, decompression, and returns either a block.Value or an error.
+func (r *Reader) doRead(
+	ctx context.Context,
+	env readBlockEnv,
+	readHandle objstorage.ReadHandle,
+	bh block.Handle,
+	initBlockMetadataFn func(*block.Metadata, []byte) error,
+) (block.Value, error) {
 	compressed := block.Alloc(int(bh.Length+block.TrailerLen), env.BufferPool)
 	readStopwatch := makeStopwatch()
 	var err error
@@ -548,17 +598,15 @@ func (r *Reader) readBlockInternal(
 	}
 	if err != nil {
 		compressed.Release()
-		return block.BufferHandle{}, err
+		return block.Value{}, err
 	}
 	env.BlockRead(bh.Length, readDuration)
-	if err := checkChecksum(r.checksumType, compressed.BlockData(), bh, r.cacheOpts.FileNum); err != nil {
+	if err = checkChecksum(r.checksumType, compressed.BlockData(), bh, r.cacheOpts.FileNum); err != nil {
 		compressed.Release()
-		return block.BufferHandle{}, err
+		return block.Value{}, err
 	}
-
 	typ := block.CompressionIndicator(compressed.BlockData()[bh.Length])
 	compressed.Truncate(int(bh.Length))
-
 	var decompressed block.Value
 	if typ == block.NoCompressionIndicator {
 		decompressed = compressed
@@ -567,23 +615,21 @@ func (r *Reader) readBlockInternal(
 		decodedLen, prefixLen, err := block.DecompressedLen(typ, compressed.BlockData())
 		if err != nil {
 			compressed.Release()
-			return block.BufferHandle{}, err
+			return block.Value{}, err
 		}
-
 		decompressed = block.Alloc(decodedLen, env.BufferPool)
 		err = block.DecompressInto(typ, compressed.BlockData()[prefixLen:], decompressed.BlockData())
 		compressed.Release()
 		if err != nil {
 			decompressed.Release()
-			return block.BufferHandle{}, err
+			return block.Value{}, err
 		}
 	}
-	if err := initBlockMetadataFn(decompressed.BlockMetadata(), decompressed.BlockData()); err != nil {
+	if err = initBlockMetadataFn(decompressed.BlockMetadata(), decompressed.BlockData()); err != nil {
 		decompressed.Release()
-		return block.BufferHandle{}, err
+		return block.Value{}, err
 	}
-	h := decompressed.MakeHandle(r.cacheOpts.Cache, r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
-	return h, nil
+	return decompressed, nil
 }
 
 func (r *Reader) readMetaindex(
diff --git a/sstable/rowblk_writer.go b/sstable/rowblk_writer.go
index d2578013..be02de82 100644
--- a/sstable/rowblk_writer.go
+++ b/sstable/rowblk_writer.go
@@ -461,19 +461,17 @@ type blockBuf struct {
 	// blockTrailerLen bytes, (5 * binary.MaxVarintLen64) bytes, and most
 	// likely large enough for a block handle with properties.
 	tmp [blockHandleLikelyMaxLen]byte
-	// compressedBuf is the destination buffer for compression. It is re-used over the
-	// lifetime of the blockBuf, avoiding the allocation of a temporary buffer for each block.
-	compressedBuf []byte
-	checksummer   block.Checksummer
+	// dataBuf is the destination buffer for compression, or (in some cases where
+	// compression is not used) for storing a copy of the data. It is re-used over
+	// the lifetime of the blockBuf, avoiding the allocation of a temporary buffer
+	// for each block.
+	dataBuf     []byte
+	checksummer block.Checksummer
 }
 
 func (b *blockBuf) clear() {
-	// We can't assign b.compressedBuf[:0] to compressedBuf because snappy relies
-	// on the length of the buffer, and not the capacity to determine if it needs
-	// to make an allocation.
-	*b = blockBuf{
-		compressedBuf: b.compressedBuf, checksummer: b.checksummer,
-	}
+	b.tmp = [blockHandleLikelyMaxLen]byte{}
+	b.dataBuf = b.dataBuf[:0]
 }
 
 // A dataBlockBuf holds all the state required to compress and write a data block to disk.
@@ -545,7 +543,7 @@ func (d *dataBlockBuf) finish() {
 }
 
 func (d *dataBlockBuf) compressAndChecksum(c block.Compression) {
-	d.physical = block.CompressAndChecksum(&d.compressedBuf, d.uncompressed, c, &d.checksummer)
+	d.physical = block.CompressAndChecksum(&d.dataBuf, d.uncompressed, c, &d.checksummer)
 }
 
 func (d *dataBlockBuf) shouldFlush(


Test patch:
diff --git a/internal/cache/read_shard_test.go b/internal/cache/read_shard_test.go
new file mode 100644
index 00000000..d6c18216
--- /dev/null
+++ b/internal/cache/read_shard_test.go
@@ -0,0 +1,264 @@
+// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package cache
+
+import (
+	"context"
+	crand "crypto/rand"
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/stretchr/testify/require"
+)
+
+type testReader struct {
+	ctx     context.Context
+	id      ID
+	fileNum base.DiskFileNum
+	offset  uint64
+	re      *readEntry
+	mu      struct {
+		*sync.Mutex
+		finishedWait     bool
+		waitedNeedToRead bool
+		waitedErr        error
+		waitedValue      string
+
+		cond *sync.Cond
+	}
+}
+
+func newTestReader(
+	ctx context.Context, id ID, fileNum base.DiskFileNum, offset uint64, mu *sync.Mutex,
+) *testReader {
+	r := &testReader{
+		ctx:     ctx,
+		id:      id,
+		fileNum: fileNum,
+		offset:  offset,
+	}
+	r.mu.Mutex = mu
+	r.mu.cond = sync.NewCond(mu)
+	return r
+}
+
+func (r *testReader) getAsync(shard *shard) *string {
+	h, re := shard.getWithMaybeReadEntry(r.id, r.fileNum, r.offset, true)
+	if h.Valid() {
+		v := string(h.RawBuffer())
+		h.Release()
+		return &v
+	}
+	r.re = re
+	go func() {
+		h, _, err := re.waitForReadPermissionOrHandle(r.ctx)
+		r.mu.Lock()
+		defer r.mu.Unlock()
+		r.mu.finishedWait = true
+		r.mu.cond.Signal()
+		if h.Valid() {
+			r.mu.waitedValue = string(h.RawBuffer())
+			h.Release()
+			return
+		}
+		if err != nil {
+			r.mu.waitedErr = err
+			return
+		}
+		r.mu.waitedNeedToRead = true
+	}()
+	return nil
+}
+
+func (r *testReader) waitUntilFinishedWait() (*string, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for !r.mu.finishedWait {
+		r.mu.cond.Wait()
+	}
+	if r.mu.waitedNeedToRead {
+		return nil, nil
+	}
+	if r.mu.waitedErr != nil {
+		return nil, r.mu.waitedErr
+	}
+	return &r.mu.waitedValue, nil
+}
+
+func (r *testReader) setReadValue(t *testing.T, v string) {
+	val := Alloc(len(v))
+	copy(val.Buf(), []byte(v))
+	h := ReadHandle{entry: r.re}.SetReadValue(val)
+	require.Equal(t, v, string(h.RawBuffer()))
+	h.Release()
+}
+
+func (r *testReader) setError(err error) {
+	ReadHandle{entry: r.re}.SetReadError(err)
+}
+
+func TestReadShard(t *testing.T) {
+	var c *shard
+	var readers map[string]*testReader
+	var mu sync.Mutex
+
+	datadriven.RunTest(t, "testdata/read_shard",
+		func(t *testing.T, td *datadriven.TestData) string {
+			switch td.Cmd {
+			case "init":
+				var maxSize int64
+				td.ScanArgs(t, "max-size", &maxSize)
+				c = &shard{}
+				c.init(maxSize)
+				if len(readers) > 0 {
+					t.Fatalf("have %d readers that have not completed", len(readers))
+				}
+				readers = map[string]*testReader{}
+				return ""
+
+			case "get":
+				var name string
+				td.ScanArgs(t, "name", &name)
+				if _, ok := readers[name]; ok {
+					t.Fatalf("reader with name %s already exists", name)
+				}
+				var id, fileNum, offset int
+				td.ScanArgs(t, "id", &id)
+				td.ScanArgs(t, "file-num", &fileNum)
+				td.ScanArgs(t, "offset", &offset)
+				ctx := context.Background()
+				if td.HasArg("cancelled-context") {
+					var cancelFunc context.CancelFunc
+					ctx, cancelFunc = context.WithCancel(ctx)
+					cancelFunc()
+				}
+				r := newTestReader(ctx, ID(id), base.DiskFileNum(fileNum), uint64(offset), &mu)
+				val := r.getAsync(c)
+				if val != nil {
+					return fmt.Sprintf("val: %s", *val)
+				}
+				readers[name] = r
+				time.Sleep(10 * time.Millisecond)
+				return fmt.Sprintf("waiting\nmap-len: %d", c.readShard.lenForTesting())
+
+			case "wait":
+				var name string
+				td.ScanArgs(t, "name", &name)
+				val, err := readers[name].waitUntilFinishedWait()
+				if val != nil || err != nil {
+					delete(readers, name)
+					if val != nil {
+						return fmt.Sprintf("val: %s\nmap-len: %d", *val, c.readShard.lenForTesting())
+					}
+					if err != nil {
+						return fmt.Sprintf("err: %s\nmap-len: %d", err.Error(), c.readShard.lenForTesting())
+					}
+				}
+				return fmt.Sprintf("turn to read\nmap-len: %d", c.readShard.lenForTesting())
+
+			case "set-read-value":
+				var name string
+				td.ScanArgs(t, "name", &name)
+				var val string
+				td.ScanArgs(t, "val", &val)
+				readers[name].setReadValue(t, val)
+				delete(readers, name)
+				time.Sleep(10 * time.Millisecond)
+				return fmt.Sprintf("map-len: %d", c.readShard.lenForTesting())
+
+			case "set-error":
+				var name string
+				td.ScanArgs(t, "name", &name)
+				readers[name].setError(errors.Errorf("read error: %s", name))
+				delete(readers, name)
+				time.Sleep(10 * time.Millisecond)
+				return fmt.Sprintf("map-len: %d", c.readShard.lenForTesting())
+
+			default:
+				return fmt.Sprintf("unknown command: %s", td.Cmd)
+
+			}
+		})
+}
+
+// testSyncReaders is the config for multiple readers concurrently reading the
+// same block.
+type testSyncReaders struct {
+	// id, fileNum, offset are the key.
+	id      ID
+	fileNum base.DiskFileNum
+	offset  uint64
+	// val will be the value read, if not found in the cache.
+	val []byte
+	// numReaders is the number of concurrent readers.
+	numReaders int
+	// readerWithErrIndex is a reader that will have a read error and hand a
+	// turn to another reader.
+	readerWithErrIndex int
+	// sleepDuration is the duration that the reader with the turns sleeps
+	// before setting the value or error.
+	sleepDuration time.Duration
+	// wg is used to wait for all reader goroutines to be done.
+	wg sync.WaitGroup
+}
+
+func TestReadShardConcurrent(t *testing.T) {
+	cache := New(rand.Int63n(20 << 10))
+	defer cache.Unref()
+	var differentReaders []*testSyncReaders
+	// 50 blocks are read.
+	for i := 0; i < 50; i++ {
+		valLen := rand.Intn(100) + 1
+		val := make([]byte, valLen)
+		crand.Read(val)
+		readers := &testSyncReaders{
+			id:                 ID(rand.Uint64()),
+			fileNum:            base.DiskFileNum(rand.Uint64()),
+			offset:             rand.Uint64(),
+			val:                val,
+			numReaders:         5,
+			readerWithErrIndex: rand.Intn(5),
+			sleepDuration:      time.Duration(rand.Intn(2)) * time.Millisecond,
+		}
+		readers.wg.Add(readers.numReaders)
+		differentReaders = append(differentReaders, readers)
+	}
+	for _, r := range differentReaders {
+		for j := 0; j < r.numReaders; j++ {
+			go func(r *testSyncReaders, index int) {
+				h, rh, _, _, err := cache.GetWithReadHandle(context.Background(), r.id, r.fileNum, r.offset)
+				require.NoError(t, err)
+				if h.Valid() {
+					require.Equal(t, r.val, h.RawBuffer())
+					h.Release()
+					r.wg.Done()
+					return
+				}
+				if r.sleepDuration > 0 {
+					time.Sleep(r.sleepDuration)
+				}
+				if r.readerWithErrIndex == index {
+					rh.SetReadError(errors.Errorf("error"))
+					r.wg.Done()
+					return
+				}
+				v := Alloc(len(r.val))
+				copy(v.Buf(), r.val)
+				h = rh.SetReadValue(v)
+				h.Release()
+				r.wg.Done()
+			}(r, j)
+		}
+	}
+	for _, r := range differentReaders {
+		r.wg.Wait()
+	}
+}
diff --git a/internal/cache/testdata/read_shard b/internal/cache/testdata/read_shard
new file mode 100644
index 00000000..48b0321b
--- /dev/null
+++ b/internal/cache/testdata/read_shard
@@ -0,0 +1,163 @@
+# Cache cannot store anything.
+init max-size=0
+----
+
+# Get misses the cache and has turn to read.
+get name=s1_f2_off3 id=1 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+wait name=s1_f2_off3
+----
+turn to read
+map-len: 1
+
+# Sets the value read, but will be forgotten by the cache.
+set-read-value name=s1_f2_off3 val=potato
+----
+map-len: 0
+
+# Another get for the same block, which gets the turn to read.
+get name=s1_f2_off3 id=1 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+wait name=s1_f2_off3
+----
+turn to read
+map-len: 1
+
+# A concurrent get for the same block.
+get name=another_s1_f2_off3 id=1 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+# A get with a different id but the same file-num and offset. It is a
+# different block.
+get name=s2_f2_off3 id=2 file-num=2 offset=3
+----
+waiting
+map-len: 2
+
+# The get with id=2 also gets a turn to read, and sets an error.
+wait name=s2_f2_off3
+----
+turn to read
+map-len: 2
+
+set-error name=s2_f2_off3
+----
+map-len: 1
+
+# The first get reads and sets a value.
+set-read-value name=s1_f2_off3 val=eggplant
+----
+map-len: 0
+
+# The concurrent get sees this value, which was never successfully stored in
+# the cache.
+wait name=another_s1_f2_off3
+----
+val: eggplant
+map-len: 0
+
+# Get with id=2 gets a turn to read.
+get name=s2_f2_off3 id=2 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+wait name=s2_f2_off3
+----
+turn to read
+map-len: 1
+
+# Two more concurrent gets, one with a cancelled context.
+get name=s2_f2_off3_2 id=2 file-num=2 offset=3 cancelled-context
+----
+waiting
+map-len: 1
+
+get name=s2_f2_off3_3 id=2 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+wait name=s2_f2_off3_2
+----
+err: context canceled
+map-len: 1
+
+# The get with the read turn sets an error.
+set-error name=s2_f2_off3
+----
+map-len: 1
+
+# The waiting reader gets a turn.
+wait name=s2_f2_off3_3
+----
+turn to read
+map-len: 1
+
+# Two more concurrent gets.
+get name=s2_f2_off3_4 id=2 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+get name=s2_f2_off3_5 id=2 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+# The get with a read turn sets a value.
+set-read-value name=s2_f2_off3_3 val=aubergine
+----
+map-len: 0
+
+wait name=s2_f2_off3_4
+----
+val: aubergine
+map-len: 0
+
+wait name=s2_f2_off3_5
+----
+val: aubergine
+map-len: 0
+
+# Cache with some storage capacity
+init max-size=100
+----
+
+# Get misses the cache and has turn to read.
+get name=s1_f2_off3 id=1 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+wait name=s1_f2_off3
+----
+turn to read
+map-len: 1
+
+# Concurrent reader.
+get name=s1_f2_off3_2 id=1 file-num=2 offset=3
+----
+waiting
+map-len: 1
+
+set-read-value name=s1_f2_off3 val=brinjal
+----
+map-len: 0
+
+wait name=s1_f2_off3_2
+----
+val: brinjal
+map-len: 0
+
+get name=s1_f2_off3_3 id=1 file-num=2 offset=3
+----
+val: brinjal
diff --git a/internal/compact/iterator_test.go b/internal/compact/iterator_test.go
index ba337628..d4c85bf2 100644
--- a/internal/compact/iterator_test.go
+++ b/internal/compact/iterator_test.go
@@ -192,15 +192,18 @@ func TestCompactionIter(t *testing.T) {
 						continue
 					}
 					var key *base.InternalKey
-					var value []byte
+					var lv base.LazyValue
 					switch parts[0] {
 					case "first":
-						key, value = iter.First()
+						key, lv = iter.First()
 					case "next":
-						key, value = iter.Next()
+						key, lv = iter.Next()
 					default:
 						d.Fatalf(t, "unknown iter command: %s", parts[0])
 					}
+					value, _, err := lv.Value(nil)
+					require.NoError(t, err)
+
 					if key != nil {
 						snapshotPinned := ""
 						if printSnapshotPinned {
diff --git a/internal/metamorphic/reduce_test.go b/internal/metamorphic/reduce_test.go
index 2acbfc15..e3e81f9a 100644
--- a/internal/metamorphic/reduce_test.go
+++ b/internal/metamorphic/reduce_test.go
@@ -176,7 +176,10 @@ func (r *reducer) try(t *testing.T, ops []string) bool {
 	t.Logf("  Log: %v", logFile)
 
 	// Try to generate a diagram.
-	diagram, err := metamorphic.TryToGenerateDiagram([]byte(strings.Join(ops, "\n")))
+	diagram, err := metamorphic.TryToGenerateDiagram(
+		metamorphic.TestkeysKeyFormat,
+		[]byte(strings.Join(ops, "\n")),
+	)
 	require.NoError(t, err)
 	if diagram != "" {
 		diagramPath := filepath.Join(testRootDir, "diagram")
@@ -210,7 +213,7 @@ func (r *reducer) Run(t *testing.T) {
 	// Try to simplify the keys.
 	opsData := []byte(strings.Join(ops, "\n"))
 	for _, retainSuffixes := range []bool{false, true} {
-		newOpsData := metamorphic.TryToSimplifyKeys(opsData, retainSuffixes)
+		newOpsData := metamorphic.TryToSimplifyKeys(metamorphic.TestkeysKeyFormat, opsData, retainSuffixes)
 		o := strings.Split(strings.TrimSpace(string(newOpsData)), "\n")
 		if r.try(t, o) {
 			return
diff --git a/metamorphic/diagram_test.go b/metamorphic/diagram_test.go
index 2557e9c4..1a660d11 100644
--- a/metamorphic/diagram_test.go
+++ b/metamorphic/diagram_test.go
@@ -15,7 +15,7 @@ func TestDiagram(t *testing.T) {
 	datadriven.RunTest(t, "testdata/diagram", func(t *testing.T, d *datadriven.TestData) string {
 		switch d.Cmd {
 		case "diagram":
-			res, err := TryToGenerateDiagram([]byte(d.Input))
+			res, err := TryToGenerateDiagram(TestkeysKeyFormat, []byte(d.Input))
 			if err != nil {
 				return err.Error()
 			}
diff --git a/metamorphic/generator_test.go b/metamorphic/generator_test.go
index e7f2ab18..d1b0dc37 100644
--- a/metamorphic/generator_test.go
+++ b/metamorphic/generator_test.go
@@ -132,7 +132,8 @@ func TestGeneratorRandom(t *testing.T) {
 	generateFromSeed := func(cfg OpConfig) string {
 		rng := rand.New(rand.NewPCG(0, seed))
 		count := ops.Uint64(rng)
-		return formatOps(generate(rng, count, cfg, newKeyManager(cfg.numInstances)))
+		km := newKeyManager(cfg.numInstances)
+		return formatOps(km.kf, generate(rng, count, cfg, km))
 	}
 
 	for i := range cfgs {
diff --git a/metamorphic/key_generator_test.go b/metamorphic/key_generator_test.go
deleted file mode 100644
index 4b8c2446..00000000
--- a/metamorphic/key_generator_test.go
+++ /dev/null
@@ -1,41 +0,0 @@
-// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-package metamorphic
-
-import (
-	"testing"
-
-	"github.com/cockroachdb/pebble"
-	"github.com/cockroachdb/pebble/internal/randvar"
-)
-
-func TestGenerateRandKeyInRange(t *testing.T) {
-	rng := randvar.NewRand()
-	g := newKeyGenerator(newKeyManager(1 /* numInstances */), rng, DefaultOpConfig())
-	// Seed 100 initial keys.
-	for i := 0; i < 100; i++ {
-		_ = g.RandKey(1.0)
-	}
-	for i := 0; i < 100; i++ {
-		a := g.RandKey(0.01)
-		b := g.RandKey(0.01)
-		// Ensure unique prefixes; required by RandKeyInRange.
-		for g.equal(g.prefix(a), g.prefix(b)) {
-			b = g.RandKey(0.01)
-		}
-		if v := g.cmp(a, b); v > 0 {
-			a, b = b, a
-		}
-		kr := pebble.KeyRange{Start: a, End: b}
-		for j := 0; j < 10; j++ {
-			k := g.RandKeyInRange(0.05, kr)
-			if g.cmp(k, a) < 0 {
-				t.Errorf("generated random key %q outside range %s", k, kr)
-			} else if g.cmp(k, b) >= 0 {
-				t.Errorf("generated random key %q outside range %s", k, kr)
-			}
-		}
-	}
-}
diff --git a/metamorphic/key_manager_test.go b/metamorphic/key_manager_test.go
index 9c5bf328..1125a501 100644
--- a/metamorphic/key_manager_test.go
+++ b/metamorphic/key_manager_test.go
@@ -8,6 +8,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/internal/randvar"
 	"github.com/stretchr/testify/require"
 )
@@ -77,6 +78,7 @@ func printKeys(w io.Writer, keys [][]byte) {
 func TestKeyManager(t *testing.T) {
 	var buf bytes.Buffer
 	km := newKeyManager(1 /* numInstances */)
+	kf := TestkeysKeyFormat
 	datadriven.RunTest(t, "testdata/key_manager", func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "reset":
@@ -130,7 +132,7 @@ func TestKeyManager(t *testing.T) {
 						t.Fatalf("expected 1 op but found %d", len(ops))
 					}
 					km.update(ops[0])
-					fmt.Fprintf(&buf, "[%s]\n", ops[0])
+					fmt.Fprintf(&buf, "[%s]\n", ops[0].formattedString(kf))
 				default:
 					return fmt.Sprintf("unrecognized subcommand %q", fields[0])
 				}
@@ -171,3 +173,38 @@ func TestLoadPrecedingKeys(t *testing.T) {
 	require.Subset(t, km.globalKeys, km2.globalKeys)
 	require.Subset(t, km.globalKeyPrefixes, km2.globalKeyPrefixes)
 }
+
+func TestGenerateRandKeyInRange(t *testing.T) {
+	rng := randvar.NewRand()
+
+	for _, kf := range knownKeyFormats {
+		t.Run(kf.Name, func(t *testing.T) {
+			km := newKeyManager(1 /* numInstances */)
+			g := kf.NewGenerator(km, rng, DefaultOpConfig())
+			// Seed 100 initial keys.
+			for i := 0; i < 100; i++ {
+				_ = g.RandKey(1.0)
+			}
+			for i := 0; i < 100; i++ {
+				a := g.RandKey(0.01)
+				b := g.RandKey(0.01)
+				// Ensure unique prefixes; required by RandKeyInRange.
+				for kf.Comparer.Equal(kf.Comparer.Split.Prefix(a), kf.Comparer.Split.Prefix(b)) {
+					b = g.RandKey(0.01)
+				}
+				if v := kf.Comparer.Compare(a, b); v > 0 {
+					a, b = b, a
+				}
+				kr := pebble.KeyRange{Start: a, End: b}
+				for j := 0; j < 10; j++ {
+					k := g.RandKeyInRange(0.05, kr)
+					if kf.Comparer.Compare(k, a) < 0 {
+						t.Errorf("generated random key %q outside range %s", k, kr)
+					} else if kf.Comparer.Compare(k, b) >= 0 {
+						t.Errorf("generated random key %q outside range %s", k, kr)
+					}
+				}
+			}
+		})
+	}
+}
diff --git a/metamorphic/options_test.go b/metamorphic/options_test.go
index c84e9ecc..52c34650 100644
--- a/metamorphic/options_test.go
+++ b/metamorphic/options_test.go
@@ -174,9 +174,10 @@ func TestBlockPropertiesParse(t *testing.T) {
 	metaDir := t.TempDir()
 
 	rng := rand.New(rand.NewPCG(0, fixedSeed))
-	ops := generate(rng, numOps, presetConfigs[0], newKeyManager(1 /* numInstances */))
+	km := newKeyManager(1 /* numInstances */)
+	ops := generate(rng, numOps, presetConfigs[0], km)
 	opsPath := filepath.Join(metaDir, "ops")
-	formattedOps := formatOps(ops)
+	formattedOps := formatOps(km.kf, ops)
 	require.NoError(t, os.WriteFile(opsPath, []byte(formattedOps), 0644))
 
 	runDir := filepath.Join(metaDir, "run")
diff --git a/metamorphic/parser_test.go b/metamorphic/parser_test.go
index 2b2352af..2978b77d 100644
--- a/metamorphic/parser_test.go
+++ b/metamorphic/parser_test.go
@@ -14,14 +14,18 @@ import (
 )
 
 func TestParser(t *testing.T) {
+	kf := TestkeysKeyFormat
 	datadriven.RunTest(t, "testdata/parser", func(t *testing.T, d *datadriven.TestData) string {
 		switch d.Cmd {
 		case "parse":
-			ops, err := parse([]byte(d.Input), parserOpts{})
+			ops, err := parse([]byte(d.Input), parserOpts{
+				parseFormattedUserKey:       kf.ParseFormattedKey,
+				parseFormattedUserKeySuffix: kf.ParseFormattedKeySuffix,
+			})
 			if err != nil {
 				return err.Error()
 			}
-			return formatOps(ops)
+			return formatOps(kf, ops)
 		default:
 			return fmt.Sprintf("unknown command: %s", d.Cmd)
 		}
@@ -37,8 +41,9 @@ func TestParserRandom(t *testing.T) {
 				cfg = multiInstanceConfig()
 				cfg.numInstances = 2
 			}
-			ops := generate(randvar.NewRand(), 10000, cfg, newKeyManager(cfg.numInstances))
-			src := formatOps(ops)
+			km := newKeyManager(cfg.numInstances)
+			ops := generate(randvar.NewRand(), 10000, cfg, km)
+			src := formatOps(km.kf, ops)
 
 			parsedOps, err := parse([]byte(src), parserOpts{})
 			if err != nil {
@@ -50,14 +55,18 @@ func TestParserRandom(t *testing.T) {
 }
 
 func TestParserNilBounds(t *testing.T) {
-	formatted := formatOps([]op{
+	kf := TestkeysKeyFormat
+	formatted := formatOps(kf, []op{
 		&newIterOp{
 			readerID: makeObjID(dbTag, 1),
 			iterID:   makeObjID(iterTag, 1),
 			iterOpts: iterOpts{},
 		},
 	})
-	parsedOps, err := parse([]byte(formatted), parserOpts{})
+	parsedOps, err := parse([]byte(formatted), parserOpts{
+		parseFormattedUserKey:       kf.ParseFormattedKey,
+		parseFormattedUserKeySuffix: kf.ParseFormattedKeySuffix,
+	})
 	require.NoError(t, err)
 	require.Equal(t, 1, len(parsedOps))
 	v := parsedOps[0].(*newIterOp)
diff --git a/metamorphic/simplify_test.go b/metamorphic/simplify_test.go
index de986bc6..9da18f4e 100644
--- a/metamorphic/simplify_test.go
+++ b/metamorphic/simplify_test.go
@@ -16,7 +16,7 @@ func TestSimplifyKeys(t *testing.T) {
 		switch d.Cmd {
 		case "simplify-keys":
 			retainSuffixes := d.HasArg("retain-suffixes")
-			res := TryToSimplifyKeys([]byte(d.Input), retainSuffixes)
+			res := TryToSimplifyKeys(TestkeysKeyFormat, []byte(d.Input), retainSuffixes)
 			return string(res)
 
 		default:
diff --git a/metamorphic/test.go b/metamorphic/test.go
index c1fe1c32..1ccf4bb4 100644
--- a/metamorphic/test.go
+++ b/metamorphic/test.go
@@ -431,7 +431,8 @@ func (t *Test) runOp(idx int, h historyRecorder) {
 			opTimeout *= 8
 		}
 		timer = time.AfterFunc(opTimeout, func() {
-			panic(fmt.Sprintf("operation took longer than %s: %s", opTimeout, op.String()))
+			panic(fmt.Sprintf("operation took longer than %s: %s",
+				opTimeout, op.formattedString(t.testOpts.KeyFormat)))
 		})
 	}
 	op.run(t, h)
@@ -586,7 +587,8 @@ func computeSynchronizationPoints(ops []op) (opsWaitOn [][]int, opsDone []chan s
 			// have been referenced by some other operation before it's used as
 			// a receiver.
 			if i != 0 && receiver.tag() != dbTag {
-				panic(fmt.Sprintf("op %s on receiver %s; first reference of %s", ops[i].String(), receiver, receiver))
+				panic(fmt.Sprintf("op %T on receiver %s; first reference of %s",
+					ops[i], receiver, receiver))
 			}
 			// The initOp is a little special. We do want to store the objects it's
 			// syncing on, in `lastOpReference`.
@@ -611,7 +613,7 @@ func computeSynchronizationPoints(ops []op) (opsWaitOn [][]int, opsDone []chan s
 		for _, syncObjID := range o.syncObjs() {
 			if vi, vok := lastOpReference[syncObjID]; vok {
 				if vi == i {
-					panic(fmt.Sprintf("%s has %s as syncObj multiple times", ops[i].String(), syncObjID))
+					panic(fmt.Sprintf("%T has %s as syncObj multiple times", ops[i], syncObjID))
 				}
 				opsWaitOn[i] = append(opsWaitOn[i], vi)
 			}
diff --git a/metamorphic/testdata/diagram b/metamorphic/testdata/diagram
index a46a94a6..07148416 100644
--- a/metamorphic/testdata/diagram
+++ b/metamorphic/testdata/diagram
@@ -5,7 +5,7 @@ snap9 = db2.NewSnapshot("a", "z")
 db2.RangeKeyDelete("d", "f")
 db2.Compact("g", "i", true /* parallelize */)
 db2.Replicate(db1, "b", "f")
-iter25 = db1.NewIter("", "", 2 /* key types */, 0, 0, false /* use L6 filters */, "@7" /* masking suffix */)
+iter25 = db1.NewIter("", "", 2 /* key types */, "", "", false /* use L6 filters */, "@7" /* masking suffix */)
 iter25.SeekGE("e", "")
 ----
                                                         Init(2 /* dbs */, 49 /* batches */, 63 /* iters */, 45 /* snapshots */, 0 /* externalObjs */)
@@ -14,7 +14,7 @@ iter25.SeekGE("e", "")
                   |-----------|                         db2.RangeKeyDelete("d", "f")
                                     |-----------|       db2.Compact("g", "i", true /* parallelize */)
       |-----------------------|                         db2.Replicate(db1, "b", "f")
-                                                        iter25 = db1.NewIter("", "", 2 /* key types */, 0, 0, false /* use L6 filters */, "@7" /* masking suffix */)
+                                                        iter25 = db1.NewIter("", "", 2 /* key types */, "", "", false /* use L6 filters */, "@7" /* masking suffix */)
                         |                               iter25.SeekGE("e", "")
 |-----|-----|-----|-----|-----|-----|-----|-----|-----|
 a     b     c     d     e     f     g     h     i     z
diff --git a/sstable/block/compression_test.go b/sstable/block/compression_test.go
index 23a950a5..108a3888 100644
--- a/sstable/block/compression_test.go
+++ b/sstable/block/compression_test.go
@@ -23,6 +23,9 @@ func TestCompressionRoundtrip(t *testing.T) {
 	rng := rand.New(rand.NewPCG(0, seed))
 
 	for compression := DefaultCompression + 1; compression < NCompression; compression++ {
+		if compression == NoCompression {
+			continue
+		}
 		t.Run(compression.String(), func(t *testing.T) {
 			payload := make([]byte, 1+rng.IntN(10<<10 /* 10 KiB */))
 			for i := range payload {
diff --git a/sstable/colblk/index_block_test.go b/sstable/colblk/index_block_test.go
index 87f5d6a6..3be6ff40 100644
--- a/sstable/colblk/index_block_test.go
+++ b/sstable/colblk/index_block_test.go
@@ -125,7 +125,7 @@ func TestIndexIterInitHandle(t *testing.T) {
 	d := (*IndexBlockDecoder)(unsafe.Pointer(v.BlockMetadata()))
 	d.Init(blockData)
 
-	h := v.MakeHandle(c, cache.ID(1), base.DiskFileNum(1), 0)
+	h := v.SetInCacheForTesting(c, cache.ID(1), base.DiskFileNum(1), 0)
 	defer h.Release()
 
 	getBlockAndIterate := func(it *IndexIter) {
diff --git a/sstable/colblk/keyspan_test.go b/sstable/colblk/keyspan_test.go
index 9592dabc..9f7c99d8 100644
--- a/sstable/colblk/keyspan_test.go
+++ b/sstable/colblk/keyspan_test.go
@@ -88,7 +88,7 @@ func TestKeyspanBlockPooling(t *testing.T) {
 	copy(v.BlockData(), b)
 	d := (*KeyspanDecoder)(unsafe.Pointer(v.BlockMetadata()))
 	d.Init(v.BlockData())
-	v.MakeHandle(c, cache.ID(1), base.DiskFileNum(1), 0).Release()
+	v.SetInCacheForTesting(c, cache.ID(1), base.DiskFileNum(1), 0).Release()
 
 	getBlockAndIterate := func() {
 		h := c.Get(cache.ID(1), base.DiskFileNum(1), 0)
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 45c7748c..be01c02f 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -531,7 +531,7 @@ func TestBlockBufClear(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	b1 := &blockBuf{}
 	b1.tmp[0] = 1
-	b1.compressedBuf = make([]byte, 1)
+	b1.dataBuf = make([]byte, 1)
 	b1.clear()
 	testBlockBufClear(t, b1, &blockBuf{})
 }
@@ -539,7 +539,7 @@ func TestBlockBufClear(t *testing.T) {
 func TestClearDataBlockBuf(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	d := newDataBlockBuf(1, block.ChecksumTypeCRC32c)
-	d.blockBuf.compressedBuf = make([]byte, 1)
+	d.blockBuf.dataBuf = make([]byte, 1)
 	d.dataBlock.Add(ikey("apple"), nil)
 	d.dataBlock.Add(ikey("banana"), nil)