CREATE DATABASE AS（CDAS）语句

示例一：整库同步

CDAS通常会配合数据源的Catalog和目标的Catalog一起使用。例如，MySQL Catalog和Hologres Catalog结合CDAS语法，完成MySQL到Hologres的全量和增量数据同步。使用MySQL Catalog可以自动解析源表的Schema及相应的参数，而不用手动编写DDL。

假设已在工作空间中注册了名为holo的Hologres Catalog和名为mysql的MySQL Catalog，MySQL中有一个名为tpcds的库。您可以使用以下语句将tpcds库下的24张表全部同步到Hologres中，包括未来的数据变更和表结构变更，无需提前在Hologres中创建表。

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds  -- 在hologres中创建holo_tpcds库。
WITH ('sink.parallelism' = '4') -- 可选，指定目标库的参数，每个holo sink默认使用4并发。
AS DATABASE mysql.tpcds INCLUDING ALL TABLES  -- 同步mysql中tpcds库下所有表。
/*+ OPTIONS('server-id'='8001-8004') */ ; -- 可选，指定mysql-cdc源表的额外参数。

示例二：分库合并同步

对于分库合并同步的场景，需要利用正则表达式的库名来匹配所要同步的多个分库。使用CDAS可以将上游多个分库下相同表名的数据合并同步到Hologres目标库对应表名的同一张表中，库名和表名会作为额外的两个字段写入到每张结果表中。为保证主键唯一性，库名、表名和原主键一起作为对应Hologres表的新联合主键。

假设MySQL实例中有order_db01~order_db99多个分库，每个分库下都有order、order_detail等多张表。您可以使用以下语句将99个分库下的order、order_detail等表全部同步到Hologres中，包括未来的数据变更和表结构变更，无需提前在Hologres中创建表。

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order--在Hologres中创建holo_order库，包括mysql中order分库的所有表。
WITH('sink.parallelism'='4')        --可选，指定目标库的参数，每个HologresSink默认并发为4。
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES --同步mysql中order_db分库下所有表。
/*+OPTIONS('server-id'='8001-8004')*/;  --可选，指定mysql-cdc源表的额外参数。

示例三：源库新增表加入数据同步

使用VVR 8.0.6及以上版本时，CDAS作业启动后，如果源库新增表，支持从作业快照重启，从而捕获到新的表，对新增表进行数据同步。

1. SQL作业开发时需要增加以下语句，开启CDAS新增表读取功能。

   SET 'table.cdas.scan.newly-added-table.enabled' = 'true';

2. 当出现新增的表需要同步时，停止作业并勾选停止前创建一次快照。

3. 在SQL开发中，重新部署这个SQL作业。

4. 在作业运维页面单击目标作业名称，状态集管理页签，单击历史。

5. 在作业快照列表中，找到停止作业时创建的快照。

6. 单击目标快照操作列，选择更多 > 从该快照恢复作业。

7. 在作业启动配置对话框，配置作业启动信息，详情请参见作业启动。

示例四：多CDAS&CTAS语句

实时计算Flink版支持使用STATEMENT SET语法将多个CTAS语句作为一个作业一起提交，并且可以对Source进行优化，复用一个Source节点读取多业务表的数据。这对于MySQL CDC数据源场景尤为适用，因为这可以减少server-id的使用，减少对数据库的连接数和读取压力。

假设MySQL实例中有tpcds、tpch、user_db01~user_db99（分库分表）多个库。您可以通过组合多条CDAS和CTAS语句，将MySQL实例下的所有库和表都同步到Hologres，只需一个Flink作业便能完成所有表的同步，只需一个Source便能读取所有表的数据，代码示例如下。

USE CATALOG holo;

BEGIN STATEMENT SET;

-- 同步user分库分表。
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

-- 同步TPCDS库。
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;
-- 同步TPCH库。
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

END;

示例五：多CDAS语句整库同步到Kafka

在使用多个CDAS语句整库同步到Kafka时，由于不同的数据库中可能存在相同的表，为了防止topic冲突，需要使用cdas.topic.pattern配置。cdas.topic.pattern定义了创建topic的名称的格式，其中可通过{table-name}占位符来替换为表名。如：当设置'cdas.topic.pattern'='db1-{table-name}'，对于上游表名为table1的表，在Kafka中对应的topic名称为db1-table1。

假设MySQL实例中有tpcds、tpch多个库。您可以通过如下方式将MySQL实例下的所有库和表都同步到Kafka，避免topic冲突，代码示例如下。

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- 同步TPCDS库。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpcds-{table-name}')
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

-- 同步TPCH库。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpch-{table-name}')
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

END;

实时计算Flink版提供MySQL整库同步到Kafka的能力，通过引入Kafka作为中间层，并使用CDAS整库同步或CTAS整表同步到Kafka来解决，具体操作请参见MySQL整库同步Kafka。