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_foreign_server.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "foreign/foreign.h"
#include "miscadmin.h"
#include "storage/lmgr.h"
#include "storage/lock.h"
#include "utils/acl.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
#include "utils/memutils.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 List *	textarray_to_stringlist (ArrayType *textarray)
void	GetPublicationsStr (List *publications, StringInfo dest, bool quote_literal)
Subscription *	GetSubscription (Oid subid, bool missing_ok, bool conninfo_needed, bool conninfo_aclcheck)
int	CountDBSubscriptions (Oid dbid)
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	HasSubscriptionTables (Oid subid)
List *	GetSubscriptionRelations (Oid subid, bool tables, bool sequences, 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 336 of file pg_subscription.c.

{
    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 relation %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 (XLogRecPtrIsValid(sublsn))
        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. */
    if (retain_lock)
    {
        table_close(rel, NoLock);
    }
    else
    {
        table_close(rel, RowExclusiveLock);
        UnlockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
    }
}

References AccessShareLock, CatalogTupleInsert(), CharGetDatum(), elog, ERROR, fb(), heap_form_tuple(), heap_freetuple(), HeapTupleIsValid, LockSharedObject(), LSNGetDatum(), NoLock, ObjectIdGetDatum(), RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy2, table_close(), table_open(), UnlockSharedObject(), values, and XLogRecPtrIsValid.

Referenced by AlterSubscription_refresh(), and binary_upgrade_add_sub_rel_state().

CountDBSubscriptions()

int CountDBSubscriptions

(

Oid

dbid

)

Definition at line 234 of file pg_subscription.c.

{
    int         nsubs = 0;
    Relation    rel;
    ScanKeyData scankey;
    SysScanDesc scan;
    HeapTuple   tup;
 
    rel = table_open(SubscriptionRelationId, RowExclusiveLock);
 
    ScanKeyInit(&scankey,
                Anum_pg_subscription_subdbid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(dbid));
 
    scan = systable_beginscan(rel, InvalidOid, false,
                              NULL, 1, &scankey);
 
    while (HeapTupleIsValid(tup = systable_getnext(scan)))
        nsubs++;
 
    systable_endscan(scan);
 
    table_close(rel, NoLock);
 
    return nsubs;
}

References BTEqualStrategyNumber, fb(), 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 266 of file pg_subscription.c.

{
    Relation    rel;
    bool        nulls[Natts_pg_subscription];
    bool        replaces[Natts_pg_subscription];
    Datum       values[Natts_pg_subscription];
    HeapTuple   tup;
 
    /* Look up the subscription in the catalog */
    rel = table_open(SubscriptionRelationId, RowExclusiveLock);
    tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
 
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for subscription %u", subid);
 
    /* Must only modify subscriptions belonging to the current database. */
    Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
 
    LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
 
    /* Form a new tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
    memset(replaces, false, sizeof(replaces));
 
    /* Set the subscription to disabled. */
    values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
    replaces[Anum_pg_subscription_subenabled - 1] = true;
 
    /* Update the catalog */
    tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
                            replaces);
    CatalogTupleUpdate(rel, &tup->t_self, tup);
    heap_freetuple(tup);
 
    table_close(rel, NoLock);
}

References AccessShareLock, Assert, BoolGetDatum(), CatalogTupleUpdate(), elog, ERROR, fb(), Form_pg_subscription, GETSTRUCT(), heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, LockSharedObject(), MyDatabaseId, NoLock, ObjectIdGetDatum(), RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy1, table_close(), table_open(), and values.

Referenced by DisableSubscriptionAndExit().

GetPublicationsStr()

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

Definition at line 50 of file pg_subscription.c.

{
    ListCell   *lc;
    bool        first = true;
 
    Assert(publications != NIL);
 
    foreach(lc, publications)
    {
        char       *pubname = strVal(lfirst(lc));
 
        if (quote_literal)
        {
            if (!first)
                appendStringInfoString(dest, ", ");
            appendStringInfoString(dest, quote_literal_cstr(pubname));
        }
        else
        {
            if (first)
                appendStringInfo(dest, _("\"%s\""), pubname);
            else
                appendStringInfo(dest, _(", \"%s\""), pubname);
        }
 
        first = false;
    }
}

References _, appendStringInfo(), appendStringInfoString(), Assert, fb(), lfirst, NIL, quote_literal(), quote_literal_cstr(), and strVal.

Referenced by check_publications_origin_sequences(), check_publications_origin_tables(), fetch_relation_list(), and fetch_remote_table_info().

GetSubscription()

Subscription * GetSubscription	(	Oid	subid,
		bool	missing_ok,
		bool	conninfo_needed,
		bool	conninfo_aclcheck
	)

Definition at line 88 of file pg_subscription.c.

{
    HeapTuple   tup;
    Subscription *sub;
    Form_pg_subscription subform;
    Datum       datum;
    bool        isnull;
    MemoryContext cxt;
    MemoryContext oldcxt;
 
    Assert(conninfo_needed || !conninfo_aclcheck);
 
    tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
 
    if (!HeapTupleIsValid(tup))
    {
        if (missing_ok)
            return NULL;
 
        elog(ERROR, "cache lookup failed for subscription %u", subid);
    }
 
    cxt = AllocSetContextCreate(CurrentMemoryContext, "subscription",
                                ALLOCSET_SMALL_SIZES);
    oldcxt = MemoryContextSwitchTo(cxt);
 
    subform = (Form_pg_subscription) GETSTRUCT(tup);
 
    /*
     * It's only safe to access subscriptions from another database from the
     * launcher process, which does not call GetSubscription().
     */
    Assert(subform->subdbid == MyDatabaseId);
 
    sub = palloc0_object(Subscription);
    sub->cxt = cxt;
    sub->oid = subid;
    sub->dbid = subform->subdbid;
    sub->skiplsn = subform->subskiplsn;
    sub->name = pstrdup(NameStr(subform->subname));
    sub->owner = subform->subowner;
    sub->enabled = subform->subenabled;
    sub->binary = subform->subbinary;
    sub->stream = subform->substream;
    sub->twophasestate = subform->subtwophasestate;
    sub->disableonerr = subform->subdisableonerr;
    sub->passwordrequired = subform->subpasswordrequired;
    sub->runasowner = subform->subrunasowner;
    sub->failover = subform->subfailover;
    sub->retaindeadtuples = subform->subretaindeadtuples;
    sub->maxretention = subform->submaxretention;
    sub->retentionactive = subform->subretentionactive;
    sub->conflictlogrelid = subform->subconflictlogrelid;
 
    if (conninfo_needed)
    {
        if (OidIsValid(subform->subserver))
        {
            AclResult   aclresult;
            ForeignServer *server;
 
            server = GetForeignServer(subform->subserver);
 
            if (conninfo_aclcheck)
            {
                /* recheck ACL if requested */
                aclresult = object_aclcheck(ForeignServerRelationId,
                                            subform->subserver,
                                            subform->subowner, ACL_USAGE);
 
                if (aclresult != ACLCHECK_OK)
                    ereport(ERROR,
                            (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                             errmsg("subscription owner \"%s\" does not have permission on foreign server \"%s\"",
                                    GetUserNameFromId(subform->subowner, false),
                                    server->servername)));
            }
 
            sub->conninfo = ForeignServerConnectionString(subform->subowner,
                                                          server);
        }
        else
        {
            datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
                                           tup,
                                           Anum_pg_subscription_subconninfo);
            sub->conninfo = TextDatumGetCString(datum);
        }
    }
 
    /* Get slotname */
    datum = SysCacheGetAttr(SUBSCRIPTIONOID,
                            tup,
                            Anum_pg_subscription_subslotname,
                            &isnull);
    if (!isnull)
        sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
    else
        sub->slotname = NULL;
 
    /* Get synccommit */
    datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
                                   tup,
                                   Anum_pg_subscription_subsynccommit);
    sub->synccommit = TextDatumGetCString(datum);
 
    /* Get walrcvtimeout */
    datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
                                   tup,
                                   Anum_pg_subscription_subwalrcvtimeout);
    sub->walrcvtimeout = TextDatumGetCString(datum);
 
    /* Get publications */
    datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
                                   tup,
                                   Anum_pg_subscription_subpublications);
    sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
 
    /* Get origin */
    datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
                                   tup,
                                   Anum_pg_subscription_suborigin);
    sub->origin = TextDatumGetCString(datum);
 
    /* Get conflict log destination */
    datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
                                   tup,
                                   Anum_pg_subscription_subconflictlogdest);
    sub->conflictlogdest = TextDatumGetCString(datum);
 
    /* Is the subscription owner a superuser? */
    sub->ownersuperuser = superuser_arg(sub->owner);
 
    ReleaseSysCache(tup);
 
    MemoryContextSwitchTo(oldcxt);
 
    return sub;
}

