From 76d44c31bf8ba88ca4f9bbc3aa9c457b99f7310b Mon Sep 17 00:00:00 2001
From: Ajin Cherian <itsajin@gmail.com>
Date: Thu, 12 Sep 2024 02:29:24 -0400
Subject: [PATCH v13 1/5] 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 reseting 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          |  88 +-----
 doc/src/sgml/monitoring.sgml                   |  13 +-
 doc/src/sgml/ref/alter_subscription.sgml       |  12 +
 doc/src/sgml/ref/create_subscription.sgml      | 178 +++++++++++
 src/backend/commands/subscriptioncmds.c        |  98 ++++++
 src/backend/parser/gram.y                      |  49 ++-
 src/backend/replication/logical/conflict.c     | 422 +++++++++++++++++++++++++
 src/bin/pg_dump/pg_dump.c                      |  96 +++++-
 src/bin/pg_dump/pg_dump.h                      | 118 ++++++-
 src/bin/psql/describe.c                        |  14 +-
 src/bin/psql/tab-complete.c                    |   6 +-
 src/include/catalog/Makefile                   |   3 +-
 src/include/catalog/meson.build                |   1 +
 src/include/catalog/pg_subscription_conflict.h |  55 ++++
 src/include/nodes/parsenodes.h                 |   6 +
 src/include/parser/kwlist.h                    |   1 +
 src/include/replication/conflict.h             |  51 +++
 src/test/regress/expected/oidjoins.out         |   1 +
 src/test/regress/expected/subscription.out     | 284 ++++++++++++-----
 src/test/regress/sql/subscription.sql          |  57 ++++
 src/tools/pgindent/typedefs.list               |   2 +
 21 files changed, 1369 insertions(+), 186 deletions(-)
 mode change 100644 => 100755 doc/src/sgml/ref/alter_subscription.sgml
 mode change 100644 => 100755 doc/src/sgml/ref/create_subscription.sgml
 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 df62eb4..f82adfc 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,83 +1582,15 @@ 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>
+   Additional logging is triggered in various conflict scenarios, each identified as a
+   conflict type, and the conflict statistics are collected (displayed in the
+   <link linkend="monitoring-pg-stat-subscription-stats"><structname>pg_stat_subscription_stats</structname></link> view).
+   Users have the option to configure a <literal>conflict_resolver</literal> for each
+   <literal>conflict_type</literal> when creating a subscription.
+   For more information on the supported <literal>conflict_types</literal> detected and
+   <literal>conflict_resolvers</literal>, refer to section
+   <link linkend="sql-createsubscription-params-with-conflict-resolver"><literal>CONFLICT RESOLVERS</literal></link>.
+
     Note that there are other conflict scenarios, such as exclusion constraint
     violations. Currently, we do not provide additional details for them in the
     log.
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 933de6f..6ead5c3 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
old mode 100644
new mode 100755
index fdc648d..e7b39f2
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -32,6 +32,7 @@ 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>] [, ...] )
 </synopsis>
  </refsynopsisdiv>
 
@@ -345,6 +346,17 @@ 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 either the default conflict resolvers or those set by <xref linkend="sql-createsubscription"/>.
+      Refer to section <link linkend="sql-createsubscription-params-with-conflict-resolver"><literal>CONFLICT RESOLVERS</literal></link>
+      for the details on supported <literal>conflict_types</literal> and <literal>conflict_resolvers</literal>.
+     </para>
+    </listitem>
+   </varlistentry>
   </variablelist>
 
   <para>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
old mode 100644
new mode 100755
index 740b7d94..e04a269
--- 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>
@@ -432,6 +433,183 @@ 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 options for conflict resolvers for different conflict_types.
+     </para>
+
+     <para>
+      The <replaceable class="parameter">conflict_type</replaceable> and their default behaviour are listed below.
+      <variablelist>
+       <varlistentry id="sql-createsubscription-params-with-conflict_type-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.
+           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 or the resolver is configured to a
+           non-default value that can automatically resolve the conflict.
+          </para>
+         </listitem>
+       </varlistentry>
+
+       <varlistentry id="sql-createsubscription-params-with-conflict_type-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.
+           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="sql-createsubscription-params-with-conflict_type-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. 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 or the resolver is configured to a
+           non-default value that can automatically resolve the conflict.
+           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="sql-createsubscription-params-with-conflict_type-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 update will simply be skipped in this scenario.
+         </para>
+        </listitem>
+       </varlistentry>
+
+       <varlistentry id="sql-createsubscription-params-with-conflict_type-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. 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="sql-createsubscription-params-with-conflict_type-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 delete will simply be skipped in this scenario.
+          </para>
+         </listitem>
+       </varlistentry>
+
+      </variablelist></para>
+
+     <para>
+      The <replaceable class="parameter">conflict_resolver</replaceable> and their behaviour
+      are listed below. Users can use any of 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. It can be used for
+           any conflict type.
+           It is the default resolver for <literal>insert_exists</literal> and
+           <literal>update_exists</literal>.
+          </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 continue replication
+           with the next change.
+           It can be used for <literal>update_missing</literal> and
+           <literal>delete_missing</literal> and is the default resolver for these types.
+          </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. It can be used for
+           <literal>insert_exists</literal>, <literal>update_exists</literal>,
+           <literal>update_origin_differs</literal> and <literal>delete_origin_differs</literal>.
+
+           It is the default resolver for <literal>insert_exists</literal> and
+           <literal>update_exists</literal>.
+          </para>
+         </listitem>
+       </varlistentry>
+
+       <varlistentry id="sql-createsubscription-params-with-conflict_resolver-keep-local">
+        <term><literal>keep_local</literal> (<type>enum</type>)</term>
+         <listitem>
+          <para>
+           With this resolver, the remote change is not applied and the local tuple is maintained.
+           It can be used for <literal>insert_exists</literal>, <literal>update_exists</literal>,
+           <literal>update_origin_differs</literal> and <literal>delete_origin_differs</literal>.
+          </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>
+
+       <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>
+       </variablelist></para>
+
+     </listitem>
+    </varlistentry>
   </variablelist>
 
   <para>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 02ccc63..131e4f5 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -14,6 +14,7 @@
 
 #include "postgres.h"
 
