Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * conflict.c
3 : : * Support routines for logging conflicts.
4 : : *
5 : : * Copyright (c) 2024-2026, PostgreSQL Global Development Group
6 : : *
7 : : * IDENTIFICATION
8 : : * src/backend/replication/logical/conflict.c
9 : : *
10 : : * This file contains the code for logging conflicts on the subscriber during
11 : : * logical replication.
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres.h"
16 : :
17 : : #include "access/commit_ts.h"
18 : : #include "access/genam.h"
19 : : #include "access/tableam.h"
20 : : #include "catalog/heap.h"
21 : : #include "catalog/pg_am.h"
22 : : #include "catalog/pg_namespace.h"
23 : : #include "catalog/toasting.h"
24 : : #include "executor/executor.h"
25 : : #include "pgstat.h"
26 : : #include "replication/conflict.h"
27 : : #include "replication/worker_internal.h"
28 : : #include "storage/lmgr.h"
29 : : #include "utils/lsyscache.h"
30 : :
31 : : /*
32 : : * String representations for the supported conflict logging destinations.
33 : : */
34 : : const char *const ConflictLogDestNames[] = {
35 : : [CONFLICT_LOG_DEST_LOG] = "log",
36 : : [CONFLICT_LOG_DEST_TABLE] = "table",
37 : : [CONFLICT_LOG_DEST_ALL] = "all"
38 : : };
39 : :
40 : : StaticAssertDecl(lengthof(ConflictLogDestNames) == CONFLICT_LOG_DEST_ALL + 1,
41 : : "ConflictLogDestNames length mismatch");
42 : :
43 : :
44 : : /* Structure to hold metadata for one column of the conflict log table */
45 : : typedef struct ConflictLogColumnDef
46 : : {
47 : : const char *attname; /* Column name */
48 : : Oid atttypid; /* Data type OID */
49 : : } ConflictLogColumnDef;
50 : :
51 : : /*
52 : : * Schema definition for conflict log tables.
53 : : *
54 : : * Defines the fixed schema of the per-subscription conflict log table created
55 : : * in the pg_conflict namespace. Each entry specifies the column name and its
56 : : * type OID; the table is created in this column order by
57 : : * create_conflict_log_table().
58 : : *
59 : : * The tuple/key columns (replica_identity, remote_tuple, local_conflicts) are
60 : : * typed json rather than jsonb on purpose: they hold an exact audit snapshot
61 : : * of the applied tuples and replica identity, and json preserves the verbatim
62 : : * representation whereas jsonb would normalize it. Indexing them (jsonb's main
63 : : * advantage) wouldn't help anyway, as the conflict log is looked up by its
64 : : * scalar columns (relid, conflict_type, commit timestamp) while these json
65 : : * columns are per-conflict payload to inspect, not search keys.
66 : : */
67 : : static const ConflictLogColumnDef ConflictLogSchema[] = {
68 : : {.attname = "relid", .atttypid = OIDOID},
69 : : {.attname = "schemaname", .atttypid = TEXTOID},
70 : : {.attname = "relname", .atttypid = TEXTOID},
71 : : {.attname = "conflict_type", .atttypid = TEXTOID},
72 : : {.attname = "remote_xid", .atttypid = XIDOID},
73 : : {.attname = "remote_commit_lsn", .atttypid = LSNOID},
74 : : {.attname = "remote_commit_ts", .atttypid = TIMESTAMPTZOID},
75 : : {.attname = "remote_origin", .atttypid = TEXTOID},
76 : : {.attname = "replica_identity_full", .atttypid = BOOLOID},
77 : : {.attname = "replica_identity", .atttypid = JSONOID},
78 : : {.attname = "remote_tuple", .atttypid = JSONOID},
79 : : {.attname = "local_conflicts", .atttypid = JSONARRAYOID}
80 : : };
81 : :
82 : : #define NUM_CONFLICT_ATTRS ((AttrNumber) lengthof(ConflictLogSchema))
83 : :
84 : : static const char *const ConflictTypeNames[] = {
85 : : [CT_INSERT_EXISTS] = "insert_exists",
86 : : [CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
87 : : [CT_UPDATE_EXISTS] = "update_exists",
88 : : [CT_UPDATE_MISSING] = "update_missing",
89 : : [CT_DELETE_ORIGIN_DIFFERS] = "delete_origin_differs",
90 : : [CT_UPDATE_DELETED] = "update_deleted",
91 : : [CT_DELETE_MISSING] = "delete_missing",
92 : : [CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts"
93 : : };
94 : :
95 : : static int errcode_apply_conflict(ConflictType type);
96 : : static void errdetail_apply_conflict(EState *estate,
97 : : ResultRelInfo *relinfo,
98 : : ConflictType type,
99 : : TupleTableSlot *searchslot,
100 : : TupleTableSlot *localslot,
101 : : TupleTableSlot *remoteslot,
102 : : Oid indexoid, TransactionId localxmin,
103 : : ReplOriginId localorigin,
104 : : TimestampTz localts, StringInfo err_msg);
105 : : static void get_tuple_desc(EState *estate, ResultRelInfo *relinfo,
106 : : ConflictType type, char **key_desc,
107 : : TupleTableSlot *localslot, char **local_desc,
108 : : TupleTableSlot *remoteslot, char **remote_desc,
109 : : TupleTableSlot *searchslot, char **search_desc,
110 : : Oid indexoid);
111 : : static char *build_index_value_desc(EState *estate, Relation localrel,
112 : : TupleTableSlot *slot, Oid indexoid);
113 : :
114 : : /*
115 : : * Builds the TupleDesc for the conflict log table.
116 : : */
117 : : static TupleDesc
118 : 14 : create_conflict_log_table_tupdesc(void)
119 : : {
120 : : TupleDesc tupdesc;
121 : :
122 : 14 : tupdesc = CreateTemplateTupleDesc(NUM_CONFLICT_ATTRS);
123 : :
124 [ + + ]: 182 : for (int i = 0; i < NUM_CONFLICT_ATTRS; i++)
125 : 168 : TupleDescInitEntry(tupdesc, i + 1,
126 : 168 : ConflictLogSchema[i].attname,
127 : 168 : ConflictLogSchema[i].atttypid,
128 : : -1, 0);
129 : :
130 : 14 : TupleDescFinalize(tupdesc);
131 : :
132 : 14 : return tupdesc;
133 : : }
134 : :
135 : : /*
136 : : * Create a structured conflict log table for a subscription.
137 : : *
138 : : * The table is created within the system-managed 'pg_conflict' namespace to
139 : : * prevent users from manually dropping or altering it. This also prevents
140 : : * accidental name collisions with user-created tables with the same name.
141 : : *
142 : : * The table name is generated automatically using the subscription's OID
143 : : * (e.g., "pg_conflict_log_<subid>") to ensure uniqueness within the
144 : : * cluster and to avoid collisions during subscription renames.
145 : : */
146 : : Oid
147 : 14 : create_conflict_log_table(Oid subid, char *subname, Oid subowner)
148 : : {
149 : : TupleDesc tupdesc;
150 : : Oid relid;
151 : : char relname[NAMEDATALEN];
152 : :
153 : 14 : snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
154 : :
155 : : /* Build the tuple descriptor for the new table. */
156 : 14 : tupdesc = create_conflict_log_table_tupdesc();
157 : :
158 : : /* Create conflict log table. */
159 : 14 : relid = heap_create_with_catalog(relname,
160 : : PG_CONFLICT_NAMESPACE,
161 : : 0, /* tablespace */
162 : : InvalidOid, /* relid */
163 : : InvalidOid, /* reltypeid */
164 : : InvalidOid, /* reloftypeid */
165 : : subowner,
166 : : HEAP_TABLE_AM_OID,
167 : : tupdesc,
168 : : NIL,
169 : : RELKIND_RELATION,
170 : : RELPERSISTENCE_PERMANENT,
171 : : false, /* shared_relation */
172 : : false, /* mapped_relation */
173 : : ONCOMMIT_NOOP,
174 : : (Datum) 0, /* reloptions */
175 : : false, /* use_user_acl */
176 : : false, /* allow_system_table_mods */
177 : : true, /* is_internal */
178 : : InvalidOid, /* relrewrite */
179 : : NULL); /* typaddress */
180 : : Assert(OidIsValid(relid));
181 : :
182 : : /* Release tuple descriptor memory. */
183 : 14 : FreeTupleDesc(tupdesc);
184 : :
185 : : /*
186 : : * We must bump the command counter to make the newly-created relation
187 : : * tuple visible for opening.
188 : : */
189 : 14 : CommandCounterIncrement();
190 : :
191 : : /*
192 : : * Create a TOAST table for the conflict log to support out-of-line
193 : : * storage of large json data.
194 : : */
195 : 14 : NewRelationCreateToastTable(relid, (Datum) 0);
196 : :
197 [ + + ]: 14 : ereport(NOTICE,
198 : : (errmsg("created conflict log table \"%s\" for subscription \"%s\"",
199 : : get_qualified_objname(PG_CONFLICT_NAMESPACE, relname),
200 : : subname)));
201 : :
202 : 14 : return relid;
203 : : }
204 : :
205 : : /*
206 : : * Convert the string representation of a conflict logging destination to its
207 : : * corresponding enum value.
208 : : */
209 : : ConflictLogDest
210 : 54 : GetConflictLogDest(const char *dest)
211 : : {
212 : : /* NULL defaults to LOG. */
213 [ + - + + ]: 54 : if (dest == NULL || pg_strcasecmp(dest, "log") == 0)
214 : 12 : return CONFLICT_LOG_DEST_LOG;
215 : :
216 [ + + ]: 42 : if (pg_strcasecmp(dest, "table") == 0)
217 : 25 : return CONFLICT_LOG_DEST_TABLE;
218 : :
219 [ + + ]: 17 : if (pg_strcasecmp(dest, "all") == 0)
220 : 9 : return CONFLICT_LOG_DEST_ALL;
221 : :
222 : : /* Unrecognized string. */
223 [ + - ]: 8 : ereport(ERROR,
224 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
225 : : errmsg("unrecognized conflict_log_destination value: \"%s\"", dest),
226 : : errhint("Valid values are \"log\", \"table\", and \"all\".")));
227 : : }
228 : :
229 : : /*
230 : : * Get the xmin and commit timestamp data (origin and timestamp) associated
231 : : * with the provided local row.
232 : : *
233 : : * Return true if the commit timestamp data was found, false otherwise.
234 : : */
235 : : bool
236 : 72340 : GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin,
237 : : ReplOriginId *localorigin, TimestampTz *localts)
238 : : {
239 : : Datum xminDatum;
240 : : bool isnull;
241 : :
242 : 72340 : xminDatum = slot_getsysattr(localslot, MinTransactionIdAttributeNumber,
243 : : &isnull);
244 : 72340 : *xmin = DatumGetTransactionId(xminDatum);
245 : : Assert(!isnull);
246 : :
247 : : /*
248 : : * The commit timestamp data is not available if track_commit_timestamp is
249 : : * disabled.
250 : : */
251 [ + + ]: 72340 : if (!track_commit_timestamp)
252 : : {
253 : 72254 : *localorigin = InvalidReplOriginId;
254 : 72254 : *localts = 0;
255 : 72254 : return false;
256 : : }
257 : :
258 : 86 : return TransactionIdGetCommitTsData(*xmin, localts, localorigin);
259 : : }
260 : :
261 : : /*
262 : : * This function is used to report a conflict while applying replication
263 : : * changes.
264 : : *
265 : : * 'searchslot' should contain the tuple used to search the local row to be
266 : : * updated or deleted.
267 : : *
268 : : * 'remoteslot' should contain the remote new tuple, if any.
269 : : *
270 : : * conflicttuples is a list of local rows that caused the conflict and the
271 : : * conflict related information. See ConflictTupleInfo.
272 : : *
273 : : * The caller must ensure that all the indexes passed in ConflictTupleInfo are
274 : : * locked so that we can fetch and display the conflicting key values.
275 : : */
276 : : void
277 : 77 : ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
278 : : ConflictType type, TupleTableSlot *searchslot,
279 : : TupleTableSlot *remoteslot, List *conflicttuples)
280 : : {
281 : 77 : Relation localrel = relinfo->ri_RelationDesc;
282 : : StringInfoData err_detail;
283 : :
284 : 77 : initStringInfo(&err_detail);
285 : :
286 : : /* Form errdetail message by combining conflicting tuples information. */
287 [ + - + + : 275 : foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
+ + ]
288 : 121 : errdetail_apply_conflict(estate, relinfo, type, searchslot,
289 : : conflicttuple->slot, remoteslot,
290 : : conflicttuple->indexoid,
291 : : conflicttuple->xmin,
292 : 121 : conflicttuple->origin,
293 : : conflicttuple->ts,
294 : : &err_detail);
295 : :
296 : 77 : pgstat_report_subscription_conflict(MySubscription->oid, type);
297 : :
298 [ + - ]: 77 : ereport(elevel,
299 : : errcode_apply_conflict(type),
300 : : errmsg("conflict detected on relation \"%s.%s\": conflict=%s",
301 : : get_namespace_name(RelationGetNamespace(localrel)),
302 : : RelationGetRelationName(localrel),
303 : : ConflictTypeNames[type]),
304 : : errdetail_internal("%s", err_detail.data));
305 : 29 : }
306 : :
307 : : /*
308 : : * Find all unique indexes to check for a conflict and store them into
309 : : * ResultRelInfo.
310 : : */
311 : : void
312 : 128639 : InitConflictIndexes(ResultRelInfo *relInfo)
313 : : {
314 : 128639 : List *uniqueIndexes = NIL;
315 : :
316 [ + + ]: 236871 : for (int i = 0; i < relInfo->ri_NumIndices; i++)
317 : : {
318 : 108232 : Relation indexRelation = relInfo->ri_IndexRelationDescs[i];
319 : :
320 [ - + ]: 108232 : if (indexRelation == NULL)
321 : 0 : continue;
322 : :
323 : : /* Detect conflict only for unique indexes */
324 [ + + ]: 108232 : if (!relInfo->ri_IndexRelationInfo[i]->ii_Unique)
325 : 29 : continue;
326 : :
327 : : /* Don't support conflict detection for deferrable index */
328 [ - + ]: 108203 : if (!indexRelation->rd_index->indimmediate)
329 : 0 : continue;
330 : :
331 : 108203 : uniqueIndexes = lappend_oid(uniqueIndexes,
332 : : RelationGetRelid(indexRelation));
333 : : }
334 : :
335 : 128639 : relInfo->ri_onConflictArbiterIndexes = uniqueIndexes;
336 : 128639 : }
337 : :
338 : : /*
339 : : * Add SQLSTATE error code to the current conflict report.
340 : : */
341 : : static int
342 : 77 : errcode_apply_conflict(ConflictType type)
343 : : {
344 [ + + - ]: 77 : switch (type)
345 : : {
346 : 48 : case CT_INSERT_EXISTS:
347 : : case CT_UPDATE_EXISTS:
348 : : case CT_MULTIPLE_UNIQUE_CONFLICTS:
349 : 48 : return errcode(ERRCODE_UNIQUE_VIOLATION);
350 : 29 : case CT_UPDATE_ORIGIN_DIFFERS:
351 : : case CT_UPDATE_MISSING:
352 : : case CT_DELETE_ORIGIN_DIFFERS:
353 : : case CT_UPDATE_DELETED:
354 : : case CT_DELETE_MISSING:
355 : 29 : return errcode(ERRCODE_T_R_SERIALIZATION_FAILURE);
356 : : }
357 : :
358 : : Assert(false);
359 : 0 : return 0; /* silence compiler warning */
360 : : }
361 : :
362 : : /*
363 : : * Helper function to build the additional details for conflicting key,
364 : : * local row, remote row, and replica identity columns.
365 : : */
366 : : static void
367 : 169 : append_tuple_value_detail(StringInfo buf, List *tuple_values)
368 : : {
369 : 169 : bool first = true;
370 : :
371 : : Assert(buf != NULL && tuple_values != NIL);
372 : :
373 [ + - + + : 675 : foreach_ptr(char, tuple_value, tuple_values)
+ + ]
374 : : {
375 : : /*
376 : : * Skip if the value is NULL. This means the current user does not
377 : : * have enough permissions to see all columns in the table. See
378 : : * get_tuple_desc().
379 : : */
380 [ + + ]: 337 : if (!tuple_value)
381 : 51 : continue;
382 : :
383 : : /* standard SQL punctuation, not translated */
384 [ + + ]: 286 : if (!first)
385 : 117 : appendStringInfoString(buf, ", ");
386 : :
387 : 286 : appendStringInfoString(buf, tuple_value);
388 : 286 : first = false;
389 : : }
390 : 169 : }
391 : :
392 : : /*
393 : : * Add an errdetail() line showing conflict detail.
394 : : *
395 : : * The DETAIL line comprises of two parts:
396 : : * 1. Explanation of the conflict type, including the origin and commit
397 : : * timestamp of the local row.
398 : : * 2. Display of conflicting key, local row, remote new row, and replica
399 : : * identity columns, if any. The remote old row is excluded as its
400 : : * information is covered in the replica identity columns.
401 : : */
402 : : static void
403 : 121 : errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo,
404 : : ConflictType type, TupleTableSlot *searchslot,
405 : : TupleTableSlot *localslot, TupleTableSlot *remoteslot,
406 : : Oid indexoid, TransactionId localxmin,
407 : : ReplOriginId localorigin, TimestampTz localts,
408 : : StringInfo err_msg)
409 : : {
410 : : StringInfoData err_detail;
411 : : StringInfoData tuple_buf;
412 : : char *origin_name;
413 : 121 : char *key_desc = NULL;
414 : 121 : char *local_desc = NULL;
415 : 121 : char *remote_desc = NULL;
416 : 121 : char *search_desc = NULL;
417 : :
418 : : /* Get key, replica identity, remote, and local value data */
419 : 121 : get_tuple_desc(estate, relinfo, type, &key_desc,
420 : : localslot, &local_desc,
421 : : remoteslot, &remote_desc,
422 : : searchslot, &search_desc,
423 : : indexoid);
424 : :
425 : 121 : initStringInfo(&err_detail);
426 : 121 : initStringInfo(&tuple_buf);
427 : :
428 : : /* Construct a detailed message describing the type of conflict */
429 [ + + + + : 121 : switch (type)
+ + - ]
430 : : {
431 : 92 : case CT_INSERT_EXISTS:
432 : : case CT_UPDATE_EXISTS:
433 : : case CT_MULTIPLE_UNIQUE_CONFLICTS:
434 : : Assert(OidIsValid(indexoid) &&
435 : : CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
436 : :
437 [ + + ]: 92 : if (err_msg->len == 0)
438 : : {
439 : 48 : append_tuple_value_detail(&tuple_buf,
440 : : list_make2(remote_desc, search_desc));
441 : :
442 [ + - ]: 48 : if (tuple_buf.len)
443 : 48 : appendStringInfo(&err_detail, _("Could not apply remote change: %s.\n"),
444 : : tuple_buf.data);
445 : : else
446 : 0 : appendStringInfo(&err_detail, _("Could not apply remote change.\n"));
447 : :
448 : :
449 : 48 : resetStringInfo(&tuple_buf);
450 : : }
451 : :
452 : 92 : append_tuple_value_detail(&tuple_buf,
453 : : list_make2(key_desc, local_desc));
454 : :
455 [ + + ]: 92 : if (localts)
456 : : {
457 [ - + ]: 3 : if (localorigin == InvalidReplOriginId)
458 : : {
459 [ # # ]: 0 : if (tuple_buf.len)
460 : 0 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified locally in transaction %u at %s: %s."),
461 : : get_rel_name(indexoid),
462 : : localxmin, timestamptz_to_str(localts),
463 : : tuple_buf.data);
464 : : else
465 : 0 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified locally in transaction %u at %s."),
466 : : get_rel_name(indexoid),
467 : : localxmin, timestamptz_to_str(localts));
468 : : }
469 [ + + ]: 3 : else if (replorigin_by_oid(localorigin, true, &origin_name))
470 : : {
471 [ + - ]: 1 : if (tuple_buf.len)
472 : 1 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by origin \"%s\" in transaction %u at %s: %s."),
473 : : get_rel_name(indexoid), origin_name,
474 : : localxmin, timestamptz_to_str(localts),
475 : : tuple_buf.data);
476 : : else
477 : 0 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by origin \"%s\" in transaction %u at %s."),
478 : : get_rel_name(indexoid), origin_name,
479 : : localxmin, timestamptz_to_str(localts));
480 : : }
481 : :
482 : : /*
483 : : * The origin that modified this row has been removed. This
484 : : * can happen if the origin was created by a different apply
485 : : * worker and its associated subscription and origin were
486 : : * dropped after updating the row, or if the origin was
487 : : * manually dropped by the user.
488 : : */
489 : : else
490 : : {
491 [ + - ]: 2 : if (tuple_buf.len)
492 : 2 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by a non-existent origin in transaction %u at %s: %s."),
493 : : get_rel_name(indexoid),
494 : : localxmin, timestamptz_to_str(localts),
495 : : tuple_buf.data);
496 : : else
497 : 0 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified by a non-existent origin in transaction %u at %s."),
498 : : get_rel_name(indexoid),
499 : : localxmin, timestamptz_to_str(localts));
500 : : }
501 : : }
502 : : else
503 : : {
504 [ + - ]: 89 : if (tuple_buf.len)
505 : 89 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified in transaction %u: %s."),
506 : : get_rel_name(indexoid), localxmin,
507 : : tuple_buf.data);
508 : : else
509 : 0 : appendStringInfo(&err_detail, _("Key already exists in unique index \"%s\", modified in transaction %u."),
510 : : get_rel_name(indexoid), localxmin);
511 : : }
512 : :
513 : 92 : break;
514 : :
515 : 3 : case CT_UPDATE_ORIGIN_DIFFERS:
516 : 3 : append_tuple_value_detail(&tuple_buf,
517 : : list_make3(local_desc, remote_desc,
518 : : search_desc));
519 : :
520 [ + + ]: 3 : if (localorigin == InvalidReplOriginId)
521 : : {
522 [ + - ]: 1 : if (tuple_buf.len)
523 : 1 : appendStringInfo(&err_detail, _("Updating the row that was modified locally in transaction %u at %s: %s."),
524 : : localxmin, timestamptz_to_str(localts),
525 : : tuple_buf.data);
526 : : else
527 : 0 : appendStringInfo(&err_detail, _("Updating the row that was modified locally in transaction %u at %s."),
528 : : localxmin, timestamptz_to_str(localts));
529 : : }
530 [ + + ]: 2 : else if (replorigin_by_oid(localorigin, true, &origin_name))
531 : : {
532 [ + - ]: 1 : if (tuple_buf.len)
533 : 1 : appendStringInfo(&err_detail, _("Updating the row that was modified by a different origin \"%s\" in transaction %u at %s: %s."),
534 : : origin_name, localxmin,
535 : : timestamptz_to_str(localts),
536 : : tuple_buf.data);
537 : : else
538 : 0 : appendStringInfo(&err_detail, _("Updating the row that was modified by a different origin \"%s\" in transaction %u at %s."),
539 : : origin_name, localxmin,
540 : : timestamptz_to_str(localts));
541 : : }
542 : :
543 : : /* The origin that modified this row has been removed. */
544 : : else
545 : : {
546 [ + - ]: 1 : if (tuple_buf.len)
547 : 1 : appendStringInfo(&err_detail, _("Updating the row that was modified by a non-existent origin in transaction %u at %s: %s."),
548 : : localxmin, timestamptz_to_str(localts),
549 : : tuple_buf.data);
550 : : else
551 : 0 : appendStringInfo(&err_detail, _("Updating the row that was modified by a non-existent origin in transaction %u at %s."),
552 : : localxmin, timestamptz_to_str(localts));
553 : : }
554 : :
555 : 3 : break;
556 : :
557 : 3 : case CT_UPDATE_DELETED:
558 : 3 : append_tuple_value_detail(&tuple_buf,
559 : : list_make2(remote_desc, search_desc));
560 : :
561 [ + - ]: 3 : if (tuple_buf.len)
562 : 3 : appendStringInfo(&err_detail, _("Could not find the row to be updated: %s.\n"),
563 : : tuple_buf.data);
564 : : else
565 : 0 : appendStringInfo(&err_detail, _("Could not find the row to be updated.\n"));
566 : :
567 [ + - ]: 3 : if (localts)
568 : : {
569 [ + - ]: 3 : if (localorigin == InvalidReplOriginId)
570 : 3 : appendStringInfo(&err_detail, _("The row to be updated was deleted locally in transaction %u at %s"),
571 : : localxmin, timestamptz_to_str(localts));
572 [ # # ]: 0 : else if (replorigin_by_oid(localorigin, true, &origin_name))
573 : 0 : appendStringInfo(&err_detail, _("The row to be updated was deleted by a different origin \"%s\" in transaction %u at %s"),
574 : : origin_name, localxmin, timestamptz_to_str(localts));
575 : :
576 : : /* The origin that modified this row has been removed. */
577 : : else
578 : 0 : appendStringInfo(&err_detail, _("The row to be updated was deleted by a non-existent origin in transaction %u at %s"),
579 : : localxmin, timestamptz_to_str(localts));
580 : : }
581 : : else
582 : 0 : appendStringInfoString(&err_detail, _("The row to be updated was deleted"));
583 : :
584 : 3 : break;
585 : :
586 : 9 : case CT_UPDATE_MISSING:
587 : 9 : append_tuple_value_detail(&tuple_buf,
588 : : list_make2(remote_desc, search_desc));
589 : :
590 [ + - ]: 9 : if (tuple_buf.len)
591 : 9 : appendStringInfo(&err_detail, _("Could not find the row to be updated: %s."),
592 : : tuple_buf.data);
593 : : else
594 : 0 : appendStringInfo(&err_detail, _("Could not find the row to be updated."));
595 : :
596 : 9 : break;
597 : :
598 : 5 : case CT_DELETE_ORIGIN_DIFFERS:
599 : 5 : append_tuple_value_detail(&tuple_buf,
600 : : list_make3(local_desc, remote_desc,
601 : : search_desc));
602 : :
603 [ + + ]: 5 : if (localorigin == InvalidReplOriginId)
604 : : {
605 [ + - ]: 4 : if (tuple_buf.len)
606 : 4 : appendStringInfo(&err_detail, _("Deleting the row that was modified locally in transaction %u at %s: %s."),
607 : : localxmin, timestamptz_to_str(localts),
608 : : tuple_buf.data);
609 : : else
610 : 0 : appendStringInfo(&err_detail, _("Deleting the row that was modified locally in transaction %u at %s."),
611 : : localxmin, timestamptz_to_str(localts));
612 : : }
613 [ + - ]: 1 : else if (replorigin_by_oid(localorigin, true, &origin_name))
614 : : {
615 [ + - ]: 1 : if (tuple_buf.len)
616 : 1 : appendStringInfo(&err_detail, _("Deleting the row that was modified by a different origin \"%s\" in transaction %u at %s: %s."),
617 : : origin_name, localxmin,
618 : : timestamptz_to_str(localts),
619 : : tuple_buf.data);
620 : : else
621 : 0 : appendStringInfo(&err_detail, _("Deleting the row that was modified by a different origin \"%s\" in transaction %u at %s."),
622 : : origin_name, localxmin,
623 : : timestamptz_to_str(localts));
624 : : }
625 : :
626 : : /* The origin that modified this row has been removed. */
627 : : else
628 : : {
629 [ # # ]: 0 : if (tuple_buf.len)
630 : 0 : appendStringInfo(&err_detail, _("Deleting the row that was modified by a non-existent origin in transaction %u at %s: %s."),
631 : : localxmin, timestamptz_to_str(localts),
632 : : tuple_buf.data);
633 : : else
634 : 0 : appendStringInfo(&err_detail, _("Deleting the row that was modified by a non-existent origin in transaction %u at %s."),
635 : : localxmin, timestamptz_to_str(localts));
636 : : }
637 : :
638 : 5 : break;
639 : :
640 : 9 : case CT_DELETE_MISSING:
641 : 9 : append_tuple_value_detail(&tuple_buf,
642 : : list_make1(search_desc));
643 : :
644 [ + - ]: 9 : if (tuple_buf.len)
645 : 9 : appendStringInfo(&err_detail, _("Could not find the row to be deleted: %s."),
646 : : tuple_buf.data);
647 : : else
648 : 0 : appendStringInfo(&err_detail, _("Could not find the row to be deleted."));
649 : :
650 : 9 : break;
651 : : }
652 : :
653 : : Assert(err_detail.len > 0);
654 : :
655 : : /*
656 : : * Insert a blank line to visually separate the new detail line from the
657 : : * existing ones.
658 : : */
659 [ + + ]: 121 : if (err_msg->len > 0)
660 : 44 : appendStringInfoChar(err_msg, '\n');
661 : :
662 : 121 : appendStringInfoString(err_msg, err_detail.data);
663 : 121 : }
664 : :
665 : : /*
666 : : * Extract conflicting key, local row, remote row, and replica identity
667 : : * columns. Results are set at xxx_desc.
668 : : *
669 : : * If the output is NULL, it indicates that the current user lacks permissions
670 : : * to view the columns involved.
671 : : */
672 : : static void
673 : 121 : get_tuple_desc(EState *estate, ResultRelInfo *relinfo, ConflictType type,
674 : : char **key_desc,
675 : : TupleTableSlot *localslot, char **local_desc,
676 : : TupleTableSlot *remoteslot, char **remote_desc,
677 : : TupleTableSlot *searchslot, char **search_desc,
678 : : Oid indexoid)
679 : : {
680 : 121 : Relation localrel = relinfo->ri_RelationDesc;
681 : 121 : Oid relid = RelationGetRelid(localrel);
682 : 121 : TupleDesc tupdesc = RelationGetDescr(localrel);
683 : 121 : char *desc = NULL;
684 : :
685 : : Assert((localslot && local_desc) || (remoteslot && remote_desc) ||
686 : : (searchslot && search_desc));
687 : :
688 : : /*
689 : : * Report the conflicting key values in the case of a unique constraint
690 : : * violation.
691 : : */
692 [ + + + + : 121 : if (type == CT_INSERT_EXISTS || type == CT_UPDATE_EXISTS ||
+ + ]
693 : : type == CT_MULTIPLE_UNIQUE_CONFLICTS)
694 : : {
695 : : Assert(OidIsValid(indexoid) && localslot);
696 : :
697 : 92 : desc = build_index_value_desc(estate, localrel, localslot,
698 : : indexoid);
699 : :
700 [ + - ]: 92 : if (desc)
701 : 92 : *key_desc = psprintf(_("key %s"), desc);
702 : : }
703 : :
704 [ + + ]: 121 : if (localslot)
705 : : {
706 : : /*
707 : : * The 'modifiedCols' only applies to the new tuple, hence we pass
708 : : * NULL for the local row.
709 : : */
710 : 100 : desc = ExecBuildSlotValueDescription(relid, localslot, tupdesc,
711 : : NULL, 64);
712 : :
713 [ + - ]: 100 : if (desc)
714 : 100 : *local_desc = psprintf(_("local row %s"), desc);
715 : : }
716 : :
717 [ + + ]: 121 : if (remoteslot)
718 : : {
719 : : Bitmapset *modifiedCols;
720 : :
721 : : /*
722 : : * Although logical replication doesn't maintain the bitmap for the
723 : : * columns being inserted, we still use it to create 'modifiedCols'
724 : : * for consistency with other calls to ExecBuildSlotValueDescription.
725 : : *
726 : : * Note that generated columns are formed locally on the subscriber.
727 : : */
728 : 107 : modifiedCols = bms_union(ExecGetInsertedCols(relinfo, estate),
729 : 107 : ExecGetUpdatedCols(relinfo, estate));
730 : 107 : desc = ExecBuildSlotValueDescription(relid, remoteslot,
731 : : tupdesc, modifiedCols,
732 : : 64);
733 : :
734 [ + - ]: 107 : if (desc)
735 : 107 : *remote_desc = psprintf(_("remote row %s"), desc);
736 : : }
737 : :
738 [ + + ]: 121 : if (searchslot)
739 : : {
740 : : /*
741 : : * Note that while index other than replica identity may be used (see
742 : : * IsIndexUsableForReplicaIdentityFull for details) to find the tuple
743 : : * when applying update or delete, such an index scan may not result
744 : : * in a unique tuple and we still compare the complete tuple in such
745 : : * cases, thus such indexes are not used here.
746 : : */
747 : 33 : Oid replica_index = GetRelationIdentityOrPK(localrel);
748 : :
749 : : Assert(type != CT_INSERT_EXISTS);
750 : :
751 : : /*
752 : : * If the table has a valid replica identity index, build the index
753 : : * key value string. Otherwise, construct the full tuple value for
754 : : * REPLICA IDENTITY FULL cases.
755 : : */
756 [ + + ]: 33 : if (OidIsValid(replica_index))
757 : 29 : desc = build_index_value_desc(estate, localrel, searchslot, replica_index);
758 : : else
759 : 4 : desc = ExecBuildSlotValueDescription(relid, searchslot, tupdesc, NULL, 64);
760 : :
761 [ + - ]: 33 : if (desc)
762 : : {
763 [ + + ]: 33 : if (OidIsValid(replica_index))
764 : 29 : *search_desc = psprintf(_("replica identity %s"), desc);
765 : : else
766 : 4 : *search_desc = psprintf(_("replica identity full %s"), desc);
767 : : }
768 : : }
769 : 121 : }
770 : :
771 : : /*
772 : : * Helper functions to construct a string describing the contents of an index
773 : : * entry. See BuildIndexValueDescription for details.
774 : : *
775 : : * The caller must ensure that the index with the OID 'indexoid' is locked so
776 : : * that we can fetch and display the conflicting key value.
777 : : */
778 : : static char *
779 : 121 : build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot,
780 : : Oid indexoid)
781 : : {
782 : : char *index_value;
783 : : Relation indexDesc;
784 : : Datum values[INDEX_MAX_KEYS];
785 : : bool isnull[INDEX_MAX_KEYS];
786 : 121 : TupleTableSlot *tableslot = slot;
787 : :
788 [ - + ]: 121 : if (!tableslot)
789 : 0 : return NULL;
790 : :
791 : : Assert(CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
792 : :
793 : 121 : indexDesc = index_open(indexoid, NoLock);
794 : :
795 : : /*
796 : : * If the slot is a virtual slot, copy it into a heap tuple slot as
797 : : * FormIndexDatum only works with heap tuple slots.
798 : : */
799 [ + + ]: 121 : if (TTS_IS_VIRTUAL(slot))
800 : : {
801 : 20 : tableslot = table_slot_create(localrel, &estate->es_tupleTable);
802 : 20 : tableslot = ExecCopySlot(tableslot, slot);
803 : : }
804 : :
805 : : /*
806 : : * Initialize ecxt_scantuple for potential use in FormIndexDatum when
807 : : * index expressions are present.
808 : : */
809 [ + - ]: 121 : GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
810 : :
811 : : /*
812 : : * The values/nulls arrays passed to BuildIndexValueDescription should be
813 : : * the results of FormIndexDatum, which are the "raw" input to the index
814 : : * AM.
815 : : */
816 : 121 : FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull);
817 : :
818 : 121 : index_value = BuildIndexValueDescription(indexDesc, values, isnull);
819 : :
820 : 121 : index_close(indexDesc, NoLock);
821 : :
822 : 121 : return index_value;
823 : : }
|
