ApsaraDB RDS:レプリケーションスロットを用いた変更追跡の設定

ステップ 1：テスト用データベースの作成

1. インスタンス ページに移動します。上部のナビゲーションバーから、ご利用の RDS インスタンスが配置されているリージョンを選択し、インスタンス ID をクリックします。

2. 左側のナビゲーションウィンドウで、データベース をクリックします。

3. データベースの作成 をクリックします。

   説明

   本例では、testdb という名前のデータベースを作成します。詳細については、「データベースの作成」をご参照ください。

   

ステップ 2：アカウントの作成と権限の設定

1. 左側のナビゲーションウィンドウで、アカウント をクリックします。

2. アカウントの作成 をクリックし、以下の 2 つのアカウントを作成します：

   • 管理用の特権アカウント：db_admin

   • 変更データを取得するための標準アカウント：cdc_user

   説明

   本例のアカウント名は参考用です。詳細については、「アカウントの作成」をご参照ください。

3. db_admin としてインスタンスに接続します。

   説明

   インスタンスのエンドポイントを確認するには、「エンドポイントおよびポート番号の表示と変更」をご参照ください。

   psql -h <instance-endpoint> -p 5432 -U db_admin -d testdb

4. cdc_user にレプリケーションロールを付与します。

   ALTER USER cdc_user WITH REPLICATION;

   付与されたことを確認します。

   SELECT rolreplication FROM pg_roles WHERE rolname='cdc_user';

   期待される出力：

   rolreplication
   ----------------
    t
   (1 行)

5. cdc_user に、すべてのテーブルに対する SELECT 権限を付与します。

   GRANT SELECT ON ALL TABLES IN SCHEMA PUBLIC to cdc_user;

ステップ 3：wal_level を logical に設定

wal_level パラメーターは、WAL ログに書き込まれる情報量を制御します。デフォルト値は replica です。CDC を有効化するには、論理デコーディングに必要な情報を追加する logical 値を指定する必要があります。

現在の値を確認するには：

SELECT name, setting, short_desc, source
FROM pg_settings
WHERE name = 'wal_level';

期待される出力：

name      | setting |                          short_desc                          |       source
----------+---------+---------------------------------------------------------------+--------------------
 wal_level | replica | Sets the level of information written to the WAL.            | configuration file
(1 行)

wal_level を logical に変更するには、以下の手順を実行します。

1. インスタンス ページに移動し、インスタンス ID をクリックします。

2. 左側のナビゲーションウィンドウで、パラメーター をクリックします。

3. wal_level を logical に設定し、変更を送信します。

ステップ 4：論理レプリケーションスロットの作成

インスタンスが再起動し、実行中 状態に戻ってから次の手順に進んでください。

1. db_admin としてインスタンスに接続します。

   psql -h <instance-endpoint> -p 5432 -U db_admin -d testdb

2. cdc_replication_slot という名前の論理レプリケーションスロットを、test_decoding 出力プラグインを用いて作成します。

   説明

   test_decoding は、オープンソース PostgreSQL が提供する出力プラグインです。cdc_replication_slot は例の名前であり、要件に合った任意の名前を使用できます。

   SELECT pg_create_logical_replication_slot('cdc_replication_slot', 'test_decoding');

   期待される出力：

   pg_create_logical_replication_slot
   ------------------------------------
    (cdc_replication_slot,1/14003428)
   (1 行)

3. レプリケーションスロットが正しく作成されたことを確認します。

   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 | two_phase
   ---------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+---------------+-----------
    cdc_replication_slot | test_decoding | logical   |  18822 | testdb   | f         | f      |            |      |        22356 | 1/140033F0  | 1/14003428          | reserved   |               | f
   (1 行)

ステップ 5：テスト用データの作成

以下のステートメントを実行して、テスト用テーブルを作成し、サンプルデータを挿入します。

CREATE TABLE public.tb_test(
    id int NOT NULL PRIMARY KEY
);

ALTER TABLE public.tb_test ADD name varchar(1) NULL;

INSERT INTO public.tb_test SELECT 1, 'A';

ステップ 6：レプリケーションスロットからの変更データの読み取り

1. cdc_user としてインスタンスに接続します。

   psql -h <instance-endpoint> -p 5432 -U cdc_user -d testdb

2. 変更データを消費せずに、レプリケーションスロットから読み取ります。

   SELECT * FROM pg_logical_slot_peek_changes('cdc_replication_slot', null, null);

   期待される出力：

   lsn        |  xid  |                                  data
   -----------+-------+------------------------------------------------------------------------
    1/14003D90 | 22376 | BEGIN 22376
    1/1401DDE8 | 22376 | COMMIT 22376
    1/1401DDE8 | 22377 | BEGIN 22377
    1/1401E100 | 22377 | COMMIT 22377
    1/1401E2A8 | 22382 | BEGIN 22382
    1/1401E2A8 | 22382 | table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
    1/1401E3C0 | 22382 | COMMIT 22382
   (7 行)

ステップ 7：pg_recvlogical を用いた変更データの消費

pg_recvlogical は、PostgreSQL に同梱されるネイティブな論理デコーディングツールです。このツールはレプリケーションスロットから読み取り、消費位置を進めることで、消費済みの WAL データをクリア可能にします。

1. 現在のデータベース接続を終了するには、\q を実行します。

2. 以下のコマンドを実行して、変更データの消費を開始します。

   説明

   pg_recvlogical は postgres ユーザーとして実行する必要があります（su - postgres）。-bash: pg_recvlogical: command not found というエラーが表示された場合は、「よくある質問」をご参照ください。

   pg_recvlogical -h <instance-endpoint> -U <privileged-account> -d <database> \
     --create-slot --if-not-exists --slot=cdc_replication_slot \
     --plugin=test_decoding --start -f -

   例：

   pg_recvlogical -h pgm-*****.pgsql.singapore.rds.aliyuncs.com -U db_admin -d testdb \
     --create-slot --if-not-exists --slot=cdc_replication_slot \
     --plugin=test_decoding --start -f -

   期待される出力：

   BEGIN 22376
   COMMIT 22376
   BEGIN 22377
   COMMIT 22377
   BEGIN 22382
   table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
   COMMIT 22382