PostgreSQL Source Code  git master
pg_subscription.c File Reference
#include "postgres.h"
#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/indexing.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "storage/lmgr.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, bool missing_ok)
 
void AddSubscriptionRelState (Oid subid, Oid relid, char state, XLogRecPtr sublsn)
 
void UpdateSubscriptionRelState (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

◆ AddSubscriptionRelState()

void AddSubscriptionRelState ( Oid  subid,
Oid  relid,
char  state,
XLogRecPtr  sublsn 
)

Definition at line 236 of file pg_subscription.c.

References AccessShareLock, CatalogTupleInsert(), CharGetDatum, elog, ERROR, heap_form_tuple(), heap_freetuple(), HeapTupleIsValid, InvalidXLogRecPtr, LockSharedObject(), LSNGetDatum, NoLock, ObjectIdGetDatum, RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy2, SUBSCRIPTIONRELMAP, table_close(), table_open(), and values.

Referenced by AlterSubscription_refresh(), and CreateSubscription().

238 {
239  Relation rel;
240  HeapTuple tup;
241  bool nulls[Natts_pg_subscription_rel];
242  Datum values[Natts_pg_subscription_rel];
243 
244  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
245 
246  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
247 
248  /* Try finding existing mapping. */
250  ObjectIdGetDatum(relid),
251  ObjectIdGetDatum(subid));
252  if (HeapTupleIsValid(tup))
253  elog(ERROR, "subscription table %u in subscription %u already exists",
254  relid, subid);
255 
256  /* Form the tuple. */
257  memset(values, 0, sizeof(values));
258  memset(nulls, false, sizeof(nulls));
259  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
260  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
261  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
262  if (sublsn != InvalidXLogRecPtr)
263  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
264  else
265  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
266 
267  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
268 
269  /* Insert tuple into catalog. */
270  CatalogTupleInsert(rel, tup);
271 
272  heap_freetuple(tup);
273 
274  /* Cleanup. */
275  table_close(rel, NoLock);
276 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
#define RelationGetDescr(relation)
Definition: rel.h:482
#define AccessShareLock
Definition: lockdefs.h:36
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1338
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
uintptr_t Datum
Definition: postgres.h:367
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1002
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
Definition: regguts.h:298
#define CharGetDatum(X)
Definition: postgres.h:416
static Datum values[MAXATTR]
Definition: bootstrap.c:167
#define elog(elevel,...)
Definition: elog.h:214
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:176
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
void CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:221

◆ CountDBSubscriptions()

int CountDBSubscriptions ( Oid  dbid)

Definition at line 114 of file pg_subscription.c.

References BTEqualStrategyNumber, HeapTupleIsValid, InvalidOid, NoLock, ObjectIdGetDatum, RowExclusiveLock, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by dropdb().

115 {
116  int nsubs = 0;
117  Relation rel;
118  ScanKeyData scankey;
119  SysScanDesc scan;
120  HeapTuple tup;
121 
122  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
123 
124  ScanKeyInit(&scankey,
125  Anum_pg_subscription_subdbid,
126  BTEqualStrategyNumber, F_OIDEQ,
127  ObjectIdGetDatum(dbid));
128 
129  scan = systable_beginscan(rel, InvalidOid, false,
130  NULL, 1, &scankey);
131 
132  while (HeapTupleIsValid(tup = systable_getnext(scan)))
133  nsubs++;
134 
135  systable_endscan(scan);
136 
137  table_close(rel, NoLock);
138 
139  return nsubs;
140 }
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:529
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:356
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:448
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ FreeSubscription()

void FreeSubscription ( Subscription sub)

Definition at line 146 of file pg_subscription.c.

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

Referenced by maybe_reread_subscription().

147 {
148  pfree(sub->name);
149  pfree(sub->conninfo);
150  if (sub->slotname)
151  pfree(sub->slotname);
153  pfree(sub);
154 }
void list_free_deep(List *list)
Definition: list.c:1390
void pfree(void *pointer)
Definition: mcxt.c:1056
List * publications

◆ get_subscription_name()

char* get_subscription_name ( Oid  subid,
bool  missing_ok 
)

Definition at line 183 of file pg_subscription.c.

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

Referenced by getObjectDescription(), and getObjectIdentityParts().

184 {
185  HeapTuple tup;
186  char *subname;
187  Form_pg_subscription subform;
188 
190 
191  if (!HeapTupleIsValid(tup))
192  {
193  if (!missing_ok)
194  elog(ERROR, "cache lookup failed for subscription %u", subid);
195  return NULL;
196  }
197 
198  subform = (Form_pg_subscription) GETSTRUCT(tup);
199  subname = pstrdup(NameStr(subform->subname));
200 
201  ReleaseSysCache(tup);
202 
203  return subname;
204 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
char * pstrdup(const char *in)
Definition: mcxt.c:1186
FormData_pg_subscription * Form_pg_subscription
NameData subname
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1116
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1164
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define elog(elevel,...)
Definition: elog.h:214
#define NameStr(name)
Definition: c.h:622

◆ get_subscription_oid()

Oid get_subscription_oid ( const char *  subname,
bool  missing_ok 
)

Definition at line 163 of file pg_subscription.c.

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

Referenced by get_object_address_unqualified().

164 {
165  Oid oid;
166 
167  oid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
169  if (!OidIsValid(oid) && !missing_ok)
170  ereport(ERROR,
171  (errcode(ERRCODE_UNDEFINED_OBJECT),
172  errmsg("subscription \"%s\" does not exist", subname)));
173  return oid;
174 }
int errcode(int sqlerrcode)
Definition: elog.c:610
unsigned int Oid
Definition: postgres_ext.h:31
NameData subname
#define OidIsValid(objectId)
Definition: c.h:651
#define ERROR
Definition: elog.h:43
#define CStringGetDatum(X)
Definition: postgres.h:578
Oid MyDatabaseId
Definition: globals.c:85
#define ereport(elevel,...)
Definition: elog.h:144
#define GetSysCacheOid2(cacheId, oidcol, key1, key2)
Definition: syscache.h:194
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ GetSubscription()

Subscription* GetSubscription ( Oid  subid,
bool  missing_ok 
)

Definition at line 42 of file pg_subscription.c.

References Assert, Subscription::binary, Subscription::conninfo, DatumGetArrayTypeP, DatumGetName, Subscription::dbid, elog, Subscription::enabled, ERROR, GETSTRUCT, HeapTupleIsValid, Subscription::name, NameStr, 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 maybe_reread_subscription().

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 
70  /* Get conninfo */
72  tup,
73  Anum_pg_subscription_subconninfo,
74  &isnull);
75  Assert(!isnull);
76  sub->conninfo = TextDatumGetCString(datum);
77 
78  /* Get slotname */
80  tup,
81  Anum_pg_subscription_subslotname,
82  &isnull);
83  if (!isnull)
84  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
85  else
86  sub->slotname = NULL;
87 
88  /* Get synccommit */
90  tup,
91  Anum_pg_subscription_subsynccommit,
92  &isnull);
93  Assert(!isnull);
94  sub->synccommit = TextDatumGetCString(datum);
95 
96  /* Get publications */
98  tup,
99  Anum_pg_subscription_subpublications,
100  &isnull);
101  Assert(!isnull);
103 
104  ReleaseSysCache(tup);
105 
106  return sub;
107 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
char * pstrdup(const char *in)
Definition: mcxt.c:1186
FormData_pg_subscription * Form_pg_subscription
#define DatumGetName(X)
Definition: postgres.h:585
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
static List * textarray_to_stringlist(ArrayType *textarray)
List * publications
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1116
#define TextDatumGetCString(d)
Definition: builtins.h:88
uintptr_t Datum
Definition: postgres.h:367
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1164
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1377
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define Assert(condition)
Definition: c.h:745
void * palloc(Size size)
Definition: mcxt.c:949
#define elog(elevel,...)
Definition: elog.h:214
#define NameStr(name)
Definition: c.h:622
#define DatumGetArrayTypeP(X)
Definition: array.h:249

