dbview_replication_log.sql

  CREATE OR REPLACE FUNCTION do_replication_log(remote_connection_info text, local_connection_info text, schema_name text, last_replication_date_hour timestamp with time zone)
    RETURNS timestamp with time zone AS
  $BODY$
  DECLARE

    RegistrosLoop RECORD;
    remote_connection_id text;
    local_connection_id text;
    last_timestamp timestamp with time zone;
    has_more_records boolean = false;
    rows_limit int = 10000;

  BEGIN
      
      local_connection_id = 'do_local_replication_log';
      remote_connection_id = 'do_remote_replication_log';
      
      PERFORM public.dblink_connect_u(remote_connection_id,remote_connection_info);
      PERFORM public.dblink_connect_u(local_connection_id,local_connection_info);

      LOOP

          has_more_records = FALSE;
          FOR RegistrosLoop IN 
          
              SELECT transaction.*
              FROM public.dblink(remote_connection_id,
                      'SELECT trl_datehour, trl_statements 
                       FROM transactionlog 
                       WHERE trl_datehour > '|| quote_literal(last_replication_date_hour)|| ' LIMIT ' || rows_limit)
              AS transaction( trl_datehour timestamp with time zone,
                             trl_statements text)

          LOOP
            
              has_more_records = true;

              RAISE DEBUG '%', RegistrosLoop.trl_statements;

              PERFORM public.dblink_exec(local_connection_id, 'set search_path to ' || schema_name || ';');
              PERFORM public.dblink_exec(local_connection_id, RegistrosLoop.trl_statements);

              last_timestamp = RegistrosLoop.trl_datehour;
              last_replication_date_hour = last_timestamp;

          END LOOP;

          RAISE DEBUG 'HAS MORE RECORDS TO IMPORT: %', has_more_records;
          EXIT WHEN has_more_records IS false;


      END LOOP;
      
      PERFORM public.dblink_disconnect(local_connection_id);
      PERFORM public.dblink_disconnect(remote_connection_id);

      RETURN last_timestamp;
  
  END;

  $BODY$
    LANGUAGE plpgsql VOLATILE
    COST 100;

Assim funciona:

...
SELECT transaction.*
FROM public.dblink(remote_connection_id,
E'SELECT trl_datehour, regexp_replace(trl_statements, '(BEGIN;|COMMIT;)', '', 'g')
FROM transactionlog
WHERE trl_datehour > '|| quote_literal(last_replication_date_hour)|| ' LIMIT ' || rows_limit)
AS transaction( trl_datehour timestamp with time zone,
trl_statements text)

...