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/lsyscache.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)
 
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 238 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().

240 {
241  Relation rel;
242  HeapTuple tup;
243  bool nulls[Natts_pg_subscription_rel];
244  Datum values[Natts_pg_subscription_rel];
245 
246  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
247 
248  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
249 
250  /* Try finding existing mapping. */
252  ObjectIdGetDatum(relid),
253  ObjectIdGetDatum(subid));
254  if (HeapTupleIsValid(tup))
255  elog(ERROR, "subscription table %u in subscription %u already exists",
256  relid, subid);
257 
258  /* Form the tuple. */
259  memset(values, 0, sizeof(values));
260  memset(nulls, false, sizeof(nulls));
261  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
262  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
263  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
264  if (sublsn != InvalidXLogRecPtr)
265  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
266  else
267  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
268 
269  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
270 
271  /* Insert tuple into catalog. */
272  CatalogTupleInsert(rel, tup);
273 
274  heap_freetuple(tup);
275 
276  /* Cleanup. */
277  table_close(rel, NoLock);
278 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:167
#define RelationGetDescr(relation)
Definition: rel.h:483
#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:551
#define ERROR
Definition: elog.h:46
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
uintptr_t Datum
Definition: postgres.h:411
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1017
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
Definition: regguts.h:317
#define CharGetDatum(X)
Definition: postgres.h:460
static Datum values[MAXATTR]
Definition: bootstrap.c:166
#define elog(elevel,...)
Definition: elog.h:232
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:177
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 116 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().

117 {
118  int nsubs = 0;
119  Relation rel;
120  ScanKeyData scankey;
121  SysScanDesc scan;
122  HeapTuple tup;
123 
124  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
125 
126  ScanKeyInit(&scankey,
127  Anum_pg_subscription_subdbid,
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  table_close(rel, NoLock);
140 
141  return nsubs;
142 }
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:167
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:595
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:383
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:502
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
#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 148 of file pg_subscription.c.

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

Referenced by maybe_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:1405
void pfree(void *pointer)
Definition: mcxt.c:1169
List * publications

◆ get_subscription_name()

char* get_subscription_name ( Oid  subid,
bool  missing_ok 
)

Definition at line 185 of file pg_subscription.c.

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

Referenced by getObjectDescription(), getObjectIdentityParts(), and RemoveSubscriptionRel().

186 {
187  HeapTuple tup;
188  char *subname;
189  Form_pg_subscription subform;
190 
192 
193  if (!HeapTupleIsValid(tup))
194  {
195  if (!missing_ok)
196  elog(ERROR, "cache lookup failed for subscription %u", subid);
197  return NULL;
198  }
199 
200  subform = (Form_pg_subscription) GETSTRUCT(tup);
201  subname = pstrdup(NameStr(subform->subname));
202 
203  ReleaseSysCache(tup);
204 
205  return subname;
206 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:654
char * pstrdup(const char *in)
Definition: mcxt.c:1299
FormData_pg_subscription * Form_pg_subscription
NameData subname
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
#define ERROR
Definition: elog.h:46
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1127
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1175
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define elog(elevel,...)
Definition: elog.h:232
#define NameStr(name)
Definition: c.h:681

◆ get_subscription_oid()

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 
169  oid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
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:698
unsigned int Oid
Definition: postgres_ext.h:31
NameData subname
#define OidIsValid(objectId)
Definition: c.h:710
#define ERROR
Definition: elog.h:46
#define CStringGetDatum(X)
Definition: postgres.h:622
Oid MyDatabaseId
Definition: globals.c:88
#define ereport(elevel,...)
Definition: elog.h:157
#define GetSysCacheOid2(cacheId, oidcol, key1, key2)
Definition: syscache.h:195
int errmsg(const char *fmt,...)
Definition: elog.c:909

◆ GetSubscription()

Subscription* GetSubscription ( Oid  subid,
bool  missing_ok 
)

Definition at line 43 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, Subscription::stream, SUBSCRIPTIONOID, Subscription::synccommit, SysCacheGetAttr(), textarray_to_stringlist(), and TextDatumGetCString.

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

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 
72  /* Get conninfo */
74  tup,
75  Anum_pg_subscription_subconninfo,
76  &isnull);
77  Assert(!isnull);
78  sub->conninfo = TextDatumGetCString(datum);
79 
80  /* Get slotname */
82  tup,
83  Anum_pg_subscription_subslotname,
84  &isnull);
85  if (!isnull)
86  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
87  else
88  sub->slotname = NULL;
89 
90  /* Get synccommit */
92  tup,
93  Anum_pg_subscription_subsynccommit,
94  &isnull);
95  Assert(!isnull);
96  sub->synccommit = TextDatumGetCString(datum);
97 
98  /* Get publications */
100  tup,
101  Anum_pg_subscription_subpublications,
102  &isnull);
103  Assert(!isnull);
105 
106  ReleaseSysCache(tup);
107 
108  return sub;
109 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:654
char * pstrdup(const char *in)
Definition: mcxt.c:1299
FormData_pg_subscription * Form_pg_subscription
#define DatumGetName(X)
Definition: postgres.h:629
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
#define ERROR
Definition: elog.h:46
static List * textarray_to_stringlist(ArrayType *textarray)
List * publications
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1127
#define TextDatumGetCString(d)
Definition: builtins.h:83
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
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define Assert(condition)
Definition: c.h:804
void * palloc(Size size)
Definition: mcxt.c:1062
#define elog(elevel,...)
Definition: elog.h:232
#define NameStr(name)
Definition: c.h:681
#define DatumGetArrayTypeP(X)
Definition: array.h:254

