PostgreSQL Source Code  git master
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))
 
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, 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 
)

Definition at line 230 of file pg_subscription.c.

232 {
233  Relation rel;
234  HeapTuple tup;
235  bool nulls[Natts_pg_subscription_rel];
236  Datum values[Natts_pg_subscription_rel];
237 
238  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
239 
240  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
241 
242  /* Try finding existing mapping. */
244  ObjectIdGetDatum(relid),
245  ObjectIdGetDatum(subid));
246  if (HeapTupleIsValid(tup))
247  elog(ERROR, "subscription table %u in subscription %u already exists",
248  relid, subid);
249 
250  /* Form the tuple. */
251  memset(values, 0, sizeof(values));
252  memset(nulls, false, sizeof(nulls));
253  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
254  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
255  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
256  if (sublsn != InvalidXLogRecPtr)
257  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
258  else
259  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
260 
261  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
262 
263  /* Insert tuple into catalog. */
264  CatalogTupleInsert(rel, tup);
265 
266  heap_freetuple(tup);
267 
268  /* Cleanup. */
269  table_close(rel, NoLock);
270 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
#define ERROR
Definition: elog.h:39
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1108
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1426
#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:1046
#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:530
Definition: regguts.h:323
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:184
@ SUBSCRIPTIONRELMAP
Definition: syscache.h:100
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, SUBSCRIPTIONRELMAP, table_close(), table_open(), and values.

Referenced by AlterSubscription_refresh(), 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:31
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 487 of file pg_subscription.c.

488 {
489  List *res = NIL;
490  Relation rel;
491  HeapTuple tup;
492  int nkeys = 0;
493  ScanKeyData skey[2];
494  SysScanDesc scan;
495 
496  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
497 
498  ScanKeyInit(&skey[nkeys++],
499  Anum_pg_subscription_rel_srsubid,
500  BTEqualStrategyNumber, F_OIDEQ,
501  ObjectIdGetDatum(subid));
502 
503  if (not_ready)
504  ScanKeyInit(&skey[nkeys++],
505  Anum_pg_subscription_rel_srsubstate,
506  BTEqualStrategyNumber, F_CHARNE,
507  CharGetDatum(SUBREL_STATE_READY));
508 
509  scan = systable_beginscan(rel, InvalidOid, false,
510  NULL, nkeys, skey);
511 
512  while (HeapTupleIsValid(tup = systable_getnext(scan)))
513  {
515  SubscriptionRelState *relstate;
516  Datum d;
517  bool isnull;
518 
519  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
520 
521  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
522  relstate->relid = subrel->srrelid;
523  relstate->state = subrel->srsubstate;
525  Anum_pg_subscription_rel_srsublsn, &isnull);
526  if (isnull)
527  relstate->lsn = InvalidXLogRecPtr;
528  else
529  relstate->lsn = DatumGetLSN(d);
530 
531  res = lappend(res, relstate);
532  }
533 
534  /* Cleanup */
535  systable_endscan(scan);
537 
538  return res;
539 }
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:599
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:506
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:338
void * palloc(Size size)
Definition: mcxt.c:1226
#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:36
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:1081

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 AlterSubscription_refresh(), DropSubscription(), and FetchTableStates().

◆ GetSubscriptionRelState()

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

Definition at line 327 of file pg_subscription.c.

328 {
329  HeapTuple tup;
330  char substate;
331  bool isnull;
332  Datum d;
333  Relation rel;
334 
335  /*
336  * This is to avoid the race condition with AlterSubscription which tries
337  * to remove this relstate.
338  */
339  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
340 
341  /* Try finding the mapping. */
343  ObjectIdGetDatum(relid),
344  ObjectIdGetDatum(subid));
345 
346  if (!HeapTupleIsValid(tup))
347  {
349  *sublsn = InvalidXLogRecPtr;
350  return SUBREL_STATE_UNKNOWN;
351  }
352 
353  /* Get the state. */
354  substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
355 
356  /* Get the LSN */
358  Anum_pg_subscription_rel_srsublsn, &isnull);
359  if (isnull)
360  *sublsn = InvalidXLogRecPtr;
361  else
362  *sublsn = DatumGetLSN(d);
363 
364  /* Cleanup */
365  ReleaseSysCache(tup);
366 
368 
369  return substate;
370 }
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:868
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:831

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

