From 04e333b30b5d599701a511029cab10a73985f5b7 Mon Sep 17 00:00:00 2001
From: Amit Kapila <akapila@postgresql.org>
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

