From e02b058f2e932a3f7e8c80aea2f63b59e419366f Mon Sep 17 00:00:00 2001
From: Ajin Cherian <itsajin@gmail.com>
Date: Mon, 14 Oct 2024 05:35:35 -0400
Subject: [PATCH v17 1/4] Add CONFLICT RESOLVERS into the syntax for CREATE and
 ALTER SUBSCRIPTION

This patch provides support for configuring subscriptions with conflict resolvers.

Syntax for CREATE SUBSCRIPTION:
CREATE SUBSCRIPTION <subname> CONNECTION <conninfo> PUBLICATION <pubname> CONFLICT RESOLVER
    (conflict_type1 = resolver1, conflict_type2 = resolver2, conflict_type3 = resolver3,...);

Syntax for ALTER SUBSCRIPTION:
ALTER SUBSCRIPTION <subname> CONFLICT RESOLVER
    (conflict_type1 = resolver1, conflict_type2 = resolver2, conflict_type3 = resolver3,...);

Syntax for resetting conflict resolvers:
 ALTER SUBSCRIPTION <subname> RESET CONFLICT RESOLVER ALL

 ALTER SUBSCRIPTION <subname> RESET CONFLICT RESOLVER FOR conflict_type

Additionally, this patch includes documentation for conflict resolvers.
---
 doc/src/sgml/logical-replication.sgml          |  95 +-----
 doc/src/sgml/monitoring.sgml                   |  13 +-
 doc/src/sgml/ref/alter_subscription.sgml       |  63 ++++
 doc/src/sgml/ref/create_subscription.sgml      | 211 ++++++++++++
 src/backend/commands/subscriptioncmds.c        |  53 +++
 src/backend/parser/gram.y                      |  49 ++-
 src/backend/replication/logical/conflict.c     | 451 ++++++++++++++++++++++++-
 src/bin/pg_dump/pg_dump.c                      | 126 ++++++-
 src/bin/pg_dump/pg_dump.h                      |   8 +
 src/bin/pg_dump/t/002_pg_dump.pl               |   3 +-
 src/bin/psql/describe.c                        | 118 ++++++-
 src/bin/psql/tab-complete.in.c                 |  22 +-
 src/include/catalog/Makefile                   |   3 +-
 src/include/catalog/meson.build                |   1 +
 src/include/catalog/pg_subscription_conflict.h |  55 +++
 src/include/nodes/parsenodes.h                 |   7 +
 src/include/parser/kwlist.h                    |   1 +
 src/include/replication/conflict.h             |  49 ++-
 src/test/regress/expected/oidjoins.out         |   1 +
 src/test/regress/expected/subscription.out     | 362 ++++++++++++++++++--
 src/test/regress/sql/subscription.sql          |  63 ++++
 src/tools/pgindent/typedefs.list               |   3 +
 22 files changed, 1627 insertions(+), 130 deletions(-)
 create mode 100644 src/include/catalog/pg_subscription_conflict.h

diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 98a7ad0..86c9b0b 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1568,7 +1568,7 @@ test_sub=# SELECT * FROM t1 ORDER BY id;
  </sect1>
 
  <sect1 id="logical-replication-conflicts">
-  <title>Conflicts</title>
+  <title>Conflicts and Conflict Resolution</title>
 
   <para>
    Logical replication behaves similarly to normal DML operations in that
@@ -1582,86 +1582,19 @@ test_sub=# SELECT * FROM t1 ORDER BY id;
   </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-missing" xreflabel="update_missing">
-     <term><literal>update_missing</literal></term>
-     <listitem>
-      <para>
-       The tuple 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 tuple to be deleted was not found. The delete will simply be
-       skipped in this scenario.
-      </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.
+   There are various conflict scenarios, each identified as a <firstterm>conflict type</firstterm>.
+   Users can configure a <firstterm>conflict resolver</firstterm> for each
+   conflict type when creating a subscription. For more information, refer to
+   <link linkend="sql-createsubscription-params-with-conflict-resolver">
+   <command>CREATE SUBSCRIPTION ... CONFLICT RESOLVER</command></link>.
+  </para>
+  <para>
+   When a conflict occurs the details about it are logged, and the conflict
+   statistics are recorded in the <link linkend="monitoring-pg-stat-subscription-stats">
+   <structname>pg_stat_subscription_stats</structname></link> view.
+   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>
 
   <para>
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 331315f..d79db76 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2181,7 +2181,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       <para>
        Number of times a row insertion violated a
        <literal>NOT DEFERRABLE</literal> unique constraint during the
-       application of changes. See <xref linkend="conflict-insert-exists"/>
+       application of changes. See <xref linkend="sql-createsubscription-params-with-conflict_type-insert-exists"/>
        for details about this conflict.
       </para></entry>
      </row>
@@ -2193,7 +2193,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       <para>
        Number of times an update was applied to a row that had been previously
        modified by another source during the application of changes. See
-       <xref linkend="conflict-update-origin-differs"/> for details about this
+       <xref linkend="sql-createsubscription-params-with-conflict_type-update-origin-differs"/> for details about this
        conflict.
       </para></entry>
      </row>
@@ -2205,7 +2205,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       <para>
        Number of times that an updated row value violated a
        <literal>NOT DEFERRABLE</literal> unique constraint during the
-       application of changes. See <xref linkend="conflict-update-exists"/>
+       application of changes. See <xref linkend="sql-createsubscription-params-with-conflict_type-update-exists"/>
        for details about this conflict.
       </para></entry>
      </row>
@@ -2216,7 +2216,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        Number of times the tuple to be updated was not found during the
-       application of changes. See <xref linkend="conflict-update-missing"/>
+       application of changes. See <xref linkend="sql-createsubscription-params-with-conflict_type-update-missing
+"/>
        for details about this conflict.
       </para></entry>
      </row>
@@ -2228,7 +2229,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       <para>
        Number of times a delete operation was applied to row that had been
        previously modified by another source during the application of changes.
-       See <xref linkend="conflict-delete-origin-differs"/> for details about
+       See <xref linkend="sql-createsubscription-params-with-conflict_type-delete-origin-differs"/> for details about
        this conflict.
       </para></entry>
      </row>
@@ -2239,7 +2240,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
       </para>
       <para>
        Number of times the tuple to be deleted was not found during the application
