v40-0001-Reworking-ParallelSlots-for-mutliple-DB-use.patch

From 8c8fd04f48ff80cfbc7331a8a819555cab38f27b Mon Sep 17 00:00:00 2001
From: Mark Dilger <mark.dilger@enterprisedb.com>
Date: Mon, 22 Feb 2021 08:55:56 -0800
Subject: [PATCH v40 1/3] Reworking ParallelSlots for mutliple DB use

The existing implementation of ParallelSlots is used by reindexdb
and vacuumdb to process tables in parallel in only one database at
a time.  The ParallelSlots interface reflects this usage pattern.
The function to set up the slots assumes all slots should be
connected to the same database, and the function for getting the
next idle slot pays no attention to which database the slot may be
connected to.

In anticipation of pg_amcheck using parallel slots to process
multiple databases in parallel, reworking the interface while
trying to remain reasonably simple for reindexdb and vacuumdb to
use:

ParallelSlotsSetup() is replaced by two functions,
ParallelSlotsSetupOneDB() and ParallelSlotsSetupMinimal().  The
former establishes database connections for all slots much as the
old ParallelSlotsSetup() did, and the latter delays connecting to
databases until a slot is requested.

ParallelSlotsGetIdle() is extended to take arguments about the
database connection desired and to manage a heterogeneous set of
slots potentially containing slots connected to varying databases
and some slots not yet connected.  The function will reuse an
existing connection or form a new connection as necessary.

The logic for determining whether a slot's connection is suitable
for reuse is unfortunately a little more complicated than I was
hoping, using a ConnParams struct to identify the desired database
rather than a simple database name.

For callers like reindexdb and vacuumdb, they pass NULL, and any
existing connection is considered suitable.  This matches their
historical behavior and is simple.

For callers like pg_amcheck, they pass the cparams they want used to
open a new connection (if necessary) or to select an existing
connection if one matches.  Byte-for-byte equality between the
connection parameter strings is used to determine if an existing
connection is suitable to satisfy a slot request.  In practice,
there are multiple ways to format connection parameter strings that
would result in the same database/host/port/user, but the
implementation does not attempt to determine ConnParams equivalence
beyond bytewise equality.
---
 src/bin/scripts/reindexdb.c          |   5 +-
 src/bin/scripts/vacuumdb.c           |  35 +-
 src/fe_utils/parallel_slot.c         | 506 +++++++++++++++++++++------
 src/include/fe_utils/parallel_slot.h |  18 +-
 src/tools/pgindent/typedefs.list     |   1 +
 5 files changed, 427 insertions(+), 138 deletions(-)

diff --git a/src/bin/scripts/reindexdb.c b/src/bin/scripts/reindexdb.c
index 9f072ac49a..712102f521 100644
--- a/src/bin/scripts/reindexdb.c
+++ b/src/bin/scripts/reindexdb.c
@@ -445,7 +445,8 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
 
 	Assert(process_list != NULL);
 
-	slots = ParallelSlotsSetup(cparams, progname, echo, conn, concurrentCons);
+	slots = ParallelSlotsSetupOneDB(cparams, progname, echo, conn,
+									concurrentCons, NULL);
 
 	cell = process_list->head;
 	do
@@ -459,7 +460,7 @@ reindex_one_database(const ConnParams *cparams, ReindexType type,
 			goto finish;
 		}
 
