Documentation PostgreSQL 14.22 » Programmation serveur » Décodage logique (Logical Decoding) » Plugins de sortie de décodage logique

Précédent | Niveau supérieur | Suivant
49.5. Catalogues systèmes liés au décodage logique | Sommaire | 49.7. Écrivains de sortie de décodage logique

## 49.6. Plugins de sortie de décodage logique

Un exemple de plugin de sortie peut être trouvé dans le sous-répertoire `contrib/test_decoding` de l'arborescence du code source de PostgreSQL.

### 49.6.1. Fonction d'initialisation

Un plugin de sortie est chargé en chargeant dynamiquement une bibliothèque partagée avec comme nom de base le nom du plugin de sortie. Le chemin de recherche de bibliothèque habituel est utilisé pour localiser cette bibliothèque. Pour fournir les callbacks de plugins de sortie requis et pour indiquer que la bibliothèque est effectivement un plugin de sortie, elle doit fournir une fonction nommée `_PG_output_plugin_init`. Une structure est passée à cette fonction qui doit la remplir avec les pointeurs des fonctions de callback pour chaque action individuelle.

    typedef struct OutputPluginCallbacks
    {
        LogicalDecodeStartupCB startup_cb;
        LogicalDecodeBeginCB begin_cb;
        LogicalDecodeChangeCB change_cb;
        LogicalDecodeTruncateCB truncate_cb;
        LogicalDecodeCommitCB commit_cb;
        LogicalDecodeMessageCB message_cb;
        LogicalDecodeFilterByOriginCB filter_by_origin_cb;
        LogicalDecodeShutdownCB shutdown_cb;
        LogicalDecodeFilterPrepareCB filter_prepare_cb;
        LogicalDecodeBeginPrepareCB begin_prepare_cb;
        LogicalDecodePrepareCB prepare_cb;
        LogicalDecodeCommitPreparedCB commit_prepared_cb;
        LogicalDecodeRollbackPreparedCB rollback_prepared_cb;
        LogicalDecodeStreamStartCB stream_start_cb;
        LogicalDecodeStreamStopCB stream_stop_cb;
        LogicalDecodeStreamAbortCB stream_abort_cb;
        LogicalDecodeStreamPrepareCB stream_prepare_cb;
        LogicalDecodeStreamCommitCB stream_commit_cb;
        LogicalDecodeStreamChangeCB stream_change_cb;
        LogicalDecodeStreamMessageCB stream_message_cb;
        LogicalDecodeStreamTruncateCB stream_truncate_cb;
    } OutputPluginCallbacks;

    typedef void (*LogicalOutputPluginInit) (struct OutputPluginCallbacks *cb);

Les callbacks `begin_cb`, `change_cb` et `commit_cb` sont obligatoires, alors que `startup_cb`, `filter_by_origin_cb`, `truncate_cb` et `shutdown_cb` sont facultatifs. Si `truncate_cb` n'est pas configuré mais que `TRUNCATE` doit être décodé, l'action sera ignorée.

Un plugin de sortie peut aussi définir les fonctions pour accepter le flux de grosses transactions en cours. Les callbacks `stream_start_cb`, `stream_stop_cb`, `stream_abort_cb`, `stream_commit_cb`, `stream_change_cb` et `stream_prepare_cb` sont obligatoires, alors que `stream_message_cb` et `stream_truncate_cb` sont facultatifs.

Un plugin de sortie peut aussi définir les fonctions pour accepter les validations en deux phases, ce qui permet de décoder un `PREPARE TRANSACTION`. Les callbacks `begin_prepare_cb`, `prepare_cb`, `stream_prepare_cb`, `commit_prepared_cb` et `rollback_prepared_cb` sont obligatoires, alors que `filter_prepare_cb` est facultatif.

### 49.6.2. Capacités

Pour décoder, formater et afficher les changements, les plugins de sortie peuvent utiliser une grande partie de l'infrastructure habituelle des processus clients, y compris l'appel aux fonctions de sortie. Les accès en lecture seule aux relations est permis du moment que les relations accédées ont été créées par `initdb` dans le schéma `pg_catalog`, ou ont été marquées comme tables du catalogue pour l'utilisateur en utilisant :

    ALTER TABLE table_catalogue_utilisateur SET (user_catalog_table = true);
    CREATE TABLE autre_table_catalogue(data text) WITH (user_catalog_table = true);

Notez que l'accès aux tables utilisateurs ou aux tables systèmes dans le plugin de sortie doit se faire uniquement via les API de parcours `systable_*`. L'accès via les API de parcours `heap_*` renverre une erreur. De plus, toute action amenant à l'affectation d'un identifiant de transaction est prohibée. Ceci, parmi d'autres, inclut l'écriture dans les tables, la réalisation de modifications DDL, et l'appel à `pg_current_xact_id()`.

### 49.6.3. Modes de sortie

Les fonctions callbacks des plugins en sortie peuvent renvoyer des données au consommateur dans des formats pratiquement arbitraires. Pour certains cas d'utilisation, comme la visualisation des changements en SQL, le renvoi des données dans un type de données qui peut contenir des données arbitraires (par exemple du `bytea`) est complexe. Si le plugin en sortie renvoie seulement les données au format texte dans l'encodage du serveur, il peut déclarer cela en configurant `OutputPluginOptions.output_type` à `OUTPUT_PLUGIN_TEXTUAL_OUTPUT` au lieu de `OUTPUT_PLUGIN_BINARY_OUTPUT` dans la fonction callback de démarrage. Dans ce cas, toutes les données doivent être dans l'encodage du serveur pour qu'un champ de type `text` puisse les contenir. Ceci est vérifié dans les constructions comprenant les assertions.

### 49.6.4. Callbacks de plugin de sortie

Un plugin de sortie est notifié des changements arrivant au travers de différents callbacks qu'il doit fournir.

Les transactions concurrentes sont décodées dans l'ordre dans lequel elles sont validées, et seuls les changements appartenant à une transaction spécifique sont décodés entre les callbacks `begin` et `commit`. Les transactions qui ont été explicitement ou implicitement annulées ne sont jamais décodées. Les savepoints validés sont inclus dans la transaction les contenant, dans l'ordre dans lequel ils ont été effectués dans la transaction. Une transaction qui est préparée pour une validation en deux phases avec `PREPARE TRANSACTION` sera aussi décodée si les callbacks du plugin de sortie nécessaires sont fournies pour le décodage. Il est possible que la transaction préparée courante en cours de décodage soit annulée en parallèle via une commande `ROLLBACK PREPARED`. Dans ce cas, le décodage logique de cette transaction sera lui-aussi annulé. Tous les changements d'une telle transaction seront ignorés une fois que l'annulation est détectée et que la fonction callback `prepare_cb` est appelée. De ce fait, même dans le cas d'une annulation en parallèle, suffisamment d'informations sont fournies au plugin de sortie pour gérer correctement un `ROLLBACK PREPARED` une fois qu'il est décodé.

### Note

Seules les transactions qui ont été synchronisées sur disque de manière sûre seront décodées. Cela peut amener à ce qu'un `COMMIT` ne soit pas immédiatement décodé lors d'un appel à `pg_logical_slot_get_changes()` juste après celui-ci quand `synchronous_commit` est positionné à `off`.

#### 49.6.4.1. Callback de démarrage

Le callback facultatif `startup_cb` est appelé chaque fois qu'un slot de réplication est créé ou qu'on lui demande de fournir les flux de changement, indépendamment du nombre de changements qui sont prêt à être fournis.

    typedef void (*LogicalDecodeStartupCB) (struct LogicalDecodingContext *ctx,
                                            OutputPluginOptions *options,
                                            bool is_init);

Le paramètre `is_init` sera positionné à true quand le slot de réplication est créé, et à false sinon. `options` pointe vers une structure d'options que le plugin de sortie peut positionner :

    typedef struct OutputPluginOptions
    {
        OutputPluginOutputType output_type;
        bool        receive_rewrites;
    } OutputPluginOptions;

`output_type` doit être positionné soit à `OUTPUT_PLUGIN_TEXTUAL_OUTPUT` ou à `OUTPUT_PLUGIN_BINARY_OUTPUT`. Voir aussi Section 49.6.3. Si `receive_rewrites` vaut true, le plugin de sortie sera aussi appelé pour les modifications réalisées par des réécritures du fichier HEAP lors de certaines opérations DDL. Ceci est intéressant pour les plugins qui gèrent la réplication DDL mais ils nécessitent une gestion particulière.

Le callback de démarrage devrait valider les options présentes dans `ctx->output_plugin_options`. Si le plugin de sortie a besoin d'avoir un état, il peut utiliser `ctx->output_plugin_private` pour le stocker.

#### 49.6.4.2. Callback d'arrêt

Le callback facultatif `shutdown_cb` est appelé chaque fois qu'un slot de réplication anciennement actif n'est plus utilisé et peut être utilisé pour désallouer les ressources privées du plugin de sortie. Le slot n'est pas nécessairement supprimé, le flux est juste arrêté.

    typedef void (*LogicalDecodeShutdownCB) (struct LogicalDecodingContext *ctx);

#### 49.6.4.3. Callback de début de transaction

Le callback obligatoire `begin_cb` est appelé chaque fois que le début d'une transaction validée a été décodé. Les transactions annulées et leur contenu ne sont pas décodés.

    typedef void (*LogicalDecodeBeginCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn);

Le paramètre `txn` contient des métadonnées sur la transaction, comme l'heure à laquelle elle a été validée et son XID.

#### 49.6.4.4. Callback de fin de transaction

Le callback obligatoire `commit_cb` est appelé chaque fois qu'une transaction validée a été décodée. Le callback `change_cb` aura été appelé avant cela pour chacune des lignes modifiées, s'il y en a eu.

    typedef void (*LogicalDecodeCommitCB) (struct LogicalDecodingContext *ctx,
                                           ReorderBufferTXN *txn,
                                           XLogRecPtr commit_lsn);

#### 49.6.4.5. Callback de modification

Le callback obligatoire `change_cb` est appelé pour chacune des modifications de ligne au sein d'une transaction, qu'il s'agisse d'un `INSERT`, `UPDATE` ou `DELETE`. Même si la commande d'origine a modifié plusieurs ligne en une seule instruction, le callback sera appelé pour chaque ligne individuellement. Le callback `change_cb` pourrait accéder aux tables systèmes et utilisateurs pour aider au traitement en sortie des détails de modification de la ligne. Dans le cas du décodage d'une transaction préparée (mais pas encore validée) ou du décodage d'une transaction non validée, ce callback de modification pourrait aussi renvoyer une erreur à cause de l'annulation simultanée de cette même transaction. Dans ce cas, le décodage logique de cette transaction annulée est arrêté proprement.

    typedef void (*LogicalDecodeChangeCB) (struct LogicalDecodingContext *ctx,
                                           ReorderBufferTXN *txn,
                                           Relation relation,
                                           ReorderBufferChange *change);

Les paramètres `ctx` et `txn` ont le même contenu que pour les callbacks `begin_cb` et `commit_cb`, mais en plus le descripteur de relation `relation` pointe vers la relation à laquelle appartient la ligne et une structure `change` décrivant les modifications de ligne y est passée.

### Note

Seules les changements dans les tables définies par les utilisateurs qui sont journalisées (voir `UNLOGGED`) et non temporaires(voir `TEMPORARY` ou `TEMP`) peuvent être extraites avec le décodage logique.

#### 49.6.4.6. Callback Truncate

La fonction callback `truncate_cb` est appelée pour la commande `TRUNCATE`.

    typedef void (*LogicalDecodeTruncateCB) (struct LogicalDecodingContext *ctx,
                                             ReorderBufferTXN *txn,
                                             int nrelations,
                                             Relation relations[],
                                             ReorderBufferChange *change);

Les paramètres sont identiques à ceux du callback `change_cb`. Néanmoins, comme les actions du `TRUNCATE` sur les tables liées par clés étrangères doivent être exécutées ensembles, ce callback reçoit un tableau de relations au lieu d'une seule relation. Voir la description de l'instruction TRUNCATE pour les détails.

#### 49.6.4.7. Fonction de filtre sur l'origine

La fonction optionnelle `filter_by_origin_cb` est appelée pour déterminer si les données rejouées à partir de `origin_id` ont un intérêt pour le plugin de sortie.

    typedef bool (*LogicalDecodeFilterByOriginCB) (
        struct LogicalDecodingContext *ctx,
        RepNodeId origin_id
    );

Le paramètre `ctx` a le même contenu que pour les autres fonctions. Aucune information mais l'origine est disponible. Pour signaler que les changements provenant du nœud sont hors de propos, elle renvoie true, ce qui permet de les filtrer. Elle renvoie false dans les autres cas. Les autres fonctions ne seront pas appelées pour les transactions et changements qui ont été filtrées.

Ceci est utile pour implémenter des solutions de réplication en cascade ou des solutions de réplication multi-directionnelles. Filtrer par rapport à l'origine perment d'empêcher la réplication dans les deux sesns des mêmes modifications dans ce type de configuration. Quand les transactions et les modifications contiennent aussi des informations sur l'origine, le filtre via cette fonction est beaucoup plus efficace.

#### 49.6.4.8. Fonctions personnalisées de message générique

La fonction (callback) `message_cb` est appelée quand un message de décodage logique a été décodé.

    typedef void (*LogicalDecodeMessageCB) (
        struct LogicalDecodingContext *,
        ReorderBufferTXN *txn,
        XLogRecPtr message_lsn,
        bool transactional,
        const char *prefix,
        Size message_size,
        const char *message
    );

Le paramètre `txn` contient des méta-informations sur la transaction, comme l'horodatage à laquelle la transaction a été validée et son identifiant (XID). Notez néanmoins qu'il peut être NULL quand le message n'est pas transactionnel et que le XID n'a pas encore été affecté dans la transaction qui a tracé le message. Le `lsn` a la position du message dans les WAL. Le paramètre `transactional` indique si le message a été envoyé de façon transactionnelle ou non. De façon similaire au callback de changement, dans le cas du décodage d'une transaction préparée (mais pas encore validée) ou du décodage d'une transaction non validée, ce callback de message pourrait renvoyer une erreur en cas d'annulation simultanée de cette même transaction. Dans ce cas, le décodage logique de cette transaction annulée est stoppé proprement. Le paramètre `prefix` est un préfixe arbitraire terminé par un caractère nul qui peut être utilisé pour identifier les messages intéressants pour le plugin courant. Et enfin, le paramètre `message` détient le message réel de taille `message_size`.

Une attention particulière doit être portée à l'unicité du préfixe que le plugin de sortie trouve intéressant. Utiliser le nom de l'extension ou du plugin de sortie est souvent un bon choix.

#### 49.6.4.9. Fonction de filtre du PREPARE

La fonction callback facultative `filter_prepare_cb` est appelée pour déterminer si les données qui font partie de la transaction de validation en deux phases en cours doivent être considérées pour le décoage à cette étape préparatoire ou plus tard comme une transaction standard en une phase au moment du `COMMIT PREPARED`. Pour signaler que le décodage doit être ignoré, renvoyez `true` ; `false` dans le cas contraire. Quand la fonction callback est définie, `false` est supposé (autrement dit, pas de filtre toutes les transactions utilisant la validation en deux phases sont décodées aussi en deux phases).

    typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx,
                                                  TransactionId xid,
                                                  const char *gid);

