PostgreSQL Source Code  git master
pg_subscription_rel.h File Reference
#include "catalog/genbki.h"
#include "catalog/pg_subscription_rel_d.h"
#include "access/xlogdefs.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)
 
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, bool missing_ok)
 
void RemoveSubscriptionRel (Oid subid, Oid relid)
 
ListGetSubscriptionRelations (Oid subid)
 
ListGetSubscriptionNotReadyRelations (Oid subid)
 

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

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

243 {
244  Relation rel;
245  HeapTuple tup;
246  bool nulls[Natts_pg_subscription_rel];
247  Datum values[Natts_pg_subscription_rel];
248 
249  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
250 
251  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
252 
253  /* Try finding existing mapping. */
255  ObjectIdGetDatum(relid),
256  ObjectIdGetDatum(subid));
257  if (HeapTupleIsValid(tup))
258  elog(ERROR, "subscription table %u in subscription %u already exists",
259  relid, subid);
260 
261  /* Form the tuple. */
262  memset(values, 0, sizeof(values));
263  memset(nulls, false, sizeof(nulls));
264  values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
265  values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
266  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
267  if (sublsn != InvalidXLogRecPtr)
268  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
269  else
270  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
271 
272  tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
273 
274  /* Insert tuple into catalog. */
275  CatalogTupleInsert(rel, tup);
276 
277  heap_freetuple(tup);
278 
279  /* Cleanup. */
280  table_close(rel, NoLock);
281 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
#define RelationGetDescr(relation)
Definition: rel.h:445
#define AccessShareLock
Definition: lockdefs.h:36
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, Datum *values, bool *isnull)
Definition: heaptuple.c:1020
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1338
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
uintptr_t Datum
Definition: postgres.h:367
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1002
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
Definition: regguts.h:298
#define CharGetDatum(X)
Definition: postgres.h:416
static Datum values[MAXATTR]
Definition: bootstrap.c:167
#define elog(elevel,...)
Definition: elog.h:226
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:176
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
void CatalogTupleInsert(Relation heapRel, HeapTuple tup)
Definition: indexing.c:183

◆ CATALOG()

CATALOG ( pg_subscription_rel  ,
6102  ,
SubscriptionRelRelationId   
)

Definition at line 32 of file pg_subscription_rel.h.

33 {
34  Oid srsubid; /* Oid of subscription */
35  Oid srrelid; /* Oid of relation */
36  char srsubstate; /* state of the relation in subscription */
37  XLogRecPtr srsublsn; /* remote lsn of the state change used for
38  * synchronization coordination */
unsigned int Oid
Definition: postgres_ext.h:31
FormData_pg_subscription_rel
uint64 XLogRecPtr
Definition: xlogdefs.h:21

◆ GetSubscriptionNotReadyRelations()

List* GetSubscriptionNotReadyRelations ( Oid  subid)

Definition at line 484 of file pg_subscription.c.

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

Referenced by process_syncing_tables_for_apply().

485 {
486  List *res = NIL;
487  Relation rel;
488  HeapTuple tup;
489  int nkeys = 0;
490  ScanKeyData skey[2];
491  SysScanDesc scan;
492 
493  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
494 
495  ScanKeyInit(&skey[nkeys++],
496  Anum_pg_subscription_rel_srsubid,
497  BTEqualStrategyNumber, F_OIDEQ,
498  ObjectIdGetDatum(subid));
499 
500  ScanKeyInit(&skey[nkeys++],
501  Anum_pg_subscription_rel_srsubstate,
502  BTEqualStrategyNumber, F_CHARNE,
503  CharGetDatum(SUBREL_STATE_READY));
504 
505  scan = systable_beginscan(rel, InvalidOid, false,
506  NULL, nkeys, skey);
507 
508  while (HeapTupleIsValid(tup = systable_getnext(scan)))
509  {
511  SubscriptionRelState *relstate;
512 
513  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
514 
515  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
516  relstate->relid = subrel->srrelid;
517  relstate->state = subrel->srsubstate;
518  relstate->lsn = subrel->srsublsn;
519 
520  res = lappend(res, relstate);
521  }
522 
523  /* Cleanup */
524  systable_endscan(scan);
526 
527  return res;
528 }
#define NIL
Definition: pg_list.h:65
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:525
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
#define AccessShareLock
Definition: lockdefs.h:36
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:352
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:444
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
List * lappend(List *list, void *datum)
Definition: list.c:322
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
#define CharGetDatum(X)
Definition: postgres.h:416
void * palloc(Size size)
Definition: mcxt.c:949
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
Definition: pg_list.h:50
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ GetSubscriptionRelations()

List* GetSubscriptionRelations ( Oid  subid)

Definition at line 437 of file pg_subscription.c.

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

Referenced by AlterSubscription_refresh().

438 {
439  List *res = NIL;
440  Relation rel;
441  HeapTuple tup;
442  int nkeys = 0;
443  ScanKeyData skey[2];
444  SysScanDesc scan;
445 
446  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
447 
448  ScanKeyInit(&skey[nkeys++],
449  Anum_pg_subscription_rel_srsubid,
450  BTEqualStrategyNumber, F_OIDEQ,
451  ObjectIdGetDatum(subid));
452 
453  scan = systable_beginscan(rel, InvalidOid, false,
454  NULL, nkeys, skey);
455 
456  while (HeapTupleIsValid(tup = systable_getnext(scan)))
457  {
459  SubscriptionRelState *relstate;
460 
461  subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
462 
463  relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
464  relstate->relid = subrel->srrelid;
465  relstate->state = subrel->srsubstate;
466  relstate->lsn = subrel->srsublsn;
467 
468  res = lappend(res, relstate);
469  }
470 
471  /* Cleanup */
472  systable_endscan(scan);
474 
475  return res;
476 }
#define NIL
Definition: pg_list.h:65
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
void systable_endscan(SysScanDesc sysscan)
Definition: genam.c:525
#define GETSTRUCT(TUP)
Definition: htup_details.h:655
#define AccessShareLock
Definition: lockdefs.h:36
SysScanDesc systable_beginscan(Relation heapRelation, Oid indexId, bool indexOK, Snapshot snapshot, int nkeys, ScanKey key)
Definition: genam.c:352
HeapTuple systable_getnext(SysScanDesc sysscan)
Definition: genam.c:444
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
List * lappend(List *list, void *datum)
Definition: list.c:322
#define InvalidOid
Definition: postgres_ext.h:36
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
FormData_pg_subscription_rel * Form_pg_subscription_rel
void * palloc(Size size)
Definition: mcxt.c:949
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
Definition: pg_list.h:50
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ GetSubscriptionRelState()

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

Definition at line 338 of file pg_subscription.c.

References AccessShareLock, Assert, DatumGetChar, DatumGetLSN, elog, ERROR, HeapTupleIsValid, InvalidXLogRecPtr, ObjectIdGetDatum, ReleaseSysCache(), SearchSysCache2(), SUBSCRIPTIONRELMAP, SysCacheGetAttr(), table_close(), and table_open().

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

340 {
341  Relation rel;
342  HeapTuple tup;
343  char substate;
344  bool isnull;
345  Datum d;
346 
347  rel = table_open(SubscriptionRelRelationId, AccessShareLock);
348 
349  /* Try finding the mapping. */
351  ObjectIdGetDatum(relid),
352  ObjectIdGetDatum(subid));
353 
354  if (!HeapTupleIsValid(tup))
355  {
356  if (missing_ok)
357  {
359  *sublsn = InvalidXLogRecPtr;
360  return SUBREL_STATE_UNKNOWN;
361  }
362 
363  elog(ERROR, "subscription table %u in subscription %u does not exist",
364  relid, subid);
365  }
366 
367  /* Get the state. */
369  Anum_pg_subscription_rel_srsubstate, &isnull);
370  Assert(!isnull);
371  substate = DatumGetChar(d);
373  Anum_pg_subscription_rel_srsublsn, &isnull);
374  if (isnull)
375  *sublsn = InvalidXLogRecPtr;
376  else
377  *sublsn = DatumGetLSN(d);
378 
379  /* Cleanup */
380  ReleaseSysCache(tup);
382 
383  return substate;
384 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
#define AccessShareLock
Definition: lockdefs.h:36
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
uintptr_t Datum
Definition: postgres.h:367
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1172
#define DatumGetChar(X)
Definition: postgres.h:409
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:1385
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define Assert(condition)
Definition: c.h:732
HeapTuple SearchSysCache2(int cacheId, Datum key1, Datum key2)
Definition: syscache.c:1135
#define elog(elevel,...)
Definition: elog.h:226
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39

