libpq_rowproc_20120223.patch

diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt
index 1af8df6..7e02497 100644
--- a/src/interfaces/libpq/exports.txt
+++ b/src/interfaces/libpq/exports.txt
@@ -160,3 +160,7 @@ PQconnectStartParams      157
 PQping                    158
 PQpingParams              159
 PQlibVersion              160
+PQsetRowProcessor	  161
+PQgetRowProcessor	  162
+PQsetRowProcessorErrMsg	  163
+PQskipResult		  164
diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c
index 27a9805..4605e49 100644
--- a/src/interfaces/libpq/fe-connect.c
+++ b/src/interfaces/libpq/fe-connect.c
@@ -2693,6 +2693,9 @@ makeEmptyPGconn(void)
 	conn->wait_ssl_try = false;
 #endif
 
+	/* set default row processor */
+	PQsetRowProcessor(conn, NULL, NULL);
+
 	/*
 	 * We try to send at least 8K at a time, which is the usual size of pipe
 	 * buffers on Unix systems.  That way, when we are sending a large amount
@@ -2711,8 +2714,13 @@ makeEmptyPGconn(void)
 	initPQExpBuffer(&conn->errorMessage);
 	initPQExpBuffer(&conn->workBuffer);
 
+	/* set up initial row buffer */
+	conn->rowBufLen = 32;
+	conn->rowBuf = (PGrowValue *)malloc(conn->rowBufLen * sizeof(PGrowValue));
+
 	if (conn->inBuffer == NULL ||
 		conn->outBuffer == NULL ||
+		conn->rowBuf == NULL ||
 		PQExpBufferBroken(&conn->errorMessage) ||
 		PQExpBufferBroken(&conn->workBuffer))
 	{
@@ -2814,6 +2822,8 @@ freePGconn(PGconn *conn)
 		free(conn->inBuffer);
 	if (conn->outBuffer)
 		free(conn->outBuffer);
+	if (conn->rowBuf)
+		free(conn->rowBuf);
 	termPQExpBuffer(&conn->errorMessage);
 	termPQExpBuffer(&conn->workBuffer);
 
@@ -5078,3 +5088,4 @@ PQregisterThreadLock(pgthreadlock_t newhandler)
 
 	return prev;
 }
+
diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c
index b743566..cd287cd 100644
--- a/src/interfaces/libpq/fe-exec.c
+++ b/src/interfaces/libpq/fe-exec.c
@@ -66,6 +66,7 @@ static PGresult *PQexecFinish(PGconn *conn);
 static int PQsendDescribe(PGconn *conn, char desc_type,
 			   const char *desc_target);
 static int	check_field_number(const PGresult *res, int field_num);
+static int	pqAddRow(PGresult *res, PGrowValue *columns, void *param);
 
 
 /* ----------------
@@ -160,6 +161,7 @@ PQmakeEmptyPGresult(PGconn *conn, ExecStatusType status)
 	result->curBlock = NULL;
 	result->curOffset = 0;
 	result->spaceLeft = 0;
+	result->rowProcessorErrMsg = NULL;
 
 	if (conn)
 	{
@@ -701,7 +703,6 @@ pqClearAsyncResult(PGconn *conn)
 	if (conn->result)
 		PQclear(conn->result);
 	conn->result = NULL;
-	conn->curTuple = NULL;
 }
 
 /*
@@ -756,7 +757,6 @@ pqPrepareAsyncResult(PGconn *conn)
 	 */
 	res = conn->result;
 	conn->result = NULL;		/* handing over ownership to caller */
-	conn->curTuple = NULL;		/* just in case */
 	if (!res)
 		res = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR);
 	else
