Age Owner Branch data TLA 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
638 michael@paquier.xyz 50 :CBC 542 : GetPublicationsStr(List *publications, StringInfo dest, bool quote_literal)
51 : : {
52 : : ListCell *lc;
53 : 542 : bool first = true;
54 : :
55 [ - + ]: 542 : Assert(publications != NIL);
56 : :
57 [ + - + + : 1392 : foreach(lc, publications)
+ + ]
58 : : {
59 : 850 : char *pubname = strVal(lfirst(lc));
60 : :
61 [ + + ]: 850 : if (quote_literal)
62 : : {
44 alvherre@kurilemu.de 63 [ + + ]: 840 : if (!first)
64 : 307 : appendStringInfoString(dest, ", ");
638 michael@paquier.xyz 65 : 840 : appendStringInfoString(dest, quote_literal_cstr(pubname));
66 : : }
67 : : else
68 : : {
44 alvherre@kurilemu.de 69 [ + + ]: 10 : if (first)
70 : 9 : appendStringInfo(dest, _("\"%s\""), pubname);
71 : : else
72 : 1 : appendStringInfo(dest, _(", \"%s\""), pubname);
73 : : }
74 : :
75 : 850 : first = false;
76 : : }
638 michael@paquier.xyz 77 : 542 : }
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 *
38 jdavis@postgresql.or 88 : 1061 : 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 [ + + - + ]: 1061 : Assert(conninfo_needed || !conninfo_aclcheck);
100 : :
3474 peter_e@gmx.net 101 : 1061 : tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
102 : :
103 [ + + ]: 1061 : if (!HeapTupleIsValid(tup))
104 : : {
105 [ + - ]: 65 : if (missing_ok)
106 : 65 : return NULL;
107 : :
3474 peter_e@gmx.net 108 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
109 : : }
110 : :
123 jdavis@postgresql.or 111 :CBC 996 : cxt = AllocSetContextCreate(CurrentMemoryContext, "subscription",
112 : : ALLOCSET_SMALL_SIZES);
113 : 996 : oldcxt = MemoryContextSwitchTo(cxt);
114 : :
3474 peter_e@gmx.net 115 : 996 : 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 : : */
7 jdavis@postgresql.or 121 [ - + ]: 996 : Assert(subform->subdbid == MyDatabaseId);
122 : :
38 123 : 996 : sub = palloc0_object(Subscription);
123 124 : 996 : sub->cxt = cxt;
3474 peter_e@gmx.net 125 : 996 : sub->oid = subid;
126 : 996 : sub->dbid = subform->subdbid;
1570 akapila@postgresql.o 127 : 996 : sub->skiplsn = subform->subskiplsn;
3474 peter_e@gmx.net 128 : 996 : sub->name = pstrdup(NameStr(subform->subname));
129 : 996 : sub->owner = subform->subowner;
130 : 996 : sub->enabled = subform->subenabled;
2198 tgl@sss.pgh.pa.us 131 : 996 : sub->binary = subform->subbinary;
2151 akapila@postgresql.o 132 : 996 : sub->stream = subform->substream;
1837 133 : 996 : sub->twophasestate = subform->subtwophasestate;
1594 134 : 996 : sub->disableonerr = subform->subdisableonerr;
1213 rhaas@postgresql.org 135 : 996 : sub->passwordrequired = subform->subpasswordrequired;
1208 136 : 996 : sub->runasowner = subform->subrunasowner;
907 akapila@postgresql.o 137 : 996 : sub->failover = subform->subfailover;
367 138 : 996 : sub->retaindeadtuples = subform->subretaindeadtuples;
326 139 : 996 : sub->maxretention = subform->submaxretention;
140 : 996 : sub->retentionactive = subform->subretentionactive;
23 akapila@postgresql.o 141 :GNC 996 : sub->conflictlogrelid = subform->subconflictlogrelid;
142 : :
38 jdavis@postgresql.or 143 [ + + ]:CBC 996 : if (conninfo_needed)
144 : : {
145 [ + + ]: 860 : if (OidIsValid(subform->subserver))
146 : : {
147 : : AclResult aclresult;
148 : : ForeignServer *server;
149 : :
150 : 5 : server = GetForeignServer(subform->subserver);
151 : :
152 [ + - ]: 5 : if (conninfo_aclcheck)
153 : : {
154 : : /* recheck ACL if requested */
155 : 5 : aclresult = object_aclcheck(ForeignServerRelationId,
156 : : subform->subserver,
157 : : subform->subowner, ACL_USAGE);
158 : :
159 [ - + ]: 5 : if (aclresult != ACLCHECK_OK)
38 jdavis@postgresql.or 160 [ # # ]:UBC 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 : :
38 jdavis@postgresql.or 167 :CBC 5 : sub->conninfo = ForeignServerConnectionString(subform->subowner,
168 : : server);
169 : : }
170 : : else
171 : : {
172 : 855 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
173 : : tup,
174 : : Anum_pg_subscription_subconninfo);
175 : 855 : sub->conninfo = TextDatumGetCString(datum);
176 : : }
177 : : }
178 : :
179 : : /* Get slotname */
3474 peter_e@gmx.net 180 : 996 : datum = SysCacheGetAttr(SUBSCRIPTIONOID,
181 : : tup,
182 : : Anum_pg_subscription_subslotname,
183 : : &isnull);
3364 184 [ + + ]: 996 : if (!isnull)
185 : 952 : sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
186 : : else
187 : 44 : sub->slotname = NULL;
188 : :
189 : : /* Get synccommit */
1218 dgustafsson@postgres 190 : 996 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
191 : : tup,
192 : : Anum_pg_subscription_subsynccommit);
3389 peter_e@gmx.net 193 : 996 : sub->synccommit = TextDatumGetCString(datum);
194 : :
195 : : /* Get walrcvtimeout */
155 fujii@postgresql.org 196 : 996 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
197 : : tup,
198 : : Anum_pg_subscription_subwalrcvtimeout);
199 : 996 : sub->walrcvtimeout = TextDatumGetCString(datum);
200 : :
201 : : /* Get publications */
1218 dgustafsson@postgres 202 : 996 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
203 : : tup,
204 : : Anum_pg_subscription_subpublications);
3474 peter_e@gmx.net 205 : 996 : sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
206 : :
207 : : /* Get origin */
1218 dgustafsson@postgres 208 : 996 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
209 : : tup,
210 : : Anum_pg_subscription_suborigin);
1465 akapila@postgresql.o 211 : 996 : sub->origin = TextDatumGetCString(datum);
212 : :
213 : : /* Get conflict log destination */
23 akapila@postgresql.o 214 :GNC 996 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
215 : : tup,
216 : : Anum_pg_subscription_subconflictlogdest);
217 : 996 : sub->conflictlogdest = TextDatumGetCString(datum);
218 : :
219 : : /* Is the subscription owner a superuser? */
1012 akapila@postgresql.o 220 :CBC 996 : sub->ownersuperuser = superuser_arg(sub->owner);
221 : :
3474 peter_e@gmx.net 222 : 996 : ReleaseSysCache(tup);
223 : :
123 jdavis@postgresql.or 224 : 996 : MemoryContextSwitchTo(oldcxt);
225 : :
3474 peter_e@gmx.net 226 : 996 : 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 : 50 : CountDBSubscriptions(Oid dbid)
235 : : {
3356 bruce@momjian.us 236 : 50 : int nsubs = 0;
237 : : Relation rel;
238 : : ScanKeyData scankey;
239 : : SysScanDesc scan;
240 : : HeapTuple tup;
241 : :
2742 andres@anarazel.de 242 : 50 : rel = table_open(SubscriptionRelationId, RowExclusiveLock);
243 : :
3474 peter_e@gmx.net 244 : 50 : ScanKeyInit(&scankey,
245 : : Anum_pg_subscription_subdbid,
246 : : BTEqualStrategyNumber, F_OIDEQ,
247 : : ObjectIdGetDatum(dbid));
248 : :
249 : 50 : scan = systable_beginscan(rel, InvalidOid, false,
250 : : NULL, 1, &scankey);
251 : :
252 [ - + ]: 50 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
3474 peter_e@gmx.net 253 :UBC 0 : nsubs++;
254 : :
3474 peter_e@gmx.net 255 :CBC 50 : systable_endscan(scan);
256 : :
2742 andres@anarazel.de 257 : 50 : table_close(rel, NoLock);
258 : :
3474 peter_e@gmx.net 259 : 50 : return nsubs;
260 : : }
261 : :
262 : : /*
263 : : * Disable the given subscription.
264 : : */
265 : : void
1594 akapila@postgresql.o 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))
1594 akapila@postgresql.o 279 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
280 : :
281 : : /* Must only modify subscriptions belonging to the current database. */
7 jdavis@postgresql.or 282 [ - + ]:CBC 4 : Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
283 : :
1594 akapila@postgresql.o 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 *
3474 peter_e@gmx.net 310 : 996 : textarray_to_stringlist(ArrayType *textarray)
311 : : {
312 : : Datum *elems;
313 : : int nelems,
314 : : i;
3356 bruce@momjian.us 315 : 996 : List *res = NIL;
316 : :
1485 peter@eisentraut.org 317 : 996 : deconstruct_array_builtin(textarray, TEXTOID, &elems, NULL, &nelems);
318 : :
3474 peter_e@gmx.net 319 [ - + ]: 996 : if (nelems == 0)
3474 peter_e@gmx.net 320 :UBC 0 : return NIL;
321 : :
3474 peter_e@gmx.net 322 [ + + ]:CBC 2394 : for (i = 0; i < nelems; i++)
3389 323 : 1398 : res = lappend(res, makeString(TextDatumGetCString(elems[i])));
324 : :
3474 325 : 996 : 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
3032 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 : :
3309 344 : 228 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
345 : :
2742 andres@anarazel.de 346 : 228 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
347 : :
348 : : /* Try finding existing mapping. */
3411 peter_e@gmx.net 349 : 228 : tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
350 : : ObjectIdGetDatum(relid),
351 : : ObjectIdGetDatum(subid));
3032 352 [ - + ]: 228 : if (HeapTupleIsValid(tup))
275 akapila@postgresql.o 353 [ # # ]:UBC 0 : elog(ERROR, "subscription relation %u in subscription %u already exists",
354 : : relid, subid);
355 : :
356 : : /* Form the tuple. */
3032 peter_e@gmx.net 357 :CBC 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);
261 alvherre@kurilemu.de 362 [ + + ]: 228 : if (XLogRecPtrIsValid(sublsn))
3032 peter_e@gmx.net 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. */
2804 andres@anarazel.de 370 : 228 : CatalogTupleInsert(rel, tup);
371 : :
3032 peter_e@gmx.net 372 : 228 : heap_freetuple(tup);
373 : :
374 : : /* Cleanup. */
935 akapila@postgresql.o 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 : : }
3032 peter_e@gmx.net 384 : 228 : }
385 : :
386 : : /*
387 : : * Update the state of a subscription table.
388 : : */
389 : : void
390 : 848 : 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 : :
358 akapila@postgresql.o 399 [ + + ]: 848 : if (already_locked)
400 : : {
401 : : #ifdef USE_ASSERT_CHECKING
402 : : LOCKTAG tag;
403 : :
404 [ - + ]: 194 : Assert(CheckRelationOidLockedByMe(SubscriptionRelRelationId,
405 : : RowExclusiveLock, true));
406 : 194 : SET_LOCKTAG_OBJECT(tag, InvalidOid, SubscriptionRelationId, subid, 0);
407 [ - + ]: 194 : Assert(LockHeldByMe(&tag, AccessShareLock, true));
408 : : #endif
409 : :
410 : 194 : rel = table_open(SubscriptionRelRelationId, NoLock);
411 : : }
412 : : else
413 : : {
414 : 654 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
415 : 653 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
416 : : }
417 : :
418 : : /* Try finding existing mapping. */
3032 peter_e@gmx.net 419 : 847 : tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
420 : : ObjectIdGetDatum(relid),
421 : : ObjectIdGetDatum(subid));
422 [ - + ]: 847 : if (!HeapTupleIsValid(tup))
262 akapila@postgresql.o 423 [ # # ]:UBC 0 : elog(ERROR, "subscription relation %u in subscription %u does not exist",
424 : : relid, subid);
425 : :
426 : : /* Update the tuple. */
3032 peter_e@gmx.net 427 :CBC 847 : memset(values, 0, sizeof(values));
428 : 847 : memset(nulls, false, sizeof(nulls));
429 : 847 : memset(replaces, false, sizeof(replaces));
430 : :
431 : 847 : replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
432 : 847 : values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
433 : :
434 : 847 : replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
261 alvherre@kurilemu.de 435 [ + + ]: 847 : if (XLogRecPtrIsValid(sublsn))
3032 peter_e@gmx.net 436 : 417 : values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
437 : : else
438 : 430 : nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
439 : :
440 : 847 : tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
441 : : replaces);
442 : :
443 : : /* Update the catalog. */
444 : 847 : CatalogTupleUpdate(rel, &tup->t_self, tup);
445 : :
446 : : /* Cleanup. */
2742 andres@anarazel.de 447 : 847 : table_close(rel, NoLock);
3411 peter_e@gmx.net 448 : 847 : }
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
2109 alvherre@alvh.no-ip. 456 : 1314 : 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 : : */
1989 akapila@postgresql.o 468 : 1314 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
469 : :
470 : : /* Try finding the mapping. */
3411 peter_e@gmx.net 471 : 1314 : tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
472 : : ObjectIdGetDatum(relid),
473 : : ObjectIdGetDatum(subid));
474 : :
475 [ + + ]: 1314 : if (!HeapTupleIsValid(tup))
476 : : {
1976 akapila@postgresql.o 477 : 32 : table_close(rel, AccessShareLock);
2109 alvherre@alvh.no-ip. 478 : 32 : *sublsn = InvalidXLogRecPtr;
479 : 32 : return SUBREL_STATE_UNKNOWN;
480 : : }
481 : :
482 : : /* Get the state. */
483 : 1282 : substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
484 : :
485 : : /* Get the LSN */
3411 peter_e@gmx.net 486 : 1282 : d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
487 : : Anum_pg_subscription_rel_srsublsn, &isnull);
488 [ + + ]: 1282 : if (isnull)
489 : 694 : *sublsn = InvalidXLogRecPtr;
490 : : else
491 : 588 : *sublsn = DatumGetLSN(d);
492 : :
493 : : /* Cleanup */
494 : 1282 : ReleaseSysCache(tup);
495 : :
1989 akapila@postgresql.o 496 : 1282 : table_close(rel, AccessShareLock);
497 : :
3411 peter_e@gmx.net 498 : 1282 : 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 : 34269 : RemoveSubscriptionRel(Oid subid, Oid relid)
507 : : {
508 : : Relation rel;
509 : : TableScanDesc scan;
510 : : ScanKeyData skey[2];
511 : : HeapTuple tup;
512 : 34269 : int nkeys = 0;
513 : :
2742 andres@anarazel.de 514 : 34269 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
515 : :
3411 peter_e@gmx.net 516 [ + + ]: 34269 : 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 [ + + ]: 34269 : if (OidIsValid(relid))
526 : : {
527 : 34101 : 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. */
2693 andres@anarazel.de 535 : 34269 : scan = table_beginscan_catalog(rel, nkeys, skey);
3411 peter_e@gmx.net 536 [ + + ]: 34399 : while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
537 : : {
538 : : Form_pg_subscription_rel subrel;
539 : :
1989 akapila@postgresql.o 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 : : */
275 551 [ + + ]: 130 : if (!OidIsValid(subid) &&
552 [ - + - - ]: 16 : subrel->srsubstate != SUBREL_STATE_READY &&
275 akapila@postgresql.o 553 :UBC 0 : get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE)
554 : : {
1989 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 : :
3328 tgl@sss.pgh.pa.us 571 :CBC 130 : CatalogTupleDelete(rel, &tup->t_self);
572 : : }
2693 andres@anarazel.de 573 : 34269 : table_endscan(scan);
574 : :
2742 575 : 34269 : table_close(rel, RowExclusiveLock);
3411 peter_e@gmx.net 576 : 34269 : }
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
282 akapila@postgresql.o 585 : 295 : HasSubscriptionTables(Oid subid)
586 : : {
587 : : Relation rel;
588 : : ScanKeyData skey[1];
589 : : SysScanDesc scan;
590 : : HeapTuple tup;
275 591 : 295 : bool has_subtables = false;
592 : :
1837 593 : 295 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
594 : :
595 : 295 : ScanKeyInit(&skey[0],
596 : : Anum_pg_subscription_rel_srsubid,
597 : : BTEqualStrategyNumber, F_OIDEQ,
598 : : ObjectIdGetDatum(subid));
599 : :
600 : 295 : scan = systable_beginscan(rel, InvalidOid, false,
601 : : NULL, 1, skey);
602 : :
275 603 [ + + ]: 371 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
604 : : {
605 : : Form_pg_subscription_rel subrel;
606 : : char relkind;
607 : :
608 : 347 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
609 : 347 : relkind = get_rel_relkind(subrel->srrelid);
610 : :
611 [ + + + + ]: 347 : if (relkind == RELKIND_RELATION ||
612 : : relkind == RELKIND_PARTITIONED_TABLE)
613 : : {
614 : 271 : has_subtables = true;
615 : 271 : break;
616 : : }
617 : : }
618 : :
619 : : /* Cleanup */
1837 620 : 295 : systable_endscan(scan);
621 : 295 : table_close(rel, AccessShareLock);
622 : :
275 623 : 295 : 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 : 1257 : GetSubscriptionRelations(Oid subid, bool tables, bool sequences,
635 : : bool not_ready)
636 : : {
3411 peter_e@gmx.net 637 : 1257 : List *res = NIL;
638 : : Relation rel;
639 : : HeapTuple tup;
640 : 1257 : int nkeys = 0;
641 : : ScanKeyData skey[2];
642 : : SysScanDesc scan;
643 : :
644 : : /* One or both of 'tables' and 'sequences' must be true. */
275 akapila@postgresql.o 645 [ + + - + ]: 1257 : Assert(tables || sequences);
646 : :
2742 andres@anarazel.de 647 : 1257 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
648 : :
3411 peter_e@gmx.net 649 : 1257 : ScanKeyInit(&skey[nkeys++],
650 : : Anum_pg_subscription_rel_srsubid,
651 : : BTEqualStrategyNumber, F_OIDEQ,
652 : : ObjectIdGetDatum(subid));
653 : :
1459 michael@paquier.xyz 654 [ + + ]: 1257 : if (not_ready)
655 : 1214 : ScanKeyInit(&skey[nkeys++],
656 : : Anum_pg_subscription_rel_srsubstate,
657 : : BTEqualStrategyNumber, F_CHARNE,
658 : : CharGetDatum(SUBREL_STATE_READY));
659 : :
3411 peter_e@gmx.net 660 : 1257 : scan = systable_beginscan(rel, InvalidOid, false,
661 : : NULL, nkeys, skey);
662 : :
663 [ + + ]: 3285 : 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 : 2028 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
672 : :
673 : : /* Relation is either a sequence or a table */
275 akapila@postgresql.o 674 : 2028 : relkind = get_rel_relkind(subrel->srrelid);
675 [ + + + + : 2028 : Assert(relkind == RELKIND_SEQUENCE || relkind == RELKIND_RELATION ||
- + ]
676 : : relkind == RELKIND_PARTITIONED_TABLE);
677 : :
678 : : /* Skip sequences if they were not requested */
679 [ + + - + ]: 2028 : if ((relkind == RELKIND_SEQUENCE) && !sequences)
275 akapila@postgresql.o 680 :UBC 0 : continue;
681 : :
682 : : /* Skip tables if they were not requested */
275 akapila@postgresql.o 683 [ + + + + ]:CBC 2028 : if ((relkind == RELKIND_RELATION ||
684 [ - + ]: 1961 : relkind == RELKIND_PARTITIONED_TABLE) && !tables)
275 akapila@postgresql.o 685 :UBC 0 : continue;
686 : :
227 michael@paquier.xyz 687 :CBC 2028 : relstate = palloc_object(SubscriptionRelState);
3411 peter_e@gmx.net 688 : 2028 : relstate->relid = subrel->srrelid;
689 : 2028 : relstate->state = subrel->srsubstate;
2196 tgl@sss.pgh.pa.us 690 : 2028 : d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
691 : : Anum_pg_subscription_rel_srsublsn, &isnull);
692 [ + + ]: 2028 : if (isnull)
693 : 1658 : relstate->lsn = InvalidXLogRecPtr;
694 : : else
695 : 370 : relstate->lsn = DatumGetLSN(d);
696 : :
3411 peter_e@gmx.net 697 : 2028 : res = lappend(res, relstate);
698 : : }
699 : :
700 : : /* Cleanup */
701 : 1257 : systable_endscan(scan);
2742 andres@anarazel.de 702 : 1257 : table_close(rel, AccessShareLock);
703 : :
3411 peter_e@gmx.net 704 : 1257 : return res;
705 : : }
706 : :
707 : : /*
708 : : * Update the dead tuple retention status for the given subscription.
709 : : */
710 : : void
326 akapila@postgresql.o 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))
326 akapila@postgresql.o 724 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
725 : :
726 : : /* Must only modify subscriptions belonging to the current database. */
7 jdavis@postgresql.or 727 [ - + ]:CBC 2 : Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
728 : :
326 akapila@postgresql.o 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. */
96 peter@eisentraut.org 737 : 2 : values[Anum_pg_subscription_subretentionactive - 1] = BoolGetDatum(active);
326 akapila@postgresql.o 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 : }
|