-       of changes. See <xref linkend="conflict-delete-missing"/> for details
+       of changes. See <xref linkend="sql-createsubscription-params-with-conflict_type-delete-missing"/> for details
        about this conflict.
       </para></entry>
      </row>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index fdc648d..8ed0b93 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -32,6 +32,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> SET ( <repl
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> SKIP ( <replaceable class="parameter">skip_option</replaceable> = <replaceable class="parameter">value</replaceable> )
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> OWNER TO { <replaceable>new_owner</replaceable> | CURRENT_ROLE | CURRENT_USER | SESSION_USER }
 ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <replaceable>new_name</replaceable>
+ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> CONFLICT RESOLVER ( <replaceable class="parameter">conflict_type</replaceable> = <replaceable class="parameter">conflict_resolver</replaceable> [, ...] )
+ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RESET CONFLICT RESOLVER ALL
+ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RESET CONFLICT RESOLVER FOR '<replaceable class="parameter">conflict_type</replaceable>'
 </synopsis>
  </refsynopsisdiv>
 
@@ -345,6 +348,66 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
      </para>
     </listitem>
    </varlistentry>
+
+   <varlistentry id="sql-altersubscription-params-conflict-resolver">
+    <term><literal>CONFLICT RESOLVER ( <replaceable class="parameter">conflict_type</replaceable> = <replaceable class="parameter">conflict_resolver</replaceable> [, ... ] )</literal></term>
+    <listitem>
+     <para>
+      This clause alters the current conflict resolver for the specified conflict types.
+      Refer to <link linkend="sql-createsubscription-params-with-conflict-resolver">
+      <command>CREATE SUBSCRIBER ... CONFLICT RESOLVER</command></link>
+      for details about different conflict types and what kinds of resolver can
+      be assigned to them.
+     </para>
+     <variablelist>
+      <varlistentry id="sql-altersubscription-params-conflict-resolver-type">
+       <term><replaceable class="parameter">conflict_type</replaceable></term>
+       <listitem>
+        <para>
+         The conflict type.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry id="sql-altersubscription-params-conflict-resolver-resolver">
+       <term><replaceable class="parameter">conflict_resolver</replaceable></term>
+       <listitem>
+        <para>
+         The conflict resolver to use for this conflict type.
+        </para>
+       </listitem>
+      </varlistentry>
+     </variablelist>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry id="sql-altersubscription-params-reset-conflict-resolver">
+    <term><literal>RESET CONFLICT RESOLVER</literal></term>
+    <listitem>
+     <para>
+     Reset all conflict types, or a specified conflict type, to their default resolvers.
+     For details on conflict types and their default resolvers, refer to
+     section <link linkend="sql-createsubscription-params-with-conflict-resolver"><literal>CREATE SUBSCRIBER ... CONFLICT RESOLVER</literal></link>.
+     </para>
+     <variablelist>
+      <varlistentry id="sql-altersubscription-params-reset-all">
+       <term><literal>ALL</literal></term>
+       <listitem>
+        <para>
+         All conflict types will be reset to their respective default resolvers.
+        </para>
+       </listitem>
+      </varlistentry>
+      <varlistentry id="sql-altersubscription-params-reset-conflict-type">
+       <term><literal>FOR</literal> '<replaceable class="parameter">conflict_type</replaceable>'</term>
+       <listitem>
+        <para>
+         The given <replaceable class="parameter">conflict_type</replaceable> will be reset to its default resolver.
+        </para>
+       </listitem>
+      </varlistentry>
+     </variablelist>
+    </listitem>
+   </varlistentry>
   </variablelist>
 
   <para>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 8a3096e..06009c3 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -24,6 +24,7 @@ PostgreSQL documentation
 CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceable>
     CONNECTION '<replaceable class="parameter">conninfo</replaceable>'
     PUBLICATION <replaceable class="parameter">publication_name</replaceable> [, ...]
+    [ CONFLICT RESOLVER ( <replaceable class="parameter">conflict_type</replaceable> = <replaceable class="parameter">conflict_resolver</replaceable> [, ...] ) ]
     [ WITH ( <replaceable class="parameter">subscription_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
 </synopsis>
  </refsynopsisdiv>
@@ -97,6 +98,216 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
     </listitem>
    </varlistentry>
 
+   <varlistentry id="sql-createsubscription-params-with-conflict-resolver">
+    <term><literal>CONFLICT RESOLVER ( <replaceable class="parameter">conflict_type</replaceable> = <replaceable class="parameter">conflict_resolver</replaceable> [, ... ] )</literal></term>
+    <listitem>
+     <para>
+      This optional clause specifies conflict resolvers for different conflict types.
+     </para>
+
+     <table id="sql-createsubscription-params-conflict-type-resolver-summary">
+      <title>Conflict type/resolver Summary</title>
+      <tgroup cols="3">
+       <thead>
+        <row><entry>Conflict type</entry>        <entry>Default resolver</entry>  <entry>Possible resolvers</entry></row>
+       </thead>
+       <tbody>
+        <row><entry>insert_exists</entry>        <entry>error</entry>             <entry>apply_remote, error, keep_local</entry></row>
+        <row><entry>update_exists</entry>        <entry>error</entry>             <entry>apply_remote, error, keep_local</entry></row>
+        <row><entry>update_missing</entry>       <entry>skip</entry>              <entry>apply_or_error, apply_or_skip, error, skip</entry></row>
+        <row><entry>update_origin_differs</entry><entry>apply_remote</entry>      <entry>apply_remote, error, keep_local</entry></row>
+        <row><entry>delete_missing</entry>       <entry>skip</entry>              <entry>error, skip</entry></row>
+        <row><entry>delete_origin_differs</entry><entry>apply_remote</entry>      <entry>apply_remote, error, keep_local</entry></row>
+       </tbody>
+      </tgroup>
+     </table>
+
+     <para>
+      The scenarios that trigger conflicts, along with the behavior for each
+      <replaceable class="parameter">conflict_type</replaceable>, are listed below.
+      <variablelist>
+       <varlistentry id="sql-createsubscription-params-with-conflict_type-insert-exists" xreflabel="insert_exists">
+        <term><literal>insert_exists</literal> (<type>enum</type>)</term>
+        <listitem>
+         <para>
+          This conflict occurs when inserting a row that violates a
+          <literal>NOT DEFERRABLE</literal> unique constraint.
+          The default behavior is an error will be raised until the conflict
+          is resolved manually or the resolver is configured to a non-default
+          value that can automatically resolve the conflict.
+         </para>
+         <para>
+          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.
+         </para>
+        </listitem>
+       </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_type-update-exists" xreflabel="update_exists">
+       <term><literal>update_exists</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This conflict occurs when the updated value of a row violates
+         a <literal>NOT DEFERRABLE</literal>
+         unique constraint.
+         The default behavior is an error will be raised until the conflict
+         is resolved manually or the resolver is configured to a non-default
+         value that can automatically resolve the conflict.
+        </para>
+        <para>
+         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>
+        <para>
+         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.
+        </para>
+        <warning>
+         <para>
+          If the local table contains multiple <literal>NOT DEFERRABLE</literal> unique constraint
+          columns and the conflict resolution strategy for an INSERT or UPDATE favors applying the changes,
+          then a remote INSERT or UPDATE could trigger multiple <literal>update_exists</literal> conflicts.
+          Each conflict will be detected and resolved in sequence, potentially leading to the deletion of
+          multiple local rows.
+         </para>
+        </warning>
+       </listitem>
+      </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_type-update-missing" xreflabel="update_missing">
+       <term><literal>update_missing</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This conflict occurs when the tuple to be updated was not found.
+         The default behavior is to simply skip the update.
+        </para>
+       </listitem>
+      </varlistentry>
+
+       <varlistentry id="sql-createsubscription-params-with-conflict_type-update-origin-differs" xreflabel="update_origin_differs">
+        <term><literal>update_origin_differs</literal> (<type>enum</type>)</term>
+        <listitem>
+         <para>
+          This conflict occurs when updating a row that was previously
+          modified by another origin.
+          The default behavoir is that the update will be applied regardless
+          of the origin of the local row.
+         </para>
+         <para>
+          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.
+         </para>
+        </listitem>
+      </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_type-delete-missing" xreflabel="delete_missing">
+       <term><literal>delete_missing</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This conflict occurs when the tuple to be deleted was not found.
+         The default behavior is to simply skip the delete.
+        </para>
+       </listitem>
+      </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_type-delete-origin-differs" xreflabel="delete_origin_differs">
+       <term><literal>delete_origin_differs</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This conflict occurs when deleting a row that was previously modified
+         by another origin.
+         The default behavior is that the delete will be applied regardless
+         of the origin of the local row.
+        </para>
+        <para>
+         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.
+        </para>
+       </listitem>
+      </varlistentry>
+
+     </variablelist>
+    </para>
+
+    <para>
+     The behavior of each <replaceable class="parameter">conflict_resolver</replaceable>
+     is described below. Users can choose from the following resolvers for automatic conflict
+     resolution.
+     <variablelist>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_resolver-error">
+       <term><literal>error</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This resolver throws an error and stops replication.
+        </para>
+       </listitem>
+      </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_resolver-skip">
+       <term><literal>skip</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This resolver skips processing the remote change and continues replication
+         with the next change.
+        </para>
+       </listitem>
+      </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_resolver-apply-remote">
+       <term><literal>apply_remote</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This resolver applies the remote change.
+        </para>
+       </listitem>
+      </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_resolver-keep-local">
+       <term><literal>keep_local</literal> (<type>enum</type>)</term>
+        <listitem>
+         <para>
+          This resolver maintains the local tuple; the remote change is not applied.
+         </para>
+        </listitem>
+       </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_resolver-apply-or-error">
+       <term><literal>apply_or_error</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This resolver is only used for <literal>update_missing</literal>.
+         An attempt is made to convert the update into an insert; if this cannot
+         be done due to missing information, then an error is thrown, and replication
+         is stopped.
+        </para>
+       </listitem>
+      </varlistentry>
+
+      <varlistentry id="sql-createsubscription-params-with-conflict_resolver-apply-or-skip">
+       <term><literal>apply_or_skip</literal> (<type>enum</type>)</term>
+       <listitem>
+        <para>
+         This resolver is only used for <literal>update_missing</literal>.
+         An attempt is made to convert the update into an insert; if this
+         cannot be done due to missing information, then the change is skipped.
+        </para>
+       </listitem>
+      </varlistentry>
+
+      </variablelist>
+     </para>
+
+    </listitem>
+   </varlistentry>
+
    <varlistentry id="sql-createsubscription-params-with">
     <term><literal>WITH ( <replaceable class="parameter">subscription_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] )</literal></term>
     <listitem>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 02ccc63..54d8ad9 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -36,6 +36,7 @@
 #include "executor/executor.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
+#include "parser/scansup.h"
 #include "pgstat.h"
 #include "replication/logicallauncher.h"
 #include "replication/logicalworker.h"
@@ -583,6 +584,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	bits32		supported_opts;
 	SubOpts		opts = {0};
 	AclResult	aclresult;
+	List	   *conflict_resolvers = NIL;
 
 	/*
 	 * Parse and check options.
@@ -597,6 +599,10 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
+	/* Parse and get conflict resolvers list. */
+	conflict_resolvers =
+		ParseAndGetSubConflictResolvers(pstate, stmt->resolvers, true);
+
 	/*
 	 * Since creating a replication slot is not transactional, rolling back
 	 * the transaction leaves the created replication slot.  So we cannot run
@@ -723,6 +729,9 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 
 	recordDependencyOnOwner(SubscriptionRelationId, subid, owner);
 
+	/* Update the Conflict Resolvers in pg_subscription_conflict */
+	SetSubConflictResolvers(subid, conflict_resolvers);
+
 	ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
 	replorigin_create(originname);
 
@@ -1581,6 +1590,47 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				update_tuple = true;
 				break;
 			}
+		case ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS:
+			{
+				List	   *conflict_resolvers = NIL;
+
+				/*
+				 * Get the list of conflict types and resolvers and validate
+				 * them.
+				 */
+				conflict_resolvers = ParseAndGetSubConflictResolvers(
+																	 pstate,
+																	 stmt->resolvers,
+																	 false);
+
+				/*
+				 * Update the conflict resolvers for the corresponding
+				 * conflict types in the pg_subscription_conflict catalog.
+				 */
+				UpdateSubConflictResolvers(conflict_resolvers, subid);
+				break;
+			}
+		case ALTER_SUBSCRIPTION_CONFLICT_RESOLVER_RESET_ALL:
+			{
+				List	   *conflict_resolvers = NIL;
+
+				/*
+				 * Create list of conflict resolvers and set them in the
+				 * catalog.
+				 */
+				conflict_resolvers = GetDefaultConflictResolvers();
+				UpdateSubConflictResolvers(conflict_resolvers, subid);
+				break;
+			}
+		case ALTER_SUBSCRIPTION_CONFLICT_RESOLVER_RESET:
+			{
+				/*
+				 * Reset the conflict resolver for this conflict type to its
+				 * default.
+				 */
+				ResetSubConflictResolver(subid, stmt->conflict_type_name);
+				break;
+			}
 
 		default:
 			elog(ERROR, "unrecognized ALTER SUBSCRIPTION kind %d",
@@ -1832,6 +1882,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	/* Remove any associated relation synchronization states. */
 	RemoveSubscriptionRel(subid, InvalidOid);
 
+	/* Remove any associated conflict resolvers */
+	RemoveSubConflictResolvers(subid);
+
 	/* Remove the origin tracking if exists. */
 	ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
 	replorigin_drop_by_name(originname, true, false);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 4aa8646..28591c9 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -425,7 +425,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				OptTableElementList TableElementList OptInherit definition
 				OptTypedTableElementList TypedTableElementList
 				reloptions opt_reloptions
-				OptWith opt_definition func_args func_args_list
+				OptWith opt_definition opt_resolver_definition func_args func_args_list
 				func_args_with_defaults func_args_with_defaults_list
 				aggr_args aggr_args_list
 				func_as createfunc_opt_list opt_createfunc_opt_list alterfunc_opt_list
@@ -613,6 +613,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type <ival>	opt_check_option
 
 %type <str>		opt_provider security_label
+%type <str>		conflict_type
 
 %type <target>	xml_attribute_el
 %type <list>	xml_attribute_list xml_attributes
@@ -770,7 +771,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 
 	RANGE READ REAL REASSIGN RECURSIVE REF_P REFERENCES REFERENCING
 	REFRESH REINDEX RELATIVE_P RELEASE RENAME REPEATABLE REPLACE REPLICA
-	RESET RESTART RESTRICT RETURN RETURNING RETURNS REVOKE RIGHT ROLE ROLLBACK ROLLUP
+	RESET RESOLVER RESTART RESTRICT RETURN RETURNING RETURNS REVOKE RIGHT ROLE ROLLBACK ROLLUP
 	ROUTINE ROUTINES ROW ROWS RULE
 
 	SAVEPOINT SCALAR SCHEMA SCHEMAS SCROLL SEARCH SECOND_P SECURITY SELECT
@@ -8813,6 +8814,11 @@ opt_definition:
 			| /*EMPTY*/								{ $$ = NIL; }
 		;
 
+opt_resolver_definition:
+			CONFLICT RESOLVER definition			{ $$ = $3; }
+			| /*EMPTY*/								{ $$ = NIL; }
+		;
+
 table_func_column:	param_name func_type
 				{
 					FunctionParameter *n = makeNode(FunctionParameter);
@@ -10747,14 +10753,15 @@ AlterPublicationStmt:
  *****************************************************************************/
 
 CreateSubscriptionStmt:
-			CREATE SUBSCRIPTION name CONNECTION Sconst PUBLICATION name_list opt_definition
+	CREATE SUBSCRIPTION name CONNECTION Sconst PUBLICATION name_list opt_resolver_definition opt_definition
 				{
 					CreateSubscriptionStmt *n =
 						makeNode(CreateSubscriptionStmt);
 					n->subname = $3;
 					n->conninfo = $5;
 					n->publication = $7;
-					n->options = $8;
+					n->resolvers = $8;
+					n->options = $9;
 					$$ = (Node *) n;
 				}
 		;
@@ -10861,6 +10868,38 @@ AlterSubscriptionStmt:
 					n->options = $5;
 					$$ = (Node *) n;
 				}
+			| ALTER SUBSCRIPTION name opt_resolver_definition
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+
+					n->kind = ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS;
+					n->subname = $3;
+					n->resolvers = $4;
+					$$ = (Node *) n;
+				}
+			| ALTER SUBSCRIPTION name RESET CONFLICT RESOLVER ALL
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+
+					n->kind = ALTER_SUBSCRIPTION_CONFLICT_RESOLVER_RESET_ALL;
+					n->subname = $3;
+					$$ = (Node *) n;
+				}
+			| ALTER SUBSCRIPTION name RESET CONFLICT RESOLVER FOR conflict_type
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+
+					n->kind = ALTER_SUBSCRIPTION_CONFLICT_RESOLVER_RESET;
+					n->subname = $3;
+					n->conflict_type_name = $8;
+					$$ = (Node *) n;
+				}
+		;
+conflict_type:
+			Sconst								{ $$ = $1; }
 		;
 
 /*****************************************************************************
@@ -17785,6 +17824,7 @@ unreserved_keyword:
 			| REPLACE
 			| REPLICA
 			| RESET
+			| RESOLVER
 			| RESTART
 			| RESTRICT
 			| RETURN
@@ -18414,6 +18454,7 @@ bare_label_keyword:
 			| REPLACE
 			| REPLICA
 			| RESET
+			| RESOLVER
 			| RESTART
 			| RESTRICT
 			| RETURN
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 5d9ff62..cb6d50c 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -8,30 +8,111 @@
  *	  src/backend/replication/logical/conflict.c
  *
  * This file contains the code for logging conflicts on the subscriber during
- * logical replication.
+ * logical replication and setting up conflict resolvers for a subscription.
  *-------------------------------------------------------------------------
  */
 
 #include "postgres.h"
 
 #include "access/commit_ts.h"