@@ -828,6 +828,87 @@ pqInternalNotice(const PGNoticeHooks *hooks, const char *fmt,...)
 }
 
 /*
+ * PQsetRowProcessor
+ *   Set function that copies column data out from network buffer.
+ */
+void
+PQsetRowProcessor(PGconn *conn, PQrowProcessor func, void *param)
+{
+	conn->rowProcessor = (func ? func : pqAddRow);
+	conn->rowProcessorParam = param;
+}
+
+/*
+ * PQgetRowProcessor
+ *   Get current row processor of conn. set pointer to current parameter for
+ *   row processor to param if not NULL.
+ */
+PQrowProcessor
+PQgetRowProcessor(PGconn *conn, void **param)
+{
+	if (param)
+		*param = conn->rowProcessorParam;
+
+	return conn->rowProcessor;
+}
+
+/*
+ * PQsetRowProcessorErrMsg
+ *    Set the error message pass back to the caller of RowProcessor.
+ *
+ *    You can replace the previous message by alternative mes, or clear
+ *    it with NULL.
+ */
+void
+PQsetRowProcessorErrMsg(PGresult *res, const char *msg)
+{
+	if (msg)
+		res->rowProcessorErrMsg = pqResultStrdup(res, msg);
+	else
+		res->rowProcessorErrMsg = NULL;
+}
+
+/*
+ * pqAddRow
+ *	  add a row to the PGresult structure, growing it if necessary
+ *	  Returns TRUE if OK, FALSE if not enough memory to add the row.
+ */
+static int
+pqAddRow(PGresult *res, PGrowValue *columns, void *param)
+{
+	PGresAttValue *tup;
+	int			nfields = res->numAttributes;
+	int			i;
+
+	tup = (PGresAttValue *)
+		pqResultAlloc(res, nfields * sizeof(PGresAttValue), TRUE);
+	if (tup == NULL)
+		return FALSE;
+
+	for (i = 0 ; i < nfields ; i++)
+	{
+		tup[i].len = columns[i].len;
+		if (tup[i].len == NULL_LEN)
+		{
+			tup[i].value = res->null_field;
+		}
+		else
+		{
+			bool isbinary = (res->attDescs[i].format != 0);
+			tup[i].value = (char *)pqResultAlloc(res, tup[i].len + 1, isbinary);
+			if (tup[i].value == NULL)
+				return FALSE;
+
+			memcpy(tup[i].value, columns[i].value, tup[i].len);
+			/* We have to terminate this ourselves */
+			tup[i].value[tup[i].len] = '\0';
+		}
+	}
+
+	return pqAddTuple(res, tup);
+}
+
+/*
  * pqAddTuple
  *	  add a row pointer to the PGresult structure, growing it if necessary
  *	  Returns TRUE if OK, FALSE if not enough memory to add the row
@@ -1223,7 +1304,6 @@ PQsendQueryStart(PGconn *conn)
 
 	/* initialize async result-accumulation state */
 	conn->result = NULL;
-	conn->curTuple = NULL;
 
 	/* ready to send command message */
 	return true;
@@ -1831,6 +1911,55 @@ PQexecFinish(PGconn *conn)
 	return lastResult;
 }
 
+
+/*
+ * Do-nothing row processor for PQskipResult
+ */
+static int
+dummyRowProcessor(PGresult *res, PGrowValue *columns, void *param)
+{
+	return 1;
+}
+
+/*
+ * Exaust remaining Data Rows in curret conn.
+ * 
+ * Exaust current result if skipAll is false and all succeeding results if
+ * true.
+ */
+int
+PQskipResult(PGconn *conn, int skipAll)
+{
+	PQrowProcessor savedRowProcessor;
+	void * savedRowProcParam;
+	PGresult *res;
+	int ret = 0;
+
+	/* save the current row processor settings and set dummy processor */
+	savedRowProcessor = PQgetRowProcessor(conn, &savedRowProcParam);
+	PQsetRowProcessor(conn, dummyRowProcessor, NULL);
+	
+	/*
+	 * Throw away the remaining rows in current result, or all succeeding
+	 * results if skipAll is not FALSE.
+	 */
+	if (skipAll)
+	{
+		while ((res = PQgetResult(conn)) != NULL)
+			PQclear(res);
+	}
+	else if ((res = PQgetResult(conn)) != NULL)
+	{
+		PQclear(res);
+		ret = 1;
+	}
+	
+	PQsetRowProcessor(conn, savedRowProcessor, savedRowProcParam);
+
+	return ret;
+}
+
+
 /*
  * PQdescribePrepared
  *	  Obtain information about a previously prepared statement
diff --git a/src/interfaces/libpq/fe-misc.c b/src/interfaces/libpq/fe-misc.c
index ce0eac3..d11cb3c 100644
--- a/src/interfaces/libpq/fe-misc.c
+++ b/src/interfaces/libpq/fe-misc.c
@@ -219,6 +219,25 @@ pqGetnchar(char *s, size_t len, PGconn *conn)
 }
 
 /*
+ * pqGetnchar:
+ *	skip len bytes in input buffer.
+ */
+int
+pqSkipnchar(size_t len, PGconn *conn)
+{
+	if (len > (size_t) (conn->inEnd - conn->inCursor))
+		return EOF;
+
+	conn->inCursor += len;
+
+	if (conn->Pfdebug)
+		fprintf(conn->Pfdebug, "From backend (%lu skipped)\n",
+				(unsigned long) len);
+
+	return 0;
+}
+
+/*
  * pqPutnchar:
  *	write exactly len bytes to the current message
  */
