On Mon, 13 Nov 2023 at 13:52, Peter Smith <smithpb2...@gmail.com> wrote:
>
> Here are some review comments for patch v13-0001
>
> ======
> src/bin/pg_dump/pg_dump.c
>
> 1. getSubscriptionTables
>
> + int i_srsublsn;
> + int i;
> + int cur_rel = 0;
> + int ntups;
>
> What is the difference between 'i' and 'cur_rel'?
>
> AFAICT these represent the same tuple index, in which case you might
> as well throw away 'cur_rel' and only keep 'i'.
Modified
> ~~~
>
> 2. getSubscriptionTables
>
> + for (i = 0; i < ntups; i++)
> + {
> + Oid cur_srsubid = atooid(PQgetvalue(res, i, i_srsubid));
> + Oid relid = atooid(PQgetvalue(res, i, i_srrelid));
> + TableInfo *tblinfo;
>
> Since this is all new code, using C99 style for loop variable
> declaration of 'i' will be better.
Modified
> ======
> src/bin/pg_upgrade/check.c
>
> 3. check_for_subscription_state
>
> +check_for_subscription_state(ClusterInfo *cluster)
> +{
> + int dbnum;
> + FILE *script = NULL;
> + char output_path[MAXPGPATH];
> + int ntup;
> +
> + /* Subscription relations state can be migrated since PG17. */
> + if (GET_MAJOR_VERSION(old_cluster.major_version) < 1700)
> + return;
> +
> + prep_status("Checking for subscription state");
> +
> + snprintf(output_path, sizeof(output_path), "%s/%s",
> + log_opts.basedir,
> + "subscription_state.txt");
>
> I felt this filename ought to be more like
> 'subscriptions_with_bad_state.txt' because the current name looks like
> a normal logfile with nothing to indicate that it is only for the
> states of the "bad" subscriptions.
I have kept the file name intentionally shorted as we noticed that
when the upgrade of the publisher patch used a longer name there were
some buildfarm failures because of longer names.
> ~~~
>
> 4.
> + for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
> + {
>
> Since this is all new code, using C99 style for loop variable
> declaration of 'dbnum' will be better.
Modified
> ~~~
>
> 5.
> + * 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
> + * tracked by the publisher does not match anymore.
>
> missing whitespace
>
> /SUBREL_STATE_DATASYNC:A relation/SUBREL_STATE_DATASYNC: A relation/
Modified
Also added a couple of missing test cases. The attached v14 version
patch has the changes for the same.
Regards,
Vignesh
From 354137c80dfacc30bd0fa85c2f993f34ae5af4b9 Mon Sep 17 00:00:00 2001
From: Vignesh C <vignes...@gmail.com>
Date: Mon, 30 Oct 2023 12:31:59 +0530
Subject: [PATCH v14] Preserve the full subscription's state during pg_upgrade
Previously, only the subscription metadata information was preserved. Without
the list of relations and their state it's impossible to re-enable the
subscriptions without missing some records as the list of relations can only be
refreshed after enabling the subscription (and therefore starting the apply
worker). Even if we added a way to refresh the subscription while enabling a
publication, we still wouldn't know which relations are new on the publication
side, and therefore should be fully synced, and which shouldn't.
To fix this problem, this patch teaches pg_dump to restore the content of
pg_subscription_rel from the old cluster by using
binary_upgrade_add_sub_rel_state SQL function. This is supported only
in binary upgrade mode.
The new SQL binary_upgrade_add_sub_rel_state function has the following
syntax:
SELECT binary_upgrade_add_sub_rel_state(subname text, relid oid, state char [,sublsn pg_lsn])
In the above, subname is the subscription name, relid is the relation
identifier, the state is the state of the relation, sublsn is subscription lsn
which is optional, and defaults to NULL/InvalidXLogRecPtr if not provided.
pg_dump will retrieve these values(subname, relid, state and sublsn) from the
old cluster.
The subscription's replication origin is needed to ensure that we don't
replicate anything twice.
To fix this problem, this patch teaches pg_dump to update the replication
origin along with create subscription by using
binary_upgrade_replorigin_advance SQL function to restore the
underlying replication origin remote LSN. This is supported only in
binary upgrade mode.
The new SQL binary_upgrade_replorigin_advance function has the following
syntax:
SELECT binary_upgrade_replorigin_advance(subname text, sublsn pg_lsn)
In the above, subname is the subscription name and sublsn is subscription lsn.
pg_dump will retrieve these values(subname and sublsn) from the old cluster.
pg_upgrade will check that all the subscription relations are in 'i' (init), 's' (data sync) or in 'r' (ready) state, and
will error out if that's not the case, logging the reason for the failure.
Author: Julien Rouhaud, Vignesh C
Reviewed-by: FIXME
Discussion: https://postgr.es/m/20230217075433.u5mjly4d5cr4hcfe@jrouhaud
---
doc/src/sgml/ref/pgupgrade.sgml | 72 ++++
src/backend/utils/adt/pg_upgrade_support.c | 130 +++++++
src/bin/pg_dump/common.c | 22 ++
src/bin/pg_dump/pg_dump.c | 197 ++++++++++-
src/bin/pg_dump/pg_dump.h | 16 +
src/bin/pg_dump/pg_dump_sort.c | 11 +-
src/bin/pg_upgrade/check.c | 169 ++++++++-
src/bin/pg_upgrade/info.c | 25 ++
src/bin/pg_upgrade/meson.build | 1 +
src/bin/pg_upgrade/pg_upgrade.h | 1 +
src/bin/pg_upgrade/t/004_subscription.pl | 392 +++++++++++++++++++++
src/include/catalog/pg_proc.dat | 10 +
src/tools/pgindent/typedefs.list | 1 +
13 files changed, 1024 insertions(+), 23 deletions(-)
create mode 100644 src/bin/pg_upgrade/t/004_subscription.pl
diff --git a/doc/src/sgml/ref/pgupgrade.sgml b/doc/src/sgml/ref/pgupgrade.sgml
index 46e8a0b746..b824097e87 100644
--- a/doc/src/sgml/ref/pgupgrade.sgml
+++ b/doc/src/sgml/ref/pgupgrade.sgml
@@ -456,6 +456,78 @@ make prefix=/usr/local/pgsql.new install
</step>
+ <step>
+ <title>Prepare for subscriber upgrades</title>
+
+ <para>
+ Setup the <link linkend="logical-replication-config-subscriber">
+ subscriber configurations</link> in the new subscriber.
+ <application>pg_upgrade</application> attempts to migrate subscription
+ dependencies which includes the subscription table information present in
+ <link linkend="catalog-pg-subscription-rel">pg_subscription_rel</link>
+ system catalog and also the subscription replication origin. This allows
+ logical replication on the new subscriber to continue from where the
+ old subscriber was up to. Migration of subscription dependencies is only
+ supported when the old cluster is version 17.0 or later. Subscription
+ dependencies on clusters before version 17.0 will silently be ignored.
+ </para>
+
+ <para>
+ There are some prerequisites for <application>pg_upgrade</application> to
+ be able to upgrade the subscriptions. If these are not met an error
+ will be reported.
+ </para>
+
+ <itemizedlist>
+ <listitem>
+ <para>
+ All the subscription tables in the old subscriber should be in state
+ <literal>i</literal> (initialize), <literal>r</literal> (ready) or
+ <literal>s</literal> (synchronized). This can be verified by checking
+ <link linkend="catalog-pg-subscription-rel">pg_subscription_rel</link>.<structfield>srsubstate</structfield>.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The replication origin entry corresponding to each of the subscriptions
+ should exist in the old cluster. This can be found by checking
+ <link linkend="catalog-pg-subscription">pg_subscription</link> and
+ <link linkend="catalog-pg-replication-origin">pg_replication_origin</link>
+ system tables.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The new cluster must have
+ <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link>
+ configured to a value greater than or equal to the number of
+ subscriptions present in the old cluster.
+ </para>
+ </listitem>
+ </itemizedlist>
+
+ <para>
+ The subscriptions will be migrated to the new cluster in a disabled state.
+ After migration, do this:
+ </para>
+
+ <itemizedlist>
+ <listitem>
+ <para>
+ Enable the subscriptions by executing
+ <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ... ENABLE</command></link>.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ Create all the new tables that were created in the publication during
+ upgrade and refresh the publication by executing
+ <link linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command></link>.
+ </para>
+ </listitem>
+ </itemizedlist>
+ </step>
+
<step>
<title>Stop both servers</title>
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index 2f6fc86c3d..75d77d8e22 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -11,15 +11,22 @@
#include "postgres.h"
+#include "access/table.h"
#include "catalog/binary_upgrade.h"
#include "catalog/heap.h"
#include "catalog/namespace.h"
+#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "commands/extension.h"
#include "miscadmin.h"
#include "replication/logical.h"
+#include "replication/origin.h"
+#include "replication/worker_internal.h"
+#include "storage/lmgr.h"
#include "utils/array.h"
#include "utils/builtins.h"
+#include "utils/pg_lsn.h"
+#include "utils/syscache.h"
#define CHECK_IS_BINARY_UPGRADE \
@@ -305,3 +312,126 @@ binary_upgrade_logical_slot_has_caught_up(PG_FUNCTION_ARGS)
PG_RETURN_BOOL(!found_pending_wal);
}
+
+/*
+ * binary_upgrade_add_sub_rel_state
+ *
+ * Add the relation with the specified relation state to pg_subscription_rel
+ * catalog.
+ */
+Datum
+binary_upgrade_add_sub_rel_state(PG_FUNCTION_ARGS)
+{
+ Relation rel;
+ HeapTuple tup;
+ Oid subid;
+ Form_pg_subscription form;
+ char *subname;
+ Oid relid;
+ char relstate;
+ XLogRecPtr sublsn;
+
+ CHECK_IS_BINARY_UPGRADE;
+
+ /* We must check these things before dereferencing the arguments */
+ if (PG_ARGISNULL(0) || PG_ARGISNULL(1) || PG_ARGISNULL(2))
+ elog(ERROR, "null argument to binary_upgrade_add_sub_rel_state is not allowed");
+
+ subname = text_to_cstring(PG_GETARG_TEXT_PP(0));
+ relid = PG_GETARG_OID(1);
+ relstate = PG_GETARG_CHAR(2);
+ sublsn = PG_ARGISNULL(3) ? InvalidXLogRecPtr : PG_GETARG_LSN(3);
+
+ if (!OidIsValid(relid))
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("invalid relation identifier used: %u", relid));
+
+ tup = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+ if (!HeapTupleIsValid(tup))
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("relation %u does not exist", relid));
+ ReleaseSysCache(tup);
+
+ rel = table_open(SubscriptionRelationId, RowExclusiveLock);
+
+ /* Fetch the existing tuple. */
+ tup = SearchSysCache2(SUBSCRIPTIONNAME, MyDatabaseId,
+ CStringGetDatum(subname));
+ if (!HeapTupleIsValid(tup))
+ ereport(ERROR,
+ errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("subscription \"%s\" does not exist", subname));
+
+ form = (Form_pg_subscription) GETSTRUCT(tup);
+ subid = form->oid;
+
+ AddSubscriptionRelState(subid, relid, relstate, sublsn);
+
+ ReleaseSysCache(tup);
+ table_close(rel, RowExclusiveLock);
+
+ PG_RETURN_VOID();
+}
+
+/*
+ * binary_upgrade_replorigin_advance
+ *
+ * Update the remote_lsn for the subscriber's replication origin.
+ */
+Datum
+binary_upgrade_replorigin_advance(PG_FUNCTION_ARGS)
+{
+ Relation rel;
+ HeapTuple tup;
+ Oid subid;
+ Form_pg_subscription form;
+ char *subname;
+ XLogRecPtr sublsn;
+ char originname[NAMEDATALEN];
+ RepOriginId originid;
+
+ CHECK_IS_BINARY_UPGRADE;
+
+ /* We must check these things before dereferencing the arguments */
+ if (PG_ARGISNULL(0))
+ elog(ERROR, "null argument to binary_upgrade_replorigin_advance is not allowed");
+
+ subname = text_to_cstring(PG_GETARG_TEXT_PP(0));
+ sublsn = PG_ARGISNULL(1) ? InvalidXLogRecPtr : PG_GETARG_LSN(1);
+
+ rel = table_open(SubscriptionRelationId, RowExclusiveLock);
+
+ /* Fetch the existing tuple. */
+ tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
+ CStringGetDatum(subname));
+ if (!HeapTupleIsValid(tup))
+ ereport(ERROR,
+ errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("subscription \"%s\" does not exist", subname));
+
+ form = (Form_pg_subscription) GETSTRUCT(tup);
+ subid = form->oid;
+
+ ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
+
+ /* lock to prevent the replication origin from vanishing */
+ LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
+ originid = replorigin_by_name(originname, false);
+
+ /*
+ * The server will be stopped after setting up the objects in the new
+ * cluster. Shutdown server will flush the origins during shutdown
+ * checkpoint.
+ */
+ replorigin_advance(originid, sublsn, InvalidXLogRecPtr,
+ false /* backward */ ,
+ false /* WAL log */ );
+
+ UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
+ heap_freetuple(tup);
+ table_close(rel, RowExclusiveLock);
+
+ PG_RETURN_VOID();
+}
diff --git a/src/bin/pg_dump/common.c b/src/bin/pg_dump/common.c
index 8b0c1e7b53..764a39fcb9 100644
--- a/src/bin/pg_dump/common.c
+++ b/src/bin/pg_dump/common.c
@@ -24,6 +24,7 @@
#include "catalog/pg_operator_d.h"
#include "catalog/pg_proc_d.h"
#include "catalog/pg_publication_d.h"
+#include "catalog/pg_subscription_d.h"
#include "catalog/pg_type_d.h"
#include "common/hashfn.h"
#include "fe_utils/string_utils.h"
@@ -265,6 +266,9 @@ getSchemaData(Archive *fout, int *numTablesPtr)
pg_log_info("reading subscriptions");
getSubscriptions(fout);
+ pg_log_info("reading subscription membership of tables");
+ getSubscriptionTables(fout);
+
free(inhinfo); /* not needed any longer */
*numTablesPtr = numTables;
@@ -978,6 +982,24 @@ findPublicationByOid(Oid oid)
return (PublicationInfo *) dobj;
}
+/*
+ * findSubscriptionByOid
+ * finds the DumpableObject for the subscription with the given oid
+ * returns NULL if not found
+ */
+SubscriptionInfo *
+findSubscriptionByOid(Oid oid)
+{
+ CatalogId catId;
+ DumpableObject *dobj;
+
+ catId.tableoid = SubscriptionRelationId;
+ catId.oid = oid;
+ dobj = findObjectByCatalogId(catId);
+ Assert(dobj == NULL || dobj->objType == DO_SUBSCRIPTION);
+ return (SubscriptionInfo *) dobj;
+}
+
/*
* recordExtensionMembership
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index e863913849..eb48f12fbe 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -296,6 +296,7 @@ static void dumpPolicy(Archive *fout, const PolicyInfo *polinfo);
static void dumpPublication(Archive *fout, const PublicationInfo *pubinfo);
static void dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo);
static void dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo);
+static void dumpSubscriptionTable(Archive *fout, const SubRelInfo *subrinfo);
static void dumpDatabase(Archive *fout);
static void dumpDatabaseConfig(Archive *AH, PQExpBuffer outbuf,
const char *dbname, Oid dboid);
@@ -4581,6 +4582,95 @@ is_superuser(Archive *fout)
return false;
}
+/*
+ * getSubscriptionTables
+ * Get information about subscription membership for dumpable tables. This
+ * will be used only in binary-upgrade mode.
+ */
+void
+getSubscriptionTables(Archive *fout)
+{
+ DumpOptions *dopt = fout->dopt;
+ SubscriptionInfo *subinfo = NULL;
+ SubRelInfo *subrinfo;
+ PQExpBuffer query;
+ PGresult *res;
+ int i_srsubid;
+ int i_srrelid;
+ int i_srsubstate;
+ int i_srsublsn;
+ int ntups;
+ Oid last_srsubid = InvalidOid;
+
+ if (dopt->no_subscriptions || !dopt->binary_upgrade ||
+ fout->remoteVersion < 170000)
+ return;
+
+ query = createPQExpBuffer();
+ appendPQExpBuffer(query, "SELECT srsubid, srrelid, srsubstate, srsublsn"
+ " FROM pg_catalog.pg_subscription_rel"
+ " ORDER BY srsubid");
+ res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
+
+ ntups = PQntuples(res);
+ if (ntups == 0)
+ goto cleanup;
+
+ /* Get pg_subscription_rel attributes */
+ i_srsubid = PQfnumber(res, "srsubid");
+ i_srrelid = PQfnumber(res, "srrelid");
+ i_srsubstate = PQfnumber(res, "srsubstate");
+ i_srsublsn = PQfnumber(res, "srsublsn");
+
+ subrinfo = pg_malloc(ntups * sizeof(SubRelInfo));
+ for (int i = 0; i < ntups; i++)
+ {
+ Oid cur_srsubid = atooid(PQgetvalue(res, i, i_srsubid));
+ Oid relid = atooid(PQgetvalue(res, i, i_srrelid));
+ TableInfo *tblinfo;
+
+ /*
+ * If we switched to a new subscription, check if the subscription
+ * exists.
+ */
+ if (cur_srsubid != last_srsubid)
+ {
+ subinfo = findSubscriptionByOid(cur_srsubid);
+ if (subinfo == NULL)
+ pg_fatal("subscription with OID %u does not exist", cur_srsubid);
+
+ last_srsubid = cur_srsubid;
+ }
+
+ tblinfo = findTableByOid(relid);
+ if (tblinfo == NULL)
+ pg_fatal("failed sanity check, table with OID %u not found",
+ relid);
+
+ /* OK, make a DumpableObject for this relationship */
+ subrinfo[i].dobj.objType = DO_SUBSCRIPTION_REL;
+ subrinfo[i].dobj.catId.tableoid = relid;
+ subrinfo[i].dobj.catId.oid = cur_srsubid;
+ AssignDumpId(&subrinfo[i].dobj);
+ subrinfo[i].dobj.name = pg_strdup(subinfo->dobj.name);
+ subrinfo[i].tblinfo = tblinfo;
+ subrinfo[i].srsubstate = PQgetvalue(res, i, i_srsubstate)[0];
+ if (PQgetisnull(res, i, i_srsublsn))
+ subrinfo[i].srsublsn = NULL;
+ else
+ subrinfo[i].srsublsn = pg_strdup(PQgetvalue(res, i, i_srsublsn));
+
+ subrinfo[i].subinfo = subinfo;
+
+ /* Decide whether we want to dump it */
+ selectDumpableObject(&(subrinfo[i].dobj), fout);
+ }
+
+cleanup:
+ PQclear(res);
+ destroyPQExpBuffer(query);
+}
+
/*
* getSubscriptions
* get information about subscriptions
@@ -4607,6 +4697,7 @@ getSubscriptions(Archive *fout)
int i_subsynccommit;
int i_subpublications;
int i_suborigin;
+ int i_suboriginremotelsn;
int i,
ntups;
@@ -4662,17 +4753,20 @@ getSubscriptions(Archive *fout)
appendPQExpBufferStr(query,
" s.subpasswordrequired,\n"
" s.subrunasowner,\n"
- " s.suborigin\n");
+ " s.suborigin,\n");
else
appendPQExpBuffer(query,
" 't' AS subpasswordrequired,\n"
" 't' AS subrunasowner,\n"
- " '%s' AS suborigin\n",
+ " '%s' AS suborigin,\n",
LOGICALREP_ORIGIN_ANY);
+ appendPQExpBufferStr(query, "o.remote_lsn AS suboriginremotelsn\n");
appendPQExpBufferStr(query,
- "FROM pg_subscription s\n"
- "WHERE s.subdbid = (SELECT oid FROM pg_database\n"
+ "FROM pg_catalog.pg_subscription s\n"
+ "LEFT JOIN pg_catalog.pg_replication_origin_status o \n"
+ " ON o.external_id = 'pg_' || s.oid::text \n"
+ "WHERE s.subdbid = (SELECT oid FROM pg_catalog.pg_database\n"
" WHERE datname = current_database())");
res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
@@ -4698,6 +4792,7 @@ getSubscriptions(Archive *fout)
i_subsynccommit = PQfnumber(res, "subsynccommit");
i_subpublications = PQfnumber(res, "subpublications");
i_suborigin = PQfnumber(res, "suborigin");
+ i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4735,6 +4830,11 @@ getSubscriptions(Archive *fout)
subinfo[i].subpublications =
pg_strdup(PQgetvalue(res, i, i_subpublications));
subinfo[i].suborigin = pg_strdup(PQgetvalue(res, i, i_suborigin));
+ if (PQgetisnull(res, i, i_suboriginremotelsn))
+ subinfo[i].suboriginremotelsn = NULL;
+ else
+ subinfo[i].suboriginremotelsn =
+ pg_strdup(PQgetvalue(res, i, i_suboriginremotelsn));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -4744,6 +4844,80 @@ getSubscriptions(Archive *fout)
destroyPQExpBuffer(query);
}
+/*
+ * dumpSubscriptionTable
+ * Dump the definition of the given subscription table mapping. This will be
+ * used only in binary-upgrade mode.
+ */
+static void
+dumpSubscriptionTable(Archive *fout, const SubRelInfo *subrinfo)
+{
+ DumpOptions *dopt = fout->dopt;
+ SubscriptionInfo *subinfo = subrinfo->subinfo;
+ PQExpBuffer query;
+ char *tag;
+
+ /* Do nothing in data-only dump */
+ if (dopt->dataOnly)
+ return;
+
+ Assert(fout->dopt->binary_upgrade);
+
+ tag = psprintf("%s %s", subinfo->dobj.name, subrinfo->dobj.name);
+
+ query = createPQExpBuffer();
+
+ if (subinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)
+ {
+ /*
+ * binary_upgrade_add_sub_rel_state will add the subscription relation
+ * to pg_subscription_rel table. This will be used only in
+ * binary-upgrade mode.
+ */
+ if (fout->remoteVersion >= 170000)
+ {
+ appendPQExpBufferStr(query,
+ "\n-- For binary upgrade, must preserve the subscriber table.\n");
+ appendPQExpBufferStr(query,
+ "SELECT pg_catalog.binary_upgrade_add_sub_rel_state(");
+ appendStringLiteralAH(query, subrinfo->dobj.name, fout);
+ appendPQExpBuffer(query,
+ ", %u, '%c'",
+ subrinfo->tblinfo->dobj.catId.oid,
+ subrinfo->srsubstate);
+
+ if (subrinfo->srsublsn && subrinfo->srsublsn[0] != '\0')
+ appendPQExpBuffer(query, ", '%s'",
+ subrinfo->srsublsn);
+ else
+ appendPQExpBuffer(query, ", NULL");
+
+ appendPQExpBufferStr(query, ");\n");
+ }
+ }
+
+ /*
+ * There is no point in creating a drop query as the drop is done by table
+ * drop. (If you think to change this, see also _printTocEntry().)
+ * Although this object doesn't really have ownership as such, set the
+ * owner field anyway to ensure that the command is run by the correct
+ * role at restore time.
+ */
+ if (subrinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)
+ ArchiveEntry(fout, subrinfo->dobj.catId, subrinfo->dobj.dumpId,
+ ARCHIVE_OPTS(.tag = tag,
+ .namespace = subrinfo->tblinfo->dobj.namespace->dobj.name,
+ .owner = subinfo->rolname,
+ .description = "SUBSCRIPTION TABLE",
+ .section = SECTION_POST_DATA,
+ .createStmt = query->data));
+
+ /* These objects can't currently have comments or seclabels */
+
+ free(tag);
+ destroyPQExpBuffer(query);
+}
+
/*
* dumpSubscription
* dump the definition of the given subscription
@@ -4824,6 +4998,17 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
appendPQExpBufferStr(query, ");\n");
+ if (dopt->binary_upgrade && fout->remoteVersion >= 170000 &&
+ subinfo->suboriginremotelsn)
+ {
+ appendPQExpBufferStr(query,
+ "\n-- For binary upgrade, must preserve the remote_lsn for the subscriber's replication origin.\n");
+ appendPQExpBufferStr(query,
+ "SELECT pg_catalog.binary_upgrade_replorigin_advance(");
+ appendStringLiteralAH(query, subinfo->dobj.name, fout);
+ appendPQExpBuffer(query, ", '%s');\n", subinfo->suboriginremotelsn);
+ }
+
if (subinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)
ArchiveEntry(fout, subinfo->dobj.catId, subinfo->dobj.dumpId,
ARCHIVE_OPTS(.tag = subinfo->dobj.name,
@@ -10442,6 +10627,9 @@ dumpDumpableObject(Archive *fout, DumpableObject *dobj)
case DO_SUBSCRIPTION:
dumpSubscription(fout, (const SubscriptionInfo *) dobj);
break;
+ case DO_SUBSCRIPTION_REL:
+ dumpSubscriptionTable(fout, (const SubRelInfo *) dobj);
+ break;
case DO_PRE_DATA_BOUNDARY:
case DO_POST_DATA_BOUNDARY:
/* never dumped, nothing to do */
@@ -18508,6 +18696,7 @@ addBoundaryDependencies(DumpableObject **dobjs, int numObjs,
case DO_PUBLICATION_REL:
case DO_PUBLICATION_TABLE_IN_SCHEMA:
case DO_SUBSCRIPTION:
+ case DO_SUBSCRIPTION_REL:
/* Post-data objects: must come after the post-data boundary */
addObjectDependency(dobj, postDataBound->dumpId);
break;
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 2fe3cbed9a..62b3d9249b 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -83,6 +83,7 @@ typedef enum
DO_PUBLICATION_REL,
DO_PUBLICATION_TABLE_IN_SCHEMA,
DO_SUBSCRIPTION,
+ DO_SUBSCRIPTION_REL,
} DumpableObjectType;
/*
@@ -671,8 +672,21 @@ typedef struct _SubscriptionInfo
char *subsynccommit;
char *subpublications;
char *suborigin;
+ char *suboriginremotelsn;
} SubscriptionInfo;
+/*
+ * The SubRelInfo struct is used to represent a subscription relation.
+ */
+typedef struct _SubRelInfo
+{
+ DumpableObject dobj;
+ SubscriptionInfo *subinfo;
+ TableInfo *tblinfo;
+ char srsubstate;
+ char *srsublsn;
+} SubRelInfo;
+
/*
* common utility functions
*/
@@ -697,6 +711,7 @@ extern CollInfo *findCollationByOid(Oid oid);
extern NamespaceInfo *findNamespaceByOid(Oid oid);
extern ExtensionInfo *findExtensionByOid(Oid oid);
extern PublicationInfo *findPublicationByOid(Oid oid);
+extern SubscriptionInfo *findSubscriptionByOid(Oid oid);
extern void recordExtensionMembership(CatalogId catId, ExtensionInfo *ext);
extern ExtensionInfo *findOwningExtension(CatalogId catalogId);
@@ -756,5 +771,6 @@ extern void getPublicationNamespaces(Archive *fout);
extern void getPublicationTables(Archive *fout, TableInfo tblinfo[],
int numTables);
extern void getSubscriptions(Archive *fout);
+extern void getSubscriptionTables(Archive *fout);
#endif /* PG_DUMP_H */
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index abfea15c09..e8d9c8ac86 100644
--- a/src/bin/pg_dump/pg_dump_sort.c
+++ b/src/bin/pg_dump/pg_dump_sort.c
@@ -94,6 +94,7 @@ enum dbObjectTypePriorities
PRIO_PUBLICATION_REL,
PRIO_PUBLICATION_TABLE_IN_SCHEMA,
PRIO_SUBSCRIPTION,
+ PRIO_SUBSCRIPTION_REL,
PRIO_DEFAULT_ACL, /* done in ACL pass */
PRIO_EVENT_TRIGGER, /* must be next to last! */
PRIO_REFRESH_MATVIEW /* must be last! */
@@ -147,10 +148,11 @@ static const int dbObjectTypePriority[] =
PRIO_PUBLICATION, /* DO_PUBLICATION */
PRIO_PUBLICATION_REL, /* DO_PUBLICATION_REL */
PRIO_PUBLICATION_TABLE_IN_SCHEMA, /* DO_PUBLICATION_TABLE_IN_SCHEMA */
- PRIO_SUBSCRIPTION /* DO_SUBSCRIPTION */
+ PRIO_SUBSCRIPTION, /* DO_SUBSCRIPTION */
+ PRIO_SUBSCRIPTION_REL /* DO_SUBSCRIPTION_REL */
};
-StaticAssertDecl(lengthof(dbObjectTypePriority) == (DO_SUBSCRIPTION + 1),
+StaticAssertDecl(lengthof(dbObjectTypePriority) == (DO_SUBSCRIPTION_REL + 1),
"array length mismatch");
static DumpId preDataBoundId;
@@ -1472,6 +1474,11 @@ describeDumpableObject(DumpableObject *obj, char *buf, int bufsize)
"SUBSCRIPTION (ID %d OID %u)",
obj->dumpId, obj->catId.oid);
return;
+ case DO_SUBSCRIPTION_REL:
+ snprintf(buf, bufsize,
+ "SUBSCRIPTION TABLE (ID %d OID %u)",
+ obj->dumpId, obj->catId.oid);
+ return;
case DO_PRE_DATA_BOUNDARY:
snprintf(buf, bufsize,
"PRE-DATA BOUNDARY (ID %d)",
diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index fa52aa2c22..9e89c3b2eb 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -20,6 +20,7 @@ static void check_is_install_user(ClusterInfo *cluster);
static void check_proper_datallowconn(ClusterInfo *cluster);
static void check_for_prepared_transactions(ClusterInfo *cluster);
static void check_for_isn_and_int8_passing_mismatch(ClusterInfo *cluster);
+static void check_for_subscription_state(ClusterInfo *cluster);
static void check_for_user_defined_postfix_ops(ClusterInfo *cluster);
static void check_for_incompatible_polymorphics(ClusterInfo *cluster);
static void check_for_tables_with_oids(ClusterInfo *cluster);
@@ -111,6 +112,7 @@ check_and_dump_old_cluster(bool live_check)
check_for_composite_data_type_usage(&old_cluster);
check_for_reg_data_type_usage(&old_cluster);
check_for_isn_and_int8_passing_mismatch(&old_cluster);
+ check_for_subscription_state(&old_cluster);
/*
* Logical replication slots can be migrated since PG17. See comments atop
@@ -812,6 +814,126 @@ check_for_isn_and_int8_passing_mismatch(ClusterInfo *cluster)
check_ok();
}
+/*
+ * check_for_subscription_state()
+ *
+ * Verify that each of the subscriptions has all their corresponding tables in
+ * i (initialize), r (ready) or s (synchronized) state.
+ */
+static void
+check_for_subscription_state(ClusterInfo *cluster)
+{
+ FILE *script = NULL;
+ char output_path[MAXPGPATH];
+ int ntup;
+
+ /* Subscription relations state can be migrated since PG17. */
+ if (GET_MAJOR_VERSION(old_cluster.major_version) < 1700)
+ return;
+
+ prep_status("Checking for subscription state");
+
+ snprintf(output_path, sizeof(output_path), "%s/%s",
+ log_opts.basedir,
+ "subscription_state.txt");
+ for (int dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
+ {
+ PGresult *res;
+ DbInfo *active_db = &cluster->dbarr.dbs[dbnum];
+ PGconn *conn = connectToServer(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\": %s",
+ output_path, strerror(errno));
+ fprintf(script, "replication origin is missing for database:%s subscription:%s\n",
+ PQgetvalue(res, i, 0),
+ PQgetvalue(res, i, 1));
+ }
+ PQclear(res);
+ }
+
+ /*
+ * A slot not created yet refers to the 'i' (initialize) state, while
+ * 'r' (ready) and 's' (synchronized) states refer to a slot created
+ * previously but already dropped. These states are supported states
+ * for upgrade. The other states listed below are not ok:
+ *
+ * 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
+ * tracked by the publisher does not match anymore.
+ *
+ * b) 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.
+ *
+ * c) 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 s.subname, n.nspname, c.relname, r.srsubstate "
+ "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', 's') "
+ "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\": %s",
+ output_path, strerror(errno));
+
+ fprintf(script, "database:%s subscription:%s schema:%s relation:%s state:%s not in required state\n",
+ active_db->db_name,
+ PQgetvalue(res, i, 0),
+ PQgetvalue(res, i, 1),
+ PQgetvalue(res, i, 2),
+ PQgetvalue(res, i, 3));
+ }
+
+ PQclear(res);
+ PQfinish(conn);
+ }
+
+ if (script)
+ {
+ fclose(script);
+ pg_log(PG_REPORT, "fatal");
+ pg_fatal("Your installation contains subscriptions without origin or having relations not in i (initialize), r (ready) or s (synchronized) state.\n"
+ "A list of the problem subscriptions is in the file:\n"
+ " %s", output_path);
+ }
+ else
+ check_ok();
+}
+
/*
* Verify that no user defined postfix operators exist.
*/
@@ -1470,7 +1592,8 @@ check_for_user_defined_encoding_conversions(ClusterInfo *cluster)
* check_new_cluster_logical_replication_slots()
*
* Verify that there are no logical replication slots on the new cluster and
- * that the parameter settings necessary for creating slots are sufficient.
+ * that the parameter settings necessary for creating slots and subscriptions
+ * are sufficient.
*/
static void
check_new_cluster_logical_replication_slots(void)
@@ -1479,6 +1602,7 @@ check_new_cluster_logical_replication_slots(void)
PGconn *conn;
int nslots_on_old;
int nslots_on_new;
+ int nsubs_on_old = old_cluster.subscription_count;
int max_replication_slots;
char *wal_level;
@@ -1488,29 +1612,35 @@ check_new_cluster_logical_replication_slots(void)
nslots_on_old = count_old_cluster_logical_slots();
- /* Quick return if there are no logical slots to be migrated. */
- if (nslots_on_old == 0)
+ /*
+ * Quick return if there are no logical slots and subscriptions to be
+ * migrated.
+ */
+ if (nslots_on_old == 0 && nsubs_on_old == 0)
return;
conn = connectToServer(&new_cluster, "template1");
- prep_status("Checking for new cluster logical replication slots");
+ if (nslots_on_old)
+ {
+ prep_status("Checking for new cluster logical replication slots");
- res = executeQueryOrDie(conn, "SELECT count(*) "
- "FROM pg_catalog.pg_replication_slots "
- "WHERE slot_type = 'logical' AND "
- "temporary IS FALSE;");
+ res = executeQueryOrDie(conn, "SELECT count(*) "
+ "FROM pg_catalog.pg_replication_slots "
+ "WHERE slot_type = 'logical' AND "
+ "temporary IS FALSE;");
- if (PQntuples(res) != 1)
- pg_fatal("could not count the number of logical replication slots");
+ if (PQntuples(res) != 1)
+ pg_fatal("could not count the number of logical replication slots");
- nslots_on_new = atoi(PQgetvalue(res, 0, 0));
+ nslots_on_new = atoi(PQgetvalue(res, 0, 0));
- if (nslots_on_new)
- pg_fatal("Expected 0 logical replication slots but found %d.",
- nslots_on_new);
+ if (nslots_on_new)
+ pg_fatal("Expected 0 logical replication slots but found %d.",
+ nslots_on_new);
- PQclear(res);
+ PQclear(res);
+ }
res = executeQueryOrDie(conn, "SELECT setting FROM pg_settings "
"WHERE name IN ('wal_level', 'max_replication_slots') "
@@ -1521,17 +1651,22 @@ check_new_cluster_logical_replication_slots(void)
wal_level = PQgetvalue(res, 0, 0);
- if (strcmp(wal_level, "logical") != 0)
+ if (nslots_on_old && strcmp(wal_level, "logical") != 0)
pg_fatal("wal_level must be \"logical\", but is set to \"%s\"",
wal_level);
max_replication_slots = atoi(PQgetvalue(res, 1, 0));
- if (nslots_on_old > max_replication_slots)
+ if (nslots_on_old && nslots_on_old > max_replication_slots)
pg_fatal("max_replication_slots (%d) must be greater than or equal to the number of "
"logical replication slots (%d) on the old cluster",
max_replication_slots, nslots_on_old);
+ if (nsubs_on_old && nsubs_on_old > max_replication_slots)
+ pg_fatal("max_replication_slots (%d) must be greater than or equal to the number of "
+ "subscriptions (%d) on the old cluster",
+ max_replication_slots, nsubs_on_old);
+
PQclear(res);
PQfinish(conn);
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index 4878aa22bf..f674ecd52e 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -21,6 +21,7 @@ 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_subscription_count(ClusterInfo *cluster);
static void get_db_infos(ClusterInfo *cluster);
static void get_rel_infos(ClusterInfo *cluster, DbInfo *dbinfo);
static void free_rel_infos(RelInfoArr *rel_arr);
@@ -286,6 +287,9 @@ get_db_rel_and_slot_infos(ClusterInfo *cluster, bool live_check)
get_template0_info(cluster);
get_db_infos(cluster);
+ if (cluster == &old_cluster)
+ get_subscription_count(cluster);
+
for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
{
DbInfo *pDbInfo = &cluster->dbarr.dbs[dbnum];
@@ -365,6 +369,27 @@ get_template0_info(ClusterInfo *cluster)
PQfinish(conn);
}
+/*
+ * Get the number of subscriptions in the old cluster.
+ */
+static void
+get_subscription_count(ClusterInfo *cluster)
+{
+ PGconn *conn;
+ PGresult *res;
+
+ if (GET_MAJOR_VERSION(cluster->major_version) < 1700)
+ return;
+
+ conn = connectToServer(cluster, "template1");
+ res = executeQueryOrDie(conn,
+ "SELECT oid FROM pg_catalog.pg_subscription");
+
+ cluster->subscription_count = PQntuples(res);
+
+ PQclear(res);
+ PQfinish(conn);
+}
/*
* get_db_infos()
diff --git a/src/bin/pg_upgrade/meson.build b/src/bin/pg_upgrade/meson.build
index 3e8a08e062..32f12f9e27 100644
--- a/src/bin/pg_upgrade/meson.build
+++ b/src/bin/pg_upgrade/meson.build
@@ -43,6 +43,7 @@ tests += {
't/001_basic.pl',
't/002_pg_upgrade.pl',
't/003_logical_slots.pl',
+ 't/004_subscription.pl',
],
'test_kwargs': {'priority': 40}, # pg_upgrade tests are slow
},
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index a710f325de..07cd6ed34c 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -292,6 +292,7 @@ typedef struct
char major_version_str[64]; /* string PG_VERSION of cluster */
uint32 bin_version; /* version returned from pg_ctl */
const char *tablespace_suffix; /* directory specification */
+ int subscription_count; /* number of subscriptions */
} ClusterInfo;
diff --git a/src/bin/pg_upgrade/t/004_subscription.pl b/src/bin/pg_upgrade/t/004_subscription.pl
new file mode 100644
index 0000000000..2823c17e82
--- /dev/null
+++ b/src/bin/pg_upgrade/t/004_subscription.pl
@@ -0,0 +1,392 @@
+# Copyright (c) 2023, PostgreSQL Global Development Group
+
+# Test for pg_upgrade of logical subscription
+use strict;
+use warnings;
+
+use File::Find qw(find);
+use File::Path qw(rmtree);
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Can be changed to test the other modes.
+my $mode = $ENV{PG_TEST_PG_UPGRADE_MODE} || '--copy';
+
+# Initialize publisher node
+my $publisher = PostgreSQL::Test::Cluster->new('publisher');
+$publisher->init(allows_streaming => 'logical');
+$publisher->start;
+
+# Initialize the old subscriber node
+my $old_sub = PostgreSQL::Test::Cluster->new('old_sub');
+$old_sub->init;
+$old_sub->start;
+
+# Initialize the new subscriber
+my $new_sub = PostgreSQL::Test::Cluster->new('new_sub');
+$new_sub->init;
+my $bindir = $new_sub->config_data('--bindir');
+
+sub insert_line_at_pub
+{
+ my $payload = shift;
+
+ foreach ("tab_upgraded", "tab_not_upgraded")
+ {
+ $publisher->safe_psql('postgres',
+ "INSERT INTO " . $_ . " (val) VALUES('$payload')");
+ }
+}
+
+# Initial setup
+foreach ("tab_upgraded", "tab_not_upgraded")
+{
+ $publisher->safe_psql('postgres',
+ "CREATE TABLE " . $_ . " (id serial, val text)");
+ $old_sub->safe_psql('postgres',
+ "CREATE TABLE " . $_ . " (id serial, val text)");
+}
+insert_line_at_pub('before initial sync');
+
+# Setup logical replication, replicating only 1 table
+my $connstr = $publisher->connstr . ' dbname=postgres';
+
+$publisher->safe_psql('postgres',
+ "CREATE PUBLICATION regress_pub FOR TABLE tab_upgraded");
+
+$old_sub->safe_psql('postgres',
+ "CREATE SUBSCRIPTION regress_sub CONNECTION '$connstr' PUBLICATION regress_pub"
+);
+
+# Wait for the catchup, as we need the subscription rel in ready state
+$old_sub->wait_for_subscription_sync($publisher, 'regress_sub');
+
+$publisher->safe_psql('postgres',
+ "INSERT INTO tab_upgraded VALUES (generate_series(2,50), 'after initial sync')"
+);
+$publisher->wait_for_catchup('regress_sub');
+
+# ------------------------------------------------------
+# Check that pg_upgrade is successful when all tables are in ready state.
+# ------------------------------------------------------
+my $synced_query =
+ "SELECT count(1) = 1 FROM pg_subscription_rel WHERE srsubstate = 'r'";
+$old_sub->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+command_ok(
+ [
+ 'pg_upgrade', '--no-sync', '-d', $old_sub->data_dir,
+ '-D', $new_sub->data_dir, '-b', $bindir,
+ '-B', $bindir, '-s', $new_sub->host,
+ '-p', $old_sub->port, '-P', $new_sub->port,
+ $mode, '--check',
+ ],
+ 'run of pg_upgrade --check for old instance when the subscription tables are in ready state'
+);
+ok( !-d $new_sub->data_dir . "/pg_upgrade_output.d",
+ "pg_upgrade_output.d/ removed after successful pg_upgrade");
+
+# Check the number of rows for each table on each server
+my $result =
+ $publisher->safe_psql('postgres', "SELECT count(*) FROM tab_upgraded");
+is($result, qq(50), "check initial tab_upgraded table data on publisher");
+$result =
+ $publisher->safe_psql('postgres', "SELECT count(*) FROM tab_not_upgraded");
+is($result, qq(1), "check initial tab_upgraded table data on publisher");
+$result =
+ $old_sub->safe_psql('postgres', "SELECT count(*) FROM tab_upgraded");
+is($result, qq(50),
+ "check initial tab_upgraded table data on the old subscriber");
+$result =
+ $old_sub->safe_psql('postgres', "SELECT count(*) FROM tab_not_upgraded");
+is($result, qq(0),
+ "check initial tab_not_upgraded table data on the old subscriber");
+
+# ------------------------------------------------------
+# Check that pg_upgrade refuses to run if there's a subscription with tables in
+# a state different than 'r' (ready), 'i' (init) and 's' (synchronized).
+# ------------------------------------------------------
+$publisher->safe_psql('postgres',
+ "CREATE TABLE tab_primary_key(id serial PRIMARY KEY, val text);");
+$old_sub->safe_psql('postgres',
+ "CREATE TABLE tab_primary_key(id serial PRIMARY KEY, val text);");
+$publisher->safe_psql('postgres',
+ "INSERT INTO tab_primary_key values(1, 'before initial sync')");
+
+# Insert the same value that is already present in publisher to the primary key
+# column of subscriber so that the table sync will fail.
+$old_sub->safe_psql('postgres',
+ "INSERT INTO tab_primary_key values(1, 'before initial sync')");
+
+$publisher->safe_psql('postgres',
+ "ALTER PUBLICATION regress_pub ADD TABLE tab_primary_key");
+$old_sub->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub REFRESH PUBLICATION");
+
+# Table will be in 'd' (data is being copied) state as table sync will fail
+# because of primary key constraint error.
+my $started_query =
+ "SELECT count(1) = 1 FROM pg_subscription_rel WHERE srsubstate = 'd';";
+$old_sub->poll_query_until('postgres', $started_query)
+ or die
+ "Timed out while waiting for the table state to become 'd' (datasync)";
+
+command_fails(
+ [
+ 'pg_upgrade', '--no-sync', '-d', $old_sub->data_dir,
+ '-D', $new_sub->data_dir, '-b', $bindir,
+ '-B', $bindir, '-s', $new_sub->host,
+ '-p', $old_sub->port, '-P', $new_sub->port,
+ $mode, '--check',
+ ],
+ 'run of pg_upgrade --check for old instance with relation in \'d\' datasync(invalid) state'
+);
+
+# Verify the reason why the subscriber cannot be upgraded
+my $sub_relstate_filename;
+
+# Find a txt file that contains a list of tables that cannot be upgraded. We
+# cannot predict the file's path because the output directory contains a
+# milliseconds timestamp. File::Find::find must be used.
+find(
+ sub {
+ if ($File::Find::name =~ m/subscription_state\.txt/)
+ {
+ $sub_relstate_filename = $File::Find::name;
+ }
+ },
+ $new_sub->data_dir . "/pg_upgrade_output.d");
+
+# Check the file content which should have tab_primary_key table in invalid
+# state.
+like(
+ slurp_file($sub_relstate_filename),
+ qr/database:postgres subscription:regress_sub schema:public relation:tab_primary_key state:d not in required state/m,
+ 'the previous test failed due to subscription table in invalid state');
+
+# Delete the table data so that the primary key violation error will not happen
+# and tab_primary_key reaches ready state.
+$old_sub->safe_psql('postgres', "DELETE FROM tab_primary_key");
+
+$synced_query =
+ "SELECT count(1) = 2 FROM pg_subscription_rel WHERE srsubstate = 'r'";
+$old_sub->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# ------------------------------------------------------
+# The incremental changes added to the publisher are replicated after upgrade.
+# ------------------------------------------------------
+
+# Stop the old subscriber, insert a row in tab_upgraded and tab_not_upgraded
+# publisher table while it's down and add tab_not_upgraded to the publication.
+my $remote_lsn = $old_sub->safe_psql('postgres',
+ "SELECT remote_lsn FROM pg_replication_origin_status");
+$old_sub->stop;
+
+insert_line_at_pub('while old_sub is down');
+
+# Run pg_upgrade
+command_ok(
+ [
+ 'pg_upgrade', '--no-sync', '-d', $old_sub->data_dir,
+ '-D', $new_sub->data_dir, '-b', $bindir,
+ '-B', $bindir, '-s', $new_sub->host,
+ '-p', $old_sub->port, '-P', $new_sub->port,
+ $mode,
+ ],
+ 'run of pg_upgrade for new sub');
+
+$publisher->safe_psql('postgres',
+ "ALTER PUBLICATION regress_pub ADD TABLE tab_not_upgraded");
+
+$new_sub->start;
+
+# Subscription relations and replication origin remote_lsn should be preserved
+$result =
+ $new_sub->safe_psql('postgres', "SELECT count(*) FROM pg_subscription_rel");
+is($result, qq(2),
+ "There should be 2 rows in pg_subscription_rel(representing tab_upgraded and tab_primary_key)"
+);
+
+$result = $new_sub->safe_psql('postgres',
+ "SELECT remote_lsn FROM pg_replication_origin_status os, pg_subscription s where os.external_id = 'pg_' || s.oid"
+);
+is($result, qq($remote_lsn), "remote_lsn should have been preserved");
+
+# There should be no new replicated rows before enabling the subscription
+$result =
+ $new_sub->safe_psql('postgres', "SELECT count(*) FROM tab_upgraded");
+is($result, qq(50),
+ "tab_upgraded table has no new replicated rows before enabling the subscription"
+);
+$result =
+ $new_sub->safe_psql('postgres', "SELECT count(*) FROM tab_not_upgraded");
+is($result, qq(0),
+ "no change in tab_not_upgraded table which is not part of the publication"
+);
+
+# Enable the subscription
+$new_sub->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub ENABLE");
+
+$publisher->wait_for_catchup('regress_sub');
+
+# Rows on tab_upgraded should have been replicated, while nothing should happen
+# for tab_not_upgraded.
+$result =
+ $new_sub->safe_psql('postgres', "SELECT count(*) FROM tab_upgraded");
+is($result, qq(51), "check replicated inserts on new subscriber");
+$result =
+ $new_sub->safe_psql('postgres', "SELECT count(*) FROM tab_not_upgraded");
+is($result, qq(0),
+ "no change in table tab_not_upgraded afer enable subscription which is not part of the publication"
+);
+
+# Refresh the subscription, only the missing row on tab_not_upgraded should be
+# replicated.
+$new_sub->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub REFRESH PUBLICATION");
+$new_sub->wait_for_subscription_sync($publisher, 'regress_sub');
+$result =
+ $new_sub->safe_psql('postgres', "SELECT count(*) FROM tab_upgraded");
+is($result, qq(51),
+ "check there is no change when there was no changes replicated");
+$result =
+ $new_sub->safe_psql('postgres', "SELECT count(*) FROM tab_not_upgraded");
+is($result, qq(2),
+ "check replicated inserts on new subscriber after refreshing");
+
+# ------------------------------------------------------
+# Check that pg_upgrade is successful when the table is in init state.
+# ------------------------------------------------------
+my $old_sub1 = PostgreSQL::Test::Cluster->new('old_sub1');
+$old_sub1->init;
+$old_sub1->append_conf('postgresql.conf',
+ "max_logical_replication_workers = 0");
+$old_sub1->start;
+
+$publisher->safe_psql('postgres',
+ "CREATE TABLE tab(id serial PRIMARY KEY, val text);");
+$old_sub1->safe_psql('postgres',
+ "CREATE TABLE tab(id serial PRIMARY KEY, val text);");
+
+$publisher->safe_psql('postgres',
+ "CREATE PUBLICATION regress_pub1 FOR TABLE tab");
+
+$old_sub1->safe_psql('postgres',
+ "CREATE SUBSCRIPTION regress_sub1 CONNECTION '$connstr' PUBLICATION regress_pub1"
+);
+
+$publisher->safe_psql('postgres',
+ "INSERT INTO tab values(1, 'before initial sync')");
+
+# The tables will be in init state as the subscriber configuration for
+# max_logical_replication_workers is set to 0.
+$synced_query =
+ "SELECT count(1) = 1 FROM pg_subscription_rel WHERE srsubstate = 'i'";
+$old_sub1->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Initialize the new subscriber
+my $new_sub1 = PostgreSQL::Test::Cluster->new('new_sub1');
+$new_sub1->init;
+
+$old_sub1->stop;
+
+command_ok(
+ [
+ 'pg_upgrade', '--no-sync', '-d', $old_sub1->data_dir,
+ '-D', $new_sub1->data_dir, '-b', $bindir,
+ '-B', $bindir, '-s', $new_sub1->host,
+ '-p', $old_sub1->port, '-P', $new_sub1->port,
+ $mode,
+ ],
+ 'run of pg_upgrade --check for old instance when the subscription tables are in ready state'
+);
+ok( !-d $new_sub1->data_dir . "/pg_upgrade_output.d",
+ "pg_upgrade_output.d/ removed after successful pg_upgrade");
+
+$new_sub1->start;
+
+$result =
+ $new_sub1->safe_psql('postgres', "SELECT srsubstate FROM pg_subscription_rel");
+is($result, qq(i), "check tab table is in init state after upgrade");
+
+# Check the number of rows in the table
+$result =
+ $new_sub1->safe_psql('postgres', "SELECT count(*) FROM tab");
+is($result, qq(0), "check initial tab table data on upgraded subscriber");
+
+# Enable the subscription
+$new_sub1->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub1 ENABLE");
+
+$new_sub1->wait_for_subscription_sync($publisher, 'regress_sub1');
+
+# Check the number of rows in the table
+$result =
+ $new_sub1->safe_psql('postgres', "SELECT count(*) FROM tab");
+is($result, qq(1), "check the data is synced after enabling the subscription");
+
+# ------------------------------------------------------
+# Check that pg_upgrade will fail when the subscription's replication origin
+# does not exist.
+# ------------------------------------------------------
+my $old_sub2 = PostgreSQL::Test::Cluster->new('old_sub2');
+$old_sub2->init;
+$old_sub2->start;
+
+$publisher->safe_psql('postgres',
+ "CREATE PUBLICATION regress_pub2");
+
+$old_sub2->safe_psql('postgres',
+ "CREATE SUBSCRIPTION regress_sub2 CONNECTION '$connstr' PUBLICATION regress_pub2"
+);
+
+$old_sub2->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_sub2 disable");
+
+my $subid = $old_sub2->safe_psql('postgres',
+ "SELECT oid FROM pg_subscription WHERE subname = 'regress_sub2'");
+my $reporigin = 'pg_'.qq($subid);
+$old_sub2->safe_psql('postgres',
+ "SELECT pg_replication_origin_drop('$reporigin')"
+);
+
+# Initialize the new subscriber
+my $new_sub2 = PostgreSQL::Test::Cluster->new('new_sub2');
+$new_sub2->init;
+
+$old_sub2->stop;
+
+command_fails(
+ [
+ 'pg_upgrade', '--no-sync', '-d', $old_sub2->data_dir,
+ '-D', $new_sub2->data_dir, '-b', $bindir,
+ '-B', $bindir, '-s', $new_sub2->host,
+ '-p', $old_sub2->port, '-P', $new_sub2->port,
+ $mode, '--check',
+ ],
+ 'run of pg_upgrade --check for old instance with missing replication origin'
+);
+
+# Find a txt file that contains a list of replication origins that is missing.
+# We cannot predict the file's path because the output directory contains a
+# milliseconds timestamp. File::Find::find must be used.
+find(
+ sub {
+ if ($File::Find::name =~ m/subscription_state\.txt/)
+ {
+ $sub_relstate_filename = $File::Find::name;
+ }
+ },
+ $new_sub2->data_dir . "/pg_upgrade_output.d");
+
+# Check the file content which should have regress_sub2 subscription.
+like(
+ slurp_file($sub_relstate_filename),
+ qr/replication origin is missing for database:postgres subscription:regress_sub2/m,
+ 'the previous test failed due to missing replication origin');
+
+done_testing();
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index bd0b8873d3..a52dc8f735 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11383,6 +11383,16 @@
provolatile => 'v', proparallel => 'u', prorettype => 'bool',
proargtypes => 'name',
prosrc => 'binary_upgrade_logical_slot_has_caught_up' },
+{ oid => '8404', descr => 'for use by pg_upgrade (relation for pg_subscription_rel)',
+ proname => 'binary_upgrade_add_sub_rel_state', proisstrict => 'f',
+ provolatile => 'v', proparallel => 'u', prorettype => 'void',
+ proargtypes => 'text oid char pg_lsn',
+ prosrc => 'binary_upgrade_add_sub_rel_state' },
+{ oid => '8405', descr => 'for use by pg_upgrade (remote_lsn for origin)',
+ proname => 'binary_upgrade_replorigin_advance', proisstrict => 'f',
+ provolatile => 'v', proparallel => 'u', prorettype => 'void',
+ proargtypes => 'text pg_lsn',
+ prosrc => 'binary_upgrade_replorigin_advance' },
# conversion functions
{ oid => '4302',
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index bf50a32119..a4946b40b1 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2660,6 +2660,7 @@ SubLinkType
SubOpts
SubPlan
SubPlanState
+SubRelInfo
SubRemoveRels
SubTransactionId
SubXactCallback
--
2.34.1
