From 76bf161f9b8267e09e4dba130441f02e915ec857 Mon Sep 17 00:00:00 2001 From: Hayato Kuroda Date: Tue, 25 Aug 2026 17:38:32 +0900 Subject: [PATCH v15 3/3] Save and restore roident and remote_lsn for new cluster --- src/bin/pg_upgrade/check.c | 51 +++++++ src/bin/pg_upgrade/pg_upgrade.c | 20 +++ src/bin/pg_upgrade/pg_upgrade.h | 2 + .../pg_upgrade/t/009_transfer_commit_ts.pl | 128 ++++++++++++++++++ 4 files changed, 201 insertions(+) diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c index 17ab7cf54af..73746486197 100644 --- a/src/bin/pg_upgrade/check.c +++ b/src/bin/pg_upgrade/check.c @@ -2470,6 +2470,57 @@ check_old_cluster_subscription_state(void) PQclear(res); PQfinish(conn); + if (user_opts.do_copy_pg_commit_ts) + { + /* + * Save pg_replication_origin.roident and + * pg_replication_origin_status.remote_lsn old cluster. + */ + conn = connectToServer(&old_cluster, old_cluster.dbarr.dbs[0].db_name); + res = executeQueryOrDie(conn, + "SELECT string_agg(CASE WHEN os.remote_lsn is not null THEN format('" + " SELECT pg_catalog.pg_replication_origin_advance(" + " (SELECT roname FROM pg_catalog.pg_replication_origin " + " WHERE roident=%%s), %%L)', o.roident, os.remote_lsn) END, ';')," + " format('UPDATE pg_catalog.pg_replication_origin r SET roident=roident::int+10000'), " + " string_agg(CASE WHEN s.subname is not null THEN " + " format('UPDATE pg_catalog.pg_replication_origin r " + " SET roident=%%s from pg_catalog.pg_subscription s WHERE r.roname=''pg_''||s.oid " + " and s.subname=%%L'" + " ,o.roident, s.subname) " + " ELSE" + " format('UPDATE pg_catalog.pg_replication_origin r " + " SET roident=%%s WHERE r.roname=%%L'" + " ,o.roident, o.roname) " + " END" + " ,';' ORDER BY o.roident DESC), " + " max(o.roident)>9999 " + "FROM pg_catalog.pg_replication_origin o " + "LEFT JOIN pg_catalog.pg_subscription s " + " ON o.roname = 'pg_' || s.oid " + "LEFT JOIN pg_catalog.pg_replication_origin_status os " + " ON os.external_id = o.roname;"); + + ntup = PQntuples(res); + if (ntup > 0) + { + if (strcmp(PQgetvalue(res, 0, 3), "t") == 0) + pg_fatal("The origin ID exceeds 9999"); + sql_roident_correction = createPQExpBuffer(); + /* Prepare roident in new cluster for execute update */ + appendPQExpBufferStr(sql_roident_correction, PQgetvalue(res, 0, 1)); + appendPQExpBufferStr(sql_roident_correction, ";\n"); + /* Restore roident */ + appendPQExpBufferStr(sql_roident_correction, PQgetvalue(res, 0, 2)); + appendPQExpBufferStr(sql_roident_correction, ";\n"); + /* Restore remote_lsn if exists */ + appendPQExpBufferStr(sql_roident_correction, PQgetvalue(res, 0, 0)); + appendPQExpBufferStr(sql_roident_correction, ";\n"); + PQclear(res); + PQfinish(conn); + } + } + /* * We don't allow upgrade if there is a risk of dangling slot or origin * corresponding to initial sync after upgrade. diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c index e27ce5a16aa..0a882c0e0a5 100644 --- a/src/bin/pg_upgrade/pg_upgrade.c +++ b/src/bin/pg_upgrade/pg_upgrade.c @@ -85,6 +85,9 @@ char *output_files[] = { NULL }; +/* list SQL commands for new cluster */ +PQExpBuffer sql_roident_correction; + int main(int argc, char **argv) { @@ -253,6 +256,23 @@ main(int argc, char **argv) issue_warnings_and_set_wal_level(); + if (user_opts.do_copy_pg_commit_ts && sql_roident_correction) + { + /* + * Correction pg_replication_origin.roident and + * pg_replication_origin_status.remote_lsn for new cluster. + */ + PGconn *conn_new_template1; + + start_postmaster(&new_cluster, true); + + conn_new_template1 = connectToServer(&new_cluster, "template1"); + PQclear(executeQueryOrDie(conn_new_template1, "%s", sql_roident_correction->data)); + PQfinish(conn_new_template1); + + stop_postmaster(false); + } + pg_log(PG_REPORT, "\n" "Upgrade Complete\n" diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h index 89042063daa..d5eddacac3f 100644 --- a/src/bin/pg_upgrade/pg_upgrade.h +++ b/src/bin/pg_upgrade/pg_upgrade.h @@ -12,6 +12,7 @@ #include "common/relpath.h" #include "libpq-fe.h" +#include "fe_utils/string_utils.h" /* For now, pg_upgrade does not use common/logging.c; use our own pg_fatal */ #undef pg_fatal @@ -361,6 +362,7 @@ extern UserOpts user_opts; extern ClusterInfo old_cluster, new_cluster; extern OSInfo os_info; +extern PQExpBuffer sql_roident_correction; /* check.c */ diff --git a/src/bin/pg_upgrade/t/009_transfer_commit_ts.pl b/src/bin/pg_upgrade/t/009_transfer_commit_ts.pl index 7479be57fd6..d74896b2270 100644 --- a/src/bin/pg_upgrade/t/009_transfer_commit_ts.pl +++ b/src/bin/pg_upgrade/t/009_transfer_commit_ts.pl @@ -65,4 +65,132 @@ my $resnew = $new->safe_psql( $new->stop; ok($resold eq $resnew, "timestamp transferred successfully"); +my $publisher = PostgreSQL::Test::Cluster->new('publisher'); +$publisher->init(allows_streaming => 'logical'); +$publisher->start; + +# Initialize the old subscriber node +my $old_sub = PostgreSQL::Test::Cluster->new('old_sub'); +$old_sub->init; +$old_sub->append_conf('postgresql.conf', 'track_commit_timestamp = on'); +$old_sub->start; +my $oldbindir = $old_sub->config_data('--bindir'); + +# Initialize the new subscriber +my $new_sub = PostgreSQL::Test::Cluster->new('new_sub'); +$new_sub->init; +$new_sub->append_conf('postgresql.conf', 'track_commit_timestamp = on'); +my $newbindir = $new_sub->config_data('--bindir'); + +# In a VPATH build, we'll be started in the source directory, but we want +# to run pg_upgrade in the build directory so that any files generated finish +# in it, like delete_old_cluster.{sh,bat}. +chdir ${PostgreSQL::Test::Utils::tmp_check}; + +# Remember a connection string for the publisher node. It would be used +# several times. +my $appname='tap_sub'; +my $connstr = $publisher->connstr . ' dbname=postgres '; + +$publisher->safe_psql('postgres', "CREATE TABLE tab (a int PRIMARY KEY)"); +$old_sub->safe_psql('postgres', "CREATE TABLE tab (a int PRIMARY KEY)"); +$publisher->safe_psql('postgres', "CREATE PUBLICATION regress_pub1 FOR TABLE tab"); +#Create 1 origin +$old_sub->safe_psql( + 'postgres', " + CREATE SUBSCRIPTION a_dummy + CONNECTION '$connstr' + PUBLICATION regress_pub1 + WITH (connect = false, enabled = false,create_slot = false)"); +#Create 2 origin +$old_sub->safe_psql('postgres', + "CREATE SUBSCRIPTION regress_sub2 CONNECTION '$connstr application_name=$appname' PUBLICATION regress_pub1 WITH(copy_data = false)" +); +#Create 3 origin +$old_sub->safe_psql( + 'postgres', " + CREATE SUBSCRIPTION z_dummy + CONNECTION '$connstr' + PUBLICATION regress_pub1 + WITH (connect = false, enabled = false,create_slot = false)"); +#Create 4,5 origin no link subscription +$old_sub->safe_psql('postgres', + "SELECT pg_replication_origin_create('no_link_sub_4'),pg_replication_origin_create('no_link_sub_5')" +); + +# Wait for initial table sync to finish +$old_sub->wait_for_subscription_sync($publisher, $appname); +$publisher->safe_psql('postgres', "INSERT INTO tab VALUES (11);"); +$publisher->wait_for_catchup($appname); +my $result = $old_sub->safe_psql('postgres', + "SELECT count(1) = 1 FROM tab"); +is($result, qq(t), "Check that the table is 1 row"); + +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_sub2'" +); + +#Delete 1 origin +$old_sub->safe_psql('postgres', "ALTER SUBSCRIPTION a_dummy DISABLE"); +$old_sub->safe_psql('postgres', "ALTER SUBSCRIPTION a_dummy SET (slot_name = NONE)"); +$old_sub->safe_psql('postgres', "DROP SUBSCRIPTION a_dummy"); + +my $origin_others= $old_sub->safe_psql('postgres', + "SELECT roident,roname FROM pg_replication_origin o LEFT JOIN pg_subscription s ON o.roname = 'pg_' || s.oid WHERE s.subname is null ORDER BY o.roident" +); + +$old_sub->stop; + +command_ok( + [ + 'pg_upgrade', + '--no-sync','--pg-commit-ts', + '--old-datadir' => $old_sub->data_dir, + '--new-datadir' => $new_sub->data_dir, + '--old-bindir' => $oldbindir, + '--new-bindir' => $newbindir, + '--socketdir' => $new_sub->host, + '--old-port' => $old_sub->port, + '--new-port' => $new_sub->port, + $mode + ], + 'run of pg_upgrade for old instance when the subscription tables not empty' +); +ok( !-d $new_sub->data_dir . "/pg_upgrade_output.d", + "pg_upgrade_output.d/ removed after successful pg_upgrade"); + +$new_sub->start; + +$result = $new_sub->safe_psql('postgres', + "SELECT roident,s.subname FROM pg_replication_origin o LEFT JOIN pg_subscription s ON o.roname = 'pg_' || s.oid WHERE s.subname is not null ORDER BY o.roident"); +is($result, qq(2|regress_sub2 +3|z_dummy), "Check that the roident this restore old cluster (subscribtions)"); + + +$result = $new_sub->safe_psql('postgres', + "SELECT roident,roname FROM pg_replication_origin o LEFT JOIN pg_subscription s ON o.roname = 'pg_' || s.oid WHERE s.subname is null ORDER BY o.roident"); +# No migrate origin create finction pg_replication_origin_create +# Comment next line if fix this bug +$origin_others=""; +is($result, $origin_others, "Check that the roident this restore old cluster (origin id without subscribtions)"); + +my $remote_lsn_new_sub = $new_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_sub2'" +); +is($remote_lsn_new_sub, qq($remote_lsn), "remote_lsn should have been preserved"); + + +my $log_offset = -s $new_sub->logfile; + +#Check replication new cluster +$publisher->safe_psql('postgres', "UPDATE tab set a=32 where a=11;"); + +$publisher->wait_for_catchup($appname); + +$result = $new_sub->safe_psql('postgres', + "SELECT a FROM tab WHERE a=32"); +is($result,32, "update row ok"); + +$new_sub->log_check("no conflict",$log_offset,log_unlike => [ qr/conflict detected on relation \"public.tab\": conflict=/, ]); + done_testing(); -- 2.52.0