tablecmds.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * tablecmds.c
 *    Commands for creating and altering table structures and settings
 *
 * Portions Copyright (c) 1996-2023, 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/attmap.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/tableam.h"
#include "access/toast_compression.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "catalog/catalog.h"
#include "catalog/heap.h"
#include "catalog/index.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/partition.h"
#include "catalog/pg_am.h"
#include "catalog/pg_attrdef.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_largeobject.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_opclass.h"
#include "catalog/pg_statistic_ext.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/fdwapi.h"
#include "foreign/foreign.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "nodes/parsenodes.h"
#include "optimizer/optimizer.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 "partitioning/partdesc.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 "tcop/utility.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/timestamp.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 */
/* We could support a RENAME COLUMN pass here, but not currently used */
#define AT_PASS_ADD_COL         4   /* ADD COLUMN */
#define AT_PASS_ADD_CONSTR      5   /* ADD constraints (initial examination) */
#define AT_PASS_COL_ATTRS       6   /* set column attributes, eg NOT NULL */
#define AT_PASS_ADD_INDEXCONSTR 7   /* ADD index-based constraints */
#define AT_PASS_ADD_INDEX       8   /* ADD indexes */
#define AT_PASS_ADD_OTHERCONSTR 9   /* ADD other constraints, defaults */
#define AT_PASS_MISC            10  /* other stuff */
#define AT_NUM_PASSES           11
 
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 */
 
    /*
     * Transiently set during Phase 2, normally set to NULL.
     *
     * ATRewriteCatalogs sets this when it starts, and closes when ATExecCmd
     * returns control.  This can be exploited by ATExecCmd subroutines to
     * close/reopen across transaction boundaries.
     */
    Relation    rel;
 
    /* 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 */
    List       *afterStmts;     /* List of utility command parsetrees */
    bool        verify_new_notnull; /* T if we should recheck NOT NULL */
    int         rewrite;        /* Reason for forced rewrite, if any */
    Oid         newAccessMethod;    /* new access method; 0 means no change */
    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 */
    char       *replicaIdentityIndex;   /* index to reset as REPLICA IDENTITY */
    char       *clusterOnIndex; /* index to use for CLUSTER */
    List       *changedStatisticsOids;  /* OIDs of statistics to rebuild */
    List       *changedStatisticsDefs;  /* 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, except when is_generated
 * is true; then it is an expression over columns of the *new* tuple.
 */
typedef struct NewColumnValue
{
    AttrNumber  attnum;         /* which column */
    Expr       *expr;           /* expression to compute */
    ExprState  *exprstate;      /* execution state */
    bool        is_generated;   /* is it a GENERATED expression? */
} 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}
};
 
/* communication between RemoveRelations and RangeVarCallbackForDropRelation */
struct DropRelationCallbackState
{
    /* These fields are set by RemoveRelations: */
    char        expected_relkind;
    LOCKMODE    heap_lockmode;
    /* These fields are state to track which subsidiary locks are held: */
    Oid         heapOid;
    Oid         partParentOid;
    /* These fields are passed back by RangeVarCallbackForDropRelation: */
    char        actual_relkind;
    char        actual_relpersistence;
};
 
/* 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
#define     ATT_SEQUENCE            0x0080
 
/*
 * ForeignTruncateInfo
 *
 * Information related to truncation of foreign tables.  This is used for
 * the elements in a hash table. It uses the server OID as lookup key,
 * and includes a per-server list of all foreign tables involved in the
 * truncation.
 */
typedef struct ForeignTruncateInfo
{
    Oid         serverid;
    List       *rels;
} ForeignTruncateInfo;
 
