feat(sink): introduce exactly once iceberg sink

Cargo.toml

@@ -151,12 +151,12 @@ otlp-embedded = { git = "https://github.com/risingwavelabs/otlp-embedded", rev =
 prost = { version = "0.13" }
 prost-build = { version = "0.13" }
 # branch dev_rebase_main_20250325
-iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "c9da916ab29b13adcb137ba40bad9e2dc10309c4", features = [
+iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "14c3387c388f64e99dd2a811cdd5f554fe7680c3", features = [
     "storage-s3",
     "storage-gcs",
 ] }
-iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "c9da916ab29b13adcb137ba40bad9e2dc10309c4" }
-iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "c9da916ab29b13adcb137ba40bad9e2dc10309c4" }
+iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "14c3387c388f64e99dd2a811cdd5f554fe7680c3" }
+iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "14c3387c388f64e99dd2a811cdd5f554fe7680c3" }
 opendal = "0.49"
 arrow-udf-js = "0.6.1"
 arrow-udf-wasm = { version = "0.5.1", features = ["build"] }

src/bench/Cargo.toml

@@ -24,6 +24,7 @@ plotters = { version = "0.3.5", default-features = false, features = [
 risingwave_common = { workspace = true }
 risingwave_connector = { workspace = true }
 risingwave_stream = { workspace = true }
+sea-orm  = { workspace = true }
 serde = { version = "1", features = ["derive"] }
 serde_yaml = "0.9"
 thiserror-ext = { workspace = true }

src/bench/sink_bench/main.rs

@@ -64,6 +64,7 @@ use risingwave_connector::source::{
 };
 use risingwave_stream::executor::test_utils::prelude::ColumnDesc;
 use risingwave_stream::executor::{Barrier, Message, MessageStreamItem, StreamExecutorError};
+use sea_orm::DatabaseConnection;
 use serde::{Deserialize, Deserializer};
 use thiserror_ext::AsReport;
 use tokio::sync::oneshot::Sender;
@@ -382,7 +383,7 @@ where
     <S as risingwave_connector::sink::Sink>::Coordinator: std::marker::Send,
     <S as risingwave_connector::sink::Sink>::Coordinator: 'static,
 {
-    if let Ok(coordinator) = sink.new_coordinator().await {
+    if let Ok(coordinator) = sink.new_coordinator(DatabaseConnection::Disconnected).await {
         sink_writer_param.meta_client = Some(SinkMetaClient::MockMetaClient(MockMetaClient::new(
             Box::new(coordinator),
         )));

src/connector/Cargo.toml

@@ -111,6 +111,7 @@ risingwave_common_estimate_size = { workspace = true }
 risingwave_common_rate_limit = { workspace = true }
 risingwave_connector_codec = { workspace = true }
 risingwave_jni_core = { workspace = true }
+risingwave_meta_model = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
 rumqttc = { version = "0.24.0", features = ["url"] }
@@ -120,6 +121,7 @@ rustls-pemfile = "2"
 rustls-pki-types = "1"
 rw_futures_util = { workspace = true }
 scopeguard = "1"
+sea-orm = { workspace = true }
 sea-schema = { version = "0.16", default-features = false, features = [
     "discovery",
     "sqlx-postgres",

src/connector/src/sink/boxed.rs

@@ -20,6 +20,7 @@ use risingwave_common::array::StreamChunk;
 use risingwave_common::bitmap::Bitmap;
 use risingwave_pb::connector_service::SinkMetadata;
 
+use super::SinkCommittedEpochSubscriber;
 use crate::sink::{SinkCommitCoordinator, SinkWriter};
 
 pub type BoxWriter<CM> = Box<dyn SinkWriter<CommitMetadata = CM> + Send + 'static>;
@@ -52,8 +53,11 @@ impl<CM: 'static + Send> SinkWriter for BoxWriter<CM> {
 
 #[async_trait]
 impl SinkCommitCoordinator for BoxCoordinator {
-    async fn init(&mut self) -> crate::sink::Result<Option<u64>> {
-        self.deref_mut().init().await
+    async fn init(
+        &mut self,
+        subscriber: SinkCommittedEpochSubscriber,
+    ) -> crate::sink::Result<Option<u64>> {
+        self.deref_mut().init(subscriber).await
     }
 
     async fn commit(&mut self, epoch: u64, metadata: Vec<SinkMetadata>) -> crate::sink::Result<()> {

src/connector/src/sink/deltalake.rs

@@ -37,6 +37,7 @@ use risingwave_common::util::iter_util::ZipEqDebug;
 use risingwave_pb::connector_service::SinkMetadata;
 use risingwave_pb::connector_service::sink_metadata::Metadata::Serialized;
 use risingwave_pb::connector_service::sink_metadata::SerializedMetadata;
+use sea_orm::DatabaseConnection;
 use serde_derive::{Deserialize, Serialize};
 use serde_with::{DisplayFromStr, serde_as};
 use with_options::WithOptions;
@@ -48,7 +49,7 @@ use super::decouple_checkpoint_log_sink::{
 use super::writer::SinkWriter;
 use super::{
     Result, SINK_TYPE_APPEND_ONLY, SINK_USER_FORCE_APPEND_ONLY_OPTION, Sink, SinkCommitCoordinator,
-    SinkError, SinkParam, SinkWriterMetrics, SinkWriterParam,
+    SinkCommittedEpochSubscriber, SinkError, SinkParam, SinkWriterMetrics, SinkWriterParam,
 };
 use crate::connector_common::AwsAuthProps;
 
@@ -384,7 +385,7 @@ impl Sink for DeltaLakeSink {
         true
     }
 
-    async fn new_coordinator(&self) -> Result<Self::Coordinator> {
+    async fn new_coordinator(&self, _db: DatabaseConnection) -> Result<Self::Coordinator> {
         Ok(DeltaLakeSinkCommitter {
             table: self.config.common.create_deltalake_client().await?,
         })
@@ -496,7 +497,7 @@ pub struct DeltaLakeSinkCommitter {
 
 #[async_trait::async_trait]
 impl SinkCommitCoordinator for DeltaLakeSinkCommitter {
-    async fn init(&mut self) -> crate::sink::Result<Option<u64>> {
+    async fn init(&mut self, _subscriber: SinkCommittedEpochSubscriber) -> Result<Option<u64>> {
         tracing::info!("DeltaLake commit coordinator inited.");
         Ok(None)
     }