From 04e333b30b5d599701a511029cab10a73985f5b7 Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Wed, 5 Aug 2026 15:57:13 +0530 Subject: [PATCH v66 3/3] Return unique-key conflicts to the apply worker. Return unique-key conflicts to apply_handle_insert_internal()/ apply_handle_update_internal() instead of reporting them in ExecSimpleRelationInsert()/ExecSimpleRelationUpdate(), so a future conflict resolution strategy has a place to act while the apply transaction is still open. --- src/backend/executor/execReplication.c | 126 +++++++++++++++-------- src/backend/replication/logical/worker.c | 50 ++++++++- src/include/executor/executor.h | 11 +- src/include/replication/conflict.h | 17 +++ 4 files changed, 150 insertions(+), 54 deletions(-) diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index b2ca5cbf117..31b397f2c6b 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -764,12 +764,16 @@ retry: /* * Check all the unique indexes in 'recheckIndexes' for conflict with the - * tuple in 'remoteslot' and report if found. + * tuple in 'remoteslot'. If a conflict is found, fill '*conflict' with the + * details and return true; the caller is responsible for reporting it (via + * ReportApplyConflict()) since only the caller knows the subscription's + * conflict resolution strategy. */ -static void -CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate, - ConflictType type, List *recheckIndexes, - TupleTableSlot *searchslot, TupleTableSlot *remoteslot) +static bool +CheckForConflict(ResultRelInfo *resultRelInfo, EState *estate, + ConflictType type, List *recheckIndexes, + TupleTableSlot *searchslot, TupleTableSlot *remoteslot, + ApplyConflictInfo *conflict) { List *conflicttuples = NIL; TupleTableSlot *conflictslot; @@ -793,11 +797,16 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate, } } - /* Report the conflict, if found */ - if (conflicttuples) - ReportApplyConflict(estate, resultRelInfo, ERROR, - list_length(conflicttuples) > 1 ? CT_MULTIPLE_UNIQUE_CONFLICTS : type, - searchslot, remoteslot, conflicttuples); + if (!conflicttuples) + return false; + + conflict->type = list_length(conflicttuples) > 1 ? + CT_MULTIPLE_UNIQUE_CONFLICTS : type; + conflict->searchslot = searchslot; + conflict->remoteslot = remoteslot; + conflict->conflicttuples = conflicttuples; + + return true; } /* @@ -805,12 +814,22 @@ CheckAndReportConflict(ResultRelInfo *resultRelInfo, EState *estate, * and execute any constraints and per-row triggers. * * Caller is responsible for opening the indexes. + * + * If a unique index conflict is detected, the tuple and index entries are + * still left in place (see the XXX below), '*conflict' is filled in with the + * conflict details, and true is returned -- the AFTER ROW trigger is + * skipped in that case, since the row is not really going to survive. The + * caller is responsible for calling ReportApplyConflict() on '*conflict'; + * this function does not do so itself, since only the caller knows how the + * subscription is configured to handle the conflict. */ -void +bool ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, - EState *estate, TupleTableSlot *slot) + EState *estate, TupleTableSlot *slot, + ApplyConflictInfo *conflict) { bool skip_tuple = false; + bool found_conflict = false; Relation rel = resultRelInfo->ri_RelationDesc; /* For now we support only tables. */ @@ -830,7 +849,7 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, { List *recheckIndexes = NIL; List *conflictindexes; - bool conflict = false; + bool index_conflict = false; /* Compute stored generated columns */ if (rel->rd_att->constr && @@ -860,40 +879,46 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, recheckIndexes = ExecInsertIndexTuples(resultRelInfo, estate, flags, slot, conflictindexes, - &conflict); + &index_conflict); } /* * Checks the conflict indexes to fetch the conflicting local row and - * reports the conflict. We perform this check here, instead of - * performing an additional index scan before the actual insertion and - * reporting the conflict if any conflicting rows are found. This is - * to avoid the overhead of executing the extra scan for each INSERT - * operation, even when no conflict arises, which could introduce - * significant overhead to replication, particularly in cases where - * conflicts are rare. + * hand the details back to the caller. We perform this check here, + * instead of performing an additional index scan before the actual + * insertion and reporting the conflict if any conflicting rows are + * found. This is to avoid the overhead of executing the extra scan + * for each INSERT operation, even when no conflict arises, which + * could introduce significant overhead to replication, particularly + * in cases where conflicts are rare. * * XXX OTOH, this could lead to clean-up effort for dead tuples added * in heap and index in case of conflicts. But as conflicts shouldn't * be a frequent thing so we preferred to save the performance * overhead of extra scan before each insertion. */ - if (conflict) - CheckAndReportConflict(resultRelInfo, estate, CT_INSERT_EXISTS, - recheckIndexes, NULL, slot); + if (index_conflict) + found_conflict = CheckForConflict(resultRelInfo, estate, + CT_INSERT_EXISTS, recheckIndexes, + NULL, slot, conflict); - /* AFTER ROW INSERT Triggers */ - ExecARInsertTriggers(estate, resultRelInfo, slot, - recheckIndexes, NULL); + if (!found_conflict) + { + /* AFTER ROW INSERT Triggers */ + ExecARInsertTriggers(estate, resultRelInfo, slot, + recheckIndexes, NULL); - /* - * XXX we should in theory pass a TransitionCaptureState object to the - * above to capture transition tuples, but after statement triggers - * don't actually get fired by replication yet anyway - */ + /* + * XXX we should in theory pass a TransitionCaptureState object to + * the above to capture transition tuples, but after statement + * triggers don't actually get fired by replication yet anyway + */ + } list_free(recheckIndexes); } + + return found_conflict; } /* @@ -901,13 +926,18 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, * update the indexes, and execute any constraints and per-row triggers. * * Caller is responsible for opening the indexes. + * + * See ExecSimpleRelationInsert() for the meaning of the return value and + * '*conflict'. */ -void +bool ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, - TupleTableSlot *searchslot, TupleTableSlot *slot) + TupleTableSlot *searchslot, TupleTableSlot *slot, + ApplyConflictInfo *conflict) { bool skip_tuple = false; + bool found_conflict = false; Relation rel = resultRelInfo->ri_RelationDesc; ItemPointer tid = &(searchslot->tts_tid); @@ -934,7 +964,7 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, List *recheckIndexes = NIL; TU_UpdateIndexes update_indexes; List *conflictindexes; - bool conflict = false; + bool index_conflict = false; /* Compute stored generated columns */ if (rel->rd_att->constr && @@ -964,26 +994,32 @@ ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, recheckIndexes = ExecInsertIndexTuples(resultRelInfo, estate, flags, slot, conflictindexes, - &conflict); + &index_conflict); } /* - * Refer to the comments above the call to CheckAndReportConflict() in + * Refer to the comments above the call to CheckForConflict() in * ExecSimpleRelationInsert to understand why this check is done at * this point. */ - if (conflict) - CheckAndReportConflict(resultRelInfo, estate, CT_UPDATE_EXISTS, - recheckIndexes, searchslot, slot); + if (index_conflict) + found_conflict = CheckForConflict(resultRelInfo, estate, + CT_UPDATE_EXISTS, recheckIndexes, + searchslot, slot, conflict); - /* AFTER ROW UPDATE Triggers */ - ExecARUpdateTriggers(estate, resultRelInfo, - NULL, NULL, - tid, NULL, slot, - recheckIndexes, NULL, false); + if (!found_conflict) + { + /* AFTER ROW UPDATE Triggers */ + ExecARUpdateTriggers(estate, resultRelInfo, + NULL, NULL, + tid, NULL, slot, + recheckIndexes, NULL, false); + } list_free(recheckIndexes); } + + return found_conflict; } /* diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index a43535e7dda..30f6ea98e29 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -2756,6 +2756,7 @@ apply_handle_insert_internal(ApplyExecutionData *edata, TupleTableSlot *remoteslot) { EState *estate = edata->estate; + ApplyConflictInfo conflict; /* Caller should have opened indexes already. */ Assert(relinfo->ri_IndexRelationDescs != NULL || @@ -2768,7 +2769,19 @@ apply_handle_insert_internal(ApplyExecutionData *edata, /* Do the insert. */ TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_INSERT); - ExecSimpleRelationInsert(relinfo, estate, remoteslot); + if (ExecSimpleRelationInsert(relinfo, estate, remoteslot, &conflict)) + { + /* + * This is where a future conflict resolution strategy could decide + * to resolve the conflict without raising an ERROR (e.g. skip the + * insert and log at LOG level). For now, always report at ERROR, + * matching the previous behavior. + */ + ReportApplyConflict(estate, relinfo, ERROR, conflict.type, + conflict.searchslot, conflict.remoteslot, + conflict.conflicttuples); + pg_unreachable(); + } } /* @@ -2951,6 +2964,7 @@ apply_handle_update_internal(ApplyExecutionData *edata, EPQState epqstate; TupleTableSlot *localslot = NULL; ConflictTupleInfo conflicttuple = {0}; + ApplyConflictInfo conflict; bool found; MemoryContext oldctx; @@ -3001,8 +3015,19 @@ apply_handle_update_internal(ApplyExecutionData *edata, /* Do the actual update. */ TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_UPDATE); - ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot, - remoteslot); + if (ExecSimpleRelationUpdate(relinfo, estate, &epqstate, localslot, + remoteslot, &conflict)) + { + /* + * This is where a future conflict resolution strategy could + * decide to resolve the conflict without raising an ERROR. For + * now, always report at ERROR, matching the previous behavior. + */ + ReportApplyConflict(estate, relinfo, ERROR, conflict.type, + conflict.searchslot, conflict.remoteslot, + conflict.conflicttuples); + pg_unreachable(); + } } else { @@ -3490,6 +3515,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, bool found; EPQState epqstate; ConflictTupleInfo conflicttuple = {0}; + ApplyConflictInfo conflict; /* Get the matching local tuple from the partition. */ found = FindReplTupleInLocalRel(edata, partrel, @@ -3584,8 +3610,22 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, EvalPlanQualSetSlot(&epqstate, remoteslot_part); TargetPrivilegesCheck(partrelinfo->ri_RelationDesc, ACL_UPDATE); - ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate, - localslot, remoteslot_part); + if (ExecSimpleRelationUpdate(partrelinfo, estate, &epqstate, + localslot, remoteslot_part, + &conflict)) + { + /* + * This is where a future conflict resolution + * strategy could decide to resolve the conflict + * without raising an ERROR. For now, always report + * at ERROR, matching the previous behavior. + */ + ReportApplyConflict(estate, partrelinfo, ERROR, + conflict.type, conflict.searchslot, + conflict.remoteslot, + conflict.conflicttuples); + pg_unreachable(); + } } else { diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index 1798e6027d4..1c09cf42f3e 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -20,6 +20,7 @@ #include "fmgr.h" #include "nodes/lockoptions.h" #include "nodes/parsenodes.h" +#include "replication/conflict.h" #include "utils/memutils.h" @@ -791,11 +792,13 @@ extern bool RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid, TransactionId *delete_xid, ReplOriginId *delete_origin, TimestampTz *delete_time); -extern void ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, - EState *estate, TupleTableSlot *slot); -extern void ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, +extern bool ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo, + EState *estate, TupleTableSlot *slot, + ApplyConflictInfo *conflict); +extern bool ExecSimpleRelationUpdate(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, - TupleTableSlot *searchslot, TupleTableSlot *slot); + TupleTableSlot *searchslot, TupleTableSlot *slot, + ApplyConflictInfo *conflict); extern void ExecSimpleRelationDelete(ResultRelInfo *resultRelInfo, EState *estate, EPQState *epqstate, TupleTableSlot *searchslot); diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h index 4db3a36d4df..458cfe21b81 100644 --- a/src/include/replication/conflict.h +++ b/src/include/replication/conflict.h @@ -80,6 +80,23 @@ typedef struct ConflictTupleInfo * conflicting local row occurred */ } ConflictTupleInfo; +/* + * Conflict detected by ExecSimpleRelationInsert()/ExecSimpleRelationUpdate() + * while rechecking unique indexes for a duplicate key. + * + * The executor functions fill this in and return control to their caller + * instead of invoking ReportApplyConflict() themselves, so that the apply + * worker call site -- which knows the subscription's conflict resolution + * strategy -- decides whether/how to report it. + */ +typedef struct ApplyConflictInfo +{ + ConflictType type; + TupleTableSlot *searchslot; + TupleTableSlot *remoteslot; + List *conflicttuples; +} ApplyConflictInfo; + /* * Defines where logical replication conflict details are recorded. * -- 2.54.0