Repository: cockroachdb/pebble
Pull request #3731 (2024-07-03 15:52:04)

colblk: add UintBuilder, DeltaEncoding
Add a UintBuilder column writer for building uint{8,16,32,64} columns. The UintBuilder will use a delta encoding when possible to represent an array of values as a constant C and an array of deltas using a smaller-width uint. This is expected to reduce the size of blocks in many instances. Internal key trailers have zero sequence numbers in ingested sstables, and almost always have zero sequence numbers in L6 sstables. This will allow most data blocks in these sstables to encode either a single 8-byte constant, or an 8-byte constant and an array of 1-byte deltas if the key kinds vary.

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


Patch:
diff --git a/internal/aligned/aligned.go b/internal/aligned/aligned.go
new file mode 100644
index 00000000..eaf4f99d
--- /dev/null
+++ b/internal/aligned/aligned.go
@@ -0,0 +1,26 @@
+// 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 aligned
+
+import (
+	"fmt"
+	"unsafe"
+)
+
+// ByteSlice allocates a new byte slice of length n, ensuring the address of the
+// beginning of the slice is word aligned. Go does not guarantee that a simple
+// make([]byte, n) is aligned. In practice it often is, especially for larger n,
+// but small n can often be misaligned.
+func ByteSlice(n int) []byte {
+	a := make([]uint64, (n+7)/8)
+	b := unsafe.Slice((*byte)(unsafe.Pointer(&a[0])), n)
+
+	// Verify alignment.
+	ptr := uintptr(unsafe.Pointer(&b[0]))
+	if ptr%unsafe.Sizeof(int(0)) != 0 {
+		panic(fmt.Sprintf("allocated []uint64 slice not %d-aligned: pointer %p", unsafe.Sizeof(int(0)), &b[0]))
+	}
+	return b
+}
diff --git a/internal/binfmt/binfmt.go b/internal/binfmt/binfmt.go
index 143ad05c..c8571c25 100644
--- a/internal/binfmt/binfmt.go
+++ b/internal/binfmt/binfmt.go
@@ -59,18 +59,18 @@ func (f *Formatter) Offset() int {
 	return f.off
 }
 