References ACL_USAGE, ACLCHECK_OK, ALLOCSET_SMALL_SIZES, AllocSetContextCreate, Assert, Subscription::binary, Subscription::conflictlogdest, Subscription::conflictlogrelid, Subscription::conninfo, CurrentMemoryContext, Subscription::cxt, DatumGetArrayTypeP, DatumGetName(), Subscription::dbid, Subscription::disableonerr, elog, Subscription::enabled, ereport, errcode(), errmsg, ERROR, Subscription::failover, fb(), ForeignServerConnectionString(), Form_pg_subscription, GetForeignServer(), GETSTRUCT(), GetUserNameFromId(), HeapTupleIsValid, Subscription::maxretention, MemoryContextSwitchTo(), MyDatabaseId, Subscription::name, NameStr, object_aclcheck(), ObjectIdGetDatum(), Subscription::oid, OidIsValid, Subscription::origin, Subscription::owner, Subscription::ownersuperuser, palloc0_object, Subscription::passwordrequired, pstrdup(), Subscription::publications, ReleaseSysCache(), Subscription::retaindeadtuples, Subscription::retentionactive, Subscription::runasowner, SearchSysCache1(), ForeignServer::servername, Subscription::skiplsn, Subscription::slotname, Subscription::stream, superuser_arg(), Subscription::synccommit, SysCacheGetAttr(), SysCacheGetAttrNotNull(), textarray_to_stringlist(), TextDatumGetCString, Subscription::twophasestate, and Subscription::walrcvtimeout.

Referenced by AlterSubscription(), InitializeLogRepWorker(), and maybe_reread_subscription().

GetSubscriptionRelations()

List * GetSubscriptionRelations	(	Oid	subid,
		bool	tables,
		bool	sequences,
		bool	not_ready
	)

Definition at line 634 of file pg_subscription.c.

{
    List       *res = NIL;
    Relation    rel;
    HeapTuple   tup;
    int         nkeys = 0;
    ScanKeyData skey[2];
    SysScanDesc scan;
 
    /* One or both of 'tables' and 'sequences' must be true. */
    Assert(tables || sequences);
 
    rel = table_open(SubscriptionRelRelationId, AccessShareLock);
 
    ScanKeyInit(&skey[nkeys++],
                Anum_pg_subscription_rel_srsubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(subid));
 
    if (not_ready)
        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;
        Datum       d;
        bool        isnull;
        char        relkind;
 
        subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
 
        /* Relation is either a sequence or a table */
        relkind = get_rel_relkind(subrel->srrelid);
        Assert(relkind == RELKIND_SEQUENCE || relkind == RELKIND_RELATION ||
               relkind == RELKIND_PARTITIONED_TABLE);
 
        /* Skip sequences if they were not requested */
        if ((relkind == RELKIND_SEQUENCE) && !sequences)
            continue;
 
        /* Skip tables if they were not requested */
        if ((relkind == RELKIND_RELATION ||
             relkind == RELKIND_PARTITIONED_TABLE) && !tables)
            continue;
 
        relstate = palloc_object(SubscriptionRelState);
        relstate->relid = subrel->srrelid;
        relstate->state = subrel->srsubstate;
        d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
                            Anum_pg_subscription_rel_srsublsn, &isnull);
        if (isnull)
            relstate->lsn = InvalidXLogRecPtr;
        else
            relstate->lsn = DatumGetLSN(d);
 
        res = lappend(res, relstate);
    }
 
    /* Cleanup */
    systable_endscan(scan);
    table_close(rel, AccessShareLock);
 
    return res;
}

