v46-0002-Implement-the-conflict-insertion-infrastructure-.patch

From e2e140de32b33e9e2bbd76392c486f1aa0d6980b Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumarb@google.com>
Date: Sat, 30 May 2026 13:35:04 +0530
Subject: [PATCH v46 2/5] Implement the conflict insertion infrastructure for
 the conflict log table

This patch introduces the core logic to populate the conflict log table whenever
a logical replication conflict is detected. It captures the remote transaction
details along with the corresponding local state at the time of the conflict.

Handling Multi-row Conflicts: A single remote tuple may conflict with multiple
local tuples (e.g., in the case of multiple_unique_conflicts). To handle this,
the infrastructure creates a single row in the conflict log table for each
remote tuple. The details of all conflicting local rows are aggregated into a
single JSON array in the local_conflicts column.

The JSON array uses the following structured format:
[ { "xid": "1001", "commit_ts": "2025-12-25 10:00:00+05:30", "origin": "node_1",
"key": {"id": 1}, "tuple": {"id": 1, "val": "old_data"} }, ... ]

Example of querying the structured conflict data:

SELECT remote_xid, relname, remote_origin, local_conflicts[1] ->> 'xid' AS local_xid,
       local_conflicts[1] ->> 'tuple' AS local_tuple
FROM pg_conflict.pg_conflict_log_16396;

 remote_xid | relname  | remote_origin | local_xid |     local_tuple
------------+----------+---------------+-----------+---------------------
        760 | test     | pg_16406      | 771       | {"a":1,"b":10}
        765 | conf_tab | pg_16406      | 775       | {"a":2,"b":2,"c":2}
---
 .../replication/logical/applyparallelworker.c |  37 +-
 src/backend/replication/logical/conflict.c    | 647 ++++++++++++++++--
 src/backend/replication/logical/launcher.c    |   1 +
 src/backend/replication/logical/worker.c      |  16 +-
 src/include/replication/conflict.h            |   3 +
 src/include/replication/worker_internal.h     |   7 +
 src/test/subscription/t/030_origin.pl         |   4 +-
 src/test/subscription/t/035_conflicts.pl      |  47 +-
 8 files changed, 713 insertions(+), 49 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 012d55e9d3d..d24c52d43e6 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -986,7 +986,42 @@ ParallelApplyWorkerMain(Datum main_arg)
 
 	set_apply_error_context_origin(originname);
 
-	LogicalParallelApplyLoop(mqh);
+	PG_TRY();
+	{
+		LogicalParallelApplyLoop(mqh);
+	}
+	PG_CATCH();
+	{
+		MemoryContext oldcontext;
+		ErrorData *edata;
+
+		/*
+		 * Copy the ErrorData before doing any further work. The error may
+		 * have been raised while running under ErrorContext, so switch to
+		 * a safe context (TopMemoryContext) to avoid assertions and ensure
+		 * the error data survives subsequent cleanup.
+		 */
+		oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+		edata = CopyErrorData();
+		MemoryContextSwitchTo(oldcontext);
+
+		FlushErrorState();
+
+		/*
+		 * Reset the origin state to prevent the advancement of origin
+		 * progress if we fail to apply. Otherwise, this will result in
+		 * transaction loss as that transaction won't be sent again by the
+		 * server.
+		 */
+		replorigin_xact_clear(true);
+
+		AbortOutOfAnyTransaction();
+		ProcessPendingConflictLogTuple();
+
+		/* Re-throw the original error. */
+		ReThrowError(edata);
+	}
+	PG_END_TRY();
 
 	/*
 	 * The parallel apply worker must not get here because the parallel apply
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 4a647277d1f..6bf3d6d5a44 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -16,17 +16,24 @@
 
 #include "access/commit_ts.h"
 #include "access/genam.h"
+#include "access/heapam.h"
 #include "access/tableam.h"
+#include "access/xact.h"
 #include "catalog/heap.h"
 #include "catalog/pg_am.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/toasting.h"
 #include "executor/executor.h"
+#include "funcapi.h"
 #include "pgstat.h"
 #include "replication/conflict.h"
 #include "replication/worker_internal.h"
 #include "storage/lmgr.h"
+#include "utils/array.h"
+#include "utils/builtins.h"
 #include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/pg_lsn.h"
 
 /*
  * String representations for the supported conflict logging destinations.
@@ -40,7 +47,6 @@ const char *const ConflictLogDestNames[] = {
 StaticAssertDecl(lengthof(ConflictLogDestNames) == CONFLICT_LOG_DEST_ALL + 1,
 				 "ConflictLogDestNames length mismatch");
 
-
 /* Structure to hold metadata for one column of the conflict log table */
 typedef struct ConflictLogColumnDef
 {
@@ -72,6 +78,18 @@ static const ConflictLogColumnDef ConflictLogSchema[] = {
 
 #define NUM_CONFLICT_ATTRS lengthof(ConflictLogSchema)
 
+/* Schema for the elements within the 'local_conflicts' JSON array */
+static const ConflictLogColumnDef LocalConflictSchema[] =
+{
+	{ .attname = "xid",       .atttypid = XIDOID },
+	{ .attname = "commit_ts", .atttypid = TIMESTAMPTZOID },
+	{ .attname = "origin",    .atttypid = TEXTOID },
+	{ .attname = "key",       .atttypid = JSONOID },
+	{ .attname = "tuple",     .atttypid = JSONOID }
+};
+
+#define NUM_LOCAL_CONFLICT_ATTRS lengthof(LocalConflictSchema)
+
 static const char *const ConflictTypeNames[] = {
 	[CT_INSERT_EXISTS] = "insert_exists",
 	[CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
@@ -99,8 +117,27 @@ static void get_tuple_desc(EState *estate, ResultRelInfo *relinfo,
 						   TupleTableSlot *remoteslot, char **remote_desc,
 						   TupleTableSlot *searchslot, char **search_desc,
 						   Oid indexoid);
+static void build_index_datums_from_slot(EState *estate, Relation localrel,
+										 TupleTableSlot *slot,
+										 Relation indexDesc, Datum *values,
+										 bool *isnull);
 static char *build_index_value_desc(EState *estate, Relation localrel,
 									TupleTableSlot *slot, Oid indexoid);
+static Datum tuple_table_slot_to_json_datum(TupleTableSlot *slot);
+static Datum tuple_table_slot_to_indextup_json(EState *estate,
+											   Relation localrel,
+											   Oid replica_index,
+											   TupleTableSlot *slot);
+static TupleDesc build_conflict_tupledesc(void);
+static Datum build_local_conflicts_json_array(EState *estate, Relation rel,
+											  ConflictType conflict_type,
+											  List *conflicttuples);
+static void prepare_conflict_log_tuple(EState *estate, Relation rel,
+									   Relation conflictlogrel,
+									   ConflictType conflict_type,
+									   TupleTableSlot *searchslot,
+									   List *conflicttuples,
+									   TupleTableSlot *remoteslot);
 
 /*
  * Builds the TupleDesc for the conflict log table.
@@ -281,30 +318,170 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
 					ConflictType type, TupleTableSlot *searchslot,
 					TupleTableSlot *remoteslot, List *conflicttuples)
 {
-	Relation	localrel = relinfo->ri_RelationDesc;
-	StringInfoData err_detail;
+	Relation		localrel = relinfo->ri_RelationDesc;
+	ConflictLogDest	dest;
+	Relation		conflictlogrel;
+	bool			log_dest_table;
+	bool 			log_dest_logfile;
 
-	initStringInfo(&err_detail);
+	pgstat_report_subscription_conflict(MySubscription->oid, type);
 
-	/* Form errdetail message by combining conflicting tuples information. */
-	foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
-		errdetail_apply_conflict(estate, relinfo, type, searchslot,
-								 conflicttuple->slot, remoteslot,
-								 conflicttuple->indexoid,
-								 conflicttuple->xmin,
-								 conflicttuple->origin,
-								 conflicttuple->ts,
-								 &err_detail);
+	/*
+	 * Get the conflict log destination. Also, (if there is one) return the
+	 * CLT relation already opened and ready for insertion.
+	 */
+	conflictlogrel = GetConflictLogDestAndTable(&dest);
 
-	pgstat_report_subscription_conflict(MySubscription->oid, type);
+	log_dest_table = CONFLICTS_LOGGED_TO_TABLE(dest);
+	log_dest_logfile = CONFLICTS_LOGGED_TO_LOG(dest);
+
+	/* Insert to table if requested. */
+	if (log_dest_table)
+	{
+		Assert(conflictlogrel != NULL);
+
+		/*
+		 * Prepare the conflict log tuple. If the error level is below ERROR,
+		 * insert it immediately. Otherwise, defer the insertion to a new
+		 * transaction after the current one aborts, ensuring the insertion of
+		 * the log tuple is not rolled back.
+		 */
+		prepare_conflict_log_tuple(estate,
+								   relinfo->ri_RelationDesc,
+								   conflictlogrel,
+								   type,
+								   searchslot,
+								   conflicttuples,
+								   remoteslot);
+		if (elevel < ERROR)
+			InsertConflictLogTuple(conflictlogrel);
+
+		if (!log_dest_logfile)
+		{
+			/*
+			 * Not logging conflict details to the server log; Report the error
+			 * msg but omit raw tuple data from server logs since it's already
+			 * captured in the conflict log table.
+			 */
+			ereport(elevel,
+					errcode_apply_conflict(type),
+					errmsg("conflict detected on relation \"%s\": conflict=%s",
+						RelationGetQualifiedRelationName(localrel),
+						ConflictTypeNames[type]),
+					errdetail("Conflict details are logged to the conflict log table: %s",
+							  RelationGetRelationName(conflictlogrel)));
+		}
+
+		table_close(conflictlogrel, RowExclusiveLock);
+	}
+
+	/* Log into the server log if requested. */
+	if (log_dest_logfile)
+	{
+		StringInfoData	err_detail;
+
+		initStringInfo(&err_detail);
+
+		/* Form errdetail message by combining conflicting tuples information. */
+		foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
+			errdetail_apply_conflict(estate, relinfo, type, searchslot,
+									conflicttuple->slot, remoteslot,
+									conflicttuple->indexoid,
+									conflicttuple->xmin,
+									conflicttuple->origin,
+									conflicttuple->ts,
+									&err_detail);
+
+		/* Standard reporting with full internal details. */
+		ereport(elevel,
+				errcode_apply_conflict(type),
+				errmsg("conflict detected on relation \"%s\": conflict=%s",
+					   RelationGetQualifiedRelationName(localrel),
+					   ConflictTypeNames[type]),
+				errdetail_internal("%s", err_detail.data));
+	}
+}
+
+/*
+ * ProcessPendingConflictLogTuple
+ *      Insert any deferred conflict log tuple in a separate transaction.
+ *
+ * For conflicts raised at ERROR level, the conflict log tuple cannot be
+ * inserted immediately because the surrounding transaction will abort.
+ * To ensure that conflict information is not lost, such tuples are prepared
+ * during error processing (see prepare_conflict_log_tuple()) but their
+ * insertion is deferred.
+ *
+ * This function is responsible for completing that deferred insertion after
+ * the failing transaction has been aborted and the system has returned to an
+ * idle state.  It executes the insertion in a new, independent transaction,
+ * ensuring that the conflict log entry is durable and not rolled back
+ * together with the failed apply transaction.
+ */
+void
+ProcessPendingConflictLogTuple(void)
+{
+	Relation	conflictlogrel;
+	ConflictLogDest dest;
+
+	/* Nothing to do */
+	if (MyLogicalRepWorker->conflict_log_tuple == NULL)
+		return;
+
+	PG_TRY();
+	{
+		StartTransactionCommand();
+		PushActiveSnapshot(GetTransactionSnapshot());
+
+		/* Open conflict log table and insert the tuple */
+		conflictlogrel = GetConflictLogDestAndTable(&dest);
+		Assert(conflictlogrel);
+
+		InsertConflictLogTuple(conflictlogrel);
+
+		table_close(conflictlogrel, RowExclusiveLock);
+
+		PopActiveSnapshot();
+		CommitTransactionCommand();
+	}
+	PG_CATCH();
+	{
+		ErrorData  *edata;
+		MemoryContext oldctx;
+
+		/* Save error info in our memory context */
+		oldctx = MemoryContextSwitchTo(TopMemoryContext);
+		edata = CopyErrorData();
+		MemoryContextSwitchTo(oldctx);
+
+		/* Clear the error state so we can continue */
+		FlushErrorState();
 
-	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));
+		/* Abort the transaction we started above */
+		AbortOutOfAnyTransaction();
+
+		/*
+		 * Report the error as a warning. We use WARNING because we don't want
+		 * this to be a fatal error for the worker, and we want to allow the
+		 * caller's original error to remain primary.
+		 */
+		ereport(WARNING,
+				(errmsg("could not log conflict to table for subscription \"%s\": %s",
+						MySubscription->name, edata->message)));
+
+		FreeErrorData(edata);
+
+		/*
+		 * Free the conflict log tuple and set it to NULL. This ensures we
+		 * don't try to insert the same problematic tuple again.
+		 */
+		if (MyLogicalRepWorker->conflict_log_tuple != NULL)
+		{
+			heap_freetuple(MyLogicalRepWorker->conflict_log_tuple);
+			MyLogicalRepWorker->conflict_log_tuple = NULL;
+		}
+	}
+	PG_END_TRY();
 }
 
 /*
@@ -338,6 +515,56 @@ InitConflictIndexes(ResultRelInfo *relInfo)
 	relInfo->ri_onConflictArbiterIndexes = uniqueIndexes;
 }
 
+/*
+ * GetConflictLogDestAndTable
+ *
+ * Fetches conflict logging metadata from the cached MySubscription pointer.
+ * Sets the destination enum in *log_dest and, if applicable, opens and
+ * returns the relation handle for the conflict log table.
+ */
+Relation
+GetConflictLogDestAndTable(ConflictLogDest *log_dest)
+{
+	Oid			conflictlogrelid;
+
+	/*
+	 * Convert the text log destination to the internal enum.  MySubscription
+	 * already contains the data from pg_subscription.
+	 */
+	*log_dest = GetConflictLogDest(MySubscription->conflictlogdest);
+
+	/* Quick exit if a conflict log table was not requested. */
+	if (!CONFLICTS_LOGGED_TO_TABLE(*log_dest))
+		return NULL;
+
+	conflictlogrelid = MySubscription->conflictlogrelid;
+
+	Assert(OidIsValid(conflictlogrelid));
+
+	return table_open(conflictlogrelid, RowExclusiveLock);
+}
+
+/*
+ * InsertConflictLogTuple
+ *
+ * Insert conflict log tuple into the conflict log table. It uses
+ * HEAP_INSERT_NO_LOGICAL to explicitly block logical decoding of the tuple
+ * inserted into the conflict log table.
+ */
+void
+InsertConflictLogTuple(Relation conflictlogrel)
+{
+	/* A valid tuple must be prepared and stored in MyLogicalRepWorker. */
+	Assert(MyLogicalRepWorker->conflict_log_tuple != NULL);
+
+	heap_insert(conflictlogrel, MyLogicalRepWorker->conflict_log_tuple,
+				GetCurrentCommandId(true), HEAP_INSERT_NO_LOGICAL, NULL);
+
+	/* Free conflict log tuple. */
+	heap_freetuple(MyLogicalRepWorker->conflict_log_tuple);
+	MyLogicalRepWorker->conflict_log_tuple = NULL;
+}
+
 /*
  * Add SQLSTATE error code to the current conflict report.
  */
@@ -771,6 +998,40 @@ get_tuple_desc(EState *estate, ResultRelInfo *relinfo, ConflictType type,
 	}
 }
 