+#include "access/commit_ts.h"
 #include "access/htup_details.h"
 #include "access/table.h"
 #include "access/twophase.h"
@@ -28,6 +29,7 @@
 #include "catalog/pg_database_d.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_subscription_rel.h"
+#include "catalog/pg_subscription_conflict.h"
 #include "catalog/pg_type.h"
 #include "commands/dbcommands.h"
 #include "commands/defrem.h"
@@ -36,7 +38,9 @@
 #include "executor/executor.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
+#include "parser/scansup.h"
 #include "pgstat.h"
+#include "replication/conflict.h"
 #include "replication/logicallauncher.h"
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
@@ -440,6 +444,50 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 }
 
 /*
+ * Parsing function for conflict resolvers in CREATE SUBSCRIPTION command.
+ * This function will report an error if mutually exclusive or duplicate
+ * options are specified.
+ */
+static void
+parse_subscription_conflict_resolvers(List *stmtresolvers,
+									  ConflictTypeResolver *resolvers)
+{
+	ListCell   *lc;
+	List	   *SeenTypes = NIL;
+
+	/* Warn users if prerequisites are not met. */
+	if (stmtresolvers)
+		conf_detection_check_prerequisites();
+
+	/* First initialize the resolvers with default values. */
+	SetDefaultResolvers(resolvers);
+
+	foreach(lc, stmtresolvers)
+	{
+		DefElem    *defel = (DefElem *) lfirst(lc);
+		ConflictType type;
+		char	   *resolver;
+
+		/* Check if the conflict type already exists in the list */
+		if (list_member(SeenTypes, makeString(defel->defname)))
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("duplicate conflict type \"%s\" found", defel->defname)));
+
+		/* Validate the conflict type and resolver. */
+		resolver = defGetString(defel);
+		type = validate_conflict_type_and_resolver(defel->defname,
+												   defGetString(defel));
+
+		/* Add the conflict type to the list of seen types */
+		SeenTypes = lappend(SeenTypes, makeString((char *)resolvers[type].conflict_type));
+
+		/* Update the corresponding resolver for the given conflict type. */
+		resolvers[type].resolver = downcase_truncate_identifier(resolver, strlen(resolver), false);
+	}
+}
+
+/*
  * Add publication names from the list to a string.
  */
 static void
@@ -583,6 +631,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	bits32		supported_opts;
 	SubOpts		opts = {0};
 	AclResult	aclresult;
+	ConflictTypeResolver conflictResolvers[CT_MAX + 1];
 
 	/*
 	 * Parse and check options.
@@ -597,6 +646,9 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
+	/* Parse and check conflict resolvers. */
+	parse_subscription_conflict_resolvers(stmt->resolvers, conflictResolvers);
+
 	/*
 	 * Since creating a replication slot is not transactional, rolling back
 	 * the transaction leaves the created replication slot.  So we cannot run
@@ -723,6 +775,9 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 
 	recordDependencyOnOwner(SubscriptionRelationId, subid, owner);
 
+	/* Update the Conflict Resolvers in pg_subscription_conflict */
+	SetSubConflictResolver(subid, conflictResolvers, CT_MAX + 1);
+
 	ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
 	replorigin_create(originname);
 
@@ -1581,6 +1636,46 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				update_tuple = true;
 				break;
 			}
+		case ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS:
+			{
+				List	   *conflict_resolvers = NIL;
+
+				conf_detection_check_prerequisites();
+
+				/* Get the list of conflict types and resolvers and validate them. */
+				conflict_resolvers = GetAndValidateSubsConflictResolverList(stmt->resolvers);
+
+				/*
+				 * Update the conflict resolvers for the corresponding
+				 * conflict types in the pg_subscription_conflict catalog.
+				 */
+				UpdateSubConflictResolvers(conflict_resolvers, subid);
+				break;
+			}
+		case ALTER_SUBSCRIPTION_RESET_ALL_CONFLICT_RESOLVERS:
+			{
+				ConflictTypeResolver conflictResolvers[CT_MAX + 1];
+
+				/* Remove the existing conflict resolvers. */
+				RemoveSubscriptionConflictResolvers(subid);
+
+				/*
+				 * Create list of conflict resolvers and set them in the
+				 * catalog.
+				 */
+				SetDefaultResolvers(conflictResolvers);
+				SetSubConflictResolver(subid, conflictResolvers, CT_MAX + 1);
+				break;
+			}
+		case ALTER_SUBSCRIPTION_RESET_CONFLICT_RESOLVER:
+			{
+				/*
+				 * Reset the conflict resolver for this conflict type to its
+				 * default.
+				 */
+				ResetConflictResolver(subid, stmt->conflict_type);
+				break;
+			}
 
 		default:
 			elog(ERROR, "unrecognized ALTER SUBSCRIPTION kind %d",
@@ -1832,6 +1927,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	/* Remove any associated relation synchronization states. */
 	RemoveSubscriptionRel(subid, InvalidOid);
 
+	/* Remove any associated conflict resolvers */
+	RemoveSubscriptionConflictResolvers(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 84cef57..3c090b0 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
@@ -612,6 +612,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
@@ -769,7 +770,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
@@ -8754,6 +8755,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);
@@ -10688,14 +10694,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;
 				}
 		;
