ApsaraDB RDS:Use a replication slot to set up change tracking

Step 1: Create a test database

1. Go to the Instances page. In the top navigation bar, select the region where your RDS instance resides. Click the instance ID.

2. In the left-side navigation pane, click Databases.

3. Click Create Database.

   Note

   This example creates a database named testdb. For instructions, see Create a database.

   

Step 2: Create accounts and configure permissions

1. In the left-side navigation pane, click Accounts.

2. Click Create Account to create two accounts:

   • A privileged account named db_admin for administration

   • A standard account named cdc_user for capturing change data

   Note

   The account names in this example are for reference only. For instructions, see Create an account.

3. Connect to the instance as db_admin:

   Note

   To find your instance endpoint, see View and change endpoints and port numbers.

   psql -h <instance-endpoint> -p 5432 -U db_admin -d testdb

4. Grant the replication role to cdc_user:

   ALTER USER cdc_user WITH REPLICATION;

   Verify the assignment:

   SELECT rolreplication FROM pg_roles WHERE rolname='cdc_user';

   Expected output:

   rolreplication
   ----------------
    t
   (1 row)

5. Grant SELECT permissions on all tables to cdc_user:

   GRANT SELECT ON ALL TABLES IN SCHEMA PUBLIC to cdc_user;

Step 3: Set wal_level to logical

The wal_level parameter controls how much information is written to WAL logs. The default value is replica. CDC requires the logical value, which adds the information needed for logical decoding.

To verify the current value:

SELECT name, setting, short_desc, source
FROM pg_settings
WHERE name = 'wal_level';

Expected output:

name      | setting |                          short_desc                          |       source
----------+---------+---------------------------------------------------------------+--------------------
 wal_level | replica | Sets the level of information written to the WAL.            | configuration file
(1 row)

To change wal_level to logical:

1. Go to the Instances page and click your instance ID.

2. In the left-side navigation pane, click Parameters.

3. Set wal_level to logical and submit the change.

Step 4: Create a logical replication slot

Wait for the instance to restart and return to the Running state before proceeding.

1. Connect to the instance as db_admin:

   psql -h <instance-endpoint> -p 5432 -U db_admin -d testdb

2. Create a logical replication slot named cdc_replication_slot using the test_decoding output plugin:

   Note

   test_decoding is an output plugin provided by open source PostgreSQL. cdc_replication_slot is an example name — use any name that fits your requirements.

   SELECT pg_create_logical_replication_slot('cdc_replication_slot', 'test_decoding');

   Expected output:

   pg_create_logical_replication_slot
   ------------------------------------
    (cdc_replication_slot,1/14003428)
   (1 row)

3. Verify the replication slot was created:

   SELECT * FROM pg_replication_slots;

   Expected output:

   slot_name            |    plugin     | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size | two_phase
   ---------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+---------------+-----------
    cdc_replication_slot | test_decoding | logical   |  18822 | testdb   | f         | f      |            |      |        22356 | 1/140033F0  | 1/14003428          | reserved   |               | f
   (1 row)

Step 5: Create test data

Run the following statements to create a test table and insert sample data:

CREATE TABLE public.tb_test(
    id int NOT NULL PRIMARY KEY
);

ALTER TABLE public.tb_test ADD name varchar(1) NULL;

INSERT INTO public.tb_test SELECT 1, 'A';

Step 6: Read change data from the replication slot

1. Connect to the instance as cdc_user:

   psql -h <instance-endpoint> -p 5432 -U cdc_user -d testdb

2. Read change data from the replication slot without consuming it:

   SELECT * FROM pg_logical_slot_peek_changes('cdc_replication_slot', null, null);

   Expected output:

   lsn        |  xid  |                                  data
   -----------+-------+------------------------------------------------------------------------
    1/14003D90 | 22376 | BEGIN 22376
    1/1401DDE8 | 22376 | COMMIT 22376
    1/1401DDE8 | 22377 | BEGIN 22377
    1/1401E100 | 22377 | COMMIT 22377
    1/1401E2A8 | 22382 | BEGIN 22382
    1/1401E2A8 | 22382 | table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
    1/1401E3C0 | 22382 | COMMIT 22382
   (7 rows)

Step 7: Consume change data using pg_recvlogical

pg_recvlogical is the native logical decoding tool included with PostgreSQL. It reads from a replication slot and advances the consumption position so that consumed WAL data can be cleared.

1. Run \q to close the current database connection.

2. Run the following command to start consuming change data:

   Note

   Run pg_recvlogical as the postgres user (su - postgres). If you see -bash: pg_recvlogical: command not found, see the FAQ.

   pg_recvlogical -h <instance-endpoint> -U <privileged-account> -d <database> \
     --create-slot --if-not-exists --slot=cdc_replication_slot \
     --plugin=test_decoding --start -f -

   Example:

   pg_recvlogical -h pgm-*****.pgsql.singapore.rds.aliyuncs.com -U db_admin -d testdb \
     --create-slot --if-not-exists --slot=cdc_replication_slot \
     --plugin=test_decoding --start -f -

   Expected output:

   BEGIN 22376
   COMMIT 22376
   BEGIN 22377
   COMMIT 22377
   BEGIN 22382
   table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
   COMMIT 22382