PostgreSQL Source Code  git master
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 "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)
 
Oid AddSubscriptionRelState (Oid subid, Oid relid, char state, XLogRecPtr sublsn)
 
Oid 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()

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

Definition at line 233 of file pg_subscription.c.

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

Referenced by AlterSubscription_refresh(), and CreateSubscription().

235 {
236  Relation rel;
237  HeapTuple tup;
238  Oid subrelid;
239  bool nulls[Natts_pg_subscription_rel];
240  Datum values[Natts_pg_subscription_rel];
241 
242  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
243 
244  rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
245 
246  /* Try finding existing mapping. */
248  ObjectIdGetDatum(relid),
249  ObjectIdGetDatum(subid));
250  if (HeapTupleIsValid(tup))
251  elog(ERROR, "subscription table %u in subscription %u already exists",
252  relid, subid);
253 
254  /* Form the tuple. */
255  memset(values, 0, sizeof(values));
256  memset(nulls, false, sizeof(nulls));
257  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
258  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
259  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
260  if (sublsn != InvalidXLogRecPtr)
261  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
262  else
263  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
264 
265  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
266 
267  /* Insert tuple into catalog. */
268  subrelid = CatalogTupleInsert(rel, tup);
269 
270  heap_freetuple(tup);
271 
272  /* Cleanup. */
273  heap_close(rel, NoLock);
274 
275  return subrelid;
276 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define RelationGetDescr(relation)
Definition: rel.h:433
#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:1074
#define heap_close(r, l)
Definition: heapam.h:97
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1773
unsigned int Oid
Definition: postgres_ext.h:31
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
#define ERROR
Definition: elog.h:43
Oid CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:163
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
uintptr_t Datum
Definition: postgres.h:365
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1294
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:897
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
Definition: regguts.h:298
#define CharGetDatum(X)
Definition: postgres.h:399
static Datum values[MAXATTR]
Definition: bootstrap.c:164
#define elog
Definition: elog.h:219
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:175

◆ CountDBSubscriptions()

int CountDBSubscriptions ( Oid  dbid)

Definition at line 118 of file pg_subscription.c.

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

Referenced by dropdb().

