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)
 
void DisableSubscription (Oid subid)
 
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)
 
bool HasSubscriptionRelations (Oid subid)
 
ListGetSubscriptionRelations (Oid subid)
 
ListGetSubscriptionNotReadyRelations (Oid subid)
 

Function Documentation

◆ AddSubscriptionRelState()

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

Definition at line 280 of file pg_subscription.c.

282 {
283  Relation rel;
284  HeapTuple tup;
285  bool nulls[Natts_pg_subscription_rel];
286  Datum values[Natts_pg_subscription_rel];
287 
288  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
289 
290  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
291 
292  /* Try finding existing mapping. */
294  ObjectIdGetDatum(relid),
295  ObjectIdGetDatum(subid));
296  if (HeapTupleIsValid(tup))
297  elog(ERROR, "subscription table %u in subscription %u already exists",
298  relid, subid);
299 
300  /* Form the tuple. */
301  memset(values, 0, sizeof(values));
302  memset(nulls, false, sizeof(nulls));
303  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
304  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
305  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
306  if (sublsn != InvalidXLogRecPtr)
307  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
308  else
309  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
310 
311  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
312 
313  /* Insert tuple into catalog. */
314  CatalogTupleInsert(rel, tup);
315 
316  heap_freetuple(tup);
317 
318  /* Cleanup. */
319  table_close(rel, NoLock);
320 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define ERROR
Definition: elog.h:33
#define elog(elevel,...)
Definition: elog.h:218
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1338
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
void CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:221
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1046
#define NoLock
Definition: lockdefs.h:34
#define AccessShareLock
Definition: lockdefs.h:36
#define RowExclusiveLock
Definition: lockdefs.h:38
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
uintptr_t Datum
Definition: postgres.h:411
#define ObjectIdGetDatum(X)
Definition: postgres.h:551
#define CharGetDatum(X)
Definition: postgres.h:460
#define RelationGetDescr(relation)
Definition: rel.h:515
Definition: regguts.h:318
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:181
@ SUBSCRIPTIONRELMAP
Definition: syscache.h:100
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:167
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
#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, SUBSCRIPTIONRELMAP, table_close(), table_open(), and values.

Referenced by AlterSubscription_refresh(), and CreateSubscription().

◆ CountDBSubscriptions()

int CountDBSubscriptions ( Oid  dbid)

Definition at line 119 of file pg_subscription.c.

120 {
121  int nsubs = 0;
122  Relation rel;
123  ScanKeyData scankey;
124  SysScanDesc scan;
125  HeapTuple tup;
126 
127  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
128 
129  ScanKeyInit(&scankey,
130  Anum_pg_subscription_subdbid,
131  BTEqualStrategyNumber, F_OIDEQ,
132  ObjectIdGetDatum(dbid));
133 
134  scan = systable_beginscan(rel, InvalidOid, false,
135  NULL, 1, &scankey);
136 
137  while (HeapTupleIsValid(tup = systable_getnext(scan)))
138  nsubs++;
139 
140  systable_endscan(scan);
141 
142  table_close(rel, NoLock);
143 
144  return nsubs;
145 }
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:598
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:505
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:386
#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 165 of file pg_subscription.c.

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

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

Referenced by DisableSubscriptionAndExit().

◆ FreeSubscription()

void FreeSubscription ( Subscription sub)

Definition at line 151 of file pg_subscription.c.

152 {
153  pfree(sub->name);
154  pfree(sub->conninfo);
155  if (sub->slotname)
156  pfree(sub->slotname);
158  pfree(sub);
159 }
void list_free_deep(List *list)
Definition: list.c:1519
void pfree(void *pointer)
Definition: mcxt.c:1175

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

Referenced by maybe_reread_subscription().

◆ get_subscription_name()

char* get_subscription_name ( Oid  subid,
bool  missing_ok 
)

Definition at line 227 of file pg_subscription.c.

228 {
229  HeapTuple tup;
230  char *subname;
231  Form_pg_subscription subform;
232 
234 
235  if (!HeapTupleIsValid(tup))
236  {
237  if (!missing_ok)
238  elog(ERROR, "cache lookup failed for subscription %u", subid);
239  return NULL;
240  }
241 
242  subform = (Form_pg_subscription) GETSTRUCT(tup);
243  subname = pstrdup(NameStr(subform->subname));
244 
245  ReleaseSysCache(tup);
246 
247  return subname;
248 }
#define NameStr(name)
Definition: c.h:681
#define GETSTRUCT(TUP)
Definition: htup_details.h:649
char * pstrdup(const char *in)
Definition: mcxt.c:1305
NameData subname
FormData_pg_subscription * Form_pg_subscription
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1221
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:1173

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

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

◆ get_subscription_oid()

Oid get_subscription_oid ( const char *  subname,
bool  missing_ok 
)

Definition at line 207 of file pg_subscription.c.

208 {
209  Oid oid;
210 
211  oid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
213  if (!OidIsValid(oid) && !missing_ok)
214  ereport(ERROR,
215  (errcode(ERRCODE_UNDEFINED_OBJECT),
216  errmsg("subscription \"%s\" does not exist", subname)));
217  return oid;
218 }
#define OidIsValid(objectId)
Definition: c.h:710
int errcode(int sqlerrcode)
Definition: elog.c:693
int errmsg(const char *fmt,...)
Definition: elog.c:904
#define ereport(elevel,...)
Definition: elog.h:143
Oid MyDatabaseId
Definition: globals.c:89
#define CStringGetDatum(X)
Definition: postgres.h:622
unsigned int Oid
Definition: postgres_ext.h:31
@ SUBSCRIPTIONNAME
Definition: syscache.h:98
#define GetSysCacheOid2(cacheId, oidcol, key1, key2)
Definition: syscache.h:199

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

Referenced by get_object_address_unqualified().

◆ GetSubscription()

Subscription* GetSubscription ( Oid  subid,
bool  missing_ok 
)

Definition at line 43 of file pg_subscription.c.

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->skiplsn = subform->subskiplsn;
67  sub->name = pstrdup(NameStr(subform->subname));
68  sub->owner = subform->subowner;
69  sub->enabled = subform->subenabled;
70  sub->binary = subform->subbinary;
71  sub->stream = subform->substream;
72  sub->twophasestate = subform->subtwophasestate;
73  sub->disableonerr = subform->subdisableonerr;
74 
75  /* Get conninfo */
77  tup,
78  Anum_pg_subscription_subconninfo,
79  &isnull);
80  Assert(!isnull);
81  sub->conninfo = TextDatumGetCString(datum);
82 
83  /* Get slotname */
85  tup,
86  Anum_pg_subscription_subslotname,
87  &isnull);
88  if (!isnull)
89  sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
90  else
91  sub->slotname = NULL;
92 
93  /* Get synccommit */
95  tup,
96  Anum_pg_subscription_subsynccommit,
97  &isnull);
98  Assert(!isnull);
99  sub->synccommit = TextDatumGetCString(datum);
100 
101  /* Get publications */
103  tup,
104  Anum_pg_subscription_subpublications,
105  &isnull);
106  Assert(!isnull);
108 
109  ReleaseSysCache(tup);
110 
111  return sub;
112 }
#define DatumGetArrayTypeP(X)
Definition: array.h:254
#define TextDatumGetCString(d)
Definition: builtins.h:86
Assert(fmt[strlen(fmt) - 1] !='\n')
void * palloc(Size size)
Definition: mcxt.c:1068
static List * textarray_to_stringlist(ArrayType *textarray)
#define DatumGetName(X)
Definition: postgres.h:629
XLogRecPtr skiplsn
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1434

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

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

◆ GetSubscriptionNotReadyRelations()

List* GetSubscriptionNotReadyRelations ( Oid  subid)

Definition at line 588 of file pg_subscription.c.

