publicationcmds.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * publicationcmds.c
 *      publication manipulation
 *
 * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *      src/backend/commands/publicationcmds.c
 *
 *-------------------------------------------------------------------------
 */
 
#include "postgres.h"
 
#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/objectaddress.h"
#include "catalog/pg_database.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_namespace.h"
#include "catalog/pg_publication_rel.h"
#include "commands/dbcommands.h"
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/publicationcmds.h"
#include "miscadmin.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_clause.h"
#include "parser/parse_collate.h"
#include "parser/parse_relation.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"
#include "utils/varlena.h"
 
 
/*
 * Information used to validate the columns in the row filter expression. See
 * contain_invalid_rfcolumn_walker for details.
 */
typedef struct rf_context
{
    Bitmapset  *bms_replident;  /* bitset of replica identity columns */
    bool        pubviaroot;     /* true if we are validating the parent
                                 * relation's row filter */
    Oid         relid;          /* relid of the relation */
    Oid         parentid;       /* relid of the parent relation */
} rf_context;
 
static List *OpenTableList(List *tables);
static void CloseTableList(List *rels);
static void LockSchemaList(List *schemalist);
static void PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
                                 AlterPublicationStmt *stmt);
static void PublicationDropTables(Oid pubid, List *rels, bool missing_ok);
static void PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
                                  AlterPublicationStmt *stmt);
static void PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok);
 
 
static void
parse_publication_options(ParseState *pstate,
                          List *options,
                          bool *publish_given,
                          PublicationActions *pubactions,
                          bool *publish_via_partition_root_given,
                          bool *publish_via_partition_root,
                          bool *publish_generated_columns_given,
                          bool *publish_generated_columns)
{
    ListCell   *lc;
 
    *publish_given = false;
    *publish_via_partition_root_given = false;
    *publish_generated_columns_given = false;
 
    /* defaults */
    pubactions->pubinsert = true;
    pubactions->pubupdate = true;
    pubactions->pubdelete = true;
    pubactions->pubtruncate = true;
    *publish_via_partition_root = false;
    *publish_generated_columns = false;
 
    /* Parse options */
    foreach(lc, options)
    {
        DefElem    *defel = (DefElem *) lfirst(lc);
 
        if (strcmp(defel->defname, "publish") == 0)
        {
            char       *publish;
            List       *publish_list;
            ListCell   *lc2;
 
            if (*publish_given)
                errorConflictingDefElem(defel, pstate);
 
            /*
             * If publish option was given only the explicitly listed actions
             * should be published.
             */
            pubactions->pubinsert = false;
            pubactions->pubupdate = false;
            pubactions->pubdelete = false;
            pubactions->pubtruncate = false;
 
            *publish_given = true;
            publish = defGetString(defel);
 
            if (!SplitIdentifierString(publish, ',', &publish_list))
                ereport(ERROR,
                        (errcode(ERRCODE_SYNTAX_ERROR),
                         errmsg("invalid list syntax in parameter \"%s\"",
                                "publish")));
 
            /* Process the option list. */
            foreach(lc2, publish_list)
            {
                char       *publish_opt = (char *) lfirst(lc2);
 
                if (strcmp(publish_opt, "insert") == 0)
                    pubactions->pubinsert = true;
                else if (strcmp(publish_opt, "update") == 0)
                    pubactions->pubupdate = true;
                else if (strcmp(publish_opt, "delete") == 0)
                    pubactions->pubdelete = true;
                else if (strcmp(publish_opt, "truncate") == 0)
                    pubactions->pubtruncate = true;
                else
                    ereport(ERROR,
                            (errcode(ERRCODE_SYNTAX_ERROR),
                             errmsg("unrecognized value for publication option \"%s\": \"%s\"",
                                    "publish", publish_opt)));
            }
        }
        else if (strcmp(defel->defname, "publish_via_partition_root") == 0)
        {
            if (*publish_via_partition_root_given)
                errorConflictingDefElem(defel, pstate);
            *publish_via_partition_root_given = true;
            *publish_via_partition_root = defGetBoolean(defel);
        }
        else if (strcmp(defel->defname, "publish_generated_columns") == 0)
        {
            if (*publish_generated_columns_given)
                errorConflictingDefElem(defel, pstate);
            *publish_generated_columns_given = true;
            *publish_generated_columns = defGetBoolean(defel);
        }
        else
            ereport(ERROR,
                    (errcode(ERRCODE_SYNTAX_ERROR),
                     errmsg("unrecognized publication parameter: \"%s\"", defel->defname)));
    }
}
 
/*
 * Convert the PublicationObjSpecType list into schema oid list and
 * PublicationTable list.
 */
static void
ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
                           List **rels, List **schemas)
{
    ListCell   *cell;
    PublicationObjSpec *pubobj;
 
    if (!pubobjspec_list)
        return;
 
    foreach(cell, pubobjspec_list)
    {
        Oid         schemaid;
        List       *search_path;
 
        pubobj = (PublicationObjSpec *) lfirst(cell);
 
        switch (pubobj->pubobjtype)
        {
            case PUBLICATIONOBJ_TABLE:
                *rels = lappend(*rels, pubobj->pubtable);
                break;
            case PUBLICATIONOBJ_TABLES_IN_SCHEMA:
                schemaid = get_namespace_oid(pubobj->name, false);
 
                /* Filter out duplicates if user specifies "sch1, sch1" */
                *schemas = list_append_unique_oid(*schemas, schemaid);
                break;
            case PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA:
                search_path = fetch_search_path(false);
                if (search_path == NIL) /* nothing valid in search_path? */
                    ereport(ERROR,
                            errcode(ERRCODE_UNDEFINED_SCHEMA),
                            errmsg("no schema has been selected for CURRENT_SCHEMA"));
 
                schemaid = linitial_oid(search_path);
                list_free(search_path);
 
                /* Filter out duplicates if user specifies "sch1, sch1" */
                *schemas = list_append_unique_oid(*schemas, schemaid);
                break;
            default:
                /* shouldn't happen */
                elog(ERROR, "invalid publication object type %d", pubobj->pubobjtype);
                break;
        }
    }
}
 
/*
 * Returns true if any of the columns used in the row filter WHERE expression is
 * not part of REPLICA IDENTITY, false otherwise.
 */
static bool
contain_invalid_rfcolumn_walker(Node *node, rf_context *context)
{
    if (node == NULL)
        return false;
 
    if (IsA(node, Var))
    {
        Var        *var = (Var *) node;
        AttrNumber  attnum = var->varattno;
 
        /*
         * If pubviaroot is true, we are validating the row filter of the
         * parent table, but the bitmap contains the replica identity
         * information of the child table. So, get the column number of the
         * child table as parent and child column order could be different.
         */
        if (context->pubviaroot)
        {
            char       *colname = get_attname(context->parentid, attnum, false);
 
            attnum = get_attnum(context->relid, colname);
        }
 
        if (!bms_is_member(attnum - FirstLowInvalidHeapAttributeNumber,
                           context->bms_replident))
            return true;
    }
 
    return expression_tree_walker(node, contain_invalid_rfcolumn_walker,
                                  context);
}
 
/*
 * Check if all columns referenced in the filter expression are part of the
 * REPLICA IDENTITY index or not.
 *
 * Returns true if any invalid column is found.
 */
