#include "postgres.h"
#include "access/genam.h"
#include "access/relscan.h"
#include "access/tableam.h"
#include "access/transam.h"
#include "access/xact.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_relation.h"
#include "parser/parsetree.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/typcache.h"

Include dependency graph for execReplication.c:

Functions
static bool	build_replindex_scan_key (ScanKey skey, Relation rel, Relation idxrel, TupleTableSlot *searchslot)

bool	RelationFindReplTupleByIndex (Relation rel, Oid idxoid, LockTupleMode lockmode, TupleTableSlot searchslot, TupleTableSlot outslot)

static bool	tuples_equal (TupleTableSlot slot1, TupleTableSlot slot2, TypeCacheEntry **eq)

bool	RelationFindReplTupleSeq (Relation rel, LockTupleMode lockmode, TupleTableSlot searchslot, TupleTableSlot outslot)

void	ExecSimpleRelationInsert (ResultRelInfo resultRelInfo, EState estate, TupleTableSlot *slot)

void	ExecSimpleRelationUpdate (ResultRelInfo resultRelInfo, EState estate, EPQState epqstate, TupleTableSlot searchslot, TupleTableSlot *slot)

void	ExecSimpleRelationDelete (ResultRelInfo resultRelInfo, EState estate, EPQState epqstate, TupleTableSlot searchslot)

void	CheckCmdReplicaIdentity (Relation rel, CmdType cmd)

void	CheckSubscriptionRelkind (char relkind, const char nspname, const char relname)

Function Documentation

◆ build_replindex_scan_key()

static bool build_replindex_scan_key	(	ScanKey	skey,
		Relation	rel,
		Relation	idxrel,
		TupleTableSlot *	searchslot
	)

static

Definition at line 50 of file execReplication.c.

 {
     int         attoff;
     bool        isnull;
     Datum       indclassDatum;
     oidvector  *opclass;
     int2vector *indkey = &idxrel->rd_index->indkey;
     bool        hasnulls = false;
  
     Assert(RelationGetReplicaIndex(rel) == RelationGetRelid(idxrel) ||
            RelationGetPrimaryKeyIndex(rel) == RelationGetRelid(idxrel));
  
     indclassDatum = SysCacheGetAttr(INDEXRELID, idxrel->rd_indextuple,
                                     Anum_pg_index_indclass, &isnull);
     Assert(!isnull);
     opclass = (oidvector *) DatumGetPointer(indclassDatum);
  
     /* Build scankey for every attribute in the index. */
     for (attoff = 0; attoff < IndexRelationGetNumberOfKeyAttributes(idxrel); attoff++)
     {
         Oid         operator;
         Oid         opfamily;
         RegProcedure regop;
         int         pkattno = attoff + 1;
         int         mainattno = indkey->values[attoff];
         Oid         optype = get_opclass_input_type(opclass->values[attoff]);
  
         /*
          * Load the operator info.  We need this to get the equality operator
          * function for the scan key.
          */
         opfamily = get_opclass_family(opclass->values[attoff]);
  
         operator = get_opfamily_member(opfamily, optype,
                                        optype,
                                        BTEqualStrategyNumber);
         if (!OidIsValid(operator))
             elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
                  BTEqualStrategyNumber, optype, optype, opfamily);
  
         regop = get_opcode(operator);
  
         /* Initialize the scankey. */
         ScanKeyInit(&skey[attoff],
                     pkattno,
                     BTEqualStrategyNumber,
                     regop,
                     searchslot->tts_values[mainattno - 1]);
  
         skey[attoff].sk_collation = idxrel->rd_indcollation[attoff];
  
         /* Check for null value. */
         if (searchslot->tts_isnull[mainattno - 1])
         {
             hasnulls = true;
             skey[attoff].sk_flags |= SK_ISNULL;
         }
     }
  
     return hasnulls;
 }

References Assert(), BTEqualStrategyNumber, DatumGetPointer, elog, ERROR, get_opclass_family(), get_opclass_input_type(), get_opcode(), get_opfamily_member(), IndexRelationGetNumberOfKeyAttributes, INDEXRELID, OidIsValid, RelationData::rd_indcollation, RelationData::rd_index, RelationData::rd_indextuple, RelationGetPrimaryKeyIndex(), RelationGetRelid, RelationGetReplicaIndex(), ScanKeyInit(), ScanKeyData::sk_collation, ScanKeyData::sk_flags, SK_ISNULL, SysCacheGetAttr(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_values, int2vector::values, and oidvector::values.

Referenced by RelationFindReplTupleByIndex().

◆ CheckCmdReplicaIdentity()

void CheckCmdReplicaIdentity	(	Relation	rel,
		CmdType	cmd
	)

Definition at line 568 of file execReplication.c.

 {
     PublicationActions *pubactions;
  
     /* We only need to do checks for UPDATE and DELETE. */
     if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
         return;
  
     /* If relation has replica identity we are always good. */
     if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
         OidIsValid(RelationGetReplicaIndex(rel)))
         return;
  
     /*
      * This is either UPDATE OR DELETE and there is no replica identity.
      *
      * Check if the table publishes UPDATES or DELETES.
      */
     pubactions = GetRelationPublicationActions(rel);
     if (cmd == CMD_UPDATE && pubactions->pubupdate)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("cannot update table \"%s\" because it does not have a replica identity and publishes updates",
                         RelationGetRelationName(rel)),
                  errhint("To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.")));
     else if (cmd == CMD_DELETE && pubactions->pubdelete)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("cannot delete from table \"%s\" because it does not have a replica identity and publishes deletes",
                         RelationGetRelationName(rel)),
                  errhint("To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.")));
 }

References CMD_DELETE, CMD_UPDATE, ereport, errcode(), errhint(), errmsg(), ERROR, GetRelationPublicationActions(), OidIsValid, PublicationActions::pubdelete, PublicationActions::pubupdate, RelationData::rd_rel, RelationGetRelationName, and RelationGetReplicaIndex().

Referenced by CheckValidResultRel(), ExecSimpleRelationDelete(), ExecSimpleRelationInsert(), and ExecSimpleRelationUpdate().

◆ CheckSubscriptionRelkind()

void CheckSubscriptionRelkind	(	char	relkind,
		const char *	nspname,
		const char *	relname
	)

Definition at line 608 of file execReplication.c.

 {
     if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE)
         ereport(ERROR,
                 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                  errmsg("cannot use relation \"%s.%s\" as logical replication target",
                         nspname, relname),
                  errdetail_relkind_not_supported(relkind)));
 }

References ereport, errcode(), errdetail_relkind_not_supported(), errmsg(), ERROR, and relname.

Referenced by AlterSubscription_refresh(), CreateSubscription(), and logicalrep_rel_open().

◆ ExecSimpleRelationDelete()

void ExecSimpleRelationDelete	(	ResultRelInfo *	resultRelInfo,
		EState *	estate,
		EPQState *	epqstate,
		TupleTableSlot *	searchslot
	)

Definition at line 534 of file execReplication.c.

 {
     bool        skip_tuple = false;
     Relation    rel = resultRelInfo->ri_RelationDesc;
     ItemPointer tid = &searchslot->tts_tid;
  
     CheckCmdReplicaIdentity(rel, CMD_DELETE);
  
     /* BEFORE ROW DELETE Triggers */
     if (resultRelInfo->ri_TrigDesc &&
         resultRelInfo->ri_TrigDesc->trig_delete_before_row)
     {
         skip_tuple = !ExecBRDeleteTriggers(estate, epqstate, resultRelInfo,
                                            tid, NULL, NULL);
  
     }
  
     if (!skip_tuple)
     {
         /* OK, delete the tuple */
         simple_table_tuple_delete(rel, tid, estate->es_snapshot);
  
         /* AFTER ROW DELETE Triggers */
         ExecARDeleteTriggers(estate, resultRelInfo,
                              tid, NULL, NULL);
     }
 }

References CheckCmdReplicaIdentity(), CMD_DELETE, EState::es_snapshot, ExecARDeleteTriggers(), ExecBRDeleteTriggers(), ResultRelInfo::ri_RelationDesc, ResultRelInfo::ri_TrigDesc, simple_table_tuple_delete(), TriggerDesc::trig_delete_before_row, and TupleTableSlot::tts_tid.

Referenced by apply_handle_delete_internal().

◆ ExecSimpleRelationInsert()

void ExecSimpleRelationInsert	(	ResultRelInfo *	resultRelInfo,
		EState *	estate,
		TupleTableSlot *	slot
	)

