PostgreSQL flashback query 闪回功能 emulate by trigger

我之前写过一篇关于利用PostgreSQL mvcc特性来模拟闪回, 找回数据的文章, 有兴趣的同学可以参考如下 :

<a target="_blank" href="http://blog.163.com/digoal@126/blog/static/163877040201251911813661/">http://blog.163.com/digoal@126/blog/static/163877040201251911813661/</a>

使用以上方法需要担心的一个问题是数据可能被VACUUM掉了, 所以我们可以定制表的autovacuum threshold, 让重点保护的表的autovacuum 阈值较大, 减少VACUUM的间隔, 或者关闭重点保护的表的autovacuum, 改为人为调度VACUUM. 或者改vacuum_defer_cleanup_age参数, 延迟多少个事务之后再回收.

但是使用以上方法也不适合生产, 因为还要停库, 改控制文件, 都是非常危险的操作. (当然你如果为了找回重要数据, 那么拿备库来做也是值得考虑的.)

如果可以在会话层欺骗数据库当前未分配事务号, 最早已提交事务号的话, 其实就不需要修改这么麻烦了. 当然这个就需要改代码了, 因为这部分数据在共享内存区, 直接改的话危险系数太高, 想办法搞成会话层面的吧还好一点.

本文要介绍另一种闪回方法, 触发器.

步骤如下 :

1. 首先要记录所有的DML以及truncate. 也就是对于insert, update, delete, truncate操作, 我们可以回退. 通过触发器来记录old value, new value.

2. 需要闪回的表必须有PK, PK列可以被更新. 如果没有PK的话, 不能唯一的定位到一条记录. 因为PG的行号无法定位到一条记录, 一条记录一旦被更新, 是会生成一个新版本的.

3. INSERT的UNDO, delete where pk=NEW.pk

UPDATE的UNDO, UPDATE set cols=OLD.* where pk=NEW.pk

DELETE和TRUNCATE的UNDO, insert into values (OLD.*)

4. 表的SCHEMA可能会变, 表名可能会变, 列的类型可能会变, 可能会新增列, 可能会删除列.

这些都必须考虑, 因为DDL不被跟踪. 所以我们不直接记录UNDO_SQL, 而是在UNDO时根据当前的数据定义来组装SQL. 并且本方法也不支持DDL的闪回.

需要DDL的闪回, 或者完美的闪回, 请使用PITR.

同时, 为了区分需要闪回的表, 我们不能把跟踪记录放在同一个表里面用schema和tablename来区分, 因为schema和tablename可能被DDL改掉, 那么就会造成取不到记录的情况. 例如TIME1, A表,执行了一些DML后, 改名为B表了, 有执行了一些DML, 然后我们要回退到TIME1的时间点, 根据当前表名B, 从统一的跟踪表undo_table里面取记录的话, 需要告诉跟踪表名字为B, XID为?然后取数据拼装UNDO SQL, 这样的话表名为A的记录时取不出来的, 因为过滤条件是tablename=B. 所以跟踪表要每个表各自一个.

tablea, undo_tablea, tableb, undo_tableb.....这样就不管表或者SCHEMA怎么变了.

注意我们不使用hstore来存储被跟踪表的记录, 原因是回退的时候很麻烦, hstore没有直接转换成record的接口. 我们直接使用表的复合类型来存储被跟踪表的记录.

例子 :

为了增加复杂度, 我们使用大写表名, 列名.

create table public."TBL" (

c1 int,

c2 int,

"C3" text,

c4 text,

c5 text,

c6 text,

c7 int,

crt_time timestamp,

primary key (c1,"C3",c6,c4)

);

创建记录表, 跟踪表的DML和truncate. 可以增加一列txid_snapshot类型存储txid_current_snapshot(), 这样就能回退到一个一致的点了.

