public inbox for [email protected]
help / color / mirror / Atom feedFrom: Etsuro Fujita <[email protected]>
To: Fujii Masao <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit
Date: Mon, 7 Feb 2022 14:35:04 +0900
Message-ID: <CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com> (raw)
In-Reply-To: <[email protected]>
References: <CAPmGK17dAZCXvwnfpr1eTfknTGdt=hYTV9405Gt5SqPOX8K84w@mail.gmail.com>
<[email protected]>
<CAPmGK17XczZmzU=0mhk2_oG9Omo1p-uZCNdZAh_3vuC6+YboiA@mail.gmail.com>
<[email protected]>
<CAPmGK14rr1R_t4xxWyJVng04nDjGnxi=J5fz-TBNBH9Xn0OFYQ@mail.gmail.com>
<[email protected]>
<CAPmGK15FuPVGx3TGHKShsbPKKtF1y58-ZLcKoxfN-nqLj1dZ=g@mail.gmail.com>
<[email protected]>
On Thu, Jan 13, 2022 at 11:54 AM Fujii Masao
<[email protected]> wrote:
> At first I'm reading the 0001 patch. Here are the comments for the patch.
Thanks for reviewing!
> 0001 patch failed to be applied. Could you rebase the patch?
Done. Attached is an updated version of the patch set.
> + entry->changing_xact_state = true;
> + do_sql_command_begin(entry->conn, "DEALLOCATE ALL");
> + pending_deallocs = lappend(pending_deallocs, entry);
>
> Originally entry->changing_xact_state is not set to true when executing DEALLOCATE ALL. But the patch do that. Why do we need this change?
>
> The source comment explains that we intentionally ignore errors in the DEALLOCATE. But the patch changes DEALLOCATE ALL so that it's executed via do_sql_command_begin() that can cause an error. Is this OK?
>
> + if (ignore_errors)
> + pgfdw_report_error(WARNING, res, conn, true, sql);
>
> When DEALLOCATE fails, originally even warning message is not logged. But the patch changes DEALLOCATE so that its result is received via do_sql_command_end() that can log warning message even when ignore_errors argument is enabled. Why do we need to change the behavior?
Yeah, we don’t need to change the behavior as discussed before, so I
fixed these. I worked on the patch after a while, so I forgot about
that. :-(
> + <para>
> + This option controls whether <filename>postgres_fdw</filename> commits
> + multiple remote (sub)transactions opened in a local (sub)transaction
> + in parallel when the local (sub)transaction commits.
>
> Since parallel_commit is an option for foreign server, how the server with this option enabled is handled by postgres_fdw should be documented, instead?
Agreed. I rewrote this slightly like the attached. Does that make sense?
> + <para>
> + Note that if many remote (sub)transactions are opened on a remote server
> + in a local (sub)transaction, this option might increase the remote
> + server’s load significantly when those remote (sub)transactions are
> + committed. So be careful when using this option.
> + </para>
>
> This paragraph should be inside the listitem for parallel_commit, shouldn't it?
I put this note outside, because it’s rewritten to a note about both
the parallel_commit and parallel_abort options in the following patch.
But it would be good to make the parallel-commit patch independent, so
I moved it into the list.
> async_capable=true also may cause the similar issue? If so, this kind of note should be documented also in async_capable?
That’s right. I think it would be good to add a similar note about
that, but I’d like to leave that for another patch.
> This explains that the remote server's load will be increased *significantly*. But "significantly" part is really true?
I think that that would depend on how many transactions are committed
on the remote side at the same time. But the word “significantly”
might be too strong, so I dropped the word.
> I'd like to know how much parallel_commit=true actually can increase the load in a remote server.
Ok, I’ll do a load test.
About the #0002 patch:
This is in preparation for the parallel-abort patch (#0003), but I’d
like to propose a minor cleanup for commit 85c696112: 1) build an
abort command to be sent to the remote in pgfdw_abort_cleanup(), using
a macro, only when/if necessary, as before, and 2) add/modify comments
a little bit.
Sorry for the delay again.
Best regards,
Etsuro Fujita
Attachments:
[application/octet-stream] v3-0001-postgres-fdw-Add-support-for-parallel-commit.patch (18.1K, ../CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com/2-v3-0001-postgres-fdw-Add-support-for-parallel-commit.patch)
download | inline diff:
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index 29fcb6a76e..d59f91f14c 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -58,6 +58,7 @@ typedef struct ConnCacheEntry
bool have_prep_stmt; /* have we prepared any stmts in this xact? */
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 invalidated; /* true if reconnect is pending */
bool keep_connections; /* setting value of keep_connections
* server option */
@@ -92,6 +93,8 @@ static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user);
static void disconnect_pg_server(ConnCacheEntry *entry);
static void check_conn_params(const char **keywords, const char **values, UserMapping *user);
static void configure_remote_session(PGconn *conn);
+static void do_sql_command_begin(PGconn *conn, const char *sql);
+static void do_sql_command_end(PGconn *conn, const char *sql);
static void begin_remote_xact(ConnCacheEntry *entry);
static void pgfdw_xact_callback(XactEvent event, void *arg);
static void pgfdw_subxact_callback(SubXactEvent event,
@@ -100,6 +103,7 @@ static void pgfdw_subxact_callback(SubXactEvent event,
void *arg);
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_exec_cleanup_query(PGconn *conn, const char *query,
bool ignore_errors);
@@ -107,6 +111,8 @@ 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_finish_pre_commit_cleanup(List *pending_entries);
+static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries);
static bool UserMappingPasswordRequired(UserMapping *user);
static bool disconnect_cached_connections(Oid serverid);
@@ -316,14 +322,20 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user)
* is changed will be closed and re-made later.
*
* By default, all the connections to any foreign servers are kept open.
+ *
+ * Also determine to commit (sub)transactions opened on the remote server
+ * in parallel at (sub)transaction end.
*/
entry->keep_connections = true;
+ entry->parallel_commit = false;
foreach(lc, server->options)
{
DefElem *def = (DefElem *) lfirst(lc);
if (strcmp(def->defname, "keep_connections") == 0)
entry->keep_connections = defGetBoolean(def);
+ if (strcmp(def->defname, "parallel_commit") == 0)
+ entry->parallel_commit = defGetBoolean(def);
}
/* Now try to make the connection */
@@ -633,6 +645,32 @@ do_sql_command(PGconn *conn, const char *sql)
PQclear(res);
}
+static void
+do_sql_command_begin(PGconn *conn, const char *sql)
+{
+ if (!PQsendQuery(conn, sql))
+ pgfdw_report_error(ERROR, NULL, conn, false, sql);
+}
+
+static void
+do_sql_command_end(PGconn *conn, const char *sql)
+{
+ PGresult *res;
+
+ /*
+ * Consume whatever data is available from the socket (Note that if all
+ * data is available, this allows us to call PQgetResult without forcing
+ * the overhead of WaitLatchOrSocket in pgfdw_get_result, which would be
+ * very large compared to the overhead of PQconsumeInput.)
+ */
+ if (!PQconsumeInput(conn))
+ pgfdw_report_error(ERROR, NULL, conn, false, sql);
+ res = pgfdw_get_result(conn, sql);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ pgfdw_report_error(ERROR, res, conn, true, sql);
+ PQclear(res);
+}
+
/*
* Start remote transaction or subtransaction, if needed.
*
@@ -888,6 +926,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
{
HASH_SEQ_STATUS scan;
ConnCacheEntry *entry;
+ List *pending_entries = NIL;
/* Quick exit if no connections were touched in this transaction. */
if (!xact_got_connection)
@@ -925,6 +964,12 @@ pgfdw_xact_callback(XactEvent event, void *arg)
/* Commit all remote transactions during pre-commit */
entry->changing_xact_state = true;
+ if (entry->parallel_commit)
+ {
+ do_sql_command_begin(entry->conn, "COMMIT TRANSACTION");
+ pending_entries = lappend(pending_entries, entry);
+ continue;
+ }
do_sql_command(entry->conn, "COMMIT TRANSACTION");
entry->changing_xact_state = false;
@@ -981,23 +1026,15 @@ pgfdw_xact_callback(XactEvent event, void *arg)
}
/* Reset state to show we're out of a transaction */
- entry->xact_depth = 0;
+ pgfdw_reset_xact_state(entry, true);
+ }
- /*
- * If the connection isn't in a good idle state, it is marked as
- * invalid or keep_connections option of its server is disabled, then
- * discard it to recover. Next GetConnection will open a new
- * connection.
- */
- if (PQstatus(entry->conn) != CONNECTION_OK ||
- PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
- entry->changing_xact_state ||
- entry->invalidated ||
- !entry->keep_connections)
- {
- elog(DEBUG3, "discarding connection %p", entry->conn);
- disconnect_pg_server(entry);
- }
+ /* If there are any pending remote transactions, finish closing them */
+ if (pending_entries)
+ {
+ Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT ||
+ event == XACT_EVENT_PRE_COMMIT);
+ pgfdw_finish_pre_commit_cleanup(pending_entries);
}
/*
@@ -1021,6 +1058,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
HASH_SEQ_STATUS scan;
ConnCacheEntry *entry;
int curlevel;
+ List *pending_entries = NIL;
/* Nothing to do at subxact start, nor after commit. */
if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
@@ -1063,6 +1101,12 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
/* Commit all remote subtransactions during pre-commit */
snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
entry->changing_xact_state = true;
+ if (entry->parallel_commit)
+ {
+ do_sql_command_begin(entry->conn, sql);
+ pending_entries = lappend(pending_entries, entry);
+ continue;
+ }
do_sql_command(entry->conn, sql);
entry->changing_xact_state = false;
}
@@ -1076,7 +1120,14 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
}
/* OK, we're outta that level of subtransaction */
- entry->xact_depth--;
+ pgfdw_reset_xact_state(entry, false);
+ }
+
+ /* If there are any pending remote subtransactions, finish closing them */
+ if (pending_entries)
+ {
+ Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB);
+ pgfdw_finish_pre_subcommit_cleanup(pending_entries);
}
}
@@ -1169,6 +1220,40 @@ pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
server->servername)));
}
+/*
+ * Reset state to show we're out of a (sub)transaction.
+ */
+static void
+pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel)
+{
+ if (toplevel)
+ {
+ /* Reset state to show we're out of a transaction */
+ entry->xact_depth = 0;
+
+ /*
+ * If the connection isn't in a good idle state, it is marked as
+ * invalid or keep_connections option of its server is disabled, then
+ * discard it to recover. Next GetConnection will open a new
+ * connection.
+ */
+ if (PQstatus(entry->conn) != CONNECTION_OK ||
+ PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
+ entry->changing_xact_state ||
+ entry->invalidated ||
+ !entry->keep_connections)
+ {
+ elog(DEBUG3, "discarding connection %p", entry->conn);
+ disconnect_pg_server(entry);
+ }
+ }
+ else
+ {
+ /* Reset state to show we're out of a subtransaction */
+ entry->xact_depth--;
+ }
+}
+
/*
* Cancel the currently-in-progress query (whose query text we do not have)
* and ignore the result. Returns true if we successfully cancel the query
@@ -1456,6 +1541,91 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel)
entry->changing_xact_state = false;
}
+static void
+pgfdw_finish_pre_commit_cleanup(List *pending_entries)
+{
+ ConnCacheEntry *entry;
+ List *pending_deallocs = NIL;
+ ListCell *lc;
+
+ Assert(pending_entries);
+
+ foreach(lc, pending_entries)
+ {
+ entry = (ConnCacheEntry *) lfirst(lc);
+
+ Assert(entry->changing_xact_state);
+ do_sql_command_end(entry->conn, "COMMIT TRANSACTION");
+ entry->changing_xact_state = false;
+
+ /* Do a DEALLOCATE ALL if needed */
+ if (entry->have_prep_stmt && entry->have_error)
+ {
+ /* Ignore errors in the DEALLOCATE (see note above) */
+ if (PQsendQuery(entry->conn, "DEALLOCATE ALL"))
+ {
+ pending_deallocs = lappend(pending_deallocs, entry);
+ continue;
+ }
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+
+ pgfdw_reset_xact_state(entry, true);
+ }
+
+ if (pending_deallocs)
+ {
+ foreach(lc, pending_deallocs)
+ {
+ entry = (ConnCacheEntry *) lfirst(lc);
+ PGresult *res;
+
+ /* Ignore errors in the DEALLOCATE (see note above) */
+ if ((res = PQgetResult(entry->conn)) != NULL)
+ {
+ PQclear(res);
+ /*
+ * Stop if the connection is lost (else we'll loop infinitely)
+ */
+ if (PQstatus(entry->conn) == CONNECTION_BAD)
+ break;
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+
+ pgfdw_reset_xact_state(entry, true);
+ }
+ }
+}
+
+static void
+pgfdw_finish_pre_subcommit_cleanup(List *pending_entries)
+{
+ ConnCacheEntry *entry;
+ int curlevel;
+ char sql[100];
+ ListCell *lc;
+
+ Assert(pending_entries);
+
+ curlevel = GetCurrentTransactionNestLevel();
+ snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
+
+ foreach(lc, pending_entries)
+ {
+ entry = (ConnCacheEntry *) lfirst(lc);
+
+ Assert(entry->changing_xact_state);
+ do_sql_command_end(entry->conn, sql);
+ entry->changing_xact_state = false;
+
+ pgfdw_reset_xact_state(entry, false);
+ }
+}
+
/*
* List active foreign server connections.
*
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index b2e02caefe..8043b207c5 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, 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, 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
@@ -10913,3 +10913,79 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity
--Clean up
RESET postgres_fdw.application_name;
RESET debug_discard_caches;
+-- ===================================================================
+-- test parallel commit
+-- ===================================================================
+ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+CREATE TABLE ploc1 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
+ SERVER loopback OPTIONS (table_name 'ploc1');
+CREATE TABLE ploc2 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem2 (f1 int, f2 text)
+ SERVER loopback2 OPTIONS (table_name 'ploc2');
+BEGIN;
+INSERT INTO prem1 VALUES (101, 'foo');
+INSERT INTO prem2 VALUES (201, 'bar');
+COMMIT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+-----
+ 101 | foo
+(1 row)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+-----
+ 201 | bar
+(1 row)
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (102, 'foofoo');
+INSERT INTO prem2 VALUES (202, 'barbar');
+RELEASE SAVEPOINT s;
+COMMIT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+--------
+ 101 | foo
+ 102 | foofoo
+(2 rows)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+--------
+ 201 | bar
+ 202 | barbar
+(2 rows)
+
+-- This tests executing DEALLOCATE ALL against foreign servers in parallel
+-- during pre-commit
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (103, 'baz');
+INSERT INTO prem2 VALUES (203, 'qux');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (104, 'bazbaz');
+INSERT INTO prem2 VALUES (204, 'quxqux');
+COMMIT;
+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 loopback2 OPTIONS (DROP parallel_commit);
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index fc3ce6a53a..a09c0b6db7 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -121,6 +121,7 @@ postgres_fdw_validator(PG_FUNCTION_ARGS)
strcmp(def->defname, "updatable") == 0 ||
strcmp(def->defname, "truncatable") == 0 ||
strcmp(def->defname, "async_capable") == 0 ||
+ strcmp(def->defname, "parallel_commit") == 0 ||
strcmp(def->defname, "keep_connections") == 0)
{
/* these accept only boolean values */
@@ -249,6 +250,7 @@ InitPgFdwOptions(void)
/* async_capable is available on both server and table */
{"async_capable", ForeignServerRelationId, false},
{"async_capable", ForeignTableRelationId, false},
+ {"parallel_commit", 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 e050639b57..2dc6386b40 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -3504,3 +3504,49 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity
--Clean up
RESET postgres_fdw.application_name;
RESET debug_discard_caches;
+
+-- ===================================================================
+-- test parallel commit
+-- ===================================================================
+ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+
+CREATE TABLE ploc1 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
+ SERVER loopback OPTIONS (table_name 'ploc1');
+CREATE TABLE ploc2 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem2 (f1 int, f2 text)
+ SERVER loopback2 OPTIONS (table_name 'ploc2');
+
+BEGIN;
+INSERT INTO prem1 VALUES (101, 'foo');
+INSERT INTO prem2 VALUES (201, 'bar');
+COMMIT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (102, 'foofoo');
+INSERT INTO prem2 VALUES (202, 'barbar');
+RELEASE SAVEPOINT s;
+COMMIT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+-- This tests executing DEALLOCATE ALL against foreign servers in parallel
+-- during pre-commit
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (103, 'baz');
+INSERT INTO prem2 VALUES (203, 'qux');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (104, 'bazbaz');
+INSERT INTO prem2 VALUES (204, 'quxqux');
+COMMIT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+ALTER SERVER loopback OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback2 OPTIONS (DROP parallel_commit);
diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml
index 2bb31f1125..7886436aae 100644
--- a/doc/src/sgml/postgres-fdw.sgml
+++ b/doc/src/sgml/postgres-fdw.sgml
@@ -455,6 +455,50 @@ OPTIONS (ADD password_required 'false');
</variablelist>
</sect3>
+ <sect3>
+ <title>Transaction Management Options</title>
+
+ <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:
+ </para>
+
+ <variablelist>
+
+ <varlistentry>
+ <term><literal>parallel_commit</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ This option controls whether <filename>postgres_fdw</filename> commits
+ remote (sub)transactions opened on a foreign server in a local
+ (sub)transaction in parallel when the local (sub)transaction commits.
+ This option can only be specified for foreign servers, not per-table.
+ The default is <literal>false</literal>.
+ </para>
+
+ <para>
+ If multiple foreign servers with this option enabled are involved in
+ a local (sub)transaction, remote (sub)transactions opened on those
+ foreign servers in the local (sub)transaction are committed in parallel
+ across those foreign servers.
+ </para>
+
+ <para>
+ For a foreign server with this option enabled, if many remote
+ (sub)transactions are opened on it in a local (sub)transaction, this
+ option might increase the remote server’s load when the local
+ (sub)transactions commits, so be careful when using this option.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ </variablelist>
+
+ </sect3>
+
<sect3>
<title>Updatability Options</title>
[application/octet-stream] v3-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch (3.2K, ../CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com/3-v3-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 d59f91f14c..22e72c99c7 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
*/
@@ -109,8 +121,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);
static bool UserMappingPasswordRequired(UserMapping *user);
@@ -1019,8 +1030,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;
}
}
@@ -1113,10 +1124,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 */
@@ -1469,10 +1477,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.
@@ -1480,8 +1485,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.
@@ -1513,8 +1520,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] v3-0003-postgres-fdw-Add-support-for-parallel-abort.patch (23.0K, ../CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com/4-v3-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 22e72c99c7..9b041361f3 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 */
@@ -117,13 +118,25 @@ static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue);
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry);
static void pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel);
static bool pgfdw_cancel_query(PGconn *conn);
+static bool pgfdw_cancel_query_begin(PGconn *conn);
+static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime);
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query,
bool ignore_errors);
+static bool pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query);
+static bool pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query,
+ bool ignore_errors,
+ TimestampTz endtime);
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
PGresult **result, bool *timed_out);
static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel);
+static bool pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel,
+ List **pending_entries,
+ List **cancel_requested);
static void pgfdw_finish_pre_commit_cleanup(List *pending_entries);
static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries);
+static void pgfdw_finish_abort_cleanup(List *pending_entries,
+ List *cancel_requested,
+ bool toplevel);
static bool UserMappingPasswordRequired(UserMapping *user);
static bool disconnect_cached_connections(Oid serverid);
@@ -334,11 +347,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user)
*
* By default, all the connections to any foreign servers are kept open.
*
- * Also determine to commit (sub)transactions opened on the remote server
- * in parallel at (sub)transaction end.
+ * Also determine to commit/abort (sub)transactions opened on the remote
+ * server in parallel at (sub)transaction end.
*/
entry->keep_connections = true;
entry->parallel_commit = false;
+ entry->parallel_abort = false;
foreach(lc, server->options)
{
DefElem *def = (DefElem *) lfirst(lc);
@@ -347,6 +361,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user)
entry->keep_connections = defGetBoolean(def);
if (strcmp(def->defname, "parallel_commit") == 0)
entry->parallel_commit = defGetBoolean(def);
+ if (strcmp(def->defname, "parallel_abort") == 0)
+ entry->parallel_abort = defGetBoolean(def);
}
/* Now try to make the connection */
@@ -938,6 +954,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)
@@ -1031,7 +1048,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;
}
}
@@ -1041,11 +1066,21 @@ pgfdw_xact_callback(XactEvent event, void *arg)
}
/* If there are any pending remote transactions, finish closing them */
- 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);
+ }
}
/*
@@ -1070,6 +1105,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 ||
@@ -1124,7 +1160,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 */
@@ -1132,10 +1176,19 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
}
/* If there are any pending remote subtransactions, finish closing them */
- if (pending_entries)
+ if (pending_entries || cancel_requested)
{
- Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB);
- pgfdw_finish_pre_subcommit_cleanup(pending_entries);
+ if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
+ {
+ Assert(cancel_requested == NIL);
+ pgfdw_finish_pre_subcommit_cleanup(pending_entries);
+ }
+ else
+ {
+ Assert(event == SUBXACT_EVENT_ABORT_SUB);
+ pgfdw_finish_abort_cleanup(pending_entries, cancel_requested,
+ false);
+ }
}
}
@@ -1279,11 +1332,7 @@ pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel)
static bool
pgfdw_cancel_query(PGconn *conn)
{
- PGcancel *cancel;
- char errbuf[256];
- PGresult *result = NULL;
TimestampTz endtime;
- bool timed_out;
/*
* If it takes too long to cancel the query and discard the result, assume
@@ -1291,6 +1340,17 @@ pgfdw_cancel_query(PGconn *conn)
*/
endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+ if (!pgfdw_cancel_query_begin(conn))
+ return false;
+ return pgfdw_cancel_query_end(conn, endtime);
+}
+
+static bool
+pgfdw_cancel_query_begin(PGconn *conn)
+{
+ PGcancel *cancel;
+ char errbuf[256];
+
/*
* Issue cancel request. Unfortunately, there's no good way to limit the
* amount of time that we might block inside PQgetCancel().
@@ -1309,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))
{
@@ -1343,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
@@ -1355,6 +1422,14 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
*/
endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+ if (!pgfdw_exec_cleanup_query_begin(conn, query))
+ return false;
+ return pgfdw_exec_cleanup_query_end(conn, query, ignore_errors, endtime);
+}
+
+static bool
+pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query)
+{
/*
* Submit a query. Since we don't use non-blocking mode, this also can
* block. But its risk is relatively small, so we ignore that for now.
@@ -1365,6 +1440,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))
{
@@ -1549,6 +1634,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;
+}
+
static void
pgfdw_finish_pre_commit_cleanup(List *pending_entries)
{
@@ -1634,6 +1769,155 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries)
}
}
+static void
+pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested,
+ bool toplevel)
+{
+ List *pending_deallocs = NIL;
+ ListCell *lc;
+
+ if (cancel_requested)
+ {
+ foreach(lc, cancel_requested)
+ {
+ ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
+ TimestampTz endtime;
+ char sql[100];
+
+ Assert(entry->changing_xact_state);
+
+ /*
+ * Set end time. You might think we should do so before issuing
+ * cancel request like in normal mode, but that is problematic,
+ * because if, for example, it took longer than 30 seconds to
+ * process the first few entries in the cancel_requested list, it
+ * would cause a timeout for each of the remaining entries in the
+ * list when processing it, leading to slamming the connection of
+ * it shut.
+ */
+ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
+ 30000);
+
+ /* Get and discard the result of the query. */
+ if (!pgfdw_cancel_query_end(entry->conn, endtime))
+ {
+ /* Unable to cancel running query */
+ pgfdw_reset_xact_state(entry, toplevel);
+ continue;
+ }
+
+ CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
+ if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql))
+ {
+ /* Unable to abort remote (sub)transaction */
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+ else
+ pending_entries = lappend(pending_entries, entry);
+ }
+ }
+
+ if (!pending_entries)
+ return;
+
+ foreach(lc, pending_entries)
+ {
+ ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
+ TimestampTz endtime;
+ char sql[100];
+
+ Assert(entry->changing_xact_state);
+
+ /*
+ * Set end time. We do this now, not before issuing the command like
+ * in normal mode, for the same reason as for the cancel_requested
+ * entries.
+ */
+ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+
+ /* Get the result of the command. */
+ CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
+ if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, false, endtime))
+ {
+ /* Unable to abort remote (sub)transaction */
+ pgfdw_reset_xact_state(entry, toplevel);
+ continue;
+ }
+
+ /*
+ * If called for cleanup at main-transaction end, do a DEALLOCATE ALL
+ * if needed.
+ */
+ if (toplevel)
+ {
+ if (entry->have_prep_stmt && entry->have_error)
+ {
+ if (!pgfdw_exec_cleanup_query_begin(entry->conn,
+ "DEALLOCATE ALL"))
+ {
+ /* Trouble clearing prepared statements */
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+ else
+ pending_deallocs = lappend(pending_deallocs, entry);
+ continue;
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+ }
+
+ /* Reset the per-connection state if needed */
+ if (entry->state.pendingAreq)
+ memset(&entry->state, 0, sizeof(entry->state));
+
+ /* We're done with this entry; unset the changing_xact_state flag */
+ entry->changing_xact_state = false;
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+
+ if (!pending_deallocs)
+ return;
+ Assert(toplevel);
+
+ foreach(lc, pending_deallocs)
+ {
+ ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
+ TimestampTz endtime;
+
+ Assert(entry->changing_xact_state);
+ Assert(entry->have_prep_stmt);
+ Assert(entry->have_error);
+
+ /*
+ * Set end time. We do this now, not before issuing the command like
+ * in normal mode, for the same reason as for the cancel_requested
+ * entries.
+ */
+ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+
+ /* Get the result of the command. */
+ if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL",
+ true, endtime))
+ {
+ /* Trouble clearing prepared statements */
+ pgfdw_reset_xact_state(entry, toplevel);
+ continue;
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+
+ /* Reset the per-connection state if needed */
+ if (entry->state.pendingAreq)
+ memset(&entry->state, 0, sizeof(entry->state));
+
+ /* We're done with this entry; unset the changing_xact_state flag */
+ entry->changing_xact_state = false;
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+}
+
/*
* List active foreign server connections.
*
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 8043b207c5..2adb0ac125 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -9509,7 +9509,7 @@ DO $d$
END;
$d$;
ERROR: invalid option "password"
-HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, keep_connections
+HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, parallel_abort, keep_connections
CONTEXT: SQL statement "ALTER SERVER loopback_nopw OPTIONS (ADD password 'dummypw')"
PL/pgSQL function inline_code_block line 3 at EXECUTE
-- If we add a password for our user mapping instead, we should get a different
@@ -10914,10 +10914,12 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity
RESET postgres_fdw.application_name;
RESET debug_discard_caches;
-- ===================================================================
--- test parallel commit
+-- test parallel commit and parallel abort
-- ===================================================================
ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true');
ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true');
CREATE TABLE ploc1 (f1 int, f2 text);
CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
SERVER loopback OPTIONS (table_name 'ploc1');
@@ -10987,5 +10989,52 @@ SELECT * FROM prem2;
204 | quxqux
(3 rows)
+BEGIN;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+--------
+ 101 | foo
+ 102 | foofoo
+ 104 | bazbaz
+(3 rows)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+--------
+ 201 | bar
+ 202 | barbar
+ 204 | quxqux
+(3 rows)
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+--------
+ 101 | foo
+ 102 | foofoo
+ 104 | bazbaz
+(3 rows)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+--------
+ 201 | bar
+ 202 | barbar
+ 204 | quxqux
+(3 rows)
+
ALTER SERVER loopback OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback OPTIONS (DROP parallel_abort);
ALTER SERVER loopback2 OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback2 OPTIONS (DROP parallel_abort);
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index a09c0b6db7..e8c8b4ab36 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -122,6 +122,7 @@ postgres_fdw_validator(PG_FUNCTION_ARGS)
strcmp(def->defname, "truncatable") == 0 ||
strcmp(def->defname, "async_capable") == 0 ||
strcmp(def->defname, "parallel_commit") == 0 ||
+ strcmp(def->defname, "parallel_abort") == 0 ||
strcmp(def->defname, "keep_connections") == 0)
{
/* these accept only boolean values */
@@ -251,6 +252,7 @@ InitPgFdwOptions(void)
{"async_capable", ForeignServerRelationId, false},
{"async_capable", ForeignTableRelationId, false},
{"parallel_commit", ForeignServerRelationId, false},
+ {"parallel_abort", ForeignServerRelationId, false},
{"keep_connections", ForeignServerRelationId, false},
{"password_required", UserMappingRelationId, false},
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 2dc6386b40..58ec80fad7 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -3506,10 +3506,12 @@ RESET postgres_fdw.application_name;
RESET debug_discard_caches;
-- ===================================================================
--- test parallel commit
+-- test parallel commit and parallel abort
-- ===================================================================
ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true');
ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true');
CREATE TABLE ploc1 (f1 int, f2 text);
CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
@@ -3548,5 +3550,26 @@ COMMIT;
SELECT * FROM prem1;
SELECT * FROM prem2;
+BEGIN;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
ALTER SERVER loopback OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback OPTIONS (DROP parallel_abort);
ALTER SERVER loopback2 OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback2 OPTIONS (DROP parallel_abort);
diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml
index 7886436aae..19e408ec9a 100644
--- a/doc/src/sgml/postgres-fdw.sgml
+++ b/doc/src/sgml/postgres-fdw.sgml
@@ -460,10 +460,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>
@@ -478,25 +478,38 @@ 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, remote (sub)transactions opened on those
- foreign servers in the local (sub)transaction are committed in parallel
- across those foreign servers.
- </para>
-
- <para>
- For a foreign server with this option enabled, if many remote
- (sub)transactions are opened on it in a local (sub)transaction, this
- option might increase the remote server’s load when the local
- (sub)transactions 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 this option enabled are involved in a
+ local (sub)transaction, remote (sub)transactions opened on those foreign
+ servers in the local (sub)transaction are committed or aborted in parallel
+ across those foreign servers.
+ </para>
+
+ <para>
+ For a foreign server with this option enabled, if many remote
+ (sub)transactions are opened on it in a local (sub)transaction, this
+ option might increase the remote server’s load when the local
+ (sub)transactions commits or aborts, so be careful when using this option.
+ </para>
+
</sect3>
<sect3>
view thread (4+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected]
Subject: Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit
In-Reply-To: <CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox