diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/backend/access/heap/vacuumlazy.c | 3 | ||||
| -rw-r--r-- | src/backend/utils/activity/pgstat_relation.c | 11 | ||||
| -rw-r--r-- | src/bin/pg_basebackup/pg_createsubscriber.c | 95 | ||||
| -rw-r--r-- | src/bin/pg_basebackup/t/040_pg_createsubscriber.pl | 54 | ||||
| -rw-r--r-- | src/include/pgstat.h | 4 |
5 files changed, 132 insertions, 35 deletions
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c index 62035b7f9c3..30778a15639 100644 --- a/src/backend/access/heap/vacuumlazy.c +++ b/src/backend/access/heap/vacuumlazy.c @@ -961,8 +961,7 @@ heap_vacuum_rel(Relation rel, const VacuumParams params, * soon in cases where the failsafe prevented significant amounts of heap * vacuuming. */ - pgstat_report_vacuum(RelationGetRelid(rel), - rel->rd_rel->relisshared, + pgstat_report_vacuum(rel, Max(vacrel->new_live_tuples, 0), vacrel->recently_dead_tuples + vacrel->missed_dead_tuples, diff --git a/src/backend/utils/activity/pgstat_relation.c b/src/backend/utils/activity/pgstat_relation.c index b90754f8578..55a10c299db 100644 --- a/src/backend/utils/activity/pgstat_relation.c +++ b/src/backend/utils/activity/pgstat_relation.c @@ -207,14 +207,13 @@ pgstat_drop_relation(Relation rel) * Report that the table was just vacuumed and flush IO statistics. */ void -pgstat_report_vacuum(Oid tableoid, bool shared, - PgStat_Counter livetuples, PgStat_Counter deadtuples, - TimestampTz starttime) +pgstat_report_vacuum(Relation rel, PgStat_Counter livetuples, + PgStat_Counter deadtuples, TimestampTz starttime) { PgStat_EntryRef *entry_ref; PgStatShared_Relation *shtabentry; PgStat_StatTabEntry *tabentry; - Oid dboid = (shared ? InvalidOid : MyDatabaseId); + Oid dboid = (rel->rd_rel->relisshared ? InvalidOid : MyDatabaseId); TimestampTz ts; PgStat_Counter elapsedtime; @@ -226,8 +225,8 @@ pgstat_report_vacuum(Oid tableoid, bool shared, elapsedtime = TimestampDifferenceMilliseconds(starttime, ts); /* block acquiring lock for the same reason as pgstat_report_autovac() */ - entry_ref = pgstat_get_entry_ref_locked(PGSTAT_KIND_RELATION, - dboid, tableoid, false); + entry_ref = pgstat_get_entry_ref_locked(PGSTAT_KIND_RELATION, dboid, + RelationGetRelid(rel), false); shtabentry = (PgStatShared_Relation *) entry_ref->shared_stats; tabentry = &shtabentry->stats; diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c index ef6deec14af..41a649297c7 100644 --- a/src/bin/pg_basebackup/pg_createsubscriber.c +++ b/src/bin/pg_basebackup/pg_createsubscriber.c @@ -116,6 +116,7 @@ static void stop_standby_server(const char *datadir); static void wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions *opt); static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo); +static bool find_publication(PGconn *conn, const char *pubname, const char *dbname); static void drop_publication(PGconn *conn, const char *pubname, const char *dbname, bool *made_publication); static void check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo); @@ -764,6 +765,39 @@ generate_object_name(PGconn *conn) } /* + * Does the publication exist in the specified database? + */ +static bool +find_publication(PGconn *conn, const char *pubname, const char *dbname) +{ + PQExpBuffer str = createPQExpBuffer(); + PGresult *res; + bool found = false; + char *pubname_esc = PQescapeLiteral(conn, pubname, strlen(pubname)); + + appendPQExpBuffer(str, + "SELECT 1 FROM pg_catalog.pg_publication " + "WHERE pubname = %s", + pubname_esc); + res = PQexec(conn, str->data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + pg_log_error("could not find publication \"%s\" in database \"%s\": %s", + pubname, dbname, PQerrorMessage(conn)); + disconnect_database(conn, true); + } + + if (PQntuples(res) == 1) + found = true; + + PQclear(res); + PQfreemem(pubname_esc); + destroyPQExpBuffer(str); + + return found; +} + +/* * Create the publications and replication slots in preparation for logical * replication. Returns the LSN from latest replication slot. It will be the * replication start point that is used to adjust the subscriptions (see @@ -799,13 +833,25 @@ setup_publisher(struct LogicalRepInfo *dbinfo) if (num_replslots == 0) dbinfo[i].replslotname = pg_strdup(dbinfo[i].subname); - /* - * Create publication on publisher. This step should be executed - * *before* promoting the subscriber to avoid any transactions between - * consistent LSN and the new publication rows (such transactions - * wouldn't see the new publication rows resulting in an error). - */ - create_publication(conn, &dbinfo[i]); + if (find_publication(conn, dbinfo[i].pubname, dbinfo[i].dbname)) + { + /* Reuse existing publication on publisher. */ + pg_log_info("use existing publication \"%s\" in database \"%s\"", + dbinfo[i].pubname, dbinfo[i].dbname); + /* Don't remove pre-existing publication if an error occurs. */ + dbinfo[i].made_publication = false; + } + else + { + /* + * Create publication on publisher. This step should be executed + * *before* promoting the subscriber to avoid any transactions + * between consistent LSN and the new publication rows (such + * transactions wouldn't see the new publication rows resulting in + * an error). + */ + create_publication(conn, &dbinfo[i]); + } /* Create replication slot on publisher */ if (lsn) @@ -1749,11 +1795,10 @@ drop_publication(PGconn *conn, const char *pubname, const char *dbname, /* * Retrieve and drop the publications. * - * Since the publications were created before the consistent LSN, they - * remain on the subscriber even after the physical replica is - * promoted. Remove these publications from the subscriber because - * they have no use. Additionally, if requested, drop all pre-existing - * publications. + * Publications copied during physical replication remain on the subscriber + * after promotion. If --clean=publications is specified, drop all existing + * publications in the subscriber database. Otherwise, only drop publications + * that were created by pg_createsubscriber during this operation. */ static void check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo) @@ -1785,14 +1830,24 @@ check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo) PQclear(res); } - - /* - * In dry-run mode, we don't create publications, but we still try to drop - * those to provide necessary information to the user. - */ - if (!drop_all_pubs || dry_run) - drop_publication(conn, dbinfo->pubname, dbinfo->dbname, - &dbinfo->made_publication); + else + { + /* Drop publication only if it was created by this tool */ + if (dbinfo->made_publication) + { + drop_publication(conn, dbinfo->pubname, dbinfo->dbname, + &dbinfo->made_publication); + } + else + { + if (dry_run) + pg_log_info("dry-run: would preserve existing publication \"%s\" in database \"%s\"", + dbinfo->pubname, dbinfo->dbname); + else + pg_log_info("preserve existing publication \"%s\" in database \"%s\"", + dbinfo->pubname, dbinfo->dbname); + } + } } /* diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl index 3d6086dc489..9e0db6cd099 100644 --- a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl +++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl @@ -443,10 +443,17 @@ is(scalar(() = $stderr =~ /would create the replication slot/g), is(scalar(() = $stderr =~ /would create subscription/g), 3, "verify subscriptions are created for all databases"); +# Create a user-defined publication, and a table that is not a member of that +# publication. +$node_p->safe_psql($db1, qq( + CREATE PUBLICATION test_pub3 FOR TABLE tbl1; + CREATE TABLE not_replicated (a int); +)); + # Run pg_createsubscriber on node S. --verbose is used twice # to show more information. -# In passing, also test the --enable-two-phase option and -# --clean option +# +# Test two phase and clean options. Use pre-existing publication. command_ok( [ 'pg_createsubscriber', @@ -456,7 +463,7 @@ command_ok( '--publisher-server' => $node_p->connstr($db1), '--socketdir' => $node_s->host, '--subscriber-port' => $node_s->port, - '--publication' => 'pub1', + '--publication' => 'test_pub3', '--publication' => 'pub2', '--replication-slot' => 'replslot1', '--replication-slot' => 'replslot2', @@ -478,13 +485,16 @@ is($result, qq(0), # Insert rows on P $node_p->safe_psql($db1, "INSERT INTO tbl1 VALUES('third row')"); $node_p->safe_psql($db2, "INSERT INTO tbl2 VALUES('row 1')"); +$node_p->safe_psql($db1, "INSERT INTO not_replicated VALUES(0)"); # Start subscriber $node_s->start; # Confirm publications are removed from the subscriber node -is($node_s->safe_psql($db1, "SELECT COUNT(*) FROM pg_publication;"), - '0', 'all publications on subscriber have been removed'); +is($node_s->safe_psql($db1, 'SELECT COUNT(*) FROM pg_publication'), + '0', 'all publications were removed from db1'); +is($node_s->safe_psql($db2, 'SELECT COUNT(*) FROM pg_publication'), + '0', 'all publications were removed from db2'); # Verify that all subtwophase states are pending or enabled, # e.g. there are no subscriptions where subtwophase is disabled ('d') @@ -525,6 +535,9 @@ is( $result, qq(first row second row third row), "logical replication works in database $db1"); +$result = $node_s->safe_psql($db1, 'SELECT * FROM not_replicated'); +is($result, qq(), + "table is not replicated in database $db1"); # Check result in database $db2 $result = $node_s->safe_psql($db2, 'SELECT * FROM tbl2'); @@ -537,6 +550,37 @@ my $sysid_s = $node_s->safe_psql('postgres', 'SELECT system_identifier FROM pg_control_system()'); isnt($sysid_p, $sysid_s, 'system identifier was changed'); +# Verify that pub2 was created in $db2 +is($node_p->safe_psql($db2, "SELECT COUNT(*) FROM pg_publication WHERE pubname = 'pub2'"), + '1', "publication pub2 was created in $db2"); + +# Get subscription and publication names +$result = $node_s->safe_psql( + 'postgres', qq( + SELECT subname, subpublications FROM pg_subscription WHERE subname ~ '^pg_createsubscriber_' + ORDER BY subpublications; +)); +like( + $result, + qr/^pg_createsubscriber_\d+_[0-9a-f]+ \|\{pub2\}\n + pg_createsubscriber_\d+_[0-9a-f]+ \|\{test_pub3\}$/x, + 'subscription and publication names are ok'); + +# Verify that the correct publications are being used +$result = $node_s->safe_psql( + 'postgres', qq( + SELECT d.datname, s.subpublications + FROM pg_subscription s + JOIN pg_database d ON d.oid = s.subdbid + WHERE subname ~ '^pg_createsubscriber_' + ORDER BY s.subdbid + ) +); + +is($result, qq($db1|{test_pub3} +$db2|{pub2}), + "subscriptions use the correct publications"); + # clean up $node_p->teardown_node; $node_s->teardown_node; diff --git a/src/include/pgstat.h b/src/include/pgstat.h index f23dd5870da..6714363144a 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -669,8 +669,8 @@ extern void pgstat_init_relation(Relation rel); extern void pgstat_assoc_relation(Relation rel); extern void pgstat_unlink_relation(Relation rel); -extern void pgstat_report_vacuum(Oid tableoid, bool shared, - PgStat_Counter livetuples, PgStat_Counter deadtuples, +extern void pgstat_report_vacuum(Relation rel, PgStat_Counter livetuples, + PgStat_Counter deadtuples, TimestampTz starttime); extern void pgstat_report_analyze(Relation rel, PgStat_Counter livetuples, PgStat_Counter deadtuples, |