Le paramètre `ctx` a le même contenu que pour les autres fonctions callbacks. Les paramètres `xid` et `gid` fournissent deux façons différentes d'identifier la transaction. La commande `COMMIT PREPARED` ou `ROLLBACK PREPARED` ultérieure intègre les identifiants, autorisant au plugin de sortie le choix de celui à utiliser.

La fonction callback pourrait être appelée plusieurs fois par transaction pour décoder, et doit fournir la même réponse statique pour une paire donnée de `xid` et de `gid` à chaque fois qu'elle est appelée.

#### 49.6.4.10. Fonction de préparation de début de la transaction

La fonction callback requise `begin_prepare_cb` est appelée à chaque fois que le début d'une transaction préparée a été décodée. Le champ `gid`, qui fait partie du paramètre `txn`, peut être utilisé dans cette fonction callback pour vérifier si le plugin a déjà reçu cette commande `PREPARE`, auquel cas elle peut renvoyer une erreur ou ignorer les changements restants de la transaction.

    typedef void (*LogicalDecodeBeginPrepareCB) (struct LogicalDecodingContext *ctx,
                                                 ReorderBufferTXN *txn);

#### 49.6.4.11. Fonction de transaction préparée

La fonction callback requise `prepare_cb` est appelée à chaque fois qu'une transaction préparée pour une validation en deux phrases a été décodée. La fonction callback `change_cb` sera appelée avant celle-ci pour toutes les lignes modifiées, à condition que des lignes aient été modifiées. Le champ `gid`, qui fait partie du paramètre `txn`, peut être utilisé dans cette fonction callback.

    typedef void (*LogicalDecodePrepareCB) (struct LogicalDecodingContext *ctx,
                                            ReorderBufferTXN *txn,
                                            XLogRecPtr prepare_lsn);

#### 49.6.4.12. Fonction pour le COMMIT PREPARED d'une transaction

La fonction callback requise `commit_prepared_cb` est appelée quand une commande `COMMIT PREPARED` d'une transaction a été décodée. Le champ `gid`, qui fait partie du paramètre `txn`, peut être utilisé dans cette fonction callback.

    typedef void (*LogicalDecodeCommitPreparedCB) (struct LogicalDecodingContext *ctx,
                                                   ReorderBufferTXN *txn,
                                                   XLogRecPtr commit_lsn);

#### 49.6.4.13. Fonction d'annulation de transaction préparée

La fonction callback requise `rollback_prepared_cb` est appelée à chaque qu'une transaction `ROLLBACK PREPARED` a été décodée. Le champ `gid`, qui fait partie du paramètre `txn`, peut être utilisé dans cette fonction callback. Les paramètres `prepare_end_lsn` et `prepare_time` peuvent être utilisés pour vérifier si le plugin a reçu la commande `PREPARE TRANSACTION` auquel cas il peut appliquer l'annulation. Sinon, il peut ignorer l'opération d'annulation. Le paramètre `gid` seul n'est pas suffisant parce que le nœud en amont peut avoir une transaction préparée avec le même identifiant.

    typedef void (*LogicalDecodeRollbackPreparedCB) (struct LogicalDecodingContext *ctx,
                                                     ReorderBufferTXN *txn,
                                                     XLogRecPtr prepare_end_lsn,
                                                     TimestampTz prepare_time);

#### 49.6.4.14. Fonction de début de flux

La fonction callback `stream_start_cb` est appelée à l'ouverture d'un bloc de changements en flux à partir d'une transaction en cours.

    typedef void (*LogicalDecodeStreamStartCB) (struct LogicalDecodingContext *ctx,
                                                ReorderBufferTXN *txn);

#### 49.6.4.15. Fonction d'arrêt de flux

La fonction callback `stream_stop_cb` est appelée lors de la fermeture d'un bloc de changements en flux provenant d'une transaction en cours.

    typedef void (*LogicalDecodeStreamStopCB) (struct LogicalDecodingContext *ctx,
                                               ReorderBufferTXN *txn);

#### 49.6.4.16. Fonction d'annulation du flux

