From fbf56a37afd3e92d4fa4da30d49cd3f8d13635de Mon Sep 17 00:00:00 2001
From: Shveta Malik <shveta.malik@gmail.com>
Date: Mon, 24 Jun 2024 09:55:15 +0530
Subject: [PATCH v3 2/5] DDL command to configure Global Conflict Resolvers

This patch adds new DDL commands to configure resolvers
at global level:

a) To set global resolver for a given conflict_type:
SET CONFLICT RESOLVER 'conflict_resolver' FOR 'conflict_type'

b) To reset to default:
RESET CONFLICT RESOLVER FOR 'conflict_type'

A new catalog table pg_conflict has been created to store
global conflict_type and conflict_resolver configurations given by
above DDL command. Initially, this catalog table holds default
configuration for resolvers which is overwritten by the ones given
by the user in 'SET CONFLICT RESOLVER' command.
'RESET CONFLICT RESOLVER' command will reset the resolver to default.

This patch provides support to configure resolvers for below conflict
types:
- insert_exists
- update_differ
- update_missing
- delete_missing
---
 src/backend/parser/gram.y                     |  42 +++-
 src/backend/replication/logical/conflict.c    | 215 +++++++++++++++++-
 src/backend/tcop/utility.c                    |  15 ++
 src/include/catalog/Makefile                  |   4 +-
 src/include/catalog/meson.build               |   2 +
 src/include/catalog/pg_conflict.dat           |  20 ++
 src/include/catalog/pg_conflict.h             |  43 ++++
 src/include/nodes/parsenodes.h                |  12 +
 src/include/parser/kwlist.h                   |   1 +
 src/include/replication/conflict.h            |  38 +++-
 src/include/tcop/cmdtaglist.h                 |   1 +
 .../regress/expected/conflict_resolver.out    |  50 ++++
 src/test/regress/parallel_schedule            |   2 +-
 src/test/regress/sql/conflict_resolver.sql    |  27 +++
 src/tools/pgindent/typedefs.list              |   2 +
 15 files changed, 467 insertions(+), 7 deletions(-)
 create mode 100644 src/include/catalog/pg_conflict.dat
 create mode 100644 src/include/catalog/pg_conflict.h
 create mode 100644 src/test/regress/expected/conflict_resolver.out
 create mode 100644 src/test/regress/sql/conflict_resolver.sql

diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index a043fd4c66..42726fe3a6 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -292,7 +292,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 		AlterRoleStmt AlterRoleSetStmt AlterPolicyStmt AlterStatsStmt
 		AlterDefaultPrivilegesStmt DefACLAction
 		AnalyzeStmt CallStmt ClosePortalStmt ClusterStmt CommentStmt
-		ConstraintsSetStmt CopyStmt CreateAsStmt CreateCastStmt
+		ConflictResolverStmt ConstraintsSetStmt CopyStmt CreateAsStmt CreateCastStmt
 		CreateDomainStmt CreateExtensionStmt CreateGroupStmt CreateOpClassStmt
 		CreateOpFamilyStmt AlterOpFamilyStmt CreatePLangStmt
 		CreateSchemaStmt CreateSeqStmt CreateStmt CreateStatsStmt CreateTableSpaceStmt
@@ -613,6 +613,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type <ival>	opt_check_option
 
 %type <str>		opt_provider security_label
+%type <str>		conflict_type conflict_resolver
 
 %type <target>	xml_attribute_el
 %type <list>	xml_attribute_list xml_attributes
@@ -772,7 +773,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 
 	RANGE READ REAL REASSIGN RECHECK 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
@@ -1039,6 +1040,7 @@ stmt:
 			| ClosePortalStmt
 			| ClusterStmt
 			| CommentStmt
+			| ConflictResolverStmt
 			| ConstraintsSetStmt
 			| CopyStmt
 			| CreateAmStmt
@@ -7266,7 +7268,41 @@ comment_text:
 			| NULL_P							{ $$ = NULL; }
 		;
 
