SQL Query

Query a Paimon table with SELECT. Choose the runtime mode and starting point before tuning parallelism or filters.

Choose a Read Mode

Goal	Runtime mode	Starting point
Read the current table once	Batch	Latest snapshot by default.
Read an older table version	Batch	Snapshot, timestamp, watermark, or tag.
Process a bounded range of changes	Batch	Incremental range, with an exclusive start and inclusive end.
Load the table and follow changes	Streaming	Latest full snapshot, then subsequent changes by default.
Follow only new changes	Streaming	`scan.mode = latest`.
Resume from a selected version	Streaming	Streaming time travel; use `from-snapshot-full` to include its full contents.

Batch reads one snapshot and stops; streaming can read a full snapshot then follow changes, or follow new changes only.

For primary-key tables, choose a changelog producer that supplies the changes required by the downstream query. For append tables, see Streaming Reads.

Batch Query

Paimon's batch read returns all the data in a snapshot of the table. By default, batch reads return the latest snapshot.

-- Flink SQL
SET 'execution.runtime-mode' = 'batch';

Batch Time Travel

Select a retained snapshot by ID, timestamp, watermark, or tag. A batch time-travel query returns the table contents at that version; it does not return only the changes made in that snapshot.

Flink (dynamic option)
Flink 1.18+

-- read the snapshot with id 1L
SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '1') */;

-- read the snapshot from specified timestamp in unix milliseconds
SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;

-- read the snapshot from specified timestamp string ,it will be automatically converted to timestamp in unix milliseconds
-- Supported formats include：yyyy-MM-dd, yyyy-MM-dd HH:mm:ss, yyyy-MM-dd HH:mm:ss.SSS, use default local time zone
SELECT * FROM t /*+ OPTIONS('scan.timestamp' = '2023-12-09 23:09:12') */;

-- read tag 'my-tag'
SELECT * FROM t /*+ OPTIONS('scan.tag-name' = 'my-tag') */;

-- Read the first retained snapshot with watermark >= the requested value.
SELECT * FROM t /*+ OPTIONS('scan.watermark' = '1678883047356') */;

Flink 1.18+ supports SQL time-travel syntax.

-- read the snapshot from specified timestamp
SELECT * FROM t FOR SYSTEM_TIME AS OF TIMESTAMP '2023-01-01 00:00:00';

-- you can also use some simple expressions (see flink document to get supported functions)
SELECT * FROM t FOR SYSTEM_TIME AS OF TIMESTAMP '2023-01-01 00:00:00' + INTERVAL '1' DAY;

Batch Incremental

Read changes in (start, end]: the start boundary is excluded and the end boundary is included. You can specify snapshot IDs, tag names, or timestamps.

An incremental read from snapshot 12 to 15 excludes snapshot 12, includes changes from 13 through 15, and stops at 15.

SET 'execution.runtime-mode' = 'batch';

-- Changes committed after snapshot 12, through snapshot 15.
SELECT * FROM t /*+ OPTIONS('incremental-between' = '12,15') */;

-- The same boundary convention applies to tags.
SELECT * FROM t /*+ OPTIONS('incremental-between' = 'TAG1,TAG3') */;

-- Timestamp boundaries can be Unix milliseconds or timestamp strings.
SELECT * FROM t /*+ OPTIONS(
    'incremental-between-timestamp' = '1692169000000,1692169900000'
) */;
SELECT * FROM t /*+ OPTIONS(
    'incremental-between-timestamp' = '2025-03-12 00:00:00,2025-03-12 00:08:00'
) */;

By default, an incremental read uses changelog files when the table produces them; otherwise it reads newly changed files. Use incremental-between-scan-mode to select the scan mode explicitly. An incremental read is not a full table snapshot.

Batch SQL drops DELETE records from an ordinary table result. To inspect deletes as data, read the audit-log system table, which exposes the row kind:

SELECT * FROM `t$audit_log` /*+ OPTIONS('incremental-between' = '12,15') */;

Batch Incremental between Auto-created Tags

An automatically created tag may be missing when data is delayed. A direct incremental-between query fails if a named boundary tag does not exist.

Use incremental-to-auto-tag when only the end tag is known. Paimon finds the preceding tag and reads the changes between them. If either the requested tag or its predecessor is missing, the result is empty.

For example, suppose the existing daily tags are 2024-12-01, 2024-12-02, and 2024-12-04:

Requested end tag	Result
`2024-12-01`	Empty: there is no preceding tag.
`2024-12-02`	Changes between `2024-12-01` and `2024-12-02`.
`2024-12-03`	Empty: this tag does not exist.
`2024-12-04`	Changes between `2024-12-02` and `2024-12-04`.

SELECT * FROM t /*+ OPTIONS('incremental-to-auto-tag' = '2024-12-04') */;

Streaming Query

On a fresh start with no time-travel options or stored consumer progress, a streaming read loads the latest full snapshot and then follows changes. It is unbounded: the query continues running until it is canceled or otherwise stopped.

-- Flink SQL
SET 'execution.runtime-mode' = 'streaming';

You can also do streaming read without the snapshot data, you can use latest scan mode:

