From 0287957686dcec0d4108be607403ea0cddecaf02 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumarb@google.com>
Date: Wed, 19 Aug 2026 23:10:08 +0800
Subject: [PATCH v70 2/3] 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.

Only resolved (LOG-level) conflicts are recorded in the conflict log table.
Conflicts that raise an ERROR (such as unique constraint violations) halt
replication and abort the transaction, so they are logged exclusively to the
server log.

Local Conflicts Column: The 'local_conflicts' column is typed as an array of
JSON objects (json[]). Although currently recorded conflicts involve a single
local tuple, the column type is preserved as a JSON array for future-proofing.
This avoids modifying the exposed table schema and dealing with upgrade handling
when multi-row conflict resolution is introduced in the future.

The JSON array uses the following structured format:
[ { "xid": "1001", "commit_ts": "2025-12-25 10:00:00+05:30", "origin": "node_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}

The remote transaction details (xid, final LSN, commit timestamp) recorded
in the conflict log table are fetched from the apply worker's remote
transaction context through GetRemoteTransactionInfoForConflict() instead of
new global variables.
---
 doc/src/sgml/logical-replication.sgml      | 697 +++++++++++++--------
 doc/src/sgml/ref/alter_subscription.sgml   |   3 +-
 doc/src/sgml/ref/create_subscription.sgml  |  22 +-
 src/backend/replication/logical/conflict.c | 569 +++++++++++++++--
 src/backend/replication/logical/worker.c   | 101 ++-
 src/include/replication/conflict.h         |   2 +
 src/include/replication/worker_internal.h  |   5 +
 src/include/utils/rel.h                    |  13 +-
 src/test/subscription/t/030_origin.pl      |   4 +-
 src/test/subscription/t/035_conflicts.pl   |  38 +-
 10 files changed, 1098 insertions(+), 356 deletions(-)

diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 3a61e2d6889..0d9892764df 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -2033,120 +2033,267 @@ Included in publications:
    operations will simply be skipped.
   </para>
 
-  <para>
-   Additional logging is triggered, and the conflict statistics are collected (displayed in the
-   <link linkend="monitoring-pg-stat-subscription-stats"><structname>pg_stat_subscription_stats</structname></link> view)
-   in the following <firstterm>conflict</firstterm> cases:
-   <variablelist>
-    <varlistentry id="conflict-insert-exists" xreflabel="insert_exists">
-     <term><literal>insert_exists</literal></term>
-     <listitem>
-      <para>
-       Inserting a row that violates a <literal>NOT DEFERRABLE</literal>
-       unique constraint. Note that to log the origin and commit
-       timestamp details of the conflicting key,
-       <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
-       should be enabled on the subscriber. In this case, an error will be
-       raised until the conflict is resolved manually.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry id="conflict-update-origin-differs" xreflabel="update_origin_differs">
-     <term><literal>update_origin_differs</literal></term>
-     <listitem>
-      <para>
-       Updating a row that was previously modified by another origin.
-       Note that this conflict can only be detected when
-       <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
-       is enabled on the subscriber. Currently, the update is always applied
-       regardless of the origin of the local row.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry id="conflict-update-exists" xreflabel="update_exists">
-     <term><literal>update_exists</literal></term>
-     <listitem>
-      <para>
-       The updated value of a row violates a <literal>NOT DEFERRABLE</literal>
-       unique constraint. Note that to log the origin and commit
-       timestamp details of the conflicting key,
-       <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
-       should be enabled on the subscriber. In this case, an error will be
-       raised until the conflict is resolved manually. Note that when updating a
-       partitioned table, if the updated row value satisfies another partition
-       constraint resulting in the row being inserted into a new partition, the
-       <literal>insert_exists</literal> conflict may arise if the new row
-       violates a <literal>NOT DEFERRABLE</literal> unique constraint.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry id="conflict-update-deleted" xreflabel="update_deleted">
-     <term><literal>update_deleted</literal></term>
-     <listitem>
-      <para>
-       The tuple to be updated was concurrently deleted by another origin. The
-       update will simply be skipped in this scenario. Note that this conflict
-       can only be detected when
-       <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
-       and <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
-       are enabled. Note that if a tuple cannot be found due to the table being
-       truncated, only a <literal>update_missing</literal> conflict will
-       arise. Additionally, if the tuple was deleted by the same origin, an
-       <literal>update_missing</literal> conflict will arise.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry id="conflict-update-missing" xreflabel="update_missing">
-     <term><literal>update_missing</literal></term>
-     <listitem>
-      <para>
-       The row to be updated was not found. The update will simply be
-       skipped in this scenario.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry id="conflict-delete-origin-differs" xreflabel="delete_origin_differs">
-     <term><literal>delete_origin_differs</literal></term>
-     <listitem>
-      <para>
-       Deleting a row that was previously modified by another origin. Note that
-       this conflict can only be detected when
-       <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
-       is enabled on the subscriber. Currently, the delete is always applied
-       regardless of the origin of the local row.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry id="conflict-delete-missing" xreflabel="delete_missing">
-     <term><literal>delete_missing</literal></term>
-     <listitem>
-      <para>
-       The row to be deleted was not found. The delete will simply be
-       skipped in this scenario.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry id="conflict-multiple-unique-conflicts" xreflabel="multiple_unique_conflicts">
-     <term><literal>multiple_unique_conflicts</literal></term>
-     <listitem>
-      <para>
-       Inserting or updating a row violates multiple
-       <literal>NOT DEFERRABLE</literal> unique constraints. Note that to log
-       the origin and commit timestamp details of conflicting keys, ensure
-       that <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
-       is enabled on the subscriber. In this case, an error will be raised until
-       the conflict is resolved manually.
-      </para>
-     </listitem>
-    </varlistentry>
-   </variablelist>
-    Note that there are other conflict scenarios, such as exclusion constraint
-    violations. Currently, we do not provide additional details for them in the
-    log.
-  </para>
+  <sect2 id="logical-replication-conflict-logging">
+   <title>Conflict logging</title>
+   <para>
+    Conflict logging is triggered, and conflict statistics are collected
+    (displayed in the
+    <link linkend="monitoring-pg-stat-subscription-stats"><structname>pg_stat_subscription_stats</structname></link>
+    view) for the conflict cases described below. Depending on the value of
+    <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>,
+    the conflict information is written to the server log, a conflict log table,
+    or both.
+   </para>
 
-  <para>
-   The log format for logical replication conflicts is as follows:
+   <para>
+    The supported conflict cases are:
+    <variablelist>
+     <varlistentry id="conflict-insert-exists" xreflabel="insert_exists">
+      <term><literal>insert_exists</literal></term>
+      <listitem>
+       <para>
+        Inserting a row that violates a <literal>NOT DEFERRABLE</literal>
+        unique constraint. Note that to log the origin and commit
+        timestamp details of the conflicting key,
+        <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+        should be enabled on the subscriber. In this case, an error will be
+        raised until the conflict is resolved manually.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry id="conflict-update-origin-differs" xreflabel="update_origin_differs">
+      <term><literal>update_origin_differs</literal></term>
+      <listitem>
+       <para>
+        Updating a row that was previously modified by another origin.
+        Note that this conflict can only be detected when
+        <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+        is enabled on the subscriber. Currently, the update is always applied
+        regardless of the origin of the local row.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry id="conflict-update-exists" xreflabel="update_exists">
+      <term><literal>update_exists</literal></term>
+      <listitem>
+       <para>
+        The updated value of a row violates a <literal>NOT DEFERRABLE</literal>
+        unique constraint. Note that to log the origin and commit
+        timestamp details of the conflicting key,
+        <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+        should be enabled on the subscriber. In this case, an error will be
+        raised until the conflict is resolved manually. Note that when updating a
+        partitioned table, if the updated row value satisfies another partition
+        constraint resulting in the row being inserted into a new partition, the
+        <literal>insert_exists</literal> conflict may arise if the new row
+        violates a <literal>NOT DEFERRABLE</literal> unique constraint.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry id="conflict-update-deleted" xreflabel="update_deleted">
+      <term><literal>update_deleted</literal></term>
+      <listitem>
+       <para>
+        The tuple to be updated was concurrently deleted by another origin. The
+        update will simply be skipped in this scenario. Note that this conflict
+        can only be detected when
+        <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+        and <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
+        are enabled. Note that if a tuple cannot be found due to the table being
+        truncated, only a <literal>update_missing</literal> conflict will
+        arise. Additionally, if the tuple was deleted by the same origin, an
+        <literal>update_missing</literal> conflict will arise.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry id="conflict-update-missing" xreflabel="update_missing">
+      <term><literal>update_missing</literal></term>
+      <listitem>
+       <para>
+        The row to be updated was not found. The update will simply be
+        skipped in this scenario.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry id="conflict-delete-origin-differs" xreflabel="delete_origin_differs">
+      <term><literal>delete_origin_differs</literal></term>
+      <listitem>
+       <para>
+        Deleting a row that was previously modified by another origin. Note that
+        this conflict can only be detected when
+        <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+        is enabled on the subscriber. Currently, the delete is always applied
+        regardless of the origin of the local row.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry id="conflict-delete-missing" xreflabel="delete_missing">
+      <term><literal>delete_missing</literal></term>
+      <listitem>
+       <para>
+        The row to be deleted was not found. The delete will simply be
+        skipped in this scenario.
+       </para>
+      </listitem>
+     </varlistentry>
+     <varlistentry id="conflict-multiple-unique-conflicts" xreflabel="multiple_unique_conflicts">
+      <term><literal>multiple_unique_conflicts</literal></term>
+      <listitem>
+       <para>
+        Inserting or updating a row violates multiple
+        <literal>NOT DEFERRABLE</literal> unique constraints. Note that to log
+        the origin and commit timestamp details of conflicting keys, ensure
+        that <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+        is enabled on the subscriber. In this case, an error will be raised until
+        the conflict is resolved manually.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+      Note that there are other conflict scenarios, such as exclusion constraint
+      violations. Currently, we do not provide additional details for them in the
+      log.
+    </para>
+
+   <sect3 id="logical-replication-conflict-table-based-logging">
+    <title>Table-based logging</title>
+    <para>
+     If the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+     parameter is set to <literal>table</literal> or <literal>all</literal>,
+     a dedicated conflict log table will be automatically created. This table is
+     created in the <literal>pg_conflict</literal> namespace. The name of the
+     conflict log table is
+     <literal>pg_conflict_log_&lt;subid&gt;</literal>. The predefined
+     schema of this table is detailed in
+     <xref linkend="logical-replication-conflict-log-schema"/>.
+    </para>
+
+    <note>
+     <para>
+      Only resolved conflicts (logged at <literal>LOG</literal> level where the
+      apply transaction continues) are recorded in the conflict log table.
+      Conflicts that raise an <literal>ERROR</literal> halt replication and are
+      reported to the server log alone.
+     </para>
+    </note>
+
+    <para>
+     The conflict log table serves as an audit trail for resolved conflicts
+     where replication continues and the transaction successfully commits
+     (such as <literal>update_origin_differs</literal>,
+     <literal>update_missing</literal>, <literal>update_deleted</literal>,
+     <literal>delete_origin_differs</literal>, and
+     <literal>delete_missing</literal>). In these cases, local data may diverge
+     or be overwritten, making a queryable audit log valuable for DBAs to
+     understand post-hoc why a row changed or was skipped.
+    </para>
+
+    <para>
+     In contrast, conflicts raised at <literal>ERROR</literal> level (such as
+     <literal>insert_exists</literal>, <literal>update_exists</literal>, and
+     <literal>multiple_unique_conflicts</literal>) halt replication and abort the
+     entire apply transaction. Because the transaction rolls back and no data
+     modifications survive on the subscriber, no data divergence occurs. These
+     errors are always reported to the server log (regardless of the
+     <literal>conflict_log_destination</literal> setting) so that monitoring and
+     alerting systems can capture the replication failure.
+    </para>
+
+    <table id="logical-replication-conflict-log-schema">
+     <title>Conflict Log Table Schema</title>
+     <tgroup cols="3">
+      <thead>
+       <row>
+        <entry>Column</entry>
+        <entry>Type</entry>
+        <entry>Description</entry>
+       </row>
+      </thead>
+      <tbody>
+       <row>
+        <entry><literal>relid</literal></entry>
+        <entry><type>oid</type></entry>
+        <entry>The OID of the local table where the conflict occurred.</entry>
+       </row>
+       <row>
+        <entry><literal>schemaname</literal></entry>
+        <entry><type>text</type></entry>
+        <entry>The schema name of the conflicting table.</entry>
+       </row>
+       <row>
+        <entry><literal>relname</literal></entry>
+        <entry><type>text</type></entry>
+        <entry>The name of the conflicting table.</entry>
+       </row>
+       <row>
+        <entry><literal>conflict_type</literal></entry>
+        <entry><type>text</type></entry>
+        <entry>The type of conflict that occurred (e.g., <literal>update_origin_differs</literal>).</entry>
+       </row>
+       <row>
+        <entry><literal>remote_xid</literal></entry>
+        <entry><type>xid</type></entry>
+        <entry>The remote transaction ID that caused the conflict.</entry>
+       </row>
+       <row>
+        <entry><literal>remote_commit_lsn</literal></entry>
+        <entry><type>pg_lsn</type></entry>
+        <entry>The final LSN of the remote transaction. This is <literal>NULL</literal> if the conflict is handled by a parallel apply worker, because in-progress transactions are streamed and applied before the remote commit LSN is known.</entry>
+       </row>
+       <row>
+        <entry><literal>remote_commit_ts</literal></entry>
+        <entry><type>timestamptz</type></entry>
+        <entry>The remote commit timestamp of the remote transaction. This is <literal>NULL</literal> if the conflict is handled by a parallel apply worker, because in-progress transactions are streamed and applied before the remote commit timestamp is known. It is also <literal>NULL</literal> for a conflict detected while applying a transaction that is later prepared with <literal>PREPARE TRANSACTION</literal>, because the remote commit timestamp is known only when the commit prepared record is processed.</entry>
+       </row>
+       <row>
+        <entry><literal>remote_origin</literal></entry>
+        <entry><type>text</type></entry>
+        <entry>The origin of the remote transaction.</entry>
+       </row>
+       <row>
+        <entry><literal>replica_identity_full</literal></entry>
+        <entry><type>boolean</type></entry>
+        <entry>Indicates whether <literal>replica_identity</literal> represents a full tuple (<literal>true</literal>) or key values of a replica identity index (<literal>false</literal>). This is <literal>NULL</literal> when <literal>replica_identity</literal> is not logged.</entry>
+       </row>
+       <row>
+        <entry><literal>replica_identity</literal></entry>
+        <entry><type>json</type></entry>
+        <entry>The JSON representation of the replica identity key values or full tuple.</entry>
+       </row>
+       <row>
+        <entry><literal>remote_tuple</literal></entry>
+        <entry><type>json</type></entry>
+        <entry>The JSON representation of the incoming remote row that caused
+        the conflict.</entry>
+       </row>
+       <row>
+        <entry><literal>local_conflicts</literal></entry>
+        <entry><type>json[]</type></entry>
+        <entry>
+         An array of JSON objects representing the state of existing local
+         row(s) that caused the conflict. Each object includes the local
+         transaction ID (<literal>xid</literal>), commit timestamp
+         (<literal>commit_ts</literal>), origin (<literal>origin</literal>),
+         and the full local row image (<literal>tuple</literal>).
+        </entry>
+       </row>
+      </tbody>
+     </tgroup>
+    </table>
+
+    <para>
+     The conflicting row data, including the incoming remote row (<literal>remote_tuple</literal>)
+     and the associated local conflict details (<literal>local_conflicts</literal>), is stored in
+     <type>JSON</type> formats for flexible querying and analysis.
+    </para>
+   </sect3>
+
+   <sect3 id="logical-replication-conflict-file-based-logging">
+    <title>File-based logging</title>
+    <para>
+     If the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+     parameter is set to <literal>log</literal> or <literal>all</literal>, conflicts
+     are logged to the server log using the following format:
 <synopsis>
 LOG:  conflict detected on relation "<replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>": conflict=<replaceable>conflict_type</replaceable>
 DETAIL:  <replaceable class="parameter">detailed_explanation</replaceable>[: <replaceable class="parameter">detail_values</replaceable> [, ... ]].
@@ -2159,182 +2306,186 @@ DETAIL:  <replaceable class="parameter">detailed_explanation</replaceable>[: <re
     <literal>replica identity</literal> {(<replaceable>column_name</replaceable> <optional>, ...</optional>)=(<replaceable>column_value</replaceable> <optional>, ...</optional>) | full <optional>(<replaceable>column_name</replaceable> <optional>, ...</optional>)=</optional>(<replaceable>column_value</replaceable> <optional>, ...</optional>)}
 </synopsis>
 
-   The log provides the following information:
-   <variablelist>
-    <varlistentry>
-     <term><literal>LOG</literal></term>
-      <listitem>
+    The log provides the following information:
+    <variablelist>
+     <varlistentry>
+      <term><literal>LOG</literal></term>
+       <listitem>
+        <itemizedlist>
+         <listitem>
+          <para>
+          <replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>
+          identifies the local relation involved in the conflict.
+          </para>
+         </listitem>
+         <listitem>
+          <para>
+          <replaceable>conflict_type</replaceable> is the type of conflict that occurred
+          (e.g., <literal>insert_exists</literal>, <literal>update_exists</literal>).
+          </para>
+         </listitem>
+        </itemizedlist>
+       </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><literal>DETAIL</literal></term>
+       <listitem>
        <itemizedlist>
         <listitem>
          <para>
-         <replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>
-         identifies the local relation involved in the conflict.
+          <replaceable class="parameter">detailed_explanation</replaceable> includes
+          the origin, transaction ID, and commit timestamp of the transaction that
+          modified the local row, if available.
+         </para>
+        </listitem>
+        <listitem>
+         <para>
+          The <literal>key</literal> section includes the key values of the local
+          row that violated a unique constraint for
+          <literal>insert_exists</literal>, <literal>update_exists</literal> or
+          <literal>multiple_unique_conflicts</literal> conflicts.
          </para>
         </listitem>
         <listitem>
          <para>
-         <replaceable>conflict_type</replaceable> is the type of conflict that occurred
-         (e.g., <literal>insert_exists</literal>, <literal>update_exists</literal>).
+          The <literal>local row</literal> section includes the local row if its
+          origin differs from the remote row for
+          <literal>update_origin_differs</literal> or <literal>delete_origin_differs</literal>
+          conflicts, or if the key value conflicts with the remote row for
+          <literal>insert_exists</literal>, <literal>update_exists</literal> or
+          <literal>multiple_unique_conflicts</literal> conflicts.
+         </para>
+        </listitem>
+        <listitem>
+         <para>
+          The <literal>remote row</literal> section includes the new row from
+          the remote insert or update operation that caused the conflict. Note that
+          for an update operation, the column value of the new row will be null
+          if the value is unchanged and toasted.
+         </para>
+        </listitem>
+        <listitem>
+         <para>
+          The <literal>replica identity</literal> section includes the replica
+          identity key values that were used to search for the existing local
+          row to be updated or deleted. This may include the full row value
+          if the local relation is marked with
+          <link linkend="sql-altertable-replica-identity-full"><literal>REPLICA IDENTITY FULL</literal></link>.
+         </para>
+        </listitem>
+        <listitem>
+         <para>
+          <replaceable class="parameter">column_name</replaceable> is the column name.
+          For <literal>local row</literal>, <literal>remote row</literal>, and
+          <literal>replica identity full</literal> cases, column names are
+          logged only if the user lacks the privilege to access all columns of
+          the table. If column names are present, they appear in the same order
+          as the corresponding column values.
+         </para>
+        </listitem>
+        <listitem>
+         <para>
+          <replaceable class="parameter">column_value</replaceable> is the column value.
+          The large column values are truncated to 64 bytes.
+         </para>
+        </listitem>
+        <listitem>
+         <para>
+          Note that in case of <literal>multiple_unique_conflicts</literal> conflict,
+          multiple <replaceable class="parameter">detailed_explanation</replaceable>
+          and <replaceable class="parameter">detail_values</replaceable> lines
+          will be generated, each detailing the conflict information associated
+          with distinct unique constraints.
          </para>
         </listitem>
        </itemizedlist>
       </listitem>
-    </varlistentry>
-
-    <varlistentry>
-     <term><literal>DETAIL</literal></term>
-      <listitem>
-      <itemizedlist>
-       <listitem>
-        <para>
-         <replaceable class="parameter">detailed_explanation</replaceable> includes
-         the origin, transaction ID, and commit timestamp of the transaction that
-         modified the local row, if available.
-        </para>
-       </listitem>
-       <listitem>
-        <para>
-         The <literal>key</literal> section includes the key values of the local
-         row that violated a unique constraint for
-         <literal>insert_exists</literal>, <literal>update_exists</literal> or
-         <literal>multiple_unique_conflicts</literal> conflicts.
-        </para>
-       </listitem>
-       <listitem>
-        <para>
-         The <literal>local row</literal> section includes the local row if its
-         origin differs from the remote row for
-         <literal>update_origin_differs</literal> or <literal>delete_origin_differs</literal>
-         conflicts, or if the key value conflicts with the remote row for
-         <literal>insert_exists</literal>, <literal>update_exists</literal> or
-         <literal>multiple_unique_conflicts</literal> conflicts.
-        </para>
-       </listitem>
-       <listitem>
-        <para>
-         The <literal>remote row</literal> section includes the new row from
-         the remote insert or update operation that caused the conflict. Note that
-         for an update operation, the column value of the new row will be null
-         if the value is unchanged and toasted.
-        </para>
-       </listitem>
-       <listitem>
-        <para>
-         The <literal>replica identity</literal> section includes the replica
-         identity key values that were used to search for the existing local
-         row to be updated or deleted. This may include the full row value
-         if the local relation is marked with
-         <link linkend="sql-altertable-replica-identity-full"><literal>REPLICA IDENTITY FULL</literal></link>.
-        </para>
-       </listitem>
-       <listitem>
-        <para>
-         <replaceable class="parameter">column_name</replaceable> is the column name.
-         For <literal>local row</literal>, <literal>remote row</literal>, and
-         <literal>replica identity full</literal> cases, column names are
-         logged only if the user lacks the privilege to access all columns of
-         the table. If column names are present, they appear in the same order
-         as the corresponding column values.
-        </para>
-       </listitem>
-       <listitem>
-        <para>
-         <replaceable class="parameter">column_value</replaceable> is the column value.
-         The large column values are truncated to 64 bytes.
-        </para>
-       </listitem>
-       <listitem>
-        <para>
-         Note that in case of <literal>multiple_unique_conflicts</literal> conflict,
-         multiple <replaceable class="parameter">detailed_explanation</replaceable>
-         and <replaceable class="parameter">detail_values</replaceable> lines
-         will be generated, each detailing the conflict information associated
-         with distinct unique
-         constraints.
-        </para>
-       </listitem>
-      </itemizedlist>
-     </listitem>
-    </varlistentry>
-   </variablelist>
-  </para>
+     </varlistentry>
+    </variablelist>
+   </para>
+   </sect3>
+  </sect2>
 
-  <para>
-   Logical replication operations are performed with the privileges of the role
-   which owns the subscription.  Permissions failures on target tables will
-   cause replication conflicts, as will enabled
-   <link linkend="ddl-rowsecurity">row-level security</link> on target tables
-   that the subscription owner is subject to, without regard to whether any
-   policy would ordinarily reject the <command>INSERT</command>,
-   <command>UPDATE</command>, <command>DELETE</command> or
-   <command>TRUNCATE</command> which is being replicated.  This restriction on
-   row-level security may be lifted in a future version of
-   <productname>PostgreSQL</productname>.
-  </para>
+  <sect2 id="logical-replication-conflict-notes">
+   <title>Notes</title>
+   <para>
+    Logical replication operations are performed with the privileges of the role
+    which owns the subscription.  Permissions failures on target tables will
+    cause replication conflicts, as will enabled
+    <link linkend="ddl-rowsecurity">row-level security</link> on target tables
+    that the subscription owner is subject to, without regard to whether any
+    policy would ordinarily reject the <command>INSERT</command>,
+    <command>UPDATE</command>, <command>DELETE</command> or
+    <command>TRUNCATE</command> which is being replicated.  This restriction on
+    row-level security may be lifted in a future version of
+    <productname>PostgreSQL</productname>.
+   </para>
 
-  <para>
-   A conflict that produces an error will stop the replication; it must be
-   resolved manually by the user.  Details about the conflict can be found in
-   the subscriber's server log.
-  </para>
+   <para>
+    A conflict that produces an error will stop the replication; it must be
+    resolved manually by the user.  Details about the conflict can be found in
+    the subscriber's server log.
+   </para>
 
-  <para>
-   The resolution can be done either by changing data or permissions on the subscriber so
-   that it does not conflict with the incoming change or by skipping the
-   transaction that conflicts with the existing data.  When a conflict produces
-   an error, the replication won't proceed, and the logical replication worker will
-   emit the following kind of message to the subscriber's server log:
+   <para>
+    The resolution can be done either by changing data or permissions on the subscriber so
+    that it does not conflict with the incoming change or by skipping the
+    transaction that conflicts with the existing data.  When a conflict produces
+    an error, the replication won't proceed, and the logical replication worker will
+    emit the following kind of message to the subscriber's server log:
 <screen>
 ERROR:  conflict detected on relation "public.test": conflict=insert_exists
 DETAIL:  Could not apply remote change: remote row (1, 'remote').
 Key already exists in unique index "test_pkey", modified locally in transaction 800 at 2026-01-16 18:15:25.652759+09: key (c)=(1), local row (1, 'local').
 CONTEXT:  processing remote data for replication origin "pg_16395" during "INSERT" for replication target relation "public.test" in transaction 725 finished at 0/014C0378
 </screen>
-   The LSN of the transaction that contains the change violating the constraint and
-   the replication origin name can be found from the server log (LSN 0/014C0378 and
-   replication origin <literal>pg_16395</literal> in the above case).  The
-   transaction that produced the conflict can be skipped by using
-   <link linkend="sql-altersubscription-params-skip"><command>ALTER SUBSCRIPTION ... SKIP</command></link>
-   with the finish LSN
-   (i.e., LSN 0/014C0378).  The finish LSN could be an LSN at which the transaction
-   is committed or prepared on the publisher.  Alternatively, the transaction can
-   also be skipped by calling the <link linkend="pg-replication-origin-advance">
-   <function>pg_replication_origin_advance()</function></link> function.
-   Before using this function, the subscription needs to be disabled temporarily
-   either by <link linkend="sql-altersubscription-params-disable">
-   <command>ALTER SUBSCRIPTION ... DISABLE</command></link> or, the
-   subscription can be used with the
-   <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>
-   option. Then, you can use <function>pg_replication_origin_advance()</function>
-   function with the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>)
-   and the next LSN of the finish LSN (i.e., 0/014C0379).  The current position of
-   origins can be seen in the <link linkend="view-pg-replication-origin-status">
-   <structname>pg_replication_origin_status</structname></link> system view.
-   Please note that skipping the whole transaction includes skipping changes that
-   might not violate any constraint.  This can easily make the subscriber
-   inconsistent.
-   The additional details regarding conflicting rows, such as their origin and
-   commit timestamp can be seen in the <literal>DETAIL</literal> line of the
-   log. But note that this information is only available when
-   <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
-   is enabled on the subscriber. Users can use this information to decide
-   whether to retain the local change or adopt the remote alteration. For
-   instance, the <literal>DETAIL</literal> line in the above log indicates that
-   the existing row was modified locally. Users can manually perform a
-   remote-change-win.
-  </para>
-
-  <para>
-   When the
-   <link linkend="sql-createsubscription-params-with-streaming"><literal>streaming</literal></link>
-   mode is <literal>parallel</literal>, the finish LSN of failed transactions
-   may not be logged. In that case, it may be necessary to change the streaming
-   mode to <literal>on</literal> or <literal>off</literal> and cause the same
-   conflicts again so the finish LSN of the failed transaction will be written
-   to the server log. For the usage of finish LSN, please refer to <link
-   linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
-   SKIP</command></link>.
-  </para>
+    The LSN of the transaction that contains the change violating the constraint and
+    the replication origin name can be found from the server log (LSN 0/014C0378 and
+    replication origin <literal>pg_16395</literal> in the above case).  The
+    transaction that produced the conflict can be skipped by using
+    <link linkend="sql-altersubscription-params-skip"><command>ALTER SUBSCRIPTION ... SKIP</command></link>
+    with the finish LSN
+    (i.e., LSN 0/014C0378).  The finish LSN could be an LSN at which the transaction
+    is committed or prepared on the publisher.  Alternatively, the transaction can
+    also be skipped by calling the <link linkend="pg-replication-origin-advance">
+    <function>pg_replication_origin_advance()</function></link> function.
+    Before using this function, the subscription needs to be disabled temporarily
+    either by <link linkend="sql-altersubscription-params-disable">
+    <command>ALTER SUBSCRIPTION ... DISABLE</command></link> or, the
+    subscription can be used with the
+    <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>
+    option. Then, you can use <function>pg_replication_origin_advance()</function>
+    function with the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>)
+    and the next LSN of the finish LSN (i.e., 0/014C0379).  The current position of
+    origins can be seen in the <link linkend="view-pg-replication-origin-status">
+    <structname>pg_replication_origin_status</structname></link> system view.
+    Please note that skipping the whole transaction includes skipping changes that
+    might not violate any constraint.  This can easily make the subscriber
+    inconsistent.
+    The additional details regarding conflicting rows, such as their origin and
+    commit timestamp can be seen in the <literal>DETAIL</literal> line of the
+    log. But note that this information is only available when
+    <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+    is enabled on the subscriber. Users can use this information to decide
+    whether to retain the local change or adopt the remote alteration. For
+    instance, the <literal>DETAIL</literal> line in the above log indicates that
+    the existing row was modified locally. Users can manually perform a
+    remote-change-win.
+   </para>
+
+   <para>
+    When the
+    <link linkend="sql-createsubscription-params-with-streaming"><literal>streaming</literal></link>
+    mode is <literal>parallel</literal>, the finish LSN of failed transactions
+    may not be logged. In that case, it may be necessary to change the streaming
+    mode to <literal>on</literal> or <literal>off</literal> and cause the same
+    conflicts again so the finish LSN of the failed transaction will be written
+    to the server log. For the usage of finish LSN, please refer to <link
+    linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
+    SKIP</command></link>.
+   </para>
+  </sect2>
  </sect1>
 
  <sect1 id="logical-replication-restrictions">
@@ -2448,6 +2599,14 @@ CONTEXT:  processing remote data for replication origin "pg_16395" during "INSER
      key or replica identity defined for it.
     </para>
    </listitem>
+
+   <listitem>
+    <para>
+     Conflict log tables (see <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link> parameter)
+     are never published, even when using <literal>FOR ALL TABLES</literal> in a
+     publication.
+    </para>
+   </listitem>
   </itemizedlist>
  </sect1>
 
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 545264e8a0a..d1c49c2c7c4 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -366,7 +366,8 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       When the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
       parameter is set to <literal>table</literal> or <literal>all</literal>, the system
       automatically creates the conflict log table. Conversely, if the destination is changed to
-      <literal>log</literal>, the conflict log table is automatically dropped.
+      <literal>log</literal>, logging to the table stops and the conflict log
+      table is automatically dropped.
      </para>
     </listitem>
    </varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 25a81e2e62a..4803344839e 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -272,8 +272,10 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
             <term><literal>log</literal></term>
             <listitem>
              <para>
-              Conflict details are recorded in the server log.
-              This is the default behavior.
+              Conflict details are recorded in the server log. This is
+              the default behavior. See
+              <xref linkend="logical-replication-conflict-file-based-logging"/>
+              for details.
              </para>
             </listitem>
            </varlistentry>
@@ -283,15 +285,20 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
              <para>
               The system automatically creates a structured table named
               <literal>pg_conflict_log_&lt;subid&gt;</literal> in the
-              <literal>pg_conflict</literal> schema. This allows for easy
-              querying and analysis of conflicts.
+              <literal>pg_conflict</literal> schema to record resolved
+              conflicts. Conflicts raised at <literal>ERROR</literal> level
+              (which halt replication and abort the transaction) are not
+              recorded in the table; they are always reported to the server log.
+              See <xref linkend="logical-replication-conflict-table-based-logging"/>
+              for details.
              </para>
              <caution>
               <para>
                The conflict log table is strictly tied to the lifecycle of the
                subscription or the <literal>conflict_log_destination</literal> setting. If
                the subscription is dropped, or if the destination is changed to
-               <literal>log</literal>, the table is <emphasis>dropped</emphasis>.
+               <literal>log</literal>, the table and all its recorded conflict data are
+               <emphasis>permanently deleted</emphasis>.
               </para>
              </caution>
             </listitem>
@@ -300,8 +307,9 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
             <term><literal>all</literal></term>
             <listitem>
              <para>
-              This is equivalent to configuring both destinations <literal>log</literal>
-              and <literal>table</literal> together.
+              Records resolved conflict details to both the server log and the
+              conflict log table. Conflicts raised at <literal>ERROR</literal> level
+              are reported to the server log alone.
              </para>
             </listitem>
            </varlistentry>
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index f0a6ed20c60..ddbff5f764a 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -16,17 +16,22 @@
 
 #include "access/commit_ts.h"
 #include "access/genam.h"
+#include "access/heapam.h"
 #include "access/tableam.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/pg_lsn.h"
 
 /*
  * String representations for the supported conflict logging destinations.
@@ -63,6 +68,11 @@ typedef struct ConflictLogColumnDef
  * advantage) wouldn't help anyway, as the conflict log is looked up by its
  * scalar columns (relid, conflict_type, commit timestamp) while these json
  * columns are per-conflict payload to inspect, not search keys.
+ *
+ * 'local_conflicts' is typed as an array of JSON objects (json[]), not a
+ * single json object, so that a future conflict type needing to record
+ * multiple local rows for one remote operation doesn't require a
+ * backward-incompatible schema change.
  */
 static const ConflictLogColumnDef ConflictLogSchema[] = {
 	{.attname = "relid", .atttypid = OIDOID},
@@ -81,6 +91,19 @@ static const ConflictLogColumnDef ConflictLogSchema[] = {
 
 #define NUM_CONFLICT_ATTRS ((AttrNumber) 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 = "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",
@@ -108,8 +131,25 @@ 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(List *conflicttuples);
+static void insert_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.
@@ -279,29 +319,108 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
 					TupleTableSlot *remoteslot, List *conflicttuples)
 {
 	Relation	localrel = relinfo->ri_RelationDesc;
-	StringInfoData err_detail;
+	ConflictLogDest dest;
+	Relation	conflictlogrel = NULL;
+	bool		log_dest_table = false;
+	bool		log_dest_logfile = false;
 
-	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);
+	/*
+	 * Only LOG-level conflicts (i.e. resolved conflicts where the transaction
+	 * continues) are recorded in the conflict log table.  ERROR-level
+	 * conflicts halt replication and abort the transaction, so they are
+	 * always reported exclusively to the server log.
+	 */
+	if (elevel < ERROR)
+	{
+		conflictlogrel = GetConflictLogDestAndTable(&dest);
+		log_dest_table = CONFLICTS_LOGGED_TO_TABLE(dest);
+		log_dest_logfile = CONFLICTS_LOGGED_TO_LOG(dest);
 
-	pgstat_report_subscription_conflict(MySubscription->oid, type);
+		/*
+		 * If a conflict log table was requested but it has been dropped
+		 * concurrently (e.g. a concurrent ALTER SUBSCRIPTION changed
+		 * conflict_log_destination), GetConflictLogDestAndTable() returned
+		 * NULL.  Fall back to logging to the server log so that the conflict
+		 * is not lost.
+		 */
+		if (log_dest_table && conflictlogrel == NULL)
+		{
+			log_dest_table = false;
+			log_dest_logfile = true;
+		}
+	}
+	else
+	{
+		/* ERROR-level conflicts always go to the server log with full detail */
+		log_dest_logfile = true;
+	}
+
+	/*
+	 * Report the conflict to the server log.  When the server log is one of
+	 * the destinations (or for ERROR-level conflicts), emit the full details.
+	 * Otherwise (table-only for LOG-level conflicts), emit a shorter message
+	 * noting that the details are captured in the conflict log table.
+	 */
+	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));
+	}
+	else if (log_dest_table)
+	{
+		/*
+		 * Not logging conflict details to the server log; report the conflict
+		 * but omit raw tuple data since it is 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",
+						  RelationGetQualifiedRelationName(conflictlogrel)));
+	}
 
-	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));
+	/*
+	 * Insert into the conflict log table for LOG-level conflicts if
+	 * requested.
+	 */
+	if (log_dest_table)
+	{
+		Assert(conflictlogrel != NULL);
+		Assert(elevel < ERROR);
+
+		insert_conflict_log_tuple(estate,
+								  relinfo->ri_RelationDesc,
+								  conflictlogrel,
+								  type,
+								  searchslot,
+								  conflicttuples,
+								  remoteslot);
+		table_close(conflictlogrel, NoLock);
+	}
 }
 
 /*
@@ -335,6 +454,47 @@ 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 a table is one of the
+ * destinations, opens and returns the relation handle for the conflict log
+ * table.
+ *
+ * The table is opened with try_table_open(), so NULL is returned if the
+ * conflict log table has been dropped concurrently (e.g. by an ALTER
+ * SUBSCRIPTION that changed conflict_log_destination).  Callers must treat a
+ * NULL result for a table destination as "table unavailable" and fall back to
+ * server-log reporting rather than failing.
+ */
+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));
+
+	/*
+	 * Use try_table_open(): the table may have been dropped concurrently by
+	 * an ALTER SUBSCRIPTION that changed conflict_log_destination.  Returning
+	 * NULL lets the caller fall back to the server log instead of failing.
+	 */
+	return try_table_open(conflictlogrelid, RowExclusiveLock);
+}
+
 /*
  * Add SQLSTATE error code to the current conflict report.
  */