+/*****************************************************************************
+ *
+ *  SET CONFLICT RESOLVER <conflict_resolver> FOR <conflict_type>
+ *  RESET CONFLICT RESOLVER FOR <conflict_type>
+ *
+ *****************************************************************************/
+
+ConflictResolverStmt:
+			SET CONFLICT RESOLVER conflict_resolver FOR conflict_type
+				{
+					ConflictResolverStmt *n = makeNode(ConflictResolverStmt);
+
+					n->conflict_resolver = $4;
+					n->conflict_type = $6;
+					n->isReset = false;
+					$$ = (Node *) n;
+				}
+			| RESET CONFLICT RESOLVER FOR conflict_type
+				{
+					ConflictResolverStmt *n = makeNode(ConflictResolverStmt);
 
+					n->conflict_type = $5;
+					n->isReset = true;
+					$$ = (Node *) n;
+				}
+		;
+conflict_type:
+			Sconst								{ $$ = $1; }
+			| NULL_P							{ $$ = NULL; }
+		;
+
+conflict_resolver:
+			Sconst								{ $$ = $1; }
+			| NULL_P							{ $$ = NULL; }
+		;
 /*****************************************************************************
  *
  *  SECURITY LABEL [FOR <provider>] ON <object> IS <label>
@@ -17797,6 +17833,7 @@ unreserved_keyword:
 			| REPLACE
 			| REPLICA
 			| RESET
+			| RESOLVER
 			| RESTART
 			| RESTRICT
 			| RETURN
@@ -18428,6 +18465,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 f24f048054..87b1ea0ab3 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -1,13 +1,13 @@
 /*-------------------------------------------------------------------------
  * conflict.c
- *	   Functionality for detecting and logging conflicts.
+ *	   Functionality for detecting, logging and configuring conflicts.
  *
  * Copyright (c) 2024, PostgreSQL Global Development Group
  *
  * IDENTIFICATION
  *	  src/backend/replication/logical/conflict.c
  *
- * This file contains the code for detecting and logging conflicts on
+ * This file contains the code for detecting and handling conflicts on
  * the subscriber during logical replication.
  *-------------------------------------------------------------------------
  */
@@ -15,10 +15,19 @@
 #include "postgres.h"
 
 #include "access/commit_ts.h"
+#include "access/genam.h"
+#include "access/htup_details.h"
+#include "access/skey.h"
+#include "access/table.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_conflict.h"
 #include "replication/conflict.h"
 #include "replication/origin.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
