valsep: create SSTBlobWriter to write external sst with blob files

SSTBlobWriter is an SST writer that can be configured to separate values into
new blob files. On close, the writer can return metadata about the written sst
and its blob files.

Author: xinhaoz

sstable/colblk_writer.go

@@ -269,6 +269,14 @@ func (w *RawColumnWriter) IsPrefixEqualPrev(k []byte) bool {
 	return w.dataBlock.KeyWriter.ComparePrev(k).PrefixEqual()
 }
 
+// PrevPointKeyKind implements the RawWriter interface.
+func (w *RawColumnWriter) PrevPointKeyKind() base.InternalKeyKind {
+	if w == nil || w.dataBlock.Rows() == 0 {
+		return base.InternalKeyKindInvalid
+	}
+	return w.prevPointKey.trailer.Kind()
+}
+
 // SetSnapshotPinnedProperties sets the properties for pinned keys. Should only
 // be used internally by Pebble.
 func (w *RawColumnWriter) SetSnapshotPinnedProperties(

sstable/rowblk_writer.go

@@ -1408,6 +1408,14 @@ func (w *RawRowWriter) IsPrefixEqualPrev(k []byte) bool {
 	return bytes.Equal(w.split.Prefix(k), w.split.Prefix(w.dataBlockBuf.dataBlock.CurUserKey()))
 }
 
+// PrevPointKeyKind implements the RawWriter interface.
+func (w *RawRowWriter) PrevPointKeyKind() base.InternalKeyKind {
+	if w == nil || w.dataBlockBuf.dataBlock.EntryCount() == 0 {
+		return base.InternalKeyKindInvalid
+	}
+	return w.dataBlockBuf.dataBlock.CurKey().Kind()
+}
+
 // EncodeSpan encodes the keys in the given span. The span can contain either
 // only RANGEDEL keys or only range keys.
 //

sstable/writer.go

@@ -363,6 +363,9 @@ type RawWriter interface {
 	//
 	// Must not be called after Writer is closed.
 	IsPrefixEqualPrev(k []byte) bool
+	// PrevPointKeyKind returns the InternalKeyKind of the last point key written
+	// to the writer. Must not be called after Writer is closed.
+	PrevPointKeyKind() base.InternalKeyKind
 
 	// SetValueSeparationProps sets the value separation props that were used when
 	// writing this sstable. This is recorded in the sstable properties.

valsep/sst_blob_writer.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 valsep
+
+import (
+	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/sstable/blob"
+	"github.com/cockroachdb/pebble/sstable/block"
+)
+
+// SSTBlobWriter writes an sstable and 0 or more blob value files
+// for ingesting into pebble. Values are extracted depending on the
+// value separation strategy defined. If ValueSeparation is
+// neverSeparateValues, this behaves like sstable.Writer and no blob
+// files are written.
+type SSTBlobWriter struct {
+	// Promote methods from sstable.Writer.
+	*sstable.Writer
+	valSep ValueSeparation
+	err    error
+
+	blobFileNum base.DiskFileNum
+	closed      bool
+	// isStrictObsolete is true if the writer is configured to write and enforce
+	// a 'strict obsolete' sstable. This includes prohibiting the addition of
+	// MERGE keys. See the documentation in format.go for more details.
+	isStrictObsolete bool
+
+	kvScratch base.InternalKV
+	// Metadata on the blob files written.
+	blobFilesMeta []blob.FileWriterStats
+}
+
+type NewBlobFile func() (objstorage.Writable, error)
+
+var neverSeparateValues = &NeverSeparateValues{}
+
+type SSTBlobWriterOptions struct {
+	SSTWriterOpts  sstable.WriterOptions
+	BlobWriterOpts blob.FileWriterOptions
+	// BlobFilesDisabled is true if value separation into blob files
+	// is disabled and the writer will behave like an sstable.Writer
+	// in this case. Note that values may still be separated into a
+	// value block in the same sstable.
+	BlobFilesDisabled bool
+	// The minimum size required for a value to be separated into a
+	// blob file. This value may be overridden by the span policy.
+	ValueSeparationMinSize int
+	// SpanPolicy specifies the specific policies applied to the table span.
+	// When using the external writer, there should be 1 span policy
+	// applied to the entire sstable.
+	SpanPolicy    base.SpanPolicy
+	NewBlobFileFn NewBlobFile
+}
+
+// NewSSTBlobWriter returns a new SSTBlobWriter that writes to the provided
+// sstHandle. The writer uses the provided options to configure both the sstable
+// writer and the blob file writer.
+func NewSSTBlobWriter(sstHandle objstorage.Writable, opts SSTBlobWriterOptions) *SSTBlobWriter {
+	writer := &SSTBlobWriter{
+		isStrictObsolete: opts.SSTWriterOpts.IsStrictObsolete,
+	}
+
+	if opts.SpanPolicy.PreferFastCompression && opts.SSTWriterOpts.Compression != block.NoCompression {
+		opts.SSTWriterOpts.Compression = block.FastestCompression
+	}
+
+	writer.Writer = sstable.NewWriter(sstHandle, opts.SSTWriterOpts)
+
+	// Create the value separator.
+	minimumValueSize := opts.ValueSeparationMinSize
+	if opts.SpanPolicy.ValueStoragePolicy.OverrideBlobSeparationMinimumSize > 0 {
+		minimumValueSize = opts.SpanPolicy.ValueStoragePolicy.OverrideBlobSeparationMinimumSize
+	}
+	if opts.BlobFilesDisabled || minimumValueSize == 0 || opts.SpanPolicy.ValueStoragePolicy.DisableBlobSeparation {
+		writer.valSep = neverSeparateValues
+	} else {
+		newBlobObject := func() (objstorage.Writable, objstorage.ObjectMetadata, error) {
+			// The ObjectMetadata collected by the value separator will not be
+			// exposed by this writer, since this store does not yet know about
+			// these objects. However, we must provide a unique file number for
+			// each new blob file because the value separator uses file ids to
+			// retrieve the index of the blob file within the sst's tracked blob
+			// references array. The reference id (array index) is what is then
+			// written to the inline blob handle.
+			newHandle, err := opts.NewBlobFileFn()
+			if err != nil {
+				return nil, objstorage.ObjectMetadata{}, err
+			}
+			nextFileNum := writer.blobFileNum
+			writer.blobFileNum++
+			return newHandle, objstorage.ObjectMetadata{DiskFileNum: nextFileNum}, nil
+		}
+		writer.valSep = NewWriteNewBlobFiles(
+			opts.SSTWriterOpts.Comparer,
+			newBlobObject,
+			opts.BlobWriterOpts,
+			minimumValueSize,
+			WriteNewBlobFilesOptions{
+				DisableValueSeparationBySuffix: opts.SpanPolicy.ValueStoragePolicy.DisableSeparationBySuffix,
+				ShortAttrExtractor:             opts.SSTWriterOpts.ShortAttributeExtractor,
+				InvalidValueCallback: func(userKey []byte, value []byte, err error) {
+					writer.err = errors.CombineErrors(writer.err, err)
+				},
+			},
+		)
+	}
+
+	return writer
+}
+
+// Error returns the current accumulated error if any.
+func (w *SSTBlobWriter) Error() error {
+	return errors.CombineErrors(w.err, w.Writer.Error())
+}
+
+// Set sets the value for the given key. The sequence number is set to 0.
+// Values may be separated into blob files depending on the value separation
+// strategy configured for the writer. Intended for use to externally construct
+// an sstable with its blob files before ingestion into a DB. For a given
+// SSTBlobWriter, the keys passed to Set must be in strictly increasing order.
+func (w *SSTBlobWriter) Set(key, value []byte) error {
+	if err := w.Error(); err != nil {
+		return err
+	}
+
+	if w.isStrictObsolete {
+		return errors.Errorf("use raw writer Add in strict obsolete mode")
+	}
+
+	w.kvScratch.K = base.MakeInternalKey(key, 0, sstable.InternalKeyKindSet)
+	w.kvScratch.V = base.MakeInPlaceValue(value)
+	isLikelyMVCCGarbage := sstable.IsLikelyMVCCGarbage(
+		key, w.Raw().PrevPointKeyKind(), sstable.InternalKeyKindSet, len(value), w.Writer.Raw().IsPrefixEqualPrev)
+	return w.valSep.Add(w.Raw(), &w.kvScratch, false, isLikelyMVCCGarbage)
+}
+
+// BlobWriterMetas returns an array of blob.FileWriterStats describing the
+// blob files written by this SSTBlobWriter. The ordering of the returned
+// slice matches the ordering of blob files as they should appear in the
+// sstable's manifest.BlobReferences. Close must be called before calling this
+// method.
+func (w *SSTBlobWriter) BlobWriterMetas() ([]blob.FileWriterStats, error) {
+	if !w.closed {
+		return nil, errors.New("blob writer not closed")
+	}
+	return w.blobFilesMeta, nil
+}
+
+// Close closes both the sstable writer and the blob file writer if any.
+func (w *SSTBlobWriter) Close() error {
+	w.err = errors.CombineErrors(w.err, w.Writer.Close())
+	meta, err := w.valSep.FinishOutput()
+	if err != nil {
+		w.err = errors.CombineErrors(w.err, err)
+	}
+	for _, blobFile := range meta.NewBlobFiles {
+		w.blobFilesMeta = append(w.blobFilesMeta, blobFile.FileStats)
+	}
+	w.closed = true
+	return w.err
+}

valsep/sst_blob_writer_test.go

@@ -0,0 +1,184 @@
+// 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 valsep
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/cockroachdb/crlib/testutils/leaktest"
+	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/sstable"
+	"github.com/cockroachdb/pebble/vfs"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSSTBlobWriter(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	runDataDriven(t, "testdata/sst_blob_writer")
+}
+
+// The span policy string is in the form "(<option1>=<val>,<option2>=<val>...)"
+func parseSpanPolicy(t *testing.T, spanPolicyStr string) base.SpanPolicy {
+	spanPolicyStr = strings.TrimPrefix(spanPolicyStr, "(")
+	spanPolicyStr = strings.TrimSuffix(spanPolicyStr, ")")
+	spanPolicyParts := strings.Split(spanPolicyStr, ",")
+	var policy base.ValueStoragePolicyAdjustment
+	var err error
+	for _, part := range spanPolicyParts {
+		fieldParts := strings.Split(part, "=")
+		switch fieldParts[0] {
+		case "no-value-separation":
+			policy.DisableBlobSeparation = true
+		case "value-separation-min-size":
+			policy.OverrideBlobSeparationMinimumSize, err = strconv.Atoi(fieldParts[1])
+			if err != nil {
+				t.Fatalf("parsing value-separation-min-size: %v", err)
+			}
+		case "disable-value-separation-by-suffix":
+			policy.DisableSeparationBySuffix = true
+		default:
+			t.Fatalf("unrecognized span policy option: %s", fieldParts[0])
+		}
+	}
+
+	return base.SpanPolicy{
+		ValueStoragePolicy: policy,
+	}
+}
+
+func parseBuildSSTBlobWriterOptions(t *testing.T, td *datadriven.TestData) SSTBlobWriterOptions {
+	opts := SSTBlobWriterOptions{}
+	td.MaybeScanArgs(t, "value-separation-min-size", &opts.ValueSeparationMinSize)
+
+	var spanPolicyStr string
+	td.MaybeScanArgs(t, "span-policy", &spanPolicyStr)
+	if spanPolicyStr != "" {
+		opts.SpanPolicy = parseSpanPolicy(t, spanPolicyStr)
+	}
+	return opts
+}
+
+func runDataDriven(t *testing.T, file string) {
+	datadriven.RunTest(t, file, func(t *testing.T, td *datadriven.TestData) string {
+		ctx := context.Background()
+		switch td.Cmd {
+		case "build":
+			var buf bytes.Buffer
+			fs := vfs.WithLogging(vfs.NewMem(), func(format string, args ...any) {
+				fmt.Fprint(&buf, "# ")
+				fmt.Fprintf(&buf, format, args...)
+				fmt.Fprintln(&buf)
+			})
+			objStore, err := objstorageprovider.Open(objstorageprovider.Settings{
+				FS: fs,
+			})
+			require.NoError(t, err)
+			blobFileCount := 0
+			opts := parseBuildSSTBlobWriterOptions(t, td)
+			opts.SSTWriterOpts.Comparer = testkeys.Comparer
+			opts.SSTWriterOpts.TableFormat = sstable.TableFormatPebblev7
+			opts.NewBlobFileFn = func() (objstorage.Writable, error) {
+				fnum := blobFileCount
+				w, _, err := objStore.Create(ctx, base.FileTypeBlob, base.DiskFileNum(fnum), objstorage.CreateOptions{})
+				if err != nil {
+					return nil, err
+				}
+				blobFileCount++
+				return w, err
+			}
+			sstHandle, _, err := objStore.Create(ctx, base.FileTypeTable, 0, objstorage.CreateOptions{})
+			require.NoError(t, err)
+			writer := NewSSTBlobWriter(sstHandle, opts)
+			defer func() {
+				if !writer.closed {
+					_ = writer.Close()
+				}
+			}()
+			kvs, err := sstable.ParseTestKVsAndSpans(td.Input, nil)
+			if err != nil {
+				return fmt.Sprintf("error parsing input: %v", err)
+			}
+
+			for _, kv := range kvs {
+				keyKind := kv.Key.Kind()
+				if kv.IsKeySpan() {
+					keyKind = kv.Span.Keys[0].Kind()
+				}
+				switch keyKind {
+				case sstable.InternalKeyKindSet, sstable.InternalKeyKindSetWithDelete:
+					if err := writer.Set(kv.Key.UserKey, kv.Value); err != nil {
+						return fmt.Sprintf("error putting key %s: %v", kv.Key.UserKey, err)
+					}
+				case sstable.InternalKeyKindDelete, sstable.InternalKeyKindDeleteSized:
+					if err := writer.Delete(kv.Key.UserKey); err != nil {
+						return fmt.Sprintf("error deleting key %s: %v", kv.Key.UserKey, err)
+					}
+				case sstable.InternalKeyKindRangeDelete:
+					if err := writer.RangeKeyDelete(kv.Span.Start, kv.Span.End); err != nil {
+						return fmt.Sprintf("error deleting range %s-%s: %v", kv.Span.Start, kv.Span.End, err)
+					}
+				case sstable.InternalKeyKindMerge:
+					if err := writer.Merge(kv.Key.UserKey, kv.Value); err != nil {
+						return fmt.Sprintf("error merging key %s: %v", kv.Key.UserKey, err)
+					}
+				case base.InternalKeyKindRangeKeySet:
+					if err := writer.RangeKeySet(kv.Span.Start, kv.Span.End, nil, kv.Value); err != nil {
+						return fmt.Sprintf("error setting range key %s-%s: %v", kv.Span.Start, kv.Span.End, err)
+					}
+				case base.InternalKeyKindRangeKeyUnset:
+					if err := writer.RangeKeyUnset(kv.Span.Start, kv.Span.End, kv.Key.UserKey); err != nil {
+						return fmt.Sprintf("error unsetting range key %s-%s: %v", kv.Span.Start, kv.Span.End, err)
+					}
+				case base.InternalKeyKindRangeKeyDelete:
+					if err := writer.RangeKeyDelete(kv.Span.Start, kv.Span.End); err != nil {
+						return fmt.Sprintf("error deleting range key %s-%s: %v", kv.Span.Start, kv.Span.End, err)
+					}
+				default:
+					return fmt.Sprintf("unsupported key kind %v", kv.Key.Kind())
+				}
+			}
+
+			if err := writer.Close(); err != nil {
+				return fmt.Sprintf("error closing writer: %v", err)
+			}
+
+			tableMeta, err := writer.Metadata()
+			if err != nil {
+				return fmt.Sprintf("error getting metadata: %v", err)
+			}
+
+			blobMetas, err := writer.BlobWriterMetas()
+			if err != nil {
+				return fmt.Sprintf("error getting blob metas: %v", err)
+			}
+
+			var outputBuf bytes.Buffer
+			// Print some sst properties.
+			fmt.Fprintf(&outputBuf, "size:%d\n", tableMeta.Size)
+			outputBuf.WriteString("blobfiles:")
+			require.Equal(t, blobFileCount, len(blobMetas))
+			if len(blobMetas) > 0 {
+				outputBuf.WriteString("\n")
+				for i, bm := range blobMetas {
+					fmt.Fprintf(&outputBuf, "%d: %s\n", i+1, bm.String())
+				}
+			} else {
+				outputBuf.WriteString(" none\n")
+			}
+			return outputBuf.String()
+		default:
+			return fmt.Sprintf("unrecognized command %s", td.Cmd)
+		}
+	})
+}