From 26fd28171d89583d2103b57e883bd7d49f169c4a Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Wed, 23 Sep 2020 16:16:36 +0900
Subject: [PATCH v26 08/11] Automatic foreign transaciton resolution on
 commit/rollback.

Co-authored-by: Masahiko Sawada, Ashutosh Bapat
---
 src/backend/access/fdwxact/fdwxact.c          | 244 +++++++++++++++++-
 src/backend/access/transam/xact.c             |  45 +++-
 src/backend/utils/misc/guc.c                  |  28 ++
 src/backend/utils/misc/postgresql.conf.sample |   2 +
 src/include/access/fdwxact.h                  |  11 +
 src/include/foreign/fdwapi.h                  |   2 +-
 6 files changed, 312 insertions(+), 20 deletions(-)

diff --git a/src/backend/access/fdwxact/fdwxact.c b/src/backend/access/fdwxact/fdwxact.c
index e3b5937054..8638a3cdf0 100644
--- a/src/backend/access/fdwxact/fdwxact.c
+++ b/src/backend/access/fdwxact/fdwxact.c
@@ -19,6 +19,32 @@
  * transaction, we collect the involved foreign transaction and wait for the resolver
  * process committing or rolling back the foreign transactions.
  *
+ * The global transaction manager support automatically foreign transaction
+ * resolution on commit and rollback.  The basic strategy is that we prepare all
+ * of the remote transactions before committing locally and commit them after
+ * committing locally.
+ *
+ * During pre-commit of local transaction, we prepare the transaction on
+ * all foreign servers.  And after committing or rolling back locally,
+ * we notify the resolver process and tell it to commit or rollback those
+ * transactions. If we ask to commit, we also tell to notify us when
+ * it's done, so that we can wait interruptibly to finish, and so that
+ * we're not trying to locally do work that might fail after foreign
+ * transaction are committed.
+ *
+ * The best performing way to manage the waiting backends is to have a
+ * queue of waiting backends, so that we can avoid searching the through all
+ * foreign transactions each time we receive a request.  We have one queue
+ * of which elements are ordered by the timestamp when they expect to be
+ * processed.  Before waiting for foreign transactions being resolved the
+ * backend enqueues with the timestamp when they expects to be processed.
+ * On failure, it enqueues again with new timestamp (last timestamp +
+ * foreign_xact_resolution_interval).
+ *
+ * If server crash occurs or user canceled waiting the prepared foreign
+ * transactions are left without a holder.	Such foreign transactions are
+ * resolved automatically by the resolver process.
+ *
  * Two-phase commit protocol is crash-safe.  We WAL logs the foreign transaction
  * information.
  *
@@ -96,6 +122,9 @@
 #include "utils/rel.h"
 #include "utils/ps_status.h"
 
+/* Foreign twophase commit is enabled and requested by user */
+#define IsForeignTwophaseCommitRequested() \
+	 (foreign_twophase_commit > FOREIGN_TWOPHASE_COMMIT_DISABLED)
 /* Check the FdwXactParticipant is capable of two-phase commit  */
 #define ServerSupportTransactionCallback(fdw_part) \
 	(((FdwXactParticipant *)(fdw_part))->commit_foreign_xact_fn != NULL)
@@ -139,6 +168,9 @@ typedef struct FdwXactParticipant
 	/* Transaction identifier used for PREPARE */
 	char	   *fdwxact_id;
 
+	/* true if modified the data on the server */
+	bool		modified;
+
 	/* Callbacks for foreign transaction */
 	CommitForeignTransaction_function commit_foreign_xact_fn;
 	RollbackForeignTransaction_function rollback_foreign_xact_fn;
@@ -161,15 +193,25 @@ typedef struct FdwXactParticipant
  * and kept them in FdwXactParticipants_tmp.  Even if an error occurs during
  * that, we don't rollback them.  In the second phase, SetFdwXactParticipants(),
  * we replace FdwXactParticipants_tmp with FdwXactParticipants and hold them.
