pg_publication.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/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaddress.h"
#include "catalog/partition.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_namespace.h"
#include "catalog/pg_publication_rel.h"
#include "catalog/pg_type.h"
#include "commands/publicationcmds.h"
#include "funcapi.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"

Include dependency graph for pg_publication.c:

Go to the source code of this file.

Data Structures
struct	published_rel

Macros
#define	NUM_PUBLICATION_TABLES_ELEM   4

Functions
static void	check_publication_add_relation (Relation targetrel)
static void	check_publication_add_schema (Oid schemaid)
static bool	is_publishable_class (Oid relid, Form_pg_class reltuple)
bool	is_publishable_relation (Relation rel)
Datum	pg_relation_is_publishable (PG_FUNCTION_ARGS)
static bool	is_ancestor_member_tableinfos (Oid ancestor, List *table_infos)
static void	filter_partitions (List *table_infos)
bool	is_schema_publication (Oid pubid)
bool	check_and_fetch_column_list (Publication *pub, Oid relid, MemoryContext mcxt, Bitmapset **cols)
List *	GetPubPartitionOptionRelations (List *result, PublicationPartOpt pub_partopt, Oid relid)
Oid	GetTopMostAncestorInPublication (Oid puboid, List *ancestors, int *ancestor_level)
static int2vector *	attnumstoint2vector (Bitmapset *attrs)
ObjectAddress	publication_add_relation (Oid pubid, PublicationRelInfo *pri, bool if_not_exists)
Bitmapset *	pub_collist_validate (Relation targetrel, List *columns)
Bitmapset *	pub_collist_to_bitmapset (Bitmapset *columns, Datum pubcols, MemoryContext mcxt)
Bitmapset *	pub_form_cols_map (Relation relation, PublishGencolsType include_gencols_type)
ObjectAddress	publication_add_schema (Oid pubid, Oid schemaid, bool if_not_exists)
List *	GetRelationPublications (Oid relid)
List *	GetPublicationRelations (Oid pubid, PublicationPartOpt pub_partopt)
List *	GetAllTablesPublications (void)
List *	GetAllTablesPublicationRelations (bool pubviaroot)
List *	GetPublicationSchemas (Oid pubid)
List *	GetSchemaPublications (Oid schemaid)
List *	GetSchemaPublicationRelations (Oid schemaid, PublicationPartOpt pub_partopt)
List *	GetAllSchemaPublicationRelations (Oid pubid, PublicationPartOpt pub_partopt)
Publication *	GetPublication (Oid pubid)
Publication *	GetPublicationByName (const char *pubname, bool missing_ok)
Datum	pg_get_publication_tables (PG_FUNCTION_ARGS)

Macro Definition Documentation

NUM_PUBLICATION_TABLES_ELEM

#define NUM_PUBLICATION_TABLES_ELEM   4

Function Documentation

attnumstoint2vector()

static int2vector * attnumstoint2vector ( Bitmapset *  attrs )

static

Definition at line 405 of file pg_publication.c.

{
    int2vector *result;
    int         n = bms_num_members(attrs);
    int         i = -1;
    int         j = 0;
 
    result = buildint2vector(NULL, n);
 
    while ((i = bms_next_member(attrs, i)) >= 0)
    {
        Assert(i <= PG_INT16_MAX);
 
        result->values[j++] = (int16) i;
    }
 
    return result;
}

References Assert(), bms_next_member(), bms_num_members(), buildint2vector(), i, j, PG_INT16_MAX, and int2vector::values.

Referenced by publication_add_relation().

check_and_fetch_column_list()

bool check_and_fetch_column_list	(	Publication *	pub,
		Oid	relid,
		MemoryContext	mcxt,
		Bitmapset **	cols
	)

Definition at line 268 of file pg_publication.c.

{
    HeapTuple   cftuple;
    bool        found = false;
 
    if (pub->alltables)
        return false;
 
    cftuple = SearchSysCache2(PUBLICATIONRELMAP,
                              ObjectIdGetDatum(relid),
                              ObjectIdGetDatum(pub->oid));
    if (HeapTupleIsValid(cftuple))
    {
        Datum       cfdatum;
        bool        isnull;
 
        /* Lookup the column list attribute. */
        cfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, cftuple,
                                  Anum_pg_publication_rel_prattrs, &isnull);
 
        /* Was a column list found? */
        if (!isnull)
        {
            /* Build the column list bitmap in the given memory context. */
            if (cols)
                *cols = pub_collist_to_bitmapset(*cols, cfdatum, mcxt);
 
            found = true;
        }
 
        ReleaseSysCache(cftuple);
    }
 
    return found;
}

References Publication::alltables, HeapTupleIsValid, ObjectIdGetDatum(), Publication::oid, pub_collist_to_bitmapset(), ReleaseSysCache(), SearchSysCache2(), and SysCacheGetAttr().

Referenced by check_and_init_gencol(), pgoutput_column_list_init(), and pub_contains_invalid_column().

check_publication_add_relation()

static void check_publication_add_relation ( Relation  targetrel )

static

Definition at line 56 of file pg_publication.c.

{
    /* Must be a regular or partitioned table */
    if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION &&
        RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot add relation \"%s\" to publication",
                        RelationGetRelationName(targetrel)),
                 errdetail_relkind_not_supported(RelationGetForm(targetrel)->relkind)));
 
    /* Can't be system table */
    if (IsCatalogRelation(targetrel))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot add relation \"%s\" to publication",
                        RelationGetRelationName(targetrel)),
                 errdetail("This operation is not supported for system tables.")));
 
    /* UNLOGGED and TEMP relations cannot be part of publication. */
    if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot add relation \"%s\" to publication",
                        RelationGetRelationName(targetrel)),
                 errdetail("This operation is not supported for temporary tables.")));
    else if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot add relation \"%s\" to publication",
                        RelationGetRelationName(targetrel)),
                 errdetail("This operation is not supported for unlogged tables.")));
}

