From 5d677ca7654f083280b2634d941e09258fa99c78 Mon Sep 17 00:00:00 2001 From: Jeff Davis Date: Tue, 2 Jan 2024 13:42:48 -0800 Subject: [PATCH v9 2/3] CREATE SUSBCRIPTION ... SERVER. Allow specifying a foreign server for CREATE SUBSCRIPTION, rather than a raw connection string with CONNECTION. Using a foreign server as a layer of indirection improves management of multiple subscriptions to the same server. It also provides integration with user mappings in case different subscriptions have different owners or a subscription changes owners. Discussion: https://postgr.es/m/61831790a0a937038f78ce09f8dd4cef7de7456a.camel@j-davis.com Reviewed-by: Ashutosh Bapat --- contrib/postgres_fdw/Makefile | 2 + .../postgres_fdw/expected/postgres_fdw.out | 8 + contrib/postgres_fdw/meson.build | 5 + contrib/postgres_fdw/sql/postgres_fdw.sql | 7 + contrib/postgres_fdw/t/010_subscription.pl | 71 ++++++++ doc/src/sgml/ref/alter_subscription.sgml | 18 +- doc/src/sgml/ref/create_subscription.sgml | 11 +- src/backend/catalog/pg_subscription.c | 39 +++- src/backend/commands/subscriptioncmds.c | 168 ++++++++++++++++-- src/backend/foreign/foreign.c | 25 +++ src/backend/parser/gram.y | 20 +++ src/backend/replication/logical/worker.c | 16 +- src/bin/pg_dump/pg_dump.c | 27 ++- src/bin/pg_dump/pg_dump.h | 1 + src/bin/psql/tab-complete.c | 2 +- src/include/catalog/pg_subscription.h | 7 +- src/include/foreign/foreign.h | 1 + src/include/nodes/parsenodes.h | 3 + src/test/regress/expected/foreign_data.out | 14 ++ src/test/regress/expected/subscription.out | 53 ++++++ src/test/regress/sql/foreign_data.sql | 18 ++ src/test/regress/sql/subscription.sql | 58 ++++++ src/test/subscription/t/001_rep_changes.pl | 60 +++++++ 23 files changed, 601 insertions(+), 33 deletions(-) create mode 100644 contrib/postgres_fdw/t/010_subscription.pl diff --git a/contrib/postgres_fdw/Makefile b/contrib/postgres_fdw/Makefile index c1b0cad453..c3498ea6b4 100644 --- a/contrib/postgres_fdw/Makefile +++ b/contrib/postgres_fdw/Makefile @@ -18,6 +18,8 @@ DATA = postgres_fdw--1.0.sql postgres_fdw--1.0--1.1.sql REGRESS = postgres_fdw +TAP_TESTS = 1 + ifdef USE_PGXS PG_CONFIG = pg_config PGXS := $(shell $(PG_CONFIG) --pgxs) diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 8a7a15cc51..ecd0230738 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -270,6 +270,14 @@ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1; -- should work again ANALYZE ft1; ALTER FOREIGN TABLE ft2 OPTIONS (use_remote_estimate 'true'); -- =================================================================== +-- test subscription +-- =================================================================== +CREATE SUBSCRIPTION regress_pgfdw_subscription SERVER testserver1 + PUBLICATION pub1 WITH (slot_name = NONE, connect = false); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. +DROP SUBSCRIPTION regress_pgfdw_subscription; +-- =================================================================== -- test error case for create publication on foreign table -- =================================================================== CREATE PUBLICATION testpub_ftbl FOR TABLE ft1; -- should fail diff --git a/contrib/postgres_fdw/meson.build b/contrib/postgres_fdw/meson.build index 2b86d8a6ee..cf7071dbf8 100644 --- a/contrib/postgres_fdw/meson.build +++ b/contrib/postgres_fdw/meson.build @@ -39,4 +39,9 @@ tests += { ], 'regress_args': ['--dlpath', meson.build_root() / 'src/test/regress'], }, + 'tap': { + 'tests': [ + 't/010_subscription.pl', + ], + }, } diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 0d8478120d..1c9c12703f 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -254,6 +254,13 @@ SELECT c3, c4 FROM ft1 ORDER BY c3, c1 LIMIT 1; -- should work again ANALYZE ft1; ALTER FOREIGN TABLE ft2 OPTIONS (use_remote_estimate 'true'); +-- =================================================================== +-- test subscription +-- =================================================================== +CREATE SUBSCRIPTION regress_pgfdw_subscription SERVER testserver1 + PUBLICATION pub1 WITH (slot_name = NONE, connect = false); +DROP SUBSCRIPTION regress_pgfdw_subscription; + -- =================================================================== -- test error case for create publication on foreign table -- =================================================================== diff --git a/contrib/postgres_fdw/t/010_subscription.pl b/contrib/postgres_fdw/t/010_subscription.pl new file mode 100644 index 0000000000..a39e8fdbba --- /dev/null +++ b/contrib/postgres_fdw/t/010_subscription.pl @@ -0,0 +1,71 @@ + +# Copyright (c) 2021-2024, PostgreSQL Global Development Group + +# Basic logical replication test +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Initialize publisher node +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +# Create subscriber node +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init; +$node_subscriber->start; + +# Create some preexisting content on publisher +$node_publisher->safe_psql('postgres', + "CREATE TABLE tab_ins AS SELECT a, a + 1 as b FROM generate_series(1,1002) AS a"); + +# Replicate the changes without columns +$node_publisher->safe_psql('postgres', "CREATE TABLE tab_no_col()"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_no_col default VALUES"); + +# Setup structure on subscriber +$node_subscriber->safe_psql('postgres', "CREATE EXTENSION postgres_fdw"); +$node_subscriber->safe_psql('postgres', "CREATE TABLE tab_ins (a int, b int)"); + +# Setup logical replication +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_pub FOR TABLE tab_ins"); + +my $publisher_host = $node_publisher->host; +my $publisher_port = $node_publisher->port; +$node_subscriber->safe_psql('postgres', + "CREATE SERVER tap_server FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host '$publisher_host', port '$publisher_port', dbname 'postgres')" +); + +$node_subscriber->safe_psql('postgres', + "CREATE USER MAPPING FOR PUBLIC SERVER tap_server" +); + +$node_subscriber->safe_psql('postgres', + "CREATE FOREIGN TABLE f_tab_ins (a int, b int) SERVER tap_server OPTIONS(table_name 'tab_ins')" +); +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub SERVER tap_server PUBLICATION tap_pub WITH (password_required=false)" +); + +# Wait for initial table sync to finish +$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub'); + +my $result = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM (SELECT f.b = l.b as match FROM tab_ins l, f_tab_ins f WHERE l.a = f.a) WHERE match"); +is($result, qq(1002), 'check initial data was copied to subscriber'); + +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_ins SELECT a, a + 1 FROM generate_series(1003,1050) a"); + +$node_publisher->wait_for_catchup('tap_sub'); + +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM (SELECT f.b = l.b as match FROM tab_ins l, f_tab_ins f WHERE l.a = f.a) WHERE match"); +is($result, qq(1050), 'check initial data was copied to subscriber'); + +done_testing(); diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 6d36ff0dc9..6d219145a9 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -21,6 +21,7 @@ PostgreSQL documentation +ALTER SUBSCRIPTION name SERVER servername ALTER SUBSCRIPTION name CONNECTION 'conninfo' ALTER SUBSCRIPTION name SET PUBLICATION publication_name [, ...] [ WITH ( publication_option [= value] [, ... ] ) ] ALTER SUBSCRIPTION name ADD PUBLICATION publication_name [, ...] [ WITH ( publication_option [= value] [, ... ] ) ] @@ -94,13 +95,24 @@ ALTER SUBSCRIPTION name RENAME TO < + + SERVER servername + + + This clause replaces the foreign server or connection string originally + set by with the foreign server + servername. + + + + CONNECTION 'conninfo' - This clause replaces the connection string originally set by - . See there for more - information. + This clause replaces the foreign server or connection string originally + set by with the connection + string conninfo. diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index c7ace922f9..24538baf98 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -22,7 +22,7 @@ PostgreSQL documentation CREATE SUBSCRIPTION subscription_name - CONNECTION 'conninfo' + { SERVER servername | CONNECTION 'conninfo' } PUBLICATION publication_name [, ...] [ WITH ( subscription_parameter [= value] [, ... ] ) ] @@ -77,6 +77,15 @@ CREATE SUBSCRIPTION subscription_name + + SERVER servername + + + A foreign server to use for the connection. + + + + CONNECTION 'conninfo' diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index c516c25ac7..5a2eaa803d 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -20,12 +20,15 @@ #include "access/tableam.h" #include "access/xact.h" #include "catalog/indexing.h" +#include "catalog/pg_foreign_server.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" #include "catalog/pg_type.h" +#include "foreign/foreign.h" #include "miscadmin.h" #include "nodes/makefuncs.h" #include "storage/lmgr.h" +#include "utils/acl.h" #include "utils/array.h" #include "utils/builtins.h" #include "utils/fmgroids.h" @@ -40,7 +43,7 @@ static List *textarray_to_stringlist(ArrayType *textarray); * Fetch the subscription from the syscache. */ Subscription * -GetSubscription(Oid subid, bool missing_ok) +GetSubscription(Oid subid, bool missing_ok, bool aclcheck) { HeapTuple tup; Subscription *sub; @@ -75,10 +78,36 @@ GetSubscription(Oid subid, bool missing_ok) sub->runasowner = subform->subrunasowner; /* Get conninfo */ - datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, - tup, - Anum_pg_subscription_subconninfo); - sub->conninfo = TextDatumGetCString(datum); + if (OidIsValid(subform->subserver)) + { + AclResult aclresult; + + /* recheck ACL if requested */ + if (aclcheck) + { + aclresult = object_aclcheck(ForeignServerRelationId, + subform->subserver, + subform->subowner, ACL_USAGE); + + if (aclresult != ACLCHECK_OK) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("subscription owner \"%s\" does not have permission on foreign server \"%s\"", + GetUserNameFromId(subform->subowner, false), + ForeignServerName(subform->subserver)))); + } + + sub->conninfo = ForeignServerConnectionString(subform->subowner, + subform->subserver, + true); + } + else + { + datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, + tup, + Anum_pg_subscription_subconninfo); + sub->conninfo = TextDatumGetCString(datum); + } /* Get slotname */ datum = SysCacheGetAttr(SUBSCRIPTIONOID, diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 75e6cd8ae3..983b5d17fe 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -25,14 +25,17 @@ #include "catalog/objectaddress.h" #include "catalog/pg_authid_d.h" #include "catalog/pg_database_d.h" +#include "catalog/pg_foreign_server.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" #include "catalog/pg_type.h" +#include "catalog/pg_user_mapping.h" #include "commands/dbcommands.h" #include "commands/defrem.h" #include "commands/event_trigger.h" #include "commands/subscriptioncmds.h" #include "executor/executor.h" +#include "foreign/foreign.h" #include "miscadmin.h" #include "nodes/makefuncs.h" #include "pgstat.h" @@ -574,6 +577,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, Datum values[Natts_pg_subscription]; Oid owner = GetUserId(); HeapTuple tup; + Oid serverid; char *conninfo; char originname[NAMEDATALEN]; List *publications; @@ -666,15 +670,40 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, if (opts.synchronous_commit == NULL) opts.synchronous_commit = "off"; - conninfo = stmt->conninfo; - publications = stmt->publication; - /* Load the library providing us libpq calls. */ load_file("libpqwalreceiver", false); + if (stmt->servername) + { + ForeignServer *server; + + Assert(!stmt->conninfo); + conninfo = NULL; + + server = GetForeignServerByName(stmt->servername, false); + aclresult = object_aclcheck(ForeignServerRelationId, server->serverid, owner, ACL_USAGE); + if (aclresult != ACLCHECK_OK) + aclcheck_error(aclresult, OBJECT_FOREIGN_SERVER, server->servername); + + /* make sure a user mapping exists */ + GetUserMapping(owner, server->serverid); + + serverid = server->serverid; + conninfo = ForeignServerConnectionString(owner, serverid, true); + } + else + { + Assert(stmt->conninfo); + + serverid = InvalidOid; + conninfo = stmt->conninfo; + } + /* Check the connection info string. */ walrcv_check_conninfo(conninfo, opts.passwordrequired && !superuser()); + publications = stmt->publication; + /* Everything ok, form a new tuple. */ memset(values, 0, sizeof(values)); memset(nulls, false, sizeof(nulls)); @@ -697,8 +726,12 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr); values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired); values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner); - values[Anum_pg_subscription_subconninfo - 1] = - CStringGetTextDatum(conninfo); + values[Anum_pg_subscription_subserver - 1] = serverid; + if (!OidIsValid(serverid)) + values[Anum_pg_subscription_subconninfo - 1] = + CStringGetTextDatum(conninfo); + else + nulls[Anum_pg_subscription_subconninfo - 1] = true; if (opts.slot_name) values[Anum_pg_subscription_subslotname - 1] = DirectFunctionCall1(namein, CStringGetDatum(opts.slot_name)); @@ -719,6 +752,17 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, recordDependencyOnOwner(SubscriptionRelationId, subid, owner); + ObjectAddressSet(myself, SubscriptionRelationId, subid); + + if (stmt->servername) + { + ObjectAddress referenced; + Assert(OidIsValid(serverid)); + + ObjectAddressSet(referenced, ForeignServerRelationId, serverid); + recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL); + } + ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname)); replorigin_create(originname); @@ -835,8 +879,6 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, if (opts.enabled) ApplyLauncherWakeupAtCommit(); - ObjectAddressSet(myself, SubscriptionRelationId, subid); - InvokeObjectPostCreateHook(SubscriptionRelationId, subid, 0); return myself; @@ -1104,7 +1146,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION, stmt->subname); - sub = GetSubscription(subid, false); + /* + * Skip ACL checks on the subscription's foreign server, if any. If + * changing the server (or replacing it with a raw connection), then the + * old one will be removed anyway. If changing something unrelated, + * there's no need to do an additional ACL check here; that will be done + * by the subscription worker anyway. + */ + sub = GetSubscription(subid, false, false); /* * Don't allow non-superuser modification of a subscription with @@ -1124,6 +1173,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, memset(nulls, false, sizeof(nulls)); memset(replaces, false, sizeof(replaces)); + ObjectAddressSet(myself, SubscriptionRelationId, subid); + switch (stmt->kind) { case ALTER_SUBSCRIPTION_OPTIONS: @@ -1244,7 +1295,80 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, break; } + case ALTER_SUBSCRIPTION_SERVER: + { + ForeignServer *new_server; + ObjectAddress referenced; + AclResult aclresult; + char *conninfo; + + /* + * Remove what was there before, either another foreign server + * or a connection string. + */ + if (form->subserver) + { + deleteDependencyRecordsForSpecific(SubscriptionRelationId, form->oid, + DEPENDENCY_NORMAL, + ForeignServerRelationId, form->subserver); + } + else + { + nulls[Anum_pg_subscription_subconninfo - 1] = true; + replaces[Anum_pg_subscription_subconninfo - 1] = true; + } + + /* + * Find the new server and user mapping. Check ACL of server + * based on current user ID, but find the user mapping based + * on the subscription owner. + */ + new_server = GetForeignServerByName(stmt->servername, false); + aclresult = object_aclcheck(ForeignServerRelationId, + new_server->serverid, + form->subowner, ACL_USAGE); + if (aclresult != ACLCHECK_OK) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("subscription owner \"%s\" does not have permission on foreign server \"%s\"", + GetUserNameFromId(form->subowner, false), + ForeignServerName(new_server->serverid)))); + + /* make sure a user mapping exists */ + GetUserMapping(form->subowner, new_server->serverid); + + conninfo = ForeignServerConnectionString(form->subowner, + new_server->serverid, + true); + + /* Load the library providing us libpq calls. */ + load_file("libpqwalreceiver", false); + /* Check the connection info string. */ + walrcv_check_conninfo(conninfo, + sub->passwordrequired && !sub->ownersuperuser); + + values[Anum_pg_subscription_subserver - 1] = new_server->serverid; + replaces[Anum_pg_subscription_subserver - 1] = true; + + ObjectAddressSet(referenced, ForeignServerRelationId, new_server->serverid); + recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL); + + update_tuple = true; + } + break; + case ALTER_SUBSCRIPTION_CONNECTION: + /* remove reference to foreign server and dependencies, if present */ + if (form->subserver) + { + deleteDependencyRecordsForSpecific(SubscriptionRelationId, form->oid, + DEPENDENCY_NORMAL, + ForeignServerRelationId, form->subserver); + + values[Anum_pg_subscription_subserver - 1] = InvalidOid; + replaces[Anum_pg_subscription_subserver - 1] = true; + } + /* Load the library providing us libpq calls. */ load_file("libpqwalreceiver", false); /* Check the connection info string. */ @@ -1455,8 +1579,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, table_close(rel, RowExclusiveLock); - ObjectAddressSet(myself, SubscriptionRelationId, subid); - InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0); /* Wake up related replication workers to handle this change quickly. */ @@ -1541,9 +1663,28 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) subname = pstrdup(NameStr(*DatumGetName(datum))); /* Get conninfo */ - datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, tup, - Anum_pg_subscription_subconninfo); - conninfo = TextDatumGetCString(datum); + if (OidIsValid(form->subserver)) + { + AclResult aclresult; + + aclresult = object_aclcheck(ForeignServerRelationId, form->subserver, + form->subowner, ACL_USAGE); + if (aclresult != ACLCHECK_OK) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("subscription owner \"%s\" does not have permission on foreign server \"%s\"", + GetUserNameFromId(form->subowner, false), + ForeignServerName(form->subserver)))); + + conninfo = ForeignServerConnectionString(form->subowner, + form->subserver, true); + } + else + { + datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, tup, + Anum_pg_subscription_subconninfo); + conninfo = TextDatumGetCString(datum); + } /* Get slotname */ datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, @@ -1644,6 +1785,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) } /* Clean up dependencies */ + deleteDependencyRecordsFor(SubscriptionRelationId, subid, false); deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0); /* Remove any associated relation synchronization states. */ diff --git a/src/backend/foreign/foreign.c b/src/backend/foreign/foreign.c index b4635d6eba..db2cf6780d 100644 --- a/src/backend/foreign/foreign.c +++ b/src/backend/foreign/foreign.c @@ -179,6 +179,31 @@ GetForeignServerExtended(Oid serverid, bits16 flags) } +/* + * ForeignServerName - get name of foreign server. + */ +char * +ForeignServerName(Oid serverid) +{ + Form_pg_foreign_server serverform; + char *servername; + HeapTuple tp; + + tp = SearchSysCache1(FOREIGNSERVEROID, ObjectIdGetDatum(serverid)); + + if (!HeapTupleIsValid(tp)) + elog(ERROR, "cache lookup failed for foreign server %u", serverid); + + serverform = (Form_pg_foreign_server) GETSTRUCT(tp); + + servername = pstrdup(NameStr(serverform->srvname)); + + ReleaseSysCache(tp); + + return servername; +} + + /* * GetForeignServerByName - look up the foreign server definition by name. */ diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 3460fea56b..c27e0b8b5d 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -10654,6 +10654,16 @@ CreateSubscriptionStmt: n->options = $8; $$ = (Node *) n; } + | CREATE SUBSCRIPTION name SERVER name PUBLICATION name_list opt_definition + { + CreateSubscriptionStmt *n = + makeNode(CreateSubscriptionStmt); + n->subname = $3; + n->servername = $5; + n->publication = $7; + n->options = $8; + $$ = (Node *) n; + } ; /***************************************************************************** @@ -10683,6 +10693,16 @@ AlterSubscriptionStmt: n->conninfo = $5; $$ = (Node *) n; } + | ALTER SUBSCRIPTION name SERVER name + { + AlterSubscriptionStmt *n = + makeNode(AlterSubscriptionStmt); + + n->kind = ALTER_SUBSCRIPTION_SERVER; + n->subname = $3; + n->servername = $5; + $$ = (Node *) n; + } | ALTER SUBSCRIPTION name REFRESH PUBLICATION opt_definition { AlterSubscriptionStmt *n = diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 9b598caf3c..0ade3150bf 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -3897,7 +3897,7 @@ maybe_reread_subscription(void) /* Ensure allocations in permanent context. */ oldctx = MemoryContextSwitchTo(ApplyContext); - newsub = GetSubscription(MyLogicalRepWorker->subid, true); + newsub = GetSubscription(MyLogicalRepWorker->subid, true, true); /* * Exit if the subscription was removed. This normally should not happen @@ -4003,7 +4003,9 @@ maybe_reread_subscription(void) } /* - * Callback from subscription syscache invalidation. + * Callback from subscription syscache invalidation. Also needed for server or + * user mapping invalidation, which can change the connection information for + * subscriptions that connect using a server object. */ static void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue) @@ -4602,7 +4604,7 @@ InitializeLogRepWorker(void) StartTransactionCommand(); oldctx = MemoryContextSwitchTo(ApplyContext); - MySubscription = GetSubscription(MyLogicalRepWorker->subid, true); + MySubscription = GetSubscription(MyLogicalRepWorker->subid, true, true); if (!MySubscription) { ereport(LOG, @@ -4639,6 +4641,14 @@ InitializeLogRepWorker(void) CacheRegisterSyscacheCallback(SUBSCRIPTIONOID, subscription_change_cb, (Datum) 0); + /* Keep us informed about subscription changes. */ + CacheRegisterSyscacheCallback(FOREIGNSERVEROID, + subscription_change_cb, + (Datum) 0); + /* Keep us informed about subscription changes. */ + CacheRegisterSyscacheCallback(USERMAPPINGOID, + subscription_change_cb, + (Datum) 0); CacheRegisterSyscacheCallback(AUTHOID, subscription_change_cb, diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index bc20a025ce..5312008a82 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -4634,6 +4634,7 @@ getSubscriptions(Archive *fout) int i_subdisableonerr; int i_subpasswordrequired; int i_subrunasowner; + int i_subservername; int i_subconninfo; int i_subslotname; int i_subsynccommit; @@ -4705,10 +4706,12 @@ getSubscriptions(Archive *fout) LOGICALREP_ORIGIN_ANY); if (dopt->binary_upgrade && fout->remoteVersion >= 170000) - appendPQExpBufferStr(query, " o.remote_lsn AS suboriginremotelsn,\n" + appendPQExpBufferStr(query, " fs.srvname AS subservername,\n" + " o.remote_lsn AS suboriginremotelsn,\n" " s.subenabled\n"); else - appendPQExpBufferStr(query, " NULL AS suboriginremotelsn,\n" + appendPQExpBufferStr(query, " NULL AS subservername,\n" + " NULL AS suboriginremotelsn,\n" " false AS subenabled\n"); appendPQExpBufferStr(query, @@ -4716,6 +4719,8 @@ getSubscriptions(Archive *fout) if (dopt->binary_upgrade && fout->remoteVersion >= 170000) appendPQExpBufferStr(query, + "LEFT JOIN pg_catalog.pg_foreign_server fs \n" + " ON fs.oid = s.subserver \n" "LEFT JOIN pg_catalog.pg_replication_origin_status o \n" " ON o.external_id = 'pg_' || s.oid::text \n"); @@ -4741,6 +4746,7 @@ getSubscriptions(Archive *fout) i_subdisableonerr = PQfnumber(res, "subdisableonerr"); i_subpasswordrequired = PQfnumber(res, "subpasswordrequired"); i_subrunasowner = PQfnumber(res, "subrunasowner"); + i_subservername = PQfnumber(res, "subservername"); i_subconninfo = PQfnumber(res, "subconninfo"); i_subslotname = PQfnumber(res, "subslotname"); i_subsynccommit = PQfnumber(res, "subsynccommit"); @@ -4760,7 +4766,10 @@ getSubscriptions(Archive *fout) AssignDumpId(&subinfo[i].dobj); subinfo[i].dobj.name = pg_strdup(PQgetvalue(res, i, i_subname)); subinfo[i].rolname = getRoleName(PQgetvalue(res, i, i_subowner)); - + if (PQgetisnull(res, i, i_subservername)) + subinfo[i].subservername = NULL; + else + subinfo[i].subservername = pg_strdup(PQgetvalue(res, i, i_subservername)); subinfo[i].subbinary = pg_strdup(PQgetvalue(res, i, i_subbinary)); subinfo[i].substream = @@ -4986,9 +4995,17 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo) appendPQExpBuffer(delq, "DROP SUBSCRIPTION %s;\n", qsubname); - appendPQExpBuffer(query, "CREATE SUBSCRIPTION %s CONNECTION ", + appendPQExpBuffer(query, "CREATE SUBSCRIPTION %s ", qsubname); - appendStringLiteralAH(query, subinfo->subconninfo, fout); + if (subinfo->subservername) + { + appendPQExpBuffer(query, "SERVER %s", fmtId(subinfo->subservername)); + } + else + { + appendPQExpBuffer(query, "CONNECTION "); + appendStringLiteralAH(query, subinfo->subconninfo, fout); + } /* Build list of quoted publications and append them to query. */ if (!parsePGArray(subinfo->subpublications, &pubnames, &npubnames)) diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index f0772d2157..849950e470 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -659,6 +659,7 @@ typedef struct _SubscriptionInfo char *subdisableonerr; char *subpasswordrequired; char *subrunasowner; + char *subservername; char *subconninfo; char *subslotname; char *subsynccommit; diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index ada711d02f..616c90c48b 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -3327,7 +3327,7 @@ psql_completion(const char *text, int start, int end) /* CREATE SUBSCRIPTION */ else if (Matches("CREATE", "SUBSCRIPTION", MatchAny)) - COMPLETE_WITH("CONNECTION"); + COMPLETE_WITH("SERVER", "CONNECTION"); else if (Matches("CREATE", "SUBSCRIPTION", MatchAny, "CONNECTION", MatchAny)) COMPLETE_WITH("PUBLICATION"); else if (Matches("CREATE", "SUBSCRIPTION", MatchAny, "CONNECTION", diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index ab206bad7d..01141febb5 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -93,9 +93,11 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW bool subrunasowner; /* True if replication should execute as the * subscription owner */ + Oid subserver; /* Set if connecting with server */ + #ifdef CATALOG_VARLEN /* variable-length fields start here */ /* Connection string to the publisher */ - text subconninfo BKI_FORCE_NOT_NULL; + text subconninfo; /* Set if connecting with connection string */ /* Slot name on publisher */ NameData subslotname BKI_FORCE_NULL; @@ -165,7 +167,8 @@ typedef struct Subscription */ #define LOGICALREP_STREAM_PARALLEL 'p' -extern Subscription *GetSubscription(Oid subid, bool missing_ok); +extern Subscription *GetSubscription(Oid subid, bool missing_ok, + bool aclcheck); extern void FreeSubscription(Subscription *sub); extern void DisableSubscription(Oid subid); diff --git a/src/include/foreign/foreign.h b/src/include/foreign/foreign.h index b5b9b97f4d..a2f04ce9af 100644 --- a/src/include/foreign/foreign.h +++ b/src/include/foreign/foreign.h @@ -65,6 +65,7 @@ typedef struct ForeignTable extern ForeignServer *GetForeignServer(Oid serverid); +extern char *ForeignServerName(Oid serverid); extern ForeignServer *GetForeignServerExtended(Oid serverid, bits16 flags); extern ForeignServer *GetForeignServerByName(const char *srvname, diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index b3181f34ae..6d6b242cec 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -4041,6 +4041,7 @@ typedef struct CreateSubscriptionStmt { NodeTag type; char *subname; /* Name of the subscription */ + char *servername; /* Server name of publisher */ char *conninfo; /* Connection string to publisher */ List *publication; /* One or more publication to subscribe to */ List *options; /* List of DefElem nodes */ @@ -4049,6 +4050,7 @@ typedef struct CreateSubscriptionStmt typedef enum AlterSubscriptionType { ALTER_SUBSCRIPTION_OPTIONS, + ALTER_SUBSCRIPTION_SERVER, ALTER_SUBSCRIPTION_CONNECTION, ALTER_SUBSCRIPTION_SET_PUBLICATION, ALTER_SUBSCRIPTION_ADD_PUBLICATION, @@ -4063,6 +4065,7 @@ typedef struct AlterSubscriptionStmt NodeTag type; AlterSubscriptionType kind; /* ALTER_SUBSCRIPTION_OPTIONS, etc */ char *subname; /* Name of the subscription */ + char *servername; /* Server name of publisher */ char *conninfo; /* Connection string to publisher */ List *publication; /* One or more publication to subscribe to */ List *options; /* List of DefElem nodes */ diff --git a/src/test/regress/expected/foreign_data.out b/src/test/regress/expected/foreign_data.out index 0211531f32..30aa23a8ff 100644 --- a/src/test/regress/expected/foreign_data.out +++ b/src/test/regress/expected/foreign_data.out @@ -733,6 +733,20 @@ SELECT pg_conninfo_from_server('connection_server', 'regress_test_role2', false) DROP USER MAPPING FOR regress_test_role SERVER connection_server; DROP USER MAPPING FOR PUBLIC SERVER connection_server; DROP SERVER connection_server; +SET ROLE regress_test_role; +CREATE SERVER t3 FOREIGN DATA WRAPPER regress_connection_fdw; -- ERROR: no permissions on FDW +ERROR: permission denied for foreign-data wrapper regress_connection_fdw +RESET ROLE; +GRANT USAGE ON FOREIGN DATA WRAPPER regress_connection_fdw TO regress_test_role; +SET ROLE regress_test_role; +CREATE SERVER t3 FOREIGN DATA WRAPPER regress_connection_fdw; +IMPORT FOREIGN SCHEMA foo FROM SERVER t3 INTO bar; -- fails +ERROR: schema "bar" does not exist +CREATE USER MAPPING FOR PUBLIC SERVER t3 OPTIONS (user 'x', password 'secret'); +DROP USER MAPPING FOR PUBLIC SERVER t3; +DROP SERVER t3; +RESET ROLE; +REVOKE USAGE ON FOREIGN DATA WRAPPER regress_connection_fdw FROM regress_test_role; DROP FOREIGN DATA WRAPPER regress_connection_fdw; -- CREATE FOREIGN TABLE CREATE SCHEMA foreign_schema; diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index b15eddbff3..b0a1a3cc26 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -144,6 +144,59 @@ ERROR: could not connect to the publisher: invalid port number: "-1" ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar'; ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string +RESET SESSION AUTHORIZATION; +GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; +CREATE SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist password=regress_fakepassword' PUBLICATION testpub WITH (slot_name = NONE, connect = false); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. +DROP SUBSCRIPTION regress_testsub6; +-- test using a server object instead of connection string +RESET SESSION AUTHORIZATION; +CREATE FOREIGN DATA WRAPPER regress_connection_fdw + VALIDATOR pg_connection_validator; +CREATE SERVER regress_testserver1 FOREIGN DATA WRAPPER regress_connection_fdw; +CREATE SERVER regress_testserver2 FOREIGN DATA WRAPPER regress_connection_fdw; +CREATE USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver1 + OPTIONS (password 'secret'); +CREATE USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver2 + OPTIONS (password 'secret'); +GRANT USAGE ON FOREIGN SERVER regress_testserver2 TO regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; +CREATE SUBSCRIPTION regress_testsub6 SERVER regress_testserver1 PUBLICATION testpub + WITH (slot_name = NONE, connect = false); -- fails +ERROR: permission denied for foreign server regress_testserver1 +CREATE SUBSCRIPTION regress_testsub6 SERVER regress_testserver2 PUBLICATION testpub + WITH (slot_name = NONE, connect = false); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription. +RESET SESSION AUTHORIZATION; +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver1; -- fails +ERROR: subscription owner "regress_subscription_user3" does not have permission on foreign server "regress_testserver1" +GRANT USAGE ON FOREIGN SERVER regress_testserver1 TO regress_subscription_user3; +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver1; +DROP USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver2; +DROP SERVER regress_testserver2; +-- test an FDW with no validator +CREATE FOREIGN DATA WRAPPER regress_fdw; +CREATE SERVER regress_testserver3 FOREIGN DATA WRAPPER regress_fdw + OPTIONS (abc 'xyz'); +CREATE USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver3 + OPTIONS (password 'secret'); +GRANT USAGE ON FOREIGN SERVER regress_testserver3 TO regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver3; +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver1; +RESET SESSION AUTHORIZATION; +DROP USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver3; +DROP SERVER regress_testserver3; +DROP FOREIGN DATA WRAPPER regress_fdw; +DROP SUBSCRIPTION regress_testsub6; +DROP USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver1; +DROP SERVER regress_testserver1; +DROP FOREIGN DATA WRAPPER regress_connection_fdw; +REVOKE CREATE ON DATABASE regression FROM regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user; \dRs+ List of subscriptions Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN diff --git a/src/test/regress/sql/foreign_data.sql b/src/test/regress/sql/foreign_data.sql index a8e2edfeee..7956705217 100644 --- a/src/test/regress/sql/foreign_data.sql +++ b/src/test/regress/sql/foreign_data.sql @@ -329,6 +329,24 @@ SELECT pg_conninfo_from_server('connection_server', 'regress_test_role2', false) DROP USER MAPPING FOR regress_test_role SERVER connection_server; DROP USER MAPPING FOR PUBLIC SERVER connection_server; DROP SERVER connection_server; + +SET ROLE regress_test_role; +CREATE SERVER t3 FOREIGN DATA WRAPPER regress_connection_fdw; -- ERROR: no permissions on FDW +RESET ROLE; +GRANT USAGE ON FOREIGN DATA WRAPPER regress_connection_fdw TO regress_test_role; +SET ROLE regress_test_role; + +CREATE SERVER t3 FOREIGN DATA WRAPPER regress_connection_fdw; + +IMPORT FOREIGN SCHEMA foo FROM SERVER t3 INTO bar; -- fails + +CREATE USER MAPPING FOR PUBLIC SERVER t3 OPTIONS (user 'x', password 'secret'); +DROP USER MAPPING FOR PUBLIC SERVER t3; +DROP SERVER t3; + +RESET ROLE; +REVOKE USAGE ON FOREIGN DATA WRAPPER regress_connection_fdw FROM regress_test_role; + DROP FOREIGN DATA WRAPPER regress_connection_fdw; -- CREATE FOREIGN TABLE diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 444e563ff3..4d44f141b7 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -88,6 +88,64 @@ CREATE SUBSCRIPTION regress_testsub5 CONNECTION 'port=-1' PUBLICATION testpub; -- fail - invalid connection string during ALTER ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar'; +RESET SESSION AUTHORIZATION; +GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; + +CREATE SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist password=regress_fakepassword' PUBLICATION testpub WITH (slot_name = NONE, connect = false); +DROP SUBSCRIPTION regress_testsub6; + +-- test using a server object instead of connection string + +RESET SESSION AUTHORIZATION; +CREATE FOREIGN DATA WRAPPER regress_connection_fdw + VALIDATOR pg_connection_validator; +CREATE SERVER regress_testserver1 FOREIGN DATA WRAPPER regress_connection_fdw; +CREATE SERVER regress_testserver2 FOREIGN DATA WRAPPER regress_connection_fdw; +CREATE USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver1 + OPTIONS (password 'secret'); +CREATE USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver2 + OPTIONS (password 'secret'); +GRANT USAGE ON FOREIGN SERVER regress_testserver2 TO regress_subscription_user3; + +SET SESSION AUTHORIZATION regress_subscription_user3; +CREATE SUBSCRIPTION regress_testsub6 SERVER regress_testserver1 PUBLICATION testpub + WITH (slot_name = NONE, connect = false); -- fails +CREATE SUBSCRIPTION regress_testsub6 SERVER regress_testserver2 PUBLICATION testpub + WITH (slot_name = NONE, connect = false); +RESET SESSION AUTHORIZATION; + +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver1; -- fails +GRANT USAGE ON FOREIGN SERVER regress_testserver1 TO regress_subscription_user3; +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver1; +DROP USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver2; +DROP SERVER regress_testserver2; + +-- test an FDW with no validator +CREATE FOREIGN DATA WRAPPER regress_fdw; +CREATE SERVER regress_testserver3 FOREIGN DATA WRAPPER regress_fdw + OPTIONS (abc 'xyz'); +CREATE USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver3 + OPTIONS (password 'secret'); +GRANT USAGE ON FOREIGN SERVER regress_testserver3 TO regress_subscription_user3; + +SET SESSION AUTHORIZATION regress_subscription_user3; +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver3; +ALTER SUBSCRIPTION regress_testsub6 SERVER regress_testserver1; + +RESET SESSION AUTHORIZATION; +DROP USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver3; +DROP SERVER regress_testserver3; +DROP FOREIGN DATA WRAPPER regress_fdw; + +DROP SUBSCRIPTION regress_testsub6; + +DROP USER MAPPING FOR regress_subscription_user3 SERVER regress_testserver1; +DROP SERVER regress_testserver1; +DROP FOREIGN DATA WRAPPER regress_connection_fdw; +REVOKE CREATE ON DATABASE regression FROM regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user; + \dRs+ ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false); diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl index 9ccebd890a..8653423d08 100644 --- a/src/test/subscription/t/001_rep_changes.pl +++ b/src/test/subscription/t/001_rep_changes.pl @@ -27,6 +27,8 @@ $node_publisher->safe_psql('postgres', "CREATE TABLE tab_notrep AS SELECT generate_series(1,10) AS a"); $node_publisher->safe_psql('postgres', "CREATE TABLE tab_ins AS SELECT generate_series(1,1002) AS a"); +$node_publisher->safe_psql('postgres', + "CREATE TABLE tab_ins2 AS SELECT generate_series(1,1002) AS a"); $node_publisher->safe_psql('postgres', "CREATE TABLE tab_full AS SELECT generate_series(1,10) AS a"); $node_publisher->safe_psql('postgres', "CREATE TABLE tab_full2 (x text)"); @@ -65,6 +67,7 @@ $node_publisher->safe_psql('postgres', # Setup structure on subscriber $node_subscriber->safe_psql('postgres', "CREATE TABLE tab_notrep (a int)"); $node_subscriber->safe_psql('postgres', "CREATE TABLE tab_ins (a int)"); +$node_subscriber->safe_psql('postgres', "CREATE TABLE tab_ins2 (a int)"); $node_subscriber->safe_psql('postgres', "CREATE TABLE tab_full (a int)"); $node_subscriber->safe_psql('postgres', "CREATE TABLE tab_full2 (x text)"); $node_subscriber->safe_psql('postgres', @@ -110,6 +113,25 @@ $node_subscriber->safe_psql('postgres', "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub, tap_pub_ins_only" ); +my $publisher_host = $node_publisher->host; +my $publisher_port = $node_publisher->port; +$node_subscriber->safe_psql('postgres', + "CREATE FOREIGN DATA WRAPPER test_connection_fdw VALIDATOR pg_connection_validator" +); +$node_subscriber->safe_psql('postgres', + "CREATE SERVER tap_sub2_server FOREIGN DATA WRAPPER test_connection_fdw OPTIONS (host '$publisher_host', port '$publisher_port', dbname 'postgres')" +); + +$node_subscriber->safe_psql('postgres', + "CREATE USER MAPPING FOR PUBLIC SERVER tap_sub2_server" +); + +$node_publisher->safe_psql('postgres', "CREATE PUBLICATION tap_simple_pub FOR TABLE tab_ins2"); + +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub2 SERVER tap_sub2_server PUBLICATION tap_simple_pub WITH (password_required=false)" +); + # Wait for initial table sync to finish $node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub'); @@ -121,11 +143,22 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_ins"); is($result, qq(1002), 'check initial data was copied to subscriber'); +$result = + $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_ins2"); +is($result, qq(1002), 'check initial data was copied to subscriber'); + +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub2 CONNECTION '$publisher_connstr'"); + $node_publisher->safe_psql('postgres', "INSERT INTO tab_ins SELECT generate_series(1,50)"); $node_publisher->safe_psql('postgres', "DELETE FROM tab_ins WHERE a > 20"); $node_publisher->safe_psql('postgres', "UPDATE tab_ins SET a = -a"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_ins2 SELECT generate_series(1,50)"); + +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION tap_sub2 SERVER tap_sub2_server"); + $node_publisher->safe_psql('postgres', "INSERT INTO tab_rep SELECT generate_series(1,50)"); $node_publisher->safe_psql('postgres', "DELETE FROM tab_rep WHERE a > 20"); @@ -158,6 +191,10 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab_ins"); is($result, qq(1052|1|1002), 'check replicated inserts on subscriber'); +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab_ins2"); +is($result, qq(1052|1|1002), 'check replicated inserts on subscriber'); + $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab_rep"); is($result, qq(20|-20|-1), 'check replicated changes on subscriber'); @@ -449,10 +486,27 @@ $node_publisher->poll_query_until('postgres', or die "Timed out while waiting for apply to restart after changing PUBLICATION"; +# test that changes to a foreign server subscription cause the worker +# to restart +$oldpid = $node_publisher->safe_psql('postgres', + "SELECT pid FROM pg_stat_replication WHERE application_name = 'tap_sub2' AND state = 'streaming';" +); +$node_subscriber->safe_psql('postgres', + "ALTER SERVER tap_sub2_server OPTIONS (sslmode 'disable')" +); +$node_publisher->poll_query_until('postgres', + "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = 'tap_sub2' AND state = 'streaming';" + ) + or die + "Timed out while waiting for apply to restart after changing PUBLICATION"; + $node_publisher->safe_psql('postgres', "INSERT INTO tab_ins SELECT generate_series(1001,1100)"); $node_publisher->safe_psql('postgres', "DELETE FROM tab_rep"); +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_ins2 SELECT generate_series(1001,1100)"); + # Restart the publisher and check the state of the subscriber which # should be in a streaming state after catching up. $node_publisher->stop('fast'); @@ -465,6 +519,11 @@ $result = $node_subscriber->safe_psql('postgres', is($result, qq(1152|1|1100), 'check replicated inserts after subscription publication change'); +$result = $node_subscriber->safe_psql('postgres', + "SELECT count(*), min(a), max(a) FROM tab_ins2"); +is($result, qq(1152|1|1100), + 'check replicated inserts after subscription publication change'); + $result = $node_subscriber->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab_rep"); is($result, qq(20|-20|-1), @@ -533,6 +592,7 @@ $node_publisher->poll_query_until('postgres', # check all the cleanup $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_renamed"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub2"); $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_subscription"); -- 2.34.1