Repository: cockroachdb/pebble
Pull request #4033 (2024-10-10 20:41:38)

sstable: cleanup readBlock usage
We move the stats and bufferPool arguments into their own
`readBlockEnv` struct. We add specific functions for different block
types.

A functional change is that now we use the buffer pool for the
top-level index block and filter block (this looks like an omission in
the existing code). This changed caused failures in some tests that
were not releasing a pool, which are now fixed.

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

RaduBerinde: TFTR!

Patch:
diff --git a/sstable/colblk/data_block.go b/sstable/colblk/data_block.go
index 3a25c3de..56ee83e4 100644
--- a/sstable/colblk/data_block.go
+++ b/sstable/colblk/data_block.go
@@ -372,9 +372,9 @@ func (ks *defaultKeySeeker) MaterializeUserKey(keyIter *PrefixBytesIter, prevRow
 		ks.prefixes.SetAt(keyIter, row)
 	}
 	suffix := ks.suffixes.At(row)
-	res := keyIter.buf[:len(keyIter.buf)+len(suffix)]
+	res := keyIter.Buf[:len(keyIter.Buf)+len(suffix)]
 	memmove(
-		unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(keyIter.buf)))+uintptr(len(keyIter.buf))),
+		unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(keyIter.Buf)))+uintptr(len(keyIter.Buf))),
 		unsafe.Pointer(unsafe.SliceData(suffix)),
 		uintptr(len(suffix)),
 	)
@@ -390,9 +390,9 @@ func (ks *defaultKeySeeker) MaterializeUserKeyWithSyntheticSuffix(
 	} else {
 		ks.prefixes.SetAt(keyIter, row)
 	}
-	res := keyIter.buf[:len(keyIter.buf)+len(suffix)]
+	res := keyIter.Buf[:len(keyIter.Buf)+len(suffix)]
 	memmove(
-		unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(keyIter.buf)))+uintptr(len(keyIter.buf))),
+		unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(keyIter.Buf)))+uintptr(len(keyIter.Buf))),
 		unsafe.Pointer(unsafe.SliceData(suffix)),
 		uintptr(len(suffix)),
 	)
@@ -698,8 +698,8 @@ func (rw *DataBlockRewriter) RewriteSuffixes(
 	// Allocate a keyIter buffer that's large enough to hold the largest user
 	// key in the block with 1 byte to spare (so that pointer arithmetic is
 	// never pointing beyond the allocation, which would violate Go rules).
-	if cap(rw.prefixBytesIter.buf) < int(rw.decoder.maximumKeyLength)+1 {
-		rw.prefixBytesIter.buf = make([]byte, rw.decoder.maximumKeyLength+1)
+	if cap(rw.prefixBytesIter.Buf) < int(rw.decoder.maximumKeyLength)+1 {
+		rw.prefixBytesIter.Buf = make([]byte, rw.decoder.maximumKeyLength+1)
 	}
 	if newMax := int(rw.decoder.maximumKeyLength) - len(from) + len(to) + 1; cap(rw.keyBuf) < newMax {
 		rw.keyBuf = make([]byte, newMax)
@@ -775,6 +775,11 @@ func (d *DataBlockDecoder) BlockDecoder() *BlockDecoder {
 	return &d.d
 }
 
+// PrefixChanged returns the prefix-changed bitmap.
+func (d *DataBlockDecoder) PrefixChanged() Bitmap {
+	return d.prefixChanged
+}
+
 // Init initializes the data block reader with the given serialized data block.
 func (d *DataBlockDecoder) Init(schema KeySchema, data []byte) {
 	d.d.Init(data, dataBlockCustomHeaderSize)
@@ -965,7 +970,7 @@ func (i *DataBlockIter) ResetForReuse() DataBlockIter {
 	return DataBlockIter{
 		keySchema:     i.keySchema,
 		getLazyValuer: i.getLazyValuer,
-		keyIter:       PrefixBytesIter{buf: i.keyIter.buf},
+		keyIter:       PrefixBytesIter{Buf: i.keyIter.Buf},
 		keySeeker:     i.keySeeker,
 	}
 }
diff --git a/sstable/colblk/prefix_bytes.go b/sstable/colblk/prefix_bytes.go
index 96851bf4..4a3d1b42 100644
--- a/sstable/colblk/prefix_bytes.go
+++ b/sstable/colblk/prefix_bytes.go
@@ -246,9 +246,9 @@ func (b *PrefixBytes) UnsafeFirstSlice() []byte {
 // within a PrefixBytes, avoiding unnecessary copying when portions of slices
 // are shared.
 type PrefixBytesIter struct {
-	// buf is used for materializing a user key. It is preallocated to the maximum
+	// Buf is used for materializing a user key. It is preallocated to the maximum
 	// key length in the data block.
-	buf                      []byte
+	Buf                      []byte
 	syntheticPrefixLen       uint32
 	sharedAndBundlePrefixLen uint32
 	offsetIndex              int
@@ -263,14 +263,14 @@ func (i *PrefixBytesIter) Init(maxKeyLength int, syntheticPrefix block.Synthetic
 	// block with 1 byte to spare (so that pointer arithmetic is never pointing
 	// beyond the allocation, which would violate Go rules).
 	n := maxKeyLength + 1
-	if cap(i.buf) < n {
+	if cap(i.Buf) < n {
 		ptr := mallocgc(uintptr(n), nil, false)
-		i.buf = unsafe.Slice((*byte)(ptr), n)
+		i.Buf = unsafe.Slice((*byte)(ptr), n)
 	}
-	i.buf = i.buf[:0]
+	i.Buf = i.Buf[:0]
 	i.syntheticPrefixLen = uint32(len(syntheticPrefix))
 	if syntheticPrefix.IsSet() {
-		i.buf = append(i.buf, syntheticPrefix...)
+		i.Buf = append(i.Buf, syntheticPrefix...)
 	}
 }
 
@@ -294,9 +294,9 @@ func (b *PrefixBytes) SetAt(it *PrefixBytesIter, i int) {
 	rowSuffixLen := rowSuffixEnd - rowSuffixStart
 
 	it.sharedAndBundlePrefixLen = it.syntheticPrefixLen + uint32(b.sharedPrefixLen) + bundlePrefixLen
-	it.buf = it.buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
+	it.Buf = it.Buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
 
-	ptr := unsafe.Pointer(unsafe.SliceData(it.buf))
+	ptr := unsafe.Pointer(unsafe.SliceData(it.Buf))
 	ptr = unsafe.Pointer(uintptr(ptr) + uintptr(it.syntheticPrefixLen))
 	// Copy the shared key prefix.
 	memmove(ptr, b.rawBytes.data, uintptr(b.sharedPrefixLen))
@@ -336,9 +336,9 @@ func (b *PrefixBytes) SetNext(it *PrefixBytesIter) {
 			// nothing left to do, we can leave buf as-is.
 			return
 		}
-		it.buf = it.buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
+		it.Buf = it.Buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
 		// Copy in the per-row suffix.
-		ptr := unsafe.Pointer(unsafe.SliceData(it.buf))
+		ptr := unsafe.Pointer(unsafe.SliceData(it.Buf))
 		memmove(
 			unsafe.Pointer(uintptr(ptr)+uintptr(it.sharedAndBundlePrefixLen)),
 			unsafe.Pointer(uintptr(b.rawBytes.data)+uintptr(rowSuffixStart)),
@@ -360,9 +360,9 @@ func (b *PrefixBytes) SetNext(it *PrefixBytesIter) {
 	it.nextBundleOffsetIndex = it.offsetIndex + (1 << b.bundleShift)
 
 	it.sharedAndBundlePrefixLen = it.syntheticPrefixLen + uint32(b.sharedPrefixLen) + bundlePrefixLen
-	it.buf = it.buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
+	it.Buf = it.Buf[:it.sharedAndBundlePrefixLen+rowSuffixLen]
 	// Copy in the new bundle suffix.
-	ptr := unsafe.Pointer(unsafe.SliceData(it.buf))
+	ptr := unsafe.Pointer(unsafe.SliceData(it.Buf))
 	ptr = unsafe.Pointer(uintptr(ptr) + uintptr(it.syntheticPrefixLen) + uintptr(b.sharedPrefixLen))
 	memmove(
 		ptr,
diff --git a/sstable/copier.go b/sstable/copier.go
index 8af2c9dd..7e5decb2 100644
--- a/sstable/copier.go
+++ b/sstable/copier.go
@@ -93,7 +93,7 @@ func CopySpan(
 		r.readable, objstorage.ReadBeforeForIndexAndFilter, &preallocRH)
 	defer rh.Close()
 	rh.SetupForCompaction()
-	indexH, err := r.readIndex(ctx, rh, nil, nil)
+	indexH, err := r.readTopLevelIndexBlock(ctx, noEnv, rh)
 	if err != nil {
 		return 0, err
 	}
@@ -106,7 +106,7 @@ func CopySpan(
 		if w.filter != nil && r.Properties.FilterPolicyName != w.filter.policyName() {
 			return 0, errors.New("mismatched filters")
 		}
-		filterBlock, err := r.readFilter(ctx, rh, nil, nil)
+		filterBlock, err := r.readFilterBlock(ctx, noEnv, rh)
 		if err != nil {
 			return 0, errors.Wrap(err, "reading filter")
 		}
@@ -270,7 +270,7 @@ func intersectingIndexEntries(
 			alloc, entry.sep.UserKey = alloc.Copy(entry.sep.UserKey)
 			res = append(res, entry)
 		} else {
-			subBlk, err := r.readBlock(ctx, bh.Handle, rh, nil, nil, nil)
+			subBlk, err := r.readIndexBlock(ctx, noEnv, rh, bh.Handle)
 			if err != nil {
 				return nil, err
 			}
diff --git a/sstable/layout.go b/sstable/layout.go
index d3b78381..4231970f 100644
--- a/sstable/layout.go
+++ b/sstable/layout.go
@@ -178,8 +178,7 @@ func (l *Layout) Describe(
 			continue
 		}
 
-		h, err := r.readBlock(
-			context.Background(), b.Handle, nil /* readHandle */, nil /* stats */, nil /* iterStats */, nil /* buffer pool */)
+		h, err := r.readBlockInternal(context.Background(), noEnv, noReadHandle, b.Handle)
 		if err != nil {
 			fmt.Fprintf(w, "  [err: %s]\n", err)
 			continue
diff --git a/sstable/reader.go b/sstable/reader.go
index 39993906..77a5eec7 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -72,7 +72,7 @@ type Reader struct {
 	rangeKeyBH   block.Handle
 	valueBIH     valueBlocksIndexHandle
 	propertiesBH block.Handle
-	metaIndexBH  block.Handle
+	metaindexBH  block.Handle
 	footerBH     block.Handle
 
 	Properties   Properties
@@ -280,7 +280,8 @@ func (r *Reader) NewRawRangeDelIter(
 	if r.rangeDelBH.Length == 0 {
 		return nil, nil
 	}
-	h, err := r.readRangeDel(ctx, nil /* stats */, nil /* iterStats */)
+	// TODO(radu): plumb stats here.
+	h, err := r.readRangeDelBlock(ctx, noEnv)
 	if err != nil {
 		return nil, err
 	}
@@ -304,7 +305,8 @@ func (r *Reader) NewRawRangeKeyIter(
 	if r.rangeKeyBH.Length == 0 {
 		return nil, nil
 	}
-	h, err := r.readRangeKey(ctx, nil /* stats */, nil /* iterStats */)
+	// TODO(radu): plumb stats here.
+	h, err := r.readRangeKeyBlock(ctx, noEnv)
 	if err != nil {
 		return nil, err
 	}
@@ -319,38 +321,111 @@ func (r *Reader) NewRawRangeKeyIter(
 	return keyspan.MaybeAssert(iter, r.Compare), nil
 }
 
-func (r *Reader) readIndex(
-	ctx context.Context,
-	readHandle objstorage.ReadHandle,
-	stats *base.InternalIteratorStats,
-	iterStats *iterStatsAccumulator,
+// readBlockEnv contains arguments used when reading a block which apply to all
+// the block reads performed by a higher-level operation.
+type readBlockEnv struct {
+	// stats and iterStats are slightly different. stats is a shared struct
+	// supplied from the outside, and represents stats for the whole iterator
+	// tree and can be reset from the outside (e.g. when the pebble.Iterator is
+	// being reused). It is currently only provided when the iterator tree is
+	// rooted at pebble.Iterator. iterStats contains an sstable iterator's
+	// private stats that are reported to a CategoryStatsCollector when this
+	// iterator is closed. In the important code paths, the CategoryStatsCollector
+	// is managed by the tableCacheContainer.
+	Stats     *base.InternalIteratorStats
+	IterStats *iterStatsAccumulator
+
+	// BufferPool is not-nil if we read blocks into a buffer pool and not into the
+	// cache. This is used during compactions.
+	BufferPool *block.BufferPool
+}
+
+// BlockServedFromCache updates the stats when a block was found in the cache.
+func (env *readBlockEnv) BlockServedFromCache(blockLength uint64) {
+	if env.Stats != nil {
+		env.Stats.BlockBytes += blockLength
+		env.Stats.BlockBytesInCache += blockLength
+	}
+	if env.IterStats != nil {
+		env.IterStats.reportStats(blockLength, blockLength, 0)
+	}
+}
+
+// BlockRead updates the stats when a block had to be read.
+func (env *readBlockEnv) BlockRead(blockLength uint64, readDuration time.Duration) {
+	if env.Stats != nil {
+		env.Stats.BlockBytes += blockLength
+		env.Stats.BlockReadDuration += readDuration
+	}
+	if env.IterStats != nil {
+		env.IterStats.reportStats(blockLength, 0, readDuration)
+	}
+}
+
+// noEnv is the empty readBlockEnv which reports no stats and does not use a
+// buffer pool.
+var noEnv = readBlockEnv{}
+
+// noReadHandle is used when we don't want to pass a ReadHandle to one of the
+// read block methods.
+var noReadHandle objstorage.ReadHandle = nil
+
+// readMetaindexBlock reads the metaindex block.
+func (r *Reader) readMetaindexBlock(
+	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle,
 ) (block.BufferHandle, error) {
 	ctx = objiotracing.WithBlockType(ctx, objiotracing.MetadataBlock)
-	return r.readBlock(ctx, r.indexBH, readHandle, stats, iterStats, nil /* buffer pool */)
+	return r.readBlockInternal(ctx, env, readHandle, r.metaindexBH)
 }
 
-func (r *Reader) readFilter(
-	ctx context.Context,
-	readHandle objstorage.ReadHandle,
-	stats *base.InternalIteratorStats,
-	iterStats *iterStatsAccumulator,
+// readTopLevelIndexBlock reads the top-level index block.
+func (r *Reader) readTopLevelIndexBlock(
+	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle,
+) (block.BufferHandle, error) {
+	return r.readIndexBlock(ctx, env, readHandle, r.indexBH)
+}
+
+// readIndexBlock reads a top-level or second-level index block.
+func (r *Reader) readIndexBlock(
+	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle, bh block.Handle,
+) (block.BufferHandle, error) {
+	ctx = objiotracing.WithBlockType(ctx, objiotracing.MetadataBlock)
+	return r.readBlockInternal(ctx, env, readHandle, bh)
+}
+
+func (r *Reader) readDataBlock(
+	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle, bh block.Handle,
+) (block.BufferHandle, error) {
+	ctx = objiotracing.WithBlockType(ctx, objiotracing.DataBlock)
+	return r.readBlockInternal(ctx, env, readHandle, bh)
+}
+
+func (r *Reader) readFilterBlock(
+	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle,
 ) (block.BufferHandle, error) {
 	ctx = objiotracing.WithBlockType(ctx, objiotracing.FilterBlock)
-	return r.readBlock(ctx, r.filterBH, readHandle, stats, iterStats, nil /* buffer pool */)
+	return r.readBlockInternal(ctx, env, readHandle, r.filterBH)
 }
 
-func (r *Reader) readRangeDel(
-	ctx context.Context, stats *base.InternalIteratorStats, iterStats *iterStatsAccumulator,
+func (r *Reader) readRangeDelBlock(
+	ctx context.Context, env readBlockEnv,
 ) (block.BufferHandle, error) {
 	ctx = objiotracing.WithBlockType(ctx, objiotracing.MetadataBlock)
-	return r.readBlock(ctx, r.rangeDelBH, nil /* readHandle */, stats, iterStats, nil /* buffer pool */)
+	return r.readBlockInternal(ctx, env, nil, r.rangeDelBH)
 }
 
-func (r *Reader) readRangeKey(
-	ctx context.Context, stats *base.InternalIteratorStats, iterStats *iterStatsAccumulator,
+func (r *Reader) readRangeKeyBlock(
+	ctx context.Context, env readBlockEnv,
 ) (block.BufferHandle, error) {
 	ctx = objiotracing.WithBlockType(ctx, objiotracing.MetadataBlock)
-	return r.readBlock(ctx, r.rangeKeyBH, nil /* readHandle */, stats, iterStats, nil /* buffer pool */)
+	return r.readBlockInternal(ctx, env, nil, r.rangeKeyBH)
+}
+
+func (r *Reader) readValueBlock(
+	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle, bh block.Handle,
+) (block.BufferHandle, error) {
+	ctx = objiotracing.WithBlockType(ctx, objiotracing.ValueBlock)
+	return r.readBlockInternal(ctx, env, readHandle, bh)
 }
 
 func checkChecksum(
@@ -388,28 +463,17 @@ func DeterministicReadBlockDurationForTesting() func() {
 
 var deterministicReadBlockDurationForTesting = false
 
-func (r *Reader) readBlock(
-	ctx context.Context,
-	bh block.Handle,
-	readHandle objstorage.ReadHandle,
-	stats *base.InternalIteratorStats,
-	iterStats *iterStatsAccumulator,
-	bufferPool *block.BufferPool,
+// readBlockInternal should not be used directly; one of the read*Block methods
+// should be used instead.
+func (r *Reader) readBlockInternal(
+	ctx context.Context, env readBlockEnv, readHandle objstorage.ReadHandle, bh block.Handle,
 ) (handle block.BufferHandle, _ error) {
 	if h := r.cacheOpts.Cache.Get(r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset); h.Get() != nil {
 		// Cache hit.
 		if readHandle != nil {
 			readHandle.RecordCacheHit(ctx, int64(bh.Offset), int64(bh.Length+block.TrailerLen))
 		}
-		if stats != nil {
-			stats.BlockBytes += bh.Length
-			stats.BlockBytesInCache += bh.Length
-		}
-		if iterStats != nil {
-			iterStats.reportStats(bh.Length, bh.Length, 0)
-		}
-		// This block is already in the cache; return a handle to existing vlaue
-		// in the cache.
+		env.BlockServedFromCache(bh.Length)
 		return block.CacheBufferHandle(h), nil
 	}
 
@@ -423,7 +487,7 @@ func (r *Reader) readBlock(
 		defer sema.Release(1)
 	}
 
-	compressed := block.Alloc(int(bh.Length+block.TrailerLen), bufferPool)
+	compressed := block.Alloc(int(bh.Length+block.TrailerLen), env.BufferPool)
 	readStopwatch := makeStopwatch()
 	var err error
 	if readHandle != nil {
@@ -443,14 +507,11 @@ func (r *Reader) readBlock(
 			filepath.Base(filepath.Dir(file2)), filepath.Base(file2), line2,
 			filepath.Base(filepath.Dir(file1)), filepath.Base(file1), line1)
 	}
-	if stats != nil {
-		stats.BlockBytes += bh.Length
-		stats.BlockReadDuration += readDuration
-	}
 	if err != nil {
 		compressed.Release()
 		return block.BufferHandle{}, err
 	}
+	env.BlockRead(bh.Length, readDuration)
 	if err := checkChecksum(r.checksumType, compressed.Get(), bh, r.cacheOpts.FileNum); err != nil {
 		compressed.Release()
 		return block.BufferHandle{}, err
@@ -470,26 +531,20 @@ func (r *Reader) readBlock(
 			return block.BufferHandle{}, err
 		}
 
-		decompressed = block.Alloc(decodedLen, bufferPool)
-		if err := block.DecompressInto(typ, compressed.Get()[prefixLen:], decompressed.Get()); err != nil {
-			compressed.Release()
+		decompressed = block.Alloc(decodedLen, env.BufferPool)
+		err = block.DecompressInto(typ, compressed.Get()[prefixLen:], decompressed.Get())
+		compressed.Release()
+		if err != nil {
+			decompressed.Release()
 			return block.BufferHandle{}, err
 		}
-		compressed.Release()
-	}
-
-	if iterStats != nil {
-		iterStats.reportStats(bh.Length, 0, readDuration)
 	}
 	h := decompressed.MakeHandle(r.cacheOpts.Cache, r.cacheOpts.CacheID, r.cacheOpts.FileNum, bh.Offset)
 	return h, nil
 }
 
 func (r *Reader) readMetaindex(
-	ctx context.Context,
-	metaindexBH block.Handle,
-	readHandle objstorage.ReadHandle,
-	filters map[string]FilterPolicy,
+	ctx context.Context, readHandle objstorage.ReadHandle, filters map[string]FilterPolicy,
 ) error {
 	// We use a BufferPool when reading metaindex blocks in order to avoid
 	// populating the block cache with these blocks. In heavy-write workloads,
@@ -503,19 +558,20 @@ func (r *Reader) readMetaindex(
 	// When we're finished, release the buffers we've allocated back to memory
 	// allocator. We don't expect to use metaBufferPool again.
 	defer r.metaBufferPool.Release()
+	metaEnv := readBlockEnv{
+		BufferPool: &r.metaBufferPool,
+	}
 
-	b, err := r.readBlock(
-		ctx, metaindexBH, readHandle, nil, /* stats */
-		nil /* iterStats */, &r.metaBufferPool)
+	b, err := r.readMetaindexBlock(ctx, metaEnv, readHandle)
 	if err != nil {
 		return err
 	}
 	data := b.Get()
 	defer b.Release()
 
-	if uint64(len(data)) != metaindexBH.Length {
+	if uint64(len(data)) != r.metaindexBH.Length {
 		return base.CorruptionErrorf("pebble/table: unexpected metaindex block size: %d vs %d",
-			errors.Safe(len(data)), errors.Safe(metaindexBH.Length))
+			errors.Safe(len(data)), errors.Safe(r.metaindexBH.Length))
 	}
 
 	var meta map[string]block.Handle
@@ -525,9 +581,7 @@ func (r *Reader) readMetaindex(
 	}
 
 	if bh, ok := meta[metaPropertiesName]; ok {
-		b, err = r.readBlock(
-			ctx, bh, readHandle, nil, /* stats */
-			nil /* iterStats */, nil /* buffer pool */)
+		b, err = r.readBlockInternal(ctx, metaEnv, readHandle, bh)
 		if err != nil {
 			return err
 		}
@@ -578,15 +632,16 @@ func (r *Reader) Layout() (*Layout, error) {
 		RangeKey:   r.rangeKeyBH,
 		ValueIndex: r.valueBIH.h,
 		Properties: r.propertiesBH,
-		MetaIndex:  r.metaIndexBH,
+		MetaIndex:  r.metaindexBH,
 		Footer:     r.footerBH,
 		Format:     r.tableFormat,
 	}
 	if r.filterBH.Length > 0 {
 		l.Filter = []NamedBlockHandle{{Name: "fullfilter." + r.tableFilter.policy.Name(), Handle: r.filterBH}}
 	}
+	ctx := context.TODO()
 
-	indexH, err := r.readIndex(context.Background(), nil, nil, nil)
+	indexH, err := r.readTopLevelIndexBlock(ctx, noEnv, noReadHandle)
 	if err != nil {
 		return nil, err
 	}
@@ -626,8 +681,7 @@ func (r *Reader) Layout() (*Layout, error) {
 			}
 			l.Index = append(l.Index, indexBH.Handle)
 
-			subIndex, err := r.readBlock(context.Background(), indexBH.Handle,
-				nil /* readHandle */, nil /* stats */, nil /* iterStats */, nil /* buffer pool */)
+			subIndex, err := r.readIndexBlock(ctx, noEnv, noReadHandle, indexBH.Handle)
 			if err != nil {
 				return nil, err
 			}
@@ -655,7 +709,7 @@ func (r *Reader) Layout() (*Layout, error) {
 		}
 	}
 	if r.valueBIH.h.Length != 0 {
-		vbiH, err := r.readBlock(context.Background(), r.valueBIH.h, nil, nil, nil, nil /* buffer pool */)
+		vbiH, err := r.readValueBlock(context.Background(), noEnv, noReadHandle, r.valueBIH.h)
 		if err != nil {
 			return nil, err
 		}
@@ -708,7 +762,7 @@ func (r *Reader) ValidateBlockChecksums() error {
 		}
 
 		// Read the block, which validates the checksum.
-		h, err := r.readBlock(context.Background(), bh, rh, nil, nil /* iterStats */, nil /* buffer pool */)
+		h, err := r.readBlockInternal(context.Background(), noEnv, rh, bh)
 		if err != nil {
 			return err
 		}
@@ -751,8 +805,9 @@ func estimateDiskUsage[I any, PI indexBlockIterator[I]](
 	if r.err != nil {
 		return 0, r.err
 	}
+	ctx := context.TODO()
 
-	indexH, err := r.readIndex(context.Background(), nil, nil, nil)
+	indexH, err := r.readTopLevelIndexBlock(ctx, noEnv, noReadHandle)
 	if err != nil {
 		return 0, err
 	}
@@ -781,8 +836,7 @@ func estimateDiskUsage[I any, PI indexBlockIterator[I]](
 		if err != nil {
 			return 0, errCorruptIndexEntry(err)
 		}
-		startIdxBlock, err := r.readBlock(context.Background(), startIndexBH.Handle,
-			nil /* readHandle */, nil /* stats */, nil /* iterStats */, nil /* buffer pool */)
+		startIdxBlock, err := r.readIndexBlock(ctx, noEnv, noReadHandle, startIndexBH.Handle)
 		if err != nil {
 			return 0, err
 		}
@@ -798,8 +852,7 @@ func estimateDiskUsage[I any, PI indexBlockIterator[I]](
 			if err != nil {
 				return 0, errCorruptIndexEntry(err)
 			}
-			endIdxBlock, err := r.readBlock(context.Background(),
-				endIndexBH.Handle, nil /* readHandle */, nil /* stats */, nil /* iterStats */, nil /* buffer pool */)
+			endIdxBlock, err := r.readIndexBlock(ctx, noEnv, noReadHandle, endIndexBH.Handle)
 			if err != nil {
 				return 0, err
 			}
@@ -901,14 +954,14 @@ func NewReader(ctx context.Context, f objstorage.Readable, o ReaderOptions) (*Re
 	}
 	r.checksumType = footer.checksum
 	r.tableFormat = footer.format
+	r.indexBH = footer.indexBH
+	r.metaindexBH = footer.metaindexBH
+	r.footerBH = footer.footerBH
 	// Read the metaindex and properties blocks.
-	if err := r.readMetaindex(ctx, footer.metaindexBH, rh, o.Filters); err != nil {
+	if err := r.readMetaindex(ctx, rh, o.Filters); err != nil {
 		r.err = err
 		return nil, r.Close()
 	}
-	r.indexBH = footer.indexBH
-	r.metaIndexBH = footer.metaindexBH
-	r.footerBH = footer.footerBH
 
 	// If the table format indicates that blocks are encoded within the columnar
 	// format, we require a key schema to interpret it correctly.
@@ -1003,7 +1056,7 @@ func (s *simpleReadable) Size() int64 {
 	return s.size
 }
 
-// NewReaddHandle is part of the objstorage.Readable interface.
+// NewReadHandle is part of the objstorage.Readable interface.
 func (s *simpleReadable) NewReadHandle(
 	readBeforeSize objstorage.ReadBeforeSize,
 ) objstorage.ReadHandle {
diff --git a/sstable/reader_iter_single_lvl.go b/sstable/reader_iter_single_lvl.go
index 47fb2205..7bda676e 100644
--- a/sstable/reader_iter_single_lvl.go
+++ b/sstable/reader_iter_single_lvl.go
@@ -17,7 +17,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
-	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
 	"github.com/cockroachdb/pebble/sstable/block"
 )
 
@@ -74,18 +73,9 @@ type singleLevelIterator[I any, PI indexBlockIterator[I], D any, PD dataBlockIte
 	vbRHPrealloc objstorageprovider.PreallocatedReadHandle
 	err          error
 	closeHook    func(i Iterator) error
-	// stats and iterStats are slightly different. stats is a shared struct
-	// supplied from the outside, and represents stats for the whole iterator
-	// tree and can be reset from the outside (e.g. when the pebble.Iterator is
-	// being reused). It is currently only provided when the iterator tree is
-	// rooted at pebble.Iterator. iterStats is this sstable iterator's private
-	// stats that are reported to a CategoryStatsCollector when this iterator is
-	// closed. More paths are instrumented with this as the
-	// CategoryStatsCollector needed for this is provided by the
-	// tableCacheContainer (which is more universally used).
-	stats      *base.InternalIteratorStats
-	iterStats  iterStatsAccumulator
-	bufferPool *block.BufferPool
+
+	iterStats    iterStatsAccumulator
+	readBlockEnv readBlockEnv
 
 	// boundsCmp and positionedUsingLatestBounds are for optimizing iteration
 	// that uses multiple adjacent bounds. The seek after setting a new bound
@@ -252,7 +242,7 @@ func newColumnBlockSingleLevelIterator(
 		i.vbRH = objstorageprovider.UsePreallocatedReadHandle(r.readable, objstorage.NoReadBefore, &i.vbRHPrealloc)
 	}
 	i.data.InitOnce(r.keySchema, i.cmp, r.Split, getLazyValuer)
-	indexH, err := r.readIndex(ctx, i.indexFilterRH, stats, &i.iterStats)
+	indexH, err := r.readTopLevelIndexBlock(ctx, i.readBlockEnv, i.indexFilterRH)
 	if err == nil {
 		err = i.index.InitHandle(i.cmp, r.Split, indexH, transforms)
 	}
@@ -319,7 +309,7 @@ func newRowBlockSingleLevelIterator(
 		i.data.SetHasValuePrefix(true)
 	}
 
-	indexH, err := r.readIndex(ctx, i.indexFilterRH, stats, &i.iterStats)
+	indexH, err := r.readTopLevelIndexBlock(ctx, i.readBlockEnv, i.indexFilterRH)
 	if err == nil {
 		err = i.index.InitHandle(i.cmp, r.Split, indexH, transforms)
 	}
@@ -352,15 +342,18 @@ func (i *singleLevelIterator[I, PI, D, PD]) init(
 	i.useFilterBlock = useFilterBlock
 	i.reader = r
 	i.cmp = r.Compare
-	i.stats = stats
 	i.transforms = transforms
-	i.bufferPool = bufferPool
 	if v != nil {
 		i.vState = v
 		i.endKeyInclusive, i.lower, i.upper = v.constrainBounds(lower, upper, false /* endInclusive */)
 	}
 
 	i.iterStats.init(categoryAndQoS, statsCollector)
+	i.readBlockEnv = readBlockEnv{
+		Stats:      stats,
+		IterStats:  &i.iterStats,
+		BufferPool: bufferPool,
+	}
 
 	i.indexFilterRH = objstorageprovider.UsePreallocatedReadHandle(
 		r.readable, objstorage.ReadBeforeForIndexAndFilter, &i.indexFilterRHPrealloc)
@@ -521,10 +514,10 @@ func (i *singleLevelIterator[I, PI, P, PD]) SetContext(ctx context.Context) {
 	i.ctx = ctx
 }
 
-// loadBlock loads the block at the current index position and leaves i.data
+// loadDataBlock loads the block at the current index position and leaves i.data
 // unpositioned. If unsuccessful, it sets i.err to any error encountered, which
 // may be nil if we have simply exhausted the entire table.
-func (i *singleLevelIterator[I, PI, P, PD]) loadBlock(dir int8) loadBlockResult {
+func (i *singleLevelIterator[I, PI, P, PD]) loadDataBlock(dir int8) loadBlockResult {
 	if !PI(&i.index).Valid() {
 		// Ensure the data block iterator is invalidated even if loading of the block
 		// fails.
@@ -539,8 +532,8 @@ func (i *singleLevelIterator[I, PI, P, PD]) loadBlock(dir int8) loadBlockResult
 		// or disk.
 		//
 		// It's safe to leave i.data in its original state here, as all callers to
-		// loadBlock make an absolute positioning call (i.e. a seek, first, or last)
-		// to `i.data` right after loadBlock returns loadBlockOK.
+		// loadDataBlock make an absolute positioning call (i.e. a seek, first, or last)
+		// to `i.data` right after loadDataBlock returns loadBlockOK.
 		i.initBounds()
 		return loadBlockOK
 	}
@@ -566,9 +559,7 @@ func (i *singleLevelIterator[I, PI, P, PD]) loadBlock(dir int8) loadBlockResult
 		}
 		// blockIntersects
 	}
-	ctx := objiotracing.WithBlockType(i.ctx, objiotracing.DataBlock)
-	block, err := i.reader.readBlock(
-		ctx, i.dataBH, i.dataRH, i.stats, &i.iterStats, i.bufferPool)
+	block, err := i.reader.readDataBlock(i.ctx, i.readBlockEnv, i.dataRH, i.dataBH)
 	if err != nil {
 		i.err = err
 		return loadBlockFailed
@@ -586,10 +577,11 @@ func (i *singleLevelIterator[I, PI, P, PD]) loadBlock(dir int8) loadBlockResult
 // readBlockForVBR implements the blockProviderWhenOpen interface for use by
 // the valueBlockReader.
 func (i *singleLevelIterator[I, PI, D, PD]) readBlockForVBR(
-	h block.Handle, stats *base.InternalIteratorStats,
+	bh block.Handle, stats *base.InternalIteratorStats,
 ) (block.BufferHandle, error) {
-	ctx := objiotracing.WithBlockType(i.ctx, objiotracing.ValueBlock)
-	return i.reader.readBlock(ctx, h, i.vbRH, stats, &i.iterStats, i.bufferPool)
+	env := i.readBlockEnv
+	env.Stats = stats
+	return i.reader.readValueBlock(i.ctx, env, i.vbRH, bh)
 }
 
 // resolveMaybeExcluded is invoked when the block-property filterer has found
@@ -827,7 +819,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) seekGEHelper(
 			PD(&i.data).Invalidate()
 			return nil
 		}
-		result := i.loadBlock(+1)
+		result := i.loadDataBlock(+1)
 		if result == loadBlockFailed {
 			return nil
 		}
@@ -959,7 +951,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) bloomFilterMayContain(prefix []byte)
 		}
 	}
 
-	dataH, err := i.reader.readFilter(i.ctx, i.indexFilterRH, i.stats, &i.iterStats)
+	dataH, err := i.reader.readFilterBlock(i.ctx, i.readBlockEnv, i.indexFilterRH)
 	if err != nil {
 		return false, err
 	}
@@ -1035,7 +1027,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) virtualLastSeekLE() *base.InternalKV
 	// > key. So there could be keys in the block > key. But the block preceding
 	// this block cannot have any keys > key, otherwise it would have been the
 	// result of the original index.SeekGE.
-	result := i.loadBlock(-1)
+	result := i.loadDataBlock(-1)
 	if result == loadBlockFailed {
 		return nil
 	}
@@ -1133,7 +1125,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) SeekLT(
 			}
 		}
 		// INVARIANT: ikey != nil.
-		result := i.loadBlock(-1)
+		result := i.loadDataBlock(-1)
 		if result == loadBlockFailed {
 			return nil
 		}
@@ -1205,7 +1197,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) firstInternal() *base.InternalKV {
 		PD(&i.data).Invalidate()
 		return nil
 	}
-	result := i.loadBlock(+1)
+	result := i.loadDataBlock(+1)
 	if result == loadBlockFailed {
 		return nil
 	}
@@ -1271,7 +1263,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) lastInternal() *base.InternalKV {
 		PD(&i.data).Invalidate()
 		return nil
 	}
-	result := i.loadBlock(-1)
+	result := i.loadDataBlock(-1)
 	if result == loadBlockFailed {
 		return nil
 	}
@@ -1372,7 +1364,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) NextPrefix(succKey []byte) *base.Int
 			return nil
 		}
 	}
-	result := i.loadBlock(+1)
+	result := i.loadDataBlock(+1)
 	if result == loadBlockFailed {
 		return nil
 	}
@@ -1433,7 +1425,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) skipForward() *base.InternalKV {
 			PD(&i.data).Invalidate()
 			break
 		}
-		result := i.loadBlock(+1)
+		result := i.loadDataBlock(+1)
 		if result != loadBlockOK {
 			if i.err != nil {
 				break
@@ -1442,7 +1434,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) skipForward() *base.InternalKV {
 				// We checked that i.index was at a valid entry, so
 				// loadBlockFailed could not have happened due to i.index
 				// being exhausted, and must be due to an error.
-				panic("loadBlock should not have failed with no error")
+				panic("loadDataBlock should not have failed with no error")
 			}
 			// result == loadBlockIrrelevant. Enforce the upper bound here
 			// since don't want to bother moving to the next block if upper
@@ -1513,7 +1505,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) skipBackward() *base.InternalKV {
 			PD(&i.data).Invalidate()
 			break
 		}
-		result := i.loadBlock(-1)
+		result := i.loadDataBlock(-1)
 		if result != loadBlockOK {
 			if i.err != nil {
 				break
@@ -1522,7 +1514,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) skipBackward() *base.InternalKV {
 				// We checked that i.index was at a valid entry, so
 				// loadBlockFailed could not have happened due to to i.index
 				// being exhausted, and must be due to an error.
-				panic("loadBlock should not have failed with no error")
+				panic("loadDataBlock should not have failed with no error")
 			}
 			// result == loadBlockIrrelevant. Enforce the lower bound here
 			// since don't want to bother moving to the previous block if lower
diff --git a/sstable/reader_iter_two_lvl.go b/sstable/reader_iter_two_lvl.go
index 7651a1f7..bc7068fd 100644
--- a/sstable/reader_iter_two_lvl.go
+++ b/sstable/reader_iter_two_lvl.go
@@ -16,7 +16,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
-	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
 	"github.com/cockroachdb/pebble/sstable/block"
 )
 
@@ -38,11 +37,11 @@ type twoLevelIterator[I any, PI indexBlockIterator[I], D any, PD dataBlockIterat
 
 var _ Iterator = (*twoLevelIteratorRowBlocks)(nil)
 
-// loadIndex loads the index block at the current top level index position and
-// leaves i.index unpositioned. If unsuccessful, it gets i.secondLevel.err to any error
-// encountered, which may be nil if we have simply exhausted the entire table.
-// This is used for two level indexes.
-func (i *twoLevelIterator[I, PI, D, PD]) loadIndex(dir int8) loadBlockResult {
+// loadSecondLevelIndexBlock loads the index block at the current top level
+// index position and leaves i.index unpositioned. If unsuccessful, it gets
+// i.secondLevel.err to any error encountered, which may be nil if we have
+// simply exhausted the entire table. This is used for two level indexes.
+func (i *twoLevelIterator[I, PI, D, PD]) loadSecondLevelIndexBlock(dir int8) loadBlockResult {
 	// Ensure the index data block iterators are invalidated even if loading of
 	// the index fails.
 	PD(&i.secondLevel.data).Invalidate()
@@ -69,9 +68,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) loadIndex(dir int8) loadBlockResult {
 		}
 		// blockIntersects
 	}
-	ctx := objiotracing.WithBlockType(i.secondLevel.ctx, objiotracing.MetadataBlock)
-	indexBlock, err := i.secondLevel.reader.readBlock(
-		ctx, bhp.Handle, i.secondLevel.indexFilterRH, i.secondLevel.stats, &i.secondLevel.iterStats, i.secondLevel.bufferPool)
+	indexBlock, err := i.secondLevel.reader.readIndexBlock(i.secondLevel.ctx, i.secondLevel.readBlockEnv, i.secondLevel.indexFilterRH, bhp.Handle)
 	if err == nil {
 		err = PI(&i.secondLevel.index).InitHandle(i.secondLevel.cmp, i.secondLevel.reader.Split, indexBlock, i.secondLevel.transforms)
 	}
@@ -198,7 +195,7 @@ func newColumnBlockTwoLevelIterator(
 	}
 	i.secondLevel.data.InitOnce(r.keySchema, r.Compare, r.Split, getLazyValuer)
 	i.useFilterBlock = shouldUseFilterBlock(r, filterBlockSizeLimit)
-	topLevelIndexH, err := r.readIndex(ctx, i.secondLevel.indexFilterRH, stats, &i.secondLevel.iterStats)
+	topLevelIndexH, err := r.readTopLevelIndexBlock(ctx, i.secondLevel.readBlockEnv, i.secondLevel.indexFilterRH)
 	if err == nil {
 		err = i.topLevelIndex.InitHandle(i.secondLevel.cmp, i.secondLevel.reader.Split, topLevelIndexH, transforms)
 	}
@@ -265,7 +262,7 @@ func newRowBlockTwoLevelIterator(
 
 	i.useFilterBlock = shouldUseFilterBlock(r, filterBlockSizeLimit)
 
-	topLevelIndexH, err := r.readIndex(ctx, i.secondLevel.indexFilterRH, stats, &i.secondLevel.iterStats)
+	topLevelIndexH, err := r.readTopLevelIndexBlock(ctx, i.secondLevel.readBlockEnv, i.secondLevel.indexFilterRH)
 	if err == nil {
 		err = i.topLevelIndex.InitHandle(i.secondLevel.cmp, i.secondLevel.reader.Split, topLevelIndexH, transforms)
 	}
@@ -341,7 +338,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) SeekGE(
 			return nil
 		}
 
-		result := i.loadIndex(+1)
+		result := i.loadSecondLevelIndexBlock(+1)
 		if result == loadBlockFailed {
 			i.secondLevel.boundsCmp = 0
 			return nil
@@ -509,7 +506,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) SeekPrefixGE(
 			return nil
 		}
 
-		result := i.loadIndex(+1)
+		result := i.loadSecondLevelIndexBlock(+1)
 		if result == loadBlockFailed {
 			i.secondLevel.boundsCmp = 0
 			return nil
@@ -633,7 +630,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) virtualLastSeekLE() *base.InternalKV {
 	if !topLevelOk {
 		return i.skipBackward()
 	}
-	result := i.loadIndex(-1)
+	result := i.loadSecondLevelIndexBlock(-1)
 	if result == loadBlockFailed {
 		i.secondLevel.boundsCmp = 0
 		return nil
@@ -688,7 +685,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) SeekLT(
 			return nil
 		}
 
-		result = i.loadIndex(-1)
+		result = i.loadSecondLevelIndexBlock(-1)
 		if result == loadBlockFailed {
 			return nil
 		}
@@ -702,7 +699,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) SeekLT(
 		}
 		// Else loadBlockIrrelevant, so fall through.
 	} else {
-		result = i.loadIndex(-1)
+		result = i.loadSecondLevelIndexBlock(-1)
 		if result == loadBlockFailed {
 			return nil
 		}
@@ -749,7 +746,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) First() *base.InternalKV {
 	if !PI(&i.topLevelIndex).First() {
 		return nil
 	}
-	result := i.loadIndex(+1)
+	result := i.loadSecondLevelIndexBlock(+1)
 	if result == loadBlockFailed {
 		return nil
 	}
@@ -799,7 +796,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) Last() *base.InternalKV {
 	if !PI(&i.topLevelIndex).Last() {
 		return nil
 	}
-	result := i.loadIndex(-1)
+	result := i.loadSecondLevelIndexBlock(-1)
 	if result == loadBlockFailed {
 		return nil
 	}
@@ -868,7 +865,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) NextPrefix(succKey []byte) *base.Intern
 		PI(&i.secondLevel.index).Invalidate()
 		return nil
 	}
-	result := i.loadIndex(+1)
+	result := i.loadSecondLevelIndexBlock(+1)
 	if result == loadBlockFailed {
 		return nil
 	}
@@ -949,7 +946,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) skipForward() *base.InternalKV {
 			PI(&i.secondLevel.index).Invalidate()
 			return nil
 		}
-		result := i.loadIndex(+1)
+		result := i.loadSecondLevelIndexBlock(+1)
 		if result == loadBlockFailed {
 			return nil
 		}
@@ -995,7 +992,7 @@ func (i *twoLevelIterator[I, PI, D, PD]) skipBackward() *base.InternalKV {
 			PI(&i.secondLevel.index).Invalidate()
 			return nil
 		}
-		result := i.loadIndex(-1)
+		result := i.loadSecondLevelIndexBlock(-1)
 		if result == loadBlockFailed {
 			return nil
 		}
diff --git a/sstable/value_block.go b/sstable/value_block.go
index 1df4e9ed..2d6a1999 100644
--- a/sstable/value_block.go
+++ b/sstable/value_block.go
@@ -692,8 +692,7 @@ func (bpwc blockProviderWhenClosed) readBlockForVBR(
 	// TODO(jackson,sumeer): Consider whether to use a buffer pool in this case.
 	// The bpwc is not allowed to outlive the iterator tree, so it cannot
 	// outlive the buffer pool.
-	return bpwc.r.readBlock(
-		ctx, h, nil, stats, nil /* iterStats */, nil /* buffer pool */)
+	return bpwc.r.readValueBlock(ctx, noEnv, noReadHandle, h)
 }
 
 // ReaderProvider supports the implementation of blockProviderWhenClosed.


Test patch:
diff --git a/internal/crdbtest/crdbtest.go b/internal/crdbtest/crdbtest.go
index dcc4e9ae..234808a4 100644
--- a/internal/crdbtest/crdbtest.go
+++ b/internal/crdbtest/crdbtest.go
@@ -11,14 +11,19 @@ import (
 	"cmp"
 	"encoding/binary"
 	"fmt"
+	"io"
 	"math/rand/v2"
 	"slices"
+	"sync"
 	"time"
+	"unsafe"
 
+	"github.com/cockroachdb/crlib/crbytes"
 	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/pebble/sstable/colblk"
 )
 
 const withWall = 9
@@ -486,3 +491,378 @@ func (g *cockroachKeyGen) randTimestamp() (wallTime uint64, logicalTime uint32)
 	}
 	return wallTime, logicalTime
 }
+
+const (
+	cockroachColRoachKey int = iota
+	cockroachColMVCCWallTime
+	cockroachColMVCCLogical
+	cockroachColUntypedVersion
+	cockroachColCount
+)
+
+var KeySchema = colblk.KeySchema{
+	ColumnTypes: []colblk.DataType{
+		cockroachColRoachKey:       colblk.DataTypePrefixBytes,
+		cockroachColMVCCWallTime:   colblk.DataTypeUint,
+		cockroachColMVCCLogical:    colblk.DataTypeUint,
+		cockroachColUntypedVersion: colblk.DataTypeBytes,
+	},
+	NewKeyWriter: func() colblk.KeyWriter {
+		kw := &cockroachKeyWriter{}
+		kw.roachKeys.Init(16)
+		kw.wallTimes.Init()
+		kw.logicalTimes.InitWithDefault()
+		kw.untypedVersions.Init()
+		return kw
+	},
+	NewKeySeeker: func() colblk.KeySeeker {
+		return &cockroachKeySeeker{}
+	},
+}
+
+type cockroachKeyWriter struct {
+	roachKeys       colblk.PrefixBytesBuilder
+	wallTimes       colblk.UintBuilder
+	logicalTimes    colblk.UintBuilder
+	untypedVersions colblk.RawBytesBuilder
+	prevSuffix      []byte
+}
+
+func (kw *cockroachKeyWriter) ComparePrev(key []byte) colblk.KeyComparison {
+	var cmpv colblk.KeyComparison
+	cmpv.PrefixLen = int32(Split(key)) // TODO(jackson): Inline
+	if kw.roachKeys.Rows() == 0 {
+		cmpv.UserKeyComparison = 1
+		return cmpv
+	}
+	lp := kw.roachKeys.UnsafeGet(kw.roachKeys.Rows() - 1)
+	cmpv.CommonPrefixLen = int32(crbytes.CommonPrefix(lp, key[:cmpv.PrefixLen-1]))
+	if cmpv.CommonPrefixLen == cmpv.PrefixLen-1 {
+		// Adjust CommonPrefixLen to include the sentinel byte,
+		cmpv.CommonPrefixLen = cmpv.PrefixLen
+		cmpv.UserKeyComparison = int32(CompareSuffixes(key[cmpv.PrefixLen:], kw.prevSuffix))
+		return cmpv
+	}
+	// The keys have different MVCC prefixes. We haven't determined which is
+	// greater, but we know the index at which they diverge. The base.Comparer
+	// contract dictates that prefixes must be lexicographically ordered.
+	if len(lp) == int(cmpv.CommonPrefixLen) {
+		// cmpv.PrefixLen > cmpv.PrefixLenShared; key is greater.
+		cmpv.UserKeyComparison = +1
+	} else {
+		// Both keys have at least 1 additional byte at which they diverge.
+		// Compare the diverging byte.
+		cmpv.UserKeyComparison = int32(cmp.Compare(key[cmpv.CommonPrefixLen], lp[cmpv.CommonPrefixLen]))
+	}
+	return cmpv
+}
+
+func (kw *cockroachKeyWriter) WriteKey(
+	row int, key []byte, keyPrefixLen, keyPrefixLenSharedWithPrev int32,
+) {
+	// TODO(jackson): Avoid copying the previous suffix.
+	// TODO(jackson): Use keyPrefixLen to speed up decoding.
+	roachKey, untypedVersion, wallTime, logicalTime := DecodeEngineKey(key)
+	kw.prevSuffix = append(kw.prevSuffix[:0], key[keyPrefixLen:]...)
+	// When the roach key is the same, keyPrefixLenSharedWithPrev includes the
+	// separator byte.
+	kw.roachKeys.Put(roachKey, min(int(keyPrefixLenSharedWithPrev), len(roachKey)))
+	kw.wallTimes.Set(row, wallTime)
+	// The w.logicalTimes builder was initialized with InitWithDefault, so if we
+	// don't set a value, the column value is implicitly zero. We only need to
+	// Set anything for non-zero values.
+	if logicalTime > 0 {
+		kw.logicalTimes.Set(row, uint64(logicalTime))
+	}
+	kw.untypedVersions.Put(untypedVersion)
+}
+
+func (kw *cockroachKeyWriter) MaterializeKey(dst []byte, i int) []byte {
+	dst = append(dst, kw.roachKeys.UnsafeGet(i)...)
+	// Append separator byte.
+	dst = append(dst, 0)
+	if untypedVersion := kw.untypedVersions.UnsafeGet(i); len(untypedVersion) > 0 {
+		dst = append(dst, untypedVersion...)
+		dst = append(dst, byte(len(untypedVersion)))
+		return dst
+	}
+	return AppendTimestamp(dst, kw.wallTimes.Get(i), uint32(kw.logicalTimes.Get(i)))
+}
+
+func (kw *cockroachKeyWriter) Reset() {
+	kw.roachKeys.Reset()
+	kw.wallTimes.Reset()
+	kw.logicalTimes.Reset()
+	kw.untypedVersions.Reset()
+}
+
+func (kw *cockroachKeyWriter) WriteDebug(dst io.Writer, rows int) {
+	fmt.Fprint(dst, "prefixes: ")
+	kw.roachKeys.WriteDebug(dst, rows)
+	fmt.Fprintln(dst)
+	fmt.Fprint(dst, "wall times: ")
+	kw.wallTimes.WriteDebug(dst, rows)
+	fmt.Fprintln(dst)
+	fmt.Fprint(dst, "logical times: ")
+	kw.logicalTimes.WriteDebug(dst, rows)
+	fmt.Fprintln(dst)
+	fmt.Fprint(dst, "untyped suffixes: ")
+	kw.untypedVersions.WriteDebug(dst, rows)
+	fmt.Fprintln(dst)
+}
+
+func (kw *cockroachKeyWriter) NumColumns() int {
+	return cockroachColCount
+}
+
+func (kw *cockroachKeyWriter) DataType(col int) colblk.DataType {
+	return KeySchema.ColumnTypes[col]
+}
+
+func (kw *cockroachKeyWriter) Size(rows int, offset uint32) uint32 {
+	offset = kw.roachKeys.Size(rows, offset)
+	offset = kw.wallTimes.Size(rows, offset)
+	offset = kw.logicalTimes.Size(rows, offset)
+	offset = kw.untypedVersions.Size(rows, offset)
+	return offset
+}
+
+func (kw *cockroachKeyWriter) Finish(
+	col int, rows int, offset uint32, buf []byte,
+) (endOffset uint32) {
+	switch col {
+	case cockroachColRoachKey:
+		return kw.roachKeys.Finish(0, rows, offset, buf)
+	case cockroachColMVCCWallTime:
+		return kw.wallTimes.Finish(0, rows, offset, buf)
+	case cockroachColMVCCLogical:
+		return kw.logicalTimes.Finish(0, rows, offset, buf)
+	case cockroachColUntypedVersion:
+		return kw.untypedVersions.Finish(0, rows, offset, buf)
+	default:
+		panic(fmt.Sprintf("unknown default key column: %d", col))
+	}
+}
+
+var cockroachKeySeekerPool = sync.Pool{
+	New: func() interface{} { return &cockroachKeySeeker{} },
+}
+
+type cockroachKeySeeker struct {
+	roachKeys       colblk.PrefixBytes
+	roachKeyChanged colblk.Bitmap
+	mvccWallTimes   colblk.UnsafeUints
+	mvccLogical     colblk.UnsafeUints
+	untypedVersions colblk.RawBytes
+}
+
+var _ colblk.KeySeeker = (*cockroachKeySeeker)(nil)
+
+// Init is part of the KeySeeker interface.
+func (ks *cockroachKeySeeker) Init(d *colblk.DataBlockDecoder) error {
+	bd := d.BlockDecoder()
+	ks.roachKeys = bd.PrefixBytes(cockroachColRoachKey)
+	ks.roachKeyChanged = d.PrefixChanged()
+	ks.mvccWallTimes = bd.Uints(cockroachColMVCCWallTime)
+	ks.mvccLogical = bd.Uints(cockroachColMVCCLogical)
+	ks.untypedVersions = bd.RawBytes(cockroachColUntypedVersion)
+	return nil
+}
+
+// CompareFirstUserKey compares the provided key to the first user key
+// contained within the data block. It's equivalent to performing
+//
+//	Compare(firstUserKey, k)
+func (ks *cockroachKeySeeker) IsLowerBound(k []byte, syntheticSuffix []byte) bool {
+	roachKey, untypedVersion, wallTime, logicalTime := DecodeEngineKey(k)
+	if v := Compare(ks.roachKeys.UnsafeFirstSlice(), roachKey); v != 0 {
+		return v > 0
+	}
+	if len(syntheticSuffix) > 0 {
+		return Compare(syntheticSuffix, k[len(roachKey)+1:]) >= 0
+	}
+	if len(untypedVersion) > 0 {
+		if invariants.Enabled && ks.mvccWallTimes.At(0) != 0 {
+			panic("comparing timestamp with untyped suffix")
+		}
+		return Compare(ks.untypedVersions.At(0), untypedVersion) >= 0
+	}
+	if v := cmp.Compare(ks.mvccWallTimes.At(0), wallTime); v != 0 {
+		return v > 0
+	}
+	return cmp.Compare(uint32(ks.mvccLogical.At(0)), logicalTime) >= 0
+}
+
+// SeekGE is part of the KeySeeker interface.
+func (ks *cockroachKeySeeker) SeekGE(
+	key []byte, boundRow int, searchDir int8,
+) (row int, equalPrefix bool) {
+	// TODO(jackson): Inline Split.
+	si := Split(key)
+	row, eq := ks.roachKeys.Search(key[:si-1])
+	if eq {
+		return ks.seekGEOnSuffix(row, key[si:]), true
+	}
+	return row, false
+}
+
+// seekGEOnSuffix is a helper function for SeekGE when a seek key's prefix
+// exactly matches a row. seekGEOnSuffix finds the first row at index or later
+// with the same prefix as index and a suffix greater than or equal to [suffix],
+// or if no such row exists, the next row with a different prefix.
+func (ks *cockroachKeySeeker) seekGEOnSuffix(index int, seekSuffix []byte) (row int) {
+	// The search key's prefix exactly matches the prefix of the row at index.
+	const withWall = 9
+	const withLogical = withWall + 4
+	const withSynthetic = withLogical + 1
+	var seekWallTime uint64
+	var seekLogicalTime uint32
+	switch len(seekSuffix) {
+	case 0:
+		// The search key has no suffix, so it's the smallest possible key with
+		// its prefix. Return the row. This is a common case where the user is
+		// seeking to the most-recent row and just wants the smallest key with
+		// the prefix.
+		return index
+	case withLogical, withSynthetic:
+		seekWallTime = binary.BigEndian.Uint64(seekSuffix)
+		seekLogicalTime = binary.BigEndian.Uint32(seekSuffix[8:])
+	case withWall:
+		seekWallTime = binary.BigEndian.Uint64(seekSuffix)
+	default:
+		// The suffix is untyped. Compare the untyped suffixes.
+		// Binary search between [index, prefixChanged.SeekSetBitGE(index+1)].
+		//
+		// Define f(i) = true iff key at i is >= seek key.
+		// Invariant: f(l-1) == false, f(u) == true.
+		l := index
+		u := ks.roachKeyChanged.SeekSetBitGE(index + 1)
+		for l < u {
+			h := int(uint(l+u) >> 1) // avoid overflow when computing h
+			// l ≤ h < u
+			if bytes.Compare(ks.untypedVersions.At(h), seekSuffix) >= 0 {
+				u = h // preserves f(u) == true
+			} else {
+				l = h + 1 // preserves f(l-1) == false
+			}
+		}
+		return l
+	}
+	// Seeking among MVCC versions using a MVCC timestamp.
+
+	// TODO(jackson): What if the row has an untyped suffix?
+
+	// First check the suffix at index, because querying for the latest value is
+	// the most common case.
+	if latestWallTime := ks.mvccWallTimes.At(index); latestWallTime < seekWallTime ||
+		(latestWallTime == seekWallTime && uint32(ks.mvccLogical.At(index)) <= seekLogicalTime) {
+		return index
+	}
+
+	// Binary search between [index+1, prefixChanged.SeekSetBitGE(index+1)].
+	//
+	// Define f(i) = true iff key at i is >= seek key.
+	// Invariant: f(l-1) == false, f(u) == true.
+	l := index + 1
+	u := ks.roachKeyChanged.SeekSetBitGE(index + 1)
+	for l < u {
+		h := int(uint(l+u) >> 1) // avoid overflow when computing h
+		// l ≤ h < u
+		hWallTime := ks.mvccWallTimes.At(h)
+		if hWallTime < seekWallTime ||
+			(hWallTime == seekWallTime && uint32(ks.mvccLogical.At(h)) <= seekLogicalTime) {
+			u = h // preserves f(u) = true
+		} else {
+			l = h + 1 // preserves f(l-1) = false
+		}
+	}
+	return l
+}
+
+// MaterializeUserKey is part of the KeySeeker interface.
+func (ks *cockroachKeySeeker) MaterializeUserKey(
+	ki *colblk.PrefixBytesIter, prevRow, row int,
+) []byte {
+	if prevRow+1 == row && prevRow >= 0 {
+		ks.roachKeys.SetNext(ki)
+	} else {
+		ks.roachKeys.SetAt(ki, row)
+	}
+
+	roachKeyLen := len(ki.Buf)
+	ptr := unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(ki.Buf))) + uintptr(roachKeyLen))
+	mvccWall := ks.mvccWallTimes.At(row)
+	mvccLogical := uint32(ks.mvccLogical.At(row))
+	if mvccWall == 0 && mvccLogical == 0 {
+		// This is not an MVCC key. Use the untyped suffix.
+		untypedVersion := ks.untypedVersions.At(row)
+		if len(untypedVersion) == 0 {
+			res := ki.Buf[:roachKeyLen+1]
+			res[roachKeyLen] = 0
+			return res
+		}
+		// Slice first, to check that the capacity is sufficient.
+		res := ki.Buf[:roachKeyLen+1+len(untypedVersion)]
+		*(*byte)(ptr) = 0
+		memmove(
+			unsafe.Pointer(uintptr(ptr)+1),
+			unsafe.Pointer(unsafe.SliceData(untypedVersion)),
+			uintptr(len(untypedVersion)),
+		)
+		return res
+	}
+
+	// Inline binary.BigEndian.PutUint64. Note that this code is converted into
+	// word-size instructions by the compiler.
+	*(*byte)(ptr) = 0
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 1)) = byte(mvccWall >> 56)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 2)) = byte(mvccWall >> 48)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 3)) = byte(mvccWall >> 40)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 4)) = byte(mvccWall >> 32)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 5)) = byte(mvccWall >> 24)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 6)) = byte(mvccWall >> 16)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 7)) = byte(mvccWall >> 8)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 8)) = byte(mvccWall)
+
+	ptr = unsafe.Pointer(uintptr(ptr) + 9)
+	// This is an MVCC key.
+	if mvccLogical == 0 {
+		*(*byte)(ptr) = 9
+		return ki.Buf[:len(ki.Buf)+10]
+	}
+
+	// Inline binary.BigEndian.PutUint32.
+	*(*byte)(ptr) = byte(mvccWall >> 24)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 1)) = byte(mvccWall >> 16)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 2)) = byte(mvccWall >> 8)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 3)) = byte(mvccWall)
+	*(*byte)(unsafe.Pointer(uintptr(ptr) + 4)) = 13
+	return ki.Buf[:len(ki.Buf)+14]
+}
+
+// MaterializeUserKeyWithSyntheticSuffix is part of the KeySeeker interface.
+func (ks *cockroachKeySeeker) MaterializeUserKeyWithSyntheticSuffix(
+	ki *colblk.PrefixBytesIter, suffix []byte, prevRow, row int,
+) []byte {
+	if prevRow+1 == row && prevRow >= 0 {
+		ks.roachKeys.SetNext(ki)
+	} else {
+		ks.roachKeys.SetAt(ki, row)
+	}
+
+	// Slice first, to check that the capacity is sufficient.
+	res := ki.Buf[:len(ki.Buf)+1+len(suffix)]
+	ptr := unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(ki.Buf))) + uintptr(len(ki.Buf)))
+	*(*byte)(ptr) = 0
+	memmove(unsafe.Pointer(uintptr(ptr)+1), unsafe.Pointer(unsafe.SliceData(suffix)), uintptr(len(suffix)))
+	return res
+}
+
+// Release is part of the KeySeeker interface.
+func (ks *cockroachKeySeeker) Release() {
+	*ks = cockroachKeySeeker{}
+	cockroachKeySeekerPool.Put(ks)
+}
+
+//go:linkname memmove runtime.memmove
+func memmove(to, from unsafe.Pointer, n uintptr)
diff --git a/level_checker_test.go b/level_checker_test.go
index da26e4b9..03c97ee5 100644
--- a/level_checker_test.go
+++ b/level_checker_test.go
@@ -20,11 +20,11 @@ import (
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
-	"github.com/cockroachdb/pebble/internal/rangedel"
 	"github.com/cockroachdb/pebble/internal/sstableinternal"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -177,8 +177,9 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 					}
 				}
 				writerOpts := sstable.WriterOptions{
-					TableFormat: FormatNewest.MaxTableFormat(),
 					Comparer:    testkeys.Comparer,
+					KeySchema:   colblk.DefaultKeySchema(testkeys.Comparer, 16),
+					TableFormat: FormatNewest.MaxTableFormat(),
 				}
 				writerOpts.SetInternal(sstableinternal.WriterOptions{
 					DisableKeyOrderChecks: disableKeyOrderChecks,
@@ -192,22 +193,26 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 						tombstones = append(tombstones, fragmented)
 					},
 				}
