为 TiDB DM 添加阿里云 RDS/DMS Online DDL 支持

2022 年 7 月 11 日
本文字数：4222 字
阅读完需：约 14 分钟

原文来源：https://tidb.net/blog/0d5a52c9

转载好友文章：

作者：LittleMagic

链接：https://www.jianshu.com/p/bb7dbebbc552

Online DDL 即无锁表结构变更，能够避免对表（尤其是大表）进行更改时，长时间阻塞 DML 操作。我们当前采用 TiDB 的 DM 组件实现上游许多业务库的合库合表与数据汇聚，DM 原生支持的 Online DDL 工具有 pt-osc（Percona 开源）与 gh-ost（GitHub 开源）两种。但是，我们的业务库绝大多数都是阿里云 RDS MySQL，采用其 DMS 工具做无锁变更。这导致我们每次都需要手动执行 DDL 语句，再指定位点恢复 DM 任务，操作起来比较繁琐，所以还是自己动手丰衣足食比较好。

gh-ost/DMS Online DDL Procedure

通过观察 DMS Online DDL 产生的 binlog，可以发现它的风格与 gh-ost 相同。简单复习一下 gh-ost 的处理流程：

在目标库上创建影子表（ghost table，后缀为 _gho ），它的结构与被变更的原表完全相同；
创建日志表（log table，后缀为 _ghc ）用于记录整个 DDL 执行过程中的状态；
在影子表上执行 DDL 语句；
gh-ost 将自身伪装成 MySQL slave，创建 binlog streamer，将原表的全量数据和增量 binlog 迁移到影子表（两个操作同时进行）；
迁移完毕后，将原表锁表，并重命名加上 _del 后缀，再将影子表重命名去掉 _gho 后缀，完成 cut-over。两个重命名操作是原子性执行的；
删除 _ghc 和 _del 表，并关闭 binlog streamer。

原子性 cut-over 的思想十分巧妙，基于 MySQL 内部锁表之后，执行 RENAME 操作的优先级比任何 DML 都要高这一简单的原理。具体的分析可参见国外大佬的这篇文章。

阿里云 DMS 的执行流程与上述一样，只是表的命名有所变化而已：

影子表： tp_[id]_ogt_[table]
日志表： tp_[id]_ogl_[table]
删除表： tp_[id]_del_[table]

下面直接上代码。

Hacking to the Code

DM 将 Online DDL 过程中的表分为 3 类，代码文件 online_ddl.go 中的定义是：

type TableType string
const (    realTable  TableType = "real table"    ghostTable TableType = "ghost table"    trashTable TableType = "trash table" // means we should ignore these tables)

复制代码

real table：原始表；
ghost table：影子表；
trash table：日志表与删除表。

另外还会通过 OnlineDDLStorage 结构体来维护执行过程中的必要信息，如数据库连接、schema/table 名称、DDL 语句等。

type OnlineDDLStorage struct {    sync.RWMutex
    cfg *config.SubTaskConfig
    db        *conn.BaseDB    dbConn    *DBConn    schema    string // schema name, set through task config    tableName string // table name with schema, now it's task name    id        string // the source ID of the upstream MySQL/MariaDB replica.
    // map ghost schema => [ghost table => ghost ddl info, ...]    ddls map[string]map[string]*GhostDDLInfo
    logCtx *tcontext.Context}

复制代码

DM 内部的 Online DDL 模块是插件化的，只需实现 OnlinePlugin 接口即可。该接口的定义如下，注释写得比较清楚，笔者就不多废话了。

// OnlinePlugin handles online ddl solutions like pt, gh-ost.type OnlinePlugin interface {    // Apply does:    // * detect online ddl    // * record changes    // * apply online ddl on real table    // returns sqls, replaced/self schema, replaced/self table, error    Apply(tctx *tcontext.Context, tables []*filter.Table, statement string, stmt ast.StmtNode) ([]string, string, string, error)    // Finish would delete online ddl from memory and storage    Finish(tctx *tcontext.Context, schema, table string) error    // TableType returns ghhost/real table    TableType(table string) TableType    // RealName returns real table name that removed ghost suffix and handled by table router    RealName(schema, table string) (string, string)    // ResetConn reset db connection    ResetConn(tctx *tcontext.Context) error    // Clear clears all online information    // TODO: not used now, check if we could remove it later    Clear(tctx *tcontext.Context) error    // Close closes online ddl plugin    Close()}

复制代码

我们需要重点实现的方法是 Apply()、TableType() 和 RealName()。后两者的逻辑比较简单，代码如下。

func (r *AliRDS) TableType(table string) TableType {    if len(table) > 8 && strings.HasPrefix(table, "tp_") {        if strings.Contains(table, "_ogt_") {            return ghostTable        }
        if strings.Contains(table, "_ogl_") || strings.Contains(table, "_del_") {            return trashTable        }    }
    return realTable}
func (r *AliRDS) RealName(schema, table string) (string, string) {    tp := r.TableType(table)    idx := -1
    if tp == ghostTable {        idx = strings.Index(table, "_ogt_")    } else if tp == trashTable {        idx = strings.Index(table, "_ogl_")        if idx == -1 {            idx = strings.Index(table, "_del_")        }    }
    if idx > 0 {        table = table[idx+5:]    }    return schema, table}

复制代码

接下来思考如何将 DMS 的步骤转化成 TiDB 的处理方式：

只执行原表上的 DML 操作；
不创建日志表；
不创建影子表，但是将要执行的 DDL 记录到 DM 元数据库（默认为 dm_meta ）以及 DM-Worker 的内存里；
在 cut-over 阶段，将上述记录的 DDL 的影子表名替换成原表名，直接执行替换后的 DDL。

然后就可以顺理成章地写出 Apply() 方法了。

func (r *AliRDS) Apply(tctx *tcontext.Context, tables []*filter.Table, statement string, stmt ast.StmtNode) ([]string, string, string, error) {    if len(tables) < 1 {        return nil, "", "", terror.ErrSyncerUnitAliRDSApplyEmptyTable.Generate()    }
    schema, table := tables[0].Schema, tables[0].Name    targetSchema, targetTable := r.RealName(schema, table)    tp := r.TableType(table)
    switch tp {    case realTable:        switch stmt.(type) {        case *ast.RenameTableStmt:            if len(tables) != parserpkg.SingleRenameTableNameNum {                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameTableNotValid.Generate()            }
            tp1 := r.TableType(tables[1].Name)            if tp1 == trashTable {                return nil, "", "", nil            } else if tp1 == ghostTable {                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameToGhostTable.Generate(statement)            }        }        return []string{statement}, schema, table, nil
    case trashTable:        switch stmt.(type) {        case *ast.RenameTableStmt:            if len(tables) != parserpkg.SingleRenameTableNameNum {                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameTableNotValid.Generate()            }
            tp1 := r.TableType(tables[1].Name)            if tp1 == ghostTable {                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameGhostTblToOther.Generate(statement)            }        }
    case ghostTable:        switch stmt.(type) {        case *ast.CreateTableStmt:            err := r.storage.Delete(tctx, schema, table)            if err != nil {                return nil, "", "", err            }        case *ast.DropTableStmt:            err := r.storage.Delete(tctx, schema, table)            if err != nil {                return nil, "", "", err            }        case *ast.RenameTableStmt:            if len(tables) != parserpkg.SingleRenameTableNameNum {                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameTableNotValid.Generate()            }
            tp1 := r.TableType(tables[1].Name)            if tp1 == realTable {                rdsInfo := r.storage.Get(schema, table)                if rdsInfo != nil {                    return rdsInfo.DDLs, tables[1].Schema, tables[1].Name, nil                }                return nil, "", "", terror.ErrSyncerUnitAliRDSOnlineDDLOnGhostTbl.Generate(schema, table)            } else if tp1 == ghostTable {                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameGhostTblToOther.Generate(statement)            }
            err := r.storage.Delete(tctx, schema, table)            if err != nil {                return nil, "", "", err            }        }
    default:        err := r.storage.Save(tctx, schema, table, targetSchema, targetTable, statement)        if err != nil {            return nil, "", "", err        }    }
    return nil, schema, table, nil}

复制代码

最后不要忘了添加 Online DDL 配置项的定义，以及相关的错误信息等边角代码，不一一列举了。

var OnlineDDLSchemes = map[string]func(*tcontext.Context, *config.SubTaskConfig) (OnlinePlugin, error){    config.PT:    NewPT,    config.GHOST: NewGhost,    config.ALIRDS: NewAliRDS,}
const (    GHOST  = "gh-ost"    PT     = "pt"    ALIRDS = "ali-rds")

复制代码

重新编译 dm-master 与 dm-worker 二进制文件，并替换掉原本通过 TiUP 部署的文件（具体路径因人而异）。

CGO_ENABLED=0 GOOS=linux GOARCH=amd64 make dm-masterCGO_ENABLED=0 GOOS=linux GOARCH=amd64 make dm-worker

复制代码

在 DM 作业的配置文件中指定 online-ddl-scheme 参数。

online-ddl-scheme: "ali-rds"

复制代码

使用 DMS 工具做无锁表变更操作，可以发现能够正常同步了。

The End

继续搬砖去了。

Enjoy~

发布于: 刚刚阅读数: 3

TiDB 社区干货传送门

关注

TiDB 社区官网:https://tidb.net/ 2021.12.15 加入

TiDB 社区干货传送门是由 TiDB 社区中布道师组委会自发组织的 TiDB 社区优质内容对外宣布的栏目，旨在加深 TiDBer 之间的交流和学习。一起构建有爱、互助、共创共建的 TiDB 社区 https://tidb.net/

发布

暂无评论

创作场景

为 TiDB DM 添加阿里云 RDS/DMS Online DDL 支持

gh-ost/DMS Online DDL Procedure

Hacking to the Code

The End

TiDB 社区干货传送门

评论