diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index 3d8aaf0..07e8900 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -4115,6 +4115,112 @@ int PQflush(PGconn *conn); read-ready and then read the response as described above. + + Above-mentioned functions always wait until full resultset has arrived + before makeing row data available as PGresult. Sometimes it's + more useful to process rows as soon as the arrive from network. + For that, following functions can be used: + + + + PQgetRow + + PQgetRow + + + + + + Waits for the next row from a prior + PQsendQuery, + PQsendQueryParams, + PQsendQueryPrepared call, and returns it. + A null pointer is returned when no more rows are available or + some error happened. + +PGresult *PQgetRow(PGconn *conn); + + + + + If this function returns non-NULL result, it is a + PGresult that contains exactly 1 row. + It needs to be freed later with PQclear. + + + On synchronous connection, the function will wait for more + data from network until all resultset is done. So it returns + NULL only if resultset has completely received or some error + happened. In both cases, call PQgetResult + next to get final status. + + + + On asynchronous connection the function does not read more data + from network. So after NULL call PQisBusy + to see whether final PGresult is avilable + or more data needs to be read from network via + PQconsumeInput. Do not call + PQisBusy before PQgetRow + has returned NULL, as PQisBusy will parse + any available rows and add them to main PGresult + that will be returned later by PQgetResult. + + + + + + + + PQrecvRow + + PQrecvRow + + + + + + Get row data without constructing PGresult for it. This is the + underlying function for PQgetRow. + +int PQrecvRow(PGconn *conn, PGresult **hdr_p, PGrowValue **row_p); + + + + + It returns row data as pointers to network buffer. + All structures are owned by libpq's + PGconn and must not be freed or stored + by user. Instead row data should be copied to user structures, before + any libpq result-processing function + is called. + + + It returns 1 when row data is available. + Argument hdr_p will contain pointer + to empty PGresult that describes + row contents. Actual data is in row_p. + For the description of structure PGrowValue + see . + + It returns 0 when no more rows are avalable. On synchronous + connection, it means resultset is fully arrived. Call + PQgetResult to get final status. + On asynchronous connection it can also mean more data + needs to be read from network. Call PQisBusy + to see whether PQgetResult + or PQconsumeInput needs to be called next. + + + it returns -1 if some network error occured. + Use connection status functions described in + to check connection state. + + + + + + diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt index a6418ec..0433e4a 100644 --- a/src/interfaces/libpq/exports.txt +++ b/src/interfaces/libpq/exports.txt @@ -163,3 +163,5 @@ PQlibVersion 160 PQsetRowProcessor 161 PQgetRowProcessor 162 PQskipResult 163 +PQrecvRow 164 +PQgetRow 165 diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 52beb07..ffcfd46 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -1969,6 +1969,133 @@ PQskipResult(PGconn *conn, int skipAll) return ret; } +/* temp buffer to pass pointers */ +struct RecvRowBuf +{ + PGresult *temp_hdr; + PGrowValue *temp_row; +}; + +/* set pointers, do early exit from PQisBusy() */ +static int +recv_row_proc(PGresult *hdr, PGrowValue *row, void *arg) +{ + struct RecvRowBuf *buf = arg; + buf->temp_hdr = hdr; + buf->temp_row = row; + return 0; +} + +/* + * PQrecvRow + * + * Wait and return next row in resultset. + * + * Returns: + * 1 - got row data, the pointers are owned by PGconn + * 0 - no rows available, either resultset complete + * or more data needed (async-only) + * -1 - some problem, check connection error + */ +int +PQrecvRow(PGconn *conn, PGresult **hdr_p, PGrowValue **row_p) +{ + struct RecvRowBuf buf; + int rc; + int ret = -1; + PQrowProcessor oldproc; + void *oldarg; + + *hdr_p = NULL; + *row_p = NULL; + + /* the query may be still pending, send it */ + while (1) + { + rc = PQflush(conn); + if (rc < 0) + return -1; + if (rc == 0) + break; + if (pqWait(FALSE, TRUE, conn)) + return -1; + } + + /* replace existing row processor */ + oldproc = PQgetRowProcessor(conn, &oldarg); + PQsetRowProcessor(conn, recv_row_proc, &buf); + + /* read data */ + while (1) + { + buf.temp_hdr = NULL; + buf.temp_row = NULL; + + /* done with resultset? */ + if (!PQisBusy(conn)) + break; + + /* new row available? */ + if (buf.temp_row) + { + *hdr_p = buf.temp_hdr; + *row_p = buf.temp_row; + ret = 1; + goto done; + } + + /* + * More data needed + */ + + if (pqIsnonblocking(conn)) + /* let user worry about new data */ + break; + if (pqWait(TRUE, FALSE, conn)) + goto done; + if (!PQconsumeInput(conn)) + goto done; + } + /* no more rows available */ + ret = 0; +done: + /* restore old row processor */ + PQsetRowProcessor(conn, oldproc, oldarg); + return ret; +} + +/* + * PQgetRow + * Returns next available row for resultset. NULL means + * no row available, either resultset is done + * or more data needed (only if async connection). + */ +PGresult * +PQgetRow(PGconn *conn) +{ + PGresult *hdr, *res; + PGrowValue *row; + + /* check if row is available */ + if (PQrecvRow(conn, &hdr, &row) != 1) + return NULL; + + /* Now make PGresult out of it */ + res = PQcopyResult(hdr, PG_COPYRES_ATTRS); + if (!res) + { + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("out of memory\n")); + pqSaveErrorResult(conn); + return NULL; + } + + /* add the row, pqAddRow sets error itself */ + if (pqAddRow(res, row, NULL)) + return res; + PQclear(res); + return NULL; +} /* * PQdescribePrepared diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h index e1d3339..50872a5 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -400,6 +400,10 @@ extern int PQsendQueryPrepared(PGconn *conn, int resultFormat); extern PGresult *PQgetResult(PGconn *conn); +/* fetch single row from resultset */ +extern PGresult *PQgetRow(PGconn *conn); +extern int PQrecvRow(PGconn *conn, PGresult **hdr_p, PGrowValue **row_p); + /* Routines for managing an asynchronous query */ extern int PQisBusy(PGconn *conn); extern int PQconsumeInput(PGconn *conn);