cleanup and add docs

loomts · loomts · commit 47aacae9fa46 · 2026-01-26T15:25:24.000+08:00
diff --git a/docs/en/snapshot/check.md b/docs/en/snapshot/check.md
@@ -21,8 +21,10 @@ sample_interval=3
 This configuration is similar to the full synchronization task. The only differences are:
 
 ```
-[sinker]
-sink_type=check
+[checker]
+db_type=mysql
+url=mysql://user:pass@host:3306/db
+batch_size=2
 
 [parallelizer]
 parallel_type=rdb_check
@@ -58,10 +60,10 @@ Missing logs include database (schema), table (tb) and primary/unique key (id_co
 
 ## Output Full Row
 
-When the business needs full row content for troubleshooting exceptions, you can enable full row logging in `[sinker]`:
+When the business needs full row content for troubleshooting exceptions, you can enable full row logging in `[checker]`:
 
 ```
-[sinker]
+[checker]
 output_full_row=true
 ```
 
@@ -99,10 +101,10 @@ After enabling, all diff.log will append `src_row` and `dst_row`, and miss.log w
 
 ## Output Revise SQL
 
-If the business needs to manually repair different data, you can enable SQL output in `[sinker]`:
+If the business needs to manually repair different data, you can enable SQL output in `[checker]`:
 
 ```
-[sinker]
+[checker]
 output_revise_sql=true
 # Optional: force WHERE clause to match the whole row
 revise_match_full_row=true
@@ -157,7 +159,7 @@ The summary log contains the overall results of the check, such as start_time, e
 
 # Reverse Check
 
-Swap the [extractor] and [sinker] configurations to perform reverse check.
+Swap the [extractor] and [checker] target configurations to perform reverse check.
 
 # Other Configurations
 
diff --git a/docs/en/snapshot/review.md b/docs/en/snapshot/review.md
@@ -17,8 +17,9 @@ While this configuration is similar to that of snapshot migration, the only diff
 extract_type=check_log
 check_log_dir=./dt-tests/tests/mysql_to_mysql/revise/basic_test/check_log
 
-[sinker]
-sink_type=check
+[checker]
+db_type=mysql
+url=mysql://user:pass@host:3306/db
 
 [parallelizer]
 parallel_type=rdb_check
diff --git a/docs/en/tutorial/mongo_to_mongo.md b/docs/en/tutorial/mongo_to_mongo.md
@@ -125,9 +125,8 @@ db_type=mongo
 extract_type=snapshot
 url=mongodb://127.0.0.1:27017
 
-[sinker]
+[checker]
 db_type=mongo
-sink_type=check
 url=mongodb://ape_dts:123456@127.0.0.1:27018
 
 [filter]
@@ -226,9 +225,8 @@ extract_type=check_log
 url=mongodb://127.0.0.1:27017
 check_log_dir=./check_data_task_log
 
-[sinker]
+[checker]
 db_type=mongo
-sink_type=check
 url=mongodb://ape_dts:123456@127.0.0.1:27018
 
 [filter]
@@ -315,4 +313,4 @@ docker exec -it dst-mongo mongosh \
   { _id: ObjectId("670cc7d95bace351d307453c"), name: 'b', age: '4' },
   { _id: ObjectId("670ccb84b6456ba2539bb75a"), name: 'b', age: '5' }
 ]
-```
+```
diff --git a/docs/en/tutorial/mysql_to_mysql.md b/docs/en/tutorial/mysql_to_mysql.md
@@ -185,9 +185,8 @@ db_type=mysql
 extract_type=snapshot
 url=mysql://root:123456@127.0.0.1:3307?ssl-mode=disabled
 
-[sinker]
+[checker]
 db_type=mysql
-sink_type=check
 url=mysql://root:123456@127.0.0.1:3308?ssl-mode=disabled
 
 [filter]
@@ -288,9 +287,8 @@ extract_type=check_log
 url=mysql://root:123456@127.0.0.1:3307?ssl-mode=disabled
 check_log_dir=./check_data_task_log
 
-[sinker]
+[checker]
 db_type=mysql