References ereport, errcode(), errdetail(), errdetail_relkind_not_supported(), errmsg(), ERROR, IsCatalogRelation(), RelationData::rd_rel, RelationGetForm, and RelationGetRelationName.

Referenced by publication_add_relation().

check_publication_add_schema()

static void check_publication_add_schema ( Oid  schemaid )

static

Definition at line 95 of file pg_publication.c.

{
    /* Can't be system namespace */
    if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot add schema \"%s\" to publication",
                        get_namespace_name(schemaid)),
                 errdetail("This operation is not supported for system schemas.")));
 
    /* Can't be temporary namespace */
    if (isAnyTempNamespace(schemaid))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot add schema \"%s\" to publication",
                        get_namespace_name(schemaid)),
                 errdetail("Temporary schemas cannot be replicated.")));
}

References ereport, errcode(), errdetail(), errmsg(), ERROR, get_namespace_name(), isAnyTempNamespace(), IsCatalogNamespace(), and IsToastNamespace().

Referenced by publication_add_schema().

filter_partitions()

static void filter_partitions ( List *  table_infos )

static

Definition at line 198 of file pg_publication.c.

{
    ListCell   *lc;
 
    foreach(lc, table_infos)
    {
        bool        skip = false;
        List       *ancestors = NIL;
        ListCell   *lc2;
        published_rel *table_info = (published_rel *) lfirst(lc);
 
        if (get_rel_relispartition(table_info->relid))
            ancestors = get_partition_ancestors(table_info->relid);
 
        foreach(lc2, ancestors)
        {
            Oid         ancestor = lfirst_oid(lc2);
 
            if (is_ancestor_member_tableinfos(ancestor, table_infos))
            {
                skip = true;
                break;
            }
        }
 
        if (skip)
            table_infos = foreach_delete_current(table_infos, lc);
    }
}

References foreach_delete_current, get_partition_ancestors(), get_rel_relispartition(), is_ancestor_member_tableinfos(), lfirst, lfirst_oid, NIL, published_rel::relid, and skip.

Referenced by pg_get_publication_tables().

GetAllSchemaPublicationRelations()

List * GetAllSchemaPublicationRelations	(	Oid	pubid,
		PublicationPartOpt	pub_partopt
	)

Definition at line 1046 of file pg_publication.c.

{
    List       *result = NIL;
    List       *pubschemalist = GetPublicationSchemas(pubid);
    ListCell   *cell;
 
    foreach(cell, pubschemalist)
    {
        Oid         schemaid = lfirst_oid(cell);
        List       *schemaRels = NIL;
 
        schemaRels = GetSchemaPublicationRelations(schemaid, pub_partopt);
        result = list_concat(result, schemaRels);
    }
 
    return result;
}

References GetPublicationSchemas(), GetSchemaPublicationRelations(), lfirst_oid, list_concat(), and NIL.

Referenced by AlterPublicationOptions(), InvalidatePubRelSyncCache(), and pg_get_publication_tables().

GetAllTablesPublicationRelations()

List * GetAllTablesPublicationRelations

(

bool

pubviaroot

)

Definition at line 864 of file pg_publication.c.

{
    Relation    classRel;
    ScanKeyData key[1];
    TableScanDesc scan;
    HeapTuple   tuple;
    List       *result = NIL;
 
    classRel = table_open(RelationRelationId, AccessShareLock);
 
    ScanKeyInit(&key[0],
                Anum_pg_class_relkind,
                BTEqualStrategyNumber, F_CHAREQ,
                CharGetDatum(RELKIND_RELATION));
 
    scan = table_beginscan_catalog(classRel, 1, key);
 
    while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
    {
        Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
        Oid         relid = relForm->oid;
 
        if (is_publishable_class(relid, relForm) &&
            !(relForm->relispartition && pubviaroot))
            result = lappend_oid(result, relid);
    }
 
    table_endscan(scan);
 
    if (pubviaroot)
    {
        ScanKeyInit(&key[0],
                    Anum_pg_class_relkind,
                    BTEqualStrategyNumber, F_CHAREQ,
                    CharGetDatum(RELKIND_PARTITIONED_TABLE));
 
        scan = table_beginscan_catalog(classRel, 1, key);
 
        while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
        {
            Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
            Oid         relid = relForm->oid;
 
            if (is_publishable_class(relid, relForm) &&
                !relForm->relispartition)
                result = lappend_oid(result, relid);
        }
 
        table_endscan(scan);
    }
 
    table_close(classRel, AccessShareLock);
    return result;
}

References AccessShareLock, BTEqualStrategyNumber, CharGetDatum(), ForwardScanDirection, GETSTRUCT(), heap_getnext(), is_publishable_class(), sort-test::key, lappend_oid(), NIL, ScanKeyInit(), table_beginscan_catalog(), table_close(), table_endscan(), and table_open().

Referenced by pg_get_publication_tables().

GetAllTablesPublications()

List * GetAllTablesPublications

(

void

)

Definition at line 823 of file pg_publication.c.

{
    List       *result;
    Relation    rel;
    ScanKeyData scankey;
    SysScanDesc scan;
    HeapTuple   tup;
 
    /* Find all publications that are marked as for all tables. */
    rel = table_open(PublicationRelationId, AccessShareLock);
 
    ScanKeyInit(&scankey,
                Anum_pg_publication_puballtables,
                BTEqualStrategyNumber, F_BOOLEQ,
                BoolGetDatum(true));
 
    scan = systable_beginscan(rel, InvalidOid, false,
                              NULL, 1, &scankey);
 
    result = NIL;
    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Oid         oid = ((Form_pg_publication) GETSTRUCT(tup))->oid;
 
        result = lappend_oid(result, oid);
    }
 
    systable_endscan(scan);
    table_close(rel, AccessShareLock);
 
    return result;
}

References AccessShareLock, BoolGetDatum(), BTEqualStrategyNumber, GETSTRUCT(), HeapTupleIsValid, InvalidOid, lappend_oid(), NIL, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by RelationBuildPublicationDesc().

