Realtime Compute for Apache Flink:CREATE DATABASE AS (CDAS)

Data synchronization

Schema evolution

When CDAS synchronizes a database, schema changes—such as adding columns—are propagated to the sink automatically. The behavior and policy are the same as for CTAS. See Schema evolution.

Syntax limitations

• Debugging an SQL draft that contains a CDAS statement is not supported.

• MiniBatch is not supported.

  Important

  Remove all MiniBatch configurations before creating an SQL draft with a CTAS or CDAS statement: 1. Go to O&M > Configurations. 2. Select the Deployment Defaults tab. 3. In the Other Configuration section, verify that MiniBatch configurations are removed. If you encounter a "Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax" error when creating or starting a deployment, see How do I fix this error?

Supported connectors

Parameters

Synchronize a database

Synchronize all tables from the tpcds MySQL database to Hologres.

Prerequisites:

• A Hologres catalog named holo is created in the workspace.

• A MySQL catalog named mysql is created in the workspace.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds          -- Create holo_tpcds in Hologres.
WITH ('sink.parallelism' = '4')                   -- Set sink parallelism (default: 4 for Hologres).
AS DATABASE mysql.tpcds INCLUDING ALL TABLES      -- Sync all tables from mysql.tpcds.
/*+ OPTIONS('server-id'='8001-8004') */;          -- Configure the MySQL CDC source options.

Options set in the WITH clause apply only to the current job and control write behavior. They are not persisted in the Hologres catalog. For supported options, see Hologres.

Consolidate and synchronize database shards

Merge all tables with identical names across multiple MySQL database shards into single Hologres tables.

Scenario: A MySQL instance has shards named order_db01 through order_db99. Each shard contains tables such as order and order_detail. All shard data—including schema changes—needs to land in Hologres.

Solution: Use a regular expression on the database name to match all shards. The database and table names are added to each sink table as two additional fields. The Hologres primary key includes the database name, table name, and source table's primary key columns to ensure uniqueness. There is no need to create target tables in advance.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order              -- Create holo_order in Hologres.
WITH('sink.parallelism'='4')                          -- Set sink parallelism (optional).
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES  -- Match all order_db shards.
/*+ OPTIONS('server-id'='8001-8004') */;              -- Configure MySQL CDC source options (optional).

Tables with identical names across shards are merged into a single Hologres table.

Synchronize new tables

After a running CDAS job has started, enable new table detection and restart from a savepoint to capture newly added tables.

New table detection requires VVR 8.0.6 or later. The source table's startup mode must be initial.

1. On the Deployments page, find the target deployment and click Cancel in the Actions column.

2. In the dialog, expand More Strategies, select Stop With Savepoint, and click OK.

3. In the SQL draft, add the following statement:

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

4. Click Deploy.

5. Recover the job from the savepoint:

   1. On the Deployments page, click the deployment name.

   2. On the deployment details page, click the State tab, then the History subtab.

   3. In the Savepoints list, find the savepoint created when you stopped the job.

   4. Choose More > Start job from this savepoint in the Actions column. See Start a job.

Run multiple CDAS and CTAS statements as one job

Synchronize multiple databases in a single job to maximize source reuse.

Scenario: Sync tpcds, tpch, and user_db01–user_db99 shards to Hologres in one job.

Solution: Wrap all statements in a STATEMENT SET block. Flink reuses a single source vertex across all statements, reducing the number of server IDs, database connections, and overall read load—especially for MySQL CDC sources.

USE CATALOG holo;

BEGIN STATEMENT SET;

-- Sync user tables from all user_db shards.
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sync tpcds database.
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sync tpch database.
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Synchronize multiple MySQL databases to Kafka

Sync tables from multiple MySQL databases to Kafka while avoiding topic name conflicts.

Scenario: Tables with the same name exist in both tpcds and tpch. Sending both to the same Kafka topic would overwrite data.

Solution: Use the cdas.topic.pattern option to generate unique topic names per database. The {table-name} placeholder is replaced at runtime by the actual table name. For example, 'cdas.topic.pattern'='tpcds-{table-name}' routes table1 from tpcds to the tpcds-table1 topic.

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- Sync tpcds database. Topic names follow the pattern "tpcds-<table_name>".
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') */;

-- Sync tpch database. Topic names follow the pattern "tpch-<table_name>".
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;

Introducing Kafka as an intermediate layer between MySQL and Flink also reduces the load on MySQL. See Synchronize MySQL database to Kafka with Flink CDC.