libpq_rowproc_20120127.patch

diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt
index 1af8df6..c47af3a 100644
--- a/src/interfaces/libpq/exports.txt
+++ b/src/interfaces/libpq/exports.txt
@@ -160,3 +160,6 @@ PQconnectStartParams      157
 PQping                    158
 PQpingParams              159
 PQlibVersion              160
+PQregisterRowProcessor	  161
+PQgetRowProcessorParam	  163
+PQsetRowProcessorErrMes	  164
diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c
index d454538..93803d5 100644
--- a/src/interfaces/libpq/fe-connect.c
+++ b/src/interfaces/libpq/fe-connect.c
@@ -2692,6 +2692,7 @@ makeEmptyPGconn(void)
 	conn->allow_ssl_try = true;
 	conn->wait_ssl_try = false;
 #endif
+	conn->rowProcessor = NULL;
 
 	/*
 	 * We try to send at least 8K at a time, which is the usual size of pipe
@@ -5076,3 +5077,10 @@ PQregisterThreadLock(pgthreadlock_t newhandler)
 
 	return prev;
 }
+
+void
+PQregisterRowProcessor(PGconn *conn, RowProcessor func, void *param)
+{
+	conn->rowProcessor = func;
+	conn->rowProcessorParam = param;
+}
diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c
index b743566..5d78b39 100644
--- a/src/interfaces/libpq/fe-exec.c
+++ b/src/interfaces/libpq/fe-exec.c
@@ -66,7 +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 void *pqAddTuple(PGresult *res, PGresAttValue *columns);
 
 /* ----------------
  * Space management for PGresult.
@@ -160,6 +160,9 @@ PQmakeEmptyPGresult(PGconn *conn, ExecStatusType status)
 	result->curBlock = NULL;
 	result->curOffset = 0;
 	result->spaceLeft = 0;
+	result->rowProcessor = pqAddTuple;
+	result->rowProcessorParam = NULL;
+	result->rowProcessorErrMes = NULL;
 
 	if (conn)
 	{
@@ -194,6 +197,12 @@ PQmakeEmptyPGresult(PGconn *conn, ExecStatusType status)
 			}
 			result->nEvents = conn->nEvents;
 		}
+
+		if (conn->rowProcessor)
+		{
+			result->rowProcessor = conn->rowProcessor;
+			result->rowProcessorParam = conn->rowProcessorParam;
+		}
 	}
 	else
 	{
@@ -445,7 +454,7 @@ PQsetvalue(PGresult *res, int tup_num, int field_num, char *value, int len)
 		}
 
 		/* add it to the array */
-		if (!pqAddTuple(res, tup))
+		if (pqAddTuple(res, tup) == NULL)
 			return FALSE;
 	}
 
@@ -701,7 +710,6 @@ pqClearAsyncResult(PGconn *conn)
 	if (conn->result)
 		PQclear(conn->result);
 	conn->result = NULL;
-	conn->curTuple = NULL;
 }
 
 /*
@@ -756,7 +764,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
@@ -829,12 +836,17 @@ pqInternalNotice(const PGNoticeHooks *hooks, const char *fmt,...)
 
 /*
  * 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
+ *	  add a row to the PGresult structure, growing it if necessary
+ *	  Returns the pointer to the new tuple if OK, NULL if not enough
+ *	  memory to add the row.
  */
