Merge #158508

158508: sql/backfill: persist distributed merge mode in job details r=spilchen a=spilchen

Previously, the choice of merge strategy during index backfills was controlled by a global cluster setting. This boolean value was evaluated at execution time, meaning the behavior could change mid-job if the setting changed.

This commit addresses that by persisting the distributed merge mode into the job details. The setting is now read only once (when the job is created) and saved in the job payload. This change applies to both legacy and declarative schema changes.

Additionally, the cluster setting has been converted from a boolean to a multi value enum. This allows us to enable this for only legacy or only declarative schema changes.

Informs: #158378
Epic: CRDB-48845

Release note: none

Co-authored-by: Matt Spilchen <[email protected]>

Author: craig[bot]

pkg/jobs/jobspb/jobs.proto

@@ -788,6 +788,13 @@ message TypeSchemaChangeProgress {
 
 }
 
+// IndexBackfillDistributedMergeMode enumerates the per-job distributed merge
+// decision for index backfills.
+enum IndexBackfillDistributedMergeMode {
+  INDEX_BACKFILL_DISTRIBUTED_MERGE_MODE_DISABLED = 0 [(gogoproto.enumvalue_customname) = "Disabled"];
+  INDEX_BACKFILL_DISTRIBUTED_MERGE_MODE_ENABLED = 1 [(gogoproto.enumvalue_customname) = "Enabled"];
+}
+
 // NewSchemaChangeDetails is the job detail information for the new schema change job.
 message NewSchemaChangeDetails {
 
@@ -806,6 +813,11 @@ message NewSchemaChangeDetails {
     (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID"
   ];
 
+  // DistributedMergeMode captures the per-job distributed merge decision for
+  // index backfills managed by this job. This is ignored if the job doesn't
+  // perform an index backfill.
+  IndexBackfillDistributedMergeMode distributed_merge_mode = 8;
+
   reserved 1, 2, 3, 5;
 }
 
@@ -1039,7 +1051,12 @@ message SchemaChangeDetails {
 
   sessiondatapb.SessionData session_data = 13;
 
-  // Next id 14.
+  // DistributedMergeMode captures the per-job distributed merge decision for
+  // index backfills managed by this job. This is ignored if the job doesn't
+  // perform an index backfill.
+  IndexBackfillDistributedMergeMode distributed_merge_mode = 14;
+
+  // Next id 15.
 }
 
 message SchemaChangeProgress {

pkg/jobs/registry.go

@@ -286,6 +286,11 @@ func (r *Registry) ID() base.SQLInstanceID {
 	return r.nodeID.SQLInstanceID()
 }
 
+// ClusterSettings returns the registry's cluster settings handle.
+func (r *Registry) ClusterSettings() *cluster.Settings {
+	return r.settings
+}
+
 // makeCtx returns a new context from r's ambient context and an associated
 // cancel func.
 func (r *Registry) makeCtx() (context.Context, func()) {

pkg/sql/backfill.go

@@ -1044,8 +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, sc.execCfg.NodeInfo.NodeID.SQLInstanceID(), &spec); err != nil {
-			return err
+		if details, ok := sc.job.Details().(jobspb.SchemaChangeDetails); ok &&
+			details.DistributedMergeMode == jobspb.IndexBackfillDistributedMergeMode_Enabled {
+			backfill.EnableDistributedMergeIndexBackfillSink(sc.execCfg.NodeInfo.NodeID.SQLInstanceID(), &spec)
 		}
 		p, err = sc.distSQLPlanner.createBackfillerPhysicalPlan(ctx, planCtx, spec, todoSpans)
 		return err

pkg/sql/backfill/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "backfill",
     srcs = [
         "backfill.go",
+        "distributed_merge_mode.go",
         "index_backfiller_cols.go",
         "mvcc_index_merger.go",
     ],
@@ -12,11 +13,13 @@ go_library(
     deps = [
         "//pkg/base",
         "//pkg/clusterversion",
+        "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/kv/kvpb",
         "//pkg/roachpb",
         "//pkg/settings",
+        "//pkg/settings/cluster",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catenumpb",
         "//pkg/sql/catalog/descpb",

pkg/sql/backfill/distributed_merge_mode.go

@@ -0,0 +1,121 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package backfill
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"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/errors"
+)
+
+// DistributedMergeConsumer identifies which backfill pipeline is attempting to
+// opt into the distributed merge infrastructure.
+type DistributedMergeConsumer int
+
+const (
+	// DistributedMergeConsumerLegacy corresponds to the legacy schema changer.
+	DistributedMergeConsumerLegacy DistributedMergeConsumer = iota
+	// DistributedMergeConsumerDeclarative corresponds to the declarative schema
+	// changer (new schema change).
+	DistributedMergeConsumerDeclarative
+)
+
+type distributedMergeIndexBackfillMode int64
+
+const (
+	distributedMergeModeDisabled distributedMergeIndexBackfillMode = iota
+	distributedMergeModeEnabled
+	distributedMergeModeLegacy
+	distributedMergeModeDeclarative
+	// aliases for synonyms.
+	distributedMergeModeAliasFalse
+	distributedMergeModeAliasTrue
+	distributedMergeModeAliasOff
+	distributedMergeModeAliasOn
+)
+
+// DistributedMergeIndexBackfillMode exposes the cluster setting used to control
+// when index backfills run through the distributed merge pipeline.
+var DistributedMergeIndexBackfillMode = settings.RegisterEnumSetting(
+	settings.ApplicationLevel,
+	"bulkio.index_backfill.distributed_merge.mode",
+	"controls when the distributed merge pipeline powers index backfills: disabled/off/false, legacy, declarative, or enabled/on/true",
+	"disabled",
+	map[distributedMergeIndexBackfillMode]string{
+		distributedMergeModeDisabled:    "disabled",
+		distributedMergeModeEnabled:     "enabled",
+		distributedMergeModeLegacy:      "legacy",
+		distributedMergeModeDeclarative: "declarative",
+		distributedMergeModeAliasFalse:  "false",
+		distributedMergeModeAliasTrue:   "true",
+		distributedMergeModeAliasOff:    "off",
+		distributedMergeModeAliasOn:     "on",
+	},
+	settings.WithRetiredName("bulkio.index_backfill.distributed_merge.enabled"),
+)
+
+// shouldEnableDistributedMergeIndexBackfill determines whether the specified
+// backfill consumer should opt into the distributed merge pipeline based on the
+// current cluster setting and version state.
+func shouldEnableDistributedMergeIndexBackfill(
+	ctx context.Context, st *cluster.Settings, consumer DistributedMergeConsumer,
+) (bool, error) {
+	mode := DistributedMergeIndexBackfillMode.Get(&st.SV)
+	var enable bool
+	switch mode {
+	case distributedMergeModeDisabled, distributedMergeModeAliasFalse, distributedMergeModeAliasOff:
+		enable = false
+	case distributedMergeModeLegacy:
+		enable = consumer == DistributedMergeConsumerLegacy
+	case distributedMergeModeDeclarative:
+		enable = consumer == DistributedMergeConsumerDeclarative
+	case distributedMergeModeEnabled, distributedMergeModeAliasTrue, distributedMergeModeAliasOn:
+		enable = true
+	default:
+		return false, errors.AssertionFailedf("unrecognized distributed merge index backfill mode %d", mode)
+	}
+	if enable && !st.Version.IsActive(ctx, clusterversion.V26_1) {
+		return false, pgerror.New(pgcode.FeatureNotSupported, "distributed merge requires cluster version 26.1")
+	}
+	return enable, nil
+}
+
+// EnableDistributedMergeIndexBackfillSink updates the backfiller spec to use the
+// distributed merge sink and file prefix for the provided SQL instance.
+func EnableDistributedMergeIndexBackfillSink(
+	nodeID base.SQLInstanceID, spec *execinfrapb.BackfillerSpec,
+) {
+	spec.UseDistributedMergeSink = true
+	spec.DistributedMergeFilePrefix = fmt.Sprintf("nodelocal://%d/index-backfill", nodeID)
+}
+
+// DetermineDistributedMergeMode evaluates the cluster setting to decide
+// whether backfills for the specified consumer should opt into the distributed
+// merge pipeline.
+func DetermineDistributedMergeMode(
+	ctx context.Context, st *cluster.Settings, consumer DistributedMergeConsumer,
+) (jobspb.IndexBackfillDistributedMergeMode, error) {
+	if st == nil {
+		return jobspb.IndexBackfillDistributedMergeMode_Disabled, nil
+	}
+	useDistributedMerge, err := shouldEnableDistributedMergeIndexBackfill(ctx, st, consumer)
+	if err != nil {
+		return jobspb.IndexBackfillDistributedMergeMode_Disabled, err
+	}
+	if useDistributedMerge {
+		return jobspb.IndexBackfillDistributedMergeMode_Enabled, nil
+	}
+	return jobspb.IndexBackfillDistributedMergeMode_Disabled, nil
+}

pkg/sql/distsql_plan_backfill.go

@@ -7,20 +7,14 @@ 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"
-	"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"
@@ -86,30 +80,6 @@ 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,
-	nodeID base.SQLInstanceID,
-	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
-	spec.DistributedMergeFilePrefix = fmt.Sprintf("nodelocal://%d/index-backfill", nodeID)
-	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/index_backfiller.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/backfill"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
@@ -121,6 +122,10 @@ func (ib *IndexBackfillPlanner) BackfillIndexes(
 	updateSSTManifests := func(newManifests []jobspb.IndexBackfillSSTManifest) {
 		progress.SSTManifests = sstManifestBuf.Append(newManifests)
 	}
+	mode, err := getIndexBackfillDistributedMergeMode(job)
+	if err != nil {
+		return err
+	}
 	updateFunc := func(
 		ctx context.Context, meta *execinfrapb.ProducerMetadata,
 	) error {
@@ -175,6 +180,7 @@ func (ib *IndexBackfillPlanner) BackfillIndexes(
 	// timestamp because other writing transactions have been writing at the
 	// appropriate timestamps in-between.
 	readAsOf := now
+	useDistributedMerge := mode == jobspb.IndexBackfillDistributedMergeMode_Enabled
 	run, retErr := ib.plan(
 		ctx,
 		descriptor,
@@ -184,6 +190,7 @@ func (ib *IndexBackfillPlanner) BackfillIndexes(
 		spansToDo,
 		progress.DestIndexIDs,
 		progress.SourceIndexID,
+		useDistributedMerge,
 		updateFunc,
 	)
 	if retErr != nil {
@@ -235,6 +242,7 @@ func (ib *IndexBackfillPlanner) plan(
 	sourceSpans []roachpb.Span,
 	indexesToBackfill []descpb.IndexID,
 	sourceIndexID descpb.IndexID,
+	useDistributedMerge bool,
 	callback func(_ context.Context, meta *execinfrapb.ProducerMetadata) error,
 ) (runFunc func(context.Context) error, _ error) {
 
@@ -258,8 +266,8 @@ func (ib *IndexBackfillPlanner) plan(
 			*td.TableDesc(), writeAsOf, writeAtRequestTimestamp, chunkSize,
 			indexesToBackfill, sourceIndexID,
 		)
-		if err := maybeEnableDistributedMergeIndexBackfill(ctx, ib.execCfg.Settings, ib.execCfg.NodeInfo.NodeID.SQLInstanceID(), &spec); err != nil {
-			return err
+		if useDistributedMerge {
+			backfill.EnableDistributedMergeIndexBackfillSink(ib.execCfg.NodeInfo.NodeID.SQLInstanceID(), &spec)
 		}
 		var err error
 		p, err = ib.execCfg.DistSQLPlanner.createBackfillerPhysicalPlan(ctx, planCtx, spec, sourceSpans)
@@ -286,3 +294,15 @@ func (ib *IndexBackfillPlanner) plan(
 		return cbw.Err()
 	}, nil
 }
+
+func getIndexBackfillDistributedMergeMode(
+	job *jobs.Job,
+) (jobspb.IndexBackfillDistributedMergeMode, error) {
+	payload := job.Payload()
+	details := payload.GetNewSchemaChange()
+	if details == nil {
+		return jobspb.IndexBackfillDistributedMergeMode_Disabled,
+			errors.AssertionFailedf("expected new schema change details on job %d", job.ID())
+	}
+	return details.DistributedMergeMode, nil
+}

pkg/sql/logictest/testdata/logic_test/create_index

@@ -712,13 +712,13 @@ subtest end
 subtest distributed_merge_index_backfill_placeholder
 
 statement ok
-CREATE TABLE dist_merge_idx (a INT PRIMARY KEY, b INT)
+CREATE TABLE dist_merge_idx (a INT NOT NULL PRIMARY KEY, b INT NOT NULL, c INT NOT NULL UNIQUE)
 
 statement ok
-INSERT INTO dist_merge_idx VALUES (1,1), (2,2), (3,3)
+INSERT INTO dist_merge_idx VALUES (1,1,1), (2,2,2), (3,3,3)
 
 statement ok
-SET CLUSTER SETTING bulkio.index_backfill.distributed_merge.enabled = true
+SET CLUSTER SETTING bulkio.index_backfill.distributed_merge.mode = enabled
 
 # 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
@@ -732,8 +732,21 @@ 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)
 
+# Try an operation that is implemented in the declarative schema changer.
+#
+# TODO(158378): Declarative schema change don't yet implement the full flow for
+# distributed merge. So, we end up with a similar situation as above where we
+# get to validation and notice no rows were ingested.
+skipif config local-mixed-25.3 local-mixed-25.4
+statement error pgcode 23505 pq: .*duplicate key value violates unique constraint.*
+ALTER TABLE dist_merge_idx ALTER PRIMARY KEY USING COLUMNS (b)
+
+onlyif config local-mixed-25.3 local-mixed-25.4
+statement error pq: .*distributed merge requires cluster version 26.*
+ALTER TABLE dist_merge_idx ALTER PRIMARY KEY USING COLUMNS (b)
+
 statement ok
-SET CLUSTER SETTING bulkio.index_backfill.distributed_merge.enabled = false
+SET CLUSTER SETTING bulkio.index_backfill.distributed_merge.mode = 'disabled'
 
 statement ok
 DROP TABLE dist_merge_idx