Repository: cockroachdb/pebble
Pull request #4207 (2024-12-19 19:54:56)

sstable/blob: new package
Add blob FileWriter type for writing out blob files, consisting of a series of
value blocks, a value block index and a footer.

Add the initial, largely unimplemented stencil of a FileReader type.

Informs #112.


Patch:
diff --git a/README.md b/README.md
index 683e06ec..c591a6af 100644
--- a/README.md
+++ b/README.md
@@ -86,16 +86,16 @@ differences.
 
 ## RocksDB Compatibility
 
-Pebble strives for forward compatibility with RocksDB 6.2.1 (the latest version
-of RocksDB used by CockroachDB). Forward compatibility means that a DB generated
-by RocksDB 6.2.1 can be upgraded for use by Pebble. Pebble versions in the `v1`
-series may open DBs generated by RocksDB 6.2.1. Since its introduction, Pebble
-has adopted various backwards-incompatible format changes that are gated behind
-new 'format major versions'. The Pebble `master` branch does not support opening
-DBs generated by RocksDB. DBs generated by RocksDB may only be used with recent
-versions of Pebble after migrating them through format major version upgrades
-using previous versions of Pebble. See the below section of format major
-versions.
+Pebble `v1` strives for forward compatibility with RocksDB 6.2.1 (the latest
+version of RocksDB used by CockroachDB). Forward compatibility means that a DB
+generated by RocksDB 6.2.1 can be upgraded for use by Pebble. Pebble versions in
+the `v1` series may open DBs generated by RocksDB 6.2.1. Since its introduction,
+Pebble has adopted various backwards-incompatible format changes that are gated
+behind new 'format major versions'. Pebble `v2` and newer does not support
+opening DBs generated by RocksDB. DBs generated by RocksDB may only be used with
+recent versions of Pebble after migrating them through format major version
+upgrades using previous versions of Pebble. See the below section of format
+major versions.
 
 Even the RocksDB-compatible versions of Pebble only provide compatibility with
 the subset of functionality and configuration used by CockroachDB. The scope of
@@ -157,10 +157,13 @@ range of Pebble versions support that format.
 | FormatMinTableFormatPebblev1       |   9   | No         | v1             |
 | FormatPrePebblev1Marked            |  10   | Background | v1             |
 | FormatSSTableValueBlocks           |  12   | No         | v1             |
-| FormatFlushableIngest              |  13   | No         | v1, master     |
-| FormatPrePebblev1MarkedCompacted   |  14   | Blocking   | v1, master     |
-| FormatDeleteSizedAndObsolete       |  15   | No         | v1, master     |
-| FormatVirtualSSTables              |  16   | No         | v1, master     |
+| FormatFlushableIngest              |  13   | No         | v1, v2, master |
+| FormatPrePebblev1MarkedCompacted   |  14   | Blocking   | v1, v2, master |
+| FormatDeleteSizedAndObsolete       |  15   | No         | v1, v2, master |
+| FormatVirtualSSTables              |  16   | No         | v1, v2, master |
+| FormatSyntheticPrefixSuffix        |  17   | No         | v2, master     |
+| FormatFlushableIngestExcises       |  18   | No         | v2, master     |
+| FormatColumnarBlocks               |  19   | No         | v2, master     |
 
 Upgrading to a format major version with 'Background' in the migration
 column may trigger background activity to rewrite physical file
@@ -171,18 +174,30 @@ writes if upgrading a live database through
 `RatchetFormatMajorVersion`, but the method call will not return until
 the migration is complete.
 
+Upgrading existing stores can be performed via the `RatchetFormatMajorVersion`
+method. If the database does not use a custom comparer, merger, or block
+property collectors, the `pebble` tool can also be used, at the latest version
+that supports the format. For example:
+```
+# WARNING: only use if no custom comparer/merger/property collector are necessary.
+go run github.com/cockroachdb/pebble/cmd/pebble@v1.1.3 db upgrade <db-dir>
+```
+
 For reference, the table below lists the range of supported Pebble format major
 versions for CockroachDB releases.
 
-| CockroachDB release | Earliest supported                 | Latest supported          |
-|---------------------|------------------------------------|---------------------------|
-| 20.1 through 21.1   | FormatMostCompatible               | FormatMostCompatible      |
-| 21.2                | FormatMostCompatible               | FormatSetWithDelete       |
-| 21.2                | FormatMostCompatible               | FormatSetWithDelete       |
-| 22.1                | FormatMostCompatible               | FormatSplitUserKeysMarked |
-| 22.2                | FormatMostCompatible               | FormatPrePebblev1Marked   |
-| 23.1                | FormatSplitUserKeysMarkedCompacted | FormatFlushableIngest     |
-| 23.2                | FormatSplitUserKeysMarkedCompacted | FormatVirtualSSTables     |
+| CockroachDB release | Earliest supported                 | Latest supported            |
+|---------------------|------------------------------------|-----------------------------|
+| 20.1 through 21.1   | FormatMostCompatible               | FormatMostCompatible        |
+| 21.2                | FormatMostCompatible               | FormatSetWithDelete         |
+| 21.2                | FormatMostCompatible               | FormatSetWithDelete         |
+| 22.1                | FormatMostCompatible               | FormatSplitUserKeysMarked   |
+| 22.2                | FormatMostCompatible               | FormatPrePebblev1Marked     |
+| 23.1                | FormatSplitUserKeysMarkedCompacted | FormatFlushableIngest       |
+| 23.2                | FormatPrePebblev1Marked            | FormatVirtualSSTables       |
+| 24.1                | FormatFlushableIngest              | FormatSyntheticPrefixSuffix |
+| 24.2                | FormatVirtualSSTables              | FormatSyntheticPrefixSuffix |
+| 24.3                | FormatSyntheticPrefixSuffix        | FormatColumnarBlocks        |
 
 ## Pedigree
 
diff --git a/cmd/pebble/ycsb.go b/cmd/pebble/ycsb.go
index 126560b6..4231c240 100644
--- a/cmd/pebble/ycsb.go
+++ b/cmd/pebble/ycsb.go
@@ -482,7 +482,7 @@ func (y *ycsb) insert(db DB, buf *ycsbBuf) {
 			log.Fatal(err)
 		}
 		if delta > 0 {
-			y.keyDist.IncMax(delta)
+			y.keyDist.IncMax(uint64(delta))
 		}
 	}
 }
diff --git a/cockroachkvs/cockroachkvs.go b/cockroachkvs/cockroachkvs.go
index cc5c61f2..b3d2860b 100644
--- a/cockroachkvs/cockroachkvs.go
+++ b/cockroachkvs/cockroachkvs.go
@@ -10,8 +10,10 @@ import (
 	"bytes"
 	"cmp"
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
 	"io"
+	"strconv"
 	"strings"
 	"unsafe"
 
@@ -63,7 +65,7 @@ var Comparer = base.Comparer{
 		}
 		return base.DefaultComparer.AbbreviatedKey(key)
 	},
