From edcdf0a8b9f780b9fdb9a0430ee88a007975399a Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Sun, 10 Jun 2018 18:46:01 +0900
Subject: [PATCH v19 3/4] postgres_fdw supports atomic commit APIs.

---
 contrib/postgres_fdw/connection.c              | 538 +++++++++++++++++++------
 contrib/postgres_fdw/expected/postgres_fdw.out | 387 +++++++++++++++++-
 contrib/postgres_fdw/option.c                  |   5 +-
 contrib/postgres_fdw/postgres_fdw.c            |  60 ++-
 contrib/postgres_fdw/postgres_fdw.h            |  10 +-
 contrib/postgres_fdw/sql/postgres_fdw.sql      | 151 ++++++-
 doc/src/sgml/postgres-fdw.sgml                 |  37 ++
 7 files changed, 1044 insertions(+), 144 deletions(-)

diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index fe4893a..28f87a6 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -14,9 +14,12 @@
 
 #include "postgres_fdw.h"
 
+#include "access/fdwxact.h"
 #include "access/htup_details.h"
-#include "catalog/pg_user_mapping.h"
 #include "access/xact.h"
+#include "catalog/pg_user_mapping.h"
+#include "commands/defrem.h"
+#include "foreign/fdwapi.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
 #include "pgstat.h"
@@ -56,6 +59,7 @@ typedef struct ConnCacheEntry
 	bool		have_error;		/* have any subxacts aborted in this xact? */
 	bool		changing_xact_state;	/* xact state change in process */
 	bool		invalidated;	/* true if reconnect is pending */
+	bool		am_participant_of_ac;	/* true if fdwxact code control the transaction */
 	uint32		server_hashvalue;	/* hash value of foreign server OID */
 	uint32		mapping_hashvalue;	/* hash value of user mapping OID */
 } ConnCacheEntry;
@@ -78,7 +82,7 @@ static void disconnect_pg_server(ConnCacheEntry *entry);
 static void check_conn_params(const char **keywords, const char **values, UserMapping *user);
 static void configure_remote_session(PGconn *conn);
 static void do_sql_command(PGconn *conn, const char *sql);
-static void begin_remote_xact(ConnCacheEntry *entry);
+static void begin_remote_xact(ConnCacheEntry *entry, Oid serverid, Oid userid);
 static void pgfdw_xact_callback(XactEvent event, void *arg);
 static void pgfdw_subxact_callback(SubXactEvent event,
 					   SubTransactionId mySubid,
@@ -91,20 +95,14 @@ static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query,
 						 bool ignore_errors);
 static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
 						 PGresult **result);
-
-
-/*
- * Get a PGconn which can be used to execute queries on the remote PostgreSQL
- * server with the user's authorization.  A new connection is established
- * if we don't already have a suitable one, and a transaction is opened at
- * the right subtransaction nesting depth if we didn't do that already.
- *
- * will_prep_stmt must be true if caller intends to create any prepared
- * statements.  Since those don't go away automatically at transaction end
- * (not even on error), we need this flag to cue manual cleanup.
- */
-PGconn *
-GetConnection(UserMapping *user, bool will_prep_stmt)
+static bool pgfdw_commit_transaction(ConnCacheEntry *entry);
+static bool pgfdw_rollback_transaction(ConnCacheEntry *entry);
+static void pgfdw_cleanup_after_transaction(ConnCacheEntry *entry);
+static ConnCacheEntry *GetConnectionState(Oid umid, bool will_prep_stmt,
+										  bool start_transaction);
+
+static ConnCacheEntry *
+GetConnectionState(Oid umid, bool will_prep_stmt, bool start_transaction)
 {
 	bool		found;
 	ConnCacheEntry *entry;
@@ -136,11 +134,8 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
 									  pgfdw_inval_callback, (Datum) 0);
 	}
 
-	/* Set flag that we did GetConnection during the current transaction */
-	xact_got_connection = true;
-
 	/* Create hash key for the entry.  Assume no pad bytes in key struct */
-	key = user->umid;
+	key = umid;
 
 	/*
 	 * Find or create cached entry for requested connection.
@@ -182,6 +177,7 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
 	 */
 	if (entry->conn == NULL)
 	{
+		UserMapping		*user = GetUserMappingByOid(umid);
 		ForeignServer *server = GetForeignServer(user->serverid);
 
 		/* Reset all transient state fields, to be sure all are clean */
@@ -190,6 +186,7 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
 		entry->have_error = false;
 		entry->changing_xact_state = false;
 		entry->invalidated = false;
+		entry->am_participant_of_ac = false;
 		entry->server_hashvalue =
 			GetSysCacheHashValue1(FOREIGNSERVEROID,
 								  ObjectIdGetDatum(server->serverid));
@@ -200,6 +197,15 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
 		/* Now try to make the connection */
 		entry->conn = connect_pg_server(server, user);
 
+		Assert(entry->conn);
+
+		if (!entry->conn)
+		{
+			elog(DEBUG3, "attempt to connection to server \"%s\" by postgres_fdw failed",
+				 server->servername);
+			return NULL;
+		}
+
 		elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)",
 			 entry->conn, server->servername, user->umid, user->userid);
 	}
@@ -207,16 +213,46 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
 	/*
 	 * Start a new transaction or subtransaction if needed.
 	 */
-	begin_remote_xact(entry);
+	if (start_transaction)
+	{
+		UserMapping		*user = GetUserMappingByOid(umid);
+
+		begin_remote_xact(entry, user->serverid, user->userid);
+
+		/* Set flag that we did GetConnection during the current transaction */
+		xact_got_connection = true;
+	}
 
 	/* Remember if caller will prepare statements */
 	entry->have_prep_stmt |= will_prep_stmt;
 
+	return entry;
+}
+
+/*
+ * Get a PGconn which can be used to execute queries on the remote PostgreSQL
+ * server with the user's authorization.  A new connection is established
+ * if we don't already have a suitable one, and a transaction is opened at
+ * the right subtransaction nesting depth if we didn't do that already.
+ *
+ * will_prep_stmt must be true if caller intends to create any prepared
+ * statements.  Since those don't go away automatically at transaction end
+ * (not even on error), we need this flag to cue manual cleanup.
+ */
+PGconn *
+GetConnection(Oid umid, bool will_prep_stmt, bool start_transaction)
+{
+	ConnCacheEntry *entry;
+
+	entry = GetConnectionState(umid, will_prep_stmt, start_transaction);
+
 	return entry->conn;
 }
 
 /*
  * Connect to remote server using specified server and user mapping properties.
+ * If the attempt to connect fails, and the caller can handle connection failure
+ * (connection_error_ok = true) return NULL, throw error otherwise.
  */
 static PGconn *
 connect_pg_server(ForeignServer *server, UserMapping *user)
