PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
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)
 
void GetPublicationsStr (List *publications, StringInfo dest, bool quote_literal)
 
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, bool already_locked)
 
char GetSubscriptionRelState (Oid subid, Oid relid, XLogRecPtr *sublsn)
 
void RemoveSubscriptionRel (Oid subid, Oid relid)
 
bool HasSubscriptionRelations (Oid subid)
 
ListGetSubscriptionRelations (Oid subid, bool not_ready)
 
void UpdateDeadTupleRetentionStatus (Oid subid, bool active)
 

Function Documentation

◆ AddSubscriptionRelState()

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

Definition at line 270 of file pg_subscription.c.

272{
273 Relation rel;
274 HeapTuple tup;
275 bool nulls[Natts_pg_subscription_rel];
276 Datum values[Natts_pg_subscription_rel];
277
278 LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
279
280 rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
281
282 /* Try finding existing mapping. */
283 tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
284 ObjectIdGetDatum(relid),
285 ObjectIdGetDatum(subid));
286 if (HeapTupleIsValid(tup))
287 elog(ERROR, "subscription table %u in subscription %u already exists",
288 relid, subid);
289
290 /* Form the tuple. */
291 memset(values, 0, sizeof(values));
292 memset(nulls, false, sizeof(nulls));
293 values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
294 values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
295 values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
296 if (sublsn != InvalidXLogRecPtr)
297 values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
298 else
299 nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
300
301 tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
302
303 /* Insert tuple into catalog. */
304 CatalogTupleInsert(rel, tup);
305
306 heap_freetuple(tup);
307
308 /* Cleanup. */
309 if (retain_lock)
310 {
311 table_close(rel, NoLock);
312 }
313 else
314 {
316 UnlockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
317 }
318}
static Datum values[MAXATTR]
Definition: bootstrap.c:153
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:226
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, const Datum *values, const bool *isnull)
Definition: heaptuple.c:1117
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1435
#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:1088
void UnlockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1148
#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:31
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:262
uint64_t Datum
Definition: postgres.h:70
static Datum CharGetDatum(char X)
Definition: postgres.h:132
#define RelationGetDescr(relation)
Definition: rel.h:540
Definition: regguts.h:323
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:93
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 157 of file pg_subscription.c.

158{
159 int nsubs = 0;
160 Relation rel;
161 ScanKeyData scankey;
162 SysScanDesc scan;
163 HeapTuple tup;
164
165 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
166
167 ScanKeyInit(&scankey,
168 Anum_pg_subscription_subdbid,
169 BTEqualStrategyNumber, F_OIDEQ,
170 ObjectIdGetDatum(dbid));
171
172 scan = systable_beginscan(rel, InvalidOid, false,
173 NULL, 1, &scankey);
174
175 while (HeapTupleIsValid(tup = systable_getnext(scan)))
176 nsubs++;
177
178 systable_endscan(scan);
179
180 table_close(rel, NoLock);
181
182 return nsubs;
183}
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:603
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:514
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:388
#define InvalidOid
Definition: postgres_ext.h:37
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 203 of file pg_subscription.c.

204{
205 Relation rel;
206 bool nulls[Natts_pg_subscription];
207 bool replaces[Natts_pg_subscription];
208 Datum values[Natts_pg_subscription];
209 HeapTuple tup;
210
211 /* Look up the subscription in the catalog */
212 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
213 tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
214
215 if (!HeapTupleIsValid(tup))
216 elog(ERROR, "cache lookup failed for subscription %u", subid);
217
218 LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
219
220 /* Form a new tuple. */
221 memset(values, 0, sizeof(values));
222 memset(nulls, false, sizeof(nulls));
223 memset(replaces, false, sizeof(replaces));
224
225 /* Set the subscription to disabled. */
226 values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
227 replaces[Anum_pg_subscription_subenabled - 1] = true;
228
229 /* Update the catalog */
230 tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
231 replaces);
232 CatalogTupleUpdate(rel, &tup->t_self, tup);
233 heap_freetuple(tup);
234
235 table_close(rel, NoLock);
236}
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, const Datum *replValues, const bool *replIsnull, const bool *doReplace)
Definition: heaptuple.c:1210
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:313
static Datum BoolGetDatum(bool X)
Definition: postgres.h:112
ItemPointerData t_self
Definition: htup.h:65
#define SearchSysCacheCopy1(cacheId, key1)
Definition: syscache.h:91

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

