Repository: cockroachdb/pebble
Pull request #1277 (2021-09-17 17:17:11)

metamorphic: add key manager abstraction and SetWithDelete bug fix
Currently, the metamorphic test generator tracks the state of keys that
it generates to determine whether or not certain keys can be used for
specific operations. Specifically, keys that have been set once in the
database are eligible to be SingleDeleted. Future enhancements to the
test suite will likely introduce other desirable sequences of operations
that rely on tracking the state of certain keys through the metamorphic
test.

State tracking for keys is currently embedded directory onto the
`generator` struct. Additional fields that are to be used for state
tracking would continue to complicate the already sizable struct. There
is also significant cognitive overhead with the existing state tracking
for keys.

This commit introduces the concept of a "key manager", dedicated to
tracking the state of keys during metamorphic test generation time.
Internally, the `keyManager` struct can be passed operations via
`ops` (e.g. `setOp`, etc.) that transform the internal state. The
`keyManager` then exposes functions that can be used to deduce which
keys are eligible for certain operations at a point in time.

As part of this change, we split the paths that are asking for keys
for reads, writes (excluding SingleDelete), SingleDelete writes.

We can now generate complex sequences involving SingleDelete, which
lead to dicovering a bug where the code diverged from the solution
outlined in the issue). The following sequence
  SET => (SINGLEDEL => SET) => SINGLEDEL
  could be transformed to
  SET => (SET => SINGLEDEL)
  to
  SET
which is incorrect. compactionIter contains the fix, and there is
a datadriven test that exercises this behavior. There are also
datadriven test cases that exercise some paths caused by snapshots.

One limitation of the keyManager is that since it used in the
generation phase it cannot account for operations that fail during
execution. We have one case of that involving multiple batches
being ingested. A failed operation can cause the keyManager to no
longer have the correct state of what will actually happen at runtime.
Which means a sequence of operations it generates could be invalid
and cause false positive test failures, since the state of different
DBs is allowed to be non-deterministic for these invalid sequences.
We have removed the troublesome failure cases to address this.

Informs #1255


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

nicktrav: Note that there are some algorithmic improvements that can be made to the `keyManager` (have left some TODOs in the code). I didn't want to burn to much time on these, as we only use the key manager once per meta test run (when generating operations).

I also ran the metamorphic tests for ~1 hour just to be certain we weren't regressing. I've also spot checked some generated output to make sure they made senes w.r.t. SingleDel.
sumeerbhola: This is ready for a review:
- I've fixed the test, and added a bug fix, in the second commit.
- These will get squashed before merge, but I am keeping them separate for now. For reviewing purposes it is fine to review them combined.
- There is a metamorphic test failure related to SingleDelete that I am still in the process of diagnosing.


Patch:
diff --git a/compaction_iter.go b/compaction_iter.go
index 17d66fcd..f1c188b3 100644
--- a/compaction_iter.go
+++ b/compaction_iter.go
@@ -350,9 +350,10 @@ func (i *compactionIter) Next() (*InternalKey, []byte) {
 
 		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 entry.
-			// setNext() does the work to move the iterator forward, preserving
-			// the original value, and potentially mutating the key kind.
+			// 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
 
@@ -525,32 +526,33 @@ func (i *compactionIter) setNext() {
 			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 under this
-				// SET, even if we did not actually encounter one.
+				// 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 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 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 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.
+				// - 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 <- .... 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.
+				// 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)
@@ -563,9 +565,11 @@ func (i *compactionIter) setNext() {
 			}
 			return
 		case sameStripeSkippable:
-			// We're still in the same stripe. If this is a DEL, we stop looking
-			// and emit a SETWITHDEL. Subsequent keys are eligible for skipping.
-			if i.iterKey.Kind() == InternalKeyKindDelete {
+			// 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
diff --git a/internal/base/internal.go b/internal/base/internal.go
index b2fbecae..056854ff 100644
--- a/internal/base/internal.go
+++ b/internal/base/internal.go
@@ -46,8 +46,9 @@ const (
 	InternalKeyKindSeparator = 17
 
 	// InternalKeyKindSetWithDelete keys are SET keys that have met with a
-	// DELETE key in a prior compaction. This key kind is specific to Pebble.
-	// See https://github.com/cockroachdb/pebble/issues/1255.
+	// 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,
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


Test patch:
diff --git a/compaction_test.go b/compaction_test.go
index a43c609a..dbdb2cc6 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -1180,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
 
@@ -1291,6 +1291,20 @@ 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)
 			}
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/testdata/compaction_iter_set_with_del b/testdata/compaction_iter_set_with_del
index 5235fe30..da782525 100644
--- a/testdata/compaction_iter_set_with_del
+++ b/testdata/compaction_iter_set_with_del
@@ -766,6 +766,8 @@ first
 ----
 .
 
+# SET that meets a SINGLEDEL is transformed into a SETWITHDEL.
+
 define
 a.SET.2:b
 a.SINGLEDEL.1:
@@ -775,7 +777,7 @@ iter
 first
 next
 ----
-a#2,1:b
+a#2,18:b
 .
 
 define
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
index 0e75649c..53bd1625 100644
--- a/testdata/manual_compaction_set_with_del
+++ b/testdata/manual_compaction_set_with_del
@@ -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/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
index 7d7e1053..29afc2e1 100644
--- a/testdata/singledel_manual_compaction_set_with_del
+++ b/testdata/singledel_manual_compaction_set_with_del
@@ -16,15 +16,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
@@ -76,3 +76,206 @@ 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
+----
+.