GetPublication()

Publication * GetPublication

(

Oid

pubid

)

Definition at line 1070 of file pg_publication.c.

{
    HeapTuple   tup;
    Publication *pub;
    Form_pg_publication pubform;
 
    tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for publication %u", pubid);
 
    pubform = (Form_pg_publication) GETSTRUCT(tup);
 
    pub = (Publication *) palloc(sizeof(Publication));
    pub->oid = pubid;
    pub->name = pstrdup(NameStr(pubform->pubname));
    pub->alltables = pubform->puballtables;
    pub->pubactions.pubinsert = pubform->pubinsert;
    pub->pubactions.pubupdate = pubform->pubupdate;
    pub->pubactions.pubdelete = pubform->pubdelete;
    pub->pubactions.pubtruncate = pubform->pubtruncate;
    pub->pubviaroot = pubform->pubviaroot;
    pub->pubgencols_type = pubform->pubgencols;
 
    ReleaseSysCache(tup);
 
    return pub;
}

References Publication::alltables, elog, ERROR, GETSTRUCT(), HeapTupleIsValid, Publication::name, NameStr, ObjectIdGetDatum(), Publication::oid, palloc(), pstrdup(), Publication::pubactions, PublicationActions::pubdelete, Publication::pubgencols_type, PublicationActions::pubinsert, PublicationActions::pubtruncate, PublicationActions::pubupdate, Publication::pubviaroot, ReleaseSysCache(), and SearchSysCache1().

Referenced by GetPublicationByName(), pg_get_publication_tables(), pub_contains_invalid_column(), publication_add_relation(), and publication_add_schema().

GetPublicationByName()

Publication * GetPublicationByName	(	const char *	pubname,
		bool	missing_ok
	)

Definition at line 1102 of file pg_publication.c.

{
    Oid         oid;
 
    oid = get_publication_oid(pubname, missing_ok);
 
    return OidIsValid(oid) ? GetPublication(oid) : NULL;
}

References get_publication_oid(), GetPublication(), and OidIsValid.

Referenced by get_object_address_publication_rel(), get_object_address_publication_schema(), LoadPublications(), and pg_get_publication_tables().

GetPublicationRelations()

List * GetPublicationRelations	(	Oid	pubid,
		PublicationPartOpt	pub_partopt
	)

Definition at line 780 of file pg_publication.c.

{
    List       *result;
    Relation    pubrelsrel;
    ScanKeyData scankey;
    SysScanDesc scan;
    HeapTuple   tup;
 
    /* Find all publications associated with the relation. */
    pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock);
 
    ScanKeyInit(&scankey,
                Anum_pg_publication_rel_prpubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(pubid));
 
    scan = systable_beginscan(pubrelsrel, PublicationRelPrpubidIndexId,
                              true, NULL, 1, &scankey);
 
    result = NIL;
    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_pg_publication_rel pubrel;
 
        pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
        result = GetPubPartitionOptionRelations(result, pub_partopt,
                                                pubrel->prrelid);
    }
 
    systable_endscan(scan);
    table_close(pubrelsrel, AccessShareLock);
 
    /* Now sort and de-duplicate the result list */
    list_sort(result, list_oid_cmp);
    list_deduplicate_oid(result);
 
    return result;
}

References AccessShareLock, BTEqualStrategyNumber, GetPubPartitionOptionRelations(), GETSTRUCT(), HeapTupleIsValid, list_deduplicate_oid(), list_oid_cmp(), list_sort(), NIL, ObjectIdGetDatum(), ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by AlterPublicationOptions(), AlterPublicationSchemas(), AlterPublicationTables(), InvalidatePubRelSyncCache(), and pg_get_publication_tables().

GetPublicationSchemas()

List * GetPublicationSchemas

(

Oid

pubid

)

Definition at line 925 of file pg_publication.c.

{
    List       *result = NIL;
    Relation    pubschsrel;
    ScanKeyData scankey;
    SysScanDesc scan;
    HeapTuple   tup;
 
    /* Find all schemas associated with the publication */
    pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
 
    ScanKeyInit(&scankey,
                Anum_pg_publication_namespace_pnpubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(pubid));
 
    scan = systable_beginscan(pubschsrel,
                              PublicationNamespacePnnspidPnpubidIndexId,
                              true, NULL, 1, &scankey);
    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_pg_publication_namespace pubsch;
 
        pubsch = (Form_pg_publication_namespace) GETSTRUCT(tup);
 
        result = lappend_oid(result, pubsch->pnnspid);
    }
 
    systable_endscan(scan);
    table_close(pubschsrel, AccessShareLock);
 
    return result;
}

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

Referenced by AlterPublicationSchemas(), and GetAllSchemaPublicationRelations().

GetPubPartitionOptionRelations()

List * GetPubPartitionOptionRelations	(	List *	result,
		PublicationPartOpt	pub_partopt,
		Oid	relid
	)

Definition at line 310 of file pg_publication.c.

{
    if (get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE &&
        pub_partopt != PUBLICATION_PART_ROOT)
    {
        List       *all_parts = find_all_inheritors(relid, NoLock,
                                                    NULL);
 
        if (pub_partopt == PUBLICATION_PART_ALL)
            result = list_concat(result, all_parts);
        else if (pub_partopt == PUBLICATION_PART_LEAF)
        {
            ListCell   *lc;
 
            foreach(lc, all_parts)
            {
                Oid         partOid = lfirst_oid(lc);
 
                if (get_rel_relkind(partOid) != RELKIND_PARTITIONED_TABLE)
                    result = lappend_oid(result, partOid);
            }
        }
        else
            Assert(false);
    }
    else
        result = lappend_oid(result, relid);
 
    return result;
}

References Assert(), find_all_inheritors(), get_rel_relkind(), lappend_oid(), lfirst_oid, list_concat(), NoLock, PUBLICATION_PART_ALL, PUBLICATION_PART_LEAF, and PUBLICATION_PART_ROOT.

