ApsaraDB for OceanBase (Deprecated):Créer un déclencheur

Contexte

Lorsque vous utilisez le service de transmission de données pour la synchronisation incrémentielle depuis une base de données PostgreSQL, seules les opérations DML, y compris INSERT, DELETE et UPDATE, peuvent être synchronisées. Toutefois, si une opération DDL, telle que l'ajout d'une colonne à une table, est effectuée dans la base de données source d'une tâche de migration de données et que le service de transmission de données n'obtient pas le nouveau schéma de table, les opérations DML liées à cet objet de table échoueront lors de la synchronisation, ce qui interrompra la tâche. Par conséquent, vous devez d'abord créer des tables d'enregistrement DDL et des déclencheurs dans la base de données source afin de capturer les informations sur les opérations DDL et d'assurer la synchronisation normale des opérations DML.

Limites

La version de la base de données PostgreSQL source doit être la V10.x ou ultérieure.

Procédure

1. Connectez-vous à l'instance de base de données PostgreSQL source et basculez vers la base de données à partir de laquelle les données sont migrées.

   Dans cet exemple, psql est utilisé pour la connexion. Exécutez la commande \c <database name> dans psql pour basculer vers la base de données à partir de laquelle les données sont migrées.

2. Créez les deux tables suivantes pour enregistrer les événements DDL.

   • Table qui enregistre les informations sur les événements autres que les suppressions

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

   • Table qui enregistre les informations sur les événements de suppression

     ----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. Créez les deux fonctions de déclencheur d'événements DDL suivantes.

   • Fonction de déclencheur pour les événements autres que les suppressions

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

   • Fonction de déclencheur pour les événements de suppression

         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. Modifiez le propriétaire des fonctions créées pour qu'il corresponde au compte utilisé par le service de transmission de données pour se connecter à la base de données source. Supposons que le nom du compte soit 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. Exécutez les instructions suivantes pour créer des déclencheurs d'événements globaux :

   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();

Étapes suivantes

Après avoir publié une tâche de migration de données, connectez-vous à la base de données source et exécutez les instructions suivantes pour supprimer les fonctions de déclencheur et les tables correspondantes :

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;