From: Nathan Bossart <nathandbossart@gmail.com>
To: Robert Haas <robertmhaas@gmail.com>
Cc: Jeff Davis <pgsql@j-davis.com>
Cc: pgsql-hackers@postgresql.org
Subject: Re: optimizing pg_upgrade's once-in-each-database steps
Date: Mon, 8 Jul 2024 10:22:08 -0500
Message-ID: <ZowEIGd0lLKh16qe@nathan> (raw)
In-Reply-To: <ZoMK9ZhOsWEKqxlM@nathan>
References: <20240516211638.GA1688936@nathanxps13>
<3b97ab8e295ec3db22baa2dfdd49073449ae1696.camel@j-davis.com>
<20240517012408.GA1691322@nathanxps13>
<ZoMHsBasBy8LqkEU@nathan>
<CA+TgmoaBof36-ZuRWtu=+8rnhGTNQFFGNcmVB0x3aQawfc_JxA@mail.gmail.com>
<ZoMK9ZhOsWEKqxlM@nathan>
As I mentioned elsewhere [0], here's a first attempt at parallelizing the
data type checks. I was worried that I might have to refactor Daniel's
work in commit 347758b quite significantly, but I was able to avoid that by
using a set of generic callbacks and providing each task step an index to
the data_types_usage_checks array.
[0] https://postgr.es/m/Zov5kHbEyDMuHJI_%40nathan
--
nathan
Attachments:download all (7 files, .tar.gz)v3-0001-introduce-framework-for-parallelizing-pg_upgrade-.patch (text/plain, 10.0K) download
From 0d3353c318ca4dcd582d67ca25c47b9e3cf24921 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathan@postgresql.org>
Date: Fri, 28 Jun 2024 11:02:44 -0500
Subject: [PATCH v3 1/7] introduce framework for parallelizing pg_upgrade tasks
---
src/bin/pg_upgrade/Makefile | 1 +
src/bin/pg_upgrade/async.c | 323 +++++++++++++++++++++++++++++++
src/bin/pg_upgrade/meson.build | 1 +
src/bin/pg_upgrade/pg_upgrade.h | 16 ++
src/tools/pgindent/typedefs.list | 4 +
5 files changed, 345 insertions(+)
create mode 100644 src/bin/pg_upgrade/async.c
diff --git a/src/bin/pg_upgrade/Makefile b/src/bin/pg_upgrade/Makefileindex bde91e2beb..3bc4f5d740 100644--- a/src/bin/pg_upgrade/Makefile+++ b/src/bin/pg_upgrade/Makefile@@ -12,6 +12,7 @@ include $(top_builddir)/src/Makefile.global
OBJS = \
$(WIN32RES) \
+ async.o \
check.o \
controldata.o \
dump.o \
diff --git a/src/bin/pg_upgrade/async.c b/src/bin/pg_upgrade/async.cnew file mode 100644index 0000000000..7df1e7712d--- /dev/null+++ b/src/bin/pg_upgrade/async.c@@ -0,0 +1,323 @@+/*+ * async.c+ *+ * parallelization via libpq's async APIs+ *+ * Copyright (c) 2024, PostgreSQL Global Development Group+ * src/bin/pg_upgrade/async.c+ */++#include "postgres_fe.h"++#include "common/connect.h"+#include "fe_utils/string_utils.h"+#include "pg_upgrade.h"++static int dbs_complete;+static int dbs_processing;++typedef struct AsyncTaskCallbacks+{+ AsyncTaskGetQueryCB query_cb;+ AsyncTaskProcessCB process_cb;+ bool free_result;+ void *arg;+} AsyncTaskCallbacks;++typedef struct AsyncTask+{+ AsyncTaskCallbacks *cbs;+ int num_cb_sets;+} AsyncTask;++typedef enum+{+ FREE,+ CONNECTING,+ SETTING_SEARCH_PATH,+ RUNNING_QUERY,+} AsyncSlotState;++typedef struct+{+ AsyncSlotState state;+ int db;+ int query;+ PGconn *conn;+} AsyncSlot;++AsyncTask *+async_task_create(void)+{+ return pg_malloc0(sizeof(AsyncTask));+}++void+async_task_free(AsyncTask *task)+{+ if (task->cbs)+ pg_free(task->cbs);++ pg_free(task);+}++void+async_task_add_step(AsyncTask *task,+ AsyncTaskGetQueryCB query_cb,+ AsyncTaskProcessCB process_cb, bool free_result,+ void *arg)+{+ AsyncTaskCallbacks *new_cbs;++ task->cbs = pg_realloc(task->cbs,+ ++task->num_cb_sets * sizeof(AsyncTaskCallbacks));++ new_cbs = &task->cbs[task->num_cb_sets - 1];+ new_cbs->query_cb = query_cb;+ new_cbs->process_cb = process_cb;+ new_cbs->free_result = free_result;+ new_cbs->arg = arg;+}++static void+conn_failure(PGconn *conn)+{+ pg_log(PG_REPORT, "%s", PQerrorMessage(conn));+ printf(_("Failure, exiting\n"));+ exit(1);+}++static void+start_conn(const ClusterInfo *cluster, AsyncSlot *slot)+{+ PQExpBufferData conn_opts;++ /* Build connection string with proper quoting */+ initPQExpBuffer(&conn_opts);+ appendPQExpBufferStr(&conn_opts, "dbname=");+ appendConnStrVal(&conn_opts, cluster->dbarr.dbs[slot->db].db_name);+ appendPQExpBufferStr(&conn_opts, " user=");+ appendConnStrVal(&conn_opts, os_info.user);+ appendPQExpBuffer(&conn_opts, " port=%d", cluster->port);+ if (cluster->sockdir)+ {+ appendPQExpBufferStr(&conn_opts, " host=");+ appendConnStrVal(&conn_opts, cluster->sockdir);+ }++ slot->conn = PQconnectStart(conn_opts.data);+ termPQExpBuffer(&conn_opts);++ if (!slot->conn)+ conn_failure(slot->conn);+}++static void+dispatch_query(const ClusterInfo *cluster, AsyncSlot *slot,+ const AsyncTask *task)+{+ AsyncTaskCallbacks *cbs = &task->cbs[slot->query];+ AsyncTaskGetQueryCB get_query = cbs->query_cb;+ DbInfo *dbinfo = &cluster->dbarr.dbs[slot->db];+ char *query = (*get_query) (dbinfo, cbs->arg);++ if (!PQsendQuery(slot->conn, query))+ conn_failure(slot->conn);++ pg_free(query);+}++static PGresult *+get_last_result(PGconn *conn)+{+ PGresult *tmp;+ PGresult *res = NULL;++ while ((tmp = PQgetResult(conn)) != NULL)+ {+ PQclear(res);+ res = tmp;+ if (PQstatus(conn) == CONNECTION_BAD)+ conn_failure(conn);+ }++ if (PQresultStatus(res) != PGRES_COMMAND_OK &&+ PQresultStatus(res) != PGRES_TUPLES_OK)+ conn_failure(conn);++ return res;+}++static void+process_query_result(const ClusterInfo *cluster, AsyncSlot *slot,+ const AsyncTask *task)+{+ AsyncTaskCallbacks *cbs = &task->cbs[slot->query];+ AsyncTaskProcessCB process_cb = cbs->process_cb;+ DbInfo *dbinfo = &cluster->dbarr.dbs[slot->db];+ PGresult *res = get_last_result(slot->conn);++ (*process_cb) (dbinfo, res, cbs->arg);++ if (cbs->free_result)+ PQclear(res);+}++static void+process_slot(const ClusterInfo *cluster, AsyncSlot *slot, const AsyncTask *task)+{+ switch (slot->state)+ {+ case FREE:+ if (dbs_processing >= cluster->dbarr.ndbs)+ return;+ slot->db = dbs_processing++;+ slot->state = CONNECTING;+ start_conn(cluster, slot);+ return;++ case CONNECTING:+ if (PQconnectPoll(slot->conn) == PGRES_POLLING_FAILED)+ conn_failure(slot->conn);+ if (PQconnectPoll(slot->conn) != PGRES_POLLING_OK)+ return;+ slot->state = SETTING_SEARCH_PATH;+ if (!PQsendQuery(slot->conn, ALWAYS_SECURE_SEARCH_PATH_SQL))+ conn_failure(slot->conn);+ return;++ case SETTING_SEARCH_PATH:+ if (!PQconsumeInput(slot->conn))+ conn_failure(slot->conn);+ if (PQisBusy(slot->conn))+ return;+ PQclear(get_last_result(slot->conn));+ slot->state = RUNNING_QUERY;+ dispatch_query(cluster, slot, task);+ return;++ case RUNNING_QUERY:+ if (!PQconsumeInput(slot->conn))+ conn_failure(slot->conn);+ if (PQisBusy(slot->conn))+ return;+ process_query_result(cluster, slot, task);+ if (++slot->query >= task->num_cb_sets)+ {+ dbs_complete++;+ PQfinish(slot->conn);+ memset(slot, 0, sizeof(AsyncSlot));+ return;+ }+ dispatch_query(cluster, slot, task);+ return;+ }+}++/*+ * Wait on the slots to either finish connecting or to receive query results if+ * possible. This avoids a tight loop in async_task_run().+ */+static void+wait_on_slots(AsyncSlot *slots, int numslots)+{+ fd_set input_mask;+ fd_set output_mask;+ fd_set except_mask;+ int maxFd = 0;++ FD_ZERO(&input_mask);+ FD_ZERO(&output_mask);+ FD_ZERO(&except_mask);++ for (int i = 0; i < numslots; i++)+ {+ int sock;+ bool read = false;++ switch (slots[i].state)+ {+ case FREE:++ /*+ * If we see a free slot, return right away so that it can be+ * reused immediately for the next database. This might cause+ * us to spin more than necessary as we finish processing the+ * last few databases, but that shouldn't cause too much harm.+ */+ return;++ case CONNECTING:++ /*+ * If we are waiting for the connection to establish, choose+ * whether to wait for reading or for writing on the socket as+ * appropriate. If neither apply, just return immediately so+ * that we can handle the slot.+ */+ {+ PostgresPollingStatusType status;++ status = PQconnectPoll(slots[i].conn);+ if (status == PGRES_POLLING_READING)+ read = true;+ else if (status != PGRES_POLLING_WRITING)+ return;+ }+ break;++ case SETTING_SEARCH_PATH:+ case RUNNING_QUERY:++ /*+ * If we've sent a query, we must wait for the socket to be+ * read-ready. Note that process_slot() handles calling+ * PQconsumeInput() as required.+ */+ read = true;+ break;+ }++ /*+ * If there's some problem retrieving the socket, just pretend this+ * slot doesn't exist. We don't expect this to happen regularly in+ * practice, so it seems unlikely to cause too much harm.+ */+ sock = PQsocket(slots[i].conn);+ if (sock < 0)+ continue;++ /*+ * Add the socket to the set.+ */+ FD_SET(sock, read ? &input_mask : &output_mask);+ FD_SET(sock, &except_mask);+ maxFd = Max(maxFd, sock);+ }++ /*+ * If we found socket(s) to wait on, wait.+ */+ if (maxFd != 0)+ (void) select(maxFd + 1, &input_mask, &output_mask, &except_mask, NULL);+}++void+async_task_run(const AsyncTask *task, const ClusterInfo *cluster)+{+ int jobs = Max(1, user_opts.jobs);+ AsyncSlot *slots = pg_malloc0(sizeof(AsyncSlot) * jobs);++ dbs_complete = 0;+ dbs_processing = 0;++ while (dbs_complete < cluster->dbarr.ndbs)+ {+ for (int i = 0; i < jobs; i++)+ process_slot(cluster, &slots[i], task);++ wait_on_slots(slots, jobs);+ }++ pg_free(slots);+}diff --git a/src/bin/pg_upgrade/meson.build b/src/bin/pg_upgrade/meson.buildindex 9825fa3305..9eb48e176c 100644--- a/src/bin/pg_upgrade/meson.build+++ b/src/bin/pg_upgrade/meson.build@@ -1,6 +1,7 @@
# Copyright (c) 2022-2024, PostgreSQL Global Development Group
pg_upgrade_sources = files(
+ 'async.c',
'check.c',
'controldata.c',
'dump.c',
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.hindex 8afe240bdf..1ebad3bd74 100644--- a/src/bin/pg_upgrade/pg_upgrade.h+++ b/src/bin/pg_upgrade/pg_upgrade.h@@ -494,3 +494,19 @@ void parallel_transfer_all_new_dbs(DbInfoArr *old_db_arr, DbInfoArr *new_db_arr
char *old_pgdata, char *new_pgdata,
char *old_tablespace);
bool reap_child(bool wait_for_child);
++/* async.c */++typedef char *(*AsyncTaskGetQueryCB) (DbInfo *dbinfo, void *arg);+typedef void (*AsyncTaskProcessCB) (DbInfo *dbinfo, PGresult *res, void *arg);++/* struct definition is private to async.c */+typedef struct AsyncTask AsyncTask;++AsyncTask *async_task_create(void);+void async_task_add_step(AsyncTask *task,+ AsyncTaskGetQueryCB query_cb,+ AsyncTaskProcessCB process_cb, bool free_result,+ void *arg);+void async_task_run(const AsyncTask *task, const ClusterInfo *cluster);+void async_task_free(AsyncTask *task);diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.listindex 9320e4d808..ef0db74d3b 100644--- a/src/tools/pgindent/typedefs.list+++ b/src/tools/pgindent/typedefs.list@@ -153,6 +153,10 @@ ArrayMetaState
ArraySubWorkspace
ArrayToken
ArrayType
+AsyncSlot+AsyncSlotState+AsyncTask+AsyncTaskCallbacks
AsyncQueueControl
AsyncQueueEntry
AsyncRequest
--
2.39.3 (Apple Git-146)
v3-0002-use-new-pg_upgrade-async-API-for-subscription-sta.patch (text/plain, 8.8K) download
From 2b651e53dcde039aec716de05ce70a0a9a69fef6 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathan@postgresql.org>
Date: Fri, 28 Jun 2024 17:21:19 -0500
Subject: [PATCH v3 2/7] use new pg_upgrade async API for subscription state
checks
---
src/bin/pg_upgrade/check.c | 200 ++++++++++++++++++++-----------------
1 file changed, 106 insertions(+), 94 deletions(-)
diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.cindex 27924159d6..f653fa25a5 100644--- a/src/bin/pg_upgrade/check.c+++ b/src/bin/pg_upgrade/check.c@@ -1906,6 +1906,75 @@ check_old_cluster_for_valid_slots(bool live_check)
check_ok();
}
+/* private state for subscription state checks */+struct substate_info+{+ FILE *script;+ char output_path[MAXPGPATH];+};++/*+ * We don't allow upgrade if there is a risk of dangling slot or origin+ * corresponding to initial sync after upgrade.+ *+ * A slot/origin not created yet refers to the 'i' (initialize) state, while+ * 'r' (ready) state refers to a slot/origin created previously but already+ * dropped. These states are supported for pg_upgrade. The other states listed+ * below are not supported:+ *+ * a) SUBREL_STATE_DATASYNC: A relation upgraded while in this state would+ * retain a replication slot, which could not be dropped by the sync worker+ * spawned after the upgrade because the subscription ID used for the slot name+ * won't match anymore.+ *+ * b) SUBREL_STATE_SYNCDONE: A relation upgraded while in this state would+ * retain the replication origin when there is a failure in tablesync worker+ * immediately after dropping the replication slot in the publisher.+ *+ * c) SUBREL_STATE_FINISHEDCOPY: A tablesync worker spawned to work on a+ * relation upgraded while in this state would expect an origin ID with the OID+ * of the subscription used before the upgrade, causing it to fail.+ *+ * d) SUBREL_STATE_SYNCWAIT, SUBREL_STATE_CATCHUP and SUBREL_STATE_UNKNOWN:+ * These states are not stored in the catalog, so we need not allow these+ * states.+ */+static char *+sub_query(DbInfo *dbinfo, void *arg)+{+ return pg_strdup("SELECT r.srsubstate, s.subname, n.nspname, c.relname "+ "FROM pg_catalog.pg_subscription_rel r "+ "LEFT JOIN pg_catalog.pg_subscription s"+ " ON r.srsubid = s.oid "+ "LEFT JOIN pg_catalog.pg_class c"+ " ON r.srrelid = c.oid "+ "LEFT JOIN pg_catalog.pg_namespace n"+ " ON c.relnamespace = n.oid "+ "WHERE r.srsubstate NOT IN ('i', 'r') "+ "ORDER BY s.subname");+}++static void+sub_process(DbInfo *dbinfo, PGresult *res, void *arg)+{+ struct substate_info *state = (struct substate_info *) arg;+ int ntup = PQntuples(res);++ for (int i = 0; i < ntup; i++)+ {+ if (state->script == NULL &&+ (state->script = fopen_priv(state->output_path, "w")) == NULL)+ pg_fatal("could not open file \"%s\": %m", state->output_path);++ fprintf(state->script, "The table sync state \"%s\" is not allowed for database:\"%s\" subscription:\"%s\" schema:\"%s\" relation:\"%s\"\n",+ PQgetvalue(res, i, 0),+ dbinfo->db_name,+ PQgetvalue(res, i, 1),+ PQgetvalue(res, i, 2),+ PQgetvalue(res, i, 3));+ }+}+
/*
* check_old_cluster_subscription_state()
*
@@ -1916,115 +1985,58 @@ check_old_cluster_for_valid_slots(bool live_check)
static void
check_old_cluster_subscription_state(void)
{
- FILE *script = NULL;- char output_path[MAXPGPATH];+ AsyncTask *task = async_task_create();+ struct substate_info state;+ PGresult *res;+ PGconn *conn;
int ntup;
prep_status("Checking for subscription state");
- snprintf(output_path, sizeof(output_path), "%s/%s",+ state.script = NULL;+ snprintf(state.output_path, sizeof(state.output_path), "%s/%s",
log_opts.basedir,
"subs_invalid.txt");
- for (int dbnum = 0; dbnum < old_cluster.dbarr.ndbs; dbnum++)- {- PGresult *res;- DbInfo *active_db = &old_cluster.dbarr.dbs[dbnum];- PGconn *conn = connectToServer(&old_cluster, active_db->db_name);- /* We need to check for pg_replication_origin only once. */- if (dbnum == 0)- {- /*- * Check that all the subscriptions have their respective- * replication origin.- */- res = executeQueryOrDie(conn,- "SELECT d.datname, s.subname "- "FROM pg_catalog.pg_subscription s "- "LEFT OUTER JOIN pg_catalog.pg_replication_origin o "- " ON o.roname = 'pg_' || s.oid "- "INNER JOIN pg_catalog.pg_database d "- " ON d.oid = s.subdbid "- "WHERE o.roname IS NULL;");-- ntup = PQntuples(res);- for (int i = 0; i < ntup; i++)- {- if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)- pg_fatal("could not open file \"%s\": %m", output_path);- fprintf(script, "The replication origin is missing for database:\"%s\" subscription:\"%s\"\n",- PQgetvalue(res, i, 0),- PQgetvalue(res, i, 1));- }- PQclear(res);- }-- /*- * We don't allow upgrade if there is a risk of dangling slot or- * origin corresponding to initial sync after upgrade.- *- * A slot/origin not created yet refers to the 'i' (initialize) state,- * while 'r' (ready) state refers to a slot/origin created previously- * but already dropped. These states are supported for pg_upgrade. The- * other states listed below are not supported:- *- * a) SUBREL_STATE_DATASYNC: A relation upgraded while in this state- * would retain a replication slot, which could not be dropped by the- * sync worker spawned after the upgrade because the subscription ID- * used for the slot name won't match anymore.- *- * b) SUBREL_STATE_SYNCDONE: A relation upgraded while in this state- * would retain the replication origin when there is a failure in- * tablesync worker immediately after dropping the replication slot in- * the publisher.- *- * c) SUBREL_STATE_FINISHEDCOPY: A tablesync worker spawned to work on- * a relation upgraded while in this state would expect an origin ID- * with the OID of the subscription used before the upgrade, causing- * it to fail.- *- * d) SUBREL_STATE_SYNCWAIT, SUBREL_STATE_CATCHUP and- * SUBREL_STATE_UNKNOWN: These states are not stored in the catalog,- * so we need not allow these states.- */- res = executeQueryOrDie(conn,- "SELECT r.srsubstate, s.subname, n.nspname, c.relname "- "FROM pg_catalog.pg_subscription_rel r "- "LEFT JOIN pg_catalog.pg_subscription s"- " ON r.srsubid = s.oid "- "LEFT JOIN pg_catalog.pg_class c"- " ON r.srrelid = c.oid "- "LEFT JOIN pg_catalog.pg_namespace n"- " ON c.relnamespace = n.oid "- "WHERE r.srsubstate NOT IN ('i', 'r') "- "ORDER BY s.subname");-- ntup = PQntuples(res);- for (int i = 0; i < ntup; i++)- {- if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)- pg_fatal("could not open file \"%s\": %m", output_path);+ /*+ * Check that all the subscriptions have their respective replication+ * origin. This check only needs to run once.+ */+ conn = connectToServer(&old_cluster, old_cluster.dbarr.dbs[0].db_name);+ res = executeQueryOrDie(conn,+ "SELECT d.datname, s.subname "+ "FROM pg_catalog.pg_subscription s "+ "LEFT OUTER JOIN pg_catalog.pg_replication_origin o "+ " ON o.roname = 'pg_' || s.oid "+ "INNER JOIN pg_catalog.pg_database d "+ " ON d.oid = s.subdbid "+ "WHERE o.roname IS NULL;");+ ntup = PQntuples(res);+ for (int i = 0; i < ntup; i++)+ {+ if (state.script == NULL &&+ (state.script = fopen_priv(state.output_path, "w")) == NULL)+ pg_fatal("could not open file \"%s\": %m", state.output_path);+ fprintf(state.script, "The replication origin is missing for database:\"%s\" subscription:\"%s\"\n",+ PQgetvalue(res, i, 0),+ PQgetvalue(res, i, 1));+ }+ PQclear(res);+ PQfinish(conn);- fprintf(script, "The table sync state \"%s\" is not allowed for database:\"%s\" subscription:\"%s\" schema:\"%s\" relation:\"%s\"\n",- PQgetvalue(res, i, 0),- active_db->db_name,- PQgetvalue(res, i, 1),- PQgetvalue(res, i, 2),- PQgetvalue(res, i, 3));- }+ async_task_add_step(task, sub_query, sub_process, true, &state);- PQclear(res);- PQfinish(conn);- }+ async_task_run(task, &old_cluster);+ async_task_free(task);- if (script)+ if (state.script)
{
- fclose(script);+ fclose(state.script);
pg_log(PG_REPORT, "fatal");
pg_fatal("Your installation contains subscriptions without origin or having relations not in i (initialize) or r (ready) state.\n"
"You can allow the initial sync to finish for all relations and then restart the upgrade.\n"
"A list of the problematic subscriptions is in the file:\n"
- " %s", output_path);+ " %s", state.output_path);
}
else
check_ok();
--
2.39.3 (Apple Git-146)
v3-0003-move-live_check-variable-to-user_opts.patch (text/plain, 11.8K) download
From 753f3399de8eabd38b79e61f6e1880547d1121c0 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathan@postgresql.org>
Date: Tue, 21 May 2024 16:35:19 -0500
Subject: [PATCH v3 3/7] move live_check variable to user_opts
---
src/bin/pg_upgrade/check.c | 32 ++++++++++++++++----------------
src/bin/pg_upgrade/controldata.c | 5 +++--
src/bin/pg_upgrade/info.c | 12 +++++-------
src/bin/pg_upgrade/option.c | 4 ++--
src/bin/pg_upgrade/pg_upgrade.c | 21 ++++++++++-----------
src/bin/pg_upgrade/pg_upgrade.h | 13 +++++++------
6 files changed, 43 insertions(+), 44 deletions(-)
diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.cindex f653fa25a5..251f3d9017 100644--- a/src/bin/pg_upgrade/check.c+++ b/src/bin/pg_upgrade/check.c@@ -29,7 +29,7 @@ static void check_for_new_tablespace_dir(void);
static void check_for_user_defined_encoding_conversions(ClusterInfo *cluster);
static void check_new_cluster_logical_replication_slots(void);
static void check_new_cluster_subscription_configuration(void);
-static void check_old_cluster_for_valid_slots(bool live_check);+static void check_old_cluster_for_valid_slots(void);
static void check_old_cluster_subscription_state(void);
/*
@@ -555,9 +555,9 @@ fix_path_separator(char *path)
}
void
-output_check_banner(bool live_check)+output_check_banner(void)
{
- if (user_opts.check && live_check)+ if (user_opts.live_check)
{
pg_log(PG_REPORT,
"Performing Consistency Checks on Old Live Server\n"
@@ -573,18 +573,18 @@ output_check_banner(bool live_check)
void
-check_and_dump_old_cluster(bool live_check)+check_and_dump_old_cluster(void)
{
/* -- OLD -- */
- if (!live_check)+ if (!user_opts.live_check)
start_postmaster(&old_cluster, true);
/*
* Extract a list of databases, tables, and logical replication slots from
* the old cluster.
*/
- get_db_rel_and_slot_infos(&old_cluster, live_check);+ get_db_rel_and_slot_infos(&old_cluster);
init_tablespaces();
@@ -605,7 +605,7 @@ check_and_dump_old_cluster(bool live_check)
* Logical replication slots can be migrated since PG17. See comments
* atop get_old_cluster_logical_slot_infos().
*/
- check_old_cluster_for_valid_slots(live_check);+ check_old_cluster_for_valid_slots();
/*
* Subscriptions and their dependencies can be migrated since PG17.
@@ -652,7 +652,7 @@ check_and_dump_old_cluster(bool live_check)
*/
if (GET_MAJOR_VERSION(old_cluster.major_version) <= 906)
{
- if (user_opts.check)+ if (user_opts.live_check)
old_9_6_invalidate_hash_indexes(&old_cluster, true);
}
@@ -667,7 +667,7 @@ check_and_dump_old_cluster(bool live_check)
if (!user_opts.check)
generate_old_dump();
- if (!live_check)+ if (!user_opts.live_check)
stop_postmaster(false);
}
@@ -675,7 +675,7 @@ check_and_dump_old_cluster(bool live_check)
void
check_new_cluster(void)
{
- get_db_rel_and_slot_infos(&new_cluster, false);+ get_db_rel_and_slot_infos(&new_cluster);
check_new_cluster_is_empty();
@@ -826,14 +826,14 @@ check_cluster_versions(void)
void
-check_cluster_compatibility(bool live_check)+check_cluster_compatibility(void)
{
/* get/check pg_control data of servers */
- get_control_data(&old_cluster, live_check);- get_control_data(&new_cluster, false);+ get_control_data(&old_cluster);+ get_control_data(&new_cluster);
check_control_data(&old_cluster.controldata, &new_cluster.controldata);
- if (live_check && old_cluster.port == new_cluster.port)+ if (user_opts.live_check && old_cluster.port == new_cluster.port)
pg_fatal("When checking a live server, "
"the old and new port numbers must be different.");
}
@@ -1839,7 +1839,7 @@ check_new_cluster_subscription_configuration(void)
* before shutdown.
*/
static void
-check_old_cluster_for_valid_slots(bool live_check)+check_old_cluster_for_valid_slots(void)
{
char output_path[MAXPGPATH];
FILE *script = NULL;
@@ -1878,7 +1878,7 @@ check_old_cluster_for_valid_slots(bool live_check)
* Note: This can be satisfied only when the old cluster has been
* shut down, so we skip this for live checks.
*/
- if (!live_check && !slot->caught_up)+ if (!user_opts.live_check && !slot->caught_up)
{
if (script == NULL &&
(script = fopen_priv(output_path, "w")) == NULL)
diff --git a/src/bin/pg_upgrade/controldata.c b/src/bin/pg_upgrade/controldata.cindex 1f0ccea3ed..cf665b9dee 100644--- a/src/bin/pg_upgrade/controldata.c+++ b/src/bin/pg_upgrade/controldata.c@@ -33,7 +33,7 @@
* return valid xid data for a running server.
*/
void
-get_control_data(ClusterInfo *cluster, bool live_check)+get_control_data(ClusterInfo *cluster)
{
char cmd[MAXPGPATH];
char bufin[MAX_STRING];
@@ -76,6 +76,7 @@ get_control_data(ClusterInfo *cluster, bool live_check)
uint32 segno = 0;
char *resetwal_bin;
int rc;
+ bool live_check = (cluster == &old_cluster && user_opts.live_check);
/*
* Because we test the pg_resetwal output as strings, it has to be in
@@ -118,7 +119,7 @@ get_control_data(ClusterInfo *cluster, bool live_check)
/*
* Check for clean shutdown
*/
- if (!live_check || cluster == &new_cluster)+ if (!live_check)
{
/* only pg_controldata outputs the cluster state */
snprintf(cmd, sizeof(cmd), "\"%s/pg_controldata\" \"%s\"",
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.cindex 95c22a7200..8f1777de59 100644--- a/src/bin/pg_upgrade/info.c+++ b/src/bin/pg_upgrade/info.c@@ -27,7 +27,7 @@ static void free_rel_infos(RelInfoArr *rel_arr);
static void print_db_infos(DbInfoArr *db_arr);
static void print_rel_infos(RelInfoArr *rel_arr);
static void print_slot_infos(LogicalSlotInfoArr *slot_arr);
-static void get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check);+static void get_old_cluster_logical_slot_infos(DbInfo *dbinfo);
static void get_db_subscription_count(DbInfo *dbinfo);
@@ -273,11 +273,9 @@ report_unmatched_relation(const RelInfo *rel, const DbInfo *db, bool is_new_db)
*
* higher level routine to generate dbinfos for the database running
* on the given "port". Assumes that server is already running.
- *- * live_check would be used only when the target is the old cluster.
*/
void
-get_db_rel_and_slot_infos(ClusterInfo *cluster, bool live_check)+get_db_rel_and_slot_infos(ClusterInfo *cluster)
{
int dbnum;
@@ -299,7 +297,7 @@ get_db_rel_and_slot_infos(ClusterInfo *cluster, bool live_check)
*/
if (cluster == &old_cluster)
{
- get_old_cluster_logical_slot_infos(pDbInfo, live_check);+ get_old_cluster_logical_slot_infos(pDbInfo);
get_db_subscription_count(pDbInfo);
}
}
@@ -645,7 +643,7 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
* are included.
*/
static void
-get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)+get_old_cluster_logical_slot_infos(DbInfo *dbinfo)
{
PGconn *conn;
PGresult *res;
@@ -681,7 +679,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
"WHERE slot_type = 'logical' AND "
"database = current_database() AND "
"temporary IS FALSE;",
- live_check ? "FALSE" :+ user_opts.live_check ? "FALSE" :
"(CASE WHEN invalidation_reason IS NOT NULL THEN FALSE "
"ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) "
"END)");
diff --git a/src/bin/pg_upgrade/option.c b/src/bin/pg_upgrade/option.cindex 548ea4e623..6f41d63eed 100644--- a/src/bin/pg_upgrade/option.c+++ b/src/bin/pg_upgrade/option.c@@ -470,10 +470,10 @@ adjust_data_dir(ClusterInfo *cluster)
* directory.
*/
void
-get_sock_dir(ClusterInfo *cluster, bool live_check)+get_sock_dir(ClusterInfo *cluster)
{
#if !defined(WIN32)
- if (!live_check)+ if (!user_opts.live_check || cluster == &new_cluster)
cluster->sockdir = user_opts.socketdir;
else
{
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.cindex af370768b6..3f4ad7d5cc 100644--- a/src/bin/pg_upgrade/pg_upgrade.c+++ b/src/bin/pg_upgrade/pg_upgrade.c@@ -65,7 +65,7 @@ static void create_new_objects(void);
static void copy_xact_xlog_xid(void);
static void set_frozenxids(bool minmxid_only);
static void make_outputdirs(char *pgdata);
-static void setup(char *argv0, bool *live_check);+static void setup(char *argv0);
static void create_logical_replication_slots(void);
ClusterInfo old_cluster,
@@ -88,7 +88,6 @@ int
main(int argc, char **argv)
{
char *deletion_script_file_name = NULL;
- bool live_check = false;
/*
* pg_upgrade doesn't currently use common/logging.c, but initialize it
@@ -123,18 +122,18 @@ main(int argc, char **argv)
*/
make_outputdirs(new_cluster.pgdata);
- setup(argv[0], &live_check);+ setup(argv[0]);- output_check_banner(live_check);+ output_check_banner();
check_cluster_versions();
- get_sock_dir(&old_cluster, live_check);- get_sock_dir(&new_cluster, false);+ get_sock_dir(&old_cluster);+ get_sock_dir(&new_cluster);- check_cluster_compatibility(live_check);+ check_cluster_compatibility();- check_and_dump_old_cluster(live_check);+ check_and_dump_old_cluster();
/* -- NEW -- */
@@ -331,7 +330,7 @@ make_outputdirs(char *pgdata)
static void
-setup(char *argv0, bool *live_check)+setup(char *argv0)
{
/*
* make sure the user has a clean environment, otherwise, we may confuse
@@ -378,7 +377,7 @@ setup(char *argv0, bool *live_check)
pg_fatal("There seems to be a postmaster servicing the old cluster.\n"
"Please shutdown that postmaster and try again.");
else
- *live_check = true;+ user_opts.live_check = true;
}
}
@@ -648,7 +647,7 @@ create_new_objects(void)
set_frozenxids(true);
/* update new_cluster info now that we have objects in the databases */
- get_db_rel_and_slot_infos(&new_cluster, false);+ get_db_rel_and_slot_infos(&new_cluster);
}
/*
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.hindex 1ebad3bd74..56d05d7eb9 100644--- a/src/bin/pg_upgrade/pg_upgrade.h+++ b/src/bin/pg_upgrade/pg_upgrade.h@@ -322,6 +322,7 @@ typedef struct
typedef struct
{
bool check; /* check clusters only, don't change any data */
+ bool live_check; /* check clusters only, old server is running */
bool do_sync; /* flush changes to disk */
transferMode transfer_mode; /* copy files or link them? */
int jobs; /* number of processes/threads to use */
@@ -366,20 +367,20 @@ extern OSInfo os_info;
/* check.c */
-void output_check_banner(bool live_check);-void check_and_dump_old_cluster(bool live_check);+void output_check_banner(void);+void check_and_dump_old_cluster(void);
void check_new_cluster(void);
void report_clusters_compatible(void);
void issue_warnings_and_set_wal_level(void);
void output_completion_banner(char *deletion_script_file_name);
void check_cluster_versions(void);
-void check_cluster_compatibility(bool live_check);+void check_cluster_compatibility(void);
void create_script_for_old_cluster_deletion(char **deletion_script_file_name);
/* controldata.c */
-void get_control_data(ClusterInfo *cluster, bool live_check);+void get_control_data(ClusterInfo *cluster);
void check_control_data(ControlData *oldctrl, ControlData *newctrl);
void disable_old_cluster(void);
@@ -428,7 +429,7 @@ void check_loadable_libraries(void);
FileNameMap *gen_db_file_maps(DbInfo *old_db,
DbInfo *new_db, int *nmaps, const char *old_pgdata,
const char *new_pgdata);
-void get_db_rel_and_slot_infos(ClusterInfo *cluster, bool live_check);+void get_db_rel_and_slot_infos(ClusterInfo *cluster);
int count_old_cluster_logical_slots(void);
int count_old_cluster_subscriptions(void);
@@ -436,7 +437,7 @@ int count_old_cluster_subscriptions(void);
void parseCommandLine(int argc, char *argv[]);
void adjust_data_dir(ClusterInfo *cluster);
-void get_sock_dir(ClusterInfo *cluster, bool live_check);+void get_sock_dir(ClusterInfo *cluster);
/* relfilenumber.c */
--
2.39.3 (Apple Git-146)
v3-0004-use-new-pg_upgrade-async-API-for-retrieving-relin.patch (text/plain, 11.1K) download
From 7b72cf76faba3532499e0f4d6ba1f5035f3fe1b6 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathan@postgresql.org>
Date: Fri, 28 Jun 2024 21:09:33 -0500
Subject: [PATCH v3 4/7] use new pg_upgrade async API for retrieving relinfos
---
src/bin/pg_upgrade/info.c | 187 +++++++++++++++++---------------------
1 file changed, 81 insertions(+), 106 deletions(-)
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.cindex 8f1777de59..d07255bd0a 100644--- a/src/bin/pg_upgrade/info.c+++ b/src/bin/pg_upgrade/info.c@@ -22,13 +22,16 @@ static void report_unmatched_relation(const RelInfo *rel, const DbInfo *db,
static void free_db_and_rel_infos(DbInfoArr *db_arr);
static void get_template0_info(ClusterInfo *cluster);
static void get_db_infos(ClusterInfo *cluster);
-static void get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo);+static char *get_rel_infos_query(DbInfo *dbinfo, void *arg);+static void get_rel_infos_result(DbInfo *dbinfo, PGresult *res, void *arg);
static void free_rel_infos(RelInfoArr *rel_arr);
static void print_db_infos(DbInfoArr *db_arr);
static void print_rel_infos(RelInfoArr *rel_arr);
static void print_slot_infos(LogicalSlotInfoArr *slot_arr);
-static void get_old_cluster_logical_slot_infos(DbInfo *dbinfo);-static void get_db_subscription_count(DbInfo *dbinfo);+static char *get_old_cluster_logical_slot_infos_query(DbInfo *dbinfo, void *arg);+static void get_old_cluster_logical_slot_infos_result(DbInfo *dbinfo, PGresult *res, void *arg);+static char *get_db_subscription_count_query(DbInfo *dbinfo, void *arg);+static void get_db_subscription_count_result(DbInfo *dbinfo, PGresult *res, void *arg);
/*
@@ -277,7 +280,7 @@ report_unmatched_relation(const RelInfo *rel, const DbInfo *db, bool is_new_db)
void
get_db_rel_and_slot_infos(ClusterInfo *cluster)
{
- int dbnum;+ AsyncTask *task = async_task_create();
if (cluster->dbarr.dbs != NULL)
free_db_and_rel_infos(&cluster->dbarr);
@@ -285,23 +288,26 @@ get_db_rel_and_slot_infos(ClusterInfo *cluster)
get_template0_info(cluster);
get_db_infos(cluster);
- for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)+ async_task_add_step(task,+ get_rel_infos_query,+ get_rel_infos_result,+ true, NULL);+ if (cluster == &old_cluster &&+ GET_MAJOR_VERSION(cluster->major_version) > 1600)
{
- DbInfo *pDbInfo = &cluster->dbarr.dbs[dbnum];-- get_rel_infos(cluster, pDbInfo);-- /*- * Retrieve the logical replication slots infos and the subscriptions- * count for the old cluster.- */- if (cluster == &old_cluster)- {- get_old_cluster_logical_slot_infos(pDbInfo);- get_db_subscription_count(pDbInfo);- }+ async_task_add_step(task,+ get_old_cluster_logical_slot_infos_query,+ get_old_cluster_logical_slot_infos_result,+ true, cluster);+ async_task_add_step(task,+ get_db_subscription_count_query,+ get_db_subscription_count_result,+ true, cluster);
}
+ async_task_run(task, cluster);+ async_task_free(task);+
if (cluster == &old_cluster)
pg_log(PG_VERBOSE, "\nsource databases:");
else
@@ -447,30 +453,10 @@ get_db_infos(ClusterInfo *cluster)
* Note: the resulting RelInfo array is assumed to be sorted by OID.
* This allows later processing to match up old and new databases efficiently.
*/
-static void-get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)+static char *+get_rel_infos_query(DbInfo *dbinfo, void *arg)
{
- PGconn *conn = connectToServer(cluster,- dbinfo->db_name);- PGresult *res;- RelInfo *relinfos;- int ntups;- int relnum;- int num_rels = 0;- char *nspname = NULL;- char *relname = NULL;- char *tablespace = NULL;- int i_spclocation,- i_nspname,- i_relname,- i_reloid,- i_indtable,- i_toastheap,- i_relfilenumber,- i_reltablespace;- char query[QUERY_ALLOC];- char *last_namespace = NULL,- *last_tablespace = NULL;+ char *query = pg_malloc(QUERY_ALLOC);
query[0] = '\0'; /* initialize query string to empty */
@@ -484,7 +470,7 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
* output, so we have to copy that system table. It's easiest to do that
* by treating it as a user table.
*/
- snprintf(query + strlen(query), sizeof(query) - strlen(query),+ snprintf(query + strlen(query), QUERY_ALLOC - strlen(query),
"WITH regular_heap (reloid, indtable, toastheap) AS ( "
" SELECT c.oid, 0::oid, 0::oid "
" FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n "
@@ -506,7 +492,7 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
* selected by the regular_heap CTE. (We have to do this separately
* because the namespace-name rules above don't work for toast tables.)
*/
- snprintf(query + strlen(query), sizeof(query) - strlen(query),+ snprintf(query + strlen(query), QUERY_ALLOC - strlen(query),
" toast_heap (reloid, indtable, toastheap) AS ( "
" SELECT c.reltoastrelid, 0::oid, c.oid "
" FROM regular_heap JOIN pg_catalog.pg_class c "
@@ -519,7 +505,7 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
* Testing indisready is necessary in 9.2, and harmless in earlier/later
* versions.
*/
- snprintf(query + strlen(query), sizeof(query) - strlen(query),+ snprintf(query + strlen(query), QUERY_ALLOC - strlen(query),
" all_index (reloid, indtable, toastheap) AS ( "
" SELECT indexrelid, indrelid, 0::oid "
" FROM pg_catalog.pg_index "
@@ -533,7 +519,7 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
* And now we can write the query that retrieves the data we want for each
* heap and index relation. Make sure result is sorted by OID.
*/
- snprintf(query + strlen(query), sizeof(query) - strlen(query),+ snprintf(query + strlen(query), QUERY_ALLOC - strlen(query),
"SELECT all_rels.*, n.nspname, c.relname, "
" c.relfilenode, c.reltablespace, "
" pg_catalog.pg_tablespace_location(t.oid) AS spclocation "
@@ -550,22 +536,30 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
" ON c.reltablespace = t.oid "
"ORDER BY 1;");
- res = executeQueryOrDie(conn, "%s", query);-- ntups = PQntuples(res);-- relinfos = (RelInfo *) pg_malloc(sizeof(RelInfo) * ntups);+ return query;+}- i_reloid = PQfnumber(res, "reloid");- i_indtable = PQfnumber(res, "indtable");- i_toastheap = PQfnumber(res, "toastheap");- i_nspname = PQfnumber(res, "nspname");- i_relname = PQfnumber(res, "relname");- i_relfilenumber = PQfnumber(res, "relfilenode");- i_reltablespace = PQfnumber(res, "reltablespace");- i_spclocation = PQfnumber(res, "spclocation");+static void+get_rel_infos_result(DbInfo *dbinfo, PGresult *res, void *arg)+{+ int ntups = PQntuples(res);+ RelInfo *relinfos = (RelInfo *) pg_malloc(sizeof(RelInfo) * ntups);+ int i_reloid = PQfnumber(res, "reloid");+ int i_indtable = PQfnumber(res, "indtable");+ int i_toastheap = PQfnumber(res, "toastheap");+ int i_nspname = PQfnumber(res, "nspname");+ int i_relname = PQfnumber(res, "relname");+ int i_relfilenumber = PQfnumber(res, "relfilenode");+ int i_reltablespace = PQfnumber(res, "reltablespace");+ int i_spclocation = PQfnumber(res, "spclocation");+ int num_rels = 0;+ char *nspname = NULL;+ char *relname = NULL;+ char *tablespace = NULL;+ char *last_namespace = NULL;+ char *last_tablespace = NULL;- for (relnum = 0; relnum < ntups; relnum++)+ for (int relnum = 0; relnum < ntups; relnum++)
{
RelInfo *curr = &relinfos[num_rels++];
@@ -618,9 +612,6 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
/* A zero reltablespace oid indicates the database tablespace. */
curr->tablespace = dbinfo->db_tablespace;
}
- PQclear(res);-- PQfinish(conn);
dbinfo->rel_arr.rels = relinfos;
dbinfo->rel_arr.nrels = num_rels;
@@ -642,20 +633,9 @@ get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo)
* check_old_cluster_for_valid_slots() would raise a FATAL error if such slots
* are included.
*/
-static void-get_old_cluster_logical_slot_infos(DbInfo *dbinfo)+static char *+get_old_cluster_logical_slot_infos_query(DbInfo *dbinfo, void *arg)
{
- PGconn *conn;- PGresult *res;- LogicalSlotInfo *slotinfos = NULL;- int num_slots;-- /* Logical slots can be migrated since PG17. */- if (GET_MAJOR_VERSION(old_cluster.major_version) <= 1600)- return;-- conn = connectToServer(&old_cluster, dbinfo->db_name);-
/*
* Fetch the logical replication slot information. The check whether the
* slot is considered caught up is done by an upgrade function. This
@@ -673,18 +653,23 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo)
* started and stopped several times causing any temporary slots to be
* removed.
*/
- res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, "- "%s as caught_up, invalidation_reason IS NOT NULL as invalid "- "FROM pg_catalog.pg_replication_slots "- "WHERE slot_type = 'logical' AND "- "database = current_database() AND "- "temporary IS FALSE;",- user_opts.live_check ? "FALSE" :- "(CASE WHEN invalidation_reason IS NOT NULL THEN FALSE "- "ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) "- "END)");-- num_slots = PQntuples(res);+ return psprintf("SELECT slot_name, plugin, two_phase, failover, "+ "%s as caught_up, invalidation_reason IS NOT NULL as invalid "+ "FROM pg_catalog.pg_replication_slots "+ "WHERE slot_type = 'logical' AND "+ "database = current_database() AND "+ "temporary IS FALSE;",+ user_opts.live_check ? "FALSE" :+ "(CASE WHEN invalidation_reason IS NOT NULL THEN FALSE "+ "ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) "+ "END)");+}++static void+get_old_cluster_logical_slot_infos_result(DbInfo *dbinfo, PGresult *res, void *arg)+{+ LogicalSlotInfo *slotinfos = NULL;+ int num_slots = PQntuples(res);
if (num_slots)
{
@@ -717,14 +702,10 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo)
}
}
- PQclear(res);- PQfinish(conn);-
dbinfo->slot_arr.slots = slotinfos;
dbinfo->slot_arr.nslots = num_slots;
}
-
/*
* count_old_cluster_logical_slots()
*
@@ -754,24 +735,18 @@ count_old_cluster_logical_slots(void)
* This is because before that the logical slots are not upgraded, so we will
* not be able to upgrade the logical replication clusters completely.
*/
-static void-get_db_subscription_count(DbInfo *dbinfo)+static char *+get_db_subscription_count_query(DbInfo *dbinfo, void *arg)
{
- PGconn *conn;- PGresult *res;-- /* Subscriptions can be migrated since PG17. */- if (GET_MAJOR_VERSION(old_cluster.major_version) < 1700)- return;+ return psprintf("SELECT count(*) "+ "FROM pg_catalog.pg_subscription WHERE subdbid = %u",+ dbinfo->db_oid);+}- conn = connectToServer(&old_cluster, dbinfo->db_name);- res = executeQueryOrDie(conn, "SELECT count(*) "- "FROM pg_catalog.pg_subscription WHERE subdbid = %u",- dbinfo->db_oid);+static void+get_db_subscription_count_result(DbInfo *dbinfo, PGresult *res, void *arg)+{
dbinfo->nsubs = atoi(PQgetvalue(res, 0, 0));
-- PQclear(res);- PQfinish(conn);
}
/*
--
2.39.3 (Apple Git-146)
v3-0005-use-new-pg_upgrade-async-API-to-parallelize-getti.patch (text/plain, 3.4K) download
From ca28a4f8fb44b1ac6921591ebafab41c6aa7ed98 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathan@postgresql.org>
Date: Fri, 28 Jun 2024 21:24:35 -0500
Subject: [PATCH v3 5/7] use new pg_upgrade async API to parallelize getting
loadable libraries
---
src/bin/pg_upgrade/function.c | 63 ++++++++++++++++++++---------------
1 file changed, 37 insertions(+), 26 deletions(-)
diff --git a/src/bin/pg_upgrade/function.c b/src/bin/pg_upgrade/function.cindex 7e3abed098..c11fce0696 100644--- a/src/bin/pg_upgrade/function.c+++ b/src/bin/pg_upgrade/function.c@@ -42,6 +42,32 @@ library_name_compare(const void *p1, const void *p2)
((const LibraryInfo *) p2)->dbnum);
}
+struct loadable_libraries_state+{+ PGresult **ress;+ int totaltups;+};++static char *+get_loadable_libraries_query(DbInfo *dbinfo, void *arg)+{+ return psprintf("SELECT DISTINCT probin "+ "FROM pg_catalog.pg_proc "+ "WHERE prolang = %u AND "+ "probin IS NOT NULL AND "+ "oid >= %u;",+ ClanguageId,+ FirstNormalObjectId);+}++static void+get_loadable_libraries_result(DbInfo *dbinfo, PGresult *res, void *arg)+{+ struct loadable_libraries_state *state = (struct loadable_libraries_state *) arg;++ state->ress[dbinfo - old_cluster.dbarr.dbs] = res;+ state->totaltups += PQntuples(res);+}
/*
* get_loadable_libraries()
@@ -54,47 +80,32 @@ library_name_compare(const void *p1, const void *p2)
void
get_loadable_libraries(void)
{
- PGresult **ress;
int totaltups;
int dbnum;
int n_libinfos;
+ AsyncTask *task = async_task_create();+ struct loadable_libraries_state state;- ress = (PGresult **) pg_malloc(old_cluster.dbarr.ndbs * sizeof(PGresult *));- totaltups = 0;+ state.ress = (PGresult **) pg_malloc(old_cluster.dbarr.ndbs * sizeof(PGresult *));+ state.totaltups = 0;- /* Fetch all library names, removing duplicates within each DB */- for (dbnum = 0; dbnum < old_cluster.dbarr.ndbs; dbnum++)- {- DbInfo *active_db = &old_cluster.dbarr.dbs[dbnum];- PGconn *conn = connectToServer(&old_cluster, active_db->db_name);+ async_task_add_step(task, get_loadable_libraries_query,+ get_loadable_libraries_result, false, &state);- /*- * Fetch all libraries containing non-built-in C functions in this DB.- */- ress[dbnum] = executeQueryOrDie(conn,- "SELECT DISTINCT probin "- "FROM pg_catalog.pg_proc "- "WHERE prolang = %u AND "- "probin IS NOT NULL AND "- "oid >= %u;",- ClanguageId,- FirstNormalObjectId);- totaltups += PQntuples(ress[dbnum]);-- PQfinish(conn);- }+ async_task_run(task, &old_cluster);+ async_task_free(task);
/*
* Allocate memory for required libraries and logical replication output
* plugins.
*/
- n_libinfos = totaltups + count_old_cluster_logical_slots();+ n_libinfos = state.totaltups + count_old_cluster_logical_slots();
os_info.libraries = (LibraryInfo *) pg_malloc(sizeof(LibraryInfo) * n_libinfos);
totaltups = 0;
for (dbnum = 0; dbnum < old_cluster.dbarr.ndbs; dbnum++)
{
- PGresult *res = ress[dbnum];+ PGresult *res = state.ress[dbnum];
int ntups;
int rowno;
LogicalSlotInfoArr *slot_arr = &old_cluster.dbarr.dbs[dbnum].slot_arr;
@@ -129,7 +140,7 @@ get_loadable_libraries(void)
}
}
- pg_free(ress);+ pg_free(state.ress);
os_info.num_libraries = totaltups;
}
--
2.39.3 (Apple Git-146)
v3-0006-use-new-pg_upgrade-async-API-to-parallelize-repor.patch (text/plain, 3.4K) download
From 8ef006b34462a51cf6c52bc1059eb7b38e26147d Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathan@postgresql.org>
Date: Fri, 28 Jun 2024 21:31:57 -0500
Subject: [PATCH v3 6/7] use new pg_upgrade async API to parallelize reporting
extension updates
---
src/bin/pg_upgrade/version.c | 82 ++++++++++++++++++------------------
1 file changed, 41 insertions(+), 41 deletions(-)
diff --git a/src/bin/pg_upgrade/version.c b/src/bin/pg_upgrade/version.cindex 2de6dffccd..12783bb2ba 100644--- a/src/bin/pg_upgrade/version.c+++ b/src/bin/pg_upgrade/version.c@@ -139,6 +139,42 @@ old_9_6_invalidate_hash_indexes(ClusterInfo *cluster, bool check_mode)
check_ok();
}
+static char *+report_extension_updates_query(DbInfo *dbinfo, void *arg)+{+ return pg_strdup("SELECT name "+ "FROM pg_available_extensions "+ "WHERE installed_version != default_version");+}++static void+report_extension_updates_result(DbInfo *dbinfo, PGresult *res, void *arg)+{+ bool db_used = false;+ int ntups = PQntuples(res);+ int i_name = PQfnumber(res, "name");+ char *output_path = "update_extensions.sql";+ FILE **script = (FILE **) arg;++ for (int rowno = 0; rowno < ntups; rowno++)+ {+ if (*script == NULL && (*script = fopen_priv(output_path, "w")) == NULL)+ pg_fatal("could not open file \"%s\": %m", output_path);+ if (!db_used)+ {+ PQExpBufferData connectbuf;++ initPQExpBuffer(&connectbuf);+ appendPsqlMetaConnect(&connectbuf, dbinfo->db_name);+ fputs(connectbuf.data, *script);+ termPQExpBuffer(&connectbuf);+ db_used = true;+ }+ fprintf(*script, "ALTER EXTENSION %s UPDATE;\n",+ quote_identifier(PQgetvalue(res, rowno, i_name)));+ }+}+
/*
* report_extension_updates()
* Report extensions that should be updated.
@@ -146,53 +182,17 @@ old_9_6_invalidate_hash_indexes(ClusterInfo *cluster, bool check_mode)
void
report_extension_updates(ClusterInfo *cluster)
{
- int dbnum;
FILE *script = NULL;
char *output_path = "update_extensions.sql";
+ AsyncTask *task = async_task_create();
prep_status("Checking for extension updates");
- for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)- {- PGresult *res;- bool db_used = false;- int ntups;- int rowno;- int i_name;- DbInfo *active_db = &cluster->dbarr.dbs[dbnum];- PGconn *conn = connectToServer(cluster, active_db->db_name);-- /* find extensions needing updates */- res = executeQueryOrDie(conn,- "SELECT name "- "FROM pg_available_extensions "- "WHERE installed_version != default_version"- );-- ntups = PQntuples(res);- i_name = PQfnumber(res, "name");- for (rowno = 0; rowno < ntups; rowno++)- {- if (script == NULL && (script = fopen_priv(output_path, "w")) == NULL)- pg_fatal("could not open file \"%s\": %m", output_path);- if (!db_used)- {- PQExpBufferData connectbuf;-- initPQExpBuffer(&connectbuf);- appendPsqlMetaConnect(&connectbuf, active_db->db_name);- fputs(connectbuf.data, script);- termPQExpBuffer(&connectbuf);- db_used = true;- }- fprintf(script, "ALTER EXTENSION %s UPDATE;\n",- quote_identifier(PQgetvalue(res, rowno, i_name)));- }+ async_task_add_step(task, report_extension_updates_query,+ report_extension_updates_result, true, &script);- PQclear(res);-- PQfinish(conn);- }+ async_task_run(task, cluster);+ async_task_free(task);
if (script)
{
--
2.39.3 (Apple Git-146)
v3-0007-parallelize-data-type-checks-in-pg_upgrade.patch (text/plain, 12.4K) download
From 203af9d3861b4a62548e98e96ee6dad35e7f1123 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathan@postgresql.org>
Date: Sat, 6 Jul 2024 21:06:31 -0500
Subject: [PATCH v3 7/7] parallelize data type checks in pg_upgrade
---
src/bin/pg_upgrade/check.c | 319 +++++++++++++++++++------------------
1 file changed, 160 insertions(+), 159 deletions(-)
diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.cindex 251f3d9017..800d944218 100644--- a/src/bin/pg_upgrade/check.c+++ b/src/bin/pg_upgrade/check.c@@ -32,6 +32,10 @@ static void check_new_cluster_subscription_configuration(void);
static void check_old_cluster_for_valid_slots(void);
static void check_old_cluster_subscription_state(void);
+static bool *data_type_check_results;+static bool data_type_check_failed;+static PQExpBufferData data_type_check_report;+
/*
* DataTypesUsageChecks - definitions of data type checks for the old cluster
* in order to determine if an upgrade can be performed. See the comment on
@@ -314,6 +318,129 @@ static DataTypesUsageChecks data_types_usage_checks[] =
}
};
+static char *+data_type_check_query(DbInfo *dbinfo, void *arg)+{+ int i = (int) (intptr_t) arg;+ DataTypesUsageChecks *check = &data_types_usage_checks[i];++ return psprintf("WITH RECURSIVE oids AS ( "+ /* start with the type(s) returned by base_query */+ " %s "+ " UNION ALL "+ " SELECT * FROM ( "+ /* inner WITH because we can only reference the CTE once */+ " WITH x AS (SELECT oid FROM oids) "+ /* domains on any type selected so far */+ " SELECT t.oid FROM pg_catalog.pg_type t, x WHERE typbasetype = x.oid AND typtype = 'd' "+ " UNION ALL "+ /* arrays over any type selected so far */+ " SELECT t.oid FROM pg_catalog.pg_type t, x WHERE typelem = x.oid AND typtype = 'b' "+ " UNION ALL "+ /* composite types containing any type selected so far */+ " SELECT t.oid FROM pg_catalog.pg_type t, pg_catalog.pg_class c, pg_catalog.pg_attribute a, x "+ " WHERE t.typtype = 'c' AND "+ " t.oid = c.reltype AND "+ " c.oid = a.attrelid AND "+ " NOT a.attisdropped AND "+ " a.atttypid = x.oid "+ " UNION ALL "+ /* ranges containing any type selected so far */+ " SELECT t.oid FROM pg_catalog.pg_type t, pg_catalog.pg_range r, x "+ " WHERE t.typtype = 'r' AND r.rngtypid = t.oid AND r.rngsubtype = x.oid"+ " ) foo "+ ") "+ /* now look for stored columns of any such type */+ "SELECT n.nspname, c.relname, a.attname "+ "FROM pg_catalog.pg_class c, "+ " pg_catalog.pg_namespace n, "+ " pg_catalog.pg_attribute a "+ "WHERE c.oid = a.attrelid AND "+ " NOT a.attisdropped AND "+ " a.atttypid IN (SELECT oid FROM oids) AND "+ " c.relkind IN ("+ CppAsString2(RELKIND_RELATION) ", "+ CppAsString2(RELKIND_MATVIEW) ", "+ CppAsString2(RELKIND_INDEX) ") AND "+ " c.relnamespace = n.oid AND "+ /* exclude possible orphaned temp tables */+ " n.nspname !~ '^pg_temp_' AND "+ " n.nspname !~ '^pg_toast_temp_' AND "+ /* exclude system catalogs, too */+ " n.nspname NOT IN ('pg_catalog', 'information_schema')",+ check->base_query);+}++static void+data_type_check_process(DbInfo *dbinfo, PGresult *res, void *arg)+{+ int i = (int) (intptr_t) arg;+ DataTypesUsageChecks *check = &data_types_usage_checks[i];+ int ntups = PQntuples(res);++ if (ntups)+ {+ char output_path[MAXPGPATH];+ int i_nspname;+ int i_relname;+ int i_attname;+ FILE *script = NULL;+ bool db_used = false;++ snprintf(output_path, sizeof(output_path), "%s/%s",+ log_opts.basedir,+ check->report_filename);++ /*+ * Make sure we have a buffer to save reports to now that we found a+ * first failing check.+ */+ if (!data_type_check_failed)+ initPQExpBuffer(&data_type_check_report);+ data_type_check_failed = true;++ /*+ * If this is the first time we see an error for the check in question+ * then print a status message of the failure.+ */+ if (!data_type_check_results[i])+ {+ pg_log(PG_REPORT, " failed check: %s", _(check->status));+ appendPQExpBuffer(&data_type_check_report, "\n%s\n%s %s\n",+ _(check->report_text),+ _("A list of the problem columns is in the file:"),+ output_path);+ }+ data_type_check_results[i] = true;++ i_nspname = PQfnumber(res, "nspname");+ i_relname = PQfnumber(res, "relname");+ i_attname = PQfnumber(res, "attname");++ for (int rowno = 0; rowno < ntups; rowno++)+ {+ if (script == NULL && (script = fopen_priv(output_path, "a")) == NULL)+ pg_fatal("could not open file \"%s\": %m", output_path);++ if (!db_used)+ {+ fprintf(script, "In database: %s\n", dbinfo->db_name);+ db_used = true;+ }+ fprintf(script, " %s.%s.%s\n",+ PQgetvalue(res, rowno, i_nspname),+ PQgetvalue(res, rowno, i_relname),+ PQgetvalue(res, rowno, i_attname));+ }++ if (script)+ {+ fclose(script);+ script = NULL;+ }+ }+}+
/*
* check_for_data_types_usage()
* Detect whether there are any stored columns depending on given type(s)
@@ -337,10 +464,9 @@ static void
check_for_data_types_usage(ClusterInfo *cluster, DataTypesUsageChecks *checks)
{
bool found = false;
- bool *results;- PQExpBufferData report;
DataTypesUsageChecks *tmp = checks;
int n_data_types_usage_checks = 0;
+ AsyncTask *task = async_task_create();
prep_status("Checking for data type usage");
@@ -352,176 +478,51 @@ check_for_data_types_usage(ClusterInfo *cluster, DataTypesUsageChecks *checks)
}
/* Prepare an array to store the results of checks in */
- results = pg_malloc0(sizeof(bool) * n_data_types_usage_checks);+ data_type_check_results = pg_malloc0(sizeof(bool) * n_data_types_usage_checks);+ data_type_check_failed = false;- /*- * Connect to each database in the cluster and run all defined checks- * against that database before trying the next one.- */- for (int dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)+ for (int i = 0; i < n_data_types_usage_checks; i++)
{
- DbInfo *active_db = &cluster->dbarr.dbs[dbnum];- PGconn *conn = connectToServer(cluster, active_db->db_name);+ DataTypesUsageChecks *check = &checks[i];- for (int checknum = 0; checknum < n_data_types_usage_checks; checknum++)+ if (check->threshold_version == MANUAL_CHECK)
{
- PGresult *res;- int ntups;- int i_nspname;- int i_relname;- int i_attname;- FILE *script = NULL;- bool db_used = false;- char output_path[MAXPGPATH];- DataTypesUsageChecks *cur_check = &checks[checknum];-- if (cur_check->threshold_version == MANUAL_CHECK)- {- Assert(cur_check->version_hook);-- /*- * Make sure that the check applies to the current cluster- * version and skip if not. If no check hook has been defined- * we run the check for all versions.- */- if (!cur_check->version_hook(cluster))- continue;- }- else if (cur_check->threshold_version != ALL_VERSIONS)- {- if (GET_MAJOR_VERSION(cluster->major_version) > cur_check->threshold_version)- continue;- }- else- Assert(cur_check->threshold_version == ALL_VERSIONS);-- snprintf(output_path, sizeof(output_path), "%s/%s",- log_opts.basedir,- cur_check->report_filename);-- /*- * The type(s) of interest might be wrapped in a domain, array,- * composite, or range, and these container types can be nested- * (to varying extents depending on server version, but that's not- * of concern here). To handle all these cases we need a- * recursive CTE.- */- res = executeQueryOrDie(conn,- "WITH RECURSIVE oids AS ( "- /* start with the type(s) returned by base_query */- " %s "- " UNION ALL "- " SELECT * FROM ( "- /* inner WITH because we can only reference the CTE once */- " WITH x AS (SELECT oid FROM oids) "- /* domains on any type selected so far */- " SELECT t.oid FROM pg_catalog.pg_type t, x WHERE typbasetype = x.oid AND typtype = 'd' "- " UNION ALL "- /* arrays over any type selected so far */- " SELECT t.oid FROM pg_catalog.pg_type t, x WHERE typelem = x.oid AND typtype = 'b' "- " UNION ALL "- /* composite types containing any type selected so far */- " SELECT t.oid FROM pg_catalog.pg_type t, pg_catalog.pg_class c, pg_catalog.pg_attribute a, x "- " WHERE t.typtype = 'c' AND "- " t.oid = c.reltype AND "- " c.oid = a.attrelid AND "- " NOT a.attisdropped AND "- " a.atttypid = x.oid "- " UNION ALL "- /* ranges containing any type selected so far */- " SELECT t.oid FROM pg_catalog.pg_type t, pg_catalog.pg_range r, x "- " WHERE t.typtype = 'r' AND r.rngtypid = t.oid AND r.rngsubtype = x.oid"- " ) foo "- ") "- /* now look for stored columns of any such type */- "SELECT n.nspname, c.relname, a.attname "- "FROM pg_catalog.pg_class c, "- " pg_catalog.pg_namespace n, "- " pg_catalog.pg_attribute a "- "WHERE c.oid = a.attrelid AND "- " NOT a.attisdropped AND "- " a.atttypid IN (SELECT oid FROM oids) AND "- " c.relkind IN ("- CppAsString2(RELKIND_RELATION) ", "- CppAsString2(RELKIND_MATVIEW) ", "- CppAsString2(RELKIND_INDEX) ") AND "- " c.relnamespace = n.oid AND "- /* exclude possible orphaned temp tables */- " n.nspname !~ '^pg_temp_' AND "- " n.nspname !~ '^pg_toast_temp_' AND "- /* exclude system catalogs, too */- " n.nspname NOT IN ('pg_catalog', 'information_schema')",- cur_check->base_query);-- ntups = PQntuples(res);+ Assert(check->version_hook);
/*
- * The datatype was found, so extract the data and log to the- * requested filename. We need to open the file for appending- * since the check might have already found the type in another- * database earlier in the loop.+ * Make sure that the check applies to the current cluster version+ * and skip it if not.
*/
- if (ntups)- {- /*- * Make sure we have a buffer to save reports to now that we- * found a first failing check.- */- if (!found)- initPQExpBuffer(&report);- found = true;-- /*- * If this is the first time we see an error for the check in- * question then print a status message of the failure.- */- if (!results[checknum])- {- pg_log(PG_REPORT, " failed check: %s", _(cur_check->status));- appendPQExpBuffer(&report, "\n%s\n%s %s\n",- _(cur_check->report_text),- _("A list of the problem columns is in the file:"),- output_path);- }- results[checknum] = true;-- i_nspname = PQfnumber(res, "nspname");- i_relname = PQfnumber(res, "relname");- i_attname = PQfnumber(res, "attname");-- for (int rowno = 0; rowno < ntups; rowno++)- {- if (script == NULL && (script = fopen_priv(output_path, "a")) == NULL)- pg_fatal("could not open file \"%s\": %m", output_path);-- if (!db_used)- {- fprintf(script, "In database: %s\n", active_db->db_name);- db_used = true;- }- fprintf(script, " %s.%s.%s\n",- PQgetvalue(res, rowno, i_nspname),- PQgetvalue(res, rowno, i_relname),- PQgetvalue(res, rowno, i_attname));- }-- if (script)- {- fclose(script);- script = NULL;- }- }-- PQclear(res);+ if (!check->version_hook(cluster))+ continue;
}
+ else if (check->threshold_version != ALL_VERSIONS)+ {+ if (GET_MAJOR_VERSION(cluster->major_version) > check->threshold_version)+ continue;+ }+ else+ Assert(check->threshold_version == ALL_VERSIONS);- PQfinish(conn);+ async_task_add_step(task, data_type_check_query,+ data_type_check_process, true,+ (void *) (intptr_t) i);
}
+ /*+ * Connect to each database in the cluster and run all defined checks+ * against that database before trying the next one.+ */+ async_task_run(task, cluster);+ async_task_free(task);+
if (found)
- pg_fatal("Data type checks failed: %s", report.data);+ {+ pg_fatal("Data type checks failed: %s", data_type_check_report.data);+ termPQExpBuffer(&data_type_check_report);+ }- pg_free(results);+ pg_free(data_type_check_results);
check_ok();
}
--
2.39.3 (Apple Git-146)
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: nathandbossart@gmail.com, robertmhaas@gmail.com, pgsql@j-davis.com
Subject: Re: optimizing pg_upgrade's once-in-each-database steps
In-Reply-To: <ZowEIGd0lLKh16qe@nathan>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox