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 "catalog/indexing.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "miscadmin.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)
 
void DisableSubscription (Oid subid)
 
void AddSubscriptionRelState (Oid subid, Oid relid, char state, XLogRecPtr sublsn, bool retain_lock)
 
void UpdateSubscriptionRelState (Oid subid, Oid relid, char state, XLogRecPtr sublsn)
 
char GetSubscriptionRelState (Oid subid, Oid relid, XLogRecPtr *sublsn)
 
void RemoveSubscriptionRel (Oid subid, Oid relid)
 
bool HasSubscriptionRelations (Oid subid)
 
ListGetSubscriptionRelations (Oid subid, bool not_ready)
 

Function Documentation

◆ AddSubscriptionRelState()

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

Definition at line 236 of file pg_subscription.c.

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. */
249  tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
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  if (retain_lock)
276  {
277  table_close(rel, NoLock);
278  }
279  else
280  {
282  UnlockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
283  }
284 }
static Datum values[MAXATTR]
Definition: bootstrap.c:152
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:224
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, const Datum *values, const bool *isnull)
Definition: heaptuple.c:1116
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1434
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
void CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:233
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1083
void UnlockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1142
#define NoLock
Definition: lockdefs.h:34
#define AccessShareLock
Definition: lockdefs.h:36
#define RowExclusiveLock
Definition: lockdefs.h:38
static Datum LSNGetDatum(XLogRecPtr X)
Definition: pg_lsn.h:28
uintptr_t Datum
Definition: postgres.h:64
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
static Datum CharGetDatum(char X)
Definition: postgres.h:122
#define RelationGetDescr(relation)
Definition: rel.h:531
Definition: regguts.h:323
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:88
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:126
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:40
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

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

Referenced by AlterSubscription_refresh(), binary_upgrade_add_sub_rel_state(), and CreateSubscription().

◆ CountDBSubscriptions()

int CountDBSubscriptions ( Oid  dbid)

Definition at line 123 of file pg_subscription.c.

124 {
125  int nsubs = 0;
126  Relation rel;
127  ScanKeyData scankey;
128  SysScanDesc scan;
129  HeapTuple tup;
130 
131  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
132 
133  ScanKeyInit(&scankey,
134  Anum_pg_subscription_subdbid,
135  BTEqualStrategyNumber, F_OIDEQ,
136  ObjectIdGetDatum(dbid));
137 
138  scan = systable_beginscan(rel, InvalidOid, false,
139  NULL, 1, &scankey);
140 
141  while (HeapTupleIsValid(tup = systable_getnext(scan)))
142  nsubs++;
143 
144  systable_endscan(scan);
145 
146  table_close(rel, NoLock);
147 
148  return nsubs;
149 }
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:596
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:503
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:384
#define InvalidOid
Definition: postgres_ext.h:36
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
#define BTEqualStrategyNumber
Definition: stratnum.h:31

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

Referenced by dropdb().

◆ DisableSubscription()

void DisableSubscription ( Oid  subid)

Definition at line 169 of file pg_subscription.c.

170 {
171  Relation rel;
172  bool nulls[Natts_pg_subscription];
173  bool replaces[Natts_pg_subscription];
174  Datum values[Natts_pg_subscription];
175  HeapTuple tup;
176 
177  /* Look up the subscription in the catalog */
178  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
179  tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
180 
181  if (!HeapTupleIsValid(tup))
182  elog(ERROR, "cache lookup failed for subscription %u", subid);
183 
184  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
185 
186  /* Form a new tuple. */
187  memset(values, 0, sizeof(values));
188  memset(nulls, false, sizeof(nulls));
189  memset(replaces, false, sizeof(replaces));
190 
191  /* Set the subscription to disabled. */
192  values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
193  replaces[Anum_pg_subscription_subenabled - 1] = true;
194 
195  /* Update the catalog */
196  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
197  replaces);
198  CatalogTupleUpdate(rel, &tup->t_self, tup);
199  heap_freetuple(tup);
200 
201  table_close(rel, NoLock);
202 }
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, const Datum *replValues, const bool *replIsnull, const bool *doReplace)
Definition: heaptuple.c:1209
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:313
static Datum BoolGetDatum(bool X)
Definition: postgres.h:102
ItemPointerData t_self
Definition: htup.h:65
#define SearchSysCacheCopy1(cacheId, key1)
Definition: syscache.h:86

References AccessShareLock, BoolGetDatum(), CatalogTupleUpdate(), elog, ERROR, heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, LockSharedObject(), NoLock, ObjectIdGetDatum(), RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy1, HeapTupleData::t_self, table_close(), table_open(), and values.

Referenced by DisableSubscriptionAndExit().

◆ FreeSubscription()

void FreeSubscription ( Subscription sub)

Definition at line 155 of file pg_subscription.c.

156 {
157  pfree(sub->name);
158  pfree(sub->conninfo);
159  if (sub->slotname)
160  pfree(sub->slotname);
162  pfree(sub);
163 }
void list_free_deep(List *list)
Definition: list.c:1560
void pfree(void *pointer)
Definition: mcxt.c:1520

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

Referenced by maybe_reread_subscription().

◆ GetSubscription()

Subscription* GetSubscription ( Oid  subid,
bool  missing_ok 
)

Definition at line 41 of file pg_subscription.c.

42 {
43  HeapTuple tup;
44  Subscription *sub;
45  Form_pg_subscription subform;
46  Datum datum;
47  bool isnull;
48 
49  tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
50 
51  if (!HeapTupleIsValid(tup))
52  {
53  if (missing_ok)
54  return NULL;
55 
56  elog(ERROR, "cache lookup failed for subscription %u", subid);
57  }
58 
59  subform = (Form_pg_subscription) GETSTRUCT(tup);
60 
61  sub = (Subscription *) palloc(sizeof(Subscription));
62  sub->oid = subid;
63  sub->dbid = subform->subdbid;
64  sub->skiplsn = subform->subskiplsn;
65  sub->name = pstrdup(NameStr(subform->subname));
66  sub->owner = subform->subowner;
67  sub->enabled = subform->subenabled;
68  sub->binary = subform->subbinary;
69  sub->stream = subform->substream;
70  sub->twophasestate = subform->subtwophasestate;
71  sub->disableonerr = subform->subdisableonerr;
72  sub->passwordrequired = subform->subpasswordrequired;
73  sub->runasowner = subform->subrunasowner;
74  sub->failover = subform->subfailover;
75 
76  /* Get conninfo */
77  datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
78  tup,
79  Anum_pg_subscription_subconninfo);
80  sub->conninfo = TextDatumGetCString(datum);
81 
82  /* Get slotname */
83  datum = SysCacheGetAttr(SUBSCRIPTIONOID,
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 */
93  datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
94  tup,
95  Anum_pg_subscription_subsynccommit);
96  sub->synccommit = TextDatumGetCString(datum);
97 
98  /* Get publications */
99  datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
100  tup,
101  Anum_pg_subscription_subpublications);
103 
104  /* Get origin */
105  datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
106  tup,
107  Anum_pg_subscription_suborigin);
108  sub->origin = TextDatumGetCString(datum);
109 
110  /* Is the subscription owner a superuser? */
111  sub->ownersuperuser = superuser_arg(sub->owner);
112 
113  ReleaseSysCache(tup);
114 
115  return sub;
116 }
#define DatumGetArrayTypeP(X)
Definition: array.h:261
#define TextDatumGetCString(d)
Definition: builtins.h:98
#define NameStr(name)
Definition: c.h:746
#define GETSTRUCT(TUP)
Definition: htup_details.h:653
char * pstrdup(const char *in)
Definition: mcxt.c:1695
void * palloc(Size size)
Definition: mcxt.c:1316
static List * textarray_to_stringlist(ArrayType *textarray)
FormData_pg_subscription * Form_pg_subscription
static Name DatumGetName(Datum X)
Definition: postgres.h:360
XLogRecPtr skiplsn
bool superuser_arg(Oid roleid)
Definition: superuser.c:56
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:266
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:218
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:479
Datum SysCacheGetAttrNotNull(int cacheId, HeapTuple tup, AttrNumber attributeNumber)
Definition: syscache.c:510

