diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt index 1af8df6..5ed083c 100644 --- a/src/interfaces/libpq/exports.txt +++ b/src/interfaces/libpq/exports.txt @@ -160,3 +160,5 @@ PQconnectStartParams 157 PQping 158 PQpingParams 159 PQlibVersion 160 +PQregisterRowProcessor 161 +PQsetRowProcessorErrMes 162 diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index d454538..4fe2f41 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -2692,6 +2692,8 @@ makeEmptyPGconn(void) conn->allow_ssl_try = true; conn->wait_ssl_try = false; #endif + conn->rowProcessor = pqAddRow; + conn->rowProcessorParam = NULL; /* * We try to send at least 8K at a time, which is the usual size of pipe @@ -5076,3 +5078,10 @@ PQregisterThreadLock(pgthreadlock_t newhandler) return prev; } + +void +PQregisterRowProcessor(PGconn *conn, RowProcessor func, void *param) +{ + conn->rowProcessor = (func ? func : pqAddRow); + conn->rowProcessorParam = param; +} diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index b743566..82914fd 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 pqAddTuple(PGresult *res, PGresAttValue *tup); /* ---------------- @@ -160,6 +161,9 @@ PQmakeEmptyPGresult(PGconn *conn, ExecStatusType status) result->curBlock = NULL; result->curOffset = 0; result->spaceLeft = 0; + result->rowProcessor = pqAddRow; + result->rowProcessorParam = NULL; + result->rowProcessorErrMes = NULL; if (conn) { @@ -194,6 +198,10 @@ PQmakeEmptyPGresult(PGconn *conn, ExecStatusType status) } result->nEvents = conn->nEvents; } + + /* copy row processor settings */ + result->rowProcessor = conn->rowProcessor; + result->rowProcessorParam = conn->rowProcessorParam; } else { @@ -701,7 +709,6 @@ pqClearAsyncResult(PGconn *conn) if (conn->result) PQclear(conn->result); conn->result = NULL; - conn->curTuple = NULL; } /* @@ -756,7 +763,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,9 +834,52 @@ pqInternalNotice(const PGNoticeHooks *hooks, const char *fmt,...) } /* + * 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. + */ +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; + + 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 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 + * 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. */ int pqAddTuple(PGresult *res, PGresAttValue *tup) @@ -1223,7 +1272,6 @@ PQsendQueryStart(PGconn *conn) /* initialize async result-accumulation state */ conn->result = NULL; - conn->curTuple = NULL; /* ready to send command message */ return true; @@ -2822,6 +2870,30 @@ PQgetisnull(const PGresult *res, int tup_num, int field_num) return 0; } +/* PQsetRowProcessorErrMes + * 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 +PQsetRowProcessorErrMes(PGresult *res, char *mes) +{ + /* Free existing message */ + if (res->rowProcessorErrMes) + free(res->rowProcessorErrMes); + + /* + * Set the duped message if mes is not NULL. Failure of strdup + * will be handled as 'Out of memory' by the caller of the + * RowProcessor. + */ + if (mes) + res->rowProcessorErrMes = strdup(mes); + else + res->rowProcessorErrMes = NULL; +} + /* 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..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..496c42e 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; + PGrowValue rowval[result->numAttributes + 1]; /* 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)) @@ -771,34 +756,31 @@ 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'; } + + /* + * Buffer content may be shifted on reloading additional + * data. So we must set all pointers on every scan. + * + * rowval[i].value always points to the next address of the + * length field even if the value length is zero or the value + * is NULL for the access safety. + */ + rowval[i].value = conn->inBuffer + conn->inCursor; + rowval[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,17 +793,33 @@ 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; + /* + * Set rowval[nfields] for the access safety. We can estimate the + * length of the buffer to store by + * + * rowval[nfields].value - rowval[0].value - 4 * (# of non-nulls)). + */ + rowval[nfields].value = conn->inBuffer + conn->inCursor; + rowval[nfields].len = NULL_LEN; + + /* Success! Pass the completed row values to rowProcessor */ + if (!result->rowProcessor(result, result->rowProcessorParam, rowval)) + goto rowProcessError; + + /* Free garbage message. */ + if (result->rowProcessorErrMes) + { + free(result->rowProcessorErrMes); + result->rowProcessorErrMes = NULL; + } if (bitmap != std_bitmap) free(bitmap); + return 0; -outOfMemory: +rowProcessError: + /* Replace partially constructed result with an error result */ /* @@ -829,8 +827,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 RowProcessor, 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..b7c6118 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; + PGrowValue rowval[result->numAttributes + 1]; 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; @@ -663,48 +653,70 @@ getAnotherTuple(PGconn *conn, int msgLength) 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 = NULL_LEN; + else if (vlen < 0) vlen = 0; - if (tup[i].value == NULL) - { - bool isbinary = (result->attDescs[i].format != 0); + + /* + * Buffer content may be shifted on reloading additional + * data. So we must set all pointers on every scan. + * + * rowval[i].value always points to the next address of the + * length field even if the value length is zero or the value + * is NULL for the access safety. + */ + rowval[i].value = conn->inBuffer + conn->inCursor; + rowval[i].len = vlen; - 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'; + /* Skip to the next length field */ + if (vlen > 0 && pqSkipnchar(vlen, conn)) + return EOF; } - /* Success! Store the completed tuple in the result */ - if (!pqAddTuple(result, tup)) - goto outOfMemory; - /* and reset for a new message */ - conn->curTuple = NULL; + /* + * Set rowval[nfields] for the access safety. We can estimate the + * length of the buffer to store by + * + * rowval[nfields].value - rowval[0].value - 4 * nfields. + */ + rowval[nfields].value = conn->inBuffer + conn->inCursor; + rowval[nfields].len = NULL_LEN; + + /* Success! Pass the completed row values to rowProcessor */ + if (!result->rowProcessor(result, result->rowProcessorParam, rowval)) + goto rowProcessError; + + /* Free garbage error message. */ + if (result->rowProcessorErrMes) + { + free(result->rowProcessorErrMes); + result->rowProcessorErrMes = NULL; + } 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..27ef007 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -149,6 +149,16 @@ typedef struct pgNotify struct pgNotify *next; /* list link */ } PGnotify; +/* PGrowValue represents a value of one tuple field in string form, + used by RowProcessor. NULL is represented as len < 0. Otherwise + value points to a string without null termination of the length of + len. */ +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 +426,32 @@ 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 1 for success and must return 0 for + * failure and may set error message by PQsetRowProcessorErrMes. 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 (*RowProcessor)(PGresult *res, void *param, + PGrowValue *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 passed to + * 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 +490,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 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..06d8b26 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -209,6 +209,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 +403,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 +447,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 +518,7 @@ 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 int pqAddRow(PGresult *res, void *param, PGrowValue *columns); extern void pqSaveMessageField(PGresult *res, char code, const char *value); extern void pqSaveParameterStatus(PGconn *conn, const char *name, @@ -560,6 +571,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);