190{
191 pfree(sub->name);
192 pfree(sub->conninfo);
193 if (sub->slotname)
194 pfree(sub->slotname);
196 pfree(sub);
197}
void list_free_deep(List *list)
Definition: list.c:1560
void pfree(void *pointer)
Definition: mcxt.c:1594

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

Referenced by maybe_reread_subscription().

◆ GetPublicationsStr()

void GetPublicationsStr ( List publications,
StringInfo  dest,
bool  quote_literal 
)

Definition at line 41 of file pg_subscription.c.

42{
43 ListCell *lc;
44 bool first = true;
45
46 Assert(publications != NIL);
47
48 foreach(lc, publications)
49 {
50 char *pubname = strVal(lfirst(lc));
51
52 if (first)
53 first = false;
54 else
56
57 if (quote_literal)
59 else
60 {
64 }
65 }
66}
Assert(PointerIsAligned(start, uint64))
#define lfirst(lc)
Definition: pg_list.h:172
#define NIL
Definition: pg_list.h:68
char * quote_literal_cstr(const char *rawstr)
Definition: quote.c:103
Datum quote_literal(PG_FUNCTION_ARGS)
Definition: quote.c:78
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:230
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:242
#define strVal(v)
Definition: value.h:82

References appendStringInfoChar(), appendStringInfoString(), Assert(), generate_unaccent_rules::dest, lfirst, NIL, quote_literal(), quote_literal_cstr(), and strVal.

Referenced by check_publications(), check_publications_origin(), fetch_remote_table_info(), and fetch_table_list().

◆ GetSubscription()

Subscription * GetSubscription ( Oid  subid,
bool  missing_ok 
)

Definition at line 72 of file pg_subscription.c.

73{
74 HeapTuple tup;
75 Subscription *sub;
77 Datum datum;
78 bool isnull;
79
80 tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
81
82 if (!HeapTupleIsValid(tup))
83 {
84 if (missing_ok)
85 return NULL;
86
87 elog(ERROR, "cache lookup failed for subscription %u", subid);
88 }
89
90 subform = (Form_pg_subscription) GETSTRUCT(tup);
91
92 sub = (Subscription *) palloc(sizeof(Subscription));
93 sub->oid = subid;
94 sub->dbid = subform->subdbid;
95 sub->skiplsn = subform->subskiplsn;
96 sub->name = pstrdup(NameStr(subform->subname));
97 sub->owner = subform->subowner;
98 sub->enabled = subform->subenabled;
99 sub->binary = subform->subbinary;
100 sub->stream = subform->substream;
101 sub->twophasestate = subform->subtwophasestate;
102 sub->disableonerr = subform->subdisableonerr;
103 sub->passwordrequired = subform->subpasswordrequired;
104 sub->runasowner = subform->subrunasowner;
105 sub->failover = subform->subfailover;
106 sub->retaindeadtuples = subform->subretaindeadtuples;
107 sub->maxretention = subform->submaxretention;
108 sub->retentionactive = subform->subretentionactive;
109
110 /* Get conninfo */
111 datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
112 tup,
113 Anum_pg_subscription_subconninfo);
114 sub->conninfo = TextDatumGetCString(datum);
115
116 /* Get slotname */
117 datum = SysCacheGetAttr(SUBSCRIPTIONOID,
118 tup,
119 Anum_pg_subscription_subslotname,
120 &isnull);
121 if (!isnull)
122 sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
123 else
124 sub->slotname = NULL;
125
126 /* Get synccommit */
127 datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
128 tup,
129 Anum_pg_subscription_subsynccommit);
130 sub->synccommit = TextDatumGetCString(datum);
131
132 /* Get publications */
133 datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
134 tup,
135 Anum_pg_subscription_subpublications);
137
138 /* Get origin */
139 datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
140 tup,
141 Anum_pg_subscription_suborigin);
142 sub->origin = TextDatumGetCString(datum);
143
144 /* Is the subscription owner a superuser? */
146
147 ReleaseSysCache(tup);
148
149 return sub;
150}
#define DatumGetArrayTypeP(X)
Definition: array.h:261
#define TextDatumGetCString(d)
Definition: builtins.h:98
#define NameStr(name)
Definition: c.h:751
static void * GETSTRUCT(const HeapTupleData *tuple)
Definition: htup_details.h:728
char * pstrdup(const char *in)
Definition: mcxt.c:1759
void * palloc(Size size)
Definition: mcxt.c:1365
static List * textarray_to_stringlist(ArrayType *textarray)
FormData_pg_subscription * Form_pg_subscription
static Name DatumGetName(Datum X)
Definition: postgres.h:370
XLogRecPtr skiplsn
bool superuser_arg(Oid roleid)
Definition: superuser.c:56
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:264
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:220
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:595
Datum SysCacheGetAttrNotNull(int cacheId, HeapTuple tup, AttrNumber attributeNumber)
Definition: syscache.c:625

