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