Using a PostgreSQL database as an AWS DMS source

Use the AWS master user account for the PostgreSQL DB instance as the user account for the PostgreSQL source endpoint. The master user account has the required roles that allow it to set up CDC.

If you use an account other than the master user account, make sure to create several objects from the master account for the account that you use. For more information, see Migrating an Amazon RDS for PostgreSQL database without using the master user account.

Set the rds.logical_replication parameter in your DB CLUSTER parameter group to 1. This static parameter requires a reboot of the DB instance to take effect. As part of applying this parameter, AWS DMS sets the wal_level, max_wal_senders, max_replication_slots, and max_connections parameters. These parameter changes can increase write ahead log (WAL) generation, so only set rds.logical_replication when you use logical replication slots.

The wal_sender_timeout parameter ends replication connections that are inactive longer than the specified number of milliseconds. The default for an AWS-managed PostgreSQL database is 30000 milliseconds (30 seconds). Setting the value to 0 (zero) disables the timeout mechanism, and is a valid setting for DMS.

When setting wal_sender_timeout to a non-zero value, a DMS task with CDC requires a minimum of 10000 milliseconds (10 seconds), and fails if the value is between 0 and 10000. Keep the value less than 5 minutes to avoid causing a delay during a Multi-AZ failover of a DMS replication instance.

Ensure the value of the max_worker_processes parameter in your DB Cluster Parameter Group is equal to, or higher than the total combined values of max_logical_replication_workers, autovacuum_max_workers, and max_parallel_workers. A high number of background worker processes might impact application workloads on small instances. So, monitor performance of your database if you set max_worker_processes higher than the default value.

When using Aurora PostgreSQL as a source with CDC, set synchronous_commit to ON.

Set the rds.logical_replication and sync_replication_slots parameters in your DB CLUSTER parameter group to 1. This static parameters require a reboot of the DB instance to take effect.

Run the following command to create the awsdms_ddl_audit table on Writer and to replace the objects_schema with the name of the schema to use:

CREATE TABLE objects_schema.awsdms_ddl_audit
(
  c_key    bigserial primary key,
  c_time   timestamp,    -- Informational
  c_user   varchar(64),  -- Informational: current_user
  c_txn    varchar(16),  -- Informational: current transaction
  c_tag    varchar(24),  -- Either 'CREATE TABLE' or 'ALTER TABLE' or 'DROP TABLE'
  c_oid    integer,      -- For future use - TG_OBJECTID
  c_name   varchar(64),  -- For future use - TG_OBJECTNAME
  c_schema varchar(64),  -- For future use - TG_SCHEMANAME. For now - holds current_schema
  c_ddlqry  text         -- The DDL query associated with the current DDL event
);

Run the following command to create the awsdms_intercept_ddl function and to replace the objects_schema with the name of the schema to use:

CREATE OR REPLACE FUNCTION objects_schema.awsdms_intercept_ddl()
  RETURNS event_trigger
LANGUAGE plpgsql
SECURITY DEFINER
  AS $$
  declare _qry text;
BEGIN
  if (tg_tag='CREATE TABLE' or tg_tag='ALTER TABLE' or tg_tag='DROP TABLE' or tg_tag = 'CREATE TABLE AS') then
         SELECT current_query() into _qry;
         insert into objects_schema.awsdms_ddl_audit
         values
         (
         default,current_timestamp,current_user,cast(TXID_CURRENT()as varchar(16)),tg_tag,0,'',current_schema,_qry
         );
         delete from objects_schema.awsdms_ddl_audit;
end if;
END;
$$;

Run the following command to create the awsdms_intercept_ddl event trigger:

CREATE EVENT TRIGGER awsdms_intercept_ddl ON ddl_command_end EXECUTE PROCEDURE objects_schema.awsdms_intercept_ddl();

Ensure that all the users and roles that access these events have the necessary DDL permissions. For example:

grant all on public.awsdms_ddl_audit to public;
grant all on public.awsdms_ddl_audit_c_key_seq to public;

Create replication slot on Writer:

SELECT * FROM pg_create_logical_replication_slot('dms_read_replica_slot', 'test_decoding', false, true);

Ensure the replication slot is available on Reader:

select * from pg_catalog.pg_replication_slots where slot_name = 'dms_read_replica_slot';

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|inactive_since               |conflicting|invalidation_reason|failover|synced|
---------------------+-------------+---------+------+--------+---------+------+----------+----+------------+-----------+-------------------+----------+-------------+---------+-----------------------------+-----------+-------------------+--------+------+
dms_read_replica_slot|test_decoding|logical  |     5|postgres|false    |false |          |    |3559        |0/180011B8 |0/180011F0         |reserved  |             |true     |2025-02-10 15:45:04.083 +0100|false      |                   |false   |false |

Create DMS source endpoint for Read Replica and set logical replication slot name via the Extra Connection Attribute:

slotName=dms_read_replica_slot;

Create and start the CDC/FL+CDC task.

Choose the schema where the objects are to be created. The default schema is public. Ensure that the schema exists and is accessible by the OtherThanMaster account.

Log in to the PostgreSQL DB instance using the user account other than the master account, here the OtherThanMaster account.

Create the table awsdms_ddl_audit by running the following command, replacing objects_schema in the following code with the name of the schema to use.