References Subscription::binary, Subscription::conninfo, DatumGetArrayTypeP, DatumGetName(), Subscription::dbid, Subscription::disableonerr, elog, Subscription::enabled, ERROR, Subscription::failover, GETSTRUCT(), HeapTupleIsValid, Subscription::maxretention, Subscription::name, NameStr, ObjectIdGetDatum(), Subscription::oid, Subscription::origin, Subscription::owner, Subscription::ownersuperuser, palloc(), Subscription::passwordrequired, pstrdup(), Subscription::publications, ReleaseSysCache(), Subscription::retaindeadtuples, Subscription::retentionactive, 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 550 of file pg_subscription.c.

551{
552 List *res = NIL;
553 Relation rel;
554 HeapTuple tup;
555 int nkeys = 0;
556 ScanKeyData skey[2];
557 SysScanDesc scan;
558
559 rel = table_open(SubscriptionRelRelationId, AccessShareLock);
560
561 ScanKeyInit(&skey[nkeys++],
562 Anum_pg_subscription_rel_srsubid,
563 BTEqualStrategyNumber, F_OIDEQ,
564 ObjectIdGetDatum(subid));
565
566 if (not_ready)
567 ScanKeyInit(&skey[nkeys++],
568 Anum_pg_subscription_rel_srsubstate,
569 BTEqualStrategyNumber, F_CHARNE,
570 CharGetDatum(SUBREL_STATE_READY));
571
572 scan = systable_beginscan(rel, InvalidOid, false,
573 NULL, nkeys, skey);
574
575 while (HeapTupleIsValid(tup = systable_getnext(scan)))
576 {
578 SubscriptionRelState *relstate;
579 Datum d;
580 bool isnull;
581
582 subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
583
584 relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
585 relstate->relid = subrel->srrelid;
586 relstate->state = subrel->srsubstate;
587 d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
588 Anum_pg_subscription_rel_srsublsn, &isnull);
589 if (isnull)
590 relstate->lsn = InvalidXLogRecPtr;
591 else
592 relstate->lsn = DatumGetLSN(d);
593
594 res = lappend(res, relstate);
595 }
596
597 /* Cleanup */
598 systable_endscan(scan);
600
601 return res;
602}
List * lappend(List *list, void *datum)
Definition: list.c:339
static XLogRecPtr DatumGetLSN(Datum X)
Definition: pg_lsn.h:25
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, 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 390 of file pg_subscription.c.

