feat(sink): set default_sink_decouple = true for all sink

e2e_test/sink/cassandra_sink.slt

@@ -10,6 +10,9 @@ CREATE TABLE t6 (v1 int primary key, v2 smallint, v3 bigint, v4 real, v5 float,
 statement ok
 CREATE TABLE t7 ("TEST_V1" int primary key, "TEST_V2" int, "TEST_V3" int);
 
+statement ok
+set sink_decouple = false;
+
 statement ok
 CREATE SINK s6
 FROM

e2e_test/sink/clickhouse_sink.slt

@@ -17,6 +17,7 @@ CREATE SINK s6 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3, mv6.v4 as v4,
  clickhouse.password = '',
  clickhouse.database = 'default',
  clickhouse.table='demo_test',
+ commit_checkpoint_interval = 1,
 );
 
 statement ok

e2e_test/sink/elasticsearch/elasticsearch_sink.slt

@@ -1,3 +1,6 @@
+statement ok
+set sink_decouple = false;
+
 statement ok
 CREATE TABLE t7 (
  v1 int primary key,

src/connector/src/lib.rs

@@ -75,27 +75,6 @@ where
  })
 }
 
-pub(crate) fn deserialize_optional_u64_from_string<'de, D>(
- deserializer: D,
-) -> Result<Option<u64>, D::Error>
-where
- D: de::Deserializer<'de>,
-{
- let s: String = de::Deserialize::deserialize(deserializer)?;
- if s.is_empty() {
- Ok(None)
- } else {
- s.parse()
- .map_err(|_| {
- de::Error::invalid_value(
- de::Unexpected::Str(&s),
- &"integer greater than or equal to 0",
- )
- })
- .map(Some)
- }
-}
-
 pub(crate) fn deserialize_optional_string_seq_from_string<'de, D>(
  deserializer: D,
 ) -> std::result::Result<Option<Vec<String>>, D::Error>

src/connector/src/sink/clickhouse.rs

@@ -30,16 +30,18 @@ use risingwave_common::types::{DataType, Decimal, ScalarRefImpl, Serial};
 use serde::ser::{SerializeSeq, SerializeStruct};
 use serde::Serialize;
 use serde_derive::Deserialize;
-use serde_with::serde_as;
+use serde_with::{serde_as, DisplayFromStr};
 use thiserror_ext::AsReport;
 use tonic::async_trait;
 use tracing::warn;
 use with_options::WithOptions;
 
-use super::decouple_checkpoint_log_sink::DecoupleCheckpointLogSinkerOf;
+use super::decouple_checkpoint_log_sink::{
+ default_commit_checkpoint_interval, DecoupleCheckpointLogSinkerOf,
+ DEFAULT_COMMIT_CHECKPOINT_INTERVAL,
+};
 use super::writer::SinkWriter;
 use super::{DummySinkCommitCoordinator, SinkWriterParam};
-use crate::deserialize_optional_u64_from_string;
 use crate::error::ConnectorResult;
 use crate::sink::catalog::desc::SinkDesc;
 use crate::sink::{
@@ -52,6 +54,7 @@ const QUERY_COLUMN: &str =
  "select distinct ?fields from system.columns where database = ? and table = ? order by ?";
 pub const CLICKHOUSE_SINK: &str = "clickhouse";
 
+#[serde_as]
 #[derive(Deserialize, Debug, Clone, WithOptions)]
 pub struct ClickHouseCommon {
  #[serde(rename = "clickhouse.url")]
@@ -66,9 +69,10 @@ pub struct ClickHouseCommon {
  pub table: String,
  #[serde(rename = "clickhouse.delete.column")]
  pub delete_column: Option<String>,
- /// 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>,
+ /// Commit every n(>0) checkpoints, default is 10.
+ #[serde(default = "default_commit_checkpoint_interval")]
+ #[serde_as(as = "DisplayFromStr")]
+ pub commit_checkpoint_interval: u64,
 }
 
 #[allow(clippy::enum_variant_names)]
@@ -494,26 +498,25 @@ impl Sink for ClickHouseSink {
  const SINK_NAME: &'static str = CLICKHOUSE_SINK;
 
  fn is_sink_decouple(desc: &SinkDesc, user_specified: &SinkDecouple) -> Result<bool> {
- let config_decouple = if let Some(interval) =
- desc.properties.get("commit_checkpoint_interval")
- && interval.parse::<u64>().unwrap_or(0) > 1
- {
- true
- } else {
- false
- };
+ let commit_checkpoint_interval =
+ if let Some(interval) = desc.properties.get("commit_checkpoint_interval") {
+  interval
+  .parse::<u64>()
+  .unwrap_or(DEFAULT_COMMIT_CHECKPOINT_INTERVAL)
+  } else {
+  DEFAULT_COMMIT_CHECKPOINT_INTERVAL
+  };
 
  match user_specified {
- SinkDecouple::Default => Ok(config_decouple),
+ SinkDecouple::Default | SinkDecouple::Enable => Ok(true),
  SinkDecouple::Disable => {
- if config_decouple {
+ if commit_checkpoint_interval > 1 {
  return Err(SinkError::Config(anyhow!(
  "config conflict: Clickhouse config `commit_checkpoint_interval` larger than 1 means that sink decouple must be enabled, but session config sink_decouple is disabled"
  )));
  }
  Ok(false)
  }
- SinkDecouple::Enable => Ok(true),
  }
  }
 
@@ -552,9 +555,9 @@ impl Sink for ClickHouseSink {
  self.check_pk_match(&clickhouse_column)?;
  }
 
- if self.config.common.commit_checkpoint_interval == Some(0) {
+ if self.config.common.commit_checkpoint_interval == 0 {
  return Err(SinkError::Config(anyhow!(
- "commit_checkpoint_interval must be greater than 0"
+ "`commit_checkpoint_interval` must be greater than 0"
  )));
  }
  Ok(())
@@ -569,7 +572,7 @@ impl Sink for ClickHouseSink {
  )
  .await?;
  let commit_checkpoint_interval =
- NonZeroU64::new(self.config.common.commit_checkpoint_interval.unwrap_or(1)).expect(
+ NonZeroU64::new(self.config.common.commit_checkpoint_interval).expect(
  "commit_checkpoint_interval should be greater than 0, and it should be checked in config validation",
  );
 

src/connector/src/sink/decouple_checkpoint_log_sink.rs

@@ -20,6 +20,11 @@ use async_trait::async_trait;
 use crate::sink::log_store::{LogStoreReadItem, TruncateOffset};
 use crate::sink::writer::SinkWriter;
 use crate::sink::{LogSinker, Result, SinkLogReader, SinkMetrics};
+pub const DEFAULT_COMMIT_CHECKPOINT_INTERVAL: u64 = 10;
+
+pub fn default_commit_checkpoint_interval() -> u64 {
+ DEFAULT_COMMIT_CHECKPOINT_INTERVAL
+}
 
 /// 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,

src/connector/src/sink/deltalake.rs

@@ -38,23 +38,26 @@ use risingwave_pb::connector_service::sink_metadata::Metadata::Serialized;
 use risingwave_pb::connector_service::sink_metadata::SerializedMetadata;
 use risingwave_pb::connector_service::SinkMetadata;
 use serde_derive::{Deserialize, Serialize};
-use serde_with::serde_as;
+use serde_with::{serde_as, DisplayFromStr};
 use with_options::WithOptions;
 
 use super::catalog::desc::SinkDesc;
 use super::coordinate::CoordinatedSinkWriter;
-use super::decouple_checkpoint_log_sink::DecoupleCheckpointLogSinkerOf;
+use super::decouple_checkpoint_log_sink::{
+ default_commit_checkpoint_interval, DecoupleCheckpointLogSinkerOf,
+ DEFAULT_COMMIT_CHECKPOINT_INTERVAL,
+};
 use super::writer::SinkWriter;
 use super::{
  Result, Sink, SinkCommitCoordinator, SinkError, SinkParam, SinkWriterParam,
  SINK_TYPE_APPEND_ONLY, SINK_USER_FORCE_APPEND_ONLY_OPTION,
 };
-use crate::deserialize_optional_u64_from_string;
 
 pub const DELTALAKE_SINK: &str = "deltalake";
 pub const DEFAULT_REGION: &str = "us-east-1";
 pub const GCS_SERVICE_ACCOUNT: &str = "service_account_key";
 
+#[serde_as]
 #[derive(Deserialize, Serialize, Debug, Clone, WithOptions)]
 pub struct DeltaLakeCommon {
  #[serde(rename = "s3.access.key")]
@@ -69,10 +72,12 @@ 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.
- #[serde(default, deserialize_with = "deserialize_optional_u64_from_string")]
- pub commit_checkpoint_interval: Option<u64>,
+ /// Commit every n(>0) checkpoints, default is 10.
+ #[serde(default = "default_commit_checkpoint_interval")]
+ #[serde_as(as = "DisplayFromStr")]
+ pub commit_checkpoint_interval: u64,
 }
+
 impl DeltaLakeCommon {
  pub async fn create_deltalake_client(&self) -> Result<DeltaTable> {
  let table = match Self::get_table_url(&self.location)? {
@@ -281,26 +286,25 @@ impl Sink for DeltaLakeSink {
  const SINK_NAME: &'static str = DELTALAKE_SINK;
 
  fn is_sink_decouple(desc: &SinkDesc, user_specified: &SinkDecouple) -> Result<bool> {
- let config_decouple = if let Some(interval) =
- desc.properties.get("commit_checkpoint_interval")
- && interval.parse::<u64>().unwrap_or(0) > 1
- {
- true
- } else {
- false
- };
+ let commit_checkpoint_interval =
+ if let Some(interval) = desc.properties.get("commit_checkpoint_interval") {
+  interval
+  .parse::<u64>()
+  .unwrap_or(DEFAULT_COMMIT_CHECKPOINT_INTERVAL)
+  } else {
+  DEFAULT_COMMIT_CHECKPOINT_INTERVAL
+  };
 
  match user_specified {
- SinkDecouple::Default => Ok(config_decouple),
+ SinkDecouple::Default | SinkDecouple::Enable => Ok(true),
  SinkDecouple::Disable => {
- if config_decouple {
+ if commit_checkpoint_interval > 1 {
  return Err(SinkError::Config(anyhow!(
  "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)
  }
- SinkDecouple::Enable => Ok(true),
  }
  }
 
@@ -328,7 +332,7 @@ impl Sink for DeltaLakeSink {
  .await?;
 
  let commit_checkpoint_interval =
- NonZeroU64::new(self.config.common.commit_checkpoint_interval.unwrap_or(1)).expect(
+ NonZeroU64::new(self.config.common.commit_checkpoint_interval).expect(
  "commit_checkpoint_interval should be greater than 0, and it should be checked in config validation",
  );
 
@@ -380,9 +384,9 @@ impl Sink for DeltaLakeSink {
  )));
  }
  }
- if self.config.common.commit_checkpoint_interval == Some(0) {
+ if self.config.common.commit_checkpoint_interval == 0 {
  return Err(SinkError::Config(anyhow!(
- "commit_checkpoint_interval must be greater than 0"
+ "`commit_checkpoint_interval` must be greater than 0"
  )));
  }
  Ok(())

src/connector/src/sink/google_pubsub.rs

@@ -29,13 +29,11 @@ use google_cloud_pubsub::client::{Client, ClientConfig};
 use google_cloud_pubsub::publisher::{Awaiter, Publisher};
 use risingwave_common::array::StreamChunk;
 use risingwave_common::catalog::Schema;
-use risingwave_common::session_config::sink_decouple::SinkDecouple;
 use serde_derive::Deserialize;
 use serde_with::serde_as;
 use tonic::Status;
 use with_options::WithOptions;
 
-use super::catalog::desc::SinkDesc;
 use super::catalog::SinkFormatDesc;
 use super::formatter::SinkFormatterImpl;
 use super::log_store::DeliveryFutureManagerAddFuture;
@@ -114,13 +112,6 @@ impl Sink for GooglePubSubSink {
 
  const SINK_NAME: &'static str = PUBSUB_SINK;
 
- fn is_sink_decouple(_desc: &SinkDesc, user_specified: &SinkDecouple) -> Result<bool> {
- match user_specified {
- SinkDecouple::Default | SinkDecouple::Enable => Ok(true),
- SinkDecouple::Disable => Ok(false),
- }
- }
-
  async fn validate(&self) -> Result<()> {
  if !self.is_append_only {
  return Err(SinkError::GooglePubSub(anyhow!(