+#include "access/heapam.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_subscription_conflict_d.h"
+#include "commands/defrem.h"
 #include "access/tableam.h"
 #include "executor/executor.h"
+#include "parser/scansup.h"
 #include "pgstat.h"
 #include "replication/conflict.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
 #include "replication/worker_internal.h"
 #include "storage/lmgr.h"
 #include "utils/lsyscache.h"
+#include "utils/syscache.h"
+
+static const char *const ConflictResolverNames[] = {
+	[CR_APPLY_REMOTE] = "apply_remote",
+	[CR_KEEP_LOCAL] = "keep_local",
+	[CR_APPLY_OR_SKIP] = "apply_or_skip",
+	[CR_APPLY_OR_ERROR] = "apply_or_error",
+	[CR_SKIP] = "skip",
+	[CR_ERROR] = "error"
+};
+
+StaticAssertDecl(lengthof(ConflictResolverNames) == CONFLICT_NUM_RESOLVERS,
+				 "array length mismatch");
 
-static const char *const ConflictTypeNames[] = {
-	[CT_INSERT_EXISTS] = "insert_exists",
-	[CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
-	[CT_UPDATE_EXISTS] = "update_exists",
-	[CT_UPDATE_MISSING] = "update_missing",
-	[CT_DELETE_ORIGIN_DIFFERS] = "delete_origin_differs",
-	[CT_DELETE_MISSING] = "delete_missing"
+/*
+ * Valid conflict resolvers for each conflict type.
+ *
+ * First member represents default resolver for each conflict_type.
+ * The same defaults are used in pg_dump.c. If any default is changed here,
+ * ensure the corresponding value is updated in pg_dump's is_default_resolver
+ * function.
+ *
+ * XXX: If we do not want to maintain different resolvers such as
+ * apply_or_skip and apply_or_error for update_missing conflict,
+ * then we can retain apply_remote and keep_local only. Then these
+ * resolvers in context of update_missing will mean:
+ *
+ * keep_local: do not apply the update as INSERT.
+ * apply_remote: apply the update as INSERT. If we could not apply,
+ * then log and skip.
+ *
+ * Similarly SKIP can be replaced with KEEP_LOCAL for both update_missing
+ * and delete_missing conflicts. For missing rows, 'SKIP' sounds more user
+ * friendly name for a resolver and thus has been added here.
+ */
+typedef struct ConflictInfo {
+		ConflictType conflict_type;
+		const char *conflict_name;
+		ConflictResolver default_resolver;
+		bool allowed_resolvers[CONFLICT_NUM_RESOLVERS];
+} ConflictInfo;
+
+static ConflictInfo ConflictInfoMap[] = {
+	[CT_INSERT_EXISTS] = {
+		.conflict_type = CT_INSERT_EXISTS,
+		.conflict_name = "insert_exists",
+		.default_resolver = CR_ERROR,
+		.allowed_resolvers = {[CR_ERROR]=true, [CR_APPLY_REMOTE]=true, [CR_KEEP_LOCAL]=true},
+	},
+	[CT_UPDATE_EXISTS] = {
+		.conflict_type = CT_UPDATE_EXISTS,
+		.conflict_name = "update_exists",
+		.default_resolver = CR_ERROR,
+		.allowed_resolvers = {[CR_ERROR]=true, [CR_APPLY_REMOTE]=true, [CR_KEEP_LOCAL]=true},
+	},
+	[CT_UPDATE_ORIGIN_DIFFERS] = {
+		.conflict_type = CT_UPDATE_ORIGIN_DIFFERS,
+		.conflict_name = "update_origin_differs",
+		.default_resolver = CR_APPLY_REMOTE,
+		.allowed_resolvers = {[CR_APPLY_REMOTE]=true, [CR_KEEP_LOCAL]=true, [CR_ERROR]=true},
+	},
+	[CT_UPDATE_MISSING] = {
+		.conflict_type = CT_UPDATE_MISSING,
+		.conflict_name = "update_missing",
+		.default_resolver = CR_SKIP,
+		.allowed_resolvers = {[CR_SKIP]=true, [CR_APPLY_OR_SKIP]=true, [CR_APPLY_OR_ERROR]=true, [CR_ERROR]=true},
+	},
+	[CT_DELETE_MISSING] = {
+		.conflict_type = CT_DELETE_MISSING,
+		.conflict_name = "delete_missing",
+		.default_resolver = CR_SKIP,
+		.allowed_resolvers = {[CR_SKIP]=true, [CR_ERROR]=true},
+	},
+	[CT_DELETE_ORIGIN_DIFFERS] = {
+		.conflict_type = CT_DELETE_ORIGIN_DIFFERS,
+		.conflict_name = "delete_origin_differs",
+		.default_resolver = CR_APPLY_REMOTE,
+		.allowed_resolvers = {[CR_APPLY_REMOTE]=true, [CR_KEEP_LOCAL]=true, [CR_ERROR]=true},
+	},
 };
 
+StaticAssertDecl(lengthof(ConflictInfoMap) == CONFLICT_NUM_TYPES,
+				 "array length mismatch");
+
 static int	errcode_apply_conflict(ConflictType type);
 static int	errdetail_apply_conflict(EState *estate,
 									 ResultRelInfo *relinfo,
@@ -122,7 +203,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
 			errmsg("conflict detected on relation \"%s.%s\": conflict=%s",
 				   get_namespace_name(RelationGetNamespace(localrel)),
 				   RelationGetRelationName(localrel),
-				   ConflictTypeNames[type]),
+				   ConflictInfoMap[type].conflict_name),
 			errdetail_apply_conflict(estate, relinfo, type, searchslot,
 									 localslot, remoteslot, indexoid,
 									 localxmin, localorigin, localts));
@@ -489,3 +570,355 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot,
 
 	return index_value;
 }
