changefeedccl: changefeed-specific option for num of sink workers #156260

fabschn · 2025-10-27T07:17:29Z

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

blathers-crl · 2025-10-27T07:17:34Z

Thank you for contributing to CockroachDB. Please ensure you have followed the guidelines for creating a PR.

My owl senses detect your PR is good for review. Please keep an eye out for any test failures in CI.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

cockroach-teamcity · 2025-10-27T07:17:40Z

This change is

cockroachlabs-cla-agent · 2025-10-27T07:17:42Z

All committers have signed the CLA.

aerfrei

Thanks for submitting this! Looks good, left a few comments.

pkg/ccl/changefeedccl/changefeedbase/options.go

pkg/ccl/changefeedccl/sink.go

aerfrei · 2025-10-27T15:07:53Z

pkg/ccl/changefeedccl/sink.go

+	} else if clusterWorkers > 0 {
+		numWorkers = clusterWorkers
+	} else {
+		numWorkers = 0


This looks like it would set numWorkers to 0, even if both the cluster setting and changefeed value were negative.

(Aside: Seems to me that we may not be currently respecting the negative value as documented which I can look at/file an issue for.)

Indeed, the current implementation does not adhere to the documentation:

"the number of workers used by changefeeds when sending requests to the sink "+
"(currently the batching versions of webhook, pubsub, and kafka sinks that are "+
"enabled by changefeed.new__sink_enabled only): <0 disables, 0 assigns "+
"a reasonable default, >0 assigns the setting value",

When a value <0 is provided, a default will be set, see current implementation:

numWorkers := changefeedbase.SinkIOWorkers.Get(&cfg.Settings.SV) if numWorkers > 0 { return int(numWorkers) } idealNumber := runtime.GOMAXPROCS(0) if idealNumber < 1 { return 1 } if idealNumber > 32 { return 32 } return idealNumber

Would you like to tackle this as part of this PR or create a separate issue for it?

I don't want to micromanage here. But I think the max that you are now using here could use a comment. But for this case where one of changefeedWorkers and clusterWorkers is zero and one is negative or both are negative, it may become useful to have the local numWorkers variable maintain that information if the current implementation doesn't respect it. I'd like to see tests for these cases also e.g. what happens if one value is negative and the other positive,

P.S. Out of curiousity, are you using AI to generate this PR?

I don't want to micromanage here.

This is not how interpret this or other comments - and after all, you're the code owner and will maintain this code. Let's make sure that it lives up to your standards.

But could you help me here on what you'd expect? My struggle with comments has been like a red thread through this PR, so I'd appreciate some guidance.

But for this case where one of changefeedWorkers and clusterWorkers is zero and one is negative or both are negative, it may become useful to have the local numWorkers variable maintain that information if the current implementation doesn't respect it.
I'd like to see tests for these cases also e.g. what happens if one value is negative and the other positive,

I added cases for

one value being negative, the other positive -> asserting that the positive value is applied

both values being negative -> asserting that a runtime.GOMAXPROCS based default is applied

Let me know that is according to your expectations - I have a hunch that I'm not really grasping what your intention in the first paragraph of the quote is.

P.S. Out of curiousity, are you using AI to generate this PR?

I used Claude Code to suggest where changes would be needed and where e.g. test cases should be added as I'm largely unfamiliar with the codebase. I had it then add some scaffolding in the form of function definitions and filled in the rest.
Also out of curiosity: what made you ask? Did you react to anything in the code or our conversation that had a "this is AI" flair to it?

blathers-crl · 2025-10-27T20:05:33Z

Thank you for updating your pull request.

My owl senses detect your PR is good for review. Please keep an eye out for any test failures in CI.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

fabschn

Thanks for the swift review! I addressed the comments and amended my initial commit as described in the Submitting your contribution section of the contribution guidelines.

Let me know if you'd like to see any other changes.

aerfrei · 2025-10-28T14:19:47Z

pkg/ccl/changefeedccl/changefeed_test.go

+		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY)`)
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1)`)
+
+		// Test 1: Per-changefeed = 5, cluster setting = 0 -> should use 5.


Nit: These comments also don't adhere to the style guide. There are other comments in the test files (including inline comments) that also do not.

Removed the comments. As these tests are now implemented as subtests, the subtest names replaced the comments. Let me know if you'd like the subtest names to be changed.

aerfrei · 2025-10-28T14:20:00Z

pkg/ccl/changefeedccl/changefeed_test.go

