changefeedccl: changefeed-specific option for num of sink workers

Previously, the number of SinkIOWorkers could only be defined in
cluster-wide setting.

This patch allows to set the number of SinkIOWorkers per changefeed.

If both (cluster-wide, per-changefeed) values are given, the min of the
two will be used.

Release note (sql change): The CREATE CHANGEFEED statement was extended
with an optional `num_sink_workers` setting that can be used to set the
number of SinkIOWorkers per changefeed. Note that the number of workers
is capped by the cluster-wide setting if present.

Fixes: #154546

Author: fabschn

pkg/ccl/changefeedccl/changefeed_test.go

@@ -12785,3 +12785,79 @@ func TestCreateTableLevelChangefeedWithDBPrivilege(t *testing.T) {
 	}
 	cdcTest(t, testFn, feedTestEnterpriseSinks)
 }
+
+// TestChangefeedNumSinkWorkersPrecedence verifies that the num_sink_workers
+// option works correctly and that the precedence logic (smaller value wins)
+// is applied when both cluster setting and per-changefeed option are set.
+func TestChangefeedNumSinkWorkersPrecedence(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		registry := s.Server.JobRegistry().(*jobs.Registry)
+		metrics := registry.MetricsStruct().Changefeed.(*Metrics).AggMetrics
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+
+		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY)`)
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1)`)
+
+		// Test 1: Per-changefeed option only (cluster setting = 0)
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 0`)
+		foo1 := feed(t, f, `CREATE CHANGEFEED FOR foo WITH num_sink_workers = '5'`)
+		defer closeFeed(t, foo1)
+
+		testutils.SucceedsSoon(t, func() error {
+			workers := metrics.ParallelIOWorkers.Value()
+			if workers != 5 {
+				return errors.Newf("expected 5 workers, got %d", workers)
+			}
+			return nil
+		})
+		require.NoError(t, foo1.Close())
+
+		// Test 2: Cluster setting = 10, per-changefeed = 3 → should use 3 (smaller)
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 10`)
+		foo2 := feed(t, f, `CREATE CHANGEFEED FOR foo WITH num_sink_workers = '3'`)
+		defer closeFeed(t, foo2)
+
+		testutils.SucceedsSoon(t, func() error {
+			workers := metrics.ParallelIOWorkers.Value()
+			if workers != 3 {
+				return errors.Newf("expected 3 workers (smaller of 10 and 3), got %d", workers)
+			}
+			return nil
+		})
+		require.NoError(t, foo2.Close())
+
+		// Test 3: Cluster setting = 2, per-changefeed = 8 → should use 2 (smaller)
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 2`)
+		foo3 := feed(t, f, `CREATE CHANGEFEED FOR foo WITH num_sink_workers = '8'`)
+		defer closeFeed(t, foo3)
+
+		testutils.SucceedsSoon(t, func() error {
+			workers := metrics.ParallelIOWorkers.Value()
+			if workers != 2 {
+				return errors.Newf("expected 2 workers (smaller of 2 and 8), got %d", workers)
+			}
+			return nil
+		})
+		require.NoError(t, foo3.Close())
+
+		// Test 4: Cluster setting only (no per-changefeed option)
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 7`)
+		foo4 := feed(t, f, `CREATE CHANGEFEED FOR foo`)
+		defer closeFeed(t, foo4)
+
+		testutils.SucceedsSoon(t, func() error {
+			workers := metrics.ParallelIOWorkers.Value()
+			if workers != 7 {
+				return errors.Newf("expected 7 workers (from cluster setting), got %d", workers)
+			}
+			return nil
+		})
+		require.NoError(t, foo4.Close())
+	}
+
+	// Test with sinks that support parallel IO
+	cdcTest(t, testFn, feedTestForceSink("pubsub"))
+}

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -115,6 +115,7 @@ const (
 	OptLaggingRangesPollingInterval       = `lagging_ranges_polling_interval`
 	OptIgnoreDisableChangefeedReplication = `ignore_disable_changefeed_replication`
 	OptEncodeJSONValueNullAsObject        = `encode_json_value_null_as_object`
+	OptNumSinkWorkers                     = `num_sink_workers`
 	// TODO(#142273): look into whether we want to add headers to pub/sub, and other
 	// sinks as well (eg cloudstorage, webhook, ..). Currently it's kafka-only.
 	OptHeadersJSONColumnName = `headers_json_column_name`
@@ -411,6 +412,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptLaggingRangesPollingInterval:       durationOption,
 	OptIgnoreDisableChangefeedReplication: flagOption,
 	OptEncodeJSONValueNullAsObject:        flagOption,
+	OptNumSinkWorkers:                     stringOption,
 	OptEnrichedProperties:                 csv(string(EnrichedPropertySource), string(EnrichedPropertySchema)),
 	OptHeadersJSONColumnName:              stringOption,
 	OptExtraHeaders:                       jsonOption,
@@ -428,6 +430,7 @@ var CommonOptions = makeStringSet(OptCursor, OptEndTime, OptEnvelope,
 	OptMinCheckpointFrequency, OptMetricsScope, OptVirtualColumns, Topics, OptExpirePTSAfter,
 	OptExecutionLocality, OptLaggingRangesThreshold, OptLaggingRangesPollingInterval,
 	OptIgnoreDisableChangefeedReplication, OptEncodeJSONValueNullAsObject, OptEnrichedProperties,
+	OptNumSinkWorkers,
 )
 
 // SQLValidOptions is options exclusive to SQL sink
@@ -1203,6 +1206,27 @@ func (s StatementOptions) GetPTSExpiration() (time.Duration, error) {
 	return *exp, nil
 }
 
+// GetNumSinkWorkers returns the number of sink IO workers to use.
+// Returns 0 if not set (which means use a reasonable default).
+// Negative values disable sink IO workers.
+func (s StatementOptions) GetNumSinkWorkers() (int64, error) {
+	v, ok := s.m[OptNumSinkWorkers]
+	if !ok {
+		return 0, nil
+	}
+	// Parse as int64
+	var result int64
+	if _, err := fmt.Sscanf(v, "%d", &result); err != nil {
+		return 0, errors.Newf("invalid integer value for %s: %q", OptNumSinkWorkers, v)
+	}
+	// Verify it's a clean parse by formatting back and comparing
+	// This rejects values like "3.14" which Sscanf would truncate to 3
+	if fmt.Sprintf("%d", result) != v {
+		return 0, errors.Newf("invalid integer value for %s: %q", OptNumSinkWorkers, v)
+	}
+	return result, nil
+}
+
 // ForceKeyInValue sets the encoding option KeyInValue to true and then validates the
 // resoluting encoding options.
 func (s StatementOptions) ForceKeyInValue() error {

pkg/ccl/changefeedccl/changefeedbase/options_test.go

@@ -51,6 +51,68 @@ func TestOptionsValidations(t *testing.T) {
 	}
 }
 
+func TestNumSinkWorkersOption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	tests := []struct {
+		name      string
+		input     map[string]string
+		expected  int64
+		expectErr bool
+	}{
+		{
+			name:      "positive value",
+			input:     map[string]string{"num_sink_workers": "5"},
+			expected:  5,
+			expectErr: false,
+		},
+		{
+			name:      "zero value (default)",
+			input:     map[string]string{"num_sink_workers": "0"},
+			expected:  0,
+			expectErr: false,
+		},
+		{
+			name:      "negative value (disable)",
+			input:     map[string]string{"num_sink_workers": "-1"},
+			expected:  -1,
+			expectErr: false,
+		},
+		{
+			name:      "not set",
+			input:     map[string]string{},
+			expected:  0,
+			expectErr: false,
+		},
+		{
+			name:      "invalid non-integer",
+			input:     map[string]string{"num_sink_workers": "abc"},
+			expected:  0,
+			expectErr: true,
+		},
+		{
+			name:      "invalid float",
+			input:     map[string]string{"num_sink_workers": "3.14"},
+			expected:  0,
+			expectErr: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			o := MakeStatementOptions(test.input)
+			val, err := o.GetNumSinkWorkers()
+			if test.expectErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.expected, val)
+			}
+		})
+	}
+}
+
 func TestEncodingOptionsValidations(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/sink.go

@@ -259,7 +259,7 @@ func getSink(
 				}
 				if KafkaV2Enabled.Get(&serverCfg.Settings.SV) {
 					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, targets, sinkOpts,
-						numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
+						numSinkIOWorkers(serverCfg, opts), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 						serverCfg.Settings, metricsBuilder, kafkaSinkV2Knobs{})
 				} else {
 					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, targets, sinkOpts, serverCfg.Settings, metricsBuilder)
