Repository: cockroachdb/pebble
Pull request #2251 (2023-01-17 22:00:53)

replay: various compaction benchmarking tweaks
**cmd/pebble: add -count to replay command**

Add a -count flag to the replay command to replay a workload multiple times,
useful for building up a statistically significant number of runs.

**replay: refactor compaction tracking**

Refactor the compaction tracking for readability and eliminating an extra
goroutine. Also, add a unit test that ensures the replayer correctly
detects when compactions quiesce.

**cmd/pebble: remove temporary run directories before exit**

**replay: fix TotalWAmp calculation**

Fix TotalWAmp calculation to use Metrics.Total and maintain consistency with
the w-amp reported by Metrics.String(). The Metrics.Total method handles
adjusting BytesIn and BytesFlushed.

**cmd/pebble: allow setting custom Pebble Options through a CLI flag**

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


Patch:
diff --git a/batch.go b/batch.go
index b00287a7..3f02a09b 100644
--- a/batch.go
+++ b/batch.go
@@ -162,6 +162,7 @@ func (d DeferredBatchOp) Finish() error {
 //
 //	InternalKeyKindDelete         varstring
 //	InternalKeyKindLogData        varstring
+//	InternalKeyKindIngestSST      varstring
 //	InternalKeyKindSet            varstring varstring
 //	InternalKeyKindMerge          varstring varstring
 //	InternalKeyKindRangeDelete    varstring varstring
@@ -210,6 +211,8 @@ type Batch struct {
 
 	// The db to which the batch will be committed. Do not change this field
 	// after the batch has been created as it might invalidate internal state.
+	// Batch.memTableSize is only refreshed if Batch.db is set. Setting db to
+	// nil once it has been set implies that the Batch has encountered an error.
 	db *DB
 
 	// The count of records in the batch. This count will be stored in the batch
@@ -255,6 +258,11 @@ type Batch struct {
 	// memtable.
 	flushable *flushableBatch
 
+	// ingestedSSTBatch indicates that the batch contains one or more key kinds
+	// of InternalKeyKindIngestSST. If the batch contains key kinds of IngestSST
+	// then it will only contain key kinds of IngestSST.
+	ingestedSSTBatch bool
+
 	// Synchronous Apply uses the commit WaitGroup for both publishing the
 	// seqnum and waiting for the WAL fsync (if needed). Asynchronous
 	// ApplyNoSyncWait, which implies WriteOptions.Sync is true, uses the commit
@@ -363,13 +371,16 @@ func (b *Batch) refreshMemTableSize() {
 		if !ok {
 			break
 		}
-		b.memTableSize += memTableEntrySize(len(key), len(value))
 		switch kind {
 		case InternalKeyKindRangeDelete:
 			b.countRangeDels++
 		case InternalKeyKindRangeKeySet, InternalKeyKindRangeKeyUnset, InternalKeyKindRangeKeyDelete:
 			b.countRangeKeys++
+		case InternalKeyKindIngestSST:
+			// This key kind doesn't contribute to the memtable size.
+			continue
 		}
+		b.memTableSize += memTableEntrySize(len(key), len(value))
 	}
 }
 
@@ -377,6 +388,9 @@ func (b *Batch) refreshMemTableSize() {
 //
 // It is safe to modify the contents of the arguments after Apply returns.
 func (b *Batch) Apply(batch *Batch, _ *WriteOptions) error {
+	if b.ingestedSSTBatch {
+		panic("pebble: invalid batch application")
+	}
 	if len(batch.data) == 0 {
 		return nil
 	}
@@ -407,6 +421,8 @@ func (b *Batch) Apply(batch *Batch, _ *WriteOptions) error {
 				b.countRangeDels++
 			case InternalKeyKindRangeKeySet, InternalKeyKindRangeKeyUnset, InternalKeyKindRangeKeyDelete:
 				b.countRangeKeys++
+			case InternalKeyKindIngestSST:
+				panic("pebble: invalid key kind for batch")
 			}
 			if b.index != nil {
 				var err error
@@ -812,6 +828,26 @@ func (b *Batch) LogData(data []byte, _ *WriteOptions) error {
 	return nil
 }
 
+// IngestSST adds an sstable path to the batch. The data will only be written to
+// the WAL (not added to memtables or sstables).
+func (b *Batch) ingestSST(data []byte) {
+	if b.Empty() {
+		b.ingestedSSTBatch = true
+	} else if !b.ingestedSSTBatch {
+		// Batch contains other key kinds.
+		panic("pebble: invalid call to ingestSST")
+	}
+
+	origMemTableSize := b.memTableSize
+	b.prepareDeferredKeyRecord(len(data), InternalKeyKindIngestSST)
+	copy(b.deferredOp.Key, data)
+	// Since IngestSST writes only to the WAL and does not affect the memtable,
+	// we restore b.memTableSize to its original value. Note that Batch.count
+	// is not reset because for the InternalKeyKindIngestSST the count is the
+	// number of sstable paths which have been added to the batch.
+	b.memTableSize = origMemTableSize
+}
+
 // Empty returns true if the batch is empty, and false otherwise.
 func (b *Batch) Empty() bool {
 	return len(b.data) <= batchHeaderLen
@@ -1175,7 +1211,9 @@ func (b *Batch) setCount(v uint32) {
 }
 
 // Count returns the count of memtable-modifying operations in this batch. All
-// operations with the except of LogData increment this count.
+// operations with the except of LogData increment this count. For IngestSSTs,
+// count is only used to indicate the number of SSTs ingested in the record, the
+// batch isn't applied to the memtable.
 func (b *Batch) Count() uint32 {
 	if b.count > math.MaxUint32 {
 		panic(ErrInvalidBatch)
diff --git a/cmd/pebble/replay.go b/cmd/pebble/replay.go
index 1b84f3c1..58bbf3dc 100644
--- a/cmd/pebble/replay.go
+++ b/cmd/pebble/replay.go
@@ -5,6 +5,7 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"flag"
 	"fmt"
@@ -14,6 +15,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"unicode"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
@@ -25,8 +27,9 @@ import (
 
 func initReplayCmd() *cobra.Command {
 	c := replayConfig{
-		pacer:            pacerFlag{Pacer: replay.Unpaced{}},
+		pacer:            pacerFlag{Pacer: replay.PaceByFixedReadAmp(10)},
 		runDir:           "",
+		count:            1,
 		streamLogs:       false,
 		ignoreCheckpoint: false,
 	}
@@ -36,10 +39,14 @@ func initReplayCmd() *cobra.Command {
 		Args:  cobra.ExactArgs(1),
 		RunE:  c.runE,
 	}
+	cmd.Flags().IntVar(
+		&c.count, "count", 1, "the number of times to replay the workload")
 	cmd.Flags().StringVar(
 		&c.name, "name", "", "the name of the workload being replayed")
 	cmd.Flags().VarPF(
 		&c.pacer, "pacer", "p", "the pacer to use: unpaced, reference-ramp, or fixed-ramp=N")
+	cmd.Flags().StringVar(
+		&c.optionsString, "options", "", "Pebble options to override, in the OPTIONS ini format but with any whitespace as field delimiters instead of newlines")
 	cmd.Flags().StringVar(
 		&c.runDir, "run-dir", c.runDir, "the directory to use for the replay data directory; defaults to a random dir in pwd")
 	cmd.Flags().BoolVar(
@@ -53,21 +60,36 @@ type replayConfig struct {
 	name             string
 	pacer            pacerFlag
 	runDir           string
+	count            int
 	streamLogs       bool
 	ignoreCheckpoint bool
+	optionsString    string
+
+	cleanUpFuncs []func() error
 }
 
 func (c *replayConfig) runE(cmd *cobra.Command, args []string) error {
 	stdout := cmd.OutOrStdout()
+	workloadPath := args[0]
+	for i := 1; i <= c.count; i++ {
+		fmt.Fprintf(stdout, "Run %d/%d:\n", i, c.count)
+		if err := c.runOnce(stdout, workloadPath); err != nil {
+			return err
+		}
+	}
+	return nil
+}
 
+func (c *replayConfig) runOnce(stdout io.Writer, workloadPath string) error {
+	defer c.cleanUp()
 	if c.name == "" {
-		c.name = vfs.Default.PathBase(args[0])
+		c.name = vfs.Default.PathBase(workloadPath)
 	}
 
 	r := &replay.Runner{
 		RunDir:       c.runDir,
 		WorkloadFS:   vfs.Default,
-		WorkloadPath: args[0],
+		WorkloadPath: workloadPath,
 		Pacer:        c.pacer,
 		Opts:         &pebble.Options{},
 	}
@@ -77,6 +99,10 @@ func (c *replayConfig) runE(cmd *cobra.Command, args []string) error {
 	if err := c.initOptions(r); err != nil {
 		return err
 	}
+	if verbose {
+		fmt.Fprintln(stdout, "Options:")
+		fmt.Fprintln(stdout, r.Opts.String())
+	}
 
 	// Begin the workload. Run does not block.
 	ctx := context.Background()
@@ -112,6 +138,9 @@ func (c *replayConfig) initRunDir(r *replay.Runner) error {
 		if err != nil {
 			return err
 		}
+		c.cleanUpFuncs = append(c.cleanUpFuncs, func() error {
+			return os.RemoveAll(r.RunDir)
+		})
 	}
 	if !c.ignoreCheckpoint {
 		checkpointDir := r.WorkloadFS.PathJoin(r.WorkloadPath, `checkpoint`)
@@ -169,6 +198,9 @@ func (c *replayConfig) initOptions(r *replay.Runner) error {
 			return err
 		}
 	}
+	if err := parseCustomOptions(c.optionsString, r.Opts); err != nil {
+		return err
+	}
 	// TODO(jackson): If r.Opts.Comparer == nil, peek at the workload's
 	// manifests and pull the comparer out of them.
 	//
@@ -183,6 +215,58 @@ func (c *replayConfig) initOptions(r *replay.Runner) error {
 	return nil
 }
 
+// parseCustomOptions parses Pebble Options passed through a CLI flag.
+// Ordinarily Pebble Options are specified through an INI file with newlines
+// delimiting fields. That doesn't translate well to a CLI interface, so this
+// function accepts fields are that delimited by any whitespace. This is the
+// same format that CockroachDB accepts Pebble Options through the --store flag,
+// and this code is copied from there.
+func parseCustomOptions(optsStr string, opts *pebble.Options) error {
+	if optsStr == "" {
+		return nil
+	}
+	// Pebble options are supplied in the Pebble OPTIONS ini-like
+	// format, but allowing any whitespace to delimit lines. Convert
+	// the options to a newline-delimited format. This isn't a trivial
+	// character replacement because whitespace may appear within a
+	// stanza, eg ["Level 0"].
+	value := strings.TrimSpace(optsStr)
+	var buf bytes.Buffer
+	for len(value) > 0 {
+		i := strings.IndexFunc(value, func(r rune) bool {
+			return r == '[' || unicode.IsSpace(r)
+		})
+		switch {
+		case i == -1:
+			buf.WriteString(value)
+			value = value[len(value):]
+		case value[i] == '[':
+			// If there's whitespace within [ ], we write it verbatim.
+			j := i + strings.IndexRune(value[i:], ']')
+			buf.WriteString(value[:j+1])
+			value = value[j+1:]
+		case unicode.IsSpace(rune(value[i])):
+			// NB: This doesn't handle multibyte whitespace.
+			buf.WriteString(value[:i])
+			buf.WriteRune('\n')
+			value = strings.TrimSpace(value[i+1:])
+		}
+	}
+	return opts.Parse(buf.String(), &pebble.ParseHooks{
+		NewComparer: makeComparer,
+		NewMerger:   makeMerger,
+	})
+}
+
+func (c *replayConfig) cleanUp() error {
+	for _, f := range c.cleanUpFuncs {
+		if err := f(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func makeComparer(name string) (*pebble.Comparer, error) {
 	switch name {
 	case base.DefaultComparer.Name:
diff --git a/commit.go b/commit.go
index 2e2a8634..7f3ae394 100644
--- a/commit.go
+++ b/commit.go
@@ -266,6 +266,23 @@ func newCommitPipeline(env commitEnv) *commitPipeline {
 	return p
 }
 
+// directWrite is used to directly write to the WAL. commitPipeline.mu must be
+// held while this is called. DB.mu must not be held. directWrite will only
+// return once the WAL sync is complete. Note that DirectWrite is a special case
+// function which is currently only used when ingesting sstables as a flushable.
+// Reason carefully about the correctness argument when calling this function
+// from any context.
+func (p *commitPipeline) directWrite(b *Batch) error {
+	var syncWG sync.WaitGroup
+	var syncErr error
+	syncWG.Add(1)
+	p.logSyncQSem <- struct{}{}
+	_, err := p.env.write(b, &syncWG, &syncErr)
+	syncWG.Wait()
+	err = firstError(err, syncErr)
+	return err
+}
+
 // Commit the specified batch, writing it to the WAL, optionally syncing the
 // WAL, and applying the batch to the memtable. Upon successful return the
 // batch's mutations will be visible for reading.
@@ -331,7 +348,9 @@ func (p *commitPipeline) Commit(b *Batch, syncWAL bool, noSyncWait bool) error {
 // sstable ingestion within the commit pipeline. The prepare callback is
 // invoked with commitPipeline.mu held, but note that DB.mu is not held and
 // must be locked if necessary.
-func (p *commitPipeline) AllocateSeqNum(count int, prepare func(), apply func(seqNum uint64)) {
+func (p *commitPipeline) AllocateSeqNum(
+	count int, prepare func(seqNum uint64), apply func(seqNum uint64),
+) {
 	// This method is similar to Commit and prepare. Be careful about trying to
 	// share additional code with those methods because Commit and prepare are
 	// performance critical code paths.
@@ -382,7 +401,7 @@ func (p *commitPipeline) AllocateSeqNum(count int, prepare func(), apply func(se
 	// Invoke the prepare callback. Note the lack of error reporting. Even if the
 	// callback internally fails, the sequence number needs to be published in
 	// order to allow the commit pipeline to proceed.
-	prepare()
+	prepare(b.SeqNum())
 
 	p.mu.Unlock()
 
diff --git a/compaction.go b/compaction.go
index 1d5047f6..f63df64f 100644
--- a/compaction.go
+++ b/compaction.go
@@ -300,6 +300,7 @@ const (
 	compactionKindElisionOnly
 	compactionKindRead
 	compactionKindRewrite
+	compactionKindIngestedFlushable
 )
 
 func (k compactionKind) String() string {
@@ -318,6 +319,8 @@ func (k compactionKind) String() string {
 		return "read"
 	case compactionKindRewrite:
 		return "rewrite"
+	case compactionKindIngestedFlushable:
+		return "ingested-flushable"
 	}
 	return "?"
 }
@@ -672,6 +675,25 @@ func newFlush(opts *Options, cur *version, baseLevel int, flushing flushableList
 	}
 	c.startLevel = &c.inputs[0]
 	c.outputLevel = &c.inputs[1]
+
+	if len(flushing) > 0 {
+		if _, ok := flushing[0].flushable.(*ingestedFlushable); ok {
+			if len(flushing) != 1 {
+				panic("pebble: ingestedFlushable must be flushed one at a time.")
+			}
+			c.kind = compactionKindIngestedFlushable
+			return c
+		}
+	}
+
+	// Make sure there's no ingestedFlushable after the first flushable in the
+	// list.
+	for _, f := range flushing {
+		if _, ok := f.flushable.(*ingestedFlushable); ok {
+			panic("pebble: flushing shouldn't contain ingestedFlushable flushable")
+		}
+	}
+
 	if cur.L0Sublevels != nil {
 		c.l0Limits = cur.L0Sublevels.FlushSplitKeys()
 	}
@@ -1642,14 +1664,106 @@ func (d *DB) flush() {
 	})
 }
 
+// runIngestFlush is used to generate a flush version edit for sstables which
+// were ingested as flushables. Both DB.mu and the manifest lock must be held
+// while runIngestFlush is called.
+func (d *DB) runIngestFlush(c *compaction) (*manifest.VersionEdit, error) {
+	if len(c.flushing) != 1 {
+		panic("pebble: ingestedFlushable must be flushed one at a time.")
+	}
+
+	// Construct the VersionEdit, levelMetrics etc.
+	c.metrics = make(map[int]*LevelMetrics, numLevels)
+	// Finding the target level for ingestion must use the latest version
+	// after the logLock has been acquired.
+	c.version = d.mu.versions.currentVersion()
+
+	baseLevel := d.mu.versions.picker.getBaseLevel()
+	iterOpts := IterOptions{logger: d.opts.Logger}
+	ve := &versionEdit{}
+	var level int
+	var err error
+	for _, file := range c.flushing[0].flushable.(*ingestedFlushable).files {
+		level, err = ingestTargetLevel(
+			d.newIters, d.tableNewRangeKeyIter, iterOpts, d.cmp,
+			c.version, baseLevel, d.mu.compact.inProgress, file,
+		)
+		if err != nil {
+			return nil, err
+		}
+		ve.NewFiles = append(ve.NewFiles, newFileEntry{Level: level, Meta: file})
+		levelMetrics := c.metrics[level]
+		if levelMetrics == nil {
+			levelMetrics = &LevelMetrics{}
+			c.metrics[level] = levelMetrics
+		}
+		levelMetrics.NumFiles++
+		levelMetrics.Size += int64(file.Size)
+		levelMetrics.BytesIngested += file.Size
+		levelMetrics.TablesIngested++
+	}
+
+	return ve, nil
+}
+
 // flush runs a compaction that copies the immutable memtables from memory to
 // disk.
 //
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
 func (d *DB) flush1() (bytesFlushed uint64, err error) {
+	// NB: The flushable queue can contain flushables of type ingestedFlushable.
+	// The sstables in ingestedFlushable.files must be placed into the appropriate
+	// level in the lsm. Let's say the flushable queue contains a prefix of
+	// regular immutable memtables, then an ingestedFlushable, and then the
+	// mutable memtable. When the flush of the ingestedFlushable is performed,
+	// it needs an updated view of the lsm. That is, the prefix of immutable
+	// memtables must have already been flushed. Similarly, if there are two
+	// contiguous ingestedFlushables in the queue, then the first flushable must
+	// be flushed, so that the second flushable can see an updated view of the
+	// lsm.
+	//
+	// Given the above, we restrict flushes to either some prefix of regular
+	// memtables, or a single flushable of type ingestedFlushable. The DB.flush
+	// function will call DB.maybeScheduleFlush again, so a new flush to finish
+	// the remaining flush work should be scheduled right away.
+	//
+	// NB: Large batches placed in the flushable queue share the WAL with the
+	// previous memtable in the queue. We must ensure the property that both the
+	// large batch and the memtable with which it shares a WAL are flushed
+	// together. The property ensures that the minimum unflushed log number
+	// isn't incremented incorrectly. Since a flushableBatch.readyToFlush always
+	// returns true, and since the large batch will always be placed right after
+	// the memtable with which it shares a WAL, the property is naturally
+	// ensured. The large batch will always be placed after the memtable with
+	// which it shares a WAL because we ensure it in DB.commitWrite by holding
+	// the commitPipeline.mu and then holding DB.mu. As an extra defensive
+	// measure, if we try to flush the memtable without also flushing the
+	// flushable batch in the same flush, since the memtable and flushableBatch
+	// have the same logNum, the errFlushInvariant check below will trigger and
+	// prevent the flush from continuing.
 	var n int
 	for ; n < len(d.mu.mem.queue)-1; n++ {
+		if f, ok := d.mu.mem.queue[n].flushable.(*ingestedFlushable); ok {
+			if n == 0 {
+				// The first flushable is of type ingestedFlushable. Since these
+				// must be flushed individually, we perform a flush for just
+				// this.
+				if !f.readyForFlush() {
+					// This check is almost unnecessary, but we guard against it
+					// just in case this invariant changes in the future.
+					panic("pebble: ingestedFlushable should always be ready to flush.")
+				}
+				// By setting n = 1, we ensure that the first flushable(n == 0)
+				// is scheduled for a flush.
+				n = 1
+				break
+			} else {
+				// There was some prefix of flushables which weren't of type
+				// ingestedFlushable. So, perform a flush for those.
+				break
+			}
+		}
 		if !d.mu.mem.queue[n].readyForFlush() {
 			break
 		}
@@ -1683,8 +1797,28 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	})
 	startTime := d.timeNow()
 
-	ve, pendingOutputs, err := d.runCompaction(jobID, c)
+	var ve *manifest.VersionEdit
+	var pendingOutputs []*manifest.FileMetadata
+	// To determine the target level of the files in the ingestedFlushable, we
+	// need to acquire the logLock, and not release it for that duration. Since,
+	// we need to acquire the logLock below to perform the logAndApply step
+	// anyway, we create the VersionEdit for ingestedFlushable outside of
+	// runCompaction. For all other flush cases, we construct the VersionEdit
+	// inside runCompaction.
+	if c.kind != compactionKindIngestedFlushable {
+		ve, pendingOutputs, err = d.runCompaction(jobID, c)
+	}
 
+	// Acquire logLock. This will be released either on an error, by way of
+	// logUnlock, or through a call to logAndApply if there is no error.
+	d.mu.versions.logLock()
+
+	if c.kind == compactionKindIngestedFlushable {
+		ve, err = d.runIngestFlush(c)
+	}
+
+	// TODO(bananabrick): Update the FlushInfo output message. The files are
+	// not necessarily being flushed to L0.
 	info := FlushInfo{
 		JobID:    jobID,
 		Input:    n,
@@ -1705,28 +1839,33 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 		// want to bump the minimum unflushed log number to the log number of the
 		// oldest unflushed memtable.
 		ve.MinUnflushedLogNum = minUnflushedLogNum
-		metrics := c.metrics[0]
-		if d.opts.DisableWAL {
-			// If the WAL is disabled, every flushable has a zero [logSize],
-			// resulting in zero bytes in. Instead, use the number of bytes we
-			// flushed as the BytesIn. This ensures we get a reasonable w-amp
-			// calculation even when the WAL is disabled.
-			metrics.BytesIn = metrics.BytesFlushed
-		} else {
-			for i := 0; i < n; i++ {
-				metrics.BytesIn += d.mu.mem.queue[i].logSize
+		if c.kind != compactionKindIngestedFlushable {
+			metrics := c.metrics[0]
+			if d.opts.DisableWAL {
+				// If the WAL is disabled, every flushable has a zero [logSize],
+				// resulting in zero bytes in. Instead, use the number of bytes we
+				// flushed as the BytesIn. This ensures we get a reasonable w-amp
+				// 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
+				}
 			}
 		}
-
-		d.mu.versions.logLock()
 		err = d.mu.versions.logAndApply(jobID, ve, c.metrics, false, /* forceRotation */
 			func() []compactionInfo { return d.getInProgressCompactionInfoLocked(c) })
 		if err != nil {
 			info.Err = err
 			// TODO(peter): untested.
 			d.mu.versions.obsoleteTables = append(d.mu.versions.obsoleteTables, pendingOutputs...)
-			d.mu.versions.incrementObsoleteTablesLocked(pendingOutputs)
+			d.mu.versions.updateObsoleteTableMetricsLocked()
 		}
+	} else {
+		// We won't be performing the logAndApply step because of the error,
+		// so logUnlock.
+		d.mu.versions.logUnlock()
 	}
 
 	bytesFlushed = c.bytesIterated
@@ -1748,6 +1887,15 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	info.TotalDuration = d.timeNow().Sub(startTime)
 	d.opts.EventListener.FlushEnd(info)
 
+	// The order of these operations matters here for ease of testing.
+	// Removing the reader reference first allows tests to be guaranteed that
+	// the memtable reservation has been released by the time a synchronous
+	// flush returns. readerUnrefLocked may also produce obsolete files so the
+	// call to deleteObsoleteFiles must happen after it.
+	for i := range flushed {
+		flushed[i].readerUnrefLocked(true)
+	}
+
 	d.deleteObsoleteFiles(jobID, false /* waitForOngoing */)
 
 	// Mark all the memtables we flushed as flushed. Note that we do this last so
@@ -1757,13 +1905,9 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 	// is not marked as completed until the deletion of obsolete files job has
 	// completed.
 	for i := range flushed {
-		// The order of these operations matters here for ease of testing. Removing
-		// the reader reference first allows tests to be guaranteed that the
-		// memtable reservation has been released by the time a synchronous flush
-		// returns.
-		flushed[i].readerUnref()
 		close(flushed[i].flushed)
 	}
+
 	return bytesFlushed, err
 }
 
@@ -2208,7 +2352,7 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 		if err != nil {
 			// TODO(peter): untested.
 			d.mu.versions.obsoleteTables = append(d.mu.versions.obsoleteTables, pendingOutputs...)
-			d.mu.versions.incrementObsoleteTablesLocked(pendingOutputs)
+			d.mu.versions.updateObsoleteTableMetricsLocked()
 		}
 	}
 
@@ -2288,6 +2432,10 @@ func (d *DB) runCompaction(
 		return ve, nil, nil
 	}
 
+	if c.kind == compactionKindIngestedFlushable {
+		panic("pebble: runCompaction cannot handle compactionKindIngestedFlushable.")
+	}
+
 	// Check for a trivial move of one table from one level to the next. We avoid
 	// such a move if there is lots of overlapping grandparent data. Otherwise,
 	// the move could create a parent file that will require a very expensive
@@ -2891,6 +3039,19 @@ func (d *DB) scanObsoleteFiles(list []string) {
 
 	liveFileNums := make(map[FileNum]struct{})
 	d.mu.versions.addLiveFileNums(liveFileNums)
+	// Protect against files which are only referred to by the ingestedFlushable
+	// from being deleted. These are added to the flushable queue on WAL replay
+	// during read only mode and aren't part of the Version. Note that if
+	// !d.opts.ReadOnly, then all flushables of type ingestedFlushable have
+	// already been flushed.
+	for _, fEntry := range d.mu.mem.queue {
+		if f, ok := fEntry.flushable.(*ingestedFlushable); ok {
+			for _, file := range f.files {
+				liveFileNums[file.FileNum] = struct{}{}
+			}
+		}
+	}
+
 	minUnflushedLogNum := d.mu.versions.minUnflushedLogNum
 	manifestFileNum := d.mu.versions.manifestFileNum
 
@@ -2950,9 +3111,9 @@ func (d *DB) scanObsoleteFiles(list []string) {
 	}
 
 	d.mu.log.queue = merge(d.mu.log.queue, obsoleteLogs)
-	d.mu.versions.metrics.WAL.Files += int64(len(obsoleteLogs))
+	d.mu.versions.metrics.WAL.Files = int64(len(d.mu.log.queue))
 	d.mu.versions.obsoleteTables = mergeFileMetas(d.mu.versions.obsoleteTables, obsoleteTables)
-	d.mu.versions.incrementObsoleteTablesLocked(obsoleteTables)
+	d.mu.versions.updateObsoleteTableMetricsLocked()
 	d.mu.versions.obsoleteManifests = merge(d.mu.versions.obsoleteManifests, obsoleteManifests)
 	d.mu.versions.obsoleteOptions = merge(d.mu.versions.obsoleteOptions, obsoleteOptions)
 }
@@ -3015,9 +3176,9 @@ func (d *DB) releaseCleaningTurn() {
 	d.mu.cleaner.cond.Broadcast()
 }
 
-// deleteObsoleteFiles deletes those files that are no longer needed. If
-// waitForOngoing is true, it waits for any ongoing cleaning turns to complete,
-// and if false, it returns rightaway if a cleaning turn is ongoing.
+// deleteObsoleteFiles deletes those files and objects that are no longer
+// needed. If waitForOngoing is true, it waits for any ongoing cleaning turns to
+// complete, and if false, it returns rightaway if a cleaning turn is ongoing.
 //
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
@@ -3165,8 +3326,10 @@ func (d *DB) paceAndDeleteObsoleteFiles(jobID int, files []obsoleteFile) {
 			d.mu.versions.metrics.Table.ObsoleteCount--
 			d.mu.versions.metrics.Table.ObsoleteSize -= of.fileSize
 			d.mu.Unlock()
+			d.deleteObsoleteObject(fileTypeTable, jobID, of.fileNum)
+		} else {
+			d.deleteObsoleteFile(of.fileType, jobID, path, of.fileNum)
 		}
-		d.deleteObsoleteFile(of.fileType, jobID, path, of.fileNum)
 	}
 }
 
@@ -3194,7 +3357,29 @@ func (d *DB) maybeScheduleObsoleteTableDeletion() {
 	}()
 }
 
-// deleteObsoleteFile deletes file that is no longer needed.
+func (d *DB) deleteObsoleteObject(fileType fileType, jobID int, fileNum FileNum) {
+	if fileType != fileTypeTable {
+		panic("not an object")
+	}
+
+	path := d.objProvider.Path(fileType, fileNum)
+	err := d.objProvider.Remove(fileType, fileNum)
+	if objstorage.IsNotExistError(err) {
+		return
+	}
+
+	switch fileType {
+	case fileTypeTable:
+		d.opts.EventListener.TableDeleted(TableDeleteInfo{
+			JobID:   jobID,
+			Path:    path,
+			FileNum: fileNum,
+			Err:     err,
+		})
+	}
+}
+
+// deleteObsoleteFile deletes a (non-object) file that is no longer needed.
 func (d *DB) deleteObsoleteFile(fileType fileType, jobID int, path string, fileNum FileNum) {
 	// TODO(peter): need to handle this error, probably by re-adding the
 	// file that couldn't be deleted to one of the obsolete slices map.
@@ -3219,12 +3404,7 @@ func (d *DB) deleteObsoleteFile(fileType fileType, jobID int, path string, fileN
 			Err:     err,
 		})
 	case fileTypeTable:
-		d.opts.EventListener.TableDeleted(TableDeleteInfo{
-			JobID:   jobID,
-			Path:    path,
-			FileNum: fileNum,
-			Err:     err,
-		})
+		panic("invalid deletion of object file")
 	}
 }
 
diff --git a/db.go b/db.go
index 8787f5b3..afd8876c 100644
--- a/db.go
+++ b/db.go
@@ -385,9 +385,6 @@ type DB struct {
 			// index is set it is never modified making a fixed slice immutable and
 			// safe for concurrent reads.
 			queue flushableList
-			// True when the memtable is actively being switched. Both mem.mutable and
-			// log.LogWriter are invalid while switching is true.
-			switching bool
 			// nextSize is the size of the next memtable. The memtable size starts at
 			// min(256KB,Options.MemTableSize) and doubles each time a new memtable
 			// is allocated up to Options.MemTableSize. This reduces the memory
@@ -879,8 +876,14 @@ func (d *DB) commitWrite(b *Batch, syncWG *sync.WaitGroup, syncErr *error) (*mem
 
 	d.mu.Lock()
 
-	// Switch out the memtable if there was not enough room to store the batch.
-	err := d.makeRoomForWrite(b)
+	var err error
+	if !b.ingestedSSTBatch {
+		// Batches which contain keys of kind InternalKeyKindIngestSST will
+		// never be applied to the memtable, so we don't need to make room for
+		// write. For the other cases, switch out the memtable if there was not
+		// enough room to store the batch.
+		err = d.makeRoomForWrite(b)
+	}
 
 	if err == nil && !d.opts.DisableWAL {
 		d.mu.log.bytesIn += uint64(len(repr))
@@ -1356,7 +1359,11 @@ func (d *DB) Close() error {
 	}
 
 	for _, mem := range d.mu.mem.queue {
-		mem.readerUnref()
+		// Usually, we'd want to delete the files returned by readerUnref. But
+		// in this case, even if we're unreferencing the flushables, the
+		// flushables aren't obsolete. They will be reconstructed during WAL
+		// replay.
+		mem.readerUnrefLocked(false)
 	}
 	if reserved := atomic.LoadInt64(&d.atomic.memTableReserved); reserved != 0 {
 		err = firstError(err, errors.Errorf("leaked memtable reservation: %d", errors.Safe(reserved)))
@@ -1643,6 +1650,9 @@ func (d *DB) Metrics() *Metrics {
 		d.opts.Logger.Infof("metrics error: %s", err)
 	}
 	metrics.Flush.WriteThroughput = d.mu.compact.flushWriteThroughput
+	if d.mu.compact.flushing {
+		metrics.Flush.NumInProgress = 1
+	}
 
 	d.mu.Unlock()
 
@@ -1832,11 +1842,13 @@ func (d *DB) newMemTable(logNum FileNum, logSeqNum uint64) (*memTable, *flushabl
 
 func (d *DB) newFlushableEntry(f flushable, logNum FileNum, logSeqNum uint64) *flushableEntry {
 	return &flushableEntry{
-		flushable:  f,
-		flushed:    make(chan struct{}),
-		logNum:     logNum,
-		logSeqNum:  logSeqNum,
-		readerRefs: 1,
+		flushable:      f,
+		flushed:        make(chan struct{}),
+		logNum:         logNum,
+		logSeqNum:      logSeqNum,
+		readerRefs:     1,
+		deleteFn:       d.mu.versions.addObsolete,
+		deleteFnLocked: d.mu.versions.addObsoleteLocked,
 	}
 }
 
@@ -1850,13 +1862,13 @@ func (d *DB) newFlushableEntry(f flushable, logNum FileNum, logSeqNum uint64) *f
 // Both DB.mu and commitPipeline.mu must be held by the caller. Note that DB.mu
 // may be released and reacquired.
 func (d *DB) makeRoomForWrite(b *Batch) error {
+	if b != nil && b.ingestedSSTBatch {
+		panic("pebble: invalid function call")
+	}
+
 	force := b == nil || b.flushable != nil
 	stalled := false
 	for {
-		if d.mu.mem.switching {
-			d.mu.mem.cond.Wait()
-			continue
-		}
 		if b != nil && b.flushable == nil {
 			err := d.mu.mem.mutable.prepare(b)
 			if err != arenaskl.ErrArenaFull {
@@ -1903,130 +1915,10 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			continue
 		}
 
-		var newLogNum FileNum
-		var newLogFile vfs.File
-		var newLogSize uint64
+		var newLogNum base.FileNum
 		var prevLogSize uint64
-		var err error
-
-		if !d.opts.DisableWAL {
-			jobID := d.mu.nextJobID
-			d.mu.nextJobID++
-			newLogNum = d.mu.versions.getNextFileNum()
-			d.mu.mem.switching = true
-
-			prevLogSize = uint64(d.mu.log.Size())
-
-			// The previous log may have grown past its original physical
-			// size. Update its file size in the queue so we have a proper
-			// accounting of its file size.
-			if d.mu.log.queue[len(d.mu.log.queue)-1].fileSize < prevLogSize {
-				d.mu.log.queue[len(d.mu.log.queue)-1].fileSize = prevLogSize
-			}
-			d.mu.Unlock()
-
-			// Close the previous log first. This writes an EOF trailer
-			// signifying the end of the file and syncs it to disk. We must
-			// close the previous log before linking the new log file,
-			// otherwise a crash could leave both logs with unclean tails, and
-			// Open will treat the previous log as corrupt.
-			err = d.mu.log.LogWriter.Close()
-			metrics := d.mu.log.LogWriter.Metrics()
-			d.mu.Lock()
-			if err := d.mu.log.metrics.Merge(metrics); err != nil {
-				d.opts.Logger.Infof("metrics error: %s", err)
-			}
-			d.mu.Unlock()
-
-			newLogName := base.MakeFilepath(d.opts.FS, d.walDirname, fileTypeLog, newLogNum)
-
-			// Try to use a recycled log file. Recycling log files is an important
-			// performance optimization as it is faster to sync a file that has
-			// already been written, than one which is being written for the first
-			// time. This is due to the need to sync file metadata when a file is
-			// being written for the first time. Note this is true even if file
-			// preallocation is performed (e.g. fallocate).
-			var recycleLog fileInfo
-			var recycleOK bool
-			if err == nil {
-				recycleLog, recycleOK = d.logRecycler.peek()
-				if recycleOK {
-					recycleLogName := base.MakeFilepath(d.opts.FS, d.walDirname, fileTypeLog, recycleLog.fileNum)
-					newLogFile, err = d.opts.FS.ReuseForWrite(recycleLogName, newLogName)
-					base.MustExist(d.opts.FS, newLogName, d.opts.Logger, err)
-				} else {
-					newLogFile, err = d.opts.FS.Create(newLogName)
-					base.MustExist(d.opts.FS, newLogName, d.opts.Logger, err)
-				}
-			}
-
-			if err == nil && recycleOK {
-				// Figure out the recycled WAL size. This Stat is necessary
-				// because ReuseForWrite's contract allows for removing the
-				// old file and creating a new one. We don't know whether the
-				// WAL was actually recycled.
-				// TODO(jackson): Adding a boolean to the ReuseForWrite return
-				// value indicating whether or not the file was actually
-				// reused would allow us to skip the stat and use
-				// recycleLog.fileSize.
-				var finfo os.FileInfo
-				finfo, err = newLogFile.Stat()
-				if err == nil {
-					newLogSize = uint64(finfo.Size())
-				}
-			}
-
-			if err == nil {
-				// TODO(peter): RocksDB delays sync of the parent directory until the
-				// first time the log is synced. Is that worthwhile?
-				err = d.walDir.Sync()
-			}
-
-			if err != nil && newLogFile != nil {
-				newLogFile.Close()
-			} else if err == nil {
-				newLogFile = vfs.NewSyncingFile(newLogFile, vfs.SyncingFileOptions{
-					NoSyncOnClose:   d.opts.NoSyncOnClose,
-					BytesPerSync:    d.opts.WALBytesPerSync,
-					PreallocateSize: d.walPreallocateSize(),
-				})
-			}
-
-			if recycleOK {
-				err = firstError(err, d.logRecycler.pop(recycleLog.fileNum))
-			}
-
-			d.opts.EventListener.WALCreated(WALCreateInfo{
-				JobID:           jobID,
-				Path:            newLogName,
-				FileNum:         newLogNum,
-				RecycledFileNum: recycleLog.fileNum,
-				Err:             err,
-			})
-
-			d.mu.Lock()
-			d.mu.mem.switching = false
-			d.mu.mem.cond.Broadcast()
-
-			d.mu.versions.metrics.WAL.Files++
-		}
-
-		if err != nil {
-			// TODO(peter): avoid chewing through file numbers in a tight loop if there
-			// is an error here.
-			//
-			// What to do here? Stumbling on doesn't seem worthwhile. If we failed to
-			// close the previous log it is possible we lost a write.
-			panic(err)
-		}
-
 		if !d.opts.DisableWAL {
-			d.mu.log.queue = append(d.mu.log.queue, fileInfo{fileNum: newLogNum, fileSize: newLogSize})
-			d.mu.log.LogWriter = record.NewLogWriter(newLogFile, newLogNum, record.LogWriterConfig{
-				WALFsyncLatency:    d.mu.log.metrics.fsyncLatency,
-				WALMinSyncInterval: d.opts.WALMinSyncInterval,
-				QueueSemChan:       d.commit.logSyncQSem,
-			})
+			newLogNum, prevLogSize = d.recycleWAL()
 		}
 
 		immMem := d.mu.mem.mutable
@@ -2046,10 +1938,10 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			// The batch is too large to fit in the memtable so add it directly to
 			// the immutable queue. The flushable batch is associated with the same
 			// log as the immutable memtable, but logically occurs after it in
-			// seqnum space. So give the flushable batch the logNum and clear it from
-			// the immutable log. This is done as a defensive measure to prevent the
-			// WAL containing the large batch from being deleted prematurely if the
-			// corresponding memtable is flushed without flushing the large batch.
+			// seqnum space. We ensure while flushing that the flushable batch
+			// is flushed along with the previous memtable in the flushable
+			// queue. See the top level comment in DB.flush1 to learn how this
+			// is ensured.
 			//
 			// See DB.commitWrite for the special handling of log writes for large
 			// batches. In particular, the large batch has already written to
@@ -2059,7 +1951,6 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			// for it until it is flushed.
 			entry.releaseMemAccounting = d.opts.Cache.Reserve(int(b.flushable.totalBytes()))
 			d.mu.mem.queue = append(d.mu.mem.queue, entry)
-			imm.logNum = 0
 		}
 
 		var logSeqNum uint64
@@ -2071,29 +1962,162 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 		} else {
 			logSeqNum = atomic.LoadUint64(&d.mu.versions.atomic.logSeqNum)
 		}
+		d.rotateMemtable(newLogNum, logSeqNum, immMem)
+		force = false
+	}
+}
 
-		// Create a new memtable, scheduling the previous one for flushing. We do
-		// this even if the previous memtable was empty because the DB.Flush
-		// mechanism is dependent on being able to wait for the empty memtable to
-		// flush. We can't just mark the empty memtable as flushed here because we
-		// also have to wait for all previous immutable tables to
-		// flush. Additionally, the memtable is tied to particular WAL file and we
-		// want to go through the flush path in order to recycle that WAL file.
-		//
-		// NB: newLogNum corresponds to the WAL that contains mutations that are
-		// present in the new memtable. When immutable memtables are flushed to
-		// disk, a VersionEdit will be created telling the manifest the minimum
-		// unflushed log number (which will be the next one in d.mu.mem.mutable
-		// that was not flushed).
-		var entry *flushableEntry
-		d.mu.mem.mutable, entry = d.newMemTable(newLogNum, logSeqNum)
-		d.mu.mem.queue = append(d.mu.mem.queue, entry)
-		d.updateReadStateLocked(nil)
-		if immMem.writerUnref() {
-			d.maybeScheduleFlush()
+// Both DB.mu and commitPipeline.mu must be held by the caller.
+func (d *DB) rotateMemtable(newLogNum FileNum, logSeqNum uint64, prev *memTable) {
+	// Create a new memtable, scheduling the previous one for flushing. We do
+	// this even if the previous memtable was empty because the DB.Flush
+	// mechanism is dependent on being able to wait for the empty memtable to
+	// flush. We can't just mark the empty memtable as flushed here because we
+	// also have to wait for all previous immutable tables to
+	// flush. Additionally, the memtable is tied to particular WAL file and we
+	// want to go through the flush path in order to recycle that WAL file.
+	//
+	// NB: newLogNum corresponds to the WAL that contains mutations that are
+	// present in the new memtable. When immutable memtables are flushed to
+	// disk, a VersionEdit will be created telling the manifest the minimum
+	// unflushed log number (which will be the next one in d.mu.mem.mutable
+	// that was not flushed).
+	//
+	// NB: prev should be the current mutable memtable.
+	var entry *flushableEntry
+	d.mu.mem.mutable, entry = d.newMemTable(newLogNum, logSeqNum)
+	d.mu.mem.queue = append(d.mu.mem.queue, entry)
+	d.updateReadStateLocked(nil)
+	if prev.writerUnref() {
+		d.maybeScheduleFlush()
+	}
+}
+
+// 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) {
+	if d.opts.DisableWAL {
+		panic("pebble: invalid function call")
+	}
+
+	jobID := d.mu.nextJobID
+	d.mu.nextJobID++
+	newLogNum = d.mu.versions.getNextFileNum()
+
+	prevLogSize = uint64(d.mu.log.Size())
+
+	// The previous log may have grown past its original physical
+	// size. Update its file size in the queue so we have a proper
+	// accounting of its file size.
+	if d.mu.log.queue[len(d.mu.log.queue)-1].fileSize < prevLogSize {
+		d.mu.log.queue[len(d.mu.log.queue)-1].fileSize = prevLogSize
+	}
+	d.mu.Unlock()
+
+	var err error
+	// Close the previous log first. This writes an EOF trailer
+	// signifying the end of the file and syncs it to disk. We must
+	// close the previous log before linking the new log file,
+	// otherwise a crash could leave both logs with unclean tails, and
+	// Open will treat the previous log as corrupt.
+	err = d.mu.log.LogWriter.Close()
+	metrics := d.mu.log.LogWriter.Metrics()
+	d.mu.Lock()
+	if err := d.mu.log.metrics.Merge(metrics); err != nil {
+		d.opts.Logger.Infof("metrics error: %s", err)
+	}
+	d.mu.Unlock()
+
+	newLogName := base.MakeFilepath(d.opts.FS, d.walDirname, fileTypeLog, newLogNum)
+
+	// Try to use a recycled log file. Recycling log files is an important
+	// performance optimization as it is faster to sync a file that has
+	// already been written, than one which is being written for the first
+	// time. This is due to the need to sync file metadata when a file is
+	// being written for the first time. Note this is true even if file
+	// preallocation is performed (e.g. fallocate).
+	var recycleLog fileInfo
+	var recycleOK bool
+	var newLogFile vfs.File
+	if err == nil {
+		recycleLog, recycleOK = d.logRecycler.peek()
+		if recycleOK {
+			recycleLogName := base.MakeFilepath(d.opts.FS, d.walDirname, fileTypeLog, recycleLog.fileNum)
+			newLogFile, err = d.opts.FS.ReuseForWrite(recycleLogName, newLogName)
+			base.MustExist(d.opts.FS, newLogName, d.opts.Logger, err)
+		} else {
+			newLogFile, err = d.opts.FS.Create(newLogName)
+			base.MustExist(d.opts.FS, newLogName, d.opts.Logger, err)
+		}
+	}
+
+	var newLogSize uint64
+	if err == nil && recycleOK {
+		// Figure out the recycled WAL size. This Stat is necessary
+		// because ReuseForWrite's contract allows for removing the
+		// old file and creating a new one. We don't know whether the
+		// WAL was actually recycled.
+		// TODO(jackson): Adding a boolean to the ReuseForWrite return
+		// value indicating whether or not the file was actually
+		// reused would allow us to skip the stat and use
+		// recycleLog.fileSize.
+		var finfo os.FileInfo
+		finfo, err = newLogFile.Stat()
+		if err == nil {
+			newLogSize = uint64(finfo.Size())
 		}
-		force = false
 	}
+
+	if err == nil {
+		// TODO(peter): RocksDB delays sync of the parent directory until the
+		// first time the log is synced. Is that worthwhile?
+		err = d.walDir.Sync()
+	}
+
+	if err != nil && newLogFile != nil {
+		newLogFile.Close()
+	} else if err == nil {
+		newLogFile = vfs.NewSyncingFile(newLogFile, vfs.SyncingFileOptions{
+			NoSyncOnClose:   d.opts.NoSyncOnClose,
+			BytesPerSync:    d.opts.WALBytesPerSync,
+			PreallocateSize: d.walPreallocateSize(),
+		})
+	}
+
+	if recycleOK {
+		err = firstError(err, d.logRecycler.pop(recycleLog.fileNum))
+	}
+
+	d.opts.EventListener.WALCreated(WALCreateInfo{
+		JobID:           jobID,
+		Path:            newLogName,
+		FileNum:         newLogNum,
+		RecycledFileNum: recycleLog.fileNum,
+		Err:             err,
+	})
+
+	d.mu.Lock()
+	d.mu.mem.cond.Broadcast()
+
+	d.mu.versions.metrics.WAL.Files++
+
+	if err != nil {
+		// TODO(peter): avoid chewing through file numbers in a tight loop if there
+		// is an error here.
+		//
+		// What to do here? Stumbling on doesn't seem worthwhile. If we failed to
+		// close the previous log it is possible we lost a write.
+		panic(err)
+	}
+
+	d.mu.log.queue = append(d.mu.log.queue, fileInfo{fileNum: newLogNum, fileSize: newLogSize})
+	d.mu.log.LogWriter = record.NewLogWriter(newLogFile, newLogNum, record.LogWriterConfig{
+		WALFsyncLatency:    d.mu.log.metrics.fsyncLatency,
+		WALMinSyncInterval: d.opts.WALMinSyncInterval,
+		QueueSemChan:       d.commit.logSyncQSem,
+	})
+
+	return
 }
 
 func (d *DB) getEarliestUnflushedSeqNumLocked() uint64 {
diff --git a/event.go b/event.go
index cec60175..2da40ff8 100644
--- a/event.go
+++ b/event.go
@@ -286,7 +286,10 @@ type TableIngestInfo struct {
 	// GlobalSeqNum is the sequence number that was assigned to all entries in
 	// the ingested table.
 	GlobalSeqNum uint64
-	Err          error
+	// flushable indicates whether the ingested sstable was treated as a
+	// flushable.
+	flushable bool
+	Err       error
 }
 
 func (i TableIngestInfo) String() string {
@@ -300,7 +303,12 @@ func (i TableIngestInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 		return
 	}
 
-	w.Printf("[JOB %d] ingested", redact.Safe(i.JobID))
+	if i.flushable {
+		w.Printf("[JOB %d] ingested as flushable", redact.Safe(i.JobID))
+	} else {
+		w.Printf("[JOB %d] ingested", redact.Safe(i.JobID))
+	}
+
 	for j := range i.Tables {
 		t := &i.Tables[j]
 		if j > 0 {
diff --git a/flushable.go b/flushable.go
index c96c91a8..7a754552 100644
--- a/flushable.go
+++ b/flushable.go
@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/pebble/internal/keyspan"
+	"github.com/cockroachdb/pebble/internal/manifest"
 )
 
 // flushable defines the interface for immutable memtables.
@@ -53,12 +54,21 @@ type flushableEntry struct {
 	// readerRefs tracks the read references on the flushable. The two sources of
 	// reader references are DB.mu.mem.queue and readState.memtables. The memory
 	// reserved by the flushable in the cache is released when the reader refs
-	// drop to zero. If the flushable is a memTable, when the reader refs drops
-	// to zero, the writer refs will already be zero because the memtable will
-	// have been flushed and that only occurs once the writer refs drops to zero.
+	// drop to zero. If the flushable is referencing sstables, then the file
+	// refount is also decreased once the reader refs drops to 0. If the
+	// flushable is a memTable, when the reader refs drops to zero, the writer
+	// refs will already be zero because the memtable will have been flushed and
+	// that only occurs once the writer refs drops to zero.
 	readerRefs int32
 	// Closure to invoke to release memory accounting.
 	releaseMemAccounting func()
+	// unrefFiles, if not nil, should be invoked to decrease the ref count of
+	// files which are backing the flushable.
+	unrefFiles func() []*fileMetadata
+	// deleteFnLocked should be called if the caller is holding DB.mu.
+	deleteFnLocked func(obsolete []*fileMetadata)
+	// deleteFn should be called if the caller is not holding DB.mu.
+	deleteFn func(obsolete []*fileMetadata)
 }
 
 func (e *flushableEntry) readerRef() {
@@ -68,7 +78,19 @@ func (e *flushableEntry) readerRef() {
 	}
 }
 
-func (e *flushableEntry) readerUnref() {
+// db.mu must not be held when this is called.
+func (e *flushableEntry) readerUnref(deleteFiles bool) {
+	e.readerUnrefHelper(deleteFiles, e.deleteFn)
+}
+
+// db.mu must be held when this is called.
+func (e *flushableEntry) readerUnrefLocked(deleteFiles bool) {
+	e.readerUnrefHelper(deleteFiles, e.deleteFnLocked)
+}
+
+func (e *flushableEntry) readerUnrefHelper(
+	deleteFiles bool, deleteFn func(obsolete []*manifest.FileMetadata),
+) {
 	switch v := atomic.AddInt32(&e.readerRefs, -1); {
 	case v < 0:
 		panic(fmt.Sprintf("pebble: inconsistent reference count: %d", v))
@@ -78,7 +100,138 @@ func (e *flushableEntry) readerUnref() {
 		}
 		e.releaseMemAccounting()
 		e.releaseMemAccounting = nil
+		if e.unrefFiles != nil {
+			obsolete := e.unrefFiles()
+			e.unrefFiles = nil
+			if deleteFiles {
+				deleteFn(obsolete)
+			}
+		}
 	}
 }
 
 type flushableList []*flushableEntry
+
+// ingestedFlushable is the implementation of the flushable interface for the
+// ingesting sstables which are added to the flushable list.
+type ingestedFlushable struct {
+	files            []*fileMetadata
+	cmp              Compare
+	split            Split
+	newIters         tableNewIters
+	newRangeKeyIters keyspan.TableNewSpanIter
+
+	// Since the level slice is immutable, we construct and set it once. It
+	// should be safe to read from slice in future reads.
+	slice manifest.LevelSlice
+	// hasRangeKeys is set on ingestedFlushable construction.
+	hasRangeKeys bool
+}
+
+func newIngestedFlushable(
+	files []*fileMetadata,
+	cmp Compare,
+	split Split,
+	newIters tableNewIters,
+	newRangeKeyIters keyspan.TableNewSpanIter,
+) *ingestedFlushable {
+	ret := &ingestedFlushable{
+		files:            files,
+		cmp:              cmp,
+		split:            split,
+		newIters:         newIters,
+		newRangeKeyIters: newRangeKeyIters,
+		// slice is immutable and can be set once and used many times.
+		slice: manifest.NewLevelSliceKeySorted(cmp, files),
+	}
+
+	for _, f := range files {
+		if f.HasRangeKeys {
+			ret.hasRangeKeys = true
+			break
+		}
+	}
+
+	return ret
+}
+
+func (s *ingestedFlushable) newIter(o *IterOptions) internalIterator {
+	var opts IterOptions
+	if o != nil {
+		opts = *o
+	}
+	// TODO(bananabrick): The manifest.Level in newLevelIter is only used for
+	// logging. Update the manifest.Level encoding to account for levels which
+	// aren't truly levels in the lsm. Right now, the encoding only supports
+	// L0 sublevels, and the rest of the levels in the lsm.
+	return newLevelIter(
+		opts, s.cmp, s.split, s.newIters, s.slice.Iter(), manifest.Level(0), nil,
+	)
+}
+
+func (s *ingestedFlushable) newFlushIter(o *IterOptions, bytesFlushed *uint64) internalIterator {
+	// newFlushIter is only used for writing memtables to disk as sstables.
+	// Since ingested sstables are already present on disk, they don't need to
+	// make use of a flush iter.
+	panic("pebble: not implemented")
+}
+
+func (s *ingestedFlushable) constructRangeDelIter(
+	file *manifest.FileMetadata, _ *keyspan.SpanIterOptions,
+) (keyspan.FragmentIterator, error) {
+	// Note that the keyspan level iter expects a non-nil iterator to be
+	// returned even if there is an error. So, we return the emptyKeyspanIter.
+	iter, rangeDelIter, err := s.newIters(file, nil, internalIterOpts{})
+	if err != nil {
+		return emptyKeyspanIter, err
+	}
+	iter.Close()
+	if rangeDelIter == nil {
+		return emptyKeyspanIter, nil
+	}
+	return rangeDelIter, nil
+}
+
+// TODO(bananabrick): Using a level iter instead of a keyspan level iter to
+// surface range deletes is more efficient.
+func (s *ingestedFlushable) newRangeDelIter(_ *IterOptions) keyspan.FragmentIterator {
+	return keyspan.NewLevelIter(
+		keyspan.SpanIterOptions{}, s.cmp,
+		s.constructRangeDelIter, s.slice.Iter(), manifest.Level(0),
+		manifest.KeyTypePoint,
+	)
+}
+
+func (s *ingestedFlushable) newRangeKeyIter(o *IterOptions) keyspan.FragmentIterator {
+	if !s.containsRangeKeys() {
+		return nil
+	}
+
+	return keyspan.NewLevelIter(
+		keyspan.SpanIterOptions{}, s.cmp, s.newRangeKeyIters,
+		s.slice.Iter(), manifest.Level(0), manifest.KeyTypeRange,
+	)
+}
+
+func (s *ingestedFlushable) containsRangeKeys() bool {
+	return s.hasRangeKeys
+}
+
+func (s *ingestedFlushable) inuseBytes() uint64 {
+	// inuseBytes is only used when memtables are flushed to disk as sstables.
+	panic("pebble: not implemented")
+}
+
+func (s *ingestedFlushable) totalBytes() uint64 {
+	// We don't allocate additional bytes for the ingestedFlushable.
+	return 0
+}
+
+func (s *ingestedFlushable) readyForFlush() bool {
+	// ingestedFlushable should always be ready to flush. However, note that
+	// memtables before the ingested sstables in the memtable queue must be
+	// flushed before an ingestedFlushable can be flushed. This is because the
+	// ingested sstables need an updated view of the Version to
+	// determine where to place the files in the lsm.
+	return true
+}
diff --git a/ingest.go b/ingest.go
index 6403a1ce..5424f39d 100644
--- a/ingest.go
+++ b/ingest.go
@@ -6,6 +6,7 @@ package pebble
 
 import (
 	"sort"
+	"sync/atomic"
 	"time"
 
 	"github.com/cockroachdb/errors"
@@ -264,26 +265,29 @@ func ingestCleanup(objProvider *objstorage.Provider, meta []*fileMetadata) error
 
 func ingestLink(
 	jobID int, opts *Options, objProvider *objstorage.Provider, paths []string, meta []*fileMetadata,
-) error {
+) ([]string, error) {
+	newPaths := make([]string, len(paths))
 	for i := range paths {
+		target := objProvider.Path(fileTypeTable, meta[i].FileNum)
+		newPaths[i] = target
 		err := objProvider.LinkOrCopyFromLocal(opts.FS, paths[i], fileTypeTable, meta[i].FileNum)
 		if err != nil {
 			if err2 := ingestCleanup(objProvider, meta[:i]); err2 != nil {
 				opts.Logger.Infof("ingest cleanup failed: %v", err2)
 			}
-			return err
+			return nil, err
 		}
 		if opts.EventListener.TableCreated != nil {
 			opts.EventListener.TableCreated(TableCreateInfo{
 				JobID:   jobID,
 				Reason:  "ingesting",
-				Path:    objProvider.Path(fileTypeTable, meta[i].FileNum),
+				Path:    target,
 				FileNum: meta[i].FileNum,
 			})
 		}
 	}
 
-	return nil
+	return newPaths, nil
 }
 
 func ingestMemtableOverlaps(cmp Compare, mem flushable, meta []*fileMetadata) bool {
@@ -291,25 +295,26 @@ func ingestMemtableOverlaps(cmp Compare, mem flushable, meta []*fileMetadata) bo
 	rangeDelIter := mem.newRangeDelIter(nil)
 	rkeyIter := mem.newRangeKeyIter(nil)
 
+	closeIters := func() error {
+		err := iter.Close()
+		if rangeDelIter != nil {
+			err = firstError(err, rangeDelIter.Close())
+		}
+		if rkeyIter != nil {
+			err = firstError(err, rkeyIter.Close())
+		}
+		return err
+	}
+
 	for _, m := range meta {
 		if overlapWithIterator(iter, &rangeDelIter, rkeyIter, m, cmp) {
+			closeIters()
 			return true
 		}
 	}
 
-	err := iter.Close()
-	if rangeDelIter != nil {
-		err = firstError(err, rangeDelIter.Close())
-	}
-	if rkeyIter != nil {
-		err = firstError(err, rkeyIter.Close())
-	}
-	if err != nil {
-		// Assume overlap if any iterator errored out.
-		return true
-	}
-
-	return false
+	// Assume overlap if any iterator errored out.
+	return closeIters() != nil
 }
 
 func ingestUpdateSeqNum(
@@ -669,6 +674,113 @@ func (d *DB) IngestWithStats(paths []string) (IngestOperationStats, error) {
 	return d.ingest(paths, ingestTargetLevel)
 }
 
+// Both DB.mu and commitPipeline.mu must be held while this is called.
+func (d *DB) newIngestedFlushableEntry(
+	meta []*fileMetadata, seqNum uint64, logNum FileNum,
+) (*flushableEntry, error) {
+	// Update the sequence number for all of the sstables in the
+	// metadata. Writing the metadata to the manifest when the
+	// version edit is applied is the mechanism that persists the
+	// sequence number. The sstables themselves are left unmodified.
+	// In this case, a version edit will only be written to the manifest
+	// when the flushable is eventually flushed. If Pebble restarts in that
+	// time, then we'll lose the ingest sequence number information. But this
+	// information will also be reconstructed on node restart.
+	if err := ingestUpdateSeqNum(
+		d.cmp, d.opts.Comparer.FormatKey, seqNum, meta,
+	); err != nil {
+		return nil, err
+	}
+
+	f := newIngestedFlushable(meta, d.cmp, d.split, d.newIters, d.tableNewRangeKeyIter)
+
+	// NB: The logNum/seqNum are the WAL number which we're writing this entry
+	// to and the sequence number within the WAL which we'll write this entry
+	// to.
+	entry := d.newFlushableEntry(f, logNum, seqNum)
+	// The flushable entry starts off with a single reader ref, so increment
+	// the FileMetadata.Refs.
+	for _, file := range f.files {
+		atomic.AddInt32(&file.Refs, 1)
+	}
+	entry.unrefFiles = func() []*fileMetadata {
+		var obsolete []*fileMetadata
+		for _, file := range f.files {
+			if val := atomic.AddInt32(&file.Refs, -1); val == 0 {
+				obsolete = append(obsolete, file)
+			}
+		}
+		return obsolete
+	}
+
+	entry.flushForced = true
+	entry.releaseMemAccounting = func() {}
+	return entry, nil
+}
+
+// Both DB.mu and commitPipeline.mu must be held while this is called. Since
+// we're holding both locks, the order in which we rotate the memtable or
+// recycle the WAL in this function is irrelevant as long as the correct log
+// numbers are assigned to the appropriate flushable.
+func (d *DB) handleIngestAsFlushable(paths []string, meta []*fileMetadata, seqNum uint64) error {
+	b := d.NewBatch()
+	for _, path := range paths {
+		b.ingestSST([]byte(path))
+	}
+	b.setSeqNum(seqNum)
+
+	// If the WAL is disabled, then the logNum used to create the flushable
+	// entry doesn't matter. We just use the logNum assigned to the current
+	// mutable memtable. If the WAL is enabled, then this logNum will be
+	// overwritten by the logNum of the log which will contain the log entry
+	// for the ingestedFlushable.
+	logNum := d.mu.mem.queue[len(d.mu.mem.queue)-1].logNum
+	if !d.opts.DisableWAL {
+		// 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()
+		d.mu.Unlock()
+		err := d.commit.directWrite(b)
+		if err != nil {
+			d.opts.Logger.Fatalf("%v", err)
+		}
+		d.mu.Lock()
+	}
+
+	entry, err := d.newIngestedFlushableEntry(meta, seqNum, logNum)
+	if err != nil {
+		return err
+	}
+	nextSeqNum := seqNum + uint64(b.Count())
+
+	// Set newLogNum to the logNum of the previous flushable. This value is
+	// irrelevant if the WAL is disabled. If the WAL is enabled, then we set
+	// 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
+		}
+	}
+
+	currMem := d.mu.mem.mutable
+	// NB: Placing ingested sstables above the current memtables
+	// requires rotating of the existing memtables/WAL. There is
+	// some concern of churning through tiny memtables due to
+	// ingested sstables being placed on top of them, but those
+	// memtables would have to be flushed anyways.
+	d.mu.mem.queue = append(d.mu.mem.queue, entry)
+	d.rotateMemtable(newLogNum, nextSeqNum, currMem)
+	d.updateReadStateLocked(d.opts.DebugCheck)
+	d.maybeScheduleFlush()
+	return nil
+}
+
 func (d *DB) ingest(
 	paths []string, targetLevelFunc ingestTargetLevelFunc,
 ) (IngestOperationStats, error) {
@@ -707,7 +819,8 @@ func (d *DB) ingest(
 	// (e.g. because the files reside on a different filesystem), ingestLink will
 	// fall back to copying, and if that fails we undo our work and return an
 	// error.
-	if err := ingestLink(jobID, d.opts, d.objProvider, paths, meta); err != nil {
+	newPaths, err := ingestLink(jobID, d.opts, d.objProvider, paths, meta)
+	if err != nil {
 		return IngestOperationStats{}, err
 	}
 	// Fsync the directory we added the tables to. We need to do this at some
@@ -719,7 +832,9 @@ func (d *DB) ingest(
 	}
 
 	var mem *flushableEntry
-	prepare := func() {
+	// asFlushable indicates whether the sstable was ingested as a flushable.
+	var asFlushable bool
+	prepare := func(seqNum uint64) {
 		// Note that d.commit.mu is held by commitPipeline when calling prepare.
 
 		d.mu.Lock()
@@ -732,12 +847,22 @@ func (d *DB) ingest(
 		for i := len(d.mu.mem.queue) - 1; i >= 0; i-- {
 			m := d.mu.mem.queue[i]
 			if ingestMemtableOverlaps(d.cmp, m, meta) {
-				mem = m
-				if mem.flushable == d.mu.mem.mutable {
-					err = d.makeRoomForWrite(nil)
+				if (len(d.mu.mem.queue) > d.opts.MemTableStopWritesThreshold-1) ||
+					!d.opts.Experimental.IngestSSTablesAsFlushable {
+					mem = m
+					if mem.flushable == d.mu.mem.mutable {
+						err = d.makeRoomForWrite(nil)
+					}
+					mem.flushForced = true
+					d.maybeScheduleFlush()
+					return
 				}
-				mem.flushForced = true
-				d.maybeScheduleFlush()
+
+				// The ingestion overlaps with the memtable. Since there aren't
+				// too many memtables already queued up, we can slide the
+				// ingested sstables on top of the existing memtables.
+				err = d.handleIngestAsFlushable(newPaths, meta, seqNum)
+				asFlushable = true
 				return
 			}
 		}
@@ -745,7 +870,7 @@ func (d *DB) ingest(
 
 	var ve *versionEdit
 	apply := func(seqNum uint64) {
-		if err != nil {
+		if err != nil || asFlushable {
 			// An error occurred during prepare.
 			return
 		}
@@ -778,6 +903,8 @@ func (d *DB) ingest(
 			d.opts.Logger.Infof("ingest cleanup failed: %v", err2)
 		}
 	} else {
+		// Since we either created a hard link to the ingesting files, or copied
+		// them over, it is safe to remove the originals paths.
 		for _, path := range paths {
 			if err2 := d.opts.FS.Remove(path); err2 != nil {
 				d.opts.Logger.Infof("ingest failed to remove original file: %s", err2)
@@ -789,6 +916,7 @@ func (d *DB) ingest(
 		JobID:        jobID,
 		GlobalSeqNum: meta[0].SmallestSeqNum,
 		Err:          err,
+		flushable:    asFlushable,
 	}
 	var stats IngestOperationStats
 	if ve != nil {
@@ -805,6 +933,15 @@ func (d *DB) ingest(
 				stats.ApproxIngestedIntoL0Bytes += e.Meta.Size
 			}
 		}
+	} else if asFlushable {
+		info.Tables = make([]struct {
+			TableInfo
+			Level int
+		}, len(meta))
+		for i, f := range meta {
+			info.Tables[i].Level = -1
+			info.Tables[i].TableInfo = f.TableInfo()
+		}
 	}
 	d.opts.EventListener.TableIngested(info)
 
diff --git a/internal.go b/internal.go
index 72375410..99fec083 100644
--- a/internal.go
+++ b/internal.go
@@ -22,6 +22,7 @@ const (
 	InternalKeyKindRangeKeySet     = base.InternalKeyKindRangeKeySet
 	InternalKeyKindRangeKeyUnset   = base.InternalKeyKindRangeKeyUnset
 	InternalKeyKindRangeKeyDelete  = base.InternalKeyKindRangeKeyDelete
+	InternalKeyKindIngestSST       = base.InternalKeyIngestSST
 	InternalKeyKindInvalid         = base.InternalKeyKindInvalid
 	InternalKeySeqNumBatch         = base.InternalKeySeqNumBatch
 	InternalKeySeqNumMax           = base.InternalKeySeqNumMax
diff --git a/internal/base/internal.go b/internal/base/internal.go
index 69865e11..208b88f8 100644
--- a/internal/base/internal.go
+++ b/internal/base/internal.go
@@ -60,6 +60,12 @@ const (
 	InternalKeyKindRangeKeyUnset InternalKeyKind = 20
 	InternalKeyKindRangeKeySet   InternalKeyKind = 21
 
+	// InternalKeyIngestSST is used to distinguish a batch that corresponds to
+	// the WAL entry for ingested sstables that are added to the flushable
+	// queue. This InternalKeyKind cannot appear, amongst other key kinds in a
+	// batch, or in an sstable.
+	InternalKeyIngestSST InternalKeyKind = 22
+
 	// This maximum value isn't part of the file format. It's unlikely,
 	// but future extensions may increase this value.
 	//
@@ -69,7 +75,7 @@ const (
 	// which sorts 'less than or equal to' any other valid internalKeyKind, when
 	// searching for any kind of internal key formed by a certain user key and
 	// seqNum.
-	InternalKeyKindMax InternalKeyKind = 21
+	InternalKeyKindMax InternalKeyKind = 22
 
 	// InternalKeyZeroSeqnumMaxTrailer is the largest trailer with a
 	// zero sequence number.
@@ -111,6 +117,7 @@ var internalKeyKindNames = []string{
 	InternalKeyKindRangeKeySet:    "RANGEKEYSET",
 	InternalKeyKindRangeKeyUnset:  "RANGEKEYUNSET",
 	InternalKeyKindRangeKeyDelete: "RANGEKEYDEL",
+	InternalKeyIngestSST:          "INGESTSST",
 	InternalKeyKindInvalid:        "INVALID",
 }
 
diff --git a/internal/keyspan/iter.go b/internal/keyspan/iter.go
index 2be919ac..4798561a 100644
--- a/internal/keyspan/iter.go
+++ b/internal/keyspan/iter.go
@@ -60,7 +60,8 @@ type FragmentIterator interface {
 	Close() error
 }
 
-// TableNewSpanIter creates a new iterator for spans for the given file.
+// TableNewSpanIter creates a new iterator for range key spans for the given
+// file.
 type TableNewSpanIter func(file *manifest.FileMetadata, iterOptions *SpanIterOptions) (FragmentIterator, error)
 
 // SpanIterOptions is a subset of IterOptions that are necessary to instantiate
diff --git a/internal/keyspan/level_iter.go b/internal/keyspan/level_iter.go
index 5da6c135..047c6a24 100644
--- a/internal/keyspan/level_iter.go
+++ b/internal/keyspan/level_iter.go
@@ -73,8 +73,8 @@ type LevelIter struct {
 // LevelIter implements the keyspan.FragmentIterator interface.
 var _ FragmentIterator = (*LevelIter)(nil)
 
-// newLevelIter returns a LevelIter.
-func newLevelIter(
+// NewLevelIter returns a LevelIter.
+func NewLevelIter(
 	opts SpanIterOptions,
 	cmp base.Compare,
 	newIter TableNewSpanIter,
@@ -103,7 +103,10 @@ func (l *LevelIter) Init(
 	l.iterFile = nil
 	l.newIter = newIter
 	switch keyType {
-	case manifest.KeyTypePoint, manifest.KeyTypeRange:
+	case manifest.KeyTypePoint:
+		l.keyType = keyType
+		l.files = files.Filter(keyType)
+	case manifest.KeyTypeRange:
 		l.keyType = keyType
 		l.files = files.Filter(keyType)
 	default:
diff --git a/internal/manifest/btree.go b/internal/manifest/btree.go
index 235b98ad..66464c98 100644
--- a/internal/manifest/btree.go
+++ b/internal/manifest/btree.go
@@ -207,7 +207,7 @@ func (n *node) decRef(contentsToo bool, obsolete *[]*FileMetadata) {
 	// nodes, and they want to preserve the existing reference count.
 	if contentsToo {
 		for _, f := range n.items[:n.count] {
-			if atomic.AddInt32(&f.refs, -1) == 0 {
+			if atomic.AddInt32(&f.Refs, -1) == 0 {
 				// There are two sources of node dereferences: tree mutations
 				// and Version dereferences. Files should only be made obsolete
 				// during Version dereferences, during which `obsolete` will be
@@ -241,7 +241,7 @@ func (n *node) clone() *node {
 	c.subtreeCount = n.subtreeCount
 	// Increase the refcount of each contained item.
 	for _, f := range n.items[:n.count] {
-		atomic.AddInt32(&f.refs, 1)
+		atomic.AddInt32(&f.Refs, 1)
 	}
 	if !c.leaf {
 		// Copy children and increase each refcount.
@@ -800,7 +800,7 @@ func (t *btree) Delete(item *FileMetadata) (obsolete bool) {
 		return false
 	}
 	if out := mut(&t.root).Remove(t.cmp, item); out != nil {
-		obsolete = atomic.AddInt32(&out.refs, -1) == 0
+		obsolete = atomic.AddInt32(&out.Refs, -1) == 0
 	}
 	if invariants.Enabled {
 		t.root.verifyInvariants()
@@ -832,7 +832,7 @@ func (t *btree) Insert(item *FileMetadata) error {
 		newRoot.subtreeCount = t.root.subtreeCount + splitNode.subtreeCount + 1
 		t.root = newRoot
 	}
-	atomic.AddInt32(&item.refs, 1)
+	atomic.AddInt32(&item.Refs, 1)
 	err := mut(&t.root).Insert(t.cmp, item)
 	if invariants.Enabled {
 		t.root.verifyInvariants()
diff --git a/internal/manifest/version.go b/internal/manifest/version.go
index 33c78ceb..e2cb384e 100644
--- a/internal/manifest/version.go
+++ b/internal/manifest/version.go
@@ -143,7 +143,7 @@ type FileMetadata struct {
 	// Reference count for the file: incremented when a file is added to a
 	// version and decremented when the version is unreferenced. The file is
 	// obsolete when the reference count falls to zero.
-	refs int32
+	Refs int32
 	// FileNum is the file number.
 	FileNum base.FileNum
 	// Size is the size of the file, in bytes.
diff --git a/internal/metamorphic/options.go b/internal/metamorphic/options.go
index 4ce3b5c6..b01c4df6 100644
--- a/internal/metamorphic/options.go
+++ b/internal/metamorphic/options.go
@@ -133,6 +133,7 @@ func defaultOptions() *pebble.Options {
 			FilterPolicy: bloom.FilterPolicy(10),
 		}},
 	}
+	opts.Experimental.IngestSSTablesAsFlushable = true
 	opts.EnsureDefaults()
 	return opts
 }
diff --git a/mem_table.go b/mem_table.go
index 21088346..c24220eb 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -210,6 +210,8 @@ func (m *memTable) apply(batch *Batch, seqNum uint64) error {
 			// Don't increment seqNum for LogData, since these are not applied
 			// to the memtable.
 			seqNum--
+		case InternalKeyKindIngestSST:
+			panic("pebble: cannot apply ingested sstable key kind to memtable")
 		default:
 			err = ins.Add(&m.skl, ikey, value)
 		}
diff --git a/metrics.go b/metrics.go
index abc585e5..2a8d68d9 100644
--- a/metrics.go
+++ b/metrics.go
@@ -184,6 +184,9 @@ type Metrics struct {
 		// The total number of flushes.
 		Count           int64
 		WriteThroughput ThroughputMetric
+		// Number of flushes that are in-progress. In the current implementation
+		// this will always be zero or one.
+		NumInProgress int64
 	}
 
 	Filter FilterMetrics
diff --git a/objstorage/provider.go b/objstorage/provider.go
index bfd58cc6..12fd1264 100644
--- a/objstorage/provider.go
+++ b/objstorage/provider.go
@@ -7,6 +7,7 @@ package objstorage
 import (
 	"io"
 
+	"github.com/cockroachdb/errors/oserror"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/vfs"
 )
@@ -77,6 +78,10 @@ type Settings struct {
 	// Local filesystem configuration.
 	FS        vfs.FS
 	FSDirName string
+	// Cleaner cleans obsolete files from the local filesystem.
+	//
+	// The default cleaner uses the DeleteCleaner.
+	FSCleaner base.Cleaner
 
 	// NoSyncOnClose decides whether the implementation will enforce a
 	// close-time synchronization (e.g., fdatasync() or sync_file_range())
@@ -97,6 +102,7 @@ func DefaultSettings(fs vfs.FS, dirName string) Settings {
 		Logger:        base.DefaultLogger,
 		FS:            fs,
 		FSDirName:     dirName,
+		FSCleaner:     base.DeleteCleaner{},
 		NoSyncOnClose: false,
 		BytesPerSync:  512 * 1024, // 512KB
 	}
@@ -160,7 +166,13 @@ func (p *Provider) Create(fileType base.FileType, fileNum base.FileNum) (Writabl
 
 // Remove removes an object.
 func (p *Provider) Remove(fileType base.FileType, fileNum base.FileNum) error {
-	return p.st.FS.Remove(p.Path(fileType, fileNum))
+	return p.st.FSCleaner.Clean(p.st.FS, fileType, p.Path(fileType, fileNum))
+}
+
+// IsNotExistError indicates whether the error is known to report that a file or
+// directory does not exist.
+func IsNotExistError(err error) bool {
+	return oserror.IsNotExist(err)
 }
 
 // LinkOrCopyFromLocal creates a new object that is either a copy of a given
diff --git a/open.go b/open.go
index 2886048d..a00b57a0 100644
--- a/open.go
+++ b/open.go
@@ -10,7 +10,10 @@ import (
 	"io"
 	"math"
 	"os"
+	"path/filepath"
 	"sort"
+	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -19,6 +22,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/pebble/internal/manifest"
 	"github.com/cockroachdb/pebble/internal/manual"
 	"github.com/cockroachdb/pebble/internal/rate"
 	"github.com/cockroachdb/pebble/objstorage"
@@ -70,6 +74,7 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 		Logger:        opts.Logger,
 		FS:            opts.FS,
 		FSDirName:     dirname,
+		FSCleaner:     opts.Cleaner,
 		NoSyncOnClose: opts.NoSyncOnClose,
 		BytesPerSync:  opts.BytesPerSync,
 	})
@@ -368,13 +373,15 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 	})
 
 	var ve versionEdit
+	var toFlush flushableList
 	for i, lf := range logFiles {
 		lastWAL := i == len(logFiles)-1
-		maxSeqNum, err := d.replayWAL(jobID, &ve, opts.FS,
+		flush, maxSeqNum, err := d.replayWAL(jobID, &ve, opts.FS,
 			opts.FS.PathJoin(d.walDirname, lf.name), lf.num, strictWALTail && !lastWAL)
 		if err != nil {
 			return nil, err
 		}
+		toFlush = append(toFlush, flush...)
 		d.mu.versions.markFileNumUsed(lf.num)
 		if d.mu.versions.atomic.logSeqNum < maxSeqNum {
 			d.mu.versions.atomic.logSeqNum = maxSeqNum
@@ -402,6 +409,10 @@ func Open(dirname string, opts *Options) (db *DB, _ error) {
 			return nil, err
 		}
 
+		for _, entry := range toFlush {
+			entry.readerUnrefLocked(true)
+		}
+
 		newLogName := base.MakeFilepath(opts.FS, d.walDirname, fileTypeLog, newLogNum)
 		d.mu.log.queue = append(d.mu.log.queue, fileInfo{fileNum: newLogNum, fileSize: 0})
 		logFile, err := opts.FS.Create(newLogName)
@@ -591,25 +602,31 @@ func GetVersion(dir string, fs vfs.FS) (string, error) {
 	return version, nil
 }
 
-// replayWAL replays the edits in the specified log file.
+// replayWAL replays the edits in the specified log file. If the DB is in
+// read only mode, then the WALs are replayed into memtables and not flushed. If
+// the DB is not in read only mode, then the contents of the WAL are guaranteed
+// to be flushed.
+//
+// The toFlush return value is a list of flushables associated with the WAL
+// being replayed which will be flushed. Once the version edit has been applied
+// to the manifest, it is up to the caller of replayWAL to unreference the
+// toFlush flushables returned by replayWAL.
 //
 // d.mu must be held when calling this, but the mutex may be dropped and
 // re-acquired during the course of this method.
 func (d *DB) replayWAL(
 	jobID int, ve *versionEdit, fs vfs.FS, filename string, logNum FileNum, strictWALTail bool,
-) (maxSeqNum uint64, err error) {
+) (toFlush flushableList, maxSeqNum uint64, err error) {
 	file, err := fs.Open(filename)
 	if err != nil {
-		return 0, err
+		return nil, 0, err
 	}
 	defer file.Close()
-
 	var (
 		b               Batch
 		buf             bytes.Buffer
 		mem             *memTable
 		entry           *flushableEntry
-		toFlush         flushableList
 		rr              = record.NewReader(file, logNum)
 		offset          int64 // byte offset in rr
 		lastFlushOffset int64
@@ -653,6 +670,23 @@ func (d *DB) replayWAL(
 			d.mu.mem.queue = append(d.mu.mem.queue, entry)
 		}
 	}
+
+	// updateVE is used to update ve with information about new files created
+	// during the flush of any flushable not of type ingestedFlushable. For the
+	// flushable of type ingestedFlushable we use custom handling below.
+	updateVE := func() error {
+		// TODO(bananabrick): See if we can use the actual base level here,
+		// instead of using 1.
+		c := newFlush(d.opts, d.mu.versions.currentVersion(),
+			1 /* base level */, toFlush)
+		newVE, _, err := d.runCompaction(jobID, c)
+		if err != nil {
+			return err
+		}
+		ve.NewFiles = append(ve.NewFiles, newVE.NewFiles...)
+		return nil
+	}
+
 	for {
 		offset = rr.Offset()
 		r, err := rr.Next()
@@ -670,11 +704,11 @@ func (d *DB) replayWAL(
 			} else if record.IsInvalidRecord(err) && !strictWALTail {
 				break
 			}
-			return 0, errors.Wrap(err, "pebble: error when replaying WAL")
+			return nil, 0, errors.Wrap(err, "pebble: error when replaying WAL")
 		}
 
 		if buf.Len() < batchHeaderLen {
-			return 0, base.CorruptionErrorf("pebble: corrupt log file %q (num %s)",
+			return nil, 0, base.CorruptionErrorf("pebble: corrupt log file %q (num %s)",
 				filename, errors.Safe(logNum))
 		}
 
@@ -685,6 +719,102 @@ func (d *DB) replayWAL(
 		seqNum := b.SeqNum()
 		maxSeqNum = seqNum + uint64(b.Count())
 
+		{
+			br := b.Reader()
+			if kind, path, _, _ := br.Next(); kind == InternalKeyKindIngestSST {
+				paths := make([]string, b.Count())
+				fileNums := make([]FileNum, b.Count())
+				addPath := func(path []byte, i int) {
+					paths[i] = string(path)
+					// TODO(bananabrick): Store the filenums in the batch as a
+					// value, so that we don't have to perform this custom
+					// parsing here.
+					fileNum, err := strconv.Atoi(
+						strings.TrimSuffix(filepath.Base(string(path)),
+							filepath.Ext(string(path))),
+					)
+					if err != nil {
+						panic("pebble: sstable file path is invalid.")
+					}
+					fileNums[i] = FileNum(fileNum)
+				}
+				addPath(path, 0)
+
+				for i := 1; i < int(b.Count()); i++ {
+					kind, path, _, ok := br.Next()
+					if kind != InternalKeyKindIngestSST {
+						panic("pebble: invalid batch key kind.")
+					}
+					if !ok {
+						panic("pebble: invalid batch count.")
+					}
+					addPath(path, i)
+				}
+
+				if _, _, _, ok := br.Next(); ok {
+					panic("pebble: invalid number of entries in batch.")
+				}
+
+				var meta []*manifest.FileMetadata
+				meta, _, err = ingestLoad(
+					d.opts, d.mu.formatVers.vers, paths, d.cacheID, fileNums,
+				)
+				if err != nil {
+					return nil, 0, err
+				}
+
+				if uint32(len(meta)) != b.Count() {
+					panic("pebble: couldn't load all files in WAL entry.")
+				}
+
+				entry, err = d.newIngestedFlushableEntry(
+					meta, seqNum, logNum,
+				)
+				if err != nil {
+					return nil, 0, err
+				}
+
+				if d.opts.ReadOnly {
+					d.mu.mem.queue = append(d.mu.mem.queue, entry)
+					// We added the IngestSST flushable to the queue. But there
+					// must be at least one WAL entry waiting to be replayed. We
+					// have to ensure this newer WAL entry isn't replayed into
+					// the current value of d.mu.mem.mutable because the current
+					// mutable memtable exists before this flushable entry in
+					// the memtable queue. To ensure this, we just need to unset
+					// d.mu.mem.mutable. When a newer WAL is replayed, we will
+					// set d.mu.mem.mutable to a newer value.
+					d.mu.mem.mutable = nil
+				} else {
+					toFlush = append(toFlush, entry)
+					// During WAL replay, the lsm only has L0, hence, the
+					// baseLevel is 1. For the sake of simplicity, we place the
+					// ingested files in L0 here, instead of finding their
+					// target levels. This is a simplification for the sake of
+					// simpler code. It is expected that WAL replay should be
+					// rare, and that flushables of type ingestedFlushable
+					// should also be rare. So, placing the ingested files in L0
+					// is alright.
+					//
+					// TODO(bananabrick): Maybe refactor this function to allow
+					// us to easily place ingested files in levels as low as
+					// possible during WAL replay. It would require breaking up
+					// the application of ve to the manifest into chunks and is
+					// not pretty w/o a refactor to this function and how it's
+					// used.
+					c := newFlush(
+						d.opts, d.mu.versions.currentVersion(),
+						1, /* base level */
+						[]*flushableEntry{entry},
+					)
+					for _, file := range c.flushing[0].flushable.(*ingestedFlushable).files {
+						ve.NewFiles = append(ve.NewFiles, newFileEntry{Level: 0, Meta: file})
+					}
+				}
+				return toFlush, maxSeqNum, nil
+			}
+		}
+
 		if b.memTableSize >= uint64(d.largeBatchThreshold) {
 			flushMem()
 			// Make a copy of the data slice since it is currently owned by buf and will
@@ -703,7 +833,7 @@ func (d *DB) replayWAL(
 		} else {
 			ensureMem(seqNum)
 			if err = mem.prepare(&b); err != nil && err != arenaskl.ErrArenaFull {
-				return 0, err
+				return nil, 0, err
 			}
 			// We loop since DB.newMemTable() slowly grows the size of allocated memtables, so the
 			// batch may not initially fit, but will eventually fit (since it is smaller than
@@ -713,11 +843,11 @@ func (d *DB) replayWAL(
 				ensureMem(seqNum)
 				err = mem.prepare(&b)
 				if err != nil && err != arenaskl.ErrArenaFull {
-					return 0, err
+					return nil, 0, err
 				}
 			}
 			if err = mem.apply(&b, seqNum); err != nil {
-				return 0, err
+				return nil, 0, err
 			}
 			mem.writerUnref()
 		}
@@ -726,18 +856,12 @@ func (d *DB) replayWAL(
 	flushMem()
 	// mem is nil here.
 	if !d.opts.ReadOnly {
-		c := newFlush(d.opts, d.mu.versions.currentVersion(),
-			1 /* base level */, toFlush)
-		newVE, _, err := d.runCompaction(jobID, c)
+		err = updateVE()
 		if err != nil {
-			return 0, err
-		}
-		ve.NewFiles = append(ve.NewFiles, newVE.NewFiles...)
-		for i := range toFlush {
-			toFlush[i].readerUnref()
+			return nil, 0, err
 		}
 	}
-	return maxSeqNum, err
+	return toFlush, maxSeqNum, err
 }
 
 func checkOptions(opts *Options, path string) (strictWALTail bool, err error) {
diff --git a/options.go b/options.go
index 482018fb..c5016b06 100644
--- a/options.go
+++ b/options.go
@@ -615,6 +615,15 @@ type Options struct {
 		// Any change in exclusion behavior takes effect only on future written
 		// sstables, and does not start rewriting existing sstables.
 		RequiredInPlaceValueBound UserKeyPrefixBound
+
+		// IngestSSTablesAsFlushable is used to determine if ingested sstables
+		// should be ingested as a flushable. By default this is false, but it
+		// is true for all metamorphic tests.
+		//
+		// TODO(bananabrick): Remove this field and enable by default once
+		// https://github.com/cockroachdb/pebble/issues/2292 and
+		// https://github.com/cockroachdb/pebble/issues/2266 are closed.
+		IngestSSTablesAsFlushable bool
 	}
 
 	// Filters is a map from filter policy name to filter policy. It is used for
@@ -965,14 +974,7 @@ func (o *Options) EnsureDefaults() *Options {
 	}
 
 	if o.FS == nil {
-		o.FS, o.private.fsCloser = vfs.WithDiskHealthChecks(vfs.Default, 5*time.Second,
-			func(name string, op vfs.OpType, duration time.Duration) {
-				o.EventListener.DiskSlow(DiskSlowInfo{
-					Path:     name,
-					OpType:   op,
-					Duration: duration,
-				})
-			})
+		o.WithFSDefaults()
 	}
 	if o.FlushSplitBytes <= 0 {
 		o.FlushSplitBytes = 2 * o.Levels[0].TargetFileSize
@@ -1004,6 +1006,23 @@ func (o *Options) EnsureDefaults() *Options {
 	return o
 }
 
+// WithFSDefaults configures the Options to wrap the configured filesystem with
+// the default virtual file system middleware, like disk-health checking.
+func (o *Options) WithFSDefaults() *Options {
+	if o.FS == nil {
+		o.FS = vfs.Default
+	}
+	o.FS, o.private.fsCloser = vfs.WithDiskHealthChecks(o.FS, 5*time.Second,
+		func(name string, op vfs.OpType, duration time.Duration) {
+			o.EventListener.DiskSlow(DiskSlowInfo{
+				Path:     name,
+				OpType:   op,
+				Duration: duration,
+			})
+		})
+	return o
+}
+
 // AddEventListener adds the provided event listener to the Options, in addition
 // to any existing event listener.
 func (o *Options) AddEventListener(l EventListener) {
diff --git a/read_state.go b/read_state.go
index 67d5a7ea..a3898171 100644
--- a/read_state.go
+++ b/read_state.go
@@ -40,7 +40,7 @@ func (s *readState) unref() {
 	}
 	s.current.Unref()
 	for _, mem := range s.memtables {
-		mem.readerUnref()
+		mem.readerUnref(true)
 	}
 
 	// The last reference to the readState was released. Check to see if there
@@ -58,7 +58,7 @@ func (s *readState) unrefLocked() {
 	}
 	s.current.UnrefLocked()
 	for _, mem := range s.memtables {
-		mem.readerUnref()
+		mem.readerUnrefLocked(true)
 	}
 
 	// NB: Unlike readState.unref(), we don't attempt to cleanup newly obsolete
@@ -78,8 +78,7 @@ func (d *DB) loadReadState() *readState {
 
 // updateReadStateLocked creates a new readState from the current version and
 // list of memtables. Requires DB.mu is held. If checker is not nil, it is
-// called after installing the new readState. If atomicFunc is not nil, it is
-// executed atomically with the transition to the new read state.
+// called after installing the new readState.
 func (d *DB) updateReadStateLocked(checker func(*DB) error) {
 	s := &readState{
 		db:        d,
diff --git a/replay/replay.go b/replay/replay.go
index 26112fd8..a40f25d6 100644
--- a/replay/replay.go
+++ b/replay/replay.go
@@ -55,14 +55,14 @@ func computeReadAmp(v *manifest.Version) int {
 // PaceByFixedReadAmp to wait on the dbMetricsNotifier condition variable if the
 // read amplification observed is greater than the specified target (refRAmp).
 func waitForReadAmpLE(r *Runner, rAmp int) {
-	r.dbMetricsNotifier.L.Lock()
+	r.dbMetricsCond.L.Lock()
 	m := r.dbMetrics
 	ra := m.ReadAmp()
 	for ra > rAmp {
-		r.dbMetricsNotifier.Wait()
+		r.dbMetricsCond.Wait()
 		ra = r.dbMetrics.ReadAmp()
 	}
-	r.dbMetricsNotifier.L.Unlock()
+	r.dbMetricsCond.L.Unlock()
 }
 
 // Unpaced implements Pacer by applying each new write as soon as possible. It
@@ -187,48 +187,33 @@ type Runner struct {
 	Opts         *pebble.Options
 
 	// Internal state.
-	cancel func()
 
-	// workloadExhausted is a channel that is used by refreshMetrics in order to
-	// receive a single message once all the workload steps have been applied.
-	workloadExhausted chan struct{}
-
-	d        *pebble.DB
-	err      atomic.Value
-	errgroup *errgroup.Group
-	metrics  struct {
+	d *pebble.DB
+	// dbMetrics and dbMetricsCond work in unison to update the metrics and
+	// notify (broadcast) to any waiting clients that metrics have been updated.
+	dbMetrics     *pebble.Metrics
+	dbMetricsCond sync.Cond
+	cancel        func()
+	err           atomic.Value
+	errgroup      *errgroup.Group
+	readerOpts    sstable.ReaderOptions
+	stagingDir    string
+	steps         chan workloadStep
+	stepsApplied  chan workloadStep
+
+	metrics struct {
 		writeStalls             uint64
 		writeStallsDurationNano uint64
 	}
-	readerOpts sstable.ReaderOptions
-	stagingDir string
-	steps      chan workloadStep
-
-	// compactionEnded is a channel used to notify the refreshMetrics method that
-	// a compaction has finished and hence it can update the metrics.
-	compactionEnded            chan struct{}
-	finishedCompactionNotifier sync.Cond
-
-	// compactionEndedCount keeps track of the cumulative number of compactions.
-	// It is guarded by the finishedCompactionNotifier condition variable lock.
-	compactionEndedCount int64
-
-	// atomicCompactionStartCount keeps track of the cumulative number of
-	// compactions started. It's incremented by a pebble.EventListener's
-	// CompactionStart handler.
-	//
-	// Must be accessed with atomics.
-	atomicCompactionStartCount int64
-
-	// compactionsHaveQuiesced is a channel that is closed once compactions have
-	// quiesced allowing the compactionNotified goroutine to complete.
-	compactionsHaveQuiesced chan struct{}
-
-	// dbMetrics and dbMetricsNotifier work in unison to update the metrics and
-	// notify (broadcast) to any waiting clients that metrics have been updated.
-	dbMetricsNotifier sync.Cond
-	dbMetrics         *pebble.Metrics
-
+	// compactionMu holds state for tracking the number of compactions
+	// started and completed and waking waiting goroutines when a new compaction
+	// completes. See nextCompactionCompletes.
+	compactionMu struct {
+		sync.Mutex
+		ch        chan struct{}
+		started   int64
+		completed int64
+	}
 	workload struct {
 		manifests []string
 		// manifest{Idx,Off} record the starting position of the workload
@@ -263,15 +248,13 @@ func (r *Runner) Run(ctx context.Context) error {
 		return err
 	}
 
-	r.dbMetricsNotifier = sync.Cond{
-		L: &sync.Mutex{},
-	}
-	r.finishedCompactionNotifier = sync.Cond{
+	r.dbMetricsCond = sync.Cond{
 		L: &sync.Mutex{},
 	}
 
 	// Extend the user-provided Options with extensions necessary for replay
 	// mechanics.
+	r.compactionMu.ch = make(chan struct{})
 	r.Opts.AddEventListener(r.eventListener())
 	r.Opts.EnsureDefaults()
 	r.readerOpts = r.Opts.MakeReaderOptions()
@@ -282,74 +265,142 @@ func (r *Runner) Run(ctx context.Context) error {
 	}
 
 	r.dbMetrics = r.d.Metrics()
-	r.workloadExhausted = make(chan struct{}, 1)
-	r.compactionEnded = make(chan struct{})
-	r.compactionsHaveQuiesced = make(chan struct{})
-
 	// Use a buffered channel to allow the prepareWorkloadSteps to read ahead,
 	// buffering up to cap(r.steps) steps ahead of the current applied state.
 	// Flushes need to be buffered and ingested sstables need to be copied, so
 	// pipelining this preparation makes it more likely the step will be ready
 	// to apply when the pacer decides to apply it.
 	r.steps = make(chan workloadStep, 5)
+	r.stepsApplied = make(chan workloadStep, 5)
 
 	ctx, r.cancel = context.WithCancel(ctx)
 	r.errgroup, ctx = errgroup.WithContext(ctx)
 	r.errgroup.Go(func() error { return r.prepareWorkloadSteps(ctx) })
 	r.errgroup.Go(func() error { return r.applyWorkloadSteps(ctx) })
 	r.errgroup.Go(func() error { return r.refreshMetrics(ctx) })
-	r.errgroup.Go(func() error { return r.compactionNotified(ctx) })
 	return nil
 }
 
+// refreshMetrics runs in its own goroutine, collecting metrics from the Pebble
+// instance whenever a) a workload step completes, or b) a compaction completes.
+// The Pacer implementations that pace based on read-amplification rely on these
+// refreshed metrics to decide when to allow the workload to proceed.
 func (r *Runner) refreshMetrics(ctx context.Context) error {
-	workloadExhausted := false
-	done := false
-
-	fetchDoneState := func() {
-		r.finishedCompactionNotifier.L.Lock()
-		done = r.compactionEndedCount == atomic.LoadInt64(&r.atomicCompactionStartCount)
-		r.finishedCompactionNotifier.L.Unlock()
-	}
-
-	for !done || !workloadExhausted {
-		select {
-		// r.workloadExhausted receives a single message when the workload is exhausted
-		case <-r.workloadExhausted:
-			workloadExhausted = true
-			fetchDoneState()
-			if done {
-				r.stopCompactionNotifier()
+	var workloadExhausted bool
+	stepsApplied := r.stepsApplied
+	compactionCount, alreadyCompleted, compactionCh := r.nextCompactionCompletes(0)
+	for {
+		if !alreadyCompleted {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-compactionCh:
+				// Fall through to refreshing dbMetrics.
+			case _, ok := <-stepsApplied:
+				if !ok {
+					workloadExhausted = true
+					// Set the [stepsApplied] channel to nil so that we'll never
+					// hit this case again, and we don't busy loop.
+					stepsApplied = nil
+				}
+				// Fall through to refreshing dbMetrics.
 			}
-		case <-ctx.Done():
-			r.stopCompactionNotifier()
-			return ctx.Err()
-		case <-r.compactionEnded:
-			r.dbMetricsNotifier.L.Lock()
-			r.dbMetrics = r.d.Metrics()
-			r.dbMetricsNotifier.Broadcast()
-			r.dbMetricsNotifier.L.Unlock()
-
-			fetchDoneState()
-			if done && workloadExhausted {
-				r.stopCompactionNotifier()
+		}
+
+		m := r.d.Metrics()
+		r.dbMetricsCond.L.Lock()
+		r.dbMetrics = m
+		r.dbMetricsCond.Broadcast()
+		r.dbMetricsCond.L.Unlock()
+
+		compactionCount, alreadyCompleted, compactionCh = r.nextCompactionCompletes(compactionCount)
+		// Consider whether replaying is complete. There are two necessary
+		// conditions:
+		//
+		//   1. The workload must be exhausted.
+		//   2. Compactions must have quiesced.
+		//
+		// The first condition is simple. The replay tool is responsible for
+		// applying the workload. The goroutine responsible for applying the
+		// workload closes the `stepsApplied` channel after the last step has
+		// been applied, and we'll flip `workloadExhausted` to true.
+		//
+		// The second condition is tricky. The replay tool doesn't control
+		// compactions and doesn't have visibility into whether the compaction
+		// picker is about to schedule a new compaction. We can tell when
+		// compactions are in progress or may be immeninent (eg, flushes in
+		// progress). If it appears that compactions have quiesced, pause for a
+		// fixed duration to see if a new one is scheduled. If not, consider
+		// compactions quiesced.
+		if workloadExhausted && !alreadyCompleted && r.compactionsAppearQuiesced(m) {
+			select {
+			case <-compactionCh:
+				// A new compaction just finished; compactions have not
+				// quiesced.
+				continue
+			case <-time.After(time.Second):
+				// No compactions completed. If it still looks like they've
+				// quiesced according to the metrics, consider them quiesced.
+				if r.compactionsAppearQuiesced(r.d.Metrics()) {
+					return nil
+				}
 			}
 		}
 	}
-	return nil
 }
 
-func (r *Runner) stopCompactionNotifier() {
-	// Close the channel which will cause the compactionNotified goroutine to exit
-	close(r.compactionsHaveQuiesced)
-
-	// Wake up compaction notifier if it is waiting on the
-	// r.finishedCompactionNotifier condition and make sure it doesn't go back
-	// to sleep
-	r.finishedCompactionNotifier.L.Lock()
-	r.compactionEndedCount++
-	r.finishedCompactionNotifier.Broadcast()
-	r.finishedCompactionNotifier.L.Unlock()
+// compactionsAppearQuiesced returns true if the database may have quiesced, and
+// there likely won't be additional compactions scheduled. Detecting quiescence
+// is a bit fraught: The various signals that Pebble makes available are
+// adjusted at different points in the compaction lifecycle, and database
+// mutexes are dropped and acquired between them. This makes it difficult to
+// reliably identify when compactions quiesce.
+//
+// For example, our call to DB.Metrics() may acquire the DB.mu mutex when a
+// compaction has just successfully completed, but before it's managed to
+// schedule the next compaction (DB.mu is dropped while it attempts to acquire
+// the manifest lock).
+func (r *Runner) compactionsAppearQuiesced(m *pebble.Metrics) bool {
+	r.compactionMu.Lock()
+	defer r.compactionMu.Unlock()
+	if m.Flush.NumInProgress > 0 {
+		return false
+	} else if m.Compact.NumInProgress > 0 && r.compactionMu.started != r.compactionMu.completed {
+		return false
+	}
+	return true
+}
+
+// nextCompactionCompletes may be used to be notified when new compactions
+// complete. The caller is responsible for holding on to a monotonically
+// increasing count representing the number of compactions that have been
+// observed, beginning at zero.
+//
+// The caller passes their current count as an argument. If a new compaction has
+// already completed since their provided count, nextCompactionCompletes returns
+// the new count and a true boolean return value. If a new compaction has not
+// yet completed, it returns a channel that will be closed when the next
+// compaction completes. This scheme allows the caller to select{...},
+// performing some action on every compaction completion.
+func (r *Runner) nextCompactionCompletes(
+	lastObserved int64,
+) (count int64, alreadyOccurred bool, ch chan struct{}) {
+	r.compactionMu.Lock()
+	defer r.compactionMu.Unlock()
+
+	if lastObserved < r.compactionMu.completed {
+		// There has already been another compaction since the last one observed
+		// by this caller. Return immediately.
+		return r.compactionMu.completed, true, nil
+	}
+
+	// The last observed compaction is still the most recent compaction.
+	// Return a channel that the caller can wait on to be notified when the
+	// next compaction occurs.
+	if r.compactionMu.ch == nil {
+		r.compactionMu.ch = make(chan struct{})
+	}
+	return lastObserved, false, r.compactionMu.ch
 }
 
 // Wait waits for the workload replay to complete. Wait returns once the entire
@@ -360,11 +411,9 @@ func (r *Runner) Wait() (Metrics, error) {
 		err = storedErr.(error)
 	}
 	pm := r.d.Metrics()
-	var total pebble.LevelMetrics
+	total := pm.Total()
 	var ingestBytesWeighted uint64
 	for l := 0; l < len(pm.Levels); l++ {
-		total.Add(&pm.Levels[l])
-		total.Sublevels += pm.Levels[l].Sublevels
 		ingestBytesWeighted += pm.Levels[l].BytesIngested * uint64(len(pm.Levels)-l-1)
 	}
 
@@ -400,17 +449,14 @@ func (r *Runner) Close() error {
 type workloadStep struct {
 	kind stepKind
 	ve   manifest.VersionEdit
-
 	// a Version describing the state of the LSM *before* the workload was
 	// collected.
 	pv *manifest.Version
 	// a Version describing the state of the LSM when the workload was
 	// collected.
 	v *manifest.Version
-
 	// non-nil for flushStepKind
-	flushBatch *pebble.Batch
-
+	flushBatch     *pebble.Batch
 	tablesToIngest []string
 }
 
@@ -440,13 +486,22 @@ func (r *Runner) eventListener() pebble.EventListener {
 				uint64(time.Since(writeStallBegin).Nanoseconds()))
 		},
 		CompactionBegin: func(_ pebble.CompactionInfo) {
-			atomic.AddInt64(&r.atomicCompactionStartCount, 1)
+			r.compactionMu.Lock()
+			defer r.compactionMu.Unlock()
+			r.compactionMu.started++
 		},
 		CompactionEnd: func(_ pebble.CompactionInfo) {
-			r.finishedCompactionNotifier.L.Lock()
-			defer r.finishedCompactionNotifier.L.Unlock()
-			r.compactionEndedCount++
-			r.finishedCompactionNotifier.Broadcast()
+			// Keep track of the number of compactions that complete and notify
+			// anyone waiting for a compaction to complete. See the function
+			// nextCompactionCompletes for the corresponding receiver side.
+			r.compactionMu.Lock()
+			defer r.compactionMu.Unlock()
+			r.compactionMu.completed++
+			if r.compactionMu.ch != nil {
+				// Signal that a compaction has completed.
+				close(r.compactionMu.ch)
+				r.compactionMu.ch = nil
+			}
 		},
 	}
 	l.EnsureDefaults(nil)
@@ -465,12 +520,12 @@ func (r *Runner) applyWorkloadSteps(ctx context.Context) error {
 		case step, ok = <-r.steps:
 			if !ok {
 				// Exhausted the workload. Exit.
-				r.workloadExhausted <- struct{}{}
+				close(r.stepsApplied)
 				return nil
 			}
 		}
 
-		// TODO(leon,jackson): Make sure to sum the duration statistics
+		// TODO(leon,jackson): Make sure to sum the duration statistics.
 		r.Pacer.pace(r, step)
 
 		switch step.kind {
@@ -478,16 +533,18 @@ func (r *Runner) applyWorkloadSteps(ctx context.Context) error {
 			if err := step.flushBatch.Commit(&pebble.WriteOptions{Sync: false}); err != nil {
 				return err
 			}
+			r.stepsApplied <- step
 		case ingestStepKind:
 			if err := r.d.Ingest(step.tablesToIngest); err != nil {
 				return err
 			}
-
+			r.stepsApplied <- step
 		case compactionStepKind:
-			// TODO(jackson,leon): Apply the step to runner state.
+			// No-op.
+			// TODO(jackson): Should we elide this earlier?
+		default:
+			panic("unreachable")
 		}
-
-		var _ = step
 	}
 }
 
@@ -669,24 +726,6 @@ func (r *Runner) prepareWorkloadSteps(ctx context.Context) error {
 	return nil
 }
 
-func (r *Runner) compactionNotified(ctx context.Context) error {
-	r.finishedCompactionNotifier.L.Lock()
-	for {
-		if r.compactionEndedCount < atomic.LoadInt64(&r.atomicCompactionStartCount) {
-			r.finishedCompactionNotifier.Wait()
-		}
-		r.finishedCompactionNotifier.L.Unlock()
-		select {
-		case <-r.compactionsHaveQuiesced:
-			return nil
-		case <-ctx.Done():
-			return ctx.Err()
-		case r.compactionEnded <- struct{}{}:
-		}
-		r.finishedCompactionNotifier.L.Lock()
-	}
-}
-
 // findWorkloadFiles finds all manifests and tables in the provided path on fs.
 func findWorkloadFiles(
 	path string, fs vfs.FS,
diff --git a/table_cache.go b/table_cache.go
index 981c74bc..e6466302 100644
--- a/table_cache.go
+++ b/table_cache.go
@@ -484,10 +484,16 @@ func (c *tableCacheShard) newRangeKeyIter(
 	// No need to hold onto a ref of the cache value.
 	c.unrefValue(v)
 
-	if err != nil || iter == nil {
+	if err != nil {
 		return nil, err
 	}
 
+	if iter == nil {
+		// NewRawRangeKeyIter can return nil even if there's no error. However,
+		// the keyspan.LevelIter expects a non-nil iterator if err is nil.
+		return emptyKeyspanIter, nil
+	}
+
 	return iter, nil
 }
 
diff --git a/tool/wal.go b/tool/wal.go
index c2e44c07..009aed69 100644
--- a/tool/wal.go
+++ b/tool/wal.go
@@ -137,6 +137,8 @@ func (w *walT) runDump(cmd *cobra.Command, args []string) {
 						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtValue.fn(ukey, value))
 					case base.InternalKeyKindLogData:
 						fmt.Fprintf(stdout, "<%d>", len(value))
+					case base.InternalKeyIngestSST:
+						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
 					case base.InternalKeyKindSingleDelete:
 						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
 					case base.InternalKeyKindSetWithDelete:
diff --git a/version_set.go b/version_set.go
index 1b63e62d..1b625fc6 100644
--- a/version_set.go
+++ b/version_set.go
@@ -607,7 +607,7 @@ func (vs *versionSet) incrementCompactions(kind compactionKind, extraLevels []*c
 		vs.metrics.Compact.Count++
 		vs.metrics.Compact.DefaultCount++
 
-	case compactionKindFlush:
+	case compactionKindFlush, compactionKindIngestedFlushable:
 		vs.metrics.Flush.Count++
 
 	case compactionKindMove:
@@ -754,7 +754,12 @@ func (vs *versionSet) addLiveFileNums(m map[FileNum]struct{}) {
 	}
 }
 
+// DB.mu must be held when addObsoleteLocked is called.
 func (vs *versionSet) addObsoleteLocked(obsolete []*manifest.FileMetadata) {
+	if len(obsolete) == 0 {
+		return
+	}
+
 	for _, fileMeta := range obsolete {
 		// Note that the obsolete tables are no longer zombie by the definition of
 		// zombie, but we leave them in the zombie tables map until they are
@@ -764,12 +769,20 @@ func (vs *versionSet) addObsoleteLocked(obsolete []*manifest.FileMetadata) {
 		}
 	}
 	vs.obsoleteTables = append(vs.obsoleteTables, obsolete...)
-	vs.incrementObsoleteTablesLocked(obsolete)
+	vs.updateObsoleteTableMetricsLocked()
 }
 
-func (vs *versionSet) incrementObsoleteTablesLocked(obsolete []*manifest.FileMetadata) {
-	for _, fileMeta := range obsolete {
-		vs.metrics.Table.ObsoleteCount++
+// addObsolete will acquire DB.mu, so DB.mu must not be held when this is
+// called.
+func (vs *versionSet) addObsolete(obsolete []*manifest.FileMetadata) {
+	vs.mu.Lock()
+	defer vs.mu.Unlock()
+	vs.addObsoleteLocked(obsolete)
+}
+
+func (vs *versionSet) updateObsoleteTableMetricsLocked() {
+	vs.metrics.Table.ObsoleteCount = int64(len(vs.obsoleteTables))
+	for _, fileMeta := range vs.obsoleteTables {
 		vs.metrics.Table.ObsoleteSize += fileMeta.Size
 	}
 }
diff --git a/vfs/disk_health.go b/vfs/disk_health.go
index b85a27be..8b6c7e54 100644
--- a/vfs/disk_health.go
+++ b/vfs/disk_health.go
@@ -280,6 +280,26 @@ func (d *diskHealthCheckingFile) timeDiskOp(opType OpType, op func()) {
 	op()
 }
 
+// diskHealthCheckingDir implements disk-health checking for directories. Unlike
+// other files, we allow directories to receive concurrent write operations
+// (Syncs are the only write operations supported by a directory.) Since the
+// diskHealthCheckingFile's timeDiskOp can only track a single in-flight
+// operation at a time, we time the operation using the filesystem-level
+// timeFilesystemOp function instead.
+type diskHealthCheckingDir struct {
+	File
+	name string
+	fs   *diskHealthCheckingFS
+}
+
+// Sync implements the io.Syncer interface.
+func (d *diskHealthCheckingDir) Sync() (err error) {
+	d.fs.timeFilesystemOp(d.name, OpTypeSync, func() {
+		err = d.File.Sync()
+	})
+	return err
+}
+
 // diskHealthCheckingFS adds disk-health checking facilities to a VFS.
 // It times disk write operations in two ways:
 //
@@ -576,11 +596,11 @@ func (d *diskHealthCheckingFS) OpenDir(name string) (File, error) {
 	}
 	// Directories opened with OpenDir must be opened with health checking,
 	// because they may be explicitly synced.
-	checkingFile := newDiskHealthCheckingFile(f, d.diskSlowThreshold, func(opType OpType, duration time.Duration) {
-		d.onSlowDisk(name, opType, duration)
-	})
-	checkingFile.startTicker()
-	return checkingFile, nil
+	return &diskHealthCheckingDir{
+		File: f,
+		name: name,
+		fs:   d,
+	}, nil
 }
 
 // PathBase implements the FS interface.


Test patch:
diff --git a/batch_test.go b/batch_test.go
index b56168c2..b7bdc5b5 100644
--- a/batch_test.go
+++ b/batch_test.go
@@ -55,6 +55,7 @@ func TestBatch(t *testing.T) {
 	// deferred variants. This is a consequence of these keys' more complex
 	// value encodings.
 	testCases := []testCase{
+		{InternalKeyKindIngestSST, "1.sst", ""},
 		{InternalKeyKindSet, "roses", "red"},
 		{InternalKeyKindSet, "violets", "blue"},
 		{InternalKeyKindDelete, "roses", ""},
@@ -97,6 +98,8 @@ func TestBatch(t *testing.T) {
 			_ = b.LogData([]byte(tc.key), nil)
 		case InternalKeyKindRangeKeyDelete:
 			_ = b.RangeKeyDelete([]byte(tc.key), []byte(tc.value), nil)
+		case InternalKeyKindIngestSST:
+			b.ingestSST([]byte(tc.key))
 		}
 	}
 	verifyTestCases(&b, testCases)
@@ -135,6 +138,8 @@ func TestBatch(t *testing.T) {
 			d.Finish()
 		case InternalKeyKindLogData:
 			_ = b.LogData([]byte(tc.key), nil)
+		case InternalKeyKindIngestSST:
+			b.ingestSST([]byte(tc.key))
 		case InternalKeyKindRangeKeyDelete:
 			d := b.RangeKeyDeleteDeferred(len(key), len(value))
 			copy(d.Key, key)
@@ -145,6 +150,18 @@ func TestBatch(t *testing.T) {
 	verifyTestCases(&b, testCases)
 }
 
+func TestBatchIngestSST(t *testing.T) {
+	// Verify that Batch.IngestSST has the correct batch count and memtable
+	// size.
+	var b Batch
+	b.ingestSST([]byte("1.sst"))
+	require.Equal(t, int(b.Count()), 1)
+	b.ingestSST([]byte("2.sst"))
+	require.Equal(t, int(b.Count()), 2)
+	require.Equal(t, int(b.memTableSize), 0)
+	require.Equal(t, b.ingestedSSTBatch, true)
+}
+
 func TestBatchLen(t *testing.T) {
 	var b Batch
 
diff --git a/cleaner_test.go b/cleaner_test.go
index a7784923..0dc91527 100644
--- a/cleaner_test.go
+++ b/cleaner_test.go
@@ -25,11 +25,11 @@ func TestArchiveCleaner(t *testing.T) {
 
 	var buf syncedBuffer
 	mem := vfs.NewMem()
-	opts := &Options{
+	opts := (&Options{
 		Cleaner: ArchiveCleaner{},
 		FS:      loggingFS{mem, &buf},
 		WALDir:  "wal",
-	}
+	}).WithFSDefaults()
 
 	datadriven.RunTest(t, "testdata/cleaner", func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
diff --git a/cmd/pebble/replay_test.go b/cmd/pebble/replay_test.go
new file mode 100644
index 00000000..576bcf21
--- /dev/null
+++ b/cmd/pebble/replay_test.go
@@ -0,0 +1,58 @@
+// Copyright 2023 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 main
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/pebble"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseOptionsStr(t *testing.T) {
+	type testCase struct {
+		optionsStr string
+		options    *pebble.Options
+	}
+
+	testCases := []testCase{
+		{
+			optionsStr: `[Options] max_concurrent_compactions=9`,
+			options:    &pebble.Options{MaxConcurrentCompactions: func() int { return 9 }},
+		},
+		{
+			optionsStr: `[Options] bytes_per_sync=90000`,
+			options:    &pebble.Options{BytesPerSync: 90000},
+		},
+		{
+			optionsStr: `[Options] [Level "0"] target_file_size=222`,
+			options: &pebble.Options{Levels: []pebble.LevelOptions{
+				{TargetFileSize: 222},
+			}},
+		},
+		{
+			optionsStr: `[Options] lbase_max_bytes=10  max_open_files=20  [Level "0"] target_file_size=30 [Level "1"] index_block_size=40`,
+			options: &pebble.Options{
+				LBaseMaxBytes: 10,
+				MaxOpenFiles:  20,
+				Levels: []pebble.LevelOptions{
+					{TargetFileSize: 30},
+					{IndexBlockSize: 40},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		o := new(pebble.Options)
+		require.NoError(t, parseCustomOptions(tc.optionsStr, o))
+		o.EnsureDefaults()
+		got := o.String()
+
+		tc.options.EnsureDefaults()
+		want := tc.options.String()
+		require.Equal(t, want, got)
+	}
+}
diff --git a/commit_test.go b/commit_test.go
index d4ae4c4d..2a8969dc 100644
--- a/commit_test.go
+++ b/commit_test.go
@@ -184,7 +184,7 @@ func TestCommitPipelineAllocateSeqNum(t *testing.T) {
 	for i := 1; i <= n; i++ {
 		go func(i int) {
 			defer wg.Done()
-			p.AllocateSeqNum(i, func() {
+			p.AllocateSeqNum(i, func(_ uint64) {
 				atomic.AddUint64(&prepareCount, uint64(1))
 			}, func(seqNum uint64) {
 				atomic.AddUint64(&applyCount, uint64(1))
diff --git a/compaction_test.go b/compaction_test.go
index ac81d767..46d7e7e0 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -534,8 +534,7 @@ func TestPickCompaction(t *testing.T) {
 }
 
 func TestElideTombstone(t *testing.T) {
-	opts := &Options{}
-	opts.EnsureDefaults()
+	opts := (&Options{}).EnsureDefaults().WithFSDefaults()
 
 	newFileMeta := func(smallest, largest base.InternalKey) *fileMetadata {
 		m := (&fileMetadata{}).ExtendPointKeyBounds(opts.Comparer.Compare, smallest, largest)
@@ -917,9 +916,7 @@ func (t *cpuPermissionGranter) CPUWorkDone(_ CPUWorkHandle) {
 // the acquired handles are returned.
 func TestCompactionCPUGranter(t *testing.T) {
 	mem := vfs.NewMem()
-	opts := &Options{
-		FS: mem,
-	}
+	opts := (&Options{FS: mem}).WithFSDefaults()
 	g := &cpuPermissionGranter{permit: true}
 	opts.Experimental.CPUWorkPermissionGranter = g
 	d, err := Open("", opts)
@@ -940,9 +937,7 @@ func TestCompactionCPUGranter(t *testing.T) {
 // Tests that there's no errors or panics when the default CPU granter is used.
 func TestCompactionCPUGranterDefault(t *testing.T) {
 	mem := vfs.NewMem()
-	opts := &Options{
-		FS: mem,
-	}
+	opts := (&Options{FS: mem}).WithFSDefaults()
 	d, err := Open("", opts)
 	if err != nil {
 		t.Fatalf("Open: %v", err)
@@ -970,7 +965,7 @@ func TestCompaction(t *testing.T) {
 		DebugCheck:            DebugCheckLevels,
 		L0CompactionThreshold: 8,
 	}
-	opts.testingRandomized()
+	opts.testingRandomized().WithFSDefaults()
 	d, err := Open("", opts)
 	if err != nil {
 		t.Fatalf("Open: %v", err)
@@ -1275,11 +1270,11 @@ func TestManualCompaction(t *testing.T) {
 		mem = vfs.NewMem()
 		require.NoError(t, mem.MkdirAll("ext", 0755))
 
-		opts := &Options{
+		opts := (&Options{
 			FS:                 mem,
 			DebugCheck:         DebugCheckLevels,
 			FormatMajorVersion: randVersion(minVersion, maxVersion),
-		}
+		}).WithFSDefaults()
 		opts.DisableAutomaticCompactions = true
 
 		var err error
@@ -1373,12 +1368,12 @@ func TestManualCompaction(t *testing.T) {
 				}
 
 				mem = vfs.NewMem()
-				opts := &Options{
+				opts := (&Options{
 					FS:                          mem,
 					DebugCheck:                  DebugCheckLevels,
 					FormatMajorVersion:          randVersion(minVersion, maxVersion),
 					DisableAutomaticCompactions: true,
-				}
+				}).WithFSDefaults()
 
 				var err error
 				if d, err = runDBDefineCmd(td, opts); err != nil {
@@ -1905,7 +1900,7 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 				return nil, err
 			}
 		}
-		opts := &Options{
+		opts := (&Options{
 			FS:         vfs.NewMem(),
 			DebugCheck: DebugCheckLevels,
 			EventListener: &EventListener{
@@ -1917,7 +1912,7 @@ func TestCompactionDeleteOnlyHints(t *testing.T) {
 				},
 			},
 			FormatMajorVersion: FormatNewest,
-		}
+		}).WithFSDefaults()
 
 		// Collection of table stats can trigger compactions. As we want full
 		// control over when compactions are run, disable stats by default.
@@ -2175,7 +2170,7 @@ func TestCompactionTombstones(t *testing.T) {
 						return err.Error()
 					}
 				}
-				opts := &Options{
+				opts := (&Options{
 					FS:         vfs.NewMem(),
 					DebugCheck: DebugCheckLevels,
 					EventListener: &EventListener{
@@ -2184,7 +2179,7 @@ func TestCompactionTombstones(t *testing.T) {
 						},
 					},
 					FormatMajorVersion: FormatNewest,
-				}
+				}).WithFSDefaults()
 				var err error
 				d, err = runDBDefineCmd(td, opts)
 				if err != nil {
@@ -2383,7 +2378,7 @@ func TestCompactionReadTriggered(t *testing.T) {
 						return err.Error()
 					}
 				}
-				opts := &Options{
+				opts := (&Options{
 					FS:         vfs.NewMem(),
 					DebugCheck: DebugCheckLevels,
 					EventListener: &EventListener{
@@ -2391,7 +2386,7 @@ func TestCompactionReadTriggered(t *testing.T) {
 							compactInfo = &info
 						},
 					},
-				}
+				}).WithFSDefaults()
 				var err error
 				d, err = runDBDefineCmd(td, opts)
 				if err != nil {
@@ -2853,7 +2848,7 @@ func TestCompactionErrorCleanup(t *testing.T) {
 
 	mem := vfs.NewMem()
 	ii := errorfs.OnIndex(math.MaxInt32) // start disabled
-	opts := &Options{
+	opts := (&Options{
 		FS:     errorfs.Wrap(mem, ii),
 		Levels: make([]LevelOptions, numLevels),
 		EventListener: &EventListener{
@@ -2871,7 +2866,7 @@ func TestCompactionErrorCleanup(t *testing.T) {
 				}
 			},
 		},
-	}
+	}).WithFSDefaults()
 	for i := range opts.Levels {
 		opts.Levels[i].TargetFileSize = 1
 	}
@@ -3157,7 +3152,7 @@ func TestFlushInvariant(t *testing.T) {
 							},
 						},
 						DebugCheck: DebugCheckLevels,
-					}))
+					}).WithFSDefaults())
 					require.NoError(t, err)
 
 					require.NoError(t, d.Set([]byte("hello"), nil, NoSync))
@@ -3208,7 +3203,7 @@ func TestCompactFlushQueuedMemTableAndFlushMetrics(t *testing.T) {
 	mem := vfs.NewMem()
 	d, err := Open("", testingRandomized(&Options{
 		FS: mem,
-	}))
+	}).WithFSDefaults())
 	require.NoError(t, err)
 
 	// Add the key "a" to the memtable, then fill up the memtable with the key
@@ -3267,7 +3262,7 @@ func TestCompactFlushQueuedLargeBatch(t *testing.T) {
 	mem := vfs.NewMem()
 	d, err := Open("", testingRandomized(&Options{
 		FS: mem,
-	}))
+	}).WithFSDefaults())
 	require.NoError(t, err)
 
 	// The default large batch threshold is slightly less than 1/2 of the
@@ -3301,7 +3296,7 @@ func TestCompactFlushQueuedLargeBatch(t *testing.T) {
 func TestCleanerCond(t *testing.T) {
 	d, err := Open("", testingRandomized(&Options{
 		FS: vfs.NewMem(),
-	}))
+	}).WithFSDefaults())
 	require.NoError(t, err)
 
 	for i := 0; i < 10; i++ {
@@ -3359,7 +3354,7 @@ func TestFlushError(t *testing.T) {
 				t.Log(err)
 			},
 		},
-	}))
+	}).WithFSDefaults())
 	require.NoError(t, err)
 	require.NoError(t, d.Set([]byte("a"), []byte("foo"), NoSync))
 	require.NoError(t, d.Flush())
@@ -3406,7 +3401,7 @@ func TestAdjustGrandparentOverlapBytesForFlush(t *testing.T) {
 func TestCompactionInvalidBounds(t *testing.T) {
 	db, err := Open("", testingRandomized(&Options{
 		FS: vfs.NewMem(),
-	}))
+	}).WithFSDefaults())
 	require.NoError(t, err)
 	defer db.Close()
 	require.NoError(t, db.Compact([]byte("a"), []byte("b"), false))
@@ -3654,7 +3649,7 @@ func TestMarkedForCompaction(t *testing.T) {
 	}()
 
 	var buf bytes.Buffer
-	opts := &Options{
+	opts := (&Options{
 		FS:                          mem,
 		DebugCheck:                  DebugCheckLevels,
 		DisableAutomaticCompactions: true,
@@ -3668,7 +3663,7 @@ func TestMarkedForCompaction(t *testing.T) {
 				fmt.Fprintln(&buf, info)
 			},
 		},
-	}
+	}).WithFSDefaults()
 
 	reset := func() {
 		if d != nil {
@@ -3823,7 +3818,7 @@ func TestCompaction_LogAndApplyFails(t *testing.T) {
 				// NB: we hold db.mu here.
 				var cur *flushableEntry
 				cur, db.mu.mem.queue = db.mu.mem.queue[0], db.mu.mem.queue[1:]
-				cur.readerUnref()
+				cur.readerUnrefLocked(true)
 			},
 		},
 		{
@@ -3836,7 +3831,7 @@ func TestCompaction_LogAndApplyFails(t *testing.T) {
 		var db *DB
 		inj := &createManifestErrorInjector{}
 		logger := &fatalCapturingLogger{}
-		opts := &Options{
+		opts := (&Options{
 			FS: errorfs.Wrap(vfs.NewMem(), inj),
 			// Rotate the manifest after each write. This is required to trigger a
 			// file creation, into which errors can be injected.
@@ -3850,7 +3845,7 @@ func TestCompaction_LogAndApplyFails(t *testing.T) {
 				},
 			},
 			DisableAutomaticCompactions: true,
-		}
+		}).WithFSDefaults()
 
 		db, err := Open("", opts)
 		require.NoError(t, err)
diff --git a/db_test.go b/db_test.go
index c091e490..a52a83e7 100644
--- a/db_test.go
+++ b/db_test.go
@@ -851,7 +851,9 @@ func TestMemTableReservationLeak(t *testing.T) {
 	d.mu.Lock()
 	last := d.mu.mem.queue[len(d.mu.mem.queue)-1]
 	last.readerRef()
-	defer last.readerUnref()
+	defer func() {
+		last.readerUnref(true)
+	}()
 	d.mu.Unlock()
 	if err := d.Close(); err == nil {
 		t.Fatalf("expected failure, but found success")
diff --git a/flushable_test.go b/flushable_test.go
new file mode 100644
index 00000000..86510fb3
--- /dev/null
+++ b/flushable_test.go
@@ -0,0 +1,163 @@
+package pebble
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
+)
+
+// Simple sanity tests for the flushable interface implementation for ingested
+// sstables.
+func TestIngestedSSTFlushableAPI(t *testing.T) {
+	var mem vfs.FS
+	var d *DB
+	defer func() {
+		require.NoError(t, d.Close())
+	}()
+	var flushable flushable
+
+	reset := func() {
+		if d != nil {
+			require.NoError(t, d.Close())
+		}
+
+		mem = vfs.NewMem()
+		require.NoError(t, mem.MkdirAll("ext", 0755))
+		opts := &Options{
+			FS:                    mem,
+			L0CompactionThreshold: 100,
+			L0StopWritesThreshold: 100,
+			DebugCheck:            DebugCheckLevels,
+			FormatMajorVersion:    FormatNewest,
+		}
+		// Disable automatic compactions because otherwise we'll race with
+		// delete-only compactions triggered by ingesting range tombstones.
+		opts.DisableAutomaticCompactions = true
+
+		var err error
+		d, err = Open("", opts)
+		require.NoError(t, err)
+		flushable = nil
+	}
+	reset()
+
+	loadFileMeta := func(paths []string) []*fileMetadata {
+		d.mu.Lock()
+		pendingOutputs := make([]FileNum, len(paths))
+		for i := range paths {
+			pendingOutputs[i] = d.mu.versions.getNextFileNum()
+		}
+		jobID := d.mu.nextJobID
+		d.mu.nextJobID++
+		d.mu.Unlock()
+
+		// We can reuse the ingestLoad function for this test even if we're
+		// not actually ingesting a file.
+		meta, paths, err := ingestLoad(
+			d.opts, d.FormatMajorVersion(), paths, d.cacheID, pendingOutputs,
+		)
+		if err != nil {
+			panic(err)
+		}
+		if len(meta) == 0 {
+			// All of the sstables to be ingested were empty. Nothing to do.
+			panic("empty sstable")
+		}
+
+		// Verify the sstables do not overlap.
+		if err := ingestSortAndVerify(d.cmp, meta, paths); err != nil {
+			panic("unsorted sstables")
+		}
+
+		// Hard link the sstables into the DB directory. Since the sstables aren't
+		// referenced by a version, they won't be used. If the hard linking fails
+		// (e.g. because the files reside on a different filesystem), ingestLink will
+		// fall back to copying, and if that fails we undo our work and return an
+		// error.
+		if _, err := ingestLink(jobID, d.opts, d.objProvider, paths, meta); err != nil {
+			panic("couldn't hard link sstables")
+		}
+
+		// Fsync the directory we added the tables to. We need to do this at some
+		// point before we update the MANIFEST (via logAndApply), otherwise a crash
+		// can have the tables referenced in the MANIFEST, but not present in the
+		// directory.
+		if err := d.dataDir.Sync(); err != nil {
+			panic("Couldn't sync data directory")
+		}
+
+		return meta
+	}
+
+	datadriven.RunTest(t, "testdata/ingested_flushable_api", func(t *testing.T, td *datadriven.TestData) string {
+		switch td.Cmd {
+		case "reset":
+			reset()
+			return ""
+		case "build":
+			if err := runBuildCmd(td, d, mem); err != nil {
+				return err.Error()
+			}
+			return ""
+		case "flushable":
+			// Creates an ingestedFlushable over the input files.
+			paths := make([]string, 0, len(td.CmdArgs))
+			for _, arg := range td.CmdArgs {
+				paths = append(paths, arg.String())
+			}
+
+			meta := loadFileMeta(paths)
+			flushable = newIngestedFlushable(
+				meta, d.cmp, d.split, d.newIters, d.tableNewRangeKeyIter,
+			)
+			return ""
+		case "iter":
+			iter := flushable.newIter(nil)
+			var buf bytes.Buffer
+			for x, _ := iter.First(); x != nil; x, _ = iter.Next() {
+				buf.WriteString(x.String())
+				buf.WriteString("\n")
+			}
+			iter.Close()
+			return buf.String()
+		case "rangekeyIter":
+			iter := flushable.newRangeKeyIter(nil)
+			var buf bytes.Buffer
+			if iter != nil {
+				for span := iter.First(); span != nil; span = iter.Next() {
+					buf.WriteString(span.String())
+					buf.WriteString("\n")
+				}
+				iter.Close()
+			}
+			return buf.String()
+		case "rangedelIter":
+			iter := flushable.newRangeDelIter(nil)
+			var buf bytes.Buffer
+			if iter != nil {
+				for span := iter.First(); span != nil; span = iter.Next() {
+					buf.WriteString(span.String())
+					buf.WriteString("\n")
+				}
+				iter.Close()
+			}
+			return buf.String()
+		case "readyForFlush":
+			if flushable.readyForFlush() {
+				return "true"
+			}
+			return "false"
+		case "containsRangeKey":
+			if flushable.containsRangeKeys() {
+				return "true"
+			}
+			return "false"
+		default:
+			return fmt.Sprintf("unknown command: %s", td.Cmd)
+		}
+	})
+}
diff --git a/format_major_version_test.go b/format_major_version_test.go
index 24edc847..ce24c0ae 100644
--- a/format_major_version_test.go
+++ b/format_major_version_test.go
@@ -32,7 +32,7 @@ func TestFormatMajorVersion_MigrationDefined(t *testing.T) {
 
 func TestRatchetFormat(t *testing.T) {
 	fs := vfs.NewMem()
-	d, err := Open("", &Options{FS: fs})
+	d, err := Open("", (&Options{FS: fs}).WithFSDefaults())
 	require.NoError(t, err)
 	require.NoError(t, d.Set([]byte("foo"), []byte("bar"), Sync))
 	require.Equal(t, FormatMostCompatible, d.FormatMajorVersion())
@@ -63,7 +63,7 @@ func TestRatchetFormat(t *testing.T) {
 
 	// If we Open the database again, leaving the default format, the
 	// database should Open using the persisted FormatNewest.
-	d, err = Open("", &Options{FS: fs})
+	d, err = Open("", (&Options{FS: fs}).WithFSDefaults())
 	require.NoError(t, err)
 	require.Equal(t, FormatNewest, d.FormatMajorVersion())
 	require.NoError(t, d.Close())
@@ -74,10 +74,10 @@ func TestRatchetFormat(t *testing.T) {
 	require.NoError(t, m.Move("999999"))
 	require.NoError(t, m.Close())
 
-	_, err = Open("", &Options{
+	_, err = Open("", (&Options{
 		FS:                 fs,
 		FormatMajorVersion: FormatVersioned,
-	})
+	}).WithFSDefaults())
 	require.Error(t, err)
 	require.EqualError(t, err, `pebble: database "" written in format major version 999999`)
 }
@@ -108,10 +108,10 @@ func TestFormatMajorVersions(t *testing.T) {
 	for vers := FormatMostCompatible; vers <= FormatNewest; vers++ {
 		t.Run(fmt.Sprintf("vers=%03d", vers), func(t *testing.T) {
 			fs := vfs.NewStrictMem()
-			opts := &Options{
+			opts := (&Options{
 				FS:                 fs,
 				FormatMajorVersion: vers,
-			}
+			}).WithFSDefaults()
 
 			// Create a database at this format major version and perform
 			// some very basic operations.
@@ -255,7 +255,7 @@ func TestSplitUserKeyMigration(t *testing.T) {
 					}
 					buf.Reset()
 				}
-				opts = &Options{
+				opts = (&Options{
 					FormatMajorVersion: FormatBlockPropertyCollector,
 					EventListener: &EventListener{
 						CompactionEnd: func(info CompactionInfo) {
@@ -267,7 +267,7 @@ func TestSplitUserKeyMigration(t *testing.T) {
 						},
 					},
 					DisableAutomaticCompactions: true,
-				}
+				}).WithFSDefaults()
 				var err error
 				if d, err = runDBDefineCmd(td, opts); err != nil {
 					return err.Error()
@@ -363,10 +363,10 @@ func TestPebblev1Migration(t *testing.T) {
 						return fmt.Sprintf("unknown argument: %s", cmd)
 					}
 				}
-				opts := &Options{
+				opts := (&Options{
 					FS:                 vfs.NewMem(),
 					FormatMajorVersion: FormatMajorVersion(version),
-				}
+				}).WithFSDefaults()
 				d, err = Open("", opts)
 				if err != nil {
 					return err.Error()
@@ -486,13 +486,13 @@ func TestPebblev1MigrationRace(t *testing.T) {
 	defer cache.Unref()
 	tableCache := NewTableCache(cache, 1, 5)
 	defer tableCache.Unref()
-	d, err := Open("", &Options{
+	d, err := Open("", (&Options{
 		Cache:              cache,
 		FS:                 vfs.NewMem(),
 		FormatMajorVersion: FormatMajorVersion(FormatPrePebblev1Marked - 1),
 		TableCache:         tableCache,
 		Levels:             []LevelOptions{{TargetFileSize: 1}},
-	})
+	}).WithFSDefaults())
 	require.NoError(t, err)
 	defer d.Close()
 
@@ -531,7 +531,7 @@ func TestPebblev1MigrationRace(t *testing.T) {
 // Regression test for #2044, where multiple concurrent compactions can lead
 // to an indefinite wait on the compaction goroutine in compactMarkedFilesLocked.
 func TestPebblev1MigrationConcurrencyRace(t *testing.T) {
-	opts := &Options{
+	opts := (&Options{
 		Comparer:           testkeys.Comparer,
 		FS:                 vfs.NewMem(),
 		FormatMajorVersion: FormatSplitUserKeysMarked,
@@ -539,7 +539,7 @@ func TestPebblev1MigrationConcurrencyRace(t *testing.T) {
 		MaxConcurrentCompactions: func() int {
 			return 4
 		},
-	}
+	}).WithFSDefaults()
 	func() {
 		d, err := Open("", opts)
 		require.NoError(t, err)
diff --git a/ingest_test.go b/ingest_test.go
index 0993d5cb..c1fbed6b 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -112,10 +112,10 @@ func TestIngestLoad(t *testing.T) {
 			}
 			w.Close()
 
-			opts := &Options{
+			opts := (&Options{
 				Comparer: DefaultComparer,
 				FS:       mem,
-			}
+			}).WithFSDefaults()
 			meta, _, err := ingestLoad(opts, dbVersion, []string{"ext"}, 0, []FileNum{1})
 			if err != nil {
 				return err.Error()
@@ -198,10 +198,10 @@ func TestIngestLoadRand(t *testing.T) {
 		}()
 	}
 
-	opts := &Options{
+	opts := (&Options{
 		Comparer: DefaultComparer,
 		FS:       mem,
-	}
+	}).WithFSDefaults()
 	meta, _, err := ingestLoad(opts, version, paths, 0, pending)
 	require.NoError(t, err)
 
@@ -219,10 +219,10 @@ func TestIngestLoadInvalid(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, f.Close())
 
-	opts := &Options{
+	opts := (&Options{
 		Comparer: DefaultComparer,
 		FS:       mem,
-	}
+	}).WithFSDefaults()
 	if _, _, err := ingestLoad(opts, FormatNewest, []string{"invalid"}, 0, []FileNum{1}); err == nil {
 		t.Fatalf("expected error, but found success")
 	}
@@ -287,11 +287,10 @@ func TestIngestLink(t *testing.T) {
 	const count = 10
 	for i := 0; i <= count; i++ {
 		t.Run("", func(t *testing.T) {
-			mem := vfs.NewMem()
-			opts := &Options{FS: mem}
-			opts.EnsureDefaults()
-			require.NoError(t, mem.MkdirAll(dir, 0755))
-			objProvider := objstorage.New(objstorage.DefaultSettings(mem, dir))
+			opts := &Options{FS: vfs.NewMem()}
+			opts.EnsureDefaults().WithFSDefaults()
+			require.NoError(t, opts.FS.MkdirAll(dir, 0755))
+			objProvider := objstorage.New(objstorage.DefaultSettings(opts.FS, dir))
 
 			paths := make([]string, 10)
 			meta := make([]*fileMetadata, len(paths))
@@ -300,7 +299,7 @@ func TestIngestLink(t *testing.T) {
 				paths[j] = fmt.Sprintf("external%d", j)
 				meta[j] = &fileMetadata{}
 				meta[j].FileNum = FileNum(j)
-				f, err := mem.Create(paths[j])
+				f, err := opts.FS.Create(paths[j])
 				require.NoError(t, err)
 
 				contents[j] = []byte(fmt.Sprintf("data%d", j))
@@ -312,10 +311,10 @@ func TestIngestLink(t *testing.T) {
 			}
 
 			if i < count {
-				mem.Remove(paths[i])
+				opts.FS.Remove(paths[i])
 			}
 
-			err := ingestLink(0 /* jobID */, opts, objProvider, paths, meta)
+			_, err := ingestLink(0 /* jobID */, opts, objProvider, paths, meta)
 			if i < count {
 				if err == nil {
 					t.Fatalf("expected error, but found success")
@@ -324,7 +323,7 @@ func TestIngestLink(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			files, err := mem.List(dir)
+			files, err := opts.FS.List(dir)
 			require.NoError(t, err)
 
 			sort.Strings(files)
@@ -339,7 +338,7 @@ func TestIngestLink(t *testing.T) {
 					t.Fatalf("expected %d files, but found:\n%s", count, strings.Join(files, "\n"))
 				}
 				for j := range files {
-					ftype, fileNum, ok := base.ParseFilename(mem, files[j])
+					ftype, fileNum, ok := base.ParseFilename(opts.FS, files[j])
 					if !ok {
 						t.Fatalf("unable to parse filename: %s", files[j])
 					}
@@ -349,7 +348,7 @@ func TestIngestLink(t *testing.T) {
 					if FileNum(j) != fileNum {
 						t.Fatalf("expected table %d, but found %d", j, fileNum)
 					}
-					f, err := mem.Open(mem.PathJoin(dir, files[j]))
+					f, err := opts.FS.Open(opts.FS.PathJoin(dir, files[j]))
 					require.NoError(t, err)
 
 					data, err := io.ReadAll(f)
@@ -372,11 +371,12 @@ func TestIngestLinkFallback(t *testing.T) {
 	require.NoError(t, err)
 
 	opts := &Options{FS: errorfs.Wrap(mem, errorfs.OnIndex(0))}
-	opts.EnsureDefaults()
+	opts.EnsureDefaults().WithFSDefaults()
 	objProvider := objstorage.New(objstorage.DefaultSettings(opts.FS, ""))
 
 	meta := []*fileMetadata{{FileNum: 1}}
-	require.NoError(t, ingestLink(0, opts, objProvider, []string{"source"}, meta))
+	_, err = ingestLink(0, opts, objProvider, []string{"source"}, meta)
+	require.NoError(t, err)
 
 	dest, err := mem.Open("000001.sst")
 	require.NoError(t, err)
@@ -391,6 +391,173 @@ func TestIngestLinkFallback(t *testing.T) {
 	}
 }
 
+func TestOverlappingIngestedSSTs(t *testing.T) {
+	dir := ""
+	var (
+		mem        vfs.FS
+		d          *DB
+		opts       *Options
+		closed     = false
+		blockFlush = false
+	)
+	defer func() {
+		if !closed {
+			require.NoError(t, d.Close())
+		}
+	}()
+
+	reset := func(strictMem bool) {
+		if d != nil && !closed {
+			require.NoError(t, d.Close())
+		}
+		blockFlush = false
+
+		if strictMem {
+			mem = vfs.NewStrictMem()
+		} else {
+			mem = vfs.NewMem()
+		}
+
+		require.NoError(t, mem.MkdirAll("ext", 0755))
+		opts = (&Options{
+			FS:                          mem,
+			MemTableStopWritesThreshold: 4,
+			L0CompactionThreshold:       100,
+			L0StopWritesThreshold:       100,
+			DebugCheck:                  DebugCheckLevels,
+			FormatMajorVersion:          FormatNewest,
+		}).WithFSDefaults()
+		// Disable automatic compactions because otherwise we'll race with
+		// delete-only compactions triggered by ingesting range tombstones.
+		opts.DisableAutomaticCompactions = true
+		opts.Experimental.IngestSSTablesAsFlushable = true
+
+		var err error
+		d, err = Open(dir, opts)
+		require.NoError(t, err)
+	}
+	waitForFlush := func() {
+		if d == nil {
+			return
+		}
+		d.mu.Lock()
+		for d.mu.compact.flushing {
+			d.mu.compact.cond.Wait()
+		}
+		d.mu.Unlock()
+	}
+	reset(false)
+
+	datadriven.RunTest(t, "testdata/flushable_ingest", func(t *testing.T, td *datadriven.TestData) string {
+		switch td.Cmd {
+		case "reset":
+			var strictMem bool
+			if len(td.CmdArgs) == 1 && td.CmdArgs[0].String() == "strictMem" {
+				strictMem = true
+			}
+			reset(strictMem)
+			return ""
+
+		case "ignoreSyncs":
+			var ignoreSyncs bool
+			if len(td.CmdArgs) == 1 && td.CmdArgs[0].String() == "true" {
+				ignoreSyncs = true
+			}
+			mem.(*vfs.MemFS).SetIgnoreSyncs(ignoreSyncs)
+			return ""
+
+		case "resetToSynced":
+			mem.(*vfs.MemFS).ResetToSyncedState()
+			files, err := mem.List(dir)
+			sort.Strings(files)
+			require.NoError(t, err)
+			return strings.Join(files, "\n")
+
+		case "batch":
+			b := d.NewIndexedBatch()
+			if err := runBatchDefineCmd(td, b); err != nil {
+				return err.Error()
+			}
+			if err := b.Commit(nil); err != nil {
+				return err.Error()
+			}
+			return ""
+
+		case "build":
+			if err := runBuildCmd(td, d, mem); err != nil {
+				return err.Error()
+			}
+			return ""
+
+		case "ingest":
+			if err := runIngestCmd(td, d, mem); err != nil {
+				return err.Error()
+			}
+			if !blockFlush {
+				waitForFlush()
+			}
+			return ""
+
+		case "iter":
+			iter := d.NewIter(nil)
+			return runIterCmd(td, iter, true)
+
+		case "lsm":
+			return runLSMCmd(td, d)
+
+		case "close":
+			if closed {
+				return "already closed"
+			}
+			require.NoError(t, d.Close())
+			closed = true
+			return ""
+
+		case "ls":
+			files, err := mem.List(dir)
+			sort.Strings(files)
+			require.NoError(t, err)
+			return strings.Join(files, "\n")
+
+		case "open":
+			if len(td.CmdArgs) == 1 && td.CmdArgs[0].String() == "readOnly" {
+				opts.ReadOnly = true
+			}
+			var err error
+			d, err = Open(dir, opts)
+			closed = false
+			require.NoError(t, err)
+			waitForFlush()
+			return ""
+
+		case "blockFlush":
+			blockFlush = true
+			d.mu.Lock()
+			d.mu.compact.flushing = true
+			d.mu.Unlock()
+			return ""
+
+		case "allowFlush":
+			blockFlush = false
+			d.mu.Lock()
+			d.mu.compact.flushing = false
+			d.mu.Unlock()
+			return ""
+
+		case "flush":
+			d.maybeScheduleFlush()
+			waitForFlush()
+			return ""
+
+		case "get":
+			return runGetCmd(td, d)
+
+		default:
+			return fmt.Sprintf("unknown command: %s", td.Cmd)
+		}
+	})
+}
+
 func TestIngestMemtableOverlaps(t *testing.T) {
 	comparers := []Comparer{
 		{Name: "default", Compare: DefaultComparer.Compare, FormatKey: DefaultComparer.FormatKey},
@@ -446,7 +613,7 @@ func TestIngestMemtableOverlaps(t *testing.T) {
 					opts := &Options{
 						Comparer: &comparer,
 					}
-					opts.EnsureDefaults()
+					opts.EnsureDefaults().WithFSDefaults()
 					if len(d.CmdArgs) > 1 {
 						return fmt.Sprintf("%s expects at most 1 argument", d.Cmd)
 					}
@@ -575,6 +742,7 @@ func TestIngestTargetLevel(t *testing.T) {
 			opts := Options{
 				FormatMajorVersion: FormatNewest,
 			}
+			opts.WithFSDefaults()
 			if d, err = runDBDefineCmd(td, &opts); err != nil {
 				return err.Error()
 			}
@@ -649,6 +817,7 @@ func TestIngest(t *testing.T) {
 		// Disable automatic compactions because otherwise we'll race with
 		// delete-only compactions triggered by ingesting range tombstones.
 		opts.DisableAutomaticCompactions = true
+		opts.Experimental.IngestSSTablesAsFlushable = true
 
 		var err error
 		d, err = Open("", opts)
diff --git a/internal/base/internal_test.go b/internal/base/internal_test.go
index 8c3b7f72..fa369fff 100644
--- a/internal/base/internal_test.go
+++ b/internal/base/internal_test.go
@@ -41,7 +41,7 @@ func TestInvalidInternalKey(t *testing.T) {
 		"\x01\x02\x03\x04\x05\x06\x07",
 		"foo",
 		"foo\x08\x07\x06\x05\x04\x03\x02",
-		"foo\x16\x07\x06\x05\x04\x03\x02\x01",
+		"foo\x17\x07\x06\x05\x04\x03\x02\x01",
 	}
 	for _, tc := range testCases {
 		k := DecodeInternalKey([]byte(tc))
diff --git a/internal/keyspan/level_iter_test.go b/internal/keyspan/level_iter_test.go
index 0161122b..0236040e 100644
--- a/internal/keyspan/level_iter_test.go
+++ b/internal/keyspan/level_iter_test.go
@@ -306,7 +306,10 @@ func TestLevelIterEquivalence(t *testing.T) {
 			b.Added[6] = metas
 			v, _, err := b.Apply(nil, base.DefaultComparer.Compare, base.DefaultFormatter, 0, 0)
 			require.NoError(t, err)
-			levelIter.Init(SpanIterOptions{}, base.DefaultComparer.Compare, tableNewIters, v.Levels[6].Iter(), 0, manifest.KeyTypeRange)
+			levelIter.Init(
+				SpanIterOptions{}, base.DefaultComparer.Compare, tableNewIters,
+				v.Levels[6].Iter(), 0, manifest.KeyTypeRange,
+			)
 			levelIters = append(levelIters, &levelIter)
 		}
 
@@ -423,7 +426,7 @@ func TestLevelIter(t *testing.T) {
 			}
 			if iter == nil {
 				var lastFileNum base.FileNum
-				tableNewIters := func(file *manifest.FileMetadata, iterOptions *SpanIterOptions) (FragmentIterator, error) {
+				tableNewIters := func(file *manifest.FileMetadata, _ *SpanIterOptions) (FragmentIterator, error) {
 					keyType := keyType
 					spans := level[file.FileNum-1]
 					if keyType == manifest.KeyTypePoint {
@@ -436,7 +439,10 @@ func TestLevelIter(t *testing.T) {
 				b.Added[6] = metas
 				v, _, err := b.Apply(nil, base.DefaultComparer.Compare, base.DefaultFormatter, 0, 0)
 				require.NoError(t, err)
-				iter = newLevelIter(SpanIterOptions{}, base.DefaultComparer.Compare, tableNewIters, v.Levels[6].Iter(), 6, keyType)
+				iter = NewLevelIter(
+					SpanIterOptions{}, base.DefaultComparer.Compare,
+					tableNewIters, v.Levels[6].Iter(), 6, keyType,
+				)
 				extraInfo = func() string {
 					return fmt.Sprintf("file = %s.sst", lastFileNum)
 				}
diff --git a/internal/metamorphic/meta_test.go b/internal/metamorphic/meta_test.go
index b03617d2..150b14a8 100644
--- a/internal/metamorphic/meta_test.go
+++ b/internal/metamorphic/meta_test.go
@@ -163,6 +163,8 @@ func testMetaRun(t *testing.T, runDir string, seed uint64, historyPath string) {
 			opts.FS = vfs.NewMem()
 		}
 	}
+	opts.WithFSDefaults()
+
 	threads := testOpts.threads
 	if *maxThreads < threads {
 		threads = *maxThreads
diff --git a/iterator_histories_test.go b/iterator_histories_test.go
index ea6298c8..80bfc94d 100644
--- a/iterator_histories_test.go
+++ b/iterator_histories_test.go
@@ -52,6 +52,7 @@ func TestIterHistories(t *testing.T) {
 			}
 			opts.DisableAutomaticCompactions = true
 			opts.EnsureDefaults()
+			opts.WithFSDefaults()
 
 			for _, cmdArg := range td.CmdArgs {
 				switch cmdArg.Key {
diff --git a/replay/replay_test.go b/replay/replay_test.go
index e0a1753d..75532ce0 100644
--- a/replay/replay_test.go
+++ b/replay/replay_test.go
@@ -6,11 +6,14 @@ import (
 	"encoding/hex"
 	"fmt"
 	"io"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -18,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/datatest"
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/rangekey"
 	"github.com/cockroachdb/pebble/vfs"
@@ -436,3 +440,111 @@ func listFiles(t *testing.T, fs vfs.FS, w io.Writer, name string) {
 		fmt.Fprintf(w, "  %s\n", dirent)
 	}
 }
+
+// TestCompactionsQuiesce replays a workload that produces a nontrivial number of
+// compactions several times. It's intended to exercise Waits termination, which
+// is dependent on compactions quiescing.
+func TestCompactionsQuiesce(t *testing.T) {
+	const replayCount = 1
+	workloadFS := getHeavyWorkload(t)
+	fs := vfs.NewMem()
+	var atomicDone [replayCount]uint32
+	for i := 0; i < replayCount; i++ {
+		func(i int) {
+			runDir := fmt.Sprintf("run%d", i)
+			require.NoError(t, fs.MkdirAll(runDir, os.ModePerm))
+			r := Runner{
+				RunDir:       runDir,
+				WorkloadFS:   workloadFS,
+				WorkloadPath: "workload",
+				Pacer:        Unpaced{},
+				Opts: &pebble.Options{
+					Comparer:           testkeys.Comparer,
+					FS:                 fs,
+					FormatMajorVersion: pebble.FormatNewest,
+					LBaseMaxBytes:      1,
+				},
+			}
+			r.Opts.Experimental.LevelMultiplier = 2
+			require.NoError(t, r.Run(context.Background()))
+			defer r.Close()
+
+			var m Metrics
+			var err error
+			go func() {
+				m, err = r.Wait()
+				atomic.StoreUint32(&atomicDone[i], 1)
+			}()
+
+			wait := 30 * time.Second
+			if invariants.Enabled {
+				wait = time.Minute
+			}
+
+			// The above call to [Wait] should eventually return. [Wait] blocks
+			// until the workload has replayed AND compactions have quiesced. A
+			// bug in either could prevent [Wait] from ever returning.
+			require.Eventually(t, func() bool { return atomic.LoadUint32(&atomicDone[i]) == 1 },
+				wait, time.Millisecond, "(*replay.Runner).Wait didn't terminate")
+			require.NoError(t, err)
+			// Require at least 5 compactions.
+			require.Greater(t, m.Final.Compact.Count, int64(5))
+			require.Equal(t, int64(0), m.Final.Compact.NumInProgress)
+			for l := 0; l < len(m.Final.Levels)-1; l++ {
+				require.Less(t, m.Final.Levels[l].Score, 1.0)
+			}
+		}(i)
+	}
+}
+
+// getHeavyWorkload returns a FS containing a workload in the `workload`
+// directory that flushes enough randomly generated keys that replaying it
+// should generate a non-trivial number of compactions.
+func getHeavyWorkload(t *testing.T) vfs.FS {
+	heavyWorkload.Once.Do(func() {
+		t.Run("buildHeavyWorkload", func(t *testing.T) {
+			heavyWorkload.fs = buildHeavyWorkload(t)
+		})
+	})
+	return heavyWorkload.fs
+}
+
+var heavyWorkload struct {
+	sync.Once
+	fs vfs.FS
+}
+
+func buildHeavyWorkload(t *testing.T) vfs.FS {
+	o := &pebble.Options{
+		Comparer:           testkeys.Comparer,
+		FS:                 vfs.NewMem(),
+		FormatMajorVersion: pebble.FormatNewest,
+	}
+	wc := NewWorkloadCollector("")
+	wc.Attach(o)
+	d, err := pebble.Open("", o)
+	require.NoError(t, err)
+
+	destFS := vfs.NewMem()
+	require.NoError(t, destFS.MkdirAll("workload", os.ModePerm))
+	wc.Start(destFS, "workload")
+
+	ks := testkeys.Alpha(5)
+	var bufKey = make([]byte, ks.MaxLen())
+	var bufVal [512]byte
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	for i := 0; i < 100; i++ {
+		b := d.NewBatch()
+		for j := 0; j < 1000; j++ {
+			rng.Read(bufVal[:])
+			n := testkeys.WriteKey(bufKey[:], ks, rng.Intn(ks.Count()))
+			require.NoError(t, b.Set(bufKey[:n], bufVal[:], pebble.NoSync))
+		}
+		require.NoError(t, b.Commit(pebble.NoSync))
+		require.NoError(t, d.Flush())
+	}
+	wc.Stop()
+
+	defer d.Close()
+	return destFS
+}
diff --git a/testdata/flushable_ingest b/testdata/flushable_ingest
new file mode 100644
index 00000000..ff51856d
--- /dev/null
+++ b/testdata/flushable_ingest
@@ -0,0 +1,627 @@
+# We create multiple SSTs, one of which overlaps with the memtable (scheduling a flush).
+# Check that the SSTs get ingested to the lowest levels possible.
+
+batch
+set a 0
+----
+
+# The SST below overlaps with memtable and thus should be placed in L0
+# post flush.
+build ext1
+set a 1
+----
+
+# The SST below overlaps with nothing and thus should be placed in L6 post
+# flush.
+build ext2
+set b 1
+----
+
+# The SST below doesn't overlap with any SSTs in the LSM and thus can be placed
+# in L6 post-flush.
+build ext3
+set d 1
+----
+
+# We block the flush, so the SSTs should still be in the flushable queue.
+blockFlush
+----
+
+ingest ext1 ext2 ext3
+----
+
+allowFlush
+----
+
+lsm
+----
+
+get
+a
+b
+d
+----
+a:1
+b:1
+d:1
+
+# We expect 1 WAL for an immutable memtable, 1 file for the ingested ssts,
+# one for the mutable memtable. We also expect 3 ssts corresponding to the
+# ingested files.
+ls
+----
+000002.log
+000004.sst
+000005.sst
+000006.sst
+000007.log
+000008.log
+CURRENT
+LOCK
+MANIFEST-000001
+OPTIONS-000003
+ext
+marker.format-version.000011.012
+marker.manifest.000001.MANIFEST-000001
+
+# Test basic WAL replay
+close
+----
+
+# In this case only the flushable was holding a reference to the sstables. Even
+# after the DB is closed, those sstables should still be hanging around.
+ls
+----
+000002.log
+000004.sst
+000005.sst
+000006.sst
+000007.log
+000008.log
+CURRENT
+LOCK
+MANIFEST-000001
+OPTIONS-000003
+ext
+marker.format-version.000011.012
+marker.manifest.000001.MANIFEST-000001
+
+open
+----
+
+# Make sure that the sstables got flushed in the correct order on a WAL replay.
+lsm
+----
+0.1:
+  000004:[a#2,SET-a#2,SET]
+0.0:
+  000009:[a#1,SET-a#1,SET]
+  000005:[b#3,SET-b#3,SET]
+  000006:[d#4,SET-d#4,SET]
+
+get
+a
+b
+d
+----
+a:1
+b:1
+d:1
+
+reset
+----
+
+# Repeat the steps above without closing Pebble. Note that the final lsm state
+# will be different because WAL replay just placed the files in L0.
+batch
+set a 0
+----
+
+build ext1
+set a 1
+----
+
+build ext2
+set b 1
+----
+
+build ext3
+set d 1
+----
+
+ingest ext1 ext2 ext3
+----
+
+lsm
+----
+0.1:
+  000004:[a#2,SET-a#2,SET]
+0.0:
+  000009:[a#1,SET-a#1,SET]
+6:
+  000005:[b#3,SET-b#3,SET]
+  000006:[d#4,SET-d#4,SET]
+
+reset
+----
+
+# Test multiple overlapping ingests interleaving batch sets, and then flushing.
+batch
+set a 0
+----
+
+build ext4
+set a 1
+----
+
+build ext5
+set a 2
+----
+
+blockFlush
+----
+
+ingest ext4
+----
+
+allowFlush
+----
+
+get
+a
+----
+a:1
+
+batch
+set b 1
+----
+
+get
+a
+b
+----
+a:1
+b:1
+
+# Should get ingested into L0 above the memtable flush.
+blockFlush
+----
+
+ingest ext5
+----
+
+allowFlush
+----
+
+get
+a
+b
+----
+a:2
+b:1
+
+batch
+set c 1
+----
+
+flush
+----
+
+lsm
+----
+0.2:
+  000007:[a#4,SET-a#4,SET]
+0.1:
+  000004:[a#2,SET-a#2,SET]
+0.0:
+  000010:[a#1,SET-a#1,SET]
+  000011:[b#3,SET-b#3,SET]
+
+# Value of a should be the value of a in the second ingested SST.
+get
+a
+b
+c
+----
+a:2
+b:1
+c:1
+
+# Test that non-overlapping ingest still works normally.
+reset
+----
+
+batch
+set a 0
+----
+
+build ext1
+set b 1
+----
+
+build ext2
+set d 1
+----
+
+ingest ext1 ext2
+----
+
+lsm
+----
+6:
+  000004:[b#2,SET-b#2,SET]
+  000005:[d#3,SET-d#3,SET]
+
+
+# Verify target level of ingestedFlushable.
+reset
+----
+
+batch
+set a 0
+----
+
+build ext1
+set a 1
+----
+
+build ext2
+set b 1
+----
+
+build ext3
+set d 1
+----
+
+ingest ext1 ext2 ext3
+----
+
+lsm
+----
+0.1:
+  000004:[a#2,SET-a#2,SET]
+0.0:
+  000009:[a#1,SET-a#1,SET]
+6:
+  000005:[b#3,SET-b#3,SET]
+  000006:[d#4,SET-d#4,SET]
+
+
+batch
+set a 3
+----
+
+build ext4
+set a 4
+----
+
+build ext5
+set b 5
+----
+
+ingest ext4 ext5
+----
+
+# Looking for the sstable with the key a to go into 0.3, and the sstable with
+# key b to go into 0.0. The sstable doesn't go into L5, because L5 isn't open
+# yet.
+lsm
+----
+0.3:
+  000010:[a#6,SET-a#6,SET]
+0.2:
+  000014:[a#5,SET-a#5,SET]
+0.1:
+  000004:[a#2,SET-a#2,SET]
+0.0:
+  000009:[a#1,SET-a#1,SET]
+  000011:[b#7,SET-b#7,SET]
+6:
+  000005:[b#3,SET-b#3,SET]
+  000006:[d#4,SET-d#4,SET]
+
+# Testing whether the new mutable memtable with data is flushed correctly during
+# WAL replay.
+reset
+----
+
+batch
+set a 0
+----
+
+# The SST below overlaps with memtable and thus should be placed in L0
+# post flush.
+build ext1
+set a 1
+----
+
+# The SST below overlaps with nothing and thus should be placed in L6 post
+# flush.
+build ext2
+set b 1
+----
+
+# The SST below doesn't overlap with any SSTs in the LSM and thus can be placed
+# in L6 post-flush.
+build ext3
+set d 1
+----
+
+# We block the flush, so the SSTs should still be in the flushable queue.
+blockFlush
+----
+
+ingest ext1 ext2 ext3
+----
+
+# Add another write which should go to the new mutable memtable.
+batch
+set f 1
+----
+
+allowFlush
+----
+
+lsm
+----
+
+get
+a
+b
+d
+f
+----
+a:1
+b:1
+d:1
+f:1
+
+# We expect 1 WAL for an immutable memtable, 1 file for the ingested ssts,
+# one for the mutable memtable. We also expect 3 ssts corresponding to the
+# ingested files.
+ls
+----
+000002.log
+000004.sst
+000005.sst
+000006.sst
+000007.log
+000008.log
+CURRENT
+LOCK
+MANIFEST-000001
+OPTIONS-000003
+ext
+marker.format-version.000011.012
+marker.manifest.000001.MANIFEST-000001
+
+close
+----
+
+# In this case only the memtable was holding a reference to the sstables. Even
+# after the DB is closed, those memtables should still be hanging around.
+ls
+----
+000002.log
+000004.sst
+000005.sst
+000006.sst
+000007.log
+000008.log
+CURRENT
+LOCK
+MANIFEST-000001
+OPTIONS-000003
+ext
+marker.format-version.000011.012
+marker.manifest.000001.MANIFEST-000001
+
+open
+----
+
+# Make sure that the sstables got flushed in the correct order on a WAL replay.
+lsm
+----
+0.1:
+  000004:[a#2,SET-a#2,SET]
+0.0:
+  000009:[a#1,SET-a#1,SET]
+  000005:[b#3,SET-b#3,SET]
+  000006:[d#4,SET-d#4,SET]
+  000010:[f#5,SET-f#5,SET]
+
+# Check if the new mutable memtable is using a new log file, and that the
+# previous log files have been deleted appropriately after the flush.
+ls
+----
+000004.sst
+000005.sst
+000006.sst
+000009.sst
+000010.sst
+000011.log
+CURRENT
+LOCK
+MANIFEST-000001
+MANIFEST-000012
+OPTIONS-000013
+ext
+marker.format-version.000011.012
+marker.manifest.000002.MANIFEST-000012
+
+# Make sure that the new mutable memtable can accept writes.
+batch
+set h 2
+----
+
+get
+h
+----
+h:2
+
+# Test correct WAL replay with read only mode. We essentially want to make sure
+# that once a flushable is added to the queue, we create a new mutable memtable
+# on top of the flushable. Otherwise, we can invert sequence number invariants.
+reset
+----
+
+batch
+set a 0
+----
+
+# The SST below overlaps with memtable and thus should be placed in L0
+# post flush.
+build ext1
+set a 1
+----
+
+# The SST below overlaps with nothing and thus should be placed in L6 post
+# flush.
+build ext2
+set b 1
+----
+
+# The SST below doesn't overlap with any SSTs in the LSM and thus can be placed
+# in L6 post-flush.
+build ext3
+set d 1
+----
+
+# We block the flush, so the SSTs should still be in the flushable queue.
+blockFlush
+----
+
+ingest ext1 ext2 ext3
+----
+
+# Add another write which should go to the new mutable memtable.
+batch
+set a 3
+----
+
+allowFlush
+----
+
+lsm
+----
+
+get
+a
+b
+d
+----
+a:3
+b:1
+d:1
+
+close
+----
+
+open readOnly
+----
+
+get
+a
+b
+d
+----
+a:3
+b:1
+d:1
+
+# Test with StrictFS
+reset strictMem
+----
+
+batch
+set a 1
+set b 1
+----
+
+build ext1
+set a 2
+set b 2
+----
+
+blockFlush
+----
+
+ingest ext1
+----
+
+get
+a
+b
+----
+a:2
+b:2
+
+ignoreSyncs true
+----
+
+lsm
+----
+
+allowFlush
+----
+
+flush
+----
+
+# The ingested file is placed above the sstable generated by memtable flush. The
+# ingested file has a lower file number, but a higher sequence number as
+# expected.
+lsm
+----
+0.1:
+  000004:[a#3,SET-b#3,SET]
+0.0:
+  000007:[a#1,SET-b#2,SET]
+
+ls
+----
+000002.log
+000004.sst
+000005.log
+000006.log
+000007.sst
+CURRENT
+LOCK
+MANIFEST-000001
+OPTIONS-000003
+ext
+marker.format-version.000011.012
+marker.manifest.000001.MANIFEST-000001
+
+close
+----
+
+# At this point, the changes to the manifest should be lost. Note that 7.sst
+# is gone because that file was never synced.
+resetToSynced
+----
+000002.log
+000004.sst
+000005.log
+000006.log
+CURRENT
+LOCK
+MANIFEST-000001
+OPTIONS-000003
+ext
+ext1
+marker.format-version.000011.012
+marker.manifest.000001.MANIFEST-000001
+
+ignoreSyncs false
+----
+
+open
+----
+
+lsm
+----
+0.1:
+  000004:[a#3,SET-b#3,SET]
+0.0:
+  000007:[a#1,SET-b#2,SET]
diff --git a/testdata/ingest b/testdata/ingest
index 4a0b7a4f..e33199fd 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -240,7 +240,7 @@ lsm
   000011:[k#8,SET-k#8,SET]
 0.0:
   000007:[a#2,SET-b#2,DEL]
-  000013:[j#6,SET-k#7,SET]
+  000014:[j#6,SET-k#7,SET]
 6:
   000006:[a#1,SET-b#1,SET]
   000010:[x#5,SET-y#5,SET]
@@ -281,10 +281,10 @@ lsm
   000011:[k#8,SET-k#8,SET]
 0.0:
   000007:[a#2,SET-b#2,DEL]
-  000013:[j#6,SET-k#7,SET]
+  000014:[j#6,SET-k#7,SET]
 6:
   000006:[a#1,SET-b#1,SET]
-  000014:[n#10,SET-n#10,SET]
+  000015:[n#10,SET-n#10,SET]
   000010:[x#5,SET-y#5,SET]
 
 get
@@ -306,7 +306,7 @@ memtable flushed
 lsm
 ----
 0.3:
-  000015:[a#11,RANGEDEL-z#inf,RANGEDEL]
+  000016:[a#11,RANGEDEL-z#inf,RANGEDEL]
 0.2:
   000009:[b#4,MERGE-c#4,DEL]
 0.1:
@@ -314,11 +314,11 @@ lsm
   000011:[k#8,SET-k#8,SET]
 0.0:
   000007:[a#2,SET-b#2,DEL]
-  000013:[j#6,SET-k#7,SET]
-  000017:[m#9,SET-m#9,SET]
+  000014:[j#6,SET-k#7,SET]
+  000019:[m#9,SET-m#9,SET]
 6:
   000006:[a#1,SET-b#1,SET]
-  000014:[n#10,SET-n#10,SET]
+  000015:[n#10,SET-n#10,SET]
   000010:[x#5,SET-y#5,SET]
 
 get
@@ -343,7 +343,7 @@ x: pebble: not found
 y: pebble: not found
 
 wait-pending-table-stats
-000015
+000016
 ----
 num-entries: 2
 num-deletions: 2
@@ -384,10 +384,10 @@ ingest ext9
 lsm
 ----
 0.4:
-  000019:[a#13,SET-g#13,SET]
-  000018:[j#12,RANGEDEL-m#12,SET]
+  000021:[a#13,SET-g#13,SET]
+  000020:[j#12,RANGEDEL-m#12,SET]
 0.3:
-  000015:[a#11,RANGEDEL-z#inf,RANGEDEL]
+  000016:[a#11,RANGEDEL-z#inf,RANGEDEL]
 0.2:
   000009:[b#4,MERGE-c#4,DEL]
 0.1:
@@ -395,11 +395,11 @@ lsm
   000011:[k#8,SET-k#8,SET]
 0.0:
   000007:[a#2,SET-b#2,DEL]
-  000013:[j#6,SET-k#7,SET]
-  000017:[m#9,SET-m#9,SET]
+  000014:[j#6,SET-k#7,SET]
+  000019:[m#9,SET-m#9,SET]
 6:
   000006:[a#1,SET-b#1,SET]
-  000014:[n#10,SET-n#10,SET]
+  000015:[n#10,SET-n#10,SET]
   000010:[x#5,SET-y#5,SET]
 
 # Overlap with sst boundary containing range del sentinel (fileNum 000015) is not considered an overlap since
@@ -437,10 +437,10 @@ d:40
 lsm
 ----
 0.4:
-  000019:[a#13,SET-g#13,SET]
-  000018:[j#12,RANGEDEL-m#12,SET]
+  000021:[a#13,SET-g#13,SET]
+  000020:[j#12,RANGEDEL-m#12,SET]
 0.3:
-  000015:[a#11,RANGEDEL-z#inf,RANGEDEL]
+  000016:[a#11,RANGEDEL-z#inf,RANGEDEL]
 0.2:
   000009:[b#4,MERGE-c#4,DEL]
 0.1:
@@ -448,15 +448,15 @@ lsm
   000011:[k#8,SET-k#8,SET]
 0.0:
   000007:[a#2,SET-b#2,DEL]
-  000013:[j#6,SET-k#7,SET]
-  000017:[m#9,SET-m#9,SET]
+  000014:[j#6,SET-k#7,SET]
+  000019:[m#9,SET-m#9,SET]
 6:
   000006:[a#1,SET-b#1,SET]
-  000021:[d#14,SET-d#14,SET]
-  000022:[i#15,RANGEDEL-j#inf,RANGEDEL]
-  000014:[n#10,SET-n#10,SET]
+  000023:[d#14,SET-d#14,SET]
+  000024:[i#15,RANGEDEL-j#inf,RANGEDEL]
+  000015:[n#10,SET-n#10,SET]
   000010:[x#5,SET-y#5,SET]
-  000020:[z#16,SET-z#16,SET]
+  000022:[z#16,SET-z#16,SET]
 
 # No overlap between fileNum 000019 that contains point key f, since f is ingested file's range del sentinel.
 
@@ -470,10 +470,10 @@ ingest ext13
 lsm
 ----
 0.4:
-  000019:[a#13,SET-g#13,SET]
-  000018:[j#12,RANGEDEL-m#12,SET]
+  000021:[a#13,SET-g#13,SET]
+  000020:[j#12,RANGEDEL-m#12,SET]
 0.3:
-  000015:[a#11,RANGEDEL-z#inf,RANGEDEL]
+  000016:[a#11,RANGEDEL-z#inf,RANGEDEL]
 0.2:
   000009:[b#4,MERGE-c#4,DEL]
 0.1:
@@ -481,16 +481,16 @@ lsm
   000011:[k#8,SET-k#8,SET]
 0.0:
   000007:[a#2,SET-b#2,DEL]
-  000013:[j#6,SET-k#7,SET]
-  000017:[m#9,SET-m#9,SET]
+  000014:[j#6,SET-k#7,SET]
+  000019:[m#9,SET-m#9,SET]
 6:
   000006:[a#1,SET-b#1,SET]
-  000021:[d#14,SET-d#14,SET]
-  000023:[e#17,RANGEDEL-f#inf,RANGEDEL]
-  000022:[i#15,RANGEDEL-j#inf,RANGEDEL]
-  000014:[n#10,SET-n#10,SET]
+  000023:[d#14,SET-d#14,SET]
+  000025:[e#17,RANGEDEL-f#inf,RANGEDEL]
+  000024:[i#15,RANGEDEL-j#inf,RANGEDEL]
+  000015:[n#10,SET-n#10,SET]
   000010:[x#5,SET-y#5,SET]
-  000020:[z#16,SET-z#16,SET]
+  000022:[z#16,SET-z#16,SET]
 
 # Overlap with range delete keys in memtable, hence memtable will be flushed.
 
@@ -509,14 +509,14 @@ memtable flushed
 lsm
 ----
 0.6:
-  000024:[b#19,SET-b#19,SET]
+  000026:[b#19,SET-b#19,SET]
 0.5:
-  000026:[a#18,RANGEDEL-d#inf,RANGEDEL]
+  000029:[a#18,RANGEDEL-d#inf,RANGEDEL]
 0.4:
-  000019:[a#13,SET-g#13,SET]
-  000018:[j#12,RANGEDEL-m#12,SET]
+  000021:[a#13,SET-g#13,SET]
+  000020:[j#12,RANGEDEL-m#12,SET]
 0.3:
-  000015:[a#11,RANGEDEL-z#inf,RANGEDEL]
+  000016:[a#11,RANGEDEL-z#inf,RANGEDEL]
 0.2:
   000009:[b#4,MERGE-c#4,DEL]
 0.1:
@@ -524,16 +524,16 @@ lsm
   000011:[k#8,SET-k#8,SET]
 0.0:
   000007:[a#2,SET-b#2,DEL]
-  000013:[j#6,SET-k#7,SET]
-  000017:[m#9,SET-m#9,SET]
+  000014:[j#6,SET-k#7,SET]
+  000019:[m#9,SET-m#9,SET]
 6:
   000006:[a#1,SET-b#1,SET]
-  000021:[d#14,SET-d#14,SET]
-  000023:[e#17,RANGEDEL-f#inf,RANGEDEL]
-  000022:[i#15,RANGEDEL-j#inf,RANGEDEL]
-  000014:[n#10,SET-n#10,SET]
+  000023:[d#14,SET-d#14,SET]
+  000025:[e#17,RANGEDEL-f#inf,RANGEDEL]
+  000024:[i#15,RANGEDEL-j#inf,RANGEDEL]
+  000015:[n#10,SET-n#10,SET]
   000010:[x#5,SET-y#5,SET]
-  000020:[z#16,SET-z#16,SET]
+  000022:[z#16,SET-z#16,SET]
 
 reset
 ----
diff --git a/testdata/ingested_flushable_api b/testdata/ingested_flushable_api
new file mode 100644
index 00000000..48f85478
--- /dev/null
+++ b/testdata/ingested_flushable_api
@@ -0,0 +1,112 @@
+build ext1
+set c c
+set e e
+----
+
+flushable ext1
+----
+
+iter
+----
+c#0,1
+e#0,1
+
+rangekeyIter
+----
+
+rangedelIter
+----
+
+containsRangeKey
+----
+false
+
+readyForFlush
+----
+true
+
+reset
+----
+
+build ext2
+range-key-set d g 1 val1
+----
+
+flushable ext2
+----
+
+iter
+----
+
+rangekeyIter
+----
+d-g:{(#0,RANGEKEYSET,1,val1)}
+
+containsRangeKey
+----
+true
+
+rangedelIter
+----
+
+reset
+----
+
+build ext3
+del-range a j
+del-range o z
+----
+
+flushable ext3
+----
+
+iter
+----
+
+rangedelIter
+----
+a-j:{(#0,RANGEDEL)}
+o-z:{(#0,RANGEDEL)}
+
+rangekeyIter
+----
+
+readyForFlush
+----
+true
+
+containsRangeKey
+----
+false
+
+reset
+----
+
+build ext4
+del-range a j
+set k kk
+range-key-set y z 1 val1
+----
+
+flushable ext4
+----
+
+iter
+----
+k#0,1
+
+rangekeyIter
+----
+y-z:{(#0,RANGEKEYSET,1,val1)}
+
+rangedelIter
+----
+a-j:{(#0,RANGEDEL)}
+
+readyForFlush
+----
+true
+
+containsRangeKey
+----
+true