+ *
  */
 static List *FdwXactParticipants = NIL;
 static List *FdwXactParticipants_tmp = NIL;
 
+/*
+ * FdwXactLocalXid is the local transaction id associated with FdwXactParticipants.
+ * ForeignTwophaseCommitIsRequired is true if the current transaction needs to
+ * be committed together with foreign servers.
+ */
+static TransactionId FdwXactLocalXid = InvalidTransactionId;
+static bool ForeignTwophaseCommitIsRequired = false;
+
 /* Guc parameter */
 int			max_prepared_foreign_xacts = 0;
 int			max_foreign_xact_resolvers = 0;
+int			foreign_twophase_commit = FOREIGN_TWOPHASE_COMMIT_DISABLED;
 
-static void FdwXactPrepareForeignTransactions(void);
+static void FdwXactPrepareForeignTransactions(bool prepare_all);
 static void FdwXactParticipantEndTransaction(FdwXactParticipant *fdw_part,
 											 bool commit);
 static FdwXact FdwXactInsertFdwXactEntry(TransactionId xid,
@@ -189,6 +231,7 @@ static char *ProcessFdwXactBuffer(Oid dbid, TransactionId xid, Oid serverid,
 static char *ReadFdwXactFile(Oid dbid, TransactionId xid, Oid serverid, Oid userid);
 static void RemoveFdwXactFile(Oid dbid, TransactionId xid, Oid serverid, Oid userid,
 							  bool giveWarning);
+static bool checkForeignTwophaseCommitRequired(bool local_modified);
 static FdwXact insert_fdwxact(Oid dbid, TransactionId xid, Oid serverid, Oid userid,
 							  Oid umid, char *fdwxact_id);
 static void remove_fdwxact(FdwXact fdwxact);
@@ -269,7 +312,7 @@ FdwXactShmemInit(void)
  * as a participant of the transaction.
  */
 void
-FdwXactRegisterXact(Oid serverid, Oid userid)
+FdwXactRegisterXact(Oid serverid, Oid userid, bool modified)
 {
 	FdwXactParticipant *fdw_part;
 	MemoryContext old_ctx;
@@ -284,6 +327,7 @@ FdwXactRegisterXact(Oid serverid, Oid userid)
 			fdw_part->usermapping->userid == userid)
 		{
 			/* Already registered */
+			fdw_part->modified |= modified;
 			return;
 		}
 	}
@@ -303,6 +347,7 @@ FdwXactRegisterXact(Oid serverid, Oid userid)
 	old_ctx = MemoryContextSwitchTo(TopTransactionContext);
 
 	fdw_part = create_fdwxact_participant(serverid, userid, routine);
+	fdw_part->modified = modified;
 
 	/* Add to the participants list */
 	FdwXactParticipants = lappend(FdwXactParticipants, fdw_part);
@@ -349,6 +394,7 @@ create_fdwxact_participant(Oid serverid, Oid userid, FdwRoutine *routine)
 	fdw_part->server = foreign_server;
 	fdw_part->usermapping = user_mapping;
 	fdw_part->fdwxact_id = NULL;
+	fdw_part->modified = false;
 	fdw_part->commit_foreign_xact_fn = routine->CommitForeignTransaction;
 	fdw_part->rollback_foreign_xact_fn = routine->RollbackForeignTransaction;
 	fdw_part->prepare_foreign_xact_fn = routine->PrepareForeignTransaction;
@@ -358,23 +404,169 @@ create_fdwxact_participant(Oid serverid, Oid userid, FdwRoutine *routine)
 }
 
 /*
- * Insert FdwXact entries and prepare foreign transactions.
+ * Prepare all foreign transactions if foreign twophase commit is required.
+ * When foreign twophase commit is enabled, the behavior depends on the value
+ * of foreign_twophase_commit; when 'required' we strictly require for all
+ * foreign servers' FDW to support two-phase commit protocol and ask them to
+ * prepare foreign transactions, and when 'disabled' we ask all foreign servers
+ * to commit foreign transaction in one-phase. If we failed to commit any of
+ * them we change to aborting.
+ *
+ * Note that non-modified foreign servers always can be committed without
+ * preparation.
+ */
+void
+PreCommit_FdwXact(void)
+{
+	TransactionId xid;
+	ListCell   *lc;
+	bool		local_modified;
+
+	/* If there are no foreign servers involved, we have no business here */
+	if (FdwXactParticipants == NIL)
+		return;
+
+	Assert(!RecoveryInProgress());
+
+	/*
+	 * Check if the current transaction did writes.	 We need to include the
+	 * local node to the distributed transaction participant and to regard it
+	 * as modified, if the current transaction has performed WAL logging and
+	 * has assigned an xid.	 The transaction can end up not writing any WAL,
+	 * even if it has an xid, if it only wrote to temporary and/or unlogged
+	 * tables.	It can end up having written WAL without an xid if did HOT
+	 * pruning.
+	 */
+	xid = GetTopTransactionIdIfAny();
+	local_modified = (TransactionIdIsValid(xid) && (XactLastRecEnd != 0));
+
+	/*
+	 * Check if we need to use foreign twophase commit. Note that we don't
+	 * support foreign twophase commit in single user mode.
+	 */
+	if (IsUnderPostmaster && checkForeignTwophaseCommitRequired(local_modified))
+	{
+		/*
+		 * We need to use two-phase commit.	 Assign a transaction id to the
+		 * current transaction if not yet. Then prepare foreign transactions
+		 * on foreign servers that support two-phase commit.  Note that we
+		 * keep FdwXactParticipants until the end of the transaction.
+		 */
+		FdwXactLocalXid = xid;
+		if (!TransactionIdIsValid(FdwXactLocalXid))
+			FdwXactLocalXid = GetTopTransactionId();
+
+		FdwXactPrepareForeignTransactions(false);
+		ForeignTwophaseCommitIsRequired = true;
+	}
+	else
+	{
+		/*
+		 * Two-phase commit is not required. Commit foreign transactions in
+		 * the participant list.
+		 */
+		foreach(lc, FdwXactParticipants)
+		{
+			FdwXactParticipant *fdw_part = (FdwXactParticipant *) lfirst(lc);
+
+			Assert(!fdw_part->fdwxact);
+
+			/* Commit the foreign transaction in one-phase */
+			if (ServerSupportTransactionCallback(fdw_part))
+				FdwXactParticipantEndTransaction(fdw_part, true);
+		}
+
+		/* All participants' transactions should be completed at this time */
+		ForgetAllFdwXactParticipants();
+	}
+}
+
+/*
+ * Return true if the current transaction modifies data on two or more servers
+ * in FdwXactParticipants and local server itself.
+ */
+static bool
+checkForeignTwophaseCommitRequired(bool local_modified)
+{
+	ListCell   *lc;
+	bool		have_notwophase = false;
+	int			nserverswritten = 0;
+
+	if (!IsForeignTwophaseCommitRequested())
+		return false;
+
+	foreach(lc, FdwXactParticipants)
+	{
+		FdwXactParticipant *fdw_part = (FdwXactParticipant *) lfirst(lc);
+
+		if (!fdw_part->modified)
+			continue;
+
+		if (!ServerSupportTwophaseCommit(fdw_part))
+			have_notwophase = true;
+
+		nserverswritten++;
+	}
+
+	/* Did we modify the local non-temporary data? */
+	if (local_modified)
+		nserverswritten++;
+
+	/*
+	 * Two-phase commit is not required if the number of servers performed
+	 * writes is less than 2.
+	 */
+	if (nserverswritten < 2)
+		return false;
+
+	Assert(foreign_twophase_commit == FOREIGN_TWOPHASE_COMMIT_REQUIRED);
+
+	/* Two-phase commit is required. Check parameters */
+	if (max_prepared_foreign_xacts == 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("foreign two-phase commit is required but prepared foreign transactions are disabled"),
+				 errhint("Set max_prepared_foreign_transactions to a nonzero value.")));
+
+	if (max_foreign_xact_resolvers == 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("foreign two-phase commit is required but prepared foreign transactions are disabled"),
+				 errhint("Set max_foreign_transaction_resolvers to a nonzero value.")));
+
+	if (have_notwophase)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("cannot process a distributed transaction that has operated on a foreign server that does not support two-phase commit protocol"),
+				 errdetail("foreign_twophase_commit is \'required\' but the transaction has some foreign servers which are not capable of two-phase commit")));
+
+	return true;
+}
+
+/*
+ * Insert FdwXact entries and prepare foreign transactions.  If prepare_all is
+ * true, we prepare all foreign transaction regardless of writes having happened
+ * on the server.
+ *
+ * We still can change to rollback here on failure. If any error occurs, we
+ * rollback non-prepared foreign transactions.
  */
 static void
-FdwXactPrepareForeignTransactions(void)
+FdwXactPrepareForeignTransactions(bool prepare_all)
 {
 	ListCell   *lc;
 
 	if (FdwXactParticipants == NIL)
 		return;
 
+	Assert(TransactionIdIsValid(FdwXactLocalXid));
+
 	/* Loop over the foreign connections */
 	foreach(lc, FdwXactParticipants)
 	{
 		FdwXactParticipant *fdw_part = (FdwXactParticipant *) lfirst(lc);
 		FdwXactRslvState state;
 		FdwXact		fdwxact;
-		TransactionId xid = GetTopTransactionId();
 
 		CHECK_FOR_INTERRUPTS();
 
@@ -382,8 +574,11 @@ FdwXactPrepareForeignTransactions(void)
 		if (!ServerSupportTwophaseCommit(fdw_part))
 			continue;
 
+		if (!prepare_all && !fdw_part->modified)
+			continue;
+
 		/* Get prepared transaction identifier */
-		fdw_part->fdwxact_id = get_fdwxact_identifier(fdw_part, xid);
+		fdw_part->fdwxact_id = get_fdwxact_identifier(fdw_part, FdwXactLocalXid);
 		Assert(fdw_part->fdwxact_id);
 
 		/*
@@ -401,7 +596,7 @@ FdwXactPrepareForeignTransactions(void)
 		 * server and will not be able to resolve it after the crash recovery.
 		 * Hence persist first then prepare.
 		 */
-		fdwxact = FdwXactInsertFdwXactEntry(xid, fdw_part);
+		fdwxact = FdwXactInsertFdwXactEntry(FdwXactLocalXid, fdw_part);
 
 		/*
 		 * Prepare the foreign transaction.
@@ -410,7 +605,7 @@ FdwXactPrepareForeignTransactions(void)
 		 * acknowledge from foreign server, the backend may abort the local
 		 * transaction (say, because of a signal).
 		 */
-		state.xid = xid;
+		state.xid = FdwXactLocalXid;
 		state.server = fdw_part->server;
 		state.usermapping = fdw_part->usermapping;
 		state.fdwxact_id = fdw_part->fdwxact_id;
@@ -739,8 +934,11 @@ PrePrepare_FdwXact(void)
 					 errmsg("cannot PREPARE a distributed transaction which has operated on a foreign server not supporting two-phase commit protocol")));
 	}
 