bool
pub_rf_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
                               bool pubviaroot)
{
    HeapTuple   rftuple;
    Oid         relid = RelationGetRelid(relation);
    Oid         publish_as_relid = RelationGetRelid(relation);
    bool        result = false;
    Datum       rfdatum;
    bool        rfisnull;
 
    /*
     * FULL means all columns are in the REPLICA IDENTITY, so all columns are
     * allowed in the row filter and we can skip the validation.
     */
    if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
        return false;
 
    /*
     * For a partition, if pubviaroot is true, find the topmost ancestor that
     * is published via this publication as we need to use its row filter
     * expression to filter the partition's changes.
     *
     * Note that even though the row filter used is for an ancestor, the
     * REPLICA IDENTITY used will be for the actual child table.
     */
    if (pubviaroot && relation->rd_rel->relispartition)
    {
        publish_as_relid
            = GetTopMostAncestorInPublication(pubid, ancestors, NULL);
 
        if (!OidIsValid(publish_as_relid))
            publish_as_relid = relid;
    }
 
    rftuple = SearchSysCache2(PUBLICATIONRELMAP,
                              ObjectIdGetDatum(publish_as_relid),
                              ObjectIdGetDatum(pubid));
 
    if (!HeapTupleIsValid(rftuple))
        return false;
 
    rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
                              Anum_pg_publication_rel_prqual,
                              &rfisnull);
 
    if (!rfisnull)
    {
        rf_context  context = {0};
        Node       *rfnode;
        Bitmapset  *bms = NULL;
 
        context.pubviaroot = pubviaroot;
        context.parentid = publish_as_relid;
        context.relid = relid;
 
        /* Remember columns that are part of the REPLICA IDENTITY */
        bms = RelationGetIndexAttrBitmap(relation,
                                         INDEX_ATTR_BITMAP_IDENTITY_KEY);
 
        context.bms_replident = bms;
        rfnode = stringToNode(TextDatumGetCString(rfdatum));
        result = contain_invalid_rfcolumn_walker(rfnode, &context);
    }
 
    ReleaseSysCache(rftuple);
 
    return result;
}
 
/*
 * Check for invalid columns in the publication table definition.
 *
 * This function evaluates two conditions:
 *
 * 1. Ensures that all columns referenced in the REPLICA IDENTITY are covered
 *    by the column list. If any column is missing, *invalid_column_list is set
 *    to true.
 * 2. Ensures that all the generated columns referenced in the REPLICA IDENTITY
 *    are published either by listing them in the column list or by enabling
 *    publish_generated_columns option. If any unpublished generated column is
 *    found, *invalid_gen_col is set to true.
 *
 * Returns true if any of the above conditions are not met.
 */
bool
pub_contains_invalid_column(Oid pubid, Relation relation, List *ancestors,
                            bool pubviaroot, bool pubgencols,
                            bool *invalid_column_list,
                            bool *invalid_gen_col)
{
    Oid         relid = RelationGetRelid(relation);
    Oid         publish_as_relid = RelationGetRelid(relation);
    Bitmapset  *idattrs;
    Bitmapset  *columns = NULL;
    TupleDesc   desc = RelationGetDescr(relation);
    Publication *pub;
    int         x;
 
    *invalid_column_list = false;
    *invalid_gen_col = false;
 
    /*
     * For a partition, if pubviaroot is true, find the topmost ancestor that
     * is published via this publication as we need to use its column list for
     * the changes.
     *
     * Note that even though the column list used is for an ancestor, the
     * REPLICA IDENTITY used will be for the actual child table.
     */
    if (pubviaroot && relation->rd_rel->relispartition)
    {
        publish_as_relid = GetTopMostAncestorInPublication(pubid, ancestors, NULL);
 
        if (!OidIsValid(publish_as_relid))
            publish_as_relid = relid;
    }
 
    /* Fetch the column list */
    pub = GetPublication(pubid);
    check_and_fetch_column_list(pub, publish_as_relid, NULL, &columns);
 
    if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
    {
        /* With REPLICA IDENTITY FULL, no column list is allowed. */
        *invalid_column_list = (columns != NULL);
 
        /*
         * As we don't allow a column list with REPLICA IDENTITY FULL, the
         * publish_generated_columns option must be set to true if the table
         * has any stored generated columns.
         */
        if (!pubgencols &&
            relation->rd_att->constr &&
            relation->rd_att->constr->has_generated_stored)
            *invalid_gen_col = true;
 
        if (*invalid_gen_col && *invalid_column_list)
            return true;
    }
 
    /* Remember columns that are part of the REPLICA IDENTITY */
    idattrs = RelationGetIndexAttrBitmap(relation,
                                         INDEX_ATTR_BITMAP_IDENTITY_KEY);
 
    /*
     * Attnums in the bitmap returned by RelationGetIndexAttrBitmap are offset
     * (to handle system columns the usual way), while column list does not
     * use offset, so we can't do bms_is_subset(). Instead, we have to loop
     * over the idattrs and check all of them are in the list.
     */
    x = -1;
    while ((x = bms_next_member(idattrs, x)) >= 0)
    {
        AttrNumber  attnum = (x + FirstLowInvalidHeapAttributeNumber);
        Form_pg_attribute att = TupleDescAttr(desc, attnum - 1);
 
        if (columns == NULL)
        {
            /*
             * The publish_generated_columns option must be set to true if the
             * REPLICA IDENTITY contains any stored generated column.
             */
            if (!pubgencols && att->attgenerated)
            {
                *invalid_gen_col = true;
                break;
            }
 
            /* Skip validating the column list since it is not defined */
            continue;
        }
 
        /*
         * If pubviaroot is true, we are validating the column list of the
         * parent table, but the bitmap contains the replica identity
         * information of the child table. The parent/child attnums may not
         * match, so translate them to the parent - get the attname from the
         * child, and look it up in the parent.
         */
        if (pubviaroot)
        {
            /* attribute name in the child table */
            char       *colname = get_attname(relid, attnum, false);
 
            /*
             * Determine the attnum for the attribute name in parent (we are
             * using the column list defined on the parent).
             */
            attnum = get_attnum(publish_as_relid, colname);
        }
 
        /* replica identity column, not covered by the column list */
        *invalid_column_list |= !bms_is_member(attnum, columns);
 
        if (*invalid_column_list && *invalid_gen_col)
            break;
    }
 
    bms_free(columns);
    bms_free(idattrs);
 
    return *invalid_column_list || *invalid_gen_col;
}
 
/* check_functions_in_node callback */
static bool
contain_mutable_or_user_functions_checker(Oid func_id, void *context)
{
    return (func_volatile(func_id) != PROVOLATILE_IMMUTABLE ||
            func_id >= FirstNormalObjectId);
}
 
/*
 * The row filter walker checks if the row filter expression is a "simple
 * expression".
 *
 * It allows only simple or compound expressions such as:
 * - (Var Op Const)
 * - (Var Op Var)
 * - (Var Op Const) AND/OR (Var Op Const)
 * - etc
 * (where Var is a column of the table this filter belongs to)
 *
 * The simple expression has the following restrictions:
 * - User-defined operators are not allowed;
 * - User-defined functions are not allowed;
 * - User-defined types are not allowed;
 * - User-defined collations are not allowed;
 * - Non-immutable built-in functions are not allowed;
 * - System columns are not allowed.
 *
 * NOTES
 *
 * We don't allow user-defined functions/operators/types/collations because
 * (a) if a user drops a user-defined object used in a row filter expression or
 * if there is any other error while using it, the logical decoding
 * infrastructure won't be able to recover from such an error even if the
 * object is recreated again because a historic snapshot is used to evaluate
 * the row filter;
 * (b) a user-defined function can be used to access tables that could have
 * unpleasant results because a historic snapshot is used. That's why only
 * immutable built-in functions are allowed in row filter expressions.
 *
 * We don't allow system columns because currently, we don't have that
 * information in the tuple passed to downstream. Also, as we don't replicate
 * those to subscribers, there doesn't seem to be a need for a filter on those
 * columns.
 *
 * We can allow other node types after more analysis and testing.
 */
