tablecmds.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * tablecmds.c
 *    Commands for creating and altering table structures and settings
 *
 * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *    src/backend/commands/tablecmds.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/heapam_xlog.h"
#include "access/multixact.h"
#include "access/reloptions.h"
#include "access/relscan.h"
#include "access/sysattr.h"
#include "access/tupconvert.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/heap.h"
#include "catalog/index.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/partition.h"
#include "catalog/pg_am.h"
#include "catalog/pg_collation.h"
#include "catalog/pg_constraint.h"
#include "catalog/pg_depend.h"
#include "catalog/pg_foreign_table.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_opclass.h"
#include "catalog/pg_tablespace.h"
#include "catalog/pg_trigger.h"
#include "catalog/pg_type.h"
#include "catalog/storage.h"
#include "catalog/storage_xlog.h"
#include "catalog/toasting.h"
#include "commands/cluster.h"
#include "commands/comment.h"
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/policy.h"
#include "commands/sequence.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "commands/trigger.h"
#include "commands/typecmds.h"
#include "commands/user.h"
#include "executor/executor.h"
#include "foreign/foreign.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "nodes/parsenodes.h"
#include "optimizer/clauses.h"
#include "optimizer/planner.h"
#include "optimizer/predtest.h"
#include "optimizer/prep.h"
#include "optimizer/var.h"
#include "parser/parse_clause.h"
#include "parser/parse_coerce.h"
#include "parser/parse_collate.h"
#include "parser/parse_expr.h"
#include "parser/parse_oper.h"
#include "parser/parse_relation.h"
#include "parser/parse_type.h"
#include "parser/parse_utilcmd.h"
#include "parser/parser.h"
#include "partitioning/partbounds.h"
#include "pgstat.h"
#include "rewrite/rewriteDefine.h"
#include "rewrite/rewriteHandler.h"
#include "rewrite/rewriteManip.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "storage/lock.h"
#include "storage/predicate.h"
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/partcache.h"
#include "utils/relcache.h"
#include "utils/ruleutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/tqual.h"
#include "utils/typcache.h"


/*
 * ON COMMIT action list
 */
typedef struct OnCommitItem
{
    Oid         relid;          /* relid of relation */
    OnCommitAction oncommit;    /* what to do at end of xact */

    /*
     * If this entry was created during the current transaction,
     * creating_subid is the ID of the creating subxact; if created in a prior
     * transaction, creating_subid is zero.  If deleted during the current
     * transaction, deleting_subid is the ID of the deleting subxact; if no
     * deletion request is pending, deleting_subid is zero.
     */
    SubTransactionId creating_subid;
    SubTransactionId deleting_subid;
} OnCommitItem;

static List *on_commits = NIL;


/*
 * State information for ALTER TABLE
 *
 * The pending-work queue for an ALTER TABLE is a List of AlteredTableInfo
 * structs, one for each table modified by the operation (the named table
 * plus any child tables that are affected).  We save lists of subcommands
 * to apply to this table (possibly modified by parse transformation steps);
 * these lists will be executed in Phase 2.  If a Phase 3 step is needed,
 * necessary information is stored in the constraints and newvals lists.
 *
 * Phase 2 is divided into multiple passes; subcommands are executed in
 * a pass determined by subcommand type.
 */

#define AT_PASS_UNSET           -1  /* UNSET will cause ERROR */
#define AT_PASS_DROP            0   /* DROP (all flavors) */
#define AT_PASS_ALTER_TYPE      1   /* ALTER COLUMN TYPE */
#define AT_PASS_OLD_INDEX       2   /* re-add existing indexes */
#define AT_PASS_OLD_CONSTR      3   /* re-add existing constraints */
#define AT_PASS_COL_ATTRS       4   /* set other column attributes */
/* We could support a RENAME COLUMN pass here, but not currently used */
#define AT_PASS_ADD_COL         5   /* ADD COLUMN */
#define AT_PASS_ADD_INDEX       6   /* ADD indexes */
#define AT_PASS_ADD_CONSTR      7   /* ADD constraints, defaults */
#define AT_PASS_MISC            8   /* other stuff */
#define AT_NUM_PASSES           9

typedef struct AlteredTableInfo
{
    /* Information saved before any work commences: */
    Oid         relid;          /* Relation to work on */
    char        relkind;        /* Its relkind */
    TupleDesc   oldDesc;        /* Pre-modification tuple descriptor */
    /* Information saved by Phase 1 for Phase 2: */
    List       *subcmds[AT_NUM_PASSES]; /* Lists of AlterTableCmd */
    /* Information saved by Phases 1/2 for Phase 3: */
    List       *constraints;    /* List of NewConstraint */
    List       *newvals;        /* List of NewColumnValue */
    bool        new_notnull;    /* T if we added new NOT NULL constraints */
    int         rewrite;        /* Reason for forced rewrite, if any */
    Oid         newTableSpace;  /* new tablespace; 0 means no change */
    bool        chgPersistence; /* T if SET LOGGED/UNLOGGED is used */
    char        newrelpersistence;  /* if above is true */
    Expr       *partition_constraint;   /* for attach partition validation */
    /* true, if validating default due to some other attach/detach */
    bool        validate_default;
    /* Objects to rebuild after completing ALTER TYPE operations */
    List       *changedConstraintOids;  /* OIDs of constraints to rebuild */
    List       *changedConstraintDefs;  /* string definitions of same */
    List       *changedIndexOids;   /* OIDs of indexes to rebuild */
    List       *changedIndexDefs;   /* string definitions of same */
} AlteredTableInfo;

/* Struct describing one new constraint to check in Phase 3 scan */
/* Note: new NOT NULL constraints are handled elsewhere */
typedef struct NewConstraint
{
    char       *name;           /* Constraint name, or NULL if none */
    ConstrType  contype;        /* CHECK or FOREIGN */
    Oid         refrelid;       /* PK rel, if FOREIGN */
    Oid         refindid;       /* OID of PK's index, if FOREIGN */
    Oid         conid;          /* OID of pg_constraint entry, if FOREIGN */
    Node       *qual;           /* Check expr or CONSTR_FOREIGN Constraint */
    ExprState  *qualstate;      /* Execution state for CHECK expr */
} NewConstraint;

/*
 * Struct describing one new column value that needs to be computed during
 * Phase 3 copy (this could be either a new column with a non-null default, or
 * a column that we're changing the type of).  Columns without such an entry
 * are just copied from the old table during ATRewriteTable.  Note that the
 * expr is an expression over *old* table values.
 */
typedef struct NewColumnValue
{
    AttrNumber  attnum;         /* which column */
    Expr       *expr;           /* expression to compute */
    ExprState  *exprstate;      /* execution state */
} NewColumnValue;

/*
 * Error-reporting support for RemoveRelations
 */
struct dropmsgstrings
{
    char        kind;
    int         nonexistent_code;
    const char *nonexistent_msg;
    const char *skipping_msg;
    const char *nota_msg;
    const char *drophint_msg;
};

static const struct dropmsgstrings dropmsgstringarray[] = {
    {RELKIND_RELATION,
        ERRCODE_UNDEFINED_TABLE,
        gettext_noop("table \"%s\" does not exist"),
        gettext_noop("table \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not a table"),
    gettext_noop("Use DROP TABLE to remove a table.")},
    {RELKIND_SEQUENCE,
        ERRCODE_UNDEFINED_TABLE,
        gettext_noop("sequence \"%s\" does not exist"),
        gettext_noop("sequence \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not a sequence"),
    gettext_noop("Use DROP SEQUENCE to remove a sequence.")},
    {RELKIND_VIEW,
        ERRCODE_UNDEFINED_TABLE,
        gettext_noop("view \"%s\" does not exist"),
        gettext_noop("view \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not a view"),
    gettext_noop("Use DROP VIEW to remove a view.")},
    {RELKIND_MATVIEW,
        ERRCODE_UNDEFINED_TABLE,
        gettext_noop("materialized view \"%s\" does not exist"),
        gettext_noop("materialized view \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not a materialized view"),
    gettext_noop("Use DROP MATERIALIZED VIEW to remove a materialized view.")},
    {RELKIND_INDEX,
        ERRCODE_UNDEFINED_OBJECT,
        gettext_noop("index \"%s\" does not exist"),
        gettext_noop("index \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not an index"),
    gettext_noop("Use DROP INDEX to remove an index.")},
    {RELKIND_COMPOSITE_TYPE,
        ERRCODE_UNDEFINED_OBJECT,
        gettext_noop("type \"%s\" does not exist"),
        gettext_noop("type \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not a type"),
    gettext_noop("Use DROP TYPE to remove a type.")},
    {RELKIND_FOREIGN_TABLE,
        ERRCODE_UNDEFINED_OBJECT,
        gettext_noop("foreign table \"%s\" does not exist"),
        gettext_noop("foreign table \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not a foreign table"),
    gettext_noop("Use DROP FOREIGN TABLE to remove a foreign table.")},
    {RELKIND_PARTITIONED_TABLE,
        ERRCODE_UNDEFINED_TABLE,
        gettext_noop("table \"%s\" does not exist"),
        gettext_noop("table \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not a table"),
    gettext_noop("Use DROP TABLE to remove a table.")},
    {RELKIND_PARTITIONED_INDEX,
        ERRCODE_UNDEFINED_OBJECT,
        gettext_noop("index \"%s\" does not exist"),
        gettext_noop("index \"%s\" does not exist, skipping"),
        gettext_noop("\"%s\" is not an index"),
    gettext_noop("Use DROP INDEX to remove an index.")},
    {'\0', 0, NULL, NULL, NULL, NULL}
};

struct DropRelationCallbackState
{
    char        relkind;
    Oid         heapOid;
    Oid         partParentOid;
    bool        concurrent;
};

/* Alter table target-type flags for ATSimplePermissions */
#define     ATT_TABLE               0x0001
#define     ATT_VIEW                0x0002
#define     ATT_MATVIEW             0x0004
#define     ATT_INDEX               0x0008
#define     ATT_COMPOSITE_TYPE      0x0010
#define     ATT_FOREIGN_TABLE       0x0020
#define     ATT_PARTITIONED_INDEX   0x0040

/*
 * Partition tables are expected to be dropped when the parent partitioned
 * table gets dropped. Hence for partitioning we use AUTO dependency.
 * Otherwise, for regular inheritance use NORMAL dependency.
 */
#define child_dependency_type(child_is_partition)   \
    ((child_is_partition) ? DEPENDENCY_AUTO : DEPENDENCY_NORMAL)

static void truncate_check_rel(Relation rel);
static List *MergeAttributes(List *schema, List *supers, char relpersistence,
                bool is_partition, List **supOids, List **supconstr,
                int *supOidCount);
static bool MergeCheckConstraint(List *constraints, char *name, Node *expr);
static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel);
static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel);
static void StoreCatalogInheritance(Oid relationId, List *supers,
                        bool child_is_partition);
static void StoreCatalogInheritance1(Oid relationId, Oid parentOid,
                         int32 seqNumber, Relation inhRelation,
                         bool child_is_partition);
static int  findAttrByName(const char *attributeName, List *schema);
static void AlterIndexNamespaces(Relation classRel, Relation rel,
                     Oid oldNspOid, Oid newNspOid, ObjectAddresses *objsMoved);
static void AlterSeqNamespaces(Relation classRel, Relation rel,
                   Oid oldNspOid, Oid newNspOid, ObjectAddresses *objsMoved,
                   LOCKMODE lockmode);
static ObjectAddress ATExecAlterConstraint(Relation rel, AlterTableCmd *cmd,
                      bool recurse, bool recursing, LOCKMODE lockmode);
static ObjectAddress ATExecValidateConstraint(Relation rel, char *constrName,
                         bool recurse, bool recursing, LOCKMODE lockmode);
static int transformColumnNameList(Oid relId, List *colList,
                        int16 *attnums, Oid *atttypids);
static int transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
                           List **attnamelist,
                           int16 *attnums, Oid *atttypids,
                           Oid *opclasses);
static Oid transformFkeyCheckAttrs(Relation pkrel,
                        int numattrs, int16 *attnums,
                        Oid *opclasses);
static void checkFkeyPermissions(Relation rel, int16 *attnums, int natts);
static CoercionPathType findFkeyCast(Oid targetTypeId, Oid sourceTypeId,
             Oid *funcid);
static void validateCheckConstraint(Relation rel, HeapTuple constrtup);
static void validateForeignKeyConstraint(char *conname,
                             Relation rel, Relation pkrel,
                             Oid pkindOid, Oid constraintOid);
static void ATController(AlterTableStmt *parsetree,
             Relation rel, List *cmds, bool recurse, LOCKMODE lockmode);
static void ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
          bool recurse, bool recursing, LOCKMODE lockmode);
static void ATRewriteCatalogs(List **wqueue, LOCKMODE lockmode);
static void ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
          AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATRewriteTables(AlterTableStmt *parsetree,
                List **wqueue, LOCKMODE lockmode);
static void ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode);
static AlteredTableInfo *ATGetQueueEntry(List **wqueue, Relation rel);
static void ATSimplePermissions(Relation rel, int allowed_targets);
static void ATWrongRelkindError(Relation rel, int allowed_targets);
static void ATSimpleRecursion(List **wqueue, Relation rel,
                  AlterTableCmd *cmd, bool recurse, LOCKMODE lockmode);
static void ATTypedTableRecursion(List **wqueue, Relation rel, AlterTableCmd *cmd,
                      LOCKMODE lockmode);
static List *find_typed_table_dependencies(Oid typeOid, const char *typeName,
                              DropBehavior behavior);
static void ATPrepAddColumn(List **wqueue, Relation rel, bool recurse, bool recursing,
                bool is_view, AlterTableCmd *cmd, LOCKMODE lockmode);
static ObjectAddress ATExecAddColumn(List **wqueue, AlteredTableInfo *tab,
                Relation rel, ColumnDef *colDef, bool isOid,
                bool recurse, bool recursing,
                bool if_not_exists, LOCKMODE lockmode);
static bool check_for_column_name_collision(Relation rel, const char *colname,
                                bool if_not_exists);
static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid);
static void add_column_collation_dependency(Oid relid, int32 attnum, Oid collid);
static void ATPrepAddOids(List **wqueue, Relation rel, bool recurse,
              AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATPrepDropNotNull(Relation rel, bool recurse, bool recursing);
static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode);
static void ATPrepSetNotNull(Relation rel, bool recurse, bool recursing);
static ObjectAddress ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
                 const char *colName, LOCKMODE lockmode);
static ObjectAddress ATExecColumnDefault(Relation rel, const char *colName,
                    Node *newDefault, LOCKMODE lockmode);
static ObjectAddress ATExecAddIdentity(Relation rel, const char *colName,
                  Node *def, LOCKMODE lockmode);
static ObjectAddress ATExecSetIdentity(Relation rel, const char *colName,
                  Node *def, LOCKMODE lockmode);
static ObjectAddress ATExecDropIdentity(Relation rel, const char *colName, bool missing_ok, LOCKMODE lockmode);
static void ATPrepSetStatistics(Relation rel, const char *colName, int16 colNum,
                    Node *newValue, LOCKMODE lockmode);
static ObjectAddress ATExecSetStatistics(Relation rel, const char *colName, int16 colNum,
                    Node *newValue, LOCKMODE lockmode);
static ObjectAddress ATExecSetOptions(Relation rel, const char *colName,
                 Node *options, bool isReset, LOCKMODE lockmode);
static ObjectAddress ATExecSetStorage(Relation rel, const char *colName,
                 Node *newValue, LOCKMODE lockmode);
static void ATPrepDropColumn(List **wqueue, Relation rel, bool recurse, bool recursing,
                 AlterTableCmd *cmd, LOCKMODE lockmode);
static ObjectAddress ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
                 DropBehavior behavior,
                 bool recurse, bool recursing,
                 bool missing_ok, LOCKMODE lockmode);
static ObjectAddress ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
               IndexStmt *stmt, bool is_rebuild, LOCKMODE lockmode);
static ObjectAddress ATExecAddConstraint(List **wqueue,
                    AlteredTableInfo *tab, Relation rel,
                    Constraint *newConstraint, bool recurse, bool is_readd,
                    LOCKMODE lockmode);
static ObjectAddress ATExecAddIndexConstraint(AlteredTableInfo *tab, Relation rel,
                         IndexStmt *stmt, LOCKMODE lockmode);
static ObjectAddress ATAddCheckConstraint(List **wqueue,
                     AlteredTableInfo *tab, Relation rel,
                     Constraint *constr,
                     bool recurse, bool recursing, bool is_readd,
                     LOCKMODE lockmode);
static ObjectAddress ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab,
                          Relation rel, Constraint *fkconstraint, Oid parentConstr,
                          bool recurse, bool recursing,
                          LOCKMODE lockmode);
static void ATExecDropConstraint(Relation rel, const char *constrName,
                     DropBehavior behavior,
                     bool recurse, bool recursing,
                     bool missing_ok, LOCKMODE lockmode);
static void ATPrepAlterColumnType(List **wqueue,
                      AlteredTableInfo *tab, Relation rel,
                      bool recurse, bool recursing,
                      AlterTableCmd *cmd, LOCKMODE lockmode);
static bool ATColumnChangeRequiresRewrite(Node *expr, AttrNumber varattno);
static ObjectAddress ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
                      AlterTableCmd *cmd, LOCKMODE lockmode);
static ObjectAddress ATExecAlterColumnGenericOptions(Relation rel, const char *colName,
                                List *options, LOCKMODE lockmode);
static void ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab,
                       LOCKMODE lockmode);
static void ATPostAlterTypeParse(Oid oldId, Oid oldRelId, Oid refRelId,
                     char *cmd, List **wqueue, LOCKMODE lockmode,
                     bool rewrite);
static void RebuildConstraintComment(AlteredTableInfo *tab, int pass,
                         Oid objid, Relation rel, List *domname,
                         const char *conname);
static void TryReuseIndex(Oid oldId, IndexStmt *stmt);
static void TryReuseForeignKey(Oid oldId, Constraint *con);
static void change_owner_fix_column_acls(Oid relationOid,
                             Oid oldOwnerId, Oid newOwnerId);
static void change_owner_recurse_to_sequences(Oid relationOid,
                                  Oid newOwnerId, LOCKMODE lockmode);
static ObjectAddress ATExecClusterOn(Relation rel, const char *indexName,
                LOCKMODE lockmode);
static void ATExecDropCluster(Relation rel, LOCKMODE lockmode);
static bool ATPrepChangePersistence(Relation rel, bool toLogged);
static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
                    const char *tablespacename, LOCKMODE lockmode);
static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
static void ATExecSetRelOptions(Relation rel, List *defList,
                    AlterTableType operation,
                    LOCKMODE lockmode);
static void ATExecEnableDisableTrigger(Relation rel, const char *trigname,
                           char fires_when, bool skip_system, LOCKMODE lockmode);
static void ATExecEnableDisableRule(Relation rel, const char *rulename,
                        char fires_when, LOCKMODE lockmode);
static void ATPrepAddInherit(Relation child_rel);
static ObjectAddress ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode);
static ObjectAddress ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode);
static void drop_parent_dependency(Oid relid, Oid refclassid, Oid refobjid,
                       DependencyType deptype);
static ObjectAddress ATExecAddOf(Relation rel, const TypeName *ofTypename, LOCKMODE lockmode);
static void ATExecDropOf(Relation rel, LOCKMODE lockmode);
static void ATExecReplicaIdentity(Relation rel, ReplicaIdentityStmt *stmt, LOCKMODE lockmode);
static void ATExecGenericOptions(Relation rel, List *options);
static void ATExecEnableRowSecurity(Relation rel);
static void ATExecDisableRowSecurity(Relation rel);
static void ATExecForceNoForceRowSecurity(Relation rel, bool force_rls);

static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
                   ForkNumber forkNum, char relpersistence);
static const char *storage_name(char c);

static void RangeVarCallbackForDropRelation(const RangeVar *rel, Oid relOid,
                                Oid oldRelOid, void *arg);
static void RangeVarCallbackForAlterRelation(const RangeVar *rv, Oid relid,
                                 Oid oldrelid, void *arg);
static PartitionSpec *transformPartitionSpec(Relation rel, PartitionSpec *partspec, char *strategy);
static void ComputePartitionAttrs(Relation rel, List *partParams, AttrNumber *partattrs,
                      List **partexprs, Oid *partopclass, Oid *partcollation, char strategy);
static void CreateInheritance(Relation child_rel, Relation parent_rel);
static void RemoveInheritance(Relation child_rel, Relation parent_rel);
static ObjectAddress ATExecAttachPartition(List **wqueue, Relation rel,
                      PartitionCmd *cmd);
static void AttachPartitionEnsureIndexes(Relation rel, Relation attachrel);
static void QueuePartitionConstraintValidation(List **wqueue, Relation scanrel,
                                   List *partConstraint,
                                   bool validate_default);
static void CloneRowTriggersToPartition(Relation parent, Relation partition);
static ObjectAddress ATExecDetachPartition(Relation rel, RangeVar *name);
static ObjectAddress ATExecAttachPartitionIdx(List **wqueue, Relation rel,
                         RangeVar *name);
static void validatePartitionedIndex(Relation partedIdx, Relation partedTbl);
static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx,
                      Relation partitionTbl);
static void update_relispartition(Relation classRel, Oid relationId,
                      bool newval);


/* ----------------------------------------------------------------
 *      DefineRelation
 *              Creates a new relation.
 *
 * stmt carries parsetree information from an ordinary CREATE TABLE statement.
 * The other arguments are used to extend the behavior for other cases:
 * relkind: relkind to assign to the new relation
 * ownerId: if not InvalidOid, use this as the new relation's owner.
 * typaddress: if not null, it's set to the pg_type entry's address.
 * queryString: for error reporting
 *
 * Note that permissions checks are done against current user regardless of
 * ownerId.  A nonzero ownerId is used when someone is creating a relation
 * "on behalf of" someone else, so we still want to see that the current user
 * has permissions to do it.
 *
 * If successful, returns the address of the new relation.
 * ----------------------------------------------------------------
 */
