Repository: cockroachdb/pebble
Pull request #1287 (2021-09-23 21:43:30)

release-21.2: backports related to SingleDelete changes
testdata: remove values from deletes 96994e6
db: improve use-after-free test coverage for compactions 670d312
base: add InternalKeyKindSeparator d55c17f
internal/base: add SetWithDelete key kind b8d37a0
metamorphic: add keyManager abstraction and SetWithDelete bug fix 1cd711f

The first two are included since not including them causes conflicts with later changes.

cockroach-teamcity: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cockroachdb/pebble/1287)
<!-- Reviewable:end -->

sumeerbhola: TFTR!

Patch:
diff --git a/compaction.go b/compaction.go
index c54acaa1..e2336880 100644
--- a/compaction.go
+++ b/compaction.go
@@ -2044,7 +2044,8 @@ func (d *DB) runCompaction(
 	}
 	c.allowedZeroSeqNum = c.allowZeroSeqNum()
 	iter := newCompactionIter(c.cmp, c.formatKey, d.merge, iiter, snapshots,
-		&c.rangeDelFrag, c.allowedZeroSeqNum, c.elideTombstone, c.elideRangeTombstone)
+		&c.rangeDelFrag, c.allowedZeroSeqNum, c.elideTombstone,
+		c.elideRangeTombstone, d.FormatMajorVersion())
 
 	var (
 		filenames []string
diff --git a/compaction_iter.go b/compaction_iter.go
index cadb9b03..f1c188b3 100644
--- a/compaction_iter.go
+++ b/compaction_iter.go
@@ -7,6 +7,7 @@ package pebble
 import (
 	"io"
 	"sort"
+	"strconv"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -122,6 +123,27 @@ import (
 // to take the range tombstones into consideration when outputting normal
 // keys. Just as with point deletions, a range deletion covering an entry can
 // cause the entry to be elided.
+//
+// A note on the stability of keys and values.
+//
+// The stability guarantees of keys and values returned by the iterator tree
+// that backs a compactionIter is nuanced and care must be taken when
+// referencing any returned items.
+//
+// Keys and values returned by exported functions (i.e. First, Next, etc.) have
+// lifetimes that fall into two categories:
+//
+// Lifetime valid for duration of compaction. Range deletion keys and values are
+// stable for the duration of the compaction, due to way in which a
+// compactionIter is typically constructed (i.e. via (*compaction).newInputIter,
+// which wraps the iterator over the range deletion block in a noCloseIter,
+// preventing the release of the backing memory until the compaction is
+// finished).
+//
+// Lifetime limited to duration of sstable block liveness. Point keys (SET, DEL,
+// etc.) and values must be cloned / copied following the return from the
+// exported function, and before a subsequent call to Next advances the iterator
+// and mutates the contents of the returned key and value.
 type compactionIter struct {
 	cmp   Compare
 	merge Merge
@@ -140,6 +162,10 @@ type compactionIter struct {
 	// determine when iteration has advanced to a new user key and thus a new
 	// snapshot stripe.
 	keyBuf []byte
+	// Temporary buffer used for storing the previous value, which may be an
+	// unsafe, i.iter-owned slice that could be altered when the iterator is
+	// advanced.
+	valueBuf []byte
 	// Is the current entry valid?
 	valid     bool
 	iterKey   *InternalKey
@@ -178,6 +204,9 @@ type compactionIter struct {
 	allowZeroSeqNum     bool
 	elideTombstone      func(key []byte) bool
 	elideRangeTombstone func(start, end []byte) bool
+	// The on-disk format major version. This informs the types of keys that
+	// may be written to disk during a compaction.
+	formatVersion FormatMajorVersion
 }
 
 func newCompactionIter(
@@ -190,6 +219,7 @@ func newCompactionIter(
 	allowZeroSeqNum bool,
 	elideTombstone func(key []byte) bool,
 	elideRangeTombstone func(start, end []byte) bool,
+	formatVersion FormatMajorVersion,
 ) *compactionIter {
 	i := &compactionIter{
 		cmp:                 cmp,
@@ -200,6 +230,7 @@ func newCompactionIter(
 		allowZeroSeqNum:     allowZeroSeqNum,
 		elideTombstone:      elideTombstone,
 		elideRangeTombstone: elideRangeTombstone,
+		formatVersion:       formatVersion,
 	}
 	i.rangeDelFrag.Cmp = cmp
 	i.rangeDelFrag.Format = formatKey
@@ -265,6 +296,20 @@ func (i *compactionIter) Next() (*InternalKey, []byte) {
 			// Although, note that `skip` may already be true before reaching here
 			// due to an earlier key in the stripe. Then it is fine to leave it set
 			// to true, as the earlier key must have had a higher sequence number.
+			//
+			// NOTE: there is a subtle invariant violation here in that calling
+			// saveKey and returning a reference to the temporary slice violates
+			// the stability guarantee for range deletion keys. A potential
+			// mediation could return the original iterKey and iterValue
+			// directly, as the backing memory is guaranteed to be stable until
+			// the compaction completes. The violation here is only minor in
+			// that the caller immediately clones the range deletion InternalKey
+			// when passing the key to the deletion fragmenter (see the
+			// call-site in compaction.go).
+			// TODO(travers): address this violation by removing the call to
+			// saveKey and instead return the original iterKey and iterValue.
+			// This goes against the comment on i.key in the struct, and
+			// therefore warrants some investigation.
 			i.saveKey()
 			i.value = i.iterValue
 			i.valid = true
@@ -303,12 +348,13 @@ func (i *compactionIter) Next() (*InternalKey, []byte) {
 				continue
 			}
 
-		case InternalKeyKindSet:
-			i.saveKey()
-			i.value = i.iterValue
-			i.valid = true
-			i.skip = true
-			i.maybeZeroSeqnum(i.curSnapshotIdx)
+		case InternalKeyKindSet, InternalKeyKindSetWithDelete:
+			// The key we emit for this entry is a function of the current key
+			// kind, and whether this entry is followed by a DEL/SINGLEDEL
+			// entry. setNext() does the work to move the iterator forward,
+			// preserving the original value, and potentially mutating the key
+			// kind.
+			i.setNext()
 			return &i.key, i.value
 
 		case InternalKeyKindMerge:
@@ -412,6 +458,12 @@ const (
 
 // nextInStripe advances the iterator and returns one of the above const ints
 // indicating how its state changed.
+//
+// Calls to nextInStripe must be preceded by a call to saveKey to retain a
+// temporary reference to the original key, so that forward iteration can
+// proceed with a reference to the original key. Care should be taken to avoid
+// overwriting or mutating the saved key or value before they have been returned
+// to the caller of the exported function (i.e. the caller of Next, First, etc.)
 func (i *compactionIter) nextInStripe() stripeChangeType {
 	if !i.iterNext() {
 		return newStripe
@@ -444,6 +496,90 @@ func (i *compactionIter) nextInStripe() stripeChangeType {
 	return newStripe
 }
 
+func (i *compactionIter) setNext() {
+	// Save the current key.
+	i.saveKey()
+	i.value = i.iterValue
+	i.valid = true
+	i.maybeZeroSeqnum(i.curSnapshotIdx)
+
+	// There are two cases where we can early return and skip the remaining
+	// records in the stripe:
+	// - If the DB does not SETWITHDEL.
+	// - If this key is already a SETWITHDEL.
+	if i.formatVersion < FormatSetWithDelete ||
+		i.iterKey.Kind() == InternalKeyKindSetWithDelete {
+		i.skip = true
+		return
+	}
+
+	// We are iterating forward. Save the current value.
+	i.valueBuf = append(i.valueBuf[:0], i.iterValue...)
+	i.value = i.valueBuf
+
+	// Else, we continue to loop through entries in the stripe looking for a
+	// DEL. Note that we may stop *before* encountering a DEL, if one exists.
+	for {
+		switch t := i.nextInStripe(); t {
+		case newStripe, sameStripeNonSkippable:
+			i.pos = iterPosNext
+			if t == sameStripeNonSkippable {
+				// We iterated onto a key that we cannot skip. We can
+				// conservatively transform the original SET into a SETWITHDEL
+				// as an indication that there *may* still be a DEL/SINGLEDEL
+				// under this SET, even if we did not actually encounter one.
+				//
+				// This is safe to do, as:
+				//
+				// - in the case that there *is not* actually a DEL/SINGLEDEL
+				// under this entry, any SINGLEDEL above this now-transformed
+				// SETWITHDEL will become a DEL when the two encounter in a
+				// compaction. The DEL will eventually be elided in a
+				// subsequent compaction. The cost for ensuring correctness is
+				// that this entry is kept around for an additional compaction
+				// cycle(s).
+				//
+				// - in the case there *is* indeed a DEL/SINGLEDEL under us
+				// (but in a different stripe or sstable), then we will have
+				// already done the work to transform the SET into a
+				// SETWITHDEL, and we will skip any additional iteration when
+				// this entry is encountered again in a subsequent compaction.
+				//
+				// Ideally, this codepath would be smart enough to handle the
+				// case of SET <- RANGEDEL <- ... <- DEL/SINGLEDEL <- ....
+				// This requires preserving any RANGEDEL entries we encounter
+				// along the way, then emitting the original (possibly
+				// transformed) key, followed by the RANGEDELs. This requires
+				// a sizable refactoring of the existing code, as nextInStripe
+				// currently returns a sameStripeNonSkippable when it
+				// encounters a RANGEDEL.
+				// TODO(travers): optimize to handle the RANGEDEL case if it
+				// turns out to be a performance problem.
+				i.key.SetKind(InternalKeyKindSetWithDelete)
+
+				// By setting i.skip=true, we are saying that after the
+				// non-skippable key is emitted (which is likely a RANGEDEL),
+				// the remaining point keys that share the same user key as this
+				// saved key should be skipped.
+				i.skip = true
+			}
+			return
+		case sameStripeSkippable:
+			// We're still in the same stripe. If this is a DEL/SINGLEDEL, we
+			// stop looking and emit a SETWITHDEL. Subsequent keys are
+			// eligible for skipping.
+			if i.iterKey.Kind() == InternalKeyKindDelete ||
+				i.iterKey.Kind() == InternalKeyKindSingleDelete {
+				i.key.SetKind(InternalKeyKindSetWithDelete)
+				i.skip = true
+				return
+			}
+		default:
+			panic("pebble: unexpected stripeChangeType: " + strconv.Itoa(int(t)))
+		}
+	}
+}
+
 func (i *compactionIter) mergeNext(valueMerger ValueMerger) stripeChangeType {
 	// Save the current key.
 	i.saveKey()
@@ -479,7 +615,7 @@ func (i *compactionIter) mergeNext(valueMerger ValueMerger) stripeChangeType {
 			i.skip = true
 			return sameStripeSkippable
 
-		case InternalKeyKindSet:
+		case InternalKeyKindSet, InternalKeyKindSetWithDelete:
 			if i.rangeDelFrag.Deleted(*key, i.curSnapshotSeqNum) {
 				// We change the kind of the result key to a Set so that it shadows
 				// keys in lower levels. That is, MERGE+RANGEDEL -> SET. This isn't
@@ -490,9 +626,10 @@ func (i *compactionIter) mergeNext(valueMerger ValueMerger) stripeChangeType {
 				return sameStripeSkippable
 			}
 
-			// We've hit a Set value. Merge with the existing value and return. We
-			// change the kind of the resulting key to a Set so that it shadows keys
-			// in lower levels. That is, MERGE+SET -> SET.
+			// We've hit a Set or SetWithDel value. Merge with the existing
+			// value and return. We change the kind of the resulting key to a
+			// Set so that it shadows keys in lower levels. That is:
+			// MERGE + (SET*) -> SET.
 			i.err = valueMerger.MergeOlder(i.iterValue)
 			if i.err != nil {
 				i.valid = false
@@ -544,8 +681,9 @@ func (i *compactionIter) singleDeleteNext() bool {
 
 		key := i.iterKey
 		switch key.Kind() {
-		case InternalKeyKindDelete, InternalKeyKindMerge:
-			// We've hit a Delete or Merge, transform the SingleDelete into a full Delete.
+		case InternalKeyKindDelete, InternalKeyKindMerge, InternalKeyKindSetWithDelete:
+			// We've hit a Delete, Merge or SetWithDelete, transform the
+			// SingleDelete into a full Delete.
 			i.key.SetKind(InternalKeyKindDelete)
 			i.skip = true
 			return true
diff --git a/format_major_version.go b/format_major_version.go
index 4eb67528..05ec13ca 100644
--- a/format_major_version.go
+++ b/format_major_version.go
@@ -62,8 +62,12 @@ const (
 	// Pebble versions will be unable to open the database unless
 	// they're aware of format versions.
 	FormatVersioned
+	// FormatSetWithDelete is a format major version that introduces a new key
+	// kind, base.InternalKeyKindSetWithDelete. Previous Pebble versions will be
+	// unable to open this database.
+	FormatSetWithDelete
 	// FormatNewest always contains the most recent format major version.
-	FormatNewest FormatMajorVersion = FormatVersioned
+	FormatNewest FormatMajorVersion = FormatSetWithDelete
 )
 
 // formatMajorVersionMigrations defines the migrations from one format
@@ -138,6 +142,11 @@ var formatMajorVersionMigrations = map[FormatMajorVersion]func(*DB) error{
 		}
 		return d.finalizeFormatVersUpgrade(FormatVersioned)
 	},
+	// As SetWithDelete is a new key kind, there is nothing to migrate. We can
+	// simply finalize the format version and we're done.
+	FormatSetWithDelete: func(d *DB) error {
+		return d.finalizeFormatVersUpgrade(FormatSetWithDelete)
+	},
 }
 
 const formatVersionMarkerName = `format-version`
diff --git a/internal.go b/internal.go
index 00e38b33..563785ee 100644
--- a/internal.go
+++ b/internal.go
@@ -18,6 +18,7 @@ const (
 	InternalKeyKindSingleDelete    = base.InternalKeyKindSingleDelete
 	InternalKeyKindRangeDelete     = base.InternalKeyKindRangeDelete
 	InternalKeyKindMax             = base.InternalKeyKindMax
+	InternalKeyKindSetWithDelete   = base.InternalKeyKindSetWithDelete
 	InternalKeyKindInvalid         = base.InternalKeyKindInvalid
 	InternalKeySeqNumBatch         = base.InternalKeySeqNumBatch
 	InternalKeySeqNumMax           = base.InternalKeySeqNumMax
diff --git a/internal/base/internal.go b/internal/base/internal.go
index 3afd6f38..056854ff 100644
--- a/internal/base/internal.go
+++ b/internal/base/internal.go
@@ -36,6 +36,21 @@ const (
 	// InternalKeyKindColumnFamilyBlobIndex                    = 16
 	// InternalKeyKindBlobIndex                                = 17
 
+	// InternalKeyKindSeparator is a key used for separator / successor keys
+	// written to sstable block indexes.
+	//
+	// NOTE: the RocksDB value has been repurposed. This was done to ensure that
+	// keys written to block indexes with value "17" (when 17 happened to be the
+	// max value, and InternalKeyKindMax was therefore set to 17), remain stable
+	// when new key kinds are supported in Pebble.
+	InternalKeyKindSeparator = 17
+
+	// InternalKeyKindSetWithDelete keys are SET keys that have met with a
+	// DELETE or SINGLEDEL key in a prior compaction. This key kind is
+	// specific to Pebble. See
+	// https://github.com/cockroachdb/pebble/issues/1255.
+	InternalKeyKindSetWithDelete InternalKeyKind = 18
+
 	// This maximum value isn't part of the file format. It's unlikely,
 	// but future extensions may increase this value.
 	//
@@ -45,7 +60,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 = 17
+	InternalKeyKindMax InternalKeyKind = 18
 
 	// A marker for an invalid key.
 	InternalKeyKindInvalid InternalKeyKind = 255
@@ -66,14 +81,15 @@ const (
 )
 
 var internalKeyKindNames = []string{
-	InternalKeyKindDelete:       "DEL",
-	InternalKeyKindSet:          "SET",
-	InternalKeyKindMerge:        "MERGE",
-	InternalKeyKindLogData:      "LOGDATA",
-	InternalKeyKindSingleDelete: "SINGLEDEL",
-	InternalKeyKindRangeDelete:  "RANGEDEL",
-	InternalKeyKindMax:          "MAX",
-	InternalKeyKindInvalid:      "INVALID",
+	InternalKeyKindDelete:        "DEL",
+	InternalKeyKindSet:           "SET",
+	InternalKeyKindMerge:         "MERGE",
+	InternalKeyKindLogData:       "LOGDATA",
+	InternalKeyKindSingleDelete:  "SINGLEDEL",
+	InternalKeyKindRangeDelete:   "RANGEDEL",
+	InternalKeyKindSeparator:     "SEPARATOR",
+	InternalKeyKindSetWithDelete: "SETWITHDEL",
+	InternalKeyKindInvalid:       "INVALID",
 }
 
 func (k InternalKeyKind) String() string {
@@ -130,13 +146,14 @@ func MakeRangeDeleteSentinelKey(userKey []byte) InternalKey {
 }
 
 var kindsMap = map[string]InternalKeyKind{
-	"DEL":       InternalKeyKindDelete,
-	"SINGLEDEL": InternalKeyKindSingleDelete,
-	"RANGEDEL":  InternalKeyKindRangeDelete,
-	"SET":       InternalKeyKindSet,
-	"MERGE":     InternalKeyKindMerge,
-	"INVALID":   InternalKeyKindInvalid,
-	"MAX":       InternalKeyKindMax,
+	"DEL":        InternalKeyKindDelete,
+	"SINGLEDEL":  InternalKeyKindSingleDelete,
+	"RANGEDEL":   InternalKeyKindRangeDelete,
+	"SET":        InternalKeyKindSet,
+	"MERGE":      InternalKeyKindMerge,
+	"INVALID":    InternalKeyKindInvalid,
+	"SEPARATOR":  InternalKeyKindSeparator,
+	"SETWITHDEL": InternalKeyKindSetWithDelete,
 }
 
 // ParseInternalKey parses the string representation of an internal key. The
@@ -224,7 +241,7 @@ func (k InternalKey) Separator(
 		// any sequence number and kind here to create a valid separator key. We
 		// use the max sequence number to match the behavior of LevelDB and
 		// RocksDB.
-		return MakeInternalKey(buf, InternalKeySeqNumMax, InternalKeyKindMax)
+		return MakeInternalKey(buf, InternalKeySeqNumMax, InternalKeyKindSeparator)
 	}
 	return k
 }
@@ -241,7 +258,7 @@ func (k InternalKey) Successor(cmp Compare, succ Successor, buf []byte) Internal
 		// any sequence number and kind here to create a valid separator key. We
 		// use the max sequence number to match the behavior of LevelDB and
 		// RocksDB.
-		return MakeInternalKey(buf, InternalKeySeqNumMax, InternalKeyKindMax)
+		return MakeInternalKey(buf, InternalKeySeqNumMax, InternalKeyKindSeparator)
 	}
 	return k
 }
diff --git a/internal/cache/value_invariants.go b/internal/cache/value_invariants.go
index bbffca45..9bea1a7f 100644
--- a/internal/cache/value_invariants.go
+++ b/internal/cache/value_invariants.go
@@ -43,9 +43,9 @@ func newValue(n int) *Value {
 func (v *Value) free() {
 	// When "invariants" are enabled set the value contents to 0xff in order to
 	// cache use-after-free bugs.
-	// for i := range v.buf {
-	// 	v.buf[i] = 0xff
-	// }
+	for i := range v.buf {
+		v.buf[i] = 0xff
+	}
 	manual.Free(v.buf)
 	// Setting Value.buf to nil is needed for correctness of the leak checking
 	// that is performed when the "invariants" or "tracing" build tags are
diff --git a/internal/metamorphic/config.go b/internal/metamorphic/config.go
index af42a1e0..36c83522 100644
--- a/internal/metamorphic/config.go
+++ b/internal/metamorphic/config.go
@@ -92,6 +92,6 @@ var defaultConfig = config{
 		writerIngest:        100,
 		writerMerge:         100,
 		writerSet:           100,
-		writerSingleDelete:  25,
+		writerSingleDelete:  50,
 	},
 }
diff --git a/internal/metamorphic/generator.go b/internal/metamorphic/generator.go
index 8e1e6eb0..34f91d30 100644
--- a/internal/metamorphic/generator.go
+++ b/internal/metamorphic/generator.go
@@ -23,16 +23,8 @@ type generator struct {
 	init *initOp
 	ops  []op
 
-	// keys that have been set in the DB. Used to reuse already generated keys
-	// during random key selection.
-	keys [][]byte
-	// singleSetKeysInDB is also in the writerToSingleSetKeys map. This tracks
-	// keys that are eligible to be single deleted.
-	singleSetKeysInDB     singleSetKeysForBatch
-	writerToSingleSetKeys map[objID]singleSetKeysForBatch
-	// Ensures no duplication of single set keys for the duration of the test.
-	generatedWriteKeys map[string]struct{}
-
+	// keyManager tracks the state of keys a operation generation time.
+	keyManager *keyManager
 	// Unordered sets of object IDs for live objects. Used to randomly select on
 	// object when generating an operation. There are 4 concrete objects: the DB
 	// (of which there is exactly 1), batches, iterators, and snapshots.
@@ -71,19 +63,16 @@ type generator struct {
 
 func newGenerator(rng *rand.Rand) *generator {
 	g := &generator{
-		rng:                   rng,
-		init:                  &initOp{},
-		singleSetKeysInDB:     makeSingleSetKeysForBatch(),
-		writerToSingleSetKeys: make(map[objID]singleSetKeysForBatch),
-		generatedWriteKeys:    make(map[string]struct{}),
-		liveReaders:           objIDSlice{makeObjID(dbTag, 0)},
-		liveWriters:           objIDSlice{makeObjID(dbTag, 0)},
-		batches:               make(map[objID]objIDSet),
-		iters:                 make(map[objID]objIDSet),
-		readers:               make(map[objID]objIDSet),
-		snapshots:             make(map[objID]objIDSet),
-	}
-	g.writerToSingleSetKeys[makeObjID(dbTag, 0)] = g.singleSetKeysInDB
+		rng:         rng,
+		init:        &initOp{},
+		keyManager:  newKeyManager(),
+		liveReaders: objIDSlice{makeObjID(dbTag, 0)},
+		liveWriters: objIDSlice{makeObjID(dbTag, 0)},
+		batches:     make(map[objID]objIDSet),
+		iters:       make(map[objID]objIDSet),
+		readers:     make(map[objID]objIDSet),
+		snapshots:   make(map[objID]objIDSet),
+	}
 	// Note that the initOp fields are populated during generation.
 	g.ops = append(g.ops, g.init)
 	return g
@@ -141,47 +130,41 @@ func generate(rng *rand.Rand, count uint64, cfg config) []op {
 
 func (g *generator) add(op op) {
 	g.ops = append(g.ops, op)
+	g.keyManager.update(op)
 }
 
+// randKeyToWrite returns a key for any write other than SingleDelete.
+//
 // TODO(peter): make the size and distribution of keys configurable. See
 // keyDist and keySizeDist in config.go.
-func (g *generator) randKey(newKey float64) []byte {
-	if n := len(g.keys); n > 0 && g.rng.Float64() > newKey {
-		return g.keys[g.rng.Intn(n)]
-	}
-	key := g.randValue(4, 12)
-	g.keys = append(g.keys, key)
-	return key
+func (g *generator) randKeyToWrite(newKey float64) []byte {
+	return g.randKeyHelper(g.keyManager.eligibleWriteKeys(), newKey)
 }
 
-func (g *generator) randKeyForWrite(newKey float64, singleSetKey float64, writerID objID) []byte {
-	if n := len(g.keys); n > 0 && g.rng.Float64() > newKey {
-		return g.keys[g.rng.Intn(n)]
-	}
-	key := g.randValue(4, 12)
-	if g.rng.Float64() < singleSetKey {
-		for {
-			if _, ok := g.generatedWriteKeys[string(key)]; ok {
-				key = g.randValue(4, 12)
-				continue
-			}
-			g.generatedWriteKeys[string(key)] = struct{}{}
-			g.writerToSingleSetKeys[writerID].addKey(key)
-			break
-		}
-	} else {
-		g.generatedWriteKeys[string(key)] = struct{}{}
-		g.keys = append(g.keys, key)
+func (g *generator) randKeyToSingleDelete(id objID) []byte {
+	keys := g.keyManager.eligibleSingleDeleteKeys(id)
+	length := len(keys)
+	if length == 0 {
+		return nil
 	}
-	return key
+	return keys[g.rng.Intn(length)]
 }
 
-func (g *generator) randKeyToSingleDelete() []byte {
-	length := len(*g.singleSetKeysInDB.keys)
-	if length == 0 {
-		return nil
+// randKeyToRead returns a key for read operations.
+func (g *generator) randKeyToRead(newKey float64) []byte {
+	return g.randKeyHelper(g.keyManager.eligibleReadKeys(), newKey)
+}
+
+func (g *generator) randKeyHelper(keys [][]byte, newKey float64) []byte {
+	if n := len(keys); n > 0 && g.rng.Float64() > newKey {
+		return keys[g.rng.Intn(n)]
+	}
+	for {
+		key := g.randValue(4, 12)
+		if g.keyManager.addNewKey(key) {
+			return key
+		}
 	}
-	return g.singleSetKeysInDB.removeKey(g.rng.Intn(length))
 }
 
 // TODO(peter): make the value size configurable. See valueSizeDist in
@@ -222,7 +205,6 @@ func (g *generator) newBatch() {
 	g.init.batchSlots++
 	g.liveBatches = append(g.liveBatches, batchID)
 	g.liveWriters = append(g.liveWriters, batchID)
-	g.writerToSingleSetKeys[batchID] = makeSingleSetKeysForBatch()
 
 	g.add(&newBatchOp{
 		batchID: batchID,
@@ -239,7 +221,6 @@ func (g *generator) newIndexedBatch() {
 	iters := make(objIDSet)
 	g.batches[batchID] = iters
 	g.readers[batchID] = iters
-	g.writerToSingleSetKeys[batchID] = makeSingleSetKeysForBatch()
 
 	g.add(&newIndexedBatchOp{
 		batchID: batchID,
@@ -261,7 +242,6 @@ func (g *generator) batchClose(batchID objID) {
 		delete(g.iters, id)
 		g.add(&closeOp{objID: id})
 	}
-	delete(g.writerToSingleSetKeys, batchID)
 }
 
 func (g *generator) batchAbort() {
@@ -281,7 +261,6 @@ func (g *generator) batchCommit() {
 	}
 
 	batchID := g.liveBatches.rand(g.rng)
-	g.writerToSingleSetKeys[batchID].transferTo(g.singleSetKeysInDB)
 	g.batchClose(batchID)
 
 	g.add(&batchCommitOp{
@@ -307,8 +286,8 @@ func (g *generator) dbCheckpoint() {
 
 func (g *generator) dbCompact() {
 	// Generate new key(s) with a 1% probability.
-	start := g.randKey(0.01)
-	end := g.randKey(0.01)
+	start := g.randKeyToRead(0.01)
+	end := g.randKeyToRead(0.01)
 	if bytes.Compare(start, end) > 0 {
 		start, end = end, start
 	}
@@ -360,11 +339,11 @@ func (g *generator) newIter() {
 	var lower, upper []byte
 	if g.rng.Float64() <= 0.1 {
 		// Generate a new key with a .1% probability.
-		lower = g.randKey(0.001)
+		lower = g.randKeyToRead(0.001)
 	}
 	if g.rng.Float64() <= 0.1 {
 		// Generate a new key with a .1% probability.
-		upper = g.randKey(0.001)
+		upper = g.randKeyToRead(0.001)
 	}
 	if bytes.Compare(lower, upper) > 0 {
 		lower, upper = upper, lower
@@ -446,11 +425,11 @@ func (g *generator) iterSetBounds() {
 		attempts++
 		if genLower {
 			// Generate a new key with a .1% probability.
-			lower = g.randKey(0.001)
+			lower = g.randKeyToRead(0.001)
 		}
 		if genUpper {
 			// Generate a new key with a .1% probability.
-			upper = g.randKey(0.001)
+			upper = g.randKeyToRead(0.001)
 		}
 		if bytes.Compare(lower, upper) > 0 {
 			lower, upper = upper, lower
@@ -548,7 +527,7 @@ func (g *generator) iterSeekGE() {
 
 	g.add(&iterSeekGEOp{
 		iterID: g.liveIters.rand(g.rng),
-		key:    g.randKey(0.001), // 0.1% new keys
+		key:    g.randKeyToRead(0.001), // 0.1% new keys
 	})
 }
 
@@ -557,7 +536,7 @@ func (g *generator) iterSeekGEWithLimit() {
 		return
 	}
 	// 0.1% new keys
-	key, limit := g.randKey(0.001), g.randKey(0.001)
+	key, limit := g.randKeyToRead(0.001), g.randKeyToRead(0.001)
 	if bytes.Compare(key, limit) > 0 {
 		key, limit = limit, key
 	}
@@ -575,7 +554,7 @@ func (g *generator) iterSeekPrefixGE() {
 
 	g.add(&iterSeekPrefixGEOp{
 		iterID: g.liveIters.rand(g.rng),
-		key:    g.randKey(0), // 0% new keys
+		key:    g.randKeyToRead(0), // 0% new keys
 	})
 }
 
@@ -586,7 +565,7 @@ func (g *generator) iterSeekLT() {
 
 	g.add(&iterSeekLTOp{
 		iterID: g.liveIters.rand(g.rng),
-		key:    g.randKey(0.001), // 0.1% new keys
+		key:    g.randKeyToRead(0.001), // 0.1% new keys
 	})
 }
 
@@ -595,7 +574,7 @@ func (g *generator) iterSeekLTWithLimit() {
 		return
 	}
 	// 0.1% new keys
-	key, limit := g.randKey(0.001), g.randKey(0.001)
+	key, limit := g.randKeyToRead(0.001), g.randKeyToRead(0.001)
 	if bytes.Compare(limit, key) > 0 {
 		key, limit = limit, key
 	}
@@ -643,7 +622,7 @@ func (g *generator) iterNextWithLimit() {
 
 	g.add(&iterNextOp{
 		iterID: g.liveIters.rand(g.rng),
-		limit:  g.randKey(0.001), // 0.1% new keys
+		limit:  g.randKeyToRead(0.001), // 0.1% new keys
 	})
 }
 
@@ -664,7 +643,7 @@ func (g *generator) iterPrevWithLimit() {
 
 	g.add(&iterPrevOp{
 		iterID: g.liveIters.rand(g.rng),
-		limit:  g.randKey(0.001), // 0.1% new keys
+		limit:  g.randKeyToRead(0.001), // 0.1% new keys
 	})
 }
 
@@ -675,7 +654,7 @@ func (g *generator) readerGet() {
 
 	g.add(&getOp{
 		readerID: g.liveReaders.rand(g.rng),
-		key:      g.randKey(0.001), // 0.1% new keys
+		key:      g.randKeyToRead(0.001), // 0.1% new keys
 	})
 }
 
@@ -733,7 +712,6 @@ func (g *generator) writerApply() {
 		}
 	}
 
-	g.writerToSingleSetKeys[batchID].transferTo(g.writerToSingleSetKeys[writerID])
 	g.batchClose(batchID)
 
 	g.add(&applyOp{
@@ -750,7 +728,7 @@ func (g *generator) writerDelete() {
 	writerID := g.liveWriters.rand(g.rng)
 	g.add(&deleteOp{
 		writerID: writerID,
-		key:      g.randKey(0.001), // 0.1% new keys
+		key:      g.randKeyToWrite(0.001), // 0.1% new keys
 	})
 	g.tryRepositionBatchIters(writerID)
 }
@@ -760,8 +738,8 @@ func (g *generator) writerDeleteRange() {
 		return
 	}
 
-	start := g.randKey(0.001)
-	end := g.randKey(0.001)
+	start := g.randKeyToWrite(0.001)
+	end := g.randKeyToWrite(0.001)
 	if bytes.Compare(start, end) > 0 {
 		start, end = end, start
 	}
@@ -780,20 +758,34 @@ func (g *generator) writerIngest() {
 		return
 	}
 
+	// TODO(nicktrav): this is resulting in too many single batch ingests.
+	// Consider alternatives. One possibility would be to pass through whether
+	// we can tolerate failure or not, and if the ingestOp encounters a
+	// failure, it would retry after splitting into single batch ingests.
+
 	// Ingest between 1 and 3 batches.
 	batchIDs := make([]objID, 0, 1+g.rng.Intn(3))
+	canFail := cap(batchIDs) > 1
 	for i := 0; i < cap(batchIDs); i++ {
 		batchID := g.liveBatches.rand(g.rng)
+		if canFail && !g.keyManager.canTolerateApplyFailure(batchID) {
+			continue
+		}
 		// After the ingest runs, it either succeeds and the keys are in the
 		// DB, or it fails and these keys never make it to the DB.
-		g.writerToSingleSetKeys[batchID].transferTo(g.singleSetKeysInDB)
 		g.batchClose(batchID)
 		batchIDs = append(batchIDs, batchID)
 		if len(g.liveBatches) == 0 {
 			break
 		}
 	}
-
+	if len(batchIDs) == 0 && len(g.liveBatches) > 0 {
+		// Unable to find multiple batches because of the
+		// canTolerateApplyFailure call above, so just pick one batch.
+		batchID := g.liveBatches.rand(g.rng)
+		g.batchClose(batchID)
+		batchIDs = append(batchIDs, batchID)
+	}
 	g.add(&ingestOp{
 		batchIDs: batchIDs,
 	})
@@ -807,8 +799,8 @@ func (g *generator) writerMerge() {
 	writerID := g.liveWriters.rand(g.rng)
 	g.add(&mergeOp{
 		writerID: writerID,
-		// 20% new keys, and none are set once.
-		key:   g.randKeyForWrite(0.2, 0, writerID),
+		// 20% new keys.
+		key:   g.randKeyToWrite(0.2),
 		value: g.randValue(0, 20),
 	})
 	g.tryRepositionBatchIters(writerID)
@@ -822,9 +814,8 @@ func (g *generator) writerSet() {
 	writerID := g.liveWriters.rand(g.rng)
 	g.add(&setOp{
 		writerID: writerID,
-		// 50% new keys, and of those 50%, half are keys that will be set
-		// once.
-		key:   g.randKeyForWrite(0.5, 0.5, writerID),
+		// 50% new keys.
+		key:   g.randKeyToWrite(0.5),
 		value: g.randValue(0, 20),
 	})
 	g.tryRepositionBatchIters(writerID)
@@ -836,7 +827,7 @@ func (g *generator) writerSingleDelete() {
 	}
 
 	writerID := g.liveWriters.rand(g.rng)
-	key := g.randKeyToSingleDelete()
+	key := g.randKeyToSingleDelete(writerID)
 	if key == nil {
 		return
 	}
diff --git a/internal/metamorphic/key_manager.go b/internal/metamorphic/key_manager.go
new file mode 100644
index 00000000..e78c8d08
--- /dev/null
+++ b/internal/metamorphic/key_manager.go
@@ -0,0 +1,418 @@
+package metamorphic
+
+import (
+	"bytes"
+	"fmt"
+	"sort"
+)
+
+// objKey is a tuple of (objID, key). This struct is used primarily as a map
+// key for keyManager. Only writer objTags can occur here, i.e., dbTag and
+// batchTag, since this is used for tracking the keys in a writer.
+type objKey struct {
+	id  objID
+	key []byte
+}
+
+// makeObjKey returns a new objKey given and id and key.
+func makeObjKey(id objID, key []byte) objKey {
+	if id.tag() != dbTag && id.tag() != batchTag {
+		panic("unexpected non-writer tag")
+	}
+	return objKey{id, key}
+}
+
+// String implements fmt.Stringer, returning a stable string representation of
+// the objKey. This string is used as map key.
+func (o objKey) String() string {
+	return fmt.Sprintf("%s:%s", o.id, o.key)
+}
+
+// keyMeta is metadata associated with an (objID, key) pair, where objID is
+// a writer containing the key.
+type keyMeta struct {
+	objKey
+
+	// The number of Sets of the key in this writer.
+	sets      int
+	// The number of Merges of the key in this writer.
+	merges    int
+	// singleDel can be true only if sets <= 1 && merges == 0 and the
+	// SingleDelete was added to this writer after the set.
+	singleDel bool
+	// The number of Deletes of the key in this writer.
+	dels int
+	// del can be true only if a Delete was added to this writer after the
+	// Sets and Merges counted above.
+	del bool
+}
+
+func (m *keyMeta) clear() {
+	m.sets = 0
+	m.merges = 0
+	m.singleDel = false
+	m.del = false
+	m.dels = 0
+}
+
+// mergeInto merges this metadata this into the metadata for other.
+func (m *keyMeta) mergeInto(other *keyMeta) {
+	if other.del && !m.del {
+		// m's Sets and Merges are later.
+		if m.sets > 0 || m.merges > 0 {
+			other.del = false
+		}
+	} else {
+		other.del = m.del
+	}
+	// Sets, merges, dels are additive.
+	other.sets += m.sets
+	other.merges += m.merges
+	other.dels += m.dels
+
+	// Single deletes are preserved. This is valid since we are also
+	// maintaining a global invariant that SingleDelete will only be added for
+	// a key that has no inflight Sets or Merges (Sets have made their way to
+	// the DB), and no subsequent Sets or Merges will happen until the
+	// SingleDelete makes its way to th DB.
+	other.singleDel = other.singleDel || m.singleDel
+	if other.singleDel {
+		if other.sets > 1 || other.merges > 0 || other.dels > 0 {
+			panic(fmt.Sprintf("invalid sets %d or merges %d or dels %d",
+				other.sets, other.merges, other.dels))
+		}
+	}
+}
+
+// keyManager tracks the write operations performed on keys in the generation
+// phase of the metamorphic test. It makes the assumption that write
+// operations do not fail, since that can cause the keyManager state to be not
+// in-sync with the actual state of the writers. This assumption is needed to
+// correctly decide when it is safe to generate a SingleDelete. This
+// assumption is violated in a single place in the metamorphic test: ingestion
+// of multiple batches. We sidestep this issue in a narrow way in
+// generator.writerIngest by not ingesting multiple batches that contain
+// deletes or single deletes, since loss of those specific operations on a key
+// are what we cannot tolerate (doing SingleDelete on a key that has not been
+// written to because the Set was lost is harmless).
+type keyManager struct {
+	// byObjKey tracks the state for each (writer, key) pair. It refers to the
+	// same *keyMeta as in the byObj slices. Using a map allows for fast state
+	// lookups when changing the state based on a writer operation on the key.
+	byObjKey map[string]*keyMeta
+	// List of keys per writer, and what has happened to it in that writer.
+	// Will be transferred when needed.
+	byObj    map[objID][]*keyMeta
+
+	// globalKeys represents all the keys that have been generated so far. Not
+	// all these keys have been written to.
+	globalKeys     [][]byte
+	// globalKeysMap contains the same keys as globalKeys. It ensures no
+	// duplication, and contains the aggregate state of the key across all
+	// writers, including inflight state that has not made its way to the DB
+	// yet.The keyMeta.objKey is uninitialized.
+	globalKeysMap map[string]*keyMeta
+
+	// Using SingleDeletes imposes some constraints on the above state, and
+	// causes some state transitions that help with generating complex but
+	// correct sequences involving SingleDeletes.
+	// - Generating a SingleDelete requires for that key: global.merges==0 &&
+	//   global.sets==1 && global.dels==0 && !global.singleDel && (db.sets==1
+	//   || writer.sets==1), where global represents the entry in
+	//   globalKeysMap[key] and db represents the entry in
+	//   byObjKey[makeObjKey(makeObjID(dbTag, 0), key)], and writer is the
+	//   entry in byObjKey[makeObjKey(writerID, key)].
+	//
+	// - We do not track state changes due to range deletes, so one should
+	//   think of these counts as upper bounds. Also we are not preventing
+	//   interactions caused by concurrently in-flight range deletes and
+	//   SingleDelete. This is acceptable since it does not cause
+	//   non-determinism.
+	//
+	// - When the SingleDelete is generated, it is recorded as
+	//   writer.singleDel=true and global.singleDel=true. No more write
+	//   operations are permitted on this key until db.singleDel transitions
+	//   to true.
+	//
+	// - When db.singleDel transitions to true, we are guaranteed that no
+	//   writer other than the DB has any writes for this key. We set
+	//   db.singleDel and global.singleDel to false and the corresponding sets
+	//   and merges counts in global and db also to 0. This allows this key to
+	//   fully participate again in write operations. This means we can
+	//   generate sequences of the form:
+	//   SET => SINGLEDEL => SET* => MERGE* => DEL
+	//   SET => SINGLEDEL => SET => SINGLEDEL, among others.
+	//
+	// - The above logic is insufficient to generate sequences of the form
+	//   SET => DEL => SET => SINGLEDEL
+	//   To do this we need to track Deletes. When db.del transitions to true,
+	//   we check if db.sets==global.sets && db.merges==global.merges &&
+	//   db.dels==global.dels. If true, there are no in-flight
+	//   sets/merges/deletes to this key. We then default initialize the
+	//   global and db entries since one can behave as if this key was never
+	//   written in this system. This enables the above sequence, among
+	//   others.
+}
+
+var dbObjID objID = makeObjID(dbTag, 0)
+
+// newKeyManager returns a pointer to a new keyManager. Callers should
+// interact with this using addNewKey, eligible*Keys, update,
+// canTolerateApplyFailure methods only.
+func newKeyManager() *keyManager {
+	m := &keyManager{
+		byObjKey:       make(map[string]*keyMeta),
+		byObj:          make(map[objID][]*keyMeta),
+		globalKeysMap: make(map[string]*keyMeta),
+	}
+	m.byObj[dbObjID] = []*keyMeta{}
+	return m
+}
+
+// addKey adds the given key to the key manager for global key tracking.
+// Returns false iff this is not a new key.
+func (k *keyManager) addNewKey(key []byte) bool {
+	_, ok := k.globalKeysMap[string(key)]
+	if ok {
+		return false
+	}
+	k.globalKeys = append(k.globalKeys, key)
+	k.globalKeysMap[string(key)] = &keyMeta{objKey: objKey{key: key}}
+	return true
+}
+
+// getOrInit returns the keyMeta for the (objID, key) pair, if it exists, else
+// allocates, initializes and returns a new value.
+func (k *keyManager) getOrInit(id objID, key []byte) *keyMeta {
+	o := makeObjKey(id, key)
+	m, ok := k.byObjKey[o.String()]
+	if ok {
+		return m
+	}
+	m = &keyMeta{objKey: makeObjKey(id, key)}
+	// Initialize the key-to-meta index.
+	k.byObjKey[o.String()] = m
+	// Add to the id-to-metas slide.
+	k.byObj[o.id] = append(k.byObj[o.id], m)
+	return m
+}
+
+// contains returns true if the (objID, key) pair is tracked by the keyManager.
+func (k *keyManager) contains(id objID, key []byte) bool {
+	_, ok := k.byObjKey[makeObjKey(id, key).String()]
+	return ok
+}
+
+// mergeKeysInto merges all metadata for all keys associated with the "from" ID
+// with the metadata for keys associated with the "to" ID.
+func (k *keyManager) mergeKeysInto(from, to objID) {
+	msFrom, ok := k.byObj[from]
+	if !ok {
+		msFrom = []*keyMeta{}
+		k.byObj[from] = msFrom
+	}
+
+	msTo, ok := k.byObj[to]
+	if !ok {
+		msTo = []*keyMeta{}
+		k.byObj[to] = msTo
+	}
+
+	// Sort to facilitate a merge.
+	sort.Slice(msFrom, func(i, j int) bool {
+		return msFrom[i].String() < msFrom[j].String()
+	})
+	sort.Slice(msTo, func(i, j int) bool {
+		return msTo[i].String() < msTo[j].String()
+	})
+
+	var msNew []*keyMeta
+	var iTo int
+	for _, m := range msFrom {
+		// Move cursor on mTo forward.
+		for iTo < len(msTo) && string(msTo[iTo].key) < string(m.key) {
+			msNew = append(msNew, msTo[iTo])
+			iTo++
+		}
+
+		var mTo *keyMeta
+		if iTo < len(msTo) && string(msTo[iTo].key) == string(m.key) {
+			mTo = msTo[iTo]
+			iTo++
+		} else {
+			mTo = &keyMeta{objKey: makeObjKey(to, m.key)}
+			k.byObjKey[mTo.String()] = mTo
+		}
+
+		m.mergeInto(mTo)
+		msNew = append(msNew, mTo)
+
+		delete(k.byObjKey, m.String()) // Unlink "from".
+	}
+
+	// Add any remaining items from the "to" set.
+	for iTo < len(msTo) {
+		msNew = append(msNew, msTo[iTo])
+		iTo++
+	}
+
+	k.byObj[to] = msNew   // Update "to".
+	delete(k.byObj, from) // Unlink "from".
+}
+
+func (k *keyManager) checkForDelOrSingleDelTransition(dbMeta *keyMeta, globalMeta *keyMeta) {
+	if dbMeta.singleDel {
+		if !globalMeta.singleDel {
+			panic("inconsistency with globalMeta")
+		}
+		if dbMeta.del || globalMeta.del || dbMeta.dels > 0 || globalMeta.dels > 0 ||
+			dbMeta.merges > 0 || globalMeta.merges > 0 || dbMeta.sets != 1 || globalMeta.sets != 1 {
+			panic("inconsistency in metas when SingleDelete applied to DB")
+		}
+		dbMeta.clear()
+		globalMeta.clear()
+		return
+	}
+	if dbMeta.del && globalMeta.sets == dbMeta.sets && globalMeta.merges == dbMeta.merges &&
+		globalMeta.dels == dbMeta.dels {
+		if dbMeta.singleDel || globalMeta.singleDel {
+			panic("Delete should not have happened given SingleDelete")
+		}
+		dbMeta.clear()
+		globalMeta.clear()
+	}
+}
+
+func (k *keyManager) checkForDelOrSingleDelTransitionInDB() {
+	keys := k.byObj[dbObjID]
+	for _, dbMeta := range keys {
+		globalMeta := k.globalKeysMap[string(dbMeta.key)]
+		k.checkForDelOrSingleDelTransition(dbMeta, globalMeta)
+	}
+}
+
+// update updates the internal state of the keyManager according to the given
+// op.
+func (k *keyManager) update(o op) {
+	switch s := o.(type) {
+	case *setOp:
+		meta := k.getOrInit(s.writerID, s.key)
+		globalMeta := k.globalKeysMap[string(s.key)]
+		meta.sets++           // Update the set count on this specific (id, key) pair.
+		meta.del = false
+		globalMeta.sets++
+		if meta.singleDel || globalMeta.singleDel {
+			panic("setting a key that has in-flight SingleDelete")
+		}
+	case *mergeOp:
+		meta := k.getOrInit(s.writerID, s.key)
+		globalMeta := k.globalKeysMap[string(s.key)]
+		meta.merges++
+		meta.del = false
+		globalMeta.merges++
+		if meta.singleDel || globalMeta.singleDel {
+			panic("merging a key that has in-flight SingleDelete")
+		}
+	case *deleteOp:
+		meta := k.getOrInit(s.writerID, s.key)
+		globalMeta := k.globalKeysMap[string(s.key)]
+		meta.del = true
+		globalMeta.del = true
+		meta.dels++
+		globalMeta.dels++
+		if s.writerID == dbObjID {
+			k.checkForDelOrSingleDelTransition(meta, globalMeta)
+		}
+	case *singleDeleteOp:
+		if !k.globalStateIndicatesEligibleForSingleDelete(s.key) {
+			panic("key ineligible for SingleDelete")
+		}
+		meta := k.getOrInit(s.writerID, s.key)
+		globalMeta := k.globalKeysMap[string(s.key)]
+		meta.singleDel = true
+		globalMeta.singleDel = true
+		if s.writerID == dbObjID {
+			k.checkForDelOrSingleDelTransition(meta, globalMeta)
+		}
+	case *ingestOp:
+		// For each batch, merge all keys with the keys in the DB.
+		for _, batchID := range s.batchIDs {
+			k.mergeKeysInto(batchID, dbObjID)
+		}
+		k.checkForDelOrSingleDelTransitionInDB()
+	case *applyOp:
+		// Merge the keys from this writer into the parent writer.
+		k.mergeKeysInto(s.batchID, s.writerID)
+		if s.writerID == dbObjID {
+			k.checkForDelOrSingleDelTransitionInDB()
+		}
+	case *batchCommitOp:
+		// Merge the keys from the batch with the keys from the DB.
+		k.mergeKeysInto(s.batchID, dbObjID)
+		k.checkForDelOrSingleDelTransitionInDB()
+	}
+}
+
+func (k *keyManager) eligibleReadKeys() (keys [][]byte) {
+	return k.globalKeys
+}
+
+func (k *keyManager) eligibleWriteKeys() (keys [][]byte) {
+	// Creating and sorting this slice of keys is wasteful given that the
+	// caller will pick one, but makes it simpler for unit testing.
+	for _, v := range k.globalKeysMap {
+		if v.singleDel {
+			continue
+		}
+		keys = append(keys, v.key)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		return bytes.Compare(keys[i], keys[j]) < 0
+	})
+	return keys
+}
+
+// eligibleSingleDeleteKeys returns a slice of keys that can be safely single
+// deleted, given the writer id.
+func (k *keyManager) eligibleSingleDeleteKeys(id objID) (keys [][]byte) {
+	// Creating and sorting this slice of keys is wasteful given that the
+	// caller will pick one, but makes it simpler for unit testing.
+	addForObjID := func(id objID) {
+		for _, m := range k.byObj[id] {
+			if m.sets == 1 && k.globalStateIndicatesEligibleForSingleDelete(m.key) {
+				keys = append(keys, m.key)
+			}
+		}
+	}
+	addForObjID(id)
+	if id != dbObjID {
+		addForObjID(dbObjID)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		return bytes.Compare(keys[i], keys[j]) < 0
+	})
+	return keys
+}
+
+func (k *keyManager) globalStateIndicatesEligibleForSingleDelete(key []byte) bool {
+	m := k.globalKeysMap[string(key)]
+	return m.merges==0 && m.sets==1 && m.dels==0 && !m.singleDel
+}
+
+// canTolerateApplyFailure is called with a batch ID and returns true iff a
+// failure to apply this batch to the DB can be tolerated.
+func (k *keyManager) canTolerateApplyFailure(id objID) bool {
+	if id.tag() != batchTag {
+		panic("called with an objID that is not a batch")
+	}
+	ms, ok := k.byObj[id]
+	if !ok {
+		return true
+	}
+	for _, m := range ms {
+		if m.singleDel || m.del {
+			return false
+		}
+	}
+	return true
+}
diff --git a/internal/metamorphic/options.go b/internal/metamorphic/options.go
index 5f9dcab1..523d2132 100644
--- a/internal/metamorphic/options.go
+++ b/internal/metamorphic/options.go
@@ -189,7 +189,7 @@ func standardOptions() []*testOptions {
 `,
 		21: `
 [TestOptions]
-  use_disk=true
+ use_disk=true
 `,
 	}
 
@@ -210,7 +210,13 @@ func randomOptions(rng *rand.Rand) *testOptions {
 	opts.Cache = cache.New(1 << uint(rng.Intn(30))) // 1B - 1GB
 	opts.DisableWAL = rng.Intn(2) == 0
 	opts.FlushSplitBytes = 1 << rng.Intn(20) // 1B - 1MB
-	opts.FormatMajorVersion = pebble.FormatMajorVersion(rng.Intn(int(pebble.FormatNewest)) + 1)
+	// The metamorphic test exercises the latest SingleDelete semantics, that
+	// require SetWithDelete so we cannot use an older FormatMajorVersion.
+	opts.FormatMajorVersion = pebble.FormatSetWithDelete
+	n := int(pebble.FormatNewest - pebble.FormatSetWithDelete)
+	if n > 0 {
+		opts.FormatMajorVersion += pebble.FormatMajorVersion(rng.Intn(n))
+	}
 	opts.Experimental.L0CompactionConcurrency = 1 + rng.Intn(4)    // 1-4
 	opts.Experimental.MinDeletionRate = 1 << uint(20+rng.Intn(10)) // 1MB - 1GB
 	opts.L0CompactionThreshold = 1 + rng.Intn(100)                 // 1 - 100
diff --git a/iterator.go b/iterator.go
index 6bfa85eb..4d7ebfb9 100644
--- a/iterator.go
+++ b/iterator.go
@@ -245,7 +245,7 @@ func (i *Iterator) findNextEntry(limit []byte) {
 			i.nextUserKey()
 			continue
 
-		case InternalKeyKindSet:
+		case InternalKeyKindSet, InternalKeyKindSetWithDelete:
 			i.keyBuf = append(i.keyBuf[:0], key.UserKey...)
 			i.key = i.keyBuf
 			i.value = i.iterValue
@@ -459,7 +459,7 @@ func (i *Iterator) findPrevEntry(limit []byte) {
 			}
 			continue
 
-		case InternalKeyKindSet:
+		case InternalKeyKindSet, InternalKeyKindSetWithDelete:
 			i.keyBuf = append(i.keyBuf[:0], key.UserKey...)
 			i.key = i.keyBuf
 			// iterValue is owned by i.iter and could change after the Prev()
@@ -570,7 +570,7 @@ func (i *Iterator) mergeNext(key InternalKey, valueMerger ValueMerger) {
 			// point.
 			return
 
-		case InternalKeyKindSet:
+		case InternalKeyKindSet, InternalKeyKindSetWithDelete:
 			// We've hit a Set value. Merge with the existing value and return.
 			i.err = valueMerger.MergeOlder(i.iterValue)
 			return
diff --git a/level_checker.go b/level_checker.go
index d779546e..dcf60109 100644
--- a/level_checker.go
+++ b/level_checker.go
@@ -167,7 +167,7 @@ func (m *simpleMergingIter) step() bool {
 					m.err = closer.Close()
 				}
 				m.valueMerger = nil
-			case InternalKeyKindSet:
+			case InternalKeyKindSet, InternalKeyKindSetWithDelete:
 				m.err = m.valueMerger.MergeOlder(item.value)
 				if m.err == nil {
 					var closer io.Closer
diff --git a/sstable/block.go b/sstable/block.go
index f0c4aa95..8ef57da2 100644
--- a/sstable/block.go
+++ b/sstable/block.go
@@ -187,6 +187,10 @@ type blockEntry struct {
 // tombstones are always encoded with a restart interval of 1. This per-block
 // key stability guarantee is sufficient for range tombstones as they are
 // always encoded in a single block.
+//
+// A blockIter also provides a value stability guarantee for range deletions
+// since there is only a single range deletion block per sstable and the
+// blockIter will not release the bytes for the block until it is closed.
 type blockIter struct {
 	cmp Compare
 	// offset is the byte index that marks where the current key/value is
diff --git a/tool/find.go b/tool/find.go
index 214f9a25..8714a5b7 100644
--- a/tool/find.go
+++ b/tool/find.go
@@ -345,7 +345,8 @@ func (f *findT) searchLogs(searchKey []byte, refs []findRef) []findRef {
 					case base.InternalKeyKindDelete,
 						base.InternalKeyKindSet,
 						base.InternalKeyKindMerge,
-						base.InternalKeyKindSingleDelete:
+						base.InternalKeyKindSingleDelete,
+						base.InternalKeyKindSetWithDelete:
 						if cmp(searchKey, ikey.UserKey) != 0 {
 							continue
 						}
diff --git a/tool/wal.go b/tool/wal.go
index 28092eef..d701174b 100644
--- a/tool/wal.go
+++ b/tool/wal.go
@@ -138,6 +138,8 @@ func (w *walT) runDump(cmd *cobra.Command, args []string) {
 						fmt.Fprintf(stdout, "<%d>", len(value))
 					case base.InternalKeyKindSingleDelete:
 						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
+					case base.InternalKeyKindSetWithDelete:
+						fmt.Fprintf(stdout, "%s", w.fmtKey.fn(ukey))
 					case base.InternalKeyKindRangeDelete:
 						fmt.Fprintf(stdout, "%s,%s", w.fmtKey.fn(ukey), w.fmtKey.fn(value))
 					}


Test patch:
diff --git a/compaction_iter_test.go b/compaction_iter_test.go
index 86015e9f..dc0c3be0 100644
--- a/compaction_iter_test.go
+++ b/compaction_iter_test.go
@@ -79,7 +79,23 @@ func TestCompactionIter(t *testing.T) {
 	var elideTombstones bool
 	var allowZeroSeqnum bool
 
-	newIter := func() *compactionIter {
+	// The input to the data-driven test is dependent on the format major
+	// version we are testing against.
+	fileFunc := func(formatVersion FormatMajorVersion) string {
+		if formatVersion < FormatSetWithDelete {
+			return "testdata/compaction_iter"
+		}
+		return "testdata/compaction_iter_set_with_del"
+	}
+
+	newIter := func(formatVersion FormatMajorVersion) *compactionIter {
+		// To adhere to the existing assumption that range deletion blocks in
+		// SSTables are not released while iterating, and therefore not
+		// susceptible to use-after-free bugs, we skip the zeroing of
+		// RangeDelete keys.
+		iter := newInvalidatingIter(&fakeIter{keys: keys, vals: vals})
+		iter.ignoreKind(InternalKeyKindRangeDelete)
+
 		return newCompactionIter(
 			DefaultComparer.Compare,
 			DefaultComparer.FormatKey,
@@ -88,7 +104,7 @@ func TestCompactionIter(t *testing.T) {
 				m.buf = append(m.buf, value...)
 				return m, nil
 			},
-			&fakeIter{keys: keys, vals: vals},
+			iter,
 			snapshots,
 			&rangedel.Fragmenter{},
 			allowZeroSeqnum,
@@ -98,96 +114,113 @@ func TestCompactionIter(t *testing.T) {
 			func(_, _ []byte) bool {
 				return elideTombstones
 			},
+			formatVersion,
 		)
 	}
 
-	datadriven.RunTest(t, "testdata/compaction_iter", func(d *datadriven.TestData) string {
-		switch d.Cmd {
-		case "define":
-			keys = keys[:0]
-			vals = vals[:0]
-			for _, key := range strings.Split(d.Input, "\n") {
-				j := strings.Index(key, ":")
-				keys = append(keys, base.ParseInternalKey(key[:j]))
-				vals = append(vals, []byte(key[j+1:]))
-			}
-			return ""
-
-		case "iter":
-			snapshots = snapshots[:0]
-			elideTombstones = false
-			allowZeroSeqnum = false
-			for _, arg := range d.CmdArgs {
-				switch arg.Key {
-				case "snapshots":
-					for _, val := range arg.Vals {
-						seqNum, err := strconv.Atoi(val)
+	runTest := func(t *testing.T, formatVersion FormatMajorVersion) {
+		datadriven.RunTest(t, fileFunc(formatVersion), func(d *datadriven.TestData) string {
+			switch d.Cmd {
+			case "define":
+				keys = keys[:0]
+				vals = vals[:0]
+				for _, key := range strings.Split(d.Input, "\n") {
+					j := strings.Index(key, ":")
+					keys = append(keys, base.ParseInternalKey(key[:j]))
+					vals = append(vals, []byte(key[j+1:]))
+				}
+				return ""
+
+			case "iter":
+				snapshots = snapshots[:0]
+				elideTombstones = false
+				allowZeroSeqnum = false
+				for _, arg := range d.CmdArgs {
+					switch arg.Key {
+					case "snapshots":
+						for _, val := range arg.Vals {
+							seqNum, err := strconv.Atoi(val)
+							if err != nil {
+								return err.Error()
+							}
+							snapshots = append(snapshots, uint64(seqNum))
+						}
+					case "elide-tombstones":
+						var err error
+						elideTombstones, err = strconv.ParseBool(arg.Vals[0])
 						if err != nil {
 							return err.Error()
 						}
-						snapshots = append(snapshots, uint64(seqNum))
-					}
-				case "elide-tombstones":
-					var err error
-					elideTombstones, err = strconv.ParseBool(arg.Vals[0])
-					if err != nil {
-						return err.Error()
-					}
-				case "allow-zero-seqnum":
-					var err error
-					allowZeroSeqnum, err = strconv.ParseBool(arg.Vals[0])
-					if err != nil {
-						return err.Error()
+					case "allow-zero-seqnum":
+						var err error
+						allowZeroSeqnum, err = strconv.ParseBool(arg.Vals[0])
+						if err != nil {
+							return err.Error()
+						}
+					default:
+						return fmt.Sprintf("%s: unknown arg: %s", d.Cmd, arg.Key)
 					}
-				default:
-					return fmt.Sprintf("%s: unknown arg: %s", d.Cmd, arg.Key)
 				}
-			}
-			sort.Slice(snapshots, func(i, j int) bool {
-				return snapshots[i] < snapshots[j]
-			})
-
-			iter := newIter()
-			var b bytes.Buffer
-			for _, line := range strings.Split(d.Input, "\n") {
-				parts := strings.Fields(line)
-				if len(parts) == 0 {
-					continue
-				}
-				switch parts[0] {
-				case "first":
-					iter.First()
-				case "next":
-					iter.Next()
-				case "tombstones":
-					var key []byte
-					if len(parts) == 2 {
-						key = []byte(parts[1])
+				sort.Slice(snapshots, func(i, j int) bool {
+					return snapshots[i] < snapshots[j]
+				})
+
+				iter := newIter(formatVersion)
+				var b bytes.Buffer
+				for _, line := range strings.Split(d.Input, "\n") {
+					parts := strings.Fields(line)
+					if len(parts) == 0 {
+						continue
 					}
-					for _, v := range iter.Tombstones(key, false) {
-						fmt.Fprintf(&b, "%s-%s#%d\n",
-							v.Start.UserKey, v.End, v.Start.SeqNum())
+					switch parts[0] {
+					case "first":
+						iter.First()
+					case "next":
+						iter.Next()
+					case "tombstones":
+						var key []byte
+						if len(parts) == 2 {
+							key = []byte(parts[1])
+						}
+						for _, v := range iter.Tombstones(key, false) {
+							fmt.Fprintf(&b, "%s-%s#%d\n",
+								v.Start.UserKey, v.End, v.Start.SeqNum())
+						}
+						fmt.Fprintf(&b, ".\n")
+						continue
+					default:
+						return fmt.Sprintf("unknown op: %s", parts[0])
 					}
-					fmt.Fprintf(&b, ".\n")
-					continue
-				default:
-					return fmt.Sprintf("unknown op: %s", parts[0])
-				}
-				if iter.Valid() {
-					fmt.Fprintf(&b, "%s:%s\n", iter.Key(), iter.Value())
-					if iter.Key().Kind() == InternalKeyKindRangeDelete {
-						iter.rangeDelFrag.Add(iter.cloneKey(iter.Key()), iter.Value())
+					if iter.Valid() {
+						fmt.Fprintf(&b, "%s:%s\n", iter.Key(), iter.Value())
+						if iter.Key().Kind() == InternalKeyKindRangeDelete {
+							iter.rangeDelFrag.Add(iter.cloneKey(iter.Key()), iter.Value())
+						}
+					} else if err := iter.Error(); err != nil {
+						fmt.Fprintf(&b, "err=%v\n", err)
+					} else {
+						fmt.Fprintf(&b, ".\n")
 					}
-				} else if err := iter.Error(); err != nil {
-					fmt.Fprintf(&b, "err=%v\n", err)
-				} else {
-					fmt.Fprintf(&b, ".\n")
 				}
+				return b.String()
+
+			default:
+				return fmt.Sprintf("unknown command: %s", d.Cmd)
 			}
-			return b.String()
+		})
+	}
 
-		default:
-			return fmt.Sprintf("unknown command: %s", d.Cmd)
-		}
-	})
+	// Rather than testing against all format version, we test against the
+	// significant boundaries.
+	formatVersions := []FormatMajorVersion{
+		FormatMostCompatible,
+		FormatSetWithDelete - 1,
+		FormatSetWithDelete,
+		FormatNewest,
+	}
+	for _, formatVersion := range formatVersions {
+		t.Run(fmt.Sprintf("version-%s", formatVersion), func(t *testing.T) {
+			runTest(t, formatVersion)
+		})
+	}
 }
diff --git a/compaction_test.go b/compaction_test.go
index f134d5f2..dbdb2cc6 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -1103,7 +1103,15 @@ func TestManualCompaction(t *testing.T) {
 		}
 	}()
 
-	reset := func() {
+	seed := time.Now().UnixNano()
+	rng := rand.New(rand.NewSource(seed))
+	t.Logf("seed: %d", seed)
+
+	randVersion := func(min, max FormatMajorVersion) FormatMajorVersion {
+		return FormatMajorVersion(int(min) + rng.Intn(int(max)-int(min)+1))
+	}
+
+	reset := func(minVersion, maxVersion FormatMajorVersion) {
 		if d != nil {
 			require.NoError(t, d.Close())
 		}
@@ -1111,26 +1119,24 @@ func TestManualCompaction(t *testing.T) {
 		require.NoError(t, mem.MkdirAll("ext", 0755))
 
 		opts := &Options{
-			FS:         mem,
-			DebugCheck: DebugCheckLevels,
+			FS:                 mem,
+			DebugCheck:         DebugCheckLevels,
+			FormatMajorVersion: randVersion(minVersion, maxVersion),
 		}
 		opts.private.disableAutomaticCompactions = true
-		opts.testingRandomized()
 
 		var err error
 		d, err = Open("", opts)
 		require.NoError(t, err)
 	}
-	reset()
-
-	var ongoingCompaction *compaction
 
-	paths := []string{"testdata/manual_compaction", "testdata/singledel_manual_compaction"}
-	for _, path := range paths {
-		datadriven.RunTest(t, path, func(td *datadriven.TestData) string {
+	runTest := func(t *testing.T, testData string, minVersion, maxVersion FormatMajorVersion) {
+		reset(minVersion, maxVersion)
+		var ongoingCompaction *compaction
+		datadriven.RunTest(t, testData, func(td *datadriven.TestData) string {
 			switch td.Cmd {
 			case "reset":
-				reset()
+				reset(minVersion, maxVersion)
 				return ""
 
 			case "batch":
@@ -1162,8 +1168,9 @@ func TestManualCompaction(t *testing.T) {
 
 				mem = vfs.NewMem()
 				opts := &Options{
-					FS:         mem,
-					DebugCheck: DebugCheckLevels,
+					FS:                 mem,
+					DebugCheck:         DebugCheckLevels,
+					FormatMajorVersion: randVersion(minVersion, maxVersion),
 				}
 				opts.private.disableAutomaticCompactions = true
 
@@ -1173,7 +1180,7 @@ func TestManualCompaction(t *testing.T) {
 				}
 
 				d.mu.Lock()
-				s := d.mu.versions.currentVersion().String()
+				s := d.mu.versions.currentVersion().DebugString(base.DefaultFormatter)
 				d.mu.Unlock()
 				return s
 
@@ -1284,11 +1291,58 @@ func TestManualCompaction(t *testing.T) {
 			case "wait-pending-table-stats":
 				return runTableStatsCmd(td, d)
 
+			case "close-snapshots":
+				d.mu.Lock()
+				var ss []*Snapshot
+				l := &d.mu.snapshots
+				for i := l.root.next; i != &l.root; i = i.next {
+					ss = append(ss, i)
+				}
+				d.mu.Unlock()
+				for i := range ss {
+					if err := ss[i].Close(); err != nil {
+						return err.Error()
+					}
+				}
+				return ""
 			default:
 				return fmt.Sprintf("unknown command: %s", td.Cmd)
 			}
 		})
 	}
+
+	testCases := []struct {
+		testData   string
+		minVersion FormatMajorVersion
+		maxVersion FormatMajorVersion // inclusive
+	}{
+		{
+			testData:   "testdata/manual_compaction",
+			minVersion: FormatMostCompatible,
+			maxVersion: FormatSetWithDelete - 1,
+		},
+		{
+			testData:   "testdata/manual_compaction_set_with_del",
+			minVersion: FormatSetWithDelete,
+			maxVersion: FormatNewest,
+		},
+		{
+			testData:   "testdata/singledel_manual_compaction",
+			minVersion: FormatMostCompatible,
+			maxVersion: FormatSetWithDelete - 1,
+		},
+		{
+			testData:   "testdata/singledel_manual_compaction_set_with_del",
+			minVersion: FormatSetWithDelete,
+			maxVersion: FormatNewest,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.testData, func(t *testing.T) {
+			runTest(t, tc.testData, tc.minVersion, tc.maxVersion)
+		})
+	}
 }
 
 func TestCompactionFindGrandparentLimit(t *testing.T) {
diff --git a/error_test.go b/error_test.go
index 165f5a83..0ba36f86 100644
--- a/error_test.go
+++ b/error_test.go
@@ -6,6 +6,7 @@ package pebble
 
 import (
 	"bytes"
+	"fmt"
 	"math"
 	"strings"
 	"sync/atomic"
@@ -164,13 +165,14 @@ func TestErrors(t *testing.T) {
 // cannot require operations fail since it involves flush/compaction, which retry
 // internally and succeed following an injected error.
 func TestRequireReadError(t *testing.T) {
-	run := func(index int32) (err error) {
+	run := func(formatVersion FormatMajorVersion, index int32) (err error) {
 		// Perform setup with error injection disabled as it involves writes/background ops.
 		inj := errorfs.OnIndex(-1)
 		fs := errorfs.Wrap(vfs.NewMem(), inj)
 		opts := &Options{
-			FS:     fs,
-			Logger: panicLogger{},
+			FS:                 fs,
+			Logger:             panicLogger{},
+			FormatMajorVersion: formatVersion,
 		}
 		opts.private.disableTableStats = true
 		d, err := Open("", opts)
@@ -192,12 +194,21 @@ func TestRequireReadError(t *testing.T) {
 		require.NoError(t, d.DeleteRange(key1, key2, nil))
 		require.NoError(t, d.Set(key1, value, nil))
 		require.NoError(t, d.Flush())
-		expectLSM(`
+		if formatVersion < FormatSetWithDelete {
+			expectLSM(`
 0.0:
   000007:[a1#4,SET-a2#72057594037927935,RANGEDEL]
 6:
   000005:[a1#1,SET-a2#2,SET]
 `, d, t)
+		} else {
+			expectLSM(`
+0.0:
+  000007:[a1#4,SETWITHDEL-a2#72057594037927935,RANGEDEL]
+6:
+  000005:[a1#1,SET-a2#2,SET]
+`, d, t)
+		}
 
 		// Now perform foreground ops with error injection enabled.
 		inj.SetIndex(index)
@@ -235,12 +246,17 @@ func TestRequireReadError(t *testing.T) {
 		return nil
 	}
 
-	for i := int32(0); ; i++ {
-		err := run(i)
-		if err == nil {
-			t.Logf("no failures reported at index %d", i)
-			break
-		}
+	versions := []FormatMajorVersion{FormatMostCompatible, FormatSetWithDelete}
+	for _, version := range versions {
+		t.Run(fmt.Sprintf("version-%s", version), func(t *testing.T) {
+			for i := int32(0); ; i++ {
+				err := run(version, i)
+				if err == nil {
+					t.Logf("no failures reported at index %d", i)
+					break
+				}
+			}
+		})
 	}
 }
 
@@ -248,15 +264,16 @@ func TestRequireReadError(t *testing.T) {
 // corruption and return an error. In this case the filesystem reads return
 // successful status but the data they return is corrupt.
 func TestCorruptReadError(t *testing.T) {
-	run := func(index int32) (err error) {
+	run := func(formatVersion FormatMajorVersion, index int32) (err error) {
 		// Perform setup with corruption injection disabled as it involves writes/background ops.
 		fs := &corruptFS{
 			FS:    vfs.NewMem(),
 			index: -1,
 		}
 		opts := &Options{
-			FS:     fs,
-			Logger: panicLogger{},
+			FS:                 fs,
+			Logger:             panicLogger{},
+			FormatMajorVersion: formatVersion,
 		}
 		opts.private.disableTableStats = true
 		d, err := Open("", opts)
@@ -279,13 +296,23 @@ func TestCorruptReadError(t *testing.T) {
 		require.NoError(t, d.DeleteRange(key1, key2, nil))
 		require.NoError(t, d.Set(key1, value, nil))
 		require.NoError(t, d.Flush())
-		expectLSM(`
+		if formatVersion < FormatSetWithDelete {
+			expectLSM(`
 0.0:
   000007:[a1#4,SET-a2#72057594037927935,RANGEDEL]
 6:
   000005:[a1#1,SET-a2#2,SET]
 `, d, t)
 
+		} else {
+			expectLSM(`
+0.0:
+  000007:[a1#4,SETWITHDEL-a2#72057594037927935,RANGEDEL]
+6:
+  000005:[a1#1,SET-a2#2,SET]
+`, d, t)
+		}
+
 		// Now perform foreground ops with corruption injection enabled.
 		atomic.StoreInt32(&fs.index, index)
 		iter := d.NewIter(nil)
@@ -322,12 +349,17 @@ func TestCorruptReadError(t *testing.T) {
 		}
 		return nil
 	}
-	for i := int32(0); ; i++ {
-		err := run(i)
-		if err == nil {
-			t.Logf("no failures reported at index %d", i)
-			break
-		}
+	versions := []FormatMajorVersion{FormatMostCompatible, FormatSetWithDelete}
+	for _, version := range versions {
+		t.Run(fmt.Sprintf("version-%s", version), func(t *testing.T) {
+			for i := int32(0); ; i++ {
+				err := run(version, i)
+				if err == nil {
+					t.Logf("no failures reported at index %d", i)
+					break
+				}
+			}
+		})
 	}
 }
 
diff --git a/format_major_version_test.go b/format_major_version_test.go
index b6776bd5..275bd600 100644
--- a/format_major_version_test.go
+++ b/format_major_version_test.go
@@ -31,13 +31,15 @@ func TestRatchetFormat(t *testing.T) {
 	require.Equal(t, FormatVersioned, d.FormatMajorVersion())
 	require.NoError(t, d.RatchetFormatMajorVersion(FormatVersioned))
 	require.Equal(t, FormatVersioned, d.FormatMajorVersion())
+	require.NoError(t, d.RatchetFormatMajorVersion(FormatSetWithDelete))
+	require.Equal(t, FormatSetWithDelete, d.FormatMajorVersion())
 	require.NoError(t, d.Close())
 
 	// If we Open the database again, leaving the default format, the
-	// database should Open using the persisted FormatVersioned format.
+	// database should Open using the persisted FormatNewest.
 	d, err = Open("", &Options{FS: fs})
 	require.NoError(t, err)
-	require.Equal(t, FormatVersioned, d.FormatMajorVersion())
+	require.Equal(t, FormatNewest, d.FormatMajorVersion())
 	require.NoError(t, d.Close())
 
 	// Move the marker to a version that does not exist.
diff --git a/get_iter_test.go b/get_iter_test.go
index 8c22157c..68598d21 100644
--- a/get_iter_test.go
+++ b/get_iter_test.go
@@ -37,7 +37,7 @@ func TestGetIter(t *testing.T) {
 		{
 			description: "empty: an empty version",
 			queries: []string{
-				"abc.MAX.101 ErrNotFound",
+				"abc.SEPARATOR.101 ErrNotFound",
 			},
 		},
 
@@ -59,21 +59,21 @@ func TestGetIter(t *testing.T) {
 				},
 			},
 			queries: []string{
-				"aaa.MAX.105 ErrNotFound",
-				"cat.MAX.105 b",
-				"hat.MAX.105 ErrNotFound",
-				"mat.MAX.105 e",
-				"the.MAX.108 g",
-				"the.MAX.107 g",
-				"the.MAX.106 ErrNotFound",
-				"the.MAX.105 d",
-				"the.MAX.104 d",
-				"the.MAX.104 d",
-				"the.MAX.103 a",
-				"the.MAX.102 a",
-				"the.MAX.101 a",
-				"the.MAX.100 ErrNotFound",
-				"zzz.MAX.105 ErrNotFound",
+				"aaa.SEPARATOR.105 ErrNotFound",
+				"cat.SEPARATOR.105 b",
+				"hat.SEPARATOR.105 ErrNotFound",
+				"mat.SEPARATOR.105 e",
+				"the.SEPARATOR.108 g",
+				"the.SEPARATOR.107 g",
+				"the.SEPARATOR.106 ErrNotFound",
+				"the.SEPARATOR.105 d",
+				"the.SEPARATOR.104 d",
+				"the.SEPARATOR.104 d",
+				"the.SEPARATOR.103 a",
+				"the.SEPARATOR.102 a",
+				"the.SEPARATOR.101 a",
+				"the.SEPARATOR.100 ErrNotFound",
+				"zzz.SEPARATOR.105 ErrNotFound",
 			},
 		},
 
@@ -118,45 +118,45 @@ func TestGetIter(t *testing.T) {
 				},
 			},
 			queries: []string{
-				"aaa.MAX.105 ErrNotFound",
-				"awk.MAX.135 w",
-				"awk.MAX.125 w",
-				"awk.MAX.115 w",
-				"awk.MAX.105 ErrNotFound",
-				"cat.MAX.135 ErrNotFound",
-				"cat.MAX.125 ErrNotFound",
-				"cat.MAX.115 x",
-				"cat.MAX.105 b",
-				"man.MAX.135 G",
-				"man.MAX.125 ErrNotFound",
-				"man.MAX.115 y",
-				"man.MAX.105 ErrNotFound",
-				"on_.MAX.135 c",
-				"on_.MAX.125 c",
-				"on_.MAX.115 c",
-				"on_.MAX.105 c",
-				"the.MAX.135 F",
-				"the.MAX.127 F",
-				"the.MAX.126 F",
-				"the.MAX.125 ErrNotFound",
-				"the.MAX.122 ErrNotFound",
-				"the.MAX.121 ErrNotFound",
-				"the.MAX.120 g",
-				"the.MAX.115 g",
-				"the.MAX.114 g",
-				"the.MAX.111 g",
-				"the.MAX.110 g",
-				"the.MAX.108 g",
-				"the.MAX.107 g",
-				"the.MAX.106 ErrNotFound",
-				"the.MAX.105 d",
-				"the.MAX.104 d",
-				"the.MAX.104 d",
-				"the.MAX.103 a",
-				"the.MAX.102 a",
-				"the.MAX.101 a",
-				"the.MAX.100 ErrNotFound",
-				"zzz.MAX.105 ErrNotFound",
+				"aaa.SEPARATOR.105 ErrNotFound",
+				"awk.SEPARATOR.135 w",
+				"awk.SEPARATOR.125 w",
+				"awk.SEPARATOR.115 w",
+				"awk.SEPARATOR.105 ErrNotFound",
+				"cat.SEPARATOR.135 ErrNotFound",
+				"cat.SEPARATOR.125 ErrNotFound",
+				"cat.SEPARATOR.115 x",
+				"cat.SEPARATOR.105 b",
+				"man.SEPARATOR.135 G",
+				"man.SEPARATOR.125 ErrNotFound",
+				"man.SEPARATOR.115 y",
+				"man.SEPARATOR.105 ErrNotFound",
+				"on_.SEPARATOR.135 c",
+				"on_.SEPARATOR.125 c",
+				"on_.SEPARATOR.115 c",
+				"on_.SEPARATOR.105 c",
+				"the.SEPARATOR.135 F",
+				"the.SEPARATOR.127 F",
+				"the.SEPARATOR.126 F",
+				"the.SEPARATOR.125 ErrNotFound",
+				"the.SEPARATOR.122 ErrNotFound",
+				"the.SEPARATOR.121 ErrNotFound",
+				"the.SEPARATOR.120 g",
+				"the.SEPARATOR.115 g",
+				"the.SEPARATOR.114 g",
+				"the.SEPARATOR.111 g",
+				"the.SEPARATOR.110 g",
+				"the.SEPARATOR.108 g",
+				"the.SEPARATOR.107 g",
+				"the.SEPARATOR.106 ErrNotFound",
+				"the.SEPARATOR.105 d",
+				"the.SEPARATOR.104 d",
+				"the.SEPARATOR.104 d",
+				"the.SEPARATOR.103 a",
+				"the.SEPARATOR.102 a",
+				"the.SEPARATOR.101 a",
+				"the.SEPARATOR.100 ErrNotFound",
+				"zzz.SEPARATOR.105 ErrNotFound",
 			},
 		},
 
@@ -201,47 +201,47 @@ func TestGetIter(t *testing.T) {
 				},
 			},
 			queries: []string{
-				"a_______.MAX.999 ErrNotFound",
-				"aardvark.MAX.999 a1",
-				"aardvark.MAX.102 a1",
-				"aardvark.MAX.101 a1",
-				"aardvark.MAX.100 ErrNotFound",
-				"alpaca__.MAX.999 a2",
-				"alpaca__.MAX.200 ErrNotFound",
-				"anteater.MAX.999 a3",
-				"anteater.MAX.302 a3",
-				"anteater.MAX.301 a3",
-				"anteater.MAX.300 ErrNotFound",
-				"anteater.MAX.000 ErrNotFound",
-				"b_______.MAX.999 ErrNotFound",
-				"baboon__.MAX.999 b3",
-				"baboon__.MAX.302 b3",
-				"baboon__.MAX.301 ErrNotFound",
-				"baboon__.MAX.202 ErrNotFound",
-				"baboon__.MAX.201 b1",
-				"baboon__.MAX.102 b1",
-				"baboon__.MAX.101 ErrNotFound",
-				"bear____.MAX.999 ErrNotFound",
-				"bear____.MAX.500 b4",
-				"bear____.MAX.000 ErrNotFound",
-				"buffalo_.MAX.999 b6",
-				"buffalo_.MAX.603 b6",
-				"buffalo_.MAX.602 b6",
-				"buffalo_.MAX.601 B1",
-				"buffalo_.MAX.104 B1",
-				"buffalo_.MAX.103 B1",
-				"buffalo_.MAX.102 ErrNotFound",
-				"buffalo_.MAX.000 ErrNotFound",
-				"c_______.MAX.999 ErrNotFound",
-				"chipmunk.MAX.999 c2",
-				"chipmunk.MAX.205 c2",
-				"chipmunk.MAX.204 c2",
-				"chipmunk.MAX.203 c1",
-				"chipmunk.MAX.105 c1",
-				"chipmunk.MAX.104 c1",
-				"chipmunk.MAX.103 ErrNotFound",
-				"chipmunk.MAX.000 ErrNotFound",
-				"d_______.MAX.999 ErrNotFound",
+				"a_______.SEPARATOR.999 ErrNotFound",
+				"aardvark.SEPARATOR.999 a1",
+				"aardvark.SEPARATOR.102 a1",
+				"aardvark.SEPARATOR.101 a1",
+				"aardvark.SEPARATOR.100 ErrNotFound",
+				"alpaca__.SEPARATOR.999 a2",
+				"alpaca__.SEPARATOR.200 ErrNotFound",
+				"anteater.SEPARATOR.999 a3",
+				"anteater.SEPARATOR.302 a3",
+				"anteater.SEPARATOR.301 a3",
+				"anteater.SEPARATOR.300 ErrNotFound",
+				"anteater.SEPARATOR.000 ErrNotFound",
+				"b_______.SEPARATOR.999 ErrNotFound",
+				"baboon__.SEPARATOR.999 b3",
+				"baboon__.SEPARATOR.302 b3",
+				"baboon__.SEPARATOR.301 ErrNotFound",
+				"baboon__.SEPARATOR.202 ErrNotFound",
+				"baboon__.SEPARATOR.201 b1",
+				"baboon__.SEPARATOR.102 b1",
+				"baboon__.SEPARATOR.101 ErrNotFound",
+				"bear____.SEPARATOR.999 ErrNotFound",
+				"bear____.SEPARATOR.500 b4",
+				"bear____.SEPARATOR.000 ErrNotFound",
+				"buffalo_.SEPARATOR.999 b6",
+				"buffalo_.SEPARATOR.603 b6",
+				"buffalo_.SEPARATOR.602 b6",
+				"buffalo_.SEPARATOR.601 B1",
+				"buffalo_.SEPARATOR.104 B1",
+				"buffalo_.SEPARATOR.103 B1",
+				"buffalo_.SEPARATOR.102 ErrNotFound",
+				"buffalo_.SEPARATOR.000 ErrNotFound",
+				"c_______.SEPARATOR.999 ErrNotFound",
+				"chipmunk.SEPARATOR.999 c2",
+				"chipmunk.SEPARATOR.205 c2",
+				"chipmunk.SEPARATOR.204 c2",
+				"chipmunk.SEPARATOR.203 c1",
+				"chipmunk.SEPARATOR.105 c1",
+				"chipmunk.SEPARATOR.104 c1",
+				"chipmunk.SEPARATOR.103 ErrNotFound",
+				"chipmunk.SEPARATOR.000 ErrNotFound",
+				"d_______.SEPARATOR.999 ErrNotFound",
 			},
 		},
 
@@ -313,62 +313,62 @@ func TestGetIter(t *testing.T) {
 				},
 			},
 			queries: []string{
-				"a________.MAX.999 ErrNotFound",
-				"alfalfa__.MAX.520 p1",
-				"alfalfa__.MAX.510 p1",
-				"alfalfa__.MAX.500 r1",
-				"alfalfa__.MAX.400 u1",
-				"alfalfa__.MAX.300 ErrNotFound",
-				"asparagus.MAX.520 q3",
-				"asparagus.MAX.510 p2",
-				"asparagus.MAX.500 r2",
-				"asparagus.MAX.400 ErrNotFound",
-				"asparagus.MAX.300 ErrNotFound",
-				"arugula__.MAX.520 r3",
-				"arugula__.MAX.510 r3",
-				"arugula__.MAX.500 r3",
-				"arugula__.MAX.400 ErrNotFound",
-				"arugula__.MAX.300 ErrNotFound",
-				"beans____.MAX.520 q4",
-				"beans____.MAX.510 s2",
-				"beans____.MAX.500 s2",
-				"beans____.MAX.400 ErrNotFound",
-				"beans____.MAX.300 ErrNotFound",
-				"bokchoy__.MAX.520 ErrNotFound",
-				"bokchoy__.MAX.510 ErrNotFound",
-				"bokchoy__.MAX.500 ErrNotFound",
-				"bokchoy__.MAX.400 u2",
-				"bokchoy__.MAX.300 ErrNotFound",
-				"broccoli_.MAX.520 q5",
-				"broccoli_.MAX.510 s4",
-				"broccoli_.MAX.500 s4",
-				"broccoli_.MAX.400 ErrNotFound",
-				"broccoli_.MAX.300 ErrNotFound",
-				"cabbage__.MAX.520 q6",
-				"cabbage__.MAX.510 ErrNotFound",
-				"cabbage__.MAX.500 t1",
-				"cabbage__.MAX.400 ErrNotFound",
-				"cabbage__.MAX.300 ErrNotFound",
-				"celery___.MAX.520 q7",
-				"celery___.MAX.510 u3",
-				"celery___.MAX.500 u3",
-				"celery___.MAX.400 u3",
-				"celery___.MAX.300 ErrNotFound",
-				"corn_____.MAX.520 ErrNotFound",
-				"corn_____.MAX.510 ErrNotFound",
-				"corn_____.MAX.500 ErrNotFound",
-				"corn_____.MAX.400 u4",
-				"corn_____.MAX.300 ErrNotFound",
-				"d________.MAX.999 ErrNotFound",
-				"spinach__.MAX.999 u5t2p3q8",
-				"spinach__.MAX.518 u5t2p3q8",
-				"spinach__.MAX.517 u5t2p3",
-				"spinach__.MAX.504 u5t2p3",
-				"spinach__.MAX.503 u5t2",
-				"spinach__.MAX.432 u5t2",
-				"spinach__.MAX.431 u5",
-				"spinach__.MAX.350 u5",
-				"spinach__.MAX.349 ErrNotFound",
+				"a________.SEPARATOR.999 ErrNotFound",
+				"alfalfa__.SEPARATOR.520 p1",
+				"alfalfa__.SEPARATOR.510 p1",
+				"alfalfa__.SEPARATOR.500 r1",
+				"alfalfa__.SEPARATOR.400 u1",
+				"alfalfa__.SEPARATOR.300 ErrNotFound",
+				"asparagus.SEPARATOR.520 q3",
+				"asparagus.SEPARATOR.510 p2",
+				"asparagus.SEPARATOR.500 r2",
+				"asparagus.SEPARATOR.400 ErrNotFound",
+				"asparagus.SEPARATOR.300 ErrNotFound",
+				"arugula__.SEPARATOR.520 r3",
+				"arugula__.SEPARATOR.510 r3",
+				"arugula__.SEPARATOR.500 r3",
+				"arugula__.SEPARATOR.400 ErrNotFound",
+				"arugula__.SEPARATOR.300 ErrNotFound",
+				"beans____.SEPARATOR.520 q4",
+				"beans____.SEPARATOR.510 s2",
+				"beans____.SEPARATOR.500 s2",
+				"beans____.SEPARATOR.400 ErrNotFound",
+				"beans____.SEPARATOR.300 ErrNotFound",
+				"bokchoy__.SEPARATOR.520 ErrNotFound",
+				"bokchoy__.SEPARATOR.510 ErrNotFound",
+				"bokchoy__.SEPARATOR.500 ErrNotFound",
+				"bokchoy__.SEPARATOR.400 u2",
+				"bokchoy__.SEPARATOR.300 ErrNotFound",
+				"broccoli_.SEPARATOR.520 q5",
+				"broccoli_.SEPARATOR.510 s4",
+				"broccoli_.SEPARATOR.500 s4",
+				"broccoli_.SEPARATOR.400 ErrNotFound",
+				"broccoli_.SEPARATOR.300 ErrNotFound",
+				"cabbage__.SEPARATOR.520 q6",
+				"cabbage__.SEPARATOR.510 ErrNotFound",
+				"cabbage__.SEPARATOR.500 t1",
+				"cabbage__.SEPARATOR.400 ErrNotFound",
+				"cabbage__.SEPARATOR.300 ErrNotFound",
+				"celery___.SEPARATOR.520 q7",
+				"celery___.SEPARATOR.510 u3",
+				"celery___.SEPARATOR.500 u3",
+				"celery___.SEPARATOR.400 u3",
+				"celery___.SEPARATOR.300 ErrNotFound",
+				"corn_____.SEPARATOR.520 ErrNotFound",
+				"corn_____.SEPARATOR.510 ErrNotFound",
+				"corn_____.SEPARATOR.500 ErrNotFound",
+				"corn_____.SEPARATOR.400 u4",
+				"corn_____.SEPARATOR.300 ErrNotFound",
+				"d________.SEPARATOR.999 ErrNotFound",
+				"spinach__.SEPARATOR.999 u5t2p3q8",
+				"spinach__.SEPARATOR.518 u5t2p3q8",
+				"spinach__.SEPARATOR.517 u5t2p3",
+				"spinach__.SEPARATOR.504 u5t2p3",
+				"spinach__.SEPARATOR.503 u5t2",
+				"spinach__.SEPARATOR.432 u5t2",
+				"spinach__.SEPARATOR.431 u5",
+				"spinach__.SEPARATOR.350 u5",
+				"spinach__.SEPARATOR.349 ErrNotFound",
 			},
 		},
 
diff --git a/internal/base/internal_test.go b/internal/base/internal_test.go
index b058d2ac..d41d0601 100644
--- a/internal/base/internal_test.go
+++ b/internal/base/internal_test.go
@@ -37,7 +37,7 @@ func TestInvalidInternalKey(t *testing.T) {
 		"\x01\x02\x03\x04\x05\x06\x07",
 		"foo",
 		"foo\x08\x07\x06\x05\x04\x03\x02",
-		"foo\x12\x07\x06\x05\x04\x03\x02\x01",
+		"foo\x13\x07\x06\x05\x04\x03\x02\x01",
 	}
 	for _, tc := range testCases {
 		k := DecodeInternalKey([]byte(tc))
@@ -124,12 +124,12 @@ func TestInternalKeySeparator(t *testing.T) {
 		{"foo.SET.100", "foo.DEL.100", "foo.SET.100"},
 		{"foo.SET.100", "foo.SET.101", "foo.SET.100"},
 		{"foo.SET.100", "bar.SET.99", "foo.SET.100"},
-		{"foo.SET.100", "hello.SET.200", "g.MAX.72057594037927935"},
-		{"ABC1AAAAA.SET.100", "ABC2ABB.SET.200", "ABC2.MAX.72057594037927935"},
-		{"AAA1AAA.SET.100", "AAA2AA.SET.200", "AAA2.MAX.72057594037927935"},
-		{"AAA1AAA.SET.100", "AAA4.SET.200", "AAA2.MAX.72057594037927935"},
-		{"AAA1AAA.SET.100", "AAA2.SET.200", "AAA1B.MAX.72057594037927935"},
-		{"AAA1AAA.SET.100", "AAA2A.SET.200", "AAA2.MAX.72057594037927935"},
+		{"foo.SET.100", "hello.SET.200", "g.SEPARATOR.72057594037927935"},
+		{"ABC1AAAAA.SET.100", "ABC2ABB.SET.200", "ABC2.SEPARATOR.72057594037927935"},
+		{"AAA1AAA.SET.100", "AAA2AA.SET.200", "AAA2.SEPARATOR.72057594037927935"},
+		{"AAA1AAA.SET.100", "AAA4.SET.200", "AAA2.SEPARATOR.72057594037927935"},
+		{"AAA1AAA.SET.100", "AAA2.SET.200", "AAA1B.SEPARATOR.72057594037927935"},
+		{"AAA1AAA.SET.100", "AAA2A.SET.200", "AAA2.SEPARATOR.72057594037927935"},
 		{"AAA1.SET.100", "AAA2.SET.200", "AAA1.SET.100"},
 		{"foo.SET.100", "foobar.SET.200", "foo.SET.100"},
 		{"foobar.SET.100", "foo.SET.200", "foobar.SET.100"},
diff --git a/internal/metamorphic/key_manager_test.go b/internal/metamorphic/key_manager_test.go
new file mode 100644
index 00000000..fccb0986
--- /dev/null
+++ b/internal/metamorphic/key_manager_test.go
@@ -0,0 +1,462 @@
+package metamorphic
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestObjKey(t *testing.T) {
+	testCases := []struct {
+		key  objKey
+		want string
+	}{
+		{
+			key:  makeObjKey(makeObjID(dbTag, 0), []byte("foo")),
+			want: "db:foo",
+		},
+		{
+			key:  makeObjKey(makeObjID(batchTag, 1), []byte("bar")),
+			want: "batch1:bar",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run("", func(t *testing.T) {
+			require.Equal(t, tc.want, tc.key.String())
+		})
+	}
+}
+
+func TestGlobalStateIndicatesEligibleForSingleDelete(t *testing.T) {
+	key := makeObjKey(makeObjID(dbTag, 0), []byte("foo"))
+	testCases := []struct {
+		meta keyMeta
+		want bool
+	}{
+		{
+			meta: keyMeta{
+				objKey: key,
+			},
+			want: false,
+		},
+		{
+			meta: keyMeta{
+				objKey: key,
+				sets:   1,
+			},
+			want: true,
+		},
+		{
+			meta: keyMeta{
+				objKey: key,
+				sets:   2,
+			},
+			want: false,
+		},
+		{
+			meta: keyMeta{
+				objKey: key,
+				sets:   1,
+				merges: 1,
+			},
+			want: false,
+		},
+		{
+			meta: keyMeta{
+				objKey: key,
+				sets:   1,
+				dels: 1,
+			},
+			want: false,
+		},
+		{
+			meta: keyMeta{
+				objKey: key,
+				sets:   1,
+				singleDel: true,
+			},
+			want: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		k := newKeyManager()
+		t.Run("", func(t *testing.T) {
+			k.globalKeysMap[string(key.key)] = &tc.meta
+			require.Equal(t, tc.want, k.globalStateIndicatesEligibleForSingleDelete(key.key))
+		})
+	}
+}
+
+func TestKeyMeta_MergeInto(t *testing.T) {
+	testCases := []struct {
+		existing keyMeta
+		toMerge keyMeta
+		expected keyMeta
+	}{
+		{
+			existing: keyMeta{
+				sets:      1,
+				merges:    0,
+				singleDel: false,
+			},
+			toMerge: keyMeta{
+				sets:      0,
+				merges:    0,
+				singleDel: true,
+			},
+			expected: keyMeta {
+				sets:      1,
+				merges:    0,
+				singleDel: true,
+			},
+		},
+		{
+			existing: keyMeta{
+				sets:      3,
+				merges:    1,
+				dels: 7,
+			},
+			toMerge: keyMeta{
+				sets:      4,
+				merges:    2,
+				dels: 8,
+				del: true,
+			},
+			expected: keyMeta {
+				sets:      7,
+				merges:    3,
+				dels: 15,
+				del: true,
+			},
+		},
+		{
+			existing: keyMeta{
+				sets:      3,
+				merges:    1,
+				dels: 7,
+				del: true,
+			},
+			toMerge: keyMeta{
+				sets:      1,
+				merges:    0,
+				dels: 8,
+				del: false,
+			},
+			expected: keyMeta {
+				sets:      4,
+				merges:    1,
+				dels: 15,
+				del: false,
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run("", func(t *testing.T) {
+			tc.toMerge.mergeInto(&tc.existing)
+			require.Equal(t, tc.expected, tc.existing)
+		})
+	}
+}
+
+func TestKeyManager_AddKey(t *testing.T) {
+	m := newKeyManager()
+	require.Empty(t, m.globalKeys)
+
+	k1 := []byte("foo")
+	require.True(t, m.addNewKey(k1))
+	require.Len(t, m.globalKeys, 1)
+	require.Contains(t, m.globalKeys, k1)
+	require.False(t, m.addNewKey(k1))
+	k2 := []byte("bar")
+	require.True(t, m.addNewKey(k2))
+	require.Len(t, m.globalKeys, 2)
+	require.Contains(t, m.globalKeys, k2)
+}
+
+func TestKeyManager_GetOrInit(t *testing.T) {
+	id := makeObjID(batchTag, 1)
+	key := []byte("foo")
+	o := makeObjKey(id, key)
+
+	m := newKeyManager()
+	require.NotContains(t, m.byObjKey, o.String())
+	require.NotContains(t, m.byObj, id)
+	require.Contains(t, m.byObj, makeObjID(dbTag, 0)) // Always contains the DB key.
+
+	meta1 := m.getOrInit(id, key)
+	require.Contains(t, m.byObjKey, o.String())
+	require.Contains(t, m.byObj, id)
+
+	// Idempotent.
+	meta2 := m.getOrInit(id, key)
+	require.Equal(t, meta1, meta2)
+}
+
+func TestKeyManager_Contains(t *testing.T) {
+	id := makeObjID(dbTag, 0)
+	key := []byte("foo")
+
+	m := newKeyManager()
+	require.False(t, m.contains(id, key))
+
+	m.getOrInit(id, key)
+	require.True(t, m.contains(id, key))
+}
+
+func TestKeyManager_MergeInto(t *testing.T) {
+	fromID := makeObjID(batchTag, 1)
+	toID := makeObjID(dbTag, 0)
+
+	m := newKeyManager()
+
+	// Two keys in "from".
+	a := m.getOrInit(fromID, []byte("foo"))
+	a.sets = 1
+	b := m.getOrInit(fromID, []byte("bar"))
+	b.merges = 2
+
+	// One key in "to", with same value as a key in "from", that will be merged.
+	m.getOrInit(toID, []byte("foo"))
+
+	// Before, there are two sets.
+	require.Len(t, m.byObj[fromID], 2)
+	require.Len(t, m.byObj[toID], 1)
+
+	m.mergeKeysInto(fromID, toID)
+
+	// Keys in "from" sets are moved to "to" set.
+	require.Len(t, m.byObj[toID], 2)
+
+	// Key "foo" was merged into "to".
+	foo := m.getOrInit(toID, []byte("foo"))
+	require.Equal(t, 1, foo.sets) // value was merged.
+
+	// Key "bar" was merged into "to".
+	bar := m.getOrInit(toID, []byte("bar"))
+	require.Equal(t, 2, bar.merges) // value was unchanged.
+
+	// Keys in "from" sets are removed from maps.
+	require.NotContains(t, m.byObjKey, makeObjKey(fromID, a.key))
+	require.NotContains(t, m.byObjKey, makeObjKey(fromID, b.key))
+	require.NotContains(t, m.byObj, fromID)
+}
+
+type seqFn func(t *testing.T, k *keyManager)
+
+func updateForOp(op op) seqFn {
+	return func(t *testing.T, k *keyManager) {
+		k.update(op)
+	}
+}
+
+func addKey(key []byte, expected bool) seqFn {
+	return func(t *testing.T, k *keyManager) {
+		require.Equal(t, expected, k.addNewKey(key))
+	}
+}
+
+func eligibleRead(key []byte, val bool) seqFn {
+	return func(t *testing.T, k *keyManager) {
+		require.Equal(t, val, contains(key, k.eligibleReadKeys()))
+	}
+}
+
+func eligibleWrite(key []byte, val bool) seqFn {
+	return func(t *testing.T, k *keyManager) {
+		require.Equal(t, val, contains(key, k.eligibleWriteKeys()))
+	}
+}
+
+func eligibleSingleDelete(key []byte, val bool, id objID) seqFn {
+	return func(t *testing.T, k *keyManager) {
+		require.Equal(t, val, contains(key, k.eligibleSingleDeleteKeys(id)))
+	}
+}
+
+func contains(key []byte, keys [][]byte) bool {
+	for _, k := range keys {
+		if bytes.Equal(key, k) {
+			return true
+		}
+	}
+	return false
+}
+
+func TestKeyManager(t *testing.T) {
+	var (
+		id1  = makeObjID(batchTag, 0)
+		id2  = makeObjID(batchTag, 1)
+		key1 = []byte("foo")
+	)
+
+	testCases := []struct {
+		description string
+		ops         []seqFn
+		wantPanic   bool
+	}{
+		{
+			description: "set, single del, on db",
+			ops: []seqFn{
+				addKey(key1, true),
+				addKey(key1, false),
+				eligibleRead(key1, true),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				eligibleSingleDelete(key1, false, id1),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleRead(key1, true),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, true, dbObjID),
+				eligibleSingleDelete(key1, true, id1),
+				updateForOp(&singleDeleteOp{writerID: dbObjID, key: key1}),
+				eligibleRead(key1, true),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+			},
+		},
+		{
+			description: "set, single del, on batch",
+			ops: []seqFn{
+				addKey(key1, true),
+				updateForOp(&setOp{writerID: id1, key: key1}),
+				eligibleRead(key1, true),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				eligibleSingleDelete(key1, true, id1),
+				eligibleSingleDelete(key1, false, id2),
+				updateForOp(&singleDeleteOp{writerID: id1, key: key1}),
+				eligibleRead(key1, true),
+				eligibleWrite(key1, false),
+				eligibleSingleDelete(key1, false, dbObjID),
+				eligibleSingleDelete(key1, false, id1),
+				updateForOp(&applyOp{batchID: id1, writerID: dbObjID}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+			},
+		},
+		{
+			description: "set on db, single del on batch",
+			ops: []seqFn{
+				addKey(key1, true),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, true, dbObjID),
+				eligibleSingleDelete(key1, true, id1),
+				updateForOp(&singleDeleteOp{writerID: id1, key: key1}),
+				eligibleWrite(key1, false),
+				eligibleSingleDelete(key1, false, dbObjID),
+				eligibleSingleDelete(key1, false, id1),
+				updateForOp(&applyOp{batchID: id1, writerID: dbObjID}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleSingleDelete(key1, true, dbObjID),
+				eligibleSingleDelete(key1, true, id1),
+			},
+		},
+		{
+			description: "set, del, set, single del, on db",
+			ops: []seqFn{
+				addKey(key1, true),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				updateForOp(&deleteOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, true, dbObjID),
+				updateForOp(&singleDeleteOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+			},
+		},
+		{
+			description: "set, del, set, del, on batches",
+			ops: []seqFn{
+				addKey(key1, true),
+				updateForOp(&setOp{writerID: id1, key: key1}),
+				updateForOp(&deleteOp{writerID: id1, key: key1}),
+				updateForOp(&setOp{writerID: id1, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, id1),
+				updateForOp(&applyOp{batchID: id1, writerID: dbObjID}),
+				eligibleWrite(key1, true),
+				// Not eligible for single del since the set count is 2.
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				// Not eligible for single del since the set count is 3.
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&deleteOp{writerID: id2, key: key1}),
+				updateForOp(&applyOp{batchID: id2, writerID: dbObjID}),
+				// Set count is 0.
+				eligibleSingleDelete(key1, false, dbObjID),
+				// Set count is 1.
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleSingleDelete(key1, true, dbObjID),
+			},
+		},
+		{
+			description: "set, merge, del, set, single del, on db",
+			ops: []seqFn{
+				addKey(key1, true),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleSingleDelete(key1, true, dbObjID),
+				updateForOp(&mergeOp{writerID: dbObjID, key: key1}),
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&deleteOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, true, dbObjID),
+				updateForOp(&singleDeleteOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+			},
+		},
+		{
+			description: "set, del on db, set, single del on batch",
+			ops: []seqFn{
+				addKey(key1, true),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleSingleDelete(key1, true, dbObjID),
+				updateForOp(&deleteOp{writerID: dbObjID, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&setOp{writerID: id1, key: key1}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				eligibleSingleDelete(key1, true, id1),
+				updateForOp(&singleDeleteOp{writerID: id1, key: key1}),
+				eligibleWrite(key1, false),
+				eligibleSingleDelete(key1, false, id1),
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&applyOp{batchID: id1, writerID: dbObjID}),
+				eligibleWrite(key1, true),
+				eligibleSingleDelete(key1, false, dbObjID),
+				updateForOp(&setOp{writerID: dbObjID, key: key1}),
+				eligibleSingleDelete(key1, true, dbObjID),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.description, func(t *testing.T) {
+			m := newKeyManager()
+			tFunc := func() {
+				for _, op := range tc.ops {
+					op(t, m)
+				}
+			}
+			if tc.wantPanic {
+				require.Panics(t, tFunc)
+			} else {
+				tFunc()
+			}
+		})
+	}
+}
diff --git a/iterator_test.go b/iterator_test.go
index 3a4616d7..62e1c004 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -215,25 +215,22 @@ func (f *fakeIter) SetBounds(lower, upper []byte) {
 // invalidatingIter tests unsafe key/value slice reuse by modifying the last
 // returned key/value to all 1s.
 type invalidatingIter struct {
-	iter      internalIterator
-	lastKey   *InternalKey
-	lastValue []byte
+	iter        internalIterator
+	lastKey     *InternalKey
+	lastValue   []byte
+	ignoreKinds [base.InternalKeyKindMax + 1]bool
 }
 
 func newInvalidatingIter(iter internalIterator) *invalidatingIter {
 	return &invalidatingIter{iter: iter}
 }
 
+func (i *invalidatingIter) ignoreKind(kind base.InternalKeyKind) {
+	i.ignoreKinds[kind] = true
+}
+
 func (i *invalidatingIter) update(key *InternalKey, value []byte) (*InternalKey, []byte) {
-	if i.lastKey != nil {
-		for j := range i.lastKey.UserKey {
-			i.lastKey.UserKey[j] = 0xff
-		}
-		i.lastKey.Trailer = 0
-	}
-	for j := range i.lastValue {
-		i.lastValue[j] = 0xff
-	}
+	i.zeroLast()
 
 	if key == nil {
 		i.lastKey = nil
@@ -248,6 +245,25 @@ func (i *invalidatingIter) update(key *InternalKey, value []byte) (*InternalKey,
 	return i.lastKey, i.lastValue
 }
 
+func (i *invalidatingIter) zeroLast() {
+	if i.lastKey == nil {
+		return
+	}
+	if i.ignoreKinds[i.lastKey.Kind()] {
+		return
+	}
+
+	if i.lastKey != nil {
+		for j := range i.lastKey.UserKey {
+			i.lastKey.UserKey[j] = 0xff
+		}
+		i.lastKey.Trailer = 0
+	}
+	for j := range i.lastValue {
+		i.lastValue[j] = 0xff
+	}
+}
+
 func (i *invalidatingIter) SeekGE(key []byte) (*InternalKey, []byte) {
 	return i.update(i.iter.SeekGE(key))
 }
diff --git a/open_test.go b/open_test.go
index 58385bae..385754c8 100644
--- a/open_test.go
+++ b/open_test.go
@@ -99,7 +99,7 @@ func TestNewDBFilenames(t *testing.T) {
 			"LOCK",
 			"MANIFEST-000001",
 			"OPTIONS-000003",
-			"marker.format-version.000002.003",
+			"marker.format-version.000003.004",
 			"marker.manifest.000001.MANIFEST-000001",
 		},
 	}
diff --git a/range_del_test.go b/range_del_test.go
index dd406e9f..db1866c3 100644
--- a/range_del_test.go
+++ b/range_del_test.go
@@ -156,73 +156,75 @@ func TestDeleteRangeFlushDelay(t *testing.T) {
 // problem is that range tombstones are not truncated to sstable boundaries on
 // disk, only in memory.
 func TestRangeDelCompactionTruncation(t *testing.T) {
-	// Use a small target file size so that there is a single key per sstable.
-	d, err := Open("", &Options{
-		FS: vfs.NewMem(),
-		Levels: []LevelOptions{
-			{TargetFileSize: 100},
-			{TargetFileSize: 100},
-			{TargetFileSize: 1},
-		},
-		DebugCheck: DebugCheckLevels,
-	})
-	require.NoError(t, err)
-	defer d.Close()
-
-	d.mu.Lock()
-	d.mu.versions.dynamicBaseLevel = false
-	d.mu.Unlock()
+	runTest := func(formatVersion FormatMajorVersion) {
+		// Use a small target file size so that there is a single key per sstable.
+		d, err := Open("", &Options{
+			FS: vfs.NewMem(),
+			Levels: []LevelOptions{
+				{TargetFileSize: 100},
+				{TargetFileSize: 100},
+				{TargetFileSize: 1},
+			},
+			DebugCheck:         DebugCheckLevels,
+			FormatMajorVersion: formatVersion,
+		})
+		require.NoError(t, err)
+		defer d.Close()
 
-	lsm := func() string {
 		d.mu.Lock()
-		s := d.mu.versions.currentVersion().DebugString(base.DefaultFormatter)
+		d.mu.versions.dynamicBaseLevel = false
 		d.mu.Unlock()
-		return s
-	}
-	expectLSM := func(expected string) {
-		t.Helper()
-		expected = strings.TrimSpace(expected)
-		actual := strings.TrimSpace(lsm())
-		if expected != actual {
-			t.Fatalf("expected\n%s\nbut found\n%s", expected, actual)
-		}
-	}
-
-	require.NoError(t, d.Set([]byte("a"), bytes.Repeat([]byte("b"), 100), nil))
-	snap1 := d.NewSnapshot()
-	defer snap1.Close()
-	// Flush so that each version of "a" ends up in its own L0 table. If we
-	// allowed both versions in the same L0 table, compaction could trivially
-	// move the single L0 table to L1.
-	require.NoError(t, d.Flush())
-	require.NoError(t, d.Set([]byte("b"), bytes.Repeat([]byte("c"), 100), nil))
 
-	snap2 := d.NewSnapshot()
-	defer snap2.Close()
-	require.NoError(t, d.DeleteRange([]byte("a"), []byte("d"), nil))
+		lsm := func() string {
+			d.mu.Lock()
+			s := d.mu.versions.currentVersion().DebugString(base.DefaultFormatter)
+			d.mu.Unlock()
+			return s
+		}
+		expectLSM := func(expected string) {
+			t.Helper()
+			expected = strings.TrimSpace(expected)
+			actual := strings.TrimSpace(lsm())
+			if expected != actual {
+				t.Fatalf("expected\n%s\nbut found\n%s", expected, actual)
+			}
+		}
 
-	// Compact to produce the L1 tables.
-	require.NoError(t, d.Compact([]byte("c"), []byte("c\x00")))
-	expectLSM(`
+		require.NoError(t, d.Set([]byte("a"), bytes.Repeat([]byte("b"), 100), nil))
+		snap1 := d.NewSnapshot()
+		defer snap1.Close()
+		// Flush so that each version of "a" ends up in its own L0 table. If we
+		// allowed both versions in the same L0 table, compaction could trivially
+		// move the single L0 table to L1.
+		require.NoError(t, d.Flush())
+		require.NoError(t, d.Set([]byte("b"), bytes.Repeat([]byte("c"), 100), nil))
+
+		snap2 := d.NewSnapshot()
+		defer snap2.Close()
+		require.NoError(t, d.DeleteRange([]byte("a"), []byte("d"), nil))
+
+		// Compact to produce the L1 tables.
+		require.NoError(t, d.Compact([]byte("c"), []byte("c\x00")))
+		expectLSM(`
 1:
   000008:[a#3,RANGEDEL-b#72057594037927935,RANGEDEL]
   000009:[b#3,RANGEDEL-d#72057594037927935,RANGEDEL]
 `)
 
-	// Compact again to move one of the tables to L2.
-	require.NoError(t, d.Compact([]byte("c"), []byte("c\x00")))
-	expectLSM(`
+		// Compact again to move one of the tables to L2.
+		require.NoError(t, d.Compact([]byte("c"), []byte("c\x00")))
+		expectLSM(`
 1:
   000008:[a#3,RANGEDEL-b#72057594037927935,RANGEDEL]
 2:
   000009:[b#3,RANGEDEL-d#72057594037927935,RANGEDEL]
 `)
 
-	// Write "b" and "c" to a new table.
-	require.NoError(t, d.Set([]byte("b"), []byte("d"), nil))
-	require.NoError(t, d.Set([]byte("c"), []byte("e"), nil))
-	require.NoError(t, d.Flush())
-	expectLSM(`
+		// Write "b" and "c" to a new table.
+		require.NoError(t, d.Set([]byte("b"), []byte("d"), nil))
+		require.NoError(t, d.Set([]byte("c"), []byte("e"), nil))
+		require.NoError(t, d.Flush())
+		expectLSM(`
 0.0:
   000011:[b#4,SET-c#5,SET]
 1:
@@ -231,36 +233,37 @@ func TestRangeDelCompactionTruncation(t *testing.T) {
   000009:[b#3,RANGEDEL-d#72057594037927935,RANGEDEL]
 `)
 
-	// "b" is still visible at this point as it should be.
-	if _, closer, err := d.Get([]byte("b")); err != nil {
-		t.Fatalf("expected success, but found %v", err)
-	} else {
-		closer.Close()
-	}
+		// "b" is still visible at this point as it should be.
+		if _, closer, err := d.Get([]byte("b")); err != nil {
+			t.Fatalf("expected success, but found %v", err)
+		} else {
+			closer.Close()
+		}
 
-	keys := func() string {
-		iter := d.NewIter(nil)
-		defer iter.Close()
-		var buf bytes.Buffer
-		var sep string
-		for iter.First(); iter.Valid(); iter.Next() {
-			fmt.Fprintf(&buf, "%s%s", sep, iter.Key())
-			sep = " "
+		keys := func() string {
+			iter := d.NewIter(nil)
+			defer iter.Close()
+			var buf bytes.Buffer
+			var sep string
+			for iter.First(); iter.Valid(); iter.Next() {
+				fmt.Fprintf(&buf, "%s%s", sep, iter.Key())
+				sep = " "
+			}
+			return buf.String()
 		}
-		return buf.String()
-	}
 
-	if expected, actual := `b c`, keys(); expected != actual {
-		t.Fatalf("expected %q, but found %q", expected, actual)
-	}
+		if expected, actual := `b c`, keys(); expected != actual {
+			t.Fatalf("expected %q, but found %q", expected, actual)
+		}
 
-	// Compact the L0 table. This will compact the L0 table into L1 and do to the
-	// sstable target size settings will create 2 tables in L1. Then L1 table
-	// containing "c" will be compacted again with the L2 table creating two
-	// tables in L2. Lastly, the L2 table containing "c" will be compacted
-	// creating the L3 table.
-	require.NoError(t, d.Compact([]byte("c"), []byte("c\x00")))
-	expectLSM(`
+		// Compact the L0 table. This will compact the L0 table into L1 and do to the
+		// sstable target size settings will create 2 tables in L1. Then L1 table
+		// containing "c" will be compacted again with the L2 table creating two
+		// tables in L2. Lastly, the L2 table containing "c" will be compacted
+		// creating the L3 table.
+		require.NoError(t, d.Compact([]byte("c"), []byte("c\x00")))
+		if formatVersion < FormatSetWithDelete {
+			expectLSM(`
 1:
   000012:[a#3,RANGEDEL-b#72057594037927935,RANGEDEL]
 3:
@@ -268,17 +271,40 @@ func TestRangeDelCompactionTruncation(t *testing.T) {
   000018:[b#3,RANGEDEL-c#72057594037927935,RANGEDEL]
   000019:[c#5,SET-d#72057594037927935,RANGEDEL]
 `)
+		} else {
+			expectLSM(`
+1:
+  000012:[a#3,RANGEDEL-b#72057594037927935,RANGEDEL]
+3:
+  000017:[b#4,SETWITHDEL-b#4,SETWITHDEL]
+  000018:[b#3,RANGEDEL-c#72057594037927935,RANGEDEL]
+  000019:[c#5,SETWITHDEL-d#72057594037927935,RANGEDEL]
+`)
+		}
 
-	// The L1 table still contains a tombstone from [a,d) which will improperly
-	// delete the newer version of "b" in L2.
-	if _, closer, err := d.Get([]byte("b")); err != nil {
-		t.Errorf("expected success, but found %v", err)
-	} else {
-		closer.Close()
+		// The L1 table still contains a tombstone from [a,d) which will improperly
+		// delete the newer version of "b" in L2.
+		if _, closer, err := d.Get([]byte("b")); err != nil {
+			t.Errorf("expected success, but found %v", err)
+		} else {
+			closer.Close()
+		}
+
+		if expected, actual := `b c`, keys(); expected != actual {
+			t.Errorf("expected %q, but found %q", expected, actual)
+		}
 	}
 
-	if expected, actual := `b c`, keys(); expected != actual {
-		t.Errorf("expected %q, but found %q", expected, actual)
+	versions := []FormatMajorVersion{
+		FormatMostCompatible,
+		FormatSetWithDelete - 1,
+		FormatSetWithDelete,
+		FormatNewest,
+	}
+	for _, version := range versions {
+		t.Run(fmt.Sprintf("version-%s", version), func(t *testing.T) {
+			runTest(version)
+		})
 	}
 }
 
diff --git a/sstable/testdata/writer b/sstable/testdata/writer
index c03a3d4a..86878e9e 100644
--- a/sstable/testdata/writer
+++ b/sstable/testdata/writer
@@ -14,11 +14,11 @@ scan-range-del
 
 build
 a.SET.1:a
-b.DEL.2:b
+b.DEL.2:
 c.MERGE.3:c
 d.RANGEDEL.4:e
 f.SET.5:f
-g.DEL.6:g
+g.DEL.6:
 h.MERGE.7:h
 i.RANGEDEL.8:j
 ----
@@ -28,11 +28,11 @@ seqnums: [1,8]
 
 build
 a.SET.1:a
-b.DEL.2:b
+b.DEL.2:
 c.MERGE.3:c
 d.RANGEDEL.4:e
 f.SET.5:f
-g.DEL.6:g
+g.DEL.6:
 h.MERGE.7:h
 i.RANGEDEL.8:j
 ----
@@ -43,10 +43,10 @@ seqnums: [1,8]
 scan
 ----
 a#1,1:a
-b#2,0:b
+b#2,0:
 c#3,2:c
 f#5,1:f
-g#6,0:g
+g#6,0:
 h#7,2:h
 
 scan-range-del
diff --git a/testdata/checkpoint b/testdata/checkpoint
index f0bcb14f..3ae93382 100644
--- a/testdata/checkpoint
+++ b/testdata/checkpoint
@@ -25,6 +25,9 @@ rename: db/CURRENT.000000.dbtmp -> db/CURRENT
 create: db/marker.format-version.000002.003
 close: db/marker.format-version.000002.003
 sync: db
+create: db/marker.format-version.000003.004
+close: db/marker.format-version.000003.004
+sync: db
 sync: db/MANIFEST-000001
 create: db/000002.log
 sync: db
@@ -89,9 +92,9 @@ close:
 open-dir: checkpoints/checkpoint1
 link: db/OPTIONS-000003 -> checkpoints/checkpoint1/OPTIONS-000003
 open-dir: checkpoints/checkpoint1
-create: checkpoints/checkpoint1/marker.format-version.000001.003
-sync: checkpoints/checkpoint1/marker.format-version.000001.003
-close: checkpoints/checkpoint1/marker.format-version.000001.003
+create: checkpoints/checkpoint1/marker.format-version.000001.004
+sync: checkpoints/checkpoint1/marker.format-version.000001.004
+close: checkpoints/checkpoint1/marker.format-version.000001.004
 sync: checkpoints/checkpoint1
 close: checkpoints/checkpoint1
 create: checkpoints/checkpoint1/MANIFEST-000001
@@ -146,7 +149,7 @@ CURRENT
 LOCK
 MANIFEST-000001
 OPTIONS-000003
-marker.format-version.000002.003
+marker.format-version.000003.004
 marker.manifest.000001.MANIFEST-000001
 
 list checkpoints/checkpoint1
@@ -156,7 +159,7 @@ list checkpoints/checkpoint1
 000007.sst
 MANIFEST-000001
 OPTIONS-000003
-marker.format-version.000001.003
+marker.format-version.000001.004
 marker.manifest.000001.MANIFEST-000001
 
 open checkpoints/checkpoint1 readonly
diff --git a/testdata/compaction_iter b/testdata/compaction_iter
index 5cf678fd..e7038e8c 100644
--- a/testdata/compaction_iter
+++ b/testdata/compaction_iter
@@ -969,9 +969,9 @@ a-c#2
 
 define
 a.RANGEDEL.3:d
-a.DEL.2:a
+a.DEL.2:
 a.SET.1:a
-d.DEL.2:a
+d.DEL.2:
 ----
 
 iter
@@ -981,7 +981,7 @@ next
 tombstones
 ----
 a#3,15:d
-d#2,0:a
+d#2,0:
 .
 a-d#3
 .
@@ -993,8 +993,8 @@ next
 next
 ----
 a#3,15:d
-a#2,0:a
-d#2,0:a
+a#2,0:
+d#2,0:
 .
 
 iter snapshots=2
@@ -1005,7 +1005,7 @@ next
 ----
 a#3,15:d
 a#1,1:a
-d#2,0:a
+d#2,0:
 .
 
 iter snapshots=1
@@ -1014,7 +1014,7 @@ next
 next
 ----
 a#3,15:d
-d#2,0:a
+d#2,0:
 .
 
 define
@@ -1060,7 +1060,7 @@ a-b#1
 
 define
 a.MERGE.5:5
-a.DEL.3:3
+a.DEL.3:
 a.MERGE.1:1
 ----
 
diff --git a/testdata/compaction_iter_set_with_del b/testdata/compaction_iter_set_with_del
new file mode 100644
index 00000000..da782525
--- /dev/null
+++ b/testdata/compaction_iter_set_with_del
@@ -0,0 +1,1421 @@
+define
+a.SET.1:b
+----
+
+iter
+first
+next
+----
+a#1,1:b
+.
+
+define
+a.SET.2:c
+a.SET.1:b
+----
+
+iter
+first
+next
+----
+a#2,1:c
+.
+
+iter snapshots=0
+first
+next
+----
+a#2,1:c
+.
+
+iter snapshots=1
+first
+next
+----
+a#2,1:c
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#2,1:c
+a#1,1:b
+.
+
+define
+a.DEL.2:
+a.SET.1:b
+----
+
+iter
+first
+next
+----
+a#2,0:
+.
+
+iter elide-tombstones=true
+first
+----
+.
+
+iter elide-tombstones=true snapshots=2
+first
+next
+next
+----
+a#2,0:
+a#1,1:b
+.
+
+iter elide-tombstones=true snapshots=1
+first
+next
+----
+a#2,0:
+.
+
+define
+a.DEL.2:
+a.SET.1:b
+b.SET.3:c
+----
+
+iter
+first
+next
+next
+----
+a#2,0:
+b#3,1:c
+.
+
+iter snapshots=1
+first
+next
+next
+----
+a#2,0:
+b#3,1:c
+.
+
+iter snapshots=2
+first
+next
+next
+next
+----
+a#2,0:
+a#1,1:b
+b#3,1:c
+.
+
+define
+a.SET.1:a
+b.SET.2:b
+c.SET.3:c
+----
+
+iter
+first
+next
+next
+next
+----
+a#1,1:a
+b#2,1:b
+c#3,1:c
+.
+
+define
+a.MERGE.3:d
+a.MERGE.2:c
+a.SET.1:b
+b.MERGE.2:b
+b.MERGE.1:a
+----
+
+iter
+first
+next
+next
+----
+a#3,1:bcd[base]
+b#2,2:ab
+.
+
+iter snapshots=3
+first
+next
+next
+next
+----
+a#3,2:d
+a#2,1:bc[base]
+b#2,2:ab
+.
+
+define
+a.SET.9:b
+a.DEL.8:
+a.SET.7:d
+a.DEL.6:
+a.SET.5:f
+----
+
+iter
+first
+next
+----
+a#9,18:b
+.
+
+iter snapshots=6
+first
+next
+next
+----
+a#9,18:b
+a#5,1:f
+.
+
+iter snapshots=7
+first
+next
+next
+----
+a#9,18:b
+a#6,0:
+.
+
+iter snapshots=8
+first
+next
+next
+----
+a#9,18:b
+a#7,18:d
+.
+
+iter snapshots=9
+first
+next
+next
+----
+a#9,1:b
+a#8,0:
+.
+
+iter snapshots=10
+first
+next
+----
+a#9,18:b
+.
+
+iter snapshots=(5,6,7,8,9)
+first
+next
+next
+next
+next
+next
+----
+a#9,1:b
+a#8,0:
+a#7,1:d
+a#6,0:
+a#5,1:f
+.
+
+define
+a.INVALID.2:b
+a.SET.1:c
+----
+
+iter
+first
+----
+err=invalid internal key kind: 255
+
+define
+a.SET.2:b
+a.INVALID.1:c
+----
+
+iter
+first
+next
+----
+a#2,18:b
+err=invalid internal key kind: 255
+
+define
+a.MERGE.2:b
+a.INVALID.1:c
+----
+
+iter
+first
+next
+----
+a#2,2:b
+err=invalid internal key kind: 255
+
+define
+a.INVALID.2:c
+a.RANGEDEL.1:d
+----
+
+iter
+first
+tombstones
+----
+err=invalid internal key kind: 255
+.
+
+define
+a.MERGE.2:b
+a.MERGE.1:c
+a.MERGE.0:d
+----
+
+iter snapshots=(1,2)
+first
+next
+next
+next
+----
+a#2,2:b
+a#1,2:c
+a#0,2:d
+.
+
+define
+a.SET.2:b
+a.RANGEDEL.1:c
+b.RANGEDEL.4:d
+b.SET.2:e
+c.SET.3:f
+----
+
+iter
+first
+next
+next
+next
+tombstones
+----
+a#2,18:b
+a#1,15:c
+b#4,15:d
+.
+a-b#1
+b-c#4
+c-d#4
+.
+
+iter snapshots=2
+first
+next
+next
+next
+tombstones
+----
+a#2,1:b
+a#1,15:c
+b#4,15:d
+.
+a-b#1
+b-c#4
+b-c#1
+c-d#4
+.
+
+iter snapshots=3
+first
+next
+next
+next
+next
+tombstones
+----
+a#2,18:b
+a#1,15:c
+b#4,15:d
+b#2,1:e
+.
+a-b#1
+b-c#4
+b-c#1
+c-d#4
+.
+
+iter snapshots=4
+first
+next
+next
+next
+next
+next
+tombstones
+----
+a#2,18:b
+a#1,15:c
+b#4,15:d
+b#2,1:e
+c#3,1:f
+.
+a-b#1
+b-c#4
+b-c#1
+c-d#4
+.
+
+define
+a.RANGEDEL.3:e
+b.SET.4:b
+c.SET.3:c
+d.SET.2:d
+e.SET.1:e
+----
+
+iter
+first
+next
+next
+next
+next
+tombstones
+----
+a#3,15:e
+b#4,1:b
+c#3,1:c
+e#1,1:e
+.
+a-e#3
+.
+
+define
+a.RANGEDEL.3:e
+b.MERGE.4:b
+c.MERGE.3:c
+d.MERGE.2:d
+e.MERGE.1:e
+----
+
+iter
+first
+next
+next
+next
+next
+tombstones
+----
+a#3,15:e
+b#4,2:b
+c#3,2:c
+e#1,2:e
+.
+a-e#3
+.
+
+define
+a.RANGEDEL.3:c
+b.MERGE.5:e
+b.MERGE.4:d
+b.MERGE.2:c
+b.MERGE.1:b
+d.MERGE.5:c
+d.MERGE.4:b
+d.RANGEDEL.3:f
+d.MERGE.2:e
+d.MERGE.1:d
+----
+
+iter
+first
+next
+next
+next
+next
+tombstones
+----
+a#3,15:c
+b#5,1:de[base]
+d#5,2:bc
+d#3,15:f
+.
+a-c#3
+d-f#3
+.
+
+define
+a.RANGEDEL.3:d
+b.RANGEDEL.2:e
+c.RANGEDEL.1:f
+----
+
+iter
+first
+next
+next
+next
+tombstones
+----
+a#3,15:d
+b#2,15:e
+c#1,15:f
+.
+a-b#3
+b-c#3
+c-d#3
+d-e#2
+e-f#1
+.
+
+iter snapshots=2
+first
+next
+next
+next
+tombstones
+----
+a#3,15:d
+b#2,15:e
+c#1,15:f
+.
+a-b#3
+b-c#3
+c-d#3
+c-d#1
+d-e#2
+d-e#1
+e-f#1
+.
+
+iter snapshots=3
+first
+next
+next
+next
+tombstones
+----
+a#3,15:d
+b#2,15:e
+c#1,15:f
+.
+a-b#3
+b-c#3
+b-c#2
+c-d#3
+c-d#2
+d-e#2
+e-f#1
+.
+
+iter snapshots=(2,3)
+first
+next
+next
+next
+tombstones
+----
+a#3,15:d
+b#2,15:e
+c#1,15:f
+.
+a-b#3
+b-c#3
+b-c#2
+c-d#3
+c-d#2
+c-d#1
+d-e#2
+d-e#1
+e-f#1
+.
+
+define
+a.RANGEDEL.10:k
+f.SET.9:f
+f.SET.8:f
+----
+
+iter snapshots=(9,10)
+first
+next
+tombstones f
+next
+tombstones
+----
+a#10,15:k
+f#9,1:f
+a-k#10
+.
+f#8,1:f
+f-k#10
+.
+
+define
+f.RANGEDEL.10:k
+f.SET.9:f
+f.SET.8:f
+----
+
+iter snapshots=(9,10)
+first
+next
+tombstones f
+next
+tombstones
+----
+f#10,15:k
+f#9,1:f
+f-k#10
+.
+f#8,1:f
+f-k#10
+.
+
+define
+a.SET.1:a
+b.RANGEDEL.2:d
+c.RANGEDEL.3:e
+d.SET.4:d
+----
+
+iter
+first
+next
+next
+next
+tombstones c
+tombstones
+----
+a#1,1:a
+b#2,15:d
+c#3,15:e
+d#4,1:d
+b-c#2
+c-d#3
+.
+c-d#3
+d-e#3
+.
+
+iter snapshots=3
+first
+next
+next
+next
+tombstones c
+tombstones
+----
+a#1,1:a
+b#2,15:d
+c#3,15:e
+d#4,1:d
+b-c#2
+c-d#3
+c-d#2
+.
+c-d#3
+c-d#2
+d-e#3
+.
+
+define
+a.SET.1:a
+b.RANGEDEL.2:d
+c.SET.4:d
+----
+
+iter
+first
+next
+next
+tombstones c
+tombstones
+----
+a#1,1:a
+b#2,15:d
+c#4,1:d
+b-d#2
+.
+c-d#2
+.
+
+define
+a.RANGEDEL.2:d
+a.SET.2:a
+b.SET.2:b
+c.SET.2:c
+----
+
+iter
+first
+next
+next
+next
+next
+----
+a#2,15:d
+a#2,1:a
+b#2,1:b
+c#2,1:c
+.
+
+define
+a.SINGLEDEL.1:
+----
+
+iter
+first
+next
+----
+a#1,7:
+.
+
+iter elide-tombstones=true
+first
+----
+.
+
+define
+a.SINGLEDEL.2:
+a.SINGLEDEL.1:
+----
+
+iter
+first
+next
+----
+a#2,7:
+.
+
+define
+a.SINGLEDEL.3:
+a.SINGLEDEL.2:
+a.SET.1:a
+----
+
+iter
+first
+----
+.
+
+define
+a.SET.3:a
+b.SINGLEDEL.2:
+b.DEL.1:
+----
+
+iter
+first
+next
+next
+----
+a#3,1:a
+b#2,0:
+.
+
+define
+a.SINGLEDEL.2:
+a.DEL.1:
+----
+
+iter
+first
+next
+----
+a#2,0:
+.
+
+iter elide-tombstones=true
+first
+----
+.
+
+define
+a.SINGLEDEL.2:
+a.MERGE.1:
+----
+
+iter
+first
+next
+----
+a#2,0:
+.
+
+iter elide-tombstones=true
+first
+----
+.
+
+define
+a.SINGLEDEL.2:
+a.SET.1:b
+----
+
+iter
+first
+----
+.
+
+# SET that meets a SINGLEDEL is transformed into a SETWITHDEL.
+
+define
+a.SET.2:b
+a.SINGLEDEL.1:
+----
+
+iter
+first
+next
+----
+a#2,18:b
+.
+
+define
+a.MERGE.6:b
+a.SINGLEDEL.5:
+a.SET.4:a
+----
+
+iter
+first
+next
+----
+a#6,1:b[base]
+.
+
+# Non-deterministic use of SINGLEDEL where there are two older SETs that have
+# not been deleted or single deleted. It is permitted to shadow both.
+define
+a.MERGE.6:b
+a.SINGLEDEL.5:
+a.SET.4:a
+a.SET.3:a
+----
+
+iter
+first
+next
+----
+a#6,1:b[base]
+.
+
+define
+a.SINGLEDEL.2:
+a.SET.1:b
+b.SET.3:c
+----
+
+iter
+first
+next
+----
+b#3,1:c
+.
+
+define
+a.SINGLEDEL.3:
+a.SET.2:b
+a.SET.1:a
+----
+
+iter
+first
+next
+----
+a#1,1:a
+.
+
+define
+a.SINGLEDEL.3:
+a.MERGE.2:b
+a.MERGE.1:a
+----
+
+iter
+first
+next
+----
+a#3,0:
+.
+
+define
+a.SINGLEDEL.4:
+a.SET.3:val
+a.SINGLEDEL.2:
+a.SET.1:val
+----
+
+iter
+first
+----
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#2,7:
+a#1,1:val
+.
+
+define
+a.SINGLEDEL.4:
+a.SET.3:val
+a.DEL.2:
+a.SET.1:val
+----
+
+iter
+first
+next
+----
+a#2,0:
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#2,0:
+a#1,1:val
+.
+
+iter snapshots=3
+first
+next
+----
+a#2,0:
+.
+
+iter snapshots=(2,3)
+first
+next
+next
+----
+a#2,0:
+a#1,1:val
+.
+
+define
+a.SINGLEDEL.4:
+a.SET.3:c
+a.MERGE.2:b
+a.SET.1:a
+----
+
+iter
+first
+next
+----
+a#2,1:ab[base]
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#2,2:b
+a#1,1:a
+.
+
+iter snapshots=3
+first
+next
+----
+a#2,1:ab[base]
+.
+
+iter snapshots=(2,3,4)
+first
+next
+next
+next
+next
+----
+a#4,7:
+a#3,1:c
+a#2,2:b
+a#1,1:a
+.
+
+define
+a.SINGLEDEL.3:
+a.RANGEDEL.2:c
+a.SET.1:val
+----
+
+iter
+first
+next
+next
+tombstones
+----
+a#3,7:
+a#2,15:c
+.
+a-c#2
+.
+
+define
+a.RANGEDEL.3:d
+a.DEL.2:
+a.SET.1:a
+d.DEL.2:
+----
+
+iter
+first
+next
+next
+tombstones
+----
+a#3,15:d
+d#2,0:
+.
+a-d#3
+.
+
+iter snapshots=3
+first
+next
+next
+next
+----
+a#3,15:d
+a#2,0:
+d#2,0:
+.
+
+iter snapshots=2
+first
+next
+next
+next
+----
+a#3,15:d
+a#1,1:a
+d#2,0:
+.
+
+iter snapshots=1
+first
+next
+next
+----
+a#3,15:d
+d#2,0:
+.
+
+define
+a.MERGE.2:a
+b.RANGEDEL.1:c
+----
+
+iter
+first
+tombstones a
+next
+next
+tombstones
+----
+a#2,2:a
+.
+b#1,15:c
+.
+b-c#1
+.
+
+define
+a.MERGE.2:v2
+a.RANGEDEL.1:b
+a.MERGE.1:v1
+----
+
+iter allow-zero-seqnum=true
+first
+next
+next
+next
+tombstones
+----
+a#2,2:v2
+a#1,15:b
+a#0,2:v1
+.
+a-b#1
+.
+
+# Verify that we transform merge+del -> set.
+
+define
+a.MERGE.5:5
+a.DEL.3:
+a.MERGE.1:1
+----
+
+iter
+first
+next
+----
+a#5,1:5[base]
+.
+
+iter allow-zero-seqnum=true
+first
+next
+----
+a#0,1:5[base]
+.
+
+iter elide-tombstones=true
+first
+next
+----
+a#5,1:5[base]
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#5,1:5[base]
+a#1,2:1
+.
+
+iter snapshots=2 elide-tombstones=true
+first
+next
+next
+----
+a#5,1:5[base]
+a#1,2:1
+.
+
+# Verify that we transform merge+rangedel -> set. This isn't strictly
+# necessary, but provides consistency with the behavior for merge+del.
+
+define
+a.RANGEDEL.3:c
+b.MERGE.5:5
+b.SET.2:2
+b.MERGE.1:1
+----
+
+iter
+first
+next
+next
+----
+a#3,15:c
+b#5,1:5[base]
+.
+
+iter allow-zero-seqnum=true
+first
+next
+next
+----
+a#3,15:c
+b#0,1:5[base]
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#3,15:c
+b#5,1:5[base]
+b#1,2:1
+
+define
+a.RANGEDEL.3:c
+b.MERGE.5:5
+b.MERGE.2:2
+b.MERGE.1:1
+----
+
+iter
+first
+next
+next
+----
+a#3,15:c
+b#5,1:5[base]
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#3,15:c
+b#5,1:5[base]
+b#1,2:1
+
+# SET that meets a DEL is transformed into a SETWITHDEL.
+
+define
+a.SET.2:b
+a.DEL.1:
+----
+
+iter
+first
+next
+----
+a#2,18:b
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#2,1:b
+a#1,0:
+.
+
+define
+a.SET.3:c
+a.DEL.2:
+a.SET.1:b
+----
+
+iter
+first
+next
+----
+a#3,18:c
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#3,18:c
+a#1,1:b
+.
+
+define
+a.SET.3:c
+a.SET.2:b
+a.DEL.1:
+----
+
+iter
+first
+next
+----
+a#3,18:c
+.
+
+iter snapshots=3
+first
+next
+next
+----
+a#3,1:c
+a#2,18:b
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#3,1:c
+a#1,0:
+.
+
+define
+a.DEL.3:
+a.SET.2:b
+a.DEL.1:
+----
+
+iter
+first
+next
+----
+a#3,0:
+.
+
+iter snapshots=3
+first
+next
+next
+----
+a#3,0:
+a#2,18:b
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#3,0:
+a#1,0:
+.
+
+# SETWITHDEL-eligible entries at or under a RANGEDEL at the same user key should
+# be skipped.
+define
+a.SET.3:c
+a.RANGEDEL.2:z
+a.SET.2:b
+a.DEL.1:
+----
+
+iter allow-zero-seqnum=true
+first
+next
+next
+----
+a#0,18:c
+a#2,15:z
+.
+
+iter allow-zero-seqnum=true snapshots=3
+first
+next
+next
+next
+----
+a#3,1:c
+a#2,15:z
+a#0,18:b
+.
+
+iter allow-zero-seqnum=true snapshots=2
+first
+next
+next
+next
+----
+a#3,18:c
+a#2,15:z
+a#1,0:
+.
+
+define
+a.SET.4:c
+a.RANGEDEL.3:z
+a.SET.2:b
+a.DEL.1:
+----
+
+iter
+first
+next
+next
+----
+a#4,18:c
+a#3,15:z
+.
+
+# Invalid keys are emitted under SETWITHDEL.
+
+define
+a.SET.2:b
+a.INVALID.1:
+----
+
+iter
+first
+next
+----
+a#2,18:b
+err=invalid internal key kind: 255
+
+define
+a.SET.3:c
+a.INVALID.2:
+a.SET.1:b
+----
+
+iter
+first
+next
+----
+a#3,18:c
+err=invalid internal key kind: 255
+
+# SINGLEDEL that meets a SETWITHDEL is transformed into a DEL.
+
+define
+a.SINGLEDEL.3:
+a.SETWITHDEL.2:d
+b.SET.1:c
+----
+
+iter
+first
+next
+next
+----
+a#3,0:
+b#1,1:c
+.
+
+iter snapshots=2
+first
+next
+next
+----
+a#3,0:
+b#1,1:c
+.
+
+iter snapshots=3
+first
+next
+next
+next
+----
+a#3,7:
+a#2,18:d
+b#1,1:c
+.
+
+define
+a.SETWITHDEL.3:3
+a.SET.2:d
+b.SET.1:c
+----
+
+iter
+first
+next
+next
+----
+a#3,18:3
+b#1,1:c
+.
+
+iter snapshots=3
+first
+next
+next
+next
+----
+a#3,18:3
+a#2,1:d
+b#1,1:c
+.
diff --git a/testdata/event_listener b/testdata/event_listener
index d0fa1da3..4e4b0539 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -30,6 +30,10 @@ create: db/marker.format-version.000002.003
 close: db/marker.format-version.000002.003
 sync: db
 upgraded to format version: 003
+create: db/marker.format-version.000003.004
+close: db/marker.format-version.000003.004
+sync: db
+upgraded to format version: 004
 create: db/MANIFEST-000003
 close: db/MANIFEST-000001
 sync: db/MANIFEST-000003
@@ -195,9 +199,9 @@ close:
 open-dir: checkpoint
 link: db/OPTIONS-000004 -> checkpoint/OPTIONS-000004
 open-dir: checkpoint
-create: checkpoint/marker.format-version.000001.003
-sync: checkpoint/marker.format-version.000001.003
-close: checkpoint/marker.format-version.000001.003
+create: checkpoint/marker.format-version.000001.004
+sync: checkpoint/marker.format-version.000001.004
+close: checkpoint/marker.format-version.000001.004
 sync: checkpoint
 close: checkpoint
 create: checkpoint/MANIFEST-000017
diff --git a/testdata/level_checker b/testdata/level_checker
index 550ce730..f0be6a1b 100644
--- a/testdata/level_checker
+++ b/testdata/level_checker
@@ -348,7 +348,7 @@ merge processing error on key a#9,DEL in L1: fileNum=000032: finish failed
 define
 L
 a.MERGE.10 a.SINGLEDEL.9
-a.MERGE.10:fail-finish a.SINGLEDEL.9:fail-finish
+a.MERGE.10:fail-finish a.SINGLEDEL.9:
 ----
 Level 1
   file 0: [a#10,2-a#9,7]
diff --git a/testdata/manual_compaction b/testdata/manual_compaction
index 594c742d..ebe49e79 100644
--- a/testdata/manual_compaction
+++ b/testdata/manual_compaction
@@ -48,8 +48,8 @@ L0
   a.SET.2:v
 ----
 0.0:
-  000005:[a-a]
-  000004:[b-b]
+  000005:[a#2,SET-a#2,SET]
+  000004:[b#1,SET-b#1,SET]
 
 compact a-b
 ----
@@ -74,12 +74,12 @@ L3
   d.SET.0:v
 ----
 1:
-  000004:[a-a]
+  000004:[a#3,SET-a#3,SET]
 2:
-  000005:[a-e]
+  000005:[a#2,RANGEDEL-e#72057594037927935,RANGEDEL]
 3:
-  000006:[a-b]
-  000007:[c-d]
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
 
 wait-pending-table-stats
 000005
@@ -128,14 +128,14 @@ L3
   g.SET.0:v
 ----
 1:
-  000004:[a-a]
+  000004:[a#3,SET-a#3,SET]
 2:
-  000005:[a-g]
+  000005:[a#2,RANGEDEL-g#72057594037927935,RANGEDEL]
 3:
-  000006:[a-b]
-  000007:[c-d]
-  000008:[e-f]
-  000009:[f-g]
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+  000008:[e#0,SET-f#1,SET]
+  000009:[f#0,SET-g#0,SET]
 
 compact a-e L1
 ----
@@ -170,13 +170,13 @@ L3
   f.SET.1:v
 ----
 1:
-  000004:[a-h]
+  000004:[a#3,SET-h#3,SET]
 2:
-  000005:[a-g]
+  000005:[a#2,RANGEDEL-g#72057594037927935,RANGEDEL]
 3:
-  000006:[a-b]
-  000007:[c-d]
-  000008:[e-f]
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+  000008:[e#0,SET-f#1,SET]
 
 compact a-e L1
 ----
@@ -205,12 +205,12 @@ L3
   d.SET.0:v
 ----
 1:
-  000004:[a-e]
+  000004:[a#3,RANGEDEL-e#72057594037927935,RANGEDEL]
 2:
-  000005:[a-a]
+  000005:[a#2,SET-a#2,SET]
 3:
-  000006:[a-b]
-  000007:[c-d]
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
 
 compact a-e L1
 ----
@@ -241,13 +241,13 @@ L3
   m.SET.0:v
 ----
 1:
-  000004:[a-d]
-  000005:[grandparent-z]
+  000004:[a#4,RANGEDEL-d#72057594037927935,RANGEDEL]
+  000005:[grandparent#2,RANGEDEL-z#72057594037927935,RANGEDEL]
 2:
-  000006:[grandparent-grandparent]
+  000006:[grandparent#1,SET-grandparent#1,SET]
 3:
-  000007:[grandparent-grandparent]
-  000008:[m-m]
+  000007:[grandparent#0,SET-grandparent#0,SET]
+  000008:[m#0,SET-m#0,SET]
 
 compact a-h L1
 ----
@@ -275,13 +275,13 @@ L3
   b.SET.0:v
 ----
 1:
-  000004:[a-c]
+  000004:[a#2,SET-c#2,SET]
 2:
-  000005:[a-c]
+  000005:[a#3,RANGEDEL-c#72057594037927935,RANGEDEL]
 3:
-  000006:[b-b]
-  000007:[b-b]
-  000008:[b-b]
+  000006:[b#2,SET-b#2,SET]
+  000007:[b#1,SET-b#1,SET]
+  000008:[b#0,SET-b#0,SET]
 
 compact a-c L1
 ----
@@ -377,13 +377,13 @@ L3
   b.SET.1:1
 ----
 0.0:
-  000004:[c-c]
+  000004:[c#4,SET-c#4,SET]
 1:
-  000005:[a-a]
+  000005:[a#3,SET-a#3,SET]
 2:
-  000006:[a-e]
+  000006:[a#2,RANGEDEL-e#72057594037927935,RANGEDEL]
 3:
-  000007:[b-b]
+  000007:[b#1,SET-b#1,SET]
 
 compact a-e L1
 ----
@@ -438,10 +438,10 @@ L3
   b.SET.1:1
 ----
 1:
-  000004:[a-a]
-  000005:[b-e]
+  000004:[a#4,SET-a#4,SET]
+  000005:[b#3,RANGEDEL-e#72057594037927935,RANGEDEL]
 3:
-  000006:[b-b]
+  000006:[b#1,SET-b#1,SET]
 
 compact a-e L1
 ----
@@ -475,10 +475,10 @@ L3
   b.SET.1:1
 ----
 1:
-  000004:[a-a]
-  000005:[b-e]
+  000004:[a#4,SET-a#4,SET]
+  000005:[b#3,RANGEDEL-e#72057594037927935,RANGEDEL]
 3:
-  000006:[b-b]
+  000006:[b#1,SET-b#1,SET]
 
 compact a-e L1
 ----
@@ -516,10 +516,10 @@ L3
   b.SET.1:1
 ----
 1:
-  000004:[a-a]
-  000005:[b-e]
+  000004:[a#4,SET-a#4,SET]
+  000005:[b#4,SET-e#72057594037927935,RANGEDEL]
 3:
-  000006:[b-b]
+  000006:[b#1,SET-b#1,SET]
 
 compact a-e L1
 ----
@@ -562,10 +562,10 @@ L3
   d.SET.0:0
 ----
 1:
-  000004:[a-e]
+  000004:[a#3,SET-e#72057594037927935,RANGEDEL]
 3:
-  000005:[a-b]
-  000006:[c-d]
+  000005:[a#2,RANGEDEL-b#72057594037927935,RANGEDEL]
+  000006:[c#0,SET-d#0,SET]
 
 iter
 last
@@ -605,9 +605,9 @@ L3
   a.RANGEDEL.2:b
 ----
 1:
-  000004:[a-e]
+  000004:[a#3,SET-e#72057594037927935,RANGEDEL]
 3:
-  000005:[a-b]
+  000005:[a#2,RANGEDEL-b#72057594037927935,RANGEDEL]
 
 compact a-f L1
 ----
@@ -630,9 +630,9 @@ L3
   b.SET.0:0
 ----
 1:
-  000004:[a-d]
+  000004:[a#0,SET-d#72057594037927935,RANGEDEL]
 3:
-  000005:[b-b]
+  000005:[b#0,SET-b#0,SET]
 
 compact a-e L1
 ----
@@ -648,8 +648,8 @@ L0
   a.SET.2:v
 ----
 0.0:
-  000005:[a-a]
-  000004:[b-b]
+  000005:[a#2,SET-a#2,SET]
+  000004:[b#1,SET-b#1,SET]
 
 add-ongoing-compaction startLevel=0 outputLevel=1
 ----
@@ -728,15 +728,15 @@ L3
   k.SET.1:foo
 ----
 1:
-  000004:[a-f]
-  000005:[g-j]
-  000006:[k-q]
+  000004:[a#4,RANGEDEL-f#72057594037927935,RANGEDEL]
+  000005:[g#6,RANGEDEL-j#72057594037927935,RANGEDEL]
+  000006:[k#5,RANGEDEL-q#72057594037927935,RANGEDEL]
 2:
-  000007:[a-a]
+  000007:[a#2,SET-a#2,SET]
 3:
-  000008:[a-c]
-  000009:[ff-ff]
-  000010:[k-k]
+  000008:[a#1,SET-c#1,SET]
+  000009:[ff#1,SET-ff#1,SET]
+  000010:[k#1,SET-k#1,SET]
 
 compact a-q L1
 ----
@@ -762,11 +762,11 @@ L3
   q.SET.6:6
 ----
 1:
-  000004:[a-r]
+  000004:[a#10,RANGEDEL-r#72057594037927935,RANGEDEL]
 2:
-  000005:[g-h]
+  000005:[g#7,RANGEDEL-h#72057594037927935,RANGEDEL]
 3:
-  000006:[q-q]
+  000006:[q#6,SET-q#6,SET]
 
 compact a-r L1
 ----
@@ -794,15 +794,15 @@ L4
   f.SET.0:0
 ----
 1:
-  000004:[a-j]
+  000004:[a#10,RANGEDEL-j#10,SET]
 2:
-  000005:[f-g]
+  000005:[f#7,RANGEDEL-g#72057594037927935,RANGEDEL]
 3:
-  000006:[c-c]
-  000007:[c-c]
-  000008:[c-c]
+  000006:[c#6,SET-c#6,SET]
+  000007:[c#5,SET-c#5,SET]
+  000008:[c#4,SET-c#4,SET]
 4:
-  000009:[a-f]
+  000009:[a#0,SET-f#0,SET]
 
 compact a-r L1
 ----
@@ -830,9 +830,9 @@ L2
   d.SET.0:foo
 ----
 1:
-  000004:[a-z]
+  000004:[a#15,RANGEDEL-z#72057594037927935,RANGEDEL]
 2:
-  000005:[c-d]
+  000005:[c#0,SET-d#0,SET]
 
 compact a-z L1
 ----
@@ -856,9 +856,9 @@ L2
   d.SET.0:foo
 ----
 1:
-  000004:[a-z]
+  000004:[a#10,RANGEDEL-z#72057594037927935,RANGEDEL]
 2:
-  000005:[c-d]
+  000005:[c#0,SET-d#0,SET]
 
 compact a-z L1
 ----
diff --git a/testdata/manual_compaction_set_with_del b/testdata/manual_compaction_set_with_del
new file mode 100644
index 00000000..53bd1625
--- /dev/null
+++ b/testdata/manual_compaction_set_with_del
@@ -0,0 +1,867 @@
+batch
+set a 1
+set b 2
+----
+
+compact a-b
+----
+6:
+  000005:[a#1,SET-b#2,SET]
+
+batch
+set c 3
+set d 4
+----
+
+compact c-d
+----
+6:
+  000005:[a#1,SET-b#2,SET]
+  000007:[c#3,SET-d#4,SET]
+
+batch
+set b 5
+set c 6
+----
+
+compact a-d
+----
+6:
+  000010:[a#0,SET-d#0,SET]
+
+# This also tests flushing a memtable that only contains range
+# deletions.
+
+batch
+del-range a e
+----
+
+compact a-d
+----
+
+# Test that a multi-output-file compaction generates non-overlapping files.
+
+define target-file-sizes=(100, 1)
+L0
+  b.SET.1:v
+L0
+  a.SET.2:v
+----
+0.0:
+  000005:[a#2,SET-a#2,SET]
+  000004:[b#1,SET-b#1,SET]
+
+compact a-b
+----
+1:
+  000006:[a#0,SET-a#0,SET]
+  000007:[b#0,SET-b#0,SET]
+
+# A range tombstone extends past the grandparent file boundary used to limit the
+# size of future compactions. Verify the range tombstone is split at that file
+# boundary.
+
+define target-file-sizes=(1, 1, 1, 1)
+L1
+  a.SET.3:v
+L2
+  a.RANGEDEL.2:e
+L3
+  a.SET.0:v
+  b.SET.0:v
+L3
+  c.SET.0:v
+  d.SET.0:v
+----
+1:
+  000004:[a#3,SET-a#3,SET]
+2:
+  000005:[a#2,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+
+wait-pending-table-stats
+000005
+----
+num-entries: 1
+num-deletions: 1
+point-deletions-bytes-estimate: 0
+range-deletions-bytes-estimate: 1552
+
+compact a-e L1
+----
+2:
+  000008:[a#3,SETWITHDEL-b#72057594037927935,RANGEDEL]
+  000009:[b#2,RANGEDEL-d#72057594037927935,RANGEDEL]
+  000010:[d#2,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+
+wait-pending-table-stats
+000008
+----
+num-entries: 2
+num-deletions: 1
+point-deletions-bytes-estimate: 0
+range-deletions-bytes-estimate: 776
+
+# Same as above, except range tombstone covers multiple grandparent file boundaries.
+
+define target-file-sizes=(1, 1, 1, 1)
+L1
+  a.SET.3:v
+L2
+  a.RANGEDEL.2:g
+L3
+  a.SET.0:v
+  b.SET.0:v
+L3
+  c.SET.0:v
+  d.SET.0:v
+L3
+  e.SET.0:v
+  f.SET.1:v
+L3
+  f.SET.0:v
+  g.SET.0:v
+----
+1:
+  000004:[a#3,SET-a#3,SET]
+2:
+  000005:[a#2,RANGEDEL-g#72057594037927935,RANGEDEL]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+  000008:[e#0,SET-f#1,SET]
+  000009:[f#0,SET-g#0,SET]
+
+compact a-e L1
+----
+2:
+  000010:[a#3,SETWITHDEL-b#72057594037927935,RANGEDEL]
+  000011:[b#2,RANGEDEL-d#72057594037927935,RANGEDEL]
+  000012:[d#2,RANGEDEL-f#72057594037927935,RANGEDEL]
+  000013:[f#2,RANGEDEL-g#72057594037927935,RANGEDEL]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+  000008:[e#0,SET-f#1,SET]
+  000009:[f#0,SET-g#0,SET]
+
+# A range tombstone covers multiple grandparent file boundaries between point keys,
+# rather than after all point keys.
+
+define target-file-sizes=(1, 1, 1, 1)
+L1
+  a.SET.3:v
+  h.SET.3:v
+L2
+  a.RANGEDEL.2:g
+L3
+  a.SET.0:v
+  b.SET.0:v
+L3
+  c.SET.0:v
+  d.SET.0:v
+L3
+  e.SET.0:v
+  f.SET.1:v
+----
+1:
+  000004:[a#3,SET-h#3,SET]
+2:
+  000005:[a#2,RANGEDEL-g#72057594037927935,RANGEDEL]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+  000008:[e#0,SET-f#1,SET]
+
+compact a-e L1
+----
+2:
+  000009:[a#3,SETWITHDEL-g#72057594037927935,RANGEDEL]
+  000010:[h#3,SET-h#3,SET]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+  000008:[e#0,SET-f#1,SET]
+
+# A range tombstone is the first and only item output by a compaction, and it
+# extends past the grandparent file boundary used to limit the size of future
+# compactions. Verify the range tombstone is split at that file boundary.
+
+define target-file-sizes=(1, 1, 1, 1)
+L1
+  a.RANGEDEL.3:e
+L2
+  a.SET.2:v
+L3
+  a.SET.0:v
+  b.SET.0:v
+L3
+  c.SET.0:v
+  d.SET.0:v
+----
+1:
+  000004:[a#3,RANGEDEL-e#72057594037927935,RANGEDEL]
+2:
+  000005:[a#2,SET-a#2,SET]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+
+compact a-e L1
+----
+2:
+  000008:[a#3,RANGEDEL-b#72057594037927935,RANGEDEL]
+  000009:[b#3,RANGEDEL-d#72057594037927935,RANGEDEL]
+  000010:[d#3,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[a#0,SET-b#0,SET]
+  000007:[c#0,SET-d#0,SET]
+
+# An elided range tombstone is the first item encountered by a compaction,
+# and the grandparent limit set by it extends to the next item, also a range
+# tombstone. The first item should be elided, and the second item should
+# reset the grandparent limit.
+
+define target-file-sizes=(100, 100, 100, 100)
+L1
+  a.RANGEDEL.4:d
+L1
+  grandparent.RANGEDEL.2:z
+  h.SET.3:v
+L2
+  grandparent.SET.1:v
+L3
+  grandparent.SET.0:v
+L3
+  m.SET.0:v
+----
+1:
+  000004:[a#4,RANGEDEL-d#72057594037927935,RANGEDEL]
+  000005:[grandparent#2,RANGEDEL-z#72057594037927935,RANGEDEL]
+2:
+  000006:[grandparent#1,SET-grandparent#1,SET]
+3:
+  000007:[grandparent#0,SET-grandparent#0,SET]
+  000008:[m#0,SET-m#0,SET]
+
+compact a-h L1
+----
+2:
+  000009:[grandparent#2,RANGEDEL-m#72057594037927935,RANGEDEL]
+  000010:[m#2,RANGEDEL-z#72057594037927935,RANGEDEL]
+3:
+  000007:[grandparent#0,SET-grandparent#0,SET]
+  000008:[m#0,SET-m#0,SET]
+
+# Setup such that grandparent overlap limit is exceeded multiple times at the same user key ("b").
+# Ensures the compaction output files are non-overlapping.
+
+define target-file-sizes=(1, 1, 1, 1)
+L1
+  a.SET.2:v
+  c.SET.2:v
+L2
+  a.RANGEDEL.3:c
+L3
+  b.SET.2:v
+L3
+  b.SET.1:v
+L3
+  b.SET.0:v
+----
+1:
+  000004:[a#2,SET-c#2,SET]
+2:
+  000005:[a#3,RANGEDEL-c#72057594037927935,RANGEDEL]
+3:
+  000006:[b#2,SET-b#2,SET]
+  000007:[b#1,SET-b#1,SET]
+  000008:[b#0,SET-b#0,SET]
+
+compact a-c L1
+----
+2:
+  000009:[a#3,RANGEDEL-c#72057594037927935,RANGEDEL]
+  000010:[c#2,SET-c#2,SET]
+3:
+  000006:[b#2,SET-b#2,SET]
+  000007:[b#1,SET-b#1,SET]
+  000008:[b#0,SET-b#0,SET]
+
+# Regression test for a bug where compaction would stop process range
+# tombstones for an input level upon finding an sstable in the input
+# level with no range tombstones. In the scenario below, sstable 6
+# does not contain any range tombstones while sstable 7 does. Both are
+# compacted together with sstable 5.
+
+reset
+----
+
+batch
+set a 1
+set b 1
+set c 1
+set d 1
+set z 1
+----
+
+compact a-z
+----
+6:
+  000005:[a#1,SET-z#5,SET]
+
+build ext1
+set a 2
+----
+
+build ext2
+set b 2
+del-range c z
+----
+
+ingest ext1 ext2
+----
+0.0:
+  000006:[a#6,SET-a#6,SET]
+  000007:[b#7,SET-z#72057594037927935,RANGEDEL]
+6:
+  000005:[a#1,SET-z#5,SET]
+
+iter
+first
+next
+next
+next
+----
+a:2
+b:2
+z:1
+.
+
+compact a-z
+----
+6:
+  000008:[a#0,SET-z#0,SET]
+
+iter
+first
+next
+next
+next
+----
+a:2
+b:2
+z:1
+.
+
+# Regresion test for a bug in sstable smallest boundary generation
+# where the smallest key for an sstable was set to a key "larger" than
+# the start key of the first range tombstone. This in turn fouled up
+# the processing logic of range tombstones used by mergingIter which
+# allowed stepping out of an sstable even though it contained a range
+# tombstone that covered keys in lower levels.
+
+define target-file-sizes=(1, 1, 1, 1)
+L0
+  c.SET.4:4
+L1
+  a.SET.3:3
+L2
+  a.RANGEDEL.2:e
+L3
+  b.SET.1:1
+----
+0.0:
+  000004:[c#4,SET-c#4,SET]
+1:
+  000005:[a#3,SET-a#3,SET]
+2:
+  000006:[a#2,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000007:[b#1,SET-b#1,SET]
+
+compact a-e L1
+----
+0.0:
+  000004:[c#4,SET-c#4,SET]
+2:
+  000008:[a#3,SETWITHDEL-b#72057594037927935,RANGEDEL]
+  000009:[b#2,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000007:[b#1,SET-b#1,SET]
+
+# We should only see a:3 and c:4 at this point.
+
+iter
+first
+next
+next
+----
+a:3
+c:4
+.
+
+# The bug allowed seeing b:1 during reverse iteration.
+
+iter
+last
+prev
+prev
+----
+c:4
+a:3
+.
+
+# This is a similar scenario to the one above, but we've arranged for
+# the compaction to produce an sstable (9) for which the smallest
+# range tombstone overlaps in key space with the largest key in the
+# previous sstable (8). This necessitates adjusting the seqnum of the
+# range tombstone. There was no actual iteration bug from earlier
+# behavior of setting the seqnum to 0, but now we set the seqnum to
+# prev.LargestKey.SeqNum-1.
+#
+# Note the target-file-size of 26 is specially tailored to get the
+# desired compaction output.
+
+define target-file-sizes=(26, 26, 26, 26) snapshots=(1, 2, 3)
+L1
+  a.SET.4:4
+L1
+  b.SET.2:2
+  b.RANGEDEL.3:e
+L3
+  b.SET.1:1
+----
+1:
+  000004:[a#4,SET-a#4,SET]
+  000005:[b#3,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[b#1,SET-b#1,SET]
+
+compact a-e L1
+----
+2:
+  000007:[a#4,SET-b#2,SET]
+  000008:[b#1,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[b#1,SET-b#1,SET]
+
+iter
+first
+next
+last
+prev
+----
+a:4
+.
+a:4
+.
+
+# Similar to the preceding scenario, except the range tombstone has
+# the same seqnum as the largest key in the preceding file.
+
+define target-file-sizes=(26, 26, 26, 26) snapshots=(1, 2, 3)
+L1
+  a.SET.4:4
+L1
+  b.SET.3:3
+  b.RANGEDEL.3:e
+L3
+  b.SET.1:1
+----
+1:
+  000004:[a#4,SET-a#4,SET]
+  000005:[b#3,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[b#1,SET-b#1,SET]
+
+compact a-e L1
+----
+2:
+  000007:[a#4,SET-b#3,SET]
+  000008:[b#2,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[b#1,SET-b#1,SET]
+
+iter
+first
+next
+next
+last
+prev
+prev
+----
+a:4
+b:3
+.
+b:3
+a:4
+.
+
+# Similar to the preceding scenario, except the range tombstone has
+# a smaller seqnum than the largest key in the preceding file.
+
+define target-file-sizes=(26, 26, 26, 26) snapshots=(1, 2, 3)
+L1
+  a.SET.4:4
+L1
+  b.SET.4:4
+  b.RANGEDEL.2:e
+L3
+  b.SET.1:1
+----
+1:
+  000004:[a#4,SET-a#4,SET]
+  000005:[b#4,SET-e#72057594037927935,RANGEDEL]
+3:
+  000006:[b#1,SET-b#1,SET]
+
+compact a-e L1
+----
+2:
+  000007:[a#4,SET-b#4,SET]
+  000008:[b#2,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000006:[b#1,SET-b#1,SET]
+
+iter
+first
+next
+next
+last
+prev
+prev
+----
+a:4
+b:4
+.
+b:4
+a:4
+.
+
+# Test a scenario where the last point key in an sstable has a seqnum
+# of 0, and the smallest range tombstone in the next sstable starts
+# with the same key. We have to tweak the kind of the boundary key in
+# sstable 8 to be a DEL rather than a RANGEDEL so that the two tables
+# are disjoint in the key space.
+
+define target-file-sizes=(1, 1, 26) snapshots=(2)
+L1
+  a.SET.3:3
+  b.RANGEDEL.3:e
+  b.SET.0:0
+L3
+  a.RANGEDEL.2:b
+L3
+  c.SET.0:0
+  d.SET.0:0
+----
+1:
+  000004:[a#3,SET-e#72057594037927935,RANGEDEL]
+3:
+  000005:[a#2,RANGEDEL-b#72057594037927935,RANGEDEL]
+  000006:[c#0,SET-d#0,SET]
+
+iter
+last
+prev
+----
+a:3
+.
+
+compact a-e L1
+----
+2:
+  000007:[a#3,SET-e#72057594037927935,RANGEDEL]
+3:
+  000005:[a#2,RANGEDEL-b#72057594037927935,RANGEDEL]
+  000006:[c#0,SET-d#0,SET]
+
+iter
+last
+prev
+----
+a:3
+.
+
+# Test a scenario where the last point key in an sstable before the
+# grandparent limit is reached has a seqnum of 0. We want to cut the
+# sstable after the next point key is added, rather than continuing to
+# add keys indefinitely (or till the size limit is reached).
+
+define target-file-sizes=(100, 1, 52) snapshots=(2)
+L1
+  a.SET.3:3
+  b.RANGEDEL.3:e
+  b.SET.0:0
+  c.SET.1:1
+  d.SET.1:1
+L3
+  a.RANGEDEL.2:b
+----
+1:
+  000004:[a#3,SET-e#72057594037927935,RANGEDEL]
+3:
+  000005:[a#2,RANGEDEL-b#72057594037927935,RANGEDEL]
+
+compact a-f L1
+----
+2:
+  000006:[a#3,SET-c#72057594037927935,RANGEDEL]
+  000007:[c#3,RANGEDEL-e#72057594037927935,RANGEDEL]
+3:
+  000005:[a#2,RANGEDEL-b#72057594037927935,RANGEDEL]
+
+# Test a scenario where we the last point key in an sstable has a
+# seqnum of 0, but there is another range tombstone later in the
+# compaction. This scenario was previously triggering an assertion due
+# to the rangedel.Fragmenter being finished prematurely.
+
+define target-file-sizes=(1, 1, 1)
+L1
+  a.SET.0:0
+  c.RANGEDEL.1:d
+L3
+  b.SET.0:0
+----
+1:
+  000004:[a#0,SET-d#72057594037927935,RANGEDEL]
+3:
+  000005:[b#0,SET-b#0,SET]
+
+compact a-e L1
+----
+2:
+  000006:[a#0,SET-a#0,SET]
+3:
+  000005:[b#0,SET-b#0,SET]
+
+define target-file-sizes=(1, 1, 1, 1)
+L0
+  b.SET.1:v
+L0
+  a.SET.2:v
+----
+0.0:
+  000005:[a#2,SET-a#2,SET]
+  000004:[b#1,SET-b#1,SET]
+
+add-ongoing-compaction startLevel=0 outputLevel=1
+----
+
+async-compact a-b L0
+----
+manual compaction blocked until ongoing finished
+1:
+  000006:[a#0,SET-a#0,SET]
+  000007:[b#0,SET-b#0,SET]
+
+compact a-b L1
+----
+2:
+  000008:[a#0,SET-a#0,SET]
+  000009:[b#0,SET-b#0,SET]
+
+add-ongoing-compaction startLevel=0 outputLevel=1
+----
+
+async-compact a-b L2
+----
+manual compaction blocked until ongoing finished
+3:
+  000010:[a#0,SET-a#0,SET]
+  000011:[b#0,SET-b#0,SET]
+
+add-ongoing-compaction startLevel=0 outputLevel=1
+----
+
+set-concurrent-compactions num=2
+----
+
+async-compact a-b L3
+----
+manual compaction did not block for ongoing
+4:
+  000012:[a#0,SET-a#0,SET]
+  000013:[b#0,SET-b#0,SET]
+
+remove-ongoing-compaction
+----
+
+add-ongoing-compaction startLevel=5 outputLevel=6
+----
+
+async-compact a-b L4
+----
+manual compaction blocked until ongoing finished
+5:
+  000014:[a#0,SET-a#0,SET]
+  000015:[b#0,SET-b#0,SET]
+
+# Test of a scenario where consecutive elided range tombstones and grandparent
+# boundaries could result in an invariant violation in the rangedel fragmenter.
+
+define target-file-sizes=(1, 1, 1, 1)
+L1
+  a.RANGEDEL.4:b
+  c.RANGEDEL.4:d
+  e.RANGEDEL.4:f
+L1
+  g.RANGEDEL.6:h
+  i.RANGEDEL.4:j
+L1
+  k.RANGEDEL.5:q
+  m.RANGEDEL.4:q
+L2
+  a.SET.2:foo
+L3
+  a.SET.1:foo
+  c.SET.1:foo
+L3
+  ff.SET.1:v
+L3
+  k.SET.1:foo
+----
+1:
+  000004:[a#4,RANGEDEL-f#72057594037927935,RANGEDEL]
+  000005:[g#6,RANGEDEL-j#72057594037927935,RANGEDEL]
+  000006:[k#5,RANGEDEL-q#72057594037927935,RANGEDEL]
+2:
+  000007:[a#2,SET-a#2,SET]
+3:
+  000008:[a#1,SET-c#1,SET]
+  000009:[ff#1,SET-ff#1,SET]
+  000010:[k#1,SET-k#1,SET]
+
+compact a-q L1
+----
+2:
+  000011:[a#4,RANGEDEL-d#72057594037927935,RANGEDEL]
+  000012:[k#5,RANGEDEL-m#72057594037927935,RANGEDEL]
+3:
+  000008:[a#1,SET-c#1,SET]
+  000009:[ff#1,SET-ff#1,SET]
+  000010:[k#1,SET-k#1,SET]
+
+# Test a case where a new output file is started, there are no previous output
+# files, there are no additional keys (key = nil) and the rangedel fragmenter
+# is non-empty.
+define target-file-sizes=(1, 1, 1)
+L1
+  a.RANGEDEL.10:b
+  d.RANGEDEL.9:e
+  q.RANGEDEL.8:r
+L2
+  g.RANGEDEL.7:h
+L3
+  q.SET.6:6
+----
+1:
+  000004:[a#10,RANGEDEL-r#72057594037927935,RANGEDEL]
+2:
+  000005:[g#7,RANGEDEL-h#72057594037927935,RANGEDEL]
+3:
+  000006:[q#6,SET-q#6,SET]
+
+compact a-r L1
+----
+2:
+  000007:[q#8,RANGEDEL-r#72057594037927935,RANGEDEL]
+3:
+  000006:[q#6,SET-q#6,SET]
+
+define target-file-sizes=(100, 100, 100)
+L1
+  a.RANGEDEL.10:b
+  b.SET.0:foo
+  d.RANGEDEL.0:e
+  j.SET.10:foo
+L2
+  f.RANGEDEL.7:g
+L3
+  c.SET.6:6
+L3
+  c.SET.5:5
+L3
+  c.SET.4:4
+L4
+  a.SET.0:0
+  f.SET.0:0
+----
+1:
+  000004:[a#10,RANGEDEL-j#10,SET]
+2:
+  000005:[f#7,RANGEDEL-g#72057594037927935,RANGEDEL]
+3:
+  000006:[c#6,SET-c#6,SET]
+  000007:[c#5,SET-c#5,SET]
+  000008:[c#4,SET-c#4,SET]
+4:
+  000009:[a#0,SET-f#0,SET]
+
+compact a-r L1
+----
+2:
+  000010:[a#10,RANGEDEL-b#0,SET]
+  000011:[d#0,RANGEDEL-j#10,SET]
+3:
+  000006:[c#6,SET-c#6,SET]
+  000007:[c#5,SET-c#5,SET]
+  000008:[c#4,SET-c#4,SET]
+4:
+  000009:[a#0,SET-f#0,SET]
+
+# Test a snapshot that separates a range deletion from all the data that it
+# deletes. Ensure that we respect the target-file-size and split into multiple
+# outputs.
+
+define target-file-sizes=(1, 1, 1) snapshots=(14)
+L1
+  a.RANGEDEL.15:z
+  b.SET.11:foo
+  c.SET.11:foo
+L2
+  c.SET.0:foo
+  d.SET.0:foo
+----
+1:
+  000004:[a#15,RANGEDEL-z#72057594037927935,RANGEDEL]
+2:
+  000005:[c#0,SET-d#0,SET]
+
+compact a-z L1
+----
+2:
+  000006:[a#15,RANGEDEL-c#72057594037927935,RANGEDEL]
+  000007:[c#15,RANGEDEL-d#72057594037927935,RANGEDEL]
+  000008:[d#15,RANGEDEL-z#72057594037927935,RANGEDEL]
+
+# Test an interaction between a range deletion that will be elided with
+# output splitting. Ensure that the output is still split (previous versions
+# of the code did not, because of intricacies around preventing a zero
+# sequence number in an output's largest key).
+
+define target-file-sizes=(1, 1, 1)
+L1
+  a.RANGEDEL.10:z
+  b.SET.11:foo
+  c.SET.11:foo
+L2
+  c.SET.0:foo
+  d.SET.0:foo
+----
+1:
+  000004:[a#10,RANGEDEL-z#72057594037927935,RANGEDEL]
+2:
+  000005:[c#0,SET-d#0,SET]
+
+compact a-z L1
+----
+2:
+  000006:[b#0,SET-b#0,SET]
+  000007:[c#0,SET-c#0,SET]
diff --git a/testdata/singledel_manual_compaction b/testdata/singledel_manual_compaction
index ff83e376..a6c7a714 100644
--- a/testdata/singledel_manual_compaction
+++ b/testdata/singledel_manual_compaction
@@ -17,15 +17,15 @@ L5
   a.SET.6:v1
 ----
 1:
-  000004:[a-a]
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
 2:
-  000005:[a-a]
+  000005:[a#9,SET-a#9,SET]
 3:
-  000006:[a-a]
+  000006:[a#8,DEL-a#8,DEL]
 4:
-  000007:[a-a]
+  000007:[a#7,SET-a#7,SET]
 5:
-  000008:[a-a]
+  000008:[a#6,SET-a#6,SET]
 
 # No data.
 iter
diff --git a/testdata/singledel_manual_compaction_set_with_del b/testdata/singledel_manual_compaction_set_with_del
new file mode 100644
index 00000000..29afc2e1
--- /dev/null
+++ b/testdata/singledel_manual_compaction_set_with_del
@@ -0,0 +1,281 @@
+# This is not actually a manual compaction test, and simply uses manual
+# compaction to demonstrate single delete semantics when used with
+# set-with-delete.
+
+# Define a sequence of SET=>SET=>DEL=>SET=>SINGLEDEL.
+define target-file-sizes=(1, 1, 1, 1, 1)
+L1
+  a.SINGLEDEL.10:
+L2
+  a.SET.9:v3
+L3
+  a.DEL.8:
+L4
+  a.SET.7:v2
+L5
+  a.SET.6:v1
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+2:
+  000005:[a#9,SET-a#9,SET]
+3:
+  000006:[a#8,DEL-a#8,DEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+5:
+  000008:[a#6,SET-a#6,SET]
+
+# No data.
+iter
+first
+----
+.
+
+# Compact away the DEL.
+compact a-b L2
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000009:[a#9,SETWITHDEL-a#9,SETWITHDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+5:
+  000008:[a#6,SET-a#6,SET]
+
+# No data.
+iter
+first
+----
+.
+
+# Do two compactions to compact away the SINGLEDEL and 1 SET.
+compact a-b L1
+----
+2:
+  000010:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000009:[a#9,SETWITHDEL-a#9,SETWITHDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+5:
+  000008:[a#6,SET-a#6,SET]
+
+compact a-b L2
+----
+3:
+  000011:[a#10,DEL-a#10,DEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+5:
+  000008:[a#6,SET-a#6,SET]
+
+# Deleted data is not resurrected.
+iter
+first
+----
+.
+
+# Define a sequence of SET=>SINGLEDEL=>SET=>SINGLEDEL.
+define target-file-sizes=(1, 1, 1, 1, 1)
+L1
+  a.SINGLEDEL.10:
+L2
+  a.SET.9:v3
+L3
+  a.SINGLEDEL.8:
+L4
+  a.SET.7:v2
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+2:
+  000005:[a#9,SET-a#9,SET]
+3:
+  000006:[a#8,SINGLEDEL-a#8,SINGLEDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data.
+iter
+first
+----
+.
+
+# Compact away the older SINGLEDEL.
+compact a-b L2
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000008:[a#9,SETWITHDEL-a#9,SETWITHDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data.
+iter
+first
+----
+.
+
+# Do two compactions to compact away the newer SINGLEDEL and 1 SET.
+compact a-b L1
+----
+2:
+  000009:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000008:[a#9,SETWITHDEL-a#9,SETWITHDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+compact a-b L2
+----
+3:
+  000010:[a#10,DEL-a#10,DEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# Deleted data is not resurrected.
+iter
+first
+----
+.
+
+# Define a sequence of SET=>DEL=>SET=>SINGLEDEL, such that the DEL and
+# SINGLEDEL meet in a compaction.
+define snapshots=(9)
+L1
+  a.SINGLEDEL.10:
+L2
+  a.SET.9:v3
+L3
+  a.DEL.8:
+L4
+  a.SET.7:v2
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+2:
+  000005:[a#9,SET-a#9,SET]
+3:
+  000006:[a#8,DEL-a#8,DEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data.
+iter
+first
+----
+.
+
+# Compact L2 and L3. The snapshot prevents the DEL=>SET from being collapsed.
+compact a-b L2
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000008:[a#9,SET-a#8,DEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data.
+iter
+first
+----
+.
+
+close-snapshots
+----
+
+compact a-b L1
+----
+2:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000008:[a#9,SET-a#8,DEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# The DEL survives.
+compact a-b L2
+----
+3:
+  000009:[a#8,DEL-a#8,DEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data
+iter
+first
+----
+.
+
+# Define a sequence of SET=>SINGLEDEL=>SET=>SINGLEDEL, such that the two
+# SINGLEDELs meet in a compaction.
+define snapshots=(9)
+L1
+  a.SINGLEDEL.10:
+L2
+  a.SET.9:v3
+L3
+  a.SINGLEDEL.8:
+L4
+  a.SET.7:v2
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+2:
+  000005:[a#9,SET-a#9,SET]
+3:
+  000006:[a#8,SINGLEDEL-a#8,SINGLEDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data.
+iter
+first
+----
+.
+
+# Compact L2 and L3. The snapshot prevents the SINGLEDEL=>SET from being collapsed.
+compact a-b L2
+----
+1:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000008:[a#9,SET-a#8,SINGLEDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data.
+iter
+first
+----
+.
+
+close-snapshots
+----
+
+compact a-b L1
+----
+2:
+  000004:[a#10,SINGLEDEL-a#10,SINGLEDEL]
+3:
+  000008:[a#9,SET-a#8,SINGLEDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# The SINGLEDEL survives.
+compact a-b L2
+----
+3:
+  000009:[a#8,SINGLEDEL-a#8,SINGLEDEL]
+4:
+  000007:[a#7,SET-a#7,SET]
+
+# No data
+iter
+first
+----
+.