LCOV - code coverage report
Current view: top level - src/backend/commands - subscriptioncmds.c (source / functions) Hit Total Coverage
Test: PostgreSQL 16beta1 Lines: 686 785 87.4 %
Date: 2023-05-30 18:12:27 Functions: 16 18 88.9 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * subscriptioncmds.c
       4             :  *      subscription catalog manipulation functions
       5             :  *
       6             :  * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
       7             :  * Portions Copyright (c) 1994, Regents of the University of California
       8             :  *
       9             :  * IDENTIFICATION
      10             :  *      src/backend/commands/subscriptioncmds.c
      11             :  *
      12             :  *-------------------------------------------------------------------------
      13             :  */
      14             : 
      15             : #include "postgres.h"
      16             : 
      17             : #include "access/htup_details.h"
      18             : #include "access/table.h"
      19             : #include "access/xact.h"
      20             : #include "catalog/catalog.h"
      21             : #include "catalog/dependency.h"
      22             : #include "catalog/indexing.h"
      23             : #include "catalog/namespace.h"
      24             : #include "catalog/objectaccess.h"
      25             : #include "catalog/objectaddress.h"
      26             : #include "catalog/pg_authid_d.h"
      27             : #include "catalog/pg_database_d.h"
      28             : #include "catalog/pg_subscription.h"
      29             : #include "catalog/pg_subscription_rel.h"
      30             : #include "catalog/pg_type.h"
      31             : #include "commands/dbcommands.h"
      32             : #include "commands/defrem.h"
      33             : #include "commands/event_trigger.h"
      34             : #include "commands/subscriptioncmds.h"
      35             : #include "executor/executor.h"
      36             : #include "miscadmin.h"
      37             : #include "nodes/makefuncs.h"
      38             : #include "pgstat.h"
      39             : #include "replication/logicallauncher.h"
      40             : #include "replication/logicalworker.h"
      41             : #include "replication/origin.h"
      42             : #include "replication/slot.h"
      43             : #include "replication/walreceiver.h"
      44             : #include "replication/walsender.h"
      45             : #include "replication/worker_internal.h"
      46             : #include "storage/lmgr.h"
      47             : #include "utils/acl.h"
      48             : #include "utils/builtins.h"
      49             : #include "utils/guc.h"
      50             : #include "utils/lsyscache.h"
      51             : #include "utils/memutils.h"
      52             : #include "utils/pg_lsn.h"
      53             : #include "utils/syscache.h"
      54             : 
      55             : /*
      56             :  * Options that can be specified by the user in CREATE/ALTER SUBSCRIPTION
      57             :  * command.
      58             :  */
      59             : #define SUBOPT_CONNECT              0x00000001
      60             : #define SUBOPT_ENABLED              0x00000002
      61             : #define SUBOPT_CREATE_SLOT          0x00000004
      62             : #define SUBOPT_SLOT_NAME            0x00000008
      63             : #define SUBOPT_COPY_DATA            0x00000010
      64             : #define SUBOPT_SYNCHRONOUS_COMMIT   0x00000020
      65             : #define SUBOPT_REFRESH              0x00000040
      66             : #define SUBOPT_BINARY               0x00000080
      67             : #define SUBOPT_STREAMING            0x00000100
      68             : #define SUBOPT_TWOPHASE_COMMIT      0x00000200
      69             : #define SUBOPT_DISABLE_ON_ERR       0x00000400
      70             : #define SUBOPT_PASSWORD_REQUIRED    0x00000800
      71             : #define SUBOPT_RUN_AS_OWNER         0x00001000
      72             : #define SUBOPT_LSN                  0x00002000
      73             : #define SUBOPT_ORIGIN               0x00004000
      74             : 
      75             : /* check if the 'val' has 'bits' set */
      76             : #define IsSet(val, bits)  (((val) & (bits)) == (bits))
      77             : 
      78             : /*
      79             :  * Structure to hold a bitmap representing the user-provided CREATE/ALTER
      80             :  * SUBSCRIPTION command options and the parsed/default values of each of them.
      81             :  */
      82             : typedef struct SubOpts
      83             : {
      84             :     bits32      specified_opts;
      85             :     char       *slot_name;
      86             :     char       *synchronous_commit;
      87             :     bool        connect;
      88             :     bool        enabled;
      89             :     bool        create_slot;
      90             :     bool        copy_data;
      91             :     bool        refresh;
      92             :     bool        binary;
      93             :     char        streaming;
      94             :     bool        twophase;
      95             :     bool        disableonerr;
      96             :     bool        passwordrequired;
      97             :     bool        runasowner;
      98             :     char       *origin;
      99             :     XLogRecPtr  lsn;
     100             : } SubOpts;
     101             : 
     102             : static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
     103             : static void check_publications_origin(WalReceiverConn *wrconn,
     104             :                                       List *publications, bool copydata,
     105             :                                       char *origin, Oid *subrel_local_oids,
     106             :                                       int subrel_count, char *subname);
     107             : static void check_duplicates_in_publist(List *publist, Datum *datums);
     108             : static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname);
     109             : static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
     110             : 
     111             : 
     112             : /*
     113             :  * Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
     114             :  *
     115             :  * Since not all options can be specified in both commands, this function
     116             :  * will report an error if mutually exclusive options are specified.
     117             :  */
     118             : static void
     119         726 : parse_subscription_options(ParseState *pstate, List *stmt_options,
     120             :                            bits32 supported_opts, SubOpts *opts)
     121             : {
     122             :     ListCell   *lc;
     123             : 
     124             :     /* Start out with cleared opts. */
     125         726 :     memset(opts, 0, sizeof(SubOpts));
     126             : 
     127             :     /* caller must expect some option */
     128             :     Assert(supported_opts != 0);
     129             : 
     130             :     /* If connect option is supported, these others also need to be. */
     131             :     Assert(!IsSet(supported_opts, SUBOPT_CONNECT) ||
     132             :            IsSet(supported_opts, SUBOPT_ENABLED | SUBOPT_CREATE_SLOT |
     133             :                  SUBOPT_COPY_DATA));
     134             : 
     135             :     /* Set default values for the supported options. */
     136         726 :     if (IsSet(supported_opts, SUBOPT_CONNECT))
     137         356 :         opts->connect = true;
     138         726 :     if (IsSet(supported_opts, SUBOPT_ENABLED))
     139         398 :         opts->enabled = true;
     140         726 :     if (IsSet(supported_opts, SUBOPT_CREATE_SLOT))
     141         356 :         opts->create_slot = true;
     142         726 :     if (IsSet(supported_opts, SUBOPT_COPY_DATA))
     143         504 :         opts->copy_data = true;
     144         726 :     if (IsSet(supported_opts, SUBOPT_REFRESH))
     145         102 :         opts->refresh = true;
     146         726 :     if (IsSet(supported_opts, SUBOPT_BINARY))
     147         512 :         opts->binary = false;
     148         726 :     if (IsSet(supported_opts, SUBOPT_STREAMING))
     149         512 :         opts->streaming = LOGICALREP_STREAM_OFF;
     150         726 :     if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
     151         356 :         opts->twophase = false;
     152         726 :     if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
     153         512 :         opts->disableonerr = false;
     154         726 :     if (IsSet(supported_opts, SUBOPT_PASSWORD_REQUIRED))
     155         512 :         opts->passwordrequired = true;
     156         726 :     if (IsSet(supported_opts, SUBOPT_RUN_AS_OWNER))
     157         512 :         opts->runasowner = false;
     158         726 :     if (IsSet(supported_opts, SUBOPT_ORIGIN))
     159         512 :         opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
     160             : 
     161             :     /* Parse options */
     162        1390 :     foreach(lc, stmt_options)
     163             :     {
     164         724 :         DefElem    *defel = (DefElem *) lfirst(lc);
     165             : 
     166         724 :         if (IsSet(supported_opts, SUBOPT_CONNECT) &&
     167         428 :             strcmp(defel->defname, "connect") == 0)
     168             :         {
     169         158 :             if (IsSet(opts->specified_opts, SUBOPT_CONNECT))
     170           0 :                 errorConflictingDefElem(defel, pstate);
     171             : 
     172         158 :             opts->specified_opts |= SUBOPT_CONNECT;
     173         158 :             opts->connect = defGetBoolean(defel);
     174             :         }
     175         566 :         else if (IsSet(supported_opts, SUBOPT_ENABLED) &&
     176         312 :                  strcmp(defel->defname, "enabled") == 0)
     177             :         {
     178          66 :             if (IsSet(opts->specified_opts, SUBOPT_ENABLED))
     179           0 :                 errorConflictingDefElem(defel, pstate);
     180             : 
     181          66 :             opts->specified_opts |= SUBOPT_ENABLED;
     182          66 :             opts->enabled = defGetBoolean(defel);
     183             :         }
     184         500 :         else if (IsSet(supported_opts, SUBOPT_CREATE_SLOT) &&
     185         246 :                  strcmp(defel->defname, "create_slot") == 0)
     186             :         {
     187          32 :             if (IsSet(opts->specified_opts, SUBOPT_CREATE_SLOT))
     188           0 :                 errorConflictingDefElem(defel, pstate);
     189             : 
     190          32 :             opts->specified_opts |= SUBOPT_CREATE_SLOT;
     191          32 :             opts->create_slot = defGetBoolean(defel);
     192             :         }
     193         468 :         else if (IsSet(supported_opts, SUBOPT_SLOT_NAME) &&
     194         372 :                  strcmp(defel->defname, "slot_name") == 0)
     195             :         {
     196         124 :             if (IsSet(opts->specified_opts, SUBOPT_SLOT_NAME))
     197           0 :                 errorConflictingDefElem(defel, pstate);
     198             : 
     199         124 :             opts->specified_opts |= SUBOPT_SLOT_NAME;
     200         124 :             opts->slot_name = defGetString(defel);
     201             : 
     202             :             /* Setting slot_name = NONE is treated as no slot name. */
     203         124 :             if (strcmp(opts->slot_name, "none") == 0)
     204         106 :                 opts->slot_name = NULL;
     205             :             else
     206          18 :                 ReplicationSlotValidateName(opts->slot_name, ERROR);
     207             :         }
     208         344 :         else if (IsSet(supported_opts, SUBOPT_COPY_DATA) &&
     209         224 :                  strcmp(defel->defname, "copy_data") == 0)
     210             :         {
     211          32 :             if (IsSet(opts->specified_opts, SUBOPT_COPY_DATA))
     212           0 :                 errorConflictingDefElem(defel, pstate);
     213             : 
     214          32 :             opts->specified_opts |= SUBOPT_COPY_DATA;
     215          32 :             opts->copy_data = defGetBoolean(defel);
     216             :         }
     217         312 :         else if (IsSet(supported_opts, SUBOPT_SYNCHRONOUS_COMMIT) &&
     218         222 :                  strcmp(defel->defname, "synchronous_commit") == 0)
     219             :         {
     220          12 :             if (IsSet(opts->specified_opts, SUBOPT_SYNCHRONOUS_COMMIT))
     221           0 :                 errorConflictingDefElem(defel, pstate);
     222             : 
     223          12 :             opts->specified_opts |= SUBOPT_SYNCHRONOUS_COMMIT;
     224          12 :             opts->synchronous_commit = defGetString(defel);
     225             : 
     226             :             /* Test if the given value is valid for synchronous_commit GUC. */
     227          12 :             (void) set_config_option("synchronous_commit", opts->synchronous_commit,
     228             :                                      PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
     229             :                                      false, 0, false);
     230             :         }
     231         300 :         else if (IsSet(supported_opts, SUBOPT_REFRESH) &&
     232          66 :                  strcmp(defel->defname, "refresh") == 0)
     233             :         {
     234          66 :             if (IsSet(opts->specified_opts, SUBOPT_REFRESH))
     235           0 :                 errorConflictingDefElem(defel, pstate);
     236             : 
     237          66 :             opts->specified_opts |= SUBOPT_REFRESH;
     238          66 :             opts->refresh = defGetBoolean(defel);
     239             :         }
     240         234 :         else if (IsSet(supported_opts, SUBOPT_BINARY) &&
     241         210 :                  strcmp(defel->defname, "binary") == 0)
     242             :         {
     243          32 :             if (IsSet(opts->specified_opts, SUBOPT_BINARY))
     244           0 :                 errorConflictingDefElem(defel, pstate);
     245             : 
     246          32 :             opts->specified_opts |= SUBOPT_BINARY;
     247          32 :             opts->binary = defGetBoolean(defel);
     248             :         }
     249         202 :         else if (IsSet(supported_opts, SUBOPT_STREAMING) &&
     250         178 :                  strcmp(defel->defname, "streaming") == 0)
     251             :         {
     252          62 :             if (IsSet(opts->specified_opts, SUBOPT_STREAMING))
     253           0 :                 errorConflictingDefElem(defel, pstate);
     254             : 
     255          62 :             opts->specified_opts |= SUBOPT_STREAMING;
     256          62 :             opts->streaming = defGetStreamingMode(defel);
     257             :         }
     258         140 :         else if (strcmp(defel->defname, "two_phase") == 0)
     259             :         {
     260             :             /*
     261             :              * Do not allow toggling of two_phase option. Doing so could cause
     262             :              * missing of transactions and lead to an inconsistent replica.
     263             :              * See comments atop worker.c
     264             :              *
     265             :              * Note: Unsupported twophase indicates that this call originated
     266             :              * from AlterSubscription.
     267             :              */
     268          36 :             if (!IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
     269           6 :                 ereport(ERROR,
     270             :                         (errcode(ERRCODE_SYNTAX_ERROR),
     271             :                          errmsg("unrecognized subscription parameter: \"%s\"", defel->defname)));
     272             : 
     273          30 :             if (IsSet(opts->specified_opts, SUBOPT_TWOPHASE_COMMIT))
     274           0 :                 errorConflictingDefElem(defel, pstate);
     275             : 
     276          30 :             opts->specified_opts |= SUBOPT_TWOPHASE_COMMIT;
     277          30 :             opts->twophase = defGetBoolean(defel);
     278             :         }
     279         104 :         else if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR) &&
     280          80 :                  strcmp(defel->defname, "disable_on_error") == 0)
     281             :         {
     282          20 :             if (IsSet(opts->specified_opts, SUBOPT_DISABLE_ON_ERR))
     283           0 :                 errorConflictingDefElem(defel, pstate);
     284             : 
     285          20 :             opts->specified_opts |= SUBOPT_DISABLE_ON_ERR;
     286          20 :             opts->disableonerr = defGetBoolean(defel);
     287             :         }
     288          84 :         else if (IsSet(supported_opts, SUBOPT_PASSWORD_REQUIRED) &&
     289          60 :                  strcmp(defel->defname, "password_required") == 0)
     290             :         {
     291          22 :             if (IsSet(opts->specified_opts, SUBOPT_PASSWORD_REQUIRED))
     292           0 :                 errorConflictingDefElem(defel, pstate);
     293             : 
     294          22 :             opts->specified_opts |= SUBOPT_PASSWORD_REQUIRED;
     295          22 :             opts->passwordrequired = defGetBoolean(defel);
     296             :         }
     297          62 :         else if (IsSet(supported_opts, SUBOPT_RUN_AS_OWNER) &&
     298          38 :                  strcmp(defel->defname, "run_as_owner") == 0)
     299             :         {
     300           2 :             if (IsSet(opts->specified_opts, SUBOPT_RUN_AS_OWNER))
     301           0 :                 errorConflictingDefElem(defel, pstate);
     302             : 
     303           2 :             opts->specified_opts |= SUBOPT_RUN_AS_OWNER;
     304           2 :             opts->runasowner = defGetBoolean(defel);
     305             :         }
     306          60 :         else if (IsSet(supported_opts, SUBOPT_ORIGIN) &&
     307          36 :                  strcmp(defel->defname, "origin") == 0)
     308             :         {
     309          30 :             if (IsSet(opts->specified_opts, SUBOPT_ORIGIN))
     310           0 :                 errorConflictingDefElem(defel, pstate);
     311             : 
     312          30 :             opts->specified_opts |= SUBOPT_ORIGIN;
     313          30 :             pfree(opts->origin);
     314             : 
     315             :             /*
     316             :              * Even though the "origin" parameter allows only "none" and "any"
     317             :              * values, it is implemented as a string type so that the
     318             :              * parameter can be extended in future versions to support
     319             :              * filtering using origin names specified by the user.
     320             :              */
     321          30 :             opts->origin = defGetString(defel);
     322             : 
     323          44 :             if ((pg_strcasecmp(opts->origin, LOGICALREP_ORIGIN_NONE) != 0) &&
     324          14 :                 (pg_strcasecmp(opts->origin, LOGICALREP_ORIGIN_ANY) != 0))
     325           6 :                 ereport(ERROR,
     326             :                         errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     327             :                         errmsg("unrecognized origin value: \"%s\"", opts->origin));
     328             :         }
     329          30 :         else if (IsSet(supported_opts, SUBOPT_LSN) &&
     330          24 :                  strcmp(defel->defname, "lsn") == 0)
     331          18 :         {
     332          24 :             char       *lsn_str = defGetString(defel);
     333             :             XLogRecPtr  lsn;
     334             : 
     335          24 :             if (IsSet(opts->specified_opts, SUBOPT_LSN))
     336           0 :                 errorConflictingDefElem(defel, pstate);
     337             : 
     338             :             /* Setting lsn = NONE is treated as resetting LSN */
     339          24 :             if (strcmp(lsn_str, "none") == 0)
     340           6 :                 lsn = InvalidXLogRecPtr;
     341             :             else
     342             :             {
     343             :                 /* Parse the argument as LSN */
     344          18 :                 lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in,
     345             :                                                       CStringGetDatum(lsn_str)));
     346             : 
     347          18 :                 if (XLogRecPtrIsInvalid(lsn))
     348           6 :                     ereport(ERROR,
     349             :                             (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
     350             :                              errmsg("invalid WAL location (LSN): %s", lsn_str)));
     351             :             }
     352             : 
     353          18 :             opts->specified_opts |= SUBOPT_LSN;
     354          18 :             opts->lsn = lsn;
     355             :         }
     356             :         else
     357           6 :             ereport(ERROR,
     358             :                     (errcode(ERRCODE_SYNTAX_ERROR),
     359             :                      errmsg("unrecognized subscription parameter: \"%s\"", defel->defname)));
     360             :     }
     361             : 
     362             :     /*
     363             :      * We've been explicitly asked to not connect, that requires some
     364             :      * additional processing.
     365             :      */
     366         666 :     if (!opts->connect && IsSet(supported_opts, SUBOPT_CONNECT))
     367             :     {
     368             :         /* Check for incompatible options from the user. */
     369         128 :         if (opts->enabled &&
     370         128 :             IsSet(opts->specified_opts, SUBOPT_ENABLED))
     371           6 :             ereport(ERROR,
     372             :                     (errcode(ERRCODE_SYNTAX_ERROR),
     373             :             /*- translator: both %s are strings of the form "option = value" */
     374             :                      errmsg("%s and %s are mutually exclusive options",
     375             :                             "connect = false", "enabled = true")));
     376             : 
     377         122 :         if (opts->create_slot &&
     378         116 :             IsSet(opts->specified_opts, SUBOPT_CREATE_SLOT))
     379           6 :             ereport(ERROR,
     380             :                     (errcode(ERRCODE_SYNTAX_ERROR),
     381             :                      errmsg("%s and %s are mutually exclusive options",
     382             :                             "connect = false", "create_slot = true")));
     383             : 
     384         116 :         if (opts->copy_data &&
     385         110 :             IsSet(opts->specified_opts, SUBOPT_COPY_DATA))
     386           6 :             ereport(ERROR,
     387             :                     (errcode(ERRCODE_SYNTAX_ERROR),
     388             :                      errmsg("%s and %s are mutually exclusive options",
     389             :                             "connect = false", "copy_data = true")));
     390             : 
     391             :         /* Change the defaults of other options. */
     392         110 :         opts->enabled = false;
     393         110 :         opts->create_slot = false;
     394         110 :         opts->copy_data = false;
     395             :     }
     396             : 
     397             :     /*
     398             :      * Do additional checking for disallowed combination when slot_name = NONE
     399             :      * was used.
     400             :      */
     401         648 :     if (!opts->slot_name &&
     402         636 :         IsSet(opts->specified_opts, SUBOPT_SLOT_NAME))
     403             :     {
     404         100 :         if (opts->enabled)
     405             :         {
     406          18 :             if (IsSet(opts->specified_opts, SUBOPT_ENABLED))
     407           6 :                 ereport(ERROR,
     408             :                         (errcode(ERRCODE_SYNTAX_ERROR),
     409             :                 /*- translator: both %s are strings of the form "option = value" */
     410             :                          errmsg("%s and %s are mutually exclusive options",
     411             :                                 "slot_name = NONE", "enabled = true")));
     412             :             else
     413          12 :                 ereport(ERROR,
     414             :                         (errcode(ERRCODE_SYNTAX_ERROR),
     415             :                 /*- translator: both %s are strings of the form "option = value" */
     416             :                          errmsg("subscription with %s must also set %s",
     417             :                                 "slot_name = NONE", "enabled = false")));
     418             :         }
     419             : 
     420          82 :         if (opts->create_slot)
     421             :         {
     422          12 :             if (IsSet(opts->specified_opts, SUBOPT_CREATE_SLOT))
     423           6 :                 ereport(ERROR,
     424             :                         (errcode(ERRCODE_SYNTAX_ERROR),
     425             :                 /*- translator: both %s are strings of the form "option = value" */
     426             :                          errmsg("%s and %s are mutually exclusive options",
     427             :                                 "slot_name = NONE", "create_slot = true")));
     428             :             else
     429           6 :                 ereport(ERROR,
     430             :                         (errcode(ERRCODE_SYNTAX_ERROR),
     431             :                 /*- translator: both %s are strings of the form "option = value" */
     432             :                          errmsg("subscription with %s must also set %s",
     433             :                                 "slot_name = NONE", "create_slot = false")));
     434             :         }
     435             :     }
     436         618 : }
     437             : 
     438             : /*
     439             :  * Add publication names from the list to a string.
     440             :  */
     441             : static void
     442         420 : get_publications_str(List *publications, StringInfo dest, bool quote_literal)
     443             : {
     444             :     ListCell   *lc;
     445         420 :     bool        first = true;
     446             : 
     447             :     Assert(publications != NIL);
     448             : 
     449         998 :     foreach(lc, publications)
     450             :     {
     451         578 :         char       *pubname = strVal(lfirst(lc));
     452             : 
     453         578 :         if (first)
     454         420 :             first = false;
     455             :         else
     456         158 :             appendStringInfoString(dest, ", ");
     457             : 
     458         578 :         if (quote_literal)
     459         566 :             appendStringInfoString(dest, quote_literal_cstr(pubname));
     460             :         else
     461             :         {
     462          12 :             appendStringInfoChar(dest, '"');
     463          12 :             appendStringInfoString(dest, pubname);
     464          12 :             appendStringInfoChar(dest, '"');
     465             :         }
     466             :     }
     467         420 : }
     468             : 
     469             : /*
     470             :  * Check that the specified publications are present on the publisher.
     471             :  */
     472             : static void
     473         178 : check_publications(WalReceiverConn *wrconn, List *publications)
     474             : {
     475             :     WalRcvExecResult *res;
     476             :     StringInfo  cmd;
     477             :     TupleTableSlot *slot;
     478         178 :     List       *publicationsCopy = NIL;
     479         178 :     Oid         tableRow[1] = {TEXTOID};
     480             : 
     481         178 :     cmd = makeStringInfo();
     482         178 :     appendStringInfoString(cmd, "SELECT t.pubname FROM\n"
     483             :                            " pg_catalog.pg_publication t WHERE\n"
     484             :                            " t.pubname IN (");
     485         178 :     get_publications_str(publications, cmd, true);
     486         178 :     appendStringInfoChar(cmd, ')');
     487             : 
     488         178 :     res = walrcv_exec(wrconn, cmd->data, 1, tableRow);
     489         178 :     pfree(cmd->data);
     490         178 :     pfree(cmd);
     491             : 
     492         178 :     if (res->status != WALRCV_OK_TUPLES)
     493           0 :         ereport(ERROR,
     494             :                 errmsg("could not receive list of publications from the publisher: %s",
     495             :                        res->err));
     496             : 
     497         178 :     publicationsCopy = list_copy(publications);
     498             : 
     499             :     /* Process publication(s). */
     500         178 :     slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
     501         408 :     while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
     502             :     {
     503             :         char       *pubname;
     504             :         bool        isnull;
     505             : 
     506         230 :         pubname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
     507             :         Assert(!isnull);
     508             : 
     509             :         /* Delete the publication present in publisher from the list. */
     510         230 :         publicationsCopy = list_delete(publicationsCopy, makeString(pubname));
     511         230 :         ExecClearTuple(slot);
     512             :     }
     513             : 
     514         178 :     ExecDropSingleTupleTableSlot(slot);
     515             : 
     516         178 :     walrcv_clear_result(res);
     517             : 
     518         178 :     if (list_length(publicationsCopy))
     519             :     {
     520             :         /* Prepare the list of non-existent publication(s) for error message. */
     521           6 :         StringInfo  pubnames = makeStringInfo();
     522             : 
     523           6 :         get_publications_str(publicationsCopy, pubnames, false);
     524           6 :         ereport(WARNING,
     525             :                 errcode(ERRCODE_UNDEFINED_OBJECT),
     526             :                 errmsg_plural("publication %s does not exist on the publisher",
     527             :                               "publications %s do not exist on the publisher",
     528             :                               list_length(publicationsCopy),
     529             :                               pubnames->data));
     530             :     }
     531         178 : }
     532             : 
     533             : /*
     534             :  * Auxiliary function to build a text array out of a list of String nodes.
     535             :  */
     536             : static Datum
     537         296 : publicationListToArray(List *publist)
     538             : {
     539             :     ArrayType  *arr;
     540             :     Datum      *datums;
     541             :     MemoryContext memcxt;
     542             :     MemoryContext oldcxt;
     543             : 
     544             :     /* Create memory context for temporary allocations. */
     545         296 :     memcxt = AllocSetContextCreate(CurrentMemoryContext,
     546             :                                    "publicationListToArray to array",
     547             :                                    ALLOCSET_DEFAULT_SIZES);
     548         296 :     oldcxt = MemoryContextSwitchTo(memcxt);
     549             : 
     550         296 :     datums = (Datum *) palloc(sizeof(Datum) * list_length(publist));
     551             : 
     552         296 :     check_duplicates_in_publist(publist, datums);
     553             : 
     554         290 :     MemoryContextSwitchTo(oldcxt);
     555             : 
     556         290 :     arr = construct_array_builtin(datums, list_length(publist), TEXTOID);
     557             : 
     558         290 :     MemoryContextDelete(memcxt);
     559             : 
     560         290 :     return PointerGetDatum(arr);
     561             : }
     562             : 
     563             : /*
     564             :  * Create new subscription.
     565             :  */
     566             : ObjectAddress
     567         356 : CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
     568             :                    bool isTopLevel)
     569             : {
     570             :     Relation    rel;
     571             :     ObjectAddress myself;
     572             :     Oid         subid;
     573             :     bool        nulls[Natts_pg_subscription];
     574             :     Datum       values[Natts_pg_subscription];
     575         356 :     Oid         owner = GetUserId();
     576             :     HeapTuple   tup;
     577             :     char       *conninfo;
     578             :     char        originname[NAMEDATALEN];
     579             :     List       *publications;
     580             :     bits32      supported_opts;
     581         356 :     SubOpts     opts = {0};
     582             :     AclResult   aclresult;
     583             : 
     584             :     /*
     585             :      * Parse and check options.
     586             :      *
     587             :      * Connection and publication should not be specified here.
     588             :      */
     589         356 :     supported_opts = (SUBOPT_CONNECT | SUBOPT_ENABLED | SUBOPT_CREATE_SLOT |
     590             :                       SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
     591             :                       SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
     592             :                       SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
     593             :                       SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
     594             :                       SUBOPT_RUN_AS_OWNER | SUBOPT_ORIGIN);
     595         356 :     parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
     596             : 
     597             :     /*
     598             :      * Since creating a replication slot is not transactional, rolling back
     599             :      * the transaction leaves the created replication slot.  So we cannot run
     600             :      * CREATE SUBSCRIPTION inside a transaction block if creating a
     601             :      * replication slot.
     602             :      */
     603         278 :     if (opts.create_slot)
     604         166 :         PreventInTransactionBlock(isTopLevel, "CREATE SUBSCRIPTION ... WITH (create_slot = true)");
     605             : 
     606             :     /*
     607             :      * We don't want to allow unprivileged users to be able to trigger
     608             :      * attempts to access arbitrary network destinations, so require the user
     609             :      * to have been specifically authorized to create subscriptions.
     610             :      */
     611         272 :     if (!has_privs_of_role(owner, ROLE_PG_CREATE_SUBSCRIPTION))
     612           6 :         ereport(ERROR,
     613             :                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
     614             :                  errmsg("permission denied to create subscription"),
     615             :                  errdetail("Only roles with privileges of the \"%s\" role may create subscriptions.",
     616             :                            "pg_create_subscription")));
     617             : 
     618             :     /*
     619             :      * Since a subscription is a database object, we also check for CREATE
     620             :      * permission on the database.
     621             :      */
     622         266 :     aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId,
     623             :                                 owner, ACL_CREATE);
     624         266 :     if (aclresult != ACLCHECK_OK)
     625          12 :         aclcheck_error(aclresult, OBJECT_DATABASE,
     626           6 :                        get_database_name(MyDatabaseId));
     627             : 
     628             :     /*
     629             :      * Non-superusers are required to set a password for authentication, and
     630             :      * that password must be used by the target server, but the superuser can
     631             :      * exempt a subscription from this requirement.
     632             :      */
     633         260 :     if (!opts.passwordrequired && !superuser_arg(owner))
     634           6 :         ereport(ERROR,
     635             :                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
     636             :                  errmsg("password_required=false is superuser-only"),
     637             :                  errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
     638             : 
     639             :     /*
     640             :      * If built with appropriate switch, whine when regression-testing
     641             :      * conventions for subscription names are violated.
     642             :      */
     643             : #ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
     644             :     if (strncmp(stmt->subname, "regress_", 8) != 0)
     645             :         elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\"");
     646             : #endif
     647             : 
     648         254 :     rel = table_open(SubscriptionRelationId, RowExclusiveLock);
     649             : 
     650             :     /* Check if name is used */
     651         254 :     subid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
     652             :                             MyDatabaseId, CStringGetDatum(stmt->subname));
     653         254 :     if (OidIsValid(subid))
     654             :     {
     655           6 :         ereport(ERROR,
     656             :                 (errcode(ERRCODE_DUPLICATE_OBJECT),
     657             :                  errmsg("subscription \"%s\" already exists",
     658             :                         stmt->subname)));
     659             :     }
     660             : 
     661         248 :     if (!IsSet(opts.specified_opts, SUBOPT_SLOT_NAME) &&
     662         222 :         opts.slot_name == NULL)
     663         222 :         opts.slot_name = stmt->subname;
     664             : 
     665             :     /* The default for synchronous_commit of subscriptions is off. */
     666         248 :     if (opts.synchronous_commit == NULL)
     667         248 :         opts.synchronous_commit = "off";
     668             : 
     669         248 :     conninfo = stmt->conninfo;
     670         248 :     publications = stmt->publication;
     671             : 
     672             :     /* Load the library providing us libpq calls. */
     673         248 :     load_file("libpqwalreceiver", false);
     674             : 
     675             :     /* Check the connection info string. */
     676         248 :     walrcv_check_conninfo(conninfo, opts.passwordrequired && !superuser());
     677             : 
     678             :     /* Everything ok, form a new tuple. */
     679         230 :     memset(values, 0, sizeof(values));
     680         230 :     memset(nulls, false, sizeof(nulls));
     681             : 
     682         230 :     subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
     683             :                                Anum_pg_subscription_oid);
     684         230 :     values[Anum_pg_subscription_oid - 1] = ObjectIdGetDatum(subid);
     685         230 :     values[Anum_pg_subscription_subdbid - 1] = ObjectIdGetDatum(MyDatabaseId);
     686         230 :     values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
     687         230 :     values[Anum_pg_subscription_subname - 1] =
     688         230 :         DirectFunctionCall1(namein, CStringGetDatum(stmt->subname));
     689         230 :     values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(owner);
     690         230 :     values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(opts.enabled);
     691         230 :     values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(opts.binary);
     692         230 :     values[Anum_pg_subscription_substream - 1] = CharGetDatum(opts.streaming);
     693         230 :     values[Anum_pg_subscription_subtwophasestate - 1] =
     694         230 :         CharGetDatum(opts.twophase ?
     695             :                      LOGICALREP_TWOPHASE_STATE_PENDING :
     696             :                      LOGICALREP_TWOPHASE_STATE_DISABLED);
     697         230 :     values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
     698         230 :     values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
     699         230 :     values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner);
     700         230 :     values[Anum_pg_subscription_subconninfo - 1] =
     701         230 :         CStringGetTextDatum(conninfo);
     702         230 :     if (opts.slot_name)
     703         210 :         values[Anum_pg_subscription_subslotname - 1] =
     704         210 :             DirectFunctionCall1(namein, CStringGetDatum(opts.slot_name));
     705             :     else
     706          20 :         nulls[Anum_pg_subscription_subslotname - 1] = true;
     707         230 :     values[Anum_pg_subscription_subsynccommit - 1] =
     708         230 :         CStringGetTextDatum(opts.synchronous_commit);
     709         224 :     values[Anum_pg_subscription_subpublications - 1] =
     710         230 :         publicationListToArray(publications);
     711         224 :     values[Anum_pg_subscription_suborigin - 1] =
     712         224 :         CStringGetTextDatum(opts.origin);
     713             : 
     714         224 :     tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
     715             : 
     716             :     /* Insert tuple into catalog. */
     717         224 :     CatalogTupleInsert(rel, tup);
     718         224 :     heap_freetuple(tup);
     719             : 
     720         224 :     recordDependencyOnOwner(SubscriptionRelationId, subid, owner);
     721             : 
     722         224 :     ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
     723         224 :     replorigin_create(originname);
     724             : 
     725             :     /*
     726             :      * Connect to remote side to execute requested commands and fetch table
     727             :      * info.
     728             :      */
     729         224 :     if (opts.connect)
     730             :     {
     731             :         char       *err;
     732             :         WalReceiverConn *wrconn;
     733             :         List       *tables;
     734             :         ListCell   *lc;
     735             :         char        table_state;
     736             :         bool        must_use_password;
     737             : 
     738             :         /* Try to connect to the publisher. */
     739         150 :         must_use_password = !superuser_arg(owner) && opts.passwordrequired;
     740         150 :         wrconn = walrcv_connect(conninfo, true, must_use_password,
     741             :                                 stmt->subname, &err);
     742         150 :         if (!wrconn)
     743           6 :             ereport(ERROR,
     744             :                     (errcode(ERRCODE_CONNECTION_FAILURE),
     745             :                      errmsg("could not connect to the publisher: %s", err)));
     746             : 
     747         144 :         PG_TRY();
     748             :         {
     749         144 :             check_publications(wrconn, publications);
     750         144 :             check_publications_origin(wrconn, publications, opts.copy_data,
     751             :                                       opts.origin, NULL, 0, stmt->subname);
     752             : 
     753             :             /*
     754             :              * Set sync state based on if we were asked to do data copy or
     755             :              * not.
     756             :              */
     757         144 :             table_state = opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
     758             : 
     759             :             /*
     760             :              * Get the table list from publisher and build local table status
     761             :              * info.
     762             :              */
     763         144 :             tables = fetch_table_list(wrconn, publications);
     764         400 :             foreach(lc, tables)
     765             :             {
     766         258 :                 RangeVar   *rv = (RangeVar *) lfirst(lc);
     767             :                 Oid         relid;
     768             : 
     769         258 :                 relid = RangeVarGetRelid(rv, AccessShareLock, false);
     770             : 
     771             :                 /* Check for supported relkind. */
     772         258 :                 CheckSubscriptionRelkind(get_rel_relkind(relid),
     773         258 :                                          rv->schemaname, rv->relname);
     774             : 
     775         258 :                 AddSubscriptionRelState(subid, relid, table_state,
     776             :                                         InvalidXLogRecPtr);
     777             :             }
     778             : 
     779             :             /*
     780             :              * If requested, create permanent slot for the subscription. We
     781             :              * won't use the initial snapshot for anything, so no need to
     782             :              * export it.
     783             :              */
     784         142 :             if (opts.create_slot)
     785             :             {
     786         140 :                 bool        twophase_enabled = false;
     787             : 
     788             :                 Assert(opts.slot_name);
     789             : 
     790             :                 /*
     791             :                  * Even if two_phase is set, don't create the slot with
     792             :                  * two-phase enabled. Will enable it once all the tables are
     793             :                  * synced and ready. This avoids race-conditions like prepared
     794             :                  * transactions being skipped due to changes not being applied
     795             :                  * due to checks in should_apply_changes_for_rel() when
     796             :                  * tablesync for the corresponding tables are in progress. See
     797             :                  * comments atop worker.c.
     798             :                  *
     799             :                  * Note that if tables were specified but copy_data is false
     800             :                  * then it is safe to enable two_phase up-front because those
     801             :                  * tables are already initially in READY state. When the
     802             :                  * subscription has no tables, we leave the twophase state as
     803             :                  * PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
     804             :                  * PUBLICATION to work.
     805             :                  */
     806         140 :                 if (opts.twophase && !opts.copy_data && tables != NIL)
     807           2 :                     twophase_enabled = true;
     808             : 
     809         140 :                 walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
     810             :                                    CRS_NOEXPORT_SNAPSHOT, NULL);
     811             : 
     812         140 :                 if (twophase_enabled)
     813           2 :                     UpdateTwoPhaseState(subid, LOGICALREP_TWOPHASE_STATE_ENABLED);
     814             : 
     815         140 :                 ereport(NOTICE,
     816             :                         (errmsg("created replication slot \"%s\" on publisher",
     817             :                                 opts.slot_name)));
     818             :             }
     819             :         }
     820           2 :         PG_FINALLY();
     821             :         {
     822         144 :             walrcv_disconnect(wrconn);
     823             :         }
     824         144 :         PG_END_TRY();
     825             :     }
     826             :     else
     827          74 :         ereport(WARNING,
     828             :                 (errmsg("subscription was created, but is not connected"),
     829             :                  errhint("To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.")));
     830             : 
     831         216 :     table_close(rel, RowExclusiveLock);
     832             : 
     833         216 :     pgstat_create_subscription(subid);
     834             : 
     835         216 :     if (opts.enabled)
     836         142 :         ApplyLauncherWakeupAtCommit();
     837             : 
     838         216 :     ObjectAddressSet(myself, SubscriptionRelationId, subid);
     839             : 
     840         216 :     InvokeObjectPostCreateHook(SubscriptionRelationId, subid, 0);
     841             : 
     842         216 :     return myself;
     843             : }
     844             : 
     845             : static void
     846          76 : AlterSubscription_refresh(Subscription *sub, bool copy_data,
     847             :                           List *validate_publications)
     848             : {
     849             :     char       *err;
     850             :     List       *pubrel_names;
     851             :     List       *subrel_states;
     852             :     Oid        *subrel_local_oids;
     853             :     Oid        *pubrel_local_oids;
     854             :     ListCell   *lc;
     855             :     int         off;
     856             :     int         remove_rel_len;
     857             :     int         subrel_count;
     858          76 :     Relation    rel = NULL;
     859             :     typedef struct SubRemoveRels
     860             :     {
     861             :         Oid         relid;
     862             :         char        state;
     863             :     } SubRemoveRels;
     864             :     SubRemoveRels *sub_remove_rels;
     865             :     WalReceiverConn *wrconn;
     866             :     bool        must_use_password;
     867             : 
     868             :     /* Load the library providing us libpq calls. */
     869          76 :     load_file("libpqwalreceiver", false);
     870             : 
     871             :     /* Try to connect to the publisher. */
     872          76 :     must_use_password = !superuser_arg(sub->owner) && sub->passwordrequired;
     873          76 :     wrconn = walrcv_connect(sub->conninfo, true, must_use_password,
     874             :                             sub->name, &err);
     875          76 :     if (!wrconn)
     876           0 :         ereport(ERROR,
     877             :                 (errcode(ERRCODE_CONNECTION_FAILURE),
     878             :                  errmsg("could not connect to the publisher: %s", err)));
     879             : 
     880          76 :     PG_TRY();
     881             :     {
     882          76 :         if (validate_publications)
     883          34 :             check_publications(wrconn, validate_publications);
     884             : 
     885             :         /* Get the table list from publisher. */
     886          76 :         pubrel_names = fetch_table_list(wrconn, sub->publications);
     887             : 
     888             :         /* Get local table list. */
     889          76 :         subrel_states = GetSubscriptionRelations(sub->oid, false);
     890          76 :         subrel_count = list_length(subrel_states);
     891             : 
     892             :         /*
     893             :          * Build qsorted array of local table oids for faster lookup. This can
     894             :          * potentially contain all tables in the database so speed of lookup
     895             :          * is important.
     896             :          */
     897          76 :         subrel_local_oids = palloc(subrel_count * sizeof(Oid));
     898          76 :         off = 0;
     899         262 :         foreach(lc, subrel_states)
     900             :         {
     901         186 :             SubscriptionRelState *relstate = (SubscriptionRelState *) lfirst(lc);
     902             : 
     903         186 :             subrel_local_oids[off++] = relstate->relid;
     904             :         }
     905          76 :         qsort(subrel_local_oids, subrel_count,
     906             :               sizeof(Oid), oid_cmp);
     907             : 
     908          76 :         check_publications_origin(wrconn, sub->publications, copy_data,
     909             :                                   sub->origin, subrel_local_oids,
     910             :                                   subrel_count, sub->name);
     911             : 
     912             :         /*
     913             :          * Rels that we want to remove from subscription and drop any slots
     914             :          * and origins corresponding to them.
     915             :          */
     916          76 :         sub_remove_rels = palloc(subrel_count * sizeof(SubRemoveRels));
     917             : 
     918             :         /*
     919             :          * Walk over the remote tables and try to match them to locally known
     920             :          * tables. If the table is not known locally create a new state for
     921             :          * it.
     922             :          *
     923             :          * Also builds array of local oids of remote tables for the next step.
     924             :          */
     925          76 :         off = 0;
     926          76 :         pubrel_local_oids = palloc(list_length(pubrel_names) * sizeof(Oid));
     927             : 
     928         260 :         foreach(lc, pubrel_names)
     929             :         {
     930         184 :             RangeVar   *rv = (RangeVar *) lfirst(lc);
     931             :             Oid         relid;
     932             : 
     933         184 :             relid = RangeVarGetRelid(rv, AccessShareLock, false);
     934             : 
     935             :             /* Check for supported relkind. */
     936         184 :             CheckSubscriptionRelkind(get_rel_relkind(relid),
     937         184 :                                      rv->schemaname, rv->relname);
     938             : 
     939         184 :             pubrel_local_oids[off++] = relid;
     940             : 
     941         184 :             if (!bsearch(&relid, subrel_local_oids,
     942             :                          subrel_count, sizeof(Oid), oid_cmp))
     943             :             {
     944          58 :                 AddSubscriptionRelState(sub->oid, relid,
     945             :                                         copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
     946             :                                         InvalidXLogRecPtr);
     947          58 :                 ereport(DEBUG1,
     948             :                         (errmsg_internal("table \"%s.%s\" added to subscription \"%s\"",
     949             :                                          rv->schemaname, rv->relname, sub->name)));
     950             :             }
     951             :         }
     952             : 
     953             :         /*
     954             :          * Next remove state for tables we should not care about anymore using
     955             :          * the data we collected above
     956             :          */
     957          76 :         qsort(pubrel_local_oids, list_length(pubrel_names),
     958             :               sizeof(Oid), oid_cmp);
     959             : 
     960          76 :         remove_rel_len = 0;
     961         262 :         for (off = 0; off < subrel_count; off++)
     962             :         {
     963         186 :             Oid         relid = subrel_local_oids[off];
     964             : 
     965         186 :             if (!bsearch(&relid, pubrel_local_oids,
     966         186 :                          list_length(pubrel_names), sizeof(Oid), oid_cmp))
     967             :             {
     968             :                 char        state;
     969             :                 XLogRecPtr  statelsn;
     970             : 
     971             :                 /*
     972             :                  * Lock pg_subscription_rel with AccessExclusiveLock to
     973             :                  * prevent any race conditions with the apply worker
     974             :                  * re-launching workers at the same time this code is trying
     975             :                  * to remove those tables.
     976             :                  *
     977             :                  * Even if new worker for this particular rel is restarted it
     978             :                  * won't be able to make any progress as we hold exclusive
     979             :                  * lock on pg_subscription_rel till the transaction end. It
     980             :                  * will simply exit as there is no corresponding rel entry.
     981             :                  *
     982             :                  * This locking also ensures that the state of rels won't
     983             :                  * change till we are done with this refresh operation.
     984             :                  */
     985          60 :                 if (!rel)
     986          30 :                     rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock);
     987             : 
     988             :                 /* Last known rel state. */
     989          60 :                 state = GetSubscriptionRelState(sub->oid, relid, &statelsn);
     990             : 
     991          60 :                 sub_remove_rels[remove_rel_len].relid = relid;
     992          60 :                 sub_remove_rels[remove_rel_len++].state = state;
     993             : 
     994          60 :                 RemoveSubscriptionRel(sub->oid, relid);
     995             : 
     996          60 :                 logicalrep_worker_stop(sub->oid, relid);
     997             : 
     998             :                 /*
     999             :                  * For READY state, we would have already dropped the
    1000             :                  * tablesync origin.
    1001             :                  */
    1002          60 :                 if (state != SUBREL_STATE_READY)
    1003             :                 {
    1004             :                     char        originname[NAMEDATALEN];
    1005             : 
    1006             :                     /*
    1007             :                      * Drop the tablesync's origin tracking if exists.
    1008             :                      *
    1009             :                      * It is possible that the origin is not yet created for
    1010             :                      * tablesync worker, this can happen for the states before
    1011             :                      * SUBREL_STATE_FINISHEDCOPY. The tablesync worker or
    1012             :                      * apply worker can also concurrently try to drop the
    1013             :                      * origin and by this time the origin might be already
    1014             :                      * removed. For these reasons, passing missing_ok = true.
    1015             :                      */
    1016           0 :                     ReplicationOriginNameForLogicalRep(sub->oid, relid, originname,
    1017             :                                                        sizeof(originname));
    1018           0 :                     replorigin_drop_by_name(originname, true, false);
    1019             :                 }
    1020             : 
    1021          60 :                 ereport(DEBUG1,
    1022             :                         (errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"",
    1023             :                                          get_namespace_name(get_rel_namespace(relid)),
    1024             :                                          get_rel_name(relid),
    1025             :                                          sub->name)));
    1026             :             }
    1027             :         }
    1028             : 
    1029             :         /*
    1030             :          * Drop the tablesync slots associated with removed tables. This has
    1031             :          * to be at the end because otherwise if there is an error while doing
    1032             :          * the database operations we won't be able to rollback dropped slots.
    1033             :          */
    1034         136 :         for (off = 0; off < remove_rel_len; off++)
    1035             :         {
    1036          60 :             if (sub_remove_rels[off].state != SUBREL_STATE_READY &&
    1037           0 :                 sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE)
    1038             :             {
    1039           0 :                 char        syncslotname[NAMEDATALEN] = {0};
    1040             : 
    1041             :                 /*
    1042             :                  * For READY/SYNCDONE states we know the tablesync slot has
    1043             :                  * already been dropped by the tablesync worker.
    1044             :                  *
    1045             :                  * For other states, there is no certainty, maybe the slot
    1046             :                  * does not exist yet. Also, if we fail after removing some of
    1047             :                  * the slots, next time, it will again try to drop already
    1048             :                  * dropped slots and fail. For these reasons, we allow
    1049             :                  * missing_ok = true for the drop.
    1050             :                  */
    1051           0 :                 ReplicationSlotNameForTablesync(sub->oid, sub_remove_rels[off].relid,
    1052             :                                                 syncslotname, sizeof(syncslotname));
    1053           0 :                 ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
    1054             :             }
    1055             :         }
    1056             :     }
    1057           0 :     PG_FINALLY();
    1058             :     {
    1059          76 :         walrcv_disconnect(wrconn);
    1060             :     }
    1061          76 :     PG_END_TRY();
    1062             : 
    1063          76 :     if (rel)
    1064          30 :         table_close(rel, NoLock);
    1065          76 : }
    1066             : 
    1067             : /*
    1068             :  * Alter the existing subscription.
    1069             :  */
    1070             : ObjectAddress
    1071         396 : AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
    1072             :                   bool isTopLevel)
    1073             : {
    1074             :     Relation    rel;
    1075             :     ObjectAddress myself;
    1076             :     bool        nulls[Natts_pg_subscription];
    1077             :     bool        replaces[Natts_pg_subscription];
    1078             :     Datum       values[Natts_pg_subscription];
    1079             :     HeapTuple   tup;
    1080             :     Oid         subid;
    1081         396 :     bool        update_tuple = false;
    1082             :     Subscription *sub;
    1083             :     Form_pg_subscription form;
    1084             :     bits32      supported_opts;
    1085         396 :     SubOpts     opts = {0};
    1086             : 
    1087         396 :     rel = table_open(SubscriptionRelationId, RowExclusiveLock);
    1088             : 
    1089             :     /* Fetch the existing tuple. */
    1090         396 :     tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
    1091             :                               CStringGetDatum(stmt->subname));
    1092             : 
    1093         396 :     if (!HeapTupleIsValid(tup))
    1094           6 :         ereport(ERROR,
    1095             :                 (errcode(ERRCODE_UNDEFINED_OBJECT),
    1096             :                  errmsg("subscription \"%s\" does not exist",
    1097             :                         stmt->subname)));
    1098             : 
    1099         390 :     form = (Form_pg_subscription) GETSTRUCT(tup);
    1100         390 :     subid = form->oid;
    1101             : 
    1102             :     /* must be owner */
    1103         390 :     if (!object_ownercheck(SubscriptionRelationId, subid, GetUserId()))
    1104           0 :         aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
    1105           0 :                        stmt->subname);
    1106             : 
    1107         390 :     sub = GetSubscription(subid, false);
    1108             : 
    1109             :     /*
    1110             :      * Don't allow non-superuser modification of a subscription with
    1111             :      * password_required=false.
    1112             :      */
    1113         390 :     if (!sub->passwordrequired && !superuser())
    1114           0 :         ereport(ERROR,
    1115             :                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
    1116             :                  errmsg("password_required=false is superuser-only"),
    1117             :                  errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
    1118             : 
    1119             :     /* Lock the subscription so nobody else can do anything with it. */
    1120         390 :     LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock);
    1121             : 
    1122             :     /* Form a new tuple. */
    1123         390 :     memset(values, 0, sizeof(values));
    1124         390 :     memset(nulls, false, sizeof(nulls));
    1125         390 :     memset(replaces, false, sizeof(replaces));
    1126             : 
    1127         390 :     switch (stmt->kind)
    1128             :     {
    1129         156 :         case ALTER_SUBSCRIPTION_OPTIONS:
    1130             :             {
    1131         156 :                 supported_opts = (SUBOPT_SLOT_NAME |
    1132             :                                   SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
    1133             :                                   SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
    1134             :                                   SUBOPT_PASSWORD_REQUIRED |
    1135             :                                   SUBOPT_RUN_AS_OWNER | SUBOPT_ORIGIN);
    1136             : 
    1137         156 :                 parse_subscription_options(pstate, stmt->options,
    1138             :                                            supported_opts, &opts);
    1139             : 
    1140         132 :                 if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
    1141             :                 {
    1142             :                     /*
    1143             :                      * The subscription must be disabled to allow slot_name as
    1144             :                      * 'none', otherwise, the apply worker will repeatedly try
    1145             :                      * to stream the data using that slot_name which neither
    1146             :                      * exists on the publisher nor the user will be allowed to
    1147             :                      * create it.
    1148             :                      */
    1149          56 :                     if (sub->enabled && !opts.slot_name)
    1150           0 :                         ereport(ERROR,
    1151             :                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1152             :                                  errmsg("cannot set %s for enabled subscription",
    1153             :                                         "slot_name = NONE")));
    1154             : 
    1155          56 :                     if (opts.slot_name)
    1156           6 :                         values[Anum_pg_subscription_subslotname - 1] =
    1157           6 :                             DirectFunctionCall1(namein, CStringGetDatum(opts.slot_name));
    1158             :                     else
    1159          50 :                         nulls[Anum_pg_subscription_subslotname - 1] = true;
    1160          56 :                     replaces[Anum_pg_subscription_subslotname - 1] = true;
    1161             :                 }
    1162             : 
    1163         132 :                 if (opts.synchronous_commit)
    1164             :                 {
    1165           6 :                     values[Anum_pg_subscription_subsynccommit - 1] =
    1166           6 :                         CStringGetTextDatum(opts.synchronous_commit);
    1167           6 :                     replaces[Anum_pg_subscription_subsynccommit - 1] = true;
    1168             :                 }
    1169             : 
    1170         132 :                 if (IsSet(opts.specified_opts, SUBOPT_BINARY))
    1171             :                 {
    1172          18 :                     values[Anum_pg_subscription_subbinary - 1] =
    1173          18 :                         BoolGetDatum(opts.binary);
    1174          18 :                     replaces[Anum_pg_subscription_subbinary - 1] = true;
    1175             :                 }
    1176             : 
    1177         132 :                 if (IsSet(opts.specified_opts, SUBOPT_STREAMING))
    1178             :                 {
    1179          30 :                     values[Anum_pg_subscription_substream - 1] =
    1180          30 :                         CharGetDatum(opts.streaming);
    1181          30 :                     replaces[Anum_pg_subscription_substream - 1] = true;
    1182             :                 }
    1183             : 
    1184         132 :                 if (IsSet(opts.specified_opts, SUBOPT_DISABLE_ON_ERR))
    1185             :                 {
    1186             :                     values[Anum_pg_subscription_subdisableonerr - 1]
    1187           6 :                         = BoolGetDatum(opts.disableonerr);
    1188             :                     replaces[Anum_pg_subscription_subdisableonerr - 1]
    1189           6 :                         = true;
    1190             :                 }
    1191             : 
    1192         132 :                 if (IsSet(opts.specified_opts, SUBOPT_PASSWORD_REQUIRED))
    1193             :                 {
    1194             :                     /* Non-superuser may not disable password_required. */
    1195          12 :                     if (!opts.passwordrequired && !superuser())
    1196           0 :                         ereport(ERROR,
    1197             :                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
    1198             :                                  errmsg("password_required=false is superuser-only"),
    1199             :                                  errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
    1200             : 
    1201             :                     values[Anum_pg_subscription_subpasswordrequired - 1]
    1202          12 :                         = BoolGetDatum(opts.passwordrequired);
    1203             :                     replaces[Anum_pg_subscription_subpasswordrequired - 1]
    1204          12 :                         = true;
    1205             :                 }
    1206             : 
    1207         132 :                 if (IsSet(opts.specified_opts, SUBOPT_ORIGIN))
    1208             :                 {
    1209           6 :                     values[Anum_pg_subscription_suborigin - 1] =
    1210           6 :                         CStringGetTextDatum(opts.origin);
    1211           6 :                     replaces[Anum_pg_subscription_suborigin - 1] = true;
    1212             :                 }
    1213             : 
    1214         132 :                 update_tuple = true;
    1215         132 :                 break;
    1216             :             }
    1217             : 
    1218          42 :         case ALTER_SUBSCRIPTION_ENABLED:
    1219             :             {
    1220          42 :                 parse_subscription_options(pstate, stmt->options,
    1221             :                                            SUBOPT_ENABLED, &opts);
    1222             :                 Assert(IsSet(opts.specified_opts, SUBOPT_ENABLED));
    1223             : 
    1224          42 :                 if (!sub->slotname && opts.enabled)
    1225           6 :                     ereport(ERROR,
    1226             :                             (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1227             :                              errmsg("cannot enable subscription that does not have a slot name")));
    1228             : 
    1229          36 :                 values[Anum_pg_subscription_subenabled - 1] =
    1230          36 :                     BoolGetDatum(opts.enabled);
    1231          36 :                 replaces[Anum_pg_subscription_subenabled - 1] = true;
    1232             : 
    1233          36 :                 if (opts.enabled)
    1234          20 :                     ApplyLauncherWakeupAtCommit();
    1235             : 
    1236          36 :                 update_tuple = true;
    1237          36 :                 break;
    1238             :             }
    1239             : 
    1240          14 :         case ALTER_SUBSCRIPTION_CONNECTION:
    1241             :             /* Load the library providing us libpq calls. */
    1242          14 :             load_file("libpqwalreceiver", false);
    1243             :             /* Check the connection info string. */
    1244          14 :             walrcv_check_conninfo(stmt->conninfo,
    1245             :                                   sub->passwordrequired && !superuser_arg(sub->owner));
    1246             : 
    1247           8 :             values[Anum_pg_subscription_subconninfo - 1] =
    1248           8 :                 CStringGetTextDatum(stmt->conninfo);
    1249           8 :             replaces[Anum_pg_subscription_subconninfo - 1] = true;
    1250           8 :             update_tuple = true;
    1251           8 :             break;
    1252             : 
    1253          48 :         case ALTER_SUBSCRIPTION_SET_PUBLICATION:
    1254             :             {
    1255          48 :                 supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
    1256          48 :                 parse_subscription_options(pstate, stmt->options,
    1257             :                                            supported_opts, &opts);
    1258             : 
    1259          48 :                 values[Anum_pg_subscription_subpublications - 1] =
    1260          48 :                     publicationListToArray(stmt->publication);
    1261          48 :                 replaces[Anum_pg_subscription_subpublications - 1] = true;
    1262             : 
    1263          48 :                 update_tuple = true;
    1264             : 
    1265             :                 /* Refresh if user asked us to. */
    1266          48 :                 if (opts.refresh)
    1267             :                 {
    1268          42 :                     if (!sub->enabled)
    1269           0 :                         ereport(ERROR,
    1270             :                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1271             :                                  errmsg("ALTER SUBSCRIPTION with refresh is not allowed for disabled subscriptions"),
    1272             :                                  errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION ... WITH (refresh = false).")));
    1273             : 
    1274             :                     /*
    1275             :                      * See ALTER_SUBSCRIPTION_REFRESH for details why this is
    1276             :                      * not allowed.
    1277             :                      */
    1278          42 :                     if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
    1279           0 :                         ereport(ERROR,
    1280             :                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1281             :                                  errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
    1282             :                                  errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
    1283             : 
    1284          42 :                     PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION with refresh");
    1285             : 
    1286             :                     /* Make sure refresh sees the new list of publications. */
    1287          30 :                     sub->publications = stmt->publication;
    1288             : 
    1289          30 :                     AlterSubscription_refresh(sub, opts.copy_data,
    1290             :                                               stmt->publication);
    1291             :                 }
    1292             : 
    1293          36 :                 break;
    1294             :             }
    1295             : 
    1296          54 :         case ALTER_SUBSCRIPTION_ADD_PUBLICATION:
    1297             :         case ALTER_SUBSCRIPTION_DROP_PUBLICATION:
    1298             :             {
    1299             :                 List       *publist;
    1300          54 :                 bool        isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;
    1301             : 
    1302          54 :                 supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
    1303          54 :                 parse_subscription_options(pstate, stmt->options,
    1304             :                                            supported_opts, &opts);
    1305             : 
    1306          54 :                 publist = merge_publications(sub->publications, stmt->publication, isadd, stmt->subname);
    1307          18 :                 values[Anum_pg_subscription_subpublications - 1] =
    1308          18 :                     publicationListToArray(publist);
    1309          18 :                 replaces[Anum_pg_subscription_subpublications - 1] = true;
    1310             : 
    1311          18 :                 update_tuple = true;
    1312             : 
    1313             :                 /* Refresh if user asked us to. */
    1314          18 :                 if (opts.refresh)
    1315             :                 {
    1316             :                     /* We only need to validate user specified publications. */
    1317           6 :                     List       *validate_publications = (isadd) ? stmt->publication : NULL;
    1318             : 
    1319           6 :                     if (!sub->enabled)
    1320           0 :                         ereport(ERROR,
    1321             :                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1322             :                                  errmsg("ALTER SUBSCRIPTION with refresh is not allowed for disabled subscriptions"),
    1323             :                         /* translator: %s is an SQL ALTER command */
    1324             :                                  errhint("Use %s instead.",
    1325             :                                          isadd ?
    1326             :                                          "ALTER SUBSCRIPTION ... ADD PUBLICATION ... WITH (refresh = false)" :
    1327             :                                          "ALTER SUBSCRIPTION ... DROP PUBLICATION ... WITH (refresh = false)")));
    1328             : 
    1329             :                     /*
    1330             :                      * See ALTER_SUBSCRIPTION_REFRESH for details why this is
    1331             :                      * not allowed.
    1332             :                      */
    1333           6 :                     if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
    1334           0 :                         ereport(ERROR,
    1335             :                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1336             :                                  errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
    1337             :                         /* translator: %s is an SQL ALTER command */
    1338             :                                  errhint("Use %s with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.",
    1339             :                                          isadd ?
    1340             :                                          "ALTER SUBSCRIPTION ... ADD PUBLICATION" :
    1341             :                                          "ALTER SUBSCRIPTION ... DROP PUBLICATION")));
    1342             : 
    1343           6 :                     PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION with refresh");
    1344             : 
    1345             :                     /* Refresh the new list of publications. */
    1346           6 :                     sub->publications = publist;
    1347             : 
    1348           6 :                     AlterSubscription_refresh(sub, opts.copy_data,
    1349             :                                               validate_publications);
    1350             :                 }
    1351             : 
    1352          18 :                 break;
    1353             :             }
    1354             : 
    1355          52 :         case ALTER_SUBSCRIPTION_REFRESH:
    1356             :             {
    1357          52 :                 if (!sub->enabled)
    1358           6 :                     ereport(ERROR,
    1359             :                             (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1360             :                              errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));
    1361             : 
    1362          46 :                 parse_subscription_options(pstate, stmt->options,
    1363             :                                            SUBOPT_COPY_DATA, &opts);
    1364             : 
    1365             :                 /*
    1366             :                  * The subscription option "two_phase" requires that
    1367             :                  * replication has passed the initial table synchronization
    1368             :                  * phase before the two_phase becomes properly enabled.
    1369             :                  *
    1370             :                  * But, having reached this two-phase commit "enabled" state
    1371             :                  * we must not allow any subsequent table initialization to
    1372             :                  * occur. So the ALTER SUBSCRIPTION ... REFRESH is disallowed
    1373             :                  * when the user had requested two_phase = on mode.
    1374             :                  *
    1375             :                  * The exception to this restriction is when copy_data =
    1376             :                  * false, because when copy_data is false the tablesync will
    1377             :                  * start already in READY state and will exit directly without
    1378             :                  * doing anything.
    1379             :                  *
    1380             :                  * For more details see comments atop worker.c.
    1381             :                  */
    1382          46 :                 if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
    1383           0 :                     ereport(ERROR,
    1384             :                             (errcode(ERRCODE_SYNTAX_ERROR),
    1385             :                              errmsg("ALTER SUBSCRIPTION ... REFRESH with copy_data is not allowed when two_phase is enabled"),
    1386             :                              errhint("Use ALTER SUBSCRIPTION ... REFRESH with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
    1387             : 
    1388          46 :                 PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH");
    1389             : 
    1390          40 :                 AlterSubscription_refresh(sub, opts.copy_data, NULL);
    1391             : 
    1392          40 :                 break;
    1393             :             }
    1394             : 
    1395          24 :         case ALTER_SUBSCRIPTION_SKIP:
    1396             :             {
    1397          24 :                 parse_subscription_options(pstate, stmt->options, SUBOPT_LSN, &opts);
    1398             : 
    1399             :                 /* ALTER SUBSCRIPTION ... SKIP supports only LSN option */
    1400             :                 Assert(IsSet(opts.specified_opts, SUBOPT_LSN));
    1401             : 
    1402             :                 /*
    1403             :                  * If the user sets subskiplsn, we do a sanity check to make
    1404             :                  * sure that the specified LSN is a probable value.
    1405             :                  */
    1406          18 :                 if (!XLogRecPtrIsInvalid(opts.lsn))
    1407             :                 {
    1408             :                     RepOriginId originid;
    1409             :                     char        originname[NAMEDATALEN];
    1410             :                     XLogRecPtr  remote_lsn;
    1411             : 
    1412          12 :                     ReplicationOriginNameForLogicalRep(subid, InvalidOid,
    1413             :                                                        originname, sizeof(originname));
    1414          12 :                     originid = replorigin_by_name(originname, false);
    1415          12 :                     remote_lsn = replorigin_get_progress(originid, false);
    1416             : 
    1417             :                     /* Check the given LSN is at least a future LSN */
    1418          12 :                     if (!XLogRecPtrIsInvalid(remote_lsn) && opts.lsn < remote_lsn)
    1419           0 :                         ereport(ERROR,
    1420             :                                 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    1421             :                                  errmsg("skip WAL location (LSN %X/%X) must be greater than origin LSN %X/%X",
    1422             :                                         LSN_FORMAT_ARGS(opts.lsn),
    1423             :                                         LSN_FORMAT_ARGS(remote_lsn))));
    1424             :                 }
    1425             : 
    1426          18 :                 values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(opts.lsn);
    1427          18 :                 replaces[Anum_pg_subscription_subskiplsn - 1] = true;
    1428             : 
    1429          18 :                 update_tuple = true;
    1430          18 :                 break;
    1431             :             }
    1432             : 
    1433           0 :         default:
    1434           0 :             elog(ERROR, "unrecognized ALTER SUBSCRIPTION kind %d",
    1435             :                  stmt->kind);
    1436             :     }
    1437             : 
    1438             :     /* Update the catalog if needed. */
    1439         288 :     if (update_tuple)
    1440             :     {
    1441         248 :         tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
    1442             :                                 replaces);
    1443             : 
    1444         248 :         CatalogTupleUpdate(rel, &tup->t_self, tup);
    1445             : 
    1446         248 :         heap_freetuple(tup);
    1447             :     }
    1448             : 
    1449         288 :     table_close(rel, RowExclusiveLock);
    1450             : 
    1451         288 :     ObjectAddressSet(myself, SubscriptionRelationId, subid);
    1452             : 
    1453         288 :     InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0);
    1454             : 
    1455             :     /* Wake up related replication workers to handle this change quickly. */
    1456         288 :     LogicalRepWorkersWakeupAtCommit(subid);
    1457             : 
    1458         288 :     return myself;
    1459             : }
    1460             : 
    1461             : /*
    1462             :  * Drop a subscription
    1463             :  */
    1464             : void
    1465         158 : DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
    1466             : {
    1467             :     Relation    rel;
    1468             :     ObjectAddress myself;
    1469             :     HeapTuple   tup;
    1470             :     Oid         subid;
    1471             :     Oid         subowner;
    1472             :     Datum       datum;
    1473             :     bool        isnull;
    1474             :     char       *subname;
    1475             :     char       *conninfo;
    1476             :     char       *slotname;
    1477             :     List       *subworkers;
    1478             :     ListCell   *lc;
    1479             :     char        originname[NAMEDATALEN];
    1480         158 :     char       *err = NULL;
    1481             :     WalReceiverConn *wrconn;
    1482             :     Form_pg_subscription form;
    1483             :     List       *rstates;
    1484             :     bool        must_use_password;
    1485             : 
    1486             :     /*
    1487             :      * Lock pg_subscription with AccessExclusiveLock to ensure that the
    1488             :      * launcher doesn't restart new worker during dropping the subscription
    1489             :      */
    1490         158 :     rel = table_open(SubscriptionRelationId, AccessExclusiveLock);
    1491             : 
    1492         158 :     tup = SearchSysCache2(SUBSCRIPTIONNAME, MyDatabaseId,
    1493         158 :                           CStringGetDatum(stmt->subname));
    1494             : 
    1495         158 :     if (!HeapTupleIsValid(tup))
    1496             :     {
    1497          12 :         table_close(rel, NoLock);
    1498             : 
    1499          12 :         if (!stmt->missing_ok)
    1500           6 :             ereport(ERROR,
    1501             :                     (errcode(ERRCODE_UNDEFINED_OBJECT),
    1502             :                      errmsg("subscription \"%s\" does not exist",
    1503             :                             stmt->subname)));
    1504             :         else
    1505           6 :             ereport(NOTICE,
    1506             :                     (errmsg("subscription \"%s\" does not exist, skipping",
    1507             :                             stmt->subname)));
    1508             : 
    1509          74 :         return;
    1510             :     }
    1511             : 
    1512         146 :     form = (Form_pg_subscription) GETSTRUCT(tup);
    1513         146 :     subid = form->oid;
    1514         146 :     subowner = form->subowner;
    1515         146 :     must_use_password = !superuser_arg(subowner) && form->subpasswordrequired;
    1516             : 
    1517             :     /* must be owner */
    1518         146 :     if (!object_ownercheck(SubscriptionRelationId, subid, GetUserId()))
    1519           0 :         aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
    1520           0 :                        stmt->subname);
    1521             : 
    1522             :     /* DROP hook for the subscription being removed */
    1523         146 :     InvokeObjectDropHook(SubscriptionRelationId, subid, 0);
    1524             : 
    1525             :     /*
    1526             :      * Lock the subscription so nobody else can do anything with it (including
    1527             :      * the replication workers).
    1528             :      */
    1529         146 :     LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock);
    1530             : 
    1531             :     /* Get subname */
    1532         146 :     datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, tup,
    1533             :                                    Anum_pg_subscription_subname);
    1534         146 :     subname = pstrdup(NameStr(*DatumGetName(datum)));
    1535             : 
    1536             :     /* Get conninfo */
    1537         146 :     datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, tup,
    1538             :                                    Anum_pg_subscription_subconninfo);
    1539         146 :     conninfo = TextDatumGetCString(datum);
    1540             : 
    1541             :     /* Get slotname */
    1542         146 :     datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup,
    1543             :                             Anum_pg_subscription_subslotname, &isnull);
    1544         146 :     if (!isnull)
    1545          76 :         slotname = pstrdup(NameStr(*DatumGetName(datum)));
    1546             :     else
    1547          70 :         slotname = NULL;
    1548             : 
    1549             :     /*
    1550             :      * Since dropping a replication slot is not transactional, the replication
    1551             :      * slot stays dropped even if the transaction rolls back.  So we cannot
    1552             :      * run DROP SUBSCRIPTION inside a transaction block if dropping the
    1553             :      * replication slot.  Also, in this case, we report a message for dropping
    1554             :      * the subscription to the cumulative stats system.
    1555             :      *
    1556             :      * XXX The command name should really be something like "DROP SUBSCRIPTION
    1557             :      * of a subscription that is associated with a replication slot", but we
    1558             :      * don't have the proper facilities for that.
    1559             :      */
    1560         146 :     if (slotname)
    1561          76 :         PreventInTransactionBlock(isTopLevel, "DROP SUBSCRIPTION");
    1562             : 
    1563         140 :     ObjectAddressSet(myself, SubscriptionRelationId, subid);
    1564         140 :     EventTriggerSQLDropAddObject(&myself, true, true);
    1565             : 
    1566             :     /* Remove the tuple from catalog. */
    1567         140 :     CatalogTupleDelete(rel, &tup->t_self);
    1568             : 
    1569         140 :     ReleaseSysCache(tup);
    1570             : 
    1571             :     /*
    1572             :      * Stop all the subscription workers immediately.
    1573             :      *
    1574             :      * This is necessary if we are dropping the replication slot, so that the
    1575             :      * slot becomes accessible.
    1576             :      *
    1577             :      * It is also necessary if the subscription is disabled and was disabled
    1578             :      * in the same transaction.  Then the workers haven't seen the disabling
    1579             :      * yet and will still be running, leading to hangs later when we want to
    1580             :      * drop the replication origin.  If the subscription was disabled before
    1581             :      * this transaction, then there shouldn't be any workers left, so this
    1582             :      * won't make a difference.
    1583             :      *
    1584             :      * New workers won't be started because we hold an exclusive lock on the
    1585             :      * subscription till the end of the transaction.
    1586             :      */
    1587         140 :     LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
    1588         140 :     subworkers = logicalrep_workers_find(subid, false);
    1589         140 :     LWLockRelease(LogicalRepWorkerLock);
    1590         218 :     foreach(lc, subworkers)
    1591             :     {
    1592          78 :         LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
    1593             : 
    1594          78 :         logicalrep_worker_stop(w->subid, w->relid);
    1595             :     }
    1596         140 :     list_free(subworkers);
    1597             : 
    1598             :     /*
    1599             :      * Remove the no-longer-useful entry in the launcher's table of apply
    1600             :      * worker start times.
    1601             :      *
    1602             :      * If this transaction rolls back, the launcher might restart a failed
    1603             :      * apply worker before wal_retrieve_retry_interval milliseconds have
    1604             :      * elapsed, but that's pretty harmless.
    1605             :      */
    1606         140 :     ApplyLauncherForgetWorkerStartTime(subid);
    1607             : 
    1608             :     /*
    1609             :      * Cleanup of tablesync replication origins.
    1610             :      *
    1611             :      * Any READY-state relations would already have dealt with clean-ups.
    1612             :      *
    1613             :      * Note that the state can't change because we have already stopped both
    1614             :      * the apply and tablesync workers and they can't restart because of
    1615             :      * exclusive lock on the subscription.
    1616             :      */
    1617         140 :     rstates = GetSubscriptionRelations(subid, true);
    1618         144 :     foreach(lc, rstates)
    1619             :     {
    1620           4 :         SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
    1621           4 :         Oid         relid = rstate->relid;
    1622             : 
    1623             :         /* Only cleanup resources of tablesync workers */
    1624           4 :         if (!OidIsValid(relid))
    1625           0 :             continue;
    1626             : 
    1627             :         /*
    1628             :          * Drop the tablesync's origin tracking if exists.
    1629             :          *
    1630             :          * It is possible that the origin is not yet created for tablesync
    1631             :          * worker so passing missing_ok = true. This can happen for the states
    1632             :          * before SUBREL_STATE_FINISHEDCOPY.
    1633             :          */
    1634           4 :         ReplicationOriginNameForLogicalRep(subid, relid, originname,
    1635             :                                            sizeof(originname));
    1636           4 :         replorigin_drop_by_name(originname, true, false);
    1637             :     }
    1638             : 
    1639             :     /* Clean up dependencies */
    1640         140 :     deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
    1641             : 
    1642             :     /* Remove any associated relation synchronization states. */
    1643         140 :     RemoveSubscriptionRel(subid, InvalidOid);
    1644             : 
    1645             :     /* Remove the origin tracking if exists. */
    1646         140 :     ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname));
    1647         140 :     replorigin_drop_by_name(originname, true, false);
    1648             : 
    1649             :     /*
    1650             :      * If there is no slot associated with the subscription, we can finish
    1651             :      * here.
    1652             :      */
    1653         140 :     if (!slotname && rstates == NIL)
    1654             :     {
    1655          68 :         table_close(rel, NoLock);
    1656          68 :         return;
    1657             :     }
    1658             : 
    1659             :     /*
    1660             :      * Try to acquire the connection necessary for dropping slots.
    1661             :      *
    1662             :      * Note: If the slotname is NONE/NULL then we allow the command to finish
    1663             :      * and users need to manually cleanup the apply and tablesync worker slots
    1664             :      * later.
    1665             :      *
    1666             :      * This has to be at the end because otherwise if there is an error while
    1667             :      * doing the database operations we won't be able to rollback dropped
    1668             :      * slot.
    1669             :      */
    1670          72 :     load_file("libpqwalreceiver", false);
    1671             : 
    1672          72 :     wrconn = walrcv_connect(conninfo, true, must_use_password,
    1673             :                             subname, &err);
    1674          72 :     if (wrconn == NULL)
    1675             :     {
    1676           0 :         if (!slotname)
    1677             :         {
    1678             :             /* be tidy */
    1679           0 :             list_free(rstates);
    1680           0 :             table_close(rel, NoLock);
    1681           0 :             return;
    1682             :         }
    1683             :         else
    1684             :         {
    1685           0 :             ReportSlotConnectionError(rstates, subid, slotname, err);
    1686             :         }
    1687             :     }
    1688             : 
    1689          72 :     PG_TRY();
    1690             :     {
    1691          76 :         foreach(lc, rstates)
    1692             :         {
    1693           4 :             SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
    1694           4 :             Oid         relid = rstate->relid;
    1695             : 
    1696             :             /* Only cleanup resources of tablesync workers */
    1697           4 :             if (!OidIsValid(relid))
    1698           0 :                 continue;
    1699             : 
    1700             :             /*
    1701             :              * Drop the tablesync slots associated with removed tables.
    1702             :              *
    1703             :              * For SYNCDONE/READY states, the tablesync slot is known to have
    1704             :              * already been dropped by the tablesync worker.
    1705             :              *
    1706             :              * For other states, there is no certainty, maybe the slot does
    1707             :              * not exist yet. Also, if we fail after removing some of the
    1708             :              * slots, next time, it will again try to drop already dropped
    1709             :              * slots and fail. For these reasons, we allow missing_ok = true
    1710             :              * for the drop.
    1711             :              */
    1712           4 :             if (rstate->state != SUBREL_STATE_SYNCDONE)
    1713             :             {
    1714           2 :                 char        syncslotname[NAMEDATALEN] = {0};
    1715             : 
    1716           2 :                 ReplicationSlotNameForTablesync(subid, relid, syncslotname,
    1717             :                                                 sizeof(syncslotname));
    1718           2 :                 ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
    1719             :             }
    1720             :         }
    1721             : 
    1722          72 :         list_free(rstates);
    1723             : 
    1724             :         /*
    1725             :          * If there is a slot associated with the subscription, then drop the
    1726             :          * replication slot at the publisher.
    1727             :          */
    1728          72 :         if (slotname)
    1729          70 :             ReplicationSlotDropAtPubNode(wrconn, slotname, false);
    1730             :     }
    1731           0 :     PG_FINALLY();
    1732             :     {
    1733          72 :         walrcv_disconnect(wrconn);
    1734             :     }
    1735          72 :     PG_END_TRY();
    1736             : 
    1737             :     /*
    1738             :      * Tell the cumulative stats system that the subscription is getting
    1739             :      * dropped.
    1740             :      */
    1741          72 :     pgstat_drop_subscription(subid);
    1742             : 
    1743          72 :     table_close(rel, NoLock);
    1744             : }
    1745             : 
    1746             : /*
    1747             :  * Drop the replication slot at the publisher node using the replication
    1748             :  * connection.
    1749             :  *
    1750             :  * missing_ok - if true then only issue a LOG message if the slot doesn't
    1751             :  * exist.
    1752             :  */
    1753             : void
    1754         382 : ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok)
    1755             : {
    1756             :     StringInfoData cmd;
    1757             : 
    1758             :     Assert(wrconn);
    1759             : 
    1760         382 :     load_file("libpqwalreceiver", false);
    1761             : 
    1762         382 :     initStringInfo(&cmd);
    1763         382 :     appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(slotname));
    1764             : 
    1765         382 :     PG_TRY();
    1766             :     {
    1767             :         WalRcvExecResult *res;
    1768             : 
    1769         382 :         res = walrcv_exec(wrconn, cmd.data, 0, NULL);
    1770             : 
    1771         382 :         if (res->status == WALRCV_OK_COMMAND)
    1772             :         {
    1773             :             /* NOTICE. Success. */
    1774         382 :             ereport(NOTICE,
    1775             :                     (errmsg("dropped replication slot \"%s\" on publisher",
    1776             :                             slotname)));
    1777             :         }
    1778           0 :         else if (res->status == WALRCV_ERROR &&
    1779           0 :                  missing_ok &&
    1780           0 :                  res->sqlstate == ERRCODE_UNDEFINED_OBJECT)
    1781             :         {
    1782             :             /* LOG. Error, but missing_ok = true. */
    1783           0 :             ereport(LOG,
    1784             :                     (errmsg("could not drop replication slot \"%s\" on publisher: %s",
    1785             :                             slotname, res->err)));
    1786             :         }
    1787             :         else
    1788             :         {
    1789             :             /* ERROR. */
    1790           0 :             ereport(ERROR,
    1791             :                     (errcode(ERRCODE_CONNECTION_FAILURE),
    1792             :                      errmsg("could not drop replication slot \"%s\" on publisher: %s",
    1793             :                             slotname, res->err)));
    1794             :         }
    1795             : 
    1796         382 :         walrcv_clear_result(res);
    1797             :     }
    1798           0 :     PG_FINALLY();
    1799             :     {
    1800         382 :         pfree(cmd.data);
    1801             :     }
    1802         382 :     PG_END_TRY();
    1803         382 : }
    1804             : 
    1805             : /*
    1806             :  * Internal workhorse for changing a subscription owner
    1807             :  */
    1808             : static void
    1809          12 : AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
    1810             : {
    1811             :     Form_pg_subscription form;
    1812             :     AclResult   aclresult;
    1813             : 
    1814          12 :     form = (Form_pg_subscription) GETSTRUCT(tup);
    1815             : 
    1816          12 :     if (form->subowner == newOwnerId)
    1817           0 :         return;
    1818             : 
    1819          12 :     if (!object_ownercheck(SubscriptionRelationId, form->oid, GetUserId()))
    1820           0 :         aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
    1821           0 :                        NameStr(form->subname));
    1822             : 
    1823             :     /*
    1824             :      * Don't allow non-superuser modification of a subscription with
    1825             :      * password_required=false.
    1826             :      */
    1827          12 :     if (!form->subpasswordrequired && !superuser())
    1828           0 :         ereport(ERROR,
    1829             :                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
    1830             :                  errmsg("password_required=false is superuser-only"),
    1831             :                  errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
    1832             : 
    1833             :     /* Must be able to become new owner */
    1834          12 :     check_can_set_role(GetUserId(), newOwnerId);
    1835             : 
    1836             :     /*
    1837             :      * current owner must have CREATE on database
    1838             :      *
    1839             :      * This is consistent with how ALTER SCHEMA ... OWNER TO works, but some
    1840             :      * other object types behave differently (e.g. you can't give a table to a
    1841             :      * user who lacks CREATE privileges on a schema).
    1842             :      */
    1843           6 :     aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId,
    1844             :                                 GetUserId(), ACL_CREATE);
    1845           6 :     if (aclresult != ACLCHECK_OK)
    1846           0 :         aclcheck_error(aclresult, OBJECT_DATABASE,
    1847           0 :                        get_database_name(MyDatabaseId));
    1848             : 
    1849           6 :     form->subowner = newOwnerId;
    1850           6 :     CatalogTupleUpdate(rel, &tup->t_self, tup);
    1851             : 
    1852             :     /* Update owner dependency reference */
    1853           6 :     changeDependencyOnOwner(SubscriptionRelationId,
    1854             :                             form->oid,
    1855             :                             newOwnerId);
    1856             : 
    1857           6 :     InvokeObjectPostAlterHook(SubscriptionRelationId,
    1858             :                               form->oid, 0);
    1859             : 
    1860             :     /* Wake up related background processes to handle this change quickly. */
    1861           6 :     ApplyLauncherWakeupAtCommit();
    1862           6 :     LogicalRepWorkersWakeupAtCommit(form->oid);
    1863             : }
    1864             : 
    1865             : /*
    1866             :  * Change subscription owner -- by name
    1867             :  */
    1868             : ObjectAddress
    1869          12 : AlterSubscriptionOwner(const char *name, Oid newOwnerId)
    1870             : {
    1871             :     Oid         subid;
    1872             :     HeapTuple   tup;
    1873             :     Relation    rel;
    1874             :     ObjectAddress address;
    1875             :     Form_pg_subscription form;
    1876             : 
    1877          12 :     rel = table_open(SubscriptionRelationId, RowExclusiveLock);
    1878             : 
    1879          12 :     tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
    1880             :                               CStringGetDatum(name));
    1881             : 
    1882          12 :     if (!HeapTupleIsValid(tup))
    1883           0 :         ereport(ERROR,
    1884             :                 (errcode(ERRCODE_UNDEFINED_OBJECT),
    1885             :                  errmsg("subscription \"%s\" does not exist", name)));
    1886             : 
    1887          12 :     form = (Form_pg_subscription) GETSTRUCT(tup);
    1888          12 :     subid = form->oid;
    1889             : 
    1890          12 :     AlterSubscriptionOwner_internal(rel, tup, newOwnerId);
    1891             : 
    1892           6 :     ObjectAddressSet(address, SubscriptionRelationId, subid);
    1893             : 
    1894           6 :     heap_freetuple(tup);
    1895             : 
    1896           6 :     table_close(rel, RowExclusiveLock);
    1897             : 
    1898           6 :     return address;
    1899             : }
    1900             : 
    1901             : /*
    1902             :  * Change subscription owner -- by OID
    1903             :  */
    1904             : void
    1905           0 : AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId)
    1906             : {
    1907             :     HeapTuple   tup;
    1908             :     Relation    rel;
    1909             : 
    1910           0 :     rel = table_open(SubscriptionRelationId, RowExclusiveLock);
    1911             : 
    1912           0 :     tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
    1913             : 
    1914           0 :     if (!HeapTupleIsValid(tup))
    1915           0 :         ereport(ERROR,
    1916             :                 (errcode(ERRCODE_UNDEFINED_OBJECT),
    1917             :                  errmsg("subscription with OID %u does not exist", subid)));
    1918             : 
    1919           0 :     AlterSubscriptionOwner_internal(rel, tup, newOwnerId);
    1920             : 
    1921           0 :     heap_freetuple(tup);
    1922             : 
    1923           0 :     table_close(rel, RowExclusiveLock);
    1924           0 : }
    1925             : 
    1926             : /*
    1927             :  * Check and log a warning if the publisher has subscribed to the same table
    1928             :  * from some other publisher. This check is required only if "copy_data = true"
    1929             :  * and "origin = none" for CREATE SUBSCRIPTION and
    1930             :  * ALTER SUBSCRIPTION ... REFRESH statements to notify the user that data
    1931             :  * having origin might have been copied.
    1932             :  *
    1933             :  * This check need not be performed on the tables that are already added
    1934             :  * because incremental sync for those tables will happen through WAL and the
    1935             :  * origin of the data can be identified from the WAL records.
    1936             :  *
    1937             :  * subrel_local_oids contains the list of relation oids that are already
    1938             :  * present on the subscriber.
    1939             :  */
    1940             : static void
    1941         220 : check_publications_origin(WalReceiverConn *wrconn, List *publications,
    1942             :                           bool copydata, char *origin, Oid *subrel_local_oids,
    1943             :                           int subrel_count, char *subname)
    1944             : {
    1945             :     WalRcvExecResult *res;
    1946             :     StringInfoData cmd;
    1947             :     TupleTableSlot *slot;
    1948         220 :     Oid         tableRow[1] = {TEXTOID};
    1949         220 :     List       *publist = NIL;
    1950             :     int         i;
    1951             : 
    1952         424 :     if (!copydata || !origin ||
    1953         204 :         (pg_strcasecmp(origin, LOGICALREP_ORIGIN_NONE) != 0))
    1954         208 :         return;
    1955             : 
    1956          12 :     initStringInfo(&cmd);
    1957          12 :     appendStringInfoString(&cmd,
    1958             :                            "SELECT DISTINCT P.pubname AS pubname\n"
    1959             :                            "FROM pg_publication P,\n"
    1960             :                            "     LATERAL pg_get_publication_tables(P.pubname) GPT\n"
    1961             :                            "     JOIN pg_subscription_rel PS ON (GPT.relid = PS.srrelid),\n"
    1962             :                            "     pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
    1963             :                            "WHERE C.oid = GPT.relid AND P.pubname IN (");
    1964          12 :     get_publications_str(publications, &cmd, true);
    1965          12 :     appendStringInfoString(&cmd, ")\n");
    1966             : 
    1967             :     /*
    1968             :      * In case of ALTER SUBSCRIPTION ... REFRESH, subrel_local_oids contains
    1969             :      * the list of relation oids that are already present on the subscriber.
    1970             :      * This check should be skipped for these tables.
    1971             :      */
    1972          18 :     for (i = 0; i < subrel_count; i++)
    1973             :     {
    1974           6 :         Oid         relid = subrel_local_oids[i];
    1975           6 :         char       *schemaname = get_namespace_name(get_rel_namespace(relid));
    1976           6 :         char       *tablename = get_rel_name(relid);
    1977             : 
    1978           6 :         appendStringInfo(&cmd, "AND NOT (N.nspname = '%s' AND C.relname = '%s')\n",
    1979             :                          schemaname, tablename);
    1980             :     }
    1981             : 
    1982          12 :     res = walrcv_exec(wrconn, cmd.data, 1, tableRow);
    1983          12 :     pfree(cmd.data);
    1984             : 
    1985          12 :     if (res->status != WALRCV_OK_TUPLES)
    1986           0 :         ereport(ERROR,
    1987             :                 (errcode(ERRCODE_CONNECTION_FAILURE),
    1988             :                  errmsg("could not receive list of replicated tables from the publisher: %s",
    1989             :                         res->err)));
    1990             : 
    1991             :     /* Process tables. */
    1992          12 :     slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
    1993          16 :     while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
    1994             :     {
    1995             :         char       *pubname;
    1996             :         bool        isnull;
    1997             : 
    1998           4 :         pubname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
    1999             :         Assert(!isnull);
    2000             : 
    2001           4 :         ExecClearTuple(slot);
    2002           4 :         publist = list_append_unique(publist, makeString(pubname));
    2003             :     }
    2004             : 
    2005             :     /*
    2006             :      * Log a warning if the publisher has subscribed to the same table from
    2007             :      * some other publisher. We cannot know the origin of data during the
    2008             :      * initial sync. Data origins can be found only from the WAL by looking at
    2009             :      * the origin id.
    2010             :      *
    2011             :      * XXX: For simplicity, we don't check whether the table has any data or
    2012             :      * not. If the table doesn't have any data then we don't need to
    2013             :      * distinguish between data having origin and data not having origin so we
    2014             :      * can avoid logging a warning in that case.
    2015             :      */
    2016          12 :     if (publist)
    2017             :     {
    2018           4 :         StringInfo  pubnames = makeStringInfo();
    2019             : 
    2020             :         /* Prepare the list of publication(s) for warning message. */
    2021           4 :         get_publications_str(publist, pubnames, false);
    2022           4 :         ereport(WARNING,
    2023             :                 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    2024             :                 errmsg("subscription \"%s\" requested copy_data with origin = NONE but might copy data that had a different origin",
    2025             :                        subname),
    2026             :                 errdetail_plural("Subscribed publication %s is subscribing to other publications.",
    2027             :                                  "Subscribed publications %s are subscribing to other publications.",
    2028             :                                  list_length(publist), pubnames->data),
    2029             :                 errhint("Verify that initial data copied from the publisher tables did not come from other origins."));
    2030             :     }
    2031             : 
    2032          12 :     ExecDropSingleTupleTableSlot(slot);
    2033             : 
    2034          12 :     walrcv_clear_result(res);
    2035             : }
    2036             : 
    2037             : /*
    2038             :  * Get the list of tables which belong to specified publications on the
    2039             :  * publisher connection.
    2040             :  *
    2041             :  * Note that we don't support the case where the column list is different for
    2042             :  * the same table in different publications to avoid sending unwanted column
    2043             :  * information for some of the rows. This can happen when both the column
    2044             :  * list and row filter are specified for different publications.
    2045             :  */
    2046             : static List *
    2047         220 : fetch_table_list(WalReceiverConn *wrconn, List *publications)
    2048             : {
    2049             :     WalRcvExecResult *res;
    2050             :     StringInfoData cmd;
    2051             :     TupleTableSlot *slot;
    2052         220 :     Oid         tableRow[3] = {TEXTOID, TEXTOID, InvalidOid};
    2053         220 :     List       *tablelist = NIL;
    2054         220 :     int         server_version = walrcv_server_version(wrconn);
    2055         220 :     bool        check_columnlist = (server_version >= 150000);
    2056             : 
    2057         220 :     initStringInfo(&cmd);
    2058             : 
    2059             :     /* Get the list of tables from the publisher. */
    2060         220 :     if (server_version >= 160000)
    2061             :     {
    2062             :         StringInfoData pub_names;
    2063             : 
    2064         220 :         tableRow[2] = INT2VECTOROID;
    2065         220 :         initStringInfo(&pub_names);
    2066         220 :         get_publications_str(publications, &pub_names, true);
    2067             : 
    2068             :         /*
    2069             :          * From version 16, we allowed passing multiple publications to the
    2070             :          * function pg_get_publication_tables. This helped to filter out the
    2071             :          * partition table whose ancestor is also published in this
    2072             :          * publication array.
    2073             :          *
    2074             :          * Join pg_get_publication_tables with pg_publication to exclude
    2075             :          * non-existing publications.
    2076             :          *
    2077             :          * Note that attrs are always stored in sorted order so we don't need
    2078             :          * to worry if different publications have specified them in a
    2079             :          * different order. See publication_translate_columns.
    2080             :          */
    2081         220 :         appendStringInfo(&cmd, "SELECT DISTINCT n.nspname, c.relname, gpt.attrs\n"
    2082             :                          "       FROM pg_class c\n"
    2083             :                          "         JOIN pg_namespace n ON n.oid = c.relnamespace\n"
    2084             :                          "         JOIN ( SELECT (pg_get_publication_tables(VARIADIC array_agg(pubname::text))).*\n"
    2085             :                          "                FROM pg_publication\n"
    2086             :                          "                WHERE pubname IN ( %s )) AS gpt\n"
    2087             :                          "             ON gpt.relid = c.oid\n",
    2088             :                          pub_names.data);
    2089             : 
    2090         220 :         pfree(pub_names.data);
    2091             :     }
    2092             :     else
    2093             :     {
    2094           0 :         tableRow[2] = NAMEARRAYOID;
    2095           0 :         appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename \n");
    2096             : 
    2097             :         /* Get column lists for each relation if the publisher supports it */
    2098           0 :         if (check_columnlist)
    2099           0 :             appendStringInfoString(&cmd, ", t.attnames\n");
    2100             : 
    2101           0 :         appendStringInfoString(&cmd, "FROM pg_catalog.pg_publication_tables t\n"
    2102             :                                " WHERE t.pubname IN (");
    2103           0 :         get_publications_str(publications, &cmd, true);
    2104           0 :         appendStringInfoChar(&cmd, ')');
    2105             :     }
    2106             : 
    2107         220 :     res = walrcv_exec(wrconn, cmd.data, check_columnlist ? 3 : 2, tableRow);
    2108         220 :     pfree(cmd.data);
    2109             : 
    2110         220 :     if (res->status != WALRCV_OK_TUPLES)
    2111           0 :         ereport(ERROR,
    2112             :                 (errcode(ERRCODE_CONNECTION_FAILURE),
    2113             :                  errmsg("could not receive list of replicated tables from the publisher: %s",
    2114             :                         res->err)));
    2115             : 
    2116             :     /* Process tables. */
    2117         220 :     slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
    2118         664 :     while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
    2119             :     {
    2120             :         char       *nspname;
    2121             :         char       *relname;
    2122             :         bool        isnull;
    2123             :         RangeVar   *rv;
    2124             : 
    2125         446 :         nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
    2126             :         Assert(!isnull);
    2127         446 :         relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
    2128             :         Assert(!isnull);
    2129             : 
    2130         446 :         rv = makeRangeVar(nspname, relname, -1);
    2131             : 
    2132         446 :         if (check_columnlist && list_member(tablelist, rv))
    2133           2 :             ereport(ERROR,
    2134             :                     errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    2135             :                     errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
    2136             :                            nspname, relname));
    2137             :         else
    2138         444 :             tablelist = lappend(tablelist, rv);
    2139             : 
    2140         444 :         ExecClearTuple(slot);
    2141             :     }
    2142         218 :     ExecDropSingleTupleTableSlot(slot);
    2143             : 
    2144         218 :     walrcv_clear_result(res);
    2145             : 
    2146         218 :     return tablelist;
    2147             : }
    2148             : 
    2149             : /*
    2150             :  * This is to report the connection failure while dropping replication slots.
    2151             :  * Here, we report the WARNING for all tablesync slots so that user can drop
    2152             :  * them manually, if required.
    2153             :  */
    2154             : static void
    2155           0 : ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err)
    2156             : {
    2157             :     ListCell   *lc;
    2158             : 
    2159           0 :     foreach(lc, rstates)
    2160             :     {
    2161           0 :         SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
    2162           0 :         Oid         relid = rstate->relid;
    2163             : 
    2164             :         /* Only cleanup resources of tablesync workers */
    2165           0 :         if (!OidIsValid(relid))
    2166           0 :             continue;
    2167             : 
    2168             :         /*
    2169             :          * Caller needs to ensure that relstate doesn't change underneath us.
    2170             :          * See DropSubscription where we get the relstates.
    2171             :          */
    2172           0 :         if (rstate->state != SUBREL_STATE_SYNCDONE)
    2173             :         {
    2174           0 :             char        syncslotname[NAMEDATALEN] = {0};
    2175             : 
    2176           0 :             ReplicationSlotNameForTablesync(subid, relid, syncslotname,
    2177             :                                             sizeof(syncslotname));
    2178           0 :             elog(WARNING, "could not drop tablesync replication slot \"%s\"",
    2179             :                  syncslotname);
    2180             :         }
    2181             :     }
    2182             : 
    2183           0 :     ereport(ERROR,
    2184             :             (errcode(ERRCODE_CONNECTION_FAILURE),
    2185             :              errmsg("could not connect to publisher when attempting to drop replication slot \"%s\": %s",
    2186             :                     slotname, err),
    2187             :     /* translator: %s is an SQL ALTER command */
    2188             :              errhint("Use %s to disassociate the subscription from the slot.",
    2189             :                      "ALTER SUBSCRIPTION ... SET (slot_name = NONE)")));
    2190             : }
    2191             : 
    2192             : /*
    2193             :  * Check for duplicates in the given list of publications and error out if
    2194             :  * found one.  Add publications to datums as text datums, if datums is not
    2195             :  * NULL.
    2196             :  */
    2197             : static void
    2198         350 : check_duplicates_in_publist(List *publist, Datum *datums)
    2199             : {
    2200             :     ListCell   *cell;
    2201         350 :     int         j = 0;
    2202             : 
    2203         822 :     foreach(cell, publist)
    2204             :     {
    2205         490 :         char       *name = strVal(lfirst(cell));
    2206             :         ListCell   *pcell;
    2207             : 
    2208         762 :         foreach(pcell, publist)
    2209             :         {
    2210         762 :             char       *pname = strVal(lfirst(pcell));
    2211             : 
    2212         762 :             if (pcell == cell)
    2213         472 :                 break;
    2214             : 
    2215         290 :             if (strcmp(name, pname) == 0)
    2216          18 :                 ereport(ERROR,
    2217             :                         (errcode(ERRCODE_DUPLICATE_OBJECT),
    2218             :                          errmsg("publication name \"%s\" used more than once",
    2219             :                                 pname)));
    2220             :         }
    2221             : 
    2222         472 :         if (datums)
    2223         386 :             datums[j++] = CStringGetTextDatum(name);
    2224             :     }
    2225         332 : }
    2226             : 
    2227             : /*
    2228             :  * Merge current subscription's publications and user-specified publications
    2229             :  * from ADD/DROP PUBLICATIONS.
    2230             :  *
    2231             :  * If addpub is true, we will add the list of publications into oldpublist.
    2232             :  * Otherwise, we will delete the list of publications from oldpublist.  The
    2233             :  * returned list is a copy, oldpublist itself is not changed.
    2234             :  *
    2235             :  * subname is the subscription name, for error messages.
    2236             :  */
    2237             : static List *
    2238          54 : merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname)
    2239             : {
    2240             :     ListCell   *lc;
    2241             : 
    2242          54 :     oldpublist = list_copy(oldpublist);
    2243             : 
    2244          54 :     check_duplicates_in_publist(newpublist, NULL);
    2245             : 
    2246          92 :     foreach(lc, newpublist)
    2247             :     {
    2248          68 :         char       *name = strVal(lfirst(lc));
    2249             :         ListCell   *lc2;
    2250          68 :         bool        found = false;
    2251             : 
    2252         134 :         foreach(lc2, oldpublist)
    2253             :         {
    2254         110 :             char       *pubname = strVal(lfirst(lc2));
    2255             : 
    2256         110 :             if (strcmp(name, pubname) == 0)
    2257             :             {
    2258          44 :                 found = true;
    2259          44 :                 if (addpub)
    2260          12 :                     ereport(ERROR,
    2261             :                             (errcode(ERRCODE_DUPLICATE_OBJECT),
    2262             :                              errmsg("publication \"%s\" is already in subscription \"%s\"",
    2263             :                                     name, subname)));
    2264             :                 else
    2265          32 :                     oldpublist = foreach_delete_current(oldpublist, lc2);
    2266             : 
    2267          32 :                 break;
    2268             :             }
    2269             :         }
    2270             : 
    2271          56 :         if (addpub && !found)
    2272          18 :             oldpublist = lappend(oldpublist, makeString(name));
    2273          38 :         else if (!addpub && !found)
    2274           6 :             ereport(ERROR,
    2275             :                     (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    2276             :                      errmsg("publication \"%s\" is not in subscription \"%s\"",
    2277             :                             name, subname)));
    2278             :     }
    2279             : 
    2280             :     /*
    2281             :      * XXX Probably no strong reason for this, but for now it's to make ALTER
    2282             :      * SUBSCRIPTION ... DROP PUBLICATION consistent with SET PUBLICATION.
    2283             :      */
    2284          24 :     if (!oldpublist)
    2285           6 :         ereport(ERROR,
    2286             :                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
    2287             :                  errmsg("cannot drop all the publications from a subscription")));
    2288             : 
    2289          18 :     return oldpublist;
    2290             : }
    2291             : 
    2292             : /*
    2293             :  * Extract the streaming mode value from a DefElem.  This is like
    2294             :  * defGetBoolean() but also accepts the special value of "parallel".
    2295             :  */
    2296             : char
    2297         132 : defGetStreamingMode(DefElem *def)
    2298             : {
    2299             :     /*
    2300             :      * If no parameter value given, assume "true" is meant.
    2301             :      */
    2302         132 :     if (!def->arg)
    2303           0 :         return LOGICALREP_STREAM_ON;
    2304             : 
    2305             :     /*
    2306             :      * Allow 0, 1, "false", "true", "off", "on" or "parallel".
    2307             :      */
    2308         132 :     switch (nodeTag(def->arg))
    2309             :     {
    2310           0 :         case T_Integer:
    2311           0 :             switch (intVal(def->arg))
    2312             :             {
    2313           0 :                 case 0:
    2314           0 :                     return LOGICALREP_STREAM_OFF;
    2315           0 :                 case 1:
    2316           0 :                     return LOGICALREP_STREAM_ON;
    2317           0 :                 default:
    2318             :                     /* otherwise, error out below */
    2319           0 :                     break;
    2320             :             }
    2321           0 :             break;
    2322         132 :         default:
    2323             :             {
    2324         132 :                 char       *sval = defGetString(def);
    2325             : 
    2326             :                 /*
    2327             :                  * The set of strings accepted here should match up with the
    2328             :                  * grammar's opt_boolean_or_string production.
    2329             :                  */
    2330         258 :                 if (pg_strcasecmp(sval, "false") == 0 ||
    2331         126 :                     pg_strcasecmp(sval, "off") == 0)
    2332           6 :                     return LOGICALREP_STREAM_OFF;
    2333         234 :                 if (pg_strcasecmp(sval, "true") == 0 ||
    2334         108 :                     pg_strcasecmp(sval, "on") == 0)
    2335          90 :                     return LOGICALREP_STREAM_ON;
    2336          36 :                 if (pg_strcasecmp(sval, "parallel") == 0)
    2337          30 :                     return LOGICALREP_STREAM_PARALLEL;
    2338             :             }
    2339           6 :             break;
    2340             :     }
    2341             : 
    2342           6 :     ereport(ERROR,
    2343             :             (errcode(ERRCODE_SYNTAX_ERROR),
    2344             :              errmsg("%s requires a Boolean value or \"parallel\"",
    2345             :                     def->defname)));
    2346             :     return LOGICALREP_STREAM_OFF;   /* keep compiler quiet */
    2347             : }

Generated by: LCOV version 1.14