Merge pull request #1155 from citusdata/feature/connection_cleanup

Remove parts of old connection / transaction infrastructure

src/backend/distributed/executor/multi_client_executor.c

@@ -19,7 +19,6 @@
 
 #include "commands/dbcommands.h"
 #include "distributed/metadata_cache.h"
-#include "distributed/connection_cache.h"
 #include "distributed/connection_management.h"
 #include "distributed/multi_client_executor.h"
 #include "distributed/multi_server_executor.h"

src/backend/distributed/executor/multi_router_executor.c

@@ -28,8 +28,6 @@
 #include "catalog/pg_type.h"
 #include "distributed/citus_clauses.h"
 #include "distributed/citus_ruleutils.h"
-#include "distributed/commit_protocol.h"
-#include "distributed/connection_cache.h"
 #include "distributed/connection_management.h"
 #include "distributed/deparse_shard_query.h"
 #include "distributed/listutils.h"

src/backend/distributed/executor/multi_utility.c

@@ -32,8 +32,6 @@
 #include "commands/prepare.h"
 #include "distributed/citus_ruleutils.h"
 #include "distributed/colocation_utils.h"
-#include "distributed/commit_protocol.h"
-#include "distributed/connection_cache.h"
 #include "distributed/master_metadata_utility.h"
 #include "distributed/master_protocol.h"
 #include "distributed/metadata_cache.h"

src/backend/distributed/master/master_citus_tools.c

@@ -18,7 +18,6 @@
 
 #include "access/htup_details.h"
 #include "catalog/pg_type.h"
-#include "distributed/connection_cache.h"
 #include "distributed/connection_management.h"
 #include "distributed/metadata_cache.h"
 #include "distributed/multi_server_executor.h"