-				keyvalues := strings.Fields(line)
-				for _, kv := range keyvalues {
-					j := strings.Index(kv, ":")
-					ikey := base.ParseInternalKey(kv[:j])
-					value := []byte(kv[j+1:])
-					var err error
-					switch ikey.Kind() {
-					case InternalKeyKindRangeDelete:
+				keyValues := strings.Fields(line)
+				for _, kv := range keyValues {
+					if strings.HasPrefix(kv, "EncodeSpan:") {
+						kv = kv[len("EncodeSpan:"):]
+						s := keyspan.ParseSpan(kv)
 						if writeUnfragmented {
-							err = w.AddWithForceObsolete(ikey, value, false /* forceObsolete */)
-							break
+							if err = w.EncodeSpan(s); err != nil {
+								return err.Error()
+							}
+						} else if s.Keys[0].Kind() == base.InternalKeyKindRangeDelete {
+							frag.Add(s)
+						} else {
+							t.Fatalf("unexpected span: %s", s.Pretty(testkeys.Comparer.FormatKey))
 						}
-						frag.Add(rangedel.Decode(ikey, value, nil))
-					default:
-						err = w.AddWithForceObsolete(ikey, value, false /* forceObsolete */)
+						continue
 					}
+					j := strings.Index(kv, ":")
+					ikey := base.ParseInternalKey(kv[:j])
+					value := []byte(kv[j+1:])
+					err = w.AddWithForceObsolete(ikey, value, false /* forceObsolete */)
 					if err != nil {
 						return err.Error()
 					}
@@ -230,7 +235,10 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 					return err.Error()
 				}
 				// Set FileNum for logging purposes.
-				readerOpts := sstable.ReaderOptions{Comparer: testkeys.Comparer}
+				readerOpts := sstable.ReaderOptions{
+					Comparer:  testkeys.Comparer,
+					KeySchema: writerOpts.KeySchema,
+				}
 				readerOpts.SetInternalCacheOpts(sstableinternal.CacheOptions{FileNum: base.DiskFileNum(fileNum - 1)})
 				r, err := sstable.NewReader(context.Background(), readable, readerOpts)
 				if err != nil {
diff --git a/sstable/block_property_test.go b/sstable/block_property_test.go
index af29a2f9..d7c4fb7a 100644
--- a/sstable/block_property_test.go
+++ b/sstable/block_property_test.go
@@ -902,7 +902,7 @@ func TestBlockProperties(t *testing.T) {
 
 				// Enumerate point key data blocks encoded into the index.
 				if f != nil {
-					indexH, err := r.readIndex(context.Background(), nil, nil, nil)
+					indexH, err := r.readTopLevelIndexBlock(context.Background(), noEnv, noReadHandle)
 					if err != nil {
 						return err.Error()
 					}
@@ -1271,7 +1271,7 @@ func runBlockPropertiesBuildCmd(td *datadriven.TestData) (r *Reader, out string)
 }
 
 func runBlockPropsCmd(r *Reader) string {
-	bh, err := r.readIndex(context.Background(), nil, nil, nil)
+	bh, err := r.readTopLevelIndexBlock(context.Background(), noEnv, noReadHandle)
 	if err != nil {
 		return err.Error()
 	}
@@ -1320,8 +1320,7 @@ func runBlockPropsCmd(r *Reader) string {
 		// If the table has a two-level index, also decode the index
 		// block that bhp points to, along with its block properties.
 		if twoLevelIndex {
-			subIndex, err := r.readBlock(
-				context.Background(), bhp.Handle, nil, nil, nil, nil)
+			subIndex, err := r.readIndexBlock(context.Background(), noEnv, noReadHandle, bhp.Handle)
 			if err != nil {
 				return err.Error()
 			}
diff --git a/sstable/colblk/cockroach_test.go b/sstable/colblk/cockroach_test.go
index 53be50b6..47bea118 100644
--- a/sstable/colblk/cockroach_test.go
+++ b/sstable/colblk/cockroach_test.go
@@ -2,399 +2,24 @@
 // of this source code is governed by a BSD-style license that can be found in
 // the LICENSE file.
 
-package colblk
+package colblk_test
 
 import (
 	"bytes"
-	"cmp"
-	"encoding/binary"
 	"fmt"
-	"io"
 	"math/rand/v2"
 	"slices"
-	"sync"
 	"testing"
 	"time"
-	"unsafe"
 
-	"github.com/cockroachdb/crlib/crbytes"
 	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/crdbtest"
-	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/stretchr/testify/require"
 )
 
-const (
-	cockroachColRoachKey int = iota
-	cockroachColMVCCWallTime
-	cockroachColMVCCLogical
-	cockroachColUntypedVersion
-	cockroachColCount
-)
-
-var cockroachKeySchema = KeySchema{
-	ColumnTypes: []DataType{
-		cockroachColRoachKey:       DataTypePrefixBytes,
-		cockroachColMVCCWallTime:   DataTypeUint,
-		cockroachColMVCCLogical:    DataTypeUint,
-		cockroachColUntypedVersion: DataTypeBytes,
-	},
-	NewKeyWriter: func() KeyWriter {
-		kw := &cockroachKeyWriter{}
-		kw.roachKeys.Init(16)
-		kw.wallTimes.Init()
-		kw.logicalTimes.InitWithDefault()
-		kw.untypedVersions.Init()
-		return kw
-	},
-	NewKeySeeker: func() KeySeeker {
-		return &cockroachKeySeeker{}
-	},
-}
-
-type cockroachKeyWriter struct {
-	roachKeys       PrefixBytesBuilder
-	wallTimes       UintBuilder
-	logicalTimes    UintBuilder
-	untypedVersions RawBytesBuilder
-	prevSuffix      []byte
-}
-
-func (kw *cockroachKeyWriter) ComparePrev(key []byte) KeyComparison {
-	var cmpv KeyComparison
-	cmpv.PrefixLen = int32(crdbtest.Split(key)) // TODO(jackson): Inline
-	if kw.roachKeys.Rows() == 0 {
-		cmpv.UserKeyComparison = 1
-		return cmpv
-	}
-	lp := kw.roachKeys.UnsafeGet(kw.roachKeys.Rows() - 1)
-	cmpv.CommonPrefixLen = int32(crbytes.CommonPrefix(lp, key[:cmpv.PrefixLen-1]))
-	if cmpv.CommonPrefixLen == cmpv.PrefixLen-1 {
-		// Adjust CommonPrefixLen to include the sentinel byte,
-		cmpv.CommonPrefixLen = cmpv.PrefixLen
-		cmpv.UserKeyComparison = int32(crdbtest.CompareSuffixes(key[cmpv.PrefixLen:], kw.prevSuffix))
-		return cmpv
-	}
-	// The keys have different MVCC prefixes. We haven't determined which is
-	// greater, but we know the index at which they diverge. The base.Comparer
-	// contract dictates that prefixes must be lexicographically ordered.
-	if len(lp) == int(cmpv.CommonPrefixLen) {
-		// cmpv.PrefixLen > cmpv.PrefixLenShared; key is greater.
-		cmpv.UserKeyComparison = +1
-	} else {
-		// Both keys have at least 1 additional byte at which they diverge.
-		// Compare the diverging byte.
-		cmpv.UserKeyComparison = int32(cmp.Compare(key[cmpv.CommonPrefixLen], lp[cmpv.CommonPrefixLen]))
-	}
-	return cmpv
-}
-
-func (kw *cockroachKeyWriter) WriteKey(
-	row int, key []byte, keyPrefixLen, keyPrefixLenSharedWithPrev int32,
-) {
-	// TODO(jackson): Avoid copying the previous suffix.
-	// TODO(jackson): Use keyPrefixLen to speed up decoding.
-	roachKey, untypedVersion, wallTime, logicalTime := crdbtest.DecodeEngineKey(key)
-	kw.prevSuffix = append(kw.prevSuffix[:0], key[keyPrefixLen:]...)
-	// When the roach key is the same, keyPrefixLenSharedWithPrev includes the
-	// separator byte.
-	kw.roachKeys.Put(roachKey, min(int(keyPrefixLenSharedWithPrev), len(roachKey)))
-	kw.wallTimes.Set(row, wallTime)
-	// The w.logicalTimes builder was initialized with InitWithDefault, so if we
-	// don't set a value, the column value is implicitly zero. We only need to
-	// Set anything for non-zero values.
-	if logicalTime > 0 {
-		kw.logicalTimes.Set(row, uint64(logicalTime))
-	}
-	kw.untypedVersions.Put(untypedVersion)
-}
-
-func (kw *cockroachKeyWriter) MaterializeKey(dst []byte, i int) []byte {
-	dst = append(dst, kw.roachKeys.UnsafeGet(i)...)
-	// Append separator byte.
-	dst = append(dst, 0)
-	if untypedVersion := kw.untypedVersions.UnsafeGet(i); len(untypedVersion) > 0 {
-		dst = append(dst, untypedVersion...)
-		dst = append(dst, byte(len(untypedVersion)))
-		return dst
-	}
-	return crdbtest.AppendTimestamp(dst, kw.wallTimes.Get(i), uint32(kw.logicalTimes.Get(i)))
-}
-
-func (kw *cockroachKeyWriter) Reset() {
-	kw.roachKeys.Reset()
-	kw.wallTimes.Reset()
-	kw.logicalTimes.Reset()
-	kw.untypedVersions.Reset()
-}
-
-func (kw *cockroachKeyWriter) WriteDebug(dst io.Writer, rows int) {
-	fmt.Fprint(dst, "prefixes: ")
-	kw.roachKeys.WriteDebug(dst, rows)
-	fmt.Fprintln(dst)
-	fmt.Fprint(dst, "wall times: ")
-	kw.wallTimes.WriteDebug(dst, rows)
-	fmt.Fprintln(dst)
-	fmt.Fprint(dst, "logical times: ")
-	kw.logicalTimes.WriteDebug(dst, rows)
-	fmt.Fprintln(dst)
-	fmt.Fprint(dst, "untyped suffixes: ")
-	kw.untypedVersions.WriteDebug(dst, rows)
-	fmt.Fprintln(dst)
-}
-
-func (kw *cockroachKeyWriter) NumColumns() int {
-	return cockroachColCount
-}
-
-func (kw *cockroachKeyWriter) DataType(col int) DataType {
-	return cockroachKeySchema.ColumnTypes[col]
-}
-
-func (kw *cockroachKeyWriter) Size(rows int, offset uint32) uint32 {
-	offset = kw.roachKeys.Size(rows, offset)
-	offset = kw.wallTimes.Size(rows, offset)
-	offset = kw.logicalTimes.Size(rows, offset)
-	offset = kw.untypedVersions.Size(rows, offset)
-	return offset
-}
-
-func (kw *cockroachKeyWriter) Finish(
-	col int, rows int, offset uint32, buf []byte,
-) (endOffset uint32) {
-	switch col {
-	case cockroachColRoachKey:
-		return kw.roachKeys.Finish(0, rows, offset, buf)
-	case cockroachColMVCCWallTime:
-		return kw.wallTimes.Finish(0, rows, offset, buf)
-	case cockroachColMVCCLogical:
-		return kw.logicalTimes.Finish(0, rows, offset, buf)
-	case cockroachColUntypedVersion:
-		return kw.untypedVersions.Finish(0, rows, offset, buf)
-	default:
-		panic(fmt.Sprintf("unknown default key column: %d", col))
-	}
-}
-
-var cockroachKeySeekerPool = sync.Pool{
-	New: func() interface{} { return &cockroachKeySeeker{} },
-}
-
-type cockroachKeySeeker struct {
-	roachKeys       PrefixBytes
-	roachKeyChanged Bitmap
-	mvccWallTimes   UnsafeUints
-	mvccLogical     UnsafeUints
-	untypedVersions RawBytes
-}
-
-var _ KeySeeker = (*cockroachKeySeeker)(nil)
-
-// Init is part of the KeySeeker interface.
-func (ks *cockroachKeySeeker) Init(d *DataBlockDecoder) error {
-	ks.roachKeys = d.d.PrefixBytes(cockroachColRoachKey)
-	ks.roachKeyChanged = d.prefixChanged
-	ks.mvccWallTimes = d.d.Uints(cockroachColMVCCWallTime)
-	ks.mvccLogical = d.d.Uints(cockroachColMVCCLogical)
-	ks.untypedVersions = d.d.RawBytes(cockroachColUntypedVersion)
-	return nil
-}
-
-// CompareFirstUserKey compares the provided key to the first user key
-// contained within the data block. It's equivalent to performing
-//
-//	Compare(firstUserKey, k)
-func (ks *cockroachKeySeeker) IsLowerBound(k []byte, syntheticSuffix []byte) bool {
-	roachKey, untypedVersion, wallTime, logicalTime := crdbtest.DecodeEngineKey(k)
-	if v := crdbtest.Compare(ks.roachKeys.UnsafeFirstSlice(), roachKey); v != 0 {
-		return v > 0
-	}
-	if len(syntheticSuffix) > 0 {
-		return crdbtest.Compare(syntheticSuffix, k[len(roachKey)+1:]) >= 0
-	}
-	if len(untypedVersion) > 0 {
-		if invariants.Enabled && ks.mvccWallTimes.At(0) != 0 {
-			panic("comparing timestamp with untyped suffix")
-		}
-		return crdbtest.Compare(ks.untypedVersions.At(0), untypedVersion) >= 0
-	}
-	if v := cmp.Compare(ks.mvccWallTimes.At(0), wallTime); v != 0 {
-		return v > 0
-	}
-	return cmp.Compare(uint32(ks.mvccLogical.At(0)), logicalTime) >= 0
-}
-
-// SeekGE is part of the KeySeeker interface.
-func (ks *cockroachKeySeeker) SeekGE(
-	key []byte, boundRow int, searchDir int8,
-) (row int, equalPrefix bool) {
-	// TODO(jackson): Inline crdbtest.Split.
-	si := crdbtest.Split(key)
-	row, eq := ks.roachKeys.Search(key[:si-1])
-	if eq {
-		return ks.seekGEOnSuffix(row, key[si:]), true
-	}
-	return row, false
-}
-
-// seekGEOnSuffix is a helper function for SeekGE when a seek key's prefix
-// exactly matches a row. seekGEOnSuffix finds the first row at index or later
-// with the same prefix as index and a suffix greater than or equal to [suffix],
-// or if no such row exists, the next row with a different prefix.
-func (ks *cockroachKeySeeker) seekGEOnSuffix(index int, seekSuffix []byte) (row int) {
-	// The search key's prefix exactly matches the prefix of the row at index.
-	const withWall = 9
-	const withLogical = withWall + 4
-	const withSynthetic = withLogical + 1
-	var seekWallTime uint64
-	var seekLogicalTime uint32
-	switch len(seekSuffix) {
-	case 0:
-		// The search key has no suffix, so it's the smallest possible key with
-		// its prefix. Return the row. This is a common case where the user is
-		// seeking to the most-recent row and just wants the smallest key with
-		// the prefix.
-		return index
-	case withLogical, withSynthetic:
-		seekWallTime = binary.BigEndian.Uint64(seekSuffix)
-		seekLogicalTime = binary.BigEndian.Uint32(seekSuffix[8:])
-	case withWall:
-		seekWallTime = binary.BigEndian.Uint64(seekSuffix)
-	default:
-		// The suffix is untyped. Compare the untyped suffixes.
-		// Binary search between [index, prefixChanged.SeekSetBitGE(index+1)].
-		//
-		// Define f(i) = true iff key at i is >= seek key.
-		// Invariant: f(l-1) == false, f(u) == true.
-		l := index
-		u := ks.roachKeyChanged.SeekSetBitGE(index + 1)
-		for l < u {
-			h := int(uint(l+u) >> 1) // avoid overflow when computing h
-			// l ≤ h < u
-			if bytes.Compare(ks.untypedVersions.At(h), seekSuffix) >= 0 {
-				u = h // preserves f(u) == true
-			} else {
-				l = h + 1 // preserves f(l-1) == false
-			}
-		}
-		return l
-	}
-	// Seeking among MVCC versions using a MVCC timestamp.
-
-	// TODO(jackson): What if the row has an untyped suffix?
-
-	// First check the suffix at index, because querying for the latest value is
-	// the most common case.
-	if latestWallTime := ks.mvccWallTimes.At(index); latestWallTime < seekWallTime ||
-		(latestWallTime == seekWallTime && uint32(ks.mvccLogical.At(index)) <= seekLogicalTime) {
-		return index
-	}
-
-	// Binary search between [index+1, prefixChanged.SeekSetBitGE(index+1)].
-	//
-	// Define f(i) = true iff key at i is >= seek key.
-	// Invariant: f(l-1) == false, f(u) == true.
-	l := index + 1
-	u := ks.roachKeyChanged.SeekSetBitGE(index + 1)
-	for l < u {
-		h := int(uint(l+u) >> 1) // avoid overflow when computing h
-		// l ≤ h < u
-		hWallTime := ks.mvccWallTimes.At(h)
-		if hWallTime < seekWallTime ||
-			(hWallTime == seekWallTime && uint32(ks.mvccLogical.At(h)) <= seekLogicalTime) {
-			u = h // preserves f(u) = true
-		} else {
-			l = h + 1 // preserves f(l-1) = false
-		}
-	}
-	return l
-}
-
-// MaterializeUserKey is part of the KeySeeker interface.
-func (ks *cockroachKeySeeker) MaterializeUserKey(ki *PrefixBytesIter, prevRow, row int) []byte {
-	if prevRow+1 == row && prevRow >= 0 {
-		ks.roachKeys.SetNext(ki)
-	} else {
-		ks.roachKeys.SetAt(ki, row)
-	}
-
-	roachKeyLen := len(ki.buf)
-	ptr := unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(ki.buf))) + uintptr(roachKeyLen))
-	mvccWall := ks.mvccWallTimes.At(row)
-	mvccLogical := uint32(ks.mvccLogical.At(row))
-	if mvccWall == 0 && mvccLogical == 0 {
-		// This is not an MVCC key. Use the untyped suffix.
-		untypedVersion := ks.untypedVersions.At(row)
-		if len(untypedVersion) == 0 {
-			res := ki.buf[:roachKeyLen+1]
-			res[roachKeyLen] = 0
-			return res
-		}
-		// Slice first, to check that the capacity is sufficient.
-		res := ki.buf[:roachKeyLen+1+len(untypedVersion)]
-		*(*byte)(ptr) = 0
-		memmove(
-			unsafe.Pointer(uintptr(ptr)+1),
-			unsafe.Pointer(unsafe.SliceData(untypedVersion)),
-			uintptr(len(untypedVersion)),
-		)
-		return res
-	}
-
-	// Inline binary.BigEndian.PutUint64. Note that this code is converted into
-	// word-size instructions by the compiler.
-	*(*byte)(ptr) = 0
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 1)) = byte(mvccWall >> 56)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 2)) = byte(mvccWall >> 48)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 3)) = byte(mvccWall >> 40)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 4)) = byte(mvccWall >> 32)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 5)) = byte(mvccWall >> 24)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 6)) = byte(mvccWall >> 16)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 7)) = byte(mvccWall >> 8)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 8)) = byte(mvccWall)
-
-	ptr = unsafe.Pointer(uintptr(ptr) + 9)
-	// This is an MVCC key.
-	if mvccLogical == 0 {
-		*(*byte)(ptr) = 9
-		return ki.buf[:len(ki.buf)+10]
-	}
-
-	// Inline binary.BigEndian.PutUint32.
-	*(*byte)(ptr) = byte(mvccWall >> 24)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 1)) = byte(mvccWall >> 16)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 2)) = byte(mvccWall >> 8)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 3)) = byte(mvccWall)
-	*(*byte)(unsafe.Pointer(uintptr(ptr) + 4)) = 13
-	return ki.buf[:len(ki.buf)+14]
-}
-
-// MaterializeUserKeyWithSyntheticSuffix is part of the KeySeeker interface.
-func (ks *cockroachKeySeeker) MaterializeUserKeyWithSyntheticSuffix(
-	ki *PrefixBytesIter, suffix []byte, prevRow, row int,
-) []byte {
-	if prevRow+1 == row && prevRow >= 0 {
-		ks.roachKeys.SetNext(ki)
-	} else {
-		ks.roachKeys.SetAt(ki, row)
-	}
-
-	// Slice first, to check that the capacity is sufficient.
-	res := ki.buf[:len(ki.buf)+1+len(suffix)]
-	ptr := unsafe.Pointer(uintptr(unsafe.Pointer(unsafe.SliceData(ki.buf))) + uintptr(len(ki.buf)))
-	*(*byte)(ptr) = 0
-	memmove(unsafe.Pointer(uintptr(ptr)+1), unsafe.Pointer(unsafe.SliceData(suffix)), uintptr(len(suffix)))
-	return res
-}
-
-// Release is part of the KeySeeker interface.
-func (ks *cockroachKeySeeker) Release() {
-	*ks = cockroachKeySeeker{}
-	cockroachKeySeekerPool.Put(ks)
-}
-
 func TestCockroachDataBlock(t *testing.T) {
 	const targetBlockSize = 32 << 10
 	const valueLen = 100
@@ -408,12 +33,12 @@ func TestCockroachDataBlock(t *testing.T) {
 		BaseWallTime:      seed,
 	}, valueLen)
 
