Repository: cockroachdb/pebble
Pull request #4078 (2024-10-19 04:52:03)

colblk: fix UintsBuilder bug in default mode
When we are in `useDefault` mode, the slice can be smaller than the
number of rows. In that case, the existing code was reading outside
the underlying slice and writing out garbage.

We should reconsider the use of `UnsafeSlice` here, as I don't think
it's buying us much performance-wise.

Informs https://github.com/cockroachdb/cockroach/issues/132964

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


Patch:
diff --git a/internal/base/comparer.go b/internal/base/comparer.go
index d2e76a8f..78a7aadd 100644
--- a/internal/base/comparer.go
+++ b/internal/base/comparer.go
@@ -26,37 +26,51 @@ import (
 //
 // 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).
+// ComparePointSuffixes).
 //
 // 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,
-//		                otherwise either 0 or CompareSuffixes(suffix(a), suffix(b))
-//
-// The "either 0" part is because CompareSuffixes is allowed to be more strict;
-// see CompareSuffixes.
+//		                otherwise ComparePointSuffixes(suffix(a), suffix(b))
 //
 // Compare defaults to using the formula above but it can be customized if there
 // is a (potentially faster) specialization or it has to compare suffixes
 // differently.
 type Compare func(a, b []byte) int
 
-// CompareSuffixes compares two key suffixes and returns -1, 0, or +1.
+// CompareRangeSuffixes compares two suffixes where either or both suffix
+// originates from a range key and returns -1, 0, or +1.
 //
 // For historical reasons (see
 // https://github.com/cockroachdb/cockroach/issues/130533 for a summary), we
 // allow this function to be more strict than Compare. Specifically, Compare may
-// treat two suffixes as equal whereas CompareSuffixes might not.
+// treat two suffixes as equal whereas CompareRangeSuffixes might not.
+//
+// CompareRangeSuffixes is allowed to be more strict than see
+// ComparePointSuffixes, meaning it may return -1 or +1 when
+// ComparePointSuffixes would return 0.
+//
+// The empty slice suffix must be 'less than' any non-empty suffix.
+type CompareRangeSuffixes func(a, b []byte) int
+
+// ComparePointSuffixes compares two point key suffixes and returns -1, 0, or +1.
+//
+// For historical reasons (see
+// https://github.com/cockroachdb/cockroach/issues/130533 for a summary), this
+// function is distinct from CompareRangeSuffixes. Specifically,
+// ComparePointSuffixes may treat two suffixes as equal whereas
+// CompareRangeSuffixes might not. Unlike CompareRangeSuffixes, this function
+// must agree with Compare.
 //
 // 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
+type ComparePointSuffixes func(a, b []byte) int
 
-// defaultCompare implements Compare in terms of Split and CompareSuffixes, as
+// defaultCompare implements Compare in terms of Split and ComparePointSuffixes, as
 // mentioned above.
-func defaultCompare(split Split, compareSuffixes CompareSuffixes, a, b []byte) int {
+func defaultCompare(split Split, compareSuffixes ComparePointSuffixes, a, b []byte) int {
 	an := split(a)
 	bn := split(b)
 	if prefixCmp := bytes.Compare(a[:an], b[:bn]); prefixCmp != 0 {
@@ -191,11 +205,13 @@ type Comparer struct {
 	// if it is not specified.
 	Split Split
 
-	// CompareSuffixes defaults to bytes.Compare if it is not specified.
-	CompareSuffixes CompareSuffixes
+	// CompareRangeSuffixes defaults to bytes.Compare if it is not specified.
+	CompareRangeSuffixes CompareRangeSuffixes
+	// ComparePointSuffixes defaults to bytes.Compare if it is not specified.
+	ComparePointSuffixes ComparePointSuffixes
 
 	// Compare defaults to a generic implementation that uses Split,
-	// bytes.Compare, and CompareSuffixes if it is not specified.
+	// bytes.Compare, and ComparePointSuffixes if it is not specified.
 	Compare Compare
 	// Equal defaults to using Compare() == 0 if it is not specified.
 	Equal Equal
@@ -225,7 +241,7 @@ func (c *Comparer) EnsureDefaults() *Comparer {
 	if c.AbbreviatedKey == nil || c.Separator == nil || c.Successor == nil || c.Name == "" {
 		panic("invalid Comparer: mandatory field not set")
 	}
-	if c.CompareSuffixes != nil && c.Compare != nil && c.Equal != nil && c.Split != nil && c.FormatKey != nil {
+	if c.CompareRangeSuffixes != nil && c.Compare != nil && c.Equal != nil && c.Split != nil && c.FormatKey != nil {
 		return c
 	}
 	n := &Comparer{}
@@ -234,17 +250,17 @@ func (c *Comparer) EnsureDefaults() *Comparer {
 	if n.Split == nil {
 		n.Split = DefaultSplit
 	}
-	if n.CompareSuffixes == nil && n.Compare == nil && n.Equal == nil {
-		n.CompareSuffixes = bytes.Compare
+	if n.CompareRangeSuffixes == nil && n.Compare == nil && n.Equal == nil {
+		n.CompareRangeSuffixes = bytes.Compare
 		n.Compare = bytes.Compare
 		n.Equal = bytes.Equal
 	} else {
-		if n.CompareSuffixes == nil {
-			n.CompareSuffixes = bytes.Compare
+		if n.CompareRangeSuffixes == nil {
+			n.CompareRangeSuffixes = bytes.Compare
 		}
 		if n.Compare == nil {
 			n.Compare = func(a, b []byte) int {
-				return defaultCompare(n.Split, n.CompareSuffixes, a, b)
+				return defaultCompare(n.Split, n.ComparePointSuffixes, a, b)
 			}
 		}
 		if n.Equal == nil {
@@ -262,9 +278,10 @@ 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{
-	CompareSuffixes: bytes.Compare,
-	Compare:         bytes.Compare,
-	Equal:           bytes.Equal,
+	ComparePointSuffixes: bytes.Compare,
+	CompareRangeSuffixes: bytes.Compare,
+	Compare:              bytes.Compare,
+	Equal:                bytes.Equal,
 
 	AbbreviatedKey: func(key []byte) uint64 {
 		if len(key) >= 8 {
@@ -394,7 +411,7 @@ func MakeAssertComparer(c Comparer) Comparer {
 		Compare: func(a []byte, b []byte) int {
 			res := c.Compare(a, b)
 			// Verify that Compare is consistent with the default implementation.
-			if expected := defaultCompare(c.Split, c.CompareSuffixes, a, b); res != expected {
+			if expected := defaultCompare(c.Split, c.ComparePointSuffixes, a, b); res != expected {
 				panic(AssertionFailedf("%s: Compare(%s, %s)=%d, expected %d",
 					c.Name, c.FormatKey(a), c.FormatKey(b), res, expected))
 			}
@@ -411,15 +428,16 @@ func MakeAssertComparer(c Comparer) Comparer {
 		},
 
 		// TODO(radu): add more checks.
-		CompareSuffixes:    c.CompareSuffixes,
-		AbbreviatedKey:     c.AbbreviatedKey,
-		Separator:          c.Separator,
-		Successor:          c.Successor,
-		ImmediateSuccessor: c.ImmediateSuccessor,
-		FormatKey:          c.FormatKey,
-		Split:              c.Split,
-		FormatValue:        c.FormatValue,
-		Name:               c.Name,
+		ComparePointSuffixes: c.ComparePointSuffixes,
+		CompareRangeSuffixes: c.CompareRangeSuffixes,
+		AbbreviatedKey:       c.AbbreviatedKey,
+		Separator:            c.Separator,
+		Successor:            c.Successor,
+		ImmediateSuccessor:   c.ImmediateSuccessor,
+		FormatKey:            c.FormatKey,
+		Split:                c.Split,
+		FormatValue:          c.FormatValue,
+		Name:                 c.Name,
 	}
 }
 
@@ -432,9 +450,16 @@ func CheckComparer(c *Comparer, prefixes [][]byte, suffixes [][]byte) error {
 	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")
+	slices.SortFunc(suffixes, c.CompareRangeSuffixes)
+	if !slices.IsSortedFunc(suffixes, c.CompareRangeSuffixes) {
+		return errors.Errorf("CompareRangeSuffixes is inconsistent")
+	}
+	// Verify the ordering imposed by CompareRangeSuffixes is considered a valid
+	// ordering for point suffixes. CompareRangeSuffixes imposes a stricter
+	// ordering than ComaprePointSuffixes, but a CompareRangesSuffixes ordering
+	// must be a valid ordering for point suffixes.
+	if !slices.IsSortedFunc(suffixes, c.ComparePointSuffixes) {
+		return errors.Errorf("ComparePointSuffixes is inconsistent")
 	}
 
 	n := len(prefixes)
@@ -456,8 +481,7 @@ func CheckComparer(c *Comparer, prefixes [][]byte, suffixes [][]byte) error {
 		for i := 1; i < len(suffixes); i++ {
 			a := slices.Concat(p, suffixes[i-1])
 			b := slices.Concat(p, suffixes[i])
-			// Make sure the Compare function agrees with CompareSuffixes, with the
-			// caveat that it can consider some consecutive suffixes to be equal.
+			// Make sure the Compare function agrees with ComparePointSuffixes.
 			if cmp := c.Compare(a, b); cmp > 0 {
 				return errors.Errorf("Compare(%s, %s)=%d, expected <= 0", c.FormatKey(a), c.FormatKey(b), cmp)
 			}
@@ -481,11 +505,18 @@ func CheckComparer(c *Comparer, prefixes [][]byte, suffixes [][]byte) error {
 							return errors.Errorf("Compare(%s, %s)=%d, expected %d", c.FormatKey(a), c.FormatKey(b), result, prefixCmp)
 						}
 					} else {
-						// We allow Compare to return equality for suffixes even when
-						// CompareSuffixes does not.
+						// The prefixes are equal, so Compare's result should
+						// agree with ComparePointSuffixes.
+						if suffixCmp := c.ComparePointSuffixes(as, bs); result != suffixCmp {
+							return errors.Errorf("Compare(%s, %s)=%d but ComparePointSuffixes(%q, %q)=%d",
+								c.FormatKey(a), c.FormatKey(b), result, as, bs, suffixCmp)
+						}
+						// If result == 0, CompareRangeSuffixes may not agree
+						// with ComparePointSuffixes, but otherwise it should.
 						if result != 0 {
-							if suffixCmp := c.CompareSuffixes(as, bs); result != suffixCmp {
-								return errors.Errorf("Compare(%s, %s)=%d, expected %d", c.FormatKey(a), c.FormatKey(b), result, suffixCmp)
+							if suffixCmp := c.CompareRangeSuffixes(as, bs); result != suffixCmp {
+								return errors.Errorf("Compare(%s, %s)=%d, expected %d",
+									c.FormatKey(a), c.FormatKey(b), result, suffixCmp)
 							}
 						}
 					}
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index b5c710ee..42f68dfa 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -149,7 +149,7 @@ import (
 // and mutates the contents of the returned key and value.
 type Iter struct {
 	cmp       base.Compare
-	suffixCmp base.CompareSuffixes
+	suffixCmp base.CompareRangeSuffixes
 
 	cfg IterConfig
 
@@ -310,7 +310,7 @@ func NewIter(
 	cfg.ensureDefaults()
 	i := &Iter{
 		cmp:       cfg.Comparer.Compare,
-		suffixCmp: cfg.Comparer.CompareSuffixes,
+		suffixCmp: cfg.Comparer.CompareRangeSuffixes,
 		cfg:       cfg,
 		// We don't want a nil keyBuf because if the first key we encounter is
 		// empty, it would become nil.
diff --git a/internal/compact/spans.go b/internal/compact/spans.go
index 03385bb5..e1b76d03 100644
--- a/internal/compact/spans.go
+++ b/internal/compact/spans.go
@@ -83,14 +83,17 @@ func (c *RangeDelSpanCompactor) Compact(span, output *keyspan.Span) {
 // for at most one "compacted" span.
 type RangeKeySpanCompactor struct {
 	cmp       base.Compare
-	suffixCmp base.CompareSuffixes
+	suffixCmp base.CompareRangeSuffixes
 	snapshots Snapshots
 	elider    rangeTombstoneElider
 }
 
 // MakeRangeKeySpanCompactor creates a new compactor for range key spans.
 func MakeRangeKeySpanCompactor(
-	cmp base.Compare, suffixCmp base.CompareSuffixes, snapshots Snapshots, elision TombstoneElision,
+	cmp base.Compare,
+	suffixCmp base.CompareRangeSuffixes,
+	snapshots Snapshots,
+	elision TombstoneElision,
 ) RangeKeySpanCompactor {
 	c := RangeKeySpanCompactor{
 		cmp:       cmp,
diff --git a/internal/keyspan/defragment.go b/internal/keyspan/defragment.go
index df8b7dac..a8a17320 100644
--- a/internal/keyspan/defragment.go
+++ b/internal/keyspan/defragment.go
@@ -29,17 +29,17 @@ 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(suffixCmp base.CompareSuffixes, left, right *Span) bool
+	ShouldDefragment(suffixCmp base.CompareRangeSuffixes, 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(suffixCmp base.CompareSuffixes, left, right *Span) bool
+type DefragmentMethodFunc func(suffixCmp base.CompareRangeSuffixes, left, right *Span) bool
 
 // ShouldDefragment calls f(equal, left, right).
 func (f DefragmentMethodFunc) ShouldDefragment(
-	suffixCmp base.CompareSuffixes, left, right *Span,
+	suffixCmp base.CompareRangeSuffixes, left, right *Span,
 ) bool {
 	return f(suffixCmp, left, right)
 }
@@ -51,7 +51,7 @@ func (f DefragmentMethodFunc) ShouldDefragment(
 // 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(suffixCmp base.CompareSuffixes, a, b *Span) bool {
+var DefragmentInternal DefragmentMethod = DefragmentMethodFunc(func(suffixCmp base.CompareRangeSuffixes, a, b *Span) bool {
 	if a.KeysOrder != ByTrailerDesc || b.KeysOrder != ByTrailerDesc {
 		panic("pebble: span keys unexpectedly not in trailer descending order")
 	}
@@ -454,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.comparer.CompareSuffixes, i.iterSpan, &i.curr)
+	return (!left.Empty() && !right.Empty()) && i.method.ShouldDefragment(i.comparer.CompareRangeSuffixes, i.iterSpan, &i.curr)
 }
 
 // defragmentForward defragments spans in the forward direction, starting from
diff --git a/internal/keyspan/keyspanimpl/merging_iter.go b/internal/keyspan/keyspanimpl/merging_iter.go
index edcd0600..3636760d 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.CompareSuffixes, m.span, &m.span); err != nil {
+	if err := m.transformer.Transform(m.comparer.CompareRangeSuffixes, 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 54fcfc7a..a142248a 100644
--- a/internal/keyspan/span.go
+++ b/internal/keyspan/span.go
@@ -100,7 +100,7 @@ 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(suffixCmp base.CompareSuffixes, b Key) bool {
+func (k Key) Equal(suffixCmp base.CompareRangeSuffixes, b Key) bool {
 	return k.Trailer == b.Trailer &&
 		suffixCmp(k.Suffix, b.Suffix) == 0 &&
 		bytes.Equal(k.Value, b.Value)
@@ -471,7 +471,7 @@ func SortKeysByTrailer(keys []Key) {
 }
 
 // SortKeysBySuffix sorts a Keys slice by suffix.
-func SortKeysBySuffix(suffixCmp base.CompareSuffixes, keys []Key) {
+func SortKeysBySuffix(suffixCmp base.CompareRangeSuffixes, keys []Key) {
 	slices.SortFunc(keys, func(a, b Key) int {
 		return suffixCmp(a.Suffix, b.Suffix)
 	})
diff --git a/internal/keyspan/transformer.go b/internal/keyspan/transformer.go
index 037a2e30..9510cf26 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(suffixCmp base.CompareSuffixes, in Span, out *Span) error
+	Transform(suffixCmp base.CompareRangeSuffixes, 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.CompareSuffixes, Span, *Span) error
+type TransformerFunc func(base.CompareRangeSuffixes, Span, *Span) error
 
 // Transform calls f(cmp, in, out).
-func (tf TransformerFunc) Transform(suffixCmp base.CompareSuffixes, in Span, out *Span) error {
+func (tf TransformerFunc) Transform(suffixCmp base.CompareRangeSuffixes, in Span, out *Span) error {
 	return tf(suffixCmp, in, out)
 }
 
 // NoopTransform is a Transformer that performs no mutations.
-var NoopTransform Transformer = TransformerFunc(func(_ base.CompareSuffixes, s Span, dst *Span) error {
+var NoopTransform Transformer = TransformerFunc(func(_ base.CompareRangeSuffixes, 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.CompareSuffixes, s S
 // VisibleTransform filters keys that are invisible at the provided snapshot
 // sequence number.
 func VisibleTransform(snapshot base.SeqNum) Transformer {
-	return TransformerFunc(func(_ base.CompareSuffixes, s Span, dst *Span) error {
+	return TransformerFunc(func(_ base.CompareRangeSuffixes, s Span, dst *Span) error {
 		dst.Start, dst.End = s.Start, s.End
 		dst.Keys = dst.Keys[:0]
 		for _, k := range s.Keys {
@@ -59,7 +59,7 @@ type TransformerIter struct {
 	// Transformer is applied on every Span returned by this iterator.
 	Transformer Transformer
 	// Suffix comparer in use for this keyspace.
-	SuffixCmp base.CompareSuffixes
+	SuffixCmp base.CompareRangeSuffixes
 
 	span Span
 }
diff --git a/internal/rangekey/coalesce.go b/internal/rangekey/coalesce.go
index c89e7542..8c3445df 100644
--- a/internal/rangekey/coalesce.go
+++ b/internal/rangekey/coalesce.go
@@ -49,7 +49,7 @@ import (
 // set or unset but not both.
 //
 // The resulting dst Keys slice is sorted by InternalKeyTrailer.
-func Coalesce(suffixCmp base.CompareSuffixes, keys []keyspan.Key, dst *[]keyspan.Key) {
+func Coalesce(suffixCmp base.CompareRangeSuffixes, keys []keyspan.Key, dst *[]keyspan.Key) {
 	// TODO(jackson): Currently, Coalesce doesn't actually perform the sequence
 	// number promotion described in the comment above.
 	*dst = CoalesceInto(suffixCmp, (*dst)[:0], math.MaxUint64, keys)
@@ -61,7 +61,7 @@ func Coalesce(suffixCmp base.CompareSuffixes, keys []keyspan.Key, dst *[]keyspan
 // 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,
+	suffixCmp base.CompareRangeSuffixes, dst []keyspan.Key, snapshot base.SeqNum, keys []keyspan.Key,
 ) []keyspan.Key {
 	dst = dst[:0]
 	// First, enforce visibility and RangeKeyDelete mechanics. We only need to
@@ -165,7 +165,7 @@ type ForeignSSTTransformer struct {
 
 // Transform implements the Transformer interface.
 func (f *ForeignSSTTransformer) Transform(
-	suffixCmp base.CompareSuffixes, s keyspan.Span, dst *keyspan.Span,
+	suffixCmp base.CompareRangeSuffixes, s keyspan.Span, dst *keyspan.Span,
 ) error {
 	// Apply shadowing of keys.
 	dst.Start = s.Start
diff --git a/internal/rangekeystack/user_iterator.go b/internal/rangekeystack/user_iterator.go
index 232dba6d..3d793510 100644
--- a/internal/rangekeystack/user_iterator.go
+++ b/internal/rangekeystack/user_iterator.go
@@ -144,7 +144,7 @@ func (ui *UserIteratorConfig) SetBounds(lower, upper []byte) {
 // sequence number, and hold their Keys sorted by Suffix (except if internalKeys
 // is true, then keys remain sorted by trailer.
 func (ui *UserIteratorConfig) Transform(
-	suffixCmp base.CompareSuffixes, s keyspan.Span, dst *keyspan.Span,
+	suffixCmp base.CompareRangeSuffixes, s keyspan.Span, dst *keyspan.Span,
 ) error {
 	// Apply shadowing of keys.
 	dst.Start = s.Start
@@ -196,7 +196,7 @@ func (ui *UserIteratorConfig) Transform(
 // 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(
-	suffixCmp base.CompareSuffixes, a, b *keyspan.Span,
+	suffixCmp base.CompareRangeSuffixes, a, b *keyspan.Span,
 ) bool {
 	// This method is not called with internalKeys = true.
 	if ui.internalKeys {
diff --git a/iterator.go b/iterator.go
index ab47c11b..4b21b936 100644
--- a/iterator.go
+++ b/iterator.go
@@ -2201,7 +2201,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.comparer.CompareSuffixes(s.Keys[j].Suffix, s.Keys[j-1].Suffix) < 0 {
+			} else if j > 0 && i.comparer.CompareRangeSuffixes(s.Keys[j].Suffix, s.Keys[j-1].Suffix) < 0 {
 				panic("pebble: user iteration encountered range keys not in suffix order")
 			}
 		}
@@ -2692,7 +2692,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.comparer.CompareSuffixes(o.RangeKeyMasking.Suffix, i.opts.RangeKeyMasking.Suffix) == 0 &&
+		i.comparer.CompareRangeSuffixes(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 ce8d127a..fbbfff2a 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -150,7 +150,7 @@ func writeSSTForIngestion(
 					}
 				}
 			}
-			rangekey.Coalesce(t.opts.Comparer.CompareSuffixes, keys, &collapsed.Keys)
+			rangekey.Coalesce(t.opts.Comparer.CompareRangeSuffixes, keys, &collapsed.Keys)
 			for i := range collapsed.Keys {
 				collapsed.Keys[i].Trailer = base.MakeTrailer(0, collapsed.Keys[i].Kind())
 			}
diff --git a/range_keys.go b/range_keys.go
index b50f5624..1fe7c32c 100644
--- a/range_keys.go
+++ b/range_keys.go
@@ -226,7 +226,7 @@ func containsAnyRangeKeys(iter manifest.LevelIterator) bool {
 
 type rangeKeyMasking struct {
 	cmp       base.Compare
-	suffixCmp base.CompareSuffixes
+	suffixCmp base.CompareRangeSuffixes
 	split     base.Split
 	filter    BlockPropertyFilterMask
 	// maskActiveSuffix holds the suffix of a range key currently acting as a
@@ -246,7 +246,7 @@ type rangeKeyMasking struct {
 
 func (m *rangeKeyMasking) init(parent *Iterator, c *base.Comparer) {
 	m.cmp = c.Compare
-	m.suffixCmp = c.CompareSuffixes
+	m.suffixCmp = c.CompareRangeSuffixes
 	m.split = c.Split
 	if parent.opts.RangeKeyMasking.Filter != nil {
 		m.filter = parent.opts.RangeKeyMasking.Filter()
diff --git a/sstable/block/buffer_pool.go b/sstable/block/buffer_pool.go
index bd7d2217..38809b41 100644
--- a/sstable/block/buffer_pool.go
+++ b/sstable/block/buffer_pool.go
@@ -38,13 +38,13 @@ func (b Value) getInternalBuf() []byte {
 	return b.v.Buf()
 }
 
-// Get returns the byte slice for the block data.
-func (b Value) Get() []byte {
+// BlockData returns the byte slice for the block data.
+func (b Value) BlockData() []byte {
 	return b.getInternalBuf()[MetadataSize:]
 }
 
 // GetMetadata returns the block metadata.
-func (b Value) GetMetadata() *Metadata {
+func (b Value) BlockMetadata() *Metadata {
 	return (*Metadata)(b.getInternalBuf())
 }
 
diff --git a/sstable/colblk/data_block.go b/sstable/colblk/data_block.go
index c25c610b..e38ef2e8 100644
--- a/sstable/colblk/data_block.go
+++ b/sstable/colblk/data_block.go
@@ -204,7 +204,7 @@ func (w *defaultKeyWriter) ComparePrev(key []byte) KeyComparison {
 	// CommonPrefixLen, it would sort after [key].)
 	if cmpv.CommonPrefixLen == cmpv.PrefixLen {
 		// The keys share the same MVCC prefix. Compare the suffixes.
-		cmpv.UserKeyComparison = int32(w.comparer.CompareSuffixes(key[cmpv.PrefixLen:],
+		cmpv.UserKeyComparison = int32(w.comparer.CompareRangeSuffixes(key[cmpv.PrefixLen:],
 			w.suffixes.UnsafeGet(w.suffixes.rows-1)))
 		if invariants.Enabled {
 			if !w.comparer.Equal(lp, key[:cmpv.PrefixLen]) {
@@ -344,7 +344,7 @@ func (ks *defaultKeySeeker) SeekGE(
 func (ks *defaultKeySeeker) seekGEOnSuffix(index int, suffix []byte) (row int) {
 	// The search key's prefix exactly matches the prefix of the row at index.
 	// If the row at index has a suffix >= [suffix], then return the row.
-	if ks.comparer.CompareSuffixes(ks.suffixes.At(index), suffix) >= 0 {
+	if ks.comparer.CompareRangeSuffixes(ks.suffixes.At(index), suffix) >= 0 {
 		return index
 	}
 	// Otherwise, the row at [index] sorts before the search key and we need to
@@ -357,7 +357,7 @@ func (ks *defaultKeySeeker) seekGEOnSuffix(index int, suffix []byte) (row int) {
 	for l < u {
 		h := int(uint(l+u) >> 1) // avoid overflow when computing h
 		// l ≤ h < u
-		if ks.comparer.CompareSuffixes(ks.suffixes.At(h), suffix) >= 0 {
+		if ks.comparer.CompareRangeSuffixes(ks.suffixes.At(h), suffix) >= 0 {
 			u = h // preserves f(u) == true
 		} else {
 			l = h + 1 // preserves f(l-1) == false
@@ -737,13 +737,11 @@ func (rw *DataBlockRewriter) RewriteSuffixes(
 	return start, end, rewritten, nil
 }
 
-// DataBlockDecoderSize is the size of a DataBlockDecoder struct. If allocating
-// memory for a data block, the caller may want to additionally allocate memory
-// for the corresponding DataBlockDecoder.
-const DataBlockDecoderSize = unsafe.Sizeof(DataBlockDecoder{})
-
 // Assert that a DataBlockDecoder can fit inside block.Metadata.
-const _ uint = block.MetadataSize - uint(DataBlockDecoderSize)
+const _ uint = block.MetadataSize - uint(unsafe.Sizeof(DataBlockDecoder{}))
+
+// Assert that an IndexBlockDecoder can fit inside block.Metadata.
+const _ uint = block.MetadataSize - uint(unsafe.Sizeof(IndexBlockDecoder{}))
 
 // InitDataBlockMetadata initializes the metadata for a data block.
 func InitDataBlockMetadata(schema KeySchema, md *block.Metadata, data []byte) (err error) {
@@ -763,6 +761,23 @@ func InitDataBlockMetadata(schema KeySchema, md *block.Metadata, data []byte) (e
 	return nil
 }
 
+// InitIndexBlockMetadata initializes the metadata for an index block.
+func InitIndexBlockMetadata(md *block.Metadata, data []byte) (err error) {
+	if uintptr(unsafe.Pointer(md))%8 != 0 {
+		return errors.AssertionFailedf("metadata is not 8-byte aligned")
+	}
+	d := (*IndexBlockDecoder)(unsafe.Pointer(md))
+	// Initialization can panic; convert panics to corruption errors (so higher
+	// layers can add file number and offset information).
+	defer func() {
+		if r := recover(); r != nil {
+			err = base.CorruptionErrorf("error initializing data block metadata: %v", r)
+		}
+	}()
+	d.Init(data)
+	return nil
+}
+
 // A DataBlockDecoder holds state for interpreting a columnar data block. It may
 // be shared among multiple DataBlockIters.
 type DataBlockDecoder struct {
@@ -899,7 +914,7 @@ func (i *DataBlockIter) InitOnce(
 }
 
 // Init initializes the data block iterator, configuring it to read from the
-// provided reader.
+// provided decoder.
 func (i *DataBlockIter) Init(d *DataBlockDecoder, transforms block.IterTransforms) error {
 	i.d = d
 	// Leave i.h unchanged.
diff --git a/sstable/colblk/index_block.go b/sstable/colblk/index_block.go
index 2fc7d294..de46035d 100644
--- a/sstable/colblk/index_block.go
+++ b/sstable/colblk/index_block.go
@@ -6,6 +6,7 @@ package colblk
 
 import (
 	"slices"
+	"unsafe"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -192,7 +193,9 @@ type IndexIter struct {
 	syntheticSuffix block.SyntheticSuffix
 	noTransforms    bool
 
-	h            block.BufferHandle
+	h block.BufferHandle
+	// TODO(radu): remove allocDecoder and require any Init callers to provide the
+	// decoder.
 	allocDecoder IndexBlockDecoder
 	keyBuf       []byte
 }
@@ -200,7 +203,7 @@ type IndexIter struct {
 // Assert that IndexIter satisfies the block.IndexBlockIterator interface.
 var _ block.IndexBlockIterator = (*IndexIter)(nil)
 
-// InitWithDecoder initializes an index iterator from the provided reader.
+// InitWithDecoder initializes an index iterator from the provided decoder.
 func (i *IndexIter) InitWithDecoder(
 	compare base.Compare, split base.Split, d *IndexBlockDecoder, transforms block.IterTransforms,
 ) {
@@ -209,6 +212,7 @@ func (i *IndexIter) InitWithDecoder(
 		split:           split,
 		d:               d,
 		n:               int(d.bd.header.Rows),
+		row:             -1,
 		h:               i.h,
 		allocDecoder:    i.allocDecoder,
 		keyBuf:          i.keyBuf,
@@ -233,15 +237,10 @@ func (i *IndexIter) Init(
 func (i *IndexIter) InitHandle(
 	cmp base.Compare, split base.Split, blk block.BufferHandle, transforms block.IterTransforms,
 ) error {
-	// TODO(jackson): If block.h != nil, use a *IndexBlockDecoder that's allocated
-	// when the block is loaded into the block cache. On cache hits, this will
-	// reduce the amount of setup necessary to use an iterator. (It's relatively
-	// common to open an iterator and perform just a few seeks, so avoiding the
-	// overhead can be material.)
 	i.h.Release()
 	i.h = blk
-	i.allocDecoder.Init(i.h.BlockData())
-	i.InitWithDecoder(cmp, split, &i.allocDecoder, transforms)
+	d := (*IndexBlockDecoder)(unsafe.Pointer(blk.BlockMetadata()))
+	i.InitWithDecoder(cmp, split, d, transforms)
 	return nil
 }
 
diff --git a/sstable/colblk/uints.go b/sstable/colblk/uints.go
index ceabbaf4..bfafd87e 100644
--- a/sstable/colblk/uints.go
+++ b/sstable/colblk/uints.go
@@ -361,15 +361,15 @@ func (b *UintBuilder) Finish(col, rows int, offset uint32, buf []byte) uint32 {
 	// values slice if there's actually a non-nil ptr.
 	var valuesSlice []uint64
 	if b.array.elems.ptr != nil {
-		valuesSlice = b.array.elems.Slice(rows)
+		valuesSlice = b.array.elems.Slice(min(rows, b.array.n))
 	}
-	return uintColumnFinish(minimum, valuesSlice, e, offset, buf)
+	return uintColumnFinish(rows, minimum, valuesSlice, e, offset, buf)
 }
 
 // uintColumnFinish finishes the column of unsigned integers of type T, applying
 // the given encoding.
 func uintColumnFinish(
-	minimum uint64, values []uint64, e UintEncoding, offset uint32, buf []byte,
+	rows int, minimum uint64, values []uint64, e UintEncoding, offset uint32, buf []byte,
 ) uint32 {
 	buf[offset] = byte(e)
 	offset++
@@ -390,28 +390,28 @@ func uintColumnFinish(
 
 	switch e.Width() {
 	case 1:
-		dest := makeUnsafeRawSlice[uint8](unsafe.Pointer(&buf[offset])).Slice(len(values))
+		dest := makeUnsafeRawSlice[uint8](unsafe.Pointer(&buf[offset])).Slice(rows)
 		reduceUints(deltaBase, values, dest)
 
 	case 2:
-		dest := makeUnsafeRawSlice[uint16](unsafe.Pointer(&buf[offset])).Slice(len(values))
+		dest := makeUnsafeRawSlice[uint16](unsafe.Pointer(&buf[offset])).Slice(rows)
 		reduceUints(deltaBase, values, dest)
 
 	case 4:
-		dest := makeUnsafeRawSlice[uint32](unsafe.Pointer(&buf[offset])).Slice(len(values))
+		dest := makeUnsafeRawSlice[uint32](unsafe.Pointer(&buf[offset])).Slice(rows)
 		reduceUints(deltaBase, values, dest)
 
 	case 8:
 		if deltaBase != 0 {
 			panic("unreachable")
 		}
-		dest := makeUnsafeRawSlice[uint64](unsafe.Pointer(&buf[offset])).Slice(len(values))
+		dest := makeUnsafeRawSlice[uint64](unsafe.Pointer(&buf[offset])).Slice(rows)
 		copy(dest, values)
 
 	default:
 		panic("unreachable")
 	}
-	return offset + uint32(len(values))*width
+	return offset + uint32(rows)*width
 }
 
 // WriteDebug implements Encoder.
@@ -425,17 +425,28 @@ func (b *UintBuilder) WriteDebug(w io.Writer, rows int) {
 //	reduceUints[uint8](10, []uint64{10, 11, 12}, dst)
 //
 // could be used to reduce a slice of uint64 values to uint8 values {0, 1, 2}.
+//
+// The values slice can be smaller than dst; in that case, the values between
+// len(values) and len(dst) are assumed to be 0.
 func reduceUints[N constraints.Integer](minimum uint64, values []uint64, dst []N) {
-	for i := 0; i < len(values); i++ {
+	for i, v := range values {
 		if invariants.Enabled {
-			if values[i] < minimum {
+			if v < minimum {
 				panic("incorrect minimum value")
 			}
-			if values[i]-minimum > uint64(N(0)-1) {
+			if v-minimum > uint64(N(0)-1) {
 				panic("incorrect target width")
 			}
 		}
-		dst[i] = N(values[i] - minimum)
+		dst[i] = N(v - minimum)
+	}
+	if invariants.Enabled && len(values) < len(dst) {
+		if minimum != 0 {
+			panic("incorrect minimum value")
+		}
+	}
+	for i := len(values); i < len(dst); i++ {
+		dst[i] = 0
 	}
 }
 
diff --git a/sstable/comparer.go b/sstable/comparer.go
index 62b52730..66a20b51 100644
--- a/sstable/comparer.go
+++ b/sstable/comparer.go
@@ -9,9 +9,6 @@ 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/copier.go b/sstable/copier.go
index 967874be..9538ad84 100644
--- a/sstable/copier.go
+++ b/sstable/copier.go
@@ -246,7 +246,7 @@ func intersectingIndexEntries(
 				if err != nil {
 					return nil, err
 				}
-				entry := indexEntry{bh: bh, sep: top.Separator()}
+				entry := indexEntry{bh: bh, sep: sub.Separator()}
 				alloc, entry.bh.Props = alloc.Copy(entry.bh.Props)
 				alloc, entry.sep = alloc.Copy(entry.sep)
 				res = append(res, entry)
diff --git a/sstable/layout.go b/sstable/layout.go
index 98be8a88..72376a09 100644
--- a/sstable/layout.go
+++ b/sstable/layout.go
@@ -321,7 +321,7 @@ var (
 )
 
 func formatColblkIndexBlock(tp treeprinter.Node, r *Reader, b NamedBlockHandle, data []byte) error {
-	iter := new(colblk.IndexIter)
+	var iter colblk.IndexIter
 	if err := iter.Init(r.Compare, r.Split, data, NoTransforms); err != nil {
 		return err
 	}
diff --git a/sstable/reader.go b/sstable/reader.go
index 132ea82b..eafdcee1 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -57,11 +57,10 @@ type Reader struct {
 	filterMetricsTracker *FilterMetricsTracker
 	logger               base.LoggerAndTracer
 
-	Comparer  *base.Comparer
-	Compare   Compare
-	SuffixCmp CompareSuffixes
-	Equal     Equal
-	Split     Split
+	Comparer *base.Comparer
+	Compare  Compare
+	Equal    Equal
+	Split    Split
 
 	tableFilter *tableFilterReader
 
@@ -289,7 +288,7 @@ func (r *Reader) NewRawRangeDelIter(
 	if r.tableFormat.BlockColumnar() {
 		iter = colblk.NewKeyspanIter(r.Compare, h, transforms)
 	} else {
-		iter, err = rowblk.NewFragmentIter(r.cacheOpts.FileNum, r.Compare, r.Comparer.CompareSuffixes, r.Split, h, transforms)
+		iter, err = rowblk.NewFragmentIter(r.cacheOpts.FileNum, r.Compare, r.Comparer.CompareRangeSuffixes, r.Split, h, transforms)
 		if err != nil {
 			return nil, err
 		}
@@ -314,7 +313,7 @@ func (r *Reader) NewRawRangeKeyIter(
 	if r.tableFormat.BlockColumnar() {
 		iter = colblk.NewKeyspanIter(r.Compare, h, transforms)
 	} else {
-		iter, err = rowblk.NewFragmentIter(r.cacheOpts.FileNum, r.Compare, r.Comparer.CompareSuffixes, r.Split, h, transforms)
+		iter, err = rowblk.NewFragmentIter(r.cacheOpts.FileNum, r.Compare, r.Comparer.CompareRangeSuffixes, r.Split, h, transforms)
 		if err != nil {
 			return nil, err
 		}
@@ -393,7 +392,16 @@ func (r *Reader) readIndexBlock(
 	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle, bh block.Handle,
 ) (block.BufferHandle, error) {
 	ctx = objiotracing.WithBlockType(ctx, objiotracing.MetadataBlock)
-	return r.readBlockInternal(ctx, env, readHandle, bh, noInitBlockMetadataFn)
+	return r.readBlockInternal(ctx, env, readHandle, bh, r.initIndexBlockMetadata)
+}
+
+// initIndexBlockMetadata initializes the Metadata for a data block. This will
+// later be used (and reused) when reading from the block.
+func (r *Reader) initIndexBlockMetadata(metadata *block.Metadata, data []byte) error {
+	if r.tableFormat.BlockColumnar() {
+		return colblk.InitIndexBlockMetadata(metadata, data)
+	}
+	return nil
 }
 
 func (r *Reader) readDataBlock(
@@ -507,9 +515,9 @@ func (r *Reader) readBlockInternal(
 	readStopwatch := makeStopwatch()
 	var err error
 	if readHandle != nil {
-		err = readHandle.ReadAt(ctx, compressed.Get(), int64(bh.Offset))
+		err = readHandle.ReadAt(ctx, compressed.BlockData(), int64(bh.Offset))
 	} else {
-		err = r.readable.ReadAt(ctx, compressed.Get(), int64(bh.Offset))
+		err = r.readable.ReadAt(ctx, compressed.BlockData(), int64(bh.Offset))
 	}
 	readDuration := readStopwatch.stop()
 	// Call IsTracingEnabled to avoid the allocations of boxing integers into an
@@ -528,12 +536,12 @@ func (r *Reader) readBlockInternal(
 		return block.BufferHandle{}, err
 	}
 	env.BlockRead(bh.Length, readDuration)
-	if err := checkChecksum(r.checksumType, compressed.Get(), bh, r.cacheOpts.FileNum); err != nil {
+	if err := checkChecksum(r.checksumType, compressed.BlockData(), bh, r.cacheOpts.FileNum); err != nil {
 		compressed.Release()
 		return block.BufferHandle{}, err
 	}
 
-	typ := block.CompressionIndicator(compressed.Get()[bh.Length])
+	typ := block.CompressionIndicator(compressed.BlockData()[bh.Length])
 	compressed.Truncate(int(bh.Length))
 
 	var decompressed block.Value
@@ -541,21 +549,21 @@ func (r *Reader) readBlockInternal(
 		decompressed = compressed
 	} else {
 		// Decode the length of the decompressed value.
-		decodedLen, prefixLen, err := block.DecompressedLen(typ, compressed.Get())
+		decodedLen, prefixLen, err := block.DecompressedLen(typ, compressed.BlockData())
 		if err != nil {
 			compressed.Release()
 			return block.BufferHandle{}, err
 		}
 
 		decompressed = block.Alloc(decodedLen, env.BufferPool)
-		err = block.DecompressInto(typ, compressed.Get()[prefixLen:], decompressed.Get())
+		err = block.DecompressInto(typ, compressed.BlockData()[prefixLen:], decompressed.BlockData())
 		compressed.Release()
 		if err != nil {
 			decompressed.Release()
 			return block.BufferHandle{}, err
 		}
 	}
-	if err := initBlockMetadataFn(decompressed.GetMetadata(), decompressed.Get()); err != nil {
+	if err := initBlockMetadataFn(decompressed.BlockMetadata(), decompressed.BlockData()); err != nil {
 		decompressed.Release()
 		return block.BufferHandle{}, err
 	}
@@ -1018,13 +1026,11 @@ func NewReader(ctx context.Context, f objstorage.Readable, o ReaderOptions) (*Re
 	if r.Properties.ComparerName == "" || o.Comparer.Name == r.Properties.ComparerName {
 		r.Comparer = o.Comparer
 		r.Compare = o.Comparer.Compare
-		r.SuffixCmp = o.Comparer.CompareSuffixes
 		r.Equal = o.Comparer.Equal
 		r.Split = o.Comparer.Split
 	} else if comparer, ok := o.Comparers[r.Properties.ComparerName]; ok {
 		r.Comparer = o.Comparer
 		r.Compare = comparer.Compare
-		r.SuffixCmp = comparer.CompareSuffixes
 		r.Equal = comparer.Equal
 		r.Split = comparer.Split
 	} else {
diff --git a/sstable/reader_virtual.go b/sstable/reader_virtual.go
index 5ab2a245..dfa84d2a 100644
--- a/sstable/reader_virtual.go
+++ b/sstable/reader_virtual.go
@@ -192,7 +192,7 @@ func (v *VirtualReader) NewRawRangeKeyIter(
 		transformIter := &keyspan.TransformerIter{
 			FragmentIterator: iter,
 			Transformer:      transform,
-			SuffixCmp:        v.reader.SuffixCmp,
+			SuffixCmp:        v.reader.Comparer.CompareRangeSuffixes,
 		}
 		iter = transformIter
 	}
diff --git a/sstable/rowblk/rowblk_fragment_iter.go b/sstable/rowblk/rowblk_fragment_iter.go
index 4aeb0a6f..3129e1f3 100644
--- a/sstable/rowblk/rowblk_fragment_iter.go
+++ b/sstable/rowblk/rowblk_fragment_iter.go
@@ -39,7 +39,7 @@ import (
 // byte slices (start, end, suffix, value) as stable for the lifetime of the
 // iterator.
 type fragmentIter struct {
-	suffixCmp base.CompareSuffixes
+	suffixCmp base.CompareRangeSuffixes
 	blockIter Iter
 	keyBuf    [2]keyspan.Key
 	span      keyspan.Span
@@ -76,7 +76,7 @@ var fragmentBlockIterPool = sync.Pool{
 func NewFragmentIter(
 	fileNum base.DiskFileNum,
 	cmp base.Compare,
-	suffixCmp base.CompareSuffixes,
+	suffixCmp base.CompareRangeSuffixes,
 	split base.Split,
 	blockHandle block.BufferHandle,
 	transforms block.FragmentIterTransforms,
diff --git a/sstable/rowblk_writer.go b/sstable/rowblk_writer.go
index 57a9e549..92785ca1 100644
--- a/sstable/rowblk_writer.go
+++ b/sstable/rowblk_writer.go
@@ -43,7 +43,7 @@ type RawRowWriter struct {
 	indexFlush block.FlushGovernor
 	// The following fields are copied from Options.
 	compare              Compare
-	suffixCmp            CompareSuffixes
+	pointSuffixCmp       base.ComparePointSuffixes
 	split                Split
 	formatKey            base.FormatKey
 	compression          block.Compression
@@ -662,7 +662,7 @@ func (w *RawRowWriter) makeAddPointDecisionV3(
 		cmpUser = cmpPrefix
 		if cmpPrefix == 0 {
 			// Need to compare suffixes to compute cmpUser.
-			cmpUser = w.suffixCmp(prevPointUserKey[prevPointKeyInfo.prefixLen:],
+			cmpUser = w.pointSuffixCmp(prevPointUserKey[prevPointKeyInfo.prefixLen:],
 				key.UserKey[w.lastPointKeyInfo.prefixLen:])
 		}
 	} else {
@@ -1707,7 +1707,7 @@ func newRowWriter(writable objstorage.Writable, o WriterOptions) *RawRowWriter {
 			SmallestSeqNum: math.MaxUint64,
 		},
 		compare:                    o.Comparer.Compare,
-		suffixCmp:                  o.Comparer.CompareSuffixes,
+		pointSuffixCmp:             o.Comparer.ComparePointSuffixes,
 		split:                      o.Comparer.Split,
 		formatKey:                  o.Comparer.FormatKey,
 		compression:                o.Compression,
diff --git a/sstable/writer.go b/sstable/writer.go
index a6b81019..6ff28b8a 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -71,7 +71,7 @@ func (w *Writer) encodeFragmentedRangeKeySpan(s keyspan.Span) {
 	//
 	// Sort the keys by suffix. Iteration doesn't *currently* depend on it, but
 	// we may want to in the future.
-	keyspan.SortKeysBySuffix(w.comparer.CompareSuffixes, s.Keys)
+	keyspan.SortKeysBySuffix(w.comparer.CompareRangeSuffixes, s.Keys)
 
 	if w.Error() == nil {
 		w.rw.EncodeSpan(s)
diff --git a/table_stats.go b/table_stats.go
index 410f0419..e951d9bb 100644
--- a/table_stats.go
+++ b/table_stats.go
@@ -853,7 +853,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.CompareSuffixes, a, b *keyspan.Span) bool { return true })
+	var equal = keyspan.DefragmentMethodFunc(func(_ base.CompareRangeSuffixes, 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.
@@ -882,7 +882,7 @@ func newCombinedDeletionKeyspanIter(
 				smallest = last
 			}
 		}
-		if largest.Equal(comparer.CompareSuffixes, smallest) {
+		if largest.Equal(comparer.CompareRangeSuffixes, smallest) {
 			current = append(current[:0], largest)
 		} else {
 			current = append(current[:0], largest, smallest)
@@ -894,7 +894,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(_ base.CompareSuffixes, in keyspan.Span, out *keyspan.Span) error {
+	var transform = keyspan.TransformerFunc(func(_ base.CompareRangeSuffixes, in keyspan.Span, out *keyspan.Span) error {
 		if in.KeysOrder != keyspan.ByTrailerDesc {
 			panic("pebble: combined deletion iter encountered keys in non-trailer descending order")
 		}


Test patch:
diff --git a/internal/compact/spans_test.go b/internal/compact/spans_test.go
index c881a2a3..bc9f71c3 100644
--- a/internal/compact/spans_test.go
+++ b/internal/compact/spans_test.go
@@ -73,7 +73,7 @@ func TestRangeKeySpanCompactor(t *testing.T) {
 
 			c = MakeRangeKeySpanCompactor(
 				base.DefaultComparer.Compare,
-				base.DefaultComparer.CompareSuffixes,
+				base.DefaultComparer.CompareRangeSuffixes,
 				s,
 				ElideTombstonesOutsideOf(keyRanges),
 			)
diff --git a/internal/crdbtest/crdbtest.go b/internal/crdbtest/crdbtest.go
index 216a02ef..c3b7d520 100644
--- a/internal/crdbtest/crdbtest.go
+++ b/internal/crdbtest/crdbtest.go
@@ -26,20 +26,40 @@ import (
 	"github.com/cockroachdb/pebble/sstable/colblk"
 )
 
-const withWall = 9
-const withLogical = withWall + 4
-const withSynthetic = withLogical + 1
-const withLockTableLen = 18
+const (
+	engineKeyNoVersion                             = 0
+	engineKeyVersionWallTimeLen                    = 8
+	engineKeyVersionWallAndLogicalTimeLen          = 12
+	engineKeyVersionWallLogicalAndSyntheticTimeLen = 13
+	engineKeyVersionLockTableLen                   = 17
+
+	mvccEncodedTimeSentinelLen  = 1
+	mvccEncodedTimeWallLen      = 8
+	mvccEncodedTimeLogicalLen   = 4
+	mvccEncodedTimeSyntheticLen = 1
+	mvccEncodedTimeLengthLen    = 1
+
+	suffixLenWithWall      = engineKeyVersionWallTimeLen + 1
+	suffixLenWithLogical   = engineKeyVersionWallAndLogicalTimeLen + 1
+	suffixLenWithSynthetic = engineKeyVersionWallLogicalAndSyntheticTimeLen + 1
+	suffixLenWithLockTable = engineKeyVersionLockTableLen + 1
+)
 
 // MaxSuffixLen is the maximum length of the CockroachDB key suffix.
-const MaxSuffixLen = max(withLockTableLen, withSynthetic, withLogical, withWall)
+const MaxSuffixLen = 1 + max(
+	engineKeyVersionLockTableLen,
+	engineKeyVersionWallLogicalAndSyntheticTimeLen,
+	engineKeyVersionWallAndLogicalTimeLen,
+	engineKeyVersionWallTimeLen,
+)
 
 // Comparer is a base.Comparer for CockroachDB keys.
 var Comparer = base.Comparer{
-	Split:           Split,
-	CompareSuffixes: CompareSuffixes,
-	Compare:         Compare,
-	Equal:           Equal,
+	Split:                Split,
+	ComparePointSuffixes: ComparePointSuffixes,
+	CompareRangeSuffixes: CompareRangeSuffixes,
+	Compare:              Compare,
+	Equal:                Equal,
 	AbbreviatedKey: func(k []byte) uint64 {
 		key, ok := getKeyPartFromEngineKey(k)
 		if !ok {
@@ -103,8 +123,8 @@ var Comparer = base.Comparer{
 
 // 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 {
-		dst = make([]byte, 0, len(key)+withSynthetic)
+	if cap(dst) < len(key)+suffixLenWithSynthetic {
+		dst = make([]byte, 0, len(key)+suffixLenWithSynthetic)
 	}
 	dst = append(dst[:0], key...)
 	return EncodeTimestamp(dst, walltime, logical)
@@ -237,44 +257,28 @@ func Compare(a, b []byte) int {
 	if len(a) == 0 || len(b) == 0 {
 		return cmp.Compare(len(a), len(b))
 	}
+	aSuffixLen := int(a[len(a)-1])
+	aSuffixStart := len(a) - aSuffixLen
+	bSuffixLen := int(b[len(b)-1])
+	bSuffixStart := len(b) - bSuffixLen
 
-	// NB: For performance, this routine manually splits the key into the
-	// user-key and version components rather than using DecodeEngineKey. In
-	// most situations, use DecodeEngineKey or GetKeyPartFromEngineKey or
-	// SplitMVCCKey instead of doing this.
-	aEnd := len(a) - 1
-	bEnd := len(b) - 1
-
-	// Compute the index of the separator between the key and the version. If the
-	// separator is found to be at -1 for both keys, then we are comparing bare
-	// suffixes without a user key part. Pebble requires bare suffixes to be
-	// 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 < 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 {
+	if c := bytes.Compare(a[:aSuffixStart], b[:bSuffixStart]); c != 0 {
 		return c
 	}
-
-	// 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.
-	a, b = a[aSep+1:], b[bSep+1:]
-	if len(a) == 0 || len(b) == 0 {
+	if aSuffixLen == 0 || bSuffixLen == 0 {
 		// Empty suffixes come before non-empty suffixes.
-		return cmp.Compare(len(a), len(b))
+		return cmp.Compare(aSuffixLen, bSuffixLen)
 	}
+
 	return bytes.Compare(
-		normalizeEngineKeyVersionForCompare(b),
-		normalizeEngineKeyVersionForCompare(a),
+		normalizeEngineSuffixForCompare(b[bSuffixStart:]),
+		normalizeEngineSuffixForCompare(a[aSuffixStart:]),
 	)
 }
 
-// CompareSuffixes compares suffixes (normally timestamps).
-func CompareSuffixes(a, b []byte) int {
+// CompareRangeSuffixes compares range key suffixes (normally timestamps).
+func CompareRangeSuffixes(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))
@@ -285,78 +289,91 @@ func CompareSuffixes(a, b []byte) int {
 	return bytes.Compare(b[:len(b)-1], a[:len(a)-1])
 }
 
+// ComparePointSuffixes compares point key suffixes (normally timestamps).
+func ComparePointSuffixes(a, b []byte) int {
+	if len(a) == 0 || len(b) == 0 {
+		// Empty suffixes sort before non-empty suffixes.
+		return cmp.Compare(len(a), len(b))
+	}
+	return bytes.Compare(normalizeEngineSuffixForCompare(b), normalizeEngineSuffixForCompare(a))
+}
+
 // Equal implements base.Equal for Cockroach keys.
 func Equal(a, b []byte) bool {
-	// TODO(radu): Pebble sometimes passes empty "keys" and we have to tolerate
-	// them until we fix that.
 	if len(a) == 0 || len(b) == 0 {
 		return len(a) == len(b)
 	}
-	aEnd := len(a) - 1
-	bEnd := len(b) - 1
-
-	// Last byte is the version length + 1 when there is a version,
-	// else it is 0.
-	aVerLen := int(a[aEnd])
-	bVerLen := int(b[bEnd])
-
-	// Fast-path. If the key version is empty or contains only a walltime
-	// component then normalizeEngineKeyVersionForCompare is a no-op, so we don't
-	// need to split the "user key" from the version suffix before comparing to
-	// compute equality. Instead, we can check for byte equality immediately.
-	if (aVerLen <= withWall && bVerLen <= withWall) || (aVerLen == withLockTableLen && bVerLen == withLockTableLen) {
+	aSuffixLen := int(a[len(a)-1])
+	aSuffixStart := len(a) - aSuffixLen
+	bSuffixLen := int(b[len(b)-1])
+	bSuffixStart := len(b) - bSuffixLen
+
+	// Fast-path: normalizeEngineSuffixForCompare doesn't strip off bytes when the
+	// length is withWall or withLockTableLen. In this case, as well as cases with
+	// no prefix, we can check for byte equality immediately.
+	const withWall = mvccEncodedTimeSentinelLen + mvccEncodedTimeWallLen
+	const withLockTableLen = mvccEncodedTimeSentinelLen + engineKeyVersionLockTableLen
+	if (aSuffixLen <= withWall && bSuffixLen <= withWall) ||
+		(aSuffixLen == withLockTableLen && bSuffixLen == withLockTableLen) ||
+		aSuffixLen == 0 || bSuffixLen == 0 {
 		return bytes.Equal(a, b)
 	}
 
-	// Compute the index of the separator between the key and the version. If the
-	// separator is found to be at -1 for both keys, then we are comparing bare
-	// suffixes without a user key part. Pebble requires bare suffixes to be
-	// comparable with the same ordering as if they had a common user key.
-	aSep := aEnd - aVerLen
-	bSep := bEnd - bVerLen
 	// Compare the "user key" part of the key.
-	if !bytes.Equal(a[:aSep], b[:bSep]) {
+	if !bytes.Equal(a[:aSuffixStart], b[:bSuffixStart]) {
 		return false
 	}
-	if aVerLen == 0 || bVerLen == 0 {
-		return aVerLen == bVerLen
-	}
 
-	// Compare the version part of the key.
-	aVer := a[aSep+1:]
-	bVer := b[bSep+1:]
-	aVer = normalizeEngineKeyVersionForCompare(aVer)
-	bVer = normalizeEngineKeyVersionForCompare(bVer)
-	return bytes.Equal(aVer, bVer)
+	return bytes.Equal(
+		normalizeEngineSuffixForCompare(a[aSuffixStart:]),
+		normalizeEngineSuffixForCompare(b[bSuffixStart:]),
+	)
 }
 
 var zeroLogical [4]byte
 
-func normalizeEngineKeyVersionForCompare(a []byte) []byte {
+// normalizeEngineSuffixForCompare takes a non-empty key suffix (including the
+// trailing sentinel byte) and returns a prefix of the buffer that should be
+// used for byte-wise comparison. It trims the trailing suffix length byte and
+// any other trailing bytes that need to be ignored (like a synthetic bit or
+// zero logical component).
+//
+//gcassert:inline
+func normalizeEngineSuffixForCompare(a []byte) []byte {
 	// Check sentinel byte.
-	if len(a) != int(a[len(a)-1]) {
+	if invariants.Enabled && 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-1 {
+	switch len(a) {
+	case engineKeyVersionWallLogicalAndSyntheticTimeLen:
 		// 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-1]
-	}
-	if len(a) == withLogical-1 {
+		a = a[:engineKeyVersionWallAndLogicalTimeLen]
+		fallthrough
+	case engineKeyVersionWallAndLogicalTimeLen:
 		// 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-1:], zeroLogical[:]) {
-			a = a[:withWall-1]
+		if bytes.Equal(a[engineKeyVersionWallTimeLen:], zeroLogical[:]) {
+			a = a[:engineKeyVersionWallTimeLen]
+		}
+		fallthrough
+	case engineKeyVersionWallTimeLen:
+		// Nothing to do.
+
+	case engineKeyVersionLockTableLen:
+		// We rely on engineKeyVersionLockTableLen being different from the other
+		// lengths above to ensure that we don't strip parts from a non-timestamp
+		// version.
+
+	default:
+		if invariants.Enabled {
+			panic(errors.AssertionFailedf("version with unexpected length: %x", a))
 		}
 	}
 	return a
@@ -541,7 +558,7 @@ func (kw *cockroachKeyWriter) ComparePrev(key []byte) colblk.KeyComparison {
 	if cmpv.CommonPrefixLen == cmpv.PrefixLen-1 {
 		// Adjust CommonPrefixLen to include the sentinel byte,
 		cmpv.CommonPrefixLen = cmpv.PrefixLen
-		cmpv.UserKeyComparison = int32(CompareSuffixes(key[cmpv.PrefixLen:], kw.prevSuffix))
+		cmpv.UserKeyComparison = int32(CompareRangeSuffixes(key[cmpv.PrefixLen:], kw.prevSuffix))
 		return cmpv
 	}
 	// The keys have different MVCC prefixes. We haven't determined which is
diff --git a/internal/crdbtest/crdbtest_test.go b/internal/crdbtest/crdbtest_test.go
index 97dbeda8..52b4ca69 100644
--- a/internal/crdbtest/crdbtest_test.go
+++ b/internal/crdbtest/crdbtest_test.go
@@ -38,11 +38,11 @@ func TestComparer(t *testing.T) {
 			suffix := key[Comparer.Split(key):]
 			suffixes = append(suffixes, suffix)
 
-			if len(suffix) == withWall {
+			if len(suffix) == suffixLenWithWall {
 				// Append a suffix that encodes a zero logical value that should be
 				// ignored in key comparisons, but not suffix comparisons.
-				newSuffix := slices.Concat(suffix[:withWall-1], zeroLogical[:], []byte{withLogical})
-				if Comparer.CompareSuffixes(suffix, newSuffix) != 1 {
+				newSuffix := slices.Concat(suffix[:suffixLenWithWall-1], zeroLogical[:], []byte{suffixLenWithLogical})
+				if Comparer.CompareRangeSuffixes(suffix, newSuffix) != 1 {
 					t.Fatalf("expected suffixes %x < %x", suffix, newSuffix)
 				}
 				if Comparer.Compare(slices.Concat(prefixes[0], suffix), slices.Concat(prefixes[0], newSuffix)) != 0 {
@@ -51,13 +51,13 @@ func TestComparer(t *testing.T) {
 				suffixes = append(suffixes, newSuffix)
 				suffix = newSuffix
 			}
-			if len(suffix) != withLogical {
+			if len(suffix) != suffixLenWithLogical {
 				t.Fatalf("unexpected suffix %x", suffix)
 			}
 			// Append a synthetic bit that should be ignored in key comparisons, but
 			// not suffix comparisons.
-			newSuffix := slices.Concat(suffix[:withLogical-1], []byte{1}, []byte{withSynthetic})
-			if Comparer.CompareSuffixes(suffix, newSuffix) != 1 {
+			newSuffix := slices.Concat(suffix[:suffixLenWithLogical-1], []byte{1}, []byte{suffixLenWithSynthetic})
+			if Comparer.CompareRangeSuffixes(suffix, newSuffix) != 1 {
 				t.Fatalf("expected suffixes %x < %x", suffix, newSuffix)
 			}
 			if Comparer.Compare(slices.Concat(prefixes[0], suffix), slices.Concat(prefixes[0], newSuffix)) != 0 {
@@ -67,8 +67,8 @@ func TestComparer(t *testing.T) {
 		}
 	}
 	// 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))
+	suffixes = append(suffixes, append(bytes.Repeat([]byte{1}, engineKeyVersionLockTableLen), suffixLenWithLockTable))
+	suffixes = append(suffixes, append(bytes.Repeat([]byte{2}, engineKeyVersionLockTableLen), suffixLenWithLockTable))
 	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 ab16ca10..45ba05c4 100644
--- a/internal/keyspan/defragment_test.go
+++ b/internal/keyspan/defragment_test.go
@@ -22,7 +22,7 @@ func TestDefragmentingIter(t *testing.T) {
 	comparer := testkeys.Comparer
 	cmp := comparer.Compare
 	internalEqual := DefragmentInternal
-	alwaysEqual := DefragmentMethodFunc(func(_ base.CompareSuffixes, _, _ *Span) bool { return true })
+	alwaysEqual := DefragmentMethodFunc(func(_ base.CompareRangeSuffixes, _, _ *Span) bool { return true })
 	staticReducer := StaticDefragmentReducer
 	collectReducer := func(cur, next []Key) []Key {
 		c := append(cur, next...)
diff --git a/internal/keyspan/interleaving_iter_test.go b/internal/keyspan/interleaving_iter_test.go
index 143404cd..45f331fb 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
-	suffixCmp  base.CompareSuffixes
+	suffixCmp  base.CompareRangeSuffixes
 	split      base.Split
 	threshold  []byte
 	maskSuffix []byte
@@ -77,7 +77,7 @@ func runInterleavingIterTest(t *testing.T, filename string) {
 	var buf bytes.Buffer
 	hooks := maskingHooks{
 		log:       &buf,
-		suffixCmp: testkeys.Comparer.CompareSuffixes,
+		suffixCmp: testkeys.Comparer.CompareRangeSuffixes,
 		split:     testkeys.Comparer.Split,
 	}
 
diff --git a/internal/rangekey/coalesce_test.go b/internal/rangekey/coalesce_test.go
index b76a6b45..285e4f73 100644
--- a/internal/rangekey/coalesce_test.go
+++ b/internal/rangekey/coalesce_test.go
@@ -22,7 +22,7 @@ func TestCoalesce(t *testing.T) {
 				Start: span.Start,
 				End:   span.End,
 			}
-			Coalesce(testkeys.Comparer.CompareSuffixes, span.Keys, &coalesced.Keys)
+			Coalesce(testkeys.Comparer.CompareRangeSuffixes, span.Keys, &coalesced.Keys)
 			return coalesced.String()
 
 		default:
diff --git a/internal/rangekeystack/user_iterator_test.go b/internal/rangekeystack/user_iterator_test.go
index e1b14291..71e44b03 100644
--- a/internal/rangekeystack/user_iterator_test.go
+++ b/internal/rangekeystack/user_iterator_test.go
@@ -45,7 +45,7 @@ func TestIter(t *testing.T) {
 			for _, line := range lines {
 				spans = append(spans, keyspan.ParseSpan(line))
 			}
-			transform := keyspan.TransformerFunc(func(suffixCmp base.CompareSuffixes, s keyspan.Span, dst *keyspan.Span) error {
+			transform := keyspan.TransformerFunc(func(suffixCmp base.CompareRangeSuffixes, 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
@@ -375,7 +375,7 @@ func BenchmarkTransform(b *testing.B) {
 					b.ResetTimer()
 
 					for i := 0; i < b.N; i++ {
-						err := ui.Transform(testkeys.Comparer.CompareSuffixes, keyspan.Span{Keys: keys}, &dst)
+						err := ui.Transform(testkeys.Comparer.CompareRangeSuffixes, keyspan.Span{Keys: keys}, &dst)
 						if err != nil {
 							b.Fatal(err)
 						}
diff --git a/internal/testkeys/testkeys.go b/internal/testkeys/testkeys.go
index d85c87d8..35eb6144 100644
--- a/internal/testkeys/testkeys.go
+++ b/internal/testkeys/testkeys.go
@@ -49,9 +49,10 @@ 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{
-	CompareSuffixes: compareSuffixes,
-	Compare:         compare,
-	Equal:           func(a, b []byte) bool { return compare(a, b) == 0 },
+	ComparePointSuffixes: compareSuffixes,
+	CompareRangeSuffixes: compareSuffixes,
+	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)])
 	},
diff --git a/internal/testkeys/testkeys_test.go b/internal/testkeys/testkeys_test.go
index 34a962c6..5616410a 100644
--- a/internal/testkeys/testkeys_test.go
+++ b/internal/testkeys/testkeys_test.go
@@ -163,7 +163,7 @@ func TestSuffix(t *testing.T) {
 	for ts := int64(2); ts < 150; ts++ {
 		an := WriteSuffix(a, ts-1)
 		bn := WriteSuffix(b, ts)
-		got := Comparer.CompareSuffixes(a[:an], b[:bn])
+		got := Comparer.CompareRangeSuffixes(a[:an], b[:bn])
 		if want := +1; got != want {
 			t.Errorf("CompareSuffixes(%q, %q) = %d, want %d", a, b, got, want)
 		}
@@ -311,8 +311,8 @@ func TestIgnorableSuffix(t *testing.T) {
 	require.Equal(t, 0, Comparer.Compare([]byte("foo@1"), []byte("foo@1_synthetic")))
 	require.Equal(t, 1, Comparer.Compare([]byte("foo@1"), []byte("foo@2_synthetic")))
 	require.Equal(t, 1, Comparer.Compare([]byte("foo@1_synthetic"), []byte("foo@2")))
-	require.Equal(t, -1, Comparer.CompareSuffixes([]byte("@1"), []byte("@1_synthetic")))
-	require.Equal(t, 1, Comparer.CompareSuffixes([]byte("@1_synthetic"), []byte("@1")))
-	require.Equal(t, 0, Comparer.CompareSuffixes([]byte("@1_synthetic"), []byte("@1_synthetic")))
-	require.Equal(t, 0, Comparer.CompareSuffixes([]byte("@1"), []byte("@1")))
+	require.Equal(t, -1, Comparer.CompareRangeSuffixes([]byte("@1"), []byte("@1_synthetic")))
+	require.Equal(t, 1, Comparer.CompareRangeSuffixes([]byte("@1_synthetic"), []byte("@1")))
+	require.Equal(t, 0, Comparer.CompareRangeSuffixes([]byte("@1_synthetic"), []byte("@1_synthetic")))
+	require.Equal(t, 0, Comparer.CompareRangeSuffixes([]byte("@1"), []byte("@1")))
 }
diff --git a/sstable/colblk/index_block_test.go b/sstable/colblk/index_block_test.go
index 6c1bad3a..9435f7e3 100644
--- a/sstable/colblk/index_block_test.go
+++ b/sstable/colblk/index_block_test.go
@@ -11,6 +11,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"unsafe"
 
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -115,12 +116,16 @@ func TestIndexIterInitHandle(t *testing.T) {
 	bh2 := block.Handle{Offset: 2008, Length: 1000}
 	w.AddBlockHandle([]byte("a"), bh1, nil)
 	w.AddBlockHandle([]byte("b"), bh2, nil)
-	b := w.Finish(w.Rows())
+	blockData := w.Finish(w.Rows())
 
 	c := cache.New(10 << 10)
 	defer c.Unref()
-	v := block.Alloc(len(b), nil)
-	copy(v.Get(), b)
+
+	v := block.Alloc(block.MetadataSize+len(blockData), nil)
+	copy(v.BlockData(), blockData)
+	d := (*IndexBlockDecoder)(unsafe.Pointer(v.BlockMetadata()))
+	d.Init(blockData)
+
 	v.MakeHandle(c, cache.ID(1), base.DiskFileNum(1), 0).Release()
 
 	getBlockAndIterate := func(it *IndexIter) {
diff --git a/sstable/colblk/keyspan_test.go b/sstable/colblk/keyspan_test.go
index 1cb0016b..b938d705 100644
--- a/sstable/colblk/keyspan_test.go
+++ b/sstable/colblk/keyspan_test.go
@@ -85,7 +85,7 @@ func TestKeyspanBlockPooling(t *testing.T) {
 	c := cache.New(10 << 10)
 	defer c.Unref()
 	v := block.Alloc(len(b), nil)
-	copy(v.Get(), b)
+	copy(v.BlockData(), b)
 	v.MakeHandle(c, cache.ID(1), base.DiskFileNum(1), 0).Release()
 
 	getBlockAndIterate := func() {
diff --git a/sstable/colblk/testdata/uints b/sstable/colblk/testdata/uints
index f2da68ef..b948ce0a 100644
--- a/sstable/colblk/testdata/uints
+++ b/sstable/colblk/testdata/uints
@@ -580,3 +580,117 @@ uints
  ├── 08-12: x 4a420f00 # data[1] = 1000010
  ├── 12-16: x 54420f00 # data[2] = 1000020
  └── 16-20: x 5e420f00 # data[3] = 1000030
+
+
+init default-zero
+----
+
+write
+0:1 1:1 2:1 3:1 4:1
+----
+
+# Verify the case where the last rows have not been set.
+finish rows=100
+----
+uints
+ ├── 000-001: x 01 # encoding: 1b
+ ├── 001-002: x 01 # data[0] = 1
+ ├── 002-003: x 01 # data[1] = 1
+ ├── 003-004: x 01 # data[2] = 1
+ ├── 004-005: x 01 # data[3] = 1
+ ├── 005-006: x 01 # data[4] = 1
+ ├── 006-007: x 00 # data[5] = 0
+ ├── 007-008: x 00 # data[6] = 0
+ ├── 008-009: x 00 # data[7] = 0
+ ├── 009-010: x 00 # data[8] = 0
+ ├── 010-011: x 00 # data[9] = 0
+ ├── 011-012: x 00 # data[10] = 0
+ ├── 012-013: x 00 # data[11] = 0
+ ├── 013-014: x 00 # data[12] = 0
+ ├── 014-015: x 00 # data[13] = 0
+ ├── 015-016: x 00 # data[14] = 0
+ ├── 016-017: x 00 # data[15] = 0
+ ├── 017-018: x 00 # data[16] = 0
+ ├── 018-019: x 00 # data[17] = 0
+ ├── 019-020: x 00 # data[18] = 0
+ ├── 020-021: x 00 # data[19] = 0
+ ├── 021-022: x 00 # data[20] = 0
+ ├── 022-023: x 00 # data[21] = 0
+ ├── 023-024: x 00 # data[22] = 0
+ ├── 024-025: x 00 # data[23] = 0
+ ├── 025-026: x 00 # data[24] = 0
+ ├── 026-027: x 00 # data[25] = 0
+ ├── 027-028: x 00 # data[26] = 0
+ ├── 028-029: x 00 # data[27] = 0
+ ├── 029-030: x 00 # data[28] = 0
+ ├── 030-031: x 00 # data[29] = 0
+ ├── 031-032: x 00 # data[30] = 0
+ ├── 032-033: x 00 # data[31] = 0
+ ├── 033-034: x 00 # data[32] = 0
+ ├── 034-035: x 00 # data[33] = 0
+ ├── 035-036: x 00 # data[34] = 0
+ ├── 036-037: x 00 # data[35] = 0
+ ├── 037-038: x 00 # data[36] = 0
+ ├── 038-039: x 00 # data[37] = 0
+ ├── 039-040: x 00 # data[38] = 0
+ ├── 040-041: x 00 # data[39] = 0
+ ├── 041-042: x 00 # data[40] = 0
+ ├── 042-043: x 00 # data[41] = 0
+ ├── 043-044: x 00 # data[42] = 0
+ ├── 044-045: x 00 # data[43] = 0
+ ├── 045-046: x 00 # data[44] = 0
+ ├── 046-047: x 00 # data[45] = 0
+ ├── 047-048: x 00 # data[46] = 0
+ ├── 048-049: x 00 # data[47] = 0
+ ├── 049-050: x 00 # data[48] = 0
+ ├── 050-051: x 00 # data[49] = 0
+ ├── 051-052: x 00 # data[50] = 0
+ ├── 052-053: x 00 # data[51] = 0
+ ├── 053-054: x 00 # data[52] = 0
+ ├── 054-055: x 00 # data[53] = 0
+ ├── 055-056: x 00 # data[54] = 0
+ ├── 056-057: x 00 # data[55] = 0
+ ├── 057-058: x 00 # data[56] = 0
+ ├── 058-059: x 00 # data[57] = 0
+ ├── 059-060: x 00 # data[58] = 0
+ ├── 060-061: x 00 # data[59] = 0
+ ├── 061-062: x 00 # data[60] = 0
+ ├── 062-063: x 00 # data[61] = 0
+ ├── 063-064: x 00 # data[62] = 0
+ ├── 064-065: x 00 # data[63] = 0
+ ├── 065-066: x 00 # data[64] = 0
+ ├── 066-067: x 00 # data[65] = 0
+ ├── 067-068: x 00 # data[66] = 0
+ ├── 068-069: x 00 # data[67] = 0
+ ├── 069-070: x 00 # data[68] = 0
+ ├── 070-071: x 00 # data[69] = 0
+ ├── 071-072: x 00 # data[70] = 0
+ ├── 072-073: x 00 # data[71] = 0
+ ├── 073-074: x 00 # data[72] = 0
+ ├── 074-075: x 00 # data[73] = 0
+ ├── 075-076: x 00 # data[74] = 0
+ ├── 076-077: x 00 # data[75] = 0
+ ├── 077-078: x 00 # data[76] = 0
+ ├── 078-079: x 00 # data[77] = 0
+ ├── 079-080: x 00 # data[78] = 0
+ ├── 080-081: x 00 # data[79] = 0
+ ├── 081-082: x 00 # data[80] = 0
+ ├── 082-083: x 00 # data[81] = 0
+ ├── 083-084: x 00 # data[82] = 0
+ ├── 084-085: x 00 # data[83] = 0
+ ├── 085-086: x 00 # data[84] = 0
+ ├── 086-087: x 00 # data[85] = 0
+ ├── 087-088: x 00 # data[86] = 0
+ ├── 088-089: x 00 # data[87] = 0
+ ├── 089-090: x 00 # data[88] = 0
+ ├── 090-091: x 00 # data[89] = 0
+ ├── 091-092: x 00 # data[90] = 0
+ ├── 092-093: x 00 # data[91] = 0
+ ├── 093-094: x 00 # data[92] = 0
+ ├── 094-095: x 00 # data[93] = 0
+ ├── 095-096: x 00 # data[94] = 0
+ ├── 096-097: x 00 # data[95] = 0
+ ├── 097-098: x 00 # data[96] = 0
+ ├── 098-099: x 00 # data[97] = 0
+ ├── 099-100: x 00 # data[98] = 0
+ └── 100-101: x 00 # data[99] = 0
diff --git a/sstable/copier_test.go b/sstable/copier_test.go
index 48304301..2fce0a57 100644
--- a/sstable/copier_test.go
+++ b/sstable/copier_test.go
@@ -7,6 +7,7 @@ package sstable
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -35,6 +36,8 @@ func TestCopySpan(t *testing.T) {
 				return err.Error()
 			}
 			tableFormat := TableFormatMax
+			blockSize := 1
+			var indexBlockSize int
 			for i := range d.CmdArgs[1:] {
 				switch d.CmdArgs[i+1].Key {
 				case "format":
@@ -44,13 +47,26 @@ func TestCopySpan(t *testing.T) {
 					case "pebblev5":
 						tableFormat = TableFormatPebblev5
 					}
+				case "block_size":
+					var err error
+					blockSize, err = strconv.Atoi(d.CmdArgs[i+1].FirstVal(t))
+					if err != nil {
+						return err.Error()
+					}
+				case "index_block_size":
+					var err error
+					indexBlockSize, err = strconv.Atoi(d.CmdArgs[i+1].FirstVal(t))
+					if err != nil {
+						return err.Error()
+					}
 				}
 			}
 			w := NewWriter(objstorageprovider.NewFileWritable(f), WriterOptions{
-				BlockSize:   1,
-				TableFormat: tableFormat,
-				Comparer:    testkeys.Comparer,
-				KeySchema:   keySchema,
+				BlockSize:      blockSize,
+				IndexBlockSize: indexBlockSize,
+				TableFormat:    tableFormat,
+				Comparer:       testkeys.Comparer,
+				KeySchema:      keySchema,
 			})
 			for _, key := range strings.Split(d.Input, "\n") {
 				j := strings.Index(key, ":")
diff --git a/sstable/data_test.go b/sstable/data_test.go
index d1357214..c55bfcd0 100644
--- a/sstable/data_test.go
+++ b/sstable/data_test.go
@@ -295,7 +295,7 @@ func runIterCmd(
 		for kv != nil {
 			k := kv.K.UserKey
 			suffix := k[testkeys.Comparer.Split(k):]
-			if len(suffix) == 0 || testkeys.Comparer.CompareSuffixes(suffix, maskingSuffix) <= 0 {
+			if len(suffix) == 0 || testkeys.Comparer.CompareRangeSuffixes(suffix, maskingSuffix) <= 0 {
 				return kv
 			}
 			if direction > 0 {
diff --git a/sstable/rowblk/rowblk_fragment_iter_test.go b/sstable/rowblk/rowblk_fragment_iter_test.go
index a6092d4f..e7facf3a 100644
--- a/sstable/rowblk/rowblk_fragment_iter_test.go
+++ b/sstable/rowblk/rowblk_fragment_iter_test.go
@@ -97,7 +97,7 @@ func TestBlockFragmentIterator(t *testing.T) {
 
 			blockHandle := block.CacheBufferHandle(c.Get(1, 0, 0))
 			require.True(t, blockHandle.Valid())
-			i, err := NewFragmentIter(0, comparer.Compare, comparer.CompareSuffixes, comparer.Split, blockHandle, transforms)
+			i, err := NewFragmentIter(0, comparer.Compare, comparer.CompareRangeSuffixes, comparer.Split, blockHandle, transforms)
 			defer i.Close()
 			require.NoError(t, err)
 
diff --git a/sstable/testdata/copy_span b/sstable/testdata/copy_span
index f68cea14..099a9b06 100644
--- a/sstable/testdata/copy_span
+++ b/sstable/testdata/copy_span
@@ -35,6 +35,54 @@ a#0,SET: foo
 b#0,SET: bar
 c#0,SET: baz
 
+
+# Try the above with small blocks.
+build test22 format=pebblev4 block_size=1 index_block_size=1
+a.SET.5:foo
+b.SET.3:bar
+c.SET.4:baz
+d.SET.5:foobar
+e.SET.5:foo
+f.SET.5:foo
+g.SET.5:foo
+h.SET.5:foo
+i.SET.5:foo
+j.SET.5:foo
+----
+
+iter test22
+----
+a#0,SET: foo
+b#0,SET: bar
+c#0,SET: baz
+d#0,SET: foobar
+e#0,SET: foo
+f#0,SET: foo
+g#0,SET: foo
+h#0,SET: foo
+i#0,SET: foo
+j#0,SET: foo
+
+copy-span test22 test23 b.SET.10 cc.SET.0
+----
+copied 661 bytes
+
+iter test23
+----
+b#0,SET: bar
+c#0,SET: baz
+d#0,SET: foobar
+
+copy-span test22 test24 a.SET.10 bb.SET.0
+----
+copied 658 bytes
+
+iter test24
+----
+a#0,SET: foo
+b#0,SET: bar
+c#0,SET: baz
+
 # Try the above with columnar blocks.
 
 build test3 format=pebblev5
@@ -70,3 +118,51 @@ iter test5
 a#0,SET: foo
 b#0,SET: bar
 c#0,SET: baz
+
+
+# Try the above with small blocks.
+build test32 format=pebblev5 block_size=1 index_block_size=1
+a.SET.5:foo
+b.SET.3:bar
+c.SET.4:baz
+d.SET.5:foobar
+e.SET.5:foo
+f.SET.5:foo
+g.SET.5:foo
+h.SET.5:foo
+i.SET.5:foo
+j.SET.5:foo
+----
+
+iter test32
+----
+a#0,SET: foo
+b#0,SET: bar
+c#0,SET: baz
+d#0,SET: foobar
+e#0,SET: foo
+f#0,SET: foo
+g#0,SET: foo
+h#0,SET: foo
+i#0,SET: foo
+j#0,SET: foo
+
+copy-span test32 test33 b.SET.10 cc.SET.0
+----
+copied 889 bytes
+
+iter test33
+----
+b#0,SET: bar
+c#0,SET: baz
+d#0,SET: foobar
+
+copy-span test32 test34 a.SET.10 bb.SET.0
+----
+copied 897 bytes
+
+iter test34
+----
+a#0,SET: foo
+b#0,SET: bar
+c#0,SET: baz
diff --git a/testdata/ingest b/testdata/ingest
index bebe1bea..b7217f3f 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: 3 entries (1.0KB)  hit rate: 18.2%
-Table cache: 1 entries (816B)  hit rate: 50.0%
+Table cache: 1 entries (808B)  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 49661c76..d46b8d2b 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -76,7 +76,7 @@ Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 1
 Block cache: 2 entries (700B)  hit rate: 0.0%
-Table cache: 1 entries (816B)  hit rate: 0.0%
+Table cache: 1 entries (808B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -218,7 +218,7 @@ Virtual tables: 0 (0B)
 Local tables size: 595B
 Compression types: snappy: 1
 Block cache: 2 entries (700B)  hit rate: 33.3%
-Table cache: 1 entries (816B)  hit rate: 66.7%
+Table cache: 1 entries (808B)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -496,7 +496,7 @@ Virtual tables: 0 (0B)
 Local tables size: 4.3KB
 Compression types: snappy: 7
 Block cache: 8 entries (2.8KB)  hit rate: 9.1%
-Table cache: 1 entries (816B)  hit rate: 53.8%
+Table cache: 1 entries (808B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -560,7 +560,7 @@ Virtual tables: 0 (0B)
 Local tables size: 6.1KB
 Compression types: snappy: 10
 Block cache: 8 entries (2.8KB)  hit rate: 9.1%
-Table cache: 1 entries (816B)  hit rate: 53.8%
+Table cache: 1 entries (808B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -835,7 +835,7 @@ Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 1
 Block cache: 0 entries (0B)  hit rate: 0.0%
-Table cache: 1 entries (816B)  hit rate: 0.0%
+Table cache: 1 entries (808B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -883,7 +883,7 @@ Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 2
 Block cache: 4 entries (1.4KB)  hit rate: 0.0%
-Table cache: 1 entries (816B)  hit rate: 50.0%
+Table cache: 1 entries (808B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -932,7 +932,7 @@ Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 3
 Block cache: 4 entries (1.4KB)  hit rate: 0.0%
-Table cache: 1 entries (816B)  hit rate: 50.0%
+Table cache: 1 entries (808B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0