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 === */