ObjectAddress
DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
               ObjectAddress *typaddress, const char *queryString)
{
    char        relname[NAMEDATALEN];
    Oid         namespaceId;
    Oid         relationId;
    Oid         tablespaceId;
    Relation    rel;
    TupleDesc   descriptor;
    List       *inheritOids;
    List       *old_constraints;
    bool        localHasOids;
    int         parentOidCount;
    List       *rawDefaults;
    List       *cookedDefaults;
    Datum       reloptions;
    ListCell   *listptr;
    AttrNumber  attnum;
    static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
    Oid         ofTypeId;
    ObjectAddress address;

    /*
     * Truncate relname to appropriate length (probably a waste of time, as
     * parser should have done this already).
     */
    StrNCpy(relname, stmt->relation->relname, NAMEDATALEN);

    /*
     * Check consistency of arguments
     */
    if (stmt->oncommit != ONCOMMIT_NOOP
        && stmt->relation->relpersistence != RELPERSISTENCE_TEMP)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                 errmsg("ON COMMIT can only be used on temporary tables")));

    if (stmt->partspec != NULL)
    {
        if (relkind != RELKIND_RELATION)
            elog(ERROR, "unexpected relkind: %d", (int) relkind);

        relkind = RELKIND_PARTITIONED_TABLE;
    }

    /*
     * Look up the namespace in which we are supposed to create the relation,
     * check we have permission to create there, lock it against concurrent
     * drop, and mark stmt->relation as RELPERSISTENCE_TEMP if a temporary
     * namespace is selected.
     */
    namespaceId =
        RangeVarGetAndCheckCreationNamespace(stmt->relation, NoLock, NULL);

    /*
     * Security check: disallow creating temp tables from security-restricted
     * code.  This is needed because calling code might not expect untrusted
     * tables to appear in pg_temp at the front of its search path.
     */
    if (stmt->relation->relpersistence == RELPERSISTENCE_TEMP
        && InSecurityRestrictedOperation())
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("cannot create temporary table within security-restricted operation")));

    /*
     * Select tablespace to use.  If not specified, use default tablespace
     * (which may in turn default to database's default).
     */
    if (stmt->tablespacename)
    {
        tablespaceId = get_tablespace_oid(stmt->tablespacename, false);
    }
    else
    {
        tablespaceId = GetDefaultTablespace(stmt->relation->relpersistence);
        /* note InvalidOid is OK in this case */
    }

    /* Check permissions except when using database's default */
    if (OidIsValid(tablespaceId) && tablespaceId != MyDatabaseTableSpace)
    {
        AclResult   aclresult;

        aclresult = pg_tablespace_aclcheck(tablespaceId, GetUserId(),
                                           ACL_CREATE);
        if (aclresult != ACLCHECK_OK)
            aclcheck_error(aclresult, OBJECT_TABLESPACE,
                           get_tablespace_name(tablespaceId));
    }

    /* In all cases disallow placing user relations in pg_global */
    if (tablespaceId == GLOBALTABLESPACE_OID)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("only shared relations can be placed in pg_global tablespace")));

    /* Identify user ID that will own the table */
    if (!OidIsValid(ownerId))
        ownerId = GetUserId();

    /*
     * Parse and validate reloptions, if any.
     */
    reloptions = transformRelOptions((Datum) 0, stmt->options, NULL, validnsps,
                                     true, false);

    if (relkind == RELKIND_VIEW)
        (void) view_reloptions(reloptions, true);
    else
        (void) heap_reloptions(relkind, reloptions, true);

    if (stmt->ofTypename)
    {
        AclResult   aclresult;

        ofTypeId = typenameTypeId(NULL, stmt->ofTypename);

        aclresult = pg_type_aclcheck(ofTypeId, GetUserId(), ACL_USAGE);
        if (aclresult != ACLCHECK_OK)
            aclcheck_error_type(aclresult, ofTypeId);
    }
    else
        ofTypeId = InvalidOid;

    /*
     * Look up inheritance ancestors and generate relation schema, including
     * inherited attributes.  (Note that stmt->tableElts is destructively
     * modified by MergeAttributes.)
     */
    stmt->tableElts =
        MergeAttributes(stmt->tableElts, stmt->inhRelations,
                        stmt->relation->relpersistence,
                        stmt->partbound != NULL,
                        &inheritOids, &old_constraints, &parentOidCount);

    /*
     * Create a tuple descriptor from the relation schema.  Note that this
     * deals with column names, types, and NOT NULL constraints, but not
     * default values or CHECK constraints; we handle those below.
     */
    descriptor = BuildDescForRelation(stmt->tableElts);

    /*
     * Notice that we allow OIDs here only for plain tables and partitioned
     * tables, even though some other relkinds can support them.  This is
     * necessary because the default_with_oids GUC must apply only to plain
     * tables and not any other relkind; doing otherwise would break existing
     * pg_dump files.  We could allow explicit "WITH OIDS" while not allowing
     * default_with_oids to affect other relkinds, but it would complicate
     * interpretOidsOption().
     */
    localHasOids = interpretOidsOption(stmt->options,
                                       (relkind == RELKIND_RELATION ||
                                        relkind == RELKIND_PARTITIONED_TABLE));
    descriptor->tdhasoid = (localHasOids || parentOidCount > 0);

    /*
     * If a partitioned table doesn't have the system OID column, then none of
     * its partitions should have it.
     */
    if (stmt->partbound && parentOidCount == 0 && localHasOids)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("cannot create table with OIDs as partition of table without OIDs")));

    /*
     * Find columns with default values and prepare for insertion of the
     * defaults.  Pre-cooked (that is, inherited) defaults go into a list of
     * CookedConstraint structs that we'll pass to heap_create_with_catalog,
     * while raw defaults go into a list of RawColumnDefault structs that will
     * be processed by AddRelationNewConstraints.  (We can't deal with raw
     * expressions until we can do transformExpr.)
     *
     * We can set the atthasdef flags now in the tuple descriptor; this just
     * saves StoreAttrDefault from having to do an immediate update of the
     * pg_attribute rows.
     */
    rawDefaults = NIL;
    cookedDefaults = NIL;
    attnum = 0;

    foreach(listptr, stmt->tableElts)
    {
        ColumnDef  *colDef = lfirst(listptr);
        Form_pg_attribute attr;

        attnum++;
        attr = TupleDescAttr(descriptor, attnum - 1);

        if (colDef->raw_default != NULL)
        {
            RawColumnDefault *rawEnt;

            Assert(colDef->cooked_default == NULL);

            rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));
            rawEnt->attnum = attnum;
            rawEnt->raw_default = colDef->raw_default;
            rawEnt->missingMode = false;
            rawDefaults = lappend(rawDefaults, rawEnt);
            attr->atthasdef = true;
        }
        else if (colDef->cooked_default != NULL)
        {
            CookedConstraint *cooked;

            cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
            cooked->contype = CONSTR_DEFAULT;
            cooked->conoid = InvalidOid;    /* until created */
            cooked->name = NULL;
            cooked->attnum = attnum;
            cooked->expr = colDef->cooked_default;
            cooked->skip_validation = false;
            cooked->is_local = true;    /* not used for defaults */
            cooked->inhcount = 0;   /* ditto */
            cooked->is_no_inherit = false;
            cookedDefaults = lappend(cookedDefaults, cooked);
            attr->atthasdef = true;
        }

        if (colDef->identity)
            attr->attidentity = colDef->identity;
    }

    /*
     * Create the relation.  Inherited defaults and constraints are passed in
     * for immediate handling --- since they don't need parsing, they can be
     * stored immediately.
     */
    relationId = heap_create_with_catalog(relname,
                                          namespaceId,
                                          tablespaceId,
                                          InvalidOid,
                                          InvalidOid,
                                          ofTypeId,
                                          ownerId,
                                          descriptor,
                                          list_concat(cookedDefaults,
                                                      old_constraints),
                                          relkind,
                                          stmt->relation->relpersistence,
                                          false,
                                          false,
                                          localHasOids,
                                          parentOidCount,
                                          stmt->oncommit,
                                          reloptions,
                                          true,
                                          allowSystemTableMods,
                                          false,
                                          InvalidOid,
                                          typaddress);

    /* Store inheritance information for new rel. */
    StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL);

    /*
     * We must bump the command counter to make the newly-created relation
     * tuple visible for opening.
     */
    CommandCounterIncrement();

    /*
     * Open the new relation and acquire exclusive lock on it.  This isn't
     * really necessary for locking out other backends (since they can't see
     * the new rel anyway until we commit), but it keeps the lock manager from
     * complaining about deadlock risks.
     */
    rel = relation_open(relationId, AccessExclusiveLock);

    /* Process and store partition bound, if any. */
    if (stmt->partbound)
    {
        PartitionBoundSpec *bound;
        ParseState *pstate;
        Oid         parentId = linitial_oid(inheritOids),
                    defaultPartOid;
        Relation    parent,
                    defaultRel = NULL;

        /* Already have strong enough lock on the parent */
        parent = heap_open(parentId, NoLock);

        /*
         * We are going to try to validate the partition bound specification
         * against the partition key of parentRel, so it better have one.
         */
        if (parent->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
                     errmsg("\"%s\" is not partitioned",
                            RelationGetRelationName(parent))));

        /*
         * The partition constraint of the default partition depends on the
         * partition bounds of every other partition. It is possible that
         * another backend might be about to execute a query on the default
         * partition table, and that the query relies on previously cached
         * default partition constraints. We must therefore take a table lock
         * strong enough to prevent all queries on the default partition from
         * proceeding until we commit and send out a shared-cache-inval notice
         * that will make them update their index lists.
         *
         * Order of locking: The relation being added won't be visible to
         * other backends until it is committed, hence here in
         * DefineRelation() the order of locking the default partition and the
         * relation being added does not matter. But at all other places we
         * need to lock the default relation before we lock the relation being
         * added or removed i.e. we should take the lock in same order at all
         * the places such that lock parent, lock default partition and then
         * lock the partition so as to avoid a deadlock.
         */
        defaultPartOid =
            get_default_oid_from_partdesc(RelationGetPartitionDesc(parent));
        if (OidIsValid(defaultPartOid))
            defaultRel = heap_open(defaultPartOid, AccessExclusiveLock);

        /* Tranform the bound values */
        pstate = make_parsestate(NULL);
        pstate->p_sourcetext = queryString;

        bound = transformPartitionBound(pstate, parent, stmt->partbound);

        /*
         * Check first that the new partition's bound is valid and does not
         * overlap with any of existing partitions of the parent.
         */
        check_new_partition_bound(relname, parent, bound);

        /*
         * If the default partition exists, its partition constraints will
         * change after the addition of this new partition such that it won't
         * allow any row that qualifies for this new partition. So, check that
         * the existing data in the default partition satisfies the constraint
         * as it will exist after adding this partition.
         */
        if (OidIsValid(defaultPartOid))
        {
            check_default_allows_bound(parent, defaultRel, bound);
            /* Keep the lock until commit. */
            heap_close(defaultRel, NoLock);
        }

        /* Update the pg_class entry. */
        StorePartitionBound(rel, parent, bound);

        heap_close(parent, NoLock);
    }

    /*
     * Process the partitioning specification (if any) and store the partition
     * key information into the catalog.
     */
    if (stmt->partspec)
    {
        char        strategy;
        int         partnatts;
        AttrNumber  partattrs[PARTITION_MAX_KEYS];
        Oid         partopclass[PARTITION_MAX_KEYS];
        Oid         partcollation[PARTITION_MAX_KEYS];
        List       *partexprs = NIL;

        partnatts = list_length(stmt->partspec->partParams);

        /* Protect fixed-size arrays here and in executor */
        if (partnatts > PARTITION_MAX_KEYS)
            ereport(ERROR,
                    (errcode(ERRCODE_TOO_MANY_COLUMNS),
                     errmsg("cannot partition using more than %d columns",
                            PARTITION_MAX_KEYS)));

        /*
         * We need to transform the raw parsetrees corresponding to partition
         * expressions into executable expression trees.  Like column defaults
         * and CHECK constraints, we could not have done the transformation
         * earlier.
         */
        stmt->partspec = transformPartitionSpec(rel, stmt->partspec,
                                                &strategy);

        ComputePartitionAttrs(rel, stmt->partspec->partParams,
                              partattrs, &partexprs, partopclass,
                              partcollation, strategy);

        StorePartitionKey(rel, strategy, partnatts, partattrs, partexprs,
                          partopclass, partcollation);

        /* make it all visible */
        CommandCounterIncrement();
    }

    /*
     * If we're creating a partition, create now all the indexes, triggers,
     * FKs defined in the parent.
     *
     * We can't do it earlier, because DefineIndex wants to know the partition
     * key which we just stored.
     */
    if (stmt->partbound)
    {
        Oid         parentId = linitial_oid(inheritOids);
        Relation    parent;
        List       *idxlist;
        ListCell   *cell;

        /* Already have strong enough lock on the parent */
        parent = heap_open(parentId, NoLock);
        idxlist = RelationGetIndexList(parent);

        /*
         * For each index in the parent table, create one in the partition
         */
        foreach(cell, idxlist)
        {
            Relation    idxRel = index_open(lfirst_oid(cell), AccessShareLock);
            AttrNumber *attmap;
            IndexStmt  *idxstmt;
            Oid         constraintOid;

            attmap = convert_tuples_by_name_map(RelationGetDescr(rel),
                                                RelationGetDescr(parent),
                                                gettext_noop("could not convert row type"));
            idxstmt =
                generateClonedIndexStmt(NULL, RelationGetRelid(rel), idxRel,
                                        attmap, RelationGetDescr(rel)->natts,
                                        &constraintOid);
            DefineIndex(RelationGetRelid(rel),
                        idxstmt,
                        InvalidOid,
                        RelationGetRelid(idxRel),
                        constraintOid,
                        false, false, false, false, false);

            index_close(idxRel, AccessShareLock);
        }

        list_free(idxlist);

        /*
         * If there are any row-level triggers, clone them to the new
         * partition.
         */
        if (parent->trigdesc != NULL)
            CloneRowTriggersToPartition(parent, rel);

        /*
         * And foreign keys too.  Note that because we're freshly creating the
         * table, there is no need to verify these new constraints.
         */
        CloneForeignKeyConstraints(parentId, relationId, NULL);

        heap_close(parent, NoLock);
    }

    /*
     * Now add any newly specified column default values and CHECK constraints
     * to the new relation.  These are passed to us in the form of raw
     * parsetrees; we need to transform them to executable expression trees
     * before they can be added. The most convenient way to do that is to
     * apply the parser's transformExpr routine, but transformExpr doesn't
     * work unless we have a pre-existing relation. So, the transformation has
     * to be postponed to this final step of CREATE TABLE.
     */
    if (rawDefaults || stmt->constraints)
        AddRelationNewConstraints(rel, rawDefaults, stmt->constraints,
                                  true, true, false);

    ObjectAddressSet(address, RelationRelationId, relationId);

    /*
     * Clean up.  We keep lock on new relation (although it shouldn't be
     * visible to anyone else anyway, until commit).
     */
    relation_close(rel, NoLock);

    return address;
}

/*
 * Emit the right error or warning message for a "DROP" command issued on a
 * non-existent relation
 */
static void
DropErrorMsgNonExistent(RangeVar *rel, char rightkind, bool missing_ok)
{
    const struct dropmsgstrings *rentry;

    if (rel->schemaname != NULL &&
        !OidIsValid(LookupNamespaceNoError(rel->schemaname)))
    {
        if (!missing_ok)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_SCHEMA),
                     errmsg("schema \"%s\" does not exist", rel->schemaname)));
        }
        else
        {
            ereport(NOTICE,
                    (errmsg("schema \"%s\" does not exist, skipping",
                            rel->schemaname)));
        }
        return;
    }

    for (rentry = dropmsgstringarray; rentry->kind != '\0'; rentry++)
    {
        if (rentry->kind == rightkind)
        {
            if (!missing_ok)
            {
                ereport(ERROR,
                        (errcode(rentry->nonexistent_code),
                         errmsg(rentry->nonexistent_msg, rel->relname)));
            }
            else
            {
                ereport(NOTICE, (errmsg(rentry->skipping_msg, rel->relname)));
                break;
            }
        }
    }

    Assert(rentry->kind != '\0');   /* Should be impossible */
}

/*
 * Emit the right error message for a "DROP" command issued on a
 * relation of the wrong type
 */
static void
DropErrorMsgWrongType(const char *relname, char wrongkind, char rightkind)
{
    const struct dropmsgstrings *rentry;
    const struct dropmsgstrings *wentry;

    for (rentry = dropmsgstringarray; rentry->kind != '\0'; rentry++)
        if (rentry->kind == rightkind)
            break;
    Assert(rentry->kind != '\0');

    for (wentry = dropmsgstringarray; wentry->kind != '\0'; wentry++)
        if (wentry->kind == wrongkind)
            break;
    /* wrongkind could be something we don't have in our table... */

    ereport(ERROR,
            (errcode(ERRCODE_WRONG_OBJECT_TYPE),
             errmsg(rentry->nota_msg, relname),
             (wentry->kind != '\0') ? errhint("%s", _(wentry->drophint_msg)) : 0));
}

/*
 * RemoveRelations
 *      Implements DROP TABLE, DROP INDEX, DROP SEQUENCE, DROP VIEW,
 *      DROP MATERIALIZED VIEW, DROP FOREIGN TABLE
 */
void
RemoveRelations(DropStmt *drop)
{
    ObjectAddresses *objects;
    char        relkind;
    ListCell   *cell;
    int         flags = 0;
    LOCKMODE    lockmode = AccessExclusiveLock;

    /* DROP CONCURRENTLY uses a weaker lock, and has some restrictions */
    if (drop->concurrent)
    {
        flags |= PERFORM_DELETION_CONCURRENTLY;
        lockmode = ShareUpdateExclusiveLock;
        Assert(drop->removeType == OBJECT_INDEX);
        if (list_length(drop->objects) != 1)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("DROP INDEX CONCURRENTLY does not support dropping multiple objects")));
        if (drop->behavior == DROP_CASCADE)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("DROP INDEX CONCURRENTLY does not support CASCADE")));
    }

    /*
     * First we identify all the relations, then we delete them in a single
     * performMultipleDeletions() call.  This is to avoid unwanted DROP
     * RESTRICT errors if one of the relations depends on another.
     */

    /* Determine required relkind */
    switch (drop->removeType)
    {
        case OBJECT_TABLE:
            relkind = RELKIND_RELATION;
            break;

        case OBJECT_INDEX:
            relkind = RELKIND_INDEX;
            break;

        case OBJECT_SEQUENCE:
            relkind = RELKIND_SEQUENCE;
            break;

        case OBJECT_VIEW:
            relkind = RELKIND_VIEW;
            break;

        case OBJECT_MATVIEW:
            relkind = RELKIND_MATVIEW;
            break;

        case OBJECT_FOREIGN_TABLE:
            relkind = RELKIND_FOREIGN_TABLE;
            break;

        default:
            elog(ERROR, "unrecognized drop object type: %d",
                 (int) drop->removeType);
            relkind = 0;        /* keep compiler quiet */
            break;
    }

    /* Lock and validate each relation; build a list of object addresses */
    objects = new_object_addresses();

    foreach(cell, drop->objects)
    {
        RangeVar   *rel = makeRangeVarFromNameList((List *) lfirst(cell));
        Oid         relOid;
        ObjectAddress obj;
        struct DropRelationCallbackState state;

        /*
         * These next few steps are a great deal like relation_openrv, but we
         * don't bother building a relcache entry since we don't need it.
         *
         * Check for shared-cache-inval messages before trying to access the
         * relation.  This is needed to cover the case where the name
         * identifies a rel that has been dropped and recreated since the
         * start of our transaction: if we don't flush the old syscache entry,
         * then we'll latch onto that entry and suffer an error later.
         */
        AcceptInvalidationMessages();

        /* Look up the appropriate relation using namespace search. */
        state.relkind = relkind;
        state.heapOid = InvalidOid;
        state.partParentOid = InvalidOid;
        state.concurrent = drop->concurrent;
        relOid = RangeVarGetRelidExtended(rel, lockmode, RVR_MISSING_OK,
                                          RangeVarCallbackForDropRelation,
                                          (void *) &state);

        /* Not there? */
        if (!OidIsValid(relOid))
        {
            DropErrorMsgNonExistent(rel, relkind, drop->missing_ok);
            continue;
        }

        /* OK, we're ready to delete this one */
        obj.classId = RelationRelationId;
        obj.objectId = relOid;
        obj.objectSubId = 0;

        add_exact_object_address(&obj, objects);
    }

    performMultipleDeletions(objects, drop->behavior, flags);

    free_object_addresses(objects);
}

/*
 * Before acquiring a table lock, check whether we have sufficient rights.
 * In the case of DROP INDEX, also try to lock the table before the index.
 * Also, if the table to be dropped is a partition, we try to lock the parent
 * first.
 */
static void
RangeVarCallbackForDropRelation(const RangeVar *rel, Oid relOid, Oid oldRelOid,
                                void *arg)
{
    HeapTuple   tuple;
    struct DropRelationCallbackState *state;
    char        relkind;
    char        expected_relkind;
    bool        is_partition;
    Form_pg_class classform;
    LOCKMODE    heap_lockmode;

    state = (struct DropRelationCallbackState *) arg;
    relkind = state->relkind;
    heap_lockmode = state->concurrent ?
        ShareUpdateExclusiveLock : AccessExclusiveLock;

    /*
     * If we previously locked some other index's heap, and the name we're
     * looking up no longer refers to that relation, release the now-useless
     * lock.
     */
    if (relOid != oldRelOid && OidIsValid(state->heapOid))
    {
        UnlockRelationOid(state->heapOid, heap_lockmode);
        state->heapOid = InvalidOid;
    }

    /*
     * Similarly, if we previously locked some other partition's heap, and the
     * name we're looking up no longer refers to that relation, release the
     * now-useless lock.
     */
    if (relOid != oldRelOid && OidIsValid(state->partParentOid))
    {
        UnlockRelationOid(state->partParentOid, AccessExclusiveLock);
        state->partParentOid = InvalidOid;
    }

    /* Didn't find a relation, so no need for locking or permission checks. */
    if (!OidIsValid(relOid))
        return;

    tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relOid));
    if (!HeapTupleIsValid(tuple))
        return;                 /* concurrently dropped, so nothing to do */
    classform = (Form_pg_class) GETSTRUCT(tuple);
    is_partition = classform->relispartition;

    /*
     * Both RELKIND_RELATION and RELKIND_PARTITIONED_TABLE are OBJECT_TABLE,
     * but RemoveRelations() can only pass one relkind for a given relation.
     * It chooses RELKIND_RELATION for both regular and partitioned tables.
     * That means we must be careful before giving the wrong type error when
     * the relation is RELKIND_PARTITIONED_TABLE.  An equivalent problem
     * exists with indexes.
     */
    if (classform->relkind == RELKIND_PARTITIONED_TABLE)
        expected_relkind = RELKIND_RELATION;
    else if (classform->relkind == RELKIND_PARTITIONED_INDEX)
        expected_relkind = RELKIND_INDEX;
    else
        expected_relkind = classform->relkind;

    if (relkind != expected_relkind)
        DropErrorMsgWrongType(rel->relname, classform->relkind, relkind);

    /* Allow DROP to either table owner or schema owner */
    if (!pg_class_ownercheck(relOid, GetUserId()) &&
        !pg_namespace_ownercheck(classform->relnamespace, GetUserId()))
        aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(relOid)),
                       rel->relname);

    if (!allowSystemTableMods && IsSystemClass(relOid, classform))
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("permission denied: \"%s\" is a system catalog",
                        rel->relname)));

    ReleaseSysCache(tuple);

    /*
     * In DROP INDEX, attempt to acquire lock on the parent table before
     * locking the index.  index_drop() will need this anyway, and since
     * regular queries lock tables before their indexes, we risk deadlock if
     * we do it the other way around.  No error if we don't find a pg_index
     * entry, though --- the relation may have been dropped.
     */
    if ((relkind == RELKIND_INDEX || relkind == RELKIND_PARTITIONED_INDEX) &&
        relOid != oldRelOid)
    {
        state->heapOid = IndexGetRelation(relOid, true);
        if (OidIsValid(state->heapOid))
            LockRelationOid(state->heapOid, heap_lockmode);
    }

    /*
     * Similarly, if the relation is a partition, we must acquire lock on its
     * parent before locking the partition.  That's because queries lock the
     * parent before its partitions, so we risk deadlock it we do it the other
     * way around.
     */
    if (is_partition && relOid != oldRelOid)
    {
        state->partParentOid = get_partition_parent(relOid);
        if (OidIsValid(state->partParentOid))
            LockRelationOid(state->partParentOid, AccessExclusiveLock);
    }
}

/*
 * ExecuteTruncate
 *      Executes a TRUNCATE command.
 *
 * This is a multi-relation truncate.  We first open and grab exclusive
 * lock on all relations involved, checking permissions and otherwise
 * verifying that the relation is OK for truncation.  In CASCADE mode,
 * relations having FK references to the targeted relations are automatically
 * added to the group; in RESTRICT mode, we check that all FK references are
 * internal to the group that's being truncated.  Finally all the relations
 * are truncated and reindexed.
 */
void
ExecuteTruncate(TruncateStmt *stmt)
{
    List       *rels = NIL;
    List       *relids = NIL;
    List       *relids_logged = NIL;
    ListCell   *cell;

    /*
     * Open, exclusive-lock, and check all the explicitly-specified relations
     */
    foreach(cell, stmt->relations)
    {
        RangeVar   *rv = lfirst(cell);
        Relation    rel;
        bool        recurse = rv->inh;
        Oid         myrelid;

        rel = heap_openrv(rv, AccessExclusiveLock);
        myrelid = RelationGetRelid(rel);
        /* don't throw error for "TRUNCATE foo, foo" */
        if (list_member_oid(relids, myrelid))
        {
            heap_close(rel, AccessExclusiveLock);
            continue;
        }
        truncate_check_rel(rel);
        rels = lappend(rels, rel);
        relids = lappend_oid(relids, myrelid);
        /* Log this relation only if needed for logical decoding */
        if (RelationIsLogicallyLogged(rel))
            relids_logged = lappend_oid(relids_logged, myrelid);

        if (recurse)
        {
            ListCell   *child;
            List       *children;

            children = find_all_inheritors(myrelid, AccessExclusiveLock, NULL);

            foreach(child, children)
            {
                Oid         childrelid = lfirst_oid(child);

                if (list_member_oid(relids, childrelid))
                    continue;

                /* find_all_inheritors already got lock */
                rel = heap_open(childrelid, NoLock);
                truncate_check_rel(rel);
                rels = lappend(rels, rel);
                relids = lappend_oid(relids, childrelid);
                /* Log this relation only if needed for logical decoding */
                if (RelationIsLogicallyLogged(rel))
                    relids_logged = lappend_oid(relids_logged, childrelid);
            }
        }
        else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg("cannot truncate only a partitioned table"),
                     errhint("Do not specify the ONLY keyword, or use TRUNCATE ONLY on the partitions directly.")));
    }

    ExecuteTruncateGuts(rels, relids, relids_logged,
                        stmt->behavior, stmt->restart_seqs);

    /* And close the rels */
    foreach(cell, rels)
    {
        Relation    rel = (Relation) lfirst(cell);

        heap_close(rel, NoLock);
    }
}

/*
 * ExecuteTruncateGuts
 *
 * Internal implementation of TRUNCATE.  This is called by the actual TRUNCATE
 * command (see above) as well as replication subscribers that execute a
 * replicated TRUNCATE action.
 *
 * explicit_rels is the list of Relations to truncate that the command
 * specified.  relids is the list of Oids corresponding to explicit_rels.
 * relids_logged is the list of Oids (a subset of relids) that require
 * WAL-logging.  This is all a bit redundant, but the existing callers have
 * this information handy in this form.
 */
void
ExecuteTruncateGuts(List *explicit_rels, List *relids, List *relids_logged,
                    DropBehavior behavior, bool restart_seqs)
{
    List       *rels;
    List       *seq_relids = NIL;
    EState     *estate;
    ResultRelInfo *resultRelInfos;
    ResultRelInfo *resultRelInfo;
    SubTransactionId mySubid;
    ListCell   *cell;
    Oid        *logrelids;

    /*
     * Open, exclusive-lock, and check all the explicitly-specified relations
     *
     * In CASCADE mode, suck in all referencing relations as well.  This
     * requires multiple iterations to find indirectly-dependent relations. At
     * each phase, we need to exclusive-lock new rels before looking for their
     * dependencies, else we might miss something.  Also, we check each rel as
     * soon as we open it, to avoid a faux pas such as holding lock for a long
     * time on a rel we have no permissions for.
     */
    rels = list_copy(explicit_rels);
    if (behavior == DROP_CASCADE)
    {
        for (;;)
        {
            List       *newrelids;

            newrelids = heap_truncate_find_FKs(relids);
            if (newrelids == NIL)
                break;          /* nothing else to add */

            foreach(cell, newrelids)
            {
                Oid         relid = lfirst_oid(cell);
                Relation    rel;

                rel = heap_open(relid, AccessExclusiveLock);
                ereport(NOTICE,
                        (errmsg("truncate cascades to table \"%s\"",
                                RelationGetRelationName(rel))));
                truncate_check_rel(rel);
                rels = lappend(rels, rel);
                relids = lappend_oid(relids, relid);
                /* Log this relation only if needed for logical decoding */
                if (RelationIsLogicallyLogged(rel))
                    relids_logged = lappend_oid(relids_logged, relid);
            }
        }
    }

    /*
     * Check foreign key references.  In CASCADE mode, this should be
     * unnecessary since we just pulled in all the references; but as a
     * cross-check, do it anyway if in an Assert-enabled build.
     */
#ifdef USE_ASSERT_CHECKING
    heap_truncate_check_FKs(rels, false);
#else
    if (behavior == DROP_RESTRICT)
        heap_truncate_check_FKs(rels, false);
#endif

    /*
     * If we are asked to restart sequences, find all the sequences, lock them
     * (we need AccessExclusiveLock for ResetSequence), and check permissions.
     * We want to do this early since it's pointless to do all the truncation
     * work only to fail on sequence permissions.
     */
    if (restart_seqs)
    {
        foreach(cell, rels)
        {
            Relation    rel = (Relation) lfirst(cell);
            List       *seqlist = getOwnedSequences(RelationGetRelid(rel), 0);
            ListCell   *seqcell;

            foreach(seqcell, seqlist)
            {
                Oid         seq_relid = lfirst_oid(seqcell);
                Relation    seq_rel;

                seq_rel = relation_open(seq_relid, AccessExclusiveLock);

                /* This check must match AlterSequence! */
                if (!pg_class_ownercheck(seq_relid, GetUserId()))
                    aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SEQUENCE,
                                   RelationGetRelationName(seq_rel));

                seq_relids = lappend_oid(seq_relids, seq_relid);

                relation_close(seq_rel, NoLock);
            }
        }
    }

    /* Prepare to catch AFTER triggers. */
    AfterTriggerBeginQuery();

    /*
     * To fire triggers, we'll need an EState as well as a ResultRelInfo for
     * each relation.  We don't need to call ExecOpenIndices, though.
     */
    estate = CreateExecutorState();
    resultRelInfos = (ResultRelInfo *)
        palloc(list_length(rels) * sizeof(ResultRelInfo));
    resultRelInfo = resultRelInfos;
    foreach(cell, rels)
    {
        Relation    rel = (Relation) lfirst(cell);

        InitResultRelInfo(resultRelInfo,
                          rel,
                          0,    /* dummy rangetable index */
                          NULL,
                          0);
        resultRelInfo++;
    }
    estate->es_result_relations = resultRelInfos;
    estate->es_num_result_relations = list_length(rels);

    /*
     * Process all BEFORE STATEMENT TRUNCATE triggers before we begin
     * truncating (this is because one of them might throw an error). Also, if
     * we were to allow them to prevent statement execution, that would need
     * to be handled here.
     */
    resultRelInfo = resultRelInfos;
    foreach(cell, rels)
    {
        estate->es_result_relation_info = resultRelInfo;
        ExecBSTruncateTriggers(estate, resultRelInfo);
        resultRelInfo++;
    }

    /*
     * OK, truncate each table.
     */
    mySubid = GetCurrentSubTransactionId();

    foreach(cell, rels)
    {
        Relation    rel = (Relation) lfirst(cell);

        /* Skip partitioned tables as there is nothing to do */
        if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
            continue;

        /*
         * Normally, we need a transaction-safe truncation here.  However, if
         * the table was either created in the current (sub)transaction or has
         * a new relfilenode in the current (sub)transaction, then we can just
         * truncate it in-place, because a rollback would cause the whole
         * table or the current physical file to be thrown away anyway.
         */
        if (rel->rd_createSubid == mySubid ||
            rel->rd_newRelfilenodeSubid == mySubid)
        {
            /* Immediate, non-rollbackable truncation is OK */
            heap_truncate_one_rel(rel);
        }
        else
        {
            Oid         heap_relid;
            Oid         toast_relid;
            MultiXactId minmulti;

            /*
             * This effectively deletes all rows in the table, and may be done
             * in a serializable transaction.  In that case we must record a
             * rw-conflict in to this transaction from each transaction
             * holding a predicate lock on the table.
             */
            CheckTableForSerializableConflictIn(rel);

            minmulti = GetOldestMultiXactId();

            /*
             * Need the full transaction-safe pushups.
             *
             * Create a new empty storage file for the relation, and assign it
             * as the relfilenode value. The old storage file is scheduled for
             * deletion at commit.
             */
            RelationSetNewRelfilenode(rel, rel->rd_rel->relpersistence,
                                      RecentXmin, minmulti);
            if (rel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED)
                heap_create_init_fork(rel);

            heap_relid = RelationGetRelid(rel);
            toast_relid = rel->rd_rel->reltoastrelid;

            /*
             * The same for the toast table, if any.
             */
            if (OidIsValid(toast_relid))
            {
                rel = relation_open(toast_relid, AccessExclusiveLock);
                RelationSetNewRelfilenode(rel, rel->rd_rel->relpersistence,
                                          RecentXmin, minmulti);
                if (rel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED)
                    heap_create_init_fork(rel);
                heap_close(rel, NoLock);
            }

            /*
             * Reconstruct the indexes to match, and we're done.
             */
            reindex_relation(heap_relid, REINDEX_REL_PROCESS_TOAST, 0);
        }

        pgstat_count_truncate(rel);
    }

    /*
     * Restart owned sequences if we were asked to.
     */
    foreach(cell, seq_relids)
    {
        Oid         seq_relid = lfirst_oid(cell);

        ResetSequence(seq_relid);
    }

    /*
     * Write a WAL record to allow this set of actions to be logically
     * decoded.
     *
     * Assemble an array of relids so we can write a single WAL record for the
     * whole action.
     */
    if (list_length(relids_logged) > 0)
    {
        xl_heap_truncate xlrec;
        int         i = 0;

        /* should only get here if wal_level >= logical */
        Assert(XLogLogicalInfoActive());

        logrelids = palloc(list_length(relids_logged) * sizeof(Oid));
        foreach(cell, relids_logged)
            logrelids[i++] = lfirst_oid(cell);

        xlrec.dbId = MyDatabaseId;
        xlrec.nrelids = list_length(relids_logged);
        xlrec.flags = 0;
        if (behavior == DROP_CASCADE)
            xlrec.flags |= XLH_TRUNCATE_CASCADE;
        if (restart_seqs)
            xlrec.flags |= XLH_TRUNCATE_RESTART_SEQS;

        XLogBeginInsert();
        XLogRegisterData((char *) &xlrec, SizeOfHeapTruncate);
        XLogRegisterData((char *) logrelids, list_length(relids_logged) * sizeof(Oid));

        XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);

        (void) XLogInsert(RM_HEAP_ID, XLOG_HEAP_TRUNCATE);
    }

    /*
     * Process all AFTER STATEMENT TRUNCATE triggers.
     */
    resultRelInfo = resultRelInfos;
    foreach(cell, rels)
    {
        estate->es_result_relation_info = resultRelInfo;
        ExecASTruncateTriggers(estate, resultRelInfo);
        resultRelInfo++;
    }

    /* Handle queued AFTER triggers */
    AfterTriggerEndQuery(estate);

    /* We can clean up the EState now */
    FreeExecutorState(estate);

    /*
     * Close any rels opened by CASCADE (can't do this while EState still
     * holds refs)
     */
    rels = list_difference_ptr(rels, explicit_rels);
    foreach(cell, rels)
    {
        Relation    rel = (Relation) lfirst(cell);

        heap_close(rel, NoLock);
    }
}

/*
 * Check that a given rel is safe to truncate.  Subroutine for ExecuteTruncate
 */
static void
truncate_check_rel(Relation rel)
{
    AclResult   aclresult;

    /*
     * Only allow truncate on regular tables and partitioned tables (although,
     * the latter are only being included here for the following checks; no
     * physical truncation will occur in their case.)
     */
    if (rel->rd_rel->relkind != RELKIND_RELATION &&
        rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("\"%s\" is not a table",
                        RelationGetRelationName(rel))));

    /* Permissions checks */
    aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
                                  ACL_TRUNCATE);
    if (aclresult != ACLCHECK_OK)
        aclcheck_error(aclresult, get_relkind_objtype(rel->rd_rel->relkind),
                       RelationGetRelationName(rel));

    if (!allowSystemTableMods && IsSystemRelation(rel))
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("permission denied: \"%s\" is a system catalog",
                        RelationGetRelationName(rel))));

    /*
     * Don't allow truncate on temp tables of other backends ... their local
     * buffer manager is not going to cope.
     */
    if (RELATION_IS_OTHER_TEMP(rel))
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot truncate temporary tables of other sessions")));

    /*
     * Also check for active uses of the relation in the current transaction,
     * including open scans and pending AFTER trigger events.
     */
    CheckTableNotInUse(rel, "TRUNCATE");
}

/*
 * storage_name
 *    returns the name corresponding to a typstorage/attstorage enum value
 */
static const char *
storage_name(char c)
{
    switch (c)
    {
        case 'p':
            return "PLAIN";
        case 'm':
            return "MAIN";
        case 'x':
            return "EXTENDED";
        case 'e':
            return "EXTERNAL";
        default:
            return "???";
    }
}

/*----------
 * MergeAttributes
 *      Returns new schema given initial schema and superclasses.
 *
 * Input arguments:
 * 'schema' is the column/attribute definition for the table. (It's a list
 *      of ColumnDef's.) It is destructively changed.
 * 'supers' is a list of names (as RangeVar nodes) of parent relations.
 * 'relpersistence' is a persistence type of the table.
 * 'is_partition' tells if the table is a partition
 *
 * Output arguments:
 * 'supOids' receives a list of the OIDs of the parent relations.
 * 'supconstr' receives a list of constraints belonging to the parents,
 *      updated as necessary to be valid for the child.
 * 'supOidCount' is set to the number of parents that have OID columns.
 *
 * Return value:
 * Completed schema list.
 *
 * Notes:
 *    The order in which the attributes are inherited is very important.
 *    Intuitively, the inherited attributes should come first. If a table
 *    inherits from multiple parents, the order of those attributes are
 *    according to the order of the parents specified in CREATE TABLE.
 *
 *    Here's an example:
 *
 *      create table person (name text, age int4, location point);
 *      create table emp (salary int4, manager text) inherits(person);
 *      create table student (gpa float8) inherits (person);
 *      create table stud_emp (percent int4) inherits (emp, student);
 *
 *    The order of the attributes of stud_emp is:
 *
 *                          person {1:name, 2:age, 3:location}
 *                          /    \
 *             {6:gpa}  student   emp {4:salary, 5:manager}
 *                          \    /
 *                         stud_emp {7:percent}
 *
 *     If the same attribute name appears multiple times, then it appears
 *     in the result table in the proper location for its first appearance.
 *
 *     Constraints (including NOT NULL constraints) for the child table
 *     are the union of all relevant constraints, from both the child schema
 *     and parent tables.
 *
 *     The default value for a child column is defined as:
 *      (1) If the child schema specifies a default, that value is used.
 *      (2) If neither the child nor any parent specifies a default, then
 *          the column will not have a default.
 *      (3) If conflicting defaults are inherited from different parents
 *          (and not overridden by the child), an error is raised.
 *      (4) Otherwise the inherited default is used.
 *      Rule (3) is new in Postgres 7.1; in earlier releases you got a
 *      rather arbitrary choice of which parent default to use.
 *----------
 */
static List *
MergeAttributes(List *schema, List *supers, char relpersistence,
                bool is_partition, List **supOids, List **supconstr,
                int *supOidCount)
{
    ListCell   *entry;
    List       *inhSchema = NIL;
    List       *parentOids = NIL;
    List       *constraints = NIL;
    int         parentsWithOids = 0;
    bool        have_bogus_defaults = false;
    int         child_attno;
    static Node bogus_marker = {0}; /* marks conflicting defaults */
    List       *saved_schema = NIL;

    /*
     * Check for and reject tables with too many columns. We perform this
     * check relatively early for two reasons: (a) we don't run the risk of
     * overflowing an AttrNumber in subsequent code (b) an O(n^2) algorithm is
     * okay if we're processing <= 1600 columns, but could take minutes to
     * execute if the user attempts to create a table with hundreds of
     * thousands of columns.
     *
     * Note that we also need to check that we do not exceed this figure after
     * including columns from inherited relations.
     */
    if (list_length(schema) > MaxHeapAttributeNumber)
        ereport(ERROR,
                (errcode(ERRCODE_TOO_MANY_COLUMNS),
                 errmsg("tables can have at most %d columns",
                        MaxHeapAttributeNumber)));

    /*
     * In case of a partition, there are no new column definitions, only dummy
     * ColumnDefs created for column constraints.  We merge them with the
     * constraints inherited from the parent.
     */
    if (is_partition)
    {
        saved_schema = schema;
        schema = NIL;
    }

    /*
     * Check for duplicate names in the explicit list of attributes.
     *
     * Although we might consider merging such entries in the same way that we
     * handle name conflicts for inherited attributes, it seems to make more
     * sense to assume such conflicts are errors.
     */
    foreach(entry, schema)
    {
        ColumnDef  *coldef = lfirst(entry);
        ListCell   *rest = lnext(entry);
        ListCell   *prev = entry;

        if (coldef->typeName == NULL)

            /*
             * Typed table column option that does not belong to a column from
             * the type.  This works because the columns from the type come
             * first in the list.
             */
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_COLUMN),
                     errmsg("column \"%s\" does not exist",
                            coldef->colname)));

        while (rest != NULL)
        {
            ColumnDef  *restdef = lfirst(rest);
            ListCell   *next = lnext(rest); /* need to save it in case we
                                             * delete it */

            if (strcmp(coldef->colname, restdef->colname) == 0)
            {
                if (coldef->is_from_type)
                {
                    /*
                     * merge the column options into the column from the type
                     */
                    coldef->is_not_null = restdef->is_not_null;
                    coldef->raw_default = restdef->raw_default;
                    coldef->cooked_default = restdef->cooked_default;
                    coldef->constraints = restdef->constraints;
                    coldef->is_from_type = false;
                    list_delete_cell(schema, rest, prev);
                }
                else
                    ereport(ERROR,
                            (errcode(ERRCODE_DUPLICATE_COLUMN),
                             errmsg("column \"%s\" specified more than once",
                                    coldef->colname)));
            }
            prev = rest;
            rest = next;
        }
    }

    /*
     * Scan the parents left-to-right, and merge their attributes to form a
     * list of inherited attributes (inhSchema).  Also check to see if we need
     * to inherit an OID column.
     */
    child_attno = 0;
    foreach(entry, supers)
    {
        RangeVar   *parent = (RangeVar *) lfirst(entry);
        Relation    relation;
        TupleDesc   tupleDesc;
        TupleConstr *constr;
        AttrNumber *newattno;
        AttrNumber  parent_attno;

        /*
         * A self-exclusive lock is needed here.  If two backends attempt to
         * add children to the same parent simultaneously, and that parent has
         * no pre-existing children, then both will attempt to update the
         * parent's relhassubclass field, leading to a "tuple concurrently
         * updated" error.  Also, this interlocks against a concurrent ANALYZE
         * on the parent table, which might otherwise be attempting to clear
         * the parent's relhassubclass field, if its previous children were
         * recently dropped.
         *
         * If the child table is a partition, then we instead grab an
         * exclusive lock on the parent because its partition descriptor will
         * be changed by addition of the new partition.
         */
        if (!is_partition)
            relation = heap_openrv(parent, ShareUpdateExclusiveLock);
        else
            relation = heap_openrv(parent, AccessExclusiveLock);

        /*
         * We do not allow partitioned tables and partitions to participate in
         * regular inheritance.
         */
        if (relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE &&
            !is_partition)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg("cannot inherit from partitioned table \"%s\"",
                            parent->relname)));
        if (relation->rd_rel->relispartition && !is_partition)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg("cannot inherit from partition \"%s\"",
                            parent->relname)));

        if (relation->rd_rel->relkind != RELKIND_RELATION &&
            relation->rd_rel->relkind != RELKIND_FOREIGN_TABLE &&
            relation->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg("inherited relation \"%s\" is not a table or foreign table",
                            parent->relname)));
        /* Permanent rels cannot inherit from temporary ones */
        if (relpersistence != RELPERSISTENCE_TEMP &&
            relation->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg(!is_partition
                            ? "cannot inherit from temporary relation \"%s\""
                            : "cannot create a permanent relation as partition of temporary relation \"%s\"",
                            parent->relname)));

        /* If existing rel is temp, it must belong to this session */
        if (relation->rd_rel->relpersistence == RELPERSISTENCE_TEMP &&
            !relation->rd_islocaltemp)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg(!is_partition
                            ? "cannot inherit from temporary relation of another session"
                            : "cannot create as partition of temporary relation of another session")));

        /*
         * We should have an UNDER permission flag for this, but for now,
         * demand that creator of a child table own the parent.
         */
        if (!pg_class_ownercheck(RelationGetRelid(relation), GetUserId()))
            aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(relation->rd_rel->relkind),
                           RelationGetRelationName(relation));

        /*
         * Reject duplications in the list of parents.
         */
        if (list_member_oid(parentOids, RelationGetRelid(relation)))
            ereport(ERROR,
                    (errcode(ERRCODE_DUPLICATE_TABLE),
                     errmsg("relation \"%s\" would be inherited from more than once",
                            parent->relname)));

        parentOids = lappend_oid(parentOids, RelationGetRelid(relation));

        if (relation->rd_rel->relhasoids)
            parentsWithOids++;

        tupleDesc = RelationGetDescr(relation);
        constr = tupleDesc->constr;

        /*
         * newattno[] will contain the child-table attribute numbers for the
         * attributes of this parent table.  (They are not the same for
         * parents after the first one, nor if we have dropped columns.)
         */
        newattno = (AttrNumber *)
            palloc0(tupleDesc->natts * sizeof(AttrNumber));

        for (parent_attno = 1; parent_attno <= tupleDesc->natts;
             parent_attno++)
        {
            Form_pg_attribute attribute = TupleDescAttr(tupleDesc,
                                                        parent_attno - 1);
            char       *attributeName = NameStr(attribute->attname);
            int         exist_attno;
            ColumnDef  *def;

            /*
             * Ignore dropped columns in the parent.
             */
            if (attribute->attisdropped)
                continue;       /* leave newattno entry as zero */

            /*
             * Does it conflict with some previously inherited column?
             */
            exist_attno = findAttrByName(attributeName, inhSchema);
            if (exist_attno > 0)
            {
                Oid         defTypeId;
                int32       deftypmod;
                Oid         defCollId;

                /*
                 * Yes, try to merge the two column definitions. They must
                 * have the same type, typmod, and collation.
                 */
                ereport(NOTICE,
                        (errmsg("merging multiple inherited definitions of column \"%s\"",
                                attributeName)));
                def = (ColumnDef *) list_nth(inhSchema, exist_attno - 1);
                typenameTypeIdAndMod(NULL, def->typeName, &defTypeId, &deftypmod);
                if (defTypeId != attribute->atttypid ||
                    deftypmod != attribute->atttypmod)
                    ereport(ERROR,
                            (errcode(ERRCODE_DATATYPE_MISMATCH),
                             errmsg("inherited column \"%s\" has a type conflict",
                                    attributeName),
                             errdetail("%s versus %s",
                                       format_type_with_typemod(defTypeId,
                                                                deftypmod),
                                       format_type_with_typemod(attribute->atttypid,
                                                                attribute->atttypmod))));
                defCollId = GetColumnDefCollation(NULL, def, defTypeId);
                if (defCollId != attribute->attcollation)
                    ereport(ERROR,
                            (errcode(ERRCODE_COLLATION_MISMATCH),
                             errmsg("inherited column \"%s\" has a collation conflict",
                                    attributeName),
                             errdetail("\"%s\" versus \"%s\"",
                                       get_collation_name(defCollId),
                                       get_collation_name(attribute->attcollation))));

                /* Copy storage parameter */
                if (def->storage == 0)
                    def->storage = attribute->attstorage;
                else if (def->storage != attribute->attstorage)
                    ereport(ERROR,
                            (errcode(ERRCODE_DATATYPE_MISMATCH),
                             errmsg("inherited column \"%s\" has a storage parameter conflict",
                                    attributeName),
                             errdetail("%s versus %s",
                                       storage_name(def->storage),
                                       storage_name(attribute->attstorage))));

                def->inhcount++;
                /* Merge of NOT NULL constraints = OR 'em together */
                def->is_not_null |= attribute->attnotnull;
                /* Default and other constraints are handled below */
                newattno[parent_attno - 1] = exist_attno;
            }
            else
            {
                /*
                 * No, create a new inherited column
                 */
                def = makeNode(ColumnDef);
                def->colname = pstrdup(attributeName);
                def->typeName = makeTypeNameFromOid(attribute->atttypid,
                                                    attribute->atttypmod);
                def->inhcount = 1;
                def->is_local = false;
                def->is_not_null = attribute->attnotnull;
                def->is_from_type = false;
                def->is_from_parent = true;
                def->storage = attribute->attstorage;
                def->raw_default = NULL;
                def->cooked_default = NULL;
                def->collClause = NULL;
                def->collOid = attribute->attcollation;
                def->constraints = NIL;
                def->location = -1;
                inhSchema = lappend(inhSchema, def);
                newattno[parent_attno - 1] = ++child_attno;
            }

            /*
             * Copy default if any
             */
            if (attribute->atthasdef)
            {
                Node       *this_default = NULL;
                AttrDefault *attrdef;
                int         i;

                /* Find default in constraint structure */
                Assert(constr != NULL);
                attrdef = constr->defval;
                for (i = 0; i < constr->num_defval; i++)
                {
                    if (attrdef[i].adnum == parent_attno)
                    {
                        this_default = stringToNode(attrdef[i].adbin);
                        break;
                    }
                }
                Assert(this_default != NULL);

                /*
                 * If default expr could contain any vars, we'd need to fix
                 * 'em, but it can't; so default is ready to apply to child.
                 *
                 * If we already had a default from some prior parent, check
                 * to see if they are the same.  If so, no problem; if not,
                 * mark the column as having a bogus default. Below, we will
                 * complain if the bogus default isn't overridden by the child
                 * schema.
                 */
                Assert(def->raw_default == NULL);
                if (def->cooked_default == NULL)
                    def->cooked_default = this_default;
                else if (!equal(def->cooked_default, this_default))
                {
                    def->cooked_default = &bogus_marker;
                    have_bogus_defaults = true;
                }
            }
        }

        /*
         * Now copy the CHECK constraints of this parent, adjusting attnos
         * using the completed newattno[] map.  Identically named constraints
         * are merged if possible, else we throw error.
         */
        if (constr && constr->num_check > 0)
        {
            ConstrCheck *check = constr->check;
            int         i;

            for (i = 0; i < constr->num_check; i++)
            {
                char       *name = check[i].ccname;
                Node       *expr;
                bool        found_whole_row;

                /* ignore if the constraint is non-inheritable */
                if (check[i].ccnoinherit)
                    continue;

                /* Adjust Vars to match new table's column numbering */
                expr = map_variable_attnos(stringToNode(check[i].ccbin),
                                           1, 0,
                                           newattno, tupleDesc->natts,
                                           InvalidOid, &found_whole_row);

                /*
                 * For the moment we have to reject whole-row variables. We
                 * could convert them, if we knew the new table's rowtype OID,
                 * but that hasn't been assigned yet.
                 */
                if (found_whole_row)
                    ereport(ERROR,
                            (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                             errmsg("cannot convert whole-row table reference"),
                             errdetail("Constraint \"%s\" contains a whole-row reference to table \"%s\".",
                                       name,
                                       RelationGetRelationName(relation))));

                /* check for duplicate */
                if (!MergeCheckConstraint(constraints, name, expr))
                {
                    /* nope, this is a new one */
                    CookedConstraint *cooked;

                    cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
                    cooked->contype = CONSTR_CHECK;
                    cooked->conoid = InvalidOid;    /* until created */
                    cooked->name = pstrdup(name);
                    cooked->attnum = 0; /* not used for constraints */
                    cooked->expr = expr;
                    cooked->skip_validation = false;
                    cooked->is_local = false;
                    cooked->inhcount = 1;
                    cooked->is_no_inherit = false;
                    constraints = lappend(constraints, cooked);
                }
            }
        }

        pfree(newattno);

        /*
         * Close the parent rel, but keep our lock on it until xact commit.
         * That will prevent someone else from deleting or ALTERing the parent
         * before the child is committed.
         */
        heap_close(relation, NoLock);
    }

    /*
     * If we had no inherited attributes, the result schema is just the
     * explicitly declared columns.  Otherwise, we need to merge the declared
     * columns into the inherited schema list.  Although, we never have any
     * explicitly declared columns if the table is a partition.
     */
    if (inhSchema != NIL)
    {
        int         schema_attno = 0;

        foreach(entry, schema)
        {
            ColumnDef  *newdef = lfirst(entry);
            char       *attributeName = newdef->colname;
            int         exist_attno;

            schema_attno++;

            /*
             * Does it conflict with some previously inherited column?
             */
            exist_attno = findAttrByName(attributeName, inhSchema);
            if (exist_attno > 0)
            {
                ColumnDef  *def;
                Oid         defTypeId,
                            newTypeId;
                int32       deftypmod,
                            newtypmod;
                Oid         defcollid,
                            newcollid;

                /*
                 * Partitions have only one parent and have no column
                 * definitions of their own, so conflict should never occur.
                 */
                Assert(!is_partition);

                /*
                 * Yes, try to merge the two column definitions. They must
                 * have the same type, typmod, and collation.
                 */
                if (exist_attno == schema_attno)
                    ereport(NOTICE,
                            (errmsg("merging column \"%s\" with inherited definition",
                                    attributeName)));
                else
                    ereport(NOTICE,
                            (errmsg("moving and merging column \"%s\" with inherited definition", attributeName),
                             errdetail("User-specified column moved to the position of the inherited column.")));
                def = (ColumnDef *) list_nth(inhSchema, exist_attno - 1);
                typenameTypeIdAndMod(NULL, def->typeName, &defTypeId, &deftypmod);
                typenameTypeIdAndMod(NULL, newdef->typeName, &newTypeId, &newtypmod);
                if (defTypeId != newTypeId || deftypmod != newtypmod)
                    ereport(ERROR,
                            (errcode(ERRCODE_DATATYPE_MISMATCH),
                             errmsg("column \"%s\" has a type conflict",
                                    attributeName),
                             errdetail("%s versus %s",
                                       format_type_with_typemod(defTypeId,
                                                                deftypmod),
                                       format_type_with_typemod(newTypeId,
                                                                newtypmod))));
                defcollid = GetColumnDefCollation(NULL, def, defTypeId);
                newcollid = GetColumnDefCollation(NULL, newdef, newTypeId);
                if (defcollid != newcollid)
                    ereport(ERROR,
                            (errcode(ERRCODE_COLLATION_MISMATCH),
                             errmsg("column \"%s\" has a collation conflict",
                                    attributeName),
                             errdetail("\"%s\" versus \"%s\"",
                                       get_collation_name(defcollid),
                                       get_collation_name(newcollid))));

                /*
                 * Identity is never inherited.  The new column can have an
                 * identity definition, so we always just take that one.
                 */
                def->identity = newdef->identity;

                /* Copy storage parameter */
                if (def->storage == 0)
                    def->storage = newdef->storage;
                else if (newdef->storage != 0 && def->storage != newdef->storage)
                    ereport(ERROR,
                            (errcode(ERRCODE_DATATYPE_MISMATCH),
                             errmsg("column \"%s\" has a storage parameter conflict",
                                    attributeName),
                             errdetail("%s versus %s",
                                       storage_name(def->storage),
                                       storage_name(newdef->storage))));

                /* Mark the column as locally defined */
                def->is_local = true;
                /* Merge of NOT NULL constraints = OR 'em together */
                def->is_not_null |= newdef->is_not_null;
                /* If new def has a default, override previous default */
                if (newdef->raw_default != NULL)
                {
                    def->raw_default = newdef->raw_default;
                    def->cooked_default = newdef->cooked_default;
                }
            }
            else
            {
                /*
                 * No, attach new column to result schema
                 */
                inhSchema = lappend(inhSchema, newdef);
            }
        }

        schema = inhSchema;

        /*
         * Check that we haven't exceeded the legal # of columns after merging
         * in inherited columns.
         */
        if (list_length(schema) > MaxHeapAttributeNumber)
            ereport(ERROR,
                    (errcode(ERRCODE_TOO_MANY_COLUMNS),
                     errmsg("tables can have at most %d columns",
                            MaxHeapAttributeNumber)));
    }

    /*
     * Now that we have the column definition list for a partition, we can
     * check whether the columns referenced in the column constraint specs
     * actually exist.  Also, we merge the constraints into the corresponding
     * column definitions.
     */
    if (is_partition && list_length(saved_schema) > 0)
    {
        schema = list_concat(schema, saved_schema);

        foreach(entry, schema)
        {
            ColumnDef  *coldef = lfirst(entry);
            ListCell   *rest = lnext(entry);
            ListCell   *prev = entry;

            /*
             * Partition column option that does not belong to a column from
             * the parent.  This works because the columns from the parent
             * come first in the list (see above).
             */
            if (coldef->typeName == NULL)
                ereport(ERROR,
                        (errcode(ERRCODE_UNDEFINED_COLUMN),
                         errmsg("column \"%s\" does not exist",
                                coldef->colname)));
            while (rest != NULL)
            {
                ColumnDef  *restdef = lfirst(rest);
                ListCell   *next = lnext(rest); /* need to save it in case we
                                                 * delete it */

                if (strcmp(coldef->colname, restdef->colname) == 0)
                {
                    /*
                     * merge the column options into the column from the
                     * parent
                     */
                    if (coldef->is_from_parent)
                    {
                        coldef->is_not_null = restdef->is_not_null;
                        coldef->raw_default = restdef->raw_default;
                        coldef->cooked_default = restdef->cooked_default;
                        coldef->constraints = restdef->constraints;
                        coldef->is_from_parent = false;
                        list_delete_cell(schema, rest, prev);
                    }
                    else
                        ereport(ERROR,
                                (errcode(ERRCODE_DUPLICATE_COLUMN),
                                 errmsg("column \"%s\" specified more than once",
                                        coldef->colname)));
                }
                prev = rest;
                rest = next;
            }
        }
    }

    /*
     * If we found any conflicting parent default values, check to make sure
     * they were overridden by the child.
     */
    if (have_bogus_defaults)
    {
        foreach(entry, schema)
        {
            ColumnDef  *def = lfirst(entry);

            if (def->cooked_default == &bogus_marker)
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                         errmsg("column \"%s\" inherits conflicting default values",
                                def->colname),
                         errhint("To resolve the conflict, specify a default explicitly.")));
        }
    }

    *supOids = parentOids;
    *supconstr = constraints;
    *supOidCount = parentsWithOids;
    return schema;
}


/*
 * MergeCheckConstraint
 *      Try to merge an inherited CHECK constraint with previous ones
 *
 * If we inherit identically-named constraints from multiple parents, we must
 * merge them, or throw an error if they don't have identical definitions.
 *
 * constraints is a list of CookedConstraint structs for previous constraints.
 *
 * Returns true if merged (constraint is a duplicate), or false if it's
 * got a so-far-unique name, or throws error if conflict.
 */
static bool
MergeCheckConstraint(List *constraints, char *name, Node *expr)
{
    ListCell   *lc;

    foreach(lc, constraints)
    {
        CookedConstraint *ccon = (CookedConstraint *) lfirst(lc);

        Assert(ccon->contype == CONSTR_CHECK);

        /* Non-matching names never conflict */
        if (strcmp(ccon->name, name) != 0)
            continue;

        if (equal(expr, ccon->expr))
        {
            /* OK to merge */
            ccon->inhcount++;
            return true;
        }

        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("check constraint name \"%s\" appears multiple times but with different expressions",
                        name)));
    }

    return false;
}


/*
 * StoreCatalogInheritance
 *      Updates the system catalogs with proper inheritance information.
 *
 * supers is a list of the OIDs of the new relation's direct ancestors.
 */
static void
StoreCatalogInheritance(Oid relationId, List *supers,
                        bool child_is_partition)
{
    Relation    relation;
    int32       seqNumber;
    ListCell   *entry;

    /*
     * sanity checks
     */
    AssertArg(OidIsValid(relationId));

    if (supers == NIL)
        return;

    /*
     * Store INHERITS information in pg_inherits using direct ancestors only.
     * Also enter dependencies on the direct ancestors, and make sure they are
     * marked with relhassubclass = true.
     *
     * (Once upon a time, both direct and indirect ancestors were found here
     * and then entered into pg_ipl.  Since that catalog doesn't exist
     * anymore, there's no need to look for indirect ancestors.)
     */
    relation = heap_open(InheritsRelationId, RowExclusiveLock);

    seqNumber = 1;
    foreach(entry, supers)
    {
        Oid         parentOid = lfirst_oid(entry);

        StoreCatalogInheritance1(relationId, parentOid, seqNumber, relation,
                                 child_is_partition);
        seqNumber++;
    }

    heap_close(relation, RowExclusiveLock);
}

/*
 * Make catalog entries showing relationId as being an inheritance child
 * of parentOid.  inhRelation is the already-opened pg_inherits catalog.
 */
static void
StoreCatalogInheritance1(Oid relationId, Oid parentOid,
                         int32 seqNumber, Relation inhRelation,
                         bool child_is_partition)
{
    ObjectAddress childobject,
                parentobject;

    /* store the pg_inherits row */
    StoreSingleInheritance(relationId, parentOid, seqNumber);

    /*
     * Store a dependency too
     */
    parentobject.classId = RelationRelationId;
    parentobject.objectId = parentOid;
    parentobject.objectSubId = 0;
    childobject.classId = RelationRelationId;
    childobject.objectId = relationId;
    childobject.objectSubId = 0;

    recordDependencyOn(&childobject, &parentobject,
                       child_dependency_type(child_is_partition));

    /*
     * Post creation hook of this inheritance. Since object_access_hook
     * doesn't take multiple object identifiers, we relay oid of parent
     * relation using auxiliary_id argument.
     */
    InvokeObjectPostAlterHookArg(InheritsRelationId,
                                 relationId, 0,
                                 parentOid, false);

    /*
     * Mark the parent as having subclasses.
     */
    SetRelationHasSubclass(parentOid, true);
}

/*
 * Look for an existing schema entry with the given name.
 *
 * Returns the index (starting with 1) if attribute already exists in schema,
 * 0 if it doesn't.
 */
static int
findAttrByName(const char *attributeName, List *schema)
{
    ListCell   *s;
    int         i = 1;

    foreach(s, schema)
    {
        ColumnDef  *def = lfirst(s);

        if (strcmp(attributeName, def->colname) == 0)
            return i;

        i++;
    }
    return 0;
}


/*
 * SetRelationHasSubclass
 *      Set the value of the relation's relhassubclass field in pg_class.
 *
 * NOTE: caller must be holding an appropriate lock on the relation.
 * ShareUpdateExclusiveLock is sufficient.
 *
 * NOTE: an important side-effect of this operation is that an SI invalidation
 * message is sent out to all backends --- including me --- causing plans
 * referencing the relation to be rebuilt with the new list of children.
 * This must happen even if we find that no change is needed in the pg_class
 * row.
 */
void
SetRelationHasSubclass(Oid relationId, bool relhassubclass)
{
    Relation    relationRelation;
    HeapTuple   tuple;
    Form_pg_class classtuple;

    /*
     * Fetch a modifiable copy of the tuple, modify it, update pg_class.
     */
    relationRelation = heap_open(RelationRelationId, RowExclusiveLock);
    tuple = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(relationId));
    if (!HeapTupleIsValid(tuple))
        elog(ERROR, "cache lookup failed for relation %u", relationId);
    classtuple = (Form_pg_class) GETSTRUCT(tuple);

    if (classtuple->relhassubclass != relhassubclass)
    {
        classtuple->relhassubclass = relhassubclass;
        CatalogTupleUpdate(relationRelation, &tuple->t_self, tuple);
    }
    else
    {
        /* no need to change tuple, but force relcache rebuild anyway */
        CacheInvalidateRelcacheByTuple(tuple);
    }

    heap_freetuple(tuple);
    heap_close(relationRelation, RowExclusiveLock);
}

/*
 *      renameatt_check         - basic sanity checks before attribute rename
 */
static void
renameatt_check(Oid myrelid, Form_pg_class classform, bool recursing)
{
    char        relkind = classform->relkind;

    if (classform->reloftype && !recursing)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("cannot rename column of typed table")));

    /*
     * Renaming the columns of sequences or toast tables doesn't actually
     * break anything from the system's point of view, since internal
     * references are by attnum.  But it doesn't seem right to allow users to
     * change names that are hardcoded into the system, hence the following
     * restriction.
     */
    if (relkind != RELKIND_RELATION &&
        relkind != RELKIND_VIEW &&
        relkind != RELKIND_MATVIEW &&
        relkind != RELKIND_COMPOSITE_TYPE &&
        relkind != RELKIND_INDEX &&
        relkind != RELKIND_PARTITIONED_INDEX &&
        relkind != RELKIND_FOREIGN_TABLE &&
        relkind != RELKIND_PARTITIONED_TABLE)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("\"%s\" is not a table, view, materialized view, composite type, index, or foreign table",
                        NameStr(classform->relname))));

    /*
     * permissions checking.  only the owner of a class can change its schema.
     */
    if (!pg_class_ownercheck(myrelid, GetUserId()))
        aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(get_rel_relkind(myrelid)),
                       NameStr(classform->relname));
    if (!allowSystemTableMods && IsSystemClass(myrelid, classform))
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("permission denied: \"%s\" is a system catalog",
                        NameStr(classform->relname))));
}

