Create table as (CTAS) statement

Example 1: Single-table synchronization

Typically, the CREATE TABLE AS statement is used with both the data source's and the destination store's catalogs. For instance, you can synchronize full and incremental data from a MySQL database to Hologres using a MySQL catalog for the source table and a Hologres catalog for the destination table, without manually writing DDL statements.

For example, if you have a Hologres catalog named 'holo' and a MySQL catalog named 'mysql' in your workspace, you can synchronize data from the MySQL 'web_sales' table to Hologres with the following code:

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales  
WITH ('jdbcWriteBatchSize' = '1024')   -- Configure the parameters of the result table. This setting is optional.
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */;  -- Specify additional parameters for the mysql-cdc source table.

Example 2: merging and synchronization of multiple tables in a sharded database

To merge and synchronize data from multiple tables in a sharded database, use a MySQL catalog and a regular expression to match the tables you want to synchronize. The CREATE TABLE AS statement can then merge the data into a single Hologres table, with the shard and table names included as additional fields. To ensure a unique primary key, combine the shard name, table name, and original primary key as the new composite primary key in the Hologres 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') */;

The diagram below shows the merging process. If a new 'age' column is added to the 'user02' table and a record is inserted, the data and schema changes on the 'user02' table are automatically synchronized to the sink table in real time, even if the source tables' schemas differ.

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

Example 3: addition of custom computed columns

This example demonstrates how to add computed columns for data conversion and computation to source tables during the merging and synchronization process of multiple tables in a sharded database. The 'user' table serves as an example.

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 effect of adding a computed column synchronization is depicted in the following diagram.

The image below illustrates the synchronization effect.

Example 4: Execution of multiple CREATE TABLE AS statements as one deployment

Realtime Compute for Apache Flink allows multiple CREATE TABLE AS statements to be committed as a single deployment using the STATEMENT SET statement. This optimizes the data of source operators, enabling a single source operator to read data from multiple business tables. This is particularly beneficial for MySQL CDC data sources as it reduces server-id usage, the number of database connections, and the database reading load.

For instance, the first code segment synchronizes data from the 'web_sales' table, while the second code segment synchronizes data from multiple 'user' tables in a sharded database. You can use the STATEMENT SET statement to commit these segments as one deployment.

USE CATALOG holo;

BEGIN STATEMENT SET;

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

-- Synchronize data from multiple tables in the specified database shards whose names start with user.
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Example 5: Synchronize data from the same data source table to different result tables by using multiple CREATE TABLE AS statements

In Realtime Compute for Apache Flink version VVR 4.0.16 or later, you can use multiple CREATE TABLE AS statements to synchronize data from the same data source table to different result tables without adding computed columns.

USE CATALOG `holo`;

BEGIN STATEMENT SET;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user table in database1 of Hologres.
CREATE TABLE IF NOT EXISTS `database1`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user table in database2 of Hologres.
CREATE TABLE IF NOT EXISTS `database2`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

If you need to add computed columns to the result table, execute the following statements for data synchronization:

-- Create a temporary table named user_with_changed_id based on the source table user. You can define a computed column. The following sample code defines the computed_id column that is calculated based on the id column of the source table.
CREATE TEMPORARY TABLE `user_with_changed_id` (
  `computed_id` AS `id` + 1000
) LIKE `mysql`.`tpcds`.`user`;

-- Create a temporary table named user_with_changed_age based on the source table user. You can define a computed column. The following sample code defines the computed_age column that is calculated based on the age column of the source table.
CREATE TEMPORARY TABLE `user_with_changed_age` (
  `computed_age` AS `age` + 1
) LIKE `mysql`.`tpcds`.`user`;

