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 (PublicationRelInfo *pri)
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)
static bool	is_publishable_table (Oid tableoid)
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	is_table_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, AlterPublicationStmt *alter_stmt)
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)
static List *	get_relation_publications (Oid relid, bool except_flag)
List *	GetRelationIncludedPublications (Oid relid)
List *	GetRelationExcludedPublications (Oid relid)
static List *	get_publication_relations (Oid pubid, PublicationPartOpt pub_partopt, bool except_flag)
List *	GetIncludedPublicationRelations (Oid pubid, PublicationPartOpt pub_partopt)
List *	GetExcludedPublicationTables (Oid pubid, PublicationPartOpt pub_partopt)
List *	GetAllTablesPublications (void)
List *	GetAllPublicationRelations (Oid pubid, char relkind, 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)
static bool	is_table_publishable_in_publication (Oid relid, Publication *pub)
static Datum	pg_get_publication_tables (FunctionCallInfo fcinfo, ArrayType *pubnames, Oid target_relid, bool filter_by_relid, bool pub_missing_ok)
Datum	pg_get_publication_tables_a (PG_FUNCTION_ARGS)
Datum	pg_get_publication_tables_b (PG_FUNCTION_ARGS)
Datum	pg_get_publication_sequences (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 510 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(), fb(), i, j, PG_INT16_MAX, and result.

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 373 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, fb(), 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 ( PublicationRelInfo *  pri )

static

Definition at line 56 of file pg_publication.c.

{
    Relation    targetrel = pri->relation;
    const char *relname;
    const char *errormsg;
 
    if (pri->except)
    {
        relname = RelationGetQualifiedRelationName(targetrel);
        errormsg = gettext_noop("cannot specify relation \"%s\" in the publication EXCEPT clause");
    }
    else
    {
        relname = RelationGetRelationName(targetrel);
        errormsg = gettext_noop("cannot add relation \"%s\" to publication");
    }
 
    /* If in EXCEPT clause, must be root partitioned table */
    if (pri->except && targetrel->rd_rel->relispartition)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg(errormsg, relname),
                 errdetail("This operation is not supported for individual partitions.")));
 
    /* 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(errormsg, relname),
                 errdetail_relkind_not_supported(RelationGetForm(targetrel)->relkind)));
 
    /* Can't be system table */
    if (IsCatalogRelation(targetrel))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg(errormsg, relname),
                 errdetail("This operation is not supported for system tables.")));
 
    /* Can't be conflict log table */
    if (IsConflictLogTableNamespace(RelationGetNamespace(targetrel)))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg(errormsg, relname),
                 errdetail("This operation is not supported for conflict log 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(errormsg, relname),
                 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(errormsg, relname),
                 errdetail("This operation is not supported for unlogged tables.")));
}

References ereport, errcode(), errdetail(), errdetail_relkind_not_supported(), errmsg, ERROR, fb(), gettext_noop, IsCatalogRelation(), IsConflictLogTableNamespace(), RelationGetForm, RelationGetNamespace, RelationGetQualifiedRelationName(), RelationGetRelationName, and relname.

Referenced by publication_add_relation().

check_publication_add_schema()

static void check_publication_add_schema ( Oid  schemaid )

static

Definition at line 120 of file pg_publication.c.

{
    /* Can't be system namespace */
    if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid) ||
        IsConflictLogTableNamespace(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, fb(), get_namespace_name(), isAnyTempNamespace(), IsCatalogNamespace(), IsConflictLogTableNamespace(), and IsToastNamespace().

Referenced by publication_add_schema().

filter_partitions()

static void filter_partitions ( List *  table_infos )

static

Definition at line 260 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 fb(), foreach_delete_current, get_partition_ancestors(), get_rel_relispartition(), is_ancestor_member_tableinfos(), lfirst, lfirst_oid, NIL, and skip.

Referenced by pg_get_publication_tables().

get_publication_relations()

static List * get_publication_relations ( Oid  pubid, PublicationPartOpt  pub_partopt, bool  except_flag  )

static

Definition at line 935 of file pg_publication.c.

{
    List       *result;
    Relation    pubrelsrel;
    ScanKeyData scankey;
    SysScanDesc scan;
    HeapTuple   tup;
 
    /* Find all relations associated with the publication. */
    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);
 
        if (except_flag == pubrel->prexcept)
            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, fb(), Form_pg_publication_rel, GetPubPartitionOptionRelations(), GETSTRUCT(), HeapTupleIsValid, list_deduplicate_oid(), list_oid_cmp(), list_sort(), NIL, ObjectIdGetDatum(), result, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by GetExcludedPublicationTables(), and GetIncludedPublicationRelations().

get_relation_publications()

static List * get_relation_publications ( Oid  relid, bool  except_flag  )

static

Definition at line 886 of file pg_publication.c.

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

References fb(), Form_pg_publication_rel, GETSTRUCT(), i, lappend_oid(), NIL, ObjectIdGetDatum(), ReleaseSysCacheList, result, and SearchSysCacheList1.

Referenced by GetRelationExcludedPublications(), and GetRelationIncludedPublications().

GetAllPublicationRelations()

List * GetAllPublicationRelations	(	Oid	pubid,
		char	relkind,
		bool	pubviaroot
	)

Definition at line 1055 of file pg_publication.c.

{
    Relation    classRel;
    ScanKeyData key[1];
    TableScanDesc scan;
    HeapTuple   tuple;
    List       *result = NIL;
    List       *exceptlist = NIL;
 
    Assert(!(relkind == RELKIND_SEQUENCE && pubviaroot));
 
    /* EXCEPT filtering applies only to relations, not sequences */
    if (relkind == RELKIND_RELATION)
        exceptlist = GetExcludedPublicationTables(pubid, pubviaroot ?
                                                  PUBLICATION_PART_ROOT :
                                                  PUBLICATION_PART_LEAF);
 
    classRel = table_open(RelationRelationId, AccessShareLock);
 
    ScanKeyInit(&key[0],
                Anum_pg_class_relkind,
                BTEqualStrategyNumber, F_CHAREQ,
                CharGetDatum(relkind));
 
    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) &&
            !list_member_oid(exceptlist, relid))
            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 &&
                !list_member_oid(exceptlist, relid))
                result = lappend_oid(result, relid);
        }
 
        table_endscan(scan);
    }
 
    table_close(classRel, AccessShareLock);
    return result;
}

References AccessShareLock, Assert, BTEqualStrategyNumber, CharGetDatum(), fb(), ForwardScanDirection, GetExcludedPublicationTables(), GETSTRUCT(), heap_getnext(), is_publishable_class(), lappend_oid(), list_member_oid(), NIL, PUBLICATION_PART_LEAF, PUBLICATION_PART_ROOT, result, ScanKeyInit(), table_beginscan_catalog(), table_close(), table_endscan(), and table_open().

Referenced by pg_get_publication_sequences(), and pg_get_publication_tables().

GetAllSchemaPublicationRelations()

List * GetAllSchemaPublicationRelations	(	Oid	pubid,
		PublicationPartOpt	pub_partopt
	)

Definition at line 1248 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 fb(), GetPublicationSchemas(), GetSchemaPublicationRelations(), lfirst_oid, list_concat(), NIL, and result.

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

GetAllTablesPublications()

List * GetAllTablesPublications

(

void

)

Definition at line 1009 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, fb(), Form_pg_publication, GETSTRUCT(), HeapTupleIsValid, InvalidOid, lappend_oid(), NIL, result, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by RelationBuildPublicationDesc().

GetExcludedPublicationTables()

List * GetExcludedPublicationTables	(	Oid	pubid,
		PublicationPartOpt	pub_partopt
	)

Definition at line 998 of file pg_publication.c.

{
    Assert(GetPublication(pubid)->alltables);
 
    return get_publication_relations(pubid, pub_partopt, true);
}

References Assert, fb(), get_publication_relations(), and GetPublication().

Referenced by AlterPublicationTables(), and GetAllPublicationRelations().

GetIncludedPublicationRelations()

List * GetIncludedPublicationRelations	(	Oid	pubid,
		PublicationPartOpt	pub_partopt
	)

Definition at line 984 of file pg_publication.c.

{
    Assert(!GetPublication(pubid)->alltables);
 
    return get_publication_relations(pubid, pub_partopt, false);
}

References Assert, fb(), get_publication_relations(), and GetPublication().

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

