В следующем примере демонстрируется управление логическим декодированием с помощью SQL-интерфейса.

Прежде чем использовать логическое декодирование, необходимо установить параметр wal_level в значение logical и параметр max_replication_slots в значение не менее 1. Затем следует подключиться к целевой базе данных (в примере ниже — postgres ) от имени суперпользователя.

postgres=# -- Создание слота с именем 'regression_slot' с использованием плагина вывода 'test_decoding'
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    |    plugin     | slot_type | database | active | 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 data(id serial primary key, data text);
CREATE TABLE

postgres=# -- DDL не реплицируется, поэтому будет видна только транзакция
postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
    lsn    |  xid  |     данные
-----------+-------+--------------
 0/BA2DA58 | 10297 | BEGIN 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 data(data) VALUES('2');
postgres=*# COMMIT;

postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
    lsn    |  xid  |                          данные
-----------+-------+---------------------------------------------------------
 0/BA5A688 | 10298 | BEGIN 10298
 0/BA5A6F0 | 10298 | таблица public.data: INSERT: id[integer]:1 данные[text]:'1'
 0/BA5A7F8 | 10298 | таблица public.data: INSERT: id[integer]:2 данные[text]:'2'
 0/BA5A8A8 | 10298 | COMMIT 10298
(4 строки)

postgres=# INSERT INTO data(data) VALUES('3');

postgres=# -- Можно также просмотреть поток изменений без их извлечения
postgres=# SELECT * FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL);
    LSN    |  xid  |                          данные
-----------+-------+---------------------------------------------------------
 0/BA5A8E0 | 10299 | BEGIN 10299
 0/BA5A8E0 | 10299 | таблица public.data: INSERT: id[integer]:3 данные[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 | BEGIN 10299
 0/BA5A8E0 | 10299 | таблица public.data: INSERT: id[integer]:3 данные[текст]:'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 | BEGIN 10299
 0/BA5A8E0 | 10299 | таблица public.data: INSERT: id[integer]:3 данные[текст]:'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 строка)

В следующих примерах показано, как управление логическим декодированием осуществляется по протоколу потоковой репликации с помощью программы pg_recvlogical , входящей в состав дистрибутива PostgreSQL. Для этого необходимо настроить аутентификацию клиента для разрешения соединений репликации (см. Раздел 3.11.2.5.1 ) и того, что max_wal_senders было достаточно большим для обеспечения возможности установления дополнительного соединения. Во втором примере показано, как осуществляется потоковая передача двухфазных транзакций. Перед использованием двухфазных команд необходимо установить значение параметра max_prepared_transactions не менее 1.

      Пример 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
таблица public.data: INSERT: id[integer]:4 data[текст]:'4'
COMMIT 693
      Control+C
      $ pg_recvlogical -d postgres --slot=test --drop-slot

Пример 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
таблица 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-интерфейс, который можно использовать для декодирования подготовленных транзакций. Перед использованием команд двухфазной фиксации необходимо установить параметр max_prepared_transactions в значение не менее 1. Также необходимо установить параметр two-phase в значение 'true' при создании слота с помощью pg_create_logical_replication_slot Обратите внимание, что если транзакция еще не была декодирована, она будет передана целиком после выполнения фиксации (commit).

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 |                          данные
-----------+-----+---------------------------------------------------------
 0/1689DC0 | 529 | BEGIN 529
 0/1689DC0 | 529 | таблица public.data: INSERT: id[integer]:3 data[text]:'5'
 0/1689FC0 | 529 | PREPARE TRANSACTION 'test_prepared1', txid 529
(3 строки)

postgres=# COMMIT PREPARED 'test_prepared1';
postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NULL);
    lsn    | xid |                    данные
-----------+-----+--------------------------------------------
 0/168A060 | 529 | COMMIT PREPARED 'test_prepared1', txid 529
(4 строки)

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 |                          данные
-----------+-----+---------------------------------------------------------
 0/168A180 | 530 | BEGIN 530
 0/168A1E8 | 530 | таблица public.data: INSERT: id[integer]:4 данные[текст]:'6'
 0/168A430 | 530 | PREPARE TRANSACTION 'test_prepared2', txid 530
(3 строки)

postgres=# ROLLBACK PREPARED 'test_prepared2';
postgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
    LSN    | xid |                     данные
-----------+-----+----------------------------------------------
 0/168A4B8 | 530 | ROLLBACK PREPARED 'test_prepared2', txid 530
(1 строка)