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

risingwave_connector/source/cdc/external/
mysql.rs