/*
 *      renameatt_internal      - workhorse for renameatt
 *
 * Return value is the attribute number in the 'myrelid' relation.
 */
static AttrNumber
renameatt_internal(Oid myrelid,
                   const char *oldattname,
                   const char *newattname,
                   bool recurse,
                   bool recursing,
                   int expected_parents,
                   DropBehavior behavior)
{
    Relation    targetrelation;
    Relation    attrelation;
    HeapTuple   atttup;
    Form_pg_attribute attform;
    AttrNumber  attnum;

    /*
     * Grab an exclusive lock on the target table, which we will NOT release
     * until end of transaction.
     */
    targetrelation = relation_open(myrelid, AccessExclusiveLock);
    renameatt_check(myrelid, RelationGetForm(targetrelation), recursing);

    /*
     * if the 'recurse' flag is set then we are supposed to rename this
     * attribute in all classes that inherit from 'relname' (as well as in
     * 'relname').
     *
     * any permissions or problems with duplicate attributes will cause the
     * whole transaction to abort, which is what we want -- all or nothing.
     */
    if (recurse)
    {
        List       *child_oids,
                   *child_numparents;
        ListCell   *lo,
                   *li;

        /*
         * we need the number of parents for each child so that the recursive
         * calls to renameatt() can determine whether there are any parents
         * outside the inheritance hierarchy being processed.
         */
        child_oids = find_all_inheritors(myrelid, AccessExclusiveLock,
                                         &child_numparents);

        /*
         * find_all_inheritors does the recursive search of the inheritance
         * hierarchy, so all we have to do is process all of the relids in the
         * list that it returns.
         */
        forboth(lo, child_oids, li, child_numparents)
        {
            Oid         childrelid = lfirst_oid(lo);
            int         numparents = lfirst_int(li);

            if (childrelid == myrelid)
                continue;
            /* note we need not recurse again */
            renameatt_internal(childrelid, oldattname, newattname, false, true, numparents, behavior);
        }
    }
    else
    {
        /*
         * If we are told not to recurse, there had better not be any child
         * tables; else the rename would put them out of step.
         *
         * expected_parents will only be 0 if we are not already recursing.
         */
        if (expected_parents == 0 &&
            find_inheritance_children(myrelid, NoLock) != NIL)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                     errmsg("inherited column \"%s\" must be renamed in child tables too",
                            oldattname)));
    }

    /* rename attributes in typed tables of composite type */
    if (targetrelation->rd_rel->relkind == RELKIND_COMPOSITE_TYPE)
    {
        List       *child_oids;
        ListCell   *lo;

        child_oids = find_typed_table_dependencies(targetrelation->rd_rel->reltype,
                                                   RelationGetRelationName(targetrelation),
                                                   behavior);

        foreach(lo, child_oids)
            renameatt_internal(lfirst_oid(lo), oldattname, newattname, true, true, 0, behavior);
    }

    attrelation = heap_open(AttributeRelationId, RowExclusiveLock);

    atttup = SearchSysCacheCopyAttName(myrelid, oldattname);
    if (!HeapTupleIsValid(atttup))
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_COLUMN),
                 errmsg("column \"%s\" does not exist",
                        oldattname)));
    attform = (Form_pg_attribute) GETSTRUCT(atttup);

    attnum = attform->attnum;
    if (attnum <= 0)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot rename system column \"%s\"",
                        oldattname)));

    /*
     * if the attribute is inherited, forbid the renaming.  if this is a
     * top-level call to renameatt(), then expected_parents will be 0, so the
     * effect of this code will be to prohibit the renaming if the attribute
     * is inherited at all.  if this is a recursive call to renameatt(),
     * expected_parents will be the number of parents the current relation has
     * within the inheritance hierarchy being processed, so we'll prohibit the
     * renaming only if there are additional parents from elsewhere.
     */
    if (attform->attinhcount > expected_parents)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                 errmsg("cannot rename inherited column \"%s\"",
                        oldattname)));

    /* new name should not already exist */
    (void) check_for_column_name_collision(targetrelation, newattname, false);

    /* apply the update */
    namestrcpy(&(attform->attname), newattname);

    CatalogTupleUpdate(attrelation, &atttup->t_self, atttup);

    InvokeObjectPostAlterHook(RelationRelationId, myrelid, attnum);

    heap_freetuple(atttup);

    heap_close(attrelation, RowExclusiveLock);

    relation_close(targetrelation, NoLock); /* close rel but keep lock */

    return attnum;
}

/*
 * Perform permissions and integrity checks before acquiring a relation lock.
 */
static void
RangeVarCallbackForRenameAttribute(const RangeVar *rv, Oid relid, Oid oldrelid,
                                   void *arg)
{
    HeapTuple   tuple;
    Form_pg_class form;

    tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
    if (!HeapTupleIsValid(tuple))
        return;                 /* concurrently dropped */
    form = (Form_pg_class) GETSTRUCT(tuple);
    renameatt_check(relid, form, false);
    ReleaseSysCache(tuple);
}

/*
 *      renameatt       - changes the name of an attribute in a relation
 *
 * The returned ObjectAddress is that of the renamed column.
 */
ObjectAddress
renameatt(RenameStmt *stmt)
{
    Oid         relid;
    AttrNumber  attnum;
    ObjectAddress address;

    /* lock level taken here should match renameatt_internal */
    relid = RangeVarGetRelidExtended(stmt->relation, AccessExclusiveLock,
                                     stmt->missing_ok ? RVR_MISSING_OK : 0,
                                     RangeVarCallbackForRenameAttribute,
                                     NULL);

    if (!OidIsValid(relid))
    {
        ereport(NOTICE,
                (errmsg("relation \"%s\" does not exist, skipping",
                        stmt->relation->relname)));
        return InvalidObjectAddress;
    }

    attnum =
        renameatt_internal(relid,
                           stmt->subname,   /* old att name */
                           stmt->newname,   /* new att name */
                           stmt->relation->inh, /* recursive? */
                           false,   /* recursing? */
                           0,   /* expected inhcount */
                           stmt->behavior);

    ObjectAddressSubSet(address, RelationRelationId, relid, attnum);

    return address;
}

/*
 * same logic as renameatt_internal
 */
static ObjectAddress
rename_constraint_internal(Oid myrelid,
                           Oid mytypid,
                           const char *oldconname,
                           const char *newconname,
                           bool recurse,
                           bool recursing,
                           int expected_parents)
{
    Relation    targetrelation = NULL;
    Oid         constraintOid;
    HeapTuple   tuple;
    Form_pg_constraint con;
    ObjectAddress address;

    AssertArg(!myrelid || !mytypid);

    if (mytypid)
    {
        constraintOid = get_domain_constraint_oid(mytypid, oldconname, false);
    }
    else
    {
        targetrelation = relation_open(myrelid, AccessExclusiveLock);

        /*
         * don't tell it whether we're recursing; we allow changing typed
         * tables here
         */
        renameatt_check(myrelid, RelationGetForm(targetrelation), false);

        constraintOid = get_relation_constraint_oid(myrelid, oldconname, false);
    }

    tuple = SearchSysCache1(CONSTROID, ObjectIdGetDatum(constraintOid));
    if (!HeapTupleIsValid(tuple))
        elog(ERROR, "cache lookup failed for constraint %u",
             constraintOid);
    con = (Form_pg_constraint) GETSTRUCT(tuple);

    if (myrelid && con->contype == CONSTRAINT_CHECK && !con->connoinherit)
    {
        if (recurse)
        {
            List       *child_oids,
                       *child_numparents;
            ListCell   *lo,
                       *li;

            child_oids = find_all_inheritors(myrelid, AccessExclusiveLock,
                                             &child_numparents);

            forboth(lo, child_oids, li, child_numparents)
            {
                Oid         childrelid = lfirst_oid(lo);
                int         numparents = lfirst_int(li);

                if (childrelid == myrelid)
                    continue;

                rename_constraint_internal(childrelid, InvalidOid, oldconname, newconname, false, true, numparents);
            }
        }
        else
        {
            if (expected_parents == 0 &&
                find_inheritance_children(myrelid, NoLock) != NIL)
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                         errmsg("inherited constraint \"%s\" must be renamed in child tables too",
                                oldconname)));
        }

        if (con->coninhcount > expected_parents)
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                     errmsg("cannot rename inherited constraint \"%s\"",
                            oldconname)));
    }

    if (con->conindid
        && (con->contype == CONSTRAINT_PRIMARY
            || con->contype == CONSTRAINT_UNIQUE
            || con->contype == CONSTRAINT_EXCLUSION))
        /* rename the index; this renames the constraint as well */
        RenameRelationInternal(con->conindid, newconname, false);
    else
        RenameConstraintById(constraintOid, newconname);

    ObjectAddressSet(address, ConstraintRelationId, constraintOid);

    ReleaseSysCache(tuple);

    if (targetrelation)
        relation_close(targetrelation, NoLock); /* close rel but keep lock */

    return address;
}

ObjectAddress
RenameConstraint(RenameStmt *stmt)
{
    Oid         relid = InvalidOid;
    Oid         typid = InvalidOid;

    if (stmt->renameType == OBJECT_DOMCONSTRAINT)
    {
        Relation    rel;
        HeapTuple   tup;

        typid = typenameTypeId(NULL, makeTypeNameFromNameList(castNode(List, stmt->object)));
        rel = heap_open(TypeRelationId, RowExclusiveLock);
        tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typid));
        if (!HeapTupleIsValid(tup))
            elog(ERROR, "cache lookup failed for type %u", typid);
        checkDomainOwner(tup);
        ReleaseSysCache(tup);
        heap_close(rel, NoLock);
    }
    else
    {
        /* lock level taken here should match rename_constraint_internal */
        relid = RangeVarGetRelidExtended(stmt->relation, AccessExclusiveLock,
                                         stmt->missing_ok ? RVR_MISSING_OK : 0,
                                         RangeVarCallbackForRenameAttribute,
                                         NULL);
        if (!OidIsValid(relid))
        {
            ereport(NOTICE,
                    (errmsg("relation \"%s\" does not exist, skipping",
                            stmt->relation->relname)));
            return InvalidObjectAddress;
        }
    }

    return
        rename_constraint_internal(relid, typid,
                                   stmt->subname,
                                   stmt->newname,
                                   (stmt->relation &&
                                    stmt->relation->inh),   /* recursive? */
                                   false,   /* recursing? */
                                   0 /* expected inhcount */ );

}

/*
 * Execute ALTER TABLE/INDEX/SEQUENCE/VIEW/MATERIALIZED VIEW/FOREIGN TABLE
 * RENAME
 */
ObjectAddress
RenameRelation(RenameStmt *stmt)
{
    Oid         relid;
    ObjectAddress address;

    /*
     * Grab an exclusive lock on the target table, index, sequence, view,
     * materialized view, or foreign table, which we will NOT release until
     * end of transaction.
     *
     * Lock level used here should match RenameRelationInternal, to avoid lock
     * escalation.
     */
    relid = RangeVarGetRelidExtended(stmt->relation, AccessExclusiveLock,
                                     stmt->missing_ok ? RVR_MISSING_OK : 0,
                                     RangeVarCallbackForAlterRelation,
                                     (void *) stmt);

    if (!OidIsValid(relid))
    {
        ereport(NOTICE,
                (errmsg("relation \"%s\" does not exist, skipping",
                        stmt->relation->relname)));
        return InvalidObjectAddress;
    }

    /* Do the work */
    RenameRelationInternal(relid, stmt->newname, false);

    ObjectAddressSet(address, RelationRelationId, relid);

    return address;
}

/*
 *      RenameRelationInternal - change the name of a relation
 *
 *      XXX - When renaming sequences, we don't bother to modify the
 *            sequence name that is stored within the sequence itself
 *            (this would cause problems with MVCC). In the future,
 *            the sequence name should probably be removed from the
 *            sequence, AFAIK there's no need for it to be there.
 */
void
RenameRelationInternal(Oid myrelid, const char *newrelname, bool is_internal)
{
    Relation    targetrelation;
    Relation    relrelation;    /* for RELATION relation */
    HeapTuple   reltup;
    Form_pg_class relform;
    Oid         namespaceId;

    /*
     * Grab an exclusive lock on the target table, index, sequence, view,
     * materialized view, or foreign table, which we will NOT release until
     * end of transaction.
     */
    targetrelation = relation_open(myrelid, AccessExclusiveLock);
    namespaceId = RelationGetNamespace(targetrelation);

    /*
     * Find relation's pg_class tuple, and make sure newrelname isn't in use.
     */
    relrelation = heap_open(RelationRelationId, RowExclusiveLock);

    reltup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(myrelid));
    if (!HeapTupleIsValid(reltup))  /* shouldn't happen */
        elog(ERROR, "cache lookup failed for relation %u", myrelid);
    relform = (Form_pg_class) GETSTRUCT(reltup);

    if (get_relname_relid(newrelname, namespaceId) != InvalidOid)
        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_TABLE),
                 errmsg("relation \"%s\" already exists",
                        newrelname)));

    /*
     * Update pg_class tuple with new relname.  (Scribbling on reltup is OK
     * because it's a copy...)
     */
    namestrcpy(&(relform->relname), newrelname);

    CatalogTupleUpdate(relrelation, &reltup->t_self, reltup);

    InvokeObjectPostAlterHookArg(RelationRelationId, myrelid, 0,
                                 InvalidOid, is_internal);

    heap_freetuple(reltup);
    heap_close(relrelation, RowExclusiveLock);

    /*
     * Also rename the associated type, if any.
     */
    if (OidIsValid(targetrelation->rd_rel->reltype))
        RenameTypeInternal(targetrelation->rd_rel->reltype,
                           newrelname, namespaceId);

    /*
     * Also rename the associated constraint, if any.
     */
    if (targetrelation->rd_rel->relkind == RELKIND_INDEX ||
        targetrelation->rd_rel->relkind == RELKIND_PARTITIONED_INDEX)
    {
        Oid         constraintId = get_index_constraint(myrelid);

        if (OidIsValid(constraintId))
            RenameConstraintById(constraintId, newrelname);
    }

    /*
     * Close rel, but keep exclusive lock!
     */
    relation_close(targetrelation, NoLock);
}

/*
 * Disallow ALTER TABLE (and similar commands) when the current backend has
 * any open reference to the target table besides the one just acquired by
 * the calling command; this implies there's an open cursor or active plan.
 * We need this check because our lock doesn't protect us against stomping
 * on our own foot, only other people's feet!
 *
 * For ALTER TABLE, the only case known to cause serious trouble is ALTER
 * COLUMN TYPE, and some changes are obviously pretty benign, so this could
 * possibly be relaxed to only error out for certain types of alterations.
 * But the use-case for allowing any of these things is not obvious, so we
 * won't work hard at it for now.
 *
 * We also reject these commands if there are any pending AFTER trigger events
 * for the rel.  This is certainly necessary for the rewriting variants of
 * ALTER TABLE, because they don't preserve tuple TIDs and so the pending
 * events would try to fetch the wrong tuples.  It might be overly cautious
 * in other cases, but again it seems better to err on the side of paranoia.
 *
 * REINDEX calls this with "rel" referencing the index to be rebuilt; here
 * we are worried about active indexscans on the index.  The trigger-event
 * check can be skipped, since we are doing no damage to the parent table.
 *
 * The statement name (eg, "ALTER TABLE") is passed for use in error messages.
 */
void
CheckTableNotInUse(Relation rel, const char *stmt)
{
    int         expected_refcnt;

    expected_refcnt = rel->rd_isnailed ? 2 : 1;
    if (rel->rd_refcnt != expected_refcnt)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_IN_USE),
        /* translator: first %s is a SQL command, eg ALTER TABLE */
                 errmsg("cannot %s \"%s\" because "
                        "it is being used by active queries in this session",
                        stmt, RelationGetRelationName(rel))));

    if (rel->rd_rel->relkind != RELKIND_INDEX &&
        rel->rd_rel->relkind != RELKIND_PARTITIONED_INDEX &&
        AfterTriggerPendingOnRel(RelationGetRelid(rel)))
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_IN_USE),
        /* translator: first %s is a SQL command, eg ALTER TABLE */
                 errmsg("cannot %s \"%s\" because "
                        "it has pending trigger events",
                        stmt, RelationGetRelationName(rel))));
}

/*
 * AlterTableLookupRelation
 *      Look up, and lock, the OID for the relation named by an alter table
 *      statement.
 */
Oid
AlterTableLookupRelation(AlterTableStmt *stmt, LOCKMODE lockmode)
{
    return RangeVarGetRelidExtended(stmt->relation, lockmode,
                                    stmt->missing_ok ? RVR_MISSING_OK : 0,
                                    RangeVarCallbackForAlterRelation,
                                    (void *) stmt);
}

/*
 * AlterTable
 *      Execute ALTER TABLE, which can be a list of subcommands
 *
 * ALTER TABLE is performed in three phases:
 *      1. Examine subcommands and perform pre-transformation checking.
 *      2. Update system catalogs.
 *      3. Scan table(s) to check new constraints, and optionally recopy
 *         the data into new table(s).
 * Phase 3 is not performed unless one or more of the subcommands requires
 * it.  The intention of this design is to allow multiple independent
 * updates of the table schema to be performed with only one pass over the
 * data.
 *
 * ATPrepCmd performs phase 1.  A "work queue" entry is created for
 * each table to be affected (there may be multiple affected tables if the
 * commands traverse a table inheritance hierarchy).  Also we do preliminary
 * validation of the subcommands, including parse transformation of those
 * expressions that need to be evaluated with respect to the old table
 * schema.
 *
 * ATRewriteCatalogs performs phase 2 for each affected table.  (Note that
 * phases 2 and 3 normally do no explicit recursion, since phase 1 already
 * did it --- although some subcommands have to recurse in phase 2 instead.)
 * Certain subcommands need to be performed before others to avoid
 * unnecessary conflicts; for example, DROP COLUMN should come before
 * ADD COLUMN.  Therefore phase 1 divides the subcommands into multiple
 * lists, one for each logical "pass" of phase 2.
 *
 * ATRewriteTables performs phase 3 for those tables that need it.
 *
 * Thanks to the magic of MVCC, an error anywhere along the way rolls back
 * the whole operation; we don't have to do anything special to clean up.
 *
 * The caller must lock the relation, with an appropriate lock level
 * for the subcommands requested, using AlterTableGetLockLevel(stmt->cmds)
 * or higher. We pass the lock level down
 * so that we can apply it recursively to inherited tables. Note that the
 * lock level we want as we recurse might well be higher than required for
 * that specific subcommand. So we pass down the overall lock requirement,
 * rather than reassess it at lower levels.
 */
void
AlterTable(Oid relid, LOCKMODE lockmode, AlterTableStmt *stmt)
{
    Relation    rel;

    /* Caller is required to provide an adequate lock. */
    rel = relation_open(relid, NoLock);

    CheckTableNotInUse(rel, "ALTER TABLE");

    ATController(stmt, rel, stmt->cmds, stmt->relation->inh, lockmode);
}

/*
 * AlterTableInternal
 *
 * ALTER TABLE with target specified by OID
 *
 * We do not reject if the relation is already open, because it's quite
 * likely that one or more layers of caller have it open.  That means it
 * is unsafe to use this entry point for alterations that could break
 * existing query plans.  On the assumption it's not used for such, we
 * don't have to reject pending AFTER triggers, either.
 */
void
AlterTableInternal(Oid relid, List *cmds, bool recurse)
{
    Relation    rel;
    LOCKMODE    lockmode = AlterTableGetLockLevel(cmds);

    rel = relation_open(relid, lockmode);

    EventTriggerAlterTableRelid(relid);

    ATController(NULL, rel, cmds, recurse, lockmode);
}

/*
 * AlterTableGetLockLevel
 *
 * Sets the overall lock level required for the supplied list of subcommands.
 * Policy for doing this set according to needs of AlterTable(), see
 * comments there for overall explanation.
 *
 * Function is called before and after parsing, so it must give same
 * answer each time it is called. Some subcommands are transformed
 * into other subcommand types, so the transform must never be made to a
 * lower lock level than previously assigned. All transforms are noted below.
 *
 * Since this is called before we lock the table we cannot use table metadata
 * to influence the type of lock we acquire.
 *
 * There should be no lockmodes hardcoded into the subcommand functions. All
 * lockmode decisions for ALTER TABLE are made here only. The one exception is
 * ALTER TABLE RENAME which is treated as a different statement type T_RenameStmt
 * and does not travel through this section of code and cannot be combined with
 * any of the subcommands given here.
 *
 * Note that Hot Standby only knows about AccessExclusiveLocks on the master
 * so any changes that might affect SELECTs running on standbys need to use
 * AccessExclusiveLocks even if you think a lesser lock would do, unless you
 * have a solution for that also.
 *
 * Also note that pg_dump uses only an AccessShareLock, meaning that anything
 * that takes a lock less than AccessExclusiveLock can change object definitions
 * while pg_dump is running. Be careful to check that the appropriate data is
 * derived by pg_dump using an MVCC snapshot, rather than syscache lookups,
 * otherwise we might end up with an inconsistent dump that can't restore.
 */
