Replace libpq's "row processor" API with a "single row" mode.

After taking awhile to digest the row-processor feature that was added to
libpq in commit 92785dac2ee7026948962cd61c4cd84a2d052772, we've concluded
it is over-complicated and too hard to use.  Leave the core infrastructure
changes in place (that is, there's still a row processor function inside
libpq), but remove the exposed API pieces, and instead provide a "single
row" mode switch that causes PQgetResult to return one row at a time in
separate PGresult objects.

This approach incurs more overhead than proper use of a row processor
callback would, since construction of a PGresult per row adds extra cycles.
However, it is far easier to use and harder to break.  The single-row mode
still affords applications the primary benefit that the row processor API
was meant to provide, namely not having to accumulate large result sets in
memory before processing them.  Preliminary testing suggests that we can
probably buy back most of the extra cycles by micro-optimizing construction
of the extra results, but that task will be left for another day.

Marko Kreen

1 files changed, 24 insertions, 92 deletions

diff --git a/src/interfaces/libpq/fe-protocol2.c b/src/interfaces/libpq/fe-protocol2.c
index 8dbd6b69823..1ba5885cd3b 100644
--- a/src/interfaces/libpq/fe-protocol2.c
+++ b/src/interfaces/libpq/fe-protocol2.c
@@ -49,19 +49,11 @@ static int	getNotify(PGconn *conn);
 PostgresPollingStatusType
 pqSetenvPoll(PGconn *conn)
 {
-	PostgresPollingStatusType result;
 	PGresult   *res;
-	PQrowProcessor savedRowProcessor;
-	void	   *savedRowProcessorParam;
 
 	if (conn == NULL || conn->status == CONNECTION_BAD)
 		return PGRES_POLLING_FAILED;
 
-	/* Ensure the standard row processor is used to collect any results */
-	savedRowProcessor = conn->rowProcessor;
-	savedRowProcessorParam = conn->rowProcessorParam;
-	PQsetRowProcessor(conn, NULL, NULL);
-
 	/* Check whether there are any data for us */
 	switch (conn->setenv_state)
 	{
@@ -77,10 +69,7 @@ pqSetenvPoll(PGconn *conn)
 				if (n < 0)
 					goto error_return;
 				if (n == 0)
-				{
-					result = PGRES_POLLING_READING;
-					goto normal_return;
-				}
+					return PGRES_POLLING_READING;
 
 				break;
 			}
@@ -94,8 +83,7 @@ pqSetenvPoll(PGconn *conn)
 
 			/* Should we raise an error if called when not active? */
 		case SETENV_STATE_IDLE:
-			result = PGRES_POLLING_OK;
-			goto normal_return;
+			return PGRES_POLLING_OK;
 
 		default:
 			printfPQExpBuffer(&conn->errorMessage,
@@ -192,10 +180,7 @@ pqSetenvPoll(PGconn *conn)
 			case SETENV_STATE_CLIENT_ENCODING_WAIT:
 				{
 					if (PQisBusy(conn))
-					{
-						result = PGRES_POLLING_READING;
-						goto normal_return;
-					}
+						return PGRES_POLLING_READING;
 
 					res = PQgetResult(conn);
 
@@ -220,10 +205,7 @@ pqSetenvPoll(PGconn *conn)
 			case SETENV_STATE_OPTION_WAIT:
 				{
 					if (PQisBusy(conn))
-					{
-						result = PGRES_POLLING_READING;
-						goto normal_return;
-					}
+						return PGRES_POLLING_READING;
 
 					res = PQgetResult(conn);
 
@@ -262,17 +244,13 @@ pqSetenvPoll(PGconn *conn)
 						goto error_return;
 
 					conn->setenv_state = SETENV_STATE_QUERY1_WAIT;
-					result = PGRES_POLLING_READING;
-					goto normal_return;
+					return PGRES_POLLING_READING;
 				}
 
 			case SETENV_STATE_QUERY1_WAIT:
 				{
 					if (PQisBusy(conn))
-					{
-						result = PGRES_POLLING_READING;
-						goto normal_return;
-					}
+						return PGRES_POLLING_READING;
 
 					res = PQgetResult(conn);
 
@@ -349,17 +327,13 @@ pqSetenvPoll(PGconn *conn)
 						goto error_return;
 
 					conn->setenv_state = SETENV_STATE_QUERY2_WAIT;
-					result = PGRES_POLLING_READING;
-					goto normal_return;
+					return PGRES_POLLING_READING;
 				}
 
 			case SETENV_STATE_QUERY2_WAIT:
 				{
 					if (PQisBusy(conn))
-					{
-						result = PGRES_POLLING_READING;
-						goto normal_return;
-					}
+						return PGRES_POLLING_READING;
 
 					res = PQgetResult(conn);
 
@@ -406,8 +380,7 @@ pqSetenvPoll(PGconn *conn)
 					{
 						/* Query finished, so we're done */
 						conn->setenv_state = SETENV_STATE_IDLE;
-						result = PGRES_POLLING_OK;
-						goto normal_return;
+						return PGRES_POLLING_OK;
 					}
 					break;
 				}
@@ -425,12 +398,7 @@ pqSetenvPoll(PGconn *conn)
 
 error_return:
 	conn->setenv_state = SETENV_STATE_IDLE;
-	result = PGRES_POLLING_FAILED;
-
-normal_return:
-	conn->rowProcessor = savedRowProcessor;
-	conn->rowProcessorParam = savedRowProcessorParam;
-	return result;
+	return PGRES_POLLING_FAILED;
 }
 
 
@@ -438,9 +406,6 @@ normal_return:
  * parseInput: if appropriate, parse input data from backend
  * until input is exhausted or a stopping state is reached.
  * Note that this function will NOT attempt to read more data from the backend.
- *
- * Note: callers of parseInput must be prepared for a longjmp exit when we are
- * in PGASYNC_BUSY state, since an external row processor might do that.
  */
 void
 pqParseInput2(PGconn *conn)
@@ -746,31 +711,16 @@ getRowDescriptions(PGconn *conn)
 	/* Success! */
 	conn->result = result;
 
-	/*
-	 * Advance inStart to show that the "T" message has been processed.  We
-	 * must do this before calling the row processor, in case it longjmps.
-	 */
+	/* Advance inStart to show that the "T" message has been processed. */
 	conn->inStart = conn->inCursor;
 
-	/* Give the row processor a chance to initialize for new result set */
-	errmsg = NULL;
-	switch ((*conn->rowProcessor) (result, NULL, &errmsg,
-								   conn->rowProcessorParam))
-	{
-		case 1:
-			/* everything is good */
-			return 0;
-
-		case -1:
-			/* error, report the errmsg below */
-			break;
+	/*
+	 * We could perform additional setup for the new result set here, but for
+	 * now there's nothing else to do.
+	 */
 
-		default:
-			/* unrecognized return code */
-			errmsg = libpq_gettext("unrecognized return value from row processor");
-			break;
-	}
-	goto set_error_result;
+	/* And we're done. */
+	return 0;
 
 advance_and_error:
 
@@ -781,8 +731,6 @@ advance_and_error:
 	 */
 	conn->inStart = conn->inEnd;
 
-set_error_result:
-
 	/*
 	 * Replace partially constructed result with an error result. First
 	 * discard the old result to try to win back some memory.
@@ -790,7 +738,7 @@ set_error_result:
 	pqClearAsyncResult(conn);
 
 	/*
-	 * If row processor didn't provide an error message, assume "out of
+	 * If preceding code didn't provide an error message, assume "out of
 	 * memory" was meant.  The advantage of having this special case is that
 	 * freeing the old result first greatly improves the odds that gettext()
 	 * will succeed in providing a translation.
@@ -937,31 +885,15 @@ getAnotherTuple(PGconn *conn, bool binary)
 		free(bitmap);
 	bitmap = NULL;
 
-	/*
-	 * Advance inStart to show that the "D" message has been processed.  We
-	 * must do this before calling the row processor, in case it longjmps.
-	 */
+	/* Advance inStart to show that the "D" message has been processed. */
 	conn->inStart = conn->inCursor;
 
-	/* Pass the completed row values to rowProcessor */
+	/* Process the collected row */
 	errmsg = NULL;
-	switch ((*conn->rowProcessor) (result, rowbuf, &errmsg,
-								   conn->rowProcessorParam))
-	{
-		case 1:
-			/* everything is good */
-			return 0;
+	if (pqRowProcessor(conn, &errmsg))
+		return 0;				/* normal, successful exit */
 
-		case -1:
-			/* error, report the errmsg below */
-			break;
-
-		default:
-			/* unrecognized return code */
-			errmsg = libpq_gettext("unrecognized return value from row processor");
-			break;
-	}
-	goto set_error_result;
+	goto set_error_result;		/* pqRowProcessor failed, report it */
 
 advance_and_error:
 
@@ -981,7 +913,7 @@ set_error_result:
 	pqClearAsyncResult(conn);
 
 	/*
-	 * If row processor didn't provide an error message, assume "out of
+	 * If preceding code didn't provide an error message, assume "out of
 	 * memory" was meant.  The advantage of having this special case is that
 	 * freeing the old result first greatly improves the odds that gettext()
 	 * will succeed in providing a translation.