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

risingwave_connector/sink/
mod.rs