+
+/*
+ * Report a warning about incomplete conflict detection and resolution if
+ * track_commit_timestamp is disabled.
+ */
+static void
+conf_detection_check_prerequisites(void)
+{
+	if (!track_commit_timestamp)
+		ereport(WARNING,
+				errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("conflict detection and resolution could be incomplete due to disabled track_commit_timestamp"),
+				errdetail("Conflict types 'update_origin_differs' and 'delete_origin_differs'"
+						  "cannot be detected unless 'track_commit_timestamp' is enabled"));
+}
+
+/*
+ * Validate the conflict type and return the corresponding ConflictType enum.
+ */
+ConflictType
+ValidateConflictType(const char *conflict_type)
+{
+	ConflictType type;
+	bool		valid = false;
+
+	/* Check conflict type validity */
+	for (type = 0; type < CONFLICT_NUM_TYPES; type++)
+	{
+		if (pg_strcasecmp(ConflictInfoMap[type].conflict_name, conflict_type) == 0)
+		{
+			valid = true;
+			break;
+		}
+	}
+
+	if (!valid)
+		ereport(ERROR,
+				errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("%s is not a valid conflict type", conflict_type));
+
+	return type;
+}
+
+/*
+ * Validate the conflict type and resolver. It returns an enum ConflictType
+ * corresponding to the conflict type string passed by the caller.
+ */
+ConflictType
+ValidateConflictTypeAndResolver(const char *conflict_type,
+								const char *conflict_resolver)
+{
+	ConflictType type;
+	ConflictResolver resolver;
+	bool		valid = false;
+
+	/* Validate conflict type */
+	type = ValidateConflictType(conflict_type);
+
+	/* Validate the conflict resolver name */
+	for (resolver = 0; resolver < CONFLICT_NUM_RESOLVERS; resolver++)
+	{
+		if (pg_strcasecmp(ConflictResolverNames[resolver], conflict_resolver) == 0)
+		{
+			valid = true;
+			break;
+		}
+	}
+
+	if (!valid)
+	{
+		ereport(ERROR,
+				errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("%s is not a valid conflict resolver", conflict_resolver));
+	}
+
+	valid = ConflictInfoMap[type].allowed_resolvers[resolver];
+
+	if (!valid)
+		ereport(ERROR,
+				errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("%s is not a valid conflict resolver for conflict type %s",
+					   conflict_resolver,
+					   conflict_type));
+
+	return type;
+}
+
+/*
+ * Common 'CONFLICT RESOLVER' parsing function for CREATE and ALTER
+ * SUBSCRIPTION commands.
+ *
+ * It reports an error if duplicate options are specified.
+ *
+ * Returns a list of conflict types along with their corresponding conflict
+ * resolvers. If 'add_defaults' is true, it appends default resolvers for any
+ * conflict types that have not been explicitly defined by the user.
+ */
+List *
+ParseAndGetSubConflictResolvers(ParseState *pstate,
+								List *stmtresolvers, bool add_defaults)
+{
+	List		   *res = NIL;
+	bool			already_seen[CONFLICT_NUM_TYPES] = {0};
+	ConflictType 	type;
+
+	/* Loop through the user provided resolvers */
+	foreach_ptr(DefElem, defel, stmtresolvers)
+	{
+		char	   *resolver;
+		ConflictTypeResolver *conftyperesolver = NULL;
+
+		/* Validate the conflict type and resolver */
+		resolver = defGetString(defel);
+		resolver = downcase_truncate_identifier(
+												resolver, strlen(resolver), false);
+		type = ValidateConflictTypeAndResolver(defel->defname,
+											   resolver);
+
+		/* Check if the conflict type has already been seen */
+		if (already_seen[type])
+			errorConflictingDefElem(defel, pstate);
+
+		already_seen[type] = true;
+
+		conftyperesolver = palloc(sizeof(ConflictTypeResolver));
+		conftyperesolver->conflict_type_name = defel->defname;
+		conftyperesolver->conflict_resolver_name = resolver;
+		res = lappend(res, conftyperesolver);
+	}
+
+	/* Once validation is complete, warn users if prerequisites are not met */
+	if (stmtresolvers)
+		conf_detection_check_prerequisites();
+
+	/*
+	 * If add_defaults is true, fill remaining conflict types with default
+	 * resolvers.
+	 */
+	if (add_defaults)
+	{
+		for (int i = 0; i < CONFLICT_NUM_TYPES; i++)
+		{
+			ConflictTypeResolver *conftyperesolver = NULL;
+
+			if (!already_seen[i])
+			{
+				ConflictResolver def_resolver = ConflictInfoMap[i].default_resolver;
+
+				conftyperesolver = palloc(sizeof(ConflictTypeResolver));
+				conftyperesolver->conflict_type_name = ConflictInfoMap[i].conflict_name;
+				conftyperesolver->conflict_resolver_name = ConflictResolverNames[def_resolver];;
+				res = lappend(res, conftyperesolver);
+			}
+		}
+	}
+
+	return res;
+}
+
+/*
+ * Get the list of conflict types and their corresponding default resolvers.
+ */
+List *
+GetDefaultConflictResolvers()
+{
+	List	   *res = NIL;
+
+	for (ConflictType type = 0; type < CONFLICT_NUM_TYPES; type++)
+	{
+		ConflictTypeResolver *resolver = NULL;
+		ConflictResolver def_resolver = ConflictInfoMap[type].default_resolver;
+
+		/* Allocate memory for each ConflictTypeResolver */
+		resolver = palloc(sizeof(ConflictTypeResolver));
+
+		resolver->conflict_type_name = ConflictInfoMap[type].conflict_name;
+		resolver->conflict_resolver_name = ConflictResolverNames[def_resolver];
+
+		/* Append to the response list */
+		res = lappend(res, resolver);
+	}
+
+	return res;
+}
+
+/*
+ * Update the Subscription's conflict resolvers in pg_subscription_conflict
+ * system catalog for the given conflict types.
+ */
+void
+UpdateSubConflictResolvers(List *conflict_resolvers, Oid subid)
+{
+	Datum		values[Natts_pg_subscription_conflict];
+	bool		nulls[Natts_pg_subscription_conflict];
+	bool		replaces[Natts_pg_subscription_conflict];
+	HeapTuple	oldtup;
+	HeapTuple	newtup = NULL;
+	Relation	pg_subscription_conflict;
+	char	   *cur_conflict_res;
+	Datum		datum;
+
+	/* Prepare to update a tuple */
+	memset(nulls, false, sizeof(nulls));
+	memset(replaces, false, sizeof(replaces));
+	memset(values, 0, sizeof(values));
+
+	pg_subscription_conflict = table_open(SubscriptionConflictId, RowExclusiveLock);
+
+	foreach_ptr(ConflictTypeResolver, conftyperesolver, conflict_resolvers)
+	{
+		/* Set up subid and conflict_type to search in cache */
+		values[Anum_pg_subscription_conflict_confsubid - 1] = ObjectIdGetDatum(subid);
+		values[Anum_pg_subscription_conflict_conftype - 1] =
+			CStringGetTextDatum(conftyperesolver->conflict_type_name);
+
+		oldtup = SearchSysCache2(SUBSCRIPTIONCONFLMAP,
+								 values[Anum_pg_subscription_conflict_confsubid - 1],
+								 values[Anum_pg_subscription_conflict_conftype - 1]);
+
+		if (!HeapTupleIsValid(oldtup))
+			elog(ERROR, "cache lookup failed for table conflict %s for subid %u",
+				 conftyperesolver->conflict_type_name, subid);
+
+		datum = SysCacheGetAttrNotNull(SUBSCRIPTIONCONFLMAP,
+									   oldtup, Anum_pg_subscription_conflict_confres);
+		cur_conflict_res = TextDatumGetCString(datum);
+
+		/*
+		 * Update system catalog only if the new resolver is not same as the
+		 * existing one.
+		 */
+		if (pg_strcasecmp(cur_conflict_res,
+						  conftyperesolver->conflict_resolver_name) != 0)
+		{
+			/* Update the new resolver */
+			values[Anum_pg_subscription_conflict_confres - 1] =
+				CStringGetTextDatum(conftyperesolver->conflict_resolver_name);
+			replaces[Anum_pg_subscription_conflict_confres - 1] = true;
+
+			newtup = heap_modify_tuple(oldtup,
+									   RelationGetDescr(pg_subscription_conflict),
+									   values, nulls, replaces);
+			CatalogTupleUpdate(pg_subscription_conflict,
+							   &oldtup->t_self, newtup);
+			heap_freetuple(newtup);
+		}
+
+		ReleaseSysCache(oldtup);
+	}
+
+	table_close(pg_subscription_conflict, RowExclusiveLock);
+}
+
+/*
+ * Reset the conflict resolver for this conflict type to its default setting.
+ */
+void
+ResetSubConflictResolver(Oid subid, char *conflict_type)
+{
+	ConflictType idx;
+	ConflictTypeResolver *conflictResolver = NULL;
+	List	   *conflictresolver_list = NIL;
+
+	/* Validate the conflict type and get the index */
+	idx = ValidateConflictType(conflict_type);
+	conflictResolver = palloc(sizeof(ConflictTypeResolver));
+	conflictResolver->conflict_type_name = conflict_type;
+
+	/* Get the default resolver for this conflict_type */
+	conflictResolver->conflict_resolver_name =
+		ConflictResolverNames[ConflictInfoMap[idx].default_resolver];
+
+	/* Create a list of conflict resolvers and update in catalog */
+	conflictresolver_list = lappend(conflictresolver_list, conflictResolver);
+	UpdateSubConflictResolvers(conflictresolver_list, subid);
+
+}
+
+/*
+ * Set Conflict Resolvers on the subscription
+ */
+void
+SetSubConflictResolvers(Oid subId, List *conflict_resolvers)
+{
+	Relation	pg_subscription_conflict;
+	Datum		values[Natts_pg_subscription_conflict];
+	bool		nulls[Natts_pg_subscription_conflict];
+	HeapTuple	newtup = NULL;
+	Oid			conflict_oid;
+
+	pg_subscription_conflict = table_open(SubscriptionConflictId, RowExclusiveLock);
+
+	/* Prepare to update a tuple */
+	memset(nulls, false, sizeof(nulls));
+
+	/* Iterate over the list of resolvers */
+	foreach_ptr(ConflictTypeResolver, conftyperesolver, conflict_resolvers)
+	{
+		values[Anum_pg_subscription_conflict_confsubid - 1] =
+			ObjectIdGetDatum(subId);
+		values[Anum_pg_subscription_conflict_conftype - 1] =
+			CStringGetTextDatum(conftyperesolver->conflict_type_name);
+		values[Anum_pg_subscription_conflict_confres - 1] =
+			CStringGetTextDatum(conftyperesolver->conflict_resolver_name);
+
+		/* Get a new oid and update the tuple into catalog */
+		conflict_oid = GetNewOidWithIndex(pg_subscription_conflict,
+										  SubscriptionConflictOidIndexId,
+										  Anum_pg_subscription_conflict_oid);
+		values[Anum_pg_subscription_conflict_oid - 1] =
+			ObjectIdGetDatum(conflict_oid);
+		newtup = heap_form_tuple(RelationGetDescr(pg_subscription_conflict),
+								 values, nulls);
+		CatalogTupleInsert(pg_subscription_conflict, newtup);
+		heap_freetuple(newtup);
+	}
+
+	table_close(pg_subscription_conflict, RowExclusiveLock);
+}
+
+/*
+ * Remove the subscription conflict resolvers for the subscription id
+ */
+void
+RemoveSubConflictResolvers(Oid subid)
+{
+	Relation	rel;
+	HeapTuple	tup;
+	TableScanDesc scan;
+	ScanKeyData skey[1];
+
+	rel = table_open(SubscriptionConflictId, RowExclusiveLock);
+
+	/*
+	 * Search using the subid to return all conflict resolvers for this
+	 * subscription.
+	 */
+	ScanKeyInit(&skey[0],
+				Anum_pg_subscription_conflict_confsubid,
+				BTEqualStrategyNumber,
+				F_OIDEQ,
+				ObjectIdGetDatum(subid));
+
+	scan = table_beginscan_catalog(rel, 1, skey);
+
+	/* Iterate through the tuples and delete them */
+	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
+		CatalogTupleDelete(rel, &tup->t_self);
+
+	table_endscan(scan);
+	table_close(rel, RowExclusiveLock);
+}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 1b47c38..921b1e5 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -395,6 +395,8 @@ static void setupDumpWorker(Archive *AH);
 static TableInfo *getRootTableInfo(const TableInfo *tbinfo);
 static bool forcePartitionRootLoad(const TableInfo *tbinfo);
 static void read_dump_filters(const char *filename, DumpOptions *dopt);
+static bool is_default_resolver(const char *confType, const char *confRes);
+static void destroyConcflictResolverList(SimplePtrList *list);
 
 
 int
@@ -4830,7 +4832,9 @@ getSubscriptions(Archive *fout)
 {
 	DumpOptions *dopt = fout->dopt;
 	PQExpBuffer query;
+	PQExpBuffer confQuery;
 	PGresult   *res;
+	PGresult   *confRes;
 	SubscriptionInfo *subinfo;
 	int			i_tableoid;
 	int			i_oid;
@@ -4851,7 +4855,9 @@ getSubscriptions(Archive *fout)
 	int			i_subenabled;
 	int			i_subfailover;
 	int			i,
-				ntups;
+				j,
+				ntups,
+				conf_ntuples;
 
 	if (dopt->no_subscriptions || fout->remoteVersion < 100000)
 		return;
@@ -5012,6 +5018,38 @@ getSubscriptions(Archive *fout)
 		subinfo[i].subfailover =
 			pg_strdup(PQgetvalue(res, i, i_subfailover));
 
+		/* Populate conflict type fields using the new query */
+		confQuery = createPQExpBuffer();
+		appendPQExpBuffer(confQuery,
+						  "SELECT conftype, confres FROM pg_catalog.pg_subscription_conflict "
+						  "WHERE confsubid = %u order by conftype;", subinfo[i].dobj.catId.oid);
+		confRes = ExecuteSqlQuery(fout, confQuery->data, PGRES_TUPLES_OK);
+
+		conf_ntuples = PQntuples(confRes);
+
+		/* Initialize pointers in the list to NULL */
+		subinfo[i].conflict_resolvers = (SimplePtrList)
+		{
+			.head = NULL, .tail = NULL 
+		
+		};
+
+		/* Store conflict types and resolvers from the query result in subinfo */
+		for (j = 0; j < conf_ntuples; j++)
+		{
+			/* Create the ConflictTypeResolver node */
+			ConflictTypeResolver *conftyperesolver = palloc(sizeof(ConflictTypeResolver));
+
+			conftyperesolver->conflict_type = pg_strdup(PQgetvalue(confRes, j, 0));
+			conftyperesolver->resolver = pg_strdup(PQgetvalue(confRes, j, 1));
+
+			/* Append the node to subinfo's list */
+			simple_ptr_list_append(&subinfo[i].conflict_resolvers, conftyperesolver);
+		}
+
+		PQclear(confRes);
+		destroyPQExpBuffer(confQuery);
+
 		/* Decide whether we want to dump it */
 		selectDumpableObject(&(subinfo[i].dobj), fout);
 	}
@@ -5222,7 +5260,43 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 		appendPQExpBufferStr(publications, fmtId(pubnames[i]));
 	}
 
-	appendPQExpBuffer(query, " PUBLICATION %s WITH (connect = false, slot_name = ", publications->data);
+	appendPQExpBuffer(query, " PUBLICATION %s ", publications->data);
+
+	/* Add conflict resolvers, if any */
+	if (fout->remoteVersion >= 180000)
+	{
+		bool		first_resolver = true;
+		SimplePtrListCell *cell = NULL;
+		ConflictTypeResolver *conftyperesolver = NULL;
+
+		for (cell = subinfo->conflict_resolvers.head; cell; cell = cell->next)
+		{
+			conftyperesolver = (ConflictTypeResolver *) cell->ptr;
+
+			if (!is_default_resolver(conftyperesolver->conflict_type,
+									 conftyperesolver->resolver))
+			{
+				if (first_resolver)
+				{
+					appendPQExpBuffer(query, "CONFLICT RESOLVER (%s = '%s'",
+									  conftyperesolver->conflict_type,
+									  conftyperesolver->resolver);
+					first_resolver = false;
+				}
+				else
+					appendPQExpBuffer(query, ", %s = '%s'",
+									  conftyperesolver->conflict_type,
+									  conftyperesolver->resolver);
+			}
+		}
+
+		/* If there was at least one resolver, close the braces */
+		if (!first_resolver)
+			appendPQExpBufferStr(query, ") ");
+	}
+
+	appendPQExpBuffer(query, "WITH (connect = false, slot_name = ");
+
 	if (subinfo->subslotname)
 		appendStringLiteralAH(query, subinfo->subslotname, fout);
 	else
@@ -5315,6 +5389,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 					 NULL, subinfo->rolname,
 					 subinfo->dobj.catId, 0, subinfo->dobj.dumpId);
 
+	/* Clean-up the conflict_resolver list */
+	destroyConcflictResolverList((SimplePtrList *) &subinfo->conflict_resolvers);
+
 	destroyPQExpBuffer(publications);
 	free(pubnames);
 
@@ -19051,3 +19128,48 @@ read_dump_filters(const char *filename, DumpOptions *dopt)
 
 	filter_free(&fstate);
 }
