Stream triggers

Stream triggers#

PGDは、ダウンストリーム/ターゲットノードでの追加データ処理に使用できる新しいタイプのトリガーを導入しました。

  • 競合トリガー

  • トランスフォームトリガー

これらのタイプのトリガーは、まとめて ストリームトリガー と呼ばれます。

ストリームトリガーは、構文がトリガーのようなものになるように設計されています。これらはPostgreSQL BEFOREトリガーアーキテクチャを活用し、PostgreSQL BEFOREトリガーと同様のパフォーマンス特性を持つ可能性があります。

通常のPostgreSQLトリガーと同様に、複数のトリガー定義で1つのトリガーファンクションを使用できます。トリガーファンクションは、CREATE FUNCTION ... RETURNS TRIGGER 形式で定義されたプログラムです。トリガーの作成に、 CREATE TRIGGER コマンドを使用する必要はありません。代わりに、特別なPGDファンクション bdr.create_conflict_trigger() および bdr.create_transform_trigger() を使用してストリームトリガーを作成します。

トリガーは、作成されると、カタログテーブルpg_trigger に表示されます。ストリームトリガーはtgisinternal = true およびtgenabled = 'D' としてマークされ、名前サフィックス’_bdrc’または’_bdrt’が付いています。 bdr.triggers ビューは、テーブルに関連するトリガー、実行されているプロシージャーの名前、それをトリガーするイベント、およびトリガータイプに関する情報を提供します。

ストリームトリガーは、通常のSQL処理では有効になっていません。このため、 ALTER TABLE ... ENABLE TRIGGER は、特定の名前バリアントとALLバリアントの両方でストリームトリガーに対してブロックされます。このメカニズムにより、トリガーは通常のSQLトリガーとして実行されません。

これらのトリガーは、ダウンストリームまたはターゲットノードで実行されます。元のノードで実行するオプションはありません。ただし、オリジンでrow_filter 式の使用を検討することができます。

また、ストリームトリガーの実行中に適用されるDMLは、他のPGDノードにレプリケートされず、標準のローカルトリガーの実行をトリガーしません。これは意図的なものです。たとえば、これを使用して、ストリームトリガーによってキャプチャされた変更または競合を、クラッシュセーフでそのノードに固有のテーブルにログ記録できます。実用的な例については、 ストリームトリガーの例 を参照してください。

適用中のトリガー実行#

変換トリガーは、トリガーテーブルに着信する変更ごとに1回実行されます。これらのトリガーは、一致するターゲット行を見つけようとする前に起動するため、非常に幅広い変換を効率的かつ一貫して適用できます。

次に、 UPDATEおよびDELETE変更の場合、ターゲット行を見つけます。ターゲット行がない場合、これらの変更タイプに対してそれ以上の処理は発生しません。

次に、テーブルレベルでレプリカトリガーとして明示的に有効にされていたノーマルトリガーを実行します。

ALTER TABLE tablename
ENABLE REPLICA TRIGGER trigger_name;

次に、潜在的な競合が存在するかどうかを判断します。その場合、そのテーブルに存在する競合トリガーを呼び出します。

欠落列の競合の解決#

変換トリガーが実行される前に、PostgreSQLは着信タプルをターゲットテーブルの行タイプと照合しようとします。

入力行には存在するがターゲットテーブルには存在しない列は、 target_column_missing タイプの競合をトリガーします。逆に、ターゲットテーブルには存在するが、入力行にない列は、 source_column_missing 競合をトリガーします。これらの2つの競合タイプのデフォルトの解決策は、それぞれignore_if_null およびuse_default_value です。

これは、ローリングスキーマのアップグレードのコンテキストで関連しますたとえば、スキーマの新しいバージョンが新しい列を導入する場合。古いバージョンのスキーマから新しいバージョンにレプリケートする場合、ソース列が欠落しており、新しく導入された列にデフォルト値を移入するuse_default_value ストラテジーが適切です。

