PostgreSQL Source Code  git master
execReplication.c File Reference
#include "postgres.h"
#include "access/genam.h"
#include "access/relscan.h"
#include "access/tableam.h"
#include "access/transam.h"
#include "access/xact.h"
#include "catalog/pg_am_d.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
#include "replication/logicalrelation.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/typcache.h"
Include dependency graph for execReplication.c:

Go to the source code of this file.

Functions

static bool tuples_equal (TupleTableSlot *slot1, TupleTableSlot *slot2, TypeCacheEntry **eq)
 
StrategyNumber get_equal_strategy_number_for_am (Oid am)
 
static StrategyNumber get_equal_strategy_number (Oid opclass)
 
static int build_replindex_scan_key (ScanKey skey, Relation rel, Relation idxrel, TupleTableSlot *searchslot)
 
bool RelationFindReplTupleByIndex (Relation rel, Oid idxoid, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot)
 
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 int build_replindex_scan_key ( ScanKey  skey,
Relation  rel,
Relation  idxrel,
TupleTableSlot searchslot 
)
static

Definition at line 96 of file execReplication.c.

98 {
99  int index_attoff;
100  int skey_attoff = 0;
101  Datum indclassDatum;
102  oidvector *opclass;
103  int2vector *indkey = &idxrel->rd_index->indkey;
104 
105  indclassDatum = SysCacheGetAttrNotNull(INDEXRELID, idxrel->rd_indextuple,
106  Anum_pg_index_indclass);
107  opclass = (oidvector *) DatumGetPointer(indclassDatum);
108 
109  /* Build scankey for every non-expression attribute in the index. */
110  for (index_attoff = 0; index_attoff < IndexRelationGetNumberOfKeyAttributes(idxrel);
111  index_attoff++)
112  {
113  Oid operator;
114  Oid optype;
115  Oid opfamily;
116  RegProcedure regop;
117  int table_attno = indkey->values[index_attoff];
118  StrategyNumber eq_strategy;
119 
120  if (!AttributeNumberIsValid(table_attno))
121  {
122  /*
123  * XXX: Currently, we don't support expressions in the scan key,
124  * see code below.
125  */
126  continue;
127  }
128 
129  /*
130  * Load the operator info. We need this to get the equality operator
131  * function for the scan key.
132  */
133  optype = get_opclass_input_type(opclass->values[index_attoff]);
134  opfamily = get_opclass_family(opclass->values[index_attoff]);
135  eq_strategy = get_equal_strategy_number(opclass->values[index_attoff]);
136 
137  operator = get_opfamily_member(opfamily, optype,
138  optype,
139  eq_strategy);
140 
141  if (!OidIsValid(operator))
142  elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
143  eq_strategy, optype, optype, opfamily);
144 
145  regop = get_opcode(operator);
146 
147  /* Initialize the scankey. */
148  ScanKeyInit(&skey[skey_attoff],
149  index_attoff + 1,
150  eq_strategy,
151  regop,
152  searchslot->tts_values[table_attno - 1]);
153 
154  skey[skey_attoff].sk_collation = idxrel->rd_indcollation[index_attoff];
155 
156  /* Check for null value. */
157  if (searchslot->tts_isnull[table_attno - 1])
158  skey[skey_attoff].sk_flags |= (SK_ISNULL | SK_SEARCHNULL);
159 
160  skey_attoff++;
161  }
162 
163  /* There must always be at least one attribute for the index scan. */
164  Assert(skey_attoff > 0);
165 
166  return skey_attoff;
167 }
#define AttributeNumberIsValid(attributeNumber)
Definition: attnum.h:34
#define Assert(condition)
Definition: c.h:858
regproc RegProcedure
Definition: c.h:650
#define OidIsValid(objectId)
Definition: c.h:775
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:224
static StrategyNumber get_equal_strategy_number(Oid opclass)
Oid get_opclass_input_type(Oid opclass)
Definition: lsyscache.c:1212
Oid get_opclass_family(Oid opclass)
Definition: lsyscache.c:1190
RegProcedure get_opcode(Oid opno)
Definition: lsyscache.c:1285
Oid get_opfamily_member(Oid opfamily, Oid lefttype, Oid righttype, int16 strategy)
Definition: lsyscache.c:166
uintptr_t Datum
Definition: postgres.h:64
static Pointer DatumGetPointer(Datum X)
Definition: postgres.h:312
unsigned int Oid
Definition: postgres_ext.h:31
#define IndexRelationGetNumberOfKeyAttributes(relation)
Definition: rel.h:524
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define SK_SEARCHNULL
Definition: skey.h:121
#define SK_ISNULL
Definition: skey.h:115
uint16 StrategyNumber
Definition: stratnum.h:22
struct HeapTupleData * rd_indextuple
Definition: rel.h:194
Form_pg_index rd_index
Definition: rel.h:192
Oid * rd_indcollation
Definition: rel.h:217
int sk_flags
Definition: skey.h:66
Oid sk_collation
Definition: skey.h:70
bool * tts_isnull
Definition: tuptable.h:127
Datum * tts_values
Definition: tuptable.h:125
Definition: c.h:715
int16 values[FLEXIBLE_ARRAY_MEMBER]
Definition: c.h:722
Definition: c.h:726
Oid values[FLEXIBLE_ARRAY_MEMBER]
Definition: c.h:733
Datum SysCacheGetAttrNotNull(int cacheId, HeapTuple tup, AttrNumber attributeNumber)
Definition: syscache.c:510

