Repository: cockroachdb/pebble
Pull request #2063 (2022-10-27 14:09:49)

crossversion: use op-count distribution with low lower bound
Use an `-ops` distribution with a low lower bound, so that some versions do very little work before upgrading to the next version. This helps exercise state from version n that survives to versions ≥ n+2.

Informs #2042.

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


Patch:
diff --git a/format_major_version.go b/format_major_version.go
index 5d94128b..9b6546e9 100644
--- a/format_major_version.go
+++ b/format_major_version.go
@@ -412,14 +412,22 @@ func (d *DB) compactMarkedFilesLocked() error {
 		if err := d.closed.Load(); err != nil {
 			return err.(error)
 		}
-		// NB: Waiting on this condition variable drops d.mu while blocked.
-		d.mu.compact.cond.Wait()
 
-		// Some flush or compaction was scheduled or completed. Loop again to
-		// check again for files that must be compacted. The next iteration may
-		// find same file again, but that's okay. It'll eventually succeed in
-		// scheduling the compaction and eventually be woken by its completion.
+		// Some flush or compaction may have scheduled or completed while we waited
+		// for the manifest lock in maybeScheduleCompactionPicker. Get the latest
+		// Version before waiting on a compaction.
 		curr = d.mu.versions.currentVersion()
+
+		// Only wait on compactions if there are files still marked for compaction.
+		// NB: Waiting on this condition variable drops d.mu while blocked.
+		if curr.Stats.MarkedForCompaction > 0 {
+			if d.mu.compact.compactingCount == 0 {
+				panic("expected a compaction of marked files in progress")
+			}
+			d.mu.compact.cond.Wait()
+			// Refresh the current version again.
+			curr = d.mu.versions.currentVersion()
+		}
 	}
 	return nil
 }
diff --git a/internal/metamorphic/ops.go b/internal/metamorphic/ops.go
index a130074f..6c1091b4 100644
--- a/internal/metamorphic/ops.go
+++ b/internal/metamorphic/ops.go
@@ -18,7 +18,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/errorfs"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/private"
-	"github.com/cockroachdb/pebble/internal/testkeys/blockprop"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 )
@@ -824,12 +823,12 @@ func iterOptions(o iterOpts) *pebble.IterOptions {
 	}
 	if opts.RangeKeyMasking.Suffix != nil {
 		opts.RangeKeyMasking.Filter = func() pebble.BlockPropertyFilterMask {
-			return blockprop.NewMaskingFilter()
+			return sstable.NewTestKeysMaskingFilter()
 		}
 	}
 	if o.filterMax > 0 {
 		opts.PointKeyFilters = []pebble.BlockPropertyFilter{
-			blockprop.NewBlockPropertyFilter(o.filterMin, o.filterMax),
+			sstable.NewTestKeysBlockPropertyFilter(o.filterMin, o.filterMax),
 		}
 	}
 	return opts
diff --git a/internal/metamorphic/options.go b/internal/metamorphic/options.go
index ffe612f5..7f6bc62b 100644
--- a/internal/metamorphic/options.go
+++ b/internal/metamorphic/options.go
@@ -17,7 +17,7 @@ import (
 	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/cockroachdb/pebble/internal/testkeys/blockprop"
+	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"golang.org/x/exp/rand"
 )
@@ -404,5 +404,5 @@ func moveLogs(fs vfs.FS, srcDir, dstDir string) error {
 }
 
 var blockPropertyCollectorConstructors = []func() pebble.BlockPropertyCollector{
-	blockprop.NewBlockPropertyCollector,
+	sstable.NewTestKeysBlockPropertyCollector,
 }
diff --git a/iterator.go b/iterator.go
index 21106ea0..f7569ec6 100644
--- a/iterator.go
+++ b/iterator.go
@@ -380,6 +380,14 @@ const (
 	seekPrefixGELastPositioningOp
 	seekGELastPositioningOp
 	seekLTLastPositioningOp
+	// invalidatedLastPositionOp is similar to unknownLastPositionOp and the
+	// only reason to distinguish this is for the wider set of SeekGE
+	// optimizations we permit for the external iterator Iterator.forwardOnly
+	// case. Most code predicates should be doing equality comparisons with one
+	// of the seek* enum values, so this duplication should not result in code
+	// of the form:
+	//  if unknownLastPositionOp || invalidLastPositionOp
+	invalidatedLastPositionOp
 )
 
 // Limited iteration mode. Not for use with prefix iteration.