@@ -768,6 +928,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.
@@ -783,41 +977,332 @@ 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);
+	index_close(indexDesc, NoLock);
+
+	/* Convert to a JSON datum. */
+	datum = DirectFunctionCall1(row_to_json, datum);
+	FreeTupleDesc(tupdesc);
+
+	return 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(List *conflicttuples)
+{
+	Datum	   *json_datum_array;
+	Datum		json_array_datum;
+	int			num_conflicts;
+	int			i = 0;
+	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);
+	num_conflicts = list_length(conflicttuples);
+	json_datum_array = palloc_array(Datum, num_conflicts);
 
-	return index_value;
+	/* 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		datum;
+		Datum		json_datum;
+		int			attno;
+
+		attno = 0;
+		if (TransactionIdIsValid(conflicttuple->xmin))
+			values[attno++] = TransactionIdGetDatum(conflicttuple->xmin);
+		else
+			nulls[attno++] = true;
+
+		if (conflicttuple->ts)
+			values[attno++] = TimestampTzGetDatum(conflicttuple->ts);
+		else
+			nulls[attno++] = true;
+
+		if (conflicttuple->origin != InvalidReplOriginId)
+			replorigin_by_oid(conflicttuple->origin, true, &origin_name);
+
+		/*
+		 * Set NULL if origin name for the tuple is InvalidReplOriginId or not
+		 * found.
+		 */
+		if (origin_name != NULL)
+			values[attno++] = CStringGetTextDatum(origin_name);
+		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 == NUM_LOCAL_CONFLICT_ATTRS);
+
+		tuple = heap_form_tuple(tupdesc, values, nulls);
+
+		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, datum);
+
+		/* Done with the temporary tuple. */
+		heap_freetuple(tuple);
+
+		/* Add to the array element. */
+		json_datum_array[i++] = json_datum;
+	}
+
+	Assert(i == num_conflicts);
+
+	/* 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;
+}
+
+/*
+ * insert_conflict_log_tuple
+ *
+ * Prepares and inserts a tuple detailing a conflict encountered during
+ * logical replication into the conflict log table.
+ */
+static void
+insert_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;
+	TransactionId remote_xid;
+	XLogRecPtr	remote_final_lsn;
+	TimestampTz remote_commit_ts;
+	HeapTuple	tuple;
+
+	/* Fetch the information of the remote transaction being applied. */
+	GetRemoteTransactionInfoForConflict(&remote_xid, &remote_final_lsn,
+										&remote_commit_ts);
+
+	/* 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;
+
+	if (XLogRecPtrIsValid(remote_final_lsn))
+		values[attno++] = LSNGetDatum(remote_final_lsn);
+	else
+		nulls[attno++] = true;
+
+	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(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++] = BoolGetDatum(false);
+			values[attno++] = tuple_table_slot_to_indextup_json(estate, rel,
+																replica_index,
+																searchslot);
+		}
+		else
+		{
+			values[attno++] = BoolGetDatum(true);
+			values[attno++] = tuple_table_slot_to_json_datum(searchslot);
+		}
+	}
+	else
+	{
+		nulls[attno++] = true;
+		nulls[attno++] = true;
+	}
+
+	if (!TupIsNull(remoteslot))
+		values[attno++] = tuple_table_slot_to_json_datum(remoteslot);
+	else
+		nulls[attno++] = true;
+
+	values[attno] = build_local_conflicts_json_array(conflicttuples);
+
+	Assert(attno + 1 == NUM_CONFLICT_ATTRS);
+
+	tuple = heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls);
+
+	heap_insert(conflictlogrel, tuple,
+				GetCurrentCommandId(true), 0, NULL);
+
+	heap_freetuple(tuple);
 }
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index d4ff81bd23e..9d8ce8d4917 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -328,14 +328,21 @@ typedef struct ApplyExecutionData
  * Context describing the remote transaction whose changes are currently
  * being applied, and the change within it.
  *
- * The remote transaction information (remote_xid and finish_lsn) is set when
- * the transaction's changes begin to be applied and is used not only for
- * error context reporting (see apply_error_callback), but also to decide
- * whether a change should be applied at all (see
- * should_apply_changes_for_rel) and to validate the transaction finish
- * messages.  finish_lsn is invalid when the final LSN of the remote
- * transaction is not yet known (e.g. while streaming an in-progress
- * transaction).
+ * The remote transaction information (remote_xid, finish_lsn and commit_ts)
+ * is set when the transaction's changes begin to be applied and is used not
+ * only for error context reporting (see apply_error_callback), but also to
+ * decide whether a change should be applied at all (see
+ * should_apply_changes_for_rel), to validate the transaction finish
+ * messages, and to populate the conflict log table if a conflict is
+ * detected while applying the transaction (see
+ * GetRemoteTransactionInfoForConflict).
+ *
+ * finish_lsn is invalid when the final LSN of the remote transaction is not
+ * yet known (e.g. while streaming an in-progress transaction).  Similarly,
+ * commit_ts is 0 when the remote commit timestamp is not yet known; for a
+ * streamed transaction it is filled in only if and when the leader apply
+ * worker processes the commit record, so it remains 0 for conflicts handled
+ * by a parallel apply worker.
  *
  * The remaining fields describe the individual change being applied and are
  * used only for error context reporting.
@@ -349,6 +356,7 @@ typedef struct ApplyRemoteCtx
 	int			remote_attnum;	/* -1 if invalid */
 	TransactionId remote_xid;
 	XLogRecPtr	finish_lsn;
