From 0637a264b1a915786f23a2eccf68c4594309f882 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 v36 5/9] 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          | 168 +++++++++++++++++-
 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, 207 insertions(+), 6 deletions(-)

diff --git a/src/backend/access/transam/fdwxact.c b/src/backend/access/transam/fdwxact.c
index 8100d4dd1d..bc4540d3ae 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 for each foreign transaction regardless of data on
  * the foreign server having been modified.  At COMMIT PREPARED and ROLLBACK PREPARED,
@@ -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"
@@ -118,6 +137,10 @@
 #define ServerSupportTwophaseCommit(fdwent) \
 	(((FdwXactEntry *)(fdwent))->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 '_'.
@@ -151,6 +174,9 @@ typedef struct FdwXactEntry
 	 */
 	FdwXactState		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;
@@ -158,6 +184,7 @@ typedef struct FdwXactEntry
 } FdwXactEntry;
 
 /*
+
  * The current distributed transaction state. Members of participants must
  * support at least both commit and rollback APIs
  * (ServerSupportTransactionCallback() is true)..
@@ -166,14 +193,20 @@ typedef struct DistributedXactStateData
 {
 	bool	all_prepared; /* all participants are prepared? */
 
+	bool	twophase_commit_required;
+
 	/* Statistics of participants */
 	int		nparticipants_no_twophase; /* how many participants doesn't support
 										* two-phase commit protocol? */
+	int		nparticipants_modified;		/* how many participants are modified? */
+
 	HTAB	*participants;
 } DistributedXactStateData;
 static DistributedXactStateData DistributedXactState = {
 	.all_prepared = false,
+	.twophase_commit_required = false,
 	.nparticipants_no_twophase = 0,
+	.nparticipants_modified = 0,
 	.participants = NULL,
 };
 
@@ -185,9 +218,11 @@ static DistributedXactStateData DistributedXactState = {
 /* 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 RemoveFdwXactEntry(Oid umid);
 static void EndFdwXactEntry(FdwXactEntry *fdwent, bool isCommit,
@@ -195,7 +230,7 @@ static void EndFdwXactEntry(FdwXactEntry *fdwent, bool isCommit,
 static char *getFdwXactIdentifier(FdwXactEntry *fdwent, TransactionId xid);
 static int ForgetAllParticipants(void);
 
-static void FdwXactPrepareForeignTransactions(TransactionId xid);
+static void FdwXactPrepareForeignTransactions(TransactionId xid, bool prepare_all);
 static FdwXactState FdwXactInsertEntry(TransactionId xid, FdwXactEntry *fdwent,
 									   char *identifier);
 static void AtProcExit_FdwXact(int code, Datum arg);
@@ -208,6 +243,7 @@ static char *ProcessFdwXactBuffer(TransactionId xid, Oid umid,
 								  XLogRecPtr insert_start_lsn, bool fromdisk);
 static char *ReadFdwXactStateFile(TransactionId xid, Oid umid);
 static void RemoveFdwXactStateFile(TransactionId xid, Oid umid, bool giveWarning);
+static bool checkForeignTwophaseCommitRequired(bool local_modified);
 
 static FdwXactState insert_fdwxact(Oid dbid, TransactionId xid, Oid umid, Oid serverid,
 							  Oid owner, char *identifier);
@@ -284,7 +320,7 @@ FdwXactShmemInit(void)
  * mapping OID as a participant of the transaction.
  */
 void
-FdwXactRegisterXact(UserMapping *usermapping)
+FdwXactRegisterXact(UserMapping *usermapping, bool modified)
 {
 	FdwXactEntry	*fdwent;
 	FdwRoutine	*routine;
@@ -310,8 +346,21 @@ FdwXactRegisterXact(UserMapping *usermapping)
 	fdwent = hash_search(DistributedXactState.participants,
 						 (void *) &umid, HASH_ENTER, &found);
 
+	/* Already registered */
 	if (found)
+	{
+		/* Update statistics if necessary  */
+		if (fdwent->modified && !modified)
+			DistributedXactState.nparticipants_modified--;
+		else if (!fdwent->modified && modified)
+			DistributedXactState.nparticipants_modified++;
+
+		fdwent->modified = modified;
+
+		Assert(DistributedXactState.nparticipants_modified <=
+		   hash_get_num_entries(DistributedXactState.participants));
 		return;
+	}
 
 	/*
 	 * The participant information needs to live until the end of the transaction
@@ -332,6 +381,7 @@ FdwXactRegisterXact(UserMapping *usermapping)
 				(errmsg("cannot register foreign server not supporting transaction callback")));
 
 	fdwent->fdwxact = NULL;
+	fdwent->modified = modified;
 	fdwent->commit_foreign_xact_fn = routine->CommitForeignTransaction;
 	fdwent->rollback_foreign_xact_fn = routine->RollbackForeignTransaction;
 	fdwent->prepare_foreign_xact_fn = routine->PrepareForeignTransaction;
@@ -341,9 +391,13 @@ FdwXactRegisterXact(UserMapping *usermapping)
 	/* Update statistics */
 	if (!ServerSupportTwophaseCommit(fdwent))
 		DistributedXactState.nparticipants_no_twophase++;
+	if (fdwent->modified)
+		DistributedXactState.nparticipants_modified++;
 
 	Assert(DistributedXactState.nparticipants_no_twophase <=
 			   hash_get_num_entries(DistributedXactState.participants));
+	Assert(DistributedXactState.nparticipants_modified <=
+		   hash_get_num_entries(DistributedXactState.participants));
 }
 
 /* Remove the foreign transaction from the current participants */
@@ -372,9 +426,13 @@ RemoveFdwXactEntry(Oid umid)
 		/* Update statistics */
 		if (!ServerSupportTwophaseCommit(fdwent))
 			DistributedXactState.nparticipants_no_twophase--;
+		if (fdwent->modified)
+			DistributedXactState.nparticipants_modified--;
 
 		Assert(DistributedXactState.nparticipants_no_twophase <=
 			   hash_get_num_entries(DistributedXactState.participants));
+		Assert(DistributedXactState.nparticipants_modified <=
+			   hash_get_num_entries(DistributedXactState.participants));
 	}
 }
 