Definition at line 407 of file execReplication.c.

 {
     bool        skip_tuple = false;
     Relation    rel = resultRelInfo->ri_RelationDesc;
  
     /* For now we support only tables. */
     Assert(rel->rd_rel->relkind == RELKIND_RELATION);
  
     CheckCmdReplicaIdentity(rel, CMD_INSERT);
  
     /* BEFORE ROW INSERT Triggers */
     if (resultRelInfo->ri_TrigDesc &&
         resultRelInfo->ri_TrigDesc->trig_insert_before_row)
     {
         if (!ExecBRInsertTriggers(estate, resultRelInfo, slot))
             skip_tuple = true;  /* "do nothing" */
     }
  
     if (!skip_tuple)
     {
         List       *recheckIndexes = NIL;
  
         /* Compute stored generated columns */
         if (rel->rd_att->constr &&
             rel->rd_att->constr->has_generated_stored)
             ExecComputeStoredGenerated(resultRelInfo, estate, slot,
                                        CMD_INSERT);
  
         /* Check the constraints of the tuple */
         if (rel->rd_att->constr)
             ExecConstraints(resultRelInfo, slot, estate);
         if (rel->rd_rel->relispartition)
             ExecPartitionCheck(resultRelInfo, slot, estate, true);
  
         /* OK, store the tuple and create index entries for it */
         simple_table_tuple_insert(resultRelInfo->ri_RelationDesc, slot);
  
         if (resultRelInfo->ri_NumIndices > 0)
             recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
                                                    slot, estate, false, false,
                                                    NULL, NIL);
  
         /* AFTER ROW INSERT Triggers */
         ExecARInsertTriggers(estate, resultRelInfo, slot,
                              recheckIndexes, NULL);
  
         /*
          * XXX we should in theory pass a TransitionCaptureState object to the
          * above to capture transition tuples, but after statement triggers
          * don't actually get fired by replication yet anyway
          */
  
         list_free(recheckIndexes);
     }
 }

References Assert(), CheckCmdReplicaIdentity(), CMD_INSERT, TupleDescData::constr, ExecARInsertTriggers(), ExecBRInsertTriggers(), ExecComputeStoredGenerated(), ExecConstraints(), ExecInsertIndexTuples(), ExecPartitionCheck(), TupleConstr::has_generated_stored, list_free(), NIL, RelationData::rd_att, RelationData::rd_rel, ResultRelInfo::ri_NumIndices, ResultRelInfo::ri_RelationDesc, ResultRelInfo::ri_TrigDesc, simple_table_tuple_insert(), and TriggerDesc::trig_insert_before_row.

Referenced by apply_handle_insert_internal().

◆ ExecSimpleRelationUpdate()

void ExecSimpleRelationUpdate	(	ResultRelInfo *	resultRelInfo,
		EState *	estate,
		EPQState *	epqstate,
		TupleTableSlot *	searchslot,
		TupleTableSlot *	slot
	)

Definition at line 471 of file execReplication.c.

 {
     bool        skip_tuple = false;
     Relation    rel = resultRelInfo->ri_RelationDesc;
     ItemPointer tid = &(searchslot->tts_tid);
  
     /* For now we support only tables. */
     Assert(rel->rd_rel->relkind == RELKIND_RELATION);
  
     CheckCmdReplicaIdentity(rel, CMD_UPDATE);
  
     /* BEFORE ROW UPDATE Triggers */
     if (resultRelInfo->ri_TrigDesc &&
         resultRelInfo->ri_TrigDesc->trig_update_before_row)
     {
         if (!ExecBRUpdateTriggers(estate, epqstate, resultRelInfo,
                                   tid, NULL, slot))
             skip_tuple = true;  /* "do nothing" */
     }
  
     if (!skip_tuple)
     {
         List       *recheckIndexes = NIL;
         bool        update_indexes;
  
         /* Compute stored generated columns */
         if (rel->rd_att->constr &&
             rel->rd_att->constr->has_generated_stored)
             ExecComputeStoredGenerated(resultRelInfo, estate, slot,
                                        CMD_UPDATE);
  
         /* Check the constraints of the tuple */
         if (rel->rd_att->constr)
             ExecConstraints(resultRelInfo, slot, estate);
         if (rel->rd_rel->relispartition)
             ExecPartitionCheck(resultRelInfo, slot, estate, true);
  
         simple_table_tuple_update(rel, tid, slot, estate->es_snapshot,
                                   &update_indexes);
  
         if (resultRelInfo->ri_NumIndices > 0 && update_indexes)
             recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
                                                    slot, estate, true, false,
                                                    NULL, NIL);
  
         /* AFTER ROW UPDATE Triggers */
         ExecARUpdateTriggers(estate, resultRelInfo,
                              tid, NULL, slot,
                              recheckIndexes, NULL);
  
         list_free(recheckIndexes);
     }
 }