ただし、新しいスキーマバージョンを持つノードから古いスキーマバージョンを持つノードに複製する場合、ターゲットテーブルから列が失われます。 ignore_if_null リゾルバーは、ユーザーがアップグレードされたノードの新しい列にNULL以外の値を含むタプルを挿入するとすぐにレプリケーションを中断するため、ローリングアップグレードには適していません。

この例を考慮すると、ローリングスキーマアップグレードの適切な設定は、target_column_missing 競合が発生した場合にignore リゾルバーを適用するように各ノードを構成することです。

これは、次のクエリーで行うことができます。これは、各ノードで個別に実行する必要があります。 node1 を実際のノード名に置き換えます。

SELECT bdr.alter_node_set_conflict_resolver(node1,
    target_column_missing, ignore);

データ損失と発散リスク#

競合リゾルバーをignore に設定すると、データの損失とクラスターの発散が発生する可能性があります。

次の例を考えます。テーブルt はノード1および2に存在しますが、その列col はノード1にのみ存在します。

競合リゾルバーがignore に設定されている場合、ノード1にc がnullでない行、たとえば (pk=1, col=100) が存在する場合があります。その行はノード2にレプリケートされ、列c の値は破棄されますたとえば、(pk=1) 。

列c がノード2のテーブルに追加される場合、最初に既存のすべての行でNULLに設定され、上記で検討される行は(pk=1, col=NULL) になります。 pk=1 を持つ行はすべてのノードで同一ではなくなり、したがって、クラスターは発散します。

ノード2にレプリケートされた行にはcol=NULL があるため、デフォルトのignore_if_null リゾルバーはこのリスクの影響を受けません。

この例に基づいて、 ignore リゾルバーが使用されたローリングスキーマアップグレードの最後に、クラスター全体に対して LiveCompare を実行することをお勧めします。この実践は、発散を確実に検出および修正するのに役立ちます。

行型の用語#

PGDは次の行タイプを使用します。

  • SOURCE_OLD は更新前の行、つまりキーです。

  • SOURCE_NEW は、別のノードからの新しい行です。

  • TARGET は、ノードに既に存在する行、つまり競合する行です。

競合トリガー#

競合トリガーは、PGDによって競合が検出されると実行されます。彼らは、紛争が発生したときに何が起こるかを決定します。

  • トリガーファンクションが行を返した場合、アクションはターゲットに適用されます。

  • トリガーファンクションがNULL行を返した場合、アクションはスキップされます。

たとえば、DELETE に対してトリガーが呼び出された場合、DELETE をスキップする場合、トリガーはNULLを返します。 DELETE を続行する場合は、行値を返します。SOURCE_OLD またはTARGET のいずれかが機能します。競合するオペレーションがINSERT またはUPDATE で、選択された解決が競合する行を削除することである場合、トリガーは明示的に削除を実行し、NULLを返す必要があります。トリガーファンクションは、選択した他のSQLアクションを実行できますが、これらのアクションはローカルにのみ適用され、レプリケートされません。

2つ以上のノード間で実際のデータの競合が発生すると、2つ以上の変更が同時に発生します。変更が適用されると、競合の解決は各ノードで独立して発生します。これは、競合の解決が各ノードで1回発生し、それらの間で大幅な時間差で発生する可能性があることを意味します。その結果、競合トリガーの複数の実行間の通信は不可能です。トリガーが関連するすべてのイベントに対してまったく同じ結果を与えることを確認するのは、競合トリガーの作成者の責任です。そうしないと、データの相違が発生します。

警告

  • 単一のテーブルに複数の競合トリガーを指定できますが、それらは個別のイベントと一致する必要があります。つまり、各競合は、単一の競合トリガーのみと一致する必要があります。 - 同じテーブルの同じイベントに一致する複数のトリガーはお勧めしません。これらは一貫性のない動作を引き起こす可能性があり、将来のリリースでは許可されません。

同じ競合トリガーが複数のイベントと一致する場合、トリガーでTG_OP 変数を使用して、競合を生成した操作を特定できます。

