Implement DDL recycle bin firewall and incremental synchronization with event triggers for PostgreSQL - ApsaraDB RDS

ApsaraDB RDS for PostgreSQL supports event triggers, which fire in response to Data Definition Language (DDL) operations at the database level. This page shows how to use event triggers to implement three features:

DDL recycle bin — record every DDL statement in an audit table so you can track schema changes and recover from accidental drops
DDL firewall — block specific DDL commands before they execute
Incremental synchronization — replicate the DDL audit log from a publisher to a subscriber using logical replication

Prerequisites

Before you begin, make sure that:

The RDS instance runs PostgreSQL
The instance uses a standard SSD or Enhanced SSD

How event triggers work

Event triggers fire at specific points in a DDL statement's lifecycle. This page uses three of these events:

Event	When it fires	Used for
`ddl_command_end`	After a DDL command completes	Recording CREATE, ALTER, and similar statements
`sql_drop`	When objects are dropped	Recording DROP statements
`ddl_command_start`	Before a DDL command runs	Blocking specific statements (firewall)

The DDL recycle bin and incremental synchronization features both write records to the dts_audit.dts_tb_ddl_command table. The table schema is:

Column	Type	Description
`event`	text	Trigger event name (`ddl_command_end` or `sql_drop`)
`tag`	text	Command tag (for example, `CREATE TABLE`)
`classid`	oid	OID of the system catalog the object belongs to
`objid`	oid	OID of the object within the catalog
`objsubid`	integer	Object sub-ID (for example, column number for column objects)
`object_type`	text	Type of the object
`schema_name`	text	Name of the schema; NULL if none. Not quoted.
`object_identity`	text	Schema-qualified text representation of the object identity
`is_extension`	boolean	True if the command is part of an extension script
`query`	text	The SQL text of the DDL statement
`username`	text	Session user; defaults to `CURRENT_USER`
`datname`	text	Database name; defaults to `current_database()`
`client_addr`	inet	Client IP address; defaults to `inet_client_addr()`
`crt_time`	timestamp	Time the record was inserted; defaults to `now()`

Run the following statements to create the schema and table before setting up any of the three features:

CREATE SCHEMA IF NOT EXISTS dts_audit;
CREATE TABLE IF NOT EXISTS dts_audit.dts_tb_ddl_command (
  event text,
  tag text,
  classid oid,
  objid oid,
  objsubid int,
  object_type text,
  schema_name text,
  object_identity text,
  is_extension bool,
  query text,
  username text default current_user,
  datname text default current_database(),
  client_addr inet default inet_client_addr(),
  crt_time timestamp default now()
);

Set up the DDL recycle bin

The DDL recycle bin captures every DDL statement executed by non-superuser sessions and writes it to dts_audit.dts_tb_ddl_command. Superuser activity is excluded by design.

Two event triggers drive the capture:

pg_get_ddl_command — fires on ddl_command_end, captures CREATE, ALTER, and similar statements (recorded when classid > 0)
pg_get_ddl_drop — fires on sql_drop, captures DROP statements; excludes ALTER TABLE and ALTER FOREIGN TABLE to avoid double-recording

Step 1: Create the audit function and event triggers

/* external/rds_ddl_pulication/rds_ddl_pulication--1.0.sql */

-- Create schema and audit table
CREATE SCHEMA IF NOT EXISTS dts_audit;
CREATE TABLE IF NOT EXISTS dts_audit.dts_tb_ddl_command (
  event text,
  tag text, classid oid, objid oid, objsubid int,
  object_type text, schema_name text, object_identity text, is_extension bool, query text,
  username text default current_user,
  datname text default current_database(),
  client_addr inet default inet_client_addr(),
  crt_time timestamp default now()
);

