From bac303e2d5be9bd4a6161e907b69404b761db284 Mon Sep 17 00:00:00 2001 From: Ajin Cherian Date: Tue, 3 Nov 2026 14:53:22 +1100 Subject: [PATCH v22 2/2] Preserve replication origin IDs during pg_upgrade Currently pg_upgrade does not migrate replication origins as such. Instead, CreateSubscription() creates a new origin for each subscription, which takes the lowest free roident in the new cluster, and for old clusters of version 17 or later the subscription's remote_lsn is then restored by binary_upgrade_replorigin_advance(). Origins not belonging to a subscription, such as those created with pg_replication_origin_create() by external replication tools, are lost altogether. As a result an origin's roident can change across the upgrade. The roident is recorded in commit timestamp data, so this is a prerequisite for preserving pg_commit_ts across pg_upgrade. If commit timestamps were carried over without it, the recorded origins would no longer match, causing spurious "update_origin_differs" conflicts after the upgrade. Fix this by dumping all replication origins as global objects via pg_dumpall in binary-upgrade mode, using a new function binary_upgrade_create_replication_origin(oid, name, lsn) to recreate each origin with its original roident and, if it was being tracked, its remote_lsn. This replaces binary_upgrade_replorigin_advance(), which is removed. CreateSubscription() no longer creates the origin in binary-upgrade mode, since it has already been restored. Because origins are now migrated from every supported old cluster version, check_old_cluster_subscription_state() now runs for all of them, not only 17 and later. It rejects subscriptions whose origin is missing, and relations not in the 'i' or 'r' state, whose tablesync origins would otherwise be migrated and never dropped (before 17, pg_subscription_rel is not migrated, so nothing in the new cluster would record that they belong to the subscription). This changes some behaviour visible to users: * The subscription state checks (missing origin, relations not in 'i' or 'r' state) now also apply to old clusters before version 17. * Origins are now migrated for old clusters before version 17 as well. Subscriptions from such clusters are still restored disabled, but once enabled they resume from the migrated remote_lsn rather than from the publisher slot's confirmed_flush_lsn. * The new cluster must not contain any replication origins. * The max_active_replication_origins check now compares against the number of origins with progress tracking state in the old cluster (those in pg_replication_origin_status), instead of the number of subscriptions. Author: Ajin Cherian Reviewer: Hayato Kuroda (Fujitsu) Reviewer: Zsolt Parragi Reviewer: Shlok Kyal Reviewer: Shveta malik Reviewer: Vignesh C Reviewer: Nisha Moond Reviewer: Rui Zhao --- doc/src/sgml/logical-replication.sgml | 26 +++- src/backend/commands/subscriptioncmds.c | 12 +- src/backend/replication/logical/origin.c | 86 +++++++++---- src/backend/utils/adt/pg_upgrade_support.c | 107 +++++++++------- src/bin/pg_dump/pg_dump.c | 56 ++------ src/bin/pg_dump/pg_dump.h | 1 - src/bin/pg_dump/pg_dumpall.c | 74 +++++++++++ src/bin/pg_upgrade/check.c | 89 +++++++++---- src/bin/pg_upgrade/info.c | 53 ++++++-- src/bin/pg_upgrade/pg_upgrade.c | 18 ++- src/bin/pg_upgrade/pg_upgrade.h | 5 +- src/bin/pg_upgrade/t/004_subscription.pl | 141 +++++++++++++++++++-- src/include/catalog/pg_proc.dat | 8 +- src/include/replication/origin.h | 6 + 14 files changed, 499 insertions(+), 183 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 4701a3d9d18..7500ba67f72 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2862,16 +2862,24 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER pg_upgrade attempts to migrate subscription dependencies which includes the subscription's table information present in pg_subscription_rel - system catalog and also the subscription's replication origin. This allows - logical replication on the new subscriber to continue from where the - old subscriber was up to. Migration of subscription dependencies is only - supported when the old cluster is version 17.0 or later. Subscription - dependencies on clusters before version 17.0 will silently be ignored. + system catalog and also the subscription's replication origin. When the + old cluster is version 17.0 or later, this allows logical replication on + the new subscriber to continue from where the old subscriber was up to. + All replication origins, including user-created ones, are migrated with + their original IDs and their replication progress + (remote_lsn). For clusters before version + 17.0, the + subscription's table information is not migrated and subscriptions are + restored in a disabled state, so replication does not resume + automatically after the upgrade. When such a subscription is enabled, + replication resumes from the position recorded in its migrated + replication origin. - Commit timestamps and origin data are not preserved during the upgrade. + Commit timestamps (which include the origin of the commit) 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 @@ -2913,7 +2921,11 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER The new cluster must have max_active_replication_origins configured to a value greater than or equal to the number of - subscriptions present in the old cluster. + replication origins in the old cluster whose progress is being tracked, + that is, those that appear in + pg_replication_origin_status. + Furthermore, the new cluster must not contain any pre-existing replication + origins. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 15d4208d6bc..4435a5f3a51 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -982,9 +982,17 @@ 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, the origin has already been restored by + * pg_dumpall with its roident from the old cluster, so don't create it + * again. pg_upgrade rejects subscriptions whose origin is missing. */ - 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 5cb18d851ae..f1de0a8e0ff 100644 --- a/src/backend/replication/logical/origin.c +++ b/src/backend/replication/logical/origin.c @@ -265,6 +265,60 @@ replorigin_by_name(const char *roname, bool missing_ok) return roident; } +/* + * replorigin_create_with_id + * + * Insert a replication origin with the given ID and name into + * pg_replication_origin. The caller is responsible for ensuring that the ID + * is not already in use and that the name is valid. + * + * If need_advance is true, the origin's progress is also set to remote_lsn, + * which creates its entry in the shared progress tracking state. This is + * allowed only in binary upgrade mode. + * + * rel must be pg_replication_origin, already opened by the caller with + * ExclusiveLock. Needs to be called in a transaction. + */ +void +replorigin_create_with_id(ReplOriginId roident, const char *roname, + XLogRecPtr remote_lsn, bool need_advance, + 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 requested, set up the origin's progress tracking state with the + * given remote_lsn. need_advance is separate from remote_lsn because an + * InvalidXLogRecPtr remote_lsn is a legitimate value to restore. + */ + if (need_advance) + { + Assert(IsBinaryUpgrade); + replorigin_advance(roident, remote_lsn, InvalidXLogRecPtr, + false /* backward */ , + false /* WAL log */ ); + } +} + /* * Create a replication origin. * @@ -274,9 +328,7 @@ ReplOriginId replorigin_create(const char *roname) { Oid roident; - HeapTuple tuple = NULL; Relation rel; - Datum roname_d; SnapshotData SnapshotDirty; SysScanDesc scan; ScanKeyData key; @@ -293,8 +345,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()); /* @@ -329,8 +379,6 @@ replorigin_create(const char *roname) for (roident = InvalidOid + 1; roident < PG_UINT16_MAX; roident++) { - bool nulls[Natts_pg_replication_origin]; - Datum values[Natts_pg_replication_origin]; bool collides; CHECK_FOR_INTERRUPTS(); @@ -344,38 +392,22 @@ replorigin_create(const char *roname) 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, false, 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 9152f5cfd76..df30d088578 100644 --- a/src/backend/utils/adt/pg_upgrade_support.c +++ b/src/backend/utils/adt/pg_upgrade_support.c @@ -388,71 +388,88 @@ 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]; + CHECK_IS_BINARY_UPGRADE; + + CreateConflictDetectionSlot(); + + ReplicationSlotRelease(); + + PG_RETURN_VOID(); +} + +/* + * binary_upgrade_create_replication_origin + * + * Create a replication origin with a specific roident and name, optionally + * restoring its remote_lsn. Used by pg_upgrade to preserve replication + * origin IDs across the upgrade, since roident values are recorded in + * commit timestamp data. + * + * A NULL remote_lsn means the origin had no entry in + * pg_replication_origin_status on the old cluster, so no progress tracking + * state is created for it. This must be distinguished from a non-NULL + * remote_lsn of 0/0, which means the origin was being tracked (and so counts + * against max_active_replication_origins) but had not yet advanced. Both + * would otherwise reach us as InvalidXLogRecPtr. + */ +Datum +binary_upgrade_create_replication_origin(PG_FUNCTION_ARGS) +{ + Oid node_oid; ReplOriginId node; - XLogRecPtr remote_commit; + Relation rel; + char *originname; + XLogRecPtr remote_lsn = InvalidXLogRecPtr; 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"); + if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) + elog(ERROR, + "null argument to binary_upgrade_create_replication_origin is not allowed"); - subname = text_to_cstring(PG_GETARG_TEXT_PP(0)); - remote_commit = PG_ARGISNULL(1) ? InvalidXLogRecPtr : PG_GETARG_LSN(1); + node_oid = PG_GETARG_OID(0); - rel = table_open(SubscriptionRelationId, RowExclusiveLock); - subid = get_subscription_oid(subname, false); + 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)); - ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname)); + node = (ReplOriginId) node_oid; - /* Lock to prevent the replication origin from vanishing */ - LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock); - node = replorigin_by_name(originname, false); + originname = text_to_cstring(PG_GETARG_TEXT_PP(1)); /* - * 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. + * 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. */ - replorigin_advance(node, remote_commit, InvalidXLogRecPtr, - false /* backward */ , - false /* WAL log */ ); + 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)); - UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock); - table_close(rel, RowExclusiveLock); + if (!PG_ARGISNULL(2)) + remote_lsn = PG_GETARG_LSN(2); - PG_RETURN_VOID(); -} + Assert(IsTransactionState()); -/* - * binary_upgrade_create_conflict_detection_slot - * - * Create a replication slot to retain information necessary for conflict - * detection such as dead tuples, commit timestamps, and origins. - */ -Datum -binary_upgrade_create_conflict_detection_slot(PG_FUNCTION_ARGS) -{ - CHECK_IS_BINARY_UPGRADE; + /* Acquire an exclusive lock before inserting the new origin. */ + rel = table_open(ReplicationOriginRelationId, ExclusiveLock); - CreateConflictDetectionSlot(); + replorigin_create_with_id(node, originname, remote_lsn, + !PG_ARGISNULL(2), rel); - ReplicationSlotRelease(); + 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 8ecc85f0736..7addcb469ed 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -5131,7 +5131,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; @@ -5200,11 +5199,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, @@ -5246,11 +5243,6 @@ getSubscriptions(Archive *fout) "LEFT JOIN pg_catalog.pg_foreign_server fs \n" " ON fs.oid = s.subserver \n"); - if (dopt->binary_upgrade && fout->remoteVersion >= 170000) - appendPQExpBufferStr(query, - "LEFT JOIN pg_catalog.pg_replication_origin_status o \n" - " ON o.external_id = 'pg_' || s.oid::text \n"); - appendPQExpBufferStr(query, "WHERE s.subdbid = (SELECT oid FROM pg_database\n" " WHERE datname = current_database())"); @@ -5284,7 +5276,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); @@ -5337,11 +5328,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); @@ -5620,37 +5606,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 4dcb6c854e8..425fa3d7c52 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -727,7 +727,6 @@ typedef struct _SubscriptionInfo char *subwalrcvtimeout; char *subpublications; char *suborigin; - char *suboriginremotelsn; } SubscriptionInfo; /* diff --git a/src/bin/pg_dump/pg_dumpall.c b/src/bin/pg_dump/pg_dumpall.c index c53e77c2878..68121f7fe7c 100644 --- a/src/bin/pg_dump/pg_dumpall.c +++ b/src/bin/pg_dump/pg_dumpall.c @@ -18,6 +18,7 @@ #include #include +#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, @@ -665,6 +667,10 @@ main(int argc, char *argv[]) dumpRoleGUCPrivs(conn); } + /* Dump replication origins */ + if (!tablespaces_only && !roles_only && binary_upgrade) + dumpReplicationOrigins(conn); + /* Dump tablespaces */ if (!roles_only && !no_tablespaces) dumpTablespaces(conn); @@ -1804,6 +1810,74 @@ dumpTimestamp(const char *msg) fprintf(OPF, "-- %s %s\n\n", msg, buf); } +/* + * dumpReplicationOrigins + * + * Dump all replication origins, for use by pg_upgrade only. Each origin is + * recreated with its original roident and, if it was being tracked, its + * remote_lsn. Origins are cluster-wide objects, so they are dumped here + * rather than by pg_dump, and they are restored before any subscriptions + * are created. Subscription OIDs are preserved too, so the restored origins + * keep matching their subscriptions by name ("pg_"). + */ +static void +dumpReplicationOrigins(PGconn *conn) +{ + PQExpBuffer buf = createPQExpBuffer(); + PGresult *res; + int i_roident; + int i_roname; + int i_remotelsn; + int ntups; + + /* Get replication origins from catalogs */ + appendPQExpBufferStr(buf, + "SELECT o.roident, o.roname, os.remote_lsn " + "FROM pg_catalog.pg_replication_origin o " + "LEFT JOIN pg_catalog.pg_replication_origin_status os ON o.roident = os.local_id " + "ORDER BY o.roident"); + + res = executeQuery(conn, buf->data); + + i_roident = PQfnumber(res, "roident"); + i_roname = PQfnumber(res, "roname"); + i_remotelsn = PQfnumber(res, "remote_lsn"); + ntups = PQntuples(res); + + if (ntups > 0) + fprintf(OPF, "--\n-- Replication Origins\n--\n\n"); + + for (int i = 0; i < ntups; i++) + { + ReplOriginId roident = atooid(PQgetvalue(res, i, i_roident)); + const char *roname = PQgetvalue(res, i, i_roname); + + resetPQExpBuffer(buf); + + appendPQExpBufferStr(buf, "\n-- For binary upgrade, must preserve pg_replication_origin roident and remote_lsn\n"); + appendPQExpBuffer(buf, + "SELECT pg_catalog.binary_upgrade_create_replication_origin(" + "'%u'::pg_catalog.oid, ", roident); + appendStringLiteralConn(buf, roname, conn); + appendPQExpBufferStr(buf, "::pg_catalog.text"); + + if (!PQgetisnull(res, i, i_remotelsn)) + { + appendPQExpBufferStr(buf, ", "); + appendStringLiteralConn(buf, PQgetvalue(res, i, i_remotelsn), conn); + appendPQExpBufferStr(buf, "::pg_catalog.pg_lsn"); + } + else + appendPQExpBufferStr(buf, ", NULL::pg_catalog.pg_lsn"); + + 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 150c4fc34b9..bf2d2cc27e3 100644 --- a/src/bin/pg_upgrade/check.c +++ b/src/bin/pg_upgrade/check.c @@ -34,7 +34,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); @@ -624,9 +624,18 @@ check_and_dump_old_cluster(void) * able to upgrade the logical replication clusters completely. */ get_subscription_info(&old_cluster); - check_old_cluster_subscription_state(); } + /* + * Replication origins are migrated for all supported old cluster + * versions, so check for subscriptions whose origin is missing or that + * could leave a stale tablesync origin behind, regardless of version. + */ + check_old_cluster_subscription_state(); + + /* Get replication origin information */ + get_replication_origin_info(&old_cluster); + check_for_data_types_usage(&old_cluster); /* @@ -745,7 +754,7 @@ check_new_cluster(void) check_new_cluster_replication_slots(); - check_new_cluster_subscription_configuration(); + check_new_cluster_replication_origins(); } @@ -2267,31 +2276,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 active 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) - return; - - /* Quick return if there are no subscriptions to be migrated. */ - if (old_cluster.nsubs == 0) + /* Quick return if there are no replication origins to migrate. */ + if (old_cluster.nrepl_origins == 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';"); @@ -2299,12 +2322,21 @@ check_new_cluster_subscription_configuration(void) pg_fatal("could not determine parameter settings on new cluster"); max_active_replication_origins = atoi(PQgetvalue(res, 0, 0)); - if (old_cluster.nsubs > max_active_replication_origins) - pg_fatal("\"max_active_replication_origins\" (%d) must be greater than or equal to the number of " - "subscriptions (%d) in the old cluster", - max_active_replication_origins, old_cluster.nsubs); - PQclear(res); + + /* + * Compare against the number of active replication origins in the old + * cluster, rather than the total number of origins in + * pg_replication_origin. An origin can exist without ever becoming + * active, in which case it does not count against + * max_active_replication_origins. + */ + if (old_cluster.nrepl_origins_active > max_active_replication_origins) + pg_fatal("\"max_active_replication_origins\" (%d) must be greater " + "than or equal to the number of active replication origins " + "(%d) in the old cluster", + max_active_replication_origins, old_cluster.nrepl_origins_active); + PQfinish(conn); check_ok(); @@ -2492,8 +2524,8 @@ check_old_cluster_subscription_state(void) * * a) SUBREL_STATE_DATASYNC: A relation upgraded while in this state would * retain a replication slot and origin. The sync worker spawned after the - * upgrade cannot drop them because the subscription ID used for the slot - * and origin name no longer matches. + * upgrade cannot drop the slot because its name embeds the system + * identifier, which differs in the new cluster. * * b) SUBREL_STATE_SYNCDONE: A relation upgraded while in this state would * retain the replication origin when there is a failure in tablesync @@ -2501,9 +2533,14 @@ check_old_cluster_subscription_state(void) * publisher. * * c) SUBREL_STATE_FINISHEDCOPY: A tablesync worker spawned to work on a - * relation upgraded while in this state would expect an origin ID with - * the OID of the subscription used before the upgrade, causing it to - * fail. + * relation upgraded while in this state would continue using the slot + * created before the upgrade, which it cannot find or drop by name for + * the same reason as in a). + * + * Before PG17, pg_subscription_rel is not migrated at all. These states + * are still rejected there because the tablesync origin would be + * migrated, but nothing in the new cluster would record that it belongs + * to the subscription, so it would never be dropped. * * d) SUBREL_STATE_SYNCWAIT, SUBREL_STATE_CATCHUP and * SUBREL_STATE_UNKNOWN: These states are not stored in the catalog, so we diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c index 5c59cfb32e8..1fb8cfc01aa 100644 --- a/src/bin/pg_upgrade/info.c +++ b/src/bin/pg_upgrade/info.c @@ -822,33 +822,68 @@ count_old_cluster_logical_slots(void) return slot_count; } +/* + * get_replication_origin_info() + * + * Gets the information of replication origins in the cluster. + */ +void +get_replication_origin_info(ClusterInfo *cluster) +{ + PGconn *conn; + PGresult *res; + + conn = connectToServer(cluster, "template1"); + + res = executeQueryOrDie(conn, "SELECT count(*) " + "FROM pg_catalog.pg_replication_origin"); + if (PQntuples(res) != 1) + pg_fatal("could not get the number of replication origins"); + + cluster->nrepl_origins = atoi(PQgetvalue(res, 0, 0)); + PQclear(res); + + /* + * Get the number of actively tracked replication origins, i.e., those + * with an entry in pg_replication_origin_status. This count can be less + * than nrepl_origins, since a replication origin can exist without ever + * becoming active. This count is what is actually bounded by + * max_active_replication_origins. + */ + res = executeQueryOrDie(conn, "SELECT count(*) " + "FROM pg_catalog.pg_replication_origin_status"); + if (PQntuples(res) != 1) + pg_fatal("could not get the number of active replication origins"); + + cluster->nrepl_origins_active = atoi(PQgetvalue(res, 0, 0)); + PQclear(res); + + PQfinish(conn); +} + /* * get_subscription_info() * - * Gets the information of subscriptions in the cluster. + * Determine whether any subscription has subretaindeadtuples enabled. */ void get_subscription_info(ClusterInfo *cluster) { PGconn *conn; PGresult *res; - int i_nsub; int i_retain_dead_tuples; conn = connectToServer(cluster, "template1"); if (GET_MAJOR_VERSION(cluster->major_version) >= 1900) - res = executeQueryOrDie(conn, "SELECT count(*) AS nsub," - "COUNT(CASE WHEN subretaindeadtuples THEN 1 END) > 0 AS retain_dead_tuples " + res = executeQueryOrDie(conn, + "SELECT COUNT(CASE WHEN subretaindeadtuples THEN 1 END) > 0 AS retain_dead_tuples " "FROM pg_catalog.pg_subscription"); else - res = executeQueryOrDie(conn, "SELECT count(*) AS nsub," - "'f' AS retain_dead_tuples " - "FROM pg_catalog.pg_subscription"); + res = executeQueryOrDie(conn, + "SELECT 'f' AS retain_dead_tuples "); - i_nsub = PQfnumber(res, "nsub"); i_retain_dead_tuples = PQfnumber(res, "retain_dead_tuples"); - cluster->nsubs = atoi(PQgetvalue(res, 0, i_nsub)); cluster->sub_retain_dead_tuples = (strcmp(PQgetvalue(res, 0, i_retain_dead_tuples), "t") == 0); PQclear(res); diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c index 4db1b7084d8..502305b4e07 100644 --- a/src/bin/pg_upgrade/pg_upgrade.c +++ b/src/bin/pg_upgrade/pg_upgrade.c @@ -38,6 +38,11 @@ * * We control assignment of pg_subscription.oid because we want the oid to * match between the old and new cluster. + * + * We control all assignments of pg_replication_origin.roident because + * roident values are recorded in commit timestamp data, and because + * subscription origins are linked to their subscriptions by a name that + * embeds pg_subscription.oid. */ @@ -230,13 +235,12 @@ main(int argc, char **argv) * as it only retains the dead tuples. It is created here for consistency. * Note that the new conflict detection slot uses the latest transaction * ID as xmin, so it cannot protect dead tuples that existed before the - * upgrade. Additionally, commit timestamps and origin data are not - * preserved during the upgrade. So, even after creating the slot, the - * upgraded subscriber may be unable to detect conflicts or log relevant - * commit timestamps and origins when applying changes from the publisher - * occurred before the upgrade especially if those changes were not - * replicated. It can only protect tuples that might be deleted after the - * new cluster starts. + * upgrade. Additionally, commit timestamps are not preserved during the + * upgrade. So, even after creating the slot, the upgraded subscriber may + * be unable to detect conflicts or log relevant commit timestamps and + * origins when applying changes from the publisher occurred before the + * upgrade especially if those changes were not replicated. It can only + * protect tuples that might be deleted after the new cluster starts. */ if (migrate_logical_slots || old_cluster.sub_retain_dead_tuples) { diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h index 30a2e263d2d..85bb9bd0eb1 100644 --- a/src/bin/pg_upgrade/pg_upgrade.h +++ b/src/bin/pg_upgrade/pg_upgrade.h @@ -286,7 +286,9 @@ typedef struct char **tablespaces; /* tablespace directories */ int num_tablespaces; const char *tablespace_suffix; /* directory specification */ - int nsubs; /* number of subscriptions */ + int nrepl_origins; /* number of replication origins */ + int nrepl_origins_active; /* number of replication origins that + * are active */ bool sub_retain_dead_tuples; /* whether a subscription enables * retain_dead_tuples. */ } ClusterInfo; @@ -427,6 +429,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 48324fd103e..f5e19992642 100644 --- a/src/bin/pg_upgrade/t/004_subscription.pl +++ b/src/bin/pg_upgrade/t/004_subscription.pl @@ -42,15 +42,23 @@ my $connstr = $publisher->connstr . ' dbname=postgres'; # ------------------------------------------------------ # Check that pg_upgrade fails when max_active_replication_origins configured -# in the new cluster is less than the number of subscriptions in the old -# cluster. +# in the new cluster is less than the number of active replication origins in +# the old cluster. # ------------------------------------------------------ -# It is sufficient to use disabled subscription to test upgrade failure. $publisher->safe_psql('postgres', "CREATE PUBLICATION regress_pub1"); $old_sub->safe_psql('postgres', - "CREATE SUBSCRIPTION regress_sub1 CONNECTION '$connstr' PUBLICATION regress_pub1 WITH (enabled = false)" + "CREATE SUBSCRIPTION regress_sub1 CONNECTION '$connstr' PUBLICATION regress_pub1" ); +# Wait until the apply worker sets up the replication origin, making it +# active. Otherwise, stopping the old cluster too early could leave the +# origin inactive, causing the check below to see no active replication +# origins and fail. +$old_sub->poll_query_until('postgres', + "SELECT count(*) > 0 FROM pg_replication_origin_status") + or die + "Timed out while waiting for the replication origin to become active"; + $old_sub->stop; $new_sub->append_conf('postgresql.conf', @@ -74,7 +82,7 @@ command_checks_all( ], 1, [ - qr/"max_active_replication_origins" \(0\) must be greater than or equal to the number of subscriptions \(1\) in the old cluster/ + qr/"max_active_replication_origins" \(0\) must be greater than or equal to the number of active replication origins \(1\) in the old cluster/ ], [qr//], 'run of pg_upgrade where the new cluster has insufficient max_active_replication_origins' @@ -259,6 +267,15 @@ $publisher->safe_psql( CREATE PUBLICATION regress_pub4 FOR TABLE tab_upgraded, tab_upgraded1; ]); +# Create a filler replication origin before the subscriptions so that their +# origins are not assigned the lowest free origin IDs. The filler is dropped +# after all the origins have been created, leaving a gap in the origin IDs. +# Without preservation, the new cluster would assign fresh (lowest free) IDs, +# which would differ from the old ones, so the roident checks below can +# detect a failure to preserve them. +$old_sub->safe_psql('postgres', + "SELECT pg_replication_origin_create('regress_filler_origin')"); + $old_sub->safe_psql( 'postgres', qq[ CREATE TABLE tab_upgraded(id int); @@ -300,9 +317,56 @@ my $result = $old_sub->safe_psql('postgres', 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'" +my $remote_lsn = $old_sub->safe_psql( + 'postgres', + "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'" +); + +# An origin advanced to exactly 0/0 is tracked on the old cluster and must stay +# tracked. This is the case that the NULL-ness of the remote_lsn argument +# exists to catch: a missing status row and a status row at 0/0 both reach the +# new cluster as InvalidXLogRecPtr. +my $zero_origin_name = 'regress_zero_origin'; +$old_sub->safe_psql('postgres', + "SELECT pg_replication_origin_create('$zero_origin_name')"); +$old_sub->safe_psql('postgres', + "SELECT pg_replication_origin_advance('$zero_origin_name', '0/0')"); + +# Drop the filler origin to leave a gap in the origin IDs. See above. +$old_sub->safe_psql('postgres', + "SELECT pg_replication_origin_drop('regress_filler_origin')"); + # Have the subscription in disabled state before upgrade $old_sub->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub5 DISABLE"); @@ -352,7 +416,9 @@ ok( !-d $new_sub->data_dir . "/pg_upgrade_output.d", # Check that the data inserted to the publisher when the new subscriber is down # will be replicated once it is started. Also check that the old subscription # states and relations origins are all preserved, and that the conflict -# detection slot is created. +# detection slot is created. In addition, verify that the replication origin +# identifiers (roident) for both subscription-related and user-created +# replication origins are preserved after the upgrade. # ------------------------------------------------------ $publisher->safe_psql( 'postgres', qq[ @@ -379,6 +445,65 @@ regress_sub5|f|f|f), "check that the subscription's running status, failover, and retain_dead_tuples are preserved" ); +# Verify that subscription replication origins retain their origin IDs 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 the user-created replication origin retains its origin ID 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" +); + +# Origins never advanced on the old cluster exist on the new cluster but do not +# appear in pg_replication_origin_status. +$result = $new_sub->safe_psql( + 'postgres', qq[ + SELECT s.local_id IS NULL FROM pg_replication_origin o + LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id + WHERE o.roname = '$user_origin_name']); +is($result, qq(t), + "never-advanced origin '$user_origin_name' exists and is still untracked after upgrade" +); + +# The origin advanced to 0/0 is still present in pg_replication_origin_status +# on the new cluster. +$result = $new_sub->safe_psql( + 'postgres', qq[ + SELECT count(*) FROM pg_replication_origin_status s + JOIN pg_replication_origin o ON o.roident = s.local_id + WHERE o.roname = '$zero_origin_name']); +is($result, qq(1), + "origin '$zero_origin_name' advanced to 0/0 is still tracked 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 d50079e65d1..2873c328317 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -12050,10 +12050,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', @@ -12063,6 +12059,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..145bd7cd21f 100644 --- a/src/include/replication/origin.h +++ b/src/include/replication/origin.h @@ -15,6 +15,9 @@ #include "access/xlogreader.h" #include "catalog/pg_replication_origin.h" +/* avoid including utils/relcache.h */ +typedef struct RelationData *Relation; + typedef struct xl_replorigin_set { XLogRecPtr remote_lsn; @@ -55,6 +58,9 @@ 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, bool need_advance, + 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