+		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`)


Nit: Please also implement these tests as separate subtests.

aerfrei · 2025-10-28T14:20:31Z

pkg/ccl/changefeedccl/sink.go

+func numSinkIOWorkers(cfg *execinfra.ServerConfig, opts changefeedbase.StatementOptions) int {
+	changefeedWorkers, err := opts.GetNumSinkWorkers()
+	if err != nil {
+		// Log error but continue with cluster setting.


When do we expect errors here? I don't think this is the right behavior to trudge forward with 0.

GetNumSinkWorkers() returns an error in case the user sets num_sink_workers to a value that cannot be parsed as an integer.
I opted for this approach as a value of 0 would lead to a sensible default number of workers while still creating the sink. But you have more context, so please let me know what you'd rather expect to happen.

Would you rather propagate the error and not create the sink?

aerfrei · 2025-10-28T14:27:11Z

pkg/ccl/changefeedccl/sink.go

+	} else if clusterWorkers > 0 {
+		numWorkers = clusterWorkers
+	} else {
+		numWorkers = 0


I don't want to micromanage here. But I think the max that you are now using here could use a comment. But for this case where one of changefeedWorkers and clusterWorkers is zero and one is negative or both are negative, it may become useful to have the local numWorkers variable maintain that information if the current implementation doesn't respect it. I'd like to see tests for these cases also e.g. what happens if one value is negative and the other positive,

P.S. Out of curiousity, are you using AI to generate this PR?

aerfrei · 2025-10-28T14:28:44Z

pkg/ccl/changefeedccl/sink_test.go

+
+			result := numSinkIOWorkers(&execCfg.DistSQLSrv.ServerConfig, opts)
+
+			if test.expectUseDefault {


In this case we should check that we're using the GOMAXPROCS-based default value exactly, not just that the result falls in a range from 1 to 32, since most of the test values fall in that range anyway.

aerfrei · 2025-10-28T14:30:10Z

pkg/ccl/changefeedccl/changefeed_test.go

+		require.NoError(t, foo4.Close())
+	}
+
+	// Test with sinks that support parallel IO.


Is pubsub the only sink that supports parallel IO?

Afaiu kafka, webhook and pubsub support parallel IO (my assumption is based on numSinkIOWorkers being called during the creation of these three types of sink).

Let me know what the intention of your question is:

For now, I updated the comment to reflect that we're testing with "...a sink that supports...".
Alternatively, it can of course be tested with all sinks that support parallel IO if you think that this provides additional value.

aerfrei · 2025-10-28T14:31:59Z

pkg/ccl/changefeedccl/sink.go

+	clusterWorkers := changefeedbase.SinkIOWorkers.Get(&cfg.Settings.SV)
+
+	// Apply precedence logic:
+	// 1. If both are positive, use the smaller value (cluster can cap)


Nit: To be clear, I think this comment is also reading the code out loud. If you are to include, please write in full sentences (it took me a sec to parse "cluster can cap". With periods and all that.

Removed the comment.

aerfrei · 2025-10-28T14:35:06Z

@fabschn please also take a look at the extended CI failure. It may be unrelated but if it is, please verify that and leave a comment here. Thanks so much!

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: cockroachdb#154546

fabschn · 2025-10-30T19:32:42Z

@fabschn please also take a look at the extended CI failure. It may be unrelated but if it is, please verify that and leave a comment here. Thanks so much!

I can't see details of the Bazel Exended CI job as I don't have access to TeamCity. Let me know if you can and notice that my changes have introduced an issue.

The Claude Code PR Review job fails with the following error:

Checking permissions for actor: fabschn
Permission level retrieved: read
Warning: Actor has insufficient permissions: read
Error: Prepare step failed with error: Actor does not have write permissions to the repository
Error: Process completed with exit code 1.

fabschn requested a review from a team as a code owner October 27, 2025 07:17

fabschn requested review from rharding6373 and removed request for a team October 27, 2025 07:17

blathers-crl bot added the O-community Originated from the community label Oct 27, 2025

asg0451 requested review from aerfrei and removed request for rharding6373 October 27, 2025 13:56

aerfrei reviewed Oct 27, 2025

View reviewed changes

fabschn force-pushed the changefeed-num-sink-workers branch from 3a810a8 to 7213958 Compare October 27, 2025 20:05

fabschn commented Oct 27, 2025

View reviewed changes

aerfrei requested changes Oct 28, 2025

View reviewed changes

fabschn force-pushed the changefeed-num-sink-workers branch from 7213958 to d035bed Compare October 30, 2025 18:48


		result := numSinkIOWorkers(&execCfg.DistSQLSrv.ServerConfig, opts)

		if test.expectUseDefault {

changefeedccl: changefeed-specific option for num of sink workers #156260

Are you sure you want to change the base?

changefeedccl: changefeed-specific option for num of sink workers #156260

Conversation

fabschn commented Oct 27, 2025

Uh oh!

blathers-crl bot commented Oct 27, 2025

Uh oh!

cockroach-teamcity commented Oct 27, 2025

Uh oh!

cockroachlabs-cla-agent bot commented Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aerfrei left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

blathers-crl bot commented Oct 27, 2025

Uh oh!

fabschn left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aerfrei commented Oct 28, 2025

Uh oh!

fabschn commented Oct 30, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

cockroachlabs-cla-agent bot commented Oct 27, 2025 •

edited

Loading