public inbox for [email protected]
help / color / mirror / Atom feedFrom: Ajin Cherian <[email protected]>
To: Shlok Kyal <[email protected]>
Cc: shveta malik <[email protected]>
Cc: Zsolt Parragi <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: shveta malik <[email protected]>
Subject: Re: [PATCH] Preserve replication origin OIDs in pg_upgrade
Date: Wed, 17 Jun 2026 19:19:19 +1000
Message-ID: <CAFPTHDbyVt_ggdi+rOP4=jBmgU=DDgYWuLxc4oQ6ux-T0H-ZBg@mail.gmail.com> (raw)
In-Reply-To: <CANhcyEVd7eDKGtsrUqwZO5McTG4dsJ+X8OxO31Xgu6sWmo--hw@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>
On Fri, Jun 5, 2026 at 8:10 PM vignesh C <[email protected]> wrote:
>
> Few comments:
> 1) pg_dumpall dumps all replication origins, not just subscription
> origins and recreates every origin:
> SELECT o.*, os.remote_lsn
> FROM pg_replication_origin o
> LEFT JOIN pg_replication_origin_status os
> ON o.roident = os.local_id
>
> Is pg_upgrade expected to preserve user-created origins?
> If so, you can mention this in the commit message
>
I explicitly state that all replication origins are migrated in the
commit message, nothing specific about subscription origins.
> 2) Should the table lock be released here, if so should we mention
> comments similar to how it is mentioned in
> binary_upgarde_add_sub_rel_state:
> + /* Restore the remote_lsn if provided, while still holding the lock */
> + if (!PG_ARGISNULL(2))
> + {
> + XLogRecPtr remote_commit = PG_GETARG_LSN(2);
>
> - ReplicationSlotRelease();
> + replorigin_advance(node, remote_commit, InvalidXLogRecPtr,
> + false /* backward */,
> + false /* WAL log */);
> + }
> +
> + table_close(rel, RowExclusiveLock);
>
Is that relevant when creating a new row? or only when updating an existing one?
> 3) We can use new style of ereport to exclude '(' before errcode:
> 3.a)
> + 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)));
>
> 3.b) Similarly here:
> + ereport(ERROR,
> + (errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin with ID %u already exists", node_oid)));
>
> 3.c) There are many others, you can search other ereport in the patch
> and change it too.
>
Done.
> 4) Since only one column will be returned here, can we use
> "PQgetvalue(res, 0, 0) and remove PQfnumber:
> + res = executeQueryOrDie(conn,
> + "SELECT count(*) AS nrepl_origins "
> + "FROM pg_catalog.pg_replication_origin");
> + i_nrepl_origins = PQfnumber(res, "nrepl_origins");
> + cluster->nrepl_origins = atoi(PQgetvalue(res, 0, i_nrepl_origins));
> + PQclear(res);
>
Done.
> 5) replorigin_create also has some duplicate code like below, will it
> be possible to have a common function so that both of them can use?
> /* Check for name collision */
> + ScanKeyInit(&key,
> + Anum_pg_replication_origin_roname,
> + BTEqualStrategyNumber, F_TEXTEQ,
> + roname_d);
> + scan = systable_beginscan(rel, ReplicationOriginNameIndex,
> + true /* indexOK */,
> + SnapshotSelf,
> + 1, &key);
> + collides = HeapTupleIsValid(systable_getnext(scan));
> + systable_endscan(scan);
> +
> + if (collides)
> + ereport(ERROR,
> + (errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin \"%s\" already exists",
> + originname)));
> +
> + memset(&nulls, 0, sizeof(nulls));
> + memset(&values, 0, sizeof(values));
> +
> + values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(node);
> + values[Anum_pg_replication_origin_roname - 1] = roname_d;
> +
> + tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls);
> + CatalogTupleInsert(rel, tuple);
> + heap_freetuple(tuple);
> + CommandCounterIncrement();
>
I have created a new function replorigin_create_with_id() which has
some common code and called that from both replorigin_create() and
binary_upgrade_create_replication_origin()
On Tue, Jun 9, 2026 at 3:06 AM Shlok Kyal <[email protected]> wrote:
>
> Hi Ajin,
> I have reviewed the patch and here are some comments:
>
> 1. I want to clarify, what is the expected behaviour when we upgrade
> the postgres instance from version <= 16 to "HEAD + patch"?
> - ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname,
> sizeof(originname));
> - replorigin_create(originname);
> + if (!IsBinaryUpgrade)
> + {
> + ReplicationOriginNameForLogicalRep(subid, InvalidOid,
> originname, sizeof(originname));
> + replorigin_create(originname);
> + }
> and
> + /* Dump replication origins */
> + if (server_version >= 170000 && binary_upgrade &&
> archDumpFormat == archNull)
> + dumpReplicationOrigins(conn);
>
> When the postgres instance with subscription is updated from PG 16 (or
> less) to HEAD, the replication origin is created as we do not have "if
> (!IsBinaryUpgrade)" check in HEAD,
> whereas when the similar instance is upgraded to "HEAD + patch", no
> replication origin is present after the upgrade.
> Is this difference in the behaviour between HEAD and "HEAD + patch" expected?
>
I have modified this as option 3 suggested by Shveta and Amit and
removed the PG17 restriction for migrating replication origins, now it
does it for all versions that have replication origins.
> 2.a) Can we replace:
> + /* Check for OID collision */
> + ScanKeyInit(&key,
> + Anum_pg_replication_origin_roident,
> + BTEqualStrategyNumber, F_OIDEQ,
> + ObjectIdGetDatum(node));
> + scan = systable_beginscan(rel, ReplicationOriginIdentIndex,
> + true /* indexOK */,
> + SnapshotSelf,
> + 1, &key);
> + collides = HeapTupleIsValid(systable_getnext(scan));
> + systable_endscan(scan);
> +
> + if (collides)
> + ereport(ERROR,
> + (errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin with ID %u already
> exists", node_oid)));
>
> with:
> if (SearchSysCacheExists1(REPLORIGIDENT, ObjectIdGetDatum(node)))
> ereport(ERROR, errcode(ERRCODE_DUPLICATE_OBJECT),
> errmsg("replication origin with ID %u already
> exists", node_oid));
>
> 2.b) Similarly, Can we replace:
> + /* Check for name collision */
> + ScanKeyInit(&key,
> + Anum_pg_replication_origin_roname,
> + BTEqualStrategyNumber, F_TEXTEQ,
> + roname_d);
> + scan = systable_beginscan(rel, ReplicationOriginNameIndex,
> + true /* indexOK */,
> + SnapshotSelf,
> + 1, &key);
> + collides = HeapTupleIsValid(systable_getnext(scan));
> + systable_endscan(scan);
> +
> + if (collides)
> + ereport(ERROR,
> + (errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin \"%s\" already exists",
> + originname)));
> with:
> if (SearchSysCacheExists1(REPLORIGNAME, roname_d))
> ereport(ERROR, errcode(ERRCODE_DUPLICATE_OBJECT),
> errmsg("replication origin \"%s\" already
> exists", originname));
>
> This will make the code smaller and easy to read. Thoughts?
Done
On Tue, Jun 9, 2026 at 2:13 PM shveta malik <[email protected]> wrote:
>
>
> Please find a few comments on v007:
>
> 1)
>
> + node = (ReplOriginId) node_oid;
> + originname = text_to_cstring(PG_GETARG_TEXT_PP(1));
> +
> + 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)));
> +
> + roname_d = CStringGetTextDatum(originname);
>
> CStringGetTextDatum is:
> #define CStringGetTextDatum(s) PointerGetDatum(cstring_to_text(s))
> ---------
>
> We are first converting text to C-string , then Cstring to text and
> then getting Datum. Double allocation and double conversion. Can we
> please try this instead and verify if it works:
>
> text *origin_text = PG_GETARG_TEXT_PP(1);
> originname = text_to_cstring(origin_text);
> roname_d = PointerGetDatum(origin_text);
>
I've restructured the code now, its a bit tough to do this now.
> 2)
> +check_new_cluster_replication_origins(void)
> +{
> + PGconn *conn;
> + PGresult *res;
> + int norigins;
> +
> + /* Quick return if there are no replication origins to migrate. */
> + if (old_cluster.nrepl_origins == 0)
> + return;
>
> Don't we need a version check simialr to what we have introduced in
> pg_dumpall for origins? (Similar to
> check_new_cluster_replication_slots as well)
>
I've removed the version check for PG17 now.
> 3)
> Since we have introduced check_new_cluster_replication_origins, do we
> even need below checks in binary_upgrade_create_replication_origin()?
> In which conditions will they be hit?
>
> + /* Check for OID collision */
> + ....
> + if (collides)
> + ereport(ERROR,
> + (errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin with ID %u already exists", node_oid)));
> +
> + /* Check for name collision */
> + ....
> + if (collides)
> + ereport(ERROR,
> + (errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin \"%s\" already exists",
> + originname)));
>
With the restructured code, its not easy to do this now.
>
> 4)
> Now since check_new_cluster_subscription_configuration() is purely
> about checking max-origin GUC configuration, I think the name is
> misleading. IMO, we should merge it to
> check_new_cluster_replication_origins(). See how
> check_new_cluster_replication_slots() does it: checking both
> new-cluster's count and the max-configuration for slots.
>
Done.
On Tue, Jun 9, 2026 at 9:32 PM Shlok Kyal <[email protected]> wrote:
>
> >
> One more comment:
>
> 3. In logical-replication.sgml, we have documentation as:
> "Commit timestamps and origin data are not preserved during the upgrade."
> Now since the origin related data is preserved after the upgrade,
> should we update the documentation here?
>
Updated.
I have addressed the comments in v8 (attached). I've also now taken
out the check for PG17 on old cluster and now all clusters which have
replication origins are migrated.
regards,
Ajin Cherian
Fujitsu Australia
Attachments:
[application/octet-stream] v8-0001-Preserve-subscription-OIDs-during-pg_upgrade.patch (9.1K, ../CAFPTHDbyVt_ggdi+rOP4=jBmgU=DDgYWuLxc4oQ6ux-T0H-ZBg@mail.gmail.com/2-v8-0001-Preserve-subscription-OIDs-during-pg_upgrade.patch)
download | inline diff:
From d7864a45be2d647ca3f58cdf6407c06cb53b5ff0 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Thu, 11 Jun 2026 12:35:13 +1000
Subject: [PATCH v8 1/2] Preserve subscription OIDs during pg_upgrade
Currently subscription OIDs can be changed when a cluster is upgraded
using pg_upgrade. This is required for a subsequent patch which will
preserve the replication oids after upgrade.
Author: Vignesh C <[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 fd026b304c2..71386fa7d9d 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -83,6 +83,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.
@@ -772,8 +778,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 a0f7f8e2168..049337de62a 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -5583,6 +5583,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 2127d297bfe..4e853096698 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 c94a82deae0..646767f2a65 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 be157a5fbe9..3a28406981d 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11969,6 +11969,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] v8-0002-Preserve-replication-origin-OIDs-during-pg_upgrad.patch (28.4K, ../CAFPTHDbyVt_ggdi+rOP4=jBmgU=DDgYWuLxc4oQ6ux-T0H-ZBg@mail.gmail.com/3-v8-0002-Preserve-replication-origin-OIDs-during-pg_upgrad.patch)
download | inline diff:
From 1e750a58fd4e247364b7c606869aae67287f84a1 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Wed, 17 Jun 2026 18:16:13 +1000
Subject: [PATCH v8 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]>
---
doc/src/sgml/logical-replication.sgml | 2 +-
src/backend/commands/subscriptioncmds.c | 11 +-
src/backend/replication/logical/origin.c | 117 +++++++++++++++------
src/backend/utils/adt/pg_upgrade_support.c | 88 +++++++---------
src/bin/pg_dump/pg_dump.c | 38 ++-----
src/bin/pg_dump/pg_dumpall.c | 64 +++++++++++
src/bin/pg_upgrade/check.c | 54 ++++++----
src/bin/pg_upgrade/info.c | 7 ++
src/bin/pg_upgrade/pg_upgrade.h | 1 +
src/bin/pg_upgrade/t/004_subscription.pl | 42 +++++++-
src/include/catalog/pg_proc.dat | 8 +-
src/include/replication/origin.h | 2 +
12 files changed, 293 insertions(+), 141 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 9e7868487de..0e484800fca 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -2819,7 +2819,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
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 71386fa7d9d..81b0ce22ee9 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"
@@ -867,9 +868,15 @@ 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. This is required to
+ * preserve the roident from the old cluster 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..b5b87f3e93b 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -265,6 +265,73 @@ 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.
+ *
+ * Needs to be called in a transaction.
+ */
+void
+replorigin_create_with_id(ReplOriginId roident, const char *roname,
+ XLogRecPtr remote_lsn)
+{
+ Datum roname_d;
+ bool nulls[Natts_pg_replication_origin];
+ Datum values[Natts_pg_replication_origin];
+ HeapTuple tuple;
+ Relation rel;
+
+ Assert(IsTransactionState());
+
+ /*
+ * 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(roname) > 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));
+
+ roname_d = CStringGetTextDatum(roname);
+
+ if (SearchSysCacheExists1(REPLORIGIDENT, ObjectIdGetDatum(roident)))
+ ereport(ERROR,
+ errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("replication origin with ID %u already exists",
+ (Oid) roident));
+
+ if (SearchSysCacheExists1(REPLORIGNAME, roname_d))
+ ereport(ERROR,
+ errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("replication origin \"%s\" already exists", roname));
+
+ rel = table_open(ReplicationOriginRelationId, RowExclusiveLock);
+
+ 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)
+ replorigin_advance(roident, remote_lsn, InvalidXLogRecPtr,
+ false /* backward */,
+ false /* WAL log */);
+
+ table_close(rel, RowExclusiveLock);
+}
+
/*
* Create a replication origin.
*
@@ -273,13 +340,12 @@ 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;
+ bool found = false;
/*
* To avoid needing a TOAST table for pg_replication_origin, we limit
@@ -293,8 +359,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 +385,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 +403,28 @@ 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();
+ found = true;
break;
}
}
- /* now release lock again, */
- table_close(rel, ExclusiveLock);
-
- if (tuple == NULL)
+ if (!found)
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("could not find free replication origin ID")));
- heap_freetuple(tuple);
+ table_close(rel, ExclusiveLock);
+
+ replorigin_create_with_id(roident, roname, InvalidXLogRecPtr);
+
return roident;
}
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index 59c3e7f0146..16b7408416e 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -11,6 +11,7 @@
#include "postgres.h"
+#include "access/genam.h"
#include "access/relation.h"
#include "access/table.h"
#include "catalog/binary_upgrade.h"
@@ -27,8 +28,10 @@
#include "storage/lmgr.h"
#include "utils/array.h"
#include "utils/builtins.h"
+#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
#include "utils/pg_lsn.h"
+#include "utils/snapmgr.h"
#define CHECK_IS_BINARY_UPGRADE \
@@ -377,71 +380,60 @@ 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)
{
- CHECK_IS_BINARY_UPGRADE;
+ Oid node_oid;
+ ReplOriginId node;
+ char *originname;
+ XLogRecPtr remote_lsn = InvalidXLogRecPtr;
- CreateConflictDetectionSlot();
+ CHECK_IS_BINARY_UPGRADE;
- ReplicationSlotRelease();
+ if (PG_ARGISNULL(0) || PG_ARGISNULL(1))
+ elog(ERROR,
+ "null argument to binary_upgrade_create_replication_origin is not allowed");
- PG_RETURN_VOID();
+ node_oid = PG_GETARG_OID(0);
+
+ if (node_oid == InvalidOid || node_oid > PG_UINT16_MAX)
+ 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));
+
+ if (!PG_ARGISNULL(2))
+ remote_lsn = PG_GETARG_LSN(2);
+
+ Assert(IsTransactionState());
+
+ replorigin_create_with_id(node, originname, remote_lsn);
+
+ PG_RETURN_VOID();
}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 049337de62a..fd13e165f7b 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -5668,37 +5668,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_dumpall.c b/src/bin/pg_dump/pg_dumpall.c
index c1f43113c53..b72dcb18c8f 100644
--- a/src/bin/pg_dump/pg_dumpall.c
+++ b/src/bin/pg_dump/pg_dumpall.c
@@ -25,6 +25,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"
@@ -76,6 +77,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, char *dbfile);
static void buildShSecLabels(PGconn *conn,
const char *catalog_name, Oid objectId,
@@ -813,6 +815,10 @@ main(int argc, char *argv[])
/* Dump role GUC privileges */
if (server_version >= 150000 && !skip_acls)
dumpRoleGUCPrivs(conn);
+
+ /* Dump replication origins */
+ if (binary_upgrade && archDumpFormat == archNull)
+ dumpReplicationOrigins(conn);
}
/* Dump tablespaces */
@@ -2339,6 +2345,64 @@ 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;
+
+ /* Get replication origins from catalogs */
+ appendPQExpBufferStr(buf,
+ "SELECT o.*, 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 ");
+
+ res = executeQuery(conn, buf->data);
+
+ i_roident = PQfnumber(res, "roident");
+ i_roname = PQfnumber(res, "roname");
+ i_remotelsn = PQfnumber(res, "remote_lsn");
+
+ if (PQntuples(res) > 0)
+ fprintf(OPF, "--\n-- Replication Origins \n--\n\n");
+
+ for (int i = 0; i < PQntuples(res); 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 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 f5c93e611d2..2ea2be2f422 100644
--- a/src/bin/pg_upgrade/check.c
+++ b/src/bin/pg_upgrade/check.c
@@ -33,7 +33,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);
@@ -797,7 +797,8 @@ check_new_cluster(void)
check_new_cluster_replication_slots();
- check_new_cluster_subscription_configuration();
+ check_new_cluster_replication_origins();
+
}
@@ -2303,31 +2304,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';");
@@ -2335,12 +2350,13 @@ 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)
+ PQclear(res);
+
+ if (old_cluster.nrepl_origins > max_active_replication_origins)
pg_fatal("\"max_active_replication_origins\" (%d) must be greater than or equal to the number of "
- "subscriptions (%d) on the old cluster",
- max_active_replication_origins, old_cluster.nsubs);
+ "replication origins (%d) on the old cluster",
+ max_active_replication_origins, old_cluster.nrepl_origins);
- PQclear(res);
PQfinish(conn);
check_ok();
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index 37fff93892f..0ed8625eb6b 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -862,6 +862,13 @@ get_subscription_info(ClusterInfo *cluster)
cluster->sub_retain_dead_tuples = (strcmp(PQgetvalue(res, 0, i_retain_dead_tuples), "t") == 0);
PQclear(res);
+
+ res = executeQueryOrDie(conn,
+ "SELECT count(*) AS nrepl_origins "
+ "FROM pg_catalog.pg_replication_origin");
+ cluster->nrepl_origins = atoi(PQgetvalue(res, 0, 0));
+ PQclear(res);
+
PQfinish(conn);
}
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index ccd1ac0d013..77e7ca1b4cd 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -311,6 +311,7 @@ typedef struct
int num_tablespaces;
const char *tablespace_suffix; /* directory specification */
int nsubs; /* number of subscriptions */
+ int nrepl_origins; /* number of replication origins */
bool sub_retain_dead_tuples; /* whether a subscription enables
* retain_dead_tuples. */
} ClusterInfo;
diff --git a/src/bin/pg_upgrade/t/004_subscription.pl b/src/bin/pg_upgrade/t/004_subscription.pl
index 646767f2a65..e8b11d39dd0 100644
--- a/src/bin/pg_upgrade/t/004_subscription.pl
+++ b/src/bin/pg_upgrade/t/004_subscription.pl
@@ -42,7 +42,7 @@ 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
+# in the new cluster is less than the number of replication origins in the old
# cluster.
# ------------------------------------------------------
# It is sufficient to use disabled subscription to test upgrade failure.
@@ -74,7 +74,7 @@ command_checks_all(
],
1,
[
- qr/"max_active_replication_origins" \(0\) must be greater than or equal to the number of subscriptions \(1\) on the old cluster/
+ qr/"max_active_replication_origins" \(0\) must be greater than or equal to the number of replication origins \(1\) on the old cluster/
],
[qr//],
'run of pg_upgrade where the new cluster has insufficient max_active_replication_origins'
@@ -301,8 +301,30 @@ 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 OIDs (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;
+}
+
# Have the subscription in disabled state before upgrade
$old_sub->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub5 DISABLE");
@@ -378,6 +400,20 @@ regress_sub5|f|f|f),
"check that the subscription's running status, failover, and retain_dead_tuples are preserved"
);
+# Verify that replication origin OIDs 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"
+);
+for my $row (split /\n/, $post_roident_rows)
+{
+ my ($subname, $roident) = split /\|/, $row;
+ is($roident, $pre_upgrade_roident{$subname},
+ "roident preserved for subscription '$subname' 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 3a28406981d..21466d926f1 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11960,10 +11960,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',
@@ -11973,6 +11969,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..6b25a967fe9 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -55,6 +55,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);
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) latest in thread
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], [email protected]
Subject: Re: [PATCH] Preserve replication origin OIDs in pg_upgrade
In-Reply-To: <CAFPTHDbyVt_ggdi+rOP4=jBmgU=DDgYWuLxc4oQ6ux-T0H-ZBg@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