デフォルトでは、 PGDは行のレプリケーションオリジンの変更を監視することにより競合を検出します。したがって、1つの変更のみが発生している場合でも、競合トリガーを呼び出すことができます。この場合、実際の競合は存在しないため、この競合検出メカニズムは偽陽性の競合を生成する可能性があります。競合トリガーは、これらをすべて同様に処理する必要があります。

場合によっては、タイムスタンプ競合検出は競合をまったく検出しない場合があります。たとえば、UPDATE の直後にDELETE が発生するコンカレントUPDATE /DELETE では、最初にUPDATE を、次にDELETE を参照するノードは競合を認識しません。競合が発生しない場合、競合トリガーは呼び出されません。同じ状況で、行バージョンの競合検出を使用すると、競合が表示され、競合トリガーで処理できます。

トリガーファンクションは、操作タイプに応じて、追加の状態情報と競合に関係するデータ行にアクセスします。

  • INSERT では、競合トリガーはソースからSOURCE_NEW 行とTARGET 行にアクセスできます。

  • UPDATE では、競合トリガーはソースからSOURCE_OLD およびSOURCE_NEW 行とTARGET 行にアクセスできます。

  • DELETE では、競合トリガーはソースからSOURCE_OLD 行とTARGET 行にアクセスできます。

そのオペレーションに値が存在する場合、ファンクションbdr.trigger_get_row() を使用して、SOURCE_OLD 、SOURCE_NEW 、またはTARGET 行を取得できます。

競合トリガーの変更はトランザクション的に発生し、構成変更のレプリケーション中にグローバルDMLロックによって保護されます。この動作は、ALTER TABLE の一部のバリアントの処理方法に似ています。

競合トリガー内で主キーが更新されると、実行タイミングの違いにより一意制約違反エラーが発生する場合があります。したがって、競合トリガーでの主キーの更新は避けてください。

Transformトリガー#

これらのトリガーは、特定のテーブルに対するデータストリームのすべての行に対して実行される点を除き、競合トリガーに似ています。戻り値と公開変数の動作は似ていますが、変換トリガーはターゲット行が識別される前に実行されるため、 TARGET 行はありません。

PGDの各テーブルで複数の変換トリガーを指定できます。変換トリガーはアルファベット順に実行されます。

変換トリガーは行をフィルターして取り除くことができ、必要に応じて追加の操作を実行できます。列の値を変更したり、NULL に設定したりできます。戻り値によって、実行される次のアクションが決まります。

  • トリガーファンクションが行を返した場合、ターゲットに適用されます。

  • トリガーファンクションがNULL 行を返した場合、それ以上に実行するアクションはありません。未実行のトリガーは実行されません。

  • トリガー機能は、自分の選択に応じて他のアクションを実行できます。

トリガーファンクションは、追加の状態情報と、競合に関係する行にアクセスします。

  • INSERT では、変換トリガーはソースからSOURCE_NEW 行にアクセスできます。

  • UPDATE では、変換トリガーはソースからSOURCE_OLD およびSOURCE_NEW 行にアクセスできます。

  • DELETE では、変換トリガーはソースからSOURCE_OLD 行にアクセスできます。

ファンクションbdr.trigger_get_row() を使用して、SOURCE_OLD またはSOURCE_NEW 行を取得できます。このタイプのトリガーは、このようなターゲット行が存在する場合が識別される前に実行されるため、 TARGET 行は使用できません。

変換トリガーは通常のBEFORE行トリガーに非常に似ていますが、次の重要な違いがあります。

  • 変換トリガーは、入ってくる変更ごとに呼び出されます。 UPDATE およびDELETE では、 BEFOREトリガーはまったく呼び出されず、テーブル内で一致する行が見つからない場合、DELETE が変更されます。

  • 変換トリガーは、パーティションテーブルルーティングが発生する前に呼び出されます。

  • 変換トリガーは、SOURCE_OLD を介してルックアップキーにアクセスできます。これは、通常のSQLトリガーでは使用できません。

行の内容#

SOURCE_NEW 、SOURCE_OLD 、およびTARGET の内容は、オペレーション、テーブルのREPLICA IDENTITY設定、およびターゲットテーブルの内容によって異なります。