@@ -10802,6 +10809,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_RESET_ALL_CONFLICT_RESOLVERS;
+					n->subname = $3;
+					$$ = (Node *) n;
+				}
+			| ALTER SUBSCRIPTION name RESET CONFLICT RESOLVER FOR conflict_type
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+
+					n->kind = ALTER_SUBSCRIPTION_RESET_CONFLICT_RESOLVER;
+					n->subname = $3;
+					n->conflict_type = $8;
+					$$ = (Node *) n;
+				}
+		;
+conflict_type:
+			Sconst								{ $$ = $1; }
 		;
 
 /*****************************************************************************
@@ -17725,6 +17764,7 @@ unreserved_keyword:
 			| REPLACE
 			| REPLICA
 			| RESET
+			| RESOLVER
 			| RESTART
 			| RESTRICT
 			| RETURN
@@ -18353,6 +18393,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..c54f6c6 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -15,13 +15,28 @@
 #include "postgres.h"
 
 #include "access/commit_ts.h"
+#include "access/heaptoast.h"
+#include "access/heapam.h"
+#include "access/table.h"
+#include "access/tableam.h"
+#include "catalog/dependency.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_subscription.h"
+#include "catalog/pg_subscription_conflict.h"
+#include "catalog/pg_subscription_conflict_d.h"
+#include "catalog/pg_inherits.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 ConflictTypeNames[] = {
 	[CT_INSERT_EXISTS] = "insert_exists",
@@ -32,6 +47,55 @@ static const char *const ConflictTypeNames[] = {
 	[CT_DELETE_MISSING] = "delete_missing"
 };
 
+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"
+};
+
+#define CONFLICT_TYPE_MAX_RESOLVERS 4
+
+/*
+ * Valid conflict resolvers for each conflict type.
+ *
+ * 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.
+ */
+static const int ConflictTypeResolverMap[][CONFLICT_TYPE_MAX_RESOLVERS] = {
+	[CT_INSERT_EXISTS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR},
+	[CT_UPDATE_EXISTS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR},
+	[CT_UPDATE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR},
+	[CT_UPDATE_MISSING] = {CR_APPLY_OR_SKIP, CR_APPLY_OR_ERROR, CR_SKIP, CR_ERROR},
+	[CT_DELETE_MISSING] = {CR_SKIP, CR_ERROR},
+	[CT_DELETE_ORIGIN_DIFFERS] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR}
+};
+
+/*
+ * Default conflict resolver for each conflict type.
+ */
+static const int ConflictTypeDefaultResolvers[] = {
+	[CT_INSERT_EXISTS] = CR_ERROR,
+	[CT_UPDATE_EXISTS] = CR_ERROR,
+	[CT_UPDATE_ORIGIN_DIFFERS] = CR_APPLY_REMOTE,
+	[CT_UPDATE_MISSING] = CR_SKIP,
+	[CT_DELETE_MISSING] = CR_SKIP,
+	[CT_DELETE_ORIGIN_DIFFERS] = CR_APPLY_REMOTE
+
+};
+
 static int	errcode_apply_conflict(ConflictType type);
 static int	errdetail_apply_conflict(EState *estate,
 									 ResultRelInfo *relinfo,
@@ -489,3 +553,361 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot,
 
 	return index_value;
 }
