PostgreSQL CDC with SQL Mapping
Doris continuous load is experimental and supported starting with Doris 4.1.5.
SQL Mapping Sync is implemented through Job + CDC Stream TVF. The target is an existing Doris table (INSERT INTO tbl SELECT * FROM cdc_stream(...)). Doris SQL provides the expressiveness to support column mapping, filtering, and data transformation. This approach is suitable for real-time sync scenarios where data needs to be processed before writing.
By integrating the read capability of Flink CDC, Doris reads change logs (WAL) from PostgreSQL to perform full + incremental sync from the source table to the target table. If you want Doris to automatically create downstream tables and sync a group of tables on a per-database basis, see PostgreSQL CDC with Auto Table Creation.
Use Cases
- Real-time ingestion of a single table into the warehouse, with column pruning, filtering, or expression-based transformation required before write.
- The downstream Doris table already exists, and you want SQL to explicitly control field mapping.
- High consistency requirements that demand exactly-once semantics.
Prerequisites
Before creating a job, confirm the following:
- Logical replication is enabled on the PostgreSQL side. See the Configuration Guide for details.
- The current user has the
Loadprivilege. - The source table is a primary key table (only primary key tables are currently supported for sync).
- The target Unique Key table has been created in Doris in advance, and its schema is compatible with the mapping SQL.
Notes
- Exactly-once semantics is supported.
- Only primary key table sync is currently supported.
- The
Loadprivilege is required. - Logical replication must be enabled on the PostgreSQL side. See the Configuration Guide.
Quick Start
The overall workflow has two steps: create the continuous load job, then check the job status.
Step 1: Create the Load Job
Use CREATE STREAMING JOB to create a continuous load job that syncs public.source_table from PostgreSQL to db1.tbl1 in Doris:
CREATE JOB pg_single_sync
ON STREAMING
DO
INSERT INTO db1.tbl1
SELECT * FROM cdc_stream(
"type" = "postgres",
"jdbc_url" = "jdbc:postgresql://127.0.0.1:5432/postgres",
"driver_url" = "postgresql-42.5.1.jar",
"driver_class" = "org.postgresql.Driver",
"user" = "postgres",
"password" = "postgres",
"database" = "postgres",
"schema" = "public",
"table" = "source_table",
"offset" = "initial"
)
Step 2: Check the Load Status
Use the jobs table function to check the running status of Streaming-type jobs:
select * from jobs("type"="insert") where ExecuteType = "STREAMING";
For other common operations (pause, resume, delete, view tasks, and so on), see the Continuous Load Overview.
Parameter Reference
Data Source Parameters
Data source parameters are configured through the cdc_stream(...) TVF. They describe the PostgreSQL table to be synced and the read behavior.
| Parameter | Required | Default | Description |
|---|---|---|---|
type | Yes | - | Data source type. Set to postgres. |
jdbc_url | Yes | - | PostgreSQL JDBC connection string. |
driver_url | Yes | - | Path to the JDBC driver jar. Supports a file name, local absolute path, or HTTP URL. For details, see JDBC Catalog Overview. |
driver_class | Yes | - | JDBC driver class name, for example, org.postgresql.Driver. |
user | Yes | - | Database user name. |
password | Yes | - | Database password. |
database | No | Database name in the JDBC URL | PostgreSQL database name. When explicitly set, it overrides the database name in the JDBC URL. It cannot exceed 63 bytes after UTF-8 encoding. |
schema | Yes | - | Schema name. |
table | Yes | - | Name of the table to sync. Each SQL Mapping job supports one source table. |
offset | Yes | - | Startup offset. initial: full and incremental sync; snapshot: full sync only; latest: sync only changes after the job starts. You can also specify an exact JSON offset, such as {"lsn":"12345678"}. PostgreSQL does not support earliest. |
snapshot_split_size | No | 40960 | Split size in rows. During full sync, the table is divided into multiple splits. Must be a positive integer. |
snapshot_parallelism | No | 1 | Parallelism of the full-sync phase, that is, the maximum number of splits scheduled by a Task at one time. Must be a positive integer. |
skip_snapshot_backfill | No | false | Whether to skip WAL backfill during the snapshot. When set to true, at-least-once semantics are used. |
slot_name | No | doris_cdc_<job_id> | Logical replication slot name. The name can contain only lowercase letters, digits, and underscores, cannot start with a digit, and is limited to 63 characters. A custom slot must be created in advance and is not deleted by Doris. |
publication_name | No | doris_pub_<job_id> | Publication name. The naming rules are the same as for slot_name. A custom publication must be created in advance, include the current source table, and is not deleted by Doris. |
ssl_mode | No | disable | SSL mode. Valid values are disable, require, and verify-ca. |
ssl_rootcert | Conditionally required | - | CA certificate file in the format FILE:<file_name>. Required when ssl_mode is verify-ca. Upload the file first using CREATE FILE. |
include_delete_sign | No | false | Whether the TVF additionally outputs the __DORIS_DELETE_SIGN__ column. Set to true to sync upstream DELETE operations as deletes in a Doris primary key table. |
When include_delete_sign is enabled, the target table must be a Merge-on-Write Unique Key table, and __DORIS_DELETE_SIGN__ must be explicitly mapped in both the INSERT target column list and the SELECT list:
CREATE JOB pg_cdc_with_delete
ON STREAMING
DO
INSERT INTO db1.target_table (id, value, __DORIS_DELETE_SIGN__)
SELECT id, value, __DORIS_DELETE_SIGN__
FROM cdc_stream(
"type" = "postgres",
"jdbc_url" = "jdbc:postgresql://127.0.0.1:5432/postgres",
"driver_url" = "postgresql-42.5.1.jar",
"driver_class" = "org.postgresql.Driver",
"user" = "postgres",
"password" = "postgres",
"schema" = "public",
"table" = "source_table",
"offset" = "initial",
"include_delete_sign" = "true"
);
When you customize slot_name or publication_name, prepare the corresponding resource before creating the Job. Doris retains user-provided resources when the Job is deleted.
Job Configuration Parameters
Set the following parameters through CREATE JOB ... PROPERTIES (...):
| Parameter | Default | Description |
|---|---|---|
max_interval | 10 | Idle scheduling interval in seconds when no new upstream data is available. Must be an integer greater than or equal to 1. |
compute_group | Current session or user default compute group | Supported only in compute-storage decoupled mode. Specifies the compute group in which the job runs. The user must have the USAGE privilege on the compute group. |
session.<variable_name> | Default value of the corresponding session variable | Sets a session variable for the INSERT task. For load variables, see Insert Into Select. |
For the Job Property offset used to reset a CDC position through ALTER JOB and its complete restrictions, see Common Job Load Configuration Parameters. The initial position for a new job must be set in cdc_stream(...).