Repository: cockroachdb/pebble
Pull request #3614 (2024-05-13 15:53:25)

db: avoid unnecessary second opening of range deletion iter
Since #3596 the levelIter began opening a file's range deletion iterator twice: once to pass off to the mergingIter and once to interleave range deletion boundaries among the levelIter output to ensure we don't advance to a subsequent file until the file's range deletions are irrelevant. The implementation in #3596 naively always requested the range deletion iterator from the table cache a second time, even when it already discovered the file contained no range deletions. This caused a small regression in the Pebble YCSB benchmarks.

cockroach-teamcity: 

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


Patch:
diff --git a/db.go b/db.go
index b4b9acd0..34c4e443 100644
--- a/db.go
+++ b/db.go
@@ -271,6 +271,10 @@ type DB struct {
 
 	// The logical size of the current WAL.
 	logSize atomic.Uint64
+	// The number of input bytes to the log. This is the raw size of the
+	// batches written to the WAL, without the overhead of the record
+	// envelopes.
+	logBytesIn atomic.Uint64
 
 	// The number of bytes available on disk.
 	diskAvailBytes atomic.Uint64
@@ -371,10 +375,6 @@ type DB struct {
 			// manager is not protected by mu, but calls to Create must be
 			// serialized, and happen after the previous writer is closed.
 			manager wal.Manager
-			// The number of input bytes to the log. This is the raw size of the
-			// batches written to the WAL, without the overhead of the record
-			// envelopes. Requires DB.mu to be held when read or written.
-			bytesIn uint64
 			// The Writer is protected by commitPipeline.mu. This allows log writes
 			// to be performed without holding DB.mu, but requires both
 			// commitPipeline.mu and DB.mu to be held when rotating the WAL/memtable
@@ -390,7 +390,15 @@ type DB struct {
 		}
 
 		mem struct {
-			// The current mutable memTable.
+			// The current mutable memTable. Readers of the pointer may hold
+			// either DB.mu or commitPipeline.mu.
+			//
+			// Its internal fields are protected by commitPipeline.mu. This
+			// allows batch commits to be performed without DB.mu as long as no
+			// memtable rotation is required.
+			//
+			// Both commitPipeline.mu and DB.mu must be held when rotating the
+			// memtable.
 			mutable *memTable
 			// Queue of flushables (the mutable memtable is at end). Elements are
 			// added to the end of the slice and removed from the beginning. Once an
@@ -930,35 +938,39 @@ func (d *DB) commitWrite(b *Batch, syncWG *sync.WaitGroup, syncErr *error) (*mem
 		}
 	}
 
-	d.mu.Lock()
-
 	var err error
+	// Grab a reference to the memtable. We don't hold DB.mu, but we do hold
+	// d.commit.mu. It's okay for readers of d.mu.mem.mutable to only hold one of
+	// d.commit.mu or d.mu, because memtable rotations require holding both.
+	mem := d.mu.mem.mutable
+	// Batches which contain keys of kind InternalKeyKindIngestSST will
+	// never be applied to the memtable, so we don't need to make room for
+	// write.
 	if !b.ingestedSSTBatch {
-		// Batches which contain keys of kind InternalKeyKindIngestSST will
-		// never be applied to the memtable, so we don't need to make room for
-		// write. For the other cases, switch out the memtable if there was not
-		// enough room to store the batch.
-		err = d.makeRoomForWrite(b)
-	}
-
-	if err == nil && !d.opts.DisableWAL {
-		d.mu.log.bytesIn += uint64(len(repr))
+		// Flushable batches will require a rotation of the memtable regardless,
+		// so only attempt an optimistic reservation of space in the current
+		// memtable if this batch is not a large flushable batch.
+		if b.flushable == nil {
+			err = d.mu.mem.mutable.prepare(b)
+		}
+		if b.flushable != nil || err == arenaskl.ErrArenaFull {
+			// Slow path.
+			// We need to acquire DB.mu and rotate the memtable.
+			func() {
+				d.mu.Lock()
+				defer d.mu.Unlock()
+				err = d.makeRoomForWrite(b)
+				mem = d.mu.mem.mutable
+			}()
+		}
 	}
-
-	// Grab a reference to the memtable while holding DB.mu. Note that for
-	// non-flushable batches (b.flushable == nil) makeRoomForWrite() added a
-	// reference to the memtable which will prevent it from being flushed until
-	// we unreference it. This reference is dropped in DB.commitApply().
-	mem := d.mu.mem.mutable
-
-	d.mu.Unlock()
 	if err != nil {
 		return nil, err
 	}
-
 	if d.opts.DisableWAL {
 		return mem, nil
 	}
+	d.logBytesIn.Add(uint64(len(repr)))
 
 	if b.flushable == nil {
 		size, err = d.mu.log.writer.WriteRecord(repr, wal.SyncOptions{Done: syncWG, Err: syncErr}, b)
@@ -1959,7 +1971,7 @@ func (d *DB) Metrics() *Metrics {
 	// d.atomic.logSize has exceeded that physical size. We allow for this
 	// anomaly.
 	metrics.WAL.PhysicalSize = walStats.LiveFileSize
-	metrics.WAL.BytesIn = d.mu.log.bytesIn // protected by d.mu
+	metrics.WAL.BytesIn = d.logBytesIn.Load()
 	metrics.WAL.Size = d.logSize.Load()
 	for i, n := 0, len(d.mu.mem.queue)-1; i < n; i++ {
 		metrics.WAL.Size += d.mu.mem.queue[i].logSize
@@ -2325,13 +2337,24 @@ func (d *DB) walPreallocateSize() int {
 	return int(size)
 }
 
-func (d *DB) newMemTable(logNum base.DiskFileNum, logSeqNum uint64) (*memTable, *flushableEntry) {
+func (d *DB) newMemTable(
+	logNum base.DiskFileNum, logSeqNum, minSize uint64,
+) (*memTable, *flushableEntry) {
+	targetSize := minSize + uint64(memTableEmptySize)
+	// The targetSize should be less than MemTableSize, because any batch >=
+	// MemTableSize/2 should be treated as a large flushable batch.
+	if targetSize > d.opts.MemTableSize {
+		panic(errors.AssertionFailedf("attempting to allocate memtable larger than MemTableSize"))
+	}
+	// Double until the next memtable size is at least large enough to fit
+	// minSize.
+	for d.mu.mem.nextSize < targetSize {
+		d.mu.mem.nextSize = min(2*d.mu.mem.nextSize, d.opts.MemTableSize)
+	}
 	size := d.mu.mem.nextSize
+	// The next memtable should be double the size, up to Options.MemTableSize.
 	if d.mu.mem.nextSize < d.opts.MemTableSize {
-		d.mu.mem.nextSize *= 2
-		if d.mu.mem.nextSize > d.opts.MemTableSize {
-			d.mu.mem.nextSize = d.opts.MemTableSize
-		}
+		d.mu.mem.nextSize = min(2*d.mu.mem.nextSize, d.opts.MemTableSize)
 	}
 
 	memtblOpts := memTableOptions{
@@ -2411,69 +2434,56 @@ func (d *DB) newFlushableEntry(
 	return fe
 }
 
-// makeRoomForWrite ensures that the memtable has room to hold the contents of
-// Batch. It reserves the space in the memtable and adds a reference to the
-// memtable. The caller must later ensure that the memtable is unreferenced. If
-// the memtable is full, or a nil Batch is provided, the current memtable is
-// rotated (marked as immutable) and a new mutable memtable is allocated. This
-// memtable rotation also causes a log rotation.
+// maybeInduceWriteStall is called before performing a memtable rotation in
+// makeRoomForWrite. In some conditions, we prefer to stall the user's write
+// workload rather than continuing to accept writes that may result in resource
+// exhaustion or prohibitively slow reads.
 //
-// Both DB.mu and commitPipeline.mu must be held by the caller. Note that DB.mu
-// may be released and reacquired.
-func (d *DB) makeRoomForWrite(b *Batch) error {
-	if b != nil && b.ingestedSSTBatch {
-		panic("pebble: invalid function call")
-	}
-
-	force := b == nil || b.flushable != nil
+// There are a couple reasons we might wait to rotate the memtable and
+// instead induce a write stall:
+//  1. If too many memtables have queued, we wait for a flush to finish before
+//     creating another memtable.
+//  2. If L0 read amplification has grown too high, we wait for compactions
+//     to reduce the read amplification before accepting more writes that will
+//     increase write pressure.
+//
+// maybeInduceWriteStall checks these stall conditions, and if present, waits
+// for them to abate.
+func (d *DB) maybeInduceWriteStall(b *Batch) {
 	stalled := false
+	// This function will call EventListener.WriteStallBegin at most once.  If
+	// it does call it, it will call EventListener.WriteStallEnd once before
+	// returning.
 	for {
-		if b != nil && b.flushable == nil {
-			err := d.mu.mem.mutable.prepare(b)
-			if err != arenaskl.ErrArenaFull {
-				if stalled {
-					d.opts.EventListener.WriteStallEnd()
-				}
-				return err
-			}
-		} else if !force {
-			if stalled {
-				d.opts.EventListener.WriteStallEnd()
-			}
-			return nil
-		}
-		// force || err == ErrArenaFull, so we need to rotate the current memtable.
-		{
-			var size uint64
-			for i := range d.mu.mem.queue {
-				size += d.mu.mem.queue[i].totalBytes()
+		var size uint64
+		for i := range d.mu.mem.queue {
+			size += d.mu.mem.queue[i].totalBytes()
+		}
+		// If ElevateWriteStallThresholdForFailover is true, we give an
+		// unlimited memory budget for memtables. This is simpler than trying to
+		// configure an explicit value, given that memory resources can vary.
+		// When using WAL failover in CockroachDB, an OOM risk is worth
+		// tolerating for workloads that have a strict latency SLO. Also, an
+		// unlimited budget here does not mean that the disk stall in the
+		// primary will go unnoticed until the OOM -- CockroachDB is monitoring
+		// disk stalls, and we expect it to fail the node after ~60s if the
+		// primary is stalled.
+		if size >= uint64(d.opts.MemTableStopWritesThreshold)*d.opts.MemTableSize &&
+			!d.mu.log.manager.ElevateWriteStallThresholdForFailover() {
+			// We have filled up the current memtable, but already queued memtables
+			// are still flushing, so we wait.
+			if !stalled {
+				stalled = true
+				d.opts.EventListener.WriteStallBegin(WriteStallBeginInfo{
+					Reason: "memtable count limit reached",
+				})
 			}
-			// If ElevateWriteStallThresholdForFailover is true, we give an
-			// unlimited memory budget for memtables. This is simpler than trying to
-			// configure an explicit value, given that memory resources can vary.
-			// When using WAL failover in CockroachDB, an OOM risk is worth
-			// tolerating for workloads that have a strict latency SLO. Also, an
-			// unlimited budget here does not mean that the disk stall in the
-			// primary will go unnoticed until the OOM -- CockroachDB is monitoring
-			// disk stalls, and we expect it to fail the node after ~60s if the
-			// primary is stalled.
-			if size >= uint64(d.opts.MemTableStopWritesThreshold)*d.opts.MemTableSize &&
-				!d.mu.log.manager.ElevateWriteStallThresholdForFailover() {
-				// We have filled up the current memtable, but already queued memtables
-				// are still flushing, so we wait.
-				if !stalled {
-					stalled = true
-					d.opts.EventListener.WriteStallBegin(WriteStallBeginInfo{
-						Reason: "memtable count limit reached",
-					})
-				}
-				now := time.Now()
-				d.mu.compact.cond.Wait()
-				if b != nil {
-					b.commitStats.MemTableWriteStallDuration += time.Since(now)
-				}
-				continue
+			now := time.Now()
+			d.mu.compact.cond.Wait()
+			if b != nil {
+				b.commitStats.MemTableWriteStallDuration += time.Since(now)
 			}
+			continue
 		}
 		l0ReadAmp := d.mu.versions.currentVersion().L0Sublevels.ReadAmplification()
 		if l0ReadAmp >= d.opts.L0StopWritesThreshold {
@@ -2491,31 +2501,52 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			}
 			continue
 		}
-
-		var newLogNum base.DiskFileNum
-		var prevLogSize uint64
-		if !d.opts.DisableWAL {
-			now := time.Now()
-			newLogNum, prevLogSize = d.rotateWAL()
-			if b != nil {
-				b.commitStats.WALRotationDuration += time.Since(now)
-			}
+		// Not stalled.
+		if stalled {
+			d.opts.EventListener.WriteStallEnd()
 		}
+		return
+	}
+}
 
-		immMem := d.mu.mem.mutable
-		imm := d.mu.mem.queue[len(d.mu.mem.queue)-1]
-		imm.logSize = prevLogSize
-		imm.flushForced = imm.flushForced || (b == nil)
+// makeRoomForWrite rotates the current mutable memtable, ensuring that the
+// resulting mutable memtable has room to hold the contents of the provided
+// Batch. The current memtable is rotated (marked as immutable) and a new
+// mutable memtable is allocated. It reserves space in the new memtable and adds
+// a reference to the memtable. The caller must later ensure that the memtable
+// is unreferenced. This memtable rotation also causes a log rotation.
+//
+// If the current memtable is not full but the caller wishes to trigger a
+// rotation regardless, the caller may pass a nil Batch, and no space in the
+// resulting mutable memtable will be reserved.
+//
+// Both DB.mu and commitPipeline.mu must be held by the caller. Note that DB.mu
+// may be released and reacquired.
+func (d *DB) makeRoomForWrite(b *Batch) error {
+	if b != nil && b.ingestedSSTBatch {
+		panic("pebble: invalid function call")
+	}
+	d.maybeInduceWriteStall(b)
 
-		// If we are manually flushing and we used less than half of the bytes in
-		// the memtable, don't increase the size for the next memtable. This
-		// reduces memtable memory pressure when an application is frequently
-		// manually flushing.
-		if (b == nil) && uint64(immMem.availBytes()) > immMem.totalBytes()/2 {
-			d.mu.mem.nextSize = immMem.totalBytes()
+	var newLogNum base.DiskFileNum
+	var prevLogSize uint64
+	if !d.opts.DisableWAL {
+		now := time.Now()
+		newLogNum, prevLogSize = d.rotateWAL()
+		if b != nil {
+			b.commitStats.WALRotationDuration += time.Since(now)
 		}
+	}
+	immMem := d.mu.mem.mutable
+	imm := d.mu.mem.queue[len(d.mu.mem.queue)-1]
+	imm.logSize = prevLogSize
 
-		if b != nil && b.flushable != nil {
+	var logSeqNum uint64
+	var minSize uint64
+	if b != nil {
+		logSeqNum = b.SeqNum()
+		if b.flushable != nil {
+			logSeqNum += uint64(b.Count())
 			// The batch is too large to fit in the memtable so add it directly to
 			// the immutable queue. The flushable batch is associated with the same
 			// log as the immutable memtable, but logically occurs after it in
@@ -2532,24 +2563,43 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			// for it until it is flushed.
 			entry.releaseMemAccounting = d.opts.Cache.Reserve(int(b.flushable.totalBytes()))
 			d.mu.mem.queue = append(d.mu.mem.queue, entry)
-		}
-
-		var logSeqNum uint64
-		if b != nil {
-			logSeqNum = b.SeqNum()
-			if b.flushable != nil {
-				logSeqNum += uint64(b.Count())
-			}
 		} else {
-			logSeqNum = d.mu.versions.logSeqNum.Load()
+			minSize = b.memTableSize
+		}
+	} else {
+		// b == nil
+		//
+		// This is a manual forced flush.
+		logSeqNum = d.mu.versions.logSeqNum.Load()
+		imm.flushForced = true
+		// If we are manually flushing and we used less than half of the bytes in
+		// the memtable, don't increase the size for the next memtable. This
+		// reduces memtable memory pressure when an application is frequently
+		// manually flushing.
+		if uint64(immMem.availBytes()) > immMem.totalBytes()/2 {
+			d.mu.mem.nextSize = immMem.totalBytes()
 		}
-		d.rotateMemtable(newLogNum, logSeqNum, immMem)
-		force = false
 	}
+	d.rotateMemtable(newLogNum, logSeqNum, immMem, minSize)
+	if b != nil && b.flushable == nil {
+		err := d.mu.mem.mutable.prepare(b)
+		// Reserving enough space for the batch after rotation must never fail.
+		// We pass in a minSize that's equal to b.memtableSize to ensure that
+		// memtable rotation allocates a memtable sufficiently large. We also
+		// held d.commit.mu for the entirety of this function, ensuring that no
+		// other committers may have reserved memory in the new memtable yet.
+		if err == arenaskl.ErrArenaFull {
+			panic(errors.AssertionFailedf("memtable still full after rotation"))
+		}
+		return err
+	}
+	return nil
 }
 
 // Both DB.mu and commitPipeline.mu must be held by the caller.
-func (d *DB) rotateMemtable(newLogNum base.DiskFileNum, logSeqNum uint64, prev *memTable) {
+func (d *DB) rotateMemtable(
+	newLogNum base.DiskFileNum, logSeqNum uint64, prev *memTable, minSize uint64,
+) {
 	// Create a new memtable, scheduling the previous one for flushing. We do
 	// this even if the previous memtable was empty because the DB.Flush
 	// mechanism is dependent on being able to wait for the empty memtable to
@@ -2566,7 +2616,7 @@ func (d *DB) rotateMemtable(newLogNum base.DiskFileNum, logSeqNum uint64, prev *
 	//
 	// NB: prev should be the current mutable memtable.
 	var entry *flushableEntry
-	d.mu.mem.mutable, entry = d.newMemTable(newLogNum, logSeqNum)
+	d.mu.mem.mutable, entry = d.newMemTable(newLogNum, logSeqNum, minSize)
 	d.mu.mem.queue = append(d.mu.mem.queue, entry)
 	// d.logSize tracks the log size of the WAL file corresponding to the most
 	// recent flushable. The log size of the previous mutable memtable no longer
diff --git a/ingest.go b/ingest.go
index c12591b2..64552d08 100644
--- a/ingest.go
+++ b/ingest.go
@@ -1301,7 +1301,7 @@ func (d *DB) handleIngestAsFlushable(
 	// ingested sstables being placed on top of them, but those
 	// memtables would have to be flushed anyways.
 	d.mu.mem.queue = append(d.mu.mem.queue, entry)
-	d.rotateMemtable(newLogNum, nextSeqNum, currMem)
+	d.rotateMemtable(newLogNum, nextSeqNum, currMem, 0 /* minSize */)
 	d.updateReadStateLocked(d.opts.DebugCheck)
 	// TODO(aaditya): is this necessary? we call this already in rotateMemtable above
 	d.maybeScheduleFlush()
diff --git a/internal/base/iterator.go b/internal/base/iterator.go
index 7cbea1b7..7531e5fb 100644
--- a/internal/base/iterator.go
+++ b/internal/base/iterator.go
@@ -107,13 +107,15 @@ type InternalIterator interface {
 	// SeekPrefixGE only checks the upper bound. It is up to the caller to ensure
 	// that key is greater than or equal to the lower bound.
 	//
-	// The prefix argument is used by some InternalIterator implementations (e.g.
-	// sstable.Reader) to avoid expensive operations. This operation is only
-	// useful when a user-defined Split function is supplied to the Comparer for
-	// the DB. The supplied prefix will be the prefix of the given key returned by
-	// that Split function. If the iterator is able to determine that no key with
-	// the prefix exists, it can return (nil,nilv). Unlike SeekGE, this is not an
-	// indication that iteration is exhausted.
+	// The prefix argument is used by some InternalIterator implementations
+	// (e.g.  sstable.Reader) to avoid expensive operations. This operation is
+	// only useful when a user-defined Split function is supplied to the
+	// Comparer for the DB. The supplied prefix will be the prefix of the given
+	// key returned by that Split function. If the iterator is able to determine
+	// that no key with the prefix exists, it can return (nil,nilv). Unlike
+	// SeekGE, this is not an indication that iteration is exhausted. The prefix
+	// byte slice is guaranteed to be stable until the next absolute positioning
+	// operation.
 	//
 	// Note that the iterator may return keys not matching the prefix. It is up
 	// to the caller to check if the prefix matches.
diff --git a/level_iter.go b/level_iter.go
index b799dd60..37a140ab 100644
--- a/level_iter.go
+++ b/level_iter.go
@@ -31,21 +31,13 @@ type internalIterOpts struct {
 //
 // levelIter is used during compaction and as part of the Iterator
 // implementation. When used as part of the Iterator implementation, level
-// iteration needs to "pause" at sstable boundaries if a range deletion iterator
-// is open. In this case, we materialize a "synthetic" InternalKV to return from
-// levelIter. This prevents mergingIter from advancing past the sstable until
-// the sstable contains the smallest (or largest for reverse iteration) key in
-// the merged heap. Note that mergingIter treats a range deletion tombstone
-// returned by the point iterator as a no-op.
-//
-// SeekPrefixGE presents the need for a second type of pausing. If an sstable
-// iterator returns "not found" for a SeekPrefixGE operation, we don't want to
-// advance to the next sstable as the "not found" does not indicate that all of
-// the keys in the sstable are less than the search key. Advancing to the next
-// sstable would cause us to skip over range tombstones, violating
-// correctness. Instead, SeekPrefixGE creates a synthetic boundary key with the
-// kind InternalKeyKindRangeDeletion which will be used to pause the levelIter
-// at the sstable until the mergingIter is ready to advance past it.
+// iteration needs to "pause" at range deletion boundaries if file contains
+// range deletions. In this case, the levelIter uses a keyspan.InterleavingIter
+// to materialize InternalKVs at start and end boundaries of range deletions.
+// This prevents mergingIter from advancing past the sstable until the sstable
+// contains the smallest (or largest for reverse iteration) key in the merged
+// heap. Note that mergingIter treats a range deletion tombstone returned by the
+// point iterator as a no-op.
 type levelIter struct {
 	// The context is stored here since (a) iterators are expected to be
 	// short-lived (since they pin sstables), (b) plumbing a context into every
@@ -60,6 +52,9 @@ type levelIter struct {
 	// recent call to SetBounds.
 	lower []byte
 	upper []byte
+	// prefix holds the iteration prefix when the most recent absolute
+	// positioning method was a SeekPrefixGE.
+	prefix []byte
 	// The iterator options for the currently open table. If
 	// tableOpts.{Lower,Upper}Bound are nil, the corresponding iteration boundary
 	// does not lie within the table bounds.
@@ -161,6 +156,7 @@ func (l *levelIter) init(
 	l.err = nil
 	l.level = level
 	l.logger = opts.getLogger()
+	l.prefix = nil
 	l.lower = opts.LowerBound
 	l.upper = opts.UpperBound
 	l.tableOpts.TableFilter = opts.TableFilter
@@ -570,7 +566,7 @@ func (l *levelIter) loadFile(file *fileMetadata, dir int) loadFileReturnIndicato
 			return noFileLoaded
 		}
 		l.iter = iters.Point()
-		if l.rangeDelIterPtr != nil {
+		if l.rangeDelIterPtr != nil && iters.rangeDeletion != nil {
 			*l.rangeDelIterPtr = iters.rangeDeletion
 			l.rangeDelIterCopy = iters.rangeDeletion
 
@@ -629,6 +625,7 @@ func (l *levelIter) SeekGE(key []byte, flags base.SeekGEFlags) *base.InternalKV
 
 	l.err = nil // clear cached iteration error
 	l.exhaustedDir = 0
+	l.prefix = nil
 	// NB: the top-level Iterator has already adjusted key based on
 	// IterOptions.LowerBound.
 	loadFileIndicator := l.loadFile(l.findFileGE(key, flags), +1)
@@ -651,9 +648,9 @@ func (l *levelIter) SeekPrefixGE(prefix, key []byte, flags base.SeekGEFlags) *ba
 	if invariants.Enabled && l.lower != nil && l.cmp(key, l.lower) < 0 {
 		panic(errors.AssertionFailedf("levelIter SeekGE to key %q violates lower bound %q", key, l.lower))
 	}
-
 	l.err = nil // clear cached iteration error
 	l.exhaustedDir = 0
+	l.prefix = prefix
 
 	// NB: the top-level Iterator has already adjusted key based on
 	// IterOptions.LowerBound.
@@ -673,19 +670,6 @@ func (l *levelIter) SeekPrefixGE(prefix, key []byte, flags base.SeekGEFlags) *ba
 	if err := l.iter.Error(); err != nil {
 		return nil
 	}
-	// It is possible that we are here because bloom filter matching failed. In
-	// that case it is likely that all keys matching the prefix are wholly
-	// within the current file and cannot be in the subsequent file. In that
-	// case we don't want to go to the next file, since loading and seeking in
-	// there has some cost. Additionally, for sparse key spaces, loading the
-	// next file will defeat the optimization for the next SeekPrefixGE that is
-	// called with flags.TrySeekUsingNext(), since for sparse key spaces it is
-	// likely that the next key will also be contained in the current file.
-	n := l.split(l.iterFile.LargestPointKey.UserKey)
-	if l.cmp(prefix, l.iterFile.LargestPointKey.UserKey[:n]) < 0 {
-		l.exhaustedForward()
-		return nil
-	}
 	return l.verify(l.skipEmptyFileForward())
 }
 
@@ -696,6 +680,7 @@ func (l *levelIter) SeekLT(key []byte, flags base.SeekLTFlags) *base.InternalKV
 
 	l.err = nil // clear cached iteration error
 	l.exhaustedDir = 0
+	l.prefix = nil
 
 	// NB: the top-level Iterator has already adjusted key based on
 	// IterOptions.UpperBound.
@@ -716,6 +701,7 @@ func (l *levelIter) First() *base.InternalKV {
 
 	l.err = nil // clear cached iteration error
 	l.exhaustedDir = 0
+	l.prefix = nil
 
 	// NB: the top-level Iterator will call SeekGE if IterOptions.LowerBound is
 	// set.
@@ -736,6 +722,7 @@ func (l *levelIter) Last() *base.InternalKV {
 
 	l.err = nil // clear cached iteration error
 	l.exhaustedDir = 0
+	l.prefix = nil
 
 	// NB: the top-level Iterator will call SeekLT if IterOptions.UpperBound is
 	// set.
@@ -835,6 +822,27 @@ func (l *levelIter) skipEmptyFileForward() *base.InternalKV {
 			l.exhaustedForward()
 			return nil
 		}
+
+		// If the iterator is in prefix iteration mode, it's possible that we
+		// are here because bloom filter matching failed. In that case it is
+		// likely that all keys matching the prefix are wholly within the
+		// current file and cannot be in a subsequent file. In that case we
+		// don't want to go to the next file, since loading and seeking in there
+		// has some cost.
+		//
+		// This is not just an optimization. We must not advance to the next
+		// file if the current file might possibly contain keys relevant to any
+		// prefix greater than our current iteration prefix. If we did, a
+		// subsequent SeekPrefixGE with TrySeekUsingNext could mistakenly skip
+		// the file's relevant keys.
+		if l.prefix != nil {
+			fileLargestPrefix := l.iterFile.LargestPointKey.UserKey[:l.split(l.iterFile.LargestPointKey.UserKey)]
+			if l.cmp(fileLargestPrefix, l.prefix) > 0 {
+				l.exhaustedForward()
+				return nil
+			}
+		}
+
 		// Current file was exhausted. Move to the next file.
 		if l.loadFile(l.files.Next(), +1) == noFileLoaded {
 			l.exhaustedForward()
diff --git a/objstorage/objstorageprovider/readahead.go b/objstorage/objstorageprovider/readahead.go
index e07017c9..e29155d8 100644
--- a/objstorage/objstorageprovider/readahead.go
+++ b/objstorage/objstorageprovider/readahead.go
@@ -40,40 +40,7 @@ func makeReadaheadState(maxReadaheadSize int64) readaheadState {
 }
 
 func (rs *readaheadState) recordCacheHit(offset, blockLength int64) {
-	currentReadEnd := offset + blockLength
-	if rs.numReads >= minFileReadsForReadahead {
-		if currentReadEnd >= rs.limit && offset <= rs.limit+rs.maxReadaheadSize {
-			// This is a read that would have resulted in a readahead, had it
-			// not been a cache hit.
-			rs.limit = currentReadEnd
-			return
-		}
-		if currentReadEnd < rs.limit-rs.prevSize || offset > rs.limit+rs.maxReadaheadSize {
-			// We read too far away from rs.limit to benefit from readahead in
-			// any scenario. Reset all variables.
-			rs.numReads = 1
-			rs.limit = currentReadEnd
-			rs.size = initialReadaheadSize
-			rs.prevSize = 0
-			return
-		}
-		// Reads in the range [rs.limit - rs.prevSize, rs.limit] end up
-		// here. This is a read that is potentially benefitting from a past
-		// readahead.
-		return
-	}
-	if currentReadEnd >= rs.limit && offset <= rs.limit+rs.maxReadaheadSize {
-		// Blocks are being read sequentially and would benefit from readahead
-		// down the line.
-		rs.numReads++
-		return
-	}
-	// We read too far ahead of the last read, or before it. This indicates
-	// a random read, where readahead is not desirable. Reset all variables.
-	rs.numReads = 1
-	rs.limit = currentReadEnd
-	rs.size = initialReadaheadSize
-	rs.prevSize = 0
+	_ = rs.maybeReadaheadOrCacheHit(offset, blockLength, false)
 }
 
 // maybeReadahead updates state and determines whether to issue a readahead /
@@ -81,6 +48,13 @@ func (rs *readaheadState) recordCacheHit(offset, blockLength int64) {
 // Returns a size value (greater than 0) that should be prefetched if readahead
 // would be beneficial.
 func (rs *readaheadState) maybeReadahead(offset, blockLength int64) int64 {
+	return rs.maybeReadaheadOrCacheHit(offset, blockLength, true)
+}
+
+// The return value should be ignored if !readahead.
+func (rs *readaheadState) maybeReadaheadOrCacheHit(
+	offset, blockLength int64, readahead bool,
+) int64 {
 	currentReadEnd := offset + blockLength
 	if rs.numReads >= minFileReadsForReadahead {
 		// The minimum threshold of sequential reads to justify reading ahead
@@ -119,18 +93,22 @@ func (rs *readaheadState) maybeReadahead(offset, blockLength int64) int64 {
 			//
 			//
 			rs.numReads++
-			rs.limit = offset + rs.size
-			rs.prevSize = rs.size
-			// Increase rs.size for the next read.
-			rs.size *= 2
-			if rs.size > rs.maxReadaheadSize {
-				rs.size = rs.maxReadaheadSize
+			if readahead {
+				rs.limit = offset + rs.size
+				rs.prevSize = rs.size
+				// Increase rs.size for the next read.
+				rs.size *= 2
+				if rs.size > rs.maxReadaheadSize {
+					rs.size = rs.maxReadaheadSize
+				}
+			} else {
+				// This is a read that would have resulted in a readahead, had it
+				// not been a cache hit.
+				rs.limit = currentReadEnd
 			}
 			return rs.prevSize
 		}
 		if currentReadEnd < rs.limit-rs.prevSize || offset > rs.limit+rs.maxReadaheadSize {
-			// The above conditional has rs.limit > rs.prevSize to confirm that
-			// rs.limit - rs.prevSize would not underflow.
 			// We read too far away from rs.limit to benefit from readahead in
 			// any scenario. Reset all variables.
 			// The case where we read too far ahead:
@@ -156,7 +134,16 @@ func (rs *readaheadState) maybeReadahead(offset, blockLength int64) int64 {
 
 			return 0
 		}
-		// Reads in the range [rs.limit - rs.prevSize, rs.limit] end up
+		// The previous if-block predicates were all false. This mechanically implies:
+		//
+		// INVARIANT:
+		//  !(currentReadEnd >= rs.limit && offset <= rs.limit+rs.maxReadaheadSize) &&
+		//  !(currentReadEnd < rs.limit-rs.prevSize || offset > rs.limit+rs.maxReadaheadSize)
+		// Which mechanically simplifies to:
+		//  currentReadEnd < rs.limit  && currentReadEnd >= rs.limit-rs.prevSize &&
+		//  offset <= rs.limit+rs.maxReadaheadSize
+		//
+		// So reads in the range [rs.limit - rs.prevSize, rs.limit] end up
 		// here. This is a read that is potentially benefitting from a past
 		// readahead, but there's no reason to issue a readahead call at the
 		// moment.
@@ -171,6 +158,8 @@ func (rs *readaheadState) maybeReadahead(offset, blockLength int64) int64 {
 		rs.numReads++
 		return 0
 	}
+	// Not yet at the numReads threshold to justify readahead. But we want to
+	// capture whether readahead will be beneficial in the future.
 	if currentReadEnd >= rs.limit && offset <= rs.limit+rs.maxReadaheadSize {
 		// Blocks are being read sequentially and would benefit from readahead
 		// down the line.
@@ -182,6 +171,16 @@ func (rs *readaheadState) maybeReadahead(offset, blockLength int64) int64 {
 		//                offset    currentReadEnd
 		//
 		rs.numReads++
+		// It is possible to fall here when rs.limit has not been initialized. If
+		// we don't initialize, rs.limit, it is possible that the first read
+		// offset was at rs.limit+rs.maxReadaheadSize-delta and the enclosing
+		// if-block predicate was true, and the next read is sequential but has
+		// offset > rs.limit+rs.maxReadaheadSize (if we left rs.limit at 0), and
+		// the enclosing if-block predicate will be false and we will incorrectly
+		// think that readahead is not beneficial. The same issue arises if
+		// rs.limit has been initialized and currentReadEnd is advancing past
+		// rs.limit.
+		rs.limit = currentReadEnd
 		return 0
 	}
 	// We read too far ahead of the last read, or before it. This indicates
diff --git a/open.go b/open.go
index 21336ac9..cab6196e 100644
--- a/open.go
+++ b/open.go
@@ -341,7 +341,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	// sequence number of the first batch that will be inserted.
 	if !d.opts.ReadOnly {
 		var entry *flushableEntry
-		d.mu.mem.mutable, entry = d.newMemTable(0 /* logNum */, d.mu.versions.logSeqNum.Load())
+		d.mu.mem.mutable, entry = d.newMemTable(0 /* logNum */, d.mu.versions.logSeqNum.Load(), 0 /* minSize */)
 		d.mu.mem.queue = append(d.mu.mem.queue, entry)
 	}
 
@@ -808,7 +808,7 @@ func (d *DB) replayWAL(
 		if mem != nil {
 			return
 		}
-		mem, entry = d.newMemTable(base.DiskFileNum(ll.Num), seqNum)
+		mem, entry = d.newMemTable(base.DiskFileNum(ll.Num), seqNum, 0 /* minSize */)
 		if d.opts.ReadOnly {
 			d.mu.mem.mutable = mem
 			d.mu.mem.queue = append(d.mu.mem.queue, entry)


Test patch:
diff --git a/data_test.go b/data_test.go
index 3ee315c1..5ea1ba66 100644
--- a/data_test.go
+++ b/data_test.go
@@ -842,6 +842,16 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 			for _, levelOpts := range opts.Levels {
 				levelOpts.BlockSize = size
 			}
+		case "bloom-bits-per-key":
+			v, err := strconv.Atoi(arg.Vals[0])
+			if err != nil {
+				return nil, err
+			}
+			fp := bloom.FilterPolicy(v)
+			opts.Filters = map[string]FilterPolicy{fp.Name(): fp}
+			for i := range opts.Levels {
+				opts.Levels[i].FilterPolicy = fp
+			}
 		case "format-major-version":
 			fmv, err := strconv.Atoi(arg.Vals[0])
 			if err != nil {
diff --git a/objstorage/objstorageprovider/readahead_test.go b/objstorage/objstorageprovider/readahead_test.go
index a44d78ac..86aff7b5 100644
--- a/objstorage/objstorageprovider/readahead_test.go
+++ b/objstorage/objstorageprovider/readahead_test.go
@@ -20,9 +20,7 @@ func TestMaybeReadahead(t *testing.T) {
 		cacheHit := false
 		switch d.Cmd {
 		case "reset":
-			rs.size = initialReadaheadSize
-			rs.limit = 0
-			rs.numReads = 0
+			rs = makeReadaheadState(rs.maxReadaheadSize)
 			return ""
 
 		case "cache-read":
diff --git a/objstorage/objstorageprovider/testdata/readahead b/objstorage/objstorageprovider/testdata/readahead
index d2bb2173..adfe4ec6 100644
--- a/objstorage/objstorageprovider/testdata/readahead
+++ b/objstorage/objstorageprovider/testdata/readahead
@@ -8,7 +8,7 @@ readahead:  0
 numReads:   1
 size:       65536
 prevSize:   0
-limit:      0
+limit:      2064
 
 read
 2096, 16
@@ -17,7 +17,7 @@ readahead:  0
 numReads:   2
 size:       65536
 prevSize:   0
-limit:      0
+limit:      2112
 
 read
 2112, 16
@@ -84,7 +84,7 @@ readahead:  0
 numReads:   2
 size:       65536
 prevSize:   0
-limit:      16208
+limit:      16209
 
 # The next read is too far ahead to benefit from readahead
 # (i.e. 540497 > 16208 (limit) + (512 << 10) (maxReadaheadSize))
@@ -126,45 +126,49 @@ readahead:  0
 numReads:   2
 size:       65536
 prevSize:   0
-limit:      16
+limit:      7796
 
 read
-7680, 16
+7800, 16
 ----
 readahead:  65536
 numReads:   3
 size:       131072
 prevSize:   65536
-limit:      73216
+limit:      73336
 
 read
-7780, 16
----
+7880, 16
+----
 readahead:  0
 numReads:   4
 size:       131072
 prevSize:   65536
-limit:      73216
+limit:      73336
 
 read
-7880, 16
+7980, 16
 ----
-expected 2 args: offset, size
+readahead:  0
+numReads:   5
+size:       131072
+prevSize:   65536
+limit:      73336
 
 read
-7980, 16
+8080, 16
 ----
 readahead:  0
-numReads:   4
+numReads:   6
 size:       131072
 prevSize:   65536
-limit:      73216
+limit:      73336
 
 read
 73416, 16
 ----
 readahead:  131072
-numReads:   5
+numReads:   7
 size:       262144
 prevSize:   131072
 limit:      204488
@@ -173,7 +177,7 @@ read
 204488, 16
 ----
 readahead:  262144
-numReads:   6
+numReads:   8
 size:       262144
 prevSize:   262144
 limit:      466632
@@ -184,7 +188,7 @@ read
 466632, 16
 ----
 readahead:  262144
-numReads:   7
+numReads:   9
 size:       262144
 prevSize:   262144
 limit:      728776
@@ -195,7 +199,7 @@ cache-read
 728770, 16
 ----
 readahead:  0
-numReads:   7
+numReads:   10
 size:       262144
 prevSize:   262144
 limit:      728786
@@ -204,7 +208,7 @@ read
 728780, 16
 ----
 readahead:  262144
-numReads:   8
+numReads:   11
 size:       262144
 prevSize:   262144
 limit:      990924
@@ -219,3 +223,36 @@ numReads:   1
 size:       65536
 prevSize:   0
 limit:      1216
+
+reset
+----
+
+# Offset 240KiB, length 32KiB.
+read
+245760, 32768
+----
+readahead:  0
+numReads:   1
+size:       65536
+prevSize:   0
+limit:      278528
+
+# Offset 280KiB, length 32KiB.
+read
+286720, 32768
+----
+readahead:  0
+numReads:   2
+size:       65536
+prevSize:   0
+limit:      319488
+
+# Offset 320KiB, length 32KiB.
+read
+327680, 32768
+----
+readahead:  65536
+numReads:   3
+size:       131072
+prevSize:   65536
+limit:      393216
diff --git a/open_test.go b/open_test.go
index 43d1237f..31f9a6a7 100644
--- a/open_test.go
+++ b/open_test.go
@@ -713,9 +713,11 @@ func TestOpenWALReplay(t *testing.T) {
 				}
 			}
 			require.Equal(t, 0, sstCount)
-			// The memtable size starts at 256KB and doubles up to 32MB so we expect 5
-			// logs (one for each doubling).
-			require.Equal(t, 7, logCount)
+			// The memtable size starts at 256KB and doubles up to 32MB. But,
+			// we'll jump to a power of two large enough if a batch motivating a
+			// rotation is larger than the next power of two. We expect 3 WALs
+			// here.
+			require.Equal(t, 3, logCount)
 
 			// Re-open the DB with a smaller memtable. Values for 1, 2 will fit in the first memtable;
 			// value for 3 will go in the next memtable; value for 4 will be in a flushable batch
diff --git a/testdata/ingest b/testdata/ingest
index fc6a64bf..4be171bd 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -53,7 +53,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 569B
 Block cache: 6 entries (945B)  hit rate: 30.8%
-Table cache: 1 entries (760B)  hit rate: 66.7%
+Table cache: 1 entries (760B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
diff --git a/testdata/iter_histories/prefix_iteration b/testdata/iter_histories/prefix_iteration
index 1ea878ea..d977b2ec 100644
--- a/testdata/iter_histories/prefix_iteration
+++ b/testdata/iter_histories/prefix_iteration
@@ -336,3 +336,29 @@ seek-prefix-ge c@9
 ----
 .
 .
+
+# Regression test for #3610.
+#
+# Similar to the above case, this test consists of two SeekPrefixGEs for
+# ascending keys, resulting in TrySeekUsingNext()=true for the second seek.
+# Previously, during the first SeekPrefixGE the mergingIter could Next the
+# levelIter beyond the file containing point keys relevant to both seeks.
+
+define bloom-bits-per-key=100
+L4
+  b@0.SET.10:b@0
+L5
+  b@8.RANGEDEL.3:b@1
+  c@3.SET.0:c@3
+----
+L4:
+  000004:[b@0#10,SET-b@0#10,SET]
+L5:
+  000005:[b@8#3,RANGEDEL-c@3#0,SET]
+
+combined-iter
+seek-prefix-ge b@10
+seek-prefix-ge c@10
+----
+b@0: (b@0, .)
+c@3: (c@3, .)
diff --git a/testdata/level_iter_seek b/testdata/level_iter_seek
index 74a4691e..8a1e0e4d 100644
--- a/testdata/level_iter_seek
+++ b/testdata/level_iter_seek
@@ -408,10 +408,10 @@ next
 f#5,SET:f
 i#4,SET:i
 
-# The below count should be 4, as we skip over the rangekey-only file.
+# The below count should be 3, as we skip over the rangekey-only file.
 # TODO(jackson): When we stop opening range deletion iterators twice, this
 # should be 2.
 
 iters-created
 ----
-4
+3
diff --git a/testdata/metrics b/testdata/metrics
index 04e93ab5..678d5180 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -73,7 +73,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 589B
 Block cache: 3 entries (484B)  hit rate: 0.0%
-Table cache: 1 entries (760B)  hit rate: 50.0%
+Table cache: 1 entries (760B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -128,7 +128,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 595B
 Block cache: 5 entries (946B)  hit rate: 33.3%
-Table cache: 2 entries (1.5KB)  hit rate: 75.0%
+Table cache: 2 entries (1.5KB)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 2
@@ -170,7 +170,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 595B
 Block cache: 5 entries (946B)  hit rate: 33.3%
-Table cache: 2 entries (1.5KB)  hit rate: 75.0%
+Table cache: 2 entries (1.5KB)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 2
@@ -209,7 +209,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 595B
 Block cache: 3 entries (484B)  hit rate: 33.3%
-Table cache: 1 entries (760B)  hit rate: 75.0%
+Table cache: 1 entries (760B)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -252,7 +252,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 595B
 Block cache: 0 entries (0B)  hit rate: 33.3%
-Table cache: 0 entries (0B)  hit rate: 75.0%
+Table cache: 0 entries (0B)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -322,7 +322,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 2.6KB
 Block cache: 0 entries (0B)  hit rate: 33.3%
-Table cache: 0 entries (0B)  hit rate: 75.0%
+Table cache: 0 entries (0B)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -376,7 +376,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 2.0KB
 Block cache: 0 entries (0B)  hit rate: 14.3%
-Table cache: 0 entries (0B)  hit rate: 64.3%
+Table cache: 0 entries (0B)  hit rate: 58.3%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -479,7 +479,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 4.3KB
 Block cache: 12 entries (1.9KB)  hit rate: 16.7%
-Table cache: 1 entries (760B)  hit rate: 64.7%
+Table cache: 1 entries (760B)  hit rate: 60.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -541,7 +541,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 6.1KB
 Block cache: 12 entries (1.9KB)  hit rate: 16.7%
-Table cache: 1 entries (760B)  hit rate: 64.7%
+Table cache: 1 entries (760B)  hit rate: 60.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0