risingwave_connector/source/cdc/external/
mysql.rs

1// Copyright 2024 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
15use std::collections::HashMap;
16
17use anyhow::{Context, anyhow};
18use chrono::{DateTime, NaiveDateTime};
19use futures::stream::BoxStream;
20use futures::{StreamExt, pin_mut, stream};
21use futures_async_stream::try_stream;
22use itertools::Itertools;
23use mysql_async::prelude::*;
24use mysql_common::params::Params;
25use mysql_common::value::Value;
26use risingwave_common::bail;
27use risingwave_common::catalog::{
28    CDC_OFFSET_COLUMN_NAME, CdcKeyComparison, ColumnDesc, ColumnId, Field, Schema,
29};
30use risingwave_common::row::OwnedRow;
31use risingwave_common::types::{DataType, Datum, Decimal, F32, ScalarImpl};
32use risingwave_common::util::iter_util::ZipEqFast;
33use sea_schema::mysql::def::{ColumnDefault, ColumnType, IndexInfo, NumericAttr};
34use sea_schema::mysql::discovery::SchemaDiscovery;
35use sea_schema::mysql::query::SchemaQueryBuilder;
36use sea_schema::sea_query::{Alias, IntoIden};
37use serde::{Deserialize, Serialize};
38use sqlx::MySqlPool;
39use sqlx::mysql::MySqlConnectOptions;
40use thiserror_ext::AsReport;
41
42use crate::connector_common::SslMode;
43// Re-export SslMode for convenience
44pub use crate::connector_common::SslMode as MySqlSslMode;
45use crate::error::{ConnectorError, ConnectorResult};
46use crate::parser::mysql_row_to_owned_row_with_strict_pk;
47use crate::source::CdcTableSnapshotSplit;
48use crate::source::cdc::external::{
49    CdcOffset, CdcOffsetParseFunc, CdcTableSnapshotSplitOption, DebeziumOffset,
50    ExternalTableConfig, ExternalTableReader, SchemaTableName,
51};
52
53/// Build MySQL connection pool with proper SSL configuration.
54///
55/// This helper function creates a `mysql_async::Pool` with all necessary configurations
56/// including SSL settings. Use this function to ensure consistent MySQL connection setup
57/// across the codebase.
58///
59/// # Arguments
60/// * `host` - MySQL server hostname or IP address
61/// * `port` - MySQL server port
62/// * `username` - MySQL username
63/// * `password` - MySQL password
64/// * `database` - Database name
65/// * `ssl_mode` - SSL mode configuration (disabled, preferred, required, verify-ca, verify-full)
66///
67/// # Returns
68/// Returns a configured `mysql_async::Pool` ready for use
69pub fn build_mysql_connection_pool(
70    host: &str,
71    port: u16,
72    username: &str,
73    password: &str,
74    database: &str,
75    ssl_mode: SslMode,
76) -> mysql_async::Pool {
77    let mut opts_builder = mysql_async::OptsBuilder::default()
78        .user(Some(username))
79        .pass(Some(password))
80        .ip_or_hostname(host)
81        .tcp_port(port)
82        .db_name(Some(database));
83
84    opts_builder = match ssl_mode {
85        SslMode::Disabled | SslMode::Preferred => opts_builder.ssl_opts(None),
86        // verify-ca and verify-full are same as required for mysql now
87        SslMode::Required | SslMode::VerifyCa | SslMode::VerifyFull => {
88            let ssl_without_verify = mysql_async::SslOpts::default()
89                .with_danger_accept_invalid_certs(true)
90                .with_danger_skip_domain_validation(true);
91            opts_builder.ssl_opts(Some(ssl_without_verify))
92        }
93    };
94
95    mysql_async::Pool::new(opts_builder)
96}
97
98#[derive(Debug, Clone, Default, PartialEq, PartialOrd, Serialize, Deserialize)]
99pub struct MySqlOffset {
100    pub filename: String,
101    pub position: u64,
102}
103
104impl MySqlOffset {
105    pub fn new(filename: String, position: u64) -> Self {
106        Self { filename, position }
107    }
108}
109
110impl MySqlOffset {
111    pub fn parse_debezium_offset(offset: &str) -> ConnectorResult<Self> {
112        let dbz_offset: DebeziumOffset = serde_json::from_str(offset)
113            .with_context(|| format!("invalid upstream offset: {}", offset))?;
114
115        Ok(Self {
116            filename: dbz_offset
117                .source_offset
118                .file
119                .context("binlog file not found in offset")?,
120            position: dbz_offset
121                .source_offset
122                .pos
123                .context("binlog position not found in offset")?,
124        })
125    }
126}
127
128pub struct MySqlExternalTable {
129    column_descs: Vec<ColumnDesc>,
130    pk_names: Vec<String>,
131    pk_comparisons: Vec<CdcKeyComparison>,
132}
133
134impl MySqlExternalTable {
135    pub async fn connect(config: ExternalTableConfig) -> ConnectorResult<Self> {
136        tracing::debug!("connect to mysql");
137        let options = MySqlConnectOptions::new()
138            .username(&config.username)
139            .password(&config.password)
140            .host(&config.host)
141            .port(config.port.parse::<u16>().unwrap())
142            .database(&config.database)
143            .ssl_mode(match config.ssl_mode {
144                SslMode::Disabled => sqlx::mysql::MySqlSslMode::Disabled,
145                SslMode::Preferred => sqlx::mysql::MySqlSslMode::Preferred,
146                SslMode::Required => sqlx::mysql::MySqlSslMode::Required,
147                _ => {
148                    return Err(anyhow!("unsupported SSL mode").into());
149                }
150            });
151
152        let connection = MySqlPool::connect_with(options).await?;
153        let mut schema_discovery = SchemaDiscovery::new(connection, config.database.as_str());
154
155        // discover system version first
156        let system_info = schema_discovery.discover_system().await?;
157        schema_discovery.query = SchemaQueryBuilder::new(system_info.clone());
158        let schema = Alias::new(config.database.as_str()).into_iden();
159        let table = Alias::new(config.table.as_str()).into_iden();
160        let columns = schema_discovery
161            .discover_columns(schema.clone(), table.clone(), &system_info)
162            .await?;
163        let indexes = schema_discovery.discover_indexes(schema, table).await?;
164        let pk_names = primary_key_names(&indexes)
165            .ok_or_else(|| anyhow!("MySQL table doesn't define the primary key"))?;
166        let pk_comparisons = pk_names
167            .iter()
168            .map(|pk_name| {
169                let column = columns
170                    .iter()
171                    .find(|column| column.name.eq_ignore_ascii_case(pk_name))
172                    .ok_or_else(|| {
173                        anyhow!("primary key column `{pk_name}` not found in upstream MySQL schema")
174                    })?;
175                Ok(if mysql_type_is_unsigned_bigint(&column.col_type) {
176                    CdcKeyComparison::UnsignedInt64
177                } else {
178                    CdcKeyComparison::Native
179                })
180            })
181            .collect::<ConnectorResult<Vec<_>>>()?;
182        let mut column_descs = vec![];
183        for col in columns {
184            let data_type = mysql_type_to_rw_type(&col.col_type)?;
185            // column name in mysql is case-insensitive, convert to lowercase
186            let col_name = col.name.to_lowercase();
187            let column_desc = if let Some(default) = col.default {
188                let snapshot_value = derive_default_value(default.clone(), &data_type)
189                    .unwrap_or_else(|e| {
190                        tracing::warn!(
191                            column = col_name,
192                            ?default,
193                            %data_type,
194                            error = %e.as_report(),
195                            "failed to derive column default value, fallback to `NULL`",
196                        );
197                        None
198                    });
199
200                ColumnDesc::named_with_default_value(
201                    col_name.clone(),
202                    ColumnId::placeholder(),
203                    data_type.clone(),
204                    snapshot_value,
205                )
206            } else {
207                ColumnDesc::named(col_name.clone(), ColumnId::placeholder(), data_type)
208            };
209
210            column_descs.push(column_desc);
211        }
212
213        Ok(Self {
214            column_descs,
215            pk_names,
216            pk_comparisons,
217        })
218    }
219
220    pub fn column_descs(&self) -> &Vec<ColumnDesc> {
221        &self.column_descs
222    }
223
224    pub fn pk_names(&self) -> &Vec<String> {
225        &self.pk_names
226    }
227
228    pub fn pk_column_comparisons(
229        &self,
230        pk_names: &[String],
231    ) -> ConnectorResult<Vec<CdcKeyComparison>> {
232        pk_names
233            .iter()
234            .map(|pk_name| {
235                self.pk_names
236                    .iter()
237                    .position(|name| name.eq_ignore_ascii_case(pk_name))
238                    .map(|idx| self.pk_comparisons[idx])
239                    .ok_or_else(|| {
240                        anyhow!("primary key column `{pk_name}` not found in upstream MySQL schema")
241                            .into()
242                    })
243            })
244            .collect()
245    }
246
247    pub async fn discover_pk_column_comparisons(
248        config: &ExternalTableConfig,
249        pk_names: &[String],
250    ) -> ConnectorResult<Vec<CdcKeyComparison>> {
251        let pool = build_mysql_connection_pool(
252            &config.host,
253            config.port.parse::<u16>().unwrap(),
254            &config.username,
255            &config.password,
256            &config.database,
257            config.ssl_mode.clone(),
258        );
259        let pk_infos = MySqlExternalTableReader::query_upstream_pk_infos(
260            &pool,
261            &config.database,
262            &config.table,
263        )
264        .await?;
265        pool.disconnect().await?;
266
267        pk_column_comparisons_from_infos(&pk_infos, pk_names)
268    }
269}
270
271fn primary_key_names(indexes: &[IndexInfo]) -> Option<Vec<String>> {
272    indexes
273        .iter()
274        .find(|index| index.name.eq_ignore_ascii_case("PRIMARY"))
275        .map(|index| {
276            index
277                .parts
278                .iter()
279                .map(|part| part.column.to_lowercase())
280                .collect()
281        })
282        .filter(|names: &Vec<_>| !names.is_empty())
283}
284
285fn derive_default_value(default: ColumnDefault, data_type: &DataType) -> ConnectorResult<Datum> {
286    let datum = match default {
287        ColumnDefault::Null => None,
288        ColumnDefault::Int(val) => match data_type {
289            DataType::Int16 => Some(ScalarImpl::Int16(val as _)),
290            DataType::Int32 => Some(ScalarImpl::Int32(val as _)),
291            DataType::Int64 => Some(ScalarImpl::Int64(val)),
292            DataType::Varchar => {
293                // should be the Enum type which is mapped to Varchar
294                Some(ScalarImpl::from(val.to_string()))
295            }
296            _ => bail!("unexpected default value type for integer"),
297        },
298        ColumnDefault::Real(val) => match data_type {
299            DataType::Float32 => Some(ScalarImpl::Float32(F32::from(val as f32))),
300            DataType::Float64 => Some(ScalarImpl::Float64(val.into())),
301            DataType::Decimal => Some(ScalarImpl::Decimal(
302                Decimal::try_from(val).context("failed to convert default value to decimal")?,
303            )),
304            _ => bail!("unexpected default value type for real"),
305        },
306        ColumnDefault::String(mut val) => {
307            // mysql timestamp is mapped to timestamptz, we use UTC timezone to
308            // interpret its value
309            if data_type == &DataType::Timestamptz {
310                val = timestamp_val_to_timestamptz(val.as_str())?;
311            }
312            Some(ScalarImpl::from_text(val.as_str(), data_type).map_err(|e| anyhow!(e)).context(
313                "failed to parse mysql default value expression, only constant is supported",
314            )?)
315        }
316        ColumnDefault::CurrentTimestamp | ColumnDefault::CustomExpr(_) => {
317            bail!("MySQL CURRENT_TIMESTAMP and custom expression default value not supported")
318        }
319    };
320    Ok(datum)
321}
322
323pub fn timestamp_val_to_timestamptz(value_text: &str) -> ConnectorResult<String> {
324    let format = "%Y-%m-%d %H:%M:%S";
325    let naive_datetime = NaiveDateTime::parse_from_str(value_text, format)
326        .map_err(|err| anyhow!("failed to parse mysql timestamp value").context(err))?;
327    let postgres_timestamptz: DateTime<chrono::Utc> =
328        DateTime::<chrono::Utc>::from_naive_utc_and_offset(naive_datetime, chrono::Utc);
329    Ok(postgres_timestamptz
330        .format("%Y-%m-%d %H:%M:%S%:z")
331        .to_string())
332}
333
334pub fn type_name_to_mysql_type(ty_name: &str) -> Option<ColumnType> {
335    // Debezium schema change message may include extra qualifiers, e.g. `BIGINT UNSIGNED`,
336    // `BIGINT(20) UNSIGNED`, `INT UNSIGNED ZEROFILL`, etc.
337    let ty = ty_name.trim().to_lowercase();
338    let tokens = ty
339        .split(|c: char| c.is_whitespace() || matches!(c, '(' | ')' | ','))
340        .filter(|token| !token.is_empty())
341        .collect_vec();
342    let base = tokens.first().copied().unwrap_or_default();
343    let second = tokens.get(1).copied();
344    let is_unsigned = tokens.contains(&"unsigned");
345    let is_zero_fill = tokens.contains(&"zerofill");
346
347    let make_numeric_attr = || {
348        let mut attr = NumericAttr::default();
349        if is_unsigned {
350            attr.unsigned = Some(true);
351        }
352        if is_zero_fill {
353            attr.zero_fill = Some(true);
354        }
355        attr
356    };
357
358    match (base, second) {
359        ("character", Some("varying")) => return Some(ColumnType::Varchar(Default::default())),
360        ("double", Some("precision")) => return Some(ColumnType::Double(make_numeric_attr())),
361        ("long", Some("varchar")) => return Some(ColumnType::MediumText(Default::default())),
362        ("long", Some("varbinary")) => return Some(ColumnType::MediumBlob),
363        _ => {}
364    }
365
366    match base {
367        "serial" => Some(ColumnType::Serial),
368        "bit" => Some(ColumnType::Bit(make_numeric_attr())),
369        "tinyint" | "int1" => Some(ColumnType::TinyInt(make_numeric_attr())),
370        "bool" | "boolean" => Some(ColumnType::Bool),
371        "smallint" | "int2" => Some(ColumnType::SmallInt(make_numeric_attr())),
372        "mediumint" | "middleint" | "int3" => Some(ColumnType::MediumInt(make_numeric_attr())),
373        "int" | "integer" | "int4" => Some(ColumnType::Int(make_numeric_attr())),
374        "bigint" | "int8" => Some(ColumnType::BigInt(make_numeric_attr())),
375        "decimal" | "dec" | "fixed" | "numeric" => Some(ColumnType::Decimal(make_numeric_attr())),
376        "float" | "float4" => Some(ColumnType::Float(make_numeric_attr())),
377        "double" | "float8" | "real" => Some(ColumnType::Double(make_numeric_attr())),
378        "time" => Some(ColumnType::Time(Default::default())),
379        "datetime" => Some(ColumnType::DateTime(Default::default())),
380        "timestamp" => Some(ColumnType::Timestamp(Default::default())),
381        "year" => Some(ColumnType::Year),
382        "char" | "character" => Some(ColumnType::Char(Default::default())),
383        "nchar" => Some(ColumnType::NChar(Default::default())),
384        "varchar" => Some(ColumnType::Varchar(Default::default())),
385        "nvarchar" => Some(ColumnType::NVarchar(Default::default())),
386        "binary" => Some(ColumnType::Binary(Default::default())),
387        "varbinary" => Some(ColumnType::Varbinary(Default::default())),
388        "text" => Some(ColumnType::Text(Default::default())),
389        "tinytext" => Some(ColumnType::TinyText(Default::default())),
390        "mediumtext" => Some(ColumnType::MediumText(Default::default())),
391        "longtext" => Some(ColumnType::LongText(Default::default())),
392        "blob" => Some(ColumnType::Blob(Default::default())),
393        "tinyblob" => Some(ColumnType::TinyBlob),
394        "mediumblob" => Some(ColumnType::MediumBlob),
395        "longblob" => Some(ColumnType::LongBlob),
396        "enum" => Some(ColumnType::Enum(Default::default())),
397        "set" => Some(ColumnType::Set(Default::default())),
398        "json" => Some(ColumnType::Json),
399        "date" => Some(ColumnType::Date),
400        "geometry" => Some(ColumnType::Geometry(Default::default())),
401        "point" => Some(ColumnType::Point(Default::default())),
402        "linestring" => Some(ColumnType::LineString(Default::default())),
403        "polygon" => Some(ColumnType::Polygon(Default::default())),
404        "multipoint" => Some(ColumnType::MultiPoint(Default::default())),
405        "multilinestring" => Some(ColumnType::MultiLineString(Default::default())),
406        "multipolygon" => Some(ColumnType::MultiPolygon(Default::default())),
407        "geometrycollection" => Some(ColumnType::GeometryCollection(Default::default())),
408        _ => None,
409    }
410}
411
412fn mysql_type_is_unsigned_bigint(col_type: &ColumnType) -> bool {
413    match col_type {
414        // MySQL SERIAL is an alias for BIGINT UNSIGNED NOT NULL AUTO_INCREMENT UNIQUE.
415        ColumnType::Serial => true,
416        ColumnType::BigInt(attr) => attr.unsigned == Some(true),
417        _ => false,
418    }
419}
420
421fn pk_column_comparisons_from_infos(
422    pk_infos: &[(String, ColumnType)],
423    pk_names: &[String],
424) -> ConnectorResult<Vec<CdcKeyComparison>> {
425    pk_names
426        .iter()
427        .map(|pk_name| {
428            pk_infos
429                .iter()
430                .find(|(name, _)| name.eq_ignore_ascii_case(pk_name))
431                .map(|(_, col_type)| {
432                    if mysql_type_is_unsigned_bigint(col_type) {
433                        CdcKeyComparison::UnsignedInt64
434                    } else {
435                        CdcKeyComparison::Native
436                    }
437                })
438                .ok_or_else(|| {
439                    anyhow!(
440                        "primary key column `{pk_name}` not found in upstream MySQL primary key info"
441                    )
442                    .into()
443                })
444        })
445        .collect()
446}
447
448pub fn mysql_type_to_rw_type(col_type: &ColumnType) -> ConnectorResult<DataType> {
449    let dtype = match col_type {
450        ColumnType::Serial => DataType::Decimal,
451        ColumnType::Bit(attr) => {
452            if let Some(1) = attr.maximum {
453                DataType::Boolean
454            } else {
455                return Err(
456                    anyhow!("BIT({}) type not supported", attr.maximum.unwrap_or(0)).into(),
457                );
458            }
459        }
460        // Unsigned integer family needs promotion to avoid overflow.
461        ColumnType::TinyInt(_) => DataType::Int16,
462        ColumnType::SmallInt(attr) => {
463            if attr.unsigned == Some(true) {
464                DataType::Int32
465            } else {
466                DataType::Int16
467            }
468        }
469        ColumnType::Bool => DataType::Boolean,
470        ColumnType::MediumInt(_) => DataType::Int32,
471        ColumnType::Int(attr) => {
472            if attr.unsigned == Some(true) {
473                DataType::Int64
474            } else {
475                DataType::Int32
476            }
477        }
478        ColumnType::BigInt(attr) => {
479            if attr.unsigned == Some(true) {
480                DataType::Decimal
481            } else {
482                DataType::Int64
483            }
484        }
485        ColumnType::Decimal(_) => DataType::Decimal,
486        ColumnType::Float(_) => DataType::Float32,
487        ColumnType::Double(_) => DataType::Float64,
488        ColumnType::Date => DataType::Date,
489        ColumnType::Time(_) => DataType::Time,
490        ColumnType::DateTime(_) => DataType::Timestamp,
491        ColumnType::Timestamp(_) => DataType::Timestamptz,
492        ColumnType::Year => DataType::Int32,
493        ColumnType::Char(_)
494        | ColumnType::NChar(_)
495        | ColumnType::Varchar(_)
496        | ColumnType::NVarchar(_) => DataType::Varchar,
497        ColumnType::Binary(_) | ColumnType::Varbinary(_) => DataType::Bytea,
498        ColumnType::Text(_)
499        | ColumnType::TinyText(_)
500        | ColumnType::MediumText(_)
501        | ColumnType::LongText(_) => DataType::Varchar,
502        ColumnType::Blob(_)
503        | ColumnType::TinyBlob
504        | ColumnType::MediumBlob
505        | ColumnType::LongBlob => DataType::Bytea,
506        ColumnType::Enum(_) => DataType::Varchar,
507        ColumnType::Json => DataType::Jsonb,
508        ColumnType::Set(_) => {
509            return Err(anyhow!("SET type not supported").into());
510        }
511        ColumnType::Geometry(_) => {
512            return Err(anyhow!("GEOMETRY type not supported").into());
513        }
514        ColumnType::Point(_) => {
515            return Err(anyhow!("POINT type not supported").into());
516        }
517        ColumnType::LineString(_) => {
518            return Err(anyhow!("LINE string type not supported").into());
519        }
520        ColumnType::Polygon(_) => {
521            return Err(anyhow!("POLYGON type not supported").into());
522        }
523        ColumnType::MultiPoint(_) => {
524            return Err(anyhow!("MULTI POINT type not supported").into());
525        }
526        ColumnType::MultiLineString(_) => {
527            return Err(anyhow!("MULTI LINE STRING type not supported").into());
528        }
529        ColumnType::MultiPolygon(_) => {
530            return Err(anyhow!("MULTI POLYGON type not supported").into());
531        }
532        ColumnType::GeometryCollection(_) => {
533            return Err(anyhow!("GEOMETRY COLLECTION type not supported").into());
534        }
535        ColumnType::Unknown(_) => {
536            return Err(anyhow!("Unknown MySQL data type").into());
537        }
538    };
539
540    Ok(dtype)
541}
542
543pub struct MySqlExternalTableReader {
544    rw_schema: Schema,
545    pk_indices: Vec<usize>,
546    field_names: String,
547    pool: mysql_async::Pool,
548    upstream_mysql_pk_infos: Vec<(String, ColumnType)>, // (column_name, column_type)
549    mysql_version: (u8, u8),
550    is_mariadb: bool,
551}
552
553impl ExternalTableReader for MySqlExternalTableReader {
554    async fn current_cdc_offset(&self) -> ConnectorResult<CdcOffset> {
555        let mut conn = self.pool.get_conn().await?;
556
557        // Choose SQL command based on MySQL version
558        let sql = if !self.is_mariadb && self.is_mysql_8_4_or_later() {
559            "SHOW BINARY LOG STATUS"
560        } else {
561            "SHOW MASTER STATUS"
562        };
563
564        tracing::debug!(
565            "Using SQL command: {} for MySQL version {}.{} (is_mariadb={})",
566            sql,
567            self.mysql_version.0,
568            self.mysql_version.1,
569            self.is_mariadb
570        );
571        let mut rs = conn.query::<mysql_async::Row, _>(sql).await?;
572        let row = Itertools::exactly_one(rs.iter_mut())
573            .ok()
574            .context("expect exactly one row when reading binlog offset")?;
575        drop(conn);
576        Ok(CdcOffset::MySql(MySqlOffset {
577            filename: row.take("File").unwrap(),
578            position: row.take("Position").unwrap(),
579        }))
580    }
581
582    fn snapshot_read(
583        &self,
584        table_name: SchemaTableName,
585        start_pk: Option<OwnedRow>,
586        primary_keys: Vec<String>,
587        limit: u32,
588    ) -> BoxStream<'_, ConnectorResult<OwnedRow>> {
589        self.snapshot_read_inner(table_name, start_pk, primary_keys, limit)
590    }
591
592    async fn disconnect(self) -> ConnectorResult<()> {
593        self.pool.disconnect().await.map_err(|e| e.into())
594    }
595
596    fn get_parallel_cdc_splits(
597        &self,
598        _options: CdcTableSnapshotSplitOption,
599    ) -> BoxStream<'_, ConnectorResult<CdcTableSnapshotSplit>> {
600        // TODO(zw): feat: impl
601        stream::empty::<ConnectorResult<CdcTableSnapshotSplit>>().boxed()
602    }
603
604    fn split_snapshot_read(
605        &self,
606        _table_name: SchemaTableName,
607        _left: OwnedRow,
608        _right: OwnedRow,
609        _split_columns: Vec<Field>,
610    ) -> BoxStream<'_, ConnectorResult<OwnedRow>> {
611        todo!("implement MySQL CDC parallelized backfill")
612    }
613}
614
615impl MySqlExternalTableReader {
616    /// Get MySQL version from the connection
617    async fn get_mysql_version(pool: &mysql_async::Pool) -> ConnectorResult<(u8, u8, bool)> {
618        let mut conn = pool.get_conn().await?;
619        let result: Option<String> = conn.query_first("SELECT VERSION()").await?;
620
621        if let Some(version_str) = result {
622            let parts: Vec<&str> = version_str.split('.').collect();
623            if parts.len() >= 2 {
624                let major_version = parts[0]
625                    .parse::<u8>()
626                    .context("Failed to parse major version")?;
627                let minor_version = parts[1]
628                    .parse::<u8>()
629                    .context("Failed to parse minor version")?;
630                let is_mariadb = version_str.to_lowercase().contains("mariadb");
631                return Ok((major_version, minor_version, is_mariadb));
632            }
633        }
634        Err(anyhow!("Failed to get MySQL version").into())
635    }
636
637    /// Check if MySQL version is 8.4 or later
638    fn is_mysql_8_4_or_later(&self) -> bool {
639        let (major, minor) = self.mysql_version;
640        major > 8 || (major == 8 && minor >= 4)
641    }
642
643    pub async fn new(
644        config: ExternalTableConfig,
645        rw_schema: Schema,
646        pk_indices: Vec<usize>,
647    ) -> ConnectorResult<Self> {
648        let database = config.database.clone();
649        let table = config.table.clone();
650        let pool = build_mysql_connection_pool(
651            &config.host,
652            config.port.parse::<u16>().unwrap(),
653            &config.username,
654            &config.password,
655            &config.database,
656            config.ssl_mode,
657        );
658
659        let field_names = rw_schema
660            .fields
661            .iter()
662            .filter(|f| f.name != CDC_OFFSET_COLUMN_NAME)
663            .map(|f| Self::quote_column(f.name.as_str()))
664            .join(",");
665
666        // Query MySQL primary key infos for type casting.
667        let upstream_mysql_pk_infos =
668            Self::query_upstream_pk_infos(&pool, &database, &table).await?;
669        Self::validate_upstream_pk_columns(&upstream_mysql_pk_infos, &rw_schema, &pk_indices)?;
670        // Get MySQL version
671        let (major_version, minor_version, is_mariadb) = Self::get_mysql_version(&pool).await?;
672        let mysql_version = (major_version, minor_version);
673        tracing::info!(
674            "MySQL version detected: {}.{} (is_mariadb={})",
675            mysql_version.0,
676            mysql_version.1,
677            is_mariadb
678        );
679
680        Ok(Self {
681            rw_schema,
682            pk_indices,
683            field_names,
684            pool,
685            upstream_mysql_pk_infos,
686            mysql_version,
687            is_mariadb,
688        })
689    }
690
691    pub fn get_normalized_table_name(table_name: &SchemaTableName) -> String {
692        // schema name is the database name in mysql
693        format!("`{}`.`{}`", table_name.schema_name, table_name.table_name)
694    }
695
696    pub fn get_cdc_offset_parser() -> CdcOffsetParseFunc {
697        Box::new(move |offset| {
698            Ok(CdcOffset::MySql(MySqlOffset::parse_debezium_offset(
699                offset,
700            )?))
701        })
702    }
703
704    /// Query upstream primary key data types, used for generating filter conditions with proper type casting.
705    async fn query_upstream_pk_infos(
706        pool: &mysql_async::Pool,
707        database: &str,
708        table: &str,
709    ) -> ConnectorResult<Vec<(String, ColumnType)>> {
710        let mut conn = pool.get_conn().await?;
711
712        // Query primary key columns and their data types.
713        let sql = "SELECT s.COLUMN_NAME, c.COLUMN_TYPE
714            FROM INFORMATION_SCHEMA.STATISTICS AS s
715            JOIN INFORMATION_SCHEMA.COLUMNS AS c
716              ON c.TABLE_SCHEMA = s.TABLE_SCHEMA
717             AND c.TABLE_NAME = s.TABLE_NAME
718             AND c.COLUMN_NAME = s.COLUMN_NAME
719            WHERE s.TABLE_SCHEMA = ?
720              AND s.TABLE_NAME = ?
721              AND s.INDEX_NAME = 'PRIMARY'
722            ORDER BY s.SEQ_IN_INDEX";
723
724        let rs: Vec<mysql_async::Row> = conn.exec(sql, (database, table)).await?;
725
726        let mut column_infos = Vec::new();
727        for row in &rs {
728            let column_name: String = row.get(0).unwrap();
729            let column_type: String = row.get(1).unwrap();
730            let column_type =
731                type_name_to_mysql_type(&column_type).unwrap_or(ColumnType::Unknown(column_type));
732            column_infos.push((column_name, column_type));
733        }
734
735        drop(conn);
736
737        Ok(column_infos)
738    }
739
740    fn validate_upstream_pk_columns(
741        upstream_pk_infos: &[(String, ColumnType)],
742        rw_schema: &Schema,
743        pk_indices: &[usize],
744    ) -> ConnectorResult<()> {
745        let expected_pk_names = pk_indices
746            .iter()
747            .map(|&index| rw_schema.fields[index].name.as_str())
748            .collect_vec();
749        let upstream_pk_names = upstream_pk_infos
750            .iter()
751            .map(|(name, _)| name.as_str())
752            .collect_vec();
753
754        if upstream_pk_names.is_empty() {
755            return Err(anyhow!(
756                "upstream MySQL primary key metadata is empty; expected columns {:?}",
757                expected_pk_names
758            )
759            .into());
760        }
761
762        if upstream_pk_names.len() != expected_pk_names.len()
763            || upstream_pk_names
764                .iter()
765                .zip_eq_fast(&expected_pk_names)
766                .any(|(upstream, expected)| !upstream.eq_ignore_ascii_case(expected))
767        {
768            return Err(anyhow!(
769                "upstream MySQL primary key columns {:?} do not match expected columns {:?}",
770                upstream_pk_names,
771                expected_pk_names
772            )
773            .into());
774        }
775
776        Ok(())
777    }
778
779    /// Check whether a column is `BIGINT UNSIGNED`.
780    ///
781    /// Frontend up-casts narrower unsigned integer types, and non-integer unsigned types
782    /// (`FLOAT`/`DOUBLE`/`DECIMAL UNSIGNED`) keep their own comparison semantics. Only
783    /// `BIGINT UNSIGNED` can be represented as a negative `i64` in RisingWave and needs
784    /// unsigned `u64` comparison/conversion.
785    fn needs_unsigned_i64_compare(&self, column_name: &str) -> ConnectorResult<bool> {
786        self.upstream_mysql_pk_infos
787            .iter()
788            .find(|(col_name, _)| col_name.eq_ignore_ascii_case(column_name))
789            .map(|(_, col_type)| mysql_type_is_unsigned_bigint(col_type))
790            .ok_or_else(|| {
791                anyhow!(
792                    "primary key column `{column_name}` not found in upstream MySQL primary key info"
793                )
794                .into()
795            })
796    }
797
798    pub(crate) fn pk_column_comparisons(
799        &self,
800        pk_names: &[String],
801    ) -> ConnectorResult<Vec<CdcKeyComparison>> {
802        pk_column_comparisons_from_infos(&self.upstream_mysql_pk_infos, pk_names)
803    }
804
805    /// Convert negative i64 to unsigned u64 based on column type
806    fn convert_negative_to_unsigned(&self, negative_val: i64) -> u64 {
807        negative_val as u64
808    }
809
810    #[try_stream(boxed, ok = OwnedRow, error = ConnectorError)]
811    async fn snapshot_read_inner(
812        &self,
813        table_name: SchemaTableName,
814        start_pk_row: Option<OwnedRow>,
815        primary_keys: Vec<String>,
816        limit: u32,
817    ) {
818        let order_key = primary_keys
819            .iter()
820            .map(|col| Self::quote_column(col))
821            .join(",");
822        let sql = if start_pk_row.is_none() {
823            format!(
824                "SELECT {} FROM {} ORDER BY {} LIMIT {limit}",
825                self.field_names,
826                Self::get_normalized_table_name(&table_name),
827                order_key,
828            )
829        } else {
830            let filter_expr = Self::filter_expression(&primary_keys);
831            format!(
832                "SELECT {} FROM {} WHERE {} ORDER BY {} LIMIT {limit}",
833                self.field_names,
834                Self::get_normalized_table_name(&table_name),
835                filter_expr,
836                order_key,
837            )
838        };
839        let mut conn = self.pool.get_conn().await?;
840        // Set session timezone to UTC
841        conn.exec_drop("SET time_zone = \"+00:00\"", ()).await?;
842
843        if let Some(start_pk_row) = start_pk_row {
844            let field_map = self
845                .rw_schema
846                .fields
847                .iter()
848                .map(|f| (f.name.as_str(), f.data_type.clone()))
849                .collect::<HashMap<_, _>>();
850
851            // fill in start primary key params
852            let params: Vec<_> = primary_keys
853                .iter()
854                .zip_eq_fast(start_pk_row.into_iter())
855                .map(|(pk, datum)| {
856                    if let Some(value) = datum {
857                        let ty = field_map.get(pk.as_str()).unwrap();
858                        let val = match ty {
859                            DataType::Boolean => Value::from(value.into_bool()),
860                            DataType::Int16 => Value::from(value.into_int16()),
861                            DataType::Int32 => Value::from(value.into_int32()),
862                            DataType::Int64 => {
863                                let int64_val = value.into_int64();
864                                if int64_val < 0 && self.needs_unsigned_i64_compare(pk.as_str())? {
865                                    Value::from(self.convert_negative_to_unsigned(int64_val))
866                                } else {
867                                    Value::from(int64_val)
868                                }
869                            }
870                            DataType::Float32 => Value::from(value.into_float32().into_inner()),
871                            DataType::Float64 => Value::from(value.into_float64().into_inner()),
872                            DataType::Varchar => Value::from(String::from(value.into_utf8())),
873                            DataType::Date => Value::from(value.into_date().0),
874                            DataType::Time => Value::from(value.into_time().0),
875                            DataType::Timestamp => Value::from(value.into_timestamp().0),
876                            DataType::Decimal => Value::from(value.into_decimal().to_string()),
877                            DataType::Timestamptz => {
878                                // Convert timestamptz to NaiveDateTime for MySQL TIMESTAMP comparison
879                                // MySQL expects NaiveDateTime for TIMESTAMP parameters
880                                let ts = value.into_timestamptz();
881                                let datetime_utc = ts.to_datetime_utc();
882                                let naive_datetime = datetime_utc.naive_utc();
883                                Value::from(naive_datetime)
884                            }
885                            _ => bail!("unsupported primary key data type: {}", ty),
886                        };
887                        ConnectorResult::Ok((pk.to_lowercase(), val))
888                    } else {
889                        bail!("primary key {} cannot be null", pk);
890                    }
891                })
892                .try_collect::<_, _, ConnectorError>()?;
893
894            tracing::debug!("snapshot read params: {:?}", &params);
895            let rs_stream = sql
896                .with(Params::from(params))
897                .stream::<mysql_async::Row, _>(&mut conn)
898                .await?;
899
900            let row_stream = rs_stream.map(|row| {
901                // convert mysql row into OwnedRow
902                let mut row = row?;
903                mysql_row_to_owned_row_with_strict_pk(&mut row, &self.rw_schema, &self.pk_indices)
904                    .map_err(ConnectorError::from)
905            });
906            pin_mut!(row_stream);
907            #[for_await]
908            for row in row_stream {
909                let row = row?;
910                yield row;
911            }
912        } else {
913            let rs_stream = sql.stream::<mysql_async::Row, _>(&mut conn).await?;
914            let row_stream = rs_stream.map(|row| {
915                // convert mysql row into OwnedRow
916                let mut row = row?;
917                mysql_row_to_owned_row_with_strict_pk(&mut row, &self.rw_schema, &self.pk_indices)
918                    .map_err(ConnectorError::from)
919            });
920            pin_mut!(row_stream);
921            #[for_await]
922            for row in row_stream {
923                let row = row?;
924                yield row;
925            }
926        }
927        drop(conn);
928    }
929
930    // mysql cannot leverage the given key to narrow down the range of scan,
931    // we need to rewrite the comparison conditions by our own.
932    // (a, b) > (x, y) => (`a` > x) OR ((`a` = x) AND (`b` > y))
933    fn filter_expression(columns: &[String]) -> String {
934        let mut conditions = vec![];
935        // push the first condition
936        conditions.push(format!(
937            "({} > :{})",
938            Self::quote_column(&columns[0]),
939            columns[0].to_lowercase()
940        ));
941        for i in 2..=columns.len() {
942            // '=' condition
943            let mut condition = String::new();
944            for (j, col) in columns.iter().enumerate().take(i - 1) {
945                if j == 0 {
946                    condition.push_str(&format!(
947                        "({} = :{})",
948                        Self::quote_column(col),
949                        col.to_lowercase()
950                    ));
951                } else {
952                    condition.push_str(&format!(
953                        " AND ({} = :{})",
954                        Self::quote_column(col),
955                        col.to_lowercase()
956                    ));
957                }
958            }
959            // '>' condition
960            condition.push_str(&format!(
961                " AND ({} > :{})",
962                Self::quote_column(&columns[i - 1]),
963                columns[i - 1].to_lowercase()
964            ));
965            conditions.push(format!("({})", condition));
966        }
967        if columns.len() > 1 {
968            conditions.join(" OR ")
969        } else {
970            conditions.join("")
971        }
972    }
973
974    fn quote_column(column: &str) -> String {
975        format!("`{}`", column)
976    }
977}
978
979#[cfg(test)]
980mod tests {
981    use std::collections::HashMap;
982
983    use futures::pin_mut;
984    use futures_async_stream::for_await;
985    use maplit::{convert_args, hashmap};
986    use risingwave_common::catalog::{CdcKeyComparison, ColumnDesc, ColumnId, Field, Schema};
987    use risingwave_common::types::DataType;
988    use sea_schema::mysql::def::{ColumnType, IndexInfo, IndexOrder, IndexPart, IndexType};
989
990    use super::{
991        mysql_type_is_unsigned_bigint, mysql_type_to_rw_type, pk_column_comparisons_from_infos,
992        primary_key_names, type_name_to_mysql_type,
993    };
994    use crate::source::cdc::external::mysql::MySqlExternalTable;
995    use crate::source::cdc::external::{
996        CdcOffset, ExternalTableConfig, ExternalTableReader, MySqlExternalTableReader, MySqlOffset,
997        SchemaTableName,
998    };
999
1000    fn parse_mysql_type_name(ty_name: &str) -> ColumnType {
1001        type_name_to_mysql_type(ty_name).unwrap()
1002    }
1003
1004    #[test]
1005    fn test_primary_key_names_preserve_index_order() {
1006        let indexes = vec![IndexInfo {
1007            unique: true,
1008            name: "PRIMARY".to_owned(),
1009            parts: ["RelatedID", "TypeID", "ClientID"]
1010                .into_iter()
1011                .map(|column| IndexPart {
1012                    column: column.to_owned(),
1013                    order: IndexOrder::Ascending,
1014                    sub_part: None,
1015                })
1016                .collect(),
1017            nullable: false,
1018            idx_type: IndexType::BTree,
1019            comment: String::new(),
1020            functional: false,
1021        }];
1022
1023        assert_eq!(
1024            primary_key_names(&indexes),
1025            Some(vec![
1026                "relatedid".to_owned(),
1027                "typeid".to_owned(),
1028                "clientid".to_owned(),
1029            ])
1030        );
1031    }
1032
1033    #[test]
1034    fn test_pk_column_comparisons_follow_requested_order() {
1035        let table = MySqlExternalTable {
1036            column_descs: vec![],
1037            pk_names: vec!["signed_id".to_owned(), "unsigned_id".to_owned()],
1038            pk_comparisons: vec![CdcKeyComparison::Native, CdcKeyComparison::UnsignedInt64],
1039        };
1040
1041        assert_eq!(
1042            table
1043                .pk_column_comparisons(&["UNSIGNED_ID".to_owned(), "SIGNED_ID".to_owned(),])
1044                .unwrap(),
1045            vec![CdcKeyComparison::UnsignedInt64, CdcKeyComparison::Native,]
1046        );
1047    }
1048
1049    #[test]
1050    fn test_reader_pk_column_comparisons_follow_requested_order() {
1051        let pk_infos = vec![
1052            ("Signed_ID".to_owned(), parse_mysql_type_name("BIGINT")),
1053            (
1054                "Unsigned_ID".to_owned(),
1055                parse_mysql_type_name("BIGINT UNSIGNED"),
1056            ),
1057            (
1058                "Narrow_Unsigned".to_owned(),
1059                parse_mysql_type_name("INTEGER UNSIGNED"),
1060            ),
1061        ];
1062
1063        assert_eq!(
1064            pk_column_comparisons_from_infos(
1065                &pk_infos,
1066                &[
1067                    "UNSIGNED_id".to_owned(),
1068                    "signed_id".to_owned(),
1069                    "narrow_unsigned".to_owned(),
1070                ],
1071            )
1072            .unwrap(),
1073            vec![
1074                CdcKeyComparison::UnsignedInt64,
1075                CdcKeyComparison::Native,
1076                CdcKeyComparison::Native,
1077            ]
1078        );
1079    }
1080
1081    #[test]
1082    fn test_validate_upstream_pk_columns() {
1083        let rw_schema = Schema::new(vec![
1084            Field::with_name(DataType::Int64, "id"),
1085            Field::with_name(DataType::Varchar, "tenant"),
1086        ]);
1087        let int_type = || ColumnType::BigInt(Default::default());
1088        let validate = |pk_infos: &[(String, ColumnType)]| {
1089            MySqlExternalTableReader::validate_upstream_pk_columns(pk_infos, &rw_schema, &[0, 1])
1090        };
1091
1092        let valid_pk_infos = vec![
1093            ("id".to_owned(), int_type()),
1094            ("tenant".to_owned(), int_type()),
1095        ];
1096        assert!(validate(&valid_pk_infos).is_ok());
1097
1098        for invalid_pk_infos in [
1099            // incorrect order
1100            vec![
1101                ("tenant".to_owned(), int_type()),
1102                ("id".to_owned(), int_type()),
1103            ],
1104            // incomplete
1105            vec![("id".to_owned(), int_type())],
1106            // extra index
1107            vec![
1108                ("id".to_owned(), int_type()),
1109                ("tenant".to_owned(), int_type()),
1110                ("extra".to_owned(), int_type()),
1111            ],
1112        ] {
1113            assert!(
1114                validate(&invalid_pk_infos).is_err(),
1115                "unexpectedly accepted {invalid_pk_infos:?}"
1116            );
1117        }
1118    }
1119
1120    #[test]
1121    fn test_mysql_unsigned_bigint_type_detection() {
1122        for ty_name in [
1123            "SERIAL",
1124            "BIGINT UNSIGNED",
1125            "BIGINT(20) UNSIGNED",
1126            "BIGINT UNSIGNED ZEROFILL",
1127            "INT8 UNSIGNED",
1128        ] {
1129            assert!(
1130                mysql_type_is_unsigned_bigint(&parse_mysql_type_name(ty_name)),
1131                "{ty_name}"
1132            );
1133        }
1134
1135        for ty_name in [
1136            "BIGINT",
1137            "INTEGER UNSIGNED",
1138            "INT4 UNSIGNED",
1139            "MEDIUMINT UNSIGNED",
1140            "DECIMAL UNSIGNED",
1141            "FLOAT8 UNSIGNED",
1142        ] {
1143            assert!(
1144                !mysql_type_is_unsigned_bigint(&parse_mysql_type_name(ty_name)),
1145                "{ty_name}"
1146            );
1147        }
1148    }
1149
1150    #[test]
1151    fn test_mysql_type_aliases() {
1152        assert!(matches!(
1153            parse_mysql_type_name("INTEGER UNSIGNED"),
1154            ColumnType::Int(attr) if attr.unsigned == Some(true)
1155        ));
1156        assert!(matches!(
1157            parse_mysql_type_name("INT1 UNSIGNED"),
1158            ColumnType::TinyInt(attr) if attr.unsigned == Some(true)
1159        ));
1160        assert!(matches!(
1161            parse_mysql_type_name("INT2 UNSIGNED"),
1162            ColumnType::SmallInt(attr) if attr.unsigned == Some(true)
1163        ));
1164        assert!(matches!(
1165            parse_mysql_type_name("INT3 UNSIGNED"),
1166            ColumnType::MediumInt(attr) if attr.unsigned == Some(true)
1167        ));
1168        assert!(matches!(
1169            parse_mysql_type_name("INT4 UNSIGNED"),
1170            ColumnType::Int(attr) if attr.unsigned == Some(true)
1171        ));
1172        assert!(matches!(
1173            parse_mysql_type_name("INT8 UNSIGNED"),
1174            ColumnType::BigInt(attr) if attr.unsigned == Some(true)
1175        ));
1176        assert!(matches!(
1177            parse_mysql_type_name("MIDDLEINT"),
1178            ColumnType::MediumInt(_)
1179        ));
1180        assert!(matches!(
1181            parse_mysql_type_name("NUMERIC"),
1182            ColumnType::Decimal(_)
1183        ));
1184        assert!(matches!(
1185            parse_mysql_type_name("CHARACTER VARYING(64)"),
1186            ColumnType::Varchar(_)
1187        ));
1188        assert!(matches!(
1189            parse_mysql_type_name("LONG VARBINARY"),
1190            ColumnType::MediumBlob
1191        ));
1192    }
1193
1194    #[test]
1195    fn test_mysql_serial_maps_as_unsigned_bigint() {
1196        let col_type = parse_mysql_type_name("SERIAL");
1197        assert!(mysql_type_is_unsigned_bigint(&col_type));
1198
1199        let serial_type = mysql_type_to_rw_type(&col_type).unwrap();
1200        let unsigned_bigint_type =
1201            mysql_type_to_rw_type(&parse_mysql_type_name("BIGINT UNSIGNED")).unwrap();
1202        assert_eq!(serial_type, DataType::Decimal);
1203        assert_eq!(serial_type, unsigned_bigint_type);
1204    }
1205
1206    #[ignore]
1207    #[tokio::test]
1208    async fn test_mysql_schema() {
1209        let config = ExternalTableConfig {
1210            connector: "mysql-cdc".to_owned(),
1211            host: "localhost".to_owned(),
1212            port: "8306".to_owned(),
1213            username: "root".to_owned(),
1214            password: "123456".to_owned(),
1215            database: "mydb".to_owned(),
1216            schema: "".to_owned(),
1217            table: "part".to_owned(),
1218            ssl_mode: Default::default(),
1219            ssl_root_cert: None,
1220            encrypt: "false".to_owned(),
1221        };
1222
1223        let table = MySqlExternalTable::connect(config).await.unwrap();
1224        println!("columns: {:?}", table.column_descs);
1225        println!("primary keys: {:?}", table.pk_names);
1226    }
1227
1228    #[test]
1229    fn test_mysql_filter_expr() {
1230        let cols = vec!["id".to_owned()];
1231        let expr = MySqlExternalTableReader::filter_expression(&cols);
1232        assert_eq!(expr, "(`id` > :id)");
1233
1234        let cols = vec!["aa".to_owned(), "bb".to_owned(), "cc".to_owned()];
1235        let expr = MySqlExternalTableReader::filter_expression(&cols);
1236        assert_eq!(
1237            expr,
1238            "(`aa` > :aa) OR ((`aa` = :aa) AND (`bb` > :bb)) OR ((`aa` = :aa) AND (`bb` = :bb) AND (`cc` > :cc))"
1239        );
1240    }
1241
1242    #[test]
1243    fn test_mysql_binlog_offset() {
1244        let off0_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000001", "pos": 105622, "snapshot": true }, "isHeartbeat": false }"#;
1245        let off1_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000007", "pos": 1062363217, "snapshot": true }, "isHeartbeat": false }"#;
1246        let off2_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000007", "pos": 659687560, "snapshot": true }, "isHeartbeat": false }"#;
1247        let off3_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000008", "pos": 7665875, "snapshot": true }, "isHeartbeat": false }"#;
1248        let off4_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000008", "pos": 7665875, "snapshot": true }, "isHeartbeat": false }"#;
1249
1250        let off0 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off0_str).unwrap());
1251        let off1 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off1_str).unwrap());
1252        let off2 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off2_str).unwrap());
1253        let off3 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off3_str).unwrap());
1254        let off4 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off4_str).unwrap());
1255
1256        assert!(off0 <= off1);
1257        assert!(off1 > off2);
1258        assert!(off2 < off3);
1259        assert_eq!(off3, off4);
1260    }
1261
1262    // manual test case
1263    #[ignore]
1264    #[tokio::test]
1265    async fn test_mysql_table_reader() {
1266        let columns = [
1267            ColumnDesc::named("v1", ColumnId::new(1), DataType::Int32),
1268            ColumnDesc::named("v2", ColumnId::new(2), DataType::Decimal),
1269            ColumnDesc::named("v3", ColumnId::new(3), DataType::Varchar),
1270            ColumnDesc::named("v4", ColumnId::new(4), DataType::Date),
1271        ];
1272        let rw_schema = Schema {
1273            fields: columns.iter().map(Field::from).collect(),
1274        };
1275        let props: HashMap<String, String> = convert_args!(hashmap!(
1276                "hostname" => "localhost",
1277                "port" => "8306",
1278                "username" => "root",
1279                "password" => "123456",
1280                "database.name" => "mytest",
1281                "table.name" => "t1"));
1282
1283        let config =
1284            serde_json::from_value::<ExternalTableConfig>(serde_json::to_value(props).unwrap())
1285                .unwrap();
1286        let reader = MySqlExternalTableReader::new(config, rw_schema, vec![0])
1287            .await
1288            .unwrap();
1289        let offset = reader.current_cdc_offset().await.unwrap();
1290        println!("BinlogOffset: {:?}", offset);
1291
1292        let off0_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000001", "pos": 105622, "snapshot": true }, "isHeartbeat": false }"#;
1293        let parser = MySqlExternalTableReader::get_cdc_offset_parser();
1294        println!("parsed offset: {:?}", parser(off0_str).unwrap());
1295        let table_name = SchemaTableName {
1296            schema_name: "mytest".to_owned(),
1297            table_name: "t1".to_owned(),
1298        };
1299
1300        let stream = reader.snapshot_read(table_name, None, vec!["v1".to_owned()], 1000);
1301        pin_mut!(stream);
1302        #[for_await]
1303        for row in stream {
1304            println!("OwnedRow: {:?}", row);
1305        }
1306    }
1307}
risingwave_connector/source/cdc/external/mysql.rs

risingwave_connector/source/cdc/external/
mysql.rs