+	TimestampTz commit_ts;		/* remote commit ts; 0 if unknown */
 	char	   *origin_name;
 } ApplyRemoteCtx;
 
@@ -483,6 +491,7 @@ static ApplyRemoteCtx remote_ctx =
 	.remote_attnum = -1,
 	.remote_xid = InvalidTransactionId,
 	.finish_lsn = InvalidXLogRecPtr,
+	.commit_ts = 0,
 	.origin_name = NULL,
 };
 
@@ -641,7 +650,9 @@ static void stop_skipping_changes(void);
 static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn);
 
 /* Functions to maintain the context of the remote transaction being applied */
-static inline void set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn);
+static inline void set_remote_transaction_info(TransactionId xid,
+											   XLogRecPtr lsn,
+											   TimestampTz commit_ts);
 static inline void reset_apply_remote_context(void);
 
 static TransApplyAction get_transaction_apply_action(TransactionId xid,
@@ -1249,7 +1260,8 @@ apply_handle_begin(StringInfo s)
 	Assert(!TransactionIdIsValid(stream_xid));
 
 	logicalrep_read_begin(s, &begin_data);
-	set_remote_transaction_info(begin_data.xid, begin_data.final_lsn);
+	set_remote_transaction_info(begin_data.xid, begin_data.final_lsn,
+								begin_data.committime);
 
 	maybe_start_skipping_changes(begin_data.final_lsn);
 
@@ -1307,7 +1319,12 @@ apply_handle_begin_prepare(StringInfo s)
 	Assert(!TransactionIdIsValid(stream_xid));
 
 	logicalrep_read_begin_prepare(s, &begin_data);
-	set_remote_transaction_info(begin_data.xid, begin_data.prepare_lsn);
+
+	/*
+	 * The commit timestamp is known only when the commit record arrives, so
+	 * it is set to 0 for now.
+	 */
+	set_remote_transaction_info(begin_data.xid, begin_data.prepare_lsn, 0);
 
 	maybe_start_skipping_changes(begin_data.prepare_lsn);
 
@@ -1437,7 +1454,8 @@ apply_handle_commit_prepared(StringInfo s)
 	char		gid[GIDSIZE];
 
 	logicalrep_read_commit_prepared(s, &prepare_data);
-	set_remote_transaction_info(prepare_data.xid, prepare_data.commit_lsn);
+	set_remote_transaction_info(prepare_data.xid, prepare_data.commit_lsn,
+								prepare_data.commit_time);
 
 	/* Compute GID for two_phase transactions. */
 	TwoPhaseTransactionGid(MySubscription->oid, prepare_data.xid,
@@ -1489,7 +1507,8 @@ apply_handle_rollback_prepared(StringInfo s)
 	char		gid[GIDSIZE];
 
 	logicalrep_read_rollback_prepared(s, &rollback_data);
-	set_remote_transaction_info(rollback_data.xid, rollback_data.rollback_end_lsn);
+	set_remote_transaction_info(rollback_data.xid,
+								rollback_data.rollback_end_lsn, 0);
 
 	/* Compute GID for two_phase transactions. */
 	TwoPhaseTransactionGid(MySubscription->oid, rollback_data.xid,
@@ -1565,7 +1584,7 @@ apply_handle_stream_prepare(StringInfo s)
 				 errmsg_internal("tablesync worker received a STREAM PREPARE message")));
 
 	logicalrep_read_stream_prepare(s, &prepare_data);
-	set_remote_transaction_info(prepare_data.xid, prepare_data.prepare_lsn);
+	set_remote_transaction_info(prepare_data.xid, prepare_data.prepare_lsn, 0);
 
 	apply_action = get_transaction_apply_action(prepare_data.xid, &winfo);
 
@@ -1780,10 +1799,10 @@ apply_handle_stream_start(StringInfo s)
 				 errmsg_internal("invalid transaction ID in streamed replication transaction")));
 
 	/*
-	 * The final LSN of the streamed transaction is known only when its commit
-	 * record arrives.
+	 * The final LSN and commit timestamp of the streamed transaction are
+	 * known only when its commit record arrives.
 	 */
