From 4c22c063930f0dd849114bfa11229f20512786c3 Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Wed, 25 Nov 2020 21:02:29 +0900
Subject: [PATCH v35 07/10] Prepare foreign transactions at commit time

With this commit, the foreign server modified within the transaction
marked as 'modified'. On the 'modified' servers, foreign transactions
are prepared automatically if foreign_twophase_commit is
'required'. Previously, users need to do PREPARE TRANSACTION and
COMMIT/ROLLBACK PREPARED to use two-phase commit protocol. This commit
enables users to use two-phase commit protocol transparently. Prepared
foreign transactions are resolved in asynchronous manner by foreign
transaction resolver process.

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

diff --git a/src/backend/access/transam/fdwxact.c b/src/backend/access/transam/fdwxact.c
index 5d1fcd80ed..e0bf3bb969 100644
--- a/src/backend/access/transam/fdwxact.c
+++ b/src/backend/access/transam/fdwxact.c
@@ -20,6 +20,23 @@
  *
  * FOREIGN TRANSACTION RESOLUTION
  *
+ * The transaction involving multiple foreign transactions uses two-phase commit
+ * protocol to commit the distributed transaction if enabled.  The basic strategy
+ * is that we prepare all of the remote transactions before committing locally and
+ * commit them after committing locally.
+ *
+ * At pre-commit of local transaction, we prepare the transactions on all foreign
+ * servers after logging the information of foreign transaction.  The result of
+ * distributed transaction is determined by the result of the corresponding local
+ * transaction.  Once the local transaction is successfully committed, all
+ * transactions on foreign servers must be committed.  In case where an error occurred
+ * before the local transaction commit all transactions must be aborted.  After
+ * committing or rolling back locally, we leave foreign transactions as in-doubt
+ * transactions and then notify the resolver process. The resolver process asynchronously
+ * resolves these foreign transactions according to the result of the corresponding local
+ * transaction.  Also, the user can use pg_resolve_foreign_xact() SQL function to
+ * resolve a foreign transaction manually.
+ *
  * At PREPARE TRANSACTION, we prepare all transactions on foreign servers by executing
  * PrepareForeignTransaction() API regardless of data on the foreign server having
  * been modified.  At COMMIT PREPARED and ROLLBACK PREPARED, we commit or rollback
@@ -96,8 +113,10 @@
 #include "storage/ipc.h"
 #include "storage/latch.h"
 #include "storage/lock.h"
+#include "storage/pmsignal.h"
 #include "storage/procarray.h"
 #include "storage/sinvaladt.h"
+#include "tcop/tcopprot.h"
 #include "utils/builtins.h"
 #include "utils/guc.h"
 #include "utils/memutils.h"
@@ -116,6 +135,10 @@
 #define ServerSupportTwophaseCommit(fdw_part) \
 	(((FdwXactParticipant *)(fdw_part))->prepare_foreign_xact_fn != NULL)
 
+/* Foreign twophase commit is enabled and requested by user */
+#define IsForeignTwophaseCommitRequested() \
+	 (foreign_twophase_commit > FOREIGN_TWOPHASE_COMMIT_DISABLED)
+
 /*
  * Name of foreign prepared transaction file is 8 bytes xid and
  * user mapping OID separated by '_'.
@@ -156,6 +179,9 @@ typedef struct FdwXactParticipant
 	 */
 	FdwXact		fdwxact;
 
+	/* 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;
@@ -170,18 +196,24 @@ typedef struct FdwXactParticipant
  * PreparedAllParticipants is true if the all foreign transaction participants
  * in FdwXactParticipants are prepared. This can be true in PREPARE TRANSACTION
  * case.
+ *
+ * ForeignTwophaseCommitIsRequired is true if the current transaction needs to
+ * be committed using two-phase commit protocol.
  */
 static HTAB *FdwXactParticipants = NULL;
 static bool PreparedAllParticipants = false;
+static bool ForeignTwophaseCommitIsRequired = false;
 
 /* Keep track of registering process exit call back. */
 static bool fdwXactExitRegistered = 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(TransactionId xid);
+static void FdwXactPrepareForeignTransactions(TransactionId xid, bool prepare_all);
 static char *getFdwXactIdentifier(FdwXactParticipant *fdw_part, TransactionId xid);
 static FdwXact FdwXactInsertEntry(TransactionId xid, FdwXactParticipant *fdw_part,
 								  char *identifier);
@@ -200,6 +232,7 @@ static char *ProcessFdwXactBuffer(TransactionId xid, Oid umid,
 								  XLogRecPtr insert_start_lsn, bool fromdisk);
 static char *ReadFdwXactFile(TransactionId xid, Oid umid);
 static void RemoveFdwXactFile(TransactionId xid, Oid umid, bool giveWarning);
+static bool checkForeignTwophaseCommitRequired(bool local_modified);
 
 static FdwXact insert_fdwxact(Oid dbid, TransactionId xid, Oid umid, Oid serverid,
 							  Oid owner, char *identifier);
@@ -274,7 +307,7 @@ FdwXactShmemInit(void)
  * mapping OID as a participant of the transaction.
  */
 void
-FdwXactRegisterXact(UserMapping *usermapping)
+FdwXactRegisterXact(UserMapping *usermapping, bool modified)
 {
 	FdwXactParticipant	*fdw_part;
 	FdwRoutine	*routine;
@@ -306,6 +339,7 @@ FdwXactRegisterXact(UserMapping *usermapping)
 
 	key = usermapping->umid;
 	fdw_part = hash_search(FdwXactParticipants, (void *) &key, HASH_ENTER, &found);
+	fdw_part->modified |= modified;
 
 	/* Already registered */
 	if (found)
@@ -368,14 +402,23 @@ RemoveFdwParticipant(FdwXactPartKey key)
 }
 
 /*
- * Pre-commit processing for foreign transactions. We commit those foreign
- * transactions with one-phase.
+ * Pre-commit processing for 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' since we use one-phase
+ * commit these foreign transactions are committed at the transaction end.
+ * If we failed to prepare any of them we change to aborting.
  */
 void
 PreCommit_FdwXact(bool is_parallel_worker)
 {
 	HASH_SEQ_STATUS scan;
 	FdwXactParticipant *fdw_part;
+	TransactionId xid;
+	bool		local_modified;
 
 	/*
 	 * If there is no foreign server involved or all foreign transactions
@@ -386,6 +429,40 @@ PreCommit_FdwXact(bool is_parallel_worker)
 
 	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))
+	{
+		/*
+		 * Two-phase commit is required.  Assign a transaction id to the
+		 * current transaction if not yet because the local transaction is
+		 * necessary to determine the result of the distributed transaction.
+		 * Then we prepare foreign transactions on foreign servers that support
+		 * two-phase commit.  Note that we keep FdwXactParticipants until the
+		 * end of the transaction.
+		 */
+		if (!TransactionIdIsValid(xid))
+			xid = GetTopTransactionId();
+		FdwXactPrepareForeignTransactions(xid, false);
+		ForeignTwophaseCommitIsRequired = true;
+
+		return;
+	}
+
 	/* Commit all foreign transactions in the participant list */
 	hash_seq_init(&scan, FdwXactParticipants);
 	while ((fdw_part = (FdwXactParticipant *) hash_seq_search(&scan)))
@@ -458,6 +535,23 @@ AtEOXact_FdwXact(bool is_commit, bool is_parallel_worker)
 		FdwXactLaunchOrWakeupResolver();
 
 	PreparedAllParticipants = false;
+	ForeignTwophaseCommitIsRequired = false;
+}
+
+/*
+ * Return true if there is at least one prepared foreign transaction
+ * which matches given arguments.
+ */
+bool
+FdwXactExists(TransactionId xid, Oid umid)
+{
+	int         idx;
+
+	LWLockAcquire(FdwXactLock, LW_SHARED);
+	idx = get_fdwxact_idx(xid, umid);
+	LWLockRelease(FdwXactLock);
+
+	return (idx >= 0);
 }
 
 /*
@@ -497,7 +591,7 @@ AtPrepare_FdwXact(void)
 	 * prepare all foreign transactions.
 	 */
 	xid = GetTopTransactionId();
