feat: decouple iceberg commit from risingwave commit #15634

ZENOTME · 2024-03-12T07:51:02Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

complete #13899

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

chenzl25 · 2024-03-15T09:56:28Z

src/connector/src/sink/iceberg/mod.rs

+ fn default_sink_decouple(desc: &SinkDesc) -> bool {
+ if let Some(interval) = desc.properties.get("commit_checkpoint_interval") {
+ interval.parse::<u64>().unwrap_or(0) > 1
+ } else {
+ false
+ }
+ }


As mentioned by @wenym1, when users can set sink_decouple = false, sink will not be decoupled. If users use commit_checkpoint_interval at the same time, it could be problematic and cause a deadlock. Please verify this case.

chenzl25

LGTM. Waiting for @wenym1 to take a look.

wenym1 · 2024-03-19T09:01:38Z

src/connector/src/sink/mod.rs

 pub trait Sink: TryFrom<SinkParam, Error = SinkError> {
 const SINK_NAME: &'static str;
 type LogSinker: LogSinker;
 type Coordinator: SinkCommitCoordinator;

- fn default_sink_decouple(_desc: &SinkDesc) -> bool {
- false
+ /// Infer the decouple mode from the sink properties.


I think we can directly change the default_sink_decouple to fn is_sink_decouple(desc: &SinkDesc, user_specified: SinkDecouple) -> bool and in frontend we can it directly.

For the enum SinkDecouple we can implement have a method like fn is_decouple(&self) -> Option<bool>, and then the default implementation of is_sink_decouple can be user_specified.is_decouple().unwrap_or(false).

If my understanding is correct, the user_specified: SinkDecouple is the session config. fn is_sink_decouple(desc: &SinkDesc, user_specified: SinkDecouple) -> bool determines whether the sink is decoupled. It also will check if there are conflicts between them. 🤔 Maybe it should be fn is_sink_decouple(desc: &SinkDesc, user_specified: SinkDecouple) -> Result<bool>. and return errors when there are conflicts.

I agree with this idea and we can remove the logic here

risingwave/src/frontend/src/optimizer/plan_node/stream_sink.rs

Line 643 in 61c025d

log_store_type: match self.base.ctx().session_ctx().config().sink_decouple() {

.

So that we can determine the decouple logic in one place rather than separate them.

wenym1 · 2024-03-19T09:09:16Z

src/connector/src/sink/iceberg/mod.rs

- } else {
- Ok(None)
- }
+ // Commit every n checkpoints, if n is not set, 0, 1, we will commit every checkpoint.


Any reason to specify this special deserialization method? What the difference to the default one?

config actually will be deserialize from string. So we need to specify a method from str to other type. Like

risingwave/src/connector/src/sink/kafka.rs

Line 217 in 61c025d

pub max_retry_num: u32,

wenym1 · 2024-03-19T09:14:05Z

src/connector/src/sink/writer.rs

@@ -113,13 +113,16 @@ pub trait FormattedSink {
 pub struct LogSinkerOf<W> {
 writer: W,
 sink_metrics: SinkMetrics,
+
+ commit_checkpoint_interval: u64,


We'd better implement this special logic in the iceberg sink writer and leave this shared default implementation unchanged. We can implement a new IcebergLogSinker to consume the log, and then for every commit_checkpoint_interval number of checkpoint barrier we then commit to iceberg and then truncate the log reader.

github-actions

license-eye has totally checked 4950 files.

Valid	Invalid	Ignored	Fixed
2128	1	2821	0

Click to see the invalid file list

src/connector/src/sink/iceberg/log_sink.rs

src/connector/src/sink/iceberg/log_sink.rs

chenzl25

LGTM

wenym1

Rest LGTM

wenym1 · 2024-03-31T14:28:28Z

src/connector/src/sink/iceberg/log_sink.rs

+ };
+ if is_checkpoint {
+ current_checkpoint += 1;
+ if commit_checkpoint_interval <= 1


Can we just assert that the commit_checkpoint_interval must not be 0 and only check current_checkpoint >= commit_checkpoint_interval ? Currently we use unwrap_or_default to get the commit_checkpoint, which will pass 0 when not set. If we use 1 as the default value, we can avoid passing the 0 value.

But we can't prevent the user from setting it as 0 so it means that we should adjust it to 1 if the user sets it to 0. Another concern is that this assumption is not type-safe. We can't prevent others follow this assumption(It's easy to miss), even for now, seems it will not cause something wrong. 🤔

If the user set it to 0, we should return with error in sink validation. And in the code here, we can assert that it must not be error, or return with error if it's 0.

I checked the config and used NonZero internally to guarantee that the interval is bigger than 0 now.

wenym1

Generally LGTM!

Do we have any report on how the checkpoint interval affects the sink performance? How much performance gain can we achieve when we combine the data of multiple checkpoints?

ZENOTME · 2024-04-08T03:04:28Z

Generally LGTM!

Do we have any report on how the checkpoint interval affects the sink performance? How much performance gain can we achieve when we combine the data of multiple checkpoints?

There is no performance report now. But I added the simple test case that can be used to compare the file size using checkpoint decouple.
Before:

After:

I think we can test it later

github-actions bot added the type/feature label Mar 12, 2024

ZENOTME force-pushed the zj/decouple_commit branch from 446f257 to 0d1380d Compare March 12, 2024 07:54

ZENOTME added the ci/run-e2e-iceberg-sink-tests label Mar 12, 2024

ZENOTME requested review from wenym1 and chenzl25 March 15, 2024 05:44

chenzl25 reviewed Mar 15, 2024

View reviewed changes

chenzl25 reviewed Mar 19, 2024

View reviewed changes

wenym1 reviewed Mar 19, 2024

View reviewed changes

ZENOTME mentioned this pull request Mar 27, 2024

refactor(frontend): refine decouple config #15939

Merged

4 tasks

ZENOTME force-pushed the zj/decouple_commit branch from 61c025d to dd0400e Compare March 28, 2024 15:48

github-actions bot reviewed Mar 28, 2024

View reviewed changes

src/connector/src/sink/iceberg/log_sink.rs Show resolved Hide resolved

chenzl25 approved these changes Mar 29, 2024

View reviewed changes

ZENOTME force-pushed the zj/decouple_commit branch from 6f21026 to 995f97f Compare March 29, 2024 12:21

wenym1 reviewed Mar 31, 2024

View reviewed changes

ZENOTME added 3 commits April 1, 2024 20:16

add decouple commit for iceberg

9b1fa5c

add license

0267a38

make commit interval nonzero

adc6317

ZENOTME force-pushed the zj/decouple_commit branch from 995f97f to adc6317 Compare April 1, 2024 12:16

wenym1 approved these changes Apr 7, 2024

View reviewed changes

ZENOTME added this pull request to the merge queue Apr 8, 2024

Merged via the queue into main with commit ff862dc Apr 8, 2024
29 of 30 checks passed

ZENOTME deleted the zj/decouple_commit branch April 8, 2024 06:09

wenym1 mentioned this pull request Apr 9, 2024

feat(sink): implement snowflake sink #15429

Merged

9 tasks

ZENOTME mentioned this pull request May 14, 2024

Decouple iceberg commit from risingwave commit. #13899

Closed

This was referenced Jun 21, 2024

Bug:missing docs of iceberg sink parameter commit_checkpoint_interval risingwavelabs/risingwave-docs#2292

Closed

fix: iceberg sink parameter commit_checkpoint_interval risingwavelabs/risingwave-docs#2293

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: decouple iceberg commit from risingwave commit #15634

feat: decouple iceberg commit from risingwave commit #15634

ZENOTME commented Mar 12, 2024 •

edited by fuyufjh

Loading

chenzl25 Mar 15, 2024

chenzl25 left a comment •

edited

Loading

wenym1 Mar 19, 2024

ZENOTME Mar 21, 2024

wenym1 Mar 19, 2024

ZENOTME Mar 21, 2024

wenym1 Mar 19, 2024

github-actions bot left a comment

chenzl25 left a comment

wenym1 left a comment

wenym1 Mar 31, 2024

ZENOTME Apr 1, 2024

wenym1 Apr 1, 2024

ZENOTME Apr 2, 2024

wenym1 left a comment

ZENOTME commented Apr 8, 2024

feat: decouple iceberg commit from risingwave commit #15634

feat: decouple iceberg commit from risingwave commit #15634

Conversation

ZENOTME commented Mar 12, 2024 • edited by fuyufjh Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

Choose a reason for hiding this comment

chenzl25 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

Choose a reason for hiding this comment

github-actions bot left a comment

Choose a reason for hiding this comment

chenzl25 left a comment

Choose a reason for hiding this comment

wenym1 left a 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

wenym1 left a comment

Choose a reason for hiding this comment

ZENOTME commented Apr 8, 2024

ZENOTME commented Mar 12, 2024 •

edited by fuyufjh

Loading

chenzl25 left a comment •

edited

Loading