Realtime Compute for Apache Flink:CREATE TABLE AS (CTAS)

Required arguments

Optional arguments

Supported schema changes

Unsupported schema changes

The following schema changes are not replicated automatically:

• Modifying constraints (such as primary keys or indexes)

• Deleting non-nullable columns

• Changing a column from NOT NULL to NULLABLE

To apply these changes, manually drop the sink table and restart the job to re-synchronize all data.

Synchronize a single table

Synchronize the web_sales table from MySQL to Hologres.

Prerequisites:

• A Hologres catalog named holo

• A MySQL catalog named mysql

Use source and destination catalogs so that Flink can parse the source table's schema and properties automatically — no explicit DDL needed.

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales          -- Sync to web_sales in the default database.
WITH ('jdbcWriteBatchSize' = '1024')          -- Optional: sink connector options.
AS TABLE mysql.tpcds.web_sales
/*+ OPTIONS('server-id'='8001-8004') */;      -- Optional: source connector options.

Merge and synchronize table shards

Consolidate sharded MySQL tables and databases, then synchronize to a single Hologres table.

Use regular expressions to match source databases and tables. The database name and table name are written to two additional columns in the sink table. The sink table's primary key includes these columns to ensure uniqueness across shards.

Carets (^) cannot be used to match the beginning of a table name.

Consolidate table shards:

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

After the job starts, schema changes to individual shard tables are replicated in real time. For example, adding an age column to user02 and inserting a record:

ALTER TABLE `user02` ADD COLUMN `age` INT;
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30);

Both the schema change and the new record are synchronized to the sink table automatically.

Add computed columns

During sharded table consolidation, add custom computed columns to the sink table.

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */
ADD COLUMN (
  `c_id` AS `id` + 10 AFTER `id`,
  `calss` AS 3 AFTER `id`
);

The computed columns are treated as physical columns in the sink table and synchronized in real time. Specify FIRST or AFTER <column_name> to control column position.

Run multiple CTAS statements in a single job

Use BEGIN STATEMENT SET ... END to commit multiple CTAS statements as one job. This reuses the source operator across all statements, reducing the number of server IDs, database connections, and overall read load on the source.

USE CATALOG holo;

BEGIN STATEMENT SET;

-- Sync the web_sales table.
CREATE TABLE IF NOT EXISTS web_sales
AS TABLE mysql.tpcds.web_sales
/*+ OPTIONS('server-id'='8001-8004') */;

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

END;

Synchronize one source to multiple sinks

Without computed columns:

USE CATALOG `holo`;

BEGIN STATEMENT SET;

-- Sync to database1.
CREATE TABLE IF NOT EXISTS `database1`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sync to database2.
CREATE TABLE IF NOT EXISTS `database2`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

With computed columns: Create a temporary table for each sink to define the computed columns, then reference each in a CTAS statement.

-- Define a temporary table with a computed_id column.
CREATE TEMPORARY TABLE `user_with_changed_id` (
  `computed_id` AS `id` + 1000
) LIKE `mysql`.`tpcds`.`user`;

-- Define a temporary table with a computed_age column.
CREATE TEMPORARY TABLE `user_with_changed_age` (
  `computed_age` AS `age` + 1
) LIKE `mysql`.`tpcds`.`user`;

BEGIN STATEMENT SET;

-- Sync to holo.tpcds.user_with_changed_id.
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
AS TABLE `user_with_changed_id`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sync to holo.tpcds.user_with_changed_age.
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_age`
AS TABLE `user_with_changed_age`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Add CTAS statements to replicate new tables

After a multi-CTAS job starts, add a statement to replicate a newly added source table without restarting from scratch.

Limitations:

• Requires VVR 8.0.1 or later.

• Only jobs started in initial mode can detect new tables when synchronizing from a CDC source.

• The new CTAS statement's source connector options must match those of the existing statements — this is required for source operator reuse.

• Job configuration (such as startup mode) must stay the same before and after adding the statement.

Procedure:

1. On the Deployments page, find the 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, enable new table detection and add the new CTAS statement.

   1. Click Deploy.

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

   -- Enable new table detection.
   SET 'table.cdas.scan.newly-added-table.enabled' = 'true';
   
   USE CATALOG holo;
   
   BEGIN STATEMENT SET;
   
   -- Existing: sync web_sales.
   CREATE TABLE IF NOT EXISTS web_sales
   AS TABLE mysql.tpcds.web_sales
   /*+ OPTIONS('server-id'='8001-8004') */;
   
   -- Existing: sync user shards.
   CREATE TABLE IF NOT EXISTS user
   AS TABLE mysql.`wp.*`.`user[0-9]+`
   /*+ OPTIONS('server-id'='8001-8004') */;
   
   -- New: sync the product table.
   CREATE TABLE IF NOT EXISTS product
   AS TABLE mysql.tpcds.product
   /*+ OPTIONS('server-id'='8001-8004') */;
   
   END;

4. Restart 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 click History.

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

   4. Click More > Start job from this savepoint in the Actions column. For more information, see Start a job.

Synchronize to a Hologres partitioned table

Replicate data from MySQL to a Hologres partitioned table.

If a primary key is defined on the Hologres table, the partition columns must be included in that primary key.

Create the source MySQL table:

CREATE TABLE orders (
    order_id   INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    city       VARCHAR(100) NOT NULL,
    order_date DATE,
    purchaser  INTEGER,
    PRIMARY KEY(order_id, product_id)
);

If the source primary key already includes the partition columns — use the CTAS statement directly. Hologres verifies that the partition column is part of the primary key automatically.

CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`
PARTITIONED BY (product_id)
AS TABLE `mysql`.`tpcds`.`orders`;

If the source primary key excludes the partition columns — declare the sink table's primary key explicitly in the CTAS statement and include the partition column.

-- Declare order_id, product_id, and city as the composite primary key.
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`(
    CONSTRAINT `PK_order_id_city` PRIMARY KEY (`order_id`, `product_id`, `city`) NOT ENFORCED
)
PARTITIONED BY (city)
AS TABLE `mysql`.`tpcds`.`orders`;

Widen data types during replication

Change a column's precision or data type during synchronization — for example, from VARCHAR(10) to VARCHAR(20), or from SMALLINT to INT.

Enable type normalization mode when you first launch the job. For existing jobs, drop the Hologres sink table and restart without states to apply type normalization.

Type normalization rules:

If the original and new data types both normalize to the same target type, the change succeeds. Otherwise, an exception is raised.

CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`
WITH (
  'connector' = 'hologres',
  'enableTypeNormalization' = 'true'    -- Enable type normalization mode.
)
AS TABLE `mysql`.`tpcds`.`orders`;

Synchronize from MongoDB to Hologres

Limitations:

• Requires VVR 8.0.6 or later and MongoDB 6.0 or later.

• scan.incremental.snapshot.enabled and scan.full-changelog must both be set to true in the source connector options.

• The preimage and postimage features must be enabled on the MongoDB database. See Document preimages.

• For multi-collection jobs, the following connector options must be identical across all source tables: hosts, scheme, username, password, connectionOptions, and scan.startup.mode.

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table1`
AS TABLE `mongodb`.`database`.`collection1`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table2`
AS TABLE `mongodb`.`database`.`collection2`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

END;