From 1e750a58fd4e247364b7c606869aae67287f84a1 Mon Sep 17 00:00:00 2001 From: Ajin Cherian 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 Reviewer: Hayato Kuroda (Fujitsu) Reviewer: Zsolt Parragi Reviewer: Shlok Kyal Reviewer: Shveta malik Reviewer: Vignesh C --- 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 - Commit timestamps and origin data are not preserved during the upgrade. + Commit timestamps are not preserved during the upgrade. As a result, even if retain_dead_tuples 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 #include +#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