-- Trigger function: records DDL statements from non-superuser sessions
create or replace function dts_audit.dts_func_ddl_command() returns event_trigger as $$
declare v1 text;
is_superuser bool = false;
r record;
begin
    -- Skip DDL commands from superusers
    select u.rolsuper into is_superuser from pg_catalog.pg_roles u where u.rolname = SESSION_USER;
    if is_superuser then
        return;
    end if;
    select query into v1 from pg_stat_activity where pid=pg_backend_pid();
    -- ddl_command_end: use pg_event_trigger_ddl_commands() to get command details
    if TG_EVENT='ddl_command_end' then
        SELECT * into r FROM pg_event_trigger_ddl_commands();
        if r.classid > 0 then
            insert into dts_audit.dts_tb_ddl_command(event, tag, classid, objid, objsubid, object_type, schema_name, object_identity, is_extension, query)
            values(TG_EVENT, TG_TAG, r.classid, r.objid, r.objsubid, r.object_type, r.schema_name, r.object_identity, r.in_extension, v1);
         end if;
    end if;
    -- sql_drop: use pg_event_trigger_dropped_objects() to get drop details
    -- ALTER TABLE and ALTER FOREIGN TABLE are excluded to avoid repeated collection
    if TG_EVENT='sql_drop' then
        if TG_TAG != 'ALTER TABLE' and TG_TAG != 'ALTER FOREIGN TABLE' then
            SELECT * into r FROM pg_event_trigger_dropped_objects();
            insert into dts_audit.dts_tb_ddl_command(event, tag, classid, objid, objsubid, object_type, schema_name, object_identity, is_extension, query)
            values(TG_EVENT, TG_TAG, r.classid, r.objid, r.objsubid, r.object_type, r.schema_name, r.object_identity, 'f', v1);
        end if;
    end if;
end;
$$ language plpgsql strict;

-- Event trigger: captures ddl_command_end events
CREATE EVENT TRIGGER pg_get_ddl_command on ddl_command_end EXECUTE PROCEDURE dts_audit.dts_func_ddl_command();

-- Event trigger: captures sql_drop events
CREATE EVENT TRIGGER pg_get_ddl_drop on sql_drop EXECUTE PROCEDURE dts_audit.dts_func_ddl_command();

-- Grant access to all users
GRANT USAGE ON SCHEMA dts_audit TO PUBLIC;
GRANT SELECT, INSERT ON TABLE dts_audit.dts_tb_ddl_command TO PUBLIC;

Note

After running the preceding statements, all DDL statements from non-superuser sessions are recorded in dts_audit.dts_tb_ddl_command.

Step 2: Verify that DDL statements are recorded

Run a test DDL statement and then query the audit table:

create table tb_test(id int);
select * from dts_audit.dts_tb_ddl_command;

The audit table shows the recorded statement:

Set up the DDL firewall

The DDL firewall uses the ddl_command_start event to intercept and block specific DDL statements before they execute. The trigger raises an exception, which aborts the statement.

Step 1: Create the trigger function

The following example blocks DDL statements for user test1:

CREATE OR REPLACE FUNCTION abort1()
  RETURNS event_trigger
 LANGUAGE plpgsql
  AS $$
BEGIN
  if current_user = 'test1' then
    RAISE EXCEPTION 'event:%, command:%', tg_event, tg_tag;
  end if;
 END;
$$;

Step 2: Create the event trigger

The following trigger blocks CREATE TABLE and DROP TABLE for user test1:

create event trigger b on ddl_command_start when TAG IN ('CREATE TABLE', 'DROP TABLE') execute procedure abort1();

Step 3: Verify that the firewall is active

Set up incremental synchronization

Incremental synchronization replicates the dts_audit.dts_tb_ddl_command table from a publisher to a subscriber using PostgreSQL logical replication. This lets you propagate DDL audit records across instances.

Step 1: Create a publication on the publisher

CREATE PUBLICATION my_ddl_publication FOR TABLE ONLY dts_audit.dts_tb_ddl_command;

Step 2: Create the target table on the subscriber

CREATE SCHEMA IF NOT EXISTS dts_audit;
CREATE TABLE IF NOT EXISTS dts_audit.dts_tb_ddl_command (
  event text,
  tag text, classid oid, objid oid, objsubid int,
  object_type text, schema_name text, object_identity text, is_extension bool, query text,
  username text, datname text, client_addr inet, crt_time timestamp
);

Step 3: Create a subscription on the subscriber

Replace the placeholders with the publisher's connection details:

CREATE SUBSCRIPTION my_ddl_subscriptin
  CONNECTION 'host=<host> port=<port> user=<user> password=<password> dbname=<dbname>'
  PUBLICATION my_ddl_publication;

Example:

CREATE SUBSCRIPTION my_ddl_subscriptin
  CONNECTION 'host=pgm-bpxxxxx.pg.rds.aliyuncs.com port=1433 user=test1 password=xxxxx dbname=testdb1'
  PUBLICATION my_ddl_publication;

Note

Make sure that the wal_level parameter is set to logical for your RDS instance. You can change the parameter value on the Parameters page in the ApsaraDB for RDS console and restart the instance for the change to take effect. For more information, see Use the logical subscription feature.

Step 4: Create a trigger on the subscriber

Create a trigger on the dts_audit.dts_tb_ddl_command table on the subscriber to perform incremental synchronization of DDL statements.

What's next

Use the logical subscription feature