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

Generated by: LCOV version 1.16