+
+/*
+ * is_default_resolver - checks if the given resolver is the default for the
+ * specified conflict type.
+ */
+static bool
+is_default_resolver(const char *confType, const char *confRes)
+{
+	/*
+	 * The default resolvers for each conflict type are taken from the
+	 * predefined mapping ConflictTypeDefaultResolvers[] in conflict.c.
+	 *
+	 * Only modify these defaults if the corresponding values in conflict.c
+	 * are changed.
+	 */
+
+	if (strcmp(confType, "insert_exists") == 0 ||
+		strcmp(confType, "update_exists") == 0)
+		return strcmp(confRes, "error") == 0;
+	else if (strcmp(confType, "update_missing") == 0 ||
+			 strcmp(confType, "delete_missing") == 0)
+		return strcmp(confRes, "skip") == 0;
+	else if (strcmp(confType, "update_origin_differs") == 0 ||
+			 strcmp(confType, "delete_origin_differs") == 0)
+		return strcmp(confRes, "apply_remote") == 0;
+
+	return false;
+}
+
+/*
+ * destroyConflictResolverList - frees up the SimplePtrList containing
+ * cells pointing to struct ConflictTypeResolver nodes.
+ */
+static void
+destroyConcflictResolverList(SimplePtrList *conflictlist)
+{
+	SimplePtrListCell *cell = NULL;
+
+	/* Free the list items */
+	for (cell = conflictlist->head; cell; cell = cell->next)
+		pfree((ConflictTypeResolver *) cell->ptr);
+
+	/* Destroy the pointer list */
+	simple_ptr_list_destroy(conflictlist);
+}
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 9f907ed..35a23eb 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -655,6 +655,13 @@ typedef struct _PublicationSchemaInfo
 /*
  * The SubscriptionInfo struct is used to represent subscription.
  */
+
+typedef struct _ConflictTypeResolver
+{
+	const char *conflict_type;
+	const char *resolver;
+} ConflictTypeResolver;
+
 typedef struct _SubscriptionInfo
 {
 	DumpableObject dobj;
@@ -673,6 +680,7 @@ typedef struct _SubscriptionInfo
 	char	   *suborigin;
 	char	   *suboriginremotelsn;
 	char	   *subfailover;
+	SimplePtrList conflict_resolvers;
 } SubscriptionInfo;
 
 /*
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index ab6c830..ec8b446 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -3012,9 +3012,10 @@ my %tests = (
 		create_order => 50,
 		create_sql => 'CREATE SUBSCRIPTION sub3
 						 CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1
+						 CONFLICT RESOLVER (insert_exists = "apply_remote", update_missing = "error")
 						 WITH (connect = false, origin = any);',
 		regexp => qr/^
-			\QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3');\E
+			\QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 CONFLICT RESOLVER (insert_exists = 'apply_remote', update_missing = 'error') WITH (connect = false, slot_name = 'sub3');\E
 			/xm,
 		like => { %full_runs, section_post_data => 1, },
 	},
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 6a36c91..636c507 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6545,9 +6545,10 @@ describeSubscriptions(const char *pattern, bool verbose)
 	PQExpBufferData buf;
 	PGresult   *res;
 	printQueryOpt myopt = pset.popt;
+	char	  **footers = NULL;
 	static const bool translate_columns[] = {false, false, false, false,
 		false, false, false, false, false, false, false, false, false, false,
-	false};
+	false, false, false, false, false, false, false};
 
 	if (pset.sversion < 100000)
 	{
@@ -6627,12 +6628,13 @@ describeSubscriptions(const char *pattern, bool verbose)
 			appendPQExpBuffer(&buf,
 							  ", subskiplsn AS \"%s\"\n",
 							  gettext_noop("Skip LSN"));
+
 	}
 
 	/* Only display subscriptions in current database. */
 	appendPQExpBufferStr(&buf,
-						 "FROM pg_catalog.pg_subscription\n"
-						 "WHERE subdbid = (SELECT oid\n"
+						 "FROM pg_catalog.pg_subscription s\n"
+						 "WHERE s.subdbid = (SELECT oid\n"
 						 "                 FROM pg_catalog.pg_database\n"
 						 "                 WHERE datname = pg_catalog.current_database())");
 
@@ -6648,7 +6650,6 @@ describeSubscriptions(const char *pattern, bool verbose)
 	appendPQExpBufferStr(&buf, "ORDER BY 1;");
 
 	res = PSQLexec(buf.data);
-	termPQExpBuffer(&buf);
 	if (!res)
 		return false;
 
@@ -6657,9 +6658,118 @@ describeSubscriptions(const char *pattern, bool verbose)
 	myopt.translate_columns = translate_columns;
 	myopt.n_translate_columns = lengthof(translate_columns);
 
+	if (verbose && pset.sversion >= 180000)
+	{
+		PGresult   *result;
+		int			i,
+					tuples,
+					max_len = 0;
+		char	   *prev_subname = NULL;
+
+		printfPQExpBuffer(&buf,
+						  "SELECT s.subname, c.conftype, c.confres \n"
+						  " FROM pg_catalog.pg_subscription_conflict c \n"
+						  " JOIN pg_catalog.pg_subscription s \n"
+						  " ON c.confsubid = s.oid \n"
+						  " WHERE s.subdbid = (SELECT oid\n"
+						  " FROM pg_catalog.pg_database\n"
+						  " WHERE datname = pg_catalog.current_database())");
+
+		if (!validateSQLNamePattern(&buf, pattern, true, false,
+									NULL, "s.subname", NULL,
+									NULL,
+									NULL, 1))
+		{
+			termPQExpBuffer(&buf);
+			return false;
+		}
+
+		appendPQExpBufferStr(&buf, "ORDER BY s.subname, c.conftype");
+
+		result = PSQLexec(buf.data);
+		if (!result)
+		{
+			termPQExpBuffer(&buf);
+			return false;
+		}
+		else
+			tuples = PQntuples(result);
+
+		/* Calculate the maximum length of the conflict type (conftype) */
+		for (i = 0; i < tuples; i++)
+		{
+			int len = strlen(PQgetvalue(result, i, 1));
+			if (len > max_len)
+				max_len = len;
+		}
+
+		if (tuples > 0)
+		{
+			/*
+			 * Allocate memory for footers. Size of footers will be
+			 * 1 (for storing "Conflict Resolvers" string) + conflict
+			 * resolver count + subscription count + 1 (for storing NULL).
+			 * Subscription count is determined by dividing the number of
+			 * conflict resolvers by 6. This needs to be changed when
+			 * CONFLICT_NUM_RESOLVERS in conflict.h changes
+			 */
+			int subscription_count = tuples / 6;
+			int footer_idx = 1;
+
+			footers = (char **) pg_malloc((1 + tuples + subscription_count + 1)
+										   * sizeof(char *));
+			footers[0] = pg_strdup(_("\nConflict Resolvers:"));
+
+			for (i = 0; i < tuples; i++)
+			{
+				const char *subname = PQgetvalue(result, i, 0);
+				const char *conftype = PQgetvalue(result, i, 1);
+				const char *confres = PQgetvalue(result, i, 2);
+
+				/*
+				 * Print the subscription name only if it changes
+				 * (group by subname).
+				 */
+				if (!prev_subname || strcmp(prev_subname, subname) != 0)
+				{
+					printfPQExpBuffer(&buf, "\n(%s)", subname);
+					footers[footer_idx++] = pg_strdup(buf.data);
+					prev_subname = pg_strdup(subname);
+				}
+
+				/* Adjust the padding to align the "=" */
+				printfPQExpBuffer(&buf, "%s%*s = \"%s\"",
+								  conftype,
+								  (int)(max_len - strlen(conftype) + 2),
+								   "",
+								   confres);
+
+				footers[footer_idx++] = pg_strdup(buf.data);
+			}
+
+			footers[footer_idx] = NULL;
+			myopt.footers = footers;
+		}
+
+		PQclear(result);
+	}
+
 	printQuery(res, &myopt, pset.queryFout, false, pset.logfile);
+	termPQExpBuffer(&buf);
 
 	PQclear(res);
+
+	/* Free the memory allocated for the footer */
+	if (footers)
+	{
+		char	**footer = NULL;
+
+		for (footer = footers; *footer; footer++)
+			pg_free(*footer);
+
+		pg_free(footers);
+	}
+
 	return true;
 }
 
diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c
index 1be0056..a083e1e 100644
--- a/src/bin/psql/tab-complete.in.c
+++ b/src/bin/psql/tab-complete.in.c
@@ -2266,7 +2266,7 @@ match_previous_words(int pattern_id,
 	else if (Matches("ALTER", "SUBSCRIPTION", MatchAny))
 		COMPLETE_WITH("CONNECTION", "ENABLE", "DISABLE", "OWNER TO",
 					  "RENAME TO", "REFRESH PUBLICATION", "SET", "SKIP (",
-					  "ADD PUBLICATION", "DROP PUBLICATION");
+					  "ADD PUBLICATION", "DROP PUBLICATION", "CONFLICT RESOLVER (");
 	/* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION */
 	else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION"))
 		COMPLETE_WITH("WITH (");
@@ -2297,6 +2297,19 @@ match_previous_words(int pattern_id,
 	else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN,
 					 "ADD|DROP|SET", "PUBLICATION", MatchAny, "WITH", "("))
 		COMPLETE_WITH("copy_data", "refresh");
+	/* ALTER SUBSCRIPTION <name> CONFLICT RESOLVER (<conflict_type> = <conflict_resolver> [, ..]) */
+	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
+			TailMatches("RESOLVER", "("))
+		COMPLETE_WITH("insert_exists", "update_origin_differs", "update_exists",
+					  "update_missing", "delete_origin_differs", "delete_missing");
+	/* ALTER SUBSCRIPTION <name> RESET CONFLICT RESOLVER ALL|FOR <conflict_type> */
+	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
+			 TailMatches("RESET", "CONFLICT", "RESOLVER"))
+		COMPLETE_WITH("ALL", "FOR (");
+	else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) &&
+			 TailMatches("RESOLVER", "FOR", "("))
+		COMPLETE_WITH("insert_exists", "update_origin_differs", "update_exists",
+					  "update_missing", "delete_origin_differs", "delete_missing");
 
 	/* ALTER SCHEMA <name> */
 	else if (Matches("ALTER", "SCHEMA", MatchAny))
