发布订阅

1.前置准备

安装两个单节点数据库
配置如下：
ip:192.168.0.20/21
端口：5432
用户：logical

2.修改配置

两个节点创建用户：

create user t1 with replication password 'Huawei@123';
grant all privileges to t1;

修改配置

gs_guc reload -D /data2/logical/data/ -h 'host all all 0.0.0.0/0 sha256'
gs_guc generate -S Huawei@123 -D $GAUSSHOME/bin -o subscription
gs_guc reload -D /data2/logical/data/ -h 'host    replication     t1       0.0.0.0/0         sha256'
gs_guc reload -D /data2/logical/data/ -c 'wal_level=logical'
修改dn目录下postgresql.conf中的参数listen_address为*

3.发布订阅

create database test;
\c test
创建表：
create table b(c1 text, c2 int);
create table a(a int, b varchar);

20节点：
create publication my_publication for all tables;
CREATE SUBSCRIPTION my_subscription CONNECTION 'host=192.168.0.21 port=5433 dbname=test user=t1 password=Huawei@123' PUBLICATION my_publication;
21节点：
create publication my_publication for all tables;
CREATE SUBSCRIPTION my_subscription CONNECTION 'host=192.168.0.20 port=5433 dbname=test user=t1 password=Huawei@123' PUBLICATION my_publication;

查看：
select * from pg_subscription;
select * from pg_publication;

4.测试：

1）DML

a. 插入

20节点：
test=# insert into a values(1,'a');
INSERT 0 1
21节点：
test=# select * from a;
 a | b 
---+---
 1 | a
(1 row)

b.修改、删除

需要设置REPLICA IDENTITY，这里暂时设置为FULL，可自行选择
ALTER TABLE a REPLICA IDENTITY FULL;

20节点：
test=# update a set a=2 where b='a';
UPDATE 1
21节点：
test=# select * from a;
 a | b 
---+---
 2 | a
(1 row)


20节点：
test=# delete from a;
DELETE 1
21节点：
test=# select * from a;
 a | b 
---+---
(0 rows)

2）DDL语句

21节点：
ALTER TABLE a ADD COLUMN c int;
test=# insert into a values(1,'a',1);
INSERT 0 1
test=# select * from a;
 a | b | c 
---+---+---
 1 | a | 1
(1 row)
20节点：
test=# select * from a; 
 a | b 
---+---
(0 rows)



21节点：
test=# drop table b;
DROP TABLE
test=# 
test=# \d
                         List of relations
 Schema | Name | Type  |  Owner  |             Storage              
--------+------+-------+---------+----------------------------------
 public | a    | table | logical | {orientation=row,compression=no}
(1 row)
20节点：
test=# \d
                         List of relations
 Schema | Name | Type  |  Owner  |             Storage              
--------+------+-------+---------+----------------------------------
 public | a    | table | logical | {orientation=row,compression=no}
 public | b    | table | logical | {orientation=row,compression=no}
 public | c    | table | logical | {orientation=row,compression=no}
(3 rows)

查看文档，发现是发布创建时的设置问题，要想支持DDL操作，需要在创建发布时指定(无法通过alter修改)，清理发布和订阅之后重建，重新测试
新建的命令为：
create publication my_publication for all tables with (ddl = 'all');

20节点：
test2=# create table a(a int);
CREATE TABLE
test2=# \d
                         List of relations
 Schema | Name | Type  |  Owner  |             Storage              
--------+------+-------+---------+----------------------------------
 public | a    | table | logical | {orientation=row,compression=no}
(1 row)

21节点：
test2=# \d
                         List of relations
 Schema | Name | Type  |  Owner  |             Storage              
--------+------+-------+---------+----------------------------------
 public | a    | table | logical | {orientation=row,compression=no}
(1 row)




20节点：
test2=# ALTER TABLE a ADD COLUMN c int;
ALTER TABLE
test2=# \d a
       Table "public.a"
 Column |  Type   | Modifiers 
--------+---------+-----------
 a      | integer | 
 c      | integer | 

test2=# insert into a values(1,1);
INSERT 0 1
21节点：
test2=# select * from a;
 a | c 
---+---
 1 | 1
(1 row)


21节点：
test2=# drop table a;
DROP TABLE
20节点：
test2=# \d
No relations found.

3）存储过程

20节点：
create table a(a int);
CREATE OR REPLACE FUNCTION insert_into_a(value INTEGER)
RETURNS VOID AS $$
BEGIN
    -- 插入新记录到表 a
    INSERT INTO public.a (a) VALUES (value);
    RAISE NOTICE 'Inserted value: %', value;
END;
$$ LANGUAGE plpgsql;

test2=# SELECT insert_into_a(10);
NOTICE:  Inserted value: 10
CONTEXT:  referenced column: insert_into_a
 insert_into_a 
---------------
 
(1 row)

21节点：
test2=# select * from a;
 a  
----
 10
(1 row)