conflict.c File Reference

#include "postgres.h"
#include "access/commit_ts.h"
#include "access/genam.h"
#include "access/tableam.h"
#include "catalog/heap.h"
#include "catalog/pg_am.h"
#include "catalog/pg_namespace.h"
#include "catalog/toasting.h"
#include "executor/executor.h"
#include "pgstat.h"
#include "replication/conflict.h"
#include "replication/worker_internal.h"
#include "storage/lmgr.h"
#include "utils/lsyscache.h"

Include dependency graph for conflict.c:

Go to the source code of this file.

Data Structures
struct	ConflictLogColumnDef

Macros
#define	NUM_CONFLICT_ATTRS   ((AttrNumber) lengthof(ConflictLogSchema))

Typedefs
typedef struct ConflictLogColumnDef	ConflictLogColumnDef

Functions
	StaticAssertDecl (lengthof(ConflictLogDestNames)==CONFLICT_LOG_DEST_ALL+1, "ConflictLogDestNames length mismatch")
static int	errcode_apply_conflict (ConflictType type)
static void	errdetail_apply_conflict (EState *estate, ResultRelInfo *relinfo, ConflictType type, TupleTableSlot *searchslot, TupleTableSlot *localslot, TupleTableSlot *remoteslot, Oid indexoid, TransactionId localxmin, ReplOriginId localorigin, TimestampTz localts, StringInfo err_msg)
static void	get_tuple_desc (EState *estate, ResultRelInfo *relinfo, ConflictType type, char **key_desc, TupleTableSlot *localslot, char **local_desc, TupleTableSlot *remoteslot, char **remote_desc, TupleTableSlot *searchslot, char **search_desc, Oid indexoid)
static char *	build_index_value_desc (EState *estate, Relation localrel, TupleTableSlot *slot, Oid indexoid)
static TupleDesc	create_conflict_log_table_tupdesc (void)
Oid	create_conflict_log_table (Oid subid, char *subname, Oid subowner)
ConflictLogDest	GetConflictLogDest (const char *dest)
bool	GetTupleTransactionInfo (TupleTableSlot *localslot, TransactionId *xmin, ReplOriginId *localorigin, TimestampTz *localts)
void	ReportApplyConflict (EState *estate, ResultRelInfo *relinfo, int elevel, ConflictType type, TupleTableSlot *searchslot, TupleTableSlot *remoteslot, List *conflicttuples)
void	InitConflictIndexes (ResultRelInfo *relInfo)
static void	append_tuple_value_detail (StringInfo buf, List *tuple_values)

Variables
const char *const	ConflictLogDestNames []
static const ConflictLogColumnDef	ConflictLogSchema []
static const char *const	ConflictTypeNames []

Macro Definition Documentation

NUM_CONFLICT_ATTRS

#define NUM_CONFLICT_ATTRS   ((AttrNumber) lengthof(ConflictLogSchema))

Definition at line 82 of file conflict.c.

Typedef Documentation

ConflictLogColumnDef

typedef struct ConflictLogColumnDef ConflictLogColumnDef

Function Documentation

append_tuple_value_detail()

static void append_tuple_value_detail ( StringInfo  buf, List *  tuple_values  )

static

Definition at line 367 of file conflict.c.

{
    bool        first = true;
 
    Assert(buf != NULL && tuple_values != NIL);
 
    foreach_ptr(char, tuple_value, tuple_values)
    {
        /*
         * Skip if the value is NULL. This means the current user does not
         * have enough permissions to see all columns in the table. See
         * get_tuple_desc().
         */
        if (!tuple_value)
            continue;
 
        /* standard SQL punctuation, not translated */
        if (!first)
            appendStringInfoString(buf, ", ");
 
        appendStringInfoString(buf, tuple_value);
        first = false;
    }
}

References appendStringInfoString(), Assert, buf, fb(), foreach_ptr, and NIL.

Referenced by errdetail_apply_conflict().

build_index_value_desc()

static char * build_index_value_desc ( EState *  estate, Relation  localrel, TupleTableSlot *  slot, Oid  indexoid  )

static

Definition at line 779 of file conflict.c.

{
    char       *index_value;
    Relation    indexDesc;
    Datum       values[INDEX_MAX_KEYS];
    bool        isnull[INDEX_MAX_KEYS];
    TupleTableSlot *tableslot = slot;
 
    if (!tableslot)
        return NULL;
 
    Assert(CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
 
    indexDesc = index_open(indexoid, NoLock);
 
    /*
     * If the slot is a virtual slot, copy it into a heap tuple slot as
     * FormIndexDatum only works with heap tuple slots.
     */
    if (TTS_IS_VIRTUAL(slot))
    {
        tableslot = table_slot_create(localrel, &estate->es_tupleTable);
        tableslot = ExecCopySlot(tableslot, slot);
    }
 
    /*
     * Initialize ecxt_scantuple for potential use in FormIndexDatum when
     * index expressions are present.
     */
    GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
 
    /*
     * The values/nulls arrays passed to BuildIndexValueDescription should be
     * the results of FormIndexDatum, which are the "raw" input to the index
     * AM.
     */
    FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull);
 
    index_value = BuildIndexValueDescription(indexDesc, values, isnull);
 
    index_close(indexDesc, NoLock);
 
    return index_value;
}

References Assert, BuildIndexInfo(), BuildIndexValueDescription(), CheckRelationOidLockedByMe(), EState::es_tupleTable, ExecCopySlot(), fb(), FormIndexDatum(), GetPerTupleExprContext, index_close(), INDEX_MAX_KEYS, index_open(), NoLock, RowExclusiveLock, table_slot_create(), TTS_IS_VIRTUAL, and values.

Referenced by get_tuple_desc().

create_conflict_log_table()

Oid create_conflict_log_table	(	Oid	subid,
		char *	subname,
		Oid	subowner
	)

Definition at line 147 of file conflict.c.

{
    TupleDesc   tupdesc;
    Oid         relid;
    char        relname[NAMEDATALEN];
 
    snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
 
    /* Build the tuple descriptor for the new table. */
    tupdesc = create_conflict_log_table_tupdesc();
 
    /* Create conflict log table. */
    relid = heap_create_with_catalog(relname,
                                     PG_CONFLICT_NAMESPACE,
                                     0, /* tablespace */
                                     InvalidOid,    /* relid */
                                     InvalidOid,    /* reltypeid */
                                     InvalidOid,    /* reloftypeid */
                                     subowner,
                                     HEAP_TABLE_AM_OID,
                                     tupdesc,
                                     NIL,
                                     RELKIND_RELATION,
                                     RELPERSISTENCE_PERMANENT,
                                     false, /* shared_relation */
                                     false, /* mapped_relation */
                                     ONCOMMIT_NOOP,
                                     (Datum) 0, /* reloptions */
                                     false, /* use_user_acl */
                                     false, /* allow_system_table_mods */
                                     true,  /* is_internal */
                                     InvalidOid,    /* relrewrite */
                                     NULL); /* typaddress */
    Assert(OidIsValid(relid));
 
    /* Release tuple descriptor memory. */
    FreeTupleDesc(tupdesc);
 
    /*
     * We must bump the command counter to make the newly-created relation
     * tuple visible for opening.
     */
    CommandCounterIncrement();
 
    /*
     * Create a TOAST table for the conflict log to support out-of-line
     * storage of large json data.
     */
    NewRelationCreateToastTable(relid, (Datum) 0);
 
    ereport(NOTICE,
            (errmsg("created conflict log table \"%s\" for subscription \"%s\"",
                    get_qualified_objname(PG_CONFLICT_NAMESPACE, relname),
                    subname)));
 
    return relid;
}

References Assert, CommandCounterIncrement(), create_conflict_log_table_tupdesc(), ereport, errmsg, fb(), FreeTupleDesc(), get_qualified_objname(), heap_create_with_catalog(), InvalidOid, NAMEDATALEN, NewRelationCreateToastTable(), NIL, NOTICE, OidIsValid, ONCOMMIT_NOOP, relname, snprintf, and subname.

Referenced by alter_sub_conflict_log_dest().

create_conflict_log_table_tupdesc()

static TupleDesc create_conflict_log_table_tupdesc ( void  )

static

Definition at line 118 of file conflict.c.

{
    TupleDesc   tupdesc;
 
    tupdesc = CreateTemplateTupleDesc(NUM_CONFLICT_ATTRS);
 
    for (int i = 0; i < NUM_CONFLICT_ATTRS; i++)
        TupleDescInitEntry(tupdesc, i + 1,
                           ConflictLogSchema[i].attname,
                           ConflictLogSchema[i].atttypid,
                           -1, 0);
 
    TupleDescFinalize(tupdesc);
 
    return tupdesc;
}

References attname, ConflictLogSchema, CreateTemplateTupleDesc(), i, NUM_CONFLICT_ATTRS, TupleDescFinalize(), and TupleDescInitEntry().

Referenced by create_conflict_log_table().

errcode_apply_conflict()

static int errcode_apply_conflict ( ConflictType  type )

static

Definition at line 342 of file conflict.c.

{
    switch (type)
    {
        case CT_INSERT_EXISTS:
        case CT_UPDATE_EXISTS:
        case CT_MULTIPLE_UNIQUE_CONFLICTS:
            return errcode(ERRCODE_UNIQUE_VIOLATION);
        case CT_UPDATE_ORIGIN_DIFFERS:
        case CT_UPDATE_MISSING:
        case CT_DELETE_ORIGIN_DIFFERS:
        case CT_UPDATE_DELETED:
        case CT_DELETE_MISSING:
            return errcode(ERRCODE_T_R_SERIALIZATION_FAILURE);
    }
 
    Assert(false);
    return 0;                   /* silence compiler warning */
}

References Assert, CT_DELETE_MISSING, CT_DELETE_ORIGIN_DIFFERS, CT_INSERT_EXISTS, CT_MULTIPLE_UNIQUE_CONFLICTS, CT_UPDATE_DELETED, CT_UPDATE_EXISTS, CT_UPDATE_MISSING, CT_UPDATE_ORIGIN_DIFFERS, errcode(), ERRCODE_T_R_SERIALIZATION_FAILURE, fb(), and type.

Referenced by ReportApplyConflict().

errdetail_apply_conflict()

static void errdetail_apply_conflict ( EState *  estate, ResultRelInfo *  relinfo, ConflictType  type, TupleTableSlot *  searchslot, TupleTableSlot *  localslot, TupleTableSlot *  remoteslot, Oid  indexoid, TransactionId  localxmin, ReplOriginId  localorigin, TimestampTz  localts, StringInfo  err_msg  )

static

Definition at line 403 of file conflict.c.

