public inbox for [email protected]
help / color / mirror / Atom feedFrom: Ajin Cherian <[email protected]>
To: shveta malik <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: Shlok Kyal <[email protected]>
Cc: Zsolt Parragi <[email protected]>
Cc: [email protected] <[email protected]>
Subject: Re: [PATCH] Preserve replication origin OIDs in pg_upgrade
Date: Thu, 9 Jul 2026 21:37:16 +1000
Message-ID: <CAFPTHDbVLyri=pehK6dT+G+QE5Kh-Zv0XEBsvAQ4PEkgmJh5SA@mail.gmail.com> (raw)
In-Reply-To: <CAJpy0uA5mbois4MAtZ+AFg0h0R21H=-stzWa7eLd6Vj0tEO=WA@mail.gmail.com>
References: <CAFPTHDa5aP8ixTZBygGAe48E1N=DvuPzcXikw7bNKixLMN3pVg@mail.gmail.com>
<OS9PR01MB12149E188221BED862E4BE9E0F5342@OS9PR01MB12149.jpnprd01.prod.outlook.com>
<CALDaNm2-uwpbJ8fnrssp+hORvOutsqRoZAsa05xVVzXe5Bt3bw@mail.gmail.com>
<CAFPTHDaNLRqPAda1RUDVfSDH7eLTaONv0UEmc9H5sdMW1Li2Bg@mail.gmail.com>
<CAN4CZFOb3urMsLPsEyVrYR7-7yA+BC5kDgQQd0nAQ8xj2zyRcA@mail.gmail.com>
<CAFPTHDYqpuZ6o2-HuCJDYqJ7GY3+zV+Xo-gT7PAgi4Bkz+oTxw@mail.gmail.com>
<CAN4CZFN4oLdxLwUXHUPV-5mFmK+4dcnppP00fV3i4qmMYCAkGA@mail.gmail.com>
<CAFPTHDbBiTfjYkQwYTNA2a4LRp8Sp7_zB59fmV0R7977ztxgmg@mail.gmail.com>
<CAFPTHDaftSwzGTGbFEw8rwDBsub0XqcDm1wxQcquj-Y3PC2qrg@mail.gmail.com>
<CANhcyEVercsNbgC5CTkMNzh_w_Jv1uK7RQg9vEvAeNQhoBHCKQ@mail.gmail.com>
<CAJpy0uAXynzWr4w_9GkJC3i=LL9WsRCZYVWAV0PgKmgs8riWzg@mail.gmail.com>
<CAFPTHDavPyxsWjK0cRO3yOaP4u8FGYrOJuXJB-4wzneAY3H3Ug@mail.gmail.com>
<CANhcyEVFSutXXdt_0XMCD37NE7Yd7ROdDVEE06D8YVyPNEBFHg@mail.gmail.com>
<CANhcyEVd7eDKGtsrUqwZO5McTG4dsJ+X8OxO31Xgu6sWmo--hw@mail.gmail.com>
<CAFPTHDbyVt_ggdi+rOP4=jBmgU=DDgYWuLxc4oQ6ux-T0H-ZBg@mail.gmail.com>
<CAHWVJhGoi+2bzZDNxtnB8Xc+DuFKB=YvqC0Xoc_HfzcgHAwMUg@mail.gmail.com>
<CAFPTHDbWKQjv4QfHaNiXosZ1TD0ZHc2hFyFFNB2RvLsJeJoH6g@mail.gmail.com>
<CAN4CZFPdFHwnew+wxt4UcXyjMYc3aXeR0XsR6_wZ=LEMGWEJFQ@mail.gmail.com>
<CAFPTHDYfDHQeQshXi7GDOQz3N8YdY6SNSmpoSAbnaQPBkVfv6A@mail.gmail.com>
<CANhcyEVV5A5qm9g7UVT99NpNqVsuzbzLsoteQ5gYOWvVsZsjCQ@mail.gmail.com>
<CAJpy0uAsgc5--4dCut2mmQd+Yxe-jfVQR=AY38dzeEe0+yaH4g@mail.gmail.com>
<OS9PR01MB12149E88B5ABBA9AEC715CBF2F5FF2@OS9PR01MB12149.jpnprd01.prod.outlook.com>
<CAA4eK1JtjsJiu+2cfWQ2dw883=Ks6c1TeTtJSs_mHVBFBo1aag@mail.gmail.com>
<OS9PR01MB121494C7792D4CB38DE67A0DDF5FF2@OS9PR01MB12149.jpnprd01.prod.outlook.com>
<CAA4eK1+TLMqD0KdZEk6vETx2=62bx=uihKCFCOtTHTFN5h5Oyw@mail.gmail.com>
<CAJpy0uA5mbois4MAtZ+AFg0h0R21H=-stzWa7eLd6Vj0tEO=WA@mail.gmail.com>
Thank you all for the comments. I have tried to address all the
comments. I am not responding to each of them here. Based on the
discussion above, I have retained the logic as mentioned by Shlok to
compare active origins against max_active_replication_origins.
Attaching v11 with all comments addressed.
regards,
Ajin Cherian
Fujitsu Australia
Attachments:
[application/octet-stream] v11-0001-Preserve-subscription-OIDs-during-pg_upgrade.patch (9.4K, ../CAFPTHDbVLyri=pehK6dT+G+QE5Kh-Zv0XEBsvAQ4PEkgmJh5SA@mail.gmail.com/2-v11-0001-Preserve-subscription-OIDs-during-pg_upgrade.patch)
download | inline diff:
From d8ddf11c4877efa2f2a5bf183a8e0e02e9737275 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Thu, 9 Jul 2026 21:06:04 +1000
Subject: [PATCH v11 1/2] Preserve subscription OIDs during pg_upgrade
Replication origins are linked to their subscriptions only by
naming convention, where the origin name embeds the subscription's
OID (e.g. "pg_<suboid>"). pg_dumpall restores origins independently
of CreateSubscription, so if subscription OIDs change during
upgrade, restored origins will no longer match their subscriptions.
This patch preserves subscription OIDs so a follow-up patch can
correctly restore replication origins.
Author: Vignesh C <[email protected]>
Reviewer: Ajin Cherian <[email protected]>
---
src/backend/commands/subscriptioncmds.c | 25 +++++++++++++++++--
src/backend/utils/adt/pg_upgrade_support.c | 10 ++++++++
src/bin/pg_dump/pg_dump.c | 8 ++++++
src/bin/pg_upgrade/pg_upgrade.c | 3 +++
src/bin/pg_upgrade/t/004_subscription.pl | 7 ++++++
src/include/catalog/binary_upgrade.h | 1 +
src/include/catalog/pg_proc.dat | 4 +++
.../expected/spgist_name_ops.out | 6 +++--
8 files changed, 60 insertions(+), 4 deletions(-)
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 4292e7fb8f4..e87c18356de 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -86,6 +86,12 @@
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
+/*
+ * This will be set by the pg_upgrade_support function --
+ * binary_upgrade_set_next_pg_subscription_oid().
+ */
+Oid binary_upgrade_next_pg_subscription_oid = InvalidOid;
+
/*
* Structure to hold a bitmap representing the user-provided CREATE/ALTER
* SUBSCRIPTION command options and the parsed/default values of each of them.
@@ -828,8 +834,23 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
- subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
- Anum_pg_subscription_oid);
+ /* Use binary-upgrade override for pg_subscription.oid? */
+ if (IsBinaryUpgrade)
+ {
+ if (!OidIsValid(binary_upgrade_next_pg_subscription_oid))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("pg_subscription OID value not set when in binary upgrade mode")));
+
+ subid = binary_upgrade_next_pg_subscription_oid;
+ binary_upgrade_next_pg_subscription_oid = InvalidOid;
+ }
+ else
+ {
+ subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
+ Anum_pg_subscription_oid);
+ }
+
values[Anum_pg_subscription_oid - 1] = ObjectIdGetDatum(subid);
values[Anum_pg_subscription_subdbid - 1] = ObjectIdGetDatum(MyDatabaseId);
values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index b505a6b4fee..59c3e7f0146 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -181,6 +181,16 @@ binary_upgrade_set_next_pg_authid_oid(PG_FUNCTION_ARGS)
PG_RETURN_VOID();
}
+Datum
+binary_upgrade_set_next_pg_subscription_oid(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+
+ CHECK_IS_BINARY_UPGRADE;
+ binary_upgrade_next_pg_subscription_oid = subid;
+ PG_RETURN_VOID();
+}
+
Datum
binary_upgrade_create_empty_extension(PG_FUNCTION_ARGS)
{
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 4d660d14b4c..75075ae080b 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -5537,6 +5537,14 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
appendPQExpBuffer(delq, "DROP SUBSCRIPTION %s;\n",
qsubname);
+ if (dopt->binary_upgrade)
+ {
+ appendPQExpBufferStr(query, "\n-- For binary upgrade, must preserve pg_subscription.oid\n");
+ appendPQExpBuffer(query,
+ "SELECT pg_catalog.binary_upgrade_set_next_pg_subscription_oid('%u'::pg_catalog.oid);\n\n",
+ subinfo->dobj.catId.oid);
+ }
+
appendPQExpBuffer(query, "CREATE SUBSCRIPTION %s ",
qsubname);
if (subinfo->subservername)
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index 7366fd4627c..26b1c8362dd 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -35,6 +35,9 @@
*
* We control all assignments of pg_database.oid because we want the directory
* names to match between the old and new cluster.
+ *
+ * We control assignment of pg_subscription.oid because we want the oid to
+ * match between the old and new cluster.
*/
diff --git a/src/bin/pg_upgrade/t/004_subscription.pl b/src/bin/pg_upgrade/t/004_subscription.pl
index 36f059faa2e..9286c1dc85e 100644
--- a/src/bin/pg_upgrade/t/004_subscription.pl
+++ b/src/bin/pg_upgrade/t/004_subscription.pl
@@ -313,6 +313,9 @@ my $tab_upgraded1_oid = $old_sub->safe_psql('postgres',
my $tab_upgraded2_oid = $old_sub->safe_psql('postgres',
"SELECT oid FROM pg_class WHERE relname = 'tab_upgraded2'");
+$sub_oid = $old_sub->safe_psql('postgres',
+ "SELECT oid FROM pg_subscription ORDER BY subname");
+
$old_sub->stop;
# Change configuration so that initial table sync does not get started
@@ -359,6 +362,10 @@ $publisher->safe_psql(
$new_sub->start;
+# The subscription oid should be preserved
+$result = $new_sub->safe_psql('postgres', "SELECT oid FROM pg_subscription ORDER BY subname");
+is($result, qq($sub_oid), "subscription oid should have been preserved");
+
# The subscription's running status, failover option, and retain_dead_tuples
# option should be preserved in the upgraded instance. So regress_sub4 should
# still have subenabled, subfailover, and subretaindeadtuples set to true,
diff --git a/src/include/catalog/binary_upgrade.h b/src/include/catalog/binary_upgrade.h
index 7bf7ae44385..b15b18e7dc9 100644
--- a/src/include/catalog/binary_upgrade.h
+++ b/src/include/catalog/binary_upgrade.h
@@ -32,6 +32,7 @@ extern PGDLLIMPORT RelFileNumber binary_upgrade_next_toast_pg_class_relfilenumbe
extern PGDLLIMPORT Oid binary_upgrade_next_pg_enum_oid;
extern PGDLLIMPORT Oid binary_upgrade_next_pg_authid_oid;
+extern PGDLLIMPORT Oid binary_upgrade_next_pg_subscription_oid;
extern PGDLLIMPORT bool binary_upgrade_record_init_privs;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 3cb84359adf..65d91e0d8de 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11994,6 +11994,10 @@
proisstrict => 'f', provolatile => 'v', proparallel => 'u',
prorettype => 'void', proargtypes => '',
prosrc => 'binary_upgrade_create_conflict_detection_slot' },
+{ oid => '9160', descr => 'for use by pg_upgrade',
+ proname => 'binary_upgrade_set_next_pg_subscription_oid', provolatile => 'v',
+ proparallel => 'r', prorettype => 'void', proargtypes => 'oid',
+ prosrc => 'binary_upgrade_set_next_pg_subscription_oid' },
# conversion functions
{ oid => '4310', descr => 'internal conversion function for KOI8R to WIN1251',
diff --git a/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out b/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
index 1ee65ede243..39d43368c42 100644
--- a/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
+++ b/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
@@ -59,11 +59,12 @@ select * from t
binary_upgrade_set_next_multirange_pg_type_oid | 1 | binary_upgrade_set_next_multirange_pg_type_oid
binary_upgrade_set_next_pg_authid_oid | | binary_upgrade_set_next_pg_authid_oid
binary_upgrade_set_next_pg_enum_oid | | binary_upgrade_set_next_pg_enum_oid
+ binary_upgrade_set_next_pg_subscription_oid | | binary_upgrade_set_next_pg_subscription_oid
binary_upgrade_set_next_pg_tablespace_oid | | binary_upgrade_set_next_pg_tablespace_oid
binary_upgrade_set_next_pg_type_oid | | binary_upgrade_set_next_pg_type_oid
binary_upgrade_set_next_toast_pg_class_oid | 1 | binary_upgrade_set_next_toast_pg_class_oid
binary_upgrade_set_next_toast_relfilenode | | binary_upgrade_set_next_toast_relfilenode
-(13 rows)
+(14 rows)
-- Verify clean failure when INCLUDE'd columns result in overlength tuple
-- The error message details are platform-dependent, so show only SQLSTATE
@@ -108,11 +109,12 @@ select * from t
binary_upgrade_set_next_multirange_pg_type_oid | 1 | binary_upgrade_set_next_multirange_pg_type_oid
binary_upgrade_set_next_pg_authid_oid | | binary_upgrade_set_next_pg_authid_oid
binary_upgrade_set_next_pg_enum_oid | | binary_upgrade_set_next_pg_enum_oid
+ binary_upgrade_set_next_pg_subscription_oid | | binary_upgrade_set_next_pg_subscription_oid
binary_upgrade_set_next_pg_tablespace_oid | | binary_upgrade_set_next_pg_tablespace_oid
binary_upgrade_set_next_pg_type_oid | | binary_upgrade_set_next_pg_type_oid
binary_upgrade_set_next_toast_pg_class_oid | 1 | binary_upgrade_set_next_toast_pg_class_oid
binary_upgrade_set_next_toast_relfilenode | | binary_upgrade_set_next_toast_relfilenode
-(13 rows)
+(14 rows)
\set VERBOSITY sqlstate
insert into t values(repeat('xyzzy', 12), 42, repeat('xyzzy', 4000));
--
2.47.3
[application/octet-stream] v11-0002-Preserve-replication-origin-OIDs-during-pg_upgra.patch (36.8K, ../CAFPTHDbVLyri=pehK6dT+G+QE5Kh-Zv0XEBsvAQ4PEkgmJh5SA@mail.gmail.com/3-v11-0002-Preserve-replication-origin-OIDs-during-pg_upgra.patch)
download | inline diff:
From 16f6719bcec4fb06a16d5d79527df68f63cbd63b Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Thu, 9 Jul 2026 21:13:09 +1000
Subject: [PATCH v11 2/2] Preserve replication origin OIDs during pg_upgrade
When pg_upgrade migrates a subscriber, replication origin OIDs
(roident) can change across the upgrade. This is a problem because
commit-timestamp records embed roident and are copied directly from
the old cluster's pg_commit_ts directory, causing spurious
"update_origin_differs" conflicts after the upgrade.
Fix this by dumping replication origins as global objects via
pg_dumpall during binary upgrade, using a new function
binary_upgrade_create_replication_origin(oid, name, lsn) to recreate
each origin with its preserved roident and remote_lsn. To avoid
conflicts with this, CreateSubscription() skips replorigin_create()
in binary-upgrade mode since the origin is already created by the
time the subscription is restored.
Author: Ajin Cherian <[email protected]>
Reviewer: Hayato Kuroda (Fujitsu) <[email protected]>
Reviewer: Zsolt Parragi <[email protected]>
Reviewer: Shlok Kyal <[email protected]>
Reviewer: Shveta malik <[email protected]>
Reviewer: Vignesh C <[email protected]>
Reviewer: Nisha Moond <[email protected]>
Reviewer: Rui Zhao <[email protected]>
---
doc/src/sgml/logical-replication.sgml | 4 +-
src/backend/commands/subscriptioncmds.c | 12 ++-
src/backend/replication/logical/origin.c | 95 +++++++++++--------
src/backend/utils/adt/pg_upgrade_support.c | 101 +++++++++++----------
src/bin/pg_dump/pg_dump.c | 51 ++---------
src/bin/pg_dump/pg_dump.h | 1 -
src/bin/pg_dump/pg_dumpall.c | 68 ++++++++++++++
src/bin/pg_upgrade/check.c | 65 +++++++++----
src/bin/pg_upgrade/info.c | 49 ++++++++--
src/bin/pg_upgrade/pg_upgrade.c | 13 ++-
src/bin/pg_upgrade/pg_upgrade.h | 4 +-
src/bin/pg_upgrade/t/004_subscription.pl | 74 +++++++++++++--
src/include/catalog/pg_proc.dat | 8 +-
src/include/replication/origin.h | 3 +
14 files changed, 374 insertions(+), 174 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 690598bff98..4db729df283 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -2844,7 +2844,7 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
<note>
<para>
- Commit timestamps and origin data are not preserved during the upgrade.
+ Commit timestamps are not preserved during the upgrade.
As a result, even if
<link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
is enabled, the upgraded subscriber may be unable to detect conflicts or
@@ -2886,7 +2886,7 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
The new cluster must have
<link linkend="guc-max-active-replication-origins"><varname>max_active_replication_origins</varname></link>
configured to a value greater than or equal to the number of
- subscriptions present in the old cluster.
+ replication origins present in the old cluster. The new cluster must contain no replication origins.
</para>
</listitem>
<listitem>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index e87c18356de..28bd0e48f2c 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -19,6 +19,7 @@
#include "access/table.h"
#include "access/twophase.h"
#include "access/xact.h"
+#include "catalog/binary_upgrade.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
@@ -957,9 +958,16 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
* apply workers initialization, and to handle origin creation dynamically
* when tables are added to the subscription. It is not clear whether
* preventing creation of origins is worth additional complexity.
+ *
+ * In binary-upgrade mode, skip origin creation here. It will be created
+ * separately so that the origin ID (roident) from the old cluster can be
+ * preserved for this subscription.
*/
- ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
- replorigin_create(originname);
+ if (!IsBinaryUpgrade)
+ {
+ ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
+ replorigin_create(originname);
+ }
/*
* Connect to remote side to execute requested commands and fetch table
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index c9dfb094c2b..6ba1a9e6b61 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -265,6 +265,51 @@ replorigin_by_name(const char *roname, bool missing_ok)
return roident;
}
+/*
+ * replorigin_create_with_id
+ *
+ * Create a replication origin with a specific ID and name, optionally
+ * restoring its remote_lsn. Used by pg_upgrade to preserve replication
+ * origin IDs across the upgrade.
+ *
+ * Caller must hold an exclusive lock on ReplicationOriginRelationId.
+ *
+ * Needs to be called in a transaction.
+ */
+void
+replorigin_create_with_id(ReplOriginId roident, const char *roname,
+ XLogRecPtr remote_lsn, Relation rel)
+{
+ Datum roname_d;
+ bool nulls[Natts_pg_replication_origin];
+ Datum values[Natts_pg_replication_origin];
+ HeapTuple tuple;
+
+ Assert(IsTransactionState());
+ Assert(CheckRelationLockedByMe(rel, ExclusiveLock, false));
+
+ roname_d = CStringGetTextDatum(roname);
+
+ memset(&nulls, 0, sizeof(nulls));
+ memset(&values, 0, sizeof(values));
+
+ values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(roident);
+ values[Anum_pg_replication_origin_roname - 1] = roname_d;
+
+ tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+ CatalogTupleInsert(rel, tuple);
+ heap_freetuple(tuple);
+ CommandCounterIncrement();
+
+ if (remote_lsn != InvalidXLogRecPtr)
+ {
+ Assert(IsBinaryUpgrade);
+ replorigin_advance(roident, remote_lsn, InvalidXLogRecPtr,
+ false /* backward */,
+ false /* WAL log */);
+ }
+}
+
/*
* Create a replication origin.
*
@@ -273,13 +318,11 @@ replorigin_by_name(const char *roname, bool missing_ok)
ReplOriginId
replorigin_create(const char *roname)
{
- Oid roident;
- HeapTuple tuple = NULL;
- Relation rel;
- Datum roname_d;
- SnapshotData SnapshotDirty;
- SysScanDesc scan;
- ScanKeyData key;
+ Oid roident;
+ Relation rel;
+ SnapshotData SnapshotDirty;
+ SysScanDesc scan;
+ ScanKeyData key;
/*
* To avoid needing a TOAST table for pg_replication_origin, we limit
@@ -293,8 +336,6 @@ replorigin_create(const char *roname)
errdetail("Replication origin names must be no longer than %d bytes.",
MAX_RONAME_LEN)));
- roname_d = CStringGetTextDatum(roname);
-
Assert(IsTransactionState());
/*
@@ -321,17 +362,15 @@ replorigin_create(const char *roname)
* snapshot. To make that safe, it needs to not have a TOAST table, since
* TOASTed data cannot be fetched without a snapshot. As of this writing,
* its only varlena column is roname, which we limit to 512 bytes to avoid
- * needing out-of-line storage. If you add a TOAST table to this catalog,
- * be sure to set up a snapshot everywhere it might be needed. For more
+ * needing out-of-line storage. If you add a TOAST table to this catalog,
+ * be sure to set up a snapshot everywhere it might be needed. For more
* information, see https://postgr.es/m/ZvMSUPOqUU-VNADN%40nathan.
*/
Assert(!OidIsValid(rel->rd_rel->reltoastrelid));
for (roident = InvalidOid + 1; roident < PG_UINT16_MAX; roident++)
{
- bool nulls[Natts_pg_replication_origin];
- Datum values[Natts_pg_replication_origin];
- bool collides;
+ bool collides;
CHECK_FOR_INTERRUPTS();
@@ -341,41 +380,25 @@ replorigin_create(const char *roname)
ObjectIdGetDatum(roident));
scan = systable_beginscan(rel, ReplicationOriginIdentIndex,
- true /* indexOK */ ,
+ true /* indexOK */,
&SnapshotDirty,
1, &key);
-
collides = HeapTupleIsValid(systable_getnext(scan));
-
systable_endscan(scan);
if (!collides)
- {
- /*
- * Ok, found an unused roident, insert the new row and do a CCI,
- * so our callers can look it up if they want to.
- */
- memset(&nulls, 0, sizeof(nulls));
-
- values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(roident);
- values[Anum_pg_replication_origin_roname - 1] = roname_d;
-
- tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls);
- CatalogTupleInsert(rel, tuple);
- CommandCounterIncrement();
break;
- }
}
- /* now release lock again, */
- table_close(rel, ExclusiveLock);
-
- if (tuple == NULL)
+ if (roident >= PG_UINT16_MAX)
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("could not find free replication origin ID")));
- heap_freetuple(tuple);
+ replorigin_create_with_id(roident, roname, InvalidXLogRecPtr, rel);
+
+ table_close(rel, ExclusiveLock);
+
return roident;
}
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index 59c3e7f0146..628b077ab69 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -28,6 +28,7 @@
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
+#include "utils/syscache.h"
#include "utils/pg_lsn.h"
@@ -377,71 +378,79 @@ binary_upgrade_add_sub_rel_state(PG_FUNCTION_ARGS)
}
/*
- * binary_upgrade_replorigin_advance
+ * binary_upgrade_create_conflict_detection_slot
*
- * Update the remote_lsn for the subscriber's replication origin.
+ * Create a replication slot to retain information necessary for conflict
+ * detection such as dead tuples, commit timestamps, and origins.
*/
Datum
-binary_upgrade_replorigin_advance(PG_FUNCTION_ARGS)
+binary_upgrade_create_conflict_detection_slot(PG_FUNCTION_ARGS)
{
- Relation rel;
- Oid subid;
- char *subname;
- char originname[NAMEDATALEN];
- ReplOriginId node;
- XLogRecPtr remote_commit;
-
CHECK_IS_BINARY_UPGRADE;
- /*
- * We must ensure a non-NULL subscription name 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));
- remote_commit = PG_ARGISNULL(1) ? InvalidXLogRecPtr : PG_GETARG_LSN(1);
-
- rel = table_open(SubscriptionRelationId, RowExclusiveLock);
- subid = get_subscription_oid(subname, false);
-
- ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
-
- /* Lock to prevent the replication origin from vanishing */
- LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
- node = replorigin_by_name(originname, false);
-
- /*
- * The server will be stopped after setting up the objects in the new
- * cluster and the origins will be flushed during the shutdown checkpoint.
- * This will ensure that the latest LSN values for origin will be
- * available after the upgrade.
- */
- replorigin_advance(node, remote_commit, InvalidXLogRecPtr,
- false /* backward */ ,
- false /* WAL log */ );
+ CreateConflictDetectionSlot();
- UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
- table_close(rel, RowExclusiveLock);
+ ReplicationSlotRelease();
PG_RETURN_VOID();
}
/*
- * binary_upgrade_create_conflict_detection_slot
+ * binary_upgrade_create_replication_origin
*
- * Create a replication slot to retain information necessary for conflict
- * detection such as dead tuples, commit timestamps, and origins.
+ * Create a replication origin with a specific OID and name, optionally
+ * restoring its remote_lsn. Used by pg_upgrade to preserve replication
+ * origin OIDs across the upgrade.
*/
Datum
-binary_upgrade_create_conflict_detection_slot(PG_FUNCTION_ARGS)
+binary_upgrade_create_replication_origin(PG_FUNCTION_ARGS)
{
+ Oid node_oid;
+ ReplOriginId node;
+ Relation rel;
+ char *originname;
+ XLogRecPtr remote_lsn = InvalidXLogRecPtr;
+
CHECK_IS_BINARY_UPGRADE;
- CreateConflictDetectionSlot();
+ if (PG_ARGISNULL(0) || PG_ARGISNULL(1))
+ elog(ERROR,
+ "null argument to binary_upgrade_create_replication_origin is not allowed");
- ReplicationSlotRelease();
+ node_oid = PG_GETARG_OID(0);
+
+ if (node_oid == InvalidOid || node_oid >= DoNotReplicateId)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("replication origin ID %u is out of range", node_oid));
+
+ node = (ReplOriginId) node_oid;
+
+ originname = text_to_cstring(PG_GETARG_TEXT_PP(1));
+
+ /*
+ * To avoid needing a TOAST table for pg_replication_origin, we limit
+ * replication origin names to 512 bytes. This should be more than enough
+ * for all practical use.
+ */
+ if (strlen(originname) > MAX_RONAME_LEN)
+ ereport(ERROR,
+ errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+ errmsg("replication origin name is too long"),
+ errdetail("Replication origin names must be no longer than %d bytes.",
+ MAX_RONAME_LEN));
+
+ if (!PG_ARGISNULL(2))
+ remote_lsn = PG_GETARG_LSN(2);
+
+ Assert(IsTransactionState());
+
+ /* Acquire an exclusive lock before inserting the new origin. */
+ rel = table_open(ReplicationOriginRelationId, ExclusiveLock);
+
+ replorigin_create_with_id(node, originname, remote_lsn, rel);
+
+ table_close(rel, ExclusiveLock);
PG_RETURN_VOID();
}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 75075ae080b..a03599538ae 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -5133,7 +5133,6 @@ getSubscriptions(Archive *fout)
int i_subwalrcvtimeout;
int i_subpublications;
int i_suborigin;
- int i_suboriginremotelsn;
int i_subenabled;
int i_subfailover;
int i_subretaindeadtuples;
@@ -5202,11 +5201,9 @@ getSubscriptions(Archive *fout)
LOGICALREP_ORIGIN_ANY);
if (dopt->binary_upgrade && fout->remoteVersion >= 170000)
- appendPQExpBufferStr(query, " o.remote_lsn AS suboriginremotelsn,\n"
- " s.subenabled,\n");
+ appendPQExpBufferStr(query, " s.subenabled,\n");
else
- appendPQExpBufferStr(query, " NULL AS suboriginremotelsn,\n"
- " false AS subenabled,\n");
+ appendPQExpBufferStr(query, " false AS subenabled,\n");
if (fout->remoteVersion >= 170000)
appendPQExpBufferStr(query,
@@ -5286,7 +5283,6 @@ getSubscriptions(Archive *fout)
i_subwalrcvtimeout = PQfnumber(res, "subwalrcvtimeout");
i_subpublications = PQfnumber(res, "subpublications");
i_suborigin = PQfnumber(res, "suborigin");
- i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
subinfo = pg_malloc_array(SubscriptionInfo, ntups);
@@ -5339,11 +5335,6 @@ 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);
@@ -5622,37 +5613,15 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
* In binary-upgrade mode, we allow the replication to continue after the
* upgrade.
*/
- if (dopt->binary_upgrade && fout->remoteVersion >= 170000)
+ if (dopt->binary_upgrade && subinfo->subenabled && fout->remoteVersion >= 170000)
{
- if (subinfo->suboriginremotelsn)
- {
- /*
- * Preserve the remote_lsn for the subscriber's replication
- * origin. This value is required to start the replication from
- * the position before the upgrade. This value will be stale if
- * the publisher gets upgraded before the subscriber node.
- * However, this shouldn't be a problem as the upgrade of the
- * publisher ensures that all the transactions were replicated
- * before upgrading it.
- */
- 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->subenabled)
- {
- /*
- * Enable the subscription to allow the replication to continue
- * after the upgrade.
- */
- appendPQExpBufferStr(query,
- "\n-- For binary upgrade, must preserve the subscriber's running state.\n");
- appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s ENABLE;\n", qsubname);
- }
+ /*
+ * Enable the subscription to allow the replication to continue
+ * after the upgrade.
+ */
+ appendPQExpBufferStr(query,
+ "\n-- For binary upgrade, must preserve the subscriber's running state.\n");
+ appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s ENABLE;\n", qsubname);
}
if (subinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 5a6726d8b12..8f1252b854e 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -729,7 +729,6 @@ typedef struct _SubscriptionInfo
char *subwalrcvtimeout;
char *subpublications;
char *suborigin;
- char *suboriginremotelsn;
} SubscriptionInfo;
/*
diff --git a/src/bin/pg_dump/pg_dumpall.c b/src/bin/pg_dump/pg_dumpall.c
index c53e77c2878..6af5bb058cd 100644
--- a/src/bin/pg_dump/pg_dumpall.c
+++ b/src/bin/pg_dump/pg_dumpall.c
@@ -18,6 +18,7 @@
#include <time.h>
#include <unistd.h>
+#include "access/xlogdefs.h"
#include "catalog/pg_authid_d.h"
#include "common/connect.h"
#include "common/file_perm.h"
@@ -67,6 +68,7 @@ static void dropDBs(PGconn *conn);
static void dumpUserConfig(PGconn *conn, const char *username);
static void dumpDatabases(PGconn *conn);
static void dumpTimestamp(const char *msg);
+static void dumpReplicationOrigins(PGconn *conn);
static int runPgDump(const char *dbname, const char *create_opts);
static void buildShSecLabels(PGconn *conn,
const char *catalog_name, Oid objectId,
@@ -663,8 +665,13 @@ main(int argc, char *argv[])
/* Dump role GUC privileges */
if (server_version >= 150000 && !skip_acls)
dumpRoleGUCPrivs(conn);
+
}
+ /* Dump replication origins */
+ if (!tablespaces_only && !roles_only && binary_upgrade)
+ dumpReplicationOrigins(conn);
+
/* Dump tablespaces */
if (!roles_only && !no_tablespaces)
dumpTablespaces(conn);
@@ -1804,6 +1811,67 @@ dumpTimestamp(const char *msg)
fprintf(OPF, "-- %s %s\n\n", msg, buf);
}
+static void
+dumpReplicationOrigins(PGconn *conn)
+{
+ PQExpBuffer buf = createPQExpBuffer();
+ PGresult *res;
+ int i_roident;
+ int i_roname;
+ int i_remotelsn;
+ int ntups;
+
+ /* Get replication origins from catalogs */
+ appendPQExpBufferStr(buf,
+ "SELECT o.roident, o.roname, os.remote_lsn "
+ "FROM pg_catalog.pg_replication_origin o "
+ "LEFT OUTER JOIN pg_catalog.pg_replication_origin_status os ON o.roident = os.local_id "
+ "ORDER BY o.roident");
+
+ res = executeQuery(conn, buf->data);
+
+ i_roident = PQfnumber(res, "roident");
+ i_roname = PQfnumber(res, "roname");
+ i_remotelsn = PQfnumber(res, "remote_lsn");
+ ntups = PQntuples(res);
+
+ if (ntups > 0)
+ fprintf(OPF, "--\n-- Replication Origins \n--\n\n");
+
+ for (int i = 0; i < ntups; i++)
+ {
+ ReplOriginId roident;
+ const char *roname;
+
+ roident = atooid(PQgetvalue(res, i, i_roident));
+ roname = PQgetvalue(res, i, i_roname);
+
+ resetPQExpBuffer(buf);
+
+ appendPQExpBufferStr(buf, "\n-- For binary upgrade, must preserve pg_replication_origin roident and remote_lsn\n");
+ appendPQExpBuffer(buf,
+ "SELECT pg_catalog.binary_upgrade_create_replication_origin("
+ "'%u'::pg_catalog.oid, ", roident);
+ appendStringLiteralConn(buf, roname, conn);
+ appendPQExpBufferStr(buf, "::pg_catalog.text");
+
+ if (!PQgetisnull(res, i, i_remotelsn))
+ {
+ appendPQExpBufferStr(buf, ", ");
+ appendStringLiteralConn(buf, PQgetvalue(res, i, i_remotelsn), conn);
+ appendPQExpBufferStr(buf, "::pg_catalog.pg_lsn");
+ }
+ else
+ appendPQExpBufferStr(buf, ", NULL");
+
+ appendPQExpBufferStr(buf, ");\n");
+ fprintf(OPF, "%s", buf->data);
+ }
+
+ PQclear(res);
+ destroyPQExpBuffer(buf);
+}
+
/*
* read_dumpall_filters - retrieve database identifier patterns from file
*
diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c
index 2155b01b11f..1d2db4527a8 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -32,7 +32,7 @@ static void check_for_new_tablespace_dir(void);
static void check_for_user_defined_encoding_conversions(ClusterInfo *cluster);
static void check_for_unicode_update(ClusterInfo *cluster);
static void check_new_cluster_replication_slots(void);
-static void check_new_cluster_subscription_configuration(void);
+static void check_new_cluster_replication_origins(void);
static void check_old_cluster_for_valid_slots(void);
static void check_old_cluster_subscription_state(void);
static void check_old_cluster_global_names(ClusterInfo *cluster);
@@ -589,6 +589,9 @@ check_and_dump_old_cluster(void)
check_old_cluster_subscription_state();
}
+ /* Get replication origin information */
+ get_replication_origin_info(&old_cluster);
+
check_for_data_types_usage(&old_cluster);
/*
@@ -707,7 +710,7 @@ check_new_cluster(void)
check_new_cluster_replication_slots();
- check_new_cluster_subscription_configuration();
+ check_new_cluster_replication_origins();
}
@@ -2150,31 +2153,45 @@ check_new_cluster_replication_slots(void)
}
/*
- * check_new_cluster_subscription_configuration()
+ * check_new_cluster_replication_origins()
+ *
+ * Verify that the new cluster has no replication origins. During upgrade,
+ * pg_upgrade restores replication origins from the old cluster with their
+ * original OIDs. If the new cluster already contains origins, those OIDs
+ * may collide, causing the upgrade to fail mid-way.
*
- * Verify that the max_active_replication_origins configuration specified is
- * enough for creating the subscriptions. This is required to create the
- * replication origin for each subscription.
+ * Also verify that the max_active_replication_origins configuration is
+ * enough for creating all the replication origins.
*/
static void
-check_new_cluster_subscription_configuration(void)
+check_new_cluster_replication_origins(void)
{
- PGresult *res;
PGconn *conn;
+ PGresult *res;
+ int norigins;
int max_active_replication_origins;
- /* Subscriptions and their dependencies can be migrated since PG17. */
- if (GET_MAJOR_VERSION(old_cluster.major_version) < 1700)
+ /* Quick return if there are no replication origins to migrate. */
+ if (old_cluster.nrepl_origins == 0)
return;
- /* Quick return if there are no subscriptions to be migrated. */
- if (old_cluster.nsubs == 0)
- return;
-
- prep_status("Checking new cluster configuration for subscriptions");
+ prep_status("Checking replication origins in new cluster");
conn = connectToServer(&new_cluster, "template1");
+ res = executeQueryOrDie(conn,
+ "SELECT count(*) "
+ "FROM pg_catalog.pg_replication_origin");
+
+ if (PQntuples(res) != 1)
+ pg_fatal("could not count the number of replication origins");
+
+ norigins = atoi(PQgetvalue(res, 0, 0));
+ PQclear(res);
+
+ if (norigins > 0)
+ pg_fatal("expected 0 replication origins but found %d", norigins);
+
res = executeQueryOrDie(conn, "SELECT setting FROM pg_settings "
"WHERE name = 'max_active_replication_origins';");
@@ -2182,12 +2199,20 @@ check_new_cluster_subscription_configuration(void)
pg_fatal("could not determine parameter settings on new cluster");
max_active_replication_origins = atoi(PQgetvalue(res, 0, 0));
- if (old_cluster.nsubs > max_active_replication_origins)
- pg_fatal("\"max_active_replication_origins\" (%d) must be greater than or equal to the number of "
- "subscriptions (%d) in the old cluster",
- max_active_replication_origins, old_cluster.nsubs);
-
PQclear(res);
+
+ /*
+ * Compare against the number of active origins in the old cluster not the
+ * total number of origins in pg_replication_origin. An origin can exist
+ * without ever becoming active, in which case it does not count against
+ * max_active_replication_origins.
+ */
+ if (old_cluster.nrepl_origins_active > max_active_replication_origins)
+ pg_fatal("\"max_active_replication_origins\" (%d) must be greater "
+ "than or equal to the number of active replication origins "
+ "(%d) in the old cluster",
+ max_active_replication_origins, old_cluster.nrepl_origins_active);
+
PQfinish(conn);
check_ok();
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index 37fff93892f..311d08fe233 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -832,6 +832,43 @@ count_old_cluster_logical_slots(void)
return slot_count;
}
+/*
+ * get_replication_origin_info()
+ *
+ * Gets the information of replication origins in the cluster.
+ */
+void
+get_replication_origin_info(ClusterInfo *cluster)
+{
+ PGconn *conn;
+ PGresult *res;
+
+ conn = connectToServer(cluster, "template1");
+
+ res = executeQueryOrDie(conn, "SELECT count(*) "
+ "FROM pg_catalog.pg_replication_origin");
+ if (PQntuples(res) != 1)
+ pg_fatal("could not get the number of replication origins");
+ cluster->nrepl_origins = atoi(PQgetvalue(res, 0, 0));
+ PQclear(res);
+
+ /*
+ * Get the number of actively tracked replication origins, i.e., those
+ * with an entry in pg_replication_origin_status. This can be less than
+ * nrepl_origins, since an origin can exist without ever becoming
+ * active. This count is what is actually bounded by
+ * max_active_replication_origins.
+ */
+ res = executeQueryOrDie(conn, "SELECT count(*) "
+ "FROM pg_catalog.pg_replication_origin_status");
+ if (PQntuples(res) != 1)
+ pg_fatal("could not get the number of active replication origins");
+ cluster->nrepl_origins_active = atoi(PQgetvalue(res, 0, 0));
+ PQclear(res);
+
+ PQfinish(conn);
+}
+
/*
* get_subscription_info()
*
@@ -842,23 +879,19 @@ 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 "
+ res = executeQueryOrDie(conn,
+ "SELECT 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");
+ res = executeQueryOrDie(conn,
+ "SELECT 'f' AS retain_dead_tuples ");
- 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);
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index 26b1c8362dd..9615460d4f6 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -219,13 +219,12 @@ main(int argc, char **argv)
* as it only retains the dead tuples. It is created here for consistency.
* Note that the new conflict detection slot uses the latest transaction
* ID as xmin, so it cannot protect dead tuples that existed before the
- * upgrade. Additionally, commit timestamps and origin data are not
- * preserved during the upgrade. So, even after creating the slot, the
- * upgraded subscriber may be unable to detect conflicts or log relevant
- * commit timestamps and origins when applying changes from the publisher
- * occurred before the upgrade especially if those changes were not
- * replicated. It can only protect tuples that might be deleted after the
- * new cluster starts.
+ * upgrade. Additionally, commit timestamps are not preserved during the
+ * upgrade. So, even after creating the slot, the upgraded subscriber may
+ * be unable to detect conflicts or log relevant commit timestamps when
+ * applying changes from the publisher occurred before the upgrade
+ * especially if those changes were not replicated. It can only protect
+ * tuples that might be deleted after the new cluster starts.
*/
if (migrate_logical_slots || old_cluster.sub_retain_dead_tuples)
{
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index d6e5bca5792..a09554ea3dc 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -286,7 +286,8 @@ typedef struct
char **tablespaces; /* tablespace directories */
int num_tablespaces;
const char *tablespace_suffix; /* directory specification */
- int nsubs; /* number of subscriptions */
+ int nrepl_origins; /* number of replication origins */
+ int nrepl_origins_active; /* number of replication origins that are active */
bool sub_retain_dead_tuples; /* whether a subscription enables
* retain_dead_tuples. */
} ClusterInfo;
@@ -425,6 +426,7 @@ FileNameMap *gen_db_file_maps(DbInfo *old_db,
void get_db_rel_and_slot_infos(ClusterInfo *cluster);
int count_old_cluster_logical_slots(void);
void get_subscription_info(ClusterInfo *cluster);
+void get_replication_origin_info(ClusterInfo *cluster);
/* option.c */
diff --git a/src/bin/pg_upgrade/t/004_subscription.pl b/src/bin/pg_upgrade/t/004_subscription.pl
index 9286c1dc85e..be93f0e3cbc 100644
--- a/src/bin/pg_upgrade/t/004_subscription.pl
+++ b/src/bin/pg_upgrade/t/004_subscription.pl
@@ -42,13 +42,12 @@ my $connstr = $publisher->connstr . ' dbname=postgres';
# ------------------------------------------------------
# Check that pg_upgrade fails when max_active_replication_origins configured
-# in the new cluster is less than the number of subscriptions in the old
-# cluster.
+# in the new cluster is less than the number of active replication origins in
+# the old cluster.
# ------------------------------------------------------
-# It is sufficient to use disabled subscription to test upgrade failure.
$publisher->safe_psql('postgres', "CREATE PUBLICATION regress_pub1");
$old_sub->safe_psql('postgres',
- "CREATE SUBSCRIPTION regress_sub1 CONNECTION '$connstr' PUBLICATION regress_pub1 WITH (enabled = false)"
+ "CREATE SUBSCRIPTION regress_sub1 CONNECTION '$connstr' PUBLICATION regress_pub1"
);
$old_sub->stop;
@@ -74,7 +73,7 @@ command_checks_all(
],
1,
[
- qr/"max_active_replication_origins" \(0\) must be greater than or equal to the number of subscriptions \(1\) in the old cluster/
+ qr/"max_active_replication_origins" \(0\) must be greater than or equal to the number of active replication origins \(1\) in the old cluster/
],
[qr//],
'run of pg_upgrade where the new cluster has insufficient max_active_replication_origins'
@@ -301,8 +300,38 @@ is($result, qq(t), "Check that the table is in init state");
# Get the replication origin's remote_lsn of the old subscriber
my $remote_lsn = $old_sub->safe_psql('postgres',
- "SELECT remote_lsn FROM pg_replication_origin_status os, pg_subscription s WHERE os.external_id = 'pg_' || s.oid AND s.subname = 'regress_sub4'"
+ "SELECT os.remote_lsn
+ FROM pg_replication_origin_status os
+ JOIN pg_replication_origin o ON o.roident = os.local_id
+ JOIN pg_subscription s ON o.roname = 'pg_' || s.oid::text
+ WHERE s.subname = 'regress_sub4'"
+);
+
+# Get the replication origin ids (roident) for all subscriptions, keyed by
+# subscription name (which is stable across upgrade, unlike suboid). These
+# must be preserved after upgrade. A mismatch would cause spurious
+# update_origin_differs conflicts.
+my %pre_upgrade_roident;
+my $roident_rows = $old_sub->safe_psql('postgres',
+ "SELECT s.subname, o.roident
+ FROM pg_subscription s
+ JOIN pg_replication_origin o ON o.roname = 'pg_' || s.oid::text
+ ORDER BY s.subname"
);
+for my $row (split /\n/, $roident_rows)
+{
+ my ($subname, $roident) = split /\|/, $row;
+ $pre_upgrade_roident{$subname} = $roident;
+}
+
+# Create a user created replication origin, which should also be preserved after upgrade.
+my $user_origin_name = 'regress_user_origin';
+$old_sub->safe_psql('postgres',
+ "SELECT pg_replication_origin_create('$user_origin_name')");
+$pre_upgrade_roident{$user_origin_name} = $old_sub->safe_psql('postgres',
+ "SELECT roident FROM pg_replication_origin WHERE roname = '$user_origin_name'"
+);
+
# Have the subscription in disabled state before upgrade
$old_sub->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub5 DISABLE");
@@ -378,6 +407,39 @@ regress_sub5|f|f|f),
"check that the subscription's running status, failover, and retain_dead_tuples are preserved"
);
+# Verify that the subscription related replication origins are preserved after upgrade.
+my $post_roident_rows = $new_sub->safe_psql('postgres',
+ "SELECT s.subname, o.roident
+ FROM pg_subscription s
+ JOIN pg_replication_origin o ON o.roname = 'pg_' || s.oid::text
+ ORDER BY s.subname"
+);
+
+my %post_upgrade_roident;
+for my $row (split /\n/, $post_roident_rows)
+{
+ my ($subname, $roident) = split /\|/, $row;
+ $post_upgrade_roident{$subname} = $roident;
+}
+
+# Iterate over the union of pre- and post-upgrade keys so a missing or
+# unexpected entry on either side is caught.
+my %all_subnames = (%pre_upgrade_roident, %post_upgrade_roident);
+for my $subname (sort keys %all_subnames)
+{
+ next if $subname eq $user_origin_name;
+ is($post_upgrade_roident{$subname}, $pre_upgrade_roident{$subname},
+ "roident preserved for subscription '$subname' after upgrade");
+}
+
+# Verify that user created replication origins are preserved after upgrade.
+my $post_user_roident = $new_sub->safe_psql('postgres',
+ "SELECT roident FROM pg_replication_origin WHERE roname = '$user_origin_name'"
+);
+is($post_user_roident, $pre_upgrade_roident{$user_origin_name},
+ "roident preserved for user-created origin '$user_origin_name' after upgrade"
+);
+
# Subscription relations should be preserved
$result = $new_sub->safe_psql('postgres',
"SELECT srrelid, srsubstate FROM pg_subscription_rel ORDER BY srrelid");
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 65d91e0d8de..bf56485c7f9 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11985,10 +11985,6 @@
provolatile => 'v', proparallel => 'u', prorettype => 'void',
proargtypes => 'text oid char pg_lsn',
prosrc => 'binary_upgrade_add_sub_rel_state' },
-{ oid => '6320', 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' },
{ oid => '6505', descr => 'for use by pg_upgrade (conflict detection slot)',
proname => 'binary_upgrade_create_conflict_detection_slot',
proisstrict => 'f', provolatile => 'v', proparallel => 'u',
@@ -11998,6 +11994,10 @@
proname => 'binary_upgrade_set_next_pg_subscription_oid', provolatile => 'v',
proparallel => 'r', prorettype => 'void', proargtypes => 'oid',
prosrc => 'binary_upgrade_set_next_pg_subscription_oid' },
+{ oid => '9161', descr => 'for use by pg_upgrade (replication origin)',
+ proname => 'binary_upgrade_create_replication_origin', proisstrict => 'f',
+ provolatile => 'v', proparallel => 'u', prorettype => 'void',
+ proargtypes => 'oid text pg_lsn', prosrc => 'binary_upgrade_create_replication_origin' },
# conversion functions
{ oid => '4310', descr => 'internal conversion function for KOI8R to WIN1251',
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index a69faf6eaaf..11ee630fb28 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -14,6 +14,7 @@
#include "access/xlogdefs.h"
#include "access/xlogreader.h"
#include "catalog/pg_replication_origin.h"
+#include "utils/relcache.h"
typedef struct xl_replorigin_set
{
@@ -55,6 +56,8 @@ extern PGDLLIMPORT int max_active_replication_origins;
/* API for querying & manipulating replication origins */
extern ReplOriginId replorigin_by_name(const char *roname, bool missing_ok);
extern ReplOriginId replorigin_create(const char *roname);
+extern void replorigin_create_with_id(ReplOriginId roident, const char *roname,
+ XLogRecPtr remote_lsn, Relation rel);
extern void replorigin_drop_by_name(const char *name, bool missing_ok, bool nowait);
extern bool replorigin_by_oid(ReplOriginId roident, bool missing_ok,
char **roname);
--
2.47.3
view thread (44+ messages)
reply
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: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: [PATCH] Preserve replication origin OIDs in pg_upgrade
In-Reply-To: <CAFPTHDbVLyri=pehK6dT+G+QE5Kh-Zv0XEBsvAQ4PEkgmJh5SA@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox