Repository: cockroachdb/pebble
Pull request #1567 (2022-03-09 16:34:48)

internal/metamorphic: add range key iteration
**db: fix determinism of limited reverse iteration with range keys**

During limited reverse iteration, the iterator may exceed a limit that is
straddled by a range key. Previously, to ensure that the iterator exposes range
keys that covers the keyspace within the limit, the iterator would not pause at
a limit if there was a covering a range key.

This condition was too broad, because the range key might not cover the
keyspace within the limit. This introduced nondeterminism. An iterator over one
LSM state could stop at a deleted point key beyond the limit but before the
next range key. An iterator over an equivalent LSM state could proceed all the
way to the next range key and surface it without pausing.

**db: move range keys from memtable to global arena atomically**

Previously, when a flush completed, the flush first updated the read state and
then copied range keys from flushed memtables into the global arena. An
iterator that acquired a read state (an operation that does not require D.mu)
and read from it between these two steps would fail to see the flushed range
keys.

This commit adds a temporary closure parameter to d.updateReadStateLocked that
is invoked while holding an exclusive lock on the read state. This ensures the
all readers observe the flushing range keys as either within a memtable or
within the global arena.

**db: avoid levelIter invariant violation on exhausted mergingIter**

The commit b34dbe97d33432dad78f9404983ebeb2cacb7560 solved an issue where the
levelIter's bounds invariants could be violated by cascading tombstone seeks.
That commit solved the issue only for levels that were not the current heap
root. This was sufficient because if the current heap root was exhausted, the
mergingIter was exhausted, and the pebble.Iterator would seek the merging
iterator to reposition it.

The range key interleaving iterator however had different behavior. It expected
to always be able to Next/Prev an exhausted merging iterator onto the

**internal/rangekey: seek exhausted iters when switching directions**

When a point key iterator is exhausted, seek it when switching directions. This
matches pebble.Iterator's treatment of these cases and ensures the range
tombstone cascading seek optimization applies.

**internal/metamorphic: add range key iteration**

Configure the metamorphic NewIter operation to randomly create iterators over
just points, just range keys, or points and range keys. When an iterator is
configured to iterate over both point and range keys, randomly enable range key
masking as well.

This required a few tweaks outside of the metamorphic tests. Previously, during
Open while replaying the WALs, any range keys written to the WAL were dropped
since there's a different code path than ordinary flushes. Also, the
metamorphic tests test some DB restart operations as hard crashes, using the
MemFS's SetIgnoreSyncs method to revert the DB to pre-Close state. This would
cause some WAL replays to fail when adding a range key second time, because the
original keys were not reverted to pre-Close state.


Patch:
diff --git a/Makefile b/Makefile
index 8c3887aa..99a86190 100644
--- a/Makefile
+++ b/Makefile
@@ -22,7 +22,7 @@ test:
 	${GO} test -mod=vendor -tags '$(TAGS)' ${testflags} -run ${TESTS} ${PKG}
 
 .PHONY: testrace
-testrace: testflags += -race
+testrace: testflags += -race -timeout 20m
 testrace: test
 
 .PHONY: stress stressrace
