changefeedccl: make range_distribution_strategy a changefeed option

Previously, range_distribution_strategy was only a cluster setting.
Adding a changefeed option gives users control per-changefeed for
different needs (eg initial scans vs long-running feeds). The cluster
setting will continue to apply to all changefeeds, but per-changefeed
range distribution options will override.

Fixes: #147628

Release note (sql change): Add changefeed setting
range_distribution_strategy with values 'default' or 'balanced_simple'
(e.g. CREATE CHANGEFEED FOR x into 'null://' WITH
range_distribution_strategy='balanced_simple').
This new per-changefeed setting will override the cluster setting
changefeed.default_range_distribution_strategy where both exist.

Author: log-head

docs/generated/settings/settings-for-tenants.txt

@@ -12,7 +12,7 @@ changefeed.aggregator.flush_jitter	float	0.1	jitter aggregator flushes as a frac
 changefeed.backfill.concurrent_scan_requests	integer	0	number of concurrent scan requests per node issued during a backfill	application
 changefeed.backfill.scan_request_size	integer	524288	the maximum number of bytes returned by each scan request	application
 changefeed.batch_reduction_retry.enabled (alias: changefeed.batch_reduction_retry_enabled)	boolean	false	if true, kafka changefeeds upon erroring on an oversized batch will attempt to resend the messages with progressively lower batch sizes	application
-changefeed.default_range_distribution_strategy	enumeration	default	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 [default = 0, balanced_simple = 1]	application
+changefeed.default_range_distribution_strategy	enumeration	default	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. this setting can be overridden by the changefeed option `range_distribution_strategy` [default = 0, balanced_simple = 1]	application
 changefeed.event_consumer_worker_queue_size	integer	16	if changefeed.event_consumer_workers is enabled, this setting sets the maxmimum number of events which a worker can buffer	application
 changefeed.event_consumer_workers	integer	0	the number of workers to use when processing events: <0 disables, 0 assigns a reasonable default, >0 assigns the setting value. for experimental/core changefeeds and changefeeds using parquet format, this is disabled	application
 changefeed.fast_gzip.enabled	boolean	true	use fast gzip implementation	application

docs/generated/settings/settings.html

@@ -17,7 +17,7 @@
 <tr><td><div id="setting-changefeed-backfill-concurrent-scan-requests" class="anchored"><code>changefeed.backfill.concurrent_scan_requests</code></div></td><td>integer</td><td><code>0</code></td><td>number of concurrent scan requests per node issued during a backfill</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-backfill-scan-request-size" class="anchored"><code>changefeed.backfill.scan_request_size</code></div></td><td>integer</td><td><code>524288</code></td><td>the maximum number of bytes returned by each scan request</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-batch-reduction-retry-enabled" class="anchored"><code>changefeed.batch_reduction_retry.enabled<br />(alias: changefeed.batch_reduction_retry_enabled)</code></div></td><td>boolean</td><td><code>false</code></td><td>if true, kafka changefeeds upon erroring on an oversized batch will attempt to resend the messages with progressively lower batch sizes</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
-<tr><td><div id="setting-changefeed-default-range-distribution-strategy" class="anchored"><code>changefeed.default_range_distribution_strategy</code></div></td><td>enumeration</td><td><code>default</code></td><td>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 [default = 0, balanced_simple = 1]</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
+<tr><td><div id="setting-changefeed-default-range-distribution-strategy" class="anchored"><code>changefeed.default_range_distribution_strategy</code></div></td><td>enumeration</td><td><code>default</code></td><td>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. this setting can be overridden by the changefeed option `range_distribution_strategy` [default = 0, balanced_simple = 1]</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-event-consumer-worker-queue-size" class="anchored"><code>changefeed.event_consumer_worker_queue_size</code></div></td><td>integer</td><td><code>16</code></td><td>if changefeed.event_consumer_workers is enabled, this setting sets the maxmimum number of events which a worker can buffer</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-event-consumer-workers" class="anchored"><code>changefeed.event_consumer_workers</code></div></td><td>integer</td><td><code>0</code></td><td>the number of workers to use when processing events: &lt;0 disables, 0 assigns a reasonable default, &gt;0 assigns the setting value. for experimental/core changefeeds and changefeeds using parquet format, this is disabled</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-changefeed-fast-gzip-enabled" class="anchored"><code>changefeed.fast_gzip.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>use fast gzip implementation</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -345,35 +345,36 @@ func startDistChangefeed(
 // The bin packing choice gives preference to leaseholder replicas if possible.
 var replicaOracleChoice = replicaoracle.BinPackingChoice
 
