changefeedccl: make range_distribution_strategy a changefeed option

Previously, range_distribution_strategy was a cluster setting. Moving
this from a cluster setting to a changefeed option gives users control
over which changefeeds it applies to (eg initial scans vs long-running
feeds).

Fixes: #147628

Release note (sql change): Replace cluster setting
changefeed.default_range_distribution_strategy="balanced_simple" with
changefeed setting range_distribution_strategy with values 'default' or
'balanced_simple'. 'balanced_simple' naively redistributes spans evenly
across aggregators with no regard for replica placement.

Author: log-head

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -42,7 +42,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
 	"github.com/cockroachdb/errors"
 )
 
@@ -345,37 +344,6 @@ func startDistChangefeed(
 // The bin packing choice gives preference to leaseholder replicas if possible.
 var replicaOracleChoice = replicaoracle.BinPackingChoice
 
-type rangeDistributionType int
-
-const (
-	// defaultDistribution employs no load balancing on the changefeed
-	// side. We defer to distsql to select nodes and distribute work.
-	defaultDistribution rangeDistributionType = 0
-	// balancedSimpleDistribution defers to distsql for selecting the
-	// set of nodes to distribute work to. However, changefeeds will try to
-	// distribute work evenly across this set of nodes.
-	balancedSimpleDistribution rangeDistributionType = 1
-	// TODO(jayant): add balancedFullDistribution which takes
-	// full control of node selection and distribution.
-)
-
-// RangeDistributionStrategy is used to determine how the changefeed balances
-// ranges between nodes.
-// TODO: deprecate this setting in favor of a changefeed option.
-var RangeDistributionStrategy = settings.RegisterEnumSetting(
-	settings.ApplicationLevel,
-	"changefeed.default_range_distribution_strategy",
-	"configures how work is distributed among nodes for a given changefeed. "+
-		"for the most balanced distribution, use `balanced_simple`. changing this setting "+
-		"will not override locality restrictions",
-	metamorphic.ConstantWithTestChoice("default_range_distribution_strategy",
-		"default", "balanced_simple"),
-	map[rangeDistributionType]string{
-		defaultDistribution:        "default",
-		balancedSimpleDistribution: "balanced_simple",
-	},
-	settings.WithPublic)
-
 var useBulkOracle = settings.RegisterBoolSetting(
 	settings.ApplicationLevel,
 	"changefeed.random_replica_selection.enabled",
@@ -410,7 +378,10 @@ func makePlan(
 			}
 		}
 
-		rangeDistribution := RangeDistributionStrategy.Get(sv)
+		rangeDistributionStrat, err := changefeedbase.MakeStatementOptions(details.Opts).GetRangeDistributionStrategy()
+		if err != nil {
+			return nil, nil, err
+		}
 		evalCtx := execCtx.ExtendedEvalContext()
 		oracle := replicaoracle.NewOracle(replicaOracleChoice, dsp.ReplicaOracleConfig(locFilter))
 		if useBulkOracle.Get(&evalCtx.Settings.SV) {
@@ -427,8 +398,8 @@ func makePlan(
 			log.Changefeed.Infof(ctx, "spans returned by DistSQL: %v", spanPartitions)
 		}
 		switch {
-		case distMode == sql.LocalDistribution || rangeDistribution == defaultDistribution:
-		case rangeDistribution == balancedSimpleDistribution:
+		case distMode == sql.LocalDistribution || rangeDistributionStrat == changefeedbase.RangeDistributionStrategyDefault:
+		case rangeDistributionStrat == changefeedbase.RangeDistributionStrategyBalancedSimple:
 			log.Changefeed.Infof(ctx, "rebalancing ranges using balanced simple distribution")
 			sender := execCtx.ExecCfg().DB.NonTransactionalSender()
 			distSender := sender.(*kv.CrossRangeTxnWrapperSender).Wrapped().(*kvcoord.DistSender)
@@ -442,8 +413,8 @@ func makePlan(
 				log.Changefeed.Infof(ctx, "spans after balanced simple distribution rebalancing: %v", spanPartitions)
 			}
 		default:
-			return nil, nil, errors.AssertionFailedf("unsupported dist strategy %d and dist mode %d",
-				rangeDistribution, distMode)
+			return nil, nil, errors.AssertionFailedf("unsupported dist strategy %s and dist mode %d",
+				rangeDistributionStrat, distMode)
 		}
 
 		if haveKnobs && maybeCfKnobs.FilterDrainingNodes != nil && len(drainingNodes) > 0 {

pkg/ccl/changefeedccl/changefeed_dist_test.go

@@ -500,7 +500,6 @@ func TestChangefeedWithNoDistributionStrategy(t *testing.T) {
 	tester := newRangeDistributionTester(t, noLocality)
 	defer tester.cleanup()
 
-	serverutils.SetClusterSetting(t, tester.tc, "changefeed.default_range_distribution_strategy", "default")
 	serverutils.SetClusterSetting(t, tester.tc, "changefeed.random_replica_selection.enabled", false)
 	tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no'")
 	partitions := tester.getPartitions()
@@ -527,10 +526,9 @@ func TestChangefeedWithSimpleDistributionStrategy(t *testing.T) {
 	// Check that we roughly assign (64 ranges / 6 nodes) ranges to each node.
 	tester := newRangeDistributionTester(t, noLocality)
 	defer tester.cleanup()
-	tester.sqlDB.Exec(t, "SET CLUSTER SETTING changefeed.default_range_distribution_strategy = 'balanced_simple'")
 	// We need to disable the bulk oracle in order to ensure the leaseholder is selected.
 	tester.sqlDB.Exec(t, "SET CLUSTER SETTING changefeed.random_replica_selection.enabled = false")
-	tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no'")
+	tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no', range_distribution_strategy='balanced_simple'")
 	partitions := tester.getPartitions()
 	counts := tester.countRangesPerNode(partitions)
 	upper := int(math.Ceil((1 + rebalanceThreshold.Get(&tester.lastNode.ClusterSettings().SV)) * 64 / 6))
@@ -548,30 +546,56 @@ func TestChangefeedWithNoDistributionStrategyAndConstrainedLocality(t *testing.T
 	skip.UnderShort(t)
 	skip.UnderDuress(t)
 
-	// The replica oracle selects the leaseholder replica for each range. Then, distsql assigns the replica
-	// to the same node which stores it. However, node of these nodes don't pass the filter. The replicas assigned
-	// to these nodes are distributed arbitrarily to any nodes which pass the filter.
-	tester := newRangeDistributionTester(t, func(i int) []roachpb.Tier {
-		if i%2 == 1 {
-			return []roachpb.Tier{{Key: "y", Value: "1"}}
+	t.Run("default specified", func(t *testing.T) {
+		// The replica oracle selects the leaseholder replica for each range. Then, distsql assigns the replica
+		// to the same node which stores it. However, node of these nodes don't pass the filter. The replicas assigned
+		// to these nodes are distributed arbitrarily to any nodes which pass the filter.
+		tester := newRangeDistributionTester(t, func(i int) []roachpb.Tier {
+			if i%2 == 1 {
+				return []roachpb.Tier{{Key: "y", Value: "1"}}
+			}
+			return []roachpb.Tier{}
+		})
+		defer tester.cleanup()
+		tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no', execution_locality='y=1', range_distribution_strategy='default'")
+		partitions := tester.getPartitions()
+		counts := tester.countRangesPerNode(partitions)
+
+		totalRanges := 0
+		for i, count := range counts {
+			if i%2 == 1 {
+				totalRanges += count
+			} else {
+				require.Equal(t, count, 0)
+			}
 		}
-		return []roachpb.Tier{}
+		require.Equal(t, totalRanges, 64)
 	})
-	defer tester.cleanup()
-	tester.sqlDB.Exec(t, "SET CLUSTER SETTING changefeed.default_range_distribution_strategy = 'default'")
-	tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no', execution_locality='y=1'")
-	partitions := tester.getPartitions()
-	counts := tester.countRangesPerNode(partitions)
-
-	totalRanges := 0
-	for i, count := range counts {
-		if i%2 == 1 {
-			totalRanges += count
-		} else {
-			require.Equal(t, count, 0)
+	t.Run("no distribution strategy specified", func(t *testing.T) {
+		// The replica oracle selects the leaseholder replica for each range. Then, distsql assigns the replica
+		// to the same node which stores it. However, node of these nodes don't pass the filter. The replicas assigned
+		// to these nodes are distributed arbitrarily to any nodes which pass the filter.
+		tester := newRangeDistributionTester(t, func(i int) []roachpb.Tier {
+			if i%2 == 1 {
+				return []roachpb.Tier{{Key: "y", Value: "1"}}
+			}
+			return []roachpb.Tier{}
+		})
+		defer tester.cleanup()
+		tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no', execution_locality='y=1'")
+		partitions := tester.getPartitions()
+		counts := tester.countRangesPerNode(partitions)
+
+		totalRanges := 0
+		for i, count := range counts {
+			if i%2 == 1 {
+				totalRanges += count
+			} else {
+				require.Equal(t, count, 0)
+			}
 		}
-	}
-	require.Equal(t, totalRanges, 64)
+		require.Equal(t, totalRanges, 64)
+	})
 }
 
 func TestChangefeedWithSimpleDistributionStrategyAndConstrainedLocality(t *testing.T) {
@@ -593,8 +617,7 @@ func TestChangefeedWithSimpleDistributionStrategyAndConstrainedLocality(t *testi
 		return []roachpb.Tier{}
 	})
 	defer tester.cleanup()
-	tester.sqlDB.Exec(t, "SET CLUSTER SETTING changefeed.default_range_distribution_strategy = 'balanced_simple'")
-	tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no', execution_locality='y=1'")
+	tester.sqlDB.Exec(t, "CREATE CHANGEFEED FOR x INTO 'null://' WITH initial_scan='no', execution_locality='y=1', range_distribution_strategy='balanced_simple'")
 	partitions := tester.getPartitions()
 	counts := tester.countRangesPerNode(partitions)
 

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -75,6 +75,19 @@ const (
 	EnrichedPropertySchema EnrichedProperty = `schema`
 )
 
+// RangeDistributionStrategy configures how the changefeed balances ranges between nodes.
+type RangeDistributionStrategy string
+
+const (
+	// RangeDistributionStrategyDefault employs no load balancing on the changefeed
+	// side. We defer to distsql to select nodes and distribute work.
+	RangeDistributionStrategyDefault RangeDistributionStrategy = `default`
+	// RangeDistributionStrategyBalancedSimple defers to distsql for selecting the
+	// set of nodes to distribute work to. However, changefeeds will try to
+	// distribute work evenly across this set of nodes.
+	RangeDistributionStrategyBalancedSimple RangeDistributionStrategy = `balanced_simple`
+)
+
 // Constants for the initial scan types
 const (
 	InitialScan InitialScanType = iota
@@ -162,6 +175,8 @@ const (
 
 	OptEnrichedProperties = `enriched_properties`
 
+	OptRangeDistributionStrategy = `range_distribution_strategy`
+
 	OptEnvelopeKeyOnly       EnvelopeType = `key_only`
 	OptEnvelopeRow           EnvelopeType = `row`
 	OptEnvelopeDeprecatedRow EnvelopeType = `deprecated_row`
@@ -412,6 +427,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptIgnoreDisableChangefeedReplication: flagOption,
 	OptEncodeJSONValueNullAsObject:        flagOption,
 	OptEnrichedProperties:                 csv(string(EnrichedPropertySource), string(EnrichedPropertySchema)),
+	OptRangeDistributionStrategy:          enum("default", "balanced_simple"),
 	OptHeadersJSONColumnName:              stringOption,
 	OptExtraHeaders:                       jsonOption,
 }
@@ -428,6 +444,7 @@ var CommonOptions = makeStringSet(OptCursor, OptEndTime, OptEnvelope,
 	OptMinCheckpointFrequency, OptMetricsScope, OptVirtualColumns, Topics, OptExpirePTSAfter,
 	OptExecutionLocality, OptLaggingRangesThreshold, OptLaggingRangesPollingInterval,
 	OptIgnoreDisableChangefeedReplication, OptEncodeJSONValueNullAsObject, OptEnrichedProperties,
+	OptRangeDistributionStrategy,
 )
 
 // SQLValidOptions is options exclusive to SQL sink
@@ -805,6 +822,17 @@ func (s StatementOptions) IsInitialScanSpecified() bool {
 	return true
 }
 
+func (s StatementOptions) GetRangeDistributionStrategy() (RangeDistributionStrategy, error) {
+	v, err := s.getEnumValue(OptRangeDistributionStrategy)
+	if err != nil {
+		return "", err
+	}
+	if v == `` {
+		return RangeDistributionStrategyDefault, nil
+	}
+	return RangeDistributionStrategy(v), nil
+}
+
 // ShouldUseFullStatementTimeName returns true if references to the table should be in db.schema.table
 // format (e.g. in Kafka topics).
 func (s StatementOptions) ShouldUseFullStatementTimeName() bool {