CREATE TABLE AS (CTAS) statement (Retiring)

Data synchronization

Schema change synchronization

While synchronizing data in real time, the CTAS statement also replicates schema changes from the source table to the sink table. Schema changes include the initial table creation and any subsequent table alterations.

• Supported schema changes

  Schema change	Description
  Add a nullable column	The corresponding column is automatically added to the end of the sink table's schema, and its data is synchronized. The new column defaults to nullable, and its value for pre-existing rows is set to NULL.
  Add a non-null column	The corresponding column is automatically added to the end of the sink table's schema, and its data is synchronized.
  Delete a nullable column	The column is not removed from the sink table. Instead, its data is automatically set to NULL.
  Rename a column	This is treated as adding a new column and deleting the old one. The renamed column is added to the end of the sink table's schema, and the data in the original column is automatically set to NULL. Note For example, if col_a is renamed to col_b, col_b is added to the end of the sink table, and the data for col_a is set to NULL.
  Change a column's data type	If the sink system supports data type changes: Currently, only Paimon supports handling data type changes. CTAS supports type changes for standard columns, such as from INT to BIGINT. The supported type changes depend on the rules of the sink. Refer to the documentation for the specific sink connector for details. If the sink system does not support data type changes: Currently, only Hologres supports the type normalization mode to handle data type changes. In this mode, a CTAS job creates a downstream table with wider data types. This approach leverages the sink's compatibility with data type changes. For more details, see Example: Synchronize data in type normalization mode. Important You must enable type normalization mode when the CTAS job is first started. Otherwise, you must drop the sink table and perform a stateless restart for this mode to take effect.

  Important

  CTAS detects schema changes by comparing the schemas of consecutive records rather than by identifying the specific DDL operation.

  • If a column is deleted and then re-added without any data changes in between, CTAS does not detect a schema change.

  • CTAS detects a schema change only after a new column is added and data changes occur. It then synchronizes the schema change to the sink table.

• Unsupported schema changes

  • Changes to constraints, such as a primary key or index.

  • Deletion of non-null columns.

  • Changing a column from NOT NULL to NULLABLE.

  Important

  If an unsupported schema change occurs, you must manually drop the sink table and restart the CTAS job. This recreates the sink table and resynchronizes the historical data.

Single-table synchronization

Scenario: Synchronize the web_sales table from MySQL to Hologres.

Prerequisites: You have registered the following catalogs in your workspace.

• A Hologres catalog named holo.

• A MySQL catalog named mysql.

Code sample:

CTAS is typically used with catalogs for both the data source and the destination. A source catalog automatically parses the source table's schema and options, eliminating the need for manual DDL, enabling full and incremental data synchronization from the source table to the sink table.

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales   -- If no database is specified, the table is synchronized to the web_sales table in the default database.
WITH ('jdbcWriteBatchSize' = '1024')   -- Optional: Specifies parameters for the sink table.
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */;  -- Specifies additional parameters for the mysql-cdc source table.

Merge sharded tables and databases

Scenario: Use CTAS to merge multiple sharded MySQL tables into a single Hologres table.

Solution: In combination with a MySQL catalog, use a regular expression to match the database and table names of the tables you want to synchronize. The database and table names are written as two additional columns in the sink table. To ensure primary key uniqueness, the database name, table name, and original primary key are combined to form a new composite primary key for the Hologres table.

Code sample and merge result:

USE CATALOG holo;

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

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

Synchronize computed columns

Scenario: Add custom computed columns to a Hologres table during the process of merging and synchronizing sharded MySQL tables.

Code sample and merge result:

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

Multiple CTAS statements as a single job

Scenario: Synchronize the MySQL web_sales table and the sharded user tables to Hologres as a single job.

Solution: Use the STATEMENT SET syntax to commit multiple CTAS statements as a single job. This approach allows a single source node to be reused for reading data from multiple business tables, which reduces the number of server IDs, database connections, and read load on the MySQL CDC source.

Code sample:

USE CATALOG holo;

BEGIN STATEMENT SET;

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

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

END;

Synchronize one source to multiple sink tables

• If sink tables do not require computed columns

  USE CATALOG `holo`;
  
  BEGIN STATEMENT SET;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user table in the database1 of the Hologres data warehouse.
  CREATE TABLE IF NOT EXISTS `database1`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user table in the database2 of the Hologres data warehouse.
  CREATE TABLE IF NOT EXISTS `database2`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  END;

