写点什么

为 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 的处理流程:



  1. 在目标库上创建影子表(ghost table,后缀为 _gho ),它的结构与被变更的原表完全相同;

  2. 创建日志表(log table,后缀为 _ghc )用于记录整个 DDL 执行过程中的状态;

  3. 在影子表上执行 DDL 语句;

  4. gh-ost 将自身伪装成 MySQL slave,创建 binlog streamer,将原表的全量数据和增量 binlog 迁移到影子表(两个操作同时进行);

  5. 迁移完毕后,将原表锁表,并重命名加上 _del 后缀,再将影子表重命名去掉 _gho 后缀,完成 cut-over。两个重命名操作是原子性执行的;

  6. 删除 _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~


用户头像

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

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

评论

发布
暂无评论
为TiDB DM添加阿里云RDS/DMS Online DDL支持_实践案例_TiDB 社区干货传送门_InfoQ写作社区