diff --git a/go.mod b/go.mod
index 26545722f7..e7016448fc 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/RaduBerinde/axisds v0.0.0-20250419182453-5135a0650657
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/cockroachdb/crlib v0.0.0-20251122031428-fe658a2dbda1
-	github.com/cockroachdb/datadriven v1.0.3-0.20250911232732-d959cf14706c
+	github.com/cockroachdb/datadriven v1.0.3-0.20251123150250-ddff6747b112
 	github.com/cockroachdb/errors v1.11.3
 	github.com/cockroachdb/metamorphic v0.0.0-20231108215700-4ba948b56895
 	github.com/cockroachdb/redact v1.1.5
diff --git a/go.sum b/go.sum
index 820510f0ff..5733965bcd 100644
--- a/go.sum
+++ b/go.sum
@@ -29,8 +29,8 @@ github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cockroachdb/crlib v0.0.0-20251122031428-fe658a2dbda1 h1:iX0YCYC5Jbt2/g7zNTP/QxhrV8Syp5kkzNiERKeN1uE=
 github.com/cockroachdb/crlib v0.0.0-20251122031428-fe658a2dbda1/go.mod h1:NjNuToN/FbhwH1cCyM9G4Rhtxx+ZaOgtoqFR+thng7w=
-github.com/cockroachdb/datadriven v1.0.3-0.20250911232732-d959cf14706c h1:a0m7gmtv2mzJQ4wP9BkxCmJAnjZ7fsvCi2IORGD1als=
-github.com/cockroachdb/datadriven v1.0.3-0.20250911232732-d959cf14706c/go.mod h1:jsaKMvD3RBCATk1/jbUZM8C9idWBJME9+VRZ5+Liq1g=
+github.com/cockroachdb/datadriven v1.0.3-0.20251123150250-ddff6747b112 h1:T1++5Vt0/4/IWZ1mHmUYl7fhQnz50QhNWIY+ITvLLIM=
+github.com/cockroachdb/datadriven v1.0.3-0.20251123150250-ddff6747b112/go.mod h1:jsaKMvD3RBCATk1/jbUZM8C9idWBJME9+VRZ5+Liq1g=
 github.com/cockroachdb/errors v1.11.3 h1:5bA+k2Y6r+oz/6Z/RFlNeVCesGARKuC6YymtcDrbC/I=
 github.com/cockroachdb/errors v1.11.3/go.mod h1:m4UIW4CDjx+R5cybPsNrRbreomiFqt8o1h1wUVazSd8=
 github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZeQy818SGhaone5OnYfxFR/+AzdY3sf5aE=