-// PeekInt reads a little-endian integer of the specified width at the current
-// offset.
-func (f *Formatter) PeekInt(w int) int {
+// PeekUint reads a little-endian unsigned integer of the specified width at the
+// current offset.
+func (f *Formatter) PeekUint(w int) uint64 {
 	switch w {
 	case 1:
-		return int(f.data[f.off])
+		return uint64(f.data[f.off])
 	case 2:
-		return int(binary.LittleEndian.Uint16(f.data[f.off:]))
+		return uint64(binary.LittleEndian.Uint16(f.data[f.off:]))
 	case 4:
-		return int(binary.LittleEndian.Uint32(f.data[f.off:]))
+		return uint64(binary.LittleEndian.Uint32(f.data[f.off:]))
 	case 8:
-		return int(binary.LittleEndian.Uint64(f.data[f.off:]))
+		return binary.LittleEndian.Uint64(f.data[f.off:])
 	default:
 		panic("unsupported width")
 	}
diff --git a/sstable/colblk/base.go b/sstable/colblk/base.go
index 4bdc5ca4..9dfbbc78 100644
--- a/sstable/colblk/base.go
+++ b/sstable/colblk/base.go
@@ -2,7 +2,6 @@
 // of this source code is governed by a BSD-style license that can be found in
 // the LICENSE file.
 
-// Package colblk implements a columnar block format.
 package colblk
 
 import "golang.org/x/exp/constraints"
diff --git a/sstable/colblk/block.go b/sstable/colblk/block.go
new file mode 100644
index 00000000..7f30bd36
--- /dev/null
+++ b/sstable/colblk/block.go
@@ -0,0 +1,16 @@
+// 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 colblk implements a columnar block format.
+//
+// TODO(jackson): Describe the format once we start merging the higher-level
+// block format.
+//
+// # Alignment
+//
+// Block buffers are required to be word-aligned, during encoding and decoding.
+// This ensures that if any individual column or piece of data requires
+// word-alignment, the writer can align the offset into the block buffer
+// appropriately to ensure that the data is word-aligned.
+package colblk
diff --git a/sstable/colblk/column.go b/sstable/colblk/column.go
index b0ed8f8f..c4bbb152 100644
--- a/sstable/colblk/column.go
+++ b/sstable/colblk/column.go
@@ -56,6 +56,17 @@ func (t DataType) String() string {
 	return dataTypeName[t]
 }
 
+func (t DataType) uintWidth() uint32 {
+	if t >= DataTypeUint8 && t <= DataTypeUint64 {
+		rv := 1 << (t - DataTypeUint8)
+		if rv > 8 {
+			panic("width greater than 8 bytes")
+		}
+		return uint32(rv)
+	}
+	panic("not a unit")
+}
+
 // ColumnDesc describes the column's data type and its encoding.
 type ColumnDesc struct {
 	DataType DataType
@@ -93,30 +104,92 @@ func (c ColumnDescs) String() string {
 
 // ColumnEncoding describes the encoding of a column.
 //
-// Null bitmap (bit 0):
+// Column data can optionally contain a NULL bitmap, where each bit corresponds
+// to a row in the column and a set bit indicates that the corresponding row is
+// NULL. Column data can optionally use a delta encoding where each value is a
+// delta from a base, constant value.
+//
+// The layout of the encoded data is as follows:
+//   - [optional] NULL bitmap
+//   - [optional] a "base" constant value for delta encodings
+//   - n data values
 //
-// The bit at position 0 indicates whether the column is prefixed with a null
-// bitmap. A null bitmap is a bitmap where each bit corresponds to a row in the
-// column, and a set bit indicates that the corresponding row is NULL.
+// The ColumnEncoding byte is a combination of the following values:
+//   - The bits 0,1,2 encode the DeltaEncoding enum variant.
+//   - The bit 3 is set if the column has a NULL bitmap.
 //
-// TODO(jackson): Add additional column encoding types.
+// The delta encoding is currently only used for uint columns.
 type ColumnEncoding uint8
 
 const (
 	// EncodingDefault indicates that the default encoding is in-use for a
-	// column, encoding n values for n rows.
+	// column, encoding n values for n rows. In EncodingDefault, no null bitmap
+	// is set, no delta encoding is used, etc.
 	EncodingDefault ColumnEncoding = 0
+
+	encodingUintDeltaMask        = 0b00000111
+	encodingUintDeltaInverseMask = 0b11111000
+	encodingNullBitmapBit        = 0b00001000
+
 	// TODO(jackson): Add additional encoding types.
-	encodingTypeCount = 1
 )
 
+// Delta returns the delta encoding of the column.
+func (e ColumnEncoding) Delta() DeltaEncoding {
+	return DeltaEncoding(e & encodingUintDeltaMask)
+}
+
+// WithDelta returns the column encoding with the provided delta encoding.
+func (e ColumnEncoding) WithDelta(d DeltaEncoding) ColumnEncoding {
+	return (e & encodingUintDeltaInverseMask) | ColumnEncoding(d)
+}
+
 // String returns the string representation of the column encoding.
 func (e ColumnEncoding) String() string {
-	return encodingName[e]
+	var sb strings.Builder
+	fmt.Fprint(&sb, e.Delta().String())
+	return sb.String()
 }
 
-var encodingName [encodingTypeCount]string = [encodingTypeCount]string{
-	EncodingDefault: "default",
+// DeltaEncoding indicates what delta encoding, if any is in use by a column to
+// reduce the per-row storage size.
+//
+// A uint delta encoding represents every non-NULL element in an array of uints
+// as a delta relative to the column's constant. The logical value of each row
+// is computed as C + D[i] where C is the column constant and D[i] is the delta.
+type DeltaEncoding uint8
+
+const (
+	// DeltaEncodingNone indicates no delta encoding is in use. N rows are
+	// represented using N values of the column's logical data type.
+	DeltaEncodingNone DeltaEncoding = 0
+	// DeltaEncodingConstant indicates that all rows of the column share the
+	// same value. The column data encodes the constant value and no deltas.
+	DeltaEncodingConstant DeltaEncoding = 1
+	// DeltaEncodingUint8 indicates each delta is represented as a 1-byte uint8.
+	DeltaEncodingUint8 DeltaEncoding = 2
+	// DeltaEncodingUint16 indicates each delta is represented as a 2-byte uint16.
+	DeltaEncodingUint16 DeltaEncoding = 3
+	// DeltaEncodingUint32 indicates each delta is represented as a 4-byte uint32.
+	DeltaEncodingUint32 DeltaEncoding = 4
+)
+
+// String implements fmt.Stringer.
+func (d DeltaEncoding) String() string {
+	switch d {
+	case DeltaEncodingNone:
+		return "none"
+	case DeltaEncodingConstant:
+		return "const"
+	case DeltaEncodingUint8:
+		return "delta8"
+	case DeltaEncodingUint16:
+		return "delta16"
+	case DeltaEncodingUint32:
+		return "delta32"
+	default:
+		panic("unreachable")
+	}
 }
 
 // ColumnWriter is an interface implemented by column encoders that accumulate a
@@ -137,6 +210,10 @@ type ColumnWriter interface {
 	//
 	// The provided column index must be less than NumColumns(). Finish is
 	// called for each index < NumColumns() in order.
+	//
+	// The provided buf must be word-aligned (at offset 0). If a column writer
+	// requires a particularly alignment, it's responsible for padding offset
+	// appropriately first.
 	Finish(col int, rows int, offset uint32, buf []byte) (nextOffset uint32, desc ColumnDesc)
 }
 
diff --git a/sstable/colblk/uints.go b/sstable/colblk/uints.go
new file mode 100644
index 00000000..c81bbdd5
--- /dev/null
+++ b/sstable/colblk/uints.go
@@ -0,0 +1,356 @@
+// 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 colblk
+
+import (
+	"fmt"
+	"io"
+	"unsafe"
+
+	"github.com/cockroachdb/pebble/internal/binfmt"
+	"github.com/cockroachdb/pebble/internal/invariants"
+	"golang.org/x/exp/constraints"
+)
+
+// Uint is a constraint that permits any unsigned integer type with an
+// explicit size.
+type Uint interface {
+	~uint8 | ~uint16 | ~uint32 | ~uint64
+}
+
+// UintBuilder builds a column of unsigned integers of the same width.
+// UintBuilder uses a delta encoding when possible to store values using
+// lower-width integers. See DeltaEncoding.
+type UintBuilder[T Uint] struct {
+	// configuration fixed on Init; preserved across Reset
+	dt         DataType
+	useDefault bool
+
+	// array holds the underlying heap-allocated array in which values are
+	// stored.
+	array struct {
+		// n is the size of the array (in count of T elements; not bytes). n is
+		// NOT the number of elements that have been populated by the user.
+		n int
+		// elems provides access to elements without bounds checking. elems is
+		// grown automatically in Set.
+		elems UnsafeRawSlice[T]
+	}
+	// delta holds state for the purpose of tracking which DeltaEncoding would
+	// be used if the caller Finished the column including all elements Set so
+	// far. The delta state is used by Size (and Finish) to cheaply determine
+	// which encoding may most concisely encode the array.
+	//
+	// Every Set(i, v) call updates minimum and maximum if necessary. If a call
+	// updates minimum, maximum or both, it sets the width to the number of
+	// bytes necessary to represent the new difference between maximum and
+	// minimum. It also sets widthRow=i, indicating which row last updated the
+	// width.
+	//
+	// Any call to Size or Finish that supplies [rows] that's inclusive of the
+	// index stored in widthRow may use the stored width. Calls with fewer
+	// [rows] must recompute the min/max. In expected usage, only Finish will be
+	// called with fewer rows and only with one less row than has been set,
+	// meaning that only if the last row updated the width is a recomputation
+	// necessary.
+	//
+	// TODO(jackson): There is a small discrete set of possible encodings, so we
+	// could instead track the index of the first row that makes each encoding
+	// impossible. This would allow us to avoid recomputing the min/max in all
+	// cases. Or, if we limit the API to only allow Finish to be called with one
+	// less than the last set row, we could maintain the width of only the last
+	// two rows.
+	delta struct {
+		minimum  T
+		maximum  T
+		width    uint32 // 0, 1, 2, 4, or 8
+		widthRow int    // index of last update to width
+	}
+}
+
+// Init initializes the UintBuilder.
+func (b *UintBuilder[T]) Init() {
+	b.init(false)
+}
+
+// InitWithDefault initializes the UintBuilder. Any rows that are not explicitly
+// set are assumed to be zero. For the purpose of determining whether a delta
+// encoding is possible, the column is assumed to contain at least 1 default
+// value.
+//
+// InitWithDefault may be preferrable when a nonzero value is uncommon, and the
+// caller can avoid explicitly Set-ing every zero value.
+func (b *UintBuilder[T]) InitWithDefault() {
+	b.init(true)
+}
+
+func (b *UintBuilder[T]) init(useDefault bool) {
+	b.useDefault = useDefault
+	switch unsafe.Sizeof(T(0)) {
+	case 1:
+		b.dt = DataTypeUint8
+	case 2:
+		b.dt = DataTypeUint16
+	case 4:
+		b.dt = DataTypeUint32
+	case 8:
+		b.dt = DataTypeUint64
+	default:
+		panic("unreachable")
+	}
+	b.Reset()
+}
+
+// NumColumns implements ColumnWriter.
+func (b *UintBuilder[T]) NumColumns() int { return 1 }
+
+// Reset implements ColumnWriter and resets the builder, reusing existing
+// allocated memory.
+func (b *UintBuilder[T]) Reset() {
+	if b.useDefault {
+		// If the caller configured a default zero, we assume that the array
+		// will include at least one default value.
+		b.delta.minimum = 0
+		b.delta.maximum = 0
+		clear(b.array.elems.Slice(b.array.n))
+	} else {
+		// Initialize the minimum to the max value that a T can represent. We
+		// subtract from zero, relying on the fact that T is unsigned and will
+		// wrap around to the maximal value.
+		b.delta.minimum = T(0) - 1
+		b.delta.maximum = 0
+		// We could reset all values as a precaution, but it has a visible cost
+		// in benchmarks.
+		if invariants.Sometimes(50) {
+			for i := 0; i < b.array.n; i++ {
+				b.array.elems.set(i, T(0)-1)
+			}
+		}
+	}
+	b.delta.widthRow = 0
+	b.delta.width = 0
+}
+
+// Set sets the value of the provided row index to v.
+func (b *UintBuilder[T]) Set(row int, v T) {
+	if b.array.n <= row {
+		// Double the size of the allocated array, or initialize it to at least
+		// 256 bytes if this is the first allocation. Then double until there's
+		// sufficient space for n bytes.
+		n2 := max(b.array.n<<1, 256/int(unsafe.Sizeof(T(0))))
+		for n2 <= row {
+			n2 <<= 1 /* double the size */
+		}
+		// NB: Go guarantees the allocated array will be T-aligned.
+		newDataTyped := make([]T, n2)
+		copy(newDataTyped, b.array.elems.Slice(b.array.n))
+		newElems := makeUnsafeRawSlice[T](unsafe.Pointer(&newDataTyped[0]))
+		b.array.n = n2
+		b.array.elems = newElems
+
+	}
+	// Maintain the running minimum and maximum for the purpose of maintaining
+	// knowledge of the delta encoding that would be used.
+	if b.delta.minimum > v || b.delta.maximum < v {
+		b.delta.minimum = min(v, b.delta.minimum)
+		b.delta.maximum = max(v, b.delta.maximum)
+		// If updating the minimum and maximum means that we now much use a
+		// wider width integer, update the width and the index of the update to
+		// it.
+		if w := deltaWidth(uint64(b.delta.maximum - b.delta.minimum)); w != b.delta.width {
+			b.delta.width = w
+			b.delta.widthRow = row
+		}
+	}
+	b.array.elems.set(row, v)
+}
+
+// Size implements ColumnWriter and returns the size of the column if its first
+// [rows] rows were serialized, serializing the column into offset [offset].
+func (b *UintBuilder[T]) Size(rows int, offset uint32) uint32 {
+	if rows == 0 {
+		return 0
+	}
+	// Determine the width of each element with delta-encoding applied.
+	// b.delta.width is the precomputed width for all rows. It's the best
+	// encoding we can use as long as b.delta.widthRow is included. If
+	// b.delta.widthRow is not included (b.delta.widthRow > rows-1), we need to
+	// scan the [rows] elements of the array to recalculate the appropriate
+	// delta.
+	w := b.delta.width
+	if b.delta.widthRow > rows-1 {
+		minimum, maximum := computeMinMax(b.array.elems.Slice(rows))
+		w = deltaWidth(uint64(maximum - minimum))
+	}
+	return uintColumnSize[T](uint32(rows), offset, w)
+}
+
+// uintColumnSize returns the size of a column of unsigned integers of type T,
+// encoded at the provided offset using the provided width. If width <
+// sizeof(T), then a delta encoding is assumed.
+func uintColumnSize[T Uint](rows, offset, width uint32) uint32 {
+	logicalWidth := uint32(unsafe.Sizeof(T(0)))
+	// Include alignment bytes necessary to align offset appropriately for
+	// elements of type T.
+	offset = align(offset, logicalWidth)
+	if width != logicalWidth {
+		// A delta encoding will be used. We need to first account for the constant
+		// that encodes the minimum. This constant is the full width of the column's
+		// logical data type.
+		offset += logicalWidth
+	}
+	// Now account for the array of [rows] x w elements encoding the deltas.
+	return offset + rows*width
+}
+
+// Finish implements ColumnWriter, serializing the column into offset [offset] of
+// [buf].
+func (b *UintBuilder[T]) Finish(col, rows int, offset uint32, buf []byte) (uint32, ColumnDesc) {
+	desc := ColumnDesc{DataType: b.dt}
+	if rows == 0 {
+		return offset, desc
+	}
+
+	// Determine the width of each element with delta-encoding applied.
+	// b.delta.width is the precomputed width for all rows. It's the best
+	// encoding we can use as long as b.delta.widthRow is included. If
+	// b.delta.widthRow is not included (b.delta.widthRow > rows-1), we need to
+	// scan the [rows] elements of the array to recalculate the appropriate
+	// delta.
+	minimum := b.delta.minimum
+	w := b.delta.width
+	if b.delta.widthRow > rows-1 {
+		var maximum T
+		minimum, maximum = computeMinMax(b.array.elems.Slice(rows))
+		w = deltaWidth(uint64(maximum - minimum))
+	}
+	offset, desc.Encoding = uintColumnFinish[T](minimum, b.array.elems.Slice(rows), w, offset, buf)
+	return offset, desc
+}
+
+// uintColumnFinish finishes the column of unsigned integers of type T, encoding
+// per-row deltas of size width if width < sizeof(T).
+func uintColumnFinish[T Uint](
+	minimum T, values []T, width, offset uint32, buf []byte,
+) (uint32, ColumnEncoding) {
+	enc := EncodingDefault
+	// Align the offset appropriately for elements of type T.
+	offset = alignWithZeroes(buf, offset, uint32(unsafe.Sizeof(T(0))))
+
+	// Compare the computed delta width to see if we're able to use an array of
+	// lower-width deltas to encode the column.
+	if uintptr(width) < unsafe.Sizeof(T(0)) {
+		// Regardless of the width, we encode a constant of size T encoding the
+		// minimum across all the values.
+		dest := makeUnsafeRawSlice[T](unsafe.Pointer(&buf[offset]))
+		dest.set(0, minimum)
+		offset += uint32(unsafe.Sizeof(T(0)))
+
+		switch width {
+		case 0:
+			// All the column values are the same and we can elide any deltas at
+			// all.
+			enc = enc.WithDelta(DeltaEncodingConstant)
+			return offset, enc
+		case 1:
+			enc = enc.WithDelta(DeltaEncodingUint8)
+			dest := makeUnsafeRawSlice[uint8](unsafe.Pointer(&buf[offset]))
+			reduceUints[T, uint8](minimum, values, dest.Slice(len(values)))
+			offset += uint32(len(values))
+			return offset, enc
+		case align16:
+			enc = enc.WithDelta(DeltaEncodingUint16)
+			dest := makeUnsafeRawSlice[uint16](unsafe.Pointer(&buf[offset]))
+			reduceUints[T, uint16](minimum, values, dest.Slice(len(values)))
+			offset += uint32(len(values)) * align16
+			return offset, enc
+		case align32:
+			enc = enc.WithDelta(DeltaEncodingUint32)
+			dest := makeUnsafeRawSlice[uint32](unsafe.Pointer(&buf[offset]))
+			reduceUints[T, uint32](minimum, values, dest.Slice(len(values)))
+			offset += uint32(len(values)) * align32
+			return offset, enc
+		default:
+			panic("unreachable")
+		}
+	}
+	dest := makeUnsafeRawSlice[T](unsafe.Pointer(&buf[offset])).Slice(len(values))
+	offset += uint32(copy(dest, values)) * uint32(unsafe.Sizeof(T(0)))
+	return offset, enc
+}
+
+// WriteDebug implements Encoder.
+func (b *UintBuilder[T]) WriteDebug(w io.Writer, rows int) {
+	fmt.Fprintf(w, "%s: %d rows", b.dt, rows)
+}
+
+// reduceUints reduces the bit-width of a slice of unsigned by subtracting a
+// minimum value from each element and writing it to dst. For example,
+//
+//	reduceUints[uint64, uint8](10, []uint64{10, 11, 12}, dst)
+//
+// could be used to reduce a slice of uint64 values to uint8 values {0, 1, 2}.
+func reduceUints[O constraints.Integer, N constraints.Integer](minimum O, values []O, dst []N) {
+	for i := 0; i < len(values); i++ {
+		dst[i] = N(values[i] - minimum)
+	}
+}
+
+// computeMinMax computes the minimum and the maximum of the provided slice of
+// unsigned integers.
+func computeMinMax[I constraints.Unsigned](values []I) (I, I) {
+	minimum := I(0) - 1
+	maximum := I(0)
+	for _, v := range values {
+		minimum = min(minimum, v)
+		maximum = max(maximum, v)
+	}
+	return minimum, maximum
+}
+
+// deltaWidth returns the width in bytes of the integer type that can represent
+// the provided value.
+func deltaWidth(delta uint64) uint32 {
+	// TODO(jackson): Consider making this generic; We could compare against
+	// unsafe.Sizeof(T(0)) to ensure that we don't overflow T and that the
+	// higher width cases get elided at compile time for the smaller width Ts.
+	switch {
+	case delta == 0:
+		return 0
+	case delta < (1 << 8):
+		return 1
+	case delta < (1 << 16):
+		return align16
+	case delta < (1 << 32):
+		return align32
+	default:
+		return align64
+	}
+}
+
+func uintsToBinFormatter(f *binfmt.Formatter, rows int, desc ColumnDesc) {
+	logicalWidth := desc.DataType.uintWidth()
+	elementWidth := int(logicalWidth)
+	if desc.Encoding.Delta() != DeltaEncodingNone {
+		f.HexBytesln(int(logicalWidth), "%d-bit constant: %d", logicalWidth*8, f.PeekUint(int(logicalWidth)))
+
+		switch desc.Encoding.Delta() {
+		case DeltaEncodingConstant:
+			// This is just a constant.
+			rows = 1
+		case DeltaEncodingUint8:
+			elementWidth = 1
+		case DeltaEncodingUint16:
+			elementWidth = align16
+		case DeltaEncodingUint32:
+			elementWidth = align32
+		default:
+			panic("unreachable")
+		}
+	}
+	for i := 0; i < rows; i++ {
+		f.HexBytesln(elementWidth, "data[%d] = %d", i, f.PeekUint(elementWidth))
+	}
+}
diff --git a/sstable/colblk/unsafe_slice.go b/sstable/colblk/unsafe_slice.go
index 64241ffe..e7b663a0 100644
--- a/sstable/colblk/unsafe_slice.go
+++ b/sstable/colblk/unsafe_slice.go
@@ -39,3 +39,75 @@ func (s UnsafeRawSlice[T]) Slice(len int) []T {
 func (s UnsafeRawSlice[T]) set(i int, v T) {
 	*(*T)(unsafe.Pointer(uintptr(s.ptr) + unsafe.Sizeof(T(0))*uintptr(i))) = v
 }
+
+// UnsafeUint8s is an UnsafeIntegerSlice of uint8s, possibly using delta
+// encoding internally.
+type UnsafeUint8s = UnsafeIntegerSlice[uint8]
+
+// UnsafeUint16s is an UnsafeIntegerSlice of uint16s, possibly using delta
+// encoding internally.
+type UnsafeUint16s = UnsafeIntegerSlice[uint16]
+
+// UnsafeUint32s is an UnsafeIntegerSlice of uint32s, possibly using delta
+// encoding internally.
+type UnsafeUint32s = UnsafeIntegerSlice[uint32]
+
+// UnsafeUint64s is an UnsafeIntegerSlice of uint64s, possibly using delta
+// encoding internally.
+type UnsafeUint64s = UnsafeIntegerSlice[uint64]
+
+// UnsafeIntegerSlice exposes a read-only slice of integers from a column. If
+// the column's values are delta-encoded, UnsafeIntegerSlice transparently
+// applies deltas.
+//
+// See DeltaEncoding and UintBuilder.
+type UnsafeIntegerSlice[T constraints.Integer] struct {
+	base       T
+	deltaPtr   unsafe.Pointer
+	deltaWidth uintptr
+}
+
+func makeUnsafeIntegerSlice[T constraints.Integer](
+	base T, deltaPtr unsafe.Pointer, deltaWidth int,
+) UnsafeIntegerSlice[T] {
+	return UnsafeIntegerSlice[T]{
+		base:       base,
+		deltaPtr:   deltaPtr,
+		deltaWidth: uintptr(deltaWidth),
+	}
+}
+
+// TODO(jackson): Remove when more of the read path is hooked up.
+var _ = makeUnsafeIntegerSlice[uint64]
+
+// At returns the `i`-th element of the slice.
+func (s UnsafeIntegerSlice[T]) At(i int) T {
+	// TODO(jackson): Experiment with other alternatives that might be faster
+	// and avoid switching on the width.
+	switch s.deltaWidth {
+	case 0:
+		return s.base
+	case 1:
+		return s.base + T(*(*uint8)(unsafe.Pointer(uintptr(s.deltaPtr) + uintptr(i))))
+	case 2:
+		return s.base + T(*(*uint16)(unsafe.Pointer(uintptr(s.deltaPtr) + uintptr(i)<<align16Shift)))
+	case 4:
+		return s.base + T(*(*uint32)(unsafe.Pointer(uintptr(s.deltaPtr) + uintptr(i)<<align32Shift)))
+	case 8:
+		// NB: The slice encodes 64-bit integers, there is no base (it doesn't
+		// save any bits to compute a delta) and T must be a 64-bit integer. We
+		// cast directly into a *T pointer and don't add the base.
+		return (*(*T)(unsafe.Pointer(uintptr(s.deltaPtr) + uintptr(i)<<align64Shift)))
+	default:
+		panic("unreachable")
+	}
+}
+
+// Clone allocates a new slice and copies the first `rows` elements.
+func (s UnsafeIntegerSlice[T]) Clone(rows int) []T {
+	result := make([]T, rows)
+	for i := 0; i < rows; i++ {
+		result[i] = s.At(i)
+	}
+	return result
+}


Test patch:
diff --git a/batchrepr/writer_test.go b/batchrepr/writer_test.go
index 1f7585c5..4efea07a 100644
--- a/batchrepr/writer_test.go
+++ b/batchrepr/writer_test.go
@@ -54,8 +54,8 @@ func prettyBinaryRepr(repr []byte) string {
 	}
 
 	f := binfmt.New(repr).LineWidth(40)
-	f.HexBytesln(8, " seqnum=%d", f.PeekInt(8))
-	f.HexBytesln(4, " count=%d", f.PeekInt(4))
+	f.HexBytesln(8, " seqnum=%d", f.PeekUint(8))
+	f.HexBytesln(4, " count=%d", f.PeekUint(4))
 	for r := Read(repr); len(r) > 0; {
 		prevLen := len(r)
 		kind, ukey, _, ok, err := r.Next()
diff --git a/sstable/colblk/column_test.go b/sstable/colblk/column_test.go
new file mode 100644
index 00000000..8b0b61f8
--- /dev/null
+++ b/sstable/colblk/column_test.go
@@ -0,0 +1,54 @@
+// 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 colblk
+
+import "testing"
+
+func TestColumnDesc(t *testing.T) {
+	type testCase struct {
+		desc  ColumnDesc
+		delta DeltaEncoding
+		str   string
+	}
+
+	testCases := []testCase{
+		{
+			desc:  ColumnDesc{DataType: DataTypeBool, Encoding: EncodingDefault},
+			delta: DeltaEncodingNone,
+			str:   "bool",
+		},
+		{
+			desc:  ColumnDesc{DataType: DataTypeUint32, Encoding: EncodingDefault},
+			delta: DeltaEncodingNone,
+			str:   "uint32",
+		},
+		{
+			desc: ColumnDesc{
+				DataType: DataTypeUint32,
+				Encoding: EncodingDefault.WithDelta(DeltaEncodingUint16),
+			},
+			delta: DeltaEncodingUint16,
+			str:   "uint32+delta16",
+		},
+		{
+			desc: ColumnDesc{
+				DataType: DataTypeUint64,
+				Encoding: EncodingDefault.WithDelta(DeltaEncodingConstant),
+			},
+			delta: DeltaEncodingConstant,
+			str:   "uint64+const",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.str, func(t *testing.T) {
+			if got := tc.desc.String(); got != tc.str {
+				t.Errorf("String() = %q; want %q", got, tc.str)
+			}
+			if d := tc.desc.Encoding.Delta(); d != tc.delta {
+				t.Errorf(".Delta() = %q; want %q", d, tc.delta)
+			}
+		})
+	}
+}
diff --git a/sstable/colblk/testdata/uints b/sstable/colblk/testdata/uints
new file mode 100644
index 00000000..456b2588
--- /dev/null
+++ b/sstable/colblk/testdata/uints
@@ -0,0 +1,617 @@
+# This test works by initializing a set of UintBuilders at configured logical
+# widths and setting the same elements to the same indexes. The builders' Sizes
+# are queried for various row sizes. Individual builders of specific widths may
+# be finished separately so the test can continue with testing higher-width
+# integers.
+
+# Initialize all four writers.
+
+init widths=(8, 16, 32, 64)
+----
+b8
+b16
+b32
+b64
+
+# Write a few zero values at index [0,4].
+
+write
+0:0 1:0 2:0 3:0 4:0
+----
+
+# At all row counts, the column should be encoded as a constant using the column
+# type width.
+
+size rows=(5, 4, 3, 2, 1, 0)
+----
+b8:
+  8: *colblk.UintBuilder[uint8].Size(5, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(4, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(3, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(2, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(1, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(0, 0) = 0
+b16:
+  16: *colblk.UintBuilder[uint16].Size(5, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(4, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(3, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(2, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(1, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(0, 0) = 0
+b32:
+  32: *colblk.UintBuilder[uint32].Size(5, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(4, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(3, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(2, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(1, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(0, 0) = 0
+b64:
+  64: *colblk.UintBuilder[uint64].Size(5, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(4, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(3, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(2, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(1, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(0, 0) = 0
+
+# Add a nonzero value. Size calls that include the new row count should
+# increase, but the size calls that don't include the new row count should not.
+# The increased sizes should reflect use of a uint8 delta encoding.
+
+write
+5:10
+6:0
+7:10
+----
+
+size rows=(8, 7, 6, 5, 4, 3, 2, 1, 0)
+----
+b8:
+  8: *colblk.UintBuilder[uint8].Size(8, 0) = 8
+  8: *colblk.UintBuilder[uint8].Size(7, 0) = 7
+  8: *colblk.UintBuilder[uint8].Size(6, 0) = 6
+  8: *colblk.UintBuilder[uint8].Size(5, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(4, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(3, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(2, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(1, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(0, 0) = 0
+b16:
+  16: *colblk.UintBuilder[uint16].Size(8, 0) = 10
+  16: *colblk.UintBuilder[uint16].Size(7, 0) = 9
+  16: *colblk.UintBuilder[uint16].Size(6, 0) = 8
+  16: *colblk.UintBuilder[uint16].Size(5, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(4, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(3, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(2, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(1, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(0, 0) = 0
+b32:
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+  32: *colblk.UintBuilder[uint32].Size(7, 0) = 11
+  32: *colblk.UintBuilder[uint32].Size(6, 0) = 10
+  32: *colblk.UintBuilder[uint32].Size(5, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(4, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(3, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(2, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(1, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(0, 0) = 0
+b64:
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+  64: *colblk.UintBuilder[uint64].Size(7, 0) = 15
+  64: *colblk.UintBuilder[uint64].Size(6, 0) = 14
+  64: *colblk.UintBuilder[uint64].Size(5, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(4, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(3, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(2, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(1, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(0, 0) = 0
+
+# Finish the b8 so we can test 16-bit encoding.
+finish widths=(8) rows=8
+----
+b8: *colblk.UintBuilder[uint8]:
+0-1: x 00 # data[0] = 0
+1-2: x 00 # data[1] = 0
+2-3: x 00 # data[2] = 0
+3-4: x 00 # data[3] = 0
+4-5: x 00 # data[4] = 0
+5-6: x 0a # data[5] = 10
+6-7: x 00 # data[6] = 0
+7-8: x 0a # data[7] = 10
+Keeping b16 open
+Keeping b32 open
+Keeping b64 open
+
+# Add 1000 which should force a 16-bit delta encoding.
+
+write
+8:1000
+----
+
+size rows=(9, 8)
+----
+b16:
+  16: *colblk.UintBuilder[uint16].Size(9, 0) = 18
+  16: *colblk.UintBuilder[uint16].Size(8, 0) = 10
+b32:
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# We should be able to write up to 2^16-1 without triggering a 32-bit encoding.
+
+write
+9:65535
+----
+
+size rows=(10, 9, 8)
+----
+b16:
+  16: *colblk.UintBuilder[uint16].Size(10, 0) = 20
+  16: *colblk.UintBuilder[uint16].Size(9, 0) = 18
+  16: *colblk.UintBuilder[uint16].Size(8, 0) = 10
+b32:
+  32: *colblk.UintBuilder[uint32].Size(10, 0) = 24
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# But 2^16 should trigger a 32-bit encoding. (Finish b16 so we can test 32-bit
+# encoding.)
+
+finish widths=(16) rows=10
+----
+b16: *colblk.UintBuilder[uint16]:
+00-02: x 0000 # data[0] = 0
+02-04: x 0000 # data[1] = 0
+04-06: x 0000 # data[2] = 0
+06-08: x 0000 # data[3] = 0
+08-10: x 0000 # data[4] = 0
+10-12: x 0a00 # data[5] = 10
+12-14: x 0000 # data[6] = 0
+14-16: x 0a00 # data[7] = 10
+16-18: x e803 # data[8] = 1000
+18-20: x ffff # data[9] = 65535
+Keeping b32 open
+Keeping b64 open
+
+write
+10:65536
+----
+
+size rows=(11, 10, 9, 8)
+----
+b32:
+  32: *colblk.UintBuilder[uint32].Size(11, 0) = 44
+  32: *colblk.UintBuilder[uint32].Size(10, 0) = 24
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(11, 0) = 52
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# We should be able to write up to 2^32-1 without triggering a 64-bit encoding.
+
+write
+11:4294967295
+----
+
+size rows=(12, 11, 10, 9, 8)
+----
+b32:
+  32: *colblk.UintBuilder[uint32].Size(12, 0) = 48
+  32: *colblk.UintBuilder[uint32].Size(11, 0) = 44
+  32: *colblk.UintBuilder[uint32].Size(10, 0) = 24
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(12, 0) = 56
+  64: *colblk.UintBuilder[uint64].Size(11, 0) = 52
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# But 2^32 should trigger a 64-bit encoding.
+
+finish widths=(32) rows=12
+----
+b32: *colblk.UintBuilder[uint32]:
+00-04: x 00000000 # data[0] = 0
+04-08: x 00000000 # data[1] = 0
+08-12: x 00000000 # data[2] = 0
+12-16: x 00000000 # data[3] = 0
+16-20: x 00000000 # data[4] = 0
+20-24: x 0a000000 # data[5] = 10
+24-28: x 00000000 # data[6] = 0
+28-32: x 0a000000 # data[7] = 10
+32-36: x e8030000 # data[8] = 1000
+36-40: x ffff0000 # data[9] = 65535
+40-44: x 00000100 # data[10] = 65536
+44-48: x ffffffff # data[11] = 4294967295
+Keeping b64 open
+
+write
+12:4294967296
+----
+
+size rows=(13, 12, 11, 10, 9, 8)
+----
+b64:
+  64: *colblk.UintBuilder[uint64].Size(13, 0) = 104
+  64: *colblk.UintBuilder[uint64].Size(12, 0) = 56
+  64: *colblk.UintBuilder[uint64].Size(11, 0) = 52
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+finish widths=(64) rows=13
+----
+b64: *colblk.UintBuilder[uint64]:
+000-008: x 0000000000000000 # data[0] = 0
+008-016: x 0000000000000000 # data[1] = 0
+016-024: x 0000000000000000 # data[2] = 0
+024-032: x 0000000000000000 # data[3] = 0
+032-040: x 0000000000000000 # data[4] = 0
+040-048: x 0a00000000000000 # data[5] = 10
+048-056: x 0000000000000000 # data[6] = 0
+056-064: x 0a00000000000000 # data[7] = 10
+064-072: x e803000000000000 # data[8] = 1000
+072-080: x ffff000000000000 # data[9] = 65535
+080-088: x 0000010000000000 # data[10] = 65536
+088-096: x ffffffff00000000 # data[11] = 4294967295
+096-104: x 0000000001000000 # data[12] = 4294967296
+
+# Repeat the above tests but with a zero default value, and without explicitly
+# setting any of the zero values.
+
+init widths=(8, 16, 32, 64) default-zero
+----
+b8
+b16
+b32
+b64
+
+# At all row counts, the column should be encoded as a constant using the column
+# type width.
+
+size rows=(5, 4, 3, 2, 1, 0)
+----
+b8:
+  8: *colblk.UintBuilder[uint8].Size(5, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(4, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(3, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(2, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(1, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(0, 0) = 0
+b16:
+  16: *colblk.UintBuilder[uint16].Size(5, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(4, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(3, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(2, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(1, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(0, 0) = 0
+b32:
+  32: *colblk.UintBuilder[uint32].Size(5, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(4, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(3, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(2, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(1, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(0, 0) = 0
+b64:
+  64: *colblk.UintBuilder[uint64].Size(5, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(4, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(3, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(2, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(1, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(0, 0) = 0
+
+# Add a couple nonzero values. Size calls that include the new row count should
+# increase, but the size calls that don't include the new row count should not.
+
+write
+5:10
+7:10
+----
+
+size rows=(8, 7, 6, 5, 4, 3, 2, 1, 0)
+----
+b8:
+  8: *colblk.UintBuilder[uint8].Size(8, 0) = 8
+  8: *colblk.UintBuilder[uint8].Size(7, 0) = 7
+  8: *colblk.UintBuilder[uint8].Size(6, 0) = 6
+  8: *colblk.UintBuilder[uint8].Size(5, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(4, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(3, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(2, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(1, 0) = 1
+  8: *colblk.UintBuilder[uint8].Size(0, 0) = 0
+b16:
+  16: *colblk.UintBuilder[uint16].Size(8, 0) = 10
+  16: *colblk.UintBuilder[uint16].Size(7, 0) = 9
+  16: *colblk.UintBuilder[uint16].Size(6, 0) = 8
+  16: *colblk.UintBuilder[uint16].Size(5, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(4, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(3, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(2, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(1, 0) = 2
+  16: *colblk.UintBuilder[uint16].Size(0, 0) = 0
+b32:
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+  32: *colblk.UintBuilder[uint32].Size(7, 0) = 11
+  32: *colblk.UintBuilder[uint32].Size(6, 0) = 10
+  32: *colblk.UintBuilder[uint32].Size(5, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(4, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(3, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(2, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(1, 0) = 4
+  32: *colblk.UintBuilder[uint32].Size(0, 0) = 0
+b64:
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+  64: *colblk.UintBuilder[uint64].Size(7, 0) = 15
+  64: *colblk.UintBuilder[uint64].Size(6, 0) = 14
+  64: *colblk.UintBuilder[uint64].Size(5, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(4, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(3, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(2, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(1, 0) = 8
+  64: *colblk.UintBuilder[uint64].Size(0, 0) = 0
+
+# Finish the b8 so we can test 16-bit encoding.
+finish widths=(8) rows=8
+----
+b8: *colblk.UintBuilder[uint8]:
+0-1: x 00 # data[0] = 0
+1-2: x 00 # data[1] = 0
+2-3: x 00 # data[2] = 0
+3-4: x 00 # data[3] = 0
+4-5: x 00 # data[4] = 0
+5-6: x 0a # data[5] = 10
+6-7: x 00 # data[6] = 0
+7-8: x 0a # data[7] = 10
+Keeping b16 open
+Keeping b32 open
+Keeping b64 open
+
+# Add 1000 which should force a 16-bit delta encoding.
+
+write
+8:1000
+----
+
+size rows=(9, 8)
+----
+b16:
+  16: *colblk.UintBuilder[uint16].Size(9, 0) = 18
+  16: *colblk.UintBuilder[uint16].Size(8, 0) = 10
+b32:
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# We should be able to write up to 2^16-1 without triggering a 32-bit encoding.
+
+write
+9:65535
+----
+
+size rows=(10, 9, 8)
+----
+b16:
+  16: *colblk.UintBuilder[uint16].Size(10, 0) = 20
+  16: *colblk.UintBuilder[uint16].Size(9, 0) = 18
+  16: *colblk.UintBuilder[uint16].Size(8, 0) = 10
+b32:
+  32: *colblk.UintBuilder[uint32].Size(10, 0) = 24
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# But 2^16 should trigger a 32-bit encoding. (Finish b16 so we can test 32-bit
+# encoding.)
+
+finish widths=(16) rows=10
+----
+b16: *colblk.UintBuilder[uint16]:
+00-02: x 0000 # data[0] = 0
+02-04: x 0000 # data[1] = 0
+04-06: x 0000 # data[2] = 0
+06-08: x 0000 # data[3] = 0
+08-10: x 0000 # data[4] = 0
+10-12: x 0a00 # data[5] = 10
+12-14: x 0000 # data[6] = 0
+14-16: x 0a00 # data[7] = 10
+16-18: x e803 # data[8] = 1000
+18-20: x ffff # data[9] = 65535
+Keeping b32 open
+Keeping b64 open
+
+write
+10:65536
+----
+
+size rows=(11, 10, 9, 8)
+----
+b32:
+  32: *colblk.UintBuilder[uint32].Size(11, 0) = 44
+  32: *colblk.UintBuilder[uint32].Size(10, 0) = 24
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(11, 0) = 52
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# We should be able to write up to 2^32-1 without triggering a 64-bit encoding.
+
+write
+11:4294967295
+----
+
+size rows=(12, 11, 10, 9, 8)
+----
+b32:
+  32: *colblk.UintBuilder[uint32].Size(12, 0) = 48
+  32: *colblk.UintBuilder[uint32].Size(11, 0) = 44
+  32: *colblk.UintBuilder[uint32].Size(10, 0) = 24
+  32: *colblk.UintBuilder[uint32].Size(9, 0) = 22
+  32: *colblk.UintBuilder[uint32].Size(8, 0) = 12
+b64:
+  64: *colblk.UintBuilder[uint64].Size(12, 0) = 56
+  64: *colblk.UintBuilder[uint64].Size(11, 0) = 52
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+# But 2^32 should trigger a 64-bit encoding.
+
+finish widths=(32) rows=12
+----
+b32: *colblk.UintBuilder[uint32]:
+00-04: x 00000000 # data[0] = 0
+04-08: x 00000000 # data[1] = 0
+08-12: x 00000000 # data[2] = 0
+12-16: x 00000000 # data[3] = 0
+16-20: x 00000000 # data[4] = 0
+20-24: x 0a000000 # data[5] = 10
+24-28: x 00000000 # data[6] = 0
+28-32: x 0a000000 # data[7] = 10
+32-36: x e8030000 # data[8] = 1000
+36-40: x ffff0000 # data[9] = 65535
+40-44: x 00000100 # data[10] = 65536
+44-48: x ffffffff # data[11] = 4294967295
+Keeping b64 open
+
+write
+12:4294967296
+----
+
+size rows=(13, 12, 11, 10, 9, 8)
+----
+b64:
+  64: *colblk.UintBuilder[uint64].Size(13, 0) = 104
+  64: *colblk.UintBuilder[uint64].Size(12, 0) = 56
+  64: *colblk.UintBuilder[uint64].Size(11, 0) = 52
+  64: *colblk.UintBuilder[uint64].Size(10, 0) = 28
+  64: *colblk.UintBuilder[uint64].Size(9, 0) = 26
+  64: *colblk.UintBuilder[uint64].Size(8, 0) = 16
+
+finish widths=(64) rows=13
+----
+b64: *colblk.UintBuilder[uint64]:
+000-008: x 0000000000000000 # data[0] = 0
+008-016: x 0000000000000000 # data[1] = 0
+016-024: x 0000000000000000 # data[2] = 0
+024-032: x 0000000000000000 # data[3] = 0
+032-040: x 0000000000000000 # data[4] = 0
+040-048: x 0a00000000000000 # data[5] = 10
+048-056: x 0000000000000000 # data[6] = 0
+056-064: x 0a00000000000000 # data[7] = 10
+064-072: x e803000000000000 # data[8] = 1000
+072-080: x ffff000000000000 # data[9] = 65535
+080-088: x 0000010000000000 # data[10] = 65536
+088-096: x ffffffff00000000 # data[11] = 4294967295
+096-104: x 0000000001000000 # data[12] = 4294967296
+
+# Test serializing a few columns using delta encoding.
+
+init widths=(8, 16, 32, 64) default-zero
+----
+b8
+b16
+b32
+b64
+
+write
+0:1 2:92 3:1 7:86 20:221
+----
+
+size rows=5
+----
+b8:
+  8: *colblk.UintBuilder[uint8].Size(5, 0) = 5
+b16:
+  16: *colblk.UintBuilder[uint16].Size(5, 0) = 7
+b32:
+  32: *colblk.UintBuilder[uint32].Size(5, 0) = 9
+b64:
+  64: *colblk.UintBuilder[uint64].Size(5, 0) = 13
+
+finish widths=(8,16,32,64) rows=5
+----
+b8: *colblk.UintBuilder[uint8]:
+0-1: x 01 # data[0] = 1
+1-2: x 00 # data[1] = 0
+2-3: x 5c # data[2] = 92
+3-4: x 01 # data[3] = 1
+4-5: x 00 # data[4] = 0
+b16: *colblk.UintBuilder[uint16]:
+0-2: x 0000 # 16-bit constant: 0
+2-3: x 01   # data[0] = 1
+3-4: x 00   # data[1] = 0
+4-5: x 5c   # data[2] = 92
+5-6: x 01   # data[3] = 1
+6-7: x 00   # data[4] = 0
+b32: *colblk.UintBuilder[uint32]:
+0-4: x 00000000 # 32-bit constant: 0
+4-5: x 01       # data[0] = 1
+5-6: x 00       # data[1] = 0
+6-7: x 5c       # data[2] = 92
+7-8: x 01       # data[3] = 1
+8-9: x 00       # data[4] = 0
+b64: *colblk.UintBuilder[uint64]:
+00-08: x 0000000000000000 # 64-bit constant: 0
+08-09: x 01               # data[0] = 1
+09-10: x 00               # data[1] = 0
+10-11: x 5c               # data[2] = 92
+11-12: x 01               # data[3] = 1
+12-13: x 00               # data[4] = 0
+
+# Test a situation where the most recently written value requirs a wider delta
+# encoding, but we Finish with few enough rows that we should serialize using
+# the smaller encoding.
+
+init widths=(64)
+----
+b64
+
+write
+0:0 1:29 2:595 3:2 4:2 5:9
+----
+
+size rows=(6)
+----
+b64:
+  64: *colblk.UintBuilder[uint64].Size(6, 0) = 20
+
+write
+6:70395
+----
+
+size rows=(7)
+----
+b64:
+  64: *colblk.UintBuilder[uint64].Size(7, 0) = 36
+
+finish widths=(64) rows=6
+----
+b64: *colblk.UintBuilder[uint64]:
+00-08: x 0000000000000000 # 64-bit constant: 0
+08-10: x 0000             # data[0] = 0
+10-12: x 1d00             # data[1] = 29
+12-14: x 5302             # data[2] = 595
+14-16: x 0200             # data[3] = 2
+16-18: x 0200             # data[4] = 2
+18-20: x 0900             # data[5] = 9
diff --git a/sstable/colblk/uints_test.go b/sstable/colblk/uints_test.go
new file mode 100644
index 00000000..f53125e8
--- /dev/null
+++ b/sstable/colblk/uints_test.go
@@ -0,0 +1,126 @@
+// 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 colblk
+
+import (
+	"bytes"
+	"fmt"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/aligned"
+	"github.com/cockroachdb/pebble/internal/binfmt"
+)
+
+func TestUints(t *testing.T) {
+	var b8 UintBuilder[uint8]
+	var b16 UintBuilder[uint16]
+	var b32 UintBuilder[uint32]
+	var b64 UintBuilder[uint64]
+
+	var out bytes.Buffer
+	var widths []int
+	var writers []ColumnWriter
+	datadriven.RunTest(t, "testdata/uints", func(t *testing.T, td *datadriven.TestData) string {
+		out.Reset()
+		switch td.Cmd {
+		case "init":
+			widths = widths[:0]
+			writers = writers[:0]
+			td.ScanArgs(t, "widths", &widths)
+			defaultZero := td.HasArg("default-zero")
+			for _, w := range widths {
+				switch w {
+				case 8:
+					b8.init(defaultZero)
+					writers = append(writers, &b8)
+				case 16:
+					b16.init(defaultZero)
+					writers = append(writers, &b16)
+				case 32:
+					b32.init(defaultZero)
+					writers = append(writers, &b32)
+				case 64:
+					b64.init(defaultZero)
+					writers = append(writers, &b64)
+				default:
+					panic(fmt.Sprintf("unknown width: %d", w))
+				}
+				fmt.Fprintf(&out, "b%d\n", w)
+			}
+			return out.String()
+		case "write":
+			for _, f := range strings.Fields(td.Input) {
+				delim := strings.IndexByte(f, ':')
+				i, err := strconv.Atoi(f[:delim])
+				if err != nil {
+					return err.Error()
+				}
+				for _, width := range widths {
+					v, err := strconv.ParseUint(f[delim+1:], 10, width)
+					if err != nil {
+						return err.Error()
+					}
+					switch width {
+					case 8:
+						b8.Set(i, uint8(v))
+					case 16:
+						b16.Set(i, uint16(v))
+					case 32:
+						b32.Set(i, uint32(v))
+					case 64:
+						b64.Set(i, v)
+					default:
+						panic(fmt.Sprintf("unknown width: %d", width))
+					}
+				}
+			}
+			return out.String()
+		case "size":
+			var rowCounts []int
+			td.ScanArgs(t, "rows", &rowCounts)
+			for wIdx, w := range writers {
+				fmt.Fprintf(&out, "b%d:\n", widths[wIdx])
+				for _, rows := range rowCounts {
+					fmt.Fprintf(&out, "  %d: %T.Size(%d, 0) = %d\n", widths[wIdx], w, rows, w.Size(rows, 0))
+				}
+			}
+			return out.String()
+		case "finish":
+			var rows int
+			var finishWidths []int
+			td.ScanArgs(t, "rows", &rows)
+			td.ScanArgs(t, "widths", &finishWidths)
+			var newWriters []ColumnWriter
+			var newWidths []int
+			for wIdx, width := range widths {
+				var shouldFinish bool
+				for _, fw := range finishWidths {
+					shouldFinish = shouldFinish || width == fw
+				}
+				if shouldFinish {
+					sz := writers[wIdx].Size(rows, 0)
+					buf := aligned.ByteSlice(int(sz))
+					_, desc := writers[wIdx].Finish(0, rows, 0, buf)
+					fmt.Fprintf(&out, "b%d: %T:\n", width, writers[wIdx])
+					f := binfmt.New(buf).LineWidth(20)
+					uintsToBinFormatter(f, rows, desc)
+					fmt.Fprintf(&out, "%s", f.String())
+				} else {
+					fmt.Fprintf(&out, "Keeping b%d open\n", width)
+					newWidths = append(newWidths, width)
+					newWriters = append(newWriters, writers[wIdx])
+				}
+			}
+			writers = newWriters
+			widths = newWidths
+			return out.String()
+		default:
+			panic(fmt.Sprintf("unknown command: %s", td.Cmd))
+		}
+	})
+}