GetPublication()

Publication * GetPublication

(

Oid

pubid

)

Definition at line 1272 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 = palloc_object(Publication);
    pub->oid = pubid;
    pub->name = pstrdup(NameStr(pubform->pubname));
    pub->alltables = pubform->puballtables;
    pub->allsequences = pubform->puballsequences;
    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::allsequences, Publication::alltables, elog, ERROR, fb(), Form_pg_publication, GETSTRUCT(), HeapTupleIsValid, Publication::name, NameStr, ObjectIdGetDatum(), Publication::oid, palloc_object, pstrdup(), Publication::pubactions, PublicationActions::pubdelete, Publication::pubgencols_type, PublicationActions::pubinsert, PublicationActions::pubtruncate, PublicationActions::pubupdate, Publication::pubviaroot, ReleaseSysCache(), and SearchSysCache1().

Referenced by GetExcludedPublicationTables(), GetIncludedPublicationRelations(), 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 1305 of file pg_publication.c.

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

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

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

GetPublicationSchemas()

List * GetPublicationSchemas

(

Oid

pubid

)

Definition at line 1127 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, fb(), Form_pg_publication_namespace, GETSTRUCT(), HeapTupleIsValid, lappend_oid(), NIL, ObjectIdGetDatum(), result, 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 415 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, fb(), find_all_inheritors(), get_rel_relkind(), lappend_oid(), lfirst_oid, list_concat(), NoLock, PUBLICATION_PART_ALL, PUBLICATION_PART_LEAF, PUBLICATION_PART_ROOT, and result.

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

GetRelationExcludedPublications()

List * GetRelationExcludedPublications

(

Oid

relid

)

Definition at line 922 of file pg_publication.c.

{
    return get_relation_publications(relid, true);
}

References get_relation_publications().

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

GetRelationIncludedPublications()

List * GetRelationIncludedPublications

(

Oid

relid

)

Definition at line 913 of file pg_publication.c.

{
    return get_relation_publications(relid, false);
}

References get_relation_publications().

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

GetSchemaPublicationRelations()

List * GetSchemaPublicationRelations	(	Oid	schemaid,
		PublicationPartOpt	pub_partopt
	)

Definition at line 1191 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,
                ObjectIdGetDatum(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, fb(), ForwardScanDirection, get_rel_relkind(), GetPubPartitionOptionRelations(), GETSTRUCT(), heap_getnext(), is_publishable_class(), lappend_oid(), list_concat_unique_oid(), NIL, ObjectIdGetDatum(), OidIsValid, result, 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 1165 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 fb(), Form_pg_publication_namespace, GETSTRUCT(), i, lappend_oid(), NIL, ObjectIdGetDatum(), ReleaseSysCacheList, result, and SearchSysCacheList1.

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

GetTopMostAncestorInPublication()

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

Definition at line 459 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 = GetRelationIncludedPublications(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 fb(), get_rel_namespace(), GetRelationIncludedPublications(), GetSchemaPublications(), InvalidOid, lfirst_oid, list_free(), list_member_oid(), and NIL.

Referenced by get_rel_sync_entry(), is_table_publishable_in_publication(), 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 241 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 fb(), and lfirst.

Referenced by filter_partitions().

is_publishable_class()

static bool is_publishable_class ( Oid  relid, Form_pg_class  reltuple  )

static

Definition at line 163 of file pg_publication.c.

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

References fb(), FirstNormalObjectId, IsCatalogRelationOid(), and IsConflictLogTableClass().

Referenced by GetAllPublicationRelations(), GetSchemaPublicationRelations(), is_publishable_relation(), is_publishable_table(), and pg_relation_is_publishable().

is_publishable_relation()

bool is_publishable_relation

(

Relation

rel

)

Definition at line 178 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_publishable_table()

static bool is_publishable_table ( Oid  tableoid )

static

Definition at line 188 of file pg_publication.c.

{
    HeapTuple   tuple;
    Form_pg_class relform;
 
    tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(tableoid));
    if (!HeapTupleIsValid(tuple))
        return false;
 
    relform = (Form_pg_class) GETSTRUCT(tuple);
 
    /*
     * is_publishable_class() includes sequences, so we need to explicitly
     * check the relkind to filter them out here.
     */
    if (relform->relkind != RELKIND_SEQUENCE &&
        is_publishable_class(tableoid, relform))
    {
        ReleaseSysCache(tuple);
        return true;
    }
 
    ReleaseSysCache(tuple);
    return false;
}

References fb(), GETSTRUCT(), HeapTupleIsValid, is_publishable_class(), ObjectIdGetDatum(), ReleaseSysCache(), and SearchSysCache1().

Referenced by pg_get_publication_tables().

is_schema_publication()

bool is_schema_publication

(

Oid

pubid

)

Definition at line 295 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, fb(), HeapTupleIsValid, ObjectIdGetDatum(), result, ScanKeyInit(), systable_beginscan(), systable_endscan(), systable_getnext(), table_close(), and table_open().

Referenced by AlterPublicationOwner_internal(), AlterPublicationTables(), and CheckAlterPublication().

is_table_publication()

bool is_table_publication

(

Oid

pubid

)

Definition at line 326 of file pg_publication.c.

{
    Relation    pubrelsrel;
    ScanKeyData scankey;
    SysScanDesc scan;
    HeapTuple   tup;
    bool        result = false;
 
    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);
    tup = systable_getnext(scan);
    if (HeapTupleIsValid(tup))
    {
        Form_pg_publication_rel pubrel;
 
        pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
 
        /*
         * For any publication, pg_publication_rel contains either only EXCEPT
         * entries or only explicitly included tables. Therefore, examining
         * the first tuple is sufficient to determine table inclusion.
         */
        result = !pubrel->prexcept;
    }
 
    systable_endscan(scan);
    table_close(pubrelsrel, AccessShareLock);
 
    return result;
}

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