@@ -3668,8 +3681,13 @@ match_previous_words(int pattern_id,
 	{
 		/* complete with nothing here as this refers to remote publications */
 	}
+	/* Complete "CREATE SUBSCRIPTION <name> ... CONFLICT RESOLVER ( <confoptions> ) */
 	else if (Matches("CREATE", "SUBSCRIPTION", MatchAnyN, "PUBLICATION", MatchAny))
-		COMPLETE_WITH("WITH (");
+		COMPLETE_WITH("CONFLICT RESOLVER", "WITH (");
+	else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("CONFLICT", "RESOLVER", "("))
+		COMPLETE_WITH("insert_exists", "update_origin_differs", "update_exists",
+					  "update_missing", "delete_origin_differs", "delete_missing");
+
 	/* Complete "CREATE SUBSCRIPTION <name> ...  WITH ( <opt>" */
 	else if (Matches("CREATE", "SUBSCRIPTION", MatchAnyN, "WITH", "("))
 		COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
diff --git a/src/include/catalog/Makefile b/src/include/catalog/Makefile
index 167f91a..f2611c1 100644
--- a/src/include/catalog/Makefile
+++ b/src/include/catalog/Makefile
@@ -81,7 +81,8 @@ CATALOG_HEADERS := \
 	pg_publication_namespace.h \
 	pg_publication_rel.h \
 	pg_subscription.h \
-	pg_subscription_rel.h
+	pg_subscription_rel.h \
+	pg_subscription_conflict.h
 
 GENERATED_HEADERS := $(CATALOG_HEADERS:%.h=%_d.h)
 
diff --git a/src/include/catalog/meson.build b/src/include/catalog/meson.build
index f70d1da..959e1d9 100644
--- a/src/include/catalog/meson.build
+++ b/src/include/catalog/meson.build
@@ -69,6 +69,7 @@ catalog_headers = [
   'pg_publication_rel.h',
   'pg_subscription.h',
   'pg_subscription_rel.h',
+  'pg_subscription_conflict.h',
 ]
 
 # The .dat files we need can just be listed alphabetically.
diff --git a/src/include/catalog/pg_subscription_conflict.h b/src/include/catalog/pg_subscription_conflict.h
new file mode 100644
index 0000000..15f02a1
--- /dev/null
+++ b/src/include/catalog/pg_subscription_conflict.h
@@ -0,0 +1,55 @@
+/* -------------------------------------------------------------------------
+ *
+ * pg_subscription_conflict.h
+ *    definition of the "subscription conflict resolver" system
+ *    catalog (pg_subscription_conflict)
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/catalog/pg_subscription_conflict.h
+ *
+ * NOTES
+ *	  The Catalog.pm module reads this file and derives schema
+ *	  information.
+ *
+ * -------------------------------------------------------------------------
+ */
+#ifndef PG_SUBSCRIPTION_CONFLICT_H
+#define PG_SUBSCRIPTION_CONFLICT_H
+
+#include "catalog/genbki.h"
+#include "catalog/pg_subscription_conflict_d.h"
+
+/* ----------------
+ *		pg_subscription_conflict definition.  cpp turns this into
+ *		typedef struct FormData_pg_subscription_conflict
+ * ----------------
+ */
+CATALOG(pg_subscription_conflict,8881,SubscriptionConflictId)
+{
+	Oid			oid;			/* OID of the object itself */
+	Oid			confsubid BKI_LOOKUP(pg_subscription);	/* OID of subscription */
+
+#ifdef CATALOG_VARLEN			/* variable-length fields start here */
+	text		conftype BKI_FORCE_NOT_NULL;	/* conflict type */
+	text		confres BKI_FORCE_NOT_NULL; /* conflict resolver */
+#endif
+} FormData_pg_subscription_conflict;
+
+/* ----------------
+ *		Form_pg_subscription_conflict corresponds to a pointer to a row with
+ *		the format of pg_subscription_conflict relation.
+ * ----------------
+ */
+typedef FormData_pg_subscription_conflict *Form_pg_subscription_conflict;
+
+DECLARE_TOAST(pg_subscription_conflict, 8882, 8883);
+
+DECLARE_UNIQUE_INDEX_PKEY(pg_subscription_conflict_oid_index, 8884, SubscriptionConflictOidIndexId, pg_subscription_conflict, btree(oid oid_ops));
+
+DECLARE_UNIQUE_INDEX(pg_subscription_conflict_confsubid_conftype_index, 8885, SubscriptionConflictSubIdTypeIndexId, pg_subscription_conflict, btree(confsubid oid_ops, conftype text_ops));
+
+MAKE_SYSCACHE(SUBSCRIPTIONCONFLMAP, pg_subscription_conflict_confsubid_conftype_index, 256);
+
+#endif							/* PG_SUBSCRIPTION_CONFLICT_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 5b62df3..162b205 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -4214,6 +4214,7 @@ typedef struct CreateSubscriptionStmt
 	char	   *conninfo;		/* Connection string to publisher */
 	List	   *publication;	/* One or more publication to subscribe to */
 	List	   *options;		/* List of DefElem nodes */
+	List	   *resolvers;		/* List of conflict resolvers */
 } CreateSubscriptionStmt;
 
 typedef enum AlterSubscriptionType
@@ -4226,6 +4227,9 @@ typedef enum AlterSubscriptionType
 	ALTER_SUBSCRIPTION_REFRESH,
 	ALTER_SUBSCRIPTION_ENABLED,
 	ALTER_SUBSCRIPTION_SKIP,
+	ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS,
+	ALTER_SUBSCRIPTION_CONFLICT_RESOLVER_RESET,
+	ALTER_SUBSCRIPTION_CONFLICT_RESOLVER_RESET_ALL,
 } AlterSubscriptionType;
 
 typedef struct AlterSubscriptionStmt
@@ -4236,6 +4240,9 @@ typedef struct AlterSubscriptionStmt
 	char	   *conninfo;		/* Connection string to publisher */
 	List	   *publication;	/* One or more publication to subscribe to */
 	List	   *options;		/* List of DefElem nodes */