BEGIN STATEMENT SET;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user_with_changed_id table of Hologres. The user_with_changed_id table contains the IDs that are obtained from the calculation based on the id column of the source table. The obtained IDs are in the computed_id column. 
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
AS TABLE `user_with_changed_id`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user_with_changed_age table of Hologres. The user_with_changed_age table contains the age values that are obtained from the calculation based on the age column of the source table. The obtained age values are in the computed_age column. 
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_age`
AS TABLE `user_with_changed_age`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Example 6: Add a CREATE TABLE AS statement to a deployment that executes multiple CREATE TABLE AS statements

In Realtime Compute for Apache Flink version VVR 8.0.1 or later, if you add a CREATE TABLE AS statement to a deployment already executing multiple CREATE TABLE AS statements, you can restart the deployment from a savepoint to include the new table and synchronize its data.

1. When developing an SQL draft, include the following statement to enable the feature for reading data from a new table:

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

2. To add a CREATE TABLE AS statement, stop the deployment on the Job Maintenance page and select Create A Snapshot Before Stopping.

3. In SQL Development, add a CREATE TABLE AS statement and Deploy the SQL deployment again.

4. On the Job Maintenance page, click the target deployment's name and select the State Management tab. Then, click History.

5. In the Job Snapshots list, locate the snapshot created when the deployment was stopped.

6. Click the Operation column of the desired snapshot and choose More > Restore Deployment From This Snapshot.

7. In the Job Startup configuration dialog box, set up the deployment's startup details. For more information, see Job Startup.

Example 7: Synchronize data from a MySQL source table to a partitioned table in Hologres by using the CREATE TABLE AS statement

When creating a partitioned table in Hologres, if a primary key is defined, the partition field must be included in the primary key. The following example shows how to synchronize data from a MySQL source table to a partitioned Hologres table using the CREATE TABLE AS statement.

Consider a MySQL table that needs to be synchronized to Hologres. The statement for creating the MySQL table is as follows:

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)
);

When synchronizing data using the CREATE TABLE AS statement, handle the relationship between the primary key and partition field of the upstream table according to Hologres' partitioned table rules.

• Scenario 1: The upstream table's primary key includes the partition field.

  If the partition field of the Hologres table, such as product_id, is part of the upstream table's primary key, you can directly synchronize data using the CREATE TABLE AS statement. For example:

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

  In this scenario, Hologres automatically verifies the inclusion of the partition field in the primary key and completes the synchronization.

• Scenario 2: The upstream table's primary key does not include the partition field.

  If the partition field of the Hologres table, such as city, is not part of the upstream table's primary key, synchronization fails because Hologres requires the partition field to be part of the primary key.

  To address this, redefine the result table's primary key in the CREATE TABLE AS statement to include the partition field. For example:

  -- You can use the following SQL statement to specify the primary key of the partitioned table in Hologres as order_id, product_id, and city.
  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`;

  This ensures the partition field city is included in the primary key definition, complying with Hologres' partitioned table requirements.

Example 8: Synchronize data to a Hologres table by using the CREATE TABLE AS statement in type normalization mode

When executing the CREATE TABLE AS statement, you may need to adjust the data type precision for existing fields, such as changing from VARCHAR(10) to VARCHAR(20).

• For Realtime Compute for Apache Flink versions earlier than vvr-6.0.5-flink-1.15, changing a field's data type in a source table's deployment may cause the deployment to fail, necessitating the recreation of the result table.

• For deployments in Realtime Compute for Apache Flink version vvr-6.0.5-flink-1.15 or later, the type normalization mode is available. It is recommended to enable this mode when starting a deployment for the first time. If not enabled initially, the mode does not take effect, and you must delete the downstream table and restart the deployment without state data.

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

  With type normalization mode, data type changes in the source table do not cause deployment failure as long as the pre- and post-change data types can be converted based on the normalization rules. The rules for type normalization mode are as follows:

  • TINYINT, SMALLINT, INT, and BIGINT data types are normalized to BIGINT.

  • CHAR, VARCHAR, and STRING data types are normalized to STRING.

  • FLOAT and DOUBLE data types are normalized to DOUBLE.

  • Other data types follow the original data type mapping rules. For more information, see Data type mappings.

  For instance, when type normalization mode is enabled, changing a SMALLINT data type to INT is considered successful because both are normalized to BIGINT, allowing the deployment to proceed as expected.

  • With type normalization enabled, SMALLINT and INT data types are normalized to BIGINT. Changing from SMALLINTto INT is considered successful, and the deployment runs smoothly.

  • With type normalization enabled, FLOAT is normalized to DOUBLE, and BIGINT remains BIGINT. Changing from FLOAT to BIGINT results in a data type incompatibility error.

Example 9: Synchronize data from a MongoDB source table to a Hologres table by using the CREATE TABLE AS statement

In Realtime Compute for Apache Flink version VVR 8.0.6 or later, the CREATE TABLE AS statement can synchronize data from a MongoDB source table to a Hologres table in real time, including schema changes. A MongoDB catalog can be used without manually defining the schema. For details on the MongoDB catalog, see Manage MongoDB Catalog.

The sample code below demonstrates how to use the CREATE TABLE AS statement to synchronize data from a MongoDB source table to a Hologres table:

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;

To synchronize data from multiple MongoDB collections in a deployment, ensure that the configurations of the following parameters for all tables are identical:

• Parameters related to the MongoDB database, such as hosts, scheme, username, password, and connectionOptions

• scan.startup.mode

Example 10: Synchronize data from all tables in a MySQL database to Kafka

When multiple deployments use the same MySQL table, the MySQL database establishes multiple connections, leading to high server and network load. To alleviate this, Realtime Compute for Apache Flink enables synchronization of data from all tables in a MySQL database to Kafka, using Kafka as an intermediate layer for data synchronization. The CREATE DATABASE AS statement can synchronize data from all tables in a MySQL database to Kafka, and the CREATE TABLE AS statement can do the same for individual tables. For more information, see Synchronize data from all tables in a MySQL database to Kafka.