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

risingwave_connector/source/cdc/external/
mod.rs