diff --git a/src/interfaces/libpq/fe-protocol2.c b/src/interfaces/libpq/fe-protocol2.c
index a7c3899..6578019 100644
--- a/src/interfaces/libpq/fe-protocol2.c
+++ b/src/interfaces/libpq/fe-protocol2.c
@@ -569,6 +569,8 @@ pqParseInput2(PGconn *conn)
 						/* Read another tuple of a normal query response */
 						if (getAnotherTuple(conn, FALSE))
 							return;
+						/* getAnotherTuple moves inStart itself */
+						continue;
 					}
 					else
 					{
@@ -585,6 +587,8 @@ pqParseInput2(PGconn *conn)
 						/* Read another tuple of a normal query response */
 						if (getAnotherTuple(conn, TRUE))
 							return;
+						/* getAnotherTuple moves inStart itself */
+						continue;
 					}
 					else
 					{
@@ -703,52 +707,51 @@ failure:
 
 /*
  * parseInput subroutine to read a 'B' or 'D' (row data) message.
- * We add another tuple to the existing PGresult structure.
+ * It fills rowbuf with column pointers and then calls row processor.
  * Returns: 0 if completed message, EOF if error or not enough data yet.
  *
  * Note that if we run out of data, we have to suspend and reprocess
- * the message after more data is received.  We keep a partially constructed
- * tuple in conn->curTuple, and avoid reallocating already-allocated storage.
+ * the message after more data is received.
  */
 static int
 getAnotherTuple(PGconn *conn, bool binary)
 {
 	PGresult   *result = conn->result;
 	int			nfields = result->numAttributes;
-	PGresAttValue *tup;
+	PGrowValue  *rowbuf;
 
 	/* the backend sends us a bitmap of which attributes are null */
 	char		std_bitmap[64]; /* used unless it doesn't fit */
 	char	   *bitmap = std_bitmap;
 	int			i;
+	int			rp;
 	size_t		nbytes;			/* the number of bytes in bitmap  */
 	char		bmap;			/* One byte of the bitmap */
 	int			bitmap_index;	/* Its index */
 	int			bitcnt;			/* number of bits examined in current byte */
 	int			vlen;			/* length of the current field value */
 
+	/* resize row buffer if needed */
+	if (nfields > conn->rowBufLen)
+	{
+		rowbuf = realloc(conn->rowBuf, nfields * sizeof(PGrowValue));
+		if (!rowbuf)
+			goto rowProcessError;
+		conn->rowBuf = rowbuf;
+		conn->rowBufLen = nfields;
+	}
+	else
+	{
+		rowbuf = conn->rowBuf;
+	}
+
 	result->binary = binary;
 
-	/* Allocate tuple space if first time for this data message */
-	if (conn->curTuple == NULL)
+	if (binary)
 	{
-		conn->curTuple = (PGresAttValue *)
-			pqResultAlloc(result, nfields * sizeof(PGresAttValue), TRUE);
-		if (conn->curTuple == NULL)
-			goto outOfMemory;
-		MemSet(conn->curTuple, 0, nfields * sizeof(PGresAttValue));
-
-		/*
-		 * If it's binary, fix the column format indicators.  We assume the
-		 * backend will consistently send either B or D, not a mix.
-		 */
-		if (binary)
-		{
-			for (i = 0; i < nfields; i++)
-				result->attDescs[i].format = 1;
-		}
+		for (i = 0; i < nfields; i++)
+			result->attDescs[i].format = 1;
 	}
-	tup = conn->curTuple;
 
 	/* Get the null-value bitmap */
 	nbytes = (nfields + BITS_PER_BYTE - 1) / BITS_PER_BYTE;
@@ -757,7 +760,7 @@ getAnotherTuple(PGconn *conn, bool binary)
 	{
 		bitmap = (char *) malloc(nbytes);
 		if (!bitmap)
-			goto outOfMemory;
+			goto rowProcessError;
 	}
 
 	if (pqGetnchar(bitmap, nbytes, conn))
@@ -771,34 +774,29 @@ getAnotherTuple(PGconn *conn, bool binary)
 	for (i = 0; i < nfields; i++)
 	{
 		if (!(bmap & 0200))
-		{
-			/* if the field value is absent, make it a null string */
-			tup[i].value = result->null_field;
-			tup[i].len = NULL_LEN;
-		}
+			vlen = NULL_LEN;
+		else if (pqGetInt(&vlen, 4, conn))
+				goto EOFexit;
 		else
 		{
-			/* get the value length (the first four bytes are for length) */
-			if (pqGetInt(&vlen, 4, conn))
-				goto EOFexit;
 			if (!binary)
 				vlen = vlen - 4;
 			if (vlen < 0)
 				vlen = 0;
-			if (tup[i].value == NULL)
-			{
-				tup[i].value = (char *) pqResultAlloc(result, vlen + 1, binary);
-				if (tup[i].value == NULL)
-					goto outOfMemory;
-			}
-			tup[i].len = vlen;
-			/* read in the value */
-			if (vlen > 0)
-				if (pqGetnchar((char *) (tup[i].value), vlen, conn))
-					goto EOFexit;
-			/* we have to terminate this ourselves */
-			tup[i].value[vlen] = '\0';
 		}
+
+		/*
+		 * rowbuf[i].value always points to the next address of the
+		 * length field even if the value is NULL, to allow safe
+		 * size estimates and data copy.
+		 */
+		rowbuf[i].value = conn->inBuffer + conn->inCursor;
+		rowbuf[i].len = vlen;
+
+		/* Skip the value */
+		if (vlen > 0 && pqSkipnchar(vlen, conn))
+			goto EOFexit;
+
 		/* advance the bitmap stuff */
 		bitcnt++;
 		if (bitcnt == BITS_PER_BYTE)
@@ -811,33 +809,56 @@ getAnotherTuple(PGconn *conn, bool binary)
 			bmap <<= 1;
 	}
 
-	/* Success!  Store the completed tuple in the result */
-	if (!pqAddTuple(result, tup))
-		goto outOfMemory;
-	/* and reset for a new message */
-	conn->curTuple = NULL;
-
 	if (bitmap != std_bitmap)
 		free(bitmap);
-	return 0;
+	bitmap = NULL;
+
+	/* tag the row as parsed */
+	conn->inStart = conn->inCursor;
+
+	/* Pass the completed row values to rowProcessor */
+	rp= conn->rowProcessor(result, rowbuf, conn->rowProcessorParam);
+	if (rp == 1)
+		return 0;
+	else if (rp == 2 && pqIsnonblocking(conn))
+		/* processor requested early exit */
+		return EOF;
+	else if (rp != 0)
+		PQsetRowProcessorErrMsg(result, libpq_gettext("invalid return value from row processor\n"));
+
+rowProcessError:
 
-outOfMemory:
 	/* Replace partially constructed result with an error result */
 
-	/*
-	 * we do NOT use pqSaveErrorResult() here, because of the likelihood that
-	 * there's not enough memory to concatenate messages...
-	 */
-	pqClearAsyncResult(conn);
-	printfPQExpBuffer(&conn->errorMessage,
-					  libpq_gettext("out of memory for query result\n"));
+	if (result->rowProcessorErrMsg)
+	{
+		printfPQExpBuffer(&conn->errorMessage, "%s", result->rowProcessorErrMsg);
+		pqSaveErrorResult(conn);
+	}
+	else
+	{
+		/*
+		 * we do NOT use pqSaveErrorResult() here, because of the likelihood that
+		 * there's not enough memory to concatenate messages...
+		 */
+		pqClearAsyncResult(conn);
+		resetPQExpBuffer(&conn->errorMessage);
 
-	/*
-	 * XXX: if PQmakeEmptyPGresult() fails, there's probably not much we can
-	 * do to recover...
-	 */
-	conn->result = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR);
+		/*
+		 * If error message is passed from RowProcessor, set it into
+		 * PGconn, assume out of memory if not.
+		 */
+		appendPQExpBufferStr(&conn->errorMessage,
+							 libpq_gettext("out of memory for query result\n"));
+
+		/*
+		 * XXX: if PQmakeEmptyPGresult() fails, there's probably not much we can
+		 * do to recover...
+		 */
+		conn->result = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR);
+	}
 	conn->asyncStatus = PGASYNC_READY;
