Data Transmission Service:Utiliser des déclencheurs et des fonctions pour implémenter la migration incrémentielle du DDL pour les bases de données PostgreSQL

Prérequis

• La base de données source doit satisfaire aux exigences suivantes :

  • Si la base de données source est une base de données PostgreSQL autogérée, la version de la base de données doit être la V9.4 ou ultérieure.

  • Si la base de données source est une instance ApsaraDB RDS for PostgreSQL, la version de l'instance ApsaraDB RDS for PostgreSQL doit être la V10 ou ultérieure.

    • ApsaraDB RDS for PostgreSQL V9.4 ne prend pas en charge les déclencheurs d'événements.

    • Les versions du noyau d'ApsaraDB RDS for PostgreSQL V10, V11 et V12 doivent être 20201130 ou ultérieures.

    • Les versions du noyau d'ApsaraDB RDS for PostgreSQL V13 doivent être 20210228 ou ultérieures.

    Remarque

    Pour plus d'informations sur la mise à niveau de la version du noyau d'ApsaraDB RDS for PostgreSQL, consultez

    Mettre à jour la version mineure du moteur

    .

• Une tâche de migration de données a été créée après le 1er octobre 2020.

Contexte

Lorsque vous utilisez DTS pour migrer des données entre des bases de données PostgreSQL, DTS synchronise uniquement les opérations du langage de manipulation de données (DML) lors de la migration incrémentielle des données. Les opérations DML incluent INSERT, DELETE et UPDATE. DTS ne synchronise pas les opérations DDL lors de la migration incrémentielle des données.

Pour synchroniser les opérations DDL, vous pouvez créer un déclencheur et une fonction afin de récupérer les informations DDL de la base de données source. Lors de la migration incrémentielle des données, DTS migre les opérations DDL vers la base de données de destination.

Procédure

1. Connectez-vous à la base de données PostgreSQL source. Pour plus d'informations, consultez Se connecter à une instance ApsaraDB RDS for PostgreSQL ou psql.

2. Basculez vers la base de données source.

   Remarque

   L'outil psql est utilisé dans cet exemple. Vous pouvez exécuter la commande

   \c <Nom de la base de données>

   pour basculer vers la base de données source, par exemple

   \c dtststdata

   .

3. Exécutez les instructions suivantes pour créer une table qui stocke les informations 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. Exécutez les instructions suivantes pour créer une fonction qui récupère les informations 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. Modifiez le propriétaire de la fonction pour le compte utilisé pour se connecter à la base de données source, par exemple postgresql.

   ALTER FUNCTION public.dts_capture_ddl()
       OWNER TO postgres;

6. Exécutez les instructions suivantes pour créer un déclencheur d'événement global :

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

Étapes suivantes

Configurez une tâche de migration de données. Pour plus d'informations, consultez les rubriques suivantes :

• Migration incrémentielle de PostgreSQL autogéré (v10.0 ou antérieure) vers ApsaraDB for PostgreSQL

• Migrer des données incrémentielles depuis PostgreSQL autogéré vers ApsaraDB RDS for PostgreSQL

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