时序表结构方案设计

异常现象截图

canal的坑：CanalParseException: column size is not match for table (表结构缓存异常阻塞问题)

背景

早期的canal版本(<=1.0.24)，在处理表结构的DDL变更时采用了一种简单的策略，在内存里维护了一个当前数据库内表结构的镜像(通过desc table获取)。

这样的内存表结构镜像的维护存在问题，如果当前在处理的binlog为历史时间段T0，当前时间为T1，存在的一些异常分支情况：

1. 假如在T0~T1的时间内，表结构A发生过增加列的DDL操作，那在处理T0时间段A表的binlog时，拿到的表结构为T1的镜像，就会出现列不匹配的情况. 比如之前的异常: column size is not match for table: xx , 12 vs 13
2. 假如在T0~T1发生了增加 C1列、删除了C2列，此时拿到的列的总数还是和T0时保持一致，但是对应的列会错位
3. 假如在T0~T1发生了drop table的DDL，此时拿表结构时会出现无法找到表的异常，一直阻塞整个binlog处理，比如not found [xx] in db

补充一下MySQL binlog的一些技术背景：

• MySQL的在记录DML(INSERT/UPDATE/DELETE)的binlog时，会由一个当前表结构snapshot的TableMap binlog来描述，然后跟着一条DML的binlog
• TableMap对象里，会记录一些基本信息：列的数量、列类型精度、后续DML binlog里的数据存储格式等，但唯独没有记录列名信息、列编码、列类型，这也是大众业务理解binlog的基本诉求(但MySQL binlog只做同构重放，可以不关注这些)，所以canal要做的一件事就是补全对应的列信息.

ps. 针对复杂的一条update中包含多张表的更新时，大家可以观察一下Table_map的特殊情况，留待有兴趣的同学发挥

方案

扯了一堆的背景之后，再来看一下我们如何解决canal上一版本存在的表结构一致性的问题，这里会把我们的思考过程都记录出来，方便大家辩证的看一下方案.

思考一

解决这个问题，第一个最直接的思考：canal在订阅binlog时，尽可能保持准实时，不做延迟回溯消费. 这样的方式会有对应的优点和缺点：

1. canal要做准实时解析，业务上可能有failover的需求，假如在业务处理离线时，原本canal基于内存ringBuffer的模型，会出现延迟解析，如果要解决这个问题，必须在canal store上支持了持久化存储的能力，比如实现或者转存到kafka/rocketmq等.
2. canal准实时解析，如果遇到canal本身的failover，比如zookeeper挂、网络异常，出现分钟级别以上的延迟，DDL变化的概率会比较高，此时就会陷入之前一样的表结构一致性的问题

整个方案上，基本是想避开表结构的问题，在遇到一些容灾场景下一定也会遇上，不是一个技术解决的方案，废弃.

思考二

经过了第一轮辩证的思考，基本确定想通过迂回的方式，简单绕过一致性的问题不是正解，所以这次的思考主要就是如何正面解决一致性的问题. 基本思路：基于binlog中DDL的变化，来动态维护一份表结构，比如DDL中增加一个列，在本地表结构中也动态增加一列，解析binlog时都从本地表结构中获取

实现方案：

1. 本地表结构的维护，每个canal进程可以带着一个二进制的MySQL版本，把收到的每条DDL，在本地MySQL中进行重放，从而维护一个本地的MySQL表结构
2. 每个canal第一次订阅或者回滚到指定位点，刚启动时需要拉取一份表结构基线，存入本地表结构MySQL库，然后在步骤1的方案上维护一个增量DDL.

整个方案上，可以绝大部分的解决DDL的问题，但也存在一些缺点：

1. 每个canal进程，维护一个隔离的MySQL实例。不论是资源成本、运维成本上都有一些瑕疵，更像是一个工程的解决方案，不是一个开源+技术产品的解决方案
2. 位点如果存在相对高频的位点回溯，每次都需要重新做表结构基线，做表结构基线也会概率遇上表结构一致性问题

思考三

有了之前的两次思考，思路基本明确了，在一次偶然的机会中和alibaba Druid的作者高铁，交流中得到了一些灵感，是否可以基于Druid对DDL的支持能力，来构建一份动态的表结构.

大致思路：

