Refactor cdbconn_discardResults() (#1804)

Previosly, `cdbconn_discardResults()` has proved to be unrealiable in case of
OOM, as it would still try to read the results, causing errors in recursion if
called when aborting a query.

This patch refactors the said function by transferring code from another
similar procedure, to make sure we're discarding results in a more robust way,
adressing all libpq pitfalls instead of mindlessly retrying, as opposed to the
past behavior.

Ticket: ADBDEV-7690

Author: bandetto

src/backend/cdb/cdbutil.c

@@ -894,7 +894,7 @@ cleanupQE(SegmentDatabaseDescriptor *segdbDesc)
 		return false;
 
 	/* Note, we cancel all "still running" queries */
-	if (!cdbconn_discardResults(segdbDesc, 20))
+	if (!cdbconn_discardResults(segdbDesc, -1))
 	{
 		elog(LOG, "cleaning up seg%d while it is still busy", segdbDesc->segindex);
 		return false;

src/backend/cdb/dispatcher/cdbconn.c

@@ -337,57 +337,64 @@ cdbconn_disconnect(SegmentDatabaseDescriptor *segdbDesc)
 }
 
 /*
- * Read result from connection and discard it.
+ * Discard any results from a connection.
  *
- * Retry at most N times.
+ * retryCount is deprecated and isn't used anymore. This function tries to avoid
+ * extra allocations and is safe to be called even in case of OOM.
  *
- * Return false if there'er still leftovers.
+ * Return false if connection is still busy or dead.
  */
 bool
 cdbconn_discardResults(SegmentDatabaseDescriptor *segdbDesc,
 					   int retryCount)
 {
-	PGresult   *pRes = NULL;
-	ExecStatusType stat;
-	int			i = 0;
-	bool retval = true;
+	PGresult   *res;
+	PGnotify   *notify;
+	PGconn	   *conn = segdbDesc->conn;
 
-	/* PQstatus() is smart enough to handle NULL */
-	while (NULL != (pRes = PQgetResult(segdbDesc->conn)))
-	{
-		stat = PQresultStatus(pRes);
-		PQclear(pRes);
+	(void) retryCount;
 
-		elog(LOG, "(%s) Leftover result at freeGang time: %s %s", segdbDesc->whoami,
-			 PQresStatus(stat),
-			 PQerrorMessage(segdbDesc->conn));
+	/* Free some memory and replace current result with a fatal error dummy. */
+	pqSaveErrorResult(conn);
 
-		if (stat == PGRES_FATAL_ERROR || stat == PGRES_BAD_RESPONSE)
-		{
-			retval = true;
-			break;
-		}
+	/* Make sure PQgetResult() calls are not blocking. */
+	PQconsumeInput(conn);
 
-		if (i++ > retryCount)
+	/*
+	 * Discard anything that is unread. Since our result contains a fatal
+	 * error, we'll just consume the entire message without actually parsing
+	 * it.
+	 */
+	while (!PQisBusy(conn) && (res = PQgetResult(conn)) != NULL)
+	{
+		switch (PQresultStatus(res))
 		{
-			retval = false;
-			break;
+			case PGRES_COPY_IN:
+			case PGRES_COPY_OUT:
+			case PGRES_COPY_BOTH:
+				PQendcopy(conn);
+				/* fallthrough */
+			default:
+				PQclear(res);
 		}
+
+		pqSaveErrorResult(conn);
 	}
 
-	/*
-	 * Clear of all the notify messages as well.
-	 */
-	PGnotify   *notify = segdbDesc->conn->notifyHead;
-	while (notify != NULL)
+	/* Free notices. */
+	while ((notify = PQnotifies(conn)) != NULL)
+		PQfreemem(notify);
+
+	/* The result is not needed anymore. */
+	pqClearAsyncResult(conn);
+
+	if (PQisBusy(conn) && PQstatus(conn) != CONNECTION_BAD)
 	{
-		PGnotify   *prev = notify;
-		notify = notify->next;
-		PQfreemem(prev);
+		/* Some work is still remaining until we can die. */
+		return false;
 	}
-	segdbDesc->conn->notifyHead = segdbDesc->conn->notifyTail = NULL;
 
-	return retval;
+	return true;
 }
 
 /* Return if it's a bad connection */

src/backend/cdb/dispatcher/cdbdisp_async.c

@@ -825,60 +825,6 @@ handlePollError(CdbDispatchCmdAsync *pParms)
 	return;
 }
 
-/*
- * Try to completely discard results from a dispatchResult's connection without
- * extra memory allocations by abusing libpq state-machine hacks.
- */
-static void
-tryToWipeResults(CdbDispatchResult *dispatchResult)
-{
-	PGresult   *res;
-	PGnotify   *notify;
-	PGconn	   *conn = dispatchResult->segdbDesc->conn;
-
-	/* Free some memory and replace current result with a fatal error dummy. */
-	pqSaveErrorResult(conn);
-
-	/* Make sure PQgetResult() calls are not blocking. */
-	PQconsumeInput(conn);
-
-	/*
-	 * Discard anything that is unread. Since our result contains a fatal
-	 * error, we'll just consume the entire message without actually parsing
-	 * it.
-	 */
-	while (!PQisBusy(conn) && (res = PQgetResult(conn)) != NULL)
-	{
-		switch (PQresultStatus(res))
-		{
-			case PGRES_COPY_IN:
-			case PGRES_COPY_OUT:
-			case PGRES_COPY_BOTH:
-				PQendcopy(conn);
-				/* fallthrough */
-			default:
-				PQclear(res);
-		}
-
-		pqSaveErrorResult(conn);
-	}
-
-	/* Free notices. */
-	while ((notify = PQnotifies(conn)) != NULL)
-		PQfreemem(notify);
-
-	/* The result is not needed anymore. */
-	pqClearAsyncResult(conn);
-
-	if (PQisBusy(conn) && PQstatus(conn) != CONNECTION_BAD)
-	{
-		/* Some work is still remaining until we can die. */
-		return;
-	}
-
-	dispatchResult->stillRunning = false;
-}
-
 /*
  * Receive and process results from QEs.
  */
@@ -936,7 +882,8 @@ handlePollSuccess(CdbDispatchCmdAsync *pParms,
 			 * might not be enough memory to discard the result properly.
 			 * Let's get the big guns out.
 			 */
-			tryToWipeResults(dispatchResult);
+			dispatchResult->stillRunning =
+				!cdbconn_discardResults(dispatchResult->segdbDesc, -1);
 
 			forwardQENotices();
 

src/include/cdb/cdbconn.h

@@ -78,7 +78,7 @@ cdbconn_doConnectComplete(SegmentDatabaseDescriptor *segdbDesc);
 /*
  * Read result from connection and discard it.
  *
- * Retry at most N times.
+ * retryCount is deprecated and isn't used anymore.
  *
  * Return false if there'er still leftovers.
  */