References Assert(), CheckCmdReplicaIdentity(), CMD_UPDATE, TupleDescData::constr, EState::es_snapshot, ExecARUpdateTriggers(), ExecBRUpdateTriggers(), ExecComputeStoredGenerated(), ExecConstraints(), ExecInsertIndexTuples(), ExecPartitionCheck(), TupleConstr::has_generated_stored, list_free(), NIL, RelationData::rd_att, RelationData::rd_rel, ResultRelInfo::ri_NumIndices, ResultRelInfo::ri_RelationDesc, ResultRelInfo::ri_TrigDesc, simple_table_tuple_update(), TriggerDesc::trig_update_before_row, and TupleTableSlot::tts_tid.

Referenced by apply_handle_tuple_routing(), and apply_handle_update_internal().

◆ RelationFindReplTupleByIndex()

bool RelationFindReplTupleByIndex	(	Relation	rel,
		Oid	idxoid,
		LockTupleMode	lockmode,
		TupleTableSlot *	searchslot,
		TupleTableSlot *	outslot
	)

Definition at line 120 of file execReplication.c.

 {
     ScanKeyData skey[INDEX_MAX_KEYS];
     IndexScanDesc scan;
     SnapshotData snap;
     TransactionId xwait;
     Relation    idxrel;
     bool        found;
  
     /* Open the index. */
     idxrel = index_open(idxoid, RowExclusiveLock);
  
     /* Start an index scan. */
     InitDirtySnapshot(snap);
     scan = index_beginscan(rel, idxrel, &snap,
                            IndexRelationGetNumberOfKeyAttributes(idxrel),
                            0);
  
     /* Build scan key. */
     build_replindex_scan_key(skey, rel, idxrel, searchslot);
  
 retry:
     found = false;
  
     index_rescan(scan, skey, IndexRelationGetNumberOfKeyAttributes(idxrel), NULL, 0);
  
     /* Try to find the tuple */
     if (index_getnext_slot(scan, ForwardScanDirection, outslot))
     {
         found = true;
         ExecMaterializeSlot(outslot);
  
         xwait = TransactionIdIsValid(snap.xmin) ?
             snap.xmin : snap.xmax;
  
         /*
          * If the tuple is locked, wait for locking transaction to finish and
          * retry.
          */
         if (TransactionIdIsValid(xwait))
         {
             XactLockTableWait(xwait, NULL, NULL, XLTW_None);
             goto retry;
         }
     }
  
     /* Found tuple, try to lock it in the lockmode. */
     if (found)
     {
         TM_FailureData tmfd;
         TM_Result   res;
  
         PushActiveSnapshot(GetLatestSnapshot());
  
         res = table_tuple_lock(rel, &(outslot->tts_tid), GetLatestSnapshot(),
                                outslot,
                                GetCurrentCommandId(false),
                                lockmode,
                                LockWaitBlock,
                                0 /* don't follow updates */ ,
                                &tmfd);
  
         PopActiveSnapshot();
  
         switch (res)
         {
             case TM_Ok:
                 break;
             case TM_Updated:
                 /* XXX: Improve handling here */
                 if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
                     ereport(LOG,
                             (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                              errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
                 else
                     ereport(LOG,
                             (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                              errmsg("concurrent update, retrying")));
                 goto retry;
             case TM_Deleted:
                 /* XXX: Improve handling here */
                 ereport(LOG,
                         (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                          errmsg("concurrent delete, retrying")));
                 goto retry;
             case TM_Invisible:
                 elog(ERROR, "attempted to lock invisible tuple");
                 break;
             default:
                 elog(ERROR, "unexpected table_tuple_lock status: %u", res);
                 break;
         }
     }
  
     index_endscan(scan);
  
     /* Don't release lock until commit. */
     index_close(idxrel, NoLock);
  
     return found;
 }

References build_replindex_scan_key(), TM_FailureData::ctid, elog, ereport, errcode(), errmsg(), ERROR, ExecMaterializeSlot(), ForwardScanDirection, GetCurrentCommandId(), GetLatestSnapshot(), index_beginscan(), index_close(), index_endscan(), index_getnext_slot(), INDEX_MAX_KEYS, index_open(), index_rescan(), IndexRelationGetNumberOfKeyAttributes, InitDirtySnapshot, ItemPointerIndicatesMovedPartitions, LockWaitBlock, LOG, NoLock, PopActiveSnapshot(), PushActiveSnapshot(), res, RowExclusiveLock, table_tuple_lock(), TM_Deleted, TM_Invisible, TM_Ok, TM_Updated, TransactionIdIsValid, TupleTableSlot::tts_tid, XactLockTableWait(), XLTW_None, SnapshotData::xmax, and SnapshotData::xmin.

Referenced by FindReplTupleInLocalRel().

◆ RelationFindReplTupleSeq()

bool RelationFindReplTupleSeq	(	Relation	rel,
		LockTupleMode	lockmode,
		TupleTableSlot *	searchslot,
		TupleTableSlot *	outslot
	)

Definition at line 295 of file execReplication.c.

 {
     TupleTableSlot *scanslot;
     TableScanDesc scan;
     SnapshotData snap;
     TypeCacheEntry **eq;
     TransactionId xwait;
     bool        found;
     TupleDesc   desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);
  
     Assert(equalTupleDescs(desc, outslot->tts_tupleDescriptor));
  
     eq = palloc0(sizeof(*eq) * outslot->tts_tupleDescriptor->natts);
  
     /* Start a heap scan. */
     InitDirtySnapshot(snap);
     scan = table_beginscan(rel, &snap, 0, NULL);
     scanslot = table_slot_create(rel, NULL);
  
 retry:
     found = false;
  
     table_rescan(scan, NULL);
  
     /* Try to find the tuple */
     while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot))
     {
         if (!tuples_equal(scanslot, searchslot, eq))
             continue;
  
         found = true;
         ExecCopySlot(outslot, scanslot);
  
         xwait = TransactionIdIsValid(snap.xmin) ?
             snap.xmin : snap.xmax;
  
         /*
          * If the tuple is locked, wait for locking transaction to finish and
          * retry.
          */
         if (TransactionIdIsValid(xwait))
         {
             XactLockTableWait(xwait, NULL, NULL, XLTW_None);
             goto retry;
         }
  
         /* Found our tuple and it's not locked */
         break;
     }
  
     /* Found tuple, try to lock it in the lockmode. */
     if (found)
     {
         TM_FailureData tmfd;
         TM_Result   res;
  
         PushActiveSnapshot(GetLatestSnapshot());
  
         res = table_tuple_lock(rel, &(outslot->tts_tid), GetLatestSnapshot(),
                                outslot,
                                GetCurrentCommandId(false),
                                lockmode,
                                LockWaitBlock,
                                0 /* don't follow updates */ ,
                                &tmfd);
  
         PopActiveSnapshot();
  
         switch (res)
         {
             case TM_Ok:
                 break;
             case TM_Updated:
                 /* XXX: Improve handling here */
                 if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
                     ereport(LOG,
                             (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                              errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
                 else
                     ereport(LOG,
                             (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                              errmsg("concurrent update, retrying")));
                 goto retry;
             case TM_Deleted:
                 /* XXX: Improve handling here */
                 ereport(LOG,
                         (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
                          errmsg("concurrent delete, retrying")));
                 goto retry;
             case TM_Invisible:
                 elog(ERROR, "attempted to lock invisible tuple");
                 break;
             default:
                 elog(ERROR, "unexpected table_tuple_lock status: %u", res);
                 break;
         }
     }
  
     table_endscan(scan);
     ExecDropSingleTupleTableSlot(scanslot);
  
     return found;
 }

References Assert(), TM_FailureData::ctid, elog, equalTupleDescs(), ereport, errcode(), errmsg(), ERROR, ExecCopySlot(), ExecDropSingleTupleTableSlot(), ForwardScanDirection, GetCurrentCommandId(), GetLatestSnapshot(), InitDirtySnapshot, ItemPointerIndicatesMovedPartitions, LockWaitBlock, LOG, TupleDescData::natts, palloc0(), PG_USED_FOR_ASSERTS_ONLY, PopActiveSnapshot(), PushActiveSnapshot(), RelationGetDescr, res, table_beginscan(), table_endscan(), table_rescan(), table_scan_getnextslot(), table_slot_create(), table_tuple_lock(), TM_Deleted, TM_Invisible, TM_Ok, TM_Updated, TransactionIdIsValid, TupleTableSlot::tts_tid, TupleTableSlot::tts_tupleDescriptor, tuples_equal(), XactLockTableWait(), XLTW_None, SnapshotData::xmax, and SnapshotData::xmin.