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