391{
392 HeapTuple tup;
393 char substate;
394 bool isnull;
395 Datum d;
396 Relation rel;
397
398 /*
399 * This is to avoid the race condition with AlterSubscription which tries
400 * to remove this relstate.
401 */
402 rel = table_open(SubscriptionRelRelationId, AccessShareLock);
403
404 /* Try finding the mapping. */
405 tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
406 ObjectIdGetDatum(relid),
407 ObjectIdGetDatum(subid));
408
409 if (!HeapTupleIsValid(tup))
410 {
412 *sublsn = InvalidXLogRecPtr;
413 return SUBREL_STATE_UNKNOWN;
414 }
415
416 /* Get the state. */
417 substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
418
419 /* Get the LSN */
420 d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
421 Anum_pg_subscription_rel_srsublsn, &isnull);
422 if (isnull)
423 *sublsn = InvalidXLogRecPtr;
424 else
425 *sublsn = DatumGetLSN(d);
426
427 /* Cleanup */
428 ReleaseSysCache(tup);
429
431
432 return substate;
433}
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:230

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

516{
517 Relation rel;
518 ScanKeyData skey[1];
519 SysScanDesc scan;
520 bool has_subrels;
521
522 rel = table_open(SubscriptionRelRelationId, AccessShareLock);
523
524 ScanKeyInit(&skey[0],
525 Anum_pg_subscription_rel_srsubid,
526 BTEqualStrategyNumber, F_OIDEQ,
527 ObjectIdGetDatum(subid));
528
529 scan = systable_beginscan(rel, InvalidOid, false,
530 NULL, 1, skey);
531
532 /* If even a single tuple exists then the subscription has tables. */
533 has_subrels = HeapTupleIsValid(systable_getnext(scan));
534
535 /* Cleanup */
536 systable_endscan(scan);
538
539 return has_subrels;
540}

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

441{
442 Relation rel;
443 TableScanDesc scan;
444 ScanKeyData skey[2];
445 HeapTuple tup;
446 int nkeys = 0;
447
448 rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
449
450 if (OidIsValid(subid))
451 {
452 ScanKeyInit(&skey[nkeys++],
453 Anum_pg_subscription_rel_srsubid,
455 F_OIDEQ,
456 ObjectIdGetDatum(subid));
457 }
458
459 if (OidIsValid(relid))
460 {
461 ScanKeyInit(&skey[nkeys++],
462 Anum_pg_subscription_rel_srrelid,
464 F_OIDEQ,
465 ObjectIdGetDatum(relid));
466 }
467
468 /* Do the search and delete what we found. */
469 scan = table_beginscan_catalog(rel, nkeys, skey);
471 {
473
474 subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
475
476 /*
477 * We don't allow to drop the relation mapping when the table
478 * synchronization is in progress unless the caller updates the
479 * corresponding subscription as well. This is to ensure that we don't
480 * leave tablesync slots or origins in the system when the
481 * corresponding table is dropped.
482 */
483 if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
484 {
486 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
487 errmsg("could not drop relation mapping for subscription \"%s\"",
488 get_subscription_name(subrel->srsubid, false)),
489 errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
490 get_rel_name(relid), subrel->srsubstate),
491
492 /*
493 * translator: first %s is a SQL ALTER command and second %s is a
494 * SQL DROP command
495 */
496 errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
497 "ALTER SUBSCRIPTION ... ENABLE",
498 "DROP SUBSCRIPTION ...")));
499 }
500
501 CatalogTupleDelete(rel, &tup->t_self);
502 }
503 table_endscan(scan);
504
506}
#define OidIsValid(objectId)
Definition: c.h:774
int errdetail(const char *fmt,...)
Definition: elog.c:1207
int errhint(const char *fmt,...)
Definition: elog.c:1321
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define ereport(elevel,...)
Definition: elog.h:150
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1346
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:365
char * get_rel_name(Oid relid)
Definition: lsyscache.c:2095
char * get_subscription_name(Oid subid, bool missing_ok)
Definition: lsyscache.c:3862
@ ForwardScanDirection
Definition: sdir.h:28
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, ScanKeyData *key)
Definition: tableam.c:113
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:985

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