@@ -1124,8 +1132,9 @@ func (i *Iterator) SeekGEWithLimit(key []byte, limit []byte) IterValidityState {
 	//
 	// TODO(jackson): This optimization should be obsolete once we introduce and
 	// use the NextPrefix iterator positioning operation.
-	if seekInternalIter && i.forwardOnly && i.pos == iterPosCurForward && !hasPrefix &&
-		i.iterValidityState == IterValid && i.cmp(key, i.iterKey.UserKey) > 0 {
+	if seekInternalIter && i.forwardOnly && lastPositioningOp != invalidatedLastPositionOp &&
+		i.pos == iterPosCurForward && !hasPrefix && i.iterValidityState == IterValid &&
+		i.cmp(key, i.iterKey.UserKey) > 0 {
 		flags = flags.EnableTrySeekUsingNext()
 		if invariants.Enabled && flags.TrySeekUsingNext() && !i.forceEnableSeekOpt && disableSeekOpt(key, uintptr(unsafe.Pointer(i))) {
 			flags = flags.DisableTrySeekUsingNext()
@@ -2262,7 +2271,7 @@ func (i *Iterator) SetOptions(o *IterOptions) {
 }
 
 func (i *Iterator) invalidate() {
-	i.lastPositioningOp = unknownLastPositionOp
+	i.lastPositioningOp = invalidatedLastPositionOp
 	i.hasPrefix = false
 	i.iterKey = nil
 	i.iterValue = nil
diff --git a/sstable/reader.go b/sstable/reader.go
index d16d8b3d..72673307 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -104,6 +104,75 @@ type Iterator interface {
 	SetCloseHook(fn func(i Iterator) error)
 }
 
+
+// Iterator positioning optimizations and singleLevelIterator and
+// twoLevelIterator:
+//
+// An iterator is absolute positioned using one of the Seek or First or Last
+// calls. After absolute positioning, there can be relative positioning done
+// by stepping using Prev or Next.
+//
+// We implement optimizations below where an absolute positioning call can in
+// some cases use the current position to do less work. To understand these,
+// we first define some terms. An iterator is bounds-exhausted if the bounds
+// (upper of lower) have been reached. An iterator is data-exhausted if it has
+// the reached the end of the data (forward or reverse) in the sstable. A
+// singleLevelIterator only knows a local-data-exhausted property since when
+// it is used as part of a twoLevelIterator, the twoLevelIterator can step to
+// the next lower-level index block.
+//
+// The bounds-exhausted property is tracked by
+// singleLevelIterator.exhaustedBounds being +1 (upper bound reached) or -1
+// (lower bound reached). The same field is reused by twoLevelIterator. Either
+// may notice the exhaustion of the bound and set it. Note that if
+// singleLevelIterator sets this property, it is not a local property (since
+// the bound has been reached regardless of whether this is in the context of
+// the twoLevelIterator or not).
+//
+// The data-exhausted property is tracked in a more subtle manner. We define
+// two predicates:
+// - partial-local-data-exhausted (PLDE):
+//   i.data.isDataInvalidated() || !i.data.valid()
+// - partial-global-data-exhausted (PGDE):
+//   i.index.isDataInvalidated() || !i.index.valid() || i.data.isDataInvalidated() ||
+//   !i.data.valid()
+//
+// PLDE is defined for a singleLevelIterator. PGDE is defined for a
+// twoLevelIterator. Oddly, in our code below the singleLevelIterator does not
+// know when it is part of a twoLevelIterator so it does not know when its
+// property is local or global.
+//
+// Now to define data-exhausted:
+// - Prerequisite: we must know that the iterator has been positioned and
+//   i.err is nil.
+// - bounds-exhausted must not be true:
+//   If bounds-exhausted is true, we have incomplete knowledge of
+//   data-exhausted since PLDE or PGDE could be true because we could have
+//   chosen not to load index block or data block and figured out that the
+//   bound is exhausted (due to block property filters filtering out index and
+//   data blocks and going past the bound on the top level index block). Note
+//   that if we tried to separate out the BPF case from others we could
+//   develop more knowledge here. - !PLDE or !PGDE of course imply that
+//   data-exhausted is not true.
+//
+// An implication of the above is that if we are going to somehow utilize
+// knowledge of data-exhausted in an optimization, we must not forget the
+// existing value of bounds-exhausted since by forgetting the latter we can
+// erroneously think that data-exhausted is true. Bug #2036 was due to this
+// forgetting.
+//
+// Now to the two categories of optimizations we currently have:
+// - Monotonic bounds optimization that reuse prior iterator position when
+//   doing seek: These only work with !data-exhausted. We could choose to make
+//   these work with data-exhausted but have not bothered because in the
+//   context of a DB if data-exhausted were true, the DB would move to the
+//   next file in the level. Note that this behavior of moving to the next
+//   file is not necessarily true for L0 files, so there could be some benefit
+//   in the future in this optimization. See the WARNING-data-exhausted
+//   comments if trying to optimize this in the future.
+// - TrySeekUsingNext optimizations: these work regardless of exhaustion
+//   state.
+
 // singleLevelIterator iterates over an entire table of data. To seek for a given
 // key, it first looks in the index for the block that contains that key, and then
 // looks inside that block.
@@ -209,7 +278,10 @@ type singleLevelIterator struct {
 	// exhaustedBounds represents whether the iterator is exhausted for
 	// iteration by reaching the upper or lower bound. +1 when exhausted
 	// the upper bound, -1 when exhausted the lower bound, and 0 when
-	// neither. It is used for invariant checking.
+	// neither. exhaustedBounds is also used for the TrySeekUsingNext
+	// optimization in twoLevelIterator and singleLevelIterator. Care should be
+	// taken in setting this in twoLevelIterator before calling into
+	// singleLevelIterator, given that these two iterators share this field.
 	exhaustedBounds int8
 
 	// maybeFilteredKeysSingleLevel indicates whether the last iterator
@@ -603,12 +675,23 @@ func (i *singleLevelIterator) recordOffset() uint64 {
 // package. Note that SeekGE only checks the upper bound. It is up to the
 // caller to ensure that key is greater than or equal to the lower bound.
 func (i *singleLevelIterator) SeekGE(key []byte, flags base.SeekGEFlags) (*InternalKey, []byte) {
-	// The i.exhaustedBounds comparison indicates that the upper bound was
-	// reached. The i.data.isDataInvalidated() indicates that the sstable was
-	// exhausted.
-	if flags.TrySeekUsingNext() && (i.exhaustedBounds == +1 || i.data.isDataInvalidated()) {
-		// Already exhausted, so return nil.
-		return nil, nil
+	if flags.TrySeekUsingNext() {
+		// The i.exhaustedBounds comparison indicates that the upper bound was
+		// reached. The i.data.isDataInvalidated() indicates that the sstable was
+		// exhausted.
+		if (i.exhaustedBounds == +1 || i.data.isDataInvalidated()) && i.err == nil {
+			// Already exhausted, so return nil.
+			return nil, nil
+		}
+		if i.err != nil {
+			// The current iterator position cannot be used.
+			flags = flags.DisableTrySeekUsingNext()
+		}
+		// INVARIANT: flags.TrySeekUsingNext() => i.err == nil &&
+		// !i.exhaustedBounds==+1 && !i.data.isDataInvalidated(). That is,
+		// data-exhausted and bounds-exhausted, as defined earlier, are both
+		// false. Ths makes it safe to clear out i.exhaustedBounds and i.err
+		// before calling into seekGEHelper.
 	}
 
 	i.exhaustedBounds = 0
@@ -748,6 +831,11 @@ func (i *singleLevelIterator) SeekPrefixGE(
 func (i *singleLevelIterator) seekPrefixGE(
 	prefix, key []byte, flags base.SeekGEFlags, checkFilter bool,
 ) (k *InternalKey, value []byte) {
+	// NOTE: prefix is only used for bloom filter checking and not later work in
+	// this method. Hence, we can use the existing iterator position if the last
+	// SeekPrefixGE did not fail bloom filter matching.
+
+	err := i.err
 	i.err = nil // clear cached iteration error
 	if checkFilter && i.reader.tableFilter != nil {
 		if !i.lastBloomFilterMatched {
@@ -775,12 +863,23 @@ func (i *singleLevelIterator) seekPrefixGE(
 		}
 		i.lastBloomFilterMatched = true
 	}
-	// The i.exhaustedBounds comparison indicates that the upper bound was
-	// reached. The i.data.isDataInvalidated() indicates that the sstable was
-	// exhausted.
-	if flags.TrySeekUsingNext() && (i.exhaustedBounds == +1 || i.data.isDataInvalidated()) {
-		// Already exhausted, so return nil.
-		return nil, nil
+	if flags.TrySeekUsingNext() {
+		// The i.exhaustedBounds comparison indicates that the upper bound was
+		// reached. The i.data.isDataInvalidated() indicates that the sstable was
+		// exhausted.
+		if (i.exhaustedBounds == +1 || i.data.isDataInvalidated()) && err == nil {
+			// Already exhausted, so return nil.
+			return nil, nil
+		}
+		if err != nil {
+			// The current iterator position cannot be used.
+			flags = flags.DisableTrySeekUsingNext()
+		}
+		// INVARIANT: flags.TrySeekUsingNext() => err == nil &&
+		// !i.exhaustedBounds==+1 && !i.data.isDataInvalidated(). That is,
+		// data-exhausted and bounds-exhausted, as defined earlier, are both
+		// false. Ths makes it safe to clear out i.exhaustedBounds and i.err
+		// before calling into seekGEHelper.
 	}
 	// Bloom filter matches, or skipped, so this method will position the
 	// iterator.
@@ -1494,9 +1593,18 @@ func (i *twoLevelIterator) MaybeFilteredKeys() bool {
 // package. Note that SeekGE only checks the upper bound. It is up to the
 // caller to ensure that key is greater than or equal to the lower bound.
 func (i *twoLevelIterator) SeekGE(key []byte, flags base.SeekGEFlags) (*InternalKey, []byte) {
-	i.exhaustedBounds = 0
+	err := i.err
 	i.err = nil // clear cached iteration error
 
+	// TODO(sumeer): we are not fully optimizing in the flags.TrySeekUsingNext()
+	// case when the twoLevelIterator is already exhausted. We will take the
+	// slow-path below, even though we could return now. We could do:
+	// if flags.TrySeekUsingNext() && (i.exhaustedBounds == +1 || (i.data.isDataInvalidated() &&
+	//	i.index.isDataInvalidated())) && i.err == nil {
+	//	// Already exhausted, so return nil.
+	//	return nil, nil
+	// }
+
 	// SeekGE performs various step-instead-of-seeking optimizations: eg enabled
 	// by trySeekUsingNext, or by monotonically increasing bounds (i.boundsCmp).
 	// Care must be taken to ensure that when performing these optimizations and
@@ -1511,9 +1619,13 @@ func (i *twoLevelIterator) SeekGE(key []byte, flags base.SeekGEFlags) (*Internal
 	// previous value of maybeFilteredKeys.
 
 	var dontSeekWithinSingleLevelIter bool
-	if i.topLevelIndex.isDataInvalidated() || !i.topLevelIndex.valid() ||
+	if i.topLevelIndex.isDataInvalidated() || !i.topLevelIndex.valid() || err != nil ||
 		(i.boundsCmp <= 0 && !flags.TrySeekUsingNext()) || i.cmp(key, i.topLevelIndex.Key().UserKey) > 0 {
 		// Slow-path: need to position the topLevelIndex.
+
+		// The previous exhausted state of singleLevelIterator is no longer
+		// relevant, since we may be moving to a different index block.
+		i.exhaustedBounds = 0
 		i.maybeFilteredKeysTwoLevel = false
 		flags = flags.DisableTrySeekUsingNext()
 		var ikey *InternalKey
@@ -1550,23 +1662,57 @@ func (i *twoLevelIterator) SeekGE(key []byte, flags base.SeekGEFlags) (*Internal
 			// unless we clear it here.
 			i.boundsCmp = 0
 		}
+	} else {
+		// INVARIANT: err == nil.
+		//
+		// Else fast-path: There are two possible cases, from
+		// (i.boundsCmp > 0 || flags.TrySeekUsingNext()):
+		//
+		// 1) The bounds have moved forward (i.boundsCmp > 0) and this SeekGE is
+		// respecting the lower bound (guaranteed by Iterator). We know that the
+		// iterator must already be positioned within or just outside the previous
+		// bounds. Therefore, the topLevelIndex iter cannot be positioned at an
+		// entry ahead of the seek position (though it can be positioned behind).
+		// The !i.cmp(key, i.topLevelIndex.Key().UserKey) > 0 confirms that it is
+		// not behind. Since it is not ahead and not behind it must be at the
+		// right position.
+		//
+		// 2) This SeekGE will land on a key that is greater than the key we are
+		// currently at (guaranteed by trySeekUsingNext), but since i.cmp(key,
+		// i.topLevelIndex.Key().UserKey) <= 0, we are at the correct lower level
+		// index block. No need to reset the state of singleLevelIterator.
+		//
+		// Note that cases 1 and 2 never overlap, and one of them must be true,
+		// but we have some test code (TestIterRandomizedMaybeFilteredKeys) that
+		// sets both to true, so we fix things here and then do an invariant
+		// check.
+		//
+		// This invariant checking is important enough that we do not gate it
+		// behind invariants.Enabled.
+		if i.boundsCmp > 0 {
+			// TODO(sumeer): fix TestIterRandomizedMaybeFilteredKeys so as to not
+			// need this behavior.
+			flags = flags.DisableTrySeekUsingNext()
+		}
+		if i.boundsCmp > 0 == flags.TrySeekUsingNext() {
+			panic(fmt.Sprintf("inconsistency in optimization case 1 %t and case 2 %t",
+				i.boundsCmp > 0, flags.TrySeekUsingNext()))
+		}
+
+		if !flags.TrySeekUsingNext() {
+			// Case 1. Bounds have changed so the previous exhausted bounds state is
+			// irrelevant.
+			// WARNING-data-exhausted: this is safe to do only because the monotonic
+			// bounds optimizations only work when !data-exhausted. If they also
+			// worked with data-exhausted, we have made it unclear whether
+			// data-exhausted is actually true. See the comment at the top of the
+			// file.
+			i.exhaustedBounds = 0
+		}
+		// Else flags.TrySeekUsingNext(). The i.exhaustedBounds is important to
+		// preserve for singleLevelIterator, and twoLevelIterator.skipForward. See
+		// bug https://github.com/cockroachdb/pebble/issues/2036.
 	}
-	// Else fast-path: There are two possible cases, from
-	// (i.boundsCmp > 0 || flags.TrySeekUsingNext()):
-	//
-	// 1) The bounds have moved forward (i.boundsCmp > 0) and this SeekGE is
-	// respecting the lower bound (guaranteed by Iterator). We know that
-	// the iterator must already be positioned within or just outside the
-	// previous bounds. Therefore the topLevelIndex iter cannot be
-	// positioned at an entry ahead of the seek position (though it can be
-	// positioned behind). The !i.cmp(key, i.topLevelIndex.Key().UserKey) > 0
-	// confirms that it is not behind. Since it is not ahead and not behind
-	// it must be at the right position.
-	//
-	// 2) This SeekGE will land on a key that is greater than the key we are
-	// currently at (guaranteed by trySeekUsingNext), but since
-	// i.cmp(key, i.topLevelIndex.Key().UserKey) <= 0, we are at the correct
-	// lower level index block. No need to reset the state of singleLevelIterator.
 
 	if !dontSeekWithinSingleLevelIter {
 		// Note that while trySeekUsingNext could be false here, singleLevelIterator
@@ -1584,8 +1730,24 @@ func (i *twoLevelIterator) SeekGE(key []byte, flags base.SeekGEFlags) (*Internal
 func (i *twoLevelIterator) SeekPrefixGE(
 	prefix, key []byte, flags base.SeekGEFlags,
 ) (*base.InternalKey, []byte) {
+	// NOTE: prefix is only used for bloom filter checking and not later work in
+	// this method. Hence, we can use the existing iterator position if the last
+	// SeekPrefixGE did not fail bloom filter matching.
+
+	err := i.err
 	i.err = nil // clear cached iteration error
 
+	// TODO(sumeer): we are not fully optimizing in the flags.TrySeekUsingNext()
+	// case when the twoLevelIterator is already exhausted. We will take the
+	// slow-path below, even though we could return now. We could do:
+	// filterUsedAndDidNotMatch :=
+	//	i.reader.tableFilter != nil && i.useFilter && !i.lastBloomFilterMatched
+	// if flags.TrySeekUsingNext() && (i.exhaustedBounds == +1 || (i.data.isDataInvalidated() &&
+	//  i.index.isDataInvalidated())) && i.err == nil {
+	//	// Already exhausted, so return nil.
+	//	return nil, nil
+	//}
+
 	// Check prefix bloom filter.
 	if i.reader.tableFilter != nil && i.useFilter {
 		if !i.lastBloomFilterMatched {
@@ -1614,7 +1776,6 @@ func (i *twoLevelIterator) SeekPrefixGE(
 	}
 
 	// Bloom filter matches.
-	i.exhaustedBounds = 0
 
 	// SeekPrefixGE performs various step-instead-of-seeking optimizations: eg
 	// enabled by trySeekUsingNext, or by monotonically increasing bounds
@@ -1631,7 +1792,7 @@ func (i *twoLevelIterator) SeekPrefixGE(
 	// remembering the previous value of maybeFilteredKeysTwoLevel.
 
 	var dontSeekWithinSingleLevelIter bool
-	if i.topLevelIndex.isDataInvalidated() || !i.topLevelIndex.valid() ||
+	if i.topLevelIndex.isDataInvalidated() || !i.topLevelIndex.valid() || err != nil ||
 		i.boundsCmp <= 0 || i.cmp(key, i.topLevelIndex.Key().UserKey) > 0 {
 		// Slow-path: need to position the topLevelIndex.
 		//
@@ -1641,7 +1802,13 @@ func (i *twoLevelIterator) SeekPrefixGE(
 		// monotonic bounds). To apply it here, we would need to confirm that
 		// the topLevelIndex can continue using the same second level index
 		// block, and in that case we don't need to invalidate and reload the
-		// singleLevelIterator state.
+		// singleLevelIterator state. Do this after release blocker bug is fixed
+		// since don't want to backport this optimization.
+
+		// The previous exhausted state of singleLevelIterator is no longer
+		// relevant, since we may be moving to a different index block.
+		i.exhaustedBounds = 0
+
 		i.maybeFilteredKeysTwoLevel = false
 		flags = flags.DisableTrySeekUsingNext()
 		var ikey *InternalKey
@@ -1678,15 +1845,36 @@ func (i *twoLevelIterator) SeekPrefixGE(
 			// unless we clear it here.
 			i.boundsCmp = 0
 		}
+	} else {
+		// INVARIANT: err == nil.
+		//
+		// Else fast-path: The bounds have moved forward and this SeekGE is
+		// respecting the lower bound (guaranteed by Iterator). We know that
+		// the iterator must already be positioned within or just outside the
+		// previous bounds. Therefore the topLevelIndex iter cannot be
+		// positioned at an entry ahead of the seek position (though it can be
+		// positioned behind). The !i.cmp(key, i.topLevelIndex.Key().UserKey) > 0
+		// confirms that it is not behind. Since it is not ahead and not behind
+		// it must be at the right position.
+		//
+		// This invariant checking is important enough that we do not gate it
+		// behind invariants.Enabled.
+		if i.boundsCmp <= 0 {
+			panic(fmt.Sprintf("boundsCmp %d is invalid for optimization", i.boundsCmp))
+		}
+		if flags.TrySeekUsingNext() {
+			panic("TrySeekUsingNext must not be true")
+		}
+
+		// Bounds have changed so the previous exhausted bounds state is
+		// irrelevant.
+		// WARNING-data-exhausted: this is safe to do only because the monotonic
+		// bounds optimizations only work when !data-exhausted. If they also
+		// worked with data-exhausted, we have made it unclear whether
+		// data-exhausted is actually true. See the comment at the top of the
+		// file.
+		i.exhaustedBounds = 0
 	}
-	// Else fast-path: The bounds have moved forward and this SeekGE is
-	// respecting the lower bound (guaranteed by Iterator). We know that
-	// the iterator must already be positioned within or just outside the
-	// previous bounds. Therefore the topLevelIndex iter cannot be
-	// positioned at an entry ahead of the seek position (though it can be
-	// positioned behind). The !i.cmp(key, i.topLevelIndex.Key().UserKey) > 0
-	// confirms that it is not behind. Since it is not ahead and not behind
-	// it must be at the right position.
 
 	if !dontSeekWithinSingleLevelIter {
 		if ikey, val := i.singleLevelIterator.seekPrefixGE(
diff --git a/version_set.go b/version_set.go
index 51417fbd..1b63e62d 100644
--- a/version_set.go
+++ b/version_set.go
@@ -110,6 +110,9 @@ type versionSet struct {
 
 	writing    bool
 	writerCond sync.Cond
+	// State for deciding when to write a snapshot. Protected by mu.
+	lastSnapshotFileCount           int64
+	editsSinceLastSnapshotFileCount int64
 }
 
 func (vs *versionSet) init(
@@ -391,13 +394,76 @@ func (vs *versionSet) logAndApply(
 	currentVersion := vs.currentVersion()
 	var newVersion *version
 
-	// Generate a new manifest if we don't currently have one, or the current one
-	// is too large.
+	// Generate a new manifest if we don't currently have one, or forceRotation
+	// is true, or the current one is too large.
+	//
+	// For largeness, we do not exclusively use MaxManifestFileSize size
+	// threshold since we have had incidents where due to either large keys or
+	// large numbers of files, each edit results in a snapshot + write of the
+	// edit. This slows the system down since each flush or compaction is
+	// writing a new manifest snapshot. The primary goal of the size-based
+	// rollover logic is to ensure that when reopening a DB, the number of edits
+	// that need to be replayed on top of the snapshot is "sane". Rolling over
+	// to a new manifest after each edit is not relevant to that goal.
+	//
+	// Consider the following cases:
+	// - The number of live files F in the DB is roughly stable: after writing
+	//   the snapshot (with F files), say we require that there be enough edits
+	//   such that the cumulative number of files in those edits, E, be greater
+	//   than F. This will ensure that the total amount of time in logAndApply
+	//   that is spent in snapshot writing is ~50%.
+	//
+	// - The number of live files F in the DB is shrinking drastically, say from
+	//   F to F/10: This can happen for various reasons, like wide range
+	//   tombstones, or large numbers of smaller than usual files that are being
+	//   merged together into larger files. And say the new files generated
+	//   during this shrinkage is insignificant compared to F/10, and so for
+	//   this example we will assume it is effectively 0. After this shrinking,
+	//   E = 0.9F, and so if we used the previous snapshot file count, F, as the
+	//   threshold that needs to be exceeded, we will further delay the snapshot
+	//   writing. Which means on DB reopen we will need to replay 0.9F edits to
+	//   get to a version with 0.1F files. It would be better to create a new
+	//   snapshot when E exceeds the number of files in the current version.
+	//
+	// - The number of live files F in the DB is growing via perfect ingests
+	//   into L6: Say we wrote the snapshot when there were F files and now we
+	//   have 10F files, so E = 9F. We will further delay writing a new
+	//   snapshot. This case can be critiqued as contrived, but we consider it
+	//   nonetheless.
+	//
+	// The logic below uses the min of the last snapshot file count and the file
+	// count in the current version.
+	editCount := int64(len(ve.DeletedFiles) + len(ve.NewFiles))
+	vs.editsSinceLastSnapshotFileCount += editCount
+	sizeExceeded := vs.manifest.Size() >= vs.opts.MaxManifestFileSize
+	requireRotation := forceRotation || vs.manifest == nil
+	computeNextSnapshotFileCount := func() int64 {
+		var count int64
+		for i := range vs.metrics.Levels {
+			count += vs.metrics.Levels[i].NumFiles
+		}
+		return count
+	}
+	var nextSnapshotFileCount int64
+	if sizeExceeded && !requireRotation {
+		if vs.editsSinceLastSnapshotFileCount > vs.lastSnapshotFileCount {
+			requireRotation = true
+		} else {
+			nextSnapshotFileCount = computeNextSnapshotFileCount()
+			if vs.editsSinceLastSnapshotFileCount > nextSnapshotFileCount {
+				requireRotation = true
+			}
+		}
+	}
 	var newManifestFileNum FileNum
 	var prevManifestFileSize uint64
-	if forceRotation || vs.manifest == nil || vs.manifest.Size() >= vs.opts.MaxManifestFileSize {
+	if requireRotation {
 		newManifestFileNum = vs.getNextFileNum()
 		prevManifestFileSize = uint64(vs.manifest.Size())
+		if nextSnapshotFileCount == 0 {
+			// Haven't computed it, or happens to be 0.
+			nextSnapshotFileCount = computeNextSnapshotFileCount()
+		}
 	}
 
 	// Grab certain values before releasing vs.mu, in case createManifest() needs
@@ -474,6 +540,11 @@ func (vs *versionSet) logAndApply(
 		return err
 	}
 
+	if requireRotation {
+		// Successfully rotated.
+		vs.lastSnapshotFileCount = nextSnapshotFileCount
+		vs.editsSinceLastSnapshotFileCount = editCount
+	}
 	// Now that DB.mu is held again, initialize compacting file info in
 	// L0Sublevels.
 	inProgress := inProgressCompactions()
diff --git a/vfs/mem_fs.go b/vfs/mem_fs.go
index d573c5d9..e343b3ee 100644
--- a/vfs/mem_fs.go
+++ b/vfs/mem_fs.go
@@ -78,10 +78,23 @@ type MemFS struct {
 
 	strict      bool
 	ignoreSyncs bool
+	// Windows has peculiar semantics with respect to hard links and deleting
+	// open files. In tests meant to exercise this behavior, this flag can be
+	// set to error if removing an open file.
+	windowsSemantics bool
 }
 
 var _ FS = &MemFS{}
 
+// UseWindowsSemantics configures whether the MemFS implements Windows-style
+// semantics, in particular with respect to whether any of an open file's links
+// may be removed. Windows semantics default to off.
+func (y *MemFS) UseWindowsSemantics(windowsSemantics bool) {
+	y.mu.Lock()
+	defer y.mu.Unlock()
+	y.windowsSemantics = windowsSemantics
+}
+
 // String dumps the contents of the MemFS.
 func (y *MemFS) String() string {
 	y.mu.Lock()
@@ -303,11 +316,14 @@ func (y *MemFS) Remove(fullname string) error {
 			if !ok {
 				return oserror.ErrNotExist
 			}
-			// Disallow removal of open files/directories which implements Windows
-			// semantics. This ensures that we don't regress in the ordering of
-			// operations and try to remove a file while it is still open.
-			if n := atomic.LoadInt32(&child.refs); n > 0 {
-				return oserror.ErrInvalid
+			if y.windowsSemantics {
+				// Disallow removal of open files/directories which implements
+				// Windows semantics. This ensures that we don't regress in the
+				// ordering of operations and try to remove a file while it is
+				// still open.
+				if n := atomic.LoadInt32(&child.refs); n > 0 {
+					return oserror.ErrInvalid
+				}
 			}
 			if len(child.children) > 0 {
 				return errNotEmpty


Test patch:
diff --git a/data_test.go b/data_test.go
index 9c22c61f..c87590dd 100644
--- a/data_test.go
+++ b/data_test.go
@@ -21,7 +21,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/rangedel"
 	"github.com/cockroachdb/pebble/internal/rangekey"
 	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/cockroachdb/pebble/internal/testkeys/blockprop"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
@@ -141,6 +140,8 @@ func runIterCmd(d *datadriven.TestData, iter *Iterator, closeIter bool) string {
 					op = "seekge"
 				case seekLTLastPositioningOp:
 					op = "seeklt"
+				case invalidatedLastPositionOp:
+					op = "invalidate"
 				}
 				fmt.Fprintf(&b, "%s=%q\n", field, op)
 			default:
@@ -294,7 +295,7 @@ func parseIterOptions(
 			opts.RangeKeyMasking.Suffix = []byte(arg[1])
 		case "mask-filter":
 			opts.RangeKeyMasking.Filter = func() BlockPropertyFilterMask {
-				return blockprop.NewMaskingFilter()
+				return sstable.NewTestKeysMaskingFilter()
 			}
 		case "table-filter":
 			switch arg[1] {
diff --git a/db_test.go b/db_test.go
index 7a6ae49a..c091e490 100644
--- a/db_test.go
+++ b/db_test.go
@@ -917,6 +917,7 @@ func TestRollManifest(t *testing.T) {
 	opts := &Options{
 		MaxManifestFileSize:   1,
 		L0CompactionThreshold: 10,
+		L0StopWritesThreshold: 1000,
 		FS:                    vfs.NewMem(),
 		NumPrevManifest:       int(toPreserve),
 	}
@@ -930,6 +931,11 @@ func TestRollManifest(t *testing.T) {
 		defer d.mu.Unlock()
 		return d.mu.versions.manifestFileNum
 	}
+	sizeRolloverState := func() (int64, int64) {
+		d.mu.Lock()
+		defer d.mu.Unlock()
+		return d.mu.versions.lastSnapshotFileCount, d.mu.versions.editsSinceLastSnapshotFileCount
+	}
 
 	current := func() string {
 		desc, err := Peek(d.dirname, d.opts.FS)
@@ -940,11 +946,46 @@ func TestRollManifest(t *testing.T) {
 	lastManifestNum := manifestFileNumber()
 	manifestNums := []base.FileNum{lastManifestNum}
 	for i := 0; i < 5; i++ {
-		require.NoError(t, d.Set([]byte("a"), nil, nil))
-		require.NoError(t, d.Flush())
+		// MaxManifestFileSize is 1, but the rollover logic also counts edits
+		// since the last snapshot to decide on rollover, so do as many flushes as
+		// it demands.
+		lastSnapshotCount, editsSinceSnapshotCount := sizeRolloverState()
+		var expectedLastSnapshotCount, expectedEditsSinceSnapshotCount int64
+		switch i {
+		case 0:
+			// DB is empty.
+			expectedLastSnapshotCount, expectedEditsSinceSnapshotCount = 0, 0
+		case 1:
+			// First edit that caused rollover is not in the snapshot.
+			expectedLastSnapshotCount, expectedEditsSinceSnapshotCount = 0, 1
+		case 2:
+			// One flush is in the snapshot. One flush in the edit.
+			expectedLastSnapshotCount, expectedEditsSinceSnapshotCount = 1, 1
+		case 3:
+			// Two flushes in the snapshot. One flush in the edit. Will need to do
+			// two more flushes, the first of which will be in the next snapshot.
+			expectedLastSnapshotCount, expectedEditsSinceSnapshotCount = 2, 1
+		case 4:
+			// Four flushes in the snapshot. One flush in the edit. Will need to do
+			// four more flushes, three of which will be in the snapshot.
+			expectedLastSnapshotCount, expectedEditsSinceSnapshotCount = 4, 1
+		}
+		require.Equal(t, expectedLastSnapshotCount, lastSnapshotCount)
+		require.Equal(t, expectedEditsSinceSnapshotCount, editsSinceSnapshotCount)
+		// Number of flushes to do to trigger the rollover.
+		steps := int(lastSnapshotCount - editsSinceSnapshotCount + 1)
+		// Steps can be <= 0, but we need to do at least one edit to trigger the
+		// rollover logic.
+		if steps <= 0 {
+			steps = 1
+		}
+		for j := 0; j < steps; j++ {
+			require.NoError(t, d.Set([]byte("a"), nil, nil))
+			require.NoError(t, d.Flush())
+		}
 		num := manifestFileNumber()
 		if lastManifestNum == num {
-			t.Fatalf("manifest failed to roll: %d == %d", lastManifestNum, num)
+			t.Fatalf("manifest failed to roll %d: %d == %d", i, lastManifestNum, num)
 		}
 
 		manifestNums = append(manifestNums, num)
@@ -955,6 +996,9 @@ func TestRollManifest(t *testing.T) {
 			t.Fatalf("expected %s, but found %s", expectedCurrent, v)
 		}
 	}
+	lastSnapshotCount, editsSinceSnapshotCount := sizeRolloverState()
+	require.EqualValues(t, 8, lastSnapshotCount)
+	require.EqualValues(t, 1, editsSinceSnapshotCount)
 
 	files, err := d.opts.FS.List("")
 	require.NoError(t, err)
@@ -983,6 +1027,37 @@ func TestRollManifest(t *testing.T) {
 	}
 	require.EqualValues(t, expected, manifests)
 
+	// Test the logic that uses the future snapshot size to rollover.
+	// Reminder: we have a snapshot with 8 files and the manifest has 1 edit
+	// (flush) with 1 file.
+	// Add 8 more files with a different key.
+	lastManifestNum = manifestFileNumber()
+	for j := 0; j < 8; j++ {
+		require.NoError(t, d.Set([]byte("c"), nil, nil))
+		require.NoError(t, d.Flush())
+	}
+	lastSnapshotCount, editsSinceSnapshotCount = sizeRolloverState()
+	// Need 16 more files in edits to trigger a rollover.
+	require.EqualValues(t, 16, lastSnapshotCount)
+	require.EqualValues(t, 1, editsSinceSnapshotCount)
+	require.NotEqual(t, manifestFileNumber(), lastManifestNum)
+	lastManifestNum = manifestFileNumber()
+	// Do a compaction that moves 8 of the files from L0 to 1 file in L6. This
+	// adds 9 files in edits. We still need 6 more files in edits based on the
+	// last snapshot. But the current version has only 9 L0 files and 1 L6 file,
+	// for a total of 10 files. So 1 flush should push us over that threshold.
+	d.Compact([]byte("c"), []byte("d"), false)
+	lastSnapshotCount, editsSinceSnapshotCount = sizeRolloverState()
+	require.EqualValues(t, 16, lastSnapshotCount)
+	require.EqualValues(t, 10, editsSinceSnapshotCount)
+	require.Equal(t, manifestFileNumber(), lastManifestNum)
+	require.NoError(t, d.Set([]byte("c"), nil, nil))
+	require.NoError(t, d.Flush())
+	lastSnapshotCount, editsSinceSnapshotCount = sizeRolloverState()
+	require.EqualValues(t, 10, lastSnapshotCount)
+	require.EqualValues(t, 1, editsSinceSnapshotCount)
+	require.NotEqual(t, manifestFileNumber(), lastManifestNum)
+
 	require.NoError(t, d.Close())
 }
 
diff --git a/external_iterator_test.go b/external_iterator_test.go
index 9fa1206c..3c2d47c4 100644
--- a/external_iterator_test.go
+++ b/external_iterator_test.go
@@ -16,7 +16,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/cockroachdb/pebble/internal/testkeys/blockprop"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
@@ -193,7 +192,7 @@ func TestIterRandomizedMaybeFilteredKeys(t *testing.T) {
 				TableFormat:    sstable.TableFormatPebblev2,
 				BlockPropertyCollectors: []func() BlockPropertyCollector{
 					func() BlockPropertyCollector {
-						return blockprop.NewBlockPropertyCollector()
+						return sstable.NewTestKeysBlockPropertyCollector()
 					},
 				},
 			})
@@ -235,7 +234,7 @@ func TestIterRandomizedMaybeFilteredKeys(t *testing.T) {
 			require.NoError(t, err)
 			defer r.Close()
 
-			filter := blockprop.NewBlockPropertyFilter(uint64(tsSeparator), math.MaxUint64)
+			filter := sstable.NewTestKeysBlockPropertyFilter(uint64(tsSeparator), math.MaxUint64)
 			filterer := sstable.NewBlockPropertiesFilterer([]BlockPropertyFilter{filter}, nil)
 			ok, err := filterer.IntersectsUserPropsAndFinishInit(r.Properties.UserProperties)
 			require.True(t, ok)
diff --git a/format_major_version_test.go b/format_major_version_test.go
index f44aad41..634ac591 100644
--- a/format_major_version_test.go
+++ b/format_major_version_test.go
@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/internal/testkeys"
@@ -520,3 +521,41 @@ func TestPebblev1MigrationRace(t *testing.T) {
 	require.NoError(t, d.RatchetFormatMajorVersion(FormatPrePebblev1Marked))
 	wg.Wait()
 }
+
+// Regression test for #2044, where multiple concurrent compactions can lead
+// to an indefinite wait on the compaction goroutine in compactMarkedFilesLocked.
+func TestPebblev1MigrationConcurrencyRace(t *testing.T) {
+	opts := &Options{
+		Comparer:           testkeys.Comparer,
+		FS:                 vfs.NewMem(),
+		FormatMajorVersion: FormatSplitUserKeysMarked,
+		Levels:             []LevelOptions{{FilterPolicy: bloom.FilterPolicy(10)}},
+		MaxConcurrentCompactions: func() int {
+			return 4
+		},
+	}
+	func() {
+		d, err := Open("", opts)
+		require.NoError(t, err)
+		defer func() {
+			require.NoError(t, d.Close())
+		}()
+
+		ks := testkeys.Alpha(3).EveryN(10)
+		var key [3]byte
+		for i := 0; i < ks.Count(); i++ {
+			n := testkeys.WriteKey(key[:], ks, i)
+			require.NoError(t, d.Set(key[:n], key[:n], nil))
+			if i%100 == 0 {
+				require.NoError(t, d.Flush())
+			}
+		}
+		require.NoError(t, d.Flush())
+	}()
+
+	opts.FormatMajorVersion = FormatPrePebblev1MarkedCompacted
+	d, err := Open("", opts)
+	require.NoError(t, err)
+	require.NoError(t, d.RatchetFormatMajorVersion(FormatPrePebblev1MarkedCompacted))
+	require.NoError(t, d.Close())
+}
diff --git a/ingest_test.go b/ingest_test.go
index 5930869d..78f1d475 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -1432,6 +1432,7 @@ func TestIngestCleanup(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run("", func(t *testing.T) {
 			mem := vfs.NewMem()
+			mem.UseWindowsSemantics(true)
 
 			// Create the files in the VFS.
 			metaMap := make(map[base.FileNum]vfs.File)
diff --git a/internal/metamorphic/crossversion/crossversion_test.go b/internal/metamorphic/crossversion/crossversion_test.go
index 03f3f6c1..c171e05e 100644
--- a/internal/metamorphic/crossversion/crossversion_test.go
+++ b/internal/metamorphic/crossversion/crossversion_test.go
@@ -255,6 +255,11 @@ func (r *metamorphicTestRun) run(ctx context.Context, output io.Writer) error {
 		"-test.run", "TestMeta$",
 		"-seed", strconv.FormatUint(r.seed, 10),
 		"-keep",
+		// Use an op-count distribution that includes a low lower bound, so that
+		// some intermediary versions do very little work besides opening the
+		// database. This helps exercise state from version n that survives to
+		// versions ≥ n+2.
+		"-ops", "uniform:1-10000",
 	}
 	// Propagate the verbose flag, if necessary.
 	if testing.Verbose() {
diff --git a/internal/testkeys/testkeys.go b/internal/testkeys/testkeys.go
index 0ea38a44..8ac89e2e 100644
--- a/internal/testkeys/testkeys.go
+++ b/internal/testkeys/testkeys.go
@@ -58,8 +58,8 @@ var Comparer *base.Comparer = &base.Comparer{
 		if bytes.Equal(a[:ai], buf) {
 			return append(dst[:n], a...)
 		}
-		// The separator is > a[:ai], so we only need to add the sentinel.
-		return append(dst, 0)
+		// The separator is > a[:ai], so return it
+		return dst
 	},
 	Successor: func(dst, a []byte) []byte {
 		ai := split(a)
@@ -73,8 +73,8 @@ var Comparer *base.Comparer = &base.Comparer{
 		if bytes.Equal(a[:ai], buf) {
 			return append(dst[:n], a...)
 		}
-		// The successor is > a[:ai], so we only need to add the sentinel.
-		return append(dst, 0)
+		// The successor is > a[:ai], so return it.
+		return dst
 	},
 	ImmediateSuccessor: func(dst, a []byte) []byte {
 		// TODO(jackson): Consider changing this Comparer to only support
diff --git a/iterator_histories_test.go b/iterator_histories_test.go
index 5980333e..7d359712 100644
--- a/iterator_histories_test.go
+++ b/iterator_histories_test.go
@@ -15,7 +15,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/cockroachdb/pebble/internal/testkeys/blockprop"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
@@ -62,7 +61,7 @@ func TestIterHistories(t *testing.T) {
 					Comparer:           testkeys.Comparer,
 					FormatMajorVersion: FormatRangeKeys,
 					BlockPropertyCollectors: []func() BlockPropertyCollector{
-						blockprop.NewBlockPropertyCollector,
+						sstable.NewTestKeysBlockPropertyCollector,
 					},
 				}
 				opts.DisableAutomaticCompactions = true
@@ -267,7 +266,7 @@ func TestIterHistories(t *testing.T) {
 						o.RangeKeyMasking.Suffix = []byte(arg.Vals[0])
 					case "mask-filter":
 						o.RangeKeyMasking.Filter = func() BlockPropertyFilterMask {
-							return blockprop.NewMaskingFilter()
+							return sstable.NewTestKeysMaskingFilter()
 						}
 					case "lower":
 						o.LowerBound = []byte(arg.Vals[0])
@@ -293,7 +292,7 @@ func TestIterHistories(t *testing.T) {
 							return err.Error()
 						}
 						o.PointKeyFilters = []sstable.BlockPropertyFilter{
-							blockprop.NewBlockPropertyFilter(min, max),
+							sstable.NewTestKeysBlockPropertyFilter(min, max),
 						}
 					}
 				}
diff --git a/iterator_test.go b/iterator_test.go
index c62df1f9..ce245298 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -23,7 +23,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/cockroachdb/pebble/internal/testkeys/blockprop"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
@@ -2325,7 +2324,7 @@ func TestRangeKeyMaskingRandomized(t *testing.T) {
 	}
 	if rng.Intn(2) == 0 {
 		randomOpts.filter = func() BlockPropertyFilterMask {
-			return blockprop.NewMaskingFilter()
+			return sstable.NewTestKeysMaskingFilter()
 		}
 	}
 
@@ -2337,7 +2336,7 @@ func TestRangeKeyMaskingRandomized(t *testing.T) {
 		FormatMajorVersion:       FormatNewest,
 		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
-			blockprop.NewBlockPropertyCollector,
+			sstable.NewTestKeysBlockPropertyCollector,
 		},
 	}
 	opts1.Levels = baseOpts.levelOpts
@@ -2350,7 +2349,7 @@ func TestRangeKeyMaskingRandomized(t *testing.T) {
 		FormatMajorVersion:       FormatNewest,
 		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
-			blockprop.NewBlockPropertyCollector,
+			sstable.NewTestKeysBlockPropertyCollector,
 		},
 	}
 	opts2.Levels = randomOpts.levelOpts
@@ -2477,7 +2476,7 @@ func BenchmarkIterator_RangeKeyMasking(b *testing.B) {
 		FormatMajorVersion:       FormatNewest,
 		MaxConcurrentCompactions: func() int { return maxProcs/2 + 1 },
 		BlockPropertyCollectors: []func() BlockPropertyCollector{
-			blockprop.NewBlockPropertyCollector,
+			sstable.NewTestKeysBlockPropertyCollector,
 		},
 	}
 	d, err := Open("", opts)
@@ -2527,7 +2526,7 @@ func BenchmarkIterator_RangeKeyMasking(b *testing.B) {
 				RangeKeyMasking: RangeKeyMasking{
 					Suffix: []byte("@100"),
 					Filter: func() BlockPropertyFilterMask {
-						return blockprop.NewMaskingFilter()
+						return sstable.NewTestKeysMaskingFilter()
 					},
 				},
 			}
diff --git a/sstable/block_property_test.go b/sstable/block_property_test.go
index dc7e592c..d0fe6f49 100644
--- a/sstable/block_property_test.go
+++ b/sstable/block_property_test.go
@@ -801,7 +801,7 @@ func (c *valueCharBlockIntervalCollector) FinishDataBlock() (lower, upper uint64
 	return l, u, nil
 }
 
-// suffixIntervalCollector maintains an interval over the timestamps in
+// testKeysSuffixIntervalCollector maintains an interval over the timestamps in
 // MVCC-like suffixes for keys (e.g. foo@123).
 type suffixIntervalCollector struct {
 	initialized  bool
diff --git a/internal/testkeys/blockprop/blockprop.go b/sstable/block_property_test_utils.go
similarity index 58%
rename from internal/testkeys/blockprop/blockprop.go
rename to sstable/block_property_test_utils.go
index 6f2da2c7..0ade68ff 100644
--- a/internal/testkeys/blockprop/blockprop.go
+++ b/sstable/block_property_test_utils.go
@@ -2,31 +2,31 @@
 // of this source code is governed by a BSD-style license that can be found in
 // the LICENSE file.
 
-// Package blockprop implements interval block property collectors and filters
-// on the suffixes of keys in the format used by the testkeys package (eg,
-// 'key@5').
-package blockprop
+package sstable
 
 import (
 	"math"
 
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/cockroachdb/pebble/sstable"
 )
 
-const blockPropertyName = `pebble.internal.testkeys.suffixes`
+// Code in this file contains utils for testing. It implements interval block
+// property collectors and filters on the suffixes of keys in the format used
+// by the testkeys package (eg, 'key@5').
 
-// NewBlockPropertyCollector constructs a sstable property collector over
-// testkey suffixes.
-func NewBlockPropertyCollector() sstable.BlockPropertyCollector {
-	return sstable.NewBlockIntervalCollector(
-		blockPropertyName,
-		&suffixIntervalCollector{},
+const testKeysBlockPropertyName = `pebble.internal.testkeys.suffixes`
+
+// NewTestKeysBlockPropertyCollector constructs a sstable property collector
+// over testkey suffixes.
+func NewTestKeysBlockPropertyCollector() BlockPropertyCollector {
+	return NewBlockIntervalCollector(
+		testKeysBlockPropertyName,
+		&testKeysSuffixIntervalCollector{},
 		nil)
 }
 
-// NewBlockPropertyFilter constructs a new block-property filter that excludes
+// NewTestKeysBlockPropertyFilter constructs a new block-property filter that excludes
 // blocks containing exclusively suffixed keys where all the suffixes fall
 // outside of the range [filterMin, filterMax).
 //
@@ -34,27 +34,27 @@ func NewBlockPropertyCollector() sstable.BlockPropertyCollector {
 // results of this block property filter are deterministic for unsuffixed keys
 // and keys with suffixes within the range [filterMin, filterMax). For keys with
 // suffixes outside the range, iteration is nondeterministic.
-func NewBlockPropertyFilter(filterMin, filterMax uint64) *sstable.BlockIntervalFilter {
-	return sstable.NewBlockIntervalFilter(blockPropertyName, filterMin, filterMax)
+func NewTestKeysBlockPropertyFilter(filterMin, filterMax uint64) *BlockIntervalFilter {
+	return NewBlockIntervalFilter(testKeysBlockPropertyName, filterMin, filterMax)
 }
 
-// NewMaskingFilter constructs a MaskingFilter that implements
+// NewTestKeysMaskingFilter constructs a TestKeysMaskingFilter that implements
 // pebble.BlockPropertyFilterMask for efficient range-key masking using the
 // testkeys block property filter. The masking filter wraps a block interval
 // filter, and modifies the configured interval when Pebble requests it.
-func NewMaskingFilter() MaskingFilter {
-	return MaskingFilter{BlockIntervalFilter: NewBlockPropertyFilter(0, math.MaxUint64)}
+func NewTestKeysMaskingFilter() TestKeysMaskingFilter {
+	return TestKeysMaskingFilter{BlockIntervalFilter: NewTestKeysBlockPropertyFilter(0, math.MaxUint64)}
 }
 
-// MaskingFilter implements BlockPropertyFilterMask and may be used to mask
+// TestKeysMaskingFilter implements BlockPropertyFilterMask and may be used to mask
 // point keys with the testkeys-style suffixes (eg, @4) that are masked by range
 // keys with testkeys-style suffixes.
-type MaskingFilter struct {
-	*sstable.BlockIntervalFilter
+type TestKeysMaskingFilter struct {
+	*BlockIntervalFilter
 }
 
 // SetSuffix implements pebble.BlockPropertyFilterMask.
-func (f MaskingFilter) SetSuffix(suffix []byte) error {
+func (f TestKeysMaskingFilter) SetSuffix(suffix []byte) error {
 	ts, err := testkeys.ParseSuffix(suffix)
 	if err != nil {
 		return err
@@ -64,15 +64,15 @@ func (f MaskingFilter) SetSuffix(suffix []byte) error {
 }
 
 // Intersects implements the BlockPropertyFilter interface.
-func (f MaskingFilter) Intersects(prop []byte) (bool, error) {
+func (f TestKeysMaskingFilter) Intersects(prop []byte) (bool, error) {
 	return f.BlockIntervalFilter.Intersects(prop)
 }
 
-var _ sstable.DataBlockIntervalCollector = (*suffixIntervalCollector)(nil)
+var _ DataBlockIntervalCollector = (*testKeysSuffixIntervalCollector)(nil)
 
-// suffixIntervalCollector maintains an interval over the timestamps in
+// testKeysSuffixIntervalCollector maintains an interval over the timestamps in
 // MVCC-like suffixes for keys (e.g. foo@123).
-type suffixIntervalCollector struct {
+type testKeysSuffixIntervalCollector struct {
 	initialized  bool
 	lower, upper uint64
 }
@@ -82,7 +82,7 @@ type suffixIntervalCollector struct {
 //
 // Note that range sets and unsets may have multiple suffixes. Range key deletes
 // do not have a suffix. All other point keys have a single suffix.
-func (c *suffixIntervalCollector) Add(key base.InternalKey, value []byte) error {
+func (c *testKeysSuffixIntervalCollector) Add(key base.InternalKey, value []byte) error {
 	i := testkeys.Comparer.Split(key.UserKey)
 	if i == len(key.UserKey) {
 		c.initialized = true
@@ -109,7 +109,7 @@ func (c *suffixIntervalCollector) Add(key base.InternalKey, value []byte) error
 }
 
 // FinishDataBlock implements DataBlockIntervalCollector.
-func (c *suffixIntervalCollector) FinishDataBlock() (lower, upper uint64, err error) {
+func (c *testKeysSuffixIntervalCollector) FinishDataBlock() (lower, upper uint64, err error) {
 	l, u := c.lower, c.upper
 	c.lower, c.upper = 0, 0
 	c.initialized = false
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index 48acd37d..8246b4d6 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/internal/errorfs"
+	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
@@ -205,7 +206,7 @@ func TestReader(t *testing.T) {
 							oName, lName, dName, iName),
 						func(t *testing.T) {
 							runTestReader(
-								t, tableOpt, testDirs[oName], nil /* Reader */, 0)
+								t, tableOpt, testDirs[oName], nil /* Reader */, 0, false)
 						})
 				}
 			}
@@ -233,7 +234,9 @@ func TestHamletReader(t *testing.T) {
 
 		t.Run(
 			fmt.Sprintf("sst=%s", prebuiltSST),
-			func(t *testing.T) { runTestReader(t, WriterOptions{}, "testdata/hamletreader", r, 0) },
+			func(t *testing.T) {
+				runTestReader(t, WriterOptions{}, "testdata/hamletreader", r, 0, false)
+			},
 		)
 	}
 }
@@ -243,7 +246,19 @@ func TestReaderStats(t *testing.T) {
 		BlockSize:      30,
 		IndexBlockSize: 30,
 	}
-	runTestReader(t, tableOpt, "testdata/readerstats", nil, 10000)
+	runTestReader(t, tableOpt, "testdata/readerstats", nil, 10000, false)
+}
+
+func TestReaderWithBlockPropertyFilter(t *testing.T) {
+	writerOpt := WriterOptions{
+		BlockSize: 1,
+		IndexBlockSize: 40,
+		Comparer: testkeys.Comparer,
+		TableFormat: TableFormatMax,
+		BlockPropertyCollectors: []func() BlockPropertyCollector{NewTestKeysBlockPropertyCollector},
+	}
+	runTestReader(
+		t, writerOpt, "testdata/reader_bpf", nil /* Reader */,0, true)
 }
 
 func TestInjectedErrors(t *testing.T) {
@@ -319,7 +334,44 @@ func TestInvalidReader(t *testing.T) {
 	}
 }
 
-func runTestReader(t *testing.T, o WriterOptions, dir string, r *Reader, cacheSize int) {
+func indexLayoutString(t *testing.T, r *Reader) string {
+	indexH, err := r.readIndex(nil /* stats */)
+	require.NoError(t, err)
+	defer indexH.Release()
+	var buf strings.Builder
+	twoLevelIndex := r.Properties.IndexType == twoLevelIndex
+	buf.WriteString("index entries:\n")
+	iter, err := newBlockIter(r.Compare, indexH.Get())
+	defer func() {
+		require.NoError(t, iter.Close())
+	}()
+	require.NoError(t, err)
+	for key, value := iter.First(); key != nil; key, value = iter.Next() {
+		bh, err := decodeBlockHandleWithProperties(value)
+		require.NoError(t, err)
+		fmt.Fprintf(&buf, " %s: size %d\n", string(key.UserKey), bh.Length)
+		if twoLevelIndex {
+			b, err := r.readBlock(bh.BlockHandle, nil, nil, nil)
+			require.NoError(t, err)
+			defer b.Release()
+			iter2, err := newBlockIter(r.Compare, b.Get())
+			defer func() {
+				require.NoError(t, iter2.Close())
+			}()
+			require.NoError(t, err)
+			for key, value := iter2.First(); key != nil; key, value = iter2.Next() {
+				bh, err := decodeBlockHandleWithProperties(value)
+				require.NoError(t, err)
+				fmt.Fprintf(&buf, "   %s: size %d\n", string(key.UserKey), bh.Length)
+			}
+		}
+	}
+	return buf.String()
+}
+
+func runTestReader(
+	t *testing.T, o WriterOptions, dir string, r *Reader, cacheSize int, printLayout bool,
+) {
 	datadriven.Walk(t, dir, func(t *testing.T, path string) {
 		defer func() {
 			if r != nil {
@@ -340,6 +392,9 @@ func runTestReader(t *testing.T, o WriterOptions, dir string, r *Reader, cacheSi
 				if err != nil {
 					return err.Error()
 				}
+				if printLayout {
+					return indexLayoutString(t, r)
+				}
 				return ""
 
 			case "iter":
@@ -349,10 +404,25 @@ func runTestReader(t *testing.T, o WriterOptions, dir string, r *Reader, cacheSi
 				}
 				var stats base.InternalIteratorStats
 				r.Properties.GlobalSeqNum = seqNum
+				var filterer *BlockPropertiesFilterer
+				if d.HasArg("block-property-filter") {
+					var filterMin, filterMax uint64
+					d.ScanArgs(t, "block-property-filter", &filterMin, &filterMax)
+					bpf := NewTestKeysBlockPropertyFilter(filterMin, filterMax)
+					filterer = NewBlockPropertiesFilterer([]BlockPropertyFilter{bpf}, nil)
+					intersects, err :=
+						filterer.IntersectsUserPropsAndFinishInit(r.Properties.UserProperties)
+					if err != nil {
+						return err.Error()
+					}
+					if !intersects {
+						return "table does not intersect BlockPropertyFilter"
+					}
+				}
 				iter, err := r.NewIterWithBlockPropertyFilters(
 					nil,  /* lower */
 					nil,  /* upper */
-					nil,  /* filterer */
+					filterer,
 					true, /* use filter block */
 					&stats,
 				)
diff --git a/sstable/testdata/reader_bpf/iter b/sstable/testdata/reader_bpf/iter
new file mode 100644
index 00000000..e527895d
--- /dev/null
+++ b/sstable/testdata/reader_bpf/iter
@@ -0,0 +1,56 @@
+# Test case for bug https://github.com/cockroachdb/pebble/issues/2036 Build
+# sstable with two-level index, with two data blocks in each lower-level index
+# block.
+build
+c@10.SET.10:cAT10
+d@7.SET.9:dAT7
+e@15.SET.8:eAT15
+f@7.SET.5:fAT7
+----
+index entries:
+ d@7: size 53
+   c@10: size 28
+   d@7: size 26
+ g: size 51
+   e@15: size 28
+   g: size 26
+
+iter
+first
+next
+next
+next
+----
+<c@10:10>
+<d@7:9>
+<e@15:8>
+<f@7:5>
+
+
+# The block property filter matches data block 2 and 4.
+iter block-property-filter=(7,8)
+first
+next
+----
+<d@7:9>
+<f@7:5>
+
+# Use the same block property filter, but use seeks to find these entries.
+# With the bug the second seek-ge below would step to the second lower-level
+# index block and only see the entry in the data block 4.
+iter block-property-filter=(7,8)
+set-bounds lower=a upper=c
+seek-ge a
+seek-ge b true
+set-bounds lower=c upper=g
+seek-ge c
+next
+next
+----
+.
+.
+.
+.
+<d@7:9>
+<f@7:5>
+.
diff --git a/testdata/event_listener b/testdata/event_listener
index c079b1c0..3d2ce3f7 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -62,20 +62,14 @@ create: db/marker.format-version.000010.011
 close: db/marker.format-version.000010.011
 sync: db
 upgraded to format version: 011
-create: db/MANIFEST-000003
-close: db/MANIFEST-000001
-sync: db/MANIFEST-000003
-create: db/marker.manifest.000002.MANIFEST-000003
-close: db/marker.manifest.000002.MANIFEST-000003
-sync: db
-[JOB 1] MANIFEST created 000003
+sync: db/MANIFEST-000001
 create: wal/000002.log
 sync: wal
 [JOB 1] WAL created 000002
-create: db/temporary.000004.dbtmp
-sync: db/temporary.000004.dbtmp
-close: db/temporary.000004.dbtmp
-rename: db/temporary.000004.dbtmp -> db/OPTIONS-000004
+create: db/temporary.000003.dbtmp
+sync: db/temporary.000003.dbtmp
+close: db/temporary.000003.dbtmp
+rename: db/temporary.000003.dbtmp -> db/OPTIONS-000003
 sync: db
 
 flush
@@ -83,110 +77,109 @@ flush
 sync: wal/000002.log
 sync: wal/000002.log
 close: wal/000002.log
-create: wal/000005.log
+create: wal/000004.log
 sync: wal
-[JOB 6] WAL created 000005
+[JOB 6] WAL created 000004
 [JOB 7] flushing 1 memtable to L0
-create: db/000006.sst
-[JOB 7] flushing: sstable created 000006
-sync: db/000006.sst
-close: db/000006.sst
+create: db/000005.sst
+[JOB 7] flushing: sstable created 000005
+sync: db/000005.sst
+close: db/000005.sst
 sync: db
-create: db/MANIFEST-000007
-close: db/MANIFEST-000003
-sync: db/MANIFEST-000007
-create: db/marker.manifest.000003.MANIFEST-000007
-close: db/marker.manifest.000003.MANIFEST-000007
+create: db/MANIFEST-000006
+close: db/MANIFEST-000001
+sync: db/MANIFEST-000006
+create: db/marker.manifest.000002.MANIFEST-000006
+close: db/marker.manifest.000002.MANIFEST-000006
 sync: db
-[JOB 7] MANIFEST created 000007
-[JOB 7] flushed 1 memtable to L0 [000006] (770 B), in 1.0s (2.0s total), output rate 770 B/s
-[JOB 7] MANIFEST deleted 000001
+[JOB 7] MANIFEST created 000006
+[JOB 7] flushed 1 memtable to L0 [000005] (770 B), in 1.0s (2.0s total), output rate 770 B/s
 
 compact
 ----
-sync: wal/000005.log
-sync: wal/000005.log
-close: wal/000005.log
-reuseForWrite: wal/000002.log -> wal/000008.log
+sync: wal/000004.log
+sync: wal/000004.log
+close: wal/000004.log
+reuseForWrite: wal/000002.log -> wal/000007.log
 sync: wal
-[JOB 8] WAL created 000008 (recycled 000002)
+[JOB 8] WAL created 000007 (recycled 000002)
 [JOB 9] flushing 1 memtable to L0
-create: db/000009.sst
-[JOB 9] flushing: sstable created 000009
-sync: db/000009.sst
-close: db/000009.sst
+create: db/000008.sst
+[JOB 9] flushing: sstable created 000008
+sync: db/000008.sst
+close: db/000008.sst
 sync: db
-create: db/MANIFEST-000010
-close: db/MANIFEST-000007
-sync: db/MANIFEST-000010
-create: db/marker.manifest.000004.MANIFEST-000010
-close: db/marker.manifest.000004.MANIFEST-000010
+create: db/MANIFEST-000009
+close: db/MANIFEST-000006
+sync: db/MANIFEST-000009
+create: db/marker.manifest.000003.MANIFEST-000009
+close: db/marker.manifest.000003.MANIFEST-000009
 sync: db
-[JOB 9] MANIFEST created 000010
-[JOB 9] flushed 1 memtable to L0 [000009] (770 B), in 1.0s (2.0s total), output rate 770 B/s
-[JOB 9] MANIFEST deleted 000003
-[JOB 10] compacting(default) L0 [000006 000009] (1.5 K) + L6 [] (0 B)
-create: db/000011.sst
-[JOB 10] compacting: sstable created 000011
-sync: db/000011.sst
-close: db/000011.sst
+[JOB 9] MANIFEST created 000009
+[JOB 9] flushed 1 memtable to L0 [000008] (770 B), in 1.0s (2.0s total), output rate 770 B/s
+[JOB 9] MANIFEST deleted 000001
+[JOB 10] compacting(default) L0 [000005 000008] (1.5 K) + L6 [] (0 B)
+create: db/000010.sst
+[JOB 10] compacting: sstable created 000010
+sync: db/000010.sst
+close: db/000010.sst
 sync: db
-create: db/MANIFEST-000012
-close: db/MANIFEST-000010
-sync: db/MANIFEST-000012
-create: db/marker.manifest.000005.MANIFEST-000012
-close: db/marker.manifest.000005.MANIFEST-000012
+create: db/MANIFEST-000011
+close: db/MANIFEST-000009
+sync: db/MANIFEST-000011
+create: db/marker.manifest.000004.MANIFEST-000011
+close: db/marker.manifest.000004.MANIFEST-000011
 sync: db
-[JOB 10] MANIFEST created 000012
-[JOB 10] compacted(default) L0 [000006 000009] (1.5 K) + L6 [] (0 B) -> L6 [000011] (770 B), in 1.0s (2.0s total), output rate 770 B/s
-[JOB 10] sstable deleted 000006
-[JOB 10] sstable deleted 000009
-[JOB 10] MANIFEST deleted 000007
+[JOB 10] MANIFEST created 000011
+[JOB 10] compacted(default) L0 [000005 000008] (1.5 K) + L6 [] (0 B) -> L6 [000010] (770 B), in 1.0s (2.0s total), output rate 770 B/s
+[JOB 10] sstable deleted 000005
+[JOB 10] sstable deleted 000008
+[JOB 10] MANIFEST deleted 000006
 
 disable-file-deletions
 ----
 
 flush
 ----
-sync: wal/000008.log
-sync: wal/000008.log
-close: wal/000008.log
-reuseForWrite: wal/000005.log -> wal/000013.log
+sync: wal/000007.log
+sync: wal/000007.log
+close: wal/000007.log
+reuseForWrite: wal/000004.log -> wal/000012.log
 sync: wal
-[JOB 11] WAL created 000013 (recycled 000005)
+[JOB 11] WAL created 000012 (recycled 000004)
 [JOB 12] flushing 1 memtable to L0
-create: db/000014.sst
-[JOB 12] flushing: sstable created 000014
-sync: db/000014.sst
-close: db/000014.sst
+create: db/000013.sst
+[JOB 12] flushing: sstable created 000013
+sync: db/000013.sst
+close: db/000013.sst
 sync: db
-create: db/MANIFEST-000015
-close: db/MANIFEST-000012
-sync: db/MANIFEST-000015
-create: db/marker.manifest.000006.MANIFEST-000015
-close: db/marker.manifest.000006.MANIFEST-000015
+create: db/MANIFEST-000014
+close: db/MANIFEST-000011
+sync: db/MANIFEST-000014
+create: db/marker.manifest.000005.MANIFEST-000014
+close: db/marker.manifest.000005.MANIFEST-000014
 sync: db
-[JOB 12] MANIFEST created 000015
-[JOB 12] flushed 1 memtable to L0 [000014] (770 B), in 1.0s (2.0s total), output rate 770 B/s
+[JOB 12] MANIFEST created 000014
+[JOB 12] flushed 1 memtable to L0 [000013] (770 B), in 1.0s (2.0s total), output rate 770 B/s
 
 enable-file-deletions
 ----
-[JOB 13] MANIFEST deleted 000010
+[JOB 13] MANIFEST deleted 000009
 
 ingest
 ----
-link: ext/0 -> db/000016.sst
-[JOB 14] ingesting: sstable created 000016
+link: ext/0 -> db/000015.sst
+[JOB 14] ingesting: sstable created 000015
 sync: db
-create: db/MANIFEST-000017
-close: db/MANIFEST-000015
-sync: db/MANIFEST-000017
-create: db/marker.manifest.000007.MANIFEST-000017
-close: db/marker.manifest.000007.MANIFEST-000017
+create: db/MANIFEST-000016
+close: db/MANIFEST-000014
+sync: db/MANIFEST-000016
+create: db/marker.manifest.000006.MANIFEST-000016
+close: db/marker.manifest.000006.MANIFEST-000016
 sync: db
-[JOB 14] MANIFEST created 000017
-[JOB 14] MANIFEST deleted 000012
-[JOB 14] ingested L0:000016 (825 B)
+[JOB 14] MANIFEST created 000016
+[JOB 14] MANIFEST deleted 000011
+[JOB 14] ingested L0:000015 (825 B)
 
 metrics
 ----
@@ -215,10 +208,10 @@ zmemtbl         0     0 B
 sstables
 ----
 0:
-  14:[a-a]
-  16:[a-a]
+  13:[a-a]
+  15:[a-a]
 6:
-  11:[a-a]
+  10:[a-a]
 
 checkpoint
 ----
@@ -227,28 +220,28 @@ open-dir:
 sync: 
 close: 
 open-dir: checkpoint
-link: db/OPTIONS-000004 -> checkpoint/OPTIONS-000004
+link: db/OPTIONS-000003 -> checkpoint/OPTIONS-000003
 open-dir: checkpoint
 create: checkpoint/marker.format-version.000001.011
 sync: checkpoint/marker.format-version.000001.011
 close: checkpoint/marker.format-version.000001.011
 sync: checkpoint
 close: checkpoint
-create: checkpoint/MANIFEST-000017
-sync: checkpoint/MANIFEST-000017
-close: checkpoint/MANIFEST-000017
+create: checkpoint/MANIFEST-000016
+sync: checkpoint/MANIFEST-000016
+close: checkpoint/MANIFEST-000016
 open-dir: checkpoint
-create: checkpoint/marker.manifest.000001.MANIFEST-000017
-sync: checkpoint/marker.manifest.000001.MANIFEST-000017
-close: checkpoint/marker.manifest.000001.MANIFEST-000017
+create: checkpoint/marker.manifest.000001.MANIFEST-000016
+sync: checkpoint/marker.manifest.000001.MANIFEST-000016
+close: checkpoint/marker.manifest.000001.MANIFEST-000016
 sync: checkpoint
 close: checkpoint
-link: db/000014.sst -> checkpoint/000014.sst
-link: db/000016.sst -> checkpoint/000016.sst
-link: db/000011.sst -> checkpoint/000011.sst
-create: checkpoint/000013.log
-sync: checkpoint/000013.log
-close: checkpoint/000013.log
+link: db/000013.sst -> checkpoint/000013.sst
+link: db/000015.sst -> checkpoint/000015.sst
+link: db/000010.sst -> checkpoint/000010.sst
+create: checkpoint/000012.log
+sync: checkpoint/000012.log
+close: checkpoint/000012.log
 sync: checkpoint
 close: checkpoint
 
@@ -259,9 +252,9 @@ pebble: file deletion disablement invariant violated
 close
 ----
 close: db
-sync: wal/000013.log
-close: wal/000013.log
-close: db/MANIFEST-000017
+sync: wal/000012.log
+close: wal/000012.log
+close: db/MANIFEST-000016
 close: db
 close: db
 close: wal
diff --git a/testdata/external_iterator b/testdata/external_iterator
index 90ed5f8b..9cc73c91 100644
--- a/testdata/external_iterator
+++ b/testdata/external_iterator
@@ -254,7 +254,7 @@ aaaa@1: (aaaa@1, .)
 aaaaa@3: (aaaaa@3, .)
 aaaaa@1: (aaaaa@1, .)
 stats: (interface (dir, seek, step): (fwd, 5, 5), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 5, 5), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 480 B, cached 0 B)), (points: (count 10, key-bytes 50, value-bytes 50, tombstoned: 0))
+(internal-stats: (block-bytes: (total 475 B, cached 0 B)), (points: (count 10, key-bytes 50, value-bytes 50, tombstoned: 0))
 
 # Note the inclusion of fwd-only. This iterator will use the TrySeekUsingNext
 # optimization and loads ~half the block-bytes as a result.
@@ -283,4 +283,4 @@ aaaa@1: (aaaa@1, .)
 aaaaa@3: (aaaaa@3, .)
 aaaaa@1: (aaaaa@1, .)
 stats: (interface (dir, seek, step): (fwd, 5, 5), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 5, 5), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 286 B, cached 0 B)), (points: (count 10, key-bytes 50, value-bytes 50, tombstoned: 0))
+(internal-stats: (block-bytes: (total 281 B, cached 0 B)), (points: (count 10, key-bytes 50, value-bytes 50, tombstoned: 0))
diff --git a/testdata/iter_histories/iter_optimizations b/testdata/iter_histories/iter_optimizations
index 5f2708e9..772d7603 100644
--- a/testdata/iter_histories/iter_optimizations
+++ b/testdata/iter_histories/iter_optimizations
@@ -200,7 +200,7 @@ stats
 lastPositioningOp="unknown"
 b@5: (b@5, .)
 stats: (interface (dir, seek, step): (fwd, 1, 0), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 1, 0), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 120 B, cached 0 B)), (points: (count 1, key-bytes 3, value-bytes 3, tombstoned: 0))
+(internal-stats: (block-bytes: (total 119 B, cached 0 B)), (points: (count 1, key-bytes 3, value-bytes 3, tombstoned: 0))
 
 mutate batch=foo
 set h@2 h@2
@@ -216,7 +216,7 @@ stats
 lastPositioningOp="seekprefixge"
 c@3: (c@3, .)
 stats: (interface (dir, seek, step): (fwd, 2, 0), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 2, 0), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 120 B, cached 0 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
+(internal-stats: (block-bytes: (total 119 B, cached 0 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
 
 mutate batch=foo
 set i@1 i@1
@@ -232,7 +232,7 @@ stats
 lastPositioningOp="seekprefixge"
 d@9: (d@9, .)
 stats: (interface (dir, seek, step): (fwd, 3, 0), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 3, 0), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 120 B, cached 0 B)), (points: (count 3, key-bytes 9, value-bytes 9, tombstoned: 0))
+(internal-stats: (block-bytes: (total 119 B, cached 0 B)), (points: (count 3, key-bytes 9, value-bytes 9, tombstoned: 0))
 
 mutate batch=foo
 set j@6 j@6
@@ -248,7 +248,7 @@ stats
 lastPositioningOp="seekprefixge"
 e@8: (e@8, .)
 stats: (interface (dir, seek, step): (fwd, 4, 0), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 4, 0), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 120 B, cached 0 B)), (points: (count 4, key-bytes 12, value-bytes 12, tombstoned: 0))
+(internal-stats: (block-bytes: (total 119 B, cached 0 B)), (points: (count 4, key-bytes 12, value-bytes 12, tombstoned: 0))
 
 # Ensure that a case eligible for TrySeekUsingNext across a SetOptions correctly
 # sees new batch mutations. The batch iterator should ignore the
@@ -340,7 +340,7 @@ seek-ge b@7
 .
 .
 .
-lastPositioningOp="unknown"
+lastPositioningOp="invalidate"
 b@6: (b@6, .)
 
 reset
diff --git a/testdata/iter_histories/range_key_changed b/testdata/iter_histories/range_key_changed
index ba4e8dda..f4e6d64c 100644
--- a/testdata/iter_histories/range_key_changed
+++ b/testdata/iter_histories/range_key_changed
@@ -272,4 +272,3 @@ first
 d: (., [d-e) @1=foo UPDATED)
 .
 c: (c, .)
-
diff --git a/testdata/iter_histories/range_key_masking b/testdata/iter_histories/range_key_masking
index 071a4b62..bcb29f5e 100644
--- a/testdata/iter_histories/range_key_masking
+++ b/testdata/iter_histories/range_key_masking
@@ -168,7 +168,7 @@ stats
 a: (., [a-z) @5=boop UPDATED)
 .
 stats: (interface (dir, seek, step): (fwd, 1, 1), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 1, 1), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 515 B, cached 515 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
+(internal-stats: (block-bytes: (total 514 B, cached 514 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
 
 # Perform a similar comparison in reverse.
 
@@ -190,7 +190,7 @@ stats
 a: (., [a-z) @5=boop UPDATED)
 .
 stats: (interface (dir, seek, step): (fwd, 0, 0), (rev, 1, 1)), (internal (dir, seek, step): (fwd, 0, 0), (rev, 1, 1)),
-(internal-stats: (block-bytes: (total 515 B, cached 515 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
+(internal-stats: (block-bytes: (total 514 B, cached 514 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
 
 # Perform similar comparisons with seeks.
 
@@ -202,7 +202,7 @@ stats
 m: (., [a-z) @5=boop UPDATED)
 .
 stats: (interface (dir, seek, step): (fwd, 1, 1), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 1, 1), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 790 B, cached 790 B)), (points: (count 13, key-bytes 39, value-bytes 39, tombstoned: 0))
+(internal-stats: (block-bytes: (total 789 B, cached 789 B)), (points: (count 13, key-bytes 39, value-bytes 39, tombstoned: 0))
 
 combined-iter mask-suffix=@9 mask-filter
 seek-ge m
@@ -212,7 +212,7 @@ stats
 m: (., [a-z) @5=boop UPDATED)
 .
 stats: (interface (dir, seek, step): (fwd, 1, 1), (rev, 0, 0)), (internal (dir, seek, step): (fwd, 1, 1), (rev, 0, 0)),
-(internal-stats: (block-bytes: (total 515 B, cached 515 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
+(internal-stats: (block-bytes: (total 514 B, cached 514 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
 
 combined-iter mask-suffix=@9
 seek-lt m
@@ -222,7 +222,7 @@ stats
 a: (., [a-z) @5=boop UPDATED)
 .
 stats: (interface (dir, seek, step): (fwd, 0, 0), (rev, 1, 1)), (internal (dir, seek, step): (fwd, 0, 0), (rev, 1, 1)),
-(internal-stats: (block-bytes: (total 790 B, cached 790 B)), (points: (count 12, key-bytes 36, value-bytes 36, tombstoned: 0))
+(internal-stats: (block-bytes: (total 789 B, cached 789 B)), (points: (count 12, key-bytes 36, value-bytes 36, tombstoned: 0))
 
 combined-iter mask-suffix=@9 mask-filter
 seek-lt m
@@ -232,4 +232,4 @@ stats
 a: (., [a-z) @5=boop UPDATED)
 .
 stats: (interface (dir, seek, step): (fwd, 0, 0), (rev, 1, 1)), (internal (dir, seek, step): (fwd, 0, 0), (rev, 1, 1)),
-(internal-stats: (block-bytes: (total 540 B, cached 540 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
+(internal-stats: (block-bytes: (total 539 B, cached 539 B)), (points: (count 2, key-bytes 6, value-bytes 6, tombstoned: 0))
diff --git a/version_set_test.go b/version_set_test.go
index 2bf0fc89..aa9ebed1 100644
--- a/version_set_test.go
+++ b/version_set_test.go
@@ -38,7 +38,10 @@ func TestVersionSetCheckpoint(t *testing.T) {
 
 	// Multiple manifest files are created such that the latest one must have a correct snapshot
 	// of the preceding state for the DB to be opened correctly and see the written data.
+	// Snapshot has no files, so first edit will cause manifest rotation.
 	writeAndIngest(t, mem, d, base.MakeInternalKey([]byte("a"), 0, InternalKeyKindSet), []byte("b"), "a")
+	// Snapshot has no files, and manifest has an edit from the previous ingest,
+	// so this second ingest will cause manifest rotation.
 	writeAndIngest(t, mem, d, base.MakeInternalKey([]byte("c"), 0, InternalKeyKindSet), []byte("d"), "c")
 	require.NoError(t, d.Close())
 	d, err = Open("", opts)
@@ -65,7 +68,10 @@ func TestVersionSetSeqNums(t *testing.T) {
 	d, err := Open("", opts)
 	require.NoError(t, err)
 
+	// Snapshot has no files, so first edit will cause manifest rotation.
 	writeAndIngest(t, mem, d, base.MakeInternalKey([]byte("a"), 0, InternalKeyKindSet), []byte("b"), "a")
+	// Snapshot has no files, and manifest has an edit from the previous ingest,
+	// so this second ingest will cause manifest rotation.
 	writeAndIngest(t, mem, d, base.MakeInternalKey([]byte("c"), 0, InternalKeyKindSet), []byte("d"), "c")
 	require.NoError(t, d.Close())
 	d, err = Open("", opts)