From 152ece6a4540981fb9e9e584af404709dc47e3fd Mon Sep 17 00:00:00 2001 From: Nisha Moond Date: Tue, 22 Oct 2024 11:01:41 +0530 Subject: [PATCH v4] 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. Also, the patch adds a new column 'confl_multiple_unique_conflicts' in view pg_stat_subscription_stats to support stats collection for this conflict type. --- doc/src/sgml/logical-replication.sgml | 13 ++ doc/src/sgml/monitoring.sgml | 12 ++ src/backend/catalog/system_views.sql | 1 + src/backend/executor/execReplication.c | 50 ++++-- src/backend/replication/logical/conflict.c | 94 +++++++++++- src/backend/utils/adt/pgstatfuncs.c | 6 +- src/include/catalog/pg_proc.dat | 6 +- src/include/replication/conflict.h | 11 +- src/test/regress/expected/rules.out | 3 +- src/test/subscription/meson.build | 1 + .../t/035_multiple_unique_conflicts.pl | 143 ++++++++++++++++++ 11 files changed, 318 insertions(+), 22 deletions(-) create mode 100644 src/test/subscription/t/035_multiple_unique_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/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index aaa6586d3a4..0960f5ba94a 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -2250,6 +2250,18 @@ description | Waiting for a newly initialized WAL file to reach durable storage + + + confl_multiple_unique_conflicts bigint + + + Number of times a row insertion or an updated row values violated multiple + NOT DEFERRABLE unique constraints during the + application of changes. See + for details about this conflict. + + + stats_reset timestamp with time zone diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index a4d2cfdcaf5..31d269b7ee0 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1384,6 +1384,7 @@ CREATE VIEW pg_stat_subscription_stats AS ss.confl_update_missing, ss.confl_delete_origin_differs, ss.confl_delete_missing, + ss.confl_multiple_unique_conflicts, ss.stats_reset FROM pg_subscription as s, pg_stat_get_subscription_stats(s.oid) as ss; diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index 0a9b880d250..2432db3a5f4 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -493,25 +493,55 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate, ConflictType type, List *recheckIndexes, TupleTableSlot *searchslot, TupleTableSlot *remoteslot) { + int conflicts = 0; + List *conflictSlots = NIL; + List *conflictIndexes = NIL; + TupleTableSlot *conflictslot; + /* Check all the unique indexes for a conflict */ foreach_oid(uniqueidx, resultRelInfo->ri_onConflictArbiterIndexes) { - TupleTableSlot *conflictslot; - if (list_member_oid(recheckIndexes, uniqueidx) && FindConflictTuple(resultRelInfo, estate, uniqueidx, remoteslot, &conflictslot)) { - RepOriginId origin; - TimestampTz committs; - TransactionId xmin; - - GetTupleTransactionInfo(conflictslot, &xmin, &origin, &committs); - ReportApplyConflict(estate, resultRelInfo, ERROR, type, - searchslot, conflictslot, remoteslot, - uniqueidx, xmin, origin, committs); + conflicts++; + + /* Add the conflict slot and index to their respective lists */ + conflictSlots = lappend(conflictSlots, conflictslot); + conflictIndexes = lappend_oid(conflictIndexes, uniqueidx); } } + + /* + * Report an INSERT_EXISTS or UPDATE_EXISTS conflict when only one unique + * constraint is violated. + */ + if (conflicts == 1) + { + Oid uniqueidx; + RepOriginId origin; + TimestampTz committs; + TransactionId xmin; + + uniqueidx = linitial_oid(conflictIndexes); + conflictslot = linitial(conflictSlots); + + GetTupleTransactionInfo(conflictslot, &xmin, &origin, &committs); + ReportApplyConflict(estate, resultRelInfo, ERROR, type, + searchslot, conflictslot, remoteslot, + uniqueidx, xmin, origin, committs); + } + + /* + * Report MULTIPLE_UNIQUE_CONFLICTS when two or more unique constraints + * are violated. + */ + else if (conflicts > 1) + ReportMultipleUniqueConflict(estate, resultRelInfo, ERROR, + CT_MULTIPLE_UNIQUE_CONFLICTS, + searchslot, remoteslot, + conflictSlots, conflictIndexes); } /* diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index 772fc83e88b..ae297d86374 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -29,7 +29,8 @@ 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); @@ -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, @@ -125,7 +126,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, ConflictTypeNames[type]), errdetail_apply_conflict(estate, relinfo, type, searchslot, localslot, remoteslot, indexoid, - localxmin, localorigin, localts)); + localxmin, localorigin, localts, NULL)); } /* @@ -169,6 +170,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: @@ -196,7 +198,8 @@ 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 +212,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,6 +295,17 @@ errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo, if (val_desc) appendStringInfo(&err_detail, "\n%s", val_desc); + /* + * If the caller provides a non-null 'err_msg' pointer, only the + * err_detail.data is requested. Append the constructed err_detail message + * to 'err_msg' and return. + */ + if (err_msg) + { + appendStringInfo(err_msg, "\n%s", err_detail.data); + return 0; + } + return errdetail_internal("%s", err_detail.data); } @@ -323,7 +338,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); @@ -489,3 +505,71 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot, return index_value; } + +/* + * Report a multiple_unique_conflicts while applying replication changes. + * + * 'searchslot' holds the tuple used to search the corresponding local + * tuple for update or deletion. + * + * 'remoteslot' contains the new tuple from the remote side, if available. + * + * 'conflictSlots' is a list of slots containing the local tuples + * that conflict with the remote tuple. + * + * The 'conflictIndexes' list stores the OIDs of the unique indexes that + * triggered the constraint violation. These indexes help identify the key + * values of the conflicting tuple. + * + * The caller must ensure that all indexes in 'conflictIndexes' are locked, + * allowing us to fetch and display the conflicting key values. + */ +void +ReportMultipleUniqueConflict(EState *estate, ResultRelInfo *relinfo, + int elevel, ConflictType type, + TupleTableSlot *searchslot, + TupleTableSlot *remoteslot, + List *conflictSlots, List *conflictIndexes) +{ + int conflictNum = 0; + Oid indexoid = linitial_oid(conflictIndexes); + Relation localrel = relinfo->ri_RelationDesc; + RepOriginId localorigin; + TimestampTz localts; + TransactionId localxmin; + StringInfoData err_detail; + + initStringInfo(&err_detail); + appendStringInfo(&err_detail, _("The remote tuple violates multiple unique constraints on the local table.")); + + foreach_ptr(TupleTableSlot, slot, conflictSlots) + { + indexoid = lfirst_oid(list_nth_cell(conflictIndexes, conflictNum)); + + Assert(!OidIsValid(indexoid) || + CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true)); + + GetTupleTransactionInfo(slot, &localxmin, &localorigin, &localts); + + /* + * 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, + localxmin, localorigin, localts, &err_detail); + + conflictNum++; + } + + pgstat_report_subscription_conflict(MySubscription->oid, type); + + ereport(elevel, + errcode_apply_conflict(type), + errmsg("conflict detected on relation \"%s.%s\": conflict=%s", + get_namespace_name(RelationGetNamespace(localrel)), + RelationGetRelationName(localrel), + ConflictTypeNames[type]), + errdetail_internal("%s", err_detail.data)); +} diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c index 662ce46cbc2..97af7c6554f 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -2171,7 +2171,7 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS) Datum pg_stat_get_subscription_stats(PG_FUNCTION_ARGS) { -#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 10 +#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 11 Oid subid = PG_GETARG_OID(0); TupleDesc tupdesc; Datum values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS] = {0}; @@ -2203,7 +2203,9 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS) INT8OID, -1, 0); TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_delete_missing", INT8OID, -1, 0); - TupleDescInitEntry(tupdesc, (AttrNumber) 10, "stats_reset", + TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_multiple_unique_conflicts", + INT8OID, -1, 0); + TupleDescInitEntry(tupdesc, (AttrNumber) 11, "stats_reset", TIMESTAMPTZOID, -1, 0); BlessTupleDesc(tupdesc); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 42e427f8fe8..68bb2f92024 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -5644,9 +5644,9 @@ { oid => '6231', descr => 'statistics: information about subscription stats', proname => 'pg_stat_get_subscription_stats', provolatile => 's', proparallel => 'r', prorettype => 'record', proargtypes => 'oid', - proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}', - proargmodes => '{i,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{subid,subid,apply_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,stats_reset}', + proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}', + proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{subid,subid,apply_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,confl_multiple_unique_conflicts,stats_reset}', prosrc => 'pg_stat_get_subscription_stats' }, { oid => '6118', descr => 'statistics: information about subscription', proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f', diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h index 37454dc9513..29c3c085283 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 @@ -48,7 +51,7 @@ typedef enum */ } ConflictType; -#define CONFLICT_NUM_TYPES (CT_DELETE_MISSING + 1) +#define CONFLICT_NUM_TYPES (CT_MULTIPLE_UNIQUE_CONFLICTS + 1) extern bool GetTupleTransactionInfo(TupleTableSlot *localslot, TransactionId *xmin, @@ -63,4 +66,10 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, RepOriginId localorigin, TimestampTz localts); extern void InitConflictIndexes(ResultRelInfo *relInfo); +extern void ReportMultipleUniqueConflict(EState *estate, ResultRelInfo *relinfo, + int elevel, ConflictType type, + TupleTableSlot *searchslot, + TupleTableSlot *remoteslot, + List *conflictslots_list, + List *conflictIndexes); #endif diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 62f69ac20b2..47478969135 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2157,9 +2157,10 @@ pg_stat_subscription_stats| SELECT ss.subid, ss.confl_update_missing, ss.confl_delete_origin_differs, ss.confl_delete_missing, + ss.confl_multiple_unique_conflicts, ss.stats_reset FROM pg_subscription s, - LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, sync_error_count, confl_insert_exists, confl_update_origin_differs, confl_update_exists, confl_update_missing, confl_delete_origin_differs, confl_delete_missing, stats_reset); + LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, sync_error_count, confl_insert_exists, confl_update_origin_differs, confl_update_exists, confl_update_missing, confl_delete_origin_differs, confl_delete_missing, confl_multiple_unique_conflicts, stats_reset); pg_stat_sys_indexes| SELECT relid, indexrelid, schemaname, diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index d40b49714f6..05fcdd08f57 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_multiple_unique_conflicts.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/035_multiple_unique_conflicts.pl b/src/test/subscription/t/035_multiple_unique_conflicts.pl new file mode 100644 index 00000000000..1eb0bd022a1 --- /dev/null +++ b/src/test/subscription/t/035_multiple_unique_conflicts.pl @@ -0,0 +1,143 @@ +# 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/DETAIL: The remote tuple violates multiple unique constraints on the local table./, + $log_offset), + 'multiple_unique_conflicts detected during insertion'); + +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/DETAIL: The remote tuple violates multiple unique constraints on the local table./, + $log_offset), + 'multiple_unique_conflicts detected during update'); + +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