配置

前提条件：源库配置

• 修改postgresql.conf文件

#逻辑级别日志
wal_level = logical
#设置日志保留个数，设置太少了，日志量太大了日志自动删除后，容易读不到日志
注：postgresql 13.0中，将wal_keep_segments重命名为wal_keep_size
wal_keep_segments = 10
#如果需要远程读日志，需要设置，至少为1，就接受1个读日志线程
max_wal_senders = 10

验证是否成功

select name,setting,category from pg_settings where name in ('wal_level','wal_keep_segments','max_wal_senders');

• 查询数据库或者zcbus用户加密方式

  show password_encryption；
  select rolname,rolpassword from pg_authid where rolname='zcbus';

  
• 如果远程读日志，需要设置读日志的权限，修改pg_hba.conf

host    replication     all             0.0.0.0/32            md5 #加密方式需要和添加的zcbus用户加密方式一致

• 修改pg_hba.conf执行pg_ctl命令生效

pg_ctl reload -D 指定pg_hba.conf所在目录
示例：pg_ctl reload -D /opt/postgresql/data

• 验证是否成功

select * from pg_hba_file_rules;

以下过程中，web页面不要点击注册按键，所有配置手工完成！

添加槽位号说明（如发布数据库未配置SLOT_NAME参数，则此步骤添加槽位号跳过）

1、创建槽位号
查看槽位号信息
  select * from pg_replication_slots;
2、创建槽位号信息,槽位名字:zcbus_slot_name
# PG系列创建槽位号
sql命令：
  添加：select pg_create_physical_replication_slot('zcbus_slot_name');
  删除：SELECT pg_drop_replication_slot('zcbus_slot_name');
系统命令：
  添加：pg_receivewal --create-slot --slot=zcbus_slot_name
  删除：pg_receivewal --drop-slot --slot=zcbus_slot_name
# OpenGaussDB创建槽位号
sql命令：
  添加：select pg_create_physical_replication_slot('zcbus_slot_name',false);
  删除：SELECT pg_drop_replication_slot('zcbus_slot_name');

3、配置页面槽位号信息
  切换POSTGRESQL日志操作
  旧版本：select pg_switch_xlog();
  新版本：select pg_switch_wal();
  ZCBUS发布配置增加SLOT_NAME参数，自动匹配指定槽位号进行日志读取。
4、监控日志即可

注：删除日志参考：
正常删除pg日志方式
1、检查可以删除wal的日志文件
83e44cfdd569:~/data$ pg_controldata -D /var/lib/postgresql/data/ |grep "REDO WAL file"
Latest checkpoint's REDO WAL file:    000000010000000100000067
2、删除日志文件操作
pg_archivecleanup /var/lib/postgresql/data/pg_wal/ 000000010000000100000067

配置说明

用postgres系统账号登录pg库，执行如下(数据库命令行执行)：

create user zcbus with password  'zcbus';
grant connect on database db01 to zcbus;
create schema zcbus AUTHORIZATION zcbus;

切换到同步库下，创建ddl_event_tbl表与zcbus_func_ddl_command函数

create table zcbus.ddl_event_tbl (  
event text,
tag text, 
objid oid,
object_type text,  
schema_name text,  
object_identity text,
query text,
op_time timestamp default now()
);

检查ddl_event_tbl表是否创建：

select * from information_schema.tables where table_schema='zcbus' and table_type='BASE TABLE' and table_name='ddl_event_tbl';

alter table "zcbus"."ddl_event_tbl" REPLICA IDENTITY FULL;

create or replace function zcbus.zcbus_func_ddl_command() returns event_trigger as $$  
declare  
  v1 text;
  r record;
begin
  select query into v1 from pg_stat_activity where pid=pg_backend_pid();
  -- RAISE NOTICE 'ddl event:%, command:%', tg_event, tg_tag; 
  if TG_EVENT='ddl_command_end' then  
    SELECT * into r FROM pg_event_trigger_ddl_commands();
    if r.classid > 0 then
      insert into zcbus.ddl_event_tbl(event, tag, objid, object_type, schema_name, object_identity, query)  
        values(TG_EVENT, TG_TAG, r.objid, r.object_type, r.schema_name, r.object_identity, v1);
    end if;
  end if;
  if TG_EVENT='sql_drop' then
    if TG_TAG != 'ALTER TABLE' then
      SELECT * into r FROM pg_event_trigger_dropped_objects();
      insert into zcbus.ddl_event_tbl(event, tag, object_type, schema_name, object_identity, query)  
        values(TG_EVENT, TG_TAG, r.object_type, r.schema_name, r.object_identity, v1);
    end if;
  end if;
end;  
$$ language plpgsql strict;

检查function是否创建：

