1/*-------------------------------------------------------------------------
2 *
3 * subscriptioncmds.c
4 * subscription catalog manipulation functions
5 *
6 * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
7 * Portions Copyright (c) 1994, Regents of the University of California
8 *
9 * IDENTIFICATION
10 * subscriptioncmds.c
11 *
12 *-------------------------------------------------------------------------
13 */
14
15#include "postgres.h"
16
17#include "miscadmin.h"
18
19#include "access/htup_details.h"
20#include "access/table.h"
21#include "access/xact.h"
22
23#include "catalog/catalog.h"
24#include "catalog/dependency.h"
25#include "catalog/indexing.h"
26#include "catalog/namespace.h"
27#include "catalog/objectaccess.h"
28#include "catalog/objectaddress.h"
29#include "catalog/pg_type.h"
30#include "catalog/pg_subscription.h"
31#include "catalog/pg_subscription_rel.h"
32
33#include "commands/defrem.h"
34#include "commands/event_trigger.h"
35#include "commands/subscriptioncmds.h"
36
37#include "executor/executor.h"
38
39#include "nodes/makefuncs.h"
40
41#include "replication/logicallauncher.h"
42#include "replication/origin.h"
43#include "replication/walreceiver.h"
44#include "replication/walsender.h"
45#include "replication/worker_internal.h"
46
47#include "storage/lmgr.h"
48
49#include "utils/builtins.h"
50#include "utils/guc.h"
51#include "utils/lsyscache.h"
52#include "utils/memutils.h"
53#include "utils/syscache.h"
54
55static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
56
57/*
58 * Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
59 *
60 * Since not all options can be specified in both commands, this function
61 * will report an error on options if the target output pointer is NULL to
62 * accommodate that.
63 */
64static void
65parse_subscription_options(List *options, bool *connect, bool *enabled_given,
66 bool *enabled, bool *create_slot,
67 bool *slot_name_given, char **slot_name,
68 bool *copy_data, char **synchronous_commit,
69 bool *refresh)
70{
71 ListCell *lc;
72 bool connect_given = false;
73 bool create_slot_given = false;
74 bool copy_data_given = false;
75 bool refresh_given = false;
76
77 /* If connect is specified, the others also need to be. */
78 Assert(!connect || (enabled && create_slot && copy_data));
79
80 if (connect)
81 *connect = true;
82 if (enabled)
83 {
84 *enabled_given = false;
85 *enabled = true;
86 }
87 if (create_slot)
88 *create_slot = true;
89 if (slot_name)
90 {
91 *slot_name_given = false;
92 *slot_name = NULL;
93 }
94 if (copy_data)
95 *copy_data = true;
96 if (synchronous_commit)
97 *synchronous_commit = NULL;
98 if (refresh)
99 *refresh = true;
100
101 /* Parse options */
102 foreach(lc, options)
103 {
104 DefElem *defel = (DefElem *) lfirst(lc);
105
106 if (strcmp(defel->defname, "connect") == 0 && connect)
107 {
108 if (connect_given)
109 ereport(ERROR,
110 (errcode(ERRCODE_SYNTAX_ERROR),
111 errmsg("conflicting or redundant options")));
112
113 connect_given = true;
114 *connect = defGetBoolean(defel);
115 }
116 else if (strcmp(defel->defname, "enabled") == 0 && enabled)
117 {
118 if (*enabled_given)
119 ereport(ERROR,
120 (errcode(ERRCODE_SYNTAX_ERROR),
121 errmsg("conflicting or redundant options")));
122
123 *enabled_given = true;
124 *enabled = defGetBoolean(defel);
125 }
126 else if (strcmp(defel->defname, "create_slot") == 0 && create_slot)
127 {
128 if (create_slot_given)
129 ereport(ERROR,
130 (errcode(ERRCODE_SYNTAX_ERROR),
131 errmsg("conflicting or redundant options")));
132
133 create_slot_given = true;
134 *create_slot = defGetBoolean(defel);
135 }
136 else if (strcmp(defel->defname, "slot_name") == 0 && slot_name)
137 {
138 if (*slot_name_given)
139 ereport(ERROR,
140 (errcode(ERRCODE_SYNTAX_ERROR),
141 errmsg("conflicting or redundant options")));
142
143 *slot_name_given = true;
144 *slot_name = defGetString(defel);
145
146 /* Setting slot_name = NONE is treated as no slot name. */
147 if (strcmp(*slot_name, "none") == 0)
148 *slot_name = NULL;
149 }
150 else if (strcmp(defel->defname, "copy_data") == 0 && copy_data)
151 {
152 if (copy_data_given)
153 ereport(ERROR,
154 (errcode(ERRCODE_SYNTAX_ERROR),
155 errmsg("conflicting or redundant options")));
156
157 copy_data_given = true;
158 *copy_data = defGetBoolean(defel);
159 }
160 else if (strcmp(defel->defname, "synchronous_commit") == 0 &&
161 synchronous_commit)
162 {
163 if (*synchronous_commit)
164 ereport(ERROR,
165 (errcode(ERRCODE_SYNTAX_ERROR),
166 errmsg("conflicting or redundant options")));
167
168 *synchronous_commit = defGetString(defel);
169
170 /* Test if the given value is valid for synchronous_commit GUC. */
171 (void) set_config_option("synchronous_commit", *synchronous_commit,
172 PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
173 false, 0, false);
174 }
175 else if (strcmp(defel->defname, "refresh") == 0 && refresh)
176 {
177 if (refresh_given)
178 ereport(ERROR,
179 (errcode(ERRCODE_SYNTAX_ERROR),
180 errmsg("conflicting or redundant options")));
181
182 refresh_given = true;
183 *refresh = defGetBoolean(defel);
184 }
185 else
186 ereport(ERROR,
187 (errcode(ERRCODE_SYNTAX_ERROR),
188 errmsg("unrecognized subscription parameter: \"%s\"", defel->defname)));
189 }
190
191 /*
192 * We've been explicitly asked to not connect, that requires some
193 * additional processing.
194 */
195 if (connect && !*connect)
196 {
197 /* Check for incompatible options from the user. */
198 if (enabled && *enabled_given && *enabled)
199 ereport(ERROR,
200 (errcode(ERRCODE_SYNTAX_ERROR),
201 /*- translator: both %s are strings of the form "option = value" */
202 errmsg("%s and %s are mutually exclusive options",
203 "connect = false", "enabled = true")));
204
205 if (create_slot && create_slot_given && *create_slot)
206 ereport(ERROR,
207 (errcode(ERRCODE_SYNTAX_ERROR),
208 errmsg("%s and %s are mutually exclusive options",
209 "connect = false", "create_slot = true")));
210
211 if (copy_data && copy_data_given && *copy_data)
212 ereport(ERROR,
213 (errcode(ERRCODE_SYNTAX_ERROR),
214 errmsg("%s and %s are mutually exclusive options",
215 "connect = false", "copy_data = true")));
216
217 /* Change the defaults of other options. */
218 *enabled = false;
219 *create_slot = false;
220 *copy_data = false;
221 }
222
223 /*
224 * Do additional checking for disallowed combination when slot_name = NONE
225 * was used.
226 */
227 if (slot_name && *slot_name_given && !*slot_name)
228 {
229 if (enabled && *enabled_given && *enabled)
230 ereport(ERROR,
231 (errcode(ERRCODE_SYNTAX_ERROR),
232 /*- translator: both %s are strings of the form "option = value" */
233 errmsg("%s and %s are mutually exclusive options",
234 "slot_name = NONE", "enabled = true")));
235
236 if (create_slot && create_slot_given && *create_slot)
237 ereport(ERROR,
238 (errcode(ERRCODE_SYNTAX_ERROR),
239 errmsg("%s and %s are mutually exclusive options",
240 "slot_name = NONE", "create_slot = true")));
241
242 if (enabled && !*enabled_given && *enabled)
243 ereport(ERROR,
244 (errcode(ERRCODE_SYNTAX_ERROR),
245 /*- translator: both %s are strings of the form "option = value" */
246 errmsg("subscription with %s must also set %s",
247 "slot_name = NONE", "enabled = false")));
248
249 if (create_slot && !create_slot_given && *create_slot)
250 ereport(ERROR,
251 (errcode(ERRCODE_SYNTAX_ERROR),
252 errmsg("subscription with %s must also set %s",
253 "slot_name = NONE", "create_slot = false")));
254 }
255}
256
257/*
258 * Auxiliary function to build a text array out of a list of String nodes.
259 */
260static Datum
261publicationListToArray(List *publist)
262{
263 ArrayType *arr;
264 Datum *datums;
265 int j = 0;
266 ListCell *cell;
267 MemoryContext memcxt;
268 MemoryContext oldcxt;
269
270 /* Create memory context for temporary allocations. */
271 memcxt = AllocSetContextCreate(CurrentMemoryContext,
272 "publicationListToArray to array",
273 ALLOCSET_DEFAULT_SIZES);
274 oldcxt = MemoryContextSwitchTo(memcxt);
275
276 datums = (Datum *) palloc(sizeof(Datum) * list_length(publist));
277
278 foreach(cell, publist)
279 {
280 char *name = strVal(lfirst(cell));
281 ListCell *pcell;
282
283 /* Check for duplicates. */
284 foreach(pcell, publist)
285 {
286 char *pname = strVal(lfirst(pcell));
287
288 if (pcell == cell)
289 break;
290
291 if (strcmp(name, pname) == 0)
292 ereport(ERROR,
293 (errcode(ERRCODE_SYNTAX_ERROR),
294 errmsg("publication name \"%s\" used more than once",
295 pname)));
296 }
297
298 datums[j++] = CStringGetTextDatum(name);
299 }
300
301 MemoryContextSwitchTo(oldcxt);
302
303 arr = construct_array(datums, list_length(publist),
304 TEXTOID, -1, false, 'i');
305
306 MemoryContextDelete(memcxt);
307
308 return PointerGetDatum(arr);
309}
310
311/*
312 * Create new subscription.
313 */
314ObjectAddress
315CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel)
316{
317 Relation rel;
318 ObjectAddress myself;
319 Oid subid;
320 bool nulls[Natts_pg_subscription];
321 Datum values[Natts_pg_subscription];
322 Oid owner = GetUserId();
323 HeapTuple tup;
324 bool connect;
325 bool enabled_given;
326 bool enabled;
327 bool copy_data;
328 char *synchronous_commit;
329 char *conninfo;
330 char *slotname;
331 bool slotname_given;
332 char originname[NAMEDATALEN];
333 bool create_slot;
334 List *publications;
335
336 /*
337 * Parse and check options.
338 *
339 * Connection and publication should not be specified here.
340 */
341 parse_subscription_options(stmt->options, &connect, &enabled_given,
342 &enabled, &create_slot, &slotname_given,
343 &slotname, &copy_data, &synchronous_commit,
344 NULL);
345
346 /*
347 * Since creating a replication slot is not transactional, rolling back
348 * the transaction leaves the created replication slot. So we cannot run
349 * CREATE SUBSCRIPTION inside a transaction block if creating a
350 * replication slot.
351 */
352 if (create_slot)
353 PreventInTransactionBlock(isTopLevel, "CREATE SUBSCRIPTION ... WITH (create_slot = true)");
354
355 if (!superuser())
356 ereport(ERROR,
357 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
358 (errmsg("must be superuser to create subscriptions"))));
359
360 /*
361 * If built with appropriate switch, whine when regression-testing
362 * conventions for subscription names are violated.
363 */
364#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
365 if (strncmp(stmt->subname, "regress_", 8) != 0)
366 elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\"");
367#endif
368
369 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
370
371 /* Check if name is used */
372 subid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
373 MyDatabaseId, CStringGetDatum(stmt->subname));
374 if (OidIsValid(subid))
375 {
376 ereport(ERROR,
377 (errcode(ERRCODE_DUPLICATE_OBJECT),
378 errmsg("subscription \"%s\" already exists",
379 stmt->subname)));
380 }
381
382 if (!slotname_given && slotname == NULL)
383 slotname = stmt->subname;
384
385 /* The default for synchronous_commit of subscriptions is off. */
386 if (synchronous_commit == NULL)
387 synchronous_commit = "off";
388
389 conninfo = stmt->conninfo;
390 publications = stmt->publication;
391
392 /* Load the library providing us libpq calls. */
393 load_file("libpqwalreceiver", false);
394
395 /* Check the connection info string. */
396 walrcv_check_conninfo(conninfo);
397
398 /* Everything ok, form a new tuple. */
399 memset(values, 0, sizeof(values));
400 memset(nulls, false, sizeof(nulls));
401
402 subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
403 Anum_pg_subscription_oid);
404 values[Anum_pg_subscription_oid - 1] = ObjectIdGetDatum(subid);
405 values[Anum_pg_subscription_subdbid - 1] = ObjectIdGetDatum(MyDatabaseId);
406 values[Anum_pg_subscription_subname - 1] =
407 DirectFunctionCall1(namein, CStringGetDatum(stmt->subname));
408 values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(owner);
409 values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled);
410 values[Anum_pg_subscription_subconninfo - 1] =
411 CStringGetTextDatum(conninfo);
412 if (slotname)
413 values[Anum_pg_subscription_subslotname - 1] =
414 DirectFunctionCall1(namein, CStringGetDatum(slotname));
415 else
416 nulls[Anum_pg_subscription_subslotname - 1] = true;
417 values[Anum_pg_subscription_subsynccommit - 1] =
418 CStringGetTextDatum(synchronous_commit);
419 values[Anum_pg_subscription_subpublications - 1] =
420 publicationListToArray(publications);
421
422 tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
423
424 /* Insert tuple into catalog. */
425 CatalogTupleInsert(rel, tup);
426 heap_freetuple(tup);
427
428 recordDependencyOnOwner(SubscriptionRelationId, subid, owner);
429
430 snprintf(originname, sizeof(originname), "pg_%u", subid);
431 replorigin_create(originname);
432
433 /*
434 * Connect to remote side to execute requested commands and fetch table
435 * info.
436 */
437 if (connect)
438 {
439 XLogRecPtr lsn;
440 char *err;
441 WalReceiverConn *wrconn;
442 List *tables;
443 ListCell *lc;
444 char table_state;
445
446 /* Try to connect to the publisher. */
447 wrconn = walrcv_connect(conninfo, true, stmt->subname, &err);
448 if (!wrconn)
449 ereport(ERROR,
450 (errmsg("could not connect to the publisher: %s", err)));
451
452 PG_TRY();
453 {
454 /*
455 * Set sync state based on if we were asked to do data copy or
456 * not.
457 */
458 table_state = copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
459
460 /*
461 * Get the table list from publisher and build local table status
462 * info.
463 */
464 tables = fetch_table_list(wrconn, publications);
465 foreach(lc, tables)
466 {
467 RangeVar *rv = (RangeVar *) lfirst(lc);
468 Oid relid;
469
470 relid = RangeVarGetRelid(rv, AccessShareLock, false);
471
472 /* Check for supported relkind. */
473 CheckSubscriptionRelkind(get_rel_relkind(relid),
474 rv->schemaname, rv->relname);
475
476 AddSubscriptionRelState(subid, relid, table_state,
477 InvalidXLogRecPtr);
478 }
479
480 /*
481 * If requested, create permanent slot for the subscription. We
482 * won't use the initial snapshot for anything, so no need to
483 * export it.
484 */
485 if (create_slot)
486 {
487 Assert(slotname);
488
489 walrcv_create_slot(wrconn, slotname, false,
490 CRS_NOEXPORT_SNAPSHOT, &lsn);
491 ereport(NOTICE,
492 (errmsg("created replication slot \"%s\" on publisher",
493 slotname)));
494 }
495 }
496 PG_CATCH();
497 {
498 /* Close the connection in case of failure. */
499 walrcv_disconnect(wrconn);
500 PG_RE_THROW();
501 }
502 PG_END_TRY();
503
504 /* And we are done with the remote side. */
505 walrcv_disconnect(wrconn);
506 }
507 else
508 ereport(WARNING,
509 /* translator: %s is an SQL ALTER statement */
510 (errmsg("tables were not subscribed, you will have to run %s to subscribe the tables",
511 "ALTER SUBSCRIPTION ... REFRESH PUBLICATION")));
512
513 table_close(rel, RowExclusiveLock);
514
515 if (enabled)
516 ApplyLauncherWakeupAtCommit();
517
518 ObjectAddressSet(myself, SubscriptionRelationId, subid);
519
520 InvokeObjectPostCreateHook(SubscriptionRelationId, subid, 0);
521
522 return myself;
523}
524
525static void
526AlterSubscription_refresh(Subscription *sub, bool copy_data)
527{
528 char *err;
529 List *pubrel_names;
530 List *subrel_states;
531 Oid *subrel_local_oids;
532 Oid *pubrel_local_oids;
533 ListCell *lc;
534 int off;
535
536 /* Load the library providing us libpq calls. */
537 load_file("libpqwalreceiver", false);
538
539 /* Try to connect to the publisher. */
540 wrconn = walrcv_connect(sub->conninfo, true, sub->name, &err);
541 if (!wrconn)
542 ereport(ERROR,
543 (errmsg("could not connect to the publisher: %s", err)));
544
545 /* Get the table list from publisher. */
546 pubrel_names = fetch_table_list(wrconn, sub->publications);
547
548 /* We are done with the remote side, close connection. */
549 walrcv_disconnect(wrconn);
550
551 /* Get local table list. */
552 subrel_states = GetSubscriptionRelations(sub->oid);
553
554 /*
555 * Build qsorted array of local table oids for faster lookup. This can
556 * potentially contain all tables in the database so speed of lookup is
557 * important.
558 */
559 subrel_local_oids = palloc(list_length(subrel_states) * sizeof(Oid));
560 off = 0;
561 foreach(lc, subrel_states)
562 {
563 SubscriptionRelState *relstate = (SubscriptionRelState *) lfirst(lc);
564
565 subrel_local_oids[off++] = relstate->relid;
566 }
567 qsort(subrel_local_oids, list_length(subrel_states),
568 sizeof(Oid), oid_cmp);
569
570 /*
571 * Walk over the remote tables and try to match them to locally known
572 * tables. If the table is not known locally create a new state for it.
573 *
574 * Also builds array of local oids of remote tables for the next step.
575 */
576 off = 0;
577 pubrel_local_oids = palloc(list_length(pubrel_names) * sizeof(Oid));
578
579 foreach(lc, pubrel_names)
580 {
581 RangeVar *rv = (RangeVar *) lfirst(lc);
582 Oid relid;
583
584 relid = RangeVarGetRelid(rv, AccessShareLock, false);
585
586 /* Check for supported relkind. */
587 CheckSubscriptionRelkind(get_rel_relkind(relid),
588 rv->schemaname, rv->relname);
589
590 pubrel_local_oids[off++] = relid;
591
592 if (!bsearch(&relid, subrel_local_oids,
593 list_length(subrel_states), sizeof(Oid), oid_cmp))
594 {
595 AddSubscriptionRelState(sub->oid, relid,
596 copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
597 InvalidXLogRecPtr);
598 ereport(DEBUG1,
599 (errmsg("table \"%s.%s\" added to subscription \"%s\"",
600 rv->schemaname, rv->relname, sub->name)));
601 }
602 }
603
604 /*
605 * Next remove state for tables we should not care about anymore using the
606 * data we collected above
607 */
608 qsort(pubrel_local_oids, list_length(pubrel_names),
609 sizeof(Oid), oid_cmp);
610
611 for (off = 0; off < list_length(subrel_states); off++)
612 {
613 Oid relid = subrel_local_oids[off];
614
615 if (!bsearch(&relid, pubrel_local_oids,
616 list_length(pubrel_names), sizeof(Oid), oid_cmp))
617 {
618 RemoveSubscriptionRel(sub->oid, relid);
619
620 logicalrep_worker_stop_at_commit(sub->oid, relid);
621
622 ereport(DEBUG1,
623 (errmsg("table \"%s.%s\" removed from subscription \"%s\"",
624 get_namespace_name(get_rel_namespace(relid)),
625 get_rel_name(relid),
626 sub->name)));
627 }
628 }
629}
630
631/*
632 * Alter the existing subscription.
633 */
634ObjectAddress
635AlterSubscription(AlterSubscriptionStmt *stmt)
636{
637 Relation rel;
638 ObjectAddress myself;
639 bool nulls[Natts_pg_subscription];
640 bool replaces[Natts_pg_subscription];
641 Datum values[Natts_pg_subscription];
642 HeapTuple tup;
643 Oid subid;
644 bool update_tuple = false;
645 Subscription *sub;
646 Form_pg_subscription form;
647
648 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
649
650 /* Fetch the existing tuple. */
651 tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
652 CStringGetDatum(stmt->subname));
653
654 if (!HeapTupleIsValid(tup))
655 ereport(ERROR,
656 (errcode(ERRCODE_UNDEFINED_OBJECT),
657 errmsg("subscription \"%s\" does not exist",
658 stmt->subname)));
659
660 form = (Form_pg_subscription) GETSTRUCT(tup);
661 subid = form->oid;
662
663 /* must be owner */
664 if (!pg_subscription_ownercheck(subid, GetUserId()))
665 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
666 stmt->subname);
667
668 sub = GetSubscription(subid, false);
669
670 /* Lock the subscription so nobody else can do anything with it. */
671 LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock);
672
673 /* Form a new tuple. */
674 memset(values, 0, sizeof(values));
675 memset(nulls, false, sizeof(nulls));
676 memset(replaces, false, sizeof(replaces));
677
678 switch (stmt->kind)
679 {
680 case ALTER_SUBSCRIPTION_OPTIONS:
681 {
682 char *slotname;
683 bool slotname_given;
684 char *synchronous_commit;
685
686 parse_subscription_options(stmt->options, NULL, NULL, NULL,
687 NULL, &slotname_given, &slotname,
688 NULL, &synchronous_commit, NULL);
689
690 if (slotname_given)
691 {
692 if (sub->enabled && !slotname)
693 ereport(ERROR,
694 (errcode(ERRCODE_SYNTAX_ERROR),
695 errmsg("cannot set %s for enabled subscription",
696 "slot_name = NONE")));
697
698 if (slotname)
699 values[Anum_pg_subscription_subslotname - 1] =
700 DirectFunctionCall1(namein, CStringGetDatum(slotname));
701 else
702 nulls[Anum_pg_subscription_subslotname - 1] = true;
703 replaces[Anum_pg_subscription_subslotname - 1] = true;
704 }
705
706 if (synchronous_commit)
707 {
708 values[Anum_pg_subscription_subsynccommit - 1] =
709 CStringGetTextDatum(synchronous_commit);
710 replaces[Anum_pg_subscription_subsynccommit - 1] = true;
711 }
712
713 update_tuple = true;
714 break;
715 }
716
717 case ALTER_SUBSCRIPTION_ENABLED:
718 {
719 bool enabled,
720 enabled_given;
721
722 parse_subscription_options(stmt->options, NULL,
723 &enabled_given, &enabled, NULL,
724 NULL, NULL, NULL, NULL, NULL);
725 Assert(enabled_given);
726
727 if (!sub->slotname && enabled)
728 ereport(ERROR,
729 (errcode(ERRCODE_SYNTAX_ERROR),
730 errmsg("cannot enable subscription that does not have a slot name")));
731
732 values[Anum_pg_subscription_subenabled - 1] =
733 BoolGetDatum(enabled);
734 replaces[Anum_pg_subscription_subenabled - 1] = true;
735
736 if (enabled)
737 ApplyLauncherWakeupAtCommit();
738
739 update_tuple = true;
740 break;
741 }
742
743 case ALTER_SUBSCRIPTION_CONNECTION:
744 /* Load the library providing us libpq calls. */
745 load_file("libpqwalreceiver", false);
746 /* Check the connection info string. */
747 walrcv_check_conninfo(stmt->conninfo);
748
749 values[Anum_pg_subscription_subconninfo - 1] =
750 CStringGetTextDatum(stmt->conninfo);
751 replaces[Anum_pg_subscription_subconninfo - 1] = true;
752 update_tuple = true;
753 break;
754
755 case ALTER_SUBSCRIPTION_PUBLICATION:
756 {
757 bool copy_data;
758 bool refresh;
759
760 parse_subscription_options(stmt->options, NULL, NULL, NULL,
761 NULL, NULL, NULL, &copy_data,
762 NULL, &refresh);
763
764 values[Anum_pg_subscription_subpublications - 1] =
765 publicationListToArray(stmt->publication);
766 replaces[Anum_pg_subscription_subpublications - 1] = true;
767
768 update_tuple = true;
769
770 /* Refresh if user asked us to. */
771 if (refresh)
772 {
773 if (!sub->enabled)
774 ereport(ERROR,
775 (errcode(ERRCODE_SYNTAX_ERROR),
776 errmsg("ALTER SUBSCRIPTION with refresh is not allowed for disabled subscriptions"),
777 errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION ... WITH (refresh = false).")));
778
779 /* Make sure refresh sees the new list of publications. */
780 sub->publications = stmt->publication;
781
782 AlterSubscription_refresh(sub, copy_data);
783 }
784
785 break;
786 }
787
788 case ALTER_SUBSCRIPTION_REFRESH:
789 {
790 bool copy_data;
791
792 if (!sub->enabled)
793 ereport(ERROR,
794 (errcode(ERRCODE_SYNTAX_ERROR),
795 errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));
796
797 parse_subscription_options(stmt->options, NULL, NULL, NULL,
798 NULL, NULL, NULL, &copy_data,
799 NULL, NULL);
800
801 AlterSubscription_refresh(sub, copy_data);
802
803 break;
804 }
805
806 default:
807 elog(ERROR, "unrecognized ALTER SUBSCRIPTION kind %d",
808 stmt->kind);
809 }
810
811 /* Update the catalog if needed. */
812 if (update_tuple)
813 {
814 tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
815 replaces);
816
817 CatalogTupleUpdate(rel, &tup->t_self, tup);
818
819 heap_freetuple(tup);
820 }
821
822 table_close(rel, RowExclusiveLock);
823
824 ObjectAddressSet(myself, SubscriptionRelationId, subid);
825
826 InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0);
827
828 return myself;
829}
830
831/*
832 * Drop a subscription
833 */
834void
835DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
836{
837 Relation rel;
838 ObjectAddress myself;
839 HeapTuple tup;
840 Oid subid;
841 Datum datum;
842 bool isnull;
843 char *subname;
844 char *conninfo;
845 char *slotname;
846 List *subworkers;
847 ListCell *lc;
848 char originname[NAMEDATALEN];
849 char *err = NULL;
850 RepOriginId originid;
851 WalReceiverConn *wrconn = NULL;
852 StringInfoData cmd;
853 Form_pg_subscription form;
854
855 /*
856 * Lock pg_subscription with AccessExclusiveLock to ensure that the
857 * launcher doesn't restart new worker during dropping the subscription
858 */
859 rel = table_open(SubscriptionRelationId, AccessExclusiveLock);
860
861 tup = SearchSysCache2(SUBSCRIPTIONNAME, MyDatabaseId,
862 CStringGetDatum(stmt->subname));
863
864 if (!HeapTupleIsValid(tup))
865 {
866 table_close(rel, NoLock);
867
868 if (!stmt->missing_ok)
869 ereport(ERROR,
870 (errcode(ERRCODE_UNDEFINED_OBJECT),
871 errmsg("subscription \"%s\" does not exist",
872 stmt->subname)));
873 else
874 ereport(NOTICE,
875 (errmsg("subscription \"%s\" does not exist, skipping",
876 stmt->subname)));
877
878 return;
879 }
880
881 form = (Form_pg_subscription) GETSTRUCT(tup);
882 subid = form->oid;
883
884 /* must be owner */
885 if (!pg_subscription_ownercheck(subid, GetUserId()))
886 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
887 stmt->subname);
888
889 /* DROP hook for the subscription being removed */
890 InvokeObjectDropHook(SubscriptionRelationId, subid, 0);
891
892 /*
893 * Lock the subscription so nobody else can do anything with it (including
894 * the replication workers).
895 */
896 LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock);
897
898 /* Get subname */
899 datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup,
900 Anum_pg_subscription_subname, &isnull);
901 Assert(!isnull);
902 subname = pstrdup(NameStr(*DatumGetName(datum)));
903
904 /* Get conninfo */
905 datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup,
906 Anum_pg_subscription_subconninfo, &isnull);
907 Assert(!isnull);
908 conninfo = TextDatumGetCString(datum);
909
910 /* Get slotname */
911 datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup,
912 Anum_pg_subscription_subslotname, &isnull);
913 if (!isnull)
914 slotname = pstrdup(NameStr(*DatumGetName(datum)));
915 else
916 slotname = NULL;
917
918 /*
919 * Since dropping a replication slot is not transactional, the replication
920 * slot stays dropped even if the transaction rolls back. So we cannot
921 * run DROP SUBSCRIPTION inside a transaction block if dropping the
922 * replication slot.
923 *
924 * XXX The command name should really be something like "DROP SUBSCRIPTION
925 * of a subscription that is associated with a replication slot", but we
926 * don't have the proper facilities for that.
927 */
928 if (slotname)
929 PreventInTransactionBlock(isTopLevel, "DROP SUBSCRIPTION");
930
931
932 ObjectAddressSet(myself, SubscriptionRelationId, subid);
933 EventTriggerSQLDropAddObject(&myself, true, true);
934
935 /* Remove the tuple from catalog. */
936 CatalogTupleDelete(rel, &tup->t_self);
937
938 ReleaseSysCache(tup);
939
940 /*
941 * Stop all the subscription workers immediately.
942 *
943 * This is necessary if we are dropping the replication slot, so that the
944 * slot becomes accessible.
945 *
946 * It is also necessary if the subscription is disabled and was disabled
947 * in the same transaction. Then the workers haven't seen the disabling
948 * yet and will still be running, leading to hangs later when we want to
949 * drop the replication origin. If the subscription was disabled before
950 * this transaction, then there shouldn't be any workers left, so this
951 * won't make a difference.
952 *
953 * New workers won't be started because we hold an exclusive lock on the
954 * subscription till the end of the transaction.
955 */
956 LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
957 subworkers = logicalrep_workers_find(subid, false);
958 LWLockRelease(LogicalRepWorkerLock);
959 foreach(lc, subworkers)
960 {
961 LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
962
963 logicalrep_worker_stop(w->subid, w->relid);
964 }
965 list_free(subworkers);
966
967 /* Clean up dependencies */
968 deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
969
970 /* Remove any associated relation synchronization states. */
971 RemoveSubscriptionRel(subid, InvalidOid);
972
973 /* Remove the origin tracking if exists. */
974 snprintf(originname, sizeof(originname), "pg_%u", subid);
975 originid = replorigin_by_name(originname, true);
976 if (originid != InvalidRepOriginId)
977 replorigin_drop(originid, false);
978
979 /*
980 * If there is no slot associated with the subscription, we can finish
981 * here.
982 */
983 if (!slotname)
984 {
985 table_close(rel, NoLock);
986 return;
987 }
988
989 /*
990 * Otherwise drop the replication slot at the publisher node using the
991 * replication connection.
992 */
993 load_file("libpqwalreceiver", false);
994
995 initStringInfo(&cmd);
996 appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s WAIT", quote_identifier(slotname));
997
998 wrconn = walrcv_connect(conninfo, true, subname, &err);
999 if (wrconn == NULL)
1000 ereport(ERROR,
1001 (errmsg("could not connect to publisher when attempting to "
1002 "drop the replication slot \"%s\"", slotname),
1003 errdetail("The error was: %s", err),
1004 /* translator: %s is an SQL ALTER command */
1005 errhint("Use %s to disassociate the subscription from the slot.",
1006 "ALTER SUBSCRIPTION ... SET (slot_name = NONE)")));
1007
1008 PG_TRY();
1009 {
1010 WalRcvExecResult *res;
1011
1012 res = walrcv_exec(wrconn, cmd.data, 0, NULL);
1013
1014 if (res->status != WALRCV_OK_COMMAND)
1015 ereport(ERROR,
1016 (errmsg("could not drop the replication slot \"%s\" on publisher",
1017 slotname),
1018 errdetail("The error was: %s", res->err)));
1019 else
1020 ereport(NOTICE,
1021 (errmsg("dropped replication slot \"%s\" on publisher",
1022 slotname)));
1023
1024 walrcv_clear_result(res);
1025 }
1026 PG_CATCH();
1027 {
1028 /* Close the connection in case of failure */
1029 walrcv_disconnect(wrconn);
1030 PG_RE_THROW();
1031 }
1032 PG_END_TRY();
1033
1034 walrcv_disconnect(wrconn);
1035
1036 pfree(cmd.data);
1037
1038 table_close(rel, NoLock);
1039}
1040
1041/*
1042 * Internal workhorse for changing a subscription owner
1043 */
1044static void
1045AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
1046{
1047 Form_pg_subscription form;
1048
1049 form = (Form_pg_subscription) GETSTRUCT(tup);
1050
1051 if (form->subowner == newOwnerId)
1052 return;
1053
1054 if (!pg_subscription_ownercheck(form->oid, GetUserId()))
1055 aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
1056 NameStr(form->subname));
1057
1058 /* New owner must be a superuser */
1059 if (!superuser_arg(newOwnerId))
1060 ereport(ERROR,
1061 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1062 errmsg("permission denied to change owner of subscription \"%s\"",
1063 NameStr(form->subname)),
1064 errhint("The owner of a subscription must be a superuser.")));
1065
1066 form->subowner = newOwnerId;
1067 CatalogTupleUpdate(rel, &tup->t_self, tup);
1068
1069 /* Update owner dependency reference */
1070 changeDependencyOnOwner(SubscriptionRelationId,
1071 form->oid,
1072 newOwnerId);
1073
1074 InvokeObjectPostAlterHook(SubscriptionRelationId,
1075 form->oid, 0);
1076}
1077
1078/*
1079 * Change subscription owner -- by name
1080 */
1081ObjectAddress
1082AlterSubscriptionOwner(const char *name, Oid newOwnerId)
1083{
1084 Oid subid;
1085 HeapTuple tup;
1086 Relation rel;
1087 ObjectAddress address;
1088 Form_pg_subscription form;
1089
1090 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
1091
1092 tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
1093 CStringGetDatum(name));
1094
1095 if (!HeapTupleIsValid(tup))
1096 ereport(ERROR,
1097 (errcode(ERRCODE_UNDEFINED_OBJECT),
1098 errmsg("subscription \"%s\" does not exist", name)));
1099
1100 form = (Form_pg_subscription) GETSTRUCT(tup);
1101 subid = form->oid;
1102
1103 AlterSubscriptionOwner_internal(rel, tup, newOwnerId);
1104
1105 ObjectAddressSet(address, SubscriptionRelationId, subid);
1106
1107 heap_freetuple(tup);
1108
1109 table_close(rel, RowExclusiveLock);
1110
1111 return address;
1112}
1113
1114/*
1115 * Change subscription owner -- by OID
1116 */
1117void
1118AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId)
1119{
1120 HeapTuple tup;
1121 Relation rel;
1122
1123 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
1124
1125 tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
1126
1127 if (!HeapTupleIsValid(tup))
1128 ereport(ERROR,
1129 (errcode(ERRCODE_UNDEFINED_OBJECT),
1130 errmsg("subscription with OID %u does not exist", subid)));
1131
1132 AlterSubscriptionOwner_internal(rel, tup, newOwnerId);
1133
1134 heap_freetuple(tup);
1135
1136 table_close(rel, RowExclusiveLock);
1137}
1138
1139/*
1140 * Get the list of tables which belong to specified publications on the
1141 * publisher connection.
1142 */
1143static List *
1144fetch_table_list(WalReceiverConn *wrconn, List *publications)
1145{
1146 WalRcvExecResult *res;
1147 StringInfoData cmd;
1148 TupleTableSlot *slot;
1149 Oid tableRow[2] = {TEXTOID, TEXTOID};
1150 ListCell *lc;
1151 bool first;
1152 List *tablelist = NIL;
1153
1154 Assert(list_length(publications) > 0);
1155
1156 initStringInfo(&cmd);
1157 appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n"
1158 " FROM pg_catalog.pg_publication_tables t\n"
1159 " WHERE t.pubname IN (");
1160 first = true;
1161 foreach(lc, publications)
1162 {
1163 char *pubname = strVal(lfirst(lc));
1164
1165 if (first)
1166 first = false;
1167 else
1168 appendStringInfoString(&cmd, ", ");
1169
1170 appendStringInfoString(&cmd, quote_literal_cstr(pubname));
1171 }
1172 appendStringInfoChar(&cmd, ')');
1173
1174 res = walrcv_exec(wrconn, cmd.data, 2, tableRow);
1175 pfree(cmd.data);
1176
1177 if (res->status != WALRCV_OK_TUPLES)
1178 ereport(ERROR,
1179 (errmsg("could not receive list of replicated tables from the publisher: %s",
1180 res->err)));
1181
1182 /* Process tables. */
1183 slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
1184 while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
1185 {
1186 char *nspname;
1187 char *relname;
1188 bool isnull;
1189 RangeVar *rv;
1190
1191 nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
1192 Assert(!isnull);
1193 relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
1194 Assert(!isnull);
1195
1196 rv = makeRangeVar(pstrdup(nspname), pstrdup(relname), -1);
1197 tablelist = lappend(tablelist, rv);
1198
1199 ExecClearTuple(slot);
1200 }
1201 ExecDropSingleTupleTableSlot(slot);
1202
1203 walrcv_clear_result(res);
1204
1205 return tablelist;
1206}
1207