◆ GetSubscriptionNotReadyRelations()

List* GetSubscriptionNotReadyRelations ( Oid  subid)

Definition at line 513 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 DropSubscription(), and process_syncing_tables_for_apply().

514 {
515  List *res = NIL;
516  Relation rel;
517  HeapTuple tup;
518  int nkeys = 0;
519  ScanKeyData skey[2];
520  SysScanDesc scan;
521 
522  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
523 
524  ScanKeyInit(&skey[nkeys++],
525  Anum_pg_subscription_rel_srsubid,
526  BTEqualStrategyNumber, F_OIDEQ,
527  ObjectIdGetDatum(subid));
528 
529  ScanKeyInit(&skey[nkeys++],
530  Anum_pg_subscription_rel_srsubstate,
531  BTEqualStrategyNumber, F_CHARNE,
532  CharGetDatum(SUBREL_STATE_READY));
533 
534  scan = systable_beginscan(rel, InvalidOid, false,
535  NULL, nkeys, skey);
536 
537  while (HeapTupleIsValid(tup = systable_getnext(scan)))
538  {
540  SubscriptionRelState *relstate;
541  Datum d;
542  bool isnull;
543 
544  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
545 
546  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
547  relstate->relid = subrel->srrelid;
548  relstate->state = subrel->srsubstate;
550  Anum_pg_subscription_rel_srsublsn, &isnull);
551  if (isnull)
552  relstate->lsn = InvalidXLogRecPtr;
553  else
554  relstate->lsn = DatumGetLSN(d);
555 
556  res = lappend(res, relstate);
557  }
558 
559  /* Cleanup */
560  systable_endscan(scan);
562 
563  return res;
564 }
#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:595
#define GETSTRUCT(TUP)
Definition: htup_details.h:654
#define AccessShareLock
Definition: lockdefs.h:36
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:383
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:502
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
List * lappend(List *list, void *datum)
Definition: list.c:336
uintptr_t Datum
Definition: postgres.h:411
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1388
#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:460
void * palloc(Size size)
Definition: mcxt.c:1062
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 459 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().

460 {
461  List *res = NIL;
462  Relation rel;
463  HeapTuple tup;
464  int nkeys = 0;
465  ScanKeyData skey[2];
466  SysScanDesc scan;
467 
468  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
469 
470  ScanKeyInit(&skey[nkeys++],
471  Anum_pg_subscription_rel_srsubid,
472  BTEqualStrategyNumber, F_OIDEQ,
473  ObjectIdGetDatum(subid));
474 
475  scan = systable_beginscan(rel, InvalidOid, false,
476  NULL, nkeys, skey);
477 
478  while (HeapTupleIsValid(tup = systable_getnext(scan)))
479  {
481  SubscriptionRelState *relstate;
482  Datum d;
483  bool isnull;
484 
485  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
486 
487  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
488  relstate->relid = subrel->srrelid;
489  relstate->state = subrel->srsubstate;
491  Anum_pg_subscription_rel_srsublsn, &isnull);
492  if (isnull)
493  relstate->lsn = InvalidXLogRecPtr;
494  else
495  relstate->lsn = DatumGetLSN(d);
496 
497  res = lappend(res, relstate);
498  }
499 
500  /* Cleanup */
501  systable_endscan(scan);
503 
504  return res;
505 }
#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:595
#define GETSTRUCT(TUP)
Definition: htup_details.h:654
#define AccessShareLock
Definition: lockdefs.h:36
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:383
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:502
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
List * lappend(List *list, void *datum)
Definition: list.c:336
uintptr_t Datum
Definition: postgres.h:411
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1388
#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:1062
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 
)

Definition at line 335 of file pg_subscription.c.

References AccessShareLock, DatumGetLSN, GETSTRUCT, HeapTupleIsValid, InvalidXLogRecPtr, ObjectIdGetDatum, ReleaseSysCache(), SearchSysCache2(), SUBSCRIPTIONRELMAP, SysCacheGetAttr(), table_close(), and table_open().

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

336 {
337  HeapTuple tup;
338  char substate;
339  bool isnull;
340  Datum d;
341  Relation rel;
342 
343  /*
344  * This is to avoid the race condition with AlterSubscription which tries
345  * to remove this relstate.
346  */
347  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
348 
349  /* Try finding the mapping. */
351  ObjectIdGetDatum(relid),
352  ObjectIdGetDatum(subid));
353 
354  if (!HeapTupleIsValid(tup))
355  {
357  *sublsn = InvalidXLogRecPtr;
358  return SUBREL_STATE_UNKNOWN;
359  }
360 
361  /* Get the state. */
362  substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
363 
364  /* Get the LSN */
366  Anum_pg_subscription_rel_srsublsn, &isnull);
367  if (isnull)
368  *sublsn = InvalidXLogRecPtr;
369  else
370  *sublsn = DatumGetLSN(d);
371 
372  /* Cleanup */
373  ReleaseSysCache(tup);
374 
376 
377  return substate;
378 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:167
#define GETSTRUCT(TUP)
Definition: htup_details.h:654
#define AccessShareLock
Definition: lockdefs.h:36
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
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
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:1138
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39

◆ RemoveSubscriptionRel()

void RemoveSubscriptionRel ( Oid  subid,
Oid  relid 
)

Definition at line 385 of file pg_subscription.c.

References BTEqualStrategyNumber, CatalogTupleDelete(), ereport, errcode(), errdetail(), errhint(), errmsg(), ERROR, ForwardScanDirection, get_rel_name(), get_subscription_name(), GETSTRUCT, 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().

386 {
387  Relation rel;
388  TableScanDesc scan;
389  ScanKeyData skey[2];
390  HeapTuple tup;
391  int nkeys = 0;
392 
393  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
394 
395  if (OidIsValid(subid))
396  {
397  ScanKeyInit(&skey[nkeys++],
398  Anum_pg_subscription_rel_srsubid,
400  F_OIDEQ,
401  ObjectIdGetDatum(subid));
402  }
403 
404  if (OidIsValid(relid))
405  {
406  ScanKeyInit(&skey[nkeys++],
407  Anum_pg_subscription_rel_srrelid,
409  F_OIDEQ,
410  ObjectIdGetDatum(relid));
411  }
412 
413  /* Do the search and delete what we found. */
414  scan = table_beginscan_catalog(rel, nkeys, skey);
416  {
418 
419  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
420 
421  /*
422  * We don't allow to drop the relation mapping when the table
423  * synchronization is in progress unless the caller updates the
424  * corresponding subscription as well. This is to ensure that we don't
425  * leave tablesync slots or origins in the system when the
426  * corresponding table is dropped.
427  */
428  if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
429  {
430  ereport(ERROR,
431  (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
432  errmsg("could not drop relation mapping for subscription \"%s\"",
433  get_subscription_name(subrel->srsubid, false)),
434  errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
435  get_rel_name(relid), subrel->srsubstate),
436  /*
437  * translator: first %s is a SQL ALTER command and second %s is a
438  * SQL DROP command
439  */
440  errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
441  "ALTER SUBSCRIPTION ... ENABLE",
442  "DROP SUBSCRIPTION ...")));
443  }
444 
445  CatalogTupleDelete(rel, &tup->t_self);
446  }
447  table_endscan(scan);
448 
450 }
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:167
int errhint(const char *fmt,...)
Definition: elog.c:1156
#define GETSTRUCT(TUP)
Definition: htup_details.h:654
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:112
int errcode(int sqlerrcode)
Definition: elog.c:698
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:350
#define OidIsValid(objectId)
Definition: c.h:710
char * get_subscription_name(Oid subid, bool missing_ok)
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
#define ERROR
Definition: elog.h:46
ItemPointerData t_self
Definition: htup.h:65
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
#define ereport(elevel,...)
Definition: elog.h:157
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:991
int errmsg(const char *fmt,...)
Definition: elog.c:909
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
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1899
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ textarray_to_stringlist()

static List * textarray_to_stringlist ( ArrayType textarray)
static

Definition at line 214 of file pg_subscription.c.

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

Referenced by GetSubscription().

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

◆ UpdateSubscriptionRelState()

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

Definition at line 284 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().

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