◆ GetSubscriptionNotReadyRelations()

List* GetSubscriptionNotReadyRelations ( Oid  subid)

Definition at line 486 of file pg_subscription.c.

References AccessShareLock, BTEqualStrategyNumber, CharGetDatum, DatumGetLSN, GETSTRUCT, HeapTupleIsValid, InvalidOid, InvalidXLogRecPtr, lappend(), SubscriptionRelState::lsn, NIL, ObjectIdGetDatum, palloc(), SubscriptionRelState::relid, ScanKeyInit(), SubscriptionRelState::state, SUBSCRIPTIONRELMAP, SysCacheGetAttr(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by process_syncing_tables_for_apply().

487 {
488  List *res = NIL;
489  Relation rel;
490  HeapTuple tup;
491  int nkeys = 0;
492  ScanKeyData skey[2];
493  SysScanDesc scan;
494 
495  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
496 
497  ScanKeyInit(&skey[nkeys++],
498  Anum_pg_subscription_rel_srsubid,
499  BTEqualStrategyNumber, F_OIDEQ,
500  ObjectIdGetDatum(subid));
501 
502  ScanKeyInit(&skey[nkeys++],
503  Anum_pg_subscription_rel_srsubstate,
504  BTEqualStrategyNumber, F_CHARNE,
505  CharGetDatum(SUBREL_STATE_READY));
506 
507  scan = systable_beginscan(rel, InvalidOid, false,
508  NULL, nkeys, skey);
509 
510  while (HeapTupleIsValid(tup = systable_getnext(scan)))
511  {
513  SubscriptionRelState *relstate;
514  Datum d;
515  bool isnull;
516 
517  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
518 
519  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
520  relstate->relid = subrel->srrelid;
521  relstate->state = subrel->srsubstate;
523  Anum_pg_subscription_rel_srsublsn, &isnull);
524  if (isnull)
525  relstate->lsn = InvalidXLogRecPtr;
526  else
527  relstate->lsn = DatumGetLSN(d);
528 
529  res = lappend(res, relstate);
530  }
531 
532  /* Cleanup */
533  systable_endscan(scan);
535 
536  return res;
537 }
#define NIL
Definition: pg_list.h:65
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:529
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
#define AccessShareLock
Definition: lockdefs.h:36
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:356
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:448
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
List * lappend(List *list, void *datum)
Definition: list.c:321
uintptr_t Datum
Definition: postgres.h:367
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1377
#define InvalidOid
Definition: postgres_ext.h:36
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
#define CharGetDatum(X)
Definition: postgres.h:416
void * palloc(Size size)
Definition: mcxt.c:949
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
Definition: pg_list.h:50
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ GetSubscriptionRelations()