119 {
120  int nsubs = 0;
121  Relation rel;
122  ScanKeyData scankey;
123  SysScanDesc scan;
124  HeapTuple tup;
125 
126  rel = heap_open(SubscriptionRelationId, RowExclusiveLock);
127 
128  ScanKeyInit(&scankey,
129  Anum_pg_subscription_subdbid,
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 }
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:502
#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:331
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:419
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1294
#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
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ FreeSubscription()

void FreeSubscription ( Subscription sub)

Definition at line 150 of file pg_subscription.c.

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

Referenced by maybe_reread_subscription().

151 {
152  pfree(sub->name);
153  pfree(sub->conninfo);
154  if (sub->slotname)
155  pfree(sub->slotname);
157  pfree(sub);
158 }
void list_free_deep(List *list)
Definition: list.c:1147
void pfree(void *pointer)
Definition: mcxt.c:1031
List * publications

◆ get_subscription_name()

char* get_subscription_name ( Oid  subid)

Definition at line 184 of file pg_subscription.c.

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

Referenced by getObjectDescription(), and getObjectIdentityParts().

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 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:673
char * pstrdup(const char *in)
Definition: mcxt.c:1161
FormData_pg_subscription * Form_pg_subscription
NameData subname
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
#define ERROR
Definition: elog.h:43
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1112
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1160
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define NameStr(name)
Definition: c.h:576
#define elog
Definition: elog.h:219

◆ get_subscription_oid()

Oid get_subscription_oid ( const char *  subname,
bool  missing_ok 
)

Definition at line 167 of file pg_subscription.c.

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

Referenced by get_object_address_unqualified().

168 {
169  Oid oid;
170 
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 }
int errcode(int sqlerrcode)
Definition: elog.c:575
unsigned int Oid
Definition: postgres_ext.h:31
NameData subname
#define OidIsValid(objectId)
Definition: c.h:605
#define GetSysCacheOid2(cacheId, key1, key2)
Definition: syscache.h:193
#define ERROR
Definition: elog.h:43
#define CStringGetDatum(X)
Definition: postgres.h:561
#define ereport(elevel, rest)
Definition: elog.h:122
Oid MyDatabaseId
Definition: globals.c:86
int errmsg(const char *fmt,...)
Definition: elog.c:797

◆ GetSubscription()

Subscription* GetSubscription ( Oid  subid,
bool  missing_ok 
)

Definition at line 47 of file pg_subscription.c.

References Assert, 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().

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,
77  Anum_pg_subscription_subconninfo,
78  &isnull);
79  Assert(!isnull);
80  sub->conninfo = TextDatumGetCString(datum);
81 
82  /* Get slotname */
84  tup,
85  Anum_pg_subscription_subslotname,
86  &isnull);
87  if (!isnull)
88  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
89  else
90  sub->slotname = NULL;
91 
92  /* Get synccommit */
94  tup,
95  Anum_pg_subscription_subsynccommit,
96  &isnull);
97  Assert(!isnull);
98  sub->synccommit = TextDatumGetCString(datum);
99 
100  /* Get publications */
102  tup,
103  Anum_pg_subscription_subpublications,
104  &isnull);
105  Assert(!isnull);
107 
108  ReleaseSysCache(tup);
109 
110  return sub;
111 }
#define GETSTRUCT(TUP)
Definition: htup_details.h:673
char * pstrdup(const char *in)
Definition: mcxt.c:1161
FormData_pg_subscription * Form_pg_subscription
#define DatumGetName(X)
Definition: postgres.h:568
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
#define ERROR
Definition: elog.h:43
static List * textarray_to_stringlist(ArrayType *textarray)
List * publications
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1112
#define TextDatumGetCString(d)
Definition: builtins.h:96
uintptr_t Datum
Definition: postgres.h:365
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1160
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1368
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define Assert(condition)
Definition: c.h:699
void * palloc(Size size)
Definition: mcxt.c:924
#define NameStr(name)
Definition: c.h:576
#define elog
Definition: elog.h:219
#define DatumGetArrayTypeP(X)
Definition: array.h:246

◆ GetSubscriptionNotReadyRelations()

List* GetSubscriptionNotReadyRelations ( Oid  subid)

Definition at line 484 of file pg_subscription.c.

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

Referenced by process_syncing_tables_for_apply().

485 {
486  List *res = NIL;
487  Relation rel;
488  HeapTuple tup;
489  int nkeys = 0;
490  ScanKeyData skey[2];
491  SysScanDesc scan;
492 
493  rel = heap_open(SubscriptionRelRelationId, AccessShareLock);
494 
495  ScanKeyInit(&skey[nkeys++],
496  Anum_pg_subscription_rel_srsubid,
497  BTEqualStrategyNumber, F_OIDEQ,
498  ObjectIdGetDatum(subid));
499 
500  ScanKeyInit(&skey[nkeys++],
501  Anum_pg_subscription_rel_srsubstate,
502  BTEqualStrategyNumber, F_CHARNE,
503  CharGetDatum(SUBREL_STATE_READY));
504 
505  scan = systable_beginscan(rel, InvalidOid, false,
506  NULL, nkeys, skey);
507 
508  while (HeapTupleIsValid(tup = systable_getnext(scan)))
509  {
511  SubscriptionRelState *relstate;
512 
513  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
514 
515  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
516  relstate->relid = subrel->srrelid;
517  relstate->state = subrel->srsubstate;
518  relstate->lsn = subrel->srsublsn;
519 
520  res = lappend(res, relstate);
521  }
522 
523  /* Cleanup */
524  systable_endscan(scan);
526 
527  return res;
528 }
#define NIL
Definition: pg_list.h:69
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:502
#define GETSTRUCT(TUP)
Definition: htup_details.h:673
#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:331
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:419
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
List * lappend(List *list, void *datum)
Definition: list.c:128
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1294
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
#define CharGetDatum(X)
Definition: postgres.h:399
void * palloc(Size size)
Definition: mcxt.c:924
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