1. 首先准备一份表结构基线数据，每条建表语句传入druid的SchemaRepository.console()，构建一份druid的初始表结构
2. 之后在收到每条DDL变更时，把alter table add/drop column等，全部传递给druid，由druid识别ddl语句并在内存里执行具体的add/drop column的行为，维护一份最终的表结构
3. 定时把druid的内存表结构，做一份checkpoint，之后的位点回溯，可以是checkpoint + 增量DDL重放的方式来快速构建任意时间点的表结构

最终方案示意图

名词解释：

1. C0为初始化的checkpoint，拿到所有满足订阅条件的表结构
2. D1为binlog日志流中的DDL，它会有时间戳T的标签，用于记录不同D1/D2之间的先后关系
3. 定时产生一个checkpoint cm，并保存对应的checkpoint时间戳
4. 用户如果回溯位点到任意时间点Tx，对应的表结构就是 checkpoint + ddl增量的结合

接口设计:

public interface TableMetaTSDB {

    /**
     * 初始化
     */
    public boolean init(String destination);

    /**
     * 获取当前的表结构
     */
    public TableMeta find(String schema, String table);

    /**
     * 添加ddl到时间表结构库中
     */
    public boolean apply(BinlogPosition position, String schema, String ddl, String extra);

    /**
     * 回滚到指定位点的表结构
     */
    public boolean rollback(BinlogPosition position);

    /**
     * 生成快照内容
     */
    public Map<String/* schema */, String> snapshot();
}

名词解释：

1. 依赖了alibaba druid的DDL SQL解析能力，维护一份MemoryTableMeta，实时内存表结构
2. 依赖DAO持久化存储的能力，记录WAL结果(每条DDL) + checkpoint

持久化存储的思考：

1. 本地嵌入式实现(H2)：提供最小化的依赖，完成时序表结构管理的能力。基于磁盘的模式，可以结合存储计算分离的技术，canal failover之后只要在另一个计算节点上拉起，并加载云盘上的DB数据，做到多机冷备。
2. 中心管控存储实现(MySQL)： 一般结合于规模化的管控系统，允许将DDL数据录入到中心MySQL进行统一运维。

canal中如何使用

1. 打开conf/canal.properties，选择持久化存储的方案，默认为H2

canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml

1. 打开instance下的instance.properties，修改对应的参数

例子：

# table meta tsdb info
canal.instance.tsdb.enable=true
canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
canal.instance.tsdb.dbUsername=canal
canal.instance.tsdb.dbPassword=canal

如果不想搭建 tsdb 可以使用mysql保存数据库结构快照

创建表结构

CREATE TABLE `meta_snapshot` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  `gmt_create` datetime NOT NULL COMMENT '创建时间',
  `gmt_modified` datetime NOT NULL COMMENT '修改时间',
  `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
  `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
  `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
  `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id',
  `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳',
  `data` longtext COMMENT '表结构数据',
  `extra` text COMMENT '额外的扩展信息',
  PRIMARY KEY (`id`),
  UNIQUE KEY `binlog_file_offest` (`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  KEY `destination` (`destination`),
  KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  KEY `gmt_modified` (`gmt_modified`)
)  COMMENT='表结构记录表快照表';
 
CREATE TABLE `meta_history` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  `gmt_create` datetime NOT NULL COMMENT '创建时间',
  `gmt_modified` datetime NOT NULL COMMENT '修改时间',
  `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
  `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
  `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
  `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id',
  `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳',
  `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema',
  `sql_schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema',
  `sql_table` varchar(1024) DEFAULT NULL COMMENT '对应的table',
  `sql_text` longtext COMMENT '执行的sql',
  `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql类型',
  `extra` text COMMENT '额外的扩展信息',
  PRIMARY KEY (`id`),
  UNIQUE KEY `binlog_file_offest` (`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  KEY `destination` (`destination`),
  KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  KEY `gmt_modified` (`gmt_modified`)
)  COMMENT='表结构变化明细表';

修改 canal.properties

canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
#canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.url = jdbc:mysql://hostname:3306/tsdb?useUnicode=true&characterEncoding=UTF-8&useSSL=false
canal.instance.tsdb.dbUsername = username
canal.instance.tsdb.dbPassword = passwd
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml