From 8388df6e8ac066b655a8ffbc502acc9ea9c071a9 Mon Sep 17 00:00:00 2001
From: Nisha Moond <nisha.moond412@gmail.com>
Date: Wed, 21 Aug 2024 13:50:52 +0530
Subject: [PATCH v9 1/3] 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,...);
---
 doc/src/sgml/logical-replication.sgml         |  89 +----
 doc/src/sgml/ref/alter_subscription.sgml      |  12 +
 doc/src/sgml/ref/create_subscription.sgml     | 167 ++++++++++
 src/backend/commands/subscriptioncmds.c       |  55 ++++
 src/backend/parser/gram.y                     |  25 +-
 src/backend/replication/logical/conflict.c    | 308 ++++++++++++++++++
 src/include/catalog/Makefile                  |   3 +-
 src/include/catalog/meson.build               |   1 +
 .../catalog/pg_subscription_conflict.h        |  55 ++++
 src/include/nodes/parsenodes.h                |   3 +
 src/include/parser/kwlist.h                   |   1 +
 src/include/replication/conflict.h            |  49 +++
 src/test/regress/expected/oidjoins.out        |   1 +
 src/test/regress/expected/subscription.out    |  61 ++++
 src/test/regress/sql/subscription.sql         |  32 ++
 15 files changed, 778 insertions(+), 84 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 885a2d70ae..f3fb53668c 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1571,7 +1571,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
@@ -1585,85 +1585,14 @@ test_sub=# SELECT * FROM t1 ORDER BY id;
   </para>
 
   <para>
-   Additional logging is triggered in the following <firstterm>conflict</firstterm>
-   cases:
-   <variablelist>
-    <varlistentry>
-     <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>
-     <term><literal>update_differ</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. Currenly, the update is always applied
-       regardless of the origin of the local row.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry>
-     <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>
-     <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>
-     <term><literal>delete_differ</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. Currenly, the delete is always applied
-       regardless of the origin of the local row.
-      </para>
-     </listitem>
-    </varlistentry>
-    <varlistentry>
-     <term><literal>delete_missing</literal></term>
-     <listitem>
-      <para>
-       The tuple to be deleted was not found. The delete will simply be
-       skipped in this scenario.
-      </para>
-     </listitem>
-    </varlistentry>
-   </variablelist>
-    Note that there are other conflict scenarios, such as exclusion constraint
-    violations. Currently, we do not provide additional details for them in the
-    log.
+   Additional logging is triggered for specific <literal>conflict_resolvers</literal>.
+   Users can also configure <literal>conflict_types</literal> while creating
+   the subscription.
+   Refer to section <link linkend="sql-createsubscription-params-with-conflict-resolver"><literal>CONFLICT RESOLVERS</literal></link>
+   for details on conflict_types and conflict_resolvers.
+
+   Note that there are other conflict scenarios, such as exclusion constraint
+   violations. Currently, we do not provide additional details for them in the log.
   </para>
 
   <para>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
old mode 100644
new mode 100755
index fdc648d007..e7b39f2000
--- 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 740b7d9421..cacce6979e
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -25,6 +25,7 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
     CONNECTION '<replaceable class="parameter">conninfo</replaceable>'
     PUBLICATION <replaceable class="parameter">publication_name</replaceable> [, ...]
     [ WITH ( <replaceable class="parameter">subscription_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
+    [ CONFLICT RESOLVER ( <replaceable class="parameter">conflict_type</replaceable> [= <replaceable class="parameter">conflict_resolver</replaceable>] [, ...] ) ]
 </synopsis>
  </refsynopsisdiv>
 
@@ -432,6 +433,172 @@ 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-differ">
+        <term><literal>update_differ</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-delete-differ">
+        <term><literal>delete_differ</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-remote-apply">
+        <term><literal>remote_apply</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_differ</literal> and <literal>delete_differ</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_differ</literal> and <literal>delete_differ</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 b925c464ae..fe7e3ae51c 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -28,6 +28,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"
@@ -37,6 +38,7 @@
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
 #include "pgstat.h"
+#include "replication/conflict.h"
 #include "replication/logicallauncher.h"
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
@@ -439,6 +441,32 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 	}
 }
 
