Repository: cockroachdb/pebble
Pull request #4408 (2025-03-20 20:51:07)

crl-release-25.1: backport changes to support metamorphic cockroachkvs
Backport the changes required to support use of cockroachkvs in the metamorphic test. Columnar blocks are enabled on 25.1 by default, so we'll benefit from having this test coverage in the metamorphic test.

cockroach-teamcity: 

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

RaduBerinde: Once you merge, we should kick off the nightly builds on this branch to make sure everything is green before bumping CRDB.
RaduBerinde: Oh, never mind, in this branch we already ran all the checks in the PR.

Patch:
diff --git a/cockroachkvs/blockproperties.go b/cockroachkvs/blockproperties.go
index 22ebd0b9..d5bc61b1 100644
--- a/cockroachkvs/blockproperties.go
+++ b/cockroachkvs/blockproperties.go
@@ -45,6 +45,14 @@ type MVCCWallTimeIntervalRangeKeyMask struct {
 	sstable.BlockIntervalFilter
 }
 
+// Init initializees the mask and its block interval filter.
+func (m *MVCCWallTimeIntervalRangeKeyMask) Init() {
+	m.BlockIntervalFilter.Init(
+		mvccWallTimeIntervalCollector,
+		0, math.MaxUint64,
+		MVCCBlockIntervalSuffixReplacer{})
+}
+
 // SetSuffix implements the pebble.BlockPropertyFilterMask interface.
 func (m *MVCCWallTimeIntervalRangeKeyMask) SetSuffix(suffix []byte) error {
 	if len(suffix) == 0 {
diff --git a/cockroachkvs/cockroachkvs.go b/cockroachkvs/cockroachkvs.go
index 3bdae183..959b7e9b 100644
--- a/cockroachkvs/cockroachkvs.go
+++ b/cockroachkvs/cockroachkvs.go
@@ -70,6 +70,9 @@ var Comparer = base.Comparer{
 	FormatKey: FormatKey,
 
 	Separator: func(dst, a, b []byte) []byte {
+		if len(a) == 0 || len(b) == 0 {
+			panic(errors.AssertionFailedf("cannot separate empty keys"))
+		}
 		aKey, ok := getKeyPartFromEngineKey(a)
 		if !ok {
 			return append(dst, a...)
@@ -94,6 +97,13 @@ var Comparer = base.Comparer{
 	},
 
 	Successor: func(dst, a []byte) []byte {
+		// If a is empty, the key consisting of just a single 0x00 byte is the
+		// smallest non-empty key ≥ a. We don't return an empty key because
+		// empty keys are problematic when combined with a synthetic prefix
+		// (they form a key that has no trailing sentinel byte).
+		if len(a) == 0 {
+			return append(dst, 0x00)
+		}
 		aKey, ok := getKeyPartFromEngineKey(a)
 		if !ok {
 			return append(dst, a...)
@@ -122,6 +132,8 @@ var Comparer = base.Comparer{
 		return append(append(dst, a...), 0)
 	},
 
+	ValidateKey: validateEngineKey,
+
 	Name: "cockroach_comparator",
 }
 
@@ -184,6 +196,26 @@ func EncodeTimestamp(key []byte, walltime uint64, logical uint32) []byte {
 	return key
 }
 
+// NewTimestampSuffix allocates and encodes a new suffix byte slice encoding the
+// provided timestamp.
+func NewTimestampSuffix(wallTime uint64, logical uint32) []byte {
+	switch {
+	case wallTime == 0 && logical == 0:
+		return nil
+	case logical == 0:
+		v := make([]byte, suffixLenWithWall)
+		binary.BigEndian.PutUint64(v, wallTime)
+		v[suffixLenWithWall-1] = suffixLenWithWall
+		return v
+	default:
+		v := make([]byte, suffixLenWithLogical)
+		binary.BigEndian.PutUint64(v, wallTime)
+		binary.BigEndian.PutUint32(v[8:], logical)
+		v[suffixLenWithLogical-1] = suffixLenWithLogical
+		return v
+	}
+}
+
 // DecodeMVCCTimestampSuffix decodes an MVCC timestamp from its Pebble representation,
 // including the length suffix.
 func DecodeMVCCTimestampSuffix(encodedTS []byte) (wallTime uint64, logical uint32, err error) {
@@ -266,7 +298,7 @@ func Split(key []byte) int {
 		return 0
 	}
 	if invariants.Enabled {
-		checkEngineKey(key)
+		validateEngineKey.MustValidate(key)
 	}
 	// Last byte is the version length + 1 when there is a version, else it is
 	// 0.
@@ -281,8 +313,8 @@ func Compare(a, b []byte) int {
 		return cmp.Compare(len(a), len(b))
 	}
 	if invariants.Enabled {
-		checkEngineKey(a)
-		checkEngineKey(b)
+		validateEngineKey.MustValidate(a)
+		validateEngineKey.MustValidate(b)
 	}
 	aSuffixLen := int(a[len(a)-1])
 	aSuffixStart := len(a) - aSuffixLen
@@ -343,8 +375,8 @@ func Equal(a, b []byte) bool {
 		return len(a) == len(b)
 	}
 	if invariants.Enabled {
-		checkEngineKey(a)
-		checkEngineKey(b)
+		validateEngineKey.MustValidate(a)
+		validateEngineKey.MustValidate(b)
 	}
 	aSuffixLen := int(a[len(a)-1])
 	aSuffixStart := len(a) - aSuffixLen
@@ -743,34 +775,60 @@ func (ks *cockroachKeySeeker) IsLowerBound(k []byte, syntheticSuffix []byte) boo
 	if len(syntheticSuffix) > 0 {
 		return ComparePointSuffixes(syntheticSuffix, k[len(roachKey)+1:]) >= 0
 	}
+	firstRowWall := ks.mvccWallTimes.At(0)
+	firstLogical := ks.mvccLogical.At(0)
+	// If the first row's WallTime and LogicalTime are both zero, the first row
+	// is either (a) unversioned (and sorts before all other suffixes) or (b) is
+	// a non-MVCC key with an untyped version.
+	if firstRowWall == 0 && firstLogical == 0 {
+		return ComparePointSuffixes(ks.untypedVersions.At(0), version) >= 0
+	}
+
 	var wallTime uint64
 	var logicalTime uint32
 	switch len(version) {
 	case engineKeyNoVersion:
+		// The zero-length version is the smallest possible suffix.
+		return true
 	case engineKeyVersionWallTimeLen:
 		wallTime = binary.BigEndian.Uint64(version[:8])
 	case engineKeyVersionWallAndLogicalTimeLen, engineKeyVersionWallLogicalAndSyntheticTimeLen:
 		wallTime = binary.BigEndian.Uint64(version[:8])
 		logicalTime = binary.BigEndian.Uint32(version[8:12])
 	default:
-		// The provided key `k` is not a MVCC key. Assert that the first key in
-		// the block is also not an MVCC key. If it were, that would mean there
-		// exists both a MVCC key and a non-MVCC key with the same prefix.
+		// The provided key `k` is not a MVCC key.
 		//
-		// TODO(jackson): Double check that we'll never produce index separators
-		// that are invalid version lengths.
-		if invariants.Enabled && ks.mvccWallTimes.At(0) != 0 {
-			panic("comparing timestamp with untyped suffix")
+		// The first row IS an MVCC key. We don't expect this to happen in
+		// practice, so this path is not performance sensitive. We reconsistute
+		// the first row's MVCC suffix and then compare it to the provided key's
+		// non-MVCC suffix.
+
+		//gcassert:noescape
+		var buf [13]byte
+		//gcassert:inline
+		binary.BigEndian.PutUint64(buf[:], firstRowWall)
+		var firstRowSuffix []byte
+		if firstLogical == 0 {
+			buf[8] = 9
+			firstRowSuffix = buf[:9]
+		} else {
+			//gcassert:inline
+			binary.BigEndian.PutUint32(buf[8:], uint32(firstLogical))
+			buf[12] = 13
+			firstRowSuffix = buf[:13]
 		}
-		return ComparePointSuffixes(ks.untypedVersions.At(0), version) >= 0
+		return ComparePointSuffixes(firstRowSuffix, version) >= 0
 	}
 
 	// NB: The sign comparison is inverted because suffixes are sorted such that
 	// the largest timestamps are "smaller" in the lexicographical ordering.
-	if v := cmp.Compare(ks.mvccWallTimes.At(0), wallTime); v != 0 {
+	if v := cmp.Compare(firstRowWall, wallTime); v != 0 {
+		// The empty-suffixed zero-timestamped key sorts first. If the first row
+		// has zero wall and logical times, it sorts before k and k is not a
+		// lower bound.
 		return v < 0
 	}
-	return cmp.Compare(uint32(ks.mvccLogical.At(0)), logicalTime) <= 0
+	return cmp.Compare(uint32(firstLogical), logicalTime) <= 0
 }
 
 // SeekGE is part of the KeySeeker interface.
@@ -983,16 +1041,20 @@ func (ks *cockroachKeySeeker) MaterializeUserKeyWithSyntheticSuffix(
 //go:linkname memmove runtime.memmove
 func memmove(to, from unsafe.Pointer, n uintptr)
 
-func checkEngineKey(k []byte) {
+var validateEngineKey base.ValidateKey = func(k []byte) error {
 	if len(k) == 0 {
-		panic(errors.AssertionFailedf("empty key"))
+		// It would be nice to assert that the key is not empty, but
+		// unfortunately separator keys within index blocks may be empty in the
+		// case that a sstable (and its single data block) has no keys.
+		return nil
 	}
 	if int(k[len(k)-1]) >= len(k) {
-		panic(errors.AssertionFailedf("malformed key terminator byte: %x", k))
+		return errors.AssertionFailedf("malformed key terminator byte: %x", k)
 	}
 	if k[len(k)-1] == 1 {
-		panic(errors.AssertionFailedf("invalid key terminator byte 1"))
+		return errors.AssertionFailedf("invalid key terminator byte 1 in key: %x", k)
 	}
+	return nil
 }
 
 // FormatKey returns a formatter for the user key.
@@ -1003,6 +1065,10 @@ func FormatKey(key []byte) fmt.Formatter {
 type formatKey []byte
 
 func (fk formatKey) Format(f fmt.State, c rune) {
+	if len(fk) == 0 {
+		fmt.Fprint(f, "<empty>")
+		return
+	}
 	i := Split(fk)
 	fmt.Fprintf(f, "%s", []byte(fk)[:i-1])
 	if i == len(fk) {
@@ -1013,6 +1079,11 @@ func (fk formatKey) Format(f fmt.State, c rune) {
 	formatKeySuffix(fk[i:]).Format(f, c)
 }
 
+// FormatKeySuffix returns a formatter for the user key suffix.
+func FormatKeySuffix(suffix []byte) fmt.Formatter {
+	return formatKeySuffix(suffix)
+}
+
 type formatKeySuffix []byte
 
 func (fk formatKeySuffix) Format(f fmt.State, c rune) {
@@ -1051,7 +1122,15 @@ func (fk formatKeySuffix) Format(f fmt.State, c rune) {
 //     c. <WALLTIME-SECONDS>,<LOGICAL>
 //     d. <WALLTIME-SECONDS>.<WALLTIME-NANOS>,<LOGICAL>
 //  2. A lock table key in the format: <STRENGTH-BYTE>,<TXNUUID>
+//  3. A raw hex-encoded string prefixed with "hex:".
 func ParseFormattedKey(formattedKey string) []byte {
+	if strings.HasPrefix(formattedKey, "hex:") {
+		decoded, err := hex.DecodeString(formattedKey[4:])
+		if err != nil {
+			panic(errors.Newf("invalid hex-encoded key: %w", err))
+		}
+		return decoded
+	}
 	i := strings.IndexByte(formattedKey, '@')
 	if i == -1 {
 		// Suffix-less.
diff --git a/compaction.go b/compaction.go
index 780283f2..b39ab15d 100644
--- a/compaction.go
+++ b/compaction.go
@@ -1500,7 +1500,7 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 		Err:        err,
 	}
 	if err == nil {
-		validateVersionEdit(ve, d.opts.Experimental.KeyValidationFunc, d.opts.Comparer.FormatKey, d.opts.Logger)
+		validateVersionEdit(ve, d.opts.Comparer.ValidateKey, d.opts.Comparer.FormatKey, d.opts.Logger)
 		for i := range ve.NewFiles {
 			e := &ve.NewFiles[i]
 			info.Output = append(info.Output, e.Meta.TableInfo())
@@ -2356,7 +2356,7 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 
 	info.Duration = d.timeNow().Sub(startTime)
 	if err == nil {
-		validateVersionEdit(ve, d.opts.Experimental.KeyValidationFunc, d.opts.Comparer.FormatKey, d.opts.Logger)
+		validateVersionEdit(ve, d.opts.Comparer.ValidateKey, d.opts.Comparer.FormatKey, d.opts.Logger)
 		err = func() error {
 			var err error
 			d.mu.versions.logLock()
@@ -3241,10 +3241,10 @@ func (d *DB) newCompactionOutput(
 // validateVersionEdit validates that start and end keys across new and deleted
 // files in a versionEdit pass the given validation function.
 func validateVersionEdit(
-	ve *versionEdit, validateFn func([]byte) error, format base.FormatKey, logger Logger,
+	ve *versionEdit, vk base.ValidateKey, format base.FormatKey, logger Logger,
 ) {
 	validateKey := func(f *manifest.FileMetadata, key []byte) {
-		if err := validateFn(key); err != nil {
+		if err := vk.Validate(key); err != nil {
 			logger.Fatalf("pebble: version edit validation failed (key=%s file=%s): %v", format(key), f, err)
 		}
 	}
diff --git a/ingest.go b/ingest.go
index 08491994..c409c289 100644
--- a/ingest.go
+++ b/ingest.go
@@ -98,6 +98,10 @@ func ingestValidateKey(opts *Options, key *InternalKey) error {
 		return base.CorruptionErrorf("pebble: external sstable has non-zero seqnum: %s",
 			key.Pretty(opts.Comparer.FormatKey))
 	}
+	if err := opts.Comparer.ValidateKey.Validate(key.UserKey); err != nil {
+		return base.CorruptionErrorf("pebble: external sstable has corrupted key: %s, %w",
+			key.Pretty(opts.Comparer.FormatKey), err)
+	}
 	return nil
 }
 
diff --git a/internal/base/comparer.go b/internal/base/comparer.go
index 8e277e60..92713bad 100644
--- a/internal/base/comparer.go
+++ b/internal/base/comparer.go
@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"unicode/utf8"
 
+	"github.com/cockroachdb/crlib/crbytes"
 	"github.com/cockroachdb/errors"
 )
 
@@ -124,11 +125,18 @@ type FormatValue func(key, value []byte) fmt.Formatter
 //
 // For example, if a and b are the []byte equivalents of the strings "black" and
 // "blue", then the function may append "blb" to dst.
+//
+// Callers must guarantee that len(a) > 0 and len(b) > 0.
 type Separator func(dst, a, b []byte) []byte
 
 // Successor appends to dst a shortened key k given a key a such that
 // Compare(a, k) <= 0. A simple implementation may return a unchanged.
 // The appended key k must be valid to pass to Compare.
+//
+// The parameter a may be an empty slice even if an empty slice was never
+// committed to Pebble and is not a valid key representation. If a is the empty
+// slice, Successor must return a valid key but otherwise has no other
+// constraints.
 type Successor func(dst, a []byte) []byte
 
 // ImmediateSuccessor is invoked with a prefix key ([Split(a) == len(a)]) and
@@ -229,6 +237,10 @@ type Comparer struct {
 	// FormatValue is optional.
 	FormatValue FormatValue
 
+	// ValidateKey is an optional function that determines whether a key is
+	// valid according to this Comparer's key encoding.
+	ValidateKey ValidateKey
+
 	// Name is the name of the comparer.
 	//
 	// The on-disk format stores the comparer name, and opening a database with a
@@ -308,14 +320,15 @@ var DefaultComparer = &Comparer{
 	FormatKey: DefaultFormatter,
 
 	Separator: func(dst, a, b []byte) []byte {
-		i, n := SharedPrefixLen(a, b), len(dst)
+		if len(a) == 0 || len(b) == 0 {
+			panic(errors.AssertionFailedf("empty keys"))
+		}
+
+		i := crbytes.CommonPrefix(a, b)
+		n := len(dst)
 		dst = append(dst, a...)
 
-		min := len(a)
-		if min > len(b) {
-			min = len(b)
-		}
-		if i >= min {
+		if i == len(a) || i == len(b) {
 			// Do not shorten if one string is a prefix of the other.
 			return dst
 		}
@@ -362,25 +375,6 @@ var DefaultComparer = &Comparer{
 	Name: "leveldb.BytewiseComparator",
 }
 
-// SharedPrefixLen returns the largest i such that a[:i] equals b[:i].
-// This function can be useful in implementing the Comparer interface.
-func SharedPrefixLen(a, b []byte) int {
-	i, n := 0, len(a)
-	if n > len(b) {
-		n = len(b)
-	}
-	asUint64 := func(c []byte, i int) uint64 {
-		return binary.LittleEndian.Uint64(c[i:])
-	}
-	for i < n-7 && asUint64(a, i) == asUint64(b, i) {
-		i += 8
-	}
-	for i < n && a[i] == b[i] {
-		i++
-	}
-	return i
-}
-
 // MinUserKey returns the smaller of two user keys. If one of the keys is nil,
 // the other one is returned.
 func MinUserKey(cmp Compare, a, b []byte) []byte {
@@ -439,13 +433,54 @@ func MakeAssertComparer(c Comparer) Comparer {
 		ComparePointSuffixes: c.ComparePointSuffixes,
 		CompareRangeSuffixes: c.CompareRangeSuffixes,
 		AbbreviatedKey:       c.AbbreviatedKey,
-		Separator:            c.Separator,
-		Successor:            c.Successor,
-		ImmediateSuccessor:   c.ImmediateSuccessor,
-		FormatKey:            c.FormatKey,
-		Split:                c.Split,
-		FormatValue:          c.FormatValue,
-		Name:                 c.Name,
+		Separator: func(dst, a, b []byte) []byte {
+			if len(a) == 0 || len(b) == 0 {
+				panic(errors.AssertionFailedf("empty keys"))
+			}
+			ret := c.Separator(dst, a, b)
+			// The Separator func must return a valid key.
+			c.ValidateKey.MustValidate(ret)
+			return ret
+		},
+		Successor: func(dst, a []byte) []byte {
+			ret := c.Successor(dst, a)
+			// The Successor func must return a valid key.
+			c.ValidateKey.MustValidate(ret)
+			return ret
+		},
+		ImmediateSuccessor: func(dst, a []byte) []byte {
+			ret := c.ImmediateSuccessor(dst, a)
+			// The ImmediateSuccessor func must return a valid key.
+			c.ValidateKey.MustValidate(ret)
+			return ret
+		},
+		FormatKey:   c.FormatKey,
+		Split:       c.Split,
+		FormatValue: c.FormatValue,
+		ValidateKey: c.ValidateKey,
+		Name:        c.Name,
+	}
+}
+
+// ValidateKey is a func that determines whether a key is valid according to a
+// particular key encoding. Returns nil if the provided key is a valid, full
+// user key. Implementations must be careful to not mutate the provided key.
+type ValidateKey func([]byte) error
+
+// Validate validates the provided user key. If the func is nil, Validate
+// returns nil.
+func (v ValidateKey) Validate(key []byte) error {
+	if v == nil {
+		return nil
+	}
+	return v(key)
+}
+
+// MustValidate validates the provided user key, panicking if the key is
+// invalid.
+func (v ValidateKey) MustValidate(key []byte) {
+	if err := v.Validate(key); err != nil {
+		panic(err)
 	}
 }
 
@@ -489,6 +524,13 @@ func CheckComparer(c *Comparer, prefixes [][]byte, suffixes [][]byte) error {
 		for i := 1; i < len(suffixes); i++ {
 			a := slices.Concat(p, suffixes[i-1])
 			b := slices.Concat(p, suffixes[i])
+			if err := c.ValidateKey.Validate(a); err != nil {
+				return err
+			}
+			if err := c.ValidateKey.Validate(b); err != nil {
+				return err
+			}
+
 			// Make sure the Compare function agrees with ComparePointSuffixes.
 			if cmp := c.Compare(a, b); cmp > 0 {
 				return errors.Errorf("Compare(%s, %s)=%d, expected <= 0", c.FormatKey(a), c.FormatKey(b), cmp)
@@ -500,9 +542,15 @@ func CheckComparer(c *Comparer, prefixes [][]byte, suffixes [][]byte) error {
 	for _, ap := range prefixes {
 		for _, as := range suffixes {
 			a := slices.Concat(ap, as)
+			if err := c.ValidateKey.Validate(a); err != nil {
+				return err
+			}
 			for _, bp := range prefixes {
 				for _, bs := range suffixes {
 					b := slices.Concat(bp, bs)
+					if err := c.ValidateKey.Validate(b); err != nil {
+						return err
+					}
 					result := c.Compare(a, b)
 					if (result == 0) != c.Equal(a, b) {
 						return errors.Errorf("Equal(%s, %s) doesn't agree with Compare", c.FormatKey(a), c.FormatKey(b))
diff --git a/internal/base/internal.go b/internal/base/internal.go
index 0c123eb4..bd6381bf 100644
--- a/internal/base/internal.go
+++ b/internal/base/internal.go
@@ -12,6 +12,8 @@ import (
 	"strings"
 	"sync/atomic"
 
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/redact"
 )
 
@@ -398,6 +400,9 @@ func (k InternalKey) EncodeTrailer() [8]byte {
 func (k InternalKey) Separator(
 	cmp Compare, sep Separator, buf []byte, other InternalKey,
 ) InternalKey {
+	if invariants.Enabled && (len(k.UserKey) == 0 || len(other.UserKey) == 0) {
+		panic(errors.AssertionFailedf("empty keys passed to Separator: %s, %s", k, other))
+	}
 	buf = sep(buf, k.UserKey, other.UserKey)
 	if len(buf) <= len(k.UserKey) && cmp(k.UserKey, buf) < 0 {
 		// The separator user key is physically shorter than k.UserKey (if it is
@@ -416,7 +421,7 @@ func (k InternalKey) Separator(
 // InternalKey.UserKey, though it is valid to pass a nil.
 func (k InternalKey) Successor(cmp Compare, succ Successor, buf []byte) InternalKey {
 	buf = succ(buf, k.UserKey)
-	if len(buf) <= len(k.UserKey) && cmp(k.UserKey, buf) < 0 {
+	if (len(k.UserKey) == 0 || len(buf) <= len(k.UserKey)) && cmp(k.UserKey, buf) < 0 {
 		// The successor user key is physically shorter that k.UserKey (if it is
 		// longer, we'll continue to use "k"), but logically after. Tack on the max
 		// sequence number to the shortened user key. Note that we could tack on
diff --git a/internal/metamorphic/metaflags/meta_flags.go b/internal/metamorphic/metaflags/meta_flags.go
index b8b5ed84..049a701a 100644
--- a/internal/metamorphic/metaflags/meta_flags.go
+++ b/internal/metamorphic/metaflags/meta_flags.go
@@ -60,7 +60,8 @@ func (c *CommonFlags) KeyFormat() metamorphic.KeyFormat {
 
 // KeyFormats is a map of available key formats.
 var KeyFormats = map[string]metamorphic.KeyFormat{
-	"testkeys": metamorphic.TestkeysKeyFormat,
+	"testkeys":     metamorphic.TestkeysKeyFormat,
+	"cockroachkvs": metamorphic.CockroachKeyFormat,
 }
 
 func initCommonFlags() *CommonFlags {
diff --git a/metamorphic/build.go b/metamorphic/build.go
index 4b5af83a..8c8bee8c 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -97,6 +97,7 @@ func writeSSTForIngestion(
 		if err != nil {
 			return nil, err
 		}
+		t.opts.Comparer.ValidateKey.MustValidate(k.K.UserKey)
 		if err := w.Raw().AddWithForceObsolete(k.K, valBytes, false); err != nil {
 			return nil, err
 		}
@@ -111,7 +112,11 @@ func writeSSTForIngestion(
 			if syntheticSuffix.IsSet() {
 				panic("synthetic suffix with RangeDel")
 			}
-			if err := w.DeleteRange(outputKey(span.Start, nil), outputKey(span.End, nil)); err != nil {
+			start := outputKey(span.Start, nil)
+			end := outputKey(span.End, nil)
+			t.opts.Comparer.ValidateKey.MustValidate(start)
+			t.opts.Comparer.ValidateKey.MustValidate(end)
+			if err := w.DeleteRange(start, end); err != nil {
 				return nil, err
 			}
 		}
@@ -139,6 +144,8 @@ func writeSSTForIngestion(
 				End:   outputKey(span.End, nil),
 				Keys:  make([]keyspan.Key, 0, len(span.Keys)),
 			}
+			t.opts.Comparer.ValidateKey.MustValidate(collapsed.Start)
+			t.opts.Comparer.ValidateKey.MustValidate(collapsed.End)
 			keys := span.Keys
 			if syntheticSuffix.IsSet() {
 				keys = slices.Clone(span.Keys)
diff --git a/metamorphic/cockroachkvs.go b/metamorphic/cockroachkvs.go
new file mode 100644
index 00000000..77b096a1
--- /dev/null
+++ b/metamorphic/cockroachkvs.go
@@ -0,0 +1,422 @@
+// Copyright 2025 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 (
+	"fmt"
+	"math/rand/v2"
+	"slices"
+
+	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/cockroachkvs"
+	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/colblk"
+)
+
+// CockroachKeyFormat provides a KeyFormat implementation that uses
+// CockroachDB's key encoding (as defined in the cockroachkvs package).
+var CockroachKeyFormat = KeyFormat{
+	Name:                    "cockroachkvs",
+	Comparer:                &cockroachkvs.Comparer,
+	KeySchema:               func() *colblk.KeySchema { return &cockroachkvs.KeySchema }(),
+	BlockPropertyCollectors: cockroachkvs.BlockPropertyCollectors,
+	FormatKey: func(k UserKey) string {
+		if len(k) == 0 {
+			return ""
+		}
+		return fmt.Sprint(cockroachkvs.FormatKey(k))
+	},
+	FormatKeySuffix: func(s UserKeySuffix) string {
+		if len(s) == 0 {
+			return ""
+		}
+		return fmt.Sprint(cockroachkvs.FormatKeySuffix(s))
+	},
+	ParseFormattedKey: func(formattedKey string) UserKey {
+		return UserKey(cockroachkvs.ParseFormattedKey(formattedKey))
+	},
+	ParseFormattedKeySuffix: func(formattedKeySuffix string) UserKeySuffix {
+		return UserKeySuffix(cockroachkvs.ParseFormattedKeySuffix(formattedKeySuffix))
+	},
+	NewGenerator: func(km *keyManager, rng *rand.Rand, cfg OpConfig) KeyGenerator {
+		return &cockroachKeyGenerator{
+			keyManager: km,
+			rng:        rng,
+			cfg:        cfg,
+			// TODO(jackson): Vary maxLogical.
+			suffixSpace: cockroachSuffixKeyspace{maxLogical: 2},
+		}
+	},
+	NewSuffixFilterMask: func() pebble.BlockPropertyFilterMask {
+		return &cockroachkvs.MVCCWallTimeIntervalRangeKeyMask{}
+	},
+	NewSuffixBlockPropertyFilter: func(minSuffix, maxSuffix []byte) sstable.BlockPropertyFilter {
+		minWallTime, _, err := cockroachkvs.DecodeMVCCTimestampSuffix(maxSuffix)
+		if err != nil {
+			panic(err)
+		}
+		maxWallTime, _, err := cockroachkvs.DecodeMVCCTimestampSuffix(minSuffix)
+		if err != nil {
+			panic(err)
+		}
+		return cockroachkvs.NewMVCCTimeIntervalFilter(minWallTime, maxWallTime)
+	},
+}
+
+type cockroachKeyGenerator struct {
+	keyManager  *keyManager
+	rng         *rand.Rand
+	cfg         OpConfig
+	suffixSpace cockroachSuffixKeyspace
+}
+
+// RecordPrecedingKey may be invoked before generating keys to inform the key
+// generator of a key that was previously generated and used within a related
+// test context.
+func (kg *cockroachKeyGenerator) RecordPrecedingKey(key []byte) {
+	// If the key has a suffix that's larger than the current max suffix,
+	// ratchet up the maximum of the distribution of suffixes.
+	if i := cockroachkvs.Comparer.Split(key); i < len(key) {
+		suffixIdx := kg.suffixSpace.ToSuffixIndex(key[i:])
+		if suffixIdx > suffixIndex(kg.cfg.writeSuffixDist.Max()) {
+			diff := uint64(suffixIdx) - kg.cfg.writeSuffixDist.Max()
+			kg.cfg.writeSuffixDist.IncMax(diff)
+		}
+	}
+}
+
+// ExtendPrefix extends the given prefix key with additional bytes,
+// returning a new prefix that sorts after the given prefix.
+func (kg *cockroachKeyGenerator) ExtendPrefix(prefix []byte) []byte {
+	// Copy prefix and strip the delimiter byte.
+	p := append(make([]byte, 0, len(prefix)+3), prefix[:len(prefix)-1]...)
+	p = append(p, randBytes(kg.rng, 1, 3)...)
+	p = append(p, 0x00) // Delimiter byte
+	return p
+}
+
+// RandKey returns a random key (either a previously known key, or a new key).
+func (kg *cockroachKeyGenerator) 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 *cockroachKeyGenerator) 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 *cockroachKeyGenerator) RandPrefix(newPrefix float64) []byte {
+	prefixes := kg.keyManager.prefixes()
+	if len(prefixes) > 0 && kg.rng.Float64() > newPrefix {
+		return pickOneUniform(kg.rng, prefixes)
+	}
+
+	// Use a new prefix.
+	for {
+		prefix := kg.generateKeyWithSuffix(4, 12, 0)
+		if !kg.keyManager.prefixExists(prefix) {
+			if !kg.keyManager.addNewKey(prefix) {
+				panic("key must not exist if prefix doesn't exist")
+			}
+			return prefix
+		}
+	}
+}
+
+// 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.
+func (kg *cockroachKeyGenerator) SkewedSuffix(incMaxProb float64) []byte {
+	if suffixIdx := kg.skewedSuffixInt(incMaxProb); suffixIdx != 0 {
+		return kg.suffixSpace.ToMaterializedSuffix(suffixIdx)
+	}
+	return nil
+}
+
+// skewedSuffixInt is a helper of SkewedSuffix which returns the unencoded
+// suffix as an integer.
+func (kg *cockroachKeyGenerator) skewedSuffixInt(incMaxProb float64) suffixIndex {
+	if kg.rng.Float64() < incMaxProb {
+		kg.cfg.writeSuffixDist.IncMax(1)
+	}
+	return suffixIndex(kg.cfg.writeSuffixDist.Uint64(kg.rng))
+}
+
+// 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 *cockroachKeyGenerator) IncMaxSuffix() []byte {
+	kg.cfg.writeSuffixDist.IncMax(1)
+	s := suffixIndex(kg.cfg.writeSuffixDist.Max())
+	return kg.suffixSpace.ToMaterializedSuffix(s)
+}
+
+// SuffixRange generates a new uniformly random range of suffixes (low, high]
+// such that high is guaranteed to be strictly greater (as defined by
+// ComparePointSuffixes) than low.
+//
+// The high suffix may be nil, in which case the suffix range represents all
+// suffixes ≥ low.
+func (kg *cockroachKeyGenerator) SuffixRange() (low, high []byte) {
+	a := kg.uniformSuffixInt()
+	b := kg.uniformSuffixInt()
+	if a < b {
+		a, b = b, a
+	} else if a == b {
+		a++
+	}
+	return kg.suffixSpace.ToMaterializedSuffix(a), kg.suffixSpace.ToMaterializedSuffix(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 *cockroachKeyGenerator) UniformSuffix() []byte {
+	if suffix := kg.uniformSuffixInt(); suffix != 0 {
+		return kg.suffixSpace.ToMaterializedSuffix(suffix)
+	}
+	return nil
+}
+
+// uniformSuffixInt is a helper of UniformSuffix which returns the suffix
+// index.
+func (kg *cockroachKeyGenerator) uniformSuffixInt() suffixIndex {
+	maxVal := kg.cfg.writeSuffixDist.Max()
+	return suffixIndex(kg.rng.Int64N(int64(maxVal)))
+}
+
+// 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 *cockroachKeyGenerator) randKey(
+	newKeyProbability float64, bounds *pebble.KeyRange,
+) []byte {
+	var knownKeys [][]byte
+	if bounds == nil {
+		knownKeys = kg.keyManager.knownKeys()
+	} else {
+		if cockroachkvs.Compare(bounds.Start, bounds.End) >= 0 {
+			panic(fmt.Sprintf("invalid bounds [%q, %q)", bounds.Start, bounds.End))
+		}
+		knownKeys = kg.keyManager.knownKeysInRange(*bounds)
+	}
+	switch {
+	case len(knownKeys) > 0 && kg.rng.Float64() > newKeyProbability:
+		// Use an existing user key.
+		return pickOneUniform(kg.rng, knownKeys)
+
+	case len(knownKeys) > 0 && kg.rng.Float64() > kg.cfg.newPrefix:
+		// Use an existing prefix but a new suffix, producing a new user key.
+		prefixes := kg.keyManager.prefixes()
+
+		// If we're constrained to a key range, find which existing prefixes
+		// fall within that key range.
+		if bounds != nil {
+			s, _ := slices.BinarySearchFunc(prefixes, bounds.Start, cockroachkvs.Compare)
+			e, _ := slices.BinarySearchFunc(prefixes, bounds.End, cockroachkvs.Compare)
+			prefixes = prefixes[s:e]
+		}
+
+		if len(prefixes) > 0 {
+			for {
+				// Pick a prefix on each iteration in case most or all suffixes are
+				// already in use for any individual prefix.
+				p := kg.rng.IntN(len(prefixes))
+				suffix := suffixIndex(kg.cfg.writeSuffixDist.Uint64(kg.rng))
+
+				var key []byte
+				if suffix > 0 {
+					key = append(append(key, prefixes[p]...), kg.suffixSpace.ToMaterializedSuffix(suffix)...)
+				} else {
+					key = append(key, prefixes[p]...)
+				}
+				if bounds == nil || (cockroachkvs.Compare(key, bounds.Start) >= 0 &&
+					cockroachkvs.Compare(key, bounds.End) < 0) {
+					if kg.keyManager.addNewKey(key) {
+						return key
+					}
+				}
+
+				// If the generated key already existed, or the generated key
+				// fell outside the provided bounds, increase the suffix
+				// distribution and loop.
+				kg.cfg.writeSuffixDist.IncMax(1)
+			}
+		}
+		// Otherwise fall through to generating a new prefix.
+	}
+
+	if bounds == nil {
+		suffixIdx := kg.skewedSuffixInt(0.01)
+		for {
+			key := kg.generateKeyWithSuffix(4, 12, suffixIdx)
+			if !kg.keyManager.prefixExists(kg.keyManager.kf.Comparer.Split.Prefix(key)) {
+				if !kg.keyManager.addNewKey(key) {
+					panic("key must not exist if prefix doesn't exist")
+				}
+				return key
+			}
+		}
+	}
+	// We need to generate a key between the bounds.
+	startPrefix, startSuffixIdx := kg.suffixSpace.Split(bounds.Start)
+	endPrefix, endSuffixIdx := kg.suffixSpace.Split(bounds.End)
+
+	var prefix []byte
+	var suffixIdx suffixIndex
+	if cockroachkvs.Equal(startPrefix, endPrefix) {
+		prefix = startPrefix
+		// Bounds have the same prefix, generate a suffix in-between.
+		if startSuffixIdx <= endSuffixIdx {
+			panic(fmt.Sprintf("invalid bounds [%q, %q)", bounds.Start, bounds.End))
+		}
+		suffixIdx = kg.skewedSuffixInt(0.01)
+		for i := 0; !(startSuffixIdx >= suffixIdx && endSuffixIdx < suffixIdx); i++ {
+			if i > 10 {
+				// This value is always >= startSuffix and < endSuffix.
+				suffixIdx = (startSuffixIdx + endSuffixIdx) / 2
+				break
+			}
+			// The suffix we want must exist in the current suffix range, we don't
+			// want to keep increasing it here.
+			suffixIdx = kg.skewedSuffixInt(0)
+		}
+	} else {
+		prefix = append(testkeys.RandomPrefixInRange(
+			startPrefix[:len(startPrefix)-1], // Strip the delimiter byte.
+			endPrefix[:len(endPrefix)-1],     // Strip the delimiter byte.
+			kg.rng,
+		), 0x00) // Add back the delimiter byte.
+		suffixIdx = kg.skewedSuffixInt(0.01)
+		if cockroachkvs.Equal(prefix, startPrefix) {
+			// We can't use a suffix which sorts before startSuffix.
+			for i := 0; suffixIdx > startSuffixIdx; i++ {
+				if i > 10 {
+					suffixIdx = startSuffixIdx
+					break
+				}
+				suffixIdx = kg.skewedSuffixInt(0)
+			}
+		}
+	}
+	key := slices.Clip(prefix)
+	if suffixIdx != 0 {
+		key = append(key, kg.suffixSpace.ToMaterializedSuffix(suffixIdx)...)
+	}
+	if cockroachkvs.Compare(key, bounds.Start) < 0 || cockroachkvs.Compare(key, bounds.End) >= 0 {
+		panic(fmt.Sprintf("invalid randKey %q; bounds: [%q, %q) %v %v",
+			key, bounds.Start, bounds.End,
+			cockroachkvs.Compare(key, bounds.Start),
+			cockroachkvs.Compare(key, bounds.End)))
+	}
+	// We might (rarely) produce an existing key here, that's ok.
+	kg.keyManager.addNewKey(key)
+	return key
+}
+
+// generateKeyWithSuffix generates a key with a random prefix and the suffix
+// corresponding to the provided suffix index. If the given suffix index is 0,
+// the key will not have a suffix.
+func (kg *cockroachKeyGenerator) generateKeyWithSuffix(
+	minPrefixLen, maxPrefixLen int, suffixIdx suffixIndex,
+) []byte {
+	prefix := randCockroachPrefix(kg.rng, minPrefixLen, maxPrefixLen)
+	if suffixIdx == 0 {
+		return prefix
+	}
+	return append(prefix, kg.suffixSpace.ToMaterializedSuffix(suffixIdx)...)
+}
+
+func randCockroachPrefix(rng *rand.Rand, minLen, maxLen int) []byte {
+	n := minLen + rng.IntN(maxLen-minLen+1)
+	if n == 0 {
+		return nil
+	}
+	// NB: The actual random values are not particularly important. We only use
+	// lowercase letters because that makes visual determination of ordering
+	// easier, rather than having to remember the lexicographic ordering of
+	// uppercase vs lowercase, or letters vs numbers vs punctuation.
+	const letters = "abcdefghijklmnopqrstuvwxyz"
+	const lettersLen = uint64(len(letters))
+	const lettersCharsPerRand = 12 // floor(log(math.MaxUint64)/log(lettersLen))
+
+	var r uint64
+	var q int
+	buf := make([]byte, n+1)
+	for i := 0; i < n; i++ {
+		if q == 0 {
+			r = rng.Uint64()
+			q = lettersCharsPerRand
+		}
+		buf[i] = letters[r%lettersLen]
+		r = r / lettersLen
+		q--
+	}
+	buf[n] = 0x00 // Delimiter byte
+	return buf
+}
+
+// A suffixIndex represents a unique suffix. The suffixIndex exists within a
+// one-dimensional space of int64s, but is remapped into a two-dimensional space
+// of MVCC timestamps of (WallTime, Logical) tuples.
+type suffixIndex int64
+
+// cockroackSuffixKeyspace defines the mapping between a one-dimensional
+// suffixIndex and the suffix it represents within the two-dimensional
+// (wallTime, logical) space.
+//
+// The mapping is configued by the value of maxLogical, with maxLogical+1
+// possible logical timestamps at each wall time.
+//
+// TODO(jackson): Update to disallow non-zero logical timestamps when the wall
+// time is zero if we begin to prohibit it.
+//
+//	+------------------------------------------------------+
+//	| suffix  |              maxLogical                    |
+//	| index   | 0      | 1      | 2      | 3      | 4      |
+//	+------------------------------------------------------+
+//	|  0      | (0,0)  | (0,0)  | (0,0)  | (0,0)  | (0,0)  |
+//	|  1      | (1,0)  | (0,1)  | (0,1)  | (0,1)  | (0,1)  |
+//	|  2      | (2,0)  | (1,0)  | (0,2)  | (0,2)  | (0,2)  |
+//	|  3      | (3,0)  | (1,1)  | (1,0)  | (0,3)  | (0,3)  |
+//	|  4      | (4,0)  | (2,0)  | (1,1)  | (1,0)  | (0,4)  |
+//	|  5      | (5,0)  | (2,1)  | (1,2)  | (1,1)  | (1,0)  |
+//	|  6      | (6,0)  | (3,0)  | (2,0)  | (1,2)  | (1,1)  |
+//	|  7      | (7,0)  | (3,1)  | (2,1)  | (1,3)  | (1,2)  |
+//	+------------------------------------------------------+
+type cockroachSuffixKeyspace struct {
+	maxLogical int64
+}
+
+func (ks cockroachSuffixKeyspace) ToMaterializedSuffix(s suffixIndex) []byte {
+	// There are maxLogical+1 possible logical timestamps at each wall time.
+	wallTime := int64(s) / (ks.maxLogical + 1)
+	logical := int64(s) % (ks.maxLogical + 1)
+	return cockroachkvs.NewTimestampSuffix(uint64(wallTime), uint32(logical))
+}
+
+func (ks cockroachSuffixKeyspace) ToSuffixIndex(suffix []byte) suffixIndex {
+	wallTime, logical, err := cockroachkvs.DecodeMVCCTimestampSuffix(suffix)
+	if err != nil {
+		panic(err)
+	}
+	return suffixIndex(int64(wallTime)*(ks.maxLogical+1) + int64(logical))
+}
+
+func (ks cockroachSuffixKeyspace) Split(key []byte) (prefix []byte, suffixIdx suffixIndex) {
+	i := cockroachkvs.Split(key)
+	return key[:i], ks.ToSuffixIndex(key[i:])
+}
diff --git a/metamorphic/config.go b/metamorphic/config.go
index 3efd2ae2..f12be213 100644
--- a/metamorphic/config.go
+++ b/metamorphic/config.go
@@ -345,6 +345,7 @@ func mustDynamic(dyn randvar.Dynamic, err error) randvar.Dynamic {
 
 var knownKeyFormats = []KeyFormat{
 	TestkeysKeyFormat,
+	CockroachKeyFormat,
 }
 
 var keyFormatsByName = func() map[string]KeyFormat {
@@ -422,8 +423,12 @@ type KeyGenerator interface {
 	// 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 generates a new uniformly random range of suffixes (low, high]
+	// such that high is guaranteed to be strictly greater (as defined by
+	// ComparePointSuffixes) than low.
+	//
+	// The high suffix may be nil, in which case the suffix range represents all
+	// suffixes ≥ 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
diff --git a/metamorphic/generator.go b/metamorphic/generator.go
index 2ceb68e0..257c420b 100644
--- a/metamorphic/generator.go
+++ b/metamorphic/generator.go
@@ -29,7 +29,13 @@ type iterOpts struct {
 	maskSuffix UserKeySuffix
 
 	// If filterMax is != nil, this iterator will filter out any keys that have
-	// suffixes that don't fall within the range [filterMin,filterMax).
+	// suffixes that don't fall within the range (filterMin,filterMax]
+	// [according to the ordering defined by ComparePointSuffixes]. Note that
+	// suffixes are used to represent MVCC timestamps, and MVCC timestamps are
+	// ordered in numerically descending order, so the timestamp associated with
+	// filterMin is more recent than that associated with filterMax. This means
+	// according to the ordering of ComparePointSuffixes, filterMin > filterMax.
+	//
 	// Additionally, the iterator will be constructed with a block-property
 	// filter that filters out blocks accordingly. Not all OPTIONS hook up the
 	// corresponding block property collector, so block-filtering may still be
@@ -1529,7 +1535,7 @@ func (g *generator) maybeMutateOptions(readerID objID, opts *iterOpts) {
 		}
 
 		// With 1/3 probability, clear existing filter.
-		if opts.filterMax != nil && g.rng.IntN(3) == 0 {
+		if opts.filterMin != nil && g.rng.IntN(3) == 0 {
 			opts.filterMax, opts.filterMin = nil, nil
 		}
 		// With 10% probability, set a filter range.
diff --git a/metamorphic/key_manager.go b/metamorphic/key_manager.go
index a3622be5..5c1cbd06 100644
--- a/metamorphic/key_manager.go
+++ b/metamorphic/key_manager.go
@@ -389,6 +389,7 @@ func newKeyManager(numInstances int, kf KeyFormat) *keyManager {
 // addNewKey adds the given key to the key manager for global key tracking.
 // Returns false iff this is not a new key.
 func (k *keyManager) addNewKey(key []byte) bool {
+	k.kf.Comparer.ValidateKey.MustValidate(key)
 	if k.globalKeysMap[string(key)] {
 		return false
 	}
@@ -409,6 +410,7 @@ func (k *keyManager) addNewKey(key []byte) bool {
 // getOrInit returns the keyMeta for the (objID, key) pair, if it exists, else
 // allocates, initializes and returns a new value.
 func (k *keyManager) getOrInit(id objID, key []byte) *keyMeta {
+	k.kf.Comparer.ValidateKey.MustValidate(key)
 	objKeys := k.objKeyMeta(id)
 	m, ok := objKeys.keys[string(key)]
 	if ok {
diff --git a/metamorphic/ops.go b/metamorphic/ops.go
index e1dae3bf..5b59e186 100644
--- a/metamorphic/ops.go
+++ b/metamorphic/ops.go
@@ -811,6 +811,7 @@ func (o *ingestOp) collapseBatch(
 	if pointIter != nil {
 		var lastUserKey []byte
 		for kv := pointIter.First(); kv != nil; kv = pointIter.Next() {
+			t.opts.Comparer.ValidateKey.MustValidate(kv.K.UserKey)
 			// Ignore duplicate keys.
 			//
 			// Note: this is necessary due to MERGE keys, otherwise it would be
@@ -882,6 +883,7 @@ func (o *ingestOp) collapseBatch(
 				continue
 			}
 			ik := base.MakeInternalKey(key, 0, kind)
+			t.opts.Comparer.ValidateKey.MustValidate(ik.UserKey)
 			if err := collapsed.AddInternalKey(&ik, value, nil); err != nil {
 				return nil, err
 			}
@@ -943,6 +945,8 @@ func (o *ingestAndExciseOp) run(t *Test, h historyRecorder) {
 		return
 	}
 
+	t.opts.Comparer.ValidateKey.MustValidate(o.exciseStart)
+	t.opts.Comparer.ValidateKey.MustValidate(o.exciseEnd)
 	if t.testOpts.useExcise {
 		err = firstError(err, t.withRetries(func() error {
 			_, err := db.IngestAndExcise(context.Background(), []string{path}, nil /* shared */, nil /* external */, pebble.KeyRange{
@@ -1043,6 +1047,9 @@ func (o *ingestExternalFilesOp) run(t *Test, h historyRecorder) {
 	} else {
 		external := make([]pebble.ExternalFile, len(o.objs))
 		for i, obj := range o.objs {
+			t.opts.Comparer.ValidateKey.MustValidate(obj.bounds.Start)
+			t.opts.Comparer.ValidateKey.MustValidate(obj.bounds.End)
+
 			meta := t.getExternalObj(obj.externalObjID)
 			external[i] = pebble.ExternalFile{
 				Locator:           "external",
@@ -1201,7 +1208,7 @@ func (o *newIterOp) run(t *Test, h historyRecorder) {
 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.keyTypes, kf.FormatKeySuffix(o.filterMax), kf.FormatKeySuffix(o.filterMin),
 		o.useL6Filters, kf.FormatKeySuffix(o.maskSuffix))
 }
 
@@ -1268,8 +1275,8 @@ func (o *newIterUsingCloneOp) run(t *Test, h historyRecorder) {
 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))
+		kf.FormatKey(o.upper), o.keyTypes, kf.FormatKeySuffix(o.filterMax),
+		kf.FormatKeySuffix(o.filterMin), o.useL6Filters, kf.FormatKeySuffix(o.maskSuffix))
 }
 
 func (o *newIterUsingCloneOp) receiver() objID { return o.existingIterID }
@@ -1365,7 +1372,7 @@ func (o *iterSetOptionsOp) run(t *Test, h historyRecorder) {
 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.keyTypes, kf.FormatKeySuffix(o.filterMax), kf.FormatKeySuffix(o.filterMin),
 		o.useL6Filters, kf.FormatKeySuffix(o.maskSuffix))
 }
 
@@ -1393,10 +1400,13 @@ func iterOptions(kf KeyFormat, o iterOpts) *pebble.IterOptions {
 	if opts.RangeKeyMasking.Suffix != nil {
 		opts.RangeKeyMasking.Filter = kf.NewSuffixFilterMask
 	}
-	if o.filterMax != nil {
+	if o.filterMin != nil {
 		opts.PointKeyFilters = []pebble.BlockPropertyFilter{
 			kf.NewSuffixBlockPropertyFilter(o.filterMin, o.filterMax),
 		}
+		if o.filterMax != nil && kf.Comparer.ComparePointSuffixes(o.filterMin, o.filterMax) >= 0 {
+			panic(errors.AssertionFailedf("filterMin >= filterMax: %q >= %q", 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.
@@ -1410,10 +1420,10 @@ func iterOptions(kf KeyFormat, o iterOpts) *pebble.IterOptions {
 				// non-timestamped keys.
 				return true
 			}
-			if kf.Comparer.ComparePointSuffixes(k[n:], o.filterMin) < 0 {
+			if kf.Comparer.ComparePointSuffixes(k[n:], o.filterMin) <= 0 {
 				return true
 			}
-			if kf.Comparer.ComparePointSuffixes(k[n:], o.filterMax) >= 0 {
+			if o.filterMax != nil && kf.Comparer.ComparePointSuffixes(k[n:], o.filterMax) > 0 {
 				return true
 			}
 			return false
@@ -1446,9 +1456,9 @@ type iterSeekGEOp struct {
 	derivedReaderID objID
 }
 
-func iteratorPos(i *retryableIter) string {
+func iteratorPos(kf KeyFormat, i *retryableIter) string {
 	var buf bytes.Buffer
-	fmt.Fprintf(&buf, "%q", i.Key())
+	fmt.Fprintf(&buf, "%q", kf.FormatKey(i.Key()))
 	hasPoint, hasRange := i.HasPointAndRange()
 	if hasPoint {
 		fmt.Fprintf(&buf, ",%q", i.Value())
@@ -1457,12 +1467,12 @@ func iteratorPos(i *retryableIter) string {
 	}
 	if hasRange {
 		start, end := i.RangeBounds()
-		fmt.Fprintf(&buf, ",[%q,%q)=>{", start, end)
+		fmt.Fprintf(&buf, ",[%q,%q)=>{", kf.FormatKey(start), kf.FormatKey(end))
 		for i, rk := range i.RangeKeys() {
 			if i > 0 {
 				fmt.Fprint(&buf, ",")
 			}
-			fmt.Fprintf(&buf, "%q=%q", rk.Suffix, rk.Value)
+			fmt.Fprintf(&buf, "%q=%q", kf.FormatKeySuffix(rk.Suffix), rk.Value)
 		}
 		fmt.Fprint(&buf, "}")
 	} else {
@@ -1503,7 +1513,7 @@ func (o *iterSeekGEOp) run(t *Test, h historyRecorder) {
 	}
 	if valid {
 		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			validStr, iteratorPos(i), i.Error())
+			validStr, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v",
 			o.formattedString(t.testOpts.KeyFormat), validStr, i.Error())
@@ -1548,7 +1558,7 @@ func (o *iterSeekPrefixGEOp) run(t *Test, h historyRecorder) {
 	valid := i.SeekPrefixGE(o.key)
 	if valid {
 		h.Recordf("%s // [%t,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			valid, iteratorPos(i), i.Error())
+			valid, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%t] %v", o.formattedString(t.testOpts.KeyFormat), valid, i.Error())
 	}
@@ -1589,7 +1599,7 @@ func (o *iterSeekLTOp) run(t *Test, h historyRecorder) {
 	}
 	if valid {
 		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			validStr, iteratorPos(i), i.Error())
+			validStr, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
 			validStr, i.Error())
@@ -1624,7 +1634,7 @@ func (o *iterFirstOp) run(t *Test, h historyRecorder) {
 	valid := i.First()
 	if valid {
 		h.Recordf("%s // [%t,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			valid, iteratorPos(i), i.Error())
+			valid, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%t] %v", o.formattedString(t.testOpts.KeyFormat),
 			valid, i.Error())
@@ -1650,7 +1660,7 @@ func (o *iterLastOp) run(t *Test, h historyRecorder) {
 	valid := i.Last()
 	if valid {
 		h.Recordf("%s // [%t,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			valid, iteratorPos(i), i.Error())
+			valid, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%t] %v", o.formattedString(t.testOpts.KeyFormat),
 			valid, i.Error())
@@ -1684,7 +1694,7 @@ func (o *iterNextOp) run(t *Test, h historyRecorder) {
 	}
 	if valid {
 		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			validStr, iteratorPos(i), i.Error())
+			validStr, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
 			validStr, i.Error())
@@ -1713,7 +1723,7 @@ func (o *iterNextPrefixOp) run(t *Test, h historyRecorder) {
 	validStr := validBoolToStr(valid)
 	if valid {
 		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			validStr, iteratorPos(i), i.Error())
+			validStr, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
 			validStr, i.Error())
@@ -1784,7 +1794,7 @@ func (o *iterPrevOp) run(t *Test, h historyRecorder) {
 	}
 	if valid {
 		h.Recordf("%s // [%s,%s] %v", o.formattedString(t.testOpts.KeyFormat),
-			validStr, iteratorPos(i), i.Error())
+			validStr, iteratorPos(t.testOpts.KeyFormat, i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o.formattedString(t.testOpts.KeyFormat),
 			validStr, i.Error())
@@ -2056,12 +2066,17 @@ func (r *replicateOp) runExternalReplicate(
 			if err != nil {
 				panic(err)
 			}
+			t.opts.Comparer.ValidateKey.MustValidate(key.UserKey)
 			return w.Raw().AddWithForceObsolete(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false)
 		},
 		func(start, end []byte, seqNum base.SeqNum) error {
+			t.opts.Comparer.ValidateKey.MustValidate(start)
+			t.opts.Comparer.ValidateKey.MustValidate(end)
 			return w.DeleteRange(start, end)
 		},
 		func(start, end []byte, keys []keyspan.Key) error {
+			t.opts.Comparer.ValidateKey.MustValidate(start)
+			t.opts.Comparer.ValidateKey.MustValidate(end)
 			return w.Raw().EncodeSpan(keyspan.Span{
 				Start: start,
 				End:   end,
@@ -2104,6 +2119,9 @@ func (r *replicateOp) runExternalReplicate(
 		return
 	}
 
+	t.opts.Comparer.ValidateKey.MustValidate(r.start)
+	t.opts.Comparer.ValidateKey.MustValidate(r.end)
+
 	_, err = dest.IngestAndExcise(context.Background(), []string{sstPath}, nil, externalSSTs /* external */, pebble.KeyRange{Start: r.start, End: r.end})
 	h.Recordf("%s // %v", r.formattedString(t.testOpts.KeyFormat), err)
 }
@@ -2136,6 +2154,8 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 		return
 	}
 
+	t.opts.Comparer.ValidateKey.MustValidate(r.start)
+	t.opts.Comparer.ValidateKey.MustValidate(r.end)
 	// 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.formattedString(t.testOpts.KeyFormat), err)
@@ -2170,6 +2190,8 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 			rangeKeys := iter.RangeKeys()
 			rkStart, rkEnd := iter.RangeBounds()
 
+			t.opts.Comparer.ValidateKey.MustValidate(rkStart)
+			t.opts.Comparer.ValidateKey.MustValidate(rkEnd)
 			span := keyspan.Span{Start: rkStart, End: rkEnd, Keys: make([]keyspan.Key, len(rangeKeys))}
 			for i := range rangeKeys {
 				span.Keys[i] = keyspan.Key{
diff --git a/metamorphic/parser.go b/metamorphic/parser.go
index 7256ba09..1da15596 100644
--- a/metamorphic/parser.go
+++ b/metamorphic/parser.go
@@ -93,9 +93,9 @@ func opArgs(op op) (receiverID *objID, targetID *objID, args []interface{}) {
 	case *newIndexedBatchOp:
 		return &t.dbID, &t.batchID, nil
 	case *newIterOp:
-		return &t.readerID, &t.iterID, []interface{}{&t.lower, &t.upper, &t.keyTypes, &t.filterMin, &t.filterMax, &t.useL6Filters, &t.maskSuffix}
+		return &t.readerID, &t.iterID, []interface{}{&t.lower, &t.upper, &t.keyTypes, &t.filterMax, &t.filterMin, &t.useL6Filters, &t.maskSuffix}
 	case *newIterUsingCloneOp:
-		return &t.existingIterID, &t.iterID, []interface{}{&t.refreshBatch, &t.lower, &t.upper, &t.keyTypes, &t.filterMin, &t.filterMax, &t.useL6Filters, &t.maskSuffix}
+		return &t.existingIterID, &t.iterID, []interface{}{&t.refreshBatch, &t.lower, &t.upper, &t.keyTypes, &t.filterMax, &t.filterMin, &t.useL6Filters, &t.maskSuffix}
 	case *newSnapshotOp:
 		return &t.dbID, &t.snapID, []interface{}{&t.bounds}
 	case *newExternalObjOp:
@@ -119,7 +119,7 @@ func opArgs(op op) (receiverID *objID, targetID *objID, args []interface{}) {
 	case *iterSetBoundsOp:
 		return &t.iterID, nil, []interface{}{&t.lower, &t.upper}
 	case *iterSetOptionsOp:
-		return &t.iterID, nil, []interface{}{&t.lower, &t.upper, &t.keyTypes, &t.filterMin, &t.filterMax, &t.useL6Filters, &t.maskSuffix}
+		return &t.iterID, nil, []interface{}{&t.lower, &t.upper, &t.keyTypes, &t.filterMax, &t.filterMin, &t.useL6Filters, &t.maskSuffix}
 	case *singleDeleteOp:
 		return &t.writerID, nil, []interface{}{&t.key, &t.maybeReplaceDelete}
 	case *rangeKeyDeleteOp:
diff --git a/options.go b/options.go
index c991648f..f5c8eb54 100644
--- a/options.go
+++ b/options.go
@@ -654,19 +654,6 @@ type Options struct {
 		// limited by runtime.GOMAXPROCS.
 		FileCacheShards int
 
-		// KeyValidationFunc is a function to validate a user key in an SSTable.
-		//
-		// Currently, this function is used to validate the smallest and largest
-		// keys in an SSTable undergoing compaction. In this case, returning an
-		// error from the validation function will result in a panic at runtime,
-		// given that there is rarely any way of recovering from malformed keys
-		// present in compacted files. By default, validation is not performed.
-		//
-		// Additional use-cases may be added in the future.
-		//
-		// NOTE: callers should take care to not mutate the key being validated.
-		KeyValidationFunc func(userKey []byte) error
-
 		// ValidateOnIngest schedules validation of sstables after they have
 		// been ingested.
 		//
@@ -1165,9 +1152,6 @@ func (o *Options) EnsureDefaults() *Options {
 	if o.Experimental.CompactionLimiter == nil {
 		o.Experimental.CompactionLimiter = &base.DefaultCompactionLimiter{}
 	}
-	if o.Experimental.KeyValidationFunc == nil {
-		o.Experimental.KeyValidationFunc = func([]byte) error { return nil }
-	}
 	if o.KeySchema == "" && len(o.KeySchemas) == 0 {
 		ks := colblk.DefaultKeySchema(o.Comparer, 16 /* bundleSize */)
 		o.KeySchema = ks.Name
diff --git a/sstable/rowblk/rowblk_rewrite.go b/sstable/rowblk/rowblk_rewrite.go
index d1667ee1..48b64cbc 100644
--- a/sstable/rowblk/rowblk_rewrite.go
+++ b/sstable/rowblk/rowblk_rewrite.go
@@ -10,6 +10,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/sstable/block"
 )
 
@@ -78,6 +79,9 @@ func (r *Rewriter) RewriteSuffixes(
 		// in the block, which includes the 1-byte prefix. This is fine since bw
 		// also does not know about the prefix and will preserve it in bw.add.
 		v := kv.InPlaceValue()
+		if invariants.Enabled && invariants.Sometimes(10) {
+			r.comparer.ValidateKey.MustValidate(r.scratchKey.UserKey)
+		}
 		r.writer.Add(r.scratchKey, v)
 		if start.UserKey == nil {
 			// Copy the first key.
diff --git a/sstable/rowblk_writer.go b/sstable/rowblk_writer.go
index 4ce0a9ef..e9aac79c 100644
--- a/sstable/rowblk_writer.go
+++ b/sstable/rowblk_writer.go
@@ -51,6 +51,7 @@ type RawRowWriter struct {
 	compression          block.Compression
 	separator            Separator
 	successor            Successor
+	validateKey          base.ValidateKey
 	tableFormat          TableFormat
 	isStrictObsolete     bool
 	writingToLowestLevel bool
@@ -1186,6 +1187,12 @@ func (w *RawRowWriter) indexEntrySep(
 	} else {
 		sep = prevKey.Separator(w.compare, w.separator, dataBlockBuf.sepScratch[:0], key)
 	}
+	if invariants.Enabled && invariants.Sometimes(25) {
+		if w.compare(prevKey.UserKey, sep.UserKey) > 0 {
+			panic(errors.AssertionFailedf("prevKey.UserKey > sep.UserKey: %s > %s",
+				prevKey.UserKey, sep.UserKey))
+		}
+	}
 	return sep
 }
 
@@ -1214,6 +1221,11 @@ func (w *RawRowWriter) addIndexEntry(
 		// In particular, it must have a non-zero length.
 		return nil
 	}
+	if invariants.Enabled {
+		if err := w.validateKey.Validate(sep.UserKey); err != nil {
+			return err
+		}
+	}
 
 	encoded := bhp.EncodeVarints(tmp)
 	if flushIndexBuf != nil {
@@ -1532,6 +1544,8 @@ func (w *RawRowWriter) Close() (err error) {
 			return err
 		}
 		prevKey := w.dataBlockBuf.dataBlock.CurKey()
+		// NB: prevKey.UserKey may be nil if there are no keys and we're forcing
+		// the creation of an empty data block.
 		if err := w.addIndexEntrySync(prevKey, InternalKey{}, bhp, w.dataBlockBuf.tmp[:]); err != nil {
 			return err
 		}
@@ -1713,6 +1727,7 @@ func newRowWriter(writable objstorage.Writable, o WriterOptions) *RawRowWriter {
 		compression:                o.Compression,
 		separator:                  o.Comparer.Separator,
 		successor:                  o.Comparer.Successor,
+		validateKey:                o.Comparer.ValidateKey,
 		tableFormat:                o.TableFormat,
 		isStrictObsolete:           o.IsStrictObsolete,
 		writingToLowestLevel:       o.WritingToLowestLevel,
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index 8387f189..67b0f34f 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -178,6 +178,12 @@ func rewriteDataBlocksInParallel(
 					if err != nil {
 						return err
 					}
+					if err := r.Comparer.ValidateKey.Validate(output[i].start.UserKey); err != nil {
+						return err
+					}
+					if err := r.Comparer.ValidateKey.Validate(output[i].end.UserKey); err != nil {
+						return err
+					}
 					compressedBuf = compressedBuf[:cap(compressedBuf)]
 					finished := block.CompressAndChecksum(&compressedBuf, outputBlock, opts.Compression, &checksummer)
 					output[i].physical = finished.CloneWithByteAlloc(&blockAlloc)
@@ -219,6 +225,12 @@ func rewriteRangeKeyBlockToWriter(r *Reader, w RawWriter, from, to []byte) error
 		if !s.Valid() {
 			break
 		}
+		if err := r.Comparer.ValidateKey.Validate(s.Start); err != nil {
+			return err
+		}
+		if err := r.Comparer.ValidateKey.Validate(s.End); err != nil {
+			return err
+		}
 		for i := range s.Keys {
 			if s.Keys[i].Kind() != base.InternalKeyKindRangeKeySet {
 				return errBadKind
@@ -328,6 +340,9 @@ func RewriteKeySuffixesViaWriter(
 		if err != nil {
 			return nil, err
 		}
+		if invariants.Enabled && invariants.Sometimes(10) {
+			r.Comparer.ValidateKey.MustValidate(scratch.UserKey)
+		}
 		if err := w.AddWithForceObsolete(scratch, val, false); err != nil {
 			return nil, err
 		}


Test patch:
diff --git a/cockroachkvs/cockroachkvs_test.go b/cockroachkvs/cockroachkvs_test.go
index 7227a8ea..1c4cb860 100644
--- a/cockroachkvs/cockroachkvs_test.go
+++ b/cockroachkvs/cockroachkvs_test.go
@@ -79,6 +79,46 @@ func TestComparer(t *testing.T) {
 	}
 }
 
+func TestComparerFuncs(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	var buf bytes.Buffer
+	datadriven.RunTest(t, "testdata/comparer_funcs",
+		func(t *testing.T, td *datadriven.TestData) string {
+			buf.Reset()
+			switch td.Cmd {
+			case "separator":
+				var keys [][]byte
+				var dst []byte
+				for _, line := range crstrings.Lines(td.Input) {
+					keys = keys[:0]
+					for _, formattedKey := range strings.Fields(line) {
+						k := ParseFormattedKey(formattedKey)
+						keys = append(keys, k)
+					}
+					if len(keys) != 2 {
+						t.Fatalf("expected two keys, but found %d", len(keys))
+					}
+					dst = Comparer.Separator(dst[:0], keys[0], keys[1])
+					fmt.Fprintf(&buf, "Separator(%q [%x], %q [%x]) = %q [%x]\n",
+						FormatKey(keys[0]), keys[0], FormatKey(keys[1]), keys[1], FormatKey(dst), dst)
+				}
+				return buf.String()
+			case "successor":
+				var dst []byte
+				for _, line := range crstrings.Lines(td.Input) {
+					k := ParseFormattedKey(line)
+					dst = Comparer.Successor(dst[:0], k)
+					fmt.Fprintf(&buf, "Successor(%q [%x]) = %q [%x]\n",
+						FormatKey(k), k, FormatKey(dst), dst)
+				}
+				return buf.String()
+			default:
+				panic(fmt.Sprintf("unknown command: %s", td.Cmd))
+			}
+		})
+}
+
 func TestKeySchema_KeyWriter(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
@@ -633,7 +673,7 @@ func parseUserKey(userKeyStr string) []byte {
 		}
 	}
 	k := append(append([]byte(roachKey), 0), parseVersion(versionStr)...)
-	checkEngineKey(k)
+	validateEngineKey.MustValidate(k)
 	return k
 }
 
@@ -657,7 +697,7 @@ func parseVersion(versionStr string) []byte {
 		if len(ret) != 14 && len(ret) != 10 {
 			panic(fmt.Sprintf("expected 10 or 14-length ret got %d", len(ret)))
 		}
-		checkEngineKey(ret)
+		validateEngineKey.MustValidate(ret)
 		// TODO(jackson): Refactor to allow us to generate a suffix without a
 		// sentinel byte rather than stripping it like this.
 		return ret[1:]
diff --git a/cockroachkvs/key_schema_test.go b/cockroachkvs/key_schema_test.go
index c3309bce..dcaae8f3 100644
--- a/cockroachkvs/key_schema_test.go
+++ b/cockroachkvs/key_schema_test.go
@@ -161,7 +161,7 @@ func TestKeySchema_RandomKeys(t *testing.T) {
 		if n := len(kv.K.UserKey); n > len(keys[k]) {
 			t.Fatalf("key %q is longer than original key %q", kv.K.UserKey, keys[k])
 		}
-		checkEngineKey(kv.K.UserKey)
+		validateEngineKey.MustValidate(kv.K.UserKey)
 
 		// We write keys[k] as the value too, so check that it's verbatim equal.
 		value, callerOwned, err := kv.V.Value(valBuf)
diff --git a/cockroachkvs/testdata/comparer_funcs b/cockroachkvs/testdata/comparer_funcs
new file mode 100644
index 00000000..f181a323
--- /dev/null
+++ b/cockroachkvs/testdata/comparer_funcs
@@ -0,0 +1,36 @@
+successor
+foo
+bar
+foo@1000000,0
+foo@1000000.000051212,0
+foo@1000000,8
+foo@1000000.000051212,7
+foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a
+----
+Successor(foo [666f6f00]) = g [6700]
+Successor(bar [62617200]) = c [6300]
+Successor(foo@1000000,0 [666f6f0000038d7ea4c6800009]) = g [6700]
+Successor(foo@1000000.000051212,0 [666f6f0000038d7ea4c7480c09]) = g [6700]
+Successor(foo@1000000,8 [666f6f0000038d7ea4c68000000000080d]) = g [6700]
+Successor(foo@1000000.000051212,7 [666f6f0000038d7ea4c7480c000000070d]) = g [6700]
+Successor(foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a [666f6f0001cf379e85637142e6acc8ed5259fa177a12]) = g [6700]
+
+# The successor for an empty string should be a key consisting only of the 0x00
+# byte.
+successor
+hex:
+----
+Successor(<empty> []) =  [00]
+
+separator
+bar foo
+foo@1000000,0 foo@1000000,0
+foo@1000000,1 foo@1000000,0
+foo@01,a65a118c-5dcb-452b-bd82-7dd17e121138 foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a
+fax@01,a65a118c-5dcb-452b-bd82-7dd17e121138 foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a
+----
+Separator(bar [62617200], foo [666f6f00]) = c [6300]
+Separator(foo@1000000,0 [666f6f0000038d7ea4c6800009], foo@1000000,0 [666f6f0000038d7ea4c6800009]) = foo@1000000,0 [666f6f0000038d7ea4c6800009]
+Separator(foo@1000000,1 [666f6f0000038d7ea4c68000000000010d], foo@1000000,0 [666f6f0000038d7ea4c6800009]) = foo@1000000,1 [666f6f0000038d7ea4c68000000000010d]
+Separator(foo@01,a65a118c-5dcb-452b-bd82-7dd17e121138 [666f6f0001a65a118c5dcb452bbd827dd17e12113812], foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a [666f6f0001cf379e85637142e6acc8ed5259fa177a12]) = foo@01,a65a118c-5dcb-452b-bd82-7dd17e121138 [666f6f0001a65a118c5dcb452bbd827dd17e12113812]
+Separator(fax@01,a65a118c-5dcb-452b-bd82-7dd17e121138 [6661780001a65a118c5dcb452bbd827dd17e12113812], foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a [666f6f0001cf379e85637142e6acc8ed5259fa177a12]) = fb [666200]
diff --git a/cockroachkvs/testdata/key_schema_key_seeker b/cockroachkvs/testdata/key_schema_key_seeker
index 9509b416..5eff3c64 100644
--- a/cockroachkvs/testdata/key_schema_key_seeker
+++ b/cockroachkvs/testdata/key_schema_key_seeker
@@ -169,3 +169,75 @@ MaterializeUserKey(-1, 0) = hex:6d6f6f0000000000b2d05e0109
 MaterializeUserKey(0, 1) = hex:6d6f6f0000000000b2d05e00000000020d
 MaterializeUserKey(1, 2) = hex:6d6f6f0000000000b2d05e00000000010d
 MaterializeUserKey(2, 3) = hex:6d6f6f0000000000b2d05e0009
+
+define-block
+moo
+moo @ 3000000002,1
+moo @ 3000000001,1
+moo @ 0000000001,0
+----
+Parse(moo) = hex:6d6f6f00
+Parse(moo @ 3000000002,1) = hex:6d6f6f0000000000b2d05e02000000010d
+Parse(moo @ 3000000001,1) = hex:6d6f6f0000000000b2d05e01000000010d
+Parse(moo @ 0000000001,0) = hex:6d6f6f00000000000000000109
+
+# IsLowerBound should return false for all keys because the first key of the
+# block sorts before all the provided keys. Previously, a bug did not account
+# for the empty suffix sorting first.
+
+is-lower-bound
+moo @ 9000000000,2
+moo @ 8000000000,2
+moo @ 8000000000,1
+moo @ 8000000000,0
+moo @ 7000000000,9
+moo @ 3000000000,2
+moo @ 3000000000,1
+moo @ 3000000000,0
+moo @ 0000000000,1
+----
+IsLowerBound(moo @ 9000000000,2, "") = false
+IsLowerBound(moo @ 8000000000,2, "") = false
+IsLowerBound(moo @ 8000000000,1, "") = false
+IsLowerBound(moo @ 8000000000,0, "") = false
+IsLowerBound(moo @ 7000000000,9, "") = false
+IsLowerBound(moo @ 3000000000,2, "") = false
+IsLowerBound(moo @ 3000000000,1, "") = false
+IsLowerBound(moo @ 3000000000,0, "") = false
+IsLowerBound(moo @ 0000000000,1, "") = false
+
+define-block
+moo @ 0000000000,1
+moo @ 3000000002,1
+moo @ 3000000001,1
+moo @ 0000000001,0
+----
+Parse(moo @ 0000000000,1) = hex:6d6f6f000000000000000000000000010d
+Parse(moo @ 3000000002,1) = hex:6d6f6f0000000000b2d05e02000000010d
+Parse(moo @ 3000000001,1) = hex:6d6f6f0000000000b2d05e01000000010d
+Parse(moo @ 0000000001,0) = hex:6d6f6f00000000000000000109
+
+is-lower-bound
+moo
+moo @ 9000000000,2
+moo @ 8000000000,2
+moo @ 8000000000,1
+moo @ 8000000000,0
+moo @ 7000000000,9
+moo @ 3000000000,2
+moo @ 3000000000,1
+moo @ 3000000000,0
+moo @ 0000000000,1
+moo @ 02073a83c45688420eaf97824255790f1e12
+----
+IsLowerBound(moo, "") = true
+IsLowerBound(moo @ 9000000000,2, "") = true
+IsLowerBound(moo @ 8000000000,2, "") = true
+IsLowerBound(moo @ 8000000000,1, "") = true
+IsLowerBound(moo @ 8000000000,0, "") = true
+IsLowerBound(moo @ 7000000000,9, "") = true
+IsLowerBound(moo @ 3000000000,2, "") = true
+IsLowerBound(moo @ 3000000000,1, "") = true
+IsLowerBound(moo @ 3000000000,0, "") = true
+IsLowerBound(moo @ 0000000000,1, "") = true
+IsLowerBound(moo @ 02073a83c45688420eaf97824255790f1e12, "") = true
diff --git a/internal/base/comparer_test.go b/internal/base/comparer_test.go
index e16d85ea..31db932b 100644
--- a/internal/base/comparer_test.go
+++ b/internal/base/comparer_test.go
@@ -12,15 +12,11 @@ import (
 	"time"
 )
 
-func TestDefAppendSeparator(t *testing.T) {
+func TestDefaultComparer_Separator(t *testing.T) {
 	testCases := []struct {
 		a, b, want string
 	}{
-		// Examples from the doc comments.
 		{"black", "blue", "blb"},
-		{"green", "", "green"},
-		// Non-empty b values. The C++ Level-DB code calls these separators.
-		{"", "2", ""},
 		{"1", "2", "1"},
 		{"1", "29", "2"},
 		{"13", "19", "14"},
@@ -33,22 +29,37 @@ func TestDefAppendSeparator(t *testing.T) {
 		{"1\xff\xff", "19", "1\xff\xff"},
 		{"1\xff\xff", "2", "1\xff\xff"},
 		{"1\xff\xff", "9", "2"},
-		// Empty b values. The C++ Level-DB code calls these successors.
-		{"", "", ""},
-		{"1", "", "1"},
-		{"11", "", "11"},
-		{"11\xff", "", "11\xff"},
-		{"1\xff", "", "1\xff"},
-		{"1\xff\xff", "", "1\xff\xff"},
-		{"\xff", "", "\xff"},
-		{"\xff\xff", "", "\xff\xff"},
-		{"\xff\xff\xff", "", "\xff\xff\xff"},
 	}
 	for _, tc := range testCases {
 		t.Run("", func(t *testing.T) {
 			got := string(DefaultComparer.Separator(nil, []byte(tc.a), []byte(tc.b)))
 			if got != tc.want {
-				t.Errorf("a, b = %q, %q: got %q, want %q", tc.a, tc.b, got, tc.want)
+				t.Errorf("DefaultComparer.Separator(nil, %q, %q) = %q, want %q", tc.a, tc.b, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestDefaultComparer_Successor(t *testing.T) {
+	testCases := []struct {
+		a, want string
+	}{
+		{"green", "h"},
+		{"", ""},
+		{"1", "2"},
+		{"11", "2"},
+		{"11\xff", "2"},
+		{"1\xff", "2"},
+		{"1\xff\xff", "2"},
+		{"\xff", "\xff"},
+		{"\xff\xff", "\xff\xff"},
+		{"\xff\xff\xff", "\xff\xff\xff"},
+	}
+	for _, tc := range testCases {
+		t.Run("", func(t *testing.T) {
+			got := string(DefaultComparer.Successor(nil, []byte(tc.a)))
+			if got != tc.want {
+				t.Errorf("DefaultComparer.Successor(nil, %q) = %q, want %q", tc.a, got, tc.want)
 			}
 		})
 	}
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index d459b1a6..5d3415aa 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -51,19 +51,28 @@ var runOnceFlags, runFlags = metaflags.InitAllFlags()
 // be customized via the --inner-binary flag (used for code coverage
 // instrumentation).
 func TestMeta(t *testing.T) {
-	runTestMeta(t, false /* multiInstance */)
+	runTestMeta(t)
 }
 
 func TestMetaTwoInstance(t *testing.T) {
-	runTestMeta(t, true /* multiInstance */)
+	runTestMeta(t, metamorphic.MultiInstance(2))
 }
 
-func runTestMeta(t *testing.T, multiInstance bool) {
+func TestMetaCockroachKVs(t *testing.T) {
+	runTestMeta(t, metamorphic.CockroachKeyFormat)
+}
+
+type option interface {
+	metamorphic.RunOnceOption
+	metamorphic.RunOption
+}
+
+func runTestMeta(t *testing.T, addtlOptions ...option) {
 	switch {
 	case runOnceFlags.Compare != "":
 		onceOpts := runOnceFlags.MakeRunOnceOptions()
-		if multiInstance {
-			onceOpts = append(onceOpts, metamorphic.MultiInstance(2))
+		for _, opt := range addtlOptions {
+			onceOpts = append(onceOpts, opt)
 		}
 		testRootDir, runSubdirs := runOnceFlags.ParseCompare()
 		if runOnceFlags.TryToReduce {
@@ -78,8 +87,8 @@ func runTestMeta(t *testing.T, multiInstance bool) {
 		// runOptions() below) or the user specified it manually in order to re-run
 		// a test.
 		onceOpts := runOnceFlags.MakeRunOnceOptions()
-		if multiInstance {
-			onceOpts = append(onceOpts, metamorphic.MultiInstance(2))
+		for _, opt := range addtlOptions {
+			onceOpts = append(onceOpts, opt)
 		}
 		if runOnceFlags.TryToReduce {
 			tryToReduce(t, runOnceFlags.Dir, runOnceFlags.RunDir, runOnceFlags.ReduceAttempts)
@@ -90,8 +99,8 @@ func runTestMeta(t *testing.T, multiInstance bool) {
 
 	default:
 		opts := runFlags.MakeRunOptions()
-		if multiInstance {
-			opts = append(opts, metamorphic.MultiInstance(2))
+		for _, opt := range addtlOptions {
+			opts = append(opts, opt)
 		}
 		metamorphic.RunAndCompare(t, runFlags.Dir, opts...)
 	}
diff --git a/internal/testkeys/testkeys.go b/internal/testkeys/testkeys.go
index 35eb6144..4ddab98b 100644
--- a/internal/testkeys/testkeys.go
+++ b/internal/testkeys/testkeys.go
@@ -22,6 +22,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 )
 
@@ -106,7 +107,20 @@ var Comparer = &base.Comparer{
 		return append(append(dst, a...), 0x00)
 	},
 	Split: split,
-	Name:  "pebble.internal.testkeys",
+	ValidateKey: func(k []byte) error {
+		// Ensure that if the key has a suffix, it's a valid integer
+		// (potentially modulo a faux synthetic bit suffix).
+		k = bytes.TrimSuffix(k, ignoreTimestampSuffix)
+		i := split(k)
+		if i == len(k) {
+			return nil
+		}
+		if _, err := parseUintBytes(k[i+1:], 10, 64); err != nil {
+			return errors.Wrapf(err, "invalid key %q", k)
+		}
+		return nil
+	},
+	Name: "pebble.internal.testkeys",
 }
 
 // The comparator is similar to the one in Cockroach; when the prefixes are
diff --git a/metamorphic/generator_test.go b/metamorphic/generator_test.go
index 12efc8e1..88114f56 100644
--- a/metamorphic/generator_test.go
+++ b/metamorphic/generator_test.go
@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/pebble/cockroachkvs"
 	"github.com/cockroachdb/pebble/internal/randvar"
 	"github.com/pmezard/go-difflib/difflib"
 	"github.com/stretchr/testify/require"
@@ -194,3 +195,88 @@ func TestGenerateDisjointKeyRanges(t *testing.T) {
 		}
 	}
 }
+
+func TestCockroachSuffixKeyspace(t *testing.T) {
+	testCasesByMaxLogical := [][]string{
+		0: {
+			"",
+			"0.000000001,0",
+			"0.000000002,0",
+			"0.000000003,0",
+			"0.000000004,0",
+			"0.000000005,0",
+			"0.000000006,0",
+			"0.000000007,0",
+		},
+		1: {
+			"",
+			"0,1",
+			"0.000000001,0",
+			"0.000000001,1",
+			"0.000000002,0",
+			"0.000000002,1",
+			"0.000000003,0",
+			"0.000000003,1",
+			"0.000000004,0",
+			"0.000000004,1",
+		},
+		2: {
+			"",
+			"0,1",
+			"0,2",
+			"0.000000001,0",
+			"0.000000001,1",
+			"0.000000001,2",
+			"0.000000002,0",
+			"0.000000002,1",
+			"0.000000002,2",
+			"0.000000003,0",
+			"0.000000003,1",
+			"0.000000003,2",
+			"0.000000004,0",
+			"0.000000004,1",
+			"0.000000004,2",
+		},
+		3: {
+			"",
+			"0,1",
+			"0,2",
+			"0,3",
+			"0.000000001,0",
+			"0.000000001,1",
+			"0.000000001,2",
+			"0.000000001,3",
+			"0.000000002,0",
+			"0.000000002,1",
+			"0.000000002,2",
+			"0.000000002,3",
+			"0.000000003,0",
+			"0.000000003,1",
+			"0.000000003,2",
+			"0.000000003,3",
+			"0.000000004,0",
+			"0.000000004,1",
+			"0.000000004,2",
+			"0.000000004,3",
+		},
+	}
+	for maxLogical, testCases := range testCasesByMaxLogical {
+		t.Run(fmt.Sprintf("maxLogical=%d", maxLogical), func(t *testing.T) {
+			ks := cockroachSuffixKeyspace{maxLogical: int64(maxLogical)}
+
+			for si, expect := range testCases {
+				suffix := ks.ToMaterializedSuffix(suffixIndex(si))
+				var prettyPrinted string
+				if len(suffix) > 0 {
+					prettyPrinted = fmt.Sprint(cockroachkvs.FormatKeySuffix(suffix))
+				}
+				if expect != prettyPrinted {
+					t.Errorf("ToMaterializedSuffix(%d) = %q, want %q", si, prettyPrinted, expect)
+				}
+				if got := ks.ToSuffixIndex(suffix); got != suffixIndex(si) {
+					t.Errorf("ToSuffixIndex(%q) = %d, want %d", prettyPrinted, got, si)
+				}
+			}
+		})
+	}
+}
diff --git a/metamorphic/testkeys.go b/metamorphic/testkeys.go
index 0e60d94c..41506fda 100644
--- a/metamorphic/testkeys.go
+++ b/metamorphic/testkeys.go
@@ -10,6 +10,7 @@ import (
 	"math/rand/v2"
 	"slices"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/sstable"
@@ -39,15 +40,26 @@ var TestkeysKeyFormat = KeyFormat{
 		return sstable.NewTestKeysMaskingFilter()
 	},
 	NewSuffixBlockPropertyFilter: func(filterMin, filterMax []byte) sstable.BlockPropertyFilter {
-		low, err := testkeys.ParseSuffix(filterMin)
-		if err != nil {
-			panic(err)
+		var low, high int64
+		var err error
+		if filterMin != nil {
+			low, err = testkeys.ParseSuffix(filterMin)
+			if err != nil {
+				panic(err)
+			}
 		}
-		high, err := testkeys.ParseSuffix(filterMax)
-		if err != nil {
-			panic(err)
+		if filterMax != nil {
+			high, err = testkeys.ParseSuffix(filterMax)
+			if err != nil {
+				panic(err)
+			}
+		}
+		// The suffixes were encoded in descending order, so low should be the
+		// max timestamp and high should be the min timestamp.
+		if low <= high {
+			panic(errors.AssertionFailedf("low <= high: %d <= %d", low, high))
 		}
-		return sstable.NewTestKeysBlockPropertyFilter(uint64(low), uint64(high))
+		return sstable.NewTestKeysBlockPropertyFilter(uint64(high), uint64(low))
 	},
 }
 
@@ -144,15 +156,28 @@ func (kg *testkeyKeyGenerator) IncMaxSuffix() []byte {
 	return testkeys.Suffix(int64(kg.cfg.writeSuffixDist.Max()))
 }
 
+// SuffixRange generates a new uniformly random range of suffixes (low, high]
+// such that high is guaranteed to be strictly greater (as defined by
+// ComparePointSuffixes) than low.
+//
+// The high suffix may be nil, in which case the suffix range represents all
+// suffixes ≥ low.
 func (kg *testkeyKeyGenerator) SuffixRange() (low, high []byte) {
 	a := kg.uniformSuffixInt()
 	b := kg.uniformSuffixInt()
-	if a > b {
+	// NB: Suffixes are sorted in descending order, so we need to generate the
+	// timestamps such that a > b. This ensures that the returned suffixes sort
+	// such that low < high.
+	if a < b {
 		a, b = b, a
 	} else if a == b {
-		b++
+		a++
+	}
+	low = testkeys.Suffix(a) // NB: a > 0
+	if b > 0 {
+		high = testkeys.Suffix(b)
 	}
-	return testkeys.Suffix(a), testkeys.Suffix(b)
+	return low, high
 }
 
 // UniformSuffix returns a suffix in the same range as SkewedSuffix but with a