トリガーの作成

本記事では、データ移行のソース側でDDL変更（例：テーブルフィールドの追加）が発生した場合、トリガーを使用してDDL関連イベント情報をキャプチャし、DML変更操作が正常に同期されるようにする方法について説明します。

背景

データ移行サービスを使用してソース側のPostgreSQLデータベースの増分同期を実行する場合、デフォルトではDMLステートメント（INSERT、DELETE、UPDATEを含む）の同期のみをサポートしています。しかし、ソース側でDDL変更（例：テーブルフィールドの追加）が発生した際、データ移行サービスが新しい構造情報を取得できないと、そのオブジェクトに関連する後続のDML同期が失敗し、タスクが中断する問題が発生します。そのため、まずソース側のPostgreSQLデータベースにDDLレコードテーブルとトリガーを作成し、DDL情報をキャプチャすることで、DMLの正常な同期を保証する必要があります。

使用上の制限

ソース側のPostgreSQLデータベースはV10.x以降のバージョンである必要があります。

操作手順

1. ソース側データベースにログインした後、移行対象のデータベースに切り替えます。

   PSQLツールを例にすると、コマンド\c <データベース名>を実行することで、移行対象のデータベースに切り替えます。

2. 以下の2つのテーブルを作成し、DDLイベントの変更を記録します。

   • 非DROP DDLイベント情報テーブル

     ----postgreSQL DDL EVENT TABLE---- 
     
         CREATE TABLE public.oms_non_dropped_ddl_command (
         id bigserial primary key,
         ddl_text text COLLATE pg_catalog."default",
         tag text COLLATE pg_catalog."default",
         database character varying COLLATE pg_catalog."default",
         schema character varying COLLATE pg_catalog."default",
         object_type character varying COLLATE pg_catalog."default",
         objid integer,
         top_objid integer,
         object_identity text COLLATE pg_catalog."default",
         top_object_identity text COLLATE pg_catalog."default",
         top_schema text COLLATE pg_catalog."default",
         all_children_table text COLLATE pg_catalog."default"
         );
         ALTER TABLE public.oms_non_dropped_ddl_command REPLICA IDENTITY FULL;
     

   • DROP DDLイベント情報テーブル

     ----postgreSQL DROP DDL EVENT TABLE----
     
         CREATE TABLE public.oms_dropped_ddl_command
         (
         id bigserial primary key,
         ddl_text text COLLATE pg_catalog."default",
         tag text COLLATE pg_catalog."default",
         database character varying COLLATE pg_catalog."default",
         schema character varying COLLATE pg_catalog."default",
         objid integer,
         object_type text COLLATE pg_catalog."default",
         object_name text COLLATE pg_catalog."default"
         );
         ALTER TABLE public.oms_dropped_ddl_command REPLICA IDENTITY FULL;
     