-	set_remote_transaction_info(stream_xid, InvalidXLogRecPtr);
+	set_remote_transaction_info(stream_xid, InvalidXLogRecPtr, 0);
 
 	/* Try to allocate a worker for the streaming transaction. */
 	if (first_segment)
@@ -2131,7 +2150,7 @@ apply_handle_stream_abort(StringInfo s)
 	 * names the subtransaction being aborted.  See
 	 * set_remote_transaction_info().
 	 */
-	set_remote_transaction_info(subxid, abort_data.abort_lsn);
+	set_remote_transaction_info(subxid, abort_data.abort_lsn, 0);
 
 	apply_action = get_transaction_apply_action(xid, &winfo);
 
@@ -2446,7 +2465,8 @@ apply_handle_stream_commit(StringInfo s)
 				 errmsg_internal("STREAM COMMIT message without STREAM STOP")));
 
 	xid = logicalrep_read_stream_commit(s, &commit_data);
-	set_remote_transaction_info(xid, commit_data.commit_lsn);
+	set_remote_transaction_info(xid, commit_data.commit_lsn,
+								commit_data.committime);
 
 	apply_action = get_transaction_apply_action(xid, &winfo);
 
@@ -6387,8 +6407,10 @@ apply_error_callback(void *arg)
  * Set information identifying the remote transaction currently being
  * applied, kept for the duration of that transaction.  Besides naming the
  * transaction in the error context, this decides whether a change is applied
