diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 2deb4b2c4a..bd11e762b7 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -59,6 +59,7 @@ typedef struct ConnCacheEntry bool have_error; /* have any subxacts aborted in this xact? */ bool changing_xact_state; /* xact state change in process */ bool parallel_commit; /* do we commit (sub)xacts in parallel? */ + bool parallel_abort; /* do we abort (sub)xacts in parallel? */ bool invalidated; /* true if reconnect is pending */ bool keep_connections; /* setting value of keep_connections * server option */ @@ -118,13 +119,25 @@ static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue); static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry); static void pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel); static bool pgfdw_cancel_query(PGconn *conn); +static bool pgfdw_cancel_query_begin(PGconn *conn); +static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime); static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors); +static bool pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query); +static bool pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query, + bool ignore_errors, + TimestampTz endtime); static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out); static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); +static bool pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel, + List **pending_entries, + List **cancel_requested); static void pgfdw_finish_pre_commit_cleanup(List *pending_entries); static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries); +static void pgfdw_finish_abort_cleanup(List *pending_entries, + List *cancel_requested, + bool toplevel); static bool UserMappingPasswordRequired(UserMapping *user); static bool disconnect_cached_connections(Oid serverid); @@ -335,11 +348,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine to commit (sub)transactions opened on the remote server - * in parallel at (sub)transaction end. + * Also determine to commit/abort (sub)transactions opened on the remote + * server in parallel at (sub)transaction end. */ entry->keep_connections = true; entry->parallel_commit = false; + entry->parallel_abort = false; foreach(lc, server->options) { DefElem *def = (DefElem *) lfirst(lc); @@ -348,6 +362,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) entry->keep_connections = defGetBoolean(def); else if (strcmp(def->defname, "parallel_commit") == 0) entry->parallel_commit = defGetBoolean(def); + else if (strcmp(def->defname, "parallel_abort") == 0) + entry->parallel_abort = defGetBoolean(def); } /* Now try to make the connection */ @@ -933,6 +949,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) HASH_SEQ_STATUS scan; ConnCacheEntry *entry; List *pending_entries = NIL; + List *cancel_requested = NIL; /* Quick exit if no connections were touched in this transaction. */ if (!xact_got_connection) @@ -1026,7 +1043,15 @@ pgfdw_xact_callback(XactEvent event, void *arg) case XACT_EVENT_PARALLEL_ABORT: case XACT_EVENT_ABORT: /* Rollback all remote transactions during abort */ - pgfdw_abort_cleanup(entry, true); + if (entry->parallel_abort) + { + if (pgfdw_abort_cleanup_begin(entry, true, + &pending_entries, + &cancel_requested)) + continue; + } + else + pgfdw_abort_cleanup(entry, true); break; } } @@ -1036,11 +1061,21 @@ pgfdw_xact_callback(XactEvent event, void *arg) } /* If there are any pending connections, finish cleaning them up */ - if (pending_entries) + if (pending_entries || cancel_requested) { - Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT || - event == XACT_EVENT_PRE_COMMIT); - pgfdw_finish_pre_commit_cleanup(pending_entries); + if (event == XACT_EVENT_PARALLEL_PRE_COMMIT || + event == XACT_EVENT_PRE_COMMIT) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_commit_cleanup(pending_entries); + } + else + { + Assert(event == XACT_EVENT_PARALLEL_ABORT || + event == XACT_EVENT_ABORT); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + true); + } } /* @@ -1065,6 +1100,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, ConnCacheEntry *entry; int curlevel; List *pending_entries = NIL; + List *cancel_requested = NIL; /* Nothing to do at subxact start, nor after commit. */ if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB || @@ -1119,7 +1155,15 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, else { /* Rollback all remote subtransactions during abort */ - pgfdw_abort_cleanup(entry, false); + if (entry->parallel_abort) + { + if (pgfdw_abort_cleanup_begin(entry, false, + &pending_entries, + &cancel_requested)) + continue; + } + else + pgfdw_abort_cleanup(entry, false); } /* OK, we're outta that level of subtransaction */ @@ -1127,10 +1171,19 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, } /* If there are any pending connections, finish cleaning them up */ - if (pending_entries) + if (pending_entries || cancel_requested) { - Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB); - pgfdw_finish_pre_subcommit_cleanup(pending_entries); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1274,11 +1327,7 @@ pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel) static bool pgfdw_cancel_query(PGconn *conn) { - PGcancel *cancel; - char errbuf[256]; - PGresult *result = NULL; TimestampTz endtime; - bool timed_out; /* * If it takes too long to cancel the query and discard the result, assume @@ -1286,6 +1335,17 @@ pgfdw_cancel_query(PGconn *conn) */ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime); +} + +static bool +pgfdw_cancel_query_begin(PGconn *conn) +{ + PGcancel *cancel; + char errbuf[256]; + /* * Issue cancel request. Unfortunately, there's no good way to limit the * amount of time that we might block inside PQgetCancel(). @@ -1304,6 +1364,15 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime) +{ + PGresult *result = NULL; + bool timed_out; + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1338,9 +1407,7 @@ pgfdw_cancel_query(PGconn *conn) static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) { - PGresult *result = NULL; TimestampTz endtime; - bool timed_out; /* * If it takes too long to execute a cleanup query, assume the connection @@ -1350,6 +1417,14 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) */ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, ignore_errors, endtime); +} + +static bool +pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query) +{ /* * Submit a query. Since we don't use non-blocking mode, this also can * block. But its risk is relatively small, so we ignore that for now. @@ -1360,6 +1435,16 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) return false; } + return true; +} + +static bool +pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query, + bool ignore_errors, TimestampTz endtime) +{ + PGresult *result = NULL; + bool timed_out; + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1544,6 +1629,56 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +static bool +pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel, + List **pending_entries, List **cancel_requested) +{ + /* + * Don't try to clean up the connection if we're already in error + * recursion trouble. + */ + if (in_error_recursion_trouble()) + entry->changing_xact_state = true; + + /* + * If connection is already unsalvageable, don't touch it further. + */ + if (entry->changing_xact_state) + return false; + + /* + * Mark this connection as in the process of changing transaction state. + */ + entry->changing_xact_state = true; + + /* Assume we might have lost track of prepared statements */ + entry->have_error = true; + + /* + * If a command has been submitted to the remote server by using an + * asynchronous execution function, the command might not have yet + * completed. Check to see if a command is still being processed by the + * remote server, and if so, request cancellation of the command. + */ + if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE) + { + if (!pgfdw_cancel_query_begin(entry->conn)) + return false; /* Unable to cancel running query */ + *cancel_requested = lappend(*cancel_requested, entry); + } + else + { + char sql[100]; + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql)) + return false; /* Unable to abort remote transaction */ + *pending_entries = lappend(*pending_entries, entry); + } + + return true; +} + /* * Finish pre-commit cleanup of connections on which we have sent a COMMIT * command. @@ -1654,6 +1789,159 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries) } } +/* + * Finish (sub)abort cleanup of connections on which we have sent a (sub)abort + * command or cancel request. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + if (cancel_requested) + { + foreach(lc, cancel_requested) + { + ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc); + TimestampTz endtime; + char sql[100]; + + Assert(entry->changing_xact_state); + + /* + * Set end time. You might think we should do so before issuing + * cancel request like in normal mode, but that is problematic, + * because if, for example, it took longer than 30 seconds to + * process the first few entries in the cancel_requested list, it + * would cause a timeout for each of the remaining entries in the + * list when processing it, leading to slamming the connection of + * it shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + 30000); + + /* Get and discard the result of the query. */ + if (!pgfdw_cancel_query_end(entry->conn, endtime)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + } + else + pending_entries = lappend(pending_entries, entry); + } + } + + if (!pending_entries) + return; + + foreach(lc, pending_entries) + { + ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc); + TimestampTz endtime; + char sql[100]; + + Assert(entry->changing_xact_state); + + /* + * Set end time. We do this now, not before issuing the command like + * in normal mode, for the same reason as for the cancel_requested + * entries. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + + /* Get the result of the command. */ + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, false, endtime)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* + * If called for cleanup at main-transaction end, do a DEALLOCATE ALL + * if needed. + */ + if (toplevel) + { + if (entry->have_prep_stmt && entry->have_error) + { + if (!pgfdw_exec_cleanup_query_begin(entry->conn, + "DEALLOCATE ALL")) + { + /* Trouble clearing prepared statements */ + pgfdw_reset_xact_state(entry, toplevel); + } + else + pending_deallocs = lappend(pending_deallocs, entry); + continue; + } + + entry->have_prep_stmt = false; + entry->have_error = false; + } + + /* Reset the per-connection state if needed */ + if (entry->state.pendingAreq) + memset(&entry->state, 0, sizeof(entry->state)); + + /* We're done with this entry; unset the changing_xact_state flag */ + entry->changing_xact_state = false; + pgfdw_reset_xact_state(entry, toplevel); + } + + if (!pending_deallocs) + return; + Assert(toplevel); + + foreach(lc, pending_deallocs) + { + ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc); + TimestampTz endtime; + + Assert(entry->changing_xact_state); + Assert(entry->have_prep_stmt); + Assert(entry->have_error); + + /* + * Set end time. We do this now, not before issuing the command like + * in normal mode, for the same reason as for the cancel_requested + * entries. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + + /* Get the result of the command. */ + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + true, endtime)) + { + /* Trouble clearing prepared statements */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + entry->have_prep_stmt = false; + entry->have_error = false; + + /* Reset the per-connection state if needed */ + if (entry->state.pendingAreq) + memset(&entry->state, 0, sizeof(entry->state)); + + /* We're done with this entry; unset the changing_xact_state flag */ + entry->changing_xact_state = false; + pgfdw_reset_xact_state(entry, toplevel); + } +} + /* * List active foreign server connections. * diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 8043b207c5..2adb0ac125 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -9509,7 +9509,7 @@ DO $d$ END; $d$; ERROR: invalid option "password" -HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, keep_connections +HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, parallel_abort, keep_connections CONTEXT: SQL statement "ALTER SERVER loopback_nopw OPTIONS (ADD password 'dummypw')" PL/pgSQL function inline_code_block line 3 at EXECUTE -- If we add a password for our user mapping instead, we should get a different @@ -10914,10 +10914,12 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity RESET postgres_fdw.application_name; RESET debug_discard_caches; -- =================================================================== --- test parallel commit +-- test parallel commit and parallel abort -- =================================================================== ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true'); ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true'); CREATE TABLE ploc1 (f1 int, f2 text); CREATE FOREIGN TABLE prem1 (f1 int, f2 text) SERVER loopback OPTIONS (table_name 'ploc1'); @@ -10987,5 +10989,52 @@ SELECT * FROM prem2; 204 | quxqux (3 rows) +BEGIN; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; + f1 | f2 +-----+-------- + 101 | foo + 102 | foofoo + 104 | bazbaz +(3 rows) + +SELECT * FROM prem2; + f1 | f2 +-----+-------- + 201 | bar + 202 | barbar + 204 | quxqux +(3 rows) + +BEGIN; +SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ROLLBACK TO SAVEPOINT s; +RELEASE SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; + f1 | f2 +-----+-------- + 101 | foo + 102 | foofoo + 104 | bazbaz +(3 rows) + +SELECT * FROM prem2; + f1 | f2 +-----+-------- + 201 | bar + 202 | barbar + 204 | quxqux +(3 rows) + ALTER SERVER loopback OPTIONS (DROP parallel_commit); +ALTER SERVER loopback OPTIONS (DROP parallel_abort); ALTER SERVER loopback2 OPTIONS (DROP parallel_commit); +ALTER SERVER loopback2 OPTIONS (DROP parallel_abort); diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c index a09c0b6db7..e8c8b4ab36 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -122,6 +122,7 @@ postgres_fdw_validator(PG_FUNCTION_ARGS) strcmp(def->defname, "truncatable") == 0 || strcmp(def->defname, "async_capable") == 0 || strcmp(def->defname, "parallel_commit") == 0 || + strcmp(def->defname, "parallel_abort") == 0 || strcmp(def->defname, "keep_connections") == 0) { /* these accept only boolean values */ @@ -251,6 +252,7 @@ InitPgFdwOptions(void) {"async_capable", ForeignServerRelationId, false}, {"async_capable", ForeignTableRelationId, false}, {"parallel_commit", ForeignServerRelationId, false}, + {"parallel_abort", ForeignServerRelationId, false}, {"keep_connections", ForeignServerRelationId, false}, {"password_required", UserMappingRelationId, false}, diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 2dc6386b40..58ec80fad7 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3506,10 +3506,12 @@ RESET postgres_fdw.application_name; RESET debug_discard_caches; -- =================================================================== --- test parallel commit +-- test parallel commit and parallel abort -- =================================================================== ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true'); ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true'); CREATE TABLE ploc1 (f1 int, f2 text); CREATE FOREIGN TABLE prem1 (f1 int, f2 text) @@ -3548,5 +3550,26 @@ COMMIT; SELECT * FROM prem1; SELECT * FROM prem2; +BEGIN; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; +SELECT * FROM prem2; + +BEGIN; +SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ROLLBACK TO SAVEPOINT s; +RELEASE SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; +SELECT * FROM prem2; + ALTER SERVER loopback OPTIONS (DROP parallel_commit); +ALTER SERVER loopback OPTIONS (DROP parallel_abort); ALTER SERVER loopback2 OPTIONS (DROP parallel_commit); +ALTER SERVER loopback2 OPTIONS (DROP parallel_abort); diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml index 238e2f84a6..196e0a986c 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -460,10 +460,10 @@ OPTIONS (ADD password_required 'false'); When multiple remote (sub)transactions are involved in a local - (sub)transaction, by default postgres_fdw commits - those remote (sub)transactions one by one when the local (sub)transaction - commits. - Performance can be improved with the following option: + (sub)transaction, by default postgres_fdw commits or + aborts those remote (sub)transactions one by one when the local + (sub)transaction commits or aborts. + Performance can be improved with the following options: @@ -478,27 +478,40 @@ OPTIONS (ADD password_required 'false'); This option can only be specified for foreign servers, not per-table. The default is false. + + + + parallel_abort (boolean) + - If multiple foreign servers with this option enabled are involved in - a local (sub)transaction, multiple remote (sub)transactions opened on - those foreign servers in the local (sub)transaction are committed in - parallel across those foreign servers when the local (sub)transaction - commits. - - - - For a foreign server with this option enabled, if many remote - (sub)transactions are opened on the foreign server in a local - (sub)transaction, this option might increase the remote server’s load - when the local (sub)transaction commits, so be careful when using this - option. + This option controls whether postgres_fdw aborts + remote (sub)transactions opened on a foreign server in a local + (sub)transaction in parallel when the local (sub)transaction aborts. + This option can only be specified for foreign servers, not per-table. + The default is false. + + If multiple foreign servers with these options enabled are involved in a + local (sub)transaction, multiple remote (sub)transactions opened on those + foreign servers in the local (sub)transaction are committed or aborted in + parallel across those foreign servers when the local (sub)transaction + commits or aborts. + + + + For a foreign server with these options enabled, if many remote + (sub)transactions are opened on the foreign server in a local + (sub)transaction, these options might increase the remote server’s load + when the local (sub)transaction commits or aborts, so be careful when + using these options. + +