## Chapter 49. Logical Decoding

Postgres Pro Enterprise provides infrastructure to stream the modifications performed via SQL to external consumers. This functionality can be used for a variety of purposes, including replication solutions and auditing.

Changes are sent out in streams identified by logical replication slots.

The format in which those changes are streamed is determined by the output plugin used. An example plugin is provided in the Postgres Pro Enterprise distribution. Additional plugins can be written to extend the choice of available formats without modifying any core code. Every output plugin has access to each individual new row produced by `INSERT` and the new row version created by `UPDATE`. Availability of old row versions for `UPDATE` and `DELETE` depends on the configured replica identity (see `REPLICA IDENTITY`).

Changes can be consumed either using the streaming replication protocol (see Section 53.4 and Section 49.3), or by calling functions via SQL (see Section 49.4). It is also possible to write additional methods of consuming the output of a replication slot without modifying core code (see Section 49.7).

## 49.1. Logical Decoding Examples

The following example demonstrates controlling logical decoding using the SQL interface.

Before you can use logical decoding, you must set wal_level to `logical` and max_replication_slots to at least 1. Then, you should connect to the target database (in the example below, `postgres`) as a superuser.

    postgres=# -- Create a slot named 'regression_slot' using the output plugin 'test_decoding'
    postgres=# SELECT * FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
        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=# 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=# 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=# 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=# 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=# 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=# 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=# SELECT pg_drop_replication_slot('regression_slot');
     pg_drop_replication_slot
    -----------------------

    (1 row)

The following example shows how logical decoding is controlled over the streaming replication protocol, using the program pg_recvlogical included in the Postgres Pro Enterprise distribution. This requires that client authentication is set up to allow replication connections (see Section 26.2.5.1) and that `max_wal_senders` is set sufficiently high to allow an additional connection.

    $ 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