+	/* Set the local transaction id */
+	FdwXactLocalXid = GetTopTransactionId();
+
 	/* Prepare transactions on participating foreign servers */
-	FdwXactPrepareForeignTransactions();
+	FdwXactPrepareForeignTransactions(true);
 
 	/*
 	 * We keep prepared foreign transaction participants to rollback them in
@@ -831,6 +1029,12 @@ SetFdwXactParticipants(TransactionId xid)
 	LWLockRelease(FdwXactLock);
 }
 
+bool
+FdwXactIsForeignTwophaseCommitRequired(void)
+{
+	return ForeignTwophaseCommitIsRequired;
+}
+
 void
 FdwXactCleanupAtProcExit(void)
 {
@@ -1165,6 +1369,7 @@ FdwXactParticipantEndTransaction(FdwXactParticipant *fdw_part, bool commit)
 
 	Assert(ServerSupportTransactionCallback(fdw_part));
 
+	state.xid = FdwXactLocalXid;
 	state.server = fdw_part->server;
 	state.usermapping = fdw_part->usermapping;
 	state.fdwxact_id = NULL;
@@ -1201,6 +1406,7 @@ ForgetAllFdwXactParticipants(void)
 	if (FdwXactParticipants == NIL)
 	{
 		Assert(FdwXactParticipants_tmp == NIL);
+		Assert(!ForeignTwophaseCommitIsRequired);
 		return;
 	}
 
@@ -1248,6 +1454,7 @@ ForgetAllFdwXactParticipants(void)
 	list_free_deep(FdwXactParticipants_tmp);
 	FdwXactParticipants = NIL;
 	FdwXactParticipants_tmp = NIL;
+	ForeignTwophaseCommitIsRequired = false;
 }
 
 /*
@@ -1257,6 +1464,7 @@ void
 AtEOXact_FdwXact(bool is_commit)
 {
 	ListCell   *lc;
+	bool		rollback_prepared = false;
 
 	/* If there are no foreign servers involved, we have no business here */
 	if (FdwXactParticipants == NIL)
