Repository: cockroachdb/pebble
Pull request #1873 (2022-08-11 21:00:09)

db: add Metrics.Keys.RangeKeySetsCount
Add a metric recording the approximate number of internal range key sets that
exist within the database. It's calculated through an annotation on the
range-key only level metadata b-trees. It's approximate since it excludes
memtable range key sets and any tables for which table stats have not yet been
loaded. I intend to export this as a time-series metric.

Informs #1848.

cockroach-teamcity: 

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


Patch:
diff --git a/compaction.go b/compaction.go
index d0dd8c6d..5b94788f 100644
--- a/compaction.go
+++ b/compaction.go
@@ -379,6 +379,7 @@ type compaction struct {
 	kind      compactionKind
 	cmp       Compare
 	equal     Equal
+	comparer  *base.Comparer
 	formatKey base.FormatKey
 	logger    Logger
 	version   *version
@@ -515,6 +516,7 @@ func newCompaction(pc *pickedCompaction, opts *Options) *compaction {
 		kind:              compactionKindDefault,
 		cmp:               pc.cmp,
 		equal:             opts.equal(),
+		comparer:          opts.Comparer,
 		formatKey:         opts.Comparer.FormatKey,
 		score:             pc.score,
 		inputs:            pc.inputs,
@@ -558,6 +560,7 @@ func newDeleteOnlyCompaction(opts *Options, cur *version, inputs []compactionLev
 		kind:      compactionKindDeleteOnly,
 		cmp:       opts.Comparer.Compare,
 		equal:     opts.equal(),
+		comparer:  opts.Comparer,
 		formatKey: opts.Comparer.FormatKey,
 		logger:    opts.Logger,
 		version:   cur,
@@ -645,6 +648,7 @@ func newFlush(opts *Options, cur *version, baseLevel int, flushing flushableList
 		kind:              compactionKindFlush,
 		cmp:               opts.Comparer.Compare,
 		equal:             opts.equal(),
+		comparer:          opts.Comparer,
 		formatKey:         opts.Comparer.FormatKey,
 		logger:            opts.Logger,
 		version:           cur,
@@ -1029,7 +1033,7 @@ func (c *compaction) newInputIter(
 			if rangeKeyIter := f.newRangeKeyIter(nil); rangeKeyIter != nil {
 				mi := &keyspan.MergingIter{}
 				mi.Init(c.cmp, rangeKeyCompactionTransform(snapshots, c.elideRangeKey), rangeKeyIter)
-				c.rangeKeyInterleaving.Init(c.cmp, base.WrapIterWithStats(iter), mi, nil /* hooks */, nil /* lowerBound */, nil /* upperBound */)
+				c.rangeKeyInterleaving.Init(c.comparer, base.WrapIterWithStats(iter), mi, nil /* hooks */, nil /* lowerBound */, nil /* upperBound */)
 				iter = &c.rangeKeyInterleaving
 			}
 			return iter, nil
@@ -1056,7 +1060,7 @@ func (c *compaction) newInputIter(
 		if len(rangeKeyIters) > 0 {
 			mi := &keyspan.MergingIter{}
 			mi.Init(c.cmp, rangeKeyCompactionTransform(snapshots, c.elideRangeKey), rangeKeyIters...)
-			c.rangeKeyInterleaving.Init(c.cmp, base.WrapIterWithStats(iter), mi, nil /* hooks */, nil /* lowerBound */, nil /* upperBound */)
+			c.rangeKeyInterleaving.Init(c.comparer, base.WrapIterWithStats(iter), mi, nil /* hooks */, nil /* lowerBound */, nil /* upperBound */)
 			iter = &c.rangeKeyInterleaving
 		}
 		return iter, nil
@@ -1266,7 +1270,7 @@ func (c *compaction) newInputIter(
 		mi.Init(c.cmp, rangeKeyCompactionTransform(snapshots, c.elideRangeKey), rangeKeyIters...)
 		di := &keyspan.DefragmentingIter{}
 		di.Init(c.cmp, mi, keyspan.DefragmentInternal, keyspan.StaticDefragmentReducer)
-		c.rangeKeyInterleaving.Init(c.cmp, pointKeyIter, di, nil /* hooks */, nil /* lowerBound */, nil /* upperBound */)
+		c.rangeKeyInterleaving.Init(c.comparer, pointKeyIter, di, nil /* hooks */, nil /* lowerBound */, nil /* upperBound */)
 		return &c.rangeKeyInterleaving, nil
 	}
 
diff --git a/db.go b/db.go
index 3deac014..d51bcd76 100644
--- a/db.go
+++ b/db.go
@@ -532,6 +532,7 @@ func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, io.Closer,
 		pointIter:    pointIter,
 		merge:        d.merge,
 		split:        d.split,
+		comparer:     d.opts.Comparer,
 		readState:    readState,
 		keyBuf:       buf.keyBuf,
 	}
@@ -906,6 +907,7 @@ func (d *DB) newIterInternal(batch *Batch, s *Snapshot, o *IterOptions) *Iterato
 		equal:               d.equal,
 		merge:               d.merge,
 		split:               d.split,
+		comparer:            d.opts.Comparer,
 		readState:           readState,
 		keyBuf:              buf.keyBuf,
 		prefixOrFullSeekKey: buf.prefixOrFullSeekKey,
@@ -1013,7 +1015,7 @@ func finishInitializingIter(buf *iterAlloc) *Iterator {
 			// NB: The interleaving iterator is always reinitialized, even if
 			// dbi already had an initialized range key iterator, in case the point
 			// iterator changed or the range key masking suffix changed.
-			dbi.rangeKey.iiter.Init(dbi.cmp, dbi.iter, dbi.rangeKey.rangeKeyIter,
+			dbi.rangeKey.iiter.Init(dbi.comparer, dbi.iter, dbi.rangeKey.rangeKeyIter,
 				&dbi.rangeKeyMasking, dbi.opts.LowerBound, dbi.opts.UpperBound)
 			dbi.iter = &dbi.rangeKey.iiter
 		}
@@ -1501,11 +1503,12 @@ func (d *DB) Metrics() *Metrics {
 	recycledLogsCount, recycledLogSize := d.logRecycler.stats()
 
 	d.mu.Lock()
+	vers := d.mu.versions.currentVersion()
 	*metrics = d.mu.versions.metrics
 	metrics.Compact.EstimatedDebt = d.mu.versions.picker.estimatedCompactionDebt(0)
 	metrics.Compact.InProgressBytes = atomic.LoadInt64(&d.mu.versions.atomic.atomicInProgressBytes)
 	metrics.Compact.NumInProgress = int64(d.mu.compact.compactingCount)
-	metrics.Compact.MarkedFiles = d.mu.versions.currentVersion().Stats.MarkedForCompaction
+	metrics.Compact.MarkedFiles = vers.Stats.MarkedForCompaction
 	for _, m := range d.mu.mem.queue {
 		metrics.MemTable.Size += m.totalBytes()
 	}
@@ -1549,6 +1552,8 @@ func (d *DB) Metrics() *Metrics {
 	}
 	metrics.private.optionsFileSize = d.optionsFileSize
 
+	metrics.Keys.RangeKeySetsCount = countRangeKeySetFragments(vers)
+
 	d.mu.versions.logLock()
 	metrics.private.manifestFileSize = uint64(d.mu.versions.manifest.Size())
 	d.mu.versions.logUnlock()
diff --git a/external_iterator.go b/external_iterator.go
index 102c442f..c16c5f95 100644
--- a/external_iterator.go
+++ b/external_iterator.go
@@ -59,6 +59,7 @@ func NewExternalIter(
 		equal:               o.equal(),
 		merge:               o.Merger.Merge,
 		split:               o.Comparer.Split,
+		comparer:            o.Comparer,
 		readState:           nil,
 		keyBuf:              buf.keyBuf,
 		prefixOrFullSeekKey: buf.prefixOrFullSeekKey,
@@ -148,9 +149,10 @@ func finishInitializingExternal(it *Iterator) {
 			it.rangeKey = iterRangeKeyStateAllocPool.Get().(*iteratorRangeKeyState)
 			it.rangeKey.init(it.cmp, it.split, &it.opts)
 			it.rangeKey.rangeKeyIter = it.rangeKey.iterConfig.Init(
-				it.cmp,
+				it.comparer,
 				base.InternalKeySeqNumMax,
 				it.opts.LowerBound, it.opts.UpperBound,
+				&it.hasPrefix, &it.prefixOrFullSeekKey,
 			)
 			for _, r := range it.externalReaders {
 				if rki, err := r.NewRawRangeKeyIter(); err != nil {
@@ -160,7 +162,7 @@ func finishInitializingExternal(it *Iterator) {
 				}
 			}
 		}
-		it.rangeKey.iiter.Init(it.cmp, it.iter, it.rangeKey.rangeKeyIter, &it.rangeKeyMasking,
+		it.rangeKey.iiter.Init(it.comparer, it.iter, it.rangeKey.rangeKeyIter, &it.rangeKeyMasking,
 			it.opts.LowerBound, it.opts.UpperBound)
 		it.iter = &it.rangeKey.iiter
 	}
diff --git a/internal/base/comparer.go b/internal/base/comparer.go
index 95706267..6e5384ea 100644
--- a/internal/base/comparer.go
+++ b/internal/base/comparer.go
@@ -66,6 +66,23 @@ type Separator func(dst, a, b []byte) []byte
 // key must be valid to pass to Compare.
 type Successor func(dst, a []byte) []byte
 
+// ImmediateSuccessor is invoked with a prefix key ([Split(a) == len(a)]) and
+// returns the smallest key that is larger than the given prefix a.
+// ImmediateSuccessor must return a prefix key k such that:
+//
+//   Split(k) == len(k) and Compare(k, a) > 0
+//
+// and there exists no representable k2 such that:
+//
+//   Split(k2) == len(k2) and Compare(k2, a) > 0 and Compare(k2, k) < 0
+//
+// As an example, an implementation built on the natural byte ordering using
+// bytes.Compare could append a `\0` to `a`.
+//
+// The dst parameter may be used to store the returned key, though it is valid
+// to pass nil. The returned key must be valid to pass to Compare.
+type ImmediateSuccessor func(dst, a []byte) []byte
+
 // Split returns the length of the prefix of the user key that corresponds to
 // the key portion of an MVCC encoding scheme to enable the use of prefix bloom
 // filters.
@@ -104,14 +121,15 @@ type Split func(a []byte) int
 // Comparer defines a total ordering over the space of []byte keys: a 'less
 // than' relationship.
 type Comparer struct {
-	Compare        Compare
-	Equal          Equal
-	AbbreviatedKey AbbreviatedKey
-	FormatKey      FormatKey
-	FormatValue    FormatValue
-	Separator      Separator
-	Split          Split
-	Successor      Successor
+	Compare            Compare
+	Equal              Equal
+	AbbreviatedKey     AbbreviatedKey
+	FormatKey          FormatKey
+	FormatValue        FormatValue
+	Separator          Separator
+	Split              Split
+	Successor          Successor
+	ImmediateSuccessor ImmediateSuccessor
 
 	// Name is the name of the comparer.
 	//
@@ -193,6 +211,10 @@ var DefaultComparer = &Comparer{
 		return append(dst, a...)
 	},
 
+	ImmediateSuccessor: func(dst, a []byte) (ret []byte) {
+		return append(append(dst, a...), 0x00)
+	},
+
 	// This name is part of the C++ Level-DB implementation's default file
 	// format, and should not be changed.
 	Name: "leveldb.BytewiseComparator",
diff --git a/internal/keyspan/bounded.go b/internal/keyspan/bounded.go
index e7a94086..2404f72e 100644
--- a/internal/keyspan/bounded.go
+++ b/internal/keyspan/bounded.go
@@ -33,21 +33,39 @@ const (
 
 // BoundedIter implements FragmentIterator and enforces bounds.
 type BoundedIter struct {
-	iter     FragmentIterator
-	iterSpan *Span
-	cmp      base.Compare
-	lower    []byte
-	upper    []byte
-	pos      boundedIterPos
+	iter      FragmentIterator
+	iterSpan  *Span
+	cmp       base.Compare
+	split     base.Split
+	lower     []byte
+	upper     []byte
+	hasPrefix *bool
+	prefix    *[]byte
+	pos       boundedIterPos
 }
 
 // Init initializes the bounded iterator.
-func (i *BoundedIter) Init(cmp base.Compare, iter FragmentIterator, lower, upper []byte) {
+//
+// In addition to the iterator bounds, Init takes pointers to a boolean
+// indicating whether the iterator is in prefix iteration mode and the prefix
+// key if it is. This is used to exclude spans that are outside the iteration
+// prefix.
+func (i *BoundedIter) Init(
+	cmp base.Compare,
+	split base.Split,
+	iter FragmentIterator,
+	lower, upper []byte,
+	hasPrefix *bool,
+	prefix *[]byte,
+) {
 	*i = BoundedIter{
-		iter:  iter,
-		cmp:   cmp,
-		lower: lower,
-		upper: upper,
+		iter:      iter,
+		cmp:       cmp,
+		split:     split,
+		lower:     lower,
+		upper:     upper,
+		hasPrefix: hasPrefix,
+		prefix:    prefix,
 	}
 }
 
@@ -91,6 +109,15 @@ func (i *BoundedIter) Next() *Span {
 			i.pos = posAtUpperLimit
 			return nil
 		}
+		// Similarly, if the span extends to the next prefix and we're in prefix
+		// iteration mode, we can avoid advancing.
+		if i.iterSpan != nil && *i.hasPrefix {
+			ei := i.split(i.iterSpan.End)
+			if i.cmp(i.iterSpan.End[:ei], *i.prefix) > 0 {
+				i.pos = posAtUpperLimit
+				return nil
+			}
+		}
 		return i.checkForwardBound(i.iter.Next())
 	case posAtUpperLimit:
 		// Already exhausted.
@@ -114,6 +141,15 @@ func (i *BoundedIter) Prev() *Span {
 			i.pos = posAtLowerLimit
 			return nil
 		}
+		// Similarly, if the span extends to or beyond the current prefix and
+		// we're in prefix iteration mode, we can avoid advancing.
+		if i.iterSpan != nil && *i.hasPrefix {
+			si := i.split(i.iterSpan.Start)
+			if i.cmp(i.iterSpan.Start[:si], *i.prefix) < 0 {
+				i.pos = posAtLowerLimit
+				return nil
+			}
+		}
 		return i.checkBackwardBound(i.iter.Prev())
 	case posAtUpperLimit:
 		// The BoundedIter had previously returned nil, because it knew from
@@ -146,6 +182,17 @@ func (i *BoundedIter) SetBounds(lower, upper []byte) {
 // span is wholly outside the upper bound. It also updates i.pos and i.iterSpan
 // to reflect the new iterator position.
 func (i *BoundedIter) checkForwardBound(span *Span) *Span {
+	// Compare to the prefix's bounds, if in prefix iteration mode.
+	if span != nil && *i.hasPrefix {
+		si := i.split(span.Start)
+		if i.cmp(span.Start[:si], *i.prefix) > 0 {
+			// This span starts at a whole prefix that sorts after our current
+			// prefix.
+			span = nil
+		}
+	}
+
+	// Compare to the upper bound.
 	if span != nil && i.upper != nil && i.cmp(span.Start, i.upper) >= 0 {
 		span = nil
 	}
@@ -160,6 +207,13 @@ func (i *BoundedIter) checkForwardBound(span *Span) *Span {
 // wholly outside the lower bound.  It also updates i.pos and i.iterSpan to
 // reflect the new iterator position.
 func (i *BoundedIter) checkBackwardBound(span *Span) *Span {
+	// Compare to the prefix's bounds, if in prefix iteration mode.
+	if span != nil && *i.hasPrefix && i.cmp(span.End, *i.prefix) <= 0 {
+		// This span ends before the current prefix.
+		span = nil
+	}
+
+	// Compare to the lower bound.
 	if span != nil && i.lower != nil && i.cmp(span.End, i.lower) <= 0 {
 		span = nil
 	}
diff --git a/internal/keyspan/interleaving_iter.go b/internal/keyspan/interleaving_iter.go
index a6ff97ec..ad1832d1 100644
--- a/internal/keyspan/interleaving_iter.go
+++ b/internal/keyspan/interleaving_iter.go
@@ -93,6 +93,7 @@ type SpanMask interface {
 // All spans containing keys are exposed during iteration.
 type InterleavingIter struct {
 	cmp         base.Compare
+	comparer    *base.Comparer
 	pointIter   base.InternalIteratorWithStats
 	keyspanIter FragmentIterator
 	mask        SpanMask
@@ -103,9 +104,13 @@ type InterleavingIter struct {
 	// to truncate a span. The byte slices backing a SeekGE/SeekPrefixGE search
 	// keys can come directly from the end user, so they're copied into keyBuf
 	// to ensure key stability.
-	keyBuf   []byte
-	pointKey *base.InternalKey
-	pointVal []byte
+	keyBuf []byte
+	// nextPrefixBuf is used during SeekPrefixGE calls to store the truncated
+	// upper bound of the returned spans. SeekPrefixGE truncates the returned
+	// spans to an upper bound of the seeked prefix's immediate successor.
+	nextPrefixBuf []byte
+	pointKey      *base.InternalKey
+	pointVal      []byte
 	// span holds the span at the keyspanIter's current position. If the span is
 	// wholly contained within the iterator bounds, this span is directly
 	// returned to the iterator consumer through Span(). If either bound needed
@@ -153,6 +158,12 @@ type InterleavingIter struct {
 	// maskSpanChangedCalled records whether or not the last call to
 	// SpanMask.SpanChanged provided the current span (i.span) or not.
 	maskSpanChangedCalled bool
+	// prefix records whether the iteator is in prefix mode. During prefix mode,
+	// Pebble will truncate spans to the next prefix. If the iterator
+	// subsequently leaves prefix mode, the existing span cached in i.span must
+	// be invalidated because its bounds do not reflect the original span's true
+	// bounds.
+	prefix bool
 	// dir indicates the direction of iteration: forward (+1) or backward (-1)
 	dir int8
 }
@@ -166,14 +177,15 @@ var _ base.InternalIterator = &InterleavingIter{}
 // The point iterator must already have the provided bounds. Init does not
 // propagate the bounds down the iterator stack.
 func (i *InterleavingIter) Init(
-	cmp base.Compare,
+	comparer *base.Comparer,
 	pointIter base.InternalIteratorWithStats,
 	keyspanIter FragmentIterator,
 	mask SpanMask,
 	lowerBound, upperBound []byte,
 ) {
 	*i = InterleavingIter{
-		cmp:         cmp,
+		cmp:         comparer.Compare,
+		comparer:    comparer,
 		pointIter:   pointIter,
 		keyspanIter: keyspanIter,
 		mask:        mask,
@@ -193,7 +205,7 @@ func (i *InterleavingIter) Init(
 // It allows for seeding the iterator with the current position of the point
 // iterator.
 func (i *InterleavingIter) InitSeekGE(
-	key []byte, pointKey *base.InternalKey, pointValue []byte,
+	prefix, key []byte, pointKey *base.InternalKey, pointValue []byte,
 ) (*base.InternalKey, []byte) {
 	i.dir = +1
 	i.clearMask()
@@ -202,8 +214,8 @@ func (i *InterleavingIter) InitSeekGE(
 	// NB: This keyspanSeekGE call will truncate the span to the seek key if
 	// necessary. This truncation is important for cases where a switch to
 	// combined iteration is made during a user-initiated SeekGE.
-	i.keyspanSeekGE(key)
-	return i.interleaveForward(key)
+	i.keyspanSeekGE(key, prefix)
+	return i.interleaveForward(key, prefix)
 }
 
 // InitSeekLT may be called after Init but before any positioning method.
@@ -239,6 +251,7 @@ func (i *InterleavingIter) InitSeekLT(
 //   i.lower ≤ key
 func (i *InterleavingIter) SeekGE(key []byte, flags base.SeekGEFlags) (*base.InternalKey, []byte) {
 	i.clearMask()
+	i.disablePrefixMode()
 	i.pointKey, i.pointVal = i.pointIter.SeekGE(key, flags)
 	i.pointKeyInterleaved = false
 
@@ -248,14 +261,14 @@ func (i *InterleavingIter) SeekGE(key []byte, flags base.SeekGEFlags) (*base.Int
 	if i.span != nil && i.cmp(key, i.span.End) < 0 && i.cmp(key, i.span.Start) >= 0 {
 		// We're seeking within the existing span's bounds. We still might need
 		// truncate the span to the iterator's bounds.
-		i.checkForwardBound()
+		i.checkForwardBound(nil /* prefix */)
 		i.savedKeyspan()
 	} else {
-		i.keyspanSeekGE(key)
+		i.keyspanSeekGE(key, nil /* prefix */)
 	}
 
 	i.dir = +1
-	return i.interleaveForward(key)
+	return i.interleaveForward(key, nil /* prefix */)
 }
 
 // SeekPrefixGE implements (base.InternalIterator).SeekPrefixGE.
@@ -274,6 +287,7 @@ func (i *InterleavingIter) SeekPrefixGE(
 	i.clearMask()
 	i.pointKey, i.pointVal = i.pointIter.SeekPrefixGE(prefix, key, flags)
 	i.pointKeyInterleaved = false
+	i.prefix = true
 
 	// We need to seek the keyspan iterator too. If the keyspan iterator was
 	// already positioned at a span, we might be able to avoid the seek if the
@@ -281,19 +295,20 @@ func (i *InterleavingIter) SeekPrefixGE(
 	if i.span != nil && i.cmp(key, i.span.End) < 0 && i.cmp(key, i.span.Start) >= 0 {
 		// We're seeking within the existing span's bounds. We still might need
 		// truncate the span to the iterator's bounds.
-		i.checkForwardBound()
+		i.checkForwardBound(prefix)
 		i.savedKeyspan()
 	} else {
-		i.keyspanSeekGE(key)
+		i.keyspanSeekGE(key, prefix)
 	}
 
 	i.dir = +1
-	return i.interleaveForward(key)
+	return i.interleaveForward(key, prefix)
 }
 
 // SeekLT implements (base.InternalIterator).SeekLT.
 func (i *InterleavingIter) SeekLT(key []byte, flags base.SeekLTFlags) (*base.InternalKey, []byte) {
 	i.clearMask()
+	i.disablePrefixMode()
 	i.pointKey, i.pointVal = i.pointIter.SeekLT(key, flags)
 	i.pointKeyInterleaved = false
 
@@ -331,18 +346,20 @@ func (i *InterleavingIter) SeekLT(key []byte, flags base.SeekLTFlags) (*base.Int
 // First implements (base.InternalIterator).First.
 func (i *InterleavingIter) First() (*base.InternalKey, []byte) {
 	i.clearMask()
+	i.disablePrefixMode()
 	i.pointKey, i.pointVal = i.pointIter.First()
 	i.pointKeyInterleaved = false
 	i.span = i.keyspanIter.First()
-	i.checkForwardBound()
+	i.checkForwardBound(nil /* prefix */)
 	i.savedKeyspan()
 	i.dir = +1
-	return i.interleaveForward(i.lower)
+	return i.interleaveForward(i.lower, nil /* prefix */)
 }
 
 // Last implements (base.InternalIterator).Last.
 func (i *InterleavingIter) Last() (*base.InternalKey, []byte) {
 	i.clearMask()
+	i.disablePrefixMode()
 	i.pointKey, i.pointVal = i.pointIter.Last()
 	i.pointKeyInterleaved = false
 	i.span = i.keyspanIter.Last()
@@ -387,7 +404,7 @@ func (i *InterleavingIter) Next() (*base.InternalKey, []byte) {
 			// There was no span in the reverse direction, but there may be
 			// a span in the forward direction.
 			i.span = i.keyspanIter.Next()
-			i.checkForwardBound()
+			i.checkForwardBound(nil /* prefix */)
 			i.savedKeyspan()
 		} else {
 			// Regardless of the current iterator state, we mark any existing
@@ -425,10 +442,10 @@ func (i *InterleavingIter) Next() (*base.InternalKey, []byte) {
 	if i.keyspanInterleaved && i.pointKey != nil && i.span != nil &&
 		i.cmp(i.pointKey.UserKey, i.span.End) >= 0 {
 		i.span = i.keyspanIter.Next()
-		i.checkForwardBound()
+		i.checkForwardBound(nil /* prefix */)
 		i.savedKeyspan()
 	}
-	return i.interleaveForward(i.lower)
+	return i.interleaveForward(i.lower, nil /* prefix */)
 }
 
 // Prev implements (base.InternalIterator).Prev.
@@ -526,7 +543,9 @@ func (i *InterleavingIter) Prev() (*base.InternalKey, []byte) {
 	return i.interleaveBackward()
 }
 
-func (i *InterleavingIter) interleaveForward(lowerBound []byte) (*base.InternalKey, []byte) {
+func (i *InterleavingIter) interleaveForward(
+	lowerBound []byte, prefix []byte,
+) (*base.InternalKey, []byte) {
 	// This loop determines whether a point key or a span marker key should be
 	// interleaved on each iteration. If masking is disabled and the span is
 	// nonempty, this loop executes for exactly one iteration. If masking is
@@ -572,7 +591,7 @@ func (i *InterleavingIter) interleaveForward(lowerBound []byte) (*base.InternalK
 			// advancing past the current point key.
 			if i.keyspanInterleaved {
 				i.span = i.keyspanIter.Next()
-				i.checkForwardBound()
+				i.checkForwardBound(prefix)
 				i.savedKeyspan()
 				if i.span == nil {
 					return i.yieldNil()
@@ -594,7 +613,7 @@ func (i *InterleavingIter) interleaveForward(lowerBound []byte) (*base.InternalK
 							// keyspanInterleaved would likely trip up the invariant check
 							// above.
 							i.span = i.keyspanIter.Next()
-							i.checkForwardBound()
+							i.checkForwardBound(prefix)
 							i.savedKeyspan()
 						} else {
 							i.keyspanInterleaved = true
@@ -636,7 +655,7 @@ func (i *InterleavingIter) interleaveForward(lowerBound []byte) (*base.InternalK
 					// reestablish it before the next iteration.
 					if i.pointKey != nil && i.cmp(i.pointKey.UserKey, i.span.End) >= 0 {
 						i.span = i.keyspanIter.Next()
-						i.checkForwardBound()
+						i.checkForwardBound(prefix)
 						i.savedKeyspan()
 					}
 					continue
@@ -733,7 +752,7 @@ func (i *InterleavingIter) interleaveBackward() (*base.InternalKey, []byte) {
 // keyspanSeekGE seeks the keyspan iterator to the first span covering k ≥ key.
 // Note that this differs from the FragmentIterator.SeekGE semantics, which
 // seek to the first span with a start key ≥ key.
-func (i *InterleavingIter) keyspanSeekGE(key []byte) {
+func (i *InterleavingIter) keyspanSeekGE(key []byte, prefix []byte) {
 	// Seek using SeekLT to look for a span that starts before key, with an end
 	// boundary extending beyond key.
 	i.span = i.keyspanIter.SeekLT(key)
@@ -742,7 +761,7 @@ func (i *InterleavingIter) keyspanSeekGE(key []byte) {
 		// found ends before key. Next to the first key with a start ≥ key.
 		i.span = i.keyspanIter.Next()
 	}
-	i.checkForwardBound()
+	i.checkForwardBound(prefix)
 	i.savedKeyspan()
 }
 
@@ -769,7 +788,7 @@ func (i *InterleavingIter) keyspanSeekLT(key []byte) {
 	i.savedKeyspan()
 }
 
-func (i *InterleavingIter) checkForwardBound() {
+func (i *InterleavingIter) checkForwardBound(prefix []byte) {
 	i.truncated = false
 	i.truncatedSpan = Span{}
 	if i.span == nil {
@@ -801,6 +820,22 @@ func (i *InterleavingIter) checkForwardBound() {
 		}
 		i.truncatedSpan.End = i.upper
 	}
+	// If this is a part of a SeekPrefixGE call, we may also need to truncate to
+	// the prefix's bounds.
+	if prefix != nil {
+		if !i.truncated {
+			i.truncated = true
+			i.truncatedSpan = *i.span
+		}
+		if i.cmp(prefix, i.truncatedSpan.Start) > 0 {
+			i.truncatedSpan.Start = prefix
+		}
+		i.nextPrefixBuf = i.comparer.ImmediateSuccessor(i.nextPrefixBuf[:0], prefix)
+		if i.truncated && i.cmp(i.nextPrefixBuf, i.truncatedSpan.End) < 0 {
+			i.truncatedSpan.End = i.nextPrefixBuf
+		}
+	}
+
 	if i.truncated && i.cmp(i.truncatedSpan.Start, i.truncatedSpan.End) == 0 {
 		i.span = nil
 	}
@@ -895,6 +930,15 @@ func (i *InterleavingIter) yieldSyntheticSpanMarker(lowerBound []byte) (*base.In
 	return i.verify(&i.spanMarker, nil)
 }
 
+func (i *InterleavingIter) disablePrefixMode() {
+	if i.prefix {
+		i.prefix = false
+		// Clear the existing span. It may not hold the true end bound of the
+		// underlying span.
+		i.span = nil
+	}
+}
+
 func (i *InterleavingIter) verify(k *base.InternalKey, v []byte) (*base.InternalKey, []byte) {
 	// Wrap the entire function body in the invariants build tag, so that
 	// production builds elide this entire function.
@@ -939,7 +983,11 @@ func (i *InterleavingIter) maybeUpdateMask(covered bool) {
 		if !covered || i.span.Empty() {
 			i.clearMask()
 		} else if !i.maskSpanChangedCalled {
-			i.mask.SpanChanged(i.span)
+			if i.truncated {
+				i.mask.SpanChanged(&i.truncatedSpan)
+			} else {
+				i.mask.SpanChanged(i.span)
+			}
 			i.maskSpanChangedCalled = true
 		}
 	}
diff --git a/internal/rangekey/coalesce.go b/internal/rangekey/coalesce.go
index 2af1096d..cb78d238 100644
--- a/internal/rangekey/coalesce.go
+++ b/internal/rangekey/coalesce.go
@@ -35,12 +35,17 @@ type UserIteratorConfig struct {
 // The snapshot sequence number parameter determines which keys are visible. Any
 // keys not visible at the provided snapshot are ignored.
 func (ui *UserIteratorConfig) Init(
-	cmp base.Compare, snapshot uint64, lower, upper []byte, iters ...keyspan.FragmentIterator,
+	comparer *base.Comparer,
+	snapshot uint64,
+	lower, upper []byte,
+	hasPrefix *bool,
+	prefix *[]byte,
+	iters ...keyspan.FragmentIterator,
 ) keyspan.FragmentIterator {
 	ui.snapshot = snapshot
-	ui.miter.Init(cmp, ui, iters...)
-	ui.biter.Init(cmp, &ui.miter, lower, upper)
-	ui.diter.Init(cmp, &ui.biter, ui, keyspan.StaticDefragmentReducer)
+	ui.miter.Init(comparer.Compare, ui, iters...)
+	ui.biter.Init(comparer.Compare, comparer.Split, &ui.miter, lower, upper, hasPrefix, prefix)
+	ui.diter.Init(comparer.Compare, &ui.biter, ui, keyspan.StaticDefragmentReducer)
 	ui.litersUsed = 0
 	return &ui.diter
 }
diff --git a/iterator.go b/iterator.go
index 35fc2785..6c7f129e 100644
--- a/iterator.go
+++ b/iterator.go
@@ -151,6 +151,7 @@ type Iterator struct {
 	equal     Equal
 	merge     Merge
 	split     Split
+	comparer  *base.Comparer
 	iter      internalIteratorWithStats
 	pointIter internalIteratorWithStats
 	readState *readState
@@ -1099,6 +1100,12 @@ func (i *Iterator) SeekGEWithLimit(key []byte, limit []byte) IterValidityState {
 //  }
 //
 // See ExampleIterator_SeekPrefixGE for a working example.
+//
+// When iterating with range keys enabled, all range keys encountered are
+// truncated to the seek key's prefix's bounds. The truncation of the upper
+// bound requires that the database's Comparer is configured with a
+// ImmediateSuccessor method. For example, a SeekPrefixGE("a@9") call with the
+// prefix "a" will truncate range key bounds to [a,ImmediateSuccessor(a)].
 func (i *Iterator) SeekPrefixGE(key []byte) bool {
 	lastPositioningOp := i.lastPositioningOp
 	// Set it to unknown, since this operation may not succeed, in which case
@@ -1115,7 +1122,9 @@ func (i *Iterator) SeekPrefixGE(key []byte) bool {
 	if i.split == nil {
 		panic("pebble: split must be provided for SeekPrefixGE")
 	}
-
+	if i.comparer.ImmediateSuccessor == nil && i.opts.KeyTypes != IterKeyTypePointsOnly {
+		panic("pebble: ImmediateSuccessor must be provided for SeekPrefixGE with range keys")
+	}
 	prefixLen := i.split(key)
 	keyPrefix := key[:prefixLen]
 	var flags base.SeekGEFlags
@@ -1220,7 +1229,6 @@ func (i *Iterator) SeekLTWithLimit(key []byte, limit []byte) IterValidityState {
 	i.lastPositioningOp = unknownLastPositionOp
 	i.requiresReposition = false
 	i.err = nil // clear cached iteration error
-	i.hasPrefix = false
 	i.stats.ReverseSeekCount[InterfaceCall]++
 	if upperBound := i.opts.GetUpperBound(); upperBound != nil && i.cmp(key, upperBound) > 0 {
 		key = upperBound
@@ -1231,6 +1239,7 @@ func (i *Iterator) SeekLTWithLimit(key []byte, limit []byte) IterValidityState {
 		i.rangeKey.updated = false
 		i.rangeKey.prevPosHadRangeKey = i.rangeKey.hasRangeKey && i.Valid()
 	}
+	i.hasPrefix = false
 	seekInternalIter := true
 	// The following noop optimization only applies when i.batch == nil, since
 	// an iterator over a batch is iterating over mutable data, that may have
@@ -1567,13 +1576,13 @@ func (i *Iterator) saveRangeKey() {
 	// key and the new one has an identical start key, then it must be the same
 	// range key and we can avoid copying and keep `i.rangeKey.updated=false`.
 	//
-	// TODO(jackson): This key comparison could be avoidable during relative
+	// TODO(jackson): These key comparisons could be avoidable during relative
 	// positioning operations continuing in the same direction, because these
 	// ops will never encounter the previous position's range key while
 	// stale=true. However, threading whether the current op is a seek or step
 	// maybe isn't worth it. This key comparison is only necessary once when we
 	// step onto a new range key, which should be relatively rare.
-	if i.rangeKey.prevPosHadRangeKey && i.equal(i.rangeKey.start, s.Start) {
+	if i.rangeKey.prevPosHadRangeKey && i.equal(i.rangeKey.start, s.Start) && i.equal(i.rangeKey.end, s.End) {
 		i.rangeKey.updated = false
 		i.rangeKey.stale = false
 		i.rangeKey.hasRangeKey = true
@@ -2157,6 +2166,7 @@ func (i *Iterator) Clone(opts CloneOptions) (*Iterator, error) {
 		equal:               i.equal,
 		merge:               i.merge,
 		split:               i.split,
+		comparer:            i.comparer,
 		readState:           readState,
 		keyBuf:              buf.keyBuf,
 		prefixOrFullSeekKey: buf.prefixOrFullSeekKey,
diff --git a/metrics.go b/metrics.go
index a856587f..93557e2b 100644
--- a/metrics.go
+++ b/metrics.go
@@ -178,6 +178,11 @@ type Metrics struct {
 		ZombieCount int64
 	}
 
+	Keys struct {
+		// The approximate count of internal range key set keys in the database.
+		RangeKeySetsCount uint64
+	}
+
 	Snapshots struct {
 		// The number of currently open snapshots.
 		Count int
diff --git a/range_keys.go b/range_keys.go
index 5f323ac9..d55e8c70 100644
--- a/range_keys.go
+++ b/range_keys.go
@@ -15,7 +15,9 @@ import (
 // constructRangeKeyIter constructs the range-key iterator stack, populating
 // i.rangeKey.rangeKeyIter with the resulting iterator.
 func (i *Iterator) constructRangeKeyIter() {
-	i.rangeKey.rangeKeyIter = i.rangeKey.iterConfig.Init(i.cmp, i.seqNum, i.opts.LowerBound, i.opts.UpperBound)
+	i.rangeKey.rangeKeyIter = i.rangeKey.iterConfig.Init(
+		i.comparer, i.seqNum, i.opts.LowerBound, i.opts.UpperBound,
+		&i.hasPrefix, &i.prefixOrFullSeekKey)
 
 	// If there's an indexed batch with range keys, include it.
 	if i.batch != nil {
@@ -455,7 +457,7 @@ func (i *lazyCombinedIter) initCombinedIteration(
 
 	// Initialize the Iterator's interleaving iterator.
 	i.parent.rangeKey.iiter.Init(
-		i.parent.cmp, i.parent.pointIter, i.parent.rangeKey.rangeKeyIter,
+		i.parent.comparer, i.parent.pointIter, i.parent.rangeKey.rangeKeyIter,
 		&i.parent.rangeKeyMasking, i.parent.opts.LowerBound, i.parent.opts.UpperBound)
 
 	// We need to determine the key to seek the range key iterator to. If
@@ -523,7 +525,11 @@ func (i *lazyCombinedIter) initCombinedIteration(
 	// which case the range key will be interleaved next instead of the point
 	// key.
 	if dir == +1 {
-		return i.parent.rangeKey.iiter.InitSeekGE(seekKey, pointKey, pointValue)
+		var prefix []byte
+		if i.parent.hasPrefix {
+			prefix = i.parent.prefixOrFullSeekKey
+		}
+		return i.parent.rangeKey.iiter.InitSeekGE(prefix, seekKey, pointKey, pointValue)
 	}
 	return i.parent.rangeKey.iiter.InitSeekLT(seekKey, pointKey, pointValue)
 }
diff --git a/table_stats.go b/table_stats.go
index c2b8d576..228ef693 100644
--- a/table_stats.go
+++ b/table_stats.go
@@ -759,3 +759,50 @@ func newCombinedDeletionKeyspanIter(
 
 	return mIter, nil
 }
+
+// rangeKeySetsAnnotator implements manifest.Annotator, annotating B-Tree nodes
+// with the sum of the files' counts of range key fragments. Its annotation type
+// is a *uint64. The count of range key sets may change once a table's stats are
+// loaded asynchronously, so its values are marked as cacheable only if a file's
+// stats have been loaded.
+type rangeKeySetsAnnotator struct{}
+
+var _ manifest.Annotator = rangeKeySetsAnnotator{}
+
+func (a rangeKeySetsAnnotator) Zero(dst interface{}) interface{} {
+	if dst == nil {
+		return new(uint64)
+	}
+	v := dst.(*uint64)
+	*v = 0
+	return v
+}
+
+func (a rangeKeySetsAnnotator) Accumulate(
+	f *fileMetadata, dst interface{},
+) (v interface{}, cacheOK bool) {
+	vptr := dst.(*uint64)
+	*vptr = *vptr + f.Stats.NumRangeKeySets
+	return vptr, f.StatsValidLocked()
+}
+
+func (a rangeKeySetsAnnotator) Merge(src interface{}, dst interface{}) interface{} {
+	srcV := src.(*uint64)
+	dstV := dst.(*uint64)
+	*dstV = *dstV + *srcV
+	return dstV
+}
+
+// countRangeKeySetFragments counts the number of RANGEKEYSET keys across all
+// files of the LSM. It only counts keys in files for which table stats have
+// been loaded. It uses a b-tree annotator to cache intermediate values between
+// calculations when possible.
+func countRangeKeySetFragments(v *version) (count uint64) {
+	for l := 0; l < numLevels; l++ {
+		if v.RangeKeyLevels[l].Empty() {
+			continue
+		}
+		count += *v.RangeKeyLevels[l].Annotation(rangeKeySetsAnnotator{}).(*uint64)
+	}
+	return count
+}


Test patch:
diff --git a/compaction_iter_test.go b/compaction_iter_test.go
index 4fd1586a..d1e5ea9a 100644
--- a/compaction_iter_test.go
+++ b/compaction_iter_test.go
@@ -100,7 +100,7 @@ func TestCompactionIter(t *testing.T) {
 		fi := &fakeIter{keys: keys, vals: vals}
 		interleavingIter = &keyspan.InterleavingIter{}
 		interleavingIter.Init(
-			base.DefaultComparer.Compare,
+			base.DefaultComparer,
 			base.WrapIterWithStats(fi),
 			keyspan.NewIter(base.DefaultComparer.Compare, rangeKeys),
 			nil, nil, nil)
diff --git a/compaction_test.go b/compaction_test.go
index 8b09f867..b120fbc3 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -666,6 +666,7 @@ func TestElideTombstone(t *testing.T) {
 	for _, tc := range testCases {
 		c := compaction{
 			cmp:      DefaultComparer.Compare,
+			comparer: DefaultComparer,
 			version:  tc.version,
 			inputs:   []compactionLevel{{level: tc.level}, {level: tc.level + 1}},
 			smallest: base.ParseInternalKey("a.SET.0"),
@@ -811,6 +812,7 @@ func TestElideRangeTombstone(t *testing.T) {
 	for _, tc := range testCases {
 		c := compaction{
 			cmp:      DefaultComparer.Compare,
+			comparer: DefaultComparer,
 			version:  tc.version,
 			inputs:   []compactionLevel{{level: tc.level}, {level: tc.level + 1}},
 			smallest: base.ParseInternalKey("a.SET.0"),
@@ -869,6 +871,7 @@ func TestCompactionTransform(t *testing.T) {
 			var outSpan keyspan.Span
 			c := compaction{
 				cmp:                base.DefaultComparer.Compare,
+				comparer:           base.DefaultComparer,
 				disableSpanElision: disableElision,
 				inuseKeyRanges:     keyRanges,
 			}
@@ -1574,6 +1577,7 @@ func TestCompactionFindGrandparentLimit(t *testing.T) {
 				c := &compaction{
 					cmp:          cmp,
 					equal:        DefaultComparer.Equal,
+					comparer:     DefaultComparer,
 					grandparents: manifest.NewLevelSliceKeySorted(cmp, grandparents),
 				}
 				if len(d.CmdArgs) != 1 {
@@ -1703,6 +1707,7 @@ func TestCompactionFindL0Limit(t *testing.T) {
 				c := &compaction{
 					cmp:      cmp,
 					equal:    DefaultComparer.Equal,
+					comparer: DefaultComparer,
 					version:  vers,
 					l0Limits: vers.L0Sublevels.FlushSplitKeys(),
 					inputs:   []compactionLevel{{level: -1}, {level: 0}},
@@ -1797,9 +1802,10 @@ func TestCompactionAtomicUnitBounds(t *testing.T) {
 
 			case "atomic-unit-bounds":
 				c := &compaction{
-					cmp:    cmp,
-					equal:  DefaultComparer.Equal,
-					inputs: []compactionLevel{{files: files}, {}},
+					cmp:      cmp,
+					equal:    DefaultComparer.Equal,
+					comparer: DefaultComparer,
+					inputs:   []compactionLevel{{files: files}, {}},
 				}
 				c.startLevel, c.outputLevel = &c.inputs[0], &c.inputs[1]
 				if len(d.CmdArgs) != 1 {
@@ -2415,6 +2421,7 @@ func TestCompactionInuseKeyRanges(t *testing.T) {
 			c = &compaction{
 				cmp:       DefaultComparer.Compare,
 				equal:     DefaultComparer.Equal,
+				comparer:  DefaultComparer,
 				formatKey: DefaultComparer.FormatKey,
 				inputs:    []compactionLevel{{}, {}},
 			}
@@ -2639,9 +2646,10 @@ func TestCompactionAllowZeroSeqNum(t *testing.T) {
 			case "allow-zero-seqnum":
 				d.mu.Lock()
 				c := &compaction{
-					cmp:     d.cmp,
-					version: d.mu.versions.currentVersion(),
-					inputs:  []compactionLevel{{}, {}},
+					cmp:      d.cmp,
+					comparer: d.opts.Comparer,
+					version:  d.mu.versions.currentVersion(),
+					inputs:   []compactionLevel{{}, {}},
 				}
 				c.startLevel, c.outputLevel = &c.inputs[0], &c.inputs[1]
 				d.mu.Unlock()
@@ -2725,6 +2733,7 @@ func TestCompactionErrorOnUserKeyOverlap(t *testing.T) {
 			case "error-on-user-key-overlap":
 				c := &compaction{
 					cmp:       DefaultComparer.Compare,
+					comparer:  DefaultComparer,
 					formatKey: DefaultComparer.FormatKey,
 				}
 				var files []manifest.NewFileEntry
@@ -2853,6 +2862,7 @@ func TestCompactionCheckOrdering(t *testing.T) {
 			case "check-ordering":
 				c := &compaction{
 					cmp:       DefaultComparer.Compare,
+					comparer:  DefaultComparer,
 					formatKey: DefaultComparer.FormatKey,
 					logger:    panicLogger{},
 					inputs:    []compactionLevel{{level: -1}, {level: -1}},
diff --git a/data_test.go b/data_test.go
index ab185455..3f91de65 100644
--- a/data_test.go
+++ b/data_test.go
@@ -313,7 +313,7 @@ func printIterState(
 			}
 			if hasRange {
 				start, end := iter.RangeBounds()
-				fmt.Fprintf(b, "[%s-%s)", start, end)
+				fmt.Fprintf(b, "[%s-%s)", formatASCIIKey(start), formatASCIIKey(end))
 				writeRangeKeys(b, iter)
 			} else {
 				fmt.Fprint(b, ".")
@@ -329,7 +329,7 @@ func printIterState(
 					panic(fmt.Sprintf("pebble: unexpected HasPointAndRange (%t, %t)", hasPoint, hasRange))
 				}
 				start, end := iter.RangeBounds()
-				fmt.Fprintf(b, "%s [%s-%s)", iter.Key(), start, end)
+				fmt.Fprintf(b, "%s [%s-%s)", iter.Key(), formatASCIIKey(start), formatASCIIKey(end))
 				writeRangeKeys(b, iter)
 			} else {
 				fmt.Fprint(b, ".")
@@ -346,6 +346,14 @@ func printIterState(
 	}
 }
 
+func formatASCIIKey(b []byte) string {
+	if bytes.IndexFunc(b, func(r rune) bool { return r < 'A' || r > 'z' }) != -1 {
+		// This key is not just ASCII letters. Quote it.
+		return fmt.Sprintf("%q", b)
+	}
+	return string(b)
+}
+
 func writeRangeKeys(b io.Writer, iter *Iterator) {
 	rangeKeys := iter.RangeKeys()
 	for j := 0; j < len(rangeKeys); j++ {
diff --git a/internal/keyspan/bounded_test.go b/internal/keyspan/bounded_test.go
index 854feebb..ba1e3d06 100644
--- a/internal/keyspan/bounded_test.go
+++ b/internal/keyspan/bounded_test.go
@@ -16,27 +16,28 @@ import (
 )
 
 func TestBoundedIter(t *testing.T) {
-	getBounds := func(td *datadriven.TestData) (lower, upper []byte, ok bool) {
+	getBounds := func(td *datadriven.TestData) (lower, upper []byte) {
 		for _, cmdArg := range td.CmdArgs {
 			switch cmdArg.Key {
 			case "lower":
 				if len(cmdArg.Vals[0]) > 0 {
 					lower = []byte(cmdArg.Vals[0])
 				}
-				ok = true
 			case "upper":
 				if len(cmdArg.Vals[0]) > 0 {
 					upper = []byte(cmdArg.Vals[0])
 				}
-				ok = true
 			}
 		}
-		return lower, upper, ok
+		return lower, upper
 	}
 
 	cmp := testkeys.Comparer.Compare
+	split := testkeys.Comparer.Split
 	var buf bytes.Buffer
 	var iter BoundedIter
+	var hasPrefix bool
+	var prefix []byte
 	datadriven.RunTest(t, "testdata/bounded_iter", func(td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "define":
@@ -46,14 +47,20 @@ func TestBoundedIter(t *testing.T) {
 				spans = append(spans, ParseSpan(line))
 			}
 			inner := &invalidatingIter{iter: NewIter(cmp, spans)}
-			lower, upper, _ := getBounds(td)
-			iter.Init(cmp, inner, lower, upper)
+			lower, upper := getBounds(td)
+			iter.Init(cmp, split, inner, lower, upper, &hasPrefix, &prefix)
 			return ""
+		case "set-prefix":
+			hasPrefix = len(td.CmdArgs) > 0
+			if hasPrefix {
+				prefix = []byte(td.CmdArgs[0].String())
+				return fmt.Sprintf("set prefix to %q\n", prefix)
+			}
+			return "cleared prefix"
 		case "iter":
 			buf.Reset()
-			if lower, upper, ok := getBounds(td); ok {
-				iter.SetBounds(lower, upper)
-			}
+			lower, upper := getBounds(td)
+			iter.SetBounds(lower, upper)
 
 			lines := strings.Split(strings.TrimSpace(td.Input), "\n")
 			for _, line := range lines {
diff --git a/internal/keyspan/interleaving_iter_test.go b/internal/keyspan/interleaving_iter_test.go
index 220edecb..2ac7a85b 100644
--- a/internal/keyspan/interleaving_iter_test.go
+++ b/internal/keyspan/interleaving_iter_test.go
@@ -107,7 +107,7 @@ func runInterleavingIterTest(t *testing.T, filename string) {
 			}
 			keyspanIter.Init(cmp, noopTransform, NewIter(cmp, spans))
 			hooks.maskSuffix = nil
-			iter.Init(cmp, base.WrapIterWithStats(&pointIter), &keyspanIter, &hooks, nil, nil)
+			iter.Init(testkeys.Comparer, base.WrapIterWithStats(&pointIter), &keyspanIter, &hooks, nil, nil)
 			return "OK"
 		case "define-pointkeys":
 			var points []base.InternalKey
@@ -117,7 +117,7 @@ func runInterleavingIterTest(t *testing.T, filename string) {
 			}
 			pointIter = pointIterator{cmp: cmp, keys: points}
 			hooks.maskSuffix = nil
-			iter.Init(cmp, base.WrapIterWithStats(&pointIter), &keyspanIter, &hooks, nil, nil)
+			iter.Init(testkeys.Comparer, base.WrapIterWithStats(&pointIter), &keyspanIter, &hooks, nil, nil)
 			return "OK"
 		case "iter":
 			buf.Reset()
@@ -143,6 +143,10 @@ func runInterleavingIterTest(t *testing.T, filename string) {
 					formatKey(iter.Prev())
 				case "seek-ge":
 					formatKey(iter.SeekGE([]byte(strings.TrimSpace(line[i:])), base.SeekGEFlagsNone))
+				case "seek-prefix-ge":
+					key := []byte(strings.TrimSpace(line[i:]))
+					prefix := key[:testkeys.Comparer.Split(key)]
+					formatKey(iter.SeekPrefixGE(prefix, key, base.SeekGEFlagsNone))
 				case "seek-lt":
 					formatKey(iter.SeekLT([]byte(strings.TrimSpace(line[i:])), base.SeekLTFlagsNone))
 				case "set-bounds":
diff --git a/internal/keyspan/testdata/bounded_iter b/internal/keyspan/testdata/bounded_iter
index 59ab347d..149cf9e4 100644
--- a/internal/keyspan/testdata/bounded_iter
+++ b/internal/keyspan/testdata/bounded_iter
@@ -115,3 +115,120 @@ b-g:{(#4,RANGEKEYSET,@3)}
 <nil>
 b-g:{(#4,RANGEKEYSET,@3)}
 <nil>
+
+set-prefix bar
+----
+set prefix to "bar"
+
+# Test seeking to a portion of the keyspace that contains no range keys with
+# start bounds ≥ the seek key such that the range key also overlaps the current
+# prefix.
+
+iter lower=a upper=z
+seek-ge bar
+prev
+prev
+----
+<nil>
+b-g:{(#4,RANGEKEYSET,@3)}
+<nil>
+
+# Test seeking to a portion of the keyspace that contains a range key with a
+# start bound < the seek key, and the range key also overlaps the current
+# prefix.
+
+iter lower=a upper=z
+seek-lt bar
+next
+prev
+prev
+----
+b-g:{(#4,RANGEKEYSET,@3)}
+<nil>
+b-g:{(#4,RANGEKEYSET,@3)}
+<nil>
+
+# Test seeking with bounds narrower than the range of the seek prefix. This is
+# possible in practice because the bounded iterator iterates over fragments, not
+# pre-defragmented range keys.
+
+iter lower=bar@9 upper=bar@3
+seek-lt bar
+next
+prev
+prev
+----
+b-g:{(#4,RANGEKEYSET,@3)}
+<nil>
+b-g:{(#4,RANGEKEYSET,@3)}
+<nil>
+
+# Test a similar scenario but on the start prefix of a key.
+
+iter lower=b@9 upper=b@3
+seek-lt b
+next
+next
+prev
+prev
+----
+<nil>
+b-g:{(#4,RANGEKEYSET,@3)}
+<nil>
+b-g:{(#4,RANGEKEYSET,@3)}
+<nil>
+
+# Test a scenario where the prefix overlaps a span, but the bounds exclude it.
+
+iter lower=z@9 upper=z@3
+seek-lt z@3
+next
+----
+<nil>
+<nil>
+
+# Test many spans matching the prefix, due to fragmentation within a prefix.
+
+define
+b-boo:{(#1,RANGEKEYSET,@1)}
+c@9-c@8:{(#1,RANGEKEYSET,@1)}
+c@8-c@7:{(#1,RANGEKEYSET,@1)}
+c@7-c@6:{(#1,RANGEKEYSET,@1)}
+c@6-c@5:{(#1,RANGEKEYSET,@1)}
+c@5-c@4:{(#1,RANGEKEYSET,@1)}
+----
+
+set-prefix c
+----
+set prefix to "c"
+
+iter
+seek-lt c
+next
+next
+next
+next
+next
+next
+----
+<nil>
+c@9-c@8:{(#1,RANGEKEYSET,@1)}
+c@8-c@7:{(#1,RANGEKEYSET,@1)}
+c@7-c@6:{(#1,RANGEKEYSET,@1)}
+c@6-c@5:{(#1,RANGEKEYSET,@1)}
+c@5-c@4:{(#1,RANGEKEYSET,@1)}
+<nil>
+
+# Test the same scenario with bounds limiting iteration to a subset of the
+# keys.
+
+iter lower=c@7 upper=c@5
+seek-lt c@7
+next
+next
+next
+----
+<nil>
+c@7-c@6:{(#1,RANGEKEYSET,@1)}
+c@6-c@5:{(#1,RANGEKEYSET,@1)}
+<nil>
diff --git a/internal/keyspan/testdata/interleaving_iter b/internal/keyspan/testdata/interleaving_iter
index 0e3e3754..9f73e7c0 100644
--- a/internal/keyspan/testdata/interleaving_iter
+++ b/internal/keyspan/testdata/interleaving_iter
@@ -88,11 +88,11 @@ next
 next
 ----
 -- SpanChanged(nil)
--- SpanChanged(a-c:{(#10,RANGEKEYSET,@5,apples) (#10,RANGEKEYDEL) (#8,RANGEKEYUNSET,@1) (#4,RANGEKEYSET,@3,bananas) (#4,RANGEKEYSET,@2,oranges)})
+-- SpanChanged(b-c:{(#10,RANGEKEYSET,@5,apples) (#10,RANGEKEYDEL) (#8,RANGEKEYUNSET,@1) (#4,RANGEKEYSET,@3,bananas) (#4,RANGEKEYSET,@2,oranges)})
 PointKey: b#72057594037927935,21
 Span: b-c:{(#10,RANGEKEYSET,@5,apples) (#10,RANGEKEYDEL) (#8,RANGEKEYUNSET,@1) (#4,RANGEKEYSET,@3,bananas) (#4,RANGEKEYSET,@2,oranges)}
 -
--- SpanChanged(c-d:{(#4,RANGEKEYSET,@3,coconut)})
+-- SpanChanged(c-carrot:{(#4,RANGEKEYSET,@3,coconut)})
 PointKey: c#72057594037927935,21
 Span: c-carrot:{(#4,RANGEKEYSET,@3,coconut)}
 -
@@ -110,11 +110,11 @@ prev
 prev
 ----
 -- SpanChanged(nil)
--- SpanChanged(c-d:{(#4,RANGEKEYSET,@3,coconut)})
+-- SpanChanged(c-carrot:{(#4,RANGEKEYSET,@3,coconut)})
 PointKey: c#72057594037927935,21
 Span: c-carrot:{(#4,RANGEKEYSET,@3,coconut)}
 -
--- SpanChanged(a-c:{(#10,RANGEKEYSET,@5,apples) (#10,RANGEKEYDEL) (#8,RANGEKEYUNSET,@1) (#4,RANGEKEYSET,@3,bananas) (#4,RANGEKEYSET,@2,oranges)})
+-- SpanChanged(b-c:{(#10,RANGEKEYSET,@5,apples) (#10,RANGEKEYDEL) (#8,RANGEKEYUNSET,@1) (#4,RANGEKEYSET,@3,bananas) (#4,RANGEKEYSET,@2,oranges)})
 PointKey: b#72057594037927935,21
 Span: b-c:{(#10,RANGEKEYSET,@5,apples) (#10,RANGEKEYDEL) (#8,RANGEKEYUNSET,@1) (#4,RANGEKEYSET,@3,bananas) (#4,RANGEKEYSET,@2,oranges)}
 -
@@ -719,7 +719,7 @@ prev
 prev
 ----
 -- SpanChanged(nil)
--- SpanChanged(a-z:{(#5,RANGEKEYSET,@1,v1)})
+-- SpanChanged(a-c:{(#5,RANGEKEYSET,@1,v1)})
 PointKey: a#72057594037927935,21
 Span: a-c:{(#5,RANGEKEYSET,@1,v1)}
 -
@@ -728,7 +728,7 @@ Span: a-c:{(#5,RANGEKEYSET,@1,v1)}
 PointKey: z#8,1
 Span: <invalid>
 -
--- SpanChanged(a-z:{(#5,RANGEKEYSET,@1,v1)})
+-- SpanChanged(c-z:{(#5,RANGEKEYSET,@1,v1)})
 PointKey: c#72057594037927935,21
 Span: c-z:{(#5,RANGEKEYSET,@1,v1)}
 -
@@ -887,3 +887,39 @@ seek-lt d
 -- SpanChanged(nil)
 -- SpanChanged(nil)
 .
+
+# Test seek-prefix-ge and its truncation of bounds to the prefix's bounds.
+
+define-rangekeys
+b-d:{(#5,RANGEKEYSET,@1,foo)}
+f-g:{(#6,RANGEKEYSET,@1,foo)}
+----
+OK
+
+define-pointkeys
+c.SET.8
+----
+OK
+
+iter
+seek-prefix-ge b
+next
+seek-prefix-ge c
+next
+----
+-- SpanChanged(nil)
+-- SpanChanged(b-b\x00:{(#5,RANGEKEYSET,@1,foo)})
+PointKey: b#72057594037927935,21
+Span: b-b\x00:{(#5,RANGEKEYSET,@1,foo)}
+-
+PointKey: c#8,1
+Span: b-b\x00:{(#5,RANGEKEYSET,@1,foo)}
+-
+-- SpanChanged(nil)
+-- SpanChanged(c-c\x00:{(#5,RANGEKEYSET,@1,foo)})
+PointKey: c#72057594037927935,21
+Span: c-c\x00:{(#5,RANGEKEYSET,@1,foo)}
+-
+PointKey: c#8,1
+Span: c-c\x00:{(#5,RANGEKEYSET,@1,foo)}
+-
diff --git a/internal/rangekey/coalesce_test.go b/internal/rangekey/coalesce_test.go
index 4469686f..7c765f14 100644
--- a/internal/rangekey/coalesce_test.go
+++ b/internal/rangekey/coalesce_test.go
@@ -122,6 +122,8 @@ func TestDefragmenting(t *testing.T) {
 
 	var buf bytes.Buffer
 	var spans []keyspan.Span
+	var hasPrefix bool
+	var prefix []byte
 	datadriven.RunTest(t, "testdata/defragmenting_iter", func(td *datadriven.TestData) string {
 		buf.Reset()
 		switch td.Cmd {
@@ -134,8 +136,10 @@ func TestDefragmenting(t *testing.T) {
 			return ""
 		case "iter":
 			var userIterCfg UserIteratorConfig
-			iter := userIterCfg.Init(cmp, base.InternalKeySeqNumMax,
-				nil /* lower */, nil /* upper */, keyspan.NewIter(cmp, spans))
+			iter := userIterCfg.Init(testkeys.Comparer, base.InternalKeySeqNumMax,
+				nil /* lower */, nil, /* upper */
+				&hasPrefix, &prefix,
+				keyspan.NewIter(cmp, spans))
 			for _, line := range strings.Split(td.Input, "\n") {
 				runIterOp(&buf, iter, line)
 			}
@@ -214,10 +218,14 @@ func testDefragmentingIteRandomizedOnce(t *testing.T, seed int64) {
 	fragmented = fragment(cmp, formatKey, fragmented)
 
 	var referenceCfg, fragmentedCfg UserIteratorConfig
-	referenceIter := referenceCfg.Init(cmp, base.InternalKeySeqNumMax,
-		nil /* lower */, nil /* upper */, keyspan.NewIter(cmp, original))
-	fragmentedIter := fragmentedCfg.Init(cmp, base.InternalKeySeqNumMax,
-		nil /* lower */, nil /* upper */, keyspan.NewIter(cmp, fragmented))
+	referenceIter := referenceCfg.Init(testkeys.Comparer, base.InternalKeySeqNumMax,
+		nil /* lower */, nil, /* upper */
+		new(bool), new([]byte),
+		keyspan.NewIter(cmp, original))
+	fragmentedIter := fragmentedCfg.Init(testkeys.Comparer, base.InternalKeySeqNumMax,
+		nil /* lower */, nil, /* upper */
+		new(bool), new([]byte),
+		keyspan.NewIter(cmp, fragmented))
 
 	// Generate 100 random operations and run them against both iterators.
 	const numIterOps = 100
diff --git a/internal/testkeys/testkeys.go b/internal/testkeys/testkeys.go
index 1753d190..0ea38a44 100644
--- a/internal/testkeys/testkeys.go
+++ b/internal/testkeys/testkeys.go
@@ -76,6 +76,15 @@ var Comparer *base.Comparer = &base.Comparer{
 		// The successor is > a[:ai], so we only need to add the sentinel.
 		return append(dst, 0)
 	},
+	ImmediateSuccessor: func(dst, a []byte) []byte {
+		// TODO(jackson): Consider changing this Comparer to only support
+		// representable prefix keys containing characters a-z.
+		ai := split(a)
+		if ai != len(a) {
+			panic("pebble: ImmediateSuccessor invoked with a non-prefix key")
+		}
+		return append(append(dst, a...), 0x00)
+	},
 	Split: split,
 	Name:  "pebble.internal.testkeys",
 }
diff --git a/iterator_test.go b/iterator_test.go
index 46354304..bca7cd5a 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -509,12 +509,13 @@ func TestIterator(t *testing.T) {
 			return merge(key, value)
 		}
 		return &Iterator{
-			opts:  opts,
-			cmp:   cmp,
-			equal: equal,
-			split: split,
-			merge: wrappedMerge,
-			iter:  newInvalidatingIter(iter),
+			opts:     opts,
+			cmp:      cmp,
+			comparer: DefaultComparer,
+			equal:    equal,
+			split:    split,
+			merge:    wrappedMerge,
+			iter:     newInvalidatingIter(iter),
 		}
 	}
 
@@ -1191,12 +1192,13 @@ func TestIteratorSeekOptErrors(t *testing.T) {
 		// NB: This Iterator cannot be cloned since it is not constructed
 		// with a readState. It suffices for this test.
 		return &Iterator{
-			opts:  opts,
-			cmp:   cmp,
-			equal: equal,
-			split: split,
-			merge: DefaultMerger.Merge,
-			iter:  base.WrapIterWithStats(&errorIter),
+			opts:     opts,
+			cmp:      cmp,
+			comparer: DefaultComparer,
+			equal:    equal,
+			split:    split,
+			merge:    DefaultMerger.Merge,
+			iter:     base.WrapIterWithStats(&errorIter),
 		}
 	}
 
@@ -2034,11 +2036,12 @@ func BenchmarkIteratorSeekGENoop(b *testing.B) {
 		b.Run(fmt.Sprintf("withLimit=%t", withLimit), func(b *testing.B) {
 			m := buildMergingIter(readers, levelSlices)
 			iter := Iterator{
-				cmp:   DefaultComparer.Compare,
-				equal: DefaultComparer.Equal,
-				split: DefaultComparer.Split,
-				merge: DefaultMerger.Merge,
-				iter:  m,
+				cmp:      DefaultComparer.Compare,
+				equal:    DefaultComparer.Equal,
+				split:    DefaultComparer.Split,
+				comparer: DefaultComparer,
+				merge:    DefaultMerger.Merge,
+				iter:     m,
 			}
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
@@ -2356,7 +2359,53 @@ func BenchmarkCombinedIteratorSeek(b *testing.B) {
 // range key that's fragmented across hundreds of files. The iterator bounds
 // should prevent defragmenting beyond the iterator's bounds.
 func BenchmarkCombinedIteratorSeek_Bounded(b *testing.B) {
-	rng := rand.New(rand.NewSource(uint64(1658872515083979000)))
+	d, keys := buildFragmentedRangeKey(b, uint64(1658872515083979000))
+
+	var lower = len(keys) / 2
+	var upper = len(keys)/2 + len(keys)/20 // 5%
+	iterOpts := IterOptions{
+		KeyTypes:   IterKeyTypePointsAndRanges,
+		LowerBound: keys[lower],
+		UpperBound: keys[upper],
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		it := d.NewIter(&iterOpts)
+		for j := lower; j < upper; j++ {
+			if !it.SeekGE(keys[j]) {
+				b.Errorf("key %q missing", keys[j])
+			}
+		}
+		require.NoError(b, it.Close())
+	}
+}
+
+// BenchmarkCombinedIteratorSeekPrefix benchmarks an iterator that
+// performs repeated prefix seeks over 5% of the middle of a keyspace covered by a
+// range key that's fragmented across hundreds of files. The seek prefix should
+// avoid defragmenting beyond the seek prefixes.
+func BenchmarkCombinedIteratorSeekPrefix(b *testing.B) {
+	d, keys := buildFragmentedRangeKey(b, uint64(1658872515083979000))
+
+	var lower = len(keys) / 2
+	var upper = len(keys)/2 + len(keys)/20 // 5%
+	iterOpts := IterOptions{
+		KeyTypes: IterKeyTypePointsAndRanges,
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		it := d.NewIter(&iterOpts)
+		for j := lower; j < upper; j++ {
+			if !it.SeekPrefixGE(keys[j]) {
+				b.Errorf("key %q missing", keys[j])
+			}
+		}
+		require.NoError(b, it.Close())
+	}
+}
+
+func buildFragmentedRangeKey(b testing.TB, seed uint64) (d *DB, keys [][]byte) {
+	rng := rand.New(rand.NewSource(seed))
 	ks := testkeys.Alpha(2)
 	opts := &Options{
 		FS:                        vfs.NewMem(),
@@ -2368,11 +2417,11 @@ func BenchmarkCombinedIteratorSeek_Bounded(b *testing.B) {
 	for l := 0; l < len(opts.Levels); l++ {
 		opts.Levels[l].TargetFileSize = 1
 	}
-	d, err := Open("", opts)
+	var err error
+	d, err = Open("", opts)
 	require.NoError(b, err)
-	defer func() { require.NoError(b, d.Close()) }()
 
-	keys := make([][]byte, ks.Count())
+	keys = make([][]byte, ks.Count())
 	for i := 0; i < ks.Count(); i++ {
 		keys[i] = testkeys.Key(ks, i)
 	}
@@ -2394,22 +2443,5 @@ func BenchmarkCombinedIteratorSeek_Bounded(b *testing.B) {
 	d.mu.Unlock()
 	require.GreaterOrEqualf(b, v.Levels[numLevels-1].Len(),
 		700, "expect many (≥700) L6 files but found %d", v.Levels[numLevels-1].Len())
-
-	var lower = ks.Count() / 2
-	var upper = ks.Count()/2 + ks.Count()/20 // 5%
-	iterOpts := IterOptions{
-		KeyTypes:   IterKeyTypePointsAndRanges,
-		LowerBound: keys[lower],
-		UpperBound: keys[upper],
-	}
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		it := d.NewIter(&iterOpts)
-		for j := lower; j < upper; j++ {
-			if !it.SeekGE(keys[j]) {
-				b.Errorf("key %q missing", keys[j])
-			}
-		}
-		require.NoError(b, it.Close())
-	}
+	return d, keys
 }
diff --git a/range_keys_test.go b/range_keys_test.go
index e1061058..64cc91df 100644
--- a/range_keys_test.go
+++ b/range_keys_test.go
@@ -114,6 +114,12 @@ func TestRangeKeys(t *testing.T) {
 			return fmt.Sprintf("created indexed batch with %d keys\n", count)
 		case "lsm":
 			return runLSMCmd(td, d)
+		case "metrics":
+			d.mu.Lock()
+			d.waitTableStats()
+			d.mu.Unlock()
+			m := d.Metrics()
+			return fmt.Sprintf("Metrics.Keys.RangeKeySetsCount = %d\n", m.Keys.RangeKeySetsCount)
 		case "commit-batch":
 			if b == nil {
 				return "no pending batch"
diff --git a/testdata/rangekeys b/testdata/rangekeys
index fc49b35a..70ed5689 100644
--- a/testdata/rangekeys
+++ b/testdata/rangekeys
@@ -242,11 +242,13 @@ b: (b2, [ace-c) @3=beep)
 c: (c2, [c-d) @1=boop UPDATED)
 .
 
+# NB: seek-prefix-ge truncates bounds to the prefix.
+
 combined-iter
 seek-prefix-ge b
 next
 ----
-b: (b2, [ace-c) @3=beep UPDATED)
+b: (b2, [b-"b\x00") @3=beep UPDATED)
 .
 
 reset
@@ -255,7 +257,7 @@ reset
 batch
 range-key-set a d @8 boop
 set a@2 a@2
-set a@3 a#3
+set a@3 a@3
 set a@9 a@9
 set a@10 a@10
 set b b
@@ -270,13 +272,15 @@ next
 next
 next
 ----
-a: (., [a-d) @8=boop UPDATED)
-a@10: (a@10, [a-d) @8=boop)
-a@9: (a@9, [a-d) @8=boop)
-a@3: (a#3, [a-d) @8=boop)
-a@2: (a@2, [a-d) @8=boop)
+a: (., [a-"a\x00") @8=boop UPDATED)
+a@10: (a@10, [a-"a\x00") @8=boop)
+a@9: (a@9, [a-"a\x00") @8=boop)
+a@3: (a@3, [a-"a\x00") @8=boop)
+a@2: (a@2, [a-"a\x00") @8=boop)
 .
 
+
+
 # Perform the above iteration with range-key masking enabled at a suffix equal
 # to the range key's. The [a,d)@8 range key should serve as a masking, obscuring
 # the points a@3 and a@2.
@@ -287,9 +291,9 @@ next
 next
 next
 ----
-a: (., [a-d) @8=boop UPDATED)
-a@10: (a@10, [a-d) @8=boop)
-a@9: (a@9, [a-d) @8=boop)
+a: (., [a-"a\x00") @8=boop UPDATED)
+a@10: (a@10, [a-"a\x00") @8=boop)
+a@9: (a@9, [a-"a\x00") @8=boop)
 .
 
 # Perform the same thing but with a mask suffix below the range key's. All the
@@ -310,18 +314,30 @@ next
 next
 next
 ----
-a: (., [a-d) @8=boop UPDATED)
-a@10: (a@10, [a-d) @8=boop)
-a@9: (a@9, [a-d) @8=boop)
-a@3: (a#3, [a-d) @8=boop)
-a@2: (a@2, [a-d) @8=boop)
+a: (., [a-"a\x00") @8=boop UPDATED)
+a@10: (a@10, [a-"a\x00") @8=boop)
+a@9: (a@9, [a-"a\x00") @8=boop)
+a@3: (a@3, [a-"a\x00") @8=boop)
+a@2: (a@2, [a-"a\x00") @8=boop)
 .
 .
-a: (., [a-d) @8=boop UPDATED)
-a@10: (a@10, [a-d) @8=boop)
-a@9: (a@9, [a-d) @8=boop)
+a: (., [a-"a\x00") @8=boop UPDATED)
+a@10: (a@10, [a-"a\x00") @8=boop)
+a@9: (a@9, [a-"a\x00") @8=boop)
 .
 
+# Test that switching out of prefix iteration correctly expands the bounds
+# beyond the scope of the previous prefix.
+
+combined-iter
+seek-prefix-ge a
+next
+seek-ge a@3
+----
+a: (., [a-"a\x00") @8=boop UPDATED)
+a@10: (a@10, [a-"a\x00") @8=boop)
+a@3: (a@3, [a-d) @8=boop UPDATED)
+
 reset
 ----
 
@@ -342,6 +358,10 @@ wrote 3 keys
 flush
 ----
 
+metrics
+----
+Metrics.Keys.RangeKeySetsCount = 1
+
 combined-iter
 seek-ge az
 next
@@ -412,6 +432,13 @@ wrote 4 keys
 flush
 ----
 
+lsm
+----
+0.1:
+  000008:[b#2111,RANGEKEYSET-z#72057594037927935,RANGEKEYSET]
+0.0:
+  000006:[a@100#3,SET-zz@1#2104,SET]
+
 scan-rangekeys
 ----
 [b, bat)
@@ -423,14 +450,23 @@ scan-rangekeys
 [e, z)
  @1000=boop
 
+# NB: There are now 8 range key sets in the database. See the 7 range keys in
+# the above scan-rangekeys. Additionally, the sstable flushed earlier up above
+# included a rangekeyset [b,c) @5=beep.
+
+metrics
+----
+Metrics.Keys.RangeKeySetsCount = 8
+
+
 combined-iter
 seek-prefix-ge ca
 next
 seek-prefix-ge ca@100
 ----
-ca: (., [c-e) @1000=boop, @1=bop UPDATED)
-ca@100: (ca@100, [c-e) @1000=boop, @1=bop)
-ca@100: (ca@100, [c-e) @1000=boop, @1=bop)
+ca: (., [ca-"ca\x00") @1000=boop, @1=bop UPDATED)
+ca@100: (ca@100, [ca-"ca\x00") @1000=boop, @1=bop)
+ca@100: (ca@100, [ca-"ca\x00") @1000=boop, @1=bop)
 
 
 # Perform the same iteration as above, but with @1000 range-key masking. The
@@ -441,9 +477,9 @@ seek-prefix-ge ca
 next
 seek-prefix-ge ca@100
 ----
-ca: (., [c-e) @1000=boop, @1=bop UPDATED)
+ca: (., [ca-"ca\x00") @1000=boop, @1=bop UPDATED)
 .
-ca@100: (., [c-e) @1000=boop, @1=bop UPDATED)
+ca@100: (., [ca-"ca\x00") @1000=boop, @1=bop UPDATED)
 
 # Test masked, non-prefixed iteration. We should see the range keys, but all the
 # points should be masked except those beginning with z which were excluded by
@@ -660,6 +696,13 @@ b:b
 c:c
 .
 
+flush
+----
+
+metrics
+----
+Metrics.Keys.RangeKeySetsCount = 3
+
 # Test Prev-ing back over a synthetic range key marker. Synthetic range-key
 # markers (the keys interleaved at 'c' during a SeekGE(c) when there's a
 # straddling range key) are ephemeral, and Prev-ing back must move back the