select a.*
FROM pg_catalog.pg_proc a
JOIN pg_catalog.pg_namespace b ON a.pronamespace = b.oid
WHERE a.proname = 'zcbus_func_ddl_command' and b.nspname='zcbus'

CREATE EVENT TRIGGER zcbus_et_ddl_command on ddl_command_end EXECUTE PROCEDURE zcbus.zcbus_func_ddl_command();
CREATE EVENT TRIGGER zcbus_et_ddl_drop on sql_drop EXECUTE PROCEDURE zcbus.zcbus_func_ddl_command();

检查EVENT TRIGGER是否创建：

select * from  pg_event_trigger  where evtname in ('zcbus_et_ddl_drop','zcbus_et_ddl_command')

grant select,trigger on table zcbus.ddl_event_tbl  to zcbus;
grant execute ON FUNCTION zcbus.zcbus_func_ddl_command to zcbus;
alter user zcbus REPLICATION;

**** 切换到db01库下执行

\c db01
grant select on all tables in schema public to zcbus;
grant execute on function pg_hba_file_rules to zcbus;

**** 添加新表，需要postgres账号开启表的补充日志模式(切换到同步库执行)，并且要再次执行select权限

ALTER TABLE public.tb02 REPLICA IDENTITY FULL;
grant select on all tables in schema public to zcbus;
grant usage on schema public to zcbus;

**** 添加新表，根据schemaname.tablename，确认postgres账号开启表的补充日志模式状态开启成功：

select relreplident from pg_class c , pg_tables tab where tab.schemaname = 'schemaname' and c.relname='tablename';

查询结果说明：

d = 默认(主键)
n = 无
f = 所有列
i = 索引的indisreplident被设置或者默认
查询结果为f，则此表满足同步条件，上一步postgres账号开启表的补充日志模式成功

**** 回退方式：

drop owned by zcbus cascade;
drop user zcbus;

查看用户的表权限

select * from information_schema.table_privileges where grantee='zcbus';

查看usage权限表

select * from information_schema.usage_privileges where grantee='zcbus';

查看存储过程函数相关权限

select * from information_schema.routine_privileges where grantee='zcbus';

开启表REPLICA IDENTITY参数

开启REPLICA IDENTITY

\c database_name;
ALTER TABLE schema_name.table_name REPLICA IDENTITY FULL;

生成修改表sql语句参考

SELECT 'ALTER TABLE '||nspname||'.'||relname||' REPLICA IDENTITY FULL;' FROM pg_class c,pg_namespace n where relnamespace = n.oid and c.relkind in ('r','p') and nspname='public';

注：表所属用户或者超级用户才有开启REPLICA IDENTITY权限，否则报错：must be owner of table

查询REPLICA IDENTITY状态

SELECT c.relreplident FROM pg_class c,pg_namespace n where relnamespace = n.oid and c.relkind in ('r','p') and nspname='schema_name' and relname='table_name'

状态说明：

• d = DEFAULT - 更新和删除将包含primary key列的现前值
• n = NOTHING - 更新和删除将不包含任何先前值
• f = FULL - 更新和删除将包含所有列的先前值
• i = INDEX index name -?更新和删除事件将包含名为index name的索引定义中包含的列的先前值。

查看分区表是否打开全列附加日志:

SELECT
   parent.relname AS partitioned_table_name,
   parent.relkind AS partitioned_table_type,
   parent.relpartbound AS partition_bound,
   child.relname,
   child.relkind,
   child.relreplident,
   'ALTER TABLE ' || nspname || '.' || child.relname || ' REPLICA IDENTITY FULL' rif
FROM
 pg_class parent
LEFT JOIN pg_inherits ON pg_inherits.inhparent = parent.oid
LEFT JOIN pg_class child ON pg_inherits.inhrelid = child.oid
LEFT JOIN pg_namespace pns ON child.relnamespace = pns.oid
WHERE
 parent.relkind = 'p'
AND parent.relname = 'energy_meter_info'

注：

#连接opengaussdb报“Invalid or unsupported by client SCRAM mechanisms”处理方法
使用show password_encryption_type可以查看密码的加密方式。当password_encryption_type=1是表示采用sha256方式对密码加密，兼容postgres客户端的MD5用户认证方式。password_encryption_type=2表示采用sha256方式对密码加密，此时不兼容postgres开源驱动和开源客户端。使用postgres开源生态工具连接会出现报错，连接GaussDB(DWS)出现如下报错：
The authentication type 10 is not supported.
Invalid or unsupported by client SCRAM mechanisms。
解决方法：
gs_guc reload -Z coordinator -Z datanode -N all -I all -c “password_encryption_type=1”
这个动作需要对方许可才行