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

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

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