+/*
+ * Helper function to extract the "raw" index key Datums and their null flags
+ * from a TupleTableSlot, given an already open index descriptor.
+ * This is the reusable core logic.
+ */
+static void
+build_index_datums_from_slot(EState *estate, Relation localrel,
+							 TupleTableSlot *slot,
+							 Relation indexDesc, Datum *values,
+							 bool *isnull)
+{
+	TupleTableSlot *tableslot = slot;
+
+	/*
+	 * If the slot is a virtual slot, copy it into a heap tuple slot as
+	 * FormIndexDatum only works with heap tuple slots.
+	 */
+	if (TTS_IS_VIRTUAL(slot))
+	{
+		/* Slot is created within the EState's tuple table */
+		tableslot = table_slot_create(localrel, &estate->es_tupleTable);
+		tableslot = ExecCopySlot(tableslot, slot);
+	}
+
+	/*
+	 * Initialize ecxt_scantuple for potential use in FormIndexDatum
+	 */
+	GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
+
+	/* Form the index datums */
+	FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values,
+				   isnull);
+}
+
 /*
  * Helper functions to construct a string describing the contents of an index
  * entry. See BuildIndexValueDescription for details.
@@ -786,41 +1047,339 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot,
 	Relation	indexDesc;
 	Datum		values[INDEX_MAX_KEYS];
 	bool		isnull[INDEX_MAX_KEYS];
-	TupleTableSlot *tableslot = slot;
 
-	if (!tableslot)
+	if (!slot)
 		return NULL;
 
 	Assert(CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
 
 	indexDesc = index_open(indexoid, NoLock);
 
-	/*
-	 * If the slot is a virtual slot, copy it into a heap tuple slot as
-	 * FormIndexDatum only works with heap tuple slots.
-	 */
-	if (TTS_IS_VIRTUAL(slot))
+	build_index_datums_from_slot(estate, localrel, slot, indexDesc, values,
+								 isnull);
+
+	index_value = BuildIndexValueDescription(indexDesc, values, isnull);
+
+	index_close(indexDesc, NoLock);
+
+	return index_value;
+}
+
+/*
+ * tuple_table_slot_to_json_datum
+ *
+ * Helper function to convert a TupleTableSlot to JSON.
+ */
+static Datum
+tuple_table_slot_to_json_datum(TupleTableSlot *slot)
+{
+	HeapTuple	tuple;
+	Datum		datum;
+	Datum		json;
+
+	Assert(slot != NULL);
+
+	tuple = ExecCopySlotHeapTuple(slot);
+	datum = heap_copy_tuple_as_datum(tuple, slot->tts_tupleDescriptor);
+
+	json = DirectFunctionCall1(row_to_json, datum);
+	heap_freetuple(tuple);
+
+	return json;
+}
+
+/*
+ * tuple_table_slot_to_indextup_json
+ *
+ * Fetch replica identity key from the tuple table slot and convert into a
+ * JSON datum.
+ */
+static Datum
+tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
+								  Oid indexid, TupleTableSlot *slot)
+{
+	Relation	indexDesc;
+	Datum		values[INDEX_MAX_KEYS];
+	bool		isnull[INDEX_MAX_KEYS];
+	HeapTuple	tuple;
+	TupleDesc	tupdesc;
+	Datum		datum;
+
+	Assert(slot != NULL);
+
+	Assert(CheckRelationOidLockedByMe(indexid, RowExclusiveLock, true));
+
+	indexDesc = index_open(indexid, NoLock);
+
+	build_index_datums_from_slot(estate, localrel, slot, indexDesc, values,
+								 isnull);
+	tupdesc = CreateTupleDescCopy(RelationGetDescr(indexDesc));
+
+	/* Bless the tupdesc so it can be looked up by row_to_json. */
+	BlessTupleDesc(tupdesc);
+
+	/* Form the replica identity tuple. */
+	tuple = heap_form_tuple(tupdesc, values, isnull);
+	datum = heap_copy_tuple_as_datum(tuple, tupdesc);
+
+	heap_freetuple(tuple);
+	FreeTupleDesc(tupdesc);
+	index_close(indexDesc, NoLock);
+
+	/* Convert to a JSON datum. */
+	return DirectFunctionCall1(row_to_json, datum);
+}
+
+/*
+ * build_conflict_tupledesc
+ *
+ * Build and bless a tuple descriptor for the conflict log table based on the
+ * predefined LocalConflictSchema.
+ */
+static TupleDesc
+build_conflict_tupledesc(void)
+{
+	static TupleDesc cached_tupdesc = NULL;
+
+	if (cached_tupdesc == NULL)
 	{
-		tableslot = table_slot_create(localrel, &estate->es_tupleTable);
-		tableslot = ExecCopySlot(tableslot, slot);
+		MemoryContext oldcxt;
+
+		oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+
+		cached_tupdesc = CreateTemplateTupleDesc(NUM_LOCAL_CONFLICT_ATTRS);
+
+		for (int i = 0; i < NUM_LOCAL_CONFLICT_ATTRS; i++)
+			TupleDescInitEntry(cached_tupdesc,
+							   (AttrNumber) (i + 1),
+							   LocalConflictSchema[i].attname,
+							   LocalConflictSchema[i].atttypid,
+							   -1, 0);
+
+		TupleDescFinalize(cached_tupdesc);
+
+		/*
+		 * Bless once so it can be used as a RECORD type (e.g. for
+		 * row_to_json or other record-based operations).
+		 */
+		BlessTupleDesc(cached_tupdesc);
+
+		MemoryContextSwitchTo(oldcxt);
 	}
 
-	/*
-	 * Initialize ecxt_scantuple for potential use in FormIndexDatum when
-	 * index expressions are present.
-	 */
-	GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
+	return cached_tupdesc;
+}
 