@@ -265,11 +301,22 @@ connect_pg_server(ForeignServer *server, UserMapping *user)
 
 		conn = PQconnectdbParams(keywords, values, false);
 		if (!conn || PQstatus(conn) != CONNECTION_OK)
+		{
+			char	   *connmessage;
+			int			msglen;
+
+			/* libpq typically appends a newline, strip that */
+			connmessage = pstrdup(PQerrorMessage(conn));
+			msglen = strlen(connmessage);
+			if (msglen > 0 && connmessage[msglen - 1] == '\n')
+				connmessage[msglen - 1] = '\0';
+
 			ereport(ERROR,
 					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
 					 errmsg("could not connect to server \"%s\"",
 							server->servername),
 					 errdetail_internal("%s", pchomp(PQerrorMessage(conn)))));
+		}
 
 		/*
 		 * Check that non-superuser has used password to establish connection;
@@ -414,15 +461,24 @@ do_sql_command(PGconn *conn, const char *sql)
  * control which remote queries share a snapshot.
  */
 static void
-begin_remote_xact(ConnCacheEntry *entry)
+begin_remote_xact(ConnCacheEntry *entry, Oid serverid, Oid userid)
 {
 	int			curlevel = GetCurrentTransactionNestLevel();
+	ForeignServer	*server = GetForeignServer(serverid);
 
 	/* Start main transaction if we haven't yet */
 	if (entry->xact_depth <= 0)
 	{
 		const char *sql;
 
+		/* Register the new foreign server if enabled */
+		if (server_uses_twophase_commit(server))
+		{
+			/* Register foreign server with auto-generated identifer */
+			FdwXactRegisterForeignTransaction(serverid, userid, NULL);
+			entry->am_participant_of_ac = true;
+		}
+
 		elog(DEBUG3, "starting remote transaction on connection %p",
 			 entry->conn);
 
@@ -650,12 +706,11 @@ static void
 pgfdw_xact_callback(XactEvent event, void *arg)
 {
 	HASH_SEQ_STATUS scan;
-	ConnCacheEntry *entry;
+	ConnCacheEntry	*entry;
 
-	/* Quick exit if no connections were touched in this transaction. */
+	/* Quick exit if no connections were touched in this transaction */
 	if (!xact_got_connection)
 		return;
-
 	/*
 	 * Scan all connection cache entries to find open remote transactions, and
 	 * close them.
@@ -663,17 +718,20 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 	hash_seq_init(&scan, ConnectionHash);
 	while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
 	{
-		PGresult   *res;
-
 		/* Ignore cache entry if no open connection right now */
 		if (entry->conn == NULL)
 			continue;
 
+		/*
+		 * Foreign transactions participating to atomic commit are ended
+		 * by two-phase commit APIs. Ignore them.
+		 */
+		if (entry->am_participant_of_ac)
+			continue;
+
 		/* If it has an open remote transaction, try to close it */
 		if (entry->xact_depth > 0)
 		{
-			bool		abort_cleanup_failure = false;
-
 			elog(DEBUG3, "closing remote transaction on connection %p",
 				 entry->conn);
 
@@ -681,40 +739,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 			{
 				case XACT_EVENT_PARALLEL_PRE_COMMIT:
 				case XACT_EVENT_PRE_COMMIT:
-
-					/*
-					 * If abort cleanup previously failed for this connection,
-					 * we can't issue any more commands against it.
-					 */
-					pgfdw_reject_incomplete_xact_state_change(entry);
-
-					/* Commit all remote transactions during pre-commit */
-					entry->changing_xact_state = true;
-					do_sql_command(entry->conn, "COMMIT TRANSACTION");
-					entry->changing_xact_state = false;
-
-					/*
-					 * If there were any errors in subtransactions, and we
-					 * made prepared statements, do a DEALLOCATE ALL to make
-					 * sure we get rid of all prepared statements. This is
-					 * annoying and not terribly bulletproof, but it's
-					 * probably not worth trying harder.
-					 *
-					 * DEALLOCATE ALL only exists in 8.3 and later, so this
-					 * constrains how old a server postgres_fdw can
-					 * communicate with.  We intentionally ignore errors in
-					 * the DEALLOCATE, so that we can hobble along to some
-					 * extent with older servers (leaking prepared statements
-					 * as we go; but we don't really support update operations
-					 * pre-8.3 anyway).
-					 */
-					if (entry->have_prep_stmt && entry->have_error)
-					{
-						res = PQexec(entry->conn, "DEALLOCATE ALL");
-						PQclear(res);
-					}
-					entry->have_prep_stmt = false;
-					entry->have_error = false;
+					pgfdw_commit_transaction(entry);
 					break;
 				case XACT_EVENT_PRE_PREPARE:
 
@@ -739,66 +764,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 					break;
 				case XACT_EVENT_PARALLEL_ABORT:
 				case XACT_EVENT_ABORT:
-
-					/*
-					 * Don't try to clean up the connection if we're already
-					 * in error recursion trouble.
-					 */
-					if (in_error_recursion_trouble())
-						entry->changing_xact_state = true;
-
-					/*
-					 * If connection is already unsalvageable, don't touch it
-					 * further.
-					 */
-					if (entry->changing_xact_state)
-						break;
-
-					/*
-					 * Mark this connection as in the process of changing
-					 * transaction state.
-					 */
-					entry->changing_xact_state = true;
-
-					/* Assume we might have lost track of prepared statements */
-					entry->have_error = true;
-
-					/*
-					 * If a command has been submitted to the remote server by
-					 * using an asynchronous execution function, the command
-					 * might not have yet completed.  Check to see if a
-					 * command is still being processed by the remote server,
-					 * and if so, request cancellation of the command.
-					 */
-					if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
-						!pgfdw_cancel_query(entry->conn))
-					{
-						/* Unable to cancel running query. */
-						abort_cleanup_failure = true;
-					}
-					else if (!pgfdw_exec_cleanup_query(entry->conn,
-													   "ABORT TRANSACTION",
-													   false))
-					{
-						/* Unable to abort remote transaction. */
-						abort_cleanup_failure = true;
-					}
-					else if (entry->have_prep_stmt && entry->have_error &&
-							 !pgfdw_exec_cleanup_query(entry->conn,
-													   "DEALLOCATE ALL",
-													   true))
-					{
-						/* Trouble clearing prepared statements. */
-						abort_cleanup_failure = true;
-					}
-					else
-					{
-						entry->have_prep_stmt = false;
-						entry->have_error = false;
-					}
-
-					/* Disarm changing_xact_state if it all worked. */
-					entry->changing_xact_state = abort_cleanup_failure;
+					pgfdw_rollback_transaction(entry);
 					break;
 			}
 		}
@@ -1193,3 +1159,329 @@ exit:	;
 		*result = last_res;
 	return timed_out;
 }
+
+/*
+ * The function prepares transaction on foreign server. This function
+ * is called only at the pre-commit phase of the local transaction. Since
+ * we should have the connection to the server that we are interested in
+ * we don't use serverid and userid that are necessary to get user mapping
+ * that is the key of the connection cache.
+ */
+bool
+postgresPrepareForeignTransaction(FdwXactResolveState *state)
+{
+	ConnCacheEntry *entry = NULL;
+	bool		result = false;
+	PGresult	*res;
+	StringInfo	command;
+
+	entry = hash_search(ConnectionHash, &(state->umid), HASH_FIND, NULL);
+
+	if (!entry->conn)
+		return false;
+
+	pgfdw_reject_incomplete_xact_state_change(entry);
+
+	command = makeStringInfo();
+	appendStringInfo(command, "PREPARE TRANSACTION '%s'", state->fdwxact_id);
+
+	/* Do commit foreign transaction */
+	entry->changing_xact_state = true;
+	res = pgfdw_exec_query(entry->conn, command->data);
+	entry->changing_xact_state = false;
+
+	if (PQresultStatus(res) == PGRES_COMMAND_OK)
+		result = true;
+
+	if (result)
+		elog(DEBUG1, "prepared foreign transaction on server %u with ID %s",
+			 state->serverid, state->fdwxact_id);
+
+	if (entry->have_prep_stmt && entry->have_error)
+	{
+		res = PQexec(entry->conn, "DEALLOCATE ALL");
+		PQclear(res);
+	}
+
+	entry->have_prep_stmt = false;
+	entry->have_error  = false;
+
+	pgfdw_cleanup_after_transaction(entry);
+
+	return result;
+}
+
+/*
+ * The function commits the transactionon foreign server. This
+ * function is called both at the pre-commit phase of the local transaction
+ * when committing and at the end of the local transaction when aborting.
+ * Since we should the connections to the server that involved with the local
+ * transaction we don't use serverid and userid that are necessary to get
+ * user mapping that is the key of connection cache.
+ */
+bool
+postgresCommitForeignTransaction(FdwXactResolveState *state)
+{
+	ConnCacheEntry *entry = NULL;
+	bool		result = false;
+
+	entry = hash_search(ConnectionHash, &(state->umid),
+						HASH_FIND, NULL);
+
+	result = pgfdw_commit_transaction(entry);
+
+	return result;
+}
+
+/*
+ * The function rollbacks the transactionon foreign server. This
+ * function is called both at the pre-commit phase of the local transaction
+ * when committing and at the end of the local transaction when aborting.
+ * Since we should the connections to the server that involved with the local
+ * transaction we don't use serverid and userid that are necessary to get
+ * user mapping that is the key of connection cache.
+ */
+bool
+postgresRollbackForeignTransaction(FdwXactResolveState *state)
+{
+	ConnCacheEntry *entry = NULL;
+	bool ret;
+
+	entry = hash_search(ConnectionHash, &(state->umid),
+						HASH_FIND, NULL);
+
+	/* Rollback a remote transaction */
+	ret = pgfdw_rollback_transaction(entry);
+
+	return ret;
+}
+
+bool
+postgresResolveForeignTransaction(FdwXactResolveState *state, bool is_commit)
+{
+	ConnCacheEntry *entry = NULL;
+	StringInfo	command;
+	bool result;
+	PGresult	*res;
+
+	entry = GetConnectionState(state->umid, false, false);
+
+	if (!entry->conn)
+		return false;
+
+	command = makeStringInfo();
+	appendStringInfo(command, "%s PREPARED '%s'",
+					 is_commit ? "COMMIT" : "ROLLBACK",
+					 state->fdwxact_id);
+
+	res = pgfdw_exec_query(entry->conn, command->data);
+
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		int		sqlstate;
+		char	*diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
+
+		/*
+		 * The command failed, raise a warning to log the reason of failure.
+		 * We may not be in a transaction here, so raising error doesn't
+		 * help. Even if we are in a transaction, it would be the resolver
+		 * transaction, which will get aborted on raising error, thus
+		 * delaying resolution of other prepared foreign transactions.
+		 */
+		pgfdw_report_error(LOG, res, entry->conn, false, command->data);
+
+		if (diag_sqlstate)
+		{
+			sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
+									 diag_sqlstate[1],
+									 diag_sqlstate[2],
+									 diag_sqlstate[3],
+									 diag_sqlstate[4]);
+		}
+		else
+			sqlstate = ERRCODE_CONNECTION_FAILURE;
+
+		/*
+		 * If we tried to COMMIT/ABORT a prepared transaction and the prepared
+		 * transaction was missing on the foreign server, it was probably
+		 * resolved by some other means. Anyway, it should be considered as resolved.
+		 */
+		result = (sqlstate == ERRCODE_UNDEFINED_OBJECT);
+	}
+	else
+		result = true;
+
+	elog(DEBUG1, "%s prepared foreign transaction on server %u with ID %s",
+		 is_commit ? "commit" : "rollback",
+		 state->serverid,
+		 state->fdwxact_id);
+
+	/* Cleanup transaction status */
+	pgfdw_cleanup_after_transaction(entry);
+
+	return result;
+}
+/* Cleanup at main-transaction end */
+static void
+pgfdw_cleanup_after_transaction(ConnCacheEntry *entry)
+{
+	/* Reset state to show we're out of a transaction */
+	entry->xact_depth = 0;
+	entry->am_participant_of_ac = false;
+
+	/*
+	 * If the connection isn't in a good idle state, discard it to
+	 * recover. Next GetConnection will open a new connection.
+	 */
+	if (PQstatus(entry->conn) != CONNECTION_OK ||
+		PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
+		entry->changing_xact_state)
+	{
+		elog(DEBUG3, "discarding connection %p", entry->conn);
+		disconnect_pg_server(entry);
+	}
+
+	entry->changing_xact_state = false;
+
+	/*
+	 * Regardless of the event type, we can now mark ourselves as out of the
+	 * transaction.
+	 */
+	xact_got_connection = false;
+
+	/* Also reset cursor numbering for next transaction */
+	cursor_number = 0;
+}
+
+static bool
+pgfdw_rollback_transaction(ConnCacheEntry *entry)
+{
+	bool abort_cleanup_failure = false;
+
+	/*
+	 * In rollback local transaction, if we don't the connection
+	 * it means any transaction started. So we can regard it as
+	 * success.
+	 */
+	if (!entry || !entry->conn)
+		return true;
+
+	/*
+	 * Don't try to clean up the connection if we're already
+	 * in error recursion trouble.
+	 */
+	if (in_error_recursion_trouble())
+		entry->changing_xact_state = true;
+
+	/*
+	 * If connection is already unsalvageable, do only the cleanup
+	 * and don't touch it further.
+	 */
+	if (entry->changing_xact_state)
+	{
+		pgfdw_cleanup_after_transaction(entry);
+		return true;
+	}
+
+	/*
+	 * Mark this connection as in the process of changing
+	 * transaction state.
+	 */
+	entry->changing_xact_state = true;
+
+	/* Assume we might have lost track of prepared statements */
+	entry->have_error = true;
+
+	/*
+	 * If a command has been submitted to the remote server by
+	 * using an asynchronous execution function, the command
+	 * might not have yet completed.  Check to see if a
+	 * command is still being processed by the remote server,
+	 * and if so, request cancellation of the command.
+	 */
+	if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE &&
+		!pgfdw_cancel_query(entry->conn))
+	{
+		/* Unable to cancel running query. */
+		abort_cleanup_failure = true;
+	}
+	else if (!pgfdw_exec_cleanup_query(entry->conn,
+									   "ABORT TRANSACTION",
+									   false))
+	{
+		/* Unable to abort remote transaction. */
+		abort_cleanup_failure = true;
+	}
+	else if (entry->have_prep_stmt && entry->have_error &&
+			 !pgfdw_exec_cleanup_query(entry->conn,
+									   "DEALLOCATE ALL",
+									   true))
+	{
+		/* Trouble clearing prepared statements. */
+		abort_cleanup_failure = true;
+	}
+	else
+	{
+		entry->have_prep_stmt = false;
+		entry->have_error = false;
+	}
+
+	/* Disarm changing_xact_state if it all worked. */
+	entry->changing_xact_state = abort_cleanup_failure;
+
+	/* Cleanup transaction status */
+	pgfdw_cleanup_after_transaction(entry);
+
+	return !abort_cleanup_failure;
+}
+
+static bool
+pgfdw_commit_transaction(ConnCacheEntry *entry)
+{
+	PGresult	*res;
+	bool result = false;
+
+	if (!entry || !entry->conn)
+		return false;
+
+	/*
+	 * If abort cleanup previously failed for this connection,
+	 * we can't issue any more commands against it.
+	 */
+	pgfdw_reject_incomplete_xact_state_change(entry);
+
+	entry->changing_xact_state = true;
+	res = pgfdw_exec_query(entry->conn, "COMMIT TRANSACTION");
+	entry->changing_xact_state = false;
+
+	if (PQresultStatus(res) == PGRES_COMMAND_OK)
+		result = true;
+
+	/*
+	 * If there were any errors in subtransactions, and we
+	 * made prepared statements, do a DEALLOCATE ALL to make
+	 * sure we get rid of all prepared statements. This is
+	 * annoying and not terribly bulletproof, but it's
+	 * probably not worth trying harder.
+	 *
+	 * DEALLOCATE ALL only exists in 8.3 and later, so this
+	 * constrains how old a server postgres_fdw can
+	 * communicate with.  We intentionally ignore errors in
+	 * the DEALLOCATE, so that we can hobble along to some
+	 * extent with older servers (leaking prepared statements
+	 * as we go; but we don't really support update operations
+	 * pre-8.3 anyway).
+	 */
+	if (entry->have_prep_stmt && entry->have_error)
+	{
+		res = PQexec(entry->conn, "DEALLOCATE ALL");
+		PQclear(res);
+	}
+	entry->have_prep_stmt = false;
+	entry->have_error = false;
+
+	/* Cleanup transaction status */
+	pgfdw_cleanup_after_transaction(entry);
+
+	return result;
+}
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 21a2ef5..15dadf4 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -13,12 +13,17 @@ DO $d$
             OPTIONS (dbname '$$||current_database()||$$',
                      port '$$||current_setting('port')||$$'
             )$$;