-	FormatKey: base.DefaultFormatter,
+	FormatKey: FormatKey,
 	Separator: func(dst, a, b []byte) []byte {
 		aKey, ok := getKeyPartFromEngineKey(a)
 		if !ok {
@@ -932,3 +934,164 @@ func checkEngineKey(k []byte) {
 		panic(errors.AssertionFailedf("invalid key terminator byte 1"))
 	}
 }
+
+// FormatKey returns a formatter for the user key.
+func FormatKey(key []byte) fmt.Formatter {
+	return formatKey(key)
+}
+
+type formatKey []byte
+
+func (fk formatKey) Format(f fmt.State, c rune) {
+	i := Split(fk)
+	fmt.Fprintf(f, "%s", []byte(fk)[:i-1])
+	if i == len(fk) {
+		// Suffix-less
+		return
+	}
+	fmt.Fprint(f, "@")
+	formatKeySuffix(fk[i:]).Format(f, c)
+}
+
+type formatKeySuffix []byte
+
+func (fk formatKeySuffix) Format(f fmt.State, c rune) {
+	switch len(fk) {
+	case suffixLenWithLockTable:
+		// Format as a UUID.
+		s := []byte(fk)
+		fmt.Fprintf(f, "%02x,%x-%x-%x-%x-%x", s[0], s[1:5], s[5:7], s[7:9], s[9:11], s[11:17])
+	case suffixLenWithSynthetic, suffixLenWithWall, suffixLenWithLogical:
+		// Format as a timestamp.
+		wallTime := binary.BigEndian.Uint64(fk)
+		logical := 0
+		if len(fk) >= suffixLenWithLogical {
+			logical = int(binary.BigEndian.Uint32(fk[8:]))
+		}
+		if wallTime%1e9 == 0 {
+			fmt.Fprintf(f, "%d,%d", wallTime/1e9, logical)
+		} else {
+			fmt.Fprintf(f, "%d.%09d,%d", wallTime/1e9, wallTime%1e9, logical)
+		}
+	default:
+		panic(errors.AssertionFailedf("invalid suffix length: %d", len(fk)))
+	}
+}
+
+// ParseFormattedKey parses a human-readable representation of a Cockroach key.
+//
+// ParseFormattedKey expects the roach key and version to be delimited by a '@'
+// character. The '@' may be omitted for version-less keys. Roach keys may be
+// quoted using Go syntax. The version may be:
+//
+//  1. A MVCC timestamp in one of the formats (these are cribbed from the
+//     Cockroach repository's hlc timestamp formatting):
+//     a. <WALLTIME-SECONDS>
+//     b. <WALLTIME-SECONDS>.<WALLTIME-NANOS>
+//     c. <WALLTIME-SECONDS>,<LOGICAL>
+//     d. <WALLTIME-SECONDS>.<WALLTIME-NANOS>,<LOGICAL>
+//  2. A lock table key in the format: <STRENGTH-BYTE>,<TXNUUID>
+func ParseFormattedKey(formattedKey string) []byte {
+	i := strings.IndexByte(formattedKey, '@')
+	if i == -1 {
+		// Suffix-less.
+		return append([]byte(formattedKey), 0x00)
+	}
+	return append(append([]byte(formattedKey[:i]), 0x00), ParseFormattedKeySuffix(formattedKey[i+1:])...)
+}
+
+// ParseFormattedKeySuffix parses a human-readable representation of a Cockroach
+// key's version. The formatted suffix may be:
+//
+//  1. A MVCC timestamp in one of the formats (these are cribbed from the
+//     Cockroach repository's hlc timestamp formatting):
+//     a. <WALLTIME-SECONDS>
+//     b. <WALLTIME-SECONDS>.<WALLTIME-NANOS>
+//     c. <WALLTIME-SECONDS>,<LOGICAL>
+//     d. <WALLTIME-SECONDS>.<WALLTIME-NANOS>,<LOGICAL>
+//  2. A lock table key in the format: <STRENGTH-BYTE>,<TXN-UUID>
+func ParseFormattedKeySuffix(formattedSuffix string) []byte {
+	switch len(formattedSuffix) {
+	case 0:
+		return nil
+	case 39:
+		// A lock-table key.
+		k := make([]byte, suffixLenWithLockTable)
+		// The lock strength is encoded as a hexadecimal digit (2 chars),
+		// followed by a comma.
+		hex.Decode(k, unsafe.Slice(unsafe.StringData(formattedSuffix), 2))
+		if err := decodeUUID(k[1:], unsafe.Slice(unsafe.StringData(formattedSuffix[3:]), len(formattedSuffix)-3)); err != nil {
+			panic(errors.Newf("invalid lock table suffix: %w", err))
+		}
+		k[len(k)-1] = suffixLenWithLockTable
+		return k
+	default:
+		nano, logical, err := parseFormattedMVCCtimestamp(formattedSuffix)
+		if err != nil {
+			panic(err)
+		}
+		if logical == 0 {
+			k := make([]byte, suffixLenWithWall)
+			binary.BigEndian.PutUint64(k, nano)
+			k[len(k)-1] = suffixLenWithWall
+			return k
+		}
+		k := make([]byte, suffixLenWithLogical)
+		binary.BigEndian.PutUint64(k, nano)
+		binary.BigEndian.PutUint32(k[8:], uint32(logical))
+		k[len(k)-1] = suffixLenWithLogical
+		return k
+	}
+}
+
+func parseFormattedMVCCtimestamp(formattedSuffix string) (nanos, logical uint64, err error) {
+	// Try to decode as a MVCC timestamp.
+	i := strings.IndexByte(formattedSuffix, ',')
+	if i == -1 {
+		return 0, 0, errors.Newf("invalid MVCC timestamp suffix: %q", formattedSuffix)
+	}
+	j := strings.IndexByte(formattedSuffix[:i], '.')
+	if j >= 0 {
+		nanos, err = strconv.ParseUint(formattedSuffix[j+1:i], 10, 32)
+		if err != nil {
+			return 0, 0, errors.Newf("invalid MVCC timestamp suffix: %q", formattedSuffix)
+		}
+	} else {
+		j = i
+	}
+	var secs uint64
+	secs, err = strconv.ParseUint(formattedSuffix[:j], 10, 64)
+	if err != nil {
+		return 0, 0, errors.Newf("invalid MVCC timestamp suffix: %q", formattedSuffix)
+	}
+	nanos += secs * 1e9
+	logical, err = strconv.ParseUint(formattedSuffix[i+1:], 10, 32)
+	if err != nil {
+		return 0, 0, errors.Newf("invalid MVCC timestamp suffix: %q", formattedSuffix)
+	}
+	return nanos, logical, nil
+}
+
+var uuidGroupIndices = [][2]int{
+	{0, 4},
+	{4, 6},
+	{6, 8},
+	{8, 10},
+	{10, 16},
+}
+
+func decodeUUID(dst []byte, s []byte) error {
+	for g, indices := range uuidGroupIndices {
+		if len(s) < indices[1] {
+			return errors.Errorf("invalid UUID string: %q", s)
+		}
+		i := g + 2*indices[0]
+		j := g + 2*indices[1]
+		m, err := hex.Decode(dst, s[i:j])
+		if err != nil {
+			return errors.Errorf("invalid UUID string: %q: %w", s, err)
+		}
+		dst = dst[m:]
+	}
+	return nil
+}
diff --git a/compaction.go b/compaction.go
index 8de36afb..0b0ca7a3 100644
--- a/compaction.go
+++ b/compaction.go
@@ -36,14 +36,6 @@ var errEmptyTable = errors.New("pebble: empty table")
 // concurrent excise or ingest-split operation.
 var ErrCancelledCompaction = errors.New("pebble: compaction cancelled by a concurrent operation, will retry compaction")
 
-var compactLabelsNoLevel = pprof.Labels("pebble", "compact")
-var compactLabelsPerLevel = func() [manifest.NumLevels]pprof.LabelSet {
-	var labels [manifest.NumLevels]pprof.LabelSet
-	for i := 0; i < manifest.NumLevels; i++ {
-		labels[i] = pprof.Labels("pebble", "compact", "output-level", fmt.Sprintf("L%d", i))
-	}
-	return labels
-}()
 var flushLabels = pprof.Labels("pebble", "flush", "output-level", "L0")
 var gcLabels = pprof.Labels("pebble", "gc")
 
@@ -2180,16 +2172,26 @@ func checkDeleteCompactionHints(
 	return compactLevels, resolvedHints, unresolvedHints
 }
 
+func (d *DB) compactionPprofLabels(c *compaction) pprof.LabelSet {
+	activity := "compact"
+	if len(c.flushing) != 0 {
+		activity = "flush"
+	}
+	level := "L?"
+	// Delete-only compactions don't have an output level.
+	if c.outputLevel != nil {
+		level = fmt.Sprintf("L%d", c.outputLevel.level)
+	}
+	if kc := d.opts.Experimental.UserKeyCategories; kc.Len() > 0 {
+		cat := kc.CategorizeKeyRange(c.smallest.UserKey, c.largest.UserKey)
+		return pprof.Labels("pebble", activity, "output-level", level, "key-type", cat)
+	}
+	return pprof.Labels("pebble", activity, "output-level", level)
+}
+
 // compact runs one compaction and maybe schedules another call to compact.
 func (d *DB) compact(c *compaction, errChannel chan error) {
-	var labels pprof.LabelSet
-	if c.outputLevel == nil {
-		// Delete-only compactions don't have an output level.
-		labels = compactLabelsNoLevel
-	} else {
-		labels = compactLabelsPerLevel[c.outputLevel.level]
-	}
-	pprof.Do(context.Background(), labels, func(context.Context) {
+	pprof.Do(context.Background(), d.compactionPprofLabels(c), func(context.Context) {
 		d.mu.Lock()
 		defer d.mu.Unlock()
 		if err := d.compact1(c, errChannel); err != nil {
@@ -2917,14 +2919,24 @@ func (d *DB) compactAndWrite(
 	}
 	c.allowedZeroSeqNum = c.allowZeroSeqNum()
 	cfg := compact.IterConfig{
-		Comparer:                               c.comparer,
-		Merge:                                  d.merge,
-		TombstoneElision:                       c.delElision,
-		RangeKeyElision:                        c.rangeKeyElision,
-		Snapshots:                              snapshots,
-		AllowZeroSeqNum:                        c.allowedZeroSeqNum,
-		IneffectualSingleDeleteCallback:        d.opts.Experimental.IneffectualSingleDeleteCallback,
-		SingleDeleteInvariantViolationCallback: d.opts.Experimental.SingleDeleteInvariantViolationCallback,
+		Comparer:         c.comparer,
+		Merge:            d.merge,
+		TombstoneElision: c.delElision,
+		RangeKeyElision:  c.rangeKeyElision,
+		Snapshots:        snapshots,
+		AllowZeroSeqNum:  c.allowedZeroSeqNum,
+		IneffectualSingleDeleteCallback: func(userKey []byte) {
+			d.opts.EventListener.PossibleAPIMisuse(PossibleAPIMisuseInfo{
+				Kind:    IneffectualSingleDelete,
+				UserKey: slices.Clone(userKey),
+			})
+		},
+		NondeterministicSingleDeleteCallback: func(userKey []byte) {
+			d.opts.EventListener.PossibleAPIMisuse(PossibleAPIMisuseInfo{
+				Kind:    NondeterministicSingleDelete,
+				UserKey: slices.Clone(userKey),
+			})
+		},
 	}
 	iter := compact.NewIter(cfg, pointIter, rangeDelIter, rangeKeyIter)
 
diff --git a/db.go b/db.go
index dd7f7db4..89ed1e69 100644
--- a/db.go
+++ b/db.go
@@ -2369,7 +2369,7 @@ func (d *DB) newMemTable(
 	// existing memory.
 	var mem *memTable
 	mem = d.memTableRecycle.Swap(nil)
-	if mem != nil && uint64(len(mem.arenaBuf)) != size {
+	if mem != nil && uint64(mem.arenaBuf.Len()) != size {
 		d.freeMemTable(mem)
 		mem = nil
 	}
@@ -2379,7 +2379,7 @@ func (d *DB) newMemTable(
 		memtblOpts.releaseAccountingReservation = mem.releaseAccountingReservation
 	} else {
 		mem = new(memTable)
-		memtblOpts.arenaBuf = manual.New(manual.MemTable, int(size))
+		memtblOpts.arenaBuf = manual.New(manual.MemTable, uintptr(size))
 		memtblOpts.releaseAccountingReservation = d.opts.Cache.Reserve(int(size))
 		d.memTableCount.Add(1)
 		d.memTableReserved.Add(int64(size))
@@ -2412,7 +2412,7 @@ func (d *DB) newMemTable(
 
 func (d *DB) freeMemTable(m *memTable) {
 	d.memTableCount.Add(-1)
-	d.memTableReserved.Add(-int64(len(m.arenaBuf)))
+	d.memTableReserved.Add(-int64(m.arenaBuf.Len()))
 	m.free()
 }
 
diff --git a/event.go b/event.go
index 8314380f..19555a30 100644
--- a/event.go
+++ b/event.go
@@ -530,6 +530,119 @@ func (i WriteStallBeginInfo) SafeFormat(w redact.SafePrinter, _ rune) {
 	w.Printf("write stall beginning: %s", redact.Safe(i.Reason))
 }
 
+// PossibleAPIMisuseInfo contains the information for a PossibleAPIMisuse event.
+type PossibleAPIMisuseInfo struct {
+	Kind APIMisuseKind
+
+	// UserKey is set for the following kinds:
+	//  - IneffectualSingleDelete,
+	//  - NondeterministicSingleDelete.
+	UserKey []byte
+}
+
+func (i PossibleAPIMisuseInfo) String() string {
+	switch i.Kind {
+	case IneffectualSingleDelete, NondeterministicSingleDelete:
+		return fmt.Sprintf("%s (key=%q)", i.Kind, i.UserKey)
+	default:
+		return "invalid"
+	}
+}
+
+// APIMisuseKind identifies the type of API misuse represented by a
+// PossibleAPIMisuse event.
+type APIMisuseKind int8
+
+const (
+	// IneffectualSingleDelete is emitted in compactions/flushes if any
+	// single delete is being elided without deleting a point set/merge.
+	//
+	// This event can sometimes be a false positive because of delete-only
+	// compactions which can cause a recent RANGEDEL to peek below an older
+	// SINGLEDEL and delete an arbitrary subset of data below that SINGLEDEL.
+	//
+	// Example:
+	//   RANGEDEL [a, c)#10 in L0
+	//   SINGLEDEL b#5 in L1
+	//   SET b#3 in L6
+	//
+	// If the L6 file containing the SET is narrow and the L1 file containing
+	// the SINGLEDEL is wide, a delete-only compaction can remove the file in
+	// L2 before the SINGLEDEL is compacted down. Then when the SINGLEDEL is
+	// compacted down, it will not find any SET to delete, resulting in the
+	// ineffectual callback.
+	IneffectualSingleDelete APIMisuseKind = iota
+
+	// NondeterministicSingleDelete is emitted in compactions/flushes if any
+	// single delete has consumed a Set/Merge, and there is another immediately
+	// older Set/SetWithDelete/Merge. The user of Pebble has violated the
+	// invariant under which SingleDelete can be used correctly.
+	//
+	// Consider the sequence SingleDelete#3, Set#2, Set#1. There are three
+	// ways some of these keys can first meet in a compaction.
+	//
+	// - All 3 keys in the same compaction: this callback will detect the
+	//   violation.
+	//
+	// - SingleDelete#3, Set#2 meet in a compaction first: Both keys will
+	//   disappear. The violation will not be detected, and the DB will have
+	//   Set#1 which is likely incorrect (from the user's perspective).
+	//
+	// - Set#2, Set#1 meet in a compaction first: The output will be Set#2,
+	//   which will later be consumed by SingleDelete#3. The violation will
+	//   not be detected and the DB will be correct.
+	//
+	// This event can sometimes be a false positive because of delete-only
+	// compactions which can cause a recent RANGEDEL to peek below an older
+	// SINGLEDEL and delete an arbitrary subset of data below that SINGLEDEL.
+	//
+	// Example:
+	//   RANGEDEL [a, z)#60 in L0
+	//   SINGLEDEL g#50 in L1
+	//   SET g#40 in L2
+	//   RANGEDEL [g,h)#30 in L3
+	//   SET g#20 in L6
+	//
+	// In this example, the two SETs represent the same user write, and the
+	// RANGEDELs are caused by the CockroachDB range being dropped. That is,
+	// the user wrote to g once, range was dropped, then added back, which
+	// caused the SET again, then at some point g was validly deleted using a
+	// SINGLEDEL, and then the range was dropped again. The older RANGEDEL can
+	// get fragmented due to compactions it has been part of. Say this L3 file
+	// containing the RANGEDEL is very narrow, while the L1, L2, L6 files are
+	// wider than the RANGEDEL in L0. Then the RANGEDEL in L3 can be dropped
+	// using a delete-only compaction, resulting in an LSM with state:
+	//
+	//   RANGEDEL [a, z)#60 in L0
+	//   SINGLEDEL g#50 in L1
+	//   SET g#40 in L2
+	//   SET g#20 in L6
+	//
+	// A multi-level compaction involving L1, L2, L6 will cause the invariant
+	// violation callback. This example doesn't need multi-level compactions:
+	// say there was a Pebble snapshot at g#21 preventing g#20 from being
+	// dropped when it meets g#40 in a compaction. That snapshot will not save
+	// RANGEDEL [g,h)#30, so we can have:
+	//
+	//   SINGLEDEL g#50 in L1
+	//   SET g#40, SET g#20 in L6
+	//
+	// And say the snapshot is removed and then the L1 and L6 compaction
+	// happens, resulting in the invariant violation callback.
+	NondeterministicSingleDelete
+)
+
+func (k APIMisuseKind) String() string {
+	switch k {
+	case IneffectualSingleDelete:
+		return "ineffectual SINGLEDEL"
+	case NondeterministicSingleDelete:
+		return "nondeterministicv SINGLEDEL"
+	default:
+		return "unknown"
+	}
+}
+
 // EventListener contains a set of functions that will be invoked when various
 // significant DB events occur. Note that the functions should not run for an
 // excessive amount of time as they are invoked synchronously by the DB and may
@@ -612,6 +725,9 @@ type EventListener struct {
 
 	// WriteStallEnd is invoked when delayed writes are released.
 	WriteStallEnd func()
+
+	// PossibleAPIMisuse is invoked when a possible API misuse is detected.
+	PossibleAPIMisuse func(PossibleAPIMisuseInfo)
 }
 
 // EnsureDefaults ensures that background error events are logged to the
@@ -685,6 +801,9 @@ func (l *EventListener) EnsureDefaults(logger Logger) {
 	if l.WriteStallEnd == nil {
 		l.WriteStallEnd = func() {}
 	}
+	if l.PossibleAPIMisuse == nil {
+		l.PossibleAPIMisuse = func(info PossibleAPIMisuseInfo) {}
+	}
 }
 
 // MakeLoggingEventListener creates an EventListener that logs all events to the
@@ -755,6 +874,9 @@ func MakeLoggingEventListener(logger Logger) EventListener {
 		WriteStallEnd: func() {
 			logger.Infof("write stall ending")
 		},
+		PossibleAPIMisuse: func(info PossibleAPIMisuseInfo) {
+			logger.Infof("API misuse: %s", info)
+		},
 	}
 }
 
@@ -843,5 +965,9 @@ func TeeEventListener(a, b EventListener) EventListener {
 			a.WriteStallEnd()
 			b.WriteStallEnd()
 		},
+		PossibleAPIMisuse: func(info PossibleAPIMisuseInfo) {
+			a.PossibleAPIMisuse(info)
+			b.PossibleAPIMisuse(info)
+		},
 	}
 }
diff --git a/file_cache.go b/file_cache.go
index 5d1f966d..efba7d7b 100644
--- a/file_cache.go
+++ b/file_cache.go
@@ -218,10 +218,11 @@ func (c *fileCacheContainer) estimateSize(
 // createCommonReader creates a Reader for this file.
 func createCommonReader(v *fileCacheValue, file *fileMetadata) sstable.CommonReader {
 	// TODO(bananabrick): We suffer an allocation if file is a virtual sstable.
-	var cr sstable.CommonReader = v.reader
+	r := v.mustSSTableReader()
+	var cr sstable.CommonReader = r
 	if file.Virtual {
 		virtualReader := sstable.MakeVirtualReader(
-			v.reader, file.VirtualMeta().VirtualReaderParams(v.isShared),
+			r, file.VirtualMeta().VirtualReaderParams(v.isShared),
 		)
 		cr = &virtualReader
 	}
@@ -247,7 +248,7 @@ func (c *fileCacheContainer) withReader(meta physicalMeta, fn func(*sstable.Read
 	if v.err != nil {
 		return v.err
 	}
-	return fn(v.reader)
+	return fn(v.reader.(*sstable.Reader))
 }
 
 // withVirtualReader fetches a VirtualReader associated with a virtual sstable.
@@ -265,7 +266,7 @@ func (c *fileCacheContainer) withVirtualReader(
 	if err != nil {
 		return err
 	}
-	return fn(sstable.MakeVirtualReader(v.reader, meta.VirtualReaderParams(objMeta.IsShared())))
+	return fn(sstable.MakeVirtualReader(v.mustSSTableReader(), meta.VirtualReaderParams(objMeta.IsShared())))
 }
 
 func (c *fileCacheContainer) iterCount() int64 {
@@ -402,8 +403,8 @@ func (c *fileCacheShard) releaseLoop() {
 // checkAndIntersectFilters checks the specific table and block property filters
 // for intersection with any available table and block-level properties. Returns
 // true for ok if this table should be read by this iterator.
-func (c *fileCacheShard) checkAndIntersectFilters(
-	v *fileCacheValue,
+func checkAndIntersectFilters(
+	r *sstable.Reader,
 	blockPropertyFilters []BlockPropertyFilter,
 	boundLimitedFilter sstable.BoundLimitedBlockPropertyFilter,
 	syntheticSuffix sstable.SyntheticSuffix,
@@ -412,7 +413,7 @@ func (c *fileCacheShard) checkAndIntersectFilters(
 		filterer, err = sstable.IntersectsTable(
 			blockPropertyFilters,
 			boundLimitedFilter,
-			v.reader.Properties.UserProperties,
+			r.Properties.UserProperties,
 			syntheticSuffix,
 		)
 		// NB: IntersectsTable will return a nil filterer if the table-level
@@ -446,15 +447,16 @@ func (c *fileCacheShard) newIters(
 		return iterSet{}, v.err
 	}
 
+	r := v.mustSSTableReader()
 	// Note: This suffers an allocation for virtual sstables.
 	cr := createCommonReader(v, file)
 	var iters iterSet
 	var err error
 	if kinds.RangeKey() && file.HasRangeKeys {
-		iters.rangeKey, err = c.newRangeKeyIter(ctx, v, file, cr, opts.SpanIterOptions())
+		iters.rangeKey, err = newRangeKeyIter(ctx, r, file, cr, opts.SpanIterOptions())
 	}
 	if kinds.RangeDeletion() && file.HasPointKeys && err == nil {
-		iters.rangeDeletion, err = c.newRangeDelIter(ctx, file, cr, dbOpts)
+		iters.rangeDeletion, err = newRangeDelIter(ctx, file, cr, dbOpts)
 	}
 	if kinds.Point() && err == nil {
 		iters.point, err = c.newPointIter(ctx, v, file, cr, opts, internalOpts, dbOpts)
@@ -498,6 +500,7 @@ func (c *fileCacheShard) newPointIter(
 		pointKeyFilters    []BlockPropertyFilter
 		filterer           *sstable.BlockPropertiesFilterer
 	)
+	r := v.mustSSTableReader()
 	if opts != nil {
 		// This code is appending (at most one filter) in-place to
 		// opts.PointKeyFilters even though the slice is shared for iterators in
@@ -519,12 +522,12 @@ func (c *fileCacheShard) newPointIter(
 		// apply the obsolete block property filter. We could optimize this by
 		// applying the filter.
 		hideObsoletePoints, pointKeyFilters =
-			v.reader.TryAddBlockPropertyFilterForHideObsoletePoints(
+			r.TryAddBlockPropertyFilterForHideObsoletePoints(
 				opts.snapshotForHideObsoletePoints, file.LargestSeqNum, opts.PointKeyFilters)
 
 		var ok bool
 		var err error
-		ok, filterer, err = c.checkAndIntersectFilters(v, pointKeyFilters,
+		ok, filterer, err = checkAndIntersectFilters(r, pointKeyFilters,
 			internalOpts.boundLimitedFilter, file.SyntheticPrefixAndSuffix.Suffix())
 		if err != nil {
 			return nil, err
@@ -550,7 +553,7 @@ func (c *fileCacheShard) newPointIter(
 			ctx = objiotracing.WithLevel(ctx, opts.layer.Level())
 		}
 	}
-	tableFormat, err := v.reader.TableFormat()
+	tableFormat, err := r.TableFormat()
 	if err != nil {
 		return nil, err
 	}
@@ -567,7 +570,7 @@ func (c *fileCacheShard) newPointIter(
 	iterStatsAccum := internalOpts.iterStatsAccumulator
 	if iterStatsAccum == nil && opts != nil && dbOpts.sstStatsCollector != nil {
 		iterStatsAccum = dbOpts.sstStatsCollector.Accumulator(
-			uint64(uintptr(unsafe.Pointer(v.reader))), opts.Category)
+			uint64(uintptr(unsafe.Pointer(r))), opts.Category)
 	}
 	if internalOpts.compaction {
 		iter, err = cr.NewCompactionIter(transforms, iterStatsAccum, &v.readerProvider, internalOpts.bufferPool)
@@ -595,7 +598,7 @@ func (c *fileCacheShard) newPointIter(
 // newRangeDelIter is an internal helper that constructs an iterator over a
 // sstable's range deletions. This function is for table-cache internal use
 // only, and callers should use newIters instead.
-func (c *fileCacheShard) newRangeDelIter(
+func newRangeDelIter(
 	ctx context.Context, file *manifest.FileMetadata, cr sstable.CommonReader, dbOpts *fileCacheOpts,
 ) (keyspan.FragmentIterator, error) {
 	// NB: range-del iterator does not maintain a reference to the table, nor
@@ -620,9 +623,9 @@ func (c *fileCacheShard) newRangeDelIter(
 // newRangeKeyIter is an internal helper that constructs an iterator over a
 // sstable's range keys. This function is for table-cache internal use only, and
 // callers should use newIters instead.
-func (c *fileCacheShard) newRangeKeyIter(
+func newRangeKeyIter(
 	ctx context.Context,
-	v *fileCacheValue,
+	r *sstable.Reader,
 	file *fileMetadata,
 	cr sstable.CommonReader,
 	opts keyspan.SpanIterOptions,
@@ -633,8 +636,8 @@ func (c *fileCacheShard) newRangeKeyIter(
 	// file's range key blocks may surface deleted range keys below. This is
 	// done here, rather than deferring to the block-property collector in order
 	// to maintain parity with point keys and the treatment of RANGEDELs.
-	if v.reader.Properties.NumRangeKeyDels == 0 && len(opts.RangeKeyFilters) > 0 {
-		ok, _, err := c.checkAndIntersectFilters(v, opts.RangeKeyFilters, nil, transforms.SyntheticSuffix())
+	if r.Properties.NumRangeKeyDels == 0 && len(opts.RangeKeyFilters) > 0 {
+		ok, _, err := checkAndIntersectFilters(r, opts.RangeKeyFilters, nil, transforms.SyntheticSuffix())
 		if err != nil {
 			return nil, err
 		} else if !ok {
@@ -697,7 +700,7 @@ func (rp *tableCacheShardReaderProvider) GetReader(
 
 	if rp.mu.v != nil {
 		rp.mu.refCount++
-		return rp.mu.v.reader, nil
+		return rp.mu.v.mustSSTableReader(), nil
 	}
 
 	// Calling findNodeInternal gives us the responsibility of decrementing v's
@@ -712,7 +715,7 @@ func (rp *tableCacheShardReaderProvider) GetReader(
 	}
 	rp.mu.v = v
 	rp.mu.refCount = 1
-	return v.reader, nil
+	return v.mustSSTableReader(), nil
 }
 
 // Close implements sstable.ReaderProvider.
@@ -729,7 +732,7 @@ func (rp *tableCacheShardReaderProvider) Close() {
 	}
 }
 
-// getTableProperties return sst table properties for target file
+// getTableProperties return sst table properties for target file.
 func (c *fileCacheShard) getTableProperties(
 	file *fileMetadata, dbOpts *fileCacheOpts,
 ) (*sstable.Properties, error) {
@@ -740,7 +743,8 @@ func (c *fileCacheShard) getTableProperties(
 	if v.err != nil {
 		return nil, v.err
 	}
-	return &v.reader.Properties, nil
+	r := v.mustSSTableReader()
+	return &r.Properties, nil
 }
 
 // releaseNode releases a node from the fileCacheShard.
@@ -813,13 +817,14 @@ func (c *fileCacheShard) unrefValue(v *fileCacheValue) {
 func (c *fileCacheShard) findNode(
 	ctx context.Context, b *fileBacking, dbOpts *fileCacheOpts,
 ) *fileCacheValue {
-	// The backing must have a positive refcount (otherwise it could be deleted at any time).
+	// The backing must have a positive refcount (otherwise it could be deleted
+	// at any time).
 	b.MustHaveRefs()
-	// Caution! Here fileMetadata can be a physical or virtual table. File cache
-	// sstable  readers are associated with the physical backings. All virtual
-	// tables with the same backing will use the same reader from the cache; so
-	// no information that can differ among these virtual tables can be passed
-	// to findNodeInternal.
+	// Caution! Here b can be a physical or virtual sstable, or a blob file.
+	// File cache sstable readers are associated with the physical backings. All
+	// virtual tables with the same backing will use the same reader from the
+	// cache; so no information that can differ among these virtual tables can
+	// be passed to findNodeInternal.
 	backingFileNum := b.DiskFileNum
 
 	return c.findNodeInternal(ctx, backingFileNum, dbOpts)
@@ -1126,7 +1131,7 @@ func (c *fileCacheShard) Close() error {
 
 type fileCacheValue struct {
 	closeHook func(i sstable.Iterator) error
-	reader    *sstable.Reader
+	reader    io.Closer // *sstable.Reader
 	err       error
 	loaded    chan struct{}
 	// Reference count for the value. The reader is closed when the reference
@@ -1142,11 +1147,18 @@ type fileCacheValue struct {
 	readerProvider tableCacheShardReaderProvider
 }
 
+// mustSSTable retrieves the value's *sstable.Reader. It panics if the cached
+// file is not a sstable (i.e., it is a blob file).
+func (v *fileCacheValue) mustSSTableReader() *sstable.Reader {
+	return v.reader.(*sstable.Reader)
+}
+
 func (v *fileCacheValue) load(
 	ctx context.Context, backingFileNum base.DiskFileNum, c *fileCacheShard, dbOpts *fileCacheOpts,
 ) {
 	// Try opening the file first.
 	var f objstorage.Readable
+	var r *sstable.Reader
 	var err error
 	f, err = dbOpts.objProvider.OpenForReading(
 		ctx, fileTypeTable, backingFileNum, objstorage.OpenOptions{MustExist: true},
@@ -1158,9 +1170,10 @@ func (v *fileCacheValue) load(
 			CacheID: dbOpts.cacheID,
 			FileNum: backingFileNum,
 		})
-		v.reader, err = sstable.NewReader(ctx, f, o)
+		r, err = sstable.NewReader(ctx, f, o)
 	}
 	if err == nil {
+		v.reader = r
 		var objMeta objstorage.ObjectMetadata
 		objMeta, err = dbOpts.objProvider.Lookup(fileTypeTable, backingFileNum)
 		v.isShared = objMeta.IsShared()
diff --git a/format_major_version.go b/format_major_version.go
index 09607996..203222a5 100644
--- a/format_major_version.go
+++ b/format_major_version.go
@@ -197,8 +197,6 @@ const (
 	// block.
 	FormatColumnarBlocks
 
-	// TODO(msbutler): add major version for synthetic suffixes
-
 	// -- Add new versions here --
 
 	// FormatNewest is the most recent format major version.
diff --git a/internal/cache/block_map.go b/internal/cache/block_map.go
index 62d50a73..55ebfe40 100644
--- a/internal/cache/block_map.go
+++ b/internal/cache/block_map.go
@@ -27,13 +27,13 @@ type blockMapAllocator struct{}
 
 func (blockMapAllocator) Alloc(n int) []swiss.Group[key, *entry] {
 	size := uintptr(n) * unsafe.Sizeof(swiss.Group[key, *entry]{})
-	buf := manual.New(manual.BlockCacheMap, int(size))
-	return unsafe.Slice((*swiss.Group[key, *entry])(unsafe.Pointer(unsafe.SliceData(buf))), n)
+	buf := manual.New(manual.BlockCacheMap, size)
+	return unsafe.Slice((*swiss.Group[key, *entry])(buf.Data()), n)
 }
 
 func (blockMapAllocator) Free(v []swiss.Group[key, *entry]) {
 	size := uintptr(len(v)) * unsafe.Sizeof(swiss.Group[key, *entry]{})
-	buf := unsafe.Slice((*byte)(unsafe.Pointer(unsafe.SliceData(v))), size)
+	buf := manual.MakeBufUnsafe(unsafe.Pointer(unsafe.SliceData(v)), size)
 	manual.Free(manual.BlockCacheMap, buf)
 }
 
diff --git a/internal/cache/entry.go b/internal/cache/entry.go
index 724cb290..fc753ac0 100644
--- a/internal/cache/entry.go
+++ b/internal/cache/entry.go
@@ -170,7 +170,7 @@ func (e *entry) acquireValue() *Value {
 // the race detector, in go1.15 and earlier).
 const entriesGoAllocated = valueEntryGoAllocated || buildtags.Race
 
-const entrySize = int(unsafe.Sizeof(entry{}))
+const entrySize = unsafe.Sizeof(entry{})
 
 func entryAllocNew() *entry {
 	if invariants.UseFinalizers {
@@ -245,7 +245,7 @@ func (c *entryAllocCache) alloc() *entry {
 			return &entry{}
 		}
 		b := manual.New(manual.BlockCacheEntry, entrySize)
-		return (*entry)(unsafe.Pointer(&b[0]))
+		return (*entry)(b.Data())
 	}
 	e := c.entries[n-1]
 	c.entries = c.entries[:n-1]
@@ -254,7 +254,7 @@ func (c *entryAllocCache) alloc() *entry {
 
 func (c *entryAllocCache) dealloc(e *entry) {
 	if !entriesGoAllocated {
-		buf := unsafe.Slice((*byte)(unsafe.Pointer(e)), entrySize)
+		buf := manual.MakeBufUnsafe(unsafe.Pointer(e), entrySize)
 		manual.Free(manual.BlockCacheEntry, buf)
 	}
 }
diff --git a/internal/cache/value.go b/internal/cache/value.go
index c477021a..c0ac12f1 100644
--- a/internal/cache/value.go
+++ b/internal/cache/value.go
@@ -26,7 +26,7 @@ var _ uint = ValueMetadataSize - uint(unsafe.Sizeof(Value{}))
 
 // Value holds a reference counted immutable value.
 type Value struct {
-	// buf is allocated using the manual package.
+	// buf is part of the slice allocated using the manual package.
 	buf []byte
 	// Reference count for the value. The value is freed when the reference count
 	// drops to zero.
@@ -48,8 +48,8 @@ func newValue(n int) *Value {
 
 	if valueEntryGoAllocated {
 		// Note: if cgo is not enabled, manual.New will do a regular Go allocation.
-		b := manual.New(manual.BlockCacheData, n)
-		v := &Value{buf: b}
+		b := manual.New(manual.BlockCacheData, uintptr(n))
+		v := &Value{buf: b.Slice()}
 		v.ref.init(1)
 		// Note: this is a no-op if invariants and tracing are disabled or race is
 		// enabled.
@@ -66,9 +66,9 @@ func newValue(n int) *Value {
 	// When we're not performing leak detection, the lifetime of the returned
 	// Value is exactly the lifetime of the backing buffer and we can manually
 	// allocate both.
-	b := manual.New(manual.BlockCacheData, ValueMetadataSize+n)
-	v := (*Value)(unsafe.Pointer(&b[0]))
-	v.buf = b[ValueMetadataSize:]
+	b := manual.New(manual.BlockCacheData, ValueMetadataSize+uintptr(n))
+	v := (*Value)(b.Data())
+	v.buf = b.Slice()[ValueMetadataSize:]
 	v.ref.init(1)
 	return v
 }
@@ -81,12 +81,13 @@ func (v *Value) free() {
 		}
 	}
 	if valueEntryGoAllocated {
-		manual.Free(manual.BlockCacheData, v.buf)
+		buf := manual.MakeBufUnsafe(unsafe.Pointer(unsafe.SliceData(v.buf)), uintptr(cap(v.buf)))
+		manual.Free(manual.BlockCacheData, buf)
 		v.buf = nil
 		return
 	}
-	n := ValueMetadataSize + cap(v.buf)
-	buf := unsafe.Slice((*byte)(unsafe.Pointer(v)), n)
+	n := ValueMetadataSize + uintptr(cap(v.buf))
+	buf := manual.MakeBufUnsafe(unsafe.Pointer(v), n)
 	v.buf = nil
 	manual.Free(manual.BlockCacheData, buf)
 }
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index 4f90dc01..bea5f090 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -269,20 +269,24 @@ type IterConfig struct {
 	AllowZeroSeqNum bool
 
 	// IneffectualPointDeleteCallback is called if a SINGLEDEL is being elided
-	// without deleting a point set/merge.
+	// without deleting a point set/merge. False positives are rare but possible
+	// (because of delete-only compactions).
 	IneffectualSingleDeleteCallback func(userKey []byte)
 
-	// SingleDeleteInvariantViolationCallback is called in compactions/flushes if any
+	// NondeterministicSingleDeleteCallback is called in compactions/flushes if any
 	// single delete has consumed a Set/Merge, and there is another immediately older
-	// Set/SetWithDelete/Merge. The user of Pebble has violated the invariant under
-	// which SingleDelete can be used correctly.
-	SingleDeleteInvariantViolationCallback func(userKey []byte)
+	// Set/SetWithDelete/Merge. False positives are rare but possible (because of
+	// delete-only compactions).
+	NondeterministicSingleDeleteCallback func(userKey []byte)
 }
 
 func (c *IterConfig) ensureDefaults() {
 	if c.IneffectualSingleDeleteCallback == nil {
 		c.IneffectualSingleDeleteCallback = func(userKey []byte) {}
 	}
+	if c.NondeterministicSingleDeleteCallback == nil {
+		c.NondeterministicSingleDeleteCallback = func(userKey []byte) {}
+	}
 }
 
 // IterStats are statistics produced by the compaction iterator.
@@ -965,15 +969,13 @@ func (i *Iter) singleDeleteNext() bool {
 				nextKind := i.iterKV.Kind()
 				switch nextKind {
 				case base.InternalKeyKindSet, base.InternalKeyKindSetWithDelete, base.InternalKeyKindMerge:
-					if i.cfg.SingleDeleteInvariantViolationCallback != nil {
-						// sameStripe keys returned by nextInStripe() are already
-						// known to not be covered by a RANGEDEL, so it is an invariant
-						// violation. The rare case is newStripeSameKey, where it is a
-						// violation if not covered by a RANGEDEL.
-						if change == sameStripe ||
-							i.tombstoneCovers(i.iterKV.K, i.curSnapshotSeqNum) == noCover {
-							i.cfg.SingleDeleteInvariantViolationCallback(i.key.UserKey)
-						}
+					// sameStripe keys returned by nextInStripe() are already
+					// known to not be covered by a RANGEDEL, so it is an invariant
+					// violation. The rare case is newStripeSameKey, where it is a
+					// violation if not covered by a RANGEDEL.
+					if change == sameStripe ||
+						i.tombstoneCovers(i.iterKV.K, i.curSnapshotSeqNum) == noCover {
+						i.cfg.NondeterministicSingleDeleteCallback(i.key.UserKey)
 					}
 				case base.InternalKeyKindDelete, base.InternalKeyKindDeleteSized, base.InternalKeyKindSingleDelete:
 				default:
@@ -1075,9 +1077,7 @@ func (i *Iter) skipDueToSingleDeleteElision() {
 					nextKind := i.iterKV.Kind()
 					switch nextKind {
 					case base.InternalKeyKindSet, base.InternalKeyKindSetWithDelete, base.InternalKeyKindMerge:
-						if i.cfg.SingleDeleteInvariantViolationCallback != nil {
-							i.cfg.SingleDeleteInvariantViolationCallback(i.key.UserKey)
-						}
+						i.cfg.NondeterministicSingleDeleteCallback(i.key.UserKey)
 					case base.InternalKeyKindDelete, base.InternalKeyKindDeleteSized, base.InternalKeyKindSingleDelete:
 					default:
 						panic(errors.AssertionFailedf(
diff --git a/internal/manual/manual.go b/internal/manual/manual.go
index 24f02574..d099e779 100644
--- a/internal/manual/manual.go
+++ b/internal/manual/manual.go
@@ -7,10 +7,40 @@ package manual
 import (
 	"fmt"
 	"sync/atomic"
+	"unsafe"
 
 	"github.com/cockroachdb/pebble/internal/invariants"
 )
 
+// Buf is a buffer allocated using this package.
+type Buf struct {
+	data unsafe.Pointer
+	n    uintptr
+}
+
+// MakeBufUnsafe should be used with caution: the given data and n must match
+// exactly the data and length of a Buf obtained from New. It is useful when
+// these are stored implicitly in another type (like a []byte) and we want to
+// reconstruct the Buf.
+func MakeBufUnsafe(data unsafe.Pointer, n uintptr) Buf {
+	return Buf{data: data, n: n}
+}
+
+// Data returns a pointer to the buffer data. If the buffer is not initialized
+// (or is the result of calling New with a zero length), returns nil.
+func (b Buf) Data() unsafe.Pointer {
+	return b.data
+}
+
+func (b Buf) Len() uintptr {
+	return b.n
+}
+
+// Slice converts the buffer to a byte slice.
+func (b Buf) Slice() []byte {
+	return unsafe.Slice((*byte)(b.data), b.n)
+}
+
 // Purpose identifies the use-case for an allocation.
 type Purpose uint8
 
@@ -41,11 +71,11 @@ var counters [NumPurposes]struct {
 	_ [7]uint64
 }
 
-func recordAlloc(purpose Purpose, n int) {
+func recordAlloc(purpose Purpose, n uintptr) {
 	counters[purpose].InUseBytes.Add(int64(n))
 }
 
-func recordFree(purpose Purpose, n int) {
+func recordFree(purpose Purpose, n uintptr) {
 	newVal := counters[purpose].InUseBytes.Add(-int64(n))
 	if invariants.Enabled && newVal < 0 {
 		panic(fmt.Sprintf("negative counter value %d", newVal))
diff --git a/internal/manual/manual_32bit.go b/internal/manual/manual_32bit.go
deleted file mode 100644
index 19369fa0..00000000
--- a/internal/manual/manual_32bit.go
+++ /dev/null
@@ -1,13 +0,0 @@
-// Copyright 2020 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-//go:build 386 || amd64p32 || arm || armbe || ppc || sparc
-// +build 386 amd64p32 arm armbe ppc sparc
-
-package manual
-
-const (
-	// MaxArrayLen is a safe maximum length for slices on this architecture.
-	MaxArrayLen = 1<<31 - 1
-)
diff --git a/internal/manual/manual_64bit.go b/internal/manual/manual_64bit.go
deleted file mode 100644
index 0709dd50..00000000
--- a/internal/manual/manual_64bit.go
+++ /dev/null
@@ -1,13 +0,0 @@
-// Copyright 2020 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-//go:build amd64 || arm64 || arm64be || ppc64 || ppc64le || mips64 || mips64le || s390x || sparc64 || riscv64 || loong64
-// +build amd64 arm64 arm64be ppc64 ppc64le mips64 mips64le s390x sparc64 riscv64 loong64
-
-package manual
-
-const (
-	// MaxArrayLen is a safe maximum length for slices on this architecture.
-	MaxArrayLen = 1<<50 - 1
-)
diff --git a/internal/manual/manual_cgo.go b/internal/manual/manual_cgo.go
index 280f9d42..b15f2f19 100644
--- a/internal/manual/manual_cgo.go
+++ b/internal/manual/manual_cgo.go
@@ -6,7 +6,6 @@ package manual
 
 // #include <stdlib.h>
 import "C"
-import "unsafe"
 
 // The go:linkname directives provides backdoor access to private functions in
 // the runtime. Below we're accessing the throw function.
@@ -21,9 +20,9 @@ func throw(s string)
 // New allocates a slice of size n. The returned slice is from manually managed
 // memory and MUST be released by calling Free. Failure to do so will result in
 // a memory leak.
-func New(purpose Purpose, n int) []byte {
+func New(purpose Purpose, n uintptr) Buf {
 	if n == 0 {
-		return make([]byte, 0)
+		return Buf{}
 	}
 	recordAlloc(purpose, n)
 	// We need to be conscious of the Cgo pointer passing rules:
@@ -45,19 +44,14 @@ func New(purpose Purpose, n int) []byte {
 		// it cannot allocate memory.
 		throw("out of memory")
 	}
-	// Interpret the C pointer as a pointer to a Go array, then slice.
-	return (*[MaxArrayLen]byte)(unsafe.Pointer(ptr))[:n:n]
+	return Buf{data: ptr, n: n}
 }
 
 // Free frees the specified slice. It has to be exactly the slice that was
 // returned by New.
-func Free(purpose Purpose, b []byte) {
-	if cap(b) != 0 {
-		recordFree(purpose, cap(b))
-		if len(b) == 0 {
-			b = b[:cap(b)]
-		}
-		ptr := unsafe.Pointer(&b[0])
-		C.free(ptr)
+func Free(purpose Purpose, b Buf) {
+	if b.n != 0 {
+		recordFree(purpose, b.n)
+		C.free(b.data)
 	}
 }
diff --git a/internal/manual/manual_mipsall.go b/internal/manual/manual_mipsall.go
deleted file mode 100644
index 08bb8808..00000000
--- a/internal/manual/manual_mipsall.go
+++ /dev/null
@@ -1,13 +0,0 @@
-// Copyright 2020 The LevelDB-Go and Pebble Authors. All rights reserved. Use
-// of this source code is governed by a BSD-style license that can be found in
-// the LICENSE file.
-
-//go:build mips || mipsle || mips64p32 || mips64p32le
-// +build mips mipsle mips64p32 mips64p32le
-
-package manual
-
-const (
-	// MaxArrayLen is a safe maximum length for slices on this architecture.
-	MaxArrayLen = 1 << 30
-)
diff --git a/internal/manual/manual_nocgo.go b/internal/manual/manual_nocgo.go
index 8fddb554..5a732b3c 100644
--- a/internal/manual/manual_nocgo.go
+++ b/internal/manual/manual_nocgo.go
@@ -7,17 +7,26 @@
 
 package manual
 
+import "unsafe"
+
 // Provides versions of New and Free when cgo is not available (e.g. cross
 // compilation).
 
 // New allocates a slice of size n.
-func New(purpose Purpose, n int) []byte {
+func New(purpose Purpose, n uintptr) Buf {
+	if n == 0 {
+		return Buf{}
+	}
 	recordAlloc(purpose, n)
-	return make([]byte, n)
+	slice := make([]byte, n)
+	return Buf{
+		data: unsafe.Pointer(unsafe.SliceData(slice)),
+		n:    n,
+	}
 }
 
 // Free frees the specified slice. It has to be exactly the slice that was
 // returned by New.
-func Free(purpose Purpose, b []byte) {
-	recordFree(purpose, cap(b))
+func Free(purpose Purpose, b Buf) {
+	recordFree(purpose, b.n)
 }
diff --git a/internal/randvar/randvar.go b/internal/randvar/randvar.go
index d2c9e21f..063d8989 100644
--- a/internal/randvar/randvar.go
+++ b/internal/randvar/randvar.go
@@ -18,7 +18,7 @@ type Dynamic interface {
 	Static
 
 	// Increment the max value the variable will return.
-	IncMax(delta int)
+	IncMax(delta uint64)
 
 	// Read the current max value the variable will return.
 	Max() uint64
diff --git a/internal/randvar/uniform.go b/internal/randvar/uniform.go
index 4fa95a2e..c7b99255 100644
--- a/internal/randvar/uniform.go
+++ b/internal/randvar/uniform.go
@@ -37,8 +37,8 @@ func NewUniform(min, max uint64) *Uniform {
 }
 
 // IncMax increments max.
-func (g *Uniform) IncMax(delta int) {
-	g.max.Add(uint64(delta))
+func (g *Uniform) IncMax(delta uint64) {
+	g.max.Add(delta)
 }
 
 // Max returns the max value of the distribution.
diff --git a/internal/randvar/zipf.go b/internal/randvar/zipf.go
index cc403afe..55b44cb9 100644
--- a/internal/randvar/zipf.go
+++ b/internal/randvar/zipf.go
@@ -111,10 +111,10 @@ func computeZetaFromScratch(n uint64, theta float64) float64 {
 
 // IncMax increments max and recomputes the internal values that depend on
 // it. Returns an error if the recomputation failed.
-func (z *Zipf) IncMax(delta int) {
+func (z *Zipf) IncMax(delta uint64) {
 	z.mu.Lock()
 	oldMax := z.mu.max
-	z.mu.max += uint64(delta)
+	z.mu.max += delta
 	z.mu.zetaN = computeZetaIncrementally(oldMax+1-z.min, z.mu.max+1-z.min, z.theta, z.mu.zetaN)
 	z.mu.eta = (1 - math.Pow(2.0/float64(z.mu.max+1-z.min), 1.0-z.theta)) / (1.0 - z.zeta2/z.mu.zetaN)
 	z.mu.Unlock()
diff --git a/mem_table.go b/mem_table.go
index 277a8353..d31799f5 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -66,7 +66,7 @@ type memTable struct {
 	cmp         Compare
 	formatKey   base.FormatKey
 	equal       Equal
-	arenaBuf    []byte
+	arenaBuf    manual.Buf
 	skl         arenaskl.Skiplist
 	rangeDelSkl arenaskl.Skiplist
 	rangeKeySkl arenaskl.Skiplist
@@ -93,7 +93,7 @@ func (m *memTable) free() {
 	if m != nil {
 		m.releaseAccountingReservation()
 		manual.Free(manual.MemTable, m.arenaBuf)
-		m.arenaBuf = nil
+		m.arenaBuf = manual.Buf{}
 	}
 }
 
@@ -102,7 +102,7 @@ func (m *memTable) free() {
 // which is used by tests.
 type memTableOptions struct {
 	*Options
-	arenaBuf                     []byte
+	arenaBuf                     manual.Buf
 	size                         int
 	logSeqNum                    base.SeqNum
 	releaseAccountingReservation func()
@@ -110,8 +110,8 @@ type memTableOptions struct {
 
 func checkMemTable(obj interface{}) {
 	m := obj.(*memTable)
-	if m.arenaBuf != nil {
-		fmt.Fprintf(os.Stderr, "%p: memTable buffer was not freed\n", m.arenaBuf)
+	if m.arenaBuf.Data() != nil {
+		fmt.Fprintf(os.Stderr, "%v: memTable buffer was not freed\n", m.arenaBuf)
 		os.Exit(1)
 	}
 }
@@ -151,11 +151,11 @@ func (m *memTable) init(opts memTableOptions) {
 		constructSpan: rangekey.Decode,
 	}
 
-	if m.arenaBuf == nil {
-		m.arenaBuf = make([]byte, opts.size)
+	if m.arenaBuf.Data() == nil {
+		m.arenaBuf = manual.New(manual.MemTable, uintptr(opts.size))
 	}
 
-	arena := arenaskl.NewArena(m.arenaBuf)
+	arena := arenaskl.NewArena(m.arenaBuf.Slice())
 	m.skl.Reset(arena, m.cmp)
 	m.rangeDelSkl.Reset(arena, m.cmp)
 	m.rangeKeySkl.Reset(arena, m.cmp)
diff --git a/metamorphic/config.go b/metamorphic/config.go
index 4c475308..e76c6a86 100644
--- a/metamorphic/config.go
+++ b/metamorphic/config.go
@@ -86,6 +86,10 @@ type OpConfig struct {
 	// writeSuffixDist defines the distribution of key suffixes during writing.
 	// It's a dynamic randvar to roughly emulate workloads with MVCC timestamps,
 	// skewing towards most recent timestamps.
+	//
+	// Some KeyFormats may map the uint64 produced by this distribution into a
+	// multi-dimensional keyspace (for example CockroachDB's MVCC timestamps
+	// consist of separate WallTime and Logical compopnents).
 	writeSuffixDist randvar.Dynamic
 
 	// numInstances defines the number of pebble instances created for this
@@ -368,6 +372,26 @@ type KeyFormat struct {
 
 // KeyGenerator is an interface for generating keys, prefixes and suffixes.
 type KeyGenerator interface {
+	// Configuration methods
+
+	// RecordPrecedingKey may be invoked before generating keys to inform the
+	// key generator of a key that was previously generated and used within a
+	// related test context.
+	//
+	// When metamorphic tests are run as a part of cross-version metamorphic
+	// tests, a database maintains state between individual runs of the
+	// metamorphic test. RecordPrecedingKey is invoked to inform the key
+	// generator of the keys used in the previous metamorphic test run's
+	// operations.
+	//
+	// Implementations may use this information differently, but one expected
+	// use is to update the key generator's record of the maximum suffix, so
+	// that the key generator can continue to generate keys with growing
+	// suffixes, modelling the typical MVCC use case.
+	RecordPrecedingKey(key []byte)
+
+	// Generation methods
+
 	// IncMaxSuffix increases the max suffix range and returns the new maximum
 	// suffix (which is guaranteed to be larger than any previously generated
 	// suffix).
diff --git a/metamorphic/generator.go b/metamorphic/generator.go
index b31eb711..a3a355d7 100644
--- a/metamorphic/generator.go
+++ b/metamorphic/generator.go
@@ -55,7 +55,7 @@ func (o iterOpts) IsZero() bool {
 func GenerateOps(rng *rand.Rand, n uint64, cfg OpConfig) Ops {
 	// Generate a new set of random ops, writing them to <dir>/ops. These will be
 	// read by the child processes when performing a test run.
-	return generate(rng, n, cfg, newKeyManager(1 /* num instances */))
+	return newGenerator(rng, cfg, newKeyManager(1 /* num instances */)).generate(n)
 }
 
 type generator struct {
@@ -149,9 +149,9 @@ func newGenerator(rng *rand.Rand, cfg OpConfig, km *keyManager) *generator {
 	return g
 }
 
-func generate(rng *rand.Rand, count uint64, cfg OpConfig, km *keyManager) []op {
-	g := newGenerator(rng, cfg, km)
-
+// generate generates [count] operations according to the generator's configured
+// distributions.
+func (g *generator) generate(count uint64) []op {
 	opGenerators := []func(){
 		OpBatchAbort:                  g.batchAbort,
 		OpBatchCommit:                 g.batchCommit,
@@ -203,11 +203,11 @@ func generate(rng *rand.Rand, count uint64, cfg OpConfig, km *keyManager) []op {
 
 	// TPCC-style deck of cards randomization. Every time the end of the deck is
 	// reached, we shuffle the deck.
-	deck := randvar.NewDeck(g.rng, cfg.ops[:]...)
+	deck := randvar.NewDeck(g.rng, g.cfg.ops[:]...)
 
 	defer func() {
 		if r := recover(); r != nil {
-			fmt.Fprintln(os.Stderr, formatOps(km.kf, g.ops))
+			fmt.Fprintln(os.Stderr, formatOps(g.keyManager.kf, g.ops))
 			panic(r)
 		}
 	}()
diff --git a/metamorphic/key_manager.go b/metamorphic/key_manager.go
index dbce836e..5cdf944f 100644
--- a/metamorphic/key_manager.go
+++ b/metamorphic/key_manager.go
@@ -13,8 +13,6 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
 	"github.com/cockroachdb/pebble/internal/base"
-	"github.com/cockroachdb/pebble/internal/testkeys"
-	"github.com/stretchr/testify/require"
 )
 
 // keyMeta is metadata associated with an (objID, key) pair, where objID is
@@ -911,25 +909,16 @@ func opWrittenKeys(untypedOp op) [][]byte {
 	return nil
 }
 
-func loadPrecedingKeys(t TestingT, ops []op, cfg *OpConfig, m *keyManager) {
+func loadPrecedingKeys(ops []op, kg KeyGenerator, m *keyManager) {
 	for _, op := range ops {
 		// Pretend we're generating all the operation's keys as potential new
 		// key, so that we update the key manager's keys and prefix sets.
 		for _, k := range opWrittenKeys(op) {
 			m.addNewKey(k)
-
-			// If the key has a suffix, ratchet up the suffix distribution if
-			// necessary.
-			if s := m.kf.Comparer.Split(k); s < len(k) {
-				suffix, err := testkeys.ParseSuffix(k[s:])
-				require.NoError(t, err)
-				if uint64(suffix) > cfg.writeSuffixDist.Max() {
-					diff := int(uint64(suffix) - cfg.writeSuffixDist.Max())
-					cfg.writeSuffixDist.IncMax(diff)
-				}
-			}
+			// Inform the key generator of the key, so it can update the
+			// distribution of key suffixes it generates.
+			kg.RecordPrecedingKey(k)
 		}
-
 		// Update key tracking state.
 		m.update(op)
 	}
diff --git a/metamorphic/meta.go b/metamorphic/meta.go
index c968e1ed..441f4912 100644
--- a/metamorphic/meta.go
+++ b/metamorphic/meta.go
@@ -177,6 +177,14 @@ func RunAndCompare(t *testing.T, rootDir string, rOpts ...RunOption) {
 	// read by the child processes when performing a test run.
 	km := newKeyManager(runOpts.numInstances)
 	cfg := presetConfigs[rng.IntN(len(presetConfigs))]
+	if runOpts.numInstances > 1 {
+		// The multi-instance variant does not support all operations yet.
+		//
+		// TODO(bilal): Address this and use the default configs.
+		cfg = multiInstancePresetConfig
+		cfg.numInstances = runOpts.numInstances
+	}
+	g := newGenerator(rng, cfg, km)
 	if runOpts.previousOpsPath != "" {
 		// During cross-version testing, we load keys from an `ops` file
 		// produced by a metamorphic test run of an earlier Pebble version.
@@ -187,16 +195,10 @@ func RunAndCompare(t *testing.T, rootDir string, rOpts ...RunOption) {
 		require.NoError(t, err)
 		ops, err := parse(opsData, parserOpts{})
 		require.NoError(t, err)
-		loadPrecedingKeys(t, ops, &cfg, km)
+		loadPrecedingKeys(ops, g.keyGenerator, km)
 	}
-	if runOpts.numInstances > 1 {
-		// The multi-instance variant does not support all operations yet.
-		//
-		// TODO(bilal): Address this and use the default configs.
-		cfg = multiInstancePresetConfig
-		cfg.numInstances = runOpts.numInstances
-	}
-	ops := generate(rng, opCount, cfg, km)
+
+	ops := g.generate(opCount)
 	opsPath := filepath.Join(metaDir, "ops")
 	formattedOps := formatOps(km.kf, ops)
 	require.NoError(t, os.WriteFile(opsPath, []byte(formattedOps), 0644))
diff --git a/open.go b/open.go
index 6e983912..4c67d90a 100644
--- a/open.go
+++ b/open.go
@@ -254,7 +254,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 				switch t := mem.flushable.(type) {
 				case *memTable:
 					manual.Free(manual.MemTable, t.arenaBuf)
-					t.arenaBuf = nil
+					t.arenaBuf = manual.Buf{}
 				}
 			}
 			if d.cleanupManager != nil {
diff --git a/options.go b/options.go
index 3726ba21..a1fcaf0f 100644
--- a/options.go
+++ b/options.go
@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"runtime"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -766,88 +767,6 @@ type Options struct {
 		// on shared storage in bytes. If it is 0, no cache is used.
 		SecondaryCacheSizeBytes int64
 
-		// NB: DO NOT crash on SingleDeleteInvariantViolationCallback or
-		// IneffectualSingleDeleteCallback, since these can be false positives
-		// even if SingleDel has been used correctly.
-		//
-		// Pebble's delete-only compactions can cause a recent RANGEDEL to peek
-		// below an older SINGLEDEL and delete an arbitrary subset of data below
-		// that SINGLEDEL. When that SINGLEDEL gets compacted (without the
-		// RANGEDEL), any of these callbacks can happen, without it being a real
-		// correctness problem.
-		//
-		// Example 1:
-		// RANGEDEL [a, c)#10 in L0
-		// SINGLEDEL b#5 in L1
-		// SET b#3 in L6
-		//
-		// If the L6 file containing the SET is narrow and the L1 file containing
-		// the SINGLEDEL is wide, a delete-only compaction can remove the file in
-		// L2 before the SINGLEDEL is compacted down. Then when the SINGLEDEL is
-		// compacted down, it will not find any SET to delete, resulting in the
-		// ineffectual callback.
-		//
-		// Example 2:
-		// RANGEDEL [a, z)#60 in L0
-		// SINGLEDEL g#50 in L1
-		// SET g#40 in L2
-		// RANGEDEL [g,h)#30 in L3
-		// SET g#20 in L6
-		//
-		// In this example, the two SETs represent the same user write, and the
-		// RANGEDELs are caused by the CockroachDB range being dropped. That is,
-		// the user wrote to g once, range was dropped, then added back, which
-		// caused the SET again, then at some point g was validly deleted using a
-		// SINGLEDEL, and then the range was dropped again. The older RANGEDEL can
-		// get fragmented due to compactions it has been part of. Say this L3 file
-		// containing the RANGEDEL is very narrow, while the L1, L2, L6 files are
-		// wider than the RANGEDEL in L0. Then the RANGEDEL in L3 can be dropped
-		// using a delete-only compaction, resulting in an LSM with state:
-		//
-		// RANGEDEL [a, z)#60 in L0
-		// SINGLEDEL g#50 in L1
-		// SET g#40 in L2
-		// SET g#20 in L6
-		//
-		// A multi-level compaction involving L1, L2, L6 will cause the invariant
-		// violation callback. This example doesn't need multi-level compactions:
-		// say there was a Pebble snapshot at g#21 preventing g#20 from being
-		// dropped when it meets g#40 in a compaction. That snapshot will not save
-		// RANGEDEL [g,h)#30, so we can have:
-		//
-		// SINGLEDEL g#50 in L1
-		// SET g#40, SET g#20 in L6
-		//
-		// And say the snapshot is removed and then the L1 and L6 compaction
-		// happens, resulting in the invariant violation callback.
-		//
-		// TODO(sumeer): rename SingleDeleteInvariantViolationCallback to remove
-		// the word "invariant".
-
-		// IneffectualPointDeleteCallback is called in compactions/flushes if any
-		// single delete is being elided without deleting a point set/merge.
-		IneffectualSingleDeleteCallback func(userKey []byte)
-
-		// SingleDeleteInvariantViolationCallback is called in compactions/flushes if any
-		// single delete has consumed a Set/Merge, and there is another immediately older
-		// Set/SetWithDelete/Merge. The user of Pebble has violated the invariant under
-		// which SingleDelete can be used correctly.
-		//
-		// Consider the sequence SingleDelete#3, Set#2, Set#1. There are three
-		// ways some of these keys can first meet in a compaction.
-		//
-		// - All 3 keys in the same compaction: this callback will detect the
-		//   violation.
-		//
-		// - SingleDelete#3, Set#2 meet in a compaction first: Both keys will
-		//   disappear. The violation will not be detected, and the DB will have
-		//   Set#1 which is likely incorrect (from the user's perspective).
-		//
-		// - Set#2, Set#1 meet in a compaction first: The output will be Set#2,
-		//   which will later be consumed by SingleDelete#3. The violation will
-		//   not be detected and the DB will be correct.
-		SingleDeleteInvariantViolationCallback func(userKey []byte)
-
 		// EnableDeleteOnlyCompactionExcises enables delete-only compactions to also
 		// apply delete-only compaction hints on sstables that partially overlap
 		// with it. This application happens through an excise, similar to
@@ -859,6 +778,8 @@ type Options struct {
 		// no limiting or pacing happens other than that controlled by other options
 		// like L0CompactionConcurrency and CompactionDebtConcurrency.
 		CompactionLimiter CompactionLimiter
+
+		UserKeyCategories UserKeyCategories
 	}
 
 	// Filters is a map from filter policy name to filter policy. It is used for
@@ -2200,3 +2121,99 @@ func resolveDefaultCompression(c Compression) Compression {
 	}
 	return c
 }
+
+// UserKeyCategories describes a partitioning of the user key space. Each
+// partition is a category with a name. The categories are used for informative
+// purposes only (like pprof labels). Pebble does not treat keys differently
+// based on the UserKeyCategories.
+//
+// The partitions are defined by their upper bounds. The last partition is
+// assumed to go until the end of keyspace; its UpperBound is ignored. The rest
+// of the partitions are ordered by their UpperBound.
+type UserKeyCategories struct {
+	categories []UserKeyCategory
+	cmp        base.Compare
+	// rangeNames[i][j] contains the string referring to the categories in the
+	// range [i, j], with j > i.
+	rangeNames [][]string
+}
+
+// UserKeyCategory describes a partition of the user key space.
+//
+// User keys >= the previous category's UpperBound and < this category's
+// UpperBound are part of this category.
+type UserKeyCategory struct {
+	Name string
+	// UpperBound is the exclusive upper bound of the category. All user keys >= the
+	// previous category's UpperBound and < this UpperBound are part of this
+	// category.
+	UpperBound []byte
+}
+
+// MakeUserKeyCategories creates a UserKeyCategories object with the given
+// categories. The object is immutable and can be reused across different
+// stores.
+func MakeUserKeyCategories(cmp base.Compare, categories ...UserKeyCategory) UserKeyCategories {
+	n := len(categories)
+	if n == 0 {
+		return UserKeyCategories{}
+	}
+	if categories[n-1].UpperBound != nil {
+		panic("last category UpperBound must be nil")
+	}
+	// Verify that the partitions are ordered as expected.
+	for i := 1; i < n-1; i++ {
+		if cmp(categories[i-1].UpperBound, categories[i].UpperBound) >= 0 {
+			panic("invalid UserKeyCategories: key prefixes must be sorted")
+		}
+	}
+
+	// Precalculate a table of range names to avoid allocations in the
+	// categorization path.
+	rangeNamesBuf := make([]string, n*n)
+	rangeNames := make([][]string, n)
+	for i := range rangeNames {
+		rangeNames[i] = rangeNamesBuf[:n]
+		rangeNamesBuf = rangeNamesBuf[n:]
+		for j := i + 1; j < n; j++ {
+			rangeNames[i][j] = categories[i].Name + "-" + categories[j].Name
+		}
+	}
+	return UserKeyCategories{
+		categories: categories,
+		cmp:        cmp,
+		rangeNames: rangeNames,
+	}
+}
+
+// Len returns the number of categories defined.
+func (kc *UserKeyCategories) Len() int {
+	return len(kc.categories)
+}
+
+// CategorizeKey returns the name of the category containing the key.
+func (kc *UserKeyCategories) CategorizeKey(userKey []byte) string {
+	idx := sort.Search(len(kc.categories)-1, func(i int) bool {
+		return kc.cmp(userKey, kc.categories[i].UpperBound) < 0
+	})
+	return kc.categories[idx].Name
+}
+
+// CategorizeKeyRange returns the name of the category containing the key range.
+// If the key range spans multiple categories, the result shows the first and
+// last category separated by a dash, e.g. `cat1-cat5`.
+func (kc *UserKeyCategories) CategorizeKeyRange(startUserKey, endUserKey []byte) string {
+	n := len(kc.categories)
+	p := sort.Search(n-1, func(i int) bool {
+		return kc.cmp(startUserKey, kc.categories[i].UpperBound) < 0
+	})
+	if p == n-1 || kc.cmp(endUserKey, kc.categories[p].UpperBound) < 0 {
+		// Fast path for a single category.
+		return kc.categories[p].Name
+	}
+	// Binary search among the remaining categories.
+	q := p + 1 + sort.Search(n-2-p, func(i int) bool {
+		return kc.cmp(endUserKey, kc.categories[p+1+i].UpperBound) < 0
+	})
+	return kc.rangeNames[p][q]
+}
diff --git a/sstable/blob/blob.go b/sstable/blob/blob.go
new file mode 100644
index 00000000..bcd1192b
--- /dev/null
+++ b/sstable/blob/blob.go
@@ -0,0 +1,365 @@
+// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package blob
+
+import (
+	"context"
+	"encoding/binary"
+	"sync"
+
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/sstable/valblk"
+	"github.com/cockroachdb/redact"
+)
+
+var (
+	errClosed = errors.New("blob: writer closed")
+)
+
+// FileFormat identifies the format of a blob file.
+type FileFormat uint8
+
+// String implements the fmt.Stringer interface.
+func (f FileFormat) String() string {
+	switch f {
+	case FileFormatV1:
+		return "blobV1"
+	default:
+		return "unknown"
+	}
+}
+
+const (
+	// FileFormatV1 is the first version of the blob file format.
+	FileFormatV1 FileFormat = 1
+)
+
+const (
+	fileFooterLength = 29
+	fileMagic        = "\xf0\x9f\xaa\xb3\xf0\x9f\xa6\x80" // 🪳🦀
+)
+
+// FileWriterOptions are used to configure the FileWriter.
+type FileWriterOptions struct {
+	Compression   block.Compression
+	ChecksumType  block.ChecksumType
+	FlushGovernor block.FlushGovernor
+}
+
+// FileWriterStats aggregates statistics about a blob file written by a
+// FileWriter.
+type FileWriterStats struct {
+	BlockCount             uint32
+	ValueCount             uint32
+	BlockLenLongest        uint64
+	UncompressedValueBytes uint64
+	FileLen                uint64
+}
+
+// Handle describes the location of a value stored within a blob file.
+type Handle struct {
+	FileNum       base.DiskFileNum
+	BlockNum      uint32
+	OffsetInBlock uint32
+	ValueLen      uint32
+}
+
+// String implements the fmt.Stringer interface.
+func (h Handle) String() string {
+	return redact.StringWithoutMarkers(h)
+}
+
+// SafeFormat implements redact.SafeFormatter.
+func (h Handle) SafeFormat(w redact.SafePrinter, _ rune) {
+	w.Printf("(%s,blk%d[%d:%d])",
+		h.FileNum, h.BlockNum, h.OffsetInBlock, h.OffsetInBlock+h.ValueLen)
+}
+
+// A FileWriter writes a blob file.
+type FileWriter struct {
+	fileNum      base.DiskFileNum
+	w            objstorage.Writable
+	b            block.Buffer
+	stats        FileWriterStats
+	flushGov     block.FlushGovernor
+	blockOffsets []uint64
+	err          error
+	checksumType block.ChecksumType
+	writeQueue   struct {
+		wg  sync.WaitGroup
+		ch  chan compressedBlock
+		err error
+	}
+}
+
+type compressedBlock struct {
+	pb block.PhysicalBlock
+	bh *block.BufHandle
+}
+
+// NewFileWriter creates a new FileWriter.
+func NewFileWriter(fn base.DiskFileNum, w objstorage.Writable, opts FileWriterOptions) *FileWriter {
+	fw := writerPool.Get().(*FileWriter)
+	fw.fileNum = fn
+	fw.w = w
+	fw.b.Init(opts.Compression, opts.ChecksumType)
+	fw.flushGov = opts.FlushGovernor
+	fw.checksumType = opts.ChecksumType
+	fw.writeQueue.ch = make(chan compressedBlock)
+	fw.writeQueue.wg.Add(1)
+	go fw.drainWriteQueue()
+	return fw
+}
+
+var writerPool = sync.Pool{
+	New: func() interface{} { return &FileWriter{} },
+}
+
+// AddValue adds the provided value to the blob file, returning a Handle
+// identifying the location of the value.
+func (w *FileWriter) AddValue(v []byte) Handle {
+	// Determine if we should first flush the block.
+	if sz := w.b.Size(); sz != 0 && w.flushGov.ShouldFlush(sz, sz+len(v)) {
+		w.flush()
+	}
+	w.stats.ValueCount++
+	w.stats.UncompressedValueBytes += uint64(len(v))
+	off := uint32(w.b.Append(v))
+	return Handle{
+		FileNum:       w.fileNum,
+		BlockNum:      uint32(w.stats.BlockCount),
+		OffsetInBlock: off,
+		ValueLen:      uint32(len(v)),
+	}
+}
+
+func (w *FileWriter) flush() {
+	pb, bh := w.b.CompressAndChecksum()
+	compressedLen := uint64(pb.LengthWithTrailer())
+	w.stats.BlockCount++
+	w.stats.BlockLenLongest = max(w.stats.BlockLenLongest, compressedLen)
+	w.stats.FileLen += compressedLen
+	w.writeQueue.ch <- compressedBlock{pb: pb, bh: bh}
+}
+
+// drainWriteQueue runs in its own goroutine and is responsible for writing
+// finished, compressed data blocks to the writable. It reads from w.writeQueue
+// until the channel is closed. All value blocks are written by this goroutine.
+func (w *FileWriter) drainWriteQueue() {
+	defer w.writeQueue.wg.Done()
+	for cb := range w.writeQueue.ch {
+		n, err := cb.pb.WriteTo(w.w)
+		if err != nil {
+			w.writeQueue.err = err
+			continue
+		}
+		w.blockOffsets = append(w.blockOffsets, uint64(n))
+		// We're done with the buffer associated with this physical block.
+		// Release it back to its pool.
+		cb.bh.Release()
+	}
+}
+
+// Close finishes writing the blob file.
+func (w *FileWriter) Close() (FileWriterStats, error) {
+	if w.w == nil {
+		return FileWriterStats{}, w.err
+	}
+	defer func() {
+		if w.w != nil {
+			w.w.Abort()
+			w.w = nil
+		}
+		if w.err == nil {
+			w.err = errClosed
+		}
+	}()
+	// Flush the last block to the write queue if it's non-empty.
+	if w.b.Size() > 0 {
+		w.flush()
+	}
+	// Inform the write queue we're finished by closing the channel and wait
+	// for it to complete.
+	close(w.writeQueue.ch)
+	w.writeQueue.wg.Wait()
+	if w.writeQueue.err != nil {
+		w.err = w.writeQueue.err
+		return FileWriterStats{}, w.err
+	}
+	stats := w.stats
+	if stats.BlockCount != uint32(len(w.blockOffsets)) {
+		panic(errors.AssertionFailedf("block count mismatch: %d vs %d",
+			stats.BlockCount, len(w.blockOffsets)))
+	}
+
+	// Write the index block.
+	vbih := valblk.IndexHandle{
+		Handle:                block.Handle{Offset: stats.FileLen},
+		BlockNumByteLength:    uint8(lenLittleEndian(uint64(stats.BlockCount - 1))),
+		BlockOffsetByteLength: uint8(lenLittleEndian(w.blockOffsets[len(w.blockOffsets)-1])),
+		BlockLengthByteLength: uint8(lenLittleEndian(stats.BlockLenLongest)),
+	}
+	indexBlockLen := vbih.RowWidth() * int(stats.BlockCount)
+	vbih.Handle.Length = uint64(indexBlockLen) + block.TrailerLen
+	{
+		w.b.Resize(indexBlockLen)
+		b := w.b.Get()
+		for i := 0; i < len(w.blockOffsets); i++ {
+			littleEndianPut(uint64(i), b, int(vbih.BlockNumByteLength))
+			b = b[int(vbih.BlockNumByteLength):]
+			littleEndianPut(w.blockOffsets[i], b, int(vbih.BlockOffsetByteLength))
+			b = b[int(vbih.BlockOffsetByteLength):]
+			var l uint64
+			if i == len(w.blockOffsets)-1 {
+				l = stats.FileLen - w.blockOffsets[i]
+			} else {
+				l = w.blockOffsets[i+1] - w.blockOffsets[i]
+			}
+			littleEndianPut(l, b, int(vbih.BlockLengthByteLength))
+			b = b[int(vbih.BlockLengthByteLength):]
+		}
+		if len(b) != 0 {
+			panic(errors.AssertionFailedf("incorrect length calculation: buffer has %d bytes remaining of expected %d byte block",
+				len(b), indexBlockLen))
+		}
+		w.b.SetCompression(block.NoCompression)
+		pb, bh := w.b.CompressAndChecksum()
+		if _, w.err = pb.WriteTo(w.w); w.err != nil {
+			return FileWriterStats{}, w.err
+		}
+		bh.Release()
+		stats.FileLen += vbih.Handle.Length
+	}
+
+	// Write the footer.
+	footer := fileFooter{
+		format:      FileFormatV1,
+		checksum:    w.checksumType,
+		indexHandle: vbih,
+	}
+	w.b.Resize(fileFooterLength)
+	footerBuf := w.b.Get()
+	footer.encode(footerBuf)
+	if w.err = w.w.Write(footerBuf); w.err != nil {
+		return FileWriterStats{}, w.err
+	}
+	stats.FileLen += fileFooterLength
+	if w.err = w.w.Finish(); w.err != nil {
+		return FileWriterStats{}, w.err
+	}
+
+	// Clean up w and return it to the pool.
+	w.b.Release()
+	blockOffsets := w.blockOffsets[:0]
+	*w = FileWriter{}
+	w.blockOffsets = blockOffsets
+	writerPool.Put(w)
+	return stats, nil
+}
+
+// fileFooter contains the information contained within the footer of a blob
+// file.
+//
+// Blob file footer format:
+//   - index block offset (8 bytes)
+//   - index block length (8 bytes)
+//   - index block block-number byte length (1 byte)
+//   - index block block-offset byte length (1 byte)
+//   - index block block-length byte length (1 byte)
+//   - checksum type (1 byte)
+//   - format (1 byte)
+//   - blob file magic string (8 bytes)
+type fileFooter struct {
+	format      FileFormat
+	checksum    block.ChecksumType
+	indexHandle valblk.IndexHandle
+}
+
+func (f *fileFooter) decode(b []byte) error {
+	if uint64(len(b)) != fileFooterLength {
+		return errors.AssertionFailedf("invalid blob file footer length")
+	}
+	f.indexHandle.Handle.Offset = binary.LittleEndian.Uint64(b[0:])
+	f.indexHandle.Handle.Length = binary.LittleEndian.Uint64(b[8:])
+	f.indexHandle.BlockNumByteLength = b[16]
+	f.indexHandle.BlockOffsetByteLength = b[17]
+	f.indexHandle.BlockLengthByteLength = b[18]
+	if f.indexHandle.BlockNumByteLength > 4 {
+		return base.CorruptionErrorf("invalid block num byte length %d", f.indexHandle.BlockNumByteLength)
+	}
+	if f.indexHandle.BlockOffsetByteLength > 8 {
+		return base.CorruptionErrorf("invalid block offset byte length %d", f.indexHandle.BlockOffsetByteLength)
+	}
+	if f.indexHandle.BlockLengthByteLength > 8 {
+		return base.CorruptionErrorf("invalid block length byte length %d", f.indexHandle.BlockLengthByteLength)
+	}
+
+	f.checksum = block.ChecksumType(b[19])
+	f.format = FileFormat(b[20])
+	if f.format != FileFormatV1 {
+		return base.CorruptionErrorf("invalid blob file format %x", f.format)
+	}
+	if string(b[21:]) != fileMagic {
+		return base.CorruptionErrorf("invalid blob file magic string %x", b[21:])
+	}
+	return nil
+}
+
+func (f *fileFooter) encode(b []byte) {
+	binary.LittleEndian.PutUint64(b[0:], f.indexHandle.Handle.Offset)
+	binary.LittleEndian.PutUint64(b[8:], f.indexHandle.Handle.Length)
+	b[16] = f.indexHandle.BlockNumByteLength
+	b[17] = f.indexHandle.BlockOffsetByteLength
+	b[18] = f.indexHandle.BlockLengthByteLength
+	b[19] = byte(f.checksum)
+	b[20] = byte(f.format)
+	copy(b[21:], fileMagic)
+}
+
+// FileReader reads a blob file.
+type FileReader struct {
+	r      objstorage.Readable
+	footer fileFooter
+}
+
+// NewFileReader opens a blob file for reading.
+func NewFileReader(ctx context.Context, r objstorage.Readable) (*FileReader, error) {
+	var footerBuf [fileFooterLength]byte
+	if err := r.ReadAt(ctx, footerBuf[:], r.Size()-fileFooterLength); err != nil {
+		return nil, err
+	}
+	fr := &FileReader{r: r}
+	if err := fr.footer.decode(footerBuf[:]); err != nil {
+		return nil, err
+	}
+	return fr, nil
+}
+
+// lenLittleEndian returns the minimum number of bytes needed to encode v
+// using little endian encoding.
+func lenLittleEndian(v uint64) int {
+	n := 0
+	for i := 0; i < 8; i++ {
+		n++
+		v = v >> 8
+		if v == 0 {
+			break
+		}
+	}
+	return n
+}
+
+// littleEndianPut writes v to b using little endian encoding, under the
+// assumption that v can be represented using n bytes.
+func littleEndianPut(v uint64, b []byte, n int) {
+	_ = b[n-1] // bounds check
+	for i := 0; i < n; i++ {
+		b[i] = byte(v)
+		v = v >> 8
+	}
+}
diff --git a/sstable/block/compression.go b/sstable/block/compression.go
index 751aae26..c7bb8a4c 100644
--- a/sstable/block/compression.go
+++ b/sstable/block/compression.go
@@ -6,10 +6,14 @@ package block
 
 import (
 	"encoding/binary"
+	"math/rand/v2"
+	"slices"
+	"sync"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
+	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/golang/snappy"
 )
@@ -289,3 +293,167 @@ func compress(
 		panic("unreachable")
 	}
 }
+
+// A Buffer is a buffer for encoding a block. The caller mutates the buffer to
+// construct the uncompressed block, and calls CompressAndChecksum to produce
+// the physical, possibly-compressed PhysicalBlock. A Buffer recycles byte
+// slices used in construction of the uncompressed block and the compressed
+// physical block.
+type Buffer struct {
+	h           *BufHandle
+	compression Compression
+	checksummer Checksummer
+}
+
+// Init configures the BlockBuffer with the specified compression and checksum
+// type.
+func (b *Buffer) Init(compression Compression, checksumType ChecksumType) {
+	b.h = uncompressedBuffers.Get()
+	b.h.b = b.h.b[:0]
+	b.compression = compression
+	b.checksummer.Type = checksumType
+}
+
+// Get returns the byte slice currently backing the Buffer.
+func (b *Buffer) Get() []byte {
+	return b.h.b
+}
+
+// Size returns the current size of the buffer.
+func (b *Buffer) Size() int {
+	return len(b.h.b)
+}
+
+// Append appends the contents of v to the buffer, returning the offset at which
+// it was appended, growing the buffer if necessary.
+func (b *Buffer) Append(v []byte) int {
+	// We may need to grow b.Buffer to accommodate the new value. If necessary,
+	// double the size of the buffer until it's sufficiently large.
+	off := len(b.h.b)
+	newLen := off + len(v)
+	if cap(b.h.b) < newLen {
+		size := max(2*cap(b.h.b), 1024)
+		for size < newLen {
+			size *= 2
+		}
+		b.h.b = slices.Grow(b.h.b, size-cap(b.h.b))
+	}
+	b.h.b = b.h.b[:newLen]
+	if n := copy(b.h.b[off:], v); n != len(v) {
+		panic("incorrect length computation")
+	}
+	return off
+}
+
+// Resize resizes the buffer to the specified length, allocating if necessary.
+func (b *Buffer) Resize(length int) {
+	if length > cap(b.h.b) {
+		b.h.b = slices.Grow(b.h.b, length-len(b.h.b))
+	}
+	b.h.b = b.h.b[:length]
+}
+
+// CompressAndChecksum compresses and checksums the block data, returning a
+// PhysicalBlock that is owned by the caller. The returned PhysicalBlock's
+// memory is backed by the returned BufHandle. If non-nil, the returned
+// BufHandle may be Released once the caller is done with the physical block to
+// recycle the block's underlying memory.
+//
+// When CompressAndChecksum returns, the callee has been reset and is ready to
+// be reused.
+func (b *Buffer) CompressAndChecksum() (PhysicalBlock, *BufHandle) {
+	// Grab a buffer to use as the destination for compression.
+	compressedBuf := compressedBuffers.Get()
+	pb := CompressAndChecksum(&compressedBuf.b, b.h.b, b.compression, &b.checksummer)
+	if pb.IsCompressed() {
+		// Compression was fruitful, and pb's data points into compressedBuf. We
+		// can reuse b.Buffer because we've copied the compressed data.
+		b.h.b = b.h.b[:0]
+		return pb, compressedBuf
+	}
+	// Compression was not fruitful, and pb's data points into b.h. The
+	// compressedBuf we retrieved from the pool isn't needed, but our b.h is.
+	// Use the compressedBuf as the new b.h.
+	pbHandle := b.h
+	b.h = compressedBuf
+	b.h.b = b.h.b[:0]
+	return pb, pbHandle
+}
+
+// SetCompression changes the compression algorithm used by CompressAndChecksum.
+func (b *Buffer) SetCompression(compression Compression) {
+	b.compression = compression
+}
+
+// Release may be called when a buffer will no longer be used. It releases to
+// pools any memory held by the Buffer so that it may be reused.
+func (b *Buffer) Release() {
+	if b.h != nil {
+		b.h.Release()
+		b.h = nil
+	}
+}
+
+// BufHandle is a handle to a buffer that can be released to a pool for reuse.
+type BufHandle struct {
+	pool *bufferSyncPool
+	b    []byte
+}
+
+// Release releases the buffer back to the pool for reuse.
+func (h *BufHandle) Release() {
+	if invariants.Enabled && (h.pool == nil) != (h.b == nil) {
+		panic(errors.AssertionFailedf("pool (%t) and buffer (%t) nilness disagree", h.pool == nil, h.b == nil))
+	}
+	if invariants.Enabled && (h.pool != nil && h.pool.Max == 0) {
+		panic(errors.AssertionFailedf("pool has no maximum size"))
+	}
+	// Note we avoid releasing buffers that are larger than the configured
+	// maximum to the pool. This avoids holding on to occassional large buffers
+	// necesary for, for example, single large values.
+	if h.b != nil && len(h.b) < h.pool.Max {
+		if invariants.Enabled {
+			// Set the bytes to a random value. Cap the number of bytes being
+			// randomized to prevent test timeouts.
+			l := min(cap(h.b), 1000)
+			h.b = h.b[:l:l]
+			for j := range h.b {
+				h.b[j] = byte(rand.Uint32())
+			}
+		}
+		h.pool.pool.Put(h)
+	}
+}
+
+var (
+	// uncompressedBuffers is a pool of buffers that were used to store
+	// uncompressed block data. These buffers should be sized right around the
+	// configured block size. If multiple Pebble engines are running in the same
+	// process, they all share a pool and the size of the buffers may vary.
+	uncompressedBuffers = bufferSyncPool{Max: 256 << 10, Default: 4 << 10}
+	// compressedBuffers is a pool of buffers that were used to store compressed
+	// block data. These buffers will vary significantly in size depending on
+	// the compressibility of the data.
+	compressedBuffers = bufferSyncPool{Max: 128 << 10, Default: 4 << 10}
+)
+
+// bufferSyncPool is a pool of block buffers for memory re-use.
+type bufferSyncPool struct {
+	Max     int
+	Default int
+	pool    sync.Pool
+}
+
+// Get retrieves a new buf from the pool, or allocates one of the configured
+// default size if the pool is empty.
+func (p *bufferSyncPool) Get() *BufHandle {
+	v := p.pool.Get()
+	if v != nil {
+		return v.(*BufHandle)
+	}
+	if invariants.Enabled && p.Default == 0 {
+		// Guard against accidentally forgetting to initialize a buffer sync pool.
+		panic(errors.AssertionFailedf("buffer pool has no default size"))
+	}
+	return &BufHandle{b: make([]byte, 0, p.Default), pool: p}
+}
diff --git a/sstable/layout.go b/sstable/layout.go
index e05700fb..8d5773e2 100644
--- a/sstable/layout.go
+++ b/sstable/layout.go
@@ -745,25 +745,15 @@ func (w *layoutWriter) WriteValueBlock(blk block.PhysicalBlock) (block.Handle, e
 }
 
 func (w *layoutWriter) WriteValueIndexBlock(
-	blk []byte, vbih valblk.IndexHandle,
+	blk block.PhysicalBlock, vbih valblk.IndexHandle,
 ) (block.Handle, error) {
-	// NB: value index blocks are already finished and contain the block
-	// trailer.
-	// TODO(jackson): can this be refactored to make value blocks less
-	// of a snowflake?
-	off := w.offset
-	w.clearFromCache(off)
-	// Write the bytes to the file.
-	if err := w.writable.Write(blk); err != nil {
+	h, err := w.writePrecompressedBlock(blk)
+	if err != nil {
 		return block.Handle{}, err
 	}
-	l := uint64(len(blk))
-	w.offset += l
-
 	n := valblk.EncodeIndexHandle(w.tmp[:], vbih)
 	w.recordToMetaindexRaw(metaValueIndexName, w.tmp[:n])
-
-	return block.Handle{Offset: off, Length: l}, nil
+	return h, nil
 }
 
 func (w *layoutWriter) writeBlock(
diff --git a/sstable/rowblk/rowblk_iter.go b/sstable/rowblk/rowblk_iter.go
index 2ea7097a..1a93c997 100644
--- a/sstable/rowblk/rowblk_iter.go
+++ b/sstable/rowblk/rowblk_iter.go
@@ -16,7 +16,6 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
-	"github.com/cockroachdb/pebble/internal/manual"
 	"github.com/cockroachdb/pebble/internal/treeprinter"
 	"github.com/cockroachdb/pebble/sstable/block"
 )
@@ -382,7 +381,7 @@ func (i *Iter) readEntry() {
 		ptr = unsafe.Pointer(uintptr(ptr) + 5)
 	}
 	shared += i.transforms.SyntheticPrefixAndSuffix.PrefixLen()
-	unsharedKey := getBytes(ptr, int(unshared))
+	unsharedKey := unsafe.Slice((*byte)(ptr), int(unshared))
 	// TODO(sumeer): move this into the else block below.
 	i.fullKey = append(i.fullKey[:shared], unsharedKey...)
 	if shared == 0 {
@@ -395,7 +394,7 @@ func (i *Iter) readEntry() {
 		i.key = i.fullKey
 	}
 	ptr = unsafe.Pointer(uintptr(ptr) + uintptr(unshared))
-	i.val = getBytes(ptr, int(value))
+	i.val = unsafe.Slice((*byte)(ptr), int(value))
 	i.nextOffset = int32(uintptr(ptr)-uintptr(i.ptr)) + int32(value)
 }
 
@@ -449,7 +448,7 @@ func (i *Iter) readFirstKey() error {
 		ptr = unsafe.Pointer(uintptr(ptr) + 5)
 	}
 
-	firstKey := getBytes(ptr, int(unshared))
+	firstKey := unsafe.Slice((*byte)(ptr), int(unshared))
 	// Manually inlining base.DecodeInternalKey provides a 5-10% speedup on
 	// BlockIter benchmarks.
 	if n := len(firstKey) - 8; n >= 0 {
@@ -611,7 +610,7 @@ func (i *Iter) SeekGE(key []byte, flags base.SeekGEFlags) *base.InternalKV {
 
 			// Manually inlining part of base.DecodeInternalKey provides a 5-10%
 			// speedup on BlockIter benchmarks.
-			s := getBytes(ptr, int(v1))
+			s := unsafe.Slice((*byte)(ptr), int(v1))
 			var k []byte
 			if n := len(s) - 8; n >= 0 {
 				k = s[:n:n]
@@ -793,7 +792,7 @@ func (i *Iter) SeekLT(key []byte, flags base.SeekLTFlags) *base.InternalKV {
 
 			// Manually inlining part of base.DecodeInternalKey provides a 5-10%
 			// speedup on BlockIter benchmarks.
-			s := getBytes(ptr, int(v1))
+			s := unsafe.Slice((*byte)(ptr), int(v1))
 			var k []byte
 			if n := len(s) - 8; n >= 0 {
 				k = s[:n:n]
@@ -1240,7 +1239,7 @@ func (i *Iter) nextPrefixV3(succKey []byte) *base.InternalKV {
 		}
 		// The trailer is written in little endian, so the key kind is the first
 		// byte in the trailer that is encoded in the slice [unshared-8:unshared].
-		keyKind := base.InternalKeyKind((*[manual.MaxArrayLen]byte)(ptr)[unshared-8])
+		keyKind := base.InternalKeyKind(*(*byte)(unsafe.Pointer(uintptr(ptr) + uintptr(unshared) - 8)))
 		keyKind = keyKind & base.InternalKeyKindSSTableInternalObsoleteMask
 		prefixChanged := false
 		if keyKind == base.InternalKeyKindSet {
@@ -1334,7 +1333,7 @@ func (i *Iter) nextPrefixV3(succKey []byte) *base.InternalKV {
 		// - (Unlikely) The prefix has not changed.
 		// We assemble the key etc. under the assumption that it is the likely
 		// case.
-		unsharedKey := getBytes(ptr, int(unshared))
+		unsharedKey := unsafe.Slice((*byte)(ptr), int(unshared))
 		// TODO(sumeer): move this into the else block below. This is a bit tricky
 		// since the current logic assumes we have always copied the latest key
 		// into fullKey, which is why when we get to the next key we can (a)
@@ -1346,7 +1345,7 @@ func (i *Iter) nextPrefixV3(succKey []byte) *base.InternalKV {
 		// too. This same comment applies to the other place where we can do this
 		// optimization, in readEntry().
 		i.fullKey = append(i.fullKey[:shared], unsharedKey...)
-		i.val = getBytes(valuePtr, int(value))
+		i.val = unsafe.Slice((*byte)(valuePtr), int(value))
 		if shared == 0 {
 			// Provide stability for the key across positioning calls if the key
 			// doesn't share a prefix with the previous key. This removes requiring the
@@ -1419,7 +1418,7 @@ start:
 		i.nextOffset = i.offset
 		e := &i.cached[n]
 		i.offset = e.offset
-		i.val = getBytes(unsafe.Pointer(uintptr(i.ptr)+uintptr(e.valStart)), int(e.valSize))
+		i.val = unsafe.Slice((*byte)(unsafe.Pointer(uintptr(i.ptr)+uintptr(e.valStart))), int(e.valSize))
 		// Manually inlined version of i.decodeInternalKey(i.key).
 		i.key = i.cachedBuf[e.keyStart:e.keyEnd]
 		if n := len(i.key) - 8; n >= 0 {
@@ -1713,10 +1712,10 @@ func (i *RawIter) readEntry() {
 	shared, ptr := decodeVarint(ptr)
 	unshared, ptr := decodeVarint(ptr)
 	value, ptr := decodeVarint(ptr)
-	i.key = append(i.key[:shared], getBytes(ptr, int(unshared))...)
+	i.key = append(i.key[:shared], unsafe.Slice((*byte)(ptr), int(unshared))...)
 	i.key = i.key[:len(i.key):len(i.key)]
 	ptr = unsafe.Pointer(uintptr(ptr) + uintptr(unshared))
-	i.val = getBytes(ptr, int(value))
+	i.val = unsafe.Slice((*byte)(ptr), int(value))
 	i.nextOffset = int32(uintptr(ptr)-uintptr(i.ptr)) + int32(value)
 }
 
@@ -1761,7 +1760,7 @@ func (i *RawIter) SeekGE(key []byte) bool {
 		// Decode the key at that restart point, and compare it to the key sought.
 		v1, ptr := decodeVarint(ptr)
 		_, ptr = decodeVarint(ptr)
-		s := getBytes(ptr, int(v1))
+		s := unsafe.Slice((*byte)(ptr), int(v1))
 		return i.cmp(key, s) < 0
 	})
 
@@ -1828,7 +1827,7 @@ func (i *RawIter) Prev() bool {
 		i.nextOffset = i.offset
 		e := &i.cached[n-1]
 		i.offset = e.offset
-		i.val = getBytes(unsafe.Pointer(uintptr(i.ptr)+uintptr(e.valStart)), int(e.valSize))
+		i.val = unsafe.Slice((*byte)(unsafe.Pointer(uintptr(i.ptr)+uintptr(e.valStart))), int(e.valSize))
 		i.ikey.UserKey = i.cachedBuf[e.keyStart:e.keyEnd]
 		i.cached = i.cached[:n]
 		return true
@@ -1939,10 +1938,6 @@ func (i *RawIter) Describe(tp treeprinter.Node, fmtKV DescribeKV) {
 	}
 }
 
-func getBytes(ptr unsafe.Pointer, length int) []byte {
-	return (*[manual.MaxArrayLen]byte)(ptr)[:length:length]
-}
-
 func decodeVarint(ptr unsafe.Pointer) (uint32, unsafe.Pointer) {
 	if a := *((*uint8)(ptr)); a < 128 {
 		return uint32(a),
diff --git a/sstable/valblk/valblk.go b/sstable/valblk/valblk.go
index 2021e582..d6463f12 100644
--- a/sstable/valblk/valblk.go
+++ b/sstable/valblk/valblk.go
@@ -168,6 +168,7 @@ package valblk
 
 import (
 	"encoding/binary"
+	"fmt"
 	"unsafe"
 
 	"github.com/cockroachdb/errors"
@@ -291,6 +292,18 @@ type IndexHandle struct {
 	BlockLengthByteLength uint8
 }
 
+// String returns a string representation of the IndexHandle.
+func (h IndexHandle) String() string {
+	return fmt.Sprintf("{Handle: {%d,%d}, DataLens:(%d,%d,%d)}",
+		h.Handle.Offset, h.Handle.Length, h.BlockNumByteLength, h.BlockOffsetByteLength, h.BlockLengthByteLength)
+}
+
+// RowWidth returns the number of bytes that the referenced index block uses per
+// value block.
+func (h IndexHandle) RowWidth() int {
+	return int(h.BlockNumByteLength + h.BlockOffsetByteLength + h.BlockLengthByteLength)
+}
+
 // EncodeIndexHandle encodes the IndexHandle into dst and returns the number of
 // bytes written.
 func EncodeIndexHandle(dst []byte, v IndexHandle) int {
diff --git a/sstable/valblk/writer.go b/sstable/valblk/writer.go
index 892ac9ef..e3f55f08 100644
--- a/sstable/valblk/writer.go
+++ b/sstable/valblk/writer.go
@@ -5,8 +5,6 @@
 package valblk
 
 import (
-	"encoding/binary"
-	"math/rand/v2"
 	"sync"
 
 	"github.com/cockroachdb/errors"
@@ -18,17 +16,11 @@ import (
 // sstable.
 type Writer struct {
 	flush block.FlushGovernor
-	// Configured compression.
-	compression block.Compression
-	// checksummer with configured checksum type.
-	checksummer block.Checksummer
 	// Block finished callback.
 	blockFinishedFunc func(compressedSize int)
 
 	// buf is the current block being written to (uncompressed).
-	buf *blockBuffer
-	// compressedBuf is used for compressing the block.
-	compressedBuf *blockBuffer
+	buf block.Buffer
 	// Sequence of blocks that are finished.
 	blocks []bufferedValueBlock
 	// Cumulative value block bytes written so far.
@@ -37,49 +29,9 @@ type Writer struct {
 }
 
 type bufferedValueBlock struct {
-	block  block.PhysicalBlock
-	buffer *blockBuffer
-	handle block.Handle
-}
-
-type blockBuffer struct {
-	b []byte
-}
-
-// Pool of block buffers that should be roughly the blockSize.
-var uncompressedValueBlockBufPool = sync.Pool{
-	New: func() interface{} {
-		return &blockBuffer{}
-	},
-}
-
-// Pool of block buffers for compressed value blocks. These may widely vary in
-// size based on compression ratios.
-var compressedValueBlockBufPool = sync.Pool{
-	New: func() interface{} {
-		return &blockBuffer{}
-	},
-}
-
-func releaseToValueBlockBufPool(pool *sync.Pool, b *blockBuffer) {
-	// Don't pool buffers larger than 128KB, in case we had some rare large
-	// values.
-	if len(b.b) > 128*1024 {
-		return
-	}
-	if invariants.Enabled {
-		// Set the bytes to a random value. Cap the number of bytes being
-		// randomized to prevent test timeouts.
-		length := cap(b.b)
-		if length > 1000 {
-			length = 1000
-		}
-		b.b = b.b[:length:length]
-		for j := range b.b {
-			b.b[j] = byte(rand.Uint32())
-		}
-	}
-	pool.Put(b)
+	block     block.PhysicalBlock
+	bufHandle *block.BufHandle
+	handle    block.Handle
 }
 
 var valueBlockWriterPool = sync.Pool{
@@ -98,18 +50,11 @@ func NewWriter(
 ) *Writer {
 	w := valueBlockWriterPool.Get().(*Writer)
 	*w = Writer{
-		flush:       flushGovernor,
-		compression: compression,
-		checksummer: block.Checksummer{
-			Type: checksumType,
-		},
+		flush:             flushGovernor,
 		blockFinishedFunc: blockFinishedFunc,
-		buf:               uncompressedValueBlockBufPool.Get().(*blockBuffer),
-		compressedBuf:     compressedValueBlockBufPool.Get().(*blockBuffer),
 		blocks:            w.blocks[:0],
 	}
-	w.buf.b = w.buf.b[:0]
-	w.compressedBuf.b = w.compressedBuf.b[:0]
+	w.buf.Init(compression, checksumType)
 	return w
 }
 
@@ -120,89 +65,44 @@ func (w *Writer) AddValue(v []byte) (Handle, error) {
 		return Handle{}, errors.Errorf("cannot write empty value to value block")
 	}
 	w.numValues++
-	blockLen := len(w.buf.b)
-	valueLen := len(v)
-	if w.flush.ShouldFlush(blockLen, blockLen+valueLen) {
-		// Block is not currently empty and adding this value will become too big,
-		// so finish this block.
+	if blockLen := w.buf.Size(); w.flush.ShouldFlush(blockLen, blockLen+len(v)) {
+		// Block is not currently empty and adding this value will become too
+		// big, so finish this block.
 		w.compressAndFlush()
-		blockLen = len(w.buf.b)
-		if invariants.Enabled && blockLen != 0 {
-			panic("blockLen of new block should be 0")
+		if invariants.Enabled && w.buf.Size() != 0 {
+			panic("buffer should be empty when starting new block")
 		}
 	}
 	vh := Handle{
-		ValueLen:      uint32(valueLen),
-		BlockNum:      uint32(len(w.blocks)),
-		OffsetInBlock: uint32(blockLen),
-	}
-	blockLen = int(vh.OffsetInBlock + vh.ValueLen)
-	if cap(w.buf.b) < blockLen {
-		size := 2 * cap(w.buf.b)
-		if size < 1024 {
-			size = 1024
-		}
-		for size < blockLen {
-			size *= 2
-		}
-		buf := make([]byte, blockLen, size)
-		_ = copy(buf, w.buf.b)
-		w.buf.b = buf
-	} else {
-		w.buf.b = w.buf.b[:blockLen]
-	}
-	buf := w.buf.b[vh.OffsetInBlock:]
-	n := copy(buf, v)
-	if n != len(buf) {
-		panic("incorrect length computation")
+		ValueLen: uint32(len(v)),
+		BlockNum: uint32(len(w.blocks)),
 	}
+	vh.OffsetInBlock = uint32(w.buf.Append(v))
 	return vh, nil
 }
 
 // Size returns the total size of currently buffered value blocks.
 func (w *Writer) Size() uint64 {
-	sz := w.totalBlockBytes
-	if w.buf != nil {
-		sz += uint64(len(w.buf.b))
-	}
-	return sz
+	return w.totalBlockBytes + uint64(w.buf.Size())
 }
 
 func (w *Writer) compressAndFlush() {
-	w.compressedBuf.b = w.compressedBuf.b[:cap(w.compressedBuf.b)]
-	physicalBlock := block.CompressAndChecksum(&w.compressedBuf.b, w.buf.b, w.compression, &w.checksummer)
+	physicalBlock, bufHandle := w.buf.CompressAndChecksum()
 	bh := block.Handle{Offset: w.totalBlockBytes, Length: uint64(physicalBlock.LengthWithoutTrailer())}
 	w.totalBlockBytes += uint64(physicalBlock.LengthWithTrailer())
 	// blockFinishedFunc length excludes the block trailer.
 	w.blockFinishedFunc(physicalBlock.LengthWithoutTrailer())
-	b := bufferedValueBlock{
-		block:  physicalBlock,
-		handle: bh,
-	}
-
-	// We'll hand off a buffer to w.blocks and get a new one. Which buffer we're
-	// handing off depends on the outcome of compression.
-	if physicalBlock.IsCompressed() {
-		b.buffer = w.compressedBuf
-		w.compressedBuf = compressedValueBlockBufPool.Get().(*blockBuffer)
-	} else {
-		b.buffer = w.buf
-		w.buf = uncompressedValueBlockBufPool.Get().(*blockBuffer)
-	}
-	w.blocks = append(w.blocks, b)
-	w.buf.b = w.buf.b[:0]
-}
-
-func (w *Writer) computeChecksum(blk []byte) {
-	n := len(blk) - block.TrailerLen
-	checksum := w.checksummer.Checksum(blk[:n], blk[n])
-	binary.LittleEndian.PutUint32(blk[n+1:], checksum)
+	w.blocks = append(w.blocks, bufferedValueBlock{
+		block:     physicalBlock,
+		bufHandle: bufHandle,
+		handle:    bh,
+	})
 }
 
 // Finish finishes writing the value blocks and value blocks index, writing the
 // buffered blocks out to the provider layout writer.
 func (w *Writer) Finish(layout LayoutWriter, fileOffset uint64) (IndexHandle, WriterStats, error) {
-	if len(w.buf.b) > 0 {
+	if w.buf.Size() > 0 {
 		w.compressAndFlush()
 	}
 	n := len(w.blocks)
@@ -247,18 +147,11 @@ func (w *Writer) Finish(layout LayoutWriter, fileOffset uint64) (IndexHandle, Wr
 }
 
 func (w *Writer) writeValueBlocksIndex(layout LayoutWriter, h IndexHandle) (IndexHandle, error) {
-	blockLen :=
-		int(h.BlockNumByteLength+h.BlockOffsetByteLength+h.BlockLengthByteLength) * len(w.blocks)
+	w.buf.SetCompression(block.NoCompression)
+	blockLen := h.RowWidth() * len(w.blocks)
 	h.Handle.Length = uint64(blockLen)
-	blockLen += block.TrailerLen
-	var buf []byte
-	if cap(w.buf.b) < blockLen {
-		buf = make([]byte, blockLen)
-		w.buf.b = buf
-	} else {
-		buf = w.buf.b[:blockLen]
-	}
-	b := buf
+	w.buf.Resize(blockLen)
+	b := w.buf.Get()
 	for i := range w.blocks {
 		littleEndianPut(uint64(i), b, int(h.BlockNumByteLength))
 		b = b[int(h.BlockNumByteLength):]
@@ -267,14 +160,14 @@ func (w *Writer) writeValueBlocksIndex(layout LayoutWriter, h IndexHandle) (Inde
 		littleEndianPut(w.blocks[i].handle.Length, b, int(h.BlockLengthByteLength))
 		b = b[int(h.BlockLengthByteLength):]
 	}
-	if len(b) != block.TrailerLen {
+	if len(b) != 0 {
 		panic("incorrect length calculation")
 	}
-	b[0] = byte(block.NoCompressionIndicator)
-	w.computeChecksum(buf)
-	if _, err := layout.WriteValueIndexBlock(buf, h); err != nil {
+	pb, bufHandle := w.buf.CompressAndChecksum()
+	if _, err := layout.WriteValueIndexBlock(pb, h); err != nil {
 		return IndexHandle{}, err
 	}
+	bufHandle.Release()
 	return h, nil
 }
 
@@ -282,22 +175,12 @@ func (w *Writer) writeValueBlocksIndex(layout LayoutWriter, h IndexHandle) (Inde
 // to a pool.
 func (w *Writer) Release() {
 	for i := range w.blocks {
-		if w.blocks[i].block.IsCompressed() {
-			releaseToValueBlockBufPool(&compressedValueBlockBufPool, w.blocks[i].buffer)
-		} else {
-			releaseToValueBlockBufPool(&uncompressedValueBlockBufPool, w.blocks[i].buffer)
-		}
-		w.blocks[i].buffer = nil
-	}
-	if w.buf != nil {
-		releaseToValueBlockBufPool(&uncompressedValueBlockBufPool, w.buf)
-	}
-	if w.compressedBuf != nil {
-		releaseToValueBlockBufPool(&compressedValueBlockBufPool, w.compressedBuf)
-	}
-	*w = Writer{
-		blocks: w.blocks[:0],
+		w.blocks[i].bufHandle.Release()
+		w.blocks[i] = bufferedValueBlock{}
 	}
+	w.buf.Release()
+	w.buf = block.Buffer{}
+	*w = Writer{blocks: w.blocks[:0]}
 	valueBlockWriterPool.Put(w)
 }
 
@@ -321,5 +204,5 @@ type LayoutWriter interface {
 	WriteValueBlock(blk block.PhysicalBlock) (block.Handle, error)
 	// WriteValueBlockIndex writes a pre-finished value block index to the
 	// writer.
-	WriteValueIndexBlock(blk []byte, vbih IndexHandle) (block.Handle, error)
+	WriteValueIndexBlock(blk block.PhysicalBlock, vbih IndexHandle) (block.Handle, error)
 }


Test patch:
diff --git a/cockroachkvs/cockroachkvs_test.go b/cockroachkvs/cockroachkvs_test.go
index 543f9289..7227a8ea 100644
--- a/cockroachkvs/cockroachkvs_test.go
+++ b/cockroachkvs/cockroachkvs_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/colblk"
 	"github.com/olekukonko/tablewriter"
+	testify "github.com/stretchr/testify/require"
 )
 
 func TestComparer(t *testing.T) {
@@ -703,3 +704,30 @@ func splitStringAt(str string, sep string) (before, after string) {
 	}
 	return str, ""
 }
+
+func TestFormatKey(t *testing.T) {
+	formattedKeys := []string{
+		"foo",                     // Bare key
+		"bar",                     // Bare key
+		"foo@1000000,0",           // MVCC key, only wall time in seconds
+		"foo@1000000.000051212,0", // MVCC key, wall time with subsecond nanos
+		"foo@1000000,8",           // MVCC key, with logical time
+		"foo@1000000.000051212,7", // MVCC key, wall time with subsecond nanos and logical time
+		"foo@01,cf379e85-6371-42e6-acc8-ed5259fa177a", // Lock table key
+	}
+	for _, formattedKey := range formattedKeys {
+		k := ParseFormattedKey(formattedKey)
+		t.Logf("ParseFormattedKey(%s) = %x", formattedKey, k)
+		reformatted := fmt.Sprint(FormatKey(k))
+		require.Equal(t, formattedKey, reformatted)
+	}
+
+	invalidFormattedKeys := []string{
+		"foo@01000000",                              // Missing logical
+		"foo@1000000.000051212",                     // Missing logical
+		"foo@01,cf379e85-6371-42e6-acc8-ed5259fa17", // Invalid UUID
+	}
+	for _, formattedKey := range invalidFormattedKeys {
+		testify.Panics(t, func() { ParseFormattedKey(formattedKey) })
+	}
+}
diff --git a/internal/compact/iterator_test.go b/internal/compact/iterator_test.go
index d4c85bf2..d114d7a2 100644
--- a/internal/compact/iterator_test.go
+++ b/internal/compact/iterator_test.go
@@ -84,7 +84,7 @@ func TestCompactionIter(t *testing.T) {
 			IneffectualSingleDeleteCallback: func(userKey []byte) {
 				ineffectualSingleDeleteKeys = append(ineffectualSingleDeleteKeys, string(userKey))
 			},
-			SingleDeleteInvariantViolationCallback: func(userKey []byte) {
+			NondeterministicSingleDeleteCallback: func(userKey []byte) {
 				invariantViolationSingleDeleteKeys = append(invariantViolationSingleDeleteKeys, string(userKey))
 			},
 		}
diff --git a/internal/randvar/skewed_latest.go b/internal/randvar/skewed_latest.go
index 787633d8..30bd05f9 100644
--- a/internal/randvar/skewed_latest.go
+++ b/internal/randvar/skewed_latest.go
@@ -52,7 +52,7 @@ func NewSkewedLatest(min, max uint64, theta float64) (*SkewedLatest, error) {
 }
 
 // IncMax increments max.
-func (z *SkewedLatest) IncMax(delta int) {
+func (z *SkewedLatest) IncMax(delta uint64) {
 	z.mu.Lock()
 	z.mu.zipf.IncMax(delta)
 	z.mu.max += uint64(delta)
diff --git a/metamorphic/generator_test.go b/metamorphic/generator_test.go
index d1b0dc37..c4b7bfc9 100644
--- a/metamorphic/generator_test.go
+++ b/metamorphic/generator_test.go
@@ -133,7 +133,8 @@ func TestGeneratorRandom(t *testing.T) {
 		rng := rand.New(rand.NewPCG(0, seed))
 		count := ops.Uint64(rng)
 		km := newKeyManager(cfg.numInstances)
-		return formatOps(km.kf, generate(rng, count, cfg, km))
+		g := newGenerator(rng, cfg, km)
+		return formatOps(km.kf, g.generate(count))
 	}
 
 	for i := range cfgs {
diff --git a/metamorphic/key_manager_test.go b/metamorphic/key_manager_test.go
index 1125a501..5bafee72 100644
--- a/metamorphic/key_manager_test.go
+++ b/metamorphic/key_manager_test.go
@@ -158,11 +158,13 @@ func TestLoadPrecedingKeys(t *testing.T) {
 	rng := randvar.NewRand()
 	cfg := DefaultOpConfig()
 	km := newKeyManager(1 /* numInstances */)
-	ops := generate(rng, 1000, cfg, km)
+	g := newGenerator(rng, cfg, km)
+	ops := g.generate(1000)
 
 	cfg2 := DefaultOpConfig()
 	km2 := newKeyManager(1 /* numInstances */)
-	loadPrecedingKeys(t, ops, &cfg2, km2)
+	g2 := newGenerator(rng, cfg2, km2)
+	loadPrecedingKeys(ops, g2.keyGenerator, km2)
 
 	// NB: We can't assert equality, because the original run may not have
 	// ever used the max of the distribution.
diff --git a/metamorphic/options_test.go b/metamorphic/options_test.go
index 9681e62a..d7362fdf 100644
--- a/metamorphic/options_test.go
+++ b/metamorphic/options_test.go
@@ -175,7 +175,8 @@ func TestBlockPropertiesParse(t *testing.T) {
 
 	rng := rand.New(rand.NewPCG(0, fixedSeed))
 	km := newKeyManager(1 /* numInstances */)
-	ops := generate(rng, numOps, presetConfigs[0], km)
+	g := newGenerator(rng, presetConfigs[0], km)
+	ops := g.generate(numOps)
 	opsPath := filepath.Join(metaDir, "ops")
 	formattedOps := formatOps(km.kf, ops)
 	require.NoError(t, os.WriteFile(opsPath, []byte(formattedOps), 0644))
diff --git a/metamorphic/parser_test.go b/metamorphic/parser_test.go
index 2978b77d..0bb07320 100644
--- a/metamorphic/parser_test.go
+++ b/metamorphic/parser_test.go
@@ -42,7 +42,8 @@ func TestParserRandom(t *testing.T) {
 				cfg.numInstances = 2
 			}
 			km := newKeyManager(cfg.numInstances)
-			ops := generate(randvar.NewRand(), 10000, cfg, km)
+			g := newGenerator(randvar.NewRand(), cfg, km)
+			ops := g.generate(10000)
 			src := formatOps(km.kf, ops)
 
 			parsedOps, err := parse([]byte(src), parserOpts{})
diff --git a/metamorphic/testkeys.go b/metamorphic/testkeys.go
index 27e75561..9fdc3d12 100644
--- a/metamorphic/testkeys.go
+++ b/metamorphic/testkeys.go
@@ -53,6 +53,24 @@ type testkeyKeyGenerator struct {
 	cfg        OpConfig
 }
 
+// RecordPrecedingKey may be invoked before generating keys to inform the key
+// generator of a key that was previously generated and used within a related
+// test context.
+func (kg *testkeyKeyGenerator) RecordPrecedingKey(key []byte) {
+	// If the key has a suffix that's larger than the current max suffix,
+	// ratchet up the maximum of the distribution of suffixes.
+	if i := testkeys.Comparer.Split(key); i < len(key) {
+		s, err := testkeys.ParseSuffix(key[i:])
+		if err != nil {
+			panic(err)
+		}
+		if uint64(s) > kg.cfg.writeSuffixDist.Max() {
+			diff := uint64(s) - kg.cfg.writeSuffixDist.Max()
+			kg.cfg.writeSuffixDist.IncMax(diff)
+		}
+	}
+}
+
 // RandKey returns a random key (either a previously known key, or a new key).
 func (kg *testkeyKeyGenerator) RandKey(newKeyProbability float64) []byte {
 	return kg.randKey(newKeyProbability, nil /* bounds */)
diff --git a/options_test.go b/options_test.go
index d840366c..3eaf4380 100644
--- a/options_test.go
+++ b/options_test.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"math/rand/v2"
 	"runtime"
+	"strings"
 	"testing"
 	"time"
 
@@ -394,3 +395,65 @@ func TestOptionsValidateCache(t *testing.T) {
 		t.Errorf("Unexpected error message")
 	}
 }
+
+func TestKeyCategories(t *testing.T) {
+	kc := MakeUserKeyCategories(base.DefaultComparer.Compare, []UserKeyCategory{
+		{Name: "b", UpperBound: []byte("b")},
+		{Name: "dd", UpperBound: []byte("dd")},
+		{Name: "e", UpperBound: []byte("e")},
+		{Name: "h", UpperBound: []byte("h")},
+		{Name: "o", UpperBound: nil},
+	}...)
+
+	for _, tc := range []struct {
+		key      string
+		expected string
+	}{
+		{key: "a", expected: "b"},
+		{key: "a123", expected: "b"},
+		{key: "az", expected: "b"},
+		{key: "b", expected: "dd"},
+		{key: "b123", expected: "dd"},
+		{key: "c", expected: "dd"},
+		{key: "d", expected: "dd"},
+		{key: "dd", expected: "e"},
+		{key: "dd0", expected: "e"},
+		{key: "de", expected: "e"},
+		{key: "e", expected: "h"},
+		{key: "e1", expected: "h"},
+		{key: "f", expected: "h"},
+		{key: "h", expected: "o"},
+		{key: "h1", expected: "o"},
+		{key: "z", expected: "o"},
+	} {
+		t.Run(tc.key, func(t *testing.T) {
+			res := kc.CategorizeKey([]byte(tc.key))
+			require.Equal(t, tc.expected, res)
+		})
+	}
+
+	for _, tc := range []struct {
+		rng      string
+		expected string
+	}{
+		{rng: "a-a", expected: "b"},
+		{rng: "a-a1", expected: "b"},
+		{rng: "a-b", expected: "b-dd"},
+		{rng: "b1-b5", expected: "dd"},
+		{rng: "b1-dd0", expected: "dd-e"},
+		{rng: "b1-g", expected: "dd-h"},
+		{rng: "b1-j", expected: "dd-o"},
+		{rng: "b1-z", expected: "dd-o"},
+		{rng: "dd-e", expected: "e-h"},
+		{rng: "h-i", expected: "o"},
+		{rng: "i-i", expected: "o"},
+		{rng: "i-j", expected: "o"},
+		{rng: "a-z", expected: "b-o"},
+	} {
+		t.Run(tc.rng, func(t *testing.T) {
+			keys := strings.SplitN(tc.rng, "-", 2)
+			res := kc.CategorizeKeyRange([]byte(keys[0]), []byte(keys[1]))
+			require.Equal(t, tc.expected, res)
+		})
+	}
+}
diff --git a/sstable/blob/blob_test.go b/sstable/blob/blob_test.go
new file mode 100644
index 00000000..c6c4f44a
--- /dev/null
+++ b/sstable/blob/blob_test.go
@@ -0,0 +1,73 @@
+// Copyright 2024 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package blob
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/cockroachdb/crlib/crstrings"
+	"github.com/cockroachdb/crlib/testutils/leaktest"
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable/block"
+)
+
+func TestBlobWriter(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	var buf bytes.Buffer
+	var obj *objstorage.MemObj
+	datadriven.RunTest(t, "testdata/writer", func(t *testing.T, td *datadriven.TestData) string {
+		buf.Reset()
+		switch td.Cmd {
+		case "build":
+			var (
+				targetBlockSize    int = 128
+				blockSizeThreshold int = 90
+				compression            = block.NoCompression
+			)
+			td.MaybeScanArgs(t, "target-block-size", &targetBlockSize)
+			td.MaybeScanArgs(t, "block-size-threshold", &blockSizeThreshold)
+			if cmdArg, ok := td.Arg("compression"); ok {
+				compression = block.CompressionFromString(cmdArg.SingleVal(t))
+			}
+
+			obj = &objstorage.MemObj{}
+			w := NewFileWriter(000001, obj, FileWriterOptions{
+				Compression:   compression,
+				ChecksumType:  block.ChecksumTypeCRC32c,
+				FlushGovernor: block.MakeFlushGovernor(targetBlockSize, blockSizeThreshold, 0, nil),
+			})
+			for _, l := range crstrings.Lines(td.Input) {
+				h := w.AddValue([]byte(l))
+				fmt.Fprintln(&buf, h)
+			}
+			stats, err := w.Close()
+			if err != nil {
+				t.Fatal(err)
+			}
+			fmt.Fprintf(&buf, "Stats:\n")
+			fmt.Fprintf(&buf, "  BlockCount: %d\n", stats.BlockCount)
+			fmt.Fprintf(&buf, "  ValueCount: %d\n", stats.ValueCount)
+			fmt.Fprintf(&buf, "  BlockLenLongest: %d\n", stats.BlockLenLongest)
+			fmt.Fprintf(&buf, "  UncompressedValueBytes: %d\n", stats.UncompressedValueBytes)
+			fmt.Fprintf(&buf, "  FileLen: %d\n", stats.FileLen)
+			return buf.String()
+		case "open":
+			r, err := NewFileReader(context.Background(), obj)
+			if err != nil {
+				t.Fatal(err)
+			}
+			fmt.Fprintf(&buf, "TableFormat: %s\n", r.footer.format)
+			fmt.Fprintf(&buf, "ChecksumType: %s\n", r.footer.checksum)
+			fmt.Fprintf(&buf, "IndexHandle: %s\n", r.footer.indexHandle.String())
+			return buf.String()
+		default:
+			panic(fmt.Sprintf("unknown command: %s", td.Cmd))
+		}
+	})
+}
diff --git a/sstable/blob/testdata/writer b/sstable/blob/testdata/writer
new file mode 100644
index 00000000..19f8483e
--- /dev/null
+++ b/sstable/blob/testdata/writer
@@ -0,0 +1,66 @@
+build target-block-size=64 block-size-threshold=90
+canteloupe
+apple
+orange
+blueberry
+kiwi
+tangerine
+pear
+pomegranate
+guava
+watermelon
+fig
+plum
+raspberry
+strawberry
+durian
+honeydew
+starfruit
+mango
+grape
+papaya
+lychee
+persimmon
+mandarin
+peach
+apricot
+nectarine
+----
+(000001,blk0[0:10])
+(000001,blk0[10:15])
+(000001,blk0[15:21])
+(000001,blk0[21:30])
+(000001,blk0[30:34])
+(000001,blk0[34:43])
+(000001,blk0[43:47])
+(000001,blk0[47:58])
+(000001,blk0[58:63])
+(000001,blk1[0:10])
+(000001,blk1[10:13])
+(000001,blk1[13:17])
+(000001,blk1[17:26])
+(000001,blk1[26:36])
+(000001,blk1[36:42])
+(000001,blk1[42:50])
+(000001,blk1[50:59])
+(000001,blk1[59:64])
+(000001,blk2[0:5])
+(000001,blk2[5:11])
+(000001,blk2[11:17])
+(000001,blk2[17:26])
+(000001,blk2[26:34])
+(000001,blk2[34:39])
+(000001,blk2[39:46])
+(000001,blk2[46:55])
+Stats:
+  BlockCount: 3
+  ValueCount: 26
+  BlockLenLongest: 69
+  UncompressedValueBytes: 182
+  FileLen: 240
+
+open
+----
+TableFormat: blobV1
+ChecksumType: crc32c
+IndexHandle: {Handle: {197,14}, DataLens:(1,1,1)}
diff --git a/sstable/rowblk/unsafe_test.go b/sstable/rowblk/unsafe_test.go
index d1ad8a3f..d186089d 100644
--- a/sstable/rowblk/unsafe_test.go
+++ b/sstable/rowblk/unsafe_test.go
@@ -14,19 +14,8 @@ import (
 	"unsafe"
 
 	"github.com/cockroachdb/crlib/testutils/leaktest"
-	"github.com/stretchr/testify/require"
 )
 
-func TestGetBytes(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	const size = (1 << 31) - 1
-	// No need to actually allocate a huge slice, which can cause OOM on small
-	// machines (like the GitHub CI runners).
-	block := make([]byte, 100)
-	data := getBytes(unsafe.Pointer(&block[0]), size)
-	require.EqualValues(t, size, len(data))
-}
-
 func TestDecodeVarint(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	vals := []uint32{
diff --git a/testdata/ingest b/testdata/ingest
index 65b0e536..13a97a90 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -54,7 +54,7 @@ Virtual tables: 0 (0B)
 Local tables size: 569B
 Compression types: snappy: 1
 Block cache: 3 entries (1.1KB)  hit rate: 18.2%
-Table cache: 1 entries (904B)  hit rate: 50.0%
+Table cache: 1 entries (912B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
diff --git a/testdata/metrics b/testdata/metrics
index 13c37479..67cbae03 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -76,7 +76,7 @@ Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 1
 Block cache: 2 entries (716B)  hit rate: 0.0%
-Table cache: 1 entries (904B)  hit rate: 0.0%
+Table cache: 1 entries (912B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -227,7 +227,7 @@ Virtual tables: 0 (0B)
 Local tables size: 595B
 Compression types: snappy: 1
 Block cache: 2 entries (716B)  hit rate: 33.3%
-Table cache: 1 entries (904B)  hit rate: 66.7%
+Table cache: 1 entries (912B)  hit rate: 66.7%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 1
@@ -510,7 +510,7 @@ Virtual tables: 0 (0B)
 Local tables size: 4.3KB
 Compression types: snappy: 7
 Block cache: 8 entries (2.8KB)  hit rate: 9.1%
-Table cache: 1 entries (904B)  hit rate: 53.8%
+Table cache: 1 entries (912B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -575,7 +575,7 @@ Virtual tables: 0 (0B)
 Local tables size: 6.1KB
 Compression types: snappy: 10
 Block cache: 8 entries (2.8KB)  hit rate: 9.1%
-Table cache: 1 entries (904B)  hit rate: 53.8%
+Table cache: 1 entries (912B)  hit rate: 53.8%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -853,7 +853,7 @@ Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 1
 Block cache: 0 entries (0B)  hit rate: 0.0%
-Table cache: 1 entries (904B)  hit rate: 0.0%
+Table cache: 1 entries (912B)  hit rate: 0.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -901,7 +901,7 @@ Virtual tables: 0 (0B)
 Local tables size: 0B
 Compression types: snappy: 2
 Block cache: 4 entries (1.4KB)  hit rate: 0.0%
-Table cache: 1 entries (904B)  hit rate: 50.0%
+Table cache: 1 entries (912B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0
@@ -950,7 +950,7 @@ Virtual tables: 0 (0B)
 Local tables size: 589B
 Compression types: snappy: 3
 Block cache: 4 entries (1.4KB)  hit rate: 0.0%
-Table cache: 1 entries (904B)  hit rate: 50.0%
+Table cache: 1 entries (912B)  hit rate: 50.0%
 Secondary cache: 0 entries (0B)  hit rate: 0.0%
 Snapshots: 0  earliest seq num: 0
 Table iters: 0