-sink_type=check
 url=mysql://root:123456@127.0.0.1:3308?ssl-mode=disabled
 
 [filter]
@@ -505,4 +503,4 @@ SELECT * FROM test_db.tb_2;
 +----+-------+
 |  1 |     1 |
 +----+-------+
-```
+```
diff --git a/docs/en/tutorial/pg_to_pg.md b/docs/en/tutorial/pg_to_pg.md
@@ -186,9 +186,8 @@ db_type=pg
 extract_type=snapshot
 url=postgres://postgres:postgres@127.0.0.1:5433/postgres?options[statement_timeout]=10s
 
-[sinker]
+[checker]
 db_type=pg
-sink_type=check
 url=postgres://postgres:postgres@127.0.0.1:5434/postgres?options[statement_timeout]=10s
 
 [filter]
@@ -296,9 +295,8 @@ extract_type=check_log
 url=postgres://postgres:postgres@127.0.0.1:5433/postgres?options[statement_timeout]=10s
 check_log_dir=./check_data_task_log
 
-[sinker]
+[checker]
 db_type=pg
-sink_type=check
 url=postgres://postgres:postgres@127.0.0.1:5434/postgres?options[statement_timeout]=10s
 
 [filter]
diff --git a/docs/templates/mongo_to_mongo.md b/docs/templates/mongo_to_mongo.md
@@ -143,9 +143,8 @@ db_type=mongo
 extract_type=snapshot
 url=mongodb://ape_dts:123456@mongo1:9042/?replicaSet=rs0
 
-[sinker]
+[checker]
 db_type=mongo
-sink_type=check
 url=mongodb://ape_dts:123456@127.0.0.1:27018
 batch_size=200
 
@@ -233,9 +232,8 @@ url=mongodb://ape_dts:123456@mongo1:9042/?replicaSet=rs0
 check_log_dir=./logs/origin_check_log
 batch_size=200
 
-[sinker]
+[checker]
 db_type=mongo
-sink_type=check
 url=mongodb://ape_dts:123456@127.0.0.1:27018
 batch_size=200
 
@@ -265,4 +263,4 @@ log4rs_file=./log4rs.yaml
 log_dir=./logs
 ```
 
-- the output will be in {log_dir}/check/
+- the output will be in {log_dir}/check/
diff --git a/docs/templates/mysql_to_mysql.md b/docs/templates/mysql_to_mysql.md
@@ -222,11 +222,9 @@ db_type=mysql
 extract_type=struct
 url=mysql://root:123456@127.0.0.1:3307?ssl-mode=disabled
 
-[sinker]
+[checker]
 db_type=mysql
-sink_type=check
 url=mysql://root:123456@127.0.0.1:3308?ssl-mode=disabled
-replace=true
 
 [filter]
 do_dbs=test_db
@@ -264,9 +262,8 @@ extract_type=snapshot
 url=mysql://root:123456@127.0.0.1:3307?ssl-mode=disabled
 batch_size=10000
 
-[sinker]
+[checker]
 db_type=mysql
-sink_type=check
 url=mysql://root:123456@127.0.0.1:3308?ssl-mode=disabled
 batch_size=200
 
@@ -356,9 +353,8 @@ url=mysql://root:123456@127.0.0.1:3307?ssl-mode=disabled
 check_log_dir=./logs/origin_check_log
 batch_size=200
 
-[sinker]
+[checker]
 db_type=mysql
-sink_type=check
 url=mysql://root:123456@127.0.0.1:3308?ssl-mode=disabled
 batch_size=200
 
diff --git a/docs/templates/pg_to_pg.md b/docs/templates/pg_to_pg.md
@@ -175,9 +175,8 @@ db_type=pg
 extract_type=struct
 url=postgres://postgres:postgres@127.0.0.1:5433/postgres?options[statement_timeout]=10s
 
-[sinker]
+[checker]
 db_type=pg
-sink_type=check
 url=postgres://postgres:postgres@127.0.0.1:5434/postgres?options[statement_timeout]=10s
 
 [filter]
@@ -216,9 +215,8 @@ extract_type=snapshot
 url=postgres://postgres:postgres@127.0.0.1:5433/postgres?options[statement_timeout]=10s
 batch_size=10000
 
-[sinker]
+[checker]
 db_type=pg
-sink_type=check
 url=postgres://postgres:postgres@127.0.0.1:5434/postgres?options[statement_timeout]=10s
 batch_size=200
 
@@ -308,9 +306,8 @@ url=postgres://postgres:postgres@127.0.0.1:5433/postgres?options[statement_timeo
 check_log_dir=./logs/origin_check_log
 batch_size=200
 
-[sinker]
+[checker]
 db_type=pg
-sink_type=check
 url=postgres://postgres:postgres@127.0.0.1:5434/postgres?options[statement_timeout]=10s
 batch_size=200
 
diff --git a/docs/zh/snapshot/check.md b/docs/zh/snapshot/check.md
@@ -21,8 +21,10 @@ sample_interval=3
 此配置和全量同步任务的基本一致，两者的不同之处是：
 
 ```