/*
 * 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(Oid relid, Form_pg_class reltuple);
static void truncate_check_perms(Oid relid, Form_pg_class reltuple);
static void truncate_check_activity(Relation rel);
static void RangeVarCallbackForTruncate(const RangeVar *relation,
                                        Oid relId, Oid oldRelId, void *arg);
static List *MergeAttributes(List *schema, List *supers, char relpersistence,
                             bool is_partition, List **supconstr);
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 bool ATExecAlterConstrRecurse(Constraint *cmdcon, Relation conrel, Relation tgrel,
                                     Relation rel, HeapTuple contuple, List **otherrelids,
                                     LOCKMODE lockmode);
static ObjectAddress ATExecValidateConstraint(List **wqueue,
                                              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 validateForeignKeyConstraint(char *conname,
                                         Relation rel, Relation pkrel,
                                         Oid pkindOid, Oid constraintOid);
static void ATController(AlterTableStmt *parsetree,
                         Relation rel, List *cmds, bool recurse, LOCKMODE lockmode,
                         AlterTableUtilityContext *context);
static void ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
                      bool recurse, bool recursing, LOCKMODE lockmode,
                      AlterTableUtilityContext *context);
static void ATRewriteCatalogs(List **wqueue, LOCKMODE lockmode,
                              AlterTableUtilityContext *context);
static void ATExecCmd(List **wqueue, AlteredTableInfo *tab,
                      AlterTableCmd *cmd, LOCKMODE lockmode, int cur_pass,
                      AlterTableUtilityContext *context);
static AlterTableCmd *ATParseTransformCmd(List **wqueue, AlteredTableInfo *tab,
                                          Relation rel, AlterTableCmd *cmd,
                                          bool recurse, LOCKMODE lockmode,
                                          int cur_pass,
                                          AlterTableUtilityContext *context);
static void ATRewriteTables(AlterTableStmt *parsetree,
                            List **wqueue, LOCKMODE lockmode,
                            AlterTableUtilityContext *context);
static void ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode);
static AlteredTableInfo *ATGetQueueEntry(List **wqueue, Relation rel);
static void ATSimplePermissions(AlterTableType cmdtype, Relation rel, int allowed_targets);
static void ATSimpleRecursion(List **wqueue, Relation rel,
                              AlterTableCmd *cmd, bool recurse, LOCKMODE lockmode,
                              AlterTableUtilityContext *context);
static void ATCheckPartitionsNotInUse(Relation rel, LOCKMODE lockmode);
static void ATTypedTableRecursion(List **wqueue, Relation rel, AlterTableCmd *cmd,
                                  LOCKMODE lockmode,
                                  AlterTableUtilityContext *context);
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,
                            AlterTableUtilityContext *context);
static ObjectAddress ATExecAddColumn(List **wqueue, AlteredTableInfo *tab,
                                     Relation rel, AlterTableCmd **cmd,
                                     bool recurse, bool recursing,
                                     LOCKMODE lockmode, int cur_pass,
                                     AlterTableUtilityContext *context);
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 ATPrepDropNotNull(Relation rel, bool recurse, bool recursing);
static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode);
static void ATPrepSetNotNull(List **wqueue, Relation rel,
                             AlterTableCmd *cmd, bool recurse, bool recursing,
                             LOCKMODE lockmode,
                             AlterTableUtilityContext *context);
static ObjectAddress ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
                                      const char *colName, LOCKMODE lockmode);
static void ATExecCheckNotNull(AlteredTableInfo *tab, Relation rel,
                               const char *colName, LOCKMODE lockmode);
static bool NotNullImpliedByRelConstraints(Relation rel, Form_pg_attribute attr);
static bool ConstraintImpliedByRelConstraint(Relation scanrel,
                                             List *testConstraint, List *provenConstraint);
static ObjectAddress ATExecColumnDefault(Relation rel, const char *colName,
                                         Node *newDefault, LOCKMODE lockmode);
static ObjectAddress ATExecCookedColumnDefault(Relation rel, AttrNumber attnum,
                                               Node *newDefault);
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 ATPrepDropExpression(Relation rel, AlterTableCmd *cmd, bool recurse, bool recursing, LOCKMODE lockmode);
static ObjectAddress ATExecDropExpression(Relation rel, const char *colName, bool missing_ok, 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,
                             AlterTableUtilityContext *context);
static ObjectAddress ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
                                      DropBehavior behavior,
                                      bool recurse, bool recursing,
                                      bool missing_ok, LOCKMODE lockmode,
                                      ObjectAddresses *addrs);
static ObjectAddress ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
                                    IndexStmt *stmt, bool is_rebuild, LOCKMODE lockmode);
static ObjectAddress ATExecAddStatistics(AlteredTableInfo *tab, Relation rel,
                                         CreateStatsStmt *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 char *ChooseForeignKeyConstraintNameAddition(List *colnames);
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,
                                               bool recurse, bool recursing,
                                               LOCKMODE lockmode);
static ObjectAddress addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint,
                                            Relation rel, Relation pkrel, Oid indexOid, Oid parentConstr,
                                            int numfks, int16 *pkattnum, int16 *fkattnum,
                                            Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators,
                                            int numfkdelsetcols, int16 *fkdelsetcols,
                                            bool old_check_ok,
                                            Oid parentDelTrigger, Oid parentUpdTrigger);
static void validateFkOnDeleteSetColumns(int numfks, const int16 *fkattnums,
                                         int numfksetcols, const int16 *fksetcolsattnums,
                                         List *fksetcols);
static void addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint,
                                    Relation rel, Relation pkrel, Oid indexOid, Oid parentConstr,
                                    int numfks, int16 *pkattnum, int16 *fkattnum,
                                    Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators,
                                    int numfkdelsetcols, int16 *fkdelsetcols,
                                    bool old_check_ok, LOCKMODE lockmode,
                                    Oid parentInsTrigger, Oid parentUpdTrigger);
static void CloneForeignKeyConstraints(List **wqueue, Relation parentRel,
                                       Relation partitionRel);
static void CloneFkReferenced(Relation parentRel, Relation partitionRel);
static void CloneFkReferencing(List **wqueue, Relation parentRel,
                               Relation partRel);
static void createForeignKeyCheckTriggers(Oid myRelOid, Oid refRelOid,
                                          Constraint *fkconstraint, Oid constraintOid,
                                          Oid indexOid,
                                          Oid parentInsTrigger, Oid parentUpdTrigger,
                                          Oid *insertTrigOid, Oid *updateTrigOid);
static void createForeignKeyActionTriggers(Relation rel, Oid refRelOid,
                                           Constraint *fkconstraint, Oid constraintOid,
                                           Oid indexOid,
                                           Oid parentDelTrigger, Oid parentUpdTrigger,
                                           Oid *deleteTrigOid, Oid *updateTrigOid);
static bool tryAttachPartitionForeignKey(ForeignKeyCacheInfo *fk,
                                         Oid partRelid,
                                         Oid parentConstrOid, int numfks,
                                         AttrNumber *mapped_conkey, AttrNumber *confkey,
                                         Oid *conpfeqop,
                                         Oid parentInsTrigger,
                                         Oid parentUpdTrigger,
                                         Relation trigrel);
static void GetForeignKeyActionTriggers(Relation trigrel,
                                        Oid conoid, Oid confrelid, Oid conrelid,
                                        Oid *deleteTriggerOid,
                                        Oid *updateTriggerOid);
static void GetForeignKeyCheckTriggers(Relation trigrel,
                                       Oid conoid, Oid confrelid, Oid conrelid,
                                       Oid *insertTriggerOid,
                                       Oid *updateTriggerOid);
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,
                                  AlterTableUtilityContext *context);
static bool ATColumnChangeRequiresRewrite(Node *expr, AttrNumber varattno);
static ObjectAddress ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
                                           AlterTableCmd *cmd, LOCKMODE lockmode);
static void RememberConstraintForRebuilding(Oid conoid, AlteredTableInfo *tab);
static void RememberIndexForRebuilding(Oid indoid, AlteredTableInfo *tab);
static void RememberStatisticsForRebuilding(Oid stxoid, AlteredTableInfo *tab);
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 ObjectAddress ATExecAlterColumnGenericOptions(Relation rel, const char *colName,
                                                     List *options, LOCKMODE lockmode);
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 void ATPrepSetAccessMethod(AlteredTableInfo *tab, Relation rel, const char *amname);
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 ATExecSetTableSpaceNoStorage(Relation rel, Oid newTableSpace);
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, bool recurse,
                                       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 ATExecSetRowSecurity(Relation rel, bool rls);
static void ATExecForceNoForceRowSecurity(Relation rel, bool force_rls);
static ObjectAddress ATExecSetCompression(Relation rel,
                                          const char *column, Node *newValue, LOCKMODE lockmode);
 
static void index_copy_data(Relation rel, RelFileLocator newrlocator);
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);
static void ComputePartitionAttrs(ParseState *pstate, Relation rel, List *partParams, AttrNumber *partattrs,
                                  List **partexprs, Oid *partopclass, Oid *partcollation,
                                  PartitionStrategy strategy);
static void CreateInheritance(Relation child_rel, Relation parent_rel);
static void RemoveInheritance(Relation child_rel, Relation parent_rel,
                              bool expect_detached);
static ObjectAddress ATExecAttachPartition(List **wqueue, Relation rel,
                                           PartitionCmd *cmd,
                                           AlterTableUtilityContext *context);
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 void DetachAddConstraintIfNeeded(List **wqueue, Relation partRel);
static void DropClonedTriggersFromPartition(Oid partitionId);
static ObjectAddress ATExecDetachPartition(List **wqueue, AlteredTableInfo *tab,
                                           Relation rel, RangeVar *name,
                                           bool concurrent);
static void DetachPartitionFinalize(Relation rel, Relation partRel,
                                    bool concurrent, Oid defaultPartOid);
static ObjectAddress ATExecDetachPartitionFinalize(Relation rel, RangeVar *name);
static ObjectAddress ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx,
                                              RangeVar *name);
static void validatePartitionedIndex(Relation partedIdx, Relation partedTbl);
static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx,
                                  Relation partitionTbl);
static List *GetParentedForeignKeyRefs(Relation partition);
static void ATDetachCheckNoForeignKeyRefs(Relation partition);
static char GetAttributeCompression(Oid atttypid, char *compression);
static char GetAttributeStorage(Oid atttypid, const char *storagemode);
 
 
/* ----------------------------------------------------------------
 *      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;
    List       *rawDefaults;
    List       *cookedDefaults;
    Datum       reloptions;
    ListCell   *listptr;
    AttrNumber  attnum;
    bool        partitioned;
    static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
    Oid         ofTypeId;
    ObjectAddress address;
    LOCKMODE    parentLockmode;
    const char *accessMethod = NULL;
    Oid         accessMethodId = InvalidOid;
 
    /*
     * Truncate relname to appropriate length (probably a waste of time, as
     * parser should have done this already).
     */
    strlcpy(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;
        partitioned = true;
    }
    else
        partitioned = false;
 
    /*
     * 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")));
 
    /*
     * Determine the lockmode to use when scanning parents.  A self-exclusive
     * lock is needed here.
     *
     * For regular inheritance, 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.
     */
    parentLockmode = (stmt->partbound != NULL ? AccessExclusiveLock :
                      ShareUpdateExclusiveLock);
 
    /* Determine the list of OIDs of the parents. */
    inheritOids = NIL;
    foreach(listptr, stmt->inhRelations)
    {
        RangeVar   *rv = (RangeVar *) lfirst(listptr);
        Oid         parentOid;
 
        parentOid = RangeVarGetRelid(rv, parentLockmode, false);
 
        /*
         * Reject duplications in the list of parents.
         */
        if (list_member_oid(inheritOids, parentOid))
            ereport(ERROR,
                    (errcode(ERRCODE_DUPLICATE_TABLE),
                     errmsg("relation \"%s\" would be inherited from more than once",
                            get_rel_name(parentOid))));
 
        inheritOids = lappend_oid(inheritOids, parentOid);
    }
 
    /*
     * Select tablespace to use: an explicitly indicated one, or (in the case
     * of a partitioned table) the parent's, if it has one.
     */
    if (stmt->tablespacename)
    {
        tablespaceId = get_tablespace_oid(stmt->tablespacename, false);
 
        if (partitioned && tablespaceId == MyDatabaseTableSpace)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("cannot specify default tablespace for partitioned relations")));
    }
    else if (stmt->partbound)
    {
        /*
         * For partitions, when no other tablespace is specified, we default
         * the tablespace to the parent partitioned table's.
         */
        Assert(list_length(inheritOids) == 1);
        tablespaceId = get_rel_tablespace(linitial_oid(inheritOids));
    }
    else
        tablespaceId = InvalidOid;
 
    /* still nothing? use the default */
    if (!OidIsValid(tablespaceId))
        tablespaceId = GetDefaultTablespace(stmt->relation->relpersistence,
                                            partitioned);
 
    /* Check permissions except when using database's default */
    if (OidIsValid(tablespaceId) && tablespaceId != MyDatabaseTableSpace)
    {
        AclResult   aclresult;
 
        aclresult = object_aclcheck(TableSpaceRelationId, 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);
 
    switch (relkind)
    {
        case RELKIND_VIEW:
            (void) view_reloptions(reloptions, true);
            break;
        case RELKIND_PARTITIONED_TABLE:
            (void) partitioned_table_reloptions(reloptions, true);
            break;
        default:
            (void) heap_reloptions(relkind, reloptions, true);
    }
 
    if (stmt->ofTypename)
    {
        AclResult   aclresult;
 
        ofTypeId = typenameTypeId(NULL, stmt->ofTypename);
 
        aclresult = object_aclcheck(TypeRelationId, 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, inheritOids,
                        stmt->relation->relpersistence,
                        stmt->partbound != NULL,
                        &old_constraints);
 
    /*
     * 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);
 
    /*
     * 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;
            rawEnt->generated = colDef->generated;
            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;
 
        if (colDef->generated)
            attr->attgenerated = colDef->generated;
 
        if (colDef->compression)
            attr->attcompression = GetAttributeCompression(attr->atttypid,
                                                           colDef->compression);
 
        if (colDef->storage_name)
            attr->attstorage = GetAttributeStorage(attr->atttypid, colDef->storage_name);
    }
 
    /*
     * If the statement hasn't specified an access method, but we're defining
     * a type of relation that needs one, use the default.
     */
    if (stmt->accessMethod != NULL)
    {
        accessMethod = stmt->accessMethod;
 
        if (partitioned)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("specifying a table access method is not supported on a partitioned table")));
    }
    else if (RELKIND_HAS_TABLE_AM(relkind))
        accessMethod = default_table_access_method;
 
    /* look up the access method, verify it is for a table */
    if (accessMethod != NULL)
        accessMethodId = get_table_am_oid(accessMethod, false);
 
    /*
     * 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,
                                          accessMethodId,
                                          descriptor,
                                          list_concat(cookedDefaults,
                                                      old_constraints),
                                          relkind,
                                          stmt->relation->relpersistence,
                                          false,
                                          false,
                                          stmt->oncommit,
                                          reloptions,
                                          true,
                                          allowSystemTableMods,
                                          false,
                                          InvalidOid,
                                          typaddress);
 
    /*
     * 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);
 
    /*
     * Now add any newly specified column default and generation expressions
     * 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.
     *
     * This needs to be before processing the partitioning clauses because
     * those could refer to generated columns.
     */
    if (rawDefaults)
        AddRelationNewConstraints(rel, rawDefaults, NIL,
                                  true, true, false, queryString);
 
    /*
     * Make column generation expressions visible for use by partitioning.
     */
    CommandCounterIncrement();
 
    /* Process and store partition bound, if any. */
    if (stmt->partbound)
    {
        PartitionBoundSpec *bound;
        ParseState *pstate;
        Oid         parentId = linitial_oid(inheritOids),
                    defaultPartOid;
        Relation    parent,
                    defaultRel = NULL;
        ParseNamespaceItem *nsitem;
 
        /* Already have strong enough lock on the parent */
        parent = table_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,
                                                                   true));
        if (OidIsValid(defaultPartOid))
            defaultRel = table_open(defaultPartOid, AccessExclusiveLock);
 
        /* Transform the bound values */
        pstate = make_parsestate(NULL);
        pstate->p_sourcetext = queryString;
 
        /*
         * Add an nsitem containing this relation, so that transformExpr
         * called on partition bound expressions is able to report errors
         * using a proper context.
         */
        nsitem = addRangeTableEntryForRelation(pstate, rel, AccessShareLock,
                                               NULL, false, false);
        addNSItemToQuery(pstate, nsitem, false, true, true);
 
        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, pstate);
 
        /*
         * 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_partition_contents(parent, defaultRel, bound);
            /* Keep the lock until commit. */
            table_close(defaultRel, NoLock);
        }
 
        /* Update the pg_class entry. */
        StorePartitionBound(rel, parent, bound);
 
        table_close(parent, NoLock);
    }
 
    /* Store inheritance information for new rel. */
    StoreCatalogInheritance(relationId, inheritOids, stmt->partbound != NULL);
 
    /*
     * Process the partitioning specification (if any) and store the partition
     * key information into the catalog.
     */
    if (partitioned)
    {
        ParseState *pstate;
        int         partnatts;
        AttrNumber  partattrs[PARTITION_MAX_KEYS];
        Oid         partopclass[PARTITION_MAX_KEYS];
        Oid         partcollation[PARTITION_MAX_KEYS];
        List       *partexprs = NIL;
 
        pstate = make_parsestate(NULL);
        pstate->p_sourcetext = queryString;
 
        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);
 
        ComputePartitionAttrs(pstate, rel, stmt->partspec->partParams,
                              partattrs, &partexprs, partopclass,
                              partcollation, stmt->partspec->strategy);
 
        StorePartitionKey(rel, stmt->partspec->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 = table_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);
            AttrMap    *attmap;
            IndexStmt  *idxstmt;
            Oid         constraintOid;
 
            if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
            {
                if (idxRel->rd_index->indisunique)
                    ereport(ERROR,
                            (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                             errmsg("cannot create foreign partition of partitioned table \"%s\"",
                                    RelationGetRelationName(parent)),
                             errdetail("Table \"%s\" contains indexes that are unique.",
                                       RelationGetRelationName(parent))));
                else
                {
                    index_close(idxRel, AccessShareLock);
                    continue;
                }
            }
 
            attmap = build_attrmap_by_name(RelationGetDescr(rel),
                                           RelationGetDescr(parent),
                                           false);
            idxstmt =
                generateClonedIndexStmt(NULL, idxRel,
                                        attmap, &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(NULL, parent, rel);
 
        table_close(parent, NoLock);
    }
 
    /*
     * Now add any newly specified CHECK constraints to the new relation. Same
     * as for defaults above, but these need to come after partitioning is set
     * up.
     */
    if (stmt->constraints)
        AddRelationNewConstraints(rel, NIL, stmt->constraints,
                                  true, true, false, queryString);
 
    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)
    {
        /*
         * Note that for temporary relations this lock may get upgraded later
         * on, but as no other session can access a temporary relation, this
         * is actually fine.
         */
        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.expected_relkind = relkind;
        state.heap_lockmode = drop->concurrent ?
            ShareUpdateExclusiveLock : AccessExclusiveLock;
        /* We must initialize these fields to show that no locks are held: */
        state.heapOid = InvalidOid;
        state.partParentOid = InvalidOid;
 
        relOid = RangeVarGetRelidExtended(rel, lockmode, RVR_MISSING_OK,
                                          RangeVarCallbackForDropRelation,
                                          (void *) &state);
 
        /* Not there? */
        if (!OidIsValid(relOid))
        {
            DropErrorMsgNonExistent(rel, relkind, drop->missing_ok);
            continue;
        }
 
        /*
         * Decide if concurrent mode needs to be used here or not.  The
         * callback retrieved the rel's persistence for us.
         */
        if (drop->concurrent &&
            state.actual_relpersistence != RELPERSISTENCE_TEMP)
        {
            Assert(list_length(drop->objects) == 1 &&
                   drop->removeType == OBJECT_INDEX);
            flags |= PERFORM_DELETION_CONCURRENTLY;
        }
 
        /*
         * Concurrent index drop cannot be used with partitioned indexes,
         * either.
         */
        if ((flags & PERFORM_DELETION_CONCURRENTLY) != 0 &&
            state.actual_relkind == RELKIND_PARTITIONED_INDEX)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("cannot drop partitioned index \"%s\" concurrently",
                            rel->relname)));
 
        /*
         * If we're told to drop a partitioned index, we must acquire lock on
         * all the children of its parent partitioned table before proceeding.
         * Otherwise we'd try to lock the child index partitions before their
         * tables, leading to potential deadlock against other sessions that
         * will lock those objects in the other order.
         */
        if (state.actual_relkind == RELKIND_PARTITIONED_INDEX)
            (void) find_all_inheritors(state.heapOid,
                                       state.heap_lockmode,
                                       NULL);
 
        /* 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        expected_relkind;
    bool        is_partition;
    Form_pg_class classform;
    LOCKMODE    heap_lockmode;
    bool        invalid_system_index = false;
 
    state = (struct DropRelationCallbackState *) arg;
    heap_lockmode = state->heap_lockmode;
 
    /*
     * 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;
 
    /* Pass back some data to save lookups in RemoveRelations */
    state->actual_relkind = classform->relkind;
    state->actual_relpersistence = classform->relpersistence;
 
    /*
     * 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 (state->expected_relkind != expected_relkind)
        DropErrorMsgWrongType(rel->relname, classform->relkind,
                              state->expected_relkind);
 
    /* Allow DROP to either table owner or schema owner */
    if (!object_ownercheck(RelationRelationId, relOid, GetUserId()) &&
        !object_ownercheck(NamespaceRelationId, classform->relnamespace, GetUserId()))
        aclcheck_error(ACLCHECK_NOT_OWNER,
                       get_relkind_objtype(classform->relkind),
                       rel->relname);
 
    /*
     * Check the case of a system index that might have been invalidated by a
     * failed concurrent process and allow its drop. For the time being, this
     * only concerns indexes of toast relations that became invalid during a
     * REINDEX CONCURRENTLY process.
     */
    if (IsSystemClass(relOid, classform) && classform->relkind == RELKIND_INDEX)
    {
        HeapTuple   locTuple;
        Form_pg_index indexform;
        bool        indisvalid;
 
        locTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(relOid));
        if (!HeapTupleIsValid(locTuple))
        {
            ReleaseSysCache(tuple);
            return;
        }
 
        indexform = (Form_pg_index) GETSTRUCT(locTuple);
        indisvalid = indexform->indisvalid;
        ReleaseSysCache(locTuple);
 
        /* Mark object as being an invalid index of system catalogs */
        if (!indisvalid)
            invalid_system_index = true;
    }
 
    /* In the case of an invalid index, it is fine to bypass this check */
    if (!invalid_system_index && !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.  Note that this
     * code will execute for either plain or partitioned indexes.
     */
    if (expected_relkind == RELKIND_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 if we do it the other
     * way around.
     */
    if (is_partition && relOid != oldRelOid)
    {
        state->partParentOid = get_partition_parent(relOid, true);
        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.  Note that if relations
 * are foreign tables, at this stage, we have not yet checked that their
 * foreign data in external data sources are OK for truncation.  These are
 * checked when foreign data are actually truncated later.  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;
        LOCKMODE    lockmode = AccessExclusiveLock;
 
        myrelid = RangeVarGetRelidExtended(rv, lockmode,
                                           0, RangeVarCallbackForTruncate,
                                           NULL);
 
        /* don't throw error for "TRUNCATE foo, foo" */
        if (list_member_oid(relids, myrelid))
            continue;
 
        /* open the relation, we already hold a lock on it */
        rel = table_open(myrelid, NoLock);
 
        /*
         * RangeVarGetRelidExtended() has done most checks with its callback,
         * but other checks with the now-opened Relation remain.
         */
        truncate_check_activity(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, lockmode, NULL);
 
            foreach(child, children)
            {
                Oid         childrelid = lfirst_oid(child);
 
                if (list_member_oid(relids, childrelid))
                    continue;
 
                /* find_all_inheritors already got lock */
                rel = table_open(childrelid, NoLock);
 
                /*
                 * It is possible that the parent table has children that are
                 * temp tables of other backends.  We cannot safely access
                 * such tables (because of buffering issues), and the best
                 * thing to do is to silently ignore them.  Note that this
                 * check is the same as one of the checks done in
                 * truncate_check_activity() called below, still it is kept
                 * here for simplicity.
                 */
                if (RELATION_IS_OTHER_TEMP(rel))
                {
                    table_close(rel, lockmode);
                    continue;
                }
 
                /*
                 * Inherited TRUNCATE commands perform access permission
                 * checks on the parent table only. So we skip checking the
                 * children's permissions and don't call
                 * truncate_check_perms() here.
                 */
                truncate_check_rel(RelationGetRelid(rel), rel->rd_rel);
                truncate_check_activity(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);
 
        table_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;
    HTAB       *ft_htab = NULL;
    EState     *estate;
    ResultRelInfo *resultRelInfos;
    ResultRelInfo *resultRelInfo;
    SubTransactionId mySubid;
    ListCell   *cell;
    Oid        *logrelids;
 
    /*
     * Check 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 = table_open(relid, AccessExclusiveLock);
                ereport(NOTICE,
                        (errmsg("truncate cascades to table \"%s\"",
                                RelationGetRelationName(rel))));
                truncate_check_rel(relid, rel->rd_rel);
                truncate_check_perms(relid, rel->rd_rel);
                truncate_check_activity(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));
            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 (!object_ownercheck(RelationRelationId, 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.
     *
     * We put the ResultRelInfos in the es_opened_result_relations list, even
     * though we don't have a range table and don't populate the
     * es_result_relations array.  That's a bit bogus, but it's enough to make
     * ExecGetTriggerResultRel() find them.
     */
    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);
        estate->es_opened_result_relations =
            lappend(estate->es_opened_result_relations, resultRelInfo);
        resultRelInfo++;
    }
 
    /*
     * 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)
    {
        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;
 
        /*
         * Build the lists of foreign tables belonging to each foreign server
         * and pass each list to the foreign data wrapper's callback function,
         * so that each server can truncate its all foreign tables in bulk.
         * Each list is saved as a single entry in a hash table that uses the
         * server OID as lookup key.
         */
        if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
        {
            Oid         serverid = GetForeignServerIdByRelId(RelationGetRelid(rel));
            bool        found;
            ForeignTruncateInfo *ft_info;
 
            /* First time through, initialize hashtable for foreign tables */
            if (!ft_htab)
            {
                HASHCTL     hctl;
 
                memset(&hctl, 0, sizeof(HASHCTL));
                hctl.keysize = sizeof(Oid);
                hctl.entrysize = sizeof(ForeignTruncateInfo);
                hctl.hcxt = CurrentMemoryContext;
 
                ft_htab = hash_create("TRUNCATE for Foreign Tables",
                                      32,   /* start small and extend */
                                      &hctl,
                                      HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
            }
 
            /* Find or create cached entry for the foreign table */
            ft_info = hash_search(ft_htab, &serverid, HASH_ENTER, &found);
            if (!found)
            {
                ft_info->serverid = serverid;
                ft_info->rels = NIL;
            }
 
            /*
             * Save the foreign table in the entry of the server that the
             * foreign table belongs to.
             */
            ft_info->rels = lappend(ft_info->rels, rel);
            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 relfilenumber 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_newRelfilelocatorSubid == mySubid)
        {
            /* Immediate, non-rollbackable truncation is OK */
            heap_truncate_one_rel(rel);
        }
        else
        {
            Oid         heap_relid;
            Oid         toast_relid;
            ReindexParams reindex_params = {0};
 
            /*
             * 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);
 
            /*
             * Need the full transaction-safe pushups.
             *
             * Create a new empty storage file for the relation, and assign it
             * as the relfilenumber value. The old storage file is scheduled
             * for deletion at commit.
             */
            RelationSetNewRelfilenumber(rel, rel->rd_rel->relpersistence);
 
            heap_relid = RelationGetRelid(rel);
 
            /*
             * The same for the toast table, if any.
             */
            toast_relid = rel->rd_rel->reltoastrelid;
            if (OidIsValid(toast_relid))
            {
                Relation    toastrel = relation_open(toast_relid,
                                                     AccessExclusiveLock);
 
                RelationSetNewRelfilenumber(toastrel,
                                            toastrel->rd_rel->relpersistence);
                table_close(toastrel, NoLock);
            }
 
            /*
             * Reconstruct the indexes to match, and we're done.
             */
            reindex_relation(heap_relid, REINDEX_REL_PROCESS_TOAST,
                             &reindex_params);
        }
 
        pgstat_count_truncate(rel);
    }
 
    /* Now go through the hash table, and truncate foreign tables */
    if (ft_htab)
    {
        ForeignTruncateInfo *ft_info;
        HASH_SEQ_STATUS seq;
 
        hash_seq_init(&seq, ft_htab);
 
        PG_TRY();
        {
            while ((ft_info = hash_seq_search(&seq)) != NULL)
            {
                FdwRoutine *routine = GetFdwRoutineByServerId(ft_info->serverid);
 
                /* truncate_check_rel() has checked that already */
                Assert(routine->ExecForeignTruncate != NULL);
 
                routine->ExecForeignTruncate(ft_info->rels,
                                             behavior,
                                             restart_seqs);
            }
        }
        PG_FINALLY();
        {
            hash_destroy(ft_htab);
        }
        PG_END_TRY();
    }
 
    /*
     * 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 (relids_logged != NIL)
    {
        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)
    {
        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);
 
        table_close(rel, NoLock);
    }
}
 
/*
 * Check that a given relation is safe to truncate.  Subroutine for
 * ExecuteTruncate() and RangeVarCallbackForTruncate().
 */
static void
truncate_check_rel(Oid relid, Form_pg_class reltuple)
{
    char       *relname = NameStr(reltuple->relname);
 
    /*
     * Only allow truncate on regular tables, foreign tables using foreign
     * data wrappers supporting TRUNCATE and partitioned tables (although, the
     * latter are only being included here for the following checks; no
     * physical truncation will occur in their case.).
     */
    if (reltuple->relkind == RELKIND_FOREIGN_TABLE)
    {
        Oid         serverid = GetForeignServerIdByRelId(relid);
        FdwRoutine *fdwroutine = GetFdwRoutineByServerId(serverid);
 
        if (!fdwroutine->ExecForeignTruncate)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("cannot truncate foreign table \"%s\"",
                            relname)));
    }
    else if (reltuple->relkind != RELKIND_RELATION &&
             reltuple->relkind != RELKIND_PARTITIONED_TABLE)
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("\"%s\" is not a table", relname)));
 
    /*
     * Most system catalogs can't be truncated at all, or at least not unless
     * allow_system_table_mods=on. As an exception, however, we allow
     * pg_largeobject to be truncated as part of pg_upgrade, because we need
     * to change its relfilenode to match the old cluster, and allowing a
     * TRUNCATE command to be executed is the easiest way of doing that.
     */
    if (!allowSystemTableMods && IsSystemClass(relid, reltuple)
        && (!IsBinaryUpgrade || relid != LargeObjectRelationId))
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("permission denied: \"%s\" is a system catalog",
                        relname)));
 
    InvokeObjectTruncateHook(relid);
}
 
