Melakukan replikasi logis untuk Amazon RDS for PostgreSQL

Cara mengaktifkan decoding logis untuk instans DB RDS for PostgreSQL

1. Pastikan akun pengguna yang Anda gunakan memiliki peran berikut:

   • Peran rds_superuser agar Anda dapat mengaktifkan replikasi logis

   • Peran rds_replication untuk memberikan izin guna mengelola slot logis dan mengalirkan data menggunakan slot logis

2. Atur parameter statis rds.logical_replication ke 1. Sebagai bagian dari penerapan parameter ini, atur juga parameterwal_level, max_wal_senders, max_replication_slots, dan max_connections. Perubahan parameter ini dapat meningkatkan pembuatan WAL, jadi atur parameter rds.logical_replication hanya saat Anda menggunakan slot logis.

3. Boot ulang instans DB agar parameter rds.logical_replication statis berlaku.

4. Buat slot replikasi logis sebagaimana dijelaskan di bagian selanjutnya. Proses ini mengharuskan Anda menentukan plugin decoding. Saat ini, RDS for PostgreSQL mendukung plugin output test_decoding dan wal2json yang dikirimkan dengan PostgreSQL.

1. Buat tabel sumber

   Connect ke database sumber di RDS Anda untuk PostgreSQL DB instance:

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

2. Masukkan data ke dalam tabel sumber

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

3. Buat publikasi untuk tabel sumber

   • Buat publikasi untuk tabel sumber:

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

   • Gunakan kueri SELECT untuk memverifikasi detail publikasi yang dibuat:

     
     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)
             

   • Verifikasi bahwa tabel sumber ditambahkan ke publikasi:

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

   • Untuk mereplikasi semua tabel dalam database, gunakan:

     
     CREATE PUBLICATION testpub FOR ALL TABLES;
               

   • Jika publikasi sudah dibuat untuk tabel individual dan Anda perlu menambahkan tabel baru, Anda dapat menjalankan kueri di bawah ini untuk menambahkan tabel baru ke publikasi yang ada:

     
     ALTER PUBLICATION <publication_name> add table <new_table_name>;
               

4. Connect ke database target dan buat tabel target

   • Connect ke database target dalam instans DB target. Buat tabel target dengan nama yang sama dengan tabel sumber:

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

   • Pastikan tidak ada data yang ada di tabel target dengan menjalankan kueri SELECT pada tabel target:

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

5. Membuat dan memverifikasi langganan di database target

   • Buat langganan di database target:

     
     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
             

   • Gunakan kueri SELECT untuk memverifikasi bahwa langganan diaktifkan:

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

   • Ketika langganan dibuat, itu memuat semua data dari tabel sumber ke tabel target. Jalankan kueri SELECT pada tabel target untuk memverifikasi bahwa data awal dimuat:

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

6. Verifikasi slot replikasi dalam database sumber

   Pembuatan langganan dalam database target menciptakan slot replikasi dalam database sumber. Verifikasi detail slot replikasi dengan menjalankan kueri SELECT berikut pada database sumber:

   
   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. Menguji replikasi

   • Uji apakah perubahan data dalam tabel sumber sedang direplikasi ke tabel target dengan menyisipkan baris ke dalam tabel sumber:

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

   • Verifikasi jumlah baris dalam tabel target untuk mengonfirmasi bahwa sisipan baru sedang direplikasi:

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

8. Menyegarkan langganan setelah menambahkan tabel

   • Saat Anda menambahkan tabel baru ke publikasi yang ada, wajib menyegarkan langganan agar perubahan diterapkan:

     
     ALTER SUBSCRIPTION <subscription_name> REFRESH PUBLICATION;
           

   • Perintah ini mengambil informasi tabel yang hilang dari penerbit dan memulai replikasi untuk tabel yang ditambahkan ke publikasi berlangganan sejak langganan dibuat atau terakhir diperbarui.