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_rel *	Form_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)
List *	GetSubscriptionRelations (Oid subid)
List *	GetSubscriptionNotReadyRelations (Oid subid)

Variables
	FormData_pg_subscription_rel

Typedef Documentation

Form_pg_subscription_rel

typedef FormData_pg_subscription_rel* Form_pg_subscription_rel

Definition at line 40 of file pg_subscription_rel.h.

SubscriptionRelState

typedef struct SubscriptionRelState SubscriptionRelState

Function Documentation

AddSubscriptionRelState()

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

Definition at line 235 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().

{
    Relation    rel;
    HeapTuple   tup;
    bool        nulls[Natts_pg_subscription_rel];
    Datum       values[Natts_pg_subscription_rel];

    LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);

    rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

    /* Try finding existing mapping. */
    tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
                              ObjectIdGetDatum(relid),
                              ObjectIdGetDatum(subid));
    if (HeapTupleIsValid(tup))
        elog(ERROR, "subscription table %u in subscription %u already exists",
             relid, subid);

    /* Form the tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
    values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(subid);
    values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(relid);
    values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);
    if (sublsn != InvalidXLogRecPtr)
        values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
    else
        nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;

    tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);

    /* Insert tuple into catalog. */
    CatalogTupleInsert(rel, tup);

    heap_freetuple(tup);

    /* Cleanup. */
    table_close(rel, NoLock);
}

CATALOG()

CATALOG	(	pg_subscription_rel	,
		6102	,
		SubscriptionRelRelationId
	)

Definition at line 31 of file pg_subscription_rel.h.

{
    Oid         srsubid;        /* Oid of subscription */
    Oid         srrelid;        /* Oid of relation */
    char        srsubstate;     /* state of the relation in subscription */
    XLogRecPtr  srsublsn;       /* remote lsn of the state change used for
                                 * synchronization coordination */
} FormData_pg_subscription_rel;

GetSubscriptionNotReadyRelations()

List* GetSubscriptionNotReadyRelations

(

Oid

subid

)

Definition at line 478 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().

{
    List       *res = NIL;
    Relation    rel;
    HeapTuple   tup;
    int         nkeys = 0;
    ScanKeyData skey[2];
    SysScanDesc scan;

    rel = table_open(SubscriptionRelRelationId, AccessShareLock);

    ScanKeyInit(&skey[nkeys++],
                Anum_pg_subscription_rel_srsubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(subid));

    ScanKeyInit(&skey[nkeys++],
                Anum_pg_subscription_rel_srsubstate,
                BTEqualStrategyNumber, F_CHARNE,
                CharGetDatum(SUBREL_STATE_READY));

    scan = systable_beginscan(rel, InvalidOid, false,
                              NULL, nkeys, skey);

    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_pg_subscription_rel subrel;
        SubscriptionRelState *relstate;

        subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);

        relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
        relstate->relid = subrel->srrelid;
        relstate->state = subrel->srsubstate;
        relstate->lsn = subrel->srsublsn;

        res = lappend(res, relstate);
    }

    /* Cleanup */
    systable_endscan(scan);
    table_close(rel, AccessShareLock);

    return res;
}

GetSubscriptionRelations()

List* GetSubscriptionRelations

(

Oid

subid

)

Definition at line 431 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().

{
    List       *res = NIL;
    Relation    rel;
    HeapTuple   tup;
    int         nkeys = 0;
    ScanKeyData skey[2];
    SysScanDesc scan;

    rel = table_open(SubscriptionRelRelationId, AccessShareLock);

    ScanKeyInit(&skey[nkeys++],
                Anum_pg_subscription_rel_srsubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(subid));

    scan = systable_beginscan(rel, InvalidOid, false,
                              NULL, nkeys, skey);

    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_pg_subscription_rel subrel;
        SubscriptionRelState *relstate;

        subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);

        relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
        relstate->relid = subrel->srrelid;
        relstate->state = subrel->srsubstate;
        relstate->lsn = subrel->srsublsn;

        res = lappend(res, relstate);
    }

    /* Cleanup */
    systable_endscan(scan);
    table_close(rel, AccessShareLock);

    return res;
}

GetSubscriptionRelState()

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

Definition at line 332 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().

{
    Relation    rel;
    HeapTuple   tup;
    char        substate;
    bool        isnull;
    Datum       d;

    rel = table_open(SubscriptionRelRelationId, AccessShareLock);

    /* Try finding the mapping. */
    tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
                          ObjectIdGetDatum(relid),
                          ObjectIdGetDatum(subid));

    if (!HeapTupleIsValid(tup))
    {
        if (missing_ok)
        {
            table_close(rel, AccessShareLock);
            *sublsn = InvalidXLogRecPtr;
            return SUBREL_STATE_UNKNOWN;
        }

        elog(ERROR, "subscription table %u in subscription %u does not exist",
             relid, subid);
    }

    /* Get the state. */
    d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
                        Anum_pg_subscription_rel_srsubstate, &isnull);
    Assert(!isnull);
    substate = DatumGetChar(d);
    d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
                        Anum_pg_subscription_rel_srsublsn, &isnull);
    if (isnull)
        *sublsn = InvalidXLogRecPtr;
    else
        *sublsn = DatumGetLSN(d);

    /* Cleanup */
    ReleaseSysCache(tup);
    table_close(rel, AccessShareLock);

    return substate;
}

RemoveSubscriptionRel()

void RemoveSubscriptionRel	(	Oid	subid,
		Oid	relid
	)

Definition at line 385 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().

{
    Relation    rel;
    TableScanDesc scan;
    ScanKeyData skey[2];
    HeapTuple   tup;
    int         nkeys = 0;

    rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

    if (OidIsValid(subid))
    {
        ScanKeyInit(&skey[nkeys++],
                    Anum_pg_subscription_rel_srsubid,
                    BTEqualStrategyNumber,
                    F_OIDEQ,
                    ObjectIdGetDatum(subid));
    }

    if (OidIsValid(relid))
    {
        ScanKeyInit(&skey[nkeys++],
                    Anum_pg_subscription_rel_srrelid,
                    BTEqualStrategyNumber,
                    F_OIDEQ,
                    ObjectIdGetDatum(relid));
    }

    /* Do the search and delete what we found. */
    scan = table_beginscan_catalog(rel, nkeys, skey);
    while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
    {
        CatalogTupleDelete(rel, &tup->t_self);
    }
    table_endscan(scan);

    table_close(rel, RowExclusiveLock);
}

UpdateSubscriptionRelState()

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

Definition at line 281 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().

{
    Relation    rel;
    HeapTuple   tup;
    bool        nulls[Natts_pg_subscription_rel];
    Datum       values[Natts_pg_subscription_rel];
    bool        replaces[Natts_pg_subscription_rel];

    LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);

    rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

    /* Try finding existing mapping. */
    tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
                              ObjectIdGetDatum(relid),
                              ObjectIdGetDatum(subid));
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "subscription table %u in subscription %u does not exist",
             relid, subid);

    /* Update the tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
    memset(replaces, false, sizeof(replaces));

    replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
    values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(state);

    replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
    if (sublsn != InvalidXLogRecPtr)
        values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(sublsn);
    else
        nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;

    tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
                            replaces);

    /* Update the catalog. */
    CatalogTupleUpdate(rel, &tup->t_self, tup);

    /* Cleanup. */
    table_close(rel, NoLock);
}

Variable Documentation

FormData_pg_subscription_rel

FormData_pg_subscription_rel

Definition at line 38 of file pg_subscription_rel.h.