Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tree: 5d4b60f2f2
Fetching contributors…

Cannot retrieve contributors at this time

10402 lines (9134 sloc) 317.144 kb
/*-------------------------------------------------------------------------
*
* tablecmds.c
* Commands for creating and altering table structures and settings
*
* Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
*
* IDENTIFICATION
* src/backend/commands/tablecmds.c
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/genam.h"
#include "access/heapam.h"
#include "access/reloptions.h"
#include "access/relscan.h"
#include "access/sysattr.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/heap.h"
#include "catalog/index.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/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_inherits_fn.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_opclass.h"
#include "catalog/pg_tablespace.h"
#include "catalog/pg_trigger.h"
#include "catalog/pg_type.h"
#include "catalog/pg_type_fn.h"
#include "catalog/storage.h"
#include "catalog/toasting.h"
#include "commands/cluster.h"
#include "commands/comment.h"
#include "commands/defrem.h"
#include "commands/sequence.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "commands/trigger.h"
#include "commands/typecmds.h"
#include "executor/executor.h"
#include "foreign/foreign.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "nodes/parsenodes.h"
#include "optimizer/clauses.h"
#include "optimizer/planner.h"
#include "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 "rewrite/rewriteDefine.h"
#include "rewrite/rewriteHandler.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "storage/lock.h"
#include "storage/predicate.h"
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/relcache.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/tqual.h"
#include "utils/typcache.h"
/*
* ON COMMIT action list
*/
typedef struct OnCommitItem
{
Oid relid; /* relid of relation */
OnCommitAction oncommit; /* what to do at end of xact */
/*
* If this entry was created during the current transaction,
* creating_subid is the ID of the creating subxact; if created in a prior
* transaction, creating_subid is zero. If deleted during the current
* transaction, deleting_subid is the ID of the deleting subxact; if no
* deletion request is pending, deleting_subid is zero.
*/
SubTransactionId creating_subid;
SubTransactionId deleting_subid;
} OnCommitItem;
static List *on_commits = NIL;
/*
* State information for ALTER TABLE
*
* The pending-work queue for an ALTER TABLE is a List of AlteredTableInfo
* structs, one for each table modified by the operation (the named table
* plus any child tables that are affected). We save lists of subcommands
* to apply to this table (possibly modified by parse transformation steps);
* these lists will be executed in Phase 2. If a Phase 3 step is needed,
* necessary information is stored in the constraints and newvals lists.
*
* Phase 2 is divided into multiple passes; subcommands are executed in
* a pass determined by subcommand type.
*/
#define AT_PASS_DROP 0 /* DROP (all flavors) */
#define AT_PASS_ALTER_TYPE 1 /* ALTER COLUMN TYPE */
#define AT_PASS_OLD_INDEX 2 /* re-add existing indexes */
#define AT_PASS_OLD_CONSTR 3 /* re-add existing constraints */
#define AT_PASS_COL_ATTRS 4 /* set other column attributes */
/* We could support a RENAME COLUMN pass here, but not currently used */
#define AT_PASS_ADD_COL 5 /* ADD COLUMN */
#define AT_PASS_ADD_INDEX 6 /* ADD indexes */
#define AT_PASS_ADD_CONSTR 7 /* ADD constraints, defaults */
#define AT_PASS_MISC 8 /* other stuff */
#define AT_NUM_PASSES 9
typedef struct AlteredTableInfo
{
/* Information saved before any work commences: */
Oid relid; /* Relation to work on */
char relkind; /* Its relkind */
TupleDesc oldDesc; /* Pre-modification tuple descriptor */
/* Information saved by Phase 1 for Phase 2: */
List *subcmds[AT_NUM_PASSES]; /* Lists of AlterTableCmd */
/* Information saved by Phases 1/2 for Phase 3: */
List *constraints; /* List of NewConstraint */
List *newvals; /* List of NewColumnValue */
bool new_notnull; /* T if we added new NOT NULL constraints */
bool rewrite; /* T if a rewrite is forced */
Oid newTableSpace; /* new tablespace; 0 means no change */
/* Objects to rebuild after completing ALTER TYPE operations */
List *changedConstraintOids; /* OIDs of constraints to rebuild */
List *changedConstraintDefs; /* string definitions of same */
List *changedIndexOids; /* OIDs of indexes to rebuild */
List *changedIndexDefs; /* string definitions of same */
} AlteredTableInfo;
/* Struct describing one new constraint to check in Phase 3 scan */
/* Note: new NOT NULL constraints are handled elsewhere */
typedef struct NewConstraint
{
char *name; /* Constraint name, or NULL if none */
ConstrType contype; /* CHECK or FOREIGN */
Oid refrelid; /* PK rel, if FOREIGN */
Oid refindid; /* OID of PK's index, if FOREIGN */
Oid conid; /* OID of pg_constraint entry, if FOREIGN */
Node *qual; /* Check expr or CONSTR_FOREIGN Constraint */
List *qualstate; /* Execution state for CHECK */
} NewConstraint;
/*
* Struct describing one new column value that needs to be computed during
* Phase 3 copy (this could be either a new column with a non-null default, or
* a column that we're changing the type of). Columns without such an entry
* are just copied from the old table during ATRewriteTable. Note that the
* expr is an expression over *old* table values.
*/
typedef struct NewColumnValue
{
AttrNumber attnum; /* which column */
Expr *expr; /* expression to compute */
ExprState *exprstate; /* execution state */
} NewColumnValue;
/*
* Error-reporting support for RemoveRelations
*/
struct dropmsgstrings
{
char kind;
int nonexistent_code;
const char *nonexistent_msg;
const char *skipping_msg;
const char *nota_msg;
const char *drophint_msg;
};
static const struct dropmsgstrings dropmsgstringarray[] = {
{RELKIND_RELATION,
ERRCODE_UNDEFINED_TABLE,
gettext_noop("table \"%s\" does not exist"),
gettext_noop("table \"%s\" does not exist, skipping"),
gettext_noop("\"%s\" is not a table"),
gettext_noop("Use DROP TABLE to remove a table.")},
{RELKIND_SEQUENCE,
ERRCODE_UNDEFINED_TABLE,
gettext_noop("sequence \"%s\" does not exist"),
gettext_noop("sequence \"%s\" does not exist, skipping"),
gettext_noop("\"%s\" is not a sequence"),
gettext_noop("Use DROP SEQUENCE to remove a sequence.")},
{RELKIND_VIEW,
ERRCODE_UNDEFINED_TABLE,
gettext_noop("view \"%s\" does not exist"),
gettext_noop("view \"%s\" does not exist, skipping"),
gettext_noop("\"%s\" is not a view"),
gettext_noop("Use DROP VIEW to remove a view.")},
{RELKIND_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.")},
{'\0', 0, NULL, NULL, NULL, NULL}
};
struct DropRelationCallbackState
{
char relkind;
Oid heapOid;
bool concurrent;
};
/* Alter table target-type flags for ATSimplePermissions */
#define ATT_TABLE 0x0001
#define ATT_VIEW 0x0002
#define ATT_INDEX 0x0004
#define ATT_COMPOSITE_TYPE 0x0008
#define ATT_FOREIGN_TABLE 0x0010
static void truncate_check_rel(Relation rel);
static List *MergeAttributes(List *schema, List *supers, char relpersistence,
List **supOids, List **supconstr, int *supOidCount);
static bool MergeCheckConstraint(List *constraints, char *name, Node *expr);
static bool change_varattnos_walker(Node *node, const AttrNumber *newattno);
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);
static void StoreCatalogInheritance1(Oid relationId, Oid parentOid,
int16 seqNumber, Relation inhRelation);
static int findAttrByName(const char *attributeName, List *schema);
static void AlterIndexNamespaces(Relation classRel, Relation rel,
Oid oldNspOid, Oid newNspOid);
static void AlterSeqNamespaces(Relation classRel, Relation rel,
Oid oldNspOid, Oid newNspOid,
const char *newNspName, LOCKMODE lockmode);
static void ATExecValidateConstraint(Relation rel, char *constrName,
bool recurse, bool recursing, LOCKMODE lockmode);
static int transformColumnNameList(Oid relId, List *colList,
int16 *attnums, Oid *atttypids);
static int transformFkeyGetPrimaryKey(Relation pkrel, Oid *indexOid,
List **attnamelist,
int16 *attnums, Oid *atttypids,
Oid *opclasses);
static Oid transformFkeyCheckAttrs(Relation pkrel,
int numattrs, int16 *attnums,
Oid *opclasses);
static void checkFkeyPermissions(Relation rel, int16 *attnums, int natts);
static CoercionPathType findFkeyCast(Oid targetTypeId, Oid sourceTypeId,
Oid *funcid);
static void validateCheckConstraint(Relation rel, HeapTuple constrtup);
static void validateForeignKeyConstraint(char *conname,
Relation rel, Relation pkrel,
Oid pkindOid, Oid constraintOid);
static void createForeignKeyTriggers(Relation rel, Constraint *fkconstraint,
Oid constraintOid, Oid indexOid);
static void ATController(Relation rel, List *cmds, bool recurse, LOCKMODE lockmode);
static void ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
bool recurse, bool recursing, LOCKMODE lockmode);
static void ATRewriteCatalogs(List **wqueue, LOCKMODE lockmode);
static void ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATRewriteTables(List **wqueue, LOCKMODE lockmode);
static void ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode);
static AlteredTableInfo *ATGetQueueEntry(List **wqueue, Relation rel);
static void ATSimplePermissions(Relation rel, int allowed_targets);
static void ATWrongRelkindError(Relation rel, int allowed_targets);
static void ATSimpleRecursion(List **wqueue, Relation rel,
AlterTableCmd *cmd, bool recurse, LOCKMODE lockmode);
static void ATTypedTableRecursion(List **wqueue, Relation rel, AlterTableCmd *cmd,
LOCKMODE lockmode);
static List *find_typed_table_dependencies(Oid typeOid, const char *typeName,
DropBehavior behavior);
static void ATPrepAddColumn(List **wqueue, Relation rel, bool recurse, bool recursing,
AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
ColumnDef *colDef, bool isOid,
bool recurse, bool recursing, LOCKMODE lockmode);
static void check_for_column_name_collision(Relation rel, const char *colname);
static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid);
static void add_column_collation_dependency(Oid relid, int32 attnum, Oid collid);
static void ATPrepAddOids(List **wqueue, Relation rel, bool recurse,
AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode);
static void ATExecSetNotNull(AlteredTableInfo *tab, Relation rel,
const char *colName, LOCKMODE lockmode);
static void ATExecColumnDefault(Relation rel, const char *colName,
Node *newDefault, LOCKMODE lockmode);
static void ATPrepSetStatistics(Relation rel, const char *colName,
Node *newValue, LOCKMODE lockmode);
static void ATExecSetStatistics(Relation rel, const char *colName,
Node *newValue, LOCKMODE lockmode);
static void ATExecSetOptions(Relation rel, const char *colName,
Node *options, bool isReset, LOCKMODE lockmode);
static void ATExecSetStorage(Relation rel, const char *colName,
Node *newValue, LOCKMODE lockmode);
static void ATPrepDropColumn(List **wqueue, Relation rel, bool recurse, bool recursing,
AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
DropBehavior behavior,
bool recurse, bool recursing,
bool missing_ok, LOCKMODE lockmode);
static void ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
IndexStmt *stmt, bool is_rebuild, LOCKMODE lockmode);
static void ATExecAddConstraint(List **wqueue,
AlteredTableInfo *tab, Relation rel,
Constraint *newConstraint, bool recurse, LOCKMODE lockmode);
static void ATExecAddIndexConstraint(AlteredTableInfo *tab, Relation rel,
IndexStmt *stmt, LOCKMODE lockmode);
static void ATAddCheckConstraint(List **wqueue,
AlteredTableInfo *tab, Relation rel,
Constraint *constr,
bool recurse, bool recursing, LOCKMODE lockmode);
static void ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
Constraint *fkconstraint, LOCKMODE lockmode);
static void ATExecDropConstraint(Relation rel, const char *constrName,
DropBehavior behavior,
bool recurse, bool recursing,
bool missing_ok, LOCKMODE lockmode);
static void ATPrepAlterColumnType(List **wqueue,
AlteredTableInfo *tab, Relation rel,
bool recurse, bool recursing,
AlterTableCmd *cmd, LOCKMODE lockmode);
static bool ATColumnChangeRequiresRewrite(Node *expr, AttrNumber varattno);
static void ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATExecAlterColumnGenericOptions(Relation rel, const char *colName,
List *options, LOCKMODE lockmode);
static void ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode);
static void ATPostAlterTypeParse(Oid oldId, char *cmd,
List **wqueue, LOCKMODE lockmode, bool rewrite);
static void TryReuseIndex(Oid oldId, IndexStmt *stmt);
static void TryReuseForeignKey(Oid oldId, Constraint *con);
static void change_owner_fix_column_acls(Oid relationOid,
Oid oldOwnerId, Oid newOwnerId);
static void change_owner_recurse_to_sequences(Oid relationOid,
Oid newOwnerId, LOCKMODE lockmode);
static void ATExecClusterOn(Relation rel, const char *indexName, LOCKMODE lockmode);
static void ATExecDropCluster(Relation rel, LOCKMODE lockmode);
static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
char *tablespacename, LOCKMODE lockmode);
static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
static void ATExecSetRelOptions(Relation rel, List *defList,
AlterTableType operation,
LOCKMODE lockmode);
static void ATExecEnableDisableTrigger(Relation rel, char *trigname,
char fires_when, bool skip_system, LOCKMODE lockmode);
static void ATExecEnableDisableRule(Relation rel, char *rulename,
char fires_when, LOCKMODE lockmode);
static void ATPrepAddInherit(Relation child_rel);
static void ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode);
static void ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode);
static void drop_parent_dependency(Oid relid, Oid refclassid, Oid refobjid);
static void ATExecAddOf(Relation rel, const TypeName *ofTypename, LOCKMODE lockmode);
static void ATExecDropOf(Relation rel, LOCKMODE lockmode);
static void ATExecGenericOptions(Relation rel, List *options);
static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
ForkNumber forkNum, char relpersistence);
static const char *storage_name(char c);
static void RangeVarCallbackForDropRelation(const RangeVar *rel, Oid relOid,
Oid oldRelOid, void *arg);
static void RangeVarCallbackForAlterRelation(const RangeVar *rv, Oid relid,
Oid oldrelid, void *arg);
/* ----------------------------------------------------------------
* 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.
*
* 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 OID of the new relation.
* ----------------------------------------------------------------
*/
Oid
DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId)
{
char relname[NAMEDATALEN];
Oid namespaceId;
List *schema = stmt->tableElts;
Oid relationId;
Oid tablespaceId;
Relation rel;
TupleDesc descriptor;
List *inheritOids;
List *old_constraints;
bool localHasOids;
int parentOidCount;
List *rawDefaults;
List *cookedDefaults;
Datum reloptions;
ListCell *listptr;
AttrNumber attnum;
static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
Oid ofTypeId;
/*
* Truncate relname to appropriate length (probably a waste of time, as
* parser should have done this already).
*/
StrNCpy(relname, stmt->relation->relname, NAMEDATALEN);
/*
* Check consistency of arguments
*/
if (stmt->oncommit != ONCOMMIT_NOOP
&& stmt->relation->relpersistence != RELPERSISTENCE_TEMP)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("ON COMMIT can only be used on temporary tables")));
if (stmt->constraints != NIL && relkind == RELKIND_FOREIGN_TABLE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("constraints on foreign tables are not supported")));
/*
* Look up the namespace in which we are supposed to create the relation,
* check we have permission to create there, lock it against concurrent
* drop, and mark stmt->relation as RELPERSISTENCE_TEMP if a temporary
* namespace is selected.
*/
namespaceId =
RangeVarGetAndCheckCreationNamespace(stmt->relation, NoLock, NULL);
/*
* Security check: disallow creating temp tables from security-restricted
* code. This is needed because calling code might not expect untrusted
* tables to appear in pg_temp at the front of its search path.
*/
if (stmt->relation->relpersistence == RELPERSISTENCE_TEMP
&& InSecurityRestrictedOperation())
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("cannot create temporary table within security-restricted operation")));
/*
* Select tablespace to use. If not specified, use default tablespace
* (which may in turn default to database's default).
*/
if (stmt->tablespacename)
{
tablespaceId = get_tablespace_oid(stmt->tablespacename, false);
}
else
{
tablespaceId = GetDefaultTablespace(stmt->relation->relpersistence);
/* note InvalidOid is OK in this case */
}
/* Check permissions except when using database's default */
if (OidIsValid(tablespaceId) && tablespaceId != MyDatabaseTableSpace)
{
AclResult aclresult;
aclresult = pg_tablespace_aclcheck(tablespaceId, GetUserId(),
ACL_CREATE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_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);
(void) heap_reloptions(relkind, reloptions, true);
if (stmt->ofTypename)
{
AclResult aclresult;
ofTypeId = typenameTypeId(NULL, stmt->ofTypename);
aclresult = pg_type_aclcheck(ofTypeId, GetUserId(), ACL_USAGE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_TYPE,
format_type_be(ofTypeId));
}
else
ofTypeId = InvalidOid;
/*
* Look up inheritance ancestors and generate relation schema, including
* inherited attributes.
*/
schema = MergeAttributes(schema, stmt->inhRelations,
stmt->relation->relpersistence,
&inheritOids, &old_constraints, &parentOidCount);
/*
* Create a tuple descriptor from the relation schema. Note that this
* deals with column names, types, and NOT NULL constraints, but not
* default values or CHECK constraints; we handle those below.
*/
descriptor = BuildDescForRelation(schema);
localHasOids = interpretOidsOption(stmt->options);
descriptor->tdhasoid = (localHasOids || parentOidCount > 0);
/*
* 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, schema)
{
ColumnDef *colDef = lfirst(listptr);
attnum++;
if (colDef->raw_default != NULL)
{
RawColumnDefault *rawEnt;
if (relkind == RELKIND_FOREIGN_TABLE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("default values on foreign tables are not supported")));
Assert(colDef->cooked_default == NULL);
rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));
rawEnt->attnum = attnum;
rawEnt->raw_default = colDef->raw_default;
rawDefaults = lappend(rawDefaults, rawEnt);
descriptor->attrs[attnum - 1]->atthasdef = true;
}
else if (colDef->cooked_default != NULL)
{
CookedConstraint *cooked;
cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
cooked->contype = CONSTR_DEFAULT;
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);
descriptor->attrs[attnum - 1]->atthasdef = true;
}
}
/*
* Create the relation. Inherited defaults and constraints are passed in
* for immediate handling --- since they don't need parsing, they can be
* stored immediately.
*/
relationId = heap_create_with_catalog(relname,
namespaceId,
tablespaceId,
InvalidOid,
InvalidOid,
ofTypeId,
ownerId,
descriptor,
list_concat(cookedDefaults,
old_constraints),
relkind,
stmt->relation->relpersistence,
false,
false,
localHasOids,
parentOidCount,
stmt->oncommit,
reloptions,
true,
allowSystemTableMods);
/* Store inheritance information for new rel. */
StoreCatalogInheritance(relationId, inheritOids);
/*
* 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 values and CHECK constraints
* to the new relation. These are passed to us in the form of raw
* parsetrees; we need to transform them to executable expression trees
* before they can be added. The most convenient way to do that is to
* apply the parser's transformExpr routine, but transformExpr doesn't
* work unless we have a pre-existing relation. So, the transformation has
* to be postponed to this final step of CREATE TABLE.
*/
if (rawDefaults || stmt->constraints)
AddRelationNewConstraints(rel, rawDefaults, stmt->constraints,
true, true);
/*
* 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 relationId;
}
/*
* Emit the right error or warning message for a "DROP" command issued on a
* non-existent relation
*/
static void
DropErrorMsgNonExistent(const char *relname, char rightkind, bool missing_ok)
{
const struct dropmsgstrings *rentry;
for (rentry = dropmsgstringarray; rentry->kind != '\0'; rentry++)
{
if (rentry->kind == rightkind)
{
if (!missing_ok)
{
ereport(ERROR,
(errcode(rentry->nonexistent_code),
errmsg(rentry->nonexistent_msg, relname)));
}
else
{
ereport(NOTICE, (errmsg(rentry->skipping_msg, 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 FOREIGN TABLE
*/
void
RemoveRelations(DropStmt *drop)
{
ObjectAddresses *objects;
char relkind;
ListCell *cell;
int flags = 0;
LOCKMODE lockmode = AccessExclusiveLock;
if (drop->concurrent)
{
lockmode = ShareUpdateExclusiveLock;
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_FOREIGN_TABLE:
relkind = RELKIND_FOREIGN_TABLE;
break;
default:
elog(ERROR, "unrecognized drop object type: %d",
(int) drop->removeType);
relkind = 0; /* keep compiler quiet */
break;
}
/* Lock and validate each relation; build a list of object addresses */
objects = new_object_addresses();
foreach(cell, drop->objects)
{
RangeVar *rel = makeRangeVarFromNameList((List *) lfirst(cell));
Oid relOid;
ObjectAddress obj;
struct DropRelationCallbackState state;
/*
* These next few steps are a great deal like relation_openrv, but we
* don't bother building a relcache entry since we don't need it.
*
* Check for shared-cache-inval messages before trying to access the
* relation. This is needed to cover the case where the name
* identifies a rel that has been dropped and recreated since the
* start of our transaction: if we don't flush the old syscache entry,
* then we'll latch onto that entry and suffer an error later.
*/
AcceptInvalidationMessages();
/* Look up the appropriate relation using namespace search. */
state.relkind = relkind;
state.heapOid = InvalidOid;
state.concurrent = drop->concurrent;
relOid = RangeVarGetRelidExtended(rel, lockmode, true,
false,
RangeVarCallbackForDropRelation,
(void *) &state);
/* Not there? */
if (!OidIsValid(relOid))
{
DropErrorMsgNonExistent(rel->relname, relkind, drop->missing_ok);
continue;
}
/* OK, we're ready to delete this one */
obj.classId = RelationRelationId;
obj.objectId = relOid;
obj.objectSubId = 0;
add_exact_object_address(&obj, objects);
}
/*
* Set options and check further requirements for concurrent drop
*/
if (drop->concurrent)
{
/*
* Confirm that concurrent behaviour is restricted in grammar.
*/
Assert(drop->removeType == OBJECT_INDEX);
flags |= PERFORM_DELETION_CONCURRENTLY;
}
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.
*/
static void
RangeVarCallbackForDropRelation(const RangeVar *rel, Oid relOid, Oid oldRelOid,
void *arg)
{
HeapTuple tuple;
struct DropRelationCallbackState *state;
char relkind;
Form_pg_class classform;
LOCKMODE heap_lockmode;
state = (struct DropRelationCallbackState *) arg;
relkind = state->relkind;
heap_lockmode = state->concurrent ?
ShareUpdateExclusiveLock : AccessExclusiveLock;
/*
* If we previously locked some other index's heap, and the name we're
* looking up no longer refers to that relation, release the now-useless
* lock.
*/
if (relOid != oldRelOid && OidIsValid(state->heapOid))
{
UnlockRelationOid(state->heapOid, heap_lockmode);
state->heapOid = InvalidOid;
}
/* 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);
if (classform->relkind != relkind)
DropErrorMsgWrongType(rel->relname, classform->relkind, relkind);
/* Allow DROP to either table owner or schema owner */
if (!pg_class_ownercheck(relOid, GetUserId()) &&
!pg_namespace_ownercheck(classform->relnamespace, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
rel->relname);
if (!allowSystemTableMods && IsSystemClass(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 droppd.
*/
if (relkind == RELKIND_INDEX && relOid != oldRelOid)
{
state->heapOid = IndexGetRelation(relOid, true);
if (OidIsValid(state->heapOid))
LockRelationOid(state->heapOid, heap_lockmode);
}
}
/*
* ExecuteTruncate
* Executes a TRUNCATE command.
*
* This is a multi-relation truncate. We first open and grab exclusive
* lock on all relations involved, checking permissions and otherwise
* verifying that the relation is OK for truncation. In CASCADE mode,
* relations having FK references to the targeted relations are automatically
* added to the group; in RESTRICT mode, we check that all FK references are
* internal to the group that's being truncated. Finally all the relations
* are truncated and reindexed.
*/
void
ExecuteTruncate(TruncateStmt *stmt)
{
List *rels = NIL;
List *relids = NIL;
List *seq_relids = NIL;
EState *estate;
ResultRelInfo *resultRelInfos;
ResultRelInfo *resultRelInfo;
SubTransactionId mySubid;
ListCell *cell;
/*
* Open, exclusive-lock, and check all the explicitly-specified relations
*/
foreach(cell, stmt->relations)
{
RangeVar *rv = lfirst(cell);
Relation rel;
bool recurse = interpretInhOption(rv->inhOpt);
Oid myrelid;
rel = heap_openrv(rv, AccessExclusiveLock);
myrelid = RelationGetRelid(rel);
/* don't throw error for "TRUNCATE foo, foo" */
if (list_member_oid(relids, myrelid))
{
heap_close(rel, AccessExclusiveLock);
continue;
}
truncate_check_rel(rel);
rels = lappend(rels, rel);
relids = lappend_oid(relids, myrelid);
if (recurse)
{
ListCell *child;
List *children;
children = find_all_inheritors(myrelid, AccessExclusiveLock, NULL);
foreach(child, children)
{
Oid childrelid = lfirst_oid(child);
if (list_member_oid(relids, childrelid))
continue;
/* find_all_inheritors already got lock */
rel = heap_open(childrelid, NoLock);
truncate_check_rel(rel);
rels = lappend(rels, rel);
relids = lappend_oid(relids, childrelid);
}
}
}
/*
* 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.
*/
if (stmt->behavior == DROP_CASCADE)
{
for (;;)
{
List *newrelids;
newrelids = heap_truncate_find_FKs(relids);
if (newrelids == NIL)
break; /* nothing else to add */
foreach(cell, newrelids)
{
Oid relid = lfirst_oid(cell);
Relation rel;
rel = heap_open(relid, AccessExclusiveLock);
ereport(NOTICE,
(errmsg("truncate cascades to table \"%s\"",
RelationGetRelationName(rel))));
truncate_check_rel(rel);
rels = lappend(rels, rel);
relids = lappend_oid(relids, relid);
}
}
}
/*
* 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 (stmt->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 (stmt->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 (!pg_class_ownercheck(seq_relid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
RelationGetRelationName(seq_rel));
seq_relids = lappend_oid(seq_relids, seq_relid);
relation_close(seq_rel, NoLock);
}
}
}
/* Prepare to catch AFTER triggers. */
AfterTriggerBeginQuery();
/*
* To fire triggers, we'll need an EState as well as a ResultRelInfo for
* each relation. We don't need to call ExecOpenIndices, though.
*/
estate = CreateExecutorState();
resultRelInfos = (ResultRelInfo *)
palloc(list_length(rels) * sizeof(ResultRelInfo));
resultRelInfo = resultRelInfos;
foreach(cell, rels)
{
Relation rel = (Relation) lfirst(cell);
InitResultRelInfo(resultRelInfo,
rel,
0, /* dummy rangetable index */
0);
resultRelInfo++;
}
estate->es_result_relations = resultRelInfos;
estate->es_num_result_relations = list_length(rels);
/*
* Process all BEFORE STATEMENT TRUNCATE triggers before we begin
* truncating (this is because one of them might throw an error). Also, if
* we were to allow them to prevent statement execution, that would need
* to be handled here.
*/
resultRelInfo = resultRelInfos;
foreach(cell, rels)
{
estate->es_result_relation_info = resultRelInfo;
ExecBSTruncateTriggers(estate, resultRelInfo);
resultRelInfo++;
}
/*
* OK, truncate each table.
*/
mySubid = GetCurrentSubTransactionId();
foreach(cell, rels)
{
Relation rel = (Relation) lfirst(cell);
/*
* Normally, we need a transaction-safe truncation here. However, if
* the table was either created in the current (sub)transaction or has
* a new relfilenode in the current (sub)transaction, then we can just
* truncate it in-place, because a rollback would cause the whole
* table or the current physical file to be thrown away anyway.
*/
if (rel->rd_createSubid == mySubid ||
rel->rd_newRelfilenodeSubid == mySubid)
{
/* Immediate, non-rollbackable truncation is OK */
heap_truncate_one_rel(rel);
}
else
{
Oid heap_relid;
Oid toast_relid;
/*
* 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 relfilenode value. The old storage file is scheduled for
* deletion at commit.
*/
RelationSetNewRelfilenode(rel, RecentXmin);
heap_relid = RelationGetRelid(rel);
toast_relid = rel->rd_rel->reltoastrelid;
/*
* The same for the toast table, if any.
*/
if (OidIsValid(toast_relid))
{
rel = relation_open(toast_relid, AccessExclusiveLock);
RelationSetNewRelfilenode(rel, RecentXmin);
heap_close(rel, NoLock);
}
/*
* Reconstruct the indexes to match, and we're done.
*/
reindex_relation(heap_relid, REINDEX_REL_PROCESS_TOAST);
}
}
/*
* Restart owned sequences if we were asked to.
*/
foreach(cell, seq_relids)
{
Oid seq_relid = lfirst_oid(cell);
ResetSequence(seq_relid);
}
/*
* Process all AFTER STATEMENT TRUNCATE triggers.
*/
resultRelInfo = resultRelInfos;
foreach(cell, rels)
{
estate->es_result_relation_info = resultRelInfo;
ExecASTruncateTriggers(estate, resultRelInfo);
resultRelInfo++;
}
/* Handle queued AFTER triggers */
AfterTriggerEndQuery(estate);
/* We can clean up the EState now */
FreeExecutorState(estate);
/* And close the rels (can't do this while EState still holds refs) */
foreach(cell, rels)
{
Relation rel = (Relation) lfirst(cell);
heap_close(rel, NoLock);
}
}
/*
* Check that a given rel is safe to truncate. Subroutine for ExecuteTruncate
*/
static void
truncate_check_rel(Relation rel)
{
AclResult aclresult;
/* Only allow truncate on regular tables */
if (rel->rd_rel->relkind != RELKIND_RELATION)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is not a table",
RelationGetRelationName(rel))));
/* Permissions checks */
aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
ACL_TRUNCATE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_CLASS,
RelationGetRelationName(rel));
if (!allowSystemTableMods && IsSystemRelation(rel))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("permission denied: \"%s\" is a system catalog",
RelationGetRelationName(rel))));
/*
* Don't allow truncate on temp tables of other backends ... their local
* buffer manager is not going to cope.
*/
if (RELATION_IS_OTHER_TEMP(rel))
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot truncate temporary tables of other sessions")));
/*
* Also check for active uses of the relation in the current transaction,
* including open scans and pending AFTER trigger events.
*/
CheckTableNotInUse(rel, "TRUNCATE");
}
/*
* storage_name
* returns the name corresponding to a typstorage/attstorage enum value
*/
static const char *
storage_name(char c)
{
switch (c)
{
case 'p':
return "PLAIN";
case 'm':
return "MAIN";
case 'x':
return "EXTENDED";
case 'e':
return "EXTERNAL";
default:
return "???";
}
}
/*----------
* MergeAttributes
* Returns new schema given initial schema and superclasses.
*
* Input arguments:
* 'schema' is the column/attribute definition for the table. (It's a list
* of ColumnDef's.) It is destructively changed.
* 'supers' is a list of names (as RangeVar nodes) of parent relations.
* 'relpersistence' is a persistence type of the table.
*
* Output arguments:
* 'supOids' receives a list of the OIDs of the parent relations.
* 'supconstr' receives a list of constraints belonging to the parents,
* updated as necessary to be valid for the child.
* 'supOidCount' is set to the number of parents that have OID columns.
*
* Return value:
* Completed schema list.
*
* Notes:
* The order in which the attributes are inherited is very important.
* Intuitively, the inherited attributes should come first. If a table
* inherits from multiple parents, the order of those attributes are
* according to the order of the parents specified in CREATE TABLE.
*
* Here's an example:
*
* create table person (name text, age int4, location point);
* create table emp (salary int4, manager text) inherits(person);
* create table student (gpa float8) inherits (person);
* create table stud_emp (percent int4) inherits (emp, student);
*
* The order of the attributes of stud_emp is:
*
* person {1:name, 2:age, 3:location}
* / \
* {6:gpa} student emp {4:salary, 5:manager}
* \ /
* stud_emp {7:percent}
*
* If the same attribute name appears multiple times, then it appears
* in the result table in the proper location for its first appearance.
*
* Constraints (including NOT NULL constraints) for the child table
* are the union of all relevant constraints, from both the child schema
* and parent tables.
*
* The default value for a child column is defined as:
* (1) If the child schema specifies a default, that value is used.
* (2) If neither the child nor any parent specifies a default, then
* the column will not have a default.
* (3) If conflicting defaults are inherited from different parents
* (and not overridden by the child), an error is raised.
* (4) Otherwise the inherited default is used.
* Rule (3) is new in Postgres 7.1; in earlier releases you got a
* rather arbitrary choice of which parent default to use.
*----------
*/
static List *
MergeAttributes(List *schema, List *supers, char relpersistence,
List **supOids, List **supconstr, int *supOidCount)
{
ListCell *entry;
List *inhSchema = NIL;
List *parentOids = NIL;
List *constraints = NIL;
int parentsWithOids = 0;
bool have_bogus_defaults = false;
int child_attno;
static Node bogus_marker = {0}; /* marks conflicting defaults */
/*
* 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 any 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.
*/
foreach(entry, schema)
{
ColumnDef *coldef = lfirst(entry);
ListCell *rest = lnext(entry);
ListCell *prev = entry;
if (coldef->typeName == NULL)
/*
* Typed table column option that does not belong to a column from
* the type. This works because the columns from the type come
* first in the list.
*/
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_COLUMN),
errmsg("column \"%s\" does not exist",
coldef->colname)));
while (rest != NULL)
{
ColumnDef *restdef = lfirst(rest);
ListCell *next = lnext(rest); /* need to save it in case we
* delete it */
if (strcmp(coldef->colname, restdef->colname) == 0)
{
if (coldef->is_from_type)
{
/*
* merge the column options into the column from the type
*/
coldef->is_not_null = restdef->is_not_null;
coldef->raw_default = restdef->raw_default;
coldef->cooked_default = restdef->cooked_default;
coldef->constraints = restdef->constraints;
coldef->is_from_type = false;
list_delete_cell(schema, rest, prev);
}
else
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_COLUMN),
errmsg("column \"%s\" specified more than once",
coldef->colname)));
}
prev = rest;
rest = next;
}
}
/*
* Scan the parents left-to-right, and merge their attributes to form a
* list of inherited attributes (inhSchema). Also check to see if we need
* to inherit an OID column.
*/
child_attno = 0;
foreach(entry, supers)
{
RangeVar *parent = (RangeVar *) lfirst(entry);
Relation relation;
TupleDesc tupleDesc;
TupleConstr *constr;
AttrNumber *newattno;
AttrNumber parent_attno;
/*
* A self-exclusive lock is needed here. If two backends attempt to
* add children to the same parent simultaneously, and that parent has
* no pre-existing children, then both will attempt to update the
* parent's relhassubclass field, leading to a "tuple concurrently
* updated" error. Also, this interlocks against a concurrent ANALYZE
* on the parent table, which might otherwise be attempting to clear
* the parent's relhassubclass field, if its previous children were
* recently dropped.
*/
relation = heap_openrv(parent, ShareUpdateExclusiveLock);
if (relation->rd_rel->relkind != RELKIND_RELATION)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("inherited relation \"%s\" is not a table",
parent->relname)));
/* Permanent rels cannot inherit from temporary ones */
if (relpersistence != RELPERSISTENCE_TEMP
&& RelationUsesTempNamespace(relation))
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot inherit from temporary relation \"%s\"",
parent->relname)));
/*
* We should have an UNDER permission flag for this, but for now,
* demand that creator of a child table own the parent.
*/
if (!pg_class_ownercheck(RelationGetRelid(relation), GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
RelationGetRelationName(relation));
/*
* Reject duplications in the list of parents.
*/
if (list_member_oid(parentOids, RelationGetRelid(relation)))
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_TABLE),
errmsg("relation \"%s\" would be inherited from more than once",
parent->relname)));
parentOids = lappend_oid(parentOids, RelationGetRelid(relation));
if (relation->rd_rel->relhasoids)
parentsWithOids++;
tupleDesc = RelationGetDescr(relation);
constr = tupleDesc->constr;
/*
* newattno[] will contain the child-table attribute numbers for the
* attributes of this parent table. (They are not the same for
* parents after the first one, nor if we have dropped columns.)
*/
newattno = (AttrNumber *)
palloc(tupleDesc->natts * sizeof(AttrNumber));
for (parent_attno = 1; parent_attno <= tupleDesc->natts;
parent_attno++)
{
Form_pg_attribute attribute = tupleDesc->attrs[parent_attno - 1];
char *attributeName = NameStr(attribute->attname);
int exist_attno;
ColumnDef *def;
/*
* Ignore dropped columns in the parent.
*/
if (attribute->attisdropped)
{
/*
* change_varattnos_of_a_node asserts that this is greater
* than zero, so if anything tries to use it, we should find
* out.
*/
newattno[parent_attno - 1] = 0;
continue;
}
/*
* Does it conflict with some previously inherited column?
*/
exist_attno = findAttrByName(attributeName, inhSchema);
if (exist_attno > 0)
{
Oid defTypeId;
int32 deftypmod;
Oid defCollId;
/*
* Yes, try to merge the two column definitions. They must
* have the same type, typmod, and collation.
*/
ereport(NOTICE,
(errmsg("merging multiple inherited definitions of column \"%s\"",
attributeName)));
def = (ColumnDef *) list_nth(inhSchema, exist_attno - 1);
typenameTypeIdAndMod(NULL, def->typeName, &defTypeId, &deftypmod);
if (defTypeId != attribute->atttypid ||
deftypmod != attribute->atttypmod)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("inherited column \"%s\" has a type conflict",
attributeName),
errdetail("%s versus %s",
TypeNameToString(def->typeName),
format_type_be(attribute->atttypid))));
defCollId = GetColumnDefCollation(NULL, def, defTypeId);
if (defCollId != attribute->attcollation)
ereport(ERROR,
(errcode(ERRCODE_COLLATION_MISMATCH),
errmsg("inherited column \"%s\" has a collation conflict",
attributeName),
errdetail("\"%s\" versus \"%s\"",
get_collation_name(defCollId),
get_collation_name(attribute->attcollation))));
/* Copy storage parameter */
if (def->storage == 0)
def->storage = attribute->attstorage;
else if (def->storage != attribute->attstorage)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("inherited column \"%s\" has a storage parameter conflict",
attributeName),
errdetail("%s versus %s",
storage_name(def->storage),
storage_name(attribute->attstorage))));
def->inhcount++;
/* Merge of NOT NULL constraints = OR 'em together */
def->is_not_null |= attribute->attnotnull;
/* Default and other constraints are handled below */
newattno[parent_attno - 1] = exist_attno;
}
else
{
/*
* No, create a new inherited column
*/
def = makeNode(ColumnDef);
def->colname = pstrdup(attributeName);
def->typeName = makeTypeNameFromOid(attribute->atttypid,
attribute->atttypmod);
def->inhcount = 1;
def->is_local = false;
def->is_not_null = attribute->attnotnull;
def->is_from_type = false;
def->storage = attribute->attstorage;
def->raw_default = NULL;
def->cooked_default = NULL;
def->collClause = NULL;
def->collOid = attribute->attcollation;
def->constraints = NIL;
inhSchema = lappend(inhSchema, def);
newattno[parent_attno - 1] = ++child_attno;
}
/*
* Copy default if any
*/
if (attribute->atthasdef)
{
Node *this_default = NULL;
AttrDefault *attrdef;
int i;
/* Find default in constraint structure */
Assert(constr != NULL);
attrdef = constr->defval;
for (i = 0; i < constr->num_defval; i++)
{
if (attrdef[i].adnum == parent_attno)
{
this_default = stringToNode(attrdef[i].adbin);
break;
}
}
Assert(this_default != NULL);
/*
* If default expr could contain any vars, we'd need to fix
* 'em, but it can't; so default is ready to apply to child.
*
* If we already had a default from some prior parent, check
* to see if they are the same. If so, no problem; if not,
* mark the column as having a bogus default. Below, we will
* complain if the bogus default isn't overridden by the child
* schema.
*/
Assert(def->raw_default == NULL);
if (def->cooked_default == NULL)
def->cooked_default = this_default;
else if (!equal(def->cooked_default, this_default))
{
def->cooked_default = &bogus_marker;
have_bogus_defaults = true;
}
}
}
/*
* Now copy the CHECK constraints of this parent, adjusting attnos
* using the completed newattno[] map. Identically named constraints
* are merged if possible, else we throw error.
*/
if (constr && constr->num_check > 0)
{
ConstrCheck *check = constr->check;
int i;
for (i = 0; i < constr->num_check; i++)
{
char *name = check[i].ccname;
Node *expr;
/* ignore if the constraint is non-inheritable */
if (check[i].ccnoinherit)
continue;
/* adjust varattnos of ccbin here */
expr = stringToNode(check[i].ccbin);
change_varattnos_of_a_node(expr, newattno);
/* 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->name = pstrdup(name);
cooked->attnum = 0; /* not used for constraints */
cooked->expr = expr;
cooked->skip_validation = false;
cooked->is_local = false;
cooked->inhcount = 1;
cooked->is_no_inherit = false;
constraints = lappend(constraints, cooked);
}
}
}
pfree(newattno);
/*
* Close the parent rel, but keep our AccessShareLock on it until xact
* commit. That will prevent someone else from deleting or ALTERing
* the parent before the child is committed.
*/
heap_close(relation, NoLock);
}
/*
* If we had no inherited attributes, the result schema is just the
* explicitly declared columns. Otherwise, we need to merge the declared
* columns into the inherited schema list.
*/
if (inhSchema != NIL)
{
foreach(entry, schema)
{
ColumnDef *newdef = lfirst(entry);
char *attributeName = newdef->colname;
int exist_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;
/*
* Yes, try to merge the two column definitions. They must
* have the same type, typmod, and collation.
*/
ereport(NOTICE,
(errmsg("merging column \"%s\" with inherited definition",
attributeName)));
def = (ColumnDef *) list_nth(inhSchema, exist_attno - 1);
typenameTypeIdAndMod(NULL, def->typeName, &defTypeId, &deftypmod);
typenameTypeIdAndMod(NULL, newdef->typeName, &newTypeId, &newtypmod);
if (defTypeId != newTypeId || deftypmod != newtypmod)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("column \"%s\" has a type conflict",
attributeName),
errdetail("%s versus %s",
TypeNameToString(def->typeName),
TypeNameToString(newdef->typeName))));
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))));
/* Copy storage parameter */
if (def->storage == 0)
def->storage = newdef->storage;
else if (newdef->storage != 0 && def->storage != newdef->storage)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("column \"%s\" has a storage parameter conflict",
attributeName),
errdetail("%s versus %s",
storage_name(def->storage),
storage_name(newdef->storage))));
/* Mark the column as locally defined */
def->is_local = true;
/* Merge of NOT NULL constraints = OR 'em together */
def->is_not_null |= newdef->is_not_null;
/* If new def has a default, override previous default */
if (newdef->raw_default != NULL)
{
def->raw_default = newdef->raw_default;
def->cooked_default = newdef->cooked_default;
}
}
else
{
/*
* No, attach new column to result schema
*/
inhSchema = lappend(inhSchema, newdef);
}
}
schema = inhSchema;
/*
* Check that we haven't exceeded the legal # of columns after merging
* in inherited columns.
*/
if (list_length(schema) > MaxHeapAttributeNumber)
ereport(ERROR,
(errcode(ERRCODE_TOO_MANY_COLUMNS),
errmsg("tables can have at most %d columns",
MaxHeapAttributeNumber)));
}
/*
* If we found any conflicting parent default values, check to make sure
* they were overridden by the child.
*/
if (have_bogus_defaults)
{
foreach(entry, schema)
{
ColumnDef *def = lfirst(entry);
if (def->cooked_default == &bogus_marker)
ereport(ERROR,
(errcode(ERRCODE_INVALID_COLUMN_DEFINITION),
errmsg("column \"%s\" inherits conflicting default values",
def->colname),
errhint("To resolve the conflict, specify a default explicitly.")));
}
}
*supOids = parentOids;
*supconstr = constraints;
*supOidCount = parentsWithOids;
return schema;
}
/*
* MergeCheckConstraint
* Try to merge an inherited CHECK constraint with previous ones
*
* If we inherit identically-named constraints from multiple parents, we must
* merge them, or throw an error if they don't have identical definitions.
*
* constraints is a list of CookedConstraint structs for previous constraints.
*
* Returns TRUE if merged (constraint is a duplicate), or FALSE if it's
* got a so-far-unique name, or throws error if conflict.
*/
static bool
MergeCheckConstraint(List *constraints, char *name, Node *expr)
{
ListCell *lc;
foreach(lc, constraints)
{
CookedConstraint *ccon = (CookedConstraint *) lfirst(lc);
Assert(ccon->contype == CONSTR_CHECK);
/* Non-matching names never conflict */
if (strcmp(ccon->name, name) != 0)
continue;
if (equal(expr, ccon->expr))
{
/* OK to merge */
ccon->inhcount++;
return true;
}
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("check constraint name \"%s\" appears multiple times but with different expressions",
name)));
}
return false;
}
/*
* Replace varattno values in an expression tree according to the given
* map array, that is, varattno N is replaced by newattno[N-1]. It is
* caller's responsibility to ensure that the array is long enough to
* define values for all user varattnos present in the tree. System column
* attnos remain unchanged.
*
* Note that the passed node tree is modified in-place!
*/
void
change_varattnos_of_a_node(Node *node, const AttrNumber *newattno)
{
/* no setup needed, so away we go */
(void) change_varattnos_walker(node, newattno);
}
static bool
change_varattnos_walker(Node *node, const AttrNumber *newattno)
{
if (node == NULL)
return false;
if (IsA(node, Var))
{
Var *var = (Var *) node;
if (var->varlevelsup == 0 && var->varno == 1 &&
var->varattno > 0)
{
/*
* ??? the following may be a problem when the node is multiply
* referenced though stringToNode() doesn't create such a node
* currently.
*/
Assert(newattno[var->varattno - 1] > 0);
var->varattno = var->varoattno = newattno[var->varattno - 1];
}
return false;
}
return expression_tree_walker(node, change_varattnos_walker,
(void *) newattno);
}
/*
* Generate a map for change_varattnos_of_a_node from old and new TupleDesc's,
* matching according to column name.
*/
AttrNumber *
varattnos_map(TupleDesc olddesc, TupleDesc newdesc)
{
AttrNumber *attmap;
int i,
j;
attmap = (AttrNumber *) palloc0(sizeof(AttrNumber) * olddesc->natts);
for (i = 1; i <= olddesc->natts; i++)
{
if (olddesc->attrs[i - 1]->attisdropped)
continue; /* leave the entry as zero */
for (j = 1; j <= newdesc->natts; j++)
{
if (strcmp(NameStr(olddesc->attrs[i - 1]->attname),
NameStr(newdesc->attrs[j - 1]->attname)) == 0)
{
attmap[i - 1] = j;
break;
}
}
}
return attmap;
}
/*
* Generate a map for change_varattnos_of_a_node from a TupleDesc and a list
* of ColumnDefs
*/
AttrNumber *
varattnos_map_schema(TupleDesc old, List *schema)
{
AttrNumber *attmap;
int i;
attmap = (AttrNumber *) palloc0(sizeof(AttrNumber) * old->natts);
for (i = 1; i <= old->natts; i++)
{
if (old->attrs[i - 1]->attisdropped)
continue; /* leave the entry as zero */
attmap[i - 1] = findAttrByName(NameStr(old->attrs[i - 1]->attname),
schema);
}
return attmap;
}
/*
* 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)
{
Relation relation;
int16 seqNumber;
ListCell *entry;
/*
* sanity checks
*/
AssertArg(OidIsValid(relationId));
if (supers == NIL)
return;
/*
* Store INHERITS information in pg_inherits using direct ancestors only.
* Also enter dependencies on the direct ancestors, and make sure they are
* marked with relhassubclass = true.
*
* (Once upon a time, both direct and indirect ancestors were found here
* and then entered into pg_ipl. Since that catalog doesn't exist
* anymore, there's no need to look for indirect ancestors.)
*/
relation = heap_open(InheritsRelationId, RowExclusiveLock);
seqNumber = 1;
foreach(entry, supers)
{
Oid parentOid = lfirst_oid(entry);
StoreCatalogInheritance1(relationId, parentOid, seqNumber, relation);
seqNumber++;
}
heap_close(relation, RowExclusiveLock);
}
/*
* Make catalog entries showing relationId as being an inheritance child
* of parentOid. inhRelation is the already-opened pg_inherits catalog.
*/
static void
StoreCatalogInheritance1(Oid relationId, Oid parentOid,
int16 seqNumber, Relation inhRelation)
{
TupleDesc desc = RelationGetDescr(inhRelation);
Datum values[Natts_pg_inherits];
bool nulls[Natts_pg_inherits];
ObjectAddress childobject,
parentobject;
HeapTuple tuple;
/*
* Make the pg_inherits entry
*/
values[Anum_pg_inherits_inhrelid - 1] = ObjectIdGetDatum(relationId);
values[Anum_pg_inherits_inhparent - 1] = ObjectIdGetDatum(parentOid);
values[Anum_pg_inherits_inhseqno - 1] = Int16GetDatum(seqNumber);
memset(nulls, 0, sizeof(nulls));
tuple = heap_form_tuple(desc, values, nulls);
simple_heap_insert(inhRelation, tuple);
CatalogUpdateIndexes(inhRelation, tuple);
heap_freetuple(tuple);
/*
* 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, DEPENDENCY_NORMAL);
/*
* Mark the parent as having subclasses.
*/
SetRelationHasSubclass(parentOid, true);
}
/*
* Look for an existing schema entry with the given name.
*
* Returns the index (starting with 1) if attribute already exists in schema,
* 0 if it doesn't.
*/
static int
findAttrByName(const char *attributeName, List *schema)
{
ListCell *s;
int i = 1;
foreach(s, schema)
{
ColumnDef *def = lfirst(s);
if (strcmp(attributeName, def->colname) == 0)
return i;
i++;
}
return 0;
}
/*
* SetRelationHasSubclass
* Set the value of the relation's relhassubclass field in pg_class.
*
* NOTE: caller must be holding an appropriate lock on the relation.
* ShareUpdateExclusiveLock is sufficient.
*
* NOTE: an important side-effect of this operation is that an SI invalidation
* message is sent out to all backends --- including me --- causing plans
* referencing the relation to be rebuilt with the new list of children.
* This must happen even if we find that no change is needed in the pg_class
* row.
*/
void
SetRelationHasSubclass(Oid relationId, bool relhassubclass)
{
Relation relationRelation;
HeapTuple tuple;
Form_pg_class classtuple;
/*
* Fetch a modifiable copy of the tuple, modify it, update pg_class.
*/
relationRelation = heap_open(RelationRelationId, RowExclusiveLock);
tuple = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(relationId));
if (!HeapTupleIsValid(tuple))
elog(ERROR, "cache lookup failed for relation %u", relationId);
classtuple = (Form_pg_class) GETSTRUCT(tuple);
if (classtuple->relhassubclass != relhassubclass)
{
classtuple->relhassubclass = relhassubclass;
simple_heap_update(relationRelation, &tuple->t_self, tuple);
/* keep the catalog indexes up to date */
CatalogUpdateIndexes(relationRelation, tuple);
}
else
{
/* no need to change tuple, but force relcache rebuild anyway */
CacheInvalidateRelcacheByTuple(tuple);
}
heap_freetuple(tuple);
heap_close(relationRelation, RowExclusiveLock);
}
/*
* renameatt_check - basic sanity checks before attribute rename
*/
static void
renameatt_check(Oid myrelid, Form_pg_class classform, bool recursing)
{
char relkind = classform->relkind;
if (classform->reloftype && !recursing)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot rename column of typed table")));
/*
* Renaming the columns of sequences or toast tables doesn't actually
* break anything from the system's point of view, since internal
* references are by attnum. But it doesn't seem right to allow users to
* change names that are hardcoded into the system, hence the following
* restriction.
*/
if (relkind != RELKIND_RELATION &&
relkind != RELKIND_VIEW &&
relkind != RELKIND_COMPOSITE_TYPE &&
relkind != RELKIND_INDEX &&
relkind != RELKIND_FOREIGN_TABLE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is not a table, view, composite type, index, or foreign table",
NameStr(classform->relname))));
/*
* permissions checking. only the owner of a class can change its schema.
*/
if (!pg_class_ownercheck(myrelid, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
NameStr(classform->relname));
if (!allowSystemTableMods && IsSystemClass(classform))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("permission denied: \"%s\" is a system catalog",
NameStr(classform->relname))));
}
/*
* renameatt_internal - workhorse for renameatt
*/
static void
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;
int attnum;
/*
* Grab an exclusive lock on the target table, which we will NOT release
* until end of transaction.
*/
targetrelation = relation_open(myrelid, AccessExclusiveLock);
renameatt_check(myrelid, RelationGetForm(targetrelation), recursing);
/*
* if the 'recurse' flag is set then we are supposed to rename this
* attribute in all classes that inherit from 'relname' (as well as in
* 'relname').
*
* any permissions or problems with duplicate attributes will cause the
* whole transaction to abort, which is what we want -- all or nothing.
*/
if (recurse)
{
List *child_oids,
*child_numparents;
ListCell *lo,
*li;
/*
* we need the number of parents for each child so that the recursive
* calls to renameatt() can determine whether there are any parents
* outside the inheritance hierarchy being processed.
*/
child_oids = find_all_inheritors(myrelid, AccessExclusiveLock,
&child_numparents);
/*
* find_all_inheritors does the recursive search of the inheritance
* hierarchy, so all we have to do is process all of the relids in the
* list that it returns.
*/
forboth(lo, child_oids, li, child_numparents)
{
Oid childrelid = lfirst_oid(lo);
int numparents = lfirst_int(li);
if (childrelid == myrelid)
continue;
/* note we need not recurse again */
renameatt_internal(childrelid, oldattname, newattname, false, true, numparents, behavior);
}
}
else
{
/*
* If we are told not to recurse, there had better not be any child
* tables; else the rename would put them out of step.
*
* expected_parents will only be 0 if we are not already recursing.
*/
if (expected_parents == 0 &&
find_inheritance_children(myrelid, NoLock) != NIL)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("inherited column \"%s\" must be renamed in child tables too",
oldattname)));
}
/* rename attributes in typed tables of composite type */
if (targetrelation->rd_rel->relkind == RELKIND_COMPOSITE_TYPE)
{
List *child_oids;
ListCell *lo;
child_oids = find_typed_table_dependencies(targetrelation->rd_rel->reltype,
RelationGetRelationName(targetrelation),
behavior);
foreach(lo, child_oids)
renameatt_internal(lfirst_oid(lo), oldattname, newattname, true, true, 0, behavior);
}
attrelation = heap_open(AttributeRelationId, RowExclusiveLock);
atttup = SearchSysCacheCopyAttName(myrelid, oldattname);
if (!HeapTupleIsValid(atttup))
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_COLUMN),
errmsg("column \"%s\" does not exist",
oldattname)));
attform = (Form_pg_attribute) GETSTRUCT(atttup);
attnum = attform->attnum;
if (attnum <= 0)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot rename system column \"%s\"",
oldattname)));
/*
* if the attribute is inherited, forbid the renaming. if this is a
* top-level call to renameatt(), then expected_parents will be 0, so the
* effect of this code will be to prohibit the renaming if the attribute
* is inherited at all. if this is a recursive call to renameatt(),
* expected_parents will be the number of parents the current relation has
* within the inheritance hierarchy being processed, so we'll prohibit the
* renaming only if there are additional parents from elsewhere.
*/
if (attform->attinhcount > expected_parents)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("cannot rename inherited column \"%s\"",
oldattname)));
/* new name should not already exist */
check_for_column_name_collision(targetrelation, newattname);
/* apply the update */
namestrcpy(&(attform->attname), newattname);
simple_heap_update(attrelation, &atttup->t_self, atttup);
/* keep system catalog indexes current */
CatalogUpdateIndexes(attrelation, atttup);
heap_freetuple(atttup);
heap_close(attrelation, RowExclusiveLock);
relation_close(targetrelation, NoLock); /* close rel but keep lock */
}
/*
* 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 a attribute in a relation
*/
void
renameatt(RenameStmt *stmt)
{
Oid relid;
/* lock level taken here should match renameatt_internal */
relid = RangeVarGetRelidExtended(stmt->relation, AccessExclusiveLock,
stmt->missing_ok, false,
RangeVarCallbackForRenameAttribute,
NULL);
if (!OidIsValid(relid))
{
ereport(NOTICE,
(errmsg("relation \"%s\" does not exist, skipping",
stmt->relation->relname)));
return;
}
renameatt_internal(relid,
stmt->subname, /* old att name */
stmt->newname, /* new att name */
interpretInhOption(stmt->relation->inhOpt), /* recursive? */
false, /* recursing? */
0, /* expected inhcount */
stmt->behavior);
}
/*
* same logic as renameatt_internal
*/
static void
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;
AssertArg(!myrelid || !mytypid);
if (mytypid)
{
constraintOid = get_domain_constraint_oid(mytypid, oldconname, false);
}
else
{
targetrelation = relation_open(myrelid, AccessExclusiveLock);
/* don't tell it whether we're recursing; we allow changing typed tables here */
renameatt_check(myrelid, RelationGetForm(targetrelation), false);
constraintOid = get_relation_constraint_oid(myrelid, oldconname, false);
}
tuple = SearchSysCache1(CONSTROID, ObjectIdGetDatum(constraintOid));
if (!HeapTupleIsValid(tuple))
elog(ERROR, "cache lookup failed for constraint %u",
constraintOid);
con = (Form_pg_constraint) GETSTRUCT(tuple);
if (myrelid && con->contype == CONSTRAINT_CHECK && !con->connoinherit)
{
if (recurse)
{
List *child_oids,
*child_numparents;
ListCell *lo,
*li;
child_oids = find_all_inheritors(myrelid, AccessExclusiveLock,
&child_numparents);
forboth(lo, child_oids, li, child_numparents)
{
Oid childrelid = lfirst_oid(lo);
int numparents = lfirst_int(li);
if (childrelid == myrelid)
continue;
rename_constraint_internal(childrelid, InvalidOid, oldconname, newconname, false, true, numparents);
}
}
else
{
if (expected_parents == 0 &&
find_inheritance_children(myrelid, NoLock) != NIL)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("inherited constraint \"%s\" must be renamed in child tables too",
oldconname)));
}
if (con->coninhcount > expected_parents)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("cannot rename inherited constraint \"%s\"",
oldconname)));
}
if (con->conindid
&& (con->contype == CONSTRAINT_PRIMARY
|| con->contype == CONSTRAINT_UNIQUE
|| con->contype == CONSTRAINT_EXCLUSION))
/* rename the index; this renames the constraint as well */
RenameRelationInternal(con->conindid, newconname);
else
RenameConstraintById(constraintOid, newconname);
ReleaseSysCache(tuple);
if (targetrelation)
relation_close(targetrelation, NoLock); /* close rel but keep lock */
}
void
RenameConstraint(RenameStmt *stmt)
{
Oid relid = InvalidOid;
Oid typid = InvalidOid;
if (stmt->relationType == OBJECT_DOMAIN)
{
Relation rel;
HeapTuple tup;
typid = typenameTypeId(NULL, makeTypeNameFromNameList(stmt->object));
rel = heap_open(TypeRelationId, RowExclusiveLock);
tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typid));
if (!HeapTupleIsValid(tup))
elog(ERROR, "cache lookup failed for type %u", typid);
checkDomainOwner(tup);
ReleaseSysCache(tup);
heap_close(rel, NoLock);
}
else
{
/* lock level taken here should match rename_constraint_internal */
relid = RangeVarGetRelidExtended(stmt->relation, AccessExclusiveLock,
false, false,
RangeVarCallbackForRenameAttribute,
NULL);
}
rename_constraint_internal(relid, typid,
stmt->subname,
stmt->newname,
stmt->relation ? interpretInhOption(stmt->relation->inhOpt) : false, /* recursive? */
false, /* recursing? */
0 /* expected inhcount */);
}
/*
* Execute ALTER TABLE/INDEX/SEQUENCE/VIEW/FOREIGN TABLE RENAME
*/
void
RenameRelation(RenameStmt *stmt)
{
Oid relid;
/*
* Grab an exclusive lock on the target table, index, sequence or view,
* which we will NOT release until end of transaction.
*
* Lock level used here should match RenameRelationInternal, to avoid
* lock escalation.
*/
relid = RangeVarGetRelidExtended(stmt->relation, AccessExclusiveLock,
stmt->missing_ok, false,
RangeVarCallbackForAlterRelation,
(void *) stmt);
if (!OidIsValid(relid))
{
ereport(NOTICE,
(errmsg("relation \"%s\" does not exist, skipping",
stmt->relation->relname)));
return;
}
/* Do the work */
RenameRelationInternal(relid, stmt->newname);
}
/*
* RenameRelationInternal - change the name of a relation
*
* XXX - When renaming sequences, we don't bother to modify the
* sequence name that is stored within the sequence itself
* (this would cause problems with MVCC). In the future,
* the sequence name should probably be removed from the
* sequence, AFAIK there's no need for it to be there.
*/
void
RenameRelationInternal(Oid myrelid, const char *newrelname)
{
Relation targetrelation;
Relation relrelation; /* for RELATION relation */
HeapTuple reltup;
Form_pg_class relform;
Oid namespaceId;
/*
* Grab an exclusive lock on the target table, index, sequence or view,
* which we will NOT release until end of transaction.
*/
targetrelation = relation_open(myrelid, AccessExclusiveLock);
namespaceId = RelationGetNamespace(targetrelation);
/*
* Find relation's pg_class tuple, and make sure newrelname isn't in use.
*/
relrelation = heap_open(RelationRelationId, RowExclusiveLock);
reltup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(myrelid));
if (!HeapTupleIsValid(reltup)) /* shouldn't happen */
elog(ERROR, "cache lookup failed for relation %u", myrelid);
relform = (Form_pg_class) GETSTRUCT(reltup);
if (get_relname_relid(newrelname, namespaceId) != InvalidOid)
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_TABLE),
errmsg("relation \"%s\" already exists",
newrelname)));
/*
* Update pg_class tuple with new relname. (Scribbling on reltup is OK
* because it's a copy...)
*/
namestrcpy(&(relform->relname), newrelname);
simple_heap_update(relrelation, &reltup->t_self, reltup);
/* keep the system catalog indexes current */
CatalogUpdateIndexes(relrelation, reltup);
heap_freetuple(reltup);
heap_close(relrelation, RowExclusiveLock);
/*
* Also rename the associated type, if any.
*/
if (OidIsValid(targetrelation->rd_rel->reltype))
RenameTypeInternal(targetrelation->rd_rel->reltype,
newrelname, namespaceId);
/*
* Also rename the associated constraint, if any.
*/
if (targetrelation->rd_rel->relkind == RELKIND_INDEX)
{
Oid constraintId = get_index_constraint(myrelid);
if (OidIsValid(constraintId))
RenameConstraintById(constraintId, newrelname);
}
/*
* Close rel, but keep exclusive lock!
*/
relation_close(targetrelation, NoLock);
}
/*
* Disallow ALTER TABLE (and similar commands) when the current backend has
* any open reference to the target table besides the one just acquired by
* the calling command; this implies there's an open cursor or active plan.
* We need this check because our lock doesn't protect us against stomping
* on our own foot, only other people's feet!
*
* For ALTER TABLE, the only case known to cause serious trouble is ALTER
* COLUMN TYPE, and some changes are obviously pretty benign, so this could
* possibly be relaxed to only error out for certain types of alterations.
* But the use-case for allowing any of these things is not obvious, so we
* won't work hard at it for now.
*
* We also reject these commands if there are any pending AFTER trigger events
* for the rel. This is certainly necessary for the rewriting variants of
* ALTER TABLE, because they don't preserve tuple TIDs and so the pending
* events would try to fetch the wrong tuples. It might be overly cautious
* in other cases, but again it seems better to err on the side of paranoia.
*
* REINDEX calls this with "rel" referencing the index to be rebuilt; here
* we are worried about active indexscans on the index. The trigger-event
* check can be skipped, since we are doing no damage to the parent table.
*
* The statement name (eg, "ALTER TABLE") is passed for use in error messages.
*/
void
CheckTableNotInUse(Relation rel, const char *stmt)
{
int expected_refcnt;
expected_refcnt = rel->rd_isnailed ? 2 : 1;
if (rel->rd_refcnt != expected_refcnt)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE),
/* translator: first %s is a SQL command, eg ALTER TABLE */
errmsg("cannot %s \"%s\" because "
"it is being used by active queries in this session",
stmt, RelationGetRelationName(rel))));
if (rel->rd_rel->relkind != RELKIND_INDEX &&
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, false,
RangeVarCallbackForAlterRelation,
(void *) stmt);
}
/*
* AlterTable
* Execute ALTER TABLE, which can be a list of subcommands
*
* ALTER TABLE is performed in three phases:
* 1. Examine subcommands and perform pre-transformation checking.
* 2. Update system catalogs.
* 3. Scan table(s) to check new constraints, and optionally recopy
* the data into new table(s).
* Phase 3 is not performed unless one or more of the subcommands requires
* it. The intention of this design is to allow multiple independent
* updates of the table schema to be performed with only one pass over the
* data.
*
* ATPrepCmd performs phase 1. A "work queue" entry is created for
* each table to be affected (there may be multiple affected tables if the
* commands traverse a table inheritance hierarchy). Also we do preliminary
* validation of the subcommands, including parse transformation of those
* expressions that need to be evaluated with respect to the old table
* schema.
*
* ATRewriteCatalogs performs phase 2 for each affected table. (Note that
* phases 2 and 3 normally do no explicit recursion, since phase 1 already
* did it --- although some subcommands have to recurse in phase 2 instead.)
* Certain subcommands need to be performed before others to avoid
* unnecessary conflicts; for example, DROP COLUMN should come before
* ADD COLUMN. Therefore phase 1 divides the subcommands into multiple
* lists, one for each logical "pass" of phase 2.
*
* ATRewriteTables performs phase 3 for those tables that need it.
*
* Thanks to the magic of MVCC, an error anywhere along the way rolls back
* the whole operation; we don't have to do anything special to clean up.
*
* The caller must lock the relation, with an appropriate lock level
* for the subcommands requested. Any subcommand that needs to rewrite
* tuples in the table forces the whole command to be executed with
* AccessExclusiveLock (actually, that is currently required always, but
* we hope to relax it at some point). We pass the lock level down
* so that we can apply it recursively to inherited tables. Note that the
* lock level we want as we recurse might well be higher than required for
* that specific subcommand. So we pass down the overall lock requirement,
* rather than reassess it at lower levels.
*/
void
AlterTable(Oid relid, LOCKMODE lockmode, AlterTableStmt *stmt)
{
Relation rel;
/* Caller is required to provide an adequate lock. */
rel = relation_open(relid, NoLock);
CheckTableNotInUse(rel, "ALTER TABLE");
ATController(rel, stmt->cmds, interpretInhOption(stmt->relation->inhOpt),
lockmode);
}
/*
* AlterTableInternal
*
* ALTER TABLE with target specified by OID
*
* We do not reject if the relation is already open, because it's quite
* likely that one or more layers of caller have it open. That means it
* is unsafe to use this entry point for alterations that could break
* existing query plans. On the assumption it's not used for such, we
* don't have to reject pending AFTER triggers, either.
*/
void
AlterTableInternal(Oid relid, List *cmds, bool recurse)
{
Relation rel;
LOCKMODE lockmode = AlterTableGetLockLevel(cmds);
rel = relation_open(relid, lockmode);
ATController(rel, cmds, recurse, lockmode);
}
/*
* AlterTableGetLockLevel
*
* Sets the overall lock level required for the supplied list of subcommands.
* Policy for doing this set according to needs of AlterTable(), see
* comments there for overall explanation.
*
* Function is called before and after parsing, so it must give same
* answer each time it is called. Some subcommands are transformed
* into other subcommand types, so the transform must never be made to a
* lower lock level than previously assigned. All transforms are noted below.
*
* Since this is called before we lock the table we cannot use table metadata
* to influence the type of lock we acquire.
*
* There should be no lockmodes hardcoded into the subcommand functions. All
* lockmode decisions for ALTER TABLE are made here only. The one exception is
* ALTER TABLE RENAME which is treated as a different statement type T_RenameStmt
* and does not travel through this section of code and cannot be combined with
* any of the subcommands given here.
*/
LOCKMODE
AlterTableGetLockLevel(List *cmds)
{
/*
* Late in 9.1 dev cycle a number of issues were uncovered with access
* to catalog relations, leading to the decision to re-enforce all DDL
* at AccessExclusiveLock level by default.
*
* The issues are that there is a pervasive assumption in the code that
* the catalogs will not be read unless an AccessExclusiveLock is held.
* If that rule is relaxed, we must protect against a number of potential
* effects - infrequent, but proven possible with test cases where
* multiple DDL operations occur in a stream against frequently accessed
* tables.
*
* 1. Catalog tables are read using SnapshotNow, which has a race bug
* that allows a scan to return no valid rows even when one is present
* in the case of a commit of a concurrent update of the catalog table.
* SnapshotNow also ignores transactions in progress, so takes the
* latest committed version without waiting for the latest changes.
*
* 2. Relcache needs to be internally consistent, so unless we lock the
* definition during reads we have no way to guarantee that.
*
* 3. Catcache access isn't coordinated at all so refreshes can occur at
* any time.
*/
#ifdef REDUCED_ALTER_TABLE_LOCK_LEVELS
ListCell *lcmd;
LOCKMODE lockmode = ShareUpdateExclusiveLock;
foreach(lcmd, cmds)
{
AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lcmd);
LOCKMODE cmd_lockmode = AccessExclusiveLock; /* default for compiler */
switch (cmd->subtype)
{
/*
* Need AccessExclusiveLock for these subcommands because they
* affect or potentially affect both read and write
* operations.
*
* New subcommand types should be added here by default.
*/
case AT_AddColumn: /* may rewrite heap, in some cases and visible
* to SELECT */
case AT_DropColumn: /* change visible to SELECT */
case AT_AddColumnToView: /* CREATE VIEW */
case AT_AlterColumnType: /* must rewrite heap */
case AT_DropConstraint: /* as DROP INDEX */
case AT_AddOids: /* must rewrite heap */
case AT_DropOids: /* calls AT_DropColumn */
case AT_EnableAlwaysRule: /* may change SELECT rules */
case AT_EnableReplicaRule: /* may change SELECT rules */
case AT_EnableRule: /* may change SELECT rules */
case AT_DisableRule: /* may change SELECT rules */
case AT_ChangeOwner: /* change visible to SELECT */
case AT_SetTableSpace: /* must rewrite heap */
case AT_DropNotNull: /* may change some SQL plans */
case AT_SetNotNull:
case AT_GenericOptions:
case AT_AlterColumnGenericOptions:
cmd_lockmode = AccessExclusiveLock;
break;
/*
* These subcommands affect write operations only.
*/
case AT_ColumnDefault:
case AT_ProcessedConstraint: /* becomes AT_AddConstraint */
case AT_AddConstraintRecurse: /* becomes AT_AddConstraint */
case AT_EnableTrig:
case AT_EnableAlwaysTrig:
case AT_EnableReplicaTrig:
case AT_EnableTrigAll:
case AT_EnableTrigUser:
case AT_DisableTrig:
case AT_DisableTrigAll:
case AT_DisableTrigUser:
case AT_AddIndex: /* from ADD CONSTRAINT */
case AT_AddIndexConstraint:
cmd_lockmode = ShareRowExclusiveLock;
break;
case 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.
*/
cmd_lockmode = ShareRowExclusiveLock;
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 = ShareRowExclusiveLock;
}
}
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.
*/
case AT_AddInherit:
case AT_DropInherit:
cmd_lockmode = ShareUpdateExclusiveLock;
break;
/*
* These subcommands affect implicit row type conversion. They
* have affects similar to CREATE/DROP CAST on queries. We
* don't provide for invalidating parse trees as a result of
* such changes. Do avoid concurrent pg_class updates,
* though.
*/
case AT_AddOf:
case AT_DropOf:
cmd_lockmode = ShareUpdateExclusiveLock;
/*
* 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:
case AT_ClusterOn:
case AT_DropCluster:
case AT_SetRelOptions:
case AT_ResetRelOptions:
case AT_ReplaceRelOptions:
case AT_SetOptions:
case AT_ResetOptions:
case AT_SetStorage:
case AT_ValidateConstraint:
cmd_lockmode = ShareUpdateExclusiveLock;
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;
}
#else
LOCKMODE lockmode = AccessExclusiveLock;
#endif
return lockmode;
}
static void
ATController(Relation rel, List *cmds, bool recurse, LOCKMODE lockmode)
{
List *wqueue = NIL;
ListCell *lcmd;
/* Phase 1: preliminary examination of commands, create work queue */
foreach(lcmd, cmds)
{
AlterTableCmd *cmd = (AlterTableCmd *) lfirst(lcmd);
ATPrepCmd(&wqueue, rel, cmd, recurse, false, lockmode);
}
/* Close the relation, but keep lock until commit */
relation_close(rel, NoLock);
/* Phase 2: update system catalogs */
ATRewriteCatalogs(&wqueue, lockmode);
/* Phase 3: scan/rewrite tables as needed */
ATRewriteTables(&wqueue, lockmode);
}
/*
* ATPrepCmd
*
* Traffic cop for ALTER TABLE Phase 1 operations, including simple
* recursion and permission checks.
*
* Caller must have acquired appropriate lock type on relation already.
* This lock should be held until commit.
*/
static void
ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
bool recurse, bool recursing, LOCKMODE lockmode)
{
AlteredTableInfo *tab;
int pass;
/* Find or create work queue entry for this table */
tab = ATGetQueueEntry(wqueue, rel);
/*
* Copy the original subcommand for each table. This avoids conflicts
* when different child tables need to make different parse
* transformations (for example, the same column may have different column
* numbers in different children).
*/
cmd = copyObject(cmd);
/*
* Do permissions checking, recursion to child tables if needed, and any
* additional phase-1 processing needed.
*/
switch (cmd->subtype)
{
case AT_AddColumn: /* ADD COLUMN */
ATSimplePermissions(rel,
ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
ATPrepAddColumn(wqueue, rel, recurse, recursing, cmd, lockmode);
/* Recursion occurs during execution phase */
pass = AT_PASS_ADD_COL;
break;
case AT_AddColumnToView: /* add column via CREATE OR REPLACE
* VIEW */
ATSimplePermissions(rel, ATT_VIEW);
ATPrepAddColumn(wqueue, rel, recurse, recursing, cmd, lockmode);
/* Recursion occurs during execution phase */
pass = AT_PASS_ADD_COL;
break;
case AT_ColumnDefault: /* ALTER COLUMN DEFAULT */
/*
* We allow defaults on views so that INSERT into a view can have
* default-ish behavior. This works because the rewriter
* substitutes default values into INSERTs before it expands
* rules.
*/
ATSimplePermissions(rel, ATT_TABLE | ATT_VIEW);
ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = cmd->def ? AT_PASS_ADD_CONSTR : AT_PASS_DROP;
break;
case AT_DropNotNull: /* ALTER COLUMN DROP NOT NULL */
ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = AT_PASS_DROP;
break;
case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */
ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);
ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = AT_PASS_ADD_CONSTR;
break;
case AT_SetStatistics: /* ALTER COLUMN SET STATISTICS */
ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* Performs own permission checks */
ATPrepSetStatistics(rel, cmd->name, cmd->def, lockmode);
pass = AT_PASS_MISC;
break;
case AT_SetOptions: /* ALTER COLUMN SET ( options ) */
case AT_ResetOptions: /* ALTER COLUMN RESET ( options ) */
ATSimplePermissions(rel, ATT_TABLE | ATT_INDEX | ATT_FOREIGN_TABLE);
/* This command never recurses */
pass = AT_PASS_MISC;
break;
case AT_SetStorage: /* ALTER COLUMN SET STORAGE */
ATSimplePermissions(rel, ATT_TABLE);
ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
case AT_DropColumn: /* DROP COLUMN */
ATSimplePermissions(rel,
ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
ATPrepDropColumn(wqueue, rel, recurse, recursing, cmd, lockmode);
/* Recursion occurs during execution phase */
pass = AT_PASS_DROP;
break;
case AT_AddIndex: /* ADD INDEX */
ATSimplePermissions(rel, ATT_TABLE);
/* This command never recurses */
/* No command-specific prep needed */
pass = AT_PASS_ADD_INDEX;
break;
case AT_AddConstraint: /* ADD CONSTRAINT */
ATSimplePermissions(rel, ATT_TABLE);
/* Recursion occurs during execution phase */
/* No command-specific prep needed except saving recurse flag */
if (recurse)
cmd->subtype = AT_AddConstraintRecurse;
pass = AT_PASS_ADD_CONSTR;
break;
case AT_AddIndexConstraint: /* ADD CONSTRAINT USING INDEX */
ATSimplePermissions(rel, ATT_TABLE);
/* This command never recurses */
/* No command-specific prep needed */
pass = AT_PASS_ADD_CONSTR;
break;
case AT_DropConstraint: /* DROP CONSTRAINT */
ATSimplePermissions(rel, ATT_TABLE);
/* Recursion occurs during execution phase */
/* No command-specific prep needed except saving recurse flag */
if (recurse)
cmd->subtype = AT_DropConstraintRecurse;
pass = AT_PASS_DROP;
break;
case AT_AlterColumnType: /* ALTER COLUMN TYPE */
ATSimplePermissions(rel,
ATT_TABLE | ATT_COMPOSITE_TYPE | ATT_FOREIGN_TABLE);
/* Performs own recursion */
ATPrepAlterColumnType(wqueue, tab, rel, recurse, recursing, cmd, lockmode);
pass = AT_PASS_ALTER_TYPE;
break;
case AT_AlterColumnGenericOptions:
ATSimplePermissions(rel, ATT_FOREIGN_TABLE);
/* This command never recurses */
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
case AT_ChangeOwner: /* ALTER OWNER */
/* This command never recurses */
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
case AT_ClusterOn: /* CLUSTER ON */
case AT_DropCluster: /* SET WITHOUT CLUSTER */
ATSimplePermissions(rel, ATT_TABLE);
/* These commands never recurse */
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
case AT_AddOids: /* SET WITH OIDS */
ATSimplePermissions(rel, ATT_TABLE);
if (!rel->rd_rel->relhasoids || recursing)
ATPrepAddOids(wqueue, rel, recurse, cmd, lockmode);
/* Recursion occurs during execution phase */
pass = AT_PASS_ADD_COL;
break;
case AT_DropOids: /* SET WITHOUT OIDS */
ATSimplePermissions(rel, ATT_TABLE);
/* Performs own recursion */
if (rel->rd_rel->relhasoids)
{
AlterTableCmd *dropCmd = makeNode(AlterTableCmd);
dropCmd->subtype = AT_DropColumn;
dropCmd->name = pstrdup("oid");
dropCmd->behavior = cmd->behavior;
ATPrepCmd(wqueue, rel, dropCmd, recurse, false, lockmode);
}
pass = AT_PASS_DROP;
break;
case AT_SetTableSpace: /* SET TABLESPACE */
ATSimplePermissions(rel, ATT_TABLE | ATT_INDEX);
/* This command never recurses */
ATPrepSetTableSpace(tab, rel, cmd->name, lockmode);
pass = AT_PASS_MISC; /* doesn't actually matter */
break;
case AT_SetRelOptions: /* SET (...) */
case AT_ResetRelOptions: /* RESET (...) */
case AT_ReplaceRelOptions: /* reset them all, then set just these */
ATSimplePermissions(rel, ATT_TABLE | ATT_INDEX | ATT_VIEW);
/* This command never recurses */
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
case AT_AddInherit: /* INHERIT */
ATSimplePermissions(rel, ATT_TABLE);
/* This command never recurses */
ATPrepAddInherit(rel);
pass = AT_PASS_MISC;
break;
case AT_ValidateConstraint: /* VALIDATE CONSTRAINT */
ATSimplePermissions(rel, ATT_TABLE);
/* Recursion occurs during execution phase */
/* No command-specific prep needed except saving recurse flag */
if (recurse)
cmd->subtype = AT_ValidateConstraintRecurse;
pass = AT_PASS_MISC;
break;
case AT_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:
case AT_EnableRule: /* ENABLE/DISABLE RULE variants */
case AT_EnableAlwaysRule:
case AT_EnableReplicaRule:
case AT_DisableRule:
case AT_DropInherit: /* NO INHERIT */
case AT_AddOf: /* OF */
case AT_DropOf: /* NOT OF */
ATSimplePermissions(rel, ATT_TABLE);
/* These commands never recurse */
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
case AT_GenericOptions:
ATSimplePermissions(rel, ATT_FOREIGN_TABLE);
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
(int) cmd->subtype);
pass = 0; /* keep compiler quiet */
break;
}
/* Add the subcommand to the appropriate list for phase 2 */
tab->subcmds[pass] = lappend(tab->subcmds[pass], cmd);
}
/*
* ATRewriteCatalogs
*
* Traffic cop for ALTER TABLE Phase 2 operations. Subcommands are
* dispatched in a "safe" execution order (designed to avoid unnecessary
* conflicts).
*/
static void
ATRewriteCatalogs(List **wqueue, LOCKMODE lockmode)
{
int pass;
ListCell *ltab;
/*
* We process all the tables "in parallel", one pass at a time. This is
* needed because we may have to propagate work from one table to another
* (specifically, ALTER TYPE on a foreign key's PK has to dispatch the
* re-adding of the foreign key constraint to the other table). Work can
* only be propagated into later passes, however.
*/
for (pass = 0; pass < AT_NUM_PASSES; pass++)
{
/* Go through each table that needs to be processed */
foreach(ltab, *wqueue)
{
AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);
List *subcmds = tab->subcmds[pass];
Relation rel;
ListCell *lcmd;
if (subcmds == NIL)
continue;
/*
* Appropriate lock was obtained by phase 1, needn't get it again
*/
rel = relation_open(tab->relid, NoLock);
foreach(lcmd, subcmds)
ATExecCmd(wqueue, tab, rel, (AlterTableCmd *) lfirst(lcmd), lockmode);
/*
* After the ALTER TYPE pass, do cleanup work (this is not done in
* ATExecAlterColumnType since it should be done only once if
* multiple columns of a table are altered).
*/
if (pass == AT_PASS_ALTER_TYPE)
ATPostAlterTypeCleanup(wqueue, tab, lockmode);
relation_close(rel, NoLock);
}
}
/* Check to see if a toast table must be added. */
foreach(ltab, *wqueue)
{
AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);
if (tab->relkind == RELKIND_RELATION)
AlterTableCreateToastTable(tab->relid, (Datum) 0);
}
}
/*
* ATExecCmd: dispatch a subcommand to appropriate execution routine
*/
static void
ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
AlterTableCmd *cmd, LOCKMODE lockmode)
{
switch (cmd->subtype)
{
case AT_AddColumn: /* ADD COLUMN */
case AT_AddColumnToView: /* add column via CREATE OR REPLACE
* VIEW */
ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
false, false, false, lockmode);
break;
case AT_AddColumnRecurse:
ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
false, true, false, lockmode);
break;
case AT_ColumnDefault: /* ALTER COLUMN DEFAULT */
ATExecColumnDefault(rel, cmd->name, cmd->def, lockmode);
break;
case AT_DropNotNull: /* ALTER COLUMN DROP NOT NULL */
ATExecDropNotNull(rel, cmd->name, lockmode);
break;
case AT_SetNotNull: /* ALTER COLUMN SET NOT NULL */
ATExecSetNotNull(tab, rel, cmd->name, lockmode);
break;
case AT_SetStatistics: /* ALTER COLUMN SET STATISTICS */
ATExecSetStatistics(rel, cmd->name, cmd->def, lockmode);
break;
case AT_SetOptions: /* ALTER COLUMN SET ( options ) */
ATExecSetOptions(rel, cmd->name, cmd->def, false, lockmode);
break;
case AT_ResetOptions: /* ALTER COLUMN RESET ( options ) */
ATExecSetOptions(rel, cmd->name, cmd->def, true, lockmode);
break;
case AT_SetStorage: /* ALTER COLUMN SET STORAGE */
ATExecSetStorage(rel, cmd->name, cmd->def, lockmode);
break;
case AT_DropColumn: /* DROP COLUMN */
ATExecDropColumn(wqueue, rel, cmd->name,
cmd->behavior, false, false, cmd->missing_ok, lockmode);
break;
case AT_DropColumnRecurse: /* DROP COLUMN with recursion */
ATExecDropColumn(wqueue, rel, cmd->name,
cmd->behavior, true, false, cmd->missing_ok, lockmode);
break;
case AT_AddIndex: /* ADD INDEX */
ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, false, lockmode);
break;
case AT_ReAddIndex: /* ADD INDEX */
ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, true, lockmode);
break;
case AT_AddConstraint: /* ADD CONSTRAINT */
ATExecAddConstraint(wqueue, tab, rel, (Constraint *) cmd->def,
false, lockmode);
break;
case AT_AddConstraintRecurse: /* ADD CONSTRAINT with recursion */
ATExecAddConstraint(wqueue, tab, rel, (Constraint *) cmd->def,
true, lockmode);
break;
case AT_AddIndexConstraint: /* ADD CONSTRAINT USING INDEX */
ATExecAddIndexConstraint(tab, rel, (IndexStmt *) cmd->def, lockmode);
break;
case AT_ValidateConstraint: /* VALIDATE CONSTRAINT */
ATExecValidateConstraint(rel, cmd->name, false, false, lockmode);
break;
case AT_ValidateConstraintRecurse: /* VALIDATE CONSTRAINT with
* recursion */
ATExecValidateConstraint(rel, cmd->name, true, false, lockmode);
break;
case AT_DropConstraint: /* DROP CONSTRAINT */
ATExecDropConstraint(rel, cmd->name, cmd->behavior,
false, false,
cmd->missing_ok, lockmode);
break;
case AT_DropConstraintRecurse: /* DROP CONSTRAINT with recursion */
ATExecDropConstraint(rel, cmd->name, cmd->behavior,
true, false,
cmd->missing_ok, lockmode);
break;
case AT_AlterColumnType: /* ALTER COLUMN TYPE */
ATExecAlterColumnType(tab, rel, cmd, lockmode);
break;
case AT_AlterColumnGenericOptions: /* ALTER COLUMN OPTIONS */
ATExecAlterColumnGenericOptions(rel, cmd->name, (List *) cmd->def, lockmode);
break;
case AT_ChangeOwner: /* ALTER OWNER */
ATExecChangeOwner(RelationGetRelid(rel),
get_role_oid(cmd->name, false),
false, lockmode);
break;
case AT_ClusterOn: /* CLUSTER ON */
ATExecClusterOn(rel, cmd->name, lockmode);
break;
case AT_DropCluster: /* SET WITHOUT CLUSTER */
ATExecDropCluster(rel, lockmode);
break;
case AT_AddOids: /* SET WITH OIDS */
/* Use the ADD COLUMN code, unless prep decided to do nothing */
if (cmd->def != NULL)
ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
true, false, false, lockmode);
break;
case AT_AddOidsRecurse: /* SET WITH OIDS */
/* Use the ADD COLUMN code, unless prep decided to do nothing */
if (cmd->def != NULL)
ATExecAddColumn(wqueue, tab, rel, (ColumnDef *) cmd->def,
true, true, false, lockmode);
break;
case AT_DropOids: /* SET WITHOUT OIDS */
/*
* Nothing to do here; we'll have generated a DropColumn
* subcommand to do the real work
*/
break;
case AT_SetTableSpace: /* SET TABLESPACE */
/*
* Nothing to do here; Phase 3 does the work
*/
break;
case AT_SetRelOptions: /* SET (...) */
case AT_ResetRelOptions: /* RESET (...) */
case AT_ReplaceRelOptions: /* replace entire option list */
ATExecSetRelOptions(rel, (List *) cmd->def, cmd->subtype, lockmode);
break;
case AT_EnableTrig: /* ENABLE TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
break;
case AT_EnableAlwaysTrig: /* ENABLE ALWAYS TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
TRIGGER_FIRES_ALWAYS, false, lockmode);
break;
case AT_EnableReplicaTrig: /* ENABLE REPLICA TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
TRIGGER_FIRES_ON_REPLICA, false, lockmode);
break;
case AT_DisableTrig: /* DISABLE TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
TRIGGER_DISABLED, false, lockmode);
break;
case AT_EnableTrigAll: /* ENABLE TRIGGER ALL */
ATExecEnableDisableTrigger(rel, NULL,
TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
break;
case AT_DisableTrigAll: /* DISABLE TRIGGER ALL */
ATExecEnableDisableTrigger(rel, NULL,
TRIGGER_DISABLED, false, lockmode);
break;
case AT_EnableTrigUser: /* ENABLE TRIGGER USER */
ATExecEnableDisableTrigger(rel, NULL,
TRIGGER_FIRES_ON_ORIGIN, true, lockmode);
break;
case AT_DisableTrigUser: /* DISABLE TRIGGER USER */
ATExecEnableDisableTrigger(rel, NULL,
TRIGGER_DISABLED, true, lockmode);
break;
case AT_EnableRule: /* ENABLE RULE name */
ATExecEnableDisableRule(rel, cmd->name,
RULE_FIRES_ON_ORIGIN, lockmode);
break;
case AT_EnableAlwaysRule: /* ENABLE ALWAYS RULE name */
ATExecEnableDisableRule(rel, cmd->name,
RULE_FIRES_ALWAYS, lockmode);
break;
case AT_EnableReplicaRule: /* ENABLE REPLICA RULE name */
ATExecEnableDisableRule(rel, cmd->name,
RULE_FIRES_ON_REPLICA, lockmode);
break;
case AT_DisableRule: /* DISABLE RULE name */
ATExecEnableDisableRule(rel, cmd->name,
RULE_DISABLED, lockmode);
break;
case AT_AddInherit:
ATExecAddInherit(rel, (RangeVar *) cmd->def, lockmode);
break;
case AT_DropInherit:
ATExecDropInherit(rel, (RangeVar *) cmd->def, lockmode);
break;
case AT_AddOf:
ATExecAddOf(rel, (TypeName *) cmd->def, lockmode);
break;
case AT_DropOf:
ATExecDropOf(rel, lockmode);
break;
case AT_GenericOptions:
ATExecGenericOptions(rel, (List *) cmd->def);
break;
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
(int) cmd->subtype);
break;
}
/*
* Bump the command counter to ensure the next subcommand in the sequence
* can see the changes so far
*/
CommandCounterIncrement();
}
/*
* ATRewriteTables: ALTER TABLE phase 3
*/
static void
ATRewriteTables(List **wqueue, LOCKMODE lockmode)
{
ListCell *ltab;
/* Go through each table that needs to be checked or rewritten */
foreach(ltab, *wqueue)
{
AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);
/* Foreign tables have no storage. */
if (tab->relkind == RELKIND_FOREIGN_TABLE)
continue;
/*
* If we change column data types or add/remove OIDs, the operation
* has to be propagated to tables that use this table's rowtype as a
* column type. tab->newvals will also be non-NULL in the case where
* we're adding a column with a default. We choose to forbid that
* case as well, since composite types might eventually support
* defaults.
*
* (Eventually we'll probably need to check for composite type
* dependencies even when we're just scanning the table without a
* rewrite, but at the moment a composite type does not enforce any
* constraints, so it's not necessary/appropriate to enforce them just
* during ALTER.)
*/
if (tab->newvals != NIL || tab->rewrite)
{
Relation rel;
rel = heap_open(tab->relid, NoLock);
find_composite_type_dependencies(rel->rd_rel->reltype, rel, NULL);
heap_close(rel, NoLock);
}
/*
* We only need to rewrite the table if at least one column needs to
* be recomputed, or we are adding/removing the OID column.
*/
if (tab->rewrite)
{
/* Build a temporary relation and copy data */
Relation OldHeap;
Oid OIDNewHeap;
Oid NewTableSpace;
OldHeap = heap_open(tab->relid, NoLock);
/*
* We don't support rewriting of system catalogs; there are too
* many corner cases and too little benefit. In particular this
* is certainly not going to work for mapped catalogs.
*/
if (IsSystemRelation(OldHeap))
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot rewrite system relation \"%s\"",
RelationGetRelationName(OldHeap))));
/*
* Don't allow rewrite on temp tables of other backends ... their
* local buffer manager is not going to cope.
*/
if (RELATION_IS_OTHER_TEMP(OldHeap))
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot rewrite temporary tables of other sessions")));
/*
* Select destination tablespace (same as original unless user
* requested a change)
*/
if (tab->newTableSpace)
NewTableSpace = tab->newTableSpace;
else
NewTableSpace = OldHeap->rd_rel->reltablespace;
heap_close(OldHeap, NoLock);
/* Create transient table that will receive the modified data */
OIDNewHeap = make_new_heap(tab->relid, NewTableSpace);
/*
* Copy the heap data into the new table with the desired
* modifications, and test the current data within the table
* against new constraints generated by ALTER TABLE commands.
*/
ATRewriteTable(tab, OIDNewHeap, lockmode);
/*
* Swap the physical files of the old and new heaps, then rebuild
* indexes and discard the old heap. We can use RecentXmin for
* the table's new relfrozenxid because we rewrote all the tuples
* in ATRewriteTable, so no older Xid remains in the table. Also,
* we never try to swap toast tables by content, since we have no
* interest in letting this code work on system catalogs.
*/
finish_heap_swap(tab->relid, OIDNewHeap,
false, false, true, RecentXmin);
}
else
{
/*
* Test the current data within the table against new constraints
* generated by ALTER TABLE commands, but don't rebuild data.
*/
if (tab->constraints != NIL || tab->new_notnull)
ATRewriteTable(tab, InvalidOid, lockmode);
/*
* If we had SET TABLESPACE but no reason to reconstruct tuples,
* just do a block-by-block copy.
*/
if (tab->newTableSpace)
ATExecSetTableSpace(tab->relid, tab->newTableSpace, lockmode);
}
}
/*
* Foreign key constraints are checked in a final pass, since (a) it's
* generally best to examine each one separately, and (b) it's at least
* theoretically possible that we have changed both relations of the
* foreign key, and we'd better have finished both rewrites before we try
* to read the tables.
*/
foreach(ltab, *wqueue)
{
AlteredTableInfo *tab = (AlteredTableInfo *) lfirst(ltab);
Relation rel = NULL;
ListCell *lcon;
foreach(lcon, tab->constraints)
{
NewConstraint *con = lfirst(lcon);
if (con->contype == CONSTR_FOREIGN)
{
Constraint *fkconstraint = (Constraint *) con->qual;
Relation refrel;
if (rel == NULL)
{
/* Long since locked, no need for another */
rel = heap_open(tab->relid, NoLock);
}
refrel = heap_open(con->refrelid, RowShareLock);
validateForeignKeyConstraint(fkconstraint->conname, rel, refrel,
con->refindid,
con->conid);
/*
* No need to mark the constraint row as validated, we did
* that when we inserted the row earlier.
*/
heap_close(refrel, NoLock);
}
}
if (rel)
heap_close(rel, NoLock);
}
}
/*
* ATRewriteTable: scan or rewrite one table
*
* OIDNewHeap is InvalidOid if we don't need to rewrite
*/
static void
ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
{
Relation oldrel;
Relation newrel;
TupleDesc oldTupDesc;
TupleDesc newTupDesc;
bool needscan = false;
List *notnull_attrs;