-int
-pqAddTuple(PGresult *res, PGresAttValue *tup)
+void *
+pqAddTuple(PGresult *res, PGresAttValue *columns)
 {
+	PGresAttValue *tup;
+	int nfields = res->numAttributes;
+	int i;
+
 	if (res->ntups >= res->tupArrSize)
 	{
 		/*
@@ -858,13 +870,39 @@ pqAddTuple(PGresult *res, PGresAttValue *tup)
 			newTuples = (PGresAttValue **)
 				realloc(res->tuples, newSize * sizeof(PGresAttValue *));
 		if (!newTuples)
-			return FALSE;		/* malloc or realloc failed */
+			return NULL;		/* malloc or realloc failed */
 		res->tupArrSize = newSize;
 		res->tuples = newTuples;
 	}
+
+	tup = (PGresAttValue *)
+		pqResultAlloc(res, nfields * sizeof(PGresAttValue), TRUE);
+	if (tup == NULL) return NULL;
+	memcpy(tup, columns, nfields * sizeof(PGresAttValue));
+
+	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 NULL;
+			memcpy(tup[i].value, columns[i].value, tup[i].len);
+			/* We have to terminate this ourselves */
+			tup[i].value[tup[i].len] = '\0';
+		}
+	}
+
 	res->tuples[res->ntups] = tup;
 	res->ntups++;
-	return TRUE;
+	return tup;
 }
 
 /*
@@ -1223,7 +1261,6 @@ PQsendQueryStart(PGconn *conn)
 
 	/* initialize async result-accumulation state */
 	conn->result = NULL;
-	conn->curTuple = NULL;
 
 	/* ready to send command message */
 	return true;
@@ -2822,6 +2859,35 @@ PQgetisnull(const PGresult *res, int tup_num, int field_num)
 		return 0;
 }
 
+/* PQgetAddRowProcessorParam
+ *	Get the pointer to the contextual parameter from PGresult which is
+ *	registered to PGconn by PQregisterRowProcessor
+ */
+void *
+PQgetRowProcessorParam(const PGresult *res)
+{
+	if (!res)
+		return NULL;
+	return res->rowProcessorParam;
+}
+
+/* PQsetRowProcessorErrMes
+ *	Set the error message pass back to the caller of RowProcessor.
+ *
+ *  mes must be a malloc'ed memory block and it will be released by
+ *  the caller of RowProcessor.  You can replace the previous message
+ *  by alternative mes, or clear it with NULL. The previous one will
+ *  be freed internally.
+ */
+void
+PQsetRowProcessorErrMes(PGresult *res, char *mes)
+{
+	/* Free existing message */
+	if (res->rowProcessorErrMes)
+		free(res->rowProcessorErrMes);
+	res->rowProcessorErrMes = mes;
+}
+
 /* PQnparams:
  *	returns the number of input parameters of a prepared statement.
  */
diff --git a/src/interfaces/libpq/fe-misc.c b/src/interfaces/libpq/fe-misc.c
index ce0eac3..546534a 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
  */
@@ -238,6 +257,7 @@ pqPutnchar(const char *s, size_t len, PGconn *conn)
 	return 0;
 }
 
+
 /*
  * pqGetInt
  *	read a 2 or 4 byte integer and convert from network byte order
diff --git a/src/interfaces/libpq/fe-protocol2.c b/src/interfaces/libpq/fe-protocol2.c
index a7c3899..9abbb29 100644
--- a/src/interfaces/libpq/fe-protocol2.c
+++ b/src/interfaces/libpq/fe-protocol2.c
@@ -715,7 +715,7 @@ getAnotherTuple(PGconn *conn, bool binary)
 {
 	PGresult   *result = conn->result;
 	int			nfields = result->numAttributes;
-	PGresAttValue *tup;
+	PGresAttValue tup[result->numAttributes];
 
 	/* the backend sends us a bitmap of which attributes are null */
 	char		std_bitmap[64]; /* used unless it doesn't fit */
@@ -729,26 +729,11 @@ getAnotherTuple(PGconn *conn, bool binary)
 
 	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 +742,7 @@ getAnotherTuple(PGconn *conn, bool binary)
 	{
 		bitmap = (char *) malloc(nbytes);
 		if (!bitmap)
-			goto outOfMemory;
+			goto rowProcessError;
 	}
 
 	if (pqGetnchar(bitmap, nbytes, conn))
