libpq_rowproc_2012-03-07v2.diff

*** a/src/interfaces/libpq/exports.txt
--- b/src/interfaces/libpq/exports.txt
***************
*** 160,162 **** PQconnectStartParams      157
--- 160,165 ----
  PQping                    158
  PQpingParams              159
  PQlibVersion              160
+ PQsetRowProcessor	  	  161
+ PQgetRowProcessor	  	  162
+ PQskipResult		  	  163
*** a/src/interfaces/libpq/fe-connect.c
--- b/src/interfaces/libpq/fe-connect.c
***************
*** 2693,2698 **** makeEmptyPGconn(void)
--- 2693,2701 ----
  	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,2718 **** makeEmptyPGconn(void)
--- 2714,2726 ----
  	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,2819 **** freePGconn(PGconn *conn)
--- 2822,2829 ----
  		free(conn->inBuffer);
  	if (conn->outBuffer)
  		free(conn->outBuffer);
+ 	if (conn->rowBuf)
+ 		free(conn->rowBuf);
  	termPQExpBuffer(&conn->errorMessage);
  	termPQExpBuffer(&conn->workBuffer);
  
***************
*** 5078,5080 **** PQregisterThreadLock(pgthreadlock_t newhandler)
--- 5088,5091 ----
  
  	return prev;
  }
+ 
*** a/src/interfaces/libpq/fe-exec.c
--- b/src/interfaces/libpq/fe-exec.c
***************
*** 66,71 **** static PGresult *PQexecFinish(PGconn *conn);
--- 66,72 ----
  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);
  
  
  /* ----------------
***************
*** 701,707 **** pqClearAsyncResult(PGconn *conn)
  	if (conn->result)
  		PQclear(conn->result);
  	conn->result = NULL;
- 	conn->curTuple = NULL;
  }
  
  /*
--- 702,707 ----
***************
*** 756,762 **** 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
--- 756,761 ----
***************
*** 828,833 **** pqInternalNotice(const PGNoticeHooks *hooks, const char *fmt,...)
--- 827,921 ----
  }
  
  /*
+  * PQsetRowProcessor
+  *   Set function that copies column data out from network buffer.
+  */
+ void
+ PQsetRowProcessor(PGconn *conn, PQrowProcessor func, void *param)
+ {
+ 	if (!conn)
+ 		return;
+ 
+ 	if (func)
+ 	{
+ 		/* set custom row processor */
+ 		conn->rowProcessor = func;
+ 		conn->rowProcessorParam = param;
+ 	}
+ 	else
+ 	{
+ 		/* set default row processor */
+ 		conn->rowProcessor = pqAddRow;
+ 		conn->rowProcessorParam = conn;
+ 	}
+ }
+ 
+ /*
+  * 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 (!conn)
+ 		return NULL;
+ 
+ 	if (param)
+ 		*param = conn->rowProcessorParam;
+ 
+ 	return conn->rowProcessor;
+ }
+ 
+ /*
+  * pqAddRow
+  *	  add a row to the PGresult structure, growing it if necessary
+  *	  Returns 1 if OK, -1 if error occurred.
+  */
+ static int
+ pqAddRow(PGresult *res, PGrowValue *columns, void *param)
+ {
+ 	PGconn		*conn = param;
+ 	PGresAttValue *tup;
+ 	int			nfields = res->numAttributes;
+ 	int			i;
+ 
+ 	tup = (PGresAttValue *)
+ 		pqResultAlloc(res, nfields * sizeof(PGresAttValue), TRUE);
+ 	if (tup == NULL)
+ 		goto no_memory;
+ 
+ 	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)
+ 				goto no_memory;
+ 
+ 			/* copy and zero-terminate */
+ 			memcpy(tup[i].value, columns[i].value, tup[i].len);
+ 			tup[i].value[tup[i].len] = '\0';
+ 		}
+ 	}
+ 
+ 	if (pqAddTuple(res, tup))
+ 		return 1;
+ 
+ no_memory:
+ 	printfPQExpBuffer(&conn->errorMessage,
+ 					  libpq_gettext("out of memory for query result\n"));
+ 	pqSaveErrorResult(conn);
+ 	return -1;
+ }
+ 
+ /*
   * 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,1229 **** PQsendQueryStart(PGconn *conn)
  
  	/* initialize async result-accumulation state */
  	conn->result = NULL;
- 	conn->curTuple = NULL;
  
  	/* ready to send command message */
  	return true;
--- 1311,1316 ----
***************
*** 1831,1836 **** PQexecFinish(PGconn *conn)
--- 1918,1975 ----
  	return lastResult;
  }
  