/*
 * Check that current user has the permission to truncate given relation.
 */
static void
truncate_check_perms(Oid relid, Form_pg_class reltuple)
{
    char       *relname = NameStr(reltuple->relname);
    AclResult   aclresult;
 
    /* Permissions checks */
    aclresult = pg_class_aclcheck(relid, GetUserId(), ACL_TRUNCATE);
    if (aclresult != ACLCHECK_OK)
        aclcheck_error(aclresult, get_relkind_objtype(reltuple->relkind),
                       relname);
}
 
/*
 * Set of extra sanity checks to check if a given relation is safe to
 * truncate.  This is split with truncate_check_rel() as
 * RangeVarCallbackForTruncate() cannot open a Relation yet.
 */
static void
truncate_check_activity(Relation 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 TYPSTORAGE_PLAIN:
            return "PLAIN";
        case TYPSTORAGE_EXTERNAL:
            return "EXTERNAL";
        case TYPSTORAGE_EXTENDED:
            return "EXTENDED";
        case TYPSTORAGE_MAIN:
            return "MAIN";
        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 OIDs of parent relations, already locked by caller.
 * 'relpersistence' is the persistence type of the table.
 * 'is_partition' tells if the table is a partition.
 *
 * Output arguments:
 * 'supconstr' receives a list of constraints belonging to the parents,
 *      updated as necessary to be valid for the child.
 *
 * 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.
 *
 *      Note that the default-value infrastructure is used for generated
 *      columns' expressions too, so most of the preceding paragraph applies
 *      to generation expressions too.  We insist that a child column be
 *      generated if and only if its parent(s) are, but it need not have
 *      the same generation expression.
 *----------
 */
static List *
MergeAttributes(List *schema, List *supers, char relpersistence,
                bool is_partition, List **supconstr)
{
    List       *inhSchema = NIL;
    List       *constraints = NIL;
    bool        have_bogus_defaults = false;
    int         child_attno;
    static Node bogus_marker = {0}; /* marks conflicting defaults */
    List       *saved_schema = NIL;
    ListCell   *entry;
 
    /*
     * 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)));
 
    /*
     * 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.
     *
     * We don't use foreach() here because we have two nested loops over the
     * schema list, with possible element deletions in the inner one.  If we
     * used foreach_delete_current() it could only fix up the state of one of
     * the loops, so it seems cleaner to use looping over list indexes for
     * both loops.  Note that any deletion will happen beyond where the outer
     * loop is, so its index never needs adjustment.
     */
    for (int coldefpos = 0; coldefpos < list_length(schema); coldefpos++)
    {
        ColumnDef  *coldef = list_nth_node(ColumnDef, schema, coldefpos);
 
        if (!is_partition && 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.  (We omit this check for partition column
             * lists; those are processed separately below.)
             */
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_COLUMN),
                     errmsg("column \"%s\" does not exist",
                            coldef->colname)));
        }
 
        /* restpos scans all entries beyond coldef; incr is in loop body */
        for (int restpos = coldefpos + 1; restpos < list_length(schema);)
        {
            ColumnDef  *restdef = list_nth_node(ColumnDef, schema, restpos);
 
            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;
                    schema = list_delete_nth_cell(schema, restpos);
                }
                else
                    ereport(ERROR,
                            (errcode(ERRCODE_DUPLICATE_COLUMN),
                             errmsg("column \"%s\" specified more than once",
                                    coldef->colname)));
            }
            else
                restpos++;
        }
    }
 
    /*
     * In case of a partition, there are no new column definitions, only dummy
     * ColumnDefs created for column constraints.  Set them aside for now and
     * process them at the end.
     */
    if (is_partition)
    {
        saved_schema = schema;
        schema = NIL;
    }
 
    /*
     * 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)
    {
        Oid         parent = lfirst_oid(entry);
        Relation    relation;
        TupleDesc   tupleDesc;
        TupleConstr *constr;
        AttrMap    *newattmap;
        List       *inherited_defaults;
        List       *cols_with_defaults;
        AttrNumber  parent_attno;
        ListCell   *lc1;
        ListCell   *lc2;
 
        /* caller already got lock */
        relation = table_open(parent, NoLock);
 
        /*
         * Check for active uses of the parent partitioned table in the
         * current transaction, such as being used in some manner by an
         * enclosing command.
         */
        if (is_partition)
            CheckTableNotInUse(relation, "CREATE TABLE .. PARTITION OF");
 
        /*
         * 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\"",
                            RelationGetRelationName(relation))));
        if (relation->rd_rel->relispartition && !is_partition)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg("cannot inherit from partition \"%s\"",
                            RelationGetRelationName(relation))));
 
        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",
                            RelationGetRelationName(relation))));
 
        /*
         * If the parent is permanent, so must be all of its partitions.  Note
         * that inheritance allows that case.
         */
        if (is_partition &&
            relation->rd_rel->relpersistence != RELPERSISTENCE_TEMP &&
            relpersistence == RELPERSISTENCE_TEMP)
            ereport(ERROR,
                    (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                     errmsg("cannot create a temporary relation as partition of permanent relation \"%s\"",
                            RelationGetRelationName(relation))));
 
        /* 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\"",
                            RelationGetRelationName(relation))));
 
        /* 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 (!object_ownercheck(RelationRelationId, RelationGetRelid(relation), GetUserId()))
            aclcheck_error(ACLCHECK_NOT_OWNER, get_relkind_objtype(relation->rd_rel->relkind),
                           RelationGetRelationName(relation));
 
        tupleDesc = RelationGetDescr(relation);
        constr = tupleDesc->constr;
 
        /*
         * newattmap->attnums[] 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.)
         */
        newattmap = make_attrmap(tupleDesc->natts);
 
        /* We can't process inherited defaults until newattmap is complete. */
        inherited_defaults = cols_with_defaults = NIL;
 
        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 newattmap->attnums 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.
                 */
                ereport(NOTICE,
                        (errmsg("merging multiple inherited definitions of column \"%s\"",
                                attributeName)));
                def = (ColumnDef *) list_nth(inhSchema, exist_attno - 1);
 
                /*
                 * Must have the same type and typmod
                 */
                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))));
 
                /*
                 * Must have the same collation
                 */
                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/check 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))));
 
                /*
                 * Copy/check compression parameter
                 */
                if (CompressionMethodIsValid(attribute->attcompression))
                {
                    const char *compression =
                    GetCompressionMethodName(attribute->attcompression);
 
                    if (def->compression == NULL)
                        def->compression = pstrdup(compression);
                    else if (strcmp(def->compression, compression) != 0)
                        ereport(ERROR,
                                (errcode(ERRCODE_DATATYPE_MISMATCH),
                                 errmsg("column \"%s\" has a compression method conflict",
                                        attributeName),
                                 errdetail("%s versus %s", def->compression, compression)));
                }
 
                /*
                 * Merge of NOT NULL constraints = OR 'em together
                 */
                def->is_not_null |= attribute->attnotnull;
 
                /*
                 * Check for GENERATED conflicts
                 */
                if (def->generated != attribute->attgenerated)
                    ereport(ERROR,
                            (errcode(ERRCODE_DATATYPE_MISMATCH),
                             errmsg("inherited column \"%s\" has a generation conflict",
                                    attributeName)));
 
                /*
                 * Default and other constraints are handled below
                 */
 
                def->inhcount++;
 
                newattmap->attnums[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->storage = attribute->attstorage;
                def->raw_default = NULL;
                def->cooked_default = NULL;
                def->generated = attribute->attgenerated;
                def->collClause = NULL;
                def->collOid = attribute->attcollation;
                def->constraints = NIL;
                def->location = -1;
                if (CompressionMethodIsValid(attribute->attcompression))
                    def->compression =
                        pstrdup(GetCompressionMethodName(attribute->attcompression));
                else
                    def->compression = NULL;
                inhSchema = lappend(inhSchema, def);
                newattmap->attnums[parent_attno - 1] = ++child_attno;
            }
 
            /*
             * Locate default/generation expression if any
             */
            if (attribute->atthasdef)
            {
                Node       *this_default = NULL;
 
                /* Find default in constraint structure */
                if (constr != NULL)
                {
                    AttrDefault *attrdef = constr->defval;
 
                    for (int i = 0; i < constr->num_defval; i++)
                    {
                        if (attrdef[i].adnum == parent_attno)
                        {
                            this_default = stringToNode(attrdef[i].adbin);
                            break;
                        }
                    }
                }
                if (this_default == NULL)
                    elog(ERROR, "default expression not found for attribute %d of relation \"%s\"",
                         parent_attno, RelationGetRelationName(relation));
 
                /*
                 * If it's a GENERATED default, it might contain Vars that
                 * need to be mapped to the inherited column(s)' new numbers.
                 * We can't do that till newattmap is ready, so just remember
                 * all the inherited default expressions for the moment.
                 */
                inherited_defaults = lappend(inherited_defaults, this_default);
                cols_with_defaults = lappend(cols_with_defaults, def);
            }
        }
 
        /*
         * Now process any inherited default expressions, adjusting attnos
         * using the completed newattmap map.
         */
        forboth(lc1, inherited_defaults, lc2, cols_with_defaults)
        {
            Node       *this_default = (Node *) lfirst(lc1);
            ColumnDef  *def = (ColumnDef *) lfirst(lc2);
            bool        found_whole_row;
 
            /* Adjust Vars to match new table's column numbering */
            this_default = map_variable_attnos(this_default,
                                               1, 0,
                                               newattmap,
                                               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.  (A variable could only appear in a
             * generation expression, so the error message is correct.)
             */
            if (found_whole_row)
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot convert whole-row table reference"),
                         errdetail("Generation expression for column \"%s\" contains a whole-row reference to table \"%s\".",
                                   def->colname,
                                   RelationGetRelationName(relation))));
 
            /*
             * 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 newattmap 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,
                                           newattmap,
                                           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);
                }
            }
        }
 
        free_attrmap(newattmap);
 
        /*
         * 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.
         */
        table_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.
                 */
                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);
 
                /*
                 * Must have the same type and typmod
                 */
                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))));
 
                /*
                 * Must have the same collation
                 */
                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))));
 
                /*
                 * Copy compression parameter
                 */
                if (def->compression == NULL)
                    def->compression = newdef->compression;
                else if (newdef->compression != NULL)
                {
                    if (strcmp(def->compression, newdef->compression) != 0)
                        ereport(ERROR,
                                (errcode(ERRCODE_DATATYPE_MISMATCH),
                                 errmsg("column \"%s\" has a compression method conflict",
                                        attributeName),
                                 errdetail("%s versus %s", def->compression, newdef->compression)));
                }
 
                /*
                 * Merge of NOT NULL constraints = OR 'em together
                 */
                def->is_not_null |= newdef->is_not_null;
 
                /*
                 * Check for conflicts related to generated columns.
                 *
                 * If the parent column is generated, the child column will be
                 * made a generated column if it isn't already.  If it is a
                 * generated column, we'll take its generation expression in
                 * preference to the parent's.  We must check that the child
                 * column doesn't specify a default value or identity, which
                 * matches the rules for a single column in parse_util.c.
                 *
                 * Conversely, if the parent column is not generated, the
                 * child column can't be either.  (We used to allow that, but
                 * it results in being able to override the generation
                 * expression via UPDATEs through the parent.)
                 */
                if (def->generated)
                {
                    if (newdef->raw_default && !newdef->generated)
                        ereport(ERROR,
                                (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                                 errmsg("column \"%s\" inherits from generated column but specifies default",
                                        def->colname)));
                    if (newdef->identity)
                        ereport(ERROR,
                                (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                                 errmsg("column \"%s\" inherits from generated column but specifies identity",
                                        def->colname)));
                }
                else
                {
                    if (newdef->generated)
                        ereport(ERROR,
                                (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                                 errmsg("child column \"%s\" specifies generation expression",
                                        def->colname),
                                 errhint("A child table column cannot be generated unless its parent column is.")));
                }
 
                /*
                 * 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;
                }
 
                /* Mark the column as locally defined */
                def->is_local = true;
            }
            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 parent's NOT NULL constraints and
     * defaults into each corresponding column definition.
     */
    if (is_partition)
    {
        foreach(entry, saved_schema)
        {
            ColumnDef  *restdef = lfirst(entry);
            bool        found = false;
            ListCell   *l;
 
            foreach(l, schema)
            {
                ColumnDef  *coldef = lfirst(l);
 
                if (strcmp(coldef->colname, restdef->colname) == 0)
                {
                    found = true;
                    coldef->is_not_null |= restdef->is_not_null;
 
                    /*
                     * Check for conflicts related to generated columns.
                     *
                     * Same rules as above: generated-ness has to match the
                     * parent, but the contents of the generation expression
                     * can be different.
                     */
                    if (coldef->generated)
                    {
                        if (restdef->raw_default && !restdef->generated)
                            ereport(ERROR,
                                    (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                                     errmsg("column \"%s\" inherits from generated column but specifies default",
                                            restdef->colname)));
                        if (restdef->identity)
                            ereport(ERROR,
                                    (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                                     errmsg("column \"%s\" inherits from generated column but specifies identity",
                                            restdef->colname)));
                    }
                    else
                    {
                        if (restdef->generated)
                            ereport(ERROR,
                                    (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                                     errmsg("child column \"%s\" specifies generation expression",
                                            restdef->colname),
                                     errhint("A child table column cannot be generated unless its parent column is.")));
                    }
 
                    /*
                     * Override the parent's default value for this column
                     * (coldef->cooked_default) with the partition's local
                     * definition (restdef->raw_default), if there's one. It
                     * should be physically impossible to get a cooked default
                     * in the local definition or a raw default in the
                     * inherited definition, but make sure they're nulls, for
                     * future-proofing.
                     */
                    Assert(restdef->cooked_default == NULL);
                    Assert(coldef->raw_default == NULL);
                    if (restdef->raw_default)
                    {
                        coldef->raw_default = restdef->raw_default;
                        coldef->cooked_default = NULL;
                    }
                }
            }
 
            /* complain for constraints on columns not in parent */
            if (!found)
                ereport(ERROR,
                        (errcode(ERRCODE_UNDEFINED_COLUMN),
                         errmsg("column \"%s\" does not exist",
                                restdef->colname)));
        }
    }
 
    /*
     * 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)
            {
                if (def->generated)
                    ereport(ERROR,
                            (errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
                             errmsg("column \"%s\" inherits conflicting generation expressions",
                                    def->colname),
                             errhint("To resolve the conflict, specify a generation expression explicitly.")));
                else
                    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.")));
            }
        }
    }
 
    *supconstr = constraints;
    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
     */
    Assert(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 = table_open(InheritsRelationId, RowExclusiveLock);
 
    seqNumber = 1;
    foreach(entry, supers)
    {
        Oid         parentOid = lfirst_oid(entry);
 
        StoreCatalogInheritance1(relationId, parentOid, seqNumber, relation,
                                 child_is_partition);
        seqNumber++;
    }
 
    table_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 = table_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);
    table_close(relationRelation, RowExclusiveLock);
}
 