static bool
check_simple_rowfilter_expr_walker(Node *node, ParseState *pstate)
{
    char       *errdetail_msg = NULL;
 
    if (node == NULL)
        return false;
 
    switch (nodeTag(node))
    {
        case T_Var:
            /* System columns are not allowed. */
            if (((Var *) node)->varattno < InvalidAttrNumber)
                errdetail_msg = _("System columns are not allowed.");
            break;
        case T_OpExpr:
        case T_DistinctExpr:
        case T_NullIfExpr:
            /* OK, except user-defined operators are not allowed. */
            if (((OpExpr *) node)->opno >= FirstNormalObjectId)
                errdetail_msg = _("User-defined operators are not allowed.");
            break;
        case T_ScalarArrayOpExpr:
            /* OK, except user-defined operators are not allowed. */
            if (((ScalarArrayOpExpr *) node)->opno >= FirstNormalObjectId)
                errdetail_msg = _("User-defined operators are not allowed.");
 
            /*
             * We don't need to check the hashfuncid and negfuncid of
             * ScalarArrayOpExpr as those functions are only built for a
             * subquery.
             */
            break;
        case T_RowCompareExpr:
            {
                ListCell   *opid;
 
                /* OK, except user-defined operators are not allowed. */
                foreach(opid, ((RowCompareExpr *) node)->opnos)
                {
                    if (lfirst_oid(opid) >= FirstNormalObjectId)
                    {
                        errdetail_msg = _("User-defined operators are not allowed.");
                        break;
                    }
                }
            }
            break;
        case T_Const:
        case T_FuncExpr:
        case T_BoolExpr:
        case T_RelabelType:
        case T_CollateExpr:
        case T_CaseExpr:
        case T_CaseTestExpr:
        case T_ArrayExpr:
        case T_RowExpr:
        case T_CoalesceExpr:
        case T_MinMaxExpr:
        case T_XmlExpr:
        case T_NullTest:
        case T_BooleanTest:
        case T_List:
            /* OK, supported */
            break;
        default:
            errdetail_msg = _("Only columns, constants, built-in operators, built-in data types, built-in collations, and immutable built-in functions are allowed.");
            break;
    }
 
    /*
     * For all the supported nodes, if we haven't already found a problem,
     * check the types, functions, and collations used in it.  We check List
     * by walking through each element.
     */
    if (!errdetail_msg && !IsA(node, List))
    {
        if (exprType(node) >= FirstNormalObjectId)
            errdetail_msg = _("User-defined types are not allowed.");
        else if (check_functions_in_node(node, contain_mutable_or_user_functions_checker,
                                         pstate))
            errdetail_msg = _("User-defined or built-in mutable functions are not allowed.");
        else if (exprCollation(node) >= FirstNormalObjectId ||
                 exprInputCollation(node) >= FirstNormalObjectId)
            errdetail_msg = _("User-defined collations are not allowed.");
    }
 
    /*
     * If we found a problem in this node, throw error now. Otherwise keep
     * going.
     */
    if (errdetail_msg)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("invalid publication WHERE expression"),
                 errdetail_internal("%s", errdetail_msg),
                 parser_errposition(pstate, exprLocation(node))));
 
    return expression_tree_walker(node, check_simple_rowfilter_expr_walker,
                                  pstate);
}
 
/*
 * Check if the row filter expression is a "simple expression".
 *
 * See check_simple_rowfilter_expr_walker for details.
 */
static bool
check_simple_rowfilter_expr(Node *node, ParseState *pstate)
{
    return check_simple_rowfilter_expr_walker(node, pstate);
}
 
/*
 * Transform the publication WHERE expression for all the relations in the list,
 * ensuring it is coerced to boolean and necessary collation information is
 * added if required, and add a new nsitem/RTE for the associated relation to
 * the ParseState's namespace list.
 *
 * Also check the publication row filter expression and throw an error if
 * anything not permitted or unexpected is encountered.
 */
static void
TransformPubWhereClauses(List *tables, const char *queryString,
                         bool pubviaroot)
{
    ListCell   *lc;
 
    foreach(lc, tables)
    {
        ParseNamespaceItem *nsitem;
        Node       *whereclause = NULL;
        ParseState *pstate;
        PublicationRelInfo *pri = (PublicationRelInfo *) lfirst(lc);
 
        if (pri->whereClause == NULL)
            continue;
 
        /*
         * If the publication doesn't publish changes via the root partitioned
         * table, the partition's row filter will be used. So disallow using
         * WHERE clause on partitioned table in this case.
         */
        if (!pubviaroot &&
            pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("cannot use publication WHERE clause for relation \"%s\"",
                            RelationGetRelationName(pri->relation)),
                     errdetail("WHERE clause cannot be used for a partitioned table when %s is false.",
                               "publish_via_partition_root")));
 
        /*
         * A fresh pstate is required so that we only have "this" table in its
         * rangetable
         */
        pstate = make_parsestate(NULL);
        pstate->p_sourcetext = queryString;
        nsitem = addRangeTableEntryForRelation(pstate, pri->relation,
                                               AccessShareLock, NULL,
                                               false, false);
        addNSItemToQuery(pstate, nsitem, false, true, true);
 
        whereclause = transformWhereClause(pstate,
                                           copyObject(pri->whereClause),
                                           EXPR_KIND_WHERE,
                                           "PUBLICATION WHERE");
 
        /* Fix up collation information */
        assign_expr_collations(pstate, whereclause);
 
        /*
         * We allow only simple expressions in row filters. See
         * check_simple_rowfilter_expr_walker.
         */
        check_simple_rowfilter_expr(whereclause, pstate);
 
        free_parsestate(pstate);
 
        pri->whereClause = whereclause;
    }
}
 
 
/*
 * Given a list of tables that are going to be added to a publication,
 * verify that they fulfill the necessary preconditions, namely: no tables
 * have a column list if any schema is published; and partitioned tables do
 * not have column lists if publish_via_partition_root is not set.
 *
 * 'publish_schema' indicates that the publication contains any TABLES IN
 * SCHEMA elements (newly added in this command, or preexisting).
 * 'pubviaroot' is the value of publish_via_partition_root.
 */
static void
CheckPubRelationColumnList(char *pubname, List *tables,
                           bool publish_schema, bool pubviaroot)
{
    ListCell   *lc;
 
    foreach(lc, tables)
    {
        PublicationRelInfo *pri = (PublicationRelInfo *) lfirst(lc);
 
        if (pri->columns == NIL)
            continue;
 
        /*
         * Disallow specifying column list if any schema is in the
         * publication.
         *
         * XXX We could instead just forbid the case when the publication
         * tries to publish the table with a column list and a schema for that
         * table. However, if we do that then we need a restriction during
         * ALTER TABLE ... SET SCHEMA to prevent such a case which doesn't
         * seem to be a good idea.
         */
        if (publish_schema)
            ereport(ERROR,
                    errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                    errmsg("cannot use column list for relation \"%s.%s\" in publication \"%s\"",
                           get_namespace_name(RelationGetNamespace(pri->relation)),
                           RelationGetRelationName(pri->relation), pubname),
                    errdetail("Column lists cannot be specified in publications containing FOR TABLES IN SCHEMA elements."));
 
        /*
         * If the publication doesn't publish changes via the root partitioned
         * table, the partition's column list will be used. So disallow using
         * a column list on the partitioned table in this case.
         */
        if (!pubviaroot &&
            pri->relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("cannot use column list for relation \"%s.%s\" in publication \"%s\"",
                            get_namespace_name(RelationGetNamespace(pri->relation)),
                            RelationGetRelationName(pri->relation), pubname),
                     errdetail("Column lists cannot be specified for partitioned tables when %s is false.",
                               "publish_via_partition_root")));
    }
}
 