+
+
+/*
+ * Set default values for CONFLICT RESOLVERS for each conflict type
+ */
+void
+SetDefaultResolvers(ConflictTypeResolver * conflictResolvers)
+{
+	ConflictType type;
+
+	for (type = CT_MIN; type <= CT_MAX; type++)
+	{
+		conflictResolvers[type].conflict_type = ConflictTypeNames[type];
+		conflictResolvers[type].resolver =
+			ConflictResolverNames[ConflictTypeDefaultResolvers[type]];
+	}
+}
+
+/*
+ * Validate the conflict type and resolver. It returns an enum ConflictType
+ * corresponding to the conflict type string passed by the caller.
+ */
+ConflictType
+validate_conflict_type_and_resolver(const char *conflict_type,
+									const char *conflict_resolver)
+{
+	ConflictType type;
+	ConflictResolver resolver;
+	bool		valid = false;
+	int			i;
+
+	/* Check conflict type validity */
+	for (type = CT_MIN; type <= CT_MAX; type++)
+	{
+		if (pg_strcasecmp(ConflictTypeNames[type], 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));
+
+	/* Reset */
+	valid = false;
+
+	/* Check conflict resolver validity. */
+	for (resolver = CR_MIN; resolver <= CR_MAX; 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));
+
+	/* Reset */
+	valid = false;
+
+	/* Check if conflict resolver is a valid one for the given conflict type */
+	for (i = 0; i < CONFLICT_TYPE_MAX_RESOLVERS; i++)
+	{
+		if (ConflictTypeResolverMap[type][i] == resolver)
+		{
+			valid = true;
+			break;
+		}
+	}
+
+	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;
+
+}
+
+/*
+ * Extract the conflict type and conflict resolvers from the
+ * ALTER SUBSCRIPTION command and return a list of ConflictTypeResolver nodes.
+ */
+List *
+GetAndValidateSubsConflictResolverList(List *stmtresolvers)
+{
+	ListCell   *lc;
+	ConflictTypeResolver *conftyperesolver = NULL;
+	List	   *res = NIL;
+	List       *conflictTypes = NIL;
+
+	foreach(lc, stmtresolvers)
+	{
+		DefElem    *defel = (DefElem *) lfirst(lc);
+		char 	   *resolver_str;
+
+		/* Check if the conflict type already exists in the list */
+		if (list_member(conflictTypes, makeString(defel->defname)))
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("duplicate conflict type \"%s\" found", defel->defname)));
+		}
+
+		conftyperesolver = palloc(sizeof(ConflictTypeResolver));
+		conftyperesolver->conflict_type = downcase_truncate_identifier(defel->defname,
+											strlen(defel->defname), false);
+		resolver_str = defGetString(defel);
+		conftyperesolver->resolver = downcase_truncate_identifier(resolver_str,
+										strlen(resolver_str), false);
+
+		/*
+		 * Validate the conflict type and that the resolver is valid for that
+		 * conflict type
+		 */
+		validate_conflict_type_and_resolver(
+										conftyperesolver->conflict_type,
+										conftyperesolver->resolver);
+
+		/* Add the conflict type to the list of seen types */
+		conflictTypes = lappend(conflictTypes,
+								makeString((char *) conftyperesolver->conflict_type));
+
+		/* Add the validated ConflictTypeResolver to the result list */
+		res = lappend(res, conftyperesolver);
+	}
+
+	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)
+{
+	ListCell   *lc;
+	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;
+
+	/* 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(lc, conflict_resolvers)
+	{
+		ConflictTypeResolver *conftyperesolver = (ConflictTypeResolver *) lfirst(lc);
+
+		/* set up subid and conflict_type to search in cache */
+		values[Anum_pg_subscription_conflict_confsubid - 1] = ObjectIdGetDatum(subid);
+		values[Anum_pg_subscription_conflict_confrtype - 1] =
+			CStringGetTextDatum(conftyperesolver->conflict_type);
+
+		oldtup = SearchSysCache2(SUBSCRIPTIONCONFLMAP,
+								 values[Anum_pg_subscription_conflict_confsubid - 1],
+								 values[Anum_pg_subscription_conflict_confrtype - 1]);
+
+		if (HeapTupleIsValid(oldtup))
+		{
+			/* Update the new resolver */
+			values[Anum_pg_subscription_conflict_confrres - 1] =
+						CStringGetTextDatum(conftyperesolver->resolver);
+			replaces[Anum_pg_subscription_conflict_confrres - 1] = true;
+
+			newtup = heap_modify_tuple(oldtup, RelationGetDescr(pg_subscription_conflict),
+									   values, nulls, replaces);
+			CatalogTupleUpdate(pg_subscription_conflict, &oldtup->t_self, newtup);
+			ReleaseSysCache(oldtup);
+			heap_freetuple(newtup);
+		}
+		else
+			elog(ERROR, "cache lookup failed for table conflict %s for subid %u",
+				 conftyperesolver->conflict_type, subid);
+
+	}
+
+	table_close(pg_subscription_conflict, RowExclusiveLock);
+}
+
+/*
+ * Reset the conflict resolver for this conflict type to its default setting.
+ */
+void
+ResetConflictResolver(Oid subid, char *conflict_type)
+{
+	ConflictType idx;
+	ConflictTypeResolver conflictResolver;
+	Datum		values[Natts_pg_subscription_conflict];
+	bool		nulls[Natts_pg_subscription_conflict];
+	bool		replaces[Natts_pg_subscription_conflict];
+	Relation	pg_subscription_conflict;
+	HeapTuple	oldtup,
+				newtup;
+	bool		valid = false;
+	char	   *cur_conflict_res;
+	Datum		datum;
+
+	/* Get the index for this conflict_type */
+	for (idx = CT_MIN; idx <= CT_MAX; idx++)
+	{
+		if (pg_strcasecmp(ConflictTypeNames[idx], 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));
+
+	/* Get the default resolver for this conflict_type. */
+	conflictResolver.resolver = ConflictResolverNames[ConflictTypeDefaultResolvers[idx]];
+
+	/* 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);
+
+	values[Anum_pg_subscription_conflict_confsubid - 1] = ObjectIdGetDatum(subid);
+	values[Anum_pg_subscription_conflict_confrtype - 1] = CStringGetTextDatum(conflict_type);
+
+	oldtup = SearchSysCache2(SUBSCRIPTIONCONFLMAP,
+							 values[Anum_pg_subscription_conflict_confsubid - 1],
+							 values[Anum_pg_subscription_conflict_confrtype - 1]);
+
+
+	if (!HeapTupleIsValid(oldtup))
+		elog(ERROR, "cache lookup failed for table conflict %s for subid %u", conflict_type, subid);
+
+	datum = SysCacheGetAttrNotNull(SUBSCRIPTIONCONFLMAP,
+								   oldtup, Anum_pg_subscription_conflict_confrres);
+	cur_conflict_res = TextDatumGetCString(datum);
+
+	/* Check if current resolver is the default one, if not update it. */
+	if (pg_strcasecmp(cur_conflict_res, conflictResolver.resolver) != 0)
+	{
+		/* Update the new resolver */
+		values[Anum_pg_subscription_conflict_confrres - 1] =
+			CStringGetTextDatum(conflictResolver.resolver);
+		replaces[Anum_pg_subscription_conflict_confrres - 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);
+}
+
+/*
+ * Report a warning about incomplete conflict detection and resolution if
+ * track_commit_timestamp is disabled.
+ */
+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("Conflicts update_origin_differs and delete_origin_differs cannot be "
+					  "detected, and the origin and commit timestamp for the local row "
+					  "will not be logged."));
+}
+
+/*
+ * Set Conflict Resolvers on the subscription
+ */
+void
+SetSubConflictResolver(Oid subId, ConflictTypeResolver * resolvers, int resolvers_cnt)
+{
+	Relation	pg_subscription_conflict;
+	Datum		values[Natts_pg_subscription_conflict];
+	bool		nulls[Natts_pg_subscription_conflict];
+	HeapTuple	newtup = NULL;
+	int			idx;
+	Oid			conflict_oid;
+
+	pg_subscription_conflict = table_open(SubscriptionConflictId, RowExclusiveLock);
+
+	/* Prepare to update a tuple. */
+	memset(nulls, false, sizeof(nulls));
+
+	for (idx = 0; idx < resolvers_cnt; idx++)
+	{
+		values[Anum_pg_subscription_conflict_confsubid - 1] = ObjectIdGetDatum(subId);
+		values[Anum_pg_subscription_conflict_confrtype - 1] =
+			CStringGetTextDatum(resolvers[idx].conflict_type);
+		values[Anum_pg_subscription_conflict_confrres - 1] =
+			CStringGetTextDatum(resolvers[idx].resolver);
+
+		/* 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
+RemoveSubscriptionConflictResolvers(Oid subid)
+{
+	Relation	rel;
+	HeapTuple	tup;
+	TableScanDesc scan;
+	ScanKeyData skey[1];
+
+	rel = table_open(SubscriptionConflictId, RowExclusiveLock);
+
+	/*
+	 * Search using the subid, this should return all conflict resolvers for
+	 * this sub
+	 */
+	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 546e7e4..27f9f30 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4827,7 +4827,9 @@ getSubscriptions(Archive *fout)
 {
 	DumpOptions *dopt = fout->dopt;
 	PQExpBuffer query;
-	PGresult   *res;
+	PQExpBuffer confQuery;
+	PGresult    *res;
+	PGresult    *confRes;
 	SubscriptionInfo *subinfo;
 	int			i_tableoid;
 	int			i_oid;
@@ -4848,7 +4850,9 @@ getSubscriptions(Archive *fout)
 	int			i_subenabled;
 	int			i_subfailover;
 	int			i,
-				ntups;
+				j,
+				ntups,
+				ntuples;
 
 	if (dopt->no_subscriptions || fout->remoteVersion < 100000)
 		return;
@@ -5009,6 +5013,67 @@ 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 confrtype, confrres FROM pg_catalog.pg_subscription_conflict "
+						  "WHERE confsubid = %u;", subinfo[i].dobj.catId.oid);
+		confRes = ExecuteSqlQuery(fout, confQuery->data, PGRES_TUPLES_OK);
+
+		ntuples = PQntuples(confRes);
+		for (j = 0; j < ntuples; j++)
+		{
+			char *confType = PQgetvalue(confRes, j, 0);
+			char *confResVal = PQgetvalue(confRes, j, 1);
+
+			if (strcmp(confType, "delete_missing") == 0)
+			{
+				subinfo[i].conflict_resolver[CT_DELETE_MISSING].resolver =
+																pg_strdup(confResVal);
+				subinfo[i].conflict_resolver[CT_DELETE_MISSING].conflict_type =
+																pg_strdup(confType);
+			}
+			else if (strcmp(confType, "delete_origin_differs") == 0)
+			{
+				subinfo[i].conflict_resolver[CT_DELETE_ORIGIN_DIFFERS].resolver =
+																pg_strdup(confResVal);
+				subinfo[i].conflict_resolver[CT_DELETE_ORIGIN_DIFFERS].conflict_type =
+																pg_strdup(confType);
+			}
+			else if (strcmp(confType, "insert_exists") == 0)
+			{
+				subinfo[i].conflict_resolver[CT_INSERT_EXISTS].resolver =
+																pg_strdup(confResVal);
+				subinfo[i].conflict_resolver[CT_INSERT_EXISTS].conflict_type =
+																pg_strdup(confType);
+			}
+			else if (strcmp(confType, "update_exists") == 0)
+			{
+				subinfo[i].conflict_resolver[CT_UPDATE_EXISTS].resolver =
+																pg_strdup(confResVal);
+				subinfo[i].conflict_resolver[CT_UPDATE_EXISTS].conflict_type =
+																pg_strdup(confType);
+			}
+			else if (strcmp(confType, "update_missing") == 0)
+			{
+				subinfo[i].conflict_resolver[CT_UPDATE_MISSING].resolver =
+																pg_strdup(confResVal);
+				subinfo[i].conflict_resolver[CT_UPDATE_MISSING].conflict_type =
+																pg_strdup(confType);
+			}
+			else if (strcmp(confType, "update_origin_differs") == 0)
+			{
+				subinfo[i].conflict_resolver[CT_UPDATE_ORIGIN_DIFFERS].resolver =
+																pg_strdup(confResVal);
+				subinfo[i].conflict_resolver[CT_UPDATE_ORIGIN_DIFFERS].conflict_type =
+																pg_strdup(confType);
+			}
+
+		}
+
+		PQclear(confRes);
+		destroyPQExpBuffer(confQuery);
+
 		/* Decide whether we want to dump it */
 		selectDumpableObject(&(subinfo[i].dobj), fout);
 	}
@@ -5254,6 +5319,33 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 	if (pg_strcasecmp(subinfo->suborigin, LOGICALREP_ORIGIN_ANY) != 0)
 		appendPQExpBuffer(query, ", origin = %s", subinfo->suborigin);
 
+	/* add conflict resolvers, if any */
+	if (fout->remoteVersion >= 180000)
+	{
+		ConflictType idx;
+		bool		 first_resolver = true;
+
+		for (idx = 0; idx < CONFLICT_NUM_TYPES; idx++)
+		{
+			if (strcmp(subinfo->conflict_resolver[idx].resolver,
+					ConflictResolverNames[ConflictTypeDefaultResolvers[idx]]) != 0)
+			{
+				if (first_resolver)
+				{
+					appendPQExpBuffer(query, ") CONFLICT RESOLVER (%s = '%s'",
+									  subinfo->conflict_resolver[idx].conflict_type,
+									  subinfo->conflict_resolver[idx].resolver);
+					first_resolver = false;
+				}
+				else
+					appendPQExpBuffer(query,", %s = '%s'",
+									  subinfo->conflict_resolver[idx].conflict_type,
+									  subinfo->conflict_resolver[idx].resolver);
+			}
+		}
+
+	}
+
 	appendPQExpBufferStr(query, ");\n");
 
 	/*
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 0b7d21b..13625c6 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -654,24 +654,112 @@ typedef struct _PublicationSchemaInfo
 /*
  * The SubscriptionInfo struct is used to represent subscription.
  */
+#define CONFLICT_NUM_TYPES 6
+
+/*
+ * Conflict types that could occur while applying remote changes.
+ */
+typedef enum
+{
+	/* The row to be inserted violates unique constraint */
+	CT_INSERT_EXISTS,
+
+	/* The row to be updated was modified by a different origin */
+	CT_UPDATE_ORIGIN_DIFFERS,
+
+	/* The updated row value violates unique constraint */
+	CT_UPDATE_EXISTS,
+
+	/* The row to be updated is missing */
+	CT_UPDATE_MISSING,
+
+	/* The row to be deleted was modified by a different origin */
+	CT_DELETE_ORIGIN_DIFFERS,
+
+	/* The row to be deleted is missing */
+	CT_DELETE_MISSING,
+
+	/*
+	 * Other conflicts, such as exclusion constraint violations, involve more
+	 * complex rules than simple equality checks. These conflicts are left for
+	 * future improvements.
+	 */
+} ConflictType;
+
+/*
+ * Conflict resolvers that can be used to resolve various conflicts.
+ *
+ * See ConflictTypeResolverMap in conflcit.c to find out which all
+ * resolvers are supported for each conflict type.
+ */
+typedef enum ConflictResolver
+{
+	/* Apply the remote change */
+	CR_APPLY_REMOTE = 1,
+
+	/* Keep the local change */
+	CR_KEEP_LOCAL,
+
+	/* Apply the remote change; skip if it can not be applied */
+	CR_APPLY_OR_SKIP,
+
+	/* Apply the remote change; emit error if it can not be applied */
+	CR_APPLY_OR_ERROR,
+
+	/* Skip applying the change */
+	CR_SKIP,
+
+	/* Error out */
+	CR_ERROR,
+}ConflictResolver;
+
+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"
+};
+
+/*
+ * Default conflict resolver for each conflict type.
+ */
+static const int ConflictTypeDefaultResolvers[] = {
+	[CT_INSERT_EXISTS] = CR_ERROR,
+	[CT_UPDATE_EXISTS] = CR_ERROR,
+	[CT_UPDATE_ORIGIN_DIFFERS] = CR_APPLY_REMOTE,
+	[CT_UPDATE_MISSING] = CR_SKIP,
+	[CT_DELETE_MISSING] = CR_SKIP,
+	[CT_DELETE_ORIGIN_DIFFERS] = CR_APPLY_REMOTE
+
+};
+
+typedef struct ConflictTypeResolver
+{
+	const char *conflict_type;
+	const char *resolver;
+}ConflictTypeResolver;
+
 typedef struct _SubscriptionInfo
 {
 	DumpableObject dobj;
-	const char *rolname;
-	char	   *subenabled;
-	char	   *subbinary;
-	char	   *substream;
-	char	   *subtwophasestate;
-	char	   *subdisableonerr;
-	char	   *subpasswordrequired;
-	char	   *subrunasowner;
-	char	   *subconninfo;
-	char	   *subslotname;
-	char	   *subsynccommit;
-	char	   *subpublications;
-	char	   *suborigin;
-	char	   *suboriginremotelsn;
-	char	   *subfailover;
+	const char				*rolname;
+	char					*subenabled;
+	char					*subbinary;
+	char					*substream;
+	char					*subtwophasestate;
+	char					*subdisableonerr;
+	char					*subpasswordrequired;
+	char					*subrunasowner;
+	char					*subconninfo;
+	char					*subslotname;
+	char					*subsynccommit;
+	char					*subpublications;
+	char					*suborigin;
+	char					*suboriginremotelsn;
+	char					*subfailover;
+	ConflictTypeResolver	 conflict_resolver[CONFLICT_NUM_TYPES];
 } SubscriptionInfo;
 
 /*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 7c9a1f2..c577b3d 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6539,7 +6539,7 @@ describeSubscriptions(const char *pattern, bool verbose)
 	printQueryOpt myopt = pset.popt;
 	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)
 	{
@@ -6619,12 +6619,20 @@ describeSubscriptions(const char *pattern, bool verbose)
 			appendPQExpBuffer(&buf,
 							  ", subskiplsn AS \"%s\"\n",
 							  gettext_noop("Skip LSN"));
+
+		/* Add conflict resolvers information from pg_subscription_conflict */
+		if (pset.sversion >= 180000)
+			appendPQExpBuffer(&buf,
+							  ", (SELECT string_agg(confrtype || ' = ' || confrres, ', ') \n"
+							  "   FROM pg_catalog.pg_subscription_conflict c \n"
+							  "   WHERE c.confsubid = s.oid) AS \"%s\"\n",
+							  gettext_noop("Conflict Resolvers"));
 	}
 
 	/* 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())");
 
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index a7ccde6..fa3245e 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -3338,8 +3338,12 @@ psql_completion(const char *text, int start, int end)
 	{
 		/* complete with nothing here as this refers to remote publications */
 	}
