PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
pg_subscription_rel.h File Reference
#include "access/xlogdefs.h"
#include "catalog/genbki.h"
#include "catalog/pg_subscription_rel_d.h"
#include "nodes/pg_list.h"
Include dependency graph for pg_subscription_rel.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  SubscriptionRelState
 

Typedefs

typedef FormData_pg_subscription_relForm_pg_subscription_rel
 
typedef struct SubscriptionRelState SubscriptionRelState
 

Functions

 CATALOG (pg_subscription_rel, 6102, SubscriptionRelRelationId)
 
 DECLARE_UNIQUE_INDEX_PKEY (pg_subscription_rel_srrelid_srsubid_index, 6117, SubscriptionRelSrrelidSrsubidIndexId, pg_subscription_rel, btree(srrelid oid_ops, srsubid oid_ops))
 
 MAKE_SYSCACHE (SUBSCRIPTIONRELMAP, pg_subscription_rel_srrelid_srsubid_index, 64)
 
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)
 

Variables

 FormData_pg_subscription_rel
 

Typedef Documentation

◆ Form_pg_subscription_rel

◆ SubscriptionRelState

Function Documentation

◆ AddSubscriptionRelState()

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

Definition at line 267 of file pg_subscription.c.

269{
270 Relation rel;
271 HeapTuple tup;
272 bool nulls[Natts_pg_subscription_rel];
273 Datum values[Natts_pg_subscription_rel];
274
275 LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
276
277 rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
278
279 /* Try finding existing mapping. */
280 tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
281 ObjectIdGetDatum(relid),
282 ObjectIdGetDatum(subid));
283 if (HeapTupleIsValid(tup))
284 elog(ERROR, "subscription table %u in subscription %u already exists",
285 relid, subid);
286
287 /* Form the tuple. */
288 memset(values, 0, sizeof(values));
289 memset(nulls, false, sizeof(nulls));
290 values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
291 values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
292 values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
293 if (sublsn != InvalidXLogRecPtr)
294 values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
295 else
296 nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
297
298 tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
299
300 /* Insert tuple into catalog. */
301 CatalogTupleInsert(rel, tup);
302
303 heap_freetuple(tup);
304
305 /* Cleanup. */
306 if (retain_lock)
307 {
308 table_close(rel, NoLock);
309 }
310 else
311 {
313 UnlockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
314 }
315}
static Datum values[MAXATTR]
Definition: bootstrap.c:151
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:225
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:1072
void UnlockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1131
#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:69
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:257
static Datum CharGetDatum(char X)
Definition: postgres.h:127
#define RelationGetDescr(relation)
Definition: rel.h:531
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().

◆ CATALOG()

CATALOG ( pg_subscription_rel  ,
6102  ,
SubscriptionRelRelationId   
)

Definition at line 31 of file pg_subscription_rel.h.

