improve

Author: digoal

201711/20171126_01.md

@@ -79,8 +79,10 @@ create table tbl2 (
 
 ```  
 create table tbl3 (  
-  imei text primary key,       -- 手机唯一标识  
-  tagids int[]      -- 标签数组  
+  imei text primary key,   -- 手机唯一标识  
+  tagids int[],            -- 标签数组  
+  ins_tags int[],          -- 合并操作需要的中间字段
+  del_tags int[]           -- 合并操作需要的中间字段
 );  
   
 create index idx_tbl3_tagids on tbl3 using gin (tagids);  
@@ -107,18 +109,20 @@ select imei from tbl3 where tagids && array[标签1, 标签2] and tagid @> array
 select v8,count(*) from tbl1 where   
   imei = any (array(   
                      select imei from tbl3 where tagids @> array[标签1, 标签2]   
-	     )     )  
+             )     )  
 group by v8;  
 ```  
 
 3、新增或追加标签  
 
+使用intarray插件，简化数组交、并、差操作。  
+   
 ```  
 create extension intarray;  
 ```  
 
 ```  
-insert into tbl3 (imei, tagids) values (?, ?[], 'insert') on conflict (imei) do update set tagids=tbl3.tagids|excluded.tagids;  
+insert into tbl3 (imei, tagids) values (?, ?[]) on conflict (imei) do update set tagids=tbl3.tagids|excluded.tagids;  
 ```  
 
 4、删标签  
@@ -133,38 +137,122 @@ update tbl3 set tagids = tagids - ?[] where imei=?;
 update tbl3 set tagids = ?[] where imei=?;  
 ```  
 
-6、批量并行新增、追加、删除、更新标签注意  
+6、批量并行新增、追加、删除、更新标签优化   
 
 如果要一次性操作很多条记录（例如1000万条记录），并且有并行的贴标签操作（同一条用户被多个SQL更新）。需要注意两个问题：  
 
 6\.1 大事务导致膨胀的问题，建议分段操作。  
 
 6\.2 行锁冲突问题，建议新增（插入），然后合并到标签表。  
 
-优化方法，将直接写标签表，改成写行为日志，任务调度，批量合并：  
+优化方法，  
+   
+实现标签最终一致性。  
+  
+将直接增、删、改标签表，改成写行为日志tag_log，采用任务调度，批量合并到标签表：    
 
 ```  
 create table tag_log (  
   imei text,    -- 手机唯一标识  
-  action text,  -- insert, update, delete  
+  action text,  -- insert, delete  表示增加、删除标签  (更新需求应该没有，如有，直接到标签表操作)
   tagids int[], -- 标签IDs  
   crt_time timestamp default clock_timestamp()   -- 时间     
 );  
   
 create index idx_tag_log_1 on tag_log (crt_time);  
-  
+```
+   
+串行任务，阅后即焚(假设-99999999是一个永远不存在的TAGID)   
+    
+```
 -- CTE语法，支持阅后即焚的批量合并方法  
 with tmp as (delete from tag_log where ctid = any ( array (  
   select ctid from tag_log order by crt_time limit 10000  -- 按时序，批量取1万条  
   )) returning * )   
-, tmp1 as (... 按时序生成合并数据 ...)  
-, ... 写入  
-, ... 更新  
-, ... 删除  
-```  
-  
-阅后即焚的处理速度，每秒百万行。  
-  
+, tmp1 as (select imei, 
+             uniq(sort(array_agg(case when action='insert' then tagids else -99999999 end))) - (-99999999) AS ins_tags,   
+	     uniq(sort(array_agg(case when action='delete' then tagids else -99999999 end))) - (-99999999) AS del_tags
+	   from (select imei, action, unnest(tagids) as tagids from tmp) t group by imei)  
+insert into tbl3 (imei, tagids, ins_tags, del_tags) 
+select imei, ins_tags-del_tags, ins_tags, del_tags from tmp1 
+ on conflict (imei) do update set tagids=((tbl3.tagids | excluded.ins_tags) - excluded.del_tags), ins_tags=excluded.ins_tags, del_tags=excluded.del_tags;  
+```  
+  
+并行任务，阅后即焚     
+   
+```
+例如开启16个并行
+
+abs(mod(hashtext(imei), 16))=?
+```
+  
+```
+-- CTE语法，支持阅后即焚的批量合并方法  
+with tmp as (delete from tag_log where ctid = any ( array (  
+  select ctid from tag_log where abs(mod(hashtext(imei), 16))=0 order by crt_time limit 10000  -- 按时序，批量取1万条，按HASH并行
+  )) returning * )   
+, tmp1 as (select imei, 
+             uniq(sort(array_agg(case when action='insert' then tagids else -99999999 end))) - (-99999999) AS ins_tags,   
+	     uniq(sort(array_agg(case when action='delete' then tagids else -99999999 end))) - (-99999999) AS del_tags
+	   from (select imei, action, unnest(tagids) as tagids from tmp) t group by imei)  
+insert into tbl3 (imei, tagids, ins_tags, del_tags) 
+select imei, ins_tags-del_tags, ins_tags, del_tags from tmp1 
+ on conflict (imei) do update set tagids=((tbl3.tagids | excluded.ins_tags) - excluded.del_tags), ins_tags=excluded.ins_tags, del_tags=excluded.del_tags;  
+```  
+  
+写成函数，方便调用  
+  
+```
+create or replace function consume_tag_log(mo int, mov int, lim int) returns void as $$
+declare
+begin
+  execute format($_$with tmp as (delete from tag_log where ctid = any ( array (  
+  select ctid from tag_log where abs(mod(hashtext(imei), %s))=%s order by crt_time limit %s 
+  )) returning * )   
+, tmp1 as (select imei, 
+             uniq(sort(array_agg(case when action='insert' then tagids else -99999999 end))) - (-99999999) AS ins_tags,   
+	     uniq(sort(array_agg(case when action='delete' then tagids else -99999999 end))) - (-99999999) AS del_tags
+	   from (select imei, action, unnest(tagids) as tagids from tmp) t group by imei)  
+insert into tbl3 (imei, tagids, ins_tags, del_tags) 
+select imei, ins_tags-del_tags, ins_tags, del_tags from tmp1 
+ on conflict (imei) do update set tagids=((tbl3.tagids | excluded.ins_tags) - excluded.del_tags), ins_tags=excluded.ins_tags, del_tags=excluded.del_tags$_$,
+ mo, mov, lim);
+end;
+$$ language plpgsql strict;
+
+
+select consume_tag_log(16,0,10000);   -- 并行处理
+select consume_tag_log(16,1,10000);
+.....
+select consume_tag_log(16,15,10000);
+``` 
+  
+```
+create or replace function consume_tag_log(lim int) returns void as $$
+declare
+begin
+  execute format($_$with tmp as (delete from tag_log where ctid = any ( array (  
+  select ctid from tag_log order by crt_time limit %s
+  )) returning * )   
+, tmp1 as (select imei, 
+             uniq(sort(array_agg(case when action='insert' then tagids else -99999999 end))) - (-99999999) AS ins_tags,   
+	     uniq(sort(array_agg(case when action='delete' then tagids else -99999999 end))) - (-99999999) AS del_tags
+	   from (select imei, action, unnest(tagids) as tagids from tmp) t group by imei)  
+insert into tbl3 (imei, tagids, ins_tags, del_tags) 
+select imei, ins_tags-del_tags, ins_tags, del_tags from tmp1 
+ on conflict (imei) do update set tagids=((tbl3.tagids | excluded.ins_tags) - excluded.del_tags), ins_tags=excluded.ins_tags, del_tags=excluded.del_tags$_$,
+ lim);
+end;
+$$ language plpgsql strict;
+
+
+select consume_tag_log(10000);  -- 每次处理1万条
+```
+  
+创建调度任务，执行消费函数调度即可。    
+   
+**阅后即焚的处理速度，每秒 百万行。**    
+    
 [《(OLTP) 高吞吐数据进出(堆存、行扫、无需索引) - 阅后即焚(读写大吞吐并测)》](../201711/20171107_32.md)    
 
 除了以上基于数组、GIN索引的设计，PostgreSQL还有一些技术，可以用在经营分析系统。  

201711/20171126_02.md

@@ -1,4 +1,4 @@
-## [未完待续] 数据库生命周期管理  
+## [未完待续] 数据库sharding
 
 ### 作者  
 digoal