[WIP] Update deps

vendor/github.com/cockroachdb/pebble/compaction.go

@@ -2142,7 +2142,6 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 				// calculation even when the WAL is disabled.
 				metrics.BytesIn = metrics.BytesFlushed
 			} else {
-				metrics := c.metrics[0]
 				for i := 0; i < n; i++ {
 					metrics.BytesIn += d.mu.mem.queue[i].logSize
 				}

vendor/github.com/cockroachdb/pebble/db.go

@@ -377,7 +377,7 @@ type DB struct {
 			queue []fileInfo
 			// The number of input bytes to the log. This is the raw size of the
 			// batches written to the WAL, without the overhead of the record
-			// envelopes.
+			// envelopes. Requires DB.mu to be held when read or written.
 			bytesIn uint64
 			// The LogWriter is protected by commitPipeline.mu. This allows log
 			// writes to be performed without holding DB.mu, but requires both
@@ -2471,7 +2471,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 		var prevLogSize uint64
 		if !d.opts.DisableWAL {
 			now := time.Now()
-			newLogNum, prevLogSize = d.recycleWAL()
+			newLogNum, prevLogSize = d.rotateWAL()
 			if b != nil {
 				b.commitStats.WALRotationDuration += time.Since(now)
 			}
@@ -2543,15 +2543,29 @@ func (d *DB) rotateMemtable(newLogNum FileNum, logSeqNum uint64, prev *memTable)
 	var entry *flushableEntry
 	d.mu.mem.mutable, entry = d.newMemTable(newLogNum, logSeqNum)
 	d.mu.mem.queue = append(d.mu.mem.queue, entry)
+	// d.logSize tracks the log size of the WAL file corresponding to the most
+	// recent flushable. The log size of the previous mutable memtable no longer
+	// applies to the current mutable memtable.
+	//
+	// It's tempting to perform this update in rotateWAL, but that would not be
+	// atomic with the enqueue of the new flushable. A call to DB.Metrics()
+	// could acquire DB.mu after the WAL has been rotated but before the new
+	// memtable has been appended; this would result in omitting the log size of
+	// the most recent flushable.
+	d.logSize.Store(0)
 	d.updateReadStateLocked(nil)
 	if prev.writerUnref() {
 		d.maybeScheduleFlush()
 	}
 }
 
+// rotateWAL creates a new write-ahead log, possibly recycling a previous WAL's
+// files. It returns the file number assigned to the new WAL, and the size of
+// the previous WAL file.
+//
 // Both DB.mu and commitPipeline.mu must be held by the caller. Note that DB.mu
 // may be released and reacquired.
-func (d *DB) recycleWAL() (newLogNum FileNum, prevLogSize uint64) {
+func (d *DB) rotateWAL() (newLogNum FileNum, prevLogSize uint64) {
 	if d.opts.DisableWAL {
 		panic("pebble: invalid function call")
 	}

vendor/github.com/cockroachdb/pebble/ingest.go

@@ -1223,7 +1223,12 @@ func (d *DB) handleIngestAsFlushable(meta []*fileMetadata, seqNum uint64) error
 		// We create a new WAL for the flushable instead of reusing the end of
 		// the previous WAL. This simplifies the increment of the minimum
 		// unflushed log number, and also simplifies WAL replay.
-		logNum, _ = d.recycleWAL()
+		var prevLogSize uint64
+		logNum, prevLogSize = d.rotateWAL()
+		// As the rotator of the WAL, we're responsible for updating the
+		// previous flushable queue tail's log size.
+		d.mu.mem.queue[len(d.mu.mem.queue)-1].logSize = prevLogSize
+
 		d.mu.Unlock()
 		err := d.commit.directWrite(b)
 		if err != nil {
@@ -1243,13 +1248,14 @@ func (d *DB) handleIngestAsFlushable(meta []*fileMetadata, seqNum uint64) error
 	// the appropriate value below.
 	newLogNum := d.mu.mem.queue[len(d.mu.mem.queue)-1].logNum
 	if !d.opts.DisableWAL {
-		// This is WAL num of the next mutable memtable which comes after the
-		// ingestedFlushable in the flushable queue. The mutable memtable
-		// will be created below.
-		newLogNum, _ = d.recycleWAL()
-		if err != nil {
-			return err
-		}
+		// newLogNum will be the WAL num of the next mutable memtable which
+		// comes after the ingestedFlushable in the flushable queue. The mutable
+		// memtable will be created below.
+		//
+		// The prevLogSize returned by rotateWAL is the WAL to which the
+		// flushable ingest keys were appended. This intermediary WAL is only
+		// used to record the flushable ingest and nothing else.
+		newLogNum, entry.logSize = d.rotateWAL()
 	}
 
 	currMem := d.mu.mem.mutable

vendor/github.com/cockroachdb/pebble/internal/base/iterator.go

@@ -7,6 +7,9 @@ package base
 import (
 	"fmt"
 	"time"
+
+	"github.com/cockroachdb/pebble/internal/humanize"
+	"github.com/cockroachdb/redact"
 )
 
 // InternalIterator iterates over a DB's key/value pairs in key order. Unlike
@@ -407,3 +410,35 @@ func (s *InternalIteratorStats) Merge(from InternalIteratorStats) {
 	s.SeparatedPointValue.ValueBytes += from.SeparatedPointValue.ValueBytes
 	s.SeparatedPointValue.ValueBytesFetched += from.SeparatedPointValue.ValueBytesFetched
 }
+
+func (s *InternalIteratorStats) String() string {
+	return redact.StringWithoutMarkers(s)
+}
+
+// SafeFormat implements the redact.SafeFormatter interface.
+func (s *InternalIteratorStats) SafeFormat(p redact.SafePrinter, verb rune) {
+	p.Printf("blocks: %s cached",
+		humanize.Bytes.Uint64(s.BlockBytesInCache),
+	)
+	if s.BlockBytes != s.BlockBytesInCache || s.BlockReadDuration != 0 {
+		p.Printf(", %s not cached (read time: %s)",
+			humanize.Bytes.Uint64(s.BlockBytes-s.BlockBytesInCache),
+			humanize.FormattedString(s.BlockReadDuration.String()),
+		)
+	}
+	p.Printf("; points: %s", humanize.Count.Uint64(s.PointCount))
+
+	if s.PointsCoveredByRangeTombstones != 0 {
+		p.Printf("(%s tombstoned)", humanize.Count.Uint64(s.PointsCoveredByRangeTombstones))
+	}
+	p.Printf(" (%s keys, %s values)",
+		humanize.Bytes.Uint64(s.KeyBytes),
+		humanize.Bytes.Uint64(s.ValueBytes),
+	)
+	if s.SeparatedPointValue.Count != 0 {
+		p.Printf("; separated: %s (%s, %s fetched)",
+			humanize.Count.Uint64(s.SeparatedPointValue.Count),
+			humanize.Bytes.Uint64(s.SeparatedPointValue.ValueBytes),
+			humanize.Bytes.Uint64(s.SeparatedPointValue.ValueBytesFetched))
+	}
+}

vendor/github.com/cockroachdb/pebble/iterator.go

@@ -154,6 +154,18 @@ func (s *RangeKeyIteratorStats) Merge(o RangeKeyIteratorStats) {
 	s.SkippedPoints += o.SkippedPoints
 }
 
+func (s *RangeKeyIteratorStats) String() string {
+	return redact.StringWithoutMarkers(s)
+}
+
+// SafeFormat implements the redact.SafeFormatter interface.
+func (s *RangeKeyIteratorStats) SafeFormat(p redact.SafePrinter, verb rune) {
+	p.Printf("range keys: %s, contained points: %s (%s skipped)",
+		humanize.Count.Uint64(uint64(s.Count)),
+		humanize.Count.Uint64(uint64(s.ContainedPoints)),
+		humanize.Count.Uint64(uint64(s.SkippedPoints)))
+}
+
 // LazyValue is a lazy value. See the long comment in base.LazyValue.
 type LazyValue = base.LazyValue
 
@@ -2788,44 +2800,44 @@ func (stats *IteratorStats) String() string {
 
 // SafeFormat implements the redact.SafeFormatter interface.
 func (stats *IteratorStats) SafeFormat(s redact.SafePrinter, verb rune) {
-	for i := range stats.ForwardStepCount {
-		switch IteratorStatsKind(i) {
-		case InterfaceCall:
-			s.SafeString("(interface (dir, seek, step): ")
-		case InternalIterCall:
-			s.SafeString(", (internal (dir, seek, step): ")
-		}
-		s.Printf("(fwd, %d, %d), (rev, %d, %d))",
-			redact.Safe(stats.ForwardSeekCount[i]), redact.Safe(stats.ForwardStepCount[i]),
-			redact.Safe(stats.ReverseSeekCount[i]), redact.Safe(stats.ReverseStepCount[i]))
-	}
-	if stats.InternalStats != (InternalIteratorStats{}) {
-		s.SafeString(",\n(internal-stats: ")
-		s.Printf("(block-bytes: (total %s, cached %s, read-time %s)), "+
-			"(points: (count %s, key-bytes %s, value-bytes %s, tombstoned %s))",
-			humanize.Bytes.Uint64(stats.InternalStats.BlockBytes),
-			humanize.Bytes.Uint64(stats.InternalStats.BlockBytesInCache),
-			humanize.FormattedString(stats.InternalStats.BlockReadDuration.String()),
-			humanize.Count.Uint64(stats.InternalStats.PointCount),
-			humanize.Bytes.Uint64(stats.InternalStats.KeyBytes),
-			humanize.Bytes.Uint64(stats.InternalStats.ValueBytes),
-			humanize.Count.Uint64(stats.InternalStats.PointsCoveredByRangeTombstones),
+	if stats.ReverseSeekCount[InterfaceCall] == 0 && stats.ReverseSeekCount[InternalIterCall] == 0 {
+		s.Printf("seeked %s times (%s internal)",
+			humanize.Count.Uint64(uint64(stats.ForwardSeekCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ForwardSeekCount[InternalIterCall])),
 		)
-		if stats.InternalStats.SeparatedPointValue.Count != 0 {
-			s.Printf(", (separated: (count %s, bytes %s, fetched %s)))",
-				humanize.Count.Uint64(stats.InternalStats.SeparatedPointValue.Count),
-				humanize.Bytes.Uint64(stats.InternalStats.SeparatedPointValue.ValueBytes),
-				humanize.Bytes.Uint64(stats.InternalStats.SeparatedPointValue.ValueBytesFetched))
-		} else {
-			s.Printf(")")
-		}
+	} else {
+		s.Printf("seeked %s times (%s fwd/%s rev, internal: %s fwd/%s rev)",
+			humanize.Count.Uint64(uint64(stats.ForwardSeekCount[InterfaceCall]+stats.ReverseSeekCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ForwardSeekCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ReverseSeekCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ForwardSeekCount[InternalIterCall])),
+			humanize.Count.Uint64(uint64(stats.ReverseSeekCount[InternalIterCall])),
+		)
+	}
+	s.SafeString("; ")
+
+	if stats.ReverseStepCount[InterfaceCall] == 0 && stats.ReverseStepCount[InternalIterCall] == 0 {
+		s.Printf("stepped %s times (%s internal)",
+			humanize.Count.Uint64(uint64(stats.ForwardStepCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ForwardStepCount[InternalIterCall])),
+		)
+	} else {
+		s.Printf("stepped %s times (%s fwd/%s rev, internal: %s fwd/%s rev)",
+			humanize.Count.Uint64(uint64(stats.ForwardStepCount[InterfaceCall]+stats.ReverseStepCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ForwardStepCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ReverseStepCount[InterfaceCall])),
+			humanize.Count.Uint64(uint64(stats.ForwardStepCount[InternalIterCall])),
+			humanize.Count.Uint64(uint64(stats.ReverseStepCount[InternalIterCall])),
+		)
+	}
+
+	if stats.InternalStats != (InternalIteratorStats{}) {
+		s.SafeString("; ")
+		stats.InternalStats.SafeFormat(s, verb)
 	}
 	if stats.RangeKeyStats != (RangeKeyIteratorStats{}) {
-		s.SafeString(",\n(range-key-stats: ")
-		s.Printf("(count %d), (contained points: (count %d, skipped %d)))",
-			stats.RangeKeyStats.Count,
-			stats.RangeKeyStats.ContainedPoints,
-			stats.RangeKeyStats.SkippedPoints)
+		s.SafeString(", ")
+		stats.RangeKeyStats.SafeFormat(s, verb)
 	}
 }
 

vendor/github.com/cockroachdb/pebble/objstorage/objstorageprovider/provider.go

@@ -92,6 +92,17 @@ type Settings struct {
 	// out a large chunk of dirty filesystem buffers.
 	BytesPerSync int
 
+	// Local contains fields that are only relevant for files stored on the local
+	// filesystem.
+	Local struct {
+		// TODO(radu): move FSCleaner, NoSyncOnClose, BytesPerSync here.
+
+		// ReadaheadConfigFn is a function used to retrieve the current readahead
+		// mode. This function is run whenever a local object is open for reading.
+		// If it is nil, DefaultReadaheadConfig is used.
+		ReadaheadConfigFn func() ReadaheadConfig
+	}
+
 	// Fields here are set only if the provider is to support remote objects
 	// (experimental).
 	Remote struct {
@@ -129,6 +140,43 @@ type Settings struct {
 	}
 }
 
+// ReadaheadConfig controls the use of read-ahead.
+type ReadaheadConfig struct {
+	// Informed is the type of read-ahead for operations that are known to read a
+	// large consecutive chunk of a file.
+	Informed ReadaheadMode
+
+	// Speculative is the type of read-ahead used automatically, when consecutive
+	// reads are detected.
+	Speculative ReadaheadMode
+}
+
+// DefaultReadaheadConfig is the readahead config used when ReadaheadConfigFn is
+// not specified.
+var DefaultReadaheadConfig = ReadaheadConfig{
+	Informed:    FadviseSequential,
+	Speculative: FadviseSequential,
+}
+
+// ReadaheadMode indicates the type of read-ahead to use, either for informed
+// read-ahead (e.g. compactions) or speculative read-ahead.
+type ReadaheadMode uint8
+
+const (
+	// NoReadahead disables readahead altogether.
+	NoReadahead ReadaheadMode = iota
+
+	// SysReadahead enables the use of SYS_READAHEAD call to prefetch data.
+	// The prefetch window grows dynamically as consecutive writes are detected.
+	SysReadahead
+
+	// FadviseSequential enables to use of FADV_SEQUENTIAL. For informed
+	// read-ahead, FADV_SEQUENTIAL is used from the beginning. For speculative
+	// read-ahead SYS_READAHEAD is first used until the window reaches the maximum
+	// size, then we siwtch to FADV_SEQUENTIAL.
+	FadviseSequential
+)
+
 // DefaultSettings initializes default settings (with no remote storage),
 // suitable for tests and tools.
 func DefaultSettings(fs vfs.FS, dirName string) Settings {

vendor/github.com/cockroachdb/pebble/objstorage/objstorageprovider/readahead.go

@@ -4,6 +4,8 @@
 
 package objstorageprovider
 
+import "github.com/cockroachdb/pebble/internal/invariants"
+
 const (
 	// Constants for dynamic readahead of data blocks. Note that the size values
 	// make sense as some multiple of the default block size; and they should
@@ -81,6 +83,9 @@ func (rs *readaheadState) recordCacheHit(offset, blockLength int64) {
 // Returns a size value (greater than 0) that should be prefetched if readahead
 // would be beneficial.
 func (rs *readaheadState) maybeReadahead(offset, blockLength int64) int64 {
+	if invariants.Enabled && rs.maxReadaheadSize == 0 {
+		panic("readaheadState not initialized")
+	}
 	currentReadEnd := offset + blockLength
 	if rs.numReads >= minFileReadsForReadahead {
 		// The minimum threshold of sequential reads to justify reading ahead

vendor/github.com/cockroachdb/pebble/objstorage/objstorageprovider/vfs.go

@@ -31,7 +31,11 @@ func (p *provider) vfsOpenForReading(
 		}
 		return nil, err
 	}
-	return newFileReadable(file, p.st.FS, filename)
+	readaheadConfig := DefaultReadaheadConfig
+	if f := p.st.Local.ReadaheadConfigFn; f != nil {
+		readaheadConfig = f()
+	}
+	return newFileReadable(file, p.st.FS, readaheadConfig, filename)
 }
 
 func (p *provider) vfsCreate(

vendor/github.com/cockroachdb/pebble/objstorage/objstorageprovider/vfs_readable.go

@@ -27,22 +27,26 @@ type fileReadable struct {
 
 	// The following fields are used to possibly open the file again using the
 	// sequential reads option (see vfsReadHandle).
-	filename string
-	fs       vfs.FS
+	filename        string
+	fs              vfs.FS
+	readaheadConfig ReadaheadConfig
 }
 
 var _ objstorage.Readable = (*fileReadable)(nil)
 
-func newFileReadable(file vfs.File, fs vfs.FS, filename string) (*fileReadable, error) {
+func newFileReadable(
+	file vfs.File, fs vfs.FS, readaheadConfig ReadaheadConfig, filename string,
+) (*fileReadable, error) {
 	info, err := file.Stat()
 	if err != nil {
 		return nil, err
 	}
 	r := &fileReadable{
-		file:     file,
-		size:     info.Size(),
-		filename: filename,
-		fs:       fs,
+		file:            file,
+		size:            info.Size(),
+		filename:        filename,
+		fs:              fs,
+		readaheadConfig: readaheadConfig,
 	}
 	invariants.SetFinalizer(r, func(obj interface{}) {
 		if obj.(*fileReadable).file != nil {
@@ -76,14 +80,14 @@ func (r *fileReadable) Size() int64 {
 // NewReadHandle is part of the objstorage.Readable interface.
 func (r *fileReadable) NewReadHandle(_ context.Context) objstorage.ReadHandle {
 	rh := readHandlePool.Get().(*vfsReadHandle)
-	rh.r = r
-	rh.rs = makeReadaheadState(fileMaxReadaheadSize)
+	rh.init(r)
 	return rh
 }
 
 type vfsReadHandle struct {
-	r  *fileReadable
-	rs readaheadState
+	r             *fileReadable
+	rs            readaheadState
+	readaheadMode ReadaheadMode
 
 	// sequentialFile holds a file descriptor to the same underlying File,
 	// except with fadvise(FADV_SEQUENTIAL) called on it to take advantage of
@@ -108,6 +112,14 @@ var readHandlePool = sync.Pool{
 	},
 }
 
+func (rh *vfsReadHandle) init(r *fileReadable) {
+	*rh = vfsReadHandle{
+		r:             r,
+		rs:            makeReadaheadState(fileMaxReadaheadSize),
+		readaheadMode: r.readaheadConfig.Speculative,
+	}
+}
+
 // Close is part of the objstorage.ReadHandle interface.
 func (rh *vfsReadHandle) Close() error {
 	var err error
@@ -121,14 +133,17 @@ func (rh *vfsReadHandle) Close() error {
 
 // ReadAt is part of the objstorage.ReadHandle interface.
 func (rh *vfsReadHandle) ReadAt(_ context.Context, p []byte, offset int64) error {
-	var n int
-	var err error
 	if rh.sequentialFile != nil {
 		// Use OS-level read-ahead.
-		n, err = rh.sequentialFile.ReadAt(p, offset)
-	} else {
+		n, err := rh.sequentialFile.ReadAt(p, offset)
+		if invariants.Enabled && err == nil && n != len(p) {
+			panic("short read")
+		}
+		return err
+	}
+	if rh.readaheadMode != NoReadahead {
 		if readaheadSize := rh.rs.maybeReadahead(offset, int64(len(p))); readaheadSize > 0 {
-			if readaheadSize >= fileMaxReadaheadSize {
+			if rh.readaheadMode == FadviseSequential && readaheadSize >= fileMaxReadaheadSize {
 				// We've reached the maximum readahead size. Beyond this point, rely on
 				// OS-level readahead.
 				rh.switchToOSReadahead()
@@ -136,8 +151,8 @@ func (rh *vfsReadHandle) ReadAt(_ context.Context, p []byte, offset int64) error
 				_ = rh.r.file.Prefetch(offset, readaheadSize)
 			}
 		}
-		n, err = rh.r.file.ReadAt(p, offset)
 	}
+	n, err := rh.r.file.ReadAt(p, offset)
 	if invariants.Enabled && err == nil && n != len(p) {
 		panic("short read")
 	}
@@ -146,10 +161,16 @@ func (rh *vfsReadHandle) ReadAt(_ context.Context, p []byte, offset int64) error
 
 // SetupForCompaction is part of the objstorage.ReadHandle interface.
 func (rh *vfsReadHandle) SetupForCompaction() {
-	rh.switchToOSReadahead()
+	rh.readaheadMode = rh.r.readaheadConfig.Informed
+	if rh.readaheadMode == FadviseSequential {
+		rh.switchToOSReadahead()
+	}
 }
 
 func (rh *vfsReadHandle) switchToOSReadahead() {
+	if invariants.Enabled && rh.readaheadMode != FadviseSequential {
+		panic("readheadMode not respected")
+	}
 	if rh.sequentialFile != nil {
 		return
 	}
@@ -164,8 +185,8 @@ func (rh *vfsReadHandle) switchToOSReadahead() {
 
 // RecordCacheHit is part of the objstorage.ReadHandle interface.
 func (rh *vfsReadHandle) RecordCacheHit(_ context.Context, offset, size int64) {
-	if rh.sequentialFile != nil {
-		// Using OS-level readahead, so do nothing.
+	if rh.sequentialFile != nil || rh.readaheadMode == NoReadahead {
+		// Using OS-level or no readahead, so do nothing.
 		return
 	}
 	rh.rs.recordCacheHit(offset, size)
@@ -208,8 +229,7 @@ func UsePreallocatedReadHandle(
 	ctx context.Context, readable objstorage.Readable, rh *PreallocatedReadHandle,
 ) objstorage.ReadHandle {
 	if r, ok := readable.(*fileReadable); ok {
-		// See fileReadable.NewReadHandle.
-		rh.vfsReadHandle = vfsReadHandle{r: r}
+		rh.init(r)
 		return rh
 	}
 	return readable.NewReadHandle(ctx)

vendor/github.com/cockroachdb/pebble/open.go

@@ -310,6 +310,7 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 		NoSyncOnClose:       opts.NoSyncOnClose,
 		BytesPerSync:        opts.BytesPerSync,
 	}
+	providerSettings.Local.ReadaheadConfigFn = opts.Local.ReadaheadConfigFn
 	providerSettings.Remote.StorageFactory = opts.Experimental.RemoteStorage
 	providerSettings.Remote.CreateOnShared = opts.Experimental.CreateOnShared
 	providerSettings.Remote.CreateOnSharedLocator = opts.Experimental.CreateOnSharedLocator

vendor/github.com/cockroachdb/pebble/options.go

@@ -14,11 +14,13 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/fifo"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/humanize"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/manifest"
+	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/rangekey"
 	"github.com/cockroachdb/pebble/sstable"
@@ -481,11 +483,25 @@ type Options struct {
 	// The default cache size is 8 MB.
 	Cache *cache.Cache
 
+	// LoadBlockSema, if set, is used to limit the number of blocks that can be
+	// loaded (i.e. read from the filesystem) in parallel. Each load acquires one
+	// unit from the semaphore for the duration of the read.
+	LoadBlockSema *fifo.Semaphore
+
 	// Cleaner cleans obsolete files.
 	//
 	// The default cleaner uses the DeleteCleaner.
 	Cleaner Cleaner
 
+	// Local contains option that pertain to files stored on the local filesystem.
+	Local struct {
+		// ReadaheadConfigFn is a function used to retrieve the current readahead
+		// mode. This function is consulted when a table enters the table cache.
+		ReadaheadConfigFn func() ReadaheadConfig
+
+		// TODO(radu): move BytesPerSync, LoadBlockSema, Cleaner here.
+	}
+
 	// Comparer defines a total ordering over the space of []byte keys: a 'less
 	// than' relationship. The same comparison algorithm must be used for reads
 	// and writes over the lifetime of the DB.
@@ -996,6 +1012,9 @@ type Options struct {
 	}
 }
 
+// ReadaheadConfig controls the use of read-ahead.
+type ReadaheadConfig = objstorageprovider.ReadaheadConfig
+
 // DebugCheckLevels calls CheckLevels on the provided database.
 // It may be set in the DebugCheck field of Options to check
 // level invariants whenever a new version is installed.
@@ -1713,6 +1732,7 @@ func (o *Options) MakeReaderOptions() sstable.ReaderOptions {
 	var readerOpts sstable.ReaderOptions
 	if o != nil {
 		readerOpts.Cache = o.Cache
+		readerOpts.LoadBlockSema = o.LoadBlockSema
 		readerOpts.Comparer = o.Comparer
 		readerOpts.Filters = o.Filters
 		if o.Merger != nil {

vendor/github.com/cockroachdb/pebble/sstable/options.go

@@ -5,6 +5,7 @@
 package sstable
 
 import (
+	"github.com/cockroachdb/fifo"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
 )
@@ -111,6 +112,11 @@ type ReaderOptions struct {
 	// The default cache size is a zero-size cache.
 	Cache *cache.Cache
 
+	// LoadBlockSema, if set, is used to limit the number of blocks that can be
+	// loaded (i.e. read from the filesystem) in parallel. Each load acquires one
+	// unit from the semaphore for the duration of the read.
+	LoadBlockSema *fifo.Semaphore
+
 	// User properties specified in this map will not be added to sst.Properties.UserProperties.
 	DeniedUserProperties map[string]struct{}
 

vendor/github.com/cockroachdb/pebble/sstable/reader.go

@@ -539,6 +539,15 @@ func (r *Reader) readBlock(
 	}
 
 	// Cache miss.
+
+	if sema := r.opts.LoadBlockSema; sema != nil {
+		if err := sema.Acquire(ctx, 1); err != nil {
+			// An error here can only come from the context.
+			return bufferHandle{}, err
+		}
+		defer sema.Release(1)
+	}
+
 	var compressed cacheValueOrBuf
 	if bufferPool != nil {
 		compressed = cacheValueOrBuf{
@@ -571,6 +580,7 @@ func (r *Reader) readBlock(
 			int(bh.Length+blockTrailerLen), readDuration.String())
 	}
 	if stats != nil {
+		stats.BlockBytes += bh.Length
 		stats.BlockReadDuration += readDuration
 	}
 	if err != nil {
@@ -628,9 +638,6 @@ func (r *Reader) readBlock(
 		decompressed = transformed
 	}
 
-	if stats != nil {
-		stats.BlockBytes += bh.Length
-	}
 	if decompressed.buf.Valid() {
 		return bufferHandle{b: decompressed.buf}, nil
 	}

vendor/github.com/cockroachdb/pebble/sstable/reader_iter_two_lvl.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
 )
 
@@ -177,7 +178,7 @@ func (i *twoLevelIterator) init(
 		_ = i.topLevelIndex.Close()
 		return err
 	}
-	i.dataRH = r.readable.NewReadHandle(ctx)
+	i.dataRH = objstorageprovider.UsePreallocatedReadHandle(ctx, r.readable, &i.dataRHPrealloc)
 	if r.tableFormat >= TableFormatPebblev3 {
 		if r.Properties.NumValueBlocks > 0 {
 			i.vbReader = &valueBlockReader{

vendor/github.com/cockroachdb/pebble/vfs/logging_fs.go

@@ -5,6 +5,7 @@
 package vfs
 
 import (
+	"fmt"
 	"io"
 	"os"
 )
@@ -38,7 +39,15 @@ func (fs *loggingFS) Create(name string) (File, error) {
 }
 
 func (fs *loggingFS) Open(name string, opts ...OpenOption) (File, error) {
-	fs.logFn("open: %s", name)
+	var optsStr string
+	if len(opts) > 0 {
+		optsStr = " (options:"
+		for i := range opts {
+			optsStr += fmt.Sprintf(" %T", opts[i])
+		}
+		optsStr += ")"
+	}
+	fs.logFn("open: %s%s", name, optsStr)
 	f, err := fs.FS.Open(name, opts...)
 	if err != nil {
 		return nil, err