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

risingwave_connector/sink/
mod.rs