Referenced by AlterPublicationOptions(), GetPublicationRelations(), GetSchemaPublicationRelations(), publication_add_relation(), and RemovePublicationRelById().

GetRelationPublications()

List * GetRelationPublications

(

Oid

relid

)

Definition at line 751 of file pg_publication.c.

{
    List       *result = NIL;
    CatCList   *pubrellist;
    int         i;
 
    /* Find all publications associated with the relation. */
    pubrellist = SearchSysCacheList1(PUBLICATIONRELMAP,
                                     ObjectIdGetDatum(relid));
    for (i = 0; i < pubrellist->n_members; i++)
    {
        HeapTuple   tup = &pubrellist->members[i]->tuple;
        Oid         pubid = ((Form_pg_publication_rel) GETSTRUCT(tup))->prpubid;
 
        result = lappend_oid(result, pubid);
    }
 
    ReleaseSysCacheList(pubrellist);
 
    return result;
}

References GETSTRUCT(), i, lappend_oid(), catclist::members, catclist::n_members, NIL, ObjectIdGetDatum(), ReleaseSysCacheList, SearchSysCacheList1, and catctup::tuple.

Referenced by ATExecSetExpression(), ATPrepChangePersistence(), get_rel_sync_entry(), GetTopMostAncestorInPublication(), and RelationBuildPublicationDesc().

GetSchemaPublicationRelations()

List * GetSchemaPublicationRelations	(	Oid	schemaid,
		PublicationPartOpt	pub_partopt
	)

Definition at line 989 of file pg_publication.c.

{
    Relation    classRel;
    ScanKeyData key[1];
    TableScanDesc scan;
    HeapTuple   tuple;
    List       *result = NIL;
 
    Assert(OidIsValid(schemaid));
 
    classRel = table_open(RelationRelationId, AccessShareLock);
 
    ScanKeyInit(&key[0],
                Anum_pg_class_relnamespace,
                BTEqualStrategyNumber, F_OIDEQ,
                schemaid);
 
    /* get all the relations present in the specified schema */
    scan = table_beginscan_catalog(classRel, 1, key);
    while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
    {
        Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
        Oid         relid = relForm->oid;
        char        relkind;
 
        if (!is_publishable_class(relid, relForm))
            continue;
 
        relkind = get_rel_relkind(relid);
        if (relkind == RELKIND_RELATION)
            result = lappend_oid(result, relid);
        else if (relkind == RELKIND_PARTITIONED_TABLE)
        {
            List       *partitionrels = NIL;
 
            /*
             * It is quite possible that some of the partitions are in a
             * different schema than the parent table, so we need to get such
             * partitions separately.
             */
            partitionrels = GetPubPartitionOptionRelations(partitionrels,
                                                           pub_partopt,
                                                           relForm->oid);
            result = list_concat_unique_oid(result, partitionrels);
        }
    }
 
    table_endscan(scan);
    table_close(classRel, AccessShareLock);
    return result;
}

References AccessShareLock, Assert(), BTEqualStrategyNumber, ForwardScanDirection, get_rel_relkind(), GetPubPartitionOptionRelations(), GETSTRUCT(), heap_getnext(), is_publishable_class(), sort-test::key, lappend_oid(), list_concat_unique_oid(), NIL, OidIsValid, ScanKeyInit(), table_beginscan_catalog(), table_close(), table_endscan(), and table_open().

Referenced by GetAllSchemaPublicationRelations(), publication_add_schema(), and RemovePublicationSchemaById().

GetSchemaPublications()

List * GetSchemaPublications

(

Oid

schemaid

)

Definition at line 963 of file pg_publication.c.

{
    List       *result = NIL;
    CatCList   *pubschlist;
    int         i;
 
    /* Find all publications associated with the schema */
    pubschlist = SearchSysCacheList1(PUBLICATIONNAMESPACEMAP,
                                     ObjectIdGetDatum(schemaid));
    for (i = 0; i < pubschlist->n_members; i++)
    {
        HeapTuple   tup = &pubschlist->members[i]->tuple;
        Oid         pubid = ((Form_pg_publication_namespace) GETSTRUCT(tup))->pnpubid;
 
        result = lappend_oid(result, pubid);
    }
 
    ReleaseSysCacheList(pubschlist);
 
    return result;
}

References GETSTRUCT(), i, lappend_oid(), catclist::members, catclist::n_members, NIL, ObjectIdGetDatum(), ReleaseSysCacheList, SearchSysCacheList1, and catctup::tuple.

Referenced by get_rel_sync_entry(), GetTopMostAncestorInPublication(), and RelationBuildPublicationDesc().

GetTopMostAncestorInPublication()

Oid GetTopMostAncestorInPublication	(	Oid	puboid,
		List *	ancestors,
		int *	ancestor_level
	)

Definition at line 354 of file pg_publication.c.

{
    ListCell   *lc;
    Oid         topmost_relid = InvalidOid;
    int         level = 0;
 
    /*
     * Find the "topmost" ancestor that is in this publication.
     */
    foreach(lc, ancestors)
    {
        Oid         ancestor = lfirst_oid(lc);
        List       *apubids = GetRelationPublications(ancestor);
        List       *aschemaPubids = NIL;
 
        level++;
 
        if (list_member_oid(apubids, puboid))
        {
            topmost_relid = ancestor;
 
            if (ancestor_level)
                *ancestor_level = level;
        }
        else
        {
            aschemaPubids = GetSchemaPublications(get_rel_namespace(ancestor));
            if (list_member_oid(aschemaPubids, puboid))
            {
                topmost_relid = ancestor;
 
                if (ancestor_level)
                    *ancestor_level = level;
            }
        }
 
        list_free(apubids);
        list_free(aschemaPubids);
    }
 
    return topmost_relid;
}

References get_rel_namespace(), GetRelationPublications(), GetSchemaPublications(), InvalidOid, lfirst_oid, list_free(), list_member_oid(), and NIL.