@@ -222,7 +221,6 @@ ExecuteCommandsInParallelAndStoreResults(StringInfo *nodeNameArray, int *nodePor
 										 int commmandCount)
 {
 	int commandIndex = 0;
-	char *nodeUser = CurrentUserName();
 	PGconn **connectionArray = palloc0(commmandCount * sizeof(PGconn *));
 	int finishedCount = 0;
 
@@ -231,20 +229,25 @@ ExecuteCommandsInParallelAndStoreResults(StringInfo *nodeNameArray, int *nodePor
 	{
 		char *nodeName = nodeNameArray[commandIndex]->data;
 		int nodePort = nodePortArray[commandIndex];
-		PGconn *connection = ConnectToNode(nodeName, nodePort, nodeUser);
+		int connectionFlags = FORCE_NEW_CONNECTION;
+		MultiConnection *multiConnection =
+			GetNodeConnection(connectionFlags, nodeName, nodePort);
+		PGconn *connection = multiConnection->pgConn;
 		StringInfo queryResultString = resultStringArray[commandIndex];
 
 		statusArray[commandIndex] = true;
 
-		connectionArray[commandIndex] = connection;
+		if (PQstatus(multiConnection->pgConn) != CONNECTION_OK)
-
-		if (connection == NULL)
 		{
 			appendStringInfo(queryResultString, "failed to connect to %s:%d", nodeName,
 							 (int) nodePort);
 			statusArray[commandIndex] = false;
 			finishedCount++;
 		}
+		else
+		{
+			connectionArray[commandIndex] = connection;
+		}
 	}
 
 	/* send queries at once */
@@ -491,37 +494,27 @@ static bool
 ExecuteRemoteQueryOrCommand(char *nodeName, uint32 nodePort, char *queryString,
 							StringInfo queryResultString)
 {
-	char *nodeUser = CurrentUserName();
+	int connectionFlags = FORCE_NEW_CONNECTION;
-	PGconn *nodeConnection = ConnectToNode(nodeName, nodePort, nodeUser);
+	MultiConnection *multiConnection =
+		GetNodeConnection(connectionFlags, nodeName, nodePort);
+	PGconn *nodeConnection = multiConnection->pgConn;
 	bool success = false;
+	PGresult *queryResult = NULL;
 
-	if (nodeConnection == NULL)
+	if (PQstatus(multiConnection->pgConn) != CONNECTION_OK)
 	{
 		appendStringInfo(queryResultString, "failed to connect to %s:%d", nodeName,
 						 (int) nodePort);
 		return false;
 	}
 
-	PG_TRY();
+	queryResult = PQexec(nodeConnection, queryString);
-	{
+	success = EvaluateQueryResult(nodeConnection, queryResult, queryResultString);
-		PGresult *queryResult = PQexec(nodeConnection, queryString);
-		success = EvaluateQueryResult(nodeConnection, queryResult, queryResultString);
 
-		PQclear(queryResult);
+	PQclear(queryResult);
 
-		/* close the connection */
+	/* close the connection */
-		CloseConnectionByPGconn(nodeConnection);
+	CloseConnection(multiConnection);
-		nodeConnection = NULL;
-	}
-	PG_CATCH();
-	{
-		StoreErrorMessage(nodeConnection, queryResultString);
-
-		/* close the connection */
-		CloseConnectionByPGconn(nodeConnection);
-		nodeConnection = NULL;
-	}
-	PG_END_TRY();
 
 	return success;
 }

src/backend/distributed/master/master_create_shards.c

@@ -26,7 +26,6 @@
 
 #include "catalog/namespace.h"
 #include "catalog/pg_class.h"
-#include "distributed/connection_cache.h"
 #include "distributed/listutils.h"
 #include "distributed/master_protocol.h"
 #include "distributed/metadata_cache.h"

src/backend/distributed/master/master_modify_multiple_shards.c

@@ -26,8 +26,6 @@
 #include "commands/event_trigger.h"
 #include "distributed/citus_clauses.h"
 #include "distributed/citus_ruleutils.h"
-#include "distributed/commit_protocol.h"
-#include "distributed/connection_cache.h"
 #include "distributed/listutils.h"
 #include "distributed/master_metadata_utility.h"
 #include "distributed/master_protocol.h"

src/backend/distributed/master/master_repair_shards.c

@@ -19,7 +19,7 @@
 
 #include "catalog/pg_class.h"
 #include "distributed/colocation_utils.h"
-#include "distributed/connection_cache.h"
+#include "distributed/connection_management.h"
 #include "distributed/listutils.h"
 #include "distributed/master_protocol.h"
 #include "distributed/metadata_cache.h"

src/backend/distributed/master/master_split_shards.c

@@ -16,7 +16,6 @@
 
 #include "catalog/pg_class.h"
 #include "distributed/colocation_utils.h"
-#include "distributed/connection_cache.h"
 #include "distributed/master_protocol.h"
 #include "distributed/metadata_cache.h"
 #include "distributed/metadata_sync.h"

src/backend/distributed/shared_library_init.c

@@ -20,7 +20,6 @@
 #include "executor/executor.h"
 #include "distributed/citus_nodefuncs.h"
 #include "distributed/connection_management.h"
-#include "distributed/commit_protocol.h"
 #include "distributed/connection_management.h"
 #include "distributed/master_protocol.h"
 #include "distributed/multi_copy.h"

src/backend/distributed/test/connection_cache.c

@@ -1,187 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * test/src/connection_cache.c
- *
- * This file contains functions to exercise Citus's connection hash
- * functionality for purposes of unit testing.
- *
- * Copyright (c) 2014-2016, Citus Data, Inc.
- *
- *-------------------------------------------------------------------------
- */
-
-#include "postgres.h"
-#include "c.h"
-#include "fmgr.h"
-#include "libpq-fe.h"
-
-#include <stddef.h>
-#include <string.h>
-#include <sys/socket.h>
-
-#include "catalog/pg_type.h"
-#include "distributed/connection_cache.h"
-#include "distributed/connection_management.h"
-#include "distributed/metadata_cache.h"
-#include "distributed/test_helper_functions.h" /* IWYU pragma: keep */
-#include "utils/elog.h"
-#include "utils/lsyscache.h"
-
-
-/* declarations for dynamic loading */
-PG_FUNCTION_INFO_V1(initialize_remote_temp_table);
-PG_FUNCTION_INFO_V1(count_remote_temp_table_rows);
-PG_FUNCTION_INFO_V1(get_and_purge_connection);
-PG_FUNCTION_INFO_V1(connect_and_purge_connection);
-PG_FUNCTION_INFO_V1(set_connection_status_bad);
-
-
-/*
- * initialize_remote_temp_table connects to a specified host on a specified
- * port and creates a temporary table with 100 rows. Because the table is
- * temporary, it will be visible if a connection is reused but not if a new
- * connection is opened to the node.
- */
-Datum
-initialize_remote_temp_table(PG_FUNCTION_ARGS)
-{
-	char *nodeName = PG_GETARG_CSTRING(0);
-	int32 nodePort = PG_GETARG_INT32(1);
-	PGresult *result = NULL;
-
-	PGconn *connection = GetOrEstablishConnection(nodeName, nodePort);
-	if (connection == NULL)
-	{
-		PG_RETURN_BOOL(false);
-	}
-
-	result = PQexec(connection, POPULATE_TEMP_TABLE);
-	if (PQresultStatus(result) != PGRES_COMMAND_OK)
-	{
-		WarnRemoteError(connection, result);
-	}
-
-	PQclear(result);
-
-	PG_RETURN_BOOL(true);
-}
-
-
-/*
- * count_remote_temp_table_rows just returns the integer count of rows in the
- * table created by initialize_remote_temp_table. If no such table exists, this
- * function emits a warning and returns -1.
- */
-Datum
-count_remote_temp_table_rows(PG_FUNCTION_ARGS)
-{
-	char *nodeName = PG_GETARG_CSTRING(0);
-	int32 nodePort = PG_GETARG_INT32(1);
-	Datum count = Int32GetDatum(-1);
-	PGresult *result = NULL;
-
-	PGconn *connection = GetOrEstablishConnection(nodeName, nodePort);
-	if (connection == NULL)
-	{
-		PG_RETURN_DATUM(count);
-	}
-
-	result = PQexec(connection, COUNT_TEMP_TABLE);
-	if (PQresultStatus(result) != PGRES_TUPLES_OK)
-	{
-		WarnRemoteError(connection, result);
-	}
-	else
-	{
-		char *countText = PQgetvalue(result, 0, 0);
-		count = StringToDatum(countText, INT4OID);
-	}
-
-	PQclear(result);
-
-	PG_RETURN_DATUM(count);
-}
-
-
-/*
- * get_and_purge_connection first gets a connection using the provided hostname
- * and port before immediately passing that connection to PurgeConnection.
- * Simply a wrapper around PurgeConnection that uses hostname/port rather than
- * PGconn.
- */
-Datum
-get_and_purge_connection(PG_FUNCTION_ARGS)
-{
-	char *nodeName = PG_GETARG_CSTRING(0);
-	int32 nodePort = PG_GETARG_INT32(1);
-
-	PGconn *connection = GetOrEstablishConnection(nodeName, nodePort);
-	if (connection == NULL)
-	{
-		PG_RETURN_BOOL(false);
-	}
-
-	CloseConnectionByPGconn(connection);
-
-	PG_RETURN_BOOL(true);
-}
-
-
-/*
- * get_and_purge_connection first gets a connection using the provided hostname
- * and port before immediately passing that connection to PurgeConnection. This
- * is to test PurgeConnection behvaior when circumventing the cache.
- */
-Datum
-connect_and_purge_connection(PG_FUNCTION_ARGS)
-{
-	char *nodeName = PG_GETARG_CSTRING(0);
-	int32 nodePort = PG_GETARG_INT32(1);
-	char *nodeUser = CurrentUserName();
-
-	PGconn *connection = ConnectToNode(nodeName, nodePort, nodeUser);
-	if (connection == NULL)
-	{
-		PG_RETURN_BOOL(false);
-	}
-
-	CloseConnectionByPGconn(connection);
-
-	PG_RETURN_BOOL(true);
-}
-
-
-/*
- * set_connection_status_bad does not remove the given connection from the connection hash.
- * It simply shuts down the underlying socket. On success, it returns true.
- */
-Datum
-set_connection_status_bad(PG_FUNCTION_ARGS)
-{
-	char *nodeName = PG_GETARG_CSTRING(0);
-	int32 nodePort = PG_GETARG_INT32(1);
-	int socket = -1;
-	int shutdownStatus = 0;
-	int pqStatus PG_USED_FOR_ASSERTS_ONLY = 0;
-
-	PGconn *connection = GetOrEstablishConnection(nodeName, nodePort);
-	if (connection == NULL)
-	{
-		PG_RETURN_BOOL(false);
-	}
-
-	/* Prevent further reads/writes... */
-	socket = PQsocket(connection);
-	shutdownStatus = shutdown(socket, SHUT_RDWR);
-	if (shutdownStatus != 0)
-	{
-		ereport(ERROR, (errcode_for_socket_access(), errmsg("shutdown failed")));
-	}
-
-	/* ... and make libpq notice by reading data. */
-	pqStatus = PQconsumeInput(connection);
-
-	Assert(pqStatus == 0); /* expect failure */
-
-	PG_RETURN_BOOL(true);
-}

src/backend/distributed/transaction/commit_protocol.c

@@ -1,256 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * commit_protocol.c
- *     This file contains functions for managing 1PC or 2PC transactions
- *     across many shard placements.
- *
- * Copyright (c) 2016, Citus Data, Inc.
- *
- *-------------------------------------------------------------------------
- */
-
-
-#include "postgres.h"
-#include "libpq-fe.h"
-#include "miscadmin.h"
-
-#include "distributed/commit_protocol.h"
-#include "distributed/connection_cache.h"
-#include "distributed/master_metadata_utility.h"
-#include "distributed/multi_shard_transaction.h"
-#include "lib/stringinfo.h"
-#include "nodes/pg_list.h"
-
-
-/* Local functions forward declarations */
-static uint32 DistributedTransactionId = 0;
-
-
-/*
- * InitializeDistributedTransaction prepares the distributed transaction ID
- * used in transaction names.
- */
-void
-InitializeDistributedTransaction(void)
-{
-	DistributedTransactionId++;
-}
-
-
-/*
- * PrepareRemoteTransactions prepares all transactions on connections in
- * connectionList for commit if the 2PC commit protocol is enabled.
- * On failure, it reports an error and stops.
- */
-void
-PrepareRemoteTransactions(List *connectionList)
-{
-	ListCell *connectionCell = NULL;
-
-	foreach(connectionCell, connectionList)
-	{
-		TransactionConnection *transactionConnection =
-			(TransactionConnection *) lfirst(connectionCell);
-		PGconn *connection = transactionConnection->connection;
-		int64 connectionId = transactionConnection->connectionId;
-
-		PGresult *result = NULL;
-		StringInfo command = makeStringInfo();
-		StringInfo transactionName = BuildTransactionName(connectionId);
-
-		appendStringInfo(command, "PREPARE TRANSACTION '%s'", transactionName->data);
-
-		result = PQexec(connection, command->data);
-		if (PQresultStatus(result) != PGRES_COMMAND_OK)
-		{
-			/* a failure to prepare is an implicit rollback */
-			transactionConnection->transactionState = TRANSACTION_STATE_CLOSED;
-
-			WarnRemoteError(connection, result);
-			PQclear(result);
-
-			ereport(ERROR, (errcode(ERRCODE_IO_ERROR),
-							errmsg("failed to prepare transaction")));
-		}
-
-		ereport(DEBUG2, (errmsg("sent PREPARE TRANSACTION over connection %ld",
-								connectionId)));
-
-		PQclear(result);
-
-		transactionConnection->transactionState = TRANSACTION_STATE_PREPARED;
-	}
-}
-
-
-/*
- * AbortRemoteTransactions aborts all transactions on connections in connectionList.
- * On failure, it reports a warning and continues to abort all of them.
- */
-void
-AbortRemoteTransactions(List *connectionList)
-{
-	ListCell *connectionCell = NULL;
-
-	foreach(connectionCell, connectionList)
-	{
-		TransactionConnection *transactionConnection =
-			(TransactionConnection *) lfirst(connectionCell);
-		PGconn *connection = transactionConnection->connection;
-		int64 connectionId = transactionConnection->connectionId;
-		PGresult *result = NULL;
-
-		if (transactionConnection->transactionState == TRANSACTION_STATE_PREPARED)
-		{
-			StringInfo command = makeStringInfo();
-			StringInfo transactionName = BuildTransactionName(connectionId);
-
-			appendStringInfo(command, "ROLLBACK PREPARED '%s'", transactionName->data);
-
-			result = PQexec(connection, command->data);
-			if (PQresultStatus(result) != PGRES_COMMAND_OK)
-			{
-				char *nodeName = ConnectionGetOptionValue(connection, "host");
-				char *nodePort = ConnectionGetOptionValue(connection, "port");
-
-				/* log a warning so the user may abort the transaction later */
-				ereport(WARNING, (errmsg("failed to roll back prepared transaction '%s'",
-										 transactionName->data),
-								  errhint("Run \"%s\" on %s:%s",
-										  command->data, nodeName, nodePort)));
-			}
-
-			ereport(DEBUG2, (errmsg("sent ROLLBACK over connection %ld", connectionId)));
-
-			PQclear(result);
-		}
-		else if (transactionConnection->transactionState == TRANSACTION_STATE_OPEN)
-		{
-			/* try to roll back cleanly, if it fails then we won't commit anyway */
-			result = PQexec(connection, "ROLLBACK");
-			PQclear(result);
-		}
-
-		transactionConnection->transactionState = TRANSACTION_STATE_CLOSED;
-	}
-}
-
-
-/*
- * CommitRemoteTransactions commits all transactions on connections in connectionList.
- * If stopOnFailure is true, then CommitRemoteTransactions reports an error on
- * failure, otherwise it reports a warning.
- * Note that if the caller of this function wants the transactions to roll back
- * on a failing commit, stopOnFailure should be used as true. On the other hand,
- * if the caller does not want the transactions to roll back on a failing commit,
- * stopOnFailure should be used as false.
- */
-void
-CommitRemoteTransactions(List *connectionList, bool stopOnFailure)
-{
-	ListCell *connectionCell = NULL;
-
-	foreach(connectionCell, connectionList)
-	{
-		TransactionConnection *transactionConnection =
-			(TransactionConnection *) lfirst(connectionCell);
-		PGconn *connection = transactionConnection->connection;
-		int64 connectionId = transactionConnection->connectionId;
-		PGresult *result = NULL;
-
-		if (transactionConnection->transactionState == TRANSACTION_STATE_PREPARED)
-		{
-			StringInfo command = makeStringInfo();
-			StringInfo transactionName = BuildTransactionName(connectionId);
-
-			/* we shouldn't be committing if any transactions are not prepared */
-			Assert(transactionConnection->transactionState == TRANSACTION_STATE_PREPARED);
-
-			appendStringInfo(command, "COMMIT PREPARED '%s'", transactionName->data);
-
-			result = PQexec(connection, command->data);
-			if (PQresultStatus(result) != PGRES_COMMAND_OK)
-			{
-				char *nodeName = ConnectionGetOptionValue(connection, "host");
-				char *nodePort = ConnectionGetOptionValue(connection, "port");
-
-				/*
-				 * If stopOnFailure is false, log a warning so the user may
-				 * commit the transaction later.
-				 */
-				if (stopOnFailure)
-				{
-					ereport(ERROR, (errmsg("failed to commit prepared transaction '%s'",
-										   transactionName->data),
-									errhint("Run \"%s\" on %s:%s",
-											command->data, nodeName, nodePort)));
-				}
-				else
-				{
-					ereport(WARNING, (errmsg("failed to commit prepared transaction '%s'",
-											 transactionName->data),
-									  errhint("Run \"%s\" on %s:%s",
-											  command->data, nodeName, nodePort)));
-				}
-			}
-
-			ereport(DEBUG2, (errmsg("sent COMMIT PREPARED over connection %ld",
-									connectionId)));
-		}
-		else
-		{
-			/* we shouldn't be committing if any transactions are not open */
-			Assert(transactionConnection->transactionState == TRANSACTION_STATE_OPEN);
-
-			/*
-			 * Try to commit, if it fails and stopOnFailure is false then
-			 * the user might lose data.
-			 */
-			result = PQexec(connection, "COMMIT");
-			if (PQresultStatus(result) != PGRES_COMMAND_OK)
-			{
-				char *nodeName = ConnectionGetOptionValue(connection, "host");
-				char *nodePort = ConnectionGetOptionValue(connection, "port");
-
-				if (stopOnFailure)
-				{
-					ereport(ERROR, (errmsg("failed to commit transaction on %s:%s",
-										   nodeName, nodePort)));
-				}
-				else
-				{
-					ereport(WARNING, (errmsg("failed to commit transaction on %s:%s",
-											 nodeName, nodePort)));
-				}
-			}
-
-			ereport(DEBUG2, (errmsg("sent COMMIT over connection %ld", connectionId)));
-		}
-
-		PQclear(result);
-
-		transactionConnection->transactionState = TRANSACTION_STATE_CLOSED;
-	}
-}
-
-
-/*
- * BuildTransactionName constructs a transaction name that ensures there are no
- * collisions with concurrent transactions by the same master node, subsequent
- * transactions by the same backend, or transactions on a different shard.
- *
- * Collisions may occur over time if transactions fail to commit or abort and
- * are left to linger. This would cause a PREPARE failure for the second
- * transaction, which causes it to be rolled back. In general, the user
- * should ensure that prepared transactions do not linger.
- */
-StringInfo
-BuildTransactionName(int connectionId)
-{
-	StringInfo commandString = makeStringInfo();
-
-	appendStringInfo(commandString, "citus_%d_%u_%d", MyProcPid,
-					 DistributedTransactionId, connectionId);
-
-	return commandString;
-}

src/backend/distributed/transaction/multi_shard_transaction.c

@@ -14,8 +14,6 @@
 #include "postgres.h"
 
 #include "distributed/colocation_utils.h"
-#include "distributed/commit_protocol.h"
-#include "distributed/connection_cache.h"
 #include "distributed/connection_management.h"
 #include "distributed/master_metadata_utility.h"
 #include "distributed/metadata_cache.h"
@@ -250,22 +248,3 @@ ResetShardPlacementTransactionState(void)
 		SavedMultiShardCommitProtocol = COMMIT_PROTOCOL_BARE;
 	}
 }
