PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
pg_subscription.c File Reference
#include "postgres.h"
#include "miscadmin.h"
#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/xact.h"
#include "catalog/indexing.h"
#include "catalog/pg_type.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "nodes/makefuncs.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/syscache.h"
Include dependency graph for pg_subscription.c:

Go to the source code of this file.

Functions

static Listtextarray_to_stringlist (ArrayType *textarray)
 
SubscriptionGetSubscription (Oid subid, bool missing_ok)
 
int CountDBSubscriptions (Oid dbid)
 
void FreeSubscription (Subscription *sub)
 
Oid get_subscription_oid (const char *subname, bool missing_ok)
 
char * get_subscription_name (Oid subid)
 
Oid SetSubscriptionRelState (Oid subid, Oid relid, char state, XLogRecPtr sublsn)
 
char GetSubscriptionRelState (Oid subid, Oid relid, XLogRecPtr *sublsn, bool missing_ok)
 
void RemoveSubscriptionRel (Oid subid, Oid relid)
 
ListGetSubscriptionRelations (Oid subid)
 
ListGetSubscriptionNotReadyRelations (Oid subid)
 

Function Documentation

int CountDBSubscriptions ( Oid  dbid)

Definition at line 116 of file pg_subscription.c.

References Anum_pg_subscription_subdbid, BTEqualStrategyNumber, heap_close, heap_open(), HeapTupleIsValid, InvalidOid, NoLock, NULL, ObjectIdGetDatum, RowExclusiveLock, ScanKeyInit(), SubscriptionRelationId, systable_beginscan(), systable_endscan(), and systable_getnext().

Referenced by dropdb().

117 {
118  int nsubs = 0;
119  Relation rel;
120  ScanKeyData scankey;
121  SysScanDesc scan;
122  HeapTuple tup;
123 
125 
126  ScanKeyInit(&scankey,
128  BTEqualStrategyNumber, F_OIDEQ,
129  ObjectIdGetDatum(dbid));
130 
131  scan = systable_beginscan(rel, InvalidOid, false,
132  NULL, 1, &scankey);
133 
134  while (HeapTupleIsValid(tup = systable_getnext(scan)))
135  nsubs++;
136 
137  systable_endscan(scan);
138 
139  heap_close(rel, NoLock);
140 
141  return nsubs;
142 }
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:499
#define heap_close(r, l)
Definition: heapam.h:97
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:328
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:416
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
#define SubscriptionRelationId
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
#define Anum_pg_subscription_subdbid
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1284
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
#define NULL
Definition: c.h:229
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define BTEqualStrategyNumber
Definition: stratnum.h:31
void FreeSubscription ( Subscription sub)

Definition at line 148 of file pg_subscription.c.

References Subscription::conninfo, list_free_deep(), Subscription::name, pfree(), Subscription::publications, and Subscription::slotname.

Referenced by reread_subscription().

149 {
150  pfree(sub->name);
151  pfree(sub->conninfo);
152  if (sub->slotname)
153  pfree(sub->slotname);
155  pfree(sub);
156 }
void list_free_deep(List *list)
Definition: list.c:1147
void pfree(void *pointer)
Definition: mcxt.c:950
List * publications
char* get_subscription_name ( Oid  subid)

Definition at line 182 of file pg_subscription.c.

References elog, ERROR, GETSTRUCT, HeapTupleIsValid, NameStr, ObjectIdGetDatum, pstrdup(), ReleaseSysCache(), SearchSysCache1, and SUBSCRIPTIONOID.

Referenced by getObjectDescription(), and getObjectIdentityParts().

