PolarDB-X CDC (Public preview)

Learn how to use the PolarDB-X connector.

Background

PolarDB for Xscale (PolarDB-X) is Alibaba Cloud's high-performance, cloud native distributed database, offering high throughput, large storage capacity, low latency, scalability, and ultra-high availability.

Important

This connector requires VVR 11.5 or later and PolarDB-X 2.0 or later.

The PolarDB-X CDC connector supports only source tables. To query a PolarDB-X instance as a dimension table or write to it as a sink table, use the MySQL connector (public preview).

Category	Description
Supported types	Source table, data ingestion source
Running mode	streaming mode only
Data format	N/A
Specific monitoring metrics	`currentFetchEventTimeLag`: The time between data generation and when the Source Operator fetches it. This metric applies only to the binlog phase and is 0 during the snapshot phase. `currentEmitEventTimeLag`: The time between data generation and when the Source Operator emits it. This metric applies only to the binlog phase and is 0 during the snapshot phase. `sourceIdleTime`: The time elapsed since the source table last produced data.
API type	SQL, data ingestion YAML jobs
Supports updating or deleting data in the sink table	No

Key features

The PolarDB-X CDC connector optimizes Binlog parsing with server-side filtering. It trims irrelevant Binlog data on the PolarDB-X server, which increases throughput and conserves network bandwidth.

On-demand Binlog subscription

With server-side Binlog filtering, the connector sends only required change logs to the client. This reduces network traffic and improves data consumption throughput.

For example, to subscribe only to changes from the db.table1 and db.table2 tables in your PolarDB-X server, configure your Flink SQL job as follows:

CREATE TABLE polardbx_table_foo (
  ... -- Define the table schema here
) WITH (
  'connector' = 'polardbx-cdc',
  'database-name' = 'db',
  'table-name' = '.*',
  ..., -- Other parameters
  'polardbx.binlog.include.tables' = 'db.table1,db.table2' -- Captures changes from only the specified tables
);

Unlike the MySQL CDC connector, which loads an entire instance's Binlog for client-side filtering, the PolarDB-X CDC connector performs server-side filtering. This capability enables on-demand client subscriptions, significantly reducing network I/O overhead.

Usage limits

Server-side Binlog filtering and table-level subscription require PolarDB-X server version 2.5.0 or later and Log Service component version 5.4.20 or later.

SQL

Syntax

CREATE TABLE polardbx_customer_table(
  `id` STRING,
  [columnName dataType,]*
  PRIMARY KEY(`id`) NOT ENFORCED
) WITH (
  'connector' = 'polardbx-cdc',
  'hosts' = 'pxc-**************-pub.polarx.rds.aliyuncs.com',
  'username' = 'pdx_user',
  'password' = 'pdx_password',
  'database' = 'full_db',
  'collection' = 'customers'
)

WITH parameters