Referenced by get_rel_sync_entry(), pub_contains_invalid_column(), and pub_rf_contains_invalid_column().

is_ancestor_member_tableinfos()

static bool is_ancestor_member_tableinfos ( Oid  ancestor, List *  table_infos  )

static

Definition at line 179 of file pg_publication.c.

{
    ListCell   *lc;
 
    foreach(lc, table_infos)
    {
        Oid         relid = ((published_rel *) lfirst(lc))->relid;
 
        if (relid == ancestor)
            return true;
    }
 
    return false;
}

References lfirst.

Referenced by filter_partitions().

is_publishable_class()

static bool is_publishable_class ( Oid  relid, Form_pg_class  reltuple  )

static

Definition at line 134 of file pg_publication.c.

{
    return (reltuple->relkind == RELKIND_RELATION ||
            reltuple->relkind == RELKIND_PARTITIONED_TABLE) &&
        !IsCatalogRelationOid(relid) &&
        reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
        relid >= FirstNormalObjectId;
}

References FirstNormalObjectId, and IsCatalogRelationOid().

Referenced by GetAllTablesPublicationRelations(), GetSchemaPublicationRelations(), is_publishable_relation(), and pg_relation_is_publishable().

is_publishable_relation()

bool is_publishable_relation

(

Relation

rel

)

Definition at line 147 of file pg_publication.c.

{
    return is_publishable_class(RelationGetRelid(rel), rel->rd_rel);
}

References is_publishable_class(), RelationData::rd_rel, and RelationGetRelid.

Referenced by pgoutput_change(), pgoutput_truncate(), and RelationBuildPublicationDesc().

is_schema_publication()

bool is_schema_publication

(

Oid

pubid

)

Definition at line 233 of file pg_publication.c.

{
    Relation    pubschsrel;
    ScanKeyData scankey;
    SysScanDesc scan;
    HeapTuple   tup;
    bool        result = false;
 
    pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
    ScanKeyInit(&scankey,
                Anum_pg_publication_namespace_pnpubid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(pubid));
 
    scan = systable_beginscan(pubschsrel,
                              PublicationNamespacePnnspidPnpubidIndexId,
                              true, NULL, 1, &scankey);
    tup = systable_getnext(scan);
    result = HeapTupleIsValid(tup);
 
    systable_endscan(scan);
    table_close(pubschsrel, AccessShareLock);
 
    return result;
}

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

Referenced by AlterPublicationOwner_internal(), and AlterPublicationTables().

pg_get_publication_tables()

Datum pg_get_publication_tables

(

PG_FUNCTION_ARGS

)

Definition at line 1117 of file pg_publication.c.

{
#define NUM_PUBLICATION_TABLES_ELEM 4
    FuncCallContext *funcctx;
    List       *table_infos = NIL;
 
    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL())
    {
        TupleDesc   tupdesc;
        MemoryContext oldcontext;
        ArrayType  *arr;
        Datum      *elems;
        int         nelems,
                    i;
        bool        viaroot = false;
 
        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();
 
        /* switch to memory context appropriate for multiple function calls */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
 
        /*
         * Deconstruct the parameter into elements where each element is a
         * publication name.
         */
        arr = PG_GETARG_ARRAYTYPE_P(0);
        deconstruct_array_builtin(arr, TEXTOID, &elems, NULL, &nelems);
 
        /* Get Oids of tables from each publication. */
        for (i = 0; i < nelems; i++)
        {
            Publication *pub_elem;
            List       *pub_elem_tables = NIL;
            ListCell   *lc;
 
            pub_elem = GetPublicationByName(TextDatumGetCString(elems[i]), false);
 
            /*
             * Publications support partitioned tables. If
             * publish_via_partition_root is false, all changes are replicated
             * using leaf partition identity and schema, so we only need
             * those. Otherwise, get the partitioned table itself.
             */
            if (pub_elem->alltables)
                pub_elem_tables = GetAllTablesPublicationRelations(pub_elem->pubviaroot);
            else
            {
                List       *relids,
                           *schemarelids;
 
                relids = GetPublicationRelations(pub_elem->oid,
                                                 pub_elem->pubviaroot ?
                                                 PUBLICATION_PART_ROOT :
                                                 PUBLICATION_PART_LEAF);
                schemarelids = GetAllSchemaPublicationRelations(pub_elem->oid,
                                                                pub_elem->pubviaroot ?
                                                                PUBLICATION_PART_ROOT :
                                                                PUBLICATION_PART_LEAF);
                pub_elem_tables = list_concat_unique_oid(relids, schemarelids);
            }
 
            /*
             * Record the published table and the corresponding publication so
             * that we can get row filters and column lists later.
             *
             * When a table is published by multiple publications, to obtain
             * all row filters and column lists, the structure related to this
             * table will be recorded multiple times.
             */
            foreach(lc, pub_elem_tables)
            {
                published_rel *table_info = (published_rel *) palloc(sizeof(published_rel));
 
                table_info->relid = lfirst_oid(lc);
                table_info->pubid = pub_elem->oid;
                table_infos = lappend(table_infos, table_info);
            }
 
            /* At least one publication is using publish_via_partition_root. */
            if (pub_elem->pubviaroot)
                viaroot = true;
        }
 
        /*
         * If the publication publishes partition changes via their respective
         * root partitioned tables, we must exclude partitions in favor of
         * including the root partitioned tables. Otherwise, the function
         * could return both the child and parent tables which could cause
         * data of the child table to be double-published on the subscriber
         * side.
         */
        if (viaroot)
            filter_partitions(table_infos);
 
        /* Construct a tuple descriptor for the result rows. */
        tupdesc = CreateTemplateTupleDesc(NUM_PUBLICATION_TABLES_ELEM);
        TupleDescInitEntry(tupdesc, (AttrNumber) 1, "pubid",
                           OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 2, "relid",
                           OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 3, "attrs",
                           INT2VECTOROID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber) 4, "qual",
                           PG_NODE_TREEOID, -1, 0);
 
        funcctx->tuple_desc = BlessTupleDesc(tupdesc);
        funcctx->user_fctx = table_infos;
 
        MemoryContextSwitchTo(oldcontext);
    }
 
    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();
    table_infos = (List *) funcctx->user_fctx;
 
    if (funcctx->call_cntr < list_length(table_infos))
    {
        HeapTuple   pubtuple = NULL;
        HeapTuple   rettuple;
        Publication *pub;
        published_rel *table_info = (published_rel *) list_nth(table_infos, funcctx->call_cntr);
        Oid         relid = table_info->relid;
        Oid         schemaid = get_rel_namespace(relid);
        Datum       values[NUM_PUBLICATION_TABLES_ELEM] = {0};
        bool        nulls[NUM_PUBLICATION_TABLES_ELEM] = {0};
 
        /*
         * Form tuple with appropriate data.
         */
 
        pub = GetPublication(table_info->pubid);
 
        values[0] = ObjectIdGetDatum(pub->oid);
        values[1] = ObjectIdGetDatum(relid);
 
        /*
         * We don't consider row filters or column lists for FOR ALL TABLES or
         * FOR TABLES IN SCHEMA publications.
         */
        if (!pub->alltables &&
            !SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
                                   ObjectIdGetDatum(schemaid),
                                   ObjectIdGetDatum(pub->oid)))
            pubtuple = SearchSysCacheCopy2(PUBLICATIONRELMAP,
                                           ObjectIdGetDatum(relid),
                                           ObjectIdGetDatum(pub->oid));
 
        if (HeapTupleIsValid(pubtuple))
        {
            /* Lookup the column list attribute. */
            values[2] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
                                        Anum_pg_publication_rel_prattrs,
                                        &(nulls[2]));
 
            /* Null indicates no filter. */
            values[3] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
                                        Anum_pg_publication_rel_prqual,
                                        &(nulls[3]));
        }
        else
        {
            nulls[2] = true;
            nulls[3] = true;
        }
 
        /* Show all columns when the column list is not specified. */
        if (nulls[2])
        {
            Relation    rel = table_open(relid, AccessShareLock);
            int         nattnums = 0;
            int16      *attnums;
            TupleDesc   desc = RelationGetDescr(rel);
            int         i;
 
            attnums = (int16 *) palloc(desc->natts * sizeof(int16));
 
            for (i = 0; i < desc->natts; i++)
            {
                Form_pg_attribute att = TupleDescAttr(desc, i);
 
                if (att->attisdropped)
                    continue;
 
                if (att->attgenerated)
                {
                    /* We only support replication of STORED generated cols. */
                    if (att->attgenerated != ATTRIBUTE_GENERATED_STORED)
                        continue;
 
                    /*
                     * User hasn't requested to replicate STORED generated
                     * cols.
                     */
                    if (pub->pubgencols_type != PUBLISH_GENCOLS_STORED)
                        continue;
                }
 
                attnums[nattnums++] = att->attnum;
            }
 
            if (nattnums > 0)
            {
                values[2] = PointerGetDatum(buildint2vector(attnums, nattnums));
                nulls[2] = false;
            }
 
            table_close(rel, AccessShareLock);
        }
 
        rettuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
 
        SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(rettuple));
    }
 
    SRF_RETURN_DONE(funcctx);
}

