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

risingwave_connector/source/cdc/external/
mod.rs