+
 	/* Discard the failed message --- good idea? */
 	conn->inStart = conn->inEnd;
 
diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c
index 892dcbc..a19ee88 100644
--- a/src/interfaces/libpq/fe-protocol3.c
+++ b/src/interfaces/libpq/fe-protocol3.c
@@ -327,6 +327,9 @@ pqParseInput3(PGconn *conn)
 						/* Read another tuple of a normal query response */
 						if (getAnotherTuple(conn, msgLength))
 							return;
+
+						/* getAnotherTuple() moves inStart itself */
+						continue;
 					}
 					else if (conn->result != NULL &&
 							 conn->result->resultStatus == PGRES_FATAL_ERROR)
@@ -613,33 +616,22 @@ failure:
 
 /*
  * parseInput subroutine to read a 'D' (row data) message.
- * We add another tuple to the existing PGresult structure.
+ * It fills rowbuf with column pointers and then calls row processor.
  * Returns: 0 if completed message, EOF if error or not enough data yet.
  *
  * Note that if we run out of data, we have to suspend and reprocess
- * the message after more data is received.  We keep a partially constructed
- * tuple in conn->curTuple, and avoid reallocating already-allocated storage.
+ * the message after more data is received.
  */
 static int
 getAnotherTuple(PGconn *conn, int msgLength)
 {
 	PGresult   *result = conn->result;
 	int			nfields = result->numAttributes;
-	PGresAttValue *tup;
+	PGrowValue  *rowbuf;
 	int			tupnfields;		/* # fields from tuple */
 	int			vlen;			/* length of the current field value */
 	int			i;
-
-	/* Allocate tuple space if first time for this data message */
-	if (conn->curTuple == NULL)
-	{
-		conn->curTuple = (PGresAttValue *)
-			pqResultAlloc(result, nfields * sizeof(PGresAttValue), TRUE);
-		if (conn->curTuple == NULL)
-			goto outOfMemory;
-		MemSet(conn->curTuple, 0, nfields * sizeof(PGresAttValue));
-	}
-	tup = conn->curTuple;
+	int			rp;
 
 	/* Get the field count and make sure it's what we expect */
 	if (pqGetInt(&tupnfields, 2, conn))
@@ -652,52 +644,88 @@ getAnotherTuple(PGconn *conn, int msgLength)
 				 libpq_gettext("unexpected field count in \"D\" message\n"));
 		pqSaveErrorResult(conn);
 		/* Discard the failed message by pretending we read it */