-
-
-/*
- * CloseConnections closes all connections in connectionList.
- */
-void
-CloseConnections(List *connectionList)
-{
-	ListCell *connectionCell = NULL;
-
-	foreach(connectionCell, connectionList)
-	{
-		TransactionConnection *transactionConnection =
-			(TransactionConnection *) lfirst(connectionCell);
-		PGconn *connection = transactionConnection->connection;
-
-		CloseConnectionByPGconn(connection);
-	}
-}

src/backend/distributed/transaction/worker_transaction.c

@@ -13,13 +13,12 @@
 
 #include "postgres.h"
 #include "miscadmin.h"
+#include "libpq-fe.h"
 
 #include <sys/stat.h>
 #include <unistd.h>
 
 #include "access/xact.h"
-#include "distributed/commit_protocol.h"
-#include "distributed/connection_cache.h"
 #include "distributed/connection_management.h"
 #include "distributed/metadata_cache.h"
 #include "distributed/multi_shard_transaction.h"

src/backend/distributed/utils/connection_cache.c

@@ -1,268 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * connection_cache.c
- *
- * Legacy connection caching layer. Will be removed entirely.
- *
- * Copyright (c) 2014-2016, Citus Data, Inc.
- *
- *-------------------------------------------------------------------------
- */
-
-#include "postgres.h" /* IWYU pragma: keep */
-#include "c.h"
-#include "libpq-fe.h"
-#include "miscadmin.h"
-
-#include <stddef.h>
-#include <stdio.h>
-#include <string.h>
-
-#include "commands/dbcommands.h"
-#include "distributed/connection_management.h"
-#include "distributed/connection_cache.h"
-#include "distributed/metadata_cache.h"
-#include "distributed/remote_commands.h"
-#include "mb/pg_wchar.h"
-#include "utils/builtins.h"
-#include "utils/elog.h"
-#include "utils/errcodes.h"
-#include "utils/hsearch.h"
-#include "utils/memutils.h"
-#include "utils/palloc.h"
-
-
-/* local function forward declarations */
-static void ReportRemoteError(PGconn *connection, PGresult *result, bool raiseError);
-
-
-/*
- * GetOrEstablishConnection returns a PGconn which can be used to execute
- * queries on a remote PostgreSQL server. If no suitable connection to the
- * specified node on the specified port yet exists, the function establishes
- * a new connection and adds it to the connection cache before returning it.
- *
- * Returned connections are guaranteed to be in the CONNECTION_OK state. If the
- * requested connection cannot be established, or if it was previously created
- * but is now in an unrecoverable bad state, this function returns NULL.
- *
- * This function throws an error if a hostname over 255 characters is provided.
- */
-PGconn *
-GetOrEstablishConnection(char *nodeName, int32 nodePort)
-{
-	int connectionFlags = SESSION_LIFESPAN;
-	PGconn *connection = NULL;
-	MultiConnection *multiConnection =
-		GetNodeConnection(connectionFlags, nodeName, nodePort);
-
-	if (PQstatus(multiConnection->pgConn) == CONNECTION_OK)
-	{
-		connection = multiConnection->pgConn;
-	}
-	else
-	{
-		ReportConnectionError(multiConnection, WARNING);
-		CloseConnection(multiConnection);
-		connection = NULL;
-	}
-
-	return connection;
-}
-
-
-/*
- * Utility method to simplify populating a connection cache key with relevant
- * fields from a provided connection.
- */
-void
-BuildKeyForConnection(PGconn *connection, NodeConnectionKey *connectionKey)
-{
-	char *nodeNameString = NULL;
-	char *nodePortString = NULL;
-	char *nodeUserString = NULL;
-
-	nodeNameString = ConnectionGetOptionValue(connection, "host");
-	if (nodeNameString == NULL)
-	{
-		ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						errmsg("connection is missing host option")));
-	}
-
-	nodePortString = ConnectionGetOptionValue(connection, "port");
-	if (nodePortString == NULL)
-	{
-		ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						errmsg("connection is missing port option")));
-	}
-
-	nodeUserString = ConnectionGetOptionValue(connection, "user");
-	if (nodeUserString == NULL)
-	{
-		ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						errmsg("connection is missing user option")));
-	}
-
-	MemSet(connectionKey, 0, sizeof(NodeConnectionKey));
-	strlcpy(connectionKey->nodeName, nodeNameString, MAX_NODE_LENGTH + 1);
-	connectionKey->nodePort = pg_atoi(nodePortString, sizeof(int32), 0);
-	strlcpy(connectionKey->nodeUser, nodeUserString, NAMEDATALEN);
-
-	pfree(nodeNameString);
-	pfree(nodePortString);
-	pfree(nodeUserString);
-}
-
-
-/*
- * WarnRemoteError retrieves error fields from a remote result and produces an
- * error report at the WARNING level after amending the error with a CONTEXT
- * field containing the remote node host and port information.
- */
-void
-WarnRemoteError(PGconn *connection, PGresult *result)
-{
-	ReportRemoteError(connection, result, false);
-}
-
-
-/*
- * ReraiseRemoteError retrieves error fields from a remote result and re-raises
- * the error after amending it with a CONTEXT field containing the remote node
- * host and port information.
- */
-void
-ReraiseRemoteError(PGconn *connection, PGresult *result)
-{
-	ReportRemoteError(connection, result, true);
-}
-
-
-/*
- * ReportRemoteError is an internal helper function which implements logic
- * needed by both WarnRemoteError and ReraiseRemoteError. They wrap this
- * function to provide explicit names for the possible behaviors.
- */
-static void
-ReportRemoteError(PGconn *connection, PGresult *result, bool raiseError)
-{
-	char *sqlStateString = PQresultErrorField(result, PG_DIAG_SQLSTATE);
-	char *messagePrimary = PQresultErrorField(result, PG_DIAG_MESSAGE_PRIMARY);
-	char *messageDetail = PQresultErrorField(result, PG_DIAG_MESSAGE_DETAIL);
-	char *messageHint = PQresultErrorField(result, PG_DIAG_MESSAGE_HINT);
-	char *messageContext = PQresultErrorField(result, PG_DIAG_CONTEXT);
-
-	char *nodeName = ConnectionGetOptionValue(connection, "host");
-	char *nodePort = ConnectionGetOptionValue(connection, "port");
-	int sqlState = ERRCODE_CONNECTION_FAILURE;
-	int errorLevel = WARNING;
-
-	if (sqlStateString != NULL)
-	{
-		sqlState = MAKE_SQLSTATE(sqlStateString[0], sqlStateString[1], sqlStateString[2],
-								 sqlStateString[3], sqlStateString[4]);
-	}
-
-	/*
-	 * If the PGresult did not contain a message, the connection may provide a
-	 * suitable top level one. At worst, this is an empty string.
-	 */
-	if (messagePrimary == NULL)
-	{
-		char *lastNewlineIndex = NULL;
-
-		messagePrimary = PQerrorMessage(connection);
-		lastNewlineIndex = strrchr(messagePrimary, '\n');
-
-		/* trim trailing newline, if any */
-		if (lastNewlineIndex != NULL)
-		{
-			*lastNewlineIndex = '\0';
-		}
-	}
-
-	/*
-	 * If requested, actually raise an error.
-	 */
-	if (raiseError)
-	{
-		errorLevel = ERROR;
-	}
-
-	if (sqlState == ERRCODE_CONNECTION_FAILURE)
-	{
-		ereport(errorLevel, (errcode(sqlState),
-							 errmsg("connection failed to %s:%s", nodeName, nodePort),
-							 errdetail("%s", messagePrimary)));
-	}
-	else
-	{
-		ereport(errorLevel, (errcode(sqlState), errmsg("%s", messagePrimary),
-							 messageDetail ? errdetail("%s", messageDetail) : 0,
-							 messageHint ? errhint("%s", messageHint) : 0,
-							 messageContext ? errcontext("%s", messageContext) : 0,
-							 errcontext("while executing command on %s:%s",
-										nodeName, nodePort)));
-	}
-}
-
-
-/*
- * ConnectToNode opens a connection to a remote PostgreSQL server. The function
- * configures the connection's fallback application name to 'citus' and sets
- * the remote encoding to match the local one.  All parameters are required to
- * be non NULL.
- *
- * This is only a thin layer over connection_management.[ch], and will be
- * removed soon.
- */
-PGconn *
-ConnectToNode(char *nodeName, int32 nodePort, char *nodeUser)
-{
-	/* don't want already established connections */
-	int connectionFlags = FORCE_NEW_CONNECTION;
-	PGconn *connection = NULL;
-	MultiConnection *multiConnection =
-		GetNodeUserDatabaseConnection(connectionFlags, nodeName, nodePort, nodeUser,
-									  NULL);
-
-	if (PQstatus(multiConnection->pgConn) == CONNECTION_OK)
-	{
-		connection = multiConnection->pgConn;
-	}
-	else
-	{
-		ReportConnectionError(multiConnection, WARNING);
-		CloseConnection(multiConnection);
-		connection = NULL;
-	}
-
-	return connection;
-}
-
-
-/*
- * ConnectionGetOptionValue inspects the provided connection for an option with
- * a given keyword and returns a new palloc'd string with that options's value.
- * The function returns NULL if the connection has no setting for an option with
- * the provided keyword.
- */
-char *
-ConnectionGetOptionValue(PGconn *connection, char *optionKeyword)
-{
-	char *optionValue = NULL;
-	PQconninfoOption *conninfoOptions = PQconninfo(connection);
-	PQconninfoOption *option = NULL;
-
-	for (option = conninfoOptions; option->keyword != NULL; option++)
-	{
-		if (strncmp(option->keyword, optionKeyword, NAMEDATALEN) == 0)
-		{
-			optionValue = pstrdup(option->val);
-		}
-	}
-
-	PQconninfoFree(conninfoOptions);
-
-	return optionValue;
-}

