Manage Paimon catalogs | Realtime Compute for Apache Flink - Realtime Compute for Apache Flink

Paimon catalogs connect Flink to data lake metadata stored in OSS or DLF. Learn how to create and manage catalogs, databases, and tables.

Before you begin

Version requirements
- VVR 8.0.5 or later: Required to create Paimon catalogs and tables.
- VVR 11.1 or later: Required to use DLF as metastore.
- VVR 11.8 or later: Required to use OSS-HDFS as storage for Paimon Filesystem Catalog.

Storage requirements

Paimon stores data and metadata in OSS. Configure your OSS bucket as follows:
- Storage class: Use Standard. Storage class.
- Bucket setup: Create a dedicated bucket in the same region as your Flink workspace. Do not reuse the default system bucket to avoid data conflicts.
Access permissions
Ensure the AccessKey pair has read and write permissions on your OSS bucket or DLF directory.
Format compatibility

Paimon catalogs support only Paimon format. Other formats such as Lance and Iceberg are not supported.
Console refresh
After creating or deleting a catalog, database, or table, click refresh to update the console.

Paimon version mapping

Apache Paimon version	VVR
1.3.1	11.5, 11.6, 11.7, 11.8
1.3	11.4
1.2	11.2, 11.3
1.1	11.1
1.0	8.0.11

Create a catalog

Choose a metastore type based on your scenario:

Filesystem catalog: Stores metadata in OSS. Best for simple deployments.
DLF catalog: Stores metadata in Data Lake Formation (DLF). Best for enterprise scenarios with unified metadata management.

Create a Filesystem catalog

Option 1: Use the console (Recommended)

Go to the Catalogs page.
1. Log on to the Management Portal. Find the target workspace and click Console in the Actions column.
2. In the left navigation pane, click Catalogs.
Click Create Catalog, select Apache Paimon, and then click Next.
Set metastore to filesystem and configure other catalog settings.

Option 2: Use SQL

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

CREATE CATALOG `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = '<warehouse>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

General parameters

Configuration item	Description	Required	Remarks
`my-catalog`	The name of the Paimon catalog.	Yes	Enter a custom name.
`type`	The type of the catalog.	Yes	Set to `paimon`.
`metastore`	The type of the metastore.	Yes	Valid values: `filesystem`: Stores metadata in OSS. `dlf`: Stores metadata in DLF.

OSS parameters

Configuration item	Description	Required	Remarks
warehouse	The data warehouse directory in OSS.	Yes	The format is `oss://<bucket>/<object>`, where: `bucket`: the name of your OSS bucket. `object`: the path where your data is stored. View your bucket and object names in the OSS console.
fs.oss.endpoint	The endpoint of the OSS service.	Yes	If Flink and DLF are in the same region, use the VPC endpoint. Otherwise, use the public network endpoint. This parameter is required if the OSS bucket specified for warehouse is not in the same region as the Flink workspace, or if you use an OSS bucket that belongs to another Alibaba Cloud account. Regions and Endpoints and Create an AccessKey.
fs.oss.accessKeyId	The AccessKey ID of the Alibaba Cloud account or RAM user that has read and write permissions on OSS.	Yes
fs.oss.accessKeySecret	The AccessKey secret of the Alibaba Cloud account or RAM user that has read and write permissions on OSS.	Yes

OSS-HDFS (VVR 11.8 or later)

To use OSS-HDFS as storage for Paimon Filesystem Catalog, use the following SQL example:

CREATE CATALOG `my_catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = 'dls://<bucket>.<dls-endpoint>/<warehouse-path>',
  'fs.oss.endpoint' = '<dls-endpoint>',
  'fs.oss.accessKeyId' = '<access-key-id>',
  'fs.oss.accessKeySecret' = '<access-key-secret>',
  'fs.dls.client.mode.enabled' = 'false',
  'fs.dls.data.endpoint' = '<oss-data-endpoint>'
);

The following table describes the parameters.

Parameter	Description	Required	Remarks
warehouse	The warehouse directory in OSS-HDFS.	Yes	Format: dls://<bucket>.<dls-endpoint>/<warehouse-path>. The <dls-endpoint> is the same as `fs.oss.endpoint`.
fs.oss.endpoint	The endpoint of the OSS-HDFS service.	Yes	In the OSS console, go to the target bucket's "Overview > Access Ports" section and check the HDFS Service Endpoint.
fs.oss.accessKeyId	The AccessKey ID of the Alibaba Cloud account or RAM user that has read and write permissions on OSS-HDFS.	Yes	—
fs.oss.accessKeySecret	The AccessKey Secret corresponding to the AccessKey ID.	Yes	—
fs.dls.client.mode.enabled	Specifies whether to enable the DLS Client Mode.	Yes	Set to `false`.
fs.dls.data.endpoint	The data endpoint used by DLS to access underlying OSS data.	Yes	In the OSS console, go to the target bucket's "Overview > Access Ports" section and check the OSS internal Endpoint. You can also refer to the "Regions and endpoints" documentation.

Create a DLF catalog

DLF

Create a catalog in the DLF console in the same region as your Flink workspace. Quickly use DLF.

Note

This operation maps your DLF catalog as metadata only. Creating or deleting this catalog in Realtime Compute for Apache Flink does not affect the underlying data in DLF.

Log on to the Realtime Compute console.
In the list of fully managed workspaces, click the name of your workspace to open its console.

You can register the catalog by using the UI or by running SQL commands.

UI

In the navigation pane on the left, click Catalogs.
On the Catalog List page, click Create Catalog.
Select Apache Paimon and click Next.
Set metastore to dlf. For catalog name, select the desired DLF catalog. Click OK to create the catalog.

SQL commands

In the Scripts editor, enter and run the following SQL statement to create the catalog.

CREATE CATALOG `flink_catalog_name` 
WITH (
  'type' = 'paimon',
  'metastore' = 'rest',
  'token.provider' = 'dlf',
  'uri' = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com',
  'warehouse' = 'dlf_test'
);

The following table describes the parameters.

Parameter	Description	Required	Example
type	The type of the catalog. This parameter is fixed to `paimon`.	Yes	paimon
metastore	The type of the metastore. This parameter is fixed to `rest`.	Yes	rest
token.provider	The token provider. This parameter is fixed to `dlf`.	Yes	dlf
uri	The URI for accessing the DLF REST Catalog Server. The format is `http://[region-id]-vpc.dlf.aliyuncs.com`. For more information about region IDs, see Endpoints.	Yes	http://cn-hangzhou-vpc.dlf.aliyuncs.com
warehouse	The name of the underlying DLF catalog to use as the warehouse.	Yes	dlf_test

DLF-Legacy

Create a catalog in the DLF console in the same region as your Flink workspace. Getting Started.

Option 1: Use the console (Recommended)

Go to the Catalogs page.
1. Log on to the Management Portal and click Console in the Actions column of the target workspace.
2. Click Catalogs.
Click Create Catalog. Select Apache Paimon and click Next.
Set metastore to DLF and select the V1.0 DLF catalog to associate from the catalog name drop-down list. Then, configure other catalog settings.

Option 2: Use SQL

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

CREATE CATALOG `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'dlf',
  'warehouse' = '<warehouse>',
  'dlf.catalog.id' = '<dlf.catalog.id>',
  'dlf.catalog.accessKeyId' = '<dlf.catalog.accessKeyId>',
  'dlf.catalog.accessKeySecret' = '<dlf.catalog.accessKeySecret>',
  'dlf.catalog.endpoint' = '<dlf.catalog.endpoint>',
  'dlf.catalog.region' = '<dlf.catalog.region>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

Parameter descriptions:

General

Configuration item	Description	Required	Remarks
my-catalog	The name of the Paimon catalog.	Yes	Enter a custom name.
type	The type of the catalog.	Yes	The value is fixed to `paimon`.
metastore	The type of the metastore.	Yes	The value is fixed to `dlf`.

OSS

Configuration item	Description	Required	Remarks
warehouse	The data warehouse directory in OSS.	Yes	The format is `oss://<bucket>/<object>`, where: `bucket`: the name of your OSS bucket. `object`: the path where your data is stored. View your bucket and object names in the OSS console.
fs.oss.endpoint	The endpoint of the OSS service.	Yes	If Flink and DLF are in the same region, use the VPC endpoint. Otherwise, use the public network endpoint. If you want to store Paimon tables in OSS-HDFS, set the value of the `fs.oss.endpoint` parameter to `cn-<region>.oss-dls.aliyuncs.com`, for example, `cn-hangzhou.oss-dls.aliyuncs.com`.
fs.oss.accessKeyId	The AccessKey ID of the Alibaba Cloud account or RAM user that has read and write permissions on OSS.	Yes	For more information, see Regions and endpoints and Create an AccessKey.
fs.oss.accessKeySecret	The AccessKey secret of the Alibaba Cloud account or RAM user that has read and write permissions on OSS.	Yes

DLF

Configuration item	Description	Required	Remarks
dlf.catalog.id	The ID of the DLF data catalog.	Yes	View the ID of the data catalog in the Data Lake Formation console.
dlf.catalog.accessKeyId	The AccessKey ID required to access the DLF service.	Yes	For more information, see Create an AccessKey.
dlf.catalog.accessKeySecret	The AccessKey secret required to access the DLF service.	Yes	For more information, see Create an AccessKey.
dlf.catalog.endpoint	The endpoint of the DLF service.	Yes	Regions and endpoints. Note If Flink and DLF are in the same region, use the VPC endpoint. Otherwise, use the public network endpoint.
dlf.catalog.region	The region where DLF is located.	Yes	Regions and endpoints. Note Make sure that the region is the same as the one specified for dlf.catalog.endpoint.