-		conn->inCursor = conn->inStart + 5 + msgLength;
+		conn->inStart += 5 + msgLength;
 		return 0;
 	}
 
+	/* resize row buffer if needed */
+	rowbuf = conn->rowBuf;
+	if (nfields > conn->rowBufLen)
+	{
+		rowbuf = realloc(conn->rowBuf, nfields * sizeof(PGrowValue));
+		if (!rowbuf)
+		{
+			goto outOfMemory1;
+		}
+		conn->rowBuf = rowbuf;
+		conn->rowBufLen = nfields;
+	}
+
 	/* Scan the fields */
 	for (i = 0; i < nfields; i++)
 	{
 		/* get the value length */
 		if (pqGetInt(&vlen, 4, conn))
-			return EOF;
+			goto protocolError;
 		if (vlen == -1)
-		{
-			/* null field */
-			tup[i].value = result->null_field;
-			tup[i].len = NULL_LEN;
-			continue;
-		}
-		if (vlen < 0)
+			vlen = NULL_LEN;
+		else if (vlen < 0)
 			vlen = 0;
-		if (tup[i].value == NULL)
-		{
-			bool		isbinary = (result->attDescs[i].format != 0);
 
-			tup[i].value = (char *) pqResultAlloc(result, vlen + 1, isbinary);
-			if (tup[i].value == NULL)
-				goto outOfMemory;
-		}
-		tup[i].len = vlen;
-		/* read in the value */
-		if (vlen > 0)
-			if (pqGetnchar((char *) (tup[i].value), vlen, conn))
-				return EOF;
-		/* we have to terminate this ourselves */
-		tup[i].value[vlen] = '\0';
+		/*
+		 * rowbuf[i].value always points to the next address of the
+		 * length field even if the value is NULL, to allow safe
+		 * size estimates and data copy.
+		 */
+		rowbuf[i].value = conn->inBuffer + conn->inCursor;
+		rowbuf[i].len = vlen;
+
+		/* Skip to the next length field */
+		if (vlen > 0 && pqSkipnchar(vlen, conn))
+			goto protocolError;
 	}
 