+        EXECUTE $$CREATE SERVER loopback3 FOREIGN DATA WRAPPER postgres_fdw
+            OPTIONS (dbname '$$||current_database()||$$',
+                     port '$$||current_setting('port')||$$'
+            )$$;
     END;
 $d$;
 CREATE USER MAPPING FOR public SERVER testserver1
 	OPTIONS (user 'value', password 'value');
 CREATE USER MAPPING FOR CURRENT_USER SERVER loopback;
 CREATE USER MAPPING FOR CURRENT_USER SERVER loopback2;
+CREATE USER MAPPING FOR CURRENT_USER SERVER loopback3;
 -- ===================================================================
 -- create objects used through FDW loopback server
 -- ===================================================================
@@ -52,6 +57,13 @@ CREATE TABLE "S 1"."T 4" (
 	c3 text,
 	CONSTRAINT t4_pkey PRIMARY KEY (c1)
 );
+CREATE TABLE "S 1"."T 5" (
+       c1 int NOT NULL
+);
+CREATE TABLE "S 1"."T 6" (
+       c1 int NOT NULL,
+       CONSTRAINT t6_pkey PRIMARY KEY (c1)
+);
 -- Disable autovacuum for these tables to avoid unexpected effects of that
 ALTER TABLE "S 1"."T 1" SET (autovacuum_enabled = 'false');
 ALTER TABLE "S 1"."T 2" SET (autovacuum_enabled = 'false');
