Repository: cockroachdb/pebble
Pull request #3632 (2024-05-23 20:02:02)

overlap: rework ingest LSM overlap check code
~This PR is only for the last commit (it is based on #3626).~

#### overlap: rework ingest LSM overlap check code

Instead of using level iterators, we directly check overlap with files
separately. This requires less machinery and there is no chance of
opening files that are outside of the bounds of interest.

We put the new code in a new `internal/overlap` package and
`overlapChecker` is now a very small wrapper which interfaces with
`overlap.Checker`.

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

RaduBerinde: *[`testdata/event_listener` line 189 at r2](https://reviewable.io/reviews/cockroachdb/pebble/3632#-NyetzpBBGWN46L_wx87:-Nyffyzn3UvUirasMCjX:b-ay96fx) ([raw file](https://github.com/cockroachdb/pebble/blob/2ac449bbd7e59d2c7a6d1402183fbdc3e02cf3e5/testdata/event_listener#L189)):*
<details><summary><i>Previously, sumeerbhola wrote…</i></summary><blockquote>

Why did we not need to open this sst for overlap checking? Glancing at `TestEventListener` it seems that both flushed data and ingested data is only using user key "a", so there will be boundary overlap, which means we will need to open the existing files to determine file overlap. What am I missing?
</blockquote></details>

We assume data overlap if the ingested file contains/encloses a file, which we now check first.


<!-- Sent from Reviewable.io -->


Patch:
diff --git a/db.go b/db.go
index f61e90fd..7a8cf541 100644
--- a/db.go
+++ b/db.go
@@ -3007,7 +3007,13 @@ func (d *DB) checkVirtualBounds(m *fileMetadata) {
 
 // DebugString returns a debugging string describing the LSM.
 func (d *DB) DebugString() string {
+	return d.DebugCurrentVersion().DebugString()
+}
+
+// DebugCurrentVersion returns the current LSM tree metadata. Should only be
+// used for testing/debugging.
+func (d *DB) DebugCurrentVersion() *manifest.Version {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	return d.mu.versions.currentVersion().DebugString()
+	return d.mu.versions.currentVersion()
 }
diff --git a/flushable.go b/flushable.go
index a2ed68c4..85aef6b4 100644
--- a/flushable.go
+++ b/flushable.go
@@ -394,3 +394,33 @@ func determineOverlapAllIters(
 	}
 	return false, nil
 }
+
+func determineOverlapPointIterator(
+	cmp base.Compare, bounds base.UserKeyBounds, iter internalIterator,
+) (bool, error) {
+	kv := iter.SeekGE(bounds.Start, base.SeekGEFlagsNone)
+	if kv == nil {
+		return false, iter.Error()
+	}
+	return bounds.End.IsUpperBoundForInternalKey(cmp, kv.K), nil
+}
+
+func determineOverlapKeyspanIterator(
+	cmp base.Compare, bounds base.UserKeyBounds, iter keyspan.FragmentIterator,
+) (bool, error) {
+	// NB: The spans surfaced by the fragment iterator are non-overlapping.
+	span, err := iter.SeekGE(bounds.Start)
+	if err != nil {
+		return false, err
+	}
+	for ; span != nil; span, err = iter.Next() {
+		if !bounds.End.IsUpperBoundFor(cmp, span.Start) {
+			// The span starts after our bounds.
+			return false, nil
+		}
+		if !span.Empty() {
+			return true, nil
+		}
+	}
+	return false, err
+}
diff --git a/ingest.go b/ingest.go
index c2df8301..7f4c9f31 100644
--- a/ingest.go
+++ b/ingest.go
@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/internal/overlap"
 	"github.com/cockroachdb/pebble/internal/private"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/remote"
@@ -832,7 +833,7 @@ func ingestUpdateSeqNum(
 func ingestTargetLevel(
 	ctx context.Context,
 	cmp base.Compare,
-	lsmOverlap lsmOverlap,
+	lsmOverlap overlap.WithLSM,
 	baseLevel int,
 	compactions map[*compaction]struct{},
 	meta *fileMetadata,
@@ -904,32 +905,32 @@ func ingestTargetLevel(
 	// existing point that falls within the ingested table bounds as being "data
 	// overlap".
 
-	if lsmOverlap[0].result == dataOverlap {
+	if lsmOverlap[0].Result == overlap.Data {
 		return 0, nil, nil
 	}
 	targetLevel = 0
 	splitFile = nil
 	for level := baseLevel; level < numLevels; level++ {
 		var candidateSplitFile *fileMetadata
-		switch lsmOverlap[level].result {
-		case dataOverlap:
+		switch lsmOverlap[level].Result {
+		case overlap.Data:
 			// We cannot ingest into or under this level; return the best target level
 			// so far.
 			return targetLevel, splitFile, nil
 
-		case noDataOverlap:
-			if !suggestSplit || lsmOverlap[level].splitFile == nil {
+		case overlap.OnlyBoundary:
+			if !suggestSplit || lsmOverlap[level].SplitFile == nil {
 				// We can ingest under this level, but not into this level.
 				continue
 			}
 			// We can ingest into this level if we split this file.
-			candidateSplitFile = lsmOverlap[level].splitFile
+			candidateSplitFile = lsmOverlap[level].SplitFile
 
-		case noBoundaryOverlap:
+		case overlap.None:
 		// We can ingest into this level.
 
 		default:
-			return 0, nil, base.AssertionFailedf("unexpected lsmOverlap result: %v", lsmOverlap[level].result)
+			return 0, nil, base.AssertionFailedf("unexpected WithLevel.Result: %v", lsmOverlap[level].Result)
 		}
 
 		// Check boundary overlap with any ongoing compactions. We consider an
@@ -2192,15 +2193,15 @@ func (d *DB) ingestApply(
 				// We check overlap against the LSM without holding DB.mu. Note that we
 				// are still holding the log lock, so the version cannot change.
 				// TODO(radu): perform this check optimistically outside of the log lock.
-				var overlap lsmOverlap
-				overlap, err = func() (lsmOverlap, error) {
+				var lsmOverlap overlap.WithLSM
+				lsmOverlap, err = func() (overlap.WithLSM, error) {
 					d.mu.Unlock()
 					defer d.mu.Lock()
 					return overlapChecker.DetermineLSMOverlap(ctx, m.UserKeyBounds())
 				}()
 				if err == nil {
 					f.Level, splitFile, err = ingestTargetLevel(
-						ctx, d.cmp, overlap, baseLevel, d.mu.compact.inProgress, m, shouldIngestSplit,
+						ctx, d.cmp, lsmOverlap, baseLevel, d.mu.compact.inProgress, m, shouldIngestSplit,
 					)
 				}
 			}
diff --git a/internal/base/key_bounds.go b/internal/base/key_bounds.go
index 9151451c..7a41f786 100644
--- a/internal/base/key_bounds.go
+++ b/internal/base/key_bounds.go
@@ -73,6 +73,23 @@ func (eb UserKeyBoundary) IsUpperBoundForInternalKey(cmp Compare, key InternalKe
 	return c < 0 || (c == 0 && (eb.Kind == Inclusive || key.IsExclusiveSentinel()))
 }
 
+// CompareUpperBounds compares two UserKeyBoundaries as upper bounds (e.g. when
+// they are used for UserKeyBounds.End).
+func (eb UserKeyBoundary) CompareUpperBounds(cmp Compare, other UserKeyBoundary) int {
+	switch c := cmp(eb.Key, other.Key); {
+	case c != 0:
+		return c
+	case eb.Kind == other.Kind:
+		return 0
+	case eb.Kind == Inclusive:
+		// eb is inclusive, other is exclusive.
+		return 1
+	default:
+		// eb is exclusive, other is inclusive.
+		return -1
+	}
+}
+
 // UserKeyBounds is a user key interval with an inclusive start boundary and
 // with an end boundary that can be either inclusive or exclusive.
 type UserKeyBounds struct {
@@ -132,8 +149,7 @@ func (b *UserKeyBounds) ContainsBounds(cmp Compare, other *UserKeyBounds) bool {
 	if cmp(b.Start, other.Start) > 0 {
 		return false
 	}
-	c := cmp(other.End.Key, b.End.Key)
-	return c < 0 || (c == 0 && (b.End.Kind == Inclusive || other.End.Kind == Exclusive))
+	return other.End.CompareUpperBounds(cmp, b.End) <= 0
 }
 
 // ContainsUserKey returns true if the user key is within the bounds.
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index 6f3d7d1d..375f56a4 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -239,6 +239,9 @@ type Iter struct {
 	// lastRangeDelSpan stores the last, not compacted tombstone span. It is used
 	// to elide points or mark them as snapshot-pinned.
 	lastRangeDelSpan keyspan.Span
+	// lastRangeDelSpanFrontier is the frontier used to clear out lastRangeDelSpan
+	// when we move beyond its end key.
+	lastRangeDelSpanFrontier frontier
 
 	// span stores the last, compacted tombstone or range key span. It is provided
 	// to the caller via Span().
@@ -327,6 +330,7 @@ func NewIter(
 	i.delElider.Init(i.cmp, cfg.TombstoneElision)
 	i.rangeDelCompactor = MakeRangeDelSpanCompactor(i.cmp, i.cfg.Comparer.Equal, cfg.Snapshots, cfg.TombstoneElision)
 	i.rangeKeyCompactor = MakeRangeKeySpanCompactor(i.cmp, i.cfg.Comparer.Equal, cfg.Snapshots, cfg.RangeKeyElision)
+	i.lastRangeDelSpanFrontier.Init(&i.frontiers, nil, i.lastRangeDelSpanFrontierReached)
 	return i
 }
 
@@ -408,6 +412,8 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 	i.pos = iterPosCurForward
 
 	for i.iterKV != nil {
+		i.frontiers.Advance(i.iterKV.K.UserKey)
+
 		// If we entered a new snapshot stripe with the same key, any key we
 		// return on this iteration is only returned because the open snapshot
 		// prevented it from being elided or merged with the key returned for
@@ -437,7 +443,7 @@ func (i *Iter) Next() (*base.InternalKey, []byte) {
 
 			if i.iterKV.Kind() == base.InternalKeyKindRangeDelete {
 				span := i.rangeDelInterleaving.Span()
-				i.lastRangeDelSpan.CopyFrom(span)
+				i.setLastRangeDelSpan(span)
 				i.rangeDelCompactor.Compact(span, &i.span)
 				if i.span.Empty() {
 					// The range del span was elided entirely; don't return this key to the caller.
@@ -1254,7 +1260,6 @@ func (i *Iter) saveKey() {
 		Trailer: i.iterKV.K.Trailer,
 	}
 	i.keyTrailer = i.key.Trailer
-	i.frontiers.Advance(i.key.UserKey)
 }
 
 // Error returns any error encountered.
@@ -1311,13 +1316,9 @@ func (i *Iter) tombstoneCovers(key base.InternalKey, snapshot uint64) cover {
 	if i.lastRangeDelSpan.Empty() {
 		return noCover
 	}
-	if invariants.Enabled && i.cmp(key.UserKey, i.lastRangeDelSpan.Start) < 0 {
+	if invariants.Enabled && (i.cmp(key.UserKey, i.lastRangeDelSpan.Start) < 0 || i.cmp(key.UserKey, i.lastRangeDelSpan.End) >= 0) {
 		panic(errors.AssertionFailedf("invalid key %q, last span %s", key, i.lastRangeDelSpan))
 	}
-	if i.cmp(key.UserKey, i.lastRangeDelSpan.End) >= 0 {
-		i.lastRangeDelSpan.Reset()
-		return noCover
-	}
 	// The Covers() check is very cheap, so we want to do that first.
 	switch {
 	case !i.lastRangeDelSpan.Covers(key.SeqNum()):
@@ -1329,6 +1330,19 @@ func (i *Iter) tombstoneCovers(key base.InternalKey, snapshot uint64) cover {
 	}
 }
 
+func (i *Iter) setLastRangeDelSpan(span *keyspan.Span) {
+	if invariants.Enabled && !i.lastRangeDelSpan.Empty() {
+		panic("last range del span overwritten")
+	}
+	i.lastRangeDelSpan.CopyFrom(span)
+	i.lastRangeDelSpanFrontier.Update(i.lastRangeDelSpan.End)
+}
+
+func (i *Iter) lastRangeDelSpanFrontierReached(key []byte) []byte {
+	i.lastRangeDelSpan.Reset()
+	return nil
+}
+
 // maybeZeroSeqnum attempts to set the seqnum for the current key to 0. Doing
 // so improves compression and enables an optimization during forward iteration
 // to skip some key comparisons. The seqnum for an entry can be zeroed if the
diff --git a/internal/manifest/level_metadata.go b/internal/manifest/level_metadata.go
index d0be0f8b..af7b012f 100644
--- a/internal/manifest/level_metadata.go
+++ b/internal/manifest/level_metadata.go
@@ -41,7 +41,8 @@ func (lm *LevelMetadata) release() (obsolete []*FileBacking) {
 	return lm.tree.Release()
 }
 
-func makeLevelMetadata(cmp Compare, level int, files []*FileMetadata) LevelMetadata {
+// MakeLevelMetadata creates a LevelMetadata with the given files.
+func MakeLevelMetadata(cmp Compare, level int, files []*FileMetadata) LevelMetadata {
 	bcmp := btreeCmpSeqNum
 	if level > 0 {
 		bcmp = btreeCmpSmallestKey(cmp)
diff --git a/internal/manifest/version.go b/internal/manifest/version.go
index af2c9de0..b584899d 100644
--- a/internal/manifest/version.go
+++ b/internal/manifest/version.go
@@ -301,6 +301,21 @@ func (m *FileMetadata) UserKeyBounds() base.UserKeyBounds {
 	return base.UserKeyBoundsFromInternal(m.Smallest, m.Largest)
 }
 
+// UserKeyBoundsByType returns the user key bounds for the given key types.
+// Note that the returned bounds are invalid when requesting KeyTypePoint but
+// HasPointKeys is false, or when requesting KeyTypeRange and HasRangeKeys is
+// false.
+func (m *FileMetadata) UserKeyBoundsByType(keyType KeyType) base.UserKeyBounds {
+	switch keyType {
+	case KeyTypePoint:
+		return base.UserKeyBoundsFromInternal(m.SmallestPointKey, m.LargestPointKey)
+	case KeyTypeRange:
+		return base.UserKeyBoundsFromInternal(m.SmallestRangeKey, m.LargestRangeKey)
+	default:
+		return base.UserKeyBoundsFromInternal(m.Smallest, m.Largest)
+	}
+}
+
 // SyntheticSeqNum returns a SyntheticSeqNum which is set when SmallestSeqNum
 // equals LargestSeqNum.
 func (m *FileMetadata) SyntheticSeqNum() sstable.SyntheticSeqNum {
diff --git a/internal/manifest/version_edit.go b/internal/manifest/version_edit.go
index 8e41280d..adb6315f 100644
--- a/internal/manifest/version_edit.go
+++ b/internal/manifest/version_edit.go
@@ -963,12 +963,12 @@ func (b *BulkVersionEdit) Apply(
 
 	for level := range v.Levels {
 		if curr == nil || curr.Levels[level].tree.root == nil {
-			v.Levels[level] = makeLevelMetadata(comparer.Compare, level, nil /* files */)
+			v.Levels[level] = MakeLevelMetadata(comparer.Compare, level, nil /* files */)
 		} else {
 			v.Levels[level] = curr.Levels[level].clone()
 		}
 		if curr == nil || curr.RangeKeyLevels[level].tree.root == nil {
-			v.RangeKeyLevels[level] = makeLevelMetadata(comparer.Compare, level, nil /* files */)
+			v.RangeKeyLevels[level] = MakeLevelMetadata(comparer.Compare, level, nil /* files */)
 		} else {
 			v.RangeKeyLevels[level] = curr.RangeKeyLevels[level].clone()
 		}
diff --git a/internal/overlap/checker.go b/internal/overlap/checker.go
new file mode 100644
index 00000000..c219b553
--- /dev/null
+++ b/internal/overlap/checker.go
@@ -0,0 +1,263 @@
+// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+// Package overlap provides facilities for checking whether tables have data
+// overlap.
+package overlap
+
+import (
+	"context"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/keyspan"
+	"github.com/cockroachdb/pebble/internal/manifest"
+)
+
+// WithLSM stores the result of checking for boundary and data overlap between a
+// region of key space and the LSM levels, starting from the top (L0) and
+// stopping at the highest level with data overlap.
+type WithLSM [manifest.NumLevels]WithLevel
+
+// WithLevel is the result of checking overlap against an LSM level.
+type WithLevel struct {
+	Result Kind
+	// SplitFile can be set only when result is OnlyBoundary. If it is set, this
+	// file can be split to free up the range of interest.
+	SplitFile *manifest.FileMetadata
+}
+
+// Kind indicates the kind of overlap detected between a key range and a level.
+// We check two types of overlap:
+//
+//   - file boundary overlap: whether the key range overlaps any of the level's
+//     user key boundaries;
+//
+//   - data overlap: whether the key range overlaps any keys or ranges in the
+//     level. Data overlap implies file boundary overlap.
+type Kind uint8
+
+const (
+	// None indicates that the key range of interest doesn't overlap any tables on
+	// the level.
+	None Kind = iota + 1
+	// OnlyBoundary indicates that there is boundary overlap but no data overlap.
+	OnlyBoundary
+	// Data indicates that at least a key or range in the level overlaps with the
+	// key range of interest. Note that the data overlap check is best-effort and
+	// there could be false positives.
+	Data
+)
+
+// Checker is used to check for data overlap between tables in the LSM and a
+// user key region of interest.
+type Checker struct {
+	cmp             base.Compare
+	iteratorFactory IteratorFactory
+}
+
+// IteratorFactory is an interface that is used by the Checker to create
+// iterators for a given table. All methods can return nil as an empty iterator.
+type IteratorFactory interface {
+	Points(ctx context.Context, m *manifest.FileMetadata) (base.InternalIterator, error)
+	RangeDels(ctx context.Context, m *manifest.FileMetadata) (keyspan.FragmentIterator, error)
+	RangeKeys(ctx context.Context, m *manifest.FileMetadata) (keyspan.FragmentIterator, error)
+}
+
+// MakeChecker initializes a new Checker.
+func MakeChecker(cmp base.Compare, iteratorFactory IteratorFactory) Checker {
+	return Checker{
+		cmp:             cmp,
+		iteratorFactory: iteratorFactory,
+	}
+}
+
+// LSMOverlap calculates the LSM overlap for the given region.
+func (c *Checker) LSMOverlap(
+	ctx context.Context, region base.UserKeyBounds, v *manifest.Version,
+) (WithLSM, error) {
+	var result WithLSM
+	result[0].Result = None
+	for sublevel := 0; sublevel < len(v.L0Sublevels.Levels); sublevel++ {
+		res, err := c.LevelOverlap(ctx, region, v.L0Sublevels.Levels[sublevel])
+		if err != nil {
+			return WithLSM{}, err
+		}
+		if res.Result == Data {
+			result[0].Result = Data
+			return result, nil
+		}
+		if res.Result == OnlyBoundary {
+			result[0].Result = OnlyBoundary
+		}
+	}
+	for level := 1; level < manifest.NumLevels; level++ {
+		var err error
+		result[level], err = c.LevelOverlap(ctx, region, v.Levels[level].Slice())
+		if err != nil {
+			return WithLSM{}, err
+		}
+		if result[level].Result == Data {
+			return result, err
+		}
+	}
+	return result, nil
+}
+
+// LevelOverlap returns true if there is possible data overlap between a user
+// key region and an L0 sublevel or L1+ level.
+func (c *Checker) LevelOverlap(
+	ctx context.Context, region base.UserKeyBounds, ls manifest.LevelSlice,
+) (WithLevel, error) {
+	// Quick check: if the target region contains any file boundaries, we assume
+	// data overlap. This is a correct assumption in most cases; it is pessimistic
+	// only for external ingestions which could have "loose" boundaries. External
+	// ingestions are also the most expensive to look at, so we don't want to do
+	// that just in the off chance that we'll find a significant empty region at
+	// the boundary.
+	//
+	// This check is important because the region can be very large in the key
+	// space and encompass many files, and we don't want to open any of them in
+	// that case.
+	startIter := ls.Iter()
+	file := startIter.SeekGE(c.cmp, region.Start)
+	if file == nil {
+		// No overlapping files.
+		return WithLevel{Result: None}, nil
+	}
+	fileBounds := file.UserKeyBounds()
+	if !region.End.IsUpperBoundFor(c.cmp, fileBounds.Start) {
+		// No overlapping files.
+		return WithLevel{Result: None}, nil
+	}
+	if c.cmp(fileBounds.Start, region.Start) >= 0 || region.End.CompareUpperBounds(c.cmp, fileBounds.End) >= 0 {
+		// The file ends or starts inside our region; we assume data overlap.
+		return WithLevel{Result: Data}, nil
+	}
+	// We have a single file to look at; its boundaries enclose our region.
+	empty, err := c.EmptyRegion(ctx, region, file)
+	if err != nil {
+		return WithLevel{}, err
+	}
+	if !empty {
+		return WithLevel{Result: Data}, nil
+	}
+	return WithLevel{
+		Result:    OnlyBoundary,
+		SplitFile: file,
+	}, nil
+}
+
+// EmptyRegion returns true if the given region doesn't overlap with any keys or
+// ranges in the given table.
+func (c *Checker) EmptyRegion(
+	ctx context.Context, region base.UserKeyBounds, m *manifest.FileMetadata,
+) (bool, error) {
+	empty, err := c.emptyRegionPointsAndRangeDels(ctx, region, m)
+	if err != nil || !empty {
+		return empty, err
+	}
+	return c.emptyRegionRangeKeys(ctx, region, m)
+}
+
+// emptyRegionPointsAndRangeDels returns true if the file doesn't contain any
+// point keys or range del spans that overlap with region.
+func (c *Checker) emptyRegionPointsAndRangeDels(
+	ctx context.Context, region base.UserKeyBounds, m *manifest.FileMetadata,
+) (bool, error) {
+	if !m.HasPointKeys {
+		return true, nil
+	}
+	pointBounds := m.UserKeyBoundsByType(manifest.KeyTypePoint)
+	if !pointBounds.Overlaps(c.cmp, &region) {
+		return true, nil
+	}
+	points, err := c.iteratorFactory.Points(ctx, m)
+	if err != nil {
+		return false, err
+	}
+	if points != nil {
+		defer points.Close()
+		var kv *base.InternalKV
+		if c.cmp(region.Start, pointBounds.Start) <= 0 {
+			kv = points.First()
+		} else {
+			kv = points.SeekGE(region.Start, base.SeekGEFlagsNone)
+		}
+		if kv == nil && points.Error() != nil {
+			return false, points.Error()
+		}
+		if kv != nil && region.End.IsUpperBoundForInternalKey(c.cmp, kv.K) {
+			// Found overlap.
+			return false, nil
+		}
+	}
+	rangeDels, err := c.iteratorFactory.RangeDels(ctx, m)
+	if err != nil {
+		return false, err
+	}
+	if rangeDels != nil {
+		defer rangeDels.Close()
+		empty, err := c.emptyFragmentRegion(region, pointBounds.Start, rangeDels)
+		if err != nil || !empty {
+			return empty, err
+		}
+	}
+	// Found no overlap.
+	return true, nil
+}
+
+// emptyRegionRangeKeys returns true if the file doesn't contain any range key
+// spans that overlap with region.
+func (c *Checker) emptyRegionRangeKeys(
+	ctx context.Context, region base.UserKeyBounds, m *manifest.FileMetadata,
+) (bool, error) {
+	if !m.HasRangeKeys {
+		return true, nil
+	}
+	rangeKeyBounds := m.UserKeyBoundsByType(manifest.KeyTypeRange)
+	if !rangeKeyBounds.Overlaps(c.cmp, &region) {
+		return true, nil
+	}
+	rangeKeys, err := c.iteratorFactory.RangeKeys(ctx, m)
+	if err != nil {
+		return false, err
+	}
+	if rangeKeys != nil {
+		defer rangeKeys.Close()
+		empty, err := c.emptyFragmentRegion(region, rangeKeyBounds.Start, rangeKeys)
+		if err != nil || !empty {
+			return empty, err
+		}
+	}
+	// Found no overlap.
+	return true, nil
+}
+
+// emptyFragmentRegion returns true if the given iterator doesn't contain any
+// spans that overlap with region. The fragmentLowerBounds is a known lower
+// bound for all the spans.
+func (c *Checker) emptyFragmentRegion(
+	region base.UserKeyBounds, fragmentLowerBound []byte, fragments keyspan.FragmentIterator,
+) (bool, error) {
+	var span *keyspan.Span
+	var err error
+	if c.cmp(region.Start, fragmentLowerBound) <= 0 {
+		// This is an optimization: we know there are no spans before region.Start,
+		// so we can use First.
+		span, err = fragments.First()
+	} else {
+		span, err = fragments.SeekGE(region.Start)
+	}
+	if err != nil {
+		return false, err
+	}
+	if span != nil && span.Empty() {
+		return false, base.AssertionFailedf("fragment iterator produced empty span")
+	}
+	if span != nil && region.End.IsUpperBoundFor(c.cmp, span.Start) {
+		// Found overlap.
+		return false, nil
+	}
+	return true, nil
+}
diff --git a/overlap.go b/overlap.go
index 561ea660..23cd4e8d 100644
--- a/overlap.go
+++ b/overlap.go
@@ -7,229 +7,61 @@ package pebble
 import (
 	"context"
 
-	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/keyspan"
-	"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
 	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/internal/overlap"
 )
 
 // An overlapChecker provides facilities for checking whether any keys within a
-// particular LSM version overlap a set of bounds.
+// particular LSM version overlap a set of bounds. It is a thin wrapper for
+// dataoverlap.Checker.
 type overlapChecker struct {
 	comparer *base.Comparer
 	newIters tableNewIters
 	opts     IterOptions
 	v        *version
-
-	// bufs; reused across calls to avoid allocations.
-	upperBoundBuf    []byte
-	pointLevelIter   levelIter
-	keyspanLevelIter keyspanimpl.LevelIter
 }
 
-// levelOverlapResult indicates the result of checking data overlap between a
-// key range and a level. We check two types of overlap:
-//
-//   - file boundary overlap: whether the key range overlaps any of the level's
-//     user key boundaries;
-//
-//   - data overlap: whether the key range overlaps any keys or ranges in the
-//     level. Data overlap implies file boundary overlap.
-type levelOverlapResult uint8
-
-const (
-	// The key range of interest doesn't overlap any tables on the level.
-	noBoundaryOverlap levelOverlapResult = iota + 1
-	// The key range of interest overlaps some tables on the level in terms of
-	// boundary, but the tables contain no data within that range.
-	noDataOverlap
-	// At least a key or range in the level overlaps with the key range of
-	// interest. Note that the data overlap check is best-effort and there could
-	// be false positives.
-	dataOverlap
-)
-
-// levelOverlap is the result of checking overlap against an LSM level.
-type levelOverlap struct {
-	result levelOverlapResult
-	// splitFile can be set only when result is noDataOverlap. If it is set, this
-	// file can be split to free up the range of interest.
-	splitFile *fileMetadata
-}
-
-// lsmOverlap stores the result of checking for data overlap for the LSM levels,
-// starting from the top (L0). The overlap checks are only populated up to the
-// first level where we find data overlap.
-//
-// This is akin to a tetris game where a horizontal bar is falling and we want
-// to determine where it lands.
-type lsmOverlap [numLevels]levelOverlap
-
-// DetermineLSMOverlap calculates the lsmOverlap for the given bounds.
+// DetermineLSMOverlap calculates the overlap.WithLSM for the given bounds.
 func (c *overlapChecker) DetermineLSMOverlap(
 	ctx context.Context, bounds base.UserKeyBounds,
-) (lsmOverlap, error) {
-	var result lsmOverlap
-	for level := 0; level < numLevels; level++ {
-		var err error
-		result[level], err = c.DetermineLevelOverlap(ctx, bounds, level)
-		if err != nil || result[level].result == dataOverlap {
-			return result, err
-		}
-	}
-	return result, nil
-}
-
-func (c *overlapChecker) DetermineLevelOverlap(
-	ctx context.Context, bounds base.UserKeyBounds, level int,
-) (levelOverlap, error) {
-	// First, check boundary overlap.
-	boundaryOverlaps := c.v.Overlaps(level, bounds)
-	if boundaryOverlaps.Empty() {
-		return levelOverlap{result: noBoundaryOverlap}, nil
-	}
-
-	overlap, err := c.DetermineAnyDataOverlapInLevel(ctx, bounds, level)
-	if err != nil || overlap {
-		return levelOverlap{result: dataOverlap}, err
-	}
-	var splitFile *fileMetadata
-	if boundaryOverlaps.Len() == 1 {
-		iter := boundaryOverlaps.Iter()
-		splitFile = iter.First()
-	}
-	return levelOverlap{
-		result:    noDataOverlap,
-		splitFile: splitFile,
-	}, nil
+) (overlap.WithLSM, error) {
+	checker := overlap.MakeChecker(c.comparer.Compare, c)
+	return checker.LSMOverlap(ctx, bounds, c.v)
 }
 
-// DetermineAnyDataOverlapInLevel checks whether any keys within the provided
-// bounds and level exist within the checker's associated LSM version. This
-// function checks for actual keys within the bounds, performing I/O if
-// necessary. It may return false when files within the level overlaps the
-// bounds, but the files do not contain any keys within the bounds.
-func (c *overlapChecker) DetermineAnyDataOverlapInLevel(
-	ctx context.Context, bounds base.UserKeyBounds, level int,
-) (bool, error) {
-	// Propagating an upper bound can prevent a levelIter from unnecessarily
-	// opening files that fall outside bounds if no files within a level overlap
-	// the provided bounds.
-	c.opts.UpperBound = nil
-	if bounds.End.Kind == base.Exclusive {
-		c.opts.UpperBound = bounds.End.Key
-	} else if c.comparer.ImmediateSuccessor != nil {
-		si := c.comparer.Split(bounds.End.Key)
-		c.upperBoundBuf = c.comparer.ImmediateSuccessor(c.upperBoundBuf[:0], bounds.End.Key[:si])
-		c.opts.UpperBound = c.upperBoundBuf
-	}
-
-	// Check for overlap over the keys of L0 by iterating over the sublevels.
-	// NB: sublevel 0 contains the newest keys, whereas sublevel n contains the
-	// oldest keys.
-	if level == 0 {
-		if c.v.L0Sublevels == nil {
-			return true, base.AssertionFailedf("could not read L0 sublevels")
-		}
-		for subLevel := 0; subLevel < len(c.v.L0SublevelFiles); subLevel++ {
-			manifestIter := c.v.L0Sublevels.Levels[subLevel].Iter()
-			pointOverlap, err := c.determinePointKeyOverlapInLevel(
-				ctx, bounds, manifest.Level(0), manifestIter)
-			if err != nil || pointOverlap {
-				return pointOverlap, err
-			}
-			rangeOverlap, err := c.determineRangeKeyOverlapInLevel(
-				ctx, bounds, manifest.Level(0), manifestIter)
-			if err != nil || rangeOverlap {
-				return rangeOverlap, err
-			}
-		}
-		return false, nil
-	}
-
-	pointManifestIter := c.v.Levels[level].Iter()
-	pointOverlap, err := c.determinePointKeyOverlapInLevel(
-		ctx, bounds, manifest.Level(level), pointManifestIter)
-	if pointOverlap || err != nil {
-		return pointOverlap, err
-	}
-	rangeManifestIter := c.v.RangeKeyLevels[level].Iter()
-	return c.determineRangeKeyOverlapInLevel(
-		ctx, bounds, manifest.Level(level), rangeManifestIter)
-}
+var _ overlap.IteratorFactory = (*overlapChecker)(nil)
 
-func (c *overlapChecker) determinePointKeyOverlapInLevel(
-	ctx context.Context,
-	bounds base.UserKeyBounds,
-	level manifest.Level,
-	metadataIter manifest.LevelIterator,
-) (bool, error) {
-	// Check for overlapping point keys.
-	{
-		c.pointLevelIter.init(ctx, c.opts, c.comparer, c.newIters, metadataIter, level, internalIterOpts{})
-		pointOverlap, err := determineOverlapPointIterator(c.comparer.Compare, bounds, &c.pointLevelIter)
-		err = errors.CombineErrors(err, c.pointLevelIter.Close())
-		if pointOverlap || err != nil {
-			return pointOverlap, err
-		}
-	}
-	// Check for overlapping range deletions.
-	{
-		c.keyspanLevelIter.Init(
-			keyspan.SpanIterOptions{}, c.comparer.Compare, tableNewRangeDelIter(ctx, c.newIters),
-			metadataIter, level, manifest.KeyTypePoint,
-		)
-		rangeDeletionOverlap, err := determineOverlapKeyspanIterator(c.comparer.Compare, bounds, &c.keyspanLevelIter)
-		err = errors.CombineErrors(err, c.keyspanLevelIter.Close())
-		if rangeDeletionOverlap || err != nil {
-			return rangeDeletionOverlap, err
-		}
+// Points is part of the overlap.IteratorFactory implementation.
+func (c *overlapChecker) Points(
+	ctx context.Context, m *manifest.FileMetadata,
+) (base.InternalIterator, error) {
+	iters, err := c.newIters(ctx, m, &c.opts, internalIterOpts{}, iterPointKeys)
+	if err != nil {
+		return nil, err
 	}
-	return false, nil
-}
-
-func (c *overlapChecker) determineRangeKeyOverlapInLevel(
-	ctx context.Context,
-	bounds base.UserKeyBounds,
-	level manifest.Level,
-	metadataIter manifest.LevelIterator,
-) (bool, error) {
-	// Check for overlapping range keys.
-	c.keyspanLevelIter.Init(
-		keyspan.SpanIterOptions{}, c.comparer.Compare, tableNewRangeKeyIter(ctx, c.newIters),
-		metadataIter, level, manifest.KeyTypeRange,
-	)
-	rangeKeyOverlap, err := determineOverlapKeyspanIterator(c.comparer.Compare, bounds, &c.keyspanLevelIter)
-	return rangeKeyOverlap, errors.CombineErrors(err, c.keyspanLevelIter.Close())
+	return iters.point, nil
 }
 
-func determineOverlapPointIterator(
-	cmp base.Compare, bounds base.UserKeyBounds, iter internalIterator,
-) (bool, error) {
-	kv := iter.SeekGE(bounds.Start, base.SeekGEFlagsNone)
-	if kv == nil {
-		return false, iter.Error()
+// RangeDels is part of the overlap.IteratorFactory implementation.
+func (c *overlapChecker) RangeDels(
+	ctx context.Context, m *manifest.FileMetadata,
+) (keyspan.FragmentIterator, error) {
+	iters, err := c.newIters(ctx, m, &c.opts, internalIterOpts{}, iterRangeDeletions)
+	if err != nil {
+		return nil, err
 	}
-	return bounds.End.IsUpperBoundForInternalKey(cmp, kv.K), nil
+	return iters.rangeDeletion, nil
 }
 
-func determineOverlapKeyspanIterator(
-	cmp base.Compare, bounds base.UserKeyBounds, iter keyspan.FragmentIterator,
-) (bool, error) {
-	// NB: The spans surfaced by the fragment iterator are non-overlapping.
-	span, err := iter.SeekGE(bounds.Start)
+// RangeKeys is part of the overlap.IteratorFactory implementation.
+func (c *overlapChecker) RangeKeys(
+	ctx context.Context, m *manifest.FileMetadata,
+) (keyspan.FragmentIterator, error) {
+	iters, err := c.newIters(ctx, m, &c.opts, internalIterOpts{}, iterRangeKeys)
 	if err != nil {
-		return false, err
-	}
-	for ; span != nil; span, err = iter.Next() {
-		if !bounds.End.IsUpperBoundFor(cmp, span.Start) {
-			// The span starts after our bounds.
-			return false, nil
-		}
-		if !span.Empty() {
-			return true, nil
-		}
+		return nil, err
 	}
-	return false, err
+	return iters.rangeKey, nil
 }


Test patch:
diff --git a/internal/base/key_bounds_test.go b/internal/base/key_bounds_test.go
index 6c7281ac..725cd903 100644
--- a/internal/base/key_bounds_test.go
+++ b/internal/base/key_bounds_test.go
@@ -10,6 +10,42 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestUserKeyBoundary(t *testing.T) {
+	cmp := DefaultComparer.Compare
+	a := []byte("a")
+	b := []byte("b")
+
+	aExclusive := UserKeyExclusive(a)
+	require.False(t, aExclusive.IsUpperBoundFor(cmp, a))
+	require.False(t, aExclusive.IsUpperBoundFor(cmp, b))
+
+	aInclusive := UserKeyInclusive(a)
+	require.True(t, aInclusive.IsUpperBoundFor(cmp, a))
+	require.False(t, aInclusive.IsUpperBoundFor(cmp, b))
+
+	bExclusive := UserKeyExclusive(b)
+	require.True(t, bExclusive.IsUpperBoundFor(cmp, a))
+	require.False(t, bExclusive.IsUpperBoundFor(cmp, b))
+
+	bInclusive := UserKeyInclusive(b)
+	require.True(t, bInclusive.IsUpperBoundFor(cmp, a))
+	require.True(t, bInclusive.IsUpperBoundFor(cmp, b))
+
+	ordered := []UserKeyBoundary{aExclusive, aInclusive, bExclusive, bInclusive}
+	for i := range ordered {
+		for j := range ordered {
+			expected := 0
+			if i < j {
+				expected = -1
+			} else if i > j {
+				expected = 1
+			}
+			require.Equalf(t, expected, ordered[i].CompareUpperBounds(cmp, ordered[j]),
+				"%v, %v", ordered[i], ordered[j])
+		}
+	}
+}
+
 func TestUserKeyBounds(t *testing.T) {
 	var ukb UserKeyBounds
 	cmp := DefaultComparer.Compare
diff --git a/internal/base/test_utils.go b/internal/base/test_utils.go
index b15001dd..08871b73 100644
--- a/internal/base/test_utils.go
+++ b/internal/base/test_utils.go
@@ -6,6 +6,7 @@ package base
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"strconv"
 	"strings"
@@ -249,3 +250,14 @@ func (f *FakeIter) SetBounds(lower, upper []byte) {
 
 // SetContext is part of the InternalIterator interface.
 func (f *FakeIter) SetContext(_ context.Context) {}
+
+// ParseUserKeyBounds parses UserKeyBounds from a string representation of the
+// form "[foo, bar]" or "[foo, bar)".
+func ParseUserKeyBounds(s string) UserKeyBounds {
+	first, last, s := s[0], s[len(s)-1], s[1:len(s)-1]
+	start, end, ok := strings.Cut(s, ", ")
+	if !ok || first != '[' || (last != ']' && last != ')') {
+		panic(fmt.Sprintf("invalid bounds %q", s))
+	}
+	return UserKeyBoundsEndExclusiveIf([]byte(start), []byte(end), last == ')')
+}
diff --git a/internal/manifest/l0_sublevels_test.go b/internal/manifest/l0_sublevels_test.go
index 503980f9..279ee5ef 100644
--- a/internal/manifest/l0_sublevels_test.go
+++ b/internal/manifest/l0_sublevels_test.go
@@ -289,7 +289,7 @@ func TestL0Sublevels(t *testing.T) {
 				SortBySmallest(fileMetas[i], base.DefaultComparer.Compare)
 			}
 
-			levelMetadata := makeLevelMetadata(base.DefaultComparer.Compare, 0, fileMetas[0])
+			levelMetadata := MakeLevelMetadata(base.DefaultComparer.Compare, 0, fileMetas[0])
 			if initialize {
 				if addL0FilesOpt {
 					SortBySeqNum(addedL0Files)
@@ -550,7 +550,7 @@ func TestAddL0FilesEquivalence(t *testing.T) {
 			continue
 		}
 
-		levelMetadata := makeLevelMetadata(testkeys.Comparer.Compare, 0, fileMetas)
+		levelMetadata := MakeLevelMetadata(testkeys.Comparer.Compare, 0, fileMetas)
 		var err error
 
 		if s2 == nil {
diff --git a/internal/manifest/version_test.go b/internal/manifest/version_test.go
index 697e3209..7feaabf3 100644
--- a/internal/manifest/version_test.go
+++ b/internal/manifest/version_test.go
@@ -22,7 +22,7 @@ import (
 )
 
 func levelMetadata(level int, files ...*FileMetadata) LevelMetadata {
-	return makeLevelMetadata(base.DefaultComparer.Compare, level, files)
+	return MakeLevelMetadata(base.DefaultComparer.Compare, level, files)
 }
 
 func ikey(s string) InternalKey {
@@ -82,7 +82,7 @@ func TestIkeyRange(t *testing.T) {
 				f = append(f, m)
 			}
 		}
-		levelMetadata := makeLevelMetadata(base.DefaultComparer.Compare, 0, f)
+		levelMetadata := MakeLevelMetadata(base.DefaultComparer.Compare, 0, f)
 
 		sm, la := KeyRange(base.DefaultComparer.Compare, levelMetadata.Iter())
 		got := string(sm.UserKey) + "-" + string(la.UserKey)
diff --git a/internal/overlap/checker_test.go b/internal/overlap/checker_test.go
new file mode 100644
index 00000000..21a2fc53
--- /dev/null
+++ b/internal/overlap/checker_test.go
@@ -0,0 +1,244 @@
+// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package overlap
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"math/rand"
+	"strings"
+	"testing"
+
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/keyspan"
+	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/stretchr/testify/require"
+)
+
+func TestChecker(t *testing.T) {
+	tables := newTestTables()
+	byName := make(map[string]*manifest.FileMetadata)
+
+	datadriven.RunTest(t, "testdata/checker", func(t *testing.T, d *datadriven.TestData) string {
+		switch d.Cmd {
+		case "define":
+			tt := testTable{
+				name: d.CmdArgs[0].String(),
+				meta: &manifest.FileMetadata{
+					FileNum: base.FileNum(1 + len(tables.tables)),
+				},
+			}
+			tt.meta.InitPhysicalBacking()
+			byName[tt.name] = tt.meta
+			var overrideBounds base.UserKeyBounds
+			for _, section := range splitLinesInSections(d.Input) {
+				fields := strings.Fields(section[0])
+				switch fields[0] {
+				case "points:":
+					for _, key := range section[1:] {
+						tt.points = append(tt.points, base.InternalKV{
+							K: base.ParseInternalKey(key),
+						})
+					}
+
+				case "range-dels:":
+					for _, span := range section[1:] {
+						tt.rangeDels = append(tt.rangeDels, keyspan.ParseSpan(span))
+					}
+
+				case "range-keys:":
+					for _, span := range section[1:] {
+						tt.rangeKeys = append(tt.rangeKeys, keyspan.ParseSpan(span))
+					}
+
+				case "override-bounds:":
+					overrideBounds = base.ParseUserKeyBounds(strings.Join(fields[1:], " "))
+
+				default:
+					d.Fatalf(t, "unknown section %q", fields[0])
+				}
+			}
+
+			if overrideBounds.Start != nil {
+				// Note: this code matches the logic used for external ingestions.
+				if len(tt.points) > 0 || len(tt.rangeDels) > 0 {
+					endKey := base.MakeInternalKey(overrideBounds.End.Key, 0, 0)
+					if overrideBounds.End.Kind == base.Exclusive {
+						endKey = base.MakeRangeDeleteSentinelKey(overrideBounds.End.Key)
+					}
+					tt.meta.ExtendPointKeyBounds(
+						bytes.Compare,
+						base.MakeInternalKey(overrideBounds.Start, 0, base.InternalKeyKindMax),
+						endKey,
+					)
+				}
+				if len(tt.rangeKeys) > 0 {
+					tt.meta.ExtendRangeKeyBounds(
+						bytes.Compare,
+						base.MakeInternalKey(overrideBounds.Start, 0, base.InternalKeyKindRangeKeyMax),
+						base.MakeExclusiveSentinelKey(base.InternalKeyKindRangeKeyMin, overrideBounds.End.Key),
+					)
+				}
+			} else {
+				if len(tt.points) > 0 {
+					tt.meta.ExtendPointKeyBounds(bytes.Compare, tt.points[0].K, tt.points[len(tt.points)-1].K)
+				}
+				if len(tt.rangeDels) > 0 {
+					smallest, largest := boundsFromSpans(tt.rangeDels)
+					tt.meta.ExtendPointKeyBounds(bytes.Compare, smallest, largest)
+				}
+				if len(tt.rangeKeys) > 0 {
+					smallest, largest := boundsFromSpans(tt.rangeKeys)
+					tt.meta.ExtendRangeKeyBounds(bytes.Compare, smallest, largest)
+				}
+			}
+			tables.tables[tt.meta] = tt
+
+		case "overlap":
+			var metas []*manifest.FileMetadata
+			lines := strings.Split(d.Input, "\n")
+			for _, arg := range d.CmdArgs {
+				name := arg.String()
+				m := byName[name]
+				if m == nil {
+					d.Fatalf(t, "unknown table %q", name)
+				}
+				metas = append(metas, m)
+			}
+			levelMeta := manifest.MakeLevelMetadata(bytes.Compare, 1 /* level */, metas)
+			c := MakeChecker(bytes.Compare, tables)
+			var buf strings.Builder
+			for _, l := range lines {
+				bounds := base.ParseUserKeyBounds(l)
+				withLevel, err := c.LevelOverlap(context.Background(), bounds, levelMeta.Slice())
+				require.NoError(t, err)
+				switch withLevel.Result {
+				case None:
+					fmt.Fprintf(&buf, "%s: no overlap", bounds)
+				case OnlyBoundary:
+					splitFile := ""
+					if withLevel.SplitFile != nil {
+						splitFile = fmt.Sprintf(" (split file: %s)", tables.tables[withLevel.SplitFile].name)
+					}
+					fmt.Fprintf(&buf, "%s: boundary overlap, no data overlap%s", bounds, splitFile)
+				case Data:
+					fmt.Fprintf(&buf, "%s: possible data overlap", bounds)
+				default:
+					d.Fatalf(t, "invalid result %v", withLevel)
+				}
+				fmt.Fprintf(&buf, "   iterators opened: %s\n", tables.OpenedIterators())
+			}
+			return buf.String()
+
+		default:
+			d.Fatalf(t, "unknown command %q", d.Cmd)
+		}
+		return ""
+	})
+}
+
+type testTable struct {
+	name      string
+	meta      *manifest.FileMetadata
+	points    []base.InternalKV
+	rangeDels []keyspan.Span
+	rangeKeys []keyspan.Span
+}
+
+var _ IteratorFactory = (*testTables)(nil)
+
+// testTables implements the IteratorFactory interface for testing purposes.
+type testTables struct {
+	tables map[*manifest.FileMetadata]testTable
+	// openedIterators keeps track of the iterators we opened. It is reset before
+	// every overlap check.
+	openedIterators []string
+}
+
+func newTestTables() *testTables {
+	return &testTables{
+		tables: make(map[*manifest.FileMetadata]testTable),
+	}
+}
+
+func (tt *testTables) OpenedIterators() string {
+	if len(tt.openedIterators) == 0 {
+		return "none"
+	}
+	res := strings.Join(tt.openedIterators, ", ")
+	tt.openedIterators = nil
+	return res
+}
+
+func (tt *testTables) Points(
+	ctx context.Context, m *manifest.FileMetadata,
+) (base.InternalIterator, error) {
+	t := tt.get(m)
+	tt.openedIterators = append(tt.openedIterators, fmt.Sprintf("%s/points", t.name))
+	if len(t.points) == 0 && rand.Intn(2) == 0 {
+		return nil, nil
+	}
+	return base.NewFakeIter(t.points), nil
+}
+
+func (tt *testTables) RangeDels(
+	ctx context.Context, m *manifest.FileMetadata,
+) (keyspan.FragmentIterator, error) {
+	t := tt.get(m)
+	tt.openedIterators = append(tt.openedIterators, fmt.Sprintf("%s/range-del", t.name))
+	if len(t.rangeDels) == 0 && rand.Intn(2) == 0 {
+		return nil, nil
+	}
+	return keyspan.NewIter(bytes.Compare, t.rangeDels), nil
+}
+
+func (tt *testTables) RangeKeys(
+	ctx context.Context, m *manifest.FileMetadata,
+) (keyspan.FragmentIterator, error) {
+	t := tt.get(m)
+	tt.openedIterators = append(tt.openedIterators, fmt.Sprintf("%s/range-key", t.name))
+	if len(t.rangeKeys) == 0 && rand.Intn(2) == 0 {
+		return nil, nil
+	}
+	return keyspan.NewIter(bytes.Compare, t.rangeKeys), nil
+}
+
+func (tt *testTables) get(m *manifest.FileMetadata) testTable {
+	t, ok := tt.tables[m]
+	if !ok {
+		panic("table not found")
+	}
+	return t
+}
+
+// splitLinesInSections parses a multi-line string where each line that is not
+// indented starts a section which also includes all indented lines that follow
+// it.
+func splitLinesInSections(input string) [][]string {
+	var res [][]string
+	for _, l := range strings.Split(input, "\n") {
+		if l[0] == ' ' || l[0] == '\t' {
+			if len(res) == 0 {
+				panic(fmt.Sprintf("invalid first line %q", l))
+			}
+			res[len(res)-1] = append(res[len(res)-1], strings.TrimSpace(l))
+		} else {
+			res = append(res, []string{l})
+		}
+	}
+	return res
+}
+
+func boundsFromSpans(spans []keyspan.Span) (smallest, largest base.InternalKey) {
+	smallest = base.InternalKey{
+		UserKey: spans[0].Start,
+		Trailer: spans[0].Keys[0].Trailer,
+	}
+	last := spans[len(spans)-1]
+	largest = base.MakeExclusiveSentinelKey(last.Keys[len(last.Keys)-1].Kind(), last.End)
+	return smallest, largest
+}
diff --git a/internal/overlap/testdata/checker b/internal/overlap/testdata/checker
new file mode 100644
index 00000000..ea7e31f6
--- /dev/null
+++ b/internal/overlap/testdata/checker
@@ -0,0 +1,119 @@
+define t1
+points:
+  b.SET.1
+  d.SET.1
+  u.SET.1
+----
+
+overlap t1
+[a, b)
+[a, b]
+[b1, b2]
+[b, c]
+[d, e]
+[e, u)
+[e, u]
+[u, w]
+[v, w]
+----
+[a, b): no overlap   iterators opened: none
+[a, b]: possible data overlap   iterators opened: none
+[b1, b2]: boundary overlap, no data overlap (split file: t1)   iterators opened: t1/points, t1/range-del
+[b, c]: possible data overlap   iterators opened: none
+[d, e]: possible data overlap   iterators opened: t1/points
+[e, u): boundary overlap, no data overlap (split file: t1)   iterators opened: t1/points, t1/range-del
+[e, u]: possible data overlap   iterators opened: none
+[u, w]: possible data overlap   iterators opened: none
+[v, w]: no overlap   iterators opened: none
+
+define t2
+range-dels:
+  b-e:{#1,RANGEDEL}
+  h-k:{#1,RANGEDEL}
+----
+
+overlap t2
+[a, b)
+[a, b]
+[d, f]
+[e, f]
+[e, h)
+[e, h]
+[j, l)
+[k, l]
+----
+[a, b): no overlap   iterators opened: none
+[a, b]: possible data overlap   iterators opened: none
+[d, f]: possible data overlap   iterators opened: t2/points, t2/range-del
+[e, f]: boundary overlap, no data overlap (split file: t2)   iterators opened: t2/points, t2/range-del
+[e, h): boundary overlap, no data overlap (split file: t2)   iterators opened: t2/points, t2/range-del
+[e, h]: possible data overlap   iterators opened: t2/points, t2/range-del
+[j, l): possible data overlap   iterators opened: none
+[k, l]: no overlap   iterators opened: none
+
+define t3
+range-keys:
+  b-e:{#1,RANGEKEYDEL}
+  h-k:{#1,RANGEKEYDEL}
+----
+
+overlap t3
+[a, b)
+[a, b]
+[d, f]
+[e, f]
+[e, h)
+[e, h]
+[j, l)
+[k, l]
+----
+[a, b): no overlap   iterators opened: none
+[a, b]: possible data overlap   iterators opened: none
+[d, f]: possible data overlap   iterators opened: t3/range-key
+[e, f]: boundary overlap, no data overlap (split file: t3)   iterators opened: t3/range-key
+[e, h): boundary overlap, no data overlap (split file: t3)   iterators opened: t3/range-key
+[e, h]: possible data overlap   iterators opened: t3/range-key
+[j, l): possible data overlap   iterators opened: none
+[k, l]: no overlap   iterators opened: none
+
+
+define tBE
+points:
+  b.SET.1
+  d.SET.1
+range-keys:
+  c-e:{#1,RANGEKEYDEL}
+----
+
+define tFM
+override-bounds: [f, m)
+points:
+  g.SET.1
+  j.SET.1
+----
+
+define tMP
+override-bounds: [m, p)
+points:
+  n.SET.1
+  o.SET.1
+----
+
+overlap tBE tFM tMP
+[b1, b2]
+[c1, c2]
+[e, f)
+[g1, g2]
+[j1, m1]
+[n1, n2]
+[n, o]
+[a, z]
+----
+[b1, b2]: boundary overlap, no data overlap (split file: tBE)   iterators opened: tBE/points, tBE/range-del
+[c1, c2]: possible data overlap   iterators opened: tBE/points, tBE/range-del, tBE/range-key
+[e, f): no overlap   iterators opened: none
+[g1, g2]: boundary overlap, no data overlap (split file: tFM)   iterators opened: tFM/points, tFM/range-del
+[j1, m1]: possible data overlap   iterators opened: none
+[n1, n2]: boundary overlap, no data overlap (split file: tMP)   iterators opened: tMP/points, tMP/range-del
+[n, o]: possible data overlap   iterators opened: tMP/points
+[a, z]: possible data overlap   iterators opened: none
diff --git a/testdata/event_listener b/testdata/event_listener
index 1a5e45ff..2d6650c9 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -186,12 +186,6 @@ close: ext/0
 link: ext/0 -> db/000015.sst
 [JOB 10] ingesting: sstable created 000015
 sync: db
-open: db/000013.sst
-read-at(537, 53): db/000013.sst
-read-at(500, 37): db/000013.sst
-read-at(53, 447): db/000013.sst
-read-at(26, 27): db/000013.sst
-read-at(0, 26): db/000013.sst
 create: db/MANIFEST-000016
 close: db/MANIFEST-000014
 sync: db/MANIFEST-000016
@@ -229,8 +223,8 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 1.7KB
 Compression types: snappy: 3
-Block cache: 6 entries (970B)  hit rate: 0.0%
-Table cache: 1 entries (760B)  hit rate: 40.0%
+Block cache: 3 entries (485B)  hit rate: 0.0%
+Table cache: 0 entries (0B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -330,8 +324,8 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 3.5KB
 Compression types: snappy: 6
-Block cache: 12 entries (1.9KB)  hit rate: 7.7%
-Table cache: 1 entries (760B)  hit rate: 50.0%
+Block cache: 9 entries (1.4KB)  hit rate: 0.0%
+Table cache: 0 entries (0B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -395,7 +389,6 @@ pebble: file deletion disablement invariant violated
 close
 ----
 close: db
-close: db/000013.sst
 sync-data: wal/000021.log
 close: wal/000021.log
 close: wal
diff --git a/testdata/ingest_external b/testdata/ingest_external
index dc6edf5a..7cca6fb2 100644
--- a/testdata/ingest_external
+++ b/testdata/ingest_external
@@ -816,9 +816,10 @@ ingest local1
 
 lsm
 ----
-L6:
+L0.0:
   000005:[a#11,SET-a#11,SET]
-  000006(000004):[aa@10#10,SET-c#inf,RANGEDEL]
+L6:
+  000004(000004):[a#10,DELSIZED-c#inf,RANGEDEL]
 
 iter
 first
diff --git a/testdata/metrics b/testdata/metrics
index b59e6ab7..6776db1c 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -487,8 +487,8 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 4.3KB
 Compression types: snappy: 7
-Block cache: 12 entries (1.9KB)  hit rate: 16.7%
-Table cache: 1 entries (760B)  hit rate: 60.0%
+Block cache: 12 entries (1.9KB)  hit rate: 9.1%
+Table cache: 1 entries (760B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -498,7 +498,7 @@ Iter category stats:
                    b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
                    c, non-latency: {BlockBytes:44 BlockBytesInCache:44 BlockReadDuration:0s}
    pebble-compaction, non-latency: {BlockBytes:301 BlockBytesInCache:44 BlockReadDuration:60ms}
-       pebble-ingest,     latency: {BlockBytes:192 BlockBytesInCache:128 BlockReadDuration:10ms}
+       pebble-ingest,     latency: {BlockBytes:64 BlockBytesInCache:0 BlockReadDuration:10ms}
 
 batch
 set g g
@@ -550,8 +550,8 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 6.1KB
 Compression types: snappy: 10
-Block cache: 12 entries (1.9KB)  hit rate: 16.7%
-Table cache: 1 entries (760B)  hit rate: 60.0%
+Block cache: 12 entries (1.9KB)  hit rate: 9.1%
+Table cache: 1 entries (760B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -561,7 +561,7 @@ Iter category stats:
                    b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
                    c, non-latency: {BlockBytes:44 BlockBytesInCache:44 BlockReadDuration:0s}
    pebble-compaction, non-latency: {BlockBytes:301 BlockBytesInCache:44 BlockReadDuration:60ms}
-       pebble-ingest,     latency: {BlockBytes:192 BlockBytesInCache:128 BlockReadDuration:10ms}
+       pebble-ingest,     latency: {BlockBytes:64 BlockBytesInCache:0 BlockReadDuration:10ms}
 
 build ext1
 set z z
@@ -638,7 +638,7 @@ Iter category stats:
                    b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
                    c, non-latency: {BlockBytes:44 BlockBytesInCache:44 BlockReadDuration:0s}
    pebble-compaction, non-latency: {BlockBytes:301 BlockBytesInCache:44 BlockReadDuration:60ms}
-       pebble-ingest,     latency: {BlockBytes:328 BlockBytesInCache:128 BlockReadDuration:30ms}
+       pebble-ingest,     latency: {BlockBytes:200 BlockBytesInCache:0 BlockReadDuration:30ms}
 
 # Virtualize a virtual sstable.
 build ext1
@@ -740,7 +740,7 @@ Iter category stats:
                    b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
                    c, non-latency: {BlockBytes:44 BlockBytesInCache:44 BlockReadDuration:0s}
    pebble-compaction, non-latency: {BlockBytes:677 BlockBytesInCache:376 BlockReadDuration:70ms}
-       pebble-ingest,     latency: {BlockBytes:400 BlockBytesInCache:200 BlockReadDuration:30ms}
+       pebble-ingest,     latency: {BlockBytes:272 BlockBytesInCache:72 BlockReadDuration:30ms}
 
 # Create a DB where lower levels are written as shared tables. All ingests also
 # become shared tables.