-	/* Success!  Store the completed tuple in the result */
-	if (!pqAddTuple(result, tup))
-		goto outOfMemory;
-	/* and reset for a new message */
-	conn->curTuple = NULL;
+	/* tag the row as parsed, check if correctly */
+	conn->inStart += 5 + msgLength;
+	if (conn->inCursor != conn->inStart)
+		goto protocolError;
 
+	/* Pass the completed row values to rowProcessor */
+	rp = conn->rowProcessor(result, rowbuf, conn->rowProcessorParam);
+	if (rp == 1)
+	{
+		/* everything is good */
+		return 0;
+	}
+	if (rp == 2 && pqIsnonblocking(conn))
+	{
+		/* processor requested early exit */
+		return EOF;
+	}
+
+	/* there was some problem */
+	if (rp == 0)
+	{
+		if (result->rowProcessorErrMsg == NULL)
+			goto outOfMemory2;
+
+		/* use supplied error message */
+		printfPQExpBuffer(&conn->errorMessage, "%s", result->rowProcessorErrMsg);
+	}
+	else
+	{
+		/* row processor messed up */
+		printfPQExpBuffer(&conn->errorMessage,
+						  libpq_gettext("invalid return value from row processor\n"));
+	}
+	pqSaveErrorResult(conn);
 	return 0;
 
-outOfMemory:
+outOfMemory1:
+	/* Discard the failed message by pretending we read it */
+	conn->inStart += 5 + msgLength;
 
+outOfMemory2:
 	/*
 	 * Replace partially constructed result with an error result. First
 	 * discard the old result to try to win back some memory.
@@ -706,9 +734,14 @@ outOfMemory:
 	printfPQExpBuffer(&conn->errorMessage,
 					  libpq_gettext("out of memory for query result\n"));
 	pqSaveErrorResult(conn);
+	return 0;
 
+protocolError:
+	printfPQExpBuffer(&conn->errorMessage,
+					  libpq_gettext("invalid row contents\n"));
+	pqSaveErrorResult(conn);
 	/* Discard the failed message by pretending we read it */
-	conn->inCursor = conn->inStart + 5 + msgLength;
+	conn->inStart += 5 + msgLength;
 	return 0;
 }
 
diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h
index ef26ab9..b7370e2 100644
--- a/src/interfaces/libpq/libpq-fe.h
+++ b/src/interfaces/libpq/libpq-fe.h
@@ -149,6 +149,17 @@ typedef struct pgNotify
 	struct pgNotify *next;		/* list link */
 } PGnotify;
 
+/* PGrowValue points a column value of in network buffer.
+ * Value is a string without null termination and length len.
+ * NULL is represented as len < 0, value points then to place
+ * where value would have been.
+ */
+typedef struct pgRowValue
+{
+	int			len;			/* length in bytes of the value */
+	char	   *value;			/* actual value, without null termination */
+} PGrowValue;
+
 /* Function types for notice-handling callbacks */
 typedef void (*PQnoticeReceiver) (void *arg, const PGresult *res);
 typedef void (*PQnoticeProcessor) (void *arg, const char *message);