@@ -87,6 +99,7 @@ ANALYZE "S 1"."T 1";
 ANALYZE "S 1"."T 2";
 ANALYZE "S 1"."T 3";
 ANALYZE "S 1"."T 4";
+ANALYZE "S 1"."T 5";
 -- ===================================================================
 -- create foreign tables
 -- ===================================================================
@@ -129,6 +142,15 @@ CREATE FOREIGN TABLE ft6 (
 	c2 int NOT NULL,
 	c3 text
 ) SERVER loopback2 OPTIONS (schema_name 'S 1', table_name 'T 4');
+CREATE FOREIGN TABLE ft7_twophase (
+       c1 int NOT NULL
+) SERVER loopback OPTIONS (schema_name 'S 1', table_name 'T 5');
+CREATE FOREIGN TABLE ft8_twophase (
+       c1 int NOT NULL
+) SERVER loopback2 OPTIONS (schema_name 'S 1', table_name 'T 5');
+CREATE FOREIGN TABLE ft9_not_twophase (
+       c1 int NOT NULL
+) SERVER loopback3 OPTIONS (schema_name 'S 1', table_name 'T 5');
 -- A table with oids. CREATE FOREIGN TABLE doesn't support the
 -- WITH OIDS option, but ALTER does.
 CREATE FOREIGN TABLE ft_pg_type (
@@ -185,16 +207,19 @@ ALTER FOREIGN TABLE ft2 OPTIONS (schema_name 'S 1', table_name 'T 1');
 ALTER FOREIGN TABLE ft1 ALTER COLUMN c1 OPTIONS (column_name 'C 1');
 ALTER FOREIGN TABLE ft2 ALTER COLUMN c1 OPTIONS (column_name 'C 1');
 \det+
-                                      List of foreign tables
- Schema |   Table    |  Server   |                   FDW options                    | Description 
---------+------------+-----------+--------------------------------------------------+-------------
- public | ft1        | loopback  | (schema_name 'S 1', table_name 'T 1')            | 
- public | ft2        | loopback  | (schema_name 'S 1', table_name 'T 1')            | 
- public | ft4        | loopback  | (schema_name 'S 1', table_name 'T 3')            | 
- public | ft5        | loopback  | (schema_name 'S 1', table_name 'T 4')            | 
- public | ft6        | loopback2 | (schema_name 'S 1', table_name 'T 4')            | 
- public | ft_pg_type | loopback  | (schema_name 'pg_catalog', table_name 'pg_type') | 
-(6 rows)
+                                         List of foreign tables
+ Schema |      Table       |  Server   |                   FDW options                    | Description 
+--------+------------------+-----------+--------------------------------------------------+-------------
+ public | ft1              | loopback  | (schema_name 'S 1', table_name 'T 1')            | 
+ public | ft2              | loopback  | (schema_name 'S 1', table_name 'T 1')            | 
+ public | ft4              | loopback  | (schema_name 'S 1', table_name 'T 3')            | 
+ public | ft5              | loopback  | (schema_name 'S 1', table_name 'T 4')            | 
+ public | ft6              | loopback2 | (schema_name 'S 1', table_name 'T 4')            | 
+ public | ft7_twophase     | loopback  | (schema_name 'S 1', table_name 'T 5')            | 
+ public | ft8_twophase     | loopback2 | (schema_name 'S 1', table_name 'T 5')            | 
+ public | ft9_not_twophase | loopback3 | (schema_name 'S 1', table_name 'T 5')            | 
+ public | ft_pg_type       | loopback  | (schema_name 'pg_catalog', table_name 'pg_type') | 
+(9 rows)
 
 -- Test that alteration of server options causes reconnection
 -- Remote's errors might be non-English, so hide them to ensure stable results
@@ -8650,3 +8675,345 @@ SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700
 
 -- Clean-up
 RESET enable_partitionwise_aggregate;
+-- ===================================================================
+-- test atomic commit across foreign servers
+-- ===================================================================
+ALTER SERVER loopback OPTIONS(ADD two_phase_commit 'on');
+ALTER SERVER loopback2 OPTIONS(ADD two_phase_commit 'on');
+ALTER SERVER loopback3 OPTIONS(ADD two_phase_commit 'off');
+\det+
+                                                 List of foreign tables
+ Schema |      Table       |  Server   |                            FDW options                            | Description 
+--------+------------------+-----------+-------------------------------------------------------------------+-------------
+ public | fpagg_tab_p1     | loopback  | (table_name 'pagg_tab_p1')                                        | 
+ public | fpagg_tab_p2     | loopback  | (table_name 'pagg_tab_p2')                                        | 
+ public | fpagg_tab_p3     | loopback  | (table_name 'pagg_tab_p3')                                        | 
+ public | ft1              | loopback  | (schema_name 'S 1', table_name 'T 1')                             | 
+ public | ft2              | loopback  | (schema_name 'S 1', table_name 'T 1', use_remote_estimate 'true') | 
+ public | ft3              | loopback  | (table_name 'loct3', use_remote_estimate 'true')                  | 
+ public | ft4              | loopback  | (schema_name 'S 1', table_name 'T 3')                             | 
+ public | ft5              | loopback  | (schema_name 'S 1', table_name 'T 4')                             | 
+ public | ft6              | loopback2 | (schema_name 'S 1', table_name 'T 4')                             | 
+ public | ft7_twophase     | loopback  | (schema_name 'S 1', table_name 'T 5')                             | 
+ public | ft8_twophase     | loopback2 | (schema_name 'S 1', table_name 'T 5')                             | 
+ public | ft9_not_twophase | loopback3 | (schema_name 'S 1', table_name 'T 5')                             | 
+ public | ft_pg_type       | loopback  | (schema_name 'pg_catalog', table_name 'pg_type')                  | 
+ public | ftprt1_p1        | loopback  | (table_name 'fprt1_p1', use_remote_estimate 'true')               | 
+ public | ftprt1_p2        | loopback  | (table_name 'fprt1_p2')                                           | 
+ public | ftprt2_p1        | loopback  | (table_name 'fprt2_p1', use_remote_estimate 'true')               | 
+ public | ftprt2_p2        | loopback  | (table_name 'fprt2_p2', use_remote_estimate 'true')               | 
+ public | rem1             | loopback  | (table_name 'loc1')                                               | 
+ public | rem2             | loopback  | (table_name 'loc2')                                               | 
+(19 rows)
+
+-- Check two_phase_commit setting
+SELECT srvname FROM pg_foreign_server WHERE 'two_phase_commit=on' = ANY(srvoptions) or 'two_phase_commit=off' = ANY(srvoptions);
+  srvname  
+-----------
+ loopback
+ loopback2
+ loopback3
+(3 rows)
+
+-- Enable atomic commit
+SET foreign_twophase_commit TO on;
+-- Modify one 2PC-capable server then commit and rollback.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(1);
+COMMIT;
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+(1 row)
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(1);
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+(1 row)
+
+-- Modify two 2PC-capable servers then commit and rollback.
+-- This requires to use 2PC when commit.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(2);
+INSERT INTO ft8_twophase VALUES(2);
+COMMIT;
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+  2
+  2
+(3 rows)
+
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+(3 rows)
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(3);
+INSERT INTO ft8_twophase VALUES(3);
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+  2
+  2
+(3 rows)
+
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+(3 rows)
+
+-- Modify both local data and 2PC-capable server then commit and rollback.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(4);
+INSERT INTO "S 1"."T 6" VALUES (4);
+COMMIT;
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+SELECT * FROM "S 1"."T 6";
+ c1 
+----
+  4
+(1 row)
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(5);
+INSERT INTO "S 1"."T 6" VALUES (5);
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+SELECT * FROM "S 1"."T 6";
+ c1 
+----
+  4
+(1 row)
+
+-- Modify foreign server and raise an error
+BEGIN;
+INSERT INTO ft7_twophase VALUES(8);
+INSERT INTO ft8_twophase VALUES(NULL); -- violation
+ERROR:  null value in column "c1" violates not-null constraint
+DETAIL:  Failing row contains (null).
+CONTEXT:  remote SQL command: INSERT INTO "S 1"."T 5"(c1) VALUES ($1)
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+-- Rollback foreign transaction that involves both 2PC-capable
+-- and 2PC-non-capable foreign servers.
+BEGIN;
+INSERT INTO ft8_twophase VALUES(7);
+INSERT INTO ft9_not_twophase VALUES(7);
+ROLLBACK;
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+SELECT * FROM ft9_not_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+-- Fails, cannot commit the distributed transaction if 2PC-non-capable
+-- server involved in.
+BEGIN;
+INSERT INTO ft8_twophase VALUES(6);
+INSERT INTO ft9_not_twophase VALUES(6);
+COMMIT; -- error
+ERROR:  cannot COMMIT a distributed transaction that has operated on foreign server that doesn't support atomic commit
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+SELECT * FROM ft9_not_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+(4 rows)
+
+-- Disables atomic commit, and success the same case as above.
+SET foreign_twophase_commit TO off;
+BEGIN;
+INSERT INTO ft8_twophase VALUES(6);
+INSERT INTO ft9_not_twophase VALUES(6);
+COMMIT; -- success
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+(5 rows)
+
+SELECT * FROM ft9_not_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+(5 rows)
+
+-- Enable atomic commit, again.
+SET foreign_twophase_commit TO on;
+-- Commit and rollback foreign transactions that are part of
+-- prepare transaction.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(9);
+INSERT INTO ft8_twophase VALUES(9);
+PREPARE TRANSACTION 'gx1';
+COMMIT PREPARED 'gx1';
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+  6
+  9
+  9
+(8 rows)
+
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+  6
+  9
+  9
+(8 rows)
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(10);
+INSERT INTO ft8_twophase VALUES(10);
+PREPARE TRANSACTION 'gx1';
+ROLLBACK PREPARED 'gx1';
+SELECT * FROM ft7_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+  6
+  9
+  9
+(8 rows)
+
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+  6
+  9
+  9
+(8 rows)
+
+-- Fails, cannot prepare the transaction if non-supporeted
+-- server involved in.
+BEGIN;
+INSERT INTO ft8_twophase VALUES(11);
+INSERT INTO ft9_not_twophase VALUES(11);
+PREPARE TRANSACTION 'gx1'; -- error
+ERROR:  cannot prepare a transaction that modified remote tables
+SELECT * FROM ft8_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+  6
+  9
+  9
+(8 rows)
+
+SELECT * FROM ft9_not_twophase;
+ c1 
+----
+  1
+  2
+  2
+  4
+  6
+  6
+  9
+  9
+(8 rows)
+
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index 6854f1b..1f45b1c 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -108,7 +108,8 @@ postgres_fdw_validator(PG_FUNCTION_ARGS)
 		 * Validate option value, when we can do so without any context.
 		 */
 		if (strcmp(def->defname, "use_remote_estimate") == 0 ||
-			strcmp(def->defname, "updatable") == 0)
+			strcmp(def->defname, "updatable") == 0 ||
+			strcmp(def->defname, "two_phase_commit") == 0)
 		{
 			/* these accept only boolean values */
 			(void) defGetBoolean(def);
@@ -177,6 +178,8 @@ InitPgFdwOptions(void)
 		/* fetch_size is available on both server and table */
 		{"fetch_size", ForeignServerRelationId, false},
 		{"fetch_size", ForeignTableRelationId, false},
+		/* two phase commit support */
+		{"two_phase_commit", ForeignServerRelationId, false},
 		{NULL, InvalidOid, false}
 	};
 
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index fd20aa9..1135046 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -14,6 +14,8 @@
 
 #include "postgres_fdw.h"
 
+#include "access/fdwxact.h"
+#include "access/xact.h"
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "catalog/pg_class.h"
@@ -359,6 +361,7 @@ static void postgresGetForeignUpperPaths(PlannerInfo *root,
 							 RelOptInfo *input_rel,
 							 RelOptInfo *output_rel,
 							 void *extra);
+static bool postgresIsTwoPhaseCommitEnabled(Oid serverid);
 
 /*
  * Helper functions
@@ -452,7 +455,6 @@ static void merge_fdw_options(PgFdwRelationInfo *fpinfo,
 				  const PgFdwRelationInfo *fpinfo_o,
 				  const PgFdwRelationInfo *fpinfo_i);
 
-
 /*
  * Foreign-data wrapper handler function: return a struct with pointers
  * to my callback routines.
@@ -506,10 +508,29 @@ postgres_fdw_handler(PG_FUNCTION_ARGS)
 	/* Support functions for upper relation push-down */
 	routine->GetForeignUpperPaths = postgresGetForeignUpperPaths;
 
+	/* Support functions for foreign transactions */
+	routine->PrepareForeignTransaction = postgresPrepareForeignTransaction;
+	routine->CommitForeignTransaction = postgresCommitForeignTransaction;
+	routine->RollbackForeignTransaction = postgresRollbackForeignTransaction;
+	routine->ResolveForeignTransaction = postgresResolveForeignTransaction;
+	routine->IsTwoPhaseCommitEnabled = postgresIsTwoPhaseCommitEnabled;
+
 	PG_RETURN_POINTER(routine);
 }
 
 /*
+ * postgresIsTwoPhaseCommitEnabled
+ */
+static bool
+postgresIsTwoPhaseCommitEnabled(Oid serverid)
+{
+	ForeignServer	*server = GetForeignServer(serverid);
+
+
+	return server_uses_twophase_commit(server);
+}
+
+/*
  * postgresGetForeignRelSize
  *		Estimate # of rows and width of the result of the scan
  *
@@ -1356,7 +1377,7 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags)
 	 * Get connection to the foreign server.  Connection manager will
 	 * establish new connection if necessary.
 	 */