LOCKMODE
AlterTableGetLockLevel(List *cmds)
{
    /*
     * This only works if we read catalog tables using MVCC snapshots.
     */
    ListCell   *lcmd;
    LOCKMODE    lockmode = ShareUpdateExclusiveLock;

    foreach(lcmd, cmds)
    {
        AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lcmd);
        LOCKMODE    cmd_lockmode = AccessExclusiveLock; /* default for compiler */

        switch (cmd->subtype)
        {
                /*
                 * These subcommands rewrite the heap, so require full locks.
                 */
            case AT_AddColumn:  /* may rewrite heap, in some cases and visible
                                 * to SELECT */
            case AT_SetTableSpace:  /* must rewrite heap */
            case AT_AlterColumnType:    /* must rewrite heap */
            case AT_AddOids:    /* must rewrite heap */
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * These subcommands may require addition of toast tables. If
                 * we add a toast table to a table currently being scanned, we
                 * might miss data added to the new toast table by concurrent
                 * insert transactions.
                 */
            case AT_SetStorage: /* may add toast tables, see
                                 * ATRewriteCatalogs() */
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * Removing constraints can affect SELECTs that have been
                 * optimised assuming the constraint holds true.
                 */
            case AT_DropConstraint: /* as DROP INDEX */
            case AT_DropNotNull:    /* may change some SQL plans */
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * Subcommands that may be visible to concurrent SELECTs
                 */
            case AT_DropColumn: /* change visible to SELECT */
            case AT_AddColumnToView:    /* CREATE VIEW */
            case AT_DropOids:   /* calls AT_DropColumn */
            case AT_EnableAlwaysRule:   /* may change SELECT rules */
            case AT_EnableReplicaRule:  /* may change SELECT rules */
            case AT_EnableRule: /* may change SELECT rules */
            case AT_DisableRule:    /* may change SELECT rules */
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * Changing owner may remove implicit SELECT privileges
                 */
            case AT_ChangeOwner:    /* change visible to SELECT */
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * Changing foreign table options may affect optimization.
                 */
            case AT_GenericOptions:
            case AT_AlterColumnGenericOptions:
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * These subcommands affect write operations only.
                 */
            case AT_EnableTrig:
            case AT_EnableAlwaysTrig:
            case AT_EnableReplicaTrig:
            case AT_EnableTrigAll:
            case AT_EnableTrigUser:
            case AT_DisableTrig:
            case AT_DisableTrigAll:
            case AT_DisableTrigUser:
                cmd_lockmode = ShareRowExclusiveLock;
                break;

                /*
                 * These subcommands affect write operations only. XXX
                 * Theoretically, these could be ShareRowExclusiveLock.
                 */
            case AT_ColumnDefault:
            case AT_AlterConstraint:
            case AT_AddIndex:   /* from ADD CONSTRAINT */
            case AT_AddIndexConstraint:
            case AT_ReplicaIdentity:
            case AT_SetNotNull:
            case AT_EnableRowSecurity:
            case AT_DisableRowSecurity:
            case AT_ForceRowSecurity:
            case AT_NoForceRowSecurity:
            case AT_AddIdentity:
            case AT_DropIdentity:
            case AT_SetIdentity:
                cmd_lockmode = AccessExclusiveLock;
                break;

            case AT_AddConstraint:
            case AT_ProcessedConstraint:    /* becomes AT_AddConstraint */
            case AT_AddConstraintRecurse:   /* becomes AT_AddConstraint */
            case AT_ReAddConstraint:    /* becomes AT_AddConstraint */
            case AT_ReAddDomainConstraint:  /* becomes AT_AddConstraint */
                if (IsA(cmd->def, Constraint))
                {
                    Constraint *con = (Constraint *) cmd->def;

                    switch (con->contype)
                    {
                        case CONSTR_EXCLUSION:
                        case CONSTR_PRIMARY:
                        case CONSTR_UNIQUE:

                            /*
                             * Cases essentially the same as CREATE INDEX. We
                             * could reduce the lock strength to ShareLock if
                             * we can work out how to allow concurrent catalog
                             * updates. XXX Might be set down to
                             * ShareRowExclusiveLock but requires further
                             * analysis.
                             */
                            cmd_lockmode = AccessExclusiveLock;
                            break;
                        case CONSTR_FOREIGN:

                            /*
                             * We add triggers to both tables when we add a
                             * Foreign Key, so the lock level must be at least
                             * as strong as CREATE TRIGGER.
                             */
                            cmd_lockmode = ShareRowExclusiveLock;
                            break;

                        default:
                            cmd_lockmode = AccessExclusiveLock;
                    }
                }
                break;

                /*
                 * These subcommands affect inheritance behaviour. Queries
                 * started before us will continue to see the old inheritance
                 * behaviour, while queries started after we commit will see
                 * new behaviour. No need to prevent reads or writes to the
                 * subtable while we hook it up though. Changing the TupDesc
                 * may be a problem, so keep highest lock.
                 */
            case AT_AddInherit:
            case AT_DropInherit:
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * These subcommands affect implicit row type conversion. They
                 * have affects similar to CREATE/DROP CAST on queries. don't
                 * provide for invalidating parse trees as a result of such
                 * changes, so we keep these at AccessExclusiveLock.
                 */
            case AT_AddOf:
            case AT_DropOf:
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * Only used by CREATE OR REPLACE VIEW which must conflict
                 * with an SELECTs currently using the view.
                 */
            case AT_ReplaceRelOptions:
                cmd_lockmode = AccessExclusiveLock;
                break;

                /*
                 * These subcommands affect general strategies for performance
                 * and maintenance, though don't change the semantic results
                 * from normal data reads and writes. Delaying an ALTER TABLE
                 * behind currently active writes only delays the point where
                 * the new strategy begins to take effect, so there is no
                 * benefit in waiting. In this case the minimum restriction
                 * applies: we don't currently allow concurrent catalog
                 * updates.
                 */
            case AT_SetStatistics:  /* Uses MVCC in getTableAttrs() */
            case AT_ClusterOn:  /* Uses MVCC in getIndexes() */
            case AT_DropCluster:    /* Uses MVCC in getIndexes() */
            case AT_SetOptions: /* Uses MVCC in getTableAttrs() */
            case AT_ResetOptions:   /* Uses MVCC in getTableAttrs() */
                cmd_lockmode = ShareUpdateExclusiveLock;
                break;

            case AT_SetLogged:
            case AT_SetUnLogged:
                cmd_lockmode = AccessExclusiveLock;
                break;

            case AT_ValidateConstraint: /* Uses MVCC in getConstraints() */
                cmd_lockmode = ShareUpdateExclusiveLock;
                break;

                /*
                 * Rel options are more complex than first appears. Options
                 * are set here for tables, views and indexes; for historical
                 * reasons these can all be used with ALTER TABLE, so we can't
                 * decide between them using the basic grammar.
                 */
            case AT_SetRelOptions:  /* Uses MVCC in getIndexes() and
                                     * getTables() */
            case AT_ResetRelOptions:    /* Uses MVCC in getIndexes() and
                                         * getTables() */
                cmd_lockmode = AlterTableGetRelOptionsLockLevel((List *) cmd->def);
                break;

            case AT_AttachPartition:
            case AT_DetachPartition:
                cmd_lockmode = AccessExclusiveLock;
                break;

            default:            /* oops */
                elog(ERROR, "unrecognized alter table type: %d",
                     (int) cmd->subtype);
                break;
        }

        /*
         * Take the greatest lockmode from any subcommand
         */
        if (cmd_lockmode > lockmode)
            lockmode = cmd_lockmode;
    }

    return lockmode;
}

/*
 * ATController provides top level control over the phases.
 *
 * parsetree is passed in to allow it to be passed to event triggers
 * when requested.
 */
static void
ATController(AlterTableStmt *parsetree,
             Relation rel, List *cmds, bool recurse, LOCKMODE lockmode)
{
    List       *wqueue = NIL;
    ListCell   *lcmd;

    /* Phase 1: preliminary examination of commands, create work queue */
    foreach(lcmd, cmds)
    {
        AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lcmd);

        ATPrepCmd(&wqueue, rel, cmd, recurse, false, lockmode);
    }

    /* Close the relation, but keep lock until commit */
    relation_close(rel, NoLock);

    /* Phase 2: update system catalogs */
    ATRewriteCatalogs(&wqueue, lockmode);

    /* Phase 3: scan/rewrite tables as needed */
    ATRewriteTables(parsetree, &wqueue, lockmode);
}

/*
 * ATPrepCmd
 *
 * Traffic cop for ALTER TABLE Phase 1 operations, including simple
 * recursion and permission checks.
 *
 * Caller must have acquired appropriate lock type on relation already.
 * This lock should be held until commit.
 */
static void
ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
          bool recurse, bool recursing, LOCKMODE lockmode)
{
    AlteredTableInfo *tab;
    int         pass = AT_PASS_UNSET;

    /* Find or create work queue entry for this table */
    tab = ATGetQueueEntry(wqueue, rel);

    /*
     * Copy the original subcommand for each table.  This avoids conflicts
     * when different child tables need to make different parse
     * transformations (for example, the same column may have different column
     * numbers in different children).
     */
    cmd = copyObject(cmd);

    /*
     * Do permissions checking, recursion to child tables if needed, and any
     * additional phase-1 processing needed.
     */
    switch (cmd->subtype)
    {
        case AT_AddColumn:      /* ADD COLUMN */
            ATSimplePermissions(rel,
                                ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
            ATPrepAddColumn(wqueue, rel, recurse, recursing, false, cmd,
                            lockmode);
            /* Recursion occurs during execution phase */
            pass = AT_PASS_ADD_COL;
            break;
        case AT_AddColumnToView:    /* add column via CREATE OR REPLACE VIEW */
            ATSimplePermissions(rel, ATT_VIEW);
            ATPrepAddColumn(wqueue, rel, recurse, recursing, true, cmd,
                            lockmode);
            /* Recursion occurs during execution phase */
            pass = AT_PASS_ADD_COL;
            break;
        case AT_ColumnDefault:  /* ALTER COLUMN DEFAULT */

            /*
             * We allow defaults on views so that INSERT into a view can have
             * default-ish behavior.  This works because the rewriter
             * substitutes default values into INSERTs before it expands
             * rules.
             */
            ATSimplePermissions(rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
            /* No command-specific prep needed */
            pass = cmd->def ? AT_PASS_ADD_CONSTR : AT_PASS_DROP;
            break;
        case AT_AddIdentity:
            ATSimplePermissions(rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            pass = AT_PASS_ADD_CONSTR;
            break;
        case AT_DropIdentity:
            ATSimplePermissions(rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            pass = AT_PASS_DROP;
            break;
        case AT_SetIdentity:
            ATSimplePermissions(rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            pass = AT_PASS_COL_ATTRS;
            break;
        case AT_DropNotNull:    /* ALTER COLUMN DROP NOT NULL */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            ATPrepDropNotNull(rel, recurse, recursing);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
            /* No command-specific prep needed */
            pass = AT_PASS_DROP;
            break;
        case AT_SetNotNull:     /* ALTER COLUMN SET NOT NULL */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            ATPrepSetNotNull(rel, recurse, recursing);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
            /* No command-specific prep needed */
            pass = AT_PASS_ADD_CONSTR;
            break;
        case AT_SetStatistics:  /* ALTER COLUMN SET STATISTICS */
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
            /* Performs own permission checks */
            ATPrepSetStatistics(rel, cmd->name, cmd->num, cmd->def, lockmode);
            pass = AT_PASS_MISC;
            break;
        case AT_SetOptions:     /* ALTER COLUMN SET ( options ) */
        case AT_ResetOptions:   /* ALTER COLUMN RESET ( options ) */
            ATSimplePermissions(rel, ATT_TABLE | ATT_MATVIEW | ATT_INDEX | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            pass = AT_PASS_MISC;
            break;
        case AT_SetStorage:     /* ALTER COLUMN SET STORAGE */
            ATSimplePermissions(rel, ATT_TABLE | ATT_MATVIEW | ATT_FOREIGN_TABLE);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_DropColumn:     /* DROP COLUMN */
            ATSimplePermissions(rel,
                                ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
            ATPrepDropColumn(wqueue, rel, recurse, recursing, cmd, lockmode);
            /* Recursion occurs during execution phase */
            pass = AT_PASS_DROP;
            break;
        case AT_AddIndex:       /* ADD INDEX */
            ATSimplePermissions(rel, ATT_TABLE);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_ADD_INDEX;
            break;
        case AT_AddConstraint:  /* ADD CONSTRAINT */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Recursion occurs during execution phase */
            /* No command-specific prep needed except saving recurse flag */
            if (recurse)
                cmd->subtype = AT_AddConstraintRecurse;
            pass = AT_PASS_ADD_CONSTR;
            break;
        case AT_AddIndexConstraint: /* ADD CONSTRAINT USING INDEX */
            ATSimplePermissions(rel, ATT_TABLE);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_ADD_CONSTR;
            break;
        case AT_DropConstraint: /* DROP CONSTRAINT */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Recursion occurs during execution phase */
            /* No command-specific prep needed except saving recurse flag */
            if (recurse)
                cmd->subtype = AT_DropConstraintRecurse;
            pass = AT_PASS_DROP;
            break;
        case AT_AlterColumnType:    /* ALTER COLUMN TYPE */
            ATSimplePermissions(rel,
                                ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
            /* Performs own recursion */
            ATPrepAlterColumnType(wqueue, tab, rel, recurse, recursing, cmd, lockmode);
            pass = AT_PASS_ALTER_TYPE;
            break;
        case AT_AlterColumnGenericOptions:
            ATSimplePermissions(rel, ATT_FOREIGN_TABLE);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_ChangeOwner:    /* ALTER OWNER */
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_ClusterOn:      /* CLUSTER ON */
        case AT_DropCluster:    /* SET WITHOUT CLUSTER */
            ATSimplePermissions(rel, ATT_TABLE | ATT_MATVIEW);
            /* These commands never recurse */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_SetLogged:      /* SET LOGGED */
            ATSimplePermissions(rel, ATT_TABLE);
            tab->chgPersistence = ATPrepChangePersistence(rel, true);
            /* force rewrite if necessary; see comment in ATRewriteTables */
            if (tab->chgPersistence)
            {
                tab->rewrite |= AT_REWRITE_ALTER_PERSISTENCE;
                tab->newrelpersistence = RELPERSISTENCE_PERMANENT;
            }
            pass = AT_PASS_MISC;
            break;
        case AT_SetUnLogged:    /* SET UNLOGGED */
            ATSimplePermissions(rel, ATT_TABLE);
            tab->chgPersistence = ATPrepChangePersistence(rel, false);
            /* force rewrite if necessary; see comment in ATRewriteTables */
            if (tab->chgPersistence)
            {
                tab->rewrite |= AT_REWRITE_ALTER_PERSISTENCE;
                tab->newrelpersistence = RELPERSISTENCE_UNLOGGED;
            }
            pass = AT_PASS_MISC;
            break;
        case AT_AddOids:        /* SET WITH OIDS */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            if (!rel->rd_rel->relhasoids || recursing)
                ATPrepAddOids(wqueue, rel, recurse, cmd, lockmode);
            /* Recursion occurs during execution phase */
            pass = AT_PASS_ADD_COL;
            break;
        case AT_DropOids:       /* SET WITHOUT OIDS */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Performs own recursion */
            if (rel->rd_rel->relhasoids)
            {
                AlterTableCmd *dropCmd = makeNode(AlterTableCmd);

                dropCmd->subtype = AT_DropColumn;
                dropCmd->name = pstrdup("oid");
                dropCmd->behavior = cmd->behavior;
                ATPrepCmd(wqueue, rel, dropCmd, recurse, false, lockmode);
            }
            pass = AT_PASS_DROP;
            break;
        case AT_SetTableSpace:  /* SET TABLESPACE */
            ATSimplePermissions(rel, ATT_TABLE | ATT_MATVIEW | ATT_INDEX);
            /* This command never recurses */
            ATPrepSetTableSpace(tab, rel, cmd->name, lockmode);
            pass = AT_PASS_MISC;    /* doesn't actually matter */
            break;
        case AT_SetRelOptions:  /* SET (...) */
        case AT_ResetRelOptions:    /* RESET (...) */
        case AT_ReplaceRelOptions:  /* reset them all, then set just these */
            ATSimplePermissions(rel, ATT_TABLE | ATT_VIEW | ATT_MATVIEW | ATT_INDEX);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_AddInherit:     /* INHERIT */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            ATPrepAddInherit(rel);
            pass = AT_PASS_MISC;
            break;
        case AT_DropInherit:    /* NO INHERIT */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_AlterConstraint:    /* ALTER CONSTRAINT */
            ATSimplePermissions(rel, ATT_TABLE);
            pass = AT_PASS_MISC;
            break;
        case AT_ValidateConstraint: /* VALIDATE CONSTRAINT */
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Recursion occurs during execution phase */
            /* No command-specific prep needed except saving recurse flag */
            if (recurse)
                cmd->subtype = AT_ValidateConstraintRecurse;
            pass = AT_PASS_MISC;
            break;
        case AT_ReplicaIdentity:    /* REPLICA IDENTITY ... */
            ATSimplePermissions(rel, ATT_TABLE | ATT_MATVIEW);
            pass = AT_PASS_MISC;
            /* This command never recurses */
            /* No command-specific prep needed */
            break;
        case AT_EnableTrig:     /* ENABLE TRIGGER variants */
        case AT_EnableAlwaysTrig:
        case AT_EnableReplicaTrig:
        case AT_EnableTrigAll:
        case AT_EnableTrigUser:
        case AT_DisableTrig:    /* DISABLE TRIGGER variants */
        case AT_DisableTrigAll:
        case AT_DisableTrigUser:
            ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            pass = AT_PASS_MISC;
            break;
        case AT_EnableRule:     /* ENABLE/DISABLE RULE variants */
        case AT_EnableAlwaysRule:
        case AT_EnableReplicaRule:
        case AT_DisableRule:
        case AT_AddOf:          /* OF */
        case AT_DropOf:         /* NOT OF */
        case AT_EnableRowSecurity:
        case AT_DisableRowSecurity:
        case AT_ForceRowSecurity:
        case AT_NoForceRowSecurity:
            ATSimplePermissions(rel, ATT_TABLE);
            /* These commands never recurse */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_GenericOptions:
            ATSimplePermissions(rel, ATT_FOREIGN_TABLE);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_AttachPartition:
            ATSimplePermissions(rel, ATT_TABLE | ATT_PARTITIONED_INDEX);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_DetachPartition:
            ATSimplePermissions(rel, ATT_TABLE);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        default:                /* oops */
            elog(ERROR, "unrecognized alter table type: %d",
                 (int) cmd->subtype);
            pass = AT_PASS_UNSET;   /* keep compiler quiet */
            break;
    }
    Assert(pass > AT_PASS_UNSET);

    /* Add the subcommand to the appropriate list for phase 2 */
    tab->subcmds[pass] = lappend(tab->subcmds[pass], cmd);
}

/*
 * ATRewriteCatalogs
 *
 * Traffic cop for ALTER TABLE Phase 2 operations.  Subcommands are
 * dispatched in a "safe" execution order (designed to avoid unnecessary
 * conflicts).
 */
static void
ATRewriteCatalogs(List **wqueue, LOCKMODE lockmode)
{
    int         pass;
    ListCell   *ltab;

    /*
     * We process all the tables "in parallel", one pass at a time.  This is
     * needed because we may have to propagate work from one table to another
     * (specifically, ALTER TYPE on a foreign key's PK has to dispatch the
     * re-adding of the foreign key constraint to the other table).  Work can
     * only be propagated into later passes, however.
     */
    for (pass = 0; pass < AT_NUM_PASSES; pass++)
    {
        /* Go through each table that needs to be processed */
        foreach(ltab, *wqueue)
        {
            AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);
            List       *subcmds = tab->subcmds[pass];
            Relation    rel;
            ListCell   *lcmd;

            if (subcmds == NIL)
                continue;

            /*
             * Appropriate lock was obtained by phase 1, needn't get it again
             */
            rel = relation_open(tab->relid, NoLock);

            foreach(lcmd, subcmds)
                ATExecCmd(wqueue, tab, rel,
                          castNode(AlterTableCmd, lfirst(lcmd)),
                          lockmode);

            /*
             * After the ALTER TYPE pass, do cleanup work (this is not done in
             * ATExecAlterColumnType since it should be done only once if
             * multiple columns of a table are altered).
             */
            if (pass == AT_PASS_ALTER_TYPE)
                ATPostAlterTypeCleanup(wqueue, tab, lockmode);

            relation_close(rel, NoLock);
        }
    }

    /* Check to see if a toast table must be added. */
    foreach(ltab, *wqueue)
    {
        AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);

        /*
         * If the table is source table of ATTACH PARTITION command, we did
         * not modify anything about it that will change its toasting
         * requirement, so no need to check.
         */
        if (((tab->relkind == RELKIND_RELATION ||
              tab->relkind == RELKIND_PARTITIONED_TABLE) &&
             tab->partition_constraint == NULL) ||
            tab->relkind == RELKIND_MATVIEW)
            AlterTableCreateToastTable(tab->relid, (Datum) 0, lockmode);
    }
}

/*
 * ATExecCmd: dispatch a subcommand to appropriate execution routine
 */
static void
ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
          AlterTableCmd *cmd, LOCKMODE lockmode)
{
    ObjectAddress address = InvalidObjectAddress;

    switch (cmd->subtype)
    {
        case AT_AddColumn:      /* ADD COLUMN */
        case AT_AddColumnToView:    /* add column via CREATE OR REPLACE VIEW */
            address = ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
                                      false, false, false,
                                      false, lockmode);
            break;
        case AT_AddColumnRecurse:
            address = ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
                                      false, true, false,
                                      cmd->missing_ok, lockmode);
            break;
        case AT_ColumnDefault:  /* ALTER COLUMN DEFAULT */
            address = ATExecColumnDefault(rel, cmd->name, cmd->def, lockmode);
            break;
        case AT_AddIdentity:
            address = ATExecAddIdentity(rel, cmd->name, cmd->def, lockmode);
            break;
        case AT_SetIdentity:
            address = ATExecSetIdentity(rel, cmd->name, cmd->def, lockmode);
            break;
        case AT_DropIdentity:
            address = ATExecDropIdentity(rel, cmd->name, cmd->missing_ok, lockmode);
            break;
        case AT_DropNotNull:    /* ALTER COLUMN DROP NOT NULL */
            address = ATExecDropNotNull(rel, cmd->name, lockmode);
            break;
        case AT_SetNotNull:     /* ALTER COLUMN SET NOT NULL */
            address = ATExecSetNotNull(tab, rel, cmd->name, lockmode);
            break;
        case AT_SetStatistics:  /* ALTER COLUMN SET STATISTICS */
            address = ATExecSetStatistics(rel, cmd->name, cmd->num, cmd->def, lockmode);
            break;
        case AT_SetOptions:     /* ALTER COLUMN SET ( options ) */
            address = ATExecSetOptions(rel, cmd->name, cmd->def, false, lockmode);
            break;
        case AT_ResetOptions:   /* ALTER COLUMN RESET ( options ) */
            address = ATExecSetOptions(rel, cmd->name, cmd->def, true, lockmode);
            break;
        case AT_SetStorage:     /* ALTER COLUMN SET STORAGE */
            address = ATExecSetStorage(rel, cmd->name, cmd->def, lockmode);
            break;
        case AT_DropColumn:     /* DROP COLUMN */
            address = ATExecDropColumn(wqueue, rel, cmd->name,
                                       cmd->behavior, false, false,
                                       cmd->missing_ok, lockmode);
            break;
        case AT_DropColumnRecurse:  /* DROP COLUMN with recursion */
            address = ATExecDropColumn(wqueue, rel, cmd->name,
                                       cmd->behavior, true, false,
                                       cmd->missing_ok, lockmode);
            break;
        case AT_AddIndex:       /* ADD INDEX */
            address = ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, false,
                                     lockmode);
            break;
        case AT_ReAddIndex:     /* ADD INDEX */
            address = ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, true,
                                     lockmode);
            break;
        case AT_AddConstraint:  /* ADD CONSTRAINT */
            address =
                ATExecAddConstraint(wqueue, tab, rel, (Constraint *) cmd->def,
                                    false, false, lockmode);
            break;
        case AT_AddConstraintRecurse:   /* ADD CONSTRAINT with recursion */
            address =
                ATExecAddConstraint(wqueue, tab, rel, (Constraint *) cmd->def,
                                    true, false, lockmode);
            break;
        case AT_ReAddConstraint:    /* Re-add pre-existing check constraint */
            address =
                ATExecAddConstraint(wqueue, tab, rel, (Constraint *) cmd->def,
                                    true, true, lockmode);
            break;
        case AT_ReAddDomainConstraint:  /* Re-add pre-existing domain check
                                         * constraint */
            address =
                AlterDomainAddConstraint(((AlterDomainStmt *) cmd->def)->typeName,
                                         ((AlterDomainStmt *) cmd->def)->def,
                                         NULL);
            break;
        case AT_ReAddComment:   /* Re-add existing comment */
            address = CommentObject((CommentStmt *) cmd->def);
            break;
        case AT_AddIndexConstraint: /* ADD CONSTRAINT USING INDEX */
            address = ATExecAddIndexConstraint(tab, rel, (IndexStmt *) cmd->def,
                                               lockmode);
            break;
        case AT_AlterConstraint:    /* ALTER CONSTRAINT */
            address = ATExecAlterConstraint(rel, cmd, false, false, lockmode);
            break;
        case AT_ValidateConstraint: /* VALIDATE CONSTRAINT */
            address = ATExecValidateConstraint(rel, cmd->name, false, false,
                                               lockmode);
            break;
        case AT_ValidateConstraintRecurse:  /* VALIDATE CONSTRAINT with
                                             * recursion */
            address = ATExecValidateConstraint(rel, cmd->name, true, false,
                                               lockmode);
            break;
        case AT_DropConstraint: /* DROP CONSTRAINT */
            ATExecDropConstraint(rel, cmd->name, cmd->behavior,
                                 false, false,
                                 cmd->missing_ok, lockmode);
            break;
        case AT_DropConstraintRecurse:  /* DROP CONSTRAINT with recursion */
            ATExecDropConstraint(rel, cmd->name, cmd->behavior,
                                 true, false,
                                 cmd->missing_ok, lockmode);
            break;
        case AT_AlterColumnType:    /* ALTER COLUMN TYPE */
            address = ATExecAlterColumnType(tab, rel, cmd, lockmode);
            break;
        case AT_AlterColumnGenericOptions:  /* ALTER COLUMN OPTIONS */
            address =
                ATExecAlterColumnGenericOptions(rel, cmd->name,
                                                (List *) cmd->def, lockmode);
            break;
        case AT_ChangeOwner:    /* ALTER OWNER */
            ATExecChangeOwner(RelationGetRelid(rel),
                              get_rolespec_oid(cmd->newowner, false),
                              false, lockmode);
            break;
        case AT_ClusterOn:      /* CLUSTER ON */
            address = ATExecClusterOn(rel, cmd->name, lockmode);
            break;
        case AT_DropCluster:    /* SET WITHOUT CLUSTER */
            ATExecDropCluster(rel, lockmode);
            break;
        case AT_SetLogged:      /* SET LOGGED */
        case AT_SetUnLogged:    /* SET UNLOGGED */
            break;
        case AT_AddOids:        /* SET WITH OIDS */
            /* Use the ADD COLUMN code, unless prep decided to do nothing */
            if (cmd->def != NULL)
                address =
                    ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
                                    true, false, false,
                                    cmd->missing_ok, lockmode);
            break;
        case AT_AddOidsRecurse: /* SET WITH OIDS */
            /* Use the ADD COLUMN code, unless prep decided to do nothing */
            if (cmd->def != NULL)
                address =
                    ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
                                    true, true, false,
                                    cmd->missing_ok, lockmode);
            break;
        case AT_DropOids:       /* SET WITHOUT OIDS */

            /*
             * Nothing to do here; we'll have generated a DropColumn
             * subcommand to do the real work
             */
            break;
        case AT_SetTableSpace:  /* SET TABLESPACE */

            /*
             * Nothing to do here; Phase 3 does the work
             */
            break;
        case AT_SetRelOptions:  /* SET (...) */
        case AT_ResetRelOptions:    /* RESET (...) */
        case AT_ReplaceRelOptions:  /* replace entire option list */
            ATExecSetRelOptions(rel, (List *) cmd->def, cmd->subtype, lockmode);
            break;
        case AT_EnableTrig:     /* ENABLE TRIGGER name */
            ATExecEnableDisableTrigger(rel, cmd->name,
                                       TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
            break;
        case AT_EnableAlwaysTrig:   /* ENABLE ALWAYS TRIGGER name */
            ATExecEnableDisableTrigger(rel, cmd->name,
                                       TRIGGER_FIRES_ALWAYS, false, lockmode);
            break;
        case AT_EnableReplicaTrig:  /* ENABLE REPLICA TRIGGER name */
            ATExecEnableDisableTrigger(rel, cmd->name,
                                       TRIGGER_FIRES_ON_REPLICA, false, lockmode);
            break;
        case AT_DisableTrig:    /* DISABLE TRIGGER name */
            ATExecEnableDisableTrigger(rel, cmd->name,
                                       TRIGGER_DISABLED, false, lockmode);
            break;
        case AT_EnableTrigAll:  /* ENABLE TRIGGER ALL */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
            break;
        case AT_DisableTrigAll: /* DISABLE TRIGGER ALL */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_DISABLED, false, lockmode);
            break;
        case AT_EnableTrigUser: /* ENABLE TRIGGER USER */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_FIRES_ON_ORIGIN, true, lockmode);
            break;
        case AT_DisableTrigUser:    /* DISABLE TRIGGER USER */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_DISABLED, true, lockmode);
            break;

        case AT_EnableRule:     /* ENABLE RULE name */
            ATExecEnableDisableRule(rel, cmd->name,
                                    RULE_FIRES_ON_ORIGIN, lockmode);
            break;
        case AT_EnableAlwaysRule:   /* ENABLE ALWAYS RULE name */
            ATExecEnableDisableRule(rel, cmd->name,
                                    RULE_FIRES_ALWAYS, lockmode);
            break;
        case AT_EnableReplicaRule:  /* ENABLE REPLICA RULE name */
            ATExecEnableDisableRule(rel, cmd->name,
                                    RULE_FIRES_ON_REPLICA, lockmode);
            break;
        case AT_DisableRule:    /* DISABLE RULE name */
            ATExecEnableDisableRule(rel, cmd->name,
                                    RULE_DISABLED, lockmode);
            break;

        case AT_AddInherit:
            address = ATExecAddInherit(rel, (RangeVar *) cmd->def, lockmode);
            break;
        case AT_DropInherit:
            address = ATExecDropInherit(rel, (RangeVar *) cmd->def, lockmode);
            break;
        case AT_AddOf:
            address = ATExecAddOf(rel, (TypeName *) cmd->def, lockmode);
            break;
        case AT_DropOf:
            ATExecDropOf(rel, lockmode);
            break;
        case AT_ReplicaIdentity:
            ATExecReplicaIdentity(rel, (ReplicaIdentityStmt *) cmd->def, lockmode);
            break;
        case AT_EnableRowSecurity:
            ATExecEnableRowSecurity(rel);
            break;
        case AT_DisableRowSecurity:
            ATExecDisableRowSecurity(rel);
            break;
        case AT_ForceRowSecurity:
            ATExecForceNoForceRowSecurity(rel, true);
            break;
        case AT_NoForceRowSecurity:
            ATExecForceNoForceRowSecurity(rel, false);
            break;
        case AT_GenericOptions:
            ATExecGenericOptions(rel, (List *) cmd->def);
            break;
        case AT_AttachPartition:
            if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
                ATExecAttachPartition(wqueue, rel, (PartitionCmd *) cmd->def);
            else
                ATExecAttachPartitionIdx(wqueue, rel,
                                         ((PartitionCmd *) cmd->def)->name);
            break;
        case AT_DetachPartition:
            /* ATPrepCmd ensures it must be a table */
            Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
            ATExecDetachPartition(rel, ((PartitionCmd *) cmd->def)->name);
            break;
        default:                /* oops */
            elog(ERROR, "unrecognized alter table type: %d",
                 (int) cmd->subtype);
            break;
    }

    /*
     * Report the subcommand to interested event triggers.
     */
    EventTriggerCollectAlterTableSubcmd((Node *) cmd, address);

    /*
     * Bump the command counter to ensure the next subcommand in the sequence
     * can see the changes so far
     */
    CommandCounterIncrement();
}

/*
 * ATRewriteTables: ALTER TABLE phase 3
 */
static void
ATRewriteTables(AlterTableStmt *parsetree, List **wqueue, LOCKMODE lockmode)
{
    ListCell   *ltab;

    /* Go through each table that needs to be checked or rewritten */
    foreach(ltab, *wqueue)
    {
        AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);

        /*
         * Foreign tables have no storage, nor do partitioned tables and
         * indexes.
         */
        if (tab->relkind == RELKIND_FOREIGN_TABLE ||
            tab->relkind == RELKIND_PARTITIONED_TABLE ||
            tab->relkind == RELKIND_PARTITIONED_INDEX)
            continue;

        /*
         * If we change column data types or add/remove OIDs, the operation
         * has to be propagated to tables that use this table's rowtype as a
         * column type.  tab->newvals will also be non-NULL in the case where
         * we're adding a column with a default.  We choose to forbid that
         * case as well, since composite types might eventually support
         * defaults.
         *
         * (Eventually we'll probably need to check for composite type
         * dependencies even when we're just scanning the table without a
         * rewrite, but at the moment a composite type does not enforce any
         * constraints, so it's not necessary/appropriate to enforce them just
         * during ALTER.)
         */
        if (tab->newvals != NIL || tab->rewrite > 0)
        {
            Relation    rel;

            rel = heap_open(tab->relid, NoLock);
            find_composite_type_dependencies(rel->rd_rel->reltype, rel, NULL);
            heap_close(rel, NoLock);
        }

        /*
         * We only need to rewrite the table if at least one column needs to
         * be recomputed, we are adding/removing the OID column, or we are
         * changing its persistence.
         *
         * There are two reasons for requiring a rewrite when changing
         * persistence: on one hand, we need to ensure that the buffers
         * belonging to each of the two relations are marked with or without
         * BM_PERMANENT properly.  On the other hand, since rewriting creates
         * and assigns a new relfilenode, we automatically create or drop an
         * init fork for the relation as appropriate.
         */
        if (tab->rewrite > 0)
        {
            /* Build a temporary relation and copy data */
            Relation    OldHeap;
            Oid         OIDNewHeap;
            Oid         NewTableSpace;
            char        persistence;

            OldHeap = heap_open(tab->relid, NoLock);

            /*
             * We don't support rewriting of system catalogs; there are too
             * many corner cases and too little benefit.  In particular this
             * is certainly not going to work for mapped catalogs.
             */
            if (IsSystemRelation(OldHeap))
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot rewrite system relation \"%s\"",
                                RelationGetRelationName(OldHeap))));

            if (RelationIsUsedAsCatalogTable(OldHeap))
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot rewrite table \"%s\" used as a catalog table",
                                RelationGetRelationName(OldHeap))));

            /*
             * Don't allow rewrite on temp tables of other backends ... their
             * local buffer manager is not going to cope.
             */
            if (RELATION_IS_OTHER_TEMP(OldHeap))
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot rewrite temporary tables of other sessions")));

            /*
             * Select destination tablespace (same as original unless user
             * requested a change)
             */
            if (tab->newTableSpace)
                NewTableSpace = tab->newTableSpace;
            else
                NewTableSpace = OldHeap->rd_rel->reltablespace;

            /*
             * Select persistence of transient table (same as original unless
             * user requested a change)
             */
            persistence = tab->chgPersistence ?
                tab->newrelpersistence : OldHeap->rd_rel->relpersistence;

            heap_close(OldHeap, NoLock);

            /*
             * Fire off an Event Trigger now, before actually rewriting the
             * table.
             *
             * We don't support Event Trigger for nested commands anywhere,
             * here included, and parsetree is given NULL when coming from
             * AlterTableInternal.
             *
             * And fire it only once.
             */
            if (parsetree)
                EventTriggerTableRewrite((Node *) parsetree,
                                         tab->relid,
                                         tab->rewrite);

            /*
             * Create transient table that will receive the modified data.
             *
             * Ensure it is marked correctly as logged or unlogged.  We have
             * to do this here so that buffers for the new relfilenode will
             * have the right persistence set, and at the same time ensure
             * that the original filenode's buffers will get read in with the
             * correct setting (i.e. the original one).  Otherwise a rollback
             * after the rewrite would possibly result with buffers for the
             * original filenode having the wrong persistence setting.
             *
             * NB: This relies on swap_relation_files() also swapping the
             * persistence. That wouldn't work for pg_class, but that can't be
             * unlogged anyway.
             */
            OIDNewHeap = make_new_heap(tab->relid, NewTableSpace, persistence,
                                       lockmode);

            /*
             * Copy the heap data into the new table with the desired
             * modifications, and test the current data within the table
             * against new constraints generated by ALTER TABLE commands.
             */
            ATRewriteTable(tab, OIDNewHeap, lockmode);

            /*
             * Swap the physical files of the old and new heaps, then rebuild
             * indexes and discard the old heap.  We can use RecentXmin for
             * the table's new relfrozenxid because we rewrote all the tuples
             * in ATRewriteTable, so no older Xid remains in the table.  Also,
             * we never try to swap toast tables by content, since we have no
             * interest in letting this code work on system catalogs.
             */
            finish_heap_swap(tab->relid, OIDNewHeap,
                             false, false, true,
                             !OidIsValid(tab->newTableSpace),
                             RecentXmin,
                             ReadNextMultiXactId(),
                             persistence);
        }
        else
        {
            /*
             * Test the current data within the table against new constraints
             * generated by ALTER TABLE commands, but don't rebuild data.
             */
            if (tab->constraints != NIL || tab->new_notnull ||
                tab->partition_constraint != NULL)
                ATRewriteTable(tab, InvalidOid, lockmode);

            /*
             * If we had SET TABLESPACE but no reason to reconstruct tuples,
             * just do a block-by-block copy.
             */
            if (tab->newTableSpace)
                ATExecSetTableSpace(tab->relid, tab->newTableSpace, lockmode);
        }
    }

    /*
     * Foreign key constraints are checked in a final pass, since (a) it's
     * generally best to examine each one separately, and (b) it's at least
     * theoretically possible that we have changed both relations of the
     * foreign key, and we'd better have finished both rewrites before we try
     * to read the tables.
     */
    foreach(ltab, *wqueue)
    {
        AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);
        Relation    rel = NULL;
        ListCell   *lcon;

        foreach(lcon, tab->constraints)
        {
            NewConstraint *con = lfirst(lcon);

            if (con->contype == CONSTR_FOREIGN)
            {
                Constraint *fkconstraint = (Constraint *) con->qual;
                Relation    refrel;

                if (rel == NULL)
                {
                    /* Long since locked, no need for another */
                    rel = heap_open(tab->relid, NoLock);
                }

                refrel = heap_open(con->refrelid, RowShareLock);

                validateForeignKeyConstraint(fkconstraint->conname, rel, refrel,
                                             con->refindid,
                                             con->conid);

                /*
                 * No need to mark the constraint row as validated, we did
                 * that when we inserted the row earlier.
                 */

                heap_close(refrel, NoLock);
            }
        }

        if (rel)
            heap_close(rel, NoLock);
    }
}

/*
 * ATRewriteTable: scan or rewrite one table
 *
 * OIDNewHeap is InvalidOid if we don't need to rewrite
 */
static void
ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
{
    Relation    oldrel;
    Relation    newrel;
    TupleDesc   oldTupDesc;
    TupleDesc   newTupDesc;
    bool        needscan = false;
    List       *notnull_attrs;
    int         i;
    ListCell   *l;
    EState     *estate;
    CommandId   mycid;
    BulkInsertState bistate;
    int         hi_options;
    ExprState  *partqualstate = NULL;

    /*
     * Open the relation(s).  We have surely already locked the existing
     * table.
     */
    oldrel = heap_open(tab->relid, NoLock);
    oldTupDesc = tab->oldDesc;
    newTupDesc = RelationGetDescr(oldrel);  /* includes all mods */

    if (OidIsValid(OIDNewHeap))
        newrel = heap_open(OIDNewHeap, lockmode);
    else
        newrel = NULL;

    /*
     * Prepare a BulkInsertState and options for heap_insert. Because we're
     * building a new heap, we can skip WAL-logging and fsync it to disk at
     * the end instead (unless WAL-logging is required for archiving or
     * streaming replication). The FSM is empty too, so don't bother using it.
     */
    if (newrel)
    {
        mycid = GetCurrentCommandId(true);
        bistate = GetBulkInsertState();

        hi_options = HEAP_INSERT_SKIP_FSM;
        if (!XLogIsNeeded())
            hi_options |= HEAP_INSERT_SKIP_WAL;
    }
    else
    {
        /* keep compiler quiet about using these uninitialized */
        mycid = 0;
        bistate = NULL;
        hi_options = 0;
    }

    /*
     * Generate the constraint and default execution states
     */

    estate = CreateExecutorState();

    /* Build the needed expression execution states */
    foreach(l, tab->constraints)
    {
        NewConstraint *con = lfirst(l);

        switch (con->contype)
        {
            case CONSTR_CHECK:
                needscan = true;
                con->qualstate = ExecPrepareExpr((Expr *) con->qual, estate);
                break;
            case CONSTR_FOREIGN:
                /* Nothing to do here */
                break;
            default:
                elog(ERROR, "unrecognized constraint type: %d",
                     (int) con->contype);
        }
    }

    /* Build expression execution states for partition check quals */
    if (tab->partition_constraint)
    {
        needscan = true;
        partqualstate = ExecPrepareExpr(tab->partition_constraint, estate);
    }

    foreach(l, tab->newvals)
    {
        NewColumnValue *ex = lfirst(l);

        /* expr already planned */
        ex->exprstate = ExecInitExpr((Expr *) ex->expr, NULL);
    }

    notnull_attrs = NIL;
    if (newrel || tab->new_notnull)
    {
        /*
         * If we are rebuilding the tuples OR if we added any new NOT NULL
         * constraints, check all not-null constraints.  This is a bit of
         * overkill but it minimizes risk of bugs, and heap_attisnull is a
         * pretty cheap test anyway.
         */
        for (i = 0; i < newTupDesc->natts; i++)
        {
            Form_pg_attribute attr = TupleDescAttr(newTupDesc, i);

            if (attr->attnotnull && !attr->attisdropped)
                notnull_attrs = lappend_int(notnull_attrs, i);
        }
        if (notnull_attrs)
            needscan = true;
    }

    if (newrel || needscan)
    {
        ExprContext *econtext;
        Datum      *values;
        bool       *isnull;
        TupleTableSlot *oldslot;
        TupleTableSlot *newslot;
        HeapScanDesc scan;
        HeapTuple   tuple;
        MemoryContext oldCxt;
        List       *dropped_attrs = NIL;
        ListCell   *lc;
        Snapshot    snapshot;

        if (newrel)
            ereport(DEBUG1,
                    (errmsg("rewriting table \"%s\"",
                            RelationGetRelationName(oldrel))));
        else
            ereport(DEBUG1,
                    (errmsg("verifying table \"%s\"",
                            RelationGetRelationName(oldrel))));

        if (newrel)
        {
            /*
             * All predicate locks on the tuples or pages are about to be made
             * invalid, because we move tuples around.  Promote them to
             * relation locks.
             */
            TransferPredicateLocksToHeapRelation(oldrel);
        }

        econtext = GetPerTupleExprContext(estate);

        /*
         * Make tuple slots for old and new tuples.  Note that even when the
         * tuples are the same, the tupDescs might not be (consider ADD COLUMN
         * without a default).
         */
        oldslot = MakeSingleTupleTableSlot(oldTupDesc);
        newslot = MakeSingleTupleTableSlot(newTupDesc);

        /* Preallocate values/isnull arrays */
        i = Max(newTupDesc->natts, oldTupDesc->natts);
        values = (Datum *) palloc(i * sizeof(Datum));
        isnull = (bool *) palloc(i * sizeof(bool));
        memset(values, 0, i * sizeof(Datum));
        memset(isnull, true, i * sizeof(bool));

        /*
         * Any attributes that are dropped according to the new tuple
         * descriptor can be set to NULL. We precompute the list of dropped
         * attributes to avoid needing to do so in the per-tuple loop.
         */
        for (i = 0; i < newTupDesc->natts; i++)
        {
            if (TupleDescAttr(newTupDesc, i)->attisdropped)
                dropped_attrs = lappend_int(dropped_attrs, i);
        }

        /*
         * Scan through the rows, generating a new row if needed and then
         * checking all the constraints.
         */
        snapshot = RegisterSnapshot(GetLatestSnapshot());
        scan = heap_beginscan(oldrel, snapshot, 0, NULL);

        /*
         * Switch to per-tuple memory context and reset it for each tuple
         * produced, so we don't leak memory.
         */
        oldCxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));

        while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
        {
            if (tab->rewrite > 0)
            {
                Oid         tupOid = InvalidOid;

                /* Extract data from old tuple */
                heap_deform_tuple(tuple, oldTupDesc, values, isnull);
                if (oldTupDesc->tdhasoid)
                    tupOid = HeapTupleGetOid(tuple);

                /* Set dropped attributes to null in new tuple */
                foreach(lc, dropped_attrs)
                    isnull[lfirst_int(lc)] = true;

                /*
                 * Process supplied expressions to replace selected columns.
                 * Expression inputs come from the old tuple.
                 */
                ExecStoreTuple(tuple, oldslot, InvalidBuffer, false);
                econtext->ecxt_scantuple = oldslot;

                foreach(l, tab->newvals)
                {
                    NewColumnValue *ex = lfirst(l);

                    values[ex->attnum - 1] = ExecEvalExpr(ex->exprstate,
                                                          econtext,
                                                          &isnull[ex->attnum - 1]);
                }

                /*
                 * Form the new tuple. Note that we don't explicitly pfree it,
                 * since the per-tuple memory context will be reset shortly.
                 */
                tuple = heap_form_tuple(newTupDesc, values, isnull);

                /* Preserve OID, if any */
                if (newTupDesc->tdhasoid)
                    HeapTupleSetOid(tuple, tupOid);

                /*
                 * Constraints might reference the tableoid column, so
                 * initialize t_tableOid before evaluating them.
                 */
                tuple->t_tableOid = RelationGetRelid(oldrel);
            }

            /* Now check any constraints on the possibly-changed tuple */
            ExecStoreTuple(tuple, newslot, InvalidBuffer, false);
            econtext->ecxt_scantuple = newslot;

            foreach(l, notnull_attrs)
            {
                int         attn = lfirst_int(l);

                if (heap_attisnull(tuple, attn + 1, newTupDesc))
                {
                    Form_pg_attribute attr = TupleDescAttr(newTupDesc, attn);

                    ereport(ERROR,
                            (errcode(ERRCODE_NOT_NULL_VIOLATION),
                             errmsg("column \"%s\" contains null values",
                                    NameStr(attr->attname)),
                             errtablecol(oldrel, attn + 1)));
                }
            }

            foreach(l, tab->constraints)
            {
                NewConstraint *con = lfirst(l);

                switch (con->contype)
                {
                    case CONSTR_CHECK:
                        if (!ExecCheck(con->qualstate, econtext))
                            ereport(ERROR,
                                    (errcode(ERRCODE_CHECK_VIOLATION),
                                     errmsg("check constraint \"%s\" is violated by some row",
                                            con->name),
                                     errtableconstraint(oldrel, con->name)));
                        break;
                    case CONSTR_FOREIGN:
                        /* Nothing to do here */
                        break;
                    default:
                        elog(ERROR, "unrecognized constraint type: %d",
                             (int) con->contype);
                }
            }

            if (partqualstate && !ExecCheck(partqualstate, econtext))
            {
                if (tab->validate_default)
                    ereport(ERROR,
                            (errcode(ERRCODE_CHECK_VIOLATION),
                             errmsg("updated partition constraint for default partition would be violated by some row")));
                else
                    ereport(ERROR,
                            (errcode(ERRCODE_CHECK_VIOLATION),
                             errmsg("partition constraint is violated by some row")));
            }

            /* Write the tuple out to the new relation */
            if (newrel)
                heap_insert(newrel, tuple, mycid, hi_options, bistate);

            ResetExprContext(econtext);

            CHECK_FOR_INTERRUPTS();
        }

        MemoryContextSwitchTo(oldCxt);
        heap_endscan(scan);
        UnregisterSnapshot(snapshot);

        ExecDropSingleTupleTableSlot(oldslot);
        ExecDropSingleTupleTableSlot(newslot);
    }

    FreeExecutorState(estate);

    heap_close(oldrel, NoLock);
    if (newrel)
    {
        FreeBulkInsertState(bistate);

        /* If we skipped writing WAL, then we need to sync the heap. */
        if (hi_options & HEAP_INSERT_SKIP_WAL)
            heap_sync(newrel);

        heap_close(newrel, NoLock);
    }
}

