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
671 michael@paquier.xyz 50 :CBC 549 : GetPublicationsStr(List *publications, StringInfo dest, bool quote_literal)
51 : : {
52 : : ListCell *lc;
53 : 549 : bool first = true;
54 : :
55 [ - + ]: 549 : Assert(publications != NIL);
56 : :
57 [ + - + + : 1411 : foreach(lc, publications)
+ + ]
58 : : {
59 : 862 : char *pubname = strVal(lfirst(lc));
60 : :
61 [ + + ]: 862 : if (quote_literal)
62 : : {
77 alvherre@kurilemu.de 63 [ + + ]: 852 : if (!first)
64 : 312 : appendStringInfoString(dest, ", ");
671 michael@paquier.xyz 65 : 852 : appendStringInfoString(dest, quote_literal_cstr(pubname));
66 : : }
67 : : else
68 : : {
77 alvherre@kurilemu.de 69 [ + + ]: 10 : if (first)
70 : 9 : appendStringInfo(dest, _("\"%s\""), pubname);
71 : : else
72 : 1 : appendStringInfo(dest, _(", \"%s\""), pubname);
73 : : }
74 : :
75 : 862 : first = false;
76 : : }
671 michael@paquier.xyz 77 : 549 : }
78 : :
79 : : /*
80 : : * Fetch the subscription from the syscache.
81 : : *
82 : : * Callers that need conninfo must call SubscriptionConninfo().
83 : : */
84 : : Subscription *
22 jdavis@postgresql.or 85 : 1114 : 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 : :
3507 peter_e@gmx.net 95 : 1114 : tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
96 : :
97 [ + + ]: 1114 : if (!HeapTupleIsValid(tup))
98 : : {
99 [ + - ]: 67 : if (missing_ok)
100 : 67 : return NULL;
101 : :
3507 peter_e@gmx.net 102 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
103 : : }
104 : :
156 jdavis@postgresql.or 105 :CBC 1047 : cxt = AllocSetContextCreate(CurrentMemoryContext, "subscription",
106 : : ALLOCSET_SMALL_SIZES);
107 : 1047 : oldcxt = MemoryContextSwitchTo(cxt);
108 : :
3507 peter_e@gmx.net 109 : 1047 : 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 : : */
40 jdavis@postgresql.or 115 [ - + ]: 1047 : Assert(subform->subdbid == MyDatabaseId);
116 : :
71 117 : 1047 : sub = palloc0_object(Subscription);
156 118 : 1047 : sub->cxt = cxt;
3507 peter_e@gmx.net 119 : 1047 : sub->oid = subid;
120 : 1047 : sub->dbid = subform->subdbid;
1603 akapila@postgresql.o 121 : 1047 : sub->skiplsn = subform->subskiplsn;
3507 peter_e@gmx.net 122 : 1047 : sub->name = pstrdup(NameStr(subform->subname));
123 : 1047 : sub->owner = subform->subowner;
124 : 1047 : sub->enabled = subform->subenabled;
2231 tgl@sss.pgh.pa.us 125 : 1047 : sub->binary = subform->subbinary;
2184 akapila@postgresql.o 126 : 1047 : sub->stream = subform->substream;
1870 127 : 1047 : sub->twophasestate = subform->subtwophasestate;
1627 128 : 1047 : sub->disableonerr = subform->subdisableonerr;
1246 rhaas@postgresql.org 129 : 1047 : sub->passwordrequired = subform->subpasswordrequired;
1241 130 : 1047 : sub->runasowner = subform->subrunasowner;
940 akapila@postgresql.o 131 : 1047 : sub->failover = subform->subfailover;
400 132 : 1047 : sub->retaindeadtuples = subform->subretaindeadtuples;
359 133 : 1047 : sub->maxretention = subform->submaxretention;
134 : 1047 : sub->retentionactive = subform->subretentionactive;
56 akapila@postgresql.o 135 :GNC 1047 : sub->conflictlogrelid = subform->subconflictlogrelid;
136 : :
137 : : /* Get slotname */
3507 peter_e@gmx.net 138 :CBC 1047 : datum = SysCacheGetAttr(SUBSCRIPTIONOID,
139 : : tup,
140 : : Anum_pg_subscription_subslotname,
141 : : &isnull);
3397 142 [ + + ]: 1047 : if (!isnull)
143 : 1003 : sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
144 : : else
145 : 44 : sub->slotname = NULL;
146 : :
147 : : /* Get synccommit */
1251 dgustafsson@postgres 148 : 1047 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
149 : : tup,
150 : : Anum_pg_subscription_subsynccommit);
3422 peter_e@gmx.net 151 : 1047 : sub->synccommit = TextDatumGetCString(datum);
152 : :
153 : : /* Get walrcvtimeout */
188 fujii@postgresql.org 154 : 1047 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
155 : : tup,
156 : : Anum_pg_subscription_subwalrcvtimeout);
157 : 1047 : sub->walrcvtimeout = TextDatumGetCString(datum);
158 : :
159 : : /* Get publications */
1251 dgustafsson@postgres 160 : 1047 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
161 : : tup,
162 : : Anum_pg_subscription_subpublications);
3507 peter_e@gmx.net 163 : 1047 : sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
164 : :
165 : : /* Get origin */
1251 dgustafsson@postgres 166 : 1047 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
167 : : tup,
168 : : Anum_pg_subscription_suborigin);
1498 akapila@postgresql.o 169 : 1047 : sub->origin = TextDatumGetCString(datum);
170 : :
171 : : /* Get conflict log destination */
56 akapila@postgresql.o 172 :GNC 1047 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
173 : : tup,
174 : : Anum_pg_subscription_subconflictlogdest);
175 : 1047 : sub->conflictlogdest = TextDatumGetCString(datum);
176 : :
177 : : /* Is the subscription owner a superuser? */
1045 akapila@postgresql.o 178 :CBC 1047 : sub->ownersuperuser = superuser_arg(sub->owner);
179 : :
3507 peter_e@gmx.net 180 : 1047 : ReleaseSysCache(tup);
181 : :
156 jdavis@postgresql.or 182 : 1047 : MemoryContextSwitchTo(oldcxt);
183 : :
3507 peter_e@gmx.net 184 : 1047 : 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 *
22 jdavis@postgresql.or 199 : 696 : SubscriptionConninfo(Subscription *sub)
200 : : {
201 : : HeapTuple tup;
202 : : Form_pg_subscription subform;
203 : : Datum datum;
204 : : char *conninfo;
205 : :
206 : 696 : tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(sub->oid));
207 [ - + ]: 696 : if (!HeapTupleIsValid(tup))
22 jdavis@postgresql.or 208 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for subscription %u", sub->oid);
209 : :
22 jdavis@postgresql.or 210 :CBC 696 : subform = (Form_pg_subscription) GETSTRUCT(tup);
211 : :
212 [ + + ]: 696 : if (OidIsValid(subform->subserver))
213 : : {
214 : : ForeignServer *server;
215 : : AclResult aclresult;
216 : :
217 : 9 : server = GetForeignServer(subform->subserver);
218 : :
219 : 9 : aclresult = object_aclcheck(ForeignServerRelationId,
220 : : subform->subserver,
221 : : sub->owner, ACL_USAGE);
222 [ + + ]: 9 : 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 : 5 : conninfo = ForeignServerConnectionString(sub->owner, server);
230 : : }
231 : : else
232 : : {
233 : 687 : datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID, tup,
234 : : Anum_pg_subscription_subconninfo);
235 : 687 : conninfo = TextDatumGetCString(datum);
236 : : }
237 : :
238 : 692 : ReleaseSysCache(tup);
239 : :
240 : 692 : 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
3507 peter_e@gmx.net 248 : 56 : CountDBSubscriptions(Oid dbid)
249 : : {
3389 bruce@momjian.us 250 : 56 : int nsubs = 0;
251 : : Relation rel;
252 : : ScanKeyData scankey;
253 : : SysScanDesc scan;
254 : : HeapTuple tup;
255 : :
2775 andres@anarazel.de 256 : 56 : rel = table_open(SubscriptionRelationId, RowExclusiveLock);
257 : :
3507 peter_e@gmx.net 258 : 56 : ScanKeyInit(&scankey,
259 : : Anum_pg_subscription_subdbid,
260 : : BTEqualStrategyNumber, F_OIDEQ,
261 : : ObjectIdGetDatum(dbid));
262 : :
263 : 56 : scan = systable_beginscan(rel, InvalidOid, false,
264 : : NULL, 1, &scankey);
265 : :
266 [ - + ]: 56 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
3507 peter_e@gmx.net 267 :UBC 0 : nsubs++;
268 : :
3507 peter_e@gmx.net 269 :CBC 56 : systable_endscan(scan);
270 : :
2775 andres@anarazel.de 271 : 56 : table_close(rel, NoLock);
272 : :
3507 peter_e@gmx.net 273 : 56 : return nsubs;
274 : : }
275 : :
276 : : /*
277 : : * Disable the given subscription.
278 : : */
279 : : void
1627 akapila@postgresql.o 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))
1627 akapila@postgresql.o 293 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
294 : :
295 : : /* Must only modify subscriptions belonging to the current database. */
40 jdavis@postgresql.or 296 [ - + ]:CBC 4 : Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
297 : :
1627 akapila@postgresql.o 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 *
3507 peter_e@gmx.net 324 : 1047 : textarray_to_stringlist(ArrayType *textarray)
325 : : {
326 : : Datum *elems;
327 : : int nelems,
328 : : i;
3389 bruce@momjian.us 329 : 1047 : List *res = NIL;
330 : :
1518 peter@eisentraut.org 331 : 1047 : deconstruct_array_builtin(textarray, TEXTOID, &elems, NULL, &nelems);
332 : :
3507 peter_e@gmx.net 333 [ - + ]: 1047 : if (nelems == 0)
3507 peter_e@gmx.net 334 :UBC 0 : return NIL;
335 : :
3507 peter_e@gmx.net 336 [ + + ]:CBC 2498 : for (i = 0; i < nelems; i++)
3422 337 : 1451 : res = lappend(res, makeString(TextDatumGetCString(elems[i])));
338 : :
3507 339 : 1047 : 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
3065 350 : 231 : 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 : :
3342 358 : 231 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
359 : :
2775 andres@anarazel.de 360 : 231 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
361 : :
362 : : /* Try finding existing mapping. */
3444 peter_e@gmx.net 363 : 231 : tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
364 : : ObjectIdGetDatum(relid),
365 : : ObjectIdGetDatum(subid));
3065 366 [ - + ]: 231 : if (HeapTupleIsValid(tup))
308 akapila@postgresql.o 367 [ # # ]:UBC 0 : elog(ERROR, "subscription relation %u in subscription %u already exists",
368 : : relid, subid);
369 : :
370 : : /* Form the tuple. */
3065 peter_e@gmx.net 371 :CBC 231 : memset(values, 0, sizeof(values));
372 : 231 : memset(nulls, false, sizeof(nulls));
373 : 231 : values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
374 : 231 : values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
375 : 231 : values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
294 alvherre@kurilemu.de 376 [ + + ]: 231 : if (XLogRecPtrIsValid(sublsn))
3065 peter_e@gmx.net 377 : 2 : values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
378 : : else
379 : 229 : nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
380 : :
381 : 231 : tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
382 : :
383 : : /* Insert tuple into catalog. */
2837 andres@anarazel.de 384 : 231 : CatalogTupleInsert(rel, tup);
385 : :
3065 peter_e@gmx.net 386 : 231 : heap_freetuple(tup);
387 : :
388 : : /* Cleanup. */
968 akapila@postgresql.o 389 [ + + ]: 231 : if (retain_lock)
390 : : {
391 : 228 : table_close(rel, NoLock);
392 : : }
393 : : else
394 : : {
395 : 3 : table_close(rel, RowExclusiveLock);
396 : 3 : UnlockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
397 : : }
3065 peter_e@gmx.net 398 : 231 : }
399 : :
400 : : /*
401 : : * Update the state of a subscription table.
402 : : */
403 : : void
404 : 851 : 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 : :
391 akapila@postgresql.o 413 [ + + ]: 851 : if (already_locked)
414 : : {
415 : : #ifdef USE_ASSERT_CHECKING
416 : : LOCKTAG tag;
417 : :
418 [ - + ]: 195 : Assert(CheckRelationOidLockedByMe(SubscriptionRelRelationId,
419 : : RowExclusiveLock, true));
420 : 195 : SET_LOCKTAG_OBJECT(tag, InvalidOid, SubscriptionRelationId, subid, 0);
421 [ - + ]: 195 : Assert(LockHeldByMe(&tag, AccessShareLock, true));
422 : : #endif
423 : :
424 : 195 : rel = table_open(SubscriptionRelRelationId, NoLock);
425 : : }
426 : : else
427 : : {
428 : 656 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
429 : 655 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
430 : : }
431 : :
432 : : /* Try finding existing mapping. */
3065 peter_e@gmx.net 433 : 850 : tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
434 : : ObjectIdGetDatum(relid),
435 : : ObjectIdGetDatum(subid));
436 [ - + ]: 850 : if (!HeapTupleIsValid(tup))
295 akapila@postgresql.o 437 [ # # ]:UBC 0 : elog(ERROR, "subscription relation %u in subscription %u does not exist",
438 : : relid, subid);
439 : :
440 : : /* Update the tuple. */
3065 peter_e@gmx.net 441 :CBC 850 : memset(values, 0, sizeof(values));
442 : 850 : memset(nulls, false, sizeof(nulls));
443 : 850 : memset(replaces, false, sizeof(replaces));
444 : :
445 : 850 : replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
446 : 850 : values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
447 : :
448 : 850 : replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
294 alvherre@kurilemu.de 449 [ + + ]: 850 : if (XLogRecPtrIsValid(sublsn))
3065 peter_e@gmx.net 450 : 419 : values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
451 : : else
452 : 431 : nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
453 : :
454 : 850 : tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
455 : : replaces);
456 : :
457 : : /* Update the catalog. */
458 : 850 : CatalogTupleUpdate(rel, &tup->t_self, tup);
459 : :
460 : : /* Cleanup. */
2775 andres@anarazel.de 461 : 850 : table_close(rel, NoLock);
3444 peter_e@gmx.net 462 : 850 : }
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
2142 alvherre@alvh.no-ip. 470 : 1299 : 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 : : */
2022 akapila@postgresql.o 482 : 1299 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
483 : :
484 : : /* Try finding the mapping. */
3444 peter_e@gmx.net 485 : 1299 : tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
486 : : ObjectIdGetDatum(relid),
487 : : ObjectIdGetDatum(subid));
488 : :
489 [ + + ]: 1299 : if (!HeapTupleIsValid(tup))
490 : : {
2009 akapila@postgresql.o 491 : 34 : table_close(rel, AccessShareLock);
2142 alvherre@alvh.no-ip. 492 : 34 : *sublsn = InvalidXLogRecPtr;
493 : 34 : return SUBREL_STATE_UNKNOWN;
494 : : }
495 : :
496 : : /* Get the state. */
497 : 1265 : substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
498 : :
499 : : /* Get the LSN */
3444 peter_e@gmx.net 500 : 1265 : d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
501 : : Anum_pg_subscription_rel_srsublsn, &isnull);
502 [ + + ]: 1265 : if (isnull)
503 : 683 : *sublsn = InvalidXLogRecPtr;
504 : : else
505 : 582 : *sublsn = DatumGetLSN(d);
506 : :
507 : : /* Cleanup */
508 : 1265 : ReleaseSysCache(tup);
509 : :
2022 akapila@postgresql.o 510 : 1265 : table_close(rel, AccessShareLock);
511 : :
3444 peter_e@gmx.net 512 : 1265 : 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 : 32844 : RemoveSubscriptionRel(Oid subid, Oid relid)
521 : : {
522 : : Relation rel;
523 : : TableScanDesc scan;
524 : : ScanKeyData skey[2];
525 : : HeapTuple tup;
526 : 32844 : int nkeys = 0;
527 : :
2775 andres@anarazel.de 528 : 32844 : rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
529 : :
3444 peter_e@gmx.net 530 [ + + ]: 32844 : if (OidIsValid(subid))
531 : : {
532 : 193 : ScanKeyInit(&skey[nkeys++],
533 : : Anum_pg_subscription_rel_srsubid,
534 : : BTEqualStrategyNumber,
535 : : F_OIDEQ,
536 : : ObjectIdGetDatum(subid));
537 : : }
538 : :
539 [ + + ]: 32844 : if (OidIsValid(relid))
540 : : {
541 : 32674 : 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. */
2726 andres@anarazel.de 549 : 32844 : scan = table_beginscan_catalog(rel, nkeys, skey);
3444 peter_e@gmx.net 550 [ + + ]: 32977 : while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
551 : : {
552 : : Form_pg_subscription_rel subrel;
553 : :
2022 akapila@postgresql.o 554 : 133 : 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 : : */
308 565 [ + + ]: 133 : if (!OidIsValid(subid) &&
566 [ - + - - ]: 18 : subrel->srsubstate != SUBREL_STATE_READY &&
308 akapila@postgresql.o 567 :UBC 0 : get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE)
568 : : {
2022 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 : :
3361 tgl@sss.pgh.pa.us 585 :CBC 133 : CatalogTupleDelete(rel, &tup->t_self);
586 : : }
2726 andres@anarazel.de 587 : 32844 : table_endscan(scan);
588 : :
2775 589 : 32844 : table_close(rel, RowExclusiveLock);
3444 peter_e@gmx.net 590 : 32844 : }
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
315 akapila@postgresql.o 599 : 315 : HasSubscriptionTables(Oid subid)
600 : : {
601 : : Relation rel;
602 : : ScanKeyData skey[1];
603 : : SysScanDesc scan;
604 : : HeapTuple tup;
308 605 : 315 : bool has_subtables = false;
606 : :
1870 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 : :
308 617 [ + + ]: 394 : while (HeapTupleIsValid(tup = systable_getnext(scan)))
618 : : {
619 : : Form_pg_subscription_rel subrel;
620 : : char relkind;
621 : :
622 : 369 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
623 : 369 : relkind = get_rel_relkind(subrel->srrelid);
624 : :
625 [ + + + + ]: 369 : if (relkind == RELKIND_RELATION ||
626 : : relkind == RELKIND_PARTITIONED_TABLE)
627 : : {
628 : 290 : has_subtables = true;
629 : 290 : break;
630 : : }
631 : : }
632 : :
633 : : /* Cleanup */
1870 634 : 315 : systable_endscan(scan);
635 : 315 : table_close(rel, AccessShareLock);
636 : :
308 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 : 1279 : GetSubscriptionRelations(Oid subid, bool tables, bool sequences,
649 : : bool not_ready)
650 : : {
3444 peter_e@gmx.net 651 : 1279 : List *res = NIL;
652 : : Relation rel;
653 : : HeapTuple tup;
654 : 1279 : int nkeys = 0;
655 : : ScanKeyData skey[2];
656 : : SysScanDesc scan;
657 : :
658 : : /* One or both of 'tables' and 'sequences' must be true. */
308 akapila@postgresql.o 659 [ + + - + ]: 1279 : Assert(tables || sequences);
660 : :
2775 andres@anarazel.de 661 : 1279 : rel = table_open(SubscriptionRelRelationId, AccessShareLock);
662 : :
3444 peter_e@gmx.net 663 : 1279 : ScanKeyInit(&skey[nkeys++],
664 : : Anum_pg_subscription_rel_srsubid,
665 : : BTEqualStrategyNumber, F_OIDEQ,
666 : : ObjectIdGetDatum(subid));
667 : :
1492 michael@paquier.xyz 668 [ + + ]: 1279 : if (not_ready)
669 : 1235 : ScanKeyInit(&skey[nkeys++],
670 : : Anum_pg_subscription_rel_srsubstate,
671 : : BTEqualStrategyNumber, F_CHARNE,
672 : : CharGetDatum(SUBREL_STATE_READY));
673 : :
3444 peter_e@gmx.net 674 : 1279 : scan = systable_beginscan(rel, InvalidOid, false,
675 : : NULL, nkeys, skey);
676 : :
677 [ + + ]: 3296 : 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 : 2017 : subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
686 : :
308 akapila@postgresql.o 687 : 2017 : relkind = get_rel_relkind(subrel->srrelid);
688 : :
689 : : /* The relation may have been dropped concurrently. */
1 690 [ - + ]: 2017 : if (relkind == '\0')
308 akapila@postgresql.o 691 :UBC 0 : continue;
692 : :
693 : : /*
694 : : * The relation must be either a sequence or a table. Anything else
695 : : * indicates an unexpected relation kind for a subscription relation.
696 : : */
1 akapila@postgresql.o 697 [ + + ]:CBC 2017 : if (relkind == RELKIND_SEQUENCE)
698 : : {
699 : : /* Skip sequences if they were not requested */
700 [ - + ]: 68 : if (!sequences)
1 akapila@postgresql.o 701 :UBC 0 : continue;
702 : : }
1 akapila@postgresql.o 703 [ + + + - ]:CBC 1949 : else if (relkind == RELKIND_RELATION ||
704 : : relkind == RELKIND_PARTITIONED_TABLE)
705 : : {
706 : : /* Skip tables if they were not requested */
707 [ - + ]: 1949 : if (!tables)
1 akapila@postgresql.o 708 :UBC 0 : continue;
709 : : }
710 : : else
711 [ # # ]: 0 : elog(ERROR, "unexpected relkind \"%c\" for relation %u in subscription %u",
712 : : relkind, subrel->srrelid, subid);
713 : :
260 michael@paquier.xyz 714 :CBC 2017 : relstate = palloc_object(SubscriptionRelState);
3444 peter_e@gmx.net 715 : 2017 : relstate->relid = subrel->srrelid;
716 : 2017 : relstate->state = subrel->srsubstate;
2229 tgl@sss.pgh.pa.us 717 : 2017 : d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
718 : : Anum_pg_subscription_rel_srsublsn, &isnull);
719 [ + + ]: 2017 : if (isnull)
720 : 1647 : relstate->lsn = InvalidXLogRecPtr;
721 : : else
722 : 370 : relstate->lsn = DatumGetLSN(d);
723 : :
3444 peter_e@gmx.net 724 : 2017 : res = lappend(res, relstate);
725 : : }
726 : :
727 : : /* Cleanup */
728 : 1279 : systable_endscan(scan);
2775 andres@anarazel.de 729 : 1279 : table_close(rel, AccessShareLock);
730 : :
3444 peter_e@gmx.net 731 : 1279 : return res;
732 : : }
733 : :
734 : : /*
735 : : * Update the dead tuple retention status for the given subscription.
736 : : */
737 : : void
359 akapila@postgresql.o 738 : 2 : UpdateDeadTupleRetentionStatus(Oid subid, bool active)
739 : : {
740 : : Relation rel;
741 : : bool nulls[Natts_pg_subscription];
742 : : bool replaces[Natts_pg_subscription];
743 : : Datum values[Natts_pg_subscription];
744 : : HeapTuple tup;
745 : :
746 : : /* Look up the subscription in the catalog */
747 : 2 : rel = table_open(SubscriptionRelationId, RowExclusiveLock);
748 : 2 : tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
749 : :
750 [ - + ]: 2 : if (!HeapTupleIsValid(tup))
359 akapila@postgresql.o 751 [ # # ]:UBC 0 : elog(ERROR, "cache lookup failed for subscription %u", subid);
752 : :
753 : : /* Must only modify subscriptions belonging to the current database. */
40 jdavis@postgresql.or 754 [ - + ]:CBC 2 : Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
755 : :
359 akapila@postgresql.o 756 : 2 : LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
757 : :
758 : : /* Form a new tuple. */
759 : 2 : memset(values, 0, sizeof(values));
760 : 2 : memset(nulls, false, sizeof(nulls));
761 : 2 : memset(replaces, false, sizeof(replaces));
762 : :
763 : : /* Set the subscription to disabled. */
129 peter@eisentraut.org 764 : 2 : values[Anum_pg_subscription_subretentionactive - 1] = BoolGetDatum(active);
359 akapila@postgresql.o 765 : 2 : replaces[Anum_pg_subscription_subretentionactive - 1] = true;
766 : :
767 : : /* Update the catalog */
768 : 2 : tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
769 : : replaces);
770 : 2 : CatalogTupleUpdate(rel, &tup->t_self, tup);
771 : 2 : heap_freetuple(tup);
772 : :
773 : 2 : table_close(rel, NoLock);
774 : 2 : }
|