Referenced by CheckAlterPublication().

is_table_publishable_in_publication()

static bool is_table_publishable_in_publication ( Oid  relid, Publication *  pub  )

static

Definition at line 1333 of file pg_publication.c.

{
    bool        relispartition;
    List       *ancestors = NIL;
 
    /*
     * For non-pubviaroot publications, a partitioned table is never the
     * effective published OID; only its leaf partitions can be.
     */
    if (!pub->pubviaroot && get_rel_relkind(relid) == RELKIND_PARTITIONED_TABLE)
        return false;
 
    relispartition = get_rel_relispartition(relid);
 
    if (relispartition)
        ancestors = get_partition_ancestors(relid);
 
    if (pub->alltables)
    {
        /*
         * ALL TABLES with pubviaroot includes only regular tables or top-most
         * partitioned tables -- never child partitions.
         */
        if (pub->pubviaroot && relispartition)
            return false;
 
        /*
         * For ALL TABLES publications, the table is published unless it
         * appears in the EXCEPT clause. Only the top-most can appear in the
         * EXCEPT clause, so exclusion must be evaluated at the top-most
         * ancestor if it has. These publications store only EXCEPT'ed tables
         * in pg_publication_rel, so checking existence is sufficient.
         *
         * Note that this existence check below would incorrectly return true
         * (published) for partitions when pubviaroot is enabled; however,
         * that case is already caught and returned false by the above check.
         */
        return !SearchSysCacheExists2(PUBLICATIONRELMAP,
                                      ObjectIdGetDatum(ancestors
                                                       ? llast_oid(ancestors) : relid),
                                      ObjectIdGetDatum(pub->oid));
    }
 
    /*
     * Non-ALL-TABLE publication cases.
     *
     * A table is published if it (or a containing schema) was explicitly
     * added, or if it is a partition whose ancestor was added.
     */
 
    /*
     * If an ancestor is published, the partition's status depends on
     * publish_via_partition_root value.
     *
     * If it's true, the ancestor's relation OID is the effective published
     * OID, so the partition itself should be excluded (return false).
     *
     * If it's false, the partition is covered by its ancestor's presence in
     * the publication, it should be included (return true).
     */
    if (relispartition &&
        OidIsValid(GetTopMostAncestorInPublication(pub->oid, ancestors, NULL)))
        return !pub->pubviaroot;
 
    /*
     * Check whether the table is explicitly published via pg_publication_rel
     * or pg_publication_namespace.
     */
    return (SearchSysCacheExists2(PUBLICATIONRELMAP,
                                  ObjectIdGetDatum(relid),
                                  ObjectIdGetDatum(pub->oid)) ||
            SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
                                  ObjectIdGetDatum(get_rel_namespace(relid)),
                                  ObjectIdGetDatum(pub->oid)));
}