-	/*
-	 * The values/nulls arrays passed to BuildIndexValueDescription should be
-	 * the results of FormIndexDatum, which are the "raw" input to the index
-	 * AM.
-	 */
-	FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull);
+/*
+ * Builds the local conflicts JSON array column from the list of
+ * ConflictTupleInfo objects.
+ *
+ * Example output structure:
+ * [ { "xid": "1001", "commit_ts": "...", "origin": "...", "tuple": {...} }, ... ]
+ */
+static Datum
+build_local_conflicts_json_array(EState *estate, Relation rel,
+								 ConflictType conflict_type,
+								 List *conflicttuples)
+{
+	ListCell   *lc;
+	List	   *json_datums = NIL;
+	Datum	   *json_datum_array;
+	Datum		json_array_datum;
+	int			num_conflicts;
+	int			i;
+	int16		typlen;
+	bool		typbyval;
+	char		typalign;
+	TupleDesc	tupdesc;
 
-	index_value = BuildIndexValueDescription(indexDesc, values, isnull);
+	/* Build local conflicts tuple descriptor. */
+	tupdesc = build_conflict_tupledesc();
 
-	index_close(indexDesc, NoLock);
+	/* Process local conflict tuple list and prepare an array of JSON. */
+	foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
+	{
+		Datum		values[NUM_LOCAL_CONFLICT_ATTRS] = {0};
+		bool		nulls[NUM_LOCAL_CONFLICT_ATTRS] = {0};
+		char	   *origin_name = NULL;
+		HeapTuple	tuple;
+		Datum		json_datum;
+		int			attno;
+
+		attno = 0;
+		values[attno++] = TransactionIdGetDatum(conflicttuple->xmin);
+
+		if (conflicttuple->ts)
+			values[attno++] = TimestampTzGetDatum(conflicttuple->ts);
+		else
+			nulls[attno++] = true;
 
-	return index_value;
+		if (conflicttuple->origin != InvalidReplOriginId)
+			replorigin_by_oid(conflicttuple->origin, true, &origin_name);
+
+		/* Store empty string if origin name for the tuple is NULL. */
+		if (origin_name != NULL)
+			values[attno++] = CStringGetTextDatum(origin_name);
+		else
+			nulls[attno++] = true;
+
+		/*
+		 * Add the conflicting key values in the case of a unique constraint
+		 * violation.
+		 */
+		if (conflict_type == CT_INSERT_EXISTS ||
+			conflict_type == CT_UPDATE_EXISTS ||
+			conflict_type == CT_MULTIPLE_UNIQUE_CONFLICTS)
+		{
+			Oid	indexoid = conflicttuple->indexoid;
+
+			Assert(OidIsValid(indexoid) && conflicttuple->slot &&
+				   CheckRelationOidLockedByMe(indexoid, RowExclusiveLock,
+											  true));
+			values[attno++] =
+					tuple_table_slot_to_indextup_json(estate, rel,
+													  indexoid,
+													  conflicttuple->slot);
+		}
+		else
+			nulls[attno++] = true;
+
+		/* Convert conflicting tuple to JSON datum. */
+		if (conflicttuple->slot)
+			values[attno] = tuple_table_slot_to_json_datum(conflicttuple->slot);
+		else
+			nulls[attno] = true;
+
+		Assert(attno + 1 == NUM_LOCAL_CONFLICT_ATTRS);
+
+		tuple = heap_form_tuple(tupdesc, values, nulls);
+
+		json_datum = heap_copy_tuple_as_datum(tuple, tupdesc);
+
+		/*
+		 * Build the higher level JSON datum in format described in function
+		 * header.
+		 */
+		json_datum = DirectFunctionCall1(row_to_json, json_datum);
+
+		/* Done with the temporary tuple. */
+		heap_freetuple(tuple);
+
+		/* Add to the array element. */
+		json_datums = lappend(json_datums, (void *) json_datum);
+	}
+
+	num_conflicts = list_length(json_datums);
+
+	json_datum_array = palloc_array(Datum, num_conflicts);
+
+	i = 0;
+	foreach(lc, json_datums)
+	{
+		json_datum_array[i] = (Datum) lfirst(lc);
+		i++;
+	}
+
+	/* Construct the JSON array Datum. */
+	get_typlenbyvalalign(JSONOID, &typlen, &typbyval, &typalign);
+	json_array_datum = PointerGetDatum(construct_array(json_datum_array,
+													   num_conflicts,
+													   JSONOID,
+													   typlen,
+													   typbyval,
+													   typalign));
+	pfree(json_datum_array);
+
+	return json_array_datum;
+}
+
+/*
+ * prepare_conflict_log_tuple
+ *
+ * This routine prepares a tuple detailing a conflict encountered during
+ * logical replication. The prepared tuple will be stored in
+ * MyLogicalRepWorker->conflict_log_tuple which should be inserted into the
+ * conflict log table by calling InsertConflictLogTuple.
+ */
+static void
+prepare_conflict_log_tuple(EState *estate, Relation rel,
+						   Relation conflictlogrel,
+						   ConflictType conflict_type,
+						   TupleTableSlot *searchslot,
+						   List *conflicttuples,
+						   TupleTableSlot *remoteslot)
+{
+	Datum		values[NUM_CONFLICT_ATTRS] = {0};
+	bool		nulls[NUM_CONFLICT_ATTRS] = {0};
+	int			attno;
+	char	   *remote_origin = NULL;
+	MemoryContext	oldctx;
+
+	Assert(MyLogicalRepWorker->conflict_log_tuple == NULL);
+
+	/* Populate the values and nulls arrays. */
+	attno = 0;
+	values[attno++] = ObjectIdGetDatum(RelationGetRelid(rel));
+
+	values[attno++] =
+			CStringGetTextDatum(get_namespace_name(RelationGetNamespace(rel)));
+
+	values[attno++] = CStringGetTextDatum(RelationGetRelationName(rel));
+
+	values[attno++] = CStringGetTextDatum(ConflictTypeNames[conflict_type]);
+
+	if (TransactionIdIsValid(remote_xid))
+		values[attno++] = TransactionIdGetDatum(remote_xid);
+	else
+		nulls[attno++] = true;
+
+	values[attno++] = LSNGetDatum(remote_final_lsn);
+
+	if (remote_commit_ts > 0)
+		values[attno++] = TimestampTzGetDatum(remote_commit_ts);
+	else
+		nulls[attno++] = true;
+
+	if (replorigin_xact_state.origin != InvalidReplOriginId)
+		replorigin_by_oid(replorigin_xact_state.origin, true, &remote_origin);
+
+	if (remote_origin != NULL)
+		values[attno++] = CStringGetTextDatum(remote_origin);
+	else
+		nulls[attno++] = true;
+
+	if (!TupIsNull(remoteslot))
+		values[attno++] = tuple_table_slot_to_json_datum(remoteslot);
+	else
+		nulls[attno++] = true;
+
+	if (!TupIsNull(searchslot))
+	{
+		Oid		replica_index = GetRelationIdentityOrPK(rel);
+
+		/*
+		 * If the table has a valid replica identity index, build the index
+		 * JSON datum from key value. Otherwise, construct it from the complete
+		 * tuple in REPLICA IDENTITY FULL cases.
+		 */
+		if (OidIsValid(replica_index))
+			values[attno++] = tuple_table_slot_to_indextup_json(estate, rel,
+																replica_index,
+																searchslot);
+		else
+			values[attno++] = tuple_table_slot_to_json_datum(searchslot);
+	}
+	else
+		nulls[attno++] = true;
+
+	values[attno] = build_local_conflicts_json_array(estate, rel,
+													 conflict_type,
+													 conflicttuples);
+
+	Assert(attno + 1 == NUM_CONFLICT_ATTRS);
+
+	oldctx = MemoryContextSwitchTo(ApplyContext);
+	MyLogicalRepWorker->conflict_log_tuple =
+		heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls);
+	MemoryContextSwitchTo(oldctx);
 }
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 313e31ff2e3..05a30342f69 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -487,6 +487,7 @@ retry:
 	worker->oldest_nonremovable_xid = retain_dead_tuples
 		? MyReplicationSlot->data.xmin
 		: InvalidTransactionId;