References Subscription::binary, Subscription::conninfo, DatumGetArrayTypeP, DatumGetName(), Subscription::dbid, Subscription::disableonerr, elog, Subscription::enabled, ERROR, Subscription::failover, GETSTRUCT, HeapTupleIsValid, Subscription::name, NameStr, ObjectIdGetDatum(), Subscription::oid, Subscription::origin, Subscription::owner, Subscription::ownersuperuser, palloc(), Subscription::passwordrequired, pstrdup(), Subscription::publications, ReleaseSysCache(), Subscription::runasowner, SearchSysCache1(), Subscription::skiplsn, Subscription::slotname, Subscription::stream, superuser_arg(), Subscription::synccommit, SysCacheGetAttr(), SysCacheGetAttrNotNull(), textarray_to_stringlist(), TextDatumGetCString, and Subscription::twophasestate.

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

◆ GetSubscriptionRelations()

List* GetSubscriptionRelations ( Oid  subid,
bool  not_ready 
)

Definition at line 501 of file pg_subscription.c.

502 {
503  List *res = NIL;
504  Relation rel;
505  HeapTuple tup;
506  int nkeys = 0;
507  ScanKeyData skey[2];
508  SysScanDesc scan;
509 
510  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
511 
512  ScanKeyInit(&skey[nkeys++],
513  Anum_pg_subscription_rel_srsubid,
514  BTEqualStrategyNumber, F_OIDEQ,
515  ObjectIdGetDatum(subid));
516 
517  if (not_ready)
518  ScanKeyInit(&skey[nkeys++],
519  Anum_pg_subscription_rel_srsubstate,
520  BTEqualStrategyNumber, F_CHARNE,
521  CharGetDatum(SUBREL_STATE_READY));
522 
523  scan = systable_beginscan(rel, InvalidOid, false,
524  NULL, nkeys, skey);
525 
526  while (HeapTupleIsValid(tup = systable_getnext(scan)))
527  {
529  SubscriptionRelState *relstate;
530  Datum d;
531  bool isnull;
532 
533  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
534 
535  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
536  relstate->relid = subrel->srrelid;
537  relstate->state = subrel->srsubstate;
538  d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
539  Anum_pg_subscription_rel_srsublsn, &isnull);
540  if (isnull)
541  relstate->lsn = InvalidXLogRecPtr;
542  else
543  relstate->lsn = DatumGetLSN(d);
544 
545  res = lappend(res, relstate);
546  }
547 
548  /* Cleanup */
549  systable_endscan(scan);
551 
552  return res;
553 }
List * lappend(List *list, void *datum)
Definition: list.c:339
#define NIL
Definition: pg_list.h:68
static XLogRecPtr DatumGetLSN(Datum X)
Definition: pg_lsn.h:22
FormData_pg_subscription_rel * Form_pg_subscription_rel
Definition: pg_list.h:54

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

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

◆ GetSubscriptionRelState()

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

Definition at line 341 of file pg_subscription.c.

342 {
343  HeapTuple tup;
344  char substate;
345  bool isnull;
346  Datum d;
347  Relation rel;
348 
349  /*
350  * This is to avoid the race condition with AlterSubscription which tries
351  * to remove this relstate.
352  */
353  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
354 
355  /* Try finding the mapping. */
356  tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
357  ObjectIdGetDatum(relid),
358  ObjectIdGetDatum(subid));
359 
360  if (!HeapTupleIsValid(tup))
361  {
363  *sublsn = InvalidXLogRecPtr;
364  return SUBREL_STATE_UNKNOWN;
365  }
366 
367  /* Get the state. */
368  substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
369 
370  /* Get the LSN */
371  d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
372  Anum_pg_subscription_rel_srsublsn, &isnull);
373  if (isnull)
374  *sublsn = InvalidXLogRecPtr;
375  else
376  *sublsn = DatumGetLSN(d);
377 
378  /* Cleanup */
379  ReleaseSysCache(tup);
380 
382 
383  return substate;
384 }
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:229

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

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

◆ HasSubscriptionRelations()

bool HasSubscriptionRelations ( Oid  subid)

Definition at line 466 of file pg_subscription.c.

467 {
468  Relation rel;
469  ScanKeyData skey[1];
470  SysScanDesc scan;
471  bool has_subrels;
472 
473  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
474 
475  ScanKeyInit(&skey[0],
476  Anum_pg_subscription_rel_srsubid,
477  BTEqualStrategyNumber, F_OIDEQ,
478  ObjectIdGetDatum(subid));
479 
480  scan = systable_beginscan(rel, InvalidOid, false,
481  NULL, 1, skey);
482 
483  /* If even a single tuple exists then the subscription has tables. */
484  has_subrels = HeapTupleIsValid(systable_getnext(scan));
485 
486  /* Cleanup */
487  systable_endscan(scan);
489 
490  return has_subrels;
491 }

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

Referenced by FetchTableStates().

◆ RemoveSubscriptionRel()

void RemoveSubscriptionRel ( Oid  subid,
Oid  relid 
)

Definition at line 391 of file pg_subscription.c.

392 {
393  Relation rel;
394  TableScanDesc scan;
395  ScanKeyData skey[2];
396  HeapTuple tup;
397  int nkeys = 0;
398 
399  rel = table_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 = table_beginscan_catalog(rel, nkeys, skey);
422  {
424 
425  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
426 
427  /*
428  * We don't allow to drop the relation mapping when the table
429  * synchronization is in progress unless the caller updates the
430  * corresponding subscription as well. This is to ensure that we don't
431  * leave tablesync slots or origins in the system when the
432  * corresponding table is dropped.
433  */
434  if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
435  {
436  ereport(ERROR,
437  (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
438  errmsg("could not drop relation mapping for subscription \"%s\"",
439  get_subscription_name(subrel->srsubid, false)),
440  errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
441  get_rel_name(relid), subrel->srsubstate),
442 
443  /*
444  * translator: first %s is a SQL ALTER command and second %s is a
445  * SQL DROP command
446  */
447  errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
448  "ALTER SUBSCRIPTION ... ENABLE",
449  "DROP SUBSCRIPTION ...")));
450  }
451 
452  CatalogTupleDelete(rel, &tup->t_self);
453  }
454  table_endscan(scan);
455 
457 }
#define OidIsValid(objectId)
Definition: c.h:775
int errdetail(const char *fmt,...)
Definition: elog.c:1203
int errhint(const char *fmt,...)
Definition: elog.c:1317
int errcode(int sqlerrcode)
Definition: elog.c:857
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ereport(elevel,...)
Definition: elog.h:149
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1251
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:365
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1928
char * get_subscription_name(Oid subid, bool missing_ok)
Definition: lsyscache.c:3695
@ ForwardScanDirection
Definition: sdir.h:28
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:112
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:1029

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().

◆ textarray_to_stringlist()

static List * textarray_to_stringlist ( ArrayType textarray)
static

Definition at line 210 of file pg_subscription.c.

211 {
212  Datum *elems;
213  int nelems,
214  i;
215  List *res = NIL;
216 
217  deconstruct_array_builtin(textarray, TEXTOID, &elems, NULL, &nelems);
218 
219  if (nelems == 0)
220  return NIL;
221 
222  for (i = 0; i < nelems; i++)
224 
225  return res;
226 }
void deconstruct_array_builtin(ArrayType *array, Oid elmtype, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3678
int i
Definition: isn.c:73
String * makeString(char *str)
Definition: value.c:63

References deconstruct_array_builtin(), i, lappend(), makeString(), NIL, res, and TextDatumGetCString.

Referenced by GetSubscription().

◆ UpdateSubscriptionRelState()

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

Definition at line 290 of file pg_subscription.c.

292 {
293  Relation rel;
294  HeapTuple tup;
295  bool nulls[Natts_pg_subscription_rel];
296  Datum values[Natts_pg_subscription_rel];
297  bool replaces[Natts_pg_subscription_rel];
298 
299  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
300 
301  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
302 
303  /* Try finding existing mapping. */
304  tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
305  ObjectIdGetDatum(relid),
306  ObjectIdGetDatum(subid));
307  if (!HeapTupleIsValid(tup))
308  elog(ERROR, "subscription table %u in subscription %u does not exist",
309  relid, subid);
310 
311  /* Update the tuple. */
312  memset(values, 0, sizeof(values));
313  memset(nulls, false, sizeof(nulls));
314  memset(replaces, false, sizeof(replaces));
315 
316  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
317  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
318 
319  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
320  if (sublsn != InvalidXLogRecPtr)
321  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
322  else
323  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
324 
325  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
326  replaces);
327 
328  /* Update the catalog. */
329  CatalogTupleUpdate(rel, &tup->t_self, tup);
330 
331  /* Cleanup. */
332  table_close(rel, NoLock);
333 }

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

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