>From a04a2f8ff32cf3095f7769eedde11ca946f024e5 Mon Sep 17 00:00:00 2001 From: Kyotaro Horiguchi Date: Fri, 28 Nov 2014 10:52:41 +0900 Subject: [PATCH] Asynchronous execution of postgres_fdw v2 This is the modified version of Asynchronous execution of postgres_fdw. - Refactored to make the code simpler. - Added comment about logic outline and struct members. --- contrib/postgres_fdw/connection.c | 84 ++++++------ contrib/postgres_fdw/postgres_fdw.c | 255 +++++++++++++++++++++++++++--------- contrib/postgres_fdw/postgres_fdw.h | 84 +++++++++++- 3 files changed, 318 insertions(+), 105 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 4e02cb2..574b08e 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -44,7 +44,7 @@ typedef struct ConnCacheKey typedef struct ConnCacheEntry { ConnCacheKey key; /* hash key (must be first) */ - PGconn *conn; /* connection to foreign server, or NULL */ + PgFdwConn *conn; /* connection to foreign server, or NULL */ int xact_depth; /* 0 = no xact open, 1 = main xact open, 2 = * one level of subxact open, etc */ bool have_prep_stmt; /* have we prepared any stmts in this xact? */ @@ -64,10 +64,10 @@ static unsigned int prep_stmt_number = 0; static bool xact_got_connection = false; /* prototypes of private functions */ -static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user); +static PgFdwConn *connect_pg_server(ForeignServer *server, UserMapping *user); static void check_conn_params(const char **keywords, const char **values); -static void configure_remote_session(PGconn *conn); -static void do_sql_command(PGconn *conn, const char *sql); +static void configure_remote_session(PgFdwConn *conn); +static void do_sql_command(PgFdwConn *conn, const char *sql); static void begin_remote_xact(ConnCacheEntry *entry); static void pgfdw_xact_callback(XactEvent event, void *arg); static void pgfdw_subxact_callback(SubXactEvent event, @@ -93,7 +93,7 @@ static void pgfdw_subxact_callback(SubXactEvent event, * be useful and not mere pedantry. We could not flush any active connections * mid-transaction anyway. */ -PGconn * +PgFdwConn * GetConnection(ForeignServer *server, UserMapping *user, bool will_prep_stmt) { @@ -161,7 +161,7 @@ GetConnection(ForeignServer *server, UserMapping *user, entry->have_error = false; entry->conn = connect_pg_server(server, user); elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\"", - entry->conn, server->servername); + PFC_PGCONN(entry->conn), server->servername); } /* @@ -169,6 +169,13 @@ GetConnection(ForeignServer *server, UserMapping *user, */ begin_remote_xact(entry); + /* + * Finish async query immediately if another foreign scan node sharing + * this connection comes. + */ + if (++entry->conn->nscans > 1 && PFC_IS_ASYNC_RUNNING(entry->conn)) + fetch_more_data(entry->conn->async_scan); + /* Remember if caller will prepare statements */ entry->have_prep_stmt |= will_prep_stmt; @@ -178,10 +185,10 @@ GetConnection(ForeignServer *server, UserMapping *user, /* * Connect to remote server using specified server and user mapping properties. */ -static PGconn * +static PgFdwConn * connect_pg_server(ForeignServer *server, UserMapping *user) { - PGconn *volatile conn = NULL; + PgFdwConn *volatile conn = NULL; /* * Use PG_TRY block to ensure closing connection on error. @@ -223,14 +230,14 @@ connect_pg_server(ForeignServer *server, UserMapping *user) /* verify connection parameters and make connection */ check_conn_params(keywords, values); - conn = PQconnectdbParams(keywords, values, false); - if (!conn || PQstatus(conn) != CONNECTION_OK) + conn = PFCconnectdbParams(keywords, values, false); + if (!conn || PFCstatus(conn) != CONNECTION_OK) { char *connmessage; int msglen; /* libpq typically appends a newline, strip that */ - connmessage = pstrdup(PQerrorMessage(conn)); + connmessage = pstrdup(PFCerrorMessage(conn)); msglen = strlen(connmessage); if (msglen > 0 && connmessage[msglen - 1] == '\n') connmessage[msglen - 1] = '\0'; @@ -246,7 +253,7 @@ connect_pg_server(ForeignServer *server, UserMapping *user) * otherwise, he's piggybacking on the postgres server's user * identity. See also dblink_security_check() in contrib/dblink. */ - if (!superuser() && !PQconnectionUsedPassword(conn)) + if (!superuser() && !PFCconnectionUsedPassword(conn)) ereport(ERROR, (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED), errmsg("password is required"), @@ -263,7 +270,7 @@ connect_pg_server(ForeignServer *server, UserMapping *user) { /* Release PGconn data structure if we managed to create one */ if (conn) - PQfinish(conn); + PFCfinish(conn); PG_RE_THROW(); } PG_END_TRY(); @@ -312,9 +319,9 @@ check_conn_params(const char **keywords, const char **values) * there are any number of ways to break things. */ static void -configure_remote_session(PGconn *conn) +configure_remote_session(PgFdwConn *conn) { - int remoteversion = PQserverVersion(conn); + int remoteversion = PFCserverVersion(conn); /* Force the search path to contain only pg_catalog (see deparse.c) */ do_sql_command(conn, "SET search_path = pg_catalog"); @@ -348,11 +355,11 @@ configure_remote_session(PGconn *conn) * Convenience subroutine to issue a non-data-returning SQL command to remote */ static void -do_sql_command(PGconn *conn, const char *sql) +do_sql_command(PgFdwConn *conn, const char *sql) { PGresult *res; - res = PQexec(conn, sql); + res = PFCexec(conn, sql); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, conn, true, sql); PQclear(res); @@ -379,7 +386,7 @@ begin_remote_xact(ConnCacheEntry *entry) const char *sql; elog(DEBUG3, "starting remote transaction on connection %p", - entry->conn); + PFC_PGCONN(entry->conn)); if (IsolationIsSerializable()) sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE"; @@ -408,13 +415,11 @@ begin_remote_xact(ConnCacheEntry *entry) * Release connection reference count created by calling GetConnection. */ void -ReleaseConnection(PGconn *conn) +ReleaseConnection(PgFdwConn *conn) { - /* - * Currently, we don't actually track connection references because all - * cleanup is managed on a transaction or subtransaction basis instead. So - * there's nothing to do here. - */ + /* ongoing async query should be canceled if no scans left */ + if (--PFC_NSCANS(conn) == 0) + finish_async_connection(conn); } /* @@ -429,7 +434,7 @@ ReleaseConnection(PGconn *conn) * collisions are highly improbable; just be sure to use %u not %d to print. */ unsigned int -GetCursorNumber(PGconn *conn) +GetCursorNumber(PgFdwConn *conn) { return ++cursor_number; } @@ -443,7 +448,7 @@ GetCursorNumber(PGconn *conn) * increasing the risk of prepared-statement name collisions by resetting. */ unsigned int -GetPrepStmtNumber(PGconn *conn) +GetPrepStmtNumber(PgFdwConn *conn) { return ++prep_stmt_number; } @@ -462,7 +467,7 @@ GetPrepStmtNumber(PGconn *conn) * marked with have_error = true. */ void -pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, +pgfdw_report_error(int elevel, PGresult *res, PgFdwConn *conn, bool clear, const char *sql) { /* If requested, PGresult must be released before leaving this function. */ @@ -490,7 +495,7 @@ pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, * return NULL, not a PGresult at all. */ if (message_primary == NULL) - message_primary = PQerrorMessage(conn); + message_primary = PFCerrorMessage(conn); ereport(elevel, (errcode(sqlstate), @@ -542,7 +547,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) if (entry->xact_depth > 0) { elog(DEBUG3, "closing remote transaction on connection %p", - entry->conn); + PFC_PGCONN(entry->conn)); switch (event) { @@ -567,7 +572,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) */ if (entry->have_prep_stmt && entry->have_error) { - res = PQexec(entry->conn, "DEALLOCATE ALL"); + res = PFCexec(entry->conn, "DEALLOCATE ALL"); PQclear(res); } entry->have_prep_stmt = false; @@ -597,7 +602,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) /* Assume we might have lost track of prepared statements */ entry->have_error = true; /* If we're aborting, abort all remote transactions too */ - res = PQexec(entry->conn, "ABORT TRANSACTION"); + res = PFCexec(entry->conn, "ABORT TRANSACTION"); /* Note: can't throw ERROR, it would be infinite loop */ if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(WARNING, res, entry->conn, true, @@ -608,7 +613,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) /* As above, make sure to clear any prepared stmts */ if (entry->have_prep_stmt && entry->have_error) { - res = PQexec(entry->conn, "DEALLOCATE ALL"); + res = PFCexec(entry->conn, "DEALLOCATE ALL"); PQclear(res); } entry->have_prep_stmt = false; @@ -620,17 +625,17 @@ pgfdw_xact_callback(XactEvent event, void *arg) /* Reset state to show we're out of a transaction */ entry->xact_depth = 0; + PFC_RESET(entry->conn); /* * If the connection isn't in a good idle state, discard it to * recover. Next GetConnection will open a new connection. */ - if (PQstatus(entry->conn) != CONNECTION_OK || - PQtransactionStatus(entry->conn) != PQTRANS_IDLE) + if (PFCstatus(entry->conn) != CONNECTION_OK || + PFCtransactionStatus(entry->conn) != PQTRANS_IDLE) { - elog(DEBUG3, "discarding connection %p", entry->conn); - PQfinish(entry->conn); - entry->conn = NULL; + elog(DEBUG3, "discarding connection %p", PFC_PGCONN(entry->conn)); + PFCfinish(entry->conn); } } @@ -676,6 +681,9 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, PGresult *res; char sql[100]; + /* Shut down asynchronous scan if running */ + PFC_RESET(entry->conn); + /* * We only care about connections with open remote subtransactions of * the current level. @@ -701,7 +709,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, snprintf(sql, sizeof(sql), "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", curlevel, curlevel); - res = PQexec(entry->conn, sql); + res = PFCexec(entry->conn, sql); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(WARNING, res, entry->conn, true, sql); else diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index d76e739..552b0d4 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -136,7 +136,7 @@ typedef struct PgFdwScanState List *retrieved_attrs; /* list of retrieved attribute numbers */ /* for remote query execution */ - PGconn *conn; /* connection for the scan */ + PgFdwConn *conn; /* connection for the scan */ unsigned int cursor_number; /* quasi-unique ID for my cursor */ bool cursor_exists; /* have we created the cursor? */ int numParams; /* number of parameters passed to query */ @@ -156,6 +156,7 @@ typedef struct PgFdwScanState /* working memory contexts */ MemoryContext batch_cxt; /* context holding current batch of tuples */ MemoryContext temp_cxt; /* context for per-tuple temporary data */ + ExprContext *econtext; /* copy of ps_ExprContext of ForeignScanState */ } PgFdwScanState; /* @@ -167,7 +168,7 @@ typedef struct PgFdwModifyState AttInMetadata *attinmeta; /* attribute datatype conversion metadata */ /* for remote query execution */ - PGconn *conn; /* connection for the scan */ + PgFdwConn *conn; /* connection for the scan */ char *p_name; /* name of prepared statement, if created */ /* extracted fdw_private data */ @@ -298,7 +299,7 @@ static void estimate_path_cost_size(PlannerInfo *root, double *p_rows, int *p_width, Cost *p_startup_cost, Cost *p_total_cost); static void get_remote_estimate(const char *sql, - PGconn *conn, + PgFdwConn *conn, double *rows, int *width, Cost *startup_cost, @@ -306,9 +307,8 @@ static void get_remote_estimate(const char *sql, static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel, EquivalenceClass *ec, EquivalenceMember *em, void *arg); -static void create_cursor(ForeignScanState *node); -static void fetch_more_data(ForeignScanState *node); -static void close_cursor(PGconn *conn, unsigned int cursor_number); +static void create_cursor(PgFdwScanState *node); +static void close_cursor(PgFdwConn *conn, unsigned int cursor_number); static void prepare_foreign_modify(PgFdwModifyState *fmstate); static const char **convert_prep_stmt_params(PgFdwModifyState *fmstate, ItemPointer tupleid, @@ -329,6 +329,18 @@ static HeapTuple make_tuple_from_result_row(PGresult *res, MemoryContext temp_context); static void conversion_error_callback(void *arg); +/* wrapper functions for libpq functions */ +PgFdwConn * +PFCconnectdbParams(const char *const * keywords, + const char *const * values, int expand_dbname) +{ + PgFdwConn *ret = PFC_ALLOCATE(); + + PFC_INIT(ret); + PFC_PGCONN(ret) = PQconnectdbParams(keywords, values, expand_dbname); + + return ret; +} /* * Foreign-data wrapper handler function: return a struct with pointers @@ -982,6 +994,15 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) fsstate->param_values = (const char **) palloc0(numParams * sizeof(char *)); else fsstate->param_values = NULL; + + fsstate->econtext = node->ss.ps.ps_ExprContext; + + /* + * Start scanning asynchronously if it is the first scan on this + * connection. + */ + if (PFC_NSCANS(fsstate->conn) == 1) + create_cursor(fsstate); } /* @@ -1000,7 +1021,7 @@ postgresIterateForeignScan(ForeignScanState *node) * cursor on the remote side. */ if (!fsstate->cursor_exists) - create_cursor(node); + create_cursor(fsstate); /* * Get some more tuples, if we've run out. @@ -1009,7 +1030,7 @@ postgresIterateForeignScan(ForeignScanState *node) { /* No point in another fetch if we already detected EOF, though. */ if (!fsstate->eof_reached) - fetch_more_data(node); + fetch_more_data(fsstate); /* If we didn't get any tuples, must be end of data. */ if (fsstate->next_tuple >= fsstate->num_tuples) return ExecClearTuple(slot); @@ -1069,7 +1090,7 @@ postgresReScanForeignScan(ForeignScanState *node) * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQexec(fsstate->conn, sql); + res = PFCexec(fsstate->conn, sql); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, fsstate->conn, true, sql); PQclear(res); @@ -1398,13 +1419,13 @@ postgresExecForeignInsert(EState *estate, * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQexecPrepared(fmstate->conn, - fmstate->p_name, - fmstate->p_nums, - p_values, - NULL, - NULL, - 0); + res = PFCexecPrepared(fmstate->conn, + fmstate->p_name, + fmstate->p_nums, + p_values, + NULL, + NULL, + 0); if (PQresultStatus(res) != (fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK)) pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query); @@ -1468,13 +1489,13 @@ postgresExecForeignUpdate(EState *estate, * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQexecPrepared(fmstate->conn, - fmstate->p_name, - fmstate->p_nums, - p_values, - NULL, - NULL, - 0); + res = PFCexecPrepared(fmstate->conn, + fmstate->p_name, + fmstate->p_nums, + p_values, + NULL, + NULL, + 0); if (PQresultStatus(res) != (fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK)) pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query); @@ -1538,13 +1559,13 @@ postgresExecForeignDelete(EState *estate, * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQexecPrepared(fmstate->conn, - fmstate->p_name, - fmstate->p_nums, - p_values, - NULL, - NULL, - 0); + res = PFCexecPrepared(fmstate->conn, + fmstate->p_name, + fmstate->p_nums, + p_values, + NULL, + NULL, + 0); if (PQresultStatus(res) != (fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK)) pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query); @@ -1594,7 +1615,7 @@ postgresEndForeignModify(EState *estate, * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQexec(fmstate->conn, sql); + res = PFCexec(fmstate->conn, sql); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, fmstate->conn, true, sql); PQclear(res); @@ -1726,7 +1747,7 @@ estimate_path_cost_size(PlannerInfo *root, List *local_join_conds; StringInfoData sql; List *retrieved_attrs; - PGconn *conn; + PgFdwConn *conn; Selectivity local_sel; QualCost local_cost; @@ -1836,7 +1857,7 @@ estimate_path_cost_size(PlannerInfo *root, * The given "sql" must be an EXPLAIN command. */ static void -get_remote_estimate(const char *sql, PGconn *conn, +get_remote_estimate(const char *sql, PgFdwConn *conn, double *rows, int *width, Cost *startup_cost, Cost *total_cost) { @@ -1852,7 +1873,7 @@ get_remote_estimate(const char *sql, PGconn *conn, /* * Execute EXPLAIN remotely. */ - res = PQexec(conn, sql); + res = PFCexec(conn, sql); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(ERROR, res, conn, false, sql); @@ -1917,13 +1938,12 @@ ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel, * Create cursor for node's query with current parameter values. */ static void -create_cursor(ForeignScanState *node) +create_cursor(PgFdwScanState *fsstate) { - PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; - ExprContext *econtext = node->ss.ps.ps_ExprContext; + ExprContext *econtext = fsstate->econtext; int numParams = fsstate->numParams; const char **values = fsstate->param_values; - PGconn *conn = fsstate->conn; + PgFdwConn *conn = fsstate->conn; StringInfoData buf; PGresult *res; @@ -1985,8 +2005,8 @@ create_cursor(ForeignScanState *node) * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQexecParams(conn, buf.data, numParams, NULL, values, - NULL, NULL, 0); + res = PFCexecParams(conn, buf.data, numParams, NULL, values, + NULL, NULL, 0); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, conn, true, fsstate->query); PQclear(res); @@ -2001,15 +2021,18 @@ create_cursor(ForeignScanState *node) /* Clean up */ pfree(buf.data); + + /* Start async scan if this is the first scan */ + if (PFC_NSCANS(conn) == 1) + fetch_more_data(fsstate); } /* * Fetch some more rows from the node's cursor. */ -static void -fetch_more_data(ForeignScanState *node) +void +fetch_more_data(PgFdwScanState *fsstate) { - PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state; PGresult *volatile res = NULL; MemoryContext oldcontext; @@ -2024,7 +2047,7 @@ fetch_more_data(ForeignScanState *node) /* PGresult must be released before leaving this function. */ PG_TRY(); { - PGconn *conn = fsstate->conn; + PgFdwConn *conn = fsstate->conn; char sql[64]; int fetch_size; int numrows; @@ -2036,9 +2059,64 @@ fetch_more_data(ForeignScanState *node) snprintf(sql, sizeof(sql), "FETCH %d FROM c%u", fetch_size, fsstate->cursor_number); - res = PQexec(conn, sql); + switch (conn->async_state) + { + case PGFDW_CONN_IDLE: + Assert(conn->async_scan == NULL); + + /* Do async fetch only when only one scan uses this connection */ + if (conn->nscans == 1) + { + if (!PFCsendQuery(conn, sql)) + pgfdw_report_error(ERROR, res, conn, false, + fsstate->query); + + conn->async_state = PGFDW_CONN_ASYNC_RUNNING; + conn->async_scan = fsstate; + goto end_of_fetch; + } + + /* Do synchronous query execution */ + conn->async_state = PGFDW_CONN_SYNC_RUNNING; + res = PFCexec(conn, sql); + break; + + case PGFDW_CONN_ASYNC_RUNNING: + Assert(conn->async_scan != NULL); + + res = PFCgetResult(conn); + if (PQntuples(res) == fetch_size) + { + /* + * Connection state doesn't go to IDLE even if all data + * has been sent to client for asynchronous query. One + * more PQgetResult() is needed to reset the state to + * IDLE. See PQexecFinish() for details. + */ + if (PFCgetResult(conn) != NULL) + elog(ERROR, "Connection status error."); + } + + if (conn->nscans == 1) + break; + + /* + * If nscans is more then 1, stop invoking command asynchronously + * for multiple scans on this connection. If nscan is zero, async + * command on this connection should be finished immediately. + */ + conn->async_state = PGFDW_CONN_SYNC_RUNNING; + conn->async_scan = NULL; + break; + + default: + elog(ERROR, "unexpected async state : %d", conn->async_state); + break; + + } + /* On error, report the original query, not the FETCH. */ - if (PQresultStatus(res) != PGRES_TUPLES_OK) + if (res && PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(ERROR, res, conn, false, fsstate->query); /* Convert the data into HeapTuples */ @@ -2066,6 +2144,33 @@ fetch_more_data(ForeignScanState *node) PQclear(res); res = NULL; + + switch(conn->async_state) + { + case PGFDW_CONN_ASYNC_RUNNING: + if (!fsstate->eof_reached) + { + /* + * We can immediately request the next bunch of tuples if + * we're on asynchronous connection. + */ + if (!PFCsendQuery(conn, sql)) + pgfdw_report_error(ERROR, res, conn, false, fsstate->query); + break; + } + + /* Fall through */ + case PGFDW_CONN_SYNC_RUNNING: + PFC_SET_IDLE(conn); + break; + + default: + elog(ERROR, "Unexpedted async state: %d", conn->async_state); + break; + } + +end_of_fetch: + ; /* Nothing to do here but needed to make compiler quiet. */ } PG_CATCH(); { @@ -2079,6 +2184,32 @@ fetch_more_data(ForeignScanState *node) } /* + * Force cancelling async command state. + */ +void +finish_async_connection(PgFdwConn *conn) +{ + PgFdwScanState *fsstate = conn->async_scan; + PgFdwConn *async_conn; + + /* Nothing to do if no async connection */ + if (fsstate == NULL) return; + async_conn = fsstate->conn; + Assert(async_conn); + + /* Finish async command if any */ + if (PFC_IS_ASYNC_RUNNING(async_conn)) + fetch_more_data(async_conn->async_scan); + + Assert(async_conn->async_state == PGFDW_CONN_IDLE && + async_conn->async_scan == NULL); + + /* Immediately discard the result */ + fsstate->next_tuple = 0; + fsstate->num_tuples = 0; +} + +/* * Force assorted GUC parameters to settings that ensure that we'll output * data values in a form that is unambiguous to the remote server. * @@ -2132,7 +2263,7 @@ reset_transmission_modes(int nestlevel) * Utility routine to close a cursor. */ static void -close_cursor(PGconn *conn, unsigned int cursor_number) +close_cursor(PgFdwConn *conn, unsigned int cursor_number) { char sql[64]; PGresult *res; @@ -2143,7 +2274,7 @@ close_cursor(PGconn *conn, unsigned int cursor_number) * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQexec(conn, sql); + res = PFCexec(conn, sql); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, conn, true, sql); PQclear(res); @@ -2175,11 +2306,11 @@ prepare_foreign_modify(PgFdwModifyState *fmstate) * We don't use a PG_TRY block here, so be careful not to throw error * without releasing the PGresult. */ - res = PQprepare(fmstate->conn, - p_name, - fmstate->query, - 0, - NULL); + res = PFCprepare(fmstate->conn, + p_name, + fmstate->query, + 0, + NULL); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query); @@ -2297,7 +2428,7 @@ postgresAnalyzeForeignTable(Relation relation, ForeignTable *table; ForeignServer *server; UserMapping *user; - PGconn *conn; + PgFdwConn *conn; StringInfoData sql; PGresult *volatile res = NULL; @@ -2329,7 +2460,7 @@ postgresAnalyzeForeignTable(Relation relation, /* In what follows, do not risk leaking any PGresults. */ PG_TRY(); { - res = PQexec(conn, sql.data); + res = PFCexec(conn, sql.data); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(ERROR, res, conn, false, sql.data); @@ -2379,7 +2510,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, ForeignTable *table; ForeignServer *server; UserMapping *user; - PGconn *conn; + PgFdwConn *conn; unsigned int cursor_number; StringInfoData sql; PGresult *volatile res = NULL; @@ -2423,7 +2554,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, /* In what follows, do not risk leaking any PGresults. */ PG_TRY(); { - res = PQexec(conn, sql.data); + res = PFCexec(conn, sql.data); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, conn, false, sql.data); PQclear(res); @@ -2453,7 +2584,7 @@ postgresAcquireSampleRowsFunc(Relation relation, int elevel, snprintf(fetch_sql, sizeof(fetch_sql), "FETCH %d FROM c%u", fetch_size, cursor_number); - res = PQexec(conn, fetch_sql); + res = PFCexec(conn, fetch_sql); /* On error, report the original query, not the FETCH. */ if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(ERROR, res, conn, false, sql.data); @@ -2582,7 +2713,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) bool import_not_null = true; ForeignServer *server; UserMapping *mapping; - PGconn *conn; + PgFdwConn *conn; StringInfoData buf; PGresult *volatile res = NULL; int numrows, @@ -2615,7 +2746,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) conn = GetConnection(server, mapping, false); /* Don't attempt to import collation if remote server hasn't got it */ - if (PQserverVersion(conn) < 90100) + if (PFCserverVersion(conn) < 90100) import_collate = false; /* Create workspace for strings */ @@ -2628,7 +2759,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) appendStringInfoString(&buf, "SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = "); deparseStringLiteral(&buf, stmt->remote_schema); - res = PQexec(conn, buf.data); + res = PFCexec(conn, buf.data); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(ERROR, res, conn, false, buf.data); @@ -2723,7 +2854,7 @@ postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid) appendStringInfo(&buf, " ORDER BY c.relname, a.attnum"); /* Fetch the data */ - res = PQexec(conn, buf.data); + res = PFCexec(conn, buf.data); if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(ERROR, res, conn, false, buf.data); diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index 950c6f7..2c81189 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -20,17 +20,91 @@ #include "libpq-fe.h" +/* + * PgFdwConnState - states of PgFdwConn + * + * PgFdwConn manages asynchronous query execution status on a PGconn + * connection. Since one PGconn cannot accept multiple asynchronous queries + * at once, the ongoing async query is immediately finished by another claim + * of the PgFdwConn to use. This state transition is represented using the + * enumeration PgFdwConnState and mainly made within fetch_more_data(). + * + * PGFDW_CONN_ASYNC_RUNNING is the state to be entered when calling + * fetch_more_data() on the PgFdwConn in IDLE state and used only one + * scan. Being called on a PgFdwConn of the state, fetch_more_data() sends the + * next FETCH request after getting the result of the previous request. + * + * PGFDW_CONN_SYNC_RUNNING is rather an internal state in + * fetch_more_data(). It indicates that the function shouldn't send the next + * fetch requst after getting the result. + */ +typedef enum PgFdwConnState +{ + PGFDW_CONN_IDLE, /* running no query */ + PGFDW_CONN_ASYNC_RUNNING, /* running a query asynchronously */ + PGFDW_CONN_SYNC_RUNNING /* running a query synchronously */ +} PgFdwConnState; + +typedef struct PgFdwConn +{ + PGconn *pgconn; /* libpq connection for this connection */ + int nscans; /* number of scans using this connection */ + PgFdwConnState async_state;/* query running state */ + struct PgFdwScanState *async_scan; /* the connection currently running + * async query on this connection */ +} PgFdwConn; + +/* Macros to operate PgFdwConn */ +#define PFC_IS_ASYNC_RUNNING(c) ((c)->async_state == PGFDW_CONN_ASYNC_RUNNING) +#define PFC_PGCONN(c) ((c)->pgconn) +#define PFC_NSCANS(c) ((c)->nscans) +#define PFC_SET_IDLE(c) ((c)->async_scan = NULL, \ + (c)->async_state = PGFDW_CONN_IDLE) +#define PFC_RESET(c) \ + ((PFC_IS_ASYNC_RUNNING(c) ? PFCconsumeInput(c):0), \ + PFC_SET_IDLE(c), PFC_NSCANS(c) = 0) +#define PFC_INIT(c) (PFC_NSCANS(c) = 0, PFC_SET_IDLE(c)) + +#define PFC_ALLOCATE() ((PgFdwConn *)malloc(sizeof(PgFdwConn))) +#define PFC_FREE(c) free(c) + +/* libpq wrappers to take PgFdwConn* instead of PGconn* */ +#define PFCsendQuery(c,q) PQsendQuery((c)->pgconn, (q)) +#define PFCexec(c, q) PQexec((c)->pgconn, (q)) +#define PFCexecParams(c, q, n, t, v, l, f, rf) \ + PQexecParams((c)->pgconn,(q),(n),(t),(v),(l),(f),(rf)) +#define PFCprepare(c, sn, q, n, t) PQprepare((c)->pgconn,(sn),(q),(n),(t)) +#define PFCexecPrepared(c, sn, n, v, l, f, rf) \ + PQexecPrepared((c)->pgconn,(sn),(n),(v),(l),(f),(rf)) +#define PFCgetResult(c) PQgetResult((c)->pgconn) +#define PFCconsumeInput(c) PQconsumeInput((c)->pgconn) +#define PFCisBusy(c) PQisBusy((c)->pgconn) +#define PFCstatus(c) PQstatus((c)->pgconn) +#define PFCtransactionStatus(c) PQtransactionStatus((c)->pgconn) +#define PFCserverVersion(c) PQserverVersion((c)->pgconn) +#define PFCerrorMessage(c) PQerrorMessage((c)->pgconn) +#define PFCconnectionUsedPassword(c) PQconnectionUsedPassword((c)->pgconn) + +/* These are not simple wrappers of PQfinish */ +#define PFCfinish(c) (PQfinish((c)->pgconn), PFC_FREE(c)) + +/* libpq wrapper functions */ +extern PgFdwConn *PFCconnectdbParams(const char *const * keywords, + const char *const * values, int expand_dbname); + /* in postgres_fdw.c */ extern int set_transmission_modes(void); extern void reset_transmission_modes(int nestlevel); +extern void fetch_more_data(struct PgFdwScanState *node); +extern void finish_async_connection(PgFdwConn *fsstate); /* in connection.c */ -extern PGconn *GetConnection(ForeignServer *server, UserMapping *user, +extern PgFdwConn *GetConnection(ForeignServer *server, UserMapping *user, bool will_prep_stmt); -extern void ReleaseConnection(PGconn *conn); -extern unsigned int GetCursorNumber(PGconn *conn); -extern unsigned int GetPrepStmtNumber(PGconn *conn); -extern void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, +extern void ReleaseConnection(PgFdwConn *conn); +extern unsigned int GetCursorNumber(PgFdwConn *conn); +extern unsigned int GetPrepStmtNumber(PgFdwConn *conn); +extern void pgfdw_report_error(int elevel, PGresult *res, PgFdwConn *conn, bool clear, const char *sql); /* in option.c */ -- 2.1.0.GIT