Performing logical replication for Amazon RDS for PostgreSQL

To turn on logical decoding for an RDS for PostgreSQL DB instance

1. Make sure that the user account that you're using has these roles:

   • The rds_superuser role so you can turn on logical replication

   • The rds_replication role to grant permissions to manage logical slots and to stream data using logical slots

2. Set the rds.logical_replication static parameter to 1. As part of applying this parameter, also set the parameters wal_level, max_wal_senders, max_replication_slots, and max_connections. These parameter changes can increase WAL generation, so set the rds.logical_replication parameter only when you are using logical slots.

3. Reboot the DB instance for the static rds.logical_replication parameter to take effect.

4. Create a logical replication slot as explained in the next section. This process requires that you specify a decoding plugin. Currently, RDS for PostgreSQL supports the test_decoding and wal2json output plugins that ship with PostgreSQL.

1. Create the source tables

   Connect to the source database in your RDS for PostgreSQL DB instance:

   
   source=> CREATE TABLE testtab (slno int primary key);
   CREATE TABLE
           

2. Insert data into the source tables

   
   source=> INSERT INTO testtab VALUES (generate_series(1,1000));
   INSERT 0 1000
           

3. Create a publication for source tables

   • Create a publication for the source tables:

     
     source=> CREATE PUBLICATION testpub FOR TABLE testtab;
     CREATE PUBLICATION
             

   • Use a SELECT query to verify the details of the publication that was created:

     
     source=> SELECT * FROM pg_publication;
       oid   | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot
     --------+---------+----------+--------------+-----------+-----------+-----------+-------------+------------
      115069 | testpub |    16395 | f            | t         | t         | t         | t           | f
     (1 row)
             

   • Verify that the source tables are added to the publication:

     
     source=> SELECT * FROM pg_publication_tables; 
     pubname | schemaname | tablename
     ---------+------------+-----------
      testpub | public     | testtab
     (1 rows)
             

   • To replicate all tables in a database, use:

     
     CREATE PUBLICATION testpub FOR ALL TABLES;
               

   • If the publication is already created for individual table and you need to add new table, you can run below query to add any new tables into the existing publication:

     
     ALTER PUBLICATION <publication_name> add table <new_table_name>;
               

4. Connect to target database and create target tables

   • Connect to the target database in the target DB instance. Create the target tables with the same names as the source tables:

     
     target=> CREATE TABLE testtab (slno int primary key);
     CREATE TABLE
     

   • Make sure that there's no data present in the target tables by running a SELECT query on the target tables:

         
     target=> SELECT count(*) FROM testtab;
      count
     -------
          0
     (1 row)
             

5. Create and verify subscription in target database

   • Create the subscription in the target database:

     
     target=> CREATE SUBSCRIPTION testsub 
     CONNECTION 'host=<source RDS/host endpoint> port=5432 dbname=<source_db_name> user=<user> password=<password>' 
     PUBLICATION testpub;
     NOTICE:  Created replication slot "testsub" on publisher
     CREATE SUBSCRIPTION
             

   • Use a SELECT query to verify that the subscription is enabled:

     
     target=> SELECT oid, subname, subenabled, subslotname, subpublications FROM pg_subscription;
       oid  | subname | subenabled | subslotname | subpublications
     -------+---------+------------+-------------+-----------------
      16434 | testsub | t          | testsub     | {testpub}
     (1 row)
             

   • When the subscription is created, it loads all data from the source tables to the target tables. Run a SELECT query on the target tables to verify that the initial data loads:

     
     target=> SELECT count(*) FROM testtab;
      count
     -------
       1000
     (1 row)
             

6. Verify replication slot in source database

   The creation of a subscription in the target database creates a replication slot in the source database. Verify the replication slot details by running the following SELECT query on the source database:

   
   source=> SELECT * FROM pg_replication_slots;
    
   slot_name |  plugin  | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size
   ----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+---------------
   testsub   | pgoutput | logical   | 115048 | source   | f         | t      |        846 |      |         6945 | 58/B4000568 | 58/B40005A0         | reserved   |
   (1 row)
           

7. Testing replication

   • Test whether data changes in the source tables are being replicated to the target tables by inserting rows into the source tables:

     
     source=> INSERT INTO testtab VALUES(generate_series(1001,2000));
     INSERT 0 1000
     
     source=> SELECT count(*) FROM testtab; 
      count
     -------
       2000
     (1 row)
           

   • Verify the number of rows in the target tables to confirm that new inserts are being replicated:

     
     target=> SELECT count(*) FROM testtab;
      count
     -------
       2000
     (1 row)
           

8. Refreshing the subscription after adding tables

   • When you add new tables to an existing publication, it is mandatory to refresh the subscription for the changes to take effect:

     
     ALTER SUBSCRIPTION <subscription_name> REFRESH PUBLICATION;
           

   • This command fetches missing table information from the publisher and starts replication for tables that were added to the subscribed-to publications since the subscription was created or last refreshed.