- * at all (see should_apply_changes_for_rel) and validates the transaction
- * finish messages.
+ * at all (see should_apply_changes_for_rel), validates the transaction
+ * finish messages, and populates the conflict log table if a conflict is
+ * detected while applying the transaction (see
+ * GetRemoteTransactionInfoForConflict).
  *
  * This must be called for every message type that begins or resumes applying
  * a remote transaction's changes (BEGIN, BEGIN PREPARE, STREAM START, STREAM
@@ -6402,13 +6424,44 @@ apply_error_callback(void *arg)
  * context should name the subtransaction that failed.  Nothing else observes
  * those, because no change is applied between a STREAM ABORT and the STREAM
  * START or STREAM COMMIT/PREPARE that follows it, and each of those calls
- * this again with the transaction's own values.
+ * this again with the transaction's own values.  The assertion in
+ * GetRemoteTransactionInfoForConflict() enforces that for conflict logging.
+ *
+ * Callers pass the remote commit timestamp when it is known (from the BEGIN
+ * or STREAM COMMIT message) and 0 otherwise.
  */
 static inline void
-set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn)
+set_remote_transaction_info(TransactionId xid, XLogRecPtr lsn,
+							TimestampTz commit_ts)
 {
 	remote_ctx.remote_xid = xid;
 	remote_ctx.finish_lsn = lsn;
+	remote_ctx.commit_ts = commit_ts;
+}
+
+/*
+ * Return information identifying the remote transaction currently being
+ * applied, for use by conflict logging (see insert_conflict_log_tuple() in
+ * conflict.c).  See set_remote_transaction_info().
+ */
+void
+GetRemoteTransactionInfoForConflict(TransactionId *remote_xid,
+									XLogRecPtr *finish_lsn,
+									TimestampTz *commit_ts)
+{
+	/*
+	 * Conflicts are only detected while applying a row change, which
+	 * guarantees the recorded values describe the top-level transaction.  In
+	 * particular they are never the subxid and abort LSN recorded by a
+	 * subtransaction abort.  See set_remote_transaction_info().
+	 */
+	Assert(remote_ctx.command == LOGICAL_REP_MSG_INSERT ||
+		   remote_ctx.command == LOGICAL_REP_MSG_UPDATE ||
+		   remote_ctx.command == LOGICAL_REP_MSG_DELETE);
+
+	*remote_xid = remote_ctx.remote_xid;
+	*finish_lsn = remote_ctx.finish_lsn;
+	*commit_ts = remote_ctx.commit_ts;
 }
 
 /* Reset all information of the remote transaction context */
