Realtime Compute for Apache Flink:Paimon connector

SQL

Use the Paimon connector in SQL jobs as a source table or a result table.

Data freshness and consistency

The Paimon sink table uses the two-phase commit protocol to commit data during each Flink job checkpoint. Therefore, data freshness is determined by the Flink job's checkpoint interval. Each commit produces up to two snapshots.

When two Flink jobs write to the same Paimon table concurrently, if the jobs write to different buckets, they achieve serializable consistency. If the jobs write to the same bucket, they only achieve snapshot isolation. This means the table's data might be a mix of results from both jobs, but no data loss occurs.

Merge engine

When a Paimon sink table receives multiple records with the same primary key, it merges them into a single record to maintain uniqueness. You can control this behavior by setting the merge-engine parameter. The following table describes the available merge engines.

CREATE TABLE MyTable (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

Changelog producer

Set the changelog-producer parameter to configure Paimon to generate a complete changelog (where every update_after record has a corresponding update_before record) for any input stream. The following table describes the available changelog producers. For more details, see the Apache Paimon official documentation.

Write mode

Paimon tables support the following write modes.

Target for Flink CDC data ingestion

Paimon tables support real-time synchronization of data from single tables or entire databases. Upstream schema changes are synchronized to Paimon tables in real time. For more information, see the Data ingestion section.

Variant read pruning

Reads only the Variant fields referenced by the query, reducing I/O and memory overhead.

How to enable

Prerequisites

• Variant data can only be written by VVR 11.6 or later. Variant data written by earlier versions or other engines does not support read pruning.

• For upstream data to take effect, you must enable the variant.inferShreddingSchema option to enable automatic Variant shredding inference, or explicitly configure variant.shreddingSchema and related options to specify which Variant members are suitable for shredding. For more information, see Coreoptions.

Supported scenarios

Read pruning takes effect when SQL accesses Variant fields by string key, for example:

SELECT v['a']            FROM t;   -- Single level
SELECT v['a']['b']       FROM t;   -- Nested
SELECT v['a'], v['b']    FROM t;   -- Multiple fields
SELECT id, v['a']        FROM t;   -- Mixed with regular columns
SELECT v['a'] + 1        FROM t;   -- Referenced in expressions

Unsupported scenarios

• Referencing the entire Variant directly, such as SELECT v FROM t.

• Combining direct reference with field access, such as SELECT v, v['a'] FROM t.

• Array index access, such as v[0].

• The table schema contains nested Row fields. In this case, read pruning does not apply to any Variant field in the table.

Blob presigned URLs

Paimon can generate short-lived public HTTPS GET presigned URLs for Blob column data stored in OSS. External services, such as multimodal model services, can use the URLs to retrieve the file content directly, without transferring the file bytes inside a Flink job.

Limits

• Only Flink compute engine VVR 11.9-preview1 and later support this feature.

• Only Paimon tables stored in OSS are supported. The metastore type does not matter. Filesystem, DLF, and other metastore types are all supported.

• The generated URL is a public HTTPS URL and must be consumed by an external service that has public network access. If the catalog uses an internal endpoint, we recommend that you change it to a public HTTPS endpoint so that the external service can access the generated URL.

• To read a regular Blob column, you must first obtain the BlobDescriptor bytes by using the blob-as-descriptor option, and then pass the bytes to the function.

Syntax

descriptor_to_presigned_url(source_table, descriptor, validity)
try_descriptor_to_presigned_url(source_table, descriptor, validity)

Parameters

Return value

Example

-- Use an SQL hint to read the Blob column as a descriptor and generate a presigned URL
SELECT sys.descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

-- Fault-tolerant mode: returns NULL on row-level errors, with other behavior unchanged
SELECT sys.try_descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

Chain table

A chain table reconstructs a full data set from a snapshot branch and the changes that follow it in a delta branch. Write periodic full data to the snapshot branch and the changes between two full loads to the delta branch. A read uses the latest snapshot partition as an anchor and merges later delta partitions to rebuild the full data, so you no longer need to rewrite the entire table. This is useful when only a small fraction of rows changes between runs, such as an ODS binlog dump. For more information, see Chain Table in the Apache Paimon documentation. Chain tables are supported only in VVR 11.9 and later.

Enable a chain table

Set chain-table.enabled to true when you create the table, create the snapshot and delta branches, and then configure the branch fallback options on the primary table and on both branch tables.

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `date` STRING,
  PRIMARY KEY (`date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd'
);

-- mycat is the name of the Paimon catalog.
CALL `mycat`.sys.create_branch('default.t', 'snapshot');
CALL `mycat`.sys.create_branch('default.t', 'delta');

ALTER TABLE `default`.`t` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_snapshot` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_delta` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

Write data

Write full data to t$branch_snapshot. Each partition of this branch is a complete view at a point in time. Write incremental data to t$branch_delta. Each partition of this branch contains only the changes of that period.

-- Full write
INSERT OVERWRITE `default`.`t$branch_snapshot` PARTITION (date = '20250810') VALUES ('1', '1', '1');

-- Incremental write
INSERT INTO `default`.`t$branch_delta` PARTITION (date = '20250811') VALUES ('2', '1', '1');

Read data

-- Full query
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811';

-- Incremental query
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

-- Hybrid query
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811'
UNION ALL
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

Streaming read

A streaming read job runs in two phases:

• Initial load phase: by default, the job reads the latest snapshot partition of each group and the delta partitions that come after it. Older snapshot partitions are excluded. To merge snapshot and delta data by anchor in this phase and resolve cross-branch DELETE records, set chain-table.streaming.merge-snapshot to true. The trade-off is a heavier scan at job startup.

• Incremental phase: the job continuously reads new commits from the delta branch.

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

INSERT INTO downstream_sink SELECT * FROM `default`.`t`;

Multiple partition keys

If the table has more than one partition key, use chain-table.chain-partition-keys to separate the chain dimension from the group dimension. The value must be a contiguous suffix of the partition keys. The suffix fields form the time-ordered chain, and the partition fields before them become the group dimension. Each group of group partition values maintains its own chain and anchor. If this option is not set, all partitions belong to a single implicit group.

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `region` STRING,
  `date` STRING,
  PRIMARY KEY (`region`, `date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`region`, `date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd',
  -- Only date forms the chain. Each region has its own chain.
  'chain-table.chain-partition-keys' = 'date'
);

Limitations

• Chain tables are supported only for primary key tables. You must specify bucket and bucket-key.

• The schema of each branch must be consistent.

• A chain table requires sequence.field. The first-row and aggregation merge engines are not supported.

• Streaming read and lookup join require the delta branch to use the deduplicate merge engine (default value). Batch read is not affected.

• Deletion vectors are supported only for chain tables that use the deduplicate merge engine.

• Only none (default value) and input are supported for changelog-producer. The lookup and full-compaction modes are not supported. If the value is none, records are normalized by the full primary key that includes the chain partition, and -D and -U records whose original records reside in another partition are dropped. Set the value to input if downstream consumers must receive cross-partition changelog records.

• Streaming read and lookup join support only the default startup mode (latest-full). An error is thrown if you specify a starting position such as scan.snapshot-id, scan.timestamp-millis, scan.mode set to latest, or consumer-id. To perform a standard streaming read without chain table logic, read a specific branch table such as t$branch_delta.

• Partition filters are not supported in chain table streaming reads and lookup joins. This includes partition columns in the WHERE clause and the scan.partitions option, because a filter interferes with the cross-branch merge logic. Use batch mode to read a specific partition.

• The join key of a lookup join must not contain partition keys. Otherwise, an error is returned.

• partition.timestamp-formatter must specify a complete date (year-month-day or year-day-of-year), use only round-trippable date and time fields (y/u, M/L, d, D, H, k, m, and s), and have a minimum step of at least one second.

Mosaic file format

Mosaic is a columnar-bucket hybrid format optimized for wide tables that contain thousands or tens of thousands of columns. Columns are sorted by name and distributed into buckets by range, stored in column-oriented format within each bucket, and compressed independently by using ZSTD. Projection pushdown works at bucket granularity, so a query reads only the buckets that contain the requested columns. Columns with similar name prefixes land in the same bucket, which improves both the compression ratio and the locality of projections. For more information, see Mosaic. The Mosaic format is supported only in VVR 11.9 and later.

Enable the format

Set file.format to mosaic when you create the table. In a data ingestion job, the corresponding option is table.properties.file.format.

CREATE TABLE `default`.`wide_table` (
  `id` BIGINT,
  `col_a` STRING,
  `col_b` DOUBLE
) WITH (
  'file.format' = 'mosaic'
);

Format options

Limitations

Mosaic does not support complex types: ARRAY, MAP, MULTISET, ROW, VARIANT, BLOB, and VECTOR.

Use Paimon as a dimension table

Paimon tables can be used as dimension tables. For the JOIN syntax, see Dimension table JOIN statements.

By default, lookup loads all data in each parallel instance. This approach is suitable only for small dimension tables. For large dimension tables, use the Shuffle Lookup solutions described below.

Partitioned dimension tables

If your dimension table is partitioned and you only need data from the latest one or two partitions, you can use the dynamic partition loading feature:

SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;

Large dimension tables: fixed-bucket tables

Supported only in VVR 8.0.8 and later. For fixed-bucket tables (bucket > 0), you can use Shuffle Lookup to distribute data by bucket key across parallel instances, so each instance loads only the data in its assigned buckets:

SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Large dimension tables: non-fixed-bucket tables

Supported only in VVR 8.0.10 and later. For dynamic-bucket tables or append tables, you can use SHUFFLE_HASH or REPLICATED_SHUFFLE_HASH so that each parallel instance reads all data but retains only the portion it needs:

-- Shuffle Hash
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

-- Replicated Shuffle Hash
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

For more information about SHUFFLE_HASH and REPLICATED_SHUFFLE_HASH, see Dimension table JOIN statements.

Data ingestion

You can use the Paimon connector as a sink in data ingestion YAML jobs.

Syntax

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

Parameters

Reuse an existing catalog

Starting with Realtime Compute for Apache Flink 11.5, you can directly reference a built-in Paimon catalog from the Data Management page in a Flink CDC data ingestion job. This reduces manual configuration.

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

Data ingestion jobs can automatically reuse all parameters in a Paimon catalog. This is equivalent to manually configuring parameters prefixed with catalog.properties. in a YAML job.

To override an automatically reused parameter, explicitly set it in the YAML job. The explicit YAML configuration has a higher priority. For example, in the preceding sample, the fs.oss.endpoint parameter uses the value from the YAML job, overriding the one in paimon_dlf_catalog.

Schema Changes

A CDC YAML pipeline job can use different strategies to handle schema changes. To configure the strategy, set the pipeline-level parameter schema.change.behavior as described in Schema evolution configurations. schema.change.behavior can be set to IGNORE, LENIENT, TRY_EVOLVE, EVOLVE, or EXCEPTION. The following sections describe how the LENIENT and EVOLVE modes, which involve schema changes, handle different schema change events.

Examples

The following examples show configurations in typical scenarios.

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
route:
  # Synchronize all tables in the MySQL test_db database to the Paimon test_db2 database and retain the table names.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
route:
  # All sharded tables in the MySQL test_db database are merged into a single Paimon test_db.user table.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # On job restart, check the new tables captured by the tables parameter and take a snapshot.
  # Note: Must be used together with scan.startup.mode: initial.
  scan.newly-added-table.enabled: true
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Tables that match this regular expression are not synchronized.
  tables.exclude: test_db.table1
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  # Specify the file format of output files as lance 
  table.properties.file.format: lance
pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline
  schema.change.behavior: evolve