/*
 * CheckRelationTableSpaceMove
 *      Check if relation can be moved to new tablespace.
 *
 * NOTE: The caller must hold AccessExclusiveLock on the relation.
 *
 * Returns true if the relation can be moved to the new tablespace; raises
 * an error if it is not possible to do the move; returns false if the move
 * would have no effect.
 */
bool
CheckRelationTableSpaceMove(Relation rel, Oid newTableSpaceId)
{
    Oid         oldTableSpaceId;
 
    /*
     * No work if no change in tablespace.  Note that MyDatabaseTableSpace is
     * stored as 0.
     */
    oldTableSpaceId = rel->rd_rel->reltablespace;
    if (newTableSpaceId == oldTableSpaceId ||
        (newTableSpaceId == MyDatabaseTableSpace && oldTableSpaceId == 0))
        return false;
 
    /*
     * We cannot support moving mapped relations into different tablespaces.
     * (In particular this eliminates all shared catalogs.)
     */
    if (RelationIsMapped(rel))
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot move system relation \"%s\"",
                        RelationGetRelationName(rel))));
 
    /* Cannot move a non-shared relation into pg_global */
    if (newTableSpaceId == GLOBALTABLESPACE_OID)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("only shared relations can be placed in pg_global tablespace")));
 
    /*
     * Do not allow moving 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 move temporary tables of other sessions")));
 
    return true;
}
 
/*
 * SetRelationTableSpace
 *      Set new reltablespace and relfilenumber in pg_class entry.
 *
 * newTableSpaceId is the new tablespace for the relation, and
 * newRelFilenumber its new filenumber.  If newRelFilenumber is
 * InvalidRelFileNumber, this field is not updated.
 *
 * NOTE: The caller must hold AccessExclusiveLock on the relation.
 *
 * The caller of this routine had better check if a relation can be
 * moved to this new tablespace by calling CheckRelationTableSpaceMove()
 * first, and is responsible for making the change visible with
 * CommandCounterIncrement().
 */