@@ -6418,7 +6471,7 @@ reset_apply_remote_context(void)
 	remote_ctx.command = 0;
 	remote_ctx.rel = NULL;
 	remote_ctx.remote_attnum = -1;
-	set_remote_transaction_info(InvalidTransactionId, InvalidXLogRecPtr);
+	set_remote_transaction_info(InvalidTransactionId, InvalidXLogRecPtr, 0);
 }
 
 /*
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 326570ed639..8b66b1876e1 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -12,6 +12,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "nodes/pg_list.h"
+#include "utils/relcache.h"
 
 /* Avoid including execnodes.h here */
 typedef struct EState EState;
@@ -115,4 +116,5 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
 								TupleTableSlot *remoteslot,
 								List *conflicttuples);
 extern void InitConflictIndexes(ResultRelInfo *relInfo);
+extern Relation GetConflictLogDestAndTable(ConflictLogDest *log_dest);
 #endif
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 88cb7c1e252..e2e9ca1f708 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -256,6 +256,11 @@ extern PGDLLIMPORT bool InitializingApplyWorker;
 
 extern PGDLLIMPORT List *table_states_not_ready;
 
+/* Remote transaction information of the transaction being applied */
+extern void GetRemoteTransactionInfoForConflict(TransactionId *remote_xid,
+												XLogRecPtr *finish_lsn,
+												TimestampTz *commit_ts);
+
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype,
 												Oid subid, Oid relid,
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 41ab4586c6b..e38e31c1b58 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -714,16 +714,19 @@ RelationCloseSmgr(Relation relation)
  *
  * We don't log information for unlogged tables (since they don't WAL log
  * anyway), for foreign tables (since they don't WAL log, either),
- * and for system tables (their content is hard to make sense of, and
- * it would complicate decoding slightly for little gain). Note that we *do*
- * log information for user defined catalog tables since they presumably are
- * interesting to the user...
+ * for system tables (their content is hard to make sense of, and
+ * it would complicate decoding slightly for little gain), and for conflict
+ * log tables in the pg_conflict namespace (which are system-managed tables
+ * used internally on subscribers). Note that we *do* log information for
+ * user defined catalog tables since they presumably are interesting to the
+ * user...
  */
 #define RelationIsLogicallyLogged(relation) \
 	(XLogLogicalInfoActive() && \
 	 RelationNeedsWAL(relation) && \
 	 (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&	\
-	 !IsCatalogRelation(relation))
+	 !IsCatalogRelation(relation) && \
+	 !IsConflictLogTableNamespace(RelationGetNamespace(relation)))
 
 /* routines in utils/cache/relcache.c */
 extern void RelationIncrementReferenceCount(Relation rel);
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 3910a49c0aa..5efef6614ef 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -84,6 +84,17 @@ $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)
+# ERROR-level conflicts halt replication and are reported only to server logs,
+# so the CLT should remain empty.
+my $subid = $node_subscriber->safe_psql('postgres',
+	"SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';");
+my $clt = "pg_conflict.pg_conflict_log_$subid";
+
+my $conflict_count = $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM $clt;");
+is($conflict_count, '0', 'Verified multiple_unique_conflicts is not logged into conflict log table');
+
 pass('multiple_unique_conflicts detected during insert');
 
 # Truncate table to get rid of the error
