libpq-single-row-0802.patch
text/x-patch
Filename: libpq-single-row-0802.patch
Type: text/x-patch
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: unified
| File | + | − |
|---|---|---|
| contrib/dblink/dblink.c | 0 | 0 |
| src/interfaces/libpq/exports.txt | 0 | 0 |
| src/interfaces/libpq/fe-connect.c | 0 | 0 |
| src/interfaces/libpq/fe-exec.c | 0 | 0 |
| src/interfaces/libpq/fe-lobj.c | 0 | 0 |
| src/interfaces/libpq/fe-protocol2.c | 0 | 0 |
| src/interfaces/libpq/fe-protocol3.c | 0 | 0 |
| src/interfaces/libpq/libpq-fe.h | 0 | 0 |
| src/interfaces/libpq/libpq-int.h | 0 | 0 |
diff --git a/contrib/dblink/dblink.c b/contrib/dblink/dblink.c
index 1e62d8091a9d2bdf60af6745d5a01ee14ee5cf5a..7cf86176c2385b9e4ee37c72d7e3c662ea079f7a 100644
*** a/contrib/dblink/dblink.c
--- b/contrib/dblink/dblink.c
*************** typedef struct storeInfo
*** 70,75 ****
--- 70,78 ----
AttInMetadata *attinmeta;
MemoryContext tmpcontext;
char **cstrs;
+ /* temp storage for results to avoid leaks on exception */
+ PGresult *last_res;
+ PGresult *cur_res;
} storeInfo;
/*
*************** static void materializeQueryResult(Funct
*** 83,90 ****
const char *conname,
const char *sql,
bool fail);
! static int storeHandler(PGresult *res, const PGdataValue *columns,
! const char **errmsgp, void *param);
static remoteConn *getConnectionByName(const char *name);
static HTAB *createConnHash(void);
static void createNewConnection(const char *name, remoteConn *rconn);
--- 86,93 ----
const char *conname,
const char *sql,
bool fail);
! static PGresult *storeQueryResult(storeInfo *sinfo, PGconn *conn, const char *sql);
! static void storeRow(storeInfo *sinfo, PGresult *res, bool first);
static remoteConn *getConnectionByName(const char *name);
static HTAB *createConnHash(void);
static void createNewConnection(const char *name, remoteConn *rconn);
*************** materializeResult(FunctionCallInfo fcinf
*** 927,934 ****
/*
* Execute the given SQL command and store its results into a tuplestore
* to be returned as the result of the current function.
* This is equivalent to PQexec followed by materializeResult, but we make
! * use of libpq's "row processor" API to reduce per-row overhead.
*/
static void
materializeQueryResult(FunctionCallInfo fcinfo,
--- 930,939 ----
/*
* Execute the given SQL command and store its results into a tuplestore
* to be returned as the result of the current function.
+ *
* This is equivalent to PQexec followed by materializeResult, but we make
! * use of libpq's single-row mode to avoid accumulating the whole result
! * inside libpq before it gets transferred to the tuplestore.
*/
static void
materializeQueryResult(FunctionCallInfo fcinfo,
*************** materializeQueryResult(FunctionCallInfo
*** 944,962 ****
/* prepTuplestoreResult must have been called previously */
Assert(rsinfo->returnMode == SFRM_Materialize);
PG_TRY();
{
! /* initialize storeInfo to empty */
! memset(&sinfo, 0, sizeof(sinfo));
! sinfo.fcinfo = fcinfo;
!
! /* We'll collect tuples using storeHandler */
! PQsetRowProcessor(conn, storeHandler, &sinfo);
!
! res = PQexec(conn, sql);
!
! /* We don't keep the custom row processor installed permanently */
! PQsetRowProcessor(conn, NULL, NULL);
if (!res ||
(PQresultStatus(res) != PGRES_COMMAND_OK &&
--- 949,962 ----
/* prepTuplestoreResult must have been called previously */
Assert(rsinfo->returnMode == SFRM_Materialize);
+ /* initialize storeInfo to empty */
+ memset(&sinfo, 0, sizeof(sinfo));
+ sinfo.fcinfo = fcinfo;
+
PG_TRY();
{
! /* execute query, collecting any tuples into the tuplestore */
! res = storeQueryResult(&sinfo, conn, sql);
if (!res ||
(PQresultStatus(res) != PGRES_COMMAND_OK &&
*************** materializeQueryResult(FunctionCallInfo
*** 975,982 ****
else if (PQresultStatus(res) == PGRES_COMMAND_OK)
{
/*
! * storeHandler didn't get called, so we need to convert the
! * command status string to a tuple manually
*/
TupleDesc tupdesc;
AttInMetadata *attinmeta;
--- 975,982 ----
else if (PQresultStatus(res) == PGRES_COMMAND_OK)
{
/*
! * storeRow didn't get called, so we need to convert the command
! * status string to a tuple manually
*/
TupleDesc tupdesc;
AttInMetadata *attinmeta;
*************** materializeQueryResult(FunctionCallInfo
*** 1008,1032 ****
tuplestore_puttuple(tupstore, tuple);
PQclear(res);
}
else
{
Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
! /* storeHandler should have created a tuplestore */
Assert(rsinfo->setResult != NULL);
PQclear(res);
}
}
PG_CATCH();
{
- /* be sure to unset the custom row processor */
- PQsetRowProcessor(conn, NULL, NULL);
/* be sure to release any libpq result we collected */
! if (res)
! PQclear(res);
/* and clear out any pending data in libpq */
! while ((res = PQskipResult(conn)) != NULL)
PQclear(res);
PG_RE_THROW();
}
--- 1008,1037 ----
tuplestore_puttuple(tupstore, tuple);
PQclear(res);
+ res = NULL;
}
else
{
Assert(PQresultStatus(res) == PGRES_TUPLES_OK);
! /* storeRow should have created a tuplestore */
Assert(rsinfo->setResult != NULL);
PQclear(res);
+ res = NULL;
}
+ PQclear(sinfo.last_res);
+ sinfo.last_res = NULL;
+ PQclear(sinfo.cur_res);
+ sinfo.cur_res = NULL;
}
PG_CATCH();
{
/* be sure to release any libpq result we collected */
! PQclear(res);
! PQclear(sinfo.last_res);
! PQclear(sinfo.cur_res);
/* and clear out any pending data in libpq */
! while ((res = PQgetResult(conn)) != NULL)
PQclear(res);
PG_RE_THROW();
}
*************** materializeQueryResult(FunctionCallInfo
*** 1034,1056 ****
}
/*
! * Custom row processor for materializeQueryResult.
! * Prototype of this function must match PQrowProcessor.
*/
! static int
! storeHandler(PGresult *res, const PGdataValue *columns,
! const char **errmsgp, void *param)
{
- storeInfo *sinfo = (storeInfo *) param;
int nfields = PQnfields(res);
- char **cstrs = sinfo->cstrs;
HeapTuple tuple;
- char *pbuf;
- int pbuflen;
int i;
MemoryContext oldcontext;
! if (columns == NULL)
{
/* Prepare for new result set */
ReturnSetInfo *rsinfo = (ReturnSetInfo *) sinfo->fcinfo->resultinfo;
--- 1039,1108 ----
}
/*
! * Execute query, and send any result rows to sinfo->tuplestore.
*/
! static PGresult *
! storeQueryResult(storeInfo *sinfo, PGconn *conn, const char *sql)
! {
! bool first = true;
! PGresult *res;
!
! if (!PQsendQuery(conn, sql))
! return PQgetResult(conn);
!
! if (!PQsetSingleRowMode(conn)) /* shouldn't fail */
! elog(ERROR, "dblink: failed to set single-row mode");
!
! for (;;)
! {
! CHECK_FOR_INTERRUPTS();
!
! sinfo->cur_res = PQgetResult(conn);
! if (!sinfo->cur_res)
! break;
!
! if (PQresultStatus(sinfo->cur_res) == PGRES_SINGLE_TUPLE)
! {
! /* got one row from possibly-bigger resultset */
! storeRow(sinfo, sinfo->cur_res, first);
!
! PQclear(sinfo->cur_res);
! sinfo->cur_res = NULL;
! first = false;
! }
! else
! {
! /* if empty resultset, fill tuplestore header */
! if (first && PQresultStatus(sinfo->cur_res) == PGRES_TUPLES_OK)
! storeRow(sinfo, sinfo->cur_res, first);
!
! /* store completed result at last_res */
! PQclear(sinfo->last_res);
! sinfo->last_res = sinfo->cur_res;
! sinfo->cur_res = NULL;
! first = true;
! }
! }
!
! /* return last_res */
! res = sinfo->last_res;
! sinfo->last_res = NULL;
! return res;
! }
!
! /*
! * Send single row to sinfo->tuplestore.
! * If "first" is true, create the tuplestore using PGresult's metadata.
! */
! static void
! storeRow(storeInfo *sinfo, PGresult *res, bool first)
{
int nfields = PQnfields(res);
HeapTuple tuple;
int i;
MemoryContext oldcontext;
! if (first)
{
/* Prepare for new result set */
ReturnSetInfo *rsinfo = (ReturnSetInfo *) sinfo->fcinfo->resultinfo;
*************** storeHandler(PGresult *res, const PGdata
*** 1098,1110 ****
sinfo->attinmeta = TupleDescGetAttInMetadata(tupdesc);
/* Create a new, empty tuplestore */
! oldcontext = MemoryContextSwitchTo(
! rsinfo->econtext->ecxt_per_query_memory);
sinfo->tuplestore = tuplestore_begin_heap(true, false, work_mem);
rsinfo->setResult = sinfo->tuplestore;
rsinfo->setDesc = tupdesc;
MemoryContextSwitchTo(oldcontext);
/*
* Set up sufficiently-wide string pointers array; this won't change
* in size so it's easy to preallocate.
--- 1150,1165 ----
sinfo->attinmeta = TupleDescGetAttInMetadata(tupdesc);
/* Create a new, empty tuplestore */
! oldcontext = MemoryContextSwitchTo(rsinfo->econtext->ecxt_per_query_memory);
sinfo->tuplestore = tuplestore_begin_heap(true, false, work_mem);
rsinfo->setResult = sinfo->tuplestore;
rsinfo->setDesc = tupdesc;
MemoryContextSwitchTo(oldcontext);
+ /* Done if empty resultset */
+ if (PQntuples(res) == 0)
+ return;
+
/*
* Set up sufficiently-wide string pointers array; this won't change
* in size so it's easy to preallocate.
*************** storeHandler(PGresult *res, const PGdata
*** 1121,1131 ****
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
-
- return 1;
}
! CHECK_FOR_INTERRUPTS();
/*
* Do the following work in a temp context that we reset after each tuple.
--- 1176,1185 ----
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
}
! /* Should have a single-row result if we get here */
! Assert(PQntuples(res) == 1);
/*
* Do the following work in a temp context that we reset after each tuple.
*************** storeHandler(PGresult *res, const PGdata
*** 1135,1180 ****
oldcontext = MemoryContextSwitchTo(sinfo->tmpcontext);
/*
! * The strings passed to us are not null-terminated, but the datatype
! * input functions we're about to call require null termination. Copy the
! * strings and add null termination. As a micro-optimization, allocate
! * all the strings with one palloc.
*/
- pbuflen = nfields; /* count the null terminators themselves */
- for (i = 0; i < nfields; i++)
- {
- int len = columns[i].len;
-
- if (len > 0)
- pbuflen += len;
- }
- pbuf = (char *) palloc(pbuflen);
-
for (i = 0; i < nfields; i++)
{
! int len = columns[i].len;
!
! if (len < 0)
! cstrs[i] = NULL;
else
! {
! cstrs[i] = pbuf;
! memcpy(pbuf, columns[i].value, len);
! pbuf += len;
! *pbuf++ = '\0';
! }
}
/* Convert row to a tuple, and add it to the tuplestore */
! tuple = BuildTupleFromCStrings(sinfo->attinmeta, cstrs);
tuplestore_puttuple(sinfo->tuplestore, tuple);
/* Clean up */
MemoryContextSwitchTo(oldcontext);
MemoryContextReset(sinfo->tmpcontext);
-
- return 1;
}
/*
--- 1189,1212 ----
oldcontext = MemoryContextSwitchTo(sinfo->tmpcontext);
/*
! * Fill cstrs with null-terminated strings of column values.
*/
for (i = 0; i < nfields; i++)
{
! if (PQgetisnull(res, 0, i))
! sinfo->cstrs[i] = NULL;
else
! sinfo->cstrs[i] = PQgetvalue(res, 0, i);
}
/* Convert row to a tuple, and add it to the tuplestore */
! tuple = BuildTupleFromCStrings(sinfo->attinmeta, sinfo->cstrs);
tuplestore_puttuple(sinfo->tuplestore, tuple);
/* Clean up */
MemoryContextSwitchTo(oldcontext);
MemoryContextReset(sinfo->tmpcontext);
}
/*
diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt
index 1251455f1f6d92c74e206b5a9b8dcdeb36ac9b98..9d95e262be3fbf26731c0926013db56dbc4e00ab 100644
*** a/src/interfaces/libpq/exports.txt
--- b/src/interfaces/libpq/exports.txt
*************** PQconnectStartParams 157
*** 160,165 ****
PQping 158
PQpingParams 159
PQlibVersion 160
! PQsetRowProcessor 161
! PQgetRowProcessor 162
! PQskipResult 163
--- 160,163 ----
PQping 158
PQpingParams 159
PQlibVersion 160
! PQsetSingleRowMode 161
diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c
index a32258a8cbab5e655484d0471c031864fa5084de..adaab7aaade60a980b7eaf5b6e9734444c72b930 100644
*** a/src/interfaces/libpq/fe-connect.c
--- b/src/interfaces/libpq/fe-connect.c
*************** makeEmptyPGconn(void)
*** 2709,2716 ****
/* Zero all pointers and booleans */
MemSet(conn, 0, sizeof(PGconn));
! /* install default row processor and notice hooks */
! PQsetRowProcessor(conn, NULL, NULL);
conn->noticeHooks.noticeRec = defaultNoticeReceiver;
conn->noticeHooks.noticeProc = defaultNoticeProcessor;
--- 2709,2715 ----
/* Zero all pointers and booleans */
MemSet(conn, 0, sizeof(PGconn));
! /* install default notice hooks */
conn->noticeHooks.noticeRec = defaultNoticeReceiver;
conn->noticeHooks.noticeProc = defaultNoticeProcessor;
*************** conninfo_uri_parse_options(PQconninfoOpt
*** 4658,4664 ****
if (p == host)
{
printfPQExpBuffer(errorMessage,
! libpq_gettext("IPv6 host address may not be empty in URI: \"%s\"\n"),
uri);
goto cleanup;
}
--- 4657,4663 ----
if (p == host)
{
printfPQExpBuffer(errorMessage,
! libpq_gettext("IPv6 host address may not be empty in URI: \"%s\"\n"),
uri);
goto cleanup;
}
*************** conninfo_uri_parse_params(char *params,
*** 4878,4884 ****
printfPQExpBuffer(errorMessage,
libpq_gettext(
! "invalid URI query parameter: \"%s\"\n"),
keyword);
return false;
}
--- 4877,4883 ----
printfPQExpBuffer(errorMessage,
libpq_gettext(
! "invalid URI query parameter: \"%s\"\n"),
keyword);
return false;
}
*************** conninfo_uri_decode(const char *str, PQE
*** 4943,4949 ****
if (!(get_hexdigit(*q++, &hi) && get_hexdigit(*q++, &lo)))
{
printfPQExpBuffer(errorMessage,
! libpq_gettext("invalid percent-encoded token: \"%s\"\n"),
str);
free(buf);
return NULL;
--- 4942,4948 ----
if (!(get_hexdigit(*q++, &hi) && get_hexdigit(*q++, &lo)))
{
printfPQExpBuffer(errorMessage,
! libpq_gettext("invalid percent-encoded token: \"%s\"\n"),
str);
free(buf);
return NULL;
*************** static void
*** 5594,5601 ****
dot_pg_pass_warning(PGconn *conn)
{
/* If it was 'invalid authorization', add .pgpass mention */
- if (conn->dot_pgpass_used && conn->password_needed && conn->result &&
/* only works with >= 9.0 servers */
strcmp(PQresultErrorField(conn->result, PG_DIAG_SQLSTATE),
ERRCODE_INVALID_PASSWORD) == 0)
{
--- 5593,5600 ----
dot_pg_pass_warning(PGconn *conn)
{
/* If it was 'invalid authorization', add .pgpass mention */
/* only works with >= 9.0 servers */
+ if (conn->dot_pgpass_used && conn->password_needed && conn->result &&
strcmp(PQresultErrorField(conn->result, PG_DIAG_SQLSTATE),
ERRCODE_INVALID_PASSWORD) == 0)
{
diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c
index badc0b32a8e8f3e33e42729a8f8899475a0de31b..53516db723492f11fc9f4bdc6d4d351693c4c22f 100644
*** a/src/interfaces/libpq/fe-exec.c
--- b/src/interfaces/libpq/fe-exec.c
*************** char *const pgresStatus[] = {
*** 38,44 ****
"PGRES_BAD_RESPONSE",
"PGRES_NONFATAL_ERROR",
"PGRES_FATAL_ERROR",
! "PGRES_COPY_BOTH"
};
/*
--- 38,45 ----
"PGRES_BAD_RESPONSE",
"PGRES_NONFATAL_ERROR",
"PGRES_FATAL_ERROR",
! "PGRES_COPY_BOTH",
! "PGRES_SINGLE_TUPLE"
};
/*
*************** static bool static_std_strings = false;
*** 51,58 ****
static PGEvent *dupEvents(PGEvent *events, int count);
static bool pqAddTuple(PGresult *res, PGresAttValue *tup);
- static int pqStdRowProcessor(PGresult *res, const PGdataValue *columns,
- const char **errmsgp, void *param);
static bool PQsendQueryStart(PGconn *conn);
static int PQsendQueryGuts(PGconn *conn,
const char *command,
--- 52,57 ----
*************** static int PQsendQueryGuts(PGconn *conn,
*** 64,71 ****
const int *paramFormats,
int resultFormat);
static void parseInput(PGconn *conn);
- static int dummyRowProcessor(PGresult *res, const PGdataValue *columns,
- const char **errmsgp, void *param);
static bool PQexecStart(PGconn *conn);
static PGresult *PQexecFinish(PGconn *conn);
static int PQsendDescribe(PGconn *conn, char desc_type,
--- 63,68 ----
*************** PQmakeEmptyPGresult(PGconn *conn, ExecSt
*** 181,186 ****
--- 178,184 ----
case PGRES_COPY_OUT:
case PGRES_COPY_IN:
case PGRES_COPY_BOTH:
+ case PGRES_SINGLE_TUPLE:
/* non-error cases */
break;
default:
*************** PQclear(PGresult *res)
*** 698,703 ****
--- 696,703 ----
/*
* Handy subroutine to deallocate any partially constructed async result.
+ *
+ * Any "next" result gets cleared too.
*/
void
pqClearAsyncResult(PGconn *conn)
*************** pqClearAsyncResult(PGconn *conn)
*** 705,710 ****
--- 705,713 ----
if (conn->result)
PQclear(conn->result);
conn->result = NULL;
+ if (conn->next_result)
+ PQclear(conn->next_result);
+ conn->next_result = NULL;
}
/*
*************** pqPrepareAsyncResult(PGconn *conn)
*** 758,764 ****
* conn->errorMessage.
*/
res = conn->result;
- conn->result = NULL; /* handing over ownership to caller */
if (!res)
res = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR);
else
--- 761,766 ----
*************** pqPrepareAsyncResult(PGconn *conn)
*** 771,776 ****
--- 773,788 ----
appendPQExpBufferStr(&conn->errorMessage,
PQresultErrorMessage(res));
}
+
+ /*
+ * Replace conn->result with next_result, if any. In the normal case
+ * there isn't a next result and we're just dropping ownership of the
+ * current result. In single-row mode this restores the situation to what
+ * it was before we created the current single-row result.
+ */
+ conn->result = conn->next_result;
+ conn->next_result = NULL;
+
return res;
}
*************** pqSaveParameterStatus(PGconn *conn, cons
*** 981,1065 ****
/*
! * PQsetRowProcessor
! * Set function that copies row data out from the network buffer,
! * along with a passthrough parameter for it.
! */
! 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 = pqStdRowProcessor;
! conn->rowProcessorParam = conn;
! }
! }
!
! /*
! * PQgetRowProcessor
! * Get current row processor of PGconn.
! * If param is not NULL, also store the passthrough parameter at *param.
! */
! PQrowProcessor
! PQgetRowProcessor(const PGconn *conn, void **param)
! {
! if (!conn)
! {
! if (param)
! *param = NULL;
! return NULL;
! }
!
! if (param)
! *param = conn->rowProcessorParam;
! return conn->rowProcessor;
! }
!
! /*
! * pqStdRowProcessor
! * Add the received row to the PGresult structure
! * Returns 1 if OK, -1 if error occurred.
*
! * Note: "param" should point to the PGconn, but we don't actually need that
! * as of the current coding.
*/
! static int
! pqStdRowProcessor(PGresult *res, const PGdataValue *columns,
! const char **errmsgp, void *param)
{
int nfields = res->numAttributes;
PGresAttValue *tup;
int i;
! if (columns == NULL)
{
! /* New result set ... we have nothing to do in this function. */
! return 1;
}
/*
* Basically we just allocate space in the PGresult for each field and
* copy the data over.
*
! * Note: on malloc failure, we return -1 leaving *errmsgp still NULL,
! * which caller will take to mean "out of memory". This is preferable to
! * trying to set up such a message here, because evidently there's not
! * enough memory for gettext() to do anything.
*/
tup = (PGresAttValue *)
pqResultAlloc(res, nfields * sizeof(PGresAttValue), TRUE);
if (tup == NULL)
! return -1;
for (i = 0; i < nfields; i++)
{
--- 993,1047 ----
/*
! * pqRowProcessor
! * Add the received row to the current async result (conn->result).
! * Returns 1 if OK, 0 if error occurred.
*
! * On error, *errmsgp can be set to an error string to be returned.
! * If it is left NULL, the error is presumed to be "out of memory".
! *
! * In single-row mode, we create a new result holding just the current row,
! * stashing the previous result in conn->next_result so that it becomes
! * active again after pqPrepareAsyncResult(). This allows the result metadata
! * (column descriptions) to be carried forward to each result row.
*/
! int
! pqRowProcessor(PGconn *conn, const char **errmsgp)
{
+ PGresult *res = conn->result;
int nfields = res->numAttributes;
+ const PGdataValue *columns = conn->rowBuf;
PGresAttValue *tup;
int i;
! /*
! * In single-row mode, make a new PGresult that will hold just this one
! * row; the original conn->result is left unchanged so that it can be used
! * again as the template for future rows.
! */
! if (conn->singleRowMode)
{
! /* Copy everything that should be in the result at this point */
! res = PQcopyResult(res,
! PG_COPYRES_ATTRS | PG_COPYRES_EVENTS |
! PG_COPYRES_NOTICEHOOKS);
! if (!res)
! return 0;
}
/*
* Basically we just allocate space in the PGresult for each field and
* copy the data over.
*
! * Note: on malloc failure, we return 0 leaving *errmsgp still NULL, which
! * caller will take to mean "out of memory". This is preferable to trying
! * to set up such a message here, because evidently there's not enough
! * memory for gettext() to do anything.
*/
tup = (PGresAttValue *)
pqResultAlloc(res, nfields * sizeof(PGresAttValue), TRUE);
if (tup == NULL)
! goto fail;
for (i = 0; i < nfields; i++)
{
*************** pqStdRowProcessor(PGresult *res, const P
*** 1078,1084 ****
val = (char *) pqResultAlloc(res, clen + 1, isbinary);
if (val == NULL)
! return -1;
/* copy and zero-terminate the data (even if it's binary) */
memcpy(val, columns[i].value, clen);
--- 1060,1066 ----
val = (char *) pqResultAlloc(res, clen + 1, isbinary);
if (val == NULL)
! goto fail;
/* copy and zero-terminate the data (even if it's binary) */
memcpy(val, columns[i].value, clen);
*************** pqStdRowProcessor(PGresult *res, const P
*** 1091,1100 ****
/* And add the tuple to the PGresult's tuple array */
if (!pqAddTuple(res, tup))
! return -1;
- /* Success */
return 1;
}
--- 1073,1102 ----
/* And add the tuple to the PGresult's tuple array */
if (!pqAddTuple(res, tup))
! goto fail;
!
! /*
! * Success. In single-row mode, make the result available to the client
! * immediately.
! */
! if (conn->singleRowMode)
! {
! /* Change result status to special single-row value */
! res->resultStatus = PGRES_SINGLE_TUPLE;
! /* Stash old result for re-use later */
! conn->next_result = conn->result;
! conn->result = res;
! /* And mark the result ready to return */
! conn->asyncStatus = PGASYNC_READY;
! }
return 1;
+
+ fail:
+ /* release locally allocated PGresult, if we made one */
+ if (res != conn->result)
+ PQclear(res);
+ return 0;
}
*************** PQsendQueryStart(PGconn *conn)
*** 1343,1348 ****
--- 1345,1354 ----
/* initialize async result-accumulation state */
conn->result = NULL;
+ conn->next_result = NULL;
+
+ /* reset single-row processing mode */
+ conn->singleRowMode = false;
/* ready to send command message */
return true;
*************** pqHandleSendFailure(PGconn *conn)
*** 1548,1553 ****
--- 1554,1584 ----
}
/*
+ * Select row-by-row processing mode
+ */
+ int
+ PQsetSingleRowMode(PGconn *conn)
+ {
+ /*
+ * Only allow setting the flag when we have launched a query and not yet
+ * received any results.
+ */
+ if (!conn)
+ return 0;
+ if (conn->asyncStatus != PGASYNC_BUSY)
+ return 0;
+ if (conn->queryclass != PGQUERY_SIMPLE &&
+ conn->queryclass != PGQUERY_EXTENDED)
+ return 0;
+ if (conn->result)
+ return 0;
+
+ /* OK, set flag */
+ conn->singleRowMode = true;
+ return 1;
+ }
+
+ /*
* Consume any available input from the backend
* 0 return: some kind of trouble
* 1 return: no problem
*************** PQconsumeInput(PGconn *conn)
*** 1587,1595 ****
* parseInput: if appropriate, parse input data from backend
* until input is exhausted or a stopping state is reached.
* Note that this function will NOT attempt to read more data from the backend.
- *
- * Note: callers of parseInput must be prepared for a longjmp exit when we are
- * in PGASYNC_BUSY state, since an external row processor might do that.
*/
static void
parseInput(PGconn *conn)
--- 1618,1623 ----
*************** PQgetResult(PGconn *conn)
*** 1737,1785 ****
return res;
}
- /*
- * PQskipResult
- * Get the next PGresult produced by a query, but discard any data rows.
- *
- * This is mainly useful for cleaning up after a longjmp out of a row
- * processor, when resuming processing of the current query result isn't
- * wanted. Note that this is of little value in an async-style application,
- * since any preceding calls to PQisBusy would have already called the regular
- * row processor.
- */
- PGresult *
- PQskipResult(PGconn *conn)
- {
- PGresult *res;
- PQrowProcessor savedRowProcessor;
-
- if (!conn)
- return NULL;
-
- /* temporarily install dummy row processor */
- savedRowProcessor = conn->rowProcessor;
- conn->rowProcessor = dummyRowProcessor;
- /* no need to save/change rowProcessorParam */
-
- /* fetch the next result */
- res = PQgetResult(conn);
-
- /* restore previous row processor */
- conn->rowProcessor = savedRowProcessor;
-
- return res;
- }
-
- /*
- * Do-nothing row processor for PQskipResult
- */
- static int
- dummyRowProcessor(PGresult *res, const PGdataValue *columns,
- const char **errmsgp, void *param)
- {
- return 1;
- }
-
/*
* PQexec
--- 1765,1770 ----
*************** PQexecStart(PGconn *conn)
*** 1886,1892 ****
* Silently discard any prior query result that application didn't eat.
* This is probably poor design, but it's here for backward compatibility.
*/
! while ((result = PQskipResult(conn)) != NULL)
{
ExecStatusType resultStatus = result->resultStatus;
--- 1871,1877 ----
* Silently discard any prior query result that application didn't eat.
* This is probably poor design, but it's here for backward compatibility.
*/
! while ((result = PQgetResult(conn)) != NULL)
{
ExecStatusType resultStatus = result->resultStatus;
diff --git a/src/interfaces/libpq/fe-lobj.c b/src/interfaces/libpq/fe-lobj.c
index 13fd98c2f913d3818758e75bac96822306981b52..f3a6d0341c13ce644a7f1b2f15919385b85d6a6e 100644
*** a/src/interfaces/libpq/fe-lobj.c
--- b/src/interfaces/libpq/fe-lobj.c
*************** lo_initialize(PGconn *conn)
*** 682,689 ****
int n;
const char *query;
const char *fname;
- PQrowProcessor savedRowProcessor;
- void *savedRowProcessorParam;
Oid foid;
if (!conn)
--- 682,687 ----
*************** lo_initialize(PGconn *conn)
*** 732,747 ****
"or proname = 'loread' "
"or proname = 'lowrite'";
- /* Ensure the standard row processor is used to collect the result */
- savedRowProcessor = conn->rowProcessor;
- savedRowProcessorParam = conn->rowProcessorParam;
- PQsetRowProcessor(conn, NULL, NULL);
-
res = PQexec(conn, query);
-
- conn->rowProcessor = savedRowProcessor;
- conn->rowProcessorParam = savedRowProcessorParam;
-
if (res == NULL)
{
free(lobjfuncs);
--- 730,736 ----
diff --git a/src/interfaces/libpq/fe-protocol2.c b/src/interfaces/libpq/fe-protocol2.c
index 8dbd6b6982395850fc167eb20cb8d5590d33122f..1ba5885cd3b419a97f4fc5ea897eac209f364897 100644
*** a/src/interfaces/libpq/fe-protocol2.c
--- b/src/interfaces/libpq/fe-protocol2.c
*************** static int getNotify(PGconn *conn);
*** 49,67 ****
PostgresPollingStatusType
pqSetenvPoll(PGconn *conn)
{
- PostgresPollingStatusType result;
PGresult *res;
- PQrowProcessor savedRowProcessor;
- void *savedRowProcessorParam;
if (conn == NULL || conn->status == CONNECTION_BAD)
return PGRES_POLLING_FAILED;
- /* Ensure the standard row processor is used to collect any results */
- savedRowProcessor = conn->rowProcessor;
- savedRowProcessorParam = conn->rowProcessorParam;
- PQsetRowProcessor(conn, NULL, NULL);
-
/* Check whether there are any data for us */
switch (conn->setenv_state)
{
--- 49,59 ----
*************** pqSetenvPoll(PGconn *conn)
*** 77,86 ****
if (n < 0)
goto error_return;
if (n == 0)
! {
! result = PGRES_POLLING_READING;
! goto normal_return;
! }
break;
}
--- 69,75 ----
if (n < 0)
goto error_return;
if (n == 0)
! return PGRES_POLLING_READING;
break;
}
*************** pqSetenvPoll(PGconn *conn)
*** 94,101 ****
/* Should we raise an error if called when not active? */
case SETENV_STATE_IDLE:
! result = PGRES_POLLING_OK;
! goto normal_return;
default:
printfPQExpBuffer(&conn->errorMessage,
--- 83,89 ----
/* Should we raise an error if called when not active? */
case SETENV_STATE_IDLE:
! return PGRES_POLLING_OK;
default:
printfPQExpBuffer(&conn->errorMessage,
*************** pqSetenvPoll(PGconn *conn)
*** 192,201 ****
case SETENV_STATE_CLIENT_ENCODING_WAIT:
{
if (PQisBusy(conn))
! {
! result = PGRES_POLLING_READING;
! goto normal_return;
! }
res = PQgetResult(conn);
--- 180,186 ----
case SETENV_STATE_CLIENT_ENCODING_WAIT:
{
if (PQisBusy(conn))
! return PGRES_POLLING_READING;
res = PQgetResult(conn);
*************** pqSetenvPoll(PGconn *conn)
*** 220,229 ****
case SETENV_STATE_OPTION_WAIT:
{
if (PQisBusy(conn))
! {
! result = PGRES_POLLING_READING;
! goto normal_return;
! }
res = PQgetResult(conn);
--- 205,211 ----
case SETENV_STATE_OPTION_WAIT:
{
if (PQisBusy(conn))
! return PGRES_POLLING_READING;
res = PQgetResult(conn);
*************** pqSetenvPoll(PGconn *conn)
*** 262,278 ****
goto error_return;
conn->setenv_state = SETENV_STATE_QUERY1_WAIT;
! result = PGRES_POLLING_READING;
! goto normal_return;
}
case SETENV_STATE_QUERY1_WAIT:
{
if (PQisBusy(conn))
! {
! result = PGRES_POLLING_READING;
! goto normal_return;
! }
res = PQgetResult(conn);
--- 244,256 ----
goto error_return;
conn->setenv_state = SETENV_STATE_QUERY1_WAIT;
! return PGRES_POLLING_READING;
}
case SETENV_STATE_QUERY1_WAIT:
{
if (PQisBusy(conn))
! return PGRES_POLLING_READING;
res = PQgetResult(conn);
*************** pqSetenvPoll(PGconn *conn)
*** 349,365 ****
goto error_return;
conn->setenv_state = SETENV_STATE_QUERY2_WAIT;
! result = PGRES_POLLING_READING;
! goto normal_return;
}
case SETENV_STATE_QUERY2_WAIT:
{
if (PQisBusy(conn))
! {
! result = PGRES_POLLING_READING;
! goto normal_return;
! }
res = PQgetResult(conn);
--- 327,339 ----
goto error_return;
conn->setenv_state = SETENV_STATE_QUERY2_WAIT;
! return PGRES_POLLING_READING;
}
case SETENV_STATE_QUERY2_WAIT:
{
if (PQisBusy(conn))
! return PGRES_POLLING_READING;
res = PQgetResult(conn);
*************** pqSetenvPoll(PGconn *conn)
*** 406,413 ****
{
/* Query finished, so we're done */
conn->setenv_state = SETENV_STATE_IDLE;
! result = PGRES_POLLING_OK;
! goto normal_return;
}
break;
}
--- 380,386 ----
{
/* Query finished, so we're done */
conn->setenv_state = SETENV_STATE_IDLE;
! return PGRES_POLLING_OK;
}
break;
}
*************** pqSetenvPoll(PGconn *conn)
*** 425,436 ****
error_return:
conn->setenv_state = SETENV_STATE_IDLE;
! result = PGRES_POLLING_FAILED;
!
! normal_return:
! conn->rowProcessor = savedRowProcessor;
! conn->rowProcessorParam = savedRowProcessorParam;
! return result;
}
--- 398,404 ----
error_return:
conn->setenv_state = SETENV_STATE_IDLE;
! return PGRES_POLLING_FAILED;
}
*************** normal_return:
*** 438,446 ****
* parseInput: if appropriate, parse input data from backend
* until input is exhausted or a stopping state is reached.
* Note that this function will NOT attempt to read more data from the backend.
- *
- * Note: callers of parseInput must be prepared for a longjmp exit when we are
- * in PGASYNC_BUSY state, since an external row processor might do that.
*/
void
pqParseInput2(PGconn *conn)
--- 406,411 ----
*************** getRowDescriptions(PGconn *conn)
*** 746,776 ****
/* Success! */
conn->result = result;
! /*
! * Advance inStart to show that the "T" message has been processed. We
! * must do this before calling the row processor, in case it longjmps.
! */
conn->inStart = conn->inCursor;
! /* Give the row processor a chance to initialize for new result set */
! errmsg = NULL;
! switch ((*conn->rowProcessor) (result, NULL, &errmsg,
! conn->rowProcessorParam))
! {
! case 1:
! /* everything is good */
! return 0;
!
! case -1:
! /* error, report the errmsg below */
! break;
! default:
! /* unrecognized return code */
! errmsg = libpq_gettext("unrecognized return value from row processor");
! break;
! }
! goto set_error_result;
advance_and_error:
--- 711,726 ----
/* Success! */
conn->result = result;
! /* Advance inStart to show that the "T" message has been processed. */
conn->inStart = conn->inCursor;
! /*
! * We could perform additional setup for the new result set here, but for
! * now there's nothing else to do.
! */
! /* And we're done. */
! return 0;
advance_and_error:
*************** advance_and_error:
*** 781,788 ****
*/
conn->inStart = conn->inEnd;
- set_error_result:
-
/*
* Replace partially constructed result with an error result. First
* discard the old result to try to win back some memory.
--- 731,736 ----
*************** set_error_result:
*** 790,796 ****
pqClearAsyncResult(conn);
/*
! * If row processor didn't provide an error message, assume "out of
* memory" was meant. The advantage of having this special case is that
* freeing the old result first greatly improves the odds that gettext()
* will succeed in providing a translation.
--- 738,744 ----
pqClearAsyncResult(conn);
/*
! * If preceding code didn't provide an error message, assume "out of
* memory" was meant. The advantage of having this special case is that
* freeing the old result first greatly improves the odds that gettext()
* will succeed in providing a translation.
*************** getAnotherTuple(PGconn *conn, bool binar
*** 937,967 ****
free(bitmap);
bitmap = NULL;
! /*
! * Advance inStart to show that the "D" message has been processed. We
! * must do this before calling the row processor, in case it longjmps.
! */
conn->inStart = conn->inCursor;
! /* Pass the completed row values to rowProcessor */
errmsg = NULL;
! switch ((*conn->rowProcessor) (result, rowbuf, &errmsg,
! conn->rowProcessorParam))
! {
! case 1:
! /* everything is good */
! return 0;
!
! case -1:
! /* error, report the errmsg below */
! break;
! default:
! /* unrecognized return code */
! errmsg = libpq_gettext("unrecognized return value from row processor");
! break;
! }
! goto set_error_result;
advance_and_error:
--- 885,899 ----
free(bitmap);
bitmap = NULL;
! /* Advance inStart to show that the "D" message has been processed. */
conn->inStart = conn->inCursor;
! /* Process the collected row */
errmsg = NULL;
! if (pqRowProcessor(conn, &errmsg))
! return 0; /* normal, successful exit */
! goto set_error_result; /* pqRowProcessor failed, report it */
advance_and_error:
*************** set_error_result:
*** 981,987 ****
pqClearAsyncResult(conn);
/*
! * If row processor didn't provide an error message, assume "out of
* memory" was meant. The advantage of having this special case is that
* freeing the old result first greatly improves the odds that gettext()
* will succeed in providing a translation.
--- 913,919 ----
pqClearAsyncResult(conn);
/*
! * If preceding code didn't provide an error message, assume "out of
* memory" was meant. The advantage of having this special case is that
* freeing the old result first greatly improves the odds that gettext()
* will succeed in providing a translation.
diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c
index 173af2e0a79ef3b443de515cda851e277deaed2d..d289f82285fea00d5de20542e43ea103493f9e58 100644
*** a/src/interfaces/libpq/fe-protocol3.c
--- b/src/interfaces/libpq/fe-protocol3.c
*************** static int build_startup_packet(const PG
*** 61,69 ****
* parseInput: if appropriate, parse input data from backend
* until input is exhausted or a stopping state is reached.
* Note that this function will NOT attempt to read more data from the backend.
- *
- * Note: callers of parseInput must be prepared for a longjmp exit when we are
- * in PGASYNC_BUSY state, since an external row processor might do that.
*/
void
pqParseInput3(PGconn *conn)
--- 61,66 ----
*************** handleSyncLoss(PGconn *conn, char id, in
*** 446,455 ****
* Returns: 0 if processed message successfully, EOF to suspend parsing
* (the latter case is not actually used currently).
* In either case, conn->inStart has been advanced past the message.
- *
- * Note: the row processor could also choose to longjmp out of libpq,
- * in which case the library's state must allow for resumption at the
- * next message.
*/
static int
getRowDescriptions(PGconn *conn, int msgLength)
--- 443,448 ----
*************** getRowDescriptions(PGconn *conn, int msg
*** 564,573 ****
/* Success! */
conn->result = result;
! /*
! * Advance inStart to show that the "T" message has been processed. We
! * must do this before calling the row processor, in case it longjmps.
! */
conn->inStart = conn->inCursor;
/*
--- 557,563 ----
/* Success! */
conn->result = result;
! /* Advance inStart to show that the "T" message has been processed. */
conn->inStart = conn->inCursor;
/*
*************** getRowDescriptions(PGconn *conn, int msg
*** 580,604 ****
return 0;
}
! /* Give the row processor a chance to initialize for new result set */
! errmsg = NULL;
! switch ((*conn->rowProcessor) (result, NULL, &errmsg,
! conn->rowProcessorParam))
! {
! case 1:
! /* everything is good */
! return 0;
!
! case -1:
! /* error, report the errmsg below */
! break;
! default:
! /* unrecognized return code */
! errmsg = libpq_gettext("unrecognized return value from row processor");
! break;
! }
! goto set_error_result;
advance_and_error:
/* Discard unsaved result, if any */
--- 570,582 ----
return 0;
}
! /*
! * We could perform additional setup for the new result set here, but for
! * now there's nothing else to do.
! */
! /* And we're done. */
! return 0;
advance_and_error:
/* Discard unsaved result, if any */
*************** advance_and_error:
*** 608,615 ****
/* Discard the failed message by pretending we read it */
conn->inStart += 5 + msgLength;
- set_error_result:
-
/*
* Replace partially constructed result with an error result. First
* discard the old result to try to win back some memory.
--- 586,591 ----
*************** set_error_result:
*** 617,624 ****
pqClearAsyncResult(conn);
/*
! * If row processor didn't provide an error message, assume "out of
! * memory" was meant.
*/
if (!errmsg)
errmsg = libpq_gettext("out of memory for query result");
--- 593,602 ----
pqClearAsyncResult(conn);
/*
! * If preceding code didn't provide an error message, assume "out of
! * memory" was meant. The advantage of having this special case is that
! * freeing the old result first greatly improves the odds that gettext()
! * will succeed in providing a translation.
*/
if (!errmsg)
errmsg = libpq_gettext("out of memory for query result");
*************** failure:
*** 695,704 ****
* Returns: 0 if processed message successfully, EOF to suspend parsing
* (the latter case is not actually used currently).
* In either case, conn->inStart has been advanced past the message.
- *
- * Note: the row processor could also choose to longjmp out of libpq,
- * in which case the library's state must allow for resumption at the
- * next message.
*/
static int
getAnotherTuple(PGconn *conn, int msgLength)
--- 673,678 ----
*************** getAnotherTuple(PGconn *conn, int msgLen
*** 778,808 ****
goto advance_and_error;
}
! /*
! * Advance inStart to show that the "D" message has been processed. We
! * must do this before calling the row processor, in case it longjmps.
! */
conn->inStart = conn->inCursor;
! /* Pass the completed row values to rowProcessor */
errmsg = NULL;
! switch ((*conn->rowProcessor) (result, rowbuf, &errmsg,
! conn->rowProcessorParam))
! {
! case 1:
! /* everything is good */
! return 0;
!
! case -1:
! /* error, report the errmsg below */
! break;
! default:
! /* unrecognized return code */
! errmsg = libpq_gettext("unrecognized return value from row processor");
! break;
! }
! goto set_error_result;
advance_and_error:
/* Discard the failed message by pretending we read it */
--- 752,766 ----
goto advance_and_error;
}
! /* Advance inStart to show that the "D" message has been processed. */
conn->inStart = conn->inCursor;
! /* Process the collected row */
errmsg = NULL;
! if (pqRowProcessor(conn, &errmsg))
! return 0; /* normal, successful exit */
! goto set_error_result; /* pqRowProcessor failed, report it */
advance_and_error:
/* Discard the failed message by pretending we read it */
*************** set_error_result:
*** 817,823 ****
pqClearAsyncResult(conn);
/*
! * If row processor didn't provide an error message, assume "out of
* memory" was meant. The advantage of having this special case is that
* freeing the old result first greatly improves the odds that gettext()
* will succeed in providing a translation.
--- 775,781 ----
pqClearAsyncResult(conn);
/*
! * If preceding code didn't provide an error message, assume "out of
* memory" was meant. The advantage of having this special case is that
* freeing the old result first greatly improves the odds that gettext()
* will succeed in providing a translation.
diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h
index 67db6119bbaa35ae31fb58bb902a455b093ea23f..9d05dd20605a84ab4c4ec9d61ef8697fb2f3b77e 100644
*** a/src/interfaces/libpq/libpq-fe.h
--- b/src/interfaces/libpq/libpq-fe.h
*************** typedef enum
*** 90,96 ****
* backend */
PGRES_NONFATAL_ERROR, /* notice or warning message */
PGRES_FATAL_ERROR, /* query failed */
! PGRES_COPY_BOTH /* Copy In/Out data transfer in progress */
} ExecStatusType;
typedef enum
--- 90,97 ----
* backend */
PGRES_NONFATAL_ERROR, /* notice or warning message */
PGRES_FATAL_ERROR, /* query failed */
! PGRES_COPY_BOTH, /* Copy In/Out data transfer in progress */
! PGRES_SINGLE_TUPLE /* single tuple from larger resultset */
} ExecStatusType;
typedef enum
*************** typedef struct pg_conn PGconn;
*** 129,145 ****
*/
typedef struct pg_result PGresult;
- /* PGdataValue represents a data field value being passed to a row processor.
- * It could be either text or binary data; text data is not zero-terminated.
- * A SQL NULL is represented by len < 0; then value is still valid but there
- * are no data bytes there.
- */
- typedef struct pgDataValue
- {
- int len; /* data length in bytes, or <0 if NULL */
- const char *value; /* data value, without zero-termination */
- } PGdataValue;
-
/* PGcancel encapsulates the information needed to cancel a running
* query on an existing connection.
* The contents of this struct are not supposed to be known to applications.
--- 130,135 ----
*************** typedef struct pgNotify
*** 161,170 ****
struct pgNotify *next; /* list link */
} PGnotify;
- /* Function type for row-processor callback */
- typedef int (*PQrowProcessor) (PGresult *res, const PGdataValue *columns,
- const char **errmsgp, void *param);
-
/* Function types for notice-handling callbacks */
typedef void (*PQnoticeReceiver) (void *arg, const PGresult *res);
typedef void (*PQnoticeProcessor) (void *arg, const char *message);
--- 151,156 ----
*************** extern int PQsendQueryPrepared(PGconn *c
*** 403,419 ****
const int *paramLengths,
const int *paramFormats,
int resultFormat);
extern PGresult *PQgetResult(PGconn *conn);
- extern PGresult *PQskipResult(PGconn *conn);
/* Routines for managing an asynchronous query */
extern int PQisBusy(PGconn *conn);
extern int PQconsumeInput(PGconn *conn);
- /* Override default per-row processing */
- extern void PQsetRowProcessor(PGconn *conn, PQrowProcessor func, void *param);
- extern PQrowProcessor PQgetRowProcessor(const PGconn *conn, void **param);
-
/* LISTEN/NOTIFY support */
extern PGnotify *PQnotifies(PGconn *conn);
--- 389,401 ----
const int *paramLengths,
const int *paramFormats,
int resultFormat);
+ extern int PQsetSingleRowMode(PGconn *conn);
extern PGresult *PQgetResult(PGconn *conn);
/* Routines for managing an asynchronous query */
extern int PQisBusy(PGconn *conn);
extern int PQconsumeInput(PGconn *conn);
/* LISTEN/NOTIFY support */
extern PGnotify *PQnotifies(PGconn *conn);
diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h
index 4bc89269fababe5e4d8ecbf6e80ca1a8625d4bd5..2bac59c3d879ecabce42ceab3b5133df03a0886a 100644
*** a/src/interfaces/libpq/libpq-int.h
--- b/src/interfaces/libpq/libpq-int.h
*************** typedef struct pgLobjfuncs
*** 277,282 ****
--- 277,293 ----
Oid fn_lo_write; /* OID of backend function LOwrite */
} PGlobjfuncs;
+ /* PGdataValue represents a data field value being passed to a row processor.
+ * It could be either text or binary data; text data is not zero-terminated.
+ * A SQL NULL is represented by len < 0; then value is still valid but there
+ * are no data bytes there.
+ */
+ typedef struct pgDataValue
+ {
+ int len; /* data length in bytes, or <0 if NULL */
+ const char *value; /* data value, without zero-termination */
+ } PGdataValue;
+
/*
* PGconn stores all the state data associated with a single connection
* to a backend.
*************** struct pg_conn
*** 324,333 ****
/* Optional file to write trace info to */
FILE *Pfdebug;
- /* Callback procedure for per-row processing */
- PQrowProcessor rowProcessor; /* function pointer */
- void *rowProcessorParam; /* passthrough argument */
-
/* Callback procedures for notice message processing */
PGNoticeHooks noticeHooks;
--- 335,340 ----
*************** struct pg_conn
*** 346,351 ****
--- 353,359 ----
bool options_valid; /* true if OK to attempt connection */
bool nonblocking; /* whether this connection is using nonblock
* sending semantics */
+ bool singleRowMode; /* return current query result row-by-row? */
char copy_is_binary; /* 1 = copy binary, 0 = copy text */
int copy_already_done; /* # bytes already returned in COPY
* OUT */
*************** struct pg_conn
*** 406,411 ****
--- 414,420 ----
/* Status for asynchronous result construction */
PGresult *result; /* result being constructed */
+ PGresult *next_result; /* next result (used in single-row mode) */
/* Assorted state for SSL, GSS, etc */
*************** extern void pqSaveMessageField(PGresult
*** 517,522 ****
--- 526,532 ----
const char *value);
extern void pqSaveParameterStatus(PGconn *conn, const char *name,
const char *value);
+ extern int pqRowProcessor(PGconn *conn, const char **errmsgp);
extern void pqHandleSendFailure(PGconn *conn);
/* === in fe-protocol2.c === */