void
SetRelationTableSpace(Relation rel,
                      Oid newTableSpaceId,
                      RelFileNumber newRelFilenumber)
{
    Relation    pg_class;
    HeapTuple   tuple;
    Form_pg_class rd_rel;
    Oid         reloid = RelationGetRelid(rel);
 
    Assert(CheckRelationTableSpaceMove(rel, newTableSpaceId));
 
    /* Get a modifiable copy of the relation's pg_class row. */
    pg_class = table_open(RelationRelationId, RowExclusiveLock);
 
    tuple = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(reloid));
    if (!HeapTupleIsValid(tuple))
        elog(ERROR, "cache lookup failed for relation %u", reloid);
    rd_rel = (Form_pg_class) GETSTRUCT(tuple);
 
    /* Update the pg_class row. */
    rd_rel->reltablespace = (newTableSpaceId == MyDatabaseTableSpace) ?
        InvalidOid : newTableSpaceId;
    if (RelFileNumberIsValid(newRelFilenumber))
        rd_rel->relfilenode = newRelFilenumber;
    CatalogTupleUpdate(pg_class, &tuple->t_self, tuple);
 
    /*
     * Record dependency on tablespace.  This is only required for relations
     * that have no physical storage.
     */
    if (!RELKIND_HAS_STORAGE(rel->rd_rel->relkind))
        changeDependencyOnTablespace(RelationRelationId, reloid,
                                     rd_rel->reltablespace);
 
    heap_freetuple(tuple);
    table_close(pg_class, 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("cannot rename columns of relation \"%s\"",
                        NameStr(classform->relname)),
                 errdetail_relkind_not_supported(relkind)));
 
    /*
     * permissions checking.  only the owner of a class can change its schema.
     */
    if (!object_ownercheck(RelationRelationId, 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 = table_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);
 
    table_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;
 
    Assert(!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, true);
    else
        RenameConstraintById(constraintOid, newconname);
 
    ObjectAddressSet(address, ConstraintRelationId, constraintOid);
 
    ReleaseSysCache(tuple);
 
    if (targetrelation)
    {
        /*
         * Invalidate relcache so as others can see the new constraint name.
         */
        CacheInvalidateRelcache(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 = table_open(TypeRelationId, RowExclusiveLock);
        tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typid));
        if (!HeapTupleIsValid(tup))
            elog(ERROR, "cache lookup failed for type %u", typid);
        checkDomainOwner(tup);
        ReleaseSysCache(tup);
        table_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)
{
    bool        is_index_stmt = stmt->renameType == OBJECT_INDEX;
    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.  However, because ALTER INDEX can be used with any relation
     * type, we mustn't believe without verification.
     */
    for (;;)
    {
        LOCKMODE    lockmode;
        char        relkind;
        bool        obj_is_index;
 
        lockmode = is_index_stmt ? ShareUpdateExclusiveLock : AccessExclusiveLock;
 
        relid = RangeVarGetRelidExtended(stmt->relation, lockmode,
                                         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;
        }
 
        /*
         * We allow mismatched statement and object types (e.g., ALTER INDEX
         * to rename a table), but we might've used the wrong lock level.  If
         * that happens, retry with the correct lock level.  We don't bother
         * if we already acquired AccessExclusiveLock with an index, however.
         */
        relkind = get_rel_relkind(relid);
        obj_is_index = (relkind == RELKIND_INDEX ||
                        relkind == RELKIND_PARTITIONED_INDEX);
        if (obj_is_index || is_index_stmt == obj_is_index)
            break;
 
        UnlockRelationOid(relid, lockmode);
        is_index_stmt = obj_is_index;
    }
 
    /* Do the work */
    RenameRelationInternal(relid, stmt->newname, false, is_index_stmt);
 
    ObjectAddressSet(address, RelationRelationId, relid);
 
    return address;
}
 