-	fsstate->conn = GetConnection(user, false);
+	fsstate->conn = GetConnection(user->umid, false, true);
 
 	/* Assign a unique ID for my cursor */
 	fsstate->cursor_number = GetCursorNumber(fsstate->conn);
@@ -2411,7 +2432,7 @@ postgresBeginDirectModify(ForeignScanState *node, int eflags)
 	 * Get connection to the foreign server.  Connection manager will
 	 * establish new connection if necessary.
 	 */
-	dmstate->conn = GetConnection(user, false);
+	dmstate->conn = GetConnection(user->umid, false, true);
 
 	/* Update the foreign-join-related fields. */
 	if (fsplan->scan.scanrelid == 0)
@@ -2704,7 +2725,7 @@ estimate_path_cost_size(PlannerInfo *root,
 								&retrieved_attrs, NULL);
 
 		/* Get the remote estimate */
-		conn = GetConnection(fpinfo->user, false);
+		conn = GetConnection(fpinfo->user->umid, false, true);
 		get_remote_estimate(sql.data, conn, &rows, &width,
 							&startup_cost, &total_cost);
 		ReleaseConnection(conn);
@@ -3321,7 +3342,7 @@ create_foreign_modify(EState *estate,
 	user = GetUserMapping(userid, table->serverid);
 
 	/* Open connection; report that we'll create a prepared statement. */
-	fmstate->conn = GetConnection(user, true);
+	fmstate->conn = GetConnection(user->umid, true, true);
 	fmstate->p_name = NULL;		/* prepared statement not made yet */
 
 	/* Set up remote query information. */
@@ -4108,7 +4129,7 @@ postgresAnalyzeForeignTable(Relation relation,
 	 */
 	table = GetForeignTable(RelationGetRelid(relation));
 	user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
-	conn = GetConnection(user, false);
+	conn = GetConnection(user->umid, false, true);
 
 	/*
 	 * Construct command to get page count for relation.
@@ -4198,7 +4219,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel,
 	table = GetForeignTable(RelationGetRelid(relation));
 	server = GetForeignServer(table->serverid);
 	user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
-	conn = GetConnection(user, false);
+	conn = GetConnection(user->umid, false, true);
 
 	/*
 	 * Construct cursor that retrieves whole rows from remote.
@@ -4421,7 +4442,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
 	 */
 	server = GetForeignServer(serverOid);
 	mapping = GetUserMapping(GetUserId(), server->serverid);
-	conn = GetConnection(mapping, false);
+	conn = GetConnection(mapping->umid, false, true);
 
 	/* Don't attempt to import collation if remote server hasn't got it */
 	if (PQserverVersion(conn) < 90100)
@@ -5803,3 +5824,26 @@ find_em_expr_for_rel(EquivalenceClass *ec, RelOptInfo *rel)
 	/* We didn't find any suitable equivalence class expression */
 	return NULL;
 }
+
+/*
+ * server_uses_twophase_commit
+ * Returns true if the foreign server is configured to support 2PC.
+ */
+bool
+server_uses_twophase_commit(ForeignServer *server)
+{
+	ListCell		*lc;
+
+	/* Check the options for two phase compliance */
+	foreach(lc, server->options)
+	{
+		DefElem    *d = (DefElem *) lfirst(lc);
+
+		if (strcmp(d->defname, "two_phase_commit") == 0)
+		{
+			return defGetBoolean(d);
+		}
+	}
+	/* By default a server is not 2PC compliant */
+	return false;
+}
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h
index 70b538e..f01a71a 100644
--- a/contrib/postgres_fdw/postgres_fdw.h
+++ b/contrib/postgres_fdw/postgres_fdw.h
@@ -13,6 +13,7 @@
 #ifndef POSTGRES_FDW_H
 #define POSTGRES_FDW_H
 
+#include "access/fdwxact.h"
 #include "foreign/foreign.h"
 #include "lib/stringinfo.h"
 #include "nodes/relation.h"
@@ -115,7 +116,8 @@ extern int	set_transmission_modes(void);
 extern void reset_transmission_modes(int nestlevel);
 
 /* in connection.c */
-extern PGconn *GetConnection(UserMapping *user, bool will_prep_stmt);
+extern PGconn *GetConnection(Oid umid, bool will_prep_stmt, bool start_transaction);
+extern PGconn *GetExistingConnection(Oid umid);
 extern void ReleaseConnection(PGconn *conn);
 extern unsigned int GetCursorNumber(PGconn *conn);
 extern unsigned int GetPrepStmtNumber(PGconn *conn);
@@ -123,6 +125,11 @@ extern PGresult *pgfdw_get_result(PGconn *conn, const char *query);
 extern PGresult *pgfdw_exec_query(PGconn *conn, const char *query);
 extern void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
 				   bool clear, const char *sql);