+ 
+ /*
+  * Do-nothing row processor for PQskipResult
+  */
+ static int
+ dummyRowProcessor(PGresult *res, PGrowValue *columns, void *param)
+ {
+ 	return 1;
+ }
+ 
+ /*
+  * Exhaust remaining Data Rows in current connection.
+  * 
+  * Exhaust only one resultset 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;
+ 
+ 	if (!conn)
+ 		return 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
*** a/src/interfaces/libpq/fe-misc.c
--- b/src/interfaces/libpq/fe-misc.c
***************
*** 219,224 **** pqGetnchar(char *s, size_t len, PGconn *conn)
--- 219,243 ----
  }
  
  /*
+  * 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
   */
*** a/src/interfaces/libpq/fe-protocol2.c
--- b/src/interfaces/libpq/fe-protocol2.c
***************
*** 569,574 **** pqParseInput2(PGconn *conn)
--- 569,576 ----
  						/* Read another tuple of a normal query response */
  						if (getAnotherTuple(conn, FALSE))
  							return;
+ 						/* getAnotherTuple moves inStart itself */
+ 						continue;
  					}
  					else
  					{
***************
*** 585,590 **** pqParseInput2(PGconn *conn)
--- 587,594 ----
  						/* Read another tuple of a normal query response */
  						if (getAnotherTuple(conn, TRUE))
  							return;
+ 						/* getAnotherTuple moves inStart itself */
+ 						continue;
  					}
  					else
  					{
***************
*** 703,721 **** failure:
  
  /*
   * parseInput subroutine to read a 'B' or 'D' (row data) message.
!  * We add another tuple to the existing PGresult structure.
   * 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.
   */
  static int
  getAnotherTuple(PGconn *conn, bool binary)
  {
  	PGresult   *result = conn->result;
  	int			nfields = result->numAttributes;
! 	PGresAttValue *tup;
  
  	/* the backend sends us a bitmap of which attributes are null */
  	char		std_bitmap[64]; /* used unless it doesn't fit */
--- 707,724 ----
  
  /*
   * parseInput subroutine to read a 'B' or 'D' (row data) message.
!  * 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.
   */
  static int
  getAnotherTuple(PGconn *conn, bool binary)
  {
  	PGresult   *result = conn->result;
  	int			nfields = result->numAttributes;
! 	PGrowValue  *rowbuf;
  
  	/* the backend sends us a bitmap of which attributes are null */
  	char		std_bitmap[64]; /* used unless it doesn't fit */
***************
*** 726,754 **** getAnotherTuple(PGconn *conn, bool binary)
  	int			bitmap_index;	/* Its index */
  	int			bitcnt;			/* number of bits examined in current byte */
  	int			vlen;			/* length of the current field value */
  
! 	result->binary = binary;
! 
! 	/* 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));
! 
! 		/*
! 		 * 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;
  		}
  	}
- 	tup = conn->curTuple;
  
  	/* Get the null-value bitmap */
  	nbytes = (nfields + BITS_PER_BYTE - 1) / BITS_PER_BYTE;
--- 729,760 ----
  	int			bitmap_index;	/* Its index */
  	int			bitcnt;			/* number of bits examined in current byte */
  	int			vlen;			/* length of the current field value */
+ 	const char *errmsg = libpq_gettext("unknown error\n");
  
! 	/* resize row buffer if needed */
! 	if (nfields > conn->rowBufLen)
  	{
! 		rowbuf = realloc(conn->rowBuf, nfields * sizeof(PGrowValue));
! 		if (!rowbuf)
  		{
! 			errmsg = libpq_gettext("out of memory for query result\n");
! 			goto error_clearresult;
  		}
+ 		conn->rowBuf = rowbuf;
+ 		conn->rowBufLen = nfields;
+ 	}
+ 	else
+ 	{
+ 		rowbuf = conn->rowBuf;
+ 	}
+ 
+ 	result->binary = binary;
+ 
+ 	if (binary)
+ 	{
+ 		for (i = 0; i < nfields; i++)
+ 			result->attDescs[i].format = 1;
  	}
  
  	/* Get the null-value bitmap */
  	nbytes = (nfields + BITS_PER_BYTE - 1) / BITS_PER_BYTE;
***************
*** 757,767 **** getAnotherTuple(PGconn *conn, bool binary)
  	{
  		bitmap = (char *) malloc(nbytes);
  		if (!bitmap)
! 			goto outOfMemory;
  	}
  
  	if (pqGetnchar(bitmap, nbytes, conn))
! 		goto EOFexit;
  
  	/* Scan the fields */
  	bitmap_index = 0;
--- 763,777 ----
  	{
  		bitmap = (char *) malloc(nbytes);
  		if (!bitmap)
! 		{
! 			errmsg = libpq_gettext("out of memory for query result\n");
! 			goto error_clearresult;
! 		}
  	}
  
  	if (pqGetnchar(bitmap, nbytes, conn))
! 		goto error_clearresult;
! 
  
  	/* Scan the fields */
  	bitmap_index = 0;
***************
*** 771,804 **** 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;
! 		}
  		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';
  		}
  		/* advance the bitmap stuff */
  		bitcnt++;
  		if (bitcnt == BITS_PER_BYTE)
--- 781,809 ----
  	for (i = 0; i < nfields; i++)
  	{
  		if (!(bmap & 0200))
! 			vlen = NULL_LEN;
! 		else if (pqGetInt(&vlen, 4, conn))
! 				goto EOFexit;
  		else
  		{
  			if (!binary)
  				vlen = vlen - 4;
  			if (vlen < 0)
  				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,843 **** 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;
  
! 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"));
  
  	/*
  	 * 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;
  
--- 816,869 ----
  			bmap <<= 1;
  	}
  
  	if (bitmap != std_bitmap)
  		free(bitmap);
! 	bitmap = NULL;
  
! 	/* tag the row as parsed */
! 	conn->inStart = conn->inCursor;
  
+ 	/* Pass the completed row values to rowProcessor */
+ 	switch (conn->rowProcessor(result, rowbuf, conn->rowProcessorParam))
+ 	{
+ 		case 1:
+ 			/* everything is good */
+ 			return 0;
+ 
+ 		case 0:
+ 			/* processor requested early exit */
+ 			return EOF;
+ 			
+ 		case -1:
+ 			/* if row processor already set error, then simply exit */
+ 			if (conn->result->resultStatus != PGRES_TUPLES_OK)
+ 				return EOF;
+ 			errmsg = libpq_gettext("error in row processor\n");
+ 			break;
+ 
+ 		default:
+ 			/* Illegal return code */
+ 			errmsg = libpq_gettext("invalid return value from row processor\n");
+ 			break;
+ 	}
+ 
+ error_clearresult:
  	/*
  	 * we do NOT use pqSaveErrorResult() here, because of the likelihood that
  	 * there's not enough memory to concatenate messages...
  	 */
  	pqClearAsyncResult(conn);
  
+ 	printfPQExpBuffer(&conn->errorMessage, "%s", errmsg);
+ 	
  	/*
  	 * 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;
  
*** a/src/interfaces/libpq/fe-protocol3.c
--- b/src/interfaces/libpq/fe-protocol3.c
***************
*** 327,332 **** pqParseInput3(PGconn *conn)
--- 327,335 ----
  						/* 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,659 **** failure:
  
  /*
   * parseInput subroutine to read a 'D' (row data) message.
!  * We add another tuple to the existing PGresult structure.
!  * 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.
   */
  static int
  getAnotherTuple(PGconn *conn, int msgLength)
  {
  	PGresult   *result = conn->result;
  	int			nfields = result->numAttributes;
! 	PGresAttValue *tup;
  	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;
  
  	/* Get the field count and make sure it's what we expect */
  	if (pqGetInt(&tupnfields, 2, conn))
! 		return EOF;
  
  	if (tupnfields != nfields)
  	{
! 		/* Replace partially constructed result with an error result */
! 		printfPQExpBuffer(&conn->errorMessage,
! 				 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;
! 		return 0;
  	}
  
  	/* Scan the fields */
--- 616,661 ----
  
  /*
   * parseInput subroutine to read a 'D' (row data) message.
!  * It fills rowbuf with column pointers and then calls row processor.
!  * Returns: 0 if completed message, EOF to cancel parsing.
   */
  static int
  getAnotherTuple(PGconn *conn, int msgLength)
  {
  	PGresult   *result = conn->result;
  	int			nfields = result->numAttributes;
! 	PGrowValue  *rowbuf;
  	int			tupnfields;		/* # fields from tuple */
  	int			vlen;			/* length of the current field value */
  	int			i;
! 	const char *errmsg = libpq_gettext("unknown error\n");
  
  	/* Get the field count and make sure it's what we expect */
  	if (pqGetInt(&tupnfields, 2, conn))
! 	{
! 		/* Whole the message must be loaded on the buffer here */
! 		errmsg = libpq_gettext("protocol error\n");
! 		goto error_and_forward;
! 	}
  
  	if (tupnfields != nfields)
  	{
! 		errmsg = libpq_gettext("unexpected field count in \"D\" message\n");
! 		goto error_and_forward;
! 	}
! 
! 	/* resize row buffer if needed */
! 	rowbuf = conn->rowBuf;
! 	if (nfields > conn->rowBufLen)
! 	{
! 		rowbuf = realloc(conn->rowBuf, nfields * sizeof(PGrowValue));
! 		if (!rowbuf)
! 		{
! 			errmsg = libpq_gettext("out of memory for query result\n");
! 			goto error_and_forward;
! 		}
! 		conn->rowBuf = rowbuf;
! 		conn->rowBufLen = nfields;
  	}
  
  	/* Scan the fields */
***************
*** 661,714 **** getAnotherTuple(PGconn *conn, int msgLength)
  	{
  		/* get the value length */
  		if (pqGetInt(&vlen, 4, conn))
- 			return EOF;
- 		if (vlen == -1)
  		{
! 			/* null field */
! 			tup[i].value = result->null_field;
! 			tup[i].len = NULL_LEN;
! 			continue;
  		}
! 		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';
  	}
  
! 	/* Success!  Store the completed tuple in the result */
! 	if (!pqAddTuple(result, tup))
! 		goto outOfMemory;
! 	/* and reset for a new message */
! 	conn->curTuple = NULL;
  
! 	return 0;
  
! outOfMemory:
  
  	/*
  	 * Replace partially constructed result with an error result. First
  	 * discard the old result to try to win back some memory.
  	 */
  	pqClearAsyncResult(conn);
! 	printfPQExpBuffer(&conn->errorMessage,
! 					  libpq_gettext("out of memory for query result\n"));
  	pqSaveErrorResult(conn);
- 
- 	/* Discard the failed message by pretending we read it */
- 	conn->inCursor = conn->inStart + 5 + msgLength;
  	return 0;
  }
  
--- 663,743 ----
  	{
  		/* get the value length */
  		if (pqGetInt(&vlen, 4, conn))
  		{
! 			/* Whole the message must be loaded on the buffer here */
! 			errmsg = libpq_gettext("protocol error\n");
! 			goto error_and_forward;
  		}
! 
! 		if (vlen == -1)
! 			vlen = NULL_LEN;
! 		else if (vlen < 0)
  			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))
! 		{
! 			/* Whole the message must be loaded on the buffer here */
! 			errmsg = libpq_gettext("protocol error\n");
! 			goto error_and_forward;
  		}
  	}
  
! 	/* tag the row as parsed, check if correctly */
! 	conn->inStart += 5 + msgLength;
! 	if (conn->inCursor != conn->inStart)
! 	{
! 		errmsg = libpq_gettext("invalid row contents\n");
! 		goto error_clearresult;
! 	}
  
! 	/* Pass the completed row values to rowProcessor */
! 	switch (conn->rowProcessor(result, rowbuf, conn->rowProcessorParam))
! 	{
! 		case 1:
! 			/* everything is good */
! 			return 0;
! 
! 		case 0:
! 			/* processor requested early exit - stop parsing without error*/
! 			return EOF;
! 
! 		case -1:
! 			/* if row processor already set error, then simply exit */
! 			if (conn->result->resultStatus != PGRES_TUPLES_OK)
! 				return EOF;
  
! 			/* set generic row processor error */
! 			errmsg = libpq_gettext("error in row processor\n");
! 			goto error_clearresult;
  
+ 		default:
+ 			/* Illegal return code */
+ 			errmsg = libpq_gettext("invalid return value from row processor\n");
+ 			goto error_clearresult;
+ 	}
+ 
+ 
+ error_and_forward:
+ 	/* Discard the failed message by pretending we read it */
+ 	conn->inCursor = conn->inStart + 5 + msgLength;
+ 
+ error_clearresult:
  	/*
  	 * Replace partially constructed result with an error result. First
  	 * discard the old result to try to win back some memory.
  	 */
  	pqClearAsyncResult(conn);
! 	printfPQExpBuffer(&conn->errorMessage, "%s", errmsg);
  	pqSaveErrorResult(conn);
  	return 0;
  }
  
*** a/src/interfaces/libpq/libpq-fe.h
--- b/src/interfaces/libpq/libpq-fe.h
***************
*** 149,154 **** typedef struct pgNotify
--- 149,165 ----
  	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,421 **** extern PGPing PQping(const char *conninfo);
--- 427,464 ----
  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 -1 for failure and the
+  * caller relreases the current PGresult for the case. Returning 0
+  * instructs libpq to exit immediately from the topmost libpq function
+  * without releasing PGresult under work.
+  */
+ 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);
  
*** a/src/interfaces/libpq/libpq-int.h
--- b/src/interfaces/libpq/libpq-int.h
***************
*** 398,404 **** 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 */
--- 398,403 ----
***************
*** 441,446 **** struct pg_conn
--- 440,453 ----
  
  	/* 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
***************
*** 558,563 **** extern int	pqGets(PQExpBuffer buf, PGconn *conn);
--- 565,571 ----
  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);