public inbox for [email protected]
help / color / mirror / Atom feedFrom: Ajin Cherian <[email protected]>
To: Zsolt Parragi <[email protected]>
Cc: [email protected]
Subject: Re: [PATCH] Preserve replication origin OIDs in pg_upgrade
Date: Wed, 1 Jul 2026 12:49:49 +1000
Message-ID: <CAFPTHDYfDHQeQshXi7GDOQz3N8YdY6SNSmpoSAbnaQPBkVfv6A@mail.gmail.com> (raw)
In-Reply-To: <CAN4CZFPdFHwnew+wxt4UcXyjMYc3aXeR0XsR6_wZ=LEMGWEJFQ@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>
On Wed, Jun 24, 2026 at 2:07 AM Rui Zhao <[email protected]> wrote:
>
> Hi Ajin,
>
> Thanks for v9, and for addressing both my earlier comments. I also went
> through the rest of v9 -- the other reviewers' comments and the
> corresponding fixes (including the replorigin_create() locking rework) look
> good to me; I didn't find any issues there.
>
> A follow-up on the version gating, though: I think it now moved a bit too
> far. get_replication_origin_info() is called unconditionally (outside the
> >= 1700 block):
>
> /* Get replication origin information */
> get_replication_origin_info(&old_cluster);
>
> and it runs "SELECT count(*) FROM pg_catalog.pg_replication_origin".
> pg_dumpall's dumpReplicationOrigins() queries the same catalog with no
> version guard either. But pg_replication_origin only exists from 9.5
> (commit 5aa2350426), while pg_upgrade still accepts old clusters back to
> 9.2:
>
> if (GET_MAJOR_VERSION(old_cluster.major_version) < 902)
> pg_fatal("This utility can only upgrade from PostgreSQL version %s
> and later.", "9.2");
>
> So upgrading from 9.2/9.3/9.4 now fails: the query hits a non-existent
> catalog and pg_fatal()s (and dumpReplicationOrigins would too).
> It should be gated on the version where origins were introduced, e.g.
>
> if (GET_MAJOR_VERSION(old_cluster.major_version) >= 905)
> get_replication_origin_info(&old_cluster);
>
> with the matching server_version >= 90500 check on the pg_dumpall side.
> That keeps the < 17 case I flagged working while not breaking 9.2-9.4
> (which have no origins to migrate anyway).
>
Thanks, I have added both the flags.
> One nit while here: there is an extra blank line after the
> check_new_cluster_replication_origins() call in check_new_cluster().
>
Removed.
On Wed, Jun 24, 2026 at 3:09 PM shveta malik <[email protected]> wrote:
>
> On Tue, Jun 23, 2026 at 5:20 PM Ajin Cherian <[email protected]> wrote:
> >
> > All the above changes have been addressed in patch v9.
> >
>
> Thanks for the patches. A few ocmments:
>
> 1)
>
> + * In binary-upgrade mode, skip origin creation here. This is required to
> + * preserve the roident from the old cluster for this subscription.
>
> subscription --> subscription's origin
>
Changed.
> 2)
> replorigin_create_with_id():
>
> + if (SearchSysCacheExists1(REPLORIGNAME, roname_d))
> + ereport(ERROR,
> + errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin \"%s\" already exists", roname));
>
> Can this ever happen? IIUC, since new-cluster should not have any
> origin (as per the check introduced) and old cluster can not have
> duplicate names, we should not hit this error. Do we want to keep it
> as sanity check? If so, can we put a comment atop it.
>
I've removed it.
> 3)
> replorigin_create_with_id:
>
> + if (remote_lsn != InvalidXLogRecPtr)
> + replorigin_advance(roident, remote_lsn, InvalidXLogRecPtr,
> + false /* backward */,
> + false /* WAL log */);
>
> We do not want to call 'advance' from regular 'replorigin_create'
> flow. It is only for binary-Upgrade flow. Do you think we should have
> a sanity check here?
>
> if (remote_lsn != InvalidXLogRecPtr)
> {
> Assert(IsBinaryUpgrade)
> replorigin_advance(..)
> }
>
Done.
> 4)
> replorigin_create:
>
> + found = true;
>
> + if (!found)
> ereport(ERROR,
> (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
> errmsg("could not find free replication origin ID")));
>
> Do you think we can make use of 'collides' itself here instead of
> adding a new 'found' variable. We can move 'collides' outside of loop
> and use it to emit above error.
>
I've removed found and instead checked if the roident has reached the max limit
> 5)
> binary_upgrade_create_replication_origin:
>
> + node = (ReplOriginId) node_oid;
> +
> + if (SearchSysCacheExists1(REPLORIGIDENT, ObjectIdGetDatum(node)))
> + ereport(ERROR,
> + errcode(ERRCODE_DUPLICATE_OBJECT),
> + errmsg("replication origin with ID %u already exists",
> + (Oid) node));
>
> How can we hit this? Same comment as #2.
>
Removed
> 6)
> -check_new_cluster_subscription_configuration(void)
> +check_new_cluster_replication_origins(void)
>
> - /* Quick return if there are no subscriptions to be migrated. */
> - if (old_cluster.nsubs == 0)
> - return;
>
> - if (old_cluster.nsubs > max_active_replication_origins)
>
> With these changes, do we even need to capture old-cluster's 'nsubs'
> in get_subscription_info(). I do not see any other usage of nsubs. Let
> me know if I am missing any case.
I've removed nsubs
On Wed, Jun 24, 2026 at 3:21 PM shveta malik <[email protected]> wrote:
>
> On Wed, Jun 24, 2026 at 10:39 AM shveta malik <[email protected]> wrote:
> >
> >
> > 2)
> > replorigin_create_with_id():
> >
> > + if (SearchSysCacheExists1(REPLORIGNAME, roname_d))
> > + ereport(ERROR,
> > + errcode(ERRCODE_DUPLICATE_OBJECT),
> > + errmsg("replication origin \"%s\" already exists", roname));
> >
> > Can this ever happen? IIUC, since new-cluster should not have any
> > origin (as per the check introduced) and old cluster can not have
> > duplicate names, we should not hit this error. Do we want to keep it
> > as sanity check? If so, can we put a comment atop it.
> >
>
> Analyzed a bit more, with this change in patch002, we see an error change:
>
> Without patch:
> postgres=# SELECT pg_replication_origin_create('abcd');
> pg_replication_origin_create
> ------------------------------
> 1
> postgres=# SELECT pg_replication_origin_create('abcd');
> ERROR: duplicate key value violates unique constraint
> "pg_replication_origin_roname_index"
> DETAIL: Key (roname)=(abcd) already exists.
>
> ----
>
> With patch:
> postgres=# SELECT pg_replication_origin_create('abcd');
> pg_replication_origin_create
> ------------------------------
> 1
>
> postgres=# SELECT pg_replication_origin_create('abcd');
> ERROR: replication origin "abcd" already exists
>
>
> So without the new check also, we can prevent duplicate name insertion
> (even if that happens during upgrade, which I can not see how). The
> new error and the check is not wrong, but slightly redundant. Or let
> me know if I am missing something.
>
I've removed that check.
On Wed, Jun 24, 2026 at 11:27 PM Shlok Kyal <[email protected]> wrote:
>
> Thanks for the updated patch. I have reviewed the patch and here are
> some my comments:
>
> 1. Since we are quering only one column, do we need to use 'i_norigins'?
>
> +void
> +get_replication_origin_info(ClusterInfo *cluster)
> +{
> + PGconn *conn;
> + PGresult *res;
> + int i_norigins;
> +
> + conn = connectToServer(cluster, "template1");
> + res = executeQueryOrDie(conn, "SELECT count(*) AS norigins "
> + "FROM pg_catalog.pg_replication_origin");
> + i_norigins = PQfnumber(res, "norigins");
> +
> + cluster->nrepl_origins = atoi(PQgetvalue(res, 0, i_norigins));
> + PQclear(res);
> +
> + PQfinish(conn);
> +}
> I think we can directly use:
> cluster->nrepl_origins = atoi(PQgetvalue(res, 0, 0));
>
> Thoughts?
Removed it.
>
> 2. In the above code should we also add a check like:
> if (PQntuples(res) != 1)
> pg_fatal("could not count the number of replication origins");
>
Added.
> 3. The brackets are not required:
> + if (!tablespaces_only && !roles_only && binary_upgrade)
> + {
> + /* Dump replication origins */
> + dumpReplicationOrigins(conn);
> }
> We can move the comment '/* Dump replication origins */' above if condition.
> It will make it consistent with the existing code.
>
I've moved the comment up but keeping the brackets as it has more than
one line, that is the recommendation from Tom Lane.
> 4. I ran the tests locally and found that the test_decoding test suit
> is failing:
> not ok 10 - replorigin 249 ms
>
> diff:
> -- ensure duplicate creations fail
> SELECT pg_replication_origin_create('regress_test_decoding: regression_slot');
> -ERROR: duplicate key value violates unique constraint
> "pg_replication_origin_roname_index"
> -DETAIL: Key (roname)=(regress_test_decoding: regression_slot) already exists.
> +ERROR: replication origin "regress_test_decoding: regression_slot"
> already exists
> -- ensure inactive origin cannot be set as session one if pid is specified
> SELECT pg_replication_origin_session_setup('regress_test_decoding:
> regression_slot', -1);
> ERROR: cannot use PID -1 for inactive replication origin with ID 1
>
> I think this is expected because this patch adds a new function
> 'replorigin_create_with_id'
> which adds a check for the already existing replication origin.
> In HEAD we don't have such a check. So I think we should update the
> expected output file.
>
Based on feedback from Shveta, I've removed that new error as changing
existing behaviour isn't required.
On Thu, Jun 25, 2026 at 2:45 AM Zsolt Parragi <[email protected]> wrote:
>
> Hello
>
> +# 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"
> +);
> +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");
> +}
> +
>
> Won't this check miss (and silently pass) if an entry is missing
> post-upgrade? It verifies that all entries that exists match the pre
> upgrade entries, but it doesn't verify that the list is the same.
Modified this to iterate over pre upgrade and post upgrade
subscriptions making sure both are present.
All the above changes are updated in patch v10.
regards,
Ajin Cherian
Fujitsu Australia
Attachments:
[application/octet-stream] v10-0001-Preserve-subscription-OIDs-during-pg_upgrade.patch (9.1K, ../CAFPTHDYfDHQeQshXi7GDOQz3N8YdY6SNSmpoSAbnaQPBkVfv6A@mail.gmail.com/2-v10-0001-Preserve-subscription-OIDs-during-pg_upgrade.patch)
download | inline diff:
From cb9b9da7b14073924b635be0025fef9154203d96 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Mon, 29 Jun 2026 21:46:29 +1000
Subject: [PATCH v10 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 ee06a726f42..5fe7440418c 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.
@@ -803,8 +809,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 c56437d6057..ea22b0604c1 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 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 402d869710b..231ce2ca91e 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11968,6 +11968,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] v10-0002-Preserve-replication-origin-OIDs-during-pg_upgra.patch (33.9K, ../CAFPTHDYfDHQeQshXi7GDOQz3N8YdY6SNSmpoSAbnaQPBkVfv6A@mail.gmail.com/3-v10-0002-Preserve-replication-origin-OIDs-during-pg_upgra.patch)
download | inline diff:
From b89f8f95a970223a3fe1672946b31060c3551355 Mon Sep 17 00:00:00 2001
From: Ajin Cherian <[email protected]>
Date: Wed, 1 Jul 2026 12:39:16 +1000
Subject: [PATCH v10 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 | 11 ++-
src/backend/replication/logical/origin.c | 96 ++++++++++++--------
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 | 57 ++++++++----
src/bin/pg_upgrade/info.c | 35 +++++--
src/bin/pg_upgrade/pg_upgrade.h | 3 +-
src/bin/pg_upgrade/t/004_subscription.pl | 69 +++++++++++++-
src/include/catalog/pg_proc.dat | 8 +-
src/include/replication/origin.h | 3 +
13 files changed, 345 insertions(+), 162 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 5befefd9c5a..1eb64259ff3 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
@@ -2861,7 +2861,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 5fe7440418c..a10cda4e0e7 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"
@@ -898,9 +899,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's origin.
*/
- 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..a9868a29b14 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -265,6 +265,52 @@ 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 +319,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 +337,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 +363,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 +381,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 ea22b0604c1..e157214b926 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -5179,7 +5179,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;
@@ -5248,11 +5247,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,
@@ -5332,7 +5329,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);
@@ -5385,11 +5381,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);
@@ -5668,37 +5659,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 b9653f0aefe..485140b9332 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,6 +665,14 @@ 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)
+ {
+ if (server_version >= 90500)
+ dumpReplicationOrigins(conn);
}
/* Dump tablespaces */
@@ -1832,6 +1842,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 7556fb3f22a..ea8451d6071 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);
@@ -651,6 +651,10 @@ check_and_dump_old_cluster(void)
check_old_cluster_subscription_state();
}
+ /* Get replication origin information for PG 9.5 and above */
+ if (GET_MAJOR_VERSION(old_cluster.major_version) >= 905)
+ get_replication_origin_info(&old_cluster);
+
check_for_data_types_usage(&old_cluster);
/*
@@ -797,7 +801,7 @@ check_new_cluster(void)
check_new_cluster_replication_slots();
- check_new_cluster_subscription_configuration();
+ check_new_cluster_replication_origins();
}
@@ -2303,31 +2307,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 +2353,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) in the old cluster",
- max_active_replication_origins, old_cluster.nsubs);
+ "replication origins (%d) in 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..ec5dd5b7e96 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -832,6 +832,30 @@ 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(*) AS norigins "
+ "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);
+
+ PQfinish(conn);
+}
+
/*
* get_subscription_info()
*
@@ -842,23 +866,20 @@ 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 "
+ res = executeQueryOrDie(conn,
+ "SELECT 'f' AS retain_dead_tuples "
"FROM pg_catalog.pg_subscription");
- i_nsub = PQfnumber(res, "nsub");
i_retain_dead_tuples = PQfnumber(res, "retain_dead_tuples");
- cluster->nsubs = atoi(PQgetvalue(res, 0, i_nsub));
cluster->sub_retain_dead_tuples = (strcmp(PQgetvalue(res, 0, i_retain_dead_tuples), "t") == 0);
PQclear(res);
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index ccd1ac0d013..e79765bf4fe 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -310,7 +310,7 @@ 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 */
bool sub_retain_dead_tuples; /* whether a subscription enables
* retain_dead_tuples. */
} ClusterInfo;
@@ -451,6 +451,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..e09450fc335 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\) in the old cluster/
+ qr/"max_active_replication_origins" \(0\) must be greater than or equal to the number of replication origins \(1\) in the old cluster/
],
[qr//],
'run of pg_upgrade where the new cluster has insufficient max_active_replication_origins'
@@ -301,8 +301,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 +408,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 231ce2ca91e..4d930641522 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11959,10 +11959,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',
@@ -11972,6 +11968,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) 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]
Subject: Re: [PATCH] Preserve replication origin OIDs in pg_upgrade
In-Reply-To: <CAFPTHDYfDHQeQshXi7GDOQz3N8YdY6SNSmpoSAbnaQPBkVfv6A@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