数据同步 Schema Evolution

Schema Evolution 的重要性

在实际的业务场景中，数据库表结构并非一成不变。随着业务的发展，我们可能需要对表结构进行调整，如添加新字段、修改字段类型、删除字段等。在数据同步过程中，如果不能很好地处理这些结构变更，就可能导致同步失败或数据不一致。Schema Evolution（模式演化）功能允许在数据同步过程中自动处理源端的表结构变更，确保同步任务能够持续稳定运行，同时保持数据的一致性。

应用场景

在数据库应用领域，对于表结构的定义往往在应用上线前就已经定义好，包括表的 DDL 、索引创建和约束检查等。但随着业务的飞速发展，我们经常会面临对已有的表结构或者索引字段等进行一些变更的需求。在业务上线后，伴随着持续的读写流量，如何高效的能支持这类变更需求成为一个常见问题。传统的解法往往需要暂停业务的读写，获取一个升级窗口来进行表结构的变更等操作。在 ProtonBase 中，我们通过对 Schema Evolution 的高效支持，来解决此类问题。

Schema Evolution，全称为模式演化。是指随着新需求和数据模型的出现，跟随时间的推移修改数据库模式结构的过程。它涉及更改模式以适应新的数据元素或对现有数据元素的更改，同时保持数据的完整性和一致性。在我们本文讨论的范畴中，主要涉及以下场景：

在数据同步产品进行增量数据同步的过程中上游数据源端出现的表结构变更，比如新建表、表中新加字段、修改字段类型等，会自动将新的表结构同步到 ProtonBase 对于 Schema Evolution 的支持，对于不同的源端数据产品有不同的方式：
半结构化的数据产品，比如 MongoDB 。数据的存储是一个 JSON ，不存在 DDL ，但是 JSON 的键值可能发生变化
结构化的关系型数据库，通过某些方式拿到 DDL 变更，在目标端直接 Apply 变更，比如 MYSQL、PostgreSQL、SqlServer 和 Oracle 等

对于 MongoDB ， ProtonBase 数据同步产品开放了前端配置能力，可以让用户指定 Schema 结构。如果遇到初始 Schema 以外的字段， ProtonBase 数据同步产品会放到 Extra 字段来支持后续的 JSON 数据中出现的键值更新。

对于结构化的关系型数据库，Teleport 处理 Schema Evolution 的基本流程一致，但不同源端捕获 DDL 的方式不同。MySQL 可以从 binlog 中获取 DDL；PostgreSQL 的逻辑复制流本身不包含完整 DDL，Teleport 默认通过 Event Trigger 捕获 DDL，并重新读取变更后的完整表结构；ProtonBase 导出当前使用 Relation Message 探测行结构变化。捕获到变化后，Teleport 会生成统一的结构差异，更新带位点的元数据版本和映射，再根据目标端类型决定是否执行 DDL。

PG 生态下的源端数据同步，通过 CDC 来实现。CDC，全称 Change Data Capture，是一种观察写入数据库的所有变更，并将其转换为可以复制到其他系统中的形式的过程。ProtonBase 数据同步产品使用逻辑复制来订阅上游的变更操作。逻辑复制是一种高级的复制技术，它可以在不同的 PostgreSQL 数据库之间同步数据，甚至可以跨版本同步数据。与传统的流复制（streaming replication）相比，逻辑复制可以更加灵活地选择需要复制的表和列，也可以过滤掉不需要复制的数据。此外，逻辑复制可以在主库上执行数据转换和过滤操作，这对于一些特殊需求非常有用，例如数据仓库和实时数据分析等。逻辑复制的增量消息解析使用 pgoutput 插件。标准的逻辑解码输出插件。它由 PostgreSQL 社区维护，并由 PostgreSQL 本身用于逻辑复制。该插件始终存在，因此无需安装额外的库。

在 PostgreSQL 逻辑复制中，Teleport 支持两种 DDL 识别路径：

Event Trigger：捕获 CREATE TABLE、ALTER TABLE、DROP TABLE、索引和注释等 DDL，再读取变更后的完整表结构。该方式能够区分加列、删列、重命名列、修改类型、键和约束等绝大多数常见演进，是 PostgreSQL 源端的默认和推荐方式。
Relation Message：比较 Relation 消息携带的列名和类型与已有元数据。它无需部署 Event Trigger，但只能可靠识别会改变行结构的部分变化，也无法完整还原 DDL 语义。

Event Trigger 所需的函数和 Trigger 必须存在于每个被同步的 Database 中。Teleport 会在预检查阶段检查这些对象；同步账户有足够权限时可以自动创建，否则应由数据库管理员按预检查提供的 SQL 提前创建。

对于 ProtonBase 而言，使用 ProtonBase 数据同步产品同步数据的时候往往对应如下场景:

一次性将源端数据同步到ProtonBase，然后基于此份快照数据进行POC验证
将生产库的全量 + 增量数据同步到 ProtonBase，ProtonBase作为一个只读备实例存在，方便业务上进行 A/B Test

PostgreSQL 源端默认使用 Event Trigger，以获得更完整的演进能力。ProtonBase 源端当前使用 Relation Message；因此 ProtonBase 导出到数据库时，对删列、重命名列等无法安全还原的变化会将受影响的单表下线，而导出到 Kafka 时只更新元数据和消息映射，不会因为目标端无法执行 DDL 而下线表。

源端配置要求

以下设置为进行PG逻辑订阅的必须条件，是PG自身逻辑复制机制的需求。实际在使用其他一些类似同步工具的时候也同样要有这些设置的要求。

设置 wal_level 为 logical。wal_level 有三种级别：minimal、replica 和 logical。minimal 记录的 WAL 日志信息最少，只包含了数据库异常关闭需要恢复时的基本 WAL 信息； logical 记录的WAL日志信息最多，包含了支持逻辑解析所需的WAL。replica记录的 WAL 信息介于二者之间。只有开启 logical 级别的 WAL 日志信息采集，才会记录 Relation Message。通过如下命令进行设置：

 ALTER SYSTEM SET wal_level TO 'logical';

表被 publication 订阅

方式一：一开始订阅全部的表

 CREATE PUBLICATION xx_pub FOR ALL TABLES;

方式二：用户建好表后，在新表写入数据前，修改订阅

ALTER PUBLICATION publication_name ADD TABLE table_name;

需要同步的表必须有主键。如果是无主键表(包括只有唯一键但没有主键的表)，需要将 replica identity 设置为 full 模式，否则会导致源端PG库无法做 update 和 delete 。建议将所有表设置为有主键表。

alter table t replica identity full;

DDL 变更检测机制

上图是ProtonBase 数据同步产品的整个导数据的流程。在没有 Schema Evolution 之前初始化获取全量信息时，直接读取源端的数据库的元信息来建表即可，不需要记录任何状态。支持Schema Eovoluiton后，需要在第一次获取表 meta 信息的 snapshot，并记录状态。并在后续发生 DDL 变更时，将变更也落入状态中。对于 ProtonBase 数据同步产品而言，slice 是内部资源调度的最小单位，一个 slice 负责启动一个同步若干条数据的任务。PG的数据同步按照 Database 粒度进行，一个 DataBase 的同步会对应一个流。 ProtonBase 数据同步产品 Schema Evolution 的基本流程如下：

第一次启动，先按流的个数分 slice，并将 slice 的id记录状态。因为各 slice 的启动位点、消费进度可能都有差别，所以没法记录统一的状态
每个 slice，获取对应的Meta信息，并且记录 snapshot 状态
Source Connector 通过增量日志、Event Trigger 事件或 Relation Message 检测 DDL 变更，并与当前 Meta 比较生成结构差异。对数据库目标执行可安全转换的 DDL；对 Kafka 等非结构化目标只刷新 Meta 和映射。变更后的 Meta 会按新版本写入状态。

核心能力

目前 ProtonBase 数据同步产品支持在线的对源端的表的 DDL 变更操作的自动同步，比如新建表、新建列、删除列和更改列数据类型等常见的操作，一般在秒级可以完成。对于遇到的不支持的异常变更操作，ProtonBase 数据同步产品也会有自我保护机制，中断个别表的同步保持整个数据增量同步流程持续进行。除此之外，相比市面上的其他同步工具 ProtonBase 数据同步产品还额外做了诸多优化。

时间旅行支持

ProtonBase 数据同步产品通过识别Relation消息的变更生成不同位点的元信息，并且记录到状态表。如果客户想从某个位点重新追增量，ProtonBas e数据同步产品可以从状态表中获取到当时位点对应的元信息，这样就可以支持从任意点位进行增量数据的回放。以如下表为例，我们讲解下 ProtonBase 数据同步产品 TimeTravel 的能力。

CREATE TABLE COMPANY(
   ID INT PRIMARY KEY     NOT NULL,
   NAME           TEXT    NOT NULL,
   AGE            INT     NOT NULL,
   ADDRESS        CHAR(50),
   SALARY         REAL
);

为方便理解，下面落入状态表的信息均按照\d展示的表结构讲解，不按照实际的存储信息展示。全量同步开始T0时刻，落入状态表表的 Schema 定义如下：

protonbase=> \d COMPANY
                  Table "public.company"
 Column  |     Type      | Collation | Nullable | Default
---------+---------------+-----------+----------+---------
 id      | integer       |           | not null |
 name    | text          |           | not null |
 age     | integer       |           | not null |
 address | character(50) |           |          |
 salary  | real          |           |          |
Indexes:
    "company_pkey" PRIMARY KEY, btree (id ASC)

T1时刻同步完所有全量 + 追完同步全量期间的增量后，开始进行实时增量同步。 T2时刻遇到加列的Schema Evolution，加了一个新列 department ，新的表定义如下，落入状态表并且更新Version。

protonbase=> \d COMPANY
                   Table "public.company"
   Column   |     Type      | Collation | Nullable | Default
------------+---------------+-----------+----------+---------
 id         | integer       |           | not null |
 name       | text          |           | not null |
 age        | integer       |           | not null |
 address    | character(50) |           |          |
 salary     | real          |           |          |
 department | text          |           |          |
