Entre de mauvaises mains, les triggers sont des instruments de malheur. Leur temps d’exécution s’immisce dans les temps de réponse de vos ordres SQL sans que vous ne puissiez rien y faire…

Seulement, voilà, ré-écrire une application est extrêmement coûteux  et il faut trouver des moyens rapides de faire d’évoluer ses architectures pour améliorer l’expérience des utilisateurs sans sacrifice démesuré. Introduire de l’asynchronisme dans les codes de bases de données est une solution simple qui permet d’arriver à des résultats intéressants sans « tout casser »…

Le dessin ci-dessous illustre l’idée d’une telle transformation qui peut dans certain cas être redoutable : elle permet de lisser la charge, d’extraire le temps dédié aux triggers du temps de réponse et garantit malgré tout qu’il n’y a pas de perte d’information :

Note importante :
Ces 2 patterns ne sont pas du tout identiques ! Au delà du caractère asynchrone, l’introduction de ce mécanisme s’effectue au prix des 3 premières propriétés ACID de la transaction originale. En outre, pour manipuler les valeurs des données de la ligne, il faut la passer en paramètre dans la file d’attente AQ et vous ne pouvez pas du tout y accéder en manipulant les paramètres « :NEW » ou « :OLD ». Autant dire que l’impact de ce type de transformation n’est pas à négliger non plus. 

Cet article présente comment extraire une procédure (f) d’un trigger et l’intégrer dans un mécanisme asynchrone tout en garantissant grâce à Oracle Streams sont exécution. Vous noterez que dans le cas qui nous intéresse, la base de données n’a même pas à être en mode ARCHIVELOG et que vous n’avez pas besoin d’être en Enterprise Edition. Vous ne capturez pas à partir des fichiers de redolog…

Schéma original

Commençons par un cas relativement classique : une table avec un trigger dans un schéma de référence. Vous prendrez l’utilisateur SCOTT pour cela ; voici le script dont il s’agit :

sqlplus / as sysdba

grant execute on dbms_lock to scott;

connect scott/tiger

create table x_track(timesecs number);

create or replace procedure f(timesecs number) is
begin
   dbms_lock.sleep(timesecs);
   insert into x_track values (timesecs);      
end;
/

create table x(text varchar2(10),
               timesecs number);

create or replace trigger x_afterinsert 
   before insert on x for each row
begin
   f(:NEW.timesecs);
end;
/

set timing on 
insert into x values ('Wait 5s',5);

1 row created.

Elapsed: 00:00:05.03

commit;

Préparer l’environnement pour utiliser Streams

Pour changer l’architecture de notre application, il faut utiliser plusieurs fonctionnalités d’Oracle Streams :

• d’abord une file d’attente Advanced Queuing (AQ)
• ensuite un processus d’apply qui se déclenchera automatiquement lorsqu’un message sera envoyé dans la file d’attente
• enfin un trigger sur la table qui alimente la file d’attente lorsqu’une données est mise à jour dans la table

Pour que cela fonctionne, il faut que le type de la file d’attente soit de type SYS.ANYDATA.

Note:
Le processus d’apply Streams est le seul mécanisme « standard » qui garantit que l’ensemble des messages sera traité et ceux même en cas de crash d’instance. En outre, vous pouvez traiter ces messages dans l’ordre dans lesquels ils sont validés.
Il est également possible de développer un programme qui soit un souscripteur de la file d’attente. Dans ce cas, il vous appartient de gérer son fonctionnement et les erreurs associées.

Créer la file d’attente

Pour commencer, on crée donc une file d’attente de type SYS.ANYDATA et on donne à l’utilisateur SCOTT l’autorisation d’utiliser le package dbms_aq pour mettre des messages dans cette file d’attente :

connect / as sysdba
grant execute on dbms_aq to scott;

begin
   -- Create a queue table to hold the event queue.
   dbms_aqadm.create_queue_table(
              queue_table        => 'SCOTT.TRIGGERQ_TABLE',
              queue_payload_type => 'SYS.ANYDATA',
              multiple_consumers => TRUE,
              comment            => 'Queue Table to Manage Trigger Events');
   -- Create the ETT event queue.
   dbms_aqadm.create_queue (
              queue_name  => 'SCOTT.TRIGGERQ',
              queue_table => 'SCOTT.TRIGGERQ_TABLE');
   -- Start the event queue.
   dbms_aqadm.start_queue (queue_name => 'SCOTT.TRIGGERQ');
end;
/