/*
 * ATGetQueueEntry: find or create an entry in the ALTER TABLE work queue
 */
static AlteredTableInfo *
ATGetQueueEntry(List **wqueue, Relation rel)
{
    Oid         relid = RelationGetRelid(rel);
    AlteredTableInfo *tab;
    ListCell   *ltab;

    foreach(ltab, *wqueue)
    {
        tab = (AlteredTableInfo *) lfirst(ltab);
        if (tab->relid == relid)
            return tab;
    }

    /*
     * Not there, so add it.  Note that we make a copy of the relation's
     * existing descriptor before anything interesting can happen to it.
     */
    tab = (AlteredTableInfo *) palloc0(sizeof(AlteredTableInfo));
    tab->relid = relid;
    tab->relkind = rel->rd_rel->relkind;
    tab->oldDesc = CreateTupleDescCopyConstr(RelationGetDescr(rel));
    tab->newrelpersistence = RELPERSISTENCE_PERMANENT;
    tab->chgPersistence = false;

    *wqueue = lappend(*wqueue, tab);

    return tab;
}

/*
 * ATSimplePermissions
 *
 * - Ensure that it is a relation (or possibly a view)
 * - Ensure this user is the owner
 * - Ensure that it is not a system table
 */
static void
ATSimplePermissions(Relation rel, int allowed_targets)
{
    int         actual_target;

    switch (rel->rd_rel->relkind)
    {
        case RELKIND_RELATION:
        case RELKIND_PARTITIONED_TABLE:
            actual_target = ATT_TABLE;
            break;
        case RELKIND_VIEW:
            actual_target = ATT_VIEW;
            break;
        case RELKIND_MATVIEW:
            actual_target = ATT_MATVIEW;
            break;
        case RELKIND_INDEX:
            actual_target = ATT_INDEX;
            break;
        case RELKIND_PARTITIONED_INDEX:
            actual_target = ATT_PARTITIONED_INDEX;
            break;
        case RELKIND_COMPOSITE_TYPE:
            actual_target = ATT_COMPOSITE_TYPE;
            break;
        case RELKIND_FOREIGN_TABLE:
            actual_target = ATT_FOREIGN_TABLE;
            break;
        default:
            actual_target = 0;
            break;
    }

    /* Wrong target type? */
    if ((actual_target & allowed_targets) == 0)
        ATWrongRelkindError(rel, allowed_targets);

    /* Permissions checks */
    if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId()))
        aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(rel->rd_rel->relkind),
                       RelationGetRelationName(rel));

    if (!allowSystemTableMods && IsSystemRelation(rel))
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("permission denied: \"%s\" is a system catalog",
                        RelationGetRelationName(rel))));
}

/*
 * ATWrongRelkindError
 *
 * Throw an error when a relation has been determined to be of the wrong
 * type.
 */
static void
ATWrongRelkindError(Relation rel, int allowed_targets)
{
    char       *msg;

    switch (allowed_targets)
    {
        case ATT_TABLE:
            msg = _("\"%s\" is not a table");
            break;
        case ATT_TABLE | ATT_VIEW:
            msg = _("\"%s\" is not a table or view");
            break;
        case ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE:
            msg = _("\"%s\" is not a table, view, or foreign table");
            break;
        case ATT_TABLE | ATT_VIEW | ATT_MATVIEW | ATT_INDEX:
            msg = _("\"%s\" is not a table, view, materialized view, or index");
            break;
        case ATT_TABLE | ATT_MATVIEW:
            msg = _("\"%s\" is not a table or materialized view");
            break;
        case ATT_TABLE | ATT_MATVIEW | ATT_INDEX:
            msg = _("\"%s\" is not a table, materialized view, or index");
            break;
        case ATT_TABLE | ATT_MATVIEW | ATT_FOREIGN_TABLE:
            msg = _("\"%s\" is not a table, materialized view, or foreign table");
            break;
        case ATT_TABLE | ATT_FOREIGN_TABLE:
            msg = _("\"%s\" is not a table or foreign table");
            break;
        case ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE:
            msg = _("\"%s\" is not a table, composite type, or foreign table");
            break;
        case ATT_TABLE | ATT_MATVIEW | ATT_INDEX | ATT_FOREIGN_TABLE:
            msg = _("\"%s\" is not a table, materialized view, index, or foreign table");
            break;
        case ATT_VIEW:
            msg = _("\"%s\" is not a view");
            break;
        case ATT_FOREIGN_TABLE:
            msg = _("\"%s\" is not a foreign table");
            break;
        default:
            /* shouldn't get here, add all necessary cases above */
            msg = _("\"%s\" is of the wrong type");
            break;
    }

    ereport(ERROR,
            (errcode(ERRCODE_WRONG_OBJECT_TYPE),
             errmsg(msg, RelationGetRelationName(rel))));
}

/*
 * ATSimpleRecursion
 *
 * Simple table recursion sufficient for most ALTER TABLE operations.
 * All direct and indirect children are processed in an unspecified order.
 * Note that if a child inherits from the original table via multiple
 * inheritance paths, it will be visited just once.
 */
static void
ATSimpleRecursion(List **wqueue, Relation rel,
                  AlterTableCmd *cmd, bool recurse, LOCKMODE lockmode)
{
    /*
     * Propagate to children if desired.  Only plain tables and foreign tables
     * have children, so no need to search for other relkinds.
     */
    if (recurse &&
        (rel->rd_rel->relkind == RELKIND_RELATION ||
         rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE ||
         rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE))
    {
        Oid         relid = RelationGetRelid(rel);
        ListCell   *child;
        List       *children;

        children = find_all_inheritors(relid, lockmode, NULL);

        /*
         * find_all_inheritors does the recursive search of the inheritance
         * hierarchy, so all we have to do is process all of the relids in the
         * list that it returns.
         */
        foreach(child, children)
        {
            Oid         childrelid = lfirst_oid(child);
            Relation    childrel;

            if (childrelid == relid)
                continue;
            /* find_all_inheritors already got lock */
            childrel = relation_open(childrelid, NoLock);
            CheckTableNotInUse(childrel, "ALTER TABLE");
            ATPrepCmd(wqueue, childrel, cmd, false, true, lockmode);
            relation_close(childrel, NoLock);
        }
    }
}

/*
 * ATTypedTableRecursion
 *
 * Propagate ALTER TYPE operations to the typed tables of that type.
 * Also check the RESTRICT/CASCADE behavior.  Given CASCADE, also permit
 * recursion to inheritance children of the typed tables.
 */
static void
ATTypedTableRecursion(List **wqueue, Relation rel, AlterTableCmd *cmd,
                      LOCKMODE lockmode)
{
    ListCell   *child;
    List       *children;

    Assert(rel->rd_rel->relkind == RELKIND_COMPOSITE_TYPE);

    children = find_typed_table_dependencies(rel->rd_rel->reltype,
                                             RelationGetRelationName(rel),
                                             cmd->behavior);

    foreach(child, children)
    {
        Oid         childrelid = lfirst_oid(child);
        Relation    childrel;

        childrel = relation_open(childrelid, lockmode);
        CheckTableNotInUse(childrel, "ALTER TABLE");
        ATPrepCmd(wqueue, childrel, cmd, true, true, lockmode);
        relation_close(childrel, NoLock);
    }
}


/*
 * find_composite_type_dependencies
 *
 * Check to see if the type "typeOid" is being used as a column in some table
 * (possibly nested several levels deep in composite types, arrays, etc!).
 * Eventually, we'd like to propagate the check or rewrite operation
 * into such tables, but for now, just error out if we find any.
 *
 * Caller should provide either the associated relation of a rowtype,
 * or a type name (not both) for use in the error message, if any.
 *
 * Note that "typeOid" is not necessarily a composite type; it could also be
 * another container type such as an array or range, or a domain over one of
 * these things.  The name of this function is therefore somewhat historical,
 * but it's not worth changing.
 *
 * We assume that functions and views depending on the type are not reasons
 * to reject the ALTER.  (How safe is this really?)
 */
void
find_composite_type_dependencies(Oid typeOid, Relation origRelation,
                                 const char *origTypeName)
{
    Relation    depRel;
    ScanKeyData key[2];
    SysScanDesc depScan;
    HeapTuple   depTup;

    /* since this function recurses, it could be driven to stack overflow */
    check_stack_depth();

    /*
     * We scan pg_depend to find those things that depend on the given type.
     * (We assume we can ignore refobjsubid for a type.)
     */
    depRel = heap_open(DependRelationId, AccessShareLock);

    ScanKeyInit(&key[0],
                Anum_pg_depend_refclassid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(TypeRelationId));
    ScanKeyInit(&key[1],
                Anum_pg_depend_refobjid,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(typeOid));

    depScan = systable_beginscan(depRel, DependReferenceIndexId, true,
                                 NULL, 2, key);

    while (HeapTupleIsValid(depTup = systable_getnext(depScan)))
    {
        Form_pg_depend pg_depend = (Form_pg_depend) GETSTRUCT(depTup);
        Relation    rel;
        Form_pg_attribute att;

        /* Check for directly dependent types */
        if (pg_depend->classid == TypeRelationId)
        {
            /*
             * This must be an array, domain, or range containing the given
             * type, so recursively check for uses of this type.  Note that
             * any error message will mention the original type not the
             * container; this is intentional.
             */
            find_composite_type_dependencies(pg_depend->objid,
                                             origRelation, origTypeName);
            continue;
        }

        /* Else, ignore dependees that aren't user columns of relations */
        /* (we assume system columns are never of interesting types) */
        if (pg_depend->classid != RelationRelationId ||
            pg_depend->objsubid <= 0)
            continue;

        rel = relation_open(pg_depend->objid, AccessShareLock);
        att = TupleDescAttr(rel->rd_att, pg_depend->objsubid - 1);

        if (rel->rd_rel->relkind == RELKIND_RELATION ||
            rel->rd_rel->relkind == RELKIND_MATVIEW ||
            rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
        {
            if (origTypeName)
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot alter type \"%s\" because column \"%s.%s\" uses it",
                                origTypeName,
                                RelationGetRelationName(rel),
                                NameStr(att->attname))));
            else if (origRelation->rd_rel->relkind == RELKIND_COMPOSITE_TYPE)
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot alter type \"%s\" because column \"%s.%s\" uses it",
                                RelationGetRelationName(origRelation),
                                RelationGetRelationName(rel),
                                NameStr(att->attname))));
            else if (origRelation->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot alter foreign table \"%s\" because column \"%s.%s\" uses its row type",
                                RelationGetRelationName(origRelation),
                                RelationGetRelationName(rel),
                                NameStr(att->attname))));
            else
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot alter table \"%s\" because column \"%s.%s\" uses its row type",
                                RelationGetRelationName(origRelation),
                                RelationGetRelationName(rel),
                                NameStr(att->attname))));
        }
        else if (OidIsValid(rel->rd_rel->reltype))
        {
            /*
             * A view or composite type itself isn't a problem, but we must
             * recursively check for indirect dependencies via its rowtype.
             */
            find_composite_type_dependencies(rel->rd_rel->reltype,
                                             origRelation, origTypeName);
        }

        relation_close(rel, AccessShareLock);
    }

    systable_endscan(depScan);

    relation_close(depRel, AccessShareLock);
}


/*
 * find_typed_table_dependencies
 *
 * Check to see if a composite type is being used as the type of a
 * typed table.  Abort if any are found and behavior is RESTRICT.
 * Else return the list of tables.
 */
static List *
find_typed_table_dependencies(Oid typeOid, const char *typeName, DropBehavior behavior)
{
    Relation    classRel;
    ScanKeyData key[1];
    HeapScanDesc scan;
    HeapTuple   tuple;
    List       *result = NIL;

    classRel = heap_open(RelationRelationId, AccessShareLock);

    ScanKeyInit(&key[0],
                Anum_pg_class_reloftype,
                BTEqualStrategyNumber, F_OIDEQ,
                ObjectIdGetDatum(typeOid));

    scan = heap_beginscan_catalog(classRel, 1, key);

    while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
    {
        if (behavior == DROP_RESTRICT)
            ereport(ERROR,
                    (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
                     errmsg("cannot alter type \"%s\" because it is the type of a typed table",
                            typeName),
                     errhint("Use ALTER ... CASCADE to alter the typed tables too.")));
        else
            result = lappend_oid(result, HeapTupleGetOid(tuple));
    }

    heap_endscan(scan);
    heap_close(classRel, AccessShareLock);

    return result;
}


/*
 * check_of_type
 *
 * Check whether a type is suitable for CREATE TABLE OF/ALTER TABLE OF.  If it
 * isn't suitable, throw an error.  Currently, we require that the type
 * originated with CREATE TYPE AS.  We could support any row type, but doing so
 * would require handling a number of extra corner cases in the DDL commands.
 * (Also, allowing domain-over-composite would open up a can of worms about
 * whether and how the domain's constraints should apply to derived tables.)
 */
void
check_of_type(HeapTuple typetuple)
{
    Form_pg_type typ = (Form_pg_type) GETSTRUCT(typetuple);
    bool        typeOk = false;

    if (typ->typtype == TYPTYPE_COMPOSITE)
    {
        Relation    typeRelation;

        Assert(OidIsValid(typ->typrelid));
        typeRelation = relation_open(typ->typrelid, AccessShareLock);
        typeOk = (typeRelation->rd_rel->relkind == RELKIND_COMPOSITE_TYPE);

        /*
         * Close the parent rel, but keep our AccessShareLock on it until xact
         * commit.  That will prevent someone else from deleting or ALTERing
         * the type before the typed table creation/conversion commits.
         */
        relation_close(typeRelation, NoLock);
    }
    if (!typeOk)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("type %s is not a composite type",
                        format_type_be(HeapTupleGetOid(typetuple)))));
}


/*
 * ALTER TABLE ADD COLUMN
 *
 * Adds an additional attribute to a relation making the assumption that
 * CHECK, NOT NULL, and FOREIGN KEY constraints will be removed from the
 * AT_AddColumn AlterTableCmd by parse_utilcmd.c and added as independent
 * AlterTableCmd's.
 *
 * ADD COLUMN cannot use the normal ALTER TABLE recursion mechanism, because we
 * have to decide at runtime whether to recurse or not depending on whether we
 * actually add a column or merely merge with an existing column.  (We can't
 * check this in a static pre-pass because it won't handle multiple inheritance
 * situations correctly.)
 */
static void
ATPrepAddColumn(List **wqueue, Relation rel, bool recurse, bool recursing,
                bool is_view, AlterTableCmd *cmd, LOCKMODE lockmode)
{
    if (rel->rd_rel->reloftype && !recursing)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("cannot add column to typed table")));

    if (rel->rd_rel->relkind == RELKIND_COMPOSITE_TYPE)
        ATTypedTableRecursion(wqueue, rel, cmd, lockmode);

    if (recurse && !is_view)
        cmd->subtype = AT_AddColumnRecurse;
}

/*
 * Add a column to a table; this handles the AT_AddOids cases as well.  The
 * return value is the address of the new column in the parent relation.
 */
static ObjectAddress
ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
                ColumnDef *colDef, bool isOid,
                bool recurse, bool recursing,
                bool if_not_exists, LOCKMODE lockmode)
{
    Oid         myrelid = RelationGetRelid(rel);
    Relation    pgclass,
                attrdesc;
    HeapTuple   reltup;
    FormData_pg_attribute attribute;
    int         newattnum;
    char        relkind;
    HeapTuple   typeTuple;
    Oid         typeOid;
    int32       typmod;
    Oid         collOid;
    Form_pg_type tform;
    Expr       *defval;
    List       *children;
    ListCell   *child;
    AclResult   aclresult;
    ObjectAddress address;

    /* At top level, permission check was done in ATPrepCmd, else do it */
    if (recursing)
        ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);

    if (rel->rd_rel->relispartition && !recursing)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("cannot add column to a partition")));

    attrdesc = heap_open(AttributeRelationId, RowExclusiveLock);

    /*
     * Are we adding the column to a recursion child?  If so, check whether to
     * merge with an existing definition for the column.  If we do merge, we
     * must not recurse.  Children will already have the column, and recursing
     * into them would mess up attinhcount.
     */
    if (colDef->inhcount > 0)
    {
        HeapTuple   tuple;

        /* Does child already have a column by this name? */
        tuple = SearchSysCacheCopyAttName(myrelid, colDef->colname);
        if (HeapTupleIsValid(tuple))
        {
            Form_pg_attribute childatt = (Form_pg_attribute) GETSTRUCT(tuple);
            Oid         ctypeId;
            int32       ctypmod;
            Oid         ccollid;

            /* Child column must match on type, typmod, and collation */
            typenameTypeIdAndMod(NULL, colDef->typeName, &ctypeId, &ctypmod);
            if (ctypeId != childatt->atttypid ||
                ctypmod != childatt->atttypmod)
                ereport(ERROR,
                        (errcode(ERRCODE_DATATYPE_MISMATCH),
                         errmsg("child table \"%s\" has different type for column \"%s\"",
                                RelationGetRelationName(rel), colDef->colname)));
            ccollid = GetColumnDefCollation(NULL, colDef, ctypeId);
            if (ccollid != childatt->attcollation)
                ereport(ERROR,
                        (errcode(ERRCODE_COLLATION_MISMATCH),
                         errmsg("child table \"%s\" has different collation for column \"%s\"",
                                RelationGetRelationName(rel), colDef->colname),
                         errdetail("\"%s\" versus \"%s\"",
                                   get_collation_name(ccollid),
                                   get_collation_name(childatt->attcollation))));

            /* If it's OID, child column must actually be OID */
            if (isOid && childatt->attnum != ObjectIdAttributeNumber)
                ereport(ERROR,
                        (errcode(ERRCODE_DATATYPE_MISMATCH),
                         errmsg("child table \"%s\" has a conflicting \"%s\" column",
                                RelationGetRelationName(rel), colDef->colname)));

            /* Bump the existing child att's inhcount */
            childatt->attinhcount++;
            CatalogTupleUpdate(attrdesc, &tuple->t_self, tuple);

            heap_freetuple(tuple);

            /* Inform the user about the merge */
            ereport(NOTICE,
                    (errmsg("merging definition of column \"%s\" for child \"%s\"",
                            colDef->colname,