245{
246 Datum *elems;
247 int nelems,
248 i;
249 List *res = NIL;
250
251 deconstruct_array_builtin(textarray, TEXTOID, &elems, NULL, &nelems);
252
253 if (nelems == 0)
254 return NIL;
255
256 for (i = 0; i < nelems; i++)
257 res = lappend(res, makeString(TextDatumGetCString(elems[i])));
258
259 return res;
260}
void deconstruct_array_builtin(ArrayType *array, Oid elmtype, Datum **elemsp, bool **nullsp, int *nelemsp)
Definition: arrayfuncs.c:3698
int i
Definition: isn.c:77
String * makeString(char *str)
Definition: value.c:63

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

Referenced by GetSubscription().

◆ UpdateDeadTupleRetentionStatus()

void UpdateDeadTupleRetentionStatus ( Oid  subid,
bool  active 
)

Definition at line 608 of file pg_subscription.c.

609{
610 Relation rel;
611 bool nulls[Natts_pg_subscription];
612 bool replaces[Natts_pg_subscription];
613 Datum values[Natts_pg_subscription];
614 HeapTuple tup;
615
616 /* Look up the subscription in the catalog */
617 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
618 tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
619
620 if (!HeapTupleIsValid(tup))
621 elog(ERROR, "cache lookup failed for subscription %u", subid);
622
623 LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
624
625 /* Form a new tuple. */
626 memset(values, 0, sizeof(values));
627 memset(nulls, false, sizeof(nulls));
628 memset(replaces, false, sizeof(replaces));
629
630 /* Set the subscription to disabled. */
631 values[Anum_pg_subscription_subretentionactive - 1] = active;
632 replaces[Anum_pg_subscription_subretentionactive - 1] = true;
633
634 /* Update the catalog */
635 tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
636 replaces);
637 CatalogTupleUpdate(rel, &tup->t_self, tup);
638 heap_freetuple(tup);
639
640 table_close(rel, NoLock);
641}

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

◆ UpdateSubscriptionRelState()

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

Definition at line 324 of file pg_subscription.c.

326{
327 Relation rel;
328 HeapTuple tup;
329 bool nulls[Natts_pg_subscription_rel];
330 Datum values[Natts_pg_subscription_rel];
331 bool replaces[Natts_pg_subscription_rel];
332
333 if (already_locked)
334 {
335#ifdef USE_ASSERT_CHECKING
336 LOCKTAG tag;
337
338 Assert(CheckRelationOidLockedByMe(SubscriptionRelRelationId,
339 RowExclusiveLock, true));
340 SET_LOCKTAG_OBJECT(tag, InvalidOid, SubscriptionRelationId, subid, 0);
341 Assert(LockHeldByMe(&tag, AccessShareLock, true));
342#endif
343
344 rel = table_open(SubscriptionRelRelationId, NoLock);
345 }
346 else
347 {
348 LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
349 rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
350 }
351
352 /* Try finding existing mapping. */
353 tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
354 ObjectIdGetDatum(relid),
355 ObjectIdGetDatum(subid));
356 if (!HeapTupleIsValid(tup))
357 elog(ERROR, "subscription table %u in subscription %u does not exist",
358 relid, subid);
359
360 /* Update the tuple. */
361 memset(values, 0, sizeof(values));
362 memset(nulls, false, sizeof(nulls));
363 memset(replaces, false, sizeof(replaces));
364
365 replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
366 values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
367
368 replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
369 if (sublsn != InvalidXLogRecPtr)
370 values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
371 else
372 nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
373
374 tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
375 replaces);
376
377 /* Update the catalog. */
378 CatalogTupleUpdate(rel, &tup->t_self, tup);
379
380 /* Cleanup. */
381 table_close(rel, NoLock);
382}
bool CheckRelationOidLockedByMe(Oid relid, LOCKMODE lockmode, bool orstronger)
Definition: lmgr.c:351
bool LockHeldByMe(const LOCKTAG *locktag, LOCKMODE lockmode, bool orstronger)
Definition: lock.c:643
#define SET_LOCKTAG_OBJECT(locktag, dboid, classoid, objoid, objsubid)
Definition: lock.h:264
Definition: lock.h:167

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

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