risingwave_connector/sink/
mod.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15feature_gated_sink_mod!(big_query, "bigquery");
16pub mod batching_log_sink;
17pub mod boxed;
18pub mod catalog;
19feature_gated_sink_mod!(clickhouse, ClickHouse, "clickhouse");
20pub mod coordinate;
21pub mod decouple_checkpoint_log_sink;
22feature_gated_sink_mod!(deltalake, DeltaLake, "deltalake");
23feature_gated_sink_mod!(doris, "doris");
24#[cfg(any(feature = "sink-doris", feature = "sink-starrocks"))]
25pub mod doris_starrocks_connector;
26pub mod dynamodb;
27pub mod elasticsearch_opensearch;
28pub mod encoder;
29pub mod file_sink;
30pub mod formatter;
31feature_gated_sink_mod!(google_pubsub, GooglePubSub, "google_pubsub");
32feature_gated_sink_mod!(lancedb, LanceDb, "lancedb");
33pub mod http;
34pub mod iceberg;
35pub mod kafka;
36pub mod kinesis;
37use risingwave_common::bail;
38use risingwave_pb::stream_plan::PbSinkSchemaChange;
39pub mod jdbc_jni_client;
40pub mod log_store;
41pub mod mock_coordination_client;
42pub mod mongodb;
43pub mod mqtt;
44pub mod nats;
45pub mod postgres;
46pub mod pulsar;
47pub mod redis;
48pub mod remote;
49pub mod snowflake_redshift;
50pub mod sqlserver;
51feature_gated_sink_mod!(starrocks, "starrocks");
52pub mod test_sink;
53pub mod trivial;
54pub mod turbopuffer;
55pub mod utils;
56pub mod writer;
57pub mod prelude {
58    pub use crate::sink::{
59        Result, SINK_TYPE_APPEND_ONLY, SINK_USER_FORCE_APPEND_ONLY_OPTION,
60        SINK_USER_FORCE_COMPACTION, Sink, SinkError, SinkParam, SinkWriterParam,
61    };
62}
63
64use std::collections::{BTreeMap, HashMap};
65use std::future::Future;
66use std::sync::{Arc, LazyLock};
67
68use ::redis::RedisError;
69use anyhow::anyhow;
70use async_trait::async_trait;
71use chrono_tz::{Tz, UTC};
72use decouple_checkpoint_log_sink::{
73    COMMIT_CHECKPOINT_INTERVAL, DEFAULT_COMMIT_CHECKPOINT_INTERVAL_WITH_SINK_DECOUPLE,
74    DEFAULT_COMMIT_CHECKPOINT_INTERVAL_WITHOUT_SINK_DECOUPLE,
75};
76use futures::future::BoxFuture;
77use opendal::Error as OpendalError;
78use prometheus::Registry;
79use risingwave_common::array::ArrayError;
80use risingwave_common::bitmap::Bitmap;
81use risingwave_common::catalog::{ColumnDesc, Field, Schema};
82use risingwave_common::config::StreamingConfig;
83use risingwave_common::hash::ActorId;
84use risingwave_common::metrics::{
85    LabelGuardedHistogram, LabelGuardedHistogramVec, LabelGuardedIntCounter,
86    LabelGuardedIntCounterVec, LabelGuardedIntGaugeVec,
87};
88use risingwave_common::monitor::GLOBAL_METRICS_REGISTRY;
89use risingwave_common::secret::{LocalSecretManager, SecretError};
90use risingwave_common::session_config::sink_decouple::SinkDecouple;
91use risingwave_common::{
92    register_guarded_histogram_vec_with_registry, register_guarded_int_counter_vec_with_registry,
93    register_guarded_int_gauge_vec_with_registry,
94};
95use risingwave_pb::catalog::PbSinkType;
96use risingwave_pb::connector_service::{PbSinkParam, SinkMetadata, TableSchema};
97use risingwave_pb::id::ExecutorId;
98use risingwave_rpc_client::MetaClient;
99use risingwave_rpc_client::error::RpcError;
100use starrocks::STARROCKS_SINK;
101use thiserror::Error;
102use thiserror_ext::AsReport;
103use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
104pub use tracing;
105
106use self::catalog::{SinkFormatDesc, SinkType};
107use self::clickhouse::CLICKHOUSE_SINK;
108use self::deltalake::DELTALAKE_SINK;
109use self::iceberg::ICEBERG_SINK;
110use self::lancedb::LANCEDB_SINK;
111use self::mock_coordination_client::{MockMetaClient, SinkCoordinationRpcClientEnum};
112use crate::WithPropertiesExt;
113use crate::connector_common::IcebergSinkCompactionUpdate;
114use crate::error::{ConnectorError, ConnectorResult};
115use crate::sink::boxed::{BoxSinglePhaseCoordinator, BoxTwoPhaseCoordinator};
116use crate::sink::catalog::desc::SinkDesc;
117use crate::sink::catalog::{SinkCatalog, SinkId};
118use crate::sink::decouple_checkpoint_log_sink::ICEBERG_DEFAULT_COMMIT_CHECKPOINT_INTERVAL;
119use crate::sink::file_sink::fs::FsSink;
120use crate::sink::log_store::{LogReader, LogStoreReadItem, LogStoreResult, TruncateOffset};
121use crate::sink::snowflake_redshift::snowflake::SNOWFLAKE_SINK_V2;
122use crate::sink::utils::feature_gated_sink_mod;
123
124const BOUNDED_CHANNEL_SIZE: usize = 16;
125#[macro_export]
126macro_rules! for_all_sinks {
127    ($macro:path $(, $arg:tt)*) => {
128        $macro! {
129            {
130                { Redis, $crate::sink::redis::RedisSink, $crate::sink::redis::RedisConfig },
131                { Kafka, $crate::sink::kafka::KafkaSink, $crate::sink::kafka::KafkaConfig },
132                { Pulsar, $crate::sink::pulsar::PulsarSink, $crate::sink::pulsar::PulsarConfig },
133                { BlackHole, $crate::sink::trivial::BlackHoleSink, () },
134                { Http, $crate::sink::http::HttpSink, $crate::sink::http::HttpConfig },
135                { Turbopuffer, $crate::sink::turbopuffer::TurbopufferSink, $crate::sink::turbopuffer::TurbopufferConfig },
136                { Kinesis, $crate::sink::kinesis::KinesisSink, $crate::sink::kinesis::KinesisSinkConfig },
137                { ClickHouse, $crate::sink::clickhouse::ClickHouseSink, $crate::sink::clickhouse::ClickHouseConfig },
138                { Iceberg, $crate::sink::iceberg::IcebergSink, $crate::sink::iceberg::IcebergConfig },
139                { Mqtt, $crate::sink::mqtt::MqttSink, $crate::sink::mqtt::MqttConfig },
140                { GooglePubSub, $crate::sink::google_pubsub::GooglePubSubSink, $crate::sink::google_pubsub::GooglePubSubConfig },
141                { Nats, $crate::sink::nats::NatsSink, $crate::sink::nats::NatsConfig },
142                { Jdbc, $crate::sink::remote::JdbcSink, () },
143                { ElasticSearch, $crate::sink::elasticsearch_opensearch::elasticsearch::ElasticSearchSink, $crate::sink::elasticsearch_opensearch::elasticsearch_opensearch_config::ElasticSearchConfig },
144                { Opensearch, $crate::sink::elasticsearch_opensearch::opensearch::OpenSearchSink, $crate::sink::elasticsearch_opensearch::elasticsearch_opensearch_config::OpenSearchConfig },
145                { Cassandra, $crate::sink::remote::CassandraSink, () },
146                { Doris, $crate::sink::doris::DorisSink, $crate::sink::doris::DorisConfig },
147                { Starrocks, $crate::sink::starrocks::StarrocksSink, $crate::sink::starrocks::StarrocksConfig },
148                { S3, $crate::sink::file_sink::opendal_sink::FileSink<$crate::sink::file_sink::s3::S3Sink>, $crate::sink::file_sink::s3::S3Config },
149
150                { Gcs, $crate::sink::file_sink::opendal_sink::FileSink<$crate::sink::file_sink::gcs::GcsSink>, $crate::sink::file_sink::gcs::GcsConfig },
151                { Azblob, $crate::sink::file_sink::opendal_sink::FileSink<$crate::sink::file_sink::azblob::AzblobSink>, $crate::sink::file_sink::azblob::AzblobConfig },
152                { Webhdfs, $crate::sink::file_sink::opendal_sink::FileSink<$crate::sink::file_sink::webhdfs::WebhdfsSink>, $crate::sink::file_sink::webhdfs::WebhdfsConfig },
153
154                { Fs, $crate::sink::file_sink::opendal_sink::FileSink<FsSink>, $crate::sink::file_sink::fs::FsConfig },
155                { SnowflakeV2, $crate::sink::snowflake_redshift::snowflake::SnowflakeV2Sink, $crate::sink::snowflake_redshift::snowflake::SnowflakeV2Config },
156                { Snowflake, $crate::sink::file_sink::opendal_sink::FileSink<$crate::sink::file_sink::s3::SnowflakeSink>, $crate::sink::file_sink::s3::SnowflakeConfig },
157                { RedShift, $crate::sink::snowflake_redshift::redshift::RedshiftSink, $crate::sink::snowflake_redshift::redshift::RedShiftConfig },
158                { DeltaLake, $crate::sink::deltalake::DeltaLakeSink, $crate::sink::deltalake::DeltaLakeConfig },
159                { LanceDb, $crate::sink::lancedb::LanceDbSink, $crate::sink::lancedb::LanceDbConfig },
160                { BigQuery, $crate::sink::big_query::BigQuerySink, $crate::sink::big_query::BigQueryConfig },
161                { DynamoDb, $crate::sink::dynamodb::DynamoDbSink, $crate::sink::dynamodb::DynamoDbConfig },
162                { Mongodb, $crate::sink::mongodb::MongodbSink, $crate::sink::mongodb::MongodbConfig },
163                { SqlServer, $crate::sink::sqlserver::SqlServerSink, $crate::sink::sqlserver::SqlServerConfig },
164                { Postgres, $crate::sink::postgres::PostgresSink, $crate::sink::postgres::PostgresConfig },
165
166                { Test, $crate::sink::test_sink::TestSink, () },
167                { Table, $crate::sink::trivial::TableSink, () }
168            }
169            $(,$arg)*
170        }
171    };
172}
173
174#[macro_export]
175macro_rules! generate_config_use_clauses {
176    ({$({ $variant_name:ident, $sink_type:ty, $($config_type:tt)+ }), *}) => {
177        $(
178            $crate::generate_config_use_single! { $($config_type)+ }
179        )*
180    };
181}
182
183#[macro_export]
184macro_rules! generate_config_use_single {
185    // Skip () config types
186    (()) => {};
187
188    // Generate use clause for actual config types
189    ($config_type:path) => {
190        #[allow(unused_imports)]
191        pub(super) use $config_type;
192    };
193}
194
195// Convenience macro that uses for_all_sinks
196#[macro_export]
197macro_rules! use_all_sink_configs {
198    () => {
199        $crate::for_all_sinks! { $crate::generate_config_use_clauses }
200    };
201}
202
203#[macro_export]
204macro_rules! dispatch_sink {
205    ({$({$variant_name:ident, $sink_type:ty, $config_type:ty}),*}, $impl:tt, $sink:tt, $body:tt) => {{
206        use $crate::sink::SinkImpl;
207
208        match $impl {
209            $(
210                SinkImpl::$variant_name($sink) => $body,
211            )*
212        }
213    }};
214    ($impl:expr, $sink:ident, $body:expr) => {{
215        $crate::for_all_sinks! {$crate::dispatch_sink, {$impl}, $sink, {$body}}
216    }};
217}
218
219#[macro_export]
220macro_rules! match_sink_name_str {
221    ({$({$variant_name:ident, $sink_type:ty, $config_type:ty}),*}, $name_str:tt, $type_name:ident, $body:tt, $on_other_closure:tt) => {{
222        use $crate::sink::Sink;
223        match $name_str {
224            $(
225                <$sink_type>::SINK_NAME => {
226                    type $type_name = $sink_type;
227                    {
228                        $body
229                    }
230                },
231            )*
232            other => ($on_other_closure)(other),
233        }
234    }};
235    ($name_str:expr, $type_name:ident, $body:expr, $on_other_closure:expr) => {{
236        $crate::for_all_sinks! {$crate::match_sink_name_str, {$name_str}, $type_name, {$body}, {$on_other_closure}}
237    }};
238}
239
240pub const CONNECTOR_TYPE_KEY: &str = "connector";
241pub const SINK_TYPE_OPTION: &str = "type";
242/// `snapshot = false` corresponds to [`risingwave_pb::stream_plan::StreamScanType::UpstreamOnly`]
243pub const SINK_SNAPSHOT_OPTION: &str = "snapshot";
244pub const SINK_TYPE_APPEND_ONLY: &str = "append-only";
245pub const SINK_TYPE_DEBEZIUM: &str = "debezium";
246pub const SINK_TYPE_UPSERT: &str = "upsert";
247pub const SINK_TYPE_RETRACT: &str = "retract";
248/// Whether to drop DELETE and convert UPDATE to INSERT in the sink executor.
249pub const SINK_USER_IGNORE_DELETE_OPTION: &str = "ignore_delete";
250/// Alias for [`SINK_USER_IGNORE_DELETE_OPTION`], kept for backward compatibility.
251pub const SINK_USER_FORCE_APPEND_ONLY_OPTION: &str = "force_append_only";
252pub const SINK_USER_FORCE_COMPACTION: &str = "force_compaction";
253/// When enabled, the sink executor preserves distinct upstream stream-key changes that map to the
254/// same downstream primary key instead of compacting them into one final-state update within a
255/// barrier. Upstream changes under the same stream key may still be compacted earlier.
256pub const SINK_USER_PRESERVE_ROW_LEVEL_CHANGES: &str = "preserve_row_level_changes";
257
258/// Return whether the configured sink uses exactly-once commit state.
259///
260/// Connector dispatch is centralized here, while each [`Sink`] implementation owns the
261/// interpretation and default of its properties.
262pub fn sink_is_exactly_once(properties: &BTreeMap<String, String>) -> Result<bool> {
263    let sink_type = properties
264        .get(CONNECTOR_TYPE_KEY)
265        .ok_or_else(|| SinkError::Config(anyhow!("missing config: {}", CONNECTOR_TYPE_KEY)))?
266        .to_lowercase();
267
268    match_sink_name_str!(
269        sink_type.as_str(),
270        SinkType,
271        SinkType::is_exactly_once(properties),
272        |other| Err(SinkError::Config(anyhow!(
273            "unsupported sink connector {}",
274            other
275        )))
276    )
277}
278
279pub trait UnknownFields {
280    /// Unrecognized fields in the `WITH` clause.
281    fn unknown_fields(&self) -> HashMap<String, String>;
282}
283
284impl UnknownFields for () {
285    fn unknown_fields(&self) -> HashMap<String, String> {
286        HashMap::new()
287    }
288}
289
290impl UnknownFields for HashMap<String, String> {
291    fn unknown_fields(&self) -> HashMap<String, String> {
292        self.clone()
293    }
294}
295
296#[macro_export]
297macro_rules! impl_sink_unknown_fields {
298    ($config_type:ty) => {
299        impl $crate::sink::UnknownFields for $config_type {
300            fn unknown_fields(&self) -> std::collections::HashMap<String, String> {
301                self.unknown_fields.clone()
302            }
303        }
304    };
305}
306
307#[macro_export]
308macro_rules! impl_validate_sink_unknown_fields {
309    () => {
310        fn validate_unknown_fields(&self) -> $crate::sink::Result<()> {
311            $crate::sink::validate_sink_unknown_fields(&self.config)
312        }
313    };
314}
315
316pub fn validate_sink_unknown_fields(config: &impl UnknownFields) -> Result<()> {
317    let mut unknown_fields = config.unknown_fields();
318    // These options are consumed by the sink DDL/planner layer. They are valid for sink creation
319    // even if a connector-specific config does not declare them.
320    for key in [
321        CONNECTOR_TYPE_KEY,
322        SINK_TYPE_OPTION,
323        SINK_SNAPSHOT_OPTION,
324        SINK_USER_IGNORE_DELETE_OPTION,
325        SINK_USER_FORCE_APPEND_ONLY_OPTION,
326        SINK_USER_FORCE_COMPACTION,
327        SINK_USER_PRESERVE_ROW_LEVEL_CHANGES,
328        "backfill_rate_limit",
329        "primary_key",
330        "sink_rate_limit",
331    ] {
332        unknown_fields.remove(key);
333    }
334    if unknown_fields.is_empty() {
335        Ok(())
336    } else {
337        Err(SinkError::Config(anyhow!(
338            "Unknown fields in the WITH clause: {:?}",
339            unknown_fields
340        )))
341    }
342}
343
344#[derive(Debug, Clone, PartialEq, Eq)]
345pub struct SinkParam {
346    pub sink_id: SinkId,
347    pub sink_name: String,
348    pub properties: BTreeMap<String, String>,
349    pub columns: Vec<ColumnDesc>,
350    /// User-defined primary key indices for upsert sink, if any.
351    pub downstream_pk: Option<Vec<usize>>,
352    pub sink_type: SinkType,
353    /// Whether to drop DELETE and convert UPDATE to INSERT in the sink executor.
354    pub ignore_delete: bool,
355    pub format_desc: Option<SinkFormatDesc>,
356    pub db_name: String,
357
358    /// - For `CREATE SINK ... FROM ...`, the name of the source table.
359    /// - For `CREATE SINK ... AS <query>`, the name of the sink itself.
360    ///
361    /// See also `gen_sink_plan`.
362    // TODO(eric): Why need these 2 fields (db_name and sink_from_name)?
363    pub sink_from_name: String,
364}
365
366impl SinkParam {
367    pub fn from_proto(pb_param: PbSinkParam) -> Self {
368        let ignore_delete = pb_param.ignore_delete();
369        let table_schema = pb_param.table_schema.expect("should contain table schema");
370        let format_desc = match pb_param.format_desc {
371            Some(f) => f.try_into().ok(),
372            None => {
373                let connector = pb_param.properties.get(CONNECTOR_TYPE_KEY);
374                let r#type = pb_param.properties.get(SINK_TYPE_OPTION);
375                match (connector, r#type) {
376                    (Some(c), Some(t)) => SinkFormatDesc::from_legacy_type(c, t).ok().flatten(),
377                    _ => None,
378                }
379            }
380        };
381        Self {
382            sink_id: SinkId::from(pb_param.sink_id),
383            sink_name: pb_param.sink_name,
384            properties: pb_param.properties,
385            columns: table_schema.columns.iter().map(ColumnDesc::from).collect(),
386            downstream_pk: if table_schema.pk_indices.is_empty() {
387                None
388            } else {
389                Some(
390                    (table_schema.pk_indices.iter())
391                        .map(|i| *i as usize)
392                        .collect(),
393                )
394            },
395            sink_type: SinkType::from_proto(
396                PbSinkType::try_from(pb_param.sink_type).expect("should be able to convert"),
397            ),
398            ignore_delete,
399            format_desc,
400            db_name: pb_param.db_name,
401            sink_from_name: pb_param.sink_from_name,
402        }
403    }
404
405    pub fn to_proto(&self) -> PbSinkParam {
406        PbSinkParam {
407            sink_id: self.sink_id,
408            sink_name: self.sink_name.clone(),
409            properties: self.properties.clone(),
410            table_schema: Some(TableSchema {
411                columns: self.columns.iter().map(|col| col.to_protobuf()).collect(),
412                pk_indices: (self.downstream_pk.as_ref())
413                    .map_or_else(Vec::new, |pk| pk.iter().map(|i| *i as u32).collect()),
414            }),
415            sink_type: self.sink_type.to_proto().into(),
416            format_desc: self.format_desc.as_ref().map(|f| f.to_proto()),
417            db_name: self.db_name.clone(),
418            sink_from_name: self.sink_from_name.clone(),
419            raw_ignore_delete: self.ignore_delete,
420        }
421    }
422
423    pub fn schema(&self) -> Schema {
424        Schema {
425            fields: self.columns.iter().map(Field::from).collect(),
426        }
427    }
428
429    /// Get the downstream primary key indices specified by the user. If not specified, return
430    /// an empty vector.
431    ///
432    /// Prefer directly accessing the `downstream_pk` field, as it uses `None` to represent
433    /// unspecified values, making it clearer.
434    pub fn downstream_pk_or_empty(&self) -> Vec<usize> {
435        self.downstream_pk.clone().unwrap_or_default()
436    }
437
438    // `SinkParams` should only be used when there is a secret context.
439    // FIXME: Use a new type for `SinkFormatDesc` with properties contain filled secrets.
440    pub fn fill_secret_for_format_desc(
441        format_desc: Option<SinkFormatDesc>,
442    ) -> Result<Option<SinkFormatDesc>> {
443        match format_desc {
444            Some(mut format_desc) => {
445                format_desc.options = LocalSecretManager::global()
446                    .fill_secrets(format_desc.options, format_desc.secret_refs.clone())?;
447                Ok(Some(format_desc))
448            }
449            None => Ok(None),
450        }
451    }
452
453    /// Try to convert a `SinkCatalog` to a `SinkParam` and fill the secrets to properties.
454    pub fn try_from_sink_catalog(sink_catalog: SinkCatalog) -> Result<Self> {
455        let columns = sink_catalog
456            .visible_columns()
457            .map(|col| col.column_desc.clone())
458            .collect();
459        let properties_with_secret = LocalSecretManager::global()
460            .fill_secrets(sink_catalog.properties, sink_catalog.secret_refs)?;
461        let format_desc_with_secret = Self::fill_secret_for_format_desc(sink_catalog.format_desc)?;
462        Ok(Self {
463            sink_id: sink_catalog.id,
464            sink_name: sink_catalog.name,
465            properties: properties_with_secret,
466            columns,
467            downstream_pk: sink_catalog.downstream_pk,
468            sink_type: sink_catalog.sink_type,
469            ignore_delete: sink_catalog.ignore_delete,
470            format_desc: format_desc_with_secret,
471            db_name: sink_catalog.db_name,
472            sink_from_name: sink_catalog.sink_from_name,
473        })
474    }
475}
476
477pub fn enforce_secret_sink(props: &impl WithPropertiesExt) -> ConnectorResult<()> {
478    use crate::enforce_secret::EnforceSecret;
479
480    let connector = props
481        .get_connector()
482        .ok_or_else(|| anyhow!("Must specify 'connector' in WITH clause"))?;
483    let key_iter = props.key_iter();
484    match_sink_name_str!(
485        connector.as_str(),
486        PropType,
487        PropType::enforce_secret(key_iter),
488        |other| bail!("connector '{}' is not supported", other)
489    )
490}
491
492pub static GLOBAL_SINK_METRICS: LazyLock<SinkMetrics> =
493    LazyLock::new(|| SinkMetrics::new(&GLOBAL_METRICS_REGISTRY));
494
495#[derive(Clone)]
496pub struct SinkMetrics {
497    pub sink_commit_duration: LabelGuardedHistogramVec,
498    pub connector_sink_rows_received: LabelGuardedIntCounterVec,
499
500    // Log store writer metrics
501    pub log_store_first_write_epoch: LabelGuardedIntGaugeVec,
502    pub log_store_latest_write_epoch: LabelGuardedIntGaugeVec,
503    pub log_store_write_rows: LabelGuardedIntCounterVec,
504
505    // Log store reader metrics
506    pub log_store_latest_read_epoch: LabelGuardedIntGaugeVec,
507    pub log_store_read_rows: LabelGuardedIntCounterVec,
508    pub log_store_read_bytes: LabelGuardedIntCounterVec,
509    pub log_store_reader_wait_new_future_duration_ns: LabelGuardedIntCounterVec,
510
511    // Iceberg metrics
512    pub iceberg_write_qps: LabelGuardedIntCounterVec,
513    pub iceberg_write_latency: LabelGuardedHistogramVec,
514    pub iceberg_rolling_unflushed_data_file: LabelGuardedIntGaugeVec,
515    pub iceberg_position_delete_cache_num: LabelGuardedIntGaugeVec,
516    pub iceberg_partition_num: LabelGuardedIntGaugeVec,
517    pub iceberg_write_bytes: LabelGuardedIntCounterVec,
518    pub iceberg_snapshot_num: LabelGuardedIntGaugeVec,
519}
520
521impl SinkMetrics {
522    pub fn new(registry: &Registry) -> Self {
523        let sink_commit_duration = register_guarded_histogram_vec_with_registry!(
524            "sink_commit_duration",
525            "Duration of commit op in sink",
526            &["actor_id", "connector", "sink_id", "sink_name"],
527            registry
528        )
529        .unwrap();
530
531        let connector_sink_rows_received = register_guarded_int_counter_vec_with_registry!(
532            "connector_sink_rows_received",
533            "Number of rows received by sink",
534            &["actor_id", "connector_type", "sink_id", "sink_name"],
535            registry
536        )
537        .unwrap();
538
539        let log_store_first_write_epoch = register_guarded_int_gauge_vec_with_registry!(
540            "log_store_first_write_epoch",
541            "The first write epoch of log store",
542            &["actor_id", "sink_id", "sink_name"],
543            registry
544        )
545        .unwrap();
546
547        let log_store_latest_write_epoch = register_guarded_int_gauge_vec_with_registry!(
548            "log_store_latest_write_epoch",
549            "The latest write epoch of log store",
550            &["actor_id", "sink_id", "sink_name"],
551            registry
552        )
553        .unwrap();
554
555        let log_store_write_rows = register_guarded_int_counter_vec_with_registry!(
556            "log_store_write_rows",
557            "The write rate of rows",
558            &["actor_id", "sink_id", "sink_name"],
559            registry
560        )
561        .unwrap();
562
563        let log_store_latest_read_epoch = register_guarded_int_gauge_vec_with_registry!(
564            "log_store_latest_read_epoch",
565            "The latest read epoch of log store",
566            &["actor_id", "connector", "sink_id", "sink_name"],
567            registry
568        )
569        .unwrap();
570
571        let log_store_read_rows = register_guarded_int_counter_vec_with_registry!(
572            "log_store_read_rows",
573            "The read rate of rows",
574            &["actor_id", "connector", "sink_id", "sink_name"],
575            registry
576        )
577        .unwrap();
578
579        let log_store_read_bytes = register_guarded_int_counter_vec_with_registry!(
580            "log_store_read_bytes",
581            "Total size of chunks read by log reader",
582            &["actor_id", "connector", "sink_id", "sink_name"],
583            registry
584        )
585        .unwrap();
586
587        let log_store_reader_wait_new_future_duration_ns =
588            register_guarded_int_counter_vec_with_registry!(
589                "log_store_reader_wait_new_future_duration_ns",
590                "Accumulated duration of LogReader to wait for next call to create future",
591                &["actor_id", "connector", "sink_id", "sink_name"],
592                registry
593            )
594            .unwrap();
595
596        let iceberg_write_qps = register_guarded_int_counter_vec_with_registry!(
597            "iceberg_write_qps",
598            "The qps of iceberg writer",
599            &["actor_id", "sink_id", "sink_name"],
600            registry
601        )
602        .unwrap();
603
604        let iceberg_write_latency = register_guarded_histogram_vec_with_registry!(
605            "iceberg_write_latency",
606            "The latency of iceberg writer",
607            &["actor_id", "sink_id", "sink_name"],
608            registry
609        )
610        .unwrap();
611
612        let iceberg_rolling_unflushed_data_file = register_guarded_int_gauge_vec_with_registry!(
613            "iceberg_rolling_unflushed_data_file",
614            "The unflushed data file count of iceberg rolling writer",
615            &["actor_id", "sink_id", "sink_name"],
616            registry
617        )
618        .unwrap();
619
620        let iceberg_position_delete_cache_num = register_guarded_int_gauge_vec_with_registry!(
621            "iceberg_position_delete_cache_num",
622            "The delete cache num of iceberg position delete writer",
623            &["actor_id", "sink_id", "sink_name"],
624            registry
625        )
626        .unwrap();
627
628        let iceberg_partition_num = register_guarded_int_gauge_vec_with_registry!(
629            "iceberg_partition_num",
630            "The partition num of iceberg partition writer",
631            &["actor_id", "sink_id", "sink_name"],
632            registry
633        )
634        .unwrap();
635
636        let iceberg_write_bytes = register_guarded_int_counter_vec_with_registry!(
637            "iceberg_write_bytes",
638            "The write bytes of iceberg writer",
639            &["actor_id", "sink_id", "sink_name"],
640            registry
641        )
642        .unwrap();
643
644        let iceberg_snapshot_num = register_guarded_int_gauge_vec_with_registry!(
645            "iceberg_snapshot_num",
646            "The snapshot number of iceberg table",
647            &["sink_name", "catalog_name", "table_name"],
648            registry
649        )
650        .unwrap();
651
652        Self {
653            sink_commit_duration,
654            connector_sink_rows_received,
655            log_store_first_write_epoch,
656            log_store_latest_write_epoch,
657            log_store_write_rows,
658            log_store_latest_read_epoch,
659            log_store_read_rows,
660            log_store_read_bytes,
661            log_store_reader_wait_new_future_duration_ns,
662            iceberg_write_qps,
663            iceberg_write_latency,
664            iceberg_rolling_unflushed_data_file,
665            iceberg_position_delete_cache_num,
666            iceberg_partition_num,
667            iceberg_write_bytes,
668            iceberg_snapshot_num,
669        }
670    }
671}
672
673#[derive(Clone)]
674pub struct SinkWriterParam {
675    // TODO(eric): deprecate executor_id
676    pub executor_id: ExecutorId,
677    pub vnode_bitmap: Option<Bitmap>,
678    pub meta_client: Option<SinkMetaClient>,
679    // The val has two effect:
680    // 1. Indicates that the sink will accpect the data chunk with extra partition value column.
681    // 2. The index of the extra partition value column.
682    // More detail of partition value column, see `PartitionComputeInfo`
683    pub extra_partition_col_idx: Option<usize>,
684
685    pub actor_id: ActorId,
686    pub sink_id: SinkId,
687    pub sink_name: String,
688    pub connector: String,
689    pub streaming_config: StreamingConfig,
690    pub time_zone: Tz,
691}
692
693#[derive(Clone)]
694pub struct SinkWriterMetrics {
695    pub sink_commit_duration: LabelGuardedHistogram,
696    pub connector_sink_rows_received: LabelGuardedIntCounter,
697}
698
699impl SinkWriterMetrics {
700    pub fn new(writer_param: &SinkWriterParam) -> Self {
701        let labels = [
702            &writer_param.actor_id.to_string(),
703            writer_param.connector.as_str(),
704            &writer_param.sink_id.to_string(),
705            writer_param.sink_name.as_str(),
706        ];
707        let sink_commit_duration = GLOBAL_SINK_METRICS
708            .sink_commit_duration
709            .with_guarded_label_values(&labels);
710        let connector_sink_rows_received = GLOBAL_SINK_METRICS
711            .connector_sink_rows_received
712            .with_guarded_label_values(&labels);
713        Self {
714            sink_commit_duration,
715            connector_sink_rows_received,
716        }
717    }
718
719    #[cfg(test)]
720    pub fn for_test() -> Self {
721        Self {
722            sink_commit_duration: LabelGuardedHistogram::test_histogram::<4>(),
723            connector_sink_rows_received: LabelGuardedIntCounter::test_int_counter::<4>(),
724        }
725    }
726}
727
728#[derive(Clone)]
729pub enum SinkMetaClient {
730    MetaClient(MetaClient),
731    MockMetaClient(MockMetaClient),
732}
733
734impl SinkMetaClient {
735    pub async fn sink_coordinate_client(&self) -> SinkCoordinationRpcClientEnum {
736        match self {
737            SinkMetaClient::MetaClient(meta_client) => {
738                SinkCoordinationRpcClientEnum::SinkCoordinationRpcClient(
739                    meta_client.sink_coordinate_client().await,
740                )
741            }
742            SinkMetaClient::MockMetaClient(mock_meta_client) => {
743                SinkCoordinationRpcClientEnum::MockSinkCoordinationRpcClient(
744                    mock_meta_client.sink_coordinate_client(),
745                )
746            }
747        }
748    }
749
750    pub async fn add_sink_fail_evet_log(
751        &self,
752        sink_id: SinkId,
753        sink_name: String,
754        connector: String,
755        error: String,
756    ) {
757        match self {
758            SinkMetaClient::MetaClient(meta_client) => {
759                match meta_client
760                    .add_sink_fail_evet(sink_id, sink_name, connector, error)
761                    .await
762                {
763                    Ok(_) => {}
764                    Err(e) => {
765                        tracing::warn!(
766                            error = %e.as_report(),
767                            %sink_id,
768                            "failed to add the sink failure event to the event log",
769                        );
770                    }
771                }
772            }
773            SinkMetaClient::MockMetaClient(_) => {}
774        }
775    }
776}
777
778impl SinkWriterParam {
779    pub fn for_test() -> Self {
780        SinkWriterParam {
781            executor_id: Default::default(),
782            vnode_bitmap: Default::default(),
783            meta_client: Default::default(),
784            extra_partition_col_idx: Default::default(),
785
786            actor_id: 1.into(),
787            sink_id: SinkId::new(1),
788            sink_name: "test_sink".to_owned(),
789            connector: "test_connector".to_owned(),
790            streaming_config: StreamingConfig::default(),
791            time_zone: UTC,
792        }
793    }
794}
795
796fn is_sink_support_commit_checkpoint_interval(sink_name: &str) -> bool {
797    matches!(
798        sink_name,
799        ICEBERG_SINK
800            | CLICKHOUSE_SINK
801            | STARROCKS_SINK
802            | DELTALAKE_SINK
803            | SNOWFLAKE_SINK_V2
804            | LANCEDB_SINK
805    )
806}
807pub trait Sink: TryFrom<SinkParam, Error = SinkError> {
808    const SINK_NAME: &'static str;
809
810    type LogSinker: LogSinker;
811
812    /// Return whether this sink uses exactly-once commit state for the loaded properties.
813    fn is_exactly_once(_properties: &BTreeMap<String, String>) -> Result<bool> {
814        Ok(false)
815    }
816
817    fn set_default_commit_checkpoint_interval(
818        desc: &mut SinkDesc,
819        user_specified: &SinkDecouple,
820    ) -> Result<()> {
821        if is_sink_support_commit_checkpoint_interval(Self::SINK_NAME) {
822            match desc.properties.get(COMMIT_CHECKPOINT_INTERVAL) {
823                Some(commit_checkpoint_interval) => {
824                    let commit_checkpoint_interval = commit_checkpoint_interval
825                        .parse::<u64>()
826                        .map_err(|e| SinkError::Config(anyhow!(e)))?;
827                    if matches!(user_specified, SinkDecouple::Disable)
828                        && commit_checkpoint_interval > 1
829                    {
830                        return Err(SinkError::Config(anyhow!(
831                            "config conflict: `commit_checkpoint_interval` larger than 1 means that sink decouple must be enabled, but session config sink_decouple is disabled"
832                        )));
833                    }
834                }
835                None => match user_specified {
836                    SinkDecouple::Default | SinkDecouple::Enable => {
837                        if matches!(Self::SINK_NAME, ICEBERG_SINK) {
838                            desc.properties.insert(
839                                COMMIT_CHECKPOINT_INTERVAL.to_owned(),
840                                ICEBERG_DEFAULT_COMMIT_CHECKPOINT_INTERVAL.to_string(),
841                            );
842                        } else {
843                            desc.properties.insert(
844                                COMMIT_CHECKPOINT_INTERVAL.to_owned(),
845                                DEFAULT_COMMIT_CHECKPOINT_INTERVAL_WITH_SINK_DECOUPLE.to_string(),
846                            );
847                        }
848                    }
849                    SinkDecouple::Disable => {
850                        desc.properties.insert(
851                            COMMIT_CHECKPOINT_INTERVAL.to_owned(),
852                            DEFAULT_COMMIT_CHECKPOINT_INTERVAL_WITHOUT_SINK_DECOUPLE.to_string(),
853                        );
854                    }
855                },
856            }
857        }
858        Ok(())
859    }
860
861    /// `user_specified` is the value of `sink_decouple` config.
862    fn is_sink_decouple(user_specified: &SinkDecouple) -> Result<bool> {
863        match user_specified {
864            SinkDecouple::Default | SinkDecouple::Enable => Ok(true),
865            SinkDecouple::Disable => Ok(false),
866        }
867    }
868
869    fn support_schema_change() -> bool {
870        false
871    }
872
873    /// Validates an `ALTER SINK` transition and the resulting configuration.
874    ///
875    /// `config` contains the merged sink properties, while `alter_props` contains only the
876    /// properties specified by the current statement.
877    fn validate_alter_config_change(
878        config: &BTreeMap<String, String>,
879        _alter_props: &BTreeMap<String, String>,
880    ) -> Result<()> {
881        Self::validate_alter_config(config)
882    }
883
884    fn validate_alter_config(_config: &BTreeMap<String, String>) -> Result<()> {
885        Ok(())
886    }
887
888    fn validate_unknown_fields(&self) -> Result<()> {
889        Ok(())
890    }
891
892    async fn validate(&self) -> Result<()>;
893    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker>;
894
895    fn is_coordinated_sink(&self) -> bool {
896        false
897    }
898
899    async fn new_coordinator(
900        &self,
901        _iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,
902    ) -> Result<SinkCommitCoordinator> {
903        Err(SinkError::Coordinator(anyhow!("no coordinator")))
904    }
905}
906
907pub trait SinkLogReader: Send {
908    fn start_from(
909        &mut self,
910        start_offset: Option<u64>,
911    ) -> impl Future<Output = LogStoreResult<()>> + Send + '_;
912    /// Emit the next item.
913    ///
914    /// The implementation should ensure that the future is cancellation safe.
915    fn next_item(
916        &mut self,
917    ) -> impl Future<Output = LogStoreResult<(u64, LogStoreReadItem)>> + Send + '_;
918
919    /// Mark that all items emitted so far have been consumed and it is safe to truncate the log
920    /// from the current offset.
921    fn truncate(&mut self, offset: TruncateOffset) -> LogStoreResult<()>;
922}
923
924impl<R: LogReader> SinkLogReader for &mut R {
925    fn next_item(
926        &mut self,
927    ) -> impl Future<Output = LogStoreResult<(u64, LogStoreReadItem)>> + Send + '_ {
928        <R as LogReader>::next_item(*self)
929    }
930
931    fn truncate(&mut self, offset: TruncateOffset) -> LogStoreResult<()> {
932        <R as LogReader>::truncate(*self, offset)
933    }
934
935    fn start_from(
936        &mut self,
937        start_offset: Option<u64>,
938    ) -> impl Future<Output = LogStoreResult<()>> + Send + '_ {
939        <R as LogReader>::start_from(*self, start_offset)
940    }
941}
942
943#[async_trait]
944pub trait LogSinker: 'static + Send {
945    // Note: Please rebuild the log reader's read stream before consuming the log store,
946    async fn consume_log_and_sink(self, log_reader: impl SinkLogReader) -> Result<!>;
947}
948pub type SinkCommittedEpochSubscriber = Arc<
949    dyn Fn(SinkId) -> BoxFuture<'static, Result<(u64, UnboundedReceiver<u64>)>>
950        + Send
951        + Sync
952        + 'static,
953>;
954
955pub enum SinkCommitCoordinator {
956    SinglePhase(BoxSinglePhaseCoordinator),
957    TwoPhase(BoxTwoPhaseCoordinator),
958}
959
960#[async_trait]
961pub trait SinglePhaseCommitCoordinator {
962    /// Initialize the sink committer coordinator.
963    async fn init(&mut self) -> Result<()>;
964
965    /// Commit data directly using single-phase strategy.
966    async fn commit_data(&mut self, epoch: u64, metadata: Vec<SinkMetadata>) -> Result<()>;
967
968    /// Idempotent implementation is required, because `commit_schema_change` in the same epoch could be called multiple
969    /// times.
970    async fn commit_schema_change(
971        &mut self,
972        _epoch: u64,
973        _schema_change: PbSinkSchemaChange,
974    ) -> Result<()> {
975        Err(SinkError::Coordinator(anyhow!(
976            "Schema change is not implemented for single-phase commit coordinator {}",
977            std::any::type_name::<Self>()
978        )))
979    }
980}
981
982#[async_trait]
983pub trait TwoPhaseCommitCoordinator {
984    /// Initialize the sink committer coordinator.
985    async fn init(&mut self) -> Result<()>;
986
987    /// Return serialized commit metadata to be passed to `commit`.
988    async fn pre_commit(
989        &mut self,
990        epoch: u64,
991        metadata: Vec<SinkMetadata>,
992        schema_change: Option<PbSinkSchemaChange>,
993    ) -> Result<Option<Vec<u8>>>;
994
995    /// Idempotent implementation is required, because `commit_data` in the same epoch could be called multiple times.
996    async fn commit_data(&mut self, epoch: u64, commit_metadata: Vec<u8>) -> Result<()>;
997
998    /// Idempotent implementation is required, because `commit_schema_change` in the same epoch could be called multiple
999    /// times.
1000    async fn commit_schema_change(
1001        &mut self,
1002        _epoch: u64,
1003        _schema_change: PbSinkSchemaChange,
1004    ) -> Result<()> {
1005        Err(SinkError::Coordinator(anyhow!(
1006            "Schema change is not implemented for two-phase commit coordinator {}",
1007            std::any::type_name::<Self>()
1008        )))
1009    }
1010
1011    /// Idempotent implementation is required, because `abort` in the same epoch could be called multiple times.
1012    async fn abort(&mut self, epoch: u64, commit_metadata: Vec<u8>);
1013}
1014
1015impl SinkImpl {
1016    pub fn new(mut param: SinkParam) -> Result<Self> {
1017        const PRIVATE_LINK_TARGET_KEY: &str = "privatelink.targets";
1018
1019        // remove privatelink related properties if any
1020        param.properties.remove(PRIVATE_LINK_TARGET_KEY);
1021
1022        let sink_type = param
1023            .properties
1024            .get(CONNECTOR_TYPE_KEY)
1025            .ok_or_else(|| SinkError::Config(anyhow!("missing config: {}", CONNECTOR_TYPE_KEY)))?;
1026
1027        let sink_type = sink_type.to_lowercase();
1028        match_sink_name_str!(
1029            sink_type.as_str(),
1030            SinkType,
1031            Ok(SinkType::try_from(param)?.into()),
1032            |other| {
1033                Err(SinkError::Config(anyhow!(
1034                    "unsupported sink connector {}",
1035                    other
1036                )))
1037            }
1038        )
1039    }
1040
1041    pub fn is_sink_into_table(&self) -> bool {
1042        matches!(self, SinkImpl::Table(_))
1043    }
1044
1045    pub fn is_blackhole(&self) -> bool {
1046        matches!(self, SinkImpl::BlackHole(_))
1047    }
1048
1049    pub fn is_coordinated_sink(&self) -> bool {
1050        dispatch_sink!(self, sink, sink.is_coordinated_sink())
1051    }
1052
1053    pub fn validate_unknown_fields(&self) -> Result<()> {
1054        dispatch_sink!(self, sink, sink.validate_unknown_fields())
1055    }
1056}
1057
1058pub fn build_sink(param: SinkParam) -> Result<SinkImpl> {
1059    SinkImpl::new(param)
1060}
1061
1062macro_rules! def_sink_impl {
1063    () => {
1064        $crate::for_all_sinks! { def_sink_impl }
1065    };
1066    ({ $({ $variant_name:ident, $sink_type:ty, $config_type:ty }),* }) => {
1067        #[derive(Debug)]
1068        pub enum SinkImpl {
1069            $(
1070                $variant_name(Box<$sink_type>),
1071            )*
1072        }
1073
1074        $(
1075            impl From<$sink_type> for SinkImpl {
1076                fn from(sink: $sink_type) -> SinkImpl {
1077                    SinkImpl::$variant_name(Box::new(sink))
1078                }
1079            }
1080        )*
1081    };
1082}
1083
1084def_sink_impl!();
1085
1086pub type Result<T> = std::result::Result<T, SinkError>;
1087
1088#[derive(Error, Debug)]
1089pub enum SinkError {
1090    #[error("Kafka error: {0}")]
1091    Kafka(#[from] rdkafka::error::KafkaError),
1092    #[error("Kinesis error: {0}")]
1093    Kinesis(
1094        #[source]
1095        #[backtrace]
1096        anyhow::Error,
1097    ),
1098    #[error("Remote sink error: {0}")]
1099    Remote(
1100        #[source]
1101        #[backtrace]
1102        anyhow::Error,
1103    ),
1104    #[error("Encode error: {0}")]
1105    Encode(String),
1106    #[error("Avro error: {0}")]
1107    Avro(#[from] apache_avro::Error),
1108    #[error("Iceberg error: {0}")]
1109    Iceberg(
1110        #[source]
1111        #[backtrace]
1112        anyhow::Error,
1113    ),
1114    #[error("config error: {0}")]
1115    Config(
1116        #[source]
1117        #[backtrace]
1118        anyhow::Error,
1119    ),
1120    #[error("coordinator error: {0}")]
1121    Coordinator(
1122        #[source]
1123        #[backtrace]
1124        anyhow::Error,
1125    ),
1126    #[error("ClickHouse error: {0}")]
1127    ClickHouse(String),
1128    #[error("Redis error: {0}")]
1129    Redis(String),
1130    #[error("Http error: {0}")]
1131    Http(
1132        #[source]
1133        #[backtrace]
1134        anyhow::Error,
1135    ),
1136    #[error("Mqtt error: {0}")]
1137    Mqtt(
1138        #[source]
1139        #[backtrace]
1140        anyhow::Error,
1141    ),
1142    #[error("Nats error: {0}")]
1143    Nats(
1144        #[source]
1145        #[backtrace]
1146        anyhow::Error,
1147    ),
1148    #[error("Google Pub/Sub error: {0}")]
1149    GooglePubSub(
1150        #[source]
1151        #[backtrace]
1152        anyhow::Error,
1153    ),
1154    #[error("Doris/Starrocks connect error: {0}")]
1155    DorisStarrocksConnect(
1156        #[source]
1157        #[backtrace]
1158        anyhow::Error,
1159    ),
1160    #[error("Doris error: {0}")]
1161    Doris(String),
1162    #[error("DeltaLake error: {0}")]
1163    DeltaLake(
1164        #[source]
1165        #[backtrace]
1166        anyhow::Error,
1167    ),
1168    #[error("LanceDB error: {0}")]
1169    LanceDb(
1170        #[source]
1171        #[backtrace]
1172        anyhow::Error,
1173    ),
1174    #[error("ElasticSearch/OpenSearch error: {0}")]
1175    ElasticSearchOpenSearch(
1176        #[source]
1177        #[backtrace]
1178        anyhow::Error,
1179    ),
1180    #[error("Starrocks error: {0}")]
1181    Starrocks(String),
1182    #[error("File error: {0}")]
1183    File(String),
1184    #[error("Pulsar error: {0}")]
1185    Pulsar(
1186        #[source]
1187        #[backtrace]
1188        anyhow::Error,
1189    ),
1190    #[error(transparent)]
1191    Internal(
1192        #[from]
1193        #[backtrace]
1194        anyhow::Error,
1195    ),
1196    #[error("BigQuery error: {0}")]
1197    BigQuery(
1198        #[source]
1199        #[backtrace]
1200        anyhow::Error,
1201    ),
1202    #[error("DynamoDB error: {0}")]
1203    DynamoDb(
1204        #[source]
1205        #[backtrace]
1206        anyhow::Error,
1207    ),
1208    #[error("SQL Server error: {0}")]
1209    SqlServer(
1210        #[source]
1211        #[backtrace]
1212        anyhow::Error,
1213    ),
1214    #[error("Postgres error: {0}")]
1215    Postgres(
1216        #[source]
1217        #[backtrace]
1218        anyhow::Error,
1219    ),
1220    #[error(transparent)]
1221    Connector(
1222        #[from]
1223        #[backtrace]
1224        ConnectorError,
1225    ),
1226    #[error("Secret error: {0}")]
1227    Secret(
1228        #[from]
1229        #[backtrace]
1230        SecretError,
1231    ),
1232    #[error("Mongodb error: {0}")]
1233    Mongodb(
1234        #[source]
1235        #[backtrace]
1236        anyhow::Error,
1237    ),
1238    #[error("Redshift error: {0}")]
1239    Redshift(
1240        #[source]
1241        #[backtrace]
1242        anyhow::Error,
1243    ),
1244}
1245
1246#[allow(clippy::disallowed_types)]
1247impl From<::iceberg::Error> for SinkError {
1248    fn from(err: ::iceberg::Error) -> Self {
1249        SinkError::Iceberg(anyhow!(err))
1250    }
1251}
1252
1253impl From<sea_orm::DbErr> for SinkError {
1254    fn from(err: sea_orm::DbErr) -> Self {
1255        SinkError::Iceberg(anyhow!(err))
1256    }
1257}
1258
1259impl From<OpendalError> for SinkError {
1260    fn from(error: OpendalError) -> Self {
1261        SinkError::File(error.to_report_string())
1262    }
1263}
1264
1265impl From<parquet::errors::ParquetError> for SinkError {
1266    fn from(error: parquet::errors::ParquetError) -> Self {
1267        SinkError::File(error.to_report_string())
1268    }
1269}
1270
1271impl From<ArrayError> for SinkError {
1272    fn from(error: ArrayError) -> Self {
1273        SinkError::File(error.to_report_string())
1274    }
1275}
1276
1277impl From<RpcError> for SinkError {
1278    fn from(value: RpcError) -> Self {
1279        SinkError::Remote(anyhow!(value))
1280    }
1281}
1282
1283impl From<RedisError> for SinkError {
1284    fn from(value: RedisError) -> Self {
1285        SinkError::Redis(value.to_report_string())
1286    }
1287}
1288
1289impl From<tiberius::error::Error> for SinkError {
1290    fn from(err: tiberius::error::Error) -> Self {
1291        SinkError::SqlServer(anyhow!(err))
1292    }
1293}
1294
1295impl From<::elasticsearch::Error> for SinkError {
1296    fn from(err: ::elasticsearch::Error) -> Self {
1297        SinkError::ElasticSearchOpenSearch(anyhow!(err))
1298    }
1299}
1300
1301impl From<::opensearch::Error> for SinkError {
1302    fn from(err: ::opensearch::Error) -> Self {
1303        SinkError::ElasticSearchOpenSearch(anyhow!(err))
1304    }
1305}
1306
1307impl From<tokio_postgres::Error> for SinkError {
1308    fn from(err: tokio_postgres::Error) -> Self {
1309        SinkError::Postgres(anyhow!(err))
1310    }
1311}
1312
1313#[cfg(test)]
1314mod tests {
1315    use std::collections::BTreeMap;
1316
1317    use super::*;
1318
1319    fn btreemap<const N: usize>(entries: [(&str, &str); N]) -> BTreeMap<String, String> {
1320        entries
1321            .into_iter()
1322            .map(|(key, value)| (key.to_owned(), value.to_owned()))
1323            .collect()
1324    }
1325
1326    #[test]
1327    fn test_validate_sink_unknown_fields() {
1328        let config = crate::sink::redis::RedisConfig::from_btreemap(btreemap([
1329            (CONNECTOR_TYPE_KEY, "redis"),
1330            (SINK_TYPE_OPTION, SINK_TYPE_APPEND_ONLY),
1331            ("primary_key", "id"),
1332            (SINK_USER_FORCE_COMPACTION, "true"),
1333            (SINK_USER_PRESERVE_ROW_LEVEL_CHANGES, "true"),
1334            ("redis.url", "redis://127.0.0.1:6379"),
1335        ]))
1336        .unwrap();
1337        validate_sink_unknown_fields(&config).unwrap();
1338
1339        let config = crate::sink::redis::RedisConfig::from_btreemap(btreemap([
1340            (CONNECTOR_TYPE_KEY, "redis"),
1341            (SINK_TYPE_OPTION, SINK_TYPE_APPEND_ONLY),
1342            ("redis.url", "redis://127.0.0.1:6379"),
1343            ("bogus_with", "1"),
1344        ]))
1345        .unwrap();
1346        let err = validate_sink_unknown_fields(&config).unwrap_err();
1347        let report = err.to_report_string();
1348        assert!(report.contains("bogus_with"), "{report}");
1349
1350        let err = crate::sink::kafka::KafkaConfig::from_btreemap(btreemap([
1351            (CONNECTOR_TYPE_KEY, "kafka"),
1352            (SINK_TYPE_OPTION, SINK_TYPE_APPEND_ONLY),
1353        ]))
1354        .unwrap_err();
1355        assert!(err.to_report_string().contains("missing field `topic`"));
1356    }
1357
1358    #[test]
1359    fn test_sink_exactly_once_policy() {
1360        assert!(sink_is_exactly_once(&btreemap([(CONNECTOR_TYPE_KEY, ICEBERG_SINK)])).unwrap());
1361        assert!(
1362            !sink_is_exactly_once(&btreemap([
1363                (CONNECTOR_TYPE_KEY, ICEBERG_SINK),
1364                ("is_exactly_once", "false"),
1365            ]))
1366            .unwrap()
1367        );
1368
1369        #[cfg(feature = "sink-deltalake")]
1370        {
1371            assert!(
1372                !sink_is_exactly_once(&btreemap([(CONNECTOR_TYPE_KEY, DELTALAKE_SINK)])).unwrap()
1373            );
1374            assert!(
1375                sink_is_exactly_once(&btreemap([
1376                    (CONNECTOR_TYPE_KEY, DELTALAKE_SINK),
1377                    ("is_exactly_once", "true"),
1378                ]))
1379                .unwrap()
1380            );
1381        }
1382
1383        #[cfg(feature = "sink-lancedb")]
1384        assert!(sink_is_exactly_once(&btreemap([(CONNECTOR_TYPE_KEY, LANCEDB_SINK)])).unwrap());
1385
1386        let error = sink_is_exactly_once(&btreemap([
1387            (CONNECTOR_TYPE_KEY, ICEBERG_SINK),
1388            ("is_exactly_once", "invalid"),
1389        ]))
1390        .unwrap_err();
1391        assert!(error.to_report_string().contains("is_exactly_once"));
1392    }
1393}
risingwave_connector/sink/mod.rs

risingwave_connector/sink/
mod.rs