@@ -785,19 +770,17 @@ getAnotherTuple(PGconn *conn, bool 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;
-			}
+
+			/*
+			 * Buffer content may be shifted on reloading data. So we must
+			 * set the pointer to the value on every scan.
+			 */
+			tup[i].value = conn->inBuffer + conn->inCursor;
 			tup[i].len = vlen;
-			/* read in the value */
+			/* Skip the value */
 			if (vlen > 0)
-				if (pqGetnchar((char *) (tup[i].value), vlen, conn))
+				if (pqSkipnchar(vlen, conn))
 					goto EOFexit;
-			/* we have to terminate this ourselves */
-			tup[i].value[vlen] = '\0';
 		}
 		/* advance the bitmap stuff */
 		bitcnt++;
@@ -812,16 +795,15 @@ getAnotherTuple(PGconn *conn, bool binary)
 	}
 
 	/* Success!  Store the completed tuple in the result */
-	if (!pqAddTuple(result, tup))
-		goto outOfMemory;
-	/* and reset for a new message */
-	conn->curTuple = NULL;
+	if (!result->rowProcessor(result, tup))
+		goto rowProcessError;
 
 	if (bitmap != std_bitmap)
 		free(bitmap);
 	return 0;
 
-outOfMemory:
+rowProcessError:
+	
 	/* Replace partially constructed result with an error result */
 
 	/*
@@ -829,8 +811,21 @@ outOfMemory:
 	 * there's not enough memory to concatenate messages...
 	 */
 	pqClearAsyncResult(conn);
-	printfPQExpBuffer(&conn->errorMessage,
-					  libpq_gettext("out of memory for query result\n"));
+	resetPQExpBuffer(&conn->errorMessage);
+
+	/*
+	 * If error message is passed from addTupleFunc, set it into
+	 * PGconn, assume out of memory if not.
+	 */
+	appendPQExpBufferStr(&conn->errorMessage,
+						 libpq_gettext(result->rowProcessorErrMes ?
+									   result->rowProcessorErrMes :
+									   "out of memory for query result\n"));
+	if (result->rowProcessorErrMes)
+	{
+		free(result->rowProcessorErrMes);
+		result->rowProcessorErrMes = NULL;
+	}
 
 	/*
 	 * XXX: if PQmakeEmptyPGresult() fails, there's probably not much we can
diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c
index 892dcbc..18342c7 100644
--- a/src/interfaces/libpq/fe-protocol3.c
+++ b/src/interfaces/libpq/fe-protocol3.c
@@ -625,22 +625,12 @@ getAnotherTuple(PGconn *conn, int msgLength)
 {
 	PGresult   *result = conn->result;
 	int			nfields = result->numAttributes;
-	PGresAttValue *tup;
+	PGresAttValue tup[result->numAttributes];
 	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;
@@ -671,40 +661,46 @@ getAnotherTuple(PGconn *conn, int msgLength)
 		}
 		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 */
+		/*
+		 * Buffer content may be shifted on reloading data. So we must
+		 * set the pointer to the value every scan.
+		 */
+		tup[i].value = conn->inBuffer + conn->inCursor;
+ 		tup[i].len = vlen;
 		if (vlen > 0)