diff --git a/internal/base/filenames.go b/internal/base/filenames.go
index ee7c585fd3..daf470258f 100644
--- a/internal/base/filenames.go
+++ b/internal/base/filenames.go
@@ -128,6 +128,12 @@ const (
 	FileTypeOldTemp
 	FileTypeTemp
 	FileTypeBlob
+	// FileTypeBlobMeta is a file that contains only the metadata portion of a
+	// blob file (used when the blob file in on cold storage). The filename for
+	// blobmeta files is of the form `<file-num>.blobmeta.<offset>`, where
+	// <offset> indicates that the file mirrors the contents of the corresponding
+	// blob file starting at this offset.
+	FileTypeBlobMeta
 )
 
 var fileTypeStrings = [...]string{
@@ -139,6 +145,7 @@ var fileTypeStrings = [...]string{
 	FileTypeOldTemp:  "old-temp",
 	FileTypeTemp:     "temp",
 	FileTypeBlob:     "blob",
+	FileTypeBlobMeta: "blobmeta",
 }
 
 // FileTypeFromName parses a FileType from its string representation.
@@ -166,6 +173,8 @@ func (ft FileType) String() string {
 }
 
 // MakeFilename builds a filename from components.
+//
+// Note that for FileTypeBlobMeta, ".<offset>" must be appended to the filename.
 func MakeFilename(fileType FileType, dfn DiskFileNum) string {
 	// Make a buffer sufficiently large for most possible filenames, especially
 	// the common case of a numbered table or blob file.
@@ -192,6 +201,8 @@ func appendFilename(buf []byte, fileType FileType, dfn DiskFileNum) []byte {
 		buf = fmt.Appendf(buf, "temporary.%06d.dbtmp", uint64(dfn))
 	case FileTypeBlob:
 		buf = fmt.Appendf(buf, "%06d.blob", uint64(dfn))
+	case FileTypeBlobMeta:
+		buf = fmt.Appendf(buf, "%06d.blobmeta", uint64(dfn))
 	default:
 		panic("unreachable")
 	}
@@ -199,11 +210,16 @@ func appendFilename(buf []byte, fileType FileType, dfn DiskFileNum) []byte {
 }
 
 // MakeFilepath builds a filepath from components.
+//
+// Note that for FileTypeBlobMeta, ".<offset>" must be appended to the filepath.
 func MakeFilepath(fs vfs.FS, dirname string, fileType FileType, dfn DiskFileNum) string {
 	return fs.PathJoin(dirname, MakeFilename(fileType, dfn))
 }
 
 // ParseFilename parses the components from a filename.
+//
+// Note that the offset component of a FileTypeBlobMeta is not parsed by this
+// function.
 func ParseFilename(fs vfs.FS, filename string) (fileType FileType, dfn DiskFileNum, ok bool) {
 	filename = fs.PathBase(filename)
 	switch {
@@ -250,6 +266,9 @@ func ParseFilename(fs vfs.FS, filename string) (fileType FileType, dfn DiskFileN
 		case "blob":
 			return FileTypeBlob, dfn, true
 		}
+		if strings.HasPrefix(filename[i+1:], "blobmeta.") {
+			return FileTypeBlobMeta, dfn, true
+		}
 	}
 	return 0, dfn, false
 }
diff --git a/internal/base/filenames_test.go b/internal/base/filenames_test.go
index df3645c170..ac6451d1c4 100644
--- a/internal/base/filenames_test.go
+++ b/internal/base/filenames_test.go
@@ -7,6 +7,7 @@ package base
 import (
 	"bytes"
 	"fmt"
+	"math/rand/v2"
 	"os"
 	"testing"
 
@@ -49,6 +50,7 @@ func TestParseFilename(t *testing.T) {
 		"000000.blob":            true,
 		"000001.blob":            true,
 		"935203523.blob":         true,
+		"000001.blobmeta.0":      true,
 	}
 	fs := vfs.NewMem()
 	for tc, want := range testCases {
@@ -95,6 +97,17 @@ func TestFilenameRoundTrip(t *testing.T) {
 	}
 }
 
+func TestFilenameBlobMeta(t *testing.T) {
+	fileNum := DiskFileNum(rand.Uint64())
+	offset := rand.Int64()
+	fs := vfs.NewMem()
+	path := fmt.Sprintf("%s.%d", MakeFilepath(fs, "foo", FileTypeBlobMeta, fileNum), offset)
+	typ, fn, ok := ParseFilename(fs, path)
+	require.True(t, ok)
+	require.Equal(t, FileTypeBlobMeta, typ)
+	require.Equal(t, fileNum, fn)
+}
+
 type bufferFataler struct {
 	buf bytes.Buffer
 }
diff --git a/objstorage/objstorage.go b/objstorage/objstorage.go
index 338a338e81..4c473bbd73 100644
--- a/objstorage/objstorage.go
+++ b/objstorage/objstorage.go
@@ -107,8 +107,22 @@ type Writable interface {
 
 	// Finish completes the object and makes the data durable.
 	// No further calls are allowed after calling Finish.
+	//
+	// If Finish fails, it is expected that the caller will delete the created
+	// object. If the process crashes during Finish, it is expected that the file
+	// will be deleted on startup.
 	Finish() error
 
+	// StartMetadataPortion signals to the writer that the metadata part of the
+	// object starts here. If the object is being written to the cold tier, data
+	// in subsequent Write() calls will also be written to the hot tier.
+	//
+	// The function should be called at most one time.
+	//
+	// An error means that we won't be able to successfully finish this object.
+	// - Any constraints on when this can be called relative to Write()
+	StartMetadataPortion() error
+
 	// Abort gives up on finishing the object. There is no guarantee about whether
 	// the object exists after calling Abort.
 	// No further calls are allowed after calling Abort.
diff --git a/objstorage/objstorageprovider/cold_readable.go b/objstorage/objstorageprovider/cold_readable.go
new file mode 100644
index 0000000000..706f35ddf7
--- /dev/null
+++ b/objstorage/objstorageprovider/cold_readable.go
@@ -0,0 +1,129 @@
+// 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 objstorageprovider
+
+import (
+	"context"
+	"sync"
+
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/vfs"
+)
+
+// newColdReadable returns an objstorage.Readable that reads the main data from
+// the wrapped "cold storage" readable, and the metadata from a separate file in
+// a local filesystem. The separate file contains a suffix of the full file,
+// starting at metaStartOffset.
+func newColdReadable(
+	cold objstorage.Readable, metaFS vfs.FS, metaFilepath string, metaStartOffset int64,
+) *coldReadable {
+	r := &coldReadable{
+		cold: cold,
+	}
+	r.meta.fs = metaFS
+	r.meta.filepath = metaFilepath
+	r.meta.startOffset = metaStartOffset
+	return r
+}
+
+type coldReadable struct {
+	cold objstorage.Readable
+
+	meta struct {
+		fs          vfs.FS
+		filepath    string
+		startOffset int64
+		once        struct {
+			sync.Once
+			file vfs.File
+			err  error
+		}
+	}
+}
+
+var _ objstorage.Readable = (*coldReadable)(nil)
+
+// readMetaAt reads from the metadata file at the given offset.
+func (r *coldReadable) readMetaAt(p []byte, off int64) error {
+	r.meta.once.Do(func() {
+		r.meta.once.file, r.meta.once.err = r.meta.fs.Open(r.meta.filepath, vfs.RandomReadsOption)
+	})
+	if r.meta.once.err != nil {
+		return r.meta.once.err
+	}
+	_, err := r.meta.once.file.ReadAt(p, off)
+	return err
+}
+
+// ReadAt is part of the objstorage.Readable interface.
+func (r *coldReadable) ReadAt(ctx context.Context, p []byte, off int64) error {
+	// We don't expect reads that span both regions, but in that case it is
+	// correct to read it all from the cold file (which contains all the data).
+	if off < r.meta.startOffset {
+		return r.cold.ReadAt(ctx, p, off)
+	}
+	return r.readMetaAt(p, off-r.meta.startOffset)
+}
+
+// Close is part of the objstorage.Readable interface.
+func (r *coldReadable) Close() error {
+	err := r.cold.Close()
+	if r.meta.once.file != nil {
+		err = firstError(err, r.meta.once.file.Close())
+		r.meta.once.file = nil
+	}
+	return err
+}
+
+// Size is part of the objstorage.Readable interface.
+func (r *coldReadable) Size() int64 {
+	return r.cold.Size()
+}
+
+// NewReadHandle is part of the objstorage.Readable interface.
+func (r *coldReadable) NewReadHandle(
+	readBeforeSize objstorage.ReadBeforeSize,
+) objstorage.ReadHandle {
+	return &coldReadHandle{
+		r:    r,
+		cold: r.cold.NewReadHandle(readBeforeSize),
+	}
+}
+
+type coldReadHandle struct {
+	r    *coldReadable
+	cold objstorage.ReadHandle
+}
+
+var _ objstorage.ReadHandle = (*coldReadHandle)(nil)
+
+// ReadAt is part of the objstorage.ReadHandle interface.
+func (rh *coldReadHandle) ReadAt(ctx context.Context, p []byte, off int64) error {
+	if off < rh.r.meta.startOffset {
+		// Read from cold storage only.
+		return rh.cold.ReadAt(ctx, p, off)
+	}
+	// Read from metadata only.
+	return rh.r.readMetaAt(p, off-rh.r.meta.startOffset)
+}
+
+// Close is part of the objstorage.ReadHandle interface.
+func (rh *coldReadHandle) Close() error {
+	return rh.cold.Close()
+}
+
+// SetupForCompaction is part of the objstorage.ReadHandle interface.
+func (rh *coldReadHandle) SetupForCompaction() {
+	rh.cold.SetupForCompaction()
+}
+
+// RecordCacheHit is part of the objstorage.ReadHandle interface.
+func (rh *coldReadHandle) RecordCacheHit(ctx context.Context, offset, size int64) {
+	// We don't use prefetching for the metadata portion, so we only need to
+	// report cache hits to the cold readable.
+	if offset < rh.r.meta.startOffset {
+		rh.cold.RecordCacheHit(ctx, offset, min(size, rh.r.meta.startOffset-offset))
+	}
+}
diff --git a/objstorage/objstorageprovider/cold_writable.go b/objstorage/objstorageprovider/cold_writable.go
new file mode 100644
index 0000000000..f2d3835f42
--- /dev/null
+++ b/objstorage/objstorageprovider/cold_writable.go
@@ -0,0 +1,167 @@
+// 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 objstorageprovider
+
+import (
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/vfs"
+)
+
+// newColdWritable returns an objstorage.Writable that writes the main data to the
+// wrapped "cold storage" writable, and all the metadata - i.e. everything after
+// StartMetadataPortion() - to a separate file in a local filesystem. This
+// allows reading metadata from the hot tier.
+//
+// When StartMetadataPortion() is called, coldWritable creates a new file with
+// path produced by p.metaPath(). All subsequent writes are written to both
+// files until Finish() or Abort() is called.
+func newColdWritable(
+	p *provider,
+	fileType base.FileType,
+	fileNum base.DiskFileNum,
+	cold objstorage.Writable,
+	diskWriteCategory vfs.DiskWriteCategory,
+) *coldWritable {
+	w := &coldWritable{
+		p:                 p,
+		fileType:          fileType,
+		fileNum:           fileNum,
+		cold:              cold,
+		diskWriteCategory: diskWriteCategory,
+	}
+	return w
+}
+
+type coldWritable struct {
+	p                 *provider
+	fileType          base.FileType
+	fileNum           base.DiskFileNum
+	cold              objstorage.Writable
+	diskWriteCategory vfs.DiskWriteCategory
+
+	meta struct {
+		started     bool
+		startOffset int64
+		file        vfs.File
+		buf         []byte
+	}
+
+	err error
+}
+
+// metaBufSize is the same as the iobuf.Writer default size.
+const metaBufSize = 4096
+
+var _ objstorage.Writable = (*coldWritable)(nil)
+
+// Write is part of the objstorage.Writable interface.
+func (w *coldWritable) Write(p []byte) error {
+	if w.err != nil {
+		return w.err
+	}
+	if w.meta.started {
+		if w.meta.buf == nil {
+			w.meta.buf = make([]byte, 0, metaBufSize)
+		}
+		// Because meta.file.Write() can mangle the buffer contents, we always have
+		// to copy first.
+		//
+		// This code is similar to iobuf.Writer, except that we always issue the
+		// write from a copy (since our writers can mangle the data).
+		for ofs := 0; ofs < len(p); {
+			n := copy(w.meta.buf[len(w.meta.buf):cap(w.meta.buf)], p[ofs:])
+			w.meta.buf = w.meta.buf[:len(w.meta.buf)+n]
+			ofs += n
+			if len(w.meta.buf) == cap(w.meta.buf) {
+				if w.err = w.flushMeta(); w.err != nil {
+					return w.err
+				}
+			}
+		}
+	} else {
+		w.meta.startOffset += int64(len(p))
+	}
+	w.err = w.cold.Write(p)
+	return w.err
+}
+
+func (w *coldWritable) StartMetadataPortion() error {
+	if w.err != nil {
+		return w.err
+	}
+	if w.meta.started {
+		return nil
+	}
+	w.meta.file, w.err = w.p.st.Local.FS.Create(w.metaPath(), w.diskWriteCategory)
+	if w.err != nil {
+		return w.err
+	}
+	w.meta.started = true
+	return w.cold.StartMetadataPortion()
+}
+
+func (w *coldWritable) flushMeta() error {
+	if _, err := w.meta.file.Write(w.meta.buf); err != nil {
+		return err
+	}
+	w.meta.buf = w.meta.buf[:0]
+	return nil
+}
+
+// Finish is part of the objstorage.Writable interface.
+//
+// Finish guarantees that on failure or crash, either the metadata file doesn't
+// exist or it will be cleaned up when the object is deleted.
+func (w *coldWritable) Finish() error {
+	if w.err != nil {
+		w.Abort()
+		return w.err
+	}
+	if w.meta.started {
+		if len(w.meta.buf) > 0 {
+			if err := w.flushMeta(); err != nil {
+				w.Abort()
+				return err
+			}
+		}
+		err := firstError(w.meta.file.Sync(), w.meta.file.Close())
+		w.meta.file = nil
+		if err != nil {
+			w.Abort()
+			return err
+		}
+	}
+	// We finish the cold tier write first, then register the metadata file. If we
+	// crash after cold.Finish() but before addColdObjectMetaFile(), the metadata
+	// file will be discovered during startup.
+	if err := w.cold.Finish(); err != nil {
+		w.deleteMetaFile()
+		return err
+	}
+	if w.meta.started {
+		w.p.addColdObjectMetaFile(w.fileType, w.fileNum, w.meta.startOffset)
+	}
+	return nil
+}
+
+func (w *coldWritable) Abort() {
+	if w.meta.started {
+		if w.meta.file != nil {
+			_ = w.meta.file.Close()
+			w.meta.file = nil
+		}
+		w.deleteMetaFile()
+	}
+	w.cold.Abort()
+}
+
+func (w *coldWritable) deleteMetaFile() {
+	_ = w.p.st.Local.FS.Remove(w.metaPath())
+}
+
+func (w *coldWritable) metaPath() string {
+	return w.p.metaPath(w.fileType, w.fileNum, w.meta.startOffset)
+}
diff --git a/objstorage/objstorageprovider/objiotracing/obj_io_tracing_on.go b/objstorage/objstorageprovider/objiotracing/obj_io_tracing_on.go
index 15bc0b6a32..5d3bc50fc3 100644
--- a/objstorage/objstorageprovider/objiotracing/obj_io_tracing_on.go
+++ b/objstorage/objstorageprovider/objiotracing/obj_io_tracing_on.go
@@ -110,6 +110,11 @@ func (w *writable) Write(p []byte) error {
 	return w.w.Write(p)
 }
 
+// Finish is part of the objstorage.Writable interface.
+func (w *writable) StartMetadataPortion() error {
+	return w.w.StartMetadataPortion()
+}
+
 // Finish is part of the objstorage.Writable interface.
 func (w *writable) Finish() error {
 	w.g.flush()
diff --git a/objstorage/objstorageprovider/provider_test.go b/objstorage/objstorageprovider/provider_test.go
index 16dedc41b9..a036924f2e 100644
--- a/objstorage/objstorageprovider/provider_test.go
+++ b/objstorage/objstorageprovider/provider_test.go
@@ -8,7 +8,6 @@ import (
 	"context"
 	"fmt"
 	"math/rand/v2"
-	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -24,6 +23,16 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// TestProvider datadriven format:
+//
+//	open dir=<dir> [cold-dir=<dir>] [creator-id=<id>]
+//	create file-num=<num> [file-type=<type>] [shared] [cold-tier] salt=<salt> size=<size> [no-ref-tracking]
+//	read file-num=<num> [file-type=<type>] [for-compaction] [readahead=<mode>]
+//	remove file-num=<num> [file-type=<type>]
+//	link-or-copy file-num=<num> [file-type=<type>] [shared] salt=<salt> size=<size> [no-ref-tracking]
+//	save-backing key=<key> file-num=<num>
+//	close-backing key=<key>
+//	switch dir=<dir>
 func TestProvider(t *testing.T) {
 	datadriven.Walk(t, "testdata/provider", func(t *testing.T, path string) {
 		var log base.InMemLogger
@@ -46,40 +55,40 @@ func TestProvider(t *testing.T) {
 		var curProvider objstorage.Provider
 		readaheadConfig := NewReadaheadConfig()
 		datadriven.RunTest(t, path, func(t *testing.T, d *datadriven.TestData) string {
+			// Arguments that are common to multiple commands.
+			var fsDir, coldDir string
+			var fileType base.FileType
+			var fileNum base.DiskFileNum
+			var salt, size int
+			var noRefTracking, shared, coldTier bool
+
+			d.MaybeScanArgs(t, "dir", &fsDir)
+			d.MaybeScanArgs(t, "cold-dir", &coldDir)
+			fileType = func() base.FileType {
+				var fileTypeStr string
+				if !d.MaybeScanArgs(t, "file-type", &fileTypeStr) {
+					return base.FileTypeTable
+				}
+				return base.FileTypeFromName(fileTypeStr)
+			}()
+			d.MaybeScanArgs(t, "file-num", &fileNum)
+			d.MaybeScanArgs(t, "salt", &salt)
+			d.MaybeScanArgs(t, "size", &size)
+			noRefTracking = d.HasArg("no-ref-tracking")
+			shared = d.HasArg("shared")
+			coldTier = d.HasArg("cold-tier")
+
 			readaheadConfig.Set(defaultReadaheadInformed, defaultReadaheadSpeculative)
-			scanArgs := func(desc string, args ...interface{}) {
-				t.Helper()
-				if len(d.CmdArgs) != len(args) {
-					d.Fatalf(t, "usage: %s %s", d.Cmd, desc)
-				}
-				for i := range args {
-					_, err := fmt.Sscan(d.CmdArgs[i].String(), args[i])
-					if err != nil {
-						d.Fatalf(t, "%s: error parsing argument '%s'", d.Cmd, d.CmdArgs[i])
-					}
-				}
-			}
-			ctx := context.Background()
 
+			ctx := context.Background()
 			log.Reset()
 			switch d.Cmd {
 			case "open":
-				var fsDir, coldDir string
+				if fsDir == "" {
+					d.Fatalf(t, "usage: open dir=<dir> [cold-dir=<dir>] [creator-id=<id>]")
+				}
 				var creatorID objstorage.CreatorID
-				d.CmdArgs = slices.DeleteFunc(d.CmdArgs, func(arg datadriven.CmdArg) bool {
-					switch arg.Key {
-					case "creator-id":
-						var id uint64
-						arg.Scan(t, 0, &id)
-						creatorID = objstorage.CreatorID(id)
-						return true
-					case "cold-tier":
-						coldDir = arg.SingleVal(t)
-						return true
-					}
-					return false
-				})
-				scanArgs("<fs-dir> [creator-id=X]", &fsDir)
+				d.MaybeScanArgs(t, "creator-id", &creatorID)
 
 				require.NoError(t, fs.MkdirAll(fsDir, 0755))
 				st := DefaultSettings(fs, fsDir)
@@ -111,8 +120,9 @@ func TestProvider(t *testing.T) {
 				return log.String()
 
 			case "switch":
-				var fsDir string
-				scanArgs("<fs-dir>", &fsDir)
+				if fsDir == "" {
+					d.Fatalf(t, "usage: switch dir=<dir>")
+				}
 				curProvider = providers[fsDir]
 				if curProvider == nil {
 					t.Fatalf("unknown provider %s", fsDir)
@@ -129,71 +139,56 @@ func TestProvider(t *testing.T) {
 				return log.String()
 
 			case "create":
+				if fileNum == 0 || size == 0 || salt == 0 {
+					d.Fatalf(t, "usage: create file-num=<num> [file-type=sstable|blob] [shared] [cold-tier] salt=<salt> size=<size> [no-ref-tracking] [meta-offset=<offset>]")
+				}
+				metaOffset := -1
+				d.MaybeScanArgs(t, "meta-offset", &metaOffset)
 				opts := objstorage.CreateOptions{
 					SharedCleanupMethod: objstorage.SharedRefTracking,
 				}
-				ft := base.FileTypeTable
-				d.CmdArgs = slices.DeleteFunc(d.CmdArgs, func(arg datadriven.CmdArg) bool {
-					switch arg.Key {
-					case "file-type":
-						ft = base.FileTypeFromName(d.CmdArgs[0].FirstVal(t))
-						return true
-					case "no-ref-tracking":
-						opts.SharedCleanupMethod = objstorage.SharedNoCleanup
-						return true
-					case "cold-tier":
-						opts.Tier = base.ColdTier
-						return true
-					}
-					return false
-				})
-				var fileNum base.DiskFileNum
-				var typ string
-				var salt, size int
-				scanArgs("[file-type=sstable|blob] <file-num> <local|shared> <salt> <size> [no-ref-tracking] [cold-tier]", &fileNum, &typ, &salt, &size)
-				switch typ {
-				case "local":
-				case "shared":
-					opts.PreferSharedStorage = true
-				default:
-					d.Fatalf(t, "'%s' should be 'local' or 'shared'", typ)
-				}
-				w, _, err := curProvider.Create(ctx, ft, fileNum, opts)
+				opts.PreferSharedStorage = shared
+				if coldTier {
+					opts.Tier = base.ColdTier
+				}
+				w, _, err := curProvider.Create(ctx, fileType, fileNum, opts)
 				if err != nil {
 					return err.Error()
 				}
 				data := make([]byte, size)
-				// TODO(radu): write in chunks?
 				genData(byte(salt), 0, data)
-				require.NoError(t, w.Write(data))
+				if metaOffset >= 0 {
+					if metaOffset > size {
+						d.Fatalf(t, "meta-offset (%d) must be <= size (%d)", metaOffset, size)
+					}
+					// Write data before metadata.
+					if metaOffset > 0 {
+						require.NoError(t, w.Write(data[:metaOffset]))
+					}
+					// Start metadata portion.
+					require.NoError(t, w.StartMetadataPortion())
+					// Write metadata.
+					if metaOffset < size {
+						require.NoError(t, w.Write(data[metaOffset:]))
+					}
+				} else {
+					require.NoError(t, w.Write(data))
+				}
 				require.NoError(t, w.Finish())
 
 				return log.String()
 
 			case "link-or-copy":
+				if fileNum == 0 || size == 0 || salt == 0 {
+					d.Fatalf(t, "usage: link-or-copy file-num=<num> [file-type=sstable|blob] [shared] [cold-tier] salt=<salt> size=<size> [no-ref-tracking]")
+				}
 				opts := objstorage.CreateOptions{
 					SharedCleanupMethod: objstorage.SharedRefTracking,
 				}
-				ft := base.FileTypeTable
-				if len(d.CmdArgs) > 0 && d.CmdArgs[0].Key == "file-type" {
-					ft = base.FileTypeFromName(d.CmdArgs[0].FirstVal(t))
-					d.CmdArgs = d.CmdArgs[1:]
-				}
-				if len(d.CmdArgs) == 5 && d.CmdArgs[4].Key == "no-ref-tracking" {
-					d.CmdArgs = d.CmdArgs[:4]
+				if noRefTracking {
 					opts.SharedCleanupMethod = objstorage.SharedNoCleanup
 				}
-				var fileNum base.DiskFileNum
-				var typ string
-				var salt, size int
-				scanArgs("[file-type=sstable|blob] <file-num> <local|shared> <salt> <size> [no-ref-tracking]", &fileNum, &typ, &salt, &size)
-				switch typ {
-				case "local":
-				case "shared":
-					opts.PreferSharedStorage = true
-				default:
-					d.Fatalf(t, "'%s' should be 'local' or 'shared'", typ)
-				}
+				opts.PreferSharedStorage = shared
 
 				tmpFileCounter++
 				tmpFilename := fmt.Sprintf("temp-file-%d", tmpFileCounter)
@@ -206,11 +201,14 @@ func TestProvider(t *testing.T) {
 				require.NoError(t, err)
 				require.NoError(t, f.Close())
 
-				_, err = curProvider.LinkOrCopyFromLocal(ctx, fs, tmpFilename, ft, fileNum, opts)
+				_, err = curProvider.LinkOrCopyFromLocal(ctx, fs, tmpFilename, fileType, fileNum, opts)
 				require.NoError(t, err)
 				return log.String()
 
 			case "read":
+				if fileNum == 0 {
+					d.Fatalf(t, "usage: read file-num=<num> [file-type=sstable|blob] <file-num> [for-compaction] [readahead|speculative-overhead=off|sys-readahead|fadvise-sequential]")
+				}
 				forCompaction := d.HasArg("for-compaction")
 				if arg, ok := d.Arg("readahead"); ok {
 					var mode ReadaheadMode
@@ -231,15 +229,7 @@ func TestProvider(t *testing.T) {
 					}
 				}
 
-				ft := base.FileTypeTable
-				if len(d.CmdArgs) > 0 && d.CmdArgs[0].Key == "file-type" {
-					ft = base.FileTypeFromName(d.CmdArgs[0].FirstVal(t))
-					d.CmdArgs = d.CmdArgs[1:]
-				}
-				d.CmdArgs = d.CmdArgs[:1]
-				var fileNum base.DiskFileNum
-				scanArgs("[file-type=sstable|blob] <file-num> [for-compaction] [readahead|speculative-overhead=off|sys-readahead|fadvise-sequential]", &fileNum)
-				r, err := curProvider.OpenForReading(ctx, ft, fileNum, objstorage.OpenOptions{})
+				r, err := curProvider.OpenForReading(ctx, fileType, fileNum, objstorage.OpenOptions{})
 				if err != nil {
 					return err.Error()
 				}
@@ -272,14 +262,10 @@ func TestProvider(t *testing.T) {
 				return log.String()
 
 			case "remove":
-				ft := base.FileTypeTable
-				if len(d.CmdArgs) > 0 && d.CmdArgs[0].Key == "file-type" {
-					ft = base.FileTypeFromName(d.CmdArgs[0].FirstVal(t))
-					d.CmdArgs = d.CmdArgs[1:]
-				}
-				var fileNum base.DiskFileNum
-				scanArgs("[file-type=sstable|blob] <file-num>", &fileNum)
-				if err := curProvider.Remove(ft, fileNum); err != nil {
+				if fileNum == 0 {
+					d.Fatalf(t, "usage: remove file-num=<num> [file-type=sstable|blob]")
+				}
+				if err := curProvider.Remove(fileType, fileNum); err != nil {
 					return err.Error()
 				}
 				return log.String()
@@ -291,9 +277,12 @@ func TestProvider(t *testing.T) {
 				return log.String()
 
 			case "save-backing":
+				if fileNum == 0 {
+					d.Fatalf(t, "usage: save-backing file-num=<num> [file-type=sstable|blob] key=<key>")
+				}
 				var key string
-				var fileNum base.DiskFileNum
-				scanArgs("<key> <file-num>", &key, &fileNum)
+				d.ScanArgs(t, "key", &key)
+
 				meta, err := curProvider.Lookup(base.FileTypeTable, fileNum)
 				require.NoError(t, err)
 				var handle objstorage.RemoteObjectBackingHandle
@@ -312,7 +301,7 @@ func TestProvider(t *testing.T) {
 
 			case "close-backing":
 				var key string
-				scanArgs("<key>", &key)
+				d.ScanArgs(t, "key", &key)
 				backingHandles[key].Close()
 				return ""
 
diff --git a/objstorage/objstorageprovider/shared_writable.go b/objstorage/objstorageprovider/shared_writable.go
index 5e8d45ad2f..c71a2345d4 100644
--- a/objstorage/objstorageprovider/shared_writable.go
+++ b/objstorage/objstorageprovider/shared_writable.go
@@ -33,6 +33,9 @@ func (w *sharedWritable) Write(p []byte) error {
 	return err
 }
 
+// StartMetadataPortion is part of the Writable interface.
+func (w *sharedWritable) StartMetadataPortion() error { return nil }
+
 // Finish is part of the Writable interface.
 func (w *sharedWritable) Finish() error {
 	err := w.storageWriter.Close()
diff --git a/objstorage/objstorageprovider/testdata/provider/cold_tier b/objstorage/objstorageprovider/testdata/provider/cold_tier
index c95337d70d..e642b614d9 100644
--- a/objstorage/objstorageprovider/testdata/provider/cold_tier
+++ b/objstorage/objstorageprovider/testdata/provider/cold_tier
@@ -1,4 +1,4 @@
-open p1 cold-tier=cold1
+open dir=p1 cold-dir=cold1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> mkdir-all: cold1 0755
@@ -6,13 +6,13 @@ open p1 cold-tier=cold1
 <local fs> open-dir: cold1
 
 # Create a cold file.
-create file-type=blob 1 local 1 1024 cold-tier
+create file-type=blob file-num=1 cold-tier salt=1 size=1024
 ----
 <local fs> create: cold1/000001.blob
 <local fs> sync-data: cold1/000001.blob
 <local fs> close: cold1/000001.blob
 
-read file-type=blob 1
+read file-type=blob file-num=1
 0 500
 512 1024
 ----
@@ -24,13 +24,13 @@ size: 1024
 512 1024: EOF
 <local fs> close: cold1/000001.blob
 
-create 2 local 2 1024
+create file-num=2 salt=2 size=1024
 ----
 <local fs> create: p1/000002.sst
 <local fs> sync-data: p1/000002.sst
 <local fs> close: p1/000002.sst
 
-read 2
+read file-num=2
 0 500
 512 1024
 ----
@@ -47,27 +47,27 @@ list
 000001 -> cold1/000001.blob
 000002 -> p1/000002.sst
 
-remove file-type=blob 1
+remove file-type=blob file-num=1
 ----
 <local fs> remove: cold1/000001.blob
 
-remove 2
+remove file-num=2
 ----
 <local fs> remove: p1/000002.sst
 
 # Verify that we can request cold tier even if it is not configured.
-open p2
+open dir=p2
 ----
 <local fs> mkdir-all: p2 0755
 <local fs> open-dir: p2
 
-create file-type=blob 3 local 3 1024 cold-tier
+create file-type=blob file-num=3 cold-tier salt=3 size=1024
 ----
 <local fs> create: p2/000003.blob
 <local fs> sync-data: p2/000003.blob
 <local fs> close: p2/000003.blob
 
-read file-type=blob 3
+read file-type=blob file-num=3
 0 500
 512 1024
 ----
@@ -79,6 +79,6 @@ size: 1024
 512 1024: EOF
 <local fs> close: p2/000003.blob
 
-remove file-type=blob 3
+remove file-type=blob file-num=3
 ----
 <local fs> remove: p2/000003.blob
diff --git a/objstorage/objstorageprovider/testdata/provider/cold_tier_metadata b/objstorage/objstorageprovider/testdata/provider/cold_tier_metadata
new file mode 100644
index 0000000000..b88e22668d
--- /dev/null
+++ b/objstorage/objstorageprovider/testdata/provider/cold_tier_metadata
@@ -0,0 +1,190 @@
+open dir=hot cold-dir=cold
+----
+<local fs> mkdir-all: hot 0755
+<local fs> mkdir-all: cold 0755
+<local fs> open-dir: hot
+<local fs> open-dir: cold
+
+# Create a blob file with metadata starting at offset 512.
+# All 1024 bytes go to the cold tier, and the last 512 bytes also go to a
+# separate metadata file on the hot tier.
+create file-type=blob file-num=1 cold-tier salt=1 size=1024 meta-offset=512
+----
+<local fs> create: cold/000001.blob
+<local fs> create: hot/000001.blobmeta.512
+<local fs> sync: hot/000001.blobmeta.512
+<local fs> close: hot/000001.blobmeta.512
+<local fs> sync-data: cold/000001.blob
+<local fs> close: cold/000001.blob
+
+# Verify we can read the data correctly. The last two reads should be served
+# from the blobmeta file.
+read file-type=blob file-num=1
+0 500
+500 100
+512 512
+1000 24
+----
+<local fs> open: cold/000001.blob (options: *vfs.randomReadsOption)
+size: 1024
+<local fs> read-at(0, 500): cold/000001.blob
+0 500: ok (salt 1)
+<local fs> read-at(500, 100): cold/000001.blob
+500 100: ok (salt 1)
+<local fs> open: hot/000001.blobmeta.512 (options: *vfs.randomReadsOption)
+<local fs> read-at(0, 512): hot/000001.blobmeta.512
+512 512: ok (salt 1)
+<local fs> read-at(488, 24): hot/000001.blobmeta.512
+1000 24: ok (salt 1)
+<local fs> close: cold/000001.blob
+<local fs> close: hot/000001.blobmeta.512
+
+# Create a blob file with metadata starting at offset 0 (all metadata).
+create file-type=blob file-num=2 cold-tier salt=2 size=2048 meta-offset=0
+----
+<local fs> create: cold/000002.blob
+<local fs> create: hot/000002.blobmeta.0
+<local fs> sync: hot/000002.blobmeta.0
+<local fs> close: hot/000002.blobmeta.0
+<local fs> sync-data: cold/000002.blob
+<local fs> close: cold/000002.blob
+
+read file-type=blob file-num=2
+0 2048
+----
+<local fs> open: cold/000002.blob (options: *vfs.randomReadsOption)
+size: 2048
+<local fs> open: hot/000002.blobmeta.0 (options: *vfs.randomReadsOption)
+<local fs> read-at(0, 2048): hot/000002.blobmeta.0
+0 2048: ok (salt 2)
+<local fs> close: cold/000002.blob
+<local fs> close: hot/000002.blobmeta.0
+
+# Create a blob with metadata starting at the end (no metadata).
+create file-type=blob file-num=3 cold-tier salt=3 size=1024 meta-offset=1024
+----
+<local fs> create: cold/000003.blob
+<local fs> create: hot/000003.blobmeta.1024
+<local fs> sync: hot/000003.blobmeta.1024
+<local fs> close: hot/000003.blobmeta.1024
+<local fs> sync-data: cold/000003.blob
+<local fs> close: cold/000003.blob
+
+read file-type=blob file-num=3
+0 1024
+----
+<local fs> open: cold/000003.blob (options: *vfs.randomReadsOption)
+size: 1024
+<local fs> read-at(0, 1024): cold/000003.blob
+0 1024: ok (salt 3)
+<local fs> close: cold/000003.blob
+
+# Create a large blob with metadata that spans multiple buffer flushes (>4KB).
+# Metadata starts at 1024, so we have 1024 bytes of data, then 5120 bytes of metadata.
+create file-type=blob file-num=4 cold-tier salt=4 size=6144 meta-offset=1024
+----
+<local fs> create: cold/000004.blob
+<local fs> create: hot/000004.blobmeta.1024
+<local fs> sync: hot/000004.blobmeta.1024
+<local fs> close: hot/000004.blobmeta.1024
+<local fs> sync-data: cold/000004.blob
+<local fs> close: cold/000004.blob
+
+read file-type=blob file-num=4
+0 1024
+1024 5120
+----
+<local fs> open: cold/000004.blob (options: *vfs.randomReadsOption)
+size: 6144
+<local fs> read-at(0, 1024): cold/000004.blob
+0 1024: ok (salt 4)
+<local fs> open: hot/000004.blobmeta.1024 (options: *vfs.randomReadsOption)
+<local fs> read-at(0, 5120): hot/000004.blobmeta.1024
+1024 5120: ok (salt 4)
+<local fs> close: cold/000004.blob
+<local fs> close: hot/000004.blobmeta.1024
+
+# Create a blob without metadata support (no meta-offset).
+create file-type=blob file-num=5 cold-tier salt=5 size=512
+----
+<local fs> create: cold/000005.blob
+<local fs> sync-data: cold/000005.blob
+<local fs> close: cold/000005.blob
+
+read file-type=blob file-num=5
+0 512
+----
+<local fs> open: cold/000005.blob (options: *vfs.randomReadsOption)
+size: 512
+<local fs> read-at(0, 512): cold/000005.blob
+0 512: ok (salt 5)
+<local fs> close: cold/000005.blob
+
+list
+----
+000001 -> cold/000001.blob
+000002 -> cold/000002.blob
+000003 -> cold/000003.blob
+000004 -> cold/000004.blob
+000005 -> cold/000005.blob
+
+# Verify that we don't lose track of the metadata file if we reopen.
+close
+----
+<local fs> sync: cold
+<local fs> close: hot
+<local fs> close: cold
+
+open dir=hot cold-dir=cold
+----
+<local fs> mkdir-all: hot 0755
+<local fs> mkdir-all: cold 0755
+<local fs> open-dir: hot
+<local fs> open-dir: cold
+
+# Verify we can read the data correctly. The last two reads should be served
+# from the blobmeta file.
+read file-type=blob file-num=1
+0 500
+500 100
+512 512
+1000 24
+----
+<local fs> open: cold/000001.blob (options: *vfs.randomReadsOption)
+size: 1024
+<local fs> read-at(0, 500): cold/000001.blob
+0 500: ok (salt 1)
+<local fs> read-at(500, 100): cold/000001.blob
+500 100: ok (salt 1)
+<local fs> open: hot/000001.blobmeta.512 (options: *vfs.randomReadsOption)
+<local fs> read-at(0, 512): hot/000001.blobmeta.512
+512 512: ok (salt 1)
+<local fs> read-at(488, 24): hot/000001.blobmeta.512
+1000 24: ok (salt 1)
+<local fs> close: cold/000001.blob
+<local fs> close: hot/000001.blobmeta.512
+
+# Clean up.
+remove file-type=blob file-num=1
+----
+<local fs> remove: cold/000001.blob
+<local fs> remove: hot/000001.blobmeta.512
+
+remove file-type=blob file-num=2
+----
+<local fs> remove: cold/000002.blob
+<local fs> remove: hot/000002.blobmeta.0
+
+remove file-type=blob file-num=3
+----
+<local fs> remove: cold/000003.blob
+<local fs> remove: hot/000003.blobmeta.1024
+
+remove file-type=blob file-num=4
+----
+<local fs> remove: cold/000004.blob
+<local fs> remove: hot/000004.blobmeta.1024
+
+remove file-type=blob file-num=5
+----
+<local fs> remove: cold/000005.blob
diff --git a/objstorage/objstorageprovider/testdata/provider/local b/objstorage/objstorageprovider/testdata/provider/local
index ff2a98f610..79dcf61b37 100644
--- a/objstorage/objstorageprovider/testdata/provider/local
+++ b/objstorage/objstorageprovider/testdata/provider/local
@@ -1,18 +1,18 @@
 # Basic provider tests without shared storage.
 
-open p0
+open dir=p0
 ----
 <local fs> mkdir-all: p0 0755
 <local fs> open-dir: p0
 
-create 1 local 1 1024
+create file-num=1 salt=1 size=1024
 foo
 ----
 <local fs> create: p0/000001.sst
 <local fs> sync-data: p0/000001.sst
 <local fs> close: p0/000001.sst
 
-read 1
+read file-num=1
 0 512
 0 1024
 512 1024
@@ -30,13 +30,13 @@ size: 1024
 
 # A provider without shared storage creates object with shared preference
 # locally.
-create 2 shared 2 1024
+create file-num=2 shared salt=2 size=1024
 ----
 <local fs> create: p0/000002.sst
 <local fs> sync-data: p0/000002.sst
 <local fs> close: p0/000002.sst
 
-read 2
+read file-num=2
 0 512
 0 1024
 512 1024
@@ -52,7 +52,7 @@ size: 1024
 512 1024: EOF
 <local fs> close: p0/000002.sst
 
-remove 1
+remove file-num=1
 ----
 <local fs> remove: p0/000001.sst
 
@@ -60,17 +60,17 @@ list
 ----
 000002 -> p0/000002.sst
 
-read 1
+read file-num=1
 ----
 file 000001 (type sstable) unknown to the objstorage provider: file does not exist
 
-link-or-copy 3 local 3 100
+link-or-copy file-num=3 salt=3 size=100
 ----
 <local fs> create: temp-file-1
 <local fs> close: temp-file-1
 <local fs> link: temp-file-1 -> p0/000003.sst
 
-read 3
+read file-num=3
 0 100
 ----
 <local fs> open: p0/000003.sst (options: *vfs.randomReadsOption)
@@ -79,13 +79,13 @@ size: 100
 0 100: ok (salt 3)
 <local fs> close: p0/000003.sst
 
-link-or-copy 4 shared 4 1234
+link-or-copy file-num=4 shared salt=4 size=1234
 ----
 <local fs> create: temp-file-2
 <local fs> close: temp-file-2
 <local fs> link: temp-file-2 -> p0/000004.sst
 
-read 4
+read file-num=4
 0 1234
 ----
 <local fs> open: p0/000004.sst (options: *vfs.randomReadsOption)
@@ -94,13 +94,13 @@ size: 1234
 0 1234: ok (salt 4)
 <local fs> close: p0/000004.sst
 
-create file-type=blob 000005 local 1 4096
+create file-type=blob file-num=000005 salt=1 size=4096
 ----
 <local fs> create: p0/000005.blob
 <local fs> sync-data: p0/000005.blob
 <local fs> close: p0/000005.blob
 
-read file-type=blob 000005
+read file-type=blob file-num=000005
 0 1024
 2048 1024
 ----
@@ -112,7 +112,7 @@ size: 4096
 2048 1024: ok (salt 1)
 <local fs> close: p0/000005.blob
 
-link-or-copy file-type=blob 000006 shared 6 1234
+link-or-copy file-type=blob file-num=000006 shared salt=6 size=1234
 ----
 <local fs> create: temp-file-3
 <local fs> close: temp-file-3
diff --git a/objstorage/objstorageprovider/testdata/provider/local_readahead b/objstorage/objstorageprovider/testdata/provider/local_readahead
index 69bf5667ec..4007ba8b2e 100644
--- a/objstorage/objstorageprovider/testdata/provider/local_readahead
+++ b/objstorage/objstorageprovider/testdata/provider/local_readahead
@@ -1,4 +1,4 @@
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -11,7 +11,7 @@ open p1 creator-id=1
 <local fs> sync: p1
 <local fs> sync: p1/REMOTE-OBJ-CATALOG-000001
 
-create 1 local 1 2000000
+create file-num=1 salt=1 size=2000000
 ----
 <local fs> create: p1/000001.sst
 <local fs> sync-data: p1/000001.sst
@@ -20,7 +20,7 @@ create 1 local 1 2000000
 
 # We should see prefetch calls, and eventually a reopen with sequential reads
 # option.
-read 1
+read file-num=1
 0 1000
 1000 15000
 16000 30000
@@ -52,7 +52,7 @@ size: 2000000
 <local fs> close: p1/000001.sst
 
 # We should directly see a reopen with sequential reads option.
-read 1 for-compaction
+read file-num=1 for-compaction
 0 1000
 1000 15000
 ----
@@ -68,7 +68,7 @@ size: 2000000
 
 # Test non-default readahead modes.
 
-read 1 readahead=off
+read file-num=1 readahead=off
 0 1000
 1000 15000
 16000 30000
@@ -95,7 +95,7 @@ size: 2000000
 140000 80000: ok (salt 1)
 <local fs> close: p1/000001.sst
 
-read 1 for-compaction readahead=off
+read file-num=1 for-compaction readahead=off
 0 1000
 1000 15000
 16000 30000
@@ -122,7 +122,7 @@ size: 2000000
 140000 80000: ok (salt 1)
 <local fs> close: p1/000001.sst
 
-read 1 readahead=sys-readahead
+read file-num=1 readahead=sys-readahead
 0 1000
 1000 15000
 16000 30000
@@ -154,7 +154,7 @@ size: 2000000
 
 # TODO(radu): for informed/sys-readahead, we should start with the maximum
 # prefetch window.
-read 1 for-compaction readahead=sys-readahead
+read file-num=1 for-compaction readahead=sys-readahead
 0 1000
 1000 15000
 16000 30000
diff --git a/objstorage/objstorageprovider/testdata/provider/shared_attach b/objstorage/objstorageprovider/testdata/provider/shared_attach
index 393d7f5f81..7517fb1728 100644
--- a/objstorage/objstorageprovider/testdata/provider/shared_attach
+++ b/objstorage/objstorageprovider/testdata/provider/shared_attach
@@ -1,7 +1,7 @@
 # Basic tests for obtaining the backing of shared objects and attaching them to
 # another provider.
 
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -14,28 +14,28 @@ open p1 creator-id=1
 <local fs> sync: p1
 <local fs> sync: p1/REMOTE-OBJ-CATALOG-000001
 
-create 1 shared 1 100
+create file-num=1 shared salt=1 size=100
 ----
 <remote> create object "61a6-1-000001.sst"
 <remote> close writer for "61a6-1-000001.sst" after 100 bytes
 <remote> create object "61a6-1-000001.sst.ref.1.000001"
 <remote> close writer for "61a6-1-000001.sst.ref.1.000001" after 0 bytes
 
-create 2 shared 2 200
+create file-num=2 shared salt=2 size=200
 ----
 <remote> create object "a629-1-000002.sst"
 <remote> close writer for "a629-1-000002.sst" after 200 bytes
 <remote> create object "a629-1-000002.sst.ref.1.000002"
 <remote> close writer for "a629-1-000002.sst.ref.1.000002" after 0 bytes
 
-create 3 shared 3 300
+create file-num=3 shared salt=3 size=300
 ----
 <remote> create object "eaac-1-000003.sst"
 <remote> close writer for "eaac-1-000003.sst" after 300 bytes
 <remote> create object "eaac-1-000003.sst.ref.1.000003"
 <remote> close writer for "eaac-1-000003.sst.ref.1.000003" after 0 bytes
 
-create 100 local 100 15
+create file-num=100 salt=100 size=15
 ----
 <local fs> create: p1/000100.sst
 <local fs> sync-data: p1/000100.sst
@@ -49,17 +49,17 @@ list
 000100 -> p1/000100.sst
 
 # Can't get backing of local object.
-save-backing foo 100
+save-backing key=foo file-num=100
 ----
 object 000100 not on remote storage
 
-save-backing b1 1
+save-backing key=b1 file-num=1
 ----
 
-save-backing b2 2
+save-backing key=b2 file-num=2
 ----
 
-save-backing b3 3
+save-backing key=b3 file-num=3
 ----
 
 close
@@ -70,7 +70,7 @@ close
 <local fs> close: p1/REMOTE-OBJ-CATALOG-000001
 <local fs> close: p1
 
-open p2 creator-id=2
+open dir=p2 creator-id=2
 ----
 <local fs> mkdir-all: p2 0755
 <local fs> open-dir: p2
@@ -83,7 +83,7 @@ open p2 creator-id=2
 <local fs> sync: p2
 <local fs> sync: p2/REMOTE-OBJ-CATALOG-000001
 
-create 100 shared 100 15
+create file-num=100 shared salt=100 size=15
 ----
 <remote> create object "fd72-2-000100.sst"
 <remote> close writer for "fd72-2-000100.sst" after 15 bytes
@@ -115,7 +115,7 @@ list
 000102 -> remote://a629-1-000002.sst
 000103 -> remote://eaac-1-000003.sst
 
-read 101
+read file-num=101
 0 100
 15 10
 ----
@@ -128,7 +128,7 @@ size: 100
 15 10: ok (salt 1)
 <remote> close reader for "61a6-1-000001.sst"
 
-read 102
+read file-num=102
 0 200
 90 100
 ----
@@ -141,7 +141,7 @@ size: 200
 90 100: ok (salt 2)
 <remote> close reader for "a629-1-000002.sst"
 
-read 103
+read file-num=103
 0 300
 ----
 <remote> size of object "eaac-1-000003.sst.ref.2.000103": 0
diff --git a/objstorage/objstorageprovider/testdata/provider/shared_attach_after_unref b/objstorage/objstorageprovider/testdata/provider/shared_attach_after_unref
index 3c2460e2d0..e1c011cdef 100644
--- a/objstorage/objstorageprovider/testdata/provider/shared_attach_after_unref
+++ b/objstorage/objstorageprovider/testdata/provider/shared_attach_after_unref
@@ -1,6 +1,6 @@
 # Tests when an object is unrefed before it is attached to another provider.
 
-open p5 creator-id=5
+open dir=p5 creator-id=5
 ----
 <local fs> mkdir-all: p5 0755
 <local fs> open-dir: p5
@@ -13,21 +13,21 @@ open p5 creator-id=5
 <local fs> sync: p5
 <local fs> sync: p5/REMOTE-OBJ-CATALOG-000001
 
-create 1 shared 1 100
+create file-num=1 shared salt=1 size=100
 ----
 <remote> create object "d632-5-000001.sst"
 <remote> close writer for "d632-5-000001.sst" after 100 bytes
 <remote> create object "d632-5-000001.sst.ref.5.000001"
 <remote> close writer for "d632-5-000001.sst.ref.5.000001" after 0 bytes
 
-save-backing p5b1 1
+save-backing key=p5b1 file-num=1
 ----
 
 # This should do nothing.
-remove 1
+remove file-num=1
 ----
 
-open p6 creator-id=6
+open dir=p6 creator-id=6
 ----
 <local fs> mkdir-all: p6 0755
 <local fs> open-dir: p6
@@ -49,34 +49,34 @@ p5b1 101
 <remote> size of object "d632-5-000001.sst.ref.5.000001": 0
 000101 -> remote://d632-5-000001.sst
 
-switch p5
+switch dir=p5
 ----
 
 # TODO(radu): after we close the backing, the unref should happen.
-close-backing p5b1
+close-backing key=p5b1
 ----
 
-create 2 shared 2 100
+create file-num=2 shared salt=2 size=100
 ----
 <remote> create object "1ab5-5-000002.sst"
 <remote> close writer for "1ab5-5-000002.sst" after 100 bytes
 <remote> create object "1ab5-5-000002.sst.ref.5.000002"
 <remote> close writer for "1ab5-5-000002.sst.ref.5.000002" after 0 bytes
 
-save-backing p5b2 2
+save-backing key=p5b2 file-num=2
 ----
 
 # Close the backing, then unref the object.
-close-backing p5b2
+close-backing key=p5b2
 ----
 
-remove 2
+remove file-num=2
 ----
 <remote> delete object "1ab5-5-000002.sst.ref.5.000002"
 <remote> list (prefix="1ab5-5-000002.sst.ref.", delimiter="")
 <remote> delete object "1ab5-5-000002.sst"
 
-switch p6
+switch dir=p6
 ----
 
 # Attach should error out because it can't find p5's ref.
diff --git a/objstorage/objstorageprovider/testdata/provider/shared_attach_multi b/objstorage/objstorageprovider/testdata/provider/shared_attach_multi
index 94d32556c9..49a3d6fb03 100644
--- a/objstorage/objstorageprovider/testdata/provider/shared_attach_multi
+++ b/objstorage/objstorageprovider/testdata/provider/shared_attach_multi
@@ -1,6 +1,6 @@
 # Tests with the same shared object attached as multiple objects.
 
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -13,17 +13,17 @@ open p1 creator-id=1
 <local fs> sync: p1
 <local fs> sync: p1/REMOTE-OBJ-CATALOG-000001
 
-create 1 shared 1 100
+create file-num=1 shared salt=1 size=100
 ----
 <remote> create object "61a6-1-000001.sst"
 <remote> close writer for "61a6-1-000001.sst" after 100 bytes
 <remote> create object "61a6-1-000001.sst.ref.1.000001"
 <remote> close writer for "61a6-1-000001.sst.ref.1.000001" after 0 bytes
 
-save-backing b1 1
+save-backing key=b1 file-num=1
 ----
 
-open p2 creator-id=2
+open dir=p2 creator-id=2
 ----
 <local fs> mkdir-all: p2 0755
 <local fs> open-dir: p2
@@ -55,14 +55,14 @@ b1 103
 000102 -> remote://61a6-1-000001.sst
 000103 -> remote://61a6-1-000001.sst
 
-close-backing b1
+close-backing key=b1
 ----
 
 # Remove original object.
-switch p1
+switch dir=p1
 ----
 
-remove 1
+remove file-num=1
 ----
 <remote> delete object "61a6-1-000001.sst.ref.1.000001"
 <remote> list (prefix="61a6-1-000001.sst.ref.", delimiter="")
@@ -70,23 +70,23 @@ remove 1
 <remote>  - 61a6-1-000001.sst.ref.2.000102
 <remote>  - 61a6-1-000001.sst.ref.2.000103
 
-switch p2
+switch dir=p2
 ----
 
-remove 101
+remove file-num=101
 ----
 <remote> delete object "61a6-1-000001.sst.ref.2.000101"
 <remote> list (prefix="61a6-1-000001.sst.ref.", delimiter="")
 <remote>  - 61a6-1-000001.sst.ref.2.000102
 <remote>  - 61a6-1-000001.sst.ref.2.000103
 
-remove 103
+remove file-num=103
 ----
 <remote> delete object "61a6-1-000001.sst.ref.2.000103"
 <remote> list (prefix="61a6-1-000001.sst.ref.", delimiter="")
 <remote>  - 61a6-1-000001.sst.ref.2.000102
 
-remove 102
+remove file-num=102
 ----
 <remote> delete object "61a6-1-000001.sst.ref.2.000102"
 <remote> list (prefix="61a6-1-000001.sst.ref.", delimiter="")
diff --git a/objstorage/objstorageprovider/testdata/provider/shared_basic b/objstorage/objstorageprovider/testdata/provider/shared_basic
index 13269be883..f4933204a4 100644
--- a/objstorage/objstorageprovider/testdata/provider/shared_basic
+++ b/objstorage/objstorageprovider/testdata/provider/shared_basic
@@ -1,6 +1,6 @@
 # Basic provider tests with shared storage.
 
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -13,13 +13,13 @@ open p1 creator-id=1
 <local fs> sync: p1
 <local fs> sync: p1/REMOTE-OBJ-CATALOG-000001
 
-create 1 local 1 100
+create file-num=1 salt=1 size=100
 ----
 <local fs> create: p1/000001.sst
 <local fs> sync-data: p1/000001.sst
 <local fs> close: p1/000001.sst
 
-read 1
+read file-num=1
 0 100
 ----
 <local fs> open: p1/000001.sst (options: *vfs.randomReadsOption)
@@ -28,14 +28,14 @@ size: 100
 0 100: ok (salt 1)
 <local fs> close: p1/000001.sst
 
-create 2 shared 2 100
+create file-num=2 shared salt=2 size=100
 ----
 <remote> create object "a629-1-000002.sst"
 <remote> close writer for "a629-1-000002.sst" after 100 bytes
 <remote> create object "a629-1-000002.sst.ref.1.000002"
 <remote> close writer for "a629-1-000002.sst.ref.1.000002" after 0 bytes
 
-read 2
+read file-num=2
 0 100
 ----
 <remote> size of object "a629-1-000002.sst.ref.1.000002": 0
@@ -59,7 +59,7 @@ close
 <local fs> close: p1
 
 # Test that the objects are there on re-open.
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -72,23 +72,23 @@ list
 000001 -> p1/000001.sst
 000002 -> remote://a629-1-000002.sst
 
-remove 1
+remove file-num=1
 ----
 <local fs> remove: p1/000001.sst
 
-remove 2
+remove file-num=2
 ----
 <remote> delete object "a629-1-000002.sst.ref.1.000002"
 <remote> list (prefix="a629-1-000002.sst.ref.", delimiter="")
 <remote> delete object "a629-1-000002.sst"
 
-link-or-copy 3 local 3 100
+link-or-copy file-num=3 salt=3 size=100
 ----
 <local fs> create: temp-file-1
 <local fs> close: temp-file-1
 <local fs> link: temp-file-1 -> p1/000003.sst
 
-read 3
+read file-num=3
 0 100
 ----
 <local fs> open: p1/000003.sst (options: *vfs.randomReadsOption)
@@ -97,7 +97,7 @@ size: 100
 0 100: ok (salt 3)
 <local fs> close: p1/000003.sst
 
-link-or-copy 4 shared 4 100
+link-or-copy file-num=4 shared salt=4 size=100
 ----
 <local fs> create: temp-file-2
 <local fs> close: temp-file-2
@@ -108,7 +108,7 @@ link-or-copy 4 shared 4 100
 <remote> close writer for "2f2f-1-000004.sst.ref.1.000004" after 0 bytes
 <local fs> close: temp-file-2
 
-read 4
+read file-num=4
 0 100
 ----
 <remote> size of object "2f2f-1-000004.sst.ref.1.000004": 0
diff --git a/objstorage/objstorageprovider/testdata/provider/shared_no_ref b/objstorage/objstorageprovider/testdata/provider/shared_no_ref
index 02cde23db1..c9b5c5deab 100644
--- a/objstorage/objstorageprovider/testdata/provider/shared_no_ref
+++ b/objstorage/objstorageprovider/testdata/provider/shared_no_ref
@@ -1,6 +1,6 @@
 # Tests with shared storage when ref tracking is disabled.
 
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -13,28 +13,34 @@ open p1 creator-id=1
 <local fs> sync: p1
 <local fs> sync: p1/REMOTE-OBJ-CATALOG-000001
 
-create 1 shared 1 100 no-ref-tracking
+create file-num=1 shared salt=1 size=100 no-ref-tracking
 ----
 <remote> create object "61a6-1-000001.sst"
 <remote> close writer for "61a6-1-000001.sst" after 100 bytes
+<remote> create object "61a6-1-000001.sst.ref.1.000001"
+<remote> close writer for "61a6-1-000001.sst.ref.1.000001" after 0 bytes
 
-read 1
+read file-num=1
 0 100
 ----
+<remote> size of object "61a6-1-000001.sst.ref.1.000001": 0
 <remote> create reader for object "61a6-1-000001.sst": 100 bytes
 size: 100
 <remote> read object "61a6-1-000001.sst" at 0 (length 100)
 0 100: ok (salt 1)
 <remote> close reader for "61a6-1-000001.sst"
 
-create 2 shared 2 100 no-ref-tracking
+create file-num=2 shared salt=2 size=100 no-ref-tracking
 ----
 <remote> create object "a629-1-000002.sst"
 <remote> close writer for "a629-1-000002.sst" after 100 bytes
+<remote> create object "a629-1-000002.sst.ref.1.000002"
+<remote> close writer for "a629-1-000002.sst.ref.1.000002" after 0 bytes
 
-read 2
+read file-num=2
 0 100
 ----
+<remote> size of object "a629-1-000002.sst.ref.1.000002": 0
 <remote> create reader for object "a629-1-000002.sst": 100 bytes
 size: 100
 <remote> read object "a629-1-000002.sst" at 0 (length 100)
@@ -46,7 +52,7 @@ list
 000001 -> remote://61a6-1-000001.sst
 000002 -> remote://a629-1-000002.sst
 
-link-or-copy 3 shared 3 100 no-ref-tracking
+link-or-copy file-num=3 shared salt=3 size=100 no-ref-tracking
 ----
 <local fs> create: temp-file-1
 <local fs> close: temp-file-1
@@ -55,7 +61,7 @@ link-or-copy 3 shared 3 100 no-ref-tracking
 <remote> close writer for "eaac-1-000003.sst" after 100 bytes
 <local fs> close: temp-file-1
 
-read 3
+read file-num=3
 0 100
 ----
 <remote> create reader for object "eaac-1-000003.sst": 100 bytes
@@ -72,7 +78,7 @@ close
 <local fs> close: p1
 
 # Test that the objects are there on re-open.
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -86,25 +92,27 @@ list
 000002 -> remote://a629-1-000002.sst
 000003 -> remote://eaac-1-000003.sst
 
-read 1
+read file-num=1
 0 100
 ----
+<remote> size of object "61a6-1-000001.sst.ref.1.000001": 0
 <remote> create reader for object "61a6-1-000001.sst": 100 bytes
 size: 100
 <remote> read object "61a6-1-000001.sst" at 0 (length 100)
 0 100: ok (salt 1)
 <remote> close reader for "61a6-1-000001.sst"
 
-read 2
+read file-num=2
 0 100
 ----
+<remote> size of object "a629-1-000002.sst.ref.1.000002": 0
 <remote> create reader for object "a629-1-000002.sst": 100 bytes
 size: 100
 <remote> read object "a629-1-000002.sst" at 0 (length 100)
 0 100: ok (salt 2)
 <remote> close reader for "a629-1-000002.sst"
 
-read 3
+read file-num=3
 0 100
 ----
 <remote> create reader for object "eaac-1-000003.sst": 100 bytes
@@ -113,13 +121,13 @@ size: 100
 0 100: ok (salt 3)
 <remote> close reader for "eaac-1-000003.sst"
 
-save-backing b1 1
+save-backing key=b1 file-num=1
 ----
 
-save-backing b2 1
+save-backing key=b2 file-num=1
 ----
 
-open p2 creator-id=2
+open dir=p2 creator-id=2
 ----
 <local fs> mkdir-all: p2 0755
 <local fs> open-dir: p2
@@ -136,6 +144,12 @@ attach
 b1 101
 b2 102
 ----
+<remote> create object "61a6-1-000001.sst.ref.2.000101"
+<remote> close writer for "61a6-1-000001.sst.ref.2.000101" after 0 bytes
+<remote> size of object "61a6-1-000001.sst.ref.1.000001": 0
+<remote> create object "61a6-1-000001.sst.ref.2.000102"
+<remote> close writer for "61a6-1-000001.sst.ref.2.000102" after 0 bytes
+<remote> size of object "61a6-1-000001.sst.ref.1.000001": 0
 000101 -> remote://61a6-1-000001.sst
 000102 -> remote://61a6-1-000001.sst
 
@@ -144,18 +158,20 @@ list
 000101 -> remote://61a6-1-000001.sst
 000102 -> remote://61a6-1-000001.sst
 
-read 101
+read file-num=101
 0 100
 ----
+<remote> size of object "61a6-1-000001.sst.ref.2.000101": 0
 <remote> create reader for object "61a6-1-000001.sst": 100 bytes
 size: 100
 <remote> read object "61a6-1-000001.sst" at 0 (length 100)
 0 100: ok (salt 1)
 <remote> close reader for "61a6-1-000001.sst"
 
-read 102
+read file-num=102
 0 100
 ----
+<remote> size of object "61a6-1-000001.sst.ref.2.000102": 0
 <remote> create reader for object "61a6-1-000001.sst": 100 bytes
 size: 100
 <remote> read object "61a6-1-000001.sst" at 0 (length 100)
@@ -163,17 +179,27 @@ size: 100
 <remote> close reader for "61a6-1-000001.sst"
 
 # In this mode, all removes should be no-ops on the shared backend.
-remove 101
+remove file-num=101
 ----
+<remote> delete object "61a6-1-000001.sst.ref.2.000101"
+<remote> list (prefix="61a6-1-000001.sst.ref.", delimiter="")
+<remote>  - 61a6-1-000001.sst.ref.1.000001
+<remote>  - 61a6-1-000001.sst.ref.2.000102
 
-remove 102
+remove file-num=102
 ----
+<remote> delete object "61a6-1-000001.sst.ref.2.000102"
+<remote> list (prefix="61a6-1-000001.sst.ref.", delimiter="")
+<remote>  - 61a6-1-000001.sst.ref.1.000001
 
-switch p1
+switch dir=p1
 ----
 
-remove 1
+remove file-num=1
 ----
 
-remove 2
+remove file-num=2
 ----
+<remote> delete object "a629-1-000002.sst.ref.1.000002"
+<remote> list (prefix="a629-1-000002.sst.ref.", delimiter="")
+<remote> delete object "a629-1-000002.sst"
diff --git a/objstorage/objstorageprovider/testdata/provider/shared_readahead b/objstorage/objstorageprovider/testdata/provider/shared_readahead
index e3fe67fad4..b4f6f01b7c 100644
--- a/objstorage/objstorageprovider/testdata/provider/shared_readahead
+++ b/objstorage/objstorageprovider/testdata/provider/shared_readahead
@@ -1,4 +1,4 @@
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -11,7 +11,7 @@ open p1 creator-id=1
 <local fs> sync: p1
 <local fs> sync: p1/REMOTE-OBJ-CATALOG-000001
 
-create 1 shared 1 2000000
+create file-num=1 shared salt=1 size=2000000
 ----
 <remote> create object "61a6-1-000001.sst"
 <remote> close writer for "61a6-1-000001.sst" after 2000000 bytes
@@ -20,7 +20,7 @@ create 1 shared 1 2000000
 
 # We should be seeing larger and larger reads. But the last read should be
 # capped to the object size.
-read 1
+read file-num=1
 0 1000
 1000 15000
 16000 30000
@@ -61,7 +61,7 @@ size: 2000000
 <remote> close reader for "61a6-1-000001.sst"
 
 # When reading for a compaction, we should be doing large reads from the start.
-read 1 for-compaction
+read file-num=1 for-compaction
 0 1000
 1000 15000
 16000 30000
@@ -88,14 +88,14 @@ size: 2000000
 <remote> close reader for "61a6-1-000001.sst"
 
 # When reading for a compaction, we should be doing 8MB reads from the start.
-create 2 shared 2 15000000
+create file-num=2 shared salt=2 size=15000000
 ----
 <remote> create object "a629-1-000002.sst"
 <remote> close writer for "a629-1-000002.sst" after 15000000 bytes
 <remote> create object "a629-1-000002.sst.ref.1.000002"
 <remote> close writer for "a629-1-000002.sst.ref.1.000002" after 0 bytes
 
-read 2 for-compaction
+read file-num=2 for-compaction
 0 100000
 9000000 3000000
 ----
diff --git a/objstorage/objstorageprovider/testdata/provider/shared_remove b/objstorage/objstorageprovider/testdata/provider/shared_remove
index 00893ae0b3..f64a6f18b9 100644
--- a/objstorage/objstorageprovider/testdata/provider/shared_remove
+++ b/objstorage/objstorageprovider/testdata/provider/shared_remove
@@ -1,4 +1,4 @@
-open p1 creator-id=1
+open dir=p1 creator-id=1
 ----
 <local fs> mkdir-all: p1 0755
 <local fs> open-dir: p1
@@ -11,34 +11,34 @@ open p1 creator-id=1
 <local fs> sync: p1
 <local fs> sync: p1/REMOTE-OBJ-CATALOG-000001
 
-create 1 shared 1 100
+create file-num=1 shared salt=1 size=100
 ----
 <remote> create object "61a6-1-000001.sst"
 <remote> close writer for "61a6-1-000001.sst" after 100 bytes
 <remote> create object "61a6-1-000001.sst.ref.1.000001"
 <remote> close writer for "61a6-1-000001.sst.ref.1.000001" after 0 bytes
 
-create 2 shared 2 100
+create file-num=2 shared salt=2 size=100
 ----
 <remote> create object "a629-1-000002.sst"
 <remote> close writer for "a629-1-000002.sst" after 100 bytes
 <remote> create object "a629-1-000002.sst.ref.1.000002"
 <remote> close writer for "a629-1-000002.sst.ref.1.000002" after 0 bytes
 
-create 3 shared 3 100
+create file-num=3 shared salt=3 size=100
 ----
 <remote> create object "eaac-1-000003.sst"
 <remote> close writer for "eaac-1-000003.sst" after 100 bytes
 <remote> create object "eaac-1-000003.sst.ref.1.000003"
 <remote> close writer for "eaac-1-000003.sst.ref.1.000003" after 0 bytes
 
-save-backing b1 1
+save-backing key=b1 file-num=1
 ----
 
-save-backing b2 2
+save-backing key=b2 file-num=2
 ----
 
-open p2 creator-id=2
+open dir=p2 creator-id=2
 ----
 <local fs> mkdir-all: p2 0755
 <local fs> open-dir: p2
@@ -51,7 +51,7 @@ open p2 creator-id=2
 <local fs> sync: p2
 <local fs> sync: p2/REMOTE-OBJ-CATALOG-000001
 
-create 4 shared 4 100
+create file-num=4 shared salt=4 size=100
 ----
 <remote> create object "4c52-2-000004.sst"
 <remote> close writer for "4c52-2-000004.sst" after 100 bytes
@@ -72,34 +72,34 @@ b2 102
 000102 -> remote://a629-1-000002.sst
 
 # Remove of object with no other refs; backing object should be removed.
-remove 4
+remove file-num=4
 ----
 <remote> delete object "4c52-2-000004.sst.ref.2.000004"
 <remote> list (prefix="4c52-2-000004.sst.ref.", delimiter="")
 <remote> delete object "4c52-2-000004.sst"
 
 # Object shared with p2; backing object should not be removed.
-remove 101
+remove file-num=101
 ----
 <remote> delete object "61a6-1-000001.sst.ref.2.000101"
 <remote> list (prefix="61a6-1-000001.sst.ref.", delimiter="")
 <remote>  - 61a6-1-000001.sst.ref.1.000001
 
-switch p1
+switch dir=p1
 ----
 
 # Object no longer shared with p1; backing object should be removed.
-remove 1
+remove file-num=1
 ----
 
 # Object shared with p1; backing object should not be removed.
-remove 2
+remove file-num=2
 ----
 
-switch p2
+switch dir=p2
 ----
 
-remove 102
+remove file-num=102
 ----
 <remote> delete object "a629-1-000002.sst.ref.2.000102"
 <remote> list (prefix="a629-1-000002.sst.ref.", delimiter="")
diff --git a/objstorage/objstorageprovider/vfs.go b/objstorage/objstorageprovider/vfs.go
index f8e7ee847e..f56ee83711 100644
--- a/objstorage/objstorageprovider/vfs.go
+++ b/objstorage/objstorageprovider/vfs.go
@@ -6,6 +6,9 @@ package objstorageprovider
 
 import (
 	"context"
+	"fmt"
+	"strconv"
+	"strings"
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
@@ -22,7 +25,7 @@ type localSubsystem struct {
 }
 
 type localLockedState struct {
-	hotTier, coldTier struct {
+	hotTier struct {
 		// objChangeCounter is incremented whenever objects are created.
 		// The purpose of this counter is to avoid syncing the local filesystem when
 		// only remote objects are changed.
@@ -31,6 +34,28 @@ type localLockedState struct {
 		// last completed sync was launched.
 		objChangeCounterLastSync uint64
 	}
+	coldTier struct {
+		// objChangeCounter is incremented whenever objects are created.
+		// The purpose of this counter is to avoid syncing the local filesystem when
+		// only remote objects are changed.
+		objChangeCounter uint64
+		// objChangeCounterLastSync is the value of objChangeCounter at the time the
+		// last completed sync was launched.
+		objChangeCounterLastSync uint64
+
+		// metaFiles stores information about all cold tier objects that have a
+		// known metadata file on the hot tier.
+		metaFiles map[fileTypeAndNum]metaFileInfo
+	}
+}
+
+type fileTypeAndNum struct {
+	fileType base.FileType
+	fileNum  base.DiskFileNum
+}
+
+type metaFileInfo struct {
+	startOffset int64
 }
 
 // objChanged is called after an object was created or deleted. It records the
@@ -52,6 +77,45 @@ func (p *provider) localPath(
 	return p.st.Local.FS, base.MakeFilepath(p.st.Local.FS, p.st.Local.FSDirName, fileType, fileNum)
 }
 
+// metaFileType returns the file type for a file that contains only the metadata
+// for an object of the given type.
+func metaFileType(fileType base.FileType) base.FileType {
+	switch fileType {
+	case base.FileTypeBlob:
+		return base.FileTypeBlobMeta
+	default:
+		panic("unsupported file type for metaFileType")
+	}
+}
+
+// metaPath returns the path to the metadata file for the given object in the
+// cold tier. It is of the form "<file-num>.<file-type>meta.<start-offset>",
+// where start-offset is the offset within the object where the metadata portion
+// starts. For example: "000123.blobmeta.1048576".
+func (p *provider) metaPath(
+	fileType base.FileType, fileNum base.DiskFileNum, startOffset int64,
+) string {
+	metaFileType := metaFileType(fileType)
+	prefix := base.MakeFilepath(p.st.Local.FS, p.st.Local.FSDirName, metaFileType, fileNum)
+	return fmt.Sprintf("%s.%d", prefix, startOffset)
+}
+
+// offsetFromMetaPath parses the start offset from a metadata filename. For
+// example, "000123.blobmeta.1048576" returns 1048576.
+//
+// Assumes the path was already validated by base.ParseFilename.
+func offsetFromMetaPath(filename string) (startOffset int64, ok bool) {
+	idx := strings.LastIndexByte(filename, '.')
+	if idx == -1 {
+		return 0, false
+	}
+	startOffset, err := strconv.ParseInt(filename[idx+1:], 10, 64)
+	if err != nil {
+		return 0, false
+	}
+	return startOffset, true
+}
+
 func (p *provider) localOpenForReading(
 	ctx context.Context,
 	fileType base.FileType,
@@ -68,7 +132,17 @@ func (p *provider) localOpenForReading(
 		}
 		return nil, err
 	}
-	return newFileReadable(file, p.st.Local.FS, p.st.Local.ReadaheadConfig, filename)
+	r, err := newFileReadable(file, p.st.Local.FS, p.st.Local.ReadaheadConfig, filename)
+	if err != nil {
+		return nil, err
+	}
+	if tier == base.ColdTier {
+		if startOffset, ok := p.getColdObjectMetaFile(fileType, fileNum); ok {
+			metaPath := p.metaPath(fileType, fileNum, startOffset)
+			return newColdReadable(r, p.st.Local.FS, metaPath, startOffset), nil
+		}
+	}
+	return r, nil
 }
 
 func (p *provider) vfsCreate(
@@ -78,6 +152,10 @@ func (p *provider) vfsCreate(
 	tier base.StorageTier,
 	category vfs.DiskWriteCategory,
 ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
+	if tier == base.ColdTier && fileType != base.FileTypeBlob {
+		return nil, objstorage.ObjectMetadata{}, errors.Errorf("cold tier not supported for file type %s", fileType)
+	}
+
 	fs, filename := p.localPath(fileType, fileNum, tier)
 	file, err := fs.Create(filename, category)
 	if err != nil {
@@ -92,14 +170,26 @@ func (p *provider) vfsCreate(
 		FileType:    fileType,
 	}
 	meta.Local.Tier = tier
-	return newFileBufferedWritable(file), meta, nil
+	w := objstorage.Writable(newFileBufferedWritable(file))
+	if tier == base.ColdTier {
+		w = newColdWritable(p, fileType, fileNum, w, category)
+	}
+	return w, meta, nil
 }
 
 func (p *provider) localRemove(
 	fileType base.FileType, fileNum base.DiskFileNum, tier base.StorageTier,
 ) error {
 	fs, path := p.localPath(fileType, fileNum, tier)
-	return p.st.Local.FSCleaner.Clean(fs, fileType, path)
+	err := p.st.Local.FSCleaner.Clean(fs, fileType, path)
+	if tier == base.ColdTier {
+		if startOffset, ok := p.popColdObjectMetaFile(fileType, fileNum); ok {
+			metaPath := p.metaPath(fileType, fileNum, startOffset)
+			metaFileType := metaFileType(fileType)
+			err = firstError(err, p.st.Local.FSCleaner.Clean(p.st.Local.FS, metaFileType, metaPath))
+		}
+	}
+	return err
 }
 
 // localInit finds any local FS objects.
@@ -139,13 +229,13 @@ func (p *provider) localInit() error {
 			return err
 		}
 		p.local.coldTier.fsDir = fsDir
-		listing, err := p.st.Local.ColdTier.FS.List(p.st.Local.ColdTier.FSDirName)
+		coldListing, err := p.st.Local.ColdTier.FS.List(p.st.Local.ColdTier.FSDirName)
 		if err != nil {
 			_ = p.localClose()
-			return errors.Wrapf(err, "pebble: could not cold tier directory")
+			return errors.Wrapf(err, "pebble: could not list cold tier directory")
 		}
 
-		for _, filename := range listing {
+		for _, filename := range coldListing {
 			if fileType, fileNum, ok := base.ParseFilename(p.st.Local.FS, filename); ok {
 				switch fileType {
 				case base.FileTypeTable, base.FileTypeBlob:
@@ -162,6 +252,28 @@ func (p *provider) localInit() error {
 				}
 			}
 		}
+		p.mu.local.coldTier.metaFiles = make(map[fileTypeAndNum]metaFileInfo)
+		// Look through the hot tier files to find any metadata files.
+		for _, filename := range listing {
+			if metaFileType, fileNum, ok := base.ParseFilename(p.st.Local.FS, filename); ok && metaFileType == base.FileTypeBlobMeta {
+				startOffset, ok := offsetFromMetaPath(filename)
+				if !ok {
+					p.st.Logger.Errorf("could not parse offset component for %q", filename)
+					continue
+				}
+				if _, ok := p.mu.knownObjects[fileNum]; !ok {
+					// This is a stray file, remove it.
+					filepath := p.st.Local.FS.PathJoin(p.st.Local.FSDirName, filename)
+					_ = p.st.Local.FSCleaner.Clean(p.st.Local.FS, metaFileType, filepath)
+					p.st.Logger.Infof("%q has no matching object, deleting", filepath)
+					continue
+				}
+				p.mu.local.coldTier.metaFiles[fileTypeAndNum{
+					fileType: base.FileTypeBlob,
+					fileNum:  fileNum,
+				}] = metaFileInfo{startOffset: startOffset}
+			}
+		}
 	}
 	return nil
 }
@@ -223,3 +335,47 @@ func (p *provider) localSize(
 	}
 	return stat.Size(), nil
 }
+
+// addColdObjectMetaFile adds an entry to coldTier.metaFiles.
+func (p *provider) addColdObjectMetaFile(
+	fileType base.FileType, fileNum base.DiskFileNum, startOffset int64,
+) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.mu.local.coldTier.metaFiles[fileTypeAndNum{
+		fileType: fileType,
+		fileNum:  fileNum,
+	}] = metaFileInfo{
+		startOffset: startOffset,
+	}
+}
+
+// getColdObjectMetaFile returns ok=true if the object has a metadata file on
+// the hot tier, along with the metadata start offset.
+func (p *provider) getColdObjectMetaFile(
+	fileType base.FileType, fileNum base.DiskFileNum,
+) (startOffset int64, ok bool) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	info, ok := p.mu.local.coldTier.metaFiles[fileTypeAndNum{
+		fileType: fileType,
+		fileNum:  fileNum,
+	}]
+	return info.startOffset, ok
+}
+
+// popColdObjectMetaFile is like getColdObjectMetaFile but also removes the file
+// from the internal map.
+func (p *provider) popColdObjectMetaFile(
+	fileType base.FileType, fileNum base.DiskFileNum,
+) (startOffset int64, ok bool) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	key := fileTypeAndNum{fileType: fileType, fileNum: fileNum}
+	info, ok := p.mu.local.coldTier.metaFiles[key]
+	if !ok {
+		return 0, false
+	}
+	delete(p.mu.local.coldTier.metaFiles, key)
+	return info.startOffset, ok
+}
diff --git a/objstorage/objstorageprovider/vfs_writable.go b/objstorage/objstorageprovider/vfs_writable.go
index 6b8c60c170..a6b7822be0 100644
--- a/objstorage/objstorageprovider/vfs_writable.go
+++ b/objstorage/objstorageprovider/vfs_writable.go
@@ -66,6 +66,9 @@ func (w *fileBufferedWritable) Abort() {
 	w.file = nil
 }
 
+// StartMetadataPortion is part of the objstorage.Writable interface.
+func (w *fileBufferedWritable) StartMetadataPortion() error { return nil }
+
 func firstError(err0, err1 error) error {
 	if err0 != nil {
 		return err0
diff --git a/objstorage/test_utils.go b/objstorage/test_utils.go
index 8e880a38b1..2052c56273 100644
--- a/objstorage/test_utils.go
+++ b/objstorage/test_utils.go
@@ -43,6 +43,9 @@ func (f *MemObj) Write(p []byte) error {
 	return err
 }
 
+// StartMetadataPortion is part of the Writable interface.
+func (f *MemObj) StartMetadataPortion() error { return nil }
+
 // Data returns the in-memory buffer behind this MemObj.
 func (f *MemObj) Data() []byte {
 	return f.buf.Bytes()
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 23220f4c5a..8af164fe24 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -857,6 +857,8 @@ func (f *discardFile) Finish() error {
 
 func (f *discardFile) Abort() {}
 
+func (f *discardFile) StartMetadataPortion() error { return nil }
+
 func (f *discardFile) Write(p []byte) error {
 	f.wrote += int64(len(p))
 	return nil