ccl/changefeedccl: add compression options for webhook sink #138872

massimo-ua · 2025-01-11T15:01:24Z

Added gzip compression option to the changefeed webhook sink to reduce both network bytes and storage bytes for the customer, leading to cost savings and potentially better performance. A minimum configuration to support for gzip.

Jira issue: CRDB-42915

Epic CRDB-39392

Issue #132279

blathers-crl · 2025-01-11T15:01:28Z

It looks like your PR touches production code but doesn't add or edit any test code. Did you consider adding tests to your PR?

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

Before a member of our team reviews your PR, I have some potential action items for you:

Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

cockroach-teamcity · 2025-01-11T15:01:29Z

All committers have signed the CLA.

cockroach-teamcity · 2025-01-11T15:01:33Z

This change is

blathers-crl · 2025-01-11T22:52:04Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

blathers-crl · 2025-01-11T23:16:17Z

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.}

asg0451

Thanks for the contribution @massimo-ua! I have a few nits and a few questions, but this looks very good overall.

pkg/ccl/changefeedccl/sink_webhook_v2.go

pkg/ccl/changefeedccl/sink_webhook_test.go

blathers-crl · 2025-01-15T12:23:58Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

blathers-crl · 2025-01-15T13:02:31Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

pkg/ccl/changefeedccl/changefeed_test.go

asg0451 · 2025-01-15T15:06:18Z

pkg/ccl/changefeedccl/compression.go

+		// since we are using decompression only for reading error response body, we can use default reader
+		return pgzip.NewReader(src)
+	case sinkCompressionZstd:
+		// zstd reader does not implement io.Closer interface, so we need to wrap it


The zstd reader does implement Close(), which we should call.

It may also be worth caching the readers instead of making a new one every time (see my next comment on doing the same for writers)... but that might be unnecessary for this case.