+	/* Complete "CREATE SUBSCRIPTION <name> ... CONFLICT RESOLVER ( <confoptions> ) */
 	else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("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 (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("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..cf35e0d
--- /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		confrtype BKI_FORCE_NOT_NULL;	/* conflict type */
+	text		confrres 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_confrtype_index, 8885, SubscriptionConflictSubIdTypeIndexId, pg_subscription_conflict, btree(confsubid oid_ops, confrtype text_ops));
+
+MAKE_SYSCACHE(SUBSCRIPTIONCONFLMAP, pg_subscription_conflict_confsubid_confrtype_index, 256);
+
+#endif							/* PG_SUBSCRIPTION_CONFLICT_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index d6f7e79..eddc457 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -4197,6 +4197,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
@@ -4209,6 +4210,9 @@ typedef enum AlterSubscriptionType
 	ALTER_SUBSCRIPTION_REFRESH,
 	ALTER_SUBSCRIPTION_ENABLED,
 	ALTER_SUBSCRIPTION_SKIP,
+	ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS,
+	ALTER_SUBSCRIPTION_RESET_ALL_CONFLICT_RESOLVERS,
+	ALTER_SUBSCRIPTION_RESET_CONFLICT_RESOLVER,
 } AlterSubscriptionType;
 
 typedef struct AlterSubscriptionStmt
