Stream triggers
===============

PGD introduces new types of triggers that you can use for additional
data processing on the downstream/target node:

-  Conflict triggers

-  Transform triggers

Together, these types of triggers are known as *stream triggers*.

Stream triggers are designed to be trigger-like in syntax. They leverage
the PostgreSQL BEFORE trigger architecture and are likely to have
similar performance characteristics as PostgreSQL BEFORE Triggers.

Multiple trigger definitions can use one trigger function, just as with
normal PostgreSQL triggers. A trigger function is a program defined in
this form: ``CREATE FUNCTION ... RETURNS TRIGGER`` . Creating the
trigger doesn’t require use of the ``CREATE TRIGGER`` command. Instead,
create stream triggers using the special PGD functions :ref:`bdr.create_conflict_trigger() <Stream triggers manipulation interfaces>`  and
 :ref:`bdr.create_transform_trigger() <Stream triggers manipulation interfaces>`  .

Once created, the trigger is visible in the catalog table ``pg_trigger``
. The stream triggers are marked as ``tgisinternal = true`` and
``tgenabled = 'D'`` and have the name suffix ‘\_bdrc’ or ‘\_bdrt’. The
view ``bdr.triggers`` provides information on the triggers in relation
to the table, the name of the procedure that’s being executed, the event
that triggers it, and the trigger type.

Stream triggers aren’t enabled for normal SQL processing. Because of
this, the ``ALTER TABLE ... ENABLE TRIGGER`` is blocked for stream
triggers in both its specific name variant and the ALL variant. This
mechanism prevents the trigger from executing as a normal SQL trigger.

These triggers execute on the downstream or target node. There’s no
option for them to execute on the origin node. However, you might want
to consider the use of ``row_filter`` expressions on the origin.

Also, any DML that’s applied while executing a stream trigger isn’t
replicated to other PGD nodes and doesn’t trigger the execution of
standard local triggers. This is intentional. You can use it, for
example, to log changes or conflicts captured by a stream trigger into a
table that is crash-safe and specific to that node. See :ref:`Stream triggers examples <Stream triggers examples>` 
for a working example.

Trigger execution during Apply
------------------------------

Transform triggers execute first—once for each incoming change in the
triggering table. These triggers fire before we attempt to locate a
matching target row, allowing a very wide range of transforms to be
applied efficiently and consistently.

Next, for UPDATE and DELETE changes, we locate the target row. If
there’s no target row, then no further processing occurs for those
change types.

We then execute any normal triggers that previously were explicitly
enabled as replica triggers at table level:

.. code:: sql

   ALTER TABLE tablename
   ENABLE REPLICA TRIGGER trigger_name;

We then decide whether a potential conflict exists. If so, we then call
any conflict trigger that exists for that table.

Missing-column conflict resolution
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Before transform triggers are executed, PostgreSQL tries to match the
incoming tuple against the row-type of the target table.

Any column that exists on the input row but not on the target table
triggers a conflict of type ``target_column_missing`` . Conversely, a
column existing on the target table but not in the incoming row triggers
a ``source_column_missing`` conflict. The default resolutions for those
two conflict types are respectively ``ignore_if_null`` and
``use_default_value`` .

This is relevant in the context of rolling schema upgrades, for example,
if the new version of the schema introduces a new column. When
replicating from an old version of the schema to a new one, the source
column is missing, and the ``use_default_value`` strategy is
appropriate, as it populates the newly introduced column with the
default value.

However, when replicating from a node having the new schema version to a
node having the old one, the column is missing from the target table.
The ``ignore_if_null`` resolver isn’t appropriate for a rolling upgrade
because it breaks replication as soon as the user inserts a tuple with a
non-NULL value in the new column in any of the upgraded nodes.

In view of this example, the appropriate setting for rolling schema
upgrades is to configure each node to apply the ``ignore`` resolver in
case of a ``target_column_missing`` conflict.

You can do this with the following query, which you must execute
separately on each node. Replace ``node1`` with the actual node name.

.. code:: sql

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

