git.8kb.co.uk Git - postgresql/table_log_pl/blob - table_log.sql

   1 --
   2 -- Glyn Astill 28/08/2015
   3 --
   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
   8 --
   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
  12 --
  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.
  16 --
  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
  19
  20
  21 -- drop old function
  22 DROP FUNCTION IF EXISTS table_log_pl (); -- ignore any error (but do not CASCADE)
  23 DROP FUNCTION IF EXISTS table_log_restore_table_pl(varchar, varchar, char, char, char, timestamptz, char, int, int, varchar, varchar);
  24
  25 -- Table_log plpgsql function; there's better ways to do this, but this is a drop in replacement for table_log C functions.
  26 CREATE OR REPLACE FUNCTION table_log_pl() RETURNS TRIGGER AS
  27 $BODY$
  28 DECLARE
  29     v_tabname text;
  30     v_loguser boolean := false;
  31     v_nspname text;
  32     v_num_col int;
  33     v_num_col_log int;
  34     v_col_trig text := '';
  35     v_val_trig text := '';
  36     v_cols text := '';
  37     v_sql text;
  38     v_col_cache text;
  39     v_enable_cache boolean := true;
  40     v_enable_prepare boolean := true;
  41     v_tmp text;
  42     v_i int;
  43
  44     v_cols_typ text;
  45     v_cols_nam text[];
  46     v_vals_old text = '';
  47     v_vals_new text = '';
  48 BEGIN
  49     -- Notes:
  50     --     - The trigger_id comes off sequence, this function is oblivious
  51     --     - 3 columns means don't log trigger_user or trigger_id
  52     --     - 4 columns means don't log trigger_user
  53     --     - 5 columns means log both
  54     --     - To use the column data caching on server versions prior to
  55     --       9.6 add custom var "table_log.column_cache = ''" to postgresql.conf
  56
  57     IF (TG_NARGS > 2) THEN
  58         v_nspname := TG_ARGV[2];
  59     ELSE
  60         v_nspname := TG_TABLE_SCHEMA;
  61     END IF;
  62
  63     IF (TG_NARGS > 1 AND TG_ARGV[1]::int = 1) THEN
  64         v_loguser := true;
  65     END IF;
  66
  67     IF (TG_NARGS > 0) THEN
  68         v_tabname := TG_ARGV[0];
  69     ELSE
  70         v_tabname := TG_TABLE_NAME || '_log';
  71     END IF;
  72
  73     -- Retrieve custom variable used as a poor mans cache for multirow statements
  74     IF (v_enable_cache) THEN
  75         IF (current_setting('server_version_num')::int >= 90600) THEN
  76             v_col_cache := current_setting('table_log.column_cache', true);
  77         ELSE
  78             v_col_cache := current_setting('table_log.column_cache');
  79         END IF;
  80     END IF;
  81
  82     -- If column caching is enabled and previous call in this transaction
  83     -- was for the same relation we can retrieve column detail.
  84     IF (v_enable_cache AND left(v_col_cache, length(TG_RELID::text)+1) = (TG_RELID::text || ':')) THEN
  85         v_cols := right(v_col_cache, (length(TG_RELID::text)+1)*-1);
  86         v_cols_nam := ('{' || right(v_col_cache, (length(TG_RELID::text)+1)*-1) || '}')::text[];
  87     ELSE -- Otherwise fetch the column detail
  88         IF (TG_WHEN != 'AFTER') THEN
  89             RAISE EXCEPTION 'table_log: must be fired after event';
  90         END IF;
  91         IF (TG_LEVEL = 'STATEMENT') THEN
  92             RAISE EXCEPTION 'table_log: can''t process STATEMENT events';
  93         END IF;
  94
  95         SELECT count(*), string_agg(quote_ident(attname),','), string_agg(format_type(atttypid, atttypmod),','), array_agg(quote_ident(attname))
  96         INTO STRICT v_num_col, v_cols, v_cols_typ, v_cols_nam
  97         FROM pg_catalog.pg_attribute
  98         WHERE attrelid = TG_RELID
  99         AND attnum > 0
 100         AND NOT attisdropped;
 101
 102         IF (v_num_col < 1) THEN
 103             RAISE EXCEPTION 'table_log: number of columns in table is < 1, can this happen?';
 104         END IF;
 105
 106         SELECT count(*) INTO STRICT v_num_col_log
 107         FROM pg_catalog.pg_attribute
 108         WHERE attrelid = (v_nspname || '.' || v_tabname)::regclass
 109         AND attnum > 0
 110         AND NOT attisdropped;
 111
 112         IF (v_num_col_log < 1) THEN
 113             RAISE EXCEPTION 'could not get number columns in relation %.%', v_nspname, v_tabname;
 114         END IF;
 115
 116         -- This is the way the original checks column count regardless of trigger_id is presence
 117         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
 118             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;
 119         END IF;
 120
 121         -- Set custom variable for use as a poor mans cache for multirow statements
 122         IF (v_enable_cache) THEN
 123             v_col_cache := (TG_RELID::text || ':' || v_cols);
 124             PERFORM set_config('table_log.column_cache', v_col_cache, true);
 125         END IF;
 126
 127         -- Create a prepared statement for the current table, deallocating
 128         -- any old statements we may have prepared.
 129         IF (v_enable_prepare) THEN
 130             FOR v_tmp IN (SELECT name FROM pg_catalog.pg_prepared_statements WHERE name ~ '^table_log_pl_') LOOP
 131                 EXECUTE format('DEALLOCATE %I', v_tmp);
 132             END LOOP;
 133
 134             SELECT '$' || string_agg(a::text, ', $') INTO v_col_trig FROM generate_series(1,v_num_col+3+v_loguser::int) a;
 135
 136             IF (v_loguser) THEN
 137                 v_sql := format('PREPARE table_log_pl_%s(%s, text, text, timestamptz, text) AS INSERT INTO %I.%I (%s, "trigger_user", "trigger_mode", "trigger_changed", "trigger_tuple") VALUES (%s)', TG_RELID, v_cols_typ, v_nspname, v_tabname, v_cols, v_col_trig);
 138             ELSE
 139                 v_sql := format('PREPARE table_log_pl_%s(%s, text, timestamptz, text) AS INSERT INTO %I.%I (%s, "trigger_mode", "trigger_changed", "trigger_tuple") VALUES (%s)', TG_RELID, v_cols_typ, v_nspname, v_tabname, v_cols, v_col_trig);
 140             END IF;
 141             EXECUTE v_sql;
 142         END IF;
 143     END IF;
 144
 145     -- If prepared statement method is enabled, construct strings for
 146     -- variable parameters and execute.
 147     IF (v_enable_prepare) THEN
 148         FOR v_i IN 1..array_upper(v_cols_nam, 1) LOOP
 149             IF (TG_OP != 'INSERT') THEN
 150                 EXECUTE 'SELECT ($1).' || v_cols_nam[v_i] || '::text' INTO v_tmp USING OLD;
 151                 v_vals_old :=  v_vals_old || quote_nullable(v_tmp) || ',';
 152             END IF;
 153             IF (TG_OP != 'DELETE') THEN
 154                 EXECUTE 'SELECT ($1).' || v_cols_nam[v_i] || '::text' INTO v_tmp USING NEW;
 155                 v_vals_new :=  v_vals_new || quote_nullable(v_tmp) || ',';
 156             END IF;
 157         END LOOP;
 158
 159         IF (v_loguser) THEN
 160             v_vals_new :=  v_vals_new || quote_literal(session_user) || ',';
 161             v_vals_old :=  v_vals_old || quote_literal(session_user) || ',';
 162         END IF;
 163
 164         IF (TG_OP != 'INSERT') THEN
 165             v_sql := format('EXECUTE table_log_pl_%s(%s%L, %L, %L)', TG_RELID, v_vals_old, TG_OP, current_timestamp, 'old');
 166             EXECUTE v_sql;
 167         END IF;
 168         IF (TG_OP != 'DELETE') THEN
 169             v_sql := format('EXECUTE table_log_pl_%s(%s%L, %L, %L)', TG_RELID, v_vals_new, TG_OP, current_timestamp, 'new');
 170             EXECUTE v_sql;
 171             RETURN NEW;
 172         ELSE
 173             RETURN OLD;
 174         END IF;
 175     ELSE -- Otherwise we can do the inserts dynamically.
 176         IF (v_loguser) THEN
 177             v_col_trig := v_col_trig || ', "trigger_user"';
 178             v_val_trig := format('%L, ', session_user);
 179         END IF;
 180         v_col_trig := v_col_trig || ', "trigger_mode", "trigger_changed", "trigger_tuple"';
 181         v_val_trig := format('%s%L, %L', v_val_trig, TG_OP, current_timestamp);
 182
 183         IF (TG_OP != 'INSERT') THEN
 184             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);
 185             EXECUTE v_sql USING OLD;
 186         END IF;
 187         IF (TG_OP != 'DELETE') THEN
 188             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);
 189             EXECUTE v_sql USING NEW;
 190             RETURN NEW;
 191         ELSE
 192             RETURN OLD;
 193         END IF;
 194     END IF;
 195
 196 END;
 197 $BODY$
 198 LANGUAGE plpgsql VOLATILE;
 199
 200 CREATE OR REPLACE FUNCTION table_log_restore_table_pl (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
 201 $BODY$
 202 DECLARE
 203     v_origtab_cols int;
 204     v_logtab_cols int;
 205     v_restoretab_cols int;
 206     v_origtab_fqn text;
 207     v_logtab_fqn text;
 208     v_sql text;
 209     v_cols text;
 210     v_pk_count int;
 211     v_rec record;
 212     v_old_pk_str text;
 213 BEGIN
 214
 215     -- Notes:
 216     --
 217     -- The original implimentation doesn't allow fully qualified table
 218     -- references in table_log_restore_table;  You can get some milage
 219     -- out of search_path if required there. For this reason the plpgsql
 220     -- version adds the following two optional parameters to those below:
 221     --
 222     --   - original table schema
 223     --   - logging table schema
 224     --
 225     -- Comments from C implimentation:
 226     --
 227     -- restore a complete table based on the logging table
 228     --
 229     -- parameter:
 230     --   - original table name
 231     --   - name of primary key in original table
 232     --   - logging table
 233     --   - name of primary key in logging table
 234     --   - restore table name
 235     --   - timestamp for restoring data
 236     --   - primary key to restore (only this key will be restored) (optional)
 237     --   - restore mode
 238     --     0: restore from blank table (default)
 239     --        needs a complete logging table
 240     --     1: restore from actual table backwards
 241     --   - dont create table temporarly
 242     --     0: create restore table temporarly (default)
 243     --     1: create restore table not temporarly
 244     --   return:
 245     --     not yet defined
 246
 247     IF origtab IS NULL THEN
 248         RAISE NOTICE 'table_log_restore_table: missing original table name';
 249     END IF;
 250     IF origtab_pk IS NULL THEN
 251         RAISE NOTICE 'table_log_restore_table: missing primary key name for original table';
 252     END IF;
 253     IF logtab IS NULL THEN
 254         RAISE NOTICE 'table_log_restore_table: missing log table name';
 255     END IF;
 256     IF logtab_pk IS NULL THEN
 257         RAISE NOTICE 'table_log_restore_table: missing primary key name for log table';
 258     END IF;
 259     IF restoretab IS NULL THEN
 260         RAISE NOTICE 'table_log_restore_table: missing copy table name';
 261     END IF;
 262     IF to_timestamp IS NULL THEN
 263         RAISE NOTICE 'table_log_restore_table: missing timestamp';
 264     END IF;
 265     IF (search_pk IS NOT NULL) THEN
 266         RAISE NOTICE 'table_log_restore_table: will restore a single key';
 267     END IF;
 268
 269     IF origtab_pk = logtab_pk THEN
 270         RAISE EXCEPTION 'pkey of logging table cannot be the pkey of the original table: % <-> %', origtab_pk, logtab_pk;
 271     END IF;
 272
 273     v_origtab_fqn := coalesce(quote_ident(origtab_schema) || '.','') || quote_ident(origtab);
 274     v_logtab_fqn := coalesce(quote_ident(logtab_schema) || '.','') || quote_ident(logtab);
 275
 276     -- Check original table and get column list
 277     SELECT string_agg(quote_ident(attname), ','), count(*), count(*) filter (where attname=origtab_pk)
 278     INTO v_cols, v_origtab_cols, v_pk_count
 279     FROM pg_catalog.pg_class c
 280     JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
 281     JOIN pg_catalog.pg_namespace n ON c.relnamespace = n.oid
 282     WHERE c.relname = origtab AND c.relkind='r' AND a.attnum > 0
 283     AND (origtab_schema IS NULL OR n.nspname = origtab_schema)
 284     AND NOT attisdropped;
 285
 286     IF v_origtab_cols = 0 OR v_cols IS NULL THEN
 287         RAISE EXCEPTION 'could not check relation: % (columns = %)', v_origtab_fqn, v_origtab_cols;
 288     ELSIF v_pk_count != 1 THEN
 289         RAISE EXCEPTION 'could not check relation: (missing pkey) % in table %', origtab_pk, v_origtab_fqn;
 290     ELSE
 291         RAISE NOTICE 'original table: OK (% columns)', v_origtab_cols;
 292     END IF;
 293
 294     -- Check log table
 295     SELECT count(*), count(*) filter (where attname=logtab_pk)
 296     INTO v_logtab_cols, v_pk_count
 297     FROM pg_catalog.pg_class c
 298     JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
 299     JOIN pg_catalog.pg_namespace n ON c.relnamespace = n.oid
 300     WHERE c.relname = logtab AND c.relkind='r' AND a.attnum > 0
 301     AND (logtab_schema IS NULL OR n.nspname = logtab_schema)
 302     AND NOT attisdropped;
 303
 304     IF v_logtab_cols = 0 THEN
 305         RAISE EXCEPTION 'could not check relation: % (columns = %)', v_logtab_fqn, v_logtab_cols;
 306     ELSIF v_pk_count != 1 THEN
 307         RAISE EXCEPTION 'could not check relation: (missing pkey) % in table %', logtab_pk, v_logtab_fqn;
 308     ELSE
 309         RAISE NOTICE 'log table: OK (% columns)', v_logtab_cols;
 310     END IF;
 311
 312     -- Check restore table
 313     IF EXISTS(SELECT 1 FROM pg_catalog.pg_class
 314               WHERE relname=restoretab AND relkind='r') THEN
 315         RAISE EXCEPTION 'restore table already exists: %', restoretab;
 316     ELSE
 317         RAISE NOTICE 'restore table: OK (doesnt exists)';
 318     END IF;
 319
 320     -- create restore table
 321     v_sql := 'CREATE';
 322     IF not_temporarly = 0 THEN
 323         v_sql := v_sql || ' TEMPORARY';
 324     END IF;
 325     v_sql := v_sql || format(' TABLE %I AS SELECT * FROM %s', restoretab, v_origtab_fqn);
 326     IF search_pk IS NOT NULL THEN
 327         v_sql := v_sql || format(' WHERE %I = %L', origtab_pk, search_pk);
 328     END IF;
 329     IF method = 0 THEN
 330         RAISE NOTICE 'need logs from start to timestamp: %', to_timestamp;
 331         v_sql := v_sql || ' LIMIT 0'; -- Create blank table to roll forward into (need all logs)
 332     ELSE
 333         RAISE NOTICE 'need logs from end to timestamp: %', to_timestamp;
 334     END IF;
 335
 336     -- RAISE NOTICE 'DDL: %', v_sql;
 337     EXECUTE v_sql;
 338
 339     -- now build query for getting logs
 340     v_sql := format('SELECT * FROM %s WHERE ', v_logtab_fqn);
 341     IF method = 0 THEN
 342         v_sql := v_sql || format('trigger_changed <= %L', to_timestamp); -- ROLL FORWARD
 343     ELSE
 344         v_sql := v_sql || format('trigger_changed >= %L', to_timestamp); -- ROLL BACK
 345     END IF;
 346
 347     IF search_pk IS NOT NULL THEN
 348         v_sql := v_sql || format(' AND %I = %L', origtab_pk, search_pk);
 349     END IF;
 350
 351     IF method = 0 THEN
 352         v_sql := v_sql || format(' ORDER BY %I ASC', logtab_pk);
 353     ELSE
 354         v_sql := v_sql || format(' ORDER BY %I DESC', logtab_pk);
 355     END IF;
 356
 357     -- RAISE NOTICE 'SQL: %', v_sql;
 358
 359     FOR v_rec IN EXECUTE v_sql
 360     LOOP
 361         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
 362             -- For previous update row versions we needn't apply anything;
 363             -- we just note the pk value for the quals when applying the
 364             -- next row change, i.e when rolling forward the old pk value,
 365             -- when rolling back the new pk value
 366             EXECUTE format('SELECT ($1::text::%s).%I', v_logtab_fqn, origtab_pk) INTO v_old_pk_str USING v_rec;
 367         ELSE
 368             -- Apply the row changes from the log table, the following is
 369             -- a mass of substitutions, but essentially we're selecting
 370             -- data out of the log table record and casting it into the
 371             -- restore table.
 372
 373             IF v_rec.trigger_mode = 'UPDATE' THEN
 374                 v_sql := format('UPDATE %I SET (%s) = (SELECT %s FROM (SELECT ($1::text::%s).*) t) WHERE %I = %L',
 375                                 restoretab, v_cols, v_cols, v_logtab_fqn, origtab_pk, v_old_pk_str);
 376             ELSIF (v_rec.trigger_mode = 'INSERT' AND method = 0) OR (v_rec.trigger_mode = 'DELETE' AND method != 0) THEN
 377                 v_sql := format('INSERT INTO %I (%s) SELECT %s FROM (SELECT ($1::text::%s).*) t',
 378                                 restoretab, v_cols, v_cols, v_logtab_fqn);
 379             ELSIF (v_rec.trigger_mode = 'INSERT' AND method != 0) OR (v_rec.trigger_mode = 'DELETE' AND method = 0) THEN
 380                 v_sql := format('DELETE FROM %I WHERE %I = ($1::text::%s).%I',
 381                                 restoretab, origtab_pk, v_logtab_fqn, origtab_pk);
 382             ELSE
 383                 RAISE EXCEPTION 'unknown trigger_mode: %', trigger_mode;
 384             END IF;
 385
 386             -- RAISE NOTICE 'DML: %', v_sql;
 387             EXECUTE v_sql USING v_rec;
 388         END IF;
 389
 390     END LOOP;
 391
 392     RETURN quote_ident(restoretab);
 393 END;
 394 $BODY$
 395 LANGUAGE plpgsql VOLATILE;