zstd reader implements func (d *[Decoder] Close() void
but io.ReadCloser requires
func (io.Closer) Close() error with error
that's why I had to wrap it

Yes, but your wrapper does not call the void Close method, where it should.

pkg/ccl/changefeedccl/sink_webhook_v2.go

massimo-ua · 2025-01-15T16:12:31Z

@asg0451 Thanks for the review. I'll work through the comments and update you once I'm done.

blathers-crl · 2025-01-15T19:29:51Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

blathers-crl · 2025-01-15T19:44:27Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

blathers-crl · 2025-01-16T23:17:27Z

Your pull request contains more than 1000 changes. It is strongly encouraged to split big PRs into smaller chunks.

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

blathers-crl · 2025-01-17T17:23:57Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

blathers-crl · 2025-01-18T01:17:51Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

massimo-ua · 2025-01-23T15:02:57Z

@asg0451 Please review recent changes
Updated encoder/decoders pool implementation

pkg/ccl/changefeedccl/compression.go

asg0451 · 2025-01-23T19:25:43Z

pkg/ccl/changefeedccl/sink_cloudstorage.go

@@ -887,6 +887,7 @@ func (f *cloudStorageSinkFile) flushToStorage(
 		if err := f.codec.Close(); err != nil {
 			return err
 		}
+		f.codec = nil // Set to nil to prevent double close


what's the reasoning behind this? did this issue come up in the tests? maybe it's worthwhile to make the wrappers more robust to double closing if this is an issue

Yes, this came up during the tests. TestCloudStorageSinkFastGzip was panicking with "invalid memory address or nil pointer dereference" while attempting to invoke (*encWrapper).Close. This helped to fix the issue. As far as I understand, the issue is with cloudStorageSinkFile and its codec reference management. It closes the writer but doesn't reset the reference.

Need some guidance here. Not sure we can address that on an encWrapper level

Maybe you could have a closed bool flag in the wrappers, set it on Close, and panic/err if it's used while closed. Reset could reset it. This could also prevent incorrect usage (like using it after closing it, which would be bad)

How do you feel about returning nil in case somebody is trying to close a closed wrapper but return error when Write or Read is called?
Returning nil on multiple attempts to Close the same reader is a default behaviour for gzip Writer and Reader

Added bool flag to indicate the underlying encoder/decoder is closed

that sounds good

pkg/ccl/changefeedccl/sink_webhook_v2.go

asg0451 · 2025-01-24T15:52:27Z

pkg/ccl/changefeedccl/compression.go

+// is recycled after closing rather than being discarded.
+func (e *encWrapper) Close() error {
+	var err error
+	if fErr := e.encoder.Flush(); fErr != nil {


Why do we need to call Flush here. Did we do it before? is Close not enough?

You're right. There is no need to call flush since according to the docs close flushes unwritten data to reader. I'll remove it

pkg/ccl/changefeedccl/compression.go

asg0451 · 2025-01-24T15:57:28Z

pkg/ccl/changefeedccl/sink_cloudstorage.go

@@ -887,6 +887,7 @@ func (f *cloudStorageSinkFile) flushToStorage(
 		if err := f.codec.Close(); err != nil {
 			return err
 		}
+		f.codec = nil // Set to nil to prevent double close


Maybe you could have a closed bool flag in the wrappers, set it on Close, and panic/err if it's used while closed. Reset could reset it. This could also prevent incorrect usage (like using it after closing it, which would be bad)

massimo-ua · 2025-01-25T20:50:03Z

I noticed the race issues during Bazel extended CI. I don't believe there is a connection to the changes I've made. The race appears when spanInner.isNoop() and Span.reset() in pkg/utils/tracing are attempting to read and write to a shared resource. Do you think I should ignore this?

asg0451 · 2025-01-27T17:57:46Z

I noticed the race issues during Bazel extended CI. I don't believe there is a connection to the changes I've made. The race appears when spanInner.isNoop() and Span.reset() in pkg/utils/tracing are attempting to read and write to a shared resource. Do you think I should ignore this?

Feel free to ignore bazel extended ci.

massimo-ua · 2025-01-27T18:03:26Z

I noticed the race issues during Bazel extended CI. I don't believe there is a connection to the changes I've made. The race appears when spanInner.isNoop() and Span.reset() in pkg/utils/tracing are attempting to read and write to a shared resource. Do you think I should ignore this?

Feel free to ignore bazel extended ci.

Then that's it. There are no more changes left in this PR. Thank you for the review. Let me know if there is anything

asg0451 · 2025-01-27T20:24:45Z

pkg/ccl/changefeedccl/compression.go

+	Reset(io.Reader) error
+}
+
+// encWrapper wraps an encoder with encoders pool.


nit: phrasing

Suggested change

// encWrapper wraps an encoder with encoders pool.

// encWrapper wraps an encoder and includes a pointer to the pool it's associated with.

asg0451 · 2025-01-27T20:25:52Z

pkg/ccl/changefeedccl/compression.go

+
+func (e *encWrapper) Write(p []byte) (int, error) {
+	if e.closed {
+		return 0, errors.New("attempt to write on a closed encoder, reset it first")


nit: make this error an errors.AssertionFailedf(..). ditto for the similar check in decWrapper.Read

pkg/ccl/changefeedccl/compression.go

asg0451 · 2025-01-27T20:32:26Z

pkg/ccl/changefeedccl/sink_webhook_test.go

+			struct {
+				key   string
+				value string
+			}{


nit: use the same key/value struct for both

asg0451 · 2025-01-27T20:43:07Z

pkg/ccl/changefeedccl/cdctest/mock_webhook_sink.go

+			return errors.Wrap(err, "failed to create gzip reader")
+		}
+		defer putDecoder(gzReader)
+		defer gzReader.Close()


nit: check this error

asg0451 · 2025-01-28T17:37:47Z

Looking really good @massimo-ua. I've asked @andyyang890 to take a look as well since it's a somewhat tricky change.

andyyang890

Looks pretty good to me! I'll defer to @asg0451 for final approval.

andyyang890 · 2025-01-29T22:06:54Z

pkg/ccl/changefeedccl/compression.go

-	),
-	settings.WithPublic)
+var (
+	useFastGzip = settings.RegisterBoolSetting(


nit: don't group this together with the sync pools

andyyang890 · 2025-01-29T22:20:43Z

pkg/ccl/changefeedccl/sink_webhook_v2.go

+func (sc *webhookSinkClient) setRequestHeaders(req *http.Request) {
+	switch sc.format {
+	case changefeedbase.OptFormatJSON:
+		req.Header.Set("Content-Type", applicationTypeJSON)


nit: maybe add a constant for Content-Type too?

asg0451 · 2025-01-31T15:34:32Z

pkg/ccl/changefeedccl/compression.go

+// Close flushes and closes the underlying encoder, resets it, and returns the wrapper to the pool.
+// This differs from the inner encoder's Close by adding pool management - the wrapper
+// is recycled after closing rather than being discarded.
+func (e *encWrapper) Close() error {


reading the cloudstorage code again, i'm wondering if the following can happen:

we make a new cloudStorageSinkFile with a compression codec. we use it and close it

the encWrapper is put back in the pool

it's handed out to somebody else, eg a webhook sink, resetting it and its close flag

due to some race or something the cloudStorageSinkFile writes to the codec again, causing data corruption

even if this can't happen now, is there a way we can prevent this from being possible? eg maybe we store the encoders in the pool and not the wrappers so that a closed encWrapper will always stay closed. maybe we just require that any f.codec.Close() is immediately followed by f.codec = nil. etc

what do you think?

I don’t think that storing the encoders instead of wrappers in the pool will help to resolve the issue with possible concurrent access because the problem is that during the close operation anyone who owns the reference to encoder through reference to wrapper can write to encoder until the boolean flag is set to false. Once the flag is false attempt to write returns error. And it might only happen on close because when clients get encoders from the pool they get reference after reset is applied so nobody can access it before that. It looks to that it can be easily fixed by adding mutex so we can set wrapper to closed state in the critical section, this in turn will enforce anyone who wants to write wait until it is unlocked and they shouldn’t be able to write to underlying code while open->closed transition is happening. Does it make sense to you?

And we won’t need to ask clients to do a f.codec = nil. The client will be getting error on their side so they can decide how to sync codec state on their side in case its okay to have multiple references to the same codec

I'm not sure i follow - how will mutexes help? If i have an encWrapper e and call Close, wait a bit, then call Write again, and in the meantime someone else has gotten it back from the pool. That means that the closed flag will be reset, as has the dest writer, and i'll write to their output, causing corruption. concurrent calls arent required for this to happen.

Ok. I've double-checked the code and yes, it makes total sense. The encWrapper might be put back in the pool while still referenced by cloudStorageSinkFile. Another goroutine could get this wrapper from the pool and reset it.
The original cloudStorageSinkFile might still try to use it. It's highly possible. The case I was trying to describe it's another potential issue when Close is called but concurrent requests might still write to the codec before the closed flag is set to true

blathers-crl · 2025-01-31T21:17:05Z

Thank you for updating your pull request.

Before a member of our team reviews your PR, I have some potential action items for you:

We notice you have more than one commit in your PR. We try break logical changes into separate commits, but commits such as "fix typo" or "address review commits" should be squashed into one commit and pushed with --force
Please ensure your git commit message contains a release note.
When CI has completed, please ensure no errors have appeared.

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

Release note (sql change): Added compression support for changefeed webhook sinks. This reduces network bandwidth and storage usage, improving performance and lowering costs. Users can enable compression by setting the compression=<algorithm> option. Supported algorithms are gzip and zstd. Epic: https://cockroachlabs.atlassian.net/browse/CRDB-39392 Resolved: https://cockroachlabs.atlassian.net/browse/CRDB-42915 Issue: cockroachdb#132279

massimo-ua requested a review from a team as a code owner January 11, 2025 15:01

massimo-ua requested review from rharding6373 and removed request for a team January 11, 2025 15:01

blathers-crl bot added the O-community Originated from the community label Jan 11, 2025

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from 90f7858 to 9ab5e87 Compare January 11, 2025 23:16

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch 3 times, most recently from dc150e3 to 3c59821 Compare January 13, 2025 00:19

wenyihu6 requested a review from a team January 13, 2025 15:46

wenyihu6 added A-cdc Change Data Capture T-cdc labels Jan 13, 2025

asg0451 self-requested a review January 13, 2025 17:55

asg0451 reviewed Jan 14, 2025

View reviewed changes

massimo-ua requested a review from asg0451 January 15, 2025 12:30

asg0451 requested changes Jan 15, 2025

View reviewed changes

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from dace026 to 4866761 Compare January 17, 2025 17:23

massimo-ua requested a review from asg0451 January 18, 2025 08:13

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from cb4fa0f to 505533e Compare January 23, 2025 14:56

massimo-ua requested a review from asg0451 January 23, 2025 15:34

asg0451 reviewed Jan 23, 2025

View reviewed changes

pkg/ccl/changefeedccl/sink_webhook_v2.go Show resolved Hide resolved

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch 2 times, most recently from 2648d6e to 3889c4f Compare January 23, 2025 22:43

asg0451 reviewed Jan 24, 2025

View reviewed changes

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch 3 times, most recently from 230a0d8 to ed1168d Compare January 25, 2025 15:10

massimo-ua requested a review from asg0451 January 25, 2025 15:18

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from ed1168d to 5bbc73a Compare January 25, 2025 17:41

asg0451 reviewed Jan 28, 2025

View reviewed changes

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from 5bbc73a to 4689f9a Compare January 28, 2025 18:26

asg0451 requested a review from andyyang890 January 29, 2025 17:34

andyyang890 reviewed Jan 29, 2025

View reviewed changes

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from 4689f9a to adcf255 Compare January 29, 2025 23:13

massimo-ua requested a review from asg0451 January 30, 2025 20:34

asg0451 reviewed Jan 31, 2025

View reviewed changes

massimo-ua requested a review from asg0451 January 31, 2025 23:27

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from d78e559 to 0acf5d0 Compare February 1, 2025 19:02

massimo-ua force-pushed the add-gzip-compression-to-changefeed-webhook-sink branch from 0acf5d0 to 50a2c2c Compare February 4, 2025 08:09

massimo-ua requested a review from andyyang890 February 4, 2025 18:41

	// encWrapper wraps an encoder with encoders pool.
	// encWrapper wraps an encoder and includes a pointer to the pool it's associated with.

ccl/changefeedccl: add compression options for webhook sink #138872

Are you sure you want to change the base?

ccl/changefeedccl: add compression options for webhook sink #138872

Conversation

massimo-ua commented Jan 11, 2025

blathers-crl bot commented Jan 11, 2025

cockroach-teamcity commented Jan 11, 2025 • edited Loading

cockroach-teamcity commented Jan 11, 2025

blathers-crl bot commented Jan 11, 2025

blathers-crl bot commented Jan 11, 2025

asg0451 left a comment

Choose a reason for hiding this comment

blathers-crl bot commented Jan 15, 2025

blathers-crl bot commented Jan 15, 2025

asg0451 Jan 15, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

massimo-ua commented Jan 15, 2025

blathers-crl bot commented Jan 15, 2025

blathers-crl bot commented Jan 15, 2025

blathers-crl bot commented Jan 16, 2025

blathers-crl bot commented Jan 17, 2025

blathers-crl bot commented Jan 18, 2025

massimo-ua commented Jan 23, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

massimo-ua Jan 24, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

massimo-ua commented Jan 25, 2025 • edited Loading

asg0451 commented Jan 27, 2025

massimo-ua commented Jan 27, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asg0451 commented Jan 28, 2025

andyyang890 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asg0451 Jan 31, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

massimo-ua Jan 31, 2025 • edited Loading

Choose a reason for hiding this comment

blathers-crl bot commented Jan 31, 2025

cockroach-teamcity commented Jan 11, 2025 •

edited

Loading

asg0451 Jan 15, 2025 •

edited

Loading

massimo-ua commented Jan 23, 2025 •

edited

Loading

massimo-ua Jan 24, 2025 •

edited

Loading

massimo-ua commented Jan 25, 2025 •

edited

Loading

andyyang890 left a comment •

edited

Loading

asg0451 Jan 31, 2025 •

edited

Loading

massimo-ua Jan 31, 2025 •

edited

Loading