-[sinker]
-sink_type=check
+[checker]
+db_type=mysql
+url=mysql://user:pass@host:3306/db
+batch_size=2
 
 [parallelizer]
 parallel_type=rdb_check
@@ -58,10 +60,10 @@ parallel_type=rdb_check
 
 ## 输出完整行
 
-当业务需要完整行内容用于排查异常，可以在 `[sinker]` 中开启全行日志：
+当业务需要完整行内容用于排查异常，可以在 `[checker]` 中开启全行日志：
 
 ```
-[sinker]
+[checker]
 output_full_row=true
 ```
 
@@ -99,10 +101,10 @@ output_full_row=true
 
 ## 输出修复 SQL
 
-业务若需要人工修复差异数据，可以在 `[sinker]` 中开启 SQL 输出：
+业务若需要人工修复差异数据，可以在 `[checker]` 中开启 SQL 输出：
 
 ```
-[sinker]
+[checker]
 output_revise_sql=true
 # 可选：强制使用全字段匹配 WHERE 条件
 revise_match_full_row=true
@@ -157,7 +159,7 @@ INSERT INTO `test_db_1`.`test_table`(`id`,`name`,`age`,`email`) VALUES(3,'Charli
 
 # 反向校验
 
-将 [extractor] 和 [sinker] 配置调换，即可进行反向校验。
+将 [extractor] 与 [checker] 目标配置调换，即可进行反向校验。
 
 # 其他配置
 
diff --git a/docs/zh/snapshot/review.md b/docs/zh/snapshot/review.md
@@ -17,8 +17,9 @@
 extract_type=check_log
 check_log_dir=./dt-tests/tests/mysql_to_mysql/revise/basic_test/check_log
 
-[sinker]
-sink_type=check
+[checker]
+db_type=mysql
+url=mysql://user:pass@host:3306/db
 
 [parallelizer]
 parallel_type=rdb_check
diff --git a/dt-common/src/config/task_config.rs b/dt-common/src/config/task_config.rs
@@ -87,7 +87,6 @@ const REVISE_MATCH_FULL_ROW: &str = "revise_match_full_row";
 const RETRY_INTERVAL_SECS: &str = "retry_interval_secs";
 const MAX_RETRIES: &str = "max_retries";
 const DB_TYPE: &str = "db_type";
-const SINK_TYPE: &str = "sink_type";
 const URL: &str = "url";
 const BATCH_SIZE: &str = "batch_size";
 const MAX_CONNECTIONS: &str = "max_connections";
@@ -454,15 +453,34 @@ impl TaskConfig {
 
     fn load_sinker_config(loader: &IniLoader) -> anyhow::Result<(BasicSinkerConfig, SinkerConfig)> {
         if !loader.ini.sections().contains(&SINKER.to_string()) {
-            if !loader.ini.sections().contains(&CHECKER.to_string()) {
-                bail!(Error::ConfigError(
-                    "config [sinker] is required when [checker] is not set".into()
-                ));
+            if loader.ini.sections().contains(&CHECKER.to_string()) {
+                let db_type: DbType = loader.get_required(CHECKER, DB_TYPE);
+                let url: String = loader.get_required(CHECKER, URL);
+                let connection_auth = ConnectionAuthConfig::from(loader, CHECKER);
+                let basic = BasicSinkerConfig {
+                    sink_type: SinkType::Dummy,
+                    db_type: db_type.clone(),
+                    url: url.clone(),
+                    connection_auth: connection_auth.clone(),
+                    batch_size: loader.get_with_default(CHECKER, BATCH_SIZE, 200),
+                    max_connections: loader.get_with_default(
+                        CHECKER,
+                        MAX_CONNECTIONS,
+                        DEFAULT_MAX_CONNECTIONS,
+                    ),
+                };
+                return Ok((basic, SinkerConfig::Dummy));
             }
+            bail!(Error::ConfigError(
+                "config [sinker] is required when [checker] is not set".into()
+            ));
+        }
+
+        let sink_type = loader.get_with_default(SINKER, "sink_type", SinkType::Write);
+        if let SinkType::Dummy = sink_type {
             return Ok((BasicSinkerConfig::default(), SinkerConfig::Dummy));
         }
 
-        let sink_type = loader.get_with_default(SINKER, SINK_TYPE, SinkType::Write);
         let db_type: DbType = loader.get_required(SINKER, DB_TYPE);
         let url: String = loader.get_optional(SINKER, URL);
         let batch_size: usize = loader.get_with_default(SINKER, BATCH_SIZE, 200);
@@ -757,11 +775,7 @@ impl TaskConfig {
                 default.retry_interval_secs,
             ),
             max_retries: loader.get_with_default(CHECKER, MAX_RETRIES, default.max_retries),
-            check_log_dir: loader.get_with_default(
-                CHECKER,
-                CHECK_LOG_DIR,
-                default.check_log_dir,
-            ),
+            check_log_dir: loader.get_with_default(CHECKER, CHECK_LOG_DIR, default.check_log_dir),
             check_log_file_size: loader.get_with_default(
                 CHECKER,
                 CHECK_LOG_FILE_SIZE,
@@ -780,6 +794,15 @@ impl TaskConfig {
         loader: &IniLoader,
         sinker_basic: &BasicSinkerConfig,
     ) -> anyhow::Result<(DbType, String, ConnectionAuthConfig)> {
+        let mut target = None;
+        if !sinker_basic.url.is_empty() {
+            target = Some((
+                sinker_basic.db_type.clone(),
+                sinker_basic.url.clone(),
+                sinker_basic.connection_auth.clone(),
+            ));
+        }
+
         let checker_target_present = loader.contains(CHECKER, DB_TYPE)
             || loader.contains(CHECKER, URL)
             || loader.contains(CHECKER, USERNAME)
@@ -788,15 +811,11 @@ impl TaskConfig {
             let db_type: DbType = loader.get_required(CHECKER, DB_TYPE);
             let url: String = loader.get_required(CHECKER, URL);
             let connection_auth = ConnectionAuthConfig::from(loader, CHECKER);
-            return Ok((db_type, url, connection_auth));
+            target = Some((db_type, url, connection_auth));
         }
 
-        if !matches!(&sinker_basic.sink_type, SinkType::Dummy) && !sinker_basic.url.is_empty() {
-            return Ok((
-                sinker_basic.db_type.clone(),
-                sinker_basic.url.clone(),
-                sinker_basic.connection_auth.clone(),
-            ));
+        if let Some(target) = target {
+            return Ok(target);
         }
 
         bail!(Error::ConfigError(
diff --git a/dt-connector/src/checker/base_checker.rs b/dt-connector/src/checker/base_checker.rs
diff --git a/dt-connector/src/checker/mysql_checker.rs b/dt-connector/src/checker/mysql_checker.rs
diff --git a/dt-connector/src/checker/pg_checker.rs b/dt-connector/src/checker/pg_checker.rs
diff --git a/dt-connector/src/extractor/mongo/mongo_cdc_extractor.rs b/dt-connector/src/extractor/mongo/mongo_cdc_extractor.rs
diff --git a/dt-task/src/task_runner.rs b/dt-task/src/task_runner.rs
