Merge #158449

158449: sql: add abstraction to index backfill for distributed merge r=spilchen a=spilchen

This change introduces an abstraction layer in the index backfill infrastructure, enabling selection between the legacy bulk adder and a planned distributed merge mechanism.

Currently, choosing the distributed merge mode results in an unimplemented error. However, supporting plumbing has been added, including:
- A cluster setting override
- Updated schema changer and new planner logic to persist the selected mode in BackfillerSpec
- A new UseDistributedMergeSink flag in the processor, which inspects this mode at runtime

Informs: #158378
Epic: CRDB-48845

Release note: None

Co-authored-by: Matt Spilchen <matt.spilchen@cockroachlabs.com>

Author: craig[bot]

pkg/sql/backfill.go

@@ -1044,6 +1044,9 @@ 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 {
+			return err
+		}
 		p, err = sc.distSQLPlanner.createBackfillerPhysicalPlan(ctx, planCtx, spec, todoSpans)
 		return err
 	}); err != nil {

pkg/sql/distsql_plan_backfill.go

@@ -10,11 +10,15 @@ import (
 	"time"
 	"unsafe"
 
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -80,6 +84,26 @@ var initialSplitsPerProcessor = settings.RegisterIntSetting(
 	settings.NonNegativeInt,
 )
 
+var distributedMergeIndexBackfillEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"bulkio.index_backfill.distributed_merge.enabled",
+	"enable the distributed merge pipeline for index backfills",
+	false,
+)
+
+func maybeEnableDistributedMergeIndexBackfill(
+	ctx context.Context, st *cluster.Settings, spec *execinfrapb.BackfillerSpec,
+) error {
+	if !distributedMergeIndexBackfillEnabled.Get(&st.SV) {
+		return nil
+	}
+	if !st.Version.IsActive(ctx, clusterversion.V26_1) {
+		return pgerror.New(pgcode.FeatureNotSupported, "distributed merge requires cluster version 26.1")
+	}
+	spec.UseDistributedMergeSink = true
+	return nil
+}
+
 // createBackfiller generates a plan consisting of index/column backfiller
 // processors, one for each node that has spans that we are reading. The plan is
 // finalized.

pkg/sql/execinfrapb/processors_bulk_io.proto

@@ -85,7 +85,9 @@ message BackfillerSpec {
   // the value of 0 is specified then the primary index will be used implicitly.
   optional uint32 source_index_id = 16 [(gogoproto.nullable) = false, (gogoproto.customname) = "SourceIndexID", (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.IndexID"];
 
-  // NEXTID: 17.
+  optional bool use_distributed_merge_sink = 17 [(gogoproto.nullable) = false];
+
+  // NEXTID: 18.
 }
 
 // JobProgress identifies the job to report progress on. This reporting

pkg/sql/index_backfiller.go

@@ -211,6 +211,9 @@ func (ib *IndexBackfillPlanner) plan(
 			*td.TableDesc(), writeAsOf, writeAtRequestTimestamp, chunkSize,
 			indexesToBackfill, sourceIndexID,
 		)
+		if err := maybeEnableDistributedMergeIndexBackfill(ctx, ib.execCfg.Settings, &spec); err != nil {
+			return err
+		}
 		var err error
 		p, err = ib.execCfg.DistSQLPlanner.createBackfillerPhysicalPlan(ctx, planCtx, spec, sourceSpans)
 		return err

pkg/sql/logictest/testdata/logic_test/create_index

@@ -705,3 +705,30 @@ statement error pgcode 42P07 index with name \"t_index_name_conflicts_with_prima
 CREATE INDEX t_index_name_conflicts_with_primary_key_pkey ON t_index_name_conflicts_with_primary_key (a);
 
 subtest end
+
+# Temporary check to ensure the distributed merge setting is wired up. When we
+# implement distributed merge, we probably want to test this by reusing as many
+# existing tests as possible rather than strictly writing new ones.
+subtest distributed_merge_index_backfill_placeholder
+
+statement ok
+CREATE TABLE dist_merge_idx (a INT PRIMARY KEY, b INT)
+
+statement ok
+SET CLUSTER SETTING bulkio.index_backfill.distributed_merge.enabled = true
+
+skipif config local-mixed-25.3 local-mixed-25.4
+statement error pq: .*index backfill distributed merge sink is not implemented yet.*
+CREATE INDEX dist_merge_idx_idx ON dist_merge_idx (b)
+
+onlyif config local-mixed-25.3 local-mixed-25.4
+statement error pq: .*distributed merge requires cluster version 26.*
+CREATE INDEX dist_merge_idx_idx ON dist_merge_idx (b)
+
+statement ok
+SET CLUSTER SETTING bulkio.index_backfill.distributed_merge.enabled = false
+
+statement ok
+DROP TABLE dist_merge_idx
+
+subtest end

pkg/sql/rowexec/BUILD.bazel

@@ -102,6 +102,7 @@ go_library(
         "//pkg/util/collatedstring",
         "//pkg/util/ctxgroup",
         "//pkg/util/encoding",
+        "//pkg/util/errorutil/unimplemented",
         "//pkg/util/hlc",
         "//pkg/util/intsets",
         "//pkg/util/log",
@@ -173,6 +174,7 @@ go_test(
         "//pkg/kv/kvclient/rangecache",
         "//pkg/kv/kvpb",
         "//pkg/kv/kvserver",
+        "//pkg/kv/kvserver/kvserverbase",
         "//pkg/multitenant/tenantcapabilitiespb",
         "//pkg/roachpb",
         "//pkg/security/securityassets",
@@ -190,6 +192,7 @@ go_test(
         "//pkg/sql/catalog/descs",
         "//pkg/sql/catalog/desctestutils",
         "//pkg/sql/catalog/fetchpb",
+        "//pkg/sql/catalog/tabledesc",
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/execversion",

pkg/sql/rowexec/indexbackfiller.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
+	"github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
@@ -48,6 +49,8 @@ type indexBackfiller struct {
 	processorID int32
 
 	filter backfill.MutationFilter
+
+	bulkAdderFactory indexBackfillBulkAdderFactory
 }
 
 var _ execinfra.Processor = &indexBackfiller{}
@@ -86,6 +89,40 @@ var indexBackfillElasticCPUControlEnabled = settings.RegisterBoolSetting(
 	false, // TODO(dt): enable this by default after more benchmarking.
 )
 
+// indexBackfillSink abstracts the destination for index backfill output so the
+// ingestion pipeline can route built KVs either to the legacy BulkAdder path or
+// to future sinks (e.g. distributed-merge SST writers) without rewriting the
+// DistSQL processor. All sinks share the same Add/Flush/progress contract.
+type indexBackfillSink interface {
+	// Add enqueues a single KV pair for eventual persistence in the sink-specific
+	// backing store.
+	Add(ctx context.Context, key roachpb.Key, value []byte) error
+	// Flush forces any buffered state to be persisted.
+	Flush(ctx context.Context) error
+	// Close releases resources owned by the sink. Implementations should be
+	// idempotent and safe to call even if Flush returns an error.
+	Close(ctx context.Context)
+	// SetOnFlush installs a callback that is invoked after the sink writes a
+	// batch (mirrors kvserverbase.BulkAdder semantics so existing progress
+	// plumbing can be reused).
+	SetOnFlush(func(summary kvpb.BulkOpSummary))
+}
+
+// indexBackfillBulkAdderFactory mirrors kvserverbase.BulkAdderFactory but is
+// injected so tests can swap in fakes and future sinks can reuse the backfiller
+// without referencing execinfra.ServerConfig directly.
+type indexBackfillBulkAdderFactory func(
+	ctx context.Context, writeAsOf hlc.Timestamp, opts kvserverbase.BulkAdderOptions,
+) (kvserverbase.BulkAdder, error)
+
+// bulkAdderIndexBackfillSink is the default sink implementation backed by
+// kvserverbase.BulkAdder.
+type bulkAdderIndexBackfillSink struct {
+	kvserverbase.BulkAdder
+}
+
+var _ indexBackfillSink = (*bulkAdderIndexBackfillSink)(nil)
+
 func newIndexBackfiller(
 	ctx context.Context,
 	flowCtx *execinfra.FlowCtx,
@@ -101,6 +138,11 @@ func newIndexBackfiller(
 		flowCtx:     flowCtx,
 		processorID: processorID,
 		filter:      backfill.IndexMutationFilter,
+		bulkAdderFactory: func(
+			ctx context.Context, writeAsOf hlc.Timestamp, opts kvserverbase.BulkAdderOptions,
+		) (kvserverbase.BulkAdder, error) {
+			return flowCtx.Cfg.BulkAdder(ctx, flowCtx.Cfg.DB.KV(), writeAsOf, opts)
+		},
 	}
 
 	if err := ib.IndexBackfiller.InitForDistributedUse(
@@ -257,15 +299,16 @@ func (ib *indexBackfiller) maybeReencodeAndWriteVectorIndexEntry(
 	return true, nil
 }
 
-// ingestIndexEntries adds the batches of built index entries to the buffering
-// adder and reports progress back to the coordinator node.
-func (ib *indexBackfiller) ingestIndexEntries(
-	ctx context.Context,
-	indexEntryCh <-chan indexEntryBatch,
-	progCh chan execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
-) error {
-	ctx, span := tracing.ChildSpan(ctx, "ingestIndexEntries")
-	defer span.Finish()
+// makeIndexBackfillSink materializes whatever sink the current backfill should
+// use (legacy BulkAdder or distributed-merge sink). The choice is driven by
+// execinfrapb.BackfillerSpec.
+func (ib *indexBackfiller) makeIndexBackfillSink(ctx context.Context) (indexBackfillSink, error) {
+	if ib.spec.UseDistributedMergeSink {
+		return nil, unimplemented.New(
+			"distributed merge index backfill sink",
+			"index backfill distributed merge sink is not implemented yet",
+		)
+	}
 
 	minBufferSize := backfillerBufferSize.Get(&ib.flowCtx.Cfg.Settings.SV)
 	maxBufferSize := func() int64 { return backfillerMaxBufferSize.Get(&ib.flowCtx.Cfg.Settings.SV) }
@@ -278,11 +321,33 @@ func (ib *indexBackfiller) ingestIndexEntries(
 		InitialSplitsIfUnordered: int(ib.spec.InitialSplits),
 		WriteAtBatchTimestamp:    ib.spec.WriteAtBatchTimestamp,
 	}
-	adder, err := ib.flowCtx.Cfg.BulkAdder(ctx, ib.flowCtx.Cfg.DB.KV(), ib.spec.WriteAsOf, opts)
+
+	adderFactory := ib.bulkAdderFactory
+	if adderFactory == nil {
+		return nil, errors.AssertionFailedf("index backfiller bulk adder factory must be configured")
+	}
+	adder, err := adderFactory(ctx, ib.spec.WriteAsOf, opts)
+	if err != nil {
+		return nil, err
+	}
+	return &bulkAdderIndexBackfillSink{BulkAdder: adder}, nil
+}
+
+// ingestIndexEntries adds the batches of built index entries to the buffering
+// adder and reports progress back to the coordinator node.
+func (ib *indexBackfiller) ingestIndexEntries(
+	ctx context.Context,
+	indexEntryCh <-chan indexEntryBatch,
+	progCh chan execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
+) error {
+	ctx, span := tracing.ChildSpan(ctx, "ingestIndexEntries")
+	defer span.Finish()
+
+	sink, err := ib.makeIndexBackfillSink(ctx)
 	if err != nil {
 		return err
 	}
-	defer adder.Close(ctx)
+	defer sink.Close(ctx)
 
 	// Synchronizes read and write access on completedSpans which is updated on a
 	// BulkAdder flush, but is read when progress is being sent back to the
@@ -302,7 +367,7 @@ func (ib *indexBackfiller) ingestIndexEntries(
 		mu.completedSpans = append(mu.completedSpans, mu.addedSpans...)
 		mu.addedSpans = nil
 	}
-	adder.SetOnFlush(flushAddedSpans)
+	sink.SetOnFlush(flushAddedSpans)
 
 	pushProgress := func() {
 		mu.Lock()
@@ -365,7 +430,7 @@ func (ib *indexBackfiller) ingestIndexEntries(
 					}
 				}
 
-				if err := adder.Add(ctx, indexEntry.Key, indexEntry.Value.RawBytes); err != nil {
+				if err := sink.Add(ctx, indexEntry.Key, indexEntry.Value.RawBytes); err != nil {
 					return ib.wrapDupError(ctx, err)
 				}
 			}
@@ -391,7 +456,7 @@ func (ib *indexBackfiller) ingestIndexEntries(
 
 			knobs := &ib.flowCtx.Cfg.TestingKnobs
 			if knobs.BulkAdderFlushesEveryBatch {
-				if err := adder.Flush(ctx); err != nil {
+				if err := sink.Flush(ctx); err != nil {
 					return ib.wrapDupError(ctx, err)
 				}
 				pushProgress()
@@ -412,7 +477,7 @@ func (ib *indexBackfiller) ingestIndexEntries(
 	if err := g.Wait(); err != nil {
 		return err
 	}
-	if err := adder.Flush(ctx); err != nil {
+	if err := sink.Flush(ctx); err != nil {
 		return ib.wrapDupError(ctx, err)
 	}