Manage databases

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

Create a database

A default database is created automatically. To create more databases:

-- Replace my-catalog with the name of your Paimon catalog.
USE CATALOG `my-catalog`;

-- Replace my_db with a custom database name in English.
CREATE DATABASE `my_db`;

Drop a database

Important

The default database cannot be deleted from a DLF catalog. However, you can delete it from a Filesystem catalog.

-- Replace my-catalog with the name of your Paimon catalog.
USE CATALOG `my-catalog`;

-- Replace my_db with the name of the database that you want to delete.
DROP DATABASE `my_db`; -- Deletes a database only if it contains no tables.
DROP DATABASE `my_db` CASCADE; -- Deletes the database and all tables in it.

Manage tables

Create a table

Note

After you register a Paimon catalog, you can reference its tables directly in your Flink jobs without creating DDL definitions. To reference a table:

Use the fully qualified name: ${Paimon-catalog-name}.${Paimon-db-name}.${Paimon-table-name}
Or, set the current catalog and database using USE CATALOG ${Paimon-catalog-name} and USE ${Paimon-db-name}. Then reference tables using only their names: ${Paimon-table-name}

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 and append-only tables.

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

CTAS and CDAS automatically synchronize data and schema changes, replicating 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 Primary key and append-only tables.
- CTAS and CDAS support configuring replicable downstream properties. 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.

Modify a table schema

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

Operation	Sample code
Add or modify table properties	`ALTER TABLE my_table SET ( 'write-buffer-size' = '256 MB', 'write-buffer-spillable' = 'true' );`
Temporarily modify table properties	The temporarily modified table properties take effect only for the current SQL job. When you write to the my_table table, temporarily set `write-buffer-size` to 256 MB and `write-buffer-spillable` to true. `INSERT INTO my_table /+ OPTIONS('write-buffer-size' = '256 MB', 'write-buffer-spillable' = 'true') / SELECT ...;` When consuming data from the my_table table, temporarily set `scan.mode` to latest and `scan.parallelism` to 10. `SELECT * FROM t /+ OPTIONS('scan.mode' = 'latest', 'scan.parallelism' = '10') /;`
Rename a table	`ALTER TABLE my_table RENAME TO my_table_new;` Important OSS rename operations are not atomic. If you store Paimon table files in OSS, use OSS-HDFS for atomic file operations.
Add a new column	Add the `c1` column of type INT and the `c2` column of type STRING to the end of the `my_table` table. `ALTER TABLE my_table ADD (c1 INT, c2 STRING);` Add the `c2` column of type STRING after the `c1` column in the `my_table` table. `ALTER TABLE my_table ADD c2 STRING AFTER c1;` Add the `c1` column of type INT to the beginning of the `my_table` table. `ALTER TABLE my_table ADD c1 INT FIRST;`
Rename a column	Rename the c0 column to c1 in `my_table`. `ALTER TABLE my_table RENAME c0 TO c1;`
Drop a column	Drop the `c1` and `c2` columns from `my_table`. `ALTER TABLE my_table DROP (c1, c2);`
Drop a partition	Drop the `dt=20240108,hh=06` and `dt=20240109,hh=07` partitions from `my_table`. ALTER TABLE my_table DROP PARTITION (`dt` = '20240108', `hh` = '08'), PARTITION (`dt` = '20240109', `hh` = '07');
Modify a column comment	Change the comment of the `buy_count` column in `my_table` to this is buy count. `ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';`
Modify column order	Move `col_a` of type DOUBLE to the beginning of `my_table`. `ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;` Move `col_a` of type DOUBLE after `col_b` in `my_table`. `ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;`
Modify a column type	Change the type of `col_a` in `my_table` to DOUBLE. `ALTER TABLE my_table MODIFY col_a DOUBLE;` Supported column type conversions (〇 = supported, empty = not supported):

Drop a table

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.

View or drop a catalog

In the Management Portal, click Console in the Actions column for the target workspace.
On the Catalogs page, view or delete an Apache Paimon catalog.
- On the Catalog List page, view the catalog name and type. To view the databases and tables in a catalog, click View in the Actions column.
- On the Catalog List page, click Delete in the Actions column for the catalog to delete.
  
  Note
  Deleting a catalog removes only its Flink registration. The underlying data files are unaffected. Re-register the catalog to restore access.
  
  Alternatively, in the SQL editor from Development > Scripts, you can enter DROP CATALOG <catalog name>;, select the code, and then click Run.

Realtime Compute for Apache Flink:Manage Paimon catalogs

Before you begin

Create a catalog

Create a Filesystem catalog

Option 1: Use the console (Recommended)

Option 2: Use SQL

Create a DLF catalog

DLF

UI

SQL commands

DLF-Legacy

Option 1: Use the console (Recommended)

Option 2: Use SQL

Manage databases

Manage tables

Create a table

Modify a table schema

Drop a table

View or drop a catalog

References