◆ GetSubscriptionRelations()

List* GetSubscriptionRelations ( Oid  subid)

Definition at line 437 of file pg_subscription.c.

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

Referenced by AlterSubscription_refresh().

438 {
439  List *res = NIL;
440  Relation rel;
441  HeapTuple tup;
442  int nkeys = 0;
443  ScanKeyData skey[2];
444  SysScanDesc scan;
445 
446  rel = heap_open(SubscriptionRelRelationId, AccessShareLock);
447 
448  ScanKeyInit(&skey[nkeys++],
449  Anum_pg_subscription_rel_srsubid,
450  BTEqualStrategyNumber, F_OIDEQ,
451  ObjectIdGetDatum(subid));
452 
453  scan = systable_beginscan(rel, InvalidOid, false,
454  NULL, nkeys, skey);
455 
456  while (HeapTupleIsValid(tup = systable_getnext(scan)))
457  {
459  SubscriptionRelState *relstate;
460 
461  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
462 
463  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
464  relstate->relid = subrel->srrelid;
465  relstate->state = subrel->srsubstate;
466  relstate->lsn = subrel->srsublsn;
467 
468  res = lappend(res, relstate);
469  }
470 
471  /* Cleanup */
472  systable_endscan(scan);
474 
475  return res;
476 }
#define NIL
Definition: pg_list.h:69
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:502
#define GETSTRUCT(TUP)
Definition: htup_details.h:673
#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:331
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:419
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
List * lappend(List *list, void *datum)
Definition: list.c:128
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1294
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
void * palloc(Size size)
Definition: mcxt.c:924
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

◆ GetSubscriptionRelState()

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

Definition at line 338 of file pg_subscription.c.

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

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

340 {
341  Relation rel;
342  HeapTuple tup;
343  char substate;
344  bool isnull;
345  Datum d;
346 
347  rel = heap_open(SubscriptionRelRelationId, AccessShareLock);
348 
349  /* Try finding the mapping. */
351  ObjectIdGetDatum(relid),
352  ObjectIdGetDatum(subid));
353 
354  if (!HeapTupleIsValid(tup))
355  {
356  if (missing_ok)
357  {
359  *sublsn = InvalidXLogRecPtr;
360  return SUBREL_STATE_UNKNOWN;
361  }
362 
363  elog(ERROR, "subscription table %u in subscription %u does not exist",
364  relid, subid);
365  }
366 
367  /* Get the state. */
369  Anum_pg_subscription_rel_srsubstate, &isnull);
370  Assert(!isnull);
371  substate = DatumGetChar(d);
373  Anum_pg_subscription_rel_srsublsn, &isnull);
374  if (isnull)
375  *sublsn = InvalidXLogRecPtr;
376  else
377  *sublsn = DatumGetLSN(d);
378 
379  /* Cleanup */
380  ReleaseSysCache(tup);
382 
383  return substate;
384 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define AccessShareLock
Definition: lockdefs.h:36
#define heap_close(r, l)
Definition: heapam.h:97
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
#define ERROR
Definition: elog.h:43
uintptr_t Datum
Definition: postgres.h:365
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1160
#define DatumGetChar(X)
Definition: postgres.h:392
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1368
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1294
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define Assert(condition)
Definition: c.h:699
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:1123
#define elog
Definition: elog.h:219

◆ RemoveSubscriptionRel()

void RemoveSubscriptionRel ( Oid  subid,
Oid  relid 
)

Definition at line 391 of file pg_subscription.c.

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

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