TARGET行は、競合トリガーでのみ使用できます。 TARGET行には、ターゲットテーブルでUPDATE またはDELETE を適用するときに行が見つかった場合にのみデータが含まれます。行が見つからない場合、TARGETはNULL です。

実行オーダー#

トリガーの実行順序

  • Transformトリガー —ターゲットへの入力行ごとに1回実行します。

  • 通常トリガー —行ごとに1回実行します。

  • 競合トリガー —競合が存在する行ごとに1回実行します。

ストリームトリガーの例#

update_if_newer 競合リゾルバーと同様の動作を提供する競合トリガー

CREATE OR REPLACE FUNCTION update_if_newer_trig_func
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
    IF (bdr.trigger_get_committs(TARGET) >
        bdr.trigger_get_committs(SOURCE_NEW)) THEN
    RETURN TARGET;
    ELSIF
        RETURN SOURCE;
    END IF;
END;
$$;

カウンター列にデルタ変更を適用し、他のすべての列にSOURCE_NEWを使用する競合トリガー。

CREATE OR REPLACE FUNCTION delta_count_trg_func
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
    DELTA bigint;
    SOURCE_OLD record;
    SOURCE_NEW record;
    TARGET record;
BEGIN
    SOURCE_OLD := bdr.trigger_get_row(SOURCE_OLD);
    SOURCE_NEW := bdr.trigger_get_row(SOURCE_NEW);
    TARGET := bdr.trigger_get_row(TARGET);

    DELTA := SOURCE_NEW.counter - SOURCE_OLD.counter;
    SOURCE_NEW.counter = TARGET.counter + DELTA;

    RETURN SOURCE_NEW;
END;
$$;

すべての変更を適用する代わりにログテーブルにログを記録する変換トリガー

CREATE OR REPLACE FUNCTION log_change
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
    SOURCE_NEW record;
    SOURCE_OLD record;
    COMMITTS timestamptz;
BEGIN
    SOURCE_NEW := bdr.trigger_get_row(SOURCE_NEW);
    SOURCE_OLD := bdr.trigger_get_row(SOURCE_OLD);
    COMMITTS := bdr.trigger_get_committs(SOURCE_NEW);

    IF (TG_OP = INSERT) THEN
        INSERT INTO log SELECT I, COMMITTS, row_to_json(SOURCE_NEW);
    ELSIF (TG_OP = UPDATE) THEN
        INSERT INTO log SELECT U, COMMITTS, row_to_json(SOURCE_NEW);
    ELSIF (TG_OP = DELETE) THEN
        INSERT INTO log SELECT D, COMMITTS, row_to_json(SOURCE_OLD);
    END IF;

    RETURN NULL; -- do not apply the change
END;
$$;

この例は、信頼できるサイト、優先ノード、またはAlways Wins解決としても知られる、信頼できるソースの競合検出を実装する競合トリガーを示しています。 bdr.trigger_get_origin_node_id() ファンクションを使用して、3つ以上のノードで動作するソリューションを提供します。

CREATE OR REPLACE FUNCTION test_conflict_trigger()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
    SOURCE  record;
    TARGET  record;

    TRUSTED_NODE    bigint;
    SOURCE_NODE     bigint;
    TARGET_NODE     bigint;
BEGIN
    TARGET := bdr.trigger_get_row(TARGET);
    IF (TG_OP = DELETE)
        SOURCE := bdr.trigger_get_row(SOURCE_OLD);
    ELSE
        SOURCE := bdr.trigger_get_row(SOURCE_NEW);
    END IF;

    TRUSTED_NODE := current_setting(customer.trusted_node_id);

    SOURCE_NODE := bdr.trigger_get_origin_node_id(SOURCE_NEW);
    TARGET_NODE := bdr.trigger_get_origin_node_id(TARGET);

    IF (TRUSTED_NODE = SOURCE_NODE) THEN
        RETURN SOURCE;
    ELSIF (TRUSTED_NODE = TARGET_NODE) THEN
        RETURN TARGET;
    ELSE
        RETURN NULL; -- do not apply the change
    END IF;
END;
$$;
目次