src/include/distributed/commit_protocol.h

@@ -1,55 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * commit_protocol.h
- *	  Type and function declarations used in performing transactions across
- *	  shard placements.
- *
- * Copyright (c) 2016, Citus Data, Inc.
- *
- *-------------------------------------------------------------------------
- */
-
-#ifndef COMMIT_PROTOCOL_H
-#define COMMIT_PROTOCOL_H
-
-
-#include "access/xact.h"
-#include "distributed/connection_management.h"
-#include "libpq-fe.h"
-#include "lib/stringinfo.h"
-#include "nodes/pg_list.h"
-
-
-/* Enumeration that defines different remote transaction states */
-typedef enum
-{
-	TRANSACTION_STATE_INVALID = 0,
-	TRANSACTION_STATE_OPEN,
-	TRANSACTION_STATE_COPY_STARTED,
-	TRANSACTION_STATE_PREPARED,
-	TRANSACTION_STATE_CLOSED
-} TransactionState;
-
-/*
- * TransactionConnection represents a connection to a remote node which is
- * used to perform a transaction on shard placements.
- */
-typedef struct TransactionConnection
-{
-	int groupId;
-	int64 connectionId;
-	TransactionState transactionState;
-	PGconn *connection;
-	const char *nodeName;
-	int nodePort;
-} TransactionConnection;
-
-
-/* Functions declarations for transaction and connection management */
-extern void InitializeDistributedTransaction(void);
-extern void PrepareRemoteTransactions(List *connectionList);
-extern void AbortRemoteTransactions(List *connectionList);
-extern void CommitRemoteTransactions(List *connectionList, bool stopOnFailure);
-extern StringInfo BuildTransactionName(int connectionId);
-
-#endif /* COMMIT_PROTOCOL_H */

src/include/distributed/connection_cache.h

@@ -1,51 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * connection_cache.h
- *
- * Declarations for public functions and types related to connection hash
- * functionality.
- *
- * Copyright (c) 2014-2016, Citus Data, Inc.
- *
- *-------------------------------------------------------------------------
- */
-
-#ifndef CONNECTION_CACHE_H
-#define CONNECTION_CACHE_H
-
-#include "c.h"
-#include "libpq-fe.h"
-
-#include "utils/hsearch.h"
-#include "distributed/connection_management.h"
-
-/*
- * NodeConnectionKey acts as the key to index into the (process-local) hash
- * keeping track of open connections. Node name and port are sufficient.
- */
-typedef struct NodeConnectionKey
-{
-	char nodeName[MAX_NODE_LENGTH + 1]; /* hostname of host to connect to */
-	int32 nodePort;                     /* port of host to connect to */
-	char nodeUser[NAMEDATALEN + 1];     /* user name to connect as */
-} NodeConnectionKey;
-
-
-/* NodeConnectionEntry keeps track of connections themselves. */
-typedef struct NodeConnectionEntry
-{
-	NodeConnectionKey cacheKey; /* hash entry key */
-	MultiConnection *connection; /* connection to remote server, if any */
-} NodeConnectionEntry;
-
-
-/* function declarations for obtaining and using a connection */
-extern PGconn * GetOrEstablishConnection(char *nodeName, int32 nodePort);
-extern void BuildKeyForConnection(PGconn *connection, NodeConnectionKey *connectionKey);
-extern void WarnRemoteError(PGconn *connection, PGresult *result);
-extern void ReraiseRemoteError(PGconn *connection, PGresult *result);
-extern PGconn * ConnectToNode(char *nodeName, int nodePort, char *nodeUser);
-extern char * ConnectionGetOptionValue(PGconn *connection, char *optionKeyword);
-
-
-#endif /* CONNECTION_CACHE_H */

