Repository: cockroachdb/pebble (go)

Issue:
cache: check value before setting referenced atomic
In a TPCC profile the `referenced.Store(true)` in the cache hit path takes about 0.1%. Not a lot, but we should try to check the value first and only store if necessary. An atomic load is a simple load, whereas an atomic store is an atomic XCHG operation which involves a memory barrier.

In the profile, the main code paths where through reading an index and a filter block; it is likely that these are accessed much more frequently than the clockpro hands make a full rotation.

<img width="1066" alt="Image" src="https://github.com/user-attachments/assets/881ab3fb-cddc-480a-bb58-37b96ae4981e" />

Jira issue: PEBBLE-340

Pull request description:
cache: Load before Store
#### cache: fix and improve Get benchmark


#### cache: Load before Store

On x86 an atomic.Load is a simple load, whereas a Store is an atomic
`XCHG` operation which involves a memory barrier. Frequently accessed
blocks are likely to be accessed many times between two sweeps of the
clockpro hand.

On a gceworker:
```
name         old time/op    new time/op    delta
CacheGet-24    20.1ns ± 4%    18.9ns ± 5%  -5.82%  (p=0.002 n=10+10)
```

Fixes #4328

Patch:
diff --git a/internal/cache/clockpro.go b/internal/cache/clockpro.go
index cbd55d4b..549552f8 100644
--- a/internal/cache/clockpro.go
+++ b/internal/cache/clockpro.go
@@ -140,7 +140,8 @@ func (c *shard) getWithMaybeReadEntry(k key, desireReadEntry bool) (*Value, *rea
 	var value *Value
 	if e, _ := c.blocks.Get(k); e != nil {
 		value = e.acquireValue()
-		if value != nil {
+		// Note: we Load first to avoid an atomic XCHG when not necessary.
+		if value != nil && !e.referenced.Load() {
 			e.referenced.Store(true)
 		}
 	}
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index a4e3cce7..57fce905 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -588,7 +588,7 @@ func (i *Iter) Next() (*base.InternalKey, base.LazyValue) {
 			origSnapshotIdx := i.curSnapshotIdx
 			var valueMerger base.ValueMerger
 			// MERGE values are always stored in-place.
-			valueMerger, i.err = i.cfg.Merge(i.iterKV.K.UserKey, i.iterKV.V.InPlaceValue())
+			valueMerger, i.err = i.cfg.Merge(i.iterKV.K.UserKey, i.iterKV.InPlaceValue())
 			if i.err == nil {
 				i.mergeNext(valueMerger)
 			}
@@ -1115,7 +1115,7 @@ func (i *Iter) deleteSizedNext() (*base.InternalKey, base.LazyValue) {
 	// In this case, we still peek forward in case there's another DELSIZED key
 	// with a lower sequence number, in which case we'll adopt its value.
 	// If the DELSIZED does have a value, it must be in-place.
-	i.valueBuf = append(i.valueBuf[:0], i.iterKV.V.InPlaceValue()...)
+	i.valueBuf = append(i.valueBuf[:0], i.iterKV.InPlaceValue()...)
 	i.value = base.MakeInPlaceValue(i.valueBuf)
 
 	// Loop through all the keys within this stripe that are skippable.
@@ -1289,7 +1289,7 @@ func (i *Iter) saveValue() {
 	// TODO(jackson): With the introduction of values stored in separate
 	// physical blob files, this should begin to Clone LazyValues that are
 	// stored in blob reference files when non-rewriting blob files.
-	v, callerOwned, err := i.iterKV.V.Value(i.valueBuf[:0])
+	v, callerOwned, err := i.iterKV.Value(i.valueBuf[:0])
 	if err != nil {
 		i.err = err
 		i.value = base.LazyValue{}
diff --git a/level_iter.go b/level_iter.go
index 11d39d37..9e8b08c3 100644
--- a/level_iter.go
+++ b/level_iter.go
@@ -26,7 +26,7 @@ type internalIterOpts struct {
 	compaction           bool
 	bufferPool           *block.BufferPool
 	stats                *base.InternalIteratorStats
-	iterStatsAccumulator block.IterStatsAccumulator
+	iterStatsAccumulator *block.CategoryStatsShard
 	boundLimitedFilter   sstable.BoundLimitedBlockPropertyFilter
 }
 
diff --git a/record/record.go b/record/record.go
index cfdfeb81..920fc3f0 100644
--- a/record/record.go
+++ b/record/record.go
@@ -28,17 +28,7 @@
 //			if err == io.EOF {
 //				break
 //			}
-//			if err != nil {
-//				log.Printf("recovering from %v", err)
-//				r.Recover()
-//				continue
-//			}
 //			s, err := io.ReadAll(rec)
-//			if err != nil {
-//				log.Printf("recovering from %v", err)
-//				r.Recover()
-//				continue
-//			}
 //			ss = append(ss, string(s))
 //		}
 //		return ss, nil
@@ -93,10 +83,6 @@
 // extra "recyclable" chunk types that map directly to the legacy chunk types
 // (i.e. full, first, middle, last). The CRC is computed over the type, log
 // number, and payload.
-//
-// The wire format allows for limited recovery in the face of data corruption:
-// on a format error (such as a checksum mismatch), the reader moves to the
-// next block and looks for the next full or first chunk.
 package record
 
 // The C++ Level-DB code calls this the log, but it has been renamed to record
@@ -227,8 +213,6 @@ type Reader struct {
 	// n is the number of bytes of buf that are valid. Once reading has started,
 	// only the final block can have n < blockSize.
 	n int
-	// recovering is true when recovering from corruption.
-	recovering bool
 	// last is whether the current chunk is the last chunk of the record.
 	last bool
 	// err is any accumulated error.
@@ -270,15 +254,6 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 					r.end = r.n
 					continue
 				}
-				if r.recovering {
-					// Skip the rest of the block, if it looks like it is all
-					// zeroes. This is common with WAL preallocation.
-					//
-					// Set r.err to be an error so r.recover actually recovers.
-					r.err = ErrZeroedChunk
-					r.recover()
-					continue
-				}
 				return ErrZeroedChunk
 			}
 
@@ -307,17 +282,9 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 			r.end = r.begin + int(length)
 			if r.end > r.n {
 				// The chunk straddles a 32KB boundary (or the end of file).
-				if r.recovering {
-					r.recover()
-					continue
-				}
 				return ErrInvalidChunk
 			}
 			if checksum != crc.New(r.buf[r.begin-headerSize+6:r.end]).Value() {
-				if r.recovering {
-					r.recover()
-					continue
-				}
 				return ErrInvalidChunk
 			}
 			if wantFirst {
@@ -326,7 +293,6 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 				}
 			}
 			r.last = chunkPosition == fullChunkPosition || chunkPosition == lastChunkPosition
-			r.recovering = false
 			return nil
 		}
 		if r.n < blockSize && r.blockNum >= 0 {
@@ -375,23 +341,6 @@ func (r *Reader) Offset() int64 {
 	return int64(r.blockNum)*blockSize + int64(r.end)
 }
 
-// recover clears any errors read so far, so that calling Next will start
-// reading from the next good 32KiB block. If there are no such blocks, Next
-// will return io.EOF. recover also marks the current reader, the one most
-// recently returned by Next, as stale. If recover is called without any
-// prior error, then recover is a no-op.
-func (r *Reader) recover() {
-	if r.err == nil {
-		return
-	}
-	r.recovering = true
-	r.err = nil
-	// Discard the rest of the current block.
-	r.begin, r.end, r.last = r.n, r.n, false
-	// Invalidate any outstanding singleReader.
-	r.seq++
-}
-
 // seekRecord seeks in the underlying io.Reader such that calling r.Next
 // returns the record whose first chunk header starts at the provided offset.
 // Its behavior is undefined if the argument given is not such an offset, as
@@ -401,12 +350,7 @@ func (r *Reader) recover() {
 // io.Seeker.
 //
 // seekRecord will fail and return an error if the Reader previously
-// encountered an error, including io.EOF. Such errors can be cleared by
-// calling Recover. Calling seekRecord after Recover will make calling Next
-// return the record at the given offset, instead of the record at the next
-// good 32KiB block as Recover normally would. Calling seekRecord before
-// Recover has no effect on Recover's semantics other than changing the
-// starting point for determining the next good 32KiB block.
+// encountered an error, including io.EOF.
 //
 // The offset is always relative to the start of the underlying io.Reader, so
 // negative values will result in an error as per io.Seeker.
@@ -429,7 +373,7 @@ func (r *Reader) seekRecord(offset int64) error {
 
 	// Clear the state of the internal reader.
 	r.begin, r.end, r.n = 0, 0, 0
-	r.blockNum, r.recovering, r.last = -1, false, false
+	r.blockNum, r.last = -1, false
 	if r.err = r.nextChunk(false); r.err != nil {
 		return r.err
 	}
diff --git a/sstable/block/block.go b/sstable/block/block.go
index 938a66ec..239ac05e 100644
--- a/sstable/block/block.go
+++ b/sstable/block/block.go
@@ -323,7 +323,7 @@ type ReadEnv struct {
 	// iterator is closed. In the important code paths, the CategoryStatsCollector
 	// is managed by the fileCacheContainer.
 	Stats     *base.InternalIteratorStats
-	IterStats IterStatsAccumulator
+	IterStats *CategoryStatsShard
 
 	// BufferPool is not-nil if we read blocks into a buffer pool and not into the
 	// cache. This is used during compactions.
diff --git a/sstable/block/category_stats.go b/sstable/block/category_stats.go
index a516371f..abe7cb9b 100644
--- a/sstable/block/category_stats.go
+++ b/sstable/block/category_stats.go
@@ -6,6 +6,7 @@ package block
 
 import (
 	"cmp"
+	"runtime"
 	"slices"
 	"sync"
 	"sync/atomic"
@@ -32,6 +33,15 @@ const CategoryUnknown Category = 0
 // number of categories that can be registered.
 const CategoryMax = 30
 
+// shardPadding pads each shard to 64 bytes so they don't share a cache line.
+const shardPadding = 64 - unsafe.Sizeof(CategoryStatsShard{})
+
+// paddedCategoryStatsShard is a single shard of a category's statistics.
+type paddedCategoryStatsShard struct {
+	CategoryStatsShard
+	_ [shardPadding]byte
+}
+
 func (c Category) String() string {
 	return categories[c].name
 }
@@ -169,20 +179,32 @@ type CategoryStatsAggregate struct {
 	CategoryStats CategoryStats
 }
 
-const numCategoryStatsShards = 16
+// numCategoryStatsShards must be a power of 2. We initialize it to GOMAXPROCS
+// (rounded up to the nearest power of 2) or 16, whichever is larger.
+var numCategoryStatsShards = func() int {
+	p := runtime.GOMAXPROCS(0)
+	n := 16
+	for n < p {
+		n *= 2
+	}
+	return n
+}()
 
-type categoryStatsWithMu struct {
-	mu sync.Mutex
-	// Protected by mu.
-	stats CategoryStats
+// CategoryStatsShard holds CategoryStats with a mutex
+// to ensure safe access.
+type CategoryStatsShard struct {
+	mu struct {
+		sync.Mutex
+		stats CategoryStats
+	}
 }
 
 // Accumulate implements the IterStatsAccumulator interface.
-func (c *categoryStatsWithMu) Accumulate(
+func (c *CategoryStatsShard) Accumulate(
 	blockBytes, blockBytesInCache uint64, blockReadDuration time.Duration,
 ) {
 	c.mu.Lock()
-	c.stats.aggregate(blockBytes, blockBytesInCache, blockReadDuration)
+	c.mu.stats.aggregate(blockBytes, blockBytesInCache, blockReadDuration)
 	c.mu.Unlock()
 }
 
@@ -199,11 +221,7 @@ type CategoryStatsCollector struct {
 // contention on the category stats mutex has been observed.
 type shardedCategoryStats struct {
 	Category Category
-	shards   [numCategoryStatsShards]struct {
-		categoryStatsWithMu
-		// Pad each shard to 64 bytes so they don't share a cache line.
-		_ [64 - unsafe.Sizeof(categoryStatsWithMu{})]byte
-	}
+	shards   []paddedCategoryStatsShard
 }
 
 // getStats retrieves the aggregated stats for the category, summing across all
@@ -214,7 +232,7 @@ func (s *shardedCategoryStats) getStats() CategoryStatsAggregate {
 	}
 	for i := range s.shards {
 		s.shards[i].mu.Lock()
-		agg.CategoryStats.aggregate(s.shards[i].stats.BlockBytes, s.shards[i].stats.BlockBytesInCache, s.shards[i].stats.BlockReadDuration)
+		agg.CategoryStats.aggregate(s.shards[i].mu.stats.BlockBytes, s.shards[i].mu.stats.BlockBytesInCache, s.shards[i].mu.stats.BlockReadDuration)
 		s.shards[i].mu.Unlock()
 	}
 	return agg
@@ -222,20 +240,21 @@ func (s *shardedCategoryStats) getStats() CategoryStatsAggregate {
 
 // Accumulator returns a stats accumulator for the given category. The provided
 // p is used to detrmine which shard to write stats to.
-func (c *CategoryStatsCollector) Accumulator(p uint64, category Category) IterStatsAccumulator {
+func (c *CategoryStatsCollector) Accumulator(p uint64, category Category) *CategoryStatsShard {
 	v, ok := c.statsMap.Load(category)
 	if !ok {
 		c.mu.Lock()
 		v, _ = c.statsMap.LoadOrStore(category, &shardedCategoryStats{
 			Category: category,
+			shards:   make([]paddedCategoryStatsShard, numCategoryStatsShards),
 		})
 		c.mu.Unlock()
 	}
 	s := v.(*shardedCategoryStats)
 	// This equation is taken from:
 	// https://en.wikipedia.org/wiki/Linear_congruential_generator#Parameters_in_common_use
-	shard := ((p * 25214903917) >> 32) & (numCategoryStatsShards - 1)
-	return &s.shards[shard].categoryStatsWithMu
+	shard := ((p * 25214903917) >> 32) & uint64(numCategoryStatsShards-1)
+	return &s.shards[shard].CategoryStatsShard
 }
 
 // GetStats returns the aggregated stats.
@@ -251,8 +270,3 @@ func (c *CategoryStatsCollector) GetStats() []CategoryStatsAggregate {
 	})
 	return stats
 }
-
-type IterStatsAccumulator interface {
-	// Accumulate accumulates the provided stats.
-	Accumulate(blockBytes, blockBytesInCache uint64, blockReadDuration time.Duration)
-}
diff --git a/sstable/rowblk/rowblk_index_iter.go b/sstable/rowblk/rowblk_index_iter.go
index d7eafcd1..56af8025 100644
--- a/sstable/rowblk/rowblk_index_iter.go
+++ b/sstable/rowblk/rowblk_index_iter.go
@@ -79,7 +79,7 @@ func (i *IndexIter) SeparatorGT(key []byte, inclusively bool) bool {
 // BlockHandleWithProperties decodes the block handle with any encoded
 // properties at the iterator's current position.
 func (i *IndexIter) BlockHandleWithProperties() (block.HandleWithProperties, error) {
-	return block.DecodeHandleWithProperties(i.iter.ikv.V.ValueOrHandle)
+	return block.DecodeHandleWithProperties(i.iter.ikv.InPlaceValue())
 }
 
 // SeekGE seeks the index iterator to the first block entry with a separator key


Test patch:
diff --git a/cockroachkvs/cockroachkvs_test.go b/cockroachkvs/cockroachkvs_test.go
index 1c4cb860..524de3ec 100644
--- a/cockroachkvs/cockroachkvs_test.go
+++ b/cockroachkvs/cockroachkvs_test.go
@@ -421,8 +421,8 @@ func testCockroachDataColBlock(t *testing.T, seed uint64, keyCfg keyGenConfig) {
 		if !bytes.Equal(kv.K.UserKey, keys[i]) {
 			t.Fatalf("expected %q, but found %q", keys[i], kv.K.UserKey)
 		}
-		if !bytes.Equal(kv.V.InPlaceValue(), values[i]) {
-			t.Fatalf("expected %x, but found %x", values[i], kv.V.InPlaceValue())
+		if !bytes.Equal(kv.InPlaceValue(), values[i]) {
+			t.Fatalf("expected %x, but found %x", values[i], kv.InPlaceValue())
 		}
 	}
 	require.Equal(t, len(keys), i)
@@ -440,7 +440,7 @@ func testCockroachDataColBlock(t *testing.T, seed uint64, keyCfg keyGenConfig) {
 			)
 		}
 		// Search for the correct value among all keys that are equal.
-		for !bytes.Equal(kv.V.InPlaceValue(), values[i]) {
+		for !bytes.Equal(kv.InPlaceValue(), values[i]) {
 			kv = it.Next()
 			if kv == nil || Compare(kv.K.UserKey, keys[i]) != 0 {
 				t.Fatalf("could not find correct value for %s", formatUserKey(keys[i]))
@@ -645,7 +645,7 @@ func formatInternalKey(key base.InternalKey) string {
 //
 //	foo @ 0001020304050607 #1,SET
 func formatInternalKV(kv base.InternalKV) string {
-	val, _, err := kv.V.Value(nil)
+	val, _, err := kv.Value(nil)
 	if err != nil {
 		panic(err)
 	}
diff --git a/cockroachkvs/key_schema_test.go b/cockroachkvs/key_schema_test.go
index dcaae8f3..de198289 100644
--- a/cockroachkvs/key_schema_test.go
+++ b/cockroachkvs/key_schema_test.go
@@ -164,7 +164,7 @@ func TestKeySchema_RandomKeys(t *testing.T) {
 		validateEngineKey.MustValidate(kv.K.UserKey)
 
 		// We write keys[k] as the value too, so check that it's verbatim equal.
-		value, callerOwned, err := kv.V.Value(valBuf)
+		value, callerOwned, err := kv.Value(valBuf)
 		require.NoError(t, err)
 		require.Equal(t, keys[k], value)
 		if callerOwned {
@@ -191,7 +191,7 @@ func TestKeySchema_RandomKeys(t *testing.T) {
 		require.Equal(t, 0, Compare(keys[i], kv.K.UserKey))
 
 		// We write keys[k] as the value too, so check that it's verbatim equal.
-		value, callerOwned, err := kv.V.Value(valBuf)
+		value, callerOwned, err := kv.Value(valBuf)
 		require.NoError(t, err)
 		require.Equal(t, keys[i], value)
 		if callerOwned {
diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go
index ef19a032..e3669106 100644
--- a/internal/cache/cache_test.go
+++ b/internal/cache/cache_test.go
@@ -251,13 +251,14 @@ func TestCacheStressSetExisting(t *testing.T) {
 func BenchmarkCacheGet(b *testing.B) {
 	const size = 100000
 
-	cache := newCache(size, 1)
+	n := runtime.GOMAXPROCS(0)
+	cache := newCache(size*int64(n), n)
 	defer cache.Unref()
 	h := cache.NewHandle()
 	defer h.Close()
 
 	for i := 0; i < size; i++ {
-		setTestValue(h, 0, 0, "a", 1)
+		setTestValue(h, 0, uint64(i), "a", 1)
 	}
 
 	b.ResetTimer()
diff --git a/internal/metamorphic/crossversion/crossversion_test.go b/internal/metamorphic/crossversion/crossversion_test.go
index 7285465e..a4796c23 100644
--- a/internal/metamorphic/crossversion/crossversion_test.go
+++ b/internal/metamorphic/crossversion/crossversion_test.go
@@ -17,6 +17,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -130,18 +131,6 @@ type pebbleVersion struct {
 	TestBinaryPath string
 }
 
-type initialState struct {
-	desc string
-	path string
-}
-
-func (s initialState) String() string {
-	if s.desc == "" {
-		return "<empty>"
-	}
-	return s.desc
-}
-
 func runCrossVersion(
 	ctx context.Context,
 	t *testing.T,
@@ -179,22 +168,22 @@ func runCrossVersion(
 	initialStates := []initialState{{}}
 	for i := range versions {
 		t.Logf("Running tests with version %s with %d initial state(s).", versions[i].SHA, len(initialStates))
-		histories, nextInitialStates, err := runVersion(ctx, t, &fatalOnce, rootDir, versions[i], versionSeeds[i], initialStates)
-		if err != nil {
-			return err
-		}
+		subrunResults := runVersion(ctx, t, &fatalOnce, rootDir, versions[i], versionSeeds[i], initialStates)
 
 		// All the initial states described the same state and all of this
 		// version's metamorphic runs used the same seed, so all of the
 		// resulting histories should be identical.
+		histories := subrunResults.historyPaths()
 		if h, diff := metamorphic.CompareHistories(t, histories); h > 0 {
-			fatalf(t, &fatalOnce, rootDir, "Metamorphic test divergence between %q and %q:\nDiff:\n%s",
-				nextInitialStates[0].desc, nextInitialStates[h].desc, diff)
+			fatalf(t, &fatalOnce, []string{subrunResults[0].runDir, subrunResults[h].runDir},
+				"Metamorphic test divergence between %q and %q:\nDiff:\n%s",
+				subrunResults[0].initialState.desc, subrunResults[h].initialState.desc, diff)
 		}
 
 		// Prune the set of initial states we collected for this version, using
 		// the deterministic randomness of prng to pick which states we keep.
-		if len(nextInitialStates) > factor {
+		nextInitialStates := subrunResults.initialStates()
+		if len(subrunResults) > factor {
 			prng.Shuffle(len(nextInitialStates), func(i, j int) {
 				nextInitialStates[i], nextInitialStates[j] = nextInitialStates[j], nextInitialStates[i]
 			})
@@ -209,6 +198,52 @@ func runCrossVersion(
 	return nil
 }
 
+// subrunResult describes the output of a single run of the metamorphic test
+// against a particular version.
+type subrunResult struct {
+	// runDir is the path to the directory containing the output of this
+	// TestMeta run.
+	runDir string
+	// historyPath is the path to the history file recording the output of this
+	// run. All the histories produced by a single call to runVersion should be
+	// semantically equivalent.
+	historyPath string
+	// initialState describes the location of the database on disk after the run
+	// completed. It can be used to seed a new run of the metamorphic test in a
+	// later Pebble version.
+	initialState initialState
+}
+
+type initialState struct {
+	desc string
+	path string
+}
+
+func (s initialState) String() string {
+	if s.desc == "" {
+		return "<empty>"
+	}
+	return s.desc
+}
+
+type subrunResults []subrunResult
+
+func (r subrunResults) historyPaths() []string {
+	paths := make([]string, len(r))
+	for i := range r {
+		paths[i] = r[i].historyPath
+	}
+	return paths
+}
+
+func (r subrunResults) initialStates() []initialState {
+	states := make([]initialState, len(r))
+	for i := range r {
+		states[i] = r[i].initialState
+	}
+	return states
+}
+
 func runVersion(
 	ctx context.Context,
 	t *testing.T,
@@ -217,7 +252,7 @@ func runVersion(
 	vers pebbleVersion,
 	seed uint64,
 	initialStates []initialState,
-) (histories []string, nextInitialStates []initialState, err error) {
+) (results subrunResults) {
 	// mu guards histories and nextInitialStates. The subtests may be run in
 	// parallel (via t.Parallel()).
 	var mu sync.Mutex
@@ -248,7 +283,8 @@ func runVersion(
 				t.Logf("  Running test with version %s with initial state %s.",
 					vers.SHA, s)
 				if err := r.run(ctx, out); err != nil {
-					fatalf(t, fatalOnce, rootDir, "Metamorphic test failed: %s\nOutput:%s\n", err, buf.String())
+					fatalf(t, fatalOnce, []string{r.dir},
+						"Metamorphic test failed: %s\nOutput:%s\n", err, buf.String())
 				}
 
 				// dir is a directory containing the ops file and subdirectories for
@@ -269,37 +305,46 @@ func runVersion(
 				mu.Lock()
 				defer mu.Unlock()
 				for _, subrunDir := range subrunDirs {
-					// Record the subrun as an initial state for the next version.
-					nextInitialStates = append(nextInitialStates, initialState{
-						path: filepath.Join(dir, subrunDir),
-						desc: fmt.Sprintf("sha=%s-seed=%d-opts=%s(%s)", vers.SHA, seed, subrunDir, s.String()),
+					results = append(results, subrunResult{
+						runDir:      dir,
+						historyPath: filepath.Join(dir, subrunDir, "history"),
+						initialState: initialState{
+							path: filepath.Join(dir, subrunDir),
+							desc: fmt.Sprintf("sha=%s-seed=%d-opts=%s(%s)", vers.SHA, seed, subrunDir, s.String()),
+						},
 					})
-					histories = append(histories, filepath.Join(dir, subrunDir, "history"))
 				}
 			})
 		}
 	})
-	return histories, nextInitialStates, err
+	return results
 }
 
-func fatalf(t testing.TB, fatalOnce *sync.Once, dir string, msg string, args ...interface{}) {
+func fatalf(t testing.TB, fatalOnce *sync.Once, dirs []string, msg string, args ...interface{}) {
 	fatalOnce.Do(func() {
 		if artifactsDir == "" {
 			var err error
 			artifactsDir, err = os.Getwd()
 			require.NoError(t, err)
 		}
-		// When run with test parallelism, other subtests may still be running
-		// within subdirectories of `dir`. We copy instead of rename so that those
-		// substests don't also fail when we remove their files out from under them.
-		// Those additional failures would confuse the test output.
-		dst := filepath.Join(artifactsDir, filepath.Base(dir))
-		t.Logf("Copying test dir %q to %q.", dir, dst)
-		_, err := vfs.Clone(vfs.Default, vfs.Default, dir, dst, vfs.CloneTryLink)
-		if err != nil {
-			t.Error(err)
+		// De-duplicate the directories.
+		slices.Sort(dirs)
+		dirs = slices.Compact(dirs)
+
+		for _, dir := range dirs {
+			// When run with test parallelism, other subtests may still be
+			// running within subdirectories of `dir`. We copy instead of rename
+			// so that those substests don't also fail when we remove their
+			// files out from under them.  Those additional failures would
+			// confuse the test output.
+			dst := filepath.Join(artifactsDir, filepath.Base(dir))
+			t.Logf("Copying test dir %q to %q.", dir, dst)
+			_, err := vfs.Clone(vfs.Default, vfs.Default, dir, dst, vfs.CloneTryLink)
+			if err != nil {
+				t.Error(err)
+			}
+			t.Fatalf(msg, args...)
 		}
-		t.Fatalf(msg, args...)
 	})
 }
 
diff --git a/record/record_test.go b/record/record_test.go
index d31f31de..d8b3b71b 100644
--- a/record/record_test.go
+++ b/record/record_test.go
@@ -15,7 +15,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
@@ -356,7 +355,7 @@ func corruptBlock(buf []byte, blockNum int) {
 	buf[blockSize*blockNum+3] = 0x00
 }
 
-func TestRecoverNoOp(t *testing.T) {
+func TestBasicReads(t *testing.T) {
 	recs, err := makeTestRecords(
 		blockSize-legacyHeaderSize,
 		blockSize-legacyHeaderSize,
@@ -366,40 +365,9 @@ func TestRecoverNoOp(t *testing.T) {
 		t.Fatalf("makeTestRecords: %v", err)
 	}
 
-	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
-	_, err = r.Next()
-	if err != nil || r.err != nil {
-		t.Fatalf("reader.Next: %v reader.err: %v", err, r.err)
-	}
-
-	seq, begin, end, n := r.seq, r.begin, r.end, r.n
-
-	// Should be a no-op since r.err == nil.
-	r.recover()
-
-	// r.err was nil, nothing should have changed.
-	if seq != r.seq || begin != r.begin || end != r.end || n != r.n {
-		t.Fatal("reader.Recover when no error existed, was not a no-op")
-	}
-}
-
-func TestBasicRecover(t *testing.T) {
-	recs, err := makeTestRecords(
-		blockSize-legacyHeaderSize,
-		blockSize-legacyHeaderSize,
-		blockSize-legacyHeaderSize,
-	)
-	if err != nil {
-		t.Fatalf("makeTestRecords: %v", err)
-	}
-
-	// Corrupt the checksum of the second record r1 in our file.
-	corruptBlock(recs.buf, 1)
-
 	underlyingReader := bytes.NewReader(recs.buf)
 	r := NewReader(underlyingReader, 0 /* logNum */)
 
-	// The first record r0 should be read just fine.
 	r0, err := r.Next()
 	if err != nil {
 		t.Fatalf("Next: %v", err)
@@ -411,27 +379,17 @@ func TestBasicRecover(t *testing.T) {
 	if !bytes.Equal(r0Data, recs.records[0]) {
 		t.Fatal("Unexpected output in r0's data")
 	}
-
-	// The next record should have a checksum mismatch.
-	_, err = r.Next()
-	if err == nil {
-		t.Fatal("Expected an error while reading a corrupted record")
-	}
-	if err != ErrInvalidChunk {
-		t.Fatalf("Unexpected error returned: %v", err)
+	r1, err := r.Next()
+	if err != nil {
+		t.Fatalf("Next: %v", err)
 	}
-
-	// Recover from that checksum mismatch.
-	r.recover()
-	currentOffset, err := underlyingReader.Seek(0, io.SeekCurrent)
+	r1Data, err := io.ReadAll(r1)
 	if err != nil {
-		t.Fatalf("current offset: %v", err)
+		t.Fatalf("ReadAll: %v", err)
 	}
-	if currentOffset != blockSize*2 {
-		t.Fatalf("current offset: got %d, want %d", currentOffset, blockSize*2)
+	if !bytes.Equal(r1Data, recs.records[1]) {
+		t.Fatal("Unexpected output in r1's data")
 	}
-
-	// The third record r2 should be read just fine.
 	r2, err := r.Next()
 	if err != nil {
 		t.Fatalf("Next: %v", err)
@@ -445,13 +403,12 @@ func TestBasicRecover(t *testing.T) {
 	}
 }
 
-func TestRecoverSingleBlock(t *testing.T) {
+func TestCorruptBlock(t *testing.T) {
 	// The first record will be blockSize * 3 bytes long. Since each block has
 	// a 7 byte header, the first record will roll over into 4 blocks.
 	recs, err := makeTestRecords(
 		blockSize*3,
 		blockSize-legacyHeaderSize,
-		blockSize/2,
 	)
 	if err != nil {
 		t.Fatalf("makeTestRecords: %v", err)
@@ -477,152 +434,6 @@ func TestRecoverSingleBlock(t *testing.T) {
 	if err != ErrInvalidChunk {
 		t.Fatalf("Unexpected error returned: %v", err)
 	}
-
-	// Recover from that checksum mismatch.
-	r.recover()
-
-	// All of the data in the second record r1 is lost because the first record
-	// r0 shared a partial block with it. The second record also overlapped
-	// into the block with the third record r2. Recovery should jump to that
-	// block, skipping over the end of the second record and start parsing the
-	// third record.
-	r2, err := r.Next()
-	if err != nil {
-		t.Fatalf("Next: %v", err)
-	}
-	r2Data, _ := io.ReadAll(r2)
-	if !bytes.Equal(r2Data, recs.records[2]) {
-		t.Fatal("Unexpected output in r2's data")
-	}
-}
-
-func TestRecoverMultipleBlocks(t *testing.T) {
-	recs, err := makeTestRecords(
-		// The first record will consume 3 entire blocks but a fraction of the 4th.
-		blockSize*3,
-		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
-		// Consume the entirety of the 5th block.
-		blockSize-legacyHeaderSize,
-		// Consume the entirety of the 6th block.
-		blockSize-legacyHeaderSize,
-		// Consume roughly half of the 7th block.
-		blockSize/2,
-	)
-	if err != nil {
-		t.Fatalf("makeTestRecords: %v", err)
-	}
-
-	// Corrupt the checksum for the portion of the first record that exists in the 4th block.
-	corruptBlock(recs.buf, 3)
-
-	// Now corrupt the two blocks in a row that correspond to recs.records[2:4].
-	corruptBlock(recs.buf, 4)
-	corruptBlock(recs.buf, 5)
-
-	// The first record should fail, but only when we read deeper beyond the first block.
-	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
-	r0, err := r.Next()
-	if err != nil {
-		t.Fatalf("Next: %v", err)
-	}
-
-	// Reading deeper should yield a checksum mismatch.
-	_, err = io.ReadAll(r0)
-	if err == nil {
-		t.Fatal("Exptected a checksum mismatch error, got nil")
-	}
-	if err != ErrInvalidChunk {
-		t.Fatalf("Unexpected error returned: %v", err)
-	}
-
-	// Recover from that checksum mismatch.
-	r.recover()
-
-	// All of the data in the second record is lost because the first
-	// record shared a partial block with it. The following two records
-	// have corrupted checksums as well, so the call above to r.Recover
-	// should result in r.Next() being a reader to the 5th record.
-	r4, err := r.Next()
-	if err != nil {
-		t.Fatalf("Next: %v", err)
-	}
-
-	r4Data, _ := io.ReadAll(r4)
-	if !bytes.Equal(r4Data, recs.records[4]) {
-		t.Fatal("Unexpected output in r4's data")
-	}
-}
-
-// verifyLastBlockRecover reads each record from recs expecting that the
-// last record will be corrupted. It will then try Recover and verify that EOF
-// is returned.
-func verifyLastBlockRecover(recs *testRecords) error {
-	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
-	// Loop to one element larger than the number of records to verify EOF.
-	for i := 0; i < len(recs.records)+1; i++ {
-		_, err := r.Next()
-		switch i {
-		case len(recs.records) - 1:
-			if err == nil {
-				return errors.New("Expected a checksum mismatch error, got nil")
-			}
-			r.recover()
-		case len(recs.records):
-			if err != io.EOF {
-				return errors.Errorf("Expected io.EOF, got %v", err)
-			}
-		default:
-			if err != nil {
-				return errors.Errorf("Next: %v", err)
-			}
-		}
-	}
-	return nil
-}
-
-func TestRecoverLastPartialBlock(t *testing.T) {
-	recs, err := makeTestRecords(
-		// The first record will consume 3 entire blocks but a fraction of the 4th.
-		blockSize*3,
-		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
-		// Consume roughly half of the 5th block.
-		blockSize/2,
-	)
-	if err != nil {
-		t.Fatalf("makeTestRecords: %v", err)
-	}
-
-	// Corrupt the 5th block.
-	corruptBlock(recs.buf, 4)
-
-	// Verify Recover works when the last block is corrupted.
-	if err := verifyLastBlockRecover(recs); err != nil {
-		t.Fatalf("verifyLastBlockRecover: %v", err)
-	}
-}
-
-func TestRecoverLastCompleteBlock(t *testing.T) {
-	recs, err := makeTestRecords(
-		// The first record will consume 3 entire blocks but a fraction of the 4th.
-		blockSize*3,
-		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
-		// Consume the entire 5th block.
-		blockSize-legacyHeaderSize,
-	)
-	if err != nil {
-		t.Fatalf("makeTestRecords: %v", err)
-	}
-
-	// Corrupt the 5th block.
-	corruptBlock(recs.buf, 4)
-
-	// Verify Recover works when the last block is corrupted.
-	if err := verifyLastBlockRecover(recs); err != nil {
-		t.Fatalf("verifyLastBlockRecover: %v", err)
-	}
 }
 
 func TestReaderOffset(t *testing.T) {
@@ -656,7 +467,7 @@ func TestReaderOffset(t *testing.T) {
 	}
 }
 
-func TestSeekRecord(t *testing.T) {
+func TestSeekRecordInvalidChunk(t *testing.T) {
 	recs, err := makeTestRecords(
 		// The first record will consume 3 entire blocks but a fraction of the 4th.
 		blockSize*3,
@@ -674,8 +485,8 @@ func TestSeekRecord(t *testing.T) {
 	}
 
 	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
-	// Seek to a valid block offset, but within a multiblock record. This should cause the next call to
-	// Next after SeekRecord to return the next valid FIRST/FULL chunk of the subsequent record.
+
+	// Seek to the FIRST/FULL chunk of the second block in the first record.
 	err = r.seekRecord(blockSize)
 	if err != nil {
 		t.Fatalf("SeekRecord: %v", err)
@@ -698,7 +509,39 @@ func TestSeekRecord(t *testing.T) {
 	if _, err = r.Next(); err == nil {
 		t.Fatalf("Expected an error seeking to an invalid chunk boundary")
 	}
-	r.recover()
+}
+
+func TestSeekRecordEndOfFile(t *testing.T) {
+	recs, err := makeTestRecords(
+		// The first record will consume 3 entire blocks but a fraction of the 4th.
+		blockSize*3,
+		// The second record will completely fill the remainder of the 4th block.
+		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
+		// Consume the entirety of the 5th block.
+		blockSize-legacyHeaderSize,
+		// Consume the entirety of the 6th block.
+		blockSize-legacyHeaderSize,
+		// Consume roughly half of the 7th block.
+		blockSize/2,
+	)
+	if err != nil {
+		t.Fatalf("makeTestRecords: %v", err)
+	}
+
+	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
+	// Seek to the FIRST/FULL chunk of the second block in the first record.
+	err = r.seekRecord(blockSize)
+	if err != nil {
+		t.Fatalf("SeekRecord: %v", err)
+	}
+	rec, err := r.Next()
+	if err != nil {
+		t.Fatalf("Next: %v", err)
+	}
+	rData, _ := io.ReadAll(rec)
+	if !bytes.Equal(rData, recs.records[1]) {
+		t.Fatalf("Unexpected output in record 1's data, got %v want %v", rData, recs.records[1])
+	}
 
 	// Seek to the fifth block and verify all records can be read as appropriate.
 	err = r.seekRecord(blockSize * 4)
@@ -736,14 +579,6 @@ func TestSeekRecord(t *testing.T) {
 	if err != io.ErrUnexpectedEOF {
 		t.Fatalf("Seeking past EOF raised unexpected error: %v", err)
 	}
-	r.recover() // Verify recovery works.
-
-	// Validate the current records are returned after seeking to a valid offset.
-	err = r.seekRecord(blockSize * 4)
-	if err != nil {
-		t.Fatalf("SeekRecord: %v", err)
-	}
-	check(2)
 }
 
 func TestLastRecordOffset(t *testing.T) {
diff --git a/sstable/copier_test.go b/sstable/copier_test.go
index 86b2d42f..5c6ccd36 100644
--- a/sstable/copier_test.go
+++ b/sstable/copier_test.go
@@ -130,8 +130,8 @@ func TestCopySpan(t *testing.T) {
 			}
 			defer iter.Close()
 			var result strings.Builder
-			for key := iter.First(); key != nil; key = iter.Next() {
-				fmt.Fprintf(&result, "%s: %s\n", key.K, key.V.InPlaceValue())
+			for kv := iter.First(); kv != nil; kv = iter.Next() {
+				fmt.Fprintf(&result, "%s: %s\n", kv.K, kv.InPlaceValue())
 			}
 			return result.String()
 
diff --git a/sstable/rowblk/rowblk_iter_test.go b/sstable/rowblk/rowblk_iter_test.go
index c76639ae..31ab1330 100644
--- a/sstable/rowblk/rowblk_iter_test.go
+++ b/sstable/rowblk/rowblk_iter_test.go
@@ -501,7 +501,7 @@ func TestSingularKVBlockRestartsOverflow(t *testing.T) {
 	// Ensure that SeekGE() finds the correct KV
 	require.NotNil(t, kv, "failed to find the key")
 	require.Equal(t, largeKey, kv.K.UserKey, "unexpected key")
-	require.Equal(t, largeValue, kv.V.ValueOrHandle, "unexpected value")
+	require.Equal(t, largeValue, kv.InPlaceValue(), "unexpected value")
 
 	// Ensure that SeekGE() does not raise panic due to integer overflow
 	// indexing problems.
@@ -510,7 +510,7 @@ func TestSingularKVBlockRestartsOverflow(t *testing.T) {
 	// Ensure that SeekLT() finds the correct KV
 	require.NotNil(t, kv, "failed to find the key")
 	require.Equal(t, largeKey, kv.K.UserKey, "unexpected key")
-	require.Equal(t, largeValue, kv.V.ValueOrHandle, "unexpected value")
+	require.Equal(t, largeValue, kv.InPlaceValue(), "unexpected value")
 }
 
 func TestBufferExceeding256MBShouldPanic(t *testing.T) {
@@ -625,7 +625,7 @@ func TestMultipleKVBlockRestartsOverflow(t *testing.T) {
 		kv := iter.SeekGE(key, base.SeekGEFlagsNone)
 		require.NotNil(t, kv, "failed to find the large key")
 		require.Equal(t, key, kv.K.UserKey, "unexpected key")
-		require.Equal(t, value, kv.V.ValueOrHandle, "unexpected value")
+		require.Equal(t, value, kv.InPlaceValue(), "unexpected value")
 	}
 }