◆ RemoveSubscriptionRel()

void RemoveSubscriptionRel ( Oid  subid,
Oid  relid 
)

Definition at line 391 of file pg_subscription.c.

References BTEqualStrategyNumber, CatalogTupleDelete(), ForwardScanDirection, 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().

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  {
423  CatalogTupleDelete(rel, &tup->t_self);
424  }
425  table_endscan(scan);
426 
428 }
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
TableScanDesc table_beginscan_catalog(Relation relation, int nkeys, struct ScanKeyData *key)
Definition: tableam.c:98
void CatalogTupleDelete(Relation heapRel, ItemPointer tid)
Definition: indexing.c:269
#define OidIsValid(objectId)
Definition: c.h:638
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
ItemPointerData t_self
Definition: htup.h:65
HeapTuple heap_getnext(TableScanDesc sscan, ScanDirection direction)
Definition: heapam.c:1290
#define RowExclusiveLock
Definition: lockdefs.h:38
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
static void table_endscan(TableScanDesc scan)
Definition: tableam.h:831
void ScanKeyInit(ScanKey entry, AttrNumber attributeNumber, StrategyNumber strategy, RegProcedure procedure, Datum argument)
Definition: scankey.c:76
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
#define BTEqualStrategyNumber
Definition: stratnum.h:31

◆ UpdateSubscriptionRelState()

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

Definition at line 287 of file pg_subscription.c.

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

289 {
290  Relation rel;
291  HeapTuple tup;
292  bool nulls[Natts_pg_subscription_rel];
293  Datum values[Natts_pg_subscription_rel];
294  bool replaces[Natts_pg_subscription_rel];
295 
296  LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
297 
298  rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
299 
300  /* Try finding existing mapping. */
302  ObjectIdGetDatum(relid),
303  ObjectIdGetDatum(subid));
304  if (!HeapTupleIsValid(tup))
305  elog(ERROR, "subscription table %u in subscription %u does not exist",
306  relid, subid);
307 
308  /* Update the tuple. */
309  memset(values, 0, sizeof(values));
310  memset(nulls, false, sizeof(nulls));
311  memset(replaces, false, sizeof(replaces));
312 
313  replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
314  values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
315 
316  replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
317  if (sublsn != InvalidXLogRecPtr)
318  values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
319  else
320  nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;
321 
322  tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
323  replaces);
324 
325  /* Update the catalog. */
326  CatalogTupleUpdate(rel, &tup->t_self, tup);
327 
328  /* Cleanup. */
329  table_close(rel, NoLock);
330 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:133
#define RelationGetDescr(relation)
Definition: rel.h:445
#define AccessShareLock
Definition: lockdefs.h:36
#define LSNGetDatum(X)
Definition: pg_lsn.h:22
#define ObjectIdGetDatum(X)
Definition: postgres.h:507
#define ERROR
Definition: elog.h:43
ItemPointerData t_self
Definition: htup.h:65
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
uintptr_t Datum
Definition: postgres.h:367
void LockSharedObject(Oid classid, Oid objid, uint16 objsubid, LOCKMODE lockmode)
Definition: lmgr.c:1002
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
Definition: regguts.h:298
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:224
#define CharGetDatum(X)
Definition: postgres.h:416
static Datum values[MAXATTR]
Definition: bootstrap.c:167
#define elog(elevel,...)
Definition: elog.h:226
#define SearchSysCacheCopy2(cacheId, key1, key2)
Definition: syscache.h:176
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:39
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:1113

Variable Documentation

◆ FormData_pg_subscription_rel

FormData_pg_subscription_rel

Definition at line 39 of file pg_subscription_rel.h.