@@ -283,7 +283,7 @@ func getSink(
 			}
 			return validateOptionsAndMakeSink(changefeedbase.WebhookValidOptions, func() (Sink, error) {
 				return makeWebhookSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, webhookOpts,
-					numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
+					numSinkIOWorkers(serverCfg, opts), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 					metricsBuilder, serverCfg.Settings)
 			})
 		case isPubsubSink(u):
@@ -292,7 +292,7 @@ func getSink(
 				testingKnobs = knobs
 			}
 			return makePubsubSink(ctx, u, encodingOpts, opts.GetPubsubConfigJSON(), targets,
-				opts.IsSet(changefeedbase.OptUnordered), numSinkIOWorkers(serverCfg),
+				opts.IsSet(changefeedbase.OptUnordered), numSinkIOWorkers(serverCfg, opts),
 				newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 				metricsBuilder, serverCfg.Settings, testingKnobs)
 		case isCloudStorageSink(u):
@@ -446,9 +446,11 @@ type encDatumRowBuffer []rowenc.EncDatumRow
 func (b *encDatumRowBuffer) IsEmpty() bool {
 	return b == nil || len(*b) == 0
 }
+
 func (b *encDatumRowBuffer) Push(r rowenc.EncDatumRow) {
 	*b = append(*b, r)
 }
