src/bin/pg_upgrade/info.c - master - githubmirror/postgres

src/bin/pg_upgrade/info.c
949 строк29 KB
Masahiko Sawada
pg_upgrade: Fix detection of invalid logical replication slots.
22 апр 2026, 19:59
22 апр 2026, 19:59e471dc5
/*
 *	info.c
 *
 *	information support functions
 *
 *	Copyright (c) 2010-2026, PostgreSQL Global Development Group
 *	src/bin/pg_upgrade/info.c
 */

#include "postgres_fe.h"

#include "access/transam.h"
#include "catalog/pg_class_d.h"
#include "pg_upgrade.h"
#include "pqexpbuffer.h"

static void create_rel_filename_map(const char *old_data, const char *new_data,
									const DbInfo *old_db, const DbInfo *new_db,
									const RelInfo *old_rel, const RelInfo *new_rel,
									FileNameMap *map);
static void report_unmatched_relation(const RelInfo *rel, const DbInfo *db,
									  bool is_new_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 char *get_rel_infos_query(void);
static void process_rel_infos(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 const char *get_old_cluster_logical_slot_infos_query(ClusterInfo *cluster);
static void process_old_cluster_logical_slot_infos(DbInfo *dbinfo, PGresult *res, void *arg);


/*
 * gen_db_file_maps()
 *
 * generates a database mapping from "old_db" to "new_db".
 *
 * Returns a malloc'ed array of mappings.  The length of the array
 * is returned into *nmaps.
 */
FileNameMap *
gen_db_file_maps(DbInfo *old_db, DbInfo *new_db,
				 int *nmaps,
				 const char *old_pgdata, const char *new_pgdata)
{
	FileNameMap *maps;
	int			old_relnum,
				new_relnum;
	int			num_maps = 0;
	bool		all_matched = true;

	/* There will certainly not be more mappings than there are old rels */
	maps = pg_malloc_array(FileNameMap, old_db->rel_arr.nrels);

	/*
	 * Each of the RelInfo arrays should be sorted by OID.  Scan through them
	 * and match them up.  If we fail to match everything, we'll abort, but
	 * first print as much info as we can about mismatches.
	 */
	old_relnum = new_relnum = 0;
	while (old_relnum < old_db->rel_arr.nrels ||
		   new_relnum < new_db->rel_arr.nrels)
	{
		RelInfo    *old_rel = (old_relnum < old_db->rel_arr.nrels) ?
			&old_db->rel_arr.rels[old_relnum] : NULL;
		RelInfo    *new_rel = (new_relnum < new_db->rel_arr.nrels) ?
			&new_db->rel_arr.rels[new_relnum] : NULL;

		/* handle running off one array before the other */
		if (!new_rel)
		{
			/*
			 * old_rel is unmatched.  This should never happen, because we
			 * force new rels to have TOAST tables if the old one did.
			 */
			report_unmatched_relation(old_rel, old_db, false);
			all_matched = false;
			old_relnum++;
			continue;
		}
		if (!old_rel)
		{
			/*
			 * new_rel is unmatched.  This shouldn't really happen either, but
			 * if it's a TOAST table, we can ignore it and continue
			 * processing, assuming that the new server made a TOAST table
			 * that wasn't needed.
			 */
			if (strcmp(new_rel->nspname, "pg_toast") != 0)
			{
				report_unmatched_relation(new_rel, new_db, true);
				all_matched = false;
			}
			new_relnum++;
			continue;
		}

		/* check for mismatched OID */
		if (old_rel->reloid < new_rel->reloid)
		{
			/* old_rel is unmatched, see comment above */
			report_unmatched_relation(old_rel, old_db, false);
			all_matched = false;
			old_relnum++;
			continue;
		}
		else if (old_rel->reloid > new_rel->reloid)
		{
			/* new_rel is unmatched, see comment above */
			if (strcmp(new_rel->nspname, "pg_toast") != 0)
			{
				report_unmatched_relation(new_rel, new_db, true);
				all_matched = false;
			}
			new_relnum++;
			continue;
		}

		/*
		 * Verify that rels of same OID have same name.  The namespace name
		 * should always match, but the relname might not match for TOAST
		 * tables (and, therefore, their indexes).
		 */
		if (strcmp(old_rel->nspname, new_rel->nspname) != 0 ||
			strcmp(old_rel->relname, new_rel->relname) != 0)
		{
			pg_log(PG_WARNING, "Relation names for OID %u in database \"%s\" do not match: "
				   "old name \"%s.%s\", new name \"%s.%s\"",
				   old_rel->reloid, old_db->db_name,
				   old_rel->nspname, old_rel->relname,
				   new_rel->nspname, new_rel->relname);
			all_matched = false;
			old_relnum++;
			new_relnum++;
			continue;
		}

		/* OK, create a mapping entry */
		create_rel_filename_map(old_pgdata, new_pgdata, old_db, new_db,
								old_rel, new_rel, maps + num_maps);
		num_maps++;
		old_relnum++;
		new_relnum++;
	}

	if (!all_matched)
		pg_fatal("Failed to match up old and new tables in database \"%s\"",
				 old_db->db_name);

	*nmaps = num_maps;
	return maps;
}


/*
 * create_rel_filename_map()
 *
 * fills a file node map structure and returns it in "map".
 */
static void
create_rel_filename_map(const char *old_data, const char *new_data,
						const DbInfo *old_db, const DbInfo *new_db,
						const RelInfo *old_rel, const RelInfo *new_rel,
						FileNameMap *map)
{
	/* In case old/new tablespaces don't match, do them separately. */
	if (strlen(old_rel->tablespace) == 0)
	{
		/*
		 * relation belongs to the default tablespace, hence relfiles should
		 * exist in the data directories.
		 */
		map->old_tablespace = old_data;
		map->old_tablespace_suffix = "/base";
	}
	else
	{
		/* relation belongs to a tablespace, so use the tablespace location */
		map->old_tablespace = old_rel->tablespace;
		map->old_tablespace_suffix = old_cluster.tablespace_suffix;
	}

	/* Do the same for new tablespaces */
	if (strlen(new_rel->tablespace) == 0)
	{
		map->new_tablespace = new_data;
		map->new_tablespace_suffix = "/base";
	}
	else
	{
		map->new_tablespace = new_rel->tablespace;
		map->new_tablespace_suffix = new_cluster.tablespace_suffix;
	}

	/* DB oid and relfilenumbers are preserved between old and new cluster */
	map->db_oid = old_db->db_oid;
	map->relfilenumber = old_rel->relfilenumber;

	/* used only for logging and error reporting, old/new are identical */
	map->nspname = old_rel->nspname;
	map->relname = old_rel->relname;
}


/*
 * Complain about a relation we couldn't match to the other database,
 * identifying it as best we can.
 */
static void
report_unmatched_relation(const RelInfo *rel, const DbInfo *db, bool is_new_db)
{
	Oid			reloid = rel->reloid;	/* we might change rel below */
	char		reldesc[1000];
	int			i;

	snprintf(reldesc, sizeof(reldesc), "\"%s.%s\"",
			 rel->nspname, rel->relname);
	if (rel->indtable)
	{
		for (i = 0; i < db->rel_arr.nrels; i++)
		{
			const RelInfo *hrel = &db->rel_arr.rels[i];

			if (hrel->reloid == rel->indtable)
			{
				snprintf(reldesc + strlen(reldesc),
						 sizeof(reldesc) - strlen(reldesc),
						 _(" which is an index on \"%s.%s\""),
						 hrel->nspname, hrel->relname);
				/* Shift attention to index's table for toast check */
				rel = hrel;
				break;
			}
		}
		if (i >= db->rel_arr.nrels)
			snprintf(reldesc + strlen(reldesc),
					 sizeof(reldesc) - strlen(reldesc),
					 _(" which is an index on OID %u"), rel->indtable);
	}
	if (rel->toastheap)
	{
		for (i = 0; i < db->rel_arr.nrels; i++)
		{
			const RelInfo *brel = &db->rel_arr.rels[i];

			if (brel->reloid == rel->toastheap)
			{
				snprintf(reldesc + strlen(reldesc),
						 sizeof(reldesc) - strlen(reldesc),
						 _(" which is the TOAST table for \"%s.%s\""),
						 brel->nspname, brel->relname);
				break;
			}
		}
		if (i >= db->rel_arr.nrels)
			snprintf(reldesc + strlen(reldesc),
					 sizeof(reldesc) - strlen(reldesc),
					 _(" which is the TOAST table for OID %u"), rel->toastheap);
	}

	if (is_new_db)
		pg_log(PG_WARNING, "No match found in old cluster for new relation with OID %u in database \"%s\": %s",
			   reloid, db->db_name, reldesc);
	else
		pg_log(PG_WARNING, "No match found in new cluster for old relation with OID %u in database \"%s\": %s",
			   reloid, db->db_name, reldesc);
}

/*
 * get_db_rel_and_slot_infos()
 *
 * higher level routine to generate dbinfos for the database running
 * on the given "port". Assumes that server is already running.
 */
void
get_db_rel_and_slot_infos(ClusterInfo *cluster)
{
	UpgradeTask *task = upgrade_task_create();
	char	   *rel_infos_query = NULL;

	if (cluster->dbarr.dbs != NULL)
		free_db_and_rel_infos(&cluster->dbarr);

	get_template0_info(cluster);
	get_db_infos(cluster);

	rel_infos_query = get_rel_infos_query();
	upgrade_task_add_step(task,
						  rel_infos_query,
						  process_rel_infos,
						  true, NULL);

	/*
	 * Logical slots are only carried over to the new cluster when the old
	 * cluster is on PG17 or newer.  This is because before that the logical
	 * slots are not saved at shutdown, so there is no guarantee that the
	 * latest confirmed_flush_lsn is saved to disk which can lead to data
	 * loss. It is still not guaranteed for manually created slots in PG17, so
	 * subsequent checks done in check_old_cluster_for_valid_slots() would
	 * raise a FATAL error if such slots are included.
	 */
	if (cluster == &old_cluster &&
		GET_MAJOR_VERSION(cluster->major_version) > 1600)
		upgrade_task_add_step(task,
							  get_old_cluster_logical_slot_infos_query(cluster),
							  process_old_cluster_logical_slot_infos,
							  true, NULL);

	upgrade_task_run(task, cluster);
	upgrade_task_free(task);

	pg_free(rel_infos_query);

	if (cluster == &old_cluster)
		pg_log(PG_VERBOSE, "\nsource databases:");
	else
		pg_log(PG_VERBOSE, "\ntarget databases:");

	if (log_opts.verbose)
		print_db_infos(&cluster->dbarr);
}


/*
 * Get information about template0, which will be copied from the old cluster
 * to the new cluster.
 */
static void
get_template0_info(ClusterInfo *cluster)
{
	PGconn	   *conn = connectToServer(cluster, "template1");
	DbLocaleInfo *locale;
	PGresult   *dbres;
	int			i_datencoding;
	int			i_datlocprovider;
	int			i_datcollate;
	int			i_datctype;
	int			i_datlocale;

	if (GET_MAJOR_VERSION(cluster->major_version) >= 1700)
		dbres = executeQueryOrDie(conn,
								  "SELECT encoding, datlocprovider, "
								  "       datcollate, datctype, datlocale "
								  "FROM	pg_catalog.pg_database "
								  "WHERE datname='template0'");
	else if (GET_MAJOR_VERSION(cluster->major_version) >= 1500)
		dbres = executeQueryOrDie(conn,
								  "SELECT encoding, datlocprovider, "
								  "       datcollate, datctype, daticulocale AS datlocale "
								  "FROM	pg_catalog.pg_database "
								  "WHERE datname='template0'");
	else
		dbres = executeQueryOrDie(conn,
								  "SELECT encoding, 'c' AS datlocprovider, "
								  "       datcollate, datctype, NULL AS datlocale "
								  "FROM	pg_catalog.pg_database "
								  "WHERE datname='template0'");


	if (PQntuples(dbres) != 1)
		pg_fatal("template0 not found");

	locale = pg_malloc_object(DbLocaleInfo);

	i_datencoding = PQfnumber(dbres, "encoding");
	i_datlocprovider = PQfnumber(dbres, "datlocprovider");
	i_datcollate = PQfnumber(dbres, "datcollate");
	i_datctype = PQfnumber(dbres, "datctype");
	i_datlocale = PQfnumber(dbres, "datlocale");

	locale->db_encoding = atoi(PQgetvalue(dbres, 0, i_datencoding));
	locale->db_collprovider = PQgetvalue(dbres, 0, i_datlocprovider)[0];
	locale->db_collate = pg_strdup(PQgetvalue(dbres, 0, i_datcollate));
	locale->db_ctype = pg_strdup(PQgetvalue(dbres, 0, i_datctype));
	if (PQgetisnull(dbres, 0, i_datlocale))
		locale->db_locale = NULL;
	else
		locale->db_locale = pg_strdup(PQgetvalue(dbres, 0, i_datlocale));

	cluster->template0 = locale;

	PQclear(dbres);
	PQfinish(conn);
}


/*
 * get_db_infos()
 *
 * Scans pg_database system catalog and populates all user
 * databases.
 */
static void
get_db_infos(ClusterInfo *cluster)
{
	PGconn	   *conn = connectToServer(cluster, "template1");
	PGresult   *res;
	int			ntups;
	int			tupnum;
	DbInfo	   *dbinfos;
	int			i_datname,
				i_oid,
				i_spclocation;
	char		query[QUERY_ALLOC];

	snprintf(query, sizeof(query),
			 "SELECT d.oid, d.datname, d.encoding, d.datcollate, d.datctype, ");
	if (GET_MAJOR_VERSION(cluster->major_version) >= 1700)
		snprintf(query + strlen(query), sizeof(query) - strlen(query),
				 "datlocprovider, datlocale, ");
	else if (GET_MAJOR_VERSION(cluster->major_version) >= 1500)
		snprintf(query + strlen(query), sizeof(query) - strlen(query),
				 "datlocprovider, daticulocale AS datlocale, ");
	else
		snprintf(query + strlen(query), sizeof(query) - strlen(query),
				 "'c' AS datlocprovider, NULL AS datlocale, ");
	snprintf(query + strlen(query), sizeof(query) - strlen(query),
			 "pg_catalog.pg_tablespace_location(t.oid) AS spclocation "
			 "FROM pg_catalog.pg_database d "
			 " LEFT OUTER JOIN pg_catalog.pg_tablespace t "
			 " ON d.dattablespace = t.oid "
			 "WHERE d.datallowconn = true "
			 "ORDER BY 1");

	res = executeQueryOrDie(conn, "%s", query);

	i_oid = PQfnumber(res, "oid");
	i_datname = PQfnumber(res, "datname");
	i_spclocation = PQfnumber(res, "spclocation");

	ntups = PQntuples(res);
	dbinfos = pg_malloc0_array(DbInfo, ntups);

	for (tupnum = 0; tupnum < ntups; tupnum++)
	{
		char	   *spcloc = PQgetvalue(res, tupnum, i_spclocation);
		bool		inplace = spcloc[0] && !is_absolute_path(spcloc);

		dbinfos[tupnum].db_oid = atooid(PQgetvalue(res, tupnum, i_oid));
		dbinfos[tupnum].db_name = pg_strdup(PQgetvalue(res, tupnum, i_datname));

		/*
		 * The tablespace location might be "", meaning the cluster default
		 * location, i.e. pg_default or pg_global.  For in-place tablespaces,
		 * pg_tablespace_location() returns a path relative to the data
		 * directory.
		 */
		if (inplace)
			snprintf(dbinfos[tupnum].db_tablespace,
					 sizeof(dbinfos[tupnum].db_tablespace),
					 "%s/%s", cluster->pgdata, spcloc);
		else
			snprintf(dbinfos[tupnum].db_tablespace,
					 sizeof(dbinfos[tupnum].db_tablespace),
					 "%s", spcloc);
	}
	PQclear(res);

	PQfinish(conn);

	cluster->dbarr.dbs = dbinfos;
	cluster->dbarr.ndbs = ntups;
}


/*
 * get_rel_infos_query()
 *
 * Returns the query for retrieving the relation information for all the user
 * tables and indexes in the database, for use by get_db_rel_and_slot_infos()'s
 * UpgradeTask.
 *
 * Note: the result is assumed to be sorted by OID.  This allows later
 * processing to match up old and new databases efficiently.
 */
static char *
get_rel_infos_query(void)
{
	PQExpBufferData query;

	initPQExpBuffer(&query);

	/*
	 * Create a CTE that collects OIDs of regular user tables and matviews,
	 * but excluding toast tables and indexes.  We assume that relations with
	 * OIDs >= FirstNormalObjectId belong to the user.  (That's probably
	 * redundant with the namespace-name exclusions, but let's be safe.)
	 *
	 * pg_largeobject contains user data that does not appear in pg_dump
	 * output, so we have to copy that system table.  It's easiest to do that
	 * by treating it as a user table.  We can do the same for
	 * pg_largeobject_metadata for upgrades from v16 and newer.  pg_upgrade
	 * can't copy/link the files from older versions because aclitem (needed
	 * by pg_largeobject_metadata.lomacl) changed its storage format in v16.
	 */
	appendPQExpBuffer(&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 "
					  "         ON c.relnamespace = n.oid "
					  "  WHERE relkind IN (" CppAsString2(RELKIND_RELATION) ", "
					  CppAsString2(RELKIND_MATVIEW) "%s) AND "
	/* exclude possible orphaned temp tables */
					  "    ((n.nspname !~ '^pg_temp_' AND "
					  "      n.nspname !~ '^pg_toast_temp_' AND "
					  "      n.nspname NOT IN ('pg_catalog', 'information_schema', "
					  "                        'binary_upgrade', 'pg_toast') AND "
					  "      c.oid >= %u::pg_catalog.oid) OR "
					  "     (n.nspname = 'pg_catalog' AND "
					  "      relname IN ('pg_largeobject'%s) ))), ",
					  (user_opts.transfer_mode == TRANSFER_MODE_SWAP) ?
					  ", " CppAsString2(RELKIND_SEQUENCE) : "",
					  FirstNormalObjectId,
					  (GET_MAJOR_VERSION(old_cluster.major_version) >= 1600) ?
					  ", 'pg_largeobject_metadata'" : "");

	/*
	 * Add a CTE that collects OIDs of toast tables belonging to the tables
	 * selected by the regular_heap CTE.  (We have to do this separately
	 * because the namespace-name rules above don't work for toast tables.)
	 */
	appendPQExpBufferStr(&query,
						 "  toast_heap (reloid, indtable, toastheap) AS ( "
						 "  SELECT c.reltoastrelid, 0::oid, c.oid "
						 "  FROM regular_heap JOIN pg_catalog.pg_class c "
						 "      ON regular_heap.reloid = c.oid "
						 "  WHERE c.reltoastrelid != 0), ");

	/*
	 * Add a CTE that collects OIDs of all valid indexes on the previously
	 * selected tables.  We can ignore invalid indexes since pg_dump does.
	 * Testing indisready is necessary in 9.2, and harmless in earlier/later
	 * versions.
	 */
	appendPQExpBufferStr(&query,
						 "  all_index (reloid, indtable, toastheap) AS ( "
						 "  SELECT indexrelid, indrelid, 0::oid "
						 "  FROM pg_catalog.pg_index "
						 "  WHERE indisvalid AND indisready "
						 "    AND indrelid IN "
						 "        (SELECT reloid FROM regular_heap "
						 "         UNION ALL "
						 "         SELECT reloid FROM toast_heap)) ");

	/*
	 * 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.
	 */
	appendPQExpBufferStr(&query,
						 "SELECT all_rels.*, n.nspname, c.relname, "
						 "  c.relfilenode, c.reltablespace, "
						 "  pg_catalog.pg_tablespace_location(t.oid) AS spclocation "
						 "FROM (SELECT * FROM regular_heap "
						 "      UNION ALL "
						 "      SELECT * FROM toast_heap "
						 "      UNION ALL "
						 "      SELECT * FROM all_index) all_rels "
						 "  JOIN pg_catalog.pg_class c "
						 "      ON all_rels.reloid = c.oid "
						 "  JOIN pg_catalog.pg_namespace n "
						 "     ON c.relnamespace = n.oid "
						 "  LEFT OUTER JOIN pg_catalog.pg_tablespace t "
						 "     ON c.reltablespace = t.oid "
						 "ORDER BY 1");

	return query.data;
}

/*
 * Callback function for processing results of the query returned by
 * get_rel_infos_query(), which is used for get_db_rel_and_slot_infos()'s
 * UpgradeTask.  This function stores the relation information for later use.
 */
static void
process_rel_infos(DbInfo *dbinfo, PGresult *res, void *arg)
{
	int			ntups = PQntuples(res);
	RelInfo    *relinfos = pg_malloc_array(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 (int relnum = 0; relnum < ntups; relnum++)
	{
		RelInfo    *curr = &relinfos[num_rels++];

		curr->reloid = atooid(PQgetvalue(res, relnum, i_reloid));
		curr->indtable = atooid(PQgetvalue(res, relnum, i_indtable));
		curr->toastheap = atooid(PQgetvalue(res, relnum, i_toastheap));

		nspname = PQgetvalue(res, relnum, i_nspname);
		curr->nsp_alloc = false;

		/*
		 * Many of the namespace and tablespace strings are identical, so we
		 * try to reuse the allocated string pointers where possible to reduce
		 * memory consumption.
		 */
		/* Can we reuse the previous string allocation? */
		if (last_namespace && strcmp(nspname, last_namespace) == 0)
			curr->nspname = last_namespace;
		else
		{
			last_namespace = curr->nspname = pg_strdup(nspname);
			curr->nsp_alloc = true;
		}

		relname = PQgetvalue(res, relnum, i_relname);
		curr->relname = pg_strdup(relname);

		curr->relfilenumber = atooid(PQgetvalue(res, relnum, i_relfilenumber));
		curr->tblsp_alloc = false;

		/* Is the tablespace oid non-default? */
		if (atooid(PQgetvalue(res, relnum, i_reltablespace)) != 0)
		{
			char	   *spcloc = PQgetvalue(res, relnum, i_spclocation);
			bool		inplace = spcloc[0] && !is_absolute_path(spcloc);

			/*
			 * The tablespace location might be "", meaning the cluster
			 * default location, i.e. pg_default or pg_global.  For in-place
			 * tablespaces, pg_tablespace_location() returns a path relative
			 * to the data directory.
			 */
			if (inplace)
				tablespace = psprintf("%s/%s",
									  os_info.running_cluster->pgdata,
									  spcloc);
			else
				tablespace = spcloc;

			/* Can we reuse the previous string allocation? */
			if (last_tablespace && strcmp(tablespace, last_tablespace) == 0)
				curr->tablespace = last_tablespace;
			else
			{
				last_tablespace = curr->tablespace = pg_strdup(tablespace);
				curr->tblsp_alloc = true;
			}

			/* Free palloc'd string for in-place tablespaces. */
			if (inplace)
				pfree(tablespace);
		}
		else
			/* A zero reltablespace oid indicates the database tablespace. */
			curr->tablespace = dbinfo->db_tablespace;
	}

	dbinfo->rel_arr.rels = relinfos;
	dbinfo->rel_arr.nrels = num_rels;
}

/*
 * get_old_cluster_logical_slot_infos_query()
 *
 * Returns the query for retrieving the logical slot information for all the
 * logical replication slots in the database, for use by
 * get_db_rel_and_slot_infos()'s UpgradeTask.  The status of each logical slot
 * is checked in check_old_cluster_for_valid_slots().
 */
static const char *
get_old_cluster_logical_slot_infos_query(ClusterInfo *cluster)
{
	/*
	 * Fetch the logical replication slot information. The check whether the
	 * slot is considered caught up is done by an upgrade function. This
	 * regards the slot as caught up if we don't find any decodable changes.
	 * The implementation of this check varies depending on the server
	 * version.
	 *
	 * We intentionally skip checking the WALs for invalidated slots as the
	 * corresponding WALs could have been removed for such slots.
	 *
	 * The temporary slots are explicitly ignored while checking because such
	 * slots cannot exist after the upgrade. During the upgrade, clusters are
	 * started and stopped several times causing any temporary slots to be
	 * removed.
	 */

	if (user_opts.live_check)
	{
		/*
		 * We skip the caught-up check during live_check. We cannot verify
		 * whether the slot is caught up in this mode, as new WAL records
		 * could be generated concurrently.
		 */
		return "SELECT slot_name, plugin, two_phase, failover, "
			"FALSE 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";
	}
	else if (GET_MAJOR_VERSION(cluster->major_version) >= 1900)
	{
		/*
		 * For PG19 and later, we optimize the slot caught-up check to avoid
		 * reading the same WAL stream multiple times: execute the caught-up
		 * check only for the slot with the minimum confirmed_flush_lsn, and
		 * apply the same result to all other slots in the same database. This
		 * limits the check to at most one logical slot per database. We also
		 * use the maximum confirmed_flush_lsn among all logical slots on the
		 * database as an early scan cutoff; finding a decodable WAL record
		 * beyond this point implies that no slot has caught up.
		 *
		 * Note that we don't distinguish slots based on their output plugin.
		 * If a plugin applies replication origin filters, we might get a
		 * false positive (i.e., erroneously considering a slot caught up).
		 * However, such cases are very rare, and the impact of a false
		 * positive is minimal.
		 */
		return "WITH check_caught_up AS ( "
			"  SELECT pg_catalog.binary_upgrade_check_logical_slot_pending_wal(slot_name, "
			"    MAX(confirmed_flush_lsn) OVER ()) as last_pending_wal "
			"  FROM pg_replication_slots "
			"  WHERE slot_type = 'logical' AND "
			"    database = current_database() AND "
			"    temporary IS FALSE AND "
			"    invalidation_reason IS NULL "
			"  ORDER BY confirmed_flush_lsn ASC "
			"  LIMIT 1 "
			") "
			"SELECT slot_name, plugin, two_phase, failover, "
			"CASE WHEN invalidation_reason IS NOT NULL THEN FALSE "
			"ELSE  last_pending_wal IS NULL OR "
			"  confirmed_flush_lsn > last_pending_wal "
			"END as caught_up, "
			"invalidation_reason IS NOT NULL as invalid "
			"FROM pg_catalog.pg_replication_slots "
			"LEFT JOIN check_caught_up ON true "
			"WHERE slot_type = 'logical' AND "
			"database = current_database() AND "
			"temporary IS FALSE ";
	}

	/*
	 * For PG18 and earlier, we call
	 * binary_upgrade_logical_slot_has_caught_up() for each logical slot.
	 */
	return "SELECT slot_name, plugin, two_phase, failover, "
		"CASE WHEN invalidation_reason IS NOT NULL THEN FALSE "
		"ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) "
		"END 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 ";
}

/*
 * Callback function for processing results of the query, which is used for
 * get_db_rel_and_slot_infos()'s UpgradeTask.  This function stores the logical
 * slot information for later use.
 */
static void
process_old_cluster_logical_slot_infos(DbInfo *dbinfo, PGresult *res, void *arg)
{
	LogicalSlotInfo *slotinfos = NULL;
	int			num_slots = PQntuples(res);

	if (num_slots)
	{
		int			i_slotname;
		int			i_plugin;
		int			i_twophase;
		int			i_failover;
		int			i_caught_up;
		int			i_invalid;

		slotinfos = pg_malloc_array(LogicalSlotInfo, num_slots);

		i_slotname = PQfnumber(res, "slot_name");
		i_plugin = PQfnumber(res, "plugin");
		i_twophase = PQfnumber(res, "two_phase");
		i_failover = PQfnumber(res, "failover");
		i_caught_up = PQfnumber(res, "caught_up");
		i_invalid = PQfnumber(res, "invalid");

		for (int slotnum = 0; slotnum < num_slots; slotnum++)
		{
			LogicalSlotInfo *curr = &slotinfos[slotnum];

			curr->slotname = pg_strdup(PQgetvalue(res, slotnum, i_slotname));
			curr->plugin = pg_strdup(PQgetvalue(res, slotnum, i_plugin));
			curr->two_phase = (strcmp(PQgetvalue(res, slotnum, i_twophase), "t") == 0);
			curr->failover = (strcmp(PQgetvalue(res, slotnum, i_failover), "t") == 0);
			curr->caught_up = (strcmp(PQgetvalue(res, slotnum, i_caught_up), "t") == 0);
			curr->invalid = (strcmp(PQgetvalue(res, slotnum, i_invalid), "t") == 0);
		}
	}

	dbinfo->slot_arr.slots = slotinfos;
	dbinfo->slot_arr.nslots = num_slots;
}


/*
 * count_old_cluster_logical_slots()
 *
 * Returns the number of logical replication slots for all databases.
 *
 * Note: this function always returns 0 if the old_cluster is PG16 and prior
 * because we gather slot information only for cluster versions greater than or
 * equal to PG17. See get_db_rel_and_slot_infos().
 */
int
count_old_cluster_logical_slots(void)
{
	int			slot_count = 0;

	for (int dbnum = 0; dbnum < old_cluster.dbarr.ndbs; dbnum++)
		slot_count += old_cluster.dbarr.dbs[dbnum].slot_arr.nslots;

	return slot_count;
}

/*
 * get_subscription_info()
 *
 * Gets the information of subscriptions in the cluster.
 */
void
get_subscription_info(ClusterInfo *cluster)
{
	PGconn	   *conn;
	PGresult   *res;
	int			i_nsub;
	int			i_retain_dead_tuples;

	conn = connectToServer(cluster, "template1");
	if (GET_MAJOR_VERSION(cluster->major_version) >= 1900)
		res = executeQueryOrDie(conn, "SELECT count(*) AS nsub,"
								"COUNT(CASE WHEN subretaindeadtuples THEN 1 END) > 0 AS retain_dead_tuples "
								"FROM pg_catalog.pg_subscription");
	else
		res = executeQueryOrDie(conn, "SELECT count(*) AS nsub,"
								"'f' AS retain_dead_tuples "
								"FROM pg_catalog.pg_subscription");

	i_nsub = PQfnumber(res, "nsub");
	i_retain_dead_tuples = PQfnumber(res, "retain_dead_tuples");

	cluster->nsubs = atoi(PQgetvalue(res, 0, i_nsub));
	cluster->sub_retain_dead_tuples = (strcmp(PQgetvalue(res, 0, i_retain_dead_tuples), "t") == 0);

	PQclear(res);
	PQfinish(conn);
}

static void
free_db_and_rel_infos(DbInfoArr *db_arr)
{
	int			dbnum;

	for (dbnum = 0; dbnum < db_arr->ndbs; dbnum++)
	{
		free_rel_infos(&db_arr->dbs[dbnum].rel_arr);
		pg_free(db_arr->dbs[dbnum].db_name);
	}
	pg_free(db_arr->dbs);
	db_arr->dbs = NULL;
	db_arr->ndbs = 0;
}


static void
free_rel_infos(RelInfoArr *rel_arr)
{
	int			relnum;

	for (relnum = 0; relnum < rel_arr->nrels; relnum++)
	{
		if (rel_arr->rels[relnum].nsp_alloc)
			pg_free(rel_arr->rels[relnum].nspname);
		pg_free(rel_arr->rels[relnum].relname);
		if (rel_arr->rels[relnum].tblsp_alloc)
			pg_free(rel_arr->rels[relnum].tablespace);
	}
	pg_free(rel_arr->rels);
	rel_arr->nrels = 0;
}


static void
print_db_infos(DbInfoArr *db_arr)
{
	int			dbnum;

	for (dbnum = 0; dbnum < db_arr->ndbs; dbnum++)
	{
		DbInfo	   *pDbInfo = &db_arr->dbs[dbnum];

		pg_log(PG_VERBOSE, "Database: \"%s\"", pDbInfo->db_name);
		print_rel_infos(&pDbInfo->rel_arr);
		print_slot_infos(&pDbInfo->slot_arr);
	}
}


static void
print_rel_infos(RelInfoArr *rel_arr)
{
	int			relnum;

	for (relnum = 0; relnum < rel_arr->nrels; relnum++)
		pg_log(PG_VERBOSE, "relname: \"%s.%s\", reloid: %u, reltblspace: \"%s\"",
			   rel_arr->rels[relnum].nspname,
			   rel_arr->rels[relnum].relname,
			   rel_arr->rels[relnum].reloid,
			   rel_arr->rels[relnum].tablespace);
}

static void
print_slot_infos(LogicalSlotInfoArr *slot_arr)
{
	/* Quick return if there are no logical slots. */
	if (slot_arr->nslots == 0)
		return;

	pg_log(PG_VERBOSE, "Logical replication slots in the database:");

	for (int slotnum = 0; slotnum < slot_arr->nslots; slotnum++)
	{
		LogicalSlotInfo *slot_info = &slot_arr->slots[slotnum];

		pg_log(PG_VERBOSE, "slot name: \"%s\", output plugin: \"%s\", two_phase: %s",
			   slot_info->slotname,
			   slot_info->plugin,
			   slot_info->two_phase ? "true" : "false");
	}
}