public inbox for [email protected]help / color / mirror / Atom feed
Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit 21+ messages / 5 participants [nested] [flat]
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-28 09:53 Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-02-28 09:53 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Thu, Feb 24, 2022 at 2:49 PM Etsuro Fujita <[email protected]> wrote: > I think the 0003 patch needs rebase. > I'll update the patch. Here is an updated version. I added to the 0003 patch a macro for defining the milliseconds to wait, as proposed by David upthread. Best regards, Etsuro Fujita Attachments: [application/octet-stream] v5-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch (3.2K, ../../CAPmGK15mbbsEEEULK9XTAy4c78JnCfhtZ_AnPUyjjGXSe-sbGA@mail.gmail.com/2-v5-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 8c64d42dda..fdf4cb90be 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -80,6 +80,18 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -110,8 +122,7 @@ static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors); static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out); -static void pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, - bool toplevel); +static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); static void pgfdw_finish_pre_commit_cleanup(List *pending_entries); static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel); @@ -1015,8 +1026,8 @@ pgfdw_xact_callback(XactEvent event, void *arg) break; case XACT_EVENT_PARALLEL_ABORT: case XACT_EVENT_ABORT: - - pgfdw_abort_cleanup(entry, "ABORT TRANSACTION", true); + /* Rollback all remote transactions during abort */ + pgfdw_abort_cleanup(entry, true); break; } } @@ -1109,10 +1120,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, else { /* Rollback all remote subtransactions during abort */ - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - curlevel, curlevel); - pgfdw_abort_cleanup(entry, sql, false); + pgfdw_abort_cleanup(entry, false); } /* OK, we're outta that level of subtransaction */ @@ -1465,10 +1473,7 @@ exit: ; } /* - * Abort remote transaction. - * - * The statement specified in "sql" is sent to the remote server, - * in order to rollback the remote transaction. + * Abort remote transaction or subtransaction. * * "toplevel" should be set to true if toplevel (main) transaction is * rollbacked, false otherwise. @@ -1476,8 +1481,10 @@ exit: ; * Set entry->changing_xact_state to false on success, true on failure. */ static void -pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) +pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) { + char sql[100]; + /* * Don't try to clean up the connection if we're already in error * recursion trouble. @@ -1509,8 +1516,9 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) - return; /* Unable to abort remote transaction */ + return; /* Unable to abort remote (sub)transaction */ if (toplevel) { [application/octet-stream] v5-0003-postgres-fdw-Add-support-for-parallel-abort.patch (24.4K, ../../CAPmGK15mbbsEEEULK9XTAy4c78JnCfhtZ_AnPUyjjGXSe-sbGA@mail.gmail.com/3-v5-0003-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index fdf4cb90be..89802cf5d0 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 */ @@ -80,6 +81,9 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* Milliseconds to wait to cancel a query or execute a cleanup query */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + /* macro for constructing abort command to be sent */ #define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ do { \ @@ -118,14 +122,26 @@ 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, int curlevel); +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); @@ -336,11 +352,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end. + * Also determine whether 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); @@ -349,6 +366,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 */ @@ -934,6 +953,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) @@ -1027,7 +1047,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; } } @@ -1037,11 +1065,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); + } } /* @@ -1066,6 +1104,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 || @@ -1120,7 +1159,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 */ @@ -1128,10 +1175,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1275,17 +1331,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + 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 @@ -1305,6 +1369,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)) { @@ -1339,9 +1412,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 @@ -1349,8 +1420,17 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + 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. @@ -1361,6 +1441,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)) { @@ -1545,6 +1635,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 each of which we've sent a * COMMIT command to the remote server. @@ -1651,6 +1791,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish (sub)abort cleanup of connections on each of which we've sent a + * (sub)abort command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * If cancel requests have been issued, get and discard the results of the + * queries first. + */ + 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 this 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send a (sub)abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the (sub)abort command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + 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 (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + 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 f210f91188..ed5bf9208f 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 @@ -10934,10 +10934,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'); @@ -11007,5 +11009,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 572591a558..59f865fac3 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 95b6b7192e..bd26739d9a 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3517,10 +3517,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) @@ -3559,5 +3561,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 8ebf0dc3a0..e5ca003e9b 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -461,10 +461,10 @@ OPTIONS (ADD password_required 'false'); <para> When multiple remote (sub)transactions are involved in a local - (sub)transaction, by default <filename>postgres_fdw</filename> 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 <filename>postgres_fdw</filename> 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: </para> <variablelist> @@ -479,27 +479,40 @@ OPTIONS (ADD password_required 'false'); This option can only be specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - 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. - </para> - - <para> - 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 <filename>postgres_fdw</filename> 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 <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + 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. + </para> + + <para> + 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. + </para> + </sect3> <sect3> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-03-05 10:32 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 2 replies; 21+ messages in thread From: Etsuro Fujita @ 2022-03-05 10:32 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Mon, Feb 28, 2022 at 6:53 PM Etsuro Fujita <[email protected]> wrote: > Here is an updated version. I added to the 0003 patch a macro for > defining the milliseconds to wait, as proposed by David upthread. I modified the 0003 patch further: 1) I added to pgfdw_cancel_query_end/pgfdw_exec_cleanup_query_end the PQconsumeInput optimization that we have in do_sql_command_end, and 2) I added/tweaked comments a bit further. Attached is an updated version. Like [1], I ran a simple performance test using the following transaction: BEGIN; SAVEPOINT s; INSERT INTO ft1 VALUES (10, 10); INSERT INTO ft2 VALUES (20, 20); ROLLBACK TO SAVEPOINT s; RELEASE SAVEPOINT s; INSERT INTO ft1 VALUES (10, 10); INSERT INTO ft2 VALUES (20, 20); ABORT; where ft1 is a foreign table created on a foreign server hosted on the same machine as the local server, and ft2 is a foreign table created on a foreign server hosted on a different machine. (In this test I used two machines, while in [1] I used three machines: one for the local server and the others for ft1 and ft2.) The average latencies for the ROLLBACK TO SAVEPOINT and ABORT commands over ten runs of the above transaction with the parallel_abort option disabled/enabled are: * ROLLBACK TO SAVEPOINT parallel_abort=0: 0.3217 ms parallel_abort=1: 0.2396 ms * ABORT parallel_abort=0: 0.4749 ms parallel_abort=1: 0.3733 ms This option reduces the latency for ROLLBACK TO SAVEPOINT by 25.5 percent, and the latency for ABORT by 21.4 percent. From the results, I think the patch is useful. Best regards, Etsuro Fujita [1] https://www.postgresql.org/message-id/CAPmGK17dAZCXvwnfpr1eTfknTGdt%3DhYTV9405Gt5SqPOX8K84w%40mail.g... Attachments: [application/octet-stream] v6-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch (3.2K, ../../CAPmGK15SkX=1NrY+GVyZan4Za+f17FFnM9qtQvjCEMaE3ftPcg@mail.gmail.com/2-v6-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 8c64d42dda..fdf4cb90be 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -80,6 +80,18 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -110,8 +122,7 @@ static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors); static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out); -static void pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, - bool toplevel); +static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); static void pgfdw_finish_pre_commit_cleanup(List *pending_entries); static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel); @@ -1015,8 +1026,8 @@ pgfdw_xact_callback(XactEvent event, void *arg) break; case XACT_EVENT_PARALLEL_ABORT: case XACT_EVENT_ABORT: - - pgfdw_abort_cleanup(entry, "ABORT TRANSACTION", true); + /* Rollback all remote transactions during abort */ + pgfdw_abort_cleanup(entry, true); break; } } @@ -1109,10 +1120,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, else { /* Rollback all remote subtransactions during abort */ - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - curlevel, curlevel); - pgfdw_abort_cleanup(entry, sql, false); + pgfdw_abort_cleanup(entry, false); } /* OK, we're outta that level of subtransaction */ @@ -1465,10 +1473,7 @@ exit: ; } /* - * Abort remote transaction. - * - * The statement specified in "sql" is sent to the remote server, - * in order to rollback the remote transaction. + * Abort remote transaction or subtransaction. * * "toplevel" should be set to true if toplevel (main) transaction is * rollbacked, false otherwise. @@ -1476,8 +1481,10 @@ exit: ; * Set entry->changing_xact_state to false on success, true on failure. */ static void -pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) +pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) { + char sql[100]; + /* * Don't try to clean up the connection if we're already in error * recursion trouble. @@ -1509,8 +1516,9 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) - return; /* Unable to abort remote transaction */ + return; /* Unable to abort remote (sub)transaction */ if (toplevel) { [application/octet-stream] v6-0003-postgres-fdw-Add-support-for-parallel-abort.patch (26.0K, ../../CAPmGK15SkX=1NrY+GVyZan4Za+f17FFnM9qtQvjCEMaE3ftPcg@mail.gmail.com/3-v6-0003-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index fdf4cb90be..7f209fd5b3 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 */ @@ -80,6 +81,9 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* Milliseconds to wait to cancel a query or execute a cleanup query */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + /* macro for constructing abort command to be sent */ #define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ do { \ @@ -118,14 +122,28 @@ 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, + bool consume_input); 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, + TimestampTz endtime, + bool consume_input, + bool ignore_errors); 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, int curlevel); +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); @@ -336,11 +354,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end. + * Also determine whether 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); @@ -349,6 +368,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 */ @@ -934,6 +955,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) @@ -1027,7 +1049,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; } } @@ -1037,11 +1067,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); + } } /* @@ -1066,6 +1106,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 || @@ -1120,7 +1161,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 */ @@ -1128,10 +1177,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1275,17 +1333,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime, false); +} + +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 @@ -1305,6 +1371,31 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + ereport(WARNING, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get result of cancel request: %s", + pchomp(PQerrorMessage(conn))))); + return false; + } + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1339,9 +1430,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 @@ -1349,8 +1438,18 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, endtime, + false, ignore_errors); +} +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. @@ -1361,6 +1460,30 @@ 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, + TimestampTz endtime, bool consume_input, + bool ignore_errors) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + pgfdw_report_error(WARNING, NULL, conn, false, query); + return false; + } + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1545,6 +1668,65 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Like pgfdw_abort_cleanup, submit an abort command or cancel request, but + * don't wait for the result. + * + * Returns true if the abort command or cancel request is successfully issued, + * false otherwise. If the abort command is successfully issued, the given + * connection cache entry is appended to *pending_entries. Othewise, if the + * cancel request is successfully issued, it's appended to *cancel_requested. + */ +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 each of which we've sent a * COMMIT command to the remote server. @@ -1651,6 +1833,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish abort cleanup of connections on each of which we've sent an abort + * command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * For each of the pending cancel requests (if any), get and discard the + * result of the query, and submit an abort command to the remote server. + */ + 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 this 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 error when processing each of the + * remaining entries in the list, leading to slamming that entry's + * connection shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime, true)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send an abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the abort command for each of the pending entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime, + true, false)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + if (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + endtime, true, true)) + { + /* 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 f210f91188..ed5bf9208f 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 @@ -10934,10 +10934,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'); @@ -11007,5 +11009,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 572591a558..59f865fac3 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 95b6b7192e..bd26739d9a 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3517,10 +3517,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) @@ -3559,5 +3561,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 8ebf0dc3a0..e5ca003e9b 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -461,10 +461,10 @@ OPTIONS (ADD password_required 'false'); <para> When multiple remote (sub)transactions are involved in a local - (sub)transaction, by default <filename>postgres_fdw</filename> 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 <filename>postgres_fdw</filename> 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: </para> <variablelist> @@ -479,27 +479,40 @@ OPTIONS (ADD password_required 'false'); This option can only be specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - 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. - </para> - - <para> - 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 <filename>postgres_fdw</filename> 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 <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + 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. + </para> + + <para> + 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. + </para> + </sect3> <sect3> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-03-12 01:02 David Zhang <[email protected]> parent: Etsuro Fujita <[email protected]> 1 sibling, 1 reply; 21+ messages in thread From: David Zhang @ 2022-03-12 01:02 UTC (permalink / raw) To: [email protected] Applied patches v6-0002 and v6-0003 to master branch, and the `make check` test is ok. Here is my test result in 10 times average on 3 virtual machines: before the patches: abort.1 = 2.5473 ms abort.2 = 4.1572 ms after the patches with OPTIONS (ADD parallel_abort 'true'): abort.1 = 1.7136 ms abort.2 = 2.5833 ms Overall, it has about 32 ~ 37 % improvement in my testing environment. On 2022-03-05 2:32 a.m., Etsuro Fujita wrote: > On Mon, Feb 28, 2022 at 6:53 PM Etsuro Fujita<[email protected]> wrote: >> Here is an updated version. I added to the 0003 patch a macro for >> defining the milliseconds to wait, as proposed by David upthread. > I modified the 0003 patch further: 1) I added to > pgfdw_cancel_query_end/pgfdw_exec_cleanup_query_end the PQconsumeInput > optimization that we have in do_sql_command_end, and 2) I > added/tweaked comments a bit further. Attached is an updated version. > > Like [1], I ran a simple performance test using the following transaction: > > BEGIN; > SAVEPOINT s; > INSERT INTO ft1 VALUES (10, 10); > INSERT INTO ft2 VALUES (20, 20); > ROLLBACK TO SAVEPOINT s; > RELEASE SAVEPOINT s; > INSERT INTO ft1 VALUES (10, 10); > INSERT INTO ft2 VALUES (20, 20); > ABORT; > > where ft1 is a foreign table created on a foreign server hosted on the > same machine as the local server, and ft2 is a foreign table created > on a foreign server hosted on a different machine. (In this test I > used two machines, while in [1] I used three machines: one for the > local server and the others for ft1 and ft2.) The average latencies > for the ROLLBACK TO SAVEPOINT and ABORT commands over ten runs of the > above transaction with the parallel_abort option disabled/enabled are: > > * ROLLBACK TO SAVEPOINT > parallel_abort=0: 0.3217 ms > parallel_abort=1: 0.2396 ms > > * ABORT > parallel_abort=0: 0.4749 ms > parallel_abort=1: 0.3733 ms > > This option reduces the latency for ROLLBACK TO SAVEPOINT by 25.5 > percent, and the latency for ABORT by 21.4 percent. From the results, > I think the patch is useful. > > Best regards, > Etsuro Fujita > > [1]https://www.postgresql.org/message-id/CAPmGK17dAZCXvwnfpr1eTfknTGdt%3DhYTV9405Gt5SqPOX8K84w%40mail.g... -- David Software Engineer Highgo Software Inc. (Canada) www.highgo.ca ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-03-13 09:28 Etsuro Fujita <[email protected]> parent: David Zhang <[email protected]> 0 siblings, 0 replies; 21+ messages in thread From: Etsuro Fujita @ 2022-03-13 09:28 UTC (permalink / raw) To: David Zhang <[email protected]>; +Cc: [email protected] On Sat, Mar 12, 2022 at 10:02 AM David Zhang <[email protected]> wrote: > Applied patches v6-0002 and v6-0003 to master branch, and the `make check` test is ok. > > Here is my test result in 10 times average on 3 virtual machines: > before the patches: > > abort.1 = 2.5473 ms > > abort.2 = 4.1572 ms > > after the patches with OPTIONS (ADD parallel_abort 'true'): > > abort.1 = 1.7136 ms > > abort.2 = 2.5833 ms > > Overall, it has about 32 ~ 37 % improvement in my testing environment. I think that is a great improvement. Thanks for testing! Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-03-24 04:34 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 1 sibling, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-03-24 04:34 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Sat, Mar 5, 2022 at 7:32 PM Etsuro Fujita <[email protected]> wrote: > Attached is an updated version. In the 0002 patch I introduced a macro for building an abort command in preparation for the parallel abort patch (0003), but I moved it to 0003. Attached is a new patch set. The new version of 0002 is just a cleanup patch (see the commit message in 0002), and I think it's committable, so I'm planning to commit it, if no objections. Best regards, Etsuro Fujita Attachments: [application/octet-stream] v7-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch (3.7K, ../../CAPmGK16ABpu+ff1Wiwj3MztbDi2qFNmFEa_andc3LmVbCvADCw@mail.gmail.com/2-v7-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch) download | inline diff: From 54a26d1e652f85a3fee23aa4d7b2849ceb0487f1 Mon Sep 17 00:00:00 2001 From: Etsuro Fujita <[email protected]> Date: Thu, 24 Mar 2022 13:16:09 +0900 Subject: [PATCH] postgres_fdw: Minor cleanup for pgfdw_abort_cleanup(). Commit 85c696112 introduced this function to deduplicate code in the transaction callback functions, but the SQL command passed as an argument to it was useless when it returned before aborting a remote transaction using the command. Modify pgfdw_abort_cleanup() so that it constructs the command when/if necessary, as before, removing the argument from it. Also update comments in both pgfdw_abort_cleanup() and the calling function. Etsuro Fujita, reviewed by David Zhang. Discussion: https://postgr.es/m/CAPmGK158hrd%3DZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A%40mail.gmail.com --- contrib/postgres_fdw/connection.c | 29 +++++++++++++++-------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 74d3e73205..129ca79221 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -110,8 +110,7 @@ static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors); static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out); -static void pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, - bool toplevel); +static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); static void pgfdw_finish_pre_commit_cleanup(List *pending_entries); static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel); @@ -1015,8 +1014,8 @@ pgfdw_xact_callback(XactEvent event, void *arg) break; case XACT_EVENT_PARALLEL_ABORT: case XACT_EVENT_ABORT: - - pgfdw_abort_cleanup(entry, "ABORT TRANSACTION", true); + /* Rollback all remote transactions during abort */ + pgfdw_abort_cleanup(entry, true); break; } } @@ -1109,10 +1108,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, else { /* Rollback all remote subtransactions during abort */ - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - curlevel, curlevel); - pgfdw_abort_cleanup(entry, sql, false); + pgfdw_abort_cleanup(entry, false); } /* OK, we're outta that level of subtransaction */ @@ -1465,10 +1461,7 @@ exit: ; } /* - * Abort remote transaction. - * - * The statement specified in "sql" is sent to the remote server, - * in order to rollback the remote transaction. + * Abort remote transaction or subtransaction. * * "toplevel" should be set to true if toplevel (main) transaction is * rollbacked, false otherwise. @@ -1476,8 +1469,10 @@ exit: ; * Set entry->changing_xact_state to false on success, true on failure. */ static void -pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) +pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) { + char sql[100]; + /* * Don't try to clean up the connection if we're already in error * recursion trouble. @@ -1509,8 +1504,14 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ + if (toplevel) + snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); + else + snprintf(sql, sizeof(sql), + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", + entry->xact_depth, entry->xact_depth); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) - return; /* Unable to abort remote transaction */ + return; /* Unable to abort remote (sub)transaction */ if (toplevel) { -- 2.14.3 (Apple Git-98) [application/octet-stream] v7-0003-postgres-fdw-Add-support-for-parallel-abort.patch (26.8K, ../../CAPmGK16ABpu+ff1Wiwj3MztbDi2qFNmFEa_andc3LmVbCvADCw@mail.gmail.com/3-v7-0003-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 129ca79221..a7672012f1 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 */ @@ -80,6 +81,21 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* Milliseconds to wait to cancel a query or execute a cleanup query */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + +/* Macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -106,14 +122,28 @@ 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, + bool consume_input); 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, + TimestampTz endtime, + bool consume_input, + bool ignore_errors); 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, int curlevel); +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); @@ -324,11 +354,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end. + * Also determine whether 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); @@ -337,6 +368,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 */ @@ -922,6 +955,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) @@ -1015,7 +1049,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; } } @@ -1025,11 +1067,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); + } } /* @@ -1054,6 +1106,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 || @@ -1108,7 +1161,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 */ @@ -1116,10 +1177,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1263,17 +1333,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime, false); +} + +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 @@ -1293,6 +1371,31 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + ereport(WARNING, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get result of cancel request: %s", + pchomp(PQerrorMessage(conn))))); + return false; + } + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1327,9 +1430,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 @@ -1337,8 +1438,18 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, endtime, + false, ignore_errors); +} +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. @@ -1349,6 +1460,30 @@ 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, + TimestampTz endtime, bool consume_input, + bool ignore_errors) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + pgfdw_report_error(WARNING, NULL, conn, false, query); + return false; + } + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1504,12 +1639,7 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ - if (toplevel) - snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); - else - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - entry->xact_depth, entry->xact_depth); + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) return; /* Unable to abort remote (sub)transaction */ @@ -1538,6 +1668,65 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Like pgfdw_abort_cleanup, submit an abort command or cancel request, but + * don't wait for the result. + * + * Returns true if the abort command or cancel request is successfully issued, + * false otherwise. If the abort command is successfully issued, the given + * connection cache entry is appended to *pending_entries. Othewise, if the + * cancel request is successfully issued, it's appended to *cancel_requested. + */ +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 each of which we've sent a * COMMIT command to the remote server. @@ -1644,6 +1833,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish abort cleanup of connections on each of which we've sent an abort + * command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * For each of the pending cancel requests (if any), get and discard the + * result of the query, and submit an abort command to the remote server. + */ + 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 this 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 error when processing each of the + * remaining entries in the list, leading to slamming that entry's + * connection shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime, true)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send an abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the abort command for each of the pending entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime, + true, false)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + if (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + endtime, true, true)) + { + /* 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 f210f91188..ed5bf9208f 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 @@ -10934,10 +10934,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'); @@ -11007,5 +11009,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 572591a558..59f865fac3 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 95b6b7192e..bd26739d9a 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3517,10 +3517,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) @@ -3559,5 +3561,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 d8dc715587..b86b3e9cc3 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -461,10 +461,10 @@ OPTIONS (ADD password_required 'false'); <para> When multiple remote (sub)transactions are involved in a local - (sub)transaction, by default <filename>postgres_fdw</filename> 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 <filename>postgres_fdw</filename> 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: </para> <variablelist> @@ -479,27 +479,40 @@ OPTIONS (ADD password_required 'false'); This option can only be specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - 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. - </para> - - <para> - 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 <filename>postgres_fdw</filename> 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 <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + 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. + </para> + + <para> + 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. + </para> + </sect3> <sect3> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-03-25 06:46 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-03-25 06:46 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Thu, Mar 24, 2022 at 1:34 PM Etsuro Fujita <[email protected]> wrote: > Attached is a new patch set. The new version of 0002 is just a > cleanup patch (see the commit message in 0002), and I think it's > committable, so I'm planning to commit it, if no objections. Done. Attached is the 0003 patch, which is the same as the one I sent yesterday. Best regards, Etsuro Fujita Attachments: [application/octet-stream] v7-0003-postgres-fdw-Add-support-for-parallel-abort.patch (26.8K, ../../CAPmGK16RWyc1nJsFnsn7-=UFzQiq4OGm70PK4-=5Y2fH_Gr4iA@mail.gmail.com/2-v7-0003-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 129ca79221..a7672012f1 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 */ @@ -80,6 +81,21 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* Milliseconds to wait to cancel a query or execute a cleanup query */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + +/* Macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -106,14 +122,28 @@ 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, + bool consume_input); 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, + TimestampTz endtime, + bool consume_input, + bool ignore_errors); 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, int curlevel); +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); @@ -324,11 +354,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end. + * Also determine whether 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); @@ -337,6 +368,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 */ @@ -922,6 +955,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) @@ -1015,7 +1049,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; } } @@ -1025,11 +1067,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); + } } /* @@ -1054,6 +1106,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 || @@ -1108,7 +1161,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 */ @@ -1116,10 +1177,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1263,17 +1333,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime, false); +} + +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 @@ -1293,6 +1371,31 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + ereport(WARNING, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get result of cancel request: %s", + pchomp(PQerrorMessage(conn))))); + return false; + } + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1327,9 +1430,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 @@ -1337,8 +1438,18 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, endtime, + false, ignore_errors); +} +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. @@ -1349,6 +1460,30 @@ 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, + TimestampTz endtime, bool consume_input, + bool ignore_errors) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + pgfdw_report_error(WARNING, NULL, conn, false, query); + return false; + } + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1504,12 +1639,7 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ - if (toplevel) - snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); - else - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - entry->xact_depth, entry->xact_depth); + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) return; /* Unable to abort remote (sub)transaction */ @@ -1538,6 +1668,65 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Like pgfdw_abort_cleanup, submit an abort command or cancel request, but + * don't wait for the result. + * + * Returns true if the abort command or cancel request is successfully issued, + * false otherwise. If the abort command is successfully issued, the given + * connection cache entry is appended to *pending_entries. Othewise, if the + * cancel request is successfully issued, it's appended to *cancel_requested. + */ +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 each of which we've sent a * COMMIT command to the remote server. @@ -1644,6 +1833,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish abort cleanup of connections on each of which we've sent an abort + * command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * For each of the pending cancel requests (if any), get and discard the + * result of the query, and submit an abort command to the remote server. + */ + 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 this 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 error when processing each of the + * remaining entries in the list, leading to slamming that entry's + * connection shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime, true)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send an abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the abort command for each of the pending entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime, + true, false)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + if (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + endtime, true, true)) + { + /* 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 f210f91188..ed5bf9208f 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 @@ -10934,10 +10934,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'); @@ -11007,5 +11009,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 572591a558..59f865fac3 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 95b6b7192e..bd26739d9a 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3517,10 +3517,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) @@ -3559,5 +3561,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 d8dc715587..b86b3e9cc3 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -461,10 +461,10 @@ OPTIONS (ADD password_required 'false'); <para> When multiple remote (sub)transactions are involved in a local - (sub)transaction, by default <filename>postgres_fdw</filename> 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 <filename>postgres_fdw</filename> 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: </para> <variablelist> @@ -479,27 +479,40 @@ OPTIONS (ADD password_required 'false'); This option can only be specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - 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. - </para> - - <para> - 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 <filename>postgres_fdw</filename> 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 <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + 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. + </para> + + <para> + 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. + </para> + </sect3> <sect3> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-04-19 19:54 David Zhang <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: David Zhang @ 2022-04-19 19:54 UTC (permalink / raw) To: Etsuro Fujita <[email protected]>; Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> I tried to apply the patch to master and plan to run some tests, but got below errors due to other commits. $ git apply --check v7-0003-postgres-fdw-Add-support-for-parallel-abort.patch error: patch failed: doc/src/sgml/postgres-fdw.sgml:479 error: doc/src/sgml/postgres-fdw.sgml: patch does not apply + * remote server in parallel at (sub)transaction end. Here, I think the comment above could potentially apply to multiple remote server(s). Not sure if there is a way to avoid repeated comments? For example, the same comment below appears in two places (line 231 and line 296). + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ On 2022-03-24 11:46 p.m., Etsuro Fujita wrote: > On Thu, Mar 24, 2022 at 1:34 PM Etsuro Fujita <[email protected]> wrote: >> Attached is a new patch set. The new version of 0002 is just a >> cleanup patch (see the commit message in 0002), and I think it's >> committable, so I'm planning to commit it, if no objections. > Done. > > Attached is the 0003 patch, which is the same as the one I sent yesterday. > > Best regards, > Etsuro Fujita -- Best regards, David Software Engineer Highgo Software Inc. (Canada) www.highgo.ca ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-05-02 08:25 Etsuro Fujita <[email protected]> parent: David Zhang <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-05-02 08:25 UTC (permalink / raw) To: David Zhang <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> Hi, On Wed, Apr 20, 2022 at 4:55 AM David Zhang <[email protected]> wrote: > I tried to apply the patch to master and plan to run some tests, but got > below errors due to other commits. I rebased the patch against HEAD. Attached is an updated patch. > + * remote server in parallel at (sub)transaction end. > > Here, I think the comment above could potentially apply to multiple > remote server(s). I agree on that point, but I think it's correct to say "the remote server" here, because we determine this for the given remote server. Maybe I'm missing something, so could you elaborate on it? > Not sure if there is a way to avoid repeated comments? For example, the > same comment below appears in two places (line 231 and line 296). > > + /* > + * If requested, consume whatever data is available from the socket. > + * (Note that if all data is available, this allows > + * pgfdw_get_cleanup_result to call PQgetResult without forcing the > + * overhead of WaitLatchOrSocket, which would be large compared to the > + * overhead of PQconsumeInput.) > + */ IMO I think it's OK to have this in multiple places, because 1) IMO it wouldn't be that long, and 2) we already duplicated comments like this in the same file in v14 and earlier. Here is such an example in pgfdw_xact_callback() and pgfdw_subxact_callback() in that file in those versions: /* * 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. */ Thanks for reviewing! Sorry for the delay. Best regards, Etsuro Fujita Attachments: [application/octet-stream] v8-postgres-fdw-Add-support-for-parallel-abort.patch (26.8K, ../../CAPmGK15DH+oqo=Lgn4S3ACnU3C07fSDYW_fKt09hVHskopQzxQ@mail.gmail.com/2-v8-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index f9b8c01f3b..64ffc2564d 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 */ @@ -80,6 +81,21 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* Milliseconds to wait to cancel a query or execute a cleanup query */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + +/* Macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -106,14 +122,28 @@ 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, + bool consume_input); 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, + TimestampTz endtime, + bool consume_input, + bool ignore_errors); 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, int curlevel); +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); @@ -324,11 +354,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end. + * Also determine whether 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); @@ -337,6 +368,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 */ @@ -922,6 +955,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) @@ -1015,7 +1049,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; } } @@ -1025,11 +1067,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); + } } /* @@ -1054,6 +1106,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 || @@ -1108,7 +1161,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 */ @@ -1116,10 +1177,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1263,17 +1333,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime, false); +} + +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 @@ -1293,6 +1371,31 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + ereport(WARNING, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get result of cancel request: %s", + pchomp(PQerrorMessage(conn))))); + return false; + } + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1327,9 +1430,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 @@ -1337,8 +1438,18 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, endtime, + false, ignore_errors); +} +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. @@ -1349,6 +1460,30 @@ 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, + TimestampTz endtime, bool consume_input, + bool ignore_errors) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + pgfdw_report_error(WARNING, NULL, conn, false, query); + return false; + } + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1504,12 +1639,7 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ - if (toplevel) - snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); - else - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - entry->xact_depth, entry->xact_depth); + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) return; /* Unable to abort remote (sub)transaction */ @@ -1538,6 +1668,65 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Like pgfdw_abort_cleanup, submit an abort command or cancel request, but + * don't wait for the result. + * + * Returns true if the abort command or cancel request is successfully issued, + * false otherwise. If the abort command is successfully issued, the given + * connection cache entry is appended to *pending_entries. Othewise, if the + * cancel request is successfully issued, it's appended to *cancel_requested. + */ +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 each of which we've sent a * COMMIT command to the remote server. @@ -1644,6 +1833,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish abort cleanup of connections on each of which we've sent an abort + * command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * For each of the pending cancel requests (if any), get and discard the + * result of the query, and submit an abort command to the remote server. + */ + 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 this 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 error when processing each of the + * remaining entries in the list, leading to slamming that entry's + * connection shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime, true)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send an abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the abort command for each of the pending entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime, + true, false)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + if (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + endtime, true, true)) + { + /* 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 44457f930c..c7ba7b7a27 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -9529,7 +9529,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 @@ -11218,10 +11218,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'); @@ -11291,5 +11293,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 572591a558..59f865fac3 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 92d1212027..7c5573ba9a 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3591,10 +3591,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) @@ -3633,5 +3635,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 b43d0aecba..29b46e1793 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -461,10 +461,10 @@ OPTIONS (ADD password_required 'false'); <para> When multiple remote (sub)transactions are involved in a local - (sub)transaction, by default <filename>postgres_fdw</filename> 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 <filename>postgres_fdw</filename> 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: </para> <variablelist> @@ -479,27 +479,40 @@ OPTIONS (ADD password_required 'false'); This option can only be specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - 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. - </para> - - <para> - 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 <filename>postgres_fdw</filename> 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 <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + 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. + </para> + + <para> + 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. + </para> + </sect3> <sect3> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-05-04 21:38 David Zhang <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: David Zhang @ 2022-05-04 21:38 UTC (permalink / raw) To: Etsuro Fujita <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> Thanks a lot for the patch update. On 2022-05-02 1:25 a.m., Etsuro Fujita wrote: > Hi, > > On Wed, Apr 20, 2022 at 4:55 AM David Zhang <[email protected]> wrote: >> I tried to apply the patch to master and plan to run some tests, but got >> below errors due to other commits. > I rebased the patch against HEAD. Attached is an updated patch. Applied the patch v8 to master branch today, and the `make check` is OK. I also repeated previous performance tests on three virtual Ubuntu 18.04, and the performance improvement of parallel abort in 10 times average is more consistent. before: abort.1 = 2.6344 ms abort.2 = 4.2799 ms after: abort.1 = 1.4105 ms abort.2 = 2.2075 ms >> + * remote server in parallel at (sub)transaction end. >> >> Here, I think the comment above could potentially apply to multiple >> remote server(s). > I agree on that point, but I think it's correct to say "the remote > server" here, because we determine this for the given remote server. > Maybe I'm missing something, so could you elaborate on it? Your understanding is correct. I was thinking `remote server(s)` would be easy for end user to understand but this is a comment in source code, so either way is fine for me. >> Not sure if there is a way to avoid repeated comments? For example, the >> same comment below appears in two places (line 231 and line 296). >> >> + /* >> + * If requested, consume whatever data is available from the socket. >> + * (Note that if all data is available, this allows >> + * pgfdw_get_cleanup_result to call PQgetResult without forcing the >> + * overhead of WaitLatchOrSocket, which would be large compared to the >> + * overhead of PQconsumeInput.) >> + */ > IMO I think it's OK to have this in multiple places, because 1) IMO it > wouldn't be that long, and 2) we already duplicated comments like this > in the same file in v14 and earlier. Here is such an example in > pgfdw_xact_callback() and pgfdw_subxact_callback() in that file in > those versions: > > /* > * 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. > */ > > Thanks for reviewing! Sorry for the delay. > > Best regards, > Etsuro Fujita -- Best regards, David Software Engineer Highgo Software Inc. (Canada) www.highgo.ca ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-05-06 10:08 Etsuro Fujita <[email protected]> parent: David Zhang <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-05-06 10:08 UTC (permalink / raw) To: David Zhang <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On Thu, May 5, 2022 at 6:39 AM David Zhang <[email protected]> wrote: > On 2022-05-02 1:25 a.m., Etsuro Fujita wrote: > > On Wed, Apr 20, 2022 at 4:55 AM David Zhang <[email protected]> wrote: > >> I tried to apply the patch to master and plan to run some tests, but got > >> below errors due to other commits. > > I rebased the patch against HEAD. Attached is an updated patch. > > Applied the patch v8 to master branch today, and the `make check` is OK. > I also repeated previous performance tests on three virtual Ubuntu > 18.04, and the performance improvement of parallel abort in 10 times > average is more consistent. > > before: > > abort.1 = 2.6344 ms > abort.2 = 4.2799 ms > > after: > > abort.1 = 1.4105 ms > abort.2 = 2.2075 ms Good to know! Thanks for testing! > >> + * remote server in parallel at (sub)transaction end. > >> > >> Here, I think the comment above could potentially apply to multiple > >> remote server(s). > > I agree on that point, but I think it's correct to say "the remote > > server" here, because we determine this for the given remote server. > > Maybe I'm missing something, so could you elaborate on it? > Your understanding is correct. I was thinking `remote server(s)` would > be easy for end user to understand but this is a comment in source code, > so either way is fine for me. Ok, but I noticed that the comment failed to mention that the parallel_commit option is disabled by default. Also, I noticed a comment above it: * It's enough to determine this only when making new connection because * all the connections to the foreign server whose keep_connections option * is changed will be closed and re-made later. This would apply to the parallel_commit option as well. How about updating these like the attached? (I simplified the latter comment and moved it to a more appropriate place.) Best regards, Etsuro Fujita Attachments: [application/octet-stream] update-comment-in-make_new_connection.patch (1.1K, ../../CAPmGK16Kg2Bf90sqzcZ4YM5cN_G-4h7wFUS01qQpqNB+2BG5_w@mail.gmail.com/2-update-comment-in-make_new_connection.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index f9b8c01f3b..541526ab80 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -318,14 +318,15 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * open even after the transaction using it ends, so that the subsequent * transactions can re-use it. * - * It's enough to determine this only when making new connection because - * all the connections to the foreign server whose keep_connections option - * is changed will be closed and re-made later. - * * By default, all the connections to any foreign servers are kept open. * * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end. + * server in parallel at (sub)transaction end, which is disabled by + * default. + * + * Note: it's enough to determine these only when making a new connection + * because these settings for it are changed, it will be closed and + * re-made later. */ entry->keep_connections = true; entry->parallel_commit = false; ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-05-11 10:39 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-05-11 10:39 UTC (permalink / raw) To: David Zhang <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On Fri, May 6, 2022 at 7:08 PM Etsuro Fujita <[email protected]> wrote: > > > On Wed, Apr 20, 2022 at 4:55 AM David Zhang <[email protected]> wrote: > > >> + * remote server in parallel at (sub)transaction end. > I noticed that the comment failed to mention that the > parallel_commit option is disabled by default. Also, I noticed a > comment above it: > > * It's enough to determine this only when making new connection because > * all the connections to the foreign server whose keep_connections option > * is changed will be closed and re-made later. > > This would apply to the parallel_commit option as well. How about > updating these like the attached? (I simplified the latter comment > and moved it to a more appropriate place.) I’m planning to commit this as a follow-up patch for commit 04e706d42. Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-05-12 08:46 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-05-12 08:46 UTC (permalink / raw) To: David Zhang <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On Wed, May 11, 2022 at 7:39 PM Etsuro Fujita <[email protected]> wrote: > On Fri, May 6, 2022 at 7:08 PM Etsuro Fujita <[email protected]> wrote: > > > > On Wed, Apr 20, 2022 at 4:55 AM David Zhang <[email protected]> wrote: > > > >> + * remote server in parallel at (sub)transaction end. > > > I noticed that the comment failed to mention that the > > parallel_commit option is disabled by default. Also, I noticed a > > comment above it: > > > > * It's enough to determine this only when making new connection because > > * all the connections to the foreign server whose keep_connections option > > * is changed will be closed and re-made later. > > > > This would apply to the parallel_commit option as well. How about > > updating these like the attached? (I simplified the latter comment > > and moved it to a more appropriate place.) > > I’m planning to commit this as a follow-up patch for commit 04e706d42. Done. Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-06-30 18:50 Jacob Champion <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Jacob Champion @ 2022-06-30 18:50 UTC (permalink / raw) To: Etsuro Fujita <[email protected]>; David Zhang <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On 5/12/22 01:46, Etsuro Fujita wrote: > On Wed, May 11, 2022 at 7:39 PM Etsuro Fujita <[email protected]> wrote: >> I’m planning to commit this as a follow-up patch for commit 04e706d42. > > Done. FYI, I think cfbot is confused about the patch under review here. (When I first opened the thread I thought the patch had already been committed.) For new reviewers: it looks like v8, upthread, is the proposal. --Jacob ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-07-07 08:22 Etsuro Fujita <[email protected]> parent: Jacob Champion <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-07-07 08:22 UTC (permalink / raw) To: Jacob Champion <[email protected]>; +Cc: David Zhang <[email protected]>; Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> Hi Jacob, On Fri, Jul 1, 2022 at 3:50 AM Jacob Champion <[email protected]> wrote: > On 5/12/22 01:46, Etsuro Fujita wrote: > > On Wed, May 11, 2022 at 7:39 PM Etsuro Fujita <[email protected]> wrote: > >> I’m planning to commit this as a follow-up patch for commit 04e706d42. > > > > Done. > > FYI, I think cfbot is confused about the patch under review here. (When > I first opened the thread I thought the patch had already been committed.) I should have attached the patch in the previous email. > For new reviewers: it looks like v8, upthread, is the proposal. The patch needs rebase due to commits 4036bcbbb, 8c8d307f8 and 82699edbf, so I updated the patch. Attached is a new version, in which I also tweaked comments a little bit. Thanks for taking care of this! Best regards, Etsuro Fujita Attachments: [application/octet-stream] v9-postgres-fdw-Add-support-for-parallel-abort.patch (27.3K, ../../CAPmGK16j4yiWnFngGAJM_05ZwPSZWHW0xwb_5YrACd40=AyJyQ@mail.gmail.com/2-v9-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index cffb6f8310..c2312c4e50 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 */ @@ -80,6 +81,25 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* + * Milliseconds to wait to cancel an in-progress query or execute a cleanup + * query; if it takes longer than 30 seconds to do these, we assume the + * connection is dead. + */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + +/* Macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -106,14 +126,28 @@ 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, + bool consume_input); 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, + TimestampTz endtime, + bool consume_input, + bool ignore_errors); 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, int curlevel); +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); @@ -320,8 +354,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end, which is disabled by + * Also determine whether to commit/abort (sub)transactions opened on the + * remote server in parallel at (sub)transaction end, which is disabled by * default. * * Note: it's enough to determine these only when making a new connection @@ -330,6 +364,7 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) */ entry->keep_connections = true; entry->parallel_commit = false; + entry->parallel_abort = false; foreach(lc, server->options) { DefElem *def = (DefElem *) lfirst(lc); @@ -338,6 +373,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 */ @@ -923,6 +960,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) @@ -1016,7 +1054,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; } } @@ -1026,11 +1072,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); + } } /* @@ -1055,6 +1111,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 || @@ -1109,7 +1166,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 */ @@ -1117,10 +1182,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1264,17 +1338,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime, false); +} + +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 @@ -1294,6 +1376,31 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + ereport(WARNING, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get result of cancel request: %s", + pchomp(PQerrorMessage(conn))))); + return false; + } + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1328,9 +1435,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 @@ -1338,8 +1443,18 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, endtime, + false, ignore_errors); +} +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. @@ -1350,6 +1465,30 @@ 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, + TimestampTz endtime, bool consume_input, + bool ignore_errors) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + pgfdw_report_error(WARNING, NULL, conn, false, query); + return false; + } + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1505,12 +1644,7 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ - if (toplevel) - snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); - else - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - entry->xact_depth, entry->xact_depth); + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) return; /* Unable to abort remote (sub)transaction */ @@ -1539,6 +1673,65 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Like pgfdw_abort_cleanup, submit an abort command or cancel a request, but + * don't wait for the result. + * + * Returns true if the abort command or cancel request is successfully issued, + * false otherwise. If the abort command is successfully issued, the given + * connection cache entry is appended to *pending_entries. Othewise, if the + * cancel request is successfully issued, it's appended to *cancel_requested. + */ +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 each of which we've sent a * COMMIT command to the remote server. @@ -1647,6 +1840,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish abort cleanup of connections on each of which we've sent an abort + * command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * For each of the pending cancel requests (if any), get and discard the + * result of the query, and submit an abort command to the remote server. + */ + 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 this 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 error when processing each of the + * remaining entries in the list, leading to slamming that entry's + * connection shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime, true)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send an abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the abort command for each of the pending entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime, + true, false)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + if (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + endtime, true, true)) + { + /* 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 44457f930c..c7ba7b7a27 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -9529,7 +9529,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 @@ -11218,10 +11218,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'); @@ -11291,5 +11293,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 572591a558..59f865fac3 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 92d1212027..7c5573ba9a 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3591,10 +3591,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) @@ -3633,5 +3635,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 bfd344cdc0..7b8b959462 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -465,12 +465,13 @@ OPTIONS (ADD password_required 'false'); corresponding remote transactions, and subtransactions are managed by creating corresponding remote subtransactions. When multiple remote transactions are involved in the current local transaction, by default - <filename>postgres_fdw</filename> commits those remote transactions - serially when the local transaction is committed. When multiple remote - subtransactions are involved in the current local subtransaction, by - default <filename>postgres_fdw</filename> commits those remote - subtransactions serially when the local subtransaction is committed. - Performance can be improved with the following option: + <filename>postgres_fdw</filename> commits or aborts those remote + transactions serially when the local transaction is committed or aborted. + When multiple remote subtransactions are involved in the current local + subtransaction, by default <filename>postgres_fdw</filename> commits or + aborts those remote subtransactions serially when the local subtransaction + is committed or abortd. + Performance can be improved with the following options: </para> <variablelist> @@ -486,24 +487,38 @@ OPTIONS (ADD password_required 'false'); specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - If multiple foreign servers with this option enabled are involved in a - local transaction, multiple remote transactions on those foreign - servers are committed in parallel across those foreign servers when - the local transaction is committed. - </para> - - <para> - When this option is enabled, a foreign server with many remote - transactions may see a negative performance impact when the local - transaction is committed. + This option controls whether <filename>postgres_fdw</filename> aborts + in parallel remote transactions opened on a foreign server in a local + transaction when the local transaction is aborted. This setting also + applies to remote and local subtransactions. This option can only be + specified for foreign servers, not per-table. The default is + <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + If multiple foreign servers with these options enabled are involved in a + local transaction, multiple remote transactions on those foreign servers + are committed or aborted in parallel across those foreign servers when + the local transaction is committed or aborted. + </para> + + <para> + When these options are enabled, a foreign server with many remote + transactions may see a negative performance impact when the local + transaction is committed or aborted. + </para> + </sect3> <sect3> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-09-30 20:54 David Zhang <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: David Zhang @ 2022-09-30 20:54 UTC (permalink / raw) To: Etsuro Fujita <[email protected]>; Jacob Champion <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> Hi Etsuro, > The patch needs rebase due to commits 4036bcbbb, 8c8d307f8 and > 82699edbf, so I updated the patch. Attached is a new version, in > which I also tweaked comments a little bit. After rebase the file `postgres_fdw.out` and applied to master branch, make and make check are all ok for postgres_fdw. -- David Software Engineer Highgo Software Inc. (Canada) www.highgo.ca ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-11-01 10:24 Etsuro Fujita <[email protected]> parent: David Zhang <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2022-11-01 10:24 UTC (permalink / raw) To: David Zhang <[email protected]>; +Cc: Jacob Champion <[email protected]>; Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> Hi David, On Sat, Oct 1, 2022 at 5:54 AM David Zhang <[email protected]> wrote: > After rebase the file `postgres_fdw.out` and applied to master branch, > make and make check are all ok for postgres_fdw. Thanks for testing! Attached is a rebased version of the patch. Best regards, Etsuro Fujita Attachments: [application/octet-stream] v10-postgres-fdw-Add-support-for-parallel-abort.patch (25.9K, ../../CAPmGK14BggQDrP-KuuNpC59bwbsUZ5Aohx+jFoU9ptgprBwStw@mail.gmail.com/2-v10-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index f0c45b00db..864f9516ad 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 */ @@ -80,6 +81,25 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* + * Milliseconds to wait to cancel an in-progress query or execute a cleanup + * query; if it takes longer than 30 seconds to do these, we assume the + * connection is dead. + */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + +/* Macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -106,14 +126,28 @@ 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, + bool consume_input); 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, + TimestampTz endtime, + bool consume_input, + bool ignore_errors); 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, int curlevel); +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); @@ -320,8 +354,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end, which is disabled by + * Also determine whether to commit/abort (sub)transactions opened on the + * remote server in parallel at (sub)transaction end, which is disabled by * default. * * Note: it's enough to determine these only when making a new connection @@ -330,6 +364,7 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) */ entry->keep_connections = true; entry->parallel_commit = false; + entry->parallel_abort = false; foreach(lc, server->options) { DefElem *def = (DefElem *) lfirst(lc); @@ -338,6 +373,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 */ @@ -923,6 +960,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) @@ -1016,7 +1054,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; } } @@ -1026,11 +1072,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); + } } /* @@ -1055,6 +1111,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 || @@ -1109,7 +1166,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 */ @@ -1117,10 +1182,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1264,17 +1338,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime, false); +} + +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 @@ -1294,6 +1376,31 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + ereport(WARNING, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get result of cancel request: %s", + pchomp(PQerrorMessage(conn))))); + return false; + } + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1328,9 +1435,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 @@ -1338,8 +1443,18 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, endtime, + false, ignore_errors); +} +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. @@ -1350,6 +1465,30 @@ 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, + TimestampTz endtime, bool consume_input, + bool ignore_errors) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + pgfdw_report_error(WARNING, NULL, conn, false, query); + return false; + } + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1505,12 +1644,7 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ - if (toplevel) - snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); - else - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - entry->xact_depth, entry->xact_depth); + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) return; /* Unable to abort remote (sub)transaction */ @@ -1539,6 +1673,65 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Like pgfdw_abort_cleanup, submit an abort command or cancel a request, but + * don't wait for the result. + * + * Returns true if the abort command or cancel request is successfully issued, + * false otherwise. If the abort command is successfully issued, the given + * connection cache entry is appended to *pending_entries. Othewise, if the + * cancel request is successfully issued, it's appended to *cancel_requested. + */ +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 each of which we've sent a * COMMIT command to the remote server. @@ -1647,6 +1840,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish abort cleanup of connections on each of which we've sent an abort + * command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * For each of the pending cancel requests (if any), get and discard the + * result of the query, and submit an abort command to the remote server. + */ + 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 this 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 error when processing each of the + * remaining entries in the list, leading to slamming that entry's + * connection shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime, true)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send an abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the abort command for each of the pending entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime, + true, false)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + if (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + endtime, true, true)) + { + /* 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 558e94b845..5ddb483cd5 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -11501,10 +11501,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'); @@ -11574,5 +11576,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 fa80ee2a55..df6680fe27 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -125,6 +125,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 */ @@ -254,6 +255,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 b0dbb41fb5..4d5140669b 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3744,10 +3744,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) @@ -3786,5 +3788,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 527f4deaaa..2b309ad002 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -469,12 +469,13 @@ OPTIONS (ADD password_required 'false'); corresponding remote transactions, and subtransactions are managed by creating corresponding remote subtransactions. When multiple remote transactions are involved in the current local transaction, by default - <filename>postgres_fdw</filename> commits those remote transactions - serially when the local transaction is committed. When multiple remote - subtransactions are involved in the current local subtransaction, by - default <filename>postgres_fdw</filename> commits those remote - subtransactions serially when the local subtransaction is committed. - Performance can be improved with the following option: + <filename>postgres_fdw</filename> commits or aborts those remote + transactions serially when the local transaction is committed or aborted. + When multiple remote subtransactions are involved in the current local + subtransaction, by default <filename>postgres_fdw</filename> commits or + aborts those remote subtransactions serially when the local subtransaction + is committed or abortd. + Performance can be improved with the following options: </para> <variablelist> @@ -490,24 +491,38 @@ OPTIONS (ADD password_required 'false'); specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - If multiple foreign servers with this option enabled are involved in a - local transaction, multiple remote transactions on those foreign - servers are committed in parallel across those foreign servers when - the local transaction is committed. - </para> - - <para> - When this option is enabled, a foreign server with many remote - transactions may see a negative performance impact when the local - transaction is committed. + This option controls whether <filename>postgres_fdw</filename> aborts + in parallel remote transactions opened on a foreign server in a local + transaction when the local transaction is aborted. This setting also + applies to remote and local subtransactions. This option can only be + specified for foreign servers, not per-table. The default is + <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + If multiple foreign servers with these options enabled are involved in a + local transaction, multiple remote transactions on those foreign servers + are committed or aborted in parallel across those foreign servers when + the local transaction is committed or aborted. + </para> + + <para> + When these options are enabled, a foreign server with many remote + transactions may see a negative performance impact when the local + transaction is committed or aborted. + </para> + </sect3> <sect3> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2023-01-04 12:19 vignesh C <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: vignesh C @ 2023-01-04 12:19 UTC (permalink / raw) To: Etsuro Fujita <[email protected]>; +Cc: David Zhang <[email protected]>; Jacob Champion <[email protected]>; Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On Tue, 1 Nov 2022 at 15:54, Etsuro Fujita <[email protected]> wrote: > > Hi David, > > On Sat, Oct 1, 2022 at 5:54 AM David Zhang <[email protected]> wrote: > > After rebase the file `postgres_fdw.out` and applied to master branch, > > make and make check are all ok for postgres_fdw. > > Thanks for testing! Attached is a rebased version of the patch. The patch does not apply on top of HEAD as in [1], please post a rebased patch: === Applying patches on top of PostgreSQL commit ID b37a0832396414e8469d4ee4daea33396bde39b0 === === applying patch ./v10-postgres-fdw-Add-support-for-parallel-abort.patch patching file contrib/postgres_fdw/connection.c patching file contrib/postgres_fdw/expected/postgres_fdw.out Hunk #1 succeeded at 11704 (offset 203 lines). Hunk #2 FAILED at 11576. 1 out of 2 hunks FAILED -- saving rejects to file contrib/postgres_fdw/expected/postgres_fdw.out.rej patching file contrib/postgres_fdw/option.c Hunk #2 succeeded at 272 (offset 17 lines). patching file contrib/postgres_fdw/sql/postgres_fdw.sql Hunk #1 succeeded at 3894 (offset 150 lines). Hunk #2 FAILED at 3788. 1 out of 2 hunks FAILED -- saving rejects to file contrib/postgres_fdw/sql/postgres_fdw.sql.rej [1] - http://cfbot.cputube.org/patch_41_3392.log Regards, Vignesh ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2023-01-18 11:06 Etsuro Fujita <[email protected]> parent: vignesh C <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2023-01-18 11:06 UTC (permalink / raw) To: vignesh C <[email protected]>; +Cc: David Zhang <[email protected]>; Jacob Champion <[email protected]>; Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> Hi Vignesh, On Wed, Jan 4, 2023 at 9:19 PM vignesh C <[email protected]> wrote: > On Tue, 1 Nov 2022 at 15:54, Etsuro Fujita <[email protected]> wrote: > > Attached is a rebased version of the patch. > > The patch does not apply on top of HEAD as in [1], please post a rebased patch: I rebased the patch. Attached is an updated patch. Thanks! Best regards, Etsuro Fujita Attachments: [application/octet-stream] v11-postgres-fdw-Add-support-for-parallel-abort.patch (26.2K, ../../CAPmGK15Q6XqGq9gHWyu98ochGexFLXqMaQaxa5ukS0P58-g_xA@mail.gmail.com/2-v11-postgres-fdw-Add-support-for-parallel-abort.patch) download | inline diff: diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index ed75ce3f79..781cbcb94e 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 */ @@ -80,6 +81,25 @@ static unsigned int prep_stmt_number = 0; /* tracks whether any work is needed in callback functions */ static bool xact_got_connection = false; +/* + * Milliseconds to wait to cancel an in-progress query or execute a cleanup + * query; if it takes longer than 30 seconds to do these, we assume the + * connection is dead. + */ +#define CONNECTION_CLEANUP_TIMEOUT 30000 + +/* Macro for constructing abort command to be sent */ +#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \ + do { \ + if (toplevel) \ + snprintf((sql), sizeof(sql), \ + "ABORT TRANSACTION"); \ + else \ + snprintf((sql), sizeof(sql), \ + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \ + (entry)->xact_depth, (entry)->xact_depth); \ + } while(0) + /* * SQL functions */ @@ -106,14 +126,28 @@ 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, + bool consume_input); 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, + TimestampTz endtime, + bool consume_input, + bool ignore_errors); 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, int curlevel); +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); @@ -320,8 +354,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine whether to commit (sub)transactions opened on the remote - * server in parallel at (sub)transaction end, which is disabled by + * Also determine whether to commit/abort (sub)transactions opened on the + * remote server in parallel at (sub)transaction end, which is disabled by * default. * * Note: it's enough to determine these only when making a new connection @@ -330,6 +364,7 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) */ entry->keep_connections = true; entry->parallel_commit = false; + entry->parallel_abort = false; foreach(lc, server->options) { DefElem *def = (DefElem *) lfirst(lc); @@ -338,6 +373,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 */ @@ -923,6 +960,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) @@ -1016,7 +1054,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; } } @@ -1026,11 +1072,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); + } } /* @@ -1055,6 +1111,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 || @@ -1109,7 +1166,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 */ @@ -1117,10 +1182,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, curlevel); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1264,17 +1338,25 @@ 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 * the connection is dead. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime, false); +} + +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 @@ -1294,6 +1376,31 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime, bool consume_input) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + ereport(WARNING, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not get result of cancel request: %s", + pchomp(PQerrorMessage(conn))))); + return false; + } + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1328,9 +1435,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 @@ -1338,8 +1443,18 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) * place (e.g. statement timeout, user cancel), so the timeout shouldn't * be too long. */ - endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, endtime, + false, ignore_errors); +} +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. @@ -1350,6 +1465,30 @@ 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, + TimestampTz endtime, bool consume_input, + bool ignore_errors) +{ + PGresult *result = NULL; + bool timed_out; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows + * pgfdw_get_cleanup_result to call PQgetResult without forcing the + * overhead of WaitLatchOrSocket, which would be large compared to the + * overhead of PQconsumeInput.) + */ + if (consume_input && !PQconsumeInput(conn)) + { + pgfdw_report_error(WARNING, NULL, conn, false, query); + return false; + } + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1505,12 +1644,7 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) !pgfdw_cancel_query(entry->conn)) return; /* Unable to cancel running query */ - if (toplevel) - snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); - else - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - entry->xact_depth, entry->xact_depth); + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) return; /* Unable to abort remote (sub)transaction */ @@ -1539,6 +1673,65 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Like pgfdw_abort_cleanup, submit an abort command or cancel a request, but + * don't wait for the result. + * + * Returns true if the abort command or cancel request is successfully issued, + * false otherwise. If the abort command is successfully issued, the given + * connection cache entry is appended to *pending_entries. Othewise, if the + * cancel request is successfully issued, it's appended to *cancel_requested. + */ +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 each of which we've sent a * COMMIT command to the remote server. @@ -1647,6 +1840,168 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) } } +/* + * Finish abort cleanup of connections on each of which we've sent an abort + * command or cancel request to the remote server. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + /* + * For each of the pending cancel requests (if any), get and discard the + * result of the query, and submit an abort command to the remote server. + */ + 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 this 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 error when processing each of the + * remaining entries in the list, leading to slamming that entry's + * connection shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_cancel_query_end(entry->conn, endtime, true)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* Send an abort command in parallel if needed */ + 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); + } + } + + /* No further work if no pending entries */ + if (!pending_entries) + return; + + /* + * Get the result of the abort command for each of the pending entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, endtime, + true, false)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + if (toplevel) + { + /* Do a DEALLOCATE ALL in parallel if needed */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + Assert(toplevel); + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + 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(), + CONNECTION_CLEANUP_TIMEOUT); + + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + endtime, true, true)) + { + /* 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 c0267a99d2..0271566fcf 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -11704,10 +11704,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'); @@ -11777,8 +11779,55 @@ 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); -- =================================================================== -- test for ANALYZE sampling -- =================================================================== diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c index 984e4d168a..ab00ca9df3 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -125,6 +125,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 */ @@ -271,6 +272,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 c37aa80383..8ef78fc190 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3894,10 +3894,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) @@ -3936,8 +3938,29 @@ 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); -- =================================================================== -- test for ANALYZE sampling diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml index 78f2d7d8d5..8b84381c27 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -504,12 +504,13 @@ OPTIONS (ADD password_required 'false'); corresponding remote transactions, and subtransactions are managed by creating corresponding remote subtransactions. When multiple remote transactions are involved in the current local transaction, by default - <filename>postgres_fdw</filename> commits those remote transactions - serially when the local transaction is committed. When multiple remote - subtransactions are involved in the current local subtransaction, by - default <filename>postgres_fdw</filename> commits those remote - subtransactions serially when the local subtransaction is committed. - Performance can be improved with the following option: + <filename>postgres_fdw</filename> commits or aborts those remote + transactions serially when the local transaction is committed or aborted. + When multiple remote subtransactions are involved in the current local + subtransaction, by default <filename>postgres_fdw</filename> commits or + aborts those remote subtransactions serially when the local subtransaction + is committed or abortd. + Performance can be improved with the following options: </para> <variablelist> @@ -525,24 +526,38 @@ OPTIONS (ADD password_required 'false'); specified for foreign servers, not per-table. The default is <literal>false</literal>. </para> + </listitem> + </varlistentry> + <varlistentry> + <term><literal>parallel_abort</literal> (<type>boolean</type>)</term> + <listitem> <para> - If multiple foreign servers with this option enabled are involved in a - local transaction, multiple remote transactions on those foreign - servers are committed in parallel across those foreign servers when - the local transaction is committed. - </para> - - <para> - When this option is enabled, a foreign server with many remote - transactions may see a negative performance impact when the local - transaction is committed. + This option controls whether <filename>postgres_fdw</filename> aborts + in parallel remote transactions opened on a foreign server in a local + transaction when the local transaction is aborted. This setting also + applies to remote and local subtransactions. This option can only be + specified for foreign servers, not per-table. The default is + <literal>false</literal>. </para> </listitem> </varlistentry> </variablelist> + <para> + If multiple foreign servers with these options enabled are involved in a + local transaction, multiple remote transactions on those foreign servers + are committed or aborted in parallel across those foreign servers when + the local transaction is committed or aborted. + </para> + + <para> + When these options are enabled, a foreign server with many remote + transactions may see a negative performance impact when the local + transaction is committed or aborted. + </para> + </sect3> <sect3 id="postgres-fdw-options-updatability"> ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2023-04-04 10:28 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 21+ messages in thread From: Etsuro Fujita @ 2023-04-04 10:28 UTC (permalink / raw) To: vignesh C <[email protected]>; +Cc: David Zhang <[email protected]>; Jacob Champion <[email protected]>; Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On Wed, Jan 18, 2023 at 8:06 PM Etsuro Fujita <[email protected]> wrote: > I rebased the patch. Attached is an updated patch. The parallel-abort patch received a review from David, and I addressed his comments. Also, he tested with the patch, and showed that it reduces time taken to abort remote transactions. So, if there are no objections, I will commit the patch. Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 21+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2023-04-06 08:41 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 0 replies; 21+ messages in thread From: Etsuro Fujita @ 2023-04-06 08:41 UTC (permalink / raw) To: vignesh C <[email protected]>; +Cc: David Zhang <[email protected]>; Jacob Champion <[email protected]>; Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On Tue, Apr 4, 2023 at 7:28 PM Etsuro Fujita <[email protected]> wrote: > The parallel-abort patch received a review from David, and I addressed > his comments. Also, he tested with the patch, and showed that it > reduces time taken to abort remote transactions. So, if there are no > objections, I will commit the patch. Pushed after adding/modifying comments a little bit. Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 21+ messages in thread
* [PATCH v4 14/19] Vacuum second pass emits XLOG_HEAP2_PRUNE record @ 2024-03-19 22:50 Melanie Plageman <[email protected]> 0 siblings, 0 replies; 21+ messages in thread From: Melanie Plageman @ 2024-03-19 22:50 UTC (permalink / raw) Remove the XLOG_HEAP2_VACUUM record and update vacuum's second pass to emit a XLOG_HEAP2_PRUNE record. This temporarily wastes some space but a future commit will streamline xl_heap_prune and ensure that no unused members are included in the WAL record. --- src/backend/access/heap/heapam.c | 94 ++++-------------------- src/backend/access/heap/pruneheap.c | 67 ++++++++++------- src/backend/access/heap/vacuumlazy.c | 12 ++- src/backend/access/rmgrdesc/heapdesc.c | 20 ----- src/backend/replication/logical/decode.c | 1 - src/include/access/heapam.h | 2 +- src/include/access/heapam_xlog.h | 33 +++++---- 7 files changed, 85 insertions(+), 144 deletions(-) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 532868039d5..16bab55ba02 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -8717,23 +8717,34 @@ heap_xlog_prune(XLogReaderState *record) BlockNumber blkno; XLogRedoAction action; bool get_cleanup_lock; + bool lp_truncate_only; XLogRecGetBlockTag(record, 0, &rlocator, NULL, &blkno); + lp_truncate_only = xlrec->flags & XLHP_LP_TRUNCATE_ONLY; + /* * If there are dead, redirected, or unused items set unused by * heap_page_prune_and_freeze(), heap_page_prune_execute() will call * PageRepairFragementation() which expects a full cleanup lock. */ get_cleanup_lock = xlrec->nredirected > 0 || - xlrec->ndead > 0 || xlrec->nunused > 0; + xlrec->ndead > 0 || + (xlrec->nunused > 0 && !lp_truncate_only); + + if (lp_truncate_only) + { + Assert(xlrec->nredirected == 0); + Assert(xlrec->ndead == 0); + Assert(xlrec->nunused > 0); + } /* * We are either about to remove tuples or freeze them. In Hot Standby * mode, ensure that there's no queries running for which any removed * tuples are still visible or which consider the frozen xids as running. */ - if (InHotStandby) + if (xlrec->flags & XLHP_HAS_CONFLICT_HORIZON && InHotStandby) ResolveRecoveryConflictWithSnapshot(xlrec->snapshotConflictHorizon, xlrec->isCatalogRel, rlocator); @@ -8772,7 +8783,7 @@ heap_xlog_prune(XLogReaderState *record) /* Update all line pointers per the record, and repair fragmentation */ if (nredirected > 0 || ndead > 0 || nunused > 0) - heap_page_prune_execute(buffer, + heap_page_prune_execute(buffer, lp_truncate_only, redirected, nredirected, nowdead, ndead, nowunused, nunused); @@ -8819,7 +8830,7 @@ heap_xlog_prune(XLogReaderState *record) UnlockReleaseBuffer(buffer); /* - * After pruning records from a page, it's useful to update the FSM + * After modifying records on a page, it's useful to update the FSM * about it, as it may cause the page become target for insertions * later even if vacuum decides not to visit it (which is possible if * gets marked all-visible.) @@ -8831,78 +8842,6 @@ heap_xlog_prune(XLogReaderState *record) } } -/* - * Handles XLOG_HEAP2_VACUUM record type. - * - * Acquires an ordinary exclusive lock only. - */ -static void -heap_xlog_vacuum(XLogReaderState *record) -{ - XLogRecPtr lsn = record->EndRecPtr; - xl_heap_vacuum *xlrec = (xl_heap_vacuum *) XLogRecGetData(record); - Buffer buffer; - BlockNumber blkno; - XLogRedoAction action; - - /* - * If we have a full-page image, restore it (without using a cleanup lock) - * and we're done. - */ - action = XLogReadBufferForRedoExtended(record, 0, RBM_NORMAL, false, - &buffer); - if (action == BLK_NEEDS_REDO) - { - Page page = (Page) BufferGetPage(buffer); - OffsetNumber *nowunused; - Size datalen; - OffsetNumber *offnum; - - nowunused = (OffsetNumber *) XLogRecGetBlockData(record, 0, &datalen); - - /* Shouldn't be a record unless there's something to do */ - Assert(xlrec->nunused > 0); - - /* Update all now-unused line pointers */ - offnum = nowunused; - for (int i = 0; i < xlrec->nunused; i++) - { - OffsetNumber off = *offnum++; - ItemId lp = PageGetItemId(page, off); - - Assert(ItemIdIsDead(lp) && !ItemIdHasStorage(lp)); - ItemIdSetUnused(lp); - } - - /* Attempt to truncate line pointer array now */ - PageTruncateLinePointerArray(page); - - PageSetLSN(page, lsn); - MarkBufferDirty(buffer); - } - - if (BufferIsValid(buffer)) - { - Size freespace = PageGetHeapFreeSpace(BufferGetPage(buffer)); - RelFileLocator rlocator; - - XLogRecGetBlockTag(record, 0, &rlocator, NULL, &blkno); - - UnlockReleaseBuffer(buffer); - - /* - * After vacuuming LP_DEAD items from a page, it's useful to update - * the FSM about it, as it may cause the page become target for - * insertions later even if vacuum decides not to visit it (which is - * possible if gets marked all-visible.) - * - * Do this regardless of a full-page image being applied, since the - * FSM data is not in the page anyway. - */ - XLogRecordPageWithFreeSpace(rlocator, blkno, freespace); - } -} - /* * Replay XLOG_HEAP2_VISIBLE record. * @@ -9943,9 +9882,6 @@ heap2_redo(XLogReaderState *record) case XLOG_HEAP2_PRUNE: heap_xlog_prune(record); break; - case XLOG_HEAP2_VACUUM: - heap_xlog_vacuum(record); - break; case XLOG_HEAP2_VISIBLE: heap_xlog_visible(record); break; diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c index 19b50931b90..135fe2dba3e 100644 --- a/src/backend/access/heap/pruneheap.c +++ b/src/backend/access/heap/pruneheap.c @@ -601,7 +601,7 @@ heap_page_prune_and_freeze(Relation relation, Buffer buffer, */ if (do_prune) { - heap_page_prune_execute(buffer, + heap_page_prune_execute(buffer, false, prstate.redirected, prstate.nredirected, prstate.nowdead, prstate.ndead, prstate.nowunused, prstate.nunused); @@ -668,12 +668,16 @@ log_heap_prune_and_freeze(Relation relation, Buffer buffer, OffsetNumber offsets[MaxHeapTuplesPerPage]; bool do_freeze = presult->nfrozen > 0; + xlrec.flags = 0; + xlrec.isCatalogRel = RelationIsAccessibleInLogicalDecoding(relation); xlrec.nredirected = prstate->nredirected; xlrec.ndead = prstate->ndead; xlrec.nunused = prstate->nunused; xlrec.nplans = 0; + xlrec.flags |= XLHP_HAS_CONFLICT_HORIZON; + /* * The snapshotConflictHorizon for the whole record should be the most * conservative of all the horizons calculated for any of the possible @@ -1149,7 +1153,7 @@ heap_prune_record_unused(PruneState *prstate, OffsetNumber offnum) * cleanup lock on the buffer. */ void -heap_page_prune_execute(Buffer buffer, +heap_page_prune_execute(Buffer buffer, bool lp_truncate_only, OffsetNumber *redirected, int nredirected, OffsetNumber *nowdead, int ndead, OffsetNumber *nowunused, int nunused) @@ -1171,6 +1175,7 @@ heap_page_prune_execute(Buffer buffer, ItemId tolp PG_USED_FOR_ASSERTS_ONLY; #ifdef USE_ASSERT_CHECKING + Assert(!lp_truncate_only); /* * Any existing item that we set as an LP_REDIRECT (any 'from' item) @@ -1226,6 +1231,7 @@ heap_page_prune_execute(Buffer buffer, ItemId lp = PageGetItemId(page, off); #ifdef USE_ASSERT_CHECKING + Assert(!lp_truncate_only); /* * An LP_DEAD line pointer must be left behind when the original item @@ -1259,23 +1265,29 @@ heap_page_prune_execute(Buffer buffer, #ifdef USE_ASSERT_CHECKING - /* - * When heap_page_prune_and_freeze() was called, mark_unused_now may - * have been passed as true, which allows would-be LP_DEAD items to be - * made LP_UNUSED instead. This is only possible if the relation has - * no indexes. If there are any dead items, then mark_unused_now was - * not true and every item being marked LP_UNUSED must refer to a - * heap-only tuple. - */ - if (ndead > 0) + if (lp_truncate_only) { - Assert(ItemIdHasStorage(lp) && ItemIdIsNormal(lp)); - htup = (HeapTupleHeader) PageGetItem(page, lp); - Assert(HeapTupleHeaderIsHeapOnly(htup)); + /* Setting LP_DEAD to LP_UNUSED in vacuum's second pass */ + Assert(ItemIdIsDead(lp) && !ItemIdHasStorage(lp)); } else { - Assert(ItemIdIsUsed(lp)); + /* + * When heap_page_prune_and_freeze() was called, mark_unused_now + * may have been passed as true, which allows would-be LP_DEAD + * items to be made LP_UNUSED instead. This is only possible if + * the relation has no indexes. If there are any dead items, then + * mark_unused_now was not true and every item being marked + * LP_UNUSED must refer to a heap-only tuple. + */ + if (ndead > 0) + { + Assert(ItemIdHasStorage(lp) && ItemIdIsNormal(lp)); + htup = (HeapTupleHeader) PageGetItem(page, lp); + Assert(HeapTupleHeaderIsHeapOnly(htup)); + } + else + Assert(ItemIdIsUsed(lp)); } #endif @@ -1283,17 +1295,22 @@ heap_page_prune_execute(Buffer buffer, ItemIdSetUnused(lp); } - /* - * Finally, repair any fragmentation, and update the page's hint bit about - * whether it has free pointers. - */ - PageRepairFragmentation(page); + if (lp_truncate_only) + PageTruncateLinePointerArray(page); + else + { + /* + * Finally, repair any fragmentation, and update the page's hint bit + * about whether it has free pointers. + */ + PageRepairFragmentation(page); - /* - * Now that the page has been modified, assert that redirect items still - * point to valid targets. - */ - page_verify_redirects(page); + /* + * Now that the page has been modified, assert that redirect items + * still point to valid targets. + */ + page_verify_redirects(page); + } } diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index c4553a4159c..9dfb56475cf 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -2394,18 +2394,24 @@ lazy_vacuum_heap_page(LVRelState *vacrel, BlockNumber blkno, Buffer buffer, /* XLOG stuff */ if (RelationNeedsWAL(vacrel->rel)) { - xl_heap_vacuum xlrec; + xl_heap_prune xlrec; XLogRecPtr recptr; + xlrec.flags = XLHP_LP_TRUNCATE_ONLY; + xlrec.snapshotConflictHorizon = InvalidTransactionId; + xlrec.nplans = 0; + xlrec.nredirected = 0; + xlrec.ndead = 0; xlrec.nunused = nunused; + xlrec.isCatalogRel = RelationIsAccessibleInLogicalDecoding(vacrel->rel); XLogBeginInsert(); - XLogRegisterData((char *) &xlrec, SizeOfHeapVacuum); + XLogRegisterData((char *) &xlrec, SizeOfHeapPrune); XLogRegisterBuffer(0, buffer, REGBUF_STANDARD); XLogRegisterBufData(0, (char *) unused, nunused * sizeof(OffsetNumber)); - recptr = XLogInsert(RM_HEAP2_ID, XLOG_HEAP2_VACUUM); + recptr = XLogInsert(RM_HEAP2_ID, XLOG_HEAP2_PRUNE); PageSetLSN(page, recptr); } diff --git a/src/backend/access/rmgrdesc/heapdesc.c b/src/backend/access/rmgrdesc/heapdesc.c index 9f0a0341d40..ea03f902fc4 100644 --- a/src/backend/access/rmgrdesc/heapdesc.c +++ b/src/backend/access/rmgrdesc/heapdesc.c @@ -242,23 +242,6 @@ heap2_desc(StringInfo buf, XLogReaderState *record) } } } - else if (info == XLOG_HEAP2_VACUUM) - { - xl_heap_vacuum *xlrec = (xl_heap_vacuum *) rec; - - appendStringInfo(buf, "nunused: %u", xlrec->nunused); - - if (XLogRecHasBlockData(record, 0)) - { - OffsetNumber *nowunused; - - nowunused = (OffsetNumber *) XLogRecGetBlockData(record, 0, NULL); - - appendStringInfoString(buf, ", unused:"); - array_desc(buf, nowunused, sizeof(OffsetNumber), xlrec->nunused, - &offset_elem_desc, NULL); - } - } else if (info == XLOG_HEAP2_VISIBLE) { xl_heap_visible *xlrec = (xl_heap_visible *) rec; @@ -360,9 +343,6 @@ heap2_identify(uint8 info) case XLOG_HEAP2_PRUNE: id = "PRUNE"; break; - case XLOG_HEAP2_VACUUM: - id = "VACUUM"; - break; case XLOG_HEAP2_VISIBLE: id = "VISIBLE"; break; diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index f77051572fd..38d1bdd825e 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -446,7 +446,6 @@ heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) * interested in. */ case XLOG_HEAP2_PRUNE: - case XLOG_HEAP2_VACUUM: case XLOG_HEAP2_VISIBLE: case XLOG_HEAP2_LOCK_UPDATED: break; diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index 321a46185e1..d5cb8f99cac 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -349,7 +349,7 @@ extern void heap_page_prune_and_freeze(Relation relation, Buffer buffer, HeapPageFreeze *pagefrz, PruneFreezeResult *presult, OffsetNumber *off_loc); -extern void heap_page_prune_execute(Buffer buffer, +extern void heap_page_prune_execute(Buffer buffer, bool lp_truncate_only, OffsetNumber *redirected, int nredirected, OffsetNumber *nowdead, int ndead, OffsetNumber *nowunused, int nunused); diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h index fe4a8ff0620..2393540cf68 100644 --- a/src/include/access/heapam_xlog.h +++ b/src/include/access/heapam_xlog.h @@ -52,11 +52,10 @@ */ #define XLOG_HEAP2_REWRITE 0x00 #define XLOG_HEAP2_PRUNE 0x10 -#define XLOG_HEAP2_VACUUM 0x20 -#define XLOG_HEAP2_VISIBLE 0x30 -#define XLOG_HEAP2_MULTI_INSERT 0x40 -#define XLOG_HEAP2_LOCK_UPDATED 0x50 -#define XLOG_HEAP2_NEW_CID 0x60 +#define XLOG_HEAP2_VISIBLE 0x20 +#define XLOG_HEAP2_MULTI_INSERT 0x30 +#define XLOG_HEAP2_LOCK_UPDATED 0x40 +#define XLOG_HEAP2_NEW_CID 0x50 /* * xl_heap_insert/xl_heap_multi_insert flag values, 8 bits are available. @@ -266,6 +265,7 @@ typedef struct xl_heap_freeze_plan */ typedef struct xl_heap_prune { + uint8 flags; TransactionId snapshotConflictHorizon; uint16 nplans; uint16 nredirected; @@ -288,19 +288,22 @@ typedef struct xl_heap_prune #define SizeOfHeapPrune (offsetof(xl_heap_prune, isCatalogRel) + sizeof(bool)) +/* Flags for xl_heap_prune */ + /* - * The vacuum page record is similar to the prune record, but can only mark - * already LP_DEAD items LP_UNUSED (during VACUUM's second heap pass) - * - * Acquires an ordinary exclusive lock only. + * During vacuum's second pass which sets LP_DEAD items LP_UNUSED, we will only + * truncate the line pointer array, not call PageRepairFragmentation. We need + * this flag to differentiate what kind of lock (exclusive or cleanup) to take + * on the buffer and whether to call PageTruncateLinePointerArray() or + * PageRepairFragementation(). */ -typedef struct xl_heap_vacuum -{ - uint16 nunused; - /* OFFSET NUMBERS are in the block reference 0 */ -} xl_heap_vacuum; +#define XLHP_LP_TRUNCATE_ONLY (1 << 1) -#define SizeOfHeapVacuum (offsetof(xl_heap_vacuum, nunused) + sizeof(uint16)) +/* + * Vacuum's first pass and on-access pruning may need to include a snapshot + * conflict horizon. + */ +#define XLHP_HAS_CONFLICT_HORIZON (1 << 2) /* flags for infobits_set */ #define XLHL_XMAX_IS_MULTI 0x01 -- 2.40.1 --tez7m2a73jtztiij Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v4-0015-Set-hastup-in-heap_page_prune.patch" ^ permalink raw reply [nested|flat] 21+ messages in thread
end of thread, other threads:[~2024-03-19 22:50 UTC | newest] Thread overview: 21+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2022-02-28 09:53 Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit Etsuro Fujita <[email protected]> 2022-03-05 10:32 ` Etsuro Fujita <[email protected]> 2022-03-12 01:02 ` David Zhang <[email protected]> 2022-03-13 09:28 ` Etsuro Fujita <[email protected]> 2022-03-24 04:34 ` Etsuro Fujita <[email protected]> 2022-03-25 06:46 ` Etsuro Fujita <[email protected]> 2022-04-19 19:54 ` David Zhang <[email protected]> 2022-05-02 08:25 ` Etsuro Fujita <[email protected]> 2022-05-04 21:38 ` David Zhang <[email protected]> 2022-05-06 10:08 ` Etsuro Fujita <[email protected]> 2022-05-11 10:39 ` Etsuro Fujita <[email protected]> 2022-05-12 08:46 ` Etsuro Fujita <[email protected]> 2022-06-30 18:50 ` Jacob Champion <[email protected]> 2022-07-07 08:22 ` Etsuro Fujita <[email protected]> 2022-09-30 20:54 ` David Zhang <[email protected]> 2022-11-01 10:24 ` Etsuro Fujita <[email protected]> 2023-01-04 12:19 ` vignesh C <[email protected]> 2023-01-18 11:06 ` Etsuro Fujita <[email protected]> 2023-04-04 10:28 ` Etsuro Fujita <[email protected]> 2023-04-06 08:41 ` Etsuro Fujita <[email protected]> 2024-03-19 22:50 [PATCH v4 14/19] Vacuum second pass emits XLOG_HEAP2_PRUNE record Melanie Plageman <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox