Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * execReplication.c
4 : : * miscellaneous executor routines for logical replication
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * IDENTIFICATION
10 : : * src/backend/executor/execReplication.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres.h"
16 : :
17 : : #include "access/amapi.h"
18 : : #include "access/commit_ts.h"
19 : : #include "access/genam.h"
20 : : #include "access/gist.h"
21 : : #include "access/relscan.h"
22 : : #include "access/tableam.h"
23 : : #include "access/transam.h"
24 : : #include "access/xact.h"
25 : : #include "access/heapam.h"
26 : : #include "catalog/pg_am_d.h"
27 : : #include "commands/trigger.h"
28 : : #include "executor/executor.h"
29 : : #include "executor/nodeModifyTable.h"
30 : : #include "replication/conflict.h"
31 : : #include "replication/logicalrelation.h"
32 : : #include "storage/lmgr.h"
33 : : #include "utils/builtins.h"
34 : : #include "utils/lsyscache.h"
35 : : #include "utils/rel.h"
36 : : #include "utils/snapmgr.h"
37 : : #include "utils/syscache.h"
38 : : #include "utils/typcache.h"
39 : :
40 : :
41 : : static bool tuples_equal(TupleTableSlot *slot1, TupleTableSlot *slot2,
42 : : TypeCacheEntry **eq, Bitmapset *columns);
43 : : static bool repl_index_key_equal(Relation idxrel, TupleTableSlot *slot1,
44 : : TupleTableSlot *slot2, ScanKey skey);
45 : :
46 : : /*
47 : : * Setup a ScanKey for a search in the relation 'rel' for a tuple 'key' that
48 : : * is setup to match 'rel' (*NOT* idxrel!).
49 : : *
50 : : * Returns how many columns to use for the index scan.
51 : : *
52 : : * This is not a generic routine, idxrel must be PK, RI, or an index that can be
53 : : * used for a REPLICA IDENTITY FULL table. See FindUsableIndexForReplicaIdentityFull()
54 : : * for details.
55 : : *
56 : : * By definition, replication identity of a rel meets all limitations associated
57 : : * with that. Note that any other index could also meet these limitations.
58 : : */
59 : : static int
60 : 72112 : build_replindex_scan_key(ScanKey skey, Relation rel, Relation idxrel,
61 : : TupleTableSlot *searchslot)
62 : : {
63 : : int index_attoff;
64 : 72112 : int skey_attoff = 0;
65 : : Datum indclassDatum;
66 : : oidvector *opclass;
67 : 72112 : int2vector *indkey = &idxrel->rd_index->indkey;
68 : :
69 : 72112 : indclassDatum = SysCacheGetAttrNotNull(INDEXRELID, idxrel->rd_indextuple,
70 : : Anum_pg_index_indclass);
71 : 72112 : opclass = (oidvector *) DatumGetPointer(indclassDatum);
72 : :
73 : : /* Build scankey for every non-expression attribute in the index. */
74 [ + + ]: 144252 : for (index_attoff = 0; index_attoff < IndexRelationGetNumberOfKeyAttributes(idxrel);
75 : 72140 : index_attoff++)
76 : : {
77 : : Oid operator;
78 : : Oid optype;
79 : : Oid opfamily;
80 : : RegProcedure regop;
81 : 72140 : int table_attno = indkey->values[index_attoff];
82 : : StrategyNumber eq_strategy;
83 : :
84 [ + + ]: 72140 : if (!AttributeNumberIsValid(table_attno))
85 : : {
86 : : /*
87 : : * XXX: Currently, we don't support expressions in the scan key,
88 : : * see code below.
89 : : */
90 : 2 : continue;
91 : : }
92 : :
93 : : /*
94 : : * Load the operator info. We need this to get the equality operator
95 : : * function for the scan key.
96 : : */
97 : 72138 : optype = get_opclass_input_type(opclass->values[index_attoff]);
98 : 72138 : opfamily = get_opclass_family(opclass->values[index_attoff]);
99 : 72138 : eq_strategy = IndexAmTranslateCompareType(COMPARE_EQ, idxrel->rd_rel->relam, opfamily, false);
100 : 72138 : operator = get_opfamily_member(opfamily, optype,
101 : : optype,
102 : : eq_strategy);
103 : :
104 [ - + ]: 72138 : if (!OidIsValid(operator))
105 [ # # ]: 0 : elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
106 : : eq_strategy, optype, optype, opfamily);
107 : :
108 : 72138 : regop = get_opcode(operator);
109 : :
110 : : /* Initialize the scankey. */
111 : 72138 : ScanKeyInit(&skey[skey_attoff],
112 : 72138 : index_attoff + 1,
113 : : eq_strategy,
114 : : regop,
115 : 72138 : searchslot->tts_values[table_attno - 1]);
116 : :
117 : 72138 : skey[skey_attoff].sk_collation = idxrel->rd_indcollation[index_attoff];
118 : :
119 : : /* Check for null value. */
120 [ + + ]: 72138 : if (searchslot->tts_isnull[table_attno - 1])
121 : 1 : skey[skey_attoff].sk_flags |= (SK_ISNULL | SK_SEARCHNULL);
122 : :
123 : 72138 : skey_attoff++;
124 : : }
125 : :
126 : : /* There must always be at least one attribute for the index scan. */
127 : : Assert(skey_attoff > 0);
128 : :
129 : 72112 : return skey_attoff;
130 : : }
131 : :
132 : :
133 : : /*
134 : : * Helper function to check if it is necessary to re-fetch and lock the tuple
135 : : * due to concurrent modifications. This function should be called after
136 : : * invoking table_tuple_lock.
137 : : */
138 : : static bool
139 : 72350 : should_refetch_tuple(TM_Result res, TM_FailureData *tmfd)
140 : : {
141 : 72350 : bool refetch = false;
142 : :
143 [ + - - - : 72350 : switch (res)
- ]
144 : : {
145 : 72350 : case TM_Ok:
146 : 72350 : break;
147 : 0 : case TM_Updated:
148 : : /* XXX: Improve handling here */
149 [ # # ]: 0 : if (ItemPointerIndicatesMovedPartitions(&tmfd->ctid))
150 [ # # ]: 0 : ereport(LOG,
151 : : (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
152 : : errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
153 : : else
154 [ # # ]: 0 : ereport(LOG,
155 : : (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
156 : : errmsg("concurrent update, retrying")));
157 : 0 : refetch = true;
158 : 0 : break;
159 : 0 : case TM_Deleted:
160 : : /* XXX: Improve handling here */
161 [ # # ]: 0 : ereport(LOG,
162 : : (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
163 : : errmsg("concurrent delete, retrying")));
164 : 0 : refetch = true;
165 : 0 : break;
166 : 0 : case TM_Invisible:
167 [ # # ]: 0 : elog(ERROR, "attempted to lock invisible tuple");
168 : : break;
169 : 0 : default:
170 [ # # ]: 0 : elog(ERROR, "unexpected table_tuple_lock status: %u", res);
171 : : break;
172 : : }
173 : :
174 : 72350 : return refetch;
175 : : }
176 : :
177 : : /*
178 : : * Search the relation 'rel' for tuple using the index.
179 : : *
180 : : * If a matching tuple is found, lock it with lockmode, fill the slot with its
181 : : * contents, and return true. Return false otherwise.
182 : : *
183 : : * 'skipduplicates' specifies whether the first matching tuple can be used
184 : : * without comparing it against 'searchslot', unless the index scan sets
185 : : * xs_recheck (as with a lossy GiST index for WITHOUT OVERLAPS), in which case
186 : : * only the index key columns are compared, using the index's own equality
187 : : * semantics. If false, all matching tuples are compared against 'searchslot',
188 : : * which must contain a complete row.
189 : : */
190 : : bool
191 : 72110 : RelationFindReplTupleByIndex(Relation rel, Oid idxoid,
192 : : bool skipduplicates,
193 : : LockTupleMode lockmode,
194 : : TupleTableSlot *searchslot,
195 : : TupleTableSlot *outslot)
196 : : {
197 : : ScanKeyData skey[INDEX_MAX_KEYS];
198 : : int skey_attoff;
199 : : IndexScanDesc scan;
200 : : SnapshotData snap;
201 : : TransactionId xwait;
202 : : Relation idxrel;
203 : : bool found;
204 : 72110 : TypeCacheEntry **eq = NULL;
205 : :
206 : : /* Open the index. */
207 : 72110 : idxrel = index_open(idxoid, RowExclusiveLock);
208 : :
209 : 72110 : InitDirtySnapshot(snap);
210 : :
211 : : /* Build scan key. */
212 : 72110 : skey_attoff = build_replindex_scan_key(skey, rel, idxrel, searchslot);
213 : :
214 : : /* Start an index scan. */
215 : 72110 : scan = index_beginscan(rel, idxrel, false,
216 : : &snap, NULL, skey_attoff, 0, SO_NONE);
217 : :
218 : 0 : retry:
219 : 72110 : found = false;
220 : :
221 : 72110 : index_rescan(scan, skey, skey_attoff, NULL, 0);
222 : :
223 : : /* Try to find the tuple */
224 [ + + ]: 72112 : while (table_index_getnext_slot(scan, ForwardScanDirection, outslot))
225 : : {
226 : : /*
227 : : * Avoid expensive equality check if the index is primary key or
228 : : * replica identity index, unless the index scan indicates that a
229 : : * recheck is needed.
230 : : */
231 [ + + ]: 72096 : if (!skipduplicates)
232 : : {
233 [ + - ]: 17 : if (eq == NULL)
234 : 17 : eq = palloc0_array(TypeCacheEntry *, outslot->tts_tupleDescriptor->natts);
235 : :
236 [ - + ]: 17 : if (!tuples_equal(outslot, searchslot, eq, NULL))
237 : 0 : continue;
238 : : }
239 [ + + ]: 72079 : else if (scan->xs_recheck &&
240 [ + + ]: 5 : !repl_index_key_equal(idxrel, outslot, searchslot, skey))
241 : 2 : continue;
242 : :
243 : 72094 : ExecMaterializeSlot(outslot);
244 : :
245 : 144188 : xwait = TransactionIdIsValid(snap.xmin) ?
246 [ - + ]: 72094 : snap.xmin : snap.xmax;
247 : :
248 : : /*
249 : : * If the tuple is locked, wait for locking transaction to finish and
250 : : * retry.
251 : : */
252 [ - + ]: 72094 : if (TransactionIdIsValid(xwait))
253 : : {
254 : 0 : XactLockTableWait(xwait, NULL, NULL, XLTW_None);
255 : 0 : goto retry;
256 : : }
257 : :
258 : : /* Found our tuple and it's not locked */
259 : 72094 : found = true;
260 : 72094 : break;
261 : : }
262 : :
263 : : /* Found tuple, try to lock it in the lockmode. */
264 [ + + ]: 72110 : if (found)
265 : : {
266 : : TM_FailureData tmfd;
267 : : TM_Result res;
268 : :
269 : 72094 : PushActiveSnapshot(GetLatestSnapshot());
270 : :
271 : 72094 : res = table_tuple_lock(rel, &(outslot->tts_tid), GetActiveSnapshot(),
272 : : outslot,
273 : : GetCurrentCommandId(false),
274 : : lockmode,
275 : : LockWaitBlock,
276 : : 0 /* don't follow updates */ ,
277 : : &tmfd);
278 : :
279 : 72094 : PopActiveSnapshot();
280 : :
281 [ - + ]: 72094 : if (should_refetch_tuple(res, &tmfd))
282 : 0 : goto retry;
283 : : }
284 : :
285 : 72110 : index_endscan(scan);
286 : :
287 : : /* Don't release lock until commit. */
288 : 72110 : index_close(idxrel, NoLock);
289 : :
290 : 72110 : return found;
291 : : }
292 : :
293 : : /*
294 : : * Compare the tuples in the slots by checking if they have equal values.
295 : : *
296 : : * If 'columns' is not null, only the columns specified within it will be
297 : : * considered for the equality check, ignoring all other columns.
298 : : */
299 : : static bool
300 : 105334 : tuples_equal(TupleTableSlot *slot1, TupleTableSlot *slot2,
301 : : TypeCacheEntry **eq, Bitmapset *columns)
302 : : {
303 : : int attrnum;
304 : :
305 : : Assert(slot1->tts_tupleDescriptor->natts ==
306 : : slot2->tts_tupleDescriptor->natts);
307 : :
308 : 105334 : slot_getallattrs(slot1);
309 : 105334 : slot_getallattrs(slot2);
310 : :
311 : : /* Check equality of the attributes. */
312 [ + + ]: 105541 : for (attrnum = 0; attrnum < slot1->tts_tupleDescriptor->natts; attrnum++)
313 : : {
314 : : Form_pg_attribute att;
315 : : TypeCacheEntry *typentry;
316 : :
317 : 105374 : att = TupleDescAttr(slot1->tts_tupleDescriptor, attrnum);
318 : :
319 : : /*
320 : : * Ignore dropped and generated columns as the publisher doesn't send
321 : : * those
322 : : */
323 [ + + - + ]: 105374 : if (att->attisdropped || att->attgenerated)
324 : 1 : continue;
325 : :
326 : : /*
327 : : * Ignore columns that are not listed for checking.
328 : : */
329 [ - + ]: 105373 : if (columns &&
330 [ # # ]: 0 : !bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
331 : : columns))
332 : 0 : continue;
333 : :
334 : : /*
335 : : * If one value is NULL and other is not, then they are certainly not
336 : : * equal
337 : : */
338 [ - + ]: 105373 : if (slot1->tts_isnull[attrnum] != slot2->tts_isnull[attrnum])
339 : 0 : return false;
340 : :
341 : : /*
342 : : * If both are NULL, they can be considered equal.
343 : : */
344 [ + + - + ]: 105373 : if (slot1->tts_isnull[attrnum] || slot2->tts_isnull[attrnum])
345 : 1 : continue;
346 : :
347 : 105372 : typentry = eq[attrnum];
348 [ + + ]: 105372 : if (typentry == NULL)
349 : : {
350 : 208 : typentry = lookup_type_cache(att->atttypid,
351 : : TYPECACHE_EQ_OPR_FINFO);
352 [ - + ]: 208 : if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
353 [ # # ]: 0 : ereport(ERROR,
354 : : (errcode(ERRCODE_UNDEFINED_FUNCTION),
355 : : errmsg("could not identify an equality operator for type %s",
356 : : format_type_be(att->atttypid))));
357 : 208 : eq[attrnum] = typentry;
358 : : }
359 : :
360 [ + + ]: 105372 : if (!DatumGetBool(FunctionCall2Coll(&typentry->eq_opr_finfo,
361 : : att->attcollation,
362 : 105372 : slot1->tts_values[attrnum],
363 : 105372 : slot2->tts_values[attrnum])))
364 : 105167 : return false;
365 : : }
366 : :
367 : 167 : return true;
368 : : }
369 : :
370 : : /*
371 : : * Compare the key columns of index 'idxrel' between the two slots, using the
372 : : * equality operators and collations of the scan key 'skey' built by
373 : : * build_replindex_scan_key().
374 : : */
375 : : static bool
376 : 6 : repl_index_key_equal(Relation idxrel, TupleTableSlot *slot1,
377 : : TupleTableSlot *slot2, ScanKey skey)
378 : : {
379 : : Assert(slot1->tts_tupleDescriptor->natts ==
380 : : slot2->tts_tupleDescriptor->natts);
381 : :
382 [ + + ]: 12 : for (int i = 0; i < idxrel->rd_index->indnkeyatts; i++)
383 : : {
384 : 9 : AttrNumber keycol = idxrel->rd_index->indkey.values[i];
385 : : Datum value1;
386 : : Datum value2;
387 : : bool isnull1;
388 : : bool isnull2;
389 : :
390 : 9 : value1 = slot_getattr(slot1, keycol, &isnull1);
391 : 9 : value2 = slot_getattr(slot2, keycol, &isnull2);
392 : :
393 : : /*
394 : : * If one value is NULL and other is not, then they are certainly not
395 : : * equal
396 : : */
397 [ - + ]: 9 : if (isnull1 != isnull2)
398 : 3 : return false;
399 : :
400 : : /*
401 : : * If both are NULL, they can be considered equal.
402 : : */
403 [ + - - + ]: 9 : if (isnull1 || isnull2)
404 : 0 : continue;
405 : :
406 [ + + ]: 9 : if (!DatumGetBool(FunctionCall2Coll(&skey[i].sk_func,
407 : 9 : skey[i].sk_collation,
408 : : value1, value2)))
409 : 3 : return false;
410 : : }
411 : :
412 : 3 : return true;
413 : : }
414 : :
415 : : /*
416 : : * Search the relation 'rel' for tuple using the sequential scan.
417 : : *
418 : : * If a matching tuple is found, lock it with lockmode, fill the slot with its
419 : : * contents, and return true. Return false otherwise.
420 : : *
421 : : * Note that this stops on the first matching tuple.
422 : : *
423 : : * This can obviously be quite slow on tables that have more than few rows.
424 : : */
425 : : bool
426 : 152 : RelationFindReplTupleSeq(Relation rel, LockTupleMode lockmode,
427 : : TupleTableSlot *searchslot, TupleTableSlot *outslot)
428 : : {
429 : : TupleTableSlot *scanslot;
430 : : TableScanDesc scan;
431 : : SnapshotData snap;
432 : : TypeCacheEntry **eq;
433 : : TransactionId xwait;
434 : : bool found;
435 : 152 : TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);
436 : :
437 : : Assert(equalTupleDescs(desc, outslot->tts_tupleDescriptor));
438 : :
439 : 152 : eq = palloc0_array(TypeCacheEntry *, outslot->tts_tupleDescriptor->natts);
440 : :
441 : : /* Start a heap scan. */
442 : 152 : InitDirtySnapshot(snap);
443 : 152 : scan = table_beginscan(rel, &snap, 0, NULL,
444 : : SO_NONE);
445 : 152 : scanslot = table_slot_create(rel, NULL);
446 : :
447 : 0 : retry:
448 : 152 : found = false;
449 : :
450 : 152 : table_rescan(scan, NULL);
451 : :
452 : : /* Try to find the tuple */
453 [ + + ]: 105317 : while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot))
454 : : {
455 [ + + ]: 105313 : if (!tuples_equal(scanslot, searchslot, eq, NULL))
456 : 105165 : continue;
457 : :
458 : 148 : found = true;
459 : 148 : ExecCopySlot(outslot, scanslot);
460 : :
461 : 296 : xwait = TransactionIdIsValid(snap.xmin) ?
462 [ - + ]: 148 : snap.xmin : snap.xmax;
463 : :
464 : : /*
465 : : * If the tuple is locked, wait for locking transaction to finish and
466 : : * retry.
467 : : */
468 [ - + ]: 148 : if (TransactionIdIsValid(xwait))
469 : : {
470 : 0 : XactLockTableWait(xwait, NULL, NULL, XLTW_None);
471 : 0 : goto retry;
472 : : }
473 : :
474 : : /* Found our tuple and it's not locked */
475 : 148 : break;
476 : : }
477 : :
478 : : /* Found tuple, try to lock it in the lockmode. */
479 [ + + ]: 152 : if (found)
480 : : {
481 : : TM_FailureData tmfd;
482 : : TM_Result res;
483 : :
484 : 148 : PushActiveSnapshot(GetLatestSnapshot());
485 : :
486 : 148 : res = table_tuple_lock(rel, &(outslot->tts_tid), GetActiveSnapshot(),
487 : : outslot,
488 : : GetCurrentCommandId(false),
489 : : lockmode,
490 : : LockWaitBlock,
491 : : 0 /* don't follow updates */ ,
492 : : &tmfd);
493 : :
494 : 148 : PopActiveSnapshot();
495 : :
496 [ - + ]: 148 : if (should_refetch_tuple(res, &tmfd))
497 : 0 : goto retry;
498 : : }
499 : :
500 : 152 : table_endscan(scan);
501 : 152 : ExecDropSingleTupleTableSlot(scanslot);
502 : :
503 : 152 : return found;
504 : : }
505 : :
506 : : /*
507 : : * Build additional index information necessary for conflict detection.
508 : : */
509 : : static void
510 : 111 : BuildConflictIndexInfo(ResultRelInfo *resultRelInfo, Oid conflictindex)
511 : : {
512 [ + + ]: 332 : for (int i = 0; i < resultRelInfo->ri_NumIndices; i++)
513 : : {
514 : 221 : Relation indexRelation = resultRelInfo->ri_IndexRelationDescs[i];
515 : 221 : IndexInfo *indexRelationInfo = resultRelInfo->ri_IndexRelationInfo[i];
516 : :
517 [ + + ]: 221 : if (conflictindex != RelationGetRelid(indexRelation))
518 : 110 : continue;
519 : :
520 : : /*
521 : : * This Assert will fail if BuildSpeculativeIndexInfo() is called
522 : : * twice for the given index.
523 : : */
524 : : Assert(indexRelationInfo->ii_UniqueOps == NULL);
525 : :
526 : 111 : BuildSpeculativeIndexInfo(indexRelation, indexRelationInfo);
527 : : }
528 : 111 : }
529 : :
530 : : /*
531 : : * If the tuple is recently dead and was deleted by a transaction with a newer
532 : : * commit timestamp than previously recorded, update the associated transaction
533 : : * ID, commit time, and origin. This helps ensure that conflict detection uses
534 : : * the most recent and relevant deletion metadata.
535 : : */
536 : : static void
537 : 3 : update_most_recent_deletion_info(TupleTableSlot *scanslot,
538 : : TransactionId oldestxmin,
539 : : TransactionId *delete_xid,
540 : : TimestampTz *delete_time,
541 : : ReplOriginId *delete_origin)
542 : : {
543 : : BufferHeapTupleTableSlot *hslot;
544 : : HeapTuple tuple;
545 : : Buffer buf;
546 : 3 : bool recently_dead = false;
547 : : TransactionId xmax;
548 : : TimestampTz localts;
549 : : ReplOriginId localorigin;
550 : :
551 : 3 : hslot = (BufferHeapTupleTableSlot *) scanslot;
552 : :
553 : 3 : tuple = ExecFetchSlotHeapTuple(scanslot, false, NULL);
554 : 3 : buf = hslot->buffer;
555 : :
556 : 3 : LockBuffer(buf, BUFFER_LOCK_SHARE);
557 : :
558 : : /*
559 : : * We do not consider HEAPTUPLE_DEAD status because it indicates either
560 : : * tuples whose inserting transaction was aborted (meaning there is no
561 : : * commit timestamp or origin), or tuples deleted by a transaction older
562 : : * than oldestxmin, making it safe to ignore them during conflict
563 : : * detection (See comments atop worker.c for details).
564 : : */
565 [ + - ]: 3 : if (HeapTupleSatisfiesVacuum(tuple, oldestxmin, buf) == HEAPTUPLE_RECENTLY_DEAD)
566 : 3 : recently_dead = true;
567 : :
568 : 3 : LockBuffer(buf, BUFFER_LOCK_UNLOCK);
569 : :
570 [ - + ]: 3 : if (!recently_dead)
571 : 0 : return;
572 : :
573 : 3 : xmax = HeapTupleHeaderGetUpdateXid(tuple->t_data);
574 [ - + ]: 3 : if (!TransactionIdIsValid(xmax))
575 : 0 : return;
576 : :
577 : : /* Select the dead tuple with the most recent commit timestamp */
578 [ + - + - ]: 6 : if (TransactionIdGetCommitTsData(xmax, &localts, &localorigin) &&
579 : 3 : TimestampDifferenceExceeds(*delete_time, localts, 0))
580 : : {
581 : 3 : *delete_xid = xmax;
582 : 3 : *delete_time = localts;
583 : 3 : *delete_origin = localorigin;
584 : : }
585 : : }
586 : :
587 : : /*
588 : : * Searches the relation 'rel' for the most recently deleted tuple that matches
589 : : * the values in 'searchslot' and is not yet removable by VACUUM. The function
590 : : * returns the transaction ID, origin, and commit timestamp of the transaction
591 : : * that deleted this tuple.
592 : : *
593 : : * If 'identidxoid' is valid, it is the replica identity or primary key
594 : : * index, and only its key columns are compared. Otherwise, all columns are
595 : : * compared.
596 : : *
597 : : * 'oldestxmin' acts as a cutoff transaction ID. Tuples deleted by transactions
598 : : * with IDs >= 'oldestxmin' are considered recently dead and are eligible for
599 : : * conflict detection.
600 : : *
601 : : * Instead of stopping at the first match, we scan all matching dead tuples to
602 : : * identify most recent deletion. This is crucial because only the latest
603 : : * deletion is relevant for resolving conflicts.
604 : : *
605 : : * For example, consider a scenario on the subscriber where a row is deleted,
606 : : * re-inserted, and then deleted again only on the subscriber:
607 : : *
608 : : * - (pk, 1) - deleted at 9:00,
609 : : * - (pk, 1) - deleted at 9:02,
610 : : *
611 : : * Now, a remote update arrives: (pk, 1) -> (pk, 2), timestamped at 9:01.
612 : : *
613 : : * If we mistakenly return the older deletion (9:00), the system may wrongly
614 : : * apply the remote update using a last-update-wins strategy. Instead, we must
615 : : * recognize the more recent deletion at 9:02 and skip the update. See
616 : : * comments atop worker.c for details. Note, as of now, conflict resolution
617 : : * is not implemented. Consequently, the system may incorrectly report the
618 : : * older tuple as the conflicted one, leading to misleading results.
619 : : *
620 : : * The commit timestamp of the deleting transaction is used to determine which
621 : : * tuple was deleted most recently.
622 : : */
623 : : bool
624 : 3 : RelationFindDeletedTupleInfoSeq(Relation rel, Oid identidxoid,
625 : : TupleTableSlot *searchslot,
626 : : TransactionId oldestxmin,
627 : : TransactionId *delete_xid,
628 : : ReplOriginId *delete_origin,
629 : : TimestampTz *delete_time)
630 : : {
631 : : TupleTableSlot *scanslot;
632 : : TableScanDesc scan;
633 : : TypeCacheEntry **eq;
634 : 3 : Bitmapset *indexbitmap = NULL;
635 : 3 : TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);
636 : :
637 : : Assert(equalTupleDescs(desc, searchslot->tts_tupleDescriptor));
638 : :
639 : 3 : *delete_xid = InvalidTransactionId;
640 : 3 : *delete_origin = InvalidReplOriginId;
641 : 3 : *delete_time = 0;
642 : :
643 : : /*
644 : : * If the caller's replica identity key or primary key is unusable for
645 : : * locating deleted tuples (see IsIndexUsableForFindingDeletedTuple), a
646 : : * full table scan becomes necessary. In such cases, comparing the entire
647 : : * tuple is not required, since the remote tuple might not include all
648 : : * column values. Instead, the indexed columns alone are sufficient to
649 : : * identify the target tuple (see logicalrep_rel_mark_updatable).
650 : : */
651 [ - + ]: 3 : if (OidIsValid(identidxoid))
652 : : {
653 : : /* The index must have been locked already */
654 : 0 : Relation idxrel = index_open(identidxoid, NoLock);
655 : :
656 : : /*
657 : : * The index may no longer be the replica identity if DROP INDEX
658 : : * CONCURRENTLY or REINDEX CONCURRENTLY ran meanwhile, but it stays
659 : : * unique and non-partial, which is all we rely on. See
660 : : * FindReplTupleInLocalRel().
661 : : */
662 : : Assert(idxrel->rd_index->indisunique);
663 : : Assert(heap_attisnull(idxrel->rd_indextuple, Anum_pg_index_indpred,
664 : : NULL));
665 : :
666 [ # # ]: 0 : for (int i = 0; i < idxrel->rd_index->indnkeyatts; i++)
667 : 0 : indexbitmap = bms_add_member(indexbitmap,
668 : 0 : idxrel->rd_index->indkey.values[i] -
669 : : FirstLowInvalidHeapAttributeNumber);
670 : :
671 : 0 : index_close(idxrel, NoLock);
672 : : }
673 : :
674 : 3 : eq = palloc0_array(TypeCacheEntry *, searchslot->tts_tupleDescriptor->natts);
675 : :
676 : : /*
677 : : * Start a heap scan using SnapshotAny to identify dead tuples that are
678 : : * not visible under a standard MVCC snapshot. Tuples from transactions
679 : : * not yet committed or those just committed prior to the scan are
680 : : * excluded in update_most_recent_deletion_info().
681 : : */
682 : 3 : scan = table_beginscan(rel, SnapshotAny, 0, NULL,
683 : : SO_NONE);
684 : 3 : scanslot = table_slot_create(rel, NULL);
685 : :
686 : 3 : table_rescan(scan, NULL);
687 : :
688 : : /* Try to find the tuple */
689 [ + + ]: 7 : while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot))
690 : : {
691 [ + + ]: 4 : if (!tuples_equal(scanslot, searchslot, eq, indexbitmap))
692 : 2 : continue;
693 : :
694 : 2 : update_most_recent_deletion_info(scanslot, oldestxmin, delete_xid,
695 : : delete_time, delete_origin);
696 : : }
697 : :
698 : 3 : table_endscan(scan);
699 : 3 : ExecDropSingleTupleTableSlot(scanslot);
700 : :
701 : 3 : return *delete_time != 0;
702 : : }
703 : :
704 : : /*
705 : : * Similar to RelationFindDeletedTupleInfoSeq() but using index scan to locate
706 : : * the deleted tuple.
707 : : *
708 : : * 'skipduplicates' works as in RelationFindReplTupleByIndex().
709 : : */
710 : : bool
711 : 2 : RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid,
712 : : bool skipduplicates,
713 : : TupleTableSlot *searchslot,
714 : : TransactionId oldestxmin,
715 : : TransactionId *delete_xid,
716 : : ReplOriginId *delete_origin,
717 : : TimestampTz *delete_time)
718 : : {
719 : : Relation idxrel;
720 : : ScanKeyData skey[INDEX_MAX_KEYS];
721 : : int skey_attoff;
722 : : IndexScanDesc scan;
723 : : TupleTableSlot *scanslot;
724 : 2 : TypeCacheEntry **eq = NULL;
725 : 2 : TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);
726 : :
727 : : Assert(equalTupleDescs(desc, searchslot->tts_tupleDescriptor));
728 : : Assert(OidIsValid(idxoid));
729 : :
730 : 2 : *delete_xid = InvalidTransactionId;
731 : 2 : *delete_time = 0;
732 : 2 : *delete_origin = InvalidReplOriginId;
733 : :
734 : 2 : scanslot = table_slot_create(rel, NULL);
735 : :
736 : 2 : idxrel = index_open(idxoid, RowExclusiveLock);
737 : :
738 : : /* Build scan key. */
739 : 2 : skey_attoff = build_replindex_scan_key(skey, rel, idxrel, searchslot);
740 : :
741 : : /*
742 : : * Start an index scan using SnapshotAny to identify dead tuples that are
743 : : * not visible under a standard MVCC snapshot. Tuples from transactions
744 : : * not yet committed or those just committed prior to the scan are
745 : : * excluded in update_most_recent_deletion_info().
746 : : */
747 : 2 : scan = index_beginscan(rel, idxrel, false,
748 : : SnapshotAny, NULL, skey_attoff, 0, SO_NONE);
749 : :
750 : 2 : index_rescan(scan, skey, skey_attoff, NULL, 0);
751 : :
752 : : /* Try to find the tuple */
753 [ + + ]: 4 : while (table_index_getnext_slot(scan, ForwardScanDirection, scanslot))
754 : : {
755 : : /*
756 : : * Avoid expensive equality check if the index is primary key or
757 : : * replica identity index, unless the index scan indicates that a
758 : : * recheck is needed.
759 : : */
760 [ - + ]: 2 : if (!skipduplicates)
761 : : {
762 [ # # ]: 0 : if (eq == NULL)
763 : 0 : eq = palloc0_array(TypeCacheEntry *, scanslot->tts_tupleDescriptor->natts);
764 : :
765 [ # # ]: 0 : if (!tuples_equal(scanslot, searchslot, eq, NULL))
766 : 0 : continue;
767 : : }
768 [ + + ]: 2 : else if (scan->xs_recheck &&
769 [ + - ]: 1 : !repl_index_key_equal(idxrel, scanslot, searchslot, skey))
770 : 1 : continue;
771 : :
772 : 1 : update_most_recent_deletion_info(scanslot, oldestxmin, delete_xid,
773 : : delete_time, delete_origin);
774 : : }
775 : :
776 : 2 : index_endscan(scan);
777 : :
778 : 2 : index_close(idxrel, NoLock);
779 : :
780 : 2 : ExecDropSingleTupleTableSlot(scanslot);
781 : :
782 : 2 : return *delete_time != 0;
783 : : }
784 : :
785 : : /*
786 : : * Find the tuple that violates the passed unique index (conflictindex).
787 : : *
788 : : * If the conflicting tuple is found return true, otherwise false.
789 : : *
790 : : * We lock the tuple to avoid getting it deleted before the caller can fetch
791 : : * the required information. Note that if the tuple is deleted before a lock
792 : : * is acquired, we will retry to find the conflicting tuple again.
793 : : */
794 : : static bool
795 : 111 : FindConflictTuple(ResultRelInfo *resultRelInfo, EState *estate,
796 : : Oid conflictindex, TupleTableSlot *slot,
797 : : TupleTableSlot **conflictslot)
798 : : {
799 : 111 : Relation rel = resultRelInfo->ri_RelationDesc;
800 : : ItemPointerData conflictTid;
801 : : TM_FailureData tmfd;
802 : : TM_Result res;
803 : :
804 : 111 : *conflictslot = NULL;
805 : :
806 : : /*
807 : : * Build additional information required to check constraints violations.
808 : : * See check_exclusion_or_unique_constraint().
809 : : */
810 : 111 : BuildConflictIndexInfo(resultRelInfo, conflictindex);
811 : :
812 : 111 : retry:
813 [ + + ]: 220 : if (ExecCheckIndexConstraints(resultRelInfo, slot, estate,
814 : 111 : &conflictTid, &slot->tts_tid,
815 : : list_make1_oid(conflictindex)))
816 : : {
817 [ - + ]: 1 : if (*conflictslot)
818 : 0 : ExecDropSingleTupleTableSlot(*conflictslot);
819 : :
820 : 1 : *conflictslot = NULL;
821 : 1 : return false;
822 : : }
823 : :
824 : 108 : *conflictslot = table_slot_create(rel, NULL);
825 : :
826 : 108 : PushActiveSnapshot(GetLatestSnapshot());
827 : :
828 : 108 : res = table_tuple_lock(rel, &conflictTid, GetActiveSnapshot(),
829 : : *conflictslot,
830 : : GetCurrentCommandId(false),
831 : : LockTupleShare,
832 : : LockWaitBlock,
833 : : 0 /* don't follow updates */ ,
834 : : &tmfd);
835 : :
836 : 108 : PopActiveSnapshot();
837 : :
838 [ - + ]: 108 : if (should_refetch_tuple(res, &tmfd))
839 : 0 : goto retry;
840 : :
841 : 108 : return true;
842 : : }
843 : :
844 : : /*
845 : : * Check all the unique indexes in 'recheckIndexes' for conflict with the
846 : : * tuple in 'remoteslot' and report if found.
847 : : */
848 : : static void
849 : 59 : CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate,
850 : : ConflictType type, List *recheckIndexes,
851 : : TupleTableSlot *searchslot, TupleTableSlot *remoteslot)
852 : : {
853 : 59 : List *conflicttuples = NIL;
854 : : TupleTableSlot *conflictslot;
855 : :
856 : : /* Check all the unique indexes for conflicts */
857 [ + - + + : 225 : foreach_oid(uniqueidx, resultRelInfo->ri_onConflictArbiterIndexes)
+ + ]
858 : : {
859 [ + - + + ]: 220 : if (list_member_oid(recheckIndexes, uniqueidx) &&
860 : 111 : FindConflictTuple(resultRelInfo, estate, uniqueidx, remoteslot,
861 : : &conflictslot))
862 : : {
863 : 108 : ConflictTupleInfo *conflicttuple = palloc0_object(ConflictTupleInfo);
864 : :
865 : 108 : conflicttuple->slot = conflictslot;
866 : 108 : conflicttuple->indexoid = uniqueidx;
867 : :
868 : 108 : GetTupleTransactionInfo(conflictslot, &conflicttuple->xmin,
869 : : &conflicttuple->origin, &conflicttuple->ts);
870 : :
871 : 108 : conflicttuples = lappend(conflicttuples, conflicttuple);
872 : : }
873 : : }
874 : :
875 : : /* Report the conflict, if found */
876 [ + + ]: 57 : if (conflicttuples)
877 [ + + ]: 56 : ReportApplyConflict(estate, resultRelInfo, ERROR,
878 : 56 : list_length(conflicttuples) > 1 ? CT_MULTIPLE_UNIQUE_CONFLICTS : type,
879 : : searchslot, remoteslot, conflicttuples);
880 : 1 : }
881 : :
882 : : /*
883 : : * Insert tuple represented in the slot to the relation, update the indexes,
884 : : * and execute any constraints and per-row triggers.
885 : : *
886 : : * Caller is responsible for opening the indexes.
887 : : */
888 : : void
889 : 96226 : ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
890 : : EState *estate, TupleTableSlot *slot)
891 : : {
892 : 96226 : bool skip_tuple = false;
893 : 96226 : Relation rel = resultRelInfo->ri_RelationDesc;
894 : :
895 : : /* For now we support only tables. */
896 : : Assert(rel->rd_rel->relkind == RELKIND_RELATION);
897 : :
898 : 96226 : CheckCmdReplicaIdentity(rel, CMD_INSERT);
899 : :
900 : : /* BEFORE ROW INSERT Triggers */
901 [ + + ]: 96226 : if (resultRelInfo->ri_TrigDesc &&
902 [ + + ]: 20 : resultRelInfo->ri_TrigDesc->trig_insert_before_row)
903 : : {
904 [ + + ]: 3 : if (!ExecBRInsertTriggers(estate, resultRelInfo, slot))
905 : 1 : skip_tuple = true; /* "do nothing" */
906 : : }
907 : :
908 [ + + ]: 96226 : if (!skip_tuple)
909 : : {
910 : 96225 : List *recheckIndexes = NIL;
911 : : List *conflictindexes;
912 : 96225 : bool conflict = false;
913 : :
914 : : /* Compute stored generated columns */
915 [ + + ]: 96225 : if (rel->rd_att->constr &&
916 [ + + ]: 66086 : rel->rd_att->constr->has_generated_stored)
917 : 4 : ExecComputeStoredGenerated(resultRelInfo, estate, slot,
918 : : CMD_INSERT);
919 : :
920 : : /* Check the constraints of the tuple */
921 [ + + ]: 96225 : if (rel->rd_att->constr)
922 : 66086 : ExecConstraints(resultRelInfo, slot, estate);
923 [ + + ]: 96225 : if (rel->rd_rel->relispartition)
924 : 106 : ExecPartitionCheck(resultRelInfo, slot, estate, true);
925 : :
926 : : /* OK, store the tuple and create index entries for it */
927 : 96225 : simple_table_tuple_insert(resultRelInfo->ri_RelationDesc, slot);
928 : :
929 : 96225 : conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
930 : :
931 [ + + ]: 96225 : if (resultRelInfo->ri_NumIndices > 0)
932 : : {
933 : : uint32 flags;
934 : :
935 [ + + ]: 80788 : if (conflictindexes != NIL)
936 : 80784 : flags = EIIT_NO_DUPE_ERROR;
937 : : else
938 : 4 : flags = 0;
939 : 80788 : recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
940 : : estate, flags,
941 : : slot, conflictindexes,
942 : : &conflict);
943 : : }
944 : :
945 : : /*
946 : : * Checks the conflict indexes to fetch the conflicting local row and
947 : : * reports the conflict. We perform this check here, instead of
948 : : * performing an additional index scan before the actual insertion and
949 : : * reporting the conflict if any conflicting rows are found. This is
950 : : * to avoid the overhead of executing the extra scan for each INSERT
951 : : * operation, even when no conflict arises, which could introduce
952 : : * significant overhead to replication, particularly in cases where
953 : : * conflicts are rare.
954 : : *
955 : : * XXX OTOH, this could lead to clean-up effort for dead tuples added
956 : : * in heap and index in case of conflicts. But as conflicts shouldn't
957 : : * be a frequent thing so we preferred to save the performance
958 : : * overhead of extra scan before each insertion.
959 : : */
960 [ + + ]: 96225 : if (conflict)
961 : 57 : CheckAndReportConflict(resultRelInfo, estate, CT_INSERT_EXISTS,
962 : : recheckIndexes, NULL, slot);
963 : :
964 : : /* AFTER ROW INSERT Triggers */
965 : 96169 : ExecARInsertTriggers(estate, resultRelInfo, slot,
966 : : recheckIndexes, NULL);
967 : :
968 : : /*
969 : : * XXX we should in theory pass a TransitionCaptureState object to the
970 : : * above to capture transition tuples, but after statement triggers
971 : : * don't actually get fired by replication yet anyway
972 : : */
973 : :
974 : 96169 : list_free(recheckIndexes);
975 : : }
976 : 96170 : }
977 : :
978 : : /*
979 : : * Find the searchslot tuple and update it with data in the slot,
980 : : * update the indexes, and execute any constraints and per-row triggers.
981 : : *
982 : : * Caller is responsible for opening the indexes.
983 : : */
984 : : void
985 : 31928 : ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo,
986 : : EState *estate, EPQState *epqstate,
987 : : TupleTableSlot *searchslot, TupleTableSlot *slot)
988 : : {
989 : 31928 : bool skip_tuple = false;
990 : 31928 : Relation rel = resultRelInfo->ri_RelationDesc;
991 : 31928 : ItemPointer tid = &(searchslot->tts_tid);
992 : :
993 : : /*
994 : : * We support only non-system tables, with
995 : : * check_publication_add_relation() accountable.
996 : : */
997 : : Assert(rel->rd_rel->relkind == RELKIND_RELATION);
998 : : Assert(!IsCatalogRelation(rel));
999 : :
1000 : 31928 : CheckCmdReplicaIdentity(rel, CMD_UPDATE);
1001 : :
1002 : : /* BEFORE ROW UPDATE Triggers */
1003 [ + + ]: 31928 : if (resultRelInfo->ri_TrigDesc &&
1004 [ + + ]: 10 : resultRelInfo->ri_TrigDesc->trig_update_before_row)
1005 : : {
1006 [ + + ]: 3 : if (!ExecBRUpdateTriggers(estate, epqstate, resultRelInfo,
1007 : : tid, NULL, slot, NULL, NULL, false))
1008 : 2 : skip_tuple = true; /* "do nothing" */
1009 : : }
1010 : :
1011 [ + + ]: 31928 : if (!skip_tuple)
1012 : : {
1013 : 31926 : List *recheckIndexes = NIL;
1014 : : TU_UpdateIndexes update_indexes;
1015 : : List *conflictindexes;
1016 : 31926 : bool conflict = false;
1017 : :
1018 : : /* Compute stored generated columns */
1019 [ + + ]: 31926 : if (rel->rd_att->constr &&
1020 [ + + ]: 31880 : rel->rd_att->constr->has_generated_stored)
1021 : 2 : ExecComputeStoredGenerated(resultRelInfo, estate, slot,
1022 : : CMD_UPDATE);
1023 : :
1024 : : /* Check the constraints of the tuple */
1025 [ + + ]: 31926 : if (rel->rd_att->constr)
1026 : 31880 : ExecConstraints(resultRelInfo, slot, estate);
1027 [ + + ]: 31926 : if (rel->rd_rel->relispartition)
1028 : 12 : ExecPartitionCheck(resultRelInfo, slot, estate, true);
1029 : :
1030 : 31926 : simple_table_tuple_update(rel, tid, slot, estate->es_snapshot,
1031 : : &update_indexes);
1032 : :
1033 : 31926 : conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
1034 : :
1035 [ + + + + ]: 31926 : if (resultRelInfo->ri_NumIndices > 0 && (update_indexes != TU_None))
1036 : : {
1037 : 20110 : uint32 flags = EIIT_IS_UPDATE;
1038 : :
1039 [ + + ]: 20110 : if (conflictindexes != NIL)
1040 : 20100 : flags |= EIIT_NO_DUPE_ERROR;
1041 [ - + ]: 20110 : if (update_indexes == TU_Summarizing)
1042 : 0 : flags |= EIIT_ONLY_SUMMARIZING;
1043 : 20110 : recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
1044 : : estate, flags,
1045 : : slot, conflictindexes,
1046 : : &conflict);
1047 : : }
1048 : :
1049 : : /*
1050 : : * Refer to the comments above the call to CheckAndReportConflict() in
1051 : : * ExecSimpleRelationInsert to understand why this check is done at
1052 : : * this point.
1053 : : */
1054 [ + + ]: 31926 : if (conflict)
1055 : 2 : CheckAndReportConflict(resultRelInfo, estate, CT_UPDATE_EXISTS,
1056 : : recheckIndexes, searchslot, slot);
1057 : :
1058 : : /* AFTER ROW UPDATE Triggers */
1059 : 31924 : ExecARUpdateTriggers(estate, resultRelInfo,
1060 : : NULL, NULL,
1061 : : tid, NULL, slot,
1062 : : recheckIndexes, NULL, false);
1063 : :
1064 : 31924 : list_free(recheckIndexes);
1065 : : }
1066 : 31926 : }
1067 : :
1068 : : /*
1069 : : * Find the searchslot tuple and delete it, and execute any constraints
1070 : : * and per-row triggers.
1071 : : *
1072 : : * Caller is responsible for opening the indexes.
1073 : : */
1074 : : void
1075 : 40314 : ExecSimpleRelationDelete(ResultRelInfo *resultRelInfo,
1076 : : EState *estate, EPQState *epqstate,
1077 : : TupleTableSlot *searchslot)
1078 : : {
1079 : 40314 : bool skip_tuple = false;
1080 : 40314 : Relation rel = resultRelInfo->ri_RelationDesc;
1081 : 40314 : ItemPointer tid = &searchslot->tts_tid;
1082 : :
1083 : 40314 : CheckCmdReplicaIdentity(rel, CMD_DELETE);
1084 : :
1085 : : /* BEFORE ROW DELETE Triggers */
1086 [ + + ]: 40314 : if (resultRelInfo->ri_TrigDesc &&
1087 [ - + ]: 10 : resultRelInfo->ri_TrigDesc->trig_delete_before_row)
1088 : : {
1089 : 0 : skip_tuple = !ExecBRDeleteTriggers(estate, epqstate, resultRelInfo,
1090 : 0 : tid, NULL, NULL, NULL, NULL, false);
1091 : : }
1092 : :
1093 [ + - ]: 40314 : if (!skip_tuple)
1094 : : {
1095 : : /* OK, delete the tuple */
1096 : 40314 : simple_table_tuple_delete(rel, tid, estate->es_snapshot);
1097 : :
1098 : : /* AFTER ROW DELETE Triggers */
1099 : 40314 : ExecARDeleteTriggers(estate, resultRelInfo,
1100 : : tid, NULL, NULL, false);
1101 : : }
1102 : 40314 : }
1103 : :
1104 : : /*
1105 : : * Check if command can be executed with current replica identity.
1106 : : */
1107 : : void
1108 : 264959 : CheckCmdReplicaIdentity(Relation rel, CmdType cmd)
1109 : : {
1110 : : PublicationDesc pubdesc;
1111 : :
1112 : : /*
1113 : : * Skip checking the replica identity for partitioned tables, because the
1114 : : * operations are actually performed on the leaf partitions.
1115 : : */
1116 [ + + ]: 264959 : if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
1117 : 249320 : return;
1118 : :
1119 : : /* We only need to do checks for UPDATE and DELETE. */
1120 [ + + + + ]: 261072 : if (cmd != CMD_UPDATE && cmd != CMD_DELETE)
1121 : 168503 : return;
1122 : :
1123 : : /*
1124 : : * It is only safe to execute UPDATE/DELETE if the relation does not
1125 : : * publish UPDATEs or DELETEs, or all the following conditions are
1126 : : * satisfied:
1127 : : *
1128 : : * 1. All columns, referenced in the row filters from publications which
1129 : : * the relation is in, are valid - i.e. when all referenced columns are
1130 : : * part of REPLICA IDENTITY.
1131 : : *
1132 : : * 2. All columns, referenced in the column lists are valid - i.e. when
1133 : : * all columns referenced in the REPLICA IDENTITY are covered by the
1134 : : * column list.
1135 : : *
1136 : : * 3. All generated columns in REPLICA IDENTITY of the relation, are valid
1137 : : * - i.e. when all these generated columns are published.
1138 : : *
1139 : : * XXX We could optimize it by first checking whether any of the
1140 : : * publications have a row filter or column list for this relation, or if
1141 : : * the relation contains a generated column. If none of these exist and
1142 : : * the relation has replica identity then we can avoid building the
1143 : : * descriptor but as this happens only one time it doesn't seem worth the
1144 : : * additional complexity.
1145 : : */
1146 : 92569 : RelationBuildPublicationDesc(rel, &pubdesc);
1147 [ + + + + ]: 92569 : if (cmd == CMD_UPDATE && !pubdesc.rf_valid_for_update)
1148 [ + - ]: 40 : ereport(ERROR,
1149 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1150 : : errmsg("cannot update table \"%s\"",
1151 : : RelationGetRelationName(rel)),
1152 : : errdetail("Column used in the publication WHERE expression is not part of the replica identity.")));
1153 [ + + + + ]: 92529 : else if (cmd == CMD_UPDATE && !pubdesc.cols_valid_for_update)
1154 [ + - ]: 72 : ereport(ERROR,
1155 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1156 : : errmsg("cannot update table \"%s\"",
1157 : : RelationGetRelationName(rel)),
1158 : : errdetail("Column list used by the publication does not cover the replica identity.")));
1159 [ + + + + ]: 92457 : else if (cmd == CMD_UPDATE && !pubdesc.gencols_valid_for_update)
1160 [ + - ]: 16 : ereport(ERROR,
1161 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1162 : : errmsg("cannot update table \"%s\"",
1163 : : RelationGetRelationName(rel)),
1164 : : errdetail("Replica identity must not contain unpublished generated columns.")));
1165 [ + + - + ]: 92441 : else if (cmd == CMD_DELETE && !pubdesc.rf_valid_for_delete)
1166 [ # # ]: 0 : ereport(ERROR,
1167 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1168 : : errmsg("cannot delete from table \"%s\"",
1169 : : RelationGetRelationName(rel)),
1170 : : errdetail("Column used in the publication WHERE expression is not part of the replica identity.")));
1171 [ + + - + ]: 92441 : else if (cmd == CMD_DELETE && !pubdesc.cols_valid_for_delete)
1172 [ # # ]: 0 : ereport(ERROR,
1173 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1174 : : errmsg("cannot delete from table \"%s\"",
1175 : : RelationGetRelationName(rel)),
1176 : : errdetail("Column list used by the publication does not cover the replica identity.")));
1177 [ + + - + ]: 92441 : else if (cmd == CMD_DELETE && !pubdesc.gencols_valid_for_delete)
1178 [ # # ]: 0 : ereport(ERROR,
1179 : : (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
1180 : : errmsg("cannot delete from table \"%s\"",
1181 : : RelationGetRelationName(rel)),
1182 : : errdetail("Replica identity must not contain unpublished generated columns.")));
1183 : :
1184 : : /* If relation has replica identity we are always good. */
1185 [ + + ]: 92441 : if (OidIsValid(RelationGetReplicaIndex(rel)))
1186 : 76691 : return;
1187 : :
1188 : : /* REPLICA IDENTITY FULL is also good for UPDATE/DELETE. */
1189 [ + + ]: 15750 : if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
1190 : 239 : return;
1191 : :
1192 : : /*
1193 : : * This is UPDATE/DELETE and there is no replica identity.
1194 : : *
1195 : : * Check if the table publishes UPDATES or DELETES.
1196 : : */
1197 [ + + + + ]: 15511 : if (cmd == CMD_UPDATE && pubdesc.pubactions.pubupdate)
1198 [ + - ]: 81 : ereport(ERROR,
1199 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1200 : : errmsg("cannot update table \"%s\" because it does not have a replica identity and publishes updates",
1201 : : RelationGetRelationName(rel)),
1202 : : errhint("To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.")));
1203 [ + + + + ]: 15430 : else if (cmd == CMD_DELETE && pubdesc.pubactions.pubdelete)
1204 [ + - ]: 9 : ereport(ERROR,
1205 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1206 : : errmsg("cannot delete from table \"%s\" because it does not have a replica identity and publishes deletes",
1207 : : RelationGetRelationName(rel)),
1208 : : errhint("To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.")));
1209 : : }
1210 : :
1211 : :
1212 : : /*
1213 : : * Check if we support writing into specific relkind of local relation and check
1214 : : * if it aligns with the relkind of the relation on the publisher.
1215 : : *
1216 : : * The nspname and relname are only needed for error reporting.
1217 : : */
1218 : : void
1219 : 1084 : CheckSubscriptionRelkind(char localrelkind, char remoterelkind,
1220 : : const char *nspname, const char *relname)
1221 : : {
1222 [ + + + + ]: 1084 : if (localrelkind != RELKIND_RELATION &&
1223 [ - + ]: 22 : localrelkind != RELKIND_PARTITIONED_TABLE &&
1224 : : localrelkind != RELKIND_SEQUENCE)
1225 [ # # ]: 0 : ereport(ERROR,
1226 : : (errcode(ERRCODE_WRONG_OBJECT_TYPE),
1227 : : errmsg("cannot use relation \"%s.%s\" as logical replication target",
1228 : : nspname, relname),
1229 : : errdetail_relkind_not_supported(localrelkind)));
1230 : :
1231 : : /*
1232 : : * Allow RELKIND_RELATION and RELKIND_PARTITIONED_TABLE to be treated
1233 : : * interchangeably, but ensure that sequences (RELKIND_SEQUENCE) match
1234 : : * exactly on both publisher and subscriber.
1235 : : */
1236 [ + + + - : 1084 : if ((localrelkind == RELKIND_SEQUENCE && remoterelkind != RELKIND_SEQUENCE) ||
+ + ]
1237 [ - + ]: 1062 : (localrelkind != RELKIND_SEQUENCE && remoterelkind == RELKIND_SEQUENCE))
1238 [ # # # # : 0 : ereport(ERROR,
# # ]
1239 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1240 : : /* translator: 3rd and 4th %s are "sequence" or "table" */
1241 : : errmsg("relation \"%s.%s\" type mismatch: source \"%s\", target \"%s\"",
1242 : : nspname, relname,
1243 : : remoterelkind == RELKIND_SEQUENCE ? "sequence" : "table",
1244 : : localrelkind == RELKIND_SEQUENCE ? "sequence" : "table"));
1245 : 1084 : }
|