@@ -416,6 +427,39 @@ extern PGPing PQping(const char *conninfo);
 extern PGPing PQpingParams(const char *const * keywords,
 			 const char *const * values, int expand_dbname);
 
+/*
+ * Typedef for alternative row processor.
+ *
+ * Columns array will contain PQnfields() entries, each one
+ * pointing to particular column data in network buffer.
+ * This function is supposed to copy data out from there
+ * and store somewhere.  NULL is signified with len<0.
+ *
+ * This function must return 1 for success and must return 0 for
+ * failure and may set error message by PQsetRowProcessorErrMsg.  It
+ * is assumed by caller as out of memory when the error message is not
+ * set on failure. This function is assumed not to throw any
+ * exception.
+ */
+typedef int (*PQrowProcessor)(PGresult *res, PGrowValue *columns,
+                              void *param);
+
+/*
+ * Set alternative row data processor for PGconn.
+ *
+ * By registering this function, pg_result disables its own result
+ * store and calls it for rows one by one.
+ *
+ * func is row processor function. See the typedef RowProcessor.
+ *
+ * rowProcessorParam is the contextual variable that passed to
+ * RowProcessor.
+ */
+extern void PQsetRowProcessor(PGconn *conn, PQrowProcessor func,
+								   void *rowProcessorParam);
+extern PQrowProcessor PQgetRowProcessor(PGconn *conn, void **param);
+extern int  PQskipResult(PGconn *conn, int skipAll);
+
 /* Force the write buffer to be written (or at least try) */
 extern int	PQflush(PGconn *conn);
 
@@ -454,6 +498,7 @@ extern char *PQcmdTuples(PGresult *res);
 extern char *PQgetvalue(const PGresult *res, int tup_num, int field_num);
 extern int	PQgetlength(const PGresult *res, int tup_num, int field_num);
 extern int	PQgetisnull(const PGresult *res, int tup_num, int field_num);
+extern void	PQsetRowProcessorErrMsg(PGresult *res, const char *msg);
 extern int	PQnparams(const PGresult *res);
 extern Oid	PQparamtype(const PGresult *res, int param_num);
 
diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h
index 987311e..1fc5aab 100644
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -209,6 +209,9 @@ struct pg_result
 	PGresult_data *curBlock;	/* most recently allocated block */
 	int			curOffset;		/* start offset of free space in block */
 	int			spaceLeft;		/* number of free bytes remaining in block */
+
+	/* temp etorage for message from row processor callback */
+	char	   *rowProcessorErrMsg;
 };
 
 /* PGAsyncStatusType defines the state of the query-execution state machine */
@@ -398,7 +401,6 @@ struct pg_conn
 
 	/* Status for asynchronous result construction */
 	PGresult   *result;			/* result being constructed */
-	PGresAttValue *curTuple;	/* tuple currently being read */
 
 #ifdef USE_SSL
 	bool		allow_ssl_try;	/* Allowed to try SSL negotiation */
@@ -443,6 +445,14 @@ struct pg_conn
 
 	/* Buffer for receiving various parts of messages */
 	PQExpBufferData workBuffer; /* expansible string */
+
+	/*
+	 * Read column data from network buffer.
+	 */
+	PQrowProcessor rowProcessor;/* Function pointer */
+	void *rowProcessorParam;	/* Contextual parameter for rowProcessor */
+	PGrowValue *rowBuf;			/* Buffer for passing values to rowProcessor */
+	int rowBufLen;				/* Number of columns allocated in rowBuf */
 };
 
 /* PGcancel stores all data necessary to cancel a connection. A copy of this
@@ -560,6 +570,7 @@ extern int	pqGets(PQExpBuffer buf, PGconn *conn);
 extern int	pqGets_append(PQExpBuffer buf, PGconn *conn);
 extern int	pqPuts(const char *s, PGconn *conn);
 extern int	pqGetnchar(char *s, size_t len, PGconn *conn);
+extern int	pqSkipnchar(size_t len, PGconn *conn);
 extern int	pqPutnchar(const char *s, size_t len, PGconn *conn);
 extern int	pqGetInt(int *result, size_t bytes, PGconn *conn);
 extern int	pqPutInt(int value, size_t bytes, PGconn *conn);