mirror of
https://github.com/postgres/postgres.git
synced 2025-11-22 12:22:45 +03:00
While reading information from the old cluster, a list of logical slots is fetched. At the later part of upgrading, pg_upgrade revisits the list and restores slots by executing pg_create_logical_replication_slot() on the new cluster. Migration of logical replication slots is only supported when the old cluster is version 17.0 or later. If the old node has invalid slots or slots with unconsumed WAL records, the pg_upgrade fails. These checks are needed to prevent data loss. The significant advantage of this commit is that it makes it easy to continue logical replication even after upgrading the publisher node. Previously, pg_upgrade allowed copying publications to a new node. With this patch, adjusting the connection string to the new publisher will cause the apply worker on the subscriber to connect to the new publisher automatically. This enables seamless continuation of logical replication, even after an upgrade. Author: Hayato Kuroda, Hou Zhijie Reviewed-by: Peter Smith, Bharath Rupireddy, Dilip Kumar, Vignesh C, Shlok Kyal Discussion: http://postgr.es/m/TYAPR01MB58664C81887B3AF2EB6B16E3F5939@TYAPR01MB5866.jpnprd01.prod.outlook.com Discussion: http://postgr.es/m/CAA4eK1+t7xYcfa0rEQw839=b2MzsfvYDPz3xbD+ZqOdP3zpKYg@mail.gmail.com
308 lines
6.7 KiB
C
308 lines
6.7 KiB
C
/*
|
|
* pg_upgrade_support.c
|
|
*
|
|
* server-side functions to set backend global variables
|
|
* to control oid and relfilenumber assignment, and do other special
|
|
* hacks needed for pg_upgrade.
|
|
*
|
|
* Copyright (c) 2010-2023, PostgreSQL Global Development Group
|
|
* src/backend/utils/adt/pg_upgrade_support.c
|
|
*/
|
|
|
|
#include "postgres.h"
|
|
|
|
#include "catalog/binary_upgrade.h"
|
|
#include "catalog/heap.h"
|
|
#include "catalog/namespace.h"
|
|
#include "catalog/pg_type.h"
|
|
#include "commands/extension.h"
|
|
#include "miscadmin.h"
|
|
#include "replication/logical.h"
|
|
#include "utils/array.h"
|
|
#include "utils/builtins.h"
|
|
|
|
|
|
#define CHECK_IS_BINARY_UPGRADE \
|
|
do { \
|
|
if (!IsBinaryUpgrade) \
|
|
ereport(ERROR, \
|
|
(errcode(ERRCODE_CANT_CHANGE_RUNTIME_PARAM), \
|
|
errmsg("function can only be called when server is in binary upgrade mode"))); \
|
|
} while (0)
|
|
|
|
Datum
|
|
binary_upgrade_set_next_pg_tablespace_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid tbspoid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_pg_tablespace_oid = tbspoid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_pg_type_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid typoid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_pg_type_oid = typoid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_array_pg_type_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid typoid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_array_pg_type_oid = typoid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_multirange_pg_type_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid typoid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_mrng_pg_type_oid = typoid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_multirange_array_pg_type_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid typoid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_mrng_array_pg_type_oid = typoid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_heap_pg_class_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid reloid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_heap_pg_class_oid = reloid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_heap_relfilenode(PG_FUNCTION_ARGS)
|
|
{
|
|
RelFileNumber relfilenumber = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_heap_pg_class_relfilenumber = relfilenumber;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_index_pg_class_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid reloid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_index_pg_class_oid = reloid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_index_relfilenode(PG_FUNCTION_ARGS)
|
|
{
|
|
RelFileNumber relfilenumber = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_index_pg_class_relfilenumber = relfilenumber;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_toast_pg_class_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid reloid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_toast_pg_class_oid = reloid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_toast_relfilenode(PG_FUNCTION_ARGS)
|
|
{
|
|
RelFileNumber relfilenumber = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_toast_pg_class_relfilenumber = relfilenumber;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_pg_enum_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid enumoid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_pg_enum_oid = enumoid;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_next_pg_authid_oid(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid authoid = PG_GETARG_OID(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_next_pg_authid_oid = authoid;
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_create_empty_extension(PG_FUNCTION_ARGS)
|
|
{
|
|
text *extName;
|
|
text *schemaName;
|
|
bool relocatable;
|
|
text *extVersion;
|
|
Datum extConfig;
|
|
Datum extCondition;
|
|
List *requiredExtensions;
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
|
|
/* We must check these things before dereferencing the arguments */
|
|
if (PG_ARGISNULL(0) ||
|
|
PG_ARGISNULL(1) ||
|
|
PG_ARGISNULL(2) ||
|
|
PG_ARGISNULL(3))
|
|
elog(ERROR, "null argument to binary_upgrade_create_empty_extension is not allowed");
|
|
|
|
extName = PG_GETARG_TEXT_PP(0);
|
|
schemaName = PG_GETARG_TEXT_PP(1);
|
|
relocatable = PG_GETARG_BOOL(2);
|
|
extVersion = PG_GETARG_TEXT_PP(3);
|
|
|
|
if (PG_ARGISNULL(4))
|
|
extConfig = PointerGetDatum(NULL);
|
|
else
|
|
extConfig = PG_GETARG_DATUM(4);
|
|
|
|
if (PG_ARGISNULL(5))
|
|
extCondition = PointerGetDatum(NULL);
|
|
else
|
|
extCondition = PG_GETARG_DATUM(5);
|
|
|
|
requiredExtensions = NIL;
|
|
if (!PG_ARGISNULL(6))
|
|
{
|
|
ArrayType *textArray = PG_GETARG_ARRAYTYPE_P(6);
|
|
Datum *textDatums;
|
|
int ndatums;
|
|
int i;
|
|
|
|
deconstruct_array_builtin(textArray, TEXTOID, &textDatums, NULL, &ndatums);
|
|
for (i = 0; i < ndatums; i++)
|
|
{
|
|
char *extName = TextDatumGetCString(textDatums[i]);
|
|
Oid extOid = get_extension_oid(extName, false);
|
|
|
|
requiredExtensions = lappend_oid(requiredExtensions, extOid);
|
|
}
|
|
}
|
|
|
|
InsertExtensionTuple(text_to_cstring(extName),
|
|
GetUserId(),
|
|
get_namespace_oid(text_to_cstring(schemaName), false),
|
|
relocatable,
|
|
text_to_cstring(extVersion),
|
|
extConfig,
|
|
extCondition,
|
|
requiredExtensions);
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_record_init_privs(PG_FUNCTION_ARGS)
|
|
{
|
|
bool record_init_privs = PG_GETARG_BOOL(0);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
binary_upgrade_record_init_privs = record_init_privs;
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
Datum
|
|
binary_upgrade_set_missing_value(PG_FUNCTION_ARGS)
|
|
{
|
|
Oid table_id = PG_GETARG_OID(0);
|
|
text *attname = PG_GETARG_TEXT_P(1);
|
|
text *value = PG_GETARG_TEXT_P(2);
|
|
char *cattname = text_to_cstring(attname);
|
|
char *cvalue = text_to_cstring(value);
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
SetAttrMissing(table_id, cattname, cvalue);
|
|
|
|
PG_RETURN_VOID();
|
|
}
|
|
|
|
/*
|
|
* Verify the given slot has already consumed all the WAL changes.
|
|
*
|
|
* Returns true if there are no decodable WAL records after the
|
|
* confirmed_flush_lsn. Otherwise false.
|
|
*
|
|
* This is a special purpose function to ensure that the given slot can be
|
|
* upgraded without data loss.
|
|
*/
|
|
Datum
|
|
binary_upgrade_logical_slot_has_caught_up(PG_FUNCTION_ARGS)
|
|
{
|
|
Name slot_name;
|
|
XLogRecPtr end_of_wal;
|
|
bool found_pending_wal;
|
|
|
|
CHECK_IS_BINARY_UPGRADE;
|
|
|
|
/* We must check before dereferencing the argument */
|
|
if (PG_ARGISNULL(0))
|
|
elog(ERROR, "null argument to binary_upgrade_validate_wal_records is not allowed");
|
|
|
|
CheckSlotPermissions();
|
|
|
|
slot_name = PG_GETARG_NAME(0);
|
|
|
|
/* Acquire the given slot */
|
|
ReplicationSlotAcquire(NameStr(*slot_name), true);
|
|
|
|
Assert(SlotIsLogical(MyReplicationSlot));
|
|
|
|
/* Slots must be valid as otherwise we won't be able to scan the WAL */
|
|
Assert(MyReplicationSlot->data.invalidated == RS_INVAL_NONE);
|
|
|
|
end_of_wal = GetFlushRecPtr(NULL);
|
|
found_pending_wal = LogicalReplicationSlotHasPendingWal(end_of_wal);
|
|
|
|
/* Clean up */
|
|
ReplicationSlotRelease();
|
|
|
|
PG_RETURN_BOOL(!found_pending_wal);
|
|
}
|