libpq_rowproc_2012-02-14.diff

*** a/src/interfaces/libpq/exports.txt
--- b/src/interfaces/libpq/exports.txt
***************
*** 160,162 **** PQconnectStartParams      157
--- 160,164 ----
  PQping                    158
  PQpingParams              159
  PQlibVersion              160
+ PQsetRowProcessor	  161
+ PQsetRowProcessorErrMsg	  162
*** 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, void *param, PGrowValue *columns);
  
  
  /* ----------------
***************
*** 160,165 **** PQmakeEmptyPGresult(PGconn *conn, ExecStatusType status)
--- 161,167 ----
  	result->curBlock = NULL;
  	result->curOffset = 0;
  	result->spaceLeft = 0;
+ 	result->rowProcessorErrMsg = NULL;
  
  	if (conn)
  	{
***************
*** 701,707 **** pqClearAsyncResult(PGconn *conn)
  	if (conn->result)
  		PQclear(conn->result);
  	conn->result = NULL;
- 	conn->curTuple = NULL;
  }
  
  /*
--- 703,708 ----
***************
*** 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
--- 757,762 ----
***************
*** 828,833 **** pqInternalNotice(const PGNoticeHooks *hooks, const char *fmt,...)
--- 828,900 ----
  }
  
  /*
+  * 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;
+ }
+ 
+ /*
+  * 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, 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, void *param, PGrowValue *columns)
+ {
+ 	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,1229 **** PQsendQueryStart(PGconn *conn)
  
  	/* initialize async result-accumulation state */
  	conn->result = NULL;
- 	conn->curTuple = NULL;
  
  	/* ready to send command message */
  	return true;
--- 1290,1295 ----
*** 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,754 **** 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 */
  	char	   *bitmap = std_bitmap;
  	int			i;
  	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 */
  
  	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;
--- 707,757 ----
  
  /*
   * 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 */
  	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;
  
! 	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,763 **** getAnotherTuple(PGconn *conn, bool binary)
  	{
  		bitmap = (char *) malloc(nbytes);
  		if (!bitmap)
! 			goto outOfMemory;
  	}
  
  	if (pqGetnchar(bitmap, nbytes, conn))
--- 760,766 ----
  	{
  		bitmap = (char *) malloc(nbytes);
  		if (!bitmap)
! 			goto rowProcessError;
  	}
  
  	if (pqGetnchar(bitmap, nbytes, conn))
***************
*** 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)
--- 774,802 ----
  	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;
  
--- 809,864 ----
  			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 */
! 	rp= conn->rowProcessor(result, conn->rowProcessorParam, rowbuf);
! 	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:
  
  	/* Replace partially constructed result with an error result */
  
! 	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);
  
! 		/*
! 		 * 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;
  
*** 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,645 **** 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))
--- 616,637 ----
  
  /*
   * 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 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, 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;
! 	int			rp;
  
  	/* Get the field count and make sure it's what we expect */
  	if (pqGetInt(&tupnfields, 2, conn))
***************
*** 652,703 **** 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;
  		return 0;
  	}
  
  	/* Scan the fields */
  	for (i = 0; i < nfields; i++)
  	{
  		/* 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.
--- 644,731 ----
  				 libpq_gettext("unexpected field count in \"D\" message\n"));
  		pqSaveErrorResult(conn);
  		/* Discard the failed message by pretending we read it */
! 		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))
! 			goto protocolError;
  		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))
! 			goto protocolError;
  	}
  
! 	/* 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, conn->rowProcessorParam, rowbuf);
+ 	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;
  
! 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,714 **** outOfMemory:
  	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;
  }
  
--- 734,747 ----
  	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->inStart += 5 + msgLength;
  	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,463 ----
  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, void *param,
+ 								PGrowValue *columns);
+ 
+ /*
+  * 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);
+ 
  /* Force the write buffer to be written (or at least try) */
  extern int	PQflush(PGconn *conn);
  
***************
*** 454,459 **** extern char *PQcmdTuples(PGresult *res);
--- 496,502 ----
  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, char *msg);
  extern int	PQnparams(const PGresult *res);
  extern Oid	PQparamtype(const PGresult *res, int param_num);
  
*** a/src/interfaces/libpq/libpq-int.h
--- b/src/interfaces/libpq/libpq-int.h
***************
*** 209,214 **** struct pg_result
--- 209,217 ----
  	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,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 */
--- 401,406 ----
***************
*** 443,448 **** struct pg_conn
--- 445,458 ----
  
  	/* 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,565 **** extern int	pqGets(PQExpBuffer buf, PGconn *conn);
--- 570,576 ----
  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);