@@ -1287,7 +1495,11 @@ AtEOXact_FdwXact(bool is_commit)
 
 		/*
 		 * Abort the foreign transaction.  For participants whose status is
-		 * FDWXACT_STATUS_PREPARING, we close the transaction in one-phase.
+		 * FDWXACT_STATUS_PREPARING, we close the transaction in one-phase. In
+		 * addition, since we are not sure that the preparation has been
+		 * completed on the foreign server, we also attempts to rollback the
+		 * prepared foreign transaction.  Note that it's FDWs responsibility
+		 * that they tolerate OBJECT_NOT_FOUND error in abort case.
 		 */
 		SpinLockAcquire(&(fdwxact->mutex));
 		status = fdwxact->status;
@@ -1296,6 +1508,18 @@ AtEOXact_FdwXact(bool is_commit)
 
 		if (status == FDWXACT_STATUS_PREPARING)
 			FdwXactParticipantEndTransaction(fdw_part, false);
+
+		rollback_prepared = true;
+	}
+
+	/*
+	 * Wait for all prepared or possibly-prepared foreign transactions to be
+	 * rolled back.
+	 */
+	if (rollback_prepared)
+	{
+		Assert(TransactionIdIsValid(FdwXactLocalXid));
+		FdwXactWaitForResolution(FdwXactLocalXid, false);
 	}
 
 	ForgetAllFdwXactParticipants();
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 0dcc3182ec..cd11d58721 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -1237,6 +1237,7 @@ RecordTransactionCommit(void)
 	SharedInvalidationMessage *invalMessages = NULL;
 	bool		RelcacheInitFileInval = false;
 	bool		wrote_xlog;
+	bool		need_fdwxact_commit;
 
 	/*
 	 * Log pending invalidations for logical decoding of in-progress
@@ -1255,6 +1256,7 @@ RecordTransactionCommit(void)
 		nmsgs = xactGetCommittedInvalidationMessages(&invalMessages,
 													 &RelcacheInitFileInval);
 	wrote_xlog = (XactLastRecEnd != 0);
+	need_fdwxact_commit = FdwXactIsForeignTwophaseCommitRequired();
 
 	/*
 	 * If we haven't been assigned an XID yet, we neither can, nor do we want
@@ -1293,12 +1295,13 @@ RecordTransactionCommit(void)
 		}
 
 		/*
-		 * If we didn't create XLOG entries, we're done here; otherwise we
-		 * should trigger flushing those entries the same as a commit record
+		 * If we didn't create XLOG entries and the transaction does not need
+		 * to be committed using two-phase commit. we're done here; otherwise
+		 * we should trigger flushing those entries the same as a commit record
 		 * would.  This will primarily happen for HOT pruning and the like; we
 		 * want these to be flushed to disk in due time.
 		 */