453 {
454  Relation rel;
455  ScanKeyData skey[1];
456  SysScanDesc scan;
457  bool has_subrels;
458 
459  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
460 
461  ScanKeyInit(&skey[0],
462  Anum_pg_subscription_rel_srsubid,
463  BTEqualStrategyNumber, F_OIDEQ,
464  ObjectIdGetDatum(subid));
465 
466  scan = systable_beginscan(rel, InvalidOid, false,
467  NULL, 1, skey);
468 
469  /* If even a single tuple exists then the subscription has tables. */
470  has_subrels = HeapTupleIsValid(systable_getnext(scan));
471 
472  /* Cleanup */
473  systable_endscan(scan);
475 
476  return has_subrels;
477 }

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

378 {
379  Relation rel;
380  TableScanDesc scan;
381  ScanKeyData skey[2];
382  HeapTuple tup;
383  int nkeys = 0;
384 
385  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
386 
387  if (OidIsValid(subid))
388  {
389  ScanKeyInit(&skey[nkeys++],
390  Anum_pg_subscription_rel_srsubid,
392  F_OIDEQ,
393  ObjectIdGetDatum(subid));
394  }
395 
396  if (OidIsValid(relid))
397  {
398  ScanKeyInit(&skey[nkeys++],
399  Anum_pg_subscription_rel_srrelid,
401  F_OIDEQ,
402  ObjectIdGetDatum(relid));
403  }
404 
405  /* Do the search and delete what we found. */
406  scan = table_beginscan_catalog(rel, nkeys, skey);
408  {
410 
411  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
412 
413  /*
414  * We don't allow to drop the relation mapping when the table
415  * synchronization is in progress unless the caller updates the
416  * corresponding subscription as well. This is to ensure that we don't
417  * leave tablesync slots or origins in the system when the
418  * corresponding table is dropped.
419  */
420  if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY)
421  {
422  ereport(ERROR,
423  (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
424  errmsg("could not drop relation mapping for subscription \"%s\"",
425  get_subscription_name(subrel->srsubid, false)),
426  errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
427  get_rel_name(relid), subrel->srsubstate),
428 
429  /*
430  * translator: first %s is a SQL ALTER command and second %s is a
431  * SQL DROP command
432  */
433  errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
434  "ALTER SUBSCRIPTION ... ENABLE",
435  "DROP SUBSCRIPTION ...")));
436  }
437 
438  CatalogTupleDelete(rel, &tup->t_self);
439  }
440  table_endscan(scan);
441 
443 }
#define OidIsValid(objectId)
Definition: c.h:764
int errdetail(const char *fmt,...)
Definition: elog.c:1202
int errhint(const char *fmt,...)
Definition: elog.c:1316
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define ereport(elevel,...)
Definition: elog.h:149
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1086
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:365
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1932
char * get_subscription_name(Oid subid, bool missing_ok)
Definition: lsyscache.c:3677
@ 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:1009

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

278 {
279  Relation rel;
280  HeapTuple tup;
281  bool nulls[Natts_pg_subscription_rel];
282  Datum values[Natts_pg_subscription_rel];
283  bool replaces[Natts_pg_subscription_rel];
284 
285  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
286 
287  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
288 
289  /* Try finding existing mapping. */
291  ObjectIdGetDatum(relid),
292  ObjectIdGetDatum(subid));
293  if (!HeapTupleIsValid(tup))
294  elog(ERROR, "subscription table %u in subscription %u does not exist",
295  relid, subid);
296 
297  /* Update the tuple. */
298  memset(values, 0, sizeof(values));
299  memset(nulls, false, sizeof(nulls));
300  memset(replaces, false, sizeof(replaces));
301 
302  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
303  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
304 
305  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
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_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
312  replaces);
313 
314  /* Update the catalog. */
315  CatalogTupleUpdate(rel, &tup->t_self, tup);
316 
317  /* Cleanup. */
318  table_close(rel, NoLock);
319 }
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:1201
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, SUBSCRIPTIONRELMAP, 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.