diff --git a/batch.go b/batch.go
index f652375c..a19b67a8 100644
--- a/batch.go
+++ b/batch.go
@@ -1747,13 +1747,19 @@ func (i flushFlushableBatchIter) valueSize() uint64 {
 // batchSort returns iterators for the sorted contents of the batch. It is
 // intended for testing use only. The batch.Sort dance is done to prevent
 // exposing this method in the public pebble interface.
-func batchSort(i interface{}) (internalIterator, internalIterator) {
+func batchSort(
+	i interface{},
+) (
+	points internalIterator,
+	rangeDels keyspan.FragmentIterator,
+	rangeKeys keyspan.FragmentIterator,
+) {
 	b := i.(*Batch)
 	if b.Indexed() {
-		return b.newInternalIter(nil), b.newRangeDelIter(nil)
+		return b.newInternalIter(nil), b.newRangeDelIter(nil), b.newRangeKeyIter(nil)
 	}
 	f := newFlushableBatch(b, b.db.opts.Comparer)
-	return f.newIter(nil), f.newRangeDelIter(nil)
+	return f.newIter(nil), f.newRangeDelIter(nil), f.newRangeKeyIter(nil)
 }
 
 func init() {
diff --git a/compaction.go b/compaction.go
index c50bebf6..07aada7a 100644
--- a/compaction.go
+++ b/compaction.go
@@ -1457,13 +1457,13 @@ func (d *DB) flush1() error {
 	if err == nil {
 		flushed = d.mu.mem.queue[:n]
 		d.mu.mem.queue = d.mu.mem.queue[n:]
-		d.updateReadStateLocked(d.opts.DebugCheck)
+		d.updateReadStateLocked(d.opts.DebugCheck, func() {
+			// TODO(jackson): Remove this, plus this updateReadStateLocked
+			// parameter when range keys are persisted to sstables.
+			err = d.applyFlushedRangeKeys(flushed)
+		})
 		d.updateTableStatsLocked(ve.NewFiles)
 	}
-	if err == nil {
-		// TODO(jackson): Remove this when range keys are persisted to sstables.
-		err = d.applyFlushedRangeKeys(flushed)
-	}
 	// Signal FlushEnd after installing the new readState. This helps for unit
 	// tests that use the callback to trigger a read using an iterator with
 	// IterOptions.OnlyReadGuaranteedDurable.
@@ -1892,7 +1892,7 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 	// there are no references obsolete tables will be added to the obsolete
 	// table list.
 	if err == nil {
-		d.updateReadStateLocked(d.opts.DebugCheck)
+		d.updateReadStateLocked(d.opts.DebugCheck, nil)
 		d.updateTableStatsLocked(ve.NewFiles)
 	}
 	d.deleteObsoleteFiles(jobID, true /* waitForOngoing */)
diff --git a/db.go b/db.go
index 47c9f872..6426a0f9 100644
--- a/db.go
+++ b/db.go
@@ -1880,7 +1880,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 		var entry *flushableEntry
 		d.mu.mem.mutable, entry = d.newMemTable(newLogNum, logSeqNum)
 		d.mu.mem.queue = append(d.mu.mem.queue, entry)
-		d.updateReadStateLocked(nil)
+		d.updateReadStateLocked(nil, nil)
 		if immMem.writerUnref() {
 			d.maybeScheduleFlush()
 		}
diff --git a/flush_external.go b/flush_external.go
index 179beb21..8bda7b97 100644
--- a/flush_external.go
+++ b/flush_external.go
@@ -85,7 +85,7 @@ func flushExternalTable(untypedDB interface{}, path string, originalMeta *fileMe
 		}
 		return err
 	}
-	d.updateReadStateLocked(d.opts.DebugCheck)
+	d.updateReadStateLocked(d.opts.DebugCheck, nil)
 	d.updateTableStatsLocked(ve.NewFiles)
 	d.deleteObsoleteFiles(jobID, true /* waitForOngoing */)
 	d.maybeScheduleCompaction()
diff --git a/ingest.go b/ingest.go
index c1ebf07b..73e697d3 100644
--- a/ingest.go
+++ b/ingest.go
@@ -769,7 +769,7 @@ func (d *DB) ingestApply(
 	}); err != nil {
 		return nil, err
 	}
-	d.updateReadStateLocked(d.opts.DebugCheck)
+	d.updateReadStateLocked(d.opts.DebugCheck, nil)
 	d.updateTableStatsLocked(ve.NewFiles)
 	d.deleteObsoleteFiles(jobID, false /* waitForOngoing */)
 	// The ingestion may have pushed a level over the threshold for compaction,
diff --git a/internal/manifest/level_metadata.go b/internal/manifest/level_metadata.go
index 6e7f0f6c..9131e697 100644
--- a/internal/manifest/level_metadata.go
+++ b/internal/manifest/level_metadata.go
@@ -260,12 +260,30 @@ func (ls LevelSlice) Reslice(resliceFunc func(start, end *LevelIterator)) LevelS
 	return s
 }
 
+// KeyType is used to specify the type of keys we're looking for in
+// LevelIterator positioning operations. Files not containing any keys of the
+// desired type are skipped.
+type KeyType int8
+
+const (
+	// KeyTypePointAndRange denotes a search among the entire keyspace, including
+	// both point keys and range keys. No sstables are skipped.
+	KeyTypePointAndRange KeyType = iota
+	// KeyTypePoint denotes a search among the point keyspace. SSTables with no
+	// point keys will be skipped. Note that the point keyspace includes rangedels.
+	KeyTypePoint
+	// KeyTypeRange denotes a search among the range keyspace. SSTables with no
+	// range keys will be skipped.
+	KeyTypeRange
+)
+
 // LevelIterator iterates over a set of files' metadata. Its zero value is an
 // empty iterator.
 type LevelIterator struct {
-	iter  iterator
-	start *iterator
-	end   *iterator
+	iter   iterator
+	start  *iterator
+	end    *iterator
+	filter KeyType
 }
 
 func (i LevelIterator) String() string {
@@ -315,9 +333,10 @@ func (i *LevelIterator) Clone() LevelIterator {
 	// The start and end iterators are not cloned and are treated as
 	// immutable.
 	return LevelIterator{
-		iter:  i.iter.clone(),
-		start: i.start,
-		end:   i.end,
+		iter:   i.iter.clone(),
+		start:  i.start,
+		end:    i.end,
+		filter: i.filter,
 	}
 }
 
@@ -333,6 +352,14 @@ func (i *LevelIterator) empty() bool {
 	return emptyWithBounds(i.iter, i.start, i.end)
 }
 
+// Filter clones the iterator and sets the desired KeyType as the key to filter
+// files on.
+func (i *LevelIterator) Filter(keyType KeyType) LevelIterator {
+	l := i.Clone()
+	l.filter = keyType
+	return l
+}
+
 func emptyWithBounds(i iterator, start, end *iterator) bool {
 	// If i.r is nil, the iterator was constructed from an empty btree.
 	// If the end bound is before the start bound, the bounds represent an
@@ -353,7 +380,7 @@ func (i *LevelIterator) First() *FileMetadata {
 	if !i.iter.valid() {
 		return nil
 	}
-	return i.iter.cur()
+	return i.filteredNextFile(i.iter.cur())
 }
 
 // Last seeks to the last file in the iterator and returns it.
@@ -369,7 +396,7 @@ func (i *LevelIterator) Last() *FileMetadata {
 	if !i.iter.valid() {
 		return nil
 	}
-	return i.iter.cur()
+	return i.filteredPrevFile(i.iter.cur())
 }
 
 // Next advances the iterator to the next file and returns it.
@@ -381,7 +408,7 @@ func (i *LevelIterator) Next() *FileMetadata {
 	if i.end != nil && cmpIter(i.iter, *i.end) > 0 {
 		return nil
 	}
-	return i.iter.cur()
+	return i.filteredNextFile(i.iter.cur())
 }
 
 // Prev moves the iterator the previous file and returns it.
@@ -393,7 +420,7 @@ func (i *LevelIterator) Prev() *FileMetadata {
 	if i.start != nil && cmpIter(i.iter, *i.start) < 0 {
 		return nil
 	}
-	return i.iter.cur()
+	return i.filteredPrevFile(i.iter.cur())
 }
 
 // SeekGE seeks to the first file in the iterator's file set with a largest
@@ -405,9 +432,10 @@ func (i *LevelIterator) SeekGE(cmp Compare, userKey []byte) *FileMetadata {
 	if i.empty() {
 		return nil
 	}
-	return i.seek(func(m *FileMetadata) bool {
+	meta := i.seek(func(m *FileMetadata) bool {
 		return cmp(m.Largest.UserKey, userKey) >= 0
 	})
+	return i.filteredNextFile(meta)
 }
 
 // SeekLT seeks to the last file in the iterator's file set with a smallest
@@ -422,7 +450,50 @@ func (i *LevelIterator) SeekLT(cmp Compare, userKey []byte) *FileMetadata {
 	i.seek(func(m *FileMetadata) bool {
 		return cmp(m.Smallest.UserKey, userKey) >= 0
 	})
-	return i.Prev()
+	meta := i.Prev()
+	return i.filteredPrevFile(meta)
+}
+
+func (i *LevelIterator) filteredNextFile(meta *FileMetadata) *FileMetadata {
+	switch i.filter {
+	case KeyTypePoint:
+		for meta != nil && !meta.HasPointKeys {
+			meta = i.Next()
+		}
+		return meta
+	case KeyTypeRange:
+		// TODO(bilal): Range keys are expected to be rare and sparse. Add an
+		// optimization to annotate the tree and efficiently skip over files that
+		// do not contain range keys right at the seek step, to reduce iterations
+		// here.
+		for meta != nil && !meta.HasRangeKeys {
+			meta = i.Next()
+		}
+		return meta
+	default:
+		return meta
+	}
+}
+
+func (i *LevelIterator) filteredPrevFile(meta *FileMetadata) *FileMetadata {
+	switch i.filter {
+	case KeyTypePoint:
+		for meta != nil && !meta.HasPointKeys {
+			meta = i.Prev()
+		}
+		return meta
+	case KeyTypeRange:
+		// TODO(bilal): Range keys are expected to be rare and sparse. Add an
+		// optimization to annotate the tree and efficiently skip over files that
+		// do not contain range keys right at the seek step, to reduce iterations
+		// here.
+		for meta != nil && !meta.HasRangeKeys {
+			meta = i.Prev()
+		}
+		return meta
+	default:
+		return meta
+	}
 }
 
 func (i *LevelIterator) seek(fn func(*FileMetadata) bool) *FileMetadata {
diff --git a/internal/metamorphic/config.go b/internal/metamorphic/config.go
index a8d671a9..6f3e3357 100644
--- a/internal/metamorphic/config.go
+++ b/internal/metamorphic/config.go
@@ -57,10 +57,10 @@ type config struct {
 	// the generated key uses a new key prefix rather than an existing prefix
 	// with a suffix.
 	newPrefix float64
-	// suffixDist defines the distribution of key suffixes. It's a dynamic
-	// randvar to roughly emulate workloads with MVCC timestamps, skewing
-	// towards most recent timestamps.
-	suffixDist randvar.Dynamic
+	// writeSuffixDist defines the distribution of key suffixes during writing.
+	// It's a dynamic randvar to roughly emulate workloads with MVCC timestamps,
+	// skewing towards most recent timestamps.
+	writeSuffixDist randvar.Dynamic
 
 	// TODO(peter): unimplemented
 	// keyDist        randvar.Dynamic
@@ -106,9 +106,9 @@ func defaultConfig() config {
 			writerDeleteRange:    50,
 			writerIngest:         100,
 			writerMerge:          100,
-			writerRangeKeySet:    20,
-			writerRangeKeyUnset:  20,
-			writerRangeKeyDelete: 10,
+			writerRangeKeySet:    10,
+			writerRangeKeyUnset:  10,
+			writerRangeKeyDelete: 5,
 			writerSet:            100,
 			writerSingleDelete:   50,
 		},
@@ -118,7 +118,7 @@ func defaultConfig() config {
 		// Use a skewed distribution of suffixes to mimic MVCC timestamps. The
 		// range will be widened whenever a suffix is found to already be in use
 		// for a particular prefix.
-		suffixDist: mustDynamic(randvar.NewSkewedLatest(0, 1, 0.99)),
+		writeSuffixDist: mustDynamic(randvar.NewSkewedLatest(0, 1, 0.99)),
 	}
 }
 
diff --git a/internal/metamorphic/generator.go b/internal/metamorphic/generator.go
index ca507c16..183eb53b 100644
--- a/internal/metamorphic/generator.go
+++ b/internal/metamorphic/generator.go
@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"fmt"
 
+	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/internal/randvar"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"golang.org/x/exp/rand"
@@ -183,21 +184,34 @@ func (g generator) randPrefixToWrite(newPrefix float64) []byte {
 	}
 }
 
-// randSuffix generates a random suffix according to the configuration's suffix
+// randSuffixToWrite generates a random suffix according to the configuration's suffix
 // distribution. It takes a probability 0 ≤ p ≤ 1.0 indicating the probability
 // with which the generator should increase the max suffix generated by the
 // generator.
 //
-// randSuffix may return a nil suffix, with the probability the configuration's
-// suffix distribution assigns to the zero suffix.
-func (g *generator) randSuffix(incMaxProb float64) []byte {
+// randSuffixToWrite may return a nil suffix, with the probability the
+// configuration's suffix distribution assigns to the zero suffix.
+func (g *generator) randSuffixToWrite(incMaxProb float64) []byte {
 	if g.rng.Float64() < incMaxProb {
-		g.cfg.suffixDist.IncMax(1)
+		g.cfg.writeSuffixDist.IncMax(1)
 	}
-	suffix := int(g.cfg.suffixDist.Uint64(g.rng))
+	return suffixFromInt(int(g.cfg.writeSuffixDist.Uint64(g.rng)))
+}
+
+// randSuffixToRead generates a random suffix used during reads. The suffixes
+// generated by this function are within the same range as suffixes generated by
+// randSuffixToWrite, however randSuffixToRead pulls from a uniform
+// distribution.
+func (g *generator) randSuffixToRead() []byte {
+	// When reading, don't apply the recency skewing in order to better exercise
+	// a reading a mix of older and newer keys.
+	max := g.cfg.writeSuffixDist.Max()
+	return suffixFromInt(int(g.rng.Uint64n(max)))
+}
 
-	// Treat zero as no suffix to match the behavior during point key generation
-	// in randKeyHelper.
+func suffixFromInt(suffix int) []byte {
+	// Treat the zero as no suffix to match the behavior during point key
+	// generation in randKeyHelper.
 	if suffix == 0 {
 		return nil
 	}
@@ -232,7 +246,7 @@ func (g *generator) randKeyHelper(keys [][]byte, newKey float64) []byte {
 			// Pick a prefix on each iteration in case most or all suffixes are
 			// already in use for any individual prefix.
 			p := g.rng.Intn(len(prefixes))
-			suffix := int(g.cfg.suffixDist.Uint64(g.rng))
+			suffix := int(g.cfg.writeSuffixDist.Uint64(g.rng))
 
 			if suffix > 0 {
 				key = resizeBuffer(key, len(prefixes[p]), testkeys.SuffixLen(suffix))
@@ -249,12 +263,12 @@ func (g *generator) randKeyHelper(keys [][]byte, newKey float64) []byte {
 			// If the generated key already existed, increase the suffix
 			// distribution to make a generating a new user key with an existing
 			// prefix more likely.
-			g.cfg.suffixDist.IncMax(1)
+			g.cfg.writeSuffixDist.IncMax(1)
 		}
 
 	default:
 		// Use a new prefix, producing a new user key.
-		suffix := int(g.cfg.suffixDist.Uint64(g.rng))
+		suffix := int(g.cfg.writeSuffixDist.Uint64(g.rng))
 		var key []byte
 		for {
 			key = g.randKeyHelperSuffix(nil, 4, 12, suffix)
@@ -481,11 +495,30 @@ func (g *generator) newIter() {
 		lower, upper = upper, lower
 	}
 
+	// Iterate over different key types.
+	var keyTypes uint32
+	var maskSuffix []byte
+	p := g.rng.Float64()
+	switch {
+	case p < 0.2: // 20% probability
+		keyTypes = uint32(pebble.IterKeyTypePointsOnly)
+	case p < 0.8: // 60% probability
+		keyTypes = uint32(pebble.IterKeyTypePointsAndRanges)
+		// With 50% probability, enable masking.
+		if g.rng.Intn(2) == 1 {
+			maskSuffix = g.randSuffixToRead()
+		}
+	default: // 20% probability
+		keyTypes = uint32(pebble.IterKeyTypeRangesOnly)
+	}
+
 	g.add(&newIterOp{
-		readerID: readerID,
-		iterID:   iterID,
-		lower:    lower,
-		upper:    upper,
+		readerID:           readerID,
+		iterID:             iterID,
+		lower:              lower,
+		upper:              upper,
+		keyTypes:           keyTypes,
+		rangeKeyMaskSuffix: maskSuffix,
 	})
 }
 
@@ -912,7 +945,7 @@ func (g *generator) writerRangeKeySet() {
 	var suffix []byte
 	if g.rng.Float64() < 0.90 {
 		// Increase the max suffix 5% of the time.
-		suffix = g.randSuffix(0.05)
+		suffix = g.randSuffixToWrite(0.05)
 	}
 
 	writerID := g.liveWriters.rand(g.rng)
@@ -936,7 +969,7 @@ func (g *generator) writerRangeKeyUnset() {
 	var suffix []byte
 	if g.rng.Float64() < 0.90 {
 		// Increase the max suffix 5% of the time.
-		suffix = g.randSuffix(0.05)
+		suffix = g.randSuffixToWrite(0.05)
 	}
 
 	// TODO(jackson): Increase probability of effective unsets? Purely random
diff --git a/internal/metamorphic/ops.go b/internal/metamorphic/ops.go
index 05ea7430..9ccc878b 100644
--- a/internal/metamorphic/ops.go
+++ b/internal/metamorphic/ops.go
@@ -5,6 +5,7 @@
 package metamorphic
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 	"path/filepath"
@@ -12,7 +13,9 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/errorfs"
+	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/private"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
@@ -336,16 +339,27 @@ func (o *ingestOp) run(t *test, h *history) {
 	if t.testOpts.ingestUsingApply && len(o.batchIDs) == 1 {
 		id := o.batchIDs[0]
 		b := t.getBatch(id)
-		c, err := o.collapseBatch(t, b)
-		if err == nil {
-			w := t.getWriter(makeObjID(dbTag, 0))
-			err = w.Apply(c, t.writeOpts)
+		iter, rangeDelIter, rangeKeyIter := private.BatchSort(b)
+		// Ingests currently discard range keys. Using apply as an alternative
+		// to ingestion would create a divergence, since batch applications do
+		// commit range keys. Only allow the ingest to be applied as a batch if
+		// it doesn't contain any range keys.
+		// TODO(jackson): When range keys are properly persisted, allow
+		// tables containing range keys to be applied as batches.
+		if rangeKeyIter != nil {
+			closeIters(iter, rangeDelIter, rangeKeyIter)
+		} else {
+			c, err := o.collapseBatch(t, iter, rangeDelIter, rangeKeyIter)
+			if err == nil {
+				w := t.getWriter(makeObjID(dbTag, 0))
+				err = w.Apply(c, t.writeOpts)
+			}
+			_ = b.Close()
+			_ = c.Close()
+			t.clearObj(id)
+			h.Recordf("%s // %v", o, err)
+			return
 		}
-		_ = b.Close()
-		_ = c.Close()
-		t.clearObj(id)
-		h.Recordf("%s // %v", o, err)
-		return
 	}
 
 	var paths []string
@@ -379,15 +393,8 @@ func (o *ingestOp) build(t *test, h *history, b *pebble.Batch, i int) (string, e
 		return "", err
 	}
 
-	iter, rangeDelIter := private.BatchSort(b)
-	defer func() {
-		if iter != nil {
-			iter.Close()
-		}
-		if rangeDelIter != nil {
-			rangeDelIter.Close()
-		}
-	}()
+	iter, rangeDelIter, rangeKeyIter := private.BatchSort(b)
+	defer closeIters(iter, rangeDelIter, rangeKeyIter)
 
 	equal := t.opts.Comparer.Equal
 	tableFormat := t.db.FormatMajorVersion().MaxTableFormat()
@@ -443,22 +450,31 @@ func (o *ingestOp) build(t *test, h *history, b *pebble.Batch, i int) (string, e
 	return path, nil
 }
 
+func closeIters(
+	pointIter base.InternalIterator,
+	rangeDelIter keyspan.FragmentIterator,
+	rangeKeyIter keyspan.FragmentIterator,
+) {
+	if pointIter != nil {
+		pointIter.Close()
+	}
+	if rangeDelIter != nil {
+		rangeDelIter.Close()
+	}
+	if rangeKeyIter != nil {
+		rangeKeyIter.Close()
+	}
+}
+
 // collapseBatch collapses the mutations in a batch to be equivalent to an
 // sstable ingesting those mutations. Duplicate updates to a key are collapsed
 // so that only the latest update is performed. All range deletions are
 // performed first in the batch to match the semantics of ingestion where a
 // range deletion does not delete a point record contained in the sstable.
-func (o *ingestOp) collapseBatch(t *test, b *pebble.Batch) (*pebble.Batch, error) {
-	iter, rangeDelIter := private.BatchSort(b)
-	defer func() {
-		if iter != nil {
-			iter.Close()
-		}
-		if rangeDelIter != nil {
-			rangeDelIter.Close()
-		}
-	}()
-
+func (o *ingestOp) collapseBatch(
+	t *test, pointIter base.InternalIterator, rangeDelIter, rangeKeyIter keyspan.FragmentIterator,
+) (*pebble.Batch, error) {
+	defer closeIters(pointIter, rangeDelIter, rangeKeyIter)
 	equal := t.opts.Comparer.Equal
 	collapsed := t.db.NewBatch()
 
@@ -486,9 +502,9 @@ func (o *ingestOp) collapseBatch(t *test, b *pebble.Batch) (*pebble.Batch, error
 		rangeDelIter = nil
 	}
 
-	if iter != nil {
+	if pointIter != nil {
 		var lastUserKey []byte
-		for key, value := iter.First(); key != nil; key, value = iter.Next() {
+		for key, value := pointIter.First(); key != nil; key, value = pointIter.Next() {
 			// Ignore duplicate keys.
 			if equal(lastUserKey, key.UserKey) {
 				continue
@@ -516,10 +532,10 @@ func (o *ingestOp) collapseBatch(t *test, b *pebble.Batch) (*pebble.Batch, error
 				return nil, err
 			}
 		}
-		if err := iter.Close(); err != nil {
+		if err := pointIter.Close(); err != nil {
 			return nil, err
 		}
-		iter = nil
+		pointIter = nil
 	}
 
 	return collapsed, nil
@@ -568,6 +584,11 @@ type newIterOp struct {
 	iterID   objID
 	lower    []byte
 	upper    []byte
+	keyTypes uint32 // pebble.IterKeyType
+
+	// rangeKeyMaskSuffix may be set if keyTypes is IterKeyTypePointsAndRanges
+	// to configure IterOptions.RangeKeyMasking.Suffix.
+	rangeKeyMaskSuffix []byte
 }
 
 func (o *newIterOp) run(t *test, h *history) {
@@ -577,6 +598,10 @@ func (o *newIterOp) run(t *test, h *history) {
 		i = r.NewIter(&pebble.IterOptions{
 			LowerBound: o.lower,
 			UpperBound: o.upper,
+			KeyTypes:   pebble.IterKeyType(o.keyTypes),
+			RangeKeyMasking: pebble.RangeKeyMasking{
+				Suffix: o.rangeKeyMaskSuffix,
+			},
 		})
 		if err := i.Error(); !errors.Is(err, errorfs.ErrInjected) {
 			break
@@ -589,8 +614,8 @@ func (o *newIterOp) run(t *test, h *history) {
 }
 
 func (o *newIterOp) String() string {
-	return fmt.Sprintf("%s = %s.NewIter(%q, %q)",
-		o.iterID, o.readerID, o.lower, o.upper)
+	return fmt.Sprintf("%s = %s.NewIter(%q, %q, %d /* key types */, %q /* masking suffix */)",
+		o.iterID, o.readerID, o.lower, o.upper, o.keyTypes, o.rangeKeyMaskSuffix)
 }
 
 // newIterUsingCloneOp models a Iterator.Clone operation.
@@ -637,6 +662,31 @@ type iterSeekGEOp struct {
 	limit  []byte
 }
 
+func iteratorPos(i *retryableIter) string {
+	var buf bytes.Buffer
+	fmt.Fprintf(&buf, "%q", i.Key())
+	hasPoint, hasRange := i.HasPointAndRange()
+	if hasPoint {
+		fmt.Fprintf(&buf, ",%q", i.Value())
+	} else {
+		fmt.Fprint(&buf, ",<no point>")
+	}
+	if hasRange {
+		start, end := i.RangeBounds()
+		fmt.Fprintf(&buf, ",[%q,%q)=>{", start, end)
+		for i, rk := range i.RangeKeys() {
+			if i > 0 {
+				fmt.Fprint(&buf, ",")
+			}
+			fmt.Fprintf(&buf, "%q=%q", rk.Suffix, rk.Value)
+		}
+		fmt.Fprint(&buf, "}")
+	} else {
+		fmt.Fprint(&buf, ",<no range>")
+	}
+	return buf.String()
+}
+
 func validBoolToStr(valid bool) string {
 	return fmt.Sprintf("%t", valid)
 }
@@ -665,7 +715,7 @@ func (o *iterSeekGEOp) run(t *test, h *history) {
 		valid, validStr = validityStateToStr(i.SeekGEWithLimit(o.key, o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%q,%q] %v", o, validStr, i.Key(), i.Value(), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
 	}
@@ -685,7 +735,7 @@ func (o *iterSeekPrefixGEOp) run(t *test, h *history) {
 	i := t.getIter(o.iterID)
 	valid := i.SeekPrefixGE(o.key)
 	if valid {
-		h.Recordf("%s // [%t,%q,%q] %v", o, valid, i.Key(), i.Value(), i.Error())
+		h.Recordf("%s // [%t,%s] %v", o, valid, iteratorPos(i), i.Error())
 	} else {
 		h.Recordf("%s // [%t] %v", o, valid, i.Error())
 	}
@@ -713,7 +763,7 @@ func (o *iterSeekLTOp) run(t *test, h *history) {
 		valid, validStr = validityStateToStr(i.SeekLTWithLimit(o.key, o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%q,%q] %v", o, validStr, i.Key(), i.Value(), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
 	}
@@ -732,7 +782,7 @@ func (o *iterFirstOp) run(t *test, h *history) {
 	i := t.getIter(o.iterID)
 	valid := i.First()
 	if valid {
-		h.Recordf("%s // [%t,%q,%q] %v", o, valid, i.Key(), i.Value(), i.Error())
+		h.Recordf("%s // [%t,%s] %v", o, valid, iteratorPos(i), i.Error())
 	} else {
 		h.Recordf("%s // [%t] %v", o, valid, i.Error())
 	}
@@ -751,7 +801,7 @@ func (o *iterLastOp) run(t *test, h *history) {
 	i := t.getIter(o.iterID)
 	valid := i.Last()
 	if valid {
-		h.Recordf("%s // [%t,%q,%q] %v", o, valid, i.Key(), i.Value(), i.Error())
+		h.Recordf("%s // [%t,%s] %v", o, valid, iteratorPos(i), i.Error())
 	} else {
 		h.Recordf("%s // [%t] %v", o, valid, i.Error())
 	}
@@ -778,7 +828,7 @@ func (o *iterNextOp) run(t *test, h *history) {
 		valid, validStr = validityStateToStr(i.NextWithLimit(o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%q,%q] %v", o, validStr, i.Key(), i.Value(), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
 	}
@@ -805,7 +855,7 @@ func (o *iterPrevOp) run(t *test, h *history) {
 		valid, validStr = validityStateToStr(i.PrevWithLimit(o.limit))
 	}
 	if valid {
-		h.Recordf("%s // [%s,%q,%q] %v", o, validStr, i.Key(), i.Value(), i.Error())
+		h.Recordf("%s // [%s,%s] %v", o, validStr, iteratorPos(i), i.Error())
 	} else {
 		h.Recordf("%s // [%s] %v", o, validStr, i.Error())
 	}
diff --git a/internal/metamorphic/parser.go b/internal/metamorphic/parser.go
index 9813c615..8529ea84 100644
--- a/internal/metamorphic/parser.go
+++ b/internal/metamorphic/parser.go
@@ -77,7 +77,7 @@ func opArgs(op op) (receiverID *objID, targetID *objID, args []interface{}) {
 	case *newIndexedBatchOp:
 		return nil, &t.batchID, nil
 	case *newIterOp:
-		return &t.readerID, &t.iterID, []interface{}{&t.lower, &t.upper}
+		return &t.readerID, &t.iterID, []interface{}{&t.lower, &t.upper, &t.keyTypes, &t.rangeKeyMaskSuffix}
 	case *newIterUsingCloneOp:
 		return &t.existingIterID, &t.iterID, nil
 	case *newSnapshotOp:
diff --git a/internal/metamorphic/retryable.go b/internal/metamorphic/retryable.go
index 791fdcd1..ced2f5fc 100644
--- a/internal/metamorphic/retryable.go
+++ b/internal/metamorphic/retryable.go
@@ -69,6 +69,18 @@ func (i *retryableIter) Key() []byte {
 	return i.iter.Key()
 }
 
+func (i *retryableIter) HasPointAndRange() (bool, bool) {
+	return i.iter.HasPointAndRange()
+}
+
+func (i *retryableIter) RangeBounds() ([]byte, []byte) {
+	return i.iter.RangeBounds()
+}
+
+func (i *retryableIter) RangeKeys() []pebble.RangeKeyData {
+	return i.iter.RangeKeys()
+}
+
 func (i *retryableIter) Last() bool {
 	var valid bool
 	i.withRetry(func() { valid = i.iter.Last() })
diff --git a/internal/private/batch.go b/internal/private/batch.go
index 3a046f65..dcdd1f1a 100644
--- a/internal/private/batch.go
+++ b/internal/private/batch.go
@@ -4,9 +4,16 @@
 
 package private
 
-import "github.com/cockroachdb/pebble/internal/base"
+import (
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/keyspan"
+)
 
 // BatchSort is a hook for constructing iterators over the point and range
 // mutations contained in a batch in sorted order. It is intended for testing
 // use only.
-var BatchSort func(interface{}) (base.InternalIterator, base.InternalIterator)
+var BatchSort func(interface{}) (
+	points base.InternalIterator,
+	rangeDels keyspan.FragmentIterator,
+	rangeKeys keyspan.FragmentIterator,
+)
diff --git a/internal/randvar/randvar.go b/internal/randvar/randvar.go
index e5998978..d7d1d90f 100644
--- a/internal/randvar/randvar.go
+++ b/internal/randvar/randvar.go
@@ -19,4 +19,7 @@ type Dynamic interface {
 
 	// Increment the max value the variable will return.
 	IncMax(delta int)
+
+	// Read the current max value the variable will return.
+	Max() uint64
 }
diff --git a/internal/randvar/uniform.go b/internal/randvar/uniform.go
index 526c17b7..e924fb49 100644
--- a/internal/randvar/uniform.go
+++ b/internal/randvar/uniform.go
@@ -40,6 +40,11 @@ func (g *Uniform) IncMax(delta int) {
 	atomic.AddUint64(&g.max, uint64(delta))
 }
 
+// Max returns the max value of the distribution.
+func (g *Uniform) Max() uint64 {
+	return atomic.LoadUint64(&g.max)
+}
+
 // Uint64 returns a random Uint64 between min and max, drawn from a uniform
 // distribution.
 func (g *Uniform) Uint64(rng *rand.Rand) uint64 {
diff --git a/internal/randvar/zipf.go b/internal/randvar/zipf.go
index 95316979..f9aa4e03 100644
--- a/internal/randvar/zipf.go
+++ b/internal/randvar/zipf.go
@@ -120,6 +120,13 @@ func (z *Zipf) IncMax(delta int) {
 	z.mu.Unlock()
 }
 
+// Max returns the max.
+func (z *Zipf) Max() uint64 {
+	z.mu.Lock()
+	defer z.mu.Unlock()
+	return z.mu.max
+}
+
 // Uint64 draws a new value between min and max, with probabilities according
 // to the Zipf distribution.
 func (z *Zipf) Uint64(rng *rand.Rand) uint64 {
diff --git a/internal/rangekey/interleaving_iter.go b/internal/rangekey/interleaving_iter.go
index 9c6d169b..d0b9842c 100644
--- a/internal/rangekey/interleaving_iter.go
+++ b/internal/rangekey/interleaving_iter.go
@@ -256,7 +256,14 @@ func (i *InterleavingIter) Next() (*base.InternalKey, []byte) {
 		//   points:    x*            z
 		//   ranges:        ([y-?))
 		// direction. Either way, we must move to the next point key.
-		i.pointKey, i.pointVal = i.pointIter.Next()
+		switch {
+		case i.pointKey == nil && i.lower == nil:
+			i.pointKey, i.pointVal = i.pointIter.First()
+		case i.pointKey == nil && i.lower != nil:
+			i.pointKey, i.pointVal = i.pointIter.SeekGE(i.lower, false)
+		default:
+			i.pointKey, i.pointVal = i.pointIter.Next()
+		}
 		i.pointKeyInterleaved = false
 
 		if i.rangeKey == nil {
@@ -363,7 +370,14 @@ func (i *InterleavingIter) Prev() (*base.InternalKey, []byte) {
 		//   points:    x             z*
 		//   ranges:        ([y-?))
 		// direction. Either way, we must move the point iterator backwards.
-		i.pointKey, i.pointVal = i.pointIter.Prev()
+		switch {
+		case i.pointKey == nil && i.upper == nil:
+			i.pointKey, i.pointVal = i.pointIter.Last()
+		case i.pointKey == nil && i.upper != nil:
+			i.pointKey, i.pointVal = i.pointIter.SeekLT(i.upper)
+		default:
+			i.pointKey, i.pointVal = i.pointIter.Prev()
+		}
 		i.pointKeyInterleaved = false
 	}
 
diff --git a/iterator.go b/iterator.go
index 4a6ffe7d..9316b136 100644
--- a/iterator.go
+++ b/iterator.go
@@ -624,7 +624,7 @@ func (i *Iterator) findPrevEntry(limit []byte) {
 		// the key space being iterated over in which there are no point keys.
 		// Since limits are best effort, ignoring the limit in this case is
 		// allowed by the contract of limit.
-		if firstLoopIter && limit != nil && i.cmp(limit, i.iterKey.UserKey) > 0 && !i.iterHasRangeKey() {
+		if firstLoopIter && limit != nil && i.cmp(limit, i.iterKey.UserKey) > 0 && !i.rangeKeyWithinLimit(limit) {
 			i.iterValidityState = IterAtLimit
 			i.pos = iterPosCurReversePaused
 			return
@@ -704,7 +704,7 @@ func (i *Iterator) findPrevEntry(limit []byte) {
 			// other than the firstLoopIter case above, where we could step
 			// to a different user key and start processing it for returning
 			// to the caller.
-			if limit != nil && i.iterKey != nil && i.cmp(limit, i.iterKey.UserKey) > 0 && !i.iterHasRangeKey() {
+			if limit != nil && i.iterKey != nil && i.cmp(limit, i.iterKey.UserKey) > 0 && !i.rangeKeyWithinLimit(limit) {
 				i.iterValidityState = IterAtLimit
 				i.pos = iterPosCurReversePaused
 				return
@@ -865,7 +865,17 @@ func (i *Iterator) SeekGE(key []byte) bool {
 	return i.SeekGEWithLimit(key, nil) == IterValid
 }
 
-// SeekGEWithLimit ...
+// SeekGEWithLimit moves the iterator to the first key/value pair whose key is
+// greater than or equal to the given key.
+//
+// If limit is provided, it serves as a best-effort exclusive limit. If the
+// first key greater than or equal to the given search key is also greater than
+// or equal to limit, the Iterator may pause and return IterAtLimit. Because
+// limits are best-effort, SeekGEWithLimit may return a key beyond limit.
+//
+// If the Iterator is configured to iterate over range keys, SeekGEWithLimit
+// guarantees it will surface any range keys with bounds overlapping the
+// keyspace [key, limit).
 func (i *Iterator) SeekGEWithLimit(key []byte, limit []byte) IterValidityState {
 	lastPositioningOp := i.lastPositioningOp
 	// Set it to unknown, since this operation may not succeed, in which case
@@ -1039,12 +1049,14 @@ func (i *Iterator) SeekPrefixGE(key []byte) bool {
 	if lowerBound := i.opts.GetLowerBound(); lowerBound != nil && i.cmp(key, lowerBound) < 0 {
 		if n := i.split(lowerBound); !bytes.Equal(i.prefixOrFullSeekKey, lowerBound[:n]) {
 			i.err = errors.New("pebble: SeekPrefixGE supplied with key outside of lower bound")
+			i.iterValidityState = IterExhausted
 			return false
 		}
 		key = lowerBound
 	} else if upperBound := i.opts.GetUpperBound(); upperBound != nil && i.cmp(key, upperBound) > 0 {
 		if n := i.split(upperBound); !bytes.Equal(i.prefixOrFullSeekKey, upperBound[:n]) {
 			i.err = errors.New("pebble: SeekPrefixGE supplied with key outside of upper bound")
+			i.iterValidityState = IterExhausted
 			return false
 		}
 		key = upperBound
@@ -1076,7 +1088,17 @@ func (i *Iterator) SeekLT(key []byte) bool {
 	return i.SeekLTWithLimit(key, nil) == IterValid
 }
 
-// SeekLTWithLimit ...
+// SeekLTWithLimit moves the iterator to the last key/value pair whose key is
+// less than the given key.
+//
+// If limit is provided, it serves as a best-effort inclusive limit. If the last
+// key less than the given search key is also less than limit, the Iterator may
+// pause and return IterAtLimit. Because limits are best-effort, SeekLTWithLimit
+// may return a key beyond limit.
+//
+// If the Iterator is configured to iterate over range keys, SeekLTWithLimit
+// guarantees it will surface any range keys with bounds overlapping the
+// keyspace up to limit.
 func (i *Iterator) SeekLTWithLimit(key []byte, limit []byte) IterValidityState {
 	lastPositioningOp := i.lastPositioningOp
 	// Set it to unknown, since this operation may not succeed, in which case
@@ -1178,7 +1200,16 @@ func (i *Iterator) Next() bool {
 	return i.NextWithLimit(nil) == IterValid
 }
 
-// NextWithLimit ...
+// NextWithLimit moves the iterator to the next key/value pair.
+//
+// If limit is provided, it serves as a best-effort exclusive limit. If the next
+// key  is greater than or equal to limit, the Iterator may pause and return
+// IterAtLimit. Because limits are best-effort, NextWithLimit may return a key
+// beyond limit.
+//
+// If the Iterator is configured to iterate over range keys, NextWithLimit
+// guarantees it will surface any range keys with bounds overlapping the
+// keyspace up to limit.
 func (i *Iterator) NextWithLimit(limit []byte) IterValidityState {
 	i.stats.ForwardStepCount[InterfaceCall]++
 	if limit != nil && i.hasPrefix {
@@ -1257,7 +1288,16 @@ func (i *Iterator) Prev() bool {
 	return i.PrevWithLimit(nil) == IterValid
 }
 
-// PrevWithLimit ...
+// PrevWithLimit moves the iterator to the previous key/value pair.
+//
+// If limit is provided, it serves as a best-effort inclusive limit. If the
+// previous key is less than limit, the Iterator may pause and return
+// IterAtLimit. Because limits are best-effort, PrevWithLimit may return a key
+// beyond limit.
+//
+// If the Iterator is configured to iterate over range keys, PrevWithLimit
+// guarantees it will surface any range keys with bounds overlapping the
+// keyspace up to limit.
 func (i *Iterator) PrevWithLimit(limit []byte) IterValidityState {
 	i.stats.ReverseStepCount[InterfaceCall]++
 	if i.err != nil {
@@ -1336,10 +1376,30 @@ type RangeKeyData struct {
 	Value  []byte
 }
 
-// iterHasRangeKey returns whether or not the internalIterator has a range key
-// covering its current position.
-func (i *Iterator) iterHasRangeKey() bool {
-	return i.rangeKey != nil && i.rangeKey.iter.HasRangeKey()
+// rangeKeyWithinLimit is called during limited reverse iteration when
+// positioned over a key beyond the limit. If there exists a range key that lies
+// within the limit, the iterator must not pause in order to ensure the user has
+// an opportunity to observe the range key within limit.
+//
+// It would be valid to ignore the limit whenever there's a range key covering
+// the key, but that would introduce nondeterminism. To preserve determinism for
+// testing, the iterator ignores the limit only if the covering range key does
+// cover the keyspace within the limit.
+//
+// This awkwardness exists because range keys are interleaved at their inclusive
+// start positions. Note that limit is inclusive.
+func (i *Iterator) rangeKeyWithinLimit(limit []byte) bool {
+	if i.rangeKey == nil || !i.rangeKey.iter.HasRangeKey() {
+		// If there are no covering range keys, it is safe to to pause
+		// immediately.
+		return false
+	}
+	// If the range key ends beyond the limit, then the range key does not cover
+	// any portion of the keyspace within the limit and it is safe to pause.
+	if _, end := i.rangeKey.iter.RangeKeyBounds(); i.cmp(end, limit) <= 0 {
+		return false
+	}
+	return true
 }
 
 // setRangeKey sets the current range key to the underlying iterator's current
@@ -1450,6 +1510,8 @@ func (i *Iterator) Key() []byte {
 // Value returns the value of the current key/value pair, or nil if done. The
 // caller should not modify the contents of the returned slice, and its
 // contents may change on the next call to Next.
+//
+// Only valid if HasPointAndRange() returns true for hasPoint.
 func (i *Iterator) Value() []byte {
 	return i.value
 }
diff --git a/level_iter.go b/level_iter.go
index 263c93b2..e165c76b 100644
--- a/level_iter.go
+++ b/level_iter.go
@@ -196,7 +196,7 @@ func (l *levelIter) init(
 	l.split = split
 	l.iterFile = nil
 	l.newIters = newIters
-	l.files = files
+	l.files = files.Filter(manifest.KeyTypePoint)
 	l.bytesIterated = bytesIterated
 }
 
@@ -231,8 +231,8 @@ func (l *levelIter) findFileGE(key []byte) *fileMetadata {
 	// (see the comment in that function).
 
 	m := l.files.SeekGE(l.cmp, key)
-	for m != nil && m.Largest.IsExclusiveSentinel() &&
-		l.cmp(m.Largest.UserKey, key) == 0 {
+	for m != nil && m.LargestPointKey.IsExclusiveSentinel() &&
+		l.cmp(m.LargestPointKey.UserKey, key) == 0 {
 		m = l.files.Next()
 	}
 	return m
diff --git a/merging_iter.go b/merging_iter.go
index 9034ed45..556e6c7a 100644
--- a/merging_iter.go
+++ b/merging_iter.go
@@ -419,8 +419,16 @@ func (m *mergingIter) switchToMinHeap() {
 	}
 
 	// Special handling for the current iterator because we were using its key
-	// above.
-	cur.iterKey, cur.iterValue = cur.iter.Next()
+	// above. The iterator cur.iter may still be exhausted at a sstable boundary
+	// sentinel. Similar to the logic applied to the other levels, in these
+	// cases we seek the iterator to the first key in order to avoid violating
+	// levelIter's invariants. See the example in the for loop above.
+	if m.lower != nil && cur.isSyntheticIterBoundsKey && cur.iterKey.IsExclusiveSentinel() &&
+		m.heap.cmp(cur.iterKey.UserKey, m.lower) <= 0 {
+		cur.iterKey, cur.iterValue = cur.iter.SeekGE(m.lower, false /* trySeekUsingNext */)
+	} else {
+		cur.iterKey, cur.iterValue = cur.iter.Next()
+	}
 	m.initMinHeap()
 }
 
@@ -493,8 +501,17 @@ func (m *mergingIter) switchToMaxHeap() {
 	}
 
 	// Special handling for the current iterator because we were using its key
+	// above. The iterator cur.iter may still be exhausted at a sstable boundary
+	// sentinel. Similar to the logic applied to the other levels, in these
+	// cases we seek the iterator to  in order to avoid violating levelIter's
+	// invariants by Prev-ing through files.  See the example in the for loop
 	// above.
-	cur.iterKey, cur.iterValue = cur.iter.Prev()
+	if m.upper != nil && cur.isSyntheticIterBoundsKey && cur.iterKey.IsExclusiveSentinel() &&
+		m.heap.cmp(cur.iterKey.UserKey, m.upper) >= 0 {
+		cur.iterKey, cur.iterValue = cur.iter.SeekLT(m.upper)
+	} else {
+		cur.iterKey, cur.iterValue = cur.iter.Prev()
+	}
 	m.initMaxHeap()
 }
 
diff --git a/open.go b/open.go
index 036df5ba..d1f0280d 100644
--- a/open.go
+++ b/open.go
@@ -401,7 +401,7 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 		d.mu.log.LogWriter.SetMinSyncInterval(d.opts.WALMinSyncInterval)
 		d.mu.versions.metrics.WAL.Files++
 	}
-	d.updateReadStateLocked(d.opts.DebugCheck)
+	d.updateReadStateLocked(d.opts.DebugCheck, nil)
 
 	if !d.opts.ReadOnly {
 		// Write the current options to disk.
@@ -683,6 +683,12 @@ func (d *DB) replayWAL(
 		if err != nil {
 			return 0, err
 		}
+		// TODO(jackson): Remove the below call to applyFlushedRangeKeys once
+		// flushes actually persist range keys to sstables.
+		err = d.applyFlushedRangeKeys(toFlush)
+		if err != nil {
+			return 0, err
+		}
 		ve.NewFiles = append(ve.NewFiles, newVE.NewFiles...)
 		for i := range toFlush {
 			toFlush[i].readerUnref()
diff --git a/range_keys.go b/range_keys.go
index a482ced7..9d5f171c 100644
--- a/range_keys.go
+++ b/range_keys.go
@@ -7,6 +7,7 @@ package pebble
 import (
 	"sync"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/arenaskl"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
@@ -31,6 +32,8 @@ type RangeKeysArena struct {
 // use a separate arena that exists beyond the lifetime of any individual
 // memtable.  For as long as we're relying on this hack, a single *pebble.DB may
 // only store as many range keys as fit in this arena.
+//
+// TODO(jackson): Remove applyFlushedRangeKeys when range keys are persisted.
 func (d *DB) applyFlushedRangeKeys(flushable []*flushableEntry) error {
 	var added uint32
 	for i := 0; i < len(flushable); i++ {
@@ -68,10 +71,19 @@ func (d *DB) applyFlushedRangeKeys(flushable []*flushableEntry) error {
 			}
 			rangekey.RecombineValue(k.Kind(), buf, s.End, s.Value)
 
-			if err := d.rangeKeys.skl.Add(*k, buf[:n]); err != nil {
+			err := d.rangeKeys.skl.Add(*k, buf[:n])
+			switch {
+			case err == nil:
+				added++
+			case errors.Is(err, arenaskl.ErrRecordExists):
+				// It's possible that we'll try to add a key to the arena twice
+				// during metamorphic tests that reset the synced state. Ignore.
+				// When range keys are actually flushed to stable storage, this
+				// will go away.
+			default:
+				// err != nil
 				return err
 			}
-			added++
 		}
 	}
 	if added > 0 {
diff --git a/read_state.go b/read_state.go
index 6735643d..6364bc6b 100644
--- a/read_state.go
+++ b/read_state.go
@@ -77,9 +77,10 @@ func (d *DB) loadReadState() *readState {
 }
 
 // updateReadStateLocked creates a new readState from the current version and
-// list of memtables. Requires DB.mu is held. If checker is not nil, it is called after installing
-// the new readState
-func (d *DB) updateReadStateLocked(checker func(*DB) error) {
+// list of memtables. Requires DB.mu is held. If checker is not nil, it is
+// called after installing the new readState. If atomicFunc is not nil, it is
+// executed atomically with the transition to the new read state.
+func (d *DB) updateReadStateLocked(checker func(*DB) error, atomicFunc func()) {
 	s := &readState{
 		db:        d,
 		refcnt:    1,
@@ -94,6 +95,11 @@ func (d *DB) updateReadStateLocked(checker func(*DB) error) {
 	d.readState.Lock()
 	old := d.readState.val
 	d.readState.val = s
+	// TODO(jackson): Remove atomicFunc. It's temporary to allow us to copy
+	// range keys from flushed memtables into the global arena atomically.
+	if atomicFunc != nil {
+		atomicFunc()
+	}
 	d.readState.Unlock()
 	if checker != nil {
 		if err := checker(d); err != nil {
diff --git a/sstable/block.go b/sstable/block.go
index d1cd634b..9048ad10 100644
--- a/sstable/block.go
+++ b/sstable/block.go
@@ -35,6 +35,16 @@ type blockWriter struct {
 	tmp             [4]byte
 }
 
+func (w *blockWriter) clear() {
+	*w = blockWriter{
+		buf:      w.buf[:0],
+		restarts: w.restarts[:0],
+		curKey:   w.curKey[:0],
+		curValue: w.curValue[:0],
+		prevKey:  w.prevKey[:0],
+	}
+}
+
 func (w *blockWriter) store(keySize int, value []byte) {
 	shared := 0
 	if w.nEntries == w.nextRestart {
diff --git a/sstable/write_queue.go b/sstable/write_queue.go
index 2940ef87..fc6b9dd1 100644
--- a/sstable/write_queue.go
+++ b/sstable/write_queue.go
@@ -25,6 +25,15 @@ type writeTask struct {
 	finishedIndexProps []byte
 }
 
+// It is not the responsibility of the writeTask to clear the
+// task.flushableIndexBlock, and task.buf.
+func (task *writeTask) clear() {
+	*task = writeTask{
+		indexEntrySep:   base.InvalidInternalKey,
+		compressionDone: task.compressionDone,
+	}
+}
+
 // Note that only the Writer client goroutine will be adding tasks to the writeQueue.
 // Both the Writer client and the compression goroutines will be able to write to
 // writeTask.compressionDone to indicate that the compression job associated with
@@ -78,16 +87,17 @@ func (w *writeQueue) performWrite(task *writeTask) error {
 // It is necessary to ensure that none of the buffers in the writeTask,
 // dataBlockBuf, indexBlockBuf, are pointed to by another struct.
 func (w *writeQueue) releaseBuffers(task *writeTask) {
+	task.buf.clear()
+	dataBlockBufPool.Put(task.buf)
+
 	// This index block is no longer used by the Writer, so we can add it back
 	// to the pool.
 	if task.flushableIndexBlock != nil {
+		task.flushableIndexBlock.clear()
 		indexBlockBufPool.Put(task.flushableIndexBlock)
 	}
 
-	*task = writeTask{
-		indexEntrySep:   base.InvalidInternalKey,
-		compressionDone: task.compressionDone,
-	}
+	task.clear()
 	writeTaskPool.Put(task)
 }
 
diff --git a/sstable/writer.go b/sstable/writer.go
index 43c2d00c..a63a4350 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -196,7 +196,7 @@ type Writer struct {
 	rkBuf      []byte
 	// dataBlockBuf consists of the state which is currently owned by and used by
 	// the Writer client goroutine. This state can be handed off to other goroutines.
-	dataBlockBuf dataBlockBuf
+	dataBlockBuf *dataBlockBuf
 	// blockBuf consists of the state which is owned by and used by the Writer client
 	// goroutine.
 	blockBuf blockBuf
@@ -273,12 +273,7 @@ func (s *sizeEstimate) written(newTotalSize uint64, inflightSize int, finalEntry
 }
 
 func (s *sizeEstimate) clear() {
-	s.numEntries = 0
-	s.inflightSize = 0
-	s.maxEstimatedSize = 0
-	s.totalSize = 0
-	s.compressedSize = 0
-	s.uncompressedSize = 0
+	*s = sizeEstimate{emptySize: s.emptySize}
 }
 
 type indexBlockBuf struct {
@@ -296,6 +291,16 @@ type indexBlockBuf struct {
 	restartInterval int
 }
 
+func (i *indexBlockBuf) clear() {
+	i.block.clear()
+	if i.size.useMutex {
+		i.size.mu.Lock()
+		defer i.size.mu.Unlock()
+	}
+	i.size.estimate.clear()
+	i.restartInterval = 0
+}
+
 var indexBlockBufPool = sync.Pool{
 	New: func() interface{} {
 		return &indexBlockBuf{}
@@ -340,11 +345,6 @@ func (i *indexBlockBuf) add(key InternalKey, value []byte, inflightSize int) {
 
 func (i *indexBlockBuf) finish() []byte {
 	b := i.block.finish()
-	if i.size.useMutex {
-		i.size.mu.Lock()
-		defer i.size.mu.Unlock()
-	}
-	i.size.estimate.clear()
 	return b
 }
 
@@ -452,6 +452,15 @@ type blockBuf struct {
 	checksummer   checksummer
 }
 
+func (b *blockBuf) clear() {
+	// We can't assign b.compressedBuf[:0] to compressedBuf because snappy relies
+	// on the length of the buffer, and not the capacity to determine if it needs
+	// to make an allocation.
+	*b = blockBuf{
+		compressedBuf: b.compressedBuf, checksummer: b.checksummer,
+	}
+}
+
 // A dataBlockBuf holds all the state required to compress and write a data block to disk.
 // A dataBlockBuf begins its lifecycle owned by the Writer client goroutine. The Writer
 // client goroutine adds keys to the sstable, writing directly into a dataBlockBuf's blockWriter
@@ -483,9 +492,27 @@ type dataBlockBuf struct {
 	sepScratch []byte
 }
 
-func (d *dataBlockBuf) init(restartInterval int, checksumType ChecksumType) {
+func (d *dataBlockBuf) clear() {
+	d.blockBuf.clear()
+	d.dataBlock.clear()
+
+	d.uncompressed = nil
+	d.compressed = nil
+	d.dataBlockProps = nil
+	d.sepScratch = d.sepScratch[:0]
+}
+
+var dataBlockBufPool = sync.Pool{
+	New: func() interface{} {
+		return &dataBlockBuf{}
+	},
+}
+
+func newDataBlockBuf(restartInterval int, checksumType ChecksumType) *dataBlockBuf {
+	d := dataBlockBufPool.Get().(*dataBlockBuf)
 	d.dataBlock.restartInterval = restartInterval
 	d.checksummer.checksumType = checksumType
+	return d
 }
 
 func (d *dataBlockBuf) finish() {
@@ -1029,7 +1056,7 @@ func (w *Writer) flush(key InternalKey) error {
 	var err error
 
 	// We're finishing a data block.
-	err = w.finishDataBlockProps(&w.dataBlockBuf)
+	err = w.finishDataBlockProps(w.dataBlockBuf)
 	if err != nil {
 		return err
 	}
@@ -1044,12 +1071,14 @@ func (w *Writer) flush(key InternalKey) error {
 	// byte slice which supports "sep" will eventually be copied when "sep" is
 	// added to the index block.
 	prevKey := base.DecodeInternalKey(w.dataBlockBuf.dataBlock.curKey)
-	sep := w.indexEntrySep(prevKey, key, &w.dataBlockBuf)
-	// We determine that we should flush an index block from the Writer client goroutine, but
-	// we actually finish the index block from the writeQueue. When we determine that an index
-	// block should be flushed, we need to call BlockPropertyCollector.FinishIndexBlock. But
-	// block property collector calls must happen sequentially from the Writer client. Therefore,
-	// we need to determine that we are going to flush the index block from the Writer client.
+	sep := w.indexEntrySep(prevKey, key, w.dataBlockBuf)
+	// We determine that we should flush an index block from the Writer client
+	// goroutine, but we actually finish the index block from the writeQueue.
+	// When we determine that an index block should be flushed, we need to call
+	// BlockPropertyCollector.FinishIndexBlock. But block property collector
+	// calls must happen sequentially from the Writer client. Therefore, we need
+	// to determine that we are going to flush the index block from the Writer
+	// client.
 	shouldFlushIndexBlock := supportsTwoLevelIndex(w.tableFormat) && w.indexBlock.shouldFlush(
 		sep, encodedBHPEstimatedSize, w.indexBlockSize, w.indexBlockSizeThreshold,
 	)
@@ -1059,15 +1088,18 @@ func (w *Writer) flush(key InternalKey) error {
 	if shouldFlushIndexBlock {
 		flushableIndexBlock = w.indexBlock
 		w.indexBlock = newIndexBlockBuf()
-		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to flush the index block.
+		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to
+		// flush the index block.
 		indexProps, err = w.finishIndexBlockProps()
 		if err != nil {
 			return err
 		}
 	}
 
-	// We've called BlockPropertyCollector.FinishDataBlock, and, if necessary, BlockPropertyCollector.FinishIndexBlock.
-	// Since we've decided to finish the data block, we can call BlockPropertyCollector.AddPrevDataBlockToIndexBlock.
+	// We've called BlockPropertyCollector.FinishDataBlock, and, if necessary,
+	// BlockPropertyCollector.FinishIndexBlock. Since we've decided to finish
+	// the data block, we can call
+	// BlockPropertyCollector.AddPrevDataBlockToIndexBlock.
 	w.addPrevDataBlockToIndexBlockProps()
 
 	// Schedule a write.
@@ -1075,7 +1107,7 @@ func (w *Writer) flush(key InternalKey) error {
 	// We're setting compressionDone to indicate that compression of this block
 	// has already been completed.
 	writeTask.compressionDone <- true
-	writeTask.buf = &w.dataBlockBuf
+	writeTask.buf = w.dataBlockBuf
 	writeTask.indexEntrySep = sep
 	writeTask.inflightSize = estimatedUncompressedSize
 	writeTask.currIndexBlock = w.indexBlock
@@ -1086,7 +1118,9 @@ func (w *Writer) flush(key InternalKey) error {
 	// The writeTask corresponds to an unwritten index entry.
 	w.indexBlock.addInflight(writeTask.indexInflightSize)
 
+	w.dataBlockBuf = nil
 	err = w.coordination.writeQueue.addSync(writeTask)
+	w.dataBlockBuf = newDataBlockBuf(w.restartInterval, w.checksumType)
 
 	return err
 }
@@ -1136,7 +1170,7 @@ func (w *Writer) finishDataBlockProps(buf *dataBlockBuf) error {
 func (w *Writer) maybeAddBlockPropertiesToBlockHandle(
 	bh BlockHandle,
 ) (BlockHandleWithProperties, error) {
-	err := w.finishDataBlockProps(&w.dataBlockBuf)
+	err := w.finishDataBlockProps(w.dataBlockBuf)
 	if err != nil {
 		return BlockHandleWithProperties{}, err
 	}
@@ -1167,6 +1201,8 @@ func (w *Writer) indexEntrySep(prevKey, key InternalKey, dataBlockBuf *dataBlock
 // 1. addIndexEntry must not store references to the sep InternalKey, the tmp
 //    byte slice, bhp.Props. That is, these must be either deep copied or
 //    encoded.
+// 2. addIndexEntry must not hold references to the flushIndexBuf, and the writeTo
+//    indexBlockBufs.
 func (w *Writer) addIndexEntry(
 	sep InternalKey,
 	bhp BlockHandleWithProperties,
@@ -1211,12 +1247,12 @@ func (w *Writer) addPrevDataBlockToIndexBlockProps() {
 // aren't being written asynchronously.
 //
 // Invariant:
-// 1. addIndexEntry must not store references to the prevKey, key InternalKey's,
+// 1. addIndexEntrySync must not store references to the prevKey, key InternalKey's,
 //    the tmp byte slice. That is, these must be either deep copied or encoded.
 func (w *Writer) addIndexEntrySync(
 	prevKey, key InternalKey, bhp BlockHandleWithProperties, tmp []byte,
 ) error {
-	sep := w.indexEntrySep(prevKey, key, &w.dataBlockBuf)
+	sep := w.indexEntrySep(prevKey, key, w.dataBlockBuf)
 	shouldFlush := supportsTwoLevelIndex(
 		w.tableFormat) && w.indexBlock.shouldFlush(
 		sep, encodedBHPEstimatedSize, w.indexBlockSize, w.indexBlockSizeThreshold,
@@ -1228,7 +1264,8 @@ func (w *Writer) addIndexEntrySync(
 		flushableIndexBlock = w.indexBlock
 		w.indexBlock = newIndexBlockBuf()
 
-		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to flush the index block.
+		// Call BlockPropertyCollector.FinishIndexBlock, since we've decided to
+		// flush the index block.
 		props, err = w.finishIndexBlockProps()
 		if err != nil {
 			return err
@@ -1236,6 +1273,10 @@ func (w *Writer) addIndexEntrySync(
 	}
 
 	err = w.addIndexEntry(sep, bhp, tmp, flushableIndexBlock, w.indexBlock, 0, props)
+	if flushableIndexBlock != nil {
+		flushableIndexBlock.clear()
+		indexBlockBufPool.Put(flushableIndexBlock)
+	}
 	w.addPrevDataBlockToIndexBlockProps()
 	return err
 }
@@ -1484,7 +1525,7 @@ func (w *Writer) Close() (err error) {
 	//    however, if a dataBlock is flushed, then we add a key to the new w.dataBlockBuf in the
 	//    addPoint function after the flush occurs.
 	if w.dataBlockBuf.dataBlock.nEntries >= 1 {
-		w.meta.SetLargestPointKey(base.DecodeInternalKey(w.dataBlockBuf.dataBlock.curKey))
+		w.meta.SetLargestPointKey(base.DecodeInternalKey(w.dataBlockBuf.dataBlock.curKey).Clone())
 	}
 
 	// Finish the last data block, or force an empty data block if there
@@ -1720,6 +1761,13 @@ func (w *Writer) Close() (err error) {
 		return err
 	}
 
+	w.dataBlockBuf.clear()
+	dataBlockBufPool.Put(w.dataBlockBuf)
+	w.dataBlockBuf = nil
+	w.indexBlock.clear()
+	indexBlockBufPool.Put(w.indexBlock)
+	w.indexBlock = nil
+
 	// Make any future calls to Set or Close return an error.
 	w.err = errWriterClosed
 	return nil
@@ -1759,6 +1807,7 @@ type PreviousPointKeyOpt struct {
 // option was passed during creation. The returned key points directly into
 // a buffer belonging to the Writer. The value's lifetime ends the next time a
 // point key is added to the Writer.
+// Invariant: UnsafeKey isn't and shouldn't be called after the Writer is closed.
 func (o PreviousPointKeyOpt) UnsafeKey() base.InternalKey {
 	if o.w == nil {
 		return base.InvalidInternalKey
@@ -1826,7 +1875,7 @@ func NewWriter(f writeCloseSyncer, o WriterOptions, extraOpts ...WriterOption) *
 		},
 	}
 
-	w.dataBlockBuf.init(w.restartInterval, w.checksumType)
+	w.dataBlockBuf = newDataBlockBuf(w.restartInterval, w.checksumType)
 
 	w.blockBuf = blockBuf{
 		checksummer: checksummer{checksumType: o.Checksum},


Test patch:
diff --git a/data_test.go b/data_test.go
index 4470ab1f..3528a8ce 100644
--- a/data_test.go
+++ b/data_test.go
@@ -677,7 +677,7 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 					entry := d.newFlushableEntry(d.mu.mem.mutable, 0, 0)
 					entry.readerRefs++
 					d.mu.mem.queue = append(d.mu.mem.queue, entry)
-					d.updateReadStateLocked(nil)
+					d.updateReadStateLocked(nil, nil)
 				}
 				mem = d.mu.mem.mutable
 				fields = fields[1:]
@@ -734,7 +734,7 @@ func runDBDefineCmd(td *datadriven.TestData, opts *Options) (*DB, error) {
 		}); err != nil {
 			return nil, err
 		}
-		d.updateReadStateLocked(nil)
+		d.updateReadStateLocked(nil, nil)
 		d.updateTableStatsLocked(ve.NewFiles)
 	}
 
diff --git a/internal/randvar/skewed_latest.go b/internal/randvar/skewed_latest.go
index f7f45c48..1a3dbb1b 100644
--- a/internal/randvar/skewed_latest.go
+++ b/internal/randvar/skewed_latest.go
@@ -60,6 +60,13 @@ func (z *SkewedLatest) IncMax(delta int) {
 	z.mu.Unlock()
 }
 
+// Max returns max.
+func (z *SkewedLatest) Max() uint64 {
+	z.mu.Lock()
+	defer z.mu.Unlock()
+	return z.mu.zipf.Max()
+}
+
 // Uint64 returns a random Uint64 between min and max, where keys near max are
 // most likely to be drawn.
 func (z *SkewedLatest) Uint64(rng *rand.Rand) uint64 {
diff --git a/level_checker_test.go b/level_checker_test.go
index ba168398..81276ff9 100644
--- a/level_checker_test.go
+++ b/level_checker_test.go
@@ -137,7 +137,7 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 				largestKey := base.ParseInternalKey(keys[1])
 				m := (&fileMetadata{
 					FileNum: fileNum,
-				}).ExtendRangeKeyBounds(cmp, smallestKey, largestKey)
+				}).ExtendPointKeyBounds(cmp, smallestKey, largestKey)
 				*li = append(*li, m)
 
 				i++
diff --git a/level_iter_test.go b/level_iter_test.go
index 06245284..120a5345 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -55,7 +55,7 @@ func TestLevelIter(t *testing.T) {
 
 				meta := (&fileMetadata{
 					FileNum: FileNum(len(metas)),
-				}).ExtendRangeKeyBounds(
+				}).ExtendPointKeyBounds(
 					DefaultComparer.Compare,
 					f.keys[0],
 					f.keys[len(f.keys)-1],
@@ -138,10 +138,11 @@ func TestLevelIter(t *testing.T) {
 }
 
 type levelIterTest struct {
-	cmp     base.Comparer
-	mem     vfs.FS
-	readers []*sstable.Reader
-	metas   []*fileMetadata
+	cmp          base.Comparer
+	mem          vfs.FS
+	readers      []*sstable.Reader
+	metas        []*fileMetadata
+	itersCreated int
 }
 
 func newLevelIterTest() *levelIterTest {
@@ -156,6 +157,7 @@ func newLevelIterTest() *levelIterTest {
 func (lt *levelIterTest) newIters(
 	file *manifest.FileMetadata, opts *IterOptions, _ *uint64,
 ) (internalIterator, keyspan.FragmentIterator, error) {
+	lt.itersCreated++
 	iter, err := lt.readers[file.FileNum].NewIter(opts.LowerBound, opts.UpperBound)
 	if err != nil {
 		return nil, nil, err
@@ -174,6 +176,7 @@ func (lt *levelIterTest) runClear(d *datadriven.TestData) string {
 	}
 	lt.readers = nil
 	lt.metas = nil
+	lt.itersCreated = 0
 	return ""
 }
 
@@ -185,10 +188,22 @@ func (lt *levelIterTest) runBuild(d *datadriven.TestData) string {
 		return err.Error()
 	}
 
+	tableFormat := sstable.TableFormatRocksDBv2
+	for _, arg := range d.CmdArgs {
+		if arg.Key == "format" {
+			switch arg.Vals[0] {
+			case "rocksdbv2":
+				tableFormat = sstable.TableFormatRocksDBv2
+			case "pebblev2":
+				tableFormat = sstable.TableFormatPebblev2
+			}
+		}
+	}
 	fp := bloom.FilterPolicy(10)
 	w := sstable.NewWriter(f0, sstable.WriterOptions{
 		Comparer:     &lt.cmp,
 		FilterPolicy: fp,
+		TableFormat:  tableFormat,
 	})
 	var tombstones []keyspan.Span
 	f := keyspan.Fragmenter{
@@ -205,6 +220,8 @@ func (lt *levelIterTest) runBuild(d *datadriven.TestData) string {
 		switch ikey.Kind() {
 		case InternalKeyKindRangeDelete:
 			f.Add(keyspan.Span{Start: ikey, End: value})
+		case InternalKeyKindRangeKeySet:
+			w.RangeKeySet(ikey.UserKey, value, nil, nil)
 		default:
 			if err := w.Add(ikey, value); err != nil {
 				return err.Error()
@@ -244,7 +261,9 @@ func (lt *levelIterTest) runBuild(d *datadriven.TestData) string {
 	}
 	if meta.HasRangeDelKeys {
 		m.ExtendPointKeyBounds(lt.cmp.Compare, meta.SmallestRangeDel, meta.LargestRangeDel)
-
+	}
+	if meta.HasRangeKeys {
+		m.ExtendRangeKeyBounds(lt.cmp.Compare, meta.SmallestRangeKey, meta.LargestRangeKey)
 	}
 	lt.metas = append(lt.metas, m)
 
@@ -394,6 +413,8 @@ func TestLevelIterSeek(t *testing.T) {
 			iter.initRangeDel(&iter.rangeDelIter)
 			return runInternalIterCmd(d, iter, iterCmdVerboseKey)
 
+		case "iters-created":
+			return fmt.Sprintf("%d", lt.itersCreated)
 		default:
 			return fmt.Sprintf("unknown command: %s", d.Cmd)
 		}
diff --git a/read_state_test.go b/read_state_test.go
index 4f9d5f57..899677de 100644
--- a/read_state_test.go
+++ b/read_state_test.go
@@ -29,7 +29,7 @@ func BenchmarkReadState(b *testing.B) {
 				for pb.Next() {
 					if rng.Float32() < updateFrac {
 						d.mu.Lock()
-						d.updateReadStateLocked(nil)
+						d.updateReadStateLocked(nil, nil)
 						d.mu.Unlock()
 					} else {
 						s := d.loadReadState()
diff --git a/sstable/block_test.go b/sstable/block_test.go
index aa5333bb..73486840 100644
--- a/sstable/block_test.go
+++ b/sstable/block_test.go
@@ -19,11 +19,11 @@ import (
 	"golang.org/x/exp/rand"
 )
 
-func TestBlockWriter(t *testing.T) {
-	ikey := func(s string) InternalKey {
-		return InternalKey{UserKey: []byte(s)}
-	}
+func ikey(s string) InternalKey {
+	return InternalKey{UserKey: []byte(s)}
+}
 
+func TestBlockWriter(t *testing.T) {
 	w := &rawBlockWriter{
 		blockWriter: blockWriter{restartInterval: 16},
 	}
@@ -42,6 +42,39 @@ func TestBlockWriter(t *testing.T) {
 	}
 }
 
+func testBlockCleared(t *testing.T, w, b *blockWriter) {
+	require.Equal(t, w.restartInterval, b.restartInterval)
+	require.Equal(t, w.nEntries, b.nEntries)
+	require.Equal(t, w.nextRestart, b.nextRestart)
+	require.Equal(t, len(w.buf), len(b.buf))
+	require.Equal(t, len(w.restarts), len(b.restarts))
+	require.Equal(t, len(w.curKey), len(b.curKey))
+	require.Equal(t, len(w.prevKey), len(b.prevKey))
+	require.Equal(t, len(w.curValue), len(b.curValue))
+	require.Equal(t, w.tmp, b.tmp)
+
+	// Make sure that we didn't lose the allocated byte slices.
+	require.True(t, cap(w.buf) > 0 && cap(b.buf) == 0)
+	require.True(t, cap(w.restarts) > 0 && cap(b.restarts) == 0)
+	require.True(t, cap(w.curKey) > 0 && cap(b.curKey) == 0)
+	require.True(t, cap(w.prevKey) > 0 && cap(b.prevKey) == 0)
+	require.True(t, cap(w.curValue) > 0 && cap(b.curValue) == 0)
+}
+
+func TestBlockClear(t *testing.T) {
+	w := blockWriter{restartInterval: 16}
+	w.add(ikey("apple"), nil)
+	w.add(ikey("apricot"), nil)
+	w.add(ikey("banana"), nil)
+
+	w.clear()
+
+	// Once a block is cleared, we expect its fields to be cleared, but we expect
+	// it to keep its allocated byte slices.
+	b := blockWriter{}
+	testBlockCleared(t, &w, &b)
+}
+
 func TestInvalidInternalKeyDecoding(t *testing.T) {
 	// Invalid keys since they don't have an 8 byte trailer.
 	testCases := []string{
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 4d5c32e0..8518aaf8 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -8,8 +8,10 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"math/rand"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/cockroachdb/errors"
@@ -18,6 +20,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/internal/humanize"
+	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -178,6 +181,73 @@ func runDataDriven(t *testing.T, file string) {
 	})
 }
 
+func testBlockBufClear(t *testing.T, b1, b2 *blockBuf) {
+	require.Equal(t, b1.tmp, b2.tmp)
+}
+
+func TestBlockBufClear(t *testing.T) {
+	b1 := &blockBuf{}
+	b1.tmp[0] = 1
+	b1.compressedBuf = make([]byte, 1)
+	b1.clear()
+	testBlockBufClear(t, b1, &blockBuf{})
+}
+
+func TestClearDataBlockBuf(t *testing.T) {
+	d := newDataBlockBuf(1, ChecksumTypeCRC32c)
+	d.blockBuf.compressedBuf = make([]byte, 1)
+	d.dataBlock.add(ikey("apple"), nil)
+	d.dataBlock.add(ikey("banana"), nil)
+
+	d.clear()
+	testBlockCleared(t, &d.dataBlock, &blockWriter{})
+	testBlockBufClear(t, &d.blockBuf, &blockBuf{})
+
+	dataBlockBufPool.Put(d)
+}
+
+func TestClearIndexBlockBuf(t *testing.T) {
+	i := newIndexBlockBuf()
+	i.block.add(ikey("apple"), nil)
+	i.block.add(ikey("banana"), nil)
+	i.clear()
+
+	testBlockCleared(t, &i.block, &blockWriter{})
+	require.Equal(
+		t, i.size.estimate, sizeEstimate{emptySize: i.size.estimate.emptySize},
+	)
+	indexBlockBufPool.Put(i)
+}
+
+func TestClearWriteTask(t *testing.T) {
+	w := writeTaskPool.Get().(*writeTask)
+	ch := make(chan bool, 1)
+	w.compressionDone = ch
+	w.buf = &dataBlockBuf{}
+	w.flushableIndexBlock = &indexBlockBuf{}
+	w.currIndexBlock = &indexBlockBuf{}
+	w.indexEntrySep = ikey("apple")
+	w.inflightSize = 1
+	w.indexInflightSize = 1
+	w.finishedIndexProps = []byte{'a', 'v'}
+
+	w.clear()
+
+	var nilDataBlockBuf *dataBlockBuf
+	var nilIndexBlockBuf *indexBlockBuf
+	// Channels should be the same(no new channel should be allocated)
+	require.Equal(t, w.compressionDone, ch)
+	require.Equal(t, w.buf, nilDataBlockBuf)
+	require.Equal(t, w.flushableIndexBlock, nilIndexBlockBuf)
+	require.Equal(t, w.currIndexBlock, nilIndexBlockBuf)
+	require.Equal(t, w.indexEntrySep, base.InvalidInternalKey)
+	require.Equal(t, w.inflightSize, 0)
+	require.Equal(t, w.indexInflightSize, 0)
+	require.Equal(t, w.finishedIndexProps, []byte(nil))
+
+	writeTaskPool.Put(w)
+}
+
 func TestSizeEstimate(t *testing.T) {
 	var sizeEstimate sizeEstimate
 	datadriven.RunTest(t, "testdata/size_estimate",
@@ -505,6 +575,61 @@ func TestWriter_TableFormatCompatibility(t *testing.T) {
 	}
 }
 
+// Tests for races, such as https://github.com/cockroachdb/cockroach/issues/77194,
+// in the Writer.
+func TestWriterRace(t *testing.T) {
+	ks := testkeys.Alpha(5)
+	ks = ks.EveryN(ks.Count() / 1_000)
+	keys := make([][]byte, ks.Count())
+	for ki := 0; ki < len(keys); ki++ {
+		keys[ki] = testkeys.Key(ks, ki)
+	}
+	readerOpts := ReaderOptions{
+		Comparer: testkeys.Comparer,
+		Filters:  map[string]base.FilterPolicy{},
+	}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 16; i++ {
+		wg.Add(1)
+		go func() {
+			val := make([]byte, rand.Intn(1000))
+			opts := WriterOptions{
+				Comparer:    testkeys.Comparer,
+				BlockSize:   rand.Intn(1 << 10),
+				Compression: NoCompression,
+			}
+			defer wg.Done()
+			f := &memFile{}
+			w := NewWriter(f, opts)
+			for ki := 0; ki < len(keys); ki++ {
+				require.NoError(
+					t,
+					w.Add(base.MakeInternalKey(keys[ki], uint64(ki), InternalKeyKindSet), val),
+				)
+				require.Equal(
+					t, base.DecodeInternalKey(w.dataBlockBuf.dataBlock.curKey).UserKey, keys[ki],
+				)
+			}
+			require.NoError(t, w.Close())
+			require.Equal(t, w.meta.LargestPoint.UserKey, keys[len(keys)-1])
+			r, err := NewMemReader(f.Bytes(), readerOpts)
+			require.NoError(t, err)
+			defer r.Close()
+			it, err := r.NewIter(nil, nil)
+			require.NoError(t, err)
+			defer it.Close()
+			ki := 0
+			for k, v := it.First(); k != nil; k, v = it.Next() {
+				require.Equal(t, k.UserKey, keys[ki])
+				require.Equal(t, v, val)
+				ki++
+			}
+		}()
+	}
+	wg.Wait()
+}
+
 func BenchmarkWriter(b *testing.B) {
 	keys := make([][]byte, 1e6)
 	const keyLen = 24
diff --git a/testdata/level_iter_seek b/testdata/level_iter_seek
index baa56b8d..ff4cab74 100644
--- a/testdata/level_iter_seek
+++ b/testdata/level_iter_seek
@@ -341,3 +341,51 @@ a/c-e#4#1,1:a
 e/<empty>#4,1:e
 i/<empty>#6,1:i
 j/<empty>#7,1:j
+
+# Verify that we do not open files that do not have point keys.
+
+clear
+----
+
+build
+a.SET.9:a
+b.SET.8:b
+----
+0: a#9,1-b#8,1
+
+build
+c.SET.7:c
+d.RANGEDEL.6:e
+f.SET.5:f
+----
+0: a#9,1-b#8,1
+1: c#7,1-f#5,1
+
+build format=pebblev2
+g.RANGEKEYSET.6:h
+----
+0: a#9,1-b#8,1
+1: c#7,1-f#5,1
+2: g#0,21-h#72057594037927935,21
+
+build
+i.SET.4:i
+j.SET.3:j
+----
+0: a#9,1-b#8,1
+1: c#7,1-f#5,1
+2: g#0,21-h#72057594037927935,21
+3: i#4,1-j#3,1
+
+iter
+seek-ge f
+next
+----
+f/<empty>#5,1:f
+i#4,1:i
+
+# The below count should be 2, as we skip over the rangekey-only file.
+
+iters-created
+----
+2
diff --git a/testdata/rangekeys b/testdata/rangekeys
index 0794e00c..7d16c65e 100644
--- a/testdata/rangekeys
+++ b/testdata/rangekeys
@@ -596,3 +596,42 @@ b: (., [b-e) @1=foo)
 a: (a, .)
 c: (., [b-e) @1=foo)
 b: (., [b-e) @1=foo)
+
+# Test a case during limited reverse iteration where a range key covers a
+# portion of the keyspace within the limit. The iterator should NOT pause and
+# should surface the range key.
+
+reset
+----
+
+batch
+del b
+range-key-set a d @1 foo
+----
+wrote 2 keys
+
+combined-iter
+seek-ge z
+prev-limit c
+----
+.
+a: valid (., [a-d) @1=foo)
+
+# Test a case during limited reverse iteration where there exists a range key
+# but it ends before the limit. The iterator should pause.
+
+reset
+----
+
+batch
+del b
+range-key-set a c @1 foo
+----
+wrote 2 keys
+
+combined-iter
+seek-ge z
+prev-limit c
+----
+.
+. at-limit