-- Continuously reads latest changes without producing a snapshot at the beginning.
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'latest') */;

Streaming Time Travel

Use a partition filter when the intended range is expressed by partition values:

SELECT * FROM t WHERE dt > '2023-06-26';

To select a position in commit history, use streaming time travel. scan.snapshot-id alone starts with changes from that snapshot; from-snapshot-full first loads its complete contents and then follows later changes. These examples apply to a fresh scan; stored consumer progress or restored Flink state can determine the start position instead.

Flink (dynamic option)
Flink 1.18+

-- read changes from snapshot id 1L
SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '1') */;

-- read changes from snapshot specified timestamp
SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;

-- read snapshot id 1L upon first startup, and continue to read the changes
SELECT * FROM t /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id' = '1') */;

Flink 1.18+ supports SQL time-travel syntax.

-- read the snapshot from specified timestamp
SELECT * FROM t FOR SYSTEM_TIME AS OF TIMESTAMP '2023-01-01 00:00:00';

-- you can also use some simple expressions (see flink document to get supported functions)
SELECT * FROM t FOR SYSTEM_TIME AS OF TIMESTAMP '2023-01-01 00:00:00' + INTERVAL '1' DAY;

Streaming time travel depends on retained snapshots or changelogs. Configure snapshot retention and use a Consumer ID when a reader needs to protect its progress from expiration.

For approximate filtering by data-file creation time, use scan.file-creation-time-millis. This filters files, not individual record timestamps, and is not a replacement for retaining the history needed for exact replay.

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

Read Overwrite

Streaming reads ignore INSERT OVERWRITE commits by default. For primary key tables, enable streaming-read-overwrite to read the changes. For append tables, use streaming-read-append-overwrite to read the added rows without retractions for replaced rows; see Append-table overwrite commits.

Read Parallelism

Set scan.parallelism to choose source parallelism explicitly:

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

Otherwise, Paimon infers source parallelism only when global parallelism is unset and scan.infer-parallelism is enabled. The inference depends on the read path:

Read path	Inference
Batch data table	Estimate planned work from partition data size and `split.target-size`; the estimate need not equal the actual split count.
Other batch sources	Use their planned splits.
Streaming, fixed buckets	Use the bucket count.
Streaming, `bucket = -1`	Fall back to Flink's configured or default parallelism.

Batch inference is capped by scan.infer-parallelism.max and can also be reduced by a query limit. That cap is not applied by the fixed-bucket streaming inference path.

Option	Default	Effect
`scan.parallelism`	Not set	Set the source parallelism explicitly.
`scan.infer-parallelism`	`true`	Enable inference when no explicit source or global parallelism is set.
`scan.infer-parallelism.max`	`1024`	Cap inferred batch parallelism.

See FlinkConnectorOptions for the full reference.

Query Optimization

Batch Streaming

It is highly recommended to specify partition and primary key filters along with the query, which will speed up the data skipping of the query.

The filter functions that can accelerate data skipping are:

=
<
<=
>
>=
IN (...)
LIKE 'abc%'
IS NULL

Primary-key tables normally sort rows by primary key, which helps point and range queries. For a composite primary key, filters on its leading columns can skip more data. Check the table's actual data layout and query performance guidance when choosing filters.

Suppose that a table has the following specification:

CREATE TABLE orders (
    catalog_id BIGINT,
    order_id BIGINT,
    amount DECIMAL(10, 2),
    PRIMARY KEY (catalog_id, order_id) NOT ENFORCED -- composite primary key
);

The query obtains a good acceleration by specifying a range filter for the leftmost prefix of the primary key.

SELECT * FROM orders WHERE catalog_id=1025;

SELECT * FROM orders WHERE catalog_id=1025 AND order_id=29495;

SELECT * FROM orders
  WHERE catalog_id=1025
  AND order_id>2035 AND order_id<6000;

However, the following filter cannot accelerate the query well.

SELECT * FROM orders WHERE order_id=29495;

SELECT * FROM orders WHERE catalog_id=1025 OR order_id=29495;

Dedicated Split Generation

When Paimon table snapshots contain large amount of source splits, Flink jobs reading from this table might endure long initialization time or even OOM in JobManagers. In this case, you can configure 'scan.dedicated-split-generation' = 'true' to avoid such problem. This option would enable executing the source split generation process in a dedicated subtask that runs on TaskManager, instead of in the source coordinator on the JobManager.

Note that this feature could have some side effects on your Flink jobs. For example:

It will change the DAG of the flink job, thus breaking checkpoint compatibility if enabled on an existing job.
It may lead to the Flink AdaptiveBatchScheduler inferring a small parallelism for the source reader operator. you can configure scan.infer-parallelism to avoid this possible drawback.
The failover strategy of the Flink job would be forced into global failover instead of regional failover, given that the dedicated source split generation task would be connected to all downstream subtasks.

So please make sure these side effects are acceptable to you before enabling it.

Choose a Read Mode​

Batch Query​

Batch Time Travel​

Batch Incremental​

Batch Incremental between Auto-created Tags​

Streaming Query​

Streaming Time Travel​

Read Overwrite​

Read Parallelism​

Query Optimization​

Dedicated Split Generation​