Indexes:
    "company_pkey" PRIMARY KEY, btree (id ASC)

T3时刻遇到修改字段类型的Schema Evolution，age字段从INT变为BIGINT，新的表定义如下，落入状态表并且更新Version。

protonbase=> \d COMPANY
                   Table "public.company"
   Column   |     Type      | Collation | Nullable | Default
------------+---------------+-----------+----------+---------
 id         | integer       |           | not null |
 name       | text          |           | not null |
 age        | bigint        |           | not null |
 address    | character(50) |           |          |
 salary     | real          |           |          |
 department | text          |           |          |
Indexes:
    "company_pkey" PRIMARY KEY, btree (id ASC)

如果因为某些原因，客户想从 T2' 时刻开始开始重追增量的时候，在[T2', T3]这个时间区间，ProtonBase数据同步产品会从状态表中获取最相近的一个时刻T2的元信息，对增量消息按那个时刻的结构来进行解析。通过类似的方式， ProtonBase 数据同步产品可以支持TimeTravel的功能，从指定的任意时刻(位点)开始追增量并且还原到当时的元信息就行增量消息解析。

事务支持

目前常见的数据同步工具，都不保证增量数据同步过程中的事务，只保证了最终一致性。ProtonBase 数据同步产品在实现的时候，按照源端的操作顺序，也保持了对事务的支持，从而保证了同步过程中和源端数据的严格一致性。同时，对于过程中出现了事务交叉的场景， ProtonBase 数据同步产品也进行了针对性处理来解决这种场景容易出现的一致性问题。比如如下场景：

begin;   -- 事务1
 insert into t1 (id, name) values(1, 'a');  -- 事务1
 alter t1 add column age int;  -- 事务1，修改表1的元数据，添加了age字段
 insert into t1 (id, name, age) values(2, 'b', 10);  -- 事务1;
 
 begin;  -- 事务2
 insert into t2 (id, name) values(1, 'a');  -- 事务2
 alter t2 add column age int; -- 事务2，修改表2的元数据，添加了age字段
 commit;   -- 事务2
 
 begin;   --事务3
 insert into t1 (id, name) values(10, 'b'); --事务3，应该看到事务1之前的t1表结构
 commit;  --事务3
 
 insert into t1 (id, name, age) values(3, 'c', 11);  -- 事务1
 rollback;  -- 事务1
 
 begin;   --事务4
 insert into t1 (id, name) values(2, 'b'); --事务4，应该看到事务1之前的t1表结构
 commit;  --事务4

如上所示，是多个并发操作的事务最后记录在数据库中的 wal log 中的顺序。如果在同步的时候有元数据同步技术未支持事务的话，当执行了t1表的变更后，t1表的变更即对其它事务均可见，后续事务1的回滚操作也未执行。会导致与预期的语义不符。

分库分表合并导入

为了解决分库分表的数据导入 ProtonBase 的问题，ProtonBase数据同步产品支持按照基于规则的前缀匹配，来将多个拆分后的表同步到同一个 ProtonBase 的表里。随着业务的发展，在后续新分出的表中，可能会比之前的表有更多的列， ProtonBase 数据同步产品也会通过合并所有表中的列，打成大宽表的方式支持所有源表的同步。除了一次性的同步，ProtonBase 数据同步产品对后续新建的满足表名匹配规则的表也会进行增量同步，更好的满足业务需求。通过该机制，可以很容易的将分库分表的业务表导入到 ProtonBase 。语法示例如下：

case
when ${expression} like ${expression} then ${expression}
when regexp_match(${expression}, ${regex}) like then ${expression}
...
when ${condition} then ${expression}
else ${expression}
end

其中 regex 为标准正则表达方式，支持将 () 内的数据作为分组提取出来，提取的变量名为 $1, $2,......, $n，可以用于 expression 表达式。例子：将源端 t1_000,t1_001,......,t1_009 映射到目标端 t1，将源端 t2_000,t2_001,......,t2_009 映射到目标端 t2 通过 like 配置分表映射。

case
when SOURCE_OBJECT like 't1_%' then concat(SOURCE_DATABASE, '.public.t1')
when SOURCE_OBJECT like 't2_%' then concat(SOURCE_DATABASE, '.public.t2')
else concat(SOURCE_DATABASE, '.', 'public', '.', SOURCE_OBJECT)
end

实例级别 Schema Evolution

不同于其他同步工具一次只能同步一个DB的配置方式，ProtonBase数据同步产品一次配置同步，可以将整个实例的数据进行同步并且同时支持多库 Schema Evolution。配合前端 UI 的简易操作，用户可以很容易的按需对整个实例数据或者部分表进行同步。

灵活开关 Schema Evolution

ProtonBase数据同步产品默认开启了Schema Evolution的能力。在开始数据同步前，通过ProtonBase产品的控制台页面，用户可以很容易的对数据库进行 Schema Evolution 能力的开启/关闭。如果对已经在持续增量订阅的数据同步任务进行变更，需要先暂停同步任务，修改配置后重新点击"恢复"按钮启动任务即可生效。