References AccessShareLock, Assert, BTEqualStrategyNumber, CharGetDatum(), DatumGetLSN(), fb(), Form_pg_subscription_rel, get_rel_relkind(), GETSTRUCT(), HeapTupleIsValid, InvalidOid, InvalidXLogRecPtr, lappend(), SubscriptionRelState::lsn, NIL, ObjectIdGetDatum(), palloc_object, SubscriptionRelState::relid, ScanKeyInit(), sequences, SubscriptionRelState::state, SysCacheGetAttr(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by AlterSubscription_refresh(), AlterSubscription_refresh_seq(), DropSubscription(), and FetchRelationStates().

GetSubscriptionRelState()

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

Definition at line 456 of file pg_subscription.c.

{
    HeapTuple   tup;
    char        substate;
    bool        isnull;
    Datum       d;
    Relation    rel;
 
    /*
     * This is to avoid the race condition with AlterSubscription which tries
     * to remove this relstate.
     */
    rel = table_open(SubscriptionRelRelationId, AccessShareLock);
 
    /* Try finding the mapping. */
    tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
                          ObjectIdGetDatum(relid),
                          ObjectIdGetDatum(subid));
 
    if (!HeapTupleIsValid(tup))
    {
        table_close(rel, AccessShareLock);
        *sublsn = InvalidXLogRecPtr;
        return SUBREL_STATE_UNKNOWN;
    }
 
    /* Get the state. */
    substate = ((Form_pg_subscription_rel) GETSTRUCT(tup))->srsubstate;
 
    /* Get the LSN */
    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;
}

References AccessShareLock, DatumGetLSN(), fb(), Form_pg_subscription_rel, GETSTRUCT(), HeapTupleIsValid, InvalidXLogRecPtr, ObjectIdGetDatum(), ReleaseSysCache(), SearchSysCache2(), SysCacheGetAttr(), table_close(), and table_open().

Referenced by AlterSubscription_refresh(), logicalrep_rel_open(), LogicalRepSyncTableStart(), and wait_for_table_state_change().

HasSubscriptionTables()

bool HasSubscriptionTables

(

Oid

subid

)

Definition at line 585 of file pg_subscription.c.

{
    Relation    rel;
    ScanKeyData skey[1];
    SysScanDesc scan;
    HeapTuple   tup;
    bool        has_subtables = false;
 
    rel = table_open(SubscriptionRelRelationId, AccessShareLock);
 
    ScanKeyInit(&skey[0],
                Anum_pg_subscription_rel_srsubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(subid));
 
    scan = systable_beginscan(rel, InvalidOid, false,
                              NULL, 1, skey);
 
    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_pg_subscription_rel subrel;
        char        relkind;
 
        subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
        relkind = get_rel_relkind(subrel->srrelid);
 
        if (relkind == RELKIND_RELATION ||
            relkind == RELKIND_PARTITIONED_TABLE)
        {
            has_subtables = true;
            break;
        }
    }
 
    /* Cleanup */
    systable_endscan(scan);
    table_close(rel, AccessShareLock);
 
    return has_subtables;
}

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

Referenced by FetchRelationStates().

RemoveSubscriptionRel()

void RemoveSubscriptionRel	(	Oid	subid,
		Oid	relid
	)

Definition at line 506 of file pg_subscription.c.

{
    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)))
    {
        Form_pg_subscription_rel subrel;
 
        subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
 
        /*
         * We don't allow to drop the relation mapping when the table
         * synchronization is in progress unless the caller updates the
         * corresponding subscription as well. This is to ensure that we don't
         * leave tablesync slots or origins in the system when the
         * corresponding table is dropped. For sequences, however, it's ok to
         * drop them since no separate slots or origins are created during
         * synchronization.
         */
        if (!OidIsValid(subid) &&
            subrel->srsubstate != SUBREL_STATE_READY &&
            get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("could not drop relation mapping for subscription \"%s\"",
                            get_subscription_name(subrel->srsubid, false)),
                     errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
                               get_rel_name(relid), subrel->srsubstate),
 
            /*
             * translator: first %s is a SQL ALTER command and second %s is a
             * SQL DROP command
             */
                     errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
                             "ALTER SUBSCRIPTION ... ENABLE",
                             "DROP SUBSCRIPTION ...")));
        }
 
        CatalogTupleDelete(rel, &tup->t_self);
    }
    table_endscan(scan);
 
    table_close(rel, RowExclusiveLock);
}

