2 -- Glyn Astill 28/08/2015
4 -- Attempt at pl/pgsql drop-in replacement for table_log C extenstion AKA
5 -- pg Table Audit / PostgreSQL Table Log / tablelog by Andreas Scherbaum
6 -- http://www.postgresql.org/ftp/projects/pgFoundry/tablelog/tablelog/
7 -- http://github.com/andreasscherbaum/table_log
9 -- A slightly more up to date version of the original C extension can
10 -- also be found here:
11 -- https://github.com/glynastill/pg_table_audit
13 -- There are now many better ways to audit DML, using json types or
14 -- advanced extensions like pgaudit (below), however if for some reason
15 -- you're stuck with table_log this may help.
17 -- http://8kb.co.uk/blog/2015/01/19/copying-pavel-stehules-simple-history-table-but-with-the-jsonb-type/
18 -- https://github.com/2ndQuadrant/pgaudit
20 SET search_path TO public;
23 DROP FUNCTION IF EXISTS table_log_pl (); -- ignore any error (but do not CASCADE)
24 DROP FUNCTION IF EXISTS table_log_pl_restore_table(varchar, varchar, char, char, char, timestamptz, char, int, int, varchar, varchar);
28 CREATE OR REPLACE FUNCTION table_log_pl() RETURNS TRIGGER AS
32 v_loguser boolean := false;
36 v_col_trig text := '';
37 v_val_trig text := '';
42 v_enable_cache boolean := true;
45 -- - The trigger_id comes off sequence, this function is oblivious
46 -- - 3 columns means don't log trigger_user or trigger_id
47 -- - 4 columns means don't log trigger_user
48 -- - 5 columns means log both
49 -- - To use the column data caching on server versions prior to
50 -- 9.6 add custom var "table_log.column_cache = ''" to postgresql.conf
52 IF (TG_NARGS > 2) THEN
53 v_nspname := TG_ARGV[2];
55 v_nspname := TG_TABLE_SCHEMA;
58 IF (TG_NARGS > 1 AND TG_ARGV[1]::int = 1) THEN
62 IF (TG_NARGS > 0) THEN
63 v_tabname := TG_ARGV[0];
65 v_tabname := TG_TABLE_NAME || '_log';
68 -- Retrieve custom variable used as a poor mans cache for multirow statements
69 IF (v_enable_cache) THEN
70 IF (current_setting('server_version_num')::int >= 90600) THEN
71 v_col_cache := current_setting('table_log.column_cache', true);
73 v_col_cache := current_setting('table_log.column_cache');
77 IF (v_enable_cache AND left(v_col_cache, length(TG_RELID::text)+1) = (TG_RELID::text || ':')) THEN
78 v_cols := right(v_col_cache, (length(TG_RELID::text)+1)*-1);
80 IF (TG_WHEN != 'AFTER') THEN
81 RAISE EXCEPTION 'table_log: must be fired after event';
83 IF (TG_LEVEL = 'STATEMENT') THEN
84 RAISE EXCEPTION 'table_log: can''t process STATEMENT events';
87 SELECT count(*), string_agg(quote_ident(attname),',') INTO STRICT v_num_col, v_cols
88 FROM pg_catalog.pg_attribute
89 WHERE attrelid = TG_RELID
93 IF (v_num_col < 1) THEN
94 RAISE EXCEPTION 'table_log: number of columns in table is < 1, can this happen?';
97 SELECT count(*) INTO STRICT v_num_col_log
98 FROM pg_catalog.pg_attribute
99 WHERE attrelid = (v_nspname || '.' || v_tabname)::regclass
101 AND NOT attisdropped;
103 IF (v_num_col_log < 1) THEN
104 RAISE EXCEPTION 'could not get number columns in relation %.%', v_nspname, v_tabname;
107 -- This is the way the original checks column count regardless of trigger_id is presence
108 IF (v_num_col_log != (v_num_col + 3 + v_loguser::int)) AND (v_num_col_log != (v_num_col + 4 + v_loguser::int)) THEN
109 RAISE EXCEPTION 'number colums in relation %.%(%) does not match columns in %.%(%)', TG_TABLE_SCHEMA, TG_TABLE_NAME, v_num_col, v_nspname, v_tabname, v_num_col_log;
112 -- Set custom variable for use as a poor mans cache for multirow statements
113 IF (v_enable_cache) THEN
114 v_col_cache := (TG_RELID::text || ':' || v_cols);
115 PERFORM set_config('table_log.column_cache', v_col_cache, true);
120 v_col_trig := v_col_trig || ', "trigger_user"';
121 v_val_trig := format('%L, ', session_user);
123 v_col_trig := v_col_trig || ', "trigger_mode", "trigger_changed", "trigger_tuple"';
124 v_val_trig := format('%s%L, %L', v_val_trig, TG_OP, current_timestamp);
126 IF (TG_OP != 'INSERT') THEN
127 v_sql := format('INSERT INTO %I.%I (%s%s) SELECT %s, %s, ''old'' FROM (SELECT ($1::text::%I).*) t', v_nspname, v_tabname, v_cols, v_col_trig, v_cols, v_val_trig, TG_RELID::regclass);
128 EXECUTE v_sql USING OLD;
130 IF (TG_OP != 'DELETE') THEN
131 v_sql := format('INSERT INTO %I.%I (%s%s) SELECT %s, %s, ''new'' FROM (SELECT ($1::text::%I).*) t', v_nspname, v_tabname, v_cols, v_col_trig, v_cols, v_val_trig, TG_RELID::regclass);
132 EXECUTE v_sql USING NEW;
140 LANGUAGE plpgsql VOLATILE;
144 CREATE OR REPLACE FUNCTION table_log_pl_restore_table (origtab varchar, origtab_pk varchar, logtab char, logtab_pk char, restoretab char, to_timestamp timestamptz, search_pk char DEFAULT NULL, method int DEFAULT 0, not_temporarly int DEFAULT 0, origtab_schema varchar DEFAULT NULL, logtab_schema varchar DEFAULT NULL) RETURNS varchar AS
149 v_restoretab_cols int;
161 -- The original implimentation doesn't allow fully qualified table
162 -- references in table_log_restore_table; You can get some milage
163 -- out of search_path if required there. For this reason the plpgsql
164 -- version adds the following two optional parameters to those below:
166 -- - original table schema
167 -- - logging table schema
169 -- Comments from C implimentation:
171 -- restore a complete table based on the logging table
174 -- - original table name
175 -- - name of primary key in original table
177 -- - name of primary key in logging table
178 -- - restore table name
179 -- - timestamp for restoring data
180 -- - primary key to restore (only this key will be restored) (optional)
182 -- 0: restore from blank table (default)
183 -- needs a complete logging table
184 -- 1: restore from actual table backwards
185 -- - dont create table temporarly
186 -- 0: create restore table temporarly (default)
187 -- 1: create restore table not temporarly
191 IF origtab IS NULL THEN
192 RAISE NOTICE 'table_log_restore_table: missing original table name';
194 IF origtab_pk IS NULL THEN
195 RAISE NOTICE 'table_log_restore_table: missing primary key name for original table';
197 IF logtab IS NULL THEN
198 RAISE NOTICE 'table_log_restore_table: missing log table name';
200 IF logtab_pk IS NULL THEN
201 RAISE NOTICE 'table_log_restore_table: missing primary key name for log table';
203 IF restoretab IS NULL THEN
204 RAISE NOTICE 'table_log_restore_table: missing copy table name';
206 IF to_timestamp IS NULL THEN
207 RAISE NOTICE 'table_log_restore_table: missing timestamp';
209 IF (search_pk IS NOT NULL) THEN
210 RAISE NOTICE 'table_log_restore_table: will restore a single key';
213 IF origtab_pk = logtab_pk THEN
214 RAISE EXCEPTION 'pkey of logging table cannot be the pkey of the original table: % <-> %', origtab_pk, logtab_pk;
217 v_origtab_fqn := coalesce(quote_ident(origtab_schema) || '.','') || quote_ident(origtab);
218 v_logtab_fqn := coalesce(quote_ident(logtab_schema) || '.','') || quote_ident(logtab);
220 -- Check original table and get column list
221 SELECT string_agg(quote_ident(attname), ','), count(*), count(*) filter (where attname=origtab_pk)
222 INTO v_cols, v_origtab_cols, v_pk_count
223 FROM pg_catalog.pg_class c
224 JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
225 JOIN pg_catalog.pg_namespace n ON c.relnamespace = n.oid
226 WHERE c.relname = origtab AND c.relkind='r' AND a.attnum > 0
227 AND (origtab_schema IS NULL OR n.nspname = origtab_schema)
228 AND NOT attisdropped;
230 IF v_origtab_cols = 0 OR v_cols IS NULL THEN
231 RAISE EXCEPTION 'could not check relation: % (columns = %)', v_origtab_fqn, v_origtab_cols;
232 ELSIF v_pk_count != 1 THEN
233 RAISE EXCEPTION 'could not check relation: (missing pkey) % in table %', origtab_pk, v_origtab_fqn;
235 RAISE NOTICE 'original table: OK (% columns)', v_origtab_cols;
239 SELECT count(*), count(*) filter (where attname=logtab_pk)
240 INTO v_logtab_cols, v_pk_count
241 FROM pg_catalog.pg_class c
242 JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
243 JOIN pg_catalog.pg_namespace n ON c.relnamespace = n.oid
244 WHERE c.relname = logtab AND c.relkind='r' AND a.attnum > 0
245 AND (logtab_schema IS NULL OR n.nspname = logtab_schema)
246 AND NOT attisdropped;
248 IF v_logtab_cols = 0 THEN
249 RAISE EXCEPTION 'could not check relation: % (columns = %)', v_logtab_fqn, v_logtab_cols;
250 ELSIF v_pk_count != 1 THEN
251 RAISE EXCEPTION 'could not check relation: (missing pkey) % in table %', logtab_pk, v_logtab_fqn;
253 RAISE NOTICE 'log table: OK (% columns)', v_logtab_cols;
256 -- Check restore table
257 IF EXISTS(SELECT 1 FROM pg_catalog.pg_class
258 WHERE relname=restoretab AND relkind='r') THEN
259 RAISE EXCEPTION 'restore table already exists: %', restoretab;
261 RAISE NOTICE 'restore table: OK (doesnt exists)';
264 -- create restore table
266 IF not_temporarly = 0 THEN
267 v_sql := v_sql || ' TEMPORARY';
269 v_sql := v_sql || format(' TABLE %I AS SELECT * FROM %s', restoretab, v_origtab_fqn);
270 IF search_pk IS NOT NULL THEN
271 v_sql := v_sql || format(' WHERE %I = %L', origtab_pk, search_pk);
274 RAISE NOTICE 'need logs from start to timestamp: %', to_timestamp;
275 v_sql := v_sql || ' LIMIT 0'; -- Create blank table to roll forward into (need all logs)
277 RAISE NOTICE 'need logs from end to timestamp: %', to_timestamp;
280 -- RAISE NOTICE 'DDL: %', v_sql;
283 -- now build query for getting logs
284 v_sql := format('SELECT * FROM %s WHERE ', v_logtab_fqn);
286 v_sql := v_sql || format('trigger_changed <= %L', to_timestamp); -- ROLL FORWARD
288 v_sql := v_sql || format('trigger_changed >= %L', to_timestamp); -- ROLL BACK
291 IF search_pk IS NOT NULL THEN
292 v_sql := v_sql || format(' AND %I = %L', origtab_pk, search_pk);
296 v_sql := v_sql || format(' ORDER BY %I ASC', logtab_pk);
298 v_sql := v_sql || format(' ORDER BY %I DESC', logtab_pk);
301 -- RAISE NOTICE 'SQL: %', v_sql;
303 FOR v_rec IN EXECUTE v_sql
305 IF v_rec.trigger_mode = 'UPDATE' AND ((method = 0 AND v_rec.trigger_tuple = 'old') OR (method = 1 AND v_rec.trigger_tuple = 'new')) THEN
306 -- For previous update row versions we needn't apply anything;
307 -- we just note the pk value for the quals when applying the
308 -- next row change, i.e when rolling forward the old pk value,
309 -- when rolling back the new pk value
310 EXECUTE format('SELECT ($1::text::%s).%I', v_logtab_fqn, origtab_pk) INTO v_old_pk_str USING v_rec;
312 -- Apply the row changes from the log table, the following is
313 -- a mass of substitutions, but essentially we're selecting
314 -- data out of the log table record and casting it into the
317 IF v_rec.trigger_mode = 'UPDATE' THEN
318 v_sql := format('UPDATE %I SET (%s) = (SELECT %s FROM (SELECT ($1::text::%s).*) t) WHERE %I = %L',
319 restoretab, v_cols, v_cols, v_logtab_fqn, origtab_pk, v_old_pk_str);
320 ELSIF (v_rec.trigger_mode = 'INSERT' AND method = 0) OR (v_rec.trigger_mode = 'DELETE' AND method != 0) THEN
321 v_sql := format('INSERT INTO %I (%s) SELECT %s FROM (SELECT ($1::text::%s).*) t',
322 restoretab, v_cols, v_cols, v_logtab_fqn);
323 ELSIF (v_rec.trigger_mode = 'INSERT' AND method != 0) OR (v_rec.trigger_mode = 'DELETE' AND method = 0) THEN
324 v_sql := format('DELETE FROM %I WHERE %I = ($1::text::%s).%I',
325 restoretab, origtab_pk, v_logtab_fqn, origtab_pk);
327 RAISE EXCEPTION 'unknown trigger_mode: %', trigger_mode;
330 -- RAISE NOTICE 'DML: %', v_sql;
331 EXECUTE v_sql USING v_rec;
336 RETURN quote_ident(restoretab);
339 LANGUAGE plpgsql VOLATILE;
344 DROP TRIGGER test_log_chg ON test; -- ignore any error
347 DROP TABLE test; -- ignore any error
351 name VARCHAR(20) NOT NULL
354 -- create the table without data from demo table
355 DROP TABLE test_log; -- ignore any error
356 SELECT * INTO test_log FROM test LIMIT 0;
357 ALTER TABLE test_log ADD COLUMN trigger_mode VARCHAR(10);
358 ALTER TABLE test_log ADD COLUMN trigger_tuple VARCHAR(5);
359 ALTER TABLE test_log ADD COLUMN trigger_changed TIMESTAMPTZ;
360 ALTER TABLE test_log ADD COLUMN trigger_id BIGINT;
361 CREATE SEQUENCE test_log_id;
362 SELECT SETVAL('test_log_id', 1, FALSE);
363 ALTER TABLE test_log ALTER COLUMN trigger_id SET DEFAULT NEXTVAL('test_log_id');
366 CREATE TRIGGER test_log_chg AFTER UPDATE OR INSERT OR DELETE ON test FOR EACH ROW
367 EXECUTE PROCEDURE table_log_pl();
370 INSERT INTO test VALUES (1, 'name');
372 SELECT * FROM test_log;
373 UPDATE test SET name='other name' WHERE id=1;
375 SELECT * FROM test_log;
377 -- create restore table
378 SELECT table_log_pl_restore_table('test', 'id', 'test_log', 'trigger_id', 'test_recover', NOW());
379 SELECT * FROM test_recover;