@@ -114,6 +125,11 @@ $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 that the CLT still does not contain ERROR-level conflicts
+$conflict_count = $node_subscriber->safe_psql('postgres',
+	"SELECT count(*) FROM $clt;");
+is($conflict_count, '0', 'Verified multiple_unique_conflicts during UPDATE is not logged into conflict log table');
+
 pass('multiple_unique_conflicts detected during update');
 
 # Truncate table to get rid of the error
@@ -183,7 +199,7 @@ $node_B->safe_psql(
 	CREATE SUBSCRIPTION $subname_BA
 	CONNECTION '$node_A_connstr application_name=$subname_BA'
 	PUBLICATION tap_pub_A
-	WITH (origin = none, retain_dead_tuples = true)");
+	WITH (origin = none, retain_dead_tuples = true, conflict_log_destination = 'all')");
 
 # node_B (pub) -> node_A (sub)
 my $node_B_connstr = $node_B->connstr . ' dbname=postgres';
@@ -193,7 +209,7 @@ $node_A->safe_psql(
 	CREATE SUBSCRIPTION $subname_AB
 	CONNECTION '$node_B_connstr application_name=$subname_AB'
 	PUBLICATION tap_pub_B
-	WITH (origin = none, copy_data = off)");
+	WITH (origin = none, copy_data = off, conflict_log_destination = 'all')");
 
 # Wait for initial table sync to finish
 $node_A->wait_for_subscription_sync($node_B, $subname_AB);
@@ -316,6 +332,13 @@ like(
 .*DETAIL:.* Deleting the row that was modified locally in transaction [0-9]+ at .*: local row \(1, 3\), replica identity \(a\)=\(1\)./,
 	'delete target row was modified in tab');
 
+my $subid_BA = $node_B->safe_psql('postgres',
+	"SELECT oid FROM pg_subscription WHERE subname = '$subname_BA';");
+my $clt_BA = "pg_conflict.pg_conflict_log_$subid_BA";
+my $clt_check_ba = $node_B->poll_query_until('postgres',
+	"SELECT count(*) > 0 FROM $clt_BA WHERE conflict_type = 'delete_origin_differs';");
+is($clt_check_ba, 1, 'delete_origin_differs logged into CLT on Node B');
+
 $log_location = -s $node_A->logfile;
 
 $node_A->safe_psql('postgres', "ALTER SUBSCRIPTION $subname_AB ENABLE;");
@@ -329,6 +352,13 @@ like(
 .*The row to be updated was deleted locally in transaction [0-9]+ at .*/,
 	'update target row was deleted in tab');
 
+my $subid_AB = $node_A->safe_psql('postgres',
+	"SELECT oid FROM pg_subscription WHERE subname = '$subname_AB';");
+my $clt_AB = "pg_conflict.pg_conflict_log_$subid_AB";
+my $clt_check_ab = $node_A->poll_query_until('postgres',
+	"SELECT count(*) > 0 FROM $clt_AB WHERE conflict_type = 'update_deleted';");
+is($clt_check_ab, 1, 'update_deleted logged into CLT on Node A');
+
 # Remember the next transaction ID to be assigned
 my $next_xid = $node_A->safe_psql('postgres', "SELECT txid_current() + 1;");
 
@@ -674,10 +704,6 @@ ok( $node_A->poll_query_until(
 # A conflict log table is system-managed and cannot be altered directly, so
 # moving it to another tablespace must be rejected.
 ###############################################################################
-my $subid = $node_subscriber->safe_psql('postgres',
-	"SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';");
-my $clt = "pg_conflict.pg_conflict_log_$subid";
-
 (undef, undef, $stderr) = $node_subscriber->psql('postgres',
 	"ALTER TABLE $clt SET TABLESPACE pg_default");
 like(
-- 
2.43.0

