Logical Decoding by SQL Functions

Procedure

1. Log in to any primary DN in the GaussDB cluster as a user with the REPLICATION permission.
2. Connect to the database through a DN port.

   gsql -U user1 -W password -d gaussdb -p 40000 -r

   In the preceding command, user1 indicates the username, password indicates the user password, gaussdb indicates the name of the database to be connected, and 40000 indicates the database DN port number. You can replace them as required. Replication slots are created on DNs. Therefore, you need to connect to a database through a DN port.

3. Create a logical replication slot named slot1.

   1 2 3 4 5

   gaussdb=> SELECT * FROM pg_create_logical_replication_slot('slot1', 'mppdb_decoding'); slotname | xlog_position ----------+--------------- slot1 | 0/601C150 (1 row)

4. Create a table t in the database and insert data into it.

   1 2	gaussdb=> CREATE TABLE t(a int PRIMARY KEY, b int); gaussdb=> INSERT INTO t VALUES(3,3);

5. Read the decoding result of slot1. The number of decoded records is 4096.
   

   For details about the logical decoding options, see Logical Decoding Options.

   1 2 3 4 5 6 7 8 9 10

   gaussdb=> SELECT * FROM pg_logical_slot_peek_changes('slot1', NULL, 4096); location | xid | data -----------+-------+------------------------------------------------------------------------------------------------------------------------------------------------- ------------------------------------------- 0/601C188 | 1010023 | BEGIN 1010023 0/601ED60 | 1010023 | COMMIT 1010023 (at 2023-09-14 16:03:51.394287+08) CSN 1010022 0/601ED60 | 1010024 | BEGIN 1010024 0/601ED60 | 1010024 | {"table_name":"public.t","op_type":"INSERT","columns_name":["a","b"],"columns_type":["integer","integer"],"columns_val":["3","3"],"old_keys_name":[],"old_keys_type":[],"old_keys_val":[]} 0/601EED8 | 1010024 | COMMIT 1010024 (at 2023-09-14 16:03:57.239821+08) CSN 1010023 (5 rows)

6. Delete the logical replication slot slot1 and delete the service table t.

   1 2 3 4 5 6 7 8

   gaussdb=> SELECT * FROM pg_drop_replication_slot('slot1'); pg_drop_replication_slot -------------------------- (1 row) gaussdb=> DROP TABLE t; DROP TABLE