PolarDB:論理デコードの例

例：

次の例では、SQLインターフェイスを使用した論理デコードの制御を示します。

論理デコードを使用する前に、wal_levelをlogicalに設定し、max_replication_slotsを少なくとも1に設定する必要があります。 次に、スーパーユーザーとしてターゲットデータベース (以下の例ではpostgres) に接続する必要があります。

postgres=# -- 出力プラグイン 'test_decoding' を使用して 'regression_slot' という名前のスロットを作成します。postgres=# SELECT * FROM pg_create_logical_replication_slot('regression_slot' 、'test_decoding' 、false、true);
    slot_name | lsn
----------------- -----------
 regression_slot | 0/16B1970
(1行)

postgres=# SELECT slot_name, plugin, slot_type, database, active, restart_lsn, confirmed_flush_lsn FROM pg_replication_slots;
    slot_name | プラグイン | slot_type | データベース | アクティブ | restart_lsn | confirmed_flush_lsn
----------------- ------------ ------------------------------------------------------- + -----------------
 regression_slot | test_decoding | logical | postgres | f | 0/16A4408 | 0/16A4440
(1行)

postgres=# -- まだ変更はありません
postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
 lsn | xid | データ
----- ----- --------
(0行)

postgres=# CREATE TABLEデータ (idシリアルプライマリキー、データテキスト);
テーブルの作成

postgres=# -- DDLはレプリケートされないため、表示されるのはトランザクションだけです
postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
    lsn | xid | データ
---------- --------- ---------------------
 0/BA2DA58 | 10297 | 10297開始
 0/BA5A5A0 | 10297 | COMMIT 10297
(2行)

postgres=# -変更が読み取られると、それらは消費され、放出されません
postgres=# -- その後の呼び出しで:
postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
 lsn | xid | データ
----- ----- --------
(0行)

postgres=# BEGIN;
postgres=*# INSERT INTO data(data) VALUES('1');
postgres=*# INSERT INTOデータ (データ) 値 ('2');
postgres=*# COMMIT;

postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
    lsn | xid | データ
---------- --------- ---------------------------------------------------------
 0/BA5A688 | 10298 | 10298開始
 0/BA5A6F0 | 10298 | table public.data: INSERT: id[integer]:1 data[text]:'1'
 0/BA5A7F8 | 10298 | table public.data: INSERT: id[integer]:2 data[text]:'2'
 0/BA5A8A8 | 10298 | 10298コミット
(4行)

postgres=# INSERT INTOデータ (データ) 値 ('3');

postgres=# -- 変更を消費せずに変更ストリームを事前に覗くこともできます
postgres=# SELECT * FROM pg_logical_slot_peek_changes('regression_slot' 、NULL、NULL);
    lsn | xid | データ
---------- --------- ---------------------------------------------------------
 0/BA5A8E0 | 10299 | 10299開始
 0/BA5A8E0 | 10299 | table public.data: INSERT: id[integer]:3 data[text]:'3'
 0/BA5A990 | 10299 | COMMIT 10299
(3行)

postgres=# -- pg_logical_slot_peek_changes() への次の呼び出しは、同じ変更を再度返します。postgres=# SELECT * FROM pg_logical_slot_peek_changes('regression_slot' 、NULL、NULL);
    lsn | xid | データ
---------- --------- ---------------------------------------------------------
 0/BA5A8E0 | 10299 | 10299開始
 0/BA5A8E0 | 10299 | table public.data: INSERT: id[integer]:3 data[text]:'3'
 0/BA5A990 | 10299 | COMMIT 10299
(3行)

postgres=# -- オプションを出力プラグインに渡して、フォーマットに影響を与えることができます
postgres=# SELECT * FROM pg_logical_slot_peek_changes('regression_slot' 、NULL、NULL、'include-timestamp' 、'on');
    lsn | xid | データ
---------- --------- ---------------------------------------------------------
 0/BA5A8E0 | 10299 | 10299開始
 0/BA5A8E0 | 10299 | table public.data: INSERT: id[integer]:3 data[text]:'3'
 0/BA5A990 | 10299 | COMMIT 10299 (at 2017-05-10 12:07:21.272494-04)
(3行)

postgres=# -- スロットを破壊することを忘れないでください。postgres=# -- サーバーリソース:
postgres=# SELECT pg_drop_replication_slot('regression_slot');
 pg_drop_replication_slot
-----------------------

(1行) 

次の例は、PostgreSQLディストリビューションに含まれているプログラムpg_recvlogicalを使用して、ストリーミングレプリケーションプロトコルで論理的なデコードを制御する方法を示しています。 これには、クライアント認証を設定してレプリケーション接続を許可し、max_wal_sendersを十分に高く設定して追加接続を許可する必要があります。

$ pg_recvlogical -d postgres -- slot=test -- create-slot
    $pg_recvlogical -d postgres -- slot=test -- start -f-
    コントロール + Z
    $psql -d postgres -c "INSERT INTO data(data) VALUES('4');"
    $fg
    BEGIN 693
    table public.data: INSERT: id[integer]:4 data[text]:'4'
    COMMIT 693
    コントロール + C
    $pg_recvlogical -d postgres -- slot=test -- drop-slot