Repository: cockroachdb/pebble
Pull request #3815 (2024-07-31 18:45:33)

sstable/block: refactor CompressAndChecksum
This commit moves logic around finishing a block (compressing and checksumming) into the block package. The logic is refactored to use a PhysicalBlock type that distinguishes a finished block from an unfinished, trailer-less block.

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


Patch:
diff --git a/comparer.go b/comparer.go
index c92cd793..5f31d240 100644
--- a/comparer.go
+++ b/comparer.go
@@ -29,3 +29,10 @@ type Comparer = base.Comparer
 
 // DefaultComparer exports the base.DefaultComparer variable.
 var DefaultComparer = base.DefaultComparer
+
+// CheckComparer is a mini test suite that verifies a comparer implementation.
+//
+// It takes strictly ordered (according to the comparator) lists of prefixes and
+// suffixes. Both lists must contain the empty slice. It is recommended that
+// both lists have at least three elements.
+var CheckComparer = base.CheckComparer
diff --git a/db.go b/db.go
index 82637466..2760f2cc 100644
--- a/db.go
+++ b/db.go
@@ -1152,7 +1152,7 @@ func finishInitializingIter(ctx context.Context, buf *iterAlloc) *Iterator {
 	}
 
 	if dbi.opts.rangeKeys() {
-		dbi.rangeKeyMasking.init(dbi, dbi.comparer.Compare, dbi.comparer.Split)
+		dbi.rangeKeyMasking.init(dbi, &dbi.comparer)
 
 		// When iterating over both point and range keys, don't create the
 		// range-key iterator stack immediately if we can avoid it. This
diff --git a/external_iterator.go b/external_iterator.go
index 08eecca0..72c7a147 100644
--- a/external_iterator.go
+++ b/external_iterator.go
@@ -194,7 +194,7 @@ func finishInitializingExternal(ctx context.Context, it *Iterator) error {
 	it.iter = it.pointIter
 
 	if it.opts.rangeKeys() {
-		it.rangeKeyMasking.init(it, it.comparer.Compare, it.comparer.Split)
+		it.rangeKeyMasking.init(it, &it.comparer)
 		var rangeKeyIters []keyspan.FragmentIterator
 		if it.rangeKey == nil {
 			// We could take advantage of the lack of overlaps in range keys within
diff --git a/internal/base/comparer.go b/internal/base/comparer.go
index 4673f648..934ba820 100644
--- a/internal/base/comparer.go
+++ b/internal/base/comparer.go
@@ -8,29 +8,50 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"slices"
 	"strconv"
 	"unicode/utf8"
+
+	"github.com/cockroachdb/errors"
 )
 
+// CompareSuffixes compares two key suffixes and returns -1, 0, or +1.
+//
+// The empty slice suffix must be 'less than' any non-empty suffix.
+//
+// A full key k is composed of a prefix k[:Split(k)] and suffix k[Split(k):].
+// Suffixes are compared to break ties between equal prefixes.
+type CompareSuffixes func(a, b []byte) int
+
 // Compare returns -1, 0, or +1 depending on whether a is 'less than', 'equal
-// to' or 'greater than' b. The empty slice must be 'less than' any non-empty
-// slice.
+// to' or 'greater than' b.
 //
-// Compare is used to compare user keys, such as those passed as arguments to
-// the various DB methods, as well as those returned from Separator, Successor,
-// and Split. It is also used to compare key suffixes, i.e. the remainder of the
-// key after Split.
+// Both a and b must be valid keys.
 //
-// The comparison of the prefix parts must be a simple byte-wise compare. In
-// other words, if a and be don't have suffixes, then
-// Compare(a, b) = bytes.Compare(a, b).
+// A key a is less than b if a's prefix is byte-wise less than b's prefix, or if
+// the prefixes are equal and a's suffix is less than b's suffix (according to
+// CompareSuffixes).
 //
-// In general, if prefix(a) = a[:Split(a)] and suffix(a) = a[Split(a):], then
+// In other words, if prefix(a) = a[:Split(a)] and suffix(a) = a[Split(a):]:
 //
 //	  Compare(a, b) = bytes.Compare(prefix(a), prefix(b)) if not 0, or
-//		                bytes.Compare(suffix(a), suffix(b)) otherwise.
+//		                CompareSuffixes(suffix(a), suffix(b)) otherwise.
+//
+// Compare defaults to using the formula above but it can be customized if there
+// is a (potentially faster) specialization.
 type Compare func(a, b []byte) int
 
+// defaultCompare implements Compare in terms of Split and CompareSuffixes, as
+// mentioned above.
+func defaultCompare(split Split, compareSuffixes CompareSuffixes, a, b []byte) int {
+	an := split(a)
+	bn := split(b)
+	if prefixCmp := bytes.Compare(a[:an], b[:bn]); prefixCmp != 0 {
+		return prefixCmp
+	}
+	return compareSuffixes(a[an:], b[bn:])
+}
+
 // Equal returns true if a and b are equivalent.
 //
 // For a given Compare, Equal(a,b)=true iff Compare(a,b)=0; that is, Equal is a
@@ -145,8 +166,7 @@ var DefaultSplit Split = func(key []byte) int { return len(key) }
 // Comparer defines a total ordering over the space of []byte keys: a 'less
 // than' relationship.
 type Comparer struct {
-	// These fields must always be specified.
-	Compare        Compare
+	// The following must always be specified.
 	AbbreviatedKey AbbreviatedKey
 	Separator      Separator
 	Successor      Successor
@@ -154,13 +174,20 @@ type Comparer struct {
 	// ImmediateSuccessor must be specified if range keys are used.
 	ImmediateSuccessor ImmediateSuccessor
 
+	// Split defaults to a trivial implementation that returns the full key length
+	// if it is not specified.
+	Split Split
+
+	// CompareSuffixes defaults to bytes.Compare if it is not specified.
+	CompareSuffixes CompareSuffixes
+
+	// Compare defaults to a generic implementation that uses Split,
+	// bytes.Compare, and CompareSuffixes if it is not specified.
+	Compare Compare
 	// Equal defaults to using Compare() == 0 if it is not specified.
 	Equal Equal
 	// FormatKey defaults to the DefaultFormatter if it is not specified.
 	FormatKey FormatKey
-	// Split defaults to a trivial implementation that returns the full key length
-	// if it is not specified.
-	Split Split
 
 	// FormatValue is optional.
 	FormatValue FormatValue
@@ -182,23 +209,37 @@ func (c *Comparer) EnsureDefaults() *Comparer {
 	if c == nil {
 		return DefaultComparer
 	}
-	if c.Compare == nil || c.AbbreviatedKey == nil || c.Separator == nil || c.Successor == nil || c.Name == "" {
+	if c.AbbreviatedKey == nil || c.Separator == nil || c.Successor == nil || c.Name == "" {
 		panic("invalid Comparer: mandatory field not set")
 	}
-	if c.Equal != nil && c.Split != nil && c.FormatKey != nil {
+	if c.CompareSuffixes != nil && c.Compare != nil && c.Equal != nil && c.Split != nil && c.FormatKey != nil {
 		return c
 	}
 	n := &Comparer{}
 	*n = *c
-	if n.Equal == nil {
-		cmp := n.Compare
-		n.Equal = func(a, b []byte) bool {
-			return cmp(a, b) == 0
-		}
-	}
+
 	if n.Split == nil {
 		n.Split = DefaultSplit
 	}
+	if n.CompareSuffixes == nil && n.Compare == nil && n.Equal == nil {
+		n.CompareSuffixes = bytes.Compare
+		n.Compare = bytes.Compare
+		n.Equal = bytes.Equal
+	} else {
+		if n.CompareSuffixes == nil {
+			n.CompareSuffixes = bytes.Compare
+		}
+		if n.Compare == nil {
+			n.Compare = func(a, b []byte) int {
+				return defaultCompare(n.Split, n.CompareSuffixes, a, b)
+			}
+		}
+		if n.Equal == nil {
+			n.Equal = func(a, b []byte) bool {
+				return n.Compare(a, b) == 0
+			}
+		}
+	}
 	if n.FormatKey == nil {
 		n.FormatKey = DefaultFormatter
 	}
@@ -208,8 +249,9 @@ func (c *Comparer) EnsureDefaults() *Comparer {
 // DefaultComparer is the default implementation of the Comparer interface.
 // It uses the natural ordering, consistent with bytes.Compare.
 var DefaultComparer = &Comparer{
-	Compare: bytes.Compare,
-	Equal:   bytes.Equal,
+	CompareSuffixes: bytes.Compare,
+	Compare:         bytes.Compare,
+	Equal:           bytes.Equal,
 
 	AbbreviatedKey: func(key []byte) uint64 {
 		if len(key) >= 8 {
@@ -338,31 +380,25 @@ func MakeAssertComparer(c Comparer) Comparer {
 	return Comparer{
 		Compare: func(a []byte, b []byte) int {
 			res := c.Compare(a, b)
-			an := c.Split(a)
-			aPrefix, aSuffix := a[:an], a[an:]
-			bn := c.Split(b)
-			bPrefix, bSuffix := b[:bn], b[bn:]
-			if prefixCmp := bytes.Compare(aPrefix, bPrefix); prefixCmp == 0 {
-				if suffixCmp := c.Compare(aSuffix, bSuffix); suffixCmp != res {
-					panic(AssertionFailedf("%s: Compare with equal prefixes not consistent with Compare of suffixes: Compare(%q, %q)=%d, Compare(%q, %q)=%d",
-						c.Name, a, b, res, aSuffix, bSuffix, suffixCmp,
-					))
-				}
-			} else if prefixCmp != res {
-				panic(AssertionFailedf("%s: Compare did not perform byte-wise comparison of prefixes", c.Name))
+			// Verify that Compare is consistent with the default implementation.
+			if expected := defaultCompare(c.Split, c.CompareSuffixes, a, b); res != expected {
+				panic(AssertionFailedf("%s: Compare(%s, %s)=%d, expected %d",
+					c.Name, c.FormatKey(a), c.FormatKey(b), res, expected))
 			}
 			return res
 		},
 
 		Equal: func(a []byte, b []byte) bool {
 			eq := c.Equal(a, b)
-			if cmp := c.Compare(a, b); eq != (cmp == 0) {
+			// Verify that Equal is consistent with Compare.
+			if expected := c.Compare(a, b); eq != (expected == 0) {
 				panic("Compare and Equal are not consistent")
 			}
 			return eq
 		},
 
 		// TODO(radu): add more checks.
+		CompareSuffixes:    c.CompareSuffixes,
 		AbbreviatedKey:     c.AbbreviatedKey,
 		Separator:          c.Separator,
 		Successor:          c.Successor,
@@ -373,3 +409,57 @@ func MakeAssertComparer(c Comparer) Comparer {
 		Name:               c.Name,
 	}
 }
+
+// CheckComparer is a mini test suite that verifies a comparer implementation.
+//
+// It takes lists of valid prefixes and suffixes. It is recommended that both
+// lists have at least three elements.
+func CheckComparer(c *Comparer, prefixes [][]byte, suffixes [][]byte) error {
+	// Empty slice is always a valid suffix.
+	suffixes = append(suffixes, nil)
+
+	// Verify the suffixes have a consistent ordering.
+	slices.SortFunc(suffixes, c.CompareSuffixes)
+	if !slices.IsSortedFunc(suffixes, c.CompareSuffixes) {
+		return errors.Errorf("CompareSuffixes is inconsistent")
+	}
+
+	// Check the split function.
+	for _, p := range prefixes {
+		for _, s := range suffixes {
+			key := slices.Concat(p, s)
+			if n := c.Split(key); n != len(p) {
+				return errors.Errorf("incorrect Split result %d on '%x' (prefix '%x' suffix '%x')", n, key, p, s)
+			}
+		}
+	}
+
+	// Check the Compare/Equals functions on all possible combinations.
+	for _, ap := range prefixes {
+		for _, as := range suffixes {
+			a := slices.Concat(ap, as)
+			for _, bp := range prefixes {
+				for _, bs := range suffixes {
+					b := slices.Concat(bp, bs)
+					result := c.Compare(a, b)
+
+					expected := bytes.Compare(ap, bp)
+					if expected == 0 {
+						expected = c.CompareSuffixes(as, bs)
+					}
+
+					if (result == 0) != c.Equal(a, b) {
+						return errors.Errorf("Equal(%s, %s) doesn't agree with Compare", c.FormatKey(a), c.FormatKey(b))
+					}
+
+					if result != expected {
+						return errors.Errorf("Compare(%s, %s)=%d, expected %d", c.FormatKey(a), c.FormatKey(b), result, expected)
+					}
+				}
+			}
+		}
+	}
+
+	// TODO(radu): check more methods.
+	return nil
+}
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index f17e88e7..b5c710ee 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -148,7 +148,8 @@ import (
 // exported function, and before a subsequent call to Next advances the iterator
 // and mutates the contents of the returned key and value.
 type Iter struct {
-	cmp base.Compare
+	cmp       base.Compare
+	suffixCmp base.CompareSuffixes
 
 	cfg IterConfig
 
@@ -308,8 +309,9 @@ func NewIter(
 ) *Iter {
 	cfg.ensureDefaults()
 	i := &Iter{
-		cmp: cfg.Comparer.Compare,
-		cfg: cfg,
+		cmp:       cfg.Comparer.Compare,
+		suffixCmp: cfg.Comparer.CompareSuffixes,
+		cfg:       cfg,
 		// We don't want a nil keyBuf because if the first key we encounter is
 		// empty, it would become nil.
 		keyBuf: make([]byte, 8),
@@ -329,7 +331,7 @@ func NewIter(
 	i.frontiers.Init(i.cmp)
 	i.delElider.Init(i.cmp, cfg.TombstoneElision)
 	i.rangeDelCompactor = MakeRangeDelSpanCompactor(i.cmp, i.cfg.Comparer.Equal, cfg.Snapshots, cfg.TombstoneElision)
-	i.rangeKeyCompactor = MakeRangeKeySpanCompactor(i.cmp, i.cfg.Comparer.Equal, cfg.Snapshots, cfg.RangeKeyElision)
+	i.rangeKeyCompactor = MakeRangeKeySpanCompactor(i.cmp, i.suffixCmp, cfg.Snapshots, cfg.RangeKeyElision)
 	i.lastRangeDelSpanFrontier.Init(&i.frontiers, nil, i.lastRangeDelSpanFrontierReached)
 	return i
 }
diff --git a/internal/compact/spans.go b/internal/compact/spans.go
index 8ca8c13a..d8d5bd47 100644
--- a/internal/compact/spans.go
+++ b/internal/compact/spans.go
@@ -83,18 +83,18 @@ func (c *RangeDelSpanCompactor) Compact(span, output *keyspan.Span) {
 // for at most one "compacted" span.
 type RangeKeySpanCompactor struct {
 	cmp       base.Compare
-	equal     base.Equal
+	suffixCmp base.CompareSuffixes
 	snapshots Snapshots
 	elider    rangeTombstoneElider
 }
 
 // MakeRangeKeySpanCompactor creates a new compactor for range key spans.
 func MakeRangeKeySpanCompactor(
-	cmp base.Compare, equal base.Equal, snapshots Snapshots, elision TombstoneElision,
+	cmp base.Compare, suffixCmp base.CompareSuffixes, snapshots Snapshots, elision TombstoneElision,
 ) RangeKeySpanCompactor {
 	c := RangeKeySpanCompactor{
 		cmp:       cmp,
-		equal:     equal,
+		suffixCmp: suffixCmp,
 		snapshots: snapshots,
 	}
 	c.elider.Init(cmp, elision)
@@ -131,7 +131,7 @@ func (c *RangeKeySpanCompactor) Compact(span, output *keyspan.Span) {
 		}
 		if y > start {
 			keysDst := output.Keys[usedLen:cap(output.Keys)]
-			rangekey.Coalesce(c.cmp, c.equal, span.Keys[start:y], &keysDst)
+			rangekey.Coalesce(c.suffixCmp, span.Keys[start:y], &keysDst)
 			if y == len(span.Keys) {
 				// This is the last snapshot stripe. Unsets and deletes can be elided.
 				keysDst = c.elideInLastStripe(span.Start, span.End, keysDst)
@@ -143,7 +143,7 @@ func (c *RangeKeySpanCompactor) Compact(span, output *keyspan.Span) {
 	}
 	if y < len(span.Keys) {
 		keysDst := output.Keys[usedLen:cap(output.Keys)]
-		rangekey.Coalesce(c.cmp, c.equal, span.Keys[y:], &keysDst)
+		rangekey.Coalesce(c.suffixCmp, span.Keys[y:], &keysDst)
 		keysDst = c.elideInLastStripe(span.Start, span.End, keysDst)
 		usedLen += len(keysDst)
 		output.Keys = append(output.Keys, keysDst...)
diff --git a/internal/keyspan/defragment.go b/internal/keyspan/defragment.go
index 35dda16d..df8b7dac 100644
--- a/internal/keyspan/defragment.go
+++ b/internal/keyspan/defragment.go
@@ -29,27 +29,29 @@ const keysReuseMaxCapacity = 100
 type DefragmentMethod interface {
 	// ShouldDefragment takes two abutting spans and returns whether the two
 	// spans should be combined into a single, defragmented Span.
-	ShouldDefragment(equal base.Equal, left, right *Span) bool
+	ShouldDefragment(suffixCmp base.CompareSuffixes, left, right *Span) bool
 }
 
 // The DefragmentMethodFunc type is an adapter to allow the use of ordinary
 // functions as DefragmentMethods. If f is a function with the appropriate
 // signature, DefragmentMethodFunc(f) is a DefragmentMethod that calls f.
-type DefragmentMethodFunc func(equal base.Equal, left, right *Span) bool
+type DefragmentMethodFunc func(suffixCmp base.CompareSuffixes, left, right *Span) bool
 
 // ShouldDefragment calls f(equal, left, right).
-func (f DefragmentMethodFunc) ShouldDefragment(equal base.Equal, left, right *Span) bool {
-	return f(equal, left, right)
+func (f DefragmentMethodFunc) ShouldDefragment(
+	suffixCmp base.CompareSuffixes, left, right *Span,
+) bool {
+	return f(suffixCmp, left, right)
 }
 
-// DefragmentInternal configures a DefragmentingIter to defragment spans
-// only if they have identical keys. It requires spans' keys to be sorted in
-// trailer descending order.
+// DefragmentInternal configures a DefragmentingIter to defragment spans only if
+// they have identical keys. It requires spans' keys to be sorted in trailer
+// descending order.
 //
 // This defragmenting method is intended for use in compactions that may see
 // internal range keys fragments that may now be joined, because the state that
 // required their fragmentation has been dropped.
-var DefragmentInternal DefragmentMethod = DefragmentMethodFunc(func(equal base.Equal, a, b *Span) bool {
+var DefragmentInternal DefragmentMethod = DefragmentMethodFunc(func(suffixCmp base.CompareSuffixes, a, b *Span) bool {
 	if a.KeysOrder != ByTrailerDesc || b.KeysOrder != ByTrailerDesc {
 		panic("pebble: span keys unexpectedly not in trailer descending order")
 	}
@@ -60,7 +62,7 @@ var DefragmentInternal DefragmentMethod = DefragmentMethodFunc(func(equal base.E
 		if a.Keys[i].Trailer != b.Keys[i].Trailer {
 			return false
 		}
-		if !equal(a.Keys[i].Suffix, b.Keys[i].Suffix) {
+		if suffixCmp(a.Keys[i].Suffix, b.Keys[i].Suffix) != 0 {
 			return false
 		}
 		if !bytes.Equal(a.Keys[i].Value, b.Keys[i].Value) {
@@ -452,7 +454,7 @@ func (i *DefragmentingIter) Prev() (*Span, error) {
 // DefragmentMethod and ensures both spans are NOT empty; not defragmenting empty
 // spans is an optimization that lets us load fewer sstable blocks.
 func (i *DefragmentingIter) checkEqual(left, right *Span) bool {
-	return (!left.Empty() && !right.Empty()) && i.method.ShouldDefragment(i.equal, i.iterSpan, &i.curr)
+	return (!left.Empty() && !right.Empty()) && i.method.ShouldDefragment(i.comparer.CompareSuffixes, i.iterSpan, &i.curr)
 }
 
 // defragmentForward defragments spans in the forward direction, starting from
diff --git a/internal/keyspan/fragmenter.go b/internal/keyspan/fragmenter.go
index 9b3bf614..4c2c1814 100644
--- a/internal/keyspan/fragmenter.go
+++ b/internal/keyspan/fragmenter.go
@@ -6,60 +6,11 @@ package keyspan
 
 import (
 	"fmt"
-	"sort"
 
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
 )
 
-type spansByStartKey struct {
-	cmp base.Compare
-	buf []Span
-}
-
-func (v *spansByStartKey) Len() int { return len(v.buf) }
-func (v *spansByStartKey) Less(i, j int) bool {
-	return v.cmp(v.buf[i].Start, v.buf[j].Start) < 0
-}
-func (v *spansByStartKey) Swap(i, j int) {
-	v.buf[i], v.buf[j] = v.buf[j], v.buf[i]
-}
-
-type spansByEndKey struct {
-	cmp base.Compare
-	buf []Span
-}
-
-func (v *spansByEndKey) Len() int { return len(v.buf) }
-func (v *spansByEndKey) Less(i, j int) bool {
-	return v.cmp(v.buf[i].End, v.buf[j].End) < 0
-}
-func (v *spansByEndKey) Swap(i, j int) {
-	v.buf[i], v.buf[j] = v.buf[j], v.buf[i]
-}
-
-// keysBySeqNumKind sorts spans by the start key's sequence number in
-// descending order. If two spans have equal sequence number, they're compared
-// by key kind in descending order. This ordering matches the ordering of
-// base.InternalCompare among keys with matching user keys.
-type keysBySeqNumKind []Key
-
-func (v *keysBySeqNumKind) Len() int           { return len(*v) }
-func (v *keysBySeqNumKind) Less(i, j int) bool { return (*v)[i].Trailer > (*v)[j].Trailer }
-func (v *keysBySeqNumKind) Swap(i, j int)      { (*v)[i], (*v)[j] = (*v)[j], (*v)[i] }
-
-// Sort the spans by start key. This is the ordering required by the
-// Fragmenter. Usually spans are naturally sorted by their start key,
-// but that isn't true for range deletion tombstones in the legacy
-// range-del-v1 block format.
-func Sort(cmp base.Compare, spans []Span) {
-	sorter := spansByStartKey{
-		cmp: cmp,
-		buf: spans,
-	}
-	sort.Sort(&sorter)
-}
-
 // Fragmenter fragments a set of spans such that overlapping spans are
 // split at their overlap points. The fragmented spans are output to the
 // supplied Output function.
@@ -80,10 +31,8 @@ type Fragmenter struct {
 	// specific key (e.g. TruncateAndFlushTo). It is cached in the Fragmenter to
 	// allow reuse.
 	doneBuf []Span
-	// sortBuf is used to sort fragments by end key when flushing.
-	sortBuf spansByEndKey
 	// flushBuf is used to sort keys by (seqnum,kind) before emitting.
-	flushBuf keysBySeqNumKind
+	flushBuf []Key
 	// flushedKey is the key that fragments have been flushed up to. Any
 	// additional spans added to the fragmenter must have a start key >=
 	// flushedKey. A nil value indicates flushedKey has not been set.
@@ -305,9 +254,7 @@ func (f *Fragmenter) flush(buf []Span, lastKey []byte) {
 
 	// Sort the spans by end key. This will allow us to walk over the spans and
 	// easily determine the next split point (the smallest end-key).
-	f.sortBuf.cmp = f.Cmp
-	f.sortBuf.buf = buf
-	sort.Sort(&f.sortBuf)
+	SortSpansByEndKey(f.Cmp, buf)
 
 	// Loop over the spans, splitting by end key.
 	for len(buf) > 0 {
@@ -324,7 +271,7 @@ func (f *Fragmenter) flush(buf []Span, lastKey []byte) {
 			f.flushBuf = append(f.flushBuf, buf[i].Keys...)
 		}
 
-		sort.Sort(&f.flushBuf)
+		SortKeysByTrailer(f.flushBuf)
 
 		f.Emit(Span{
 			Start: buf[0].Start,
diff --git a/internal/keyspan/keyspanimpl/merging_iter.go b/internal/keyspan/keyspanimpl/merging_iter.go
index 9fd87960..edcd0600 100644
--- a/internal/keyspan/keyspanimpl/merging_iter.go
+++ b/internal/keyspan/keyspanimpl/merging_iter.go
@@ -1047,7 +1047,7 @@ func (m *MergingIter) synthesizeKeys(dir int8) (bool, *keyspan.Span, error) {
 		Keys:      m.keys,
 		KeysOrder: keyspan.ByTrailerDesc,
 	}
-	if err := m.transformer.Transform(m.comparer.Compare, m.span, &m.span); err != nil {
+	if err := m.transformer.Transform(m.comparer.CompareSuffixes, m.span, &m.span); err != nil {
 		return false, nil, err
 	}
 	return found, &m.span, nil
diff --git a/internal/keyspan/span.go b/internal/keyspan/span.go
index 335b10ed..8f41ef96 100644
--- a/internal/keyspan/span.go
+++ b/internal/keyspan/span.go
@@ -6,9 +6,9 @@ package keyspan // import "github.com/cockroachdb/pebble/internal/keyspan"
 
 import (
 	"bytes"
+	"cmp"
 	"fmt"
 	"slices"
-	"sort"
 	"strings"
 	"unicode"
 
@@ -93,9 +93,9 @@ func (k Key) Kind() base.InternalKeyKind {
 // Equal returns true if this Key is equal to the given key. Two keys are said
 // to be equal if the two Keys have equal trailers, suffix and value. Suffix
 // comparison uses the provided base.Compare func. Value comparison is bytewise.
-func (k Key) Equal(equal base.Equal, b Key) bool {
+func (k Key) Equal(suffixCmp base.CompareSuffixes, b Key) bool {
 	return k.Trailer == b.Trailer &&
-		equal(k.Suffix, b.Suffix) &&
+		suffixCmp(k.Suffix, b.Suffix) == 0 &&
 		bytes.Equal(k.Value, b.Value)
 }
 
@@ -455,24 +455,38 @@ func (s prettySpan) Format(fs fmt.State, c rune) {
 	fmt.Fprintf(fs, "}")
 }
 
-// SortKeysByTrailer sorts a keys slice by trailer.
-func SortKeysByTrailer(keys *[]Key) {
-	// NB: keys is a pointer to a slice instead of a slice to avoid `sorted`
-	// escaping to the heap.
-	sorted := (*keysBySeqNumKind)(keys)
-	sort.Sort(sorted)
+// SortKeysByTrailer sorts a Keys slice by trailer.
+func SortKeysByTrailer(keys []Key) {
+	slices.SortFunc(keys, func(a, b Key) int {
+		// Trailer are ordered in decreasing number order.
+		return -cmp.Compare(a.Trailer, b.Trailer)
+	})
+}
+
+// SortKeysBySuffix sorts a Keys slice by suffix.
+func SortKeysBySuffix(suffixCmp base.CompareSuffixes, keys []Key) {
+	slices.SortFunc(keys, func(a, b Key) int {
+		return suffixCmp(a.Suffix, b.Suffix)
+	})
 }
 
-// KeysBySuffix implements sort.Interface, sorting its member Keys slice to by
-// Suffix in the order dictated by Cmp.
-type KeysBySuffix struct {
-	Cmp  base.Compare
-	Keys []Key
+// SortSpansByStartKey sorts the spans by start key.
+//
+// This is the ordering required by the Fragmenter. Usually spans are naturally
+// sorted by their start key, but that isn't true for range deletion tombstones
+// in the legacy range-del-v1 block format.
+func SortSpansByStartKey(cmp base.Compare, spans []Span) {
+	slices.SortFunc(spans, func(a, b Span) int {
+		return cmp(a.Start, b.Start)
+	})
 }
 
-func (s *KeysBySuffix) Len() int           { return len(s.Keys) }
-func (s *KeysBySuffix) Less(i, j int) bool { return s.Cmp(s.Keys[i].Suffix, s.Keys[j].Suffix) < 0 }
-func (s *KeysBySuffix) Swap(i, j int)      { s.Keys[i], s.Keys[j] = s.Keys[j], s.Keys[i] }
+// SortSpansByEndKey sorts the spans by the end key.
+func SortSpansByEndKey(cmp base.Compare, spans []Span) {
+	slices.SortFunc(spans, func(a, b Span) int {
+		return cmp(a.End, b.End)
+	})
+}
 
 // ParseSpan parses the string representation of a Span. It's intended for
 // tests. ParseSpan panics if passed a malformed span representation.
diff --git a/internal/keyspan/transformer.go b/internal/keyspan/transformer.go
index df7c096c..037a2e30 100644
--- a/internal/keyspan/transformer.go
+++ b/internal/keyspan/transformer.go
@@ -11,21 +11,21 @@ type Transformer interface {
 	// Transform takes a Span as input and writes the transformed Span to the
 	// provided output *Span pointer. The output Span's Keys slice may be reused
 	// by Transform to reduce allocations.
-	Transform(cmp base.Compare, in Span, out *Span) error
+	Transform(suffixCmp base.CompareSuffixes, in Span, out *Span) error
 }
 
 // The TransformerFunc type is an adapter to allow the use of ordinary functions
 // as Transformers. If f is a function with the appropriate signature,
 // TransformerFunc(f) is a Transformer that calls f.
-type TransformerFunc func(base.Compare, Span, *Span) error
+type TransformerFunc func(base.CompareSuffixes, Span, *Span) error
 
 // Transform calls f(cmp, in, out).
-func (tf TransformerFunc) Transform(cmp base.Compare, in Span, out *Span) error {
-	return tf(cmp, in, out)
+func (tf TransformerFunc) Transform(suffixCmp base.CompareSuffixes, in Span, out *Span) error {
+	return tf(suffixCmp, in, out)
 }
 
 // NoopTransform is a Transformer that performs no mutations.
-var NoopTransform Transformer = TransformerFunc(func(_ base.Compare, s Span, dst *Span) error {
+var NoopTransform Transformer = TransformerFunc(func(_ base.CompareSuffixes, s Span, dst *Span) error {
 	dst.Start, dst.End = s.Start, s.End
 	dst.Keys = append(dst.Keys[:0], s.Keys...)
 	return nil
@@ -34,7 +34,7 @@ var NoopTransform Transformer = TransformerFunc(func(_ base.Compare, s Span, dst
 // VisibleTransform filters keys that are invisible at the provided snapshot
 // sequence number.
 func VisibleTransform(snapshot base.SeqNum) Transformer {
-	return TransformerFunc(func(_ base.Compare, s Span, dst *Span) error {
+	return TransformerFunc(func(_ base.CompareSuffixes, s Span, dst *Span) error {
 		dst.Start, dst.End = s.Start, s.End
 		dst.Keys = dst.Keys[:0]
 		for _, k := range s.Keys {
@@ -58,8 +58,8 @@ type TransformerIter struct {
 
 	// Transformer is applied on every Span returned by this iterator.
 	Transformer Transformer
-	// Comparer in use for this keyspace.
-	Compare base.Compare
+	// Suffix comparer in use for this keyspace.
+	SuffixCmp base.CompareSuffixes
 
 	span Span
 }
@@ -73,7 +73,7 @@ func (t *TransformerIter) applyTransform(span *Span) (*Span, error) {
 		End:   t.span.End[:0],
 		Keys:  t.span.Keys[:0],
 	}
-	if err := t.Transformer.Transform(t.Compare, *span, &t.span); err != nil {
+	if err := t.Transformer.Transform(t.SuffixCmp, *span, &t.span); err != nil {
 		return nil, err
 	}
 	return &t.span, nil
diff --git a/internal/rangekey/coalesce.go b/internal/rangekey/coalesce.go
index c95809fd..c89e7542 100644
--- a/internal/rangekey/coalesce.go
+++ b/internal/rangekey/coalesce.go
@@ -6,7 +6,7 @@ package rangekey
 
 import (
 	"math"
-	"sort"
+	"slices"
 
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -49,25 +49,21 @@ import (
 // set or unset but not both.
 //
 // The resulting dst Keys slice is sorted by InternalKeyTrailer.
-func Coalesce(cmp base.Compare, eq base.Equal, keys []keyspan.Key, dst *[]keyspan.Key) {
+func Coalesce(suffixCmp base.CompareSuffixes, keys []keyspan.Key, dst *[]keyspan.Key) {
 	// TODO(jackson): Currently, Coalesce doesn't actually perform the sequence
 	// number promotion described in the comment above.
-	keysBySuffix := keyspan.KeysBySuffix{
-		Cmp:  cmp,
-		Keys: (*dst)[:0],
-	}
-	CoalesceIntoKeysBySuffix(eq, &keysBySuffix, math.MaxUint64, keys)
+	*dst = CoalesceInto(suffixCmp, (*dst)[:0], math.MaxUint64, keys)
 	// Update the span with the (potentially reduced) keys slice. coalesce left
 	// the keys in *dst sorted by suffix. Re-sort them by trailer.
-	*dst = keysBySuffix.Keys
-	keyspan.SortKeysByTrailer(dst)
+	keyspan.SortKeysByTrailer(*dst)
 }
 
-// CoalesceIntoKeysBySuffix is a variant of Coalesce which outputs the results into
-// keyspan.KeysBySuffix without sorting them.
-func CoalesceIntoKeysBySuffix(
-	equal base.Equal, keysBySuffix *keyspan.KeysBySuffix, snapshot base.SeqNum, keys []keyspan.Key,
-) {
+// CoalesceInto is a variant of Coalesce which outputs the results into dst
+// without sorting them.
+func CoalesceInto(
+	suffixCmp base.CompareSuffixes, dst []keyspan.Key, snapshot base.SeqNum, keys []keyspan.Key,
+) []keyspan.Key {
+	dst = dst[:0]
 	// First, enforce visibility and RangeKeyDelete mechanics. We only need to
 	// consider the prefix of keys before and including the first
 	// RangeKeyDelete. We also must skip any keys that aren't visible at the
@@ -94,7 +90,7 @@ func CoalesceIntoKeysBySuffix(
 			deleteIdx = i
 			break
 		}
-		keysBySuffix.Keys = append(keysBySuffix.Keys, keys[i])
+		dst = append(dst, keys[i])
 	}
 
 	// Sort the accumulated keys by suffix. There may be duplicates within a
@@ -103,12 +99,14 @@ func CoalesceIntoKeysBySuffix(
 	// We use a stable sort so that the first key with a given suffix is the one
 	// that with the highest InternalKeyTrailer (because the input `keys` was sorted by
 	// trailer descending).
-	sort.Stable(keysBySuffix)
+	slices.SortStableFunc(dst, func(a, b keyspan.Key) int {
+		return suffixCmp(a.Suffix, b.Suffix)
+	})
 
 	// Grab a handle of the full sorted slice, before reslicing
-	// keysBySuffix.keys to accumulate the final coalesced keys.
-	sorted := keysBySuffix.Keys
-	keysBySuffix.Keys = keysBySuffix.Keys[:0]
+	// dst to accumulate the final coalesced keys.
+	sorted := dst
+	dst = dst[:0]
 
 	var (
 		// prevSuffix is updated on each iteration of the below loop, and
@@ -123,31 +121,32 @@ func CoalesceIntoKeysBySuffix(
 		shadowing bool
 	)
 	for i := range sorted {
-		if i > 0 && equal(prevSuffix, sorted[i].Suffix) {
+		if i > 0 && suffixCmp(prevSuffix, sorted[i].Suffix) == 0 {
 			// Skip; this key is shadowed by the predecessor that had a larger
 			// InternalKeyTrailer. If this is the first shadowed key, set shadowing=true
 			// and reslice keysBySuffix.keys to hold the entire unshadowed
 			// prefix.
 			if !shadowing {
-				keysBySuffix.Keys = keysBySuffix.Keys[:i]
+				dst = dst[:i]
 				shadowing = true
 			}
 			continue
 		}
 		prevSuffix = sorted[i].Suffix
 		if shadowing {
-			keysBySuffix.Keys = append(keysBySuffix.Keys, sorted[i])
+			dst = append(dst, sorted[i])
 		}
 	}
-	// If there was no shadowing, keysBySuffix.keys is untouched. We can simply
-	// set it to the existing `sorted` slice (also backed by keysBySuffix.keys).
+	// If there was no shadowing, dst.keys is untouched. We can simply set it to
+	// the existing `sorted` slice (also backed by dst).
 	if !shadowing {
-		keysBySuffix.Keys = sorted
+		dst = sorted
 	}
 	// If the original input `keys` slice contained a RangeKeyDelete, add it.
 	if deleteIdx >= 0 {
-		keysBySuffix.Keys = append(keysBySuffix.Keys, keys[deleteIdx])
+		dst = append(dst, keys[deleteIdx])
 	}
+	return dst
 }
 
 // ForeignSSTTransformer implements a keyspan.Transformer for range keys in
@@ -161,31 +160,28 @@ func CoalesceIntoKeysBySuffix(
 type ForeignSSTTransformer struct {
 	Equal   base.Equal
 	SeqNum  base.SeqNum
-	sortBuf keyspan.KeysBySuffix
+	sortBuf []keyspan.Key
 }
 
 // Transform implements the Transformer interface.
 func (f *ForeignSSTTransformer) Transform(
-	cmp base.Compare, s keyspan.Span, dst *keyspan.Span,
+	suffixCmp base.CompareSuffixes, s keyspan.Span, dst *keyspan.Span,
 ) error {
 	// Apply shadowing of keys.
 	dst.Start = s.Start
 	dst.End = s.End
-	f.sortBuf = keyspan.KeysBySuffix{
-		Cmp:  cmp,
-		Keys: f.sortBuf.Keys[:0],
-	}
-	CoalesceIntoKeysBySuffix(f.Equal, &f.sortBuf, math.MaxUint64, s.Keys)
-	keys := f.sortBuf.Keys
+	f.sortBuf = f.sortBuf[:0]
+	f.sortBuf = CoalesceInto(suffixCmp, f.sortBuf, math.MaxUint64, s.Keys)
+	keys := f.sortBuf
 	dst.Keys = dst.Keys[:0]
 	for i := range keys {
 		switch keys[i].Kind() {
 		case base.InternalKeyKindRangeKeySet:
-			if invariants.Enabled && len(dst.Keys) > 0 && cmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
+			if invariants.Enabled && len(dst.Keys) > 0 && suffixCmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
 				panic("pebble: keys unexpectedly not in ascending suffix order")
 			}
 		case base.InternalKeyKindRangeKeyUnset:
-			if invariants.Enabled && len(dst.Keys) > 0 && cmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
+			if invariants.Enabled && len(dst.Keys) > 0 && suffixCmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
 				panic("pebble: keys unexpectedly not in ascending suffix order")
 			}
 		case base.InternalKeyKindRangeKeyDelete:
@@ -201,7 +197,7 @@ func (f *ForeignSSTTransformer) Transform(
 	}
 	// coalesce results in dst.Keys being sorted by Suffix. Change it back to
 	// ByTrailerDesc, as that's what the iterator stack will expect.
-	keyspan.SortKeysByTrailer(&dst.Keys)
+	keyspan.SortKeysByTrailer(dst.Keys)
 	dst.KeysOrder = keyspan.ByTrailerDesc
 	return nil
 }
diff --git a/internal/rangekeystack/user_iterator.go b/internal/rangekeystack/user_iterator.go
index 510a8476..232dba6d 100644
--- a/internal/rangekeystack/user_iterator.go
+++ b/internal/rangekeystack/user_iterator.go
@@ -68,7 +68,7 @@ type UserIteratorConfig struct {
 type Buffers struct {
 	merging       keyspanimpl.MergingBuffers
 	defragmenting keyspan.DefragmentingBuffers
-	sortBuf       keyspan.KeysBySuffix
+	sortBuf       []keyspan.Key
 }
 
 // PrepareForReuse discards any excessively large buffers.
@@ -143,36 +143,34 @@ func (ui *UserIteratorConfig) SetBounds(lower, upper []byte) {
 // only contain RangeKeySets describing the state visible at the provided
 // sequence number, and hold their Keys sorted by Suffix (except if internalKeys
 // is true, then keys remain sorted by trailer.
-func (ui *UserIteratorConfig) Transform(cmp base.Compare, s keyspan.Span, dst *keyspan.Span) error {
+func (ui *UserIteratorConfig) Transform(
+	suffixCmp base.CompareSuffixes, s keyspan.Span, dst *keyspan.Span,
+) error {
 	// Apply shadowing of keys.
 	dst.Start = s.Start
 	dst.End = s.End
-	ui.bufs.sortBuf = keyspan.KeysBySuffix{
-		Cmp:  cmp,
-		Keys: ui.bufs.sortBuf.Keys[:0],
-	}
-	rangekey.CoalesceIntoKeysBySuffix(ui.comparer.Equal, &ui.bufs.sortBuf, ui.snapshot, s.Keys)
+	ui.bufs.sortBuf = rangekey.CoalesceInto(suffixCmp, ui.bufs.sortBuf[:0], ui.snapshot, s.Keys)
 	if ui.internalKeys {
 		if s.KeysOrder != keyspan.ByTrailerDesc {
 			panic("unexpected key ordering in UserIteratorTransform with internalKeys = true")
 		}
-		dst.Keys = ui.bufs.sortBuf.Keys
-		keyspan.SortKeysByTrailer(&dst.Keys)
+		dst.Keys = ui.bufs.sortBuf
+		keyspan.SortKeysByTrailer(dst.Keys)
 		return nil
 	}
 	// During user iteration over range keys, unsets and deletes don't matter. This
 	// step helps logical defragmentation during iteration.
-	keys := ui.bufs.sortBuf.Keys
+	keys := ui.bufs.sortBuf
 	dst.Keys = dst.Keys[:0]
 	for i := range keys {
 		switch keys[i].Kind() {
 		case base.InternalKeyKindRangeKeySet:
-			if invariants.Enabled && len(dst.Keys) > 0 && cmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
+			if invariants.Enabled && len(dst.Keys) > 0 && suffixCmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
 				panic("pebble: keys unexpectedly not in ascending suffix order")
 			}
 			dst.Keys = append(dst.Keys, keys[i])
 		case base.InternalKeyKindRangeKeyUnset:
-			if invariants.Enabled && len(dst.Keys) > 0 && cmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
+			if invariants.Enabled && len(dst.Keys) > 0 && suffixCmp(dst.Keys[len(dst.Keys)-1].Suffix, keys[i].Suffix) > 0 {
 				panic("pebble: keys unexpectedly not in ascending suffix order")
 			}
 			// Skip.
@@ -197,7 +195,9 @@ func (ui *UserIteratorConfig) Transform(cmp base.Compare, s keyspan.Span, dst *k
 // defragmenter checks for equality between set suffixes and values (ignoring
 // sequence numbers). It's intended for use during user iteration, when the
 // wrapped keyspan iterator is merging spans across all levels of the LSM.
-func (ui *UserIteratorConfig) ShouldDefragment(equal base.Equal, a, b *keyspan.Span) bool {
+func (ui *UserIteratorConfig) ShouldDefragment(
+	suffixCmp base.CompareSuffixes, a, b *keyspan.Span,
+) bool {
 	// This method is not called with internalKeys = true.
 	if ui.internalKeys {
 		panic("unexpected call to ShouldDefragment with internalKeys = true")
@@ -221,12 +221,12 @@ func (ui *UserIteratorConfig) ShouldDefragment(equal base.Equal, a, b *keyspan.S
 				b.Keys[i].Kind() != base.InternalKeyKindRangeKeySet {
 				panic("pebble: unexpected non-RangeKeySet during defragmentation")
 			}
-			if i > 0 && (ui.comparer.Compare(a.Keys[i].Suffix, a.Keys[i-1].Suffix) < 0 ||
-				ui.comparer.Compare(b.Keys[i].Suffix, b.Keys[i-1].Suffix) < 0) {
+			if i > 0 && (suffixCmp(a.Keys[i].Suffix, a.Keys[i-1].Suffix) < 0 ||
+				suffixCmp(b.Keys[i].Suffix, b.Keys[i-1].Suffix) < 0) {
 				panic("pebble: range keys not ordered by suffix during defragmentation")
 			}
 		}
-		if !equal(a.Keys[i].Suffix, b.Keys[i].Suffix) {
+		if suffixCmp(a.Keys[i].Suffix, b.Keys[i].Suffix) != 0 {
 			ret = false
 			break
 		}
diff --git a/iterator.go b/iterator.go
index e37de717..ee85d4d0 100644
--- a/iterator.go
+++ b/iterator.go
@@ -2199,7 +2199,7 @@ func (i *Iterator) saveRangeKey() {
 		if invariants.Enabled {
 			if s.Keys[j].Kind() != base.InternalKeyKindRangeKeySet {
 				panic("pebble: user iteration encountered non-RangeKeySet key kind")
-			} else if j > 0 && i.cmp(s.Keys[j].Suffix, s.Keys[j-1].Suffix) < 0 {
+			} else if j > 0 && i.comparer.CompareSuffixes(s.Keys[j].Suffix, s.Keys[j-1].Suffix) < 0 {
 				panic("pebble: user iteration encountered range keys not in suffix order")
 			}
 		}
@@ -2685,7 +2685,7 @@ func (i *Iterator) SetOptions(o *IterOptions) {
 	if boundsEqual && o.KeyTypes == i.opts.KeyTypes &&
 		(i.pointIter != nil || !i.opts.pointKeys()) &&
 		(i.rangeKey != nil || !i.opts.rangeKeys() || i.opts.KeyTypes == IterKeyTypePointsAndRanges) &&
-		i.equal(o.RangeKeyMasking.Suffix, i.opts.RangeKeyMasking.Suffix) &&
+		i.comparer.CompareSuffixes(o.RangeKeyMasking.Suffix, i.opts.RangeKeyMasking.Suffix) == 0 &&
 		o.UseL6Filters == i.opts.UseL6Filters {
 		// The options are identical, so we can likely use the fast path. In
 		// addition to all the above constraints, we cannot use the fast path if
diff --git a/metamorphic/build.go b/metamorphic/build.go
index d627dbd4..3973929c 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -135,13 +135,11 @@ func writeSSTForIngestion(
 				End:   outputKey(span.End),
 				Keys:  make([]keyspan.Key, 0, len(span.Keys)),
 			}
-			rangekey.Coalesce(
-				t.opts.Comparer.Compare, t.opts.Comparer.Equal, span.Keys, &collapsed.Keys,
-			)
+			rangekey.Coalesce(t.opts.Comparer.CompareSuffixes, span.Keys, &collapsed.Keys)
 			for i := range collapsed.Keys {
 				collapsed.Keys[i].Trailer = base.MakeTrailer(0, collapsed.Keys[i].Kind())
 			}
-			keyspan.SortKeysByTrailer(&collapsed.Keys)
+			keyspan.SortKeysByTrailer(collapsed.Keys)
 			if err := w.Raw().EncodeSpan(collapsed); err != nil {
 				return nil, err
 			}
diff --git a/metamorphic/ops.go b/metamorphic/ops.go
index f1f2cf26..06a1b17c 100644
--- a/metamorphic/ops.go
+++ b/metamorphic/ops.go
@@ -2113,7 +2113,7 @@ func (r *replicateOp) run(t *Test, h historyRecorder) {
 					Value:   rangeKeys[i].Value,
 				}
 			}
-			keyspan.SortKeysByTrailer(&span.Keys)
+			keyspan.SortKeysByTrailer(span.Keys)
 			if err := w.Raw().EncodeSpan(span); err != nil {
 				panic(err)
 			}
diff --git a/range_keys.go b/range_keys.go
index 52640a41..b50f5624 100644
--- a/range_keys.go
+++ b/range_keys.go
@@ -225,9 +225,10 @@ func containsAnyRangeKeys(iter manifest.LevelIterator) bool {
 // result is ignored, and the block is read.
 
 type rangeKeyMasking struct {
-	cmp    base.Compare
-	split  base.Split
-	filter BlockPropertyFilterMask
+	cmp       base.Compare
+	suffixCmp base.CompareSuffixes
+	split     base.Split
+	filter    BlockPropertyFilterMask
 	// maskActiveSuffix holds the suffix of a range key currently acting as a
 	// mask, hiding point keys with suffixes greater than it. maskActiveSuffix
 	// is only ever non-nil if IterOptions.RangeKeyMasking.Suffix is non-nil.
@@ -243,9 +244,10 @@ type rangeKeyMasking struct {
 	parent   *Iterator
 }
 
-func (m *rangeKeyMasking) init(parent *Iterator, cmp base.Compare, split base.Split) {
-	m.cmp = cmp
-	m.split = split
+func (m *rangeKeyMasking) init(parent *Iterator, c *base.Comparer) {
+	m.cmp = c.Compare
+	m.suffixCmp = c.CompareSuffixes
+	m.split = c.Split
 	if parent.opts.RangeKeyMasking.Filter != nil {
 		m.filter = parent.opts.RangeKeyMasking.Filter()
 	}
@@ -270,10 +272,10 @@ func (m *rangeKeyMasking) SpanChanged(s *keyspan.Span) {
 				if s.Keys[j].Suffix == nil {
 					continue
 				}
-				if m.cmp(s.Keys[j].Suffix, m.parent.opts.RangeKeyMasking.Suffix) < 0 {
+				if m.suffixCmp(s.Keys[j].Suffix, m.parent.opts.RangeKeyMasking.Suffix) < 0 {
 					continue
 				}
-				if len(m.maskActiveSuffix) == 0 || m.cmp(m.maskActiveSuffix, s.Keys[j].Suffix) > 0 {
+				if len(m.maskActiveSuffix) == 0 || m.suffixCmp(m.maskActiveSuffix, s.Keys[j].Suffix) > 0 {
 					m.maskSpan = s
 					m.maskActiveSuffix = append(m.maskActiveSuffix[:0], s.Keys[j].Suffix...)
 				}
@@ -352,7 +354,7 @@ func (m *rangeKeyMasking) SkipPoint(userKey []byte) bool {
 	// the InterleavingIter). Skip the point key if the range key's suffix is
 	// greater than the point key's suffix.
 	pointSuffix := userKey[m.split(userKey):]
-	if len(pointSuffix) > 0 && m.cmp(m.maskActiveSuffix, pointSuffix) < 0 {
+	if len(pointSuffix) > 0 && m.suffixCmp(m.maskActiveSuffix, pointSuffix) < 0 {
 		m.parent.stats.RangeKeyStats.SkippedPoints++
 		return true
 	}
diff --git a/scan_internal.go b/scan_internal.go
index 7124600d..a72750b2 100644
--- a/scan_internal.go
+++ b/scan_internal.go
@@ -788,7 +788,7 @@ func scanInternalImpl(
 					keysCopy[i].CopyFrom(span.Keys[i])
 					keysCopy[i].Trailer = base.MakeTrailer(0, span.Keys[i].Kind())
 				}
-				keyspan.SortKeysByTrailer(&keysCopy)
+				keyspan.SortKeysByTrailer(keysCopy)
 				if err := opts.visitRangeKey(span.Start, span.End, keysCopy); err != nil {
 					return err
 				}
diff --git a/sstable/block/compression.go b/sstable/block/compression.go
index 10a62c49..8f649345 100644
--- a/sstable/block/compression.go
+++ b/sstable/block/compression.go
@@ -9,7 +9,8 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
-	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/bytealloc"
+	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/golang/snappy"
 )
 
@@ -158,31 +159,90 @@ func DecompressInto(algo CompressionIndicator, compressed []byte, buf []byte) er
 	return nil
 }
 
-// Decompress decompresses an sstable block into memory manually allocated with
-// `cache.Alloc`.  NB: If Decompress returns (nil, nil), no decompression was
-// necessary and the caller may use `b` directly.
-func Decompress(algo CompressionIndicator, b []byte) (*cache.Value, error) {
-	if algo == NoCompressionIndicator {
-		return nil, nil
+// PhysicalBlock represents a block (possibly compressed) as it is stored
+// physically on disk, including its trailer.
+type PhysicalBlock struct {
+	// data contains the possibly compressed block data.
+	data    []byte
+	trailer Trailer
+}
+
+// LengthWithTrailer returns the length of the data block, including the trailer.
+func (b *PhysicalBlock) LengthWithTrailer() int {
+	return len(b.data) + TrailerLen
+}
+
+// LengthWithoutTrailer returns the length of the data block, excluding the trailer.
+func (b *PhysicalBlock) LengthWithoutTrailer() int {
+	return len(b.data)
+}
+
+// CloneWithByteAlloc returns a deep copy of the block, using the provided
+// bytealloc.A to allocate memory for the new copy.
+func (b *PhysicalBlock) CloneWithByteAlloc(a *bytealloc.A) PhysicalBlock {
+	var data []byte
+	*a, data = (*a).Alloc(len(b.data))
+	copy(data, b.data)
+	return PhysicalBlock{
+		data:    data,
+		trailer: b.trailer,
 	}
-	// first obtain the decoded length.
-	decodedLen, prefixLen, err := DecompressedLen(algo, b)
-	if err != nil {
-		return nil, err
+}
+
+// IsCompressed returns true if the block is compressed.
+func (b *PhysicalBlock) IsCompressed() bool {
+	return CompressionIndicator(b.trailer[0]) != NoCompressionIndicator
+}
+
+// WriteTo writes the block (including its trailer) to the provided Writable. If
+// err == nil, n is the number of bytes successfully written to the Writable.
+func (b *PhysicalBlock) WriteTo(w objstorage.Writable) (n int, err error) {
+	if err := w.Write(b.data); err != nil {
+		return 0, err
+	}
+	if err := w.Write(b.trailer[:]); err != nil {
+		return 0, err
 	}
-	b = b[prefixLen:]
-	// Allocate sufficient space from the cache.
-	decoded := cache.Alloc(decodedLen)
-	decodedBuf := decoded.Buf()
-	if err := DecompressInto(algo, b, decodedBuf); err != nil {
-		cache.Free(decoded)
-		return nil, err
+	return len(b.data) + len(b.trailer), nil
+}
+
+// CompressAndChecksum compresses and checksums the provided block, returning
+// the compressed block and its trailer. The dst argument is used for the
+// compressed payload if it's sufficiently large. If it's not, a new buffer is
+// allocated and *dst is updated to point to it.
+//
+// If the compressed block is not sufficiently smaller than the original block,
+// the compressed payload is discarded and the original, uncompressed block is
+// used to avoid unnecessary decompression overhead at read time.
+func CompressAndChecksum(
+	dst *[]byte, block []byte, compression Compression, checksummer *Checksummer,
+) PhysicalBlock {
+	// Compress the buffer, discarding the result if the improvement isn't at
+	// least 12.5%.
+	algo := NoCompressionIndicator
+	if compression != NoCompression {
+		var compressed []byte
+		algo, compressed = compress(compression, block, *dst)
+		if algo != NoCompressionIndicator && cap(compressed) > cap(*dst) {
+			*dst = compressed[:cap(compressed)]
+		}
+		if len(compressed) < len(block)-len(block)/8 {
+			block = compressed
+		} else {
+			algo = NoCompressionIndicator
+		}
 	}
-	return decoded, nil
+
+	// Calculate the checksum.
+	pb := PhysicalBlock{data: block}
+	pb.trailer[0] = byte(algo)
+	checksum := checksummer.Checksum(block, pb.trailer[:1])
+	pb.trailer = MakeTrailer(byte(algo), checksum)
+	return pb
 }
 
-// Compress compresses a sstable block, using dstBuf as the desired destination.
-func Compress(
+// compress compresses a sstable block, using dstBuf as the desired destination.
+func compress(
 	compression Compression, b []byte, dstBuf []byte,
 ) (indicator CompressionIndicator, compressed []byte) {
 	switch compression {
@@ -190,16 +250,13 @@ func Compress(
 		return SnappyCompressionIndicator, snappy.Encode(dstBuf, b)
 	case NoCompression:
 		return NoCompressionIndicator, b
-	}
-
-	if len(dstBuf) < binary.MaxVarintLen64 {
-		dstBuf = append(dstBuf, make([]byte, binary.MaxVarintLen64-len(dstBuf))...)
-	}
-	varIntLen := binary.PutUvarint(dstBuf, uint64(len(b)))
-	switch compression {
 	case ZstdCompression:
+		if len(dstBuf) < binary.MaxVarintLen64 {
+			dstBuf = append(dstBuf, make([]byte, binary.MaxVarintLen64-len(dstBuf))...)
+		}
+		varIntLen := binary.PutUvarint(dstBuf, uint64(len(b)))
 		return ZstdCompressionIndicator, encodeZstd(dstBuf, varIntLen, b)
 	default:
-		return NoCompressionIndicator, b
+		panic("unreachable")
 	}
 }
diff --git a/sstable/comparer.go b/sstable/comparer.go
index 66a20b51..62b52730 100644
--- a/sstable/comparer.go
+++ b/sstable/comparer.go
@@ -9,6 +9,9 @@ import "github.com/cockroachdb/pebble/internal/base"
 // Compare exports the base.Compare type.
 type Compare = base.Compare
 
+// CompareSuffixes exports the base.CompareSuffixes type.
+type CompareSuffixes = base.CompareSuffixes
+
 // Equal exports the base.Equal type.
 type Equal = base.Equal
 
diff --git a/sstable/layout.go b/sstable/layout.go
index c3bde01d..92c46ddb 100644
--- a/sstable/layout.go
+++ b/sstable/layout.go
@@ -350,20 +350,8 @@ func (w *layoutWriter) WriteDataBlock(b []byte, buf *blockBuf) (block.Handle, er
 
 // WritePrecompressedDataBlock writes a pre-compressed data block and its
 // pre-computed trailer to the writer, returning it's block handle.
-func (w *layoutWriter) WritePrecompressedDataBlock(
-	blk []byte, trailer block.Trailer,
-) (block.Handle, error) {
-	// Write the bytes to the file.
-	if err := w.writable.Write(blk); err != nil {
-		return block.Handle{}, err
-	}
-	bh := block.Handle{Offset: w.offset, Length: uint64(len(blk))}
-	w.offset += uint64(len(blk))
-	if err := w.writable.Write(trailer[:]); err != nil {
-		return block.Handle{}, err
-	}
-	w.offset += block.TrailerLen
-	return bh, nil
+func (w *layoutWriter) WritePrecompressedDataBlock(blk block.PhysicalBlock) (block.Handle, error) {
+	return w.writePrecompressedBlock(blk)
 }
 
 // WriteIndexBlock constructs a trailer for the provided index (first or
@@ -421,19 +409,8 @@ func (w *layoutWriter) writeNamedBlock(b []byte, name string) (bh block.Handle,
 
 // WriteValueBlock writes a pre-finished value block (with the trailer) to the
 // writer.
-func (w *layoutWriter) WriteValueBlock(blk []byte) (block.Handle, error) {
-	// NB: value blocks are already finished and contain the block trailer.
-	// TODO(jackson): can this be refactored to make value blocks less of a
-	// snowflake?
-	off := w.offset
-	w.clearFromCache(off)
-	// Write the bytes to the file.
-	if err := w.writable.Write(blk); err != nil {
-		return block.Handle{}, err
-	}
-	l := uint64(len(blk))
-	w.offset += l
-	return block.Handle{Offset: off, Length: l}, nil
+func (w *layoutWriter) WriteValueBlock(blk block.PhysicalBlock) (block.Handle, error) {
+	return w.writePrecompressedBlock(blk)
 }
 
 func (w *layoutWriter) WriteValueIndexBlock(
@@ -461,19 +438,21 @@ func (w *layoutWriter) WriteValueIndexBlock(
 func (w *layoutWriter) writeBlock(
 	b []byte, compression block.Compression, buf *blockBuf,
 ) (block.Handle, error) {
-	blk, trailer := compressAndChecksum(b, compression, buf)
-	bh := block.Handle{Offset: w.offset, Length: uint64(len(blk))}
-	w.clearFromCache(bh.Offset)
+	return w.writePrecompressedBlock(block.CompressAndChecksum(
+		&buf.compressedBuf, b, compression, &buf.checksummer))
+}
 
+// writePrecompressedBlock writes a pre-compressed block and its
+// pre-computed trailer to the writer, returning it's block handle.
+func (w *layoutWriter) writePrecompressedBlock(blk block.PhysicalBlock) (block.Handle, error) {
+	w.clearFromCache(w.offset)
 	// Write the bytes to the file.
-	if err := w.writable.Write(blk); err != nil {
-		return block.Handle{}, err
-	}
-	w.offset += uint64(len(blk))
-	if err := w.writable.Write(trailer[:]); err != nil {
+	n, err := blk.WriteTo(w.writable)
+	if err != nil {
 		return block.Handle{}, err
 	}
-	w.offset += block.TrailerLen
+	bh := block.Handle{Offset: w.offset, Length: uint64(blk.LengthWithoutTrailer())}
+	w.offset += uint64(n)
 	return bh, nil
 }
 
diff --git a/sstable/raw_writer.go b/sstable/raw_writer.go
index a40221bf..79212875 100644
--- a/sstable/raw_writer.go
+++ b/sstable/raw_writer.go
@@ -11,7 +11,6 @@ import (
 	"math"
 	"runtime"
 	"slices"
-	"sort"
 	"sync"
 
 	"github.com/cockroachdb/errors"
@@ -132,6 +131,7 @@ type RawWriter struct {
 	indexBlockOptions flushDecisionOptions
 	// The following fields are copied from Options.
 	compare              Compare
+	suffixCmp            CompareSuffixes
 	split                Split
 	formatKey            base.FormatKey
 	compression          block.Compression
@@ -183,9 +183,8 @@ type RawWriter struct {
 	// in indexPartitions. These live until the index finishes.
 	indexSepAlloc bytealloc.A
 
-	rangeKeyEncoder   rangekey.Encoder
-	rangeKeysBySuffix keyspan.KeysBySuffix
-	rangeKeySpan      keyspan.Span
+	rangeKeyEncoder rangekey.Encoder
+	rangeKeySpan    keyspan.Span
 	// dataBlockBuf consists of the state which is currently owned by and used by
 	// the Writer client goroutine. This state can be handed off to other goroutines.
 	dataBlockBuf *dataBlockBuf
@@ -575,13 +574,12 @@ type dataBlockBuf struct {
 	// next byte slice to be compressed. The uncompressed byte slice will be backed by the
 	// dataBlock.buf.
 	uncompressed []byte
-	// compressed is a reference to a byte slice which is owned by the dataBlockBuf. It is the
-	// compressed byte slice which must be written to disk. The compressed byte slice may be
-	// backed by the dataBlock.buf, or the dataBlockBuf.compressedBuf, depending on whether
-	// we use the result of the compression.
-	compressed []byte
-	// trailer is the block trailer encoding the compression type and checksum.
-	trailer block.Trailer
+
+	// physical holds the (possibly) compressed block and its trailer. The
+	// underlying block data's byte slice is owned by the dataBlockBuf. It  may
+	// be backed by the dataBlock.buf, or the dataBlockBuf.compressedBuf,
+	// depending on whether we use the result of the compression.
+	physical block.PhysicalBlock
 
 	// We're making calls to BlockPropertyCollectors from the Writer client goroutine. We need to
 	// pass the encoded block properties over to the write queue. To prevent copies, and allocations,
@@ -600,7 +598,7 @@ func (d *dataBlockBuf) clear() {
 	d.dataBlock.Reset()
 
 	d.uncompressed = nil
-	d.compressed = nil
+	d.physical = block.PhysicalBlock{}
 	d.dataBlockProps = nil
 	d.sepScratch = d.sepScratch[:0]
 }
@@ -623,7 +621,7 @@ func (d *dataBlockBuf) finish() {
 }
 
 func (d *dataBlockBuf) compressAndChecksum(c block.Compression) {
-	d.compressed, d.trailer = compressAndChecksum(d.uncompressed, c, &d.blockBuf)
+	d.physical = block.CompressAndChecksum(&d.compressedBuf, d.uncompressed, c, &d.checksummer)
 }
 
 func (d *dataBlockBuf) shouldFlush(
@@ -752,7 +750,7 @@ func (w *RawWriter) makeAddPointDecisionV3(
 		cmpUser = cmpPrefix
 		if cmpPrefix == 0 {
 			// Need to compare suffixes to compute cmpUser.
-			cmpUser = w.compare(prevPointUserKey[prevPointKeyInfo.prefixLen:],
+			cmpUser = w.suffixCmp(prevPointUserKey[prevPointKeyInfo.prefixLen:],
 				key.UserKey[w.lastPointKeyInfo.prefixLen:])
 		}
 	} else {
@@ -1026,16 +1024,13 @@ func (w *RawWriter) encodeFragmentedRangeKeySpan(span keyspan.Span) {
 	// NB: The span should only contain range keys and be internally consistent
 	// (eg, no duplicate suffixes, no additional keys after a RANGEKEYDEL).
 	//
-	// We use w.rangeKeysBySuffix and w.rangeKeySpan to avoid allocations.
+	// We use w.rangeKeySpan to avoid allocations.
 
 	// Sort the keys by suffix. Iteration doesn't *currently* depend on it, but
 	// we may want to in the future.
-	w.rangeKeysBySuffix.Cmp = w.compare
-	w.rangeKeysBySuffix.Keys = span.Keys
-	sort.Sort(&w.rangeKeysBySuffix)
-
 	w.rangeKeySpan = span
-	w.rangeKeySpan.Keys = w.rangeKeysBySuffix.Keys
+	keyspan.SortKeysBySuffix(w.suffixCmp, w.rangeKeySpan.Keys)
+
 	if w.err == nil {
 		w.err = w.EncodeSpan(w.rangeKeySpan)
 	}
@@ -1149,7 +1144,7 @@ func (w *RawWriter) flush(key InternalKey) error {
 	w.dataBlockBuf.compressAndChecksum(w.compression)
 	// Since dataBlockEstimates.addInflightDataBlock was never called, the
 	// inflightSize is set to 0.
-	w.coordination.sizeEstimate.dataBlockCompressed(len(w.dataBlockBuf.compressed), 0)
+	w.coordination.sizeEstimate.dataBlockCompressed(w.dataBlockBuf.physical.LengthWithoutTrailer(), 0)
 
 	// Determine if the index block should be flushed. Since we're accessing the
 	// dataBlockBuf.dataBlock.curKey here, we have to make sure that once we start
@@ -1583,27 +1578,6 @@ func (w *RawWriter) writeTwoLevelIndex() (block.Handle, error) {
 	return w.layout.WriteIndexBlock(w.topLevelIndexBlock.Finish())
 }
 
-func compressAndChecksum(
-	b []byte, compression block.Compression, blockBuf *blockBuf,
-) (compressed []byte, trailer block.Trailer) {
-	// Compress the buffer, discarding the result if the improvement isn't at
-	// least 12.5%.
-	algo, compressed := block.Compress(compression, b, blockBuf.compressedBuf)
-	if algo != block.NoCompressionIndicator && cap(compressed) > cap(blockBuf.compressedBuf) {
-		blockBuf.compressedBuf = compressed[:cap(compressed)]
-	}
-	if len(compressed) < len(b)-len(b)/8 {
-		b = compressed
-	} else {
-		algo = block.NoCompressionIndicator
-	}
-
-	// Calculate the checksum.
-	trailer[0] = byte(algo)
-	checksum := blockBuf.checksummer.Checksum(b, trailer[:1])
-	return b, block.MakeTrailer(byte(algo), checksum)
-}
-
 // assertFormatCompatibility ensures that the features present on the table are
 // compatible with the table format version.
 func (w *RawWriter) assertFormatCompatibility() error {
@@ -1912,6 +1886,7 @@ func NewRawWriter(writable objstorage.Writable, o WriterOptions) *RawWriter {
 			sizeClassAwareThreshold: (o.IndexBlockSize*o.SizeClassAwareThreshold + 99) / 100,
 		},
 		compare:               o.Comparer.Compare,
+		suffixCmp:             o.Comparer.CompareSuffixes,
 		split:                 o.Comparer.Split,
 		formatKey:             o.Comparer.FormatKey,
 		compression:           o.Compression,
diff --git a/sstable/reader.go b/sstable/reader.go
index cbec87d4..029fe0c0 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -101,6 +101,7 @@ type Reader struct {
 	logger               base.LoggerAndTracer
 
 	Compare   Compare
+	SuffixCmp CompareSuffixes
 	Equal     Equal
 	FormatKey base.FormatKey
 	Split     Split
@@ -979,11 +980,13 @@ func NewReader(ctx context.Context, f objstorage.Readable, o ReaderOptions) (*Re
 
 	if r.Properties.ComparerName == "" || o.Comparer.Name == r.Properties.ComparerName {
 		r.Compare = o.Comparer.Compare
+		r.SuffixCmp = o.Comparer.CompareSuffixes
 		r.Equal = o.Comparer.Equal
 		r.FormatKey = o.Comparer.FormatKey
 		r.Split = o.Comparer.Split
 	} else if comparer, ok := o.Comparers[r.Properties.ComparerName]; ok {
 		r.Compare = comparer.Compare
+		r.SuffixCmp = comparer.CompareSuffixes
 		r.Equal = comparer.Equal
 		r.FormatKey = comparer.FormatKey
 		r.Split = comparer.Split
diff --git a/sstable/reader_virtual.go b/sstable/reader_virtual.go
index 771cb6a3..37dd6770 100644
--- a/sstable/reader_virtual.go
+++ b/sstable/reader_virtual.go
@@ -190,7 +190,7 @@ func (v *VirtualReader) NewRawRangeKeyIter(
 		transformIter := &keyspan.TransformerIter{
 			FragmentIterator: iter,
 			Transformer:      transform,
-			Compare:          v.reader.Compare,
+			SuffixCmp:        v.reader.SuffixCmp,
 		}
 		iter = transformIter
 	}
diff --git a/sstable/rowblk/rowblk_fragment_iter.go b/sstable/rowblk/rowblk_fragment_iter.go
index 2504e74e..180b5b60 100644
--- a/sstable/rowblk/rowblk_fragment_iter.go
+++ b/sstable/rowblk/rowblk_fragment_iter.go
@@ -253,7 +253,7 @@ func (i *fragmentIter) gatherBackward(kv *base.InternalKV) (*keyspan.Span, error
 	// span.
 
 	// Backwards iteration encounters internal keys in the wrong order.
-	keyspan.SortKeysByTrailer(&i.span.Keys)
+	keyspan.SortKeysByTrailer(i.span.Keys)
 
 	i.applySpanTransforms()
 	return &i.span, nil
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index 289eb7ab..dc6c2187 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -146,8 +146,7 @@ var errBadKind = errors.New("key does not have expected kind (set)")
 
 type blockWithSpan struct {
 	start, end InternalKey
-	data       []byte
-	trailer    block.Trailer
+	physical   block.PhysicalBlock
 }
 
 func rewriteBlocks(
@@ -246,12 +245,10 @@ func rewriteBlocks(
 
 		keyAlloc, output[i].end = cloneKeyWithBuf(scratch, keyAlloc)
 
-		finished, trailer := compressAndChecksum(bw.Finish(), compression, &buf)
+		finished := block.CompressAndChecksum(&buf.compressedBuf, bw.Finish(), compression, &buf.checksummer)
 
 		// copy our finished block into the output buffer.
-		blockAlloc, output[i].data = blockAlloc.Alloc(len(finished))
-		copy(output[i].data, finished)
-		output[i].trailer = trailer
+		output[i].physical = finished.CloneWithByteAlloc(&blockAlloc)
 	}
 	return nil
 }
@@ -340,7 +337,7 @@ func rewriteDataBlocksToWriter(
 
 	for i := range blocks {
 		// Write the rewritten block to the file.
-		bh, err := w.layout.WritePrecompressedDataBlock(blocks[i].data, blocks[i].trailer)
+		bh, err := w.layout.WritePrecompressedDataBlock(blocks[i].physical)
 		if err != nil {
 			return err
 		}
diff --git a/sstable/value_block.go b/sstable/value_block.go
index 0b3f0796..777ce3d4 100644
--- a/sstable/value_block.go
+++ b/sstable/value_block.go
@@ -346,16 +346,16 @@ type valueBlockWriter struct {
 	// compressedBuf is used for compressing the block.
 	compressedBuf *blockBuffer
 	// Sequence of blocks that are finished.
-	blocks []blockAndHandle
+	blocks []bufferedValueBlock
 	// Cumulative value block bytes written so far.
 	totalBlockBytes uint64
 	numValues       uint64
 }
 
-type blockAndHandle struct {
-	block      *blockBuffer
-	handle     block.Handle
-	compressed bool
+type bufferedValueBlock struct {
+	block  block.PhysicalBlock
+	buffer *blockBuffer
+	handle block.Handle
 }
 
 type blockBuffer struct {
@@ -430,12 +430,12 @@ func newValueBlockWriter(
 
 func releaseValueBlockWriter(w *valueBlockWriter) {
 	for i := range w.blocks {
-		if w.blocks[i].compressed {
-			releaseToValueBlockBufPool(&compressedValueBlockBufPool, w.blocks[i].block)
+		if w.blocks[i].block.IsCompressed() {
+			releaseToValueBlockBufPool(&compressedValueBlockBufPool, w.blocks[i].buffer)
 		} else {
-			releaseToValueBlockBufPool(&uncompressedValueBlockBufPool, w.blocks[i].block)
+			releaseToValueBlockBufPool(&uncompressedValueBlockBufPool, w.blocks[i].buffer)
 		}
-		w.blocks[i].block = nil
+		w.blocks[i].buffer = nil
 	}
 	if w.buf != nil {
 		releaseToValueBlockBufPool(&uncompressedValueBlockBufPool, w.buf)
@@ -495,45 +495,27 @@ func (w *valueBlockWriter) addValue(v []byte) (valueHandle, error) {
 }
 
 func (w *valueBlockWriter) compressAndFlush() {
-	// Compress the buffer, discarding the result if the improvement isn't at
-	// least 12.5%.
-	algo := block.NoCompressionIndicator
-	b := w.buf
-	if w.compression != block.NoCompression {
-		algo, w.compressedBuf.b =
-			block.Compress(w.compression, w.buf.b, w.compressedBuf.b[:cap(w.compressedBuf.b)])
-		if len(w.compressedBuf.b) < len(w.buf.b)-len(w.buf.b)/8 {
-			b = w.compressedBuf
-		} else {
-			algo = block.NoCompressionIndicator
-		}
-	}
-	n := len(b.b)
-	if n+block.TrailerLen > cap(b.b) {
-		block := make([]byte, n+block.TrailerLen)
-		copy(block, b.b)
-		b.b = block
-	} else {
-		b.b = b.b[:n+block.TrailerLen]
-	}
-	b.b[n] = byte(algo)
-	w.computeChecksum(b.b)
-	bh := block.Handle{Offset: w.totalBlockBytes, Length: uint64(n)}
-	w.totalBlockBytes += uint64(len(b.b))
+	w.compressedBuf.b = w.compressedBuf.b[:cap(w.compressedBuf.b)]
+	physicalBlock := block.CompressAndChecksum(&w.compressedBuf.b, w.buf.b, w.compression, &w.checksummer)
+	bh := block.Handle{Offset: w.totalBlockBytes, Length: uint64(physicalBlock.LengthWithoutTrailer())}
+	w.totalBlockBytes += uint64(physicalBlock.LengthWithTrailer())
 	// blockFinishedFunc length excludes the block trailer.
-	w.blockFinishedFunc(n)
-	compressed := algo != block.NoCompressionIndicator
-	w.blocks = append(w.blocks, blockAndHandle{
-		block:      b,
-		handle:     bh,
-		compressed: compressed,
-	})
-	// Handed off a buffer to w.blocks, so need get a new one.
-	if compressed {
+	w.blockFinishedFunc(physicalBlock.LengthWithoutTrailer())
+	b := bufferedValueBlock{
+		block:  physicalBlock,
+		handle: bh,
+	}
+
+	// We'll hand off a buffer to w.blocks and get a new one. Which buffer we're
+	// handing off depends on the outcome of compression.
+	if physicalBlock.IsCompressed() {
+		b.buffer = w.compressedBuf
 		w.compressedBuf = compressedValueBlockBufPool.Get().(*blockBuffer)
 	} else {
+		b.buffer = w.buf
 		w.buf = uncompressedValueBlockBufPool.Get().(*blockBuffer)
 	}
+	w.blocks = append(w.blocks, b)
 	w.buf.b = w.buf.b[:0]
 }
 
@@ -556,7 +538,7 @@ func (w *valueBlockWriter) finish(
 	largestOffset := uint64(0)
 	largestLength := uint64(0)
 	for i := range w.blocks {
-		_, err := layout.WriteValueBlock(w.blocks[i].block.b)
+		_, err := layout.WriteValueBlock(w.blocks[i].block)
 		if err != nil {
 			return valueBlocksIndexHandle{}, valueBlocksAndIndexStats{}, err
 		}
diff --git a/sstable/write_queue.go b/sstable/write_queue.go
index 70f8b6b1..7dac6478 100644
--- a/sstable/write_queue.go
+++ b/sstable/write_queue.go
@@ -61,7 +61,7 @@ func newWriteQueue(size int, writer *RawWriter) *writeQueue {
 func (w *writeQueue) performWrite(task *writeTask) error {
 	var bhp BlockHandleWithProperties
 	var err error
-	if bhp.Handle, err = w.writer.layout.WritePrecompressedDataBlock(task.buf.compressed, task.buf.trailer); err != nil {
+	if bhp.Handle, err = w.writer.layout.WritePrecompressedDataBlock(task.buf.physical); err != nil {
 		return err
 	}
 	bhp = BlockHandleWithProperties{Handle: bhp.Handle, Props: task.buf.dataBlockProps}
diff --git a/table_stats.go b/table_stats.go
index 16b0cdeb..cfd1af0f 100644
--- a/table_stats.go
+++ b/table_stats.go
@@ -850,7 +850,7 @@ func newCombinedDeletionKeyspanIter(
 ) (keyspan.FragmentIterator, error) {
 	// The range del iter and range key iter are each wrapped in their own
 	// defragmenting iter. For each iter, abutting spans can always be merged.
-	var equal = keyspan.DefragmentMethodFunc(func(_ base.Equal, a, b *keyspan.Span) bool { return true })
+	var equal = keyspan.DefragmentMethodFunc(func(_ base.CompareSuffixes, a, b *keyspan.Span) bool { return true })
 	// Reduce keys by maintaining a slice of at most length two, corresponding to
 	// the largest and smallest keys in the defragmented span. This maintains the
 	// contract that the emitted slice is sorted by (SeqNum, Kind) descending.
@@ -879,7 +879,7 @@ func newCombinedDeletionKeyspanIter(
 				smallest = last
 			}
 		}
-		if largest.Equal(comparer.Equal, smallest) {
+		if largest.Equal(comparer.CompareSuffixes, smallest) {
 			current = append(current[:0], largest)
 		} else {
 			current = append(current[:0], largest, smallest)
@@ -891,7 +891,7 @@ func newCombinedDeletionKeyspanIter(
 	// merging iter to join the keyspaces into a single keyspace. The separate
 	// iters are only added if the particular key kind is present.
 	mIter := &keyspanimpl.MergingIter{}
-	var transform = keyspan.TransformerFunc(func(cmp base.Compare, in keyspan.Span, out *keyspan.Span) error {
+	var transform = keyspan.TransformerFunc(func(_ base.CompareSuffixes, in keyspan.Span, out *keyspan.Span) error {
 		if in.KeysOrder != keyspan.ByTrailerDesc {
 			panic("pebble: combined deletion iter encountered keys in non-trailer descending order")
 		}
@@ -903,7 +903,7 @@ func newCombinedDeletionKeyspanIter(
 		// sstable into the same keyspanimpl.MergingIter. The MergingIter will
 		// return the keys in the order that the child iterators were provided.
 		// Sort the keys to ensure they're sorted by trailer descending.
-		keyspan.SortKeysByTrailer(&out.Keys)
+		keyspan.SortKeysByTrailer(out.Keys)
 		return nil
 	})
 	mIter.Init(comparer, transform, new(keyspanimpl.MergingBuffers))


Test patch:
diff --git a/internal/base/comparer_test.go b/internal/base/comparer_test.go
index ae49a314..cb1d7ea9 100644
--- a/internal/base/comparer_test.go
+++ b/internal/base/comparer_test.go
@@ -55,6 +55,12 @@ func TestDefAppendSeparator(t *testing.T) {
 	}
 }
 
+func TestDefaultComparer(t *testing.T) {
+	if err := CheckComparer(DefaultComparer, [][]byte{{}, []byte("abc"), []byte("d"), []byte("ef")}, [][]byte{{}}); err != nil {
+		t.Error(err)
+	}
+}
+
 func TestAbbreviatedKey(t *testing.T) {
 	rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
 	randBytes := func(size int) []byte {
diff --git a/internal/compact/spans_test.go b/internal/compact/spans_test.go
index a42fe65d..2a45c28d 100644
--- a/internal/compact/spans_test.go
+++ b/internal/compact/spans_test.go
@@ -71,7 +71,7 @@ func TestRangeKeySpanCompactor(t *testing.T) {
 
 			c = MakeRangeKeySpanCompactor(
 				base.DefaultComparer.Compare,
-				base.DefaultComparer.Equal,
+				base.DefaultComparer.CompareSuffixes,
 				s,
 				ElideTombstonesOutsideOf(keyRanges),
 			)
diff --git a/internal/crdbtest/crdbtest.go b/internal/crdbtest/crdbtest.go
index 588779f4..57763362 100644
--- a/internal/crdbtest/crdbtest.go
+++ b/internal/crdbtest/crdbtest.go
@@ -8,6 +8,7 @@ package crdbtest
 
 import (
 	"bytes"
+	"cmp"
 	"encoding/binary"
 
 	"github.com/cockroachdb/errors"
@@ -24,8 +25,10 @@ const MaxSuffixLen = max(withLockTableLen, withSynthetic, withLogical, withWall)
 
 // Comparer is a base.Comparer for CockroachDB keys.
 var Comparer = base.Comparer{
-	Compare: Compare,
-	Equal:   Equal,
+	Split:           Split,
+	CompareSuffixes: CompareSuffixes,
+	Compare:         Compare,
+	Equal:           Equal,
 	AbbreviatedKey: func(k []byte) uint64 {
 		key, ok := getKeyPartFromEngineKey(k)
 		if !ok {
@@ -84,16 +87,13 @@ var Comparer = base.Comparer{
 		// zero-length version.
 		return append(append(dst, a...), 0)
 	},
-	Split: Split,
-	Name:  "cockroach_comparator",
+	Name: "cockroach_comparator",
 }
 
 // EncodeMVCCKey encodes a MVCC key into dst, growing dst as necessary.
 func EncodeMVCCKey(dst []byte, key []byte, walltime uint64, logical uint32) []byte {
 	if cap(dst) < len(key)+withSynthetic {
-		newKey := make([]byte, len(key), len(key)+withSynthetic)
-		copy(newKey, key)
-		key = newKey
+		dst = make([]byte, 0, len(key)+withSynthetic)
 	}
 	dst = append(dst[:0], key...)
 	return EncodeTimestamp(dst, walltime, logical)
@@ -150,25 +150,23 @@ func DecodeTimestamp(mvccKey []byte) ([]byte, []byte, uint64, uint32) {
 
 // Split implements base.Split for CockroachDB keys.
 func Split(key []byte) int {
-	keyLen := len(key)
-	if keyLen == 0 {
-		return 0
+	if len(key) == 0 {
+		panic(errors.AssertionFailedf("empty key"))
 	}
 	// Last byte is the version length + 1 when there is a version, else it is
 	// 0.
-	versionLen := int(key[keyLen-1])
-	keyPartEnd := keyLen - 1 - versionLen
-	if keyPartEnd < 0 {
-		return keyLen
+	versionLen := int(key[len(key)-1])
+	if versionLen >= len(key) {
+		panic(errors.AssertionFailedf("empty key"))
 	}
-	return keyPartEnd + 1
+	return len(key) - versionLen
 }
 
 // Compare compares cockroach keys, including the version (which could be MVCC
 // timestamps).
 func Compare(a, b []byte) int {
 	if len(a) == 0 || len(b) == 0 {
-		return bytes.Compare(a, b)
+		panic(errors.AssertionFailedf("malformed key: %x, %x", a, b))
 	}
 
 	// NB: For performance, this routine manually splits the key into the
@@ -189,13 +187,8 @@ func Compare(a, b []byte) int {
 	// comparable with the same ordering as if they had a common user key.
 	aSep := aEnd - int(a[aEnd])
 	bSep := bEnd - int(b[bEnd])
-	if aSep == -1 && bSep == -1 {
-		aSep, bSep = 0, 0 // comparing bare suffixes
-	}
-	if aSep < 0 || bSep < 0 {
-		// This should never happen unless there is some sort of corruption of
-		// the keys.
-		return bytes.Compare(a, b)
+	if aSep < 0 || bSep < 0 || a[aSep] != 0 || b[bSep] != 0 {
+		panic(errors.AssertionFailedf("malformed key: %x, %x", a, b))
 	}
 	// Compare the "user key" part of the key.
 	if c := bytes.Compare(a[:aSep], b[:bSep]); c != 0 {
@@ -205,19 +198,19 @@ func Compare(a, b []byte) int {
 	// Compare the version part of the key. Note that when the version is a
 	// timestamp, the timestamp encoding causes byte comparison to be equivalent
 	// to timestamp comparison.
-	aVer := a[aSep:aEnd]
-	bVer := b[bSep:bEnd]
-	if len(aVer) == 0 {
-		if len(bVer) == 0 {
-			return 0
-		}
-		return -1
-	} else if len(bVer) == 0 {
-		return 1
+	return CompareSuffixes(a[aSep+1:], b[bSep+1:])
+}
+
+// CompareSuffixes compares suffixes (normally timestamps).
+func CompareSuffixes(a, b []byte) int {
+	if len(a) == 0 || len(b) == 0 {
+		// Empty suffixes come before non-empty suffixes.
+		return cmp.Compare(len(a), len(b))
 	}
-	aVer = normalizeEngineKeyVersionForCompare(aVer)
-	bVer = normalizeEngineKeyVersionForCompare(bVer)
-	return bytes.Compare(bVer, aVer)
+	return bytes.Compare(
+		normalizeEngineKeyVersionForCompare(b),
+		normalizeEngineKeyVersionForCompare(a),
+	)
 }
 
 // Equal implements base.Equal for Cockroach keys.
@@ -247,23 +240,17 @@ func Equal(a, b []byte) bool {
 	// comparable with the same ordering as if they had a common user key.
 	aSep := aEnd - aVerLen
 	bSep := bEnd - bVerLen
-	if aSep == -1 && bSep == -1 {
-		aSep, bSep = 0, 0 // comparing bare suffixes
-	}
-	if aSep < 0 || bSep < 0 {
-		// This should never happen unless there is some sort of corruption of
-		// the keys.
-		return bytes.Equal(a, b)
-	}
-
 	// Compare the "user key" part of the key.
 	if !bytes.Equal(a[:aSep], b[:bSep]) {
 		return false
 	}
+	if aVerLen == 0 || bVerLen == 0 {
+		return aVerLen == bVerLen
+	}
 
 	// Compare the version part of the key.
-	aVer := a[aSep:aEnd]
-	bVer := b[bSep:bEnd]
+	aVer := a[aSep+1:]
+	bVer := b[bSep+1:]
 	aVer = normalizeEngineKeyVersionForCompare(aVer)
 	bVer = normalizeEngineKeyVersionForCompare(bVer)
 	return bytes.Equal(aVer, bVer)
@@ -272,24 +259,30 @@ func Equal(a, b []byte) bool {
 var zeroLogical [4]byte
 
 func normalizeEngineKeyVersionForCompare(a []byte) []byte {
+	// Check sentinel byte.
+	if len(a) != int(a[len(a)-1]) {
+		panic(errors.AssertionFailedf("malformed suffix: %x", a))
+	}
+	// Strip off sentinel byte.
+	a = a[:len(a)-1]
 	// In general, the version could also be a non-timestamp version, but we know
 	// that engineKeyVersionLockTableLen+mvccEncodedTimeSentinelLen is a different
 	// constant than the above, so there is no danger here of stripping parts from
 	// a non-timestamp version.
-	if len(a) == withSynthetic {
+	if len(a) == withSynthetic-1 {
 		// Strip the synthetic bit component from the timestamp version. The
 		// presence of the synthetic bit does not affect key ordering or equality.
-		a = a[:withLogical]
+		a = a[:withLogical-1]
 	}
-	if len(a) == withLogical {
+	if len(a) == withLogical-1 {
 		// If the timestamp version contains a logical timestamp component that is
 		// zero, strip the component. encodeMVCCTimestampToBuf will typically omit
 		// the entire logical component in these cases as an optimization, but it
 		// does not guarantee to never include a zero logical component.
 		// Additionally, we can fall into this case after stripping off other
 		// components of the key version earlier on in this function.
-		if bytes.Equal(a[withWall:], zeroLogical[:]) {
-			a = a[:withWall]
+		if bytes.Equal(a[withWall-1:], zeroLogical[:]) {
+			a = a[:withWall-1]
 		}
 	}
 	return a
diff --git a/internal/crdbtest/crdbtest_test.go b/internal/crdbtest/crdbtest_test.go
new file mode 100644
index 00000000..e8adc69c
--- /dev/null
+++ b/internal/crdbtest/crdbtest_test.go
@@ -0,0 +1,56 @@
+// 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 crdbtest
+
+import (
+	"bytes"
+	"slices"
+	"testing"
+
+	"github.com/cockroachdb/pebble/internal/base"
+)
+
+func TestComparer(t *testing.T) {
+	prefixes := [][]byte{
+		EncodeMVCCKey(nil, []byte("abc"), 0, 0),
+		EncodeMVCCKey(nil, []byte("d"), 0, 0),
+		EncodeMVCCKey(nil, []byte("ef"), 0, 0),
+	}
+
+	suffixes := [][]byte{{}}
+	for walltime := 3; walltime > 0; walltime-- {
+		for logical := 2; logical >= 0; logical-- {
+			key := EncodeMVCCKey(nil, []byte("foo"), uint64(walltime), uint32(logical))
+			suffix := key[Comparer.Split(key):]
+			suffixes = append(suffixes, suffix)
+
+			if len(suffix) == withWall {
+				// Append a suffix that encodes a zero logical value that should be
+				// ignored in comparisons.
+				newSuffix := slices.Concat(suffix[:withWall-1], zeroLogical[:], []byte{withLogical})
+				if Comparer.CompareSuffixes(suffix, newSuffix) != 0 {
+					t.Fatalf("expected suffixes %x and %x to be equal", suffix, newSuffix)
+				}
+				suffixes = append(suffixes, newSuffix)
+				suffix = newSuffix
+			}
+			if len(suffix) != withLogical {
+				t.Fatalf("unexpected suffix %x", suffix)
+			}
+			// Append a synthetic bit that should be ignored in comparisons.
+			newSuffix := slices.Concat(suffix[:withLogical-1], []byte{1}, []byte{withSynthetic})
+			if Comparer.CompareSuffixes(suffix, newSuffix) != 0 {
+				t.Fatalf("expected suffixes %x and %x to be equal", suffix, newSuffix)
+			}
+			suffixes = append(suffixes, newSuffix)
+		}
+	}
+	// Add some lock table suffixes.
+	suffixes = append(suffixes, append(bytes.Repeat([]byte{1}, withLockTableLen-1), withLockTableLen))
+	suffixes = append(suffixes, append(bytes.Repeat([]byte{2}, withLockTableLen-1), withLockTableLen))
+	if err := base.CheckComparer(&Comparer, prefixes, suffixes); err != nil {
+		t.Error(err)
+	}
+}
diff --git a/internal/keyspan/defragment_test.go b/internal/keyspan/defragment_test.go
index 8ea6afcd..8e017816 100644
--- a/internal/keyspan/defragment_test.go
+++ b/internal/keyspan/defragment_test.go
@@ -8,7 +8,6 @@ import (
 	"bytes"
 	"fmt"
 	"math/rand"
-	"sort"
 	"strings"
 	"testing"
 	"time"
@@ -23,11 +22,11 @@ func TestDefragmentingIter(t *testing.T) {
 	comparer := testkeys.Comparer
 	cmp := comparer.Compare
 	internalEqual := DefragmentInternal
-	alwaysEqual := DefragmentMethodFunc(func(_ base.Equal, _, _ *Span) bool { return true })
+	alwaysEqual := DefragmentMethodFunc(func(_ base.CompareSuffixes, _, _ *Span) bool { return true })
 	staticReducer := StaticDefragmentReducer
 	collectReducer := func(cur, next []Key) []Key {
-		c := keysBySeqNumKind(append(cur, next...))
-		sort.Sort(&c)
+		c := append(cur, next...)
+		SortKeysByTrailer(c)
 		return c
 	}
 
@@ -217,7 +216,7 @@ func testDefragmentingIteRandomizedOnce(t *testing.T, seed int64) {
 }
 
 func fragment(cmp base.Compare, formatKey base.FormatKey, spans []Span) []Span {
-	Sort(cmp, spans)
+	SortSpansByStartKey(cmp, spans)
 	var fragments []Span
 	f := Fragmenter{
 		Cmp:    cmp,
diff --git a/internal/keyspan/interleaving_iter_test.go b/internal/keyspan/interleaving_iter_test.go
index 4d6ff575..143404cd 100644
--- a/internal/keyspan/interleaving_iter_test.go
+++ b/internal/keyspan/interleaving_iter_test.go
@@ -30,7 +30,7 @@ func TestInterleavingIter_Masking(t *testing.T) {
 
 type maskingHooks struct {
 	log        io.Writer
-	cmp        base.Compare
+	suffixCmp  base.CompareSuffixes
 	split      base.Split
 	threshold  []byte
 	maskSuffix []byte
@@ -55,10 +55,10 @@ func (m *maskingHooks) SpanChanged(s *Span) {
 		if s.Keys[i].Suffix == nil {
 			continue
 		}
-		if m.cmp(s.Keys[i].Suffix, m.threshold) < 0 {
+		if m.suffixCmp(s.Keys[i].Suffix, m.threshold) < 0 {
 			continue
 		}
-		if m.maskSuffix == nil || m.cmp(m.maskSuffix, s.Keys[i].Suffix) > 0 {
+		if m.maskSuffix == nil || m.suffixCmp(m.maskSuffix, s.Keys[i].Suffix) > 0 {
 			m.maskSuffix = s.Keys[i].Suffix
 		}
 	}
@@ -66,7 +66,7 @@ func (m *maskingHooks) SpanChanged(s *Span) {
 
 func (m *maskingHooks) SkipPoint(userKey []byte) bool {
 	pointSuffix := userKey[m.split(userKey):]
-	return m.maskSuffix != nil && len(pointSuffix) > 0 && m.cmp(m.maskSuffix, pointSuffix) < 0
+	return m.maskSuffix != nil && len(pointSuffix) > 0 && m.suffixCmp(m.maskSuffix, pointSuffix) < 0
 }
 
 func runInterleavingIterTest(t *testing.T, filename string) {
@@ -76,9 +76,9 @@ func runInterleavingIterTest(t *testing.T, filename string) {
 	var iter InterleavingIter
 	var buf bytes.Buffer
 	hooks := maskingHooks{
-		log:   &buf,
-		cmp:   testkeys.Comparer.Compare,
-		split: testkeys.Comparer.Split,
+		log:       &buf,
+		suffixCmp: testkeys.Comparer.CompareSuffixes,
+		split:     testkeys.Comparer.Split,
 	}
 
 	var prevKV *base.InternalKV
diff --git a/internal/keyspan/keyspanimpl/merging_iter_test.go b/internal/keyspan/keyspanimpl/merging_iter_test.go
index e5bca673..9094e0b4 100644
--- a/internal/keyspan/keyspanimpl/merging_iter_test.go
+++ b/internal/keyspan/keyspanimpl/merging_iter_test.go
@@ -160,7 +160,7 @@ func testFragmenterEquivalenceOnce(t *testing.T, seed int64) {
 			allFragmented = append(allFragmented, span)
 		},
 	}
-	keyspan.Sort(f.Cmp, allSpans)
+	keyspan.SortSpansByStartKey(f.Cmp, allSpans)
 	for _, s := range allSpans {
 		f.Add(s)
 	}
diff --git a/internal/rangekey/coalesce_test.go b/internal/rangekey/coalesce_test.go
index 177969dc..b76a6b45 100644
--- a/internal/rangekey/coalesce_test.go
+++ b/internal/rangekey/coalesce_test.go
@@ -5,7 +5,6 @@
 package rangekey
 
 import (
-	"bytes"
 	"fmt"
 	"testing"
 
@@ -15,22 +14,17 @@ import (
 )
 
 func TestCoalesce(t *testing.T) {
-	var buf bytes.Buffer
-	eq := testkeys.Comparer.Equal
-	cmp := testkeys.Comparer.Compare
-
 	datadriven.RunTest(t, "testdata/coalesce", func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "coalesce":
-			buf.Reset()
 			span := keyspan.ParseSpan(td.Input)
 			coalesced := keyspan.Span{
 				Start: span.Start,
 				End:   span.End,
 			}
-			Coalesce(cmp, eq, span.Keys, &coalesced.Keys)
-			fmt.Fprintln(&buf, coalesced)
-			return buf.String()
+			Coalesce(testkeys.Comparer.CompareSuffixes, span.Keys, &coalesced.Keys)
+			return coalesced.String()
+
 		default:
 			return fmt.Sprintf("unrecognized command %q", td.Cmd)
 		}
diff --git a/internal/rangekeystack/user_iterator_test.go b/internal/rangekeystack/user_iterator_test.go
index 10a0d66e..134acc1c 100644
--- a/internal/rangekeystack/user_iterator_test.go
+++ b/internal/rangekeystack/user_iterator_test.go
@@ -25,7 +25,6 @@ import (
 )
 
 func TestIter(t *testing.T) {
-	eq := testkeys.Comparer.Equal
 	cmp := testkeys.Comparer.Compare
 	var iter keyspanimpl.MergingIter
 	var buf bytes.Buffer
@@ -46,16 +45,12 @@ func TestIter(t *testing.T) {
 			for _, line := range lines {
 				spans = append(spans, keyspan.ParseSpan(line))
 			}
-			transform := keyspan.TransformerFunc(func(cmp base.Compare, s keyspan.Span, dst *keyspan.Span) error {
-				keysBySuffix := keyspan.KeysBySuffix{
-					Cmp:  cmp,
-					Keys: dst.Keys[:0],
-				}
-				rangekey.CoalesceIntoKeysBySuffix(eq, &keysBySuffix, visibleSeqNum, s.Keys)
-				// Update the span with the (potentially reduced) keys slice.  coalesce left
-				// the keys in *dst sorted by suffix. Re-sort them by trailer.
-				dst.Keys = keysBySuffix.Keys
-				keyspan.SortKeysByTrailer(&dst.Keys)
+			transform := keyspan.TransformerFunc(func(suffixCmp base.CompareSuffixes, s keyspan.Span, dst *keyspan.Span) error {
+				dst.Keys = rangekey.CoalesceInto(suffixCmp, dst.Keys[:0], visibleSeqNum, s.Keys)
+				// Update the span with the (potentially reduced) keys slice.
+				// CoalesceInto() left the keys sorted by suffix. Re-sort them by
+				// trailer.
+				keyspan.SortKeysByTrailer(dst.Keys)
 				dst.Start = s.Start
 				dst.End = s.End
 				return nil
@@ -263,7 +258,7 @@ func testDefragmentingIteRandomizedOnce(t *testing.T, seed int64) {
 }
 
 func fragment(cmp base.Compare, formatKey base.FormatKey, spans []keyspan.Span) []keyspan.Span {
-	keyspan.Sort(cmp, spans)
+	keyspan.SortSpansByStartKey(cmp, spans)
 	var fragments []keyspan.Span
 	f := keyspan.Fragmenter{
 		Cmp:    cmp,
@@ -380,7 +375,7 @@ func BenchmarkTransform(b *testing.B) {
 					b.ResetTimer()
 
 					for i := 0; i < b.N; i++ {
-						err := ui.Transform(testkeys.Comparer.Compare, keyspan.Span{Keys: keys}, &dst)
+						err := ui.Transform(testkeys.Comparer.CompareSuffixes, keyspan.Span{Keys: keys}, &dst)
 						if err != nil {
 							b.Fatal(err)
 						}
diff --git a/internal/testkeys/testkeys.go b/internal/testkeys/testkeys.go
index afaf6b2f..b3ed5c70 100644
--- a/internal/testkeys/testkeys.go
+++ b/internal/testkeys/testkeys.go
@@ -44,8 +44,9 @@ var MaxSuffixLen = 1 + len(fmt.Sprintf("%d", int64(math.MaxInt64)))
 
 // Comparer is the comparer for test keys generated by this package.
 var Comparer = &base.Comparer{
-	Compare: compare,
-	Equal:   func(a, b []byte) bool { return compare(a, b) == 0 },
+	CompareSuffixes: compareTimestamps,
+	Compare:         compare,
+	Equal:           func(a, b []byte) bool { return compare(a, b) == 0 },
 	AbbreviatedKey: func(k []byte) uint64 {
 		return base.DefaultComparer.AbbreviatedKey(k[:split(k)])
 	},
@@ -109,35 +110,42 @@ var Comparer = &base.Comparer{
 //     value is smaller.
 func compare(a, b []byte) int {
 	ai, bi := split(a), split(b)
+	if ai == 0 && len(a) > 0 {
+		panic(fmt.Sprintf("Compare called with bare suffix %s", a))
+	}
+	if bi == 0 && len(b) > 0 {
+		panic(fmt.Sprintf("Compare called with bare suffix %s", b))
+	}
 	if v := bytes.Compare(a[:ai], b[:bi]); v != 0 {
 		return v
 	}
-
-	if len(a[ai:]) == 0 {
-		if len(b[bi:]) == 0 {
-			return 0
-		}
-		return -1
-	} else if len(b[bi:]) == 0 {
-		return +1
-	}
 	return compareTimestamps(a[ai:], b[bi:])
 }
 
 func split(a []byte) int {
 	i := bytes.LastIndexByte(a, suffixDelim)
-	if i >= 0 {
+	if i == 0 {
+		panic("split called on bare prefix")
+	}
+	if i > 0 {
 		return i
 	}
 	return len(a)
 }
 
 func compareTimestamps(a, b []byte) int {
-	ai, err := parseUintBytes(bytes.TrimPrefix(a, []byte{suffixDelim}), 10, 64)
+	if len(a) == 0 || len(b) == 0 {
+		// The empty suffix sorts first.
+		return cmp.Compare(len(a), len(b))
+	}
+	if a[0] != suffixDelim || b[0] != suffixDelim {
+		panic(fmt.Sprintf("invalid suffixes %q %q", a, b))
+	}
+	ai, err := parseUintBytes(a[1:], 10, 64)
 	if err != nil {
 		panic(fmt.Sprintf("invalid test mvcc timestamp %q", a))
 	}
-	bi, err := parseUintBytes(bytes.TrimPrefix(b, []byte{suffixDelim}), 10, 64)
+	bi, err := parseUintBytes(b[1:], 10, 64)
 	if err != nil {
 		panic(fmt.Sprintf("invalid test mvcc timestamp %q", b))
 	}
diff --git a/internal/testkeys/testkeys_test.go b/internal/testkeys/testkeys_test.go
index 7573eeb7..75f9a221 100644
--- a/internal/testkeys/testkeys_test.go
+++ b/internal/testkeys/testkeys_test.go
@@ -11,6 +11,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
 )
@@ -157,12 +158,15 @@ func TestSuffix(t *testing.T) {
 		}
 	}
 
-	// Suffixes should be comparable on their own too.
+	// Test CompareSuffixes.
 	a, b := make([]byte, MaxSuffixLen), make([]byte, MaxSuffixLen)
 	for ts := int64(2); ts < 150; ts++ {
 		an := WriteSuffix(a, ts-1)
 		bn := WriteSuffix(b, ts)
-		assertCmp(+1, a[:an], b[:bn])
+		got := Comparer.CompareSuffixes(a[:an], b[:bn])
+		if want := +1; got != want {
+			t.Errorf("CompareSuffixes(%q, %q) = %d, want %d", a, b, got, want)
+		}
 	}
 }
 
@@ -294,3 +298,11 @@ func TestOverflowPanic(t *testing.T) {
 	}()
 	keyCount(6, len(alpha))
 }
+
+func TestComparer(t *testing.T) {
+	if err := base.CheckComparer(Comparer,
+		[][]byte{[]byte("abc"), []byte("d"), []byte("ef")},
+		[][]byte{{}, []byte("@3"), []byte("@2"), []byte("@1")}); err != nil {
+		t.Error(err)
+	}
+}
diff --git a/scan_internal_test.go b/scan_internal_test.go
index 0158c0a7..92f6ae7a 100644
--- a/scan_internal_test.go
+++ b/scan_internal_test.go
@@ -425,7 +425,7 @@ func TestScanInternal(t *testing.T) {
 							keys = append(keys, span.Keys[i])
 							keys[i].Trailer = base.MakeTrailer(0, keys[i].Kind())
 						}
-						keyspan.SortKeysByTrailer(&keys)
+						keyspan.SortKeysByTrailer(keys)
 						require.NoError(t, w.Raw().EncodeSpan(keyspan.Span{
 							Start: span.Start,
 							End:   span.End,
diff --git a/sstable/block/compression_test.go b/sstable/block/compression_test.go
index aa8029a1..5548f8f3 100644
--- a/sstable/block/compression_test.go
+++ b/sstable/block/compression_test.go
@@ -27,8 +27,8 @@ func TestCompressionRoundtrip(t *testing.T) {
 			// not sufficient, Compress should allocate one that is.
 			compressedBuf := make([]byte, rng.Intn(1<<10 /* 1 KiB */))
 
-			btyp, compressed := Compress(compression, payload, compressedBuf)
-			v, err := Decompress(btyp, compressed)
+			btyp, compressed := compress(compression, payload, compressedBuf)
+			v, err := decompress(btyp, compressed)
 			require.NoError(t, err)
 			got := payload
 			if v != nil {
@@ -53,8 +53,31 @@ func TestDecompressionError(t *testing.T) {
 	fauxCompressed = fauxCompressed[:n+compressedPayloadLen]
 	rng.Read(fauxCompressed[n:])
 
-	v, err := Decompress(ZstdCompressionIndicator, fauxCompressed)
+	v, err := decompress(ZstdCompressionIndicator, fauxCompressed)
 	t.Log(err)
 	require.Error(t, err)
 	require.Nil(t, v)
 }
+
+// decompress decompresses an sstable block into memory manually allocated with
+// `cache.Alloc`.  NB: If Decompress returns (nil, nil), no decompression was
+// necessary and the caller may use `b` directly.
+func decompress(algo CompressionIndicator, b []byte) (*cache.Value, error) {
+	if algo == NoCompressionIndicator {
+		return nil, nil
+	}
+	// first obtain the decoded length.
+	decodedLen, prefixLen, err := DecompressedLen(algo, b)
+	if err != nil {
+		return nil, err
+	}
+	b = b[prefixLen:]
+	// Allocate sufficient space from the cache.
+	decoded := cache.Alloc(decodedLen)
+	decodedBuf := decoded.Buf()
+	if err := DecompressInto(algo, b, decodedBuf); err != nil {
+		cache.Free(decoded)
+		return nil, err
+	}
+	return decoded, nil
+}
diff --git a/sstable/data_test.go b/sstable/data_test.go
index cdcad083..6c03acac 100644
--- a/sstable/data_test.go
+++ b/sstable/data_test.go
@@ -293,7 +293,7 @@ func runIterCmd(
 		for iter.Valid() {
 			k := iter.Key().UserKey
 			suffix := k[testkeys.Comparer.Split(k):]
-			if len(suffix) == 0 || testkeys.Comparer.Compare(suffix, maskingSuffix) <= 0 {
+			if len(suffix) == 0 || testkeys.Comparer.CompareSuffixes(suffix, maskingSuffix) <= 0 {
 				return
 			}
 			if direction > 0 {
diff --git a/testdata/ingest b/testdata/ingest
index 354e50e9..48930711 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -54,7 +54,7 @@ Virtual tables: 0 (0B)
 Local tables size: 569B
 Compression types: snappy: 1
 Block cache: 6 entries (945B)  hit rate: 30.8%
-Table cache: 1 entries (728B)  hit rate: 50.0%
+Table cache: 1 entries (736B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
diff --git a/testdata/metrics b/testdata/metrics
index d583567c..5760286d 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -75,7 +75,7 @@ Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 1
 Block cache: 3 entries (484B)  hit rate: 0.0%
-Table cache: 1 entries (728B)  hit rate: 0.0%
+Table cache: 1 entries (736B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -214,7 +214,7 @@ Virtual tables: 0 (0B)
 Local tables size: 595B
 Compression types: snappy: 1
 Block cache: 3 entries (484B)  hit rate: 33.3%
-Table cache: 1 entries (728B)  hit rate: 66.7%
+Table cache: 1 entries (736B)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -488,7 +488,7 @@ Virtual tables: 0 (0B)
 Local tables size: 4.3KB
 Compression types: snappy: 7
 Block cache: 12 entries (1.9KB)  hit rate: 9.1%
-Table cache: 1 entries (728B)  hit rate: 53.8%
+Table cache: 1 entries (736B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -551,7 +551,7 @@ Virtual tables: 0 (0B)
 Local tables size: 6.1KB
 Compression types: snappy: 10
 Block cache: 12 entries (1.9KB)  hit rate: 9.1%
-Table cache: 1 entries (728B)  hit rate: 53.8%
+Table cache: 1 entries (736B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -822,7 +822,7 @@ Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 1
 Block cache: 1 entries (440B)  hit rate: 0.0%
-Table cache: 1 entries (728B)  hit rate: 0.0%
+Table cache: 1 entries (736B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -869,7 +869,7 @@ Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 2
 Block cache: 6 entries (996B)  hit rate: 0.0%
-Table cache: 1 entries (728B)  hit rate: 50.0%
+Table cache: 1 entries (736B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -917,7 +917,7 @@ Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 3
 Block cache: 6 entries (996B)  hit rate: 0.0%
-Table cache: 1 entries (728B)  hit rate: 50.0%
+Table cache: 1 entries (736B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0