References Assert, AttributeNumberIsValid, DatumGetPointer(), elog, ERROR, get_equal_strategy_number(), get_opclass_family(), get_opclass_input_type(), get_opcode(), get_opfamily_member(), IndexRelationGetNumberOfKeyAttributes, OidIsValid, RelationData::rd_indcollation, RelationData::rd_index, RelationData::rd_indextuple, ScanKeyInit(), ScanKeyData::sk_collation, ScanKeyData::sk_flags, SK_ISNULL, SK_SEARCHNULL, SysCacheGetAttrNotNull(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_values, int2vector::values, and oidvector::values.

Referenced by RelationFindReplTupleByIndex().

◆ CheckCmdReplicaIdentity()

void CheckCmdReplicaIdentity ( Relation  rel,
CmdType  cmd 
)

Definition at line 652 of file execReplication.c.

653 {
654  PublicationDesc pubdesc;
655 
656  /*
657  * Skip checking the replica identity for partitioned tables, because the
658  * operations are actually performed on the leaf partitions.
659  */
660  if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
661  return;
662 
663  /* We only need to do checks for UPDATE and DELETE. */
664  if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
665  return;
666 
667  /*
668  * It is only safe to execute UPDATE/DELETE when all columns, referenced
669  * in the row filters from publications which the relation is in, are
670  * valid - i.e. when all referenced columns are part of REPLICA IDENTITY
671  * or the table does not publish UPDATEs or DELETEs.
672  *
673  * XXX We could optimize it by first checking whether any of the
674  * publications have a row filter for this relation. If not and relation
675  * has replica identity then we can avoid building the descriptor but as
676  * this happens only one time it doesn't seem worth the additional
677  * complexity.
678  */
679  RelationBuildPublicationDesc(rel, &pubdesc);
680  if (cmd == CMD_UPDATE && !pubdesc.rf_valid_for_update)
681  ereport(ERROR,
682  (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
683  errmsg("cannot update table \"%s\"",
685  errdetail("Column used in the publication WHERE expression is not part of the replica identity.")));
686  else if (cmd == CMD_UPDATE && !pubdesc.cols_valid_for_update)
687  ereport(ERROR,
688  (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
689  errmsg("cannot update table \"%s\"",
691  errdetail("Column list used by the publication does not cover the replica identity.")));
692  else if (cmd == CMD_DELETE && !pubdesc.rf_valid_for_delete)
693  ereport(ERROR,
694  (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
695  errmsg("cannot delete from table \"%s\"",
697  errdetail("Column used in the publication WHERE expression is not part of the replica identity.")));
698  else if (cmd == CMD_DELETE && !pubdesc.cols_valid_for_delete)
699  ereport(ERROR,
700  (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
701  errmsg("cannot delete from table \"%s\"",
703  errdetail("Column list used by the publication does not cover the replica identity.")));
704 
705  /* If relation has replica identity we are always good. */
707  return;
708 
709  /* REPLICA IDENTITY FULL is also good for UPDATE/DELETE. */
710  if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
711  return;
712 
713  /*
714  * This is UPDATE/DELETE and there is no replica identity.
715  *
716  * Check if the table publishes UPDATES or DELETES.
717  */
718  if (cmd == CMD_UPDATE && pubdesc.pubactions.pubupdate)
719  ereport(ERROR,
720  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
721  errmsg("cannot update table \"%s\" because it does not have a replica identity and publishes updates",
723  errhint("To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.")));
724  else if (cmd == CMD_DELETE && pubdesc.pubactions.pubdelete)
725  ereport(ERROR,
726  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
727  errmsg("cannot delete from table \"%s\" because it does not have a replica identity and publishes deletes",
729  errhint("To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.")));
730 }
int errdetail(const char *fmt,...)
Definition: elog.c:1203
int errhint(const char *fmt,...)
Definition: elog.c:1317
int errcode(int sqlerrcode)
Definition: elog.c:857
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ereport(elevel,...)
Definition: elog.h:149
@ CMD_DELETE
Definition: nodes.h:268
@ CMD_UPDATE
Definition: nodes.h:266
#define RelationGetRelationName(relation)
Definition: rel.h:539
void RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
Definition: relcache.c:5725
Oid RelationGetReplicaIndex(Relation relation)
Definition: relcache.c:5015
PublicationActions pubactions
bool cols_valid_for_delete
bool cols_valid_for_update
Form_pg_class rd_rel
Definition: rel.h:111

References CMD_DELETE, CMD_UPDATE, PublicationDesc::cols_valid_for_delete, PublicationDesc::cols_valid_for_update, ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, OidIsValid, PublicationDesc::pubactions, PublicationActions::pubdelete, PublicationActions::pubupdate, RelationData::rd_rel, RelationBuildPublicationDesc(), RelationGetRelationName, RelationGetReplicaIndex(), PublicationDesc::rf_valid_for_delete, and PublicationDesc::rf_valid_for_update.

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

◆ CheckSubscriptionRelkind()

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

Definition at line 739 of file execReplication.c.

741 {
742  if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE)
743  ereport(ERROR,
744  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
745  errmsg("cannot use relation \"%s.%s\" as logical replication target",
746  nspname, relname),
748 }
int errdetail_relkind_not_supported(char relkind)
Definition: pg_class.c:24
NameData relname
Definition: pg_class.h:38

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

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

◆ ExecSimpleRelationDelete()

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

Definition at line 619 of file execReplication.c.

622 {
623  bool skip_tuple = false;
624  Relation rel = resultRelInfo->ri_RelationDesc;
625  ItemPointer tid = &searchslot->tts_tid;
626 
628 
629  /* BEFORE ROW DELETE Triggers */
630  if (resultRelInfo->ri_TrigDesc &&
631  resultRelInfo->ri_TrigDesc->trig_delete_before_row)
632  {
633  skip_tuple = !ExecBRDeleteTriggers(estate, epqstate, resultRelInfo,
634  tid, NULL, NULL, NULL, NULL);
635  }
636 
637  if (!skip_tuple)
638  {
639  /* OK, delete the tuple */
640  simple_table_tuple_delete(rel, tid, estate->es_snapshot);
641 
642  /* AFTER ROW DELETE Triggers */
643  ExecARDeleteTriggers(estate, resultRelInfo,
644  tid, NULL, NULL, false);
645  }
646 }
void CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
Snapshot es_snapshot
Definition: execnodes.h:624
Relation ri_RelationDesc
Definition: execnodes.h:456
TriggerDesc * ri_TrigDesc
Definition: execnodes.h:486
bool trig_delete_before_row
Definition: reltrigger.h:66
ItemPointerData tts_tid
Definition: tuptable.h:129
void simple_table_tuple_delete(Relation rel, ItemPointer tid, Snapshot snapshot)
Definition: tableam.c:290
void ExecARDeleteTriggers(EState *estate, ResultRelInfo *relinfo, ItemPointer tupleid, HeapTuple fdw_trigtuple, TransitionCaptureState *transition_capture, bool is_crosspart_update)
Definition: trigger.c:2773
bool ExecBRDeleteTriggers(EState *estate, EPQState *epqstate, ResultRelInfo *relinfo, ItemPointer tupleid, HeapTuple fdw_trigtuple, TupleTableSlot **epqslot, TM_Result *tmresult, TM_FailureData *tmfd)
Definition: trigger.c:2682

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 490 of file execReplication.c.

492 {
493  bool skip_tuple = false;
494  Relation rel = resultRelInfo->ri_RelationDesc;
495 
496  /* For now we support only tables. */
497  Assert(rel->rd_rel->relkind == RELKIND_RELATION);
498 
500 
501  /* BEFORE ROW INSERT Triggers */
502  if (resultRelInfo->ri_TrigDesc &&
503  resultRelInfo->ri_TrigDesc->trig_insert_before_row)
504  {
505  if (!ExecBRInsertTriggers(estate, resultRelInfo, slot))
506  skip_tuple = true; /* "do nothing" */
507  }
508 
509  if (!skip_tuple)
510  {
511  List *recheckIndexes = NIL;
512 
513  /* Compute stored generated columns */
514  if (rel->rd_att->constr &&
516  ExecComputeStoredGenerated(resultRelInfo, estate, slot,
517  CMD_INSERT);
518 
519  /* Check the constraints of the tuple */
520  if (rel->rd_att->constr)
521  ExecConstraints(resultRelInfo, slot, estate);
522  if (rel->rd_rel->relispartition)
523  ExecPartitionCheck(resultRelInfo, slot, estate, true);
524 
525  /* OK, store the tuple and create index entries for it */
526  simple_table_tuple_insert(resultRelInfo->ri_RelationDesc, slot);
527 
528  if (resultRelInfo->ri_NumIndices > 0)
529  recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
530  slot, estate, false, false,
531  NULL, NIL, false);
532 
533  /* AFTER ROW INSERT Triggers */
534  ExecARInsertTriggers(estate, resultRelInfo, slot,
535  recheckIndexes, NULL);
536 
537  /*
538  * XXX we should in theory pass a TransitionCaptureState object to the
539  * above to capture transition tuples, but after statement triggers
540  * don't actually get fired by replication yet anyway
541  */
542 
543  list_free(recheckIndexes);
544  }
545 }
List * ExecInsertIndexTuples(ResultRelInfo *resultRelInfo, TupleTableSlot *slot, EState *estate, bool update, bool noDupErr, bool *specConflict, List *arbiterIndexes, bool onlySummarizing)
Definition: execIndexing.c:298
bool ExecPartitionCheck(ResultRelInfo *resultRelInfo, TupleTableSlot *slot, EState *estate, bool emitError)
Definition: execMain.c:1792
void ExecConstraints(ResultRelInfo *resultRelInfo, TupleTableSlot *slot, EState *estate)
Definition: execMain.c:1916
void list_free(List *list)
Definition: list.c:1546
void ExecComputeStoredGenerated(ResultRelInfo *resultRelInfo, EState *estate, TupleTableSlot *slot, CmdType cmdtype)
@ CMD_INSERT
Definition: nodes.h:267
#define NIL
Definition: pg_list.h:68
Definition: pg_list.h:54
TupleDesc rd_att
Definition: rel.h:112
int ri_NumIndices
Definition: execnodes.h:459
bool trig_insert_before_row
Definition: reltrigger.h:56
bool has_generated_stored
Definition: tupdesc.h:45
TupleConstr * constr
Definition: tupdesc.h:85
void simple_table_tuple_insert(Relation rel, TupleTableSlot *slot)
Definition: tableam.c:276
bool ExecBRInsertTriggers(EState *estate, ResultRelInfo *relinfo, TupleTableSlot *slot)
Definition: trigger.c:2457
void ExecARInsertTriggers(EState *estate, ResultRelInfo *relinfo, TupleTableSlot *slot, List *recheckIndexes, TransitionCaptureState *transition_capture)
Definition: trigger.c:2533

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 554 of file execReplication.c.

557 {
558  bool skip_tuple = false;
559  Relation rel = resultRelInfo->ri_RelationDesc;
560  ItemPointer tid = &(searchslot->tts_tid);
561 
562  /* For now we support only tables. */
563  Assert(rel->rd_rel->relkind == RELKIND_RELATION);
564 
566 
567  /* BEFORE ROW UPDATE Triggers */
568  if (resultRelInfo->ri_TrigDesc &&
569  resultRelInfo->ri_TrigDesc->trig_update_before_row)
570  {
571  if (!ExecBRUpdateTriggers(estate, epqstate, resultRelInfo,
572  tid, NULL, slot, NULL, NULL))
573  skip_tuple = true; /* "do nothing" */
574  }
575 
576  if (!skip_tuple)
577  {
578  List *recheckIndexes = NIL;
579  TU_UpdateIndexes update_indexes;
580 
581  /* Compute stored generated columns */
582  if (rel->rd_att->constr &&
584  ExecComputeStoredGenerated(resultRelInfo, estate, slot,
585  CMD_UPDATE);
586 
587  /* Check the constraints of the tuple */
588  if (rel->rd_att->constr)
589  ExecConstraints(resultRelInfo, slot, estate);
590  if (rel->rd_rel->relispartition)
591  ExecPartitionCheck(resultRelInfo, slot, estate, true);
592 
593  simple_table_tuple_update(rel, tid, slot, estate->es_snapshot,
594  &update_indexes);
595 
596  if (resultRelInfo->ri_NumIndices > 0 && (update_indexes != TU_None))
597  recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
598  slot, estate, true, false,
599  NULL, NIL,
600  (update_indexes == TU_Summarizing));
601 
602  /* AFTER ROW UPDATE Triggers */
603  ExecARUpdateTriggers(estate, resultRelInfo,
604  NULL, NULL,
605  tid, NULL, slot,
606  recheckIndexes, NULL, false);
607 
608  list_free(recheckIndexes);
609  }
610 }
bool trig_update_before_row
Definition: reltrigger.h:61
void simple_table_tuple_update(Relation rel, ItemPointer otid, TupleTableSlot *slot, Snapshot snapshot, TU_UpdateIndexes *update_indexes)
Definition: tableam.c:335
TU_UpdateIndexes
Definition: tableam.h:118
@ TU_Summarizing
Definition: tableam.h:126
@ TU_None
Definition: tableam.h:120
bool ExecBRUpdateTriggers(EState *estate, EPQState *epqstate, ResultRelInfo *relinfo, ItemPointer tupleid, HeapTuple fdw_trigtuple, TupleTableSlot *newslot, TM_Result *tmresult, TM_FailureData *tmfd)
Definition: trigger.c:2933
void ExecARUpdateTriggers(EState *estate, ResultRelInfo *relinfo, ResultRelInfo *src_partinfo, ResultRelInfo *dst_partinfo, ItemPointer tupleid, HeapTuple fdw_trigtuple, TupleTableSlot *newslot, List *recheckIndexes, TransitionCaptureState *transition_capture, bool is_crosspart_update)
Definition: trigger.c:3096

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, TupleTableSlot::tts_tid, TU_None, and TU_Summarizing.

Referenced by apply_handle_tuple_routing(), and apply_handle_update_internal().

◆ get_equal_strategy_number()

static StrategyNumber get_equal_strategy_number ( Oid  opclass)
static

Definition at line 75 of file execReplication.c.

76 {
77  Oid am = get_opclass_method(opclass);
78 
80 }
StrategyNumber get_equal_strategy_number_for_am(Oid am)
Oid get_opclass_method(Oid opclass)
Definition: lsyscache.c:1260

References get_equal_strategy_number_for_am(), and get_opclass_method().

Referenced by build_replindex_scan_key().

◆ get_equal_strategy_number_for_am()

StrategyNumber get_equal_strategy_number_for_am ( Oid  am)

Definition at line 49 of file execReplication.c.

50 {
51  int ret;
52 
53  switch (am)
54  {
55  case BTREE_AM_OID:
57  break;
58  case HASH_AM_OID:
60  break;
61  default:
62  /* XXX: Only Btree and Hash indexes are supported */
63  ret = InvalidStrategy;
64  break;
65  }
66 
67  return ret;
68 }
#define InvalidStrategy
Definition: stratnum.h:24
#define HTEqualStrategyNumber
Definition: stratnum.h:41
#define BTEqualStrategyNumber
Definition: stratnum.h:31

References BTEqualStrategyNumber, HTEqualStrategyNumber, and InvalidStrategy.

Referenced by get_equal_strategy_number(), and IsIndexUsableForReplicaIdentityFull().

◆ RelationFindReplTupleByIndex()

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

Definition at line 176 of file execReplication.c.

180 {
182  int skey_attoff;
183  IndexScanDesc scan;
184  SnapshotData snap;
185  TransactionId xwait;
186  Relation idxrel;
187  bool found;
188  TypeCacheEntry **eq = NULL;
189  bool isIdxSafeToSkipDuplicates;
190 
191  /* Open the index. */
192  idxrel = index_open(idxoid, RowExclusiveLock);
193 
194  isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid);
195 
196  InitDirtySnapshot(snap);
197 
198  /* Build scan key. */
199  skey_attoff = build_replindex_scan_key(skey, rel, idxrel, searchslot);
200 
201  /* Start an index scan. */
202  scan = index_beginscan(rel, idxrel, &snap, skey_attoff, 0);
203 
204 retry:
205  found = false;
206 
207  index_rescan(scan, skey, skey_attoff, NULL, 0);
208 
209  /* Try to find the tuple */
210  while (index_getnext_slot(scan, ForwardScanDirection, outslot))
211  {
212  /*
213  * Avoid expensive equality check if the index is primary key or
214  * replica identity index.
215  */
216  if (!isIdxSafeToSkipDuplicates)
217  {
218  if (eq == NULL)
219  eq = palloc0(sizeof(*eq) * outslot->tts_tupleDescriptor->natts);
220 
221  if (!tuples_equal(outslot, searchslot, eq))
222  continue;
223  }
224 
225  ExecMaterializeSlot(outslot);
226 
227  xwait = TransactionIdIsValid(snap.xmin) ?
228  snap.xmin : snap.xmax;
229 
230  /*
231  * If the tuple is locked, wait for locking transaction to finish and
232  * retry.
233  */
234  if (TransactionIdIsValid(xwait))
235  {
236  XactLockTableWait(xwait, NULL, NULL, XLTW_None);
237  goto retry;
238  }
239 
240  /* Found our tuple and it's not locked */
241  found = true;
242  break;
243  }
244 
245  /* Found tuple, try to lock it in the lockmode. */
246  if (found)
247  {
248  TM_FailureData tmfd;
249  TM_Result res;
250 
252 
253  res = table_tuple_lock(rel, &(outslot->tts_tid), GetLatestSnapshot(),
254  outslot,
255  GetCurrentCommandId(false),
256  lockmode,
258  0 /* don't follow updates */ ,
259  &tmfd);
260 
262 
263  switch (res)
264  {
265  case TM_Ok:
266  break;
267  case TM_Updated:
268  /* XXX: Improve handling here */
270  ereport(LOG,
272  errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
273  else
274  ereport(LOG,
276  errmsg("concurrent update, retrying")));
277  goto retry;
278  case TM_Deleted:
279  /* XXX: Improve handling here */
280  ereport(LOG,
282  errmsg("concurrent delete, retrying")));
283  goto retry;
284  case TM_Invisible:
285  elog(ERROR, "attempted to lock invisible tuple");
286  break;
287  default:
288  elog(ERROR, "unexpected table_tuple_lock status: %u", res);
289  break;
290  }
291  }
292 
293  index_endscan(scan);
294 
295  /* Don't release lock until commit. */
296  index_close(idxrel, NoLock);
297 
298  return found;
299 }
uint32 TransactionId
Definition: c.h:652
#define LOG
Definition: elog.h:31
static bool tuples_equal(TupleTableSlot *slot1, TupleTableSlot *slot2, TypeCacheEntry **eq)
static int build_replindex_scan_key(ScanKey skey, Relation rel, Relation idxrel, TupleTableSlot *searchslot)
bool index_getnext_slot(IndexScanDesc scan, ScanDirection direction, TupleTableSlot *slot)
Definition: indexam.c:673
void index_close(Relation relation, LOCKMODE lockmode)
Definition: indexam.c:177
IndexScanDesc index_beginscan(Relation heapRelation, Relation indexRelation, Snapshot snapshot, int nkeys, int norderbys)
Definition: indexam.c:256
void index_endscan(IndexScanDesc scan)
Definition: indexam.c:378
Relation index_open(Oid relationId, LOCKMODE lockmode)
Definition: indexam.c:133
void index_rescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys)
Definition: indexam.c:352
static bool ItemPointerIndicatesMovedPartitions(const ItemPointerData *pointer)
Definition: itemptr.h:197
void XactLockTableWait(TransactionId xid, Relation rel, ItemPointer ctid, XLTW_Oper oper)
Definition: lmgr.c:667
@ XLTW_None
Definition: lmgr.h:26
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
@ LockWaitBlock
Definition: lockoptions.h:39
void * palloc0(Size size)
Definition: mcxt.c:1346
#define INDEX_MAX_KEYS
#define ERRCODE_T_R_SERIALIZATION_FAILURE
Definition: pgbench.c:76
@ ForwardScanDirection
Definition: sdir.h:28
Snapshot GetLatestSnapshot(void)
Definition: snapmgr.c:291
void PushActiveSnapshot(Snapshot snapshot)
Definition: snapmgr.c:648
void PopActiveSnapshot(void)
Definition: snapmgr.c:743
#define InitDirtySnapshot(snapshotdata)
Definition: snapmgr.h:40
Oid GetRelationIdentityOrPK(Relation rel)
Definition: relation.c:851
TransactionId xmin
Definition: snapshot.h:157
TransactionId xmax
Definition: snapshot.h:158
ItemPointerData ctid
Definition: tableam.h:150
TupleDesc tts_tupleDescriptor
Definition: tuptable.h:123
TM_Result
Definition: tableam.h:80
@ TM_Ok
Definition: tableam.h:85
@ TM_Deleted
Definition: tableam.h:100
@ TM_Updated
Definition: tableam.h:97
@ TM_Invisible
Definition: tableam.h:88
static TM_Result table_tuple_lock(Relation rel, ItemPointer tid, Snapshot snapshot, TupleTableSlot *slot, CommandId cid, LockTupleMode mode, LockWaitPolicy wait_policy, uint8 flags, TM_FailureData *tmfd)
Definition: tableam.h:1580
#define TransactionIdIsValid(xid)
Definition: transam.h:41
static void ExecMaterializeSlot(TupleTableSlot *slot)
Definition: tuptable.h:472
CommandId GetCurrentCommandId(bool used)
Definition: xact.c:826

References build_replindex_scan_key(), TM_FailureData::ctid, elog, ereport, errcode(), ERRCODE_T_R_SERIALIZATION_FAILURE, errmsg(), ERROR, ExecMaterializeSlot(), ForwardScanDirection, GetCurrentCommandId(), GetLatestSnapshot(), GetRelationIdentityOrPK(), index_beginscan(), index_close(), index_endscan(), index_getnext_slot(), INDEX_MAX_KEYS, index_open(), index_rescan(), InitDirtySnapshot, ItemPointerIndicatesMovedPartitions(), LockWaitBlock, LOG, TupleDescData::natts, NoLock, palloc0(), PopActiveSnapshot(), PushActiveSnapshot(), res, RowExclusiveLock, 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.

Referenced by FindReplTupleInLocalRel().

◆ RelationFindReplTupleSeq()

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

Definition at line 378 of file execReplication.c.

380 {
381  TupleTableSlot *scanslot;
382  TableScanDesc scan;
383  SnapshotData snap;
384  TypeCacheEntry **eq;
385  TransactionId xwait;
386  bool found;
388 
389  Assert(equalTupleDescs(desc, outslot->tts_tupleDescriptor));
390 
391  eq = palloc0(sizeof(*eq) * outslot->tts_tupleDescriptor->natts);
392 
393  /* Start a heap scan. */
394  InitDirtySnapshot(snap);
395  scan = table_beginscan(rel, &snap, 0, NULL);
396  scanslot = table_slot_create(rel, NULL);
397 
398 retry:
399  found = false;
400 
401  table_rescan(scan, NULL);
402 
403  /* Try to find the tuple */
404  while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot))
405  {
406  if (!tuples_equal(scanslot, searchslot, eq))
407  continue;
408 
409  found = true;
410  ExecCopySlot(outslot, scanslot);
411 
412  xwait = TransactionIdIsValid(snap.xmin) ?
413  snap.xmin : snap.xmax;
414 
415  /*
416  * If the tuple is locked, wait for locking transaction to finish and
417  * retry.
418  */
419  if (TransactionIdIsValid(xwait))
420  {
421  XactLockTableWait(xwait, NULL, NULL, XLTW_None);
422  goto retry;
423  }
424 
425  /* Found our tuple and it's not locked */
426  break;
427  }
428 
429  /* Found tuple, try to lock it in the lockmode. */
430  if (found)
431  {
432  TM_FailureData tmfd;
433  TM_Result res;
434 
436 
437  res = table_tuple_lock(rel, &(outslot->tts_tid), GetLatestSnapshot(),
438  outslot,
439  GetCurrentCommandId(false),
440  lockmode,
442  0 /* don't follow updates */ ,
443  &tmfd);
444 
446 
447  switch (res)
448  {
449  case TM_Ok:
450  break;
451  case TM_Updated:
452  /* XXX: Improve handling here */
454  ereport(LOG,
456  errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
457  else
458  ereport(LOG,
460  errmsg("concurrent update, retrying")));
461  goto retry;
462  case TM_Deleted:
463  /* XXX: Improve handling here */
464  ereport(LOG,
466  errmsg("concurrent delete, retrying")));
467  goto retry;
468  case TM_Invisible:
469  elog(ERROR, "attempted to lock invisible tuple");
470  break;
471  default:
472  elog(ERROR, "unexpected table_tuple_lock status: %u", res);
473  break;
474  }
475  }
476 
477  table_endscan(scan);
479 
480  return found;
481 }
#define PG_USED_FOR_ASSERTS_ONLY
Definition: c.h:182
void ExecDropSingleTupleTableSlot(TupleTableSlot *slot)
Definition: execTuples.c:1341
#define RelationGetDescr(relation)
Definition: rel.h:531
TupleTableSlot * table_slot_create(Relation relation, List **reglist)
Definition: tableam.c:91
static TableScanDesc table_beginscan(Relation rel, Snapshot snapshot, int nkeys, struct ScanKeyData *key)
Definition: tableam.h:908
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:1019
static void table_rescan(TableScanDesc scan, struct ScanKeyData *key)
Definition: tableam.h:1028
static bool table_scan_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot)
Definition: tableam.h:1055
bool equalTupleDescs(TupleDesc tupdesc1, TupleDesc tupdesc2)
Definition: tupdesc.c:419
static TupleTableSlot * ExecCopySlot(TupleTableSlot *dstslot, TupleTableSlot *srcslot)
Definition: tuptable.h:509