+#include "utils/timestamp.h"
 
 const char *const ConflictTypeNames[] = {
 	[CT_INSERT_EXISTS] = "insert_exists",
@@ -27,6 +36,51 @@ const char *const ConflictTypeNames[] = {
 	[CT_DELETE_MISSING] = "delete_missing"
 };
 
+const char *const ConflictResolverNames[] = {
+	[CR_REMOTE_APPLY] = "remote_apply",
+	[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 remote_apply and keep_local only. Then these
+ * resolvers in context of update_missing will mean:
+ *
+ * keep_local: do not apply the update as INSERT.
+ * remote_apply: 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_REMOTE_APPLY, CR_KEEP_LOCAL, CR_ERROR},
+	[CT_UPDATE_DIFFER] = {CR_REMOTE_APPLY, 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}
+};
+
+/*
+ * Default conflict resolver for each conflict type.
+ * If this changes, change it in pg_conflict.dat as well.
+ */
+const int	ConflictTypeDefaultResolvers[] = {
+	[CT_INSERT_EXISTS] = CR_REMOTE_APPLY,
+	[CT_UPDATE_DIFFER] = CR_REMOTE_APPLY,
+	[CT_UPDATE_MISSING] = CR_APPLY_OR_SKIP,
+	[CT_DELETE_MISSING] = CR_SKIP
+};
+
 static char *build_index_value_desc(Oid indexoid, TupleTableSlot *conflictslot);
 static int	errdetail_apply_conflict(ConflictType type, Oid conflictidx,
 									 TransactionId localxmin,
@@ -34,6 +88,7 @@ static int	errdetail_apply_conflict(ConflictType type, Oid conflictidx,
 									 TimestampTz localts,
 									 TupleTableSlot *conflictslot);
 
+
 /*
  * Get the xmin and commit timestamp data (origin and timestamp) associated
  * with the provided local tuple.
@@ -185,3 +240,159 @@ build_index_value_desc(Oid indexoid, TupleTableSlot *conflictslot)
 
 	return conflict_row;
 }
+
+/*
+ * Validate the conflict type and resolver.
+ *
+ * Return ConflictType enum value corresponding to given conflict_type.
+ */
+static ConflictType
+validate_conflict_type_and_resolver(char *conflict_type, char *conflict_resolver,
+									bool isReset)
+{
+	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));
+
+	/* If it is RESET statement, we are done */
+	if (isReset)
+		return 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;
+		}
+	}
+
+	/*
+	 * XXX: Shall we give supported resolvers list for given type in error
+	 * hint?
+	 */
+	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;
+		}
+	}
+
+	/*
+	 * XXX: Shall we give supported resolvers list for given type in error
+	 * hint?
+	 */
+	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;
+}
+
+/*
+ * Execute Conflict Resolver Stmt
+ *
+ * Validate conflict_type and conflict_resolver given by the user
+ * and save the info in pg_conflict system catalog.
+ */
+void
+ExecConflictResolverStmt(ConflictResolverStmt *stmt)
+{
+	Relation	pg_conflict;
+	Datum		values[Natts_pg_conflict];
+	bool		nulls[Natts_pg_conflict];
+	bool		replaces[Natts_pg_conflict];
+	ScanKeyData keys[1];
+	SysScanDesc scan;
+	HeapTuple	oldtup;
+	HeapTuple	newtup = NULL;
+	ConflictType type;
+
+	type = validate_conflict_type_and_resolver(stmt->conflict_type,
+											   stmt->conflict_resolver,
+											   stmt->isReset);
+
+
+	/* Prepare to update a tuple in pg_conflict system catalog */
+	memset(values, 0, sizeof(values));
+	memset(nulls, false, sizeof(nulls));
+	memset(replaces, false, sizeof(replaces));
+
+	values[Anum_pg_conflict_conftype - 1] =
+			CStringGetTextDatum(stmt->conflict_type);
+
+	if (stmt->isReset)
+	{
+		/* Set resolver to default */
+		ConflictResolver resolver = ConflictTypeDefaultResolvers[type];
+
+		values[Anum_pg_conflict_confres - 1] =
+			CStringGetTextDatum(ConflictResolverNames[resolver]);
+	}
+	else
+		values[Anum_pg_conflict_confres - 1] =
+			CStringGetTextDatum(stmt->conflict_resolver);
+
+	/* Use the index to search for a matching old tuple */
+	ScanKeyInit(&keys[0],
+				Anum_pg_conflict_conftype,
+				BTEqualStrategyNumber, F_TEXTEQ,
+				values[Anum_pg_conflict_conftype - 1]);
+
+	pg_conflict = table_open(ConflictResRelationId, RowExclusiveLock);
+
+	scan = systable_beginscan(pg_conflict, ConflictTypeIndexId, true,
+							  NULL, 1, keys);
+
+	oldtup = systable_getnext(scan);
+	if (HeapTupleIsValid(oldtup))
+	{
+		replaces[Anum_pg_conflict_confres - 1] = true;
+		newtup = heap_modify_tuple(oldtup, RelationGetDescr(pg_conflict),
+								   values, nulls, replaces);
+		CatalogTupleUpdate(pg_conflict, &oldtup->t_self, newtup);
+	}
+	else
+		Assert(oldtup);			/* We must have a tuple for given
+								 * conflict_type */
+
+	systable_endscan(scan);
+
+	if (newtup != NULL)
+		heap_freetuple(newtup);
+
+	table_close(pg_conflict, RowExclusiveLock);
+}
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index fa66b8017e..4ddf4aaff0 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -59,6 +59,7 @@
 #include "miscadmin.h"
 #include "parser/parse_utilcmd.h"
 #include "postmaster/bgwriter.h"
+#include "replication/conflict.h"
 #include "rewrite/rewriteDefine.h"
 #include "storage/fd.h"
 #include "tcop/utility.h"
@@ -163,6 +164,7 @@ ClassifyUtilityCommandAsReadOnly(Node *parsetree)
 		case T_AlterUserMappingStmt:
 		case T_CommentStmt:
 		case T_CompositeTypeStmt:
+		case T_ConflictResolverStmt:
 		case T_CreateAmStmt:
 		case T_CreateCastStmt:
 		case T_CreateConversionStmt:
@@ -1062,6 +1064,11 @@ standard_ProcessUtility(PlannedStmt *pstmt,
 				break;
 			}
 
+		case T_ConflictResolverStmt:
+			/* no event triggers for conflict types and resolvers */
+			ExecConflictResolverStmt((ConflictResolverStmt *) parsetree);
+			break;
+
 		default:
 			/* All other statement types have event trigger support */
 			ProcessUtilitySlow(pstate, pstmt, queryString,
@@ -2659,6 +2666,10 @@ CreateCommandTag(Node *parsetree)
 			tag = CMDTAG_COMMENT;
 			break;
 
+		case T_ConflictResolverStmt:
+			tag = CMDTAG_CONFLICT_RESOLVER;
+			break;
+
 		case T_SecLabelStmt:
 			tag = CMDTAG_SECURITY_LABEL;
 			break;
@@ -3334,6 +3345,10 @@ GetCommandLogLevel(Node *parsetree)
 			lev = LOGSTMT_DDL;
 			break;
 
+		case T_ConflictResolverStmt:
+			lev = LOGSTMT_DDL;
+			break;
+
 		case T_SecLabelStmt:
 			lev = LOGSTMT_DDL;
 			break;
diff --git a/src/include/catalog/Makefile b/src/include/catalog/Makefile
index 167f91a6e3..8619d73e5a 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_conflict.h
 
 GENERATED_HEADERS := $(CATALOG_HEADERS:%.h=%_d.h)
 
@@ -97,6 +98,7 @@ POSTGRES_BKI_DATA = \
 	pg_cast.dat \
 	pg_class.dat \
 	pg_collation.dat \
+	pg_conflict.dat \
 	pg_conversion.dat \
 	pg_database.dat \
 	pg_language.dat \
diff --git a/src/include/catalog/meson.build b/src/include/catalog/meson.build
index f70d1daba5..4d6732a303 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_conflict.h',
 ]
 
 # The .dat files we need can just be listed alphabetically.
@@ -81,6 +82,7 @@ bki_data = [
   'pg_cast.dat',
   'pg_class.dat',
   'pg_collation.dat',
+  'pg_conflict.dat',
   'pg_conversion.dat',
   'pg_database.dat',
   'pg_language.dat',
diff --git a/src/include/catalog/pg_conflict.dat b/src/include/catalog/pg_conflict.dat
new file mode 100644
index 0000000000..a3e5e10d97
--- /dev/null
+++ b/src/include/catalog/pg_conflict.dat
@@ -0,0 +1,20 @@
+#----------------------------------------------------------------------
+#
+# pg_conflict.dat
+#    Initial contents of the pg_conflict system catalog.
+#
+# Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+# Portions Copyright (c) 1994, Regents of the University of California
+#
+# src/include/catalog/pg_conflict.dat
+#
+#----------------------------------------------------------------------
+
+[
+
+{ conftype => 'insert_exists', confres => 'remote_apply' },
+{ conftype => 'update_differ', confres => 'remote_apply' },
+{ conftype => 'update_missing', confres => 'apply_or_skip' },
+{ conftype => 'delete_missing', confres => 'skip' }
+
+]
diff --git a/src/include/catalog/pg_conflict.h b/src/include/catalog/pg_conflict.h
new file mode 100644
index 0000000000..e3fe3e6d30
--- /dev/null
+++ b/src/include/catalog/pg_conflict.h
@@ -0,0 +1,43 @@
+/* -------------------------------------------------------------------------
+ *
+ * pg_conflict.h
+ *	  definition of the "global conflict detection" system catalog
+ *	  (pg_conflict)
+ *
+ * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/catalog/pg_conflict.h
+ *
+ * NOTES
+ *	  The Catalog.pm module reads this file and derives schema
+ *	  information.
+ *
+ * -------------------------------------------------------------------------
+ */
+#ifndef PG_CONFLICT_H
+#define PG_CONFLICT_H
+
+#include "catalog/genbki.h"
+#include "catalog/pg_conflict_d.h"
+
+/* ----------------
+ *		pg_conflict definition.  cpp turns this into
+ *		typedef struct FormData_pg_conflict
+ * ----------------
+ */
+CATALOG(pg_conflict,8688,ConflictResRelationId)
+{
+#ifdef CATALOG_VARLEN			/* variable-length fields start here */
+	text		conftype BKI_FORCE_NOT_NULL;	/* conflict type */
+	text		confres BKI_FORCE_NOT_NULL; /* conflict resolver */
+#endif
+} FormData_pg_conflict;
+
+DECLARE_TOAST(pg_conflict, 8689, 8690);
+
+DECLARE_UNIQUE_INDEX_PKEY(pg_conflict_type_index, 8691, ConflictTypeIndexId, pg_conflict, btree(conftype text_ops));
+
+MAKE_SYSCACHE(CONFLICTTYPE, pg_conflict_type_index, 16);
+
+#endif							/* PG_CONFLICT_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 85a62b538e..1d41169d5e 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3262,6 +3262,18 @@ typedef struct CommentStmt
 	char	   *comment;		/* Comment to insert, or NULL to remove */
 } CommentStmt;
 
+/* ----------------------
+ *				CONFLICT RESOLVER Statement
+ * ----------------------
+ */
+typedef struct ConflictResolverStmt
+{
+	NodeTag		type;
+	char	   *conflict_type;
+	char	   *conflict_resolver;
+	bool		isReset;
+} ConflictResolverStmt;
+
 /* ----------------------
  *				SECURITY LABEL Statement
  * ----------------------
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index f7fe834cf4..21e5cb162d 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -377,6 +377,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 0bc9db991e..7eff93eff3 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -10,6 +10,7 @@
 #define CONFLICT_H
 
 #include "access/xlogdefs.h"
+#include "catalog/objectaddress.h"
 #include "executor/tuptable.h"
 #include "nodes/execnodes.h"
 #include "utils/relcache.h"
@@ -33,6 +34,41 @@ typedef enum
 	CT_DELETE_MISSING,
 } 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_REMOTE_APPLY = 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_REMOTE_APPLY
+#define CR_MAX CR_ERROR
+
 extern bool GetTupleCommitTs(TupleTableSlot *localslot, TransactionId *xmin,
 							 RepOriginId *localorigin, TimestampTz *localts);
 extern void ReportApplyConflict(int elevel, ConflictType type,
@@ -40,5 +76,5 @@ extern void ReportApplyConflict(int elevel, ConflictType type,
 								TransactionId localxmin, RepOriginId localorigin,
 								TimestampTz localts, TupleTableSlot *conflictslot);
 extern void InitConflictIndexes(ResultRelInfo *relInfo);
-
+extern void ExecConflictResolverStmt(ConflictResolverStmt *stmt);
 #endif
diff --git a/src/include/tcop/cmdtaglist.h b/src/include/tcop/cmdtaglist.h
index 7fdcec6dd9..562478e232 100644
--- a/src/include/tcop/cmdtaglist.h
+++ b/src/include/tcop/cmdtaglist.h
@@ -80,6 +80,7 @@ PG_CMDTAG(CMDTAG_CLUSTER, "CLUSTER", false, false, false)
 PG_CMDTAG(CMDTAG_COMMENT, "COMMENT", true, false, false)
 PG_CMDTAG(CMDTAG_COMMIT, "COMMIT", false, false, false)
 PG_CMDTAG(CMDTAG_COMMIT_PREPARED, "COMMIT PREPARED", false, false, false)
+PG_CMDTAG(CMDTAG_CONFLICT_RESOLVER, "CONFLICT RESOLVER", false, false, false)
 PG_CMDTAG(CMDTAG_COPY, "COPY", false, false, true)
 PG_CMDTAG(CMDTAG_COPY_FROM, "COPY FROM", false, false, false)
 PG_CMDTAG(CMDTAG_CREATE_ACCESS_METHOD, "CREATE ACCESS METHOD", true, false, false)
diff --git a/src/test/regress/expected/conflict_resolver.out b/src/test/regress/expected/conflict_resolver.out
new file mode 100644
index 0000000000..15c8c0cdba
--- /dev/null
+++ b/src/test/regress/expected/conflict_resolver.out
@@ -0,0 +1,50 @@
+--check default global resolvers in system catalog
+select * from pg_conflict order by conftype;
+    conftype    |    confres    
+----------------+---------------
+ delete_missing | skip
+ insert_exists  | remote_apply
+ update_differ  | remote_apply
+ update_missing | apply_or_skip
+(4 rows)
+
+--
+-- Test of SET/RESET CONFLICT RESOLVER with invalid names
+--
+SET CONFLICT RESOLVER 'keep_local' for 'aaaa'; -- fail 
+ERROR:  aaaa is not a valid conflict type
+SET CONFLICT RESOLVER 'bbbbb' for 'delete_missing'; -- fail 
+ERROR:  bbbbb is not a valid conflict resolver
+SET CONFLICT RESOLVER 'remote_apply' for 'delete_missing'; -- fail 
+ERROR:  remote_apply is not a valid conflict resolver for conflict type delete_missing
+RESET CONFLICT RESOLVER for 'ct'; -- fail 
+ERROR:  ct is not a valid conflict type
+--
+-- Test of SET/RESET CONFLICT RESOLVER with valid names
+--
+SET CONFLICT RESOLVER 'error' for 'delete_missing';
+SET CONFLICT RESOLVER 'keep_local' for 'insert_exists';
+SET CONFLICT RESOLVER 'keep_local' for 'update_differ';
+SET CONFLICT RESOLVER 'apply_or_error' for 'update_missing';
+--check new resolvers are saved
+select * from pg_conflict order by conftype;
+    conftype    |    confres     
+----------------+----------------
+ delete_missing | error
+ insert_exists  | keep_local
+ update_differ  | keep_local
+ update_missing | apply_or_error
+(4 rows)
+
+RESET CONFLICT RESOLVER for 'delete_missing';
+RESET CONFLICT RESOLVER for 'insert_exists';
+--check resolvers are reset to default for delete_missing and insert_exists
+select * from pg_conflict order by conftype;
+    conftype    |    confres     
+----------------+----------------
+ delete_missing | skip
+ insert_exists  | remote_apply
+ update_differ  | keep_local
+ update_missing | apply_or_error
+(4 rows)
+
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 2429ec2bba..76ea2d359b 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -48,7 +48,7 @@ test: create_index create_index_spgist create_view index_including index_includi
 # ----------
 # Another group of parallel tests
 # ----------
-test: create_aggregate create_function_sql create_cast constraints triggers select inherit typed_table vacuum drop_if_exists updatable_views roleattributes create_am hash_func errors infinite_recurse
+test: create_aggregate create_function_sql create_cast conflict_resolver constraints triggers select inherit typed_table vacuum drop_if_exists updatable_views roleattributes create_am hash_func errors infinite_recurse
 
 # ----------
 # sanity_check does a vacuum, affecting the sort order of SELECT *
diff --git a/src/test/regress/sql/conflict_resolver.sql b/src/test/regress/sql/conflict_resolver.sql
new file mode 100644
index 0000000000..e3540d7ac8
--- /dev/null
+++ b/src/test/regress/sql/conflict_resolver.sql
@@ -0,0 +1,27 @@
+--check default global resolvers in system catalog
+select * from pg_conflict order by conftype;
+
+--
+-- Test of SET/RESET CONFLICT RESOLVER with invalid names
+--
+SET CONFLICT RESOLVER 'keep_local' for 'aaaa'; -- fail
+SET CONFLICT RESOLVER 'bbbbb' for 'delete_missing'; -- fail
+SET CONFLICT RESOLVER 'remote_apply' for 'delete_missing'; -- fail
+RESET CONFLICT RESOLVER for 'ct'; -- fail
+
+--
+-- Test of SET/RESET CONFLICT RESOLVER with valid names
+--
+SET CONFLICT RESOLVER 'error' for 'delete_missing';
+SET CONFLICT RESOLVER 'keep_local' for 'insert_exists';
+SET CONFLICT RESOLVER 'keep_local' for 'update_differ';
+SET CONFLICT RESOLVER 'apply_or_error' for 'update_missing';
+
+--check new resolvers are saved
+select * from pg_conflict order by conftype;
+
+RESET CONFLICT RESOLVER for 'delete_missing';
+RESET CONFLICT RESOLVER for 'insert_exists';
+
+--check resolvers are reset to default for delete_missing and insert_exists
+select * from pg_conflict order by conftype;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 2485a80833..784194903a 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -466,6 +466,8 @@ ConditionVariableMinimallyPadded
 ConditionalStack
 ConfigData
 ConfigVariable
+ConflictResolver
+ConflictResolverStmt
 ConflictType
 ConnCacheEntry
 ConnCacheKey
-- 
2.34.1