List* GetSubscriptionRelations ( Oid  subid)

Definition at line 432 of file pg_subscription.c.

References AccessShareLock, BTEqualStrategyNumber, DatumGetLSN, GETSTRUCT, HeapTupleIsValid, InvalidOid, InvalidXLogRecPtr, lappend(), SubscriptionRelState::lsn, NIL, ObjectIdGetDatum, palloc(), SubscriptionRelState::relid, ScanKeyInit(), SubscriptionRelState::state, SUBSCRIPTIONRELMAP, SysCacheGetAttr(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by AlterSubscription_refresh().

433 {
434  List *res = NIL;
435  Relation rel;
436  HeapTuple tup;
437  int nkeys = 0;
438  ScanKeyData skey[2];
439  SysScanDesc scan;
440 
441  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
442 
443  ScanKeyInit(&skey[nkeys++],
444  Anum_pg_subscription_rel_srsubid,
445  BTEqualStrategyNumber, F_OIDEQ,
446  ObjectIdGetDatum(subid));
447 
448  scan = systable_beginscan(rel, InvalidOid, false,
449  NULL, nkeys, skey);
450 
451  while (HeapTupleIsValid(tup = systable_getnext(scan)))
452  {
454  SubscriptionRelState *relstate;
455  Datum d;
456  bool isnull;
457 
458  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
459 
460  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
461  relstate->relid = subrel->srrelid;
462  relstate->state = subrel->srsubstate;
464  Anum_pg_subscription_rel_srsublsn, &isnull);
465  if (isnull)
466  relstate->lsn = InvalidXLogRecPtr;
467  else
468  relstate->lsn = DatumGetLSN(d);
469 
470  res = lappend(res, relstate);
471  }
472 
473  /* Cleanup */
474  systable_endscan(scan);
476 
477  return res;
478 }
#define NIL
Definition: pg_list.h:65
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:529
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
#define AccessShareLock
Definition: lockdefs.h:36
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:356
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:448
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
List * lappend(List *list, void *datum)
Definition: list.c:321
uintptr_t Datum
Definition: postgres.h:367
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1377
#define InvalidOid
Definition: postgres_ext.h:36
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
void * palloc(Size size)
Definition: mcxt.c:949
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
Definition: pg_list.h:50
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ GetSubscriptionRelState()

char GetSubscriptionRelState ( Oid  subid,
Oid  relid,
XLogRecPtr sublsn,
bool  missing_ok 
)

Definition at line 333 of file pg_subscription.c.

References AccessShareLock, Assert, DatumGetChar, DatumGetLSN, elog, ERROR, HeapTupleIsValid, InvalidXLogRecPtr, ObjectIdGetDatum, ReleaseSysCache(), SearchSysCache2(), SUBSCRIPTIONRELMAP, SysCacheGetAttr(), table_close(), and table_open().

Referenced by logicalrep_rel_open(), LogicalRepSyncTableStart(), and wait_for_relation_state_change().

