以下はロジカルデコーディングをSQLを使って制御する例です。
ロジカルデコーディングを使う前に、wal_levelをlogical
に、そしてmax_replication_slotsを少なくとも1に設定しなければなりません。
次に、使用するデータベースにスーパーユーザ(以下の例ではpostgres
)として接続します。
postgres=# -- Create a slot named 'regression_slot' using the output plugin 'test_decoding' 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 row) postgres=# SELECT slot_name, plugin, slot_type, database, active, restart_lsn, confirmed_flush_lsn FROM pg_replication_slots; slot_name | plugin | slot_type | database | active | restart_lsn | confirmed_flush_lsn -----------------+---------------+-----------+----------+--------+-------------+----------------- regression_slot | test_decoding | logical | postgres | f | 0/16A4408 | 0/16A4440 (1 row) postgres=# -- There are no changes to see yet postgres=# -- まだ変更はありません postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----+-----+------ (0 rows) postgres=# CREATE TABLE data(id serial primary key, data text); CREATE TABLE postgres=# -- DDL isn't replicated, so all you'll see is the transaction postgres=# -- DDLはレプリケーションされないので、見えるのはトランザクションだけです postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-------+-------------- 0/BA2DA58 | 10297 | BEGIN 10297 0/BA5A5A0 | 10297 | COMMIT 10297 (2 rows) postgres=# -- Once changes are read, they're consumed and not emitted postgres=# -- in a subsequent call: postgres=# -- 変更が読み込まれると、それらは消費され、次の呼び出しで送出されません: postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----+-----+------ (0 rows) postgres=# BEGIN; postgres=*# INSERT INTO data(data) VALUES('1'); postgres=*# INSERT INTO data(data) VALUES('2'); postgres=*# COMMIT; postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-------+--------------------------------------------------------- 0/BA5A688 | 10298 | BEGIN 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 | COMMIT 10298 (4 rows) postgres=# INSERT INTO data(data) VALUES('3'); postgres=# -- You can also peek ahead in the change stream without consuming changes postgres=# -- 変更を消費せずに変更ストリームを先読みすることもできます postgres=# SELECT * FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-------+--------------------------------------------------------- 0/BA5A8E0 | 10299 | BEGIN 10299 0/BA5A8E0 | 10299 | table public.data: INSERT: id[integer]:3 data[text]:'3' 0/BA5A990 | 10299 | COMMIT 10299 (3 rows) postgres=# -- The next call to pg_logical_slot_peek_changes() returns the same changes again postgres=# -- pg_logical_slot_peek_changes()の次の呼び出しでも同じ変更が返されます postgres=# SELECT * FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-------+--------------------------------------------------------- 0/BA5A8E0 | 10299 | BEGIN 10299 0/BA5A8E0 | 10299 | table public.data: INSERT: id[integer]:3 data[text]:'3' 0/BA5A990 | 10299 | COMMIT 10299 (3 rows) postgres=# -- options can be passed to output plugin, to influence the formatting postgres=# -- 出力プラグインにオプションを渡すことで、フォーマットに影響を与えることができます postgres=# SELECT * FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'include-timestamp', 'on'); lsn | xid | data -----------+-------+--------------------------------------------------------- 0/BA5A8E0 | 10299 | BEGIN 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 rows) postgres=# -- Remember to destroy a slot you no longer need to stop it consuming postgres=# -- server resources: postgres=# -- サーバのリソースの消費を停止するためにもう必要ないスロットを破棄することを忘れないでください postgres=# SELECT pg_drop_replication_slot('regression_slot'); pg_drop_replication_slot ----------------------- (1 row)
以下はPostgreSQLに付属するプログラムpg_recvlogicalを用いてロジカルデコーディングをストリーミングレプリケーションのプロトコルによって制御する例です。
この方法を使うには、レプリケーション接続を許すようにクライアント認証を設定し(27.2.5.1参照)、max_wal_senders
を十分に大きくして追加の接続ができるようにしておかなければなりません。
2番目の例は、2相トランザクションをストリームする例です。
2相コマンドを使用する前に、max_prepared_transactionsを少なくとも1に設定する必要があります。
Example 1: $ pg_recvlogical -d postgres --slot=test --create-slot $ pg_recvlogical -d postgres --slot=test --start -f - Control+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 Control+C $ pg_recvlogical -d postgres --slot=test --drop-slot Example 2: $ pg_recvlogical -d postgres --slot=test --create-slot --two-phase $ pg_recvlogical -d postgres --slot=test --start -f - Control+Z $ psql -d postgres -c "BEGIN;INSERT INTO data(data) VALUES('5');PREPARE TRANSACTION 'test';" $ fg BEGIN 694 table public.data: INSERT: id[integer]:5 data[text]:'5' PREPARE TRANSACTION 'test', txid 694 Control+Z $ psql -d postgres -c "COMMIT PREPARED 'test';" $ fg COMMIT PREPARED 'test', txid 694 Control+C $ pg_recvlogical -d postgres --slot=test --drop-slot
以下の例では、準備されたトランザクションのデコードに使用できるSQLインタフェースを示します。
2相コミットコマンドを使用する前に、max_prepared_transactions
を少なくとも1に設定しなければなりません。
pg_create_logical_replication_slot
を使用してスロットを作成する際に、2相パラメータを'true'に設定しておく必要もあります。
トランザクションがまだデコードされていない場合は、コミット後にトランザクション全体をストリームすることに注意してください。
postgres=# BEGIN; postgres=*# INSERT INTO data(data) VALUES('5'); postgres=*# PREPARE TRANSACTION 'test_prepared1'; postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-----+--------------------------------------------------------- 0/1689DC0 | 529 | BEGIN 529 0/1689DC0 | 529 | table public.data: INSERT: id[integer]:3 data[text]:'5' 0/1689FC0 | 529 | PREPARE TRANSACTION 'test_prepared1', txid 529 (3 rows) postgres=# COMMIT PREPARED 'test_prepared1'; postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-----+-------------------------------------------- 0/168A060 | 529 | COMMIT PREPARED 'test_prepared1', txid 529 (4 row) postgres=#-- you can also rollback a prepared transaction postgres=#-- 準備されたトランザクションをロールバックすることもできます postgres=# BEGIN; postgres=*# INSERT INTO data(data) VALUES('6'); postgres=*# PREPARE TRANSACTION 'test_prepared2'; postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-----+--------------------------------------------------------- 0/168A180 | 530 | BEGIN 530 0/168A1E8 | 530 | table public.data: INSERT: id[integer]:4 data[text]:'6' 0/168A430 | 530 | PREPARE TRANSACTION 'test_prepared2', txid 530 (3 rows) postgres=# ROLLBACK PREPARED 'test_prepared2'; postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NULL); lsn | xid | data -----------+-----+---------------------------------------------- 0/168A4B8 | 530 | ROLLBACK PREPARED 'test_prepared2', txid 530 (1 row)