Data Transmission Service:Use triggers e funções para implementar migração incremental de DDL em bancos de dados PostgreSQL

Pré-requisitos

• O banco de dados source deve atender aos seguintes requisitos:

  • Se o banco de dados source for uma instância PostgreSQL autogerenciada, a versão do banco de dados deve ser V9.4 ou superior.

  • Se o banco de dados source for uma instância ApsaraDB RDS for PostgreSQL, a versão da instância deve ser V10 ou superior.

    • O ApsaraDB RDS for PostgreSQL V9.4 não oferece suporte a event triggers.

    • As versões de kernel do ApsaraDB RDS for PostgreSQL V10, V11 e V12 devem ser 20201130 ou posteriores.

    • A versão de kernel do ApsaraDB RDS for PostgreSQL V13 deve ser 20210228 ou posterior.

    Nota

    Para obter mais informações sobre como atualizar a versão do kernel do ApsaraDB RDS for PostgreSQL, consulte

    Atualizar a versão secundária do mecanismo

    .

• A tarefa de migração de dados foi criada após 1º de outubro de 2020.

Informações de contexto

Ao usar o DTS para migrar dados entre bancos de dados PostgreSQL, apenas as operações de linguagem de manipulação de dados (DML) são sincronizadas durante a migração incremental. As operações DML incluem INSERT, DELETE e UPDATE. O DTS não sincroniza operações de DDL nessa fase.

Para sincronizar operações de DDL, crie um trigger e uma função que capturem as informações de DDL do banco de dados source. Durante a migração incremental, o DTS replica essas operações para o banco de dados de destino.

Procedimento

1. Faça login no banco de dados PostgreSQL source. Para mais informações, consulte Conectar-se a uma instância PostgreSQL ou psql.

2. Mude para o banco de dados source.

   Nota

   Este exemplo usa a ferramenta psql. Execute o comando

   \c <Database name>

   para mudar para o banco de dados source, por exemplo:

   \c dtststdata

   .

3. Execute as instruções a seguir para criar uma tabela que armazene as informações de 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. Execute as instruções a seguir para criar uma função que capture as informações de 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 -- ALTER TABLE schema.TABLE REPLICA IDENTITY 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(16)),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. Altere o proprietário da função para a conta usada na conexão com o banco de dados source, por exemplo, postgresql.

   ALTER FUNCTION public.dts_capture_ddl()
       OWNER TO postgres;

6. Execute as instruções a seguir para criar um event trigger global:

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

Próximas etapas

Configure uma tarefa de migração de dados. Para mais informações, consulte os tópicos a seguir:

• Migrar incrementalmente PostgreSQL autogerenciado (v10.0 ou anterior) para ApsaraDB for PostgreSQL

• Migrar dados incrementais de PostgreSQL autogerenciado para ApsaraDB RDS for PostgreSQL

drop EVENT trigger dts_intercept_ddl;
drop function public.dts_capture_ddl();
drop table public.dts_ddl_command;