/*
 *      RenameRelationInternal - change the name of a relation
 */
void
RenameRelationInternal(Oid myrelid, const char *newrelname, bool is_internal, bool is_index)
{
    Relation    targetrelation;
    Relation    relrelation;    /* for RELATION relation */
    HeapTuple   reltup;
    Form_pg_class relform;
    Oid         namespaceId;
 
    /*
     * Grab a lock on the target relation, which we will NOT release until end
     * of transaction.  We need at least a self-exclusive lock so that
     * concurrent DDL doesn't overwrite the rename if they start updating
     * while still seeing the old version.  The lock also guards against
     * triggering relcache reloads in concurrent sessions, which might not
     * handle this information changing under them.  For indexes, we can use a
     * reduced lock level because RelationReloadIndexInfo() handles indexes
     * specially.
     */
    targetrelation = relation_open(myrelid, is_index ? ShareUpdateExclusiveLock : AccessExclusiveLock);
    namespaceId = RelationGetNamespace(targetrelation);
 
    /*
     * Find relation's pg_class tuple, and make sure newrelname isn't in use.
     */
    relrelation = table_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)));
 
    /*
     * RenameRelation is careful not to believe the caller's idea of the
     * relation kind being handled.  We don't have to worry about this, but
     * let's not be totally oblivious to it.  We can process an index as
     * not-an-index, but not the other way around.
     */
    Assert(!is_index ||
           is_index == (targetrelation->rd_rel->relkind == RELKIND_INDEX ||
                        targetrelation->rd_rel->relkind == RELKIND_PARTITIONED_INDEX));
 
    /*
     * 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);
    table_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 lock!
     */
    relation_close(targetrelation, NoLock);
}
 
/*
 *      ResetRelRewrite - reset relrewrite
 */
void
ResetRelRewrite(Oid myrelid)
{
    Relation    relrelation;    /* for RELATION relation */
    HeapTuple   reltup;
    Form_pg_class relform;
 
    /*
     * Find relation's pg_class tuple.
     */
    relrelation = table_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);
 
    /*
     * Update pg_class tuple.
     */
    relform->relrewrite = InvalidOid;
 
    CatalogTupleUpdate(relrelation, &reltup->t_self, reltup);
 
    heap_freetuple(reltup);
    table_close(relrelation, RowExclusiveLock);
}
 
/*
 * 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. Validate and transform subcommands, and 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.  Because earlier subcommands may change
 * the catalog state seen by later commands, there are limits to what can
 * be done in this phase.  Generally, this phase acquires table locks,
 * checks permissions and relkind, and recurses to find child tables.
 *
 * ATRewriteCatalogs performs phase 2 for each affected table.
 * 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.
 *
 * For most subcommand types, phases 2 and 3 do no explicit recursion,
 * since phase 1 already does it.  However, for certain subcommand types
 * it is only possible to determine how to recurse at phase 2 time; for
 * those cases, phase 1 sets the cmd->recurse flag.
 *
 * 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.
 *
 * The caller also provides a "context" which is to be passed back to
 * utility.c when we need to execute a subcommand such as CREATE INDEX.
 * Some of the fields therein, such as the relid, are used here as well.
 */
void
AlterTable(AlterTableStmt *stmt, LOCKMODE lockmode,
           AlterTableUtilityContext *context)
{
    Relation    rel;
 
    /* Caller is required to provide an adequate lock. */
    rel = relation_open(context->relid, NoLock);
 
    CheckTableNotInUse(rel, "ALTER TABLE");
 
    ATController(stmt, rel, stmt->cmds, stmt->relation->inh, lockmode, context);
}
 
/*
 * 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.
 *
 * Also, since we don't have an AlterTableUtilityContext, this cannot be
 * used for any subcommand types that require parse transformation or
 * could generate subcommands that have to be passed to ProcessUtility.
 */
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, NULL);
}
 
/*
 * 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 primary
 * 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_SetAccessMethod:    /* must rewrite heap */
            case AT_SetTableSpace:  /* must rewrite heap */
            case AT_AlterColumnType:    /* 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
                 * optimized assuming the constraint holds true. See also
                 * CloneFkReferenced.
                 */
            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:   /* used to equiv to 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_CookedColumnDefault:
            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:
            case AT_DropExpression:
            case AT_SetCompression:
                cmd_lockmode = AccessExclusiveLock;
                break;
 
            case 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:
                cmd_lockmode = ShareUpdateExclusiveLock;
                break;
 
            case AT_DetachPartition:
                if (((PartitionCmd *) cmd->def)->concurrent)
                    cmd_lockmode = ShareUpdateExclusiveLock;
                else
                    cmd_lockmode = AccessExclusiveLock;
                break;
 
            case AT_DetachPartitionFinalize:
                cmd_lockmode = ShareUpdateExclusiveLock;
                break;
 
            case AT_CheckNotNull:
 
                /*
                 * This only examines the table's schema; but lock must be
                 * strong enough to prevent concurrent DROP NOT NULL.
                 */
                cmd_lockmode = AccessShareLock;
                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,
             AlterTableUtilityContext *context)
{
    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, context);
    }
 
    /* Close the relation, but keep lock until commit */
    relation_close(rel, NoLock);
 
    /* Phase 2: update system catalogs */
    ATRewriteCatalogs(&wqueue, lockmode, context);
 
    /* Phase 3: scan/rewrite tables as needed, and run afterStmts */
    ATRewriteTables(parsetree, &wqueue, lockmode, context);
}
 
