Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * pg_subscription.c
4 : : * replication subscriptions
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : * IDENTIFICATION
10 : : * src/backend/catalog/pg_subscription.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres.h"
16 : :
17 : : #include "access/genam.h"
18 : : #include "access/heapam.h"
19 : : #include "access/htup_details.h"
20 : : #include "access/tableam.h"
21 : : #include "catalog/indexing.h"
22 : : #include "catalog/pg_foreign_server.h"
23 : : #include "catalog/pg_subscription.h"
24 : : #include "catalog/pg_subscription_rel.h"
25 : : #include "catalog/pg_type.h"
26 : : #include "foreign/foreign.h"
27 : : #include "miscadmin.h"
28 : : #include "storage/lmgr.h"
29 : : #include "storage/lock.h"
30 : : #include "utils/acl.h"
31 : : #include "utils/array.h"
32 : : #include "utils/builtins.h"
33 : : #include "utils/fmgroids.h"
34 : : #include "utils/lsyscache.h"
35 : : #include "utils/memutils.h"
36 : : #include "utils/pg_lsn.h"
37 : : #include "utils/rel.h"
38 : : #include "utils/syscache.h"
39 : :
40 : : static List *textarray_to_stringlist(ArrayType *textarray);
41 : :
42 : : /*
43 : : * Add a comma-separated list of publication names to the 'dest' string.
44 : : *
45 : : * If quote_literal is true, the returned list can be used to construct an SQL
46 : : * command, thus no translation is applied. Otherwise, the string can be used
47 : : * to create a user-facing message, so translatable quote marks are added.
48 : : */
49 : : void
50 : 546 : GetPublicationsStr(List *publications, StringInfo dest, bool quote_literal)
51 : : {
52 : : ListCell *lc;
53 : 546 : bool first = true;
54 : :
55 : : Assert(publications != NIL);
56 : :
57 [ + - + + : 1400 : foreach(lc, publications)
+ + ]
58 : : {
59 : 854 : char *pubname = strVal(lfirst(lc));
60 : :
61 [ + + ]: 854 : if (quote_literal)
62 : : {
63 [ + + ]: 844 : if (!first)
64 : 307 : appendStringInfoString(dest, ", ");
65 : 844 : appendStringInfoString(dest, quote_literal_cstr(pubname));
66 : : }
67 : : else
68 : : {
69 [ + + ]: 10 : if (first)
70 : 9 : appendStringInfo(dest, _("\"%s\""), pubname);
71 : : else
72 : 1 : appendStringInfo(dest, _(", \"%s\""), pubname);
73 : : }
74 : :
75 : 854 : first = false;
76 : : }
77 : 546 : }
78 : :
79 : : /*
80 : : * Fetch the subscription from the syscache.
81 : : *
82 : : * Callers that need conninfo must call SubscriptionConninfo().
83 : : */
84 : : Subscription *
85 : 1139 : GetSubscription(Oid subid, bool missing_ok)
86 : : {
87 : : HeapTuple tup;
88 : : Subscription *sub;
89 : : Form_pg_subscription subform;
90 : : Datum datum;
91 : : bool isnull;
92 : : MemoryContext cxt;
93 : : MemoryContext oldcxt;
94 : :
95 : 1139 : tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
96 : :
97 [ + + ]: 1139 : if (!HeapTupleIsValid(tup))
98 : : {
99 [ + - ]: 68 : if (missing_ok)
100 : 68 : return NULL;
101 : :
102 [ # # ]: 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
103 : : }
104 : :
105 : 1071 : cxt = AllocSetContextCreate(CurrentMemoryContext, "subscription",
106 : : ALLOCSET_SMALL_SIZES);
107 : 1071 : oldcxt = MemoryContextSwitchTo(cxt);
108 : :
109 : 1071 : subform = (Form_pg_subscription) GETSTRUCT(tup);
110 : :
111 : : /*
112 : : * It's only safe to access subscriptions from another database from the
113 : : * launcher process, which does not call GetSubscription().
114 : : */
115 : : Assert(subform->subdbid == MyDatabaseId);
116 : :
117 : 1071 : sub = palloc0_object(Subscription);
118 : 1071 : sub->cxt = cxt;
119 : 1071 : sub->oid = subid;
120 : 1071 : sub->dbid = subform->subdbid;
121 : 1071 : sub->skiplsn = subform->subskiplsn;
122 : 1071 : sub->name = pstrdup(NameStr(subform->subname));
123 : 1071 : sub->owner = subform->subowner;
124 : 1071 : sub->enabled = subform->subenabled;
125 : 1071 : sub->binary = subform->subbinary;
126 : 1071 : sub->stream = subform->substream;
127 : 1071 : sub->twophasestate = subform->subtwophasestate;
128 : 1071 : sub->disableonerr = subform->subdisableonerr;
129 : 1071 : sub->passwordrequired = subform->subpasswordrequired;
130 : 1071 : sub->runasowner = subform->subrunasowner;
131 : 1071 : sub->failover = subform->subfailover;
132 : 1071 : sub->retaindeadtuples = subform->subretaindeadtuples;
133 : 1071 : sub->maxretention = subform->submaxretention;
134 : 1071 : sub->retentionactive = subform->subretentionactive;
135 : 1071 : sub->conflictlogrelid = subform->subconflictlogrelid;
136 : :
137 : : /* Get slotname */
138 : 1071 : datum = SysCacheGetAttr(SUBSCRIPTIONOID,
139 : : tup,
140 : : Anum_pg_subscription_subslotname,
141 : : &isnull);
142 [ + + ]: 1071 : if (!isnull)
143 : 1027 : sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
144 : : else
145 : 44 : sub->slotname = NULL;
146 : :
147 : : /* Get synccommit */
148 : 1071 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
149 : : tup,
150 : : Anum_pg_subscription_subsynccommit);
151 : 1071 : sub->synccommit = TextDatumGetCString(datum);
152 : :
153 : : /* Get walrcvtimeout */
154 : 1071 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
155 : : tup,
156 : : Anum_pg_subscription_subwalrcvtimeout);
157 : 1071 : sub->walrcvtimeout = TextDatumGetCString(datum);
158 : :
159 : : /* Get publications */
160 : 1071 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
161 : : tup,
162 : : Anum_pg_subscription_subpublications);
163 : 1071 : sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
164 : :
165 : : /* Get origin */
166 : 1071 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
167 : : tup,
168 : : Anum_pg_subscription_suborigin);
169 : 1071 : sub->origin = TextDatumGetCString(datum);
170 : :
171 : : /* Get conflict log destination */
172 : 1071 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
173 : : tup,
174 : : Anum_pg_subscription_subconflictlogdest);
175 : 1071 : sub->conflictlogdest = TextDatumGetCString(datum);
176 : :
177 : : /* Is the subscription owner a superuser? */
178 : 1071 : sub->ownersuperuser = superuser_arg(sub->owner);
179 : :
180 : 1071 : ReleaseSysCache(tup);
181 : :
182 : 1071 : MemoryContextSwitchTo(oldcxt);
183 : :
184 : 1071 : return sub;
185 : : }
186 : :
187 : : /*
188 : : * Generate the connection string for a subscription.
189 : : *
190 : : * This is deliberately separate from GetSubscription() because resolving
191 : : * conninfo for a server-based subscription has its own error paths (foreign
192 : : * server USAGE, user mapping, ForeignServerConnectionString()). Keeping it
193 : : * separate lets a caller load the subscription and decide whether a
194 : : * connection is actually needed, and check things such as whether the
195 : : * subscription is enabled, before risking those errors. Callers that never
196 : : * connect thus never hit them, which matters during restore.
197 : : */
198 : : char *
199 : 720 : SubscriptionConninfo(Subscription *sub)
200 : : {
201 : : HeapTuple tup;
202 : : Form_pg_subscription subform;
203 : : Datum datum;
204 : : char *conninfo;
205 : :
206 : 720 : tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(sub->oid));
207 [ - + ]: 720 : if (!HeapTupleIsValid(tup))
208 [ # # ]: 0 : elog(ERROR, "cache lookup failed for subscription %u", sub->oid);
209 : :
210 : 720 : subform = (Form_pg_subscription) GETSTRUCT(tup);
211 : :
212 [ + + ]: 720 : if (OidIsValid(subform->subserver))
213 : : {
214 : : ForeignServer *server;
215 : : AclResult aclresult;
216 : :
217 : 10 : server = GetForeignServer(subform->subserver);
218 : :
219 : 10 : aclresult = object_aclcheck(ForeignServerRelationId,
220 : : subform->subserver,
221 : : sub->owner, ACL_USAGE);
222 [ + + ]: 10 : if (aclresult != ACLCHECK_OK)
223 [ + - ]: 4 : ereport(ERROR,
224 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
225 : : errmsg("subscription owner \"%s\" does not have permission on foreign server \"%s\"",
226 : : GetUserNameFromId(sub->owner, false),
227 : : server->servername)));
228 : :
229 : 6 : conninfo = ForeignServerConnectionString(sub->owner, server);
230 : : }
231 : : else
232 : : {
233 : 710 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, tup,
234 : : Anum_pg_subscription_subconninfo);
235 : 710 : conninfo = TextDatumGetCString(datum);
236 : : }
237 : :
238 : 716 : ReleaseSysCache(tup);
239 : :
240 : 716 : return conninfo;
241 : : }
242 : :
243 : : /*
244 : : * Return number of subscriptions defined in given database.
245 : : * Used by dropdb() to check if database can indeed be dropped.
246 : : */
247 : : int
248 : 57 : CountDBSubscriptions(Oid dbid)
249 : : {
250 : 57 : int nsubs = 0;
251 : : Relation rel;
252 : : ScanKeyData scankey;
253 : : SysScanDesc scan;
254 : : HeapTuple tup;
255 : :
256 : 57 : rel = table_open(SubscriptionRelationId, RowExclusiveLock);
257 : :
258 : 57 : ScanKeyInit(&scankey,
259 : : Anum_pg_subscription_subdbid,
260 : : BTEqualStrategyNumber, F_OIDEQ,
261 : : ObjectIdGetDatum(dbid));
262 : :
263 : 57 : scan = systable_beginscan(rel, InvalidOid, false,
264 : : NULL, 1, &scankey);
265 : :
266 [ - + ]: 57 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
267 : 0 : nsubs++;
268 : :
269 : 57 : systable_endscan(scan);
270 : :
271 : 57 : table_close(rel, NoLock);
272 : :
273 : 57 : return nsubs;
274 : : }
275 : :
276 : : /*
277 : : * Disable the given subscription.
278 : : */
279 : : void
280 : 4 : DisableSubscription(Oid subid)
281 : : {
282 : : Relation rel;
283 : : bool nulls[Natts_pg_subscription];
284 : : bool replaces[Natts_pg_subscription];
285 : : Datum values[Natts_pg_subscription];
286 : : HeapTuple tup;
287 : :
288 : : /* Look up the subscription in the catalog */
289 : 4 : rel = table_open(SubscriptionRelationId, RowExclusiveLock);
290 : 4 : tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
291 : :
292 [ - + ]: 4 : if (!HeapTupleIsValid(tup))
293 [ # # ]: 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
294 : :
295 : : /* Must only modify subscriptions belonging to the current database. */
296 : : Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
297 : :
298 : 4 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
299 : :
300 : : /* Form a new tuple. */
301 : 4 : memset(values, 0, sizeof(values));
302 : 4 : memset(nulls, false, sizeof(nulls));
303 : 4 : memset(replaces, false, sizeof(replaces));
304 : :
305 : : /* Set the subscription to disabled. */
306 : 4 : values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
307 : 4 : replaces[Anum_pg_subscription_subenabled - 1] = true;
308 : :
309 : : /* Update the catalog */
310 : 4 : tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
311 : : replaces);
312 : 4 : CatalogTupleUpdate(rel, &tup->t_self, tup);
313 : 4 : heap_freetuple(tup);
314 : :
315 : 4 : table_close(rel, NoLock);
316 : 4 : }
317 : :
318 : : /*
319 : : * Convert text array to list of strings.
320 : : *
321 : : * Note: the resulting list of strings is pallocated here.
322 : : */
323 : : static List *
324 : 1071 : textarray_to_stringlist(ArrayType *textarray)
325 : : {
326 : : Datum *elems;
327 : : int nelems,
328 : : i;
329 : 1071 : List *res = NIL;
330 : :
331 : 1071 : deconstruct_array_builtin(textarray, TEXTOID, &elems, NULL, &nelems);
332 : :
333 [ - + ]: 1071 : if (nelems == 0)
334 : 0 : return NIL;
335 : :
336 [ + + ]: 2543 : for (i = 0; i < nelems; i++)
337 : 1472 : res = lappend(res, makeString(TextDatumGetCString(elems[i])));
338 : :
339 : 1071 : return res;
340 : : }
341 : :
342 : : /*
343 : : * Add new state record for a subscription table.
344 : : *
345 : : * If retain_lock is true, then don't release the locks taken in this function.
346 : : * We normally release the locks at the end of transaction but in binary-upgrade
347 : : * mode, we expect to release those immediately.
348 : : */
349 : : void
350 : 228 : AddSubscriptionRelState(Oid subid, Oid relid, char state,
351 : : XLogRecPtr sublsn, bool retain_lock)
352 : : {
353 : : Relation rel;
354 : : HeapTuple tup;
355 : : bool nulls[Natts_pg_subscription_rel];
356 : : Datum values[Natts_pg_subscription_rel];
357 : :
358 : 228 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
359 : :
360 : 228 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
361 : :
362 : : /* Try finding existing mapping. */
363 : 228 : tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
364 : : ObjectIdGetDatum(relid),
365 : : ObjectIdGetDatum(subid));
366 [ - + ]: 228 : if (HeapTupleIsValid(tup))
367 [ # # ]: 0 : elog(ERROR, "subscription relation %u in subscription %u already exists",
368 : : relid, subid);
369 : :
370 : : /* Form the tuple. */
371 : 228 : memset(values, 0, sizeof(values));
372 : 228 : memset(nulls, false, sizeof(nulls));
373 : 228 : values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
374 : 228 : values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
375 : 228 : values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
376 [ + + ]: 228 : if (XLogRecPtrIsValid(sublsn))
377 : 2 : values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
378 : : else
379 : 226 : nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
380 : :
381 : 228 : tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
382 : :
383 : : /* Insert tuple into catalog. */
384 : 228 : CatalogTupleInsert(rel, tup);
385 : :
386 : 228 : heap_freetuple(tup);
387 : :
388 : : /* Cleanup. */
389 [ + + ]: 228 : if (retain_lock)
390 : : {
391 : 225 : table_close(rel, NoLock);
392 : : }
393 : : else
394 : : {
395 : 3 : table_close(rel, RowExclusiveLock);
396 : 3 : UnlockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
397 : : }
398 : 228 : }
399 : :
400 : : /*
401 : : * Update the state of a subscription table.
402 : : */
403 : : void
404 : 858 : UpdateSubscriptionRelState(Oid subid, Oid relid, char state,
405 : : XLogRecPtr sublsn, bool already_locked)
406 : : {
407 : : Relation rel;
408 : : HeapTuple tup;
409 : : bool nulls[Natts_pg_subscription_rel];
410 : : Datum values[Natts_pg_subscription_rel];
411 : : bool replaces[Natts_pg_subscription_rel];
412 : :
413 [ + + ]: 858 : if (already_locked)
414 : : {
415 : : #ifdef USE_ASSERT_CHECKING
416 : : LOCKTAG tag;
417 : :
418 : : Assert(CheckRelationOidLockedByMe(SubscriptionRelRelationId,
419 : : RowExclusiveLock, true));
420 : : SET_LOCKTAG_OBJECT(tag, InvalidOid, SubscriptionRelationId, subid, 0);
421 : : Assert(LockHeldByMe(&tag, AccessShareLock, true));
422 : : #endif
423 : :
424 : 196 : rel = table_open(SubscriptionRelRelationId, NoLock);
425 : : }
426 : : else
427 : : {
428 : 662 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
429 : 661 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
430 : : }
431 : :
432 : : /* Try finding existing mapping. */
433 : 857 : tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
434 : : ObjectIdGetDatum(relid),
435 : : ObjectIdGetDatum(subid));
436 [ - + ]: 857 : if (!HeapTupleIsValid(tup))
437 [ # # ]: 0 : elog(ERROR, "subscription relation %u in subscription %u does not exist",
438 : : relid, subid);
439 : :
440 : : /* Update the tuple. */
441 : 857 : memset(values, 0, sizeof(values));
442 : 857 : memset(nulls, false, sizeof(nulls));
443 : 857 : memset(replaces, false, sizeof(replaces));
444 : :
445 : 857 : replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
446 : 857 : values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
447 : :
448 : 857 : replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
449 [ + + ]: 857 : if (XLogRecPtrIsValid(sublsn))
450 : 421 : values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
451 : : else
452 : 436 : nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
453 : :
454 : 857 : tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
455 : : replaces);
456 : :
457 : : /* Update the catalog. */
458 : 857 : CatalogTupleUpdate(rel, &tup->t_self, tup);
459 : :
460 : : /* Cleanup. */
461 : 857 : table_close(rel, NoLock);
462 : 857 : }
463 : :
464 : : /*
465 : : * Get state of subscription table.
466 : : *
467 : : * Returns SUBREL_STATE_UNKNOWN when the table is not in the subscription.
468 : : */
469 : : char
470 : 4309 : GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn)
471 : : {
472 : : HeapTuple tup;
473 : : char substate;
474 : : bool isnull;
475 : : Datum d;
476 : : Relation rel;
477 : :
478 : : /*
479 : : * This is to avoid the race condition with AlterSubscription which tries
480 : : * to remove this relstate.
481 : : */
482 : 4309 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
483 : :
484 : : /* Try finding the mapping. */
485 : 4309 : tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
486 : : ObjectIdGetDatum(relid),
487 : : ObjectIdGetDatum(subid));
488 : :
489 [ + + ]: 4309 : if (!HeapTupleIsValid(tup))
490 : : {
491 : 33 : table_close(rel, AccessShareLock);
492 : 33 : *sublsn = InvalidXLogRecPtr;
493 : 33 : return SUBREL_STATE_UNKNOWN;
494 : : }
495 : :
496 : : /* Get the state. */
497 : 4276 : substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
498 : :
499 : : /* Get the LSN */
500 : 4276 : d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
501 : : Anum_pg_subscription_rel_srsublsn, &isnull);
502 [ + + ]: 4276 : if (isnull)
503 : 3699 : *sublsn = InvalidXLogRecPtr;
504 : : else
505 : 577 : *sublsn = DatumGetLSN(d);
506 : :
507 : : /* Cleanup */
508 : 4276 : ReleaseSysCache(tup);
509 : :
510 : 4276 : table_close(rel, AccessShareLock);
511 : :
512 : 4276 : return substate;
513 : : }
514 : :
515 : : /*
516 : : * Drop subscription relation mapping. These can be for a particular
517 : : * subscription, or for a particular relation, or both.
518 : : */
519 : : void
520 : 34434 : RemoveSubscriptionRel(Oid subid, Oid relid)
521 : : {
522 : : Relation rel;
523 : : TableScanDesc scan;
524 : : ScanKeyData skey[2];
525 : : HeapTuple tup;
526 : 34434 : int nkeys = 0;
527 : :
528 : 34434 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
529 : :
530 [ + + ]: 34434 : if (OidIsValid(subid))
531 : : {
532 : 189 : ScanKeyInit(&skey[nkeys++],
533 : : Anum_pg_subscription_rel_srsubid,
534 : : BTEqualStrategyNumber,
535 : : F_OIDEQ,
536 : : ObjectIdGetDatum(subid));
537 : : }
538 : :
539 [ + + ]: 34434 : if (OidIsValid(relid))
540 : : {
541 : 34266 : ScanKeyInit(&skey[nkeys++],
542 : : Anum_pg_subscription_rel_srrelid,
543 : : BTEqualStrategyNumber,
544 : : F_OIDEQ,
545 : : ObjectIdGetDatum(relid));
546 : : }
547 : :
548 : : /* Do the search and delete what we found. */
549 : 34434 : scan = table_beginscan_catalog(rel, nkeys, skey);
550 [ + + ]: 34564 : while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
551 : : {
552 : : Form_pg_subscription_rel subrel;
553 : :
554 : 130 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
555 : :
556 : : /*
557 : : * We don't allow to drop the relation mapping when the table
558 : : * synchronization is in progress unless the caller updates the
559 : : * corresponding subscription as well. This is to ensure that we don't
560 : : * leave tablesync slots or origins in the system when the
561 : : * corresponding table is dropped. For sequences, however, it's ok to
562 : : * drop them since no separate slots or origins are created during
563 : : * synchronization.
564 : : */
565 [ + + ]: 130 : if (!OidIsValid(subid) &&
566 [ - + - - ]: 16 : subrel->srsubstate != SUBREL_STATE_READY &&
567 : 0 : get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE)
568 : : {
569 [ # # ]: 0 : ereport(ERROR,
570 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
571 : : errmsg("could not drop relation mapping for subscription \"%s\"",
572 : : get_subscription_name(subrel->srsubid, false)),
573 : : errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
574 : : get_rel_name(relid), subrel->srsubstate),
575 : :
576 : : /*
577 : : * translator: first %s is a SQL ALTER command and second %s is a
578 : : * SQL DROP command
579 : : */
580 : : errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
581 : : "ALTER SUBSCRIPTION ... ENABLE",
582 : : "DROP SUBSCRIPTION ...")));
583 : : }
584 : :
585 : 130 : CatalogTupleDelete(rel, &tup->t_self);
586 : : }
587 : 34434 : table_endscan(scan);
588 : :
589 : 34434 : table_close(rel, RowExclusiveLock);
590 : 34434 : }
591 : :
592 : : /*
593 : : * Does the subscription have any tables?
594 : : *
595 : : * Use this function only to know true/false, and when you have no need for the
596 : : * List returned by GetSubscriptionRelations.
597 : : */
598 : : bool
599 : 315 : HasSubscriptionTables(Oid subid)
600 : : {
601 : : Relation rel;
602 : : ScanKeyData skey[1];
603 : : SysScanDesc scan;
604 : : HeapTuple tup;
605 : 315 : bool has_subtables = false;
606 : :
607 : 315 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
608 : :
609 : 315 : ScanKeyInit(&skey[0],
610 : : Anum_pg_subscription_rel_srsubid,
611 : : BTEqualStrategyNumber, F_OIDEQ,
612 : : ObjectIdGetDatum(subid));
613 : :
614 : 315 : scan = systable_beginscan(rel, InvalidOid, false,
615 : : NULL, 1, skey);
616 : :
617 [ + + ]: 391 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
618 : : {
619 : : Form_pg_subscription_rel subrel;
620 : : char relkind;
621 : :
622 : 367 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
623 : 367 : relkind = get_rel_relkind(subrel->srrelid);
624 : :
625 [ + + + + ]: 367 : if (relkind == RELKIND_RELATION ||
626 : : relkind == RELKIND_PARTITIONED_TABLE)
627 : : {
628 : 291 : has_subtables = true;
629 : 291 : break;
630 : : }
631 : : }
632 : :
633 : : /* Cleanup */
634 : 315 : systable_endscan(scan);
635 : 315 : table_close(rel, AccessShareLock);
636 : :
637 : 315 : return has_subtables;
638 : : }
639 : :
640 : : /*
641 : : * Get the relations for the subscription.
642 : : *
643 : : * If not_ready is true, return only the relations that are not in a ready
644 : : * state, otherwise return all the relations of the subscription. The
645 : : * returned list is palloc'ed in the current memory context.
646 : : */
647 : : List *
648 : 1268 : GetSubscriptionRelations(Oid subid, bool tables, bool sequences,
649 : : bool not_ready)
650 : : {
651 : 1268 : List *res = NIL;
652 : : Relation rel;
653 : : HeapTuple tup;
654 : 1268 : int nkeys = 0;
655 : : ScanKeyData skey[2];
656 : : SysScanDesc scan;
657 : :
658 : : /* One or both of 'tables' and 'sequences' must be true. */
659 : : Assert(tables || sequences);
660 : :
661 : 1268 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
662 : :
663 : 1268 : ScanKeyInit(&skey[nkeys++],
664 : : Anum_pg_subscription_rel_srsubid,
665 : : BTEqualStrategyNumber, F_OIDEQ,
666 : : ObjectIdGetDatum(subid));
667 : :
668 [ + + ]: 1268 : if (not_ready)
669 : 1225 : ScanKeyInit(&skey[nkeys++],
670 : : Anum_pg_subscription_rel_srsubstate,
671 : : BTEqualStrategyNumber, F_CHARNE,
672 : : CharGetDatum(SUBREL_STATE_READY));
673 : :
674 : 1268 : scan = systable_beginscan(rel, InvalidOid, false,
675 : : NULL, nkeys, skey);
676 : :
677 [ + + ]: 3265 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
678 : : {
679 : : Form_pg_subscription_rel subrel;
680 : : SubscriptionRelState *relstate;
681 : : Datum d;
682 : : bool isnull;
683 : : char relkind;
684 : :
685 : 1997 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
686 : :
687 : : /* Relation is either a sequence or a table */
688 : 1997 : relkind = get_rel_relkind(subrel->srrelid);
689 : : Assert(relkind == RELKIND_SEQUENCE || relkind == RELKIND_RELATION ||
690 : : relkind == RELKIND_PARTITIONED_TABLE);
691 : :
692 : : /* Skip sequences if they were not requested */
693 [ + + - + ]: 1997 : if ((relkind == RELKIND_SEQUENCE) && !sequences)
694 : 0 : continue;
695 : :
696 : : /* Skip tables if they were not requested */
697 [ + + + + ]: 1997 : if ((relkind == RELKIND_RELATION ||
698 [ - + ]: 1931 : relkind == RELKIND_PARTITIONED_TABLE) && !tables)
699 : 0 : continue;
700 : :
701 : 1997 : relstate = palloc_object(SubscriptionRelState);
702 : 1997 : relstate->relid = subrel->srrelid;
703 : 1997 : relstate->state = subrel->srsubstate;
704 : 1997 : d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
705 : : Anum_pg_subscription_rel_srsublsn, &isnull);
706 [ + + ]: 1997 : if (isnull)
707 : 1628 : relstate->lsn = InvalidXLogRecPtr;
708 : : else
709 : 369 : relstate->lsn = DatumGetLSN(d);
710 : :
711 : 1997 : res = lappend(res, relstate);
712 : : }
713 : :
714 : : /* Cleanup */
715 : 1268 : systable_endscan(scan);
716 : 1268 : table_close(rel, AccessShareLock);
717 : :
718 : 1268 : return res;
719 : : }
720 : :
721 : : /*
722 : : * Update the dead tuple retention status for the given subscription.
723 : : */
724 : : void
725 : 2 : UpdateDeadTupleRetentionStatus(Oid subid, bool active)
726 : : {
727 : : Relation rel;
728 : : bool nulls[Natts_pg_subscription];
729 : : bool replaces[Natts_pg_subscription];
730 : : Datum values[Natts_pg_subscription];
731 : : HeapTuple tup;
732 : :
733 : : /* Look up the subscription in the catalog */
734 : 2 : rel = table_open(SubscriptionRelationId, RowExclusiveLock);
735 : 2 : tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
736 : :
737 [ - + ]: 2 : if (!HeapTupleIsValid(tup))
738 [ # # ]: 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
739 : :
740 : : /* Must only modify subscriptions belonging to the current database. */
741 : : Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
742 : :
743 : 2 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
744 : :
745 : : /* Form a new tuple. */
746 : 2 : memset(values, 0, sizeof(values));
747 : 2 : memset(nulls, false, sizeof(nulls));
748 : 2 : memset(replaces, false, sizeof(replaces));
749 : :
750 : : /* Set the subscription to disabled. */
751 : 2 : values[Anum_pg_subscription_subretentionactive - 1] = BoolGetDatum(active);
752 : 2 : replaces[Anum_pg_subscription_subretentionactive - 1] = true;
753 : :
754 : : /* Update the catalog */
755 : 2 : tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
756 : : replaces);
757 : 2 : CatalogTupleUpdate(rel, &tup->t_self, tup);
758 : 2 : heap_freetuple(tup);
759 : :
760 : 2 : table_close(rel, NoLock);
761 : 2 : }
|