/*
 * Create new publication.
 */
ObjectAddress
CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
{
    Relation    rel;
    ObjectAddress myself;
    Oid         puboid;
    bool        nulls[Natts_pg_publication];
    Datum       values[Natts_pg_publication];
    HeapTuple   tup;
    bool        publish_given;
    PublicationActions pubactions;
    bool        publish_via_partition_root_given;
    bool        publish_via_partition_root;
    bool        publish_generated_columns_given;
    bool        publish_generated_columns;
    AclResult   aclresult;
    List       *relations = NIL;
    List       *schemaidlist = NIL;
 
    /* must have CREATE privilege on database */
    aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, GetUserId(), ACL_CREATE);
    if (aclresult != ACLCHECK_OK)
        aclcheck_error(aclresult, OBJECT_DATABASE,
                       get_database_name(MyDatabaseId));
 
    /* FOR ALL TABLES requires superuser */
    if (stmt->for_all_tables && !superuser())
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("must be superuser to create FOR ALL TABLES publication")));
 
    rel = table_open(PublicationRelationId, RowExclusiveLock);
 
    /* Check if name is used */
    puboid = GetSysCacheOid1(PUBLICATIONNAME, Anum_pg_publication_oid,
                             CStringGetDatum(stmt->pubname));
    if (OidIsValid(puboid))
        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("publication \"%s\" already exists",
                        stmt->pubname)));
 
    /* Form a tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
 
    values[Anum_pg_publication_pubname - 1] =
        DirectFunctionCall1(namein, CStringGetDatum(stmt->pubname));
    values[Anum_pg_publication_pubowner - 1] = ObjectIdGetDatum(GetUserId());
 
    parse_publication_options(pstate,
                              stmt->options,
                              &publish_given, &pubactions,
                              &publish_via_partition_root_given,
                              &publish_via_partition_root,
                              &publish_generated_columns_given,
                              &publish_generated_columns);
 
    puboid = GetNewOidWithIndex(rel, PublicationObjectIndexId,
                                Anum_pg_publication_oid);
    values[Anum_pg_publication_oid - 1] = ObjectIdGetDatum(puboid);
    values[Anum_pg_publication_puballtables - 1] =
        BoolGetDatum(stmt->for_all_tables);
    values[Anum_pg_publication_pubinsert - 1] =
        BoolGetDatum(pubactions.pubinsert);
    values[Anum_pg_publication_pubupdate - 1] =
        BoolGetDatum(pubactions.pubupdate);
    values[Anum_pg_publication_pubdelete - 1] =
        BoolGetDatum(pubactions.pubdelete);
    values[Anum_pg_publication_pubtruncate - 1] =
        BoolGetDatum(pubactions.pubtruncate);
    values[Anum_pg_publication_pubviaroot - 1] =
        BoolGetDatum(publish_via_partition_root);
    values[Anum_pg_publication_pubgencols - 1] =
        BoolGetDatum(publish_generated_columns);
 
    tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
    /* Insert tuple into catalog. */
    CatalogTupleInsert(rel, tup);
    heap_freetuple(tup);
 
    recordDependencyOnOwner(PublicationRelationId, puboid, GetUserId());
 
    ObjectAddressSet(myself, PublicationRelationId, puboid);
 
    /* Make the changes visible. */
    CommandCounterIncrement();
 
    /* Associate objects with the publication. */
    if (stmt->for_all_tables)
    {
        /* Invalidate relcache so that publication info is rebuilt. */
        CacheInvalidateRelcacheAll();
    }
    else
    {
        ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
                                   &schemaidlist);
 
        /* FOR TABLES IN SCHEMA requires superuser */
        if (schemaidlist != NIL && !superuser())
            ereport(ERROR,
                    errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                    errmsg("must be superuser to create FOR TABLES IN SCHEMA publication"));
 
        if (relations != NIL)
        {
            List       *rels;
 
            rels = OpenTableList(relations);
            TransformPubWhereClauses(rels, pstate->p_sourcetext,
                                     publish_via_partition_root);
 
            CheckPubRelationColumnList(stmt->pubname, rels,
                                       schemaidlist != NIL,
                                       publish_via_partition_root);
 
            PublicationAddTables(puboid, rels, true, NULL);
            CloseTableList(rels);
        }
 
        if (schemaidlist != NIL)
        {
            /*
             * Schema lock is held until the publication is created to prevent
             * concurrent schema deletion.
             */
            LockSchemaList(schemaidlist);
            PublicationAddSchemas(puboid, schemaidlist, true, NULL);
        }
    }
 
    table_close(rel, RowExclusiveLock);
 
    InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0);
 
    if (wal_level != WAL_LEVEL_LOGICAL)
        ereport(WARNING,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("\"wal_level\" is insufficient to publish logical changes"),
                 errhint("Set \"wal_level\" to \"logical\" before creating subscriptions.")));
 
    return myself;
}
 
/*
 * Change options of a publication.
 */
static void
AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
                        Relation rel, HeapTuple tup)
{
    bool        nulls[Natts_pg_publication];
    bool        replaces[Natts_pg_publication];
    Datum       values[Natts_pg_publication];
    bool        publish_given;
    PublicationActions pubactions;
    bool        publish_via_partition_root_given;
    bool        publish_via_partition_root;
    bool        publish_generated_columns_given;
    bool        publish_generated_columns;
    ObjectAddress obj;
    Form_pg_publication pubform;
    List       *root_relids = NIL;
    ListCell   *lc;
 
    parse_publication_options(pstate,
                              stmt->options,
                              &publish_given, &pubactions,
                              &publish_via_partition_root_given,
                              &publish_via_partition_root,
                              &publish_generated_columns_given,
                              &publish_generated_columns);
 
    pubform = (Form_pg_publication) GETSTRUCT(tup);
 
    /*
     * If the publication doesn't publish changes via the root partitioned
     * table, the partition's row filter and column list will be used. So
     * disallow using WHERE clause and column lists on partitioned table in
     * this case.
     */
    if (!pubform->puballtables && publish_via_partition_root_given &&
        !publish_via_partition_root)
    {
        /*
         * Lock the publication so nobody else can do anything with it. This
         * prevents concurrent alter to add partitioned table(s) with WHERE
         * clause(s) and/or column lists which we don't allow when not
         * publishing via root.
         */
        LockDatabaseObject(PublicationRelationId, pubform->oid, 0,
                           AccessShareLock);
 
        root_relids = GetPublicationRelations(pubform->oid,
                                              PUBLICATION_PART_ROOT);
 
        foreach(lc, root_relids)
        {
            Oid         relid = lfirst_oid(lc);
            HeapTuple   rftuple;
            char        relkind;
            char       *relname;
            bool        has_rowfilter;
            bool        has_collist;
 
            /*
             * Beware: we don't have lock on the relations, so cope silently
             * with the cache lookups returning NULL.
             */
 
            rftuple = SearchSysCache2(PUBLICATIONRELMAP,
                                      ObjectIdGetDatum(relid),
                                      ObjectIdGetDatum(pubform->oid));
            if (!HeapTupleIsValid(rftuple))
                continue;
            has_rowfilter = !heap_attisnull(rftuple, Anum_pg_publication_rel_prqual, NULL);
            has_collist = !heap_attisnull(rftuple, Anum_pg_publication_rel_prattrs, NULL);
            if (!has_rowfilter && !has_collist)
            {
                ReleaseSysCache(rftuple);
                continue;
            }
 
            relkind = get_rel_relkind(relid);
            if (relkind != RELKIND_PARTITIONED_TABLE)
            {
                ReleaseSysCache(rftuple);
                continue;
            }
            relname = get_rel_name(relid);
            if (relname == NULL)    /* table concurrently dropped */
            {
                ReleaseSysCache(rftuple);
                continue;
            }
 
            if (has_rowfilter)
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("cannot set parameter \"%s\" to false for publication \"%s\"",
                                "publish_via_partition_root",
                                stmt->pubname),
                         errdetail("The publication contains a WHERE clause for partitioned table \"%s\", which is not allowed when \"%s\" is false.",
                                   relname, "publish_via_partition_root")));
            Assert(has_collist);
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("cannot set parameter \"%s\" to false for publication \"%s\"",
                            "publish_via_partition_root",
                            stmt->pubname),
                     errdetail("The publication contains a column list for partitioned table \"%s\", which is not allowed when \"%s\" is false.",
                               relname, "publish_via_partition_root")));
        }
    }
 
    /* Everything ok, form a new tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
    memset(replaces, false, sizeof(replaces));
 
    if (publish_given)
    {
        values[Anum_pg_publication_pubinsert - 1] = BoolGetDatum(pubactions.pubinsert);
        replaces[Anum_pg_publication_pubinsert - 1] = true;
 
        values[Anum_pg_publication_pubupdate - 1] = BoolGetDatum(pubactions.pubupdate);
        replaces[Anum_pg_publication_pubupdate - 1] = true;
 
        values[Anum_pg_publication_pubdelete - 1] = BoolGetDatum(pubactions.pubdelete);
        replaces[Anum_pg_publication_pubdelete - 1] = true;
 
        values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate);
        replaces[Anum_pg_publication_pubtruncate - 1] = true;
    }
 
    if (publish_via_partition_root_given)
    {
        values[Anum_pg_publication_pubviaroot - 1] = BoolGetDatum(publish_via_partition_root);
        replaces[Anum_pg_publication_pubviaroot - 1] = true;
    }
 
    if (publish_generated_columns_given)
    {
        values[Anum_pg_publication_pubgencols - 1] = BoolGetDatum(publish_generated_columns);
        replaces[Anum_pg_publication_pubgencols - 1] = true;
    }
 
    tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
                            replaces);
 
    /* Update the catalog. */
    CatalogTupleUpdate(rel, &tup->t_self, tup);
 
    CommandCounterIncrement();
 
    pubform = (Form_pg_publication) GETSTRUCT(tup);
 
    /* Invalidate the relcache. */
    if (pubform->puballtables)
    {
        CacheInvalidateRelcacheAll();
    }
    else
    {
        List       *relids = NIL;
        List       *schemarelids = NIL;
 
        /*
         * For any partitioned tables contained in the publication, we must
         * invalidate all partitions contained in the respective partition
         * trees, not just those explicitly mentioned in the publication.
         */
        if (root_relids == NIL)
            relids = GetPublicationRelations(pubform->oid,
                                             PUBLICATION_PART_ALL);
        else
        {
            /*
             * We already got tables explicitly mentioned in the publication.
             * Now get all partitions for the partitioned table in the list.
             */
            foreach(lc, root_relids)
                relids = GetPubPartitionOptionRelations(relids,
                                                        PUBLICATION_PART_ALL,
                                                        lfirst_oid(lc));
        }
 
        schemarelids = GetAllSchemaPublicationRelations(pubform->oid,
                                                        PUBLICATION_PART_ALL);
        relids = list_concat_unique_oid(relids, schemarelids);
 
        InvalidatePublicationRels(relids);
    }
 
    ObjectAddressSet(obj, PublicationRelationId, pubform->oid);
    EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
                                     (Node *) stmt);
 
    InvokeObjectPostAlterHook(PublicationRelationId, pubform->oid, 0);
}
 
/*
 * Invalidate the relations.
 */
void
InvalidatePublicationRels(List *relids)
{
    /*
     * We don't want to send too many individual messages, at some point it's
     * cheaper to just reset whole relcache.
     */
    if (list_length(relids) < MAX_RELCACHE_INVAL_MSGS)
    {
        ListCell   *lc;
 
        foreach(lc, relids)
            CacheInvalidateRelcacheByRelid(lfirst_oid(lc));
    }
    else
        CacheInvalidateRelcacheAll();
}
 
/*
 * Add or remove table to/from publication.
 */
static void
AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
                       List *tables, const char *queryString,
                       bool publish_schema)
{
    List       *rels = NIL;
    Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
    Oid         pubid = pubform->oid;
 
    /*
     * Nothing to do if no objects, except in SET: for that it is quite
     * possible that user has not specified any tables in which case we need
     * to remove all the existing tables.
     */
    if (!tables && stmt->action != AP_SetObjects)
        return;
 
    rels = OpenTableList(tables);
 
    if (stmt->action == AP_AddObjects)
    {
        TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
 
        publish_schema |= is_schema_publication(pubid);
 
        CheckPubRelationColumnList(stmt->pubname, rels, publish_schema,
                                   pubform->pubviaroot);
 
        PublicationAddTables(pubid, rels, false, stmt);
    }
    else if (stmt->action == AP_DropObjects)
        PublicationDropTables(pubid, rels, false);
    else                        /* AP_SetObjects */
    {
        List       *oldrelids = GetPublicationRelations(pubid,
                                                        PUBLICATION_PART_ROOT);
        List       *delrels = NIL;
        ListCell   *oldlc;
 
        TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
 
        CheckPubRelationColumnList(stmt->pubname, rels, publish_schema,
                                   pubform->pubviaroot);
 
        /*
         * To recreate the relation list for the publication, look for
         * existing relations that do not need to be dropped.
         */
        foreach(oldlc, oldrelids)
        {
            Oid         oldrelid = lfirst_oid(oldlc);
            ListCell   *newlc;
            PublicationRelInfo *oldrel;
            bool        found = false;
            HeapTuple   rftuple;
            Node       *oldrelwhereclause = NULL;
            Bitmapset  *oldcolumns = NULL;
 
            /* look up the cache for the old relmap */
            rftuple = SearchSysCache2(PUBLICATIONRELMAP,
                                      ObjectIdGetDatum(oldrelid),
                                      ObjectIdGetDatum(pubid));
 
            /*
             * See if the existing relation currently has a WHERE clause or a
             * column list. We need to compare those too.
             */
            if (HeapTupleIsValid(rftuple))
            {
                bool        isnull = true;
                Datum       whereClauseDatum;
                Datum       columnListDatum;
 
                /* Load the WHERE clause for this table. */
                whereClauseDatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
                                                   Anum_pg_publication_rel_prqual,
                                                   &isnull);
                if (!isnull)
                    oldrelwhereclause = stringToNode(TextDatumGetCString(whereClauseDatum));
 
                /* Transform the int2vector column list to a bitmap. */
                columnListDatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
                                                  Anum_pg_publication_rel_prattrs,
                                                  &isnull);
 
                if (!isnull)
                    oldcolumns = pub_collist_to_bitmapset(NULL, columnListDatum, NULL);
 
                ReleaseSysCache(rftuple);
            }
 
            foreach(newlc, rels)
            {
                PublicationRelInfo *newpubrel;
                Oid         newrelid;
                Bitmapset  *newcolumns = NULL;
 
                newpubrel = (PublicationRelInfo *) lfirst(newlc);
                newrelid = RelationGetRelid(newpubrel->relation);
 
                /*
                 * Validate the column list.  If the column list or WHERE
                 * clause changes, then the validation done here will be
                 * duplicated inside PublicationAddTables().  The validation
                 * is cheap enough that that seems harmless.
                 */
                newcolumns = pub_collist_validate(newpubrel->relation,
                                                  newpubrel->columns);
 
                /*
                 * Check if any of the new set of relations matches with the
                 * existing relations in the publication. Additionally, if the
                 * relation has an associated WHERE clause, check the WHERE
                 * expressions also match. Same for the column list. Drop the
                 * rest.
                 */
                if (newrelid == oldrelid)
                {
                    if (equal(oldrelwhereclause, newpubrel->whereClause) &&
                        bms_equal(oldcolumns, newcolumns))
                    {
                        found = true;
                        break;
                    }
                }
            }
 
            /*
             * Add the non-matched relations to a list so that they can be
             * dropped.
             */
            if (!found)
            {
                oldrel = palloc(sizeof(PublicationRelInfo));
                oldrel->whereClause = NULL;
                oldrel->columns = NIL;
                oldrel->relation = table_open(oldrelid,
                                              ShareUpdateExclusiveLock);
                delrels = lappend(delrels, oldrel);
            }
        }
 
        /* And drop them. */
        PublicationDropTables(pubid, delrels, true);
 
        /*
         * Don't bother calculating the difference for adding, we'll catch and
         * skip existing ones when doing catalog update.
         */
        PublicationAddTables(pubid, rels, true, stmt);
 
        CloseTableList(delrels);
    }
 
    CloseTableList(rels);
}
 