Type, Message Handler et Apply

Pour la suite, il faut créer un type qui permette d’échanger des informations entre les évènements capturés sur la table, i.e. les ordres DML et le processus d’apply qui contiendra la logique du « trigger asynchrone ». Le moyen le plus générique consiste à réutiliser le type SYS.ROW$_RECORD puisqu’il peut contenir n’importe quelle information d’un ordre DML sur n’importe quelle table (sauf restrictions sur les type de Streams).

Pour simplifier cet exemple qui ne met en jeu qu’une seule table, vous utiliserez simplement un type qui contient :

• l’ordre DML
• la nouvelle valeur de la colonne timesecs

connect scott/tiger
create or replace type x_type is object
   (operation varchar2(10),
    timesecs  number);
/

Le type créé, vous allez développer la logique qui s’exécute lorsque le message est envoyé ; il s’agit d’un « custom message handler » dont voici un exemple ; celui-ci déclenche simplement la fonction f après quelques vérifications :

create or replace procedure manage_xevent (event_msg IN sys.anydata)
as
   payload x_type;
   x number;
begin
   -- retrieve and process message body
   x:=event_msg.getobject(payload);
   -- if the message is due to an insert, trigger f
   if (payload.operation='INSERT') then
      f(payload.timesecs);
   end if;
end manage_xevent;
/

Dernière étape du paramétrage, vous devez vous connecter SYS, ou un administrateur Streams, pour créer un processus d’APPLY qui soit abonné à la file d’attente et déclenche le handler créé précédemment :

begin
   -- Create the apply
   dbms_apply_adm.create_apply(
       queue_name=>'SCOTT.TRIGGERQ',
       apply_name=>'SCOTT_TRIGGERQ_APPLY',
       message_handler=>'SCOTT.manage_xevent');
   -- set commit_serialization to FULL
   dbms_apply_adm.set_parameter('SCOTT_TRIGGERQ_APPLY','commit_serialization', 'FULL');
   -- Start the Apply
   dbms_apply_adm.start_apply(apply_name=>'SCOTT_TRIGGERQ_APPLY');
end;
/

Trigger simple

Pour terminer, la configuration qui ne s’appuie pas sur les fichiers journaux, il faut créer un trigger simple qui se déclenche sur les ordres INSERT et mette un message dans la file d’attente. Pour cela, on crée une procédure enqueue :

connect scott/tiger

create or replace procedure enqueue_x_msg(
                      operation varchar2,
                      timesecs  number) is
         l_enqueue_options     DBMS_AQ.enqueue_options_t;
         l_message_properties  DBMS_AQ.message_properties_t;

       l_message_handle      RAW(16);
         v_payload             x_type;
         v_payload_anydata     sys.anydata;
begin
   if operation='INSERT' then
      v_payload:=x_type(operation,timesecs);
      v_payload_anydata:=ANYDATA.ConvertObject(v_payload);

      -- Create an agent to send the message
      l_message_properties.SENDER_ID := SYS.AQ$_AGENT(
            name     => 'LOCAL_AGENT',
            address  => NULL,
            protocol => NULL);

       -- Enqueue the message message
       DBMS_AQ.enqueue(queue_name => 'SCOTT.TRIGGERQ',
            enqueue_options       => l_enqueue_options,
            message_properties    => l_message_properties,
            payload               => v_payload_anydata,
            msgid                 => l_message_handle);
       -- The COMMIT has to be managed by the source transaction
    end if;
end;
/

Puis on crée le trigger qui utilise la procédure précédente :

create or replace trigger x_afterinsert 
   before insert on x for each row
begin
   enqueue_x_msg('INSERT', :NEW.timesecs);
end;
/

Tester le fonctionnement du « trigger asynchrone »

Vous pouvez tester votre application en exécutant des ordres DML suivant sur la table x :

delete from x_track;
commit;

set timing on
insert into x values ('Wait 10s',10);
commit;
Elapsed: 00:00:00.00

select * from x_track;

no rows selected

Attendez 10 secondes et relancez l’ordre précédent :

select * from x_track;

  TIMESECS
----------
        10

Par ailleurs, vous pouvez effectuer un test avec un crash d’instance comme celui-ci ; ça fonctionne toujours :

connect / as sysdba
delete from scott.x_track;
commit;
insert into scott.x values ('Wait 60s',60);
commit;
startup force;
select * from scott.x_track;

Pour superviser vos « triggers asynchrones », vous superviserez votre processus d’apply…