blob: vary max cached readers by read amplification

The blob.ValueFetcher maintains a small cache of open blob file readers. This
allows a scan that's retrieving values to avoid repeatedly re-opening the same
blob file and re-retrieving the index block from the block cache. This cache
previously used a fixed size of 5 in all contexts. This commit updates the
fetcher to configure the size of the cache dynamically. Iterator construction
now sets a cache size of 5 times the number of non-empty levels in the LSM,
capturing the fact that deeper LSMs will reference more blob files over a span.

Close #5181.

```
                   │   all.txt    │
                   │   ops/sec    │
ycsb/A/values=1024   383.6k ± ∞ ¹
ycsb/B/values=1024   619.7k ± ∞ ¹
ycsb/C/values=1024   1.005M ± ∞ ¹
ycsb/D/values=1024   232.8k ± ∞ ¹
ycsb/E/values=1024   96.19k ± ∞ ¹
ycsb/F/values=1024   82.91k ± ∞ ¹
geomean              276.2k
```

Author: jbowens

blob_rewrite_test.go

@@ -410,7 +410,8 @@ func TestBlobRewriteRandomized(t *testing.T) {
 		// Verify that the rewritten blob file contains the correct values, and
 		// that they may still be accessed using the original handles.
 		var valueFetcher blob.ValueFetcher
-		valueFetcher.Init(constantFileMapping(newBlobFileNum), fch, readEnv)
+		valueFetcher.Init(constantFileMapping(newBlobFileNum), fch, readEnv,
+			blob.SuggestedCachedReaders(1))
 		func() {
 			defer func() { _ = valueFetcher.Close() }()
 			for _, valueIndex := range newFile.valueIndices {

compaction.go

@@ -3321,7 +3321,8 @@ func (d *DB) compactAndWrite(
 		),
 	}
 	if c.version != nil {
-		c.iterationState.valueFetcher.Init(&c.version.BlobFiles, d.fileCache, blockReadEnv)
+		c.iterationState.valueFetcher.Init(&c.version.BlobFiles, d.fileCache, blockReadEnv,
+			blob.SuggestedCachedReaders(len(c.inputs)))
 	}
 	iiopts := internalIterOpts{
 		compaction:       true,

compaction_test.go

@@ -1175,7 +1175,8 @@ func TestCompaction(t *testing.T) {
 					}
 				}
 				vf := &blob.ValueFetcher{}
-				vf.Init(&d.mu.versions.currentVersion().BlobFiles, d.fileCache, block.ReadEnv{})
+				vf.Init(&d.mu.versions.currentVersion().BlobFiles, d.fileCache, block.ReadEnv{},
+					blob.SuggestedCachedReaders(d.mu.versions.currentVersion().MaxReadAmp()))
 				defer func() { _ = vf.Close() }()
 				err := validateBlobValueLiveness(inputTables, d.fileCache, block.ReadEnv{}, vf)
 				if err != nil {

db.go

@@ -33,6 +33,7 @@ import (
 	"github.com/cockroachdb/pebble/rangekey"
 	"github.com/cockroachdb/pebble/record"
 	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/atomicfs"
@@ -1217,9 +1218,11 @@ func (i *Iterator) constructPointIter(
 		),
 	}
 	if i.readState != nil {
-		i.blobValueFetcher.Init(&i.readState.current.BlobFiles, i.fc, readEnv)
+		i.blobValueFetcher.Init(&i.readState.current.BlobFiles, i.fc, readEnv,
+			blob.SuggestedCachedReaders(i.readState.current.MaxReadAmp()))
 	} else if i.version != nil {
-		i.blobValueFetcher.Init(&i.version.BlobFiles, i.fc, readEnv)
+		i.blobValueFetcher.Init(&i.version.BlobFiles, i.fc, readEnv,
+			blob.SuggestedCachedReaders(i.version.MaxReadAmp()))
 	}
 	internalOpts := internalIterOpts{
 		readEnv:          sstable.ReadEnv{Block: readEnv},
@@ -1245,20 +1248,13 @@ func (i *Iterator) constructPointIter(
 	var current *manifest.Version
 	if !i.batchOnlyIter {
 		numMergingLevels += len(memtables)
-
 		current = i.version
 		if current == nil {
 			current = i.readState.current
 		}
-		numMergingLevels += len(current.L0SublevelFiles)
-		numLevelIters += len(current.L0SublevelFiles)
-		for level := 1; level < len(current.Levels); level++ {
-			if current.Levels[level].Empty() {
-				continue
-			}
-			numMergingLevels++
-			numLevelIters++
-		}
+		maxReadAmp := current.MaxReadAmp()
+		numMergingLevels += maxReadAmp
+		numLevelIters += maxReadAmp
 	}
 
 	if numMergingLevels > cap(mlevels) {

get.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
+	"github.com/cockroachdb/pebble/sstable/blob"
 	"github.com/cockroachdb/pebble/sstable/block"
 )
 
@@ -120,7 +121,8 @@ func (d *DB) getInternal(key []byte, b *Batch, s *Snapshot) ([]byte, io.Closer,
 		keyBuf:    buf.keyBuf,
 	}
 	// Set up a blob value fetcher to use for retrieving values from blob files.
-	i.blobValueFetcher.Init(&readState.current.BlobFiles, d.fileCache, block.NoReadEnv)
+	i.blobValueFetcher.Init(&readState.current.BlobFiles, d.fileCache, block.NoReadEnv,
+		blob.SuggestedCachedReaders(readState.current.MaxReadAmp()))
 	get.iiopts.blobValueFetcher = &i.blobValueFetcher
 
 	if !i.First() {

internal/manifest/version.go

@@ -311,6 +311,19 @@ func (v *Version) unrefFiles() ObsoleteFiles {
 	return obsoleteFiles
 }
 
+// MaxReadAmp returns the number of non-empty sublevels and levels in the
+// Version. This is the maximum number of sstables that a point read must
+// consult.
+func (v *Version) MaxReadAmp() int {
+	readAmp := len(v.L0SublevelFiles)
+	for i := 1; i < len(v.Levels); i++ {
+		if !v.Levels[i].Empty() {
+			readAmp++
+		}
+	}
+	return readAmp
+}
+
 // ObsoleteFiles holds a set of files that are no longer referenced by any
 // referenced Version.
 type ObsoleteFiles struct {

level_checker.go

@@ -554,7 +554,11 @@ func (d *DB) CheckLevels(stats *CheckLevelsStats) error {
 		},
 		fileCache: d.fileCache,
 	}
-	checkConfig.blobValueFetcher.Init(&readState.current.BlobFiles, checkConfig.fileCache, checkConfig.readEnv)
+	checkConfig.blobValueFetcher.Init(
+		&readState.current.BlobFiles,
+		checkConfig.fileCache,
+		checkConfig.readEnv,
+		blob.SuggestedCachedReaders(readState.current.MaxReadAmp()))
 	defer func() { _ = checkConfig.blobValueFetcher.Close() }()
 	return checkLevelsInternal(checkConfig)
 }

scan_internal.go

@@ -195,7 +195,8 @@ func (d *DB) newInternalIter(
 		seqNum:          seqNum,
 		mergingIter:     &buf.merging,
 	}
-	dbi.blobValueFetcher.Init(&vers.BlobFiles, d.fileCache, block.ReadEnv{})
+	dbi.blobValueFetcher.Init(&vers.BlobFiles, d.fileCache, block.ReadEnv{},
+		blob.SuggestedCachedReaders(vers.MaxReadAmp()))
 
 	dbi.opts = *o
 	dbi.opts.logger = d.opts.Logger

sstable/blob/fetcher.go

@@ -6,6 +6,7 @@ package blob
 
 import (
 	"context"
+	"slices"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -15,8 +16,6 @@ import (
 	"github.com/cockroachdb/pebble/sstable/block"
 )
 
-const maxCachedReaders = 5
-
 // A ValueReader is an interface defined over a file that can be used to read
 // value blocks.
 type ValueReader interface {
@@ -47,6 +46,12 @@ type ReaderProvider interface {
 	) (r ValueReader, closeFunc func(), err error)
 }
 
+// SuggestedCachedReaders returns the suggested default number of cached readers
+// to use for ValueFetcher.Init given the number of non-empty levels in the LSM.
+func SuggestedCachedReaders(readAmp int) int {
+	return 5 * max(1, readAmp)
+}
+
 // A ValueFetcher retrieves values stored out-of-band in separate blob files.
 // The ValueFetcher caches accessed file readers to avoid redundant file cache
 // and block cache lookups when performing consecutive value retrievals.
@@ -62,7 +67,7 @@ type ValueFetcher struct {
 	env            block.ReadEnv
 	fetchCount     int
 	bufMangler     invariants.BufMangler
-	readers        [maxCachedReaders]cachedReader
+	readers        []cachedReader
 }
 
 // TODO(jackson): Support setting up a read handle for compaction when relevant.
@@ -71,13 +76,16 @@ type ValueFetcher struct {
 var _ base.ValueFetcher = (*ValueFetcher)(nil)
 
 // Init initializes the ValueFetcher.
-func (r *ValueFetcher) Init(fm base.BlobFileMapping, rp ReaderProvider, env block.ReadEnv) {
+func (r *ValueFetcher) Init(
+	fm base.BlobFileMapping, rp ReaderProvider, env block.ReadEnv, maxCachedReaders int,
+) {
 	r.fileMapping = fm
 	r.readerProvider = rp
 	r.env = env
 	if r.readerProvider == nil {
 		panic("readerProvider is nil")
 	}
+	r.readers = slices.Grow(r.readers[:0], maxCachedReaders)[:maxCachedReaders]
 }
 
 // FetchHandle returns the value, given the handle. FetchHandle must not be
@@ -189,6 +197,7 @@ func (r *ValueFetcher) Close() error {
 	r.env = block.ReadEnv{}
 	r.fetchCount = 0
 	r.bufMangler = invariants.BufMangler{}
+	r.readers = r.readers[:0]
 	return err
 }
 

sstable/blob/fetcher_test.go

@@ -123,8 +123,10 @@ func TestValueFetcher(t *testing.T) {
 		case "new-fetcher":
 			var name string
 			td.ScanArgs(t, "name", &name)
+			maxCachedReaders := 5
+			td.MaybeScanArgs(t, "maxCachedReaders", &maxCachedReaders)
 			fetchers[name] = &ValueFetcher{}
-			fetchers[name].Init(identityFileMapping{}, rp, block.ReadEnv{})
+			fetchers[name].Init(identityFileMapping{}, rp, block.ReadEnv{}, maxCachedReaders)
 			return ""
 		case "fetch":
 			var (
@@ -209,7 +211,7 @@ func TestValueFetcherRetrieveRandomized(t *testing.T) {
 
 	t.Run("sequential", func(t *testing.T) {
 		var fetcher ValueFetcher
-		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{})
+		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{}, 5)
 		defer fetcher.Close()
 		for i := 0; i < len(handles); i++ {
 			val, err := fetcher.retrieve(ctx, handles[i])
@@ -219,7 +221,7 @@ func TestValueFetcherRetrieveRandomized(t *testing.T) {
 	})
 	t.Run("random", func(t *testing.T) {
 		var fetcher ValueFetcher
-		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{})
+		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{}, 5)
 		defer fetcher.Close()
 		for _, i := range rng.Perm(len(handles)) {
 			val, err := fetcher.retrieve(ctx, handles[i])
@@ -281,7 +283,7 @@ func benchmarkValueFetcherRetrieve(b *testing.B, valueSize int, cacheSize int64)
 		rp := makeMockReaderProvider(b, obj, cacheSize, handles)
 		defer rp.Close()
 		var fetcher ValueFetcher
-		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{})
+		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{}, 5)
 		defer fetcher.Close()
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
@@ -301,7 +303,7 @@ func benchmarkValueFetcherRetrieve(b *testing.B, valueSize int, cacheSize int64)
 			indices[i] = testutils.RandIntInRange(rng, 0, len(handles))
 		}
 		var fetcher ValueFetcher
-		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{})
+		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{}, 5)
 		defer fetcher.Close()
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
@@ -339,7 +341,7 @@ func makeMockReaderProvider(
 	// blocks.
 	if cacheSize > 0 {
 		var fetcher ValueFetcher
-		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{})
+		fetcher.Init(identityFileMapping{}, rp, block.ReadEnv{}, 5)
 		defer fetcher.Close()
 		for i, h := range handles {
 			if i > 0 && handles[i-1].BlockID == h.BlockID {