Data Transmission Service:Tetikleyiciler ve işlevler kullanarak PostgreSQL için artımlı DDL geçişi uygulayın

Ön koşullar

• Kaynak veritabanı aşağıdaki gereksinimleri karşılamalıdır:

  • Kendi yönetiminizdeki PostgreSQL: sürüm 9,4 veya üzeri.

  • ApsaraDB RDS for PostgreSQL: sürüm 10 veya üzeri.

    • ApsaraDB RDS for PostgreSQL 9,4, olay tetikleyicilerini desteklemez.

    • ApsaraDB RDS for PostgreSQL 10, 11 ve 12'nin küçük motor sürümü 20201130 veya üzeri olmalıdır.

    • ApsaraDB RDS for PostgreSQL 13'ün küçük motor sürümü 20210228 veya üzeri olmalıdır.

    Not

    Küçük motor sürümünü güncelleyin.

• Veri geçişi görevi 1 Ekim 2020 tarihinden sonra oluşturulmuş olmalıdır.

Genel bakış

PostgreSQL veritabanları arasında artımlı geçiş sırasında DTS, varsayılan olarak yalnızca DML işlemlerini (INSERT, DELETE, UPDATE) senkronize eder. DDL işlemleri senkronize edilmez.

Kaynak veritabanında bir tetikleyici ve bir işlev oluşturarak DTS'nin artımlı geçiş sırasında DDL işlemlerini yakalamasını ve senkronize etmesini sağlayabilirsiniz.

Yordam

1. Kaynak PostgreSQL veritabanında oturum açın. Bir ApsaraDB RDS for PostgreSQL örneğine bağlanın veya psql kullanın.

2. Geçiş yapılacak veritabanına geçin.

   Not

   psql'de veritabanlarını değiştirmek için \c <database_name> komutunu çalıştırın. Örnek: \c dtststdata.

3. DDL bilgilerini depolamak için bir tablo oluşturun.

   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 bilgilerini yakalamak için bir işlev oluşturun.

   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. İşlev sahibini, DTS'nin kaynak veritabanına bağlanmak için kullandığı hesapla değiştirin. Bu örnekte postgres kullanılmaktadır.

   ALTER FUNCTION public.dts_capture_ddl()
       OWNER TO postgres;

6. Genel bir olay tetikleyicisi oluşturun.

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

Sonraki adımlar

Kaynak veritabanı sürümünüze göre artımlı bir veri geçişi görevi yapılandırın. Kendi yönetiminizdeki PostgreSQL veritabanından bir ApsaraDB RDS for PostgreSQL örneğine veri geçirin.