On Tue, Mar 22, 2022 at 5:29 AM Andres Freund <and...@anarazel.de> wrote: > > On 2022-02-13 19:34:05 +0530, vignesh C wrote: > > Thanks for the comments, the attached v14 patch has the changes for the > > same. > > The patch needs a rebase, it currently fails to apply: > http://cfbot.cputube.org/patch_37_2957.log
The attached v15 patch is rebased on top of HEAD. Regards, Vignesh
From fa175c7c823dc9fbcca7676ddec944430da81022 Mon Sep 17 00:00:00 2001 From: Vigneshwaran C <vignes...@gmail.com> Date: Tue, 22 Mar 2022 15:09:13 +0530 Subject: [PATCH v15] Identify missing publications from publisher while create/alter subscription. Creating/altering subscription is successful when we specify a publication which does not exist in the publisher. This patch checks if the specified publications are present in the publisher and throws an error if any of the publication is missing in the publisher. Author: Vignesh C Reviewed-by: Bharath Rupireddy, Japin Li, Dilip Kumar, Euler Taveira, Ashutosh Sharma Discussion: https://www.postgresql.org/message-id/flat/20220321235957.i4jtjn4wyjucex6b%40alap3.anarazel.de#b846aaaafd4ef657cfaa8c9890f044e4 --- doc/src/sgml/ref/alter_subscription.sgml | 13 ++ doc/src/sgml/ref/create_subscription.sgml | 20 ++- src/backend/commands/subscriptioncmds.c | 201 +++++++++++++++++++--- src/bin/psql/tab-complete.c | 15 +- src/test/subscription/t/007_ddl.pl | 54 ++++++ 5 files changed, 270 insertions(+), 33 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index ac2db249cb..995c1f270d 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -172,6 +172,19 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO < </para> </listitem> </varlistentry> + + <varlistentry> + <term><literal>validate_publication</literal> (<type>boolean</type>)</term> + <listitem> + <para> + When true, the command verifies if all the specified publications + that are being subscribed to are present in the publisher and throws + an error if any of the publications doesn't exist. The default is + <literal>false</literal>. + </para> + </listitem> + </varlistentry> + </variablelist></para> </listitem> </varlistentry> diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index b701752fc9..f2e7e8744d 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -110,12 +110,14 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl command should connect to the publisher at all. The default is <literal>true</literal>. Setting this to <literal>false</literal> will force the values of - <literal>create_slot</literal>, <literal>enabled</literal> and - <literal>copy_data</literal> to <literal>false</literal>. + <literal>create_slot</literal>, <literal>enabled</literal>, + <literal>copy_data</literal> and + <literal>validate_publication</literal> to <literal>false</literal>. (You cannot combine setting <literal>connect</literal> to <literal>false</literal> with setting <literal>create_slot</literal>, <literal>enabled</literal>, - or <literal>copy_data</literal> to <literal>true</literal>.) + <literal>copy_data</literal> or + <literal>validate_publication</literal> to <literal>true</literal>.) </para> <para> @@ -170,6 +172,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl </para> </listitem> </varlistentry> + + <varlistentry> + <term><literal>validate_publication</literal> (<type>boolean</type>)</term> + <listitem> + <para> + When true, the command verifies if all the specified publications + that are being subscribed to are present in the publisher and throws + an error if any of the publications doesn't exist. The default is + <literal>false</literal>. + </para> + </listitem> + </varlistentry> </variablelist> </para> diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index e16f04626d..6c066a1dfc 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -64,6 +64,7 @@ #define SUBOPT_TWOPHASE_COMMIT 0x00000200 #define SUBOPT_DISABLE_ON_ERR 0x00000400 #define SUBOPT_LSN 0x00000800 +#define SUBOPT_VALIDATE_PUB 0x00001000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -86,6 +87,7 @@ typedef struct SubOpts bool streaming; bool twophase; bool disableonerr; + bool validate_publication; XLogRecPtr lsn; } SubOpts; @@ -137,6 +139,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->twophase = false; if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR)) opts->disableonerr = false; + if (IsSet(supported_opts, SUBOPT_VALIDATE_PUB)) + opts->validate_publication = false; /* Parse options */ foreach(lc, stmt_options) @@ -292,6 +296,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->specified_opts |= SUBOPT_LSN; opts->lsn = lsn; } + else if (IsSet(supported_opts, SUBOPT_VALIDATE_PUB) && + strcmp(defel->defname, "validate_publication") == 0) + { + if (IsSet(opts->specified_opts, SUBOPT_VALIDATE_PUB)) + errorConflictingDefElem(defel, pstate); + + opts->specified_opts |= SUBOPT_VALIDATE_PUB; + opts->validate_publication = defGetBoolean(defel); + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -327,10 +340,19 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, errmsg("%s and %s are mutually exclusive options", "connect = false", "copy_data = true"))); + if (opts->validate_publication && + IsSet(supported_opts, SUBOPT_VALIDATE_PUB) && + IsSet(opts->specified_opts, SUBOPT_VALIDATE_PUB)) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("%s and %s are mutually exclusive options", + "connect = false", "validate_publication = true"))); + /* Change the defaults of other options. */ opts->enabled = false; opts->create_slot = false; opts->copy_data = false; + opts->validate_publication = false; } /* @@ -374,6 +396,133 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, } } +/* + * Add publication names from the list to a string. + */ +static void +get_publications_str(List *publications, StringInfo dest, bool quote_literal) +{ + ListCell *lc; + bool first = true; + + Assert(list_length(publications) > 0); + + foreach(lc, publications) + { + char *pubname = strVal(lfirst(lc)); + + if (first) + first = false; + else + appendStringInfoString(dest, ", "); + + if (quote_literal) + appendStringInfoString(dest, quote_literal_cstr(pubname)); + else + { + appendStringInfoChar(dest, '"'); + appendStringInfoString(dest, pubname); + appendStringInfoChar(dest, '"'); + } + } +} + +/* + * Check the specified publication(s) is(are) present in the publisher. + */ +static void +check_publications(WalReceiverConn *wrconn, List *publications) +{ + WalRcvExecResult *res; + StringInfo cmd; + TupleTableSlot *slot; + List *publicationsCopy = NIL; + Oid tableRow[1] = {TEXTOID}; + + cmd = makeStringInfo(); + appendStringInfoString(cmd, "SELECT t.pubname FROM\n" + " pg_catalog.pg_publication t WHERE\n" + " t.pubname IN ("); + get_publications_str(publications, cmd, true); + appendStringInfoChar(cmd, ')'); + + res = walrcv_exec(wrconn, cmd->data, 1, tableRow); + pfree(cmd->data); + pfree(cmd); + + if (res->status != WALRCV_OK_TUPLES) + ereport(ERROR, + errmsg_plural("could not receive publication from the publisher: %s", + "could not receive list of publications from the publisher: %s", + list_length(publications), + res->err)); + + publicationsCopy = list_copy(publications); + + /* Process publication(s). */ + slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); + while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) + { + char *pubname; + bool isnull; + + pubname = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + /* Delete the publication present in publisher from the list. */ + publicationsCopy = list_delete(publicationsCopy, makeString(pubname)); + ExecClearTuple(slot); + } + + ExecDropSingleTupleTableSlot(slot); + + walrcv_clear_result(res); + + if (list_length(publicationsCopy)) + { + /* Prepare the list of non-existent publication(s) for error message. */ + StringInfo pubnames = makeStringInfo(); + + get_publications_str(publicationsCopy, pubnames, false); + ereport(ERROR, + errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg_plural("publication %s does not exist in the publisher", + "publications %s do not exist in the publisher", + list_length(publicationsCopy), + pubnames->data)); + } +} + +/* + * Connect to the publisher and see if the given publication(s) is(are) present. + */ +static void +connect_and_check_pubs(Subscription *sub, List *publications) +{ + char *err; + WalReceiverConn *wrconn; + + /* Load the library providing us libpq calls. */ + load_file("libpqwalreceiver", false); + + /* Try to connect to the publisher. */ + wrconn = walrcv_connect(sub->conninfo, true, sub->name, &err); + if (!wrconn) + ereport(ERROR, + errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("could not connect to the publisher: %s", err)); + + PG_TRY(); + { + check_publications(wrconn, publications); + } + PG_FINALLY(); + { + walrcv_disconnect(wrconn); + } + PG_END_TRY(); +} + /* * Auxiliary function to build a text array out of a list of String nodes. */ @@ -434,7 +583,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA | SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY | SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT | - SUBOPT_DISABLE_ON_ERR); + SUBOPT_DISABLE_ON_ERR | SUBOPT_VALIDATE_PUB); + parse_subscription_options(pstate, stmt->options, supported_opts, &opts); /* @@ -554,6 +704,9 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, PG_TRY(); { + if (opts.validate_publication) + check_publications(wrconn, publications); + /* * Set sync state based on if we were asked to do data copy or * not. @@ -646,7 +799,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, } static void -AlterSubscription_refresh(Subscription *sub, bool copy_data) +AlterSubscription_refresh(Subscription *sub, bool copy_data, + bool validate_publication) { char *err; List *pubrel_names; @@ -677,6 +831,9 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data) PG_TRY(); { + if (validate_publication) + check_publications(wrconn, sub->publications); + /* Get the table list from publisher. */ pubrel_names = fetch_table_list(wrconn, sub->publications); @@ -1007,7 +1164,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, case ALTER_SUBSCRIPTION_SET_PUBLICATION: { - supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH; + supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH | SUBOPT_VALIDATE_PUB; parse_subscription_options(pstate, stmt->options, supported_opts, &opts); @@ -1016,6 +1173,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, replaces[Anum_pg_subscription_subpublications - 1] = true; update_tuple = true; + if (opts.validate_publication) + connect_and_check_pubs(sub, stmt->publication); /* Refresh if user asked us to. */ if (opts.refresh) @@ -1042,7 +1201,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, /* Make sure refresh sees the new list of publications. */ sub->publications = stmt->publication; - AlterSubscription_refresh(sub, opts.copy_data); + AlterSubscription_refresh(sub, opts.copy_data, false); } break; @@ -1055,6 +1214,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, bool isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION; supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA; + if (isadd) + supported_opts |= SUBOPT_VALIDATE_PUB; + parse_subscription_options(pstate, stmt->options, supported_opts, &opts); @@ -1064,6 +1226,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, replaces[Anum_pg_subscription_subpublications - 1] = true; update_tuple = true; + if (isadd && opts.validate_publication) + connect_and_check_pubs(sub, stmt->publication); /* Refresh if user asked us to. */ if (opts.refresh) @@ -1090,7 +1254,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, /* Refresh the new list of publications. */ sub->publications = publist; - AlterSubscription_refresh(sub, opts.copy_data); + AlterSubscription_refresh(sub, opts.copy_data, false); } break; @@ -1104,7 +1268,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions"))); parse_subscription_options(pstate, stmt->options, - SUBOPT_COPY_DATA, &opts); + SUBOPT_COPY_DATA | SUBOPT_VALIDATE_PUB, + &opts); /* * The subscription option "two_phase" requires that @@ -1132,7 +1297,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH"); - AlterSubscription_refresh(sub, opts.copy_data); + AlterSubscription_refresh(sub, opts.copy_data, + opts.validate_publication); break; } @@ -1653,28 +1819,13 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) StringInfoData cmd; TupleTableSlot *slot; Oid tableRow[2] = {TEXTOID, TEXTOID}; - ListCell *lc; - bool first; List *tablelist = NIL; - Assert(list_length(publications) > 0); - initStringInfo(&cmd); appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n" - " FROM pg_catalog.pg_publication_tables t\n" - " WHERE t.pubname IN ("); - first = true; - foreach(lc, publications) - { - char *pubname = strVal(lfirst(lc)); - - if (first) - first = false; - else - appendStringInfoString(&cmd, ", "); - - appendStringInfoString(&cmd, quote_literal_cstr(pubname)); - } + " FROM pg_catalog.pg_publication_tables t\n" + " WHERE t.pubname IN ("); + get_publications_str(publications, &cmd, true); appendStringInfoChar(&cmd, ')'); res = walrcv_exec(wrconn, cmd.data, 2, tableRow); diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index 5c064595a9..fa304b68ea 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1861,7 +1861,7 @@ psql_completion(const char *text, int start, int end) /* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION WITH ( */ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("REFRESH", "PUBLICATION", "WITH", "(")) - COMPLETE_WITH("copy_data"); + COMPLETE_WITH("copy_data", "validate_publication"); /* ALTER SUBSCRIPTION <name> SET */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, "SET")) COMPLETE_WITH("(", "PUBLICATION"); @@ -1880,11 +1880,15 @@ psql_completion(const char *text, int start, int end) else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("ADD|DROP|SET", "PUBLICATION", MatchAny)) COMPLETE_WITH("WITH ("); - /* ALTER SUBSCRIPTION <name> ADD|DROP|SET PUBLICATION <name> WITH ( */ + /* ALTER SUBSCRIPTION <name> ADD|SET PUBLICATION <name> WITH ( */ else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && - TailMatches("ADD|DROP|SET", "PUBLICATION", MatchAny, "WITH", "(")) - COMPLETE_WITH("copy_data", "refresh"); + TailMatches("ADD|SET", "PUBLICATION", MatchAny, "WITH", "(")) + COMPLETE_WITH("copy_data", "refresh", "validate_publication"); + /* ALTER SUBSCRIPTION <name> DROP PUBLICATION <name> WITH ( */ + else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && + TailMatches("DROP", "PUBLICATION", MatchAny, "WITH", "(")) + COMPLETE_WITH("copy_data", "refresh"); /* ALTER SCHEMA <name> */ else if (Matches("ALTER", "SCHEMA", MatchAny)) COMPLETE_WITH("OWNER TO", "RENAME TO"); @@ -3145,7 +3149,8 @@ psql_completion(const char *text, int start, int end) else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "(")) COMPLETE_WITH("binary", "connect", "copy_data", "create_slot", "enabled", "slot_name", "streaming", - "synchronous_commit", "two_phase", "disable_on_error"); + "synchronous_commit", "two_phase", "disable_on_error", + "validate_publication"); /* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */ diff --git a/src/test/subscription/t/007_ddl.pl b/src/test/subscription/t/007_ddl.pl index 1144b005f6..2936b12fda 100644 --- a/src/test/subscription/t/007_ddl.pl +++ b/src/test/subscription/t/007_ddl.pl @@ -41,6 +41,60 @@ COMMIT; pass "subscription disable and drop in same transaction did not hang"; +# One of the specified publications exists. +my ($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "CREATE SUBSCRIPTION mysub1 CONNECTION '$publisher_connstr' PUBLICATION mypub, non_existent_pub WITH (VALIDATE_PUBLICATION = TRUE)" +); +ok( $stderr =~ + m/ERROR: publication "non_existent_pub" does not exist in the publisher/, + "Create subscription fails with single non-existent publication"); + +# Specifying multiple non-existent publications. +($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "CREATE SUBSCRIPTION mysub1 CONNECTION '$publisher_connstr' PUBLICATION non_existent_pub, non_existent_pub1 WITH (VALIDATE_PUBLICATION = TRUE)" +); +ok( $stderr =~ + m/ERROR: publications "non_existent_pub", "non_existent_pub1" do not exist in the publisher/, + "Create subscription fails with multiple non-existent publications"); + +# Specifying mutually exclusive options. +($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "CREATE SUBSCRIPTION mysub1 CONNECTION '$publisher_connstr' PUBLICATION non_existent_pub, non_existent_pub1 WITH (CONNECT = FALSE, VALIDATE_PUBLICATION = TRUE)" +); +ok( $stderr =~ + m/ERROR: connect = false and validate_publication = true are mutually exclusive options/, + "Create subscription fails with mutually exclusive options"); + +# Specifying non-existent publication along with add publication. +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION mysub1 CONNECTION '$publisher_connstr' PUBLICATION mypub;" +); +($ret, $stdout, $stderr) = ($ret, $stdout, $stderr) = $node_subscriber->psql( + 'postgres', + "ALTER SUBSCRIPTION mysub1 ADD PUBLICATION non_existent_pub WITH (REFRESH = FALSE, VALIDATE_PUBLICATION = TRUE)" +); +ok( $stderr =~ + m/ERROR: publication "non_existent_pub" does not exist in the publisher/, + "Alter subscription add publication fails with non-existent publication"); + +# Specifying non-existent publication along with set publication. +($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION mysub1 SET PUBLICATION non_existent_pub WITH (VALIDATE_PUBLICATION = TRUE)" +); +ok( $stderr =~ + m/ERROR: publication "non_existent_pub" does not exist in the publisher/, + "Alter subscription set publication fails with non-existent publication"); + +# Specifying non-existent database. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION mysub1 CONNECTION 'dbname=regress_doesnotexist2'"); +($ret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION mysub1 SET PUBLICATION non_existent_pub WITH (REFRESH = FALSE, VALIDATE_PUBLICATION = TRUE)" +); +ok( $stderr =~ m/ERROR: could not connect to the publisher/, + "Alter subscription set publication fails with connection to a non-existent database" +); + $node_subscriber->stop; $node_publisher->stop; -- 2.32.0
