sql: add SST writer to distributed index backfill merge pipeline

This change continues the integration of the distributed merge
pipeline into the index backfill flow. When enabled, the backfiller
writes its KV output batches to SSTs (backed by ExternalStorage) via a
new SST sink.

- Processor: Updated rowexec backfiller to route KV batches
through the SST sink.
- Job progress: Persist SST manifest metadata in job progress via
IndexBackfillMapProgress.

Informs: #158378
Epic: CRDB-48845
Release note: none

Author: spilchen

pkg/jobs/jobspb/jobs.proto

@@ -843,6 +843,19 @@ message BackfillProgress {
   // backfilled into the destination indexes. Note that this will never contain
   // tenant prefixes even if the data corresponds to a secondary tenant.
   repeated roachpb.Span completed_spans = 5 [(gogoproto.nullable) = false];
+
+  // SSTManifests contains metadata about SST files produced while backfilling
+  // via the distributed merge pipeline. Each entry corresponds to a flushed SST.
+  repeated IndexBackfillSSTManifest sst_manifests = 6 [(gogoproto.nullable) = false, (gogoproto.customname) = "SSTManifests"];
+}
+
+// IndexBackfillSSTManifest describes an SST produced by the backfill map stage.
+message IndexBackfillSSTManifest {
+  string uri = 1 [(gogoproto.customname) = "URI"];
+  roachpb.Span span = 2;
+  uint64 file_size = 3;
+  bytes row_sample = 4 [(gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/roachpb.Key"];
+  util.hlc.Timestamp write_timestamp = 5;
 }
 
 // MergeProgress is used to track index merge progress in the declarative

pkg/sql/backfill.go

@@ -1044,7 +1044,7 @@ func (sc *SchemaChanger) distIndexBackfill(
 		indexBatchSize := indexBackfillBatchSize.Get(&sc.execCfg.Settings.SV)
 		chunkSize := sc.getChunkSize(indexBatchSize)
 		spec := initIndexBackfillerSpec(*tableDesc.TableDesc(), writeAsOf, writeAtRequestTimestamp, chunkSize, addedIndexes, 0)
-		if err := maybeEnableDistributedMergeIndexBackfill(ctx, sc.execCfg.Settings, &spec); err != nil {
+		if err := maybeEnableDistributedMergeIndexBackfill(ctx, sc.execCfg.Settings, sc.execCfg.NodeInfo.NodeID.SQLInstanceID(), &spec); err != nil {
 			return err
 		}
 		p, err = sc.distSQLPlanner.createBackfillerPhysicalPlan(ctx, planCtx, spec, todoSpans)

pkg/sql/bulksst/sst_file_allocator.go

@@ -53,10 +53,11 @@ func (f *fileAllocatorBase) addFile(
 	uri string, span roachpb.Span, rowSample roachpb.Key, fileSize uint64,
 ) {
 	f.fileInfo.SST = append(f.fileInfo.SST, &SSTFileInfo{
-		URI:      uri,
-		StartKey: span.Key,
-		EndKey:   span.EndKey,
-		FileSize: fileSize,
+		URI:       uri,
+		StartKey:  span.Key,
+		EndKey:    span.EndKey,
+		FileSize:  fileSize,
+		RowSample: rowSample,
 	})
 	f.fileInfo.TotalSize += fileSize
 	f.recordRowSample(rowSample)

pkg/sql/bulksst/sst_info.proto

@@ -14,6 +14,7 @@ message SSTFileInfo {
     required bytes start_key = 2 [(gogoproto.nullable) = false, (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/roachpb.Key"];
     required bytes end_key = 3 [(gogoproto.nullable) = false, (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/roachpb.Key"];
     required uint64 file_size = 4 [(gogoproto.nullable) = false];
+    optional bytes row_sample = 5 [(gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/roachpb.Key"];
 }
 
 message SSTFiles {

pkg/sql/bulksst/sst_writer.go

@@ -38,7 +38,7 @@ var _ kvserverbase.BulkAdder = &Writer{}
 
 var BatchSize = settings.RegisterByteSizeSetting(settings.ApplicationLevel,
 	"bulkio.sst_writer.batch_size",
-	"Writer in memory batch size",
+	"bytes to buffer in-memory per SST before the bulk writer flushes",
 	128*1024*1024)
 
 // NewUnsortedSSTBatcher creates a new SST batcher, a file allocator must be

pkg/sql/distsql_plan_backfill.go

@@ -7,9 +7,11 @@ package sql
 
 import (
 	"context"
+	"fmt"
 	"time"
 	"unsafe"
 
+	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
@@ -92,7 +94,10 @@ var distributedMergeIndexBackfillEnabled = settings.RegisterBoolSetting(
 )
 
 func maybeEnableDistributedMergeIndexBackfill(
-	ctx context.Context, st *cluster.Settings, spec *execinfrapb.BackfillerSpec,
+	ctx context.Context,
+	st *cluster.Settings,
+	nodeID base.SQLInstanceID,
+	spec *execinfrapb.BackfillerSpec,
 ) error {
 	if !distributedMergeIndexBackfillEnabled.Get(&st.SV) {
 		return nil
@@ -101,6 +106,7 @@ func maybeEnableDistributedMergeIndexBackfill(
 		return pgerror.New(pgcode.FeatureNotSupported, "distributed merge requires cluster version 26.1")
 	}
 	spec.UseDistributedMergeSink = true
+	spec.DistributedMergeFilePrefix = fmt.Sprintf("nodelocal://%d/index-backfill", nodeID)
 	return nil
 }
 

pkg/sql/execinfrapb/processors_bulk_io.proto

@@ -87,7 +87,11 @@ message BackfillerSpec {
 
   optional bool use_distributed_merge_sink = 17 [(gogoproto.nullable) = false];
 
-  // NEXTID: 18.
+  // distributed_merge_file_prefix is the ExternalStorage URI prefix to use when
+  // writing map outputs for the distributed merge pipeline.
+  optional string distributed_merge_file_prefix = 18 [(gogoproto.nullable) = false];
+
+  // NEXTID: 19.
 }
 
 // JobProgress identifies the job to report progress on. This reporting
@@ -609,3 +613,9 @@ message BulkMergeSpec {
 
 	// NEXT ID: 4.
 }
+
+// IndexBackfillMapProgress is emitted by map-stage processors to describe newly
+// produced SST manifests.
+message IndexBackfillMapProgress {
+  repeated jobs.jobspb.IndexBackfillSSTManifest sst_manifests = 1 [(gogoproto.nullable) = false, (gogoproto.customname) = "SSTManifests"];
+}

pkg/sql/index_backfiller.go

@@ -9,6 +9,7 @@ import (
 	"context"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -25,6 +26,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/errors"
+	gogotypes "github.com/gogo/protobuf/types"
 )
 
 // IndexBackfillPlanner holds dependencies for an index backfiller
@@ -33,6 +35,39 @@ type IndexBackfillPlanner struct {
 	execCfg *ExecutorConfig
 }
 
+type sstManifestBuffer struct {
+	syncutil.Mutex
+	manifests []jobspb.IndexBackfillSSTManifest
+}
+
+func newSSTManifestBuffer(initial []jobspb.IndexBackfillSSTManifest) *sstManifestBuffer {
+	buf := &sstManifestBuffer{}
+	buf.manifests = append(buf.manifests, initial...)
+	return buf
+}
+
+func (b *sstManifestBuffer) snapshotLocked() []jobspb.IndexBackfillSSTManifest {
+	return append([]jobspb.IndexBackfillSSTManifest(nil), b.manifests...)
+}
+
+func (b *sstManifestBuffer) Snapshot() []jobspb.IndexBackfillSSTManifest {
+	b.Lock()
+	defer b.Unlock()
+	return b.snapshotLocked()
+}
+
+func (b *sstManifestBuffer) Append(
+	newManifests []jobspb.IndexBackfillSSTManifest,
+) []jobspb.IndexBackfillSSTManifest {
+	if len(newManifests) == 0 {
+		return b.Snapshot()
+	}
+	b.Lock()
+	defer b.Unlock()
+	b.manifests = append(b.manifests, newManifests...)
+	return b.snapshotLocked()
+}
+
 // NewIndexBackfiller creates a new IndexBackfillPlanner.
 func NewIndexBackfiller(execCfg *ExecutorConfig) *IndexBackfillPlanner {
 	return &IndexBackfillPlanner{execCfg: execCfg}
@@ -81,13 +116,25 @@ func (ib *IndexBackfillPlanner) BackfillIndexes(
 	}
 	// Add spans that were already completed before the job resumed.
 	addCompleted(progress.CompletedSpans...)
+	sstManifestBuf := newSSTManifestBuffer(progress.SSTManifests)
+	progress.SSTManifests = sstManifestBuf.Snapshot()
+	updateSSTManifests := func(newManifests []jobspb.IndexBackfillSSTManifest) {
+		progress.SSTManifests = sstManifestBuf.Append(newManifests)
+	}
 	updateFunc := func(
 		ctx context.Context, meta *execinfrapb.ProducerMetadata,
 	) error {
 		if meta.BulkProcessorProgress == nil {
 			return nil
 		}
 		progress.CompletedSpans = addCompleted(meta.BulkProcessorProgress.CompletedSpans...)
+		var mapProgress execinfrapb.IndexBackfillMapProgress
+		if gogotypes.Is(&meta.BulkProcessorProgress.ProgressDetails, &mapProgress) {
+			if err := gogotypes.UnmarshalAny(&meta.BulkProcessorProgress.ProgressDetails, &mapProgress); err != nil {
+				return err
+			}
+			updateSSTManifests(mapProgress.SSTManifests)
+		}
 		// Make sure the progress update does not contain overlapping spans.
 		// This is a sanity check that only runs in test configurations, since it
 		// is an expensive n^2 check.
@@ -211,7 +258,7 @@ func (ib *IndexBackfillPlanner) plan(
 			*td.TableDesc(), writeAsOf, writeAtRequestTimestamp, chunkSize,
 			indexesToBackfill, sourceIndexID,
 		)
-		if err := maybeEnableDistributedMergeIndexBackfill(ctx, ib.execCfg.Settings, &spec); err != nil {
+		if err := maybeEnableDistributedMergeIndexBackfill(ctx, ib.execCfg.Settings, ib.execCfg.NodeInfo.NodeID.SQLInstanceID(), &spec); err != nil {
 			return err
 		}
 		var err error

pkg/sql/logictest/testdata/logic_test/create_index

@@ -714,11 +714,18 @@ subtest distributed_merge_index_backfill_placeholder
 statement ok
 CREATE TABLE dist_merge_idx (a INT PRIMARY KEY, b INT)
 
+statement ok
+INSERT INTO dist_merge_idx VALUES (1,1), (2,2), (3,3)
+
 statement ok
 SET CLUSTER SETTING bulkio.index_backfill.distributed_merge.enabled = true
 
+# TODO(158378): The end-to-end flow for create index using distributed
+# merge isn't implemented yet, so we get to the validation step and notice
+# we didn't ingest any rows. This is expected until the entire flow is
+# implemented.
 skipif config local-mixed-25.3 local-mixed-25.4
-statement error pq: .*index backfill distributed merge sink is not implemented yet.*
+statement error pgcode XX000 pq: .*validation of non-unique index dist_merge_idx_idx failed: expected 3 rows, found 0.*
 CREATE INDEX dist_merge_idx_idx ON dist_merge_idx (b)
 
 onlyif config local-mixed-25.3 local-mixed-25.4

pkg/sql/rowexec/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "hashgroupjoiner.go",
         "hashjoiner.go",
         "indexbackfiller.go",
+        "indexbackfiller_sst_sink.go",
         "inverted_expr_evaluator.go",
         "inverted_filterer.go",
         "inverted_joiner.go",
@@ -42,6 +43,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/rowexec",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/cloud",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
@@ -51,9 +53,11 @@ go_library(
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/roachpb",
+        "//pkg/security/username",
         "//pkg/server/telemetry",
         "//pkg/settings",
         "//pkg/sql/backfill",
+        "//pkg/sql/bulksst",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catenumpb",
         "//pkg/sql/catalog/colinfo",
@@ -102,7 +106,6 @@ go_library(
         "//pkg/util/collatedstring",
         "//pkg/util/ctxgroup",
         "//pkg/util/encoding",
-        "//pkg/util/errorutil/unimplemented",
         "//pkg/util/hlc",
         "//pkg/util/intsets",
         "//pkg/util/log",
@@ -123,6 +126,7 @@ go_library(
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_redact//:redact",
+        "@com_github_gogo_protobuf//types",
     ],
 )
 
@@ -135,6 +139,7 @@ go_test(
         "distinct_test.go",
         "filterer_test.go",
         "hashjoiner_test.go",
+        "indexbackfiller_sst_sink_test.go",
         "indexbackfiller_test.go",
         "inverted_expr_evaluator_test.go",
         "inverted_filterer_test.go",
@@ -166,6 +171,9 @@ go_test(
     }),
     deps = [
         "//pkg/base",
+        "//pkg/cloud",
+        "//pkg/cloud/cloudpb",
+        "//pkg/cloud/nodelocal",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
@@ -184,6 +192,7 @@ go_test(
         "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/backfill",
+        "//pkg/sql/bulksst",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catalogkeys",
         "//pkg/sql/catalog/catenumpb",
@@ -237,6 +246,8 @@ go_test(
         "//pkg/util/tracing/tracingpb",
         "@com_github_axiomhq_hyperloglog//:hyperloglog",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_pebble//objstorage",
+        "@com_github_cockroachdb_pebble//objstorage/objstorageprovider",
         "@com_github_gogo_protobuf//types",
         "@com_github_jackc_pgx_v5//:pgx",
         "@com_github_stretchr_testify//require",