作者:lan
本文为 DM 源码阅读系列文章的第五篇。上篇文章 介绍了 dump 和 load 两个数据同步处理单元的设计实现,对核心 interface 实现、数据导入并发模型、数据导入暂停或中断的恢复进行了分析。本篇文章将详细地介绍 DM 核心处理单元 Binlog replication,内容包含 binlog 读取、过滤、路由、转换,以及执行等逻辑。 文内涉及到 shard merge 相关逻辑功能,如 column mapping、shard DDL 同步处理,会在 shard merge 篇单独详细讲解,这里就不赘述了。
从上图可以大致了解到 Binlog replication 的逻辑处理流程,对应的 逻辑入口代码。
从 relay log 或者 MySQL/MariaDB 读取 binlog events。
对 binlog events 进行处理转换( transformation ),这里可以做三类操作:
| 操作 | 说明 | |:-------|:-----------------| | Filter | 根据 库 /表同步黑白名单 对库 /表进行过滤;根据 binlog event 类型过滤。| | Routing | 根据 库 /表 路由规则 对库 /表名进行转换,用于合库合表。 | | Convert | 将 binlog 转换为 job 对象,发送到 executor。 |
executor 对 job 进行冲突检测,然后根据固定规则分发给对应的 worker 执行。
定期保存 binlog position/gtid 到 checkpoint。
Binlog replication 支持两种方式读取 binlog events:
两种方式都提供了同样的读取方法,处理核心都是 go-mysql。该库主要提供了两个功能:
更多的处理细节会在下篇关于 relay log 的文章中进行介绍,迫不及待的小伙伴可以先翻阅一下相关代码实现。
处理程序拿到解析好的 binlog event 后,根据 binlog 的类型来对 binlog 进行分类处理。Binlog replication 主要关心以下类型的 binlog event:
| 类型 | 说明 |
|:------------------|:---------------------|
| rotate event
| 消费完一个 binlog 文件,开始消费下一个 binlog 文件,用于更新 checkpoint 的 binlog position。 |
| row event
| 包含 insert/update/delete DML 数据。 |
| query event
| 包含 DDL 或者 statement DML 等数据。 |
| xid event
| 代表一个 transaction 的 commit,经过 go-mysql 的处理后带有对应 transaction
结束位置的 binlog position 和 gtid ,可以用来保存 checkpoint。 |
Binlog replication 数据处理单元会对每一类 binlog event 进行以下的处理步骤,具体实现的处理顺序可能略有差异,以代码实现为准。
Binlog replication 会从两个维度对 binlog event 来进行过滤:
row event
过滤处理 和 query event
过滤处理 的实现在逻辑上面存在一些差异:
row event
包含 库名和表名 信息;query event
需要通过 tidb parser 解析 event 里面包含的 query statement 来获取需要的库名,表名以及其他信息。[schema-pattern: *, table-pattern: *]
的 binlog event 过滤规则,来跳过 parser 不支持的 query statement。query event
里面也会包含 statement format binlog event,此时 Binlog replication 就可以利用 parser 解析出来具体的 statement 类型,对不支持的 statement format binlog event 作出相应的处理: 对于需要同步的表,进行报错处理;不需要同步的表,忽略继续同步。binlog 过滤完成之后,对于需要同步的表就会根据过滤步骤获得的库名和表名,通过 路由规则 转换得到需要同步到的目标库名和表名,在接下来的转换步骤来使用目标库名和表名来转换出正确的 DML 和 DDL statement。
row event
转换处理和 query event
转换处理的实现存在一些差异,这里分开来讲述。
row event
转换处理通过三个转换函数生成对应的 statements:
generate insert sqls
:将 write rows event
转换为 replace into statements
。generate update sqls
:
safe mode = true
,将 update rows event 转换为 delete + replace statements。safe mode = false
,将 update row event 转换为 update statements。generate delete sqls
:将 delete rows event 转换为 delete statements。query event
转换处理:
因为 TiDB 目前不支持一条 DDL 语句包含多个 DDL 操作,query event 转换处理会首先尝试将 包含多个 DDL 变更操作的单条 DDL 语句 拆分成 只包含一个 DDL 操作的多条 DDL 语句(具体代码实现)。
使用 parser 将 DDL statement 对应的 ast 结构里面的库名和表名替换成对应的目标库名和表名(具体代码实现)。
通过转换处理之后,将不同的 binlog event 包装成不同的 job 发送到 executor 执行:
binlog 顺序同步模型要求按照 binlog 顺序一个一个来同步 binlog event,这样的顺序同步势必不能满足高 QPS 低同步延迟的同步需求,并且不是所有的 binlog 涉及到的操作都存在冲突。Binlog replication 采用冲突检测机制,鉴别出来需要顺序执行的 jobs,在确保这些 jobs 的顺序执行的基础上,最大程度地保持其他 job 的并发执行来满足性能方面的要求。
冲突检测流程如下:
冲突检测实现比较简单,根据转换步骤获得每条 statement 对应的 primary/unique key
信息,来进行交集检测,如果存在交集那么认定是需要顺序的执行两条 statement,请参考 具体实现代码。
job 分发到对应的 worker 后,worker 根据一定的规则来批量执行这些 job,如下:
根据上面三个规则可以很快地将已经分发的 jobs 应用到下游 TiDB。
本篇文章详细地介绍 DM 核心处理单元 Binlog replication,内容包含 binlog 读取、过滤、路由、转换,以及执行等逻辑。下一篇我们会对 relay log 数据处理单元的设计进行详细的讲解。