changefeedccl: add sink_io_workers option to CREATE statement

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 changefeed
value will be given precedence.

Release note (sql change): The CREATE CHANGEFEED statement was extended
with an optional `io_sink_workers` setting that can be used to set the
number of SinkIOWorkers per changefeed.

Fixes: #154546

Author: fabschn

pkg/ccl/changefeedccl/changefeed_test.go

@@ -21,6 +21,7 @@ import (
 	"path"
 	"reflect"
 	"regexp"
+	"runtime"
 	"slices"
 	"sort"
 	"strconv"
@@ -13084,3 +13085,117 @@ func TestDatabaseLevelChangefeedWithInitialScanOptions(t *testing.T) {
 		})
 	}
 }
+
+// TestChangefeedNumSinkWorkersPrecedence verifies that the sink_io_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)`)
+		KafkaV2Enabled.Override(context.Background(), &s.Server.ClusterSettings().SV, true)
+
+		t.Run("uses per-changefeed option when cluster setting is zero", func(t *testing.T) {
+			sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 0`)
+			cf := feed(t, f, `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '5'`)
+			defer closeFeed(t, cf)
+
+			testutils.SucceedsSoon(t, func() error {
+				workers := metrics.ParallelIOWorkers.Value()
+				if workers != 5 {
+					return errors.Newf("expected 5 workers, got %d", workers)
+				}
+				return nil
+			})
+		})
+
+		t.Run("uses per-changefeed option when it is set", func(t *testing.T) {
+			sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 10`)
+			cf := feed(t, f, `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '3'`)
+			defer closeFeed(t, cf)
+
+			testutils.SucceedsSoon(t, func() error {
+				workers := metrics.ParallelIOWorkers.Value()
+				if workers != 3 {
+					return errors.Newf("expected 3 workers (from changefeed setting), got %d", workers)
+				}
+				return nil
+			})
+		})
+
+		t.Run("uses cluster setting when per-changefeed option is not set", func(t *testing.T) {
+			sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 7`)
+			cf := feed(t, f, `CREATE CHANGEFEED FOR foo`)
+			defer closeFeed(t, cf)
+
+			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
+			})
+		})
+
+		t.Run("uses GOMAXPROCS default when changefeed setting is negative", func(t *testing.T) {
+			sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 10`)
+			cf := feed(t, f, `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '-1'`)
+			defer closeFeed(t, cf)
+
+			testutils.SucceedsSoon(t, func() error {
+				workers := metrics.ParallelIOWorkers.Value()
+				expectedWorkers := runtime.GOMAXPROCS(0)
+				if workers != int64(expectedWorkers) {
+					return errors.Newf("expected %d workers (GOMAXPROCS-based), got %d", expectedWorkers, workers)
+				}
+				return nil
+			})
+		})
+
+		t.Run("uses GOMAXPROCS default when cluster setting is negative and no changefeed option is set", func(t *testing.T) {
+			sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = -1`)
+			cf := feed(t, f, `CREATE CHANGEFEED FOR foo`)
+			defer closeFeed(t, cf)
+
+			testutils.SucceedsSoon(t, func() error {
+				workers := metrics.ParallelIOWorkers.Value()
+				expectedWorkers := runtime.GOMAXPROCS(0)
+				if workers != int64(expectedWorkers) {
+					return errors.Newf("expected %d workers (GOMAXPROCS-based), got %d", expectedWorkers, workers)
+				}
+				return nil
+			})
+		})
+
+		t.Run("uses GOMAXPROCS default when both sink_io_workers settings are negative", func(t *testing.T) {
+			sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = -1`)
+			cf := feed(t, f, `CREATE CHANGEFEED FOR foo WITH sink_io_workers = '-1'`)
+			defer closeFeed(t, cf)
+
+			testutils.SucceedsSoon(t, func() error {
+				workers := metrics.ParallelIOWorkers.Value()
+				expectedWorkers := runtime.GOMAXPROCS(0)
+				if workers != int64(expectedWorkers) {
+					return errors.Newf("expected %d workers (GOMAXPROCS-based), got %d", expectedWorkers, workers)
+				}
+				return nil
+			})
+		})
+
+		t.Run("rejects invalid sink_io_workers value", func(t *testing.T) {
+			sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.sink_io_workers = 5`)
+			sqlDB.ExpectErrWithTimeout(t, "invalid integer value",
+				`CREATE CHANGEFEED FOR foo WITH sink_io_workers = 'invalid'`)
+		})
+	}
+
+	// Test with sinks that support parallel IO.
+	cdcTest(t, testFn, feedTestRestrictSinks("pubsub", "webhook", "kafka"))
+}

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"strconv"
 	"strings"
 	"time"
 
@@ -132,6 +133,7 @@ const (
 	OptLaggingRangesPollingInterval       = `lagging_ranges_polling_interval`
 	OptIgnoreDisableChangefeedReplication = `ignore_disable_changefeed_replication`
 	OptEncodeJSONValueNullAsObject        = `encode_json_value_null_as_object`
+	OptSinkIOWorkers                      = `sink_io_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`
@@ -430,6 +432,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptLaggingRangesPollingInterval:       durationOption,
 	OptIgnoreDisableChangefeedReplication: flagOption,
 	OptEncodeJSONValueNullAsObject:        flagOption,
+	OptSinkIOWorkers:                      stringOption,
 	OptEnrichedProperties:                 csv(string(EnrichedPropertySource), string(EnrichedPropertySchema)),
 	OptRangeDistributionStrategy:          enum(string(ChangefeedRangeDistributionStrategyDefault), string(ChangefeedRangeDistributionStrategyBalancedSimple)),
 	OptHeadersJSONColumnName:              stringOption,
@@ -449,6 +452,7 @@ var CommonOptions = makeStringSet(OptCursor, OptEndTime, OptEnvelope,
 	OptExecutionLocality, OptLaggingRangesThreshold, OptLaggingRangesPollingInterval,
 	OptIgnoreDisableChangefeedReplication, OptEncodeJSONValueNullAsObject, OptEnrichedProperties,
 	OptRangeDistributionStrategy,
+	OptSinkIOWorkers,
 )
 
 // SQLValidOptions is options exclusive to SQL sink
@@ -1235,6 +1239,21 @@ func (s StatementOptions) GetPTSExpiration() (time.Duration, error) {
 	return *exp, nil
 }
 
+// GetNumSinkWorkers returns the number of sink IO workers to use.
+// Returns nil if not set.
+// Returns an error if the set value is not a valid integer.
+func (s StatementOptions) GetNumSinkWorkers() (*int64, error) {
+	v, ok := s.m[OptSinkIOWorkers]
+	if !ok {
+		return nil, nil
+	}
+	result, err := strconv.ParseInt(v, 10, 64)
+	if err != nil {
+		return nil, errors.Newf("invalid integer value for %s: %q", OptSinkIOWorkers, v)
+	}
+	return &result, nil
+}
+
 // ForceKeyInValue sets the encoding option KeyInValue to true and then validates the
 // resoluting encoding options.
 func (s StatementOptions) ForceKeyInValue() error {
@@ -1377,6 +1396,10 @@ func (s StatementOptions) ValidateForCreateChangefeed(isPredicateChangefeed bool
 			}
 		}
 	}
+	// Validate that sink_io_workers is an integer value if it is set.
+	if _, err := s.GetNumSinkWorkers(); err != nil {
+		return err
+	}
 	return nil
 }
 

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{"sink_io_workers": "5"},
+			expected:  5,
+			expectErr: false,
+		},
+		{
+			name:      "zero value (default)",
+			input:     map[string]string{"sink_io_workers": "0"},
+			expected:  0,
+			expectErr: false,
+		},
+		{
+			name:      "negative value (disable)",
+			input:     map[string]string{"sink_io_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{"sink_io_workers": "abc"},
+			expected:  0,
+			expectErr: true,
+		},
+		{
+			name:      "invalid float",
+			input:     map[string]string{"sink_io_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

@@ -258,8 +258,12 @@ func getSink(
 					return nil, err
 				}
 				if KafkaV2Enabled.Get(&serverCfg.Settings.SV) {
+					numWorkers, err := numSinkIOWorkers(serverCfg, opts)
+					if err != nil {
+						return nil, err
+					}
 					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, targets, sinkOpts,
-						numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
+						numWorkers, newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 						serverCfg.Settings, metricsBuilder, kafkaSinkV2Knobs{})
 				} else {
 					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, targets, sinkOpts, serverCfg.Settings, metricsBuilder)
@@ -282,17 +286,25 @@ func getSink(
 				return nil, err
 			}
 			return validateOptionsAndMakeSink(changefeedbase.WebhookValidOptions, func() (Sink, error) {
+				numWorkers, err := numSinkIOWorkers(serverCfg, opts)
+				if err != nil {
+					return nil, err
+				}
 				return makeWebhookSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, webhookOpts,
-					numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
+					numWorkers, newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 					metricsBuilder, serverCfg.Settings)
 			})
 		case isPubsubSink(u):
 			var testingKnobs *TestingKnobs
 			if knobs, ok := serverCfg.TestingKnobs.Changefeed.(*TestingKnobs); ok {
 				testingKnobs = knobs
 			}
+			numWorkers, err := numSinkIOWorkers(serverCfg, opts)
+			if err != nil {
+				return nil, err
+			}
 			return makePubsubSink(ctx, u, encodingOpts, opts.GetPubsubConfigJSON(), targets,
-				opts.IsSet(changefeedbase.OptUnordered), numSinkIOWorkers(serverCfg),
+				opts.IsSet(changefeedbase.OptUnordered), numWorkers,
 				newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 				metricsBuilder, serverCfg.Settings, testingKnobs)
 		case isCloudStorageSink(u):
@@ -446,9 +458,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 +759,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,20 +816,25 @@ func (j *jsonMaxRetries) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
-func numSinkIOWorkers(cfg *execinfra.ServerConfig) int {
-	numWorkers := changefeedbase.SinkIOWorkers.Get(&cfg.Settings.SV)
-	if numWorkers > 0 {
-		return int(numWorkers)
+func numSinkIOWorkers(
+	cfg *execinfra.ServerConfig, opts changefeedbase.StatementOptions,
+) (int, error) {
+	changefeedWorkers, err := opts.GetNumSinkWorkers()
+	if err != nil {
+		return 0, errors.Wrap(err, "Invalid sink config. num_sink_workers must be an integer")
 	}
 
-	idealNumber := runtime.GOMAXPROCS(0)
-	if idealNumber < 1 {
-		return 1
+	clusterWorkers := changefeedbase.SinkIOWorkers.Get(&cfg.Settings.SV)
+
+	value := int(clusterWorkers)
+	if changefeedWorkers != nil {
+		value = int(*changefeedWorkers)
 	}
-	if idealNumber > 32 {
-		return 32
+	if value <= 0 {
+		value = min(runtime.GOMAXPROCS(0), 32)
 	}
-	return idealNumber
+
+	return value, nil
 }
 
 func newCPUPacerFactory(ctx context.Context, cfg *execinfra.ServerConfig) func() *admission.Pacer {