References Assert, TM_FailureData::ctid, elog, equalTupleDescs(), ereport, errcode(), ERRCODE_T_R_SERIALIZATION_FAILURE, 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.

Referenced by FindReplTupleInLocalRel().

◆ tuples_equal()

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

Definition at line 305 of file execReplication.c.

307 {
308  int attrnum;
309 
310  Assert(slot1->tts_tupleDescriptor->natts ==
311  slot2->tts_tupleDescriptor->natts);
312 
313  slot_getallattrs(slot1);
314  slot_getallattrs(slot2);
315 
316  /* Check equality of the attributes. */
317  for (attrnum = 0; attrnum < slot1->tts_tupleDescriptor->natts; attrnum++)
318  {
319  Form_pg_attribute att;
320  TypeCacheEntry *typentry;
321 
322  att = TupleDescAttr(slot1->tts_tupleDescriptor, attrnum);
323 
324  /*
325  * Ignore dropped and generated columns as the publisher doesn't send
326  * those
327  */
328  if (att->attisdropped || att->attgenerated)
329  continue;
330 
331  /*
332  * If one value is NULL and other is not, then they are certainly not
333  * equal
334  */
335  if (slot1->tts_isnull[attrnum] != slot2->tts_isnull[attrnum])
336  return false;
337 
338  /*
339  * If both are NULL, they can be considered equal.
340  */
341  if (slot1->tts_isnull[attrnum] || slot2->tts_isnull[attrnum])
342  continue;
343 
344  typentry = eq[attrnum];
345  if (typentry == NULL)
346  {
347  typentry = lookup_type_cache(att->atttypid,
349  if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
350  ereport(ERROR,
351  (errcode(ERRCODE_UNDEFINED_FUNCTION),
352  errmsg("could not identify an equality operator for type %s",
353  format_type_be(att->atttypid))));
354  eq[attrnum] = typentry;
355  }
356 
358  att->attcollation,
359  slot1->tts_values[attrnum],
360  slot2->tts_values[attrnum])))
361  return false;
362  }
363 
364  return true;
365 }
Datum FunctionCall2Coll(FmgrInfo *flinfo, Oid collation, Datum arg1, Datum arg2)
Definition: fmgr.c:1149
char * format_type_be(Oid type_oid)
Definition: format_type.c:343
FormData_pg_attribute * Form_pg_attribute
Definition: pg_attribute.h:209
static bool DatumGetBool(Datum X)
Definition: postgres.h:90
Oid fn_oid
Definition: fmgr.h:59
FmgrInfo eq_opr_finfo
Definition: typcache.h:75
#define TupleDescAttr(tupdesc, i)
Definition: tupdesc.h:92
static void slot_getallattrs(TupleTableSlot *slot)
Definition: tuptable.h:368
TypeCacheEntry * lookup_type_cache(Oid type_id, int flags)
Definition: typcache.c:346
#define TYPECACHE_EQ_OPR_FINFO
Definition: typcache.h:142

References Assert, DatumGetBool(), TypeCacheEntry::eq_opr_finfo, ereport, errcode(), errmsg(), ERROR, FmgrInfo::fn_oid, format_type_be(), FunctionCall2Coll(), lookup_type_cache(), TupleDescData::natts, OidIsValid, slot_getallattrs(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_tupleDescriptor, TupleTableSlot::tts_values, TupleDescAttr, and TYPECACHE_EQ_OPR_FINFO.

Referenced by RelationFindReplTupleByIndex(), and RelationFindReplTupleSeq().