+	worker->conflict_log_tuple = NULL;
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a3f2406ed83..7b462904c9b 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -487,7 +487,9 @@ static bool MySubscriptionValid = false;
 static List *on_commit_wakeup_workers_subids = NIL;
 
 bool		in_remote_transaction = false;
-static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
+XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
+TransactionId	remote_xid = InvalidTransactionId;
+TimestampTz	remote_commit_ts = 0;
 
 /* fields valid only when processing streamed transaction */
 static bool in_streamed_transaction = false;
@@ -1236,6 +1238,8 @@ apply_handle_begin(StringInfo s)
 	set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn);
 
 	remote_final_lsn = begin_data.final_lsn;
+	remote_commit_ts = begin_data.committime;
+	remote_xid = begin_data.xid;
 
 	maybe_start_skipping_changes(begin_data.final_lsn);
 
@@ -1767,6 +1771,10 @@ apply_handle_stream_start(StringInfo s)
 				(errcode(ERRCODE_PROTOCOL_VIOLATION),
 				 errmsg_internal("invalid transaction ID in streamed replication transaction")));
 
+	remote_xid = stream_xid;
+	remote_final_lsn = InvalidXLogRecPtr;
+	remote_commit_ts = 0;
+
 	set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
 
 	/* Try to allocate a worker for the streaming transaction. */
