public inbox for [email protected]help / color / mirror / Atom feed
Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit 9+ messages / 4 participants [nested] [flat]
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-11 12:59 Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Etsuro Fujita @ 2022-02-11 12:59 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Tue, Feb 8, 2022 at 3:49 AM Fujii Masao <[email protected]> wrote: > Here are the review comments for 0001 patch. > > I got the following compiler warning. > > [16:58:07.120] connection.c: In function ‘pgfdw_finish_pre_commit_cleanup’: > [16:58:07.120] connection.c:1726:4: error: ISO C90 forbids mixed declarations and code [-Werror=declaration-after-statement] > [16:58:07.120] 1726 | PGresult *res; > [16:58:07.120] | ^~~~~~~~ Sorry, I didn’t notice this, because my compiler doesn’t produce it. I tried to fix it. Attached is an updated version of the patch set. I hope this works for you. > + /* Ignore errors in the DEALLOCATE (see note above) */ > + if ((res = PQgetResult(entry->conn)) != NULL) > > Doesn't PQgetResult() need to be called repeatedly until it returns NULL or the connection is lost because there can be more than one messages to receive? Yeah, we would receive a single message here, but PQgetResult must be called repeatedly until it returns NULL (see the documentation note about it in libpq.sgml); else the PQtransactionStatus of the connection would remain PQTRANS_ACTIVE, causing the connection to be closed at transaction end, because we do this in pgfdw_reset_xact_state called from pgfdw_xact_callback: /* * 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); } But I noticed a brown-paper-bag bug in the bit you showed above: the if test should be modified as a while loop. :-( I fixed this in the attached. > + if (pending_deallocs) > + { > + foreach(lc, pending_deallocs) > > If pending_deallocs is NIL, we don't enter this foreach loop. So probably "if (pending_deallocs)" seems not necessary. Yeah, I think we could omit the if test, but I added it to match other places (see e.g., foreign_grouping_ok() in postgres_fdw.c). It looks cleaner to me to have it before the loop. > entry->keep_connections = defGetBoolean(def); > + if (strcmp(def->defname, "parallel_commit") == 0) > + entry->parallel_commit = defGetBoolean(def); > > Isn't it better to use "else if" here, instead? Yeah, that would be better. Done. > +static void do_sql_command_begin(PGconn *conn, const char *sql); > +static void do_sql_command_end(PGconn *conn, const char *sql); > > To simplify the code more, I'm tempted to change do_sql_command() so that it just calls the above two functions, instead of calling PQsendQuery() and pgfw_get_result() directly. Thought? If we do this, probably we also need to change do_sql_command_end() so that it accepts boolean flag which specifies whether PQconsumeInput() is called or not, as follows. Done. Actually, I was planning to do this for consistency with a similar refactoring for pgfdw_cancel_query and pgfdw_exec_cleanup_query that had been done in the parallel-abort patch. I tweaked comments/docs a little bit as well. Thanks for reviewing! Best regards, Etsuro Fujita Attachments: [application/octet-stream] v4-0001-postgres-fdw-Add-support-for-parallel-commit.patch (19.0K, ../../CAPmGK17nfEZkKhqoP8fAPKbnfSpf2JVgV5+_Xjv2mn8P-4X0hQ@mail.gmail.com/2-v4-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..3f4e71da9e 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,9 @@ 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, + bool consume_input); 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 +104,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 +112,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 +323,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); + else if (strcmp(def->defname, "parallel_commit") == 0) + entry->parallel_commit = defGetBoolean(def); } /* Now try to make the connection */ @@ -623,10 +636,30 @@ configure_remote_session(PGconn *conn) void do_sql_command(PGconn *conn, const char *sql) { - PGresult *res; + do_sql_command_begin(conn, sql); + do_sql_command_end(conn, sql, false); +} +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, bool consume_input) +{ + PGresult *res; + + /* + * Consume whatever data is available from the socket if requested. 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 (consume_input && !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); @@ -888,6 +921,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 +959,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 +1021,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 connections, finish cleaning them up */ + if (pending_entries) + { + Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT || + event == XACT_EVENT_PRE_COMMIT); + pgfdw_finish_pre_commit_cleanup(pending_entries); } /* @@ -1021,6 +1053,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 +1096,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 +1115,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 connections, finish cleaning them up */ + if (pending_entries) + { + Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB); + pgfdw_finish_pre_subcommit_cleanup(pending_entries); } } @@ -1169,6 +1215,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 +1536,116 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) entry->changing_xact_state = false; } +/* + * Finish pre-commit cleanup of connections on which we have sent a COMMIT + * command. + */ +static void +pgfdw_finish_pre_commit_cleanup(List *pending_entries) +{ + ConnCacheEntry *entry; + List *pending_deallocs = NIL; + ListCell *lc; + + Assert(pending_entries); + + /* + * Get the result of the COMMIT command for each of the pending entries + */ + foreach(lc, pending_entries) + { + entry = (ConnCacheEntry *) lfirst(lc); + + Assert(entry->changing_xact_state); + /* + * We might already have received the result on the socket, so pass + * consume_input=true to try to consume it first + */ + do_sql_command_end(entry->conn, "COMMIT TRANSACTION", true); + entry->changing_xact_state = false; + + /* Do a DEALLOCATE ALL in parallel 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); + } + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries if any + */ + if (pending_deallocs) + { + foreach(lc, pending_deallocs) + { + PGresult *res; + entry = (ConnCacheEntry *) lfirst(lc); + + /* Ignore errors in the DEALLOCATE (see note above) */ + while ((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); + } + } +} + +/* + * Finish pre-subcommit cleanup of connections on which we have sent a RELEASE + * command. + */ +static void +pgfdw_finish_pre_subcommit_cleanup(List *pending_entries) +{ + ConnCacheEntry *entry; + int curlevel; + char sql[100]; + ListCell *lc; + + Assert(pending_entries); + + /* + * Get the result of the RELEASE command for each of the 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); + /* + * We might already have received the result on the socket, so pass + * consume_input=true to try to consume it first + */ + do_sql_command_end(entry->conn, sql, true); + 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..238e2f84a6 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -455,6 +455,52 @@ 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, 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. + </para> + </listitem> + </varlistentry> + + </variablelist> + + </sect3> + <sect3> <title>Updatability Options</title> [application/octet-stream] v4-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch (3.2K, ../../CAPmGK17nfEZkKhqoP8fAPKbnfSpf2JVgV5+_Xjv2mn8P-4X0hQ@mail.gmail.com/3-v4-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 3f4e71da9e..2deb4b2c4a 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); static bool UserMappingPasswordRequired(UserMapping *user); @@ -1014,8 +1025,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; } } @@ -1108,10 +1119,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 */ @@ -1464,10 +1472,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. @@ -1475,8 +1480,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. @@ -1508,8 +1515,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] v4-0003-postgres-fdw-Add-support-for-parallel-abort.patch (23.3K, ../../CAPmGK17nfEZkKhqoP8fAPKbnfSpf2JVgV5+_Xjv2mn8P-4X0hQ@mail.gmail.com/4-v4-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 2deb4b2c4a..bd11e762b7 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -59,6 +59,7 @@ typedef struct ConnCacheEntry bool have_error; /* have any subxacts aborted in this xact? */ bool changing_xact_state; /* xact state change in process */ bool parallel_commit; /* do we commit (sub)xacts in parallel? */ + bool parallel_abort; /* do we abort (sub)xacts in parallel? */ bool invalidated; /* true if reconnect is pending */ bool keep_connections; /* setting value of keep_connections * server option */ @@ -118,13 +119,25 @@ static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue); static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry); static void pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel); static bool pgfdw_cancel_query(PGconn *conn); +static bool pgfdw_cancel_query_begin(PGconn *conn); +static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime); static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors); +static bool pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query); +static bool pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query, + bool ignore_errors, + TimestampTz endtime); static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out); static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); +static bool pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel, + List **pending_entries, + List **cancel_requested); static void pgfdw_finish_pre_commit_cleanup(List *pending_entries); static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries); +static void pgfdw_finish_abort_cleanup(List *pending_entries, + List *cancel_requested, + bool toplevel); static bool UserMappingPasswordRequired(UserMapping *user); static bool disconnect_cached_connections(Oid serverid); @@ -335,11 +348,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) * * By default, all the connections to any foreign servers are kept open. * - * Also determine to commit (sub)transactions opened on the remote server - * in parallel at (sub)transaction end. + * Also determine to commit/abort (sub)transactions opened on the remote + * server in parallel at (sub)transaction end. */ entry->keep_connections = true; entry->parallel_commit = false; + entry->parallel_abort = false; foreach(lc, server->options) { DefElem *def = (DefElem *) lfirst(lc); @@ -348,6 +362,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user) entry->keep_connections = defGetBoolean(def); else if (strcmp(def->defname, "parallel_commit") == 0) entry->parallel_commit = defGetBoolean(def); + else if (strcmp(def->defname, "parallel_abort") == 0) + entry->parallel_abort = defGetBoolean(def); } /* Now try to make the connection */ @@ -933,6 +949,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) HASH_SEQ_STATUS scan; ConnCacheEntry *entry; List *pending_entries = NIL; + List *cancel_requested = NIL; /* Quick exit if no connections were touched in this transaction. */ if (!xact_got_connection) @@ -1026,7 +1043,15 @@ pgfdw_xact_callback(XactEvent event, void *arg) case XACT_EVENT_PARALLEL_ABORT: case XACT_EVENT_ABORT: /* Rollback all remote transactions during abort */ - pgfdw_abort_cleanup(entry, true); + if (entry->parallel_abort) + { + if (pgfdw_abort_cleanup_begin(entry, true, + &pending_entries, + &cancel_requested)) + continue; + } + else + pgfdw_abort_cleanup(entry, true); break; } } @@ -1036,11 +1061,21 @@ pgfdw_xact_callback(XactEvent event, void *arg) } /* If there are any pending connections, finish cleaning them up */ - if (pending_entries) + if (pending_entries || cancel_requested) { - Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT || - event == XACT_EVENT_PRE_COMMIT); - pgfdw_finish_pre_commit_cleanup(pending_entries); + if (event == XACT_EVENT_PARALLEL_PRE_COMMIT || + event == XACT_EVENT_PRE_COMMIT) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_commit_cleanup(pending_entries); + } + else + { + Assert(event == XACT_EVENT_PARALLEL_ABORT || + event == XACT_EVENT_ABORT); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + true); + } } /* @@ -1065,6 +1100,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, ConnCacheEntry *entry; int curlevel; List *pending_entries = NIL; + List *cancel_requested = NIL; /* Nothing to do at subxact start, nor after commit. */ if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB || @@ -1119,7 +1155,15 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, else { /* Rollback all remote subtransactions during abort */ - pgfdw_abort_cleanup(entry, false); + if (entry->parallel_abort) + { + if (pgfdw_abort_cleanup_begin(entry, false, + &pending_entries, + &cancel_requested)) + continue; + } + else + pgfdw_abort_cleanup(entry, false); } /* OK, we're outta that level of subtransaction */ @@ -1127,10 +1171,19 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, } /* If there are any pending connections, finish cleaning them up */ - if (pending_entries) + if (pending_entries || cancel_requested) { - Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB); - pgfdw_finish_pre_subcommit_cleanup(pending_entries); + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) + { + Assert(cancel_requested == NIL); + pgfdw_finish_pre_subcommit_cleanup(pending_entries); + } + else + { + Assert(event == SUBXACT_EVENT_ABORT_SUB); + pgfdw_finish_abort_cleanup(pending_entries, cancel_requested, + false); + } } } @@ -1274,11 +1327,7 @@ pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel) static bool pgfdw_cancel_query(PGconn *conn) { - PGcancel *cancel; - char errbuf[256]; - PGresult *result = NULL; TimestampTz endtime; - bool timed_out; /* * If it takes too long to cancel the query and discard the result, assume @@ -1286,6 +1335,17 @@ pgfdw_cancel_query(PGconn *conn) */ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + if (!pgfdw_cancel_query_begin(conn)) + return false; + return pgfdw_cancel_query_end(conn, endtime); +} + +static bool +pgfdw_cancel_query_begin(PGconn *conn) +{ + PGcancel *cancel; + char errbuf[256]; + /* * Issue cancel request. Unfortunately, there's no good way to limit the * amount of time that we might block inside PQgetCancel(). @@ -1304,6 +1364,15 @@ pgfdw_cancel_query(PGconn *conn) PQfreeCancel(cancel); } + return true; +} + +static bool +pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime) +{ + PGresult *result = NULL; + bool timed_out; + /* Get and discard the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1338,9 +1407,7 @@ pgfdw_cancel_query(PGconn *conn) static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) { - PGresult *result = NULL; TimestampTz endtime; - bool timed_out; /* * If it takes too long to execute a cleanup query, assume the connection @@ -1350,6 +1417,14 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) */ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + if (!pgfdw_exec_cleanup_query_begin(conn, query)) + return false; + return pgfdw_exec_cleanup_query_end(conn, query, ignore_errors, endtime); +} + +static bool +pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query) +{ /* * Submit a query. Since we don't use non-blocking mode, this also can * block. But its risk is relatively small, so we ignore that for now. @@ -1360,6 +1435,16 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) return false; } + return true; +} + +static bool +pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query, + bool ignore_errors, TimestampTz endtime) +{ + PGresult *result = NULL; + bool timed_out; + /* Get the result of the query. */ if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) { @@ -1544,6 +1629,56 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +static bool +pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel, + List **pending_entries, List **cancel_requested) +{ + /* + * Don't try to clean up the connection if we're already in error + * recursion trouble. + */ + if (in_error_recursion_trouble()) + entry->changing_xact_state = true; + + /* + * If connection is already unsalvageable, don't touch it further. + */ + if (entry->changing_xact_state) + return false; + + /* + * Mark this connection as in the process of changing transaction state. + */ + entry->changing_xact_state = true; + + /* Assume we might have lost track of prepared statements */ + entry->have_error = true; + + /* + * If a command has been submitted to the remote server by using an + * asynchronous execution function, the command might not have yet + * completed. Check to see if a command is still being processed by the + * remote server, and if so, request cancellation of the command. + */ + if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE) + { + if (!pgfdw_cancel_query_begin(entry->conn)) + return false; /* Unable to cancel running query */ + *cancel_requested = lappend(*cancel_requested, entry); + } + else + { + char sql[100]; + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql)) + return false; /* Unable to abort remote transaction */ + *pending_entries = lappend(*pending_entries, entry); + } + + return true; +} + /* * Finish pre-commit cleanup of connections on which we have sent a COMMIT * command. @@ -1654,6 +1789,159 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries) } } +/* + * Finish (sub)abort cleanup of connections on which we have sent a (sub)abort + * command or cancel request. + */ +static void +pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested, + bool toplevel) +{ + List *pending_deallocs = NIL; + ListCell *lc; + + if (cancel_requested) + { + foreach(lc, cancel_requested) + { + ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc); + TimestampTz endtime; + char sql[100]; + + Assert(entry->changing_xact_state); + + /* + * Set end time. You might think we should do so before issuing + * cancel request like in normal mode, but that is problematic, + * because if, for example, it took longer than 30 seconds to + * process the first few entries in the cancel_requested list, it + * would cause a timeout for each of the remaining entries in the + * list when processing it, leading to slamming the connection of + * it shut. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + 30000); + + /* Get and discard the result of the query. */ + if (!pgfdw_cancel_query_end(entry->conn, endtime)) + { + /* Unable to cancel running query */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + } + else + pending_entries = lappend(pending_entries, entry); + } + } + + if (!pending_entries) + return; + + foreach(lc, pending_entries) + { + ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc); + TimestampTz endtime; + char sql[100]; + + Assert(entry->changing_xact_state); + + /* + * Set end time. We do this now, not before issuing the command like + * in normal mode, for the same reason as for the cancel_requested + * entries. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + + /* Get the result of the command. */ + CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel); + if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, false, endtime)) + { + /* Unable to abort remote (sub)transaction */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + /* + * If called for cleanup at main-transaction end, do a DEALLOCATE ALL + * if needed. + */ + if (toplevel) + { + if (entry->have_prep_stmt && entry->have_error) + { + if (!pgfdw_exec_cleanup_query_begin(entry->conn, + "DEALLOCATE ALL")) + { + /* Trouble clearing prepared statements */ + pgfdw_reset_xact_state(entry, toplevel); + } + else + pending_deallocs = lappend(pending_deallocs, entry); + continue; + } + + entry->have_prep_stmt = false; + entry->have_error = false; + } + + /* Reset the per-connection state if needed */ + if (entry->state.pendingAreq) + memset(&entry->state, 0, sizeof(entry->state)); + + /* We're done with this entry; unset the changing_xact_state flag */ + entry->changing_xact_state = false; + pgfdw_reset_xact_state(entry, toplevel); + } + + if (!pending_deallocs) + return; + Assert(toplevel); + + foreach(lc, pending_deallocs) + { + ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc); + TimestampTz endtime; + + Assert(entry->changing_xact_state); + Assert(entry->have_prep_stmt); + Assert(entry->have_error); + + /* + * Set end time. We do this now, not before issuing the command like + * in normal mode, for the same reason as for the cancel_requested + * entries. + */ + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000); + + /* Get the result of the command. */ + if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL", + true, endtime)) + { + /* Trouble clearing prepared statements */ + pgfdw_reset_xact_state(entry, toplevel); + continue; + } + + entry->have_prep_stmt = false; + entry->have_error = false; + + /* Reset the per-connection state if needed */ + if (entry->state.pendingAreq) + memset(&entry->state, 0, sizeof(entry->state)); + + /* We're done with this entry; unset the changing_xact_state flag */ + entry->changing_xact_state = false; + pgfdw_reset_xact_state(entry, toplevel); + } +} + /* * List active foreign server connections. * diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 8043b207c5..2adb0ac125 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -9509,7 +9509,7 @@ DO $d$ END; $d$; ERROR: invalid option "password" -HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, keep_connections +HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, parallel_abort, keep_connections CONTEXT: SQL statement "ALTER SERVER loopback_nopw OPTIONS (ADD password 'dummypw')" PL/pgSQL function inline_code_block line 3 at EXECUTE -- If we add a password for our user mapping instead, we should get a different @@ -10914,10 +10914,12 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity RESET postgres_fdw.application_name; RESET debug_discard_caches; -- =================================================================== --- test parallel commit +-- test parallel commit and parallel abort -- =================================================================== ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true'); ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true'); CREATE TABLE ploc1 (f1 int, f2 text); CREATE FOREIGN TABLE prem1 (f1 int, f2 text) SERVER loopback OPTIONS (table_name 'ploc1'); @@ -10987,5 +10989,52 @@ SELECT * FROM prem2; 204 | quxqux (3 rows) +BEGIN; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; + f1 | f2 +-----+-------- + 101 | foo + 102 | foofoo + 104 | bazbaz +(3 rows) + +SELECT * FROM prem2; + f1 | f2 +-----+-------- + 201 | bar + 202 | barbar + 204 | quxqux +(3 rows) + +BEGIN; +SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ROLLBACK TO SAVEPOINT s; +RELEASE SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; + f1 | f2 +-----+-------- + 101 | foo + 102 | foofoo + 104 | bazbaz +(3 rows) + +SELECT * FROM prem2; + f1 | f2 +-----+-------- + 201 | bar + 202 | barbar + 204 | quxqux +(3 rows) + ALTER SERVER loopback OPTIONS (DROP parallel_commit); +ALTER SERVER loopback OPTIONS (DROP parallel_abort); ALTER SERVER loopback2 OPTIONS (DROP parallel_commit); +ALTER SERVER loopback2 OPTIONS (DROP parallel_abort); diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c index a09c0b6db7..e8c8b4ab36 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -122,6 +122,7 @@ postgres_fdw_validator(PG_FUNCTION_ARGS) strcmp(def->defname, "truncatable") == 0 || strcmp(def->defname, "async_capable") == 0 || strcmp(def->defname, "parallel_commit") == 0 || + strcmp(def->defname, "parallel_abort") == 0 || strcmp(def->defname, "keep_connections") == 0) { /* these accept only boolean values */ @@ -251,6 +252,7 @@ InitPgFdwOptions(void) {"async_capable", ForeignServerRelationId, false}, {"async_capable", ForeignTableRelationId, false}, {"parallel_commit", ForeignServerRelationId, false}, + {"parallel_abort", ForeignServerRelationId, false}, {"keep_connections", ForeignServerRelationId, false}, {"password_required", UserMappingRelationId, false}, diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 2dc6386b40..58ec80fad7 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3506,10 +3506,12 @@ RESET postgres_fdw.application_name; RESET debug_discard_caches; -- =================================================================== --- test parallel commit +-- test parallel commit and parallel abort -- =================================================================== ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true'); ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true'); +ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true'); CREATE TABLE ploc1 (f1 int, f2 text); CREATE FOREIGN TABLE prem1 (f1 int, f2 text) @@ -3548,5 +3550,26 @@ COMMIT; SELECT * FROM prem1; SELECT * FROM prem2; +BEGIN; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; +SELECT * FROM prem2; + +BEGIN; +SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ROLLBACK TO SAVEPOINT s; +RELEASE SAVEPOINT s; +INSERT INTO prem1 VALUES (105, 'test1'); +INSERT INTO prem2 VALUES (205, 'test2'); +ABORT; +SELECT * FROM prem1; +SELECT * FROM prem2; + ALTER SERVER loopback OPTIONS (DROP parallel_commit); +ALTER SERVER loopback OPTIONS (DROP parallel_abort); ALTER SERVER loopback2 OPTIONS (DROP parallel_commit); +ALTER SERVER loopback2 OPTIONS (DROP parallel_abort); diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml index 238e2f84a6..196e0a986c 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -460,10 +460,10 @@ OPTIONS (ADD password_required 'false'); <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,27 +478,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] 9+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-17 16:46 Fujii Masao <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 2 replies; 9+ messages in thread From: Fujii Masao @ 2022-02-17 16:46 UTC (permalink / raw) To: Etsuro Fujita <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On 2022/02/11 21:59, Etsuro Fujita wrote: > I tweaked comments/docs a little bit as well. Thanks for updating the patches! I reviewed 0001 patch. It looks good to me except the following minor things. If these are addressed, I think that the 001 patch can be marked as ready for committer. + * Also determine to commit (sub)transactions opened on the remote server + * in parallel at (sub)transaction end. Like the comment "Determine whether to keep the connection ...", "determine to commit" should be "determine whether to commit"? "remote server" should be "remote servers"? + curlevel = GetCurrentTransactionNestLevel(); + snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); Why does pgfdw_finish_pre_subcommit_cleanup() need to call GetCurrentTransactionNestLevel() and construct the "RELEASE SAVEPOINT" query string again? pgfdw_subxact_callback() already does them and probably we can make it pass either of them to pgfdw_finish_pre_subcommit_cleanup() as its argument. + 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. "a foreign server" should be "foreign servers"? "in a local (sub)transaction" part seems not to be necessary. Regards, -- Fujii Masao Advanced Computing Technology Center Research and Development Headquarters NTT DATA CORPORATION ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-18 21:55 David Zhang <[email protected]> parent: Fujii Masao <[email protected]> 1 sibling, 1 reply; 9+ messages in thread From: David Zhang @ 2022-02-18 21:55 UTC (permalink / raw) To: Fujii Masao <[email protected]>; Etsuro Fujita <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> Thanks a lot for updating the patch. Tried to apply the patches to master branch, no warning found and regression test passed. Now, we have many places (5) calling the same function with a constant number 30000. Is this a good time to consider redefine this number a macro somewhere? Thank you, On 2022-02-17 8:46 a.m., Fujii Masao wrote: > > > On 2022/02/11 21:59, Etsuro Fujita wrote: >> I tweaked comments/docs a little bit as well. > > Thanks for updating the patches! > > I reviewed 0001 patch. It looks good to me except the following minor > things. If these are addressed, I think that the 001 patch can be > marked as ready for committer. > > + * Also determine to commit (sub)transactions opened on the > remote server > + * in parallel at (sub)transaction end. > > Like the comment "Determine whether to keep the connection ...", > "determine to commit" should be "determine whether to commit"? > > "remote server" should be "remote servers"? > > > + curlevel = GetCurrentTransactionNestLevel(); > + snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); > > Why does pgfdw_finish_pre_subcommit_cleanup() need to call > GetCurrentTransactionNestLevel() and construct the "RELEASE SAVEPOINT" > query string again? pgfdw_subxact_callback() already does them and > probably we can make it pass either of them to > pgfdw_finish_pre_subcommit_cleanup() as its argument. > > > + 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. > > "a foreign server" should be "foreign servers"? > > "in a local (sub)transaction" part seems not to be necessary. > > Regards, > -- David Software Engineer Highgo Software Inc. (Canada) www.highgo.ca ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-21 05:45 Etsuro Fujita <[email protected]> parent: Fujii Masao <[email protected]> 1 sibling, 1 reply; 9+ messages in thread From: Etsuro Fujita @ 2022-02-21 05:45 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Fri, Feb 18, 2022 at 1:46 AM Fujii Masao <[email protected]> wrote: > I reviewed 0001 patch. It looks good to me except the following minor things. If these are addressed, I think that the 001 patch can be marked as ready for committer. OK > + * Also determine to commit (sub)transactions opened on the remote server > + * in parallel at (sub)transaction end. > > Like the comment "Determine whether to keep the connection ...", "determine to commit" should be "determine whether to commit"? Agreed. I’ll change it as such. > "remote server" should be "remote servers"? Maybe I’m missing something, but we determine this for the given remote server, so it seems to me correct to say “the remote server”, not “the remote servers“. > + curlevel = GetCurrentTransactionNestLevel(); > + snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); > > Why does pgfdw_finish_pre_subcommit_cleanup() need to call GetCurrentTransactionNestLevel() and construct the "RELEASE SAVEPOINT" query string again? pgfdw_subxact_callback() already does them and probably we can make it pass either of them to pgfdw_finish_pre_subcommit_cleanup() as its argument. Yeah, that would save cycles, but I think that that makes code a bit unclean IMO. (To save cycles, I think we could also modify pgfdw_subxact_callback() to reuse the query in the while loop in that function when processing multiple open remote subtransactions there, but that would make code a bit complicated, so I don’t think it’s a good idea to do so, either.) So I’d vote for reconstructing the query in pgfdw_finish_pre_subcommit_cleanup() as we do in pgfdw_subxact_callback(). To avoid calling GetCurrentTransactionNestLevel() again, I think we could pass the curlevel variable to that function. > + 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. > > "a foreign server" should be "foreign servers"? I thought it would be good to say “a foreign server”, not “foreign servers”, because it makes clear that even remote transactions opened on a single foreign server are committed in parallel. (To say that this option is not for a specific foreign server, I added to the documentation “This option can only be specified for foreign servers”.) > "in a local (sub)transaction" part seems not to be necessary. And I thought adding it would make clear which remote transactions are committed in parallel. But maybe I’m missing something, so could you elaborate a bit more on these? Thanks for reviewing! Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-21 05:55 Etsuro Fujita <[email protected]> parent: David Zhang <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Etsuro Fujita @ 2022-02-21 05:55 UTC (permalink / raw) To: David Zhang <[email protected]>; +Cc: Fujii Masao <[email protected]>; PostgreSQL Hackers <[email protected]> On Sat, Feb 19, 2022 at 6:55 AM David Zhang <[email protected]> wrote: > Tried to apply the patches to master branch, no warning found and > regression test passed. Thanks for testing! > Now, we have many places (5) calling the same function with a constant > number 30000. Is this a good time to consider redefine this number a > macro somewhere? Yeah, I think that is a good idea. I’ll do so in the next version of the parallel-abort patch (#0003) if no objections. Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-21 16:03 Fujii Masao <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Fujii Masao @ 2022-02-21 16:03 UTC (permalink / raw) To: Etsuro Fujita <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On 2022/02/21 14:45, Etsuro Fujita wrote: > On Fri, Feb 18, 2022 at 1:46 AM Fujii Masao <[email protected]> wrote: >> I reviewed 0001 patch. It looks good to me except the following minor things. If these are addressed, I think that the 001 patch can be marked as ready for committer. > > OK > >> + * Also determine to commit (sub)transactions opened on the remote server >> + * in parallel at (sub)transaction end. >> >> Like the comment "Determine whether to keep the connection ...", "determine to commit" should be "determine whether to commit"? > > Agreed. I’ll change it as such. Thanks! If that's updated, IMO it's ok to commit the 0001 patch. After the commit, I will review 0002 and 0003 patches. Regards, -- Fujii Masao Advanced Computing Technology Center Research and Development Headquarters NTT DATA CORPORATION ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-23 06:30 Etsuro Fujita <[email protected]> parent: Fujii Masao <[email protected]> 0 siblings, 1 reply; 9+ messages in thread From: Etsuro Fujita @ 2022-02-23 06:30 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Tue, Feb 22, 2022 at 1:03 AM Fujii Masao <[email protected]> wrote: > On 2022/02/21 14:45, Etsuro Fujita wrote: > > On Fri, Feb 18, 2022 at 1:46 AM Fujii Masao <[email protected]> wrote: > >> I reviewed 0001 patch. It looks good to me except the following minor things. If these are addressed, I think that the 001 patch can be marked as ready for committer. > >> + * Also determine to commit (sub)transactions opened on the remote server > >> + * in parallel at (sub)transaction end. > >> > >> Like the comment "Determine whether to keep the connection ...", "determine to commit" should be "determine whether to commit"? > > > > Agreed. I’ll change it as such. Done. > Thanks! If that's updated, IMO it's ok to commit the 0001 patch. Cool! Attached is an updated patch. Other changes other than that: 1) I added the curlevel parameter to pgfdw_finish_pre_subcommit_cleanup() to avoid doing GetCurrentTransactionNestLevel() there, as proposed, and 2) tweaked comments a bit further, mostly for/in pgfdw_finish_pre_commit_cleanup() and pgfdw_finish_pre_subcommit_cleanup(). Barring objections, I’ll commit the patch. Thanks for reviewing! Best regards, Etsuro Fujita Attachments: [application/octet-stream] v5-0001-postgres-fdw-Add-support-for-parallel-commit.patch (19.1K, ../../CAPmGK16sRsm30a_1dpWPkVoTANg=F3H1vLRLyek-Xcw_r7JPkQ@mail.gmail.com/2-v5-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 f753c6e232..8c64d42dda 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,9 @@ 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, + bool consume_input); 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 +104,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 +112,9 @@ 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, + int curlevel); static bool UserMappingPasswordRequired(UserMapping *user); static bool disconnect_cached_connections(Oid serverid); @@ -316,14 +324,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 whether 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); + else if (strcmp(def->defname, "parallel_commit") == 0) + entry->parallel_commit = defGetBoolean(def); } /* Now try to make the connection */ @@ -623,10 +637,30 @@ configure_remote_session(PGconn *conn) void do_sql_command(PGconn *conn, const char *sql) { - PGresult *res; + do_sql_command_begin(conn, sql); + do_sql_command_end(conn, sql, false); +} +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, bool consume_input) +{ + PGresult *res; + + /* + * If requested, consume whatever data is available from the socket. + * (Note that if all data is available, this allows pgfdw_get_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(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); @@ -888,6 +922,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 +960,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 +1022,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 connections, finish cleaning them up */ + if (pending_entries) + { + Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT || + event == XACT_EVENT_PRE_COMMIT); + pgfdw_finish_pre_commit_cleanup(pending_entries); } /* @@ -1021,6 +1054,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 +1097,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 +1116,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 connections, finish cleaning them up */ + if (pending_entries) + { + Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB); + pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); } } @@ -1169,6 +1216,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 +1537,112 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel) entry->changing_xact_state = false; } +/* + * Finish pre-commit cleanup of connections on each of which we've sent a + * COMMIT command to the remote server. + */ +static void +pgfdw_finish_pre_commit_cleanup(List *pending_entries) +{ + ConnCacheEntry *entry; + List *pending_deallocs = NIL; + ListCell *lc; + + Assert(pending_entries); + + /* + * Get the result of the COMMIT command for each of the pending entries + */ + foreach(lc, pending_entries) + { + entry = (ConnCacheEntry *) lfirst(lc); + + Assert(entry->changing_xact_state); + /* + * We might already have received the result on the socket, so pass + * consume_input=true to try to consume it first + */ + do_sql_command_end(entry->conn, "COMMIT TRANSACTION", true); + entry->changing_xact_state = false; + + /* Do a DEALLOCATE ALL in parallel if needed */ + if (entry->have_prep_stmt && entry->have_error) + { + /* Ignore errors (see notes in pgfdw_xact_callback) */ + 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); + } + + /* No further work if no pending entries */ + if (!pending_deallocs) + return; + + /* + * Get the result of the DEALLOCATE command for each of the pending + * entries + */ + foreach(lc, pending_deallocs) + { + PGresult *res; + + entry = (ConnCacheEntry *) lfirst(lc); + + /* Ignore errors (see notes in pgfdw_xact_callback) */ + while ((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); + } +} + +/* + * Finish pre-subcommit cleanup of connections on each of which we've sent a + * RELEASE command to the remote server. + */ +static void +pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel) +{ + ConnCacheEntry *entry; + char sql[100]; + ListCell *lc; + + Assert(pending_entries); + + /* + * Get the result of the RELEASE command for each of the pending entries + */ + snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); + foreach(lc, pending_entries) + { + entry = (ConnCacheEntry *) lfirst(lc); + + Assert(entry->changing_xact_state); + /* + * We might already have received the result on the socket, so pass + * consume_input=true to try to consume it first + */ + do_sql_command_end(entry->conn, sql, true); + 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 057342083c..f210f91188 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 @@ -10933,3 +10933,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 af38e956e7..76d0b6dd0f 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 6c9f579c41..95b6b7192e 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3515,3 +3515,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 dc57fe4b0d..8ebf0dc3a0 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -456,6 +456,52 @@ 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, 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. + </para> + </listitem> + </varlistentry> + + </variablelist> + + </sect3> + <sect3> <title>Updatability Options</title> ^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit @ 2022-02-24 05:49 Etsuro Fujita <[email protected]> parent: Etsuro Fujita <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Etsuro Fujita @ 2022-02-24 05:49 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Wed, Feb 23, 2022 at 3:30 PM Etsuro Fujita <[email protected]> wrote: > Attached is an updated patch. > Barring objections, I’ll commit > the patch. I have committed the patch. I think the 0003 patch needs rebase. I'll update the patch. Best regards, Etsuro Fujita ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH v22 2/8] Row pattern recognition patch (parse/analysis). @ 2024-09-19 04:48 Tatsuo Ishii <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Tatsuo Ishii @ 2024-09-19 04:48 UTC (permalink / raw) --- src/backend/parser/parse_agg.c | 7 + src/backend/parser/parse_clause.c | 296 +++++++++++++++++++++++++++++- src/backend/parser/parse_expr.c | 6 + src/backend/parser/parse_func.c | 3 + 4 files changed, 311 insertions(+), 1 deletion(-) diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c index efa730c167..a80263f90d 100644 --- a/src/backend/parser/parse_agg.c +++ b/src/backend/parser/parse_agg.c @@ -580,6 +580,10 @@ check_agglevels_and_constraints(ParseState *pstate, Node *expr) errkind = true; break; + case EXPR_KIND_RPR_DEFINE: + errkind = true; + break; + /* * There is intentionally no default: case here, so that the * compiler will warn if we add a new ParseExprKind without @@ -970,6 +974,9 @@ transformWindowFuncCall(ParseState *pstate, WindowFunc *wfunc, case EXPR_KIND_CYCLE_MARK: errkind = true; break; + case EXPR_KIND_RPR_DEFINE: + errkind = true; + break; /* * There is intentionally no default: case here, so that the diff --git a/src/backend/parser/parse_clause.c b/src/backend/parser/parse_clause.c index 8118036495..9762dce81f 100644 --- a/src/backend/parser/parse_clause.c +++ b/src/backend/parser/parse_clause.c @@ -98,7 +98,14 @@ static WindowClause *findWindowClause(List *wclist, const char *name); static Node *transformFrameOffset(ParseState *pstate, int frameOptions, Oid rangeopfamily, Oid rangeopcintype, Oid *inRangeFunc, Node *clause); - +static void transformRPR(ParseState *pstate, WindowClause *wc, WindowDef *windef, + List **targetlist); +static List *transformDefineClause(ParseState *pstate, WindowClause *wc, WindowDef *windef, + List **targetlist); +static void transformPatternClause(ParseState *pstate, WindowClause *wc, + WindowDef *windef); +static List *transformMeasureClause(ParseState *pstate, WindowClause *wc, + WindowDef *windef); /* * transformFromClause - @@ -2956,6 +2963,10 @@ transformWindowDefinitions(ParseState *pstate, rangeopfamily, rangeopcintype, &wc->endInRangeFunc, windef->endOffset); + + /* Process Row Pattern Recognition related clauses */ + transformRPR(pstate, wc, windef, targetlist); + wc->winref = winref; result = lappend(result, wc); @@ -3820,3 +3831,286 @@ transformFrameOffset(ParseState *pstate, int frameOptions, return node; } + +/* + * transformRPR + * Process Row Pattern Recognition related clauses + */ +static void +transformRPR(ParseState *pstate, WindowClause *wc, WindowDef *windef, + List **targetlist) +{ + /* + * Window definition exists? + */ + if (windef == NULL) + return; + + /* + * Row Pattern Common Syntax clause exists? + */ + if (windef->rpCommonSyntax == NULL) + return; + + /* Check Frame option. Frame must start at current row */ + if ((wc->frameOptions & FRAMEOPTION_START_CURRENT_ROW) == 0) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("FRAME must start at current row when row patttern recognition is used"))); + + /* Transform AFTER MACH SKIP TO clause */ + wc->rpSkipTo = windef->rpCommonSyntax->rpSkipTo; + + /* Transform AFTER MACH SKIP TO variable */ + wc->rpSkipVariable = windef->rpCommonSyntax->rpSkipVariable; + + /* Transform SEEK or INITIAL clause */ + wc->initial = windef->rpCommonSyntax->initial; + + /* Transform DEFINE clause into list of TargetEntry's */ + wc->defineClause = transformDefineClause(pstate, wc, windef, targetlist); + + /* Check PATTERN clause and copy to patternClause */ + transformPatternClause(pstate, wc, windef); + + /* Transform MEASURE clause */ + transformMeasureClause(pstate, wc, windef); +} + +/* + * transformDefineClause Process DEFINE clause and transform ResTarget into + * list of TargetEntry. + * + * XXX we only support column reference in row pattern definition search + * condition, e.g. "price". <row pattern definition variable name>.<column + * reference> is not supported, e.g. "A.price". + */ +static List * +transformDefineClause(ParseState *pstate, WindowClause *wc, WindowDef *windef, + List **targetlist) +{ + /* DEFINE variable name initials */ + static char *defineVariableInitials = "abcdefghijklmnopqrstuvwxyz"; + + ListCell *lc, + *l; + ResTarget *restarget, + *r; + List *restargets; + List *defineClause; + char *name; + int initialLen; + int i; + + /* + * If Row Definition Common Syntax exists, DEFINE clause must exist. (the + * raw parser should have already checked it.) + */ + Assert(windef->rpCommonSyntax->rpDefs != NULL); + + /* + * Check and add "A AS A IS TRUE" if pattern variable is missing in DEFINE + * per the SQL standard. + */ + restargets = NIL; + foreach(lc, windef->rpCommonSyntax->rpPatterns) + { + A_Expr *a; + bool found = false; + + if (!IsA(lfirst(lc), A_Expr)) + ereport(ERROR, + errmsg("node type is not A_Expr")); + + a = (A_Expr *) lfirst(lc); + name = strVal(a->lexpr); + + foreach(l, windef->rpCommonSyntax->rpDefs) + { + restarget = (ResTarget *) lfirst(l); + + if (!strcmp(restarget->name, name)) + { + found = true; + break; + } + } + + if (!found) + { + /* + * "name" is missing. So create "name AS name IS TRUE" ResTarget + * node and add it to the temporary list. + */ + A_Const *n; + + restarget = makeNode(ResTarget); + n = makeNode(A_Const); + n->val.boolval.type = T_Boolean; + n->val.boolval.boolval = true; + n->location = -1; + restarget->name = pstrdup(name); + restarget->indirection = NIL; + restarget->val = (Node *) n; + restarget->location = -1; + restargets = lappend((List *) restargets, restarget); + } + } + + if (list_length(restargets) >= 1) + { + /* add missing DEFINEs */ + windef->rpCommonSyntax->rpDefs = + list_concat(windef->rpCommonSyntax->rpDefs, restargets); + list_free(restargets); + } + + /* + * Check for duplicate row pattern definition variables. The standard + * requires that no two row pattern definition variable names shall be + * equivalent. + */ + restargets = NIL; + foreach(lc, windef->rpCommonSyntax->rpDefs) + { + restarget = (ResTarget *) lfirst(lc); + name = restarget->name; + + /* + * Add DEFINE expression (Restarget->val) to the targetlist as a + * TargetEntry if it does not exist yet. Planner will add the column + * ref var node to the outer plan's target list later on. This makes + * DEFINE expression could access the outer tuple while evaluating + * PATTERN. + * + * XXX: adding whole expressions of DEFINE to the plan.targetlist is + * not so good, because it's not necessary to evalute the expression + * in the target list while running the plan. We should extract the + * var nodes only then add them to the plan.targetlist. + */ + findTargetlistEntrySQL99(pstate, (Node *) restarget->val, + targetlist, EXPR_KIND_RPR_DEFINE); + + /* + * Make sure that the row pattern definition search condition is a + * boolean expression. + */ + transformWhereClause(pstate, restarget->val, + EXPR_KIND_RPR_DEFINE, "DEFINE"); + + foreach(l, restargets) + { + char *n; + + r = (ResTarget *) lfirst(l); + n = r->name; + + if (!strcmp(n, name)) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("row pattern definition variable name \"%s\" appears more than once in DEFINE clause", + name), + parser_errposition(pstate, exprLocation((Node *) r)))); + } + restargets = lappend(restargets, restarget); + } + list_free(restargets); + + /* + * Create list of row pattern DEFINE variable name's initial. We assign + * [a-z] to them (up to 26 variable names are allowed). + */ + restargets = NIL; + i = 0; + initialLen = strlen(defineVariableInitials); + + foreach(lc, windef->rpCommonSyntax->rpDefs) + { + char initial[2]; + + restarget = (ResTarget *) lfirst(lc); + name = restarget->name; + + if (i >= initialLen) + { + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("number of row pattern definition variable names exceeds %d", + initialLen), + parser_errposition(pstate, + exprLocation((Node *) restarget)))); + } + initial[0] = defineVariableInitials[i++]; + initial[1] = '\0'; + wc->defineInitial = lappend(wc->defineInitial, + makeString(pstrdup(initial))); + } + + defineClause = transformTargetList(pstate, windef->rpCommonSyntax->rpDefs, + EXPR_KIND_RPR_DEFINE); + + /* mark column origins */ + markTargetListOrigins(pstate, defineClause); + + /* mark all nodes in the DEFINE clause tree with collation information */ + assign_expr_collations(pstate, (Node *) defineClause); + + return defineClause; +} + +/* + * transformPatternClause + * Process PATTERN clause and return PATTERN clause in the raw parse tree + */ +static void +transformPatternClause(ParseState *pstate, WindowClause *wc, + WindowDef *windef) +{ + ListCell *lc; + + /* + * Row Pattern Common Syntax clause exists? + */ + if (windef->rpCommonSyntax == NULL) + return; + + wc->patternVariable = NIL; + wc->patternRegexp = NIL; + foreach(lc, windef->rpCommonSyntax->rpPatterns) + { + A_Expr *a; + char *name; + char *regexp; + + if (!IsA(lfirst(lc), A_Expr)) + ereport(ERROR, + errmsg("node type is not A_Expr")); + + a = (A_Expr *) lfirst(lc); + name = strVal(a->lexpr); + + wc->patternVariable = lappend(wc->patternVariable, makeString(pstrdup(name))); + regexp = strVal(lfirst(list_head(a->name))); + + wc->patternRegexp = lappend(wc->patternRegexp, makeString(pstrdup(regexp))); + } +} + +/* + * transformMeasureClause + * Process MEASURE clause + * XXX MEASURE clause is not supported yet + */ +static List * +transformMeasureClause(ParseState *pstate, WindowClause *wc, + WindowDef *windef) +{ + if (windef->rowPatternMeasures == NIL) + return NIL; + + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("%s", "MEASURE clause is not supported yet"), + parser_errposition(pstate, exprLocation((Node *) windef->rowPatternMeasures)))); + return NIL; +} diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c index 36c1b7a88f..fe154bcaa0 100644 --- a/src/backend/parser/parse_expr.c +++ b/src/backend/parser/parse_expr.c @@ -577,6 +577,7 @@ transformColumnRef(ParseState *pstate, ColumnRef *cref) case EXPR_KIND_COPY_WHERE: case EXPR_KIND_GENERATED_COLUMN: case EXPR_KIND_CYCLE_MARK: + case EXPR_KIND_RPR_DEFINE: /* okay */ break; @@ -1860,6 +1861,9 @@ transformSubLink(ParseState *pstate, SubLink *sublink) case EXPR_KIND_GENERATED_COLUMN: err = _("cannot use subquery in column generation expression"); break; + case EXPR_KIND_RPR_DEFINE: + err = _("cannot use subquery in DEFINE expression"); + break; /* * There is intentionally no default: case here, so that the @@ -3199,6 +3203,8 @@ ParseExprKindName(ParseExprKind exprKind) return "GENERATED AS"; case EXPR_KIND_CYCLE_MARK: return "CYCLE"; + case EXPR_KIND_RPR_DEFINE: + return "DEFINE"; /* * There is intentionally no default: case here, so that the diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c index 9b23344a3b..4c482abb30 100644 --- a/src/backend/parser/parse_func.c +++ b/src/backend/parser/parse_func.c @@ -2658,6 +2658,9 @@ check_srf_call_placement(ParseState *pstate, Node *last_srf, int location) case EXPR_KIND_CYCLE_MARK: errkind = true; break; + case EXPR_KIND_RPR_DEFINE: + errkind = true; + break; /* * There is intentionally no default: case here, so that the -- 2.25.1 ----Next_Part(Thu_Sep_19_13_59_47_2024_608)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v22-0003-Row-pattern-recognition-patch-rewriter.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2024-09-19 04:48 UTC | newest] Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2022-02-11 12:59 Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit Etsuro Fujita <[email protected]> 2022-02-17 16:46 ` Fujii Masao <[email protected]> 2022-02-18 21:55 ` David Zhang <[email protected]> 2022-02-21 05:55 ` Etsuro Fujita <[email protected]> 2022-02-21 05:45 ` Etsuro Fujita <[email protected]> 2022-02-21 16:03 ` Fujii Masao <[email protected]> 2022-02-23 06:30 ` Etsuro Fujita <[email protected]> 2022-02-24 05:49 ` Etsuro Fujita <[email protected]> 2024-09-19 04:48 [PATCH v22 2/8] Row pattern recognition patch (parse/analysis). Tatsuo Ishii <[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