Repository: cockroachdb/pebble
Pull request #4461 (2025-03-31 22:25:07)

db: add support for reading blob values
Initialize iterators and compactions with a blob.ValueFetcher that's propagated
through to sstable iterators. When an sstable iterator encounters a blob value
handle, it constructs an internal value with the blob.ValueFetcher configured
as the ValueFetcher.

Close https://github.com/cockroachdb/pebble/issues/4394.
Informs https://github.com/cockroachdb/pebble/issues/112.

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


Patch:
diff --git a/compaction.go b/compaction.go
index b3007be4..85c996f2 100644
--- a/compaction.go
+++ b/compaction.go
@@ -29,6 +29,7 @@ import (
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 )
@@ -216,6 +217,9 @@ type compaction struct {
 	// blob files. This consumes more write bandwidth because all values are
 	// rewritten. However it restores locality.
 	getValueSeparation func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation
+	// valueFetcher is used to fetch values from blob files. It's propagated
+	// down the iterator tree through the internal iterator options.
+	valueFetcher blob.ValueFetcher
 
 	// startLevel is the level that is being compacted. Inputs from startLevel
 	// and outputLevel will be merged to produce a set of outputLevel files.
@@ -3085,16 +3089,19 @@ func (d *DB) compactAndWrite(
 	// translate to 3 MiB per compaction.
 	c.bufferPool.Init(12)
 	defer c.bufferPool.Release()
+	env := block.ReadEnv{
+		BufferPool: &c.bufferPool,
+		Stats:      &c.stats,
+		IterStats: d.fileCache.SSTStatsCollector().Accumulator(
+			uint64(uintptr(unsafe.Pointer(c))),
+			categoryCompaction,
+		),
+	}
+	c.valueFetcher.Init(d.fileCache, env)
 	iiopts := internalIterOpts{
-		compaction: true,
-		readEnv: block.ReadEnv{
-			BufferPool: &c.bufferPool,
-			Stats:      &c.stats,
-			IterStats: d.fileCache.SSTStatsCollector().Accumulator(
-				uint64(uintptr(unsafe.Pointer(c))),
-				categoryCompaction,
-			),
-		},
+		compaction:       true,
+		readEnv:          env,
+		blobValueFetcher: &c.valueFetcher,
 	}
 
 	pointIter, rangeDelIter, rangeKeyIter, err := c.newInputIters(d.newIters, d.tableNewRangeKeyIter, iiopts)
diff --git a/db.go b/db.go
index ddc7b2ee..c23590e3 100644
--- a/db.go
+++ b/db.go
@@ -1349,6 +1349,9 @@ type internalIterOpts struct {
 	compaction         bool
 	readEnv            block.ReadEnv
 	boundLimitedFilter sstable.BoundLimitedBlockPropertyFilter
+	// blobValueFetcher is the base.ValueFetcher to use when constructing
+	// internal values to represent values stored externally in blob files.
+	blobValueFetcher base.ValueFetcher
 }
 
 func finishInitializingInternalIter(
@@ -1401,18 +1404,21 @@ func (i *Iterator) constructPointIter(
 		// Already have one.
 		return
 	}
+	readEnv := block.ReadEnv{
+		Stats: &i.stats.InternalStats,
+		// If the file cache has a sstable stats collector, ask it for an
+		// accumulator for this iterator's configured category and QoS. All SSTable
+		// iterators created by this Iterator will accumulate their stats to it as
+		// they Close during iteration.
+		IterStats: i.fc.SSTStatsCollector().Accumulator(
+			uint64(uintptr(unsafe.Pointer(i))),
+			i.opts.Category,
+		),
+	}
+	i.blobValueFetcher.Init(i.fc, readEnv)
 	internalOpts := internalIterOpts{
-		readEnv: block.ReadEnv{
-			Stats: &i.stats.InternalStats,
-			// If the file cache has a sstable stats collector, ask it for an
-			// accumulator for this iterator's configured category and QoS. All SSTable
-			// iterators created by this Iterator will accumulate their stats to it as
-			// they Close during iteration.
-			IterStats: i.fc.SSTStatsCollector().Accumulator(
-				uint64(uintptr(unsafe.Pointer(i))),
-				i.opts.Category,
-			),
-		},
+		readEnv:          readEnv,
+		blobValueFetcher: &i.blobValueFetcher,
 	}
 	if i.opts.RangeKeyMasking.Filter != nil {
 		internalOpts.boundLimitedFilter = &i.rangeKeyMasking
diff --git a/file_cache.go b/file_cache.go
index 4c0ec440..f6761f81 100644
--- a/file_cache.go
+++ b/file_cache.go
@@ -677,7 +677,11 @@ func (h *fileCacheHandle) newPointIter(
 		internalOpts.readEnv.IterStats = handle.SSTStatsCollector().Accumulator(uint64(uintptr(unsafe.Pointer(r))), opts.Category)
 	}
 	if internalOpts.compaction {
-		iter, err = cr.NewCompactionIter(transforms, internalOpts.readEnv, &v.readerProvider)
+		iter, err = cr.NewCompactionIter(transforms, internalOpts.readEnv,
+			&v.readerProvider, sstable.TableBlobContext{
+				ValueFetcher: internalOpts.blobValueFetcher,
+				References:   file.BlobReferences,
+			})
 	} else {
 		iter, err = cr.NewPointIter(ctx, sstable.IterOptions{
 			Lower:                opts.GetLowerBound(),
@@ -687,6 +691,10 @@ func (h *fileCacheHandle) newPointIter(
 			Filterer:             filterer,
 			Env:                  internalOpts.readEnv,
 			ReaderProvider:       &v.readerProvider,
+			BlobContext: sstable.TableBlobContext{
+				ValueFetcher: internalOpts.blobValueFetcher,
+				References:   file.BlobReferences,
+			},
 		})
 	}
 	if err != nil {
diff --git a/ingest.go b/ingest.go
index 96c8690c..e29d3b9a 100644
--- a/ingest.go
+++ b/ingest.go
@@ -360,7 +360,7 @@ func ingestLoad1(
 	maybeSetStatsFromProperties(meta.PhysicalMeta(), &r.Properties)
 
 	{
-		iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+		iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 		if err != nil {
 			return nil, keyspan.Span{}, err
 		}
diff --git a/internal/base/iterator.go b/internal/base/iterator.go
index 3c204144..1da0a27d 100644
--- a/internal/base/iterator.go
+++ b/internal/base/iterator.go
@@ -416,6 +416,9 @@ type InternalIteratorStats struct {
 		// ValueBytesFetched is the total byte length of the values (in value
 		// blocks) that were retrieved.
 		ValueBytesFetched uint64
+
+		// TODO(jackson): Add stats for distinguishing between value-block
+		// values and blob values.
 	}
 }
 
diff --git a/internal/base/lazy_value.go b/internal/base/lazy_value.go
index e3cedca7..dc71e77b 100644
--- a/internal/base/lazy_value.go
+++ b/internal/base/lazy_value.go
@@ -4,7 +4,11 @@
 
 package base
 
-import "context"
+import (
+	"context"
+
+	"github.com/cockroachdb/errors"
+)
 
 // A value can have user-defined attributes that are a function of the value
 // byte slice. For now, we only support "short attributes", which can be
@@ -278,3 +282,26 @@ func (lv *LazyValue) Clone(buf []byte, fetcher *LazyFetcher) (LazyValue, []byte)
 	lvCopy.ValueOrHandle = buf[bufLen : bufLen+vLen]
 	return lvCopy, buf
 }
+
+// NoBlobFetches is a ValueFetcher that returns an error. It's intended to be
+// used in situations where sstables should not encode a blob value, or the
+// caller should not fetch the handle's value.
+var NoBlobFetches = &errValueFetcher{
+	Err: errors.AssertionFailedf("unexpected blob value"),
+}
+
+// errValueFetcher is a ValueFetcher that returns an error.
+type errValueFetcher struct {
+	Err error
+}
+
+// Assert that *errValueFetcher implements base.ValueFetcher.
+var _ ValueFetcher = (*errValueFetcher)(nil)
+
+// Fetch implements base.ValueFetcher.
+func (e *errValueFetcher) Fetch(
+	_ context.Context, _ []byte, blobFileNum DiskFileNum, valLen uint32, _ []byte,
+) (val []byte, callerOwned bool, err error) {
+	err = errors.Wrapf(e.Err, "fetching %d-byte value from %s", valLen, blobFileNum)
+	return nil, false, err
+}
diff --git a/internal/manifest/blob_metadata.go b/internal/manifest/blob_metadata.go
index 239036ed..481485af 100644
--- a/internal/manifest/blob_metadata.go
+++ b/internal/manifest/blob_metadata.go
@@ -12,6 +12,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/strparse"
+	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/redact"
 )
@@ -224,6 +225,9 @@ type BlobReferenceDepth int
 // persisted to the manifest.
 type BlobReferences []BlobReference
 
+// Assert that BlobReferences implements sstable.BlobReferences.
+var _ sstable.BlobReferences = BlobReferences{}
+
 // FileNumByID returns the FileNum for the identified BlobReference.
 func (br BlobReferences) FileNumByID(i blob.ReferenceID) base.DiskFileNum {
 	return br[i].FileNum
diff --git a/iterator.go b/iterator.go
index eaa2e227..bc83f7b0 100644
--- a/iterator.go
+++ b/iterator.go
@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/rangekeystack"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
+	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/redact"
 )
 
@@ -224,6 +225,9 @@ type Iterator struct {
 	// For use in LazyValue.Value.
 	lazyValueBuf []byte
 	valueCloser  io.Closer
+	// blobValueFetcher is the ValueFetcher to use when retrieving values stored
+	// externally in blob files.
+	blobValueFetcher blob.ValueFetcher
 	// boundsBuf holds two buffers used to store the lower and upper bounds.
 	// Whenever the Iterator's bounds change, the new bounds are copied into
 	// boundsBuf[boundsBufIdx]. The two bounds share a slice to reduce
@@ -2357,6 +2361,7 @@ func (i *Iterator) Close() error {
 		if i.rangeKey != nil && i.rangeKey.rangeKeyIter != nil {
 			i.rangeKey.rangeKeyIter.Close()
 		}
+		i.err = firstError(i.err, i.blobValueFetcher.Close())
 	}
 	err := i.err
 
diff --git a/metamorphic/build.go b/metamorphic/build.go
index 6b4ec8a9..3a207100 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -274,7 +274,7 @@ func openExternalObj(
 		start = syntheticPrefix.Invert(start)
 		end = syntheticPrefix.Invert(end)
 	}
-	pointIter, err = reader.NewIter(sstable.NoTransforms, start, end)
+	pointIter, err = reader.NewIter(sstable.NoTransforms, start, end, sstable.AssertNoBlobHandles)
 	panicIfErr(err)
 
 	rangeDelIter, err = reader.NewRawRangeDelIter(context.Background(), sstable.NoFragmentTransforms, block.NoReadEnv)
diff --git a/replay/replay.go b/replay/replay.go
index e7e174ba..415e4869 100644
--- a/replay/replay.go
+++ b/replay/replay.go
@@ -962,7 +962,7 @@ func findManifestStart(
 }
 
 // loadFlushedSSTableKeys copies keys from the sstables specified by `fileNums`
-// in the directory specified by `path` into the provided the batch. Keys are
+// in the directory specified by `path` into the provided batch. Keys are
 // applied to the batch in the order dictated by their sequence numbers within
 // the sstables, ensuring the relative relationship between sequence numbers is
 // maintained.
@@ -999,7 +999,13 @@ func loadFlushedSSTableKeys(
 			defer r.Close()
 
 			// Load all the point keys.
-			iter, err := r.NewIter(sstable.NoTransforms, nil, nil)
+			iter, err := r.NewIter(sstable.NoTransforms, nil, nil,
+				// TODO(jackson): We should update this to use a
+				// blob.ValueFetcher configured with a custom reader provider
+				// that opens the appropriate blob files. We'll also need to
+				// update the workload capture-side to collect blob files.
+				// See #4459.
+				sstable.AssertNoBlobHandles)
 			if err != nil {
 				return err
 			}
diff --git a/replay/workload_capture.go b/replay/workload_capture.go
index fded3d36..18848150 100644
--- a/replay/workload_capture.go
+++ b/replay/workload_capture.go
@@ -179,6 +179,7 @@ func (w *WorkloadCollector) onFlushEnd(info pebble.FlushInfo) {
 	}
 	w.mu.Lock()
 	defer w.mu.Unlock()
+	// TODO(jackson): Copy blob files.
 	for _, table := range info.Output {
 		w.enqueueCopyLocked(base.PhysicalTableDiskFileNum(table.FileNum))
 	}
diff --git a/sstable/blob/blob.go b/sstable/blob/blob.go
index 69f8b9eb..49b04275 100644
--- a/sstable/blob/blob.go
+++ b/sstable/blob/blob.go
@@ -165,6 +165,19 @@ func (w *FileWriter) EstimatedSize() uint64 {
 	return sz
 }
 
+// FlushForTesting flushes the current block to the write queue. Writers should
+// generally not call FlushForTesting, and instead let the heuristics configured
+// through FileWriterOptions handle flushing.
+//
+// It's exposed so that tests can force flushes to construct blob files with
+// arbitrary structures.
+func (w *FileWriter) FlushForTesting() {
+	if w.b.Size() == 0 {
+		return
+	}
+	w.flush()
+}
+
 func (w *FileWriter) flush() {
 	pb, bh := w.b.CompressAndChecksum()
 	compressedLen := uint64(pb.LengthWithoutTrailer())
diff --git a/sstable/blob/fetcher.go b/sstable/blob/fetcher.go
index 78de7e80..4a53eef2 100644
--- a/sstable/blob/fetcher.go
+++ b/sstable/blob/fetcher.go
@@ -96,6 +96,7 @@ func (r *ValueFetcher) retrieve(ctx context.Context, vh Handle) (val []byte, err
 	// vh.FileNum.
 	var cr *cachedReader
 	var oldestFetchIndex int
+	// TODO(jackson): Reconsider this O(len(readers)) scan.
 	for i := range r.readers {
 		if r.readers[i].fileNum == vh.FileNum && r.readers[i].r != nil {
 			cr = &r.readers[i]
@@ -121,6 +122,10 @@ func (r *ValueFetcher) retrieve(ctx context.Context, vh Handle) (val []byte, err
 		cr.rh = cr.r.InitReadHandle(&cr.preallocRH)
 	}
 
+	if r.env.Stats != nil {
+		r.env.Stats.SeparatedPointValue.ValueBytesFetched += uint64(vh.ValueLen)
+	}
+
 	r.fetchCount++
 	cr.lastFetchCount = r.fetchCount
 	val, err = cr.GetUnsafeValue(ctx, vh, r.env)
@@ -190,8 +195,8 @@ func (cr *cachedReader) GetUnsafeValue(
 	}
 	data := cr.currentBlockBuf.BlockData()
 	if len(data) < int(vh.OffsetInBlock+vh.ValueLen) {
-		return nil, base.CorruptionErrorf("blob file %s: block %d: expected block length %d, got %d",
-			vh.FileNum, vh.BlockNum, vh.OffsetInBlock+vh.ValueLen, len(data))
+		return nil, base.CorruptionErrorf("blob file %s: block %d: value offset %d plus len %d exceeds block length %d",
+			vh.FileNum, vh.BlockNum, vh.OffsetInBlock, vh.ValueLen, len(data))
 	}
 	return data[vh.OffsetInBlock : vh.OffsetInBlock+vh.ValueLen], nil
 }
diff --git a/sstable/blob/handle.go b/sstable/blob/handle.go
index 268150eb..c475c3fb 100644
--- a/sstable/blob/handle.go
+++ b/sstable/blob/handle.go
@@ -106,9 +106,9 @@ func (h InlineHandle) Encode(b []byte) int {
 	return n
 }
 
-// DecodeInlineHandlePrefix decodes the blob reference index and value length
+// DecodeInlineHandlePreface decodes the blob reference index and value length
 // from the beginning of a variable-width encoded InlineHandle.
-func DecodeInlineHandlePrefix(src []byte) (InlineHandlePreface, []byte) {
+func DecodeInlineHandlePreface(src []byte) (InlineHandlePreface, []byte) {
 	ptr := unsafe.Pointer(&src[0])
 	var refIdx uint32
 	if a := *((*uint8)(ptr)); a < 128 {
diff --git a/sstable/block/block.go b/sstable/block/block.go
index b61b2f8c..62118dda 100644
--- a/sstable/block/block.go
+++ b/sstable/block/block.go
@@ -569,7 +569,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.opts.CacheOpts.FileNum)
+		err = errors.Wrapf(err, "pebble: file %s", r.opts.CacheOpts.FileNum)
 		return Value{}, err
 	}
 	typ := CompressionIndicator(compressed.BlockData()[bh.Length])
@@ -646,7 +646,7 @@ func ReadRaw(
 ) ([]byte, error) {
 	size := f.Size()
 	if size < int64(len(buf)) {
-		return nil, base.CorruptionErrorf("pebble/table: invalid table %s (file size is too small)", errors.Safe(fileNum))
+		return nil, base.CorruptionErrorf("pebble: invalid file %s (file size is too small)", errors.Safe(fileNum))
 	}
 
 	readStopwatch := makeStopwatch()
@@ -664,7 +664,7 @@ func ReadRaw(
 			len(buf), readDuration.String())
 	}
 	if err != nil {
-		return nil, errors.Wrap(err, "pebble/table: invalid table (could not read footer)")
+		return nil, errors.Wrap(err, "pebble: invalid file (could not read footer)")
 	}
 	return buf, nil
 }
diff --git a/sstable/block/compressor.go b/sstable/block/compressor.go
index f785c883..5c03d2c9 100644
--- a/sstable/block/compressor.go
+++ b/sstable/block/compressor.go
@@ -78,7 +78,7 @@ func (snappyDecompressor) DecompressInto(buf, compressed []byte) error {
 		return err
 	}
 	if len(result) != len(buf) || (len(result) > 0 && &result[0] != &buf[0]) {
-		return base.CorruptionErrorf("pebble/table: decompressed into unexpected buffer: %p != %p",
+		return base.CorruptionErrorf("pebble: decompressed into unexpected buffer: %p != %p",
 			errors.Safe(result), errors.Safe(buf))
 	}
 	return nil
@@ -94,7 +94,7 @@ func (zstdDecompressor) DecompressedLen(b []byte) (decompressedLen int, err erro
 	// if we implement these algorithms in the future.
 	decodedLenU64, varIntLen := binary.Uvarint(b)
 	if varIntLen <= 0 {
-		return 0, base.CorruptionErrorf("pebble/table: compression block has invalid length")
+		return 0, base.CorruptionErrorf("pebble: compression block has invalid length")
 	}
 	return int(decodedLenU64), nil
 }
diff --git a/sstable/reader.go b/sstable/reader.go
index ae63099c..e1b7b354 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -85,6 +85,7 @@ type IterOptions struct {
 	FilterBlockSizeLimit FilterBlockSizeLimit
 	Env                  block.ReadEnv
 	ReaderProvider       valblk.ReaderProvider
+	BlobContext          TableBlobContext
 }
 
 // NewPointIter returns an iterator for the point keys in the table.
@@ -151,7 +152,9 @@ func (r *Reader) newPointIter(
 //
 // NewIter must only be used when the Reader is guaranteed to outlive any
 // LazyValues returned from the iter.
-func (r *Reader) NewIter(transforms IterTransforms, lower, upper []byte) (Iterator, error) {
+func (r *Reader) NewIter(
+	transforms IterTransforms, lower, upper []byte, blobContext TableBlobContext,
+) (Iterator, error) {
 	// TODO(radu): we should probably not use bloom filters in this case, as there
 	// likely isn't a cache set up.
 	opts := IterOptions{
@@ -162,6 +165,7 @@ func (r *Reader) NewIter(transforms IterTransforms, lower, upper []byte) (Iterat
 		FilterBlockSizeLimit: AlwaysUseFilterBlock,
 		Env:                  block.NoReadEnv,
 		ReaderProvider:       MakeTrivialReaderProvider(r),
+		BlobContext:          blobContext,
 	}
 	return r.NewPointIter(context.TODO(), opts)
 }
@@ -170,13 +174,20 @@ func (r *Reader) NewIter(transforms IterTransforms, lower, upper []byte) (Iterat
 // 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, env block.ReadEnv, rp valblk.ReaderProvider,
+	transforms IterTransforms,
+	env block.ReadEnv,
+	rp valblk.ReaderProvider,
+	blobContext TableBlobContext,
 ) (Iterator, error) {
-	return r.newCompactionIter(transforms, env, rp, nil)
+	return r.newCompactionIter(transforms, env, rp, nil, blobContext)
 }
 
 func (r *Reader) newCompactionIter(
-	transforms IterTransforms, env block.ReadEnv, rp valblk.ReaderProvider, vState *virtualState,
+	transforms IterTransforms,
+	env block.ReadEnv,
+	rp valblk.ReaderProvider,
+	vState *virtualState,
+	blobContext TableBlobContext,
 ) (Iterator, error) {
 	if vState != nil && vState.isSharedIngested {
 		transforms.HideObsoletePoints = true
@@ -188,6 +199,7 @@ func (r *Reader) newCompactionIter(
 		FilterBlockSizeLimit: NeverUseFilterBlock,
 		Env:                  env,
 		ReaderProvider:       rp,
+		BlobContext:          blobContext,
 	}
 
 	if r.Properties.IndexType == twoLevelIndex {
diff --git a/sstable/reader_common.go b/sstable/reader_common.go
index dfd44773..89bc3f23 100644
--- a/sstable/reader_common.go
+++ b/sstable/reader_common.go
@@ -31,6 +31,7 @@ type CommonReader interface {
 		transforms IterTransforms,
 		env block.ReadEnv,
 		rp valblk.ReaderProvider,
+		blobContext TableBlobContext,
 	) (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 fd173a3b..ecc4a86d 100644
--- a/sstable/reader_iter_single_lvl.go
+++ b/sstable/reader_iter_single_lvl.go
@@ -64,8 +64,8 @@ type singleLevelIterator[I any, PI indexBlockIterator[I], D any, PD dataBlockIte
 	// dataBH refers to the last data block that the iterator considered
 	// loading. It may not actually have loaded the block, due to an error or
 	// because it was considered irrelevant.
-	dataBH   block.Handle
-	vbReader valblk.Reader
+	dataBH                   block.Handle
+	internalValueConstructor defaultInternalValueConstructor
 	// vbRH is the read handle for value blocks, which are in a different
 	// part of the sstable than data blocks.
 	vbRH         objstorage.ReadHandle
@@ -74,7 +74,6 @@ type singleLevelIterator[I any, PI indexBlockIterator[I], D any, PD dataBlockIte
 	closeHook    func()
 
 	readBlockEnv block.ReadEnv
-
 	// boundsCmp and positionedUsingLatestBounds are for optimizing iteration
 	// that uses multiple adjacent bounds. The seek after setting a new bound
 	// can use the fact that the iterator is either within the previous bounds
@@ -212,13 +211,11 @@ func newColumnBlockSingleLevelIterator(
 	}
 	i := singleLevelIterColumnBlockPool.Get().(*singleLevelIteratorColumnBlocks)
 	i.init(ctx, r, v, opts)
-	var getInternalValuer block.GetInternalValueForPrefixAndValueHandler
 	if r.Properties.NumValueBlocks > 0 {
-		i.vbReader = valblk.MakeReader(i, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
-		getInternalValuer = &i.vbReader
+		i.internalValueConstructor.vbReader = valblk.MakeReader(i, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
 		i.vbRH = r.blockReader.UsePreallocatedReadHandle(objstorage.NoReadBefore, &i.vbRHPrealloc)
 	}
-	i.data.InitOnce(r.keySchema, r.Comparer, getInternalValuer)
+	i.data.InitOnce(r.keySchema, r.Comparer, &i.internalValueConstructor)
 	indexH, err := r.readTopLevelIndexBlock(ctx, i.readBlockEnv, i.indexFilterRH)
 	if err == nil {
 		err = i.index.InitHandle(r.Comparer, indexH, opts.Transforms)
@@ -250,8 +247,10 @@ func newRowBlockSingleLevelIterator(
 	i.init(ctx, r, v, opts)
 	if r.tableFormat >= TableFormatPebblev3 {
 		if r.Properties.NumValueBlocks > 0 {
-			i.vbReader = valblk.MakeReader(i, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
-			(&i.data).SetGetLazyValuer(&i.vbReader)
+			i.internalValueConstructor.vbReader = valblk.MakeReader(i, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
+			// We can set the GetLazyValuer directly to the vbReader because
+			// rowblk sstables never contain blob value handles.
+			(&i.data).SetGetLazyValuer(&i.internalValueConstructor.vbReader)
 			i.vbRH = r.blockReader.UsePreallocatedReadHandle(objstorage.NoReadBefore, &i.vbRHPrealloc)
 		}
 		i.data.SetHasValuePrefix(true)
@@ -281,11 +280,13 @@ func (i *singleLevelIterator[I, PI, D, PD]) init(
 	i.reader = r
 	i.cmp = r.Comparer.Compare
 	i.transforms = opts.Transforms
+	i.readBlockEnv = opts.Env
+	i.internalValueConstructor.blobContext = opts.BlobContext
+	i.internalValueConstructor.env = &i.readBlockEnv
 	if v != nil {
 		i.vState = v
 		i.endKeyInclusive, i.lower, i.upper = v.constrainBounds(opts.Lower, opts.Upper, false /* endInclusive */)
 	}
-	i.readBlockEnv = opts.Env
 
 	i.indexFilterRH = r.blockReader.UsePreallocatedReadHandle(
 		objstorage.ReadBeforeForIndexAndFilter, &i.indexFilterRHPrealloc)
@@ -1531,7 +1532,7 @@ func (i *singleLevelIterator[I, PI, D, PD]) closeInternal() error {
 	if i.bpfs != nil {
 		releaseBlockPropertiesFilterer(i.bpfs)
 	}
-	i.vbReader.Close()
+	i.internalValueConstructor.vbReader.Close()
 	if i.vbRH != nil {
 		err = firstError(err, i.vbRH.Close())
 		i.vbRH = nil
diff --git a/sstable/reader_iter_two_lvl.go b/sstable/reader_iter_two_lvl.go
index fa9fbf48..2e1361d3 100644
--- a/sstable/reader_iter_two_lvl.go
+++ b/sstable/reader_iter_two_lvl.go
@@ -15,7 +15,6 @@ import (
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 	"github.com/cockroachdb/pebble/objstorage"
-	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/valblk"
 )
 
@@ -167,7 +166,6 @@ func newColumnBlockTwoLevelIterator(
 	i.useFilterBlock = i.secondLevel.useFilterBlock
 	i.secondLevel.useFilterBlock = false
 
-	var getInternalValuer block.GetInternalValueForPrefixAndValueHandler
 	if r.Properties.NumValueBlocks > 0 {
 		// NB: we cannot avoid this ~248 byte allocation, since valueBlockReader
 		// can outlive the singleLevelIterator due to be being embedded in a
@@ -178,12 +176,11 @@ 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, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
-		getInternalValuer = &i.secondLevel.vbReader
+		i.secondLevel.internalValueConstructor.vbReader = valblk.MakeReader(&i.secondLevel, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
 		i.secondLevel.vbRH = r.blockReader.UsePreallocatedReadHandle(
 			objstorage.NoReadBefore, &i.secondLevel.vbRHPrealloc)
 	}
-	i.secondLevel.data.InitOnce(r.keySchema, r.Comparer, getInternalValuer)
+	i.secondLevel.data.InitOnce(r.keySchema, r.Comparer, &i.secondLevel.internalValueConstructor)
 	topLevelIndexH, err := r.readTopLevelIndexBlock(ctx, i.secondLevel.readBlockEnv, i.secondLevel.indexFilterRH)
 	if err == nil {
 		err = i.topLevelIndex.InitHandle(r.Comparer, topLevelIndexH, opts.Transforms)
@@ -216,6 +213,7 @@ func newRowBlockTwoLevelIterator(
 	// Only check the bloom filter at the top level.
 	i.useFilterBlock = i.secondLevel.useFilterBlock
 	i.secondLevel.useFilterBlock = false
+
 	if r.tableFormat >= TableFormatPebblev3 {
 		if r.Properties.NumValueBlocks > 0 {
 			// NB: we cannot avoid this ~248 byte allocation, since valueBlockReader
@@ -227,8 +225,10 @@ 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, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
-			i.secondLevel.data.SetGetLazyValuer(&i.secondLevel.vbReader)
+			i.secondLevel.internalValueConstructor.vbReader = valblk.MakeReader(&i.secondLevel, opts.ReaderProvider, r.valueBIH, opts.Env.Stats)
+			// We can set the GetLazyValuer directly to the vbReader because
+			// rowblk sstables never contain blob value handles.
+			i.secondLevel.data.SetGetLazyValuer(&i.secondLevel.internalValueConstructor.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 dc942cd7..ceefb4a9 100644
--- a/sstable/reader_virtual.go
+++ b/sstable/reader_virtual.go
@@ -96,10 +96,13 @@ func MakeVirtualReader(reader *Reader, p VirtualReaderParams) VirtualReader {
 
 // NewCompactionIter is the compaction iterator function for virtual readers.
 func (v *VirtualReader) NewCompactionIter(
-	transforms IterTransforms, env block.ReadEnv, rp valblk.ReaderProvider,
+	transforms IterTransforms,
+	env block.ReadEnv,
+	rp valblk.ReaderProvider,
+	blobContext TableBlobContext,
 ) (Iterator, error) {
 	return v.reader.newCompactionIter(
-		transforms, env, rp, &v.vState)
+		transforms, env, rp, &v.vState, blobContext)
 }
 
 // NewPointIter returns an iterator for the point keys in the table.
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index ce786467..b94694d2 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -308,6 +308,9 @@ func RewriteKeySuffixesViaWriter(
 	if o.Comparer == nil || o.Comparer.Split == nil {
 		return nil, errors.New("a valid splitter is required to rewrite suffixes")
 	}
+	if r.Properties.NumValuesInBlobFiles > 0 {
+		return nil, errors.New("cannot rewrite suffixes of sstable with blob values")
+	}
 
 	o.IsStrictObsolete = false
 	w := NewRawWriter(out, o)
@@ -316,7 +319,7 @@ func RewriteKeySuffixesViaWriter(
 			w.Close()
 		}
 	}()
-	i, err := r.NewIter(NoTransforms, nil, nil)
+	i, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 	if err != nil {
 		return nil, err
 	}
@@ -336,13 +339,14 @@ func RewriteKeySuffixesViaWriter(
 		scratch.UserKey = append(scratch.UserKey, to...)
 		scratch.Trailer = kv.K.Trailer
 
+		if invariants.Enabled && invariants.Sometimes(10) {
+			r.Comparer.ValidateKey.MustValidate(scratch.UserKey)
+		}
+
 		val, _, err := kv.Value(nil)
 		if err != nil {
 			return nil, err
 		}
-		if invariants.Enabled && invariants.Sometimes(10) {
-			r.Comparer.ValidateKey.MustValidate(scratch.UserKey)
-		}
 		if err := w.Add(scratch, val, false); err != nil {
 			return nil, err
 		}
diff --git a/sstable/values.go b/sstable/values.go
new file mode 100644
index 00000000..0f1b4aa0
--- /dev/null
+++ b/sstable/values.go
@@ -0,0 +1,116 @@
+// Copyright 2025 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package sstable
+
+import (
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/sstable/blob"
+	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/sstable/valblk"
+)
+
+// AssertNoBlobHandles is a TableBlobContext that configures a sstable iterator
+// to panic if it ever encounters a value that references an external blob file.
+var AssertNoBlobHandles = TableBlobContext{
+	ValueFetcher: base.NoBlobFetches,
+	// Passing a nil BlobReferences will cause any attempt to construct an
+	// InternalValue from a blob handle to panic.
+	References: nil,
+}
+
+// BlobReferences provides a mapping from an index to a file number for a
+// sstable's blob references. In practice, this is implemented by
+// manifest.BlobReferences.
+type BlobReferences interface {
+	// FileNumByID returns the FileNum for the identified BlobReference.
+	FileNumByID(i blob.ReferenceID) base.DiskFileNum
+	// IDByFileNum returns the reference ID for the given FileNum. If the file
+	// number is not found, the second return value is false.
+	IDByFileNum(fileNum base.DiskFileNum) (blob.ReferenceID, bool)
+}
+
+// TableBlobContext configures how values that reference external blob files
+// should be retrieved and handled.
+type TableBlobContext struct {
+	// ValueFetcher is used as the base.ValueFetcher for values returned that
+	// reference external blob files.
+	ValueFetcher base.ValueFetcher
+	// References provides a mapping from an index to a file number for a
+	// sstable's blob references.
+	References BlobReferences
+}
+
+// defaultInternalValueConstructor is the default implementation of the
+// block.GetInternalValueForPrefixAndValueHandler interface.
+type defaultInternalValueConstructor struct {
+	blobContext TableBlobContext
+	env         *block.ReadEnv
+	vbReader    valblk.Reader
+
+	// lazyFetcher is the LazyFetcher value embedded in any LazyValue that we
+	// return. It is used to avoid having a separate allocation for that.
+	lazyFetcher base.LazyFetcher
+}
+
+// Assert that defaultInternalValueConstructor implements the
+// block.GetInternalValueForPrefixAndValueHandler interface.
+var _ block.GetInternalValueForPrefixAndValueHandler = (*defaultInternalValueConstructor)(nil)
+
+// GetInternalValueForPrefixAndValueHandle returns a InternalValue for the
+// given value prefix and value.
+//
+// The result is only valid until the next call to
+// GetInternalValueForPrefixAndValueHandle. Use InternalValue.Clone if the
+// lifetime of the InternalValue needs to be extended. For more details, see
+// the "memory management" comment where LazyValue is declared.
+func (i *defaultInternalValueConstructor) GetInternalValueForPrefixAndValueHandle(
+	handle []byte,
+) base.InternalValue {
+	vp := block.ValuePrefix(handle[0])
+	if vp.IsValueBlockHandle() {
+		return i.vbReader.GetInternalValueForPrefixAndValueHandle(handle)
+	} else if !vp.IsBlobValueHandle() {
+		panic(errors.AssertionFailedf("block: %x is neither a valblk or blob handle prefix", vp))
+	}
+
+	// We can't convert a blob handle into an InternalValue without
+	// BlobReferences providing the mapping of a reference index to a blob file
+	// number.
+	if i.blobContext.References == nil {
+		panic(errors.AssertionFailedf("blob references not configured"))
+	}
+
+	// The first byte of [handle] is the valuePrefix byte.
+	//
+	// After that, is the inline-handle preface encoding a) the length of the
+	// value and b) the blob reference index. We need to map the blob reference
+	// index into a file number,
+	//
+	// The remainder of the handle (the suffix) encodes the value's location
+	// within the blob file. We defer parsing of it until the user retrieves the
+	// value. We propagate it as LazyValue.ValueOrHandle.
+	preface, remainder := blob.DecodeInlineHandlePreface(handle[1:])
+
+	if i.env.Stats != nil {
+		// TODO(jackson): Add stats to differentiate between blob values and
+		// value-block values.
+		i.env.Stats.SeparatedPointValue.Count++
+		i.env.Stats.SeparatedPointValue.ValueBytes += uint64(preface.ValueLen)
+	}
+
+	i.lazyFetcher = base.LazyFetcher{
+		Fetcher: i.blobContext.ValueFetcher,
+		Attribute: base.AttributeAndLen{
+			ValueLen:       preface.ValueLen,
+			ShortAttribute: vp.ShortAttribute(),
+		},
+		BlobFileNum: i.blobContext.References.FileNumByID(preface.ReferenceID),
+	}
+	return base.MakeLazyValue(base.LazyValue{
+		ValueOrHandle: remainder,
+		Fetcher:       &i.lazyFetcher,
+	})
+}
diff --git a/tool/find.go b/tool/find.go
index a3772a2e..72a77d9f 100644
--- a/tool/find.go
+++ b/tool/find.go
@@ -468,7 +468,10 @@ func (f *findT) searchTables(stdout io.Writer, searchKey []byte, refs []findRef)
 				fragTransforms = m.FragmentIterTransforms()
 			}
 
-			iter, err := r.NewIter(transforms, nil, nil)
+			// TODO(jackson): Adjust to support two modes: one that surfaces the
+			// raw blob value handles, and one that fetches the blob values from
+			// blob files uncovered by scanning the directory entries. See #4448.
+			iter, err := r.NewIter(transforms, nil, nil, sstable.AssertNoBlobHandles)
 			if err != nil {
 				return err
 			}
diff --git a/tool/sstable.go b/tool/sstable.go
index 0dba6bc6..97cafc63 100644
--- a/tool/sstable.go
+++ b/tool/sstable.go
@@ -164,14 +164,17 @@ func (s *sstableT) runCheck(cmd *cobra.Command, args []string) {
 		s.fmtKey.setForComparer(r.Properties.ComparerName, s.comparers)
 		s.fmtValue.setForComparer(r.Properties.ComparerName, s.comparers)
 
-		iter, err := r.NewIter(sstable.NoTransforms, nil, nil)
+		// TODO(jackson): Adjust to support two modes: one that surfaces the raw
+		// blob value handles, and one that fetches the blob values from blob
+		// files uncovered by scanning the directory entries. See #4448.
+		iter, err := r.NewIter(sstable.NoTransforms, nil, nil, sstable.AssertNoBlobHandles)
 		if err != nil {
 			fmt.Fprintf(stderr, "%s\n", err)
 			return
 		}
 
 		// Verify that SeekPrefixGE can find every key in the table.
-		prefixIter, err := r.NewIter(sstable.NoTransforms, nil, nil)
+		prefixIter, err := r.NewIter(sstable.NoTransforms, nil, nil, sstable.AssertNoBlobHandles)
 		if err != nil {
 			fmt.Fprintf(stderr, "%s\n", err)
 			return
@@ -332,7 +335,10 @@ func (s *sstableT) runScan(cmd *cobra.Command, args []string) {
 			prefix = fmt.Sprintf("%s: ", path)
 		}
 
-		iter, err := r.NewIter(sstable.NoTransforms, nil, s.end)
+		// TODO(jackson): Adjust to support two modes: one that surfaces the raw
+		// blob value handles, and one that fetches the blob values from blob
+		// files uncovered by scanning the directory entries. See #4448.
+		iter, err := r.NewIter(sstable.NoTransforms, nil, s.end, sstable.AssertNoBlobHandles)
 		if err != nil {
 			fmt.Fprintf(stderr, "%s%s\n", prefix, err)
 			return
diff --git a/value_separation.go b/value_separation.go
index c73a6854..da4e0242 100644
--- a/value_separation.go
+++ b/value_separation.go
@@ -251,7 +251,7 @@ func (vs *preserveBlobReferences) Add(
 	lv := kv.V.LazyValue()
 	fn := lv.Fetcher.BlobFileNum
 
-	refIdx, found := vs.currReferences.IDByFileNum(fn)
+	refID, found := vs.currReferences.IDByFileNum(fn)
 	if !found {
 		// This is the first time we're seeing this blob file for this sstable.
 		// Find the blob file metadata for this file among the input metadatas.
@@ -259,21 +259,21 @@ func (vs *preserveBlobReferences) Add(
 		if !found {
 			return errors.AssertionFailedf("pebble: blob file %s not found among input sstables", fn)
 		}
-		refIdx = blob.ReferenceID(len(vs.currReferences))
+		refID = blob.ReferenceID(len(vs.currReferences))
 		vs.currReferences = append(vs.currReferences, manifest.BlobReference{
 			FileNum:  fn,
 			Metadata: vs.inputBlobMetadatas[idx],
 		})
 	}
 
-	if invariants.Enabled && vs.currReferences[refIdx].Metadata.FileNum != fn {
+	if invariants.Enabled && vs.currReferences[refID].Metadata.FileNum != fn {
 		panic("wrong reference index")
 	}
 
 	handleSuffix := blob.DecodeHandleSuffix(lv.ValueOrHandle)
 	inlineHandle := blob.InlineHandle{
 		InlineHandlePreface: blob.InlineHandlePreface{
-			ReferenceID: refIdx,
+			ReferenceID: refID,
 			ValueLen:    lv.Fetcher.Attribute.ValueLen,
 		},
 		HandleSuffix: handleSuffix,
@@ -282,7 +282,7 @@ func (vs *preserveBlobReferences) Add(
 	if err != nil {
 		return err
 	}
-	vs.currReferences[refIdx].ValueSize += uint64(lv.Fetcher.Attribute.ValueLen)
+	vs.currReferences[refID].ValueSize += uint64(lv.Fetcher.Attribute.ValueLen)
 	vs.totalValueSize += uint64(lv.Fetcher.Attribute.ValueLen)
 	return nil
 }


Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index 45732891..7a136897 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -610,7 +610,7 @@ func TestCompaction(t *testing.T) {
 					return "", "", errors.WithStack(err)
 				}
 				defer r.Close()
-				iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+				iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 				if err != nil {
 					return "", "", errors.WithStack(err)
 				}
diff --git a/internal/blobtest/handles.go b/internal/blobtest/handles.go
index 6a40ce19..080bf053 100644
--- a/internal/blobtest/handles.go
+++ b/internal/blobtest/handles.go
@@ -26,7 +26,8 @@ import (
 // human-readable string describing a blob handle, synthesizing unspecified
 // fields, and tracking the blob handle to support future fetches.
 type Values struct {
-	mostRecentHandle blob.Handle
+	mostRecentFileNum base.DiskFileNum
+	mostRecentHandles map[base.DiskFileNum]blob.Handle
 	// trackedHandles maps from a blob handle to its value. The value may be nil
 	// if the value was not specified (in which case Fetch will
 	// deterministically derive a random value from the handle itself.)
@@ -108,6 +109,7 @@ func (bv *Values) IsBlobHandle(input string) bool {
 func (bv *Values) Parse(input string) (h blob.Handle, err error) {
 	if bv.trackedHandles == nil {
 		bv.trackedHandles = make(map[blob.Handle]string)
+		bv.mostRecentHandles = make(map[base.DiskFileNum]blob.Handle)
 	}
 
 	defer func() {
@@ -156,13 +158,13 @@ func (bv *Values) Parse(input string) (h blob.Handle, err error) {
 	}
 
 	if !fileNumSet {
-		h.FileNum = max(bv.mostRecentHandle.FileNum, 1)
+		h.FileNum = bv.mostRecentFileNum
 	}
 	if !blockNumSet {
-		h.BlockNum = bv.mostRecentHandle.BlockNum
+		h.BlockNum = bv.mostRecentHandles[h.FileNum].BlockNum
 	}
 	if !offsetSet {
-		h.OffsetInBlock = bv.mostRecentHandle.OffsetInBlock + bv.mostRecentHandle.ValueLen
+		h.OffsetInBlock = bv.mostRecentHandles[h.FileNum].OffsetInBlock + bv.mostRecentHandles[h.FileNum].ValueLen
 	}
 	if !valueLenSet {
 		if len(value) > 0 {
@@ -171,7 +173,8 @@ func (bv *Values) Parse(input string) (h blob.Handle, err error) {
 			h.ValueLen = 12
 		}
 	}
-	bv.mostRecentHandle = h
+	bv.mostRecentFileNum = h.FileNum
+	bv.mostRecentHandles[h.FileNum] = h
 	bv.trackedHandles[h] = value
 	return h, nil
 }
@@ -228,7 +231,10 @@ func WriteFiles(
 			return err
 		}
 		writer := blob.NewFileWriter(fileNum, writable, writerOpts)
-		for _, handle := range handles {
+		for i, handle := range handles {
+			if i > 0 && handles[i-1].BlockNum != handle.BlockNum {
+				writer.FlushForTesting()
+			}
 			if value, ok := bv.trackedHandles[handle]; ok {
 				writer.AddValue([]byte(value))
 			} else {
diff --git a/internal/compact/spans_test.go b/internal/compact/spans_test.go
index 4a2a9f3c..d628b4a7 100644
--- a/internal/compact/spans_test.go
+++ b/internal/compact/spans_test.go
@@ -135,7 +135,7 @@ func TestSplitAndEncodeSpan(t *testing.T) {
 			_, rangeDels, rangeKeys, err := sstable.ReadAll(obj, sstable.ReaderOptions{
 				Comparer:   wo.Comparer,
 				KeySchemas: sstable.MakeKeySchemas(wo.KeySchema),
-			})
+			}, base.NoBlobFetches)
 			require.NoError(t, err)
 			require.LessOrEqual(t, len(rangeDels)+len(rangeKeys), 1)
 			s := "."
diff --git a/level_checker_test.go b/level_checker_test.go
index 2f322a5c..6c817815 100644
--- a/level_checker_test.go
+++ b/level_checker_test.go
@@ -109,7 +109,7 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 			if err != nil {
 				return iterSet{}, err
 			}
-			iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+			iter, err := r.NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 			if err != nil {
 				return iterSet{}, err
 			}
diff --git a/level_iter_test.go b/level_iter_test.go
index 1f7436b4..4f9ba5ca 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -190,6 +190,10 @@ func (lt *levelIterTest) newIters(
 			FilterBlockSizeLimit: sstable.AlwaysUseFilterBlock,
 			Env:                  iio.readEnv,
 			ReaderProvider:       sstable.MakeTrivialReaderProvider(lt.readers[file.FileNum]),
+			BlobContext: sstable.TableBlobContext{
+				ValueFetcher: iio.blobValueFetcher,
+				References:   file.BlobReferences,
+			},
 		})
 		if err != nil {
 			return iterSet{}, errors.CombineErrors(err, set.CloseAll())
@@ -575,7 +579,7 @@ func buildLevelIterTables(
 
 	meta := make([]*tableMetadata, len(readers))
 	for i := range readers {
-		iter, err := readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+		iter, err := readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 		require.NoError(b, err)
 		smallest := iter.First()
 		meta[i] = &tableMetadata{}
@@ -602,7 +606,7 @@ func BenchmarkLevelIterSeekGE(b *testing.B) {
 							newIters := func(
 								_ context.Context, file *manifest.TableMetadata, _ *IterOptions, _ internalIterOpts, _ iterKinds,
 							) (iterSet, error) {
-								iter, err := readers[file.FileNum].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+								iter, err := readers[file.FileNum].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 								return iterSet{point: iter}, err
 							}
 							l := newLevelIter(context.Background(), IterOptions{}, DefaultComparer, newIters, metas.Iter(), manifest.Level(level), internalIterOpts{})
@@ -644,8 +648,7 @@ func BenchmarkLevelIterSeqSeekGEWithBounds(b *testing.B) {
 								_ context.Context, file *manifest.TableMetadata, opts *IterOptions, _ internalIterOpts, _ iterKinds,
 							) (iterSet, error) {
 								iter, err := readers[file.FileNum].NewIter(
-									sstable.NoTransforms,
-									opts.LowerBound, opts.UpperBound)
+									sstable.NoTransforms, opts.LowerBound, opts.UpperBound, sstable.AssertNoBlobHandles)
 								return iterSet{point: iter}, err
 							}
 							l := newLevelIter(context.Background(), IterOptions{}, DefaultComparer, newIters, metas.Iter(), manifest.Level(level), internalIterOpts{})
@@ -691,8 +694,7 @@ func BenchmarkLevelIterSeqSeekPrefixGE(b *testing.B) {
 		_ context.Context, file *manifest.TableMetadata, opts *IterOptions, _ internalIterOpts, _ iterKinds,
 	) (iterSet, error) {
 		iter, err := readers[file.FileNum].NewIter(
-			sstable.NoTransforms,
-			opts.LowerBound, opts.UpperBound)
+			sstable.NoTransforms, opts.LowerBound, opts.UpperBound, sstable.AssertNoBlobHandles)
 		return iterSet{point: iter}, err
 	}
 
@@ -746,7 +748,8 @@ func BenchmarkLevelIterNext(b *testing.B) {
 							newIters := func(
 								_ context.Context, file *manifest.TableMetadata, _ *IterOptions, _ internalIterOpts, _ iterKinds,
 							) (iterSet, error) {
-								iter, err := readers[file.FileNum].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+								iter, err := readers[file.FileNum].NewIter(sstable.NoTransforms,
+									nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 								return iterSet{point: iter}, err
 							}
 							l := newLevelIter(context.Background(), IterOptions{}, testkeys.Comparer, newIters, metas.Iter(), manifest.Level(level), internalIterOpts{})
@@ -780,7 +783,8 @@ func BenchmarkLevelIterPrev(b *testing.B) {
 							newIters := func(
 								_ context.Context, file *manifest.TableMetadata, _ *IterOptions, _ internalIterOpts, _ iterKinds,
 							) (iterSet, error) {
-								iter, err := readers[file.FileNum].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+								iter, err := readers[file.FileNum].NewIter(
+									sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 								return iterSet{point: iter}, err
 							}
 							l := newLevelIter(context.Background(), IterOptions{}, DefaultComparer, newIters, metas.Iter(), manifest.Level(level), internalIterOpts{})
diff --git a/merging_iter_test.go b/merging_iter_test.go
index 273608b1..7a1311f2 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -184,6 +184,10 @@ func TestMergingIterDataDriven(t *testing.T) {
 					FilterBlockSizeLimit: sstable.AlwaysUseFilterBlock,
 					Env:                  iio.readEnv,
 					ReaderProvider:       sstable.MakeTrivialReaderProvider(r),
+					BlobContext: sstable.TableBlobContext{
+						ValueFetcher: iio.blobValueFetcher,
+						References:   file.BlobReferences,
+					},
 				})
 				if err != nil {
 					return iterSet{}, errors.CombineErrors(err, set.CloseAll())
@@ -418,7 +422,7 @@ func BenchmarkMergingIterSeekGE(b *testing.B) {
 							iters := make([]internalIterator, len(readers))
 							for i := range readers {
 								var err error
-								iters[i], err = readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+								iters[i], err = readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 								require.NoError(b, err)
 							}
 							var stats base.InternalIteratorStats
@@ -451,7 +455,7 @@ func BenchmarkMergingIterNext(b *testing.B) {
 							iters := make([]internalIterator, len(readers))
 							for i := range readers {
 								var err error
-								iters[i], err = readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+								iters[i], err = readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 								require.NoError(b, err)
 							}
 							var stats base.InternalIteratorStats
@@ -487,7 +491,7 @@ func BenchmarkMergingIterPrev(b *testing.B) {
 							iters := make([]internalIterator, len(readers))
 							for i := range readers {
 								var err error
-								iters[i], err = readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+								iters[i], err = readers[i].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 								require.NoError(b, err)
 							}
 							var stats base.InternalIteratorStats
@@ -650,7 +654,7 @@ func buildLevelsForMergingIterSeqSeek(
 	for i := range readers {
 		meta := make([]*tableMetadata, len(readers[i]))
 		for j := range readers[i] {
-			iter, err := readers[i][j].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */)
+			iter, err := readers[i][j].NewIter(sstable.NoTransforms, nil /* lower */, nil /* upper */, sstable.AssertNoBlobHandles)
 			require.NoError(b, err)
 			smallest := iter.First()
 			meta[j] = &tableMetadata{}
@@ -676,7 +680,7 @@ func buildMergingIter(readers [][]*sstable.Reader, levelSlices []manifest.LevelS
 			_ context.Context, file *manifest.TableMetadata, opts *IterOptions, iio internalIterOpts, _ iterKinds,
 		) (iterSet, error) {
 			iter, err := readers[levelIndex][file.FileNum].NewIter(
-				sstable.NoTransforms, opts.LowerBound, opts.UpperBound)
+				sstable.NoTransforms, opts.LowerBound, opts.UpperBound, sstable.AssertNoBlobHandles)
 			if err != nil {
 				return iterSet{}, err
 			}
diff --git a/sstable/blob/blob_test.go b/sstable/blob/blob_test.go
index 989d954f..92108c0e 100644
--- a/sstable/blob/blob_test.go
+++ b/sstable/blob/blob_test.go
@@ -110,7 +110,7 @@ func TestHandleRoundtrip(t *testing.T) {
 	for _, h := range handles {
 		var buf [MaxInlineHandleLength]byte
 		n := h.Encode(buf[:])
-		preface, rem := DecodeInlineHandlePrefix(buf[:n])
+		preface, rem := DecodeInlineHandlePreface(buf[:n])
 		suffix := DecodeHandleSuffix(rem)
 		require.Equal(t, h.InlineHandlePreface, preface)
 		require.Equal(t, h.HandleSuffix, suffix)
diff --git a/sstable/block_property_test.go b/sstable/block_property_test.go
index d755730e..a8bfe2cc 100644
--- a/sstable/block_property_test.go
+++ b/sstable/block_property_test.go
@@ -994,6 +994,7 @@ func TestBlockProperties(t *testing.T) {
 				FilterBlockSizeLimit: NeverUseFilterBlock,
 				Env:                  block.ReadEnv{Stats: &stats, IterStats: nil},
 				ReaderProvider:       MakeTrivialReaderProvider(r),
+				BlobContext:          AssertNoBlobHandles,
 			})
 			if err != nil {
 				return err.Error()
@@ -1083,6 +1084,7 @@ func TestBlockProperties_BoundLimited(t *testing.T) {
 				FilterBlockSizeLimit: NeverUseFilterBlock,
 				Env:                  block.ReadEnv{Stats: &stats, IterStats: nil},
 				ReaderProvider:       MakeTrivialReaderProvider(r),
+				BlobContext:          AssertNoBlobHandles,
 			})
 			if err != nil {
 				return err.Error()
diff --git a/sstable/copier_test.go b/sstable/copier_test.go
index e80c6c89..b4df471c 100644
--- a/sstable/copier_test.go
+++ b/sstable/copier_test.go
@@ -126,7 +126,7 @@ func TestCopySpan(t *testing.T) {
 			if err != nil {
 				return err.Error()
 			}
-			iter, err := r.NewIter(block.NoTransforms, start, end)
+			iter, err := r.NewIter(block.NoTransforms, start, end, AssertNoBlobHandles)
 			if err != nil {
 				return err.Error()
 			}
diff --git a/sstable/random_test.go b/sstable/random_test.go
index 61071ead..f159c70e 100644
--- a/sstable/random_test.go
+++ b/sstable/random_test.go
@@ -111,6 +111,7 @@ func runErrorInjectionTest(t *testing.T, seed int64) {
 		FilterBlockSizeLimit: filterBlockSizeLimit,
 		Env:                  block.ReadEnv{Stats: &stats, IterStats: nil},
 		ReaderProvider:       MakeTrivialReaderProvider(r),
+		BlobContext:          AssertNoBlobHandles,
 	})
 	require.NoError(t, err)
 	defer it.Close()
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index ad1b8ff3..a86555d7 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -66,7 +66,7 @@ func (r *Reader) get(key []byte) (value []byte, err error) {
 		}
 	}
 
-	i, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+	i, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 	if err != nil {
 		return nil, err
 	}
@@ -237,7 +237,7 @@ func runVirtualReaderTest(t *testing.T, path string, blockSize, indexBlockSize i
 			transforms := IterTransforms{
 				SyntheticPrefixAndSuffix: block.MakeSyntheticPrefixAndSuffix(nil, syntheticSuffix),
 			}
-			iter, err := v.NewCompactionIter(transforms, block.ReadEnv{BufferPool: &bp}, rp)
+			iter, err := v.NewCompactionIter(transforms, block.ReadEnv{BufferPool: &bp}, rp, AssertNoBlobHandles)
 			if err != nil {
 				return err.Error()
 			}
@@ -369,6 +369,7 @@ func runVirtualReaderTest(t *testing.T, path string, blockSize, indexBlockSize i
 				FilterBlockSizeLimit: NeverUseFilterBlock,
 				Env:                  block.ReadEnv{Stats: &stats, IterStats: nil},
 				ReaderProvider:       MakeTrivialReaderProvider(r),
+				BlobContext:          AssertNoBlobHandles,
 			})
 			if err != nil {
 				return err.Error()
@@ -584,7 +585,7 @@ func TestInjectedErrors(t *testing.T) {
 				return err
 			}
 
-			iter, err := r.NewIter(NoTransforms, nil, nil)
+			iter, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 			if err != nil {
 				return err
 			}
@@ -777,6 +778,7 @@ func runTestReader(t *testing.T, o WriterOptions, dir string, r *Reader, printVa
 					FilterBlockSizeLimit: AlwaysUseFilterBlock,
 					Env:                  block.ReadEnv{Stats: &stats, IterStats: nil},
 					ReaderProvider:       MakeTrivialReaderProvider(r),
+					BlobContext:          AssertNoBlobHandles,
 				})
 				if err != nil {
 					return err.Error()
@@ -920,7 +922,8 @@ func TestCompactionIteratorSetupForCompaction(t *testing.T) {
 				var pool block.BufferPool
 				pool.Init(5)
 				citer, err := r.NewCompactionIter(
-					NoTransforms, block.ReadEnv{BufferPool: &pool}, MakeTrivialReaderProvider(r))
+					NoTransforms, block.ReadEnv{BufferPool: &pool},
+					MakeTrivialReaderProvider(r), AssertNoBlobHandles)
 				require.NoError(t, err)
 				switch i := citer.(type) {
 				case *singleLevelIteratorRowBlocks:
@@ -980,7 +983,8 @@ func TestReadaheadSetupForV3TablesWithMultipleVersions(t *testing.T) {
 		pool.Init(5)
 		defer pool.Release()
 		citer, err := r.NewCompactionIter(
-			NoTransforms, block.ReadEnv{BufferPool: &pool}, MakeTrivialReaderProvider(r))
+			NoTransforms, block.ReadEnv{BufferPool: &pool},
+			MakeTrivialReaderProvider(r), AssertNoBlobHandles)
 		require.NoError(t, err)
 		defer citer.Close()
 		i := citer.(*singleLevelIteratorRowBlocks)
@@ -988,7 +992,7 @@ func TestReadaheadSetupForV3TablesWithMultipleVersions(t *testing.T) {
 		require.True(t, objstorageprovider.TestingCheckMaxReadahead(i.vbRH))
 	}
 	{
-		iter, err := r.NewIter(NoTransforms, nil, nil)
+		iter, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 		require.NoError(t, err)
 		defer iter.Close()
 		i := iter.(*singleLevelIteratorRowBlocks)
@@ -1277,10 +1281,12 @@ func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
 			},
 			FilterBlockSizeLimit: AlwaysUseFilterBlock,
 			ReaderProvider:       MakeTrivialReaderProvider(eReader),
+			BlobContext:          AssertNoBlobHandles,
 		}, &virtualState{
 			lower: base.MakeInternalKey([]byte("_"), base.SeqNumMax, base.InternalKeyKindSet),
 			upper: base.MakeRangeDeleteSentinelKey([]byte("~~~~~~~~~~~~~~~~")),
 		})
+
 		require.NoError(t, err)
 		return iter, func() {
 			require.NoError(t, iter.Close())
@@ -1496,14 +1502,14 @@ func TestReaderChecksumErrors(t *testing.T) {
 
 								if corruptionType == "first-byte" {
 									require.NoError(t, err)
-									iter, err := r.NewIter(NoTransforms, nil, nil)
+									iter, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 									require.NoError(t, err)
 									for kv := iter.First(); kv != nil; kv = iter.Next() {
 									}
 									require.Regexp(t, `checksum mismatch`, iter.Error())
 									require.Regexp(t, `checksum mismatch`, iter.Close())
 
-									iter, err = r.NewIter(NoTransforms, nil, nil)
+									iter, err = r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 									require.NoError(t, err)
 									for kv := iter.Last(); kv != nil; kv = iter.Prev() {
 									}
@@ -1531,7 +1537,7 @@ func TestReaderChecksumErrors(t *testing.T) {
 									if checkBitFlipErr(err) {
 										break
 									}
-									iter, err := r.NewIter(NoTransforms, nil, nil)
+									iter, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 									if checkBitFlipErr(err) {
 										break
 									}
@@ -1539,7 +1545,7 @@ func TestReaderChecksumErrors(t *testing.T) {
 									}
 									if checkBitFlipErr(iter.Error()) && checkBitFlipErr(iter.Close()) {
 									}
-									iter, err = r.NewIter(NoTransforms, nil, nil)
+									iter, err = r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 									if checkBitFlipErr(err) {
 										break
 									}
@@ -1913,7 +1919,7 @@ func BenchmarkTableIterSeekGE(b *testing.B) {
 				c := cache.New(128 << 20)
 				ch := c.NewHandle()
 				r, keys := buildBenchmarkTable(b, bm.options, false, 0, ch)
-				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 				require.NoError(b, err)
 				rng := rand.New(rand.NewPCG(0, uint64(time.Now().UnixNano())))
 
@@ -1938,7 +1944,7 @@ func BenchmarkTableIterSeekLT(b *testing.B) {
 				c := cache.New(128 << 20)
 				ch := c.NewHandle()
 				r, keys := buildBenchmarkTable(b, bm.options, false, 0, ch)
-				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 				require.NoError(b, err)
 				rng := rand.New(rand.NewPCG(0, uint64(time.Now().UnixNano())))
 
@@ -1963,7 +1969,7 @@ func BenchmarkTableIterNext(b *testing.B) {
 				c := cache.New(128 << 20)
 				ch := c.NewHandle()
 				r, _ := buildBenchmarkTable(b, bm.options, false, 0, ch)
-				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 				require.NoError(b, err)
 
 				b.ResetTimer()
@@ -1996,7 +2002,7 @@ func BenchmarkTableIterPrev(b *testing.B) {
 				c := cache.New(128 << 20)
 				ch := c.NewHandle()
 				r, _ := buildBenchmarkTable(b, bm.options, false, 0, ch)
-				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+				it, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 				require.NoError(b, err)
 
 				b.ResetTimer()
@@ -2084,7 +2090,7 @@ func BenchmarkSeqSeekGEExhausted(b *testing.B) {
 						} else {
 							seekKeys = postKeys
 						}
-						it, err := reader.NewIter(NoTransforms, nil /* lower */, upper)
+						it, err := reader.NewIter(NoTransforms, nil /* lower */, upper, AssertNoBlobHandles)
 						require.NoError(b, err)
 						b.ResetTimer()
 						pos := 0
@@ -2201,14 +2207,14 @@ func BenchmarkIteratorScanManyVersions(b *testing.B) {
 						}()
 						b.Run("NewIter", func(b *testing.B) {
 							for i := 0; i < b.N; i++ {
-								iter, err := r.NewIter(NoTransforms, nil, nil)
+								iter, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 								require.NoError(b, err)
 								require.NoError(b, iter.Close())
 							}
 						})
 						for _, readValue := range []bool{false, true} {
 							b.Run(fmt.Sprintf("read-value=%t", readValue), func(b *testing.B) {
-								iter, err := r.NewIter(NoTransforms, nil, nil)
+								iter, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 								require.NoError(b, err)
 								var kv *base.InternalKV
 								var valBuf [100]byte
@@ -2360,7 +2366,7 @@ func BenchmarkIteratorScanNextPrefix(b *testing.B) {
 				b.Run(fmt.Sprintf("method=%s", method), func(b *testing.B) {
 					for _, readValue := range []bool{false, true} {
 						b.Run(fmt.Sprintf("read-value=%t", readValue), func(b *testing.B) {
-							iter, err := r.NewIter(NoTransforms, nil, nil)
+							iter, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 							require.NoError(b, err)
 							var nextFunc func(index int) *base.InternalKV
 							switch method {
@@ -2502,6 +2508,7 @@ func BenchmarkIteratorScanObsolete(b *testing.B) {
 									Transforms:     IterTransforms{HideObsoletePoints: hideObsoletePoints},
 									Filterer:       filterer,
 									ReaderProvider: MakeTrivialReaderProvider(r),
+									BlobContext:    AssertNoBlobHandles,
 								})
 								require.NoError(b, err)
 								b.ResetTimer()
diff --git a/sstable/table_test.go b/sstable/table_test.go
index 3b3ced83..d0944d35 100644
--- a/sstable/table_test.go
+++ b/sstable/table_test.go
@@ -62,7 +62,7 @@ func check(fs vfs.FS, filename string, comparer *Comparer, fp FilterPolicy) erro
 		}
 
 		// Check using SeekGE.
-		iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+		iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 		if err != nil {
 			return err
 		}
@@ -104,7 +104,7 @@ func check(fs vfs.FS, filename string, comparer *Comparer, fp FilterPolicy) erro
 		}
 
 		// Check using Find.
-		iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+		iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 		if err != nil {
 			return err
 		}
@@ -133,7 +133,7 @@ func check(fs vfs.FS, filename string, comparer *Comparer, fp FilterPolicy) erro
 		{0, "~"},
 	}
 	for _, ct := range countTests {
-		iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+		iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 		if err != nil {
 			return err
 		}
@@ -186,7 +186,7 @@ func check(fs vfs.FS, filename string, comparer *Comparer, fp FilterPolicy) erro
 			upper = []byte(words[upperIdx])
 		}
 
-		iter, err := r.NewIter(NoTransforms, lower, upper)
+		iter, err := r.NewIter(NoTransforms, lower, upper, AssertNoBlobHandles)
 		if err != nil {
 			return err
 		}
@@ -474,7 +474,7 @@ func TestFinalBlockIsWritten(t *testing.T) {
 					if err != nil {
 						t.Errorf("nk=%d, vLen=%d: reader open: %v", nk, vLen, err)
 					}
-					iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+					iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 					require.NoError(t, err)
 					for kv := iter.First(); kv != nil; kv = iter.Next() {
 						got++
@@ -509,7 +509,7 @@ func TestReaderSymtheticSeqNum(t *testing.T) {
 	const syntheticSeqNum = 42
 	transforms := IterTransforms{SyntheticSeqNum: syntheticSeqNum}
 
-	iter, err := r.NewIter(transforms, nil /* lower */, nil /* upper */)
+	iter, err := r.NewIter(transforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 	require.NoError(t, err)
 	for kv := iter.First(); kv != nil; kv = iter.Next() {
 		if syntheticSeqNum != kv.K.SeqNum() {
diff --git a/sstable/test_utils.go b/sstable/test_utils.go
index 23481850..fb05b3d5 100644
--- a/sstable/test_utils.go
+++ b/sstable/test_utils.go
@@ -16,14 +16,14 @@ import (
 // ReadAll returns all point keys, range del spans, and range key spans from an
 // sstable. Closes the Readable. Panics on errors.
 func ReadAll(
-	r objstorage.Readable, ro ReaderOptions,
+	r objstorage.Readable, ro ReaderOptions, blobValueFetcher base.ValueFetcher,
 ) (points []base.InternalKV, rangeDels, rangeKeys []keyspan.Span, err error) {
 	reader, err := NewReader(context.Background(), r, ro)
 	if err != nil {
 		return nil, nil, nil, err
 	}
 	defer reader.Close()
-	pointIter, err := reader.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+	pointIter, err := reader.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 	if err != nil {
 		return nil, nil, nil, err
 	}
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 8803ff1f..f429be57 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -193,7 +193,7 @@ func runDataDriven(t *testing.T, file string, tableFormat TableFormat) {
 			return formatWriterMetadata(td, meta)
 
 		case "scan":
-			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 			if err != nil {
 				return err.Error()
 			}
@@ -398,12 +398,12 @@ func TestWriterWithValueBlocks(t *testing.T) {
 
 		case "scan-raw":
 			// Raw scan does not fetch from value blocks.
-			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 			if err != nil {
 				return err.Error()
 			}
 			forceRowIterIgnoreValueBlocks := func(i *singleLevelIteratorRowBlocks) {
-				i.vbReader = valblk.Reader{}
+				i.internalValueConstructor.vbReader = valblk.Reader{}
 				i.data.SetGetLazyValuer(nil)
 				i.data.SetHasValuePrefix(false)
 			}
@@ -442,7 +442,7 @@ func TestWriterWithValueBlocks(t *testing.T) {
 			return buf.String()
 
 		case "scan":
-			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 			if err != nil {
 				return err.Error()
 			}
@@ -458,7 +458,7 @@ func TestWriterWithValueBlocks(t *testing.T) {
 			return buf.String()
 
 		case "scan-cloned-lazy-values":
-			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */)
+			iter, err := r.NewIter(NoTransforms, nil /* lower */, nil /* upper */, AssertNoBlobHandles)
 			if err != nil {
 				return err.Error()
 			}
@@ -1128,7 +1128,7 @@ func TestWriterRace(t *testing.T) {
 			r, err := NewMemReader(f.Data(), readerOpts)
 			require.NoError(t, err)
 			defer r.Close()
-			it, err := r.NewIter(NoTransforms, nil, nil)
+			it, err := r.NewIter(NoTransforms, nil, nil, AssertNoBlobHandles)
 			require.NoError(t, err)
 			defer it.Close()
 			ki := 0
diff --git a/testdata/iter_histories/blob_references b/testdata/iter_histories/blob_references
index 6a8b0267..ccac33c3 100644
--- a/testdata/iter_histories/blob_references
+++ b/testdata/iter_histories/blob_references
@@ -1,3 +1,6 @@
+# Test a simple scenario where two distinct sstables contain blob handles
+# referencing the same file.
+
 define verbose format-major-version=21
 L5
   b@9.SET.9:v
@@ -12,3 +15,176 @@ L5:
   000004:[b@9#9,SET-d@9#9,SET] seqnums:[9-9] points:[b@9#9,SET-d@9#9,SET] size:885 blobrefs:[(000921: 10); depth:1]
 L6:
   000005:[b@2#2,SET-d@2#2,SET] seqnums:[2-2] points:[b@2#2,SET-d@2#2,SET] size:863 blobrefs:[(000921: 6); depth:1]
+
+combined-iter
+first
+next
+next
+next
+next
+next
+next
+stats
+----
+b@9: (v, .)
+b@2: (v, .)
+c@9: (helloworld, .)
+c@2: (foobar, .)
+d@9: (v, .)
+d@2: (v, .)
+.
+stats: seeked 1 times (1 internal); stepped 6 times (6 internal); blocks: 0B cached, 342B not cached (read time: 0s); points: 6 (18B keys, 8B values); separated: 2 (16B, 16B fetched)
+
+# Try the same but avoid fetching one of the values (by using NextPrefix to step
+# over it).
+
+combined-iter
+first
+next
+next
+next-prefix
+next
+next
+stats
+----
+b@9: (v, .)
+b@2: (v, .)
+c@9: (helloworld, .)
+d@9: (v, .)
+d@2: (v, .)
+.
+stats: seeked 1 times (1 internal); stepped 5 times (6 internal); blocks: 342B cached; points: 6 (18B keys, 8B values); separated: 2 (16B, 10B fetched)
+
+# Test a couple of blob files interleaved.
+
+define verbose format-major-version=21
+L5
+  b@9.SETWITHDEL.9:blob{fileNum=000039 value=orange}
+  c@9.SETWITHDEL.9:blob{fileNum=000921 value=canteloupe}
+  d@9.SETWITHDEL.9:blob{fileNum=000039 value=honeydew}
+  e@1.SETWITHDEL.9:blob{fileNum=000921 value=watermelon}
+L6
+  b@2.SETWITHDEL.9:blob{fileNum=000039 value=lemon}
+  c@2.SETWITHDEL.2:blob{fileNum=000921 value=kiwi}
+  d@2.SETWITHDEL.2:blob{fileNum=000921 value=tangerine}
+  f@2.SETWITHDEL.3:blob{fileNum=000039 value=grapes}
+----
+L5:
+  000004:[b@9#9,SETWITHDEL-e@1#9,SETWITHDEL] seqnums:[9-9] points:[b@9#9,SETWITHDEL-e@1#9,SETWITHDEL] size:866 blobrefs:[(000039: 14), (000921: 20); depth:2]
+L6:
+  000005:[b@2#9,SETWITHDEL-f@2#3,SETWITHDEL] seqnums:[2-9] points:[b@2#9,SETWITHDEL-f@2#3,SETWITHDEL] size:875 blobrefs:[(000039: 11), (000921: 13); depth:2]
+
+# The iterator stats should indicate that only the first value from each file
+# should trigger a read of the blob file. Once loaded, subsequent reads of the
+# values within the same blocks should not be recorded as block loads cached or
+# uncached, because the blob value fetcher retains its handle on the block.
+#
+# TODO(jackson): Add stats on the count of blob value retrievals that hit on the
+# existing cached readers.
+
+combined-iter
+first
+stats
+next
+stats
+next
+stats
+next
+next
+next
+stats
+next
+next
+next
+stats
+----
+b@9: (orange, .)
+stats: seeked 1 times (1 internal); stepped 0 times (0 internal); blocks: 0B cached, 345B not cached (read time: 0s); points: 1 (3B keys, 2B values); separated: 2 (11B, 6B fetched)
+b@2: (lemon, .)
+stats: seeked 1 times (1 internal); stepped 1 times (1 internal); blocks: 0B cached, 345B not cached (read time: 0s); points: 2 (6B keys, 4B values); separated: 3 (21B, 11B fetched)
+c@9: (canteloupe, .)
+stats: seeked 1 times (1 internal); stepped 2 times (2 internal); blocks: 0B cached, 381B not cached (read time: 0s); points: 3 (9B keys, 6B values); separated: 4 (25B, 21B fetched)
+c@2: (kiwi, .)
+d@9: (honeydew, .)
+d@2: (tangerine, .)
+stats: seeked 1 times (1 internal); stepped 5 times (5 internal); blocks: 0B cached, 381B not cached (read time: 0s); points: 6 (18B keys, 12B values); separated: 7 (52B, 42B fetched)
+e@1: (watermelon, .)
+f@2: (grapes, .)
+.
+stats: seeked 1 times (1 internal); stepped 8 times (8 internal); blocks: 0B cached, 381B not cached (read time: 0s); points: 8 (24B keys, 16B values); separated: 8 (58B, 58B fetched)
+
+# Test scanning a table, stepping into new blocks of the blob file. The stats
+# should reflect that a block is only loaded when stepping into a new block.
+
+define verbose format-major-version=21
+L6
+  a.SETWITHDEL.2:blob{fileNum=000009 value=lemonmeringue}
+  b.SETWITHDEL.2:blob{fileNum=000009 value=keylime}
+  c.SETWITHDEL.2:blob{fileNum=000009 value=pecan blockNum=1 offset=0}
+  d.SETWITHDEL.2:blob{fileNum=000009 value=cherry}
+  e.SETWITHDEL.2:blob{fileNum=000009 value=apple}
+  f.SETWITHDEL.2:blob{fileNum=000009 value=bananacream blockNum=2 offset=0}
+  g.SETWITHDEL.2:blob{fileNum=000009 value=chocolate}
+  h.SETWITHDEL.2:blob{fileNum=000009 value=strawberry blockNum=3 offset=0}
+  i.SETWITHDEL.2:blob{fileNum=000009 value=custard}
+  j.SETWITHDEL.2:blob{fileNum=000009 value=blueberry blockNum=4 offset=0}
+  k.SETWITHDEL.2:blob{fileNum=000009 value=raspberry blockNum=5 offset=0}
+  l.SETWITHDEL.2:blob{fileNum=000009 value=peach blockNum=6 offset=0}
+----
+L6:
+  000004:[a#2,SETWITHDEL-l#2,SETWITHDEL] seqnums:[2-2] points:[a#2,SETWITHDEL-l#2,SETWITHDEL] size:945 blobrefs:[(000009: 96); depth:1]
+
+combined-iter
+first
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+next
+stats
+----
+a: (lemonmeringue, .)
+stats: seeked 1 times (1 internal); stepped 0 times (0 internal); blocks: 0B cached, 264B not cached (read time: 0s); points: 1 (1B keys, 2B values); separated: 1 (13B, 13B fetched)
+b: (keylime, .)
+stats: seeked 1 times (1 internal); stepped 1 times (1 internal); blocks: 0B cached, 264B not cached (read time: 0s); points: 2 (2B keys, 4B values); separated: 2 (20B, 20B fetched)
+c: (pecan, .)
+stats: seeked 1 times (1 internal); stepped 2 times (2 internal); blocks: 0B cached, 280B not cached (read time: 0s); points: 3 (3B keys, 6B values); separated: 3 (25B, 25B fetched)
+d: (cherry, .)
+stats: seeked 1 times (1 internal); stepped 3 times (3 internal); blocks: 0B cached, 280B not cached (read time: 0s); points: 4 (4B keys, 8B values); separated: 4 (31B, 31B fetched)
+e: (apple, .)
+stats: seeked 1 times (1 internal); stepped 4 times (4 internal); blocks: 0B cached, 280B not cached (read time: 0s); points: 5 (5B keys, 10B values); separated: 5 (36B, 36B fetched)
+f: (bananacream, .)
+stats: seeked 1 times (1 internal); stepped 5 times (5 internal); blocks: 0B cached, 300B not cached (read time: 0s); points: 6 (6B keys, 12B values); separated: 6 (47B, 47B fetched)
+g: (chocolate, .)
+stats: seeked 1 times (1 internal); stepped 6 times (6 internal); blocks: 0B cached, 300B not cached (read time: 0s); points: 7 (7B keys, 14B values); separated: 7 (56B, 56B fetched)
+h: (strawberry, .)
+stats: seeked 1 times (1 internal); stepped 7 times (7 internal); blocks: 0B cached, 317B not cached (read time: 0s); points: 8 (8B keys, 16B values); separated: 8 (66B, 66B fetched)
+i: (custard, .)
+stats: seeked 1 times (1 internal); stepped 8 times (8 internal); blocks: 0B cached, 317B not cached (read time: 0s); points: 9 (9B keys, 18B values); separated: 9 (73B, 73B fetched)
+j: (blueberry, .)
+stats: seeked 1 times (1 internal); stepped 9 times (9 internal); blocks: 0B cached, 326B not cached (read time: 0s); points: 10 (10B keys, 20B values); separated: 10 (82B, 82B fetched)
+k: (raspberry, .)
+stats: seeked 1 times (1 internal); stepped 10 times (10 internal); blocks: 0B cached, 335B not cached (read time: 0s); points: 11 (11B keys, 22B values); separated: 11 (91B, 91B fetched)
+l: (peach, .)
+stats: seeked 1 times (1 internal); stepped 11 times (11 internal); blocks: 0B cached, 340B not cached (read time: 0s); points: 12 (12B keys, 24B values); separated: 12 (96B, 96B fetched)
+.
+stats: seeked 1 times (1 internal); stepped 12 times (12 internal); blocks: 0B cached, 340B not cached (read time: 0s); points: 12 (12B keys, 24B values); separated: 12 (96B, 96B fetched)
diff --git a/testdata/value_separation_policy b/testdata/value_separation_policy
index 1144f613..310ff2bd 100644
--- a/testdata/value_separation_policy
+++ b/testdata/value_separation_policy
@@ -106,7 +106,7 @@ c#9,SET:paddy
 ----
 # create: 000005.sst
 RawWriter.AddWithBlobHandle("a#9,SET", "(f0,blk0[0:5])", 5, false)
-RawWriter.AddWithBlobHandle("a#5,SET", "(f1,blk0[5:9])", 4, false)
+RawWriter.AddWithBlobHandle("a#5,SET", "(f1,blk0[0:4])", 4, false)
 RawWriter.Add("b#2,DEL", "", false)
 RawWriter.Add("c#9,SET", "paddy", false)
 
@@ -130,7 +130,7 @@ e#9,SET:blob{value=yayoi}
 ----
 # create: 000006.sst
 RawWriter.Add("d#2,DEL", "", false)
-RawWriter.AddWithBlobHandle("e#9,SET", "(f0,blk0[9:14])", 5, false)
+RawWriter.AddWithBlobHandle("e#9,SET", "(f0,blk0[4:9])", 5, false)
 
 estimated-sizes
 ----
diff --git a/tool/testdata/sstable_check b/tool/testdata/sstable_check
index 07910aa5..455e5f0c 100644
--- a/tool/testdata/sstable_check
+++ b/tool/testdata/sstable_check
@@ -44,7 +44,7 @@ WARNING: OUT OF ORDER KEYS!
 sstable check
 testdata/corrupted.sst
 ----
-corrupted.sst: pebble/table: table 000000: block 87/465: crc32c checksum mismatch c8539ba5 != b972e324
+corrupted.sst: pebble: file 000000: block 87/465: crc32c checksum mismatch c8539ba5 != b972e324
 
 sstable check
 testdata/bad-magic.sst