32{
33 Oid srsubid BKI_LOOKUP(pg_subscription); /* Oid of subscription */
34 Oid srrelid BKI_LOOKUP(pg_class); /* Oid of relation */
35 char srsubstate; /* state of the relation in subscription */
36
37 /*
38 * Although srsublsn is a fixed-width type, it is allowed to be NULL, so
39 * we prevent direct C code access to it just as for a varlena field.
40 */
41#ifdef CATALOG_VARLEN /* variable-length fields start here */
42
43 XLogRecPtr srsublsn BKI_FORCE_NULL; /* remote LSN of the state change
44 * used for synchronization
45 * coordination, or NULL if not
46 * valid */
47#endif
#define BKI_LOOKUP(catalog)
Definition: genbki.h:46
#define BKI_FORCE_NULL
Definition: genbki.h:32
FormData_pg_subscription_rel
unsigned int Oid
Definition: postgres_ext.h:32
uint64 XLogRecPtr
Definition: xlogdefs.h:21

References BKI_FORCE_NULL, and BKI_LOOKUP.

◆ DECLARE_UNIQUE_INDEX_PKEY()

DECLARE_UNIQUE_INDEX_PKEY ( pg_subscription_rel_srrelid_srsubid_index  ,
6117  ,
SubscriptionRelSrrelidSrsubidIndexId  ,
pg_subscription_rel  ,
btree(srrelid oid_ops, srsubid oid_ops)   
)

◆ GetSubscriptionRelations()

List * GetSubscriptionRelations ( Oid  subid,
bool  not_ready 
)

Definition at line 532 of file pg_subscription.c.

533{
534 List *res = NIL;
535 Relation rel;
536 HeapTuple tup;
537 int nkeys = 0;
538 ScanKeyData skey[2];
539 SysScanDesc scan;
540
541 rel = table_open(SubscriptionRelRelationId, AccessShareLock);
542
543 ScanKeyInit(&skey[nkeys++],
544 Anum_pg_subscription_rel_srsubid,
545 BTEqualStrategyNumber, F_OIDEQ,
546 ObjectIdGetDatum(subid));
547
548 if (not_ready)
549 ScanKeyInit(&skey[nkeys++],
550 Anum_pg_subscription_rel_srsubstate,
551 BTEqualStrategyNumber, F_CHARNE,
552 CharGetDatum(SUBREL_STATE_READY));
553
554 scan = systable_beginscan(rel, InvalidOid, false,
555 NULL, nkeys, skey);
556
557 while (HeapTupleIsValid(tup = systable_getnext(scan)))
558 {
560 SubscriptionRelState *relstate;
561 Datum d;
562 bool isnull;
563
564 subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
565
566 relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
567 relstate->relid = subrel->srrelid;
568 relstate->state = subrel->srsubstate;
569 d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
570 Anum_pg_subscription_rel_srsublsn, &isnull);
571 if (isnull)
572 relstate->lsn = InvalidXLogRecPtr;
573 else
574 relstate->lsn = DatumGetLSN(d);
575
576 res = lappend(res, relstate);
577 }
578
579 /* Cleanup */
580 systable_endscan(scan);
582
583 return res;
584}
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:606
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:513
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:387
#define GETSTRUCT(TUP)
Definition: htup_details.h:653
List * lappend(List *list, void *datum)
Definition: list.c:339
void * palloc(Size size)
Definition: mcxt.c:1317
#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
#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
Definition: pg_list.h:54
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:600

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

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

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

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

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

Referenced by FetchTableStates().

◆ MAKE_SYSCACHE()

MAKE_SYSCACHE ( SUBSCRIPTIONRELMAP  ,
pg_subscription_rel_srrelid_srsubid_index  ,
64   
)

◆ RemoveSubscriptionRel()

void RemoveSubscriptionRel ( Oid  subid,
Oid  relid 
)

Definition at line 422 of file pg_subscription.c.

423{
424 Relation rel;
425 TableScanDesc scan;
426 ScanKeyData skey[2];
427 HeapTuple tup;
428 int nkeys = 0;
429
430 rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
431
432 if (OidIsValid(subid))
433 {
434 ScanKeyInit(&skey[nkeys++],
435 Anum_pg_subscription_rel_srsubid,
437 F_OIDEQ,
438 ObjectIdGetDatum(subid));
439 }
440
441 if (OidIsValid(relid))
442 {
443 ScanKeyInit(&skey[nkeys++],
444 Anum_pg_subscription_rel_srrelid,
446 F_OIDEQ,
447 ObjectIdGetDatum(relid));
448 }
449
450 /* Do the search and delete what we found. */
451 scan = table_beginscan_catalog(rel, nkeys, skey);
453 {
455
456 subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
457
458 /*
459 * We don't allow to drop the relation mapping when the table
460 * synchronization is in progress unless the caller updates the
461 * corresponding subscription as well. This is to ensure that we don't
462 * leave tablesync slots or origins in the system when the
463 * corresponding table is dropped.
464 */
465 if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
466 {
468 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
469 errmsg("could not drop relation mapping for subscription \"%s\"",
470 get_subscription_name(subrel->srsubid, false)),
471 errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
472 get_rel_name(relid), subrel->srsubstate),
473
474 /*
475 * translator: first %s is a SQL ALTER command and second %s is a
476 * SQL DROP command
477 */
478 errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
479 "ALTER SUBSCRIPTION ... ENABLE",
480 "DROP SUBSCRIPTION ...")));
481 }
482
483 CatalogTupleDelete(rel, &tup->t_self);
484 }
485 table_endscan(scan);
486
488}
#define OidIsValid(objectId)
Definition: c.h:732
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:853
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:1264
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
ItemPointerData t_self
Definition: htup.h:65
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:112
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:1024

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

◆ UpdateSubscriptionRelState()

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

Definition at line 321 of file pg_subscription.c.

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

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

Variable Documentation

◆ FormData_pg_subscription_rel

FormData_pg_subscription_rel

Definition at line 48 of file pg_subscription_rel.h.