À propos de Openflow Connector for PostgreSQL¶

Comment les tables sont-elles répliquées ?¶

1. Introspection du schéma : le connecteur découvre les colonnes de la table source, leurs noms, leurs types, puis les valide par rapport aux limites de Snowflake et du connecteur. Les échecs de validation entraînent l’échec de cette préparation et le cycle s’achève. Une fois l’introspection du schéma correctement terminée, le connecteur crée une table de destination vide.

2. Chargement d’un instantané : le connecteur copie toutes les données disponibles dans la table source dans la table de destination. L’échec de cette préparation met fin au cycle et plus aucune donnée n’est répliquée. Une fois l’opération réussie, l’ensemble des données de la table source est disponible dans la table de destination.

3. Chargement incrémentiel : le connecteur suit en permanence les modifications apportées à la table source et les copie dans la table de destination. Cette opération se poursuit jusqu’à ce que la table soit retirée de la réplication. Une mise en échec à ce stade arrête définitivement la réplication de la table source, jusqu’à ce que le problème soit résolu.

   Note

   Ce connecteur peut être configuré pour démarrer immédiatement la réplication des modifications incrémentielles pour les tables nouvellement ajoutées, en contournant la phase de chargement des instantanés. Cette option est souvent utile lorsque vous réinstallez le connecteur dans un compte où des données répliquées existent déjà et que vous souhaitez poursuivre la réplication sans avoir à refaire des instantanés des tables.

   Pour plus de détails sur le contournement du chargement des instantanés et l’utilisation du processus de chargement incrémentiel, consultez Réplication incrémentielle.

Prise en charge des valeurs TOASTed¶

Le connecteur prend en charge la réplication de tables avec les valeurs TOAST pour les colonnes de types : array, bytea, json, jsonb, text, varchar, xml.

Chaque fois que le connecteur rencontre une valeur TOASTed dans le flux CDC, il remplace un espace réservé par défaut de __previous_value_unchanged, formaté pour le type de colonne donné, et le stocke dans la table de journal. La requête MERGE prend alors en compte les valeurs d’espaces réservés, de sorte que la table de destination contienne toujours la dernière valeur non-TOASTed.

Prise en charge des valeurs hors limites¶

Le connecteur prend en charge la réplication de tables avec des colonnes de types date, timestamp et timestamptz contenant des valeurs hors limites. Si le connecteur rencontre une valeur hors limites dans le flux CDC, il remplace un espace réservé par défaut en fonction du type de la colonne.

Statut de réplication de la table¶

Les défaillances intermédiaires, telles que les erreurs de connexion, n’empêchent pas la réplication de la table. Toutefois, les défaillances permanentes, telles que les types de données non pris en charge, empêchent la réplication de la table.

Pour résoudre les problèmes de réplication ou vérifier qu’une table a été correctement supprimée du flux de réplication, consultez le magasin d’états des tables :

1. Dans le canevas de l’environnement d’exécution Openflow, faites un clic droit sur un groupe de processeurs et choisissez Controller Services. Une table répertoriant les services du contrôleur s’affiche.

2. Localisez la ligne intitulée Table State Store, cliquez sur le bouton More sur le côté droit de la ligne, puis choisissez View State.

Une liste des tables et de leurs états actuels s’affiche. Renseignez le champ de recherche pour filtrer la liste par nom de table. Les états possibles sont :

• NEW : La table est planifiée pour la réplication, mais la réplication n’a pas commencé.

• SNAPSHOT_REPLICATION : Le connecteur copie des données existantes. Cet état s’affiche jusqu’à ce que tous les enregistrements soient stockés dans la table de destination.

• INCREMENTAL_REPLICATION : Le connecteur réplique activement les modifications. Cet état s’affiche après la fin de la réplication des instantanés et continue de s’afficher indéfiniment jusqu’à ce qu’une table soit supprimée de la réplication ou que la réplication échoue.

• FAILED : La réplication s’est définitivement arrêtée en raison d’une erreur.

Replication state for table <database_name>.<schema_name>.<table_name> changed from <old_state> to <new_state>

Si une défaillance permanente empêche la réplication d’une table, supprimez la table de la réplication. Après avoir résolu le problème à l’origine de l’échec, vous pouvez ajouter à nouveau la table à la réplication. Pour plus d’informations, consultez Redémarrer la réplication des tables.