/*
 * Alter the publication schemas.
 *
 * Add or remove schemas to/from publication.
 */
static void
AlterPublicationSchemas(AlterPublicationStmt *stmt,
                        HeapTuple tup, List *schemaidlist)
{
    Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
 
    /*
     * Nothing to do if no objects, except in SET: for that it is quite
     * possible that user has not specified any schemas in which case we need
     * to remove all the existing schemas.
     */
    if (!schemaidlist && stmt->action != AP_SetObjects)
        return;
 
    /*
     * Schema lock is held until the publication is altered to prevent
     * concurrent schema deletion.
     */
    LockSchemaList(schemaidlist);
    if (stmt->action == AP_AddObjects)
    {
        ListCell   *lc;
        List       *reloids;
 
        reloids = GetPublicationRelations(pubform->oid, PUBLICATION_PART_ROOT);
 
        foreach(lc, reloids)
        {
            HeapTuple   coltuple;
 
            coltuple = SearchSysCache2(PUBLICATIONRELMAP,
                                       ObjectIdGetDatum(lfirst_oid(lc)),
                                       ObjectIdGetDatum(pubform->oid));
 
            if (!HeapTupleIsValid(coltuple))
                continue;
 
            /*
             * Disallow adding schema if column list is already part of the
             * publication. See CheckPubRelationColumnList.
             */
            if (!heap_attisnull(coltuple, Anum_pg_publication_rel_prattrs, NULL))
                ereport(ERROR,
                        errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                        errmsg("cannot add schema to publication \"%s\"",
                               stmt->pubname),
                        errdetail("Schemas cannot be added if any tables that specify a column list are already part of the publication."));
 
            ReleaseSysCache(coltuple);
        }
 
        PublicationAddSchemas(pubform->oid, schemaidlist, false, stmt);
    }
    else if (stmt->action == AP_DropObjects)
        PublicationDropSchemas(pubform->oid, schemaidlist, false);
    else                        /* AP_SetObjects */
    {
        List       *oldschemaids = GetPublicationSchemas(pubform->oid);
        List       *delschemas = NIL;
 
        /* Identify which schemas should be dropped */
        delschemas = list_difference_oid(oldschemaids, schemaidlist);
 
        /*
         * Schema lock is held until the publication is altered to prevent
         * concurrent schema deletion.
         */
        LockSchemaList(delschemas);
 
        /* And drop them */
        PublicationDropSchemas(pubform->oid, delschemas, true);
 
        /*
         * Don't bother calculating the difference for adding, we'll catch and
         * skip existing ones when doing catalog update.
         */
        PublicationAddSchemas(pubform->oid, schemaidlist, true, stmt);
    }
}
 
/*
 * Check if relations and schemas can be in a given publication and throw
 * appropriate error if not.
 */
static void
CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
                      List *tables, List *schemaidlist)
{
    Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
 
    if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) &&
        schemaidlist && !superuser())
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("must be superuser to add or set schemas")));
 
    /*
     * Check that user is allowed to manipulate the publication tables in
     * schema
     */
    if (schemaidlist && pubform->puballtables)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("publication \"%s\" is defined as FOR ALL TABLES",
                        NameStr(pubform->pubname)),
                 errdetail("Schemas cannot be added to or dropped from FOR ALL TABLES publications.")));
 
    /* Check that user is allowed to manipulate the publication tables. */
    if (tables && pubform->puballtables)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("publication \"%s\" is defined as FOR ALL TABLES",
                        NameStr(pubform->pubname)),
                 errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications.")));
}
 
/*
 * Alter the existing publication.
 *
 * This is dispatcher function for AlterPublicationOptions,
 * AlterPublicationSchemas and AlterPublicationTables.
 */
void
AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
{
    Relation    rel;
    HeapTuple   tup;
    Form_pg_publication pubform;
 
    rel = table_open(PublicationRelationId, RowExclusiveLock);
 
    tup = SearchSysCacheCopy1(PUBLICATIONNAME,
                              CStringGetDatum(stmt->pubname));
 
    if (!HeapTupleIsValid(tup))
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("publication \"%s\" does not exist",
                        stmt->pubname)));
 
    pubform = (Form_pg_publication) GETSTRUCT(tup);
 
    /* must be owner */
    if (!object_ownercheck(PublicationRelationId, pubform->oid, GetUserId()))
        aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
                       stmt->pubname);
 
    if (stmt->options)
        AlterPublicationOptions(pstate, stmt, rel, tup);
    else
    {
        List       *relations = NIL;
        List       *schemaidlist = NIL;
        Oid         pubid = pubform->oid;
 
        ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
                                   &schemaidlist);
 
        CheckAlterPublication(stmt, tup, relations, schemaidlist);
 
        heap_freetuple(tup);
 
        /* Lock the publication so nobody else can do anything with it. */
        LockDatabaseObject(PublicationRelationId, pubid, 0,
                           AccessExclusiveLock);
 
        /*
         * It is possible that by the time we acquire the lock on publication,
         * concurrent DDL has removed it. We can test this by checking the
         * existence of publication. We get the tuple again to avoid the risk
         * of any publication option getting changed.
         */
        tup = SearchSysCacheCopy1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
        if (!HeapTupleIsValid(tup))
            ereport(ERROR,
                    errcode(ERRCODE_UNDEFINED_OBJECT),
                    errmsg("publication \"%s\" does not exist",
                           stmt->pubname));
 
        AlterPublicationTables(stmt, tup, relations, pstate->p_sourcetext,
                               schemaidlist != NIL);
        AlterPublicationSchemas(stmt, tup, schemaidlist);
    }
 
    /* Cleanup. */
    heap_freetuple(tup);
    table_close(rel, RowExclusiveLock);
}
 
/*
 * Remove relation from publication by mapping OID.
 */
void
RemovePublicationRelById(Oid proid)
{
    Relation    rel;
    HeapTuple   tup;
    Form_pg_publication_rel pubrel;
    List       *relids = NIL;
 
    rel = table_open(PublicationRelRelationId, RowExclusiveLock);
 
    tup = SearchSysCache1(PUBLICATIONREL, ObjectIdGetDatum(proid));
 
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for publication table %u",
             proid);
 
    pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
 
    /*
     * 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,
                                            pubrel->prrelid);
 
    InvalidatePublicationRels(relids);
 
    CatalogTupleDelete(rel, &tup->t_self);
 
    ReleaseSysCache(tup);
 
    table_close(rel, RowExclusiveLock);
}
 
/*
 * Remove the publication by mapping OID.
 */
