摘要：作者王相本文為源碼閱讀系列文章的第七篇，在上篇文章中我們介紹了的實現，主要包括目錄結構定義數據的處理流程主從切換支持的讀取等邏輯。本篇文章我們將會對的定制化數據同步功能進行詳細的講解。

作者：王相

本文為 DM 源碼閱讀系列文章的第七篇，在 上篇文章 中我們介紹了 relay log 的實現，主要包括 relay log 目錄結構定義、relay log 數據的處理流程、主從切換支持、relay log 的讀取等邏輯。本篇文章我們將會對 DM 的定制化數據同步功能進行詳細的講解。

在一般的數據同步中，上下游的數據是一一對應的，即上下游的庫名、表名、列名以及每一列的值都是相同的，但是很多用戶因為業務的原因希望 DM 在同步數據到 TiDB 時進行一些定制化的轉化。下面我們將主要介紹數據同步定制化中的庫表路由（Table routing）、黑白名單（Black & white table lists）、列值轉化（Column mapping）、binlog 過濾（Binlog event filter）四個主要功能的實現。值得注意的是，由于其他一些工具（例如 TiDB Lightning 和 TiDB Binlog）也需要類似的功能，所以這四個功能都以 package 的形式維護在 tidb-tools 項目下，這樣方便使用和維護。

庫表路由顧名思義就是對庫名和表名根據一定的路由規則進行轉換。比如用戶在上游多個 MySQL 實例或者 schema 有多個邏輯上相同的表，需要把這些表的數據同步到 TiDB 集群的同一個表中，這個時候就可以使用 table-router 功能，如下圖所示：

該功能實現在 pkg/table-router 中，庫表路由的規則定義在結構 TableRule 中，其中的屬性 SchemaPattern 和 TablePattern 用于配置原庫名和表名的模式，TargetSchema 和 TargetTable 用于配置目標庫和表名，即符合指定 pattern 的庫和表名都將轉化成目標庫名和表名。

使用結構 Table 對路由規則進行維護，Table 提供了如下方法：

Table 結構中組合了 Selector，Selector 用于管理指定模式的庫、表的規則，提供如下方法：

Selector 的底層實現是 trieSelector，使用了單詞查找樹的結構來維護庫、表與規則的對應關系，感興趣的同學可以閱讀代碼深入了解一下。 trieSelector 中使用 cache 緩存了庫、表到規則的映射關系，這樣可以減少相同庫、表匹配規則的資源消耗。除了 table routing，以下的列值轉化和 binlog 過濾功能也都使用了 Selector，在下面的介紹中就不再贅述。

黑白名單功能用來選擇同步哪些庫和表，以及不同步哪些庫和表，這部分代碼維護在 pkg/filter 中。

黑白名單規則配置在 Rules 結構中，該結構包括 DoTables、DoDBs、IgnoreTables 和 IgnoreDBs 四個屬性，下面以判斷表 test.t 是否應該被過濾的例子說明配置的作用：

首先 schema 過濾判斷。

如果 do-dbs 不為空，則判斷 do-dbs 中是否存在一個匹配的 schema。

如果存在，則進入 table 過濾判斷。

如果不存在，則過濾 test.t。

如果 do-dbs 為空并且 ignore-dbs 不為空，則判斷 ignore-dbs 中是否存在一個匹配的 schema。

如果存在，則過濾 test.t。

如果不存在，則進入 table 過濾判斷。

如果 do-dbs 和 ignore-dbs 都為空，則進入 table 過濾判斷。

進行 table 過濾判斷。

如果 do-tables 不為空，則判斷 do-tables 中是否存在一個匹配的 table。

- 如果存在，則同步 `test.t`。
- 如果不存在，則過濾 `test.t`。

如果 ignore-tables 不為空，則判斷 ignore-tables 中是否存在一個匹配的 table。

- 如果存在，則過濾 `test.t`。
- 如果不存在，則同步 `test.t`。

如果 do-tables 和 ignore-tables 都為空，則同步 test.t。

使用 Filter 對黑白名單進行管理，Filter 提供了 ApplyOn 方法來判斷一組 table 中哪些表可以同步。

列值轉化功能用于對指定列的值做一些轉化，主要用于分庫分表的同步場景。比較典型的場景是：在上游分表中使用自增列作為主鍵，這樣數據在同步到 TiDB 的一個表時會出現主鍵沖突，因此我們需要根據一定規則對主鍵做轉化，保證每個主鍵在全局仍然是唯一的。

該功能實現在 pkg/column-mapping 中的 PartitionID：修改列的值的最高幾位為 PartitionID 的值（只能作用于 Int64 類型的列）。

代碼中使用 Rule 來設置 column mapping 的規則，Rule 的屬性及說明如下表所示：

Expression 為 PartitionID 的配置和轉化的計算方式都較為復雜，下面舉個例子說明。

例如 Arguments 為 [1, “test”, “t”, “_”]，1 表示數據庫實例的 InstanceID，“test” 為庫名稱的前綴，“t” 為表名稱的前綴，“_” 為前綴與 ID 的分隔符，則表 test_1.t_2 的 SchemaID 為 1，TableID 為 2。轉化列值時需要對 InstanceID、SchemaID、TableID 進行一定的位移計算，然后與原始的值進行或運算得出一個新的值。對于具體的計算方式，可以查看代碼 partitionID 和 computePartitionID。下面是一個 PartitionID 邏輯簡化后的示意圖：

使用 Mapping 結構對 column mapping 的規則進行管理，Mapping 提供列如下方法：

binlog 過濾功能支持過濾指定類型的 binlog，或者指定模式的 query，該功能維護在 pkg/binlog-filter 中。某些用戶不希望同步一些指定類型的 binlog，例如 drop table 和 truncate table，這樣就可以在下游仍然保存這些表的數據作為備份，或者某些 SQL 語句在 TiDB 中不兼容，希望可以在同步中過濾掉，都可以通過配置 binlog event filter 功能來實現。

首先需要對 binlog 進行分類，可以查看代碼 Event Type List。然后再定義過濾規則 BinlogEventRule，包括以下屬性：

例如，TiDB 對 ADD PARTITION 和 DROP PARTITION 語句不兼容，在同步時需要過濾掉相關的 SQL 語句，就可以在 DM 中使用如下配置：

filter-partition-rule:
    schema-pattern: "*"
    sql-pattern: ["ALTERs+TABLE[sS]*ADDs+PARTITION", "ALTERs+TABLE[sS]*DROPs+PARTITION"]
    action: Ignore

如果需要過濾掉所有的 DROP DATABASE 語句，則可以在 DM 中使用如下配置：

 filter-schema-rule:
    schema-pattern: "*"
    events: ["drop database"]
    action: Ignore

代碼中通過 BinlogEvent 結構對 binlog event 過濾規則做統一的管理，BinlogEvent 提供了如下的方法：

以上就是定制化數據同步功能中庫表路由（Table routing）、黑白名單（Black & white table lists）、列值轉化（Column mapping）、binlog 過濾（Binlog event filter）的實現介紹。歡迎大家閱讀相關代碼深入了解，也歡迎給我們提 pr 優化代碼。下一篇我們將介紹 DM 是如何支持上游 online DDL 工具（pt-osc，gh-ost）的 DDL 同步場景的。

原文閱讀：https://www.pingcap.com/blog-cn/dm-source-code-reading-7/