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..53da93c 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, void *param, PGrowValue *columns); /* ---------------- @@ -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, 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,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,48 @@ PQexecFinish(PGconn *conn) return lastResult; } + +/* + * Do-nothing row processor for PQskipResult + */ +static int +dummyRowProcessor(PGresult *res, void *param, PGrowValue *columns) +{ + 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; + 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 (PQgetResult(conn)); + else if (PQgetResult(conn)) + 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..ae4d7b0 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, 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: -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..0260ba6 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, 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; -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..ffe88e2 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, 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); +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, 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);