void
RemovePublicationById(Oid pubid)
{
    Relation    rel;
    HeapTuple   tup;
    Form_pg_publication pubform;
 
    rel = table_open(PublicationRelationId, RowExclusiveLock);
 
    tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for publication %u", pubid);
 
    pubform = (Form_pg_publication) GETSTRUCT(tup);
 
    /* Invalidate relcache so that publication info is rebuilt. */
    if (pubform->puballtables)
        CacheInvalidateRelcacheAll();
 
    CatalogTupleDelete(rel, &tup->t_self);
 
    ReleaseSysCache(tup);
 
    table_close(rel, RowExclusiveLock);
}
 
/*
 * Remove schema from publication by mapping OID.
 */
void
RemovePublicationSchemaById(Oid psoid)
{
    Relation    rel;
    HeapTuple   tup;
    List       *schemaRels = NIL;
    Form_pg_publication_namespace pubsch;
 
    rel = table_open(PublicationNamespaceRelationId, RowExclusiveLock);
 
    tup = SearchSysCache1(PUBLICATIONNAMESPACE, ObjectIdGetDatum(psoid));
 
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for publication schema %u", psoid);
 
    pubsch = (Form_pg_publication_namespace) GETSTRUCT(tup);
 
    /*
     * Invalidate relcache so that publication info is rebuilt. See
     * RemovePublicationRelById for why we need to consider all the
     * partitions.
     */
    schemaRels = GetSchemaPublicationRelations(pubsch->pnnspid,
                                               PUBLICATION_PART_ALL);
    InvalidatePublicationRels(schemaRels);
 
    CatalogTupleDelete(rel, &tup->t_self);
 
    ReleaseSysCache(tup);
 
    table_close(rel, RowExclusiveLock);
}
 
/*
 * Open relations specified by a PublicationTable list.
 * The returned tables are locked in ShareUpdateExclusiveLock mode in order to
 * add them to a publication.
 */
static List *
OpenTableList(List *tables)
{
    List       *relids = NIL;
    List       *rels = NIL;
    ListCell   *lc;
    List       *relids_with_rf = NIL;
    List       *relids_with_collist = NIL;
 
    /*
     * Open, share-lock, and check all the explicitly-specified relations
     */
    foreach(lc, tables)
    {
        PublicationTable *t = lfirst_node(PublicationTable, lc);
        bool        recurse = t->relation->inh;
        Relation    rel;
        Oid         myrelid;
        PublicationRelInfo *pub_rel;
 
        /* Allow query cancel in case this takes a long time */
        CHECK_FOR_INTERRUPTS();
 
        rel = table_openrv(t->relation, ShareUpdateExclusiveLock);
        myrelid = RelationGetRelid(rel);
 
        /*
         * Filter out duplicates if user specifies "foo, foo".
         *
         * Note that this algorithm is known to not be very efficient (O(N^2))
         * but given that it only works on list of tables given to us by user
         * it's deemed acceptable.
         */
        if (list_member_oid(relids, myrelid))
        {
            /* Disallow duplicate tables if there are any with row filters. */
            if (t->whereClause || list_member_oid(relids_with_rf, myrelid))
                ereport(ERROR,
                        (errcode(ERRCODE_DUPLICATE_OBJECT),
                         errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
                                RelationGetRelationName(rel))));
 
            /* Disallow duplicate tables if there are any with column lists. */
            if (t->columns || list_member_oid(relids_with_collist, myrelid))
                ereport(ERROR,
                        (errcode(ERRCODE_DUPLICATE_OBJECT),
                         errmsg("conflicting or redundant column lists for table \"%s\"",
                                RelationGetRelationName(rel))));
 
            table_close(rel, ShareUpdateExclusiveLock);
            continue;
        }
 
        pub_rel = palloc(sizeof(PublicationRelInfo));
        pub_rel->relation = rel;
        pub_rel->whereClause = t->whereClause;
        pub_rel->columns = t->columns;
        rels = lappend(rels, pub_rel);
        relids = lappend_oid(relids, myrelid);
 
        if (t->whereClause)
            relids_with_rf = lappend_oid(relids_with_rf, myrelid);
 
        if (t->columns)
            relids_with_collist = lappend_oid(relids_with_collist, myrelid);
 
        /*
         * Add children of this rel, if requested, so that they too are added
         * to the publication.  A partitioned table can't have any inheritance
         * children other than its partitions, which need not be explicitly
         * added to the publication.
         */
        if (recurse && rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
        {
            List       *children;
            ListCell   *child;
 
            children = find_all_inheritors(myrelid, ShareUpdateExclusiveLock,
                                           NULL);
 
            foreach(child, children)
            {
                Oid         childrelid = lfirst_oid(child);
 
                /* Allow query cancel in case this takes a long time */
                CHECK_FOR_INTERRUPTS();
 
                /*
                 * Skip duplicates if user specified both parent and child
                 * tables.
                 */
                if (list_member_oid(relids, childrelid))
                {
                    /*
                     * We don't allow to specify row filter for both parent
                     * and child table at the same time as it is not very
                     * clear which one should be given preference.
                     */
                    if (childrelid != myrelid &&
                        (t->whereClause || list_member_oid(relids_with_rf, childrelid)))
                        ereport(ERROR,
                                (errcode(ERRCODE_DUPLICATE_OBJECT),
                                 errmsg("conflicting or redundant WHERE clauses for table \"%s\"",
                                        RelationGetRelationName(rel))));
 
                    /*
                     * We don't allow to specify column list for both parent
                     * and child table at the same time as it is not very
                     * clear which one should be given preference.
                     */
                    if (childrelid != myrelid &&
                        (t->columns || list_member_oid(relids_with_collist, childrelid)))
                        ereport(ERROR,
                                (errcode(ERRCODE_DUPLICATE_OBJECT),
                                 errmsg("conflicting or redundant column lists for table \"%s\"",
                                        RelationGetRelationName(rel))));
 
                    continue;
                }
 
                /* find_all_inheritors already got lock */
                rel = table_open(childrelid, NoLock);
                pub_rel = palloc(sizeof(PublicationRelInfo));
                pub_rel->relation = rel;
                /* child inherits WHERE clause from parent */
                pub_rel->whereClause = t->whereClause;
 
                /* child inherits column list from parent */
                pub_rel->columns = t->columns;
                rels = lappend(rels, pub_rel);
                relids = lappend_oid(relids, childrelid);
 
                if (t->whereClause)
                    relids_with_rf = lappend_oid(relids_with_rf, childrelid);
 
                if (t->columns)
                    relids_with_collist = lappend_oid(relids_with_collist, childrelid);
            }
        }
    }
 
    list_free(relids);
    list_free(relids_with_rf);
 
    return rels;
}
 
/*
 * Close all relations in the list.
 */
static void
CloseTableList(List *rels)
{
    ListCell   *lc;
 
    foreach(lc, rels)
    {
        PublicationRelInfo *pub_rel;
 
        pub_rel = (PublicationRelInfo *) lfirst(lc);
        table_close(pub_rel->relation, NoLock);
    }
 
    list_free_deep(rels);
}
 
/*
 * Lock the schemas specified in the schema list in AccessShareLock mode in
 * order to prevent concurrent schema deletion.
 */
