Repository: cockroachdb/pebble
Pull request #4264 (2025-01-16 17:17:43)

cockroachkvs: minor updates
Bringing in a few comments from the CRDB code.

Informs [139068](https://github.com/cockroachdb/cockroach/issues/139068)

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


Patch:
diff --git a/cockroachkvs/cockroachkvs.go b/cockroachkvs/cockroachkvs.go
index b3d2860b..d3585907 100644
--- a/cockroachkvs/cockroachkvs.go
+++ b/cockroachkvs/cockroachkvs.go
@@ -58,6 +58,7 @@ var Comparer = base.Comparer{
 	CompareRangeSuffixes: CompareRangeSuffixes,
 	Compare:              Compare,
 	Equal:                Equal,
+
 	AbbreviatedKey: func(k []byte) uint64 {
 		key, ok := getKeyPartFromEngineKey(k)
 		if !ok {
@@ -65,7 +66,9 @@ var Comparer = base.Comparer{
 		}
 		return base.DefaultComparer.AbbreviatedKey(key)
 	},
+
 	FormatKey: FormatKey,
+
 	Separator: func(dst, a, b []byte) []byte {
 		aKey, ok := getKeyPartFromEngineKey(a)
 		if !ok {
@@ -89,6 +92,7 @@ var Comparer = base.Comparer{
 		// The separator is > aKey, so we only need to add the sentinel.
 		return append(dst, 0)
 	},
+
 	Successor: func(dst, a []byte) []byte {
 		aKey, ok := getKeyPartFromEngineKey(a)
 		if !ok {
@@ -106,6 +110,7 @@ var Comparer = base.Comparer{
 		// The successor is > aKey, so we only need to add the sentinel.
 		return append(dst, 0)
 	},
+
 	ImmediateSuccessor: func(dst, a []byte) []byte {
 		// The key `a` is guaranteed to be a bare prefix: It's a
 		// `engineKeyNoVersion` key without a version—just a trailing 0-byte to
@@ -116,6 +121,7 @@ var Comparer = base.Comparer{
 		// zero-length version.
 		return append(append(dst, a...), 0)
 	},
+
 	Name: "cockroach_comparator",
 }
 
@@ -230,9 +236,6 @@ func Split(key []byte) int {
 	// Last byte is the version length + 1 when there is a version, else it is
 	// 0.
 	versionLen := int(key[len(key)-1])
-	if versionLen > len(key) {
-		panic(errors.AssertionFailedf("invalid version length"))
-	}
 	return len(key) - versionLen
 }
 
@@ -266,7 +269,11 @@ func Compare(a, b []byte) int {
 	)
 }
 
-// CompareRangeSuffixes compares range key suffixes (normally timestamps).
+// CompareRangeSuffixes implements Comparer.CompareRangeSuffixes. It compares
+// cockroach suffixes (which are composed of the version and a trailing sentinel
+// byte); the version can be an MVCC timestamp or a lock key. It is stricter
+// than ComparePointSuffixes due to historical reasons; see
+// https://github.com/cockroachdb/cockroach/issues/130533
 func CompareRangeSuffixes(a, b []byte) int {
 	if len(a) == 0 || len(b) == 0 {
 		// Empty suffixes come before non-empty suffixes.
@@ -278,7 +285,15 @@ func CompareRangeSuffixes(a, b []byte) int {
 	return bytes.Compare(b[:len(b)-1], a[:len(a)-1])
 }
 
-// ComparePointSuffixes compares point key suffixes (normally timestamps).
+// ComparePointSuffixes compares suffixes of Cockroach point keys (which are
+// composed of the version and a trailing version-length byte); the version can
+// be an MVCC timestamp or a lock key. ComparePointSuffixes differs from
+// CompareRangeSuffixes in that the ComparePointSuffixes normalizes the
+// suffixes. Ideally we'd have one function that implemented the semantics of
+// EnginePointSuffixCompare, but due to historical reasons, range key suffix
+// comparisons must not perform normalization.
+//
+// See https://github.com/cockroachdb/cockroach/issues/130533
 func ComparePointSuffixes(a, b []byte) int {
 	if len(a) == 0 || len(b) == 0 {
 		// Empty suffixes sort before non-empty suffixes.
@@ -390,8 +405,16 @@ func getKeyPartFromEngineKey(engineKey []byte) (key []byte, ok bool) {
 
 const (
 	cockroachColRoachKey int = iota
+	// cockroachColMVCCWallTime is the wall time component of a MVCC timestamp,
+	// or zero if not an MVCC key.
 	cockroachColMVCCWallTime
+	// cockroachColMVCCLogical is the logical time component of a MVCC
+	// timestamp, or zero if not an MVCC key.
 	cockroachColMVCCLogical
+	// cockroachColUntypedVersion holds any non-empty, non-MVCC version. It does
+	// NOT include the 0x00 separator byte that delimits the prefix and suffix
+	// in a serialized engine key. In practice, this column is used to store the
+	// version of lock-table keys.
 	cockroachColUntypedVersion
 	cockroachColCount
 )
@@ -459,6 +482,8 @@ type cockroachKeyWriter struct {
 	prevSuffix      []byte
 }
 
+var _ colblk.KeyWriter = (*cockroachKeyWriter)(nil)
+
 func makeCockroachKeyWriter() *cockroachKeyWriter {
 	kw := &cockroachKeyWriter{}
 	kw.roachKeys.Init(16)
diff --git a/external_iterator.go b/external_iterator.go
index 03dca978..4d136c85 100644
--- a/external_iterator.go
+++ b/external_iterator.go
@@ -12,6 +12,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 // NewExternalIter takes an input 2d array of sstable files which may overlap
@@ -164,14 +165,14 @@ func createExternalPointIter(ctx context.Context, it *Iterator) (topLevelIterato
 			pointIter, err = r.NewPointIter(
 				ctx, transforms, it.opts.LowerBound, it.opts.UpperBound, nil, /* BlockPropertiesFilterer */
 				sstable.NeverUseFilterBlock,
-				&it.stats.InternalStats, nil,
+				block.ReadEnv{Stats: &it.stats.InternalStats, IterStats: nil},
 				sstable.MakeTrivialReaderProvider(r))
 			if err != nil {
 				return nil, err
 			}
 			rangeDelIter, err = r.NewRawRangeDelIter(ctx, sstable.FragmentIterTransforms{
 				SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum),
-			})
+			}, block.ReadEnv{Stats: &it.stats.InternalStats, IterStats: nil})
 			if err != nil {
 				return nil, err
 			}
@@ -220,7 +221,7 @@ func finishInitializingExternal(ctx context.Context, it *Iterator) error {
 				for _, r := range readers {
 					transforms := sstable.FragmentIterTransforms{SyntheticSeqNum: sstable.SyntheticSeqNum(seqNum)}
 					seqNum--
-					if rki, err := r.NewRawRangeKeyIter(ctx, transforms); err != nil {
+					if rki, err := r.NewRawRangeKeyIter(ctx, transforms, block.ReadEnv{Stats: &it.stats.InternalStats, IterStats: nil}); err != nil {
 						return err
 					} else if rki != nil {
 						rangeKeyIters = append(rangeKeyIters, rki)
diff --git a/file_cache.go b/file_cache.go
index 2628fa22..e09bda2c 100644
--- a/file_cache.go
+++ b/file_cache.go
@@ -454,10 +454,10 @@ func (c *fileCacheShard) newIters(
 	var iters iterSet
 	var err error
 	if kinds.RangeKey() && file.HasRangeKeys {
-		iters.rangeKey, err = newRangeKeyIter(ctx, r, file, cr, opts.SpanIterOptions())
+		iters.rangeKey, err = newRangeKeyIter(ctx, r, file, cr, opts.SpanIterOptions(), internalOpts)
 	}
 	if kinds.RangeDeletion() && file.HasPointKeys && err == nil {
-		iters.rangeDeletion, err = newRangeDelIter(ctx, file, cr, dbOpts)
+		iters.rangeDeletion, err = newRangeDelIter(ctx, file, cr, dbOpts, internalOpts)
 	}
 	if kinds.Point() && err == nil {
 		iters.point, err = c.newPointIter(ctx, v, file, cr, opts, internalOpts, dbOpts)
@@ -574,11 +574,11 @@ func (c *fileCacheShard) newPointIter(
 			uint64(uintptr(unsafe.Pointer(r))), opts.Category)
 	}
 	if internalOpts.compaction {
-		iter, err = cr.NewCompactionIter(transforms, iterStatsAccum, &v.readerProvider, internalOpts.bufferPool)
+		iter, err = cr.NewCompactionIter(transforms, block.ReadEnv{IterStats: iterStatsAccum, BufferPool: internalOpts.bufferPool}, &v.readerProvider)
 	} else {
 		iter, err = cr.NewPointIter(
 			ctx, transforms, opts.GetLowerBound(), opts.GetUpperBound(), filterer, filterBlockSizeLimit,
-			internalOpts.stats, iterStatsAccum, &v.readerProvider)
+			block.ReadEnv{Stats: internalOpts.stats, IterStats: iterStatsAccum}, &v.readerProvider)
 	}
 	if err != nil {
 		return nil, err
@@ -600,11 +600,20 @@ func (c *fileCacheShard) newPointIter(
 // sstable's range deletions. This function is for table-cache internal use
 // only, and callers should use newIters instead.
 func newRangeDelIter(
-	ctx context.Context, file *manifest.FileMetadata, cr sstable.CommonReader, dbOpts *fileCacheOpts,
+	ctx context.Context,
+	file *manifest.FileMetadata,
+	cr sstable.CommonReader,
+	dbOpts *fileCacheOpts,
+	internalOpts internalIterOpts,
 ) (keyspan.FragmentIterator, error) {
 	// NB: range-del iterator does not maintain a reference to the table, nor
 	// does it need to read from it after creation.
-	rangeDelIter, err := cr.NewRawRangeDelIter(ctx, file.FragmentIterTransforms())
+	readBlockEnv := block.ReadEnv{
+		Stats:      internalOpts.stats,
+		IterStats:  internalOpts.iterStatsAccumulator,
+		BufferPool: nil,
+	}
+	rangeDelIter, err := cr.NewRawRangeDelIter(ctx, file.FragmentIterTransforms(), readBlockEnv)
 	if err != nil {
 		return nil, err
 	}
@@ -630,6 +639,7 @@ func newRangeKeyIter(
 	file *fileMetadata,
 	cr sstable.CommonReader,
 	opts keyspan.SpanIterOptions,
+	internalOpts internalIterOpts,
 ) (keyspan.FragmentIterator, error) {
 	transforms := file.FragmentIterTransforms()
 	// Don't filter a table's range keys if the file contains RANGEKEYDELs.
@@ -646,7 +656,12 @@ func newRangeKeyIter(
 		}
 	}
 	// TODO(radu): wrap in an AssertBounds.
-	return cr.NewRawRangeKeyIter(ctx, transforms)
+	readBlockEnv := block.ReadEnv{
+		Stats:      internalOpts.stats,
+		IterStats:  internalOpts.iterStatsAccumulator,
+		BufferPool: nil,
+	}
+	return cr.NewRawRangeKeyIter(ctx, transforms, readBlockEnv)
 }
 
 // tableCacheShardReaderProvider implements sstable.ReaderProvider for a
@@ -1166,11 +1181,11 @@ func (v *fileCacheValue) load(
 	)
 	if err == nil {
 		o := dbOpts.readerOpts
-		o.SetInternalCacheOpts(sstableinternal.CacheOptions{
+		o.CacheOpts = sstableinternal.CacheOptions{
 			Cache:   dbOpts.cache,
 			CacheID: dbOpts.cacheID,
 			FileNum: backingFileNum,
-		})
+		}
 		r, err = sstable.NewReader(ctx, f, o)
 	}
 	if err == nil {
diff --git a/ingest.go b/ingest.go
index 6373b948..08491994 100644
--- a/ingest.go
+++ b/ingest.go
@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 func sstableKeyCompare(userCmp Compare, a, b InternalKey) int {
@@ -304,11 +305,11 @@ func ingestLoad1(
 	rangeKeyValidator rangeKeyIngestValidator,
 ) (meta *fileMetadata, lastRangeKey keyspan.Span, err error) {
 	o := opts.MakeReaderOptions()
-	o.SetInternalCacheOpts(sstableinternal.CacheOptions{
+	o.CacheOpts = sstableinternal.CacheOptions{
 		Cache:   opts.Cache,
 		CacheID: cacheID,
 		FileNum: base.PhysicalTableDiskFileNum(fileNum),
-	})
+	}
 	r, err := sstable.NewReader(ctx, readable, o)
 	if err != nil {
 		return nil, keyspan.Span{}, err
@@ -378,7 +379,7 @@ func ingestLoad1(
 		}
 	}
 
-	iter, err := r.NewRawRangeDelIter(ctx, sstable.NoFragmentTransforms)
+	iter, err := r.NewRawRangeDelIter(ctx, sstable.NoFragmentTransforms, block.NoReadEnv)
 	if err != nil {
 		return nil, keyspan.Span{}, err
 	}
@@ -408,7 +409,7 @@ func ingestLoad1(
 
 	// Update the range-key bounds for the table.
 	{
-		iter, err := r.NewRawRangeKeyIter(ctx, sstable.NoFragmentTransforms)
+		iter, err := r.NewRawRangeKeyIter(ctx, sstable.NoFragmentTransforms, block.NoReadEnv)
 		if err != nil {
 			return nil, keyspan.Span{}, err
 		}
diff --git a/metamorphic/build.go b/metamorphic/build.go
index fbbfff2a..4b5af83a 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 )
 
@@ -269,7 +270,7 @@ func openExternalObj(
 	pointIter, err = reader.NewIter(sstable.NoTransforms, start, end)
 	panicIfErr(err)
 
-	rangeDelIter, err = reader.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms)
+	rangeDelIter, err = reader.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms, block.NoReadEnv)
 	panicIfErr(err)
 	if rangeDelIter != nil {
 		rangeDelIter = keyspan.Truncate(
@@ -279,7 +280,7 @@ func openExternalObj(
 		)
 	}
 
-	rangeKeyIter, err = reader.NewRawRangeKeyIter(context.Background(), sstable.NoFragmentTransforms)
+	rangeKeyIter, err = reader.NewRawRangeKeyIter(context.Background(), sstable.NoFragmentTransforms, block.NoReadEnv)
 	panicIfErr(err)
 	if rangeKeyIter != nil {
 		rangeKeyIter = keyspan.Truncate(
diff --git a/replay/replay.go b/replay/replay.go
index 7a1eb40f..e1f2b8bd 100644
--- a/replay/replay.go
+++ b/replay/replay.go
@@ -31,6 +31,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/rangekey"
 	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"golang.org/x/perf/benchfmt"
 	"golang.org/x/sync/errgroup"
@@ -1022,7 +1023,7 @@ func loadFlushedSSTableKeys(
 			}
 
 			// Load all the range tombstones.
-			if iter, err := r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms); err != nil {
+			if iter, err := r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms, block.NoReadEnv); err != nil {
 				return err
 			} else if iter != nil {
 				defer iter.Close()
@@ -1045,7 +1046,7 @@ func loadFlushedSSTableKeys(
 			}
 
 			// Load all the range keys.
-			if iter, err := r.NewRawRangeKeyIter(context.Background(), sstable.NoFragmentTransforms); err != nil {
+			if iter, err := r.NewRawRangeKeyIter(context.Background(), sstable.NoFragmentTransforms, block.NoReadEnv); err != nil {
 				return err
 			} else if iter != nil {
 				defer iter.Close()
diff --git a/sstable/blob/blob.go b/sstable/blob/blob.go
index bcd1192b..fec88b2f 100644
--- a/sstable/blob/blob.go
+++ b/sstable/blob/blob.go
@@ -12,6 +12,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/valblk"
 	"github.com/cockroachdb/redact"
@@ -323,23 +324,60 @@ func (f *fileFooter) encode(b []byte) {
 
 // FileReader reads a blob file.
 type FileReader struct {
-	r      objstorage.Readable
+	r      block.Reader
 	footer fileFooter
 }
 
+// FileReaderOptions configures a reader of a blob file.
+type FileReaderOptions struct {
+	block.ReaderOptions
+}
+
+func (o FileReaderOptions) ensureDefaults() FileReaderOptions {
+	if o.LoggerAndTracer == nil {
+		o.LoggerAndTracer = base.NoopLoggerAndTracer{}
+	}
+	return o
+}
+
 // NewFileReader opens a blob file for reading.
-func NewFileReader(ctx context.Context, r objstorage.Readable) (*FileReader, error) {
+func NewFileReader(
+	ctx context.Context, r objstorage.Readable, ro FileReaderOptions,
+) (*FileReader, error) {
+	ro = ro.ensureDefaults()
+
+	fileNum := ro.CacheOpts.FileNum
+
 	var footerBuf [fileFooterLength]byte
-	if err := r.ReadAt(ctx, footerBuf[:], r.Size()-fileFooterLength); err != nil {
+	size := r.Size()
+	off := size - fileFooterLength
+	if size < fileFooterLength {
+		return nil, base.CorruptionErrorf("pebble: invalid blob file %s (file size is too small)",
+			errors.Safe(fileNum))
+	}
+	var preallocRH objstorageprovider.PreallocatedReadHandle
+	rh := objstorageprovider.UsePreallocatedReadHandle(
+		r, objstorage.ReadBeforeForNewReader, &preallocRH)
+	defer rh.Close()
+
+	encodedFooter, err := block.ReadRaw(ctx, r, rh, ro.LoggerAndTracer, fileNum, footerBuf[:], off)
+	if err != nil {
 		return nil, err
 	}
-	fr := &FileReader{r: r}
-	if err := fr.footer.decode(footerBuf[:]); err != nil {
+
+	fr := &FileReader{}
+	if err := fr.footer.decode(encodedFooter); err != nil {
 		return nil, err
 	}
+	fr.r.Init(r, ro.ReaderOptions, fr.footer.checksum)
 	return fr, nil
 }
 
+// Close implements io.Closer, closing the underlying Readable.
+func (r *FileReader) Close() error {
+	return r.r.Close()
+}
+
 // lenLittleEndian returns the minimum number of bytes needed to encode v
 // using little endian encoding.
 func lenLittleEndian(v uint64) int {
diff --git a/sstable/block/block.go b/sstable/block/block.go
index ece228b3..06993e45 100644
--- a/sstable/block/block.go
+++ b/sstable/block/block.go
@@ -323,7 +323,7 @@ type ReadEnv struct {
 	// iterator is closed. In the important code paths, the CategoryStatsCollector
 	// is managed by the fileCacheContainer.
 	Stats     *base.InternalIteratorStats
-	IterStats *ChildIterStatsAccumulator
+	IterStats IterStatsAccumulator
 
 	// BufferPool is not-nil if we read blocks into a buffer pool and not into the
 	// cache. This is used during compactions.
@@ -355,39 +355,42 @@ func (env *ReadEnv) BlockRead(blockLength uint64, readDuration time.Duration) {
 // A Reader reads blocks from a single file, handling caching, checksum
 // validation and decompression.
 type Reader struct {
-	readable      objstorage.Readable
-	cacheOpts     sstableinternal.CacheOptions
-	loadBlockSema *fifo.Semaphore
-	logger        base.LoggerAndTracer
-	checksumType  ChecksumType
+	readable     objstorage.Readable
+	opts         ReaderOptions
+	checksumType ChecksumType
+}
+
+// ReaderOptions configures a block reader.
+type ReaderOptions struct {
+	// CacheOpts contains the information needed to interact with the block
+	// cache.
+	CacheOpts sstableinternal.CacheOptions
+	// LoadBlockSema, if set, is used to limit the number of blocks that can be
+	// loaded (i.e. read from the filesystem) in parallel. Each load acquires
+	// one unit from the semaphore for the duration of the read.
+	LoadBlockSema *fifo.Semaphore
+	// LoggerAndTracer is an optional logger and tracer.
+	LoggerAndTracer base.LoggerAndTracer
 }
 
 // Init initializes the Reader to read blocks from the provided Readable.
-func (r *Reader) Init(
-	readable objstorage.Readable,
-	cacheOpts sstableinternal.CacheOptions,
-	sema *fifo.Semaphore,
-	logger base.LoggerAndTracer,
-	checksumType ChecksumType,
-) {
+func (r *Reader) Init(readable objstorage.Readable, ro ReaderOptions, checksumType ChecksumType) {
 	r.readable = readable
-	r.cacheOpts = cacheOpts
-	r.loadBlockSema = sema
-	r.logger = logger
+	r.opts = ro
 	r.checksumType = checksumType
-	if r.cacheOpts.Cache == nil {
-		r.cacheOpts.Cache = cache.New(0)
+	if r.opts.CacheOpts.Cache == nil {
+		r.opts.CacheOpts.Cache = cache.New(0)
 	} else {
-		r.cacheOpts.Cache.Ref()
+		r.opts.CacheOpts.Cache.Ref()
 	}
-	if r.cacheOpts.CacheID == 0 {
-		r.cacheOpts.CacheID = r.cacheOpts.Cache.NewID()
+	if r.opts.CacheOpts.CacheID == 0 {
+		r.opts.CacheOpts.CacheID = r.opts.CacheOpts.Cache.NewID()
 	}
 }
 
 // FileNum returns the file number of the file being read.
 func (r *Reader) FileNum() base.DiskFileNum {
-	return r.cacheOpts.FileNum
+	return r.opts.CacheOpts.FileNum
 }
 
 // ChecksumType returns the checksum type used by the reader.
@@ -410,10 +413,10 @@ func (r *Reader) Read(
 	if env.BufferPool == nil {
 		var errorDuration time.Duration
 		var err error
-		cv, crh, errorDuration, hit, err = r.cacheOpts.Cache.GetWithReadHandle(
-			ctx, r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
-		if errorDuration > 5*time.Millisecond && r.logger.IsTracingEnabled(ctx) {
-			r.logger.Eventf(
+		cv, crh, errorDuration, hit, err = r.opts.CacheOpts.Cache.GetWithReadHandle(
+			ctx, r.opts.CacheOpts.CacheID, r.opts.CacheOpts.FileNum, bh.Offset)
+		if errorDuration > 5*time.Millisecond && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
+			r.opts.LoggerAndTracer.Eventf(
 				ctx, "waited for turn when %s time wasted by failed reads", errorDuration.String())
 		}
 		// TODO(sumeer): consider tracing when waited longer than some duration
@@ -425,7 +428,7 @@ func (r *Reader) Read(
 		// The compaction path uses env.BufferPool, and does not coordinate read
 		// using a cache.ReadHandle. This is ok since only a single compaction is
 		// reading a block.
-		cv = r.cacheOpts.Cache.Get(r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
+		cv = r.opts.CacheOpts.Cache.Get(r.opts.CacheOpts.CacheID, r.opts.CacheOpts.FileNum, bh.Offset)
 		if cv != nil {
 			hit = true
 		}
@@ -446,7 +449,7 @@ func (r *Reader) Read(
 	}
 
 	// Need to read. First acquire loadBlockSema, if needed.
-	if sema := r.loadBlockSema; sema != nil {
+	if sema := r.opts.LoadBlockSema; sema != nil {
 		if err := sema.Acquire(ctx, 1); err != nil {
 			// An error here can only come from the context.
 			return BufferHandle{}, err
@@ -460,7 +463,7 @@ func (r *Reader) Read(
 		}
 		return BufferHandle{}, err
 	}
-	h := value.MakeHandle(crh, r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
+	h := value.MakeHandle(crh, r.opts.CacheOpts.CacheID, r.opts.CacheOpts.FileNum, bh.Offset)
 	return h, nil
 }
 
@@ -487,12 +490,12 @@ func (r *Reader) doRead(
 	readDuration := readStopwatch.stop()
 	// Call IsTracingEnabled to avoid the allocations of boxing integers into an
 	// interface{}, unless necessary.
-	if readDuration >= slowReadTracingThreshold && r.logger.IsTracingEnabled(ctx) {
+	if readDuration >= slowReadTracingThreshold && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
 		_, file1, line1, _ := runtime.Caller(1)
 		_, file2, line2, _ := runtime.Caller(2)
-		r.logger.Eventf(ctx, "reading block of %d bytes took %s (fileNum=%s; %s/%s:%d -> %s/%s:%d)",
+		r.opts.LoggerAndTracer.Eventf(ctx, "reading block of %d bytes took %s (fileNum=%s; %s/%s:%d -> %s/%s:%d)",
 			int(bh.Length+TrailerLen), readDuration.String(),
-			r.cacheOpts.FileNum,
+			r.opts.CacheOpts.FileNum,
 			filepath.Base(filepath.Dir(file2)), filepath.Base(file2), line2,
 			filepath.Base(filepath.Dir(file1)), filepath.Base(file1), line1)
 	}
@@ -503,7 +506,7 @@ func (r *Reader) doRead(
 	env.BlockRead(bh.Length, readDuration)
 	if err = ValidateChecksum(r.checksumType, compressed.BlockData(), bh); err != nil {
 		compressed.Release()
-		err = errors.Wrapf(err, "pebble/table: table %s", r.cacheOpts.FileNum)
+		err = errors.Wrapf(err, "pebble/table: table %s", r.opts.CacheOpts.FileNum)
 		return Value{}, err
 	}
 	typ := CompressionIndicator(compressed.BlockData()[bh.Length])
@@ -545,7 +548,7 @@ func (r *Reader) Readable() objstorage.Readable {
 // Users should prefer using Read, which handles reading from object storage on
 // a cache miss.
 func (r *Reader) GetFromCache(bh Handle) *cache.Value {
-	return r.cacheOpts.Cache.Get(r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
+	return r.opts.CacheOpts.Cache.Get(r.opts.CacheOpts.CacheID, r.opts.CacheOpts.FileNum, bh.Offset)
 }
 
 // UsePreallocatedReadHandle returns a ReadHandle that reads from the reader and
@@ -559,7 +562,7 @@ func (r *Reader) UsePreallocatedReadHandle(
 
 // Close releases resources associated with the Reader.
 func (r *Reader) Close() error {
-	r.cacheOpts.Cache.Unref()
+	r.opts.CacheOpts.Cache.Unref()
 	var err error
 	if r.readable != nil {
 		err = r.readable.Close()
diff --git a/sstable/block/category_stats.go b/sstable/block/category_stats.go
index fa575f24..a516371f 100644
--- a/sstable/block/category_stats.go
+++ b/sstable/block/category_stats.go
@@ -155,10 +155,12 @@ type CategoryStats struct {
 	BlockReadDuration time.Duration
 }
 
-func (s *CategoryStats) aggregate(a CategoryStats) {
-	s.BlockBytes += a.BlockBytes
-	s.BlockBytesInCache += a.BlockBytesInCache
-	s.BlockReadDuration += a.BlockReadDuration
+func (s *CategoryStats) aggregate(
+	blockBytes, blockBytesInCache uint64, blockReadDuration time.Duration,
+) {
+	s.BlockBytes += blockBytes
+	s.BlockBytesInCache += blockBytesInCache
+	s.BlockReadDuration += blockReadDuration
 }
 
 // CategoryStatsAggregate is the aggregate for the given category.
@@ -176,9 +178,11 @@ type categoryStatsWithMu struct {
 }
 
 // Accumulate implements the IterStatsAccumulator interface.
-func (c *categoryStatsWithMu) Accumulate(stats CategoryStats) {
+func (c *categoryStatsWithMu) Accumulate(
+	blockBytes, blockBytesInCache uint64, blockReadDuration time.Duration,
+) {
 	c.mu.Lock()
-	c.stats.aggregate(stats)
+	c.stats.aggregate(blockBytes, blockBytesInCache, blockReadDuration)
 	c.mu.Unlock()
 }
 
@@ -210,7 +214,7 @@ func (s *shardedCategoryStats) getStats() CategoryStatsAggregate {
 	}
 	for i := range s.shards {
 		s.shards[i].mu.Lock()
-		agg.CategoryStats.aggregate(s.shards[i].stats)
+		agg.CategoryStats.aggregate(s.shards[i].stats.BlockBytes, s.shards[i].stats.BlockBytesInCache, s.shards[i].stats.BlockReadDuration)
 		s.shards[i].mu.Unlock()
 	}
 	return agg
@@ -250,34 +254,5 @@ func (c *CategoryStatsCollector) GetStats() []CategoryStatsAggregate {
 
 type IterStatsAccumulator interface {
 	// Accumulate accumulates the provided stats.
-	Accumulate(cas CategoryStats)
-}
-
-// ChildIterStatsAccumulator is a helper for a sstable iterator to accumulate
-// stats, which are reported to the CategoryStatsCollector when the accumulator
-// is closed.
-type ChildIterStatsAccumulator struct {
-	stats  CategoryStats
-	parent IterStatsAccumulator
-}
-
-// Init initializes the accumulator with the parent accumulator.
-func (a *ChildIterStatsAccumulator) Init(parent IterStatsAccumulator) {
-	a.parent = parent
-}
-
-// Accumulate accumulates the provided stats.
-func (a *ChildIterStatsAccumulator) Accumulate(
-	blockBytes, blockBytesInCache uint64, blockReadDuration time.Duration,
-) {
-	a.stats.BlockBytes += blockBytes
-	a.stats.BlockBytesInCache += blockBytesInCache
-	a.stats.BlockReadDuration += blockReadDuration
-}
-
-// Close closes the accumulator, accumulating the stats to the parent.
-func (a *ChildIterStatsAccumulator) Close() {
-	if a.parent != nil {
-		a.parent.Accumulate(a.stats)
-	}
+	Accumulate(blockBytes, blockBytesInCache uint64, blockReadDuration time.Duration)
 }
diff --git a/sstable/colblk_writer.go b/sstable/colblk_writer.go
index 0021d153..1a49c2d8 100644
--- a/sstable/colblk_writer.go
+++ b/sstable/colblk_writer.go
@@ -1066,7 +1066,9 @@ func (w *RawColumnWriter) rewriteSuffixes(
 			w.filterBlock = copyFilterWriter{
 				origPolicyName: w.filterBlock.policyName(),
 				origMetaName:   w.filterBlock.metaName(),
-				data:           filterBlock,
+				// Clone the filter block, because readBlockBuf allows the
+				// returned byte slice to point directly into sst.
+				data: slices.Clone(filterBlock),
 			}
 		}
 	}
diff --git a/sstable/options.go b/sstable/options.go
index 3d3ac9a8..b4f70bbf 100644
--- a/sstable/options.go
+++ b/sstable/options.go
@@ -7,7 +7,6 @@ package sstable
 import (
 	"fmt"
 
-	"github.com/cockroachdb/crlib/fifo"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/sstableinternal"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -81,10 +80,7 @@ func MakeKeySchemas(keySchemas ...*colblk.KeySchema) KeySchemas {
 
 // ReaderOptions holds the parameters needed for reading an sstable.
 type ReaderOptions struct {
-	// LoadBlockSema, if set, is used to limit the number of blocks that can be
-	// loaded (i.e. read from the filesystem) in parallel. Each load acquires one
-	// unit from the semaphore for the duration of the read.
-	LoadBlockSema *fifo.Semaphore
+	block.ReaderOptions
 
 	// User properties specified in this map will not be added to sst.Properties.UserProperties.
 	DeniedUserProperties map[string]struct{}
@@ -110,28 +106,8 @@ type ReaderOptions struct {
 	// policies that are not in this map will be ignored.
 	Filters map[string]FilterPolicy
 
-	// Logger is an optional logger and tracer.
-	LoggerAndTracer base.LoggerAndTracer
-
 	// FilterMetricsTracker is optionally used to track filter metrics.
 	FilterMetricsTracker *FilterMetricsTracker
-
-	// internal options can only be used from within the pebble package.
-	internal sstableinternal.ReaderOptions
-}
-
-// SetInternal sets the internal reader options. Note that even though this
-// method is public, a caller outside the pebble package can't construct a value
-// to pass to it.
-func (o *ReaderOptions) SetInternal(internalOpts sstableinternal.ReaderOptions) {
-	o.internal = internalOpts
-}
-
-// SetInternalCacheOpts sets the internal cache options. Note that even though
-// this method is public, a caller outside the pebble package can't construct a
-// value to pass to it.
-func (o *ReaderOptions) SetInternalCacheOpts(cacheOpts sstableinternal.CacheOptions) {
-	o.internal.CacheOpts = cacheOpts
 }
 
 func (o ReaderOptions) ensureDefaults() ReaderOptions {
diff --git a/sstable/reader.go b/sstable/reader.go
index 7ac36dec..028b2598 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -101,13 +101,12 @@ func (r *Reader) NewPointIter(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	filterBlockSizeLimit FilterBlockSizeLimit,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
+	env block.ReadEnv,
 	rp valblk.ReaderProvider,
 ) (Iterator, error) {
 	return r.newPointIter(
 		ctx, transforms, lower, upper, filterer, filterBlockSizeLimit,
-		stats, statsAccum, rp, nil)
+		env, rp, nil)
 }
 
 // TryAddBlockPropertyFilterForHideObsoletePoints is expected to be called
@@ -132,8 +131,7 @@ func (r *Reader) newPointIter(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	filterBlockSizeLimit FilterBlockSizeLimit,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
+	env block.ReadEnv,
 	rp valblk.ReaderProvider,
 	vState *virtualState,
 ) (Iterator, error) {
@@ -146,21 +144,21 @@ func (r *Reader) newPointIter(
 		if r.tableFormat.BlockColumnar() {
 			res, err = newColumnBlockTwoLevelIterator(
 				ctx, r, vState, transforms, lower, upper, filterer, filterBlockSizeLimit,
-				stats, statsAccum, rp, nil /* bufferPool */)
+				env, rp)
 		} else {
 			res, err = newRowBlockTwoLevelIterator(
 				ctx, r, vState, transforms, lower, upper, filterer, filterBlockSizeLimit,
-				stats, statsAccum, rp, nil /* bufferPool */)
+				env, rp)
 		}
 	} else {
 		if r.tableFormat.BlockColumnar() {
 			res, err = newColumnBlockSingleLevelIterator(
 				ctx, r, vState, transforms, lower, upper, filterer, filterBlockSizeLimit,
-				stats, statsAccum, rp, nil /* bufferPool */)
+				env, rp)
 		} else {
 			res, err = newRowBlockSingleLevelIterator(
 				ctx, r, vState, transforms, lower, upper, filterer, filterBlockSizeLimit,
-				stats, statsAccum, rp, nil /* bufferPool */)
+				env, rp)
 		}
 	}
 	if err != nil {
@@ -182,37 +180,31 @@ func (r *Reader) NewIter(transforms IterTransforms, lower, upper []byte) (Iterat
 	// likely isn't a cache set up.
 	return r.NewPointIter(
 		context.TODO(), transforms, lower, upper, nil, AlwaysUseFilterBlock,
-		nil /* stats */, nil /* statsAccum */, MakeTrivialReaderProvider(r))
+		block.NoReadEnv, MakeTrivialReaderProvider(r))
 }
 
 // NewCompactionIter returns an iterator similar to NewIter but it also increments
 // the number of bytes iterated. If an error occurs, NewCompactionIter cleans up
 // after itself and returns a nil iterator.
 func (r *Reader) NewCompactionIter(
-	transforms IterTransforms,
-	statsAccum block.IterStatsAccumulator,
-	rp valblk.ReaderProvider,
-	bufferPool *block.BufferPool,
+	transforms IterTransforms, env block.ReadEnv, rp valblk.ReaderProvider,
 ) (Iterator, error) {
-	return r.newCompactionIter(transforms, statsAccum, rp, nil, bufferPool)
+	return r.newCompactionIter(transforms, env, rp, nil)
 }
 
 func (r *Reader) newCompactionIter(
-	transforms IterTransforms,
-	statsAccum block.IterStatsAccumulator,
-	rp valblk.ReaderProvider,
-	vState *virtualState,
-	bufferPool *block.BufferPool,
+	transforms IterTransforms, env block.ReadEnv, rp valblk.ReaderProvider, vState *virtualState,
 ) (Iterator, error) {
 	if vState != nil && vState.isSharedIngested {
 		transforms.HideObsoletePoints = true
 	}
+
 	if r.Properties.IndexType == twoLevelIndex {
 		if !r.tableFormat.BlockColumnar() {
 			i, err := newRowBlockTwoLevelIterator(
 				context.Background(),
 				r, vState, transforms, nil /* lower */, nil /* upper */, nil,
-				NeverUseFilterBlock, nil /* stats */, statsAccum, rp, bufferPool)
+				NeverUseFilterBlock, env, rp)
 			if err != nil {
 				return nil, err
 			}
@@ -222,7 +214,7 @@ func (r *Reader) newCompactionIter(
 		i, err := newColumnBlockTwoLevelIterator(
 			context.Background(),
 			r, vState, transforms, nil /* lower */, nil /* upper */, nil,
-			NeverUseFilterBlock, nil /* stats */, statsAccum, rp, bufferPool)
+			NeverUseFilterBlock, env, rp)
 		if err != nil {
 			return nil, err
 		}
@@ -232,7 +224,7 @@ func (r *Reader) newCompactionIter(
 	if !r.tableFormat.BlockColumnar() {
 		i, err := newRowBlockSingleLevelIterator(
 			context.Background(), r, vState, transforms, nil /* lower */, nil, /* upper */
-			nil, NeverUseFilterBlock, nil /* stats */, statsAccum, rp, bufferPool)
+			nil, NeverUseFilterBlock, env, rp)
 		if err != nil {
 			return nil, err
 		}
@@ -241,7 +233,7 @@ func (r *Reader) newCompactionIter(
 	}
 	i, err := newColumnBlockSingleLevelIterator(
 		context.Background(), r, vState, transforms, nil /* lower */, nil, /* upper */
-		nil, NeverUseFilterBlock, nil /* stats */, statsAccum, rp, bufferPool)
+		nil, NeverUseFilterBlock, env, rp)
 	if err != nil {
 		return nil, err
 	}
@@ -253,13 +245,13 @@ func (r *Reader) newCompactionIter(
 // range-del block for the table. Returns nil if the table does not contain
 // any range deletions.
 func (r *Reader) NewRawRangeDelIter(
-	ctx context.Context, transforms FragmentIterTransforms,
+	ctx context.Context, transforms FragmentIterTransforms, env block.ReadEnv,
 ) (iter keyspan.FragmentIterator, err error) {
 	if r.rangeDelBH.Length == 0 {
 		return nil, nil
 	}
 	// TODO(radu): plumb stats here.
-	h, err := r.readRangeDelBlock(ctx, block.NoReadEnv, noReadHandle, r.rangeDelBH)
+	h, err := r.readRangeDelBlock(ctx, env, noReadHandle, r.rangeDelBH)
 	if err != nil {
 		return nil, err
 	}
@@ -278,13 +270,13 @@ func (r *Reader) NewRawRangeDelIter(
 // range-key block for the table. Returns nil if the table does not contain any
 // range keys.
 func (r *Reader) NewRawRangeKeyIter(
-	ctx context.Context, transforms FragmentIterTransforms,
+	ctx context.Context, transforms FragmentIterTransforms, env block.ReadEnv,
 ) (iter keyspan.FragmentIterator, err error) {
 	if r.rangeKeyBH.Length == 0 {
 		return nil, nil
 	}
 	// TODO(radu): plumb stats here.
-	h, err := r.readRangeKeyBlock(ctx, block.NoReadEnv, noReadHandle, r.rangeKeyBH)
+	h, err := r.readRangeKeyBlock(ctx, env, noReadHandle, r.rangeKeyBH)
 	if err != nil {
 		return nil, err
 	}
@@ -823,11 +815,11 @@ func NewReader(ctx context.Context, f objstorage.Readable, o ReaderOptions) (*Re
 		f, objstorage.ReadBeforeForNewReader, &preallocRH)
 	defer rh.Close()
 
-	footer, err := readFooter(ctx, f, rh, o.LoggerAndTracer, o.internal.CacheOpts.FileNum)
+	footer, err := readFooter(ctx, f, rh, o.LoggerAndTracer, o.CacheOpts.FileNum)
 	if err != nil {
 		return nil, errors.CombineErrors(err, f.Close())
 	}
-	r.blockReader.Init(f, o.internal.CacheOpts, o.LoadBlockSema, o.LoggerAndTracer, footer.checksum)
+	r.blockReader.Init(f, o.ReaderOptions, footer.checksum)
 	r.tableFormat = footer.format
 	r.indexBH = footer.indexBH
 	r.metaindexBH = footer.metaindexBH
diff --git a/sstable/reader_common.go b/sstable/reader_common.go
index 6dd8e3eb..12ffc704 100644
--- a/sstable/reader_common.go
+++ b/sstable/reader_common.go
@@ -8,7 +8,6 @@ import (
 	"context"
 	"math"
 
-	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/valblk"
@@ -19,11 +18,11 @@ import (
 // virtual reader.
 type CommonReader interface {
 	NewRawRangeKeyIter(
-		ctx context.Context, transforms FragmentIterTransforms,
+		ctx context.Context, transforms FragmentIterTransforms, env block.ReadEnv,
 	) (keyspan.FragmentIterator, error)
 
 	NewRawRangeDelIter(
-		ctx context.Context, transforms FragmentIterTransforms,
+		ctx context.Context, transforms FragmentIterTransforms, env block.ReadEnv,
 	) (keyspan.FragmentIterator, error)
 
 	NewPointIter(
@@ -32,16 +31,14 @@ type CommonReader interface {
 		lower, upper []byte,
 		filterer *BlockPropertiesFilterer,
 		filterBlockSizeLimit FilterBlockSizeLimit,
-		stats *base.InternalIteratorStats,
-		statsAccum block.IterStatsAccumulator,
+		env block.ReadEnv,
 		rp valblk.ReaderProvider,
 	) (Iterator, error)
 
 	NewCompactionIter(
 		transforms IterTransforms,
-		statsAccum block.IterStatsAccumulator,
+		env block.ReadEnv,
 		rp valblk.ReaderProvider,
-		bufferPool *block.BufferPool,
 	) (Iterator, error)
 
 	EstimateDiskUsage(start, end []byte) (uint64, error)
diff --git a/sstable/reader_iter_single_lvl.go b/sstable/reader_iter_single_lvl.go
index 9b27b672..559becd3 100644
--- a/sstable/reader_iter_single_lvl.go
+++ b/sstable/reader_iter_single_lvl.go
@@ -73,7 +73,6 @@ type singleLevelIterator[I any, PI indexBlockIterator[I], D any, PD dataBlockIte
 	err          error
 	closeHook    func(i Iterator) error
 
-	iterStats    block.ChildIterStatsAccumulator
 	readBlockEnv block.ReadEnv
 
 	// boundsCmp and positionedUsingLatestBounds are for optimizing iteration
@@ -210,10 +209,8 @@ func newColumnBlockSingleLevelIterator(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	filterBlockSizeLimit FilterBlockSizeLimit,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
+	env block.ReadEnv,
 	rp valblk.ReaderProvider,
-	bufferPool *block.BufferPool,
 ) (*singleLevelIteratorColumnBlocks, error) {
 	if r.err != nil {
 		return nil, r.err
@@ -225,11 +222,11 @@ func newColumnBlockSingleLevelIterator(
 	useFilterBlock := shouldUseFilterBlock(r, filterBlockSizeLimit)
 	i.init(
 		ctx, r, v, transforms, lower, upper, filterer, useFilterBlock,
-		stats, statsAccum, bufferPool,
+		env,
 	)
 	var getLazyValuer block.GetLazyValueForPrefixAndValueHandler
 	if r.Properties.NumValueBlocks > 0 {
-		i.vbReader = valblk.MakeReader(i, rp, r.valueBIH, stats)
+		i.vbReader = valblk.MakeReader(i, rp, r.valueBIH, env.Stats)
 		getLazyValuer = &i.vbReader
 		i.vbRH = r.blockReader.UsePreallocatedReadHandle(objstorage.NoReadBefore, &i.vbRHPrealloc)
 	}
@@ -260,10 +257,8 @@ func newRowBlockSingleLevelIterator(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	filterBlockSizeLimit FilterBlockSizeLimit,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
+	env block.ReadEnv,
 	rp valblk.ReaderProvider,
-	bufferPool *block.BufferPool,
 ) (*singleLevelIteratorRowBlocks, error) {
 	if r.err != nil {
 		return nil, r.err
@@ -275,11 +270,11 @@ func newRowBlockSingleLevelIterator(
 	useFilterBlock := shouldUseFilterBlock(r, filterBlockSizeLimit)
 	i.init(
 		ctx, r, v, transforms, lower, upper, filterer, useFilterBlock,
-		stats, statsAccum, bufferPool,
+		env,
 	)
 	if r.tableFormat >= TableFormatPebblev3 {
 		if r.Properties.NumValueBlocks > 0 {
-			i.vbReader = valblk.MakeReader(i, rp, r.valueBIH, stats)
+			i.vbReader = valblk.MakeReader(i, rp, r.valueBIH, env.Stats)
 			(&i.data).SetGetLazyValuer(&i.vbReader)
 			i.vbRH = r.blockReader.UsePreallocatedReadHandle(objstorage.NoReadBefore, &i.vbRHPrealloc)
 		}
@@ -306,9 +301,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) init(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	useFilterBlock bool,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
-	bufferPool *block.BufferPool,
+	env block.ReadEnv,
 ) {
 	i.inPool = false
 	i.ctx = ctx
@@ -323,12 +316,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) init(
 		i.vState = v
 		i.endKeyInclusive, i.lower, i.upper = v.constrainBounds(lower, upper, false /* endInclusive */)
 	}
-	i.iterStats.Init(statsAccum)
-	i.readBlockEnv = block.ReadEnv{
-		Stats:      stats,
-		IterStats:  &i.iterStats,
-		BufferPool: bufferPool,
-	}
+	i.readBlockEnv = env
 
 	i.indexFilterRH = r.blockReader.UsePreallocatedReadHandle(
 		objstorage.ReadBeforeForIndexAndFilter, &i.indexFilterRHPrealloc)
@@ -1554,7 +1542,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) closeInternal() error {
 	if invariants.Enabled && i.inPool {
 		panic("Close called on interator in pool")
 	}
-	i.iterStats.Close()
+
 	var err error
 	if i.closeHook != nil {
 		err = firstError(err, i.closeHook(i))
diff --git a/sstable/reader_iter_two_lvl.go b/sstable/reader_iter_two_lvl.go
index bcceef7a..437d12eb 100644
--- a/sstable/reader_iter_two_lvl.go
+++ b/sstable/reader_iter_two_lvl.go
@@ -160,10 +160,8 @@ func newColumnBlockTwoLevelIterator(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	filterBlockSizeLimit FilterBlockSizeLimit,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
+	env block.ReadEnv,
 	rp valblk.ReaderProvider,
-	bufferPool *block.BufferPool,
 ) (*twoLevelIteratorColumnBlocks, error) {
 	if r.err != nil {
 		return nil, r.err
@@ -174,7 +172,7 @@ func newColumnBlockTwoLevelIterator(
 	i := twoLevelIterColumnBlockPool.Get().(*twoLevelIteratorColumnBlocks)
 	i.secondLevel.init(ctx, r, v, transforms, lower, upper, filterer,
 		false, // Disable the use of the filter block in the second level.
-		stats, statsAccum, bufferPool)
+		env)
 	var getLazyValuer block.GetLazyValueForPrefixAndValueHandler
 	if r.Properties.NumValueBlocks > 0 {
 		// NB: we cannot avoid this ~248 byte allocation, since valueBlockReader
@@ -186,7 +184,7 @@ func newColumnBlockTwoLevelIterator(
 		// versions of keys, and therefore never expose a LazyValue that is
 		// separated to their callers, they can put this valueBlockReader into a
 		// sync.Pool.
-		i.secondLevel.vbReader = valblk.MakeReader(&i.secondLevel, rp, r.valueBIH, stats)
+		i.secondLevel.vbReader = valblk.MakeReader(&i.secondLevel, rp, r.valueBIH, env.Stats)
 		getLazyValuer = &i.secondLevel.vbReader
 		i.secondLevel.vbRH = r.blockReader.UsePreallocatedReadHandle(
 			objstorage.NoReadBefore, &i.secondLevel.vbRHPrealloc)
@@ -219,10 +217,8 @@ func newRowBlockTwoLevelIterator(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	filterBlockSizeLimit FilterBlockSizeLimit,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
+	env block.ReadEnv,
 	rp valblk.ReaderProvider,
-	bufferPool *block.BufferPool,
 ) (*twoLevelIteratorRowBlocks, error) {
 	if r.err != nil {
 		return nil, r.err
@@ -233,7 +229,7 @@ func newRowBlockTwoLevelIterator(
 	i := twoLevelIterRowBlockPool.Get().(*twoLevelIteratorRowBlocks)
 	i.secondLevel.init(ctx, r, v, transforms, lower, upper, filterer,
 		false, // Disable the use of the filter block in the second level.
-		stats, statsAccum, bufferPool)
+		env)
 	if r.tableFormat >= TableFormatPebblev3 {
 		if r.Properties.NumValueBlocks > 0 {
 			// NB: we cannot avoid this ~248 byte allocation, since valueBlockReader
@@ -245,7 +241,7 @@ func newRowBlockTwoLevelIterator(
 			// versions of keys, and therefore never expose a LazyValue that is
 			// separated to their callers, they can put this valueBlockReader into a
 			// sync.Pool.
-			i.secondLevel.vbReader = valblk.MakeReader(&i.secondLevel, rp, r.valueBIH, stats)
+			i.secondLevel.vbReader = valblk.MakeReader(&i.secondLevel, rp, r.valueBIH, env.Stats)
 			i.secondLevel.data.SetGetLazyValuer(&i.secondLevel.vbReader)
 			i.secondLevel.vbRH = r.blockReader.UsePreallocatedReadHandle(
 				objstorage.NoReadBefore, &i.secondLevel.vbRHPrealloc)
diff --git a/sstable/reader_virtual.go b/sstable/reader_virtual.go
index 298c76a2..7fc5bd68 100644
--- a/sstable/reader_virtual.go
+++ b/sstable/reader_virtual.go
@@ -94,13 +94,10 @@ func MakeVirtualReader(reader *Reader, p VirtualReaderParams) VirtualReader {
 
 // NewCompactionIter is the compaction iterator function for virtual readers.
 func (v *VirtualReader) NewCompactionIter(
-	transforms IterTransforms,
-	statsAccum block.IterStatsAccumulator,
-	rp valblk.ReaderProvider,
-	bufferPool *block.BufferPool,
+	transforms IterTransforms, env block.ReadEnv, rp valblk.ReaderProvider,
 ) (Iterator, error) {
 	return v.reader.newCompactionIter(
-		transforms, statsAccum, rp, &v.vState, bufferPool)
+		transforms, env, rp, &v.vState)
 }
 
 // NewPointIter returns an iterator for the point keys in the table.
@@ -118,13 +115,12 @@ func (v *VirtualReader) NewPointIter(
 	lower, upper []byte,
 	filterer *BlockPropertiesFilterer,
 	filterBlockSizeLimit FilterBlockSizeLimit,
-	stats *base.InternalIteratorStats,
-	statsAccum block.IterStatsAccumulator,
+	env block.ReadEnv,
 	rp valblk.ReaderProvider,
 ) (Iterator, error) {
 	return v.reader.newPointIter(
 		ctx, transforms, lower, upper, filterer, filterBlockSizeLimit,
-		stats, statsAccum, rp, &v.vState)
+		env, rp, &v.vState)
 }
 
 // ValidateBlockChecksumsOnBacking will call ValidateBlockChecksumsOnBacking on the underlying reader.
@@ -135,9 +131,9 @@ func (v *VirtualReader) ValidateBlockChecksumsOnBacking() error {
 
 // NewRawRangeDelIter wraps Reader.NewRawRangeDelIter.
 func (v *VirtualReader) NewRawRangeDelIter(
-	ctx context.Context, transforms FragmentIterTransforms,
+	ctx context.Context, transforms FragmentIterTransforms, env block.ReadEnv,
 ) (keyspan.FragmentIterator, error) {
-	iter, err := v.reader.NewRawRangeDelIter(ctx, transforms)
+	iter, err := v.reader.NewRawRangeDelIter(ctx, transforms, env)
 	if err != nil {
 		return nil, err
 	}
@@ -161,7 +157,7 @@ func (v *VirtualReader) NewRawRangeDelIter(
 
 // NewRawRangeKeyIter wraps Reader.NewRawRangeKeyIter.
 func (v *VirtualReader) NewRawRangeKeyIter(
-	ctx context.Context, transforms FragmentIterTransforms,
+	ctx context.Context, transforms FragmentIterTransforms, env block.ReadEnv,
 ) (keyspan.FragmentIterator, error) {
 	syntheticSeqNum := transforms.SyntheticSeqNum
 	if v.vState.isSharedIngested {
@@ -170,7 +166,7 @@ func (v *VirtualReader) NewRawRangeKeyIter(
 		// appropriate sequence number substitution below.
 		transforms.SyntheticSeqNum = 0
 	}
-	iter, err := v.reader.NewRawRangeKeyIter(ctx, transforms)
+	iter, err := v.reader.NewRawRangeKeyIter(ctx, transforms, env)
 	if err != nil {
 		return nil, err
 	}
diff --git a/sstable/rowblk_writer.go b/sstable/rowblk_writer.go
index 9c96aeb9..4ce0a9ef 100644
--- a/sstable/rowblk_writer.go
+++ b/sstable/rowblk_writer.go
@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"math"
 	"runtime"
+	"slices"
 	"sync"
 
 	"github.com/cockroachdb/errors"
@@ -1903,7 +1904,11 @@ func (w *RawRowWriter) rewriteSuffixes(
 				return errors.Wrap(err, "reading filter")
 			}
 			w.filter = copyFilterWriter{
-				origPolicyName: w.filter.policyName(), origMetaName: w.filter.metaName(), data: filterBlock,
+				origPolicyName: w.filter.policyName(),
+				origMetaName:   w.filter.metaName(),
+				// Clone the filter block, because readBlockBuf allows the
+				// returned byte slice to point directly into sst.
+				data: slices.Clone(filterBlock),
 			}
 		}
 	}
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index 9d421574..8387f189 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -204,7 +204,7 @@ func rewriteDataBlocksInParallel(
 }
 
 func rewriteRangeKeyBlockToWriter(r *Reader, w RawWriter, from, to []byte) error {
-	iter, err := r.NewRawRangeKeyIter(context.TODO(), NoFragmentTransforms)
+	iter, err := r.NewRawRangeKeyIter(context.TODO(), NoFragmentTransforms, block.NoReadEnv)
 	if err != nil {
 		return err
 	}
@@ -351,6 +351,10 @@ func NewMemReader(sst []byte, o ReaderOptions) (*Reader, error) {
 	return NewReader(context.Background(), newMemReader(sst), o)
 }
 
+// readBlockBuf may return a byte slice that points directly into sstBytes. If
+// the caller is going to expect that sstBytes remain stable, it should copy the
+// returned slice before writing it out to a objstorage.Writable which may
+// mangle it.
 func readBlockBuf(
 	sstBytes []byte, bh block.Handle, checksumType block.ChecksumType, buf []byte,
 ) ([]byte, []byte, error) {
diff --git a/table_stats.go b/table_stats.go
index d9430f2c..89b2f17d 100644
--- a/table_stats.go
+++ b/table_stats.go
@@ -913,7 +913,7 @@ func newCombinedDeletionKeyspanIter(
 	})
 	mIter.Init(comparer, transform, new(keyspanimpl.MergingBuffers))
 
-	iter, err := cr.NewRawRangeDelIter(context.TODO(), m.FragmentIterTransforms())
+	iter, err := cr.NewRawRangeDelIter(context.TODO(), m.FragmentIterTransforms(), block.NoReadEnv)
 	if err != nil {
 		return nil, err
 	}
@@ -964,7 +964,7 @@ func newCombinedDeletionKeyspanIter(
 		mIter.AddLevel(iter)
 	}
 
-	iter, err = cr.NewRawRangeKeyIter(context.TODO(), m.FragmentIterTransforms())
+	iter, err = cr.NewRawRangeKeyIter(context.TODO(), m.FragmentIterTransforms(), block.NoReadEnv)
 	if err != nil {
 		return nil, err
 	}
diff --git a/tool/db.go b/tool/db.go
index ee234817..b7664435 100644
--- a/tool/db.go
+++ b/tool/db.go
@@ -962,11 +962,9 @@ func (d *dbT) addProps(
 	opts := d.opts.MakeReaderOptions()
 	opts.Mergers = d.mergers
 	opts.Comparers = d.comparers
-	opts.SetInternal(sstableinternal.ReaderOptions{
-		CacheOpts: sstableinternal.CacheOptions{
-			FileNum: m.FileBacking.DiskFileNum,
-		},
-	})
+	opts.ReaderOptions.CacheOpts = sstableinternal.CacheOptions{
+		FileNum: m.FileBacking.DiskFileNum,
+	}
 	r, err := sstable.NewReader(ctx, f, opts)
 	if err != nil {
 		_ = f.Close()
diff --git a/tool/find.go b/tool/find.go
index a7cd02cb..85574471 100644
--- a/tool/find.go
+++ b/tool/find.go
@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/sstableinternal"
 	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/wal"
 	"github.com/spf13/cobra"
 )
@@ -441,12 +442,10 @@ func (f *findT) searchTables(stdout io.Writer, searchKey []byte, refs []findRef)
 			opts := f.opts.MakeReaderOptions()
 			opts.Comparers = f.comparers
 			opts.Mergers = f.mergers
-			opts.SetInternal(sstableinternal.ReaderOptions{
-				CacheOpts: sstableinternal.CacheOptions{
-					Cache:   cache,
-					FileNum: fl.DiskFileNum,
-				},
-			})
+			opts.CacheOpts = sstableinternal.CacheOptions{
+				Cache:   cache,
+				FileNum: fl.DiskFileNum,
+			}
 			readable, err := sstable.NewSimpleReadable(tf)
 			if err != nil {
 				return err
@@ -478,7 +477,7 @@ func (f *findT) searchTables(stdout io.Writer, searchKey []byte, refs []findRef)
 			// bit more work here to put them in a form that can be iterated in
 			// parallel with the point records.
 			rangeDelIter, err := func() (keyspan.FragmentIterator, error) {
-				iter, err := r.NewRawRangeDelIter(context.Background(), fragTransforms)
+				iter, err := r.NewRawRangeDelIter(context.Background(), fragTransforms, block.NoReadEnv)
 				if err != nil {
 					return nil, err
 				}
diff --git a/tool/sstable.go b/tool/sstable.go
index 3e0ae0f6..fa99824e 100644
--- a/tool/sstable.go
+++ b/tool/sstable.go
@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/rangedel"
 	"github.com/cockroachdb/pebble/internal/sstableinternal"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/spf13/cobra"
 )
@@ -152,11 +153,7 @@ func (s *sstableT) newReader(f vfs.File) (*sstable.Reader, error) {
 	o.Mergers = s.mergers
 	c := pebble.NewCache(128 << 20 /* 128 MB */)
 	defer c.Unref()
-	o.SetInternal(sstableinternal.ReaderOptions{
-		CacheOpts: sstableinternal.CacheOptions{
-			Cache: c,
-		},
-	})
+	o.CacheOpts = sstableinternal.CacheOptions{Cache: c}
 	return sstable.NewReader(context.Background(), readable, o)
 }
 
@@ -408,7 +405,7 @@ func (s *sstableT) runScan(cmd *cobra.Command, args []string) {
 		// bit more work here to put them in a form that can be iterated in
 		// parallel with the point records.
 		rangeDelIter, err := func() (keyspan.FragmentIterator, error) {
-			iter, err := r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms)
+			iter, err := r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms, block.NoReadEnv)
 			if err != nil {
 				return nil, err
 			}
@@ -510,7 +507,7 @@ func (s *sstableT) runScan(cmd *cobra.Command, args []string) {
 		}
 
 		// Handle range keys.
-		rkIter, err := r.NewRawRangeKeyIter(context.Background(), sstable.NoFragmentTransforms)
+		rkIter, err := r.NewRawRangeKeyIter(context.Background(), sstable.NoFragmentTransforms, block.NoReadEnv)
 		if err != nil {
 			fmt.Fprintf(stdout, "%s\n", err)
 			os.Exit(1)


Test patch:
diff --git a/cockroachkvs/cockroachkvs_bench_test.go b/cockroachkvs/cockroachkvs_bench_test.go
index 84f386b8..e3fdb4f3 100644
--- a/cockroachkvs/cockroachkvs_bench_test.go
+++ b/cockroachkvs/cockroachkvs_bench_test.go
@@ -100,14 +100,16 @@ func benchmarkRandSeekInSST(
 	defer c.Unref()
 	ctx := context.Background()
 	readerOpts := sstable.ReaderOptions{
+		ReaderOptions: block.ReaderOptions{
+			CacheOpts: sstableinternal.CacheOptions{
+				Cache:   c,
+				CacheID: c.NewID(),
+				FileNum: 1,
+			},
+		},
 		Comparer:   writerOpts.Comparer,
 		KeySchemas: sstable.MakeKeySchemas(&KeySchema),
 	}
-	readerOpts.SetInternalCacheOpts(sstableinternal.CacheOptions{
-		Cache:   c,
-		CacheID: c.NewID(),
-		FileNum: 1,
-	})
 	reader, err := sstable.NewReader(ctx, obj, readerOpts)
 	require.NoError(b, err)
 	defer reader.Close()
@@ -117,7 +119,7 @@ func benchmarkRandSeekInSST(
 	rp := sstable.MakeTrivialReaderProvider(reader)
 	iter, err := reader.NewPointIter(
 		ctx, sstable.NoTransforms, nil, nil, nil, sstable.NeverUseFilterBlock,
-		&stats, nil, rp)
+		block.ReadEnv{Stats: &stats, IterStats: nil}, rp)
 	require.NoError(b, err)
 	n := 0
 	for kv := iter.First(); kv != nil; kv = iter.Next() {
@@ -133,7 +135,7 @@ func benchmarkRandSeekInSST(
 		key := queryKeys[i%numQueryKeys]
 		iter, err := reader.NewPointIter(
 			ctx, sstable.NoTransforms, nil, nil, nil, sstable.NeverUseFilterBlock,
-			&stats, nil, rp)
+			block.ReadEnv{Stats: &stats, IterStats: nil}, rp)
 		if err != nil {
 			b.Fatal(err)
 		}
diff --git a/data_test.go b/data_test.go
index c43aff8e..7f179602 100644
--- a/data_test.go
+++ b/data_test.go
@@ -1177,11 +1177,11 @@ func runSSTablePropertiesCmd(t *testing.T, td *datadriven.TestData, d *DB) strin
 	// TODO(bananabrick): cacheOpts is used to set the file number on a Reader,
 	// and virtual sstables expect this file number to be set. Split out the
 	// opts into fileNum opts, and cache opts.
-	readerOpts.SetInternalCacheOpts(sstableinternal.CacheOptions{
+	readerOpts.CacheOpts = sstableinternal.CacheOptions{
 		Cache:   d.opts.Cache,
 		CacheID: 0,
 		FileNum: backingFileNum,
-	})
+	}
 	r, err := sstable.NewReader(context.Background(), readable, readerOpts)
 	if err != nil {
 		return err.Error()
diff --git a/level_checker_test.go b/level_checker_test.go
index dd2ac48c..6961c7a9 100644
--- a/level_checker_test.go
+++ b/level_checker_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
@@ -103,9 +104,9 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 
 	var fileNum FileNum
 	newIters :=
-		func(_ context.Context, file *manifest.FileMetadata, _ *IterOptions, _ internalIterOpts, _ iterKinds) (iterSet, error) {
+		func(_ context.Context, file *manifest.FileMetadata, _ *IterOptions, iio internalIterOpts, _ iterKinds) (iterSet, error) {
 			r := readers[file.FileNum]
-			rangeDelIter, err := r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms)
+			rangeDelIter, err := r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms, block.ReadEnv{Stats: iio.stats})
 			if err != nil {
 				return iterSet{}, err
 			}
@@ -240,7 +241,7 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 					Comparer:   testkeys.Comparer,
 					KeySchemas: sstable.KeySchemas{writerOpts.KeySchema.Name: writerOpts.KeySchema},
 				}
-				readerOpts.SetInternalCacheOpts(sstableinternal.CacheOptions{FileNum: base.DiskFileNum(fileNum - 1)})
+				readerOpts.CacheOpts = sstableinternal.CacheOptions{FileNum: base.DiskFileNum(fileNum - 1)}
 				r, err := sstable.NewReader(context.Background(), readable, readerOpts)
 				if err != nil {
 					return err.Error()
diff --git a/level_iter_test.go b/level_iter_test.go
index eb40b6a9..0d3f1d76 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -184,15 +185,14 @@ func (lt *levelIterTest) newIters(
 	if kinds.Point() {
 		iter, err := lt.readers[file.FileNum].NewPointIter(
 			ctx, transforms,
-			opts.LowerBound, opts.UpperBound, nil, sstable.AlwaysUseFilterBlock, iio.stats,
-			nil, sstable.MakeTrivialReaderProvider(lt.readers[file.FileNum]))
+			opts.LowerBound, opts.UpperBound, nil, sstable.AlwaysUseFilterBlock, block.ReadEnv{Stats: iio.stats, IterStats: nil}, sstable.MakeTrivialReaderProvider(lt.readers[file.FileNum]))
 		if err != nil {
 			return iterSet{}, errors.CombineErrors(err, set.CloseAll())
 		}
 		set.point = iter
 	}
 	if kinds.RangeDeletion() {
-		rangeDelIter, err := lt.readers[file.FileNum].NewRawRangeDelIter(context.Background(), file.FragmentIterTransforms())
+		rangeDelIter, err := lt.readers[file.FileNum].NewRawRangeDelIter(context.Background(), file.FragmentIterTransforms(), block.NoReadEnv)
 		if err != nil {
 			return iterSet{}, errors.CombineErrors(err, set.CloseAll())
 		}
@@ -543,9 +543,7 @@ func buildLevelIterTables(
 	c := NewCache(128 << 20 /* 128MB */)
 	defer c.Unref()
 	opts := sstable.ReaderOptions{Comparer: DefaultComparer}
-	opts.SetInternalCacheOpts(sstableinternal.CacheOptions{
-		Cache: c,
-	})
+	opts.CacheOpts = sstableinternal.CacheOptions{Cache: c}
 	readers := make([]*sstable.Reader, len(files))
 	for i := range files {
 		f, err := mem.Open(fmt.Sprintf("bench%d", i))
diff --git a/merging_iter_test.go b/merging_iter_test.go
index d8378ee5..afd766af 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/testutils/indenttree"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -169,7 +170,7 @@ func TestMergingIterDataDriven(t *testing.T) {
 			var err error
 			r := readers[file.FileNum]
 			if kinds.RangeDeletion() {
-				set.rangeDeletion, err = r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms)
+				set.rangeDeletion, err = r.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms, block.ReadEnv{Stats: iio.stats})
 				if err != nil {
 					return iterSet{}, errors.CombineErrors(err, set.CloseAll())
 				}
@@ -178,8 +179,7 @@ func TestMergingIterDataDriven(t *testing.T) {
 				set.point, err = r.NewPointIter(
 					context.Background(),
 					sstable.NoTransforms,
-					opts.GetLowerBound(), opts.GetUpperBound(), nil, sstable.AlwaysUseFilterBlock, iio.stats,
-					nil, sstable.MakeTrivialReaderProvider(r))
+					opts.GetLowerBound(), opts.GetUpperBound(), nil, sstable.AlwaysUseFilterBlock, block.ReadEnv{Stats: iio.stats, IterStats: nil}, sstable.MakeTrivialReaderProvider(r))
 				if err != nil {
 					return iterSet{}, errors.CombineErrors(err, set.CloseAll())
 				}
@@ -372,9 +372,7 @@ func buildMergingIterTables(
 	defer c.Unref()
 
 	var opts sstable.ReaderOptions
-	opts.SetInternalCacheOpts(sstableinternal.CacheOptions{
-		Cache: c,
-	})
+	opts.CacheOpts = sstableinternal.CacheOptions{Cache: c}
 
 	readers := make([]*sstable.Reader, len(files))
 	for i := range files {
@@ -617,9 +615,7 @@ func buildLevelsForMergingIterSeqSeek(
 	defer c.Unref()
 
 	opts := sstable.ReaderOptions{Comparer: DefaultComparer}
-	opts.SetInternalCacheOpts(sstableinternal.CacheOptions{
-		Cache: c,
-	})
+	opts.CacheOpts = sstableinternal.CacheOptions{Cache: c}
 
 	if writeBloomFilters {
 		opts.Filters = make(map[string]FilterPolicy)
@@ -671,14 +667,14 @@ func buildMergingIter(readers [][]*sstable.Reader, levelSlices []manifest.LevelS
 		levelIndex := i
 		level := len(readers) - 1 - i
 		newIters := func(
-			_ context.Context, file *manifest.FileMetadata, opts *IterOptions, _ internalIterOpts, _ iterKinds,
+			_ context.Context, file *manifest.FileMetadata, opts *IterOptions, iio internalIterOpts, _ iterKinds,
 		) (iterSet, error) {
 			iter, err := readers[levelIndex][file.FileNum].NewIter(
 				sstable.NoTransforms, opts.LowerBound, opts.UpperBound)
 			if err != nil {
 				return iterSet{}, err
 			}
-			rdIter, err := readers[levelIndex][file.FileNum].NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms)
+			rdIter, err := readers[levelIndex][file.FileNum].NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms, block.ReadEnv{Stats: iio.stats})
 			if err != nil {
 				iter.Close()
 				return iterSet{}, err
diff --git a/objstorage/test_utils.go b/objstorage/test_utils.go
index 8326b42c..8e880a38 100644
--- a/objstorage/test_utils.go
+++ b/objstorage/test_utils.go
@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"context"
 
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/pkg/errors"
 )
 
@@ -32,6 +33,13 @@ func (f *MemObj) Abort() { f.buf.Reset() }
 // Write is part of the Writable interface.
 func (f *MemObj) Write(p []byte) error {
 	_, err := f.buf.Write(p)
+	// Write is allowed to mangle the buffer. Do it sometimes in invariant
+	// builds to catch callers that don't handle this.
+	if invariants.Enabled && invariants.Sometimes(1) {
+		for i := range p {
+			p[i] = 0xFF
+		}
+	}
 	return err
 }
 
diff --git a/sstable/blob/blob_test.go b/sstable/blob/blob_test.go
index c6c4f44a..3003ae47 100644
--- a/sstable/blob/blob_test.go
+++ b/sstable/blob/blob_test.go
@@ -58,10 +58,11 @@ func TestBlobWriter(t *testing.T) {
 			fmt.Fprintf(&buf, "  FileLen: %d\n", stats.FileLen)
 			return buf.String()
 		case "open":
-			r, err := NewFileReader(context.Background(), obj)
+			r, err := NewFileReader(context.Background(), obj, FileReaderOptions{})
 			if err != nil {
 				t.Fatal(err)
 			}
+			defer r.Close()
 			fmt.Fprintf(&buf, "TableFormat: %s\n", r.footer.format)
 			fmt.Fprintf(&buf, "ChecksumType: %s\n", r.footer.checksum)
 			fmt.Fprintf(&buf, "IndexHandle: %s\n", r.footer.indexHandle.String())
diff --git a/sstable/block_property_test.go b/sstable/block_property_test.go
index 2a186da9..026be11d 100644
--- a/sstable/block_property_test.go
+++ b/sstable/block_property_test.go
@@ -988,8 +988,7 @@ func TestBlockProperties(t *testing.T) {
 			}
 			iter, err := r.NewPointIter(
 				context.Background(),
-				NoTransforms, lower, upper, filterer, NeverUseFilterBlock, &stats,
-				nil, MakeTrivialReaderProvider(r))
+				NoTransforms, lower, upper, filterer, NeverUseFilterBlock, block.ReadEnv{Stats: &stats, IterStats: nil}, MakeTrivialReaderProvider(r))
 			if err != nil {
 				return err.Error()
 			}
@@ -1072,8 +1071,7 @@ func TestBlockProperties_BoundLimited(t *testing.T) {
 			}
 			iter, err := r.NewPointIter(
 				context.Background(),
-				NoTransforms, lower, upper, filterer, NeverUseFilterBlock, &stats,
-				nil, MakeTrivialReaderProvider(r))
+				NoTransforms, lower, upper, filterer, NeverUseFilterBlock, block.ReadEnv{Stats: &stats, IterStats: nil}, MakeTrivialReaderProvider(r))
 			if err != nil {
 				return err.Error()
 			}
diff --git a/sstable/copier_test.go b/sstable/copier_test.go
index f5b7ce37..628470d5 100644
--- a/sstable/copier_test.go
+++ b/sstable/copier_test.go
@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/internal/sstableinternal"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable/block"
@@ -107,12 +108,17 @@ func TestCopySpan(t *testing.T) {
 				}
 			}
 			rOpts := ReaderOptions{
+				ReaderOptions: block.ReaderOptions{
+					CacheOpts: sstableinternal.CacheOptions{
+						Cache:   blockCache,
+						CacheID: cacheID,
+						FileNum: base.DiskFileNum(fileNameToNum[d.CmdArgs[0].Key]),
+					},
+				},
 				Comparer:   testkeys.Comparer,
 				KeySchemas: KeySchemas{keySchema.Name: &keySchema},
 			}
-			rOpts.internal.CacheOpts.Cache = blockCache
-			rOpts.internal.CacheOpts.CacheID = cacheID
-			rOpts.internal.CacheOpts.FileNum = base.DiskFileNum(fileNameToNum[d.CmdArgs[0].Key])
+
 			r, err := NewReader(context.TODO(), readable, rOpts)
 			defer r.Close()
 			if err != nil {
@@ -156,12 +162,16 @@ func TestCopySpan(t *testing.T) {
 				return err.Error()
 			}
 			rOpts := ReaderOptions{
+				ReaderOptions: block.ReaderOptions{
+					CacheOpts: sstableinternal.CacheOptions{
+						Cache:   blockCache,
+						CacheID: cacheID,
+						FileNum: base.DiskFileNum(fileNameToNum[d.CmdArgs[0].Key]),
+					},
+				},
 				Comparer:   testkeys.Comparer,
 				KeySchemas: KeySchemas{keySchema.Name: &keySchema},
 			}
-			rOpts.internal.CacheOpts.Cache = blockCache
-			rOpts.internal.CacheOpts.CacheID = cacheID
-			rOpts.internal.CacheOpts.FileNum = base.DiskFileNum(fileNameToNum[inputFile])
 			r, err := NewReader(context.TODO(), readable, rOpts)
 			if err != nil {
 				return err.Error()
diff --git a/sstable/data_test.go b/sstable/data_test.go
index 5c9addf9..3ef693af 100644
--- a/sstable/data_test.go
+++ b/sstable/data_test.go
@@ -179,9 +179,7 @@ func openReader(obj *objstorage.MemObj, writerOpts *WriterOptions, cacheSize int
 	if cacheSize > 0 {
 		c := cache.New(int64(cacheSize))
 		defer c.Unref()
-		readerOpts.SetInternal(sstableinternal.ReaderOptions{
-			CacheOpts: sstableinternal.CacheOptions{Cache: c},
-		})
+		readerOpts.CacheOpts = sstableinternal.CacheOptions{Cache: c}
 	}
 	r, err := NewMemReader(obj.Data(), readerOpts)
 	if err != nil {
diff --git a/sstable/random_test.go b/sstable/random_test.go
index 36bc9ae0..25e2d490 100644
--- a/sstable/random_test.go
+++ b/sstable/random_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/bytealloc"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/errorfs"
@@ -110,8 +111,7 @@ func runErrorInjectionTest(t *testing.T, seed int64) {
 		nil /* lower TODO */, nil, /* upper TODO */
 		filterer,
 		filterBlockSizeLimit,
-		&stats,
-		nil, /* IterStatsAccumulator */
+		block.ReadEnv{Stats: &stats, IterStats: nil},
 		MakeTrivialReaderProvider(r),
 	)
 	require.NoError(t, err)
diff --git a/sstable/reader_iter_test.go b/sstable/reader_iter_test.go
index 505fd5f8..9d52ce3b 100644
--- a/sstable/reader_iter_test.go
+++ b/sstable/reader_iter_test.go
@@ -62,10 +62,8 @@ func TestIteratorErrorOnInit(t *testing.T) {
 				NoTransforms,
 				nil /* lower */, nil, /* upper */
 				nil /* filterer */, NeverUseFilterBlock,
-				&stats,
-				nil, /* statsAccum */
+				block.ReadEnv{Stats: &stats, BufferPool: &pool},
 				MakeTrivialReaderProvider(r),
-				&pool,
 			)
 			require.Error(t, err)
 		} else {
@@ -76,10 +74,8 @@ func TestIteratorErrorOnInit(t *testing.T) {
 				NoTransforms,
 				nil /* lower */, nil, /* upper */
 				nil /* filterer */, NeverUseFilterBlock,
-				&stats,
-				nil, /* statsAccum */
+				block.ReadEnv{Stats: &stats, BufferPool: &pool},
 				MakeTrivialReaderProvider(r),
-				&pool,
 			)
 			require.Error(t, err)
 		}
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index efdc97cb..b3a7314f 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -235,7 +235,7 @@ func runVirtualReaderTest(t *testing.T, path string, blockSize, indexBlockSize i
 			transforms := IterTransforms{
 				SyntheticPrefixAndSuffix: block.MakeSyntheticPrefixAndSuffix(nil, syntheticSuffix),
 			}
-			iter, err := v.NewCompactionIter(transforms, nil, rp, &bp)
+			iter, err := v.NewCompactionIter(transforms, block.ReadEnv{BufferPool: &bp}, rp)
 			if err != nil {
 				return err.Error()
 			}
@@ -277,7 +277,7 @@ func runVirtualReaderTest(t *testing.T, path string, blockSize, indexBlockSize i
 			transforms := FragmentIterTransforms{
 				SyntheticPrefixAndSuffix: block.MakeSyntheticPrefixAndSuffix(nil, syntheticSuffix),
 			}
-			iter, err := v.NewRawRangeDelIter(context.Background(), transforms)
+			iter, err := v.NewRawRangeDelIter(context.Background(), transforms, block.NoReadEnv)
 			if err != nil {
 				return err.Error()
 			}
@@ -303,7 +303,7 @@ func runVirtualReaderTest(t *testing.T, path string, blockSize, indexBlockSize i
 			transforms := FragmentIterTransforms{
 				SyntheticPrefixAndSuffix: block.MakeSyntheticPrefixAndSuffix(nil, syntheticSuffix),
 			}
-			iter, err := v.NewRawRangeKeyIter(context.Background(), transforms)
+			iter, err := v.NewRawRangeKeyIter(context.Background(), transforms, block.NoReadEnv)
 			if err != nil {
 				return err.Error()
 			}
@@ -362,7 +362,7 @@ func runVirtualReaderTest(t *testing.T, path string, blockSize, indexBlockSize i
 			}
 			iter, err := v.NewPointIter(
 				context.Background(), transforms, lower, upper, filterer, NeverUseFilterBlock,
-				&stats, nil, MakeTrivialReaderProvider(r))
+				block.ReadEnv{Stats: &stats, IterStats: nil}, MakeTrivialReaderProvider(r))
 			if err != nil {
 				return err.Error()
 			}
@@ -755,8 +755,7 @@ func runTestReader(t *testing.T, o WriterOptions, dir string, r *Reader, printVa
 					nil, /* upper */
 					filterer,
 					AlwaysUseFilterBlock,
-					&stats,
-					nil,
+					block.ReadEnv{Stats: &stats, IterStats: nil},
 					MakeTrivialReaderProvider(r),
 				)
 				if err != nil {
@@ -899,7 +898,7 @@ func TestCompactionIteratorSetupForCompaction(t *testing.T) {
 				var pool block.BufferPool
 				pool.Init(5)
 				citer, err := r.NewCompactionIter(
-					NoTransforms, nil, MakeTrivialReaderProvider(r), &pool)
+					NoTransforms, block.ReadEnv{BufferPool: &pool}, MakeTrivialReaderProvider(r))
 				require.NoError(t, err)
 				switch i := citer.(type) {
 				case *singleLevelIteratorRowBlocks:
@@ -957,7 +956,7 @@ func TestReadaheadSetupForV3TablesWithMultipleVersions(t *testing.T) {
 		pool.Init(5)
 		defer pool.Release()
 		citer, err := r.NewCompactionIter(
-			NoTransforms, nil, MakeTrivialReaderProvider(r), &pool)
+			NoTransforms, block.ReadEnv{BufferPool: &pool}, MakeTrivialReaderProvider(r))
 		require.NoError(t, err)
 		defer citer.Close()
 		i := citer.(*singleLevelIteratorRowBlocks)
@@ -1254,7 +1253,7 @@ func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
 				SyntheticPrefixAndSuffix: block.MakeSyntheticPrefixAndSuffix(syntheticPrefix, syntheticSuffix),
 			},
 			nil, nil, nil,
-			AlwaysUseFilterBlock, nil, nil,
+			AlwaysUseFilterBlock, block.NoReadEnv,
 			MakeTrivialReaderProvider(eReader), &virtualState{
 				lower: base.MakeInternalKey([]byte("_"), base.SeqNumMax, base.InternalKeyKindSet),
 				upper: base.MakeRangeDeleteSentinelKey([]byte("~~~~~~~~~~~~~~~~")),
@@ -1752,7 +1751,7 @@ func buildTestTableWithProvider(
 	c := cache.New(128 << 20)
 	defer c.Unref()
 	r, err := NewReader(context.Background(), f1, ReaderOptions{
-		internal: sstableinternal.ReaderOptions{
+		ReaderOptions: block.ReaderOptions{
 			CacheOpts: sstableinternal.CacheOptions{
 				Cache: c,
 			},
@@ -1795,7 +1794,7 @@ func buildBenchmarkTable(
 	c := cache.New(128 << 20)
 	defer c.Unref()
 	r, err := newReader(f1, ReaderOptions{
-		internal: sstableinternal.ReaderOptions{
+		ReaderOptions: block.ReaderOptions{
 			CacheOpts: sstableinternal.CacheOptions{
 				Cache: c,
 			},
@@ -2082,7 +2081,7 @@ func BenchmarkIteratorScanManyVersions(b *testing.B) {
 		require.NoError(b, err)
 		r, err := newReader(f0, ReaderOptions{
 			Comparer: testkeys.Comparer,
-			internal: sstableinternal.ReaderOptions{
+			ReaderOptions: block.ReaderOptions{
 				CacheOpts: sstableinternal.CacheOptions{
 					Cache: c,
 				},
@@ -2202,7 +2201,7 @@ func BenchmarkIteratorScanNextPrefix(b *testing.B) {
 		require.NoError(b, err)
 		r, err = newReader(f0, ReaderOptions{
 			Comparer: testkeys.Comparer,
-			internal: sstableinternal.ReaderOptions{
+			ReaderOptions: block.ReaderOptions{
 				CacheOpts: sstableinternal.CacheOptions{
 					Cache: c,
 				},
@@ -2364,7 +2363,7 @@ func BenchmarkIteratorScanObsolete(b *testing.B) {
 		require.NoError(b, err)
 		r, err := newReader(f0, ReaderOptions{
 			Comparer: testkeys.Comparer,
-			internal: sstableinternal.ReaderOptions{
+			ReaderOptions: block.ReaderOptions{
 				CacheOpts: sstableinternal.CacheOptions{
 					Cache: c,
 				},
@@ -2401,7 +2400,7 @@ func BenchmarkIteratorScanObsolete(b *testing.B) {
 								transforms := IterTransforms{HideObsoletePoints: hideObsoletePoints}
 								iter, err := r.NewPointIter(
 									context.Background(), transforms, nil, nil, filterer,
-									AlwaysUseFilterBlock, nil, nil,
+									AlwaysUseFilterBlock, block.NoReadEnv,
 									MakeTrivialReaderProvider(r))
 								require.NoError(b, err)
 								b.ResetTimer()
diff --git a/sstable/suffix_rewriter_test.go b/sstable/suffix_rewriter_test.go
index def874fe..51777660 100644
--- a/sstable/suffix_rewriter_test.go
+++ b/sstable/suffix_rewriter_test.go
@@ -91,72 +91,82 @@ func TestRewriteSuffixProps(t *testing.T) {
 			var sstBytes [2][]byte
 			for i, byBlocks := range []bool{false, true} {
 				t.Run(fmt.Sprintf("byBlocks=%v", byBlocks), func(t *testing.T) {
-					rewrittenSST := &objstorage.MemObj{}
-					if byBlocks {
-						_, rewriteFormat, err := rewriteKeySuffixesInBlocks(
-							r, sst, rewrittenSST, rwOpts, from, to, 8)
-						require.NoError(t, err)
-						// rewriteFormat is equal to the original format, since
-						// rwOpts.TableFormat is ignored.
-						require.Equal(t, wOpts.TableFormat, rewriteFormat)
-					} else {
-						_, err := RewriteKeySuffixesViaWriter(r, rewrittenSST, rwOpts, from, to)
-						require.NoError(t, err)
-					}
+					fn := func() {
+						rewrittenSST := &objstorage.MemObj{}
+						if byBlocks {
+							_, rewriteFormat, err := rewriteKeySuffixesInBlocks(
+								r, sst, rewrittenSST, rwOpts, from, to, 8)
+							require.NoError(t, err)
+							// rewriteFormat is equal to the original format, since
+							// rwOpts.TableFormat is ignored.
+							require.Equal(t, wOpts.TableFormat, rewriteFormat)
+						} else {
+							_, err := RewriteKeySuffixesViaWriter(r, rewrittenSST, rwOpts, from, to)
+							require.NoError(t, err)
+						}
 
-					sstBytes[i] = rewrittenSST.Data()
-					// Check that a reader on the rewritten STT has the expected props.
-					rRewritten, err := NewMemReader(rewrittenSST.Data(), readerOpts)
-					require.NoError(t, err)
-					defer rRewritten.Close()
+						sstBytes[i] = rewrittenSST.Data()
+						// Check that a reader on the rewritten STT has the expected props.
+						rRewritten, err := NewMemReader(rewrittenSST.Data(), readerOpts)
+						require.NoError(t, err)
+						defer rRewritten.Close()
 
-					foundValues := make(map[string][]byte)
-					for k, v := range rRewritten.Properties.UserProperties {
-						if k == "obsolete-key" {
-							continue
+						foundValues := make(map[string][]byte)
+						for k, v := range rRewritten.Properties.UserProperties {
+							if k == "obsolete-key" {
+								continue
+							}
+							require.Contains(t, newCollectors, k)
+							require.Equal(t, uint8(slices.Index(newCollectors, k)), v[0], "shortID should match")
+							foundValue := []byte(v[1:])
+							foundValues[k] = foundValue
+							t.Logf("%q => %q", k, foundValues[k])
 						}
-						require.Contains(t, newCollectors, k)
-						require.Equal(t, uint8(slices.Index(newCollectors, k)), v[0], "shortID should match")
-						foundValue := []byte(v[1:])
-						foundValues[k] = foundValue
-						t.Logf("%q => %q", k, foundValues[k])
-					}
-					require.Equal(t, expectedProps, foundValues)
-					require.False(t, rRewritten.Properties.IsStrictObsolete)
+						require.Equal(t, expectedProps, foundValues)
+						require.False(t, rRewritten.Properties.IsStrictObsolete)
 
-					// Compare the block level props from the data blocks in the layout,
-					// only if we did not do a rewrite from one format to another. If the
-					// format changes, the block boundaries change slightly.
-					if !byBlocks && wOpts.TableFormat != rwOpts.TableFormat {
-						return
-					}
-					layout, err := r.Layout()
-					require.NoError(t, err)
-					newLayout, err := rRewritten.Layout()
-					require.NoError(t, err)
-
-					for i := range layout.Data {
-						oldProps := make([][]byte, len(wOpts.BlockPropertyCollectors))
-						oldDecoder := makeBlockPropertiesDecoder(len(oldProps), layout.Data[i].Props)
-						for !oldDecoder.Done() {
-							id, val, err := oldDecoder.Next()
-							require.NoError(t, err)
-							oldProps[id] = val
+						// Compare the block level props from the data blocks in the layout,
+						// only if we did not do a rewrite from one format to another. If the
+						// format changes, the block boundaries change slightly.
+						if !byBlocks && wOpts.TableFormat != rwOpts.TableFormat {
+							return
 						}
-						newProps := make([][]byte, len(newCollectors))
-						newDecoder := makeBlockPropertiesDecoder(len(newProps), newLayout.Data[i].Props)
-						for !newDecoder.Done() {
-							id, val, err := newDecoder.Next()
-							require.NoError(t, err)
-							newProps[id] = val
-							switch newCollectors[id] {
-							case "count":
-								require.Equal(t, oldProps[slices.Index(originalCollectors, "count")], val)
-							default:
-								require.Equal(t, allExpectedProps[newCollectors[id]], val)
+						layout, err := r.Layout()
+						require.NoError(t, err)
+						newLayout, err := rRewritten.Layout()
+						require.NoError(t, err)
+
+						for i := range layout.Data {
+							oldProps := make([][]byte, len(wOpts.BlockPropertyCollectors))
+							oldDecoder := makeBlockPropertiesDecoder(len(oldProps), layout.Data[i].Props)
+							for !oldDecoder.Done() {
+								id, val, err := oldDecoder.Next()
+								require.NoError(t, err)
+								oldProps[id] = val
+							}
+							newProps := make([][]byte, len(newCollectors))
+							newDecoder := makeBlockPropertiesDecoder(len(newProps), newLayout.Data[i].Props)
+							for !newDecoder.Done() {
+								id, val, err := newDecoder.Next()
+								require.NoError(t, err)
+								newProps[id] = val
+								switch newCollectors[id] {
+								case "count":
+									require.Equal(t, oldProps[slices.Index(originalCollectors, "count")], val)
+								default:
+									require.Equal(t, allExpectedProps[newCollectors[id]], val)
+								}
 							}
 						}
 					}
+					// Perform the rewrite multiple times. This helps ensure
+					// idempotence. This helps catch bugs in suffix rewriting
+					// that might mangle the in-memory source sstable's buffer
+					// by improperly assuming that Write/WriteTo leaves the
+					// input buffer unmodified.
+					for j := 0; j < 5; j++ {
+						fn()
+					}
 				})
 			}
 			if wOpts.TableFormat == rwOpts.TableFormat {
diff --git a/sstable/test_utils.go b/sstable/test_utils.go
index 578fdf6f..c570bcc7 100644
--- a/sstable/test_utils.go
+++ b/sstable/test_utils.go
@@ -11,6 +11,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 // ReadAll returns all point keys, range del spans, and range key spans from an
@@ -32,14 +33,14 @@ func ReadAll(
 	}
 
 	ctx := context.Background()
-	if rangeDelIter := testutils.CheckErr(reader.NewRawRangeDelIter(ctx, NoFragmentTransforms)); rangeDelIter != nil {
+	if rangeDelIter := testutils.CheckErr(reader.NewRawRangeDelIter(ctx, NoFragmentTransforms, block.NoReadEnv)); rangeDelIter != nil {
 		defer rangeDelIter.Close()
 		for s := testutils.CheckErr(rangeDelIter.First()); s != nil; s = testutils.CheckErr(rangeDelIter.Next()) {
 			rangeDels = append(rangeDels, s.Clone())
 		}
 	}
 
-	if rangeKeyIter := testutils.CheckErr(reader.NewRawRangeKeyIter(ctx, NoFragmentTransforms)); rangeKeyIter != nil {
+	if rangeKeyIter := testutils.CheckErr(reader.NewRawRangeKeyIter(ctx, NoFragmentTransforms, block.NoReadEnv)); rangeKeyIter != nil {
 		defer rangeKeyIter.Close()
 		for s := testutils.CheckErr(rangeKeyIter.First()); s != nil; s = testutils.CheckErr(rangeKeyIter.Next()) {
 			rangeKeys = append(rangeKeys, s.Clone())
diff --git a/sstable/writer_rangekey_test.go b/sstable/writer_rangekey_test.go
index 232eec61..63384ede 100644
--- a/sstable/writer_rangekey_test.go
+++ b/sstable/writer_rangekey_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
@@ -121,7 +122,7 @@ func TestWriter_RangeKeys(t *testing.T) {
 						return err.Error()
 					}
 
-					iter, err := r.NewRawRangeKeyIter(context.Background(), NoFragmentTransforms)
+					iter, err := r.NewRawRangeKeyIter(context.Background(), NoFragmentTransforms, block.NoReadEnv)
 					if err != nil {
 						return err.Error()
 					}
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 4867c468..a5897c6c 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -239,7 +239,7 @@ func runDataDriven(t *testing.T, file string, tableFormat TableFormat, paralleli
 			return buf.String()
 
 		case "scan-range-del":
-			iter, err := r.NewRawRangeDelIter(context.Background(), NoFragmentTransforms)
+			iter, err := r.NewRawRangeDelIter(context.Background(), NoFragmentTransforms, block.NoReadEnv)
 			if err != nil {
 				return err.Error()
 			}
@@ -259,7 +259,7 @@ func runDataDriven(t *testing.T, file string, tableFormat TableFormat, paralleli
 			return buf.String()
 
 		case "scan-range-key":
-			iter, err := r.NewRawRangeKeyIter(context.Background(), NoFragmentTransforms)
+			iter, err := r.NewRawRangeKeyIter(context.Background(), NoFragmentTransforms, block.NoReadEnv)
 			if err != nil {
 				return err.Error()
 			}
@@ -708,10 +708,8 @@ func TestWriterClearCache(t *testing.T) {
 	defer cacheOpts.Cache.Unref()
 
 	readerOpts := ReaderOptions{
-		Comparer: testkeys.Comparer,
-		internal: sstableinternal.ReaderOptions{
-			CacheOpts: cacheOpts,
-		},
+		ReaderOptions: block.ReaderOptions{CacheOpts: cacheOpts},
+		Comparer:      testkeys.Comparer,
 	}
 
 	writerOpts := WriterOptions{
diff --git a/testdata/merging_iter b/testdata/merging_iter
index feb6e9cf..1de103a7 100644
--- a/testdata/merging_iter
+++ b/testdata/merging_iter
@@ -50,7 +50,7 @@ c#27,SET:27
 e#10,SET:10
 g#20,SET:20
 .
-{BlockBytes:116 BlockBytesInCache:0 BlockReadDuration:0s KeyBytes:4 ValueBytes:8 PointCount:4 PointsCoveredByRangeTombstones:0 SeparatedPointValue:{Count:0 ValueBytes:0 ValueBytesFetched:0}}
+{BlockBytes:200 BlockBytesInCache:0 BlockReadDuration:0s KeyBytes:4 ValueBytes:8 PointCount:4 PointsCoveredByRangeTombstones:0 SeparatedPointValue:{Count:0 ValueBytes:0 ValueBytesFetched:0}}
 {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s KeyBytes:0 ValueBytes:0 PointCount:0 PointsCoveredByRangeTombstones:0 SeparatedPointValue:{Count:0 ValueBytes:0 ValueBytesFetched:0}}
 
 # seekGE() should not allow the rangedel to act on points in the lower sstable that are after it.
@@ -651,7 +651,7 @@ next
 stats
 ----
 a#30,SET:30
-{BlockBytes:97 BlockBytesInCache:0 BlockReadDuration:0s KeyBytes:1 ValueBytes:2 PointCount:1 PointsCoveredByRangeTombstones:0 SeparatedPointValue:{Count:0 ValueBytes:0 ValueBytesFetched:0}}
+{BlockBytes:139 BlockBytesInCache:0 BlockReadDuration:0s KeyBytes:1 ValueBytes:2 PointCount:1 PointsCoveredByRangeTombstones:0 SeparatedPointValue:{Count:0 ValueBytes:0 ValueBytesFetched:0}}
 {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s KeyBytes:0 ValueBytes:0 PointCount:0 PointsCoveredByRangeTombstones:0 SeparatedPointValue:{Count:0 ValueBytes:0 ValueBytesFetched:0}}
 f#21,SET:21
 {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s KeyBytes:5 ValueBytes:10 PointCount:5 PointsCoveredByRangeTombstones:4 SeparatedPointValue:{Count:0 ValueBytes:0 ValueBytesFetched:0}}
diff --git a/testdata/metrics b/testdata/metrics
index adf23bd9..a9172d45 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -83,7 +83,7 @@ Ingestions: 0  as flushable: 0 (0B in 0 tables)
 Cgo memory usage: 0B  block cache: 0B (data: 0B, maps: 0B, entries: 0B)  memtables: 0B
 Iter category stats:
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
-                   b,     latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
+                   b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
 
 disk-usage
 ----
@@ -143,8 +143,8 @@ Cgo memory usage: 0B  block cache: 0B (data: 0B, maps: 0B, entries: 0B)  memtabl
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:88 BlockBytesInCache:44 BlockReadDuration:10ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
-                   b,     latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
-                   c, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
+                   b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
+                   c, non-latency: {BlockBytes:44 BlockBytesInCache:44 BlockReadDuration:0s}
 
 disk-usage
 ----
@@ -189,8 +189,8 @@ Cgo memory usage: 0B  block cache: 0B (data: 0B, maps: 0B, entries: 0B)  memtabl
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:88 BlockBytesInCache:44 BlockReadDuration:10ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
-                   b,     latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
-                   c, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
+                   b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
+                   c, non-latency: {BlockBytes:44 BlockBytesInCache:44 BlockReadDuration:0s}
 
 # Closing iter c will release one of the zombie sstables. The other
 # zombie sstable is still referenced by iter b.
@@ -232,7 +232,7 @@ Cgo memory usage: 0B  block cache: 0B (data: 0B, maps: 0B, entries: 0B)  memtabl
 Iter category stats:
    pebble-compaction, non-latency: {BlockBytes:88 BlockBytesInCache:44 BlockReadDuration:10ms}
                    a, non-latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
-                   b,     latency: {BlockBytes:0 BlockBytesInCache:0 BlockReadDuration:0s}
+                   b,     latency: {BlockBytes:44 BlockBytesInCache:0 BlockReadDuration:10ms}
                    c, non-latency: {BlockBytes:44 BlockBytesInCache:44 BlockReadDuration:0s}
 
 disk-usage