Data Transmission Service:トリガーと関数を使用した PostgreSQL の増分 DDL 移行の実装

手順

1. ソース PostgreSQL データベースにログインします。 ApsaraDB RDS for PostgreSQL インスタンスに接続するか、psql を使用してください。

2. 移行対象のデータベースに切り替えます。

   説明

   psql で \c <database_name> を実行してデータベースを切り替えます。 例：\c dtststdata。

3. DDL 情報を格納するためのテーブルを作成します。

   CREATE TABLE public.dts_ddl_command
   (
       ddl_text text COLLATE pg_catalog."default",
      id bigserial primary key,
      event text COLLATE pg_catalog."default",
      tag text COLLATE pg_catalog."default",
      username character varying 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",
      object_name character varying COLLATE pg_catalog."default",
      client_address character varying COLLATE pg_catalog."default",
      client_port integer,
      event_time timestamp with time zone,
      txid_current character varying(128) COLLATE pg_catalog."default",
      message text COLLATE pg_catalog."default"
   );

4. DDL 情報をキャプチャするための関数を作成します。

   CREATE FUNCTION public.dts_capture_ddl()
       RETURNS event_trigger
       LANGUAGE 'plpgsql'
       COST 100
       VOLATILE NOT LEAKPROOF SECURITY DEFINER
   AS $BODY$
     declare ddl_text text;
     declare max_rows int := 10000;
     declare current_rows int;
     declare pg_version_95 int := 90500;
     declare pg_version_10 int := 100000;
     declare current_version int;
     declare object_id varchar;
     declare alter_table varchar;
     declare record_object record;
     declare message text;
     declare pub RECORD;
   begin
   
     select current_query() into ddl_text;
   
     if TG_TAG = 'CREATE TABLE' then -- テーブルのレプリケーションアイデンティティを FULL に設定します。
       show server_version_num into current_version;
       if current_version >= pg_version_95 then
         for record_object in (select * from pg_event_trigger_ddl_commands()) loop
           if record_object.command_tag = 'CREATE TABLE' then
             object_id := record_object.object_identity;
           end if;
         end loop;
       else
         select btrim(substring(ddl_text from '[ \t\r\n\v\f]*[c|C][r|R][e|E][a|A][t|T][e|E][ \t\r\n\v\f]*.*[ \t\r\n\v\f]*[t|T][a|A][b|B][l|L][e|E][ \t\r\n\v\f]+(.*)\(.*'),' \t\r\n\v\f') into object_id;
       end if;
       if object_id = '' or object_id is null then
         message := 'CREATE TABLE, but ddl_text=' || ddl_text || ', current_query=' || current_query();
       else
         alter_table := 'ALTER TABLE ' || object_id || ' REPLICA IDENTITY FULL';
         message := 'alter_sql=' || alter_table;
         execute alter_table;
       end if;
       if current_version >= pg_version_10 then
         for pub in (select * from pg_publication where pubname like 'dts_sync_%') loop
           raise notice 'pubname=%',pub.pubname;
           BEGIN
             execute 'alter publication ' || pub.pubname || ' add table ' || object_id;
           EXCEPTION WHEN OTHERS THEN
           END;
         end loop;
       end if;
     end if;
   
     insert into public.dts_ddl_command(id,event,tag,username,database,schema,object_type,object_name,client_address,client_port,event_time,ddl_text,txid_current,message)
     values (default,TG_EVENT,TG_TAG,current_user,current_database(),current_schema,'','',inet_client_addr(),inet_client_port(),current_timestamp,ddl_text,cast(TXID_CURRENT() as varchar(128)),message);
   
     select count(id) into current_rows from public.dts_ddl_command;
     if current_rows > max_rows then
       delete from public.dts_ddl_command where id in (select min(id) from public.dts_ddl_command);
     end if;
   end
   $BODY$;

5. 関数の所有者を、DTS がソースデータベースへの接続に使用するアカウントに変更します。 この例では postgres を使用します。

   ALTER FUNCTION public.dts_capture_ddl()
       OWNER TO postgres;

6. グローバルイベントトリガーを作成します。

   CREATE EVENT TRIGGER dts_intercept_ddl ON ddl_command_end
   EXECUTE PROCEDURE public.dts_capture_ddl();