@@ -441,7 +499,9 @@ AtEOXact_FdwXact(bool isCommit, bool is_parallel_worker)
 
 	/* Reset all fields */
 	DistributedXactState.all_prepared = false;
+	DistributedXactState.twophase_commit_required = false;
 	DistributedXactState.nparticipants_no_twophase = 0;
+	DistributedXactState.nparticipants_modified = 0;
 }
 
 /*
@@ -505,7 +565,7 @@ AtPrepare_FdwXact(void)
 	 * prepare all foreign transactions.
 	 */
 	xid = GetTopTransactionId();
-	FdwXactPrepareForeignTransactions(xid);
+	FdwXactPrepareForeignTransactions(xid, true);
 
 	/*
 	 * Remember we already prepared all participants.  We keep participants
@@ -524,6 +584,8 @@ PreCommit_FdwXact(bool is_parallel_worker)
 {
 	HASH_SEQ_STATUS scan;
 	FdwXactEntry *fdwent;
+	TransactionId xid;
+	bool		local_modified;
 
 	/*
 	 * If there is no foreign server involved or all foreign transactions
@@ -535,6 +597,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));
+
+	/*
+	 * Perform twophase commit if required. 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);
+		DistributedXactState.twophase_commit_required = true;
+
+		return;
+	}
+
 	/* Commit all foreign transactions in the participant list */
 	hash_seq_init(&scan, DistributedXactState.participants);
 	while ((fdwent = (FdwXactEntry *) hash_seq_search(&scan)))
@@ -650,11 +746,70 @@ CheckPointFdwXacts(XLogRecPtr redo_horizon)
 							   serialized_fdwxacts)));
 }
 
+/* Return true if the current transaction needs to use two-phase commit */
+bool
+FdwXactIsForeignTwophaseCommitRequired(void)
+{
+	return DistributedXactState.twophase_commit_required;
+}
+
 /*
- * Insert FdwXactState entries and prepare foreign transactions.
+ * 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)
+{
+	int		nserverswritten;
+
+	if (!IsForeignTwophaseCommitRequested())
+		return false;
+
+	nserverswritten = DistributedXactState.nparticipants_modified;
+
+	/* 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;
+
+	if (DistributedXactState.nparticipants_no_twophase > 0)
+		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")));
+
+	/* 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.")));
+
+	return true;
+}
+
+/*
+ * Insert FdwXactState 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)
 {
 	FdwXactEntry *fdwent;
 	HASH_SEQ_STATUS scan;
@@ -673,6 +828,9 @@ FdwXactPrepareForeignTransactions(TransactionId xid)
 
 		CHECK_FOR_INTERRUPTS();
 
+		if (!prepare_all && !fdwent->modified)
+			continue;
+
 		/* Get prepared transaction identifier */
 		identifier = getFdwXactIdentifier(fdwent, xid);
 		Assert(identifier);
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index bfe5e11245..cec4813f37 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 853ffb8f97..a79fb503e8 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -523,6 +523,24 @@ static struct config_enum_entry default_toast_compression_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
  */
@@ -4798,6 +4816,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 d5abe7d4a7..89738b6632 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -754,6 +754,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 05fc1beb2e..0d71e48a59 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 aa93ddc7ae..55298f8dae 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -303,7 +303,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.24.3 (Apple Git-128)