Data loss and divergence risk
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Setting the conflict resolver to ``ignore`` can lead to data loss and
cluster divergence.

Consider the following example: table ``t`` exists on nodes 1 and 2, but
its column ``col`` exists only on node 1.

If the conflict resolver is set to ``ignore`` , then there can be rows
on node 1 where ``c`` isn’t null, for example, ``(pk=1, col=100)`` .
That row is replicated to node 2, and the value in column ``c`` is
discarded, for example, ``(pk=1)`` .

If column ``c`` is then added to the table on node 2, it’s at first set
to NULL on all existing rows, and the row considered above becomes
``(pk=1, col=NULL)`` . The row having ``pk=1`` is no longer identical on
all nodes, and the cluster is therefore divergent.

The default ``ignore_if_null`` resolver isn’t affected by this risk
because any row replicated to node 2 has ``col=NULL`` .

Based on this example, we recommend running LiveCompare against the
whole cluster at the end of a rolling schema upgrade where the
``ignore`` resolver was used. This practice helps to ensure that you
detect and fix any divergence.

Terminology of row-types
------------------------

We use these row-types:

-  ``SOURCE_OLD`` is the row before update, that is, the key.

-  ``SOURCE_NEW`` is the new row coming from another node.

-  ``TARGET`` is the row that exists on the node already, that is, the
   conflicting row.

Conflict triggers
-----------------

Conflict triggers execute when a conflict is detected by PGD. They
decide what happens when the conflict occurs.

-  If the trigger function returns a row, the action is applied to the
   target.

-  If the trigger function returns a NULL row, the action is skipped.

For example, if the trigger is called for a ``DELETE`` , the trigger
returns NULL if it wants to skip the ``DELETE`` . If you want the
``DELETE`` to proceed, then return a row value: either ``SOURCE_OLD`` or
``TARGET`` works. When the conflicting operation is either ``INSERT`` or
``UPDATE`` , and the chosen resolution is to delete the conflicting row,
the trigger must explicitly perform the deletion and return NULL. The
trigger function can perform other SQL actions as it chooses, but those
actions are only applied locally, not replicated.

When a real data conflict occurs between two or more nodes, two or more
concurrent changes are occurring. When the changes are applied, the
conflict resolution occurs independently on each node. This means the
conflict resolution occurs once on each node and can occur with a
significant time difference between them. As a result, communication
between the multiple executions of the conflict trigger isn’t possible.
It’s the responsibility of the author of the conflict trigger to ensure
that the trigger gives exactly the same result for all related events.
Otherwise, data divergence occurs. Technical Support recommends that you
formally test all conflict triggers using the isolationtester tool
supplied with PGD.

..  Warning::
   -   You can specify multiple conflict triggers on a single table, but they must match a distinct event. That is, each conflict must match only a single conflict trigger. -   We don't recommend multiple triggers matching the same event on the same table. They might result in inconsistent behavior and will not be allowed in a future release.

If the same conflict trigger matches more than one event, you can use
the ``TG_OP`` variable in the trigger to identify the operation that
produced the conflict.

By default, PGD detects conflicts by observing a change of replication
origin for a row. Hence, you can call a conflict trigger even when only
one change is occurring. Since, in this case, there’s no real conflict,
this conflict detection mechanism can generate false-positive conflicts.
The conflict trigger must handle all of those identically.

In some cases, timestamp conflict detection doesn’t detect a conflict at
all. For example, in a concurrent ``UPDATE`` /``DELETE`` where the
``DELETE`` occurs just after the ``UPDATE`` , any nodes that see first
the ``UPDATE`` and then the ``DELETE`` don’t see any conflict. If no
conflict is seen, the conflict trigger are never called. In the same
situation but using row version conflict detection, a conflict is seen,
which a conflict trigger can then handle.

The trigger function has access to additional state information as well
as the data row involved in the conflict, depending on the operation
type:

-  On ``INSERT`` , conflict triggers can access the ``SOURCE_NEW`` row
   from the source and ``TARGET`` row.