-			if (pqGetnchar((char *) (tup[i].value), vlen, conn))
+			if (pqSkipnchar(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;
-
+	if (!result->rowProcessor(result, tup))
+		goto rowProcessError;
+	
 	return 0;
 
-outOfMemory:
+rowProcessError:
 
 	/*
 	 * 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"));
+	resetPQExpBuffer(&conn->errorMessage);
+
+	/*
+	 * If error message is passed from addTupleFunc, set it into
+	 * PGconn, assume out of memory if not.
+	 */
+	appendPQExpBufferStr(&conn->errorMessage,
+						 libpq_gettext(result->rowProcessorErrMes ?
+									   result->rowProcessorErrMes : 
+									   "out of memory for query result\n"));
+	if (result->rowProcessorErrMes)
+	{
+		free(result->rowProcessorErrMes);
+		result->rowProcessorErrMes = NULL;
+	}
 	pqSaveErrorResult(conn);
 
 	/* Discard the failed message by pretending we read it */
diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h
index ef26ab9..0931211 100644
--- a/src/interfaces/libpq/libpq-fe.h
+++ b/src/interfaces/libpq/libpq-fe.h
@@ -149,6 +149,15 @@ typedef struct pgNotify
 	struct pgNotify *next;		/* list link */
 } PGnotify;
 
+/* PGresAttValue represents a value of one tuple field in string form.
+   NULL is represented as len < 0. Otherwise value points to a string
+   without null termination of the length of len. */
+typedef struct pgresAttValue
+{
+	int			len;			/* length in bytes of the value */
+	char	   *value;			/* actual value, without null termination */
+} PGresAttValue;
+
 /* Function types for notice-handling callbacks */
 typedef void (*PQnoticeReceiver) (void *arg, const PGresult *res);
 typedef void (*PQnoticeProcessor) (void *arg, const char *message);
@@ -416,6 +425,31 @@ 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.
+ *
+ * This function must return non-NULL value for success and must
+ * return NULL for failure and may set error message by
+ * PQsetRowProcessorErrMes. It is assumed by caller as out of memory
+ * when the error message is NULL on failure. This function is assumed
+ * not to throw any exception.
+ */
+typedef void *(*RowProcessor)(PGresult *res, PGresAttValue *columns);
+
+/*
+ * Register alternative result store function to 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 can be get with
+ * PQgetRowProcessorParam in RowProcessor.
+ */
+extern void PQregisterRowProcessor(PGconn *conn, RowProcessor func,
+								   void *rowProcessorParam);
+
 /* Force the write buffer to be written (or at least try) */
 extern int	PQflush(PGconn *conn);
 
@@ -454,6 +488,8 @@ 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 *PQgetRowProcessorParam(const PGresult *res);
+extern void	PQsetRowProcessorErrMes(PGresult *res, char *mes);
 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 d967d60..51ac927 100644
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -134,12 +134,6 @@ typedef struct pgresParamDesc
 
 #define NULL_LEN		(-1)	/* pg_result len for NULL value */
 
-typedef struct pgresAttValue
-{
-	int			len;			/* length in bytes of the value */
-	char	   *value;			/* actual value, plus terminating zero byte */
-} PGresAttValue;
-
 /* Typedef for message-field list entries */
 typedef struct pgMessageField
 {
@@ -209,6 +203,11 @@ 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 */
+
+	RowProcessor rowProcessor;  /* Result row processor handler. See
+								 * RowProcessor for details. */
+	void *rowProcessorParam;    /* Contextual parameter for rowProcessor */
+	char *rowProcessorErrMes;   /* Error message from rowProcessor */
 };
 
 /* PGAsyncStatusType defines the state of the query-execution state machine */
@@ -398,7 +397,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 +441,13 @@ struct pg_conn
 
 	/* Buffer for receiving various parts of messages */
 	PQExpBufferData workBuffer; /* expansible string */
+
+    /* Tuple store handler. The two fields below is copied to newly
+	 * created PGresult if rowProcessor is not NULL. Use default
+	 * function if NULL. */
+	RowProcessor rowProcessor;   /* Result row processor. See
+								  * RowProcessor for details. */
+	void *rowProcessorParam;     /* Contextual parameter for rowProcessor */
 };
 
 /* PGcancel stores all data necessary to cancel a connection. A copy of this
@@ -507,7 +512,6 @@ extern void
 pqInternalNotice(const PGNoticeHooks *hooks, const char *fmt,...)
 /* This lets gcc check the format string for consistency. */
 __attribute__((format(PG_PRINTF_ATTRIBUTE, 2, 3)));
-extern int	pqAddTuple(PGresult *res, PGresAttValue *tup);
 extern void pqSaveMessageField(PGresult *res, char code,
 				   const char *value);
 extern void pqSaveParameterStatus(PGconn *conn, const char *name,
@@ -560,6 +564,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);