Repository: cockroachdb/pebble
Pull request #3760 (2024-07-15 20:26:19)

manifest: improve Annotator interface with generics
Refactors `manifest.Annotator` to use generics and a simplified API. This eliminates the need to perform pointer manipulation and unsafe typecasting when defining a new Annotator. The goal of this change is to improve the `Annotator` interface while not changing any existing behavior.

`BenchmarkNumFilesAnnotator` shows roughly the same performance as master when compared
to the equivalent implementation of `orderStatistic`:
```
pkg: github.com/cockroachdb/pebble/internal/manifest
                     │     old     │             new              │
                     │   sec/op    │   sec/op     vs base         │
NumFilesAnnotator-10   1.635µ ± 1%   1.572µ ± 7%  ~ (p=0.065 n=6)

                     │    old     │              new              │
                     │    B/op    │    B/op     vs base           │
NumFilesAnnotator-10   536.0 ± 0%   536.0 ± 0%  ~ (p=1.000 n=6) ¹
¹ all samples are equal

                     │    old     │              new              │
                     │ allocs/op  │ allocs/op   vs base           │
NumFilesAnnotator-10   7.000 ± 0%   7.000 ± 0%  ~ (p=1.000 n=6) ¹
¹ all samples are equal
```

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


Patch:
diff --git a/compaction.go b/compaction.go
index 71186e86..44cd103d 100644
--- a/compaction.go
+++ b/compaction.go
@@ -807,9 +807,8 @@ func (c *compaction) newInputIters(
 				}
 			}
 			if hasRangeKeys {
-				li := &keyspanimpl.LevelIter{}
-				newRangeKeyIterWrapper := func(file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
-					rangeKeyIter, err := newRangeKeyIter(file, iterOptions)
+				newRangeKeyIterWrapper := func(ctx context.Context, file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
+					rangeKeyIter, err := newRangeKeyIter(ctx, file, iterOptions)
 					if err != nil {
 						return nil, err
 					} else if rangeKeyIter == nil {
@@ -829,7 +828,10 @@ func (c *compaction) newInputIters(
 					// in this sstable must wholly lie within the file's bounds.
 					return noCloseIter, err
 				}
-				li.Init(keyspan.SpanIterOptions{}, c.cmp, newRangeKeyIterWrapper, level.files.Iter(), l, manifest.KeyTypeRange)
+				li := keyspanimpl.NewLevelIter(
+					context.Background(), keyspan.SpanIterOptions{}, c.cmp,
+					newRangeKeyIterWrapper, level.files.Iter(), l, manifest.KeyTypeRange,
+				)
 				rangeKeyIters = append(rangeKeyIters, li)
 			}
 			return nil
diff --git a/compaction_picker.go b/compaction_picker.go
index 68cb9c55..273772af 100644
--- a/compaction_picker.go
+++ b/compaction_picker.go
@@ -639,39 +639,13 @@ func compensatedSize(f *fileMetadata) uint64 {
 	return f.Size + fileCompensation(f)
 }
 
-// compensatedSizeAnnotator implements manifest.Annotator, annotating B-Tree
-// nodes with the sum of the files' compensated sizes. Its annotation type is
-// a *uint64. Compensated sizes may change once a table's stats are loaded
-// asynchronously, so its values are marked as cacheable only if a file's
-// stats have been loaded.
-type compensatedSizeAnnotator struct {
-}
-
-var _ manifest.Annotator = compensatedSizeAnnotator{}
-
-func (a compensatedSizeAnnotator) Zero(dst interface{}) interface{} {
-	if dst == nil {
-		return new(uint64)
-	}
-	v := dst.(*uint64)
-	*v = 0
-	return v
-}
-
-func (a compensatedSizeAnnotator) Accumulate(
-	f *fileMetadata, dst interface{},
-) (v interface{}, cacheOK bool) {
-	vptr := dst.(*uint64)
-	*vptr = *vptr + compensatedSize(f)
-	return vptr, f.StatsValid()
-}
-
-func (a compensatedSizeAnnotator) Merge(src interface{}, dst interface{}) interface{} {
-	srcV := src.(*uint64)
-	dstV := dst.(*uint64)
-	*dstV = *dstV + *srcV
-	return dstV
-}
+// compensatedSizeAnnotator is a manifest.Annotator that annotates B-Tree
+// nodes with the sum of the files' compensated sizes. Compensated sizes may
+// change once a table's stats are loaded asynchronously, so its values are
+// marked as cacheable only if a file's stats have been loaded.
+var compensatedSizeAnnotator = manifest.SumAnnotator(func(f *fileMetadata) (uint64, bool) {
+	return compensatedSize(f), f.StatsValid()
+})
 
 // totalCompensatedSize computes the compensated size over a file metadata
 // iterator. Note that this function is linear in the files available to the
@@ -912,10 +886,6 @@ func calculateSizeAdjust(inProgressCompactions []compactionInfo) [numLevels]leve
 	return sizeAdjust
 }
 
-func levelCompensatedSize(lm manifest.LevelMetadata) uint64 {
-	return *lm.Annotation(compensatedSizeAnnotator{}).(*uint64)
-}
-
 func (p *compactionPickerByScore) calculateLevelScores(
 	inProgressCompactions []compactionInfo,
 ) [numLevels]candidateLevelInfo {
@@ -932,7 +902,7 @@ func (p *compactionPickerByScore) calculateLevelScores(
 	}
 	sizeAdjust := calculateSizeAdjust(inProgressCompactions)
 	for level := 1; level < numLevels; level++ {
-		compensatedLevelSize := levelCompensatedSize(p.vers.Levels[level]) + sizeAdjust[level].compensated()
+		compensatedLevelSize := *compensatedSizeAnnotator.LevelAnnotation(p.vers.Levels[level]) + sizeAdjust[level].compensated()
 		scores[level].compensatedScore = float64(compensatedLevelSize) / float64(p.levelMaxBytes[level])
 		scores[level].uncompensatedScore = float64(p.vers.Levels[level].Size()+sizeAdjust[level].actual()) / float64(p.levelMaxBytes[level])
 	}
@@ -1393,109 +1363,56 @@ func (p *compactionPickerByScore) addScoresToPickedCompactionMetrics(
 	}
 }
 
-// elisionOnlyAnnotator implements the manifest.Annotator interface,
-// annotating B-Tree nodes with the *fileMetadata of a file meeting the
-// obsolete keys criteria for an elision-only compaction within the subtree.
-// If multiple files meet the criteria, it chooses whichever file has the
-// lowest LargestSeqNum. The lowest LargestSeqNum file will be the first
-// eligible for an elision-only compaction once snapshots less than or equal
-// to its LargestSeqNum are closed.
-type elisionOnlyAnnotator struct{}
-
-var _ manifest.Annotator = elisionOnlyAnnotator{}
-
-func (a elisionOnlyAnnotator) Zero(interface{}) interface{} {
-	return nil
-}
-
-func (a elisionOnlyAnnotator) Accumulate(f *fileMetadata, dst interface{}) (interface{}, bool) {
-	if f.IsCompacting() {
-		return dst, true
-	}
-	if !f.StatsValid() {
-		return dst, false
-	}
-	// Bottommost files are large and not worthwhile to compact just
-	// to remove a few tombstones. Consider a file ineligible if its
-	// own range deletions delete less than 10% of its data and its
-	// deletion tombstones make up less than 10% of its entries.
-	//
-	// TODO(jackson): This does not account for duplicate user keys
-	// which may be collapsed. Ideally, we would have 'obsolete keys'
-	// statistics that would include tombstones, the keys that are
-	// dropped by tombstones and duplicated user keys. See #847.
-	//
-	// Note that tables that contain exclusively range keys (i.e. no point keys,
-	// `NumEntries` and `RangeDeletionsBytesEstimate` are both zero) are excluded
-	// from elision-only compactions.
-	// TODO(travers): Consider an alternative heuristic for elision of range-keys.
-	if f.Stats.RangeDeletionsBytesEstimate*10 < f.Size &&
-		f.Stats.NumDeletions*10 <= f.Stats.NumEntries {
-		return dst, true
-	}
-	if dst == nil {
-		return f, true
-	} else if dstV := dst.(*fileMetadata); dstV.LargestSeqNum > f.LargestSeqNum {
-		return f, true
-	}
-	return dst, true
-}
-
-func (a elisionOnlyAnnotator) Merge(v interface{}, accum interface{}) interface{} {
-	if v == nil {
-		return accum
-	}
-	// If we haven't accumulated an eligible file yet, or f's LargestSeqNum is
-	// less than the accumulated file's, use f.
-	if accum == nil {
-		return v
-	}
-	f := v.(*fileMetadata)
-	accumV := accum.(*fileMetadata)
-	if accumV == nil || accumV.LargestSeqNum > f.LargestSeqNum {
-		return f
-	}
-	return accumV
-}
-
-// markedForCompactionAnnotator implements the manifest.Annotator interface,
-// annotating B-Tree nodes with the *fileMetadata of a file that is marked for
-// compaction within the subtree. If multiple files meet the criteria, it
-// chooses whichever file has the lowest LargestSeqNum.
-type markedForCompactionAnnotator struct{}
-
-var _ manifest.Annotator = markedForCompactionAnnotator{}
-
-func (a markedForCompactionAnnotator) Zero(interface{}) interface{} {
-	return nil
-}
-
-func (a markedForCompactionAnnotator) Accumulate(
-	f *fileMetadata, dst interface{},
-) (interface{}, bool) {
-	if !f.MarkedForCompaction {
-		// Not marked for compaction; return dst.
-		return dst, true
-	}
-	return markedMergeHelper(f, dst)
-}
-
-func (a markedForCompactionAnnotator) Merge(v interface{}, accum interface{}) interface{} {
-	if v == nil {
-		return accum
-	}
-	accum, _ = markedMergeHelper(v.(*fileMetadata), accum)
-	return accum
-}
-
-// REQUIRES: f is non-nil, and f.MarkedForCompaction=true.
-func markedMergeHelper(f *fileMetadata, dst interface{}) (interface{}, bool) {
-	if dst == nil {
-		return f, true
-	} else if dstV := dst.(*fileMetadata); dstV.LargestSeqNum > f.LargestSeqNum {
-		return f, true
-	}
-	return dst, true
+// elisionOnlyAnnotator is a manifest.Annotator that annotates B-Tree
+// nodes with the *fileMetadata of a file meeting the obsolete keys criteria
+// for an elision-only compaction within the subtree. If multiple files meet
+// the criteria, it chooses whichever file has the lowest LargestSeqNum. The
+// lowest LargestSeqNum file will be the first eligible for an elision-only
+// compaction once snapshots less than or equal to its LargestSeqNum are closed.
+var elisionOnlyAnnotator = &manifest.Annotator[fileMetadata]{
+	Aggregator: manifest.PickFileAggregator{
+		Filter: func(f *fileMetadata) (eligible bool, cacheOK bool) {
+			if f.IsCompacting() {
+				return false, true
+			}
+			if !f.StatsValid() {
+				return false, false
+			}
+			// Bottommost files are large and not worthwhile to compact just
+			// to remove a few tombstones. Consider a file eligible only if
+			// either its own range deletions delete at least 10% of its data or
+			// its deletion tombstones make at least 10% of its entries.
+			//
+			// TODO(jackson): This does not account for duplicate user keys
+			// which may be collapsed. Ideally, we would have 'obsolete keys'
+			// statistics that would include tombstones, the keys that are
+			// dropped by tombstones and duplicated user keys. See #847.
+			//
+			// Note that tables that contain exclusively range keys (i.e. no point keys,
+			// `NumEntries` and `RangeDeletionsBytesEstimate` are both zero) are excluded
+			// from elision-only compactions.
+			// TODO(travers): Consider an alternative heuristic for elision of range-keys.
+			return f.Stats.RangeDeletionsBytesEstimate*10 >= f.Size || f.Stats.NumDeletions*10 > f.Stats.NumEntries, true
+		},
+		Compare: func(f1 *fileMetadata, f2 *fileMetadata) bool {
+			return f1.LargestSeqNum < f2.LargestSeqNum
+		},
+	},
+}
+
+// markedForCompactionAnnotator is a manifest.Annotator that annotates B-Tree
+// nodes with the *fileMetadata of a file that is marked for compaction
+// within the subtree. If multiple files meet the criteria, it chooses
+// whichever file has the lowest LargestSeqNum.
+var markedForCompactionAnnotator = &manifest.Annotator[fileMetadata]{
+	Aggregator: manifest.PickFileAggregator{
+		Filter: func(f *fileMetadata) (eligible bool, cacheOK bool) {
+			return f.MarkedForCompaction, true
+		},
+		Compare: func(f1 *fileMetadata, f2 *fileMetadata) bool {
+			return f1.LargestSeqNum < f2.LargestSeqNum
+		},
+	},
 }
 
 // pickElisionOnlyCompaction looks for compactions of sstables in the
@@ -1506,11 +1423,10 @@ func (p *compactionPickerByScore) pickElisionOnlyCompaction(
 	if p.opts.private.disableElisionOnlyCompactions {
 		return nil
 	}
-	v := p.vers.Levels[numLevels-1].Annotation(elisionOnlyAnnotator{})
-	if v == nil {
+	candidate := elisionOnlyAnnotator.LevelAnnotation(p.vers.Levels[numLevels-1])
+	if candidate == nil {
 		return nil
 	}
-	candidate := v.(*fileMetadata)
 	if candidate.IsCompacting() || candidate.LargestSeqNum >= env.earliestSnapshotSeqNum {
 		return nil
 	}
@@ -1542,12 +1458,11 @@ func (p *compactionPickerByScore) pickElisionOnlyCompaction(
 // the input level.
 func (p *compactionPickerByScore) pickRewriteCompaction(env compactionEnv) (pc *pickedCompaction) {
 	for l := numLevels - 1; l >= 0; l-- {
-		v := p.vers.Levels[l].Annotation(markedForCompactionAnnotator{})
-		if v == nil {
+		candidate := markedForCompactionAnnotator.LevelAnnotation(p.vers.Levels[l])
+		if candidate == nil {
 			// Try the next level.
 			continue
 		}
-		candidate := v.(*fileMetadata)
 		if candidate.IsCompacting() {
 			// Try the next level.
 			continue
diff --git a/db.go b/db.go
index d879f688..4bb30aec 100644
--- a/db.go
+++ b/db.go
@@ -1995,8 +1995,8 @@ func (d *DB) Metrics() *Metrics {
 	metrics.private.optionsFileSize = d.optionsFileSize
 
 	// TODO(jackson): Consider making these metrics optional.
-	metrics.Keys.RangeKeySetsCount = countRangeKeySetFragments(vers)
-	metrics.Keys.TombstoneCount = countTombstones(vers)
+	metrics.Keys.RangeKeySetsCount = *rangeKeySetsAnnotator.MultiLevelAnnotation(vers.RangeKeyLevels[:])
+	metrics.Keys.TombstoneCount = *tombstonesAnnotator.MultiLevelAnnotation(vers.Levels[:])
 
 	d.mu.versions.logLock()
 	metrics.private.manifestFileSize = uint64(d.mu.versions.manifest.Size())
@@ -2014,12 +2014,12 @@ func (d *DB) Metrics() *Metrics {
 		metrics.Flush.NumInProgress = 1
 	}
 	for i := 0; i < numLevels; i++ {
-		metrics.Levels[i].Additional.ValueBlocksSize = valueBlocksSizeForLevel(vers, i)
-		unknown, snappy, none, zstd := compressionTypesForLevel(vers, i)
-		metrics.Table.CompressedCountUnknown += int64(unknown)
-		metrics.Table.CompressedCountSnappy += int64(snappy)
-		metrics.Table.CompressedCountZstd += int64(zstd)
-		metrics.Table.CompressedCountNone += int64(none)
+		metrics.Levels[i].Additional.ValueBlocksSize = *valueBlockSizeAnnotator.LevelAnnotation(vers.Levels[i])
+		compressionTypes := compressionTypeAnnotator.LevelAnnotation(vers.Levels[i])
+		metrics.Table.CompressedCountUnknown += int64(compressionTypes.unknown)
+		metrics.Table.CompressedCountSnappy += int64(compressionTypes.snappy)
+		metrics.Table.CompressedCountZstd += int64(compressionTypes.zstd)
+		metrics.Table.CompressedCountNone += int64(compressionTypes.none)
 	}
 
 	d.mu.Unlock()
diff --git a/error_iter.go b/error_iter.go
index 61911c9f..3c741f86 100644
--- a/error_iter.go
+++ b/error_iter.go
@@ -90,6 +90,7 @@ func (i *errorKeyspanIter) First() (*keyspan.Span, error)            { return ni
 func (i *errorKeyspanIter) Last() (*keyspan.Span, error)             { return nil, i.err }
 func (i *errorKeyspanIter) Next() (*keyspan.Span, error)             { return nil, i.err }
 func (i *errorKeyspanIter) Prev() (*keyspan.Span, error)             { return nil, i.err }
+func (i *errorKeyspanIter) SetContext(ctx context.Context)           {}
 func (i *errorKeyspanIter) Close()                                   {}
 func (*errorKeyspanIter) String() string                             { return "error" }
 func (*errorKeyspanIter) WrapChildren(wrap keyspan.WrapFn)           {}
diff --git a/flushable.go b/flushable.go
index bc2915c2..b06f7012 100644
--- a/flushable.go
+++ b/flushable.go
@@ -233,9 +233,9 @@ func (s *ingestedFlushable) newFlushIter(*IterOptions) internalIterator {
 }
 
 func (s *ingestedFlushable) constructRangeDelIter(
-	file *manifest.FileMetadata, _ keyspan.SpanIterOptions,
+	ctx context.Context, file *manifest.FileMetadata, _ keyspan.SpanIterOptions,
 ) (keyspan.FragmentIterator, error) {
-	iters, err := s.newIters(context.Background(), file, nil, internalIterOpts{}, iterRangeDeletions)
+	iters, err := s.newIters(ctx, file, nil, internalIterOpts{}, iterRangeDeletions)
 	if err != nil {
 		return nil, err
 	}
@@ -250,6 +250,7 @@ func (s *ingestedFlushable) constructRangeDelIter(
 // the point iterator in constructRangeDeIter is not tracked.
 func (s *ingestedFlushable) newRangeDelIter(_ *IterOptions) keyspan.FragmentIterator {
 	return keyspanimpl.NewLevelIter(
+		context.TODO(),
 		keyspan.SpanIterOptions{}, s.comparer.Compare,
 		s.constructRangeDelIter, s.slice.Iter(), manifest.Level(0),
 		manifest.KeyTypePoint,
@@ -263,6 +264,7 @@ func (s *ingestedFlushable) newRangeKeyIter(o *IterOptions) keyspan.FragmentIter
 	}
 
 	return keyspanimpl.NewLevelIter(
+		context.TODO(),
 		keyspan.SpanIterOptions{}, s.comparer.Compare, s.newRangeKeyIters,
 		s.slice.Iter(), manifest.Level(0), manifest.KeyTypeRange,
 	)
diff --git a/format_major_version.go b/format_major_version.go
index b7c501f7..90ef9f0b 100644
--- a/format_major_version.go
+++ b/format_major_version.go
@@ -517,7 +517,7 @@ func (d *DB) markFilesLocked(findFn findFilesFunc) error {
 		// annotations will be out of date. Clear the compaction-picking
 		// annotation, so that it's recomputed the next time the compaction
 		// picker looks for a file marked for compaction.
-		vers.Levels[l].InvalidateAnnotation(markedForCompactionAnnotator{})
+		markedForCompactionAnnotator.InvalidateLevelAnnotation(vers.Levels[l])
 	}
 
 	// The 'marked-for-compaction' bit is persisted in the MANIFEST file
diff --git a/internal/keyspan/assert_iter.go b/internal/keyspan/assert_iter.go
index 42f76bb9..ded00db9 100644
--- a/internal/keyspan/assert_iter.go
+++ b/internal/keyspan/assert_iter.go
@@ -5,6 +5,7 @@
 package keyspan
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cockroachdb/errors"
@@ -165,6 +166,11 @@ func (i *assertIter) Prev() (*Span, error) {
 	return span, err
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *assertIter) SetContext(ctx context.Context) {
+	i.iter.SetContext(ctx)
+}
+
 // Close implements FragmentIterator.
 func (i *assertIter) Close() {
 	i.iter.Close()
diff --git a/internal/keyspan/bounded.go b/internal/keyspan/bounded.go
index 580cb74e..379676b7 100644
--- a/internal/keyspan/bounded.go
+++ b/internal/keyspan/bounded.go
@@ -5,6 +5,8 @@
 package keyspan
 
 import (
+	"context"
+
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 )
@@ -209,6 +211,11 @@ func (i *BoundedIter) Close() {
 	i.iter.Close()
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *BoundedIter) SetContext(ctx context.Context) {
+	i.iter.SetContext(ctx)
+}
+
 // SetBounds modifies the FragmentIterator's bounds.
 func (i *BoundedIter) SetBounds(lower, upper []byte) {
 	i.lower, i.upper = lower, upper
diff --git a/internal/keyspan/defragment.go b/internal/keyspan/defragment.go
index eb9917d2..35dda16d 100644
--- a/internal/keyspan/defragment.go
+++ b/internal/keyspan/defragment.go
@@ -6,6 +6,7 @@ package keyspan
 
 import (
 	"bytes"
+	"context"
 
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
@@ -196,6 +197,11 @@ func (i *DefragmentingIter) Init(
 	}
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *DefragmentingIter) SetContext(ctx context.Context) {
+	i.iter.SetContext(ctx)
+}
+
 // Close closes the underlying iterators.
 func (i *DefragmentingIter) Close() {
 	i.iter.Close()
diff --git a/internal/keyspan/filter.go b/internal/keyspan/filter.go
index efba323d..8b09b2fd 100644
--- a/internal/keyspan/filter.go
+++ b/internal/keyspan/filter.go
@@ -5,6 +5,8 @@
 package keyspan
 
 import (
+	"context"
+
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 )
@@ -93,6 +95,11 @@ func (i *filteringIter) Prev() (*Span, error) {
 	return i.filter(s, -1)
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *filteringIter) SetContext(ctx context.Context) {
+	i.iter.SetContext(ctx)
+}
+
 // Close implements FragmentIterator.
 func (i *filteringIter) Close() {
 	i.iter.Close()
diff --git a/internal/keyspan/interleaving_iter.go b/internal/keyspan/interleaving_iter.go
index 5c02ed49..39a3759f 100644
--- a/internal/keyspan/interleaving_iter.go
+++ b/internal/keyspan/interleaving_iter.go
@@ -1140,6 +1140,7 @@ func (i *InterleavingIter) SetBounds(lower, upper []byte) {
 // SetContext implements (base.InternalIterator).SetContext.
 func (i *InterleavingIter) SetContext(ctx context.Context) {
 	i.pointIter.SetContext(ctx)
+	i.keyspanIter.SetContext(ctx)
 }
 
 // DebugTree is part of the InternalIterator interface.
diff --git a/internal/keyspan/iter.go b/internal/keyspan/iter.go
index 06ff261a..5d13e065 100644
--- a/internal/keyspan/iter.go
+++ b/internal/keyspan/iter.go
@@ -5,6 +5,8 @@
 package keyspan
 
 import (
+	"context"
+
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 )
@@ -69,6 +71,10 @@ type FragmentIterator interface {
 	// stack. Used only for debug logging.
 	WrapChildren(wrap WrapFn)
 
+	// SetContext replaces the context provided at iterator creation, or the last
+	// one provided by SetContext.
+	SetContext(ctx context.Context)
+
 	base.IteratorDebug
 }
 
@@ -211,6 +217,9 @@ func (i *Iter) Prev() (*Span, error) {
 	return &i.spans[i.index], nil
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *Iter) SetContext(ctx context.Context) {}
+
 // Close implements FragmentIterator.Close.
 func (i *Iter) Close() {}
 
diff --git a/internal/keyspan/keyspanimpl/level_iter.go b/internal/keyspan/keyspanimpl/level_iter.go
index e4b266e2..2fc5b836 100644
--- a/internal/keyspan/keyspanimpl/level_iter.go
+++ b/internal/keyspan/keyspanimpl/level_iter.go
@@ -5,6 +5,7 @@
 package keyspanimpl
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cockroachdb/pebble/internal/base"
@@ -17,7 +18,7 @@ import (
 // TableNewSpanIter creates a new iterator for range key spans for the given
 // file.
 type TableNewSpanIter func(
-	file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions,
+	ctx context.Context, file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions,
 ) (keyspan.FragmentIterator, error)
 
 // LevelIter provides a merged view of spans from sstables in an L1+ level or an
@@ -46,6 +47,8 @@ type LevelIter struct {
 	// newIter creates a range del iterator if keyType is KeyTypePoint or a range
 	// key iterator if keyType is KeyTypeRange.
 	newIter TableNewSpanIter
+	// ctx is passed to TableNewSpanIter.
+	ctx context.Context
 
 	// The options that were passed in.
 	tableOpts keyspan.SpanIterOptions
@@ -79,6 +82,7 @@ var _ keyspan.FragmentIterator = (*LevelIter)(nil)
 // newIter must create a range del iterator for the given file if keyType is
 // KeyTypePoint or a range key iterator if keyType is KeyTypeRange.
 func NewLevelIter(
+	ctx context.Context,
 	opts keyspan.SpanIterOptions,
 	cmp base.Compare,
 	newIter TableNewSpanIter,
@@ -87,7 +91,7 @@ func NewLevelIter(
 	keyType manifest.KeyType,
 ) *LevelIter {
 	l := &LevelIter{}
-	l.Init(opts, cmp, newIter, files, level, keyType)
+	l.Init(ctx, opts, cmp, newIter, files, level, keyType)
 	return l
 }
 
@@ -96,6 +100,7 @@ func NewLevelIter(
 // newIter must create a range del iterator for the given file if keyType is
 // KeyTypePoint or a range key iterator if keyType is KeyTypeRange.
 func (l *LevelIter) Init(
+	ctx context.Context,
 	opts keyspan.SpanIterOptions,
 	cmp base.Compare,
 	newIter TableNewSpanIter,
@@ -111,6 +116,7 @@ func (l *LevelIter) Init(
 		keyType:   keyType,
 		level:     level,
 		newIter:   newIter,
+		ctx:       ctx,
 		tableOpts: opts,
 		files:     files.Filter(keyType),
 	}
@@ -387,6 +393,14 @@ func (l *LevelIter) moveToPrevFile() (*keyspan.Span, error) {
 	}
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (l *LevelIter) SetContext(ctx context.Context) {
+	l.ctx = ctx
+	if l.lastIter != nil {
+		l.lastIter.SetContext(ctx)
+	}
+}
+
 // Close implements keyspan.FragmentIterator.
 func (l *LevelIter) Close() {
 	l.file = nil
@@ -442,7 +456,7 @@ func (l *LevelIter) setPosAtFile(f *manifest.FileMetadata) error {
 			l.lastIter = nil
 			l.lastIterFile = nil
 		}
-		iter, err := l.newIter(l.file, l.tableOpts)
+		iter, err := l.newIter(l.ctx, l.file, l.tableOpts)
 		if err != nil {
 			return err
 		}
diff --git a/internal/keyspan/keyspanimpl/merging_iter.go b/internal/keyspan/keyspanimpl/merging_iter.go
index 5cb07254..9fd87960 100644
--- a/internal/keyspan/keyspanimpl/merging_iter.go
+++ b/internal/keyspan/keyspanimpl/merging_iter.go
@@ -7,6 +7,7 @@ package keyspanimpl
 import (
 	"bytes"
 	"cmp"
+	"context"
 	"fmt"
 	"slices"
 
@@ -676,6 +677,13 @@ func (m *MergingIter) Prev() (*keyspan.Span, error) {
 	return m.findPrevFragmentSet()
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (m *MergingIter) SetContext(ctx context.Context) {
+	for i := range m.levels {
+		m.levels[i].iter.SetContext(ctx)
+	}
+}
+
 // Close closes the iterator, releasing all acquired resources.
 func (m *MergingIter) Close() {
 	for i := range m.levels {
diff --git a/internal/keyspan/logging_iter.go b/internal/keyspan/logging_iter.go
index a0a4b1ab..250d6691 100644
--- a/internal/keyspan/logging_iter.go
+++ b/internal/keyspan/logging_iter.go
@@ -5,6 +5,7 @@
 package keyspan
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cockroachdb/pebble/internal/base"
@@ -132,6 +133,11 @@ func (i *loggingIter) Prev() (*Span, error) {
 	return span, err
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *loggingIter) SetContext(ctx context.Context) {
+	i.iter.SetContext(ctx)
+}
+
 // Close implements FragmentIterator.
 func (i *loggingIter) Close() {
 	opEnd := i.opStartf("Close()")
diff --git a/internal/keyspan/truncate.go b/internal/keyspan/truncate.go
index 20ea12da..721d514f 100644
--- a/internal/keyspan/truncate.go
+++ b/internal/keyspan/truncate.go
@@ -5,6 +5,8 @@
 package keyspan
 
 import (
+	"context"
+
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
@@ -112,6 +114,11 @@ func (i *truncatingIter) Prev() (*Span, error) {
 	return span, err
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *truncatingIter) SetContext(ctx context.Context) {
+	i.iter.SetContext(ctx)
+}
+
 // Close implements FragmentIterator.
 func (i *truncatingIter) Close() {
 	i.iter.Close()
diff --git a/internal/manifest/annotator.go b/internal/manifest/annotator.go
new file mode 100644
index 00000000..cfbe2f4f
--- /dev/null
+++ b/internal/manifest/annotator.go
@@ -0,0 +1,253 @@
+// 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 manifest
+
+// The Annotator type defined below is used by other packages to lazily
+// compute a value over a B-Tree. Each node of the B-Tree stores one
+// `annotation` per annotator, containing the result of the computation over
+// the node's subtree.
+//
+// An annotation is marked as valid if it's current with the current subtree
+// state. Annotations are marked as invalid whenever a node will be mutated
+// (in mut).  Annotators may also return `false` from `Accumulate` to signal
+// that a computation for a file is not stable and may change in the future.
+// Annotations that include these unstable values are also marked as invalid
+// on the node, ensuring that future queries for the annotation will recompute
+// the value.
+
+// An Annotator defines a computation over a level's FileMetadata. If the
+// computation is stable and uses inputs that are fixed for the lifetime of
+// a FileMetadata, the LevelMetadata's internal data structures are annotated
+// with the intermediary computations. This allows the computation to be
+// computed incrementally as edits are applied to a level.
+type Annotator[T any] struct {
+	Aggregator AnnotationAggregator[T]
+}
+
+// An AnnotationAggregator defines how an annotation should be accumulated
+// from a single FileMetadata and merged with other annotated values.
+type AnnotationAggregator[T any] interface {
+	// Zero returns the zero value of an annotation. This value is returned
+	// when a LevelMetadata is empty. The dst argument, if non-nil, is an
+	// obsolete value previously returned by this Annotator and may be
+	// overwritten and reused to avoid a memory allocation.
+	Zero(dst *T) *T
+
+	// Accumulate computes the annotation for a single file in a level's
+	// metadata. It merges the file's value into dst and returns a bool flag
+	// indicating whether or not the value is stable and okay to cache as an
+	// annotation. If the file's value may change over the life of the file,
+	// the annotator must return false.
+	//
+	// Implementations may modify dst and return it to avoid an allocation.
+	Accumulate(f *FileMetadata, dst *T) (v *T, cacheOK bool)
+
+	// Merge combines two values src and dst, returning the result.
+	// Implementations may modify dst and return it to avoid an allocation.
+	Merge(src *T, dst *T) *T
+}
+
+type annotation struct {
+	// annotator is a pointer to the Annotator that computed this annotation.
+	// NB: This is untyped to allow AnnotationAggregator to use Go generics,
+	// since annotations are stored in a slice on each node and a single
+	// slice cannot contain elements with different type parameters.
+	annotator interface{}
+	// v is contains the annotation value, the output of either
+	// AnnotationAggregator.Accumulate or AnnotationAggregator.Merge.
+	// NB: This is untyped for the same reason as annotator above.
+	v interface{}
+	// valid indicates whether future reads of the annotation may use the
+	// value as-is. If false, v will be zeroed and recalculated.
+	valid bool
+}
+
+// findAnnotation finds this Annotator's annotation on a node, creating
+// one if it doesn't already exist.
+func (a *Annotator[T]) findAnnotation(n *node) *annotation {
+	for i := range n.annot {
+		if n.annot[i].annotator == a {
+			return &n.annot[i]
+		}
+	}
+
+	// This node has never been annotated by a. Create a new annotation.
+	n.annot = append(n.annot, annotation{
+		annotator: a,
+		v:         a.Aggregator.Zero(nil),
+	})
+	return &n.annot[len(n.annot)-1]
+}
+
+// nodeAnnotation computes this annotator's annotation of this node across all
+// files in the node's subtree. The second return value indicates whether the
+// annotation is stable and thus cacheable.
+func (a *Annotator[T]) nodeAnnotation(n *node) (v *T, cacheOK bool) {
+	annot := a.findAnnotation(n)
+	vtyped := annot.v.(*T)
+	// If the annotation is already marked as valid, we can return it without
+	// recomputing anything.
+	if annot.valid {
+		return vtyped, true
+	}
+
+	annot.v = a.Aggregator.Zero(vtyped)
+	annot.valid = true
+
+	for i := int16(0); i <= n.count; i++ {
+		if !n.leaf {
+			v, ok := a.nodeAnnotation(n.children[i])
+			annot.v = a.Aggregator.Merge(v, vtyped)
+			annot.valid = annot.valid && ok
+		}
+
+		if i < n.count {
+			v, ok := a.Aggregator.Accumulate(n.items[i], vtyped)
+			annot.v = v
+			annot.valid = annot.valid && ok
+		}
+	}
+
+	return annot.v.(*T), annot.valid
+}
+
+// InvalidateAnnotation removes any existing cached annotations from this
+// annotator from a node's subtree.
+func (a *Annotator[T]) invalidateNodeAnnotation(n *node) {
+	annot := a.findAnnotation(n)
+	annot.valid = false
+	if !n.leaf {
+		for i := int16(0); i <= n.count; i++ {
+			a.invalidateNodeAnnotation(n.children[i])
+		}
+	}
+}
+
+// LevelAnnotation calculates the annotation defined by this Annotator for all
+// files in the given LevelMetadata. A pointer to the Annotator is used as the
+// key for pre-calculated values, so the same Annotator must be used to avoid
+// duplicate computation. Annotation must not be called concurrently, and in
+// practice this is achieved by requiring callers to hold DB.mu.
+func (a *Annotator[T]) LevelAnnotation(lm LevelMetadata) *T {
+	if lm.Empty() {
+		return a.Aggregator.Zero(nil)
+	}
+
+	v, _ := a.nodeAnnotation(lm.tree.root)
+	return v
+}
+
+// LevelAnnotation calculates the annotation defined by this Annotator for all
+// files across the given levels. A pointer to the Annotator is used as the
+// key for pre-calculated values, so the same Annotator must be used to avoid
+// duplicate computation. Annotation must not be called concurrently, and in
+// practice this is achieved by requiring callers to hold DB.mu.
+func (a *Annotator[T]) MultiLevelAnnotation(lms []LevelMetadata) *T {
+	aggregated := a.Aggregator.Zero(nil)
+	for l := 0; l < len(lms); l++ {
+		if !lms[l].Empty() {
+			v := a.LevelAnnotation(lms[l])
+			aggregated = a.Aggregator.Merge(v, aggregated)
+		}
+	}
+	return aggregated
+}
+
+// InvalidateAnnotation clears any cached annotations defined by Annotator. A
+// pointer to the Annotator is used as the key for pre-calculated values, so
+// the same Annotator must be used to clear the appropriate cached annotation.
+// InvalidateAnnotation must not be called concurrently, and in practice this
+// is achieved by requiring callers to hold DB.mu.
+func (a *Annotator[T]) InvalidateLevelAnnotation(lm LevelMetadata) {
+	if lm.Empty() {
+		return
+	}
+	a.invalidateNodeAnnotation(lm.tree.root)
+}
+
+// sumAggregator defines an Aggregator which sums together a uint64 value
+// across files.
+type sumAggregator struct {
+	accumulateFunc func(f *FileMetadata) (v uint64, cacheOK bool)
+}
+
+func (sa sumAggregator) Zero(dst *uint64) *uint64 {
+	if dst == nil {
+		return new(uint64)
+	}
+	*dst = 0
+	return dst
+}
+
+func (sa sumAggregator) Accumulate(f *FileMetadata, dst *uint64) (v *uint64, cacheOK bool) {
+	accumulated, ok := sa.accumulateFunc(f)
+	*dst += accumulated
+	return dst, ok
+}
+
+func (sa sumAggregator) Merge(src *uint64, dst *uint64) *uint64 {
+	*dst += *src
+	return dst
+}
+
+// SumAnnotator takes a function that computes a uint64 value from a single
+// FileMetadata and returns an Annotator that sums together the values across
+// files.
+func SumAnnotator(accumulate func(f *FileMetadata) (v uint64, cacheOK bool)) *Annotator[uint64] {
+	return &Annotator[uint64]{
+		Aggregator: sumAggregator{
+			accumulateFunc: accumulate,
+		},
+	}
+}
+
+// PickFileAggregator implements the AnnotationAggregator interface. It defines
+// an aggregator that picks a single file from a set of eligible files.
+type PickFileAggregator struct {
+	// Filter takes a FileMetadata and returns whether it is eligible to be
+	// picked by this PickFileAggregator. The second return value indicates
+	// whether this eligibility is stable and thus cacheable.
+	Filter func(f *FileMetadata) (eligible bool, cacheOK bool)
+	// Compare compares two instances of FileMetadata and returns true if
+	// the first one should be picked over the second one. It may assume
+	// that both arguments are non-nil.
+	Compare func(f1 *FileMetadata, f2 *FileMetadata) bool
+}
+
+// Zero implements AnnotationAggregator.Zero, returning nil as the zero value.
+func (fa PickFileAggregator) Zero(dst *FileMetadata) *FileMetadata {
+	return nil
+}
+
+func (fa PickFileAggregator) mergePickedFiles(src *FileMetadata, dst *FileMetadata) *FileMetadata {
+	switch {
+	case src == nil:
+		return dst
+	case dst == nil:
+		return src
+	case fa.Compare(src, dst):
+		return src
+	default:
+		return dst
+	}
+}
+
+// Accumulate implements AnnotationAggregator.Accumulate, accumulating a single
+// file as long as it is eligible to be picked.
+func (fa PickFileAggregator) Accumulate(
+	f *FileMetadata, dst *FileMetadata,
+) (v *FileMetadata, cacheOK bool) {
+	eligible, ok := fa.Filter(f)
+	if eligible {
+		return fa.mergePickedFiles(f, dst), ok
+	}
+	return dst, ok
+}
+
+// Merge implements AnnotationAggregator.Merge by picking a single file based
+// on the output of PickFileAggregator.Compare.
+func (fa PickFileAggregator) Merge(src *FileMetadata, dst *FileMetadata) *FileMetadata {
+	return fa.mergePickedFiles(src, dst)
+}
diff --git a/internal/manifest/btree.go b/internal/manifest/btree.go
index b098a179..350200b6 100644
--- a/internal/manifest/btree.go
+++ b/internal/manifest/btree.go
@@ -16,45 +16,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/invariants"
 )
 
-// The Annotator type defined below is used by other packages to lazily
-// compute a value over a B-Tree. Each node of the B-Tree stores one
-// `annotation` per annotator, containing the result of the computation over
-// the node's subtree.
-//
-// An annotation is marked as valid if it's current with the current subtree
-// state. Annotations are marked as invalid whenever a node will be mutated
-// (in mut).  Annotators may also return `false` from `Accumulate` to signal
-// that a computation for a file is not stable and may change in the future.
-// Annotations that include these unstable values are also marked as invalid
-// on the node, ensuring that future queries for the annotation will recompute
-// the value.
-
-// An Annotator defines a computation over a level's FileMetadata. If the
-// computation is stable and uses inputs that are fixed for the lifetime of
-// a FileMetadata, the LevelMetadata's internal data structures are annotated
-// with the intermediary computations. This allows the computation to be
-// computed incrementally as edits are applied to a level.
-type Annotator interface {
-	// Zero returns the zero value of an annotation. This value is returned
-	// when a LevelMetadata is empty. The dst argument, if non-nil, is an
-	// obsolete value previously returned by this Annotator and may be
-	// overwritten and reused to avoid a memory allocation.
-	Zero(dst interface{}) (v interface{})
-
-	// Accumulate computes the annotation for a single file in a level's
-	// metadata. It merges the file's value into dst and returns a bool flag
-	// indicating whether or not the value is stable and okay to cache as an
-	// annotation. If the file's value may change over the life of the file,
-	// the annotator must return false.
-	//
-	// Implementations may modify dst and return it to avoid an allocation.
-	Accumulate(m *FileMetadata, dst interface{}) (v interface{}, cacheOK bool)
-
-	// Merge combines two values src and dst, returning the result.
-	// Implementations may modify dst and return it to avoid an allocation.
-	Merge(src interface{}, dst interface{}) interface{}
-}
-
 type btreeCmp func(*FileMetadata, *FileMetadata) int
 
 func btreeCmpSeqNum(a, b *FileMetadata) int {
@@ -91,16 +52,6 @@ const (
 	minItems = degree - 1
 )
 
-type annotation struct {
-	annotator Annotator
-	// v is an annotation value, the output of either
-	// annotator.Value or annotator.Merge.
-	v interface{}
-	// valid indicates whether future reads of the annotation may use v as-is.
-	// If false, v will be zeroed and recalculated.
-	valid bool
-}
-
 type leafNode struct {
 	ref   atomic.Int32
 	count int16
@@ -659,78 +610,6 @@ func (n *node) rebalanceOrMerge(i int) {
 	}
 }
 
-// InvalidateAnnotation removes any existing cached annotations for the provided
-// annotator from this node's subtree.
-func (n *node) InvalidateAnnotation(a Annotator) {
-	// Find this annotator's annotation on this node.
-	var annot *annotation
-	for i := range n.annot {
-		if n.annot[i].annotator == a {
-			annot = &n.annot[i]
-		}
-	}
-
-	if annot != nil && annot.valid {
-		annot.valid = false
-		annot.v = a.Zero(annot.v)
-	}
-	if !n.leaf {
-		for i := int16(0); i <= n.count; i++ {
-			n.children[i].InvalidateAnnotation(a)
-		}
-	}
-}
-
-// Annotation retrieves, computing if not already computed, the provided
-// annotator's annotation of this node. The second return value indicates
-// whether the future reads of this annotation may use the first return value
-// as-is. If false, the annotation is not stable and may change on a subsequent
-// computation.
-func (n *node) Annotation(a Annotator) (interface{}, bool) {
-	// Find this annotator's annotation on this node.
-	var annot *annotation
-	for i := range n.annot {
-		if n.annot[i].annotator == a {
-			annot = &n.annot[i]
-		}
-	}
-
-	// If it exists and is marked as valid, we can return it without
-	// recomputing anything.
-	if annot != nil && annot.valid {
-		return annot.v, true
-	}
-
-	if annot == nil {
-		// This is n's first time being annotated by a.
-		// Create a new zeroed annotation.
-		n.annot = append(n.annot, annotation{
-			annotator: a,
-			v:         a.Zero(nil),
-		})
-		annot = &n.annot[len(n.annot)-1]
-	} else {
-		// There's an existing annotation that must be recomputed.
-		// Zero its value.
-		annot.v = a.Zero(annot.v)
-	}
-
-	annot.valid = true
-	for i := int16(0); i <= n.count; i++ {
-		if !n.leaf {
-			v, ok := n.children[i].Annotation(a)
-			annot.v = a.Merge(v, annot.v)
-			annot.valid = annot.valid && ok
-		}
-		if i < n.count {
-			v, ok := a.Accumulate(n.items[i], annot.v)
-			annot.v = v
-			annot.valid = annot.valid && ok
-		}
-	}
-	return annot.v, annot.valid
-}
-
 func (n *node) verifyInvariants() {
 	recomputedSubtreeCount := int(n.count)
 	if !n.leaf {
diff --git a/internal/manifest/level_metadata.go b/internal/manifest/level_metadata.go
index c3008e80..7cba73e1 100644
--- a/internal/manifest/level_metadata.go
+++ b/internal/manifest/level_metadata.go
@@ -140,31 +140,6 @@ func (lm *LevelMetadata) Find(cmp base.Compare, m *FileMetadata) LevelSlice {
 	return LevelSlice{}
 }
 
-// Annotation lazily calculates and returns the annotation defined by
-// Annotator. The Annotator is used as the key for pre-calculated
-// values, so equal Annotators must be used to avoid duplicate computations
-// and cached annotations. Annotation must not be called concurrently, and in
-// practice this is achieved by requiring callers to hold DB.mu.
-func (lm *LevelMetadata) Annotation(annotator Annotator) interface{} {
-	if lm.Empty() {
-		return annotator.Zero(nil)
-	}
-	v, _ := lm.tree.root.Annotation(annotator)
-	return v
-}
-
-// InvalidateAnnotation clears any cached annotations defined by Annotator. The
-// Annotator is used as the key for pre-calculated values, so equal Annotators
-// must be used to clear the appropriate cached annotation. InvalidateAnnotation
-// must not be called concurrently, and in practice this is achieved by
-// requiring callers to hold DB.mu.
-func (lm *LevelMetadata) InvalidateAnnotation(annotator Annotator) {
-	if lm.Empty() {
-		return
-	}
-	lm.tree.root.InvalidateAnnotation(annotator)
-}
-
 // LevelFile holds a file's metadata along with its position
 // within a level of the LSM.
 type LevelFile struct {
diff --git a/open.go b/open.go
index c52e80d6..cb77a52b 100644
--- a/open.go
+++ b/open.go
@@ -406,7 +406,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		opts.TableCache, d.cacheID, d.objProvider, d.opts, tableCacheSize,
 		&sstable.CategoryStatsCollector{})
 	d.newIters = d.tableCache.newIters
-	d.tableNewRangeKeyIter = tableNewRangeKeyIter(context.TODO(), d.newIters)
+	d.tableNewRangeKeyIter = tableNewRangeKeyIter(d.newIters)
 
 	var previousOptionsFileNum base.DiskFileNum
 	var previousOptionsFilename string
diff --git a/range_keys.go b/range_keys.go
index 8a91dbbb..52640a41 100644
--- a/range_keys.go
+++ b/range_keys.go
@@ -92,6 +92,7 @@ func (i *Iterator) constructRangeKeyIter() {
 
 				li := i.rangeKey.iterConfig.NewLevelIter()
 				li.Init(
+					i.ctx,
 					i.opts.SpanIterOptions(),
 					i.cmp,
 					i.newIterRangeKey,
@@ -110,7 +111,7 @@ func (i *Iterator) constructRangeKeyIter() {
 			}
 			li := i.rangeKey.iterConfig.NewLevelIter()
 			spanIterOpts := i.opts.SpanIterOptions()
-			li.Init(spanIterOpts, i.cmp, i.newIterRangeKey, current.RangeKeyLevels[level].Iter(),
+			li.Init(i.ctx, spanIterOpts, i.cmp, i.newIterRangeKey, current.RangeKeyLevels[level].Iter(),
 				manifest.Level(level), manifest.KeyTypeRange)
 			i.rangeKey.iterConfig.AddLevel(li)
 		}
diff --git a/scan_internal.go b/scan_internal.go
index 7f7ca327..6081c172 100644
--- a/scan_internal.go
+++ b/scan_internal.go
@@ -908,8 +908,8 @@ func (i *scanInternalIterator) constructPointIter(
 			i.ctx, i.opts.IterOptions, i.comparer, i.newIters, files, level,
 			internalIterOpts{})
 		mlevels[mlevelsIndex].iter = li
-		rli.Init(keyspan.SpanIterOptions{RangeKeyFilters: i.opts.RangeKeyFilters},
-			i.comparer.Compare, tableNewRangeDelIter(i.ctx, i.newIters), files, level,
+		rli.Init(i.ctx, keyspan.SpanIterOptions{RangeKeyFilters: i.opts.RangeKeyFilters},
+			i.comparer.Compare, tableNewRangeDelIter(i.newIters), files, level,
 			manifest.KeyTypePoint)
 		rangeDelIters = append(rangeDelIters, rli)
 
@@ -1030,7 +1030,7 @@ func (i *scanInternalIterator) constructRangeKeyIter() error {
 	// around Key InternalKeyTrailer order.
 	iter := current.RangeKeyLevels[0].Iter()
 	for f := iter.Last(); f != nil; f = iter.Prev() {
-		spanIter, err := i.newIterRangeKey(f, i.opts.SpanIterOptions())
+		spanIter, err := i.newIterRangeKey(i.ctx, f, i.opts.SpanIterOptions())
 		if err != nil {
 			return err
 		}
@@ -1065,7 +1065,7 @@ func (i *scanInternalIterator) constructRangeKeyIter() error {
 			levSlice := manifest.NewLevelSliceKeySorted(i.db.cmp, nonRemoteFiles)
 			levIter = levSlice.Iter()
 		}
-		li.Init(spanIterOpts, i.comparer.Compare, i.newIterRangeKey, levIter,
+		li.Init(i.ctx, spanIterOpts, i.comparer.Compare, i.newIterRangeKey, levIter,
 			manifest.Level(level), manifest.KeyTypeRange)
 		i.rangeKey.iterConfig.AddLevel(li)
 	}
diff --git a/sstable/colblk/column.go b/sstable/colblk/column.go
index d69db595..9f28ac2c 100644
--- a/sstable/colblk/column.go
+++ b/sstable/colblk/column.go
@@ -72,8 +72,7 @@ type ColumnWriter interface {
 	DataType(col int) DataType
 	// Finish serializes the column at the specified index, writing the column's
 	// data to buf at offset, and returning the offset at which the next column
-	// should be encoded. Finish also returns a column descriptor describing the
-	// encoding of the column, which will be serialized within the block header.
+	// should be encoded.
 	//
 	// The supplied buf must have enough space at the provided offset to fit the
 	// column. The caller may use Size() to calculate the exact size required.
diff --git a/sstable/colblk/prefix_bytes.go b/sstable/colblk/prefix_bytes.go
index 7469cafa..559ee96f 100644
--- a/sstable/colblk/prefix_bytes.go
+++ b/sstable/colblk/prefix_bytes.go
@@ -26,9 +26,15 @@ import (
 // distinguished only by different timestamp suffixes. With columnar blocks
 // enabling the timestamp to be placed in a separate column, the multiple
 // version problem becomes one of efficiently handling exact duplicate keys.
-// PrefixBytes builds off of the RawBytes encoding, introducing n/bundleSize+1
-// additional slices for encoding n/bundleSize bundle prefixes and 1 block-level
-// shared prefix for the column.
+// PrefixBytes builds off of the RawBytes encoding, introducing additional
+// slices for encoding (n+bundleSize-1)/bundleSize bundle prefixes and 1
+// block-level shared prefix for the column.
+//
+// Unlike the original prefix compression performed by rowblk (inherited from
+// LevelDB and RocksDB), PrefixBytes does not perform all prefix compression
+// relative to the previous key. Rather it performs prefix compression relative
+// to the first key of a key's bundle. This can result in less compression, but
+// simplifies reverse iteration and allows iteration to be largely stateless.
 //
 // To understand the PrefixBytes layout, we'll work through an example using
 // these 15 keys:
@@ -87,17 +93,25 @@ import (
 //	    18 |    13 |         36 | ........
 //	    19 |    14 |         36 | ........
 //
-// The offset column in the table points to the start and end index within the
-// RawBytes data array for each of the 20 slices defined above (the 15 key
-// suffixes + 4 bundle key prefixes + block key prefix). Offset[0] is the length
-// of the first slice which is always anchored at data[0]. The data columns
-// display the portion of the data array the slice covers. For row slices, an
-// empty suffix column indicates that the slice is identical to the slice at the
-// previous index which is indicated by the slice's offset being equal to the
-// previous slice's offset. Due to the lexicographic sorting, the key at row i
-// can't be a prefix of the key at row i-1 or it would have sorted before the
-// key at row i-1. And if the key differs then only the differing bytes will be
-// part of the suffix and not contained in the bundle prefix.
+// The 'end offset' column in the table encodes the exclusive offset within the
+// string data section where each of the slices end. Each slice starts at the
+// previous slice's end offset. The first slice (the block prefix)'s start
+// offset is implicitly zero. Note that this differs from the plain RawBytes
+// encoding which always stores a zero offset at the beginning of the offsets
+// array to avoid special-casing the first slice. The block prefix already
+// requires special-casing, so materializing the zero start offset is not
+// needed.
+//
+// The table above defines 20 slices: the 1 block key prefix, the 4 bundle key
+// prefixes and the 15 key suffixes. Offset[0] is the length of the first slice
+// which is always anchored at data[0]. The data columns display the portion of
+// the data array the slice covers. For row slices, an empty suffix column
+// indicates that the slice is identical to the slice at the previous index
+// which is indicated by the slice's offset being equal to the previous slice's
+// offset. Due to the lexicographic sorting, the key at row i can't be a prefix
+// of the key at row i-1 or it would have sorted before the key at row i-1. And
+// if the key differs then only the differing bytes will be part of the suffix
+// and not contained in the bundle prefix.
 //
 // The end result of this encoding is that we can store the 119 bytes of the 15
 // keys plus their start and end offsets (which would naively consume 15*4=60
@@ -117,7 +131,14 @@ import (
 //	|                            RawBytes                              |
 //	|                                                                  |
 //	| A modified RawBytes encoding is used to store the data slices. A |
-//	| PrefixBytes column storing n keys will encode 2+n+n/bundleSize   |
+//	| PrefixBytes column storing n keys will encode                    |
+//	|                                                                  |
+//	|                        1 block prefix                            |
+//	|                               +                                  |
+//	|          (n + bundleSize-1)/bundleSize bundle prefixes           |
+//	|                               +                                  |
+//	|                         n row suffixes                           |
+//	|                                                                  |
 //	| slices. Unlike the RawBytes encoding, the first offset encoded   |
 //	| is not guaranteed to be zero. In the PrefixBytes encoding, the   |
 //	| first offset encodes the length of the column-wide prefix. The   |
@@ -148,17 +169,18 @@ import (
 // # Reads
 //
 // This encoding provides O(1) access to any row by calculating the bundle for
-// the row (5*(row/4)), then the row's index within the bundle (1+(row%4)). If
-// the slice's offset equals the previous slice's offset then we step backward
-// until we find a non-empty slice or the start of the bundle (a variable number
-// of steps, but bounded by the bundle size).
+// the row (see bundleOffsetIndexForRow), then the per-row's suffix (see
+// rowSuffixIndex). If the per-row suffix's end offset equals the previous
+// offset, then the row is a duplicate key and we need to step backward until we
+// find a non-empty slice or the start of the bundle (a variable number of
+// steps, but bounded by the bundle size).
 //
 // Forward iteration can easily reuse the previous row's key with a check on
-// whether the row's slice is empty. Reverse iteration can reuse the next row's
-// key by looking at the next row's offset to determine whether we are in the
-// middle of a run of equal keys or at an edge. When reverse iteration steps
-// over an edge it has to continue backward until a non-empty slice is found
-// (just as in absolute positioning).
+// whether the row's slice is empty. Reverse iteration within a run of equal
+// keys can reuse the next row's key. When reverse iteration steps backward from
+// a non-empty slice onto an empty slice, it must continue backward until a
+// non-empty slice is found (just as in absolute positioning) to discover the
+// row suffix that is duplicated.
 //
 // The Seek{GE,LT} routines first binary search on the first key of each bundle
 // which can be retrieved without data movement because the bundle prefix is
diff --git a/sstable/reader_iter_single_lvl.go b/sstable/reader_iter_single_lvl.go
index b32293c6..4707bad6 100644
--- a/sstable/reader_iter_single_lvl.go
+++ b/sstable/reader_iter_single_lvl.go
@@ -1447,7 +1447,15 @@ func (i *singleLevelIterator) skipBackward() *base.InternalKV {
 		}
 		kv := i.data.Last()
 		if kv == nil {
-			return nil
+			// The block iter could have hid some obsolete points, so it isn't
+			// safe to assume that there are no keys if we keep skipping backwards.
+			// Check the previous block, but check the lower bound before doing
+			// that.
+			if i.lower != nil && i.cmp(indexKey.K.UserKey, i.lower) < 0 {
+				i.exhaustedBounds = -1
+				return nil
+			}
+			continue
 		}
 		if i.blockLower != nil && i.cmp(kv.K.UserKey, i.blockLower) < 0 {
 			i.exhaustedBounds = -1
diff --git a/sstable/rowblk/rowblk_fragment_iter.go b/sstable/rowblk/rowblk_fragment_iter.go
index b4648b5e..2504e74e 100644
--- a/sstable/rowblk/rowblk_fragment_iter.go
+++ b/sstable/rowblk/rowblk_fragment_iter.go
@@ -6,6 +6,7 @@ package rowblk
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"os"
 	"sync"
@@ -258,6 +259,9 @@ func (i *fragmentIter) gatherBackward(kv *base.InternalKV) (*keyspan.Span, error
 	return &i.span, nil
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *fragmentIter) SetContext(ctx context.Context) {}
+
 // Close implements (keyspan.FragmentIterator).Close.
 func (i *fragmentIter) Close() {
 	i.blockIter.Close()
diff --git a/table_cache.go b/table_cache.go
index aee22ac3..dba3c4b5 100644
--- a/table_cache.go
+++ b/table_cache.go
@@ -62,10 +62,8 @@ type tableNewIters func(
 
 // tableNewRangeDelIter takes a tableNewIters and returns a TableNewSpanIter
 // for the rangedel iterator returned by tableNewIters.
-func tableNewRangeDelIter(
-	ctx context.Context, newIters tableNewIters,
-) keyspanimpl.TableNewSpanIter {
-	return func(file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
+func tableNewRangeDelIter(newIters tableNewIters) keyspanimpl.TableNewSpanIter {
+	return func(ctx context.Context, file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
 		iters, err := newIters(ctx, file, nil, internalIterOpts{}, iterRangeDeletions)
 		if err != nil {
 			return nil, err
@@ -76,10 +74,8 @@ func tableNewRangeDelIter(
 
 // tableNewRangeKeyIter takes a tableNewIters and returns a TableNewSpanIter
 // for the range key iterator returned by tableNewIters.
-func tableNewRangeKeyIter(
-	ctx context.Context, newIters tableNewIters,
-) keyspanimpl.TableNewSpanIter {
-	return func(file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
+func tableNewRangeKeyIter(newIters tableNewIters) keyspanimpl.TableNewSpanIter {
+	return func(ctx context.Context, file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
 		iters, err := newIters(ctx, file, nil, internalIterOpts{}, iterRangeKeys)
 		if err != nil {
 			return nil, err
diff --git a/table_stats.go b/table_stats.go
index bc1af0bd..5eaf5d31 100644
--- a/table_stats.go
+++ b/table_stats.go
@@ -990,206 +990,76 @@ func newCombinedDeletionKeyspanIter(
 	return mIter, nil
 }
 
-// rangeKeySetsAnnotator implements manifest.Annotator, annotating B-Tree nodes
-// with the sum of the files' counts of range key fragments. Its annotation type
-// is a *uint64. The count of range key sets may change once a table's stats are
+// rangeKeySetsAnnotator is a manifest.Annotator that annotates B-Tree nodes
+// with the sum of the files' counts of range key fragments. The count of range
+// key sets may change once a table's stats are loaded asynchronously, so its
+// values are marked as cacheable only if a file's stats have been loaded.
+var rangeKeySetsAnnotator = manifest.SumAnnotator(func(f *manifest.FileMetadata) (uint64, bool) {
+	return f.Stats.NumRangeKeySets, f.StatsValid()
+})
+
+// tombstonesAnnotator is a manifest.Annotator that annotates B-Tree nodes
+// with the sum of the files' counts of tombstones (DEL, SINGLEDEL and RANGEDEL
+// keys). The count of tombstones may change once a table's stats are loaded
+// asynchronously, so its values are marked as cacheable only if a file's stats
+// have been loaded.
+var tombstonesAnnotator = manifest.SumAnnotator(func(f *manifest.FileMetadata) (uint64, bool) {
+	return f.Stats.NumDeletions, f.StatsValid()
+})
+
+// valueBlocksSizeAnnotator is a manifest.Annotator that annotates B-Tree
+// nodes with the sum of the files' Properties.ValueBlocksSize. The value block
+// size may change once a table's stats are loaded asynchronously, so its
+// values are marked as cacheable only if a file's stats have been loaded.
+var valueBlockSizeAnnotator = manifest.SumAnnotator(func(f *fileMetadata) (uint64, bool) {
+	return f.Stats.ValueBlocksSize, f.StatsValid()
+})
+
+// compressionTypeAnnotator is a manifest.Annotator that annotates B-tree
+// nodes with the compression type of the file. Its annotation type is
+// compressionTypes. The compression type may change once a table's stats are
 // loaded asynchronously, so its values are marked as cacheable only if a file's
 // stats have been loaded.
-type rangeKeySetsAnnotator struct{}
-
-var _ manifest.Annotator = rangeKeySetsAnnotator{}
-
-func (a rangeKeySetsAnnotator) Zero(dst interface{}) interface{} {
-	if dst == nil {
-		return new(uint64)
-	}
-	v := dst.(*uint64)
-	*v = 0
-	return v
-}
-
-func (a rangeKeySetsAnnotator) Accumulate(
-	f *fileMetadata, dst interface{},
-) (v interface{}, cacheOK bool) {
-	vptr := dst.(*uint64)
-	*vptr = *vptr + f.Stats.NumRangeKeySets
-	return vptr, f.StatsValid()
-}
-
-func (a rangeKeySetsAnnotator) Merge(src interface{}, dst interface{}) interface{} {
-	srcV := src.(*uint64)
-	dstV := dst.(*uint64)
-	*dstV = *dstV + *srcV
-	return dstV
-}
-
-// countRangeKeySetFragments counts the number of RANGEKEYSET keys across all
-// files of the LSM. It only counts keys in files for which table stats have
-// been loaded. It uses a b-tree annotator to cache intermediate values between
-// calculations when possible.
-func countRangeKeySetFragments(v *version) (count uint64) {
-	for l := 0; l < numLevels; l++ {
-		if v.RangeKeyLevels[l].Empty() {
-			continue
-		}
-		count += *v.RangeKeyLevels[l].Annotation(rangeKeySetsAnnotator{}).(*uint64)
-	}
-	return count
-}
-
-// tombstonesAnnotator implements manifest.Annotator, annotating B-Tree nodes
-// with the sum of the files' counts of tombstones (DEL, SINGLEDEL and RANGEDELk
-// eys). Its annotation type is a *uint64. The count of tombstones may change
-// once a table's stats are loaded asynchronously, so its values are marked as
-// cacheable only if a file's stats have been loaded.
-type tombstonesAnnotator struct{}
-
-var _ manifest.Annotator = tombstonesAnnotator{}
-
-func (a tombstonesAnnotator) Zero(dst interface{}) interface{} {
-	if dst == nil {
-		return new(uint64)
-	}
-	v := dst.(*uint64)
-	*v = 0
-	return v
-}
-
-func (a tombstonesAnnotator) Accumulate(
-	f *fileMetadata, dst interface{},
-) (v interface{}, cacheOK bool) {
-	vptr := dst.(*uint64)
-	*vptr = *vptr + f.Stats.NumDeletions
-	return vptr, f.StatsValid()
-}
-
-func (a tombstonesAnnotator) Merge(src interface{}, dst interface{}) interface{} {
-	srcV := src.(*uint64)
-	dstV := dst.(*uint64)
-	*dstV = *dstV + *srcV
-	return dstV
-}
-
-// countTombstones counts the number of tombstone (DEL, SINGLEDEL and RANGEDEL)
-// internal keys across all files of the LSM. It only counts keys in files for
-// which table stats have been loaded. It uses a b-tree annotator to cache
-// intermediate values between calculations when possible.
-func countTombstones(v *version) (count uint64) {
-	for l := 0; l < numLevels; l++ {
-		if v.Levels[l].Empty() {
-			continue
-		}
-		count += *v.Levels[l].Annotation(tombstonesAnnotator{}).(*uint64)
-	}
-	return count
-}
-
-// valueBlocksSizeAnnotator implements manifest.Annotator, annotating B-Tree
-// nodes with the sum of the files' Properties.ValueBlocksSize. Its annotation
-// type is a *uint64. The value block size may change once a table's stats are
-// loaded asynchronously, so its values are marked as cacheable only if a
-// file's stats have been loaded.
-type valueBlocksSizeAnnotator struct{}
-
-var _ manifest.Annotator = valueBlocksSizeAnnotator{}
-
-func (a valueBlocksSizeAnnotator) Zero(dst interface{}) interface{} {
-	if dst == nil {
-		return new(uint64)
-	}
-	v := dst.(*uint64)
-	*v = 0
-	return v
-}
-
-func (a valueBlocksSizeAnnotator) Accumulate(
-	f *fileMetadata, dst interface{},
-) (v interface{}, cacheOK bool) {
-	vptr := dst.(*uint64)
-	*vptr = *vptr + f.Stats.ValueBlocksSize
-	return vptr, f.StatsValid()
-}
-
-func (a valueBlocksSizeAnnotator) Merge(src interface{}, dst interface{}) interface{} {
-	srcV := src.(*uint64)
-	dstV := dst.(*uint64)
-	*dstV = *dstV + *srcV
-	return dstV
-}
-
-// valueBlocksSizeForLevel returns the Properties.ValueBlocksSize across all
-// files for a level of the LSM. It only includes the size for files for which
-// table stats have been loaded. It uses a b-tree annotator to cache
-// intermediate values between calculations when possible. It must not be
-// called concurrently.
-//
-// REQUIRES: 0 <= level <= numLevels.
-func valueBlocksSizeForLevel(v *version, level int) (count uint64) {
-	if v.Levels[level].Empty() {
-		return 0
-	}
-	return *v.Levels[level].Annotation(valueBlocksSizeAnnotator{}).(*uint64)
+var compressionTypeAnnotator = manifest.Annotator[compressionTypes]{
+	Aggregator: compressionTypeAggregator{},
 }
 
-// compressionTypeAnnotator implements manifest.Annotator, annotating B-tree
-// nodes with the compression type of the file. Its annotation type is a
-// *compressionTypes. The compression type may change once a table's stats are
-// loaded asynchronously, so its values are marked as cacheable only if a file's
-// stats have been loaded.
-type compressionTypeAnnotator struct{}
+type compressionTypeAggregator struct{}
 
 type compressionTypes struct {
 	snappy, zstd, none, unknown uint64
 }
 
-var _ manifest.Annotator = compressionTypeAnnotator{}
-
-func (a compressionTypeAnnotator) Zero(dst interface{}) interface{} {
+func (a compressionTypeAggregator) Zero(dst *compressionTypes) *compressionTypes {
 	if dst == nil {
 		return new(compressionTypes)
 	}
-	v := dst.(*compressionTypes)
-	*v = compressionTypes{}
-	return v
+	*dst = compressionTypes{}
+	return dst
 }
 
-func (a compressionTypeAnnotator) Accumulate(
-	f *fileMetadata, dst interface{},
-) (v interface{}, cacheOK bool) {
-	vptr := dst.(*compressionTypes)
+func (a compressionTypeAggregator) Accumulate(
+	f *fileMetadata, dst *compressionTypes,
+) (v *compressionTypes, cacheOK bool) {
 	switch f.Stats.CompressionType {
 	case sstable.SnappyCompression:
-		vptr.snappy++
+		dst.snappy++
 	case sstable.ZstdCompression:
-		vptr.zstd++
+		dst.zstd++
 	case sstable.NoCompression:
-		vptr.none++
+		dst.none++
 	default:
-		vptr.unknown++
+		dst.unknown++
 	}
-	return vptr, f.StatsValid()
+	return dst, f.StatsValid()
 }
 
-func (a compressionTypeAnnotator) Merge(src interface{}, dst interface{}) interface{} {
-	srcV := src.(*compressionTypes)
-	dstV := dst.(*compressionTypes)
-	dstV.snappy = dstV.snappy + srcV.snappy
-	dstV.zstd = dstV.zstd + srcV.zstd
-	dstV.none = dstV.none + srcV.none
-	dstV.unknown = dstV.unknown + srcV.unknown
-	return dstV
-}
-
-// compressionTypesForLevel returns the count of sstables by compression type
-// used for a level in the LSM. Sstables with compression type snappy or zstd
-// are returned, while others are ignored.
-func compressionTypesForLevel(v *version, level int) (unknown, snappy, none, zstd uint64) {
-	if v.Levels[level].Empty() {
-		return
-	}
-	compression := v.Levels[level].Annotation(compressionTypeAnnotator{}).(*compressionTypes)
-	if compression == nil {
-		return
-	}
-	return compression.unknown, compression.snappy, compression.none, compression.zstd
+func (a compressionTypeAggregator) Merge(
+	src *compressionTypes, dst *compressionTypes,
+) *compressionTypes {
+	dst.snappy += src.snappy
+	dst.zstd += src.zstd
+	dst.none += src.none
+	dst.unknown += src.unknown
+	return dst
 }


Test patch:
diff --git a/compaction_picker_test.go b/compaction_picker_test.go
index d305b167..bd2edb32 100644
--- a/compaction_picker_test.go
+++ b/compaction_picker_test.go
@@ -563,7 +563,7 @@ func TestCompactionPickerL0(t *testing.T) {
 					}
 					f.MarkedForCompaction = true
 					picker.vers.Stats.MarkedForCompaction++
-					picker.vers.Levels[l].InvalidateAnnotation(markedForCompactionAnnotator{})
+					markedForCompactionAnnotator.InvalidateLevelAnnotation(picker.vers.Levels[l])
 					return fmt.Sprintf("marked L%d.%s", l, f.FileNum)
 				}
 			}
diff --git a/compaction_test.go b/compaction_test.go
index 980f2279..990e9ea6 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -2504,7 +2504,7 @@ func TestMarkedForCompaction(t *testing.T) {
 					}
 					f.MarkedForCompaction = true
 					vers.Stats.MarkedForCompaction++
-					vers.Levels[l].InvalidateAnnotation(markedForCompactionAnnotator{})
+					markedForCompactionAnnotator.InvalidateLevelAnnotation(vers.Levels[l])
 					return fmt.Sprintf("marked L%d.%s", l, f.FileNum)
 				}
 			}
diff --git a/db_test.go b/db_test.go
index 69217ec1..5fd308b3 100644
--- a/db_test.go
+++ b/db_test.go
@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"io"
 	"path/filepath"
+	"regexp"
 	"runtime"
 	"slices"
 	"strconv"
@@ -24,6 +25,8 @@ import (
 	"github.com/cockroachdb/fifo"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
@@ -1405,8 +1408,11 @@ func (t *testTracer) Eventf(ctx context.Context, format string, args ...interfac
 	if t.enabledOnlyForNonBackgroundContext && ctx == context.Background() {
 		return
 	}
-	fmt.Fprintf(&t.buf, format, args...)
-	fmt.Fprint(&t.buf, "\n")
+	str := fmt.Sprintf(format, args...)
+	// Redact known strings that depend on source code line numbers.
+	str = regexp.MustCompile(`\(fileNum=[^)]+\)$`).ReplaceAllString(str, "(<redacted>)")
+	t.buf.WriteString(str)
+	t.buf.WriteString("\n")
 }
 
 func (t *testTracer) IsTracingEnabled(ctx context.Context) bool {
@@ -1418,68 +1424,76 @@ func (t *testTracer) IsTracingEnabled(ctx context.Context) bool {
 
 func TestTracing(t *testing.T) {
 	defer sstable.DeterministicReadBlockDurationForTesting()()
+
 	var tracer testTracer
-	c := NewCache(0)
-	defer c.Unref()
-	d, err := Open("", &Options{
-		FS:              vfs.NewMem(),
-		Cache:           c,
-		LoggerAndTracer: &tracer,
-	})
-	require.NoError(t, err)
+	buf := &tracer.buf
+	var db *DB
 	defer func() {
-		require.NoError(t, d.Close())
+		if db != nil {
+			db.Close()
+		}
 	}()
-
-	// Create a sstable.
-	require.NoError(t, d.Set([]byte("hello"), nil, nil))
-	require.NoError(t, d.Flush())
-	_, closer, err := d.Get([]byte("hello"))
-	require.NoError(t, err)
-	closer.Close()
-	readerInitRegexp := `reading footer of 53 bytes took 5ms
-reading block of 37 bytes took 5ms \([^)]*\)
-reading block of 419 bytes took 5ms \([^)]*\)
-`
-	iterTraceRegexp :=
-		`reading block of 27 bytes took 5ms \([^)]*\)
-reading block of 29 bytes took 5ms \([^)]*\)
-`
-
-	require.Regexp(t, readerInitRegexp+iterTraceRegexp, tracer.buf.String())
-
-	// Get again, but since it currently uses context.Background(), no trace
-	// output is produced.
-	tracer.buf.Reset()
-	tracer.enabledOnlyForNonBackgroundContext = true
-	_, closer, err = d.Get([]byte("hello"))
-	require.NoError(t, err)
-	closer.Close()
-	require.Equal(t, "", tracer.buf.String())
-
+	cache := NewCache(0)
+	defer cache.Unref()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	iter, _ := d.NewIterWithContext(ctx, nil)
-	iter.SeekGE([]byte("hello"))
-	iter.Close()
-	require.Regexp(t, iterTraceRegexp, tracer.buf.String())
-
-	tracer.buf.Reset()
-	snap := d.NewSnapshot()
-	iter, _ = snap.NewIterWithContext(ctx, nil)
-	iter.SeekGE([]byte("hello"))
-	iter.Close()
-	require.Regexp(t, iterTraceRegexp, tracer.buf.String())
-	snap.Close()
-
-	tracer.buf.Reset()
-	b := d.NewIndexedBatch()
-	iter, err = b.NewIterWithContext(ctx, nil)
-	require.NoError(t, err)
-	iter.SeekGE([]byte("hello"))
-	iter.Close()
-	require.Regexp(t, iterTraceRegexp, tracer.buf.String())
-	b.Close()
+	datadriven.RunTest(t, "testdata/tracing", func(t *testing.T, td *datadriven.TestData) string {
+		buf.Reset()
+		tracer.enabledOnlyForNonBackgroundContext = td.HasArg("disable-background-tracing")
+		ctx := ctx
+		switch td.Cmd {
+		case "build":
+			if db != nil {
+				db.Close()
+			}
+			db = testutils.CheckErr(Open("", &Options{
+				FS:              vfs.NewMem(),
+				Comparer:        testkeys.Comparer,
+				Cache:           cache,
+				LoggerAndTracer: &tracer,
+			}))
+
+			b := db.NewBatch()
+			require.NoError(t, runBatchDefineCmd(td, b))
+			require.NoError(t, b.Commit(nil))
+			require.NoError(t, db.Flush())
+			return ""
+
+		case "get":
+			for _, key := range strings.Split(td.Input, "\n") {
+				v, closer, err := db.Get([]byte(key))
+				require.NoError(t, err)
+				fmt.Fprintf(buf, "%s:%s\n", key, v)
+				closer.Close()
+			}
+
+		case "iter":
+			iter, _ := db.NewIterWithContext(ctx, &IterOptions{
+				KeyTypes: IterKeyTypePointsAndRanges,
+			})
+			buf.WriteString(runIterCmd(td, iter, true))
+
+		case "snapshot-iter":
+			snap := db.NewSnapshot()
+			defer snap.Close()
+			iter, _ := snap.NewIterWithContext(ctx, &IterOptions{
+				KeyTypes: IterKeyTypePointsAndRanges,
+			})
+			buf.WriteString(runIterCmd(td, iter, true))
+
+		case "indexed-batch-iter":
+			b := db.NewIndexedBatch()
+			defer b.Close()
+			iter, _ := b.NewIterWithContext(ctx, &IterOptions{
+				KeyTypes: IterKeyTypePointsAndRanges,
+			})
+			buf.WriteString(runIterCmd(td, iter, true))
+
+		default:
+			td.Fatalf(t, "unknown command: %s", td.Cmd)
+		}
+		return buf.String()
+	})
 }
 
 func TestMemtableIngestInversion(t *testing.T) {
diff --git a/internal/keyspan/keyspanimpl/level_iter_test.go b/internal/keyspan/keyspanimpl/level_iter_test.go
index 955fe5c2..3ef71a12 100644
--- a/internal/keyspan/keyspanimpl/level_iter_test.go
+++ b/internal/keyspan/keyspanimpl/level_iter_test.go
@@ -5,6 +5,7 @@
 package keyspanimpl
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"testing"
@@ -301,7 +302,7 @@ func TestLevelIterEquivalence(t *testing.T) {
 				metas = append(metas, meta)
 			}
 
-			tableNewIters := func(file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
+			tableNewIters := func(ctx context.Context, file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
 				return keyspan.NewIter(base.DefaultComparer.Compare, tc.levels[j][file.FileNum-1]), nil
 			}
 			// Add all the fileMetadatas to L6.
@@ -314,6 +315,7 @@ func TestLevelIterEquivalence(t *testing.T) {
 			v, err := b.Apply(nil, base.DefaultComparer, 0, 0)
 			require.NoError(t, err)
 			levelIter.Init(
+				context.Background(),
 				keyspan.SpanIterOptions{}, base.DefaultComparer.Compare, tableNewIters,
 				v.Levels[6].Iter(), 0, manifest.KeyTypeRange,
 			)
@@ -414,7 +416,7 @@ func TestLevelIter(t *testing.T) {
 					keyType = manifest.KeyTypePoint
 				}
 			}
-			tableNewIters := func(file *manifest.FileMetadata, _ keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
+			tableNewIters := func(ctx context.Context, file *manifest.FileMetadata, _ keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
 				f := files[file.FileNum-1]
 				if keyType == manifest.KeyTypePoint {
 					return keyspan.NewIter(cmp, f.rangeDels), nil
@@ -426,7 +428,7 @@ func TestLevelIter(t *testing.T) {
 				metas[i] = files[i].meta
 			}
 			lm := manifest.MakeLevelMetadata(cmp, 6, metas)
-			iter := NewLevelIter(keyspan.SpanIterOptions{}, cmp, tableNewIters, lm.Iter(), 6, keyType)
+			iter := NewLevelIter(context.Background(), keyspan.SpanIterOptions{}, cmp, tableNewIters, lm.Iter(), 6, keyType)
 			extraInfo := func() string {
 				return iter.String()
 			}
diff --git a/internal/keyspan/test_utils.go b/internal/keyspan/test_utils.go
index bbff49a7..060aadd3 100644
--- a/internal/keyspan/test_utils.go
+++ b/internal/keyspan/test_utils.go
@@ -6,6 +6,7 @@ package keyspan
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"go/token"
 	"io"
@@ -365,6 +366,11 @@ func (p *probeIterator) Prev() (*Span, error) {
 	return p.handleOp(op)
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (p *probeIterator) SetContext(ctx context.Context) {
+	p.iter.SetContext(ctx)
+}
+
 func (p *probeIterator) Close() {
 	op := op{Kind: OpClose}
 	if p.iter != nil {
@@ -561,6 +567,11 @@ func (i *invalidatingIter) Last() (*Span, error)             { return i.invalida
 func (i *invalidatingIter) Next() (*Span, error)             { return i.invalidate(i.iter.Next()) }
 func (i *invalidatingIter) Prev() (*Span, error)             { return i.invalidate(i.iter.Prev()) }
 
+// SetContext is part of the FragmentIterator interface.
+func (i *invalidatingIter) SetContext(ctx context.Context) {
+	i.iter.SetContext(ctx)
+}
+
 func (i *invalidatingIter) Close() {
 	_, _ = i.invalidate(nil, nil)
 	i.iter.Close()
diff --git a/internal/manifest/annotator_test.go b/internal/manifest/annotator_test.go
new file mode 100644
index 00000000..1765188b
--- /dev/null
+++ b/internal/manifest/annotator_test.go
@@ -0,0 +1,58 @@
+// 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 manifest
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/stretchr/testify/require"
+)
+
+func makeTestLevelMetadata(count int) (LevelMetadata, []*FileMetadata) {
+	files := make([]*FileMetadata, count)
+	for i := 0; i < count; i++ {
+		files[i] = newItem(key(i))
+	}
+
+	lm := MakeLevelMetadata(base.DefaultComparer.Compare, 6, files)
+	return lm, files
+}
+
+// NumFilesAnnotator is an Annotator which computes an annotation value
+// equal to the number of files included in the annotation.
+var NumFilesAnnotator = SumAnnotator(func(f *FileMetadata) (uint64, bool) {
+	return 1, true
+})
+
+func TestNumFilesAnnotator(t *testing.T) {
+	const count = 1000
+	lm, _ := makeTestLevelMetadata(0)
+
+	for i := 1; i <= count; i++ {
+		lm.tree.Insert(newItem(key(i)))
+		numFiles := *NumFilesAnnotator.LevelAnnotation(lm)
+		require.EqualValues(t, i, numFiles)
+	}
+
+	numFiles := *NumFilesAnnotator.LevelAnnotation(lm)
+	require.EqualValues(t, count, numFiles)
+
+	numFiles = *NumFilesAnnotator.LevelAnnotation(lm)
+	require.EqualValues(t, count, numFiles)
+
+	lm.tree.Delete(newItem(key(count / 2)))
+	numFiles = *NumFilesAnnotator.LevelAnnotation(lm)
+	require.EqualValues(t, count-1, numFiles)
+}
+
+func BenchmarkNumFilesAnnotator(b *testing.B) {
+	lm, _ := makeTestLevelMetadata(0)
+	for i := 1; i <= b.N; i++ {
+		lm.tree.Insert(newItem(key(i)))
+		numFiles := *NumFilesAnnotator.LevelAnnotation(lm)
+		require.EqualValues(b, uint64(i), numFiles)
+	}
+}
diff --git a/internal/manifest/btree_test.go b/internal/manifest/btree_test.go
index cce22a25..1c4b2ed2 100644
--- a/internal/manifest/btree_test.go
+++ b/internal/manifest/btree_test.go
@@ -145,10 +145,10 @@ func (n *node) recurse(f func(child *node, pos int16)) {
 //////////////////////////////////////////
 
 func key(i int) InternalKey {
-	if i < 0 || i > 99999 {
+	if i < 0 || i > 9999999 {
 		panic("key out of bounds")
 	}
-	return base.MakeInternalKey([]byte(fmt.Sprintf("%05d", i)), 0, base.InternalKeyKindSet)
+	return base.MakeInternalKey([]byte(fmt.Sprintf("%07d", i)), 0, base.InternalKeyKindSet)
 }
 
 func keyWithMemo(i int, memo map[int]InternalKey) InternalKey {
@@ -534,56 +534,6 @@ func TestIterEndSentinel(t *testing.T) {
 	require.True(t, iter.iter.valid())
 }
 
-type orderStatistic struct{}
-
-func (o orderStatistic) Zero(dst interface{}) interface{} {
-	if dst == nil {
-		return new(int)
-	}
-	v := dst.(*int)
-	*v = 0
-	return v
-}
-
-func (o orderStatistic) Accumulate(meta *FileMetadata, dst interface{}) (interface{}, bool) {
-	v := dst.(*int)
-	*v++
-	return v, true
-}
-
-func (o orderStatistic) Merge(src interface{}, dst interface{}) interface{} {
-	srcv := src.(*int)
-	dstv := dst.(*int)
-	*dstv = *dstv + *srcv
-	return dstv
-}
-
-func TestAnnotationOrderStatistic(t *testing.T) {
-	const count = 1000
-	ann := orderStatistic{}
-
-	var tr btree
-	tr.cmp = cmp
-	for i := 1; i <= count; i++ {
-		require.NoError(t, tr.Insert(newItem(key(i))))
-
-		v, ok := tr.root.Annotation(ann)
-		require.True(t, ok)
-		vtyped := v.(*int)
-		require.Equal(t, i, *vtyped)
-	}
-
-	v, ok := tr.root.Annotation(ann)
-	require.True(t, ok)
-	vtyped := v.(*int)
-	require.Equal(t, count, *vtyped)
-
-	v, ok = tr.root.Annotation(ann)
-	vtyped = v.(*int)
-	require.True(t, ok)
-	require.Equal(t, count, *vtyped)
-}
-
 // TestRandomizedBTree tests a random set of Insert, Delete and iteration
 // operations, checking for equivalence with a map of filenums.
 func TestRandomizedBTree(t *testing.T) {
diff --git a/iterator_test.go b/iterator_test.go
index d69b9c34..651ef208 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -2721,3 +2721,163 @@ func BenchmarkSeekPrefixTombstones(b *testing.B) {
 		iter.SeekPrefixGE(seekKey)
 	}
 }
+
+func runBenchmarkQueueWorkload(b *testing.B, deleteRatio float32, initOps int, valueSize int) {
+	const queueCount = 8
+	// These should be large enough to assign a unique key to each item in the
+	// queue.
+	const maxQueueIDLen = 1
+	const maxItemLen = 7
+	const maxKeyLen = maxQueueIDLen + 1 + maxItemLen
+	queueIDKeyspace := testkeys.Alpha(maxQueueIDLen)
+	itemKeyspace := testkeys.Alpha(maxItemLen)
+	key := make([]byte, maxKeyLen)
+	val := make([]byte, valueSize)
+	rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
+
+	getKey := func(q int, i int) []byte {
+		n := testkeys.WriteKey(key, queueIDKeyspace, int64(q))
+		key[n] = '/'
+		prefixLen := n + 1
+		n = testkeys.WriteKey(key[prefixLen:], itemKeyspace, int64(i))
+		return key[:prefixLen+n]
+	}
+
+	type Queue struct {
+		start int
+		end   int // exclusive
+	}
+	var queues = make([]*Queue, queueCount)
+	for i := 0; i < queueCount; i++ {
+		queues[i] = &Queue{}
+	}
+
+	o := (&Options{
+		DisableWAL:         true,
+		FS:                 vfs.NewMem(),
+		Comparer:           testkeys.Comparer,
+		FormatMajorVersion: FormatNewest,
+	}).EnsureDefaults()
+
+	d, err := Open("", o)
+	require.NoError(b, err)
+
+	processQueueOnce := func(batch *Batch) {
+		for {
+			// Randomly pick a queue to process.
+			q := rng.Intn(queueCount)
+			queue := queues[q]
+
+			isDelete := rng.Float32() < deleteRatio
+
+			if isDelete {
+				// Only process the queue if it's not empty. Otherwise, retry
+				// with a different queue.
+				if queue.start != queue.end {
+					require.NoError(b, batch.Delete(getKey(q, queue.start), nil))
+					queue.start = (queue.start + 1) % int(itemKeyspace.Count())
+					break
+				}
+			} else {
+				// Append to the queue.
+				require.NoError(b, batch.Set(getKey(q, queue.end), val, nil))
+				queue.end = (queue.end + 1) % int(itemKeyspace.Count())
+				break
+			}
+		}
+	}
+
+	// First, process queues initialOps times.
+	batch := d.NewBatch()
+	for i := 0; i < initOps; i++ {
+		processQueueOnce(batch)
+		// Use a large batch size to speed up initialization.
+		if batch.Len() >= 10<<24 /* 167 MiB */ {
+			require.NoError(b, batch.Commit(NoSync))
+			batch = d.NewBatch()
+		}
+	}
+	require.NoError(b, batch.Commit(NoSync))
+	// Manually flush in case the last batch was small.
+	_, err = d.AsyncFlush()
+	require.NoError(b, err)
+
+	waitForCompactions := func() {
+		d.mu.Lock()
+		// NB: Wait for table stats because some compaction types rely
+		// on table stats to be collected.
+		d.waitTableStats()
+		for d.mu.compact.compactingCount > 0 {
+			d.mu.compact.cond.Wait()
+			d.waitTableStats()
+		}
+		d.mu.Unlock()
+	}
+
+	waitForCompactions()
+
+	// Log the number of tombstones and live keys in each level after
+	// background compactions are complete.
+	b.Log("LSM after compactions:")
+	firstIter, _ := d.NewIter(nil)
+	firstIter.First()
+	lastIter, _ := d.NewIter(nil)
+	lastIter.Last()
+	stats, _ := d.ScanStatistics(context.Background(), firstIter.Key(), lastIter.Key(), ScanStatisticsOptions{})
+	require.NoError(b, firstIter.Close())
+	require.NoError(b, lastIter.Close())
+	metrics := d.Metrics()
+	for i := 0; i < numLevels; i++ {
+		numTombstones := stats.Levels[i].KindsCount[base.InternalKeyKindDelete]
+		numSets := stats.Levels[i].KindsCount[base.InternalKeyKindSet]
+		numTables := metrics.Levels[i].NumFiles
+		if numSets > 0 {
+			b.Logf("L%d: %d tombstones, %d sets, %d sstables\n", i, numTombstones, numSets, numTables)
+		}
+	}
+
+	// Seek to the start of each queue.
+	b.Run("seek", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			for q := 0; q < queueCount; q++ {
+				iter, _ := d.NewIter(nil)
+				iter.SeekGE(getKey(q, 0))
+				require.NoError(b, iter.Close())
+			}
+		}
+	})
+
+	require.NoError(b, d.Close())
+}
+
+// BenchmarkQueueWorkload benchmarks a workload consisting of multiple queues
+// that are all being processed at the same time. Processing a queue entails
+// either appending to the end of the queue (a Set operation) or deleting from
+// the start of the queue (a Delete operation). The goal is to detect cases
+// where we see a large buildup of point tombstones at the beginning of each
+// queue, which leads to the slowdown of SeekGE(<start of queue>). To that end,
+// the test subbenchmarks a series of configurations that each 1) process the
+// queues a certain number of times and then 2) benchmark both the queue
+// processing throughput and SeekGE performance. See
+// https://github.com/facebook/rocksdb/wiki/Implement-Queue-Service-Using-RocksDB
+// for more information.
+func BenchmarkQueueWorkload(b *testing.B) {
+	// The portion of processing ops that will be deletes for each subbenchmark.
+	var deleteRatios = []float32{0.1, 0.3, 0.5}
+	// The number of times queues will be processed before running each
+	// subbenchmark.
+	var initOps = []int{400_000, 800_000, 1_200_000, 2_000_000, 3_500_000, 5_000_000, 7_500_000, 10_000_000, 50_000_000}
+	// We vary the value size to identify how compaction behaves when the
+	// relative sizes of tombstones and the keys they delete are different.
+	var valueSizes = []int{128, 2048}
+
+	for _, deleteRatio := range deleteRatios {
+		for _, valueSize := range valueSizes {
+			for _, numInitOps := range initOps {
+				b.Run(fmt.Sprintf("initial_ops=%d/deleteRatio=%.2f/valueSize=%d", numInitOps, deleteRatio, valueSize), func(b *testing.B) {
+					runBenchmarkQueueWorkload(b, deleteRatio, numInitOps, valueSize)
+				})
+			}
+		}
+	}
+}
diff --git a/keyspan_probe_test.go b/keyspan_probe_test.go
index 759ca3b2..afee56a4 100644
--- a/keyspan_probe_test.go
+++ b/keyspan_probe_test.go
@@ -5,6 +5,7 @@
 package pebble
 
 import (
+	"context"
 	"fmt"
 	"go/token"
 	"io"
@@ -377,6 +378,11 @@ func (p *probeKeyspanIterator) DebugTree(tp treeprinter.Node) {
 	}
 }
 
+// SetContext is part of the FragmentIterator interface.
+func (p *probeKeyspanIterator) SetContext(ctx context.Context) {
+	p.iter.SetContext(ctx)
+}
+
 func (p *probeKeyspanIterator) Close() {
 	op := keyspanOp{Kind: opSpanClose}
 	if p.iter != nil {
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index bcb4763f..d3b0ced3 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -545,6 +545,17 @@ func TestReaderHideObsolete(t *testing.T) {
 				t, opts, "testdata/reader_hide_obsolete",
 				nil /* Reader */, true)
 		})
+		opts = WriterOptions{
+			TableFormat:    TableFormatPebblev4,
+			BlockSize:      blockSize,
+			IndexBlockSize: 1 << 30, // Force a single-level index block.
+			Comparer:       testkeys.Comparer,
+		}
+		t.Run(fmt.Sprintf("singleLevel/blockSize=%s", dName), func(t *testing.T) {
+			runTestReader(
+				t, opts, "testdata/reader_hide_obsolete",
+				nil /* Reader */, true)
+		})
 	}
 }
 
@@ -797,6 +808,13 @@ func runTestReader(t *testing.T, o WriterOptions, dir string, r *Reader, printVa
 						require.True(t, retHideObsoletePoints)
 					}
 				}
+				if d.HasArg("hide-obsolete-points-without-filter") {
+					var hideObsoletePoints bool
+					d.ScanArgs(t, "hide-obsolete-points-without-filter", &hideObsoletePoints)
+					if hideObsoletePoints {
+						transforms.HideObsoletePoints = true
+					}
+				}
 				var filterer *BlockPropertiesFilterer
 				if len(bpfs) > 0 {
 					filterer = newBlockPropertiesFilterer(bpfs, nil, nil)
diff --git a/sstable/testdata/reader_hide_obsolete/iter b/sstable/testdata/reader_hide_obsolete/iter
index 5f6eab79..1aba92f3 100644
--- a/sstable/testdata/reader_hide_obsolete/iter
+++ b/sstable/testdata/reader_hide_obsolete/iter
@@ -316,3 +316,29 @@ d.SINGLEDEL.40:
 d.MERGE.30:D30
 ----
 MERGE not supported in a strict-obsolete sstable
+
+# Regression test for #3761. If an entire block contains obsolete points,
+# skipBackward should still skip to blocks earlier in the sstable.
+
+build writing-to-lowest-level
+a.SET.10:A10
+force-obsolete: b.DEL.5:
+force-obsolete: b.SETWITHDEL.2:foo
+force-obsolete: c.DEL.0:
+force-obsolete: cc.DEL.5:
+----
+
+iter hide-obsolete-points-without-filter=true
+last
+----
+<a:10>:A10
+
+iter hide-obsolete-points-without-filter=true
+seek-lt d
+----
+<a:10>:A10
+
+iter hide-obsolete-points-without-filter=true
+seek-lt c
+----
+<a:10>:A10
diff --git a/testdata/tracing b/testdata/tracing
new file mode 100644
index 00000000..b24f8aef
--- /dev/null
+++ b/testdata/tracing
@@ -0,0 +1,88 @@
+build
+set hello foo
+----
+
+get
+hello
+----
+reading footer of 53 bytes took 5ms
+reading block of 37 bytes took 5ms (<redacted>)
+reading block of 417 bytes took 5ms (<redacted>)
+reading block of 31 bytes took 5ms (<redacted>)
+reading block of 32 bytes took 5ms (<redacted>)
+hello:foo
+
+# If we disable background tracing, we should see no traces with Get.
+get disable-background-tracing
+hello
+----
+hello:foo
+
+iter
+seek-ge hello
+----
+reading block of 31 bytes took 5ms (<redacted>)
+reading block of 32 bytes took 5ms (<redacted>)
+hello: (foo, .)
+
+snapshot-iter
+seek-ge hello
+----
+reading block of 31 bytes took 5ms (<redacted>)
+reading block of 32 bytes took 5ms (<redacted>)
+hello: (foo, .)
+
+indexed-batch-iter
+seek-ge hello
+----
+reading block of 31 bytes took 5ms (<redacted>)
+reading block of 32 bytes took 5ms (<redacted>)
+hello: (foo, .)
+
+build
+set hello@2 foo1
+range-key-set a z @4 foo2
+----
+
+# This operation should not read the range key block.
+get
+hello@2
+----
+reading footer of 53 bytes took 5ms
+reading block of 62 bytes took 5ms (<redacted>)
+reading block of 504 bytes took 5ms (<redacted>)
+reading block of 27 bytes took 5ms (<redacted>)
+reading block of 35 bytes took 5ms (<redacted>)
+hello@2:foo1
+
+# These operations should read the range key block, which shows up as a third
+# log.
+iter
+seek-ge hello
+next
+----
+reading block of 27 bytes took 5ms (<redacted>)
+reading block of 35 bytes took 5ms (<redacted>)
+reading block of 35 bytes took 5ms (<redacted>)
+hello: (., [a-z) @4=foo2 UPDATED)
+hello@2: (foo1, [a-z) @4=foo2)
+
+snapshot-iter
+seek-ge hello
+next
+----
+reading block of 27 bytes took 5ms (<redacted>)
+reading block of 35 bytes took 5ms (<redacted>)
+reading block of 35 bytes took 5ms (<redacted>)
+hello: (., [a-z) @4=foo2 UPDATED)
+hello@2: (foo1, [a-z) @4=foo2)
+
+indexed-batch-iter
+seek-ge hello
+next
+----
+reading block of 27 bytes took 5ms (<redacted>)
+reading block of 35 bytes took 5ms (<redacted>)
+reading block of 35 bytes took 5ms (<redacted>)
+hello: (., [a-z) @4=foo2 UPDATED)
+hello@2: (foo1, [a-z) @4=foo2)