• If sink tables require computed columns

  -- Create a temporary table user_with_changed_id based on the source table user. It supports defining computed columns, such as computed_id, which is calculated based on the id from the source table.
  CREATE TEMPORARY TABLE `user_with_changed_id` (
    `computed_id` AS `id` + 1000
  ) LIKE `mysql`.`tpcds`.`user`;
  
  -- Create a temporary table user_with_changed_age based on the source table user. It supports defining computed columns, such as computed_age, which is calculated based on the age from the source table.
  CREATE TEMPORARY TABLE `user_with_changed_age` (
    `computed_age` AS `age` + 1
  ) LIKE `mysql`.`tpcds`.`user`;
  
  BEGIN STATEMENT SET;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user_with_changed_id table in the Hologres data warehouse. The table will contain the computed ID 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') */;
  
  -- Use a CTAS statement to synchronize the MySQL user table to the user_with_changed_age table in the Hologres data warehouse. The table will contain the computed age 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;

Synchronize a new table

Scenario: After a job with multiple CTAS statements is started, you need to add a new CTAS statement to synchronize data from a newly added table.

Solution: Enable the new table detection feature in the SQL job, add the new CTAS statement, and then restart the job from a savepoint. Once the new table is captured, its data will be synchronized.

Limitations:

• The new table detection feature is supported in VVR 8.0.1 and later.

• When synchronizing from a CDC source table, the new table detection feature is only supported for jobs with the source table startup mode set to initial.

• The configuration of the new source table in the added CTAS statement must be identical to the existing source table configurations to ensure that the source can be reused.

• The job configuration parameters, such as the startup mode, cannot be changed before and after adding the new CTAS statement.

Procedure:

1. When you need to add a new CTAS statement, go to the O&M (Operations and Maintenance) page, stop the job, and select Stop With Savepoint.

2. In the SQL job, enable new table detection, add the new CTAS statement, and then Deploy the job again.

   1. Add the following statement to the SQL job to enable new table detection.

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

   2. Add the new CTAS statement to the SQL job. The final complete code is as follows.

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

   3. Click Deploy.

3. Restore the job from the savepoint.

   1. On the O&M (Operations and Maintenance) page, click the name of the target job, go to the State Management tab, and click History.

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

   3. In the Actions column for the target savepoint, select More > Start Job from This Savepoint to start the job. For more information, see Start a job.

Synchronize to a Hologres partitioned table

Scenario: Use a CTAS statement to synchronize a MySQL source table to a Hologres partitioned table.

Hologres partition rule: In Hologres, if a primary key is defined for the sink table, the partition columns must be included in the primary key.

Code sample:

The DDL statement for the MySQL source 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)
);

The approach differs depending on whether the source table's primary key includes the partition key.

• If the source primary key includes the partition column, you can synchronize directly with a CTAS statement.

  Hologres automatically verifies that the partition column is part of the primary key.

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

• If the source primary key does not include the partition column, redeclare the sink table's primary key in the CTAS statement.

  If a partition column (for example, city) is not part of the source table's primary key, the job fails. You must redeclare the sink table's primary key in the CTAS statement to ensure the partition column is part of the primary key.

  -- You can use the following SQL to specify the primary key of the Hologres partitioned table 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`;

Synchronize data in type normalization mode

Scenario: When using a CTAS statement to synchronize data to a Hologres table, you need to support scenarios that require adjusting the precision of existing data types (for example, from VARCHAR(10) to VARCHAR(20)) or changing the data type (for example, from SMALLINT to INT).

Solution: Use the Hologres type normalization mode to synchronize data. The type normalization mode should be enabled when the CTAS job is first started. If it is not enabled at startup, you must drop the downstream table and perform a stateless restart of the job for it to take effect.

Type normalization rules:

After an upstream data type is changed, if the new and original types normalize to the same target type, the job runs normally. Otherwise, the change is considered incompatible, and the CTAS job throws an exception. The specific rules are as follows:

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

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

• FLOAT and DOUBLE are normalized to DOUBLE.

• Other data types are created according to the original type mapping rules. For more information, see Type mapping.

Code sample:

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

Synchronize MongoDB to Hologres

Limitations:

• Requires Realtime Compute for Apache Flink VVR 8.0.6 or later and MongoDB 6.0 or later.

• In SQL Hints, set scan.incremental.snapshot.enabled and scan.full-changelog to true.

• The pre- and post-images feature must be enabled in the MongoDB database. For information on how to enable it, see Document Preimages.

• Synchronizing multiple MongoDB collections in a single job has the following requirements:

  • The MongoDB configuration must be identical for each table, including hosts, scheme, username, password, and connectionOptions.

  • The scan.startup.mode configuration must be identical for each table.

Code sample:

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;