From 88239bd809783173832bc98227007fb7ebb2b902 Mon Sep 17 00:00:00 2001 From: Nisha Moond Date: Tue, 18 Mar 2025 12:27:48 +0530 Subject: [PATCH v6] Implement the conflict detection for multiple_unique_conflicts in logical replication Introduce a new conflict type, multiple_unique_conflicts, to handle cases where an incoming row during logical replication violates multiple UNIQUE constraints. Previously, the apply worker detected and reported only the first encountered key conflict (insert_exists/update_exists), causing repeated failures as each constraint violation need to be handled one by one making the process slow and error-prone. Now, the apply worker checks all unique constraints upfront and reports multiple_unique_conflicts if multiple violations exist. This allows users to resolve all conflicts at once by deleting all conflicting tuples rather than dealing with them individually or skipping the transaction. The CONFLICT_NUM_TYPES is not incremented since subscription stats do not support multiple_unique_conflicts in this patch. --- doc/src/sgml/logical-replication.sgml | 13 ++ src/backend/executor/execReplication.c | 34 +++++- src/backend/replication/logical/conflict.c | 93 ++++++++++---- src/backend/replication/logical/worker.c | 36 +++--- src/include/replication/conflict.h | 10 +- src/test/subscription/meson.build | 1 + src/test/subscription/t/035_conflicts.pl | 133 +++++++++++++++++++++ 7 files changed, 270 insertions(+), 50 deletions(-) create mode 100644 src/test/subscription/t/035_conflicts.pl diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 3d18e507bbc..4817206af7d 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -1877,6 +1877,19 @@ test_sub=# SELECT * from tab_gen_to_gen; + + multiple_unique_conflicts + + + Inserting a row or updating values of a row violates more than one + NOT DEFERRABLE unique constraint. Note that to log + the origin and commit timestamp details of the conflicting key, + track_commit_timestamp + should be enabled on the subscriber. In this case, an error will be + raised until the conflict is resolved manually. + + + Note that there are other conflict scenarios, such as exclusion constraint violations. Currently, we do not provide additional details for them in the diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index 0a9b880d250..efd93cda69a 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -493,11 +493,17 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate, ConflictType type, List *recheckIndexes, TupleTableSlot *searchslot, TupleTableSlot *remoteslot) { - /* Check all the unique indexes for a conflict */ + int conflicts = 0; + List *conflictSlots = NIL; + List *conflictIndexes = NIL; + List *localxmins = NIL; + List *localorigins = NIL; + List *localts = NIL; + TupleTableSlot *conflictslot; + + /* Check all the unique indexes for conflicts */ foreach_oid(uniqueidx, resultRelInfo->ri_onConflictArbiterIndexes) { - TupleTableSlot *conflictslot; - if (list_member_oid(recheckIndexes, uniqueidx) && FindConflictTuple(resultRelInfo, estate, uniqueidx, remoteslot, &conflictslot)) @@ -507,11 +513,27 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate, TransactionId xmin; GetTupleTransactionInfo(conflictslot, &xmin, &origin, &committs); - ReportApplyConflict(estate, resultRelInfo, ERROR, type, - searchslot, conflictslot, remoteslot, - uniqueidx, xmin, origin, committs); + + /* + * Add the conflict slot, index, and the transaction info to the + * respective lists. + */ + conflictSlots = lappend(conflictSlots, conflictslot); + conflictIndexes = lappend_oid(conflictIndexes, uniqueidx); + localxmins = lappend_xid(localxmins, xmin); + localorigins = lappend_int(localorigins, origin); + localts = lappend(localts, DatumGetPointer(Int64GetDatum(committs))); + + conflicts++; } } + + /* Report the conflict if found */ + if (conflicts) + ReportApplyConflict(estate, resultRelInfo, ERROR, + conflicts > 1 ? CT_MULTIPLE_UNIQUE_CONFLICTS : type, + searchslot, conflictSlots, remoteslot, + conflictIndexes, localxmins, localorigins, localts); } /* diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index 772fc83e88b..2074ef17321 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -29,11 +29,12 @@ static const char *const ConflictTypeNames[] = { [CT_UPDATE_EXISTS] = "update_exists", [CT_UPDATE_MISSING] = "update_missing", [CT_DELETE_ORIGIN_DIFFERS] = "delete_origin_differs", - [CT_DELETE_MISSING] = "delete_missing" + [CT_DELETE_MISSING] = "delete_missing", + [CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts" }; static int errcode_apply_conflict(ConflictType type); -static int errdetail_apply_conflict(EState *estate, +static void errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo, ConflictType type, TupleTableSlot *searchslot, @@ -41,7 +42,7 @@ static int errdetail_apply_conflict(EState *estate, TupleTableSlot *remoteslot, Oid indexoid, TransactionId localxmin, RepOriginId localorigin, - TimestampTz localts); + TimestampTz localts, StringInfo err_msg); static char *build_tuple_value_details(EState *estate, ResultRelInfo *relinfo, ConflictType type, TupleTableSlot *searchslot, @@ -90,15 +91,15 @@ GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin, * 'searchslot' should contain the tuple used to search the local tuple to be * updated or deleted. * - * 'localslot' should contain the existing local tuple, if any, that conflicts - * with the remote tuple. 'localxmin', 'localorigin', and 'localts' provide the - * transaction information related to this existing local tuple. + * 'conflictSlots' list contain the existing local tuples, if any, that + * conflicts with the remote tuple. 'localxmins', 'localorigins', and 'localts' + * provide the transaction information related to the existing local tuples. * * 'remoteslot' should contain the remote new tuple, if any. * - * The 'indexoid' represents the OID of the unique index that triggered the - * constraint violation error. We use this to report the key values for - * conflicting tuple. + * The 'conflictIndexes' list represents the OIDs of the unique index that + * triggered the constraint violation error. We use this to report the key + * values for conflicting tuple. * * The caller must ensure that the index with the OID 'indexoid' is locked so * that we can fetch and display the conflicting key value. @@ -106,16 +107,62 @@ GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin, void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, ConflictType type, TupleTableSlot *searchslot, - TupleTableSlot *localslot, TupleTableSlot *remoteslot, - Oid indexoid, TransactionId localxmin, - RepOriginId localorigin, TimestampTz localts) + List *conflictSlots, TupleTableSlot *remoteslot, + List *conflictIndexes, List *localxmins, + List *localorigins, List *localts) { + int conflictNum = 0; Relation localrel = relinfo->ri_RelationDesc; + ListCell *slotlc = list_head(conflictSlots); + StringInfoData err_detail; + + initStringInfo(&err_detail); + + do + { + Oid indexoid = InvalidOid; + TimestampTz committs = 0; + RepOriginId origin = InvalidRepOriginId; + TransactionId xmin = InvalidTransactionId; + TupleTableSlot *slot = NULL; + + if (slotlc) + { + Assert(localxmins && localorigins && localts); + + slot = lfirst(slotlc); + origin = list_nth_int(localorigins, conflictNum); + xmin = lfirst_xid(list_nth_cell(localxmins, conflictNum)); + committs = (TimestampTz) lfirst(list_nth_cell(localts, conflictNum)); + + slotlc = lnext(conflictSlots, slotlc); + } + + if (conflictIndexes) + indexoid = list_nth_oid(conflictIndexes, conflictNum); + + Assert(!OidIsValid(indexoid) || + CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true)); - Assert(!OidIsValid(indexoid) || - CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true)); + /* + * Build the error detail message containing the conflicting key and + * tuple information. The details for each conflict will be appended + * to err_detail. + */ + errdetail_apply_conflict(estate, relinfo, type, searchslot, + slot, remoteslot, indexoid, + xmin, origin, committs, &err_detail); - pgstat_report_subscription_conflict(MySubscription->oid, type); + conflictNum++; + + } while (slotlc); + + /* Conflict stats are not gathered for multiple_unique_conflicts */ + if (type != CT_MULTIPLE_UNIQUE_CONFLICTS) + pgstat_report_subscription_conflict(MySubscription->oid, type); + + /* Remove the extra newline at the end of err_detail */ + err_detail.data[err_detail.len - 1] = '\0'; ereport(elevel, errcode_apply_conflict(type), @@ -123,9 +170,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, get_namespace_name(RelationGetNamespace(localrel)), RelationGetRelationName(localrel), ConflictTypeNames[type]), - errdetail_apply_conflict(estate, relinfo, type, searchslot, - localslot, remoteslot, indexoid, - localxmin, localorigin, localts)); + errdetail_internal("%s", err_detail.data)); } /* @@ -169,6 +214,7 @@ errcode_apply_conflict(ConflictType type) { case CT_INSERT_EXISTS: case CT_UPDATE_EXISTS: + case CT_MULTIPLE_UNIQUE_CONFLICTS: return errcode(ERRCODE_UNIQUE_VIOLATION); case CT_UPDATE_ORIGIN_DIFFERS: case CT_UPDATE_MISSING: @@ -191,12 +237,13 @@ errcode_apply_conflict(ConflictType type) * replica identity columns, if any. The remote old tuple is excluded as its * information is covered in the replica identity columns. */ -static int +static void errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo, ConflictType type, TupleTableSlot *searchslot, TupleTableSlot *localslot, TupleTableSlot *remoteslot, Oid indexoid, TransactionId localxmin, - RepOriginId localorigin, TimestampTz localts) + RepOriginId localorigin, TimestampTz localts, + StringInfo err_msg) { StringInfoData err_detail; char *val_desc; @@ -209,6 +256,7 @@ errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo, { case CT_INSERT_EXISTS: case CT_UPDATE_EXISTS: + case CT_MULTIPLE_UNIQUE_CONFLICTS: Assert(OidIsValid(indexoid)); if (localts) @@ -291,7 +339,7 @@ errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo, if (val_desc) appendStringInfo(&err_detail, "\n%s", val_desc); - return errdetail_internal("%s", err_detail.data); + appendStringInfo(err_msg, "%s\n", err_detail.data); } /* @@ -323,7 +371,8 @@ build_tuple_value_details(EState *estate, ResultRelInfo *relinfo, * Report the conflicting key values in the case of a unique constraint * violation. */ - if (type == CT_INSERT_EXISTS || type == CT_UPDATE_EXISTS) + if (type == CT_INSERT_EXISTS || type == CT_UPDATE_EXISTS || + type == CT_MULTIPLE_UNIQUE_CONFLICTS) { Assert(OidIsValid(indexoid) && localslot); diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 31ab69ea13a..4a595e7906f 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -2711,8 +2711,10 @@ apply_handle_update_internal(ApplyExecutionData *edata, slot_store_data(newslot, relmapentry, newtup); ReportApplyConflict(estate, relinfo, LOG, CT_UPDATE_ORIGIN_DIFFERS, - remoteslot, localslot, newslot, - InvalidOid, localxmin, localorigin, localts); + remoteslot, list_make1(localslot), newslot, + NULL, list_make1_xid(localxmin), + list_make1_int(localorigin), + list_make1(DatumGetPointer(Int64GetDatum(localts)))); } /* Process and store remote tuple in the slot */ @@ -2741,9 +2743,8 @@ apply_handle_update_internal(ApplyExecutionData *edata, * emitting a log message. */ ReportApplyConflict(estate, relinfo, LOG, CT_UPDATE_MISSING, - remoteslot, NULL, newslot, - InvalidOid, InvalidTransactionId, - InvalidRepOriginId, 0); + remoteslot, NULL, newslot, NULL, + NULL, NULL, NULL); } /* Cleanup. */ @@ -2887,8 +2888,10 @@ apply_handle_delete_internal(ApplyExecutionData *edata, if (GetTupleTransactionInfo(localslot, &localxmin, &localorigin, &localts) && localorigin != replorigin_session_origin) ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_ORIGIN_DIFFERS, - remoteslot, localslot, NULL, - InvalidOid, localxmin, localorigin, localts); + remoteslot, list_make1(localslot), NULL, + NULL, list_make1_xid(localxmin), + list_make1_int(localorigin), + list_make1(DatumGetPointer(Int64GetDatum(localts)))); EvalPlanQualSetSlot(&epqstate, localslot); @@ -2903,9 +2906,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata, * emitting a log message. */ ReportApplyConflict(estate, relinfo, LOG, CT_DELETE_MISSING, - remoteslot, NULL, NULL, - InvalidOid, InvalidTransactionId, - InvalidRepOriginId, 0); + remoteslot, NULL, NULL, NULL, NULL, NULL, NULL); } /* Cleanup. */ @@ -3093,11 +3094,9 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, * The tuple to be updated could not be found. Do nothing * except for emitting a log message. */ - ReportApplyConflict(estate, partrelinfo, - LOG, CT_UPDATE_MISSING, - remoteslot_part, NULL, newslot, - InvalidOid, InvalidTransactionId, - InvalidRepOriginId, 0); + ReportApplyConflict(estate, partrelinfo, LOG, + CT_UPDATE_MISSING, remoteslot_part, + NULL, newslot, NULL, NULL, NULL, NULL); return; } @@ -3116,9 +3115,10 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, slot_store_data(newslot, part_entry, newtup); ReportApplyConflict(estate, partrelinfo, LOG, CT_UPDATE_ORIGIN_DIFFERS, - remoteslot_part, localslot, newslot, - InvalidOid, localxmin, localorigin, - localts); + remoteslot_part, list_make1(localslot), + newslot, NULL, list_make1_xid(localxmin), + list_make1_int(localorigin), + list_make1(DatumGetPointer(Int64GetDatum(localts)))); } /* diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h index 37454dc9513..bba7f7156b5 100644 --- a/src/include/replication/conflict.h +++ b/src/include/replication/conflict.h @@ -41,6 +41,9 @@ typedef enum /* The row to be deleted is missing */ CT_DELETE_MISSING, + /* The row to be inserted/updated violates multiple unique constraint */ + CT_MULTIPLE_UNIQUE_CONFLICTS, + /* * Other conflicts, such as exclusion constraint violations, involve more * complex rules than simple equality checks. These conflicts are left for @@ -57,10 +60,9 @@ extern bool GetTupleTransactionInfo(TupleTableSlot *localslot, extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, ConflictType type, TupleTableSlot *searchslot, - TupleTableSlot *localslot, + List *conflictSlots, TupleTableSlot *remoteslot, - Oid indexoid, TransactionId localxmin, - RepOriginId localorigin, TimestampTz localts); + List *conflictIndexes, List *localxmin, + List *localorigin, List *localts); extern void InitConflictIndexes(ResultRelInfo *relInfo); - #endif diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index d40b49714f6..586ffba434e 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -41,6 +41,7 @@ tests += { 't/032_subscribe_use_index.pl', 't/033_run_as_table_owner.pl', 't/034_temporal.pl', + 't/035_conflicts.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl new file mode 100644 index 00000000000..f1417e313db --- /dev/null +++ b/src/test/subscription/t/035_conflicts.pl @@ -0,0 +1,133 @@ +# Copyright (c) 2025, PostgreSQL Global Development Group + +# Test the conflict detection of conflict type 'multiple_unique_conflicts'. +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +############################### +# Setup +############################### + +# Create a publisher node +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +# Create a subscriber node +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init; +$node_subscriber->start; + +# Create a table on publisher +$node_publisher->safe_psql('postgres', + "CREATE TABLE conf_tab (a int PRIMARY KEY, b int UNIQUE, c int UNIQUE);"); + +# Create same table on subscriber +$node_subscriber->safe_psql('postgres', + "CREATE TABLE conf_tab (a int PRIMARY key, b int unique, c int unique);"); + +# Setup logical replication +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION pub_tab FOR TABLE conf_tab"); + +# Create the subscription +my $appname = 'sub_tab'; +$node_subscriber->safe_psql( + 'postgres', + "CREATE SUBSCRIPTION sub_tab + CONNECTION '$publisher_connstr application_name=$appname' + PUBLICATION pub_tab;"); + +# Wait for initial table sync to finish +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname); + +################################################## +# INSERT data on Pub and Sub +################################################## + +# Insert data in the publisher table +$node_publisher->safe_psql('postgres', + "INSERT INTO conf_tab VALUES (1,1,1);"); + +# Insert data in the subscriber table +$node_subscriber->safe_psql('postgres', + "INSERT INTO conf_tab VALUES (2,2,2), (3,3,3), (4,4,4);"); + +################################################## +# Test multiple_unique_conflicts due to INSERT +################################################## +my $log_offset = -s $node_subscriber->logfile; + +$node_publisher->safe_psql('postgres', + "INSERT INTO conf_tab VALUES (2,3,4);"); + +# Confirm that this causes an error on the subscriber +$node_subscriber->wait_for_log( + qr/ERROR: conflict detected on relation \"public.conf_tab\": conflict=multiple_unique_conflicts/, + $log_offset); + +ok( $node_subscriber->log_contains( + qr/Key already exists in unique index \"conf_tab_pkey\".*\n.*Key \(a\)=\(2\); existing local tuple \(2, 2, 2\); remote tuple \(2, 3, 4\)./, + $log_offset), + 'multiple_unique_conflicts detected during insertion for conf_tab_pkey (a) = (2)' +); + +ok( $node_subscriber->log_contains( + qr/Key already exists in unique index \"conf_tab_b_key\".*\n.*Key \(b\)=\(3\); existing local tuple \(3, 3, 3\); remote tuple \(2, 3, 4\)./, + $log_offset), + 'multiple_unique_conflicts detected during insertion for conf_tab_b_key (b) = (3)' +); + +ok( $node_subscriber->log_contains( + qr/Key already exists in unique index \"conf_tab_c_key\".*\n.*Key \(c\)=\(4\); existing local tuple \(4, 4, 4\); remote tuple \(2, 3, 4\)./, + $log_offset), + 'multiple_unique_conflicts detected during insertion for conf_tab_c_key (c) = (4)' +); + +# Truncate table to get rid of the error +$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;"); + +################################################## +# Test multiple_unique_conflicts due to UPDATE +################################################## +$log_offset = -s $node_subscriber->logfile; + +# Insert data in the publisher table +$node_publisher->safe_psql('postgres', + "INSERT INTO conf_tab VALUES (5,5,5);"); + +# Insert data in the subscriber table +$node_subscriber->safe_psql('postgres', + "INSERT INTO conf_tab VALUES (6,6,6), (7,7,7), (8,8,8);"); + +$node_publisher->safe_psql('postgres', + "UPDATE conf_tab set a=6, b=7, c=8 where a=5;"); + +# Confirm that this causes an error on the subscriber +$node_subscriber->wait_for_log( + qr/ERROR: conflict detected on relation \"public.conf_tab\": conflict=multiple_unique_conflicts/, + $log_offset); + +ok( $node_subscriber->log_contains( + qr/Key already exists in unique index \"conf_tab_pkey\".*\n.*Key \(a\)=\(6\); existing local tuple \(6, 6, 6\); remote tuple \(6, 7, 8\)./, + $log_offset), + 'multiple_unique_conflicts detected during update for conf_tab_pkey (a) = (6)' +); + +ok( $node_subscriber->log_contains( + qr/Key already exists in unique index \"conf_tab_b_key\".*\n.*Key \(b\)=\(7\); existing local tuple \(7, 7, 7\); remote tuple \(6, 7, 8\)./, + $log_offset), + 'multiple_unique_conflicts detected during update for conf_tab_b_key (b) = (7)' +); + +ok( $node_subscriber->log_contains( + qr/Key already exists in unique index \"conf_tab_c_key\".*\n.*Key \(c\)=\(8\); existing local tuple \(8, 8, 8\); remote tuple \(6, 7, 8\)./, + $log_offset), + 'multiple_unique_conflicts detected during update for conf_tab_c_key (c) = (8)' +); + +done_testing(); -- 2.34.1