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ノードに複製されず、標準のローカルトリガーの実行をトリガーしません。これは意図的です。これを使用して、たとえば、ストリームトリガーによってキャプチャされた変更または競合を、クラッシュセーフでそのノードに固有のテーブルにログ記録できます。実例については、: ストリームトリガーの例を参照してください。

Apply中に実行をトリガーする¶

変換トリガーは、トリガーテーブル内の受信変更ごとに最初に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を実行することをお勧めします。このプラクティスは、発散を確実に検出して修正するのに役立ちます。

行型の用語¶

これらの行タイプを使用します。

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

SOURCE_NEW は、別のノードからの新しい行です。
TARGET は、ノードにすでに存在する行、つまり競合している行です。

競合トリガー¶

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

トリガー関数が行を返す場合、アクションはターゲットに適用されます。
トリガー関数がNULL行を返した場合、アクションはスキップされます。

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

2つ以上のノード間で実際のデータの競合が発生する場合、2つ以上の同時変更が発生しています。変更が適用されると、競合の解決は各ノードで独立して行われます。これは、競合の解決が各ノードで1回発生し、ノード間に大きな時間差がある場合に発生する可能性があることを意味します。その結果、競合トリガーの複数の実行間の通信はできません。トリガーが関連するすべてのイベントに対してまったく同じ結果をもたらすことを保証するのは、競合トリガーの作成者の責任です。そうしないと、データの相違が発生します。テクニカルサポートは、PGDで提供されるisolationtesterツールを使用して、すべての競合トリガーを正式にテストすることをお勧めします。

警告

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

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

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

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

トリガー関数は、操作の種類に応じて、追加の状態情報と競合に関連するデータ行にアクセスできます。

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

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

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

競合トリガー内で主キーが更新されると、実行タイミングの違いによりユニーク制約違反エラーにつながる場合があります。したがって、競合トリガーで主キーを更新しないでください。

トリガーの変換¶

これらのトリガーは、特定のテーブルに対してデータストリームのすべての行に対して実行されることを除いて、競合トリガーに似ています。戻り値と公開された変数の動作は似ていますが、変換トリガーはターゲット行が識別される前に実行されるため、 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 行トリガーと非常によく似ていますが、次の重要な違いがあります。

変換トリガーは、着信変更ごとに呼び出されます。 BEFORE UPDATE に対してトリガーはまったく呼び出されません。テーブル内に一致する行が見つからない場合、DELETE が変更されます。
パーティションテーブルのルーティングが発生する前に、変換トリガーが呼び出されます。
変換トリガーは、通常のSQLトリガーでは使用できないSOURCE_OLD を介してルックアップキーにアクセスできます。

行の内容¶

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

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

実行順序¶

トリガーの実行順序：

トリガーを変換する -ターゲットの受信行ごとに1回実行します。
通常のトリガー - 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;
$$;