CREATE TABLE objects_schema.awsdms_ddl_audit
(
  c_key    bigserial primary key,
  c_time   timestamp,    -- Informational
  c_user   varchar(64),  -- Informational: current_user
  c_txn    varchar(16),  -- Informational: current transaction
  c_tag    varchar(24),  -- Either 'CREATE TABLE' or 'ALTER TABLE' or 'DROP TABLE'
  c_oid    integer,      -- For future use - TG_OBJECTID
  c_name   varchar(64),  -- For future use - TG_OBJECTNAME
  c_schema varchar(64),  -- For future use - TG_SCHEMANAME. For now - holds current_schema
  c_ddlqry  text         -- The DDL query associated with the current DDL event
);
                           
                        

Create the function awsdms_intercept_ddl by running the following command, replacing objects_schema in the code following with the name of the schema to use.

CREATE OR REPLACE FUNCTION objects_schema.awsdms_intercept_ddl()
  RETURNS event_trigger
LANGUAGE plpgsql
SECURITY DEFINER
  AS $$
  declare _qry text;
BEGIN
  if (tg_tag='CREATE TABLE' or tg_tag='ALTER TABLE' or tg_tag='DROP TABLE' or tg_tag = 'CREATE TABLE AS') then
         SELECT current_query() into _qry;
         insert into objects_schema.awsdms_ddl_audit
         values
         (
         default,current_timestamp,current_user,cast(TXID_CURRENT()as varchar(16)),tg_tag,0,'',current_schema,_qry
         );
         delete from objects_schema.awsdms_ddl_audit;
end if;
END;
$$;
                        

Log out of the OtherThanMaster account and log in with an account that has the rds_superuser role assigned to it.

Create the event trigger awsdms_intercept_ddl by running the following command.

CREATE EVENT TRIGGER awsdms_intercept_ddl ON ddl_command_end 
EXECUTE PROCEDURE objects_schema.awsdms_intercept_ddl();
                        

Make sure that all users and roles that access these events have the necessary DDL permissions. For example:

grant all on public.awsdms_ddl_audit to public;
grant all on public.awsdms_ddl_audit_c_key_seq to public;

Create a pglogical extension on your source PostgreSQL database:

1. Set the correct parameter:

   • For self-managed PostgreSQL databases, set the database parameter shared_preload_libraries= 'pglogical'.

   • For PostgreSQL on Amazon RDS and Amazon Aurora PostgreSQL-Compatible Edition databases, set the parameter shared_preload_libraries to pglogical in the same RDS parameter group.

2. Restart your PostgreSQL source database.

3. On the PostgreSQL database, run the command, create extension pglogical;

Run the following command to verify that pglogical installed successfully:

select * FROM pg_catalog.pg_extension

Identify the logical replication slot used by an earlier replication task (a parent task) that you want to use as a start point. Then query the pg_replication_slots view on your source database to make sure that this slot does not have any active connections. If it does, resolve and close them before proceeding.

For the following steps, assume that your logical replication slot is abc1d2efghijk_34567890_z0yx98w7_6v54_32ut_1srq_1a2b34c5d67ef.

Create a new source endpoint that includes the following extra connection attribute setting.

slotName=abc1d2efghijk_34567890_z0yx98w7_6v54_32ut_1srq_1a2b34c5d67ef;

Create a new CDC-only task using the console, AWS CLI or AWS DMS API. For example, using the CLI you might run the following create-replication-task command.

aws dms create-replication-task --replication-task-identifier postgresql-slot-name-test 
--source-endpoint-arn arn:aws:dms:us-west-2:012345678901:endpoint:ABCD1EFGHIJK2LMNOPQRST3UV4 
--target-endpoint-arn arn:aws:dms:us-west-2:012345678901:endpoint:ZYX9WVUTSRQONM8LKJIHGF7ED6 
--replication-instance-arn arn:aws:dms:us-west-2:012345678901:rep:AAAAAAAAAAA5BB4CCC3DDDD2EE 
--migration-type cdc --table-mappings "file://mappings.json" --cdc-start-position "4AF/B00000D0" 
--replication-task-settings "file://task-pg.json"

To enable custom CDC start mode when creating a new CDC-only task using the AWS DMS console, do the following:

When this CDC task runs, AWS DMS raises an error if the specified logical replication slot does not exist. It also raises an error if the task isn't created with a valid setting for cdc-start-position.

Create a replication slot, as shown following:

SELECT * FROM pg_create_logical_replication_slot('replication_slot_name', 'pglogical');
                    

After the database creates the replication slot, get and note the restart_lsn and confirmed_flush_lsn values for the slot:

select * from pg_replication_slots where slot_name like 'replication_slot_name';

Note that the Native CDC Start position for a CDC task created after the replication slot can't be older than the confirmed_flush_lsn value.

For information about the restart_lsn and confirmed_flush_lsn values, see pg_replication_slots

Create a pglogical node.

SELECT pglogical.create_node(node_name := 'node_name', dsn := 'your_dsn_name');
                    

Create two replication sets using the pglogical.create_replication_set function. The first replication set tracks updates and deletes for tables that have primary keys. The second replication set tracks only inserts, and has the same name as the first replication set, with the added prefix 'i'.

SELECT pglogical.create_replication_set('replication_slot_name', false, true, true, false);
SELECT pglogical.create_replication_set('ireplication_slot_name', true, false, false, true);

Add a table to the replication set.

SELECT pglogical.replication_set_add_table('replication_slot_name', 'schemaname.tablename', true);
SELECT pglogical.replication_set_add_table('ireplication_slot_name', 'schemaname.tablename', true);
                    

Set the extra connection attribute (ECA) following when you create your source endpoint.

PluginName=PGLOGICAL;slotName=slot_name;