-	FdwXactPrepareForeignTransactions(xid);
+	FdwXactPrepareForeignTransactions(xid, true);
 
 	/*
 	 * Remember we already prepared all participants.  We keep FdwXactParticipants
@@ -507,22 +601,6 @@ AtPrepare_FdwXact(void)
 	PreparedAllParticipants = true;
 }
 
-/*
- * Return true if there is at least one prepared foreign transaction
- * which matches given arguments.
- */
-bool
-FdwXactExists(TransactionId xid, Oid umid)
-{
-	int			idx;
-
-	LWLockAcquire(FdwXactLock, LW_SHARED);
-	idx = get_fdwxact_idx(xid, umid);
-	LWLockRelease(FdwXactLock);
-
-	return (idx >= 0);
-}
-
 /*
  * We must fsync the foreign transaction state file that is valid or generated
  * during redo and has a inserted LSN <= the checkpoint's redo horizon.
@@ -607,11 +685,85 @@ CheckPointFdwXacts(XLogRecPtr redo_horizon)
 							   serialized_fdwxacts)));
 }
 
+/* Return true if the current transaction needs to use two-phase commit */
+bool
+FdwXactIsForeignTwophaseCommitRequired(void)
+{
+	return ForeignTwophaseCommitIsRequired;
+}
+
+/*
+ * 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)
+{
+	FdwXactParticipant *fdw_part;
+	HASH_SEQ_STATUS scan;
+	bool	have_no_twophase = false;
+	int		nserverswritten = 0;
+
+	if (!IsForeignTwophaseCommitRequested())
+		return false;
+
+	hash_seq_init(&scan, FdwXactParticipants);
+	while ((fdw_part = (FdwXactParticipant *) hash_seq_search(&scan)))
+	{
+		if (!fdw_part->modified)
+			continue;
+
+		if (!ServerSupportTwophaseCommit(fdw_part))
+			have_no_twophase = 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 performing
+	 * 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_no_twophase)
+		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.
+ * 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(TransactionId xid)
+FdwXactPrepareForeignTransactions(TransactionId xid, bool prepare_all)
 {
 	FdwXactParticipant *fdw_part;
 	HASH_SEQ_STATUS scan;
@@ -630,6 +782,9 @@ FdwXactPrepareForeignTransactions(TransactionId xid)
 
 		CHECK_FOR_INTERRUPTS();
 
+		if (!prepare_all && !fdw_part->modified)
+			continue;
+
 		/* Get prepared transaction identifier */
 		identifier = getFdwXactIdentifier(fdw_part, xid);
 		Assert(identifier);
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index cf4d5f2574..c064fc936f 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -22,6 +22,7 @@
 
 #include "access/commit_ts.h"
 #include "access/fdwxact.h"
+#include "access/fdwxact_launcher.h"
 #include "access/multixact.h"
 #include "access/parallel.h"
 #include "access/subtrans.h"
@@ -1456,6 +1457,9 @@ RecordTransactionCommit(void)
 	if (wrote_xlog && markXidCommitted)
 		SyncRepWaitForLSN(XactLastRecEnd, true);
 
+	if (FdwXactIsForeignTwophaseCommitRequired())
+		FdwXactLaunchOrWakeupResolver();
+
 	/* remember end of last commit record */
 	XactLastCommitEnd = XactLastRecEnd;
 
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 7d2320834b..517ebdf5ca 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -501,6 +501,24 @@ static struct config_enum_entry shared_memory_options[] = {
 	{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}
+};
+
 /*
  * Options for enum values stored in other modules
  */
@@ -4713,6 +4731,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 3e2c013680..a0fcc053fb 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -747,6 +747,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 9d6d66ac6f..58ba7cf848 100644
--- a/src/include/access/fdwxact.h
+++ b/src/include/access/fdwxact.h
@@ -22,6 +22,14 @@
 											 * without preparation */
 #define FDWXACT_FLAG_PARALLEL_WORKER	0x02	/* is parallel worker? */
 
+/* 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
 {
@@ -103,6 +111,7 @@ extern int	max_prepared_foreign_xacts;
 extern int	max_foreign_xact_resolvers;
 extern int	foreign_xact_resolution_retry_interval;
 extern int	foreign_xact_resolver_timeout;
+extern int	foreign_twophase_commit;
 
 /* Function declarations */
 extern void PreCommit_FdwXact(bool is_parallel_worker);
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 4bb6f3738b..26e439e0cc 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -284,7 +284,7 @@ extern bool IsImportableForeignTable(const char *tablename,
 extern Path *GetExistingLocalJoinPath(RelOptInfo *joinrel);
 
 /* Functions in fdwxact/fdwxact.c */
-extern void FdwXactRegisterXact(UserMapping *usermapping);
+extern void FdwXactRegisterXact(UserMapping *usermapping, bool modified);
 extern void FdwXactUnregisterXact(UserMapping *usermapping);
 
 #endif							/* FDWAPI_H */
-- 
2.27.0

