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

Generated by: LCOV version 1.14