183 {
184  HeapTuple tup;
185  char *subname;
186  Form_pg_subscription subform;
187 
189 
190  if (!HeapTupleIsValid(tup))
191  elog(ERROR, "cache lookup failed for subscription %u", subid);
192 
193  subform = (Form_pg_subscription) GETSTRUCT(tup);
194  subname = pstrdup(NameStr(subform->subname));
195 
196  ReleaseSysCache(tup);
197 
198  return subname;
199 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
char * pstrdup(const char *in)
Definition: mcxt.c:1077
FormData_pg_subscription * Form_pg_subscription
#define SearchSysCache1(cacheId, key1)
Definition: syscache.h:156
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
#define ERROR
Definition: elog.h:43
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1117
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
#define NameStr(name)
Definition: c.h:499
#define elog
Definition: elog.h:219
Oid get_subscription_oid ( const char *  subname,
bool  missing_ok 
)

Definition at line 165 of file pg_subscription.c.

References CStringGetDatum, ereport, errcode(), errmsg(), ERROR, GetSysCacheOid2, MyDatabaseId, OidIsValid, and SUBSCRIPTIONNAME.

Referenced by get_object_address_unqualified().

166 {
167  Oid oid;
168 
170  CStringGetDatum(subname));
171  if (!OidIsValid(oid) && !missing_ok)
172  ereport(ERROR,
173  (errcode(ERRCODE_UNDEFINED_OBJECT),
174  errmsg("subscription \"%s\" does not exist", subname)));
175  return oid;
176 }
int errcode(int sqlerrcode)
Definition: elog.c:575
unsigned int Oid
Definition: postgres_ext.h:31
#define OidIsValid(objectId)
Definition: c.h:538
#define GetSysCacheOid2(cacheId, key1, key2)
Definition: syscache.h:185
#define ERROR
Definition: elog.h:43
#define CStringGetDatum(X)
Definition: postgres.h:584
#define ereport(elevel, rest)
Definition: elog.h:122
Oid MyDatabaseId
Definition: globals.c:76
int errmsg(const char *fmt,...)
Definition: elog.c:797
Subscription* GetSubscription ( Oid  subid,
bool  missing_ok 
)

Definition at line 45 of file pg_subscription.c.

References Anum_pg_subscription_subconninfo, Anum_pg_subscription_subpublications, Anum_pg_subscription_subslotname, Anum_pg_subscription_subsynccommit, Assert, Subscription::conninfo, DatumGetArrayTypeP, DatumGetName, Subscription::dbid, elog, Subscription::enabled, ERROR, GETSTRUCT, HeapTupleIsValid, Subscription::name, NameStr, NULL, ObjectIdGetDatum, Subscription::oid, Subscription::owner, palloc(), pstrdup(), Subscription::publications, ReleaseSysCache(), SearchSysCache1, Subscription::slotname, SUBSCRIPTIONOID, Subscription::synccommit, SysCacheGetAttr(), textarray_to_stringlist(), and TextDatumGetCString.

Referenced by AlterSubscription(), ApplyWorkerMain(), and reread_subscription().

46 {
47  HeapTuple tup;
48  Subscription *sub;
49  Form_pg_subscription subform;
50  Datum datum;
51  bool isnull;
52 
54 
55  if (!HeapTupleIsValid(tup))
56  {
57  if (missing_ok)
58  return NULL;
59 
60  elog(ERROR, "cache lookup failed for subscription %u", subid);
61  }
62 
63  subform = (Form_pg_subscription) GETSTRUCT(tup);
64 
65  sub = (Subscription *) palloc(sizeof(Subscription));
66  sub->oid = subid;
67  sub->dbid = subform->subdbid;
68  sub->name = pstrdup(NameStr(subform->subname));
69  sub->owner = subform->subowner;
70  sub->enabled = subform->subenabled;
71 
72  /* Get conninfo */
74  tup,
76  &isnull);
77  Assert(!isnull);
78  sub->conninfo = TextDatumGetCString(datum);
79 
80  /* Get slotname */
82  tup,
84  &isnull);
85  if (!isnull)
86  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
87  else
88  sub->slotname = NULL;
89 
90  /* Get synccommit */
92  tup,
94  &isnull);
95  Assert(!isnull);
96  sub->synccommit = TextDatumGetCString(datum);
97 
98  /* Get publications */
100  tup,
102  &isnull);
103  Assert(!isnull);
105 
106  ReleaseSysCache(tup);
107 
108  return sub;
109 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
#define Anum_pg_subscription_subpublications
char * pstrdup(const char *in)
Definition: mcxt.c:1077
FormData_pg_subscription * Form_pg_subscription
#define SearchSysCache1(cacheId, key1)
Definition: syscache.h:156
#define DatumGetName(X)
Definition: postgres.h:591
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
#define ERROR
Definition: elog.h:43
static List * textarray_to_stringlist(ArrayType *textarray)
List * publications
#define Anum_pg_subscription_subslotname
#define TextDatumGetCString(d)
Definition: builtins.h:92
uintptr_t Datum
Definition: postgres.h:372
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1117
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1279
#define Anum_pg_subscription_subconninfo
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
void * palloc(Size size)
Definition: mcxt.c:849
#define NameStr(name)
Definition: c.h:499
#define elog
Definition: elog.h:219
#define Anum_pg_subscription_subsynccommit
#define DatumGetArrayTypeP(X)
Definition: array.h:242
List* GetSubscriptionNotReadyRelations ( Oid  subid)

Definition at line 460 of file pg_subscription.c.

References AccessShareLock, Anum_pg_subscription_rel_srsubid, Anum_pg_subscription_rel_srsubstate, BTEqualStrategyNumber, CharGetDatum, GETSTRUCT, heap_close, heap_open(), HeapTupleIsValid, InvalidOid, lappend(), SubscriptionRelState::lsn, NIL, NULL, ObjectIdGetDatum, palloc(), SubscriptionRelState::relid, ScanKeyInit(), SubscriptionRelState::state, SUBREL_STATE_READY, SubscriptionRelRelationId, systable_beginscan(), systable_endscan(), and systable_getnext().

Referenced by process_syncing_tables_for_apply().

461 {
462  List *res = NIL;
463  Relation rel;
464  HeapTuple tup;
465  int nkeys = 0;
466  ScanKeyData skey[2];
467  SysScanDesc scan;
468 
470 
471  ScanKeyInit(&skey[nkeys++],
473  BTEqualStrategyNumber, F_OIDEQ,
474  ObjectIdGetDatum(subid));
475 
476  ScanKeyInit(&skey[nkeys++],
478  BTEqualStrategyNumber, F_CHARNE,
480 
481  scan = systable_beginscan(rel, InvalidOid, false,
482  NULL, nkeys, skey);
483 
484  while (HeapTupleIsValid(tup = systable_getnext(scan)))
485  {
487  SubscriptionRelState *relstate;
488 
489  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
490 
491  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
492  relstate->relid = subrel->srrelid;
493  relstate->state = subrel->srsubstate;
494  relstate->lsn = subrel->srsublsn;
495 
496  res = lappend(res, relstate);
497  }
498 
499  /* Cleanup */
500  systable_endscan(scan);
502 
503  return res;
504 }
#define NIL
Definition: pg_list.h:69
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:499
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
#define Anum_pg_subscription_rel_srsubid
#define SubscriptionRelRelationId
#define AccessShareLock
Definition: lockdefs.h:36
#define heap_close(r, l)
Definition: heapam.h:97
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:328
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:416
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
List * lappend(List *list, void *datum)
Definition: list.c:128
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1284
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
#define NULL
Definition: c.h:229
FormData_pg_subscription_rel * Form_pg_subscription_rel
#define CharGetDatum(X)
Definition: postgres.h:422
void * palloc(Size size)
Definition: mcxt.c:849
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define Anum_pg_subscription_rel_srsubstate
Definition: pg_list.h:45
#define SUBREL_STATE_READY
#define BTEqualStrategyNumber
Definition: stratnum.h:31
List* GetSubscriptionRelations ( Oid  subid)

Definition at line 413 of file pg_subscription.c.

References AccessShareLock, Anum_pg_subscription_rel_srsubid, BTEqualStrategyNumber, GETSTRUCT, heap_close, heap_open(), HeapTupleIsValid, InvalidOid, lappend(), SubscriptionRelState::lsn, NIL, NULL, ObjectIdGetDatum, palloc(), SubscriptionRelState::relid, ScanKeyInit(), SubscriptionRelState::state, SubscriptionRelRelationId, systable_beginscan(), systable_endscan(), and systable_getnext().

Referenced by AlterSubscription_refresh().

414 {
415  List *res = NIL;
416  Relation rel;
417  HeapTuple tup;
418  int nkeys = 0;
419  ScanKeyData skey[2];
420  SysScanDesc scan;
421 
423 
424  ScanKeyInit(&skey[nkeys++],
426  BTEqualStrategyNumber, F_OIDEQ,
427  ObjectIdGetDatum(subid));
428 
429  scan = systable_beginscan(rel, InvalidOid, false,
430  NULL, nkeys, skey);
431 
432  while (HeapTupleIsValid(tup = systable_getnext(scan)))
433  {
435  SubscriptionRelState *relstate;
436 
437  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
438 
439  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
440  relstate->relid = subrel->srrelid;
441  relstate->state = subrel->srsubstate;
442  relstate->lsn = subrel->srsublsn;
443 
444  res = lappend(res, relstate);
445  }
446 
447  /* Cleanup */
448  systable_endscan(scan);
450 
451  return res;
452 }
#define NIL
Definition: pg_list.h:69
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:499
#define GETSTRUCT(TUP)
Definition: htup_details.h:656
#define Anum_pg_subscription_rel_srsubid
#define SubscriptionRelRelationId
#define AccessShareLock
Definition: lockdefs.h:36
#define heap_close(r, l)
Definition: heapam.h:97
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:328
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:416
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
List * lappend(List *list, void *datum)
Definition: list.c:128
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1284
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
#define NULL
Definition: c.h:229
FormData_pg_subscription_rel * Form_pg_subscription_rel
void * palloc(Size size)
Definition: mcxt.c:849
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Definition: pg_list.h:45
#define BTEqualStrategyNumber
Definition: stratnum.h:31
char GetSubscriptionRelState ( Oid  subid,
Oid  relid,
XLogRecPtr sublsn,
bool  missing_ok 
)

Definition at line 314 of file pg_subscription.c.

References AccessShareLock, Anum_pg_subscription_rel_srsublsn, Anum_pg_subscription_rel_srsubstate, Assert, DatumGetChar, DatumGetLSN, elog, ERROR, heap_close, heap_open(), HeapTupleIsValid, InvalidXLogRecPtr, ObjectIdGetDatum, ReleaseSysCache(), SearchSysCache2, SUBREL_STATE_UNKNOWN, SUBSCRIPTIONRELMAP, SubscriptionRelRelationId, and SysCacheGetAttr().

Referenced by logicalrep_rel_open(), and LogicalRepSyncTableStart().

316 {
317  Relation rel;
318  HeapTuple tup;
319  char substate;
320  bool isnull;
321  Datum d;
322 
324 
325  /* Try finding the mapping. */
327  ObjectIdGetDatum(relid),
328  ObjectIdGetDatum(subid));
329 
330  if (!HeapTupleIsValid(tup))
331  {
332  if (missing_ok)
333  {
335  *sublsn = InvalidXLogRecPtr;
336  return SUBREL_STATE_UNKNOWN;
337  }
338 
339  elog(ERROR, "subscription table %u in subscription %u does not exist",
340  relid, subid);
341  }
342 
343  /* Get the state. */
346  Assert(!isnull);
347  substate = DatumGetChar(d);
350  if (isnull)
351  *sublsn = InvalidXLogRecPtr;
352  else
353  *sublsn = DatumGetLSN(d);
354 
355  /* Cleanup */
356  ReleaseSysCache(tup);
358 
359  return substate;
360 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define SubscriptionRelRelationId
#define AccessShareLock
Definition: lockdefs.h:36
#define heap_close(r, l)
Definition: heapam.h:97
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
#define ERROR
Definition: elog.h:43
#define SUBREL_STATE_UNKNOWN
#define Anum_pg_subscription_rel_srsublsn
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:1279
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1284
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
#define Assert(condition)
Definition: c.h:675
#define Anum_pg_subscription_rel_srsubstate
#define elog
Definition: elog.h:219
#define SearchSysCache2(cacheId, key1, key2)
Definition: syscache.h:158
void RemoveSubscriptionRel ( Oid  subid,
Oid  relid 
)

Definition at line 367 of file pg_subscription.c.

References Anum_pg_subscription_rel_srrelid, Anum_pg_subscription_rel_srsubid, BTEqualStrategyNumber, ForwardScanDirection, heap_beginscan_catalog(), heap_close, heap_endscan(), heap_getnext(), heap_open(), HeapTupleIsValid, ObjectIdGetDatum, OidIsValid, RowExclusiveLock, ScanKeyInit(), simple_heap_delete(), SubscriptionRelRelationId, and HeapTupleData::t_self.

Referenced by AlterSubscription_refresh(), DropSubscription(), and heap_drop_with_catalog().

368 {
369  Relation rel;
370  HeapScanDesc scan;
371  ScanKeyData skey[2];
372  HeapTuple tup;
373  int nkeys = 0;
374 
376 
377  if (OidIsValid(subid))
378  {
379  ScanKeyInit(&skey[nkeys++],
382  F_OIDEQ,
383  ObjectIdGetDatum(subid));
384  }
385 
386  if (OidIsValid(relid))
387  {
388  ScanKeyInit(&skey[nkeys++],
391  F_OIDEQ,
392  ObjectIdGetDatum(relid));
393  }
394 
395  /* Do the search and delete what we found. */
396  scan = heap_beginscan_catalog(rel, nkeys, skey);
398  {
399  simple_heap_delete(rel, &tup->t_self);
400  }
401  heap_endscan(scan);
402 
404 }
#define Anum_pg_subscription_rel_srsubid
void heap_endscan(HeapScanDesc scan)
Definition: heapam.c:1578
#define SubscriptionRelRelationId
#define heap_close(r, l)
Definition: heapam.h:97
#define OidIsValid(objectId)
Definition: c.h:538
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
ItemPointerData t_self
Definition: htup.h:65
#define RowExclusiveLock
Definition: lockdefs.h:38
HeapScanDesc heap_beginscan_catalog(Relation relation, int nkeys, ScanKey key)
Definition: heapam.c:1399
HeapTuple heap_getnext(HeapScanDesc scan, ScanDirection direction)
Definition: heapam.c:1794
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1284
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
void simple_heap_delete(Relation relation, ItemPointer tid)
Definition: heapam.c:3398
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define BTEqualStrategyNumber
Definition: stratnum.h:31
#define Anum_pg_subscription_rel_srrelid
Oid SetSubscriptionRelState ( Oid  subid,
Oid  relid,
char  state,
XLogRecPtr  sublsn 
)

Definition at line 235 of file pg_subscription.c.

References Anum_pg_subscription_rel_srrelid, Anum_pg_subscription_rel_srsubid, Anum_pg_subscription_rel_srsublsn, Anum_pg_subscription_rel_srsubstate, CatalogTupleInsert(), CatalogTupleUpdate(), CharGetDatum, heap_close, heap_form_tuple(), heap_freetuple(), heap_modify_tuple(), heap_open(), HeapTupleGetOid, HeapTupleIsValid, InvalidXLogRecPtr, LSNGetDatum, Natts_pg_subscription_rel, NoLock, ObjectIdGetDatum, RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy2, SUBSCRIPTIONRELMAP, SubscriptionRelRelationId, HeapTupleData::t_self, and values.

Referenced by AlterSubscription_refresh(), CreateSubscription(), LogicalRepSyncTableStart(), process_syncing_tables_for_apply(), and process_syncing_tables_for_sync().

237 {
238  Relation rel;
239  HeapTuple tup;
240  Oid subrelid;
241  bool nulls[Natts_pg_subscription_rel];
243 
245 
246  /* Try finding existing mapping. */
248  ObjectIdGetDatum(relid),
249  ObjectIdGetDatum(subid));
250 
251  /*
252  * If the record for given table does not exist yet create new record,
253  * otherwise update the existing one.
254  */
255  if (!HeapTupleIsValid(tup))
256  {
257  /* Form the tuple. */
258  memset(values, 0, sizeof(values));
259  memset(nulls, false, sizeof(nulls));
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  subrelid = CatalogTupleInsert(rel, tup);
272 
273  heap_freetuple(tup);
274  }
275  else
276  {
277  bool replaces[Natts_pg_subscription_rel];
278 
279  /* Update the tuple. */
280  memset(values, 0, sizeof(values));
281  memset(nulls, false, sizeof(nulls));
282  memset(replaces, false, sizeof(replaces));
283 
284  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
286 
287  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
288  if (sublsn != InvalidXLogRecPtr)
289  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
290  else
291  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
292 
293  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
294  replaces);
295 
296  /* Update the catalog. */
297  CatalogTupleUpdate(rel, &tup->t_self, tup);
298 
299  subrelid = HeapTupleGetOid(tup);
300  }
301 
302  /* Cleanup. */
303  heap_close(rel, NoLock);
304 
305  return subrelid;
306 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define Anum_pg_subscription_rel_srsubid
#define RelationGetDescr(relation)
Definition: rel.h:429
#define SubscriptionRelRelationId
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:692
#define heap_close(r, l)
Definition: heapam.h:97
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1372
unsigned int Oid
Definition: postgres_ext.h:31
#define ObjectIdGetDatum(X)
Definition: postgres.h:513
Oid CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:162
ItemPointerData t_self
Definition: htup.h:65
#define NoLock
Definition: lockdefs.h:34
#define Anum_pg_subscription_rel_srsublsn
#define RowExclusiveLock
Definition: lockdefs.h:38
#define Natts_pg_subscription_rel
uintptr_t Datum
Definition: postgres.h:372
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1284
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
Definition: regguts.h:298
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:210
#define CharGetDatum(X)
Definition: postgres.h:422
static Datum values[MAXATTR]
Definition: bootstrap.c:163
#define Anum_pg_subscription_rel_srsubstate
#define HeapTupleGetOid(tuple)
Definition: htup_details.h:695
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:167
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:791
#define Anum_pg_subscription_rel_srrelid
static List * textarray_to_stringlist ( ArrayType textarray)
static

Definition at line 207 of file pg_subscription.c.

References deconstruct_array(), i, lappend(), makeString(), NIL, NULL, TextDatumGetCString, and TEXTOID.

Referenced by GetSubscription().

208 {
209  Datum *elems;
210  int nelems,
211  i;
212  List *res = NIL;
213 
214  deconstruct_array(textarray,
215  TEXTOID, -1, false, 'i',
216  &elems, NULL, &nelems);
217 
218  if (nelems == 0)
219  return NIL;
220 
221  for (i = 0; i < nelems; i++)
222  res = lappend(res, makeString(TextDatumGetCString(elems[i])));
223 
224  return res;
225 }
Value * makeString(char *str)
Definition: value.c:53
#define NIL
Definition: pg_list.h:69
#define TEXTOID
Definition: pg_type.h:324
List * lappend(List *list, void *datum)
Definition: list.c:128
#define TextDatumGetCString(d)
Definition: builtins.h:92
uintptr_t Datum
Definition: postgres.h:372
#define NULL
Definition: c.h:229
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3475
int i
Definition: pg_list.h:45