La fonction callback `stream_abort_cb` est appelée lors de l'annulation d'une transaction en flux précédente.

    typedef void (*LogicalDecodeStreamAbortCB) (struct LogicalDecodingContext *ctx,
                                                ReorderBufferTXN *txn,
                                                XLogRecPtr abort_lsn);

#### 49.6.4.17. Fonction de préparation du flux

La fonction callback `stream_prepare_cb` est appelée pour préparer une transaction en flux précédente pour faire partie d'une validation en deux phases.

    typedef void (*LogicalDecodeStreamPrepareCB) (struct LogicalDecodingContext *ctx,
                                                  ReorderBufferTXN *txn,
                                                  XLogRecPtr prepare_lsn);

#### 49.6.4.18. Fonction de validation d'un flux

La fonction callback `stream_commit_cb` est appelée pour valider une transaction en flux précédente.

    typedef void (*LogicalDecodeStreamCommitCB) (struct LogicalDecodingContext *ctx,
                                                 ReorderBufferTXN *txn,
                                                 XLogRecPtr commit_lsn);

#### 49.6.4.19. Fonction de changement du flux

La fonction callback `stream_change_cb` est appelée pour un changement dans un bloc de changements de flux (démarqué par des appels à `stream_start_cb` et `stream_stop_cb`). Les modifications réelles ne sont pas affichées car la transaction peut annuler plus tard et que nous ne décodons par les changements pour les transactions annulées.

    typedef void (*LogicalDecodeStreamChangeCB) (struct LogicalDecodingContext *ctx,
                                                 ReorderBufferTXN *txn,
                                                 Relation relation,
                                                 ReorderBufferChange *change);

#### 49.6.4.20. Fonction de message du flux

La fonction callback `stream_message_cb` est appelée lors de l'envoi d'un message générique dans un bloc de changements en flux (démarqué par des appels à `stream_start_cb` et `stream_stop_cb`). Le contenu du message pour des messages transactionnels n'est pas affiché car la transaction peut s'annuler plus tard et que nous ne décodons pas les changements pour les transactions annulées.

    typedef void (*LogicalDecodeStreamMessageCB) (struct LogicalDecodingContext *ctx,
                                                  ReorderBufferTXN *txn,
                                                  XLogRecPtr message_lsn,
                                                  bool transactional,
                                                  const char *prefix,
                                                  Size message_size,
                                                  const char *message);

#### 49.6.4.21. Fonction troncage du flux

La fonction callback `stream_truncate_cb` est appelée pour une commande `TRUNCATE` dans un bloc de changements en flux (démarqué par des appels à `stream_start_cb` et `stream_stop_cb`).

    typedef void (*LogicalDecodeStreamTruncateCB) (struct LogicalDecodingContext *ctx,
                                                   ReorderBufferTXN *txn,
                                                   int nrelations,
                                                   Relation relations[],
                                                   ReorderBufferChange *change);

Les paramètres sont analogues à ceux de la fonction callback `stream_change_cb`. Néanmoins, comme les actions de la commande `TRUNCATE` sur des tables connectées par des clés étrangères doivent s'exécuter ensemble, cette fonction callback reçoit un tableau de relations au lieu d'une seule relation. Voir la description de l'instruction TRUNCATE pour les détails.

### 49.6.5. Fonction pour produire une sortie

Pour pouvoir produire une sortie, les plugins de sortie peuvent écrire des données dans le tampon de sortie `StringInfo` dans `ctx->out` dans les callbacks `begin_cb`, `commit_cb` ou `change_cb`. Avant d'écrire dans le tampon de sortie, `OutputPluginWrite(ctx, last_write)` doit avoir été appelé pour effectuer l'écriture. `last_write` indique si une écriture particulière était la dernière écriture du callback.

L'exemple suivant montre comment sortir des données pour le consommateur d'un plugin de sortie :

    OutputPluginPrepareWrite(ctx, true);
    appendStringInfo(ctx->out, "BEGIN %u", txn->xid);
    OutputPluginWrite(ctx, true);