feat(sink): decouple starrocks commit from risingwave commit (#16816)

Co-authored-by: xxhZs <[email protected]> Co-authored-by: Xinhao Xu <[email protected]> Co-authored-by: TennyZhuang <[email protected]>
risingwavelabs · Jun 5, 2024 · e01eb2e · e01eb2e
1 parent 030f2fa
commit e01eb2e
Show file tree

Hide file tree

Showing 5 changed files with 733 additions and 129 deletions.
diff --git a/src/connector/src/sink/decouple_checkpoint_log_sink.rs b/src/connector/src/sink/decouple_checkpoint_log_sink.rs
@@ -21,6 +21,9 @@ use crate::sink::log_store::{LogStoreReadItem, TruncateOffset};
 use crate::sink::writer::SinkWriter;
 use crate::sink::{LogSinker, Result, SinkLogReader, SinkMetrics};
 
+/// The `LogSinker` implementation used for commit-decoupled sinks (such as `Iceberg`, `DeltaLake` and `StarRocks`).
+/// The concurrent/frequent commit capability of these sinks is poor, so by leveraging the decoupled log reader,
+/// we delay the checkpoint barrier to make commits less frequent.
 pub struct DecoupleCheckpointLogSinkerOf<W> {
  writer: W,
  sink_metrics: SinkMetrics,

diff --git a/src/connector/src/sink/deltalake.rs b/src/connector/src/sink/deltalake.rs
@@ -69,7 +69,7 @@ pub struct DeltaLakeCommon {
  pub s3_endpoint: Option<String>,
  #[serde(rename = "gcs.service.account")]
  pub gcs_service_account: Option<String>,
- // Commit every n(>0) checkpoints, if n is not set, we will commit every checkpoint.
+ /// Commit every n(>0) checkpoints, if n is not set, we will commit every checkpoint.
  #[serde(default, deserialize_with = "deserialize_optional_u64_from_string")]
  pub commit_checkpoint_interval: Option<u64>,
 }
@@ -295,7 +295,7 @@ impl Sink for DeltaLakeSink {
  SinkDecouple::Disable => {
  if config_decouple {
  return Err(SinkError::Config(anyhow!(
- "config conflict: DeltaLake config `commit_checkpoint_interval` bigger than 1 which means that must enable sink decouple, but session config sink decouple is disabled"
+ "config conflict: DeltaLake config `commit_checkpoint_interval` larger than 1 means that sink decouple must be enabled, but session config sink_decouple is disabled"
  )));
  }
  Ok(false)