diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 17f8c1e8f31..7451c6e6bbe 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -411,7 +411,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) entry->mapping_hashvalue = GetSysCacheHashValue1(USERMAPPINGOID, ObjectIdGetDatum(user->umid)); - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; + entry->state.entry = entry; /* * Determine whether to keep the connection that we're about to make here @@ -1075,6 +1076,20 @@ GetPrepStmtNumber(PGconn *conn) return ++prep_stmt_number; } +/* + * Exported version of begin_remote_xact(). + * + * This can be called for connections on which begin_remote_xact() has started + * a remote transaction. + */ +void +pgfdw_begin_remote_xact(ConnCacheEntry *entry) +{ + Assert(entry); + Assert(entry->xact_depth > 0); + begin_remote_xact(entry); +} + /* * Submit a query and wait for the result. * @@ -1091,6 +1106,13 @@ pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state) if (state && state->pendingAreq) process_pending_request(state->pendingAreq); + /* + * Second, synchronize the local/remote transactions. Note that we need + * to do this because this function can be called from open cursors. + */ + if (state) + pgfdw_begin_remote_xact(state->entry); + if (!PQsendQuery(conn, query)) return NULL; return pgfdw_get_result(conn); @@ -1194,6 +1216,16 @@ pgfdw_xact_callback(XactEvent event, void *arg) List *pending_entries = NIL; List *cancel_requested = NIL; + /* + * If we are called for post-commit cleanup, all we need to do is clean up + * foreign scans. + */ + if (event == XACT_EVENT_PARALLEL_COMMIT || event == XACT_EVENT_COMMIT) + { + AtEOXact_ForeignScan(true); + return; + } + /* Quick exit if no connections were touched in this transaction. */ if (!xact_got_connection) return; @@ -1357,6 +1389,10 @@ pgfdw_xact_callback(XactEvent event, void *arg) /* Likewise for read_only_level */ read_only_level = 0; + + /* If we are called for abort cleanup, clean up foreign scans as well */ + if (event == XACT_EVENT_PARALLEL_ABORT || event == XACT_EVENT_ABORT) + AtEOXact_ForeignScan(false); } /* @@ -1372,8 +1408,9 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, List *pending_entries = NIL; List *cancel_requested = NIL; - /* Nothing to do at subxact start, nor after commit. */ + /* Nothing to do at subxact start. */ if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB || + event == SUBXACT_EVENT_COMMIT_SUB || event == SUBXACT_EVENT_ABORT_SUB)) return; @@ -1381,11 +1418,22 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, if (!xact_got_connection) return; + curlevel = GetCurrentTransactionNestLevel(); + + /* + * If we are called for post-subcommit cleanup, all we need to do is clean + * up foreign scans. + */ + if (event == SUBXACT_EVENT_COMMIT_SUB) + { + AtEOSubXact_ForeignScan(curlevel, true); + return; + } + /* * Scan all connection cache entries to find open remote subtransactions * of the current level, and close them. */ - curlevel = GetCurrentTransactionNestLevel(); hash_seq_init(&scan, ConnectionHash); while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) { @@ -1459,6 +1507,10 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, /* If in read_only_level, reset it */ if (curlevel == read_only_level) read_only_level = 0; + + /* If we are called for subabort cleanup, clean up foreign scans as well */ + if (event == SUBXACT_EVENT_ABORT_SUB) + AtEOSubXact_ForeignScan(curlevel, false); } /* @@ -1966,10 +2018,10 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) * If pendingAreq of the per-connection state is not NULL, it means that * an asynchronous fetch begun by fetch_more_data_begin() was not done * successfully and thus the per-connection state was not reset in - * fetch_more_data(); in that case reset the per-connection state here. + * fetch_more_data(); in that case reset pendingAreq here. */ if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; /* Disarm changing_xact_state if it all worked */ entry->changing_xact_state = false; @@ -2263,9 +2315,9 @@ pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, entry->have_error = false; } - /* Reset the per-connection state if needed */ + /* Reset pendingAreq here if any */ if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; /* We're done with this entry; unset the changing_xact_state flag */ entry->changing_xact_state = false; @@ -2308,9 +2360,9 @@ pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, entry->have_prep_stmt = false; entry->have_error = false; - /* Reset the per-connection state if needed */ + /* Reset pendingAreq here if any */ if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + entry->state.pendingAreq = NULL; /* We're done with this entry; unset the changing_xact_state flag */ entry->changing_xact_state = false; diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 3045774124e..0d5f030dd87 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -5332,6 +5332,9 @@ ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE user_enum; -- =================================================================== -- subtransaction -- + local/remote error doesn't break cursor +-- + we allow cursors opened before a savepoint to be first fetched +-- after the savepoint, but if that subtransaction aborts, do not +-- allow them to be fetched further. -- =================================================================== BEGIN; DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; @@ -5370,6 +5373,61 @@ SELECT * FROM ft1 ORDER BY c1 LIMIT 1; 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo (1 row) +COMMIT; +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +ERROR: cannot scan foreign table "ft1" anymore +ABORT; +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +CLOSE c; +COMMIT; +-- cursor created in a subtransaction is handed to its parent subtransaction +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo +(1 row) + +SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo +(1 row) + +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; + c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 +----+----+-------+------------------------------+--------------------------+----+------------+----- + 3 | 3 | 00003 | Sun Jan 04 00:00:00 1970 PST | Sun Jan 04 00:00:00 1970 | 3 | 3 | foo +(1 row) + +CLOSE c; COMMIT; -- =================================================================== -- test handling of collations diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 2bcff4b26b4..1d798328948 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -17,6 +17,7 @@ #include "access/htup_details.h" #include "access/sysattr.h" #include "access/table.h" +#include "access/xact.h" #include "catalog/pg_opfamily.h" #include "commands/defrem.h" #include "commands/explain_format.h" @@ -185,6 +186,7 @@ typedef struct PgFdwScanState PgFdwConnState *conn_state; /* extra per-connection state */ unsigned int cursor_number; /* quasi-unique ID for my cursor */ bool cursor_exists; /* have we created the cursor? */ + int cursor_level; /* xact depth at which it was created */ int numParams; /* number of parameters passed to query */ FmgrInfo *param_flinfo; /* output conversion functions for them */ List *param_exprs; /* executable expressions for param values */ @@ -207,8 +209,21 @@ typedef struct PgFdwScanState MemoryContext temp_cxt; /* context for per-tuple temporary data */ int fetch_size; /* number of tuples per fetch */ + + bool scan_allowed; /* is scanning the foreign table allowed? */ } PgFdwScanState; +typedef struct ForeignScanEntry +{ + ForeignScanState *fs_node; + int fs_level; +} ForeignScanEntry; + +/* + * Foreign scan hash table (initialized on first use) + */ +static HTAB *ForeignScanHash = NULL; + /* * Execution state of a foreign insert/update/delete operation. */ @@ -539,6 +554,8 @@ static void create_cursor(ForeignScanState *node); static void fetch_more_data(ForeignScanState *node); static void close_cursor(PGconn *conn, unsigned int cursor_number, PgFdwConnState *conn_state); +static void ForeignScanHashInsert(ForeignScanState *node); +static void ForeignScanHashDelete(ForeignScanState *node); static PgFdwModifyState *create_foreign_modify(EState *estate, RangeTblEntry *rte, ResultRelInfo *resultRelInfo, @@ -1762,6 +1779,7 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Assign a unique ID for my cursor */ fsstate->cursor_number = GetCursorNumber(fsstate->conn); fsstate->cursor_exists = false; + fsstate->cursor_level = 0; /* Get private info created by planner functions. */ fsstate->query = strVal(list_nth(fsplan->fdw_private, @@ -1822,6 +1840,10 @@ postgresBeginForeignScan(ForeignScanState *node, int eflags) /* Set the async-capable flag */ fsstate->async_capable = node->ss.ps.async_capable; + + fsstate->scan_allowed = true; + + ForeignScanHashInsert(node); } /* @@ -1904,6 +1926,7 @@ postgresReScanForeignScan(ForeignScanState *node) if (node->ss.ps.chgParam != NULL) { fsstate->cursor_exists = false; + fsstate->cursor_level = 0; snprintf(sql, sizeof(sql), "CLOSE c%u", fsstate->cursor_number); } @@ -1915,6 +1938,7 @@ postgresReScanForeignScan(ForeignScanState *node) else { fsstate->cursor_exists = false; + fsstate->cursor_level = 0; snprintf(sql, sizeof(sql), "CLOSE c%u", fsstate->cursor_number); } @@ -1961,6 +1985,8 @@ postgresEndForeignScan(ForeignScanState *node) ReleaseConnection(fsstate->conn); fsstate->conn = NULL; + ForeignScanHashDelete(node); + /* MemoryContexts will be deleted automatically. */ } @@ -4054,10 +4080,22 @@ create_cursor(ForeignScanState *node) StringInfoData buf; PGresult *res; + if (!fsstate->scan_allowed) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot scan foreign table \"%s\" anymore", + RelationGetRelationName(fsstate->rel)))); + /* First, process a pending asynchronous request, if any. */ if (fsstate->conn_state->pendingAreq) process_pending_request(fsstate->conn_state->pendingAreq); + /* + * Second, synchronize the local/remote transactions. Note that we need + * to do this because this function can be called from open cursors. + */ + pgfdw_begin_remote_xact(fsstate->conn_state->entry); + /* * Construct array of query parameter values in text format. We do the * conversions in the short-lived per-tuple context, so as not to cause a @@ -4103,6 +4141,7 @@ create_cursor(ForeignScanState *node) /* Mark the cursor as created, and show no tuples have been retrieved */ fsstate->cursor_exists = true; + fsstate->cursor_level = GetCurrentTransactionNestLevel(); fsstate->tuples = NULL; fsstate->num_tuples = 0; fsstate->next_tuple = 0; @@ -4147,7 +4186,7 @@ fetch_more_data(ForeignScanState *node) if (PQresultStatus(res) != PGRES_TUPLES_OK) pgfdw_report_error(res, conn, fsstate->query); - /* Reset per-connection state */ + /* Reset the pending asynchronous request */ fsstate->conn_state->pendingAreq = NULL; } else @@ -4270,6 +4309,131 @@ close_cursor(PGconn *conn, unsigned int cursor_number, PQclear(res); } +/* + * Add a ForeignScan node to the ForeignScanHash table + */ +static void +ForeignScanHashInsert(ForeignScanState *node) +{ + ForeignScanEntry *entry; + bool found; + + /* First time through, initialize foreign scan hashtable */ + if (ForeignScanHash == NULL) + { + HASHCTL ctl; + + ctl.keysize = sizeof(ForeignScanState *); + ctl.entrysize = sizeof(ForeignScanEntry); + ForeignScanHash = hash_create("postgres_fdw ForeignScan nodes", + 256L, + &ctl, + HASH_ELEM | HASH_BLOBS); + } + + entry = (ForeignScanEntry *) hash_search(ForeignScanHash, + &node, + HASH_ENTER, + &found); + if (found) + elog(ERROR, "duplicate ForeignScan node"); + entry->fs_node = node; + entry->fs_level = GetCurrentTransactionNestLevel(); +} + +/* + * Remove a ForeignScan node from the ForeignScanHash table + */ +static void +ForeignScanHashDelete(ForeignScanState *node) +{ + ForeignScanEntry *entry; + + Assert(ForeignScanHash); + + entry = (ForeignScanEntry *) hash_search(ForeignScanHash, + &node, + HASH_REMOVE, + NULL); + if (entry == NULL) + elog(WARNING, "trying to delete ForeignScan node that does not exist"); +} + +/* + * Clean up foreign scans after a subtransaction commit/abort + */ +void +AtEOXact_ForeignScan(bool is_commit) +{ + HASH_SEQ_STATUS scan; + ForeignScanEntry *entry; + + if (ForeignScanHash == NULL) + return; + + hash_seq_init(&scan, ForeignScanHash); + while ((entry = hash_seq_search(&scan)) != NULL) + { + ForeignScanState *node = entry->fs_node; + + /* On commit, complain about leftover ForeignScan nodes */ + if (is_commit) + elog(WARNING, "ForeignScan node %p still existing", node); + ForeignScanHashDelete(node); + } +} + +/* + * Clean up foreign scans after a subtransaction subcommit/subabort + */ +void +AtEOSubXact_ForeignScan(int level, bool is_subcommit) +{ + HASH_SEQ_STATUS scan; + ForeignScanEntry *entry; + + if (ForeignScanHash == NULL) + return; + + hash_seq_init(&scan, ForeignScanHash); + while ((entry = hash_seq_search(&scan)) != NULL) + { + ForeignScanState *node = entry->fs_node; + PgFdwScanState *fsstate; + + if (is_subcommit) + { + /* + * Relabel foreign scans created in this subxact as though they + * are owned by the parent subxact + */ + if (entry->fs_level >= level) + entry->fs_level = level - 1; + /* Likewise for remote cursors declared in this subxact */ + fsstate = (PgFdwScanState *) node->fdw_state; + if (fsstate->cursor_exists && fsstate->cursor_level >= level) + fsstate->cursor_level = level - 1; + } + else + { + /* Forget foreign scans created in this subxact */ + if (entry->fs_level >= level) + { + ForeignScanHashDelete(node); + continue; + } + /* Disable remote cursors declared in this subxact */ + fsstate = (PgFdwScanState *) node->fdw_state; + if (fsstate->cursor_exists && fsstate->cursor_level >= level) + { + fsstate->cursor_exists = false; + fsstate->cursor_level = 0; + fsstate->scan_allowed = false; + } + } + } +} + /* * create_foreign_modify * Construct an execution state of a foreign insert/update/delete @@ -4394,6 +4558,9 @@ create_foreign_modify(EState *estate, * result if any. (This is the shared guts of postgresExecForeignInsert, * postgresExecForeignBatchInsert, postgresExecForeignUpdate, and * postgresExecForeignDelete.) + * + * Note: this function is never called from open cursors, thus no need to + * synchronize the local/remote transactions. */ static TupleTableSlot ** execute_foreign_modify(EState *estate, @@ -4832,6 +4999,9 @@ rebuild_fdw_scan_tlist(ForeignScan *fscan, List *tlist) /* * Execute a direct UPDATE/DELETE statement. + * + * Note: this function is never called from open cursors, thus no need to + * synchronize the local/remote transactions. */ static void execute_dml_stmt(ForeignScanState *node) @@ -8840,9 +9010,14 @@ fetch_more_data_begin(AsyncRequest *areq) Assert(!fsstate->conn_state->pendingAreq); - /* Create the cursor synchronously. */ + /* + * Create the cursor synchronously if not already done. Otherwise, + * synchronize the local/remote transactions before the data fetch. + */ if (!fsstate->cursor_exists) create_cursor(node); + else + pgfdw_begin_remote_xact(fsstate->conn_state->entry); /* We will send this query, but not wait for the response. */ snprintf(sql, sizeof(sql), "FETCH %d FROM c%u", diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index da7da1c2ea9..16c32497903 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -146,7 +146,8 @@ typedef struct PgFdwRelationInfo */ typedef struct PgFdwConnState { - AsyncRequest *pendingAreq; /* pending async request */ + AsyncRequest *pendingAreq; /* pending async request */ + struct ConnCacheEntry *entry; /* link to containing ConnCacheEntry */ } PgFdwConnState; /* @@ -165,6 +166,8 @@ typedef enum PgFdwSamplingMethod extern int set_transmission_modes(void); extern void reset_transmission_modes(int nestlevel); extern void process_pending_request(AsyncRequest *areq); +extern void AtEOXact_ForeignScan(bool is_commit); +extern void AtEOSubXact_ForeignScan(int level, bool is_subcommit); /* in connection.c */ extern PGconn *GetConnection(UserMapping *user, bool will_prep_stmt, @@ -173,6 +176,7 @@ extern void ReleaseConnection(PGconn *conn); extern unsigned int GetCursorNumber(PGconn *conn); extern unsigned int GetPrepStmtNumber(PGconn *conn); extern void do_sql_command(PGconn *conn, const char *sql); +extern void pgfdw_begin_remote_xact(struct ConnCacheEntry *entry); extern PGresult *pgfdw_get_result(PGconn *conn); extern PGresult *pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state); diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index ef2726722b8..de1a27257cf 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -1635,6 +1635,9 @@ ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE user_enum; -- =================================================================== -- subtransaction -- + local/remote error doesn't break cursor +-- + we allow cursors opened before a savepoint to be first fetched +-- after the savepoint, but if that subtransaction aborts, do not +-- allow them to be fetched further. -- =================================================================== BEGIN; DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; @@ -1650,6 +1653,38 @@ FETCH c; SELECT * FROM ft1 ORDER BY c1 LIMIT 1; COMMIT; +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +ABORT; + +BEGIN; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +CLOSE c; +COMMIT; + +-- cursor created in a subtransaction is handed to its parent subtransaction +BEGIN; +SAVEPOINT s; +DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; +RELEASE SAVEPOINT s; +FETCH c; +SAVEPOINT s; +FETCH c; +ROLLBACK TO s; +RELEASE SAVEPOINT s; +FETCH c; +CLOSE c; +COMMIT; + -- =================================================================== -- test handling of collations -- ===================================================================