References AccessShareLock, Publication::alltables, BlessTupleDesc(), buildint2vector(), FuncCallContext::call_cntr, CreateTemplateTupleDesc(), deconstruct_array_builtin(), filter_partitions(), get_rel_namespace(), GetAllSchemaPublicationRelations(), GetAllTablesPublicationRelations(), GetPublication(), GetPublicationByName(), GetPublicationRelations(), heap_form_tuple(), HeapTupleGetDatum(), HeapTupleIsValid, i, if(), lappend(), lfirst_oid, list_concat_unique_oid(), list_length(), list_nth(), MemoryContextSwitchTo(), FuncCallContext::multi_call_memory_ctx, TupleDescData::natts, NIL, NUM_PUBLICATION_TABLES_ELEM, ObjectIdGetDatum(), Publication::oid, palloc(), PG_GETARG_ARRAYTYPE_P, PointerGetDatum(), Publication::pubgencols_type, published_rel::pubid, PUBLICATION_PART_LEAF, PUBLICATION_PART_ROOT, Publication::pubviaroot, RelationGetDescr, published_rel::relid, SearchSysCacheCopy2, SearchSysCacheExists2, SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, SRF_RETURN_NEXT, SysCacheGetAttr(), table_close(), table_open(), TextDatumGetCString, FuncCallContext::tuple_desc, TupleDescAttr(), TupleDescInitEntry(), FuncCallContext::user_fctx, and values.

pg_relation_is_publishable()

Datum pg_relation_is_publishable

(

PG_FUNCTION_ARGS

)

Definition at line 160 of file pg_publication.c.

{
    Oid         relid = PG_GETARG_OID(0);
    HeapTuple   tuple;
    bool        result;
 
    tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
    if (!HeapTupleIsValid(tuple))
        PG_RETURN_NULL();
    result = is_publishable_class(relid, (Form_pg_class) GETSTRUCT(tuple));
    ReleaseSysCache(tuple);
    PG_RETURN_BOOL(result);
}

References GETSTRUCT(), HeapTupleIsValid, is_publishable_class(), ObjectIdGetDatum(), PG_GETARG_OID, PG_RETURN_BOOL, PG_RETURN_NULL, ReleaseSysCache(), and SearchSysCache1().

pub_collist_to_bitmapset()

Bitmapset * pub_collist_to_bitmapset	(	Bitmapset *	columns,
		Datum	pubcols,
		MemoryContext	mcxt
	)

Definition at line 605 of file pg_publication.c.