-type rangeDistributionType int
+type clusterSettingRangeDistributionType int
 
 const (
 	// defaultDistribution employs no load balancing on the changefeed
 	// side. We defer to distsql to select nodes and distribute work.
-	defaultDistribution rangeDistributionType = 0
+	defaultDistribution clusterSettingRangeDistributionType = 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
+	balancedSimpleDistribution clusterSettingRangeDistributionType = 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 rangeDistributionStrategyStrings = map[clusterSettingRangeDistributionType]string{
+	defaultDistribution:        string(changefeedbase.ChangefeedRangeDistributionStrategyDefault),
+	balancedSimpleDistribution: string(changefeedbase.ChangefeedRangeDistributionStrategyBalancedSimple),
+}
+
 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",
+		"will not override locality restrictions. this setting can be overridden by the "+
+		"changefeed option `range_distribution_strategy`",
 	metamorphic.ConstantWithTestChoice("default_range_distribution_strategy",
-		"default", "balanced_simple"),
-	map[rangeDistributionType]string{
-		defaultDistribution:        "default",
-		balancedSimpleDistribution: "balanced_simple",
-	},
+		string(changefeedbase.ChangefeedRangeDistributionStrategyDefault),
+		string(changefeedbase.ChangefeedRangeDistributionStrategyBalancedSimple)),
+	rangeDistributionStrategyStrings,
 	settings.WithPublic)
 
 var useBulkOracle = settings.RegisterBoolSetting(
@@ -410,7 +411,6 @@ func makePlan(
 			}
 		}
 