@@ -2440,6 +2448,9 @@ apply_handle_stream_commit(StringInfo s)
 			/* Unlink the files with serialized changes and subxact info. */
 			stream_cleanup_files(MyLogicalRepWorker->subid, xid);
 
+			/* Set remote_commit_ts for conflict logging. */
+			remote_commit_ts = commit_data.committime;
+
 			elog(DEBUG1, "finished processing the STREAM COMMIT command");
 			break;
 
@@ -5666,6 +5677,7 @@ start_apply(XLogRecPtr origin_startpos)
 			 */
 			AbortOutOfAnyTransaction();
 			pgstat_report_subscription_error(MySubscription->oid);
+			ProcessPendingConflictLogTuple();
 
 			PG_RE_THROW();
 		}
@@ -6040,6 +6052,8 @@ DisableSubscriptionAndExit(void)
 	 */
 	pgstat_report_subscription_error(MyLogicalRepWorker->subid);
 
+	ProcessPendingConflictLogTuple();
+
 	/* Disable the subscription */
 	StartTransactionCommand();
 
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index a017e1e6cb5..b727cc285e9 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -115,5 +115,8 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
 								TupleTableSlot *searchslot,
 								TupleTableSlot *remoteslot,
 								List *conflicttuples);
+extern void ProcessPendingConflictLogTuple(void);
 extern void InitConflictIndexes(ResultRelInfo *relInfo);