{
    StringInfoData err_detail;
    StringInfoData tuple_buf;
    char       *origin_name;
    char       *key_desc = NULL;
    char       *local_desc = NULL;
    char       *remote_desc = NULL;
    char       *search_desc = NULL;
 
    /* Get key, replica identity, remote, and local value data */
    get_tuple_desc(estate, relinfo, type, &key_desc,
                   localslot, &local_desc,
                   remoteslot, &remote_desc,
                   searchslot, &search_desc,
                   indexoid);
 
    initStringInfo(&err_detail);
    initStringInfo(&tuple_buf);
 
    /* Construct a detailed message describing the type of conflict */
    switch (type)
    {
        case CT_INSERT_EXISTS:
        case CT_UPDATE_EXISTS:
        case CT_MULTIPLE_UNIQUE_CONFLICTS:
            Assert(OidIsValid(indexoid) &&
                   CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
 
            if (err_msg->len == 0)
            {
                append_tuple_value_detail(&tuple_buf,
                                          list_make2(remote_desc, search_desc));
 
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Could not apply remote change: %s.\n"),
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Could not apply remote change.\n"));
 
 
                resetStringInfo(&tuple_buf);
            }
 
            append_tuple_value_detail(&tuple_buf,
                                      list_make2(key_desc, local_desc));
 
            if (localts)
            {
                if (localorigin == InvalidReplOriginId)
                {
                    if (tuple_buf.len)
                        appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified locally in transaction %u at %s: %s."),
                                         get_rel_name(indexoid),
                                         localxmin, timestamptz_to_str(localts),
                                         tuple_buf.data);
                    else
                        appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified locally in transaction %u at %s."),
                                         get_rel_name(indexoid),
                                         localxmin, timestamptz_to_str(localts));
                }
                else if (replorigin_by_oid(localorigin, true, &origin_name))
                {
                    if (tuple_buf.len)
                        appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by origin \"%s\" in transaction %u at %s: %s."),
                                         get_rel_name(indexoid), origin_name,
                                         localxmin, timestamptz_to_str(localts),
                                         tuple_buf.data);
                    else
                        appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by origin \"%s\" in transaction %u at %s."),
                                         get_rel_name(indexoid), origin_name,
                                         localxmin, timestamptz_to_str(localts));
                }
 
                /*
                 * The origin that modified this row has been removed. This
                 * can happen if the origin was created by a different apply
                 * worker and its associated subscription and origin were
                 * dropped after updating the row, or if the origin was
                 * manually dropped by the user.
                 */
                else
                {
                    if (tuple_buf.len)
                        appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by a non-existent origin in transaction %u at %s: %s."),
                                         get_rel_name(indexoid),
                                         localxmin, timestamptz_to_str(localts),
                                         tuple_buf.data);
                    else
                        appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by a non-existent origin in transaction %u at %s."),
                                         get_rel_name(indexoid),
                                         localxmin, timestamptz_to_str(localts));
                }
            }
            else
            {
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified in transaction %u: %s."),
                                     get_rel_name(indexoid), localxmin,
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified in transaction %u."),
                                     get_rel_name(indexoid), localxmin);
            }
 
            break;
 
        case CT_UPDATE_ORIGIN_DIFFERS:
            append_tuple_value_detail(&tuple_buf,
                                      list_make3(local_desc, remote_desc,
                                                 search_desc));
 
            if (localorigin == InvalidReplOriginId)
            {
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Updating the row that was modified locally in transaction %u at %s: %s."),
                                     localxmin, timestamptz_to_str(localts),
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Updating the row that was modified locally in transaction %u at %s."),
                                     localxmin, timestamptz_to_str(localts));
            }
            else if (replorigin_by_oid(localorigin, true, &origin_name))
            {
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Updating the row that was modified by a different origin \"%s\" in transaction %u at %s: %s."),
                                     origin_name, localxmin,
                                     timestamptz_to_str(localts),
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Updating the row that was modified by a different origin \"%s\" in transaction %u at %s."),
                                     origin_name, localxmin,
                                     timestamptz_to_str(localts));
            }
 
            /* The origin that modified this row has been removed. */
            else
            {
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Updating the row that was modified by a non-existent origin in transaction %u at %s: %s."),
                                     localxmin, timestamptz_to_str(localts),
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Updating the row that was modified by a non-existent origin in transaction %u at %s."),
                                     localxmin, timestamptz_to_str(localts));
            }
 
            break;
 
        case CT_UPDATE_DELETED:
            append_tuple_value_detail(&tuple_buf,
                                      list_make2(remote_desc, search_desc));
 
            if (tuple_buf.len)
                appendStringInfo(&err_detail, _("Could not find the row to be updated: %s.\n"),
                                 tuple_buf.data);
            else
                appendStringInfo(&err_detail, _("Could not find the row to be updated.\n"));
 
            if (localts)
            {
                if (localorigin == InvalidReplOriginId)
                    appendStringInfo(&err_detail, _("The row to be updated was deleted locally in transaction %u at %s"),
                                     localxmin, timestamptz_to_str(localts));
                else if (replorigin_by_oid(localorigin, true, &origin_name))
                    appendStringInfo(&err_detail, _("The row to be updated was deleted by a different origin \"%s\" in transaction %u at %s"),
                                     origin_name, localxmin, timestamptz_to_str(localts));
 
                /* The origin that modified this row has been removed. */
                else
                    appendStringInfo(&err_detail, _("The row to be updated was deleted by a non-existent origin in transaction %u at %s"),
                                     localxmin, timestamptz_to_str(localts));
            }
            else
                appendStringInfoString(&err_detail, _("The row to be updated was deleted"));
 
            break;
 
        case CT_UPDATE_MISSING:
            append_tuple_value_detail(&tuple_buf,
                                      list_make2(remote_desc, search_desc));
 
            if (tuple_buf.len)
                appendStringInfo(&err_detail, _("Could not find the row to be updated: %s."),
                                 tuple_buf.data);
            else
                appendStringInfo(&err_detail, _("Could not find the row to be updated."));
 
            break;
 
        case CT_DELETE_ORIGIN_DIFFERS:
            append_tuple_value_detail(&tuple_buf,
                                      list_make3(local_desc, remote_desc,
                                                 search_desc));
 
            if (localorigin == InvalidReplOriginId)
            {
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Deleting the row that was modified locally in transaction %u at %s: %s."),
                                     localxmin, timestamptz_to_str(localts),
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Deleting the row that was modified locally in transaction %u at %s."),
                                     localxmin, timestamptz_to_str(localts));
            }
            else if (replorigin_by_oid(localorigin, true, &origin_name))
            {
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Deleting the row that was modified by a different origin \"%s\" in transaction %u at %s: %s."),
                                     origin_name, localxmin,
                                     timestamptz_to_str(localts),
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Deleting the row that was modified by a different origin \"%s\" in transaction %u at %s."),
                                     origin_name, localxmin,
                                     timestamptz_to_str(localts));
            }
 
            /* The origin that modified this row has been removed. */
            else
            {
                if (tuple_buf.len)
                    appendStringInfo(&err_detail, _("Deleting the row that was modified by a non-existent origin in transaction %u at %s: %s."),
                                     localxmin, timestamptz_to_str(localts),
                                     tuple_buf.data);
                else
                    appendStringInfo(&err_detail, _("Deleting the row that was modified by a non-existent origin in transaction %u at %s."),
                                     localxmin, timestamptz_to_str(localts));
            }
 
            break;
 
        case CT_DELETE_MISSING:
            append_tuple_value_detail(&tuple_buf,
                                      list_make1(search_desc));
 
            if (tuple_buf.len)
                appendStringInfo(&err_detail, _("Could not find the row to be deleted: %s."),
                                 tuple_buf.data);
            else
                appendStringInfo(&err_detail, _("Could not find the row to be deleted."));
 
            break;
    }
 
    Assert(err_detail.len > 0);
 
    /*
     * Insert a blank line to visually separate the new detail line from the
     * existing ones.
     */
    if (err_msg->len > 0)
        appendStringInfoChar(err_msg, '\n');
 
    appendStringInfoString(err_msg, err_detail.data);
}

References _, append_tuple_value_detail(), appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), Assert, CheckRelationOidLockedByMe(), CT_DELETE_MISSING, CT_DELETE_ORIGIN_DIFFERS, CT_INSERT_EXISTS, CT_MULTIPLE_UNIQUE_CONFLICTS, CT_UPDATE_DELETED, CT_UPDATE_EXISTS, CT_UPDATE_MISSING, CT_UPDATE_ORIGIN_DIFFERS, StringInfoData::data, fb(), get_rel_name(), get_tuple_desc(), initStringInfo(), InvalidReplOriginId, StringInfoData::len, list_make1, list_make2, list_make3, OidIsValid, replorigin_by_oid(), resetStringInfo(), RowExclusiveLock, timestamptz_to_str(), and type.

Referenced by ReportApplyConflict().

get_tuple_desc()

static void get_tuple_desc ( EState *  estate, ResultRelInfo *  relinfo, ConflictType  type, char **  key_desc, TupleTableSlot *  localslot, char **  local_desc, TupleTableSlot *  remoteslot, char **  remote_desc, TupleTableSlot *  searchslot, char **  search_desc, Oid  indexoid  )

static

Definition at line 673 of file conflict.c.

{
    Relation    localrel = relinfo->ri_RelationDesc;
    Oid         relid = RelationGetRelid(localrel);
    TupleDesc   tupdesc = RelationGetDescr(localrel);
    char       *desc = NULL;
 
    Assert((localslot && local_desc) || (remoteslot && remote_desc) ||
           (searchslot && search_desc));
 
    /*
     * Report the conflicting key values in the case of a unique constraint
     * violation.
     */
    if (type == CT_INSERT_EXISTS || type == CT_UPDATE_EXISTS ||
        type == CT_MULTIPLE_UNIQUE_CONFLICTS)
    {
        Assert(OidIsValid(indexoid) && localslot);
 
        desc = build_index_value_desc(estate, localrel, localslot,
                                      indexoid);
 
        if (desc)
            *key_desc = psprintf(_("key %s"), desc);
    }
 
    if (localslot)
    {
        /*
         * The 'modifiedCols' only applies to the new tuple, hence we pass
         * NULL for the local row.
         */
        desc = ExecBuildSlotValueDescription(relid, localslot, tupdesc,
                                             NULL, 64);
 
        if (desc)
            *local_desc = psprintf(_("local row %s"), desc);
    }
 
    if (remoteslot)
    {
        Bitmapset  *modifiedCols;
 
        /*
         * Although logical replication doesn't maintain the bitmap for the
         * columns being inserted, we still use it to create 'modifiedCols'
         * for consistency with other calls to ExecBuildSlotValueDescription.
         *
         * Note that generated columns are formed locally on the subscriber.
         */
        modifiedCols = bms_union(ExecGetInsertedCols(relinfo, estate),
                                 ExecGetUpdatedCols(relinfo, estate));
        desc = ExecBuildSlotValueDescription(relid, remoteslot,
                                             tupdesc, modifiedCols,
                                             64);
 
        if (desc)
            *remote_desc = psprintf(_("remote row %s"), desc);
    }
 
    if (searchslot)
    {
        /*
         * Note that while index other than replica identity may be used (see
         * IsIndexUsableForReplicaIdentityFull for details) to find the tuple
         * when applying update or delete, such an index scan may not result
         * in a unique tuple and we still compare the complete tuple in such
         * cases, thus such indexes are not used here.
         */
        Oid         replica_index = GetRelationIdentityOrPK(localrel);
 
        Assert(type != CT_INSERT_EXISTS);
 
        /*
         * If the table has a valid replica identity index, build the index
         * key value string. Otherwise, construct the full tuple value for
         * REPLICA IDENTITY FULL cases.
         */
        if (OidIsValid(replica_index))
            desc = build_index_value_desc(estate, localrel, searchslot, replica_index);
        else
            desc = ExecBuildSlotValueDescription(relid, searchslot, tupdesc, NULL, 64);
 
        if (desc)
        {
            if (OidIsValid(replica_index))
                *search_desc = psprintf(_("replica identity %s"), desc);
            else
                *search_desc = psprintf(_("replica identity full %s"), desc);
        }
    }
}

References _, Assert, bms_union(), build_index_value_desc(), CT_INSERT_EXISTS, CT_MULTIPLE_UNIQUE_CONFLICTS, CT_UPDATE_EXISTS, ExecBuildSlotValueDescription(), ExecGetInsertedCols(), ExecGetUpdatedCols(), fb(), GetRelationIdentityOrPK(), OidIsValid, psprintf(), RelationGetDescr, RelationGetRelid, and type.

Referenced by errdetail_apply_conflict().

GetConflictLogDest()

ConflictLogDest GetConflictLogDest

(

const char *

dest

)

Definition at line 210 of file conflict.c.

{
    /* NULL defaults to LOG. */
    if (dest == NULL || pg_strcasecmp(dest, "log") == 0)
        return CONFLICT_LOG_DEST_LOG;
 
    if (pg_strcasecmp(dest, "table") == 0)
        return CONFLICT_LOG_DEST_TABLE;
 
    if (pg_strcasecmp(dest, "all") == 0)
        return CONFLICT_LOG_DEST_ALL;
 
    /* Unrecognized string. */
    ereport(ERROR,
            (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
             errmsg("unrecognized conflict_log_destination value: \"%s\"", dest),
             errhint("Valid values are \"log\", \"table\", and \"all\".")));
}

References CONFLICT_LOG_DEST_ALL, CONFLICT_LOG_DEST_LOG, CONFLICT_LOG_DEST_TABLE, ereport, errcode(), errhint(), errmsg, ERROR, fb(), and pg_strcasecmp().

Referenced by AlterSubscription(), and parse_subscription_options().

GetTupleTransactionInfo()

bool GetTupleTransactionInfo	(	TupleTableSlot *	localslot,
		TransactionId *	xmin,
		ReplOriginId *	localorigin,
		TimestampTz *	localts
	)

Definition at line 236 of file conflict.c.

{
    Datum       xminDatum;
    bool        isnull;
 
    xminDatum = slot_getsysattr(localslot, MinTransactionIdAttributeNumber,
                                &isnull);
    *xmin = DatumGetTransactionId(xminDatum);
    Assert(!isnull);
 
    /*
     * The commit timestamp data is not available if track_commit_timestamp is
     * disabled.
     */
    if (!track_commit_timestamp)
    {
        *localorigin = InvalidReplOriginId;
        *localts = 0;
        return false;
    }
 
    return TransactionIdGetCommitTsData(*xmin, localts, localorigin);
}

References Assert, DatumGetTransactionId(), fb(), InvalidReplOriginId, MinTransactionIdAttributeNumber, slot_getsysattr(), track_commit_timestamp, and TransactionIdGetCommitTsData().

Referenced by apply_handle_delete_internal(), apply_handle_tuple_routing(), apply_handle_update_internal(), and CheckAndReportConflict().

InitConflictIndexes()

void InitConflictIndexes

(

ResultRelInfo *

relInfo

)

Definition at line 312 of file conflict.c.

{
    List       *uniqueIndexes = NIL;
 
    for (int i = 0; i < relInfo->ri_NumIndices; i++)
    {
        Relation    indexRelation = relInfo->ri_IndexRelationDescs[i];
 
        if (indexRelation == NULL)
            continue;
 
        /* Detect conflict only for unique indexes */
        if (!relInfo->ri_IndexRelationInfo[i]->ii_Unique)
            continue;
 
        /* Don't support conflict detection for deferrable index */
        if (!indexRelation->rd_index->indimmediate)
            continue;
 
        uniqueIndexes = lappend_oid(uniqueIndexes,
                                    RelationGetRelid(indexRelation));
    }
 
    relInfo->ri_onConflictArbiterIndexes = uniqueIndexes;
}

References fb(), i, lappend_oid(), NIL, RelationData::rd_index, and RelationGetRelid.

Referenced by apply_handle_insert_internal(), apply_handle_tuple_routing(), and apply_handle_update_internal().

ReportApplyConflict()

void ReportApplyConflict	(	EState *	estate,
		ResultRelInfo *	relinfo,
		int	elevel,
		ConflictType	type,
		TupleTableSlot *	searchslot,
		TupleTableSlot *	remoteslot,
		List *	conflicttuples
	)

Definition at line 277 of file conflict.c.

{
    Relation    localrel = relinfo->ri_RelationDesc;
    StringInfoData err_detail;
 
    initStringInfo(&err_detail);
 
    /* Form errdetail message by combining conflicting tuples information. */
    foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
        errdetail_apply_conflict(estate, relinfo, type, searchslot,
                                 conflicttuple->slot, remoteslot,
                                 conflicttuple->indexoid,
                                 conflicttuple->xmin,
                                 conflicttuple->origin,
                                 conflicttuple->ts,
                                 &err_detail);
 
    pgstat_report_subscription_conflict(MySubscription->oid, type);
 
    ereport(elevel,
            errcode_apply_conflict(type),
            errmsg("conflict detected on relation \"%s.%s\": conflict=%s",
                   get_namespace_name(RelationGetNamespace(localrel)),
                   RelationGetRelationName(localrel),
                   ConflictTypeNames[type]),
            errdetail_internal("%s", err_detail.data));
}

References ConflictTypeNames, ereport, errcode_apply_conflict(), errdetail_apply_conflict(), errdetail_internal(), errmsg, fb(), foreach_ptr, get_namespace_name(), initStringInfo(), MySubscription, Subscription::oid, pgstat_report_subscription_conflict(), RelationGetNamespace, RelationGetRelationName, and type.

Referenced by apply_handle_delete_internal(), apply_handle_tuple_routing(), apply_handle_update_internal(), and CheckAndReportConflict().

StaticAssertDecl()

StaticAssertDecl	(	lengthof(ConflictLogDestNames)	= =CONFLICT_LOG_DEST_ALL+1,
		"ConflictLogDestNames length mismatch"
	)

Variable Documentation

ConflictLogDestNames

const char* const ConflictLogDestNames[]

Initial value:

= {
    [CONFLICT_LOG_DEST_LOG] = "log",
    [CONFLICT_LOG_DEST_TABLE] = "table",
    [CONFLICT_LOG_DEST_ALL] = "all"
}

Definition at line 34 of file conflict.c.

                                           {
    [CONFLICT_LOG_DEST_LOG] = "log",
    [CONFLICT_LOG_DEST_TABLE] = "table",
    [CONFLICT_LOG_DEST_ALL] = "all"
};

Referenced by AlterSubscription().

ConflictLogSchema

const ConflictLogColumnDef ConflictLogSchema[]

static

Initial value:

= {
    {.attname = "relid", .atttypid = OIDOID},
    {.attname = "schemaname", .atttypid = TEXTOID},
    {.attname = "relname", .atttypid = TEXTOID},
    {.attname = "conflict_type", .atttypid = TEXTOID},
    {.attname = "remote_xid", .atttypid = XIDOID},
    {.attname = "remote_commit_lsn", .atttypid = LSNOID},
    {.attname = "remote_commit_ts", .atttypid = TIMESTAMPTZOID},
    {.attname = "remote_origin", .atttypid = TEXTOID},
    {.attname = "replica_identity_full", .atttypid = BOOLOID},
    {.attname = "replica_identity", .atttypid = JSONOID},
    {.attname = "remote_tuple", .atttypid = JSONOID},
    {.attname = "local_conflicts", .atttypid = JSONARRAYOID}
}

Definition at line 67 of file conflict.c.

                                                        {
    {.attname = "relid", .atttypid = OIDOID},
    {.attname = "schemaname", .atttypid = TEXTOID},
    {.attname = "relname", .atttypid = TEXTOID},
    {.attname = "conflict_type", .atttypid = TEXTOID},
    {.attname = "remote_xid", .atttypid = XIDOID},
    {.attname = "remote_commit_lsn", .atttypid = LSNOID},
    {.attname = "remote_commit_ts", .atttypid = TIMESTAMPTZOID},
    {.attname = "remote_origin", .atttypid = TEXTOID},
    {.attname = "replica_identity_full", .atttypid = BOOLOID},
    {.attname = "replica_identity", .atttypid = JSONOID},
    {.attname = "remote_tuple", .atttypid = JSONOID},
    {.attname = "local_conflicts", .atttypid = JSONARRAYOID}
};

Referenced by create_conflict_log_table_tupdesc().

ConflictTypeNames

const char* const ConflictTypeNames[]

static

Initial value:

= {
    [CT_INSERT_EXISTS] = "insert_exists",
    [CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
    [CT_UPDATE_EXISTS] = "update_exists",
    [CT_UPDATE_MISSING] = "update_missing",
    [CT_DELETE_ORIGIN_DIFFERS] = "delete_origin_differs",
    [CT_UPDATE_DELETED] = "update_deleted",
    [CT_DELETE_MISSING] = "delete_missing",
    [CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts"
}

Definition at line 84 of file conflict.c.

                                               {
    [CT_INSERT_EXISTS] = "insert_exists",
    [CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
    [CT_UPDATE_EXISTS] = "update_exists",
    [CT_UPDATE_MISSING] = "update_missing",
    [CT_DELETE_ORIGIN_DIFFERS] = "delete_origin_differs",
    [CT_UPDATE_DELETED] = "update_deleted",
    [CT_DELETE_MISSING] = "delete_missing",
    [CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts"
};

Referenced by ReportApplyConflict().