src/include/distributed/multi_shard_transaction.h

@@ -33,7 +33,6 @@ extern ShardConnections * GetShardConnections(int64 shardId, bool *shardConnecti
 extern ShardConnections * GetShardHashConnections(HTAB *connectionHash, int64 shardId,
 												  bool *connectionsFound);
 extern List * ShardConnectionList(HTAB *connectionHash);
-extern void CloseConnections(List *connectionList);
 extern void ResetShardPlacementTransactionState(void);
 
 

src/include/distributed/remote_commands.h

@@ -39,7 +39,7 @@ extern void ExecuteCriticalRemoteCommand(MultiConnection *connection,
 										 const char *command);
 extern int ExecuteOptionalRemoteCommand(MultiConnection *connection,
 										const char *command,
-										PGresult **result);
+										struct pg_result **result);
 extern int SendRemoteCommand(MultiConnection *connection, const char *command);
 extern int SendRemoteCommandParams(MultiConnection *connection, const char *command,
 								   int parameterCount, const Oid *parameterTypes,

src/test/regress/expected/multi_connection_cache.out

@@ -1,118 +0,0 @@
-ALTER SEQUENCE pg_catalog.pg_dist_shardid_seq RESTART 410000;
-ALTER SEQUENCE pg_catalog.pg_dist_jobid_seq RESTART 410000;
--- ===================================================================
--- create test functions
--- ===================================================================
-CREATE FUNCTION initialize_remote_temp_table(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
-CREATE FUNCTION count_remote_temp_table_rows(cstring, integer)
-	RETURNS integer
-	AS 'citus'
-	LANGUAGE C STRICT;
-CREATE FUNCTION get_and_purge_connection(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
-CREATE FUNCTION connect_and_purge_connection(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
-CREATE FUNCTION set_connection_status_bad(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
--- ===================================================================
--- test connection hash functionality
--- ===================================================================
--- worker port number is set in pg_regress_multi.pl
-\set worker_port 57638
--- reduce verbosity to squelch chatty warnings
-\set VERBOSITY terse
--- connect to non-existent host
-SELECT initialize_remote_temp_table('dummy-host-name', 12345);
-WARNING:  connection error: dummy-host-name:12345
- initialize_remote_temp_table 
-------------------------------
- f
-(1 row)
-
-\set VERBOSITY default
--- try to use hostname over 255 characters
-SELECT initialize_remote_temp_table(repeat('a', 256)::cstring, :worker_port);
-ERROR:  hostname exceeds the maximum length of 255
--- connect to localhost and build a temp table
-SELECT initialize_remote_temp_table('localhost', :worker_port);
- initialize_remote_temp_table 
-------------------------------
- t
-(1 row)
-
--- table should still be visible since session is reused
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
- count_remote_temp_table_rows 
-------------------------------
-                          100
-(1 row)
-
--- purge existing connection to localhost
-SELECT get_and_purge_connection('localhost', :worker_port);
- get_and_purge_connection 
---------------------------
- t
-(1 row)
-
--- squelch WARNINGs that contain worker_port
-SET client_min_messages TO ERROR;
--- should not be able to see table anymore
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
- count_remote_temp_table_rows 
-------------------------------
-                           -1
-(1 row)
-
--- recreate once more
-SELECT initialize_remote_temp_table('localhost', :worker_port);
- initialize_remote_temp_table 
-------------------------------
- t
-(1 row)
-
--- set the connection status to bad
-SELECT set_connection_status_bad('localhost', :worker_port);
- set_connection_status_bad 
----------------------------
- t
-(1 row)
-
--- should get connection failure (cached connection bad)
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
- count_remote_temp_table_rows 
-------------------------------
-                           -1
-(1 row)
-
--- should get result failure (reconnected, so no temp table)
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
- count_remote_temp_table_rows 
-------------------------------
-                           -1
-(1 row)
-
--- purge the connection so that we clean up the bad connection
-SELECT get_and_purge_connection('localhost', :worker_port);
- get_and_purge_connection 
---------------------------
- t
-(1 row)
-
-SET client_min_messages TO DEFAULT;
-\c
--- purge existing connection to localhost
-SELECT connect_and_purge_connection('localhost', :worker_port);
- connect_and_purge_connection 
-------------------------------
- t
-(1 row)
-

src/test/regress/multi_schedule

@@ -126,7 +126,7 @@ test: multi_outer_join
 # is independed from the rest of the group, it is added to increase parallelism.
 # ---
 test: multi_create_fdw
-test: multi_connection_cache multi_complex_count_distinct
+test: multi_complex_count_distinct
 test: multi_distribution_metadata
 test: multi_generate_ddl_commands
 test: multi_create_shards

src/test/regress/multi_task_tracker_extra_schedule

@@ -81,7 +81,6 @@ test: multi_outer_join
 # Note that the order of the following tests are important.
 # ---
 test: multi_create_fdw
-test: multi_connection_cache
 test: multi_distribution_metadata
 test: multi_generate_ddl_commands
 test: multi_create_shards

src/test/regress/sql/multi_connection_cache.sql

@@ -1,88 +0,0 @@
-
-ALTER SEQUENCE pg_catalog.pg_dist_shardid_seq RESTART 410000;
-ALTER SEQUENCE pg_catalog.pg_dist_jobid_seq RESTART 410000;
-
-
--- ===================================================================
--- create test functions
--- ===================================================================
-
-CREATE FUNCTION initialize_remote_temp_table(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
-
-CREATE FUNCTION count_remote_temp_table_rows(cstring, integer)
-	RETURNS integer
-	AS 'citus'
-	LANGUAGE C STRICT;
-
-CREATE FUNCTION get_and_purge_connection(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
-
-CREATE FUNCTION connect_and_purge_connection(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
-
-CREATE FUNCTION set_connection_status_bad(cstring, integer)
-	RETURNS bool
-	AS 'citus'
-	LANGUAGE C STRICT;
-
--- ===================================================================
--- test connection hash functionality
--- ===================================================================
-
--- worker port number is set in pg_regress_multi.pl
-\set worker_port 57638
-
--- reduce verbosity to squelch chatty warnings
-\set VERBOSITY terse
-
--- connect to non-existent host
-SELECT initialize_remote_temp_table('dummy-host-name', 12345);
-
-\set VERBOSITY default
-
--- try to use hostname over 255 characters
-SELECT initialize_remote_temp_table(repeat('a', 256)::cstring, :worker_port);
-
--- connect to localhost and build a temp table
-SELECT initialize_remote_temp_table('localhost', :worker_port);
-
--- table should still be visible since session is reused
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
-
--- purge existing connection to localhost
-SELECT get_and_purge_connection('localhost', :worker_port);
-
--- squelch WARNINGs that contain worker_port
-SET client_min_messages TO ERROR;
-
--- should not be able to see table anymore
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
-
--- recreate once more
-SELECT initialize_remote_temp_table('localhost', :worker_port);
-
--- set the connection status to bad
-SELECT set_connection_status_bad('localhost', :worker_port);
-
--- should get connection failure (cached connection bad)
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
-
--- should get result failure (reconnected, so no temp table)
-SELECT count_remote_temp_table_rows('localhost', :worker_port);
-
--- purge the connection so that we clean up the bad connection
-SELECT get_and_purge_connection('localhost', :worker_port);
-
-SET client_min_messages TO DEFAULT;
-
-\c
-
--- purge existing connection to localhost
-SELECT connect_and_purge_connection('localhost', :worker_port);