+extern Relation GetConflictLogDestAndTable(ConflictLogDest *log_dest);
+extern void InsertConflictLogTuple(Relation conflictlogrel);
 #endif
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 745b7d9e969..6b6525dc2e2 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -100,6 +100,9 @@ typedef struct LogicalRepWorker
 	 */
 	TransactionId oldest_nonremovable_xid;
 
+	/* A conflict log tuple that is prepared but not yet inserted. */
+	HeapTuple	conflict_log_tuple;
+
 	/* Stats. */
 	XLogRecPtr	last_lsn;
 	TimestampTz last_send_time;
@@ -255,6 +258,10 @@ extern PGDLLIMPORT bool InitializingApplyWorker;
 
 extern PGDLLIMPORT List *table_states_not_ready;
 
+extern XLogRecPtr remote_final_lsn;
+extern TimestampTz remote_commit_ts;
+extern TransactionId remote_xid;
+
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype,
 												Oid subid, Oid relid,
diff --git a/src/test/subscription/t/030_origin.pl b/src/test/subscription/t/030_origin.pl
index 6bc6b7874c2..5f4d00bdd33 100644
--- a/src/test/subscription/t/030_origin.pl
+++ b/src/test/subscription/t/030_origin.pl
@@ -166,7 +166,7 @@ is($result, qq(32), 'The node_A data replicated to node_B');
 $node_C->safe_psql('postgres', "UPDATE $tab SET a = 33 WHERE a = 32;");
 
 $node_B->wait_for_log(
-	qr/conflict detected on relation "public.$tab_unquoted": conflict=update_origin_differs.*\n.*DETAIL:.* Updating the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(32\), remote row \(33\), replica identity \(a\)=\(32\)./
+	qr/conflict detected on relation "public.$tab": conflict=update_origin_differs.*\n.*DETAIL:.* Updating the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(32\), remote row \(33\), replica identity \(a\)=\(32\)./
 );
 
 $node_B->safe_psql('postgres', "DELETE FROM $tab;");
@@ -182,7 +182,7 @@ is($result, qq(33), 'The node_A data replicated to node_B');
 $node_C->safe_psql('postgres', "DELETE FROM $tab WHERE a = 33;");
 
 $node_B->wait_for_log(
-	qr/conflict detected on relation "public.$tab_unquoted": conflict=delete_origin_differs.*\n.*DETAIL:.* Deleting the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(33\), replica identity \(a\)=\(33\).*/
+	qr/conflict detected on relation "public.$tab": conflict=delete_origin_differs.*\n.*DETAIL:.* Deleting the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(33\), replica identity \(a\)=\(33\).*/
 );
 
 # The remaining tests no longer test conflict detection.
diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index f23fe6af2a5..de2d37c8754 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -50,7 +50,7 @@ $node_subscriber->safe_psql(
 	'postgres',
 	"CREATE SUBSCRIPTION sub_tab
 	 CONNECTION '$publisher_connstr application_name=$appname'
-	 PUBLICATION pub_tab;");
+	 PUBLICATION pub_tab WITH (conflict_log_destination=all)");
 
 # Wait for initial table sync to finish
 $node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
@@ -84,10 +84,35 @@ $node_subscriber->wait_for_log(
 .*Key already exists in unique index \"conf_tab_c_key\", modified in transaction .*: key \(c\)=\(4\), local row \(4, 4, 4\)./,
 	$log_offset);
 
+# Verify the contents of the Conflict Log Table (CLT)
+# This section ensures that the CLT contains the expected
+# type and specific key data.
+my $subid = $node_subscriber->safe_psql('postgres',
+	"SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';");
+my $clt = "pg_conflict.pg_conflict_log_$subid";
+
+# Wait for the conflict to be logged in the CLT
+my $log_check = $node_subscriber->poll_query_until(
+    'postgres',
+    "SELECT count(*) > 0 FROM $clt;"
+);
+
+my $conflict_check = $node_subscriber->safe_psql('postgres',
+    "SELECT count(*) >= 1 FROM $clt WHERE conflict_type = 'multiple_unique_conflicts';");
+is($conflict_check, 't', 'Verified multiple_unique_conflicts logged into conflict log table');
+
+my $json_query = "SELECT local_conflicts FROM $clt;";
+my $raw_json = $node_subscriber->safe_psql('postgres', $json_query);
+
+# Verify that '2' is present inside the JSON structure using a regex
+# This matches the key/value pattern for "a": 2
+like($raw_json, qr/\\"a\\":2/, 'Verified that key 2 exists in the local_conflicts');
+
 pass('multiple_unique_conflicts detected during insert');
 
 # Truncate table to get rid of the error
 $node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
+$node_subscriber->safe_psql('postgres', "DELETE FROM $clt");
 
 ##################################################
 # Test multiple_unique_conflicts due to UPDATE
@@ -114,6 +139,26 @@ $node_subscriber->wait_for_log(
 .*Key already exists in unique index \"conf_tab_c_key\", modified in transaction .*: key \(c\)=\(8\), local row \(8, 8, 8\)./,
 	$log_offset);
 
+# Verify the contents of the Conflict Log Table (CLT)
+# This section ensures that the CLT contains the expected
+# type and specific key data.
+
+# Wait for the conflict to be logged in the CLT
+$log_check = $node_subscriber->poll_query_until(
+    'postgres',
+    "SELECT count(*) > 0 FROM $clt;"
+);
+
+$conflict_check = $node_subscriber->safe_psql('postgres',
+    "SELECT count(*) >= 1 FROM $clt WHERE conflict_type = 'multiple_unique_conflicts';");
+is($conflict_check, 't', 'Verified multiple_unique_conflicts logged into conflict log table');
+
+$raw_json = $node_subscriber->safe_psql('postgres', $json_query);
+
+# Verify that '6' is present inside the JSON structure using a regex
+# This matches the key/value pattern for "a": 6
+like($raw_json, qr/\\"a\\":6/, 'Verified that key 6 exists in the local_conflicts');
+
 pass('multiple_unique_conflicts detected during update');
 
 # Truncate table to get rid of the error
-- 
2.49.0