+extern bool postgresPrepareForeignTransaction(FdwXactResolveState *state);
+extern bool postgresCommitForeignTransaction(FdwXactResolveState *state);
+extern bool postgresRollbackForeignTransaction(FdwXactResolveState *state);
+extern bool postgresResolveForeignTransaction(FdwXactResolveState *state,
+											  bool is_commit);
 
 /* in option.c */
 extern int ExtractConnectionOptions(List *defelems,
@@ -181,6 +188,7 @@ extern void deparseSelectStmtForRel(StringInfo buf, PlannerInfo *root,
 						List *remote_conds, List *pathkeys, bool is_subquery,
 						List **retrieved_attrs, List **params_list);
 extern const char *get_jointype_name(JoinType jointype);
+extern bool server_uses_twophase_commit(ForeignServer *server);
 
 /* in shippable.c */
 extern bool is_builtin(Oid objectId);
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 88c4cb4..2554c9c 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -15,6 +15,10 @@ DO $d$
             OPTIONS (dbname '$$||current_database()||$$',
                      port '$$||current_setting('port')||$$'
             )$$;
+        EXECUTE $$CREATE SERVER loopback3 FOREIGN DATA WRAPPER postgres_fdw
+            OPTIONS (dbname '$$||current_database()||$$',
+                     port '$$||current_setting('port')||$$'
+            )$$;
     END;
 $d$;
 
@@ -22,6 +26,7 @@ CREATE USER MAPPING FOR public SERVER testserver1
 	OPTIONS (user 'value', password 'value');
 CREATE USER MAPPING FOR CURRENT_USER SERVER loopback;
 CREATE USER MAPPING FOR CURRENT_USER SERVER loopback2;
+CREATE USER MAPPING FOR CURRENT_USER SERVER loopback3;
 
 -- ===================================================================
 -- create objects used through FDW loopback server
@@ -56,6 +61,14 @@ CREATE TABLE "S 1"."T 4" (
 	c3 text,
 	CONSTRAINT t4_pkey PRIMARY KEY (c1)
 );
+CREATE TABLE "S 1"."T 5" (
+       c1 int NOT NULL
+);
+
+CREATE TABLE "S 1"."T 6" (
+       c1 int NOT NULL,
+       CONSTRAINT t6_pkey PRIMARY KEY (c1)
+);
 
 -- Disable autovacuum for these tables to avoid unexpected effects of that
 ALTER TABLE "S 1"."T 1" SET (autovacuum_enabled = 'false');
@@ -94,6 +107,7 @@ ANALYZE "S 1"."T 1";
 ANALYZE "S 1"."T 2";
 ANALYZE "S 1"."T 3";
 ANALYZE "S 1"."T 4";
+ANALYZE "S 1"."T 5";
 
 -- ===================================================================
 -- create foreign tables
@@ -142,6 +156,19 @@ CREATE FOREIGN TABLE ft6 (
 	c3 text
 ) SERVER loopback2 OPTIONS (schema_name 'S 1', table_name 'T 4');
 
+CREATE FOREIGN TABLE ft7_twophase (
+       c1 int NOT NULL
+) SERVER loopback OPTIONS (schema_name 'S 1', table_name 'T 5');
+
+CREATE FOREIGN TABLE ft8_twophase (
+       c1 int NOT NULL
+) SERVER loopback2 OPTIONS (schema_name 'S 1', table_name 'T 5');
+
+CREATE FOREIGN TABLE ft9_not_twophase (
+       c1 int NOT NULL
+) SERVER loopback3 OPTIONS (schema_name 'S 1', table_name 'T 5');
+
+
 -- A table with oids. CREATE FOREIGN TABLE doesn't support the
 -- WITH OIDS option, but ALTER does.
 CREATE FOREIGN TABLE ft_pg_type (
@@ -2304,7 +2331,6 @@ SELECT t1.a, t2.b FROM fprt1 t1 INNER JOIN fprt2 t2 ON (t1.a = t2.b) WHERE t1.a
 
 RESET enable_partitionwise_join;
 
-
 -- ===================================================================
 -- test partitionwise aggregates
 -- ===================================================================
@@ -2354,3 +2380,126 @@ SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700
 
 -- Clean-up
 RESET enable_partitionwise_aggregate;
+
+-- ===================================================================
+-- test atomic commit across foreign servers
+-- ===================================================================
+
+ALTER SERVER loopback OPTIONS(ADD two_phase_commit 'on');
+ALTER SERVER loopback2 OPTIONS(ADD two_phase_commit 'on');
+ALTER SERVER loopback3 OPTIONS(ADD two_phase_commit 'off');
+
+\det+
+
+-- Check two_phase_commit setting
+SELECT srvname FROM pg_foreign_server WHERE 'two_phase_commit=on' = ANY(srvoptions) or 'two_phase_commit=off' = ANY(srvoptions);
+
+-- Enable atomic commit
+SET foreign_twophase_commit TO on;
+
+-- Modify one 2PC-capable server then commit and rollback.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(1);
+COMMIT;
+SELECT * FROM ft7_twophase;
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(1);
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+
+-- Modify two 2PC-capable servers then commit and rollback.
+-- This requires to use 2PC when commit.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(2);
+INSERT INTO ft8_twophase VALUES(2);
+COMMIT;
+SELECT * FROM ft7_twophase;
+SELECT * FROM ft8_twophase;
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(3);
+INSERT INTO ft8_twophase VALUES(3);
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+SELECT * FROM ft8_twophase;
+
+-- Modify both local data and 2PC-capable server then commit and rollback.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(4);
+INSERT INTO "S 1"."T 6" VALUES (4);
+COMMIT;
+SELECT * FROM ft7_twophase;
+SELECT * FROM "S 1"."T 6";
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(5);
+INSERT INTO "S 1"."T 6" VALUES (5);
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+SELECT * FROM "S 1"."T 6";
+
+-- Modify foreign server and raise an error
+BEGIN;
+INSERT INTO ft7_twophase VALUES(8);
+INSERT INTO ft8_twophase VALUES(NULL); -- violation
+ROLLBACK;
+SELECT * FROM ft7_twophase;
+SELECT * FROM ft8_twophase;
+
+-- Rollback foreign transaction that involves both 2PC-capable
+-- and 2PC-non-capable foreign servers.
+BEGIN;
+INSERT INTO ft8_twophase VALUES(7);
+INSERT INTO ft9_not_twophase VALUES(7);
+ROLLBACK;
+SELECT * FROM ft8_twophase;
+SELECT * FROM ft9_not_twophase;
+
+-- Fails, cannot commit the distributed transaction if 2PC-non-capable
+-- server involved in.
+BEGIN;
+INSERT INTO ft8_twophase VALUES(6);
+INSERT INTO ft9_not_twophase VALUES(6);
+COMMIT; -- error
+SELECT * FROM ft8_twophase;
+SELECT * FROM ft9_not_twophase;
+
+-- Disables atomic commit, and success the same case as above.
+SET foreign_twophase_commit TO off;
+BEGIN;
+INSERT INTO ft8_twophase VALUES(6);
+INSERT INTO ft9_not_twophase VALUES(6);
+COMMIT; -- success
+SELECT * FROM ft8_twophase;
+SELECT * FROM ft9_not_twophase;
+
+-- Enable atomic commit, again.
+SET foreign_twophase_commit TO on;
+
+-- Commit and rollback foreign transactions that are part of
+-- prepare transaction.
+BEGIN;
+INSERT INTO ft7_twophase VALUES(9);
+INSERT INTO ft8_twophase VALUES(9);
+PREPARE TRANSACTION 'gx1';
+COMMIT PREPARED 'gx1';
+SELECT * FROM ft7_twophase;
+SELECT * FROM ft8_twophase;
+
+BEGIN;
+INSERT INTO ft7_twophase VALUES(10);
+INSERT INTO ft8_twophase VALUES(10);
+PREPARE TRANSACTION 'gx1';
+ROLLBACK PREPARED 'gx1';
+SELECT * FROM ft7_twophase;
+SELECT * FROM ft8_twophase;
+
+-- Fails, cannot prepare the transaction if non-supporeted
+-- server involved in.
+BEGIN;
+INSERT INTO ft8_twophase VALUES(11);
+INSERT INTO ft9_not_twophase VALUES(11);
+PREPARE TRANSACTION 'gx1'; -- error
+SELECT * FROM ft8_twophase;
+SELECT * FROM ft9_not_twophase;
diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml
index 54b5e98..f4a9ff5 100644
--- a/doc/src/sgml/postgres-fdw.sgml
+++ b/doc/src/sgml/postgres-fdw.sgml
@@ -436,6 +436,43 @@
    </para>
 
   </sect3>
+
+  <sect3>
+   <title>Transaction Management Options</title>
+
+   <para>
+    By default, if the transaction involves with multiple remote server,
+    each transaction on remote server is committed or aborted independently.
+    Some of transactions may fail to commit on remote server while other
+    transactions commit successfully. This may be overridden using
+    following option:
+   </para>
+
+   <variablelist>
+
+    <varlistentry>
+     <term><literal>two_phase_commit</literal></term>
+     <listitem>
+      <para>
+       This option controls whether <filename>postgres_fdw</filename> allows
+       to use two-phase-commit when transaction commits. This option can
+       only be specified for foreign servers, not per-table.
+       The default is <literal>false</literal>.
+      </para>
+
+      <para>
+       If this option is enabled, <filename>postgres_fdw</filename> prepares
+       transaction on remote server and <productname>PostgreSQL</productname>
+       keeps track of the distributed transaction.
+       <xref linkend="guc-max-prepared-foreign-transactions"/> must be set more
+       than 1 on local server and <xref linkend="guc-max-prepared-transactions"/>
+       must set to more than 1 on remote server.
+      </para>
+     </listitem>
+    </varlistentry>
+
+   </variablelist>
+  </sect3>
  </sect2>
 
  <sect2>
-- 
2.10.5

