ApsaraDB RDS:Use event triggers to implement a DDL recycle bin, firewall, and incremental subscription and synchronization

DDL recycle bin

1. Execute the following commands to create the table, function, and related triggers.

   /* external/rds_ddl_publication/rds_ddl_publication--1.0.sql */
   
   -- Create a schema.
   CREATE SCHEMA IF NOT EXISTS dts_audit;
   -- Create a table for DDL records.
   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()
   );
   -- Create a function for event triggers.
   create or replace function dts_audit.dts_func_ddl_command() returns event_trigger as $$
   declare v1 text;
   is_superuser bool = false;
   r record; 
   begin
       -- DDL commands from superusers are not recorded. 
       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();
       -- RAISE NOTICE 'ddl event:%, command:%', tg_event, tg_tag;
       -- Note: Because ddl_command_end cannot collect the details of the drop statement, sql_drop is used.
       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;
       if TG_EVENT='sql_drop' then
           -- To avoid repeated collection, 'ALTER TABLE' and 'ALTER FOREIGN TABLE' are filtered out.
           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;
   -- ddl_command_end event trigger.
   CREATE EVENT TRIGGER pg_get_ddl_command on ddl_command_end EXECUTE PROCEDURE dts_audit.dts_func_ddl_command();
   -- pg_get_ddl_drop event trigger.
   CREATE EVENT TRIGGER pg_get_ddl_drop on sql_drop EXECUTE PROCEDURE dts_audit.dts_func_ddl_command();
   -- Grant privileges 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 you execute the preceding commands, the DDL statements that you execute are recorded in the `dts_audit.dts_tb_ddl_command` table.

2. Execute a DDL statement to test whether the change is recorded.

   CREATE TABLE tb_test(id int);
   SELECT * FROM dts_audit.dts_tb_ddl_command;

DDL firewall

You can create event triggers based on your requirements. Use the `ddl_command_start` event type to block the execution of specific DDL statements.

1. Create the trigger function.

   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;
   $$;

2. Create a trigger to block statements that create or delete tables.

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

3. Log on to the instance as the user `test1` and try to create a table to test whether the operation is blocked.

   Note

   The DDL statement is successfully blocked.

DDL incremental subscription and synchronization

On the publisher, the executed DDL statements are stored in the `dts_audit.dts_tb_ddl_command` table. The subscriber can then read these records to perform synchronization.

1. Run the publish command on the publisher.

   CREATE PUBLICATION my_ddl_publication FOR TABLE ONLY dts_audit.dts_tb_ddl_command;

2. On the subscriber, create the same 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, datname text, client_addr inet , crt_time timestamp );

3. On the subscriber, create a subscription.

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

   Note

   Make sure that the wal_level parameter of the instance is set to logical. You can modify this parameter on the Parameters page in the RDS console. After you modify the parameter, you must restart the instance for the change to take effect. For more information, see Logical subscription.

   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;

4. On the subscriber, create a trigger for the `dts_audit.dts_tb_ddl_command` table to implement incremental DDL synchronization.