-	var decoder DataBlockDecoder
-	var it DataBlockIter
-	it.InitOnce(cockroachKeySchema, crdbtest.Compare, crdbtest.Split, getLazyValuer(func([]byte) base.LazyValue {
+	var decoder colblk.DataBlockDecoder
+	var it colblk.DataBlockIter
+	it.InitOnce(crdbtest.KeySchema, crdbtest.Compare, crdbtest.Split, getLazyValuer(func([]byte) base.LazyValue {
 		return base.LazyValue{ValueOrHandle: []byte("mock external value")}
 	}))
-	decoder.Init(cockroachKeySchema, serializedBlock)
+	decoder.Init(crdbtest.KeySchema, serializedBlock)
 	if err := it.Init(&decoder, block.IterTransforms{}); err != nil {
 		t.Fatal(err)
 	}
@@ -455,8 +80,8 @@ func generateDataBlock(
 ) (data []byte, keys [][]byte, values [][]byte) {
 	keys, values = crdbtest.RandomKVs(rng, targetBlockSize/valueLen, cfg, valueLen)
 
-	var w DataBlockEncoder
-	w.Init(cockroachKeySchema)
+	var w colblk.DataBlockEncoder
+	w.Init(crdbtest.KeySchema)
 	count := 0
 	for w.Size() < targetBlockSize {
 		ik := base.MakeInternalKey(keys[count], base.SeqNum(rng.Uint64N(uint64(base.SeqNumMax))), base.InternalKeyKindSet)
@@ -497,8 +122,8 @@ func benchmarkCockroachDataBlockWriter(b *testing.B, keyConfig crdbtest.KeyConfi
 	rng := rand.New(rand.NewPCG(0, seed))
 	_, keys, values := generateDataBlock(rng, targetBlockSize, keyConfig, valueLen)
 
-	var w DataBlockEncoder
-	w.Init(cockroachKeySchema)
+	var w colblk.DataBlockEncoder
+	w.Init(crdbtest.KeySchema)
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
@@ -634,12 +259,12 @@ func benchmarkCockroachDataBlockIter(
 
 	serializedBlock, keys, _ := generateDataBlock(rng, targetBlockSize, cfg.KeyConfig, cfg.ValueLen)
 
-	var decoder DataBlockDecoder
-	var it DataBlockIter
-	it.InitOnce(cockroachKeySchema, crdbtest.Compare, crdbtest.Split, getLazyValuer(func([]byte) base.LazyValue {
+	var decoder colblk.DataBlockDecoder
+	var it colblk.DataBlockIter
+	it.InitOnce(crdbtest.KeySchema, crdbtest.Compare, crdbtest.Split, getLazyValuer(func([]byte) base.LazyValue {
 		return base.LazyValue{ValueOrHandle: []byte("mock external value")}
 	}))
-	decoder.Init(cockroachKeySchema, serializedBlock)
+	decoder.Init(crdbtest.KeySchema, serializedBlock)
 	if err := it.Init(&decoder, transforms); err != nil {
 		b.Fatal(err)
 	}
@@ -689,3 +314,9 @@ func benchmarkCockroachDataBlockIter(
 		})
 	}
 }
+
+type getLazyValuer func([]byte) base.LazyValue
+
+func (g getLazyValuer) GetLazyValueForPrefixAndValueHandle(handle []byte) base.LazyValue {
+	return g(handle)
+}
diff --git a/sstable/colblk/prefix_bytes_test.go b/sstable/colblk/prefix_bytes_test.go
index 8858959d..31351a5a 100644
--- a/sstable/colblk/prefix_bytes_test.go
+++ b/sstable/colblk/prefix_bytes_test.go
@@ -283,7 +283,7 @@ func BenchmarkPrefixBytes(b *testing.B) {
 			pb, _ := DecodePrefixBytes(buf, 0, n)
 			b.ResetTimer()
 			var pbi PrefixBytesIter
-			pbi.buf = make([]byte, 0, maxLen)
+			pbi.Buf = make([]byte, 0, maxLen)
 			for i := 0; i < b.N; i++ {
 				j := i % n
 				if j == 0 {
@@ -291,9 +291,9 @@ func BenchmarkPrefixBytes(b *testing.B) {
 				} else {
 					pb.SetNext(&pbi)
 				}
-				if invariants.Enabled && !bytes.Equal(pbi.buf, userKeys[j]) {
+				if invariants.Enabled && !bytes.Equal(pbi.Buf, userKeys[j]) {
 					b.Fatalf("Constructed key %q (%q, %q, %q) for index %d; expected %q",
-						pbi.buf, pb.SharedPrefix(), pb.RowBundlePrefix(j), pb.RowSuffix(j), j, userKeys[j])
+						pbi.Buf, pb.SharedPrefix(), pb.RowBundlePrefix(j), pb.RowSuffix(j), j, userKeys[j])
 				}
 			}
 		})
diff --git a/sstable/reader_iter_test.go b/sstable/reader_iter_test.go
index ad1331da..a16793ce 100644
--- a/sstable/reader_iter_test.go
+++ b/sstable/reader_iter_test.go
@@ -51,6 +51,7 @@ func TestIteratorErrorOnInit(t *testing.T) {
 
 	var pool block.BufferPool
 	pool.Init(5)
+	defer pool.Release()
 
 	var stats base.InternalIteratorStats
 	for k := 0; k < 20; k++ {
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index c6652036..d736828b 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -47,7 +47,7 @@ func (r *Reader) get(key []byte) (value []byte, err error) {
 	}
 
 	if r.tableFilter != nil {
-		dataH, err := r.readFilter(context.Background(), nil, nil /* stats */, nil)
+		dataH, err := r.readFilterBlock(context.Background(), noEnv, noReadHandle)
 		if err != nil {
 			return nil, err
 		}
@@ -629,7 +629,7 @@ func TestInvalidReader(t *testing.T) {
 }
 
 func indexLayoutString(t *testing.T, r *Reader) string {
-	indexH, err := r.readIndex(context.Background(), nil, nil, nil)
+	indexH, err := r.readTopLevelIndexBlock(context.Background(), noEnv, noReadHandle)
 	require.NoError(t, err)
 	defer indexH.Release()
 	var buf strings.Builder
@@ -646,7 +646,7 @@ func indexLayoutString(t *testing.T, r *Reader) string {
 		require.NoError(t, err)
 		fmt.Fprintf(&buf, " %s: size %d\n", string(iter.Separator()), bh.Length)
 		if twoLevelIndex {
-			b, err := r.readBlock(context.Background(), bh.Handle, nil, nil, nil, nil)
+			b, err := r.readIndexBlock(context.Background(), noEnv, noReadHandle, bh.Handle)
 			require.NoError(t, err)
 			defer b.Release()
 			iter2 := r.tableFormat.newIndexIter()
@@ -948,6 +948,7 @@ func TestReadaheadSetupForV3TablesWithMultipleVersions(t *testing.T) {
 	{
 		var pool block.BufferPool
 		pool.Init(5)
+		defer pool.Release()
 		citer, err := r.NewCompactionIter(
 			NoTransforms, CategoryAndQoS{}, nil, MakeTrivialReaderProvider(r), &pool)
 		require.NoError(t, err)
diff --git a/sstable/table_test.go b/sstable/table_test.go
index 2c5e4b14..4ecf7243 100644
--- a/sstable/table_test.go
+++ b/sstable/table_test.go
@@ -532,8 +532,7 @@ func TestMetaIndexEntriesSorted(t *testing.T) {
 	r, err := newReader(f, ReaderOptions{})
 	require.NoError(t, err)
 
-	b, err := r.readBlock(
-		context.Background(), r.metaIndexBH, nil, nil, nil, nil)
+	b, err := r.readMetaindexBlock(context.Background(), noEnv, noReadHandle)
 	require.NoError(t, err)
 	defer b.Release()
 
diff --git a/testdata/checkpoint b/testdata/checkpoint
index ef8088a3..29156e02 100644
--- a/testdata/checkpoint
+++ b/testdata/checkpoint
@@ -482,6 +482,7 @@ k k
 open: db/000014.sst (options: *vfs.randomReadsOption)
 read-at(509, 53): db/000014.sst
 read-at(472, 37): db/000014.sst
+read-at(53, 419): db/000014.sst
 z z
 .
 
diff --git a/testdata/event_listener b/testdata/event_listener
index 44118fd4..6bc9b66f 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -227,7 +227,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 1.7KB
 Compression types: snappy: 3
-Block cache: 3 entries (485B)  hit rate: 0.0%
+Block cache: 2 entries (43B)  hit rate: 0.0%
 Table cache: 0 entries (0B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -329,7 +329,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 3.5KB
 Compression types: snappy: 6
-Block cache: 9 entries (1.4KB)  hit rate: 0.0%
+Block cache: 6 entries (129B)  hit rate: 0.0%
 Table cache: 0 entries (0B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
diff --git a/testdata/ingest b/testdata/ingest
index 609c84ba..e798b782 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -53,7 +53,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 569B
 Compression types: snappy: 1
-Block cache: 6 entries (945B)  hit rate: 30.8%
+Block cache: 4 entries (88B)  hit rate: 23.1%
 Table cache: 1 entries (784B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
diff --git a/testdata/level_checker b/testdata/level_checker
index f2c661ac..45b3a8ab 100644
--- a/testdata/level_checker
+++ b/testdata/level_checker
@@ -16,10 +16,10 @@
 define
 L
 a.SET.30 e.RANGEDEL.inf
-a.SET.30:30 c.SET.27:27 a.RANGEDEL.8:f
+a.SET.30:30 c.SET.27:27 EncodeSpan:a-f:{(#8,RANGEDEL)}
 L
 e.SET.10 g.SET.20
-e.SET.10:10 g.SET.20:20 e.RANGEDEL.8:f
+e.SET.10:10 g.SET.20:20 EncodeSpan:e-f:{(#8,RANGEDEL)}
 ----
 Level 1
   file 0: [a#30,SET-e#inf,RANGEDEL]
@@ -34,7 +34,7 @@ check
 define
 L
 a.SET.15 f.SET.16
-a.SET.15:15 c.SET.13:13 f.SET.16:16 a.RANGEDEL.12:h
+a.SET.15:15 c.SET.13:13 f.SET.16:16 EncodeSpan:a-h:{(#12,RANGEDEL)}
 L
 e.SET.10 g.SET.15
 e.SET.10:10 g.SET.15:15
@@ -52,10 +52,10 @@ check
 define
 L
 c.SET.30 f.RANGEDEL.0
-c.SET.30:30 d.SET.27:27 a.RANGEDEL.8:f
+c.SET.30:30 d.SET.27:27 EncodeSpan:a-f:{(#8,RANGEDEL)}
 L
 a.SET.10 c.RANGEDEL.inf
-a.SET.10:10 b.SET.12:12 a.RANGEDEL.8:f
+a.SET.10:10 b.SET.12:12 EncodeSpan:a-f:{(#8,RANGEDEL)}
 ----
 Level 1
   file 0: [c#30,SET-f#0,RANGEDEL]
@@ -70,7 +70,7 @@ check
 define
 L
 c.SET.15 g.SET.16
-c.SET.15:15 f.SET.13:13 g.SET.16:16 a.RANGEDEL.12:h
+c.SET.15:15 f.SET.13:13 g.SET.16:16 EncodeSpan:a-h:{(#12,RANGEDEL)}
 L
 b.SET.14 d.SET.10
 b.SET.14:14 d.SET.10:10
@@ -87,10 +87,10 @@ check
 define
 L
 a.SET.30 e.RANGEDEL.inf
-a.SET.30:30 c.SET.27:27 a.RANGEDEL.8:g
+a.SET.30:30 c.SET.27:27 EncodeSpan:a-g:{(#8,RANGEDEL)}
 L
 e.SET.10 g.SET.20
-e.SET.10:10 g.SET.20:20 e.RANGEDEL.8:g
+e.SET.10:10 g.SET.20:20 EncodeSpan:e-g:{(#8,RANGEDEL)}
 ----
 Level 1
   file 0: [a#30,SET-e#inf,RANGEDEL]
@@ -103,7 +103,7 @@ check
 define
 L
 a.SET.30 e.RANGEDEL.inf
-a.SET.30:30 c.SET.27:27 a.RANGEDEL.8:g
+a.SET.30:30 c.SET.27:27 EncodeSpan:a-g:{(#8,RANGEDEL)}
 L
 a.SET.10 g.SET.20
 a.SET.10:10 c.SET.28:28 g.SET.20:20
@@ -122,7 +122,7 @@ found InternalKey c#27,SET in L1: fileNum=000010 and InternalKey c#28,SET in L2:
 define
 L
 a.SET.30 g.RANGEDEL.inf
-a.SET.30:30 c.SET.27:27 a.RANGEDEL.8:g
+a.SET.30:30 c.SET.27:27 EncodeSpan:a-g:{(#8,RANGEDEL)}
 L
 g.SET.10 j.SET.20
 g.SET.10:10 j.SET.20:20
@@ -138,7 +138,7 @@ check
 define
 L
 a.SET.30 g.SET.8
-a.SET.30:30 c.SET.27:27 a.RANGEDEL.8:g g.SET.8:8
+a.SET.30:30 c.SET.27:27 EncodeSpan:a-g:{(#8,RANGEDEL)} g.SET.8:8
 L
 g.SET.10 j.SET.20
 g.SET.10:10 j.SET.20:20
@@ -158,7 +158,7 @@ a.SET.30 g.SET.30
 a.SET.30:30 c.SET.8:8 g.SET.30:30
 L
 a.SET.10 j.SET.20
-a.SET.10:10 j.SET.20:20 b.RANGEDEL.10:g
+a.SET.10:10 j.SET.20:20 EncodeSpan:b-g:{(#10,RANGEDEL)}
 ----
 Level 1
   file 0: [a#30,SET-g#30,SET]
@@ -172,10 +172,10 @@ tombstone b-g:{(#10,RANGEDEL)} in L2: fileNum=000017 deletes key c#8,SET in L1:
 define
 L
 a.RANGEDEL.8 c.RANGEDEL.inf
-a.RANGEDEL.8:c
+EncodeSpan:a-c:{(#8,RANGEDEL)}
 L
 a.RANGEDEL.6 d.RANGEDEL.inf
-a.RANGEDEL.6:d b.RANGEDEL.10:c
+EncodeSpan:a-d:{(#6,RANGEDEL)} EncodeSpan:b-c:{(#10,RANGEDEL)}
 ----
 Level 1
   file 0: [a#8,RANGEDEL-c#inf,RANGEDEL]
@@ -227,7 +227,7 @@ out of order keys b#2,SET >= b#3,SET in L1: fileNum=000023
 define write-unfragmented disable-key-order-checks
 L
 a.RANGEDEL.1 g.RANGEDEL.inf
-d.RANGEDEL.2:e d.RANGEDEL.1:e f.RANGEDEL.3:g a.RANGEDEL.4:b
+EncodeSpan:d-e:{(#2,RANGEDEL)} EncodeSpan:d-e:{(#1,RANGEDEL)} EncodeSpan:f-g:{(#3,RANGEDEL)} EncodeSpan:a-b:{(#4,RANGEDEL)}
 ----
 Level 1
   file 0: [a#1,RANGEDEL-g#inf,RANGEDEL]
@@ -240,7 +240,7 @@ unordered or unfragmented range delete tombstones f-g:{(#3,RANGEDEL)}, a-b:{(#4,
 define write-unfragmented disable-key-order-checks
 L
 a.RANGEDEL.1 d.RANGEDEL.inf
-a.RANGEDEL.1:d b.RANGEDEL.2:c
+EncodeSpan:a-d:{(#1,RANGEDEL)} EncodeSpan:b-c:{(#2,RANGEDEL)}
 ----
 Level 1
   file 0: [a#1,RANGEDEL-d#inf,RANGEDEL]
@@ -364,7 +364,7 @@ merge processing error on key a#9,SINGLEDEL in L1: fileNum=000033: finish failed
 define
 L
 a@9.SET.9 f.RANGEDEL.inf
-a@9.SET.9:a9 a@6.SET.6:a6 a.RANGEDEL.5:f
+a@9.SET.9:a9 a@6.SET.6:a6 EncodeSpan:a-f:{(#5,RANGEDEL)}
 f@6.SET.6 f@6.SET.6
 f@6.SET.6:f6
 ----
diff --git a/testdata/metrics b/testdata/metrics
index cfe9215e..9c74ca85 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -75,7 +75,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 1
-Block cache: 3 entries (484B)  hit rate: 0.0%
+Block cache: 2 entries (44B)  hit rate: 0.0%
 Table cache: 1 entries (784B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -132,7 +132,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 595B
 Compression types: snappy: 1
-Block cache: 5 entries (946B)  hit rate: 33.3%
+Block cache: 2 entries (44B)  hit rate: 33.3%
 Table cache: 2 entries (1.5KB)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -176,7 +176,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 595B
 Compression types: snappy: 1
-Block cache: 5 entries (946B)  hit rate: 33.3%
+Block cache: 2 entries (44B)  hit rate: 33.3%
 Table cache: 2 entries (1.5KB)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -217,7 +217,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 595B
 Compression types: snappy: 1
-Block cache: 3 entries (484B)  hit rate: 33.3%
+Block cache: 2 entries (44B)  hit rate: 33.3%
 Table cache: 1 entries (784B)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -495,7 +495,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 4.3KB
 Compression types: snappy: 7
-Block cache: 12 entries (1.9KB)  hit rate: 9.1%
+Block cache: 8 entries (204B)  hit rate: 9.1%
 Table cache: 1 entries (784B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -559,7 +559,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 6.1KB
 Compression types: snappy: 10
-Block cache: 12 entries (1.9KB)  hit rate: 9.1%
+Block cache: 8 entries (204B)  hit rate: 9.1%
 Table cache: 1 entries (784B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -834,7 +834,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 1
-Block cache: 1 entries (440B)  hit rate: 0.0%
+Block cache: 0 entries (0B)  hit rate: 0.0%
 Table cache: 1 entries (784B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -882,7 +882,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 2
-Block cache: 6 entries (996B)  hit rate: 0.0%
+Block cache: 4 entries (116B)  hit rate: 0.0%
 Table cache: 1 entries (784B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
@@ -931,7 +931,7 @@ Backing tables: 0 (0B)
 Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 3
-Block cache: 6 entries (996B)  hit rate: 0.0%
+Block cache: 4 entries (116B)  hit rate: 0.0%
 Table cache: 1 entries (784B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0