PostgreSQL Source Code  git master
pg_subscription.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * pg_subscription.c
4  * replication subscriptions
5  *
6  * Portions Copyright (c) 1996-2021, 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 "access/xact.h"
22 #include "catalog/indexing.h"
25 #include "catalog/pg_type.h"
26 #include "miscadmin.h"
27 #include "nodes/makefuncs.h"
28 #include "storage/lmgr.h"
29 #include "utils/array.h"
30 #include "utils/builtins.h"
31 #include "utils/fmgroids.h"
32 #include "utils/lsyscache.h"
33 #include "utils/pg_lsn.h"
34 #include "utils/rel.h"
35 #include "utils/syscache.h"
36 
37 static List *textarray_to_stringlist(ArrayType *textarray);
38 
39 /*
40  * Fetch the subscription from the syscache.
41  */
43 GetSubscription(Oid subid, bool missing_ok)
44 {
45  HeapTuple tup;
46  Subscription *sub;
47  Form_pg_subscription subform;
48  Datum datum;
49  bool isnull;
50 
52 
53  if (!HeapTupleIsValid(tup))
54  {
55  if (missing_ok)
56  return NULL;
57 
58  elog(ERROR, "cache lookup failed for subscription %u", subid);
59  }
60 
61  subform = (Form_pg_subscription) GETSTRUCT(tup);
62 
63  sub = (Subscription *) palloc(sizeof(Subscription));
64  sub->oid = subid;
65  sub->dbid = subform->subdbid;
66  sub->name = pstrdup(NameStr(subform->subname));
67  sub->owner = subform->subowner;
68  sub->enabled = subform->subenabled;
69  sub->binary = subform->subbinary;
70  sub->stream = subform->substream;
71  sub->twophasestate = subform->subtwophasestate;
72 
73  /* Get conninfo */
75  tup,
76  Anum_pg_subscription_subconninfo,
77  &isnull);
78  Assert(!isnull);
79  sub->conninfo = TextDatumGetCString(datum);
80 
81  /* Get slotname */
83  tup,
84  Anum_pg_subscription_subslotname,
85  &isnull);
86  if (!isnull)
87  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
88  else
89  sub->slotname = NULL;
90 
91  /* Get synccommit */
93  tup,
94  Anum_pg_subscription_subsynccommit,
95  &isnull);
96  Assert(!isnull);
97  sub->synccommit = TextDatumGetCString(datum);
98 
99  /* Get publications */
101  tup,
102  Anum_pg_subscription_subpublications,
103  &isnull);
104  Assert(!isnull);
106 
107  ReleaseSysCache(tup);
108 
109  return sub;
110 }
111 
112 /*
113  * Return number of subscriptions defined in given database.
114  * Used by dropdb() to check if database can indeed be dropped.
115  */
116 int
118 {
119  int nsubs = 0;
120  Relation rel;
121  ScanKeyData scankey;
122  SysScanDesc scan;
123  HeapTuple tup;
124 
125  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
126 
127  ScanKeyInit(&scankey,
128  Anum_pg_subscription_subdbid,
129  BTEqualStrategyNumber, F_OIDEQ,
130  ObjectIdGetDatum(dbid));
131 
132  scan = systable_beginscan(rel, InvalidOid, false,
133  NULL, 1, &scankey);
134 
135  while (HeapTupleIsValid(tup = systable_getnext(scan)))
136  nsubs++;
137 
138  systable_endscan(scan);
139 
140  table_close(rel, NoLock);
141 
142  return nsubs;
143 }
144 
145 /*
146  * Free memory allocated by subscription struct.
147  */
148 void
150 {
151  pfree(sub->name);
152  pfree(sub->conninfo);
153  if (sub->slotname)
154  pfree(sub->slotname);
156  pfree(sub);
157 }
158 
159 /*
160  * get_subscription_oid - given a subscription name, look up the OID
161  *
162  * If missing_ok is false, throw an error if name not found. If true, just
163  * return InvalidOid.
164  */
165 Oid
166 get_subscription_oid(const char *subname, bool missing_ok)
167 {
168  Oid oid;
169 
170  oid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
171  MyDatabaseId, CStringGetDatum(subname));
172  if (!OidIsValid(oid) && !missing_ok)
173  ereport(ERROR,
174  (errcode(ERRCODE_UNDEFINED_OBJECT),
175  errmsg("subscription \"%s\" does not exist", subname)));
176  return oid;
177 }
178 
179 /*
180  * get_subscription_name - given a subscription OID, look up the name
181  *
182  * If missing_ok is false, throw an error if name not found. If true, just
183  * return NULL.
184  */
185 char *
186 get_subscription_name(Oid subid, bool missing_ok)
187 {
188  HeapTuple tup;
189  char *subname;
190  Form_pg_subscription subform;
191 
193 
194  if (!HeapTupleIsValid(tup))
195  {
196  if (!missing_ok)
197  elog(ERROR, "cache lookup failed for subscription %u", subid);
198  return NULL;
199  }
200 
201  subform = (Form_pg_subscription) GETSTRUCT(tup);
202  subname = pstrdup(NameStr(subform->subname));
203 
204  ReleaseSysCache(tup);
205 
206  return subname;
207 }
208 
209 /*
210  * Convert text array to list of strings.
211  *
212  * Note: the resulting list of strings is pallocated here.
213  */
214 static List *
216 {
217  Datum *elems;
218  int nelems,
219  i;
220  List *res = NIL;
221 
222  deconstruct_array(textarray,
223  TEXTOID, -1, false, TYPALIGN_INT,
224  &elems, NULL, &nelems);
225 
226  if (nelems == 0)
227  return NIL;
228 
229  for (i = 0; i < nelems; i++)
230  res = lappend(res, makeString(TextDatumGetCString(elems[i])));
231 
232  return res;
233 }
234 
235 /*
236  * Add new state record for a subscription table.
237  */
238 void
240  XLogRecPtr sublsn)
241 {
242  Relation rel;
243  HeapTuple tup;
244  bool nulls[Natts_pg_subscription_rel];
245  Datum values[Natts_pg_subscription_rel];
246 
247  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
248 
249  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
250 
251  /* Try finding existing mapping. */
253  ObjectIdGetDatum(relid),
254  ObjectIdGetDatum(subid));
255  if (HeapTupleIsValid(tup))
256  elog(ERROR, "subscription table %u in subscription %u already exists",
257  relid, subid);
258 
259  /* Form the tuple. */
260  memset(values, 0, sizeof(values));
261  memset(nulls, false, sizeof(nulls));
262  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
263  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
264  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
265  if (sublsn != InvalidXLogRecPtr)
266  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
267  else
268  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
269 
270  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
271 
272  /* Insert tuple into catalog. */
273  CatalogTupleInsert(rel, tup);
274 
275  heap_freetuple(tup);
276 
277  /* Cleanup. */
278  table_close(rel, NoLock);
279 }
280 
281 /*
282  * Update the state of a subscription table.
283  */
284 void
286  XLogRecPtr sublsn)
287 {
288  Relation rel;
289  HeapTuple tup;
290  bool nulls[Natts_pg_subscription_rel];
291  Datum values[Natts_pg_subscription_rel];
292  bool replaces[Natts_pg_subscription_rel];
293 
294  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
295 
296  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
297 
298  /* Try finding existing mapping. */
300  ObjectIdGetDatum(relid),
301  ObjectIdGetDatum(subid));
302  if (!HeapTupleIsValid(tup))
303  elog(ERROR, "subscription table %u in subscription %u does not exist",
304  relid, subid);
305 
306  /* Update the tuple. */
307  memset(values, 0, sizeof(values));
308  memset(nulls, false, sizeof(nulls));
309  memset(replaces, false, sizeof(replaces));
310 
311  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
312  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
313 
314  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
315  if (sublsn != InvalidXLogRecPtr)
316  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
317  else
318  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
319 
320  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
321  replaces);
322 
323  /* Update the catalog. */
324  CatalogTupleUpdate(rel, &tup->t_self, tup);
325 
326  /* Cleanup. */
327  table_close(rel, NoLock);
328 }
329 
330 /*
331  * Get state of subscription table.
332  *
333  * Returns SUBREL_STATE_UNKNOWN when the table is not in the subscription.
334  */
335 char
337 {
338  HeapTuple tup;
339  char substate;
340  bool isnull;
341  Datum d;
342  Relation rel;
343 
344  /*
345  * This is to avoid the race condition with AlterSubscription which tries
346  * to remove this relstate.
347  */
348  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
349 
350  /* Try finding the mapping. */
352  ObjectIdGetDatum(relid),
353  ObjectIdGetDatum(subid));
354 
355  if (!HeapTupleIsValid(tup))
356  {
358  *sublsn = InvalidXLogRecPtr;
359  return SUBREL_STATE_UNKNOWN;
360  }
361 
362  /* Get the state. */
363  substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
364 
365  /* Get the LSN */
367  Anum_pg_subscription_rel_srsublsn, &isnull);
368  if (isnull)
369  *sublsn = InvalidXLogRecPtr;
370  else
371  *sublsn = DatumGetLSN(d);
372 
373  /* Cleanup */
374  ReleaseSysCache(tup);
375 
377 
378  return substate;
379 }
380 
381 /*
382  * Drop subscription relation mapping. These can be for a particular
383  * subscription, or for a particular relation, or both.
384  */
385 void
387 {
388  Relation rel;
389  TableScanDesc scan;
390  ScanKeyData skey[2];
391  HeapTuple tup;
392  int nkeys = 0;
393 
394  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
395 
396  if (OidIsValid(subid))
397  {
398  ScanKeyInit(&skey[nkeys++],
399  Anum_pg_subscription_rel_srsubid,
401  F_OIDEQ,
402  ObjectIdGetDatum(subid));
403  }
404 
405  if (OidIsValid(relid))
406  {
407  ScanKeyInit(&skey[nkeys++],
408  Anum_pg_subscription_rel_srrelid,
410  F_OIDEQ,
411  ObjectIdGetDatum(relid));
412  }
413 
414  /* Do the search and delete what we found. */
415  scan = table_beginscan_catalog(rel, nkeys, skey);
417  {
419 
420  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
421 
422  /*
423  * We don't allow to drop the relation mapping when the table
424  * synchronization is in progress unless the caller updates the
425  * corresponding subscription as well. This is to ensure that we don't
426  * leave tablesync slots or origins in the system when the
427  * corresponding table is dropped.
428  */
429  if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
430  {
431  ereport(ERROR,
432  (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
433  errmsg("could not drop relation mapping for subscription \"%s\"",
434  get_subscription_name(subrel->srsubid, false)),
435  errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
436  get_rel_name(relid), subrel->srsubstate),
437 
438  /*
439  * translator: first %s is a SQL ALTER command and second %s is a
440  * SQL DROP command
441  */
442  errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
443  "ALTER SUBSCRIPTION ... ENABLE",
444  "DROP SUBSCRIPTION ...")));
445  }
446 
447  CatalogTupleDelete(rel, &tup->t_self);
448  }
449  table_endscan(scan);
450 
452 }
453 
454 /*
455  * Does the subscription have any relations?
456  *
457  * Use this function only to know true/false, and when you have no need for the
458  * List returned by GetSubscriptionRelations.
459  */
460 bool
462 {
463  Relation rel;
464  ScanKeyData skey[1];
465  SysScanDesc scan;
466  bool has_subrels;
467 
468  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
469 
470  ScanKeyInit(&skey[0],
471  Anum_pg_subscription_rel_srsubid,
472  BTEqualStrategyNumber, F_OIDEQ,
473  ObjectIdGetDatum(subid));
474 
475  scan = systable_beginscan(rel, InvalidOid, false,
476  NULL, 1, skey);
477 
478  /* If even a single tuple exists then the subscription has tables. */
479  has_subrels = HeapTupleIsValid(systable_getnext(scan));
480 
481  /* Cleanup */
482  systable_endscan(scan);
484 
485  return has_subrels;
486 }
487 
488 /*
489  * Get all relations for subscription.
490  *
491  * Returned list is palloc'ed in current memory context.
492  */
493 List *
495 {
496  List *res = NIL;
497  Relation rel;
498  HeapTuple tup;
499  ScanKeyData skey[1];
500  SysScanDesc scan;
501 
502  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
503 
504  ScanKeyInit(&skey[0],
505  Anum_pg_subscription_rel_srsubid,
506  BTEqualStrategyNumber, F_OIDEQ,
507  ObjectIdGetDatum(subid));
508 
509  scan = systable_beginscan(rel, InvalidOid, false,
510  NULL, 1, skey);
511 
512  while (HeapTupleIsValid(tup = systable_getnext(scan)))
513  {
515  SubscriptionRelState *relstate;
516  Datum d;
517  bool isnull;
518 
519  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
520 
521  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
522  relstate->relid = subrel->srrelid;
523  relstate->state = subrel->srsubstate;
525  Anum_pg_subscription_rel_srsublsn, &isnull);
526  if (isnull)
527  relstate->lsn = InvalidXLogRecPtr;
528  else
529  relstate->lsn = DatumGetLSN(d);
530 
531  res = lappend(res, relstate);
532  }
533 
534  /* Cleanup */
535  systable_endscan(scan);
537 
538  return res;
539 }
540 
541 /*
542  * Get all relations for subscription that are not in a ready state.
543  *
544  * Returned list is palloc'ed in current memory context.
545  */
546 List *
548 {
549  List *res = NIL;
550  Relation rel;
551  HeapTuple tup;
552  int nkeys = 0;
553  ScanKeyData skey[2];
554  SysScanDesc scan;
555 
556  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
557 
558  ScanKeyInit(&skey[nkeys++],
559  Anum_pg_subscription_rel_srsubid,
560  BTEqualStrategyNumber, F_OIDEQ,
561  ObjectIdGetDatum(subid));
562 
563  ScanKeyInit(&skey[nkeys++],
564  Anum_pg_subscription_rel_srsubstate,
565  BTEqualStrategyNumber, F_CHARNE,
566  CharGetDatum(SUBREL_STATE_READY));
567 
568  scan = systable_beginscan(rel, InvalidOid, false,
569  NULL, nkeys, skey);
570 
571  while (HeapTupleIsValid(tup = systable_getnext(scan)))
572  {
574  SubscriptionRelState *relstate;
575  Datum d;
576  bool isnull;
577 
578  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
579 
580  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
581  relstate->relid = subrel->srrelid;
582  relstate->state = subrel->srsubstate;
584  Anum_pg_subscription_rel_srsublsn, &isnull);
585  if (isnull)
586  relstate->lsn = InvalidXLogRecPtr;
587  else
588  relstate->lsn = DatumGetLSN(d);
589 
590  res = lappend(res, relstate);
591  }
592 
593  /* Cleanup */
594  systable_endscan(scan);
596 
597  return res;
598 }
Value * makeString(char *str)
Definition: value.c:53
#define NIL
Definition: pg_list.h:65
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:167
int errhint(const char *fmt,...)
Definition: elog.c:1156
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:595
#define GETSTRUCT(TUP)
Definition: htup_details.h:654
void RemoveSubscriptionRel(Oid subid, Oid relid)
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:112
#define RelationGetDescr(relation)
Definition: rel.h:503
char * pstrdup(const char *in)
Definition: mcxt.c:1299
int CountDBSubscriptions(Oid dbid)
#define AccessShareLock
Definition: lockdefs.h:36
int errcode(int sqlerrcode)
Definition: elog.c:698
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:350
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
FormData_pg_subscription * Form_pg_subscription
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1338
unsigned int Oid
Definition: postgres_ext.h:31
NameData subname
#define OidIsValid(objectId)
Definition: c.h:710
void list_free_deep(List *list)
Definition: list.c:1405
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:383
char * get_subscription_name(Oid subid, bool missing_ok)
#define DatumGetName(X)
Definition: postgres.h:629
Subscription * GetSubscription(Oid subid, bool missing_ok)
bool HasSubscriptionRelations(Oid subid)
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:502
void pfree(void *pointer)
Definition: mcxt.c:1169
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
#define ERROR
Definition: elog.h:46
static List * textarray_to_stringlist(ArrayType *textarray)
ItemPointerData t_self
Definition: htup.h:65
void AddSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn)
#define NoLock
Definition: lockdefs.h:34
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1340
#define RowExclusiveLock
Definition: lockdefs.h:38
int errdetail(const char *fmt,...)
Definition: elog.c:1042
List * GetSubscriptionRelations(Oid subid)
#define CStringGetDatum(X)
Definition: postgres.h:622
List * lappend(List *list, void *datum)
Definition: list.c:336
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1127
#define TextDatumGetCString(d)
Definition: builtins.h:83
char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn)
uintptr_t Datum
Definition: postgres.h:411
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1175
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1388
Oid MyDatabaseId
Definition: globals.c:88
void UpdateSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn)
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1017
#define InvalidOid
Definition: postgres_ext.h:36
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define ereport(elevel,...)
Definition: elog.h:157
Oid get_subscription_oid(const char *subname, bool missing_ok)
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:804
Definition: regguts.h:317
#define GetSysCacheOid2(cacheId, oidcol, key1, key2)
Definition: syscache.h:195
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:301
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:1138
#define CharGetDatum(X)
Definition: postgres.h:460
void FreeSubscription(Subscription *sub)
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3490
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:991
static Datum values[MAXATTR]
Definition: bootstrap.c:166
void * palloc(Size size)
Definition: mcxt.c:1062
int errmsg(const char *fmt,...)
Definition: elog.c:909
#define elog(elevel,...)
Definition: elog.h:232
int i
#define NameStr(name)
Definition: c.h:681
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:177
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:1113
Definition: pg_list.h:50
List * GetSubscriptionNotReadyRelations(Oid subid)
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1899
void CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:221
#define BTEqualStrategyNumber
Definition: stratnum.h:31
#define DatumGetArrayTypeP(X)
Definition: array.h:254