ApsaraDB RDS:読み取り専用インスタンス上の論理レプリケーションスロット

読み取り専用インスタンス上にローカル論理レプリケーションスロットを作成します。

-- [RO] ローカル論理レプリケーションスロットを作成します。
SELECT pg_create_logical_replication_slot('my_slot', 'test_decoding');

スロットが作成されたことを確認します。

-- [RO] スロットが作成されたことを確認します。
SELECT slot_name, plugin FROM pg_replication_slots WHERE slot_name = 'my_slot';

出力例：

 slot_name |    plugin     
-----------+---------------
 my_slot   | test_decoding 
(1 row)

プライマリインスタンスにデータを書き込みます。

-- [RW] プライマリインスタンスにデータを書き込みます。
INSERT INTO my_table VALUES (1), (2), (3);

WAL が読み取り専用インスタンスでリプレイされるのを待ちます。

プライマリインスタンスで、現在の WAL LSN を照会します：

-- [RW] プライマリインスタンスで現在の WAL LSN を取得します。
SELECT pg_current_wal_lsn();

読み取り専用インスタンスで、リプレイ LSN がプライマリインスタンスの WAL LSN 以上になるまでリプレイの進行状況を照会します：

-- [RO] 読み取り専用インスタンスのリプレイ進行状況を確認します。
SELECT pg_last_wal_replay_lsn();

読み取り専用インスタンスのスロットから変更を利用します。

-- [RO] スロットから変更を利用します。
SELECT * FROM pg_logical_slot_get_changes('my_slot', NULL, NULL);

出力例：

    lsn    | xid |                  data
-----------+-----+----------------------------------------
 0/XXXXXXX | 123 | table public.my_table: INSERT: ...
(3 rows)

スロットが不要になったら、削除します。

-- [RO] スロットを削除します。
SELECT pg_drop_replication_slot('my_slot');

プライマリインスタンス上にテーブルとパブリケーションを作成します。

-- [RW] プライマリインスタンス上にテーブルとパブリケーションを作成します。
CREATE TABLE tab_rep (id int PRIMARY KEY, data text);
CREATE PUBLICATION my_pub FOR TABLE tab_rep;

WAL が読み取り専用インスタンスでリプレイされるのを待ち、パブリケーションが表示されることを確認します。手順については、「基本的な使用法」セクションの手順 4 をご参照ください。

下流のサブスクライバー上に同じテーブル構造を作成します。

-- [サブスクライバー] サブスクライバーインスタンス上にテーブル構造を作成します。
CREATE TABLE tab_rep (id int PRIMARY KEY, data text);

下流のサブスクライバー上で、読み取り専用インスタンスに接続するサブスクリプションを作成します。

-- [サブスクライバー] 読み取り専用インスタンスに接続するサブスクリプションを作成します。
-- サブスクリプションは、読み取り専用インスタンス上に 'my_sub' という名前のスロットを自動的に作成します。
CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=<RO_HOST> port=<RO_PORT> dbname=postgres'
    PUBLICATION my_pub
    WITH (copy_data = off);

プライマリインスタンスにデータを書き込み、同期を確認します。

-- [RW] プライマリインスタンスにデータを書き込みます。
INSERT INTO tab_rep VALUES (1, 'hello'), (2, 'world');

データフロー：プライマリインスタンス → (ストリーミングレプリケーション / WAL) → 読み取り専用インスタンス → (論理レプリケーション) → 下流のサブスクライバー。

-- [サブスクライバー] データが同期されていることを確認します。
SELECT * FROM tab_rep;

出力例：

 id | data
----+-------
  1 | hello
  2 | world
(2 rows)

サブスクリプションが不要になったら、下流のサブスクライバーで削除します。このアクションにより、読み取り専用インスタンスからスロットが自動的に削除されます。

-- [サブスクライバー] サブスクリプションを削除します。これにより、読み取り専用インスタンス上のスロットが自動的に削除されます。
DROP SUBSCRIPTION my_sub;

読み取り専用インスタンス上にスロットを手動で作成します。

-- [RO] 読み取り専用インスタンス上にスロットを手動で作成します。
SELECT pg_create_logical_replication_slot('my_sub', 'pgoutput');

下流のサブスクライバー上で、既存のスロットを使用するサブスクリプションを作成します。

-- [サブスクライバー] サブスクリプションを作成し、既存のスロットを使用します。
CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=<RO_HOST> port=<RO_PORT> dbname=postgres'
    PUBLICATION my_pub
    WITH (copy_data = off, create_slot = false);