589 {
590  List *res = NIL;
591  Relation rel;
592  HeapTuple tup;
593  int nkeys = 0;
594  ScanKeyData skey[2];
595  SysScanDesc scan;
596 
597  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
598 
599  ScanKeyInit(&skey[nkeys++],
600  Anum_pg_subscription_rel_srsubid,
601  BTEqualStrategyNumber, F_OIDEQ,
602  ObjectIdGetDatum(subid));
603 
604  ScanKeyInit(&skey[nkeys++],
605  Anum_pg_subscription_rel_srsubstate,
606  BTEqualStrategyNumber, F_CHARNE,
607  CharGetDatum(SUBREL_STATE_READY));
608 
609  scan = systable_beginscan(rel, InvalidOid, false,
610  NULL, nkeys, skey);
611 
612  while (HeapTupleIsValid(tup = systable_getnext(scan)))
613  {
615  SubscriptionRelState *relstate;
616  Datum d;
617  bool isnull;
618 
619  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
620 
621  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
622  relstate->relid = subrel->srrelid;
623  relstate->state = subrel->srsubstate;
625  Anum_pg_subscription_rel_srsublsn, &isnull);
626  if (isnull)
627  relstate->lsn = InvalidXLogRecPtr;
628  else
629  relstate->lsn = DatumGetLSN(d);
630 
631  res = lappend(res, relstate);
632  }
633 
634  /* Cleanup */
635  systable_endscan(scan);
637 
638  return res;
639 }
List * lappend(List *list, void *datum)
Definition: list.c:336
#define NIL
Definition: pg_list.h:65
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
FormData_pg_subscription_rel * Form_pg_subscription_rel
Definition: pg_list.h:51

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

Referenced by DropSubscription(), and FetchTableStates().

◆ GetSubscriptionRelations()

List* GetSubscriptionRelations ( Oid  subid)

Definition at line 535 of file pg_subscription.c.

536 {
537  List *res = NIL;
538  Relation rel;
539  HeapTuple tup;
540  ScanKeyData skey[1];
541  SysScanDesc scan;
542 
543  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
544 
545  ScanKeyInit(&skey[0],
546  Anum_pg_subscription_rel_srsubid,
547  BTEqualStrategyNumber, F_OIDEQ,
548  ObjectIdGetDatum(subid));
549 
550  scan = systable_beginscan(rel, InvalidOid, false,
551  NULL, 1, skey);
552 
553  while (HeapTupleIsValid(tup = systable_getnext(scan)))
554  {
556  SubscriptionRelState *relstate;
557  Datum d;
558  bool isnull;
559 
560  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
561 
562  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
563  relstate->relid = subrel->srrelid;
564  relstate->state = subrel->srsubstate;
566  Anum_pg_subscription_rel_srsublsn, &isnull);
567  if (isnull)
568  relstate->lsn = InvalidXLogRecPtr;
569  else
570  relstate->lsn = DatumGetLSN(d);
571 
572  res = lappend(res, relstate);
573  }
574 
575  /* Cleanup */
576  systable_endscan(scan);
578 
579  return res;
580 }

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

Referenced by AlterSubscription_refresh().

◆ GetSubscriptionRelState()

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

Definition at line 377 of file pg_subscription.c.

378 {
379  HeapTuple tup;
380  char substate;
381  bool isnull;
382  Datum d;
383  Relation rel;
384 
385  /*
386  * This is to avoid the race condition with AlterSubscription which tries
387  * to remove this relstate.
388  */
389  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
390 
391  /* Try finding the mapping. */
393  ObjectIdGetDatum(relid),
394  ObjectIdGetDatum(subid));
395 
396  if (!HeapTupleIsValid(tup))
397  {
399  *sublsn = InvalidXLogRecPtr;
400  return SUBREL_STATE_UNKNOWN;
401  }
402 
403  /* Get the state. */
404  substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
405 
406  /* Get the LSN */
408  Anum_pg_subscription_rel_srsublsn, &isnull);
409  if (isnull)
410  *sublsn = InvalidXLogRecPtr;
411  else
412  *sublsn = DatumGetLSN(d);
413 
414  /* Cleanup */
415  ReleaseSysCache(tup);
416 
418 
419  return substate;
420 }
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:1184

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

◆ HasSubscriptionRelations()

bool HasSubscriptionRelations ( Oid  subid)

Definition at line 502 of file pg_subscription.c.

503 {
504  Relation rel;
505  ScanKeyData skey[1];
506  SysScanDesc scan;
507  bool has_subrels;
508 
509  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
510 
511  ScanKeyInit(&skey[0],
512  Anum_pg_subscription_rel_srsubid,
513  BTEqualStrategyNumber, F_OIDEQ,
514  ObjectIdGetDatum(subid));
515 
516  scan = systable_beginscan(rel, InvalidOid, false,
517  NULL, 1, skey);
518 
519  /* If even a single tuple exists then the subscription has tables. */
520  has_subrels = HeapTupleIsValid(systable_getnext(scan));
521 
522  /* Cleanup */
523  systable_endscan(scan);
525 
526  return has_subrels;
527 }

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 427 of file pg_subscription.c.

428 {
429  Relation rel;
430  TableScanDesc scan;
431  ScanKeyData skey[2];
432  HeapTuple tup;
433  int nkeys = 0;
434 
435  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
436 
437  if (OidIsValid(subid))
438  {
439  ScanKeyInit(&skey[nkeys++],
440  Anum_pg_subscription_rel_srsubid,
442  F_OIDEQ,
443  ObjectIdGetDatum(subid));
444  }
445 
446  if (OidIsValid(relid))
447  {
448  ScanKeyInit(&skey[nkeys++],
449  Anum_pg_subscription_rel_srrelid,
451  F_OIDEQ,
452  ObjectIdGetDatum(relid));
453  }
454 
455  /* Do the search and delete what we found. */
456  scan = table_beginscan_catalog(rel, nkeys, skey);
458  {
460 
461  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
462 
463  /*
464  * We don't allow to drop the relation mapping when the table
465  * synchronization is in progress unless the caller updates the
466  * corresponding subscription as well. This is to ensure that we don't
467  * leave tablesync slots or origins in the system when the
468  * corresponding table is dropped.
469  */
470  if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
471  {
472  ereport(ERROR,
473  (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
474  errmsg("could not drop relation mapping for subscription \"%s\"",
475  get_subscription_name(subrel->srsubid, false)),
476  errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
477  get_rel_name(relid), subrel->srsubstate),
478 
479  /*
480  * translator: first %s is a SQL ALTER command and second %s is a
481  * SQL DROP command
482  */
483  errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
484  "ALTER SUBSCRIPTION ... ENABLE",
485  "DROP SUBSCRIPTION ...")));
486  }
487 
488  CatalogTupleDelete(rel, &tup->t_self);
489  }
490  table_endscan(scan);
491 
493 }
int errdetail(const char *fmt,...)
Definition: elog.c:1037
int errhint(const char *fmt,...)
Definition: elog.c:1151
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1296
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:350
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1909
char * get_subscription_name(Oid subid, bool missing_ok)
@ ForwardScanDirection
Definition: sdir.h:26
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:112
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:993

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 256 of file pg_subscription.c.

257 {
258  Datum *elems;
259  int nelems,
260  i;
261  List *res = NIL;
262 
263  deconstruct_array(textarray,
264  TEXTOID, -1, false, TYPALIGN_INT,
265  &elems, NULL, &nelems);
266 
267  if (nelems == 0)
268  return NIL;
269 
270  for (i = 0; i < nelems; i++)
272 
273  return res;
274 }
void deconstruct_array(ArrayType *array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3491
int i
Definition: isn.c:73
String * makeString(char *str)
Definition: value.c:63

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

Referenced by GetSubscription().

◆ UpdateSubscriptionRelState()

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

Definition at line 326 of file pg_subscription.c.

328 {
329  Relation rel;
330  HeapTuple tup;
331  bool nulls[Natts_pg_subscription_rel];
332  Datum values[Natts_pg_subscription_rel];
333  bool replaces[Natts_pg_subscription_rel];
334 
335  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
336 
337  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
338 
339  /* Try finding existing mapping. */
341  ObjectIdGetDatum(relid),
342  ObjectIdGetDatum(subid));
343  if (!HeapTupleIsValid(tup))
344  elog(ERROR, "subscription table %u in subscription %u does not exist",
345  relid, subid);
346 
347  /* Update the tuple. */
348  memset(values, 0, sizeof(values));
349  memset(nulls, false, sizeof(nulls));
350  memset(replaces, false, sizeof(replaces));
351 
352  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
353  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
354 
355  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
356  if (sublsn != InvalidXLogRecPtr)
357  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
358  else
359  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
360 
361  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
362  replaces);
363 
364  /* Update the catalog. */
365  CatalogTupleUpdate(rel, &tup->t_self, tup);
366 
367  /* Cleanup. */
368  table_close(rel, NoLock);
369 }

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