-  On ``UPDATE`` , conflict triggers can access the ``SOURCE_OLD`` and
   ``SOURCE_NEW`` row from the source and ``TARGET`` row.

-  On ``DELETE`` , conflict triggers can access the ``SOURCE_OLD`` row
   from the source and ``TARGET`` row.

You can use the function ``bdr.trigger_get_row()`` to retrieve
``SOURCE_OLD`` , ``SOURCE_NEW`` , or ``TARGET`` rows, if a value exists
for that operation.

Changes to conflict triggers happen transactionally and are protected by
global DML locks during replication of the configuration change,
similarly to how some variants of ``ALTER TABLE`` are handled.

If primary keys are updated inside a conflict trigger, it can sometimes
lead to unique constraint violations errors due to a difference in
timing of execution. Hence, avoid updating primary keys in conflict
triggers.

Transform triggers
------------------

These triggers are similar to conflict triggers, except they’re executed
for every row on the data stream against the specific table. The
behavior of return values and the exposed variables is similar, but
transform triggers execute before a target row is identified, so there’s
no ``TARGET`` row.

You can specify multiple transform triggers on each table in PGD.
Transform triggers execute in alphabetical order.

A transform trigger can filter away rows, and it can do additional
operations as needed. It can alter the values of any column or set them
to ``NULL`` . The return value decides the further action taken:

-  If the trigger function returns a row, it’s applied to the target.

-  If the trigger function returns a ``NULL`` row, there’s no further
   action to perform. Unexecuted triggers never execute.

-  The trigger function can perform other actions as it chooses.

The trigger function has access to additional state information as well
as rows involved in the conflict:

-  On ``INSERT`` , transform triggers can access the ``SOURCE_NEW`` row
   from the source.

-  On ``UPDATE`` , transform triggers can access the ``SOURCE_OLD`` and
   ``SOURCE_NEW`` row from the source.

-  On ``DELETE`` , transform triggers can access the ``SOURCE_OLD`` row
   from the source.

You can use the function ``bdr.trigger_get_row()`` to retrieve
``SOURCE_OLD`` or ``SOURCE_NEW`` rows. ``TARGET`` row isn’t available,
since this type of trigger executes before such a target row is
identified, if any.

Transform triggers look very similar to normal BEFORE row triggers but
have these important differences:

-  A transform trigger gets called for every incoming change. BEFORE
   triggers aren’t called at all for ``UPDATE`` and ``DELETE`` changes
   if a matching row in a table isn’t found.

-  Transform triggers are called before partition table routing occurs.

-  Transform triggers have access to the lookup key via ``SOURCE_OLD`` ,
   which isn’t available to normal SQL triggers.

Row contents
------------

The ``SOURCE_NEW`` , ``SOURCE_OLD`` , and ``TARGET`` contents depend on
the operation, REPLICA IDENTITY setting of a table, and the contents of
the target table.

The TARGET row is available only in conflict triggers. The TARGET row
contains data only if a row was found when applying ``UPDATE`` or
``DELETE`` in the target table. If the row isn’t found, the TARGET is
``NULL`` .

Execution order
---------------

Execution order for triggers:

-  Transform triggers — Execute once for each incoming row on the
   target.

-  Normal triggers — Execute once per row.

-  Conflict triggers — Execute once per row where a conflict exists.

Stream triggers examples
------------------------

A conflict trigger that provides similar behavior as the
``update_if_newer`` conflict resolver:

.. code:: sql

   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;
   $$;

A conflict trigger that applies a delta change on a counter column and
uses SOURCE_NEW for all other columns:

.. code:: sql

   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;
   $$;

A transform trigger that logs all changes to a log table instead of
applying them:

.. code:: sql

   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;
   $$;

This example shows a conflict trigger that implements trusted source
conflict detection, also known as trusted site, preferred node, or
Always Wins resolution. It uses the ``bdr.trigger_get_origin_node_id()``
function to provide a solution that works with three or more nodes.

.. code:: sql

   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;
   $$;