risingwave_connector/source/cdc/external/
mod.rs

1// Copyright 2023 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
15pub mod mock_external_table;
16pub mod postgres;
17pub mod sql_server;
18
19pub mod mysql;
20
21use std::collections::{BTreeMap, HashMap};
22
23use anyhow::{Context, anyhow};
24use futures::pin_mut;
25use futures::stream::BoxStream;
26use futures_async_stream::try_stream;
27use risingwave_common::bail;
28use risingwave_common::catalog::{ColumnDesc, Field, Schema};
29use risingwave_common::row::OwnedRow;
30use risingwave_common::secret::LocalSecretManager;
31use risingwave_pb::catalog::table::CdcTableType as PbCdcTableType;
32use risingwave_pb::secret::PbSecretRef;
33use serde::{Deserialize, Serialize};
34
35use crate::WithPropertiesExt;
36use crate::connector_common::{PgConnectionConfig, PostgresExternalTable, SslMode};
37use crate::error::{ConnectorError, ConnectorResult};
38use crate::source::CdcTableSnapshotSplit;
39use crate::source::cdc::CdcSourceType;
40use crate::source::cdc::external::mock_external_table::MockExternalTableReader;
41use crate::source::cdc::external::mysql::{
42    MySqlExternalTable, MySqlExternalTableReader, MySqlOffset,
43};
44use crate::source::cdc::external::postgres::{PostgresExternalTableReader, PostgresOffset};
45use crate::source::cdc::external::sql_server::{
46    SqlServerExternalTable, SqlServerExternalTableReader, SqlServerOffset,
47};
48
49#[derive(Debug, Clone, PartialEq, Eq, Hash)]
50pub enum ExternalCdcTableType {
51    Undefined,
52    Mock,
53    MySql,
54    Postgres,
55    SqlServer,
56    Citus,
57    Mongo,
58}
59
60impl ExternalCdcTableType {
61    pub fn from_properties(with_properties: &impl WithPropertiesExt) -> Self {
62        let connector = with_properties.get_connector().unwrap_or_default();
63        match connector.as_str() {
64            "mysql-cdc" => Self::MySql,
65            "postgres-cdc" => Self::Postgres,
66            "citus-cdc" => Self::Citus,
67            "sqlserver-cdc" => Self::SqlServer,
68            "mongodb-cdc" => Self::Mongo,
69            _ => Self::Undefined,
70        }
71    }
72
73    pub fn can_backfill(&self) -> bool {
74        matches!(self, Self::MySql | Self::Postgres | Self::SqlServer)
75    }
76
77    pub fn enable_transaction_metadata(&self) -> bool {
78        // In Debezium, transactional metadata cause delay of the newest events, as the `END` message is never sent unless a new transaction starts.
79        // So we only allow transactional metadata for MySQL and Postgres.
80        // See more in https://debezium.io/documentation/reference/2.6/connectors/sqlserver.html#sqlserver-transaction-metadata
81        matches!(self, Self::MySql | Self::Postgres)
82    }
83
84    pub async fn create_table_reader(
85        &self,
86        config: ExternalTableConfig,
87        schema: Schema,
88        pk_indices: Vec<usize>,
89        schema_table_name: SchemaTableName,
90    ) -> ConnectorResult<ExternalTableReaderImpl> {
91        match self {
92            Self::MySql => Ok(ExternalTableReaderImpl::MySql(
93                MySqlExternalTableReader::new(config, schema, pk_indices).await?,
94            )),
95            Self::Postgres => Ok(ExternalTableReaderImpl::Postgres(
96                PostgresExternalTableReader::new(config, schema, pk_indices, schema_table_name)
97                    .await?,
98            )),
99            Self::SqlServer => Ok(ExternalTableReaderImpl::SqlServer(
100                SqlServerExternalTableReader::new(config, schema, pk_indices).await?,
101            )),
102            // citus is never supported for cdc backfill (create source + create table).
103            Self::Mock => Ok(ExternalTableReaderImpl::Mock(MockExternalTableReader::new())),
104            _ => bail!("invalid external table type: {:?}", *self),
105        }
106    }
107}
108
109impl From<ExternalCdcTableType> for PbCdcTableType {
110    fn from(cdc_table_type: ExternalCdcTableType) -> Self {
111        match cdc_table_type {
112            ExternalCdcTableType::Postgres => Self::Postgres,
113            ExternalCdcTableType::MySql => Self::Mysql,
114            ExternalCdcTableType::SqlServer => Self::Sqlserver,
115
116            ExternalCdcTableType::Citus => Self::Citus,
117            ExternalCdcTableType::Mongo => Self::Mongo,
118            ExternalCdcTableType::Undefined | ExternalCdcTableType::Mock => Self::Unspecified,
119        }
120    }
121}
122
123impl From<PbCdcTableType> for ExternalCdcTableType {
124    fn from(cdc_table_type: PbCdcTableType) -> Self {
125        match cdc_table_type {
126            PbCdcTableType::Postgres => Self::Postgres,
127            PbCdcTableType::Mysql => Self::MySql,
128            PbCdcTableType::Sqlserver => Self::SqlServer,
129            PbCdcTableType::Mongo => Self::Mongo,
130            PbCdcTableType::Citus => Self::Citus,
131            PbCdcTableType::Unspecified => Self::Undefined,
132        }
133    }
134}
135
136#[derive(Debug, Clone, PartialEq)]
137pub struct SchemaTableName {
138    // namespace of the table, e.g. database in mysql, schema in postgres
139    pub schema_name: String,
140    pub table_name: String,
141}
142
143pub const TABLE_NAME_KEY: &str = "table.name";
144pub const SCHEMA_NAME_KEY: &str = "schema.name";
145pub const DATABASE_NAME_KEY: &str = "database.name";
146
147impl SchemaTableName {
148    pub fn from_properties(properties: &BTreeMap<String, String>) -> Self {
149        let table_type = ExternalCdcTableType::from_properties(properties);
150        let table_name = properties.get(TABLE_NAME_KEY).cloned().unwrap_or_default();
151
152        let schema_name = match table_type {
153            ExternalCdcTableType::MySql => properties
154                .get(DATABASE_NAME_KEY)
155                .cloned()
156                .unwrap_or_default(),
157            ExternalCdcTableType::Postgres | ExternalCdcTableType::Citus => {
158                properties.get(SCHEMA_NAME_KEY).cloned().unwrap_or_default()
159            }
160            ExternalCdcTableType::SqlServer => {
161                properties.get(SCHEMA_NAME_KEY).cloned().unwrap_or_default()
162            }
163            _ => {
164                unreachable!("invalid external table type: {:?}", table_type);
165            }
166        };
167
168        Self {
169            schema_name,
170            table_name,
171        }
172    }
173}
174
175#[derive(Debug, Clone, PartialEq, PartialOrd, Serialize, Deserialize)]
176pub enum CdcOffset {
177    MySql(MySqlOffset),
178    Postgres(PostgresOffset),
179    SqlServer(SqlServerOffset),
180}
181
182// Example debezium offset for Postgres:
183// {
184//     "sourcePartition":
185//     {
186//         "server": "RW_CDC_1004"
187//     },
188//     "sourceOffset":
189//     {
190//         "last_snapshot_record": false,
191//         "lsn": 29973552,
192//         "txId": 1046,
193//         "ts_usec": 1670826189008456,
194//         "snapshot": true
195//     }
196// }
197#[derive(Debug, Clone, Serialize, Deserialize)]
198pub struct DebeziumOffset {
199    #[serde(rename = "sourcePartition")]
200    pub source_partition: HashMap<String, String>,
201    #[serde(rename = "sourceOffset")]
202    pub source_offset: DebeziumSourceOffset,
203    #[serde(rename = "isHeartbeat")]
204    pub is_heartbeat: bool,
205}
206
207#[derive(Debug, Default, Clone, Serialize, Deserialize)]
208pub struct DebeziumSourceOffset {
209    // postgres snapshot progress
210    pub last_snapshot_record: Option<bool>,
211    // mysql snapshot progress
212    pub snapshot: Option<bool>,
213
214    // mysql binlog offset
215    pub file: Option<String>,
216    pub pos: Option<u64>,
217
218    // postgres offset
219    pub lsn: Option<u64>,
220    #[serde(rename = "txId")]
221    pub txid: Option<i64>,
222    pub tx_usec: Option<u64>,
223    pub lsn_commit: Option<u64>,
224    pub lsn_proc: Option<u64>,
225
226    // sql server offset
227    pub commit_lsn: Option<String>,
228    pub change_lsn: Option<String>,
229}
230
231pub type CdcOffsetParseFunc = Box<dyn Fn(&str) -> ConnectorResult<CdcOffset> + Send>;
232
233pub trait ExternalTableReader: Sized {
234    async fn current_cdc_offset(&self) -> ConnectorResult<CdcOffset>;
235
236    // Currently, MySQL cdc uses a connection pool to manage connections to MySQL, and other CDC processes do not require the disconnect step for now.
237
238    async fn disconnect(self) -> ConnectorResult<()> {
239        Ok(())
240    }
241
242    fn snapshot_read(
243        &self,
244        table_name: SchemaTableName,
245        start_pk: Option<OwnedRow>,
246        primary_keys: Vec<String>,
247        limit: u32,
248    ) -> BoxStream<'_, ConnectorResult<OwnedRow>>;
249
250    fn get_parallel_cdc_splits(
251        &self,
252        options: CdcTableSnapshotSplitOption,
253    ) -> BoxStream<'_, ConnectorResult<CdcTableSnapshotSplit>>;
254
255    fn split_snapshot_read(
256        &self,
257        table_name: SchemaTableName,
258        left: OwnedRow,
259        right: OwnedRow,
260        split_columns: Vec<Field>,
261    ) -> BoxStream<'_, ConnectorResult<OwnedRow>>;
262}
263
264pub struct CdcTableSnapshotSplitOption {
265    pub backfill_num_rows_per_split: u64,
266    pub backfill_as_even_splits: bool,
267    pub backfill_split_pk_column_index: u32,
268}
269
270pub enum ExternalTableReaderImpl {
271    MySql(MySqlExternalTableReader),
272    Postgres(PostgresExternalTableReader),
273    SqlServer(SqlServerExternalTableReader),
274    Mock(MockExternalTableReader),
275}
276
277#[derive(Debug, Default, Clone, Deserialize)]
278pub struct ExternalTableConfig {
279    pub connector: String,
280
281    #[serde(rename = "hostname")]
282    pub host: String,
283    pub port: String,
284    pub username: String,
285    pub password: String,
286    #[serde(rename = "database.name")]
287    pub database: String,
288    #[serde(rename = "schema.name", default = "Default::default")]
289    pub schema: String,
290    #[serde(rename = "table.name")]
291    pub table: String,
292    /// `ssl.mode` specifies the SSL/TLS encryption level for secure communication with Postgres.
293    /// Choices include `disabled`, `preferred`, and `required`.
294    /// This field is optional.
295    #[serde(rename = "ssl.mode", default = "postgres_ssl_mode_default")]
296    #[serde(alias = "debezium.database.sslmode")]
297    pub ssl_mode: SslMode,
298
299    #[serde(rename = "ssl.root.cert")]
300    #[serde(alias = "debezium.database.sslrootcert")]
301    pub ssl_root_cert: Option<String>,
302
303    /// `encrypt` specifies whether connect to SQL Server using SSL.
304    /// Only "true" means using SSL. All other values are treated as "false".
305    #[serde(rename = "database.encrypt", default = "Default::default")]
306    pub encrypt: String,
307}
308
309fn postgres_ssl_mode_default() -> SslMode {
310    // NOTE(StrikeW): Default to `disabled` for backward compatibility
311    SslMode::Disabled
312}
313
314impl ExternalTableConfig {
315    pub fn try_from_btreemap(
316        connect_properties: BTreeMap<String, String>,
317        secret_refs: BTreeMap<String, PbSecretRef>,
318    ) -> ConnectorResult<Self> {
319        let options_with_secret =
320            LocalSecretManager::global().fill_secrets(connect_properties, secret_refs)?;
321        let json_value = serde_json::to_value(options_with_secret)?;
322        let config = serde_json::from_value::<ExternalTableConfig>(json_value)?;
323        Ok(config)
324    }
325
326    /// Project the Postgres-specific subset of this config so it can drive the
327    /// shared `create_pg_client` / `PostgresExternalTable` helpers. Only meaningful
328    /// for the Postgres CDC connector; other connectors ignore the fields.
329    pub fn pg_connection_config(&self) -> ConnectorResult<PgConnectionConfig> {
330        let port = self
331            .port
332            .parse::<u16>()
333            .with_context(|| format!("invalid postgres port `{}`", self.port))?;
334        Ok(PgConnectionConfig {
335            host: self.host.clone(),
336            port,
337            user: self.username.clone(),
338            password: self.password.clone(),
339            database: self.database.clone(),
340            ssl_mode: self.ssl_mode.clone(),
341            ssl_root_cert: self.ssl_root_cert.clone(),
342        })
343    }
344}
345
346impl ExternalTableReader for ExternalTableReaderImpl {
347    async fn current_cdc_offset(&self) -> ConnectorResult<CdcOffset> {
348        match self {
349            ExternalTableReaderImpl::MySql(mysql) => mysql.current_cdc_offset().await,
350            ExternalTableReaderImpl::Postgres(postgres) => postgres.current_cdc_offset().await,
351            ExternalTableReaderImpl::SqlServer(sql_server) => sql_server.current_cdc_offset().await,
352            ExternalTableReaderImpl::Mock(mock) => mock.current_cdc_offset().await,
353        }
354    }
355
356    fn snapshot_read(
357        &self,
358        table_name: SchemaTableName,
359        start_pk: Option<OwnedRow>,
360        primary_keys: Vec<String>,
361        limit: u32,
362    ) -> BoxStream<'_, ConnectorResult<OwnedRow>> {
363        self.snapshot_read_inner(table_name, start_pk, primary_keys, limit)
364    }
365
366    fn get_parallel_cdc_splits(
367        &self,
368        options: CdcTableSnapshotSplitOption,
369    ) -> BoxStream<'_, ConnectorResult<CdcTableSnapshotSplit>> {
370        self.get_parallel_cdc_splits_inner(options)
371    }
372
373    fn split_snapshot_read(
374        &self,
375        table_name: SchemaTableName,
376        left: OwnedRow,
377        right: OwnedRow,
378        split_columns: Vec<Field>,
379    ) -> BoxStream<'_, ConnectorResult<OwnedRow>> {
380        self.split_snapshot_read_inner(table_name, left, right, split_columns)
381    }
382}
383
384impl ExternalTableReaderImpl {
385    /// For each given primary key column (by name), returns whether comparing the RisingWave
386    /// `i64` value needs upstream unsigned `BIGINT` semantics. Only MySQL `BIGINT UNSIGNED` can
387    /// overflow into a negative `i64`; other connectors are always false.
388    pub fn pk_column_unsigned_i64_compare_flags(
389        &self,
390        pk_names: &[String],
391    ) -> ConnectorResult<Vec<bool>> {
392        match self {
393            ExternalTableReaderImpl::MySql(mysql) => {
394                mysql.pk_column_unsigned_i64_compare_flags(pk_names)
395            }
396            _ => Ok(vec![false; pk_names.len()]),
397        }
398    }
399
400    pub fn get_cdc_offset_parser(&self) -> CdcOffsetParseFunc {
401        match self {
402            ExternalTableReaderImpl::MySql(_) => MySqlExternalTableReader::get_cdc_offset_parser(),
403            ExternalTableReaderImpl::Postgres(_) => {
404                PostgresExternalTableReader::get_cdc_offset_parser()
405            }
406            ExternalTableReaderImpl::SqlServer(_) => {
407                SqlServerExternalTableReader::get_cdc_offset_parser()
408            }
409            ExternalTableReaderImpl::Mock(_) => MockExternalTableReader::get_cdc_offset_parser(),
410        }
411    }
412
413    #[try_stream(boxed, ok = OwnedRow, error = ConnectorError)]
414    async fn snapshot_read_inner(
415        &self,
416        table_name: SchemaTableName,
417        start_pk: Option<OwnedRow>,
418        primary_keys: Vec<String>,
419        limit: u32,
420    ) {
421        let stream = match self {
422            ExternalTableReaderImpl::MySql(mysql) => {
423                mysql.snapshot_read(table_name, start_pk, primary_keys, limit)
424            }
425            ExternalTableReaderImpl::Postgres(postgres) => {
426                postgres.snapshot_read(table_name, start_pk, primary_keys, limit)
427            }
428            ExternalTableReaderImpl::SqlServer(sql_server) => {
429                sql_server.snapshot_read(table_name, start_pk, primary_keys, limit)
430            }
431            ExternalTableReaderImpl::Mock(mock) => {
432                mock.snapshot_read(table_name, start_pk, primary_keys, limit)
433            }
434        };
435
436        pin_mut!(stream);
437        #[for_await]
438        for row in stream {
439            let row = row?;
440            yield row;
441        }
442    }
443
444    #[try_stream(boxed, ok = CdcTableSnapshotSplit, error = ConnectorError)]
445    async fn get_parallel_cdc_splits_inner(&self, options: CdcTableSnapshotSplitOption) {
446        let stream = match self {
447            ExternalTableReaderImpl::MySql(e) => e.get_parallel_cdc_splits(options),
448            ExternalTableReaderImpl::Postgres(e) => e.get_parallel_cdc_splits(options),
449            ExternalTableReaderImpl::SqlServer(e) => e.get_parallel_cdc_splits(options),
450            ExternalTableReaderImpl::Mock(e) => e.get_parallel_cdc_splits(options),
451        };
452        pin_mut!(stream);
453        #[for_await]
454        for row in stream {
455            let row = row?;
456            yield row;
457        }
458    }
459
460    #[try_stream(boxed, ok = OwnedRow, error = ConnectorError)]
461    async fn split_snapshot_read_inner(
462        &self,
463        table_name: SchemaTableName,
464        left: OwnedRow,
465        right: OwnedRow,
466        split_columns: Vec<Field>,
467    ) {
468        let stream = match self {
469            ExternalTableReaderImpl::MySql(mysql) => {
470                mysql.split_snapshot_read(table_name, left, right, split_columns)
471            }
472            ExternalTableReaderImpl::Postgres(postgres) => {
473                postgres.split_snapshot_read(table_name, left, right, split_columns)
474            }
475            ExternalTableReaderImpl::SqlServer(sql_server) => {
476                sql_server.split_snapshot_read(table_name, left, right, split_columns)
477            }
478            ExternalTableReaderImpl::Mock(mock) => {
479                mock.split_snapshot_read(table_name, left, right, split_columns)
480            }
481        };
482
483        pin_mut!(stream);
484        #[for_await]
485        for row in stream {
486            let row = row?;
487            yield row;
488        }
489    }
490}
491
492pub enum ExternalTableImpl {
493    MySql(MySqlExternalTable),
494    Postgres(PostgresExternalTable),
495    SqlServer(SqlServerExternalTable),
496}
497
498impl ExternalTableImpl {
499    pub async fn connect(config: ExternalTableConfig) -> ConnectorResult<Self> {
500        let cdc_source_type = CdcSourceType::from(config.connector.as_str());
501        match cdc_source_type {
502            CdcSourceType::Mysql => Ok(ExternalTableImpl::MySql(
503                MySqlExternalTable::connect(config).await?,
504            )),
505            CdcSourceType::Postgres => {
506                let pg_conn = config.pg_connection_config()?;
507                Ok(ExternalTableImpl::Postgres(
508                    PostgresExternalTable::connect(
509                        &pg_conn,
510                        &config.schema,
511                        &config.table,
512                        false,
513                        Some("SELECT"),
514                    )
515                    .await?,
516                ))
517            }
518            CdcSourceType::SqlServer => Ok(ExternalTableImpl::SqlServer(
519                SqlServerExternalTable::connect(config).await?,
520            )),
521            _ => Err(anyhow!("Unsupported cdc connector type: {}", config.connector).into()),
522        }
523    }
524
525    pub fn column_descs(&self) -> &Vec<ColumnDesc> {
526        match self {
527            ExternalTableImpl::MySql(mysql) => mysql.column_descs(),
528            ExternalTableImpl::Postgres(postgres) => postgres.column_descs(),
529            ExternalTableImpl::SqlServer(sql_server) => sql_server.column_descs(),
530        }
531    }
532
533    pub fn pk_names(&self) -> &Vec<String> {
534        match self {
535            ExternalTableImpl::MySql(mysql) => mysql.pk_names(),
536            ExternalTableImpl::Postgres(postgres) => postgres.pk_names(),
537            ExternalTableImpl::SqlServer(sql_server) => sql_server.pk_names(),
538        }
539    }
540}
541
542pub const CDC_TABLE_SPLIT_ID_START: i64 = 1;
risingwave_connector/source/cdc/external/mod.rs

risingwave_connector/source/cdc/external/
mod.rs