Parameter	Description	Type	Required	Default	Remarks
connector	The name of the connector.	STRING	Yes	None	The value must be `polardbx-cdc`.
hostname	The IP address or hostname of the PolarDB-X database.	STRING	Yes	None	Use the cluster address from the instance connection information.
port	The port number of the PolarDB-X database.	INTEGER	No	3306	None
username	The username for the PolarDB-X database.	STRING	Yes	None	None
password	The password for the PolarDB-X database.	STRING	Yes	None	None
database-name	The name of the PolarDB-X database.	STRING	Yes	None	You can use a regular expression to read data from multiple databases. Note When using a regular expression, do not use the ^ and $ characters to match the start and end of the string.
table-name	The name of the PolarDB-X table.	STRING	Yes	None	You can use a regular expression to read data from multiple tables. Note When using a regular expression, do not use the ^ and $ characters to match the start and end of the string.
server-time-zone	The session time zone for the database connection.	STRING	No	The job's runtime environment time zone.	Specify an IANA time zone identifier, such as `Asia/Shanghai`. This parameter controls how `TIMESTAMP` types from the source table are converted to `STRING`.
scan.incremental.snapshot.chunk.size	The number of rows in each chunk for incremental snapshot reads.	INTEGER	No	8096	During the snapshot phase, the connector splits a table into chunks and caches them in memory. A smaller chunk size increases the total number of chunks. While this improves fault tolerance granularity, it also increases the risk of out of memory (OOM) errors and reduces throughput. Set an appropriate value to balance performance, fault tolerance, and memory consumption.
scan.snapshot.fetch.size	The maximum number of rows to fetch at a time when reading a snapshot of the table.	INTEGER	No	1024	None
connect.timeout	The timeout for establishing a connection to the PolarDB-X database.	DURATION	No	30s	None
connection.pool.size	The size of the database connection pool.	INTEGER	No	20	A database connection pool reuses connections to reduce connection overhead.
connect.max-retries	The maximum number of retries after a failed connection attempt to the PolarDB-X database.	INTEGER	No	3	None
scan.startup.mode	The startup mode for data consumption.	STRING	No	initial	Valid values: initial (default): On its first startup, the connector takes a full snapshot of the tables and then reads incremental changes from the Binlog. latest-offset: Does not perform a snapshot. On first startup, the connector starts reading from the end of the Binlog, capturing only changes made after the job starts. earliest-offset: Never performs a snapshot. Starts reading from the earliest available Binlog offset. specific-offset: Never performs a snapshot. Starts reading from a specific offset. Specify the offset by using either scan.startup.specific-offset.file and scan.startup.specific-offset.pos, or by using scan.startup.specific-offset.gtid-set. timestamp: Never performs a snapshot. Starts reading the Binlog from a specified timestamp. The timestamp is specified in milliseconds using scan.startup.timestamp-millis. Important For the earliest-offset, specific-offset, and timestamp startup modes, the table schema at startup must match the schema at the specified offset. A schema mismatch will cause the job to fail. Ensure the table schema does not change between the specified Binlog offset and the job startup.
scan.startup.specific-offset.file	The name of the Binlog file to start from when `scan.startup.mode` is `specific-offset`.	STRING	No	None	This parameter is only used when scan.startup.mode is set to specific-offset. Example: `mysql-bin.000003`.
scan.startup.specific-offset.pos	The offset within the specified Binlog file when `scan.startup.mode` is set to `specific-offset`.	INTEGER	No	None	This parameter is only used when scan.startup.mode is set to specific-offset.
scan.startup.specific-offset.gtid-set	The GTID set to start from when `scan.startup.mode` is set to `specific-offset`.	STRING	No	None	This parameter is only used when scan.startup.mode is set to specific-offset. Example: `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.
scan.startup.timestamp-millis	The startup timestamp in milliseconds when `scan.startup.mode` is set to `timestamp`.	LONG	No	None	This parameter is only used when scan.startup.mode is set to timestamp. The unit is milliseconds.
scan.startup.specific-offset.skip-events	The number of Binlog events to skip when reading from a specific offset.	INTEGER	No	None	This parameter is only used when scan.startup.mode is set to specific-offset.
scan.startup.specific-offset.skip-rows	The number of row changes to skip when reading from a specific offset. A single Binlog event can contain multiple row changes.	INTEGER	No	None	This parameter is only used when scan.startup.mode is set to specific-offset.
heartbeat.interval	The interval at which the source sends a heartbeat event to advance the Binlog offset.	DURATION	No	None	A heartbeat event advances the source's Binlog offset. This prevents the Binlog from being purged on an idle source. If the Binlog is purged, the job fails and requires a stateless restart to recover.
chunk-meta.group.size	The size of the chunk metadata.	INTEGER	No	1000	If the metadata size exceeds this value, it is split into multiple parts for transmission.
chunk-key.even-distribution.factor.upper-bound	The upper bound of the chunk distribution factor for even partitioning.	DOUBLE	No	1000.0	If the distribution factor is greater than this value, uneven chunking is used. Chunk distribution factor = (MAX(chunk-key) - MIN(chunk-key) + 1) / total number of rows.
chunk-key.even-distribution.factor.lower-bound	The lower bound of the chunk distribution factor for even partitioning.	DOUBLE	No	0.05	If the distribution factor is less than this value, uneven chunking is used. Chunk distribution factor = (MAX(chunk-key) - MIN(chunk-key) + 1) / total number of rows.
scan.newly-added-table.enabled	Specifies whether to scan for newly added tables that match the capture criteria when restarting from a Checkpoint.	BOOLEAN	No	false	When enabled, a job restarted from a Checkpoint or Savepoint scans for and synchronizes newly added tables that match the 'table-name' pattern. It also stops tracking tables that no longer match.
scan.incremental.snapshot.chunk.key-column	Specifies the column used for data chunking during the snapshot phase.	STRING	See Remarks	None	Required for tables without a primary key. The specified column must be non-nullable. Optional for tables with a primary key. Only one column from the primary key is supported.
scan.incremental.close-idle-reader.enabled	Specifies whether to close idle readers after the snapshot phase is complete.	BOOLEAN	No	false	For this setting to take effect, `execution.checkpointing.checkpoints-after-tasks-finish.enabled` must also be set to `true`.
scan.incremental.snapshot.backfill.skip	Specifies whether to skip the backfill process during the snapshot reading phase.	BOOLEAN	No	false	Valid values: true: Skips the backfill process during the snapshot reading phase. false (default): Does not skip the backfill process during the snapshot reading phase. If backfill is skipped, changes made to the table during the snapshot phase are read in the later incremental phase instead of being merged into the snapshot. Important Skipping backfill can cause data inconsistencies because changes that occur during the snapshot are replayed in the incremental phase. This mode only guarantees at-least-once semantics.
scan.parse.online.schema.changes.enabled	Specifies whether to attempt to parse DDL events for RDS lock-free changes during the incremental phase.	BOOLEAN	No	false	Valid values: true: Parses DDL events for RDS lock-free changes. false (default): Does not parse DDL events for RDS lock-free changes. This is an experimental feature. Before performing lock-free schema changes in production, create a Savepoint for the Flink job to ensure it can be recovered.
scan.only.deserialize.captured.tables.changelog.enabled	Specifies whether to deserialize change events only for the specified captured tables during the incremental phase.	BOOLEAN	No	true	Valid values: true (default): Deserializes change data only for the target tables, which can speed up Binlog reading. false: Deserializes change data for all tables.
scan.read-changelog-as-append-only.enabled	Specifies whether to convert the changelog stream to an append-only stream.	BOOLEAN	No	false	Valid values: true: Converts all message types (including `INSERT`, `DELETE`, `UPDATE_BEFORE`, and `UPDATE_AFTER`) to `INSERT` messages. Use this option only in specific cases, for example, to retain a record of deleted rows from the source. false (default): All message types are sent downstream as they are.
scan.parallel-deserialize-changelog.enabled	Specifies whether to use multiple threads to deserialize change events during the incremental phase.	BOOLEAN	No	false	Valid values: true: Uses multiple threads for deserialization to speed up reading while preserving the order of Binlog events. false (default): Uses a single thread for deserialization.
scan.parallel-deserialize-changelog.handler.size	The number of event handlers to use when deserializing change events in parallel.	INTEGER	No	2	None
scan.incremental.snapshot.unbounded-chunk-first.enabled	Specifies whether to distribute unbounded chunks first during the snapshot reading phase.	BOOLEAN	No	false	Valid values: true: Distributes unbounded chunks first during the snapshot reading phase. false (default): Does not distribute unbounded chunks first during the snapshot reading phase. This is an experimental feature. Enabling it can reduce the risk of an out of memory (OOM) error on the TaskManager when synchronizing the last chunk of a snapshot. This setting should be configured before the job's first startup.
polardbx.binlog.ignore.archive-events.enabled	Specifies whether to ignore archive events (primarily `DELETE` events) in the PolarDB-X Binlog.	BOOLEAN	No	false
polardbx.binlog.ignore.query-events.enabled	Specifies whether to ignore `ROWS_QUERY_LOG_EVENT` events in the PolarDB-X Binlog.	BOOLEAN	No	false
polardbx.binlog.include.tables	Reads Binlog events only from the specified tables. Use commas (`,`) to separate multiple table names.	STRING	No	None
polardbx.binlog.exclude.tables	Ignores Binlog events from the specified tables. Use commas (`,`) to separate multiple table names.	STRING	No	None

Type mapping

PolarDB-X type	Flink type
TINYINT	TINYINT
SMALLINT	SMALLINT
TINYINT UNSIGNED
TINYINT UNSIGNED ZEROFILL
INT	INT
MEDIUMINT
SMALLINT UNSIGNED
SMALLINT UNSIGNED ZEROFILL
BIGINT	BIGINT
INT UNSIGNED
INT UNSIGNED ZEROFILL
MEDIUMINT UNSIGNED
MEDIUMINT UNSIGNED ZEROFILL
BIGINT UNSIGNED	DECIMAL(20, 0)
BIGINT UNSIGNED ZEROFILL
SERIAL
FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
REAL [UNSIGNED] [ZEROFILL]
NUMERIC(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
DECIMAL(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
BOOLEAN	BOOLEAN
TINYINT(1)	BOOLEAN
DATE	DATE
TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
TIMESTAMP [(p)]	TIMESTAMP_LTZ [(p)]
TIMESTAMP [(p)]	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
CHAR(n)	STRING
VARCHAR(n)
TEXT
BINARY	BYTES
VARBINARY
BLOB

Data ingestion

Realtime Compute for Apache Flink 11.6 now supports the PolarDB-X connector as a data source for data ingestion YAML jobs.

Syntax

source:
   type: polardbx
   name: PolarDB-X Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: pdb.order_table
   # Ignore archive events in the binlog
   polardbx.binlog.ignore.archive-events.enabled: true
   # Ignore query events in the binlog
   polardbx.binlog.ignore.query-events.enabled: true
   # Subscribe only to the binlog for pdb.order_table to save bandwidth.
   polardbx.binlog.include.tables: pdb.order_table

sink:
  type: values

Parameter

Parameter	Description	Required	Type	Default	Remarks
type	The type of the data source.	Yes	STRING	N/A	The value must be `polardbx`.
name	The name of the data source.	No	STRING	N/A	N/A
hostname	The IP address or hostname of the PolarDB-X instance.	Yes	STRING	N/A	The VPC address is recommended. Note If your PolarDB-X instance and Realtime Compute for Apache Flink workspace are not in the same VPC, you must establish cross-VPC connectivity or access the instance over the public network. For details, see Workspace Management and Operations and How can a fully-managed Flink cluster access the public network?.
username	The username to connect to the PolarDB-X database.	Yes	STRING	N/A	N/A
password	The password for the specified username.	Yes	STRING	N/A	N/A
tables	The PolarDB-X tables to synchronize.	Yes	STRING	N/A	Supports regular expressions to read data from multiple tables. You can separate multiple regular expressions with a comma (,). Note Do not use the start and end anchors `^` and `$` in regular expressions. A period (.) separates the database name and table name. To match a literal period, you must escape it with a backslash. For example: `db0.\.`, `db1.user_table_[0-9]+`, or `db[1-2].[app\|web]order_\.`.
tables.exclude	The tables to exclude from synchronization.	No	STRING	N/A	Supports regular expressions to exclude multiple tables. You can separate multiple regular expressions with a comma (,). Note A period (.) separates the database name and table name. To match a literal period, you must escape it with a backslash. For example: `db0.\.`, `db1.user_table_[0-9]+`, or `db[1-2].[app\|web]order_\.`.
port	The port of the PolarDB-X instance.	No	INTEGER	3306	N/A
schema-change.enabled	Specifies whether to emit schema change events.	No	BOOLEAN	true	N/A
jdbc.properties.*	Custom connection parameters for the JDBC URL.	No	STRING	N/A	You can pass custom connection parameters. For example, to disable SSL, set 'jdbc.properties.useSSL' = 'false'.
debezium.*	Custom Debezium parameters for reading the binlog.	No	STRING	N/A	You can pass custom Debezium parameters. For example, set 'debezium.event.deserialization.failure.handling.mode' = 'ignore' to define how the connector handles deserialization errors.
scan.incremental.snapshot.chunk.size	The size of each chunk, in number of rows.	No	INTEGER	8096	A PolarDB-X table is split into multiple chunks for reading. The data in each chunk is cached in memory until it is fully read. A smaller chunk size increases the total number of chunks, which provides finer-grained fault recovery but can increase the risk of OOM and reduce overall throughput. You must balance these trade-offs and configure a reasonable chunk size.
scan.snapshot.fetch.size	The maximum number of records fetched at once during a full table snapshot.	No	INTEGER	1024	N/A
scan.startup.mode	The startup mode for consuming data.	No	STRING	initial	Valid values: initial (Default): Takes an initial snapshot of the table and then reads the latest changes from the binlog. latest-offset: Skips the snapshot phase and reads the binlog from the latest position, capturing only changes made after the connector starts. earliest-offset: Skips the snapshot phase and starts reading from the earliest available binlog position. specific-offset: Skips the snapshot phase and starts reading from a specific position. Specify the position by using either `scan.startup.specific-offset.file` and `scan.startup.specific-offset.pos`, or by using `scan.startup.specific-offset.gtid-set`. timestamp: Skips the snapshot phase and starts reading the binlog from a specific timestamp, which is specified in milliseconds by the scan.startup.timestamp-millis parameter. Important For the earliest-offset, specific-offset, and timestamp startup modes, the job fails with an error if the table schema at the startup time is different from the table schema at the time of the specified startup position. In other words, when you use these three startup modes, you must ensure that the schema of the corresponding table does not change between the specified Binlog consumption position and the job startup time.
scan.startup.specific-offset.file	The binlog filename for the startup position.	No	STRING	N/A	Required when scan.startup.mode is set to specific-offset. Example format: `mysql-bin.000003`.
scan.startup.specific-offset.pos	The offset within the specified binlog file to start from.	No	INTEGER	N/A	Required when scan.startup.mode is set to specific-offset.
scan.startup.specific-offset.gtid-set	The GTID set for the startup position.	No	STRING	N/A	Used when scan.startup.mode is set to specific-offset. Example GTID set format: `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.
scan.startup.timestamp-millis	The timestamp in milliseconds for the startup position.	No	LONG	N/A	Required when scan.startup.mode is set to timestamp. Important When you specify a timestamp, the connector locates the correct starting file by reading the first event of each binlog file to check its timestamp. Ensure the target binlog file for the specified timestamp has not been purged from the database and remains readable.
server-time-zone	The session time zone used by the database.	No	STRING	The time zone of the Flink job's runtime environment (the availability zone of your workspace).	Example: Asia/Shanghai. This parameter controls how `TIMESTAMP` types in PolarDB-X are converted to `STRING`. For more information, see Debezium documentation on time types.
scan.startup.specific-offset.skip-events	The number of binlog events to skip when reading from a specific position.	No	INTEGER	N/A	Used when scan.startup.mode is set to specific-offset.
scan.startup.specific-offset.skip-rows	The number of row changes to skip when reading from a specific position. A single binlog event can contain multiple row changes.	No	INTEGER	N/A	Used when scan.startup.mode is set to specific-offset.
connect.timeout	The maximum time to wait for a database connection before retrying.	No	DURATION	30s	N/A
connect.max-retries	The maximum number of retries for a failed database connection.	No	INTEGER	3	N/A
connection.pool.size	The size of the database connection pool.	No	INTEGER	20	The connection pool reuses connections to reduce connection overhead.
heartbeat.interval	The interval at which the source sends a heartbeat event to advance the binlog position.	No	DURATION	30s	Heartbeat events advance the binlog position for tables that are updated infrequently. Without them, the binlog position might not advance, which can lead to binlog expiration. An expired binlog causes the job to fail, requiring a stateless restart to recover.
scan.incremental.snapshot.chunk.key-column	Specifies a column to use as the splitting key for chunks during the snapshot phase.	No	STRING	N/A	You can only select one column from the primary key.
chunk-meta.group.size	The size of the chunk metadata.	No	INTEGER	1000	If the metadata size exceeds this value, it is split into multiple parts for transmission.
chunk-key.even-distribution.factor.lower-bound	The lower bound of the chunk distribution factor for determining whether to use even chunking.	No	DOUBLE	0.05	If the distribution factor is less than this value, uneven chunking is used. Chunk distribution factor = (MAX(chunk-key) - MIN(chunk-key) + 1) / total number of data rows.
chunk-key.even-distribution.factor.upper-bound	The upper bound of the chunk distribution factor for determining whether to use even chunking.	No	DOUBLE	1000.0	If the distribution factor is greater than this value, uneven chunking is used. Chunk distribution factor = (MAX(chunk-key) - MIN(chunk-key) + 1) / total number of data rows.
scan.incremental.close-idle-reader.enabled	Specifies whether to close idle readers after the snapshot phase is complete.	No	BOOLEAN	false	For this setting to take effect, you must set `execution.checkpointing.checkpoints-after-tasks-finish.enabled` to `true`.
scan.only.deserialize.captured.tables.changelog.enabled	Specifies whether to deserialize change events only for the specified tables during the incremental phase.	No	BOOLEAN	true	Valid values: true: Deserializes change data only for the target tables, which speeds up binlog reading. false: Deserializes change data for all tables.
scan.parallel-deserialize-changelog.enabled	Specifies whether to use multiple threads to parse change events during the incremental phase.	No	BOOLEAN	false	Valid values: true: Uses multiple threads for change event deserialization while preserving binlog event order, which speeds up reading. false (Default): Uses a single thread for event deserialization.
scan.parallel-deserialize-changelog.handler.size	The number of event handlers to use when parsing change events in parallel.	No	INTEGER	2	N/A
metadata-column.include-list	The metadata columns to pass to the downstream sink.	No	STRING	N/A	Available metadata includes `op_ts`, `es_ts`, `query_log`, `file`, and `pos`. You can use commas to separate multiple metadata columns. Note The PolarDB-X CDC YAML connector does not require or support adding metadata columns for the database name, table name, or `op_type`. You can use `__data_event_type__` in a transform expression to get the change data type, or use `__schema_name__` and `__table_name__` to get the database and table names. Important The `file` column contains the binlog filename (incremental phase) or an empty string (snapshot phase). The `pos` column contains the binlog offset (incremental phase) or `0` (snapshot phase).
scan.newly-added-table.enabled	When restarting from a checkpoint or savepoint, specifies whether to synchronize newly added tables that match the tables pattern and stop tracking tables that no longer match.	No	BOOLEAN	false	This setting takes effect only when restarting from a checkpoint or savepoint.
scan.binlog.newly-added-table.enabled	During the incremental phase, specifies whether to send data for newly matched tables.	No	BOOLEAN	false	Cannot be enabled at the same time as `scan.newly-added-table.enabled`.
scan.incremental.snapshot.chunk.key-column	Specifies a splitting key column for chunking during the snapshot phase for specific tables.	No	STRING	N/A	Use a colon (`:`) to connect a table name and a column name to define a rule. Table names can be regular expressions. You can define multiple rules separated by semicolons (`;`). For example: `db1.user_table_[0-9]+:col1;db[1-2].[app\|web]_order_\\.*:col2`. This parameter is required for tables without a primary key, and the selected column must be of a non-nullable (`NOT NULL`) type. For tables with a primary key, this parameter is optional, and you can only select one column from the primary key.
scan.parse.online.schema.changes.enabled	During the incremental phase, specifies whether to parse DDL events for ApsaraDB RDS lock-free schema changes.	No	BOOLEAN	false	Valid values: true: Parses DDL events for ApsaraDB RDS lock-free schema changes. false (Default): Does not parse DDL events for ApsaraDB RDS lock-free schema changes. This is an experimental feature. Before performing an online lock-free change, we recommend that you create a savepoint for the Flink job to ensure it can be restored if needed.
scan.incremental.snapshot.backfill.skip	Specifies whether to skip backfill during the snapshot reading phase.	No	BOOLEAN	false	Valid values: true: Skips backfill during the snapshot reading phase. false (Default): Does not skip backfill during the snapshot reading phase. If backfill is skipped, changes to the table that occur during the snapshot phase are read in the later incremental phase instead of being merged into the snapshot. Important Skipping backfill can cause data inconsistencies because changes from the snapshot phase may be replayed. This mode provides at-least-once semantics.
treat-tinyint1-as-boolean.enabled	Specifies whether to treat the `TINYINT(1)` data type as `BOOLEAN`.	No	BOOLEAN	true	Valid values: true (Default): Treats the `TINYINT(1)` type as `BOOLEAN`. false: Does not treat the `TINYINT(1)` type as `BOOLEAN`.
treat-timestamp-as-datetime-enabled	Specifies whether to treat the `TIMESTAMP` data type as `DATETIME`.	No	BOOLEAN	false	Valid values: true: Treats the PolarDB-X `TIMESTAMP` type as `DATETIME` and maps it to the CDC `TIMESTAMP` type. false (Default): Maps the PolarDB-X `TIMESTAMP` type to the CDC `TIMESTAMP_LTZ` type. The PolarDB-X `TIMESTAMP` type stores UTC time and is affected by the time zone. The PolarDB-X `DATETIME` type stores literal time and is not affected by the time zone. When enabled, this parameter converts PolarDB-X `TIMESTAMP` data to the `DATETIME` type based on the `server-time-zone`.
include-comments.enabled	Specifies whether to synchronize table and column comments.	No	BOOLEAN	false	Valid values: true: Synchronizes table and column comments. false (Default): Does not synchronize table and column comments. Enabling this parameter increases the memory usage of the job.
scan.incremental.snapshot.unbounded-chunk-first.enabled	Specifies whether to dispatch unbounded chunks first during the snapshot reading phase.	No	BOOLEAN	false	Valid values: true: Dispatches unbounded chunks first during the snapshot reading phase. false (Default): Does not dispatch unbounded chunks first during the snapshot reading phase. This is an experimental feature. Enabling it can reduce the risk of an OOM error on the TaskManager when the last chunk is being synchronized. We recommend that you add this parameter before the first startup of the job.
binlog.session.network.timeout	The network timeout for the binlog connection.	No	DURATION	10m	If set to `0s`, the server-side default timeout is used.
scan.rate-limit.records-per-second	The maximum number of records the source can emit per second.	No	LONG	N/A	Use this parameter to limit data reading throughput. This limit applies to both the snapshot and incremental phases. The source's `numRecordsOutPerSecond` metric shows the number of records emitted per second for the entire data stream. You can use this metric to tune this parameter. During the snapshot reading phase, you may also need to reduce the number of rows read per batch by decreasing the value of the `scan.incremental.snapshot.chunk.size` parameter.
include-binlog-meta.enable	Specifies whether to include raw PolarDB-X binlog metadata, such as GTID and binlog position, in the emitted records.	No	BOOLEAN	false	Applicable to raw binlog synchronization scenarios, such as replacing an existing Canal-based synchronization pipeline.
polardbx.binlog.ignore.archive-events.enabled	Specifies whether to ignore archive events (primarily `DELETE` events) in the PolarDB-X binlog.	No	BOOLEAN	false
polardbx.binlog.ignore.query-events.enabled	Specifies whether to ignore `ROWS_QUERY_LOG_EVENT` events in the PolarDB-X binlog.	No	BOOLEAN	false
polardbx.binlog.include.tables	A comma-separated list of tables to subscribe to for binlog events.	No	STRING	N/A	Note This parameter affects only the incremental (binlog) phase and does not apply to the snapshot phase.
polardbx.binlog.exclude.tables	A comma-separated list of tables whose binlog events should be ignored.	No	STRING	N/A	Note This parameter affects only the incremental (binlog) phase and does not apply to the snapshot phase.