1, sink mysql json columns. 2, sink redis rdb data by rewrite besides…

… restore. 3, sink basic ddls for mysql
apecloud · Aug 15, 2023 · fdb1c30 · fdb1c30
1 parent 71477b4
commit fdb1c30
Show file tree

Hide file tree

Showing 101 changed files with 2,005 additions and 563 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -52,4 +52,5 @@ uuid = { version = "1.3.1", features = ["v4"] }
 nom = "7.1.3"
 mongodb = { version = "2.5.0" }
 dotenv = "0.15.0"
-redis = {git = "https://github.com/qianyiwen2019/redis-rs", features = ["tokio-comp"]}
+redis =  { version = "0.23.1", features = ["tokio-comp"] } 
+thiserror = "1.0.44"
diff --git a/dt-common/Cargo.toml b/dt-common/Cargo.toml
@@ -22,4 +22,5 @@ futures = { workspace = true }
 serde_yaml = { workspace = true }
 regex = { workspace = true }
 nom = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true }
+thiserror = { workspace = true }
diff --git a/dt-common/src/config/config_enums.rs b/dt-common/src/config/config_enums.rs
@@ -4,7 +4,7 @@ use strum::{Display, EnumString, IntoStaticStr};
 
 use crate::error::Error;
 
-#[derive(Clone, Display, EnumString, IntoStaticStr, Debug)]
+#[derive(Clone, Display, EnumString, IntoStaticStr, Debug, PartialEq, Eq)]
 pub enum DbType {
     #[strum(serialize = "mysql")]
     Mysql,
@@ -36,7 +36,7 @@ pub enum ExtractType {
     Basic,
 }
 
-#[derive(EnumString, IntoStaticStr)]
+#[derive(Display, EnumString, IntoStaticStr)]
 pub enum SinkType {
     #[strum(serialize = "write")]
     Write,
@@ -73,7 +73,7 @@ pub enum RouteType {
     Tb,
 }
 
-#[derive(Clone, IntoStaticStr)]
+#[derive(Clone, Debug, IntoStaticStr)]
 pub enum ConflictPolicyEnum {
     #[strum(serialize = "ignore")]
     Ignore,

diff --git a/dt-common/src/config/extractor_config.rs b/dt-common/src/config/extractor_config.rs
@@ -1,6 +1,6 @@
 use super::config_enums::DbType;
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub enum ExtractorConfig {
     Basic {
         url: String,

diff --git a/dt-common/src/config/sinker_config.rs b/dt-common/src/config/sinker_config.rs
@@ -1,6 +1,6 @@
 use super::config_enums::{ConflictPolicyEnum, DbType};
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub enum SinkerConfig {
     Basic {
         url: String,
@@ -69,6 +69,7 @@ pub enum SinkerConfig {
     Redis {
         url: String,
         batch_size: usize,
+        method: String,
     },
 }
 

diff --git a/dt-common/src/config/task_config.rs b/dt-common/src/config/task_config.rs
@@ -157,9 +157,10 @@ impl TaskConfig {
 
                 ExtractType::Basic => Ok(ExtractorConfig::Basic { url, db_type }),
 
-                t => Err(Error::Unexpected {
-                    error: format!("extract type: {} not supported", t),
-                }),
+                extract_type => Err(Error::ConfigError(format!(
+                    "extract type: {} not supported",
+                    extract_type
+                ))),
             },
 
             DbType::Redis => {
@@ -179,15 +180,17 @@ impl TaskConfig {
                         now_db_id: ini.getint(EXTRACTOR, "now_db_id").unwrap().unwrap(),
                     }),
 
-                    t => Err(Error::Unexpected {
-                        error: format!("extract type: {} not supported", t),
-                    }),
+                    extract_type => Err(Error::ConfigError(format!(
+                        "extract type: {} not supported",
+                        extract_type
+                    ))),
                 }
             }
 
-            _ => Err(Error::Unexpected {
-                error: "extractor db type not supported".to_string(),
-            }),
+            db_type => Err(Error::ConfigError(format!(
+                "extractor db type: {} not supported",
+                db_type
+            ))),
         }
     }
 
@@ -240,9 +243,10 @@ impl TaskConfig {
 
                 SinkType::Basic => Ok(SinkerConfig::Basic { url, db_type }),
 
-                _ => Err(Error::Unexpected {
-                    error: "sinker db type not supported".to_string(),
-                }),
+                db_type => Err(Error::ConfigError(format!(
+                    "sinker db type: {} not supported",
+                    db_type
+                ))),
             },
 
             DbType::Kafka => Ok(SinkerConfig::Kafka {
@@ -267,7 +271,11 @@ impl TaskConfig {
                 root_dir: ini.get(SINKER, "root_dir").unwrap(),
             }),
 
-            DbType::Redis => Ok(SinkerConfig::Redis { url, batch_size }),
+            DbType::Redis => Ok(SinkerConfig::Redis {
+                url,
+                batch_size,
+                method: Self::get_optional_value(ini, SINKER, "method"),
+            }),
         }
     }
 

diff --git a/dt-common/src/error.rs b/dt-common/src/error.rs
@@ -1,70 +1,40 @@
-#[derive(Debug)]
+use thiserror::Error;
+
+#[derive(Error, Debug)]
 pub enum Error {
-    ConfigError {
-        error: String,
-    },
+    #[error("config error: {0}")]
+    ConfigError(String),
 
-    BinlogError {
-        error: mysql_binlog_connector_rust::binlog_error::BinlogError,
-    },
+    #[error("extractor error: {0}")]
+    ExtractorError(String),
 
-    SqlxError {
-        error: sqlx::Error,
-    },
+    #[error("sinker error: {0}")]
+    SinkerError(String),
 
-    Unexpected {
-        error: String,
-    },
+    #[error("pull mysql binlog error: {0}")]
+    BinlogError(#[from] mysql_binlog_connector_rust::binlog_error::BinlogError),
 
-    MetadataError {
-        error: String,
-    },
+    #[error("sqlx error: {0}")]
+    SqlxError(#[from] sqlx::Error),
 
-    IoError {
-        error: std::io::Error,
-    },
+    #[error("unexpected error: {0}")]
+    Unexpected(String),
 
-    YamlError {
-        error: serde_yaml::Error,
-    },
+    #[error("parse redis rdb error: {0}")]
+    RedisRdbError(String),
 
-    EnvVarError {
-        error: std::env::VarError,
-    },
+    #[error("metadata error: {0}")]
+    MetadataError(String),
 
-    StructError {
-        error: String,
-    },
+    #[error("io error: {0}")]
+    IoError(#[from] std::io::Error),
 
-    ColumnNotMatch,
-}
+    #[error("yaml error: {0}")]
+    YamlError(#[from] serde_yaml::Error),
 
-impl From<mysql_binlog_connector_rust::binlog_error::BinlogError> for Error {
-    fn from(err: mysql_binlog_connector_rust::binlog_error::BinlogError) -> Self {
-        Self::BinlogError { error: err }
-    }
-}
-
-impl From<sqlx::Error> for Error {
-    fn from(err: sqlx::Error) -> Self {
-        Self::SqlxError { error: err }
-    }
-}
-
-impl From<std::io::Error> for Error {
-    fn from(err: std::io::Error) -> Self {
-        Self::IoError { error: err }
-    }
-}
-
-impl From<serde_yaml::Error> for Error {
-    fn from(err: serde_yaml::Error) -> Self {
-        Self::YamlError { error: err }
-    }
-}
+    #[error("from utf8 error: {0}")]
+    FromUtf8Error(#[from] std::string::FromUtf8Error),
 
-impl From<std::env::VarError> for Error {
-    fn from(err: std::env::VarError) -> Self {
-        Self::EnvVarError { error: err }
-    }
+    #[error("struct error: {0}")]
+    StructError(String),
 }
diff --git a/dt-common/src/lib.rs b/dt-common/src/lib.rs
@@ -1,5 +1,3 @@
-// extern crate dt-meta;
-
 pub mod config;
 pub mod constants;
 pub mod error;

diff --git a/dt-common/src/utils/rdb_filter.rs b/dt-common/src/utils/rdb_filter.rs
@@ -21,6 +21,8 @@ pub struct RdbFilter {
     pub cache: HashMap<(String, String), bool>,
 }
 
+const DDL: &str = "ddl";
+
 impl RdbFilter {
     pub fn from_config(config: &FilterConfig, db_type: DbType) -> Result<Self, Error> {
         match config {
@@ -80,12 +82,20 @@ impl RdbFilter {
     }
 
     pub fn filter_event(&mut self, db: &str, tb: &str, row_type: &str) -> bool {
-        if self.do_events.is_empty() || !self.do_events.contains(&row_type.to_string()) {
+        if self.do_events.is_empty() || !self.do_events.contains(row_type) {
             return false;
         }
         self.filter_tb(db, tb)
     }
 
+    pub fn filter_ddl(&mut self) -> bool {
+        // filter ddl by default
+        if self.do_events.is_empty() {
+            return true;
+        }
+        !self.do_events.contains(DDL)
+    }
+
     fn contain_tb(
         set: &HashSet<(String, String)>,
         db: &str,
@@ -175,9 +185,10 @@ impl RdbFilter {
         let tokens = ConfigTokenParser::parse(config_str, &delimiters, escape_pairs);
         for token in tokens.iter() {
             if !SqlUtil::is_valid_token(token, db_type, escape_pairs) {
-                return Err(Error::ConfigError {
-                    error: format!("invalid filter config, check error near: {}", token),
-                });
+                return Err(Error::ConfigError(format!(
+                    "invalid filter config, check error near: {}",
+                    token
+                )));
             }
         }
         Ok(tokens)

diff --git a/dt-connector/src/extractor/mongo/mongo_cdc_extractor.rs b/dt-connector/src/extractor/mongo/mongo_cdc_extractor.rs
@@ -40,10 +40,6 @@ impl Extractor for MongoCdcExtractor {
         );
         self.extract_internal().await
     }
-
-    async fn close(&mut self) -> Result<(), Error> {
-        Ok(())
-    }
 }
 
 impl MongoCdcExtractor {

diff --git a/dt-connector/src/extractor/mongo/mongo_snapshot_extractor.rs b/dt-connector/src/extractor/mongo/mongo_snapshot_extractor.rs
@@ -40,10 +40,6 @@ impl Extractor for MongoSnapshotExtractor {
         );
         self.extract_internal().await
     }
-
-    async fn close(&mut self) -> Result<(), Error> {
-        Ok(())
-    }
 }
 
 impl MongoSnapshotExtractor {
-Original file line number
+Diff line change
@@ Expand Up / @@ -40,10 +40,6 @@ impl Extractor for MongoCdcExtractor { @@
             );
             self.extract_internal().await
         }
-        async fn close(&mut self) -> Result<(), Error> {
-            Ok(())
-        }
     }
     impl MongoCdcExtractor {
@@ Expand Down @@