+/*
+ * Parsing function for conflict resolvers in CREATE SUBSCRIPTION command
+ */
+static void
+parse_subscription_conflict_resolvers(List *stmtresolvers,
+									  ConflictTypeResolver * resolvers)
+{
+	ListCell   *lc;
+
+	if (!stmtresolvers)
+		return;
+
+	foreach(lc, stmtresolvers)
+	{
+		DefElem    *defel = (DefElem *) lfirst(lc);
+		ConflictType type;
+
+		/* validate the conflict type and resolver */
+		type = validate_conflict_type_and_resolver(defel->defname,
+												   defGetString(defel));
+
+		/* update the corresponding resolver for the given conflict type */
+		resolvers[type].resolver = defGetString(defel);
+	}
+}
+
 /*
  * Add publication names from the list to a string.
  */
@@ -583,6 +611,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 +626,12 @@ 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. Initialize with default values
+	 */
+	SetDefaultResolvers(conflictResolvers);
+	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
@@ -726,6 +761,9 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
 	replorigin_create(originname);
 
+	/* Update the Conflict Resolvers in pg_subscription_conflict */
+	SetSubConflictResolver(subid, conflictResolvers, CT_MAX + 1);
+
 	/*
 	 * Connect to remote side to execute requested commands and fetch table
 	 * info.
@@ -1581,6 +1619,20 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				update_tuple = true;
 				break;
 			}
+		case ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS:
+			{
+				List	   *conflict_resolvers = NIL;
+
+				/* get 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;
+			}
 
 		default:
 			elog(ERROR, "unrecognized ALTER SUBSCRIPTION kind %d",
@@ -1832,6 +1884,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	/* Remove any associated relation synchronization states. */
 	RemoveSubscriptionRel(subid, InvalidOid);
 
+	/* Remove any associated conflict resolvers */
+	RemoveSubscriptionConflictBySubid(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 c3f25582c3..628feefd92 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -426,7 +426,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
@@ -772,7 +772,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
@@ -8806,6 +8806,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);
@@ -10740,7 +10745,7 @@ AlterPublicationStmt:
  *****************************************************************************/
 
 CreateSubscriptionStmt:
-			CREATE SUBSCRIPTION name CONNECTION Sconst PUBLICATION name_list opt_definition
+	CREATE SUBSCRIPTION name CONNECTION Sconst PUBLICATION name_list opt_definition opt_resolver_definition
 				{
 					CreateSubscriptionStmt *n =
 						makeNode(CreateSubscriptionStmt);
@@ -10748,6 +10753,7 @@ CreateSubscriptionStmt:
 					n->conninfo = $5;
 					n->publication = $7;
 					n->options = $8;
+					n->resolvers = $9;
 					$$ = (Node *) n;
 				}
 		;
