PostgreSQL、Greenplum DML合并操作 最佳实践
背景
在很多场景中会涉及到数据的合并，比如
1. 某业务系统的总用户有1亿，每天的活跃用户有100万，新增用户10万，每天需要将新增、活跃用户的数据（比如他们的余额变化、等等）合并到数据仓库的用户信息表。
2. 物化视图，某个表被用户不断的增、删、改。需要将这个表（基表）的某些字段或者某部分数据提取到一个物化视图中。这个物化视图不需要对每一笔基表的DML都实施操作，比如对单条记录的操作，合并成一次操作。
3. 数据同步，将OLTP的数据，同步到OLAP系统，由于OLAP系统的事务处理能力没有TP系统强，所以也必须采用合并的方法，同一条记录被多次更新时，需要将多次更新合并成一次更新。
4. 基于REDO日志的逻辑数据复制，优化手段除了并行复制，还有一种就是合并复制。
不管是哪种数据合并，被合并的表最好是有主键的，本文也假设有主键来处理。否则会增加复杂度（需要使用整行记录来区分），而且整行记录有一个缺陷，例如根据行号定位重复记录中的一条，这样变更后，合并时可能会出错。
数据合并的方法
对于以上几种情况，比较复杂的是逻辑数据复制，它可能涉及到任意操作，单个KEY可能被删除，后续这个KEY又被插入、多次更新的情况。
同时还需要考虑事务一致性的问题，每一次合并操作都需要保证一致性。例如基于REDO的逻辑复制，对于未结束的事务产生的REDO，不能参与合并。
保证单个KEY，在合并时只操作一次，同时确保未结束的事务不参与合并。
REDO要素
table : 库\schema\表名
old : 主键值
new : 新插入的值 、 被变更的字段变更后的值
tag : insert 、 update 、 delete 、 truncate
例子
以逻辑复制为例，分解一下数据合并的过程。
创建测试表
create table tbl (pk1 int, pk2 int, c1 int, c2 text, crt_time timestamp, primary key(pk1,pk2));
产生一些DML
...忽略中间部分, 假设忽略的这部分已经同步到目标表...
delete from tbl where pk1=2 and pk2=2; -- 说明目标表已经存在pk1=2,pk2=2的记录
insert into tbl (pk1,pk2,c1,c2,crt_time) values (2,2,2,'test22','2017-02-14');
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,1,2,'test23','2017-02-14');
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,2,3,'test34','2017-02-14');
update tbl set c2='new', crt_time=null where pk1=1 and pk2=2;
delete from tbl where pk1=1 and pk2=1;
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,1,5,'test56','2017-02-14');
update tbl set c2='new11', crt_time=null where pk1=1 and pk2=1;
update tbl set c2='new12', crt_time='2017-02-15' where pk1=1 and pk2=2;
将以上DML转换为REDO要素如下（通常以下信息可以在数据库的REDO日志得到）
old: pk1=2,pk2=2
new: null
tag: delete
old: null
new: pk1=2,pk2=2,c1=2,c2='test22',crt_time='2017-02-14'
tag: insert
old: null
new: pk1=1,pk2=1,c1=2,c2='test23',crt_time='2017-02-14'
tag: insert
old: null
new: pk1=1,pk2=2,c1=3,c2='test34',crt_time='2017-02-14'
tag: insert
old: pk1=1,pk2=2
new: c2='new',crt_time=null
tag: update
old: pk1=1,pk2=1
new: null
tag: delete
old: null
new: pk1=1,pk2=1,c1=5,c2='test56',crt_time='2017-02-14'
tag: insert
old: pk1=1,pk2=1
new: c2='new11',crt_time=null
tag: update
old: pk1=1,pk2=2
new: c2='new12',crt_time='2017-02-15'
tag: update
合并过程，对已提交的记录，按PK进行分组，按执行先后顺序排序
delete from tbl where pk1=2 and pk2=2; -- 说明目标表已经存在pk1=2,pk2=2的记录
insert into tbl (pk1,pk2,c1,c2,crt_time) values (2,2,2,'test22','2017-02-14');
合并后，目标表应该仅仅执行如下SQL
update tbl set c1=2,c2='test22',crt_time='2017-02-14' where pk1=2 and pk2=2;
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,1,2,'test23','2017-02-14');
delete from tbl where pk1=1 and pk2=1;
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,1,5,'test56','2017-02-14');
update tbl set c2='new11', crt_time=null where pk1=1 and pk2=1;
合并后，目标表应该仅仅执行如下SQL
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,1,5,'new11',null);
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,2,3,'test34','2017-02-14');
update tbl set c2='new', crt_time=null where pk1=1 and pk2=2;
update tbl set c2='new12', crt_time='2017-02-15' where pk1=1 and pk2=2;
合并后，目标表应该仅仅执行如下SQL
insert into tbl (pk1,pk2,c1,c2,crt_time) values (1,2,3,'new12','2017-02-15');
如果涉及到PK的变更，需要将其分解为delete和insert两条
例如
old: pk1=1,pk2=1
new: pk1=1,pk2=3,c2='new11',crt_time=null
tag: update
需要分解为
delete from tbl where pk1=1,pk2=1;
insert into tbl (pk1,pk2,c2,crt_time) values (1,3,'new11',null);
OLAP数据合并更新、删除的例子1
逻辑复制的合并相对来说比较复杂，但是PostgreSQL是一个功能强大的数据库，它支持窗口查询，编程能力强大的plpgsql函数语言（还有python, java, perl等数据库函数语言），使用SQL还是比较方便的可以完成以上合并的。
除了逻辑复制，在OLAP中也经常要用到合并更新，主要的目的是减少OLAP系统SQL的执行次数（因为OLAP系统并不是为TP业务设计，而是为批处理或大量运算设计的，多次SQL如果能合并成一次的话，可以大幅提升效率）
下面就以更新的合并为例，简单的讲解一下数据合并的例子。
比如一张表有1亿记录，每天要更新其中的10万条记录。我们要做的是将10万条UPDATE语句，合并成一条UPDATE语句。
合并方法
1. 首先将更新语句转换为数据，插入一张临时表
2. 然后使用join update来更新目标表
过程如下
创建一个生产表（目标表，必须有PK），假设它有1亿（为了演示，仅使用100万记录）用户数据。
create table prod(id int primary key, c1 int, info text, crt_time timestamp, mod_time timestamp);
创建一张临时表，用来存储合并前的DML，表结构如下，需要包含一个新增的序列PK，以及目标表的所有字段，以及每个字段对应的SET位（表示该字段是否被更新）
我们这里假设一条记录，可能被多次更新。
create table tmp1
(
pk serial8 primary key, -- 标记插入顺序
id int, c1 int, info text, crt_time timestamp, mod_time timestamp, -- 更新后的值
set_id boolean, set_c1 boolean, set_info boolean, set_crt_time boolean, set_mod_time boolean -- 被更新的字段
);
插入100万数据到prod表
insert into prod select generate_series(1,1000000), 1, 'test', now(), null;
TP系统中的UPDATE语句，我们将它转换为目标值，插入临时表
insert into tmp1 (id,c1,info,crt_time,mod_time,set_id,set_c1,set_info,set_crt_time,set_mod_time)
select random()*10000, 2, null, null, clock_timestamp(), true,true,false,false,true from generate_series(1,10000); -- c1=2, mod_time=clock_timestamp()
insert into tmp1 (id,c1,info,crt_time,mod_time,set_id,set_c1,set_info,set_crt_time,set_mod_time)
select random()*10000, 3, 'new', null, clock_timestamp(), true,true,true,false,true from generate_series(1,10000); -- c1=3, info='new', mod_time=clock_timestamp()
insert into tmp1 (id,c1,info,crt_time,mod_time,set_id,set_c1,set_info,set_crt_time,set_mod_time)
select random()*10000, null, 'new1', null, clock_timestamp(), true,false,true,false,true from generate_series(1,10000); -- info='new1', mod_time=clock_timestamp()
insert into tmp1 (id,c1,info,crt_time,mod_time,set_id,set_c1,set_info,set_crt_time,set_mod_time)
select random()*10000, 5, null, null, clock_timestamp(), true,true,true,false,true from generate_series(1,10000); -- c1=5, info=null, mod_time=clock_timestamp()
true大于false，将用于多条记录的合并
postgres=# select true > false;
?column?
----------
t
(1 row)
如果一条记录被多次UPDATE，需要将多个UPDATE合并为一个UPDATE
用到了窗口查询，以目标表的PK为分组，按不同字段的set位优先取true的最后一条值，以及它的set状态。
select id, c1, set_c1, info, set_info, crt_time, set_crt_time, mod_time, set_mod_time from
(
select
row_number() over (partition by id) as rn,
id,
first_value(c1) over (partition by id order by set_c1 desc, pk desc) c1,
first_value(set_c1) over (partition by id order by set_c1 desc, pk desc) set_c1,
first_value(info) over (partition by id order by set_info desc, pk desc) info,
first_value(set_info) over (partition by id order by set_info desc, pk desc) set_info,
first_value(crt_time) over (partition by id order by set_crt_time desc, pk desc) crt_time,
first_value(set_crt_time) over (partition by id order by set_crt_time desc, pk desc) set_crt_time,
first_value(mod_time) over (partition by id order by set_mod_time desc, pk desc) mod_time,
first_value(set_mod_time) over (partition by id order by set_mod_time desc, pk desc) set_mod_time
from tmp1
) t
where t.rn=1;
以上就是合并后的数据
更新时，使用case，将字段set位为true的值更新为新的值，false的不变。
update prod set
c1=(case when t.set_c1 then t.c1 else prod.c1 end) , -- 将字段set位为true的值更新为新的值，false的不变。
info=(case when t.set_info then t.info else prod.info end) ,
crt_time=(case when t.set_crt_time then t.crt_time else prod.crt_time end) ,
mod_time=(case when t.set_mod_time then t.mod_time else prod.mod_time end)
from
(
select id, c1, set_c1, info, set_info, crt_time, set_crt_time, mod_time, set_mod_time
from
(
select
row_number() over (partition by id) as rn,
id,
first_value(c1) over (partition by id order by set_c1 desc, pk desc) c1,
first_value(set_c1) over (partition by id order by set_c1 desc, pk desc) set_c1,
first_value(info) over (partition by id order by set_info desc, pk desc) info,
first_value(set_info) over (partition by id order by set_info desc, pk desc) set_info,
first_value(crt_time) over (partition by id order by set_crt_time desc, pk desc) crt_time,
first_value(set_crt_time) over (partition by id order by set_crt_time desc, pk desc) set_crt_time,
first_value(mod_time) over (partition by id order by set_mod_time desc, pk desc) mod_time,
first_value(set_mod_time) over (partition by id order by set_mod_time desc, pk desc) set_mod_time
from tmp1
) t
where t.rn=1
) t
where prod.id=t.id;
删除更加简单，只需要将ID记录下来，delete from tbl where id in (…)即可，不再列举。
验证以上合并方法的一致性
使用两张目标表，一张为合并更新（合并更新的数据来源于实时更新的触发器日志），一张为实时更新。
drop table IF EXISTS prod;
drop table IF EXISTS prod_ck;
drop table IF EXISTS tmp1;
create table prod(id int primary key, c1 int, info text, crt_time timestamp, mod_time timestamp);
insert into prod select generate_series(1,1000000), 1, 'test', now(), null;
create table prod_ck(id int primary key, c1 int, info text, crt_time timestamp, mod_time timestamp);
insert into prod_ck select * from prod;
create table tmp1
(
pk serial8 primary key, -- 标记插入顺序
id int, c1 int, info text, crt_time timestamp, mod_time timestamp, -- 更新后的值
set_id boolean, set_c1 boolean, set_info boolean, set_crt_time boolean, set_mod_time boolean -- 被更新的字段
);
create or replace function f_tg() returns trigger as $$
declare
begin
insert into tmp1 (id,c1,info,crt_time,mod_time,set_id,set_c1,set_info,set_crt_time,set_mod_time) values
(NEW.id, NEW.c1, NEW.info, NEW.crt_time, NEW.mod_time, true, true, true, true, true);
return null;
end;
$$ language plpgsql strict;
create trigger tg after update on prod_ck for each row execute procedure f_tg();
使用pgbench，不断更新prod_ck
vi test.sql
\set id random(1,10000)
\set c1 random(1,1000000)
update prod_ck set c1=:c1, crt_time=clock_timestamp(), mod_time=null where id=:id;
update prod_ck set c1=:c1, crt_time=null, mod_time=clock_timestamp() where id=:id+1;
pgbench -M prepared -n -r -P 1 -f ./test.sql -c 32 -j 32 -T 30
progress: 3.0 s, 26379.3 tps, lat 1.211 ms stddev 0.660
progress: 4.0 s, 26339.9 tps, lat 1.213 ms stddev 0.700
将tmp1的更新合并到prod
update prod set
c1=(case when t.set_c1 then t.c1 else prod.c1 end) , -- 将字段set位为true的值更新为新的值，false的不变。
info=(case when t.set_info then t.info else prod.info end) ,
crt_time=(case when t.set_crt_time then t.crt_time else prod.crt_time end) ,
mod_time=(case when t.set_mod_time then t.mod_time else prod.mod_time end)
from
(
select id, c1, set_c1, info, set_info, crt_time, set_crt_time, mod_time, set_mod_time
from
(
select
row_number() over (partition by id) as rn,
id,
first_value(c1) over (partition by id order by set_c1 desc, pk desc) c1,
first_value(set_c1) over (partition by id order by set_c1 desc, pk desc) set_c1,
first_value(info) over (partition by id order by set_info desc, pk desc) info,
first_value(set_info) over (partition by id order by set_info desc, pk desc) set_info,
first_value(crt_time) over (partition by id order by set_crt_time desc, pk desc) crt_time,
first_value(set_crt_time) over (partition by id order by set_crt_time desc, pk desc) set_crt_time,
first_value(mod_time) over (partition by id order by set_mod_time desc, pk desc) mod_time,
first_value(set_mod_time) over (partition by id order by set_mod_time desc, pk desc) set_mod_time
from tmp1
) t
where t.rn=1
) t
where prod.id=t.id;
UPDATE 10001
验证合并更新后prod和prod_ck是否一致
postgres=# select sum(hashtext(t.*::text)) from prod t;
sum
----------------
-2538529730583
(1 row)
postgres=# select sum(hashtext(t.*::text)) from prod_ck t;
sum
----------------
-2538529730583
(1 row)
小结
数据合并的目标是将多条DML语句合并成一条，
包括将单条记录的多次更新、插入、删除合并为一次更新、插入或删除操作，
也包括将多条记录的多次DML合并成一条DML语句。
在数据逻辑复制、TP到AP业务系统的同步、物化视图 等场景有着广泛的应用。
特别是OLAP系统，由于并不是针对TP场景涉及，使用合并操作，可以大幅提升AP系统的操作效率。(Greenplum更新和删除都是表级锁, 效率也一般)
在greenplum单条记录，基于PK的更新速度测试
pgbench -M simple -n -r -f ./test.sql -P 1 -c 4 -j 4 -T 100 -h 127.0.0.1 -p 29999 -U digoal
progress: 1.0 s, 203.0 tps, lat 19.449 ms stddev 29.442
progress: 2.0 s, 290.0 tps, lat 13.739 ms stddev 0.337
progress: 3.0 s, 291.0 tps, lat 13.763 ms stddev 0.627
progress: 4.0 s, 276.0 tps, lat 14.035 ms stddev 3.919
progress: 5.0 s, 280.0 tps, lat 14.791 ms stddev 5.954
progress: 6.0 s, 296.0 tps, lat 13.493 ms stddev 1.720
progress: 7.0 s, 300.0 tps, lat 13.347 ms stddev 1.433
1万次更新需要十几多秒才能完成，而使用合并更新，只需要0.几秒
而插入方面，Greenplum没有表级锁，效率比更新高很多。
^C
pgbench -M simple -n -r -f ./test.sql -P 1 -c 2 -j 2 -T 100
progress: 1.0 s, 11124.9 tps, lat 0.178 ms stddev 0.018
progress: 2.0 s, 10910.9 tps, lat 0.182 ms stddev 0.011
progress: 3.0 s, 10903.3 tps, lat 0.182 ms stddev 0.011
^C
pgbench -M simple -n -r -f ./test.sql -P 1 -c 1 -j 1 -T 100
progress: 1.0 s, 4998.3 tps, lat 0.170 ms stddev 0.018
progress: 2.0 s, 4041.6 tps, lat 0.282 ms stddev 7.163
progress: 3.0 s, 5894.7 tps, lat 0.169 ms stddev 0.006
^C
pgbench -M simple -n -r -f ./test.sql -P 1 -c 8 -j 8 -T 100
progress: 1.0 s, 39995.8 tps, lat 0.198 ms stddev 0.146
progress: 2.0 s, 41325.8 tps, lat 0.192 ms stddev 0.037
progress: 3.0 s, 40775.6 tps, lat 0.195 ms stddev 0.149
progress: 4.0 s, 41638.2 tps, lat 0.191 ms stddev 0.046