Colonnes de métadonnées de la table de destination¶

Chaque table de destination comprend les colonnes de métadonnées suivantes qui suivent les informations de réplication :

Lignes supprimées de manière douce¶

Lorsqu’une ligne est supprimée de la table source, le connecteur ne la supprime pas physiquement de la table de destination. Au lieu de cela, la ligne est marquée comme supprimée en définissant la colonne de métadonnées _SNOWFLAKE_DELETED sur true.

Cette approche vous permet de :

• Conserver les données historiques à des fins d’audit ou de conformité.

• Interroger les enregistrements supprimés lorsque cela est nécessaire.

• Décider quand et comment supprimer définitivement des données en fonction de vos besoins.

Pour interroger uniquement les lignes actives (non supprimées), filtrez sur la colonne _SNOWFLAKE_DELETED :

SELECT * FROM my_table WHERE _SNOWFLAKE_DELETED = FALSE;

Pour interroger les lignes supprimées :

SELECT * FROM my_table WHERE _SNOWFLAKE_DELETED = TRUE;

Colonnes détruites¶

Lorsqu’une colonne est supprimée de la table source, le connecteur ne supprime pas la colonne correspondante de la table de destination. Au lieu de cela, la colonne est renommée en ajoutant le suffixe __SNOWFLAKE_DELETED pour préserver les valeurs historiques.

Par exemple, si une colonne nommée EMAIL est supprimée de la table source, elle est renommée en``EMAIL__SNOWFLAKE_DELETED`` dans la table de destination. Les lignes qui existaient avant la suppression de la colonne conservent leurs valeurs d’origine, tandis que les lignes ajoutées après la suppression affichent NULL dans cette colonne.

Vous pouvez toujours interroger les valeurs historiques à partir de la colonne renommée :

SELECT EMAIL__SNOWFLAKE_DELETED FROM my_table;

Colonnes renommées¶

En raison de limitations dans les mécanismes CDC (Change Data Capture), le connecteur ne peut pas faire la distinction entre une colonne renommée et une colonne supprimée, suivie de l’ajout d’une nouvelle colonne. Par conséquent, lorsque vous renommez une colonne dans la table source, le connecteur traite cela comme deux opérations distinctes : supprimer la colonne d’origine et ajouter une nouvelle colonne avec le nouveau nom.

Par exemple, si vous renommez une colonne A en B dans la table source, la table de destination contiendra :

• A__SNOWFLAKE_DELETED : Contient les valeurs d’avant le changement de nom. Les lignes ajoutées après le changement de nom affichent NULL dans cette colonne.

• B : Contient les valeurs d’après le changement de nom. Les lignes qui existaient avant le changement de nom affichent NULL dans cette colonne.

SELECT
    COALESCE(B, A__SNOWFLAKE_DELETED) AS A_RENAMED_TO_B
FROM my_table;

SELECT
    CASE
        WHEN B IS NOT NULL THEN B
        ELSE A__SNOWFLAKE_DELETED
    END AS A_RENAMED_TO_B
FROM my_table;

CREATE VIEW my_table_unified AS
SELECT
    *,
    COALESCE(B, A__SNOWFLAKE_DELETED) AS A_RENAMED_TO_B
FROM my_table;

Tables de journal¶

Lors de la réplication incrémentielle, les modifications de la base de données source sont d’abord écrites dans les tables de journal avant d’être fusionnées dans les tables de destination. Le connecteur ne supprime pas automatiquement les données des tables de journal, car ces données peuvent être utiles à des fins d’audit, de débogage ou de retraitement.

Les tables de journal sont créées dans le même schéma que les tables de destination correspondantes et suivent cette convention de dénomination :

<TABLE_NAME>_JOURNAL_<timestamp>_<number>

Où :

• <TABLE_NAME> est le nom de la table de destination.

• <timestamp> est l’horodatage de création au format d’époque Unix (secondes depuis le 1er janvier 1970), garantissant l’unicité.

• <number> commence à 1 et incrémente chaque fois que le schéma de la table de destination change, soit en raison de modifications de schéma dans la table source, soit en raison de modifications des filtres de colonnes.

Par exemple, si votre table de destination est SALES.ORDERS, la table du journal peut être nommée``SALES.ORDERS_JOURNAL_1705320000_1``.