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-2020, 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/pg_lsn.h"
33 #include "utils/rel.h"
34 #include "utils/syscache.h"
35 
36 static List *textarray_to_stringlist(ArrayType *textarray);
37 
38 /*
39  * Fetch the subscription from the syscache.
40  */
42 GetSubscription(Oid subid, bool missing_ok)
43 {
44  HeapTuple tup;
45  Subscription *sub;
46  Form_pg_subscription subform;
47  Datum datum;
48  bool isnull;
49 
51 
52  if (!HeapTupleIsValid(tup))
53  {
54  if (missing_ok)
55  return NULL;
56 
57  elog(ERROR, "cache lookup failed for subscription %u", subid);
58  }
59 
60  subform = (Form_pg_subscription) GETSTRUCT(tup);
61 
62  sub = (Subscription *) palloc(sizeof(Subscription));
63  sub->oid = subid;
64  sub->dbid = subform->subdbid;
65  sub->name = pstrdup(NameStr(subform->subname));
66  sub->owner = subform->subowner;
67  sub->enabled = subform->subenabled;
68  sub->binary = subform->subbinary;
69  sub->stream = subform->substream;
70 
71  /* Get conninfo */
73  tup,
74  Anum_pg_subscription_subconninfo,
75  &isnull);
76  Assert(!isnull);
77  sub->conninfo = TextDatumGetCString(datum);
78 
79  /* Get slotname */
81  tup,
82  Anum_pg_subscription_subslotname,
83  &isnull);
84  if (!isnull)
85  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
86  else
87  sub->slotname = NULL;
88 
89  /* Get synccommit */
91  tup,
92  Anum_pg_subscription_subsynccommit,
93  &isnull);
94  Assert(!isnull);
95  sub->synccommit = TextDatumGetCString(datum);
96 
97  /* Get publications */
99  tup,
100  Anum_pg_subscription_subpublications,
101  &isnull);
102  Assert(!isnull);
104 
105  ReleaseSysCache(tup);
106 
107  return sub;
108 }
109 
110 /*
111  * Return number of subscriptions defined in given database.
112  * Used by dropdb() to check if database can indeed be dropped.
113  */
114 int
116 {
117  int nsubs = 0;
118  Relation rel;
119  ScanKeyData scankey;
120  SysScanDesc scan;
121  HeapTuple tup;
122 
123  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
124 
125  ScanKeyInit(&scankey,
126  Anum_pg_subscription_subdbid,
127  BTEqualStrategyNumber, F_OIDEQ,
128  ObjectIdGetDatum(dbid));
129 
130  scan = systable_beginscan(rel, InvalidOid, false,
131  NULL, 1, &scankey);
132 
133  while (HeapTupleIsValid(tup = systable_getnext(scan)))
134  nsubs++;
135 
136  systable_endscan(scan);
137 
138  table_close(rel, NoLock);
139 
140  return nsubs;
141 }
142 
143 /*
144  * Free memory allocated by subscription struct.
145  */
146 void
148 {
149  pfree(sub->name);
150  pfree(sub->conninfo);
151  if (sub->slotname)
152  pfree(sub->slotname);
154  pfree(sub);
155 }
156 
157 /*
158  * get_subscription_oid - given a subscription name, look up the OID
159  *
160  * If missing_ok is false, throw an error if name not found. If true, just
161  * return InvalidOid.
162  */
163 Oid
164 get_subscription_oid(const char *subname, bool missing_ok)
165 {
166  Oid oid;
167 
168  oid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
169  MyDatabaseId, CStringGetDatum(subname));
170  if (!OidIsValid(oid) && !missing_ok)
171  ereport(ERROR,
172  (errcode(ERRCODE_UNDEFINED_OBJECT),
173  errmsg("subscription \"%s\" does not exist", subname)));
174  return oid;
175 }
176 
177 /*
178  * get_subscription_name - given a subscription OID, look up the name
179  *
180  * If missing_ok is false, throw an error if name not found. If true, just
181  * return NULL.
182  */
183 char *
184 get_subscription_name(Oid subid, bool missing_ok)
185 {
186  HeapTuple tup;
187  char *subname;
188  Form_pg_subscription subform;
189 
191 
192  if (!HeapTupleIsValid(tup))
193  {
194  if (!missing_ok)
195  elog(ERROR, "cache lookup failed for subscription %u", subid);
196  return NULL;
197  }
198 
199  subform = (Form_pg_subscription) GETSTRUCT(tup);
200  subname = pstrdup(NameStr(subform->subname));
201 
202  ReleaseSysCache(tup);
203 
204  return subname;
205 }
206 
207 /*
208  * Convert text array to list of strings.
209  *
210  * Note: the resulting list of strings is pallocated here.
211  */
212 static List *
214 {
215  Datum *elems;
216  int nelems,
217  i;
218  List *res = NIL;
219 
220  deconstruct_array(textarray,
221  TEXTOID, -1, false, TYPALIGN_INT,
222  &elems, NULL, &nelems);
223 
224  if (nelems == 0)
225  return NIL;
226 
227  for (i = 0; i < nelems; i++)
228  res = lappend(res, makeString(TextDatumGetCString(elems[i])));
229 
230  return res;
231 }
232 
233 /*
234  * Add new state record for a subscription table.
235  */
236 void
238  XLogRecPtr sublsn)
239 {
240  Relation rel;
241  HeapTuple tup;
242  bool nulls[Natts_pg_subscription_rel];
243  Datum values[Natts_pg_subscription_rel];
244 
245  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
246 
247  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
248 
249  /* Try finding existing mapping. */
251  ObjectIdGetDatum(relid),
252  ObjectIdGetDatum(subid));
253  if (HeapTupleIsValid(tup))
254  elog(ERROR, "subscription table %u in subscription %u already exists",
255  relid, subid);
256 
257  /* Form the tuple. */
258  memset(values, 0, sizeof(values));
259  memset(nulls, false, sizeof(nulls));
260  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
261  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
262  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
263  if (sublsn != InvalidXLogRecPtr)
264  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
265  else
266  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
267 
268  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
269 
270  /* Insert tuple into catalog. */
271  CatalogTupleInsert(rel, tup);
272 
273  heap_freetuple(tup);
274 
275  /* Cleanup. */
276  table_close(rel, NoLock);
277 }
278 
279 /*
280  * Update the state of a subscription table.
281  */
282 void
284  XLogRecPtr sublsn)
285 {
286  Relation rel;
287  HeapTuple tup;
288  bool nulls[Natts_pg_subscription_rel];
289  Datum values[Natts_pg_subscription_rel];
290  bool replaces[Natts_pg_subscription_rel];
291 
292  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
293 
294  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
295 
296  /* Try finding existing mapping. */
298  ObjectIdGetDatum(relid),
299  ObjectIdGetDatum(subid));
300  if (!HeapTupleIsValid(tup))
301  elog(ERROR, "subscription table %u in subscription %u does not exist",
302  relid, subid);
303 
304  /* Update the tuple. */
305  memset(values, 0, sizeof(values));
306  memset(nulls, false, sizeof(nulls));
307  memset(replaces, false, sizeof(replaces));
308 
309  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
310  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
311 
312  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
313  if (sublsn != InvalidXLogRecPtr)
314  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
315  else
316  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
317 
318  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
319  replaces);
320 
321  /* Update the catalog. */
322  CatalogTupleUpdate(rel, &tup->t_self, tup);
323 
324  /* Cleanup. */
325  table_close(rel, NoLock);
326 }
327 
328 /*
329  * Get state of subscription table.
330  *
331  * Returns SUBREL_STATE_UNKNOWN when not found and missing_ok is true.
332  */
333 char
335  bool missing_ok)
336 {
337  Relation rel;
338  HeapTuple tup;
339  char substate;
340  bool isnull;
341  Datum d;
342 
343  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
344 
345  /* Try finding the mapping. */
347  ObjectIdGetDatum(relid),
348  ObjectIdGetDatum(subid));
349 
350  if (!HeapTupleIsValid(tup))
351  {
352  if (missing_ok)
353  {
355  *sublsn = InvalidXLogRecPtr;
356  return SUBREL_STATE_UNKNOWN;
357  }
358 
359  elog(ERROR, "subscription table %u in subscription %u does not exist",
360  relid, subid);
361  }
362 
363  /* Get the state. */
365  Anum_pg_subscription_rel_srsubstate, &isnull);
366  Assert(!isnull);
367  substate = DatumGetChar(d);
369  Anum_pg_subscription_rel_srsublsn, &isnull);
370  if (isnull)
371  *sublsn = InvalidXLogRecPtr;
372  else
373  *sublsn = DatumGetLSN(d);
374 
375  /* Cleanup */
376  ReleaseSysCache(tup);
378 
379  return substate;
380 }
381 
382 /*
383  * Drop subscription relation mapping. These can be for a particular
384  * subscription, or for a particular relation, or both.
385  */
386 void
388 {
389  Relation rel;
390  TableScanDesc scan;
391  ScanKeyData skey[2];
392  HeapTuple tup;
393  int nkeys = 0;
394 
395  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
396 
397  if (OidIsValid(subid))
398  {
399  ScanKeyInit(&skey[nkeys++],
400  Anum_pg_subscription_rel_srsubid,
402  F_OIDEQ,
403  ObjectIdGetDatum(subid));
404  }
405 
406  if (OidIsValid(relid))
407  {
408  ScanKeyInit(&skey[nkeys++],
409  Anum_pg_subscription_rel_srrelid,
411  F_OIDEQ,
412  ObjectIdGetDatum(relid));
413  }
414 
415  /* Do the search and delete what we found. */
416  scan = table_beginscan_catalog(rel, nkeys, skey);
418  {
419  CatalogTupleDelete(rel, &tup->t_self);
420  }
421  table_endscan(scan);
422 
424 }
425 
426 
427 /*
428  * Get all relations for subscription.
429  *
430  * Returned list is palloc'ed in current memory context.
431  */
432 List *
434 {
435  List *res = NIL;
436  Relation rel;
437  HeapTuple tup;
438  int nkeys = 0;
439  ScanKeyData skey[2];
440  SysScanDesc scan;
441 
442  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
443 
444  ScanKeyInit(&skey[nkeys++],
445  Anum_pg_subscription_rel_srsubid,
446  BTEqualStrategyNumber, F_OIDEQ,
447  ObjectIdGetDatum(subid));
448 
449  scan = systable_beginscan(rel, InvalidOid, false,
450  NULL, nkeys, skey);
451 
452  while (HeapTupleIsValid(tup = systable_getnext(scan)))
453  {
455  SubscriptionRelState *relstate;
456  Datum d;
457  bool isnull;
458 
459  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
460 
461  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
462  relstate->relid = subrel->srrelid;
463  relstate->state = subrel->srsubstate;
465  Anum_pg_subscription_rel_srsublsn, &isnull);
466  if (isnull)
467  relstate->lsn = InvalidXLogRecPtr;
468  else
469  relstate->lsn = DatumGetLSN(d);
470 
471  res = lappend(res, relstate);
472  }
473 
474  /* Cleanup */
475  systable_endscan(scan);
477 
478  return res;
479 }
480 
481 /*
482  * Get all relations for subscription that are not in a ready state.
483  *
484  * Returned list is palloc'ed in current memory context.
485  */
486 List *
488 {
489  List *res = NIL;
490  Relation rel;
491  HeapTuple tup;
492  int nkeys = 0;
493  ScanKeyData skey[2];
494  SysScanDesc scan;
495 
496  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
497 
498  ScanKeyInit(&skey[nkeys++],
499  Anum_pg_subscription_rel_srsubid,
500  BTEqualStrategyNumber, F_OIDEQ,
501  ObjectIdGetDatum(subid));
502 
503  ScanKeyInit(&skey[nkeys++],
504  Anum_pg_subscription_rel_srsubstate,
505  BTEqualStrategyNumber, F_CHARNE,
506  CharGetDatum(SUBREL_STATE_READY));
507 
508  scan = systable_beginscan(rel, InvalidOid, false,
509  NULL, nkeys, skey);
510 
511  while (HeapTupleIsValid(tup = systable_getnext(scan)))
512  {
514  SubscriptionRelState *relstate;
515  Datum d;
516  bool isnull;
517 
518  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
519 
520  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
521  relstate->relid = subrel->srrelid;
522  relstate->state = subrel->srsubstate;
524  Anum_pg_subscription_rel_srsublsn, &isnull);
525  if (isnull)
526  relstate->lsn = InvalidXLogRecPtr;
527  else
528  relstate->lsn = DatumGetLSN(d);
529 
530  res = lappend(res, relstate);
531  }
532 
533  /* Cleanup */
534  systable_endscan(scan);
536 
537  return res;
538 }
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
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:569
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
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:482
char * pstrdup(const char *in)
Definition: mcxt.c:1187
int CountDBSubscriptions(Oid dbid)
#define AccessShareLock
Definition: lockdefs.h:36
int errcode(int sqlerrcode)
Definition: elog.c:610
#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:651
void list_free_deep(List *list)
Definition: list.c:1390
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:357
char * get_subscription_name(Oid subid, bool missing_ok)
#define DatumGetName(X)
Definition: postgres.h:585
Subscription * GetSubscription(Oid subid, bool missing_ok)
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:476
void pfree(void *pointer)
Definition: mcxt.c:1057
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
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:1286
#define RowExclusiveLock
Definition: lockdefs.h:38
List * GetSubscriptionRelations(Oid subid)
#define CStringGetDatum(X)
Definition: postgres.h:578
List * publications
List * lappend(List *list, void *datum)
Definition: list.c:321
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1116
#define TextDatumGetCString(d)
Definition: builtins.h:87
uintptr_t Datum
Definition: postgres.h:367
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1164
#define DatumGetChar(X)
Definition: postgres.h:409
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1377
Oid MyDatabaseId
Definition: globals.c:85
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:144
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:745
Definition: regguts.h:298
#define GetSysCacheOid2(cacheId, oidcol, key1, key2)
Definition: syscache.h:194
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:301
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:1127
char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn, bool missing_ok)
#define CharGetDatum(X)
Definition: postgres.h:416
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:3483
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:863
static Datum values[MAXATTR]
Definition: bootstrap.c:165
void * palloc(Size size)
Definition: mcxt.c:950
int errmsg(const char *fmt,...)
Definition: elog.c:824
#define elog(elevel,...)
Definition: elog.h:214
int i
#define NameStr(name)
Definition: c.h:622
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:176
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)
void CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:221
#define BTEqualStrategyNumber
Definition: stratnum.h:31
#define DatumGetArrayTypeP(X)
Definition: array.h:249