References BTEqualStrategyNumber, CatalogTupleDelete(), ereport, errcode(), errdetail(), errhint(), errmsg, ERROR, fb(), Form_pg_subscription_rel, ForwardScanDirection, get_rel_name(), get_rel_relkind(), get_subscription_name(), GETSTRUCT(), heap_getnext(), HeapTupleIsValid, ObjectIdGetDatum(), OidIsValid, RowExclusiveLock, ScanKeyInit(), 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 310 of file pg_subscription.c.

{
    Datum      *elems;
    int         nelems,
                i;
    List       *res = NIL;
 
    deconstruct_array_builtin(textarray, TEXTOID, &elems, NULL, &nelems);
 
    if (nelems == 0)
        return NIL;
 
    for (i = 0; i < nelems; i++)
        res = lappend(res, makeString(TextDatumGetCString(elems[i])));
 
    return res;
}

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

Referenced by GetSubscription().

UpdateDeadTupleRetentionStatus()

void UpdateDeadTupleRetentionStatus	(	Oid	subid,
		bool	active
	)

Definition at line 711 of file pg_subscription.c.

{
    Relation    rel;
    bool        nulls[Natts_pg_subscription];
    bool        replaces[Natts_pg_subscription];
    Datum       values[Natts_pg_subscription];
    HeapTuple   tup;
 
    /* Look up the subscription in the catalog */
    rel = table_open(SubscriptionRelationId, RowExclusiveLock);
    tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
 
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for subscription %u", subid);
 
    /* Must only modify subscriptions belonging to the current database. */
    Assert(((Form_pg_subscription) GETSTRUCT(tup))->subdbid == MyDatabaseId);
 
    LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
 
    /* Form a new tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
    memset(replaces, false, sizeof(replaces));
 
    /* Set the subscription to disabled. */
    values[Anum_pg_subscription_subretentionactive - 1] = BoolGetDatum(active);
    replaces[Anum_pg_subscription_subretentionactive - 1] = true;
 
    /* Update the catalog */
    tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
                            replaces);
    CatalogTupleUpdate(rel, &tup->t_self, tup);
    heap_freetuple(tup);
 
    table_close(rel, NoLock);
}

References AccessShareLock, Assert, BoolGetDatum(), CatalogTupleUpdate(), elog, ERROR, fb(), Form_pg_subscription, GETSTRUCT(), heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, LockSharedObject(), MyDatabaseId, NoLock, ObjectIdGetDatum(), RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy1, 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 390 of file pg_subscription.c.

{
    Relation    rel;
    HeapTuple   tup;
    bool        nulls[Natts_pg_subscription_rel];
    Datum       values[Natts_pg_subscription_rel];
    bool        replaces[Natts_pg_subscription_rel];
 
    if (already_locked)
    {
#ifdef USE_ASSERT_CHECKING
        LOCKTAG     tag;
 
        Assert(CheckRelationOidLockedByMe(SubscriptionRelRelationId,
                                          RowExclusiveLock, true));
        SET_LOCKTAG_OBJECT(tag, InvalidOid, SubscriptionRelationId, subid, 0);
        Assert(LockHeldByMe(&tag, AccessShareLock, true));
#endif
 
        rel = table_open(SubscriptionRelRelationId, NoLock);
    }
    else
    {
        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 relation %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 (XLogRecPtrIsValid(sublsn))
        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);
}

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

Referenced by AlterSubscription_refresh_seq(), copy_sequence(), LogicalRepSyncTableStart(), ProcessSyncingTablesForApply(), and ProcessSyncingTablesForSync().