-		free_slot = ParallelSlotsGetIdle(slots, concurrentCons);
+		free_slot = ParallelSlotsGetIdle(slots, concurrentCons, cparams, progname, echo, NULL);
 		if (!free_slot)
 		{
 			failed = true;
diff --git a/src/bin/scripts/vacuumdb.c b/src/bin/scripts/vacuumdb.c
index 602fd45c42..ddd104aa25 100644
--- a/src/bin/scripts/vacuumdb.c
+++ b/src/bin/scripts/vacuumdb.c
@@ -428,6 +428,7 @@ vacuum_one_database(const ConnParams *cparams,
 	bool		failed = false;
 	bool		tables_listed = false;
 	bool		has_where = false;
+	const char *initcmd;
 	const char *stage_commands[] = {
 		"SET default_statistics_target=1; SET vacuum_cost_delay=0;",
 		"SET default_statistics_target=10; RESET vacuum_cost_delay;",
@@ -684,26 +685,25 @@ vacuum_one_database(const ConnParams *cparams,
 		concurrentCons = 1;
 
 	/*
-	 * Setup the database connections. We reuse the connection we already have
-	 * for the first slot.  If not in parallel mode, the first slot in the
-	 * array contains the connection.
+	 * All slots need to be prepared to run the appropriate analyze stage, if
+	 * caller requested that mode.  We have to prepare the initial connection
+	 * ourselves before setting up the slots.
 	 */
-	slots = ParallelSlotsSetup(cparams, progname, echo, conn, concurrentCons);
+	if (stage == ANALYZE_NO_STAGE)
+		initcmd = NULL;
+	else
+	{
+		initcmd = stage_commands[stage];
+		executeCommand(conn, initcmd, echo);
+	}
 
 	/*
-	 * Prepare all the connections to run the appropriate analyze stage, if
-	 * caller requested that mode.
+	 * Setup the database connections. We reuse the connection we already have
+	 * for the first slot.  If not in parallel mode, the first slot in the
+	 * array contains the connection.
 	 */
-	if (stage != ANALYZE_NO_STAGE)
-	{
-		int			j;
-
-		/* We already emitted the message above */
-
-		for (j = 0; j < concurrentCons; j++)
-			executeCommand((slots + j)->connection,
-						   stage_commands[stage], echo);
-	}
+	slots = ParallelSlotsSetupOneDB(cparams, progname, echo, conn,
+									concurrentCons, initcmd);
 
 	initPQExpBuffer(&sql);
 
@@ -719,7 +719,8 @@ vacuum_one_database(const ConnParams *cparams,
 			goto finish;
 		}
 
-		free_slot = ParallelSlotsGetIdle(slots, concurrentCons);
+		free_slot = ParallelSlotsGetIdle(slots, concurrentCons, cparams,
+										 progname, echo, initcmd);
 		if (!free_slot)
 		{
 			failed = true;
diff --git a/src/fe_utils/parallel_slot.c b/src/fe_utils/parallel_slot.c
index b625deb254..e71104ff78 100644
--- a/src/fe_utils/parallel_slot.c
+++ b/src/fe_utils/parallel_slot.c
@@ -25,20 +25,91 @@
 #include "common/logging.h"
 #include "fe_utils/cancel.h"
 #include "fe_utils/parallel_slot.h"
+#include "fe_utils/query_utils.h"
 
 #define ERRCODE_UNDEFINED_TABLE  "42P01"
 
-static void init_slot(ParallelSlot *slot, PGconn *conn);
 static int	select_loop(int maxFd, fd_set *workerset);
 static bool processQueryResult(ParallelSlot *slot, PGresult *result);
 
+/*
+ * Copy the fields from a source ConnParams struct without sharing any state
+ * that would allow changes to the source to affect the copy.
+ */
 static void
-init_slot(ParallelSlot *slot, PGconn *conn)
+deep_copy_cparams(ConnParams *dst, const ConnParams *src)
 {
-	slot->connection = conn;
-	/* Initially assume connection is idle */
-	slot->isFree = true;
-	ParallelSlotClearHandler(slot);
+	memset(dst, 0, sizeof(*dst));
+	if (src == NULL)
+		return;
+	if (src->dbname)
+		dst->dbname = pstrdup(src->dbname);
+	if (src->pghost)
+		dst->pghost = pstrdup(src->pghost);
+	if (src->pgport)
+		dst->pgport = pstrdup(src->pgport);
+	if (src->pguser)
+		dst->pguser = pstrdup(src->pguser);
+	dst->prompt_password = src->prompt_password;
+	if (src->override_dbname)
+		dst->override_dbname = pstrdup(src->override_dbname);
+}
+
+/*
+ * Free a ConnParams struct that was made using deep_copy_cparams.  Beware that
+ * ConnParams structs often contain const data from the command line.  This
+ * function must only be used on copies where we own the data to be freed.
+ */
+static void
+free_cparams_copy(ConnParams *copy)
+{
+	/* We need to cast away const before freeing each field. */
+	if (copy->dbname)
+		pfree((void *) copy->dbname);
+	if (copy->pghost)
+		pfree((void *) copy->pghost);
+	if (copy->pgport)
+		pfree((void *) copy->pgport);
+	if (copy->pguser)
+		pfree((void *) copy->pguser);
+	if (copy->override_dbname)
+		pfree((void *) copy->override_dbname);
+}
+
+/* Macro for comparing string fields that might be NULL */
+#define equalstr(a, b)	\
+	(((a) != NULL && (b) != NULL) ? (strcmp(a, b) == 0) : (a) == (b))
+
+/* Compare a field that is a pointer to a C string, or perhaps NULL */
+#define COMPARE_STRING_FIELD(fldname) \
+	do { \
+		if (!equalstr(a->fldname, b->fldname)) \
+			return false; \
+	} while (0)
+
+/* Compare a simple scalar field (int, float, bool, enum, etc) */
+#define COMPARE_SCALAR_FIELD(fldname) \
+	do { \
+		if (a->fldname != b->fldname) \
+			return false; \
+	} while (0)
+
+/*
+ * Return whether two ConnParams structs contain equal parameters.
+ */
+static bool
+equalConnParams(const ConnParams *a, const ConnParams *b)
+{
+	Assert(a != NULL);
+	Assert(b != NULL);
+
+	COMPARE_STRING_FIELD(dbname);
+	COMPARE_STRING_FIELD(pghost);
+	COMPARE_STRING_FIELD(pgport);
+	COMPARE_STRING_FIELD(pguser);
+	COMPARE_SCALAR_FIELD(prompt_password);
+	COMPARE_STRING_FIELD(override_dbname);
+	return true;
 }
 
 /*
@@ -50,6 +121,7 @@ static bool
 processQueryResult(ParallelSlot *slot, PGresult *result)
 {
 	Assert(slot->handler != NULL);
+	Assert(slot->connection != NULL);
 
 	/* On failure, the handler should return NULL after freeing the result */
 	if (!slot->handler(result, slot->connection, slot->handler_context))
@@ -71,6 +143,9 @@ consumeQueryResult(ParallelSlot *slot)
 	bool		ok = true;
 	PGresult   *result;
 
+	Assert(slot != NULL);
+	Assert(slot->connection != NULL);
+
 	SetCancelConn(slot->connection);
 	while ((result = PQgetResult(slot->connection)) != NULL)
 	{
@@ -82,10 +157,9 @@ consumeQueryResult(ParallelSlot *slot)
 }
 
 /*
- * Wait until a file descriptor from the given set becomes readable.
- *
- * Returns the number of ready descriptors, or -1 on failure (including
- * getting a cancel request).
+ * Wait until a file descriptor from the given set becomes readable.  Returns
+ * the number of ready descriptors, or -1 on failure (including getting a
+ * cancel request).
  */
 static int
 select_loop(int maxFd, fd_set *workerset)
@@ -137,153 +211,352 @@ select_loop(int maxFd, fd_set *workerset)
 }
 
 /*
- * ParallelSlotsGetIdle
- *		Return a connection slot that is ready to execute a command.
- *
- * This returns the first slot we find that is marked isFree, if one is;
- * otherwise, we loop on select() until one socket becomes available.  When
- * this happens, we read the whole set and mark as free all sockets that
- * become available.  If an error occurs, NULL is returned.
+ * Return the offset of a suitable idle slot, or -1 if none are available.  If
+ * the given connection parameters are not null, only idle slots connected
+ * using equivalent parameters are considered suitable, otherwise all idle
+ * connected slots are considered suitable.
  */
-ParallelSlot *
-ParallelSlotsGetIdle(ParallelSlot *slots, int numslots)
+static int
+find_matching_idle_slot(const ParallelSlot *slots, int numslots,
+						const ConnParams *cparams)
 {
 	int			i;
-	int			firstFree = -1;
 
-	/*
-	 * Look for any connection currently free.  If there is one, mark it as
-	 * taken and let the caller know the slot to use.
-	 */
+	Assert(slots != NULL);
+
 	for (i = 0; i < numslots; i++)
 	{
-		if (slots[i].isFree)
-		{
-			slots[i].isFree = false;
-			return slots + i;
-		}
+		if (slots[i].inUse)
+			continue;
+
+		if (slots[i].connection == NULL)
+			continue;
+
+		if (cparams == NULL || equalConnParams(&slots[i].cparams, cparams))
+			return i;
+	}
+	return -1;
+}
+
+/*
+ * Return the offset of the first slot without a database connection, or -1 if
+ * all slots are connected.
+ */
+static int
+find_unconnected_slot(const ParallelSlot *slots, int numslots)
+{
+	int			i;
+
+	Assert(slots != NULL);
+
+	for (i = 0; i < numslots; i++)
+	{
+		if (slots[i].inUse)
+			continue;
+
+		if (slots[i].connection == NULL)
+			return i;
+	}
+
+	return -1;
+}
+
+/*
+ * Return the offset of the first idle slot, or -1 if all slots are busy.
+ */
+static int
+find_any_idle_slot(const ParallelSlot *slots, int numslots)
+{
+	int			i;
+
+	Assert(slots != NULL);
+
+	for (i = 0; i < numslots; i++)
+		if (!slots[i].inUse)
+			return i;
+
+	return -1;
+}
+
+/*
+ * Wait for any slot's connection to have query results, consume the results,
+ * and update the slot's status as appropriate.  Returns true on success,
+ * false on cancellation, on error, or if no slots are connected.
+ */
+static bool
+wait_on_slots(ParallelSlot *slots, int numslots, const char *progname)
+{
+	int			i;
+	fd_set		slotset;
+	int			maxFd = 0;
+	PGconn	   *cancelconn = NULL;
+
+	Assert(slots != NULL);
+	Assert(progname != NULL);
+
+	/* We must reconstruct the fd_set for each call to select_loop */
+	FD_ZERO(&slotset);
+
+	for (i = 0; i < numslots; i++)
+	{
+		int			sock;
+
+		/* We shouldn't get here if we still have slots without connections */
+		Assert(slots[i].connection != NULL);
+
+		sock = PQsocket(slots[i].connection);
+
+		/*
+		 * We don't really expect any connections to lose their sockets after
+		 * startup, but just in case, cope by ignoring them.
+		 */
+		if (sock < 0)
+			continue;
+
+		/* Keep track of the first valid connection we see. */
+		if (cancelconn == NULL)
+			cancelconn = slots[i].connection;
+
+		FD_SET(sock, &slotset);
+		if (sock > maxFd)
+			maxFd = sock;
 	}
 
 	/*
-	 * No free slot found, so wait until one of the connections has finished
-	 * its task and return the available slot.
+	 * If we get this far with no valid connections, processing cannot
+	 * continue.
 	 */
-	while (firstFree < 0)
+	if (cancelconn == NULL)
+		return false;
+
+	SetCancelConn(slots->connection);
+	i = select_loop(maxFd, &slotset);
+	ResetCancelConn();
+
+	/* failure? */
+	if (i < 0)
+		return false;
+
+	for (i = 0; i < numslots; i++)
 	{
-		fd_set		slotset;
-		int			maxFd = 0;
+		int			sock;
 
-		/* We must reconstruct the fd_set for each call to select_loop */
-		FD_ZERO(&slotset);
+		sock = PQsocket(slots[i].connection);
 
-		for (i = 0; i < numslots; i++)
+		if (sock >= 0 && FD_ISSET(sock, &slotset))
 		{
-			int			sock = PQsocket(slots[i].connection);
-
-			/*
-			 * We don't really expect any connections to lose their sockets
-			 * after startup, but just in case, cope by ignoring them.
-			 */
-			if (sock < 0)
-				continue;
-
-			FD_SET(sock, &slotset);
-			if (sock > maxFd)
-				maxFd = sock;
+			/* select() says input is available, so consume it */
+			PQconsumeInput(slots[i].connection);
 		}
 
-		SetCancelConn(slots->connection);
-		i = select_loop(maxFd, &slotset);
-		ResetCancelConn();
-
-		/* failure? */
-		if (i < 0)
-			return NULL;
-
-		for (i = 0; i < numslots; i++)
+		/* Collect result(s) as long as any are available */
+		while (!PQisBusy(slots[i].connection))
 		{
-			int			sock = PQsocket(slots[i].connection);
+			PGresult   *result = PQgetResult(slots[i].connection);
 
-			if (sock >= 0 && FD_ISSET(sock, &slotset))
+			if (result != NULL)
 			{
-				/* select() says input is available, so consume it */
-				PQconsumeInput(slots[i].connection);
+				/* Handle and discard the command result */
+				if (!processQueryResult(slots + i, result))
+					return false;
 			}
-
-			/* Collect result(s) as long as any are available */
-			while (!PQisBusy(slots[i].connection))
+			else
 			{
-				PGresult   *result = PQgetResult(slots[i].connection);
-
-				if (result != NULL)
-				{
-					/* Handle and discard the command result */
-					if (!processQueryResult(slots + i, result))
-						return NULL;
-				}
-				else
-				{
-					/* This connection has become idle */
-					slots[i].isFree = true;
-					ParallelSlotClearHandler(slots + i);
-					if (firstFree < 0)
-						firstFree = i;
-					break;
-				}
+				/* This connection has become idle */
+				slots[i].inUse = false;
+				ParallelSlotClearHandler(slots + i);
+				break;
 			}
 		}
 	}
+	return true;
+}
+
+/*
+ * Close a slot's database connection.
+ */
+static void
+disconnect_slot(ParallelSlot *slot)
+{
+	Assert(slot);
+	Assert(slot->connection);
 
-	slots[firstFree].isFree = false;
-	return slots + firstFree;
+	disconnectDatabase(slot->connection);
+	slot->connection = NULL;
+	free_cparams_copy(&slot->cparams);
+	memset(&slot->cparams, 0, sizeof(ConnParams));
 }
 
 /*
- * ParallelSlotsSetup
- *		Prepare a set of parallel slots to use on a given database.
+ * Open a new database connection using the given connection parameters,
+ * execute an initial command if supplied, and associate the new connection
+ * with the given slot.
+ */
+static void
+connect_slot(ParallelSlot *slot, const ConnParams *cparams,
+			 const char *progname, bool echo, const char *initcmd)
+{
+	Assert(slot);
+	Assert(slot->connection == NULL);
+
+	slot->connection = connectDatabase(cparams, progname, echo, false, true);
+	if (PQsocket(slot->connection) >= FD_SETSIZE)
+	{
+		pg_log_fatal("too many jobs for this platform");
+		exit(1);
+	}
+
+	/*
+	 * The caller is at liberty to reuse the cparams struct, overwriting
+	 * fields (in particular, override_dbname).  We need our own deep copy of
+	 * the parameter struct fields.
+	 */
+	deep_copy_cparams(&slot->cparams, cparams);
+
+	/* Setup the connection using the supplied command, if any. */
+	if (initcmd)
+		executeCommand(slot->connection, initcmd, echo);
+}
+
+/*
+ * ParallelSlotsGetIdle
+ *		Return a connection slot that is ready to execute a command.
+ *
+ * The slot returned is chosen as follows:
+ *
+ * If any idle slot already has an open connection, and if either cparams is
+ * null or the connection was formed using connection parameter string values
+ * identical to those in cparams, that slot will be returned allowing the
+ * connection to be reused.
+ *
+ * Otherwise, if cparams is not null, and if any idle slot is not yet connected
+ * to a database, the slot will be returned with it's connection opened using
+ * the supplied cparams.
+ *
+ * Otherwise, if cparams is not null, and if any idle slot exists, an idle slot
+ * will be chosen and returned after having it's connection disconnected and
+ * reconnected using the supplied cparams.
+ *
+ * Otherwise, if any slots have connections that are busy, we loop on select()
+ * until one socket becomes available.  When this happens, we read the whole
+ * set and mark as free all sockets that become available.  We then select a
+ * slot using the same rules as above.
+ *
+ * Otherwise, we cannot return a slot, which is an error, and NULL is returned.
+ *
+ * For any connection created, if "initcmd" is not null, it will be executed as
+ * a command on the newly formed connection before the slot is returned.
  *
- * This creates and initializes a set of connections to the database
- * using the information given by the caller, marking all parallel slots
- * as free and ready to use.  "conn" is an initial connection set up
- * by the caller and is associated with the first slot in the parallel
- * set.
+ * If an error occurs, NULL is returned.
  */
 ParallelSlot *
-ParallelSlotsSetup(const ConnParams *cparams,
-				   const char *progname, bool echo,
-				   PGconn *conn, int numslots)
+ParallelSlotsGetIdle(ParallelSlot *slots, int numslots,
+					 const ConnParams *cparams, const char *progname,
+					 bool echo, const char *initcmd)
 {
-	ParallelSlot *slots;
-	int			i;
+	int			offset;
 
-	Assert(conn != NULL);
+	Assert(slots);
+	Assert(numslots > 0);
+	Assert(cparams);
+	Assert(progname);
 
-	slots = (ParallelSlot *) pg_malloc(sizeof(ParallelSlot) * numslots);
-	init_slot(slots, conn);
-	if (numslots > 1)
+	while (1)
 	{
-		for (i = 1; i < numslots; i++)
+		/* First choice: a slot already connected to the desired database. */
+		offset = find_matching_idle_slot(slots, numslots, cparams);
+		if (offset >= 0)
 		{
-			conn = connectDatabase(cparams, progname, echo, false, true);
-
-			/*
-			 * Fail and exit immediately if trying to use a socket in an
-			 * unsupported range.  POSIX requires open(2) to use the lowest
-			 * unused file descriptor and the hint given relies on that.
-			 */
-			if (PQsocket(conn) >= FD_SETSIZE)
-			{
-				pg_log_fatal("too many jobs for this platform -- try %d", i);
-				exit(1);
-			}
+			slots[offset].inUse = true;
+			return slots + offset;
+		}
 
-			init_slot(slots + i, conn);
+		/* Second choice: a slot not connected to any database. */
+		offset = find_unconnected_slot(slots, numslots);
+		if (offset >= 0)
+		{
+			connect_slot(slots + offset, cparams, progname, echo, initcmd);
+			slots[offset].inUse = true;
+			return slots + offset;
+		}
+
+		/* Third choice: a slot connected to the wrong database. */
+		offset = find_any_idle_slot(slots, numslots);
+		if (offset >= 0)
+		{
+			disconnect_slot(slots + offset);
+			connect_slot(slots + offset, cparams, progname, echo, initcmd);
+			slots[offset].inUse = true;
+			return slots + offset;
 		}
+
+		/*
+		 * Fourth choice: block until one or more slots become available. If
+		 * any slot's hit a fatal error, we'll find out about that here and
+		 * return NULL.
+		 */
+		if (!wait_on_slots(slots, numslots, progname))
+			return NULL;
+	}
+}
+
+/*
+ * ParallelSlotsSetupMinimal
+ *		Prepare a set of parallel slots but do not connect to any database.
+ *
+ * This creates and initializes a set of slots, marking all parallel slots
+ * as free and ready to use.  Establishing connections is delayed until
+ * requesting a free slot, but in the event that an existing connection is
+ * provided in "conn", that connection will be associated with the first
+ * slot and saved for reuse.  In this case, "cparams" must contain the
+ * parameters that were used for opening "conn".
+ */
+ParallelSlot *
+ParallelSlotsSetupMinimal(int numslots, PGconn *conn,
+						  const ConnParams *cparams)
+{
+	ParallelSlot *slots;
+
+	Assert(numslots > 0);
+
+	slots = (ParallelSlot *) palloc0(sizeof(ParallelSlot) * numslots);
+	if (conn != NULL)
+	{
+		slots[0].connection = conn;
+		deep_copy_cparams(&slots[0].cparams, cparams);
 	}
 
 	return slots;
 }
 
+/*
+ * ParallelSlotsSetupOneDB
+ *		Prepare a set of parallel slots to use on a given database.
+ *
+ * This creates and initializes a set of connections to the database using the
+ * information given by the caller, marking all parallel slots as free and
+ * ready to use.  If not null, "conn" is an initial connection set up by the
+ * caller and is associated with the first slot in the parallel set.  "cparams"
+ * is used to form the remaining connections, and must be the same as was used
+ * for creating the inital connection "conn".  If not null, "initcmd" is run
+ * on each connection opened, not including "conn".
+ */
+ParallelSlot *
+ParallelSlotsSetupOneDB(const ConnParams *cparams, const char *progname,
+						bool echo, PGconn *conn, int numslots,
+						const char *initcmd)
+{
+	int			i = 0;
+	ParallelSlot *slots = ParallelSlotsSetupMinimal(numslots, conn, cparams);
+
+	if (conn)
+		i++;					/* first slot already assigned "conn" */
+	for (; i < numslots; i++)
+		connect_slot(&slots[i], cparams, progname, echo, initcmd);
+
+	return slots;
+}
+
 /*
  * ParallelSlotsTerminate
  *		Clean up a set of parallel slots
@@ -320,6 +593,8 @@ ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots)
 
 	for (i = 0; i < numslots; i++)
 	{
+		if (slots[i].connection == NULL)
+			continue;
 		if (!consumeQueryResult(slots + i))
 			return false;
 	}
@@ -350,6 +625,9 @@ ParallelSlotsWaitCompletion(ParallelSlot *slots, int numslots)
 bool
 TableCommandResultHandler(PGresult *res, PGconn *conn, void *context)
 {
+	Assert(res != NULL);
+	Assert(conn != NULL);
+
 	/*
 	 * If it's an error, report it.  Errors about a missing table are harmless
 	 * so we continue processing; but die for other errors.
diff --git a/src/include/fe_utils/parallel_slot.h b/src/include/fe_utils/parallel_slot.h
index 8902f8d4f4..d42bc33c87 100644
--- a/src/include/fe_utils/parallel_slot.h
+++ b/src/include/fe_utils/parallel_slot.h
@@ -21,7 +21,8 @@ typedef bool (*ParallelSlotResultHandler) (PGresult *res, PGconn *conn,
 typedef struct ParallelSlot
 {
 	PGconn	   *connection;		/* One connection */
-	bool		isFree;			/* Is it known to be idle? */
+	ConnParams	cparams;		/* Params used to form connection */
+	bool		inUse;			/* Is the slot being used? */
 
 	/*
 	 * Prior to issuing a command or query on 'connection', a handler callback
@@ -48,11 +49,18 @@ ParallelSlotClearHandler(ParallelSlot *slot)
 	slot->handler_context = NULL;
 }
 
-extern ParallelSlot *ParallelSlotsGetIdle(ParallelSlot *slots, int numslots);
+extern ParallelSlot *ParallelSlotsGetIdle(ParallelSlot *slots, int numslots,
+										  const ConnParams *cparams,
+										  const char *progname, bool echo,
+										  const char *initcmd);
 
-extern ParallelSlot *ParallelSlotsSetup(const ConnParams *cparams,
-										const char *progname, bool echo,
-										PGconn *conn, int numslots);
+extern ParallelSlot *ParallelSlotsSetupMinimal(int numslots, PGconn *conn,
+											   const ConnParams *cparams);
+
+extern ParallelSlot *ParallelSlotsSetupOneDB(const ConnParams *cparams,
+											 const char *progname, bool echo,
+											 PGconn *conn, int numslots,
+											 const char *initcmd);
 
 extern void ParallelSlotsTerminate(ParallelSlot *slots, int numslots);
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index bab4f3adb3..caae8cbd5b 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -403,6 +403,7 @@ ConfigData
 ConfigVariable
 ConnCacheEntry
 ConnCacheKey
+ConnParams
 ConnStatusType
 ConnType
 ConnectionStateEnum
-- 
2.21.1 (Apple Git-122.3)