CREATE TABLE public.undo_t (

id serial8 primary key,

xid int8,

relid oid,

table_schema text,

table_name text,

when_tg text,

level text,

op text,

encoding name,

old_rec public."TBL",

new_rec public."TBL",

crt_time timestamp without time zone DEFAULT now(),

username text,

client_addr inet,

client_port int

创建触发器函数, 将DML, TRUNCATE的数据插入跟踪表

CREATE OR REPLACE FUNCTION public.undo_t_trace()

RETURNS trigger

LANGUAGE plpgsql

AS $BODY$

DECLARE

v_username text := session_user;

v_client_addr inet := inet_client_addr();

v_client_port int := inet_client_port();

v_xid bigint := txid_current(); -- 记录事务号, 回退时以事务号为界限.

v_encoding name := pg_client_encoding();

BEGIN

case TG_OP

when 'DELETE' then

insert into public.undo_t (xid, relid, table_schema, table_name, when_tg, level, op, encoding, old_rec, username, client_addr, client_port)

values (v_xid, tg_relid, tg_table_schema, tg_table_name, tg_when, tg_level, tg_op, v_encoding, OLD, v_username, v_client_addr, v_client_port);

when 'INSERT' then

insert into public.undo_t (xid, relid, table_schema, table_name, when_tg, level, op, encoding, new_rec, username, client_addr, client_port)

values (v_xid, tg_relid, tg_table_schema, tg_table_name, tg_when, tg_level, tg_op, v_encoding, NEW, v_username, v_client_addr, v_client_port);

when 'UPDATE' then

insert into public.undo_t (xid, relid, table_schema, table_name, when_tg, level, op, encoding, old_rec, new_rec, username, client_addr, client_port)

values (v_xid, tg_relid, tg_table_schema, tg_table_name, tg_when, tg_level, tg_op, v_encoding, OLD, NEW, v_username, v_client_addr, v_client_port);

when 'TRUNCATE' then

select v_xid, tg_relid, tg_table_schema, tg_table_name, tg_when, tg_level, tg_op, v_encoding, t, v_username, v_client_addr, v_client_port from public."TBL" AS t;

else

return null;

end case;

RETURN null;

END;

$BODY$ strict volatile;

添加触发器, 记录表的dml和truncate.

CREATE TRIGGER tg1 AFTER DELETE or INSERT or UPDATE ON public."TBL" FOR EACH ROW EXECUTE PROCEDURE public.undo_t_trace();

CREATE TRIGGER tg2 BEFORE TRUNCATE ON public."TBL" FOR EACH STATEMENT EXECUTE PROCEDURE public.undo_t_trace();

插入测试数据, 为了增加难度, 我们使用了转义字符. 确保前后数据一致.

insert into "TBL" values (1,1,'te\\s\t','c4','c5','c6',1,now());

insert into "TBL" values (2,1,'te\\s\t','c4','c5','c6',1,now());

insert into "TBL" values (3,1,'te\\s\t','c4','c5','c6',1,now());

insert into "TBL" values (4,1,'te\\s\t','c4','c5','c6',1,now());

insert into "TBL" values (5,1,'te\\s\t','c4','c5','c6',1,now());

insert into "TBL" values (6,1,'te\\s\t','c4','c5','c6',1,now());

插入后, 可以看到 INSERT被跟踪了, 并且我们存储了插入数据时的客户端编码. 方便解决编码问题.

postgres=# select * from undo_t;

rec | crt_time | username | client_addr | client_port

----+---------+----------+--------------+------------+---------+-------+--------+----------+---------+------------------------------

------------------------------+----------------------------+----------+-------------+-------------

1 | 1301665 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 | | (1,1,"te\\\\s\\t",c4,c5,c6,1,

"2014-08-28 23:06:09.790227") | 2014-08-28 23:06:09.790227 | postgres | |

2 | 1301666 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 | | (2,1,"te\\\\s\\t",c4,c5,c6,1,

"2014-08-28 23:06:09.79597") | 2014-08-28 23:06:09.79597 | postgres | |

3 | 1301667 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 | | (3,1,"te\\\\s\\t",c4,c5,c6,1,

"2014-08-28 23:06:09.80206") | 2014-08-28 23:06:09.80206 | postgres | |

4 | 1301668 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 | | (4,1,"te\\\\s\\t",c4,c5,c6,1,

"2014-08-28 23:06:09.80903") | 2014-08-28 23:06:09.80903 | postgres | |

5 | 1301669 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 | | (5,1,"te\\\\s\\t",c4,c5,c6,1,

"2014-08-28 23:06:09.819092") | 2014-08-28 23:06:09.819092 | postgres | |

6 | 1301670 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 | | (6,1,"te\\\\s\\t",c4,c5,c6,1,

"2014-08-28 23:06:10.228624") | 2014-08-28 23:06:10.228624 | postgres | |

(6 rows)

postgres=# select * from "TBL";

c1 | c2 | C3 | c4 | c5 | c6 | c7 | crt_time

----+----+---------+----+----+----+----+----------------------------

1 | 1 | te\\s\t | c4 | c5 | c6 | 1 | 2014-08-28 23:06:09.790227

2 | 1 | te\\s\t | c4 | c5 | c6 | 1 | 2014-08-28 23:06:09.79597

3 | 1 | te\\s\t | c4 | c5 | c6 | 1 | 2014-08-28 23:06:09.80206

4 | 1 | te\\s\t | c4 | c5 | c6 | 1 | 2014-08-28 23:06:09.80903

5 | 1 | te\\s\t | c4 | c5 | c6 | 1 | 2014-08-28 23:06:09.819092

6 | 1 | te\\s\t | c4 | c5 | c6 | 1 | 2014-08-28 23:06:10.228624

回退操作我们这里用一个inline plpgsql 代码来处理, 如果你要写成函数也可以, 只需要传入一个XID即可.

回退最后一个事务, 即c1=6的那条记录. 以事务号1301670为界限.

注意变量使用标量, 因为在for 和 cursor fetch到一个变量时, 变量必须是标量.

参考代码 :

src/pl/plpgsql/src/pl_gram.y

do language plpgsql $$

declare

v_op text;

v_encoding_curr text := pg_client_encoding();

v_encoding_tmp text;

v_old text; -- 本来这里打算用public."TBL"来作为变量类型, 不过for, cursor都不允许存储非标量类型, 所以还是选择了标量text, 使用时转换.

v_new text;

v_xid int8 := 1301670;

begin

for v_op, v_encoding_tmp, v_old, v_new in

select op,encoding,old_rec::text,new_rec::text from undo_t where xid>=v_xid order by xid desc,id desc

LOOP

execute 'set client_encoding='''||v_encoding_tmp||'''';

case v_op

when 'INSERT' then

delete from public."TBL" t where t=v_new::public."TBL";

when 'DELETE' then

insert into public."TBL" values ((v_old::public."TBL").*);

when 'TRUNCATE' then

when 'UPDATE' then

else

end case;

end loop;

execute 'set client_encoding='''||v_encoding_curr||'''';

end;

$$;

回退成功

(5 rows)

回退操作同样会产生undo记录.

| new_rec | crt_time | username | client_ad

dr | client_port

----+---------+----------+--------------+------------+---------+-------+--------+----------+----------------------------------------

--------------------+------------------------------------------------------------+----------------------------+----------+----------

---+-------------

1 | 1301665 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 |

| (1,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.790227") | 2014-08-28 23:06:09.790227 | postgres |

2 | 1301666 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 |

| (2,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.79597") | 2014-08-28 23:06:09.79597 | postgres |

3 | 1301667 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 |

| (3,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.80206") | 2014-08-28 23:06:09.80206 | postgres |

4 | 1301668 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 |

| (4,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.80903") | 2014-08-28 23:06:09.80903 | postgres |

5 | 1301669 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 |

| (5,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.819092") | 2014-08-28 23:06:09.819092 | postgres |

6 | 1301670 | 50534894 | public | TBL | AFTER | ROW | INSERT | UTF8 |

| (6,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:10.228624") | 2014-08-28 23:06:10.228624 | postgres |

7 | 1301671 | 50534894 | public | TBL | AFTER | ROW | DELETE | UTF8 | (6,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-2

8 23:06:10.228624") | | 2014-08-28 23:07:07.750644 | postgres |

(7 rows)

现在执行一个UPDATE, 把所有的记录更新掉.

postgres=# update "TBL" set c7=100;

UPDATE 5

c1 | c2 | C3 | c4 | c5 | c6 | c7 | crt_time

----+----+---------+----+----+----+-----+----------------------------

1 | 1 | te\\s\t | c4 | c5 | c6 | 100 | 2014-08-28 23:06:09.790227

2 | 1 | te\\s\t | c4 | c5 | c6 | 100 | 2014-08-28 23:06:09.79597

3 | 1 | te\\s\t | c4 | c5 | c6 | 100 | 2014-08-28 23:06:09.80206

4 | 1 | te\\s\t | c4 | c5 | c6 | 100 | 2014-08-28 23:06:09.80903

5 | 1 | te\\s\t | c4 | c5 | c6 | 100 | 2014-08-28 23:06:09.819092

| new_rec | crt_time | username | client_

addr | client_port

--------------------+--------------------------------------------------------------+----------------------------+----------+--------

-----+-------------

| (1,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.790227") | 2014-08-28 23:06:09.790227 | postgres |

| (2,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.79597") | 2014-08-28 23:06:09.79597 | postgres |

| (3,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.80206") | 2014-08-28 23:06:09.80206 | postgres |

| (4,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.80903") | 2014-08-28 23:06:09.80903 | postgres |

| (5,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:09.819092") | 2014-08-28 23:06:09.819092 | postgres |

| (6,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-28 23:06:10.228624") | 2014-08-28 23:06:10.228624 | postgres |

8 23:06:10.228624") | | 2014-08-28 23:07:07.750644 | postgres |

8 | 1301672 | 50534894 | public | TBL | AFTER | ROW | UPDATE | UTF8 | (1,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-2

8 23:06:09.790227") | (1,1,"te\\\\s\\t",c4,c5,c6,100,"2014-08-28 23:06:09.790227") | 2014-08-28 23:08:52.887568 | postgres |

9 | 1301672 | 50534894 | public | TBL | AFTER | ROW | UPDATE | UTF8 | (2,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-2

8 23:06:09.79597") | (2,1,"te\\\\s\\t",c4,c5,c6,100,"2014-08-28 23:06:09.79597") | 2014-08-28 23:08:52.887568 | postgres |

10 | 1301672 | 50534894 | public | TBL | AFTER | ROW | UPDATE | UTF8 | (3,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-2

8 23:06:09.80206") | (3,1,"te\\\\s\\t",c4,c5,c6,100,"2014-08-28 23:06:09.80206") | 2014-08-28 23:08:52.887568 | postgres |

11 | 1301672 | 50534894 | public | TBL | AFTER | ROW | UPDATE | UTF8 | (4,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-2

8 23:06:09.80903") | (4,1,"te\\\\s\\t",c4,c5,c6,100,"2014-08-28 23:06:09.80903") | 2014-08-28 23:08:52.887568 | postgres |

12 | 1301672 | 50534894 | public | TBL | AFTER | ROW | UPDATE | UTF8 | (5,1,"te\\\\s\\t",c4,c5,c6,1,"2014-08-2

8 23:06:09.819092") | (5,1,"te\\\\s\\t",c4,c5,c6,100,"2014-08-28 23:06:09.819092") | 2014-08-28 23:08:52.887568 | postgres |

(12 rows)

回退到更新前, 即1301672 这个XID需要回退掉.

v_encoding_curr text := pg_client_encoding();

v_old text;

v_xid int8 := 1301672;

execute 'set client_encoding='''||v_encoding_tmp||'''';

现在把所有记录删除掉

postgres=# delete from "TBL";

DELETE 5

| new_rec | crt_time | username | clien

t_addr | client_port

----------------------+--------------------------------------------------------------+----------------------------+----------+------

-------+-------------