Repository: cockroachdb/pebble
Pull request #3695 (2024-06-25 15:07:28)

base: rename Trailer to InternalKeyTrailer
This type needs to be used by Cockroach and is now exported as
`pebble.InternalKeyTrailer`.

cockroach-teamcity: 

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


Patch:
diff --git a/internal.go b/internal.go
index 90718d02..050e234d 100644
--- a/internal.go
+++ b/internal.go
@@ -6,6 +6,9 @@ package pebble
 
 import "github.com/cockroachdb/pebble/internal/base"
 
+// SeqNum exports the base.SeqNum type.
+type SeqNum = base.SeqNum
+
 // InternalKeyKind exports the base.InternalKeyKind type.
 type InternalKeyKind = base.InternalKeyKind
 
@@ -32,11 +35,17 @@ const (
 	InternalKeyRangeDeleteSentinel = base.InternalKeyRangeDeleteSentinel
 )
 
+// InternalKeyTrailer exports the base.InternalKeyTrailer type.
+type InternalKeyTrailer = base.InternalKeyTrailer
+
 // InternalKey exports the base.InternalKey type.
 type InternalKey = base.InternalKey
 
-// SeqNum exports the base.SeqNum type.
-type SeqNum = base.SeqNum
+// MakeInternalKey constructs an internal key from a specified user key,
+// sequence number and kind.
+func MakeInternalKey(userKey []byte, seqNum SeqNum, kind InternalKeyKind) InternalKey {
+	return base.MakeInternalKey(userKey, seqNum, kind)
+}
 
 type internalIterator = base.InternalIterator
 
