Logical Decoding by SQL Function Interfaces

Procedure

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

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

   In the preceding command, user1 indicates the username, 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 on all DNs. 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=# EXECUTE DIRECT ON DATANODES 'SELECT * FROM pg_logical_slot_peek_changes(''slot1'', NULL, 4096);'; location | xid | data -----------+-------+------------------------------------------------------------------------------------------------------------------------------------------------- ------------------------------------------- 0/601C188 | 1010023 | BEGIN 1010023 0/601ED60 | 1010023 | COMMIT 1010023 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 CSN 1010023 (5 rows)

6. Delete the logical replication slot slot1.

   1 2 3 4 5

   gaussdb=# SELECT * FROM pg_drop_replication_slot('slot1'); pg_drop_replication_slot -------------------------- (1 row)