Write and consume Paimon table data - Realtime Compute for Apache Flink

Prerequisites

You must first create a Paimon catalog and a Paimon table. For more information, see Manage Paimon catalogs.

Limitations

Paimon tables are supported only in Ververica Runtime (VVR) 8.0.5 and later.

Write data to a Paimon table

Synchronize data and schema with CTAS/CDAS

For more information, see Manage Paimon catalogs.

Insert or update data with INSERT INTO

You can use an INSERT INTO statement to directly insert or update data in a Paimon table.

A Paimon primary key table can accept messages of all types, including INSERT, UPDATE_BEFORE, UPDATE_AFTER, and DELETE. It merges data with the same primary key on write, based on the data merge mechanism.
A Paimon append-only table, also known as a non-primary key table, accepts only INSERT-type messages.

Overwrite data with INSERT OVERWRITE

Overwriting clears existing data and writes new data. You can use an INSERT OVERWRITE statement to overwrite an entire Paimon table or a specific partition. The following code provides examples.

Note

The INSERT OVERWRITE statement is supported only in batch jobs.
By default, an INSERT OVERWRITE operation does not generate changelog data. Downstream streaming jobs cannot consume the deleted and imported data. If you need to consume this type of data, see Stream and consume the results of an INSERT OVERWRITE statement.

Overwrite the entire non-partitioned table my_table.
```
INSERT OVERWRITE my_table SELECT ...;
```

Overwrite the dt=20240108,hh=06 partition in the my_table table.

INSERT OVERWRITE my_table PARTITION (`dt` = '20240108', `hh` = '06') SELECT ...;

Dynamically overwrite partitions in the my_table table. Partitions that appear in the result of the SELECT statement are overwritten, while other partitions remain unchanged.
```
INSERT OVERWRITE my_table SELECT ...;
```

Overwrite the entire partitioned table my_table.

INSERT OVERWRITE my_table /*+ OPTIONS('dynamic-partition-overwrite' = 'false') */ SELECT ...;

Delete data by using DELETE

You can use a DELETE statement to delete data from a Paimon primary key table. You can execute DELETE statements only in Data Exploration.

-- Delete all data where currency = 'UNKNOWN' from the my_table table.
DELETE FROM my_table WHERE currency = 'UNKNOWN';

Filter delete messages

When you use a Paimon primary key table, DELETE messages remove data with the corresponding primary key by default. To prevent the Paimon table from processing these messages, use a SQL hint to set the following parameter to true. This filters delete messages.

Parameter	Description	Type	Default
ignore-delete	Specifies whether to filter delete messages.	Boolean	false

Adjust sink parallelism

You can use a SQL hint to set the following parameter to adjust the parallelism of the sink operator.

Parameter	Description	Type	Default
sink.parallelism	Sets the parallelism of the Paimon sink operator.	Integer	None

For example, the following SQL statement sets the parallelism of the Paimon sink operator to 10.

INSERT INTO t /*+ OPTIONS('sink.parallelism' = '10') */ SELECT * FROM s;

Consume data from a Paimon table

Streaming jobs

Note

For Paimon primary key tables consumed by streaming jobs, you must configure a changelog producer.

By default, a Paimon source operator in a streaming job first produces the table's full data when the job starts, and then the incremental data from that point onward.

Consume data from a specified offset

You can consume data from a specified offset in one of the following ways:

If you do not need to consume the full data in a Paimon table at job startup and only need to consume subsequent incremental data, you can set 'scan.mode' = 'latest' via a SQL hint.
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'latest') */;
```
If you do not want to consume the full data and want to consume only incremental data from a specific point in time, you can use a SQL Hint to set the scan.timestamp-millis parameter. The parameter value represents the number of milliseconds from the Unix Epoch (1970-01-01 00:00:00 UTC) to the specified point in time.
```
SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;
```
To consume data written after a specific time and then continuously consume subsequent incremental data, use one of the following methods.

Note
This consumption method reads data files modified after the specified time. Due to compaction, data files might contain a small amount of data written before the specified time. You can add a WHERE clause to your SQL job to filter the data as needed.
- Do not set any SQL hints. When starting the job, you can select Specify source's start time. In the Job Start dialog box, select Stateless start-up, enable the Specify source's start time switch, and set the target time.
- Set the scan.file-creation-time-millis parameter using an SQL hint.
```
SELECT * FROM t /*+ OPTIONS('scan.file-creation-time-millis' = '1678883047356') */;
```
If you do not want to consume the full data and only want to consume incremental data starting from a specific snapshot file, you can use a SQL hint to set the scan.snapshot-id parameter. The value of this parameter is the ID of the specified snapshot file.
```
SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;
```
If you want to consume the full data of a specific snapshot file and continue to consume incremental data, you can use a SQL hint to set the 'scan.mode' = 'from-snapshot-full' and scan.snapshot-id parameters. The value of the scan.snapshot-id parameter is the ID of the specified snapshot file.
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'from-snapshot-full', 'scan.snapshot-id' = '1') */;
```

Specify a consumer ID

