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 をご参照ください。

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

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

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

-- [Subscriber] 読み取り専用インスタンスに接続するサブスクリプションを作成します。
-- サブスクリプションは、読み取り専用インスタンスで '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) → 読み取り専用インスタンス → (論理レプリケーション) → 下流のサブスクライバー。

-- [Subscriber] データが同期されたことを確認します。
SELECT * FROM tab_rep;

出力例：

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

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

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

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

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

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

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