@@ -10854,6 +10860,17 @@ 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;
+				}
+
 		;
 
 /*****************************************************************************
@@ -17778,6 +17795,7 @@ unreserved_keyword:
 			| REPLACE
 			| REPLICA
 			| RESET
+			| RESOLVER
 			| RESTART
 			| RESTRICT
 			| RETURN
@@ -18408,6 +18426,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 0bc7959980..ad79f816dd 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -15,12 +15,26 @@
 #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 "replication/conflict.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
 #include "replication/logicalrelation.h"
 #include "storage/lmgr.h"
 #include "utils/lsyscache.h"
+#include "utils/syscache.h"
 
 static const char *const ConflictTypeNames[] = {
 	[CT_INSERT_EXISTS] = "insert_exists",
@@ -31,6 +45,55 @@ static const char *const ConflictTypeNames[] = {
 	[CT_DELETE_MISSING] = "delete_missing"
 };
 
+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.
+ */
+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_DIFFER] = {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_DIFFER] = {CR_APPLY_REMOTE, CR_KEEP_LOCAL, CR_ERROR}
+};
+
+/*
+ * Default conflict resolver for each conflict type.
+ */
+const int	ConflictTypeDefaultResolvers[] = {
+	[CT_INSERT_EXISTS] = CR_ERROR,
+	[CT_UPDATE_EXISTS] = CR_ERROR,
+	[CT_UPDATE_DIFFER] = CR_APPLY_REMOTE,
+	[CT_UPDATE_MISSING] = CR_SKIP,
+	[CT_DELETE_MISSING] = CR_SKIP,
+	[CT_DELETE_DIFFER] = CR_APPLY_REMOTE
+
+};
+
 static int	errcode_apply_conflict(ConflictType type);
 static int	errdetail_apply_conflict(EState *estate,
 									 ResultRelInfo *relinfo,
@@ -486,3 +549,248 @@ 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.
+ */
+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 (strcmp(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 (strcmp(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 *CTR = NULL;
+	List	   *res = NIL;
+
+	foreach(lc, stmtresolvers)
+	{
+		DefElem    *defel = (DefElem *) lfirst(lc);
+
+		CTR = palloc(sizeof(ConflictTypeResolver));
+		CTR->conflict_type = defel->defname;
+		CTR->resolver = defGetString(defel);
+
+		/*
+		 * Validate the conflict type and that the resolver is valid for that
+		 * conflict type
+		 */
+		validate_conflict_type_and_resolver(CTR->conflict_type, CTR->resolver);
+
+		res = lappend(res, CTR);
+	}
+
+	return res;
+}
+
+/*
+ * Update the Subscription's conflict resolver for a conflict type in
+ * pg_subscription_conflict system catalog
+ */
+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 *CTR = (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(CTR->conflict_type);
+
+		oldtup = SearchSysCache2(SUBSCRIPTIONCONFLICTSUBOID,
+								 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(CTR->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",
+				 CTR->conflict_type, subid);
+
+	}
+
+	table_close(pg_subscription_conflict, RowExclusiveLock);
+}
+
+/*
+ * 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];
+	bool		replaces[Natts_pg_subscription_conflict];
+	HeapTuple	newtup = NULL;
+	int			type;
+	Oid			conflict_oid;
+
+	pg_subscription_conflict = table_open(SubscriptionConflictId, RowExclusiveLock);
+
+	for (type = 0; type < resolvers_cnt; type++)
+	{
+		/* Prepare to update a tuple. */
+		memset(nulls, false, sizeof(nulls));
+		memset(replaces, false, sizeof(replaces));
+
+		values[Anum_pg_subscription_conflict_confsubid - 1] = ObjectIdGetDatum(subId);
+		values[Anum_pg_subscription_conflict_confrtype - 1] =
+			CStringGetTextDatum(resolvers[type].conflict_type);
+		values[Anum_pg_subscription_conflict_confrres - 1] =
+			CStringGetTextDatum(resolvers[type].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
+RemoveSubscriptionConflictBySubid(Oid subid)
+{
+	Relation	rel;
+	HeapTuple	tup;
+	TableScanDesc scan;
+	ScanKeyData skey[1];
+	int			nkeys = 0;
+
+	rel = table_open(SubscriptionConflictId, RowExclusiveLock);
+
+	/*
+	 * Search using the subid, this should return all conflict resolvers for
+	 * this sub
+	 */
+	ScanKeyInit(&skey[nkeys++],
+				Anum_pg_subscription_conflict_confsubid,
+				BTEqualStrategyNumber,
+				F_OIDEQ,
+				ObjectIdGetDatum(subid));
+
+	scan = table_beginscan_catalog(rel, nkeys, 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/include/catalog/Makefile b/src/include/catalog/Makefile
index 167f91a6e3..f2611c1424 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 f70d1daba5..959e1d9ded 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 0000000000..c8b37c2bef
--- /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_sub_index, 8885, SubscriptionConflictSubIndexId, pg_subscription_conflict, btree(confsubid oid_ops, confrtype text_ops));
+
+MAKE_SYSCACHE(SUBSCRIPTIONCONFLICTSUBOID, pg_subscription_conflict_sub_index, 256);
+
+#endif							/* PG_SUBSCRIPTION_CONFLICT_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 85a62b538e..2e1eefc3a7 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -4203,6 +4203,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
@@ -4215,6 +4216,7 @@ typedef enum AlterSubscriptionType
 	ALTER_SUBSCRIPTION_REFRESH,
 	ALTER_SUBSCRIPTION_ENABLED,
 	ALTER_SUBSCRIPTION_SKIP,
+	ALTER_SUBSCRIPTION_CONFLICT_RESOLVERS,
 } AlterSubscriptionType;
 
 typedef struct AlterSubscriptionStmt
@@ -4225,6 +4227,7 @@ 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 */
 } AlterSubscriptionStmt;
 
 typedef struct DropSubscriptionStmt
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index bb191b1f46..7bffc9401b 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -376,6 +376,7 @@ PG_KEYWORD("repeatable", REPEATABLE, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("replace", REPLACE, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("replica", REPLICA, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("reset", RESET, UNRESERVED_KEYWORD, BARE_LABEL)
+PG_KEYWORD("resolver", RESOLVER, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("restart", RESTART, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("restrict", RESTRICT, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("return", RETURN, UNRESERVED_KEYWORD, BARE_LABEL)
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 02cb84da7e..fac3f8335e 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -42,6 +42,47 @@ typedef enum
 	 */
 } ConflictType;
 
+/* 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 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;
+
+/* 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,
@@ -54,5 +95,13 @@ 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 RemoveSubscriptionConflictBySubid(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);
 
 #endif
diff --git a/src/test/regress/expected/oidjoins.out b/src/test/regress/expected/oidjoins.out
index 215eb899be..42bf2cce92 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 17d48b1685..fbf355c20f 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -399,6 +399,67 @@ HINT:  To initiate replication, you must manually create the replication slot, e
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | on        | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
 (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 WITH (connect = false) CONFLICT RESOLVER (insert_exists = foo);
+ERROR:  foo is not a valid conflict resolver
+-- fail - invalid conflict types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false) CONFLICT RESOLVER (foo = 'keep_local');
+ERROR:  foo is not a valid conflict type
+-- 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_differ  | apply_remote
+ delete_missing | skip
+ insert_exists  | error
+ update_differ  | apply_remote
+ update_exists  | error
+ update_missing | skip
+(6 rows)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+--try setting resolvers for few types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false) CONFLICT RESOLVER (insert_exists = 'keep_local', update_missing = 'skip', delete_differ = 'keep_local' );
+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_differ  | keep_local
+ delete_missing | skip
+ insert_exists  | keep_local
+ update_differ  | apply_remote
+ update_exists  | error
+ update_missing | skip
+(6 rows)
+
+-- fail - altering with invalid conflict type
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (foo = 'keep_local');
+ERROR:  foo is not a valid conflict type
+-- fail - altering with invalid conflict resolver
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'foo');
+ERROR:  foo is not a valid conflict resolver
+-- ok - valid conflict type and resolver
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'apply_remote', update_missing = 'skip', delete_differ = 'keep_local' );
+SELECT confrtype, confrres FROM pg_subscription_conflict ORDER BY confrtype;
+   confrtype    |   confrres   
+----------------+--------------
+ delete_differ  | keep_local
+ delete_missing | skip
+ insert_exists  | apply_remote
+ update_differ  | apply_remote
+ update_exists  | error
+ update_missing | skip
+(6 rows)
+
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 -- fail - disable_on_error must be boolean
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 007c9e7037..b7ccc02a51 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -272,6 +272,38 @@ 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 WITH (connect = false) CONFLICT RESOLVER (insert_exists = foo);
+
+-- fail - invalid conflict types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false) CONFLICT RESOLVER (foo = '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;
+
+--try setting resolvers for few types
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false) CONFLICT RESOLVER (insert_exists = 'keep_local', update_missing = 'skip', delete_differ = 'keep_local' );
+ 
+--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');
+
+-- ok - valid conflict type and resolver
+ALTER SUBSCRIPTION regress_testsub CONFLICT RESOLVER (insert_exists = 'apply_remote', update_missing = 'skip', delete_differ = 'keep_local' );
+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);
 
-- 
2.34.1