A consumer ID saves the consumption progress for a Paimon table. This is mainly used in the following scenarios:

Setting a consumer ID saves its corresponding consumption progress to the Paimon table's metadata file. This allows the job to resume consumption from the point of interruption, even if it is later started in stateless mode.
After you set a consumer ID, snapshots that have not been consumed are not deleted due to expiration. This prevents errors caused when the consumption speed cannot keep up with the snapshot expiration speed.

You can set the consumer-id parameter to assign a Consumer ID to a Paimon source operator in a streaming job. The value of the Consumer ID can be any string. When a Consumer ID is created for the first time, its starting offset is determined according to the rules in Consume a Paimon Table from a Specified Offset. You can then resume consumption of the Paimon table by continuing to use the same Consumer ID.

For example, the following SQL statement sets a consumer ID named test-id for a Paimon source operator. If you want to reset the offset for a specific consumer ID, you can also set 'consumer.ignore-progress' = 'true'.

SELECT * FROM t /*+ OPTIONS('consumer-id' = 'test-id') */;

Note

Snapshot files that have not been consumed by a consumer ID are not deleted upon expiration. If obsolete consumer IDs are not cleaned up, the corresponding snapshot files and historical data files will never be deleted and will consume storage space. You can set the consumer.expiration-time table parameter to clean up consumer IDs that have been unused for a specified period. For example, 'consumer.expiration-time' = '3d' indicates that consumer IDs that are unused for 3 days will be cleaned up.

Stream and consume INSERT OVERWRITE results

By default, INSERT OVERWRITE operations do not generate changelog data, and the data that is deleted and imported cannot be consumed by downstream streaming jobs. If you need to consume such data, you can set 'streaming-read-overwrite' = 'true' in a streaming consumption job by using an SQL hint.

SELECT * FROM t /*+ OPTIONS('streaming-read-overwrite' = 'true') */;

Batch jobs

By default, a Paimon source operator in a batch job reads the latest snapshot and outputs the latest state data of the Paimon table.

Batch time travel

You can query the state of a Paimon table at a specific point in time by setting the scan.timestamp-millis parameter in a SQL hint. The parameter value represents the number of milliseconds from the Unix Epoch (1970-01-01 00:00:00 UTC) to the specified time.

SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;

You can query the state of a Paimon table at the time a snapshot was created by using a SQL Hint to set the scan.snapshot-id parameter to the ID of the specified snapshot.

SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;

Query changes between snapshots

If you want to query for data changes in a Paimon table between two snapshots, you can set the incremental-between parameter by using a SQL hint. For example, to view all data that has changed between snapshot 20 and snapshot 12, the SQL statement is as follows.

SELECT * FROM t /*+ OPTIONS('incremental-between' = '12,20') */;

Note

Because batch jobs do not support consuming DELETE messages, these messages are discarded by default. If you want to consume DELETE messages in a batch job, query the Audit Log system table. For example: SELECT * FROM `t$audit_log ` /*+ OPTIONS('incremental-between' = '12,20') */;.

Adjust source parallelism

By default, Paimon automatically infers the source operator parallelism based on information like the number of partitions and buckets. You can use a SQL hint to set the following parameters to adjust the parallelism.

Parameter	Type	Default	Description
scan.parallelism	Integer	None	Sets the parallelism of the Paimon source operator.
scan.infer-parallelism	Boolean	true	Specifies whether to automatically infer the parallelism of the Paimon source operator.
scan.infer-parallelism.max	Integer	1024	The upper limit of the automatically inferred parallelism of the Paimon source operator.

The following SQL statement provides an example of how to set the parallelism of the Paimon source operator to 10.

SELECT * FROM t /*+ OPTIONS('scan.parallelism' = '10') */;

Use Paimon tables as dimension tables

Paimon tables can also be used as dimension tables. For more information about the syntax for dimension table JOINs, see Dimension table JOIN statement.

Write and consume the VARIANT type

In Ververica Runtime (VVR) 11.1 and later, Paimon tables support the VARIANT semi-structured data type. This type lets you convert VARCHAR JSON strings to the VARIANT type using PARSE_JSON or TRY_PARSE_JSON. Writing and consuming the VARIANT type directly significantly improves JSON query and processing performance.

The following code provides an example:

CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  k BIGINT,
  info VARIANT
);
INSERT INTO `my-catalog`.`my_db`.`my_tbl` 
SELECT k, PARSE_JSON(jsonStr) FROM T;

Realtime Compute for Apache Flink:Write and consume data in Paimon tables

Prerequisites

Limitations

Write data to a Paimon table

Synchronize data and schema with CTAS/CDAS

Insert or update data with INSERT INTO

Overwrite data with INSERT OVERWRITE

Delete data by using DELETE

Filter delete messages

Adjust sink parallelism

Consume data from a Paimon table

Streaming jobs

Consume data from a specified offset

Specify a consumer ID

Stream and consume INSERT OVERWRITE results

Batch jobs

Batch time travel

Query changes between snapshots

Adjust source parallelism

Use Paimon tables as dimension tables

Write and consume the VARIANT type

Related documentation