/*
 * 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,
          AlterTableUtilityContext *context)
{
    AlteredTableInfo *tab;
    int         pass = AT_PASS_UNSET;
 
    /* Find or create work queue entry for this table */
    tab = ATGetQueueEntry(wqueue, rel);
 
    /*
     * Disallow any ALTER TABLE other than ALTER TABLE DETACH FINALIZE on
     * partitions that are pending detach.
     */
    if (rel->rd_rel->relispartition &&
        cmd->subtype != AT_DetachPartitionFinalize &&
        PartitionHasPendingDetach(RelationGetRelid(rel)))
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("cannot alter partition \"%s\" with an incomplete detach",
                       RelationGetRelationName(rel)),
                errhint("Use ALTER TABLE ... DETACH PARTITION ... FINALIZE to complete the pending detach operation."));
 
    /*
     * Copy the original subcommand for each table, so we can scribble on it.
     * 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 and relkind checking, recursion to child tables if
     * needed, and any additional phase-1 processing needed.  (But beware of
     * adding any processing that looks at table details that another
     * subcommand could change.  In some cases we reject multiple subcommands
     * that could try to change the same state in contrary ways.)
     */
    switch (cmd->subtype)
    {
        case AT_AddColumn:      /* ADD COLUMN */
            ATSimplePermissions(cmd->subtype, rel,
                                ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
            ATPrepAddColumn(wqueue, rel, recurse, recursing, false, cmd,
                            lockmode, context);
            /* Recursion occurs during execution phase */
            pass = AT_PASS_ADD_COL;
            break;
        case AT_AddColumnToView:    /* add column via CREATE OR REPLACE VIEW */
            ATSimplePermissions(cmd->subtype, rel, ATT_VIEW);
            ATPrepAddColumn(wqueue, rel, recurse, recursing, true, cmd,
                            lockmode, context);
            /* 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(cmd->subtype, rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context);
            /* No command-specific prep needed */
            pass = cmd->def ? AT_PASS_ADD_OTHERCONSTR : AT_PASS_DROP;
            break;
        case AT_CookedColumnDefault:    /* add a pre-cooked default */
            /* This is currently used only in CREATE TABLE */
            /* (so the permission check really isn't necessary) */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            pass = AT_PASS_ADD_OTHERCONSTR;
            break;
        case AT_AddIdentity:
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            pass = AT_PASS_ADD_OTHERCONSTR;
            break;
        case AT_SetIdentity:
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            /* This should run after AddIdentity, so do it in MISC pass */
            pass = AT_PASS_MISC;
            break;
        case AT_DropIdentity:
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_VIEW | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            pass = AT_PASS_DROP;
            break;
        case AT_DropNotNull:    /* ALTER COLUMN DROP NOT NULL */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            ATPrepDropNotNull(rel, recurse, recursing);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context);
            pass = AT_PASS_DROP;
            break;
        case AT_SetNotNull:     /* ALTER COLUMN SET NOT NULL */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Need command-specific recursion decision */
            ATPrepSetNotNull(wqueue, rel, cmd, recurse, recursing,
                             lockmode, context);
            pass = AT_PASS_COL_ATTRS;
            break;
        case AT_CheckNotNull:   /* check column is already marked NOT NULL */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context);
            /* No command-specific prep needed */
            pass = AT_PASS_COL_ATTRS;
            break;
        case AT_DropExpression: /* ALTER COLUMN DROP EXPRESSION */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context);
            ATPrepDropExpression(rel, cmd, recurse, recursing, lockmode);
            pass = AT_PASS_DROP;
            break;
        case AT_SetStatistics:  /* ALTER COLUMN SET STATISTICS */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_MATVIEW | ATT_INDEX | ATT_PARTITIONED_INDEX | ATT_FOREIGN_TABLE);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_SetOptions:     /* ALTER COLUMN SET ( options ) */
        case AT_ResetOptions:   /* ALTER COLUMN RESET ( options ) */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_MATVIEW | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            pass = AT_PASS_MISC;
            break;
        case AT_SetStorage:     /* ALTER COLUMN SET STORAGE */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_MATVIEW | ATT_FOREIGN_TABLE);
            ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode, context);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_SetCompression: /* ALTER COLUMN SET COMPRESSION */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_MATVIEW);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_DropColumn:     /* DROP COLUMN */
            ATSimplePermissions(cmd->subtype, rel,
                                ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
            ATPrepDropColumn(wqueue, rel, recurse, recursing, cmd,
                             lockmode, context);
            /* Recursion occurs during execution phase */
            pass = AT_PASS_DROP;
            break;
        case AT_AddIndex:       /* ADD INDEX */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_ADD_INDEX;
            break;
        case AT_AddConstraint:  /* ADD CONSTRAINT */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Recursion occurs during execution phase */
            /* No command-specific prep needed except saving recurse flag */
            if (recurse)
                cmd->recurse = true;
            pass = AT_PASS_ADD_CONSTR;
            break;
        case AT_AddIndexConstraint: /* ADD CONSTRAINT USING INDEX */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE);
            /* This command never recurses */
            /* No command-specific prep needed */
            pass = AT_PASS_ADD_INDEXCONSTR;
            break;
        case AT_DropConstraint: /* DROP CONSTRAINT */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            ATCheckPartitionsNotInUse(rel, lockmode);
            /* Other recursion occurs during execution phase */
            /* No command-specific prep needed except saving recurse flag */
            if (recurse)
                cmd->recurse = true;
            pass = AT_PASS_DROP;
            break;
        case AT_AlterColumnType:    /* ALTER COLUMN TYPE */
            ATSimplePermissions(cmd->subtype, rel,
                                ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
            /* See comments for ATPrepAlterColumnType */
            cmd = ATParseTransformCmd(wqueue, tab, rel, cmd, recurse, lockmode,
                                      AT_PASS_UNSET, context);
            Assert(cmd != NULL);
            /* Performs own recursion */
            ATPrepAlterColumnType(wqueue, tab, rel, recurse, recursing, cmd,
                                  lockmode, context);
            pass = AT_PASS_ALTER_TYPE;
            break;
        case AT_AlterColumnGenericOptions:
            ATSimplePermissions(cmd->subtype, 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(cmd->subtype, 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(cmd->subtype, rel, ATT_TABLE | ATT_SEQUENCE);
            if (tab->chgPersistence)
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot change persistence setting twice")));
            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(cmd->subtype, rel, ATT_TABLE | ATT_SEQUENCE);
            if (tab->chgPersistence)
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot change persistence setting twice")));
            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_DropOids:       /* SET WITHOUT OIDS */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            pass = AT_PASS_DROP;
            break;
        case AT_SetAccessMethod:    /* SET ACCESS METHOD */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_MATVIEW);
 
            /* partitioned tables don't have an access method */
            if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
                ereport(ERROR,
                        (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                         errmsg("cannot change access method of a partitioned table")));
 
            /* check if another access method change was already requested */
            if (OidIsValid(tab->newAccessMethod))
                ereport(ERROR,
                        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                         errmsg("cannot have multiple SET ACCESS METHOD subcommands")));
 
            ATPrepSetAccessMethod(tab, rel, cmd->name);
            pass = AT_PASS_MISC;    /* does not matter; no work in Phase 2 */
            break;
        case AT_SetTableSpace:  /* SET TABLESPACE */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_MATVIEW | ATT_INDEX |
                                ATT_PARTITIONED_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(cmd->subtype, 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(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* This command never recurses */
            ATPrepAddInherit(rel);
            pass = AT_PASS_MISC;
            break;
        case AT_DropInherit:    /* NO INHERIT */
            ATSimplePermissions(cmd->subtype, 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(cmd->subtype, rel, ATT_TABLE);
            /* Recursion occurs during execution phase */
            pass = AT_PASS_MISC;
            break;
        case AT_ValidateConstraint: /* VALIDATE CONSTRAINT */
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Recursion occurs during execution phase */
            /* No command-specific prep needed except saving recurse flag */
            if (recurse)
                cmd->recurse = true;
            pass = AT_PASS_MISC;
            break;
        case AT_ReplicaIdentity:    /* REPLICA IDENTITY ... */
            ATSimplePermissions(cmd->subtype, 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(cmd->subtype, rel, ATT_TABLE | ATT_FOREIGN_TABLE);
            /* Set up recursion for phase 2; no other prep needed */
            if (recurse)
                cmd->recurse = true;
            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(cmd->subtype, rel, ATT_TABLE);
            /* These commands never recurse */
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_GenericOptions:
            ATSimplePermissions(cmd->subtype, rel, ATT_FOREIGN_TABLE);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_AttachPartition:
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE | ATT_PARTITIONED_INDEX);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_DetachPartition:
            ATSimplePermissions(cmd->subtype, rel, ATT_TABLE);
            /* No command-specific prep needed */
            pass = AT_PASS_MISC;
            break;
        case AT_DetachPartitionFinalize:
            ATSimplePermissions(cmd->subtype, 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,
                  AlterTableUtilityContext *context)
{
    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];
            ListCell   *lcmd;
 
            if (subcmds == NIL)
                continue;
 
            /*
             * Open the relation and store it in tab.  This allows subroutines
             * close and reopen, if necessary.  Appropriate lock was obtained
             * by phase 1, needn't get it again.
             */
            tab->rel = relation_open(tab->relid, NoLock);
 
            foreach(lcmd, subcmds)
                ATExecCmd(wqueue, tab,
                          lfirst_node(AlterTableCmd, lcmd),
                          lockmode, pass, context);
 
            /*
             * 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);
 
            if (tab->rel)
            {
                relation_close(tab->rel, NoLock);
                tab->rel = NULL;
            }
        }
    }
 
    /* 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,
          AlterTableCmd *cmd, LOCKMODE lockmode, int cur_pass,
          AlterTableUtilityContext *context)
{
    ObjectAddress address = InvalidObjectAddress;
    Relation    rel = tab->rel;
 
    switch (cmd->subtype)
    {
        case AT_AddColumn:      /* ADD COLUMN */
        case AT_AddColumnToView:    /* add column via CREATE OR REPLACE VIEW */
            address = ATExecAddColumn(wqueue, tab, rel, &cmd,
                                      cmd->recurse, false,
                                      lockmode, cur_pass, context);
            break;
        case AT_ColumnDefault:  /* ALTER COLUMN DEFAULT */
            address = ATExecColumnDefault(rel, cmd->name, cmd->def, lockmode);
            break;
        case AT_CookedColumnDefault:    /* add a pre-cooked default */
            address = ATExecCookedColumnDefault(rel, cmd->num, cmd->def);
            break;
        case AT_AddIdentity:
            cmd = ATParseTransformCmd(wqueue, tab, rel, cmd, false, lockmode,
                                      cur_pass, context);
            Assert(cmd != NULL);
            address = ATExecAddIdentity(rel, cmd->name, cmd->def, lockmode);
            break;
        case AT_SetIdentity:
            cmd = ATParseTransformCmd(wqueue, tab, rel, cmd, false, lockmode,
                                      cur_pass, context);
            Assert(cmd != NULL);
            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_CheckNotNull:   /* check column is already marked NOT NULL */
            ATExecCheckNotNull(tab, rel, cmd->name, lockmode);
            break;
        case AT_DropExpression:
            address = ATExecDropExpression(rel, cmd->name, cmd->missing_ok, 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_SetCompression: /* ALTER COLUMN SET COMPRESSION */
            address = ATExecSetCompression(rel, cmd->name, cmd->def,
                                           lockmode);
            break;
        case AT_DropColumn:     /* DROP COLUMN */
            address = ATExecDropColumn(wqueue, rel, cmd->name,
                                       cmd->behavior, cmd->recurse, false,
                                       cmd->missing_ok, lockmode,
                                       NULL);
            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_ReAddStatistics:    /* ADD STATISTICS */
            address = ATExecAddStatistics(tab, rel, (CreateStatsStmt *) cmd->def,
                                          true, lockmode);
            break;
        case AT_AddConstraint:  /* ADD CONSTRAINT */
            /* Transform the command only during initial examination */
            if (cur_pass == AT_PASS_ADD_CONSTR)
                cmd = ATParseTransformCmd(wqueue, tab, rel, cmd,
                                          cmd->recurse, lockmode,
                                          cur_pass, context);
            /* Depending on constraint type, might be no more work to do now */
            if (cmd != NULL)
                address =
                    ATExecAddConstraint(wqueue, tab, rel,
                                        (Constraint *) cmd->def,
                                        cmd->recurse, 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(wqueue, rel, cmd->name, cmd->recurse,
                                               false, lockmode);
            break;
        case AT_DropConstraint: /* DROP CONSTRAINT */
            ATExecDropConstraint(rel, cmd->name, cmd->behavior,
                                 cmd->recurse, false,
                                 cmd->missing_ok, lockmode);
            break;
        case AT_AlterColumnType:    /* ALTER COLUMN TYPE */
            /* parse transformation was done earlier */
            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_DropOids:       /* SET WITHOUT OIDS */
            /* nothing to do here, oid columns don't exist anymore */
            break;
        case AT_SetAccessMethod:    /* SET ACCESS METHOD */
            /* handled specially in Phase 3 */
            break;
        case AT_SetTableSpace:  /* SET TABLESPACE */
 
            /*
             * Only do this for partitioned tables and indexes, for which this
             * is just a catalog change.  Other relation types which have
             * storage are handled by Phase 3.
             */
            if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ||
                rel->rd_rel->relkind == RELKIND_PARTITIONED_INDEX)
                ATExecSetTableSpaceNoStorage(rel, tab->newTableSpace);
 
            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,
                                       cmd->recurse,
                                       lockmode);
            break;
        case AT_EnableAlwaysTrig:   /* ENABLE ALWAYS TRIGGER name */
            ATExecEnableDisableTrigger(rel, cmd->name,
                                       TRIGGER_FIRES_ALWAYS, false,
                                       cmd->recurse,
                                       lockmode);
            break;
        case AT_EnableReplicaTrig:  /* ENABLE REPLICA TRIGGER name */
            ATExecEnableDisableTrigger(rel, cmd->name,
                                       TRIGGER_FIRES_ON_REPLICA, false,
                                       cmd->recurse,
                                       lockmode);
            break;
        case AT_DisableTrig:    /* DISABLE TRIGGER name */
            ATExecEnableDisableTrigger(rel, cmd->name,
                                       TRIGGER_DISABLED, false,
                                       cmd->recurse,
                                       lockmode);
            break;
        case AT_EnableTrigAll:  /* ENABLE TRIGGER ALL */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_FIRES_ON_ORIGIN, false,
                                       cmd->recurse,
                                       lockmode);
            break;
        case AT_DisableTrigAll: /* DISABLE TRIGGER ALL */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_DISABLED, false,
                                       cmd->recurse,
                                       lockmode);
            break;
        case AT_EnableTrigUser: /* ENABLE TRIGGER USER */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_FIRES_ON_ORIGIN, true,
                                       cmd->recurse,
                                       lockmode);
            break;
        case AT_DisableTrigUser:    /* DISABLE TRIGGER USER */
            ATExecEnableDisableTrigger(rel, NULL,
                                       TRIGGER_DISABLED, true,
                                       cmd->recurse,
                                       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:
            ATExecSetRowSecurity(rel, true);
            break;
        case AT_DisableRowSecurity:
            ATExecSetRowSecurity(rel, false);
            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:
            cmd = ATParseTransformCmd(wqueue, tab, rel, cmd, false, lockmode,
                                      cur_pass, context);
            Assert(cmd != NULL);
            if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)