diff --git a/internal/arenaskl/node.go b/internal/arenaskl/node.go
index 43307de9..ad66d2fe 100644
--- a/internal/arenaskl/node.go
+++ b/internal/arenaskl/node.go
@@ -47,7 +47,7 @@ type node struct {
 	// Immutable fields, so no need to lock to access key.
 	keyOffset  uint32
 	keySize    uint32
-	keyTrailer base.Trailer
+	keyTrailer base.InternalKeyTrailer
 	valueSize  uint32
 	allocSize  uint32
 
diff --git a/internal/base/internal.go b/internal/base/internal.go
index 40188def..d8b531a5 100644
--- a/internal/base/internal.go
+++ b/internal/base/internal.go
@@ -133,7 +133,7 @@ const (
 
 	// InternalKeyZeroSeqnumMaxTrailer is the largest trailer with a
 	// zero sequence number.
-	InternalKeyZeroSeqnumMaxTrailer Trailer = 255
+	InternalKeyZeroSeqnumMaxTrailer InternalKeyTrailer = 255
 
 	// A marker for an invalid key.
 	InternalKeyKindInvalid InternalKeyKind = InternalKeyKindSSTableInternalObsoleteMask
@@ -150,13 +150,13 @@ const (
 	// when a range deletion tombstone is the largest key in an sstable. This is
 	// necessary because sstable boundaries are inclusive, while the end key of a
 	// range deletion tombstone is exclusive.
-	InternalKeyRangeDeleteSentinel = (Trailer(InternalKeySeqNumMax) << 8) | Trailer(InternalKeyKindRangeDelete)
+	InternalKeyRangeDeleteSentinel = (InternalKeyTrailer(InternalKeySeqNumMax) << 8) | InternalKeyTrailer(InternalKeyKindRangeDelete)
 
 	// InternalKeyBoundaryRangeKey is the marker for a range key boundary. This
 	// sequence number and kind are used during interleaved range key and point
 	// iteration to allow an iterator to stop at range key start keys where
 	// there exists no point key.
-	InternalKeyBoundaryRangeKey = (Trailer(InternalKeySeqNumMax) << 8) | Trailer(InternalKeyKindRangeKeySet)
+	InternalKeyBoundaryRangeKey = (InternalKeyTrailer(InternalKeySeqNumMax) << 8) | InternalKeyTrailer(InternalKeyKindRangeKeySet)
 )
 
 // Assert InternalKeyKindSSTableInternalObsoleteBit > InternalKeyKindMax
@@ -191,22 +191,22 @@ func (k InternalKeyKind) SafeFormat(w redact.SafePrinter, _ rune) {
 	w.Print(redact.SafeString(k.String()))
 }
 
-// Trailer encodes a SeqNum and an InternalKeyKind.
-type Trailer uint64
+// InternalKeyTrailer encodes a SeqNum and an InternalKeyKind.
+type InternalKeyTrailer uint64
 
 // MakeTrailer constructs an internal key trailer from the specified sequence
 // number and kind.
-func MakeTrailer(seqNum SeqNum, kind InternalKeyKind) Trailer {
-	return (Trailer(seqNum) << 8) | Trailer(kind)
+func MakeTrailer(seqNum SeqNum, kind InternalKeyKind) InternalKeyTrailer {
+	return (InternalKeyTrailer(seqNum) << 8) | InternalKeyTrailer(kind)
 }
 
 // SeqNum returns the sequence number component of the trailer.
-func (t Trailer) SeqNum() SeqNum {
+func (t InternalKeyTrailer) SeqNum() SeqNum {
 	return SeqNum(t >> 8)
 }
 
 // Kind returns the key kind component of the trailer.
-func (t Trailer) Kind() InternalKeyKind {
+func (t InternalKeyTrailer) Kind() InternalKeyKind {
 	return InternalKeyKind(t & 0xff)
 }
 
@@ -219,7 +219,7 @@ func (t Trailer) Kind() InternalKeyKind {
 //   - 7 bytes for a uint56 sequence number, in little-endian format.
 type InternalKey struct {
 	UserKey []byte
-	Trailer Trailer
+	Trailer InternalKeyTrailer
 }
 
 // InvalidInternalKey is an invalid internal key for which Valid() will return
@@ -331,12 +331,12 @@ const InternalTrailerLen = 8
 // DecodeInternalKey decodes an encoded internal key. See InternalKey.Encode().
 func DecodeInternalKey(encodedKey []byte) InternalKey {
 	n := len(encodedKey) - InternalTrailerLen
-	var trailer Trailer
+	var trailer InternalKeyTrailer
 	if n >= 0 {
-		trailer = Trailer(binary.LittleEndian.Uint64(encodedKey[n:]))
+		trailer = InternalKeyTrailer(binary.LittleEndian.Uint64(encodedKey[n:]))
 		encodedKey = encodedKey[:n:n]
 	} else {
-		trailer = Trailer(InternalKeyKindInvalid)
+		trailer = InternalKeyTrailer(InternalKeyKindInvalid)
 		encodedKey = nil
 	}
 	return InternalKey{
@@ -416,7 +416,7 @@ func (k InternalKey) Size() int {
 
 // SetSeqNum sets the sequence number component of the key.
 func (k *InternalKey) SetSeqNum(seqNum SeqNum) {
-	k.Trailer = (Trailer(seqNum) << 8) | (k.Trailer & 0xff)
+	k.Trailer = (InternalKeyTrailer(seqNum) << 8) | (k.Trailer & 0xff)
 }
 
 // SeqNum returns the sequence number component of the key.
@@ -460,7 +460,7 @@ func Visible(seqNum SeqNum, snapshot, batchSnapshot SeqNum) bool {
 
 // SetKind sets the kind component of the key.
 func (k *InternalKey) SetKind(kind InternalKeyKind) {
-	k.Trailer = (k.Trailer &^ 0xff) | Trailer(kind)
+	k.Trailer = (k.Trailer &^ 0xff) | InternalKeyTrailer(kind)
 }
 
 // Kind returns the kind component of the key.
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index 69f287ad..f17e88e7 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -167,7 +167,7 @@ type Iter struct {
 	rangeKeyCompactor RangeKeySpanCompactor
 	err               error
 	// `key.UserKey` is set to `keyBuf` caused by saving `i.iterKV.UserKey`
-	// and `key.Trailer` is set to `i.iterKV.Trailer`. This is the
+	// and `key.InternalKeyTrailer` is set to `i.iterKV.InternalKeyTrailer`. This is the
 	// case on return from all public methods -- these methods return `key`.
 	// Additionally, it is the internal state when the code is moving to the
 	// next key so it can determine whether the user key has changed from
@@ -176,7 +176,7 @@ type Iter struct {
 	// keyTrailer is updated when `i.key` is updated and holds the key's
 	// original trailer (eg, before any sequence-number zeroing or changes to
 	// key kind).
-	keyTrailer  base.Trailer
+	keyTrailer  base.InternalKeyTrailer
 	value       []byte
 	valueCloser io.Closer
 	// Temporary buffer used for storing the previous user key in order to
diff --git a/internal/keyspan/fragmenter.go b/internal/keyspan/fragmenter.go
index 5b48ad06..9b3bf614 100644
--- a/internal/keyspan/fragmenter.go
+++ b/internal/keyspan/fragmenter.go
@@ -184,7 +184,7 @@ func (f *Fragmenter) checkInvariants(buf []Span) {
 // stability, typically callers only need to perform a shallow clone of the Span
 // before Add-ing it to the fragmenter.
 //
-// Add requires the provided span's keys are sorted in Trailer descending order.
+// Add requires the provided span's keys are sorted in InternalKeyTrailer descending order.
 func (f *Fragmenter) Add(s Span) {
 	if f.finished {
 		panic("pebble: span fragmenter already finished")
diff --git a/internal/keyspan/keyspanimpl/merging_iter.go b/internal/keyspan/keyspanimpl/merging_iter.go
index ad54e725..9ddf9b68 100644
--- a/internal/keyspan/keyspanimpl/merging_iter.go
+++ b/internal/keyspan/keyspanimpl/merging_iter.go
@@ -200,7 +200,7 @@ type MergingIter struct {
 // MergingBuffers holds buffers used while merging keyspans.
 type MergingBuffers struct {
 	// keys holds all of the keys across all levels that overlap the key span
-	// [start, end), sorted by Trailer descending. This slice is reconstituted
+	// [start, end), sorted by InternalKeyTrailer descending. This slice is reconstituted
 	// in synthesizeKeys from each mergingIterLevel's keys every time the
 	// [start, end) bounds change.
 	//
diff --git a/internal/keyspan/span.go b/internal/keyspan/span.go
index af84a1bb..e88412da 100644
--- a/internal/keyspan/span.go
+++ b/internal/keyspan/span.go
@@ -47,7 +47,7 @@ type Span struct {
 type KeysOrder int8
 
 const (
-	// ByTrailerDesc indicates a Span's keys are sorted by Trailer descending.
+	// ByTrailerDesc indicates a Span's keys are sorted by InternalKeyTrailer descending.
 	// This is the default ordering, and the ordering used during physical
 	// storage.
 	ByTrailerDesc KeysOrder = iota
@@ -61,7 +61,7 @@ const (
 // is applied.
 type Key struct {
 	// Trailer contains the key kind and sequence number.
-	Trailer base.Trailer
+	Trailer base.InternalKeyTrailer
 	// Suffix holds an optional suffix associated with the key. This is only
 	// non-nil for RANGEKEYSET and RANGEKEYUNSET keys.
 	Suffix []byte
diff --git a/internal/manifest/version.go b/internal/manifest/version.go
index 607bc034..39f87f17 100644
--- a/internal/manifest/version.go
+++ b/internal/manifest/version.go
@@ -1685,7 +1685,7 @@ func CheckOrdering(cmp Compare, format base.FormatKey, level Level, files LevelI
 				// In all supported format major version, split user keys are
 				// prohibited, so both files cannot contain keys with the same user
 				// keys. If the bounds have the same user key, the previous file's
-				// boundary must have a Trailer indicating that it's exclusive.
+				// boundary must have a InternalKeyTrailer indicating that it's exclusive.
 				if v := cmp(prev.Largest.UserKey, f.Smallest.UserKey); v > 0 || (v == 0 && !prev.Largest.IsExclusiveSentinel()) {
 					return base.CorruptionErrorf("%s files %s and %s have overlapping ranges: [%s-%s] vs [%s-%s]",
 						errors.Safe(level), errors.Safe(prev.FileNum), errors.Safe(f.FileNum),
diff --git a/internal/rangekey/coalesce.go b/internal/rangekey/coalesce.go
index db7e7ea2..c95809fd 100644
--- a/internal/rangekey/coalesce.go
+++ b/internal/rangekey/coalesce.go
@@ -48,7 +48,7 @@ import (
 // consistent with respect to the set/unset suffixes: A given suffix should be
 // set or unset but not both.
 //
-// The resulting dst Keys slice is sorted by Trailer.
+// The resulting dst Keys slice is sorted by InternalKeyTrailer.
 func Coalesce(cmp base.Compare, eq base.Equal, keys []keyspan.Key, dst *[]keyspan.Key) {
 	// TODO(jackson): Currently, Coalesce doesn't actually perform the sequence
 	// number promotion described in the comment above.
@@ -101,7 +101,7 @@ func CoalesceIntoKeysBySuffix(
 	// suffix, in which case the one with a larger trailer survives.
 	//
 	// We use a stable sort so that the first key with a given suffix is the one
-	// that with the highest Trailer (because the input `keys` was sorted by
+	// that with the highest InternalKeyTrailer (because the input `keys` was sorted by
 	// trailer descending).
 	sort.Stable(keysBySuffix)
 
@@ -125,7 +125,7 @@ func CoalesceIntoKeysBySuffix(
 	for i := range sorted {
 		if i > 0 && equal(prevSuffix, sorted[i].Suffix) {
 			// Skip; this key is shadowed by the predecessor that had a larger
-			// Trailer. If this is the first shadowed key, set shadowing=true
+			// InternalKeyTrailer. If this is the first shadowed key, set shadowing=true
 			// and reslice keysBySuffix.keys to hold the entire unshadowed
 			// prefix.
 			if !shadowing {
diff --git a/iterator.go b/iterator.go
index 9da81228..f9c2ab4a 100644
--- a/iterator.go
+++ b/iterator.go
@@ -766,7 +766,7 @@ func (i *Iterator) nextUserKey() {
 		// NB: We're guaranteed to be on the next user key if the previous key
 		// had a zero sequence number (`done`), or the new key has a trailer
 		// greater or equal to the previous key's trailer. This is true because
-		// internal keys with the same user key are sorted by Trailer in
+		// internal keys with the same user key are sorted by InternalKeyTrailer in
 		// strictly monotonically descending order. We expect the trailer
 		// optimization to trigger around 50% of the time with randomly
 		// distributed writes. We expect it to trigger very frequently when
diff --git a/scan_internal.go b/scan_internal.go
index bb7e52e5..7c5db143 100644
--- a/scan_internal.go
+++ b/scan_internal.go
@@ -1020,7 +1020,7 @@ func (i *scanInternalIterator) constructRangeKeyIter() error {
 	// NB: We iterate L0's files in reverse order. They're sorted by
 	// LargestSeqNum ascending, and we need to add them to the merging iterator
 	// in LargestSeqNum descending to preserve the merging iterator's invariants
-	// around Key Trailer order.
+	// around Key InternalKeyTrailer order.
 	iter := current.RangeKeyLevels[0].Iter()
 	for f := iter.Last(); f != nil; f = iter.Prev() {
 		spanIter, err := i.newIterRangeKey(f, i.opts.SpanIterOptions())
diff --git a/sstable/block/block.go b/sstable/block/block.go
index c811d7f5..28d0b8a9 100644
--- a/sstable/block/block.go
+++ b/sstable/block/block.go
@@ -6,11 +6,84 @@ package block
 
 import (
 	"bytes"
+	"encoding/binary"
 	"fmt"
 
+	"github.com/cespare/xxhash/v2"
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/crc"
 )
 
+// TrailerLen is the length of the trailer at the end of a block.
+const TrailerLen = 5
+
+// Trailer is the trailer at the end of a block, encoding the block type
+// (compression) and a checksum.
+type Trailer = [TrailerLen]byte
+
+// MakeTrailer constructs a trailer from a block type and a checksum.
+func MakeTrailer(blockType byte, checksum uint32) (t Trailer) {
+	t[0] = blockType
+	binary.LittleEndian.PutUint32(t[1:5], checksum)
+	return t
+}
+
+// ChecksumType specifies the checksum used for blocks.
+type ChecksumType byte
+
+// The available checksum types. These values are part of the durable format and
+// should not be changed.
+const (
+	ChecksumTypeNone     ChecksumType = 0
+	ChecksumTypeCRC32c   ChecksumType = 1
+	ChecksumTypeXXHash   ChecksumType = 2
+	ChecksumTypeXXHash64 ChecksumType = 3
+)
+
+// String implements fmt.Stringer.
+func (t ChecksumType) String() string {
+	switch t {
+	case ChecksumTypeCRC32c:
+		return "crc32c"
+	case ChecksumTypeNone:
+		return "none"
+	case ChecksumTypeXXHash:
+		return "xxhash"
+	case ChecksumTypeXXHash64:
+		return "xxhash64"
+	default:
+		panic(errors.Newf("sstable: unknown checksum type: %d", t))
+	}
+}
+
+// A Checksummer calculates checksums for blocks.
+type Checksummer struct {
+	Type     ChecksumType
+	xxHasher *xxhash.Digest
+}
+
+// Checksum computes a checksum over the provided block and block type.
+func (c *Checksummer) Checksum(block []byte, blockType []byte) (checksum uint32) {
+	// Calculate the checksum.
+	switch c.Type {
+	case ChecksumTypeCRC32c:
+		checksum = crc.New(block).Update(blockType).Value()
+	case ChecksumTypeXXHash64:
+		if c.xxHasher == nil {
+			c.xxHasher = xxhash.New()
+		} else {
+			c.xxHasher.Reset()
+		}
+		c.xxHasher.Write(block)
+		c.xxHasher.Write(blockType)
+		checksum = uint32(c.xxHasher.Sum64())
+	default:
+		panic(errors.Newf("unsupported checksum type: %d", c.Type))
+	}
+	return checksum
+}
+
 // IterTransforms allow on-the-fly transformation of data at iteration time.
 //
 // These transformations could in principle be implemented as block transforms
diff --git a/sstable/block/buffer_pool.go b/sstable/block/buffer_pool.go
index 36543d52..6a12b2b5 100644
--- a/sstable/block/buffer_pool.go
+++ b/sstable/block/buffer_pool.go
@@ -6,22 +6,22 @@ package block
 
 import (
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
 )
 
-// MakeCacheValue constructs a CacheValueOrBuf backed by a block cache Value.
-func MakeCacheValue(v *cache.Value) CacheValueOrBuf {
-	return CacheValueOrBuf{v: v}
-}
-
-// MakeBlockBuf constructs a CacheValueOrBuf backed by a BufferPool buffer.
-func MakeBlockBuf(buf Buf) CacheValueOrBuf {
-	return CacheValueOrBuf{buf: buf}
+// Alloc allocates a new Value for a block of length n (excluding the block
+// trailer). If bufferPool is non-nil, Alloc allocates the buffer from the pool.
+// Otherwise it allocates it from the block cache.
+func Alloc(n int, p *BufferPool) Value {
+	if p != nil {
+		return Value{buf: p.Alloc(int(n))}
+	}
+	return Value{v: cache.Alloc(int(n))}
 }
 
-// CacheValueOrBuf is a handle to a block, either backed by the block cache or a
-// BufferPool.
-type CacheValueOrBuf struct {
+// Value is a block buffer, either backed by the block cache or a BufferPool.
+type Value struct {
 	// buf.Valid() returns true if backed by a BufferPool.
 	buf Buf
 	// v is non-nil if backed by the block cache.
@@ -29,21 +29,27 @@ type CacheValueOrBuf struct {
 }
 
 // Get returns the underlying block's byte slice.
-func (b CacheValueOrBuf) Get() []byte {
+func (b Value) Get() []byte {
 	if b.buf.Valid() {
 		return b.buf.p.pool[b.buf.i].b
 	}
 	return b.v.Buf()
 }
 
-// Unpack returns the underlying block's Buf and cache.Value. For a valid
-// CacheValueOrBuf, either Buf.Valid() or cache.Value is non-nil.
-func (b CacheValueOrBuf) Unpack() (Buf, *cache.Value) {
-	return b.buf, b.v
+// MakeHandle constructs a BufferHandle from the Value. If the Value is not
+// backed by a buffer pool, MakeHandle inserts the value into the block cache,
+// returning a handle to the now resident value.
+func (b Value) MakeHandle(
+	c *cache.Cache, cacheID uint64, fileNum base.DiskFileNum, offset uint64,
+) BufferHandle {
+	if b.buf.Valid() {
+		return BufferHandle{b: b.buf}
+	}
+	return BufferHandle{h: c.Set(cacheID, fileNum, offset, b.v)}
 }
 
 // Release releases the handle.
-func (b CacheValueOrBuf) Release() {
+func (b Value) Release() {
 	if b.buf.Valid() {
 		b.buf.Release()
 	} else {
@@ -52,7 +58,7 @@ func (b CacheValueOrBuf) Release() {
 }
 
 // Truncate truncates the block to n bytes.
-func (b CacheValueOrBuf) Truncate(n int) {
+func (b Value) Truncate(n int) {
 	if b.buf.Valid() {
 		b.buf.p.pool[b.buf.i].b = b.buf.p.pool[b.buf.i].b[:n]
 	} else {
@@ -68,11 +74,6 @@ type BufferHandle struct {
 	b Buf
 }
 
-// PooledBufferHandle returns a BufferHandle for a pooled buffer.
-func PooledBufferHandle(b Buf) BufferHandle {
-	return BufferHandle{b: b}
-}
-
 // CacheBufferHandle constructs a BufferHandle from a block cache Handle.
 func CacheBufferHandle(h cache.Handle) BufferHandle {
 	return BufferHandle{h: h}
diff --git a/sstable/block/kv.go b/sstable/block/kv.go
new file mode 100644
index 00000000..1165e9e9
--- /dev/null
+++ b/sstable/block/kv.go
@@ -0,0 +1,70 @@
+// 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 block
+
+import "github.com/cockroachdb/pebble/internal/base"
+
+// ValuePrefix is the single byte prefix in values indicating either an in-place
+// value or a value encoding a valueHandle. It encodes multiple kinds of
+// information (see below).
+type ValuePrefix byte
+
+const (
+	// 2 most-significant bits of valuePrefix encodes the value-kind.
+	valueKindMask           ValuePrefix = 0xC0
+	valueKindIsValueHandle  ValuePrefix = 0x80
+	valueKindIsInPlaceValue ValuePrefix = 0x00
+
+	// 1 bit indicates SET has same key prefix as immediately preceding key that
+	// is also a SET. If the immediately preceding key in the same block is a
+	// SET, AND this bit is 0, the prefix must have changed.
+	//
+	// Note that the current policy of only storing older MVCC versions in value
+	// blocks means that valueKindIsValueHandle => SET has same prefix. But no
+	// code should rely on this behavior. Also, SET has same prefix does *not*
+	// imply valueKindIsValueHandle.
+	setHasSameKeyPrefixMask ValuePrefix = 0x20
+
+	// 3 least-significant bits for the user-defined base.ShortAttribute.
+	// Undefined for valueKindIsInPlaceValue.
+	userDefinedShortAttributeMask ValuePrefix = 0x07
+)
+
+// IsValueHandle returns true if the ValuePrefix is for a valueHandle.
+func (vp ValuePrefix) IsValueHandle() bool {
+	return vp&valueKindMask == valueKindIsValueHandle
+}
+
+// SetHasSamePrefix returns true if the ValuePrefix encodes that the key is a
+// set with the same prefix as the preceding key which also is a set.
+func (vp ValuePrefix) SetHasSamePrefix() bool {
+	return vp&setHasSameKeyPrefixMask == setHasSameKeyPrefixMask
+}
+
+// ShortAttribute returns the user-defined base.ShortAttribute encoded in the
+// ValuePrefix.
+//
+// REQUIRES: IsValueHandle()
+func (vp ValuePrefix) ShortAttribute() base.ShortAttribute {
+	return base.ShortAttribute(vp & userDefinedShortAttributeMask)
+}
+
+// ValueHandlePrefix returns the ValuePrefix for a valueHandle.
+func ValueHandlePrefix(setHasSameKeyPrefix bool, attribute base.ShortAttribute) ValuePrefix {
+	prefix := valueKindIsValueHandle | ValuePrefix(attribute)
+	if setHasSameKeyPrefix {
+		prefix = prefix | setHasSameKeyPrefixMask
+	}
+	return prefix
+}
+
+// InPlaceValuePrefix returns the ValuePrefix for an in-place value.
+func InPlaceValuePrefix(setHasSameKeyPrefix bool) ValuePrefix {
+	prefix := valueKindIsInPlaceValue
+	if setHasSameKeyPrefix {
+		prefix = prefix | setHasSameKeyPrefixMask
+	}
+	return prefix
+}
diff --git a/sstable/block_iter.go b/sstable/block_iter.go
index c088ab8d..4935196c 100644
--- a/sstable/block_iter.go
+++ b/sstable/block_iter.go
@@ -437,14 +437,14 @@ func (i *blockIter) readFirstKey() error {
 
 // The sstable internal obsolete bit is set when writing a block and unset by
 // blockIter, so no code outside block writing/reading code ever sees it.
-const trailerObsoleteBit = base.Trailer(base.InternalKeyKindSSTableInternalObsoleteBit)
-const trailerObsoleteMask = (base.Trailer(InternalKeySeqNumMax) << 8) | base.Trailer(base.InternalKeyKindSSTableInternalObsoleteMask)
+const trailerObsoleteBit = base.InternalKeyTrailer(base.InternalKeyKindSSTableInternalObsoleteBit)
+const trailerObsoleteMask = (base.InternalKeyTrailer(InternalKeySeqNumMax) << 8) | base.InternalKeyTrailer(base.InternalKeyKindSSTableInternalObsoleteMask)
 
 func (i *blockIter) decodeInternalKey(key []byte) (hiddenPoint bool) {
 	// Manually inlining base.DecodeInternalKey provides a 5-10% speedup on
 	// BlockIter benchmarks.
 	if n := len(key) - 8; n >= 0 {
-		trailer := base.Trailer(binary.LittleEndian.Uint64(key[n:]))
+		trailer := base.InternalKeyTrailer(binary.LittleEndian.Uint64(key[n:]))
 		hiddenPoint = i.transforms.HideObsoletePoints &&
 			(trailer&trailerObsoleteBit != 0)
 		i.ikv.K.Trailer = trailer & trailerObsoleteMask
@@ -453,7 +453,7 @@ func (i *blockIter) decodeInternalKey(key []byte) (hiddenPoint bool) {
 			i.ikv.K.SetSeqNum(base.SeqNum(n))
 		}
 	} else {
-		i.ikv.K.Trailer = base.Trailer(InternalKeyKindInvalid)
+		i.ikv.K.Trailer = base.InternalKeyTrailer(InternalKeyKindInvalid)
 		i.ikv.K.UserKey = nil
 	}
 	return hiddenPoint
@@ -655,7 +655,7 @@ func (i *blockIter) SeekGE(key []byte, flags base.SeekGEFlags) *base.InternalKV
 		if !i.lazyValueHandling.hasValuePrefix ||
 			i.ikv.K.Kind() != InternalKeyKindSet {
 			i.ikv.V = base.MakeInPlaceValue(i.val)
-		} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+		} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 			i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 		} else {
 			i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
@@ -938,7 +938,7 @@ func (i *blockIter) SeekLT(key []byte, flags base.SeekLTFlags) *base.InternalKV
 	if !i.lazyValueHandling.hasValuePrefix ||
 		i.ikv.K.Kind() != InternalKeyKindSet {
 		i.ikv.V = base.MakeInPlaceValue(i.val)
-	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+	} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 		i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 	} else {
 		i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
@@ -967,7 +967,7 @@ func (i *blockIter) First() *base.InternalKV {
 	if !i.lazyValueHandling.hasValuePrefix ||
 		i.ikv.K.Kind() != InternalKeyKindSet {
 		i.ikv.V = base.MakeInPlaceValue(i.val)
-	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+	} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 		i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 	} else {
 		i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
@@ -1010,7 +1010,7 @@ func (i *blockIter) Last() *base.InternalKV {
 	if !i.lazyValueHandling.hasValuePrefix ||
 		i.ikv.K.Kind() != InternalKeyKindSet {
 		i.ikv.V = base.MakeInPlaceValue(i.val)
-	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+	} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 		i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 	} else {
 		i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
@@ -1044,7 +1044,7 @@ start:
 	i.readEntry()
 	// Manually inlined version of i.decodeInternalKey(i.key).
 	if n := len(i.key) - 8; n >= 0 {
-		trailer := base.Trailer(binary.LittleEndian.Uint64(i.key[n:]))
+		trailer := base.InternalKeyTrailer(binary.LittleEndian.Uint64(i.key[n:]))
 		hiddenPoint := i.transforms.HideObsoletePoints &&
 			(trailer&trailerObsoleteBit != 0)
 		i.ikv.K.Trailer = trailer & trailerObsoleteMask
@@ -1063,13 +1063,13 @@ start:
 			i.ikv.K.UserKey = i.synthSuffixBuf
 		}
 	} else {
-		i.ikv.K.Trailer = base.Trailer(InternalKeyKindInvalid)
+		i.ikv.K.Trailer = base.InternalKeyTrailer(InternalKeyKindInvalid)
 		i.ikv.K.UserKey = nil
 	}
 	if !i.lazyValueHandling.hasValuePrefix ||
 		i.ikv.K.Kind() != InternalKeyKindSet {
 		i.ikv.V = base.MakeInPlaceValue(i.val)
-	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+	} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 		i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 	} else {
 		i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
@@ -1212,8 +1212,8 @@ func (i *blockIter) nextPrefixV3(succKey []byte) *base.InternalKV {
 			if invariants.Enabled && value == 0 {
 				panic(errors.AssertionFailedf("value is of length 0, but we expect a valuePrefix"))
 			}
-			valPrefix := *((*valuePrefix)(valuePtr))
-			if setHasSamePrefix(valPrefix) {
+			valPrefix := *((*block.ValuePrefix)(valuePtr))
+			if valPrefix.SetHasSamePrefix() {
 				// Fast-path. No need to assemble i.fullKey, or update i.key. We know
 				// that subsequent keys will not have a shared length that is greater
 				// than the prefix of the current key, which is also the prefix of
@@ -1324,7 +1324,7 @@ func (i *blockIter) nextPrefixV3(succKey []byte) *base.InternalKV {
 		// Manually inlined version of i.decodeInternalKey(i.key).
 		hiddenPoint := false
 		if n := len(i.key) - 8; n >= 0 {
-			trailer := base.Trailer(binary.LittleEndian.Uint64(i.key[n:]))
+			trailer := base.InternalKeyTrailer(binary.LittleEndian.Uint64(i.key[n:]))
 			hiddenPoint = i.transforms.HideObsoletePoints &&
 				(trailer&trailerObsoleteBit != 0)
 			i.ikv.K = base.InternalKey{
@@ -1342,7 +1342,7 @@ func (i *blockIter) nextPrefixV3(succKey []byte) *base.InternalKV {
 				i.ikv.K.UserKey = i.synthSuffixBuf
 			}
 		} else {
-			i.ikv.K.Trailer = base.Trailer(InternalKeyKindInvalid)
+			i.ikv.K.Trailer = base.InternalKeyTrailer(InternalKeyKindInvalid)
 			i.ikv.K.UserKey = nil
 		}
 		nextCmpCount++
@@ -1360,7 +1360,7 @@ func (i *blockIter) nextPrefixV3(succKey []byte) *base.InternalKV {
 			}
 			if i.ikv.K.Kind() != InternalKeyKindSet {
 				i.ikv.V = base.MakeInPlaceValue(i.val)
-			} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+			} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 				i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 			} else {
 				i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
@@ -1388,7 +1388,7 @@ start:
 		// Manually inlined version of i.decodeInternalKey(i.key).
 		i.key = i.cachedBuf[e.keyStart:e.keyEnd]
 		if n := len(i.key) - 8; n >= 0 {
-			trailer := base.Trailer(binary.LittleEndian.Uint64(i.key[n:]))
+			trailer := base.InternalKeyTrailer(binary.LittleEndian.Uint64(i.key[n:]))
 			hiddenPoint := i.transforms.HideObsoletePoints &&
 				(trailer&trailerObsoleteBit != 0)
 			if hiddenPoint {
@@ -1411,14 +1411,14 @@ start:
 				i.ikv.K.UserKey = i.synthSuffixBuf
 			}
 		} else {
-			i.ikv.K.Trailer = base.Trailer(InternalKeyKindInvalid)
+			i.ikv.K.Trailer = base.InternalKeyTrailer(InternalKeyKindInvalid)
 			i.ikv.K.UserKey = nil
 		}
 		i.cached = i.cached[:n]
 		if !i.lazyValueHandling.hasValuePrefix ||
 			i.ikv.K.Kind() != InternalKeyKindSet {
 			i.ikv.V = base.MakeInPlaceValue(i.val)
-		} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+		} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 			i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 		} else {
 			i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
@@ -1499,7 +1499,7 @@ start:
 	if !i.lazyValueHandling.hasValuePrefix ||
 		i.ikv.K.Kind() != InternalKeyKindSet {
 		i.ikv.V = base.MakeInPlaceValue(i.val)
-	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+	} else if i.lazyValueHandling.vbr == nil || !block.ValuePrefix(i.val[0]).IsValueHandle() {
 		i.ikv.V = base.MakeInPlaceValue(i.val[1:])
 	} else {
 		i.ikv.V = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
diff --git a/sstable/block_writer.go b/sstable/block_writer.go
index 9f8d52d4..126543bc 100644
--- a/sstable/block_writer.go
+++ b/sstable/block_writer.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 type blockWriter struct {
@@ -86,7 +87,7 @@ func (w *blockWriter) storeWithOptionalValuePrefix(
 	value []byte,
 	maxSharedKeyLen int,
 	addValuePrefix bool,
-	valuePrefix valuePrefix,
+	valuePrefix block.ValuePrefix,
 	setHasSameKeyPrefix bool,
 ) {
 	shared := 0
@@ -210,7 +211,7 @@ func (w *blockWriter) addWithOptionalValuePrefix(
 	value []byte,
 	maxSharedKeyLen int,
 	addValuePrefix bool,
-	valuePrefix valuePrefix,
+	valuePrefix block.ValuePrefix,
 	setHasSameKeyPrefix bool,
 ) {
 	w.curKey, w.prevKey = w.prevKey, w.curKey
diff --git a/sstable/copier.go b/sstable/copier.go
index fb4e3810..a949c284 100644
--- a/sstable/copier.go
+++ b/sstable/copier.go
@@ -156,7 +156,7 @@ func CopySpan(
 	// The block lengths don't include their trailers, which just sit after the
 	// block length, before the next offset; We get the ones between the blocks
 	// we copy implicitly but need to explicitly add the last trailer to length.
-	length := blocks[len(blocks)-1].bh.Offset + blocks[len(blocks)-1].bh.Length + blockTrailerLen - offset
+	length := blocks[len(blocks)-1].bh.Offset + blocks[len(blocks)-1].bh.Length + block.TrailerLen - offset
 
 	if spanEnd := length + offset; spanEnd < offset {
 		return 0, base.AssertionFailedf("invalid intersecting span for CopySpan [%d, %d)", offset, spanEnd)
diff --git a/sstable/layout.go b/sstable/layout.go
index 12af5147..e4cbf995 100644
--- a/sstable/layout.go
+++ b/sstable/layout.go
@@ -16,6 +16,7 @@ import (
 	"unsafe"
 
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 // Layout describes the block organization of an sstable.
@@ -44,51 +45,51 @@ func (l *Layout) Describe(
 	w io.Writer, verbose bool, r *Reader, fmtRecord func(key *base.InternalKey, value []byte),
 ) {
 	ctx := context.TODO()
-	type block struct {
+	type namedBlockHandle struct {
 		BlockHandle
 		name string
 	}
-	var blocks []block
+	var blocks []namedBlockHandle
 
 	for i := range l.Data {
-		blocks = append(blocks, block{l.Data[i].BlockHandle, "data"})
+		blocks = append(blocks, namedBlockHandle{l.Data[i].BlockHandle, "data"})
 	}
 	for i := range l.Index {
-		blocks = append(blocks, block{l.Index[i], "index"})
+		blocks = append(blocks, namedBlockHandle{l.Index[i], "index"})
 	}
 	if l.TopIndex.Length != 0 {
-		blocks = append(blocks, block{l.TopIndex, "top-index"})
+		blocks = append(blocks, namedBlockHandle{l.TopIndex, "top-index"})
 	}
 	if l.Filter.Length != 0 {
-		blocks = append(blocks, block{l.Filter, "filter"})
+		blocks = append(blocks, namedBlockHandle{l.Filter, "filter"})
 	}
 	if l.RangeDel.Length != 0 {
-		blocks = append(blocks, block{l.RangeDel, "range-del"})
+		blocks = append(blocks, namedBlockHandle{l.RangeDel, "range-del"})
 	}
 	if l.RangeKey.Length != 0 {
-		blocks = append(blocks, block{l.RangeKey, "range-key"})
+		blocks = append(blocks, namedBlockHandle{l.RangeKey, "range-key"})
 	}
 	for i := range l.ValueBlock {
-		blocks = append(blocks, block{l.ValueBlock[i], "value-block"})
+		blocks = append(blocks, namedBlockHandle{l.ValueBlock[i], "value-block"})
 	}
 	if l.ValueIndex.Length != 0 {
-		blocks = append(blocks, block{l.ValueIndex, "value-index"})
+		blocks = append(blocks, namedBlockHandle{l.ValueIndex, "value-index"})
 	}
 	if l.Properties.Length != 0 {
-		blocks = append(blocks, block{l.Properties, "properties"})
+		blocks = append(blocks, namedBlockHandle{l.Properties, "properties"})
 	}
 	if l.MetaIndex.Length != 0 {
-		blocks = append(blocks, block{l.MetaIndex, "meta-index"})
+		blocks = append(blocks, namedBlockHandle{l.MetaIndex, "meta-index"})
 	}
 	if l.Footer.Length != 0 {
 		if l.Footer.Length == levelDBFooterLen {
-			blocks = append(blocks, block{l.Footer, "leveldb-footer"})
+			blocks = append(blocks, namedBlockHandle{l.Footer, "leveldb-footer"})
 		} else {
-			blocks = append(blocks, block{l.Footer, "footer"})
+			blocks = append(blocks, namedBlockHandle{l.Footer, "footer"})
 		}
 	}
 
-	slices.SortFunc(blocks, func(a, b block) int {
+	slices.SortFunc(blocks, func(a, b namedBlockHandle) int {
 		return cmp.Compare(a.Offset, b.Offset)
 	})
 	for i := range blocks {
@@ -107,7 +108,7 @@ func (l *Layout) Describe(
 			_ = r.readable.ReadAt(ctx, trailer, int64(offset))
 
 			if b.name == "footer" {
-				checksumType := ChecksumType(trailer[0])
+				checksumType := block.ChecksumType(trailer[0])
 				fmt.Fprintf(w, "%10d    checksum type: %s\n", offset, checksumType)
 				trailer, offset = trailer[1:], offset+1
 			}
@@ -175,7 +176,7 @@ func (l *Layout) Describe(
 		}
 
 		formatTrailer := func() {
-			trailer := make([]byte, blockTrailerLen)
+			trailer := make([]byte, block.TrailerLen)
 			offset := int64(b.Offset + b.Length)
 			_ = r.readable.ReadAt(ctx, trailer, offset)
 			bt := blockType(trailer[0])
@@ -220,7 +221,7 @@ func (l *Layout) Describe(
 						v := kv.InPlaceValue()
 						if kv.K.Kind() != InternalKeyKindSet {
 							fmtRecord(&kv.K, v)
-						} else if !isValueHandle(valuePrefix(v[0])) {
+						} else if !block.ValuePrefix(v[0]).IsValueHandle() {
 							fmtRecord(&kv.K, v[1:])
 						} else {
 							vh := decodeValueHandle(v[1:])
diff --git a/sstable/options.go b/sstable/options.go
index 98013739..aa503177 100644
--- a/sstable/options.go
+++ b/sstable/options.go
@@ -8,6 +8,7 @@ import (
 	"github.com/cockroachdb/fifo"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/cache"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 // Compression is the per-block compression algorithm to use.
@@ -234,7 +235,7 @@ type WriterOptions struct {
 	BlockPropertyCollectors []func() BlockPropertyCollector
 
 	// Checksum specifies which checksum to use.
-	Checksum ChecksumType
+	Checksum block.ChecksumType
 
 	// Parallelism is used to indicate that the sstable Writer is allowed to
 	// compress data blocks and write datablocks to disk in parallel with the
@@ -292,8 +293,8 @@ func (o WriterOptions) ensureDefaults() WriterOptions {
 	if o.MergerName == "" {
 		o.MergerName = base.DefaultMerger.Name
 	}
-	if o.Checksum == ChecksumTypeNone {
-		o.Checksum = ChecksumTypeCRC32c
+	if o.Checksum == block.ChecksumTypeNone {
+		o.Checksum = block.ChecksumTypeCRC32c
 	}
 	// By default, if the table format is not specified, fall back to using the
 	// most compatible format that is supported by Pebble.
diff --git a/sstable/reader.go b/sstable/reader.go
index 317904de..b9010885 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -200,7 +200,7 @@ type Reader struct {
 	tableFormat   TableFormat
 	rawTombstones bool
 	mergerOK      bool
-	checksumType  ChecksumType
+	checksumType  block.ChecksumType
 	// metaBufferPool is a buffer pool used exclusively when opening a table and
 	// loading its meta blocks. metaBufferPoolAlloc is used to batch-allocate
 	// the BufferPool.pool slice as a part of the Reader allocation. It's
@@ -473,14 +473,14 @@ func (r *Reader) readRangeKey(
 }
 
 func checkChecksum(
-	checksumType ChecksumType, b []byte, bh BlockHandle, fileNum base.DiskFileNum,
+	checksumType block.ChecksumType, b []byte, bh BlockHandle, fileNum base.DiskFileNum,
 ) error {
 	expectedChecksum := binary.LittleEndian.Uint32(b[bh.Length+1:])
 	var computedChecksum uint32
 	switch checksumType {
-	case ChecksumTypeCRC32c:
+	case block.ChecksumTypeCRC32c:
 		computedChecksum = crc.New(b[:bh.Length+1]).Value()
-	case ChecksumTypeXXHash64:
+	case block.ChecksumTypeXXHash64:
 		computedChecksum = uint32(xxhash.Sum64(b[:bh.Length+1]))
 	default:
 		return errors.Errorf("unsupported checksum type: %d", checksumType)
@@ -519,7 +519,7 @@ func (r *Reader) readBlock(
 	if h := r.opts.Cache.Get(r.cacheID, r.fileNum, bh.Offset); h.Get() != nil {
 		// Cache hit.
 		if readHandle != nil {
-			readHandle.RecordCacheHit(ctx, int64(bh.Offset), int64(bh.Length+blockTrailerLen))
+			readHandle.RecordCacheHit(ctx, int64(bh.Offset), int64(bh.Length+block.TrailerLen))
 		}
 		if stats != nil {
 			stats.BlockBytes += bh.Length
@@ -543,13 +543,7 @@ func (r *Reader) readBlock(
 		defer sema.Release(1)
 	}
 
-	var compressed block.CacheValueOrBuf
-	if bufferPool != nil {
-		compressed = block.MakeBlockBuf(bufferPool.Alloc(int(bh.Length + blockTrailerLen)))
-	} else {
-		compressed = block.MakeCacheValue(cache.Alloc(int(bh.Length + blockTrailerLen)))
-	}
-
+	compressed := block.Alloc(int(bh.Length+block.TrailerLen), bufferPool)
 	readStartTime := time.Now()
 	var err error
 	if readHandle != nil {
@@ -568,7 +562,7 @@ func (r *Reader) readBlock(
 	// interface{}, unless necessary.
 	if readDuration >= slowReadTracingThreshold && r.opts.LoggerAndTracer.IsTracingEnabled(ctx) {
 		r.opts.LoggerAndTracer.Eventf(ctx, "reading %d bytes took %s",
-			int(bh.Length+blockTrailerLen), readDuration.String())
+			int(bh.Length+block.TrailerLen), readDuration.String())
 	}
 	if stats != nil {
 		stats.BlockBytes += bh.Length
@@ -586,7 +580,7 @@ func (r *Reader) readBlock(
 	typ := blockType(compressed.Get()[bh.Length])
 	compressed.Truncate(int(bh.Length))
 
-	var decompressed block.CacheValueOrBuf
+	var decompressed block.Value
 	if typ == noCompressionBlockType {
 		decompressed = compressed
 	} else {
@@ -597,11 +591,7 @@ func (r *Reader) readBlock(
 			return block.BufferHandle{}, err
 		}
 
-		if bufferPool != nil {
-			decompressed = block.MakeBlockBuf(bufferPool.Alloc(decodedLen))
-		} else {
-			decompressed = block.MakeCacheValue(cache.Alloc(decodedLen))
-		}
+		decompressed = block.Alloc(decodedLen, bufferPool)
 		if err := decompressInto(typ, compressed.Get()[prefixLen:], decompressed.Get()); err != nil {
 			compressed.Release()
 			return block.BufferHandle{}, err
@@ -618,12 +608,7 @@ func (r *Reader) readBlock(
 			return block.BufferHandle{}, err
 		}
 
-		var transformed block.CacheValueOrBuf
-		if bufferPool != nil {
-			transformed = block.MakeBlockBuf(bufferPool.Alloc(len(tmpTransformed)))
-		} else {
-			transformed = block.MakeCacheValue(cache.Alloc(len(tmpTransformed)))
-		}
+		transformed := block.Alloc(len(tmpTransformed), bufferPool)
 		copy(transformed.Get(), tmpTransformed)
 		decompressed.Release()
 		decompressed = transformed
@@ -632,12 +617,8 @@ func (r *Reader) readBlock(
 	if iterStats != nil {
 		iterStats.reportStats(bh.Length, 0, readDuration)
 	}
-	pooledBuf, cacheV := decompressed.Unpack()
-	if pooledBuf.Valid() {
-		return block.PooledBufferHandle(pooledBuf), nil
-	}
-	h := r.opts.Cache.Set(r.cacheID, r.fileNum, bh.Offset, cacheV)
-	return block.CacheBufferHandle(h), nil
+	h := decompressed.MakeHandle(r.opts.Cache, r.cacheID, r.fileNum, bh.Offset)
+	return h, nil
 }
 
 func (r *Reader) transformRangeDelV1(b []byte) ([]byte, error) {
@@ -1087,7 +1068,7 @@ func (r *Reader) EstimateDiskUsage(start, end []byte) (uint64, error) {
 		return 0, errCorruptIndexEntry(err)
 	}
 	return includeInterpolatedValueBlocksSize(
-		endBH.Offset + endBH.Length + blockTrailerLen - startBH.Offset), nil
+		endBH.Offset + endBH.Length + block.TrailerLen - startBH.Offset), nil
 }
 
 // TableFormat returns the format version for the table.
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index 76ad840d..9efeb4cf 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -6,13 +6,13 @@ import (
 	"math"
 	"sync"
 
-	"github.com/cespare/xxhash/v2"
 	"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/internal/rangekey"
 	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 // RewriteKeySuffixesAndReturnFormat copies the content of the passed SSTable
@@ -152,7 +152,7 @@ type blockWithSpan struct {
 func rewriteBlocks(
 	r *Reader,
 	restartInterval int,
-	checksumType ChecksumType,
+	checksumType block.ChecksumType,
 	compression Compression,
 	input []BlockHandleWithProperties,
 	output []blockWithSpan,
@@ -163,11 +163,7 @@ func rewriteBlocks(
 	bw := blockWriter{
 		restartInterval: restartInterval,
 	}
-	buf := blockBuf{checksummer: checksummer{checksumType: checksumType}}
-	if checksumType == ChecksumTypeXXHash {
-		buf.checksummer.xxHasher = xxhash.New()
-	}
-
+	buf := blockBuf{checksummer: block.Checksummer{Type: checksumType}}
 	var blockAlloc bytealloc.A
 	var keyAlloc bytealloc.A
 	var scratch InternalKey
@@ -230,11 +226,11 @@ func rewriteBlocks(
 				if len(v) < 1 {
 					return errors.Errorf("value has no prefix")
 				}
-				prefix := valuePrefix(v[0])
-				if isValueHandle(prefix) {
+				prefix := block.ValuePrefix(v[0])
+				if prefix.IsValueHandle() {
 					return errors.Errorf("value prefix is incorrect")
 				}
-				if setHasSamePrefix(prefix) {
+				if prefix.SetHasSamePrefix() {
 					return errors.Errorf("multiple keys with same key prefix")
 				}
 			}
@@ -247,12 +243,12 @@ func rewriteBlocks(
 
 		keyAlloc, output[i].end = cloneKeyWithBuf(scratch, keyAlloc)
 
-		finished := compressAndChecksum(bw.finish(), compression, &buf)
+		finished, trailer := compressAndChecksum(bw.finish(), compression, &buf)
 
 		// copy our finished block into the output buffer.
-		blockAlloc, output[i].data = blockAlloc.Alloc(len(finished) + blockTrailerLen)
+		blockAlloc, output[i].data = blockAlloc.Alloc(len(finished) + block.TrailerLen)
 		copy(output[i].data, finished)
-		copy(output[i].data[len(finished):], buf.tmp[:blockTrailerLen])
+		copy(output[i].data[len(finished):], trailer[:])
 	}
 	return nil
 }
@@ -297,7 +293,7 @@ func rewriteDataBlocksToWriter(
 			err := rewriteBlocks(
 				r,
 				w.dataBlockBuf.dataBlock.restartInterval,
-				w.blockBuf.checksummer.checksumType,
+				w.blockBuf.checksummer.Type,
 				w.compression,
 				data,
 				blocks,
@@ -346,7 +342,7 @@ func rewriteDataBlocksToWriter(
 		}
 
 		n := len(blocks[i].data)
-		bh := BlockHandle{Offset: w.meta.Size, Length: uint64(n) - blockTrailerLen}
+		bh := BlockHandle{Offset: w.meta.Size, Length: uint64(n) - block.TrailerLen}
 		// Update the overall size.
 		w.meta.Size += uint64(n)
 
@@ -514,7 +510,7 @@ func NewMemReader(sst []byte, o ReaderOptions) (*Reader, error) {
 }
 
 func readBlockBuf(r *Reader, bh BlockHandle, buf []byte) ([]byte, []byte, error) {
-	raw := r.readable.(*memReader).b[bh.Offset : bh.Offset+bh.Length+blockTrailerLen]
+	raw := r.readable.(*memReader).b[bh.Offset : bh.Offset+bh.Length+block.TrailerLen]
 	if err := checkChecksum(r.checksumType, raw, bh, 0); err != nil {
 		return nil, buf, err
 	}
diff --git a/sstable/table.go b/sstable/table.go
index a64ed9ca..7fcac56b 100644
--- a/sstable/table.go
+++ b/sstable/table.go
@@ -74,6 +74,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 /*
@@ -184,7 +185,6 @@ Data blocks have some additional features:
 */
 
 const (
-	blockTrailerLen                    = 5
 	blockHandleMaxLenWithoutProperties = 10 + 10
 	// blockHandleLikelyMaxLen can be used for pre-allocating buffers to
 	// reduce memory copies. It is not guaranteed that a block handle will not
@@ -231,33 +231,6 @@ const (
 	rocksDBCompressionOptions = "window_bits=-14; level=32767; strategy=0; max_dict_bytes=0; zstd_max_train_bytes=0; enabled=0; "
 )
 
-// ChecksumType specifies the checksum used for blocks.
-type ChecksumType byte
-
-// The available checksum types.
-const (
-	ChecksumTypeNone     ChecksumType = 0
-	ChecksumTypeCRC32c   ChecksumType = 1
-	ChecksumTypeXXHash   ChecksumType = 2
-	ChecksumTypeXXHash64 ChecksumType = 3
-)
-
-// String implements fmt.Stringer.
-func (t ChecksumType) String() string {
-	switch t {
-	case ChecksumTypeCRC32c:
-		return "crc32c"
-	case ChecksumTypeNone:
-		return "none"
-	case ChecksumTypeXXHash:
-		return "xxhash"
-	case ChecksumTypeXXHash64:
-		return "xxhash64"
-	default:
-		panic(errors.Newf("sstable: unknown checksum type: %d", t))
-	}
-}
-
 type blockType byte
 
 const (
@@ -318,7 +291,7 @@ func (t blockType) String() string {
 //	table_magic_number (8 bytes)
 type footer struct {
 	format      TableFormat
-	checksum    ChecksumType
+	checksum    block.ChecksumType
 	metaindexBH BlockHandle
 	indexBH     BlockHandle
 	footerBH    BlockHandle
@@ -360,7 +333,7 @@ func readFooter(
 		buf = buf[len(buf)-levelDBFooterLen:]
 		footer.footerBH.Length = uint64(len(buf))
 		footer.format = TableFormatLevelDB
-		footer.checksum = ChecksumTypeCRC32c
+		footer.checksum = block.ChecksumTypeCRC32c
 
 	case rocksDBMagic, pebbleDBMagic:
 		// NOTE: The Pebble magic string implies the same footer format as that used
@@ -379,11 +352,11 @@ func readFooter(
 		}
 		footer.format = format
 
-		switch ChecksumType(buf[0]) {
-		case ChecksumTypeCRC32c:
-			footer.checksum = ChecksumTypeCRC32c
-		case ChecksumTypeXXHash64:
-			footer.checksum = ChecksumTypeXXHash64
+		switch block.ChecksumType(buf[0]) {
+		case block.ChecksumTypeCRC32c:
+			footer.checksum = block.ChecksumTypeCRC32c
+		case block.ChecksumTypeXXHash64:
+			footer.checksum = block.ChecksumTypeXXHash64
 		default:
 			return footer, base.CorruptionErrorf("pebble/table: unsupported checksum type %d", errors.Safe(footer.checksum))
 		}
@@ -424,14 +397,14 @@ func (f footer) encode(buf []byte) []byte {
 		buf = buf[:rocksDBFooterLen]
 		clear(buf)
 		switch f.checksum {
-		case ChecksumTypeNone:
-			buf[0] = byte(ChecksumTypeNone)
-		case ChecksumTypeCRC32c:
-			buf[0] = byte(ChecksumTypeCRC32c)
-		case ChecksumTypeXXHash:
-			buf[0] = byte(ChecksumTypeXXHash)
-		case ChecksumTypeXXHash64:
-			buf[0] = byte(ChecksumTypeXXHash64)
+		case block.ChecksumTypeNone:
+			buf[0] = byte(block.ChecksumTypeNone)
+		case block.ChecksumTypeCRC32c:
+			buf[0] = byte(block.ChecksumTypeCRC32c)
+		case block.ChecksumTypeXXHash:
+			buf[0] = byte(block.ChecksumTypeXXHash)
+		case block.ChecksumTypeXXHash64:
+			buf[0] = byte(block.ChecksumTypeXXHash64)
 		default:
 			panic("unknown checksum type")
 		}
diff --git a/sstable/value_block.go b/sstable/value_block.go
index 90303d68..b8fd6c28 100644
--- a/sstable/value_block.go
+++ b/sstable/value_block.go
@@ -105,7 +105,7 @@ import (
 //
 // Data blocks contain two kinds of SET keys: those with in-place values and
 // those with a value handle. To distinguish these two cases we use a single
-// byte prefix (valuePrefix). This single byte prefix is split into multiple
+// byte prefix (ValuePrefix). This single byte prefix is split into multiple
 // parts, where nb represents information that is encoded in n bits.
 //
 // +---------------+--------------------+-----------+--------------------+
@@ -189,31 +189,6 @@ type valueHandle struct {
 	offsetInBlock uint32
 }
 
-// valuePrefix is the single byte prefix for either the in-place value or the
-// encoded valueHandle. It encoded multiple kinds of information.
-type valuePrefix byte
-
-const (
-	// 2 most-significant bits of valuePrefix encodes the value-kind.
-	valueKindMask           valuePrefix = '\xC0'
-	valueKindIsValueHandle  valuePrefix = '\x80'
-	valueKindIsInPlaceValue valuePrefix = '\x00'
-
-	// 1 bit indicates SET has same key prefix as immediately preceding key that
-	// is also a SET. If the immediately preceding key in the same block is a
-	// SET, AND this bit is 0, the prefix must have changed.
-	//
-	// Note that the current policy of only storing older MVCC versions in value
-	// blocks means that valueKindIsValueHandle => SET has same prefix. But no
-	// code should rely on this behavior. Also, SET has same prefix does *not*
-	// imply valueKindIsValueHandle.
-	setHasSameKeyPrefixMask valuePrefix = '\x20'
-
-	// 3 least-significant bits for the user-defined base.ShortAttribute.
-	// Undefined for valueKindIsInPlaceValue.
-	userDefinedShortAttributeMask valuePrefix = '\x07'
-)
-
 // valueHandle fields are varint encoded, so maximum 5 bytes each, plus 1 byte
 // for the valuePrefix. This could alternatively be group varint encoded, but
 // experiments were inconclusive
@@ -231,35 +206,6 @@ func encodeValueHandle(dst []byte, v valueHandle) int {
 	return n
 }
 
-func makePrefixForValueHandle(setHasSameKeyPrefix bool, attribute base.ShortAttribute) valuePrefix {
-	prefix := valueKindIsValueHandle | valuePrefix(attribute)
-	if setHasSameKeyPrefix {
-		prefix = prefix | setHasSameKeyPrefixMask
-	}
-	return prefix
-}
-
-func makePrefixForInPlaceValue(setHasSameKeyPrefix bool) valuePrefix {
-	prefix := valueKindIsInPlaceValue
-	if setHasSameKeyPrefix {
-		prefix = prefix | setHasSameKeyPrefixMask
-	}
-	return prefix
-}
-
-func isValueHandle(b valuePrefix) bool {
-	return b&valueKindMask == valueKindIsValueHandle
-}
-
-// REQUIRES: isValueHandle(b)
-func getShortAttribute(b valuePrefix) base.ShortAttribute {
-	return base.ShortAttribute(b & userDefinedShortAttributeMask)
-}
-
-func setHasSamePrefix(b valuePrefix) bool {
-	return b&setHasSameKeyPrefixMask == setHasSameKeyPrefixMask
-}
-
 func decodeLenFromValueHandle(src []byte) (uint32, []byte) {
 	ptr := unsafe.Pointer(&src[0])
 	var v uint32
@@ -392,7 +338,7 @@ type valueBlockWriter struct {
 	// Configured compression.
 	compression Compression
 	// checksummer with configured checksum type.
-	checksummer checksummer
+	checksummer block.Checksummer
 	// Block finished callback.
 	blockFinishedFunc func(compressedSize int)
 
@@ -461,7 +407,7 @@ func newValueBlockWriter(
 	blockSize int,
 	blockSizeThreshold int,
 	compression Compression,
-	checksumType ChecksumType,
+	checksumType block.ChecksumType,
 	// compressedSize should exclude the block trailer.
 	blockFinishedFunc func(compressedSize int),
 ) *valueBlockWriter {
@@ -470,8 +416,8 @@ func newValueBlockWriter(
 		blockSize:          blockSize,
 		blockSizeThreshold: blockSizeThreshold,
 		compression:        compression,
-		checksummer: checksummer{
-			checksumType: checksumType,
+		checksummer: block.Checksummer{
+			Type: checksumType,
 		},
 		blockFinishedFunc: blockFinishedFunc,
 		buf:               uncompressedValueBlockBufPool.Get().(*blockBuffer),
@@ -564,12 +510,12 @@ func (w *valueBlockWriter) compressAndFlush() {
 		}
 	}
 	n := len(b.b)
-	if n+blockTrailerLen > cap(b.b) {
-		block := make([]byte, n+blockTrailerLen)
+	if n+block.TrailerLen > cap(b.b) {
+		block := make([]byte, n+block.TrailerLen)
 		copy(block, b.b)
 		b.b = block
 	} else {
-		b.b = b.b[:n+blockTrailerLen]
+		b.b = b.b[:n+block.TrailerLen]
 	}
 	b.b[n] = byte(blockType)
 	w.computeChecksum(b.b)
@@ -592,10 +538,10 @@ func (w *valueBlockWriter) compressAndFlush() {
 	w.buf.b = w.buf.b[:0]
 }
 
-func (w *valueBlockWriter) computeChecksum(block []byte) {
-	n := len(block) - blockTrailerLen
-	checksum := w.checksummer.checksum(block[:n], block[n:n+1])
-	binary.LittleEndian.PutUint32(block[n+1:], checksum)
+func (w *valueBlockWriter) computeChecksum(blk []byte) {
+	n := len(blk) - block.TrailerLen
+	checksum := w.checksummer.Checksum(blk[:n], blk[n:n+1])
+	binary.LittleEndian.PutUint32(blk[n+1:], checksum)
 }
 
 func (w *valueBlockWriter) finish(
@@ -638,7 +584,7 @@ func (w *valueBlockWriter) finish(
 	stats := valueBlocksAndIndexStats{
 		numValueBlocks:          uint64(n),
 		numValuesInValueBlocks:  w.numValues,
-		valueBlocksAndIndexSize: w.totalBlockBytes + vbih.h.Length + blockTrailerLen,
+		valueBlocksAndIndexSize: w.totalBlockBytes + vbih.h.Length + block.TrailerLen,
 	}
 	return vbih, stats, err
 }
@@ -649,7 +595,7 @@ func (w *valueBlockWriter) writeValueBlocksIndex(
 	blockLen :=
 		int(h.blockNumByteLength+h.blockOffsetByteLength+h.blockLengthByteLength) * len(w.blocks)
 	h.h.Length = uint64(blockLen)
-	blockLen += blockTrailerLen
+	blockLen += block.TrailerLen
 	var buf []byte
 	if cap(w.buf.b) < blockLen {
 		buf = make([]byte, blockLen)
@@ -666,7 +612,7 @@ func (w *valueBlockWriter) writeValueBlocksIndex(
 		littleEndianPut(w.blocks[i].handle.Length, b, int(h.blockLengthByteLength))
 		b = b[int(h.blockLengthByteLength):]
 	}
-	if len(b) != blockTrailerLen {
+	if len(b) != block.TrailerLen {
 		panic("incorrect length calculation")
 	}
 	b[0] = byte(noCompressionBlockType)
@@ -822,7 +768,7 @@ func (r *valueBlockReader) getLazyValueForPrefixAndValueHandle(handle []byte) ba
 		Fetcher: r,
 		Attribute: base.AttributeAndLen{
 			ValueLen:       int32(valLen),
-			ShortAttribute: getShortAttribute(valuePrefix(handle[0])),
+			ShortAttribute: block.ValuePrefix(handle[0]).ShortAttribute(),
 		},
 	}
 	if r.stats != nil {
diff --git a/sstable/write_queue.go b/sstable/write_queue.go
index 898f3275..68bf0737 100644
--- a/sstable/write_queue.go
+++ b/sstable/write_queue.go
@@ -63,7 +63,7 @@ func (w *writeQueue) performWrite(task *writeTask) error {
 	var bhp BlockHandleWithProperties
 
 	var err error
-	if bh, err = w.writer.writeCompressedBlock(task.buf.compressed, task.buf.tmp[:]); err != nil {
+	if bh, err = w.writer.writeCompressedBlock(task.buf.compressed, task.buf.trailer); err != nil {
 		return err
 	}
 
diff --git a/sstable/writer.go b/sstable/writer.go
index 84b34f0b..8cda449a 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -14,18 +14,17 @@ import (
 	"sort"
 	"sync"
 
-	"github.com/cespare/xxhash/v2"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/bytealloc"
 	"github.com/cockroachdb/pebble/internal/cache"
-	"github.com/cockroachdb/pebble/internal/crc"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/private"
 	"github.com/cockroachdb/pebble/internal/rangedel"
 	"github.com/cockroachdb/pebble/internal/rangekey"
 	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable/block"
 )
 
 // encodedBHPEstimatedSize estimates the size of the encoded BlockHandleWithProperties.
@@ -148,7 +147,7 @@ type Writer struct {
 	writingToLowestLevel bool
 	cache                *cache.Cache
 	restartInterval      int
-	checksumType         ChecksumType
+	checksumType         block.ChecksumType
 	// disableKeyOrderChecks disables the checks that keys are added to an
 	// sstable in order. It is intended for internal use only in the construction
 	// of invalid sstables for testing. See tool/make_test_sstables.go.
@@ -227,7 +226,7 @@ type Writer struct {
 }
 
 type pointKeyInfo struct {
-	trailer base.Trailer
+	trailer base.InternalKeyTrailer
 	// Only computed when w.valueBlockWriter is not nil.
 	userKeyLen int
 	// prefixLen uses w.split, if not nil. Only computed when w.valueBlockWriter
@@ -520,7 +519,7 @@ func (d *dataBlockEstimates) dataBlockCompressed(compressedSize int, inflightSiz
 		d.mu.Lock()
 		defer d.mu.Unlock()
 	}
-	d.estimate.writtenWithDelta(compressedSize+blockTrailerLen, inflightSize)
+	d.estimate.writtenWithDelta(compressedSize+block.TrailerLen, inflightSize)
 }
 
 // size is an estimated size of datablock data which has been written to disk.
@@ -559,31 +558,6 @@ var writeTaskPool = sync.Pool{
 	},
 }
 
-type checksummer struct {
-	checksumType ChecksumType
-	xxHasher     *xxhash.Digest
-}
-
-func (c *checksummer) checksum(block []byte, blockType []byte) (checksum uint32) {
-	// Calculate the checksum.
-	switch c.checksumType {
-	case ChecksumTypeCRC32c:
-		checksum = crc.New(block).Update(blockType).Value()
-	case ChecksumTypeXXHash64:
-		if c.xxHasher == nil {
-			c.xxHasher = xxhash.New()
-		} else {
-			c.xxHasher.Reset()
-		}
-		c.xxHasher.Write(block)
-		c.xxHasher.Write(blockType)
-		checksum = uint32(c.xxHasher.Sum64())
-	default:
-		panic(errors.Newf("unsupported checksum type: %d", c.checksumType))
-	}
-	return checksum
-}
-
 type blockBuf struct {
 	// tmp is a scratch buffer, large enough to hold either footerLen bytes,
 	// blockTrailerLen bytes, (5 * binary.MaxVarintLen64) bytes, and most
@@ -592,7 +566,7 @@ type blockBuf struct {
 	// compressedBuf is the destination buffer for compression. It is re-used over the
 	// lifetime of the blockBuf, avoiding the allocation of a temporary buffer for each block.
 	compressedBuf []byte
-	checksummer   checksummer
+	checksummer   block.Checksummer
 }
 
 func (b *blockBuf) clear() {
@@ -622,6 +596,8 @@ type dataBlockBuf struct {
 	// backed by the dataBlock.buf, or the dataBlockBuf.compressedBuf, depending on whether
 	// we use the result of the compression.
 	compressed []byte
+	// trailer is the block trailer encoding the compression type and checksum.
+	trailer block.Trailer
 
 	// We're making calls to BlockPropertyCollectors from the Writer client goroutine. We need to
 	// pass the encoded block properties over to the write queue. To prevent copies, and allocations,
@@ -651,10 +627,10 @@ var dataBlockBufPool = sync.Pool{
 	},
 }
 
-func newDataBlockBuf(restartInterval int, checksumType ChecksumType) *dataBlockBuf {
+func newDataBlockBuf(restartInterval int, checksumType block.ChecksumType) *dataBlockBuf {
 	d := dataBlockBufPool.Get().(*dataBlockBuf)
 	d.dataBlock.restartInterval = restartInterval
-	d.checksummer.checksumType = checksumType
+	d.checksummer.Type = checksumType
 	return d
 }
 
@@ -663,7 +639,7 @@ func (d *dataBlockBuf) finish() {
 }
 
 func (d *dataBlockBuf) compressAndChecksum(c Compression) {
-	d.compressed = compressAndChecksum(d.uncompressed, c, &d.blockBuf)
+	d.compressed, d.trailer = compressAndChecksum(d.uncompressed, c, &d.blockBuf)
 }
 
 func (d *dataBlockBuf) shouldFlush(
@@ -961,7 +937,7 @@ func (w *Writer) addPoint(key InternalKey, value []byte, forceObsolete bool) err
 	isObsolete = w.tableFormat >= TableFormatPebblev4 && (isObsolete || forceObsolete)
 	w.lastPointKeyInfo.isObsolete = isObsolete
 	var valueStoredWithKey []byte
-	var prefix valuePrefix
+	var prefix block.ValuePrefix
 	var valueStoredWithKeyLen int
 	if writeToValueBlock {
 		vh, err := w.valueBlockWriter.addValue(value)
@@ -982,14 +958,14 @@ func (w *Writer) addPoint(key InternalKey, value []byte, forceObsolete bool) err
 				return err
 			}
 		}
-		prefix = makePrefixForValueHandle(setHasSameKeyPrefix, attribute)
+		prefix = block.ValueHandlePrefix(setHasSameKeyPrefix, attribute)
 	} else {
 		valueStoredWithKey = value
 		valueStoredWithKeyLen = len(value)
 		if addPrefixToValueStoredWithKey {
 			valueStoredWithKeyLen++
 		}
-		prefix = makePrefixForInPlaceValue(setHasSameKeyPrefix)
+		prefix = block.InPlaceValuePrefix(setHasSameKeyPrefix)
 	}
 
 	if err := w.maybeFlush(key, valueStoredWithKeyLen); err != nil {
@@ -1822,12 +1798,14 @@ func (w *Writer) writeTwoLevelIndex() (BlockHandle, error) {
 	// index size property.
 	w.props.IndexPartitions = uint64(len(w.indexPartitions))
 	w.props.TopLevelIndexSize = uint64(w.topLevelIndexBlock.estimatedSize())
-	w.props.IndexSize += w.props.TopLevelIndexSize + blockTrailerLen
+	w.props.IndexSize += w.props.TopLevelIndexSize + block.TrailerLen
 
 	return w.writeBlock(w.topLevelIndexBlock.finish(), w.compression, &w.blockBuf)
 }
 
-func compressAndChecksum(b []byte, compression Compression, blockBuf *blockBuf) []byte {
+func compressAndChecksum(
+	b []byte, compression Compression, blockBuf *blockBuf,
+) (compressed []byte, trailer block.Trailer) {
 	// Compress the buffer, discarding the result if the improvement isn't at
 	// least 12.5%.
 	blockType, compressed := compressBlock(compression, b, blockBuf.compressedBuf)
@@ -1840,16 +1818,14 @@ func compressAndChecksum(b []byte, compression Compression, blockBuf *blockBuf)
 		blockType = noCompressionBlockType
 	}
 
-	blockBuf.tmp[0] = byte(blockType)
-
 	// Calculate the checksum.
-	checksum := blockBuf.checksummer.checksum(b, blockBuf.tmp[:1])
-	binary.LittleEndian.PutUint32(blockBuf.tmp[1:5], checksum)
-	return b
+	trailer[0] = byte(blockType)
+	checksum := blockBuf.checksummer.Checksum(b, trailer[:1])
+	return b, block.MakeTrailer(byte(blockType), checksum)
 }
 
-func (w *Writer) writeCompressedBlock(block []byte, blockTrailerBuf []byte) (BlockHandle, error) {
-	bh := BlockHandle{Offset: w.meta.Size, Length: uint64(len(block))}
+func (w *Writer) writeCompressedBlock(blk []byte, trailer block.Trailer) (BlockHandle, error) {
+	bh := BlockHandle{Offset: w.meta.Size, Length: uint64(len(blk))}
 
 	if w.cacheID != 0 && w.fileNum != 0 {
 		// Remove the block being written from the cache. This provides defense in
@@ -1861,14 +1837,14 @@ func (w *Writer) writeCompressedBlock(block []byte, blockTrailerBuf []byte) (Blo
 	}
 
 	// Write the bytes to the file.
-	if err := w.writable.Write(block); err != nil {
+	if err := w.writable.Write(blk); err != nil {
 		return BlockHandle{}, err
 	}
-	w.meta.Size += uint64(len(block))
-	if err := w.writable.Write(blockTrailerBuf[:blockTrailerLen]); err != nil {
+	w.meta.Size += uint64(len(blk))
+	if err := w.writable.Write(trailer[:]); err != nil {
 		return BlockHandle{}, err
 	}
-	w.meta.Size += blockTrailerLen
+	w.meta.Size += block.TrailerLen
 
 	return bh, nil
 }
@@ -1896,8 +1872,8 @@ func (w *Writer) Write(blockWithTrailer []byte) (n int, err error) {
 func (w *Writer) writeBlock(
 	b []byte, compression Compression, blockBuf *blockBuf,
 ) (BlockHandle, error) {
-	b = compressAndChecksum(b, compression, blockBuf)
-	return w.writeCompressedBlock(b, blockBuf.tmp[:])
+	b, trailer := compressAndChecksum(b, compression, blockBuf)
+	return w.writeCompressedBlock(b, trailer)
 }
 
 // assertFormatCompatibility ensures that the features present on the table are
@@ -2058,7 +2034,7 @@ func (w *Writer) Close() (err error) {
 		// NB: RocksDB includes the block trailer length in the index size
 		// property, though it doesn't include the trailer in the filter size
 		// property.
-		w.props.IndexSize = uint64(w.indexBlock.estimatedSize()) + blockTrailerLen
+		w.props.IndexSize = uint64(w.indexBlock.estimatedSize()) + block.TrailerLen
 		w.props.NumDataBlocks = uint64(w.indexBlock.block.nEntries)
 
 		// Write the single level index block.
@@ -2209,7 +2185,7 @@ func (w *Writer) Close() (err error) {
 	// Write the table footer.
 	footer := footer{
 		format:      w.tableFormat,
-		checksum:    w.blockBuf.checksummer.checksumType,
+		checksum:    w.blockBuf.checksummer.Type,
 		metaindexBH: metaindexBH,
 		indexBH:     indexBH,
 	}
@@ -2333,7 +2309,7 @@ func NewWriter(writable objstorage.Writable, o WriterOptions, extraOpts ...Write
 	w.dataBlockBuf = newDataBlockBuf(w.restartInterval, w.checksumType)
 
 	w.blockBuf = blockBuf{
-		checksummer: checksummer{checksumType: o.Checksum},
+		checksummer: block.Checksummer{Type: o.Checksum},
 	}
 
 	w.coordination.init(o.Parallelism, w)


Test patch:
diff --git a/sstable/block/kv_test.go b/sstable/block/kv_test.go
new file mode 100644
index 00000000..a1693369
--- /dev/null
+++ b/sstable/block/kv_test.go
@@ -0,0 +1,55 @@
+// 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 block
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/stretchr/testify/require"
+)
+
+func TestValuePrefix(t *testing.T) {
+	testCases := []struct {
+		isHandle         bool
+		setHasSamePrefix bool
+		attr             base.ShortAttribute
+	}{
+		{
+			isHandle:         false,
+			setHasSamePrefix: false,
+		},
+		{
+			isHandle:         false,
+			setHasSamePrefix: true,
+		},
+		{
+			isHandle:         true,
+			setHasSamePrefix: false,
+			attr:             5,
+		},
+		{
+			isHandle:         true,
+			setHasSamePrefix: true,
+			attr:             2,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(fmt.Sprintf("%+v", tc), func(t *testing.T) {
+			var prefix ValuePrefix
+			if tc.isHandle {
+				prefix = ValueHandlePrefix(tc.setHasSamePrefix, tc.attr)
+			} else {
+				prefix = InPlaceValuePrefix(tc.setHasSamePrefix)
+			}
+			require.Equal(t, tc.isHandle, prefix.IsValueHandle())
+			require.Equal(t, tc.setHasSamePrefix, prefix.SetHasSamePrefix())
+			if tc.isHandle {
+				require.Equal(t, tc.attr, prefix.ShortAttribute())
+			}
+		})
+	}
+}
diff --git a/sstable/block_test.go b/sstable/block_test.go
index 8f038b58..abe94233 100644
--- a/sstable/block_test.go
+++ b/sstable/block_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/itertest"
 	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
 )
@@ -54,7 +55,7 @@ func TestBlockWriterWithPrefix(t *testing.T) {
 		key InternalKey,
 		value []byte,
 		addValuePrefix bool,
-		valuePrefix valuePrefix,
+		valuePrefix block.ValuePrefix,
 		setHasSameKeyPrefix bool) {
 		w.addWithOptionalValuePrefix(
 			key, false, value, len(key.UserKey), addValuePrefix, valuePrefix, setHasSameKeyPrefix)
@@ -147,7 +148,7 @@ func TestInvalidInternalKeyDecoding(t *testing.T) {
 		i := blockIter{}
 		i.decodeInternalKey([]byte(tc))
 		require.Nil(t, i.ikv.K.UserKey)
-		require.Equal(t, base.Trailer(InternalKeyKindInvalid), i.ikv.K.Trailer)
+		require.Equal(t, base.InternalKeyTrailer(InternalKeyKindInvalid), i.ikv.K.Trailer)
 	}
 }
 
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index c7300357..f3f6a39e 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -1411,7 +1411,7 @@ func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
 }
 
 func TestReaderChecksumErrors(t *testing.T) {
-	for _, checksumType := range []ChecksumType{ChecksumTypeCRC32c, ChecksumTypeXXHash64} {
+	for _, checksumType := range []block.ChecksumType{block.ChecksumTypeCRC32c, block.ChecksumTypeXXHash64} {
 		t.Run(fmt.Sprintf("checksum-type=%d", checksumType), func(t *testing.T) {
 			for _, twoLevelIndex := range []bool{false, true} {
 				t.Run(fmt.Sprintf("two-level-index=%t", twoLevelIndex), func(t *testing.T) {
diff --git a/sstable/table_test.go b/sstable/table_test.go
index afe9ae2a..c1a70544 100644
--- a/sstable/table_test.go
+++ b/sstable/table_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/pebble/bloom"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/kr/pretty"
 	"github.com/stretchr/testify/require"
@@ -551,9 +552,9 @@ func TestFooterRoundTrip(t *testing.T) {
 	buf := make([]byte, 100+maxFooterLen)
 	for format := TableFormatLevelDB; format < TableFormatMax; format++ {
 		t.Run(fmt.Sprintf("format=%s", format), func(t *testing.T) {
-			checksums := []ChecksumType{ChecksumTypeCRC32c}
+			checksums := []block.ChecksumType{block.ChecksumTypeCRC32c}
 			if format != TableFormatLevelDB {
-				checksums = []ChecksumType{ChecksumTypeCRC32c, ChecksumTypeXXHash64}
+				checksums = []block.ChecksumType{block.ChecksumTypeCRC32c, block.ChecksumTypeXXHash64}
 			}
 			for _, checksum := range checksums {
 				t.Run(fmt.Sprintf("checksum=%d", checksum), func(t *testing.T) {
@@ -603,7 +604,7 @@ func TestFooterRoundTrip(t *testing.T) {
 }
 
 func TestReadFooter(t *testing.T) {
-	encode := func(format TableFormat, checksum ChecksumType) string {
+	encode := func(format TableFormat, checksum block.ChecksumType) string {
 		f := footer{
 			format:   format,
 			checksum: checksum,
@@ -620,8 +621,8 @@ func TestReadFooter(t *testing.T) {
 		{strings.Repeat("a", rocksDBFooterLen), "bad magic number"},
 		{encode(TableFormatLevelDB, 0)[1:], "file size is too small"},
 		{encode(TableFormatRocksDBv2, 0)[1:], "footer too short"},
-		{encode(TableFormatRocksDBv2, ChecksumTypeNone), "unsupported checksum type"},
-		{encode(TableFormatRocksDBv2, ChecksumTypeXXHash), "unsupported checksum type"},
+		{encode(TableFormatRocksDBv2, block.ChecksumTypeNone), "unsupported checksum type"},
+		{encode(TableFormatRocksDBv2, block.ChecksumTypeXXHash), "unsupported checksum type"},
 	}
 	for _, c := range testCases {
 		t.Run("", func(t *testing.T) {
diff --git a/sstable/value_block_test.go b/sstable/value_block_test.go
index f292c8ac..1dea47ab 100644
--- a/sstable/value_block_test.go
+++ b/sstable/value_block_test.go
@@ -10,7 +10,6 @@ import (
 	"math/rand"
 	"testing"
 
-	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/stretchr/testify/require"
 )
 
@@ -29,48 +28,6 @@ func TestValueHandleEncodeDecode(t *testing.T) {
 	}
 }
 
-func TestValuePrefix(t *testing.T) {
-	testCases := []struct {
-		isHandle         bool
-		setHasSamePrefix bool
-		attr             base.ShortAttribute
-	}{
-		{
-			isHandle:         false,
-			setHasSamePrefix: false,
-		},
-		{
-			isHandle:         false,
-			setHasSamePrefix: true,
-		},
-		{
-			isHandle:         true,
-			setHasSamePrefix: false,
-			attr:             5,
-		},
-		{
-			isHandle:         true,
-			setHasSamePrefix: true,
-			attr:             2,
-		},
-	}
-	for _, tc := range testCases {
-		t.Run(fmt.Sprintf("%+v", tc), func(t *testing.T) {
-			var prefix valuePrefix
-			if tc.isHandle {
-				prefix = makePrefixForValueHandle(tc.setHasSamePrefix, tc.attr)
-			} else {
-				prefix = makePrefixForInPlaceValue(tc.setHasSamePrefix)
-			}
-			require.Equal(t, tc.isHandle, isValueHandle(prefix))
-			require.Equal(t, tc.setHasSamePrefix, setHasSamePrefix(prefix))
-			if tc.isHandle {
-				require.Equal(t, tc.attr, getShortAttribute(prefix))
-			}
-		})
-	}
-}
-
 func TestValueBlocksIndexHandleEncodeDecode(t *testing.T) {
 	testCases := []valueBlocksIndexHandle{
 		{
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 1d72f87e..3c4c5973 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -343,10 +344,10 @@ func TestWriterWithValueBlocks(t *testing.T) {
 			for valid := iter.First(); valid; valid = iter.Next() {
 				v := iter.Value()
 				if iter.Key().Kind() == InternalKeyKindSet {
-					prefix := valuePrefix(v[0])
-					setWithSamePrefix := setHasSamePrefix(prefix)
-					if isValueHandle(prefix) {
-						attribute := getShortAttribute(prefix)
+					prefix := block.ValuePrefix(v[0])
+					setWithSamePrefix := prefix.SetHasSamePrefix()
+					if prefix.IsValueHandle() {
+						attribute := prefix.ShortAttribute()
 						vh := decodeValueHandle(v[1:])
 						fmt.Fprintf(&buf, "%s:value-handle len %d block %d offset %d, att %d, same-pre %t\n",
 							iter.Key(), vh.valueLen, vh.blockNum, vh.offsetInBlock, attribute, setWithSamePrefix)
@@ -434,7 +435,7 @@ func TestBlockBufClear(t *testing.T) {
 }
 
 func TestClearDataBlockBuf(t *testing.T) {
-	d := newDataBlockBuf(1, ChecksumTypeCRC32c)
+	d := newDataBlockBuf(1, block.ChecksumTypeCRC32c)
 	d.blockBuf.compressedBuf = make([]byte, 1)
 	d.dataBlock.add(ikey("apple"), nil)
 	d.dataBlock.add(ikey("banana"), nil)