-		rangeDistribution := RangeDistributionStrategy.Get(sv)
 		evalCtx := execCtx.ExtendedEvalContext()
 		oracle := replicaoracle.NewOracle(replicaOracleChoice, dsp.ReplicaOracleConfig(locFilter))
 		if useBulkOracle.Get(&evalCtx.Settings.SV) {
@@ -426,9 +426,27 @@ func makePlan(
 		if log.ExpensiveLogEnabled(ctx, 2) {
 			log.Changefeed.Infof(ctx, "spans returned by DistSQL: %v", spanPartitions)
 		}
+		// Preference for the range distribution strategy is given to the
+		// changefeed option. If none is specified, the cluster setting,
+		// defaulting to 'default', is used. The default behavior is to defer
+		// to distsql for range distribution and not rebalance.
+		changefeedRangeDistribution, err := changefeedbase.MakeStatementOptions(details.Opts).GetChangefeedRangeDistributionStrategy()
+		if err != nil {
+			return nil, nil, err
+		}
+		clusterRangeDistribution := changefeedbase.ChangefeedRangeDistributionStrategy(
+			rangeDistributionStrategyStrings[RangeDistributionStrategy.Get(sv)])
+
+		rangeDistributionStrategy := changefeedRangeDistribution
+		if rangeDistributionStrategy == changefeedbase.ChangefeedRangeDistributionStrategyNotSpecified {
+			rangeDistributionStrategy = clusterRangeDistribution
+		}
+		if haveKnobs && maybeCfKnobs.RangeDistributionStrategyCallback != nil {
+			maybeCfKnobs.RangeDistributionStrategyCallback(rangeDistributionStrategy)
+		}
 		switch {
-		case distMode == sql.LocalDistribution || rangeDistribution == defaultDistribution:
-		case rangeDistribution == balancedSimpleDistribution:
+		case distMode == sql.LocalDistribution || rangeDistributionStrategy == changefeedbase.ChangefeedRangeDistributionStrategyDefault:
+		case rangeDistributionStrategy == changefeedbase.ChangefeedRangeDistributionStrategyBalancedSimple:
 			log.Changefeed.Infof(ctx, "rebalancing ranges using balanced simple distribution")
 			sender := execCtx.ExecCfg().DB.NonTransactionalSender()
 			distSender := sender.(*kv.CrossRangeTxnWrapperSender).Wrapped().(*kvcoord.DistSender)
@@ -442,8 +460,11 @@ 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 (cluster setting: %q, changefeed option: %q)",
+				rangeDistributionStrategy, distMode,
+				clusterRangeDistribution, changefeedRangeDistribution,
+			)
 		}
 
 		if haveKnobs && maybeCfKnobs.FilterDrainingNodes != nil && len(drainingNodes) > 0 {

pkg/ccl/changefeedccl/changefeed_dist_test.go

@@ -16,6 +16,8 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
@@ -480,6 +482,64 @@ func (rdt *rangeDistributionTester) balancedDistributionUpperBound(numNodes int)
 	return int(math.Ceil((1 + rebalanceThreshold.Get(&rdt.lastNode.ClusterSettings().SV) + 0.1) * 64 / float64(numNodes)))
 }
 
+func TestChangefeedWithDistributionStrategyOptions(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	clusterOptions := []changefeedbase.ChangefeedRangeDistributionStrategy{
+		changefeedbase.ChangefeedRangeDistributionStrategyDefault,
+		changefeedbase.ChangefeedRangeDistributionStrategyBalancedSimple,
+	}
+	changefeedOptions := []changefeedbase.ChangefeedRangeDistributionStrategy{
+		changefeedbase.ChangefeedRangeDistributionStrategyDefault,
+		changefeedbase.ChangefeedRangeDistributionStrategyBalancedSimple,
+		changefeedbase.ChangefeedRangeDistributionStrategyNotSpecified,
+	}
+	testutils.RunValues(t, "cluster option", clusterOptions, func(t *testing.T, clusterOption changefeedbase.ChangefeedRangeDistributionStrategy) {
+		testutils.RunValues(t, "changefeed option", changefeedOptions, func(t *testing.T, changefeedOption changefeedbase.ChangefeedRangeDistributionStrategy) {
+			expectedStrat := changefeedOption
+			if changefeedOption == changefeedbase.ChangefeedRangeDistributionStrategyNotSpecified {
+				expectedStrat = clusterOption
+			}
+
+			testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+				sqlDB := sqlutils.MakeSQLRunner(s.DB)
+				sqlDB.Exec(t, fmt.Sprintf("SET CLUSTER SETTING changefeed.default_range_distribution_strategy = '%s'", clusterOption))
+				sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
+				sqlDB.Exec(t, `INSERT INTO foo VALUES (0, 'initial')`)
+
+				stratCh := make(chan changefeedbase.ChangefeedRangeDistributionStrategy, 1)
+
+				knobs := s.TestingKnobs.DistSQL.(*execinfra.TestingKnobs).Changefeed.(*TestingKnobs)
+				knobs.RangeDistributionStrategyCallback = func(strat changefeedbase.ChangefeedRangeDistributionStrategy) {
+					select {
+					case stratCh <- strat:
+					default:
+						fmt.Printf("skipping strat: %s\n", strat)
+					}
+				}
+
+				createStmt := "CREATE CHANGEFEED FOR foo"
+				if changefeedOption != changefeedbase.ChangefeedRangeDistributionStrategyNotSpecified {
+					createStmt += fmt.Sprintf(" WITH range_distribution_strategy='%s'", changefeedOption)
+				}
+				feed := feed(t, f, createStmt)
+				defer closeFeed(t, feed)
+				testutils.SucceedsSoon(t, func() error {
+					select {
+					case strat := <-stratCh:
+						require.Equal(t, expectedStrat, strat)
+						return nil
+					default:
+						return errors.New("no range distribution strategy callback found")
+					}
+				})
+			}
+			cdcTest(t, testFn)
+		})
+	})
+}
+
 func TestChangefeedWithNoDistributionStrategy(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -75,6 +75,23 @@ const (
 	EnrichedPropertySchema EnrichedProperty = `schema`
 )
 
+// ChangefeedRangeDistributionStrategy configures how the changefeed balances
+// ranges between nodes.
+type ChangefeedRangeDistributionStrategy string
+
+const (
+	// ChangefeedRangeDistributionStrategyDefault employs no load balancing on
+	// the changefeed side. We defer to distsql to select nodes and distribute work.
+	ChangefeedRangeDistributionStrategyDefault ChangefeedRangeDistributionStrategy = `default`
+	// ChangefeedRangeDistributionStrategyBalancedSimple 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.
+	ChangefeedRangeDistributionStrategyBalancedSimple ChangefeedRangeDistributionStrategy = `balanced_simple`
+	// ChangefeedRangeDistributionStrategyNotSpecified is used to indicate that
+	// the changefeed range distribution strategy is not specified.
+	ChangefeedRangeDistributionStrategyNotSpecified ChangefeedRangeDistributionStrategy = ``
+)
+
 // Constants for the initial scan types
 const (
 	InitialScan InitialScanType = iota
@@ -162,6 +179,8 @@ const (
 
 	OptEnrichedProperties = `enriched_properties`
 
+	OptRangeDistributionStrategy = `range_distribution_strategy`
+
 	OptEnvelopeKeyOnly       EnvelopeType = `key_only`
 	OptEnvelopeRow           EnvelopeType = `row`
 	OptEnvelopeDeprecatedRow EnvelopeType = `deprecated_row`
@@ -412,6 +431,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptIgnoreDisableChangefeedReplication: flagOption,
 	OptEncodeJSONValueNullAsObject:        flagOption,
 	OptEnrichedProperties:                 csv(string(EnrichedPropertySource), string(EnrichedPropertySchema)),
+	OptRangeDistributionStrategy:          enum(string(ChangefeedRangeDistributionStrategyDefault), string(ChangefeedRangeDistributionStrategyBalancedSimple)),
 	OptHeadersJSONColumnName:              stringOption,
 	OptExtraHeaders:                       jsonOption,
 }
@@ -428,6 +448,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 +826,17 @@ func (s StatementOptions) IsInitialScanSpecified() bool {
 	return true
 }
 
+func (s StatementOptions) GetChangefeedRangeDistributionStrategy() (
+	ChangefeedRangeDistributionStrategy,
+	error,
+) {
+	v, err := s.getEnumValue(OptRangeDistributionStrategy)
+	if err != nil {
+		return "", err
+	}
+	return ChangefeedRangeDistributionStrategy(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 {

pkg/ccl/changefeedccl/testing_knobs.go

@@ -8,6 +8,7 @@ package changefeedccl
 import (
 	"context"
 
+	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/kvevent"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/kvfeed"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/resolvedspan"
@@ -82,6 +83,10 @@ type TestingKnobs struct {
 	// when the changefeed is planned.
 	SpanPartitionsCallback func([]sql.SpanPartition)
 
+	// RangeDistributionStrategyCallback is called with the resolved
+	// range distribution strategy when the changefeed is planned.
+	RangeDistributionStrategyCallback func(changefeedbase.ChangefeedRangeDistributionStrategy)
+
 	// PreserveDeprecatedPts is used to prevent a changefeed from upgrading
 	// its PTS record from the deprecated style to the new style.
 	PreserveDeprecatedPts func() bool