storage: batch pebbleResults allocation

Embed a `pebbleResults` struct on the `pebbleMVCCScanner` to avoid an
allocation.

```
                                                                     │  23.1.txt   │           23.1-wip.txt            │
                                                                     │   sec/op    │   sec/op     vs base              │
MVCCGet_Pebble/batch=true/versions=100/valueSize=8/numRangeKeys=0-24   109.3µ ± 1%   105.2µ ± 3%  -3.73% (p=0.009 n=6)

                                                                     │   23.1.txt   │          23.1-wip.txt          │
                                                                     │     B/s      │      B/s       vs base         │
MVCCGet_Pebble/batch=true/versions=100/valueSize=8/numRangeKeys=0-24   68.36Ki ± 0%   78.12Ki ± 12%  ~ (p=0.061 n=6)

                                                                     │  23.1.txt   │            23.1-wip.txt            │
                                                                     │    B/op     │    B/op      vs base               │
MVCCGet_Pebble/batch=true/versions=100/valueSize=8/numRangeKeys=0-24   354.0 ± 22%   228.5 ± 28%  -35.45% (p=0.002 n=6)

                                                                     │  23.1.txt  │           23.1-wip.txt            │
                                                                     │ allocs/op  │ allocs/op   vs base               │
MVCCGet_Pebble/batch=true/versions=100/valueSize=8/numRangeKeys=0-24   3.000 ± 0%   2.000 ± 0%  -33.33% (p=0.002 n=6)
```

Informs cockroachdb#96960.
Epic: None
Release note: None

pkg/storage/col_mvcc.go

@@ -421,10 +421,11 @@ func mvccScanToCols(
 	opts MVCCScanOptions,
 	st *cluster.Settings,
 ) (MVCCScanResult, error) {
+	mvccScanner := pebbleMVCCScannerPool.Get().(*pebbleMVCCScanner)
 	adapter := mvccScanFetchAdapter{machine: onNextKVSeek}
 	adapter.results.maxKeysPerRow = indexFetchSpec.MaxKeysPerRow
 	adapter.results.maxFamilyID = uint32(indexFetchSpec.MaxFamilyID)
-	ok, mvccScanner, res, err := mvccScanInit(iter, key, endKey, timestamp, opts, &adapter.results)
+	ok, res, err := mvccScanInit(mvccScanner, iter, key, endKey, timestamp, opts, &adapter.results)
 	if !ok {
 		return res, err
 	}

pkg/storage/mvcc.go

@@ -1156,8 +1156,9 @@ func mvccGetWithValueHeader(
 		keyBuf:           mvccScanner.keyBuf,
 	}
 
-	var results pebbleResults
-	mvccScanner.init(opts.Txn, opts.Uncertainty, &results)
+	results := &mvccScanner.alloc.pebbleResults
+	*results = pebbleResults{}
+	mvccScanner.init(opts.Txn, opts.Uncertainty, results)
 	mvccScanner.get(ctx)
 
 	// If we're tracking the ScanStats, include the stats from this Get.
@@ -3665,33 +3666,32 @@ func recordIteratorStats(iter MVCCIterator, scanStats *kvpb.ScanStats) {
 // If ok=false is returned, then the returned result and the error are the
 // result of the scan.
 func mvccScanInit(
+	mvccScanner *pebbleMVCCScanner,
 	iter MVCCIterator,
 	key, endKey roachpb.Key,
 	timestamp hlc.Timestamp,
 	opts MVCCScanOptions,
 	results results,
-) (ok bool, _ *pebbleMVCCScanner, _ MVCCScanResult, _ error) {
+) (ok bool, _ MVCCScanResult, _ error) {
 	if len(endKey) == 0 {
-		return false, nil, MVCCScanResult{}, emptyKeyError()
+		return false, MVCCScanResult{}, emptyKeyError()
 	}
 	if err := opts.validate(); err != nil {
-		return false, nil, MVCCScanResult{}, err
+		return false, MVCCScanResult{}, err
 	}
 	if opts.MaxKeys < 0 {
-		return false, nil, MVCCScanResult{
+		return false, MVCCScanResult{
 			ResumeSpan:   &roachpb.Span{Key: key, EndKey: endKey},
 			ResumeReason: kvpb.RESUME_KEY_LIMIT,
 		}, nil
 	}
 	if opts.TargetBytes < 0 {
-		return false, nil, MVCCScanResult{
+		return false, MVCCScanResult{
 			ResumeSpan:   &roachpb.Span{Key: key, EndKey: endKey},
 			ResumeReason: kvpb.RESUME_BYTE_LIMIT,
 		}, nil
 	}
 
-	mvccScanner := pebbleMVCCScannerPool.Get().(*pebbleMVCCScanner)
-
 	*mvccScanner = pebbleMVCCScanner{
 		parent:           iter,
 		memAccount:       opts.MemoryAccount,
@@ -3710,10 +3710,16 @@ func mvccScanInit(
 		tombstones:       opts.Tombstones,
 		failOnMoreRecent: opts.FailOnMoreRecent,
 		keyBuf:           mvccScanner.keyBuf,
+		// NB: If the `results` argument passed to this function is a pointer to
+		// mvccScanner.alloc.pebbleResults, we don't want to overwrite any
+		// initialization of the pebbleResults struct performed by the caller.
+		// The struct should not contain any stale buffers from previous uses,
+		// because pebbleMVCCScanner.release zeros it.
+		alloc: mvccScanner.alloc,
 	}
 
 	mvccScanner.init(opts.Txn, opts.Uncertainty, results)
-	return true /* ok */, mvccScanner, MVCCScanResult{}, nil
+	return true /* ok */, MVCCScanResult{}, nil
 }
 
 func mvccScanToBytes(
@@ -3723,12 +3729,14 @@ func mvccScanToBytes(
 	timestamp hlc.Timestamp,
 	opts MVCCScanOptions,
 ) (MVCCScanResult, error) {
-	var results pebbleResults
+	mvccScanner := pebbleMVCCScannerPool.Get().(*pebbleMVCCScanner)
+	results := &mvccScanner.alloc.pebbleResults
+	*results = pebbleResults{}
 	if opts.WholeRowsOfSize > 1 {
 		results.lastOffsetsEnabled = true
 		results.lastOffsets = make([]int, opts.WholeRowsOfSize)
 	}
-	ok, mvccScanner, res, err := mvccScanInit(iter, key, endKey, timestamp, opts, &results)
+	ok, res, err := mvccScanInit(mvccScanner, iter, key, endKey, timestamp, opts, results)
 	if !ok {
 		return res, err
 	}

pkg/storage/pebble_mvcc_scanner.go

@@ -472,6 +472,15 @@ type pebbleMVCCScanner struct {
 		// advancing the iterator at the new key. It is backed by keyBuf.
 		origKey []byte
 	}
+	// alloc holds fields embedded within the scanner struct only to reduce
+	// allocations in common cases.
+	alloc struct {
+		// Typically pebbleMVCCScanner.results points to pebbleResults.
+		// Embedding the pebbleResults within the pebbleMVCCScanner avoids an
+		// extra allocation, at the cost of higher allocated bytes when we use a
+		// different implementation of the results interface.
+		pebbleResults pebbleResults
+	}
 }
 
 type advanceFn int
@@ -515,6 +524,9 @@ func (p *pebbleMVCCScanner) release() {
 	// Discard most memory references before placing in pool.
 	*p = pebbleMVCCScanner{
 		keyBuf: p.keyBuf,
+		// NB: This clears p.alloc.pebbleResults too, which should be maintained
+		// to avoid delaying GC of contained byte slices and avoid accidental
+		// misuse.
 	}
 	pebbleMVCCScannerPool.Put(p)
 }