@@ -4219,6 +4223,8 @@ 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 resolvers */
+	char	   *conflict_type;	/* conflict_type to be reset */
 } AlterSubscriptionStmt;
 
 typedef struct DropSubscriptionStmt
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index f865907..d661a06 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -375,6 +375,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..296270a 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -50,6 +50,47 @@ typedef enum
 
 #define CONFLICT_NUM_TYPES (CT_DELETE_MISSING + 1)
 
+/* Min and max conflict type */
+#define CT_MIN CT_INSERT_EXISTS
+#define CT_MAX CT_DELETE_MISSING
+
+/*
+ * 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 = 1,
+
+	/* Keep the local change */
+	CR_KEEP_LOCAL,
+
+	/* Apply the remote change; skip if it can not be applied */
+	CR_APPLY_OR_SKIP,
+
+	/* Apply the remote change; emit error if it can not be applied */
+	CR_APPLY_OR_ERROR,
+
+	/* Skip applying the change */
+	CR_SKIP,
+
+	/* Error out */
+	CR_ERROR,
+} ConflictResolver;
+
+/* Min and max conflict resolver */
+#define CR_MIN CR_APPLY_REMOTE
+#define CR_MAX CR_ERROR
+
+typedef struct ConflictTypeResolver
+{
+	const char *conflict_type;
+	const char *resolver;
+}			ConflictTypeResolver;
+
 extern bool GetTupleTransactionInfo(TupleTableSlot *localslot,
 									TransactionId *xmin,
 									RepOriginId *localorigin,
@@ -62,5 +103,15 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
 								Oid indexoid, TransactionId localxmin,
 								RepOriginId localorigin, TimestampTz localts);
 extern void InitConflictIndexes(ResultRelInfo *relInfo);