static void
LockSchemaList(List *schemalist)
{
    ListCell   *lc;
 
    foreach(lc, schemalist)
    {
        Oid         schemaid = lfirst_oid(lc);
 
        /* Allow query cancel in case this takes a long time */
        CHECK_FOR_INTERRUPTS();
        LockDatabaseObject(NamespaceRelationId, schemaid, 0, AccessShareLock);
 
        /*
         * It is possible that by the time we acquire the lock on schema,
         * concurrent DDL has removed it. We can test this by checking the
         * existence of schema.
         */
        if (!SearchSysCacheExists1(NAMESPACEOID, ObjectIdGetDatum(schemaid)))
            ereport(ERROR,
                    errcode(ERRCODE_UNDEFINED_SCHEMA),
                    errmsg("schema with OID %u does not exist", schemaid));
    }
}
 
/*
 * Add listed tables to the publication.
 */
static void
PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
                     AlterPublicationStmt *stmt)
{
    ListCell   *lc;
 
    Assert(!stmt || !stmt->for_all_tables);
 
    foreach(lc, rels)
    {
        PublicationRelInfo *pub_rel = (PublicationRelInfo *) lfirst(lc);
        Relation    rel = pub_rel->relation;
        ObjectAddress obj;
 
        /* Must be owner of the table or superuser. */
        if (!object_ownercheck(RelationRelationId, RelationGetRelid(rel), GetUserId()))
            aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
                           RelationGetRelationName(rel));
 
        obj = publication_add_relation(pubid, pub_rel, if_not_exists);
        if (stmt)
        {
            EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
                                             (Node *) stmt);
 
            InvokeObjectPostCreateHook(PublicationRelRelationId,
                                       obj.objectId, 0);
        }
    }
}
 
/*
 * Remove listed tables from the publication.
 */
static void
PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
{
    ObjectAddress obj;
    ListCell   *lc;
    Oid         prid;
 
    foreach(lc, rels)
    {
        PublicationRelInfo *pubrel = (PublicationRelInfo *) lfirst(lc);
        Relation    rel = pubrel->relation;
        Oid         relid = RelationGetRelid(rel);
 
        if (pubrel->columns)
            ereport(ERROR,
                    errcode(ERRCODE_SYNTAX_ERROR),
                    errmsg("column list must not be specified in ALTER PUBLICATION ... DROP"));
 
        prid = GetSysCacheOid2(PUBLICATIONRELMAP, Anum_pg_publication_rel_oid,
                               ObjectIdGetDatum(relid),
                               ObjectIdGetDatum(pubid));
        if (!OidIsValid(prid))
        {
            if (missing_ok)
                continue;
 
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_OBJECT),
                     errmsg("relation \"%s\" is not part of the publication",
                            RelationGetRelationName(rel))));
        }
 
        if (pubrel->whereClause)
            ereport(ERROR,
                    (errcode(ERRCODE_SYNTAX_ERROR),
                     errmsg("cannot use a WHERE clause when removing a table from a publication")));
 
        ObjectAddressSet(obj, PublicationRelRelationId, prid);
        performDeletion(&obj, DROP_CASCADE, 0);
    }
}
 
/*
 * Add listed schemas to the publication.
 */
static void
PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
                      AlterPublicationStmt *stmt)
{
    ListCell   *lc;
 
    Assert(!stmt || !stmt->for_all_tables);
 
    foreach(lc, schemas)
    {
        Oid         schemaid = lfirst_oid(lc);
        ObjectAddress obj;
 
        obj = publication_add_schema(pubid, schemaid, if_not_exists);
        if (stmt)
        {
            EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
                                             (Node *) stmt);
 
            InvokeObjectPostCreateHook(PublicationNamespaceRelationId,
                                       obj.objectId, 0);
        }
    }
}
 
/*
 * Remove listed schemas from the publication.
 */
static void
PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok)
{
    ObjectAddress obj;
    ListCell   *lc;
    Oid         psid;
 
    foreach(lc, schemas)
    {
        Oid         schemaid = lfirst_oid(lc);
 
        psid = GetSysCacheOid2(PUBLICATIONNAMESPACEMAP,
                               Anum_pg_publication_namespace_oid,
                               ObjectIdGetDatum(schemaid),
                               ObjectIdGetDatum(pubid));
        if (!OidIsValid(psid))
        {
            if (missing_ok)
                continue;
 
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_OBJECT),
                     errmsg("tables from schema \"%s\" are not part of the publication",
                            get_namespace_name(schemaid))));
        }
 
        ObjectAddressSet(obj, PublicationNamespaceRelationId, psid);
        performDeletion(&obj, DROP_CASCADE, 0);
    }
}
 
/*
 * Internal workhorse for changing a publication owner
 */
static void
AlterPublicationOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
{
    Form_pg_publication form;
 
    form = (Form_pg_publication) GETSTRUCT(tup);
 
    if (form->pubowner == newOwnerId)
        return;
 
    if (!superuser())
    {
        AclResult   aclresult;
 
        /* Must be owner */
        if (!object_ownercheck(PublicationRelationId, form->oid, GetUserId()))
            aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_PUBLICATION,
                           NameStr(form->pubname));
 
        /* Must be able to become new owner */
        check_can_set_role(GetUserId(), newOwnerId);
 
        /* New owner must have CREATE privilege on database */
        aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, newOwnerId, ACL_CREATE);
        if (aclresult != ACLCHECK_OK)
            aclcheck_error(aclresult, OBJECT_DATABASE,
                           get_database_name(MyDatabaseId));
 
        if (form->puballtables && !superuser_arg(newOwnerId))
            ereport(ERROR,
                    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                     errmsg("permission denied to change owner of publication \"%s\"",
                            NameStr(form->pubname)),
                     errhint("The owner of a FOR ALL TABLES publication must be a superuser.")));
 
        if (!superuser_arg(newOwnerId) && is_schema_publication(form->oid))
            ereport(ERROR,
                    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                     errmsg("permission denied to change owner of publication \"%s\"",
                            NameStr(form->pubname)),
                     errhint("The owner of a FOR TABLES IN SCHEMA publication must be a superuser.")));
    }
 
    form->pubowner = newOwnerId;
    CatalogTupleUpdate(rel, &tup->t_self, tup);
 
    /* Update owner dependency reference */
    changeDependencyOnOwner(PublicationRelationId,
                            form->oid,
                            newOwnerId);
 
    InvokeObjectPostAlterHook(PublicationRelationId,
                              form->oid, 0);
}
 
/*
 * Change publication owner -- by name
 */
ObjectAddress
AlterPublicationOwner(const char *name, Oid newOwnerId)
{
    Oid         subid;
    HeapTuple   tup;
    Relation    rel;
    ObjectAddress address;
    Form_pg_publication pubform;
 
    rel = table_open(PublicationRelationId, RowExclusiveLock);
 
    tup = SearchSysCacheCopy1(PUBLICATIONNAME, CStringGetDatum(name));
 
    if (!HeapTupleIsValid(tup))
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("publication \"%s\" does not exist", name)));
 
    pubform = (Form_pg_publication) GETSTRUCT(tup);
    subid = pubform->oid;
 
    AlterPublicationOwner_internal(rel, tup, newOwnerId);
 
    ObjectAddressSet(address, PublicationRelationId, subid);
 
    heap_freetuple(tup);
 
    table_close(rel, RowExclusiveLock);
 
    return address;
}
 
/*
 * Change publication owner -- by OID
 */
void
AlterPublicationOwner_oid(Oid subid, Oid newOwnerId)
{
    HeapTuple   tup;
    Relation    rel;
 
    rel = table_open(PublicationRelationId, RowExclusiveLock);
 
    tup = SearchSysCacheCopy1(PUBLICATIONOID, ObjectIdGetDatum(subid));
 
    if (!HeapTupleIsValid(tup))
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("publication with OID %u does not exist", subid)));
 
    AlterPublicationOwner_internal(rel, tup, newOwnerId);
 
    heap_freetuple(tup);
 
    table_close(rel, RowExclusiveLock);
}