-		if (!wrote_xlog)
+		if (!wrote_xlog && !need_fdwxact_commit)
 			goto cleanup;
 	}
 	else
@@ -1445,16 +1448,37 @@ RecordTransactionCommit(void)
 	latestXid = TransactionIdLatest(xid, nchildren, children);
 
 	/*
-	 * Wait for synchronous replication, if required. Similar to the decision
-	 * above about using committing asynchronously we only want to wait if
-	 * this backend assigned an xid and wrote WAL.  No need to wait if an xid
-	 * was assigned due to temporary/unlogged tables or due to HOT pruning.
+	 * Wait for both synchronous replication and prepared foreign transaction
+	 * to be committed, if required.  We must wait for synchrnous replication
+	 * first because we need to make sure that the fate of the current
+	 * transaction is consistent between the primary and sync replicas before
+	 * resolving foreign transaction.  Otherwise, we will end up violating
+	 * atomic commit if a fail-over happens after some of foreign transactions
+	 * are committed.
 	 *
 	 * Note that at this stage we have marked clog, but still show as running
 	 * in the procarray and continue to hold locks.
 	 */
-	if (wrote_xlog && markXidCommitted)
-		SyncRepWaitForLSN(XactLastRecEnd, true);
+	if (markXidCommitted)
+	{
+		bool canceled = false;
+
+		/*
+		 * Similar to the decision above about using committing asynchronously
+		 * we only want to wait if this backend assigned an xid, wrote WAL,
+		 * and not received a query cancel.  No need to wait if an xid was
+		 * assigned due to temporary/unlogged tables or due to HOT pruning.
+		 */
+		if (wrote_xlog)
+			canceled = SyncRepWaitForLSN(XactLastRecEnd, true);
+
+		/*
+		 * We only want to wait if we prepared foreign transactions in this
+		 * transaction and not received query cancel.
+		 */
+		if (!canceled && need_fdwxact_commit)
+			FdwXactWaitForResolution(xid, true);
+	}
 
 	/* remember end of last commit record */
 	XactLastCommitEnd = XactLastRecEnd;
@@ -2115,6 +2139,9 @@ CommitTransaction(void)
 			break;
 	}
 
+	/* Pre-commit step for foreign transactions */
+	PreCommit_FdwXact();
+
 	CallXactCallbacks(is_parallel_worker ? XACT_EVENT_PARALLEL_PRE_COMMIT
 					  : XACT_EVENT_PRE_COMMIT);
 
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index b3960e9a1b..298ab461ed 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -483,6 +483,24 @@ const struct config_enum_entry ssl_protocol_versions_info[] = {
 	{NULL, 0, false}
 };
 
+/*
+ * Although only "required" and "disabled" are documented, we accept all
+ * the likely variants of "on" and "off".
+ */
+static const struct config_enum_entry foreign_twophase_commit_options[] = {
+	{"required", FOREIGN_TWOPHASE_COMMIT_REQUIRED, false},
+	{"disabled", FOREIGN_TWOPHASE_COMMIT_DISABLED, false},
+	{"on", FOREIGN_TWOPHASE_COMMIT_REQUIRED, false},
+	{"off", FOREIGN_TWOPHASE_COMMIT_DISABLED, false},
+	{"true", FOREIGN_TWOPHASE_COMMIT_REQUIRED, true},
+	{"false", FOREIGN_TWOPHASE_COMMIT_DISABLED, true},
+	{"yes", FOREIGN_TWOPHASE_COMMIT_REQUIRED, true},
+	{"no", FOREIGN_TWOPHASE_COMMIT_DISABLED, true},
+	{"1", FOREIGN_TWOPHASE_COMMIT_REQUIRED, true},
+	{"0", FOREIGN_TWOPHASE_COMMIT_DISABLED, true},
+	{NULL, 0, false}
+};
+
 StaticAssertDecl(lengthof(ssl_protocol_versions_info) == (PG_TLS1_3_VERSION + 2),
 				 "array length mismatch");
 
