agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH v27 2/5] Add pg_depend.refobjversion. 3+ messages / 3 participants [nested] [flat]
* [PATCH v27 2/5] Add pg_depend.refobjversion. @ 2019-05-28 18:16 Thomas Munro <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Thomas Munro @ 2019-05-28 18:16 UTC (permalink / raw) Provide a place for the version of referenced database objects to be recorded. Later commits will be able to use this to record dependencies on collation versions for indexes and maybe more. Author: Thomas Munro <[email protected]> Reviewed-by: Julien Rouhaud <[email protected]> Reviewed-by: Peter Eisentraut <[email protected]> Reviewed-by: Michael Paquier <[email protected]> Discussion: https://postgr.es/m/CAEepm%3D0uEQCpfq_%2BLYFBdArCe4Ot98t1aR4eYiYTe%3DyavQygiQ%40mail.gmail.com --- doc/src/sgml/catalogs.sgml | 12 +++++++ src/backend/catalog/dependency.c | 14 +++++--- src/backend/catalog/pg_depend.c | 14 +++++--- src/bin/initdb/initdb.c | 44 +++++++++++------------ src/include/catalog/dependency.h | 1 + src/include/catalog/pg_depend.h | 4 +++ src/include/catalog/toasting.h | 1 + src/test/regress/expected/misc_sanity.out | 4 +-- 8 files changed, 62 insertions(+), 32 deletions(-) diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index a28ae89e13..2567863dd9 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -3293,6 +3293,18 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l A code defining the specific semantics of this dependency relationship; see text </para></entry> </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>refobjversion</structfield> <type>text</type> + </para> + <para> + An optional version for the referenced object. The only current use of + <structfield>refobjversion</structfield> is to record dependencies + between indexes and collation versions. + </para></entry> + </row> + </tbody> </tgroup> </table> diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c index f515e2c308..1a927377e7 100644 --- a/src/backend/catalog/dependency.c +++ b/src/backend/catalog/dependency.c @@ -1600,7 +1600,9 @@ recordDependencyOnExpr(const ObjectAddress *depender, /* And record 'em */ recordMultipleDependencies(depender, - context.addrs->refs, context.addrs->numrefs, + context.addrs->refs, + context.addrs->numrefs, + NULL, behavior); free_object_addresses(context.addrs); @@ -1687,7 +1689,9 @@ recordDependencyOnSingleRelExpr(const ObjectAddress *depender, /* Record the self-dependencies with the appropriate direction */ if (!reverse_self) recordMultipleDependencies(depender, - self_addrs->refs, self_addrs->numrefs, + self_addrs->refs, + self_addrs->numrefs, + NULL, self_behavior); else { @@ -1707,7 +1711,9 @@ recordDependencyOnSingleRelExpr(const ObjectAddress *depender, /* Record the external dependencies */ recordMultipleDependencies(depender, - context.addrs->refs, context.addrs->numrefs, + context.addrs->refs, + context.addrs->numrefs, + NULL, behavior); free_object_addresses(context.addrs); @@ -2679,7 +2685,7 @@ record_object_address_dependencies(const ObjectAddress *depender, { eliminate_duplicate_dependencies(referenced); recordMultipleDependencies(depender, - referenced->refs, referenced->numrefs, + referenced->refs, referenced->numrefs, NULL, behavior); } diff --git a/src/backend/catalog/pg_depend.c b/src/backend/catalog/pg_depend.c index 70baf03178..d86f85d142 100644 --- a/src/backend/catalog/pg_depend.c +++ b/src/backend/catalog/pg_depend.c @@ -24,6 +24,7 @@ #include "catalog/pg_extension.h" #include "commands/extension.h" #include "miscadmin.h" +#include "utils/builtins.h" #include "utils/fmgroids.h" #include "utils/lsyscache.h" #include "utils/rel.h" @@ -44,7 +45,7 @@ recordDependencyOn(const ObjectAddress *depender, const ObjectAddress *referenced, DependencyType behavior) { - recordMultipleDependencies(depender, referenced, 1, behavior); + recordMultipleDependencies(depender, referenced, 1, NULL, behavior); } /* @@ -55,6 +56,7 @@ void recordMultipleDependencies(const ObjectAddress *depender, const ObjectAddress *referenced, int nreferenced, + const char *version, DependencyType behavior) { Relation dependDesc; @@ -79,8 +81,6 @@ recordMultipleDependencies(const ObjectAddress *depender, /* Don't open indexes unless we need to make an update */ indstate = NULL; - memset(nulls, false, sizeof(nulls)); - for (i = 0; i < nreferenced; i++, referenced++) { /* @@ -94,6 +94,9 @@ recordMultipleDependencies(const ObjectAddress *depender, * Record the Dependency. Note we don't bother to check for * duplicate dependencies; there's no harm in them. */ + memset(nulls, false, sizeof(nulls)); + memset(values, 0, sizeof(values)); + values[Anum_pg_depend_classid - 1] = ObjectIdGetDatum(depender->classId); values[Anum_pg_depend_objid - 1] = ObjectIdGetDatum(depender->objectId); values[Anum_pg_depend_objsubid - 1] = Int32GetDatum(depender->objectSubId); @@ -101,9 +104,12 @@ recordMultipleDependencies(const ObjectAddress *depender, values[Anum_pg_depend_refclassid - 1] = ObjectIdGetDatum(referenced->classId); values[Anum_pg_depend_refobjid - 1] = ObjectIdGetDatum(referenced->objectId); values[Anum_pg_depend_refobjsubid - 1] = Int32GetDatum(referenced->objectSubId); + if (version) + values[Anum_pg_depend_refobjversion - 1] = CStringGetTextDatum(version); + else + nulls[Anum_pg_depend_refobjversion - 1] = true; values[Anum_pg_depend_deptype - 1] = CharGetDatum((char) behavior); - tup = heap_form_tuple(dependDesc->rd_att, values, nulls); /* fetch index info only when we know we need it */ diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c index 786672b1b6..4b2a58cf18 100644 --- a/src/bin/initdb/initdb.c +++ b/src/bin/initdb/initdb.c @@ -1572,55 +1572,55 @@ setup_depend(FILE *cmdfd) "DELETE FROM pg_shdepend;\n\n", "VACUUM pg_shdepend;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_class;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_proc;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_type;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_cast;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_constraint;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_conversion;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_attrdef;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_language;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_operator;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_opclass;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_opfamily;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_am;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_amop;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_amproc;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_rewrite;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_trigger;\n\n", /* * restriction here to avoid pinning the public namespace */ - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_namespace " " WHERE nspname LIKE 'pg%';\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_ts_parser;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_ts_dict;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_ts_template;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_ts_config;\n\n", - "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p' " + "INSERT INTO pg_depend SELECT 0,0,0, tableoid,oid,0, 'p',NULL" " FROM pg_collation;\n\n", "INSERT INTO pg_shdepend SELECT 0,0,0,0, tableoid,oid, 'p' " " FROM pg_authid;\n\n", diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h index a8f7e9965b..3baa5e498a 100644 --- a/src/include/catalog/dependency.h +++ b/src/include/catalog/dependency.h @@ -189,6 +189,7 @@ extern void recordDependencyOn(const ObjectAddress *depender, extern void recordMultipleDependencies(const ObjectAddress *depender, const ObjectAddress *referenced, int nreferenced, + const char *version, DependencyType behavior); extern void recordDependencyOnCurrentExtension(const ObjectAddress *object, diff --git a/src/include/catalog/pg_depend.h b/src/include/catalog/pg_depend.h index ccf0a98330..7489022795 100644 --- a/src/include/catalog/pg_depend.h +++ b/src/include/catalog/pg_depend.h @@ -61,6 +61,10 @@ CATALOG(pg_depend,2608,DependRelationId) * field. See DependencyType in catalog/dependency.h. */ char deptype; /* see codes in dependency.h */ +#ifdef CATALOG_VARLEN + text refobjversion; /* version tracking, NULL if not used or + * unknown */ +#endif } FormData_pg_depend; /* ---------------- diff --git a/src/include/catalog/toasting.h b/src/include/catalog/toasting.h index 8f131893dc..e320d82203 100644 --- a/src/include/catalog/toasting.h +++ b/src/include/catalog/toasting.h @@ -53,6 +53,7 @@ DECLARE_TOAST(pg_aggregate, 4159, 4160); DECLARE_TOAST(pg_attrdef, 2830, 2831); DECLARE_TOAST(pg_constraint, 2832, 2833); DECLARE_TOAST(pg_default_acl, 4143, 4144); +DECLARE_TOAST(pg_depend, 8888, 8889); DECLARE_TOAST(pg_description, 2834, 2835); DECLARE_TOAST(pg_event_trigger, 4145, 4146); DECLARE_TOAST(pg_extension, 4147, 4148); diff --git a/src/test/regress/expected/misc_sanity.out b/src/test/regress/expected/misc_sanity.out index 8538173ff8..d40afeef78 100644 --- a/src/test/regress/expected/misc_sanity.out +++ b/src/test/regress/expected/misc_sanity.out @@ -18,8 +18,8 @@ WHERE refclassid = 0 OR refobjid = 0 OR deptype NOT IN ('a', 'e', 'i', 'n', 'p') OR (deptype != 'p' AND (classid = 0 OR objid = 0)) OR (deptype = 'p' AND (classid != 0 OR objid != 0 OR objsubid != 0)); - classid | objid | objsubid | refclassid | refobjid | refobjsubid | deptype ----------+-------+----------+------------+----------+-------------+--------- + classid | objid | objsubid | refclassid | refobjid | refobjsubid | deptype | refobjversion +---------+-------+----------+------------+----------+-------------+---------+--------------- (0 rows) -- **************** pg_shdepend **************** -- 2.20.1 --o56keubw4jpehgac Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename="v27-0003-Track-collation-versions-for-indexes.patch" ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Refactoring postgres_fdw/connection.c @ 2022-08-04 04:11 Amul Sul <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Amul Sul @ 2022-08-04 04:11 UTC (permalink / raw) To: Fujii Masao <[email protected]>; +Cc: Kyotaro Horiguchi <[email protected]>; PostgreSQL Hackers <[email protected]> On Thu, Jul 28, 2022 at 11:56 AM Fujii Masao <[email protected]> wrote: > > > > On 2022/07/27 10:36, Kyotaro Horiguchi wrote: > > At Tue, 26 Jul 2022 18:33:04 +0900, Fujii Masao <[email protected]> wrote in > >>> I'm not sure the two are similar with each other. The new function > >>> pgfdw_exec_pre_commit() looks like a merger of two isolated code paths > >>> intended to share a seven-line codelet. I feel the code gets a bit > >>> harder to understand after the change. I mildly oppose to this part. > >> > >> If so, we can pgfdw_exec_pre_commit() into two, one is the common > >> function that sends or executes the command (i.e., calls > >> do_sql_command_begin() or do_sql_command()), and another is > >> the function only for toplevel. The latter function calls > >> the common function and then executes DEALLOCATE ALL things. > >> > >> But this is not the way that other functions like > >> pgfdw_abort_cleanup() > >> is implemented. Those functions have both codes for toplevel and > >> !toplevel (i.e., subxact), and run the processings depending > >> on the argument "toplevel". So I'm thinking that > >> pgfdw_exec_pre_commit() implemented in the same way is better. > > > > I didn't see it from that viewpoint but I don't think that > > unconditionally justifies other refactoring. If we merge > > pgfdw_finish_pre_(sub)?commit_cleanup()s this way, in turn > > pgfdw_subxact_callback() and pgfdw_xact_callback() are going to be > > almost identical except event IDs to handle. But I don't think we > > would want to merge them. > > I don't think they are so identical because (as you say) they have to handle different event IDs. So I agree we don't want to merge them. > > > > A concern on 0002 is that it is hiding the subxact-specific steps from > > the subxact callback. It would look reasonable if it were called from > > two or more places for each topleve and !toplevel, but actually it has > > only one caller for each. So I think that pgfdw_exec_pre_commit > > should not do that and should be renamed to pgfdw_commit_remote() or > > something. On the other hand pgfdw_finish_pre_commit() hides > > toplevel-specific steps from the caller so the same argument holds. > > So you conclusion is to rename pgfdw_exec_pre_commit() to pgfdw_commit_remote() or something? > > > > Another point that makes me concern about the patch is the new > > function takes an SQL statement, along with the toplevel flag. I guess > > the reason is that the command for subxact (RELEASE SAVEPOINT %d) > > requires the current transaction level. However, the values > > isobtainable very cheap within the cleanup functions. So I propose to > > get rid of the parameter "sql" from the two functions. > > Yes, that's possible. That is, pgfdw_exec_pre_commit() can construct the query string by executing the following codes, instead of accepting the query as an argument. But one downside of this approach is that the following codes are executed for every remote subtransaction entries. Maybe it's cheap to construct the query string as follows, but I'd like to avoid any unnecessray overhead if possible. So the patch makes the caller, pgfdw_subxact_callback(), construct the query string only once and give it to pgfdw_exec_pre_commit(). > > curlevel = GetCurrentTransactionNestLevel(); > snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); > Another possibility I can see is that instead of calling pgfdw_exec_pre_commit() (similarly pgfdw_abort_cleanup) for every connection entry, we should call that once from the callback function, and for that we need to move the hash table loop inside that function. The structure of the callback function looks a little fuzzy to me where the same event is checked for every entry of the connection hash table. Instead of simply move that loop should be inside those function (e.g. pgfdw_exec_pre_commit and pgfdw_abort_cleanup), and let called those function called once w.r.t to event and that function should take care of every entry of the connection hash table. The benefit is that we would save a few processing cycles that needed to match events and call the same function for each connection entry. I tried this refactoring in 0004 patch which is not complete, and reattaching other patches too to make CFboat happy. Thoughts? Suggestions? Regards, Amul Attachments: [application/x-patch] v2-0003-Merge-pgfdw_finish_pre_commit_cleanup-and-pgfdw_f.patch (5.0K, ../../CAAJ_b97PdG0ArsaUpezDpbFbJhP6FPyYTQPpRyJFDLoPwQKQBg@mail.gmail.com/2-v2-0003-Merge-pgfdw_finish_pre_commit_cleanup-and-pgfdw_f.patch) download | inline diff: From d6e241cb946afe3b74c2893bda6dab8d3288716b Mon Sep 17 00:00:00 2001 From: Fujii Masao <[email protected]> Date: Mon, 25 Jul 2022 23:27:14 +0900 Subject: [PATCH v2 3/4] Merge pgfdw_finish_pre_commit_cleanup and pgfdw_finish_pre_subcommit_cleanup into one. --- contrib/postgres_fdw/connection.c | 78 ++++++++++--------------------- 1 file changed, 25 insertions(+), 53 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index ec290459be3..6e23046ad69 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -113,9 +113,8 @@ static bool pgfdw_get_result_timed(PGconn *conn, TimestampTz endtime, static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); static bool pgfdw_exec_pre_commit(ConnCacheEntry *entry, const char *sql, List **pending_entries, 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 void pgfdw_finish_pre_commit(List *pending_entries, const char *sql, + bool toplevel); static bool UserMappingPasswordRequired(UserMapping *user); static bool disconnect_cached_connections(Oid serverid); @@ -954,7 +953,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) { Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT || event == XACT_EVENT_PRE_COMMIT); - pgfdw_finish_pre_commit_cleanup(pending_entries); + pgfdw_finish_pre_commit(pending_entries, "COMMIT TRANSACTION", true); } /* @@ -1031,7 +1030,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, if (pending_entries) { Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB); - pgfdw_finish_pre_subcommit_cleanup(pending_entries, curlevel); + pgfdw_finish_pre_commit(pending_entries, sql, false); } } @@ -1523,11 +1522,14 @@ pgfdw_exec_pre_commit(ConnCacheEntry *entry, const char *sql, } /* - * Finish pre-commit cleanup of connections on each of which we've sent a - * COMMIT command to the remote server. + * Wait for all remote transactions or subtransactions to be committed + * and finish pre-commit. + * + * "toplevel" should be set to true if toplevel (main) transaction is + * committed, false otherwise. */ static void -pgfdw_finish_pre_commit_cleanup(List *pending_entries) +pgfdw_finish_pre_commit(List *pending_entries, const char *sql, bool toplevel) { ConnCacheEntry *entry; List *pending_deallocs = NIL; @@ -1536,7 +1538,8 @@ pgfdw_finish_pre_commit_cleanup(List *pending_entries) Assert(pending_entries); /* - * Get the result of the COMMIT command for each of the pending entries + * Get the result of COMMIT or RELEASE command for each of the pending + * entries. */ foreach(lc, pending_entries) { @@ -1548,23 +1551,26 @@ pgfdw_finish_pre_commit_cleanup(List *pending_entries) * 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); + do_sql_command_end(entry->conn, sql, true); entry->changing_xact_state = false; /* Do a DEALLOCATE ALL in parallel if needed */ - if (entry->have_prep_stmt && entry->have_error) + if (toplevel) { - /* Ignore errors (see notes in pgfdw_xact_callback) */ - if (PQsendQuery(entry->conn, "DEALLOCATE ALL")) + if (entry->have_prep_stmt && entry->have_error) { - pending_deallocs = lappend(pending_deallocs, entry); - continue; + /* 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; } - entry->have_prep_stmt = false; - entry->have_error = false; - pgfdw_reset_xact_state(entry, true); + pgfdw_reset_xact_state(entry, toplevel); } /* No further work if no pending entries */ @@ -1588,41 +1594,7 @@ pgfdw_finish_pre_commit_cleanup(List *pending_entries) 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); + pgfdw_reset_xact_state(entry, toplevel); } } -- 2.18.0 [application/x-patch] v2-0002-Add-common-function-to-commit-xact-or-subxact-dur.patch (6.8K, ../../CAAJ_b97PdG0ArsaUpezDpbFbJhP6FPyYTQPpRyJFDLoPwQKQBg@mail.gmail.com/3-v2-0002-Add-common-function-to-commit-xact-or-subxact-dur.patch) download | inline diff: From 087d1aa83850577ee75792e78f310fe45e3240dd Mon Sep 17 00:00:00 2001 From: Fujii Masao <[email protected]> Date: Mon, 25 Jul 2022 22:45:06 +0900 Subject: [PATCH v2 2/4] Add common function to commit xact or subxact during pre-commit. --- contrib/postgres_fdw/connection.c | 122 ++++++++++++++++-------------- 1 file changed, 66 insertions(+), 56 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index cbee2854803..ec290459be3 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -111,6 +111,8 @@ static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, static bool pgfdw_get_result_timed(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out); static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); +static bool pgfdw_exec_pre_commit(ConnCacheEntry *entry, const char *sql, + List **pending_entries, bool toplevel); static void pgfdw_finish_pre_commit_cleanup(List *pending_entries); static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries, int curlevel); @@ -894,8 +896,6 @@ pgfdw_xact_callback(XactEvent event, void *arg) hash_seq_init(&scan, ConnectionHash); while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) { - PGresult *res; - /* Ignore cache entry if no open connection right now */ if (entry->conn == NULL) continue; @@ -911,45 +911,10 @@ pgfdw_xact_callback(XactEvent event, void *arg) case XACT_EVENT_PARALLEL_PRE_COMMIT: case XACT_EVENT_PRE_COMMIT: - /* - * If abort cleanup previously failed for this connection, - * we can't issue any more commands against it. - */ - pgfdw_reject_incomplete_xact_state_change(entry); - /* 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); + if (pgfdw_exec_pre_commit(entry, "COMMIT TRANSACTION", + &pending_entries, true)) continue; - } - do_sql_command(entry->conn, "COMMIT TRANSACTION"); - entry->changing_xact_state = false; - - /* - * If there were any errors in subtransactions, and we - * made prepared statements, do a DEALLOCATE ALL to make - * sure we get rid of all prepared statements. This is - * annoying and not terribly bulletproof, but it's - * probably not worth trying harder. - * - * DEALLOCATE ALL only exists in 8.3 and later, so this - * constrains how old a server postgres_fdw can - * communicate with. We intentionally ignore errors in - * the DEALLOCATE, so that we can hobble along to some - * extent with older servers (leaking prepared statements - * as we go; but we don't really support update operations - * pre-8.3 anyway). - */ - if (entry->have_prep_stmt && entry->have_error) - { - res = PQexec(entry->conn, "DEALLOCATE ALL"); - PQclear(res); - } - entry->have_prep_stmt = false; - entry->have_error = false; break; case XACT_EVENT_PRE_PREPARE: @@ -1014,6 +979,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, ConnCacheEntry *entry; int curlevel; List *pending_entries = NIL; + char sql[100]; /* Nothing to do at subxact start, nor after commit. */ if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB || @@ -1029,11 +995,11 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, * of the current level, and close them. */ curlevel = GetCurrentTransactionNestLevel(); + snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); + hash_seq_init(&scan, ConnectionHash); while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) { - char sql[100]; - /* * We only care about connections with open remote subtransactions of * the current level. @@ -1047,23 +1013,9 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) { - /* - * If abort cleanup previously failed for this connection, we - * can't issue any more commands against it. - */ - pgfdw_reject_incomplete_xact_state_change(entry); - /* 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); + if (pgfdw_exec_pre_commit(entry, sql, &pending_entries, false)) continue; - } - do_sql_command(entry->conn, sql); - entry->changing_xact_state = false; } else { @@ -1512,6 +1464,64 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) entry->changing_xact_state = false; } +/* + * Commit all remote transactions or subtransactions during pre-commit. + * + * If parallel_commit is enabled at this connection cache entry and + * the result of "sql" needs to be gotten later, return true and append + * this entry to "pending_entries". + * + * "toplevel" should be set to true if toplevel (main) transaction is + * committed, false otherwise. + */ +static bool +pgfdw_exec_pre_commit(ConnCacheEntry *entry, const char *sql, + List **pending_entries, bool toplevel) +{ + PGresult *res; + + /* + * If abort cleanup previously failed for this connection, we can't issue + * any more commands against it. + */ + pgfdw_reject_incomplete_xact_state_change(entry); + + entry->changing_xact_state = true; + if (entry->parallel_commit) + { + do_sql_command_begin(entry->conn, sql); + *pending_entries = lappend(*pending_entries, entry); + return true; + } + do_sql_command(entry->conn, sql); + entry->changing_xact_state = false; + + if (!toplevel) + return false; + + /* + * If there were any errors in subtransactions, and we made prepared + * statements, do a DEALLOCATE ALL to make sure we get rid of all prepared + * statements. This is annoying and not terribly bulletproof, but it's + * probably not worth trying harder. + * + * DEALLOCATE ALL only exists in 8.3 and later, so this constrains how old + * a server postgres_fdw can communicate with. We intentionally ignore + * errors in the DEALLOCATE, so that we can hobble along to some extent + * with older servers (leaking prepared statements as we go; but we don't + * really support update operations pre-8.3 anyway). + */ + if (entry->have_prep_stmt && entry->have_error) + { + res = PQexec(entry->conn, "DEALLOCATE ALL"); + PQclear(res); + } + entry->have_prep_stmt = false; + entry->have_error = false; + + return false; +} + /* * Finish pre-commit cleanup of connections on each of which we've sent a * COMMIT command to the remote server. -- 2.18.0 [application/x-patch] v2-0004-TRIAL-cleanup-xact-callback-WIP.patch (16.7K, ../../CAAJ_b97PdG0ArsaUpezDpbFbJhP6FPyYTQPpRyJFDLoPwQKQBg@mail.gmail.com/4-v2-0004-TRIAL-cleanup-xact-callback-WIP.patch) download | inline diff: From f8aaf91d0fdb88c6726c45f924d8dea19bf98cff Mon Sep 17 00:00:00 2001 From: Amul Sul <[email protected]> Date: Wed, 3 Aug 2022 10:17:09 -0400 Subject: [PATCH v2 4/4] TRIAL: cleanup xact callback - WIP --- contrib/postgres_fdw/connection.c | 420 +++++++++++++++--------------- 1 file changed, 213 insertions(+), 207 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 6e23046ad69..181d2f83e32 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -110,9 +110,8 @@ static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors); static bool pgfdw_get_result_timed(PGconn *conn, TimestampTz endtime, PGresult **result, bool *timed_out); -static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel); -static bool pgfdw_exec_pre_commit(ConnCacheEntry *entry, const char *sql, - List **pending_entries, bool toplevel); +static void pgfdw_abort_cleanup(bool toplevel); +static void pgfdw_exec_pre_commit(bool toplevel); static void pgfdw_finish_pre_commit(List *pending_entries, const char *sql, bool toplevel); static bool UserMappingPasswordRequired(UserMapping *user); @@ -880,80 +879,47 @@ pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, static void 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) return; - /* - * Scan all connection cache entries to find open remote transactions, and - * close them. - */ - hash_seq_init(&scan, ConnectionHash); - while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) + switch (event) { - /* Ignore cache entry if no open connection right now */ - if (entry->conn == NULL) - continue; - - /* If it has an open remote transaction, try to close it */ - if (entry->xact_depth > 0) - { - elog(DEBUG3, "closing remote transaction on connection %p", - entry->conn); + case XACT_EVENT_PARALLEL_PRE_COMMIT: + case XACT_EVENT_PRE_COMMIT: - switch (event) - { - case XACT_EVENT_PARALLEL_PRE_COMMIT: - case XACT_EVENT_PRE_COMMIT: - - /* Commit all remote transactions during pre-commit */ - if (pgfdw_exec_pre_commit(entry, "COMMIT TRANSACTION", - &pending_entries, true)) - continue; - break; - case XACT_EVENT_PRE_PREPARE: - - /* - * We disallow any remote transactions, since it's not - * very reasonable to hold them open until the prepared - * transaction is committed. For the moment, throw error - * unconditionally; later we might allow read-only cases. - * Note that the error will cause us to come right back - * here with event == XACT_EVENT_ABORT, so we'll clean up - * the connection state at that point. - */ - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot PREPARE a transaction that has operated on postgres_fdw foreign tables"))); - break; - case XACT_EVENT_PARALLEL_COMMIT: - case XACT_EVENT_COMMIT: - case XACT_EVENT_PREPARE: - /* Pre-commit should have closed the open transaction */ - elog(ERROR, "missed cleaning up connection during pre-commit"); - break; - case XACT_EVENT_PARALLEL_ABORT: - case XACT_EVENT_ABORT: - /* Rollback all remote transactions during abort */ - pgfdw_abort_cleanup(entry, true); - break; - } - } + /* Commit all remote transactions during pre-commit */ + pgfdw_exec_pre_commit(true); + break; - /* Reset state to show we're out of a transaction */ - pgfdw_reset_xact_state(entry, true); - } + case XACT_EVENT_PRE_PREPARE: - /* 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(pending_entries, "COMMIT TRANSACTION", true); + /* + * We disallow any remote transactions, since it's not + * very reasonable to hold them open until the prepared + * transaction is committed. For the moment, throw error + * unconditionally; later we might allow read-only cases. + * Note that the error will cause us to come right back + * here with event == XACT_EVENT_ABORT, so we'll clean up + * the connection state at that point. + */ + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot PREPARE a transaction that has operated on postgres_fdw foreign tables"))); + break; + + case XACT_EVENT_PARALLEL_COMMIT: + case XACT_EVENT_COMMIT: + case XACT_EVENT_PREPARE: + /* Pre-commit should have closed the open transaction */ + elog(ERROR, "missed cleaning up connection during pre-commit"); + break; + + case XACT_EVENT_PARALLEL_ABORT: + case XACT_EVENT_ABORT: + /* Rollback all remote transactions during abort */ + pgfdw_abort_cleanup(true); + break; } /* @@ -974,12 +940,6 @@ static void pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, SubTransactionId parentSubid, void *arg) { - HASH_SEQ_STATUS scan; - ConnCacheEntry *entry; - int curlevel; - List *pending_entries = NIL; - char sql[100]; - /* Nothing to do at subxact start, nor after commit. */ if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB || event == SUBXACT_EVENT_ABORT_SUB)) @@ -989,48 +949,15 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, if (!xact_got_connection) return; - /* - * Scan all connection cache entries to find open remote subtransactions - * of the current level, and close them. - */ - curlevel = GetCurrentTransactionNestLevel(); - snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); - - hash_seq_init(&scan, ConnectionHash); - while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) + if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) { - /* - * We only care about connections with open remote subtransactions of - * the current level. - */ - if (entry->conn == NULL || entry->xact_depth < curlevel) - continue; - - if (entry->xact_depth > curlevel) - elog(ERROR, "missed cleaning up remote subtransaction at level %d", - entry->xact_depth); - - if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) - { - /* Commit all remote subtransactions during pre-commit */ - if (pgfdw_exec_pre_commit(entry, sql, &pending_entries, false)) - continue; - } - else - { - /* Rollback all remote subtransactions during abort */ - pgfdw_abort_cleanup(entry, false); - } - - /* OK, we're outta that level of subtransaction */ - pgfdw_reset_xact_state(entry, false); + /* Commit all remote subtransactions during pre-commit */ + pgfdw_exec_pre_commit(false); } - - /* If there are any pending connections, finish cleaning them up */ - if (pending_entries) + else { - Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB); - pgfdw_finish_pre_commit(pending_entries, sql, false); + /* Rollback all remote subtransactions during abort */ + pgfdw_abort_cleanup(false); } } @@ -1394,131 +1321,210 @@ exit: ; * Set entry->changing_xact_state to false on success, true on failure. */ static void -pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel) +pgfdw_abort_cleanup(bool toplevel) { + HASH_SEQ_STATUS scan; + ConnCacheEntry *entry; char sql[100]; /* - * Don't try to clean up the connection if we're already in error - * recursion trouble. + * Scan all connection cache entries to find open remote transactions, and + * close them. */ - if (in_error_recursion_trouble()) - entry->changing_xact_state = true; + hash_seq_init(&scan, ConnectionHash); + while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) + { + /* Ignore cache entry if no open connection right now */ + if (entry->conn == NULL) + continue; - /* - * If connection is already unsalvageable, don't touch it further. - */ - if (entry->changing_xact_state) - return; + /* Sanity check for subtransaction */ + if (!toplevel) + { + int curlevel = GetCurrentTransactionNestLevel(); - /* - * Mark this connection as in the process of changing transaction state. - */ - entry->changing_xact_state = true; + if (entry->xact_depth < curlevel) + continue; - /* Assume we might have lost track of prepared statements */ - entry->have_error = true; + if (entry->xact_depth > curlevel) + elog(ERROR, "missed cleaning up remote subtransaction at level %d", + entry->xact_depth); + } - /* - * 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 && - !pgfdw_cancel_query(entry->conn)) - return; /* Unable to cancel running query */ + /* If it has an open remote transaction, try to close it */ + if (entry->xact_depth > 0) + { + /* + * 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 (toplevel) - snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); - else - snprintf(sql, sizeof(sql), - "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", - entry->xact_depth, entry->xact_depth); - if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) - return; /* Unable to abort remote (sub)transaction */ + /* + * If connection is already unsalvageable, don't touch it further. + */ + if (entry->changing_xact_state) + goto xact_abort_end; - if (toplevel) - { - if (entry->have_prep_stmt && entry->have_error && - !pgfdw_exec_cleanup_query(entry->conn, - "DEALLOCATE ALL", - true)) - return; /* Trouble clearing prepared statements */ + /* + * Mark this connection as in the process of changing transaction state. + */ + entry->changing_xact_state = true; - entry->have_prep_stmt = false; - entry->have_error = false; - } + /* Assume we might have lost track of prepared statements */ + entry->have_error = true; - /* - * If pendingAreq of the per-connection state is not NULL, it means that - * an asynchronous fetch begun by fetch_more_data_begin() was not done - * successfully and thus the per-connection state was not reset in - * fetch_more_data(); in that case reset the per-connection state here. - */ - if (entry->state.pendingAreq) - memset(&entry->state, 0, sizeof(entry->state)); + /* + * 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 && + !pgfdw_cancel_query(entry->conn)) + goto xact_abort_end; /* Unable to cancel running query */ - /* Disarm changing_xact_state if it all worked */ - entry->changing_xact_state = false; + if (toplevel) + snprintf(sql, sizeof(sql), "ABORT TRANSACTION"); + else + snprintf(sql, sizeof(sql), + "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", + entry->xact_depth, entry->xact_depth); + if (!pgfdw_exec_cleanup_query(entry->conn, sql, false)) + goto xact_abort_end; /* Unable to abort remote (sub)transaction */ + + if (toplevel) + { + if (entry->have_prep_stmt && entry->have_error && + !pgfdw_exec_cleanup_query(entry->conn, + "DEALLOCATE ALL", + true)) + goto xact_abort_end; /* Trouble clearing prepared statements */ + + entry->have_prep_stmt = false; + entry->have_error = false; + } + + /* + * If pendingAreq of the per-connection state is not NULL, it means that + * an asynchronous fetch begun by fetch_more_data_begin() was not done + * successfully and thus the per-connection state was not reset in + * fetch_more_data(); in that case reset the per-connection state here. + */ + if (entry->state.pendingAreq) + memset(&entry->state, 0, sizeof(entry->state)); + + /* Disarm changing_xact_state if it all worked */ + entry->changing_xact_state = false; + } + +xact_abort_end: + /* Reset state to show we're out of a transaction */ + pgfdw_reset_xact_state(entry, toplevel); + } } /* * Commit all remote transactions or subtransactions during pre-commit. * - * If parallel_commit is enabled at this connection cache entry and - * the result of "sql" needs to be gotten later, return true and append - * this entry to "pending_entries". - * * "toplevel" should be set to true if toplevel (main) transaction is * committed, false otherwise. */ -static bool -pgfdw_exec_pre_commit(ConnCacheEntry *entry, const char *sql, - List **pending_entries, bool toplevel) +static void +pgfdw_exec_pre_commit(bool toplevel) { - PGresult *res; - - /* - * If abort cleanup previously failed for this connection, we can't issue - * any more commands against it. - */ - pgfdw_reject_incomplete_xact_state_change(entry); + HASH_SEQ_STATUS scan; + ConnCacheEntry *entry; + List *pending_entries = NIL; + int curlevel = 0; + char sql[100]; - entry->changing_xact_state = true; - if (entry->parallel_commit) + /* Form SQL query */ + if (toplevel) { - do_sql_command_begin(entry->conn, sql); - *pending_entries = lappend(*pending_entries, entry); - return true; + snprintf(sql, sizeof(sql), "COMMIT TRANSACTION"); + } + else + { + curlevel = GetCurrentTransactionNestLevel(); + snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); } - do_sql_command(entry->conn, sql); - entry->changing_xact_state = false; - - if (!toplevel) - return false; /* - * If there were any errors in subtransactions, and we made prepared - * statements, do a DEALLOCATE ALL to make sure we get rid of all prepared - * statements. This is annoying and not terribly bulletproof, but it's - * probably not worth trying harder. - * - * DEALLOCATE ALL only exists in 8.3 and later, so this constrains how old - * a server postgres_fdw can communicate with. We intentionally ignore - * errors in the DEALLOCATE, so that we can hobble along to some extent - * with older servers (leaking prepared statements as we go; but we don't - * really support update operations pre-8.3 anyway). + * Scan all connection cache entries to find open remote transactions, and + * close them. */ - if (entry->have_prep_stmt && entry->have_error) + hash_seq_init(&scan, ConnectionHash); + while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) { - res = PQexec(entry->conn, "DEALLOCATE ALL"); - PQclear(res); + /* Ignore cache entry if no open connection right now */ + if (entry->conn == NULL) + continue; + + /* Sanity check for subtransaction */ + if (!toplevel) + { + if (entry->xact_depth < curlevel) + continue; + + if (entry->xact_depth > curlevel) + elog(ERROR, "missed cleaning up remote subtransaction at level %d", + entry->xact_depth); + } + + /* If it has an open remote transaction, try to close it */ + if (entry->xact_depth > 0) + { + /* + * If abort cleanup previously failed for this connection, we can't issue + * any more commands against it. + */ + pgfdw_reject_incomplete_xact_state_change(entry); + + 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; + + if (toplevel) + { + /* + * If there were any errors in subtransactions, and we made prepared + * statements, do a DEALLOCATE ALL to make sure we get rid of all prepared + * statements. This is annoying and not terribly bulletproof, but it's + * probably not worth trying harder. + * + * DEALLOCATE ALL only exists in 8.3 and later, so this constrains how old + * a server postgres_fdw can communicate with. We intentionally ignore + * errors in the DEALLOCATE, so that we can hobble along to some extent + * with older servers (leaking prepared statements as we go; but we don't + * really support update operations pre-8.3 anyway). + */ + if (entry->have_prep_stmt && entry->have_error) + { + PGresult *res; + + res = PQexec(entry->conn, "DEALLOCATE ALL"); + PQclear(res); + } + entry->have_prep_stmt = false; + entry->have_error = false; + } + } + + /* Reset state to show we're out of a transaction */ + pgfdw_reset_xact_state(entry, toplevel); } - entry->have_prep_stmt = false; - entry->have_error = false; - return false; + /* If there are any pending connections, finish cleaning them up */ + if (pending_entries) + pgfdw_finish_pre_commit(pending_entries, sql, toplevel); } /* -- 2.18.0 [application/x-patch] v2-0001-Refactor-pgfdw_get_result-and-pgfdw_get_cleanup_r.patch (6.7K, ../../CAAJ_b97PdG0ArsaUpezDpbFbJhP6FPyYTQPpRyJFDLoPwQKQBg@mail.gmail.com/5-v2-0001-Refactor-pgfdw_get_result-and-pgfdw_get_cleanup_r.patch) download | inline diff: From 79068f8f60c04612bd4a086e001a6b0aa11daa12 Mon Sep 17 00:00:00 2001 From: Fujii Masao <[email protected]> Date: Mon, 25 Jul 2022 17:25:24 +0900 Subject: [PATCH v2 1/4] Refactor pgfdw_get_result() and pgfdw_get_cleanup_result(). --- contrib/postgres_fdw/connection.c | 125 +++++++++++------------------- 1 file changed, 47 insertions(+), 78 deletions(-) diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 939d114f02e..cbee2854803 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -108,8 +108,8 @@ 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); -static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, - PGresult **result, bool *timed_out); +static bool pgfdw_get_result_timed(PGconn *conn, TimestampTz endtime, + PGresult **result, bool *timed_out); 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, @@ -799,53 +799,12 @@ pgfdw_exec_query(PGconn *conn, const char *query, PgFdwConnState *state) PGresult * pgfdw_get_result(PGconn *conn, const char *query) { - PGresult *volatile last_res = NULL; - - /* In what follows, do not leak any PGresults on an error. */ - PG_TRY(); - { - for (;;) - { - PGresult *res; - - while (PQisBusy(conn)) - { - int wc; - - /* Sleep until there's something to do */ - wc = WaitLatchOrSocket(MyLatch, - WL_LATCH_SET | WL_SOCKET_READABLE | - WL_EXIT_ON_PM_DEATH, - PQsocket(conn), - -1L, PG_WAIT_EXTENSION); - ResetLatch(MyLatch); - - CHECK_FOR_INTERRUPTS(); - - /* Data available in socket? */ - if (wc & WL_SOCKET_READABLE) - { - if (!PQconsumeInput(conn)) - pgfdw_report_error(ERROR, NULL, conn, false, query); - } - } - - res = PQgetResult(conn); - if (res == NULL) - break; /* query is complete */ + PGresult *result = NULL; - PQclear(last_res); - last_res = res; - } - } - PG_CATCH(); - { - PQclear(last_res); - PG_RE_THROW(); - } - PG_END_TRY(); + if (pgfdw_get_result_timed(conn, 0, &result, NULL)) + pgfdw_report_error(ERROR, NULL, conn, false, query); - return last_res; + return result; } /* @@ -1295,7 +1254,7 @@ pgfdw_cancel_query(PGconn *conn) } /* Get and discard the result of the query. */ - if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) + if (pgfdw_get_result_timed(conn, endtime, &result, &timed_out)) { if (timed_out) ereport(WARNING, @@ -1351,7 +1310,7 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) } /* Get the result of the query. */ - if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out)) + if (pgfdw_get_result_timed(conn, endtime, &result, &timed_out)) { if (timed_out) ereport(WARNING, @@ -1375,24 +1334,33 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors) } /* - * Get, during abort cleanup, the result of a query that is in progress. This - * might be a query that is being interrupted by transaction abort, or it might - * be a query that was initiated as part of transaction abort to get the remote - * side back to the appropriate state. + * Get the result of a query. + * + * This function offers quick responsiveness by checking for any interruptions. + * + * If timed_out is NULL, the timeout does not occur. Otherwise, the timeout is + * enabled and endtime is used as the time at which this function should + * give up and assume the remote side is dead. + * + * Return true if the timeout expired or connection trouble occurred. Otherwise + * return false and set *result to the last result of a query. Set timed_out to + * true only when the timeout expired. + * + * This function emulates PQexec()'s behavior of returning the last result + * when there are many. + * + * Caller is responsible for the error handling on the result. * - * endtime is the time at which we should give up and assume the remote - * side is dead. Returns true if the timeout expired or connection trouble - * occurred, false otherwise. Sets *result except in case of a timeout. - * Sets timed_out to true only when the timeout expired. */ static bool -pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, - bool *timed_out) +pgfdw_get_result_timed(PGconn *conn, TimestampTz endtime, PGresult **result, + bool *timed_out) { volatile bool failed = false; PGresult *volatile last_res = NULL; - *timed_out = false; + if (timed_out != NULL) + *timed_out = false; /* In what follows, do not leak any PGresults on an error. */ PG_TRY(); @@ -1404,23 +1372,27 @@ pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime, PGresult **result, while (PQisBusy(conn)) { int wc; - TimestampTz now = GetCurrentTimestamp(); - long cur_timeout; + long cur_timeout = -1; + int wakeEvents = WL_LATCH_SET | WL_SOCKET_READABLE | + WL_EXIT_ON_PM_DEATH; /* If timeout has expired, give up, else get sleep time. */ - cur_timeout = TimestampDifferenceMilliseconds(now, endtime); - if (cur_timeout <= 0) + if (timed_out != NULL) { - *timed_out = true; - failed = true; - goto exit; + TimestampTz now = GetCurrentTimestamp(); + + cur_timeout = TimestampDifferenceMilliseconds(now, endtime); + if (cur_timeout <= 0) + { + *timed_out = true; + failed = true; + goto exit; + } + wakeEvents |= WL_TIMEOUT; } /* Sleep until there's something to do */ - wc = WaitLatchOrSocket(MyLatch, - WL_LATCH_SET | WL_SOCKET_READABLE | - WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, - PQsocket(conn), + wc = WaitLatchOrSocket(MyLatch, wakeEvents, PQsocket(conn), cur_timeout, PG_WAIT_EXTENSION); ResetLatch(MyLatch); @@ -1458,6 +1430,7 @@ exit: ; PQclear(last_res); else *result = last_res; + return failed; } @@ -1599,13 +1572,9 @@ pgfdw_finish_pre_commit_cleanup(List *pending_entries) 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; - } + pgfdw_get_result_timed(entry->conn, 0, &res, NULL); + PQclear(res); + entry->have_prep_stmt = false; entry->have_error = false; -- 2.18.0 ^ permalink raw reply [nested|flat] 3+ messages in thread
* [PATCH v51 01/10] Make index_concurrently_create_copy more general @ 2026-03-24 18:02 Álvaro Herrera <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Álvaro Herrera @ 2026-03-24 18:02 UTC (permalink / raw) Add a 'boolean concurrent' option, and make it work for both cases. Also rename it to index_create_copy. This allows it to be reused for other purposes -- specifically, for REPACK CONCURRENTLY. With the CONCURRENTLY option, REPACK cannot simply swap the heap file and rebuild its indexes. Instead, it needs to build a separate set of indexes (including system catalog entries) *before* the actual swap, to reduce the time AccessExclusiveLock needs to be held for. This approach is different from what CREATE INDEX CONCURRENTLY does. Per a suggestion from Mihail Nikalayeu. Author: Antonin Houska <[email protected]> Discussion: https://postgr.es/m/41104.1754922120@localhost --- src/backend/catalog/index.c | 41 ++++++++++++++++++++++---------- src/backend/commands/indexcmds.c | 9 +++---- src/include/catalog/index.h | 7 +++--- 3 files changed, 36 insertions(+), 21 deletions(-) diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 1ccfa687f05..e418d67e8e4 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -1289,17 +1289,17 @@ index_create(Relation heapRelation, } /* - * index_concurrently_create_copy + * index_create_copy * - * Create concurrently an index based on the definition of the one provided by - * caller. The index is inserted into catalogs and needs to be built later - * on. This is called during concurrent reindex processing. + * Create an index based on the definition of the one provided by caller. The + * index is inserted into catalogs. If 'concurrently' is TRUE, it needs to be + * built later on; otherwise it's built immediately. * * "tablespaceOid" is the tablespace to use for this index. */ Oid -index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, - Oid tablespaceOid, const char *newName) +index_create_copy(Relation heapRelation, bool concurrently, + Oid oldIndexId, Oid tablespaceOid, const char *newName) { Relation indexRelation; IndexInfo *oldInfo, @@ -1318,6 +1318,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, List *indexColNames = NIL; List *indexExprs = NIL; List *indexPreds = NIL; + int flags = 0; indexRelation = index_open(oldIndexId, RowExclusiveLock); @@ -1328,7 +1329,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, * Concurrent build of an index with exclusion constraints is not * supported. */ - if (oldInfo->ii_ExclusionOps != NULL) + if (oldInfo->ii_ExclusionOps != NULL && concurrently) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("concurrent index creation for exclusion constraints is not supported"))); @@ -1384,9 +1385,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, } /* - * Build the index information for the new index. Note that rebuild of - * indexes with exclusion constraints is not supported, hence there is no - * need to fill all the ii_Exclusion* fields. + * Build the index information for the new index. */ newInfo = makeIndexInfo(oldInfo->ii_NumIndexAttrs, oldInfo->ii_NumIndexKeyAttrs, @@ -1395,11 +1394,24 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, indexPreds, oldInfo->ii_Unique, oldInfo->ii_NullsNotDistinct, - false, /* not ready for inserts */ - true, + !concurrently, /* isready */ + concurrently, /* concurrent */ indexRelation->rd_indam->amsummarizing, oldInfo->ii_WithoutOverlaps); + /* fetch exclusion constraint info if any */ + if (indexRelation->rd_index->indisexclusion) + { + /* + * XXX Beware: we're making newInfo point to oldInfo-owned memory. It + * would be more orthodox to palloc+memcpy, but we don't need that + * here at present. + */ + newInfo->ii_ExclusionOps = oldInfo->ii_ExclusionOps; + newInfo->ii_ExclusionProcs = oldInfo->ii_ExclusionProcs; + newInfo->ii_ExclusionStrats = oldInfo->ii_ExclusionStrats; + } + /* * Extract the list of column names and the column numbers for the new * index information. All this information will be used for the index @@ -1436,6 +1448,9 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, stattargets[i].isnull = isnull; } + if (concurrently) + flags = INDEX_CREATE_SKIP_BUILD | INDEX_CREATE_CONCURRENT; + /* * Now create the new index. * @@ -1459,7 +1474,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, indcoloptions->values, stattargets, reloptionsDatum, - INDEX_CREATE_SKIP_BUILD | INDEX_CREATE_CONCURRENT, + flags, 0, true, /* allow table to be a system catalog? */ false, /* is_internal? */ diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c index 373e8234794..cba379810c7 100644 --- a/src/backend/commands/indexcmds.c +++ b/src/backend/commands/indexcmds.c @@ -3989,10 +3989,11 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein tablespaceid = indexRel->rd_rel->reltablespace; /* Create new index definition based on given index */ - newIndexId = index_concurrently_create_copy(heapRel, - idx->indexId, - tablespaceid, - concurrentName); + newIndexId = index_create_copy(heapRel, + true, + idx->indexId, + tablespaceid, + concurrentName); /* * Now open the relation of the new index, a session-level lock is diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index a38e95bc0eb..ed9e4c37d27 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -101,10 +101,9 @@ extern Oid index_create(Relation heapRelation, #define INDEX_CONSTR_CREATE_REMOVE_OLD_DEPS (1 << 4) #define INDEX_CONSTR_CREATE_WITHOUT_OVERLAPS (1 << 5) -extern Oid index_concurrently_create_copy(Relation heapRelation, - Oid oldIndexId, - Oid tablespaceOid, - const char *newName); +extern Oid index_create_copy(Relation heapRelation, bool concurrently, + Oid oldIndexId, Oid tablespaceOid, + const char *newName); extern void index_concurrently_build(Oid heapRelationId, Oid indexRelationId); -- 2.47.3 --r2slln3zpmilwu22 Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v51-0002-Rename-cluster.c-h-repack.c-h.patch" ^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2026-03-24 18:02 UTC | newest] Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2019-05-28 18:16 [PATCH v27 2/5] Add pg_depend.refobjversion. Thomas Munro <[email protected]> 2022-08-04 04:11 Re: Refactoring postgres_fdw/connection.c Amul Sul <[email protected]> 2026-03-24 18:02 [PATCH v51 01/10] Make index_concurrently_create_copy more general Álvaro Herrera <[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