335 {
336  Relation rel;
337  HeapTuple tup;
338  char substate;
339  bool isnull;
340  Datum d;
341 
342  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
343 
344  /* Try finding the mapping. */
346  ObjectIdGetDatum(relid),
347  ObjectIdGetDatum(subid));
348 
349  if (!HeapTupleIsValid(tup))
350  {
351  if (missing_ok)
352  {
354  *sublsn = InvalidXLogRecPtr;
355  return SUBREL_STATE_UNKNOWN;
356  }
357 
358  elog(ERROR, "subscription table %u in subscription %u does not exist",
359  relid, subid);
360  }
361 
362  /* Get the state. */
364  Anum_pg_subscription_rel_srsubstate, &isnull);
365  Assert(!isnull);
366  substate = DatumGetChar(d);
368  Anum_pg_subscription_rel_srsublsn, &isnull);
369  if (isnull)
370  *sublsn = InvalidXLogRecPtr;
371  else
372  *sublsn = DatumGetLSN(d);
373 
374  /* Cleanup */
375  ReleaseSysCache(tup);
377 
378  return substate;
379 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
#define AccessShareLock
Definition: lockdefs.h:36
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
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
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define Assert(condition)
Definition: c.h:745
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:1127
#define elog(elevel,...)
Definition: elog.h:214
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39

◆ RemoveSubscriptionRel()

void RemoveSubscriptionRel ( Oid  subid,
Oid  relid 
)

Definition at line 386 of file pg_subscription.c.

References BTEqualStrategyNumber, CatalogTupleDelete(), ForwardScanDirection, heap_getnext(), HeapTupleIsValid, ObjectIdGetDatum, OidIsValid, RowExclusiveLock, ScanKeyInit(), HeapTupleData::t_self, table_beginscan_catalog(), table_close(), table_endscan(), and table_open().

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

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  {
418  CatalogTupleDelete(rel, &tup->t_self);
419  }
420  table_endscan(scan);
421 
423 }
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:112
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:350
#define OidIsValid(objectId)
Definition: c.h:651
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
ItemPointerData t_self
Definition: htup.h:65
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1286
#define RowExclusiveLock
Definition: lockdefs.h:38
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:862
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ textarray_to_stringlist()

static List * textarray_to_stringlist ( ArrayType textarray)
static

Definition at line 212 of file pg_subscription.c.

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

Referenced by GetSubscription().

213 {
214  Datum *elems;
215  int nelems,
216  i;
217  List *res = NIL;
218 
219  deconstruct_array(textarray,
220  TEXTOID, -1, false, TYPALIGN_INT,
221  &elems, NULL, &nelems);
222 
223  if (nelems == 0)
224  return NIL;
225 
226  for (i = 0; i < nelems; i++)
227  res = lappend(res, makeString(TextDatumGetCString(elems[i])));
228 
229  return res;
230 }
Value * makeString(char *str)
Definition: value.c:53
#define NIL
Definition: pg_list.h:65
List * lappend(List *list, void *datum)
Definition: list.c:321
#define TextDatumGetCString(d)
Definition: builtins.h:88
uintptr_t Datum
Definition: postgres.h:367
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3483
int i
Definition: pg_list.h:50

◆ UpdateSubscriptionRelState()

void UpdateSubscriptionRelState ( Oid  subid,
Oid  relid,
char  state,
XLogRecPtr  sublsn 
)

Definition at line 282 of file pg_subscription.c.

References AccessShareLock, CatalogTupleUpdate(), CharGetDatum, elog, ERROR, heap_modify_tuple(), HeapTupleIsValid, InvalidXLogRecPtr, LockSharedObject(), LSNGetDatum, NoLock, ObjectIdGetDatum, RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy2, SUBSCRIPTIONRELMAP, HeapTupleData::t_self, table_close(), table_open(), and values.

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

284 {
285  Relation rel;
286  HeapTuple tup;
287  bool nulls[Natts_pg_subscription_rel];
288  Datum values[Natts_pg_subscription_rel];
289  bool replaces[Natts_pg_subscription_rel];
290 
291  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
292 
293  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
294 
295  /* Try finding existing mapping. */
297  ObjectIdGetDatum(relid),
298  ObjectIdGetDatum(subid));
299  if (!HeapTupleIsValid(tup))
300  elog(ERROR, "subscription table %u in subscription %u does not exist",
301  relid, subid);
302 
303  /* Update the tuple. */
304  memset(values, 0, sizeof(values));
305  memset(nulls, false, sizeof(nulls));
306  memset(replaces, false, sizeof(replaces));
307 
308  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
309  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
310 
311  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
312  if (sublsn != InvalidXLogRecPtr)
313  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
314  else
315  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
316 
317  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
318  replaces);
319 
320  /* Update the catalog. */
321  CatalogTupleUpdate(rel, &tup->t_self, tup);
322 
323  /* Cleanup. */
324  table_close(rel, NoLock);
325 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
#define RelationGetDescr(relation)
Definition: rel.h:482
#define AccessShareLock
Definition: lockdefs.h:36
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
ItemPointerData t_self
Definition: htup.h:65
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
uintptr_t Datum
Definition: postgres.h:367
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1002
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
Definition: regguts.h:298
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:301
#define CharGetDatum(X)
Definition: postgres.h:416
static Datum values[MAXATTR]
Definition: bootstrap.c:167
#define elog(elevel,...)
Definition: elog.h:214
#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