1use 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};
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::error::{ConnectorError, ConnectorResult};
41use crate::source::CdcTableSnapshotSplit;
42use crate::source::cdc::external::{
43 CdcOffset, CdcOffsetParseFunc, CdcTableSnapshotSplitOption, DebeziumOffset,
44 ExternalTableConfig, ExternalTableReader, SchemaTableName, SslMode, mysql_row_to_owned_row,
45};
46
47#[derive(Debug, Clone, Default, PartialEq, PartialOrd, Serialize, Deserialize)]
48pub struct MySqlOffset {
49 pub filename: String,
50 pub position: u64,
51}
52
53impl MySqlOffset {
54 pub fn new(filename: String, position: u64) -> Self {
55 Self { filename, position }
56 }
57}
58
59impl MySqlOffset {
60 pub fn parse_debezium_offset(offset: &str) -> ConnectorResult<Self> {
61 let dbz_offset: DebeziumOffset = serde_json::from_str(offset)
62 .with_context(|| format!("invalid upstream offset: {}", offset))?;
63
64 Ok(Self {
65 filename: dbz_offset
66 .source_offset
67 .file
68 .context("binlog file not found in offset")?,
69 position: dbz_offset
70 .source_offset
71 .pos
72 .context("binlog position not found in offset")?,
73 })
74 }
75}
76
77pub struct MySqlExternalTable {
78 column_descs: Vec<ColumnDesc>,
79 pk_names: Vec<String>,
80}
81
82impl MySqlExternalTable {
83 pub async fn connect(config: ExternalTableConfig) -> ConnectorResult<Self> {
84 tracing::debug!("connect to mysql");
85 let options = MySqlConnectOptions::new()
86 .username(&config.username)
87 .password(&config.password)
88 .host(&config.host)
89 .port(config.port.parse::<u16>().unwrap())
90 .database(&config.database)
91 .ssl_mode(match config.ssl_mode {
92 SslMode::Disabled => sqlx::mysql::MySqlSslMode::Disabled,
93 SslMode::Preferred => sqlx::mysql::MySqlSslMode::Preferred,
94 SslMode::Required => sqlx::mysql::MySqlSslMode::Required,
95 _ => {
96 return Err(anyhow!("unsupported SSL mode").into());
97 }
98 });
99
100 let connection = MySqlPool::connect_with(options).await?;
101 let mut schema_discovery = SchemaDiscovery::new(connection, config.database.as_str());
102
103 let system_info = schema_discovery.discover_system().await?;
105 schema_discovery.query = SchemaQueryBuilder::new(system_info.clone());
106 let schema = Alias::new(config.database.as_str()).into_iden();
107 let table = Alias::new(config.table.as_str()).into_iden();
108 let columns = schema_discovery
109 .discover_columns(schema, table, &system_info)
110 .await?;
111 let mut column_descs = vec![];
112 let mut pk_names = vec![];
113 for col in columns {
114 let data_type = mysql_type_to_rw_type(&col.col_type)?;
115 let col_name = col.name.to_lowercase();
117 let column_desc = if let Some(default) = col.default {
118 let snapshot_value = derive_default_value(default.clone(), &data_type)
119 .unwrap_or_else(|e| {
120 tracing::warn!(
121 column = col_name,
122 ?default,
123 %data_type,
124 error = %e.as_report(),
125 "failed to derive column default value, fallback to `NULL`",
126 );
127 None
128 });
129
130 ColumnDesc::named_with_default_value(
131 col_name.clone(),
132 ColumnId::placeholder(),
133 data_type.clone(),
134 snapshot_value,
135 )
136 } else {
137 ColumnDesc::named(col_name.clone(), ColumnId::placeholder(), data_type)
138 };
139
140 column_descs.push(column_desc);
141 if matches!(col.key, ColumnKey::Primary) {
142 pk_names.push(col_name);
143 }
144 }
145
146 if pk_names.is_empty() {
147 return Err(anyhow!("MySQL table doesn't define the primary key").into());
148 }
149 Ok(Self {
150 column_descs,
151 pk_names,
152 })
153 }
154
155 pub fn column_descs(&self) -> &Vec<ColumnDesc> {
156 &self.column_descs
157 }
158
159 pub fn pk_names(&self) -> &Vec<String> {
160 &self.pk_names
161 }
162}
163
164fn derive_default_value(default: ColumnDefault, data_type: &DataType) -> ConnectorResult<Datum> {
165 let datum = match default {
166 ColumnDefault::Null => None,
167 ColumnDefault::Int(val) => match data_type {
168 DataType::Int16 => Some(ScalarImpl::Int16(val as _)),
169 DataType::Int32 => Some(ScalarImpl::Int32(val as _)),
170 DataType::Int64 => Some(ScalarImpl::Int64(val)),
171 DataType::Varchar => {
172 Some(ScalarImpl::from(val.to_string()))
174 }
175 _ => bail!("unexpected default value type for integer"),
176 },
177 ColumnDefault::Real(val) => match data_type {
178 DataType::Float32 => Some(ScalarImpl::Float32(F32::from(val as f32))),
179 DataType::Float64 => Some(ScalarImpl::Float64(val.into())),
180 DataType::Decimal => Some(ScalarImpl::Decimal(
181 Decimal::try_from(val).context("failed to convert default value to decimal")?,
182 )),
183 _ => bail!("unexpected default value type for real"),
184 },
185 ColumnDefault::String(mut val) => {
186 if data_type == &DataType::Timestamptz {
189 val = timestamp_val_to_timestamptz(val.as_str())?;
190 }
191 Some(ScalarImpl::from_text(val.as_str(), data_type).map_err(|e| anyhow!(e)).context(
192 "failed to parse mysql default value expression, only constant is supported",
193 )?)
194 }
195 ColumnDefault::CurrentTimestamp | ColumnDefault::CustomExpr(_) => {
196 bail!("MySQL CURRENT_TIMESTAMP and custom expression default value not supported")
197 }
198 };
199 Ok(datum)
200}
201
202pub fn timestamp_val_to_timestamptz(value_text: &str) -> ConnectorResult<String> {
203 let format = "%Y-%m-%d %H:%M:%S";
204 let naive_datetime = NaiveDateTime::parse_from_str(value_text, format)
205 .map_err(|err| anyhow!("failed to parse mysql timestamp value").context(err))?;
206 let postgres_timestamptz: DateTime<chrono::Utc> =
207 DateTime::<chrono::Utc>::from_naive_utc_and_offset(naive_datetime, chrono::Utc);
208 Ok(postgres_timestamptz
209 .format("%Y-%m-%d %H:%M:%S%:z")
210 .to_string())
211}
212
213pub fn type_name_to_mysql_type(ty_name: &str) -> Option<ColumnType> {
214 macro_rules! column_type {
215 ($($name:literal => $variant:ident),* $(,)?) => {
216 match ty_name.to_lowercase().as_str() {
217 $(
218 $name => Some(ColumnType::$variant(Default::default())),
219 )*
220 "json" => Some(ColumnType::Json),
221 "date" => Some(ColumnType::Date),
222 "bool" => Some(ColumnType::Bool),
223 "tinyblob" => Some(ColumnType::TinyBlob),
224 "mediumblob" => Some(ColumnType::MediumBlob),
225 "longblob" => Some(ColumnType::LongBlob),
226 _ => None,
227 }
228 };
229 }
230
231 column_type! {
232 "bit" => Bit,
233 "tinyint" => TinyInt,
234 "smallint" => SmallInt,
235 "mediumint" => MediumInt,
236 "int" => Int,
237 "bigint" => BigInt,
238 "decimal" => Decimal,
239 "float" => Float,
240 "double" => Double,
241 "time" => Time,
242 "datetime" => DateTime,
243 "timestamp" => Timestamp,
244 "char" => Char,
245 "nchar" => NChar,
246 "varchar" => Varchar,
247 "nvarchar" => NVarchar,
248 "binary" => Binary,
249 "varbinary" => Varbinary,
250 "text" => Text,
251 "tinytext" => TinyText,
252 "mediumtext" => MediumText,
253 "longtext" => LongText,
254 "blob" => Blob,
255 "enum" => Enum,
256 "set" => Set,
257 "geometry" => Geometry,
258 "point" => Point,
259 "linestring" => LineString,
260 "polygon" => Polygon,
261 "multipoint" => MultiPoint,
262 "multilinestring" => MultiLineString,
263 "multipolygon" => MultiPolygon,
264 "geometrycollection" => GeometryCollection,
265 }
266}
267
268pub fn mysql_type_to_rw_type(col_type: &ColumnType) -> ConnectorResult<DataType> {
269 let dtype = match col_type {
270 ColumnType::Serial => DataType::Int32,
271 ColumnType::Bit(attr) => {
272 if let Some(1) = attr.maximum {
273 DataType::Boolean
274 } else {
275 return Err(
276 anyhow!("BIT({}) type not supported", attr.maximum.unwrap_or(0)).into(),
277 );
278 }
279 }
280 ColumnType::TinyInt(_) | ColumnType::SmallInt(_) => DataType::Int16,
281 ColumnType::Bool => DataType::Boolean,
282 ColumnType::MediumInt(_) => DataType::Int32,
283 ColumnType::Int(_) => DataType::Int32,
284 ColumnType::BigInt(_) => DataType::Int64,
285 ColumnType::Decimal(_) => DataType::Decimal,
286 ColumnType::Float(_) => DataType::Float32,
287 ColumnType::Double(_) => DataType::Float64,
288 ColumnType::Date => DataType::Date,
289 ColumnType::Time(_) => DataType::Time,
290 ColumnType::DateTime(_) => DataType::Timestamp,
291 ColumnType::Timestamp(_) => DataType::Timestamptz,
292 ColumnType::Year => DataType::Int32,
293 ColumnType::Char(_)
294 | ColumnType::NChar(_)
295 | ColumnType::Varchar(_)
296 | ColumnType::NVarchar(_) => DataType::Varchar,
297 ColumnType::Binary(_) | ColumnType::Varbinary(_) => DataType::Bytea,
298 ColumnType::Text(_)
299 | ColumnType::TinyText(_)
300 | ColumnType::MediumText(_)
301 | ColumnType::LongText(_) => DataType::Varchar,
302 ColumnType::Blob(_)
303 | ColumnType::TinyBlob
304 | ColumnType::MediumBlob
305 | ColumnType::LongBlob => DataType::Bytea,
306 ColumnType::Enum(_) => DataType::Varchar,
307 ColumnType::Json => DataType::Jsonb,
308 ColumnType::Set(_) => {
309 return Err(anyhow!("SET type not supported").into());
310 }
311 ColumnType::Geometry(_) => {
312 return Err(anyhow!("GEOMETRY type not supported").into());
313 }
314 ColumnType::Point(_) => {
315 return Err(anyhow!("POINT type not supported").into());
316 }
317 ColumnType::LineString(_) => {
318 return Err(anyhow!("LINE string type not supported").into());
319 }
320 ColumnType::Polygon(_) => {
321 return Err(anyhow!("POLYGON type not supported").into());
322 }
323 ColumnType::MultiPoint(_) => {
324 return Err(anyhow!("MULTI POINT type not supported").into());
325 }
326 ColumnType::MultiLineString(_) => {
327 return Err(anyhow!("MULTI LINE STRING type not supported").into());
328 }
329 ColumnType::MultiPolygon(_) => {
330 return Err(anyhow!("MULTI POLYGON type not supported").into());
331 }
332 ColumnType::GeometryCollection(_) => {
333 return Err(anyhow!("GEOMETRY COLLECTION type not supported").into());
334 }
335 ColumnType::Unknown(_) => {
336 return Err(anyhow!("Unknown MySQL data type").into());
337 }
338 };
339
340 Ok(dtype)
341}
342
343pub struct MySqlExternalTableReader {
344 rw_schema: Schema,
345 field_names: String,
346 pool: mysql_async::Pool,
347}
348
349impl ExternalTableReader for MySqlExternalTableReader {
350 async fn current_cdc_offset(&self) -> ConnectorResult<CdcOffset> {
351 let mut conn = self.pool.get_conn().await?;
352
353 let sql = "SHOW MASTER STATUS".to_owned();
354 let mut rs = conn.query::<mysql_async::Row, _>(sql).await?;
355 let row = rs
356 .iter_mut()
357 .exactly_one()
358 .ok()
359 .context("expect exactly one row when reading binlog offset")?;
360 drop(conn);
361 Ok(CdcOffset::MySql(MySqlOffset {
362 filename: row.take("File").unwrap(),
363 position: row.take("Position").unwrap(),
364 }))
365 }
366
367 fn snapshot_read(
368 &self,
369 table_name: SchemaTableName,
370 start_pk: Option<OwnedRow>,
371 primary_keys: Vec<String>,
372 limit: u32,
373 ) -> BoxStream<'_, ConnectorResult<OwnedRow>> {
374 self.snapshot_read_inner(table_name, start_pk, primary_keys, limit)
375 }
376
377 async fn disconnect(self) -> ConnectorResult<()> {
378 self.pool.disconnect().await.map_err(|e| e.into())
379 }
380
381 fn get_parallel_cdc_splits(
382 &self,
383 _options: CdcTableSnapshotSplitOption,
384 ) -> BoxStream<'_, ConnectorResult<CdcTableSnapshotSplit>> {
385 stream::empty::<ConnectorResult<CdcTableSnapshotSplit>>().boxed()
387 }
388
389 fn split_snapshot_read(
390 &self,
391 _table_name: SchemaTableName,
392 _left: OwnedRow,
393 _right: OwnedRow,
394 _split_columns: Vec<Field>,
395 ) -> BoxStream<'_, ConnectorResult<OwnedRow>> {
396 todo!("implement MySQL CDC parallelized backfill")
397 }
398}
399
400impl MySqlExternalTableReader {
401 pub fn new(config: ExternalTableConfig, rw_schema: Schema) -> ConnectorResult<Self> {
402 let mut opts_builder = mysql_async::OptsBuilder::default()
403 .user(Some(config.username))
404 .pass(Some(config.password))
405 .ip_or_hostname(config.host)
406 .tcp_port(config.port.parse::<u16>().unwrap())
407 .db_name(Some(config.database));
408
409 opts_builder = match config.ssl_mode {
410 SslMode::Disabled | SslMode::Preferred => opts_builder.ssl_opts(None),
411 SslMode::Required | SslMode::VerifyCa | SslMode::VerifyFull => {
413 let ssl_without_verify = mysql_async::SslOpts::default()
414 .with_danger_accept_invalid_certs(true)
415 .with_danger_skip_domain_validation(true);
416 opts_builder.ssl_opts(Some(ssl_without_verify))
417 }
418 };
419 let pool = mysql_async::Pool::new(opts_builder);
420
421 let field_names = rw_schema
422 .fields
423 .iter()
424 .filter(|f| f.name != CDC_OFFSET_COLUMN_NAME)
425 .map(|f| Self::quote_column(f.name.as_str()))
426 .join(",");
427
428 Ok(Self {
429 rw_schema,
430 field_names,
431 pool,
432 })
433 }
434
435 pub fn get_normalized_table_name(table_name: &SchemaTableName) -> String {
436 format!("`{}`.`{}`", table_name.schema_name, table_name.table_name)
438 }
439
440 pub fn get_cdc_offset_parser() -> CdcOffsetParseFunc {
441 Box::new(move |offset| {
442 Ok(CdcOffset::MySql(MySqlOffset::parse_debezium_offset(
443 offset,
444 )?))
445 })
446 }
447
448 #[try_stream(boxed, ok = OwnedRow, error = ConnectorError)]
449 async fn snapshot_read_inner(
450 &self,
451 table_name: SchemaTableName,
452 start_pk_row: Option<OwnedRow>,
453 primary_keys: Vec<String>,
454 limit: u32,
455 ) {
456 let order_key = primary_keys
457 .iter()
458 .map(|col| Self::quote_column(col))
459 .join(",");
460 let sql = if start_pk_row.is_none() {
461 format!(
462 "SELECT {} FROM {} ORDER BY {} LIMIT {limit}",
463 self.field_names,
464 Self::get_normalized_table_name(&table_name),
465 order_key,
466 )
467 } else {
468 let filter_expr = Self::filter_expression(&primary_keys);
469 format!(
470 "SELECT {} FROM {} WHERE {} ORDER BY {} LIMIT {limit}",
471 self.field_names,
472 Self::get_normalized_table_name(&table_name),
473 filter_expr,
474 order_key,
475 )
476 };
477
478 let mut conn = self.pool.get_conn().await?;
479 conn.exec_drop("SET time_zone = \"+00:00\"", ()).await?;
481
482 if start_pk_row.is_none() {
483 let rs_stream = sql.stream::<mysql_async::Row, _>(&mut conn).await?;
484 let row_stream = rs_stream.map(|row| {
485 let mut row = row?;
487 Ok::<_, ConnectorError>(mysql_row_to_owned_row(&mut row, &self.rw_schema))
488 });
489 pin_mut!(row_stream);
490 #[for_await]
491 for row in row_stream {
492 let row = row?;
493 yield row;
494 }
495 } else {
496 let field_map = self
497 .rw_schema
498 .fields
499 .iter()
500 .map(|f| (f.name.as_str(), f.data_type.clone()))
501 .collect::<HashMap<_, _>>();
502
503 let params: Vec<_> = primary_keys
505 .iter()
506 .zip_eq_fast(start_pk_row.unwrap().into_iter())
507 .map(|(pk, datum)| {
508 if let Some(value) = datum {
509 let ty = field_map.get(pk.as_str()).unwrap();
510 let val = match ty {
511 DataType::Boolean => Value::from(value.into_bool()),
512 DataType::Int16 => Value::from(value.into_int16()),
513 DataType::Int32 => Value::from(value.into_int32()),
514 DataType::Int64 => Value::from(value.into_int64()),
515 DataType::Float32 => Value::from(value.into_float32().into_inner()),
516 DataType::Float64 => Value::from(value.into_float64().into_inner()),
517 DataType::Varchar => Value::from(String::from(value.into_utf8())),
518 DataType::Date => Value::from(value.into_date().0),
519 DataType::Time => Value::from(value.into_time().0),
520 DataType::Timestamp => Value::from(value.into_timestamp().0),
521 _ => bail!("unsupported primary key data type: {}", ty),
522 };
523 ConnectorResult::Ok((pk.to_lowercase(), val))
524 } else {
525 bail!("primary key {} cannot be null", pk);
526 }
527 })
528 .try_collect::<_, _, ConnectorError>()?;
529
530 tracing::debug!("snapshot read params: {:?}", ¶ms);
531 let rs_stream = sql
532 .with(Params::from(params))
533 .stream::<mysql_async::Row, _>(&mut conn)
534 .await?;
535
536 let row_stream = rs_stream.map(|row| {
537 let mut row = row?;
539 Ok::<_, ConnectorError>(mysql_row_to_owned_row(&mut row, &self.rw_schema))
540 });
541 pin_mut!(row_stream);
542 #[for_await]
543 for row in row_stream {
544 let row = row?;
545 yield row;
546 }
547 };
548 drop(conn);
549 }
550
551 fn filter_expression(columns: &[String]) -> String {
555 let mut conditions = vec![];
556 conditions.push(format!(
558 "({} > :{})",
559 Self::quote_column(&columns[0]),
560 columns[0].to_lowercase()
561 ));
562 for i in 2..=columns.len() {
563 let mut condition = String::new();
565 for (j, col) in columns.iter().enumerate().take(i - 1) {
566 if j == 0 {
567 condition.push_str(&format!(
568 "({} = :{})",
569 Self::quote_column(col),
570 col.to_lowercase()
571 ));
572 } else {
573 condition.push_str(&format!(
574 " AND ({} = :{})",
575 Self::quote_column(col),
576 col.to_lowercase()
577 ));
578 }
579 }
580 condition.push_str(&format!(
582 " AND ({} > :{})",
583 Self::quote_column(&columns[i - 1]),
584 columns[i - 1].to_lowercase()
585 ));
586 conditions.push(format!("({})", condition));
587 }
588 if columns.len() > 1 {
589 conditions.join(" OR ")
590 } else {
591 conditions.join("")
592 }
593 }
594
595 fn quote_column(column: &str) -> String {
596 format!("`{}`", column)
597 }
598}
599
600#[cfg(test)]
601mod tests {
602 use std::collections::HashMap;
603
604 use futures::pin_mut;
605 use futures_async_stream::for_await;
606 use maplit::{convert_args, hashmap};
607 use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema};
608 use risingwave_common::types::DataType;
609
610 use crate::source::cdc::external::mysql::MySqlExternalTable;
611 use crate::source::cdc::external::{
612 CdcOffset, ExternalTableConfig, ExternalTableReader, MySqlExternalTableReader, MySqlOffset,
613 SchemaTableName,
614 };
615
616 #[ignore]
617 #[tokio::test]
618 async fn test_mysql_schema() {
619 let config = ExternalTableConfig {
620 connector: "mysql-cdc".to_owned(),
621 host: "localhost".to_owned(),
622 port: "8306".to_owned(),
623 username: "root".to_owned(),
624 password: "123456".to_owned(),
625 database: "mydb".to_owned(),
626 schema: "".to_owned(),
627 table: "part".to_owned(),
628 ssl_mode: Default::default(),
629 ssl_root_cert: None,
630 encrypt: "false".to_owned(),
631 };
632
633 let table = MySqlExternalTable::connect(config).await.unwrap();
634 println!("columns: {:?}", &table.column_descs);
635 println!("primary keys: {:?}", &table.pk_names);
636 }
637
638 #[test]
639 fn test_mysql_filter_expr() {
640 let cols = vec!["id".to_owned()];
641 let expr = MySqlExternalTableReader::filter_expression(&cols);
642 assert_eq!(expr, "(`id` > :id)");
643
644 let cols = vec!["aa".to_owned(), "bb".to_owned(), "cc".to_owned()];
645 let expr = MySqlExternalTableReader::filter_expression(&cols);
646 assert_eq!(
647 expr,
648 "(`aa` > :aa) OR ((`aa` = :aa) AND (`bb` > :bb)) OR ((`aa` = :aa) AND (`bb` = :bb) AND (`cc` > :cc))"
649 );
650 }
651
652 #[test]
653 fn test_mysql_binlog_offset() {
654 let off0_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000001", "pos": 105622, "snapshot": true }, "isHeartbeat": false }"#;
655 let off1_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000007", "pos": 1062363217, "snapshot": true }, "isHeartbeat": false }"#;
656 let off2_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000007", "pos": 659687560, "snapshot": true }, "isHeartbeat": false }"#;
657 let off3_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000008", "pos": 7665875, "snapshot": true }, "isHeartbeat": false }"#;
658 let off4_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000008", "pos": 7665875, "snapshot": true }, "isHeartbeat": false }"#;
659
660 let off0 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off0_str).unwrap());
661 let off1 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off1_str).unwrap());
662 let off2 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off2_str).unwrap());
663 let off3 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off3_str).unwrap());
664 let off4 = CdcOffset::MySql(MySqlOffset::parse_debezium_offset(off4_str).unwrap());
665
666 assert!(off0 <= off1);
667 assert!(off1 > off2);
668 assert!(off2 < off3);
669 assert_eq!(off3, off4);
670 }
671
672 #[ignore]
674 #[tokio::test]
675 async fn test_mysql_table_reader() {
676 let columns = vec![
677 ColumnDesc::named("v1", ColumnId::new(1), DataType::Int32),
678 ColumnDesc::named("v2", ColumnId::new(2), DataType::Decimal),
679 ColumnDesc::named("v3", ColumnId::new(3), DataType::Varchar),
680 ColumnDesc::named("v4", ColumnId::new(4), DataType::Date),
681 ];
682 let rw_schema = Schema {
683 fields: columns.iter().map(Field::from).collect(),
684 };
685 let props: HashMap<String, String> = convert_args!(hashmap!(
686 "hostname" => "localhost",
687 "port" => "8306",
688 "username" => "root",
689 "password" => "123456",
690 "database.name" => "mytest",
691 "table.name" => "t1"));
692
693 let config =
694 serde_json::from_value::<ExternalTableConfig>(serde_json::to_value(props).unwrap())
695 .unwrap();
696 let reader = MySqlExternalTableReader::new(config, rw_schema).unwrap();
697 let offset = reader.current_cdc_offset().await.unwrap();
698 println!("BinlogOffset: {:?}", offset);
699
700 let off0_str = r#"{ "sourcePartition": { "server": "test" }, "sourceOffset": { "ts_sec": 1670876905, "file": "binlog.000001", "pos": 105622, "snapshot": true }, "isHeartbeat": false }"#;
701 let parser = MySqlExternalTableReader::get_cdc_offset_parser();
702 println!("parsed offset: {:?}", parser(off0_str).unwrap());
703 let table_name = SchemaTableName {
704 schema_name: "mytest".to_owned(),
705 table_name: "t1".to_owned(),
706 };
707
708 let stream = reader.snapshot_read(table_name, None, vec!["v1".to_owned()], 1000);
709 pin_mut!(stream);
710 #[for_await]
711 for row in stream {
712 println!("OwnedRow: {:?}", row);
713 }
714 }
715}