+	List	   *resolvers;		/* List of conflict type names and resolver
+								 * names */
+	char	   *conflict_type_name; /* The conflict type name to be reset */
 } AlterSubscriptionStmt;
 
 typedef struct DropSubscriptionStmt
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index 899d64a..e62b0c6 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -376,6 +376,7 @@ PG_KEYWORD("repeatable", REPEATABLE, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("replace", REPLACE, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("replica", REPLICA, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("reset", RESET, UNRESERVED_KEYWORD, BARE_LABEL)
+PG_KEYWORD("resolver", RESOLVER, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("restart", RESTART, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("restrict", RESTRICT, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("return", RETURN, UNRESERVED_KEYWORD, BARE_LABEL)
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index c759677..c5865a1 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -1,6 +1,6 @@
 /*-------------------------------------------------------------------------
  * conflict.h
- *	   Exports for conflicts logging.
+ *	   Exports for conflicts logging and resolvers configuration.
  *
  * Copyright (c) 2024, PostgreSQL Global Development Group
  *
@@ -10,6 +10,7 @@
 #define CONFLICT_H
 
 #include "nodes/execnodes.h"
+#include "parser/parse_node.h"
 #include "utils/timestamp.h"
 
 /*
@@ -50,6 +51,41 @@ typedef enum
 
 #define CONFLICT_NUM_TYPES (CT_DELETE_MISSING + 1)
 
+/*
+ * Conflict resolvers that can be used to resolve various conflicts.
+ *
+ * See ConflictTypeResolverMap in conflict.c to find out which all
+ * resolvers are supported for each conflict type.
+ */
+typedef enum ConflictResolver
+{
+	/* Apply the remote change */
+	CR_APPLY_REMOTE,
+
+	/* Keep the local change */
+	CR_KEEP_LOCAL,
+
+	/* Apply the remote change; skip if it cannot be applied */
+	CR_APPLY_OR_SKIP,
+
+	/* Apply the remote change; emit error if it cannot be applied */
+	CR_APPLY_OR_ERROR,
+
+	/* Skip applying the change */
+	CR_SKIP,
+
+	/* Error out */
+	CR_ERROR,
+} ConflictResolver;
+
+#define CONFLICT_NUM_RESOLVERS (CR_ERROR + 1)
+
+typedef struct ConflictTypeResolver
+{
+	const char *conflict_type_name;
+	const char *conflict_resolver_name;
+} ConflictTypeResolver;
+
 extern bool GetTupleTransactionInfo(TupleTableSlot *localslot,
 									TransactionId *xmin,
 									RepOriginId *localorigin,
@@ -62,5 +98,16 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
 								Oid indexoid, TransactionId localxmin,
 								RepOriginId localorigin, TimestampTz localts);
 extern void InitConflictIndexes(ResultRelInfo *relInfo);
+extern void SetSubConflictResolvers(Oid subId, List *resolvers);
+extern void RemoveSubConflictResolvers(Oid confid);
+extern List *ParseAndGetSubConflictResolvers(ParseState *pstate,
+											 List *stmtresolvers,
+											 bool add_defaults);
+extern void UpdateSubConflictResolvers(List *conflict_resolvers, Oid subid);
+extern ConflictType ValidateConflictType(const char *conflict_type);
+extern ConflictType ValidateConflictTypeAndResolver(const char *conflict_type,
+													const char *conflict_resolver);
+extern List *GetDefaultConflictResolvers(void);
+extern void ResetSubConflictResolver(Oid subid, char *conflict_type);
 
 #endif
diff --git a/src/test/regress/expected/oidjoins.out b/src/test/regress/expected/oidjoins.out
index 215eb89..42bf2cc 100644
--- a/src/test/regress/expected/oidjoins.out
+++ b/src/test/regress/expected/oidjoins.out
@@ -266,3 +266,4 @@ NOTICE:  checking pg_subscription {subdbid} => pg_database {oid}
 NOTICE:  checking pg_subscription {subowner} => pg_authid {oid}
 NOTICE:  checking pg_subscription_rel {srsubid} => pg_subscription {oid}
 NOTICE:  checking pg_subscription_rel {srrelid} => pg_class {oid}
+NOTICE:  checking pg_subscription_conflict {confsubid} => pg_subscription {oid}
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 17d48b1..bcf4bcd 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -120,7 +120,16 @@ HINT:  To initiate replication, you must manually create the replication slot, e
        Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | none   | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub4)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
 \dRs+ regress_testsub4
@@ -128,7 +137,16 @@ ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
        Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub4)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 DROP SUBSCRIPTION regress_testsub3;
 DROP SUBSCRIPTION regress_testsub4;
@@ -149,7 +167,16 @@ ERROR:  invalid connection string syntax: missing "=" after "foobar" in connecti
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
 ALTER SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist2';
@@ -161,7 +188,16 @@ ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
       Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | f                 | t             | f        | off                | dbname=regress_doesnotexist2 | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
 ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = false);
@@ -180,7 +216,16 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
       Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/12345
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 -- ok - with lsn = NONE
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
@@ -192,7 +237,16 @@ ERROR:  invalid WAL location (LSN): 0/0
       Name       |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 BEGIN;
 ALTER SUBSCRIPTION regress_testsub ENABLE;
@@ -227,7 +281,16 @@ HINT:  Available values: local, remote_write, remote_apply, on, off.
         Name         |           Owner           | Enabled |     Publication     | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |           Conninfo           | Skip LSN 
 ---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
  regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | local              | dbname=regress_doesnotexist2 | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub_foo)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 -- rename back to keep the rest simple
 ALTER SUBSCRIPTION regress_testsub_foo RENAME TO regress_testsub;
@@ -259,7 +322,16 @@ HINT:  To initiate replication, you must manually create the replication slot, e
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | t      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (binary = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -268,7 +340,16 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 DROP SUBSCRIPTION regress_testsub;
 -- fail - streaming must be boolean or 'parallel'
@@ -283,7 +364,16 @@ HINT:  To initiate replication, you must manually create the replication slot, e
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
 \dRs+
@@ -291,7 +381,16 @@ ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -300,7 +399,16 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 -- fail - publication already exists
 ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub WITH (refresh = false);
@@ -318,7 +426,16 @@ ERROR:  publication "testpub1" is already in subscription "regress_testsub"
       Name       |           Owner           | Enabled |         Publication         | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 -- fail - publication used more than once
 ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub1 WITH (refresh = false);
@@ -336,7 +453,16 @@ ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (ref
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 DROP SUBSCRIPTION regress_testsub;
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION mypub
@@ -375,7 +501,16 @@ HINT:  To initiate replication, you must manually create the replication slot, e
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 -- we can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
@@ -384,7 +519,16 @@ ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
@@ -397,7 +541,169 @@ HINT:  To initiate replication, you must manually create the replication slot, e
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- fail - invalid conflict type
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (foo = 'keep_local') WITH (connect = false);
+ERROR:  foo is not a valid conflict type
+-- fail - invalid conflict resolver
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = foo) WITH (connect = false);
+ERROR:  foo is not a valid conflict resolver
+-- fail - invalid resolver for that conflict type
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub
+ CONFLICT RESOLVER (update_missing = 'apply_remote') WITH (connect = false);
+ERROR:  apply_remote is not a valid conflict resolver for conflict type update_missing
+-- fail - duplicate conflict type
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = 'keep_local', insert_exists = 'keep_local');
+ERROR:  conflicting or redundant options
+LINE 1: ... CONFLICT RESOLVER (insert_exists = 'keep_local', insert_exi...
+                                                             ^
+-- creating subscription with no explicit conflict resolvers should
+-- configure default conflict resolvers
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+WARNING:  subscription was created, but is not connected
+HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+\dRs+
+                                                                                                                List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- ok - valid conflict types and resolvers
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = 'keep_local', update_missing = 'skip', delete_origin_differs = 'keep_local' ) WITH (connect = false);
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflict types 'update_origin_differs' and 'delete_origin_differs'cannot be detected unless 'track_commit_timestamp' is enabled
+WARNING:  subscription was created, but is not connected
+HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+--check if above are configured; for non specified conflict types, default resolvers should be seen
+\dRs+
+                                                                                                                List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "keep_local"
+insert_exists           = "keep_local"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
+
+-- fail - altering with invalid conflict type
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (foo = 'keep_local');
+ERROR:  foo is not a valid conflict type
+-- fail - altering with invalid conflict resolver
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'foo');
+ERROR:  foo is not a valid conflict resolver
+-- fail - altering with duplicate conflict type
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'apply_remote', insert_exists = 'apply_remote');
+ERROR:  conflicting or redundant options
+LINE 1: ...ONFLICT RESOLVER (insert_exists = 'apply_remote', insert_exi...
+                                                             ^
+-- ok - valid conflict types and resolvers
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'apply_remote', update_missing = 'skip', delete_origin_differs = 'keep_local' );
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflict types 'update_origin_differs' and 'delete_origin_differs'cannot be detected unless 'track_commit_timestamp' is enabled
+\dRs+
+                                                                                                                List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "keep_local"
+insert_exists           = "apply_remote"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
+
+-- ok - valid conflict types and resolvers
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (update_exists = 'keep_local', delete_missing = 'error', update_origin_differs = 'error');
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflict types 'update_origin_differs' and 'delete_origin_differs'cannot be detected unless 'track_commit_timestamp' is enabled
+\dRs+
+                                                                                                                List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "error"
+delete_origin_differs   = "keep_local"
+insert_exists           = "apply_remote"
+update_exists           = "keep_local"
+update_missing          = "skip"
+update_origin_differs   = "error"
+
+-- fail - reset with an invalid conflit type
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER for 'foo';
+ERROR:  foo is not a valid conflict type
+-- ok - valid conflict type
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER for 'insert_exists';
+\dRs+
+                                                                                                                List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "error"
+delete_origin_differs   = "keep_local"
+insert_exists           = "error"
+update_exists           = "keep_local"
+update_missing          = "skip"
+update_origin_differs   = "error"
+
+-- ok - reset ALL
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER ALL;
+\dRs+
+                                                                                                                List of subscriptions
+      Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
@@ -413,7 +719,16 @@ HINT:  To initiate replication, you must manually create the replication slot, e
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
 \dRs+
@@ -421,7 +736,16 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
       Name       |           Owner           | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit |          Conninfo           | Skip LSN 
 -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | t                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
-(1 row)
+
+Conflict Resolvers:
+
+(regress_testsub)
+delete_missing          = "skip"
+delete_origin_differs   = "apply_remote"
+insert_exists           = "error"
+update_exists           = "error"
+update_missing          = "skip"
+update_origin_differs   = "apply_remote"
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 007c9e7..3f08869 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -272,6 +272,69 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 
+-- fail - invalid conflict type
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (foo = 'keep_local') WITH (connect = false);
+
+-- fail - invalid conflict resolver
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = foo) WITH (connect = false);
+
+-- fail - invalid resolver for that conflict type
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub
+ CONFLICT RESOLVER (update_missing = 'apply_remote') WITH (connect = false);
+
+-- fail - duplicate conflict type
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = 'keep_local', insert_exists = 'keep_local');
+
+-- creating subscription with no explicit conflict resolvers should
+-- configure default conflict resolvers
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
+-- ok - valid conflict types and resolvers
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = 'keep_local', update_missing = 'skip', delete_origin_differs = 'keep_local' ) WITH (connect = false);
+
+--check if above are configured; for non specified conflict types, default resolvers should be seen
+\dRs+
+
+-- fail - altering with invalid conflict type
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (foo = 'keep_local');
+
+-- fail - altering with invalid conflict resolver
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'foo');
+
+-- fail - altering with duplicate conflict type
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'apply_remote', insert_exists = 'apply_remote');
+
+-- ok - valid conflict types and resolvers
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'apply_remote', update_missing = 'skip', delete_origin_differs = 'keep_local' );
+
+\dRs+
+
+-- ok - valid conflict types and resolvers
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (update_exists = 'keep_local', delete_missing = 'error', update_origin_differs = 'error');
+
+\dRs+
+
+-- fail - reset with an invalid conflit type
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER for 'foo';
+
+-- ok - valid conflict type
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER for 'insert_exists';
+
+\dRs+
+
+-- ok - reset ALL
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER ALL;
+
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
 -- fail - disable_on_error must be boolean
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = foo);
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index a65e1c0..bf1ea95 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -468,7 +468,9 @@ ConditionVariableMinimallyPadded
 ConditionalStack
 ConfigData
 ConfigVariable
+ConflictResolver
 ConflictType
+ConflictTypeResolver
 ConnCacheEntry
 ConnCacheKey
 ConnParams
@@ -863,6 +865,7 @@ FormData_pg_statistic
 FormData_pg_statistic_ext
 FormData_pg_statistic_ext_data
 FormData_pg_subscription
+FormData_pg_subscription_conflict
 FormData_pg_subscription_rel
 FormData_pg_tablespace
 FormData_pg_transform
-- 
1.8.3.1

