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

risingwave_connector/source/cdc/external/
mod.rs
