Repository: cockroachdb/pebble

Issue:
db: add new columns to datablock
As a part of tiered storage, we should add tiering (spanId, attribute) columns to our DataBlock. See: https://github.com/cockroachdb/pebble/commit/f8876c6d9e184584df6a51cab8c1d3773833d3f1, https://github.com/cockroachdb/pebble/commit/2a099c206139ac2f67844a447ab1f5b8eca1c1c6

Jira issue: PEBBLE-1227

Epic CRDB-48982

Hints from issue comments:

Solution patch:
diff --git a/internal/compact/iterator.go b/internal/compact/iterator.go
index 44647845..378ebe82 100644
--- a/internal/compact/iterator.go
+++ b/internal/compact/iterator.go
@@ -163,6 +163,10 @@ type Iter struct {
 	// keys.
 	iter base.InternalIterator
 
+	// metaIter is the MetaIterator version of iter, if available. This allows
+	// access to tiering metadata without adding overhead to the common iteration path.
+	metaIter base.InternalIteratorWithKVMeta
+
 	delElider         pointTombstoneElider
 	rangeDelCompactor RangeDelSpanCompactor
 	rangeKeyCompactor RangeKeySpanCompactor
@@ -181,6 +185,13 @@ type Iter struct {
 	// when moving to the next key so it can determine whether the user key has
 	// changed from the previous key.
 	kv base.InternalKV
+	// currentMeta holds the metadata for the key-value pair in i.kv (the key that
+	// will be returned to the caller). This is updated when saveKey() is called
+	// to ensure GetCurrentMeta() returns the metadata for the returned key.
+	currentMeta base.KVMeta
+	// nextMeta holds the metadata for i.iterKV (the next key being peeked at).
+	// This is updated by iterNext() and then copied to currentMeta when saveKey() is called.
+	nextMeta base.KVMeta
 	// keyTrailer is updated when `i.kv` is updated and holds the key's original
 	// trailer (eg, before any sequence-number zeroing or changes to key kind).
 	keyTrailer  base.InternalKeyTrailer
@@ -356,6 +367,10 @@ func NewIter(
 	}
 	i.iter = invalidating.MaybeWrapIfInvariants(iter)
 
+	if metaIter, ok := i.iter.(base.InternalIteratorWithKVMeta); ok {
+		i.metaIter = metaIter
+	}
+
 	i.frontiers.Init(i.cmp)
 	i.delElider.Init(i.cmp, cfg.TombstoneElision)
 	i.rangeDelCompactor = MakeRangeDelSpanCompactor(i.cmp, i.cfg.Comparer.Equal, cfg.Snapshots, cfg.TombstoneElision)
@@ -389,12 +404,26 @@ func (i *Iter) Stats() IterStats {
 	return i.stats
 }
 
+// GetCurrentMeta returns the metadata for the current key-value pair, if available.
+// This method provides access to tiering metadata without adding overhead to the
+// common iteration path.
+func (i *Iter) GetCurrentMeta() base.KVMeta {
+	return i.currentMeta
+}
+
 // First has the same semantics as InternalIterator.First.
 func (i *Iter) First() *base.InternalKV {
 	if i.err != nil {
 		return nil
 	}
-	i.iterKV = i.iter.First()
+
+	if i.metaIter != nil {
+		i.iterKV, i.nextMeta = i.metaIter.FirstWithMeta()
+	} else {
+		i.iterKV = i.iter.First()
+		i.nextMeta = base.KVMeta{}
+	}
+
 	if i.iterKV != nil {
 		i.curSnapshotIdx, i.curSnapshotSeqNum = i.cfg.Snapshots.IndexAndSeqNum(i.iterKV.SeqNum())
 	}
@@ -707,7 +736,12 @@ func (i *Iter) skipInStripe() {
 }
 
 func (i *Iter) iterNext() bool {
-	i.iterKV = i.iter.Next()
+	if i.metaIter != nil {
+		i.iterKV, i.nextMeta = i.metaIter.NextWithMeta()
+	} else {
+		i.iterKV = i.iter.Next()
+		i.nextMeta = base.KVMeta{}
+	}
 	if i.iterKV == nil {
 		i.err = i.iter.Error()
 	}
@@ -1328,6 +1362,8 @@ func (i *Iter) deleteSizedNext() *base.InternalKV {
 }
 
 // saveKey saves the key in iterKV to i.kv.K using i.keyBuf's memory.
+// It also updates currentMeta to nextMeta so that GetCurrentMeta()
+// returns the metadata for the returned key.
 func (i *Iter) saveKey() {
 	i.keyBuf = append(i.keyBuf[:0], i.iterKV.K.UserKey...)
 	i.kv.K = base.InternalKey{
@@ -1335,6 +1371,7 @@ func (i *Iter) saveKey() {
 		Trailer: i.iterKV.K.Trailer,
 	}
 	i.keyTrailer = i.kv.K.Trailer
+	i.currentMeta = i.nextMeta
 }
 
 // saveValue saves the value in iterKV to i.kv. It must be called before
diff --git a/internal/compact/run.go b/internal/compact/run.go
index baa7f79a..f82bb404 100644
--- a/internal/compact/run.go
+++ b/internal/compact/run.go
@@ -305,7 +305,7 @@ func (r *Runner) writeKeysToTable(
 		// If the key might be garbage (all requirements of
 		// sstable.IsLikelyMVCCGarbage are met), we eagerly separate the value
 		// into a blob file.
-		if err := valueSeparation.Add(tw, kv, r.iter.ForceObsoleteDueToRangeDel(), isLikelyMVCCGarbage); err != nil {
+		if err := valueSeparation.Add(tw, kv, r.iter.ForceObsoleteDueToRangeDel(), isLikelyMVCCGarbage, r.iter.GetCurrentMeta()); err != nil {
 			return nil, err
 		}
 		if r.iter.SnapshotPinned() {
diff --git a/metamorphic/build.go b/metamorphic/build.go
index b8060efb..18d417e1 100644
--- a/metamorphic/build.go
+++ b/metamorphic/build.go
@@ -98,7 +98,7 @@ func writeSSTForIngestion(
 			return nil, err
 		}
 		t.opts.Comparer.ValidateKey.MustValidate(k.K.UserKey)
-		if err := w.Raw().Add(k.K, valBytes, false); err != nil {
+		if err := w.Raw().Add(k.K, valBytes, false, base.KVMeta{}); err != nil {
 			return nil, err
 		}
 	}
diff --git a/metamorphic/ops.go b/metamorphic/ops.go
index 2a82c19c..71fab143 100644
--- a/metamorphic/ops.go
+++ b/metamorphic/ops.go
@@ -2017,7 +2017,7 @@ func (r *replicateOp) runSharedReplicate(
 			if err != nil {
 				panic(err)
 			}
-			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false)
+			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false, base.KVMeta{})
 		},
 		VisitRangeDel: func(start, end []byte, seqNum base.SeqNum) error {
 			return w.DeleteRange(start, end)
@@ -2085,7 +2085,7 @@ func (r *replicateOp) runExternalReplicate(
 				panic(err)
 			}
 			t.opts.Comparer.ValidateKey.MustValidate(key.UserKey)
-			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false)
+			return w.Raw().Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false, base.KVMeta{})
 		},
 		VisitRangeDel: func(start, end []byte, seqNum base.SeqNum) error {
 			t.opts.Comparer.ValidateKey.MustValidate(start)
diff --git a/sstable/block/flush_governor.go b/sstable/block/flush_governor.go
index 9d8b54c6..48205501 100644
--- a/sstable/block/flush_governor.go
+++ b/sstable/block/flush_governor.go
@@ -42,6 +42,12 @@ type FlushGovernor struct {
 // within a 1024B class. However, when loaded into the block cache we also
 // allocate space for the cache entry metadata. The new allocation may now only
 // fit within a 2048B class, which increases internal fragmentation.
+//
+// TODO: Consider adding extra slack here for future metadata growth (tradeoff:
+// wasted memory). Metadata size increases can cause internal fragmentation when
+// reading blocks written by older versions that were aligned to size classes
+// according to the old metadata size. An alternative design (e.g. a separate
+// slab allocator for metadata) could also avoid this issue.
 const blockAllocationOverhead = cache.ValueMetadataSize + MetadataSize
 
 // MakeFlushGovernor initializes a flush controller.
diff --git a/sstable/colblk/data_block.go b/sstable/colblk/data_block.go
index baa24eb2..72cd4b65 100644
--- a/sstable/colblk/data_block.go
+++ b/sstable/colblk/data_block.go
@@ -477,12 +477,28 @@ type DataBlockEncoder struct {
 	// when a key is known to be obsolete/non-live (i.e., shadowed by another
 	// identical point key or range deletion with a higher sequence number).
 	isObsolete BitmapBuilder
+	// These two columns are only accessed when the sstable is rewritten, so we
+	// stick them at the end of the data block. Both will be accessed together
+	// for each key, and we expect that in most cases there will be at most one
+	// unique spanID.
+	//
+	// Alternatively, we could have added these fields to the custom
+	// serialization we do for values when it is a value block handle or blob
+	// handle, but then we need to add a custom serialization for the in-place
+	// value case too, and all that serialization predated support for columnar
+	// blocks.
+	tieringSpanIDs    UintBuilder
+	tieringAttributes UintBuilder
 
 	enc              BlockEncoder
 	rows             int
 	maximumKeyLength int
 	valuePrefixTmp   [1]byte
 	lastUserKeyTmp   []byte
+
+	// columnConfig is a configuration specifying whether optional columns are
+	// present.
+	columnConfig OptionalColumnConfig
 }
 
 const (
@@ -491,9 +507,87 @@ const (
 	dataBlockColumnValue
 	dataBlockColumnIsValueExternal
 	dataBlockColumnIsObsolete
-	dataBlockColumnMax
+	dataBlockColumnTieringSpanID
+	dataBlockColumnTieringAttribute
+	dataBlockColumnMaxV1 = dataBlockColumnTieringSpanID
+	dataBlockColumnMaxV2 = dataBlockColumnTieringAttribute + 1
 )
 
+// OptionalColumnConfig describes the columns that are present in a columnar
+// block.
+type OptionalColumnConfig struct {
+	// --- Tiering columns ---
+	//
+	// Why spanID? An attribute extraction policy for a SQL index is immutable
+	// in CockroachDB. It can be represented by the [start, end) span at the
+	// Pebble layer, and/or a spanID. The age at which data becomes cold is the
+	// tiering policy and is mutable. Pebble needs the following information:
+	//
+	//   - Extract the attribute from the key-value pair for the first time. We
+	//     want to do this once when first generating a sstable (and possibly a
+	//     blob file) containing that key-value pair. With key and value
+	//     separation, we are moving into a world where we want to continue to be
+	//     able to rewrite sstables without reading the value from the blob file
+	//     and rewrite blob files without reading the key from the sstable. And
+	//     even though we start with only cold blob files, we plan to extend this
+	//     to cold sstables too, so the attribute is stored with the key in the
+	//     sstable, and with the value in the blob file.
+	//
+	//   - Get the current tiering policy for a key (sstable rewriting) or for the
+	//     value (when rewriting a blob file for policy enforcement or for space
+	//     reclamation). A spanID permits easier/faster policy lookup for the key
+	//     case, but is not necessary. For the value case, we need a way to lookup
+	//     the policy. With a single conceptual spanID in an sstable or blob file,
+	//     we can store a single key belonging to that span in the blob file and
+	//     use it for policy lookup. So a spanID is again not necessary. If we
+	//     allow multiple conceptual spanIDs in a sstable or blob file, having a
+	//     spanID becomes necessary for a blob file since we can store the spanID
+	//     concisely with each value. For blob files in higher levels, splitting
+	//     per spanID may result in unnecessarily small files. Arguably, we could
+	//     have two approaches where blob files in higher levels are not split by
+	//     spanID, since these blob files may never need to be cold, and when we
+	//     write blob files for lower level sstable compactions we would extract
+	//     the tiering policy using the sstable key and make the blob file have a
+	//     single spanID. Having blob files that are not self contained wrt us
+	//     being able to compute how much cold data is in them (by reading them)
+	//     is not ideal, IMHO, so we don't do this (yet).
+	//
+	//   - TODO(annie): Store the histogram of attributes inside a sstable and blob
+	//     file. This is needed to decide when to rewrite to make a hot => cold
+	//     transition. With a single spanID per file, we don't need a spanID. There
+	//     is some concern that users could easily abuse the system by creating
+	//     1000s of tiny SQL tables with different attribute extraction policies,
+	//     resulting in many tiny files. With multiple extraction policies per file,
+	//     a spanID provides a convenient key.
+	//
+	//   - Tiering policy change: At the Pebble layer we will keep aggregate
+	//     histograms of each attribute extraction policy, so we can compare these
+	//     to the current tiering policy (which may have changed since the last
+	//     time we checked), to decide if there is substantial amount of data in
+	//     the wrong tier. A spanID is a convenient key, though we could also
+	//     accomplish this with the start key of the policy.
+	HasColumnSpanID    bool
+	HasColumnAttribute bool
+}
+
+func (c *OptionalColumnConfig) SupportsTiering() bool {
+	return c.HasColumnSpanID && c.HasColumnAttribute
+}
+
+// NoTieringColumns returns a TieringColumnConfig with no tiering columns present.
+// Useful for tests and older table formats that don't support tiering metadata.
+func NoTieringColumns() OptionalColumnConfig {
+	return OptionalColumnConfig{}
+}
+
+// WithTieringColumns returns a TieringColumnConfig with all tiering columns present.
+func WithTieringColumns() OptionalColumnConfig {
+	return OptionalColumnConfig{
+		HasColumnSpanID:    true,
+		HasColumnAttribute: true,
+	}
+}
+
 // The data block header is a 4-byte uint32 encoding the maximum length of a key
 // contained within the block. This is used by iterators to avoid the need to
 // grow key buffers while iterating over the block, ensuring that the key buffer
@@ -502,7 +596,7 @@ const (
 const DataBlockCustomHeaderSize = 4
 
 // Init initializes the data block writer.
-func (w *DataBlockEncoder) Init(schema *KeySchema) {
+func (w *DataBlockEncoder) Init(schema *KeySchema, optionalColConfig OptionalColumnConfig) {
 	w.Schema = schema
 	w.KeyWriter = schema.NewKeyWriter()
 	w.trailers.Init()
@@ -510,6 +604,11 @@ func (w *DataBlockEncoder) Init(schema *KeySchema) {
 	w.values.Init()
 	w.isValueExternal.Reset()
 	w.isObsolete.Reset()
+	w.columnConfig = optionalColConfig
+	if optionalColConfig.SupportsTiering() {
+		w.tieringSpanIDs.InitWithDefault()
+		w.tieringAttributes.InitWithDefault()
+	}
 	w.rows = 0
 	w.maximumKeyLength = 0
 	w.lastUserKeyTmp = w.lastUserKeyTmp[:0]
@@ -524,6 +623,10 @@ func (w *DataBlockEncoder) Reset() {
 	w.values.Reset()
 	w.isValueExternal.Reset()
 	w.isObsolete.Reset()
+	if w.columnConfig.SupportsTiering() {
+		w.tieringSpanIDs.Reset()
+		w.tieringAttributes.Reset()
+	}
 	w.rows = 0
 	w.maximumKeyLength = 0
 	w.lastUserKeyTmp = w.lastUserKeyTmp[:0]
@@ -557,6 +660,16 @@ func (w *DataBlockEncoder) String() string {
 	w.isObsolete.WriteDebug(&buf, w.rows)
 	fmt.Fprintln(&buf)
 
+	if w.columnConfig.SupportsTiering() {
+		fmt.Fprintf(&buf, "%d: tiering-span-id: ", len(w.Schema.ColumnTypes)+dataBlockColumnTieringSpanID)
+		w.tieringSpanIDs.WriteDebug(&buf, w.rows)
+		fmt.Fprintln(&buf)
+
+		fmt.Fprintf(&buf, "%d: tiering-attribute: ", len(w.Schema.ColumnTypes)+dataBlockColumnTieringAttribute)
+		w.tieringAttributes.WriteDebug(&buf, w.rows)
+		fmt.Fprintln(&buf)
+	}
+
 	return buf.String()
 }
 
@@ -574,6 +687,7 @@ func (w *DataBlockEncoder) Add(
 	valuePrefix block.ValuePrefix,
 	kcmp KeyComparison,
 	isObsolete bool,
+	meta base.KVMeta,
 ) {
 	w.KeyWriter.WriteKey(w.rows, ikey.UserKey, kcmp.PrefixLen, kcmp.CommonPrefixLen)
 	if kcmp.PrefixEqual() {
@@ -593,6 +707,10 @@ func (w *DataBlockEncoder) Add(
 		// bitmap and know there is no value prefix byte if !isValueExternal.
 		w.values.Put(value)
 	}
+	if w.columnConfig.SupportsTiering() && meta != (base.KVMeta{}) {
+		w.tieringSpanIDs.Set(w.rows, meta.TieringSpanID)
+		w.tieringAttributes.Set(w.rows, uint64(meta.TieringAttribute))
+	}
 	if len(ikey.UserKey) > int(w.maximumKeyLength) {
 		w.maximumKeyLength = len(ikey.UserKey)
 	}
@@ -606,17 +724,28 @@ func (w *DataBlockEncoder) Rows() int {
 
 // Size returns the size of the current pending data block.
 func (w *DataBlockEncoder) Size() int {
-	off := HeaderSize(len(w.Schema.ColumnTypes)+dataBlockColumnMax, DataBlockCustomHeaderSize+w.Schema.HeaderSize)
+	off := HeaderSize(len(w.Schema.ColumnTypes)+w.numFormatColumns(), DataBlockCustomHeaderSize+w.Schema.HeaderSize)
 	off = w.KeyWriter.Size(w.rows, off)
 	off = w.trailers.Size(w.rows, off)
 	off = w.prefixSame.InvertedSize(w.rows, off)
 	off = w.values.Size(w.rows, off)
 	off = w.isValueExternal.Size(w.rows, off)
 	off = w.isObsolete.Size(w.rows, off)
+	if w.columnConfig.SupportsTiering() {
+		off = w.tieringSpanIDs.Size(w.rows, off)
+		off = w.tieringAttributes.Size(w.rows, off)
+	}
 	off++ // trailer padding byte
 	return int(off)
 }
 
+func (w *DataBlockEncoder) numFormatColumns() int {
+	if w.columnConfig.SupportsTiering() {
+		return dataBlockColumnMaxV2
+	}
+	return dataBlockColumnMaxV1
+}
+
 // MaterializeLastUserKey materializes the last added user key.
 func (w *DataBlockEncoder) MaterializeLastUserKey(appendTo []byte) []byte {
 	return w.KeyWriter.MaterializeKey(appendTo, w.rows-1)
@@ -636,7 +765,7 @@ func (w *DataBlockEncoder) Finish(rows, size int) (finished []byte, lastKey base
 		panic(errors.AssertionFailedf("data block has %d rows; asked to finish %d", w.rows, rows))
 	}
 
-	cols := len(w.Schema.ColumnTypes) + dataBlockColumnMax
+	cols := len(w.Schema.ColumnTypes) + w.numFormatColumns()
 	h := Header{
 		Version: Version1,
 		Columns: uint16(cols),
@@ -659,6 +788,10 @@ func (w *DataBlockEncoder) Finish(rows, size int) (finished []byte, lastKey base
 	w.enc.Encode(rows, &w.values)
 	w.enc.Encode(rows, &w.isValueExternal)
 	w.enc.Encode(rows, &w.isObsolete)
+	if w.columnConfig.SupportsTiering() {
+		w.enc.Encode(rows, &w.tieringSpanIDs)
+		w.enc.Encode(rows, &w.tieringAttributes)
+	}
 	finished = w.enc.Finish()
 
 	w.lastUserKeyTmp = w.lastUserKeyTmp[:0]
@@ -672,7 +805,8 @@ func (w *DataBlockEncoder) Finish(rows, size int) (finished []byte, lastKey base
 
 // DataBlockRewriter rewrites data blocks. See RewriteSuffixes.
 type DataBlockRewriter struct {
-	KeySchema *KeySchema
+	tieringColConfig *OptionalColumnConfig
+	KeySchema        *KeySchema
 
 	encoder   DataBlockEncoder
 	decoder   DataBlockDecoder
@@ -687,10 +821,13 @@ type DataBlockRewriter struct {
 }
 
 // NewDataBlockRewriter creates a block rewriter.
-func NewDataBlockRewriter(keySchema *KeySchema, comparer *base.Comparer) *DataBlockRewriter {
+func NewDataBlockRewriter(
+	keySchema *KeySchema, comparer *base.Comparer, tieringColConfig OptionalColumnConfig,
+) *DataBlockRewriter {
 	return &DataBlockRewriter{
-		KeySchema: keySchema,
-		comparer:  comparer,
+		tieringColConfig: &tieringColConfig,
+		KeySchema:        keySchema,
+		comparer:         comparer,
 	}
 }
 
@@ -719,8 +856,8 @@ func (rw *DataBlockRewriter) RewriteSuffixes(
 	input []byte, from []byte, to []byte,
 ) (start, end base.InternalKey, rewritten []byte, err error) {
 	if !rw.initialized {
-		rw.iter.InitOnce(rw.KeySchema, rw.comparer, assertNoExternalValues{})
-		rw.encoder.Init(rw.KeySchema)
+		rw.iter.InitOnce(rw.KeySchema, rw.comparer, assertNoExternalValues{}, *rw.tieringColConfig)
+		rw.encoder.Init(rw.KeySchema, *rw.tieringColConfig)
 		rw.initialized = true
 	}
 
@@ -750,7 +887,7 @@ func (rw *DataBlockRewriter) RewriteSuffixes(
 	rw.KeySchema.InitKeySeekerMetadata(meta, &rw.decoder, bd)
 	rw.keySeeker = rw.KeySchema.KeySeeker(meta)
 	rw.encoder.Reset()
-	if err = rw.iter.Init(&rw.decoder, bd, blockiter.Transforms{}); err != nil {
+	if err = rw.iter.Init(&rw.decoder, bd, blockiter.Transforms{}, *rw.tieringColConfig); err != nil {
 		return base.InternalKey{}, base.InternalKey{}, nil, err
 	}
 
@@ -785,7 +922,11 @@ func (rw *DataBlockRewriter) RewriteSuffixes(
 			start.Trailer = kv.K.Trailer
 		}
 		k := base.InternalKey{UserKey: rw.keyBuf, Trailer: kv.K.Trailer}
-		rw.encoder.Add(k, value, valuePrefix, kcmp, rw.decoder.isObsolete.At(i))
+		var meta base.KVMeta
+		if rw.tieringColConfig.SupportsTiering() {
+			meta = rw.iter.decodeMeta()
+		}
+		rw.encoder.Add(k, value, valuePrefix, kcmp, rw.decoder.isObsolete.At(i), meta)
 	}
 	rewritten, end = rw.encoder.Finish(int(bd.header.Rows), rw.encoder.Size())
 	end.UserKey, rw.keyAlloc = rw.keyAlloc.Copy(end.UserKey)
@@ -858,6 +999,15 @@ func InitKeyspanBlockMetadata(md *block.Metadata, data []byte) (err error) {
 
 // A DataBlockDecoder holds state for interpreting a columnar data block. It may
 // be shared among multiple DataBlockIters.
+//
+// Tiering metadata (tieringSpanID, tieringAttribute) is intentionally stored
+// in DataBlockIter rather than below. DataBlockDecoder is embedded in
+// block.Metadata, which is allocated alongside every cached block. Enlarging
+// DataBlockDecoder would require increasing block.MetadataSize, which would
+// increase the allocation size for every cached block—including old blocks
+// written without tiering columns. This could push those old blocks across
+// allocator size class boundaries, causing significant internal fragmentation
+// in the block cache.
 type DataBlockDecoder struct {
 	// trailers holds an array of the InternalKey trailers, encoding the key
 	// kind and sequence number of each key.
@@ -1009,6 +1159,9 @@ type DataBlockIter struct {
 	// getLazyValuer configures the DataBlockIterConfig to initialize the
 	// DataBlockIter to use the provided handler for retrieving lazy values.
 	getLazyValuer block.GetInternalValueForPrefixAndValueHandler
+	// tieringConfig indicates whether tiering columns are present in blocks.
+	// This is set once in InitOnce.
+	tieringConfig OptionalColumnConfig
 
 	// -- Fields that are initialized for each block --
 	// For any changes to these fields, InitHandle should be updated.
@@ -1020,6 +1173,19 @@ type DataBlockIter struct {
 	noTransforms bool
 	keySeeker    KeySeeker
 
+	// -- Tiering metadata (lazily initialized) --
+	// These fields are decoded on-demand when *WithMeta methods are first called.
+	// This design preserves alignment with size classes used by existing blocks,
+	// avoiding internal fragmentation when reading blocks written by older versions.
+
+	// tieringSpanIDs and tieringAttributes store the decoded tiering columns
+	// for the current block.
+	tieringSpanIDs    UnsafeUints
+	tieringAttributes UnsafeUints
+	// blockData stores a reference to the block data for lazy tiering column
+	// decoding. This is only set when Init() is called (not InitHandle()).
+	blockData []byte
+
 	// -- State --
 	// For any changes to these fields, InitHandle (which resets them) should be
 	// updated.
@@ -1042,17 +1208,23 @@ func (i *DataBlockIter) InitOnce(
 	keySchema *KeySchema,
 	comparer *base.Comparer,
 	getLazyValuer block.GetInternalValueForPrefixAndValueHandler,
+	tieringConfig OptionalColumnConfig,
 ) {
 	i.keySchema = keySchema
 	i.suffixCmp = comparer.ComparePointSuffixes
 	i.split = comparer.Split
 	i.getLazyValuer = getLazyValuer
+	i.tieringConfig = tieringConfig
 }
 
 // Init initializes the data block iterator, configuring it to read from the
-// provided decoder.
+// provided decoder. The tieringConfig parameter indicates whether the block
+// contains tiering columns that should be decoded lazily.
 func (i *DataBlockIter) Init(
-	d *DataBlockDecoder, bd BlockDecoder, transforms blockiter.Transforms,
+	d *DataBlockDecoder,
+	bd BlockDecoder,
+	transforms blockiter.Transforms,
+	tieringConfig OptionalColumnConfig,
 ) error {
 	i.d = d
 	// Leave i.h unchanged.
@@ -1077,6 +1249,13 @@ func (i *DataBlockIter) Init(
 	i.kv = base.InternalKV{}
 	i.kvRow = math.MinInt
 	i.nextObsoletePoint = 0
+
+	// Reset tiering state for lazy initialization.
+	i.tieringConfig = tieringConfig
+	i.tieringSpanIDs = UnsafeUints{}
+	i.tieringAttributes = UnsafeUints{}
+	// Store block data for lazy tiering column decoding.
+	i.blockData = bd.Data()
 	return nil
 }
 
@@ -1110,6 +1289,12 @@ func (i *DataBlockIter) InitHandle(
 	i.kvRow = math.MinInt
 	i.nextObsoletePoint = 0
 	i.keySeeker = i.keySchema.KeySeeker(keySeekerMeta)
+
+	// Reset tiering state for lazy initialization. Block data will be obtained
+	// from h.BlockData() when needed.
+	i.tieringSpanIDs = UnsafeUints{}
+	i.tieringAttributes = UnsafeUints{}
+	i.blockData = nil
 	return nil
 }
 
@@ -1297,11 +1482,45 @@ func (i *DataBlockIter) SeekGEWithMeta(
 	return kv, i.decodeMeta()
 }
 
+// initTieringMetadata lazily initializes the tiering columns if they haven't
+// been decoded yet for the current block, returning true if they are initialized
+// or have already been decoded.
+func (i *DataBlockIter) initTieringMetadata() bool {
+	if !i.tieringConfig.SupportsTiering() {
+		return false
+	}
+	if i.tieringSpanIDs.ptr != nil {
+		return true
+	}
+	// Get block data from either the stored blockData (Init path) or the
+	// buffer handle (InitHandle path).
+	var data []byte
+	if i.blockData != nil {
+		data = i.blockData
+	} else if i.h.Valid() {
+		data = i.h.BlockData()
+	}
+	if data == nil {
+		// No data available; tiering columns cannot be decoded.
+		return false
+	}
+
+	// Decode the tiering columns.
+	bd := DecodeBlock(data, DataBlockCustomHeaderSize+i.keySchema.HeaderSize)
+	i.tieringSpanIDs = bd.Uints(len(i.keySchema.ColumnTypes) + dataBlockColumnTieringSpanID)
+	i.tieringAttributes = bd.Uints(len(i.keySchema.ColumnTypes) + dataBlockColumnTieringAttribute)
+	return true
+}
+
 // decodeMeta extracts the KVMeta for the current row.
 func (i *DataBlockIter) decodeMeta() base.KVMeta {
-	// TODO (annie): Implement tiering metadata extraction when the fields are
-	// available.
-	return base.KVMeta{}
+	if !i.initTieringMetadata() {
+		return base.KVMeta{}
+	}
+	return base.KVMeta{
+		TieringSpanID:    i.tieringSpanIDs.At(i.row),
+		TieringAttribute: base.TieringAttribute(i.tieringAttributes.At(i.row)),
+	}
 }
 
 // Last implements the base.InternalIterator interface.
diff --git a/sstable/colblk_writer.go b/sstable/colblk_writer.go
index 5fbde21f..6f94ef75 100644
--- a/sstable/colblk_writer.go
+++ b/sstable/colblk_writer.go
@@ -138,7 +138,7 @@ func newColumnarWriter(
 	w.layout.Init(writable, o)
 	w.dataFlush = block.MakeFlushGovernor(o.BlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses)
 	w.indexFlush = block.MakeFlushGovernor(o.IndexBlockSize, o.BlockSizeThreshold, o.SizeClassAwareThreshold, o.AllocatorSizeClasses)
-	w.dataBlock.Init(o.KeySchema)
+	w.dataBlock.Init(o.KeySchema, o.TableFormat.TieringColumnConfig())
 	w.indexBlock.Init()
 	w.topLevelIndexBlock.Init()
 	w.rangeDelBlock.Init(w.comparer.Equal)
@@ -364,7 +364,9 @@ func (w *RawColumnWriter) EncodeSpan(span keyspan.Span) error {
 // that strict-obsolete ssts must satisfy. S2, due to RANGEDELs, is solely the
 // responsibility of the caller. S1 is solely the responsibility of the
 // callee.
-func (w *RawColumnWriter) Add(key InternalKey, value []byte, forceObsolete bool) error {
+func (w *RawColumnWriter) Add(
+	key InternalKey, value []byte, forceObsolete bool, meta base.KVMeta,
+) error {
 	switch key.Kind() {
 	case base.InternalKeyKindRangeDelete, base.InternalKeyKindRangeKeySet,
 		base.InternalKeyKindRangeKeyUnset, base.InternalKeyKindRangeKeyDelete:
@@ -410,12 +412,16 @@ func (w *RawColumnWriter) Add(key InternalKey, value []byte, forceObsolete bool)
 			valuePrefix = block.InPlaceValuePrefix(eval.kcmp.PrefixEqual())
 		}
 	}
-	return w.add(key, len(value), valueStoredWithKey, valuePrefix, eval)
+	return w.add(key, len(value), valueStoredWithKey, valuePrefix, eval, meta)
 }
 
 // AddWithBlobHandle implements the RawWriter interface.
 func (w *RawColumnWriter) AddWithBlobHandle(
-	key InternalKey, h blob.InlineHandle, attr base.ShortAttribute, forceObsolete bool,
+	key InternalKey,
+	h blob.InlineHandle,
+	attr base.ShortAttribute,
+	forceObsolete bool,
+	meta base.KVMeta,
 ) error {
 	// Blob value handles require at least TableFormatPebblev6.
 	if w.opts.TableFormat <= TableFormatPebblev5 {
@@ -441,7 +447,7 @@ func (w *RawColumnWriter) AddWithBlobHandle(
 	n := h.Encode(w.tmp[:])
 	valueStoredWithKey := w.tmp[:n]
 	valuePrefix := block.BlobValueHandlePrefix(eval.kcmp.PrefixEqual(), attr)
-	err = w.add(key, int(h.ValueLen), valueStoredWithKey, valuePrefix, eval)
+	err = w.add(key, int(h.ValueLen), valueStoredWithKey, valuePrefix, eval, meta)
 	if err != nil {
 		return err
 	}
@@ -458,12 +464,13 @@ func (w *RawColumnWriter) add(
 	valueStoredWithKey []byte,
 	valuePrefix block.ValuePrefix,
 	eval pointKeyEvaluation,
+	meta base.KVMeta,
 ) error {
 	// Append the key to the data block. We have NOT yet committed to
 	// including the key in the block. The data block writer permits us to
 	// finish the block excluding the last-appended KV.
 	entriesWithoutKV := w.dataBlock.Rows()
-	w.dataBlock.Add(key, valueStoredWithKey, valuePrefix, eval.kcmp, eval.isObsolete)
+	w.dataBlock.Add(key, valueStoredWithKey, valuePrefix, eval.kcmp, eval.isObsolete, meta)
 
 	// Now that we've appended the KV pair, we can compute the exact size of the
 	// block with this key-value pair included. Check to see if we should flush
@@ -477,7 +484,7 @@ func (w *RawColumnWriter) add(
 		}
 		// flushDataBlockWithoutNextKey reset the data block builder, and we can
 		// add the key to this next block now.
-		w.dataBlock.Add(key, valueStoredWithKey, valuePrefix, eval.kcmp, eval.isObsolete)
+		w.dataBlock.Add(key, valueStoredWithKey, valuePrefix, eval.kcmp, eval.isObsolete, meta)
 		w.pendingDataBlockSize = w.dataBlock.Size()
 	} else {
 		// We're not flushing the data block, and we're committing to including
@@ -1081,7 +1088,7 @@ func (w *RawColumnWriter) rewriteSuffixes(
 	}
 	// Copy data blocks in parallel, rewriting suffixes as we go.
 	blocks, err := rewriteDataBlocksInParallel(r, sstBytes, wo, l.Data, from, to, concurrency, w.layout.physBlockMaker.Compressor.Stats(), func() blockRewriter {
-		return colblk.NewDataBlockRewriter(wo.KeySchema, w.comparer)
+		return colblk.NewDataBlockRewriter(wo.KeySchema, w.comparer, wo.TableFormat.TieringColumnConfig())
 	})
 	if err != nil {
 		return errors.Wrap(err, "rewriting data blocks")
diff --git a/sstable/format.go b/sstable/format.go
index 4d0232b1..c587641a 100644
--- a/sstable/format.go
+++ b/sstable/format.go
@@ -56,6 +56,10 @@ const (
 	// Supported by CockroachDB v25.3 and later.
 	TableFormatPebblev7
 
+	// TableFormatPebblev8 adds:
+	//  - support for tiering metadata (spanID, key).
+	TableFormatPebblev8
+
 	NumTableFormats
 
 	TableFormatMax = NumTableFormats - 1
@@ -76,6 +80,7 @@ var footerSizes [NumTableFormats]int = [NumTableFormats]int{
 	TableFormatPebblev5:  rocksDBFooterLen,
 	TableFormatPebblev6:  checkedPebbleDBFooterLen,
 	TableFormatPebblev7:  pebbleDBv7FooterLen,
+	TableFormatPebblev8:  pebbleDBv7FooterLen,
 }
 
 // TableFormatPebblev4, in addition to DELSIZED, introduces the use of
@@ -279,6 +284,8 @@ func parseTableFormat(magic []byte, version uint32) (TableFormat, error) {
 			return TableFormatPebblev6, nil
 		case 7:
 			return TableFormatPebblev7, nil
+		case 8:
+			return TableFormatPebblev8, nil
 		default:
 			return TableFormatUnspecified, base.CorruptionErrorf(
 				"(unsupported pebble format version %d)", errors.Safe(version))
@@ -295,6 +302,19 @@ func (f TableFormat) BlockColumnar() bool {
 	return f >= TableFormatPebblev5
 }
 
+// TieringMetadata returns true iff the table format supports tiering metadata.
+func (f TableFormat) TieringMetadata() bool {
+	return f >= TableFormatPebblev8
+}
+
+// TieringColumnConfig returns the TieringColumnConfig for this table format.
+func (f TableFormat) TieringColumnConfig() colblk.OptionalColumnConfig {
+	if f.TieringMetadata() {
+		return colblk.WithTieringColumns()
+	}
+	return colblk.NoTieringColumns()
+}
+
 // FooterSize returns the maximum size of the footer for the table format.
 func (f TableFormat) FooterSize() int {
 	return footerSizes[f]
@@ -328,6 +348,8 @@ func (f TableFormat) AsTuple() (string, uint32) {
 		return pebbleDBMagic, 6
 	case TableFormatPebblev7:
 		return pebbleDBMagic, 7
+	case TableFormatPebblev8:
+		return pebbleDBMagic, 8
 	default:
 		panic("sstable: unknown table format version tuple")
 	}
@@ -356,6 +378,8 @@ func (f TableFormat) String() string {
 		return "(Pebble,v6)"
 	case TableFormatPebblev7:
 		return "(Pebble,v7)"
+	case TableFormatPebblev8:
+		return "(Pebble,v8)"
 	default:
 		panic("sstable: unknown table format version tuple")
 	}
diff --git a/sstable/layout.go b/sstable/layout.go
index 0b220de0..6871feb4 100644
--- a/sstable/layout.go
+++ b/sstable/layout.go
@@ -450,8 +450,9 @@ func formatColblkDataBlock(
 
 	if fmtKV != nil {
 		var iter colblk.DataBlockIter
-		iter.InitOnce(r.keySchema, r.Comparer, describingLazyValueHandler{})
-		if err := iter.Init(&decoder, bd, blockiter.Transforms{}); err != nil {
+		tieringConfig := r.tableFormat.TieringColumnConfig()
+		iter.InitOnce(r.keySchema, r.Comparer, describingLazyValueHandler{}, tieringConfig)
+		if err := iter.Init(&decoder, bd, blockiter.Transforms{}, tieringConfig); err != nil {
 			return err
 		}
 		defer func() { _ = iter.Close() }()
diff --git a/sstable/reader_iter_single_lvl.go b/sstable/reader_iter_single_lvl.go
index 76b7d261..e8724ac9 100644
--- a/sstable/reader_iter_single_lvl.go
+++ b/sstable/reader_iter_single_lvl.go
@@ -228,7 +228,7 @@ func newColumnBlockSingleLevelIterator(
 			i, opts.ReaderProvider, r.valueBIH, opts.Env.Block.Stats, opts.Env.Block.IterStats)
 		i.vbRH = r.blockReader.UsePreallocatedReadHandle(objstorage.NoReadBefore, &i.vbRHPrealloc)
 	}
-	i.data.InitOnce(r.keySchema, r.Comparer, &i.internalValueConstructor)
+	i.data.InitOnce(r.keySchema, r.Comparer, &i.internalValueConstructor, r.tableFormat.TieringColumnConfig())
 
 	return i, nil
 }
diff --git a/sstable/reader_iter_two_lvl.go b/sstable/reader_iter_two_lvl.go
index 768a2f67..d4c1873d 100644
--- a/sstable/reader_iter_two_lvl.go
+++ b/sstable/reader_iter_two_lvl.go
@@ -201,7 +201,8 @@ func newColumnBlockTwoLevelIterator(
 		i.secondLevel.vbRH = r.blockReader.UsePreallocatedReadHandle(
 			objstorage.NoReadBefore, &i.secondLevel.vbRHPrealloc)
 	}
-	i.secondLevel.data.InitOnce(r.keySchema, r.Comparer, &i.secondLevel.internalValueConstructor)
+	i.secondLevel.data.InitOnce(r.keySchema,
+		r.Comparer, &i.secondLevel.internalValueConstructor, r.tableFormat.TieringColumnConfig())
 
 	return i, nil
 }
diff --git a/sstable/rowblk_writer.go b/sstable/rowblk_writer.go
index daf004d6..fd4daa9c 100644
--- a/sstable/rowblk_writer.go
+++ b/sstable/rowblk_writer.go
@@ -515,7 +515,7 @@ type bufferedIndexBlock struct {
 // that strict-obsolete ssts must satisfy. S2, due to RANGEDELs, is solely the
 // responsibility of the caller. S1 is solely the responsibility of the
 // callee.
-func (w *RawRowWriter) Add(key InternalKey, value []byte, forceObsolete bool) error {
+func (w *RawRowWriter) Add(key InternalKey, value []byte, forceObsolete bool, _ base.KVMeta) error {
 	if w.err != nil {
 		return w.err
 	}
@@ -536,7 +536,7 @@ func (w *RawRowWriter) Add(key InternalKey, value []byte, forceObsolete bool) er
 // AddWithBlobHandle implements the RawWriter interface. This implementation
 // does not support writing blob value handles.
 func (w *RawRowWriter) AddWithBlobHandle(
-	key InternalKey, h blob.InlineHandle, attr base.ShortAttribute, forceObsolete bool,
+	key InternalKey, h blob.InlineHandle, attr base.ShortAttribute, forceObsolete bool, _ base.KVMeta,
 ) error {
 	w.err = errors.Newf("pebble: blob value handles are not supported in %s", w.tableFormat.String())
 	return w.err
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index cec02a0c..48b8e9f5 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -355,7 +355,7 @@ func RewriteKeySuffixesViaWriter(
 		if err != nil {
 			return nil, err
 		}
-		if err := w.Add(scratch, val, false); err != nil {
+		if err := w.Add(scratch, val, false, base.KVMeta{}); err != nil {
 			return nil, err
 		}
 		kv = i.Next()
diff --git a/sstable/writer.go b/sstable/writer.go
index 6d9440c9..97f5374c 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -126,7 +126,7 @@ func (w *Writer) Set(key, value []byte) error {
 	}
 	// forceObsolete is false based on the assumption that no RANGEDELs in the
 	// sstable delete the added points.
-	return w.rw.Add(base.MakeInternalKey(key, 0, InternalKeyKindSet), value, false)
+	return w.rw.Add(base.MakeInternalKey(key, 0, InternalKeyKindSet), value, false, base.KVMeta{})
 }
 
 // Delete deletes the value for the given key. The sequence number is set to
@@ -143,7 +143,7 @@ func (w *Writer) Delete(key []byte) error {
 	}
 	// forceObsolete is false based on the assumption that no RANGEDELs in the
 	// sstable delete the added points.
-	return w.rw.Add(base.MakeInternalKey(key, 0, InternalKeyKindDelete), nil, false)
+	return w.rw.Add(base.MakeInternalKey(key, 0, InternalKeyKindDelete), nil, false, base.KVMeta{})
 }
 
 // DeleteRange deletes all of the keys (and values) in the range [start,end)
@@ -184,7 +184,7 @@ func (w *Writer) Merge(key, value []byte) error {
 	// forceObsolete is false based on the assumption that no RANGEDELs in the
 	// sstable that delete the added points. If the user configured this writer
 	// to be strict-obsolete, addPoint will reject the addition of this MERGE.
-	return w.rw.Add(base.MakeInternalKey(key, 0, InternalKeyKindMerge), value, false)
+	return w.rw.Add(base.MakeInternalKey(key, 0, InternalKeyKindMerge), value, false, base.KVMeta{})
 }
 
 // RangeKeySet sets a range between start (inclusive) and end (exclusive) with
@@ -325,11 +325,12 @@ type RawWriter interface {
 	// that strict-obsolete ssts must satisfy. S2, due to RANGEDELs, is solely the
 	// responsibility of the caller. S1 is solely the responsibility of the
 	// callee.
-	Add(key InternalKey, value []byte, forceObsolete bool) error
+	Add(key InternalKey, value []byte, forceObsolete bool, meta base.KVMeta) error
 	// AddWithBlobHandle adds a key to the sstable, but encoding a blob value
 	// handle instead of an in-place value. See Add for more details. The caller
 	// must provide the already-extracted ShortAttribute for the value.
-	AddWithBlobHandle(key InternalKey, h blob.InlineHandle, attr base.ShortAttribute, forceObsolete bool) error
+	AddWithBlobHandle(key InternalKey, h blob.InlineHandle, attr base.ShortAttribute,
+		forceObsolete bool, meta base.KVMeta) error
 	// EncodeSpan encodes the keys in the given span. The span can contain
 	// either only RANGEDEL keys or only range keys.
 	//
@@ -482,14 +483,20 @@ type LoggingRawWriter struct {
 	RawWriter
 }
 
-func (w *LoggingRawWriter) Add(key base.InternalKey, value []byte, forceObsolete bool) error {
+func (w *LoggingRawWriter) Add(
+	key base.InternalKey, value []byte, forceObsolete bool, meta base.KVMeta,
+) error {
 	fmt.Fprintf(w.LogWriter, "RawWriter.Add(%q, %q, %t)\n", key, value, forceObsolete)
-	return w.RawWriter.Add(key, value, forceObsolete)
+	return w.RawWriter.Add(key, value, forceObsolete, meta)
 }
 
 func (w *LoggingRawWriter) AddWithBlobHandle(
-	key base.InternalKey, h blob.InlineHandle, attr base.ShortAttribute, forceObsolete bool,
+	key base.InternalKey,
+	h blob.InlineHandle,
+	attr base.ShortAttribute,
+	forceObsolete bool,
+	meta base.KVMeta,
 ) error {
 	fmt.Fprintf(w.LogWriter, "RawWriter.AddWithBlobHandle(%q, %q, %x, %t)\n", key, h, attr, forceObsolete)
-	return w.RawWriter.AddWithBlobHandle(key, h, attr, forceObsolete)
+	return w.RawWriter.AddWithBlobHandle(key, h, attr, forceObsolete, meta)
 }
diff --git a/valsep/sst_blob_writer.go b/valsep/sst_blob_writer.go
index a34a662a..63d6f276 100644
--- a/valsep/sst_blob_writer.go
+++ b/valsep/sst_blob_writer.go
@@ -142,7 +142,7 @@ func (w *SSTBlobWriter) Set(key, value []byte) error {
 	w.kvScratch.K = base.MakeInternalKey(key, 0, sstable.InternalKeyKindSet)
 	w.kvScratch.V = base.MakeInPlaceValue(value)
 	isLikelyMVCCGarbage := w.SSTWriter.Raw().IsLikelyMVCCGarbage(w.kvScratch.K.UserKey, w.kvScratch.Kind())
-	return w.valSep.Add(w.SSTWriter.Raw(), &w.kvScratch, false, isLikelyMVCCGarbage)
+	return w.valSep.Add(w.SSTWriter.Raw(), &w.kvScratch, false, isLikelyMVCCGarbage, base.KVMeta{})
 }
 
 // BlobWriterMetas returns a slice of blob.FileWriterStats describing the
diff --git a/valsep/value_separation.go b/valsep/value_separation.go
index b645ec74..5a1ba74a 100644
--- a/valsep/value_separation.go
+++ b/valsep/value_separation.go
@@ -48,8 +48,9 @@ type ValueSeparation interface {
 	// the current output sstable's blob references so far.
 	EstimatedReferenceSize() uint64
 	// Add adds the provided key-value pair to the provided sstable writer,
-	// possibly separating the value into a blob file.
-	Add(tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, isLikelyMVCCGarbage bool) error
+	// possibly separating the value into a blob file. The provided meta is
+	// the iterator-derived KV metadata associated with the current key/value.
+	Add(tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, isLikelyMVCCGarbage bool, meta base.KVMeta) error
 	// FinishOutput is called when a compaction is finishing an output sstable.
 	// It returns the table's blob references, which will be added to the
 	// table's TableMetadata, and stats and metadata describing a newly
@@ -103,13 +104,13 @@ func (NeverSeparateValues) EstimatedReferenceSize() uint64 { return 0 }
 
 // Add implements the ValueSeparation interface.
 func (NeverSeparateValues) Add(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ bool,
+	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ bool, meta base.KVMeta,
 ) error {
 	v, _, err := kv.Value(nil)
 	if err != nil {
 		return err
 	}
-	return tw.Add(kv.K, v, forceObsolete)
+	return tw.Add(kv.K, v, forceObsolete, meta)
 }
 
 // FinishOutput implements the ValueSeparation interface.
diff --git a/valsep/value_separator.go b/valsep/value_separator.go
index a98067a8..c614f7db 100644
--- a/valsep/value_separator.go
+++ b/valsep/value_separator.go
@@ -219,10 +219,14 @@ func (vs *ValueSeparator) EstimatedReferenceSize() uint64 {
 
 // Add implements ValueSeparation.
 func (vs *ValueSeparator) Add(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, isLikelyMVCCGarbage bool,
+	tw sstable.RawWriter,
+	kv *base.InternalKV,
+	forceObsolete bool,
+	isLikelyMVCCGarbage bool,
+	meta base.KVMeta,
 ) error {
 	if kv.V.IsBlobValueHandle() && vs.mode == preserveAllHotBlobReferences {
-		return vs.preserveBlobReference(tw, kv, forceObsolete)
+		return vs.preserveBlobReference(tw, kv, forceObsolete, meta)
 	}
 
 	// Fetch the value since either not a reference, or not preserving the reference.
@@ -240,7 +244,7 @@ func (vs *ValueSeparator) Add(
 		// sstable fileWriter may still decide to put the value in a value block,
 		// but regardless the value will be written to the sstable itself and
 		// not a blob file.
-		return tw.Add(kv.K, v, forceObsolete)
+		return tw.Add(kv.K, v, forceObsolete, meta)
 	}
 	// We are rewriting all hot blob references. Check that the value meets the criteria
 	// for separation.
@@ -249,18 +253,18 @@ func (vs *ValueSeparator) Add(
 	keyKind := kv.K.Kind()
 	if keyKind != base.InternalKeyKindSet && keyKind != base.InternalKeyKindSetWithDelete {
 		// Only SET and SETWITHDEL can be separated.
-		return tw.Add(kv.K, v, forceObsolete)
+		return tw.Add(kv.K, v, forceObsolete, meta)
 	}
 
 	// Values that are too small are never separated; however, likely MVCC
 	// garbage (determined by sstable.IsLikelyMVCCGarbage) is separated as long
 	// as DisableValueSeparationBySuffix is not enabled.
 	if len(v) < vs.currentConfig.MinimumSize && (vs.currentConfig.DisableValueSeparationBySuffix || !isLikelyMVCCGarbage) {
-		return tw.Add(kv.K, v, forceObsolete)
+		return tw.Add(kv.K, v, forceObsolete, meta)
 	}
 
 	// This KV met all the criteria and its value will be separated.
-	return vs.separateValue(tw, kv, v, forceObsolete, isLikelyMVCCGarbage)
+	return vs.separateValue(tw, kv, v, forceObsolete, isLikelyMVCCGarbage, meta)
 }
 
 // separateValue separates the value into a blob file and writes a blob handle
@@ -272,6 +276,7 @@ func (vs *ValueSeparator) separateValue(
 	rawValue []byte,
 	forceObsolete bool,
 	isLikelyMVCCGarbage bool,
+	meta base.KVMeta,
 ) (err error) {
 	if vs.mode == preserveAllHotBlobReferences {
 		return errors.AssertionFailedf("separateValue called in preserveAllHotBlobReferences mode")
@@ -294,7 +299,7 @@ func (vs *ValueSeparator) separateValue(
 			// fallback to writing the value verbatim to the sstable. Otherwise
 			// a flush could busy loop, repeatedly attempting to write the same
 			// memtable and repeatedly unable to extract a key's short attribute.
-			return tw.Add(kv.K, rawValue, forceObsolete)
+			return tw.Add(kv.K, rawValue, forceObsolete, meta)
 		}
 	}
 
@@ -317,14 +322,14 @@ func (vs *ValueSeparator) separateValue(
 			ValueID: handle.ValueID,
 		},
 	}
-	return tw.AddWithBlobHandle(kv.K, inlineHandle, shortAttr, forceObsolete)
+	return tw.AddWithBlobHandle(kv.K, inlineHandle, shortAttr, forceObsolete, meta)
 }
 
 // preserveBlobReference preserves an existing blob reference by copying it
 // into the output sstable. The provided kv must have a value that is an
 // existing blob handle.
 func (vs *ValueSeparator) preserveBlobReference(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool,
+	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, meta base.KVMeta,
 ) error {
 	// We are preserving blob references and the value is an existing blob handle.
 	// We can copy it into the output sstable, taking note of the reference for
@@ -357,7 +362,8 @@ func (vs *ValueSeparator) preserveBlobReference(
 		},
 		HandleSuffix: handleSuffix,
 	}
-	err := tw.AddWithBlobHandle(kv.K, inlineHandle, lv.Fetcher.Attribute.ShortAttribute, forceObsolete)
+	err := tw.AddWithBlobHandle(kv.K, inlineHandle, lv.Fetcher.Attribute.ShortAttribute,
+		forceObsolete, meta)
 	if err != nil {
 		return err
 	}



Test patch:
diff --git a/blob_rewrite_test.go b/blob_rewrite_test.go
index a4a54d06..4a5eed25 100644
--- a/blob_rewrite_test.go
+++ b/blob_rewrite_test.go
@@ -125,7 +125,7 @@ func TestBlobRewrite(t *testing.T) {
 						ikv.V, err = bv.ParseInternalValue(value)
 						require.NoError(t, err)
 					}
-					require.NoError(t, vs.Add(tw, &ikv, false /* forceObsolete */, false /* isLikeyMVCCGarbage */))
+					require.NoError(t, vs.Add(tw, &ikv, false /* forceObsolete */, false /* isLikelyMVCCGarbage */, base.KVMeta{}))
 				}
 				return buf.String()
 			case "close-output":
@@ -317,6 +317,7 @@ func TestBlobRewriteRandomized(t *testing.T) {
 			},
 			base.ShortAttribute(0),
 			false, /* forceObsolete */
+			base.KVMeta{},
 		))
 		require.NoError(t, tw.Close())
 		originalValueIndices[i] = i
diff --git a/cockroachkvs/cockroachkvs_bench_test.go b/cockroachkvs/cockroachkvs_bench_test.go
index d06e8c2d..149f944d 100644
--- a/cockroachkvs/cockroachkvs_bench_test.go
+++ b/cockroachkvs/cockroachkvs_bench_test.go
@@ -193,7 +193,7 @@ func benchmarkCockroachDataColBlockWriter(b *testing.B, keyConfig KeyGenConfig,
 	_, keys, values := generateDataBlock(rng, targetBlockSize, keyConfig, valueLen)
 
 	var w colblk.DataBlockEncoder
-	w.Init(&KeySchema)
+	w.Init(&KeySchema, colblk.NoTieringColumns())
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
@@ -202,7 +202,8 @@ func benchmarkCockroachDataColBlockWriter(b *testing.B, keyConfig KeyGenConfig,
 		for w.Size() < targetBlockSize {
 			ik := base.MakeInternalKey(keys[count], base.SeqNum(rng.Uint64N(uint64(base.SeqNumMax))), base.InternalKeyKindSet)
 			kcmp := w.KeyWriter.ComparePrev(ik.UserKey)
-			w.Add(ik, values[count], block.InPlaceValuePrefix(kcmp.PrefixEqual()), kcmp, false /* isObsolete */)
+			w.Add(ik, values[count], block.InPlaceValuePrefix(kcmp.PrefixEqual()), kcmp,
+				false /* isObsolete */, base.KVMeta{})
 			count++
 		}
 		_, _ = w.Finish(w.Rows(), w.Size())
@@ -317,9 +318,9 @@ func benchmarkCockroachDataColBlockIter(
 	var it colblk.DataBlockIter
 	it.InitOnce(&KeySchema, &Comparer, getInternalValuer(func([]byte) base.InternalValue {
 		return base.MakeInPlaceValue([]byte("mock external value"))
-	}))
+	}), colblk.NoTieringColumns())
 	bd := decoder.Init(&KeySchema, serializedBlock)
-	if err := it.Init(&decoder, bd, transforms); err != nil {
+	if err := it.Init(&decoder, bd, transforms, colblk.NoTieringColumns()); err != nil {
 		b.Fatal(err)
 	}
 	avgRowSize := float64(len(serializedBlock)) / float64(len(keys))
@@ -382,12 +383,12 @@ func BenchmarkInitDataBlockMetadata(b *testing.B) {
 	}, 8)
 
 	var w colblk.DataBlockEncoder
-	w.Init(&KeySchema)
+	w.Init(&KeySchema, colblk.NoTieringColumns())
 	for j := 0; w.Size() < targetBlockSize; j++ {
 		ik := base.MakeInternalKey(keys[j], base.SeqNum(rng.Uint64N(uint64(base.SeqNumMax))), base.InternalKeyKindSet)
 		kcmp := w.KeyWriter.ComparePrev(ik.UserKey)
 		vp := block.InPlaceValuePrefix(kcmp.PrefixEqual())
-		w.Add(ik, values[j], vp, kcmp, false /* isObsolete */)
+		w.Add(ik, values[j], vp, kcmp, false /* isObsolete */, base.KVMeta{})
 	}
 	finished, _ := w.Finish(w.Rows(), w.Size())
 
diff --git a/cockroachkvs/cockroachkvs_test.go b/cockroachkvs/cockroachkvs_test.go
index 582d6c35..72b012c3 100644
--- a/cockroachkvs/cockroachkvs_test.go
+++ b/cockroachkvs/cockroachkvs_test.go
@@ -240,7 +240,7 @@ func TestKeySchema_KeySeeker(t *testing.T) {
 	var bd colblk.BlockDecoder
 	var ks colblk.KeySeeker
 	var maxKeyLen int
-	enc.Init(&KeySchema)
+	enc.Init(&KeySchema, colblk.NoTieringColumns())
 
 	initKeySeeker := func() {
 		ksPointer := &cockroachKeySeeker{}
@@ -264,7 +264,7 @@ func TestKeySchema_KeySeeker(t *testing.T) {
 					UserKey: k,
 					Trailer: pebble.MakeInternalKeyTrailer(0, base.InternalKeyKindSet),
 				}
-				enc.Add(ikey, k, block.InPlaceValuePrefix(false), kcmp, false /* isObsolete */)
+				enc.Add(ikey, k, block.InPlaceValuePrefix(false), kcmp, false /* isObsolete */, base.KVMeta{})
 				rows++
 			}
 			blk, _ := enc.Finish(rows, enc.Size())
@@ -335,7 +335,7 @@ func TestKeySeekerIsLowerBound(t *testing.T) {
 		}
 	}
 	var enc colblk.DataBlockEncoder
-	enc.Init(&KeySchema)
+	enc.Init(&KeySchema, colblk.NoTieringColumns())
 	for _, k := range keys {
 		enc.Reset()
 		kcmp := enc.KeyWriter.ComparePrev(k)
@@ -343,7 +343,7 @@ func TestKeySeekerIsLowerBound(t *testing.T) {
 			UserKey: k,
 			Trailer: pebble.MakeInternalKeyTrailer(0, base.InternalKeyKindSet),
 		}
-		enc.Add(ikey, k, block.InPlaceValuePrefix(false), kcmp, false /* isObsolete */)
+		enc.Add(ikey, k, block.InPlaceValuePrefix(false), kcmp, false /* isObsolete */, base.KVMeta{})
 		blk, _ := enc.Finish(1, enc.Size())
 		var dec colblk.DataBlockDecoder
 		bd := dec.Init(&KeySchema, blk)
@@ -480,9 +480,9 @@ func testCockroachDataColBlock(t *testing.T, seed uint64, keyCfg KeyGenConfig) {
 	var it colblk.DataBlockIter
 	it.InitOnce(&KeySchema, &Comparer, getInternalValuer(func([]byte) base.InternalValue {
 		return base.MakeInPlaceValue([]byte("mock external value"))
-	}))
+	}), colblk.NoTieringColumns())
 	bd := decoder.Init(&KeySchema, serializedBlock)
-	if err := it.Init(&decoder, bd, blockiter.Transforms{}); err != nil {
+	if err := it.Init(&decoder, bd, blockiter.Transforms{}, colblk.NoTieringColumns()); err != nil {
 		t.Fatal(err)
 	}
 	// Scan the block using Next and ensure that all the keys values match.
@@ -528,12 +528,12 @@ func generateDataBlock(
 	keys, values = RandomKVs(rng, targetBlockSize/valueLen, cfg, valueLen)
 
 	var w colblk.DataBlockEncoder
-	w.Init(&KeySchema)
+	w.Init(&KeySchema, colblk.NoTieringColumns())
 	count := 0
 	for w.Size() < targetBlockSize {
 		ik := base.MakeInternalKey(keys[count], base.SeqNum(rng.Uint64N(uint64(base.SeqNumMax))), base.InternalKeyKindSet)
 		kcmp := w.KeyWriter.ComparePrev(ik.UserKey)
-		w.Add(ik, values[count], block.InPlaceValuePrefix(kcmp.PrefixEqual()), kcmp, false /* isObsolete */)
+		w.Add(ik, values[count], block.InPlaceValuePrefix(kcmp.PrefixEqual()), kcmp, false /* isObsolete */, base.KVMeta{})
 		count++
 	}
 	data, _ = w.Finish(w.Rows(), w.Size())
diff --git a/cockroachkvs/key_schema_test.go b/cockroachkvs/key_schema_test.go
index f0ff36f1..3d508c81 100644
--- a/cockroachkvs/key_schema_test.go
+++ b/cockroachkvs/key_schema_test.go
@@ -40,9 +40,9 @@ func TestKeySchema(t *testing.T) {
 func runDataDrivenTest(t *testing.T, path string) {
 	var blockData []byte
 	var e colblk.DataBlockEncoder
-	e.Init(&KeySchema)
+	e.Init(&KeySchema, colblk.NoTieringColumns())
 	var iter colblk.DataBlockIter
-	iter.InitOnce(&KeySchema, &Comparer, nil)
+	iter.InitOnce(&KeySchema, &Comparer, nil, colblk.NoTieringColumns())
 
 	datadriven.RunTest(t, path, func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
@@ -52,7 +52,7 @@ func runDataDrivenTest(t *testing.T, path string) {
 			for l := range crstrings.LinesSeq(td.Input) {
 				key, value := parseInternalKV(l)
 				kcmp := e.KeyWriter.ComparePrev(key.UserKey)
-				e.Add(key, value, 0, kcmp, false /* isObsolete */)
+				e.Add(key, value, 0, kcmp, false /* isObsolete */, base.KVMeta{})
 				buf = e.MaterializeLastUserKey(buf[:0])
 				if !Comparer.Equal(key.UserKey, buf) {
 					td.Fatalf(t, "incorrect MaterializeLastKey: %s instead of %s", formatUserKey(buf), formatUserKey(key.UserKey))
@@ -82,7 +82,7 @@ func runDataDrivenTest(t *testing.T, path string) {
 		case "keys":
 			var d colblk.DataBlockDecoder
 			bd := d.Init(&KeySchema, blockData)
-			require.NoError(t, iter.Init(&d, bd, blockiter.Transforms{}))
+			require.NoError(t, iter.Init(&d, bd, blockiter.Transforms{}, colblk.NoTieringColumns()))
 			defer iter.Close()
 			var buf bytes.Buffer
 			var prevKey base.InternalKey
@@ -99,7 +99,7 @@ func runDataDrivenTest(t *testing.T, path string) {
 		case "seek":
 			var d colblk.DataBlockDecoder
 			bd := d.Init(&KeySchema, blockData)
-			require.NoError(t, iter.Init(&d, bd, blockiter.Transforms{}))
+			require.NoError(t, iter.Init(&d, bd, blockiter.Transforms{}, colblk.NoTieringColumns()))
 			defer iter.Close()
 			var buf strings.Builder
 			for l := range crstrings.LinesSeq(td.Input) {
@@ -133,13 +133,13 @@ func TestKeySchema_RandomKeys(t *testing.T) {
 	slices.SortFunc(keys, Compare)
 
 	var enc colblk.DataBlockEncoder
-	enc.Init(&KeySchema)
+	enc.Init(&KeySchema, colblk.NoTieringColumns())
 	for i := range keys {
 		ikey := pebble.InternalKey{
 			UserKey: keys[i],
 			Trailer: pebble.MakeInternalKeyTrailer(0, pebble.InternalKeyKindSet),
 		}
-		enc.Add(ikey, keys[i], block.InPlaceValuePrefix(false), enc.KeyWriter.ComparePrev(keys[i]), false /* isObsolete */)
+		enc.Add(ikey, keys[i], block.InPlaceValuePrefix(false), enc.KeyWriter.ComparePrev(keys[i]), false /* isObsolete */, base.KVMeta{})
 	}
 	blk, _ := enc.Finish(len(keys), enc.Size())
 	blk = crbytes.CopyAligned(blk)
@@ -147,8 +147,8 @@ func TestKeySchema_RandomKeys(t *testing.T) {
 	var dec colblk.DataBlockDecoder
 	bd := dec.Init(&KeySchema, blk)
 	var it colblk.DataBlockIter
-	it.InitOnce(&KeySchema, &Comparer, nil)
-	require.NoError(t, it.Init(&dec, bd, blockiter.NoTransforms))
+	it.InitOnce(&KeySchema, &Comparer, nil, colblk.NoTieringColumns())
+	require.NoError(t, it.Init(&dec, bd, blockiter.NoTransforms, colblk.NoTieringColumns()))
 	// Ensure that a scan across the block finds all the relevant keys.
 	var valBuf []byte
 	for k, kv := 0, it.First(); kv != nil; k, kv = k+1, it.Next() {
diff --git a/data_test.go b/data_test.go
index 71c59528..5ff14524 100644
--- a/data_test.go
+++ b/data_test.go
@@ -673,13 +673,13 @@ func runBuildCmd(
 			if err != nil {
 				return err
 			}
-			if err := w.Raw().AddWithBlobHandle(tmp, handle, base.ShortAttribute(0), false); err != nil {
+			if err := w.Raw().AddWithBlobHandle(tmp, handle, base.ShortAttribute(0), false, base.KVMeta{}); err != nil {
 				return err
 			}
 			continue
 		}
 		// Otherwise add it as an ordinary value.
-		if err := w.Raw().Add(tmp, v, false); err != nil {
+		if err := w.Raw().Add(tmp, v, false, base.KVMeta{}); err != nil {
 			return err
 		}
 	}
@@ -1970,14 +1970,14 @@ func (vs *defineDBValueSeparator) EstimatedReferenceSize() uint64 {
 // Add adds the provided key-value pair to the sstable, possibly separating the
 // value into a blob file.
 func (vs *defineDBValueSeparator) Add(
-	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ bool,
+	tw sstable.RawWriter, kv *base.InternalKV, forceObsolete bool, _ bool, kvMeta base.KVMeta,
 ) error {
 	// In datadriven tests, all defined values are in-place initially. See
 	// runDBDefineCmdReuseFS.
 	v := kv.V.InPlaceValue()
 	// If the value doesn't begin with "blob", don't separate it.
 	if !bytes.HasPrefix(v, []byte("blob")) {
-		return tw.Add(kv.K, v, forceObsolete)
+		return tw.Add(kv.K, v, forceObsolete, kvMeta)
 	}
 
 	// This looks like a blob reference. Parse it.
@@ -2002,7 +2002,7 @@ func (vs *defineDBValueSeparator) Add(
 	// Return a KV that uses the original key but our constructed blob reference.
 	vs.kv.K = kv.K
 	vs.kv.V = iv
-	return vs.pbr.Add(tw, &vs.kv, forceObsolete, false /* isLikelyMVCCGarbage */)
+	return vs.pbr.Add(tw, &vs.kv, forceObsolete, false /* isLikelyMVCCGarbage */, kvMeta)
 }
 
 // FinishOutput implements valsep.ValueSeparation.
diff --git a/excise_test.go b/excise_test.go
index 659b2510..f24532ec 100644
--- a/excise_test.go
+++ b/excise_test.go
@@ -592,7 +592,8 @@ func TestConcurrentExcise(t *testing.T) {
 				VisitPointKey: func(key *InternalKey, value LazyValue, _ IteratorLevel) error {
 					val, _, err := value.Value(nil)
 					require.NoError(t, err)
-					require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false /* forceObsolete */))
+					require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val,
+						false /* forceObsolete */, base.KVMeta{}))
 					return nil
 				},
 				VisitRangeDel: func(start, end []byte, seqNum base.SeqNum) error {
diff --git a/file_cache_test.go b/file_cache_test.go
index 8f1575ee..0b302682 100644
--- a/file_cache_test.go
+++ b/file_cache_test.go
@@ -219,7 +219,7 @@ func (t *fileCacheTest) newTestHandle() (*fileCacheHandle, *fileCacheTestFS) {
 		}
 		tw := sstable.NewWriter(w, sstable.WriterOptions{TableFormat: sstable.TableFormatPebblev2})
 		ik := base.MakeInternalKey([]byte("k"), base.SeqNum(i), InternalKeyKindSet)
-		if err := tw.Raw().Add(ik, xxx[:i], false); err != nil {
+		if err := tw.Raw().Add(ik, xxx[:i], false, base.KVMeta{}); err != nil {
 			t.Fatal(err)
 		}
 		if err := tw.RangeKeySet([]byte("k"), []byte("l"), nil, xxx[:i]); err != nil {
diff --git a/ingest_test.go b/ingest_test.go
index 973e7a77..27ae3e54 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -130,12 +130,12 @@ func TestIngestLoad(t *testing.T) {
 				if blobtest.IsBlobHandle(data[j+1:]) {
 					ih, _, err := bv.ParseInlineHandle(data[j+1:])
 					require.NoError(t, err)
-					if err := w.AddWithBlobHandle(key, ih, base.ShortAttribute(0), false /* forceObsolete */); err != nil {
+					if err := w.AddWithBlobHandle(key, ih, base.ShortAttribute(0), false /* forceObsolete */, base.KVMeta{}); err != nil {
 						return err.Error()
 					}
 				} else {
 					value := []byte(data[j+1:])
-					if err := w.Add(key, value, false /* forceObsolete */); err != nil {
+					if err := w.Add(key, value, false /* forceObsolete */, base.KVMeta{}); err != nil {
 						return err.Error()
 					}
 				}
@@ -237,7 +237,7 @@ func TestIngestLoadRand(t *testing.T) {
 					// Duplicate key, ignore.
 					continue
 				}
-				require.NoError(t, w.Add(keys[i], nil, false /* forceObsolete */))
+				require.NoError(t, w.Add(keys[i], nil, false /* forceObsolete */, base.KVMeta{}))
 				count++
 				rawKeySize += uint64(keys[i].Size())
 			}
@@ -316,7 +316,7 @@ func TestIngestLocalWithBlobs(t *testing.T) {
 			sstWriterOpts := sstable.WriterOptions{
 				Comparer:    testkeys.Comparer,
 				KeySchema:   &keySchema,
-				TableFormat: sstable.TableFormatMax,
+				TableFormat: internalFormatNewest.MaxTableFormat(),
 			}
 			sstFileName := td.CmdArgs[0].Key
 			if sstFileName == "" {
@@ -1046,7 +1046,7 @@ func testIngestSharedImpl(
 				VisitPointKey: func(key *InternalKey, value LazyValue, _ IteratorLevel) error {
 					val, _, err := value.Value(nil)
 					require.NoError(t, err)
-					require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false /* forceObsolete */))
+					require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false /* forceObsolete */, base.KVMeta{}))
 					return nil
 				},
 				VisitRangeDel: func(start, end []byte, seqNum base.SeqNum) error {
@@ -1566,7 +1566,7 @@ func TestIngestExternal(t *testing.T) {
 				VisitPointKey: func(key *InternalKey, value LazyValue, _ IteratorLevel) error {
 					val, _, err := value.Value(nil)
 					require.NoError(t, err)
-					require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false /* forceObsolete */))
+					require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val, false /* forceObsolete */, base.KVMeta{}))
 					return nil
 				},
 				VisitRangeDel: func(start, end []byte, seqNum base.SeqNum) error {
@@ -2178,7 +2178,7 @@ func TestIngestCompact(t *testing.T) {
 
 	w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
 	key := []byte("a")
-	require.NoError(t, w.Add(base.MakeInternalKey(key, 0, InternalKeyKindSet), nil, false /* forceObsolete */))
+	require.NoError(t, w.Add(base.MakeInternalKey(key, 0, InternalKeyKindSet), nil, false /* forceObsolete */, base.KVMeta{}))
 	require.NoError(t, w.Close())
 
 	// Make N copies of the sstable.
@@ -2815,7 +2815,7 @@ func TestIngest_UpdateSequenceNumber(t *testing.T) {
 			}
 			key := base.ParseInternalKey(data[:j])
 			value := []byte(data[j+1:])
-			if err := w.Add(key, value, false /* forceObsolete */); err != nil {
+			if err := w.Add(key, value, false /* forceObsolete */, base.KVMeta{}); err != nil {
 				return nil, err
 			}
 		}
diff --git a/level_checker_test.go b/level_checker_test.go
index 9ad399f1..3870d445 100644
--- a/level_checker_test.go
+++ b/level_checker_test.go
@@ -213,7 +213,7 @@ func TestCheckLevelsCornerCases(t *testing.T) {
 						}
 						continue
 					}
-					err = w.Add(kv.Key, kv.Value, false /* forceObsolete */)
+					err = w.Add(kv.Key, kv.Value, false /* forceObsolete */, base.KVMeta{})
 					if err != nil {
 						return err.Error()
 					}
diff --git a/level_iter_test.go b/level_iter_test.go
index d4c5e8a7..91968444 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -269,7 +269,7 @@ func (lt *levelIterTest) runBuild(d *datadriven.TestData) string {
 				return err.Error()
 			}
 		default:
-			if err := w.Add(ikey, value, false /* forceObsolete */); err != nil {
+			if err := w.Add(ikey, value, false /* forceObsolete */, base.KVMeta{}); err != nil {
 				return err.Error()
 			}
 		}
@@ -544,7 +544,7 @@ func buildLevelIterTables(
 			key := []byte(fmt.Sprintf("%08d", i))
 			keys = append(keys, key)
 			ikey := base.MakeInternalKey(key, 0, InternalKeyKindSet)
-			require.NoError(b, w.Add(ikey, nil, false /* forceObsolete */))
+			require.NoError(b, w.Add(ikey, nil, false /* forceObsolete */, base.KVMeta{}))
 		}
 		if err := w.Close(); err != nil {
 			b.Fatal(err)
diff --git a/merging_iter_test.go b/merging_iter_test.go
index 1ee28b95..7a84b86b 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -247,7 +247,7 @@ func TestMergingIterDataDriven(t *testing.T) {
 							case InternalKeyKindRangeDelete:
 								frag.Add(keyspan.Span{Start: ikey.UserKey, End: value, Keys: []keyspan.Key{{Trailer: ikey.Trailer}}})
 							default:
-								if err := w.Add(ikey, value, false /* forceObsolete */); err != nil {
+								if err := w.Add(ikey, value, false /* forceObsolete */, base.KVMeta{}); err != nil {
 									return err.Error()
 								}
 							}
@@ -375,7 +375,7 @@ func buildMergingIterTables(
 		ikey.UserKey = key
 		j := rand.IntN(len(writers))
 		w := writers[j]
-		w.Add(ikey, nil, false /* forceObsolete */)
+		w.Add(ikey, nil, false /* forceObsolete */, base.KVMeta{})
 	}
 
 	for _, w := range writers {
@@ -605,7 +605,7 @@ func buildLevelsForMergingIterSeqSeek(
 		key := makeKey(i)
 		keys = append(keys, key)
 		ikey := base.MakeInternalKey(key, 0, InternalKeyKindSet)
-		require.NoError(b, w.Add(ikey, nil, false /* forceObsolete */))
+		require.NoError(b, w.Add(ikey, nil, false /* forceObsolete */, base.KVMeta{}))
 	}
 	if writeRangeTombstoneToLowestLevel {
 		require.NoError(b, w.EncodeSpan(keyspan.Span{
@@ -619,14 +619,14 @@ func buildLevelsForMergingIterSeqSeek(
 	for j := 1; j < len(files); j++ {
 		for _, k := range []int{0, len(keys) - 1} {
 			ikey := base.MakeInternalKey(keys[k], base.SeqNum(j), InternalKeyKindSet)
-			require.NoError(b, writers[j][0].Add(ikey, nil, false /* forceObsolete */))
+			require.NoError(b, writers[j][0].Add(ikey, nil, false /* forceObsolete */, base.KVMeta{}))
 		}
 	}
 	lastKey := makeKey(i)
 	keys = append(keys, lastKey)
 	for j := 0; j < len(files); j++ {
 		lastIKey := base.MakeInternalKey(lastKey, base.SeqNum(j), InternalKeyKindSet)
-		require.NoError(b, writers[j][1].Add(lastIKey, nil, false /* forceObsolete */))
+		require.NoError(b, writers[j][1].Add(lastIKey, nil, false /* forceObsolete */, base.KVMeta{}))
 	}
 	for _, levelWriters := range writers {
 		for j, w := range levelWriters {
diff --git a/range_del_test.go b/range_del_test.go
index d3cfa424..2da55410 100644
--- a/range_del_test.go
+++ b/range_del_test.go
@@ -626,7 +626,7 @@ func benchmarkRangeDelIterate(b *testing.B, entries, deleted int, snapshotCompac
 	})
 	for i := 0; i < entries; i++ {
 		key := base.MakeInternalKey(makeKey(i), 0, InternalKeyKindSet)
-		if err := w.Add(key, nil, false /* forceObsolete */); err != nil {
+		if err := w.Add(key, nil, false /* forceObsolete */, base.KVMeta{}); err != nil {
 			b.Fatal(err)
 		}
 	}
diff --git a/scan_internal_test.go b/scan_internal_test.go
index 0484f705..12b0369d 100644
--- a/scan_internal_test.go
+++ b/scan_internal_test.go
@@ -31,6 +31,22 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func iterateWithMeta(
+	iter base.InternalIterator, fn func(kv *base.InternalKV, meta base.KVMeta) bool,
+) {
+	if metaIter, ok := iter.(base.InternalIteratorWithKVMeta); ok {
+		kv, meta := metaIter.FirstWithMeta()
+		for kv != nil && fn(kv, meta) {
+			kv, meta = metaIter.NextWithMeta()
+		}
+	} else {
+		kv := iter.First()
+		for kv != nil && fn(kv, base.KVMeta{}) {
+			kv = iter.Next()
+		}
+	}
+}
+
 func TestScanStatistics(t *testing.T) {
 	var d *DB
 	type scanInternalReader interface {
@@ -428,14 +444,15 @@ func TestScanInternal(t *testing.T) {
 					require.NoError(t, err)
 				}
 				rangeKeys.Close()
-				for kv := points.First(); kv != nil; kv = points.Next() {
+				iterateWithMeta(points, func(kv *base.InternalKV, meta base.KVMeta) bool {
 					t.Logf("writing %s", kv.K)
 					var value []byte
 					var err error
 					value, _, err = kv.Value(value)
 					require.NoError(t, err)
-					require.NoError(t, w.Raw().Add(kv.K, value, false))
-				}
+					require.NoError(t, w.Raw().Add(kv.K, value, false /* forceObsolete */, meta))
+					return true
+				})
 				points.Close()
 				require.NoError(t, w.Close())
 			}
diff --git a/sstable/block/testdata/flush_governor b/sstable/block/testdata/flush_governor
index efef1945..ff748f29 100644
--- a/sstable/block/testdata/flush_governor
+++ b/sstable/block/testdata/flush_governor
@@ -25,7 +25,7 @@ should-flush size-before=899 size-after=10000
 ----
 should not flush
 
-# Size classes. Note that the block allocation overhead is 360.
+# Size classes. Note that the block allocation overhead is 312.
 init target-block-size=800 size-class-aware-threshold=60 size-classes=(820, 1020, 1320, 1820)
 ----
 low watermark: 480
diff --git a/sstable/colblk/data_block_meta_test.go b/sstable/colblk/data_block_meta_test.go
new file mode 100644
index 00000000..116b0b8c
--- /dev/null
+++ b/sstable/colblk/data_block_meta_test.go
@@ -0,0 +1,73 @@
+// Copyright 2025 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"
+
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/sstable/block"
+	"github.com/cockroachdb/pebble/sstable/blockiter"
+	"github.com/stretchr/testify/require"
+)
+
+// TestDataBlockIterWithMeta tests that the *WithMeta methods correctly
+// extract and return KVMeta for keys with tiering metadata.
+func TestDataBlockIterWithMeta(t *testing.T) {
+	schema := DefaultKeySchema(testkeys.Comparer, 16)
+	comparer := testkeys.Comparer.EnsureDefaults()
+	tieringColConfig := WithTieringColumns()
+
+	var w DataBlockEncoder
+	w.Init(&schema, tieringColConfig)
+	testKVs := []struct {
+		key  string
+		meta base.KVMeta
+	}{
+		{"a#1,SET", base.KVMeta{TieringSpanID: 42, TieringAttribute: 100}},
+		{"b#2,SET", base.KVMeta{TieringSpanID: 43, TieringAttribute: 200}},
+		{"c#3,SET", base.KVMeta{TieringSpanID: 0, TieringAttribute: 0}},
+	}
+	for _, kv := range testKVs {
+		ikey := base.ParseInternalKey(kv.key)
+		kcmp := w.KeyWriter.ComparePrev(ikey.UserKey)
+		w.Add(ikey, []byte("value"), block.InPlaceValuePrefix(kcmp.PrefixEqual()), kcmp, false, kv.meta)
+	}
+	block, _ := w.Finish(w.Rows(), w.Size())
+
+	var d DataBlockDecoder
+	bd := d.Init(&schema, block)
+	var it DataBlockIter
+	it.InitOnce(&schema, comparer, nil, tieringColConfig)
+	require.NoError(t, it.Init(&d, bd, blockiter.Transforms{}, tieringColConfig))
+
+	// Test FirstWithMeta and NextWithMeta.
+	for i, expected := range testKVs {
+		var kv *base.InternalKV
+		var meta base.KVMeta
+		if i == 0 {
+			kv, meta = it.FirstWithMeta()
+		} else {
+			kv, meta = it.NextWithMeta()
+		}
+		require.NotNil(t, kv)
+		require.Equal(t, expected.meta, meta)
+	}
+
+	// Test exhaustion.
+	kv, meta := it.NextWithMeta()
+	require.Nil(t, kv)
+	require.Equal(t, base.KVMeta{}, meta)
+
+	// Test SeekGEWithMeta.
+	kv, meta = it.SeekGEWithMeta([]byte("b"), base.SeekGEFlagsNone)
+	require.NotNil(t, kv)
+	require.Equal(t, testKVs[1].meta, meta)
+
+	kv, meta = it.SeekGEWithMeta([]byte("z"), base.SeekGEFlagsNone)
+	require.Nil(t, kv)
+	require.Equal(t, base.KVMeta{}, meta)
+}
diff --git a/sstable/colblk/data_block_test.go b/sstable/colblk/data_block_test.go
index 1f840515..e1bc394b 100644
--- a/sstable/colblk/data_block_test.go
+++ b/sstable/colblk/data_block_test.go
@@ -33,12 +33,13 @@ func TestDataBlock(t *testing.T) {
 	var bd BlockDecoder
 	var v DataBlockValidator
 	var it DataBlockIter
-	rw := NewDataBlockRewriter(&testKeysSchema, testkeys.Comparer.EnsureDefaults())
+	tieringConfig := NoTieringColumns()
+	rw := NewDataBlockRewriter(&testKeysSchema, testkeys.Comparer.EnsureDefaults(), tieringConfig)
 	var sizes []int
 	it.InitOnce(&testKeysSchema, testkeys.Comparer,
 		getInternalValuer(func([]byte) base.InternalValue {
 			return base.MakeInPlaceValue([]byte("mock external value"))
-		}))
+		}), tieringConfig)
 
 	datadriven.Walk(t, "testdata/data_block", func(t *testing.T, path string) {
 		datadriven.RunTest(t, path, func(t *testing.T, td *datadriven.TestData) string {
@@ -48,9 +49,9 @@ func TestDataBlock(t *testing.T) {
 				var bundleSize int
 				if td.MaybeScanArgs(t, "bundle-size", &bundleSize) {
 					s := DefaultKeySchema(testkeys.Comparer, bundleSize)
-					w.Init(&s)
+					w.Init(&s, NoTieringColumns())
 				} else {
-					w.Init(&testKeysSchema)
+					w.Init(&testKeysSchema, NoTieringColumns())
 				}
 				fmt.Fprint(&buf, &w)
 				sizes = sizes[:0]
@@ -59,7 +60,7 @@ func TestDataBlock(t *testing.T) {
 				// write-block does init/write/finish in a single command, and doesn't
 				// print anything.
 				if td.Cmd == "write-block" {
-					w.Init(&testKeysSchema)
+					w.Init(&testKeysSchema, NoTieringColumns())
 				}
 				var prevKey base.InternalKey
 				for line := range crstrings.LinesSeq(td.Input) {
@@ -80,7 +81,7 @@ func TestDataBlock(t *testing.T) {
 						isObsolete = true
 					}
 					v := []byte(line[j+1:])
-					w.Add(ik, v, vp, kcmp, isObsolete)
+					w.Add(ik, v, vp, kcmp, isObsolete, base.KVMeta{})
 					prevKey = ik
 					sizes = append(sizes, w.Size())
 				}
@@ -132,7 +133,7 @@ func TestDataBlock(t *testing.T) {
 					HideObsoletePoints:       td.HasArg("hide-obsolete-points"),
 					SyntheticPrefixAndSuffix: blockiter.MakeSyntheticPrefixAndSuffix([]byte(syntheticPrefix), []byte(syntheticSuffix)),
 				}
-				if err := it.Init(&r, bd, transforms); err != nil {
+				if err := it.Init(&r, bd, transforms, tieringConfig); err != nil {
 					return err.Error()
 				}
 
@@ -168,7 +169,7 @@ func benchmarkDataBlockWriter(b *testing.B, prefixSize, valueSize int) {
 	keys, values := makeTestKeyRandomKVs(rng, prefixSize, valueSize, targetBlockSize)
 
 	var w DataBlockEncoder
-	w.Init(&testKeysSchema)
+	w.Init(&testKeysSchema, NoTieringColumns())
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
@@ -178,7 +179,7 @@ func benchmarkDataBlockWriter(b *testing.B, prefixSize, valueSize int) {
 			ik := base.MakeInternalKey(keys[j], base.SeqNum(rng.Uint64N(uint64(base.SeqNumMax))), base.InternalKeyKindSet)
 			kcmp := w.KeyWriter.ComparePrev(ik.UserKey)
 			vp := block.InPlaceValuePrefix(kcmp.PrefixEqual())
-			w.Add(ik, values[j], vp, kcmp, false /* isObsolete */)
+			w.Add(ik, values[j], vp, kcmp, false /* isObsolete */, base.KVMeta{})
 			j++
 		}
 		w.Finish(w.Rows(), w.Size())
@@ -227,12 +228,12 @@ func BenchmarkDataBlockDecoderInit(b *testing.B) {
 	keys, values := makeTestKeyRandomKVs(rng, 8, 8, targetBlockSize)
 
 	var w DataBlockEncoder
-	w.Init(&testKeysSchema)
+	w.Init(&testKeysSchema, NoTieringColumns())
 	for j := 0; w.Size() < targetBlockSize; j++ {
 		ik := base.MakeInternalKey(keys[j], base.SeqNum(rng.Uint64N(uint64(base.SeqNumMax))), base.InternalKeyKindSet)
 		kcmp := w.KeyWriter.ComparePrev(ik.UserKey)
 		vp := block.InPlaceValuePrefix(kcmp.PrefixEqual())
-		w.Add(ik, values[j], vp, kcmp, false /* isObsolete */)
+		w.Add(ik, values[j], vp, kcmp, false /* isObsolete */, base.KVMeta{})
 	}
 	finished, _ := w.Finish(w.Rows(), w.Size())
 
diff --git a/sstable/format_test.go b/sstable/format_test.go
index 6f301236..e24ea65a 100644
--- a/sstable/format_test.go
+++ b/sstable/format_test.go
@@ -77,6 +77,12 @@ func TestTableFormat_RoundTrip(t *testing.T) {
 			version: 7,
 			want:    TableFormatPebblev7,
 		},
+		{
+			name:    "PebbleDBv8",
+			magic:   pebbleDBMagic,
+			version: 8,
+			want:    TableFormatPebblev8,
+		},
 		// Invalid cases.
 		{
 			name:    "Invalid RocksDB version",
@@ -87,8 +93,8 @@ func TestTableFormat_RoundTrip(t *testing.T) {
 		{
 			name:    "Invalid PebbleDB version",
 			magic:   pebbleDBMagic,
-			version: 8,
-			wantErr: "pebble/table: invalid table 000001: (unsupported pebble format version 8)",
+			version: 9,
+			wantErr: "pebble/table: invalid table 000001: (unsupported pebble format version 9)",
 		},
 		{
 			name:    "Unknown magic string",
diff --git a/sstable/random_test.go b/sstable/random_test.go
index f8e8601c..d08e97e6 100644
--- a/sstable/random_test.go
+++ b/sstable/random_test.go
@@ -417,7 +417,7 @@ func buildRandomSSTable(f vfs.File, cfg randomTableConfig) (*WriterMetadata, err
 				value[j] = byte(cfg.rng.Uint32())
 			}
 		}
-		if err := w.Add(keys[i], value, false /* forceObsolete */); err != nil {
+		if err := w.Add(keys[i], value, false /* forceObsolete */, base.KVMeta{}); err != nil {
 			return nil, err
 		}
 	}
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index 0f417ca2..68a449cd 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -1849,7 +1849,7 @@ func buildTestTableWithProvider(
 		value := make([]byte, i%100)
 		key = binary.BigEndian.AppendUint64(key, i)
 		ikey.UserKey = key
-		require.NoError(t, w.Add(ikey, value, false /* forceObsolete */))
+		require.NoError(t, w.Add(ikey, value, false /* forceObsolete */, base.KVMeta{}))
 	}
 
 	require.NoError(t, w.Close())
@@ -1887,7 +1887,7 @@ func buildBenchmarkTable(
 		binary.BigEndian.PutUint64(key, i+uint64(offset))
 		keys = append(keys, key)
 		ikey.UserKey = key
-		require.NoError(b, w.Add(ikey, nil, false /* forceObsolete */))
+		require.NoError(b, w.Add(ikey, nil, false /* forceObsolete */, base.KVMeta{}))
 	}
 
 	if err := w.Close(); err != nil {
@@ -2487,7 +2487,7 @@ func BenchmarkIteratorScanObsolete(b *testing.B) {
 				forceObsolete = false
 			}
 			require.NoError(b, w.Add(
-				base.MakeInternalKey(key, 0, InternalKeyKindSet), val, forceObsolete))
+				base.MakeInternalKey(key, 0, InternalKeyKindSet), val, forceObsolete, base.KVMeta{}))
 		}
 		require.NoError(b, w.Close())
 		// Re-open the Filename for reading.
diff --git a/sstable/suffix_rewriter_test.go b/sstable/suffix_rewriter_test.go
index 3adafe00..df698c10 100644
--- a/sstable/suffix_rewriter_test.go
+++ b/sstable/suffix_rewriter_test.go
@@ -198,7 +198,7 @@ func makeTestkeySSTable(
 		n := testkeys.WriteKey(keyBuf[len(sharedPrefix):], alphabet, uint64(i))
 		key := append(keyBuf[:len(sharedPrefix)+n], suffix...)
 		err := w.Raw().Add(
-			base.MakeInternalKey(key, 0, InternalKeyKindSet), key, false)
+			base.MakeInternalKey(key, 0, InternalKeyKindSet), key, false, base.KVMeta{})
 		if err != nil {
 			t.Fatal(err)
 		}
diff --git a/sstable/table_test.go b/sstable/table_test.go
index a40a8b00..d5e9d7f8 100644
--- a/sstable/table_test.go
+++ b/sstable/table_test.go
@@ -456,7 +456,8 @@ func TestFinalBlockIsWritten(t *testing.T) {
 						IndexBlockSize: indexBlockSize,
 					})
 					for _, k := range keys[:nk] {
-						if err := w.Add(InternalKey{UserKey: []byte(k)}, xxx[:vLen], false /* forceObsolete */); err != nil {
+						if err := w.Add(InternalKey{UserKey: []byte(k)}, xxx[:vLen],
+							false /* forceObsolete */, base.KVMeta{}); err != nil {
 							t.Errorf("nk=%d, vLen=%d: set: %v", nk, vLen, err)
 							continue loop
 						}
diff --git a/sstable/test_utils.go b/sstable/test_utils.go
index 79cc3f91..05374601 100644
--- a/sstable/test_utils.go
+++ b/sstable/test_utils.go
@@ -213,9 +213,9 @@ func ParseTestSST(w RawWriter, input string, bv *blobtest.Values) error {
 		case kv.IsKeySpan():
 			err = w.EncodeSpan(*kv.Span)
 		case kv.HasBlobValue():
-			err = w.AddWithBlobHandle(kv.Key, kv.BlobHandle, kv.Attr, kv.ForceObsolete)
+			err = w.AddWithBlobHandle(kv.Key, kv.BlobHandle, kv.Attr, kv.ForceObsolete, base.KVMeta{})
 		default:
-			err = w.Add(kv.Key, kv.Value, kv.ForceObsolete)
+			err = w.Add(kv.Key, kv.Value, kv.ForceObsolete, base.KVMeta{})
 		}
 		if err != nil {
 			return errors.Wrapf(err, "failed to write %s", kv)
diff --git a/sstable/testdata/treesteps_single_level_iter b/sstable/testdata/treesteps_single_level_iter
index 554cc05c..4047dfc2 100644
--- a/sstable/testdata/treesteps_single_level_iter
+++ b/sstable/testdata/treesteps_single_level_iter
@@ -17,7 +17,7 @@ seek-ge b@6
 next
 next
 ----
-https://raduberinde.github.io/treesteps/decode.html#eNrUVMuK2zAU3fcrLuomAZPGLU3AEAiloQ2UbNJdHYIcXyciqmSsSwmUgTCLWc_Cy_k6f8kgeQjOY4InMHloeXyPzkOS_7MR_4ssYISGYk78E2WIhjA1UyPUXOJU4j-UU0GYBf5X5rExYWpJhgV_mFCCBJcWRlz-GIBWYAzxSGKr5P-y9CFhxklnYIhnhDHzmHHzP1GmmNVnKR3jsdEAhIpxBZHUsyVIzeOaNDv6nRP_ZokNvwk6SQxSrw0S1ZwWPf9L542uE6GEWTj9euVUCCNcUf1W6kyfZ_NJeT_c3ThCCBUAKE2QaiNIaIWxgxIhMYC2W6GyfR8RheLhEcpmGyGL-p2QedBuvufeJV69BJWv1yBe5Pf78hbcNgBFnhf5usjXMNMyksvW0D4b68DZdWt3n2tw_fS6azvvFqcA4kMZN-_7IjmhB7OPvjce_H4xa31uI2eJuyV7YtZbjWL_ZY3mTZq2nff9z_u174LnC1FRPq38G7PujqB74AS6l0qxEWaTuw_PAQAA__-Hxnt3
+https://raduberinde.github.io/treesteps/decode.html#eNrUVMuK2zAU3fcrLuomAZPGpWnAEAiloQ2UbNJdHYIcXyciqmSsSwmUgTCLWc_Cy_k6f8kgeQjOY4InMHloeXyPzkOS_7MR_4ssYISGYk78E2WIhjA1UyPUXOJU4j-UU0GYBX6HeWxMmFqSYcEfJpQgwaWFEZc_BqAVGEM8ktgq-b8sfUiYcdIZGOIZYcw8Ztz8T5QpZvVZSsd4bDQAoWJcQST1bAlS87gmzY5-58S_WWLDb4JOEoPUa4NENadFz__SeaPrRChhFk6_XjkVwghXVL-VOtPn2XxS3g93N44QQgUAShOk2ggSWmHsoERIDKDtVqhs30dEoXh4hLLZRsii_teQedBuvufeJV69BJWv1yBe5Pf78hbcNgBFnhf5usjXMNMyksvW0D4b68DZdWt3n2tw_fS6azvvFqcA4kMZN-_7IjmhB7OPvjce_H4xa31uI2eJuyV7YtZbjWL_ZY3mTZq2nff9z_u174LnC1FRPq38G7PujqB74AS6l0qxEWaTuw_PAQAA__-Hx3t3
 
 iter-treesteps
 seek-prefix-ge a
@@ -25,4 +25,4 @@ next
 next-prefix
 next-prefix
 ----
-https://raduberinde.github.io/treesteps/decode.html#eNrcVMFq3DAQvfcrBvWyC0tqp2wpBkMoXdJACYX0VodFtseJsCoZayiGUgg99NyDj_06f0mRnAbX3i42BW82uhiGefPe0xvrK7vkn5EFjNBQyom_oBLREBZma4S6kbiV-AXlVhCWwekpW7ErwsKCDAs-MaEECS5tGTH_UGImqvMNaAXGEI8lnrRT3tshF4QlJ12CIV4SpmzFDGJ-vnmHssByPErpFPe1BiBUihXEUic5SM3TkTDb-pYTf2OBC38JOssMUuiBRHVDt6H_8tVE1ZlQwtw6_ilX1IFdYkXj72ZMd294q2kaxTjM_ETX7YK65dwDiBQAKE1QaCNIaIWpK2VCYgCeO5GyUe8hhebHT-iGuogYj9gK7j_eciaatqW7lX9avEeloqm_D3XY4t9KoKnrpr5r6jtItIxlfnJhf2crxel2pz_nUcn_9W_5tt8dTgGku8w-PECHNQwh8DPfe-6vrjYf74VbzYPiLO77zP_v_ujt2ddxsTxi6S6C9Y4E1odysZ5qol0vt1xR5XkRWz4ZIxBCfPZ6kE6_NpupDvFkU_GOdI7bCISQDLJJDmPogZZdf3v2OwAA__9XPWHp
+https://raduberinde.github.io/treesteps/decode.html#eNrcVMGK1EAQvfsVRXuZgWHNLAYkEFjEYV2QRVhvZhk6SWW3Sdsd0oUERFg8ePaQo1-XL5HurEtMxiFByMzYl0BRr957_Sr9hV3zT8gCRmgo5cRfUIloCAuzNULdSdxK_IxyKwjL4PycrdgNYWFBhgUfmVCCBJe2jJi_LzET1eUGtAJjiMcSz9op7-yQK8KSky7BEC8JU7ZiBjG_3LxFWWA5HqV0ivtaAxAqxQpiqZMcpObpSJhtfcOJv7bAxXoJOssMUuiBRHVH9-H6pT9RdSaUMPeOf8oVdWDXWNH4uxnT3RveappGMQ4zP9Ftu6BuOfcAIgUAShMU2ggSWmHqSpmQGIDnTqRs1HtIofn-A7qhLiLGI7aCx4-3nImmbelu5e8W76hUNPW3oQ5b_FMJNHXd1A9N_QCJlrHMz67s72ylON3u9Occlfyff5dv-93hFEC6y-zTA3RYwxACv1h7z9erm82HR-FW86A4i_s-87-7P3l79nVcLE9YuovA35GAfygX_lQT7Xq55Yoqz4vY8r8xAiHEF68G6fRrs5nqEE82Fe9I57SNQAjJIJvkMIaeaNnt12e_AgAA__9XPmHp
diff --git a/sstable/testdata/treesteps_two_level_iter b/sstable/testdata/treesteps_two_level_iter
index 3f6055cd..19a5707a 100644
--- a/sstable/testdata/treesteps_two_level_iter
+++ b/sstable/testdata/treesteps_two_level_iter
@@ -64,4 +64,4 @@ next
 next
 next
 ----
-https://raduberinde.github.io/treesteps/decode.html#eNrslkGL2kAUx-_9FMP0oiBbYzerBgQplW2hLAV7q4uMk-fu4HRGMo-uUApLDz334HE_nZ-kJEaJMbGxi4lmnYuQzMz7_3_zMn9_0Bv2DahDEQy6DNkb9AAMwtQM8UEPJXwHORQInnNl0RrtI0z9FYY6X6lQAgWT_mOAyWcPxmJ23SNaEWOQjSRc4IP-5O_wEcFjqD1ikHkILq1RpV1InecQ1FMSFCdCuTAjI6n5hEjN3Cyr_Xl94Fq5yxf-Fu_8HSpWlejx2AB2LtuNt0SCusP7TtOmNWoAJte9DyCn4EVNGKHuJGTykTB1KeY9QxYXUF9Vb1_uWX0slDD3QfnM5CNrbmCGGU8pPvVQLNrtFYxWI2PZfQwdt3mrdbV2b704941mY_0ZHtT97fL6Cq6utNkDRRbzX0RpJFNtBAqtwA0ezhfzx8X8cZesgSLh2FofjrGQ4JB6MAbK_-rTVJPF7z8k-nFXBpQNaI2EP_VqKaQ-hVK5liM5uQguan_jcEWywtI5ZOgQ3m3uay2x4vrNpr3V83m6nOiIb3tCdJZ6onG60lMPtZzhFQHv6R_wNiSzrpXOfZ0iebFfKXpt1fq9LwWxJ51NDWtoW-qO4iDiqp53EHHz-Z1KCUn7f5kq1TPFk6MYK1xqonnm3TbRA4ebnU-2JbSonV-HBne2nZZX9oGa1X5Gr9p7tmrMYW6Uy0nzTPCUCO7OohLRLHEQjbqtgoJo1G3lGkTRehutmviiQLQxTdmDaJNoTpTLSfNM8JQI7gqiUtEscRDxgmKI5xpCPDmC-EFalP93g_L9w4dvRQ_P99o8Xoa3P1_9DQAA__-Ta-Op
+https://raduberinde.github.io/treesteps/decode.html#eNrslkGL2kAUx-_9FMP0soJs4y7ZjQFBSmVbKEvB3uoi4-S5OzidkcyjK5TC0kPPPXjcT-cnKYlRYkxs7GLUqXMRkpl5__9vXubvd3rLvgL1KYLBgCF7gyGAQRibPj7qvoRvIPsCIfSvGrROuwjjaIWh_hcqlEDBZPQYYPQphKGY3HSIVsQYZAMJ5_ioP0Y7fEAIGeqQGGQhQkDrVOkACuf5BPWYxMWJUAFMyEBqPiJSs6DM6mheF7hWwfxFtMXbaIezRo3o4dAAtlzHuyAS1D0-tK5dWqcGYHTTeQ9yDGHahBHqXkIpHzlT52LeMWRZAc6ietPZsvpQKGEe4vKlyafW3MIES55SduquWDTdJYzLkmW3MXTY5hvNy4V7z_vv3F94V0v3Oz37u_n1FV9dRbN7isymP4nSSMbaCBRaQRA_nM6mT7Pp0yZZPUWSsbY-GUMhwSdOPHoq-uqLVJPZr98k_XGf9Sjr0TpJfpyaFVKfE6lcy4EcnccXdbRxsiJfoXUOGfqEt6-3tZZbcflm1d7i-bRYTnpktz0iOnM96Thd6HESLSd4-4D3_Bd4K5JZu1HMfZkiVbFfKHrdqHc7n_fEnrRWNSyhrak7iIPIqnrZQWTNV3cqFpKO_jKd1U4Uj45iprDVRKvMu3WiOw43t5psy2lRt7oOje9styiv3B01q_uCXnW3bNWMw8oo20nzRPCYCG7OIotoWhxEg7a3pyAatL1Kgyhdb6VVc1_sEW1GU_kgWiVaEWU7aZ4IHhPBTUFkFU2Lg4jvKYZ4pSHE8yOI76RF-T83KN8-fPha9PBqr83DZXj349WfAAAA___w2-Ot
diff --git a/sstable/testdata/virtual_reader_props b/sstable/testdata/virtual_reader_props
index f2042a1e..496bec06 100644
--- a/sstable/testdata/virtual_reader_props
+++ b/sstable/testdata/virtual_reader_props
@@ -263,7 +263,7 @@ seqnums:  [1-1]
 virtualize lower=a#1,DEL upper=a0#1,SET show-size
 ----
 bounds:  [a#1,DEL-a0#1,SET]
-size: 89 B / backing size: 840 B
+size: 99 B / backing size: 850 B
 
 # Test virtual sstable with a synthetic prefix.
 build
@@ -278,4 +278,4 @@ seqnums:  [1-1]
 virtualize lower=poi-b#1,SET upper=poi-c#1,SET prefix=poi- show-size
 ----
 bounds:  [poi-b#1,SET-poi-c#1,SET]
-size: 87 B / backing size: 719 B
+size: 100 B / backing size: 732 B
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 8af164fe..9022a41e 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -974,7 +974,7 @@ func TestWriterBlockPropertiesErrors(t *testing.T) {
 				}
 			}()
 
-			err = w.Add(k1, v1, false /* forceObsolete */)
+			err = w.Add(k1, v1, false /* forceObsolete */, base.KVMeta{})
 			switch tc {
 			case errSiteAdd:
 				require.Error(t, err)
@@ -983,18 +983,18 @@ func TestWriterBlockPropertiesErrors(t *testing.T) {
 			case errSiteFinishBlock:
 				require.NoError(t, err)
 				// Addition of a second key completes the first block.
-				err = w.Add(k2, v2, false /* forceObsolete */)
+				err = w.Add(k2, v2, false /* forceObsolete */, base.KVMeta{})
 				require.Error(t, err)
 				require.Equal(t, blockPropErr, err)
 				return
 			case errSiteFinishIndex:
 				require.NoError(t, err)
 				// Addition of a second key completes the first block.
-				err = w.Add(k2, v2, false /* forceObsolete */)
+				err = w.Add(k2, v2, false /* forceObsolete */, base.KVMeta{})
 				require.NoError(t, err)
 				// The index entry for the first block is added after the completion of
 				// the second block, which is triggered by adding a third key.
-				err = w.Add(k3, v3, false /* forceObsolete */)
+				err = w.Add(k3, v3, false /* forceObsolete */, base.KVMeta{})
 				require.Error(t, err)
 				require.Equal(t, blockPropErr, err)
 				return
@@ -1101,7 +1101,8 @@ func TestWriterRace(t *testing.T) {
 			w := newRowWriter(f, opts)
 			for ki := 0; ki < len(keys); ki++ {
 				require.NoError(t, w.Add(
-					base.MakeInternalKey(keys[ki], base.SeqNum(ki), InternalKeyKindSet), val, false /* forceObsolete */))
+					base.MakeInternalKey(keys[ki], base.SeqNum(ki), InternalKeyKindSet), val,
+					false /* forceObsolete */, base.KVMeta{}))
 				require.Equal(
 					t, w.dataBlockBuf.dataBlock.CurKey().UserKey, keys[ki],
 				)
diff --git a/testdata/blob_rewrite b/testdata/blob_rewrite
index d244f9c3..01eb7880 100644
--- a/testdata/blob_rewrite
+++ b/testdata/blob_rewrite
@@ -58,15 +58,15 @@ rewrite-blob 000002 000001 000003
 ----
 # create: 000004.blob
 # open: 000001.sst (options: *vfs.randomReadsOption)
-# read-at(737, 61): 000001.sst
-# read-at(660, 77): 000001.sst
-# read-at(209, 451): 000001.sst
-# read-at(184, 25): 000001.sst
+# read-at(747, 61): 000001.sst
+# read-at(670, 77): 000001.sst
+# read-at(219, 451): 000001.sst
+# read-at(194, 25): 000001.sst
 # open: 000003.sst (options: *vfs.randomReadsOption)
-# read-at(761, 61): 000003.sst
-# read-at(684, 77): 000003.sst
-# read-at(193, 491): 000003.sst
-# read-at(168, 25): 000003.sst
+# read-at(772, 61): 000003.sst
+# read-at(695, 77): 000003.sst
+# read-at(203, 492): 000003.sst
+# read-at(178, 25): 000003.sst
 # open: 000002.blob (options: *vfs.randomReadsOption)
 # read-at(35, 70): 000002.blob
 # read-at(37, 30): 000002.blob
diff --git a/valsep/value_separation_test.go b/valsep/value_separation_test.go
index b0ecd103..f6bdc78f 100644
--- a/valsep/value_separation_test.go
+++ b/valsep/value_separation_test.go
@@ -138,7 +138,7 @@ func TestValueSeparationPolicy(t *testing.T) {
 					} else {
 						ikv.V = base.MakeInPlaceValue([]byte(parts[1]))
 					}
-					require.NoError(t, vs.Add(tw, &ikv, false /* forceObsolete */, false /* isLikelyMVCCGarbage */))
+					require.NoError(t, vs.Add(tw, &ikv, false /* forceObsolete */, false /* isLikelyMVCCGarbage */, base.KVMeta{}))
 				}
 				return buf.String()
 			case "estimated-sizes":
@@ -187,5 +187,5 @@ func errShortAttrExtractor(
 	return 0, errors.New("short attribute extractor error")
 }
 
-// Assert that errShortAttrExtractor implements the ShortAttributeExtractor
+// Assert that errShortAttrExtractor implements the ShortAttributeExtractor.
 var _ base.ShortAttributeExtractor = errShortAttrExtractor
diff --git a/version_set_test.go b/version_set_test.go
index 7e25bb9e..000fc5dd 100644
--- a/version_set_test.go
+++ b/version_set_test.go
@@ -43,7 +43,7 @@ func writeAndIngest(t *testing.T, mem vfs.FS, d *DB, k InternalKey, v []byte, fi
 	f, err := mem.Create(path, vfs.WriteCategoryUnspecified)
 	require.NoError(t, err)
 	w := sstable.NewRawWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
-	require.NoError(t, w.Add(k, v, false /* forceObsolete */))
+	require.NoError(t, w.Add(k, v, false /* forceObsolete */, base.KVMeta{}))
 	require.NoError(t, w.Close())
 	require.NoError(t, d.Ingest(context.Background(), []string{path}))
 }