+extern void SetSubConflictResolver(Oid subId, ConflictTypeResolver * resolvers, int max_types);
+extern void RemoveSubscriptionConflictById(Oid confid);
+extern void RemoveSubscriptionConflictResolvers(Oid confid);
+extern List *GetAndValidateSubsConflictResolverList(List *stmtresolvers);
+extern void UpdateSubConflictResolvers(List *conflict_resolvers, Oid subid);
+extern ConflictType validate_conflict_type_and_resolver(const char *conflict_type,
+														const char *conflict_resolver);
+extern void SetDefaultResolvers(ConflictTypeResolver * conflictResolvers);
+extern void ResetConflictResolver(Oid subid, char *conflict_type);
+extern void conf_detection_check_prerequisites(void);
 
 #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..595f496 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
 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+ regress_testsub4
-                                                                                                                 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_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | none   | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | none   | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
 \dRs+ regress_testsub4
-                                                                                                                 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_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub3;
@@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
 ERROR:  invalid connection string syntax: missing "=" after "foobar" in connection info string
 
 \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
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
 ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
 ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
 \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       | {testpub2,testpub3} | f      | off       | d                | f                | any    | f                 | t             | f        | off                | dbname=regress_doesnotexist2 | 0/0
+                                                                                                                                                                                                         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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | f                 | t             | f        | off                | dbname=regress_doesnotexist2 | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -176,10 +176,10 @@ ERROR:  unrecognized subscription parameter: "create_slot"
 -- ok
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
 \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       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/12345
+                                                                                                                                                                                                         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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/12345  | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 -- ok - with lsn = NONE
@@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
 ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
 ERROR:  invalid WAL location (LSN): 0/0
 \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       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/0
+                                                                                                                                                                                                         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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist2 | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 BEGIN;
@@ -223,10 +223,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
 ERROR:  invalid value for parameter "synchronous_commit": "foobar"
 HINT:  Available values: local, remote_write, remote_apply, on, off.
 \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_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | local              | dbname=regress_doesnotexist2 | 0/0
