PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
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-2017, 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 "miscadmin.h"
18 
19 #include "access/genam.h"
20 #include "access/heapam.h"
21 #include "access/htup_details.h"
22 #include "access/xact.h"
23 
24 #include "catalog/indexing.h"
25 #include "catalog/pg_type.h"
28 
29 #include "nodes/makefuncs.h"
30 
31 #include "storage/lmgr.h"
32 
33 #include "utils/array.h"
34 #include "utils/builtins.h"
35 #include "utils/fmgroids.h"
36 #include "utils/pg_lsn.h"
37 #include "utils/rel.h"
38 #include "utils/syscache.h"
39 
40 
41 static List *textarray_to_stringlist(ArrayType *textarray);
42 
43 /*
44  * Fetch the subscription from the syscache.
45  */
47 GetSubscription(Oid subid, bool missing_ok)
48 {
49  HeapTuple tup;
50  Subscription *sub;
51  Form_pg_subscription subform;
52  Datum datum;
53  bool isnull;
54 
56 
57  if (!HeapTupleIsValid(tup))
58  {
59  if (missing_ok)
60  return NULL;
61 
62  elog(ERROR, "cache lookup failed for subscription %u", subid);
63  }
64 
65  subform = (Form_pg_subscription) GETSTRUCT(tup);
66 
67  sub = (Subscription *) palloc(sizeof(Subscription));
68  sub->oid = subid;
69  sub->dbid = subform->subdbid;
70  sub->name = pstrdup(NameStr(subform->subname));
71  sub->owner = subform->subowner;
72  sub->enabled = subform->subenabled;
73 
74  /* Get conninfo */
76  tup,
78  &isnull);
79  Assert(!isnull);
80  sub->conninfo = TextDatumGetCString(datum);
81 
82  /* Get slotname */
84  tup,
86  &isnull);
87  if (!isnull)
88  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
89  else
90  sub->slotname = NULL;
91 
92  /* Get synccommit */
94  tup,
96  &isnull);
97  Assert(!isnull);
98  sub->synccommit = TextDatumGetCString(datum);
99 
100  /* Get publications */
102  tup,
104  &isnull);
105  Assert(!isnull);
107 
108  ReleaseSysCache(tup);
109 
110  return sub;
111 }
112 
113 /*
114  * Return number of subscriptions defined in given database.
115  * Used by dropdb() to check if database can indeed be dropped.
116  */
117 int
119 {
120  int nsubs = 0;
121  Relation rel;
122  ScanKeyData scankey;
123  SysScanDesc scan;
124  HeapTuple tup;
125 
127 
128  ScanKeyInit(&scankey,
130  BTEqualStrategyNumber, F_OIDEQ,
131  ObjectIdGetDatum(dbid));
132 
133  scan = systable_beginscan(rel, InvalidOid, false,
134  NULL, 1, &scankey);
135 
136  while (HeapTupleIsValid(tup = systable_getnext(scan)))
137  nsubs++;
138 
139  systable_endscan(scan);
140 
141  heap_close(rel, NoLock);
142 
143  return nsubs;
144 }
145 
146 /*
147  * Free memory allocated by subscription struct.
148  */
149 void
151 {
152  pfree(sub->name);
153  pfree(sub->conninfo);
154  if (sub->slotname)
155  pfree(sub->slotname);
157  pfree(sub);
158 }
159 
160 /*
161  * get_subscription_oid - given a subscription name, look up the OID
162  *
163  * If missing_ok is false, throw an error if name not found. If true, just
164  * return InvalidOid.
165  */
166 Oid
167 get_subscription_oid(const char *subname, bool missing_ok)
168 {
169  Oid oid;
170 
172  CStringGetDatum(subname));
173  if (!OidIsValid(oid) && !missing_ok)
174  ereport(ERROR,
175  (errcode(ERRCODE_UNDEFINED_OBJECT),
176  errmsg("subscription \"%s\" does not exist", subname)));
177  return oid;
178 }
179 
180 /*
181  * get_subscription_name - given a subscription OID, look up the name
182  */
183 char *
185 {
186  HeapTuple tup;
187  char *subname;
188  Form_pg_subscription subform;
189 
191 
192  if (!HeapTupleIsValid(tup))
193  elog(ERROR, "cache lookup failed for subscription %u", subid);
194 
195  subform = (Form_pg_subscription) GETSTRUCT(tup);
196  subname = pstrdup(NameStr(subform->subname));
197 
198  ReleaseSysCache(tup);
199 
200  return subname;
201 }
202 
203 /*
204  * Convert text array to list of strings.
205  *
206  * Note: the resulting list of strings is pallocated here.
207  */
208 static List *
210 {
211  Datum *elems;
212  int nelems,
213  i;
214  List *res = NIL;
215 
216  deconstruct_array(textarray,
217  TEXTOID, -1, false, 'i',
218  &elems, NULL, &nelems);
219 
220  if (nelems == 0)
221  return NIL;
222 
223  for (i = 0; i < nelems; i++)
224  res = lappend(res, makeString(TextDatumGetCString(elems[i])));
225 
226  return res;
227 }
228 
229 /*
230  * Set the state of a subscription table.
231  *
232  * If update_only is true and the record for given table doesn't exist, do
233  * nothing. This can be used to avoid inserting a new record that was deleted
234  * by someone else. Generally, subscription DDL commands should use false,
235  * workers should use true.
236  *
237  * The insert-or-update logic in this function is not concurrency safe so it
238  * might raise an error in rare circumstances. But if we took a stronger lock
239  * such as ShareRowExclusiveLock, we would risk more deadlocks.
240  */
241 Oid
243  XLogRecPtr sublsn, bool update_only)
244 {
245  Relation rel;
246  HeapTuple tup;
247  Oid subrelid = InvalidOid;
248  bool nulls[Natts_pg_subscription_rel];
250 
252 
254 
255  /* Try finding existing mapping. */
257  ObjectIdGetDatum(relid),
258  ObjectIdGetDatum(subid));
259 
260  /*
261  * If the record for given table does not exist yet create new record,
262  * otherwise update the existing one.
263  */
264  if (!HeapTupleIsValid(tup) && !update_only)
265  {
266  /* Form the tuple. */
267  memset(values, 0, sizeof(values));
268  memset(nulls, false, sizeof(nulls));
272  if (sublsn != InvalidXLogRecPtr)
273  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
274  else
275  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
276 
277  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
278 
279  /* Insert tuple into catalog. */
280  subrelid = CatalogTupleInsert(rel, tup);
281 
282  heap_freetuple(tup);
283  }
284  else if (HeapTupleIsValid(tup))
285  {
286  bool replaces[Natts_pg_subscription_rel];
287 
288  /* Update the tuple. */
289  memset(values, 0, sizeof(values));
290  memset(nulls, false, sizeof(nulls));
291  memset(replaces, false, sizeof(replaces));
292 
293  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
295 
296  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
297  if (sublsn != InvalidXLogRecPtr)
298  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
299  else
300  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
301 
302  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
303  replaces);
304 
305  /* Update the catalog. */
306  CatalogTupleUpdate(rel, &tup->t_self, tup);
307 
308  subrelid = HeapTupleGetOid(tup);
309  }
310 
311  /* Cleanup. */
312  heap_close(rel, NoLock);
313 
314  return subrelid;
315 }
316 
317 /*
318  * Get state of subscription table.
319  *
320  * Returns SUBREL_STATE_UNKNOWN when not found and missing_ok is true.
321  */
322 char
324  bool missing_ok)
325 {
326  Relation rel;
327  HeapTuple tup;
328  char substate;
329  bool isnull;
330  Datum d;
331 
333 
334  /* Try finding the mapping. */
336  ObjectIdGetDatum(relid),
337  ObjectIdGetDatum(subid));
338 
339  if (!HeapTupleIsValid(tup))
340  {
341  if (missing_ok)
342  {
344  *sublsn = InvalidXLogRecPtr;
345  return SUBREL_STATE_UNKNOWN;
346  }
347 
348  elog(ERROR, "subscription table %u in subscription %u does not exist",
349  relid, subid);
350  }
351 
352  /* Get the state. */
355  Assert(!isnull);
356  substate = DatumGetChar(d);
359  if (isnull)
360  *sublsn = InvalidXLogRecPtr;
361  else
362  *sublsn = DatumGetLSN(d);
363 
364  /* Cleanup */
365  ReleaseSysCache(tup);
367 
368  return substate;
369 }
370 
371 /*
372  * Drop subscription relation mapping. These can be for a particular
373  * subscription, or for a particular relation, or both.
374  */
375 void
377 {
378  Relation rel;
379  HeapScanDesc scan;
380  ScanKeyData skey[2];
381  HeapTuple tup;
382  int nkeys = 0;
383 
385 
386  if (OidIsValid(subid))
387  {
388  ScanKeyInit(&skey[nkeys++],
391  F_OIDEQ,
392  ObjectIdGetDatum(subid));
393  }
394 
395  if (OidIsValid(relid))
396  {
397  ScanKeyInit(&skey[nkeys++],
400  F_OIDEQ,
401  ObjectIdGetDatum(relid));
402  }
403 
404  /* Do the search and delete what we found. */
405  scan = heap_beginscan_catalog(rel, nkeys, skey);
407  {
408  CatalogTupleDelete(rel, &tup->t_self);
409  }
410  heap_endscan(scan);
411 
413 }
414 
415 
416 /*
417  * Get all relations for subscription.
418  *
419  * Returned list is palloc'ed in current memory context.
420  */
421 List *
423 {
424  List *res = NIL;
425  Relation rel;
426  HeapTuple tup;
427  int nkeys = 0;
428  ScanKeyData skey[2];
429  SysScanDesc scan;
430 
432 
433  ScanKeyInit(&skey[nkeys++],
435  BTEqualStrategyNumber, F_OIDEQ,
436  ObjectIdGetDatum(subid));
437 
438  scan = systable_beginscan(rel, InvalidOid, false,
439  NULL, nkeys, skey);
440 
441  while (HeapTupleIsValid(tup = systable_getnext(scan)))
442  {
444  SubscriptionRelState *relstate;
445 
446  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
447 
448  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
449  relstate->relid = subrel->srrelid;
450  relstate->state = subrel->srsubstate;
451  relstate->lsn = subrel->srsublsn;
452 
453  res = lappend(res, relstate);
454  }
455 
456  /* Cleanup */
457  systable_endscan(scan);
459 
460  return res;
461 }
462 
463 /*
464  * Get all relations for subscription that are not in a ready state.
465  *
466  * Returned list is palloc'ed in current memory context.
467  */
468 List *
470 {
471  List *res = NIL;
472  Relation rel;
473  HeapTuple tup;
474  int nkeys = 0;
475  ScanKeyData skey[2];
476  SysScanDesc scan;
477 
479 
480  ScanKeyInit(&skey[nkeys++],
482  BTEqualStrategyNumber, F_OIDEQ,
483  ObjectIdGetDatum(subid));
484 
485  ScanKeyInit(&skey[nkeys++],
487  BTEqualStrategyNumber, F_CHARNE,
489 
490  scan = systable_beginscan(rel, InvalidOid, false,
491  NULL, nkeys, skey);
492 
493  while (HeapTupleIsValid(tup = systable_getnext(scan)))
494  {
496  SubscriptionRelState *relstate;
497 
498  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
499 
500  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
501  relstate->relid = subrel->srrelid;
502  relstate->state = subrel->srsubstate;
503  relstate->lsn = subrel->srsublsn;
504 
505  res = lappend(res, relstate);
506  }
507 
508  /* Cleanup */
509  systable_endscan(scan);
511 
512  return res;
513 }
Value * makeString(char *str)
Definition: value.c:53
#define NIL
Definition: pg_list.h:69
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:499
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
#define Anum_pg_subscription_rel_srsubid
#define Anum_pg_subscription_subpublications
void heap_endscan(HeapScanDesc scan)
Definition: heapam.c:1565
void RemoveSubscriptionRel(Oid subid, Oid relid)
#define RelationGetDescr(relation)
Definition: rel.h:428
#define SubscriptionRelRelationId
#define TEXTOID
Definition: pg_type.h:324
char * pstrdup(const char *in)
Definition: mcxt.c:1076
int CountDBSubscriptions(Oid dbid)
#define AccessShareLock
Definition: lockdefs.h:36
int errcode(int sqlerrcode)
Definition: elog.c:575
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:255
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:695
#define heap_close(r, l)
Definition: heapam.h:97
FormData_pg_subscription * Form_pg_subscription
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1373
unsigned int Oid
Definition: postgres_ext.h:31
#define OidIsValid(objectId)
Definition: c.h:532
void list_free_deep(List *list)
Definition: list.c:1147
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:328
#define SearchSysCache1(cacheId, key1)
Definition: syscache.h:159
#define GetSysCacheOid2(cacheId, key1, key2)
Definition: syscache.h:188
#define DatumGetName(X)
Definition: postgres.h:591
Subscription * GetSubscription(Oid subid, bool missing_ok)
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:416
void pfree(void *pointer)
Definition: mcxt.c:949
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
#define ERROR
Definition: elog.h:43
Oid CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:162
static List * textarray_to_stringlist(ArrayType *textarray)
ItemPointerData t_self
Definition: htup.h:65
#define SUBREL_STATE_UNKNOWN
#define SubscriptionRelationId
#define NoLock
Definition: lockdefs.h:34
#define Anum_pg_subscription_rel_srsublsn
#define RowExclusiveLock
Definition: lockdefs.h:38
List * GetSubscriptionRelations(Oid subid)
#define CStringGetDatum(X)
Definition: postgres.h:584
List * publications
HeapScanDesc heap_beginscan_catalog(Relation relation, int nkeys, ScanKey key)
Definition: heapam.c:1405
#define ereport(elevel, rest)
Definition: elog.h:122
#define Natts_pg_subscription_rel
#define Anum_pg_subscription_subslotname
List * lappend(List *list, void *datum)
Definition: list.c:128
#define Anum_pg_subscription_subdbid
#define TextDatumGetCString(d)
Definition: builtins.h:92
Oid SetSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn, bool update_only)
uintptr_t Datum
Definition: postgres.h:372
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1117
#define DatumGetChar(X)
Definition: postgres.h:415
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1325
Oid MyDatabaseId
Definition: globals.c:77
HeapTuple heap_getnext(HeapScanDesc scan, ScanDirection direction)
Definition: heapam.c:1808
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1290
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:871
#define InvalidOid
Definition: postgres_ext.h:36
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define Anum_pg_subscription_subconninfo
Oid get_subscription_oid(const char *subname, bool missing_ok)
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
FormData_pg_subscription_rel * Form_pg_subscription_rel
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:664
Definition: regguts.h:298
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:210
char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn, bool missing_ok)
#define CharGetDatum(X)
Definition: postgres.h:422
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:3476
static Datum values[MAXATTR]
Definition: bootstrap.c:164
void * palloc(Size size)
Definition: mcxt.c:848
int errmsg(const char *fmt,...)
Definition: elog.c:797
int i
#define NameStr(name)
Definition: c.h:493
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define Anum_pg_subscription_rel_srsubstate
#define elog
Definition: elog.h:219
#define HeapTupleGetOid(tuple)
Definition: htup_details.h:695
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:170
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:794
Definition: pg_list.h:45
List * GetSubscriptionNotReadyRelations(Oid subid)
#define SUBREL_STATE_READY
#define Anum_pg_subscription_subsynccommit
#define BTEqualStrategyNumber
Definition: stratnum.h:31
#define Anum_pg_subscription_rel_srrelid
char * get_subscription_name(Oid subid)
#define DatumGetArrayTypeP(X)
Definition: array.h:242
#define SearchSysCache2(cacheId, key1, key2)
Definition: syscache.h:161