References Publication::alltables, fb(), get_partition_ancestors(), get_rel_namespace(), get_rel_relispartition(), get_rel_relkind(), GetTopMostAncestorInPublication(), llast_oid, NIL, ObjectIdGetDatum(), Publication::oid, OidIsValid, Publication::pubviaroot, and SearchSysCacheExists2.

Referenced by pg_get_publication_tables().

pg_get_publication_sequences()

Datum pg_get_publication_sequences

(

PG_FUNCTION_ARGS

)

Definition at line 1696 of file pg_publication.c.

{
    FuncCallContext *funcctx;
    List       *sequences = NIL;
 
    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL())
    {
        char       *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0));
        Publication *publication;
        MemoryContext oldcontext;
 
        /* 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);
 
        publication = GetPublicationByName(pubname, false);
 
        if (publication->allsequences)
            sequences = GetAllPublicationRelations(publication->oid,
                                                   RELKIND_SEQUENCE,
                                                   false);
 
        funcctx->user_fctx = sequences;
 
        MemoryContextSwitchTo(oldcontext);
    }
 
    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();
    sequences = (List *) funcctx->user_fctx;
 
    if (funcctx->call_cntr < list_length(sequences))
    {
        Oid         relid = list_nth_oid(sequences, funcctx->call_cntr);
 
        SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid));
    }
 
    SRF_RETURN_DONE(funcctx);
}

References Publication::allsequences, fb(), GetAllPublicationRelations(), GetPublicationByName(), list_length(), list_nth_oid(), MemoryContextSwitchTo(), NIL, ObjectIdGetDatum(), Publication::oid, PG_GETARG_TEXT_PP, sequences, SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, SRF_RETURN_NEXT, and text_to_cstring().

pg_get_publication_tables()

static Datum pg_get_publication_tables ( FunctionCallInfo  fcinfo, ArrayType *  pubnames, Oid  target_relid, bool  filter_by_relid, bool  pub_missing_ok  )

static

Definition at line 1422 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;
        Datum      *elems;
        int         nelems,
                    i;
        bool        viaroot = false;
 
        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();
 
        /*
         * Preliminary check if the specified table can be published in the
         * first place. If not, we can return early without checking the given
         * publications and the table.
         */
        if (filter_by_relid && !is_publishable_table(target_relid))
            SRF_RETURN_DONE(funcctx);
 
        /* 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.
         */
        deconstruct_array_builtin(pubnames, 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]),
                                            pub_missing_ok);
 
            if (pub_elem == NULL)
                continue;
 
            if (filter_by_relid)
            {
                /* Check if the given table is published for the publication */
                if (is_table_publishable_in_publication(target_relid, pub_elem))
                {
                    pub_elem_tables = list_make1_oid(target_relid);
                }
            }
            else
            {
                /*
                 * 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 = GetAllPublicationRelations(pub_elem->oid,
                                                                 RELKIND_RELATION,
                                                                 pub_elem->pubviaroot);
                else
                {
                    List       *relids,
                               *schemarelids;
 
                    relids = GetIncludedPublicationRelations(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 = palloc_object(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);
 
        TupleDescFinalize(tupdesc);
        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 = palloc_array(int16, desc->natts);
 
            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(), CreateTemplateTupleDesc(), deconstruct_array_builtin(), fb(), filter_partitions(), get_rel_namespace(), GetAllPublicationRelations(), GetAllSchemaPublicationRelations(), GetIncludedPublicationRelations(), GetPublication(), GetPublicationByName(), heap_form_tuple(), HeapTupleGetDatum(), HeapTupleIsValid, i, is_publishable_table(), is_table_publishable_in_publication(), lappend(), lfirst_oid, list_concat_unique_oid(), list_length(), list_make1_oid, list_nth(), MemoryContextSwitchTo(), TupleDescData::natts, NIL, NUM_PUBLICATION_TABLES_ELEM, ObjectIdGetDatum(), Publication::oid, palloc_array, palloc_object, PointerGetDatum, Publication::pubgencols_type, PUBLICATION_PART_LEAF, PUBLICATION_PART_ROOT, 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, TupleDescAttr(), TupleDescFinalize(), TupleDescInitEntry(), and values.

Referenced by pg_get_publication_tables_a(), and pg_get_publication_tables_b().

pg_get_publication_tables_a()

Datum pg_get_publication_tables_a

(

PG_FUNCTION_ARGS

)

Definition at line 1669 of file pg_publication.c.

{
    /*
     * Get information for all tables in the given publications.
     * filter_by_relid is false so all tables are returned; pub_missing_ok is
     * false for backward compatibility.
     */
    return pg_get_publication_tables(fcinfo, PG_GETARG_ARRAYTYPE_P(0),
                                     InvalidOid, false, false);
}

References InvalidOid, pg_get_publication_tables(), and PG_GETARG_ARRAYTYPE_P.

pg_get_publication_tables_b()

Datum pg_get_publication_tables_b

(

PG_FUNCTION_ARGS

)

Definition at line 1681 of file pg_publication.c.

{
    /*
     * Get information for the specified table in the given publications. The
     * SQL-level function is declared STRICT, so target_relid is guaranteed to
     * be non-NULL here.
     */
    return pg_get_publication_tables(fcinfo, PG_GETARG_ARRAYTYPE_P(0),
                                     PG_GETARG_OID(1), true, true);
}

References pg_get_publication_tables(), PG_GETARG_ARRAYTYPE_P, and PG_GETARG_OID.

pg_relation_is_publishable()

Datum pg_relation_is_publishable

(

PG_FUNCTION_ARGS

)

Definition at line 222 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 fb(), GETSTRUCT(), HeapTupleIsValid, is_publishable_class(), ObjectIdGetDatum(), PG_GETARG_OID, PG_RETURN_BOOL, PG_RETURN_NULL, ReleaseSysCache(), result, and SearchSysCache1().

pub_collist_to_bitmapset()

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

Definition at line 734 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, fb(), i, MemoryContextSwitchTo(), and result.

Referenced by AlterPublicationTables(), and check_and_fetch_column_list().

pub_collist_validate()

Bitmapset * pub_collist_validate	(	Relation	targetrel,
		List *	columns
	)

Definition at line 685 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, fb(), 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 766 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(), fb(), i, TupleDescData::natts, RelationGetDescr, result, and TupleDescAttr().

Referenced by pgoutput_column_list_init().

publication_add_relation()

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

Definition at line 533 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;
    bool        inval_except_table;
 
    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(pri);
 
    /* 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);
    values[Anum_pg_publication_rel_prexcept - 1] =
        BoolGetDatum(pri->except);
 
    /* 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);
 
    /*
     * Determine whether EXCEPT tables require explicit relcache invalidation.
     *
     * For CREATE PUBLICATION with EXCEPT tables, invalidation is skipped
     * here, as CreatePublication() function invalidates all relations as part
     * of defining a FOR ALL TABLES publication.
     *
     * For ALTER PUBLICATION, invalidation is needed only when adding an
     * EXCEPT table to a publication already marked as ALL TABLES. For
     * publications that were originally empty or defined as ALL SEQUENCES and
     * are being converted to ALL TABLES, invalidation is skipped here, as
     * AlterPublicationAllFlags() function invalidates all relations while
     * marking the publication as ALL TABLES publication.
     */
    inval_except_table = (alter_stmt != NULL) && pub->alltables &&
        (alter_stmt->for_all_tables && pri->except);
 
    if (!pri->except || inval_except_table)
    {
        /*
         * 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 Publication::alltables, attnumstoint2vector(), bms_next_member(), BoolGetDatum(), CatalogTupleInsert(), check_publication_add_relation(), CStringGetTextDatum, DEPENDENCY_AUTO, DEPENDENCY_NORMAL, ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg, ERROR, fb(), 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(), RelationGetDescr, RelationGetRelationName, RelationGetRelid, RowExclusiveLock, SearchSysCacheExists2, table_close(), table_open(), and values.

Referenced by PublicationAddTables().

publication_add_schema()

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

Definition at line 799 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, fb(), 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().