PolarDB:Configure CDC to synchronize data changes

In a PolarDB for PostgreSQL Distributed Edition cluster, the key to setting up CDC lies in understanding where the data originates:

• Distributed tables: Data changes occur on individual data nodes (DNs), so the downstream system must subscribe to all DN nodes to obtain the complete change data.

• Replicated tables: Data changes are synchronized to all nodes, but to avoid duplication, the change stream is published exclusively by the primary compute node (primary CN). The downstream system only needs to subscribe to the primary CN.

Prerequisites: Check and configure

Before you begin, make sure the relevant parameters for the database cluster are properly configured. These parameters are prerequisites for enabling logical replication.

• Run the following SQL on the primary CN node to check whether polar_cluster.enable_change_data_capture is set to on on all nodes.

  SELECT success, result FROM run_command_on_all_nodes($$ SHOW polar_cluster.enable_change_data_capture $$);

  Expected result: the result column should be on for all nodes.

   success | result 
  ---------+--------
   t       | on
   t       | on
   t       | on
   t       | on

• Run the following SQL on the primary CN node to check whether wal_level is set to logical on all nodes.

  SELECT success, result FROM run_command_on_all_nodes($$ SHOW wal_level $$);

  Expected result: the result column should be logical for all nodes.

  success | result  
  ---------+---------
   t       | logical
   t       | logical
   t       | logical
   t       | logical

Step 1: Create a publication and replication slots on the publisher

The publisher (your PolarDB for PostgreSQL Distributed Edition cluster) is configured in two steps: creating a publication and creating replication slots.

CREATE PUBLICATION <publication_name> FOR TABLE <table_name1>, <table_name2>;

WITH nodes AS (
    SELECT
        nodename,
        nodeport,
        $$ SELECT pg_create_logical_replication_slot('<publication_slot_name>', 'pgoutput', false) $$ AS cmd
    FROM pg_dist_node
    WHERE (nodeid = 1 OR shouldhaveshards = true) AND noderole = 'primary'
)
SELECT result.*
FROM (
    SELECT
        array_agg(nodename) as nodenames,
        array_agg(nodeport) as nodeports,
        array_agg(cmd) as cmds
    FROM nodes
) params,
LATERAL master_run_on_worker(nodenames, nodeports, cmds, true) AS result;

   node_name    | node_port | success |               result               
----------------+-----------+---------+------------------------------------
 10.xxx.xxx.xxx |      3007 | t       | (<publication_slot_name>,0/C024D7D0)
 10.xxx.xxx.xxx |      3006 | t       | (<publication_slot_name>,0/C33B6668)
 10.xxx.xxx.xxx |      3003 | t       | (<publication_slot_name>,0/C33949B0)
(3 rows)

Step 2: Create subscriptions on the subscriber

The subscriber (such as Debezium, Flink, or another PostgreSQL instance) must create a separate subscription connection for each node that has a replication slot (the primary CN and all DNs) to receive the complete change data.

{
    "name": "xxx",
    "config": {
        "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname" : "<DN1_primary_address>",
        "database.port" : "<DN1_port>",
        "database.user" : "<your_username>",
        "database.password" : "<your_password>",
        "database.dbname" : "postgres",
        "slot.name": "<replication_slot_name>",
        "publication.name": "<publication_name>",
        ...
    }
}

CREATE SUBSCRIPTION test_subscription
    CONNECTION 'dbname=postgres host=<DN1_primary_address> port=<DN1_port> user=<your_username> password=<your_password>'
    PUBLICATION <publication_name>
    WITH (create_slot=false, slot_name='<replication_slot_name>');

Step 3: Verify the CDC pipeline

After configuring all subscribers, you can run the following SQL on the primary CN to check whether the replication slots on all publishing nodes are active.

WITH nodes AS (
    SELECT
        nodename,
        nodeport,
        $$ SELECT active FROM pg_replication_slots WHERE slot_name = '<publication_slot_name>' $$ AS cmd
    FROM pg_dist_node
    WHERE (nodeid = 1 OR shouldhaveshards = true) AND noderole = 'primary'
)
SELECT result.*
FROM (
    SELECT
        array_agg(nodename) as nodenames,
        array_agg(nodeport) as nodeports,
        array_agg(cmd) as cmds
    FROM nodes
) params,
LATERAL master_run_on_worker(nodenames, nodeports, cmds, true) AS result;

Expected result: if the result column is t, the replication relationship between the corresponding node and the subscriber has been established. At this point, you can insert or modify data in the source table and observe whether the subscriber receives the changes.

  node_name     | node_port | success | result 
----------------+-----------+---------+--------
 10.xxx.xxx.xxx |      3007 | t       | t
 10.xxx.xxx.xxx |      3006 | t       | t
 10.xxx.xxx.xxx |      3003 | t       | t
(3 rows)

Maintenance and cleanup

If the downstream system no longer needs to subscribe to data, you should stop the subscription on the downstream side and then promptly clean up the replication slots on the publisher to free WAL log space.

Run the following SQL on the primary CN to drop replication slots from all relevant nodes in a single batch:

WITH nodes AS (
    SELECT
        nodename,
        nodeport,
        $$ SELECT pg_drop_replication_slot('<publication_slot_name>') $$ AS cmd
    FROM pg_dist_node
    WHERE (nodeid = 1 OR shouldhaveshards = true) AND noderole = 'primary'
)
SELECT result.*
FROM (
    SELECT
        array_agg(nodename) as nodenames,
        array_agg(nodeport) as nodeports,
        array_agg(cmd) as cmds
    FROM nodes
) params,
LATERAL master_run_on_worker(nodenames, nodeports, cmds, true) AS result;