392 {
393  Relation rel;
394  HeapScanDesc scan;
395  ScanKeyData skey[2];
396  HeapTuple tup;
397  int nkeys = 0;
398 
399  rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
400 
401  if (OidIsValid(subid))
402  {
403  ScanKeyInit(&skey[nkeys++],
404  Anum_pg_subscription_rel_srsubid,
406  F_OIDEQ,
407  ObjectIdGetDatum(subid));
408  }
409 
410  if (OidIsValid(relid))
411  {
412  ScanKeyInit(&skey[nkeys++],
413  Anum_pg_subscription_rel_srrelid,
415  F_OIDEQ,
416  ObjectIdGetDatum(relid));
417  }
418 
419  /* Do the search and delete what we found. */
420  scan = heap_beginscan_catalog(rel, nkeys, skey);
422  {
423  CatalogTupleDelete(rel, &tup->t_self);
424  }
425  heap_endscan(scan);
426 
428 }
void heap_endscan(HeapScanDesc scan)
Definition: heapam.c:1572
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:256
#define heap_close(r, l)
Definition: heapam.h:97
#define OidIsValid(objectId)
Definition: c.h:605
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
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:1412
HeapTuple heap_getnext(HeapScanDesc scan, ScanDirection direction)
Definition: heapam.c:1835
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1294
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ textarray_to_stringlist()

static List * textarray_to_stringlist ( ArrayType textarray)
static

Definition at line 209 of file pg_subscription.c.

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

Referenced by GetSubscription().

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 }
Value * makeString(char *str)
Definition: value.c:53
#define NIL
Definition: pg_list.h:69
List * lappend(List *list, void *datum)
Definition: list.c:128
#define TextDatumGetCString(d)
Definition: builtins.h:96
uintptr_t Datum
Definition: postgres.h:365
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3449
int i
Definition: pg_list.h:45

◆ UpdateSubscriptionRelState()

Oid 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_close, heap_modify_tuple(), heap_open(), HeapTupleGetOid, HeapTupleIsValid, InvalidXLogRecPtr, LockSharedObject(), LSNGetDatum, NoLock, ObjectIdGetDatum, RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy2, SUBSCRIPTIONRELMAP, HeapTupleData::t_self, and values.

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

284 {
285  Relation rel;
286  HeapTuple tup;
287  Oid subrelid;
288  bool nulls[Natts_pg_subscription_rel];
289  Datum values[Natts_pg_subscription_rel];
290  bool replaces[Natts_pg_subscription_rel];
291 
292  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
293 
294  rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
295 
296  /* Try finding existing mapping. */
298  ObjectIdGetDatum(relid),
299  ObjectIdGetDatum(subid));
300  if (!HeapTupleIsValid(tup))
301  elog(ERROR, "subscription table %u in subscription %u does not exist",
302  relid, subid);
303 
304  /* Update the tuple. */
305  memset(values, 0, sizeof(values));
306  memset(nulls, false, sizeof(nulls));
307  memset(replaces, false, sizeof(replaces));
308 
309  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
310  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
311 
312  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
313  if (sublsn != InvalidXLogRecPtr)
314  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
315  else
316  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
317 
318  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
319  replaces);
320 
321  /* Update the catalog. */
322  CatalogTupleUpdate(rel, &tup->t_self, tup);
323 
324  subrelid = HeapTupleGetOid(tup);
325 
326  /* Cleanup. */
327  heap_close(rel, NoLock);
328 
329  return subrelid;
330 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define RelationGetDescr(relation)
Definition: rel.h:433
#define AccessShareLock
Definition: lockdefs.h:36
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
#define heap_close(r, l)
Definition: heapam.h:97
unsigned int Oid
Definition: postgres_ext.h:31
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
#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:365
Relation heap_open(Oid relationId, LOCKMODE lockmode)
Definition: heapam.c:1294
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:897
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
Definition: regguts.h:298
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:211
#define CharGetDatum(X)
Definition: postgres.h:399
static Datum values[MAXATTR]
Definition: bootstrap.c:164
#define elog
Definition: elog.h:219
#define HeapTupleGetOid(tuple)
Definition: htup_details.h:712
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:175
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:1173