3. 以下の2種類のイベントトリガー関数を作成します。

   • 非DROP DDLイベントトリガー関数

      CREATE OR REPLACE FUNCTION public.oms_capture_ddl_for_non_dropped()
          RETURNS event_trigger
          LANGUAGE 'plpgsql'
          COST 100
          VOLATILE NOT LEAKPROOF SECURITY DEFINER
          AS $BODY$
          DECLARE ddl_text text;
          DECLARE record_object record;
          DECLARE obj record;
          DECLARE top_objid integer;
          DECLARE parent_oid integer;
          DECLARE top_object_identity text; 
          DECLARE top_schema text;
          DECLARE all_children_table text;
     
          BEGIN
          SELECT current_query() INTO ddl_text;
          FOR obj IN (SELECT * FROM pg_event_trigger_ddl_commands() WHERE TG_TAG IN ('CREATE TABLE', 'ALTER TABLE','CREATE INDEX','CREATE SCHEMA', 'COMMENT')) LOOP
          record_object:=obj;
          SELECT inhparent FROM pg_inherits WHERE inhrelid = obj.objid
          AND inhparent::regclass::text NOT LIKE 'pg_%' ORDER BY inhseqno DESC LIMIT 1 INTO parent_oid;
            WHILE parent_oid IS NOT NULL LOOP
              SELECT inhparent
              FROM pg_inherits
              WHERE inhrelid = parent_oid
              AND inhparent::regclass::text NOT LIKE 'pg_%'
              ORDER BY inhseqno DESC
              LIMIT 1
              INTO top_objid;
     
              IF top_objid IS NULL THEN
                  top_objid := parent_oid;
                exit;
              END IF;
              parent_oid := top_objid;
          END LOOP;
          IF top_objid IS NULL THEN
          top_objid :=record_object.objid;
          ELSE
          END IF;
          IF regexp_matches(ddl_text, 'ALTER\s+TABLE\s+.+\s+RENAME\s+TO\s+.+', 'i') IS NOT NULL THEN
          WITH RECURSIVE child_tables AS (
          SELECT oid AS table_objid
          FROM pg_class
          WHERE oid = top_objid
          UNION
              SELECT c.oid AS table_objid
          FROM pg_class c
          JOIN pg_inherits i ON c.oid = i.inhrelid
          JOIN child_tables ct ON i.inhparent = ct.table_objid
          )
          SELECT string_agg(table_objid::text, ', ') INTO all_children_table FROM child_tables WHERE table_objid <> top_objid;
          ELSE
          END IF;
          SELECT pn.nspname, ss.relname INTO obj from pg_catalog.pg_class ss JOIN pg_catalog.pg_namespace pn ON ss.relnamespace = pn.oid WHERE ss.oid = top_objid;
          top_object_identity:=obj.relname;
          top_schema:=obj.nspname;
          IF TG_TAG ='CREATE TABLE' AND record_object.object_type='table' THEN
          EXECUTE 'ALTER TABLE ' || record_object.object_identity || ' REPLICA IDENTITY FULL';
          ELSE
          END IF;
          INSERT INTO public.oms_non_dropped_ddl_command(id,ddl_text,tag,database,schema,object_type, objid, top_objid, object_identity,top_object_identity,top_schema,all_children_table)
          VALUES (default,ddl_text, TG_TAG,current_database(),current_schema,record_object.object_type, record_object.objid, top_objid ,record_object.object_identity,top_object_identity,top_schema,all_children_table);
          END LOOP;
          EXCEPTION 
          WHEN OTHERS THEN
              RAISE LOG 'OMS ddl trigger error during command process: %', SQLERRM;
          END
          $BODY$;
     

   • DROP DDLイベントトリガー関数

     CREATE OR REPLACE FUNCTION public.oms_capture_ddl_for_dropped()
     RETURNS event_trigger
     LANGUAGE 'plpgsql'
     COST 100
     VOLATILE NOT LEAKPROOF SECURITY DEFINER
     AS $BODY$
     DECLARE ddl_text text;
     DECLARE record_object record;
     BEGIN
     SELECT current_query() INTO ddl_text;
     FOR record_object in (select * from pg_event_trigger_dropped_objects() WHERE TG_TAG NOT IN ('ALTER TABLE','DROP PUBLICATION','ALTER PUBLICATION')) LOOP
     IF record_object.object_type = 'type' THEN
     ELSE
     INSERT INTO public.oms_dropped_ddl_command(id,ddl_text,tag,database,schema, objid, object_type, object_name)
     VALUES (default,ddl_text, TG_TAG,current_database(),current_schema, record_object.objid, record_object.object_type, record_object.object_name);
     END IF;
     END LOOP;
     EXCEPTION
     WHEN OTHERS THEN
     RAISE LOG 'OMS drop ddl trigger error during command process: %', SQLERRM;
     END
     $BODY$;
     

4. 作成した関数の所有者を、データ移行サービスがソース側データベースに接続するアカウントに変更します。ここではpostgresを例にします：

   ALTER FUNCTION public.oms_capture_ddl_for_non_dropped()
   OWNER TO postgres;
   ALTER FUNCTION public.oms_capture_ddl_for_dropped()
   OWNER TO postgres;
   

5. 以下のコマンドを実行し、グローバルイベントトリガーを作成します。

   CREATE EVENT TRIGGER oms_capture_ddl_for_non_dropped
   ON ddl_command_end
   WHEN TAG IN ('CREATE TABLE', 'ALTER TABLE','CREATE INDEX','CREATE SCHEMA','COMMENT')
   EXECUTE PROCEDURE public.oms_capture_ddl_for_non_dropped();
   
   CREATE EVENT TRIGGER oms_capture_ddl_for_dropped ON sql_drop EXECUTE PROCEDURE public.oms_capture_ddl_for_dropped();
   

次のステップ

データ移行タスク終了後、ソースデータベースにログインし、以下のコマンドを実行してトリガー関数と対応するテーブルを削除する必要があります。

drop EVENT trigger oms_capture_ddl_for_dropped;
drop EVENT trigger oms_capture_ddl_for_non_dropped;
drop function public.oms_capture_ddl_for_non_dropped();
drop function public.oms_capture_ddl_for_dropped();
drop table public.oms_non_dropped_ddl_command;
drop table public.oms_dropped_ddl_command;