@@ -4657,6 +4675,16 @@ static struct config_enum ConfigureNamesEnum[] =
 		NULL, assign_synchronous_commit, NULL
 	},
 
+	{
+		{"foreign_twophase_commit", PGC_USERSET, FOREIGN_TRANSACTION,
+		 gettext_noop("Use of foreign twophase commit for the current transaction."),
+			NULL
+		},
+		&foreign_twophase_commit,
+		FOREIGN_TWOPHASE_COMMIT_DISABLED, foreign_twophase_commit_options,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"archive_mode", PGC_POSTMASTER, WAL_ARCHIVING,
 			gettext_noop("Allows archiving of WAL files using archive_command."),
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 2ed09cb347..5a73443be1 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -744,6 +744,8 @@
 							# retrying to resolve
 							# foreign transactions
 							# after a failed attempt
+#foreign_twophase_commit = disabled	# use two-phase commit for distributed transactions:
+					# disabled or required
 
 #------------------------------------------------------------------------------
 # VERSION AND PLATFORM COMPATIBILITY
diff --git a/src/include/access/fdwxact.h b/src/include/access/fdwxact.h
index 1ae70cbed6..13962b4156 100644
--- a/src/include/access/fdwxact.h
+++ b/src/include/access/fdwxact.h
@@ -25,6 +25,14 @@
 #define FDWXACT_FLAG_ONEPHASE		0x01	/* transaction can commit/rollback
 											 * without preparation */
 
+/* Enum for foreign_twophase_commit parameter */
+typedef enum
+{
+	FOREIGN_TWOPHASE_COMMIT_DISABLED,	/* disable foreign twophase commit */
+	FOREIGN_TWOPHASE_COMMIT_REQUIRED	/* all foreign servers have to support
+										 * twophase commit */
+}			ForeignTwophaseCommitLevel;
+
 /* Enum to track the status of foreign transaction */
 typedef enum
 {
@@ -122,15 +130,18 @@ extern int	max_foreign_xact_resolvers;
 extern int	foreign_xact_resolution_retry_interval;
 extern int	foreign_xact_resolver_timeout;
 extern int	foreign_twophase_commit;
+extern int	foreign_twophase_commit;
 
 /* Function declarations */
 extern Size FdwXactShmemSize(void);
 extern void FdwXactShmemInit(void);
 extern void AtEOXact_FdwXact(bool is_commit);
+extern void PreCommit_FdwXact(void);
 extern void PrePrepare_FdwXact(void);
 extern void PostPrepare_FdwXact(void);
 extern bool CollectFdwXactParticipants(TransactionId xid);
 extern void SetFdwXactParticipants(TransactionId xid);
+extern bool FdwXactIsForeignTwophaseCommitRequired(void);
 extern void FdwXactCleanupAtProcExit(void);
 extern void FdwXactWaitForResolution(TransactionId wait_xid, bool commit);
 extern PGPROC *FdwXactGetWaiter(TimestampTz now, TimestampTz *nextResolutionTs_p,
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 91db4f5bfc..7a444d0590 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -273,7 +273,7 @@ extern bool IsImportableForeignTable(const char *tablename,
 extern Path *GetExistingLocalJoinPath(RelOptInfo *joinrel);
 
 /* Functions in fdwxact/fdwxact.c */
-extern void FdwXactRegisterXact(Oid serverid, Oid userid);
+extern void FdwXactRegisterXact(Oid serverid, Oid userid, bool modified);
 extern void FdwXactUnregisterXact(Oid serverid, Oid userid);
 
 #endif							/* FDWAPI_H */
-- 
2.23.0