+                                                                                                                                                                                                           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 |                                                                           Conflict Resolvers                                                                           
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub_foo | regress_subscription_user | f       | {testpub2,testpub3} | f      | off       | d                | f                | any    | t                 | f             | f        | local              | dbname=regress_doesnotexist2 | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 -- rename back to keep the rest simple
@@ -255,19 +255,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 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}   | t      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | t      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (binary = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \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
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -279,27 +279,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 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      | on        | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
 \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      | parallel  | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | parallel  | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 \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
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 -- fail - publication already exists
@@ -314,10 +314,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
 ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
 ERROR:  publication "testpub1" is already in subscription "regress_testsub"
 \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,testpub1,testpub2} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                             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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub,testpub1,testpub2} | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 -- fail - publication used more than once
@@ -332,10 +332,10 @@ ERROR:  publication "testpub3" is not in subscription "regress_testsub"
 -- ok - delete publications
 ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
 \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
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 DROP SUBSCRIPTION regress_testsub;
@@ -371,19 +371,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 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       | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 -- we can alter streaming when two_phase enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 \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      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -393,14 +393,146 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 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      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
+-- fail - invalid conflict resolvers
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = foo) WITH (connect = false);
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+ERROR:  foo is not a valid conflict resolver
+-- fail - invalid conflict types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (foo = 'keep_local') WITH (connect = false);
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+ERROR:  foo is not a valid conflict type
+-- 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);
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+ERROR:  apply_remote is not a valid conflict resolver for conflict type update_missing
+-- fail - duplicate conflict types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = 'keep_local', insert_exists = 'keep_local');
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+ERROR:  duplicate conflict type "insert_exists" found
+-- creating subscription should create 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.
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+       confrtype       |   confrres   
+-----------------------+--------------
+ delete_missing        | skip
+ delete_origin_differs | apply_remote
+ insert_exists         | error
+ update_exists         | error
+ update_missing        | skip
+ update_origin_differs | apply_remote
+(6 rows)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- ok - valid conflict type 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:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+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
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+       confrtype       |   confrres   
+-----------------------+--------------
+ delete_missing        | skip
+ delete_origin_differs | keep_local
+ insert_exists         | keep_local
+ update_exists         | error
+ update_missing        | skip
+ update_origin_differs | apply_remote
+(6 rows)
+
+-- fail - altering with invalid conflict type
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (foo = 'keep_local');
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+ERROR:  foo is not a valid conflict type
+-- fail - altering with invalid conflict resolver
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'foo');
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+ERROR:  foo is not a valid conflict resolver
+-- fail - altering with duplicate conflict types
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'apply_remote', insert_exists = 'apply_remote');
+WARNING:  conflict detection and resolution could be incomplete due to disabled track_commit_timestamp
+DETAIL:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+ERROR:  duplicate conflict type "insert_exists" found
+-- 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:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+       confrtype       |   confrres   
+-----------------------+--------------
+ delete_missing        | skip
+ delete_origin_differs | keep_local
+ insert_exists         | apply_remote
+ update_exists         | error
+ update_missing        | skip
+ update_origin_differs | apply_remote
+(6 rows)
+
+-- 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:  Conflicts update_origin_differs and delete_origin_differs cannot be detected, and the origin and commit timestamp for the local row will not be logged.
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+       confrtype       |   confrres   
+-----------------------+--------------
+ delete_missing        | error
+ delete_origin_differs | keep_local
+ insert_exists         | apply_remote
+ update_exists         | keep_local
+ update_missing        | skip
+ update_origin_differs | error
+(6 rows)
+
+-- 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';
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+       confrtype       |  confrres  
+-----------------------+------------
+ delete_missing        | error
+ delete_origin_differs | keep_local
+ insert_exists         | error
+ update_exists         | keep_local
+ update_missing        | skip
+ update_origin_differs | error
+(6 rows)
+
+-- ok - reset ALL
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER ALL;
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+       confrtype       |   confrres   
+-----------------------+--------------
+ delete_missing        | skip
+ delete_origin_differs | apply_remote
+ insert_exists         | error
+ update_exists         | error
+ update_missing        | skip
+ update_origin_differs | apply_remote
+(6 rows)
+
+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);
 ERROR:  disable_on_error requires a Boolean value
@@ -409,18 +541,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 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
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
 \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                | t                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
+                                                                                                                                                                                                     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 |                                                                           Conflict Resolvers                                                                           
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | t                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0      | insert_exists = error, update_origin_differs = apply_remote, update_exists = error, update_missing = skip, delete_origin_differs = apply_remote, delete_missing = skip
 (1 row)
 
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 007c9e7..596f2e1 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -272,6 +272,63 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 
+-- fail - invalid conflict resolvers
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = foo) WITH (connect = false);
+
+-- fail - invalid conflict types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (foo = 'keep_local') 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 types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub CONFLICT RESOLVER (insert_exists = 'keep_local', insert_exists = 'keep_local');
+
+-- creating subscription should create default conflict resolvers
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
+-- ok - valid conflict type 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
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+
+-- 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 types
+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' );
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+
+-- ok - valid conflict types and resolvers
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (update_exists = 'keep_local', delete_missing = 'error', update_origin_differs = 'error');
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+
+-- 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';
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+
+-- ok - reset ALL
+ALTER SUBSCRIPTION regress_testsub RESET CONFLICT RESOLVER ALL;
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+
+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 e9ebddd..ebb7b9f 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -469,6 +469,7 @@ ConditionalStack
 ConfigData
 ConfigVariable
 ConflictType
+ConflictTypeResolver
 ConnCacheEntry
 ConnCacheKey
 ConnParams
@@ -863,6 +864,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