+
 func (b *encDatumRowBuffer) Pop() rowenc.EncDatumRow {
 	ret := (*b)[0]
 	*b = (*b)[1:]
@@ -745,7 +747,7 @@ func getSinkConfigFromJson(
 ) (batchCfg sinkBatchConfig, retryCfg retry.Options, err error) {
 	retryCfg = defaultRetryConfig()
 
-	var cfg = baseConfig
+	cfg := baseConfig
 	cfg.Retry.Max = jsonMaxRetries(retryCfg.MaxRetries)
 	cfg.Retry.Backoff = jsonDuration(retryCfg.InitialBackoff)
 	if jsonStr != `` {
@@ -802,8 +804,38 @@ func (j *jsonMaxRetries) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
-func numSinkIOWorkers(cfg *execinfra.ServerConfig) int {
-	numWorkers := changefeedbase.SinkIOWorkers.Get(&cfg.Settings.SV)
+func numSinkIOWorkers(cfg *execinfra.ServerConfig, opts changefeedbase.StatementOptions) int {
+	// Get per-changefeed option
+	changefeedWorkers, err := opts.GetNumSinkWorkers()
+	if err != nil {
+		// Log error but continue with cluster setting
+		log.Changefeed.Warningf(context.Background(), "error getting num_sink_workers option: %v", err)
+		changefeedWorkers = 0
+	}
+
+	// Get cluster setting
+	clusterWorkers := changefeedbase.SinkIOWorkers.Get(&cfg.Settings.SV)
+
+	// Apply precedence logic:
+	// 1. If both are positive, use the smaller value (cluster can cap)
+	// 2. If one is positive and the other is <=0, use the positive value
+	// 3. If both are <=0, use automatic default based on GOMAXPROCS
+
+	var numWorkers int64
+	if changefeedWorkers > 0 && clusterWorkers > 0 {
+		if changefeedWorkers < clusterWorkers {
+			numWorkers = changefeedWorkers
+		} else {
+			numWorkers = clusterWorkers
+		}
+	} else if changefeedWorkers > 0 {
+		numWorkers = changefeedWorkers
+	} else if clusterWorkers > 0 {
+		numWorkers = clusterWorkers
+	} else {
+		numWorkers = 0
+	}
+
 	if numWorkers > 0 {
 		return int(numWorkers)
 	}

pkg/ccl/changefeedccl/sink_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/kvevent"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/testutils/pgurlutils"
@@ -1014,3 +1015,101 @@ func TestChangefeedConsistentPartitioning(t *testing.T) {
 	}
 
 }
+
+func TestNumSinkIOWorkers(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	s := serverutils.StartServerOnly(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(ctx)
+
+	execCfg := s.ExecutorConfig().(sql.ExecutorConfig)
+
+	tests := []struct {
+		name             string
+		clusterSetting   int64
+		changefeedOption string // empty means not set
+		expectedWorkers  int
+		expectUseDefault bool // true if we expect GOMAXPROCS-based default
+	}{
+		{
+			name:             "both zero - use default",
+			clusterSetting:   0,
+			changefeedOption: "0",
+			expectUseDefault: true,
+		},
+		{
+			name:             "cluster positive, option not set - use cluster",
+			clusterSetting:   10,
+			changefeedOption: "",
+			expectedWorkers:  10,
+		},
+		{
+			name:             "cluster zero, option positive - use option",
+			clusterSetting:   0,
+			changefeedOption: "15",
+			expectedWorkers:  15,
+		},
+		{
+			name:             "both positive, option smaller - use option",
+			clusterSetting:   10,
+			changefeedOption: "5",
+			expectedWorkers:  5,
+		},
+		{
+			name:             "both positive, cluster smaller - use cluster",
+			clusterSetting:   10,
+			changefeedOption: "15",
+			expectedWorkers:  10,
+		},
+		{
+			name:             "cluster negative - use default",
+			clusterSetting:   -1,
+			changefeedOption: "",
+			expectUseDefault: true,
+		},
+		{
+			name:             "option negative - use default",
+			clusterSetting:   0,
+			changefeedOption: "-1",
+			expectUseDefault: true,
+		},
+		{
+			name:             "both negative - use default",
+			clusterSetting:   -1,
+			changefeedOption: "-1",
+			expectUseDefault: true,
+		},
+		{
+			name:             "option not set, cluster zero - use default",
+			clusterSetting:   0,
+			changefeedOption: "",
+			expectUseDefault: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Set cluster setting
+			changefeedbase.SinkIOWorkers.Override(ctx, &execCfg.Settings.SV, test.clusterSetting)
+
+			// Create statement options
+			optsMap := make(map[string]string)
+			if test.changefeedOption != "" {
+				optsMap[changefeedbase.OptNumSinkWorkers] = test.changefeedOption
+			}
+			opts := changefeedbase.MakeStatementOptions(optsMap)
+
+			result := numSinkIOWorkers(&execCfg.DistSQLSrv.ServerConfig, opts)
+
+			if test.expectUseDefault {
+				// Should be between 1 and 32 (GOMAXPROCS-based)
+				require.GreaterOrEqual(t, result, 1)
+				require.LessOrEqual(t, result, 32)
+			} else {
+				require.Equal(t, test.expectedWorkers, result)
+			}
+		})
+	}
+}