Realtime Compute for Apache Flink:Manage Paimon catalogs

• Create a table using the CREATE TABLE statement

  Run the following command in the SQL editor from Development > Scripts.

  The following example creates a partitioned table in the my_db database of my-catalog.

  -- Replace my-catalog with the name of your Paimon catalog.
  -- Replace my_db with the name of the database that you want to use.
  -- You can also replace my_tbl with a custom name in English.
  CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
    dt STRING,
    shop_id BIGINT,
    user_id BIGINT,
    num_orders INT,
    total_amount INT,
    PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
  ) PARTITIONED BY (dt) WITH (
    'bucket' = '4'
  );

  For more information, see Apache Paimon connector and Primary key tables and append-only tables.

• Create a table using the CREATE TABLE AS (CTAS) or CREATE DATABASE AS (CDAS) statement

  The CTAS and CDAS statements automatically synchronize data and table schema changes. Use these statements to easily replicate tables from sources like MySQL and Kafka to Paimon.

  Note

  • Enable dynamic bucketing by setting 'bucket' = '-1' is not supported for Primary key tables or Append-only tables (without primary keys).

  • The CTAS and CDAS statements allow you to configure properties that can be replicated downstream. For more information, see Paimon connector.

  • Replicate a specific table

    The following example automatically creates a Paimon table `my-catalog`.`my_db`.`web_sales` based on the schema of the mysql.tpcds.web_sales table and replicates the data.

    CREATE TABLE IF NOT EXISTS `<catalog name>`.`<db name>`.`<table name>`
    WITH (
      'bucket' = '4',
      'changelog-producer' = 'input'
    ) AS TABLE mysql.tpcds.web_sales;

  • Replicate an entire database

    The following example automatically creates Paimon tables in the `my-catalog`.`my_db` database based on the schema of each table in the mysql.tpcds database and replicates the data.

    CREATE DATABASE IF NOT EXISTS `<catalog name>`.`<db name>`
    WITH (
      'changelog-producer' = 'input'
    ) AS DATABASE mysql.tpcds INCLUDING ALL TABLES;

  • Replicate column type changes

    Paimon tables created using CTAS or CDAS statements support specific column type changes.

    • Default type replication

      By default, Paimon tables preserve the exact column types of the source. However, certain type promotions are supported without explicit configuration:

      • Integer types: TINYINT, SMALLINT, INT, and BIGINT can be promoted to an integer type of equal or higher precision. (TINYINT < SMALLINT < INT < BIGINT).

      • Floating-point types: FLOAT and DOUBLE can be promoted to a floating-point type of equal or higher precision. (FLOAT < DOUBLE).

      • String types: CHAR, VARCHAR, and STRING can be promoted to a string type of equal or higher precision.

    • Type normalization

      To enable more flexible type changes and prevent job failures from compatible schema evolutions, set 'enableTypeNormalization' = 'true' in the WITH clause of your CTAS/CDAS statement. In this mode, Flink normalizes data types during schema evolution. A job will only fail if the old and new types cannot be normalized to the same underlying type. Rules:

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

      • FLOAT, DOUBLE are normalized to DOUBLE.

      • CHAR, VARCHAR, STRING are normalized to STRING.

      • Other data types are not normalized.

    • Impact of normalization:

      When type normalization is active, the data types stored in the Paimon table will be the normalized type. For example, if source columns c1 (SMALLINT) and c2 (INT) are both normalized to BIGINT, both c1 and c2 will be stored as BIGINT in the Paimon table.

    • Examples:

      • Success: If SMALLINT in the source changes to INT, both types normalize to BIGINT. This is a compatible change, and the job continues normally.

      • Failure: If FLOAT in the source changes to BIGINT, their normalized types (DOUBLE and BIGINT) are incompatible, which will cause an exception.

Run the following commands in the SQL editor from Development > Scripts.

ALTER TABLE my_table SET (
  'write-buffer-size' = '256 MB',
  'write-buffer-spillable' = 'true'
);

ALTER TABLE my_table RENAME TO my_table_new;

ALTER TABLE my_table RENAME c0 TO c1;

ALTER TABLE my_table DROP (c1, c2);

ALTER TABLE my_table DROP PARTITION (`dt` = '20240108', `hh` = '08'), PARTITION (`dt` = '20240109', `hh` = '07');

ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';

ALTER TABLE my_table MODIFY col_a DOUBLE;

Run the following command in the SQL editor from Development > Scripts.

-- Replace my-catalog with the name of your Paimon catalog.
-- Replace my_db with the name of the database that you want to use.
-- Replace my_tbl with the name of the Paimon catalog table that you created.
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;

If the message The following statement has been executed successfully! is returned, the Paimon table is dropped.