{
    Bitmapset  *result = columns;
    ArrayType  *arr;
    int         nelems;
    int16      *elems;
    MemoryContext oldcxt = NULL;
 
    arr = DatumGetArrayTypeP(pubcols);
    nelems = ARR_DIMS(arr)[0];
    elems = (int16 *) ARR_DATA_PTR(arr);
 
    /* If a memory context was specified, switch to it. */
    if (mcxt)
        oldcxt = MemoryContextSwitchTo(mcxt);
 
    for (int i = 0; i < nelems; i++)
        result = bms_add_member(result, elems[i]);
 
    if (mcxt)
        MemoryContextSwitchTo(oldcxt);
 
    return result;
}

References ARR_DATA_PTR, ARR_DIMS, bms_add_member(), DatumGetArrayTypeP, i, and MemoryContextSwitchTo().

Referenced by AlterPublicationTables(), and check_and_fetch_column_list().

pub_collist_validate()

Bitmapset * pub_collist_validate	(	Relation	targetrel,
		List *	columns
	)

Definition at line 556 of file pg_publication.c.

{
    Bitmapset  *set = NULL;
    ListCell   *lc;
    TupleDesc   tupdesc = RelationGetDescr(targetrel);
 
    foreach(lc, columns)
    {
        char       *colname = strVal(lfirst(lc));
        AttrNumber  attnum = get_attnum(RelationGetRelid(targetrel), colname);
 
        if (attnum == InvalidAttrNumber)
            ereport(ERROR,
                    errcode(ERRCODE_UNDEFINED_COLUMN),
                    errmsg("column \"%s\" of relation \"%s\" does not exist",
                           colname, RelationGetRelationName(targetrel)));
 
        if (!AttrNumberIsForUserDefinedAttr(attnum))
            ereport(ERROR,
                    errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
                    errmsg("cannot use system column \"%s\" in publication column list",
                           colname));
 
        if (TupleDescAttr(tupdesc, attnum - 1)->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL)
            ereport(ERROR,
                    errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
                    errmsg("cannot use virtual generated column \"%s\" in publication column list",
                           colname));
 
        if (bms_is_member(attnum, set))
            ereport(ERROR,
                    errcode(ERRCODE_DUPLICATE_OBJECT),
                    errmsg("duplicate column \"%s\" in publication column list",
                           colname));
 
        set = bms_add_member(set, attnum);
    }
 
    return set;
}

References attnum, AttrNumberIsForUserDefinedAttr, bms_add_member(), bms_is_member(), ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg(), ERROR, get_attnum(), InvalidAttrNumber, lfirst, RelationGetDescr, RelationGetRelationName, RelationGetRelid, strVal, and TupleDescAttr().

Referenced by AlterPublicationTables(), and publication_add_relation().

pub_form_cols_map()

Bitmapset * pub_form_cols_map	(	Relation	relation,
		PublishGencolsType	include_gencols_type
	)

Definition at line 637 of file pg_publication.c.

{
    Bitmapset  *result = NULL;
    TupleDesc   desc = RelationGetDescr(relation);
 
    for (int i = 0; i < desc->natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(desc, i);
 
        if (att->attisdropped)
            continue;
 
        if (att->attgenerated)
        {
            /* We only support replication of STORED generated cols. */
            if (att->attgenerated != ATTRIBUTE_GENERATED_STORED)
                continue;
 
            /* User hasn't requested to replicate STORED generated cols. */
            if (include_gencols_type != PUBLISH_GENCOLS_STORED)
                continue;
        }
 
        result = bms_add_member(result, att->attnum);
    }
 
    return result;
}

References bms_add_member(), i, TupleDescData::natts, RelationGetDescr, and TupleDescAttr().

Referenced by pgoutput_column_list_init().

publication_add_relation()

ObjectAddress publication_add_relation	(	Oid	pubid,
		PublicationRelInfo *	pri,
		bool	if_not_exists
	)

Definition at line 428 of file pg_publication.c.

{
    Relation    rel;
    HeapTuple   tup;
    Datum       values[Natts_pg_publication_rel];
    bool        nulls[Natts_pg_publication_rel];
    Relation    targetrel = pri->relation;
    Oid         relid = RelationGetRelid(targetrel);
    Oid         pubreloid;
    Bitmapset  *attnums;
    Publication *pub = GetPublication(pubid);
    ObjectAddress myself,
                referenced;
    List       *relids = NIL;
    int         i;
 
    rel = table_open(PublicationRelRelationId, RowExclusiveLock);
 
    /*
     * Check for duplicates. Note that this does not really prevent
     * duplicates, it's here just to provide nicer error message in common
     * case. The real protection is the unique key on the catalog.
     */
    if (SearchSysCacheExists2(PUBLICATIONRELMAP, ObjectIdGetDatum(relid),
                              ObjectIdGetDatum(pubid)))
    {
        table_close(rel, RowExclusiveLock);
 
        if (if_not_exists)
            return InvalidObjectAddress;
 
        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("relation \"%s\" is already member of publication \"%s\"",
                        RelationGetRelationName(targetrel), pub->name)));
    }
 
    check_publication_add_relation(targetrel);
 
    /* Validate and translate column names into a Bitmapset of attnums. */
    attnums = pub_collist_validate(pri->relation, pri->columns);
 
    /* Form a tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
 
    pubreloid = GetNewOidWithIndex(rel, PublicationRelObjectIndexId,
                                   Anum_pg_publication_rel_oid);
    values[Anum_pg_publication_rel_oid - 1] = ObjectIdGetDatum(pubreloid);
    values[Anum_pg_publication_rel_prpubid - 1] =
        ObjectIdGetDatum(pubid);
    values[Anum_pg_publication_rel_prrelid - 1] =
        ObjectIdGetDatum(relid);
 
    /* Add qualifications, if available */
    if (pri->whereClause != NULL)
        values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(pri->whereClause));
    else
        nulls[Anum_pg_publication_rel_prqual - 1] = true;
 
    /* Add column list, if available */
    if (pri->columns)
        values[Anum_pg_publication_rel_prattrs - 1] = PointerGetDatum(attnumstoint2vector(attnums));
    else
        nulls[Anum_pg_publication_rel_prattrs - 1] = true;
 
    tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
    /* Insert tuple into catalog. */
    CatalogTupleInsert(rel, tup);
    heap_freetuple(tup);
 
    /* Register dependencies as needed */
    ObjectAddressSet(myself, PublicationRelRelationId, pubreloid);
 
    /* Add dependency on the publication */
    ObjectAddressSet(referenced, PublicationRelationId, pubid);
    recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
 
    /* Add dependency on the relation */
    ObjectAddressSet(referenced, RelationRelationId, relid);
    recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
 
    /* Add dependency on the objects mentioned in the qualifications */
    if (pri->whereClause)
        recordDependencyOnSingleRelExpr(&myself, pri->whereClause, relid,
                                        DEPENDENCY_NORMAL, DEPENDENCY_NORMAL,
                                        false);
 
    /* Add dependency on the columns, if any are listed */
    i = -1;
    while ((i = bms_next_member(attnums, i)) >= 0)
    {
        ObjectAddressSubSet(referenced, RelationRelationId, relid, i);
        recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
    }
 
    /* Close the table. */
    table_close(rel, RowExclusiveLock);
 
    /*
     * Invalidate relcache so that publication info is rebuilt.
     *
     * For the partitioned tables, we must invalidate all partitions contained
     * in the respective partition hierarchies, not just the one explicitly
     * mentioned in the publication. This is required because we implicitly
     * publish the child tables when the parent table is published.
     */
    relids = GetPubPartitionOptionRelations(relids, PUBLICATION_PART_ALL,
                                            relid);
 
    InvalidatePublicationRels(relids);
 
    return myself;
}

References attnumstoint2vector(), bms_next_member(), CatalogTupleInsert(), check_publication_add_relation(), PublicationRelInfo::columns, CStringGetTextDatum, DEPENDENCY_AUTO, DEPENDENCY_NORMAL, ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg(), ERROR, GetNewOidWithIndex(), GetPublication(), GetPubPartitionOptionRelations(), heap_form_tuple(), heap_freetuple(), i, InvalidatePublicationRels(), InvalidObjectAddress, Publication::name, NIL, nodeToString(), ObjectAddressSet, ObjectAddressSubSet, ObjectIdGetDatum(), PointerGetDatum(), pub_collist_validate(), PUBLICATION_PART_ALL, recordDependencyOn(), recordDependencyOnSingleRelExpr(), PublicationRelInfo::relation, RelationGetDescr, RelationGetRelationName, RelationGetRelid, RowExclusiveLock, SearchSysCacheExists2, table_close(), table_open(), values, and PublicationRelInfo::whereClause.

Referenced by PublicationAddTables().

publication_add_schema()

ObjectAddress publication_add_schema	(	Oid	pubid,
		Oid	schemaid,
		bool	if_not_exists
	)

Definition at line 670 of file pg_publication.c.

{
    Relation    rel;
    HeapTuple   tup;
    Datum       values[Natts_pg_publication_namespace];
    bool        nulls[Natts_pg_publication_namespace];
    Oid         psschid;
    Publication *pub = GetPublication(pubid);
    List       *schemaRels = NIL;
    ObjectAddress myself,
                referenced;
 
    rel = table_open(PublicationNamespaceRelationId, RowExclusiveLock);
 
    /*
     * Check for duplicates. Note that this does not really prevent
     * duplicates, it's here just to provide nicer error message in common
     * case. The real protection is the unique key on the catalog.
     */
    if (SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
                              ObjectIdGetDatum(schemaid),
                              ObjectIdGetDatum(pubid)))
    {
        table_close(rel, RowExclusiveLock);
 
        if (if_not_exists)
            return InvalidObjectAddress;
 
        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("schema \"%s\" is already member of publication \"%s\"",
                        get_namespace_name(schemaid), pub->name)));
    }
 
    check_publication_add_schema(schemaid);
 
    /* Form a tuple */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
 
    psschid = GetNewOidWithIndex(rel, PublicationNamespaceObjectIndexId,
                                 Anum_pg_publication_namespace_oid);
    values[Anum_pg_publication_namespace_oid - 1] = ObjectIdGetDatum(psschid);
    values[Anum_pg_publication_namespace_pnpubid - 1] =
        ObjectIdGetDatum(pubid);
    values[Anum_pg_publication_namespace_pnnspid - 1] =
        ObjectIdGetDatum(schemaid);
 
    tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
    /* Insert tuple into catalog */
    CatalogTupleInsert(rel, tup);
    heap_freetuple(tup);
 
    ObjectAddressSet(myself, PublicationNamespaceRelationId, psschid);
 
    /* Add dependency on the publication */
    ObjectAddressSet(referenced, PublicationRelationId, pubid);
    recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
 
    /* Add dependency on the schema */
    ObjectAddressSet(referenced, NamespaceRelationId, schemaid);
    recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
 
    /* Close the table */
    table_close(rel, RowExclusiveLock);
 
    /*
     * Invalidate relcache so that publication info is rebuilt. See
     * publication_add_relation for why we need to consider all the
     * partitions.
     */
    schemaRels = GetSchemaPublicationRelations(schemaid,
                                               PUBLICATION_PART_ALL);
    InvalidatePublicationRels(schemaRels);
 
    return myself;
}

References CatalogTupleInsert(), check_publication_add_schema(), DEPENDENCY_AUTO, ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg(), ERROR, get_namespace_name(), GetNewOidWithIndex(), GetPublication(), GetSchemaPublicationRelations(), heap_form_tuple(), heap_freetuple(), InvalidatePublicationRels(), InvalidObjectAddress, Publication::name, NIL, ObjectAddressSet, ObjectIdGetDatum(), PUBLICATION_PART_ALL, recordDependencyOn(), RelationGetDescr, RowExclusiveLock, SearchSysCacheExists2, table_close(), table_open(), and values.

Referenced by PublicationAddSchemas().