* tablecmds.c
* Commands for creating and altering table structures and settings
*
- * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.203 2006/10/04 00:29:51 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.273 2008/12/13 19:13:44 tgl Exp $
*
*-------------------------------------------------------------------------
*/
#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/pg_depend.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
+#include "catalog/pg_opclass.h"
+#include "catalog/pg_tablespace.h"
#include "catalog/pg_trigger.h"
#include "catalog/pg_type.h"
+#include "catalog/pg_type_fn.h"
+#include "catalog/storage.h"
#include "catalog/toasting.h"
#include "commands/cluster.h"
#include "commands/defrem.h"
+#include "commands/sequence.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
+#include "nodes/nodeFuncs.h"
+#include "nodes/parsenodes.h"
#include "optimizer/clauses.h"
#include "optimizer/plancat.h"
#include "optimizer/prep.h"
-#include "parser/analyze.h"
#include "parser/gramparse.h"
#include "parser/parse_clause.h"
#include "parser/parse_coerce.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/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.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"
/*
char *name; /* Constraint name, or NULL if none */
ConstrType contype; /* CHECK or FOREIGN */
Oid refrelid; /* PK rel, if FOREIGN */
+ Oid conid; /* OID of pg_constraint entry, if FOREIGN */
Node *qual; /* Check expr or FkConstraint struct */
List *qualstate; /* Execution state for CHECK */
} NewConstraint;
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.")},
+ {'\0', 0, NULL, NULL, NULL, NULL}
+};
+
static void truncate_check_rel(Relation rel);
static List *MergeAttributes(List *schema, List *supers, bool istemp,
List **supOids, List **supconstr, int *supOidCount);
-static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel);
-static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel);
+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 catalogRelation);
+ int16 seqNumber, Relation inhRelation);
static int findAttrByName(const char *attributeName, List *schema);
static void setRelhassubclassInRelation(Oid relationId, bool relhassubclass);
static void AlterIndexNamespaces(Relation classRel, Relation rel,
int numattrs, int16 *attnums,
Oid *opclasses);
static void validateForeignKeyConstraint(FkConstraint *fkconstraint,
- Relation rel, Relation pkrel);
+ Relation rel, Relation pkrel, Oid constraintOid);
static void createForeignKeyTriggers(Relation rel, FkConstraint *fkconstraint,
- Oid constrOid);
-static char *fkMatchTypeToString(char match_type);
+ Oid constraintOid);
static void ATController(Relation rel, List *cmds, bool recurse);
static void ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
bool recurse, bool recursing);
static void ATRewriteCatalogs(List **wqueue);
-static void ATExecCmd(AlteredTableInfo *tab, Relation rel, AlterTableCmd *cmd);
+static void ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
+ AlterTableCmd *cmd);
static void ATRewriteTables(List **wqueue);
static void ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap);
static AlteredTableInfo *ATGetQueueEntry(List **wqueue, Relation rel);
AlterTableCmd *cmd, bool recurse);
static void ATOneLevelRecursion(List **wqueue, Relation rel,
AlterTableCmd *cmd);
-static void find_composite_type_dependencies(Oid typeOid,
- const char *origTblName);
static void ATPrepAddColumn(List **wqueue, Relation rel, bool recurse,
AlterTableCmd *cmd);
static void ATExecAddColumn(AlteredTableInfo *tab, Relation rel,
bool recurse, bool recursing);
static void ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
IndexStmt *stmt, bool is_rebuild);
-static void ATExecAddConstraint(AlteredTableInfo *tab, Relation rel,
- Node *newConstraint);
+static void ATExecAddConstraint(List **wqueue,
+ AlteredTableInfo *tab, Relation rel,
+ Node *newConstraint, bool recurse);
+static void ATAddCheckConstraint(List **wqueue,
+ AlteredTableInfo *tab, Relation rel,
+ Constraint *constr,
+ bool recurse, bool recursing);
static void ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
FkConstraint *fkconstraint);
-static void ATPrepDropConstraint(List **wqueue, Relation rel,
- bool recurse, AlterTableCmd *cmd);
static void ATExecDropConstraint(Relation rel, const char *constrName,
- DropBehavior behavior, bool quiet);
+ DropBehavior behavior,
+ bool recurse, bool recursing);
static void ATPrepAlterColumnType(List **wqueue,
AlteredTableInfo *tab, Relation rel,
bool recurse, bool recursing,
static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace);
static void ATExecSetRelOptions(Relation rel, List *defList, bool isReset);
static void ATExecEnableDisableTrigger(Relation rel, char *trigname,
- bool enable, bool skip_system);
-static void ATExecAddInherits(Relation rel, RangeVar *parent);
-static void ATExecDropInherits(Relation rel, RangeVar *parent);
-static void copy_relation_data(Relation rel, SMgrRelation dst);
-static void update_ri_trigger_args(Oid relid,
- const char *oldname,
- const char *newname,
- bool fk_scan,
- bool update_relname);
+ char fires_when, bool skip_system);
+static void ATExecEnableDisableRule(Relation rel, char *rulename,
+ char fires_when);
+static void ATExecAddInherit(Relation rel, RangeVar *parent);
+static void ATExecDropInherit(Relation rel, RangeVar *parent);
+static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
+ ForkNumber forkNum, bool istemp);
/* ----------------------------------------------------------------
bool localHasOids;
int parentOidCount;
List *rawDefaults;
+ List *cookedDefaults;
Datum reloptions;
ListCell *listptr;
- int i;
AttrNumber attnum;
/*
}
/*
- * Select tablespace to use. If not specified, use default_tablespace
+ * Select tablespace to use. If not specified, use default tablespace
* (which may in turn default to database's default).
*/
if (stmt->tablespacename)
}
else
{
- tablespaceId = GetDefaultTablespace();
+ tablespaceId = GetDefaultTablespace(stmt->relation->istemp);
/* note InvalidOid is OK in this case */
}
- /*
- * Parse and validate reloptions, if any.
- */
- reloptions = transformRelOptions((Datum) 0, stmt->options, true, false);
-
- (void) heap_reloptions(relkind, reloptions, true);
-
/* Check permissions except when using database's default */
- if (OidIsValid(tablespaceId))
+ if (OidIsValid(tablespaceId) && tablespaceId != MyDatabaseTableSpace)
{
AclResult aclresult;
get_tablespace_name(tablespaceId));
}
+ /*
+ * Parse and validate reloptions, if any.
+ */
+ reloptions = transformRelOptions((Datum) 0, stmt->options, true, false);
+
+ (void) heap_reloptions(relkind, reloptions, true);
+
/*
* Look up inheritance ancestors and generate relation schema, including
* inherited attributes.
&inheritOids, &old_constraints, &parentOidCount);
/*
- * Create a relation descriptor from the relation schema and create the
- * relation. Note that in this stage only inherited (pre-cooked) defaults
- * and constraints will be included into the new relation.
- * (BuildDescForRelation takes care of the inherited defaults, but we have
- * to copy inherited constraints here.)
+ * 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);
- if (old_constraints != NIL)
+ /*
+ * 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)
{
- ConstrCheck *check = (ConstrCheck *)
- palloc0(list_length(old_constraints) * sizeof(ConstrCheck));
- int ncheck = 0;
+ ColumnDef *colDef = lfirst(listptr);
- foreach(listptr, old_constraints)
+ attnum++;
+
+ if (colDef->raw_default != NULL)
{
- Constraint *cdef = (Constraint *) lfirst(listptr);
- bool dup = false;
+ RawColumnDefault *rawEnt;
- if (cdef->contype != CONSTR_CHECK)
- continue;
- Assert(cdef->name != NULL);
- Assert(cdef->raw_expr == NULL && cdef->cooked_expr != NULL);
+ Assert(colDef->cooked_default == NULL);
- /*
- * In multiple-inheritance situations, it's possible to inherit
- * the same grandparent constraint through multiple parents.
- * Hence, discard inherited constraints that match as to both name
- * and expression. Otherwise, gripe if the names conflict.
- */
- for (i = 0; i < ncheck; i++)
- {
- if (strcmp(check[i].ccname, cdef->name) != 0)
- continue;
- if (strcmp(check[i].ccbin, cdef->cooked_expr) == 0)
- {
- dup = true;
- break;
- }
- ereport(ERROR,
- (errcode(ERRCODE_DUPLICATE_OBJECT),
- errmsg("duplicate check constraint name \"%s\"",
- cdef->name)));
- }
- if (!dup)
- {
- check[ncheck].ccname = cdef->name;
- check[ncheck].ccbin = pstrdup(cdef->cooked_expr);
- ncheck++;
- }
+ rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));
+ rawEnt->attnum = attnum;
+ rawEnt->raw_default = colDef->raw_default;
+ rawDefaults = lappend(rawDefaults, rawEnt);
+ descriptor->attrs[attnum - 1]->atthasdef = true;
}
- if (ncheck > 0)
+ else if (colDef->cooked_default != NULL)
{
- if (descriptor->constr == NULL)
- {
- descriptor->constr = (TupleConstr *) palloc(sizeof(TupleConstr));
- descriptor->constr->defval = NULL;
- descriptor->constr->num_defval = 0;
- descriptor->constr->has_not_null = false;
- }
- descriptor->constr->num_check = ncheck;
- descriptor->constr->check = check;
+ CookedConstraint *cooked;
+
+ cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
+ cooked->contype = CONSTR_DEFAULT;
+ cooked->name = NULL;
+ cooked->attnum = attnum;
+ cooked->expr = stringToNode(colDef->cooked_default);
+ cooked->is_local = true; /* not used for defaults */
+ cooked->inhcount = 0; /* ditto */
+ 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,
GetUserId(),
descriptor,
+ list_concat(cookedDefaults,
+ old_constraints),
relkind,
false,
localHasOids,
* 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.
- *
- * First, scan schema to find new column defaults.
- */
- rawDefaults = NIL;
- attnum = 0;
-
- foreach(listptr, schema)
- {
- ColumnDef *colDef = lfirst(listptr);
-
- attnum++;
-
- if (colDef->raw_default != NULL)
- {
- RawColumnDefault *rawEnt;
-
- Assert(colDef->cooked_default == NULL);
-
- rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));
- rawEnt->attnum = attnum;
- rawEnt->raw_default = colDef->raw_default;
- rawDefaults = lappend(rawDefaults, rawEnt);
- }
- }
-
- /*
- * Parse and add the defaults/constraints, if any.
*/
if (rawDefaults || stmt->constraints)
- AddRelationRawConstraints(rel, rawDefaults, stmt->constraints);
+ AddRelationNewConstraints(rel, rawDefaults, stmt->constraints,
+ true, true);
/*
* Clean up. We keep lock on new relation (although it shouldn't be
}
/*
- * RemoveRelation
- * Deletes a relation.
+ * 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(wentry->drophint_msg) : 0));
+}
+
+/*
+ * RemoveRelations
+ * Implements DROP TABLE, DROP INDEX, DROP SEQUENCE, DROP VIEW
*/
void
-RemoveRelation(const RangeVar *relation, DropBehavior behavior)
+RemoveRelations(DropStmt *drop)
{
- Oid relOid;
- ObjectAddress object;
+ ObjectAddresses *objects;
+ char relkind;
+ ListCell *cell;
- relOid = RangeVarGetRelid(relation, false);
+ /*
+ * 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.
+ */
- object.classId = RelationRelationId;
- object.objectId = relOid;
- object.objectSubId = 0;
+ /* Determine required relkind */
+ switch (drop->removeType)
+ {
+ case OBJECT_TABLE:
+ relkind = RELKIND_RELATION;
+ break;
- performDeletion(&object, behavior);
+ case OBJECT_INDEX:
+ relkind = RELKIND_INDEX;
+ break;
+
+ case OBJECT_SEQUENCE:
+ relkind = RELKIND_SEQUENCE;
+ break;
+
+ case OBJECT_VIEW:
+ relkind = RELKIND_VIEW;
+ 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;
+ HeapTuple tuple;
+ Form_pg_class classform;
+ ObjectAddress obj;
+
+ /*
+ * 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 */
+ relOid = RangeVarGetRelid(rel, true);
+
+ /* Not there? */
+ if (!OidIsValid(relOid))
+ {
+ DropErrorMsgNonExistent(rel->relname, relkind, drop->missing_ok);
+ continue;
+ }
+
+ /*
+ * 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 --- that most likely means it isn't an
+ * index, and we'll fail below.
+ */
+ if (relkind == RELKIND_INDEX)
+ {
+ tuple = SearchSysCache(INDEXRELID,
+ ObjectIdGetDatum(relOid),
+ 0, 0, 0);
+ if (HeapTupleIsValid(tuple))
+ {
+ Form_pg_index index = (Form_pg_index) GETSTRUCT(tuple);
+
+ LockRelationOid(index->indrelid, AccessExclusiveLock);
+ ReleaseSysCache(tuple);
+ }
+ }
+
+ /* Get the lock before trying to fetch the syscache entry */
+ LockRelationOid(relOid, AccessExclusiveLock);
+
+ tuple = SearchSysCache(RELOID,
+ ObjectIdGetDatum(relOid),
+ 0, 0, 0);
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for relation %u", relOid);
+ 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)));
+
+ /* OK, we're ready to delete this one */
+ obj.classId = RelationRelationId;
+ obj.objectId = relOid;
+ obj.objectSubId = 0;
+
+ add_exact_object_address(&obj, objects);
+
+ ReleaseSysCache(tuple);
+ }
+
+ performMultipleDeletions(objects, drop->behavior);
+
+ free_object_addresses(objects);
}
/*
{
List *rels = NIL;
List *relids = NIL;
+ List *seq_relids = NIL;
+ EState *estate;
+ ResultRelInfo *resultRelInfos;
+ ResultRelInfo *resultRelInfo;
ListCell *cell;
/*
Relation rel;
rel = heap_openrv(rv, AccessExclusiveLock);
+ /* don't throw error for "TRUNCATE foo, foo" */
+ if (list_member_oid(relids, RelationGetRelid(rel)))
+ {
+ heap_close(rel, AccessExclusiveLock);
+ continue;
+ }
truncate_check_rel(rel);
rels = lappend(rels, rel);
relids = lappend_oid(relids, RelationGetRelid(rel));
#endif
/*
- * Also check for pending AFTER trigger events on the target relations. We
- * can't just leave those be, since they will try to fetch tuples that the
- * TRUNCATE removes.
+ * If we are asked to restart sequences, find all the sequences,
+ * lock them (we only need AccessShareLock because that's all that
+ * ALTER SEQUENCE takes), 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, AccessShareLock);
+
+ /* 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.
+ */
+ 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 */
+ CMD_DELETE, /* don't need any index info */
+ false);
+ 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.
*/
- AfterTriggerCheckTruncate(relids);
+ resultRelInfo = resultRelInfos;
+ foreach(cell, rels)
+ {
+ estate->es_result_relation_info = resultRelInfo;
+ ExecBSTruncateTriggers(estate, resultRelInfo);
+ resultRelInfo++;
+ }
/*
* OK, truncate each table.
* the relfilenode value. The old storage file is scheduled for
* deletion at commit.
*/
- setNewRelfilenode(rel);
+ setNewRelfilenode(rel, RecentXmin);
heap_relid = RelationGetRelid(rel);
toast_relid = rel->rd_rel->reltoastrelid;
- heap_close(rel, NoLock);
-
/*
* The same for the toast table, if any.
*/
if (OidIsValid(toast_relid))
{
rel = relation_open(toast_relid, AccessExclusiveLock);
- setNewRelfilenode(rel);
+ setNewRelfilenode(rel, RecentXmin);
heap_close(rel, NoLock);
}
*/
reindex_relation(heap_relid, true);
}
+
+ /*
+ * 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);
+ }
+
+ /*
+ * Lastly, restart any owned sequences if we were asked to. This is done
+ * last because it's nontransactional: restarts will not roll back if
+ * we abort later. Hence it's important to postpone them as long as
+ * possible. (This is also a big reason why we locked and
+ * permission-checked the sequences beforehand.)
+ */
+ if (stmt->restart_seqs)
+ {
+ List *options = list_make1(makeDefElem("restart", NULL));
+
+ foreach(cell, seq_relids)
+ {
+ Oid seq_relid = lfirst_oid(cell);
+
+ AlterSequenceInternal(seq_relid, options);
+ }
+ }
}
/*
static void
truncate_check_rel(Relation rel)
{
+ AclResult aclresult;
+
/* Only allow truncate on regular tables */
if (rel->rd_rel->relkind != RELKIND_RELATION)
ereport(ERROR,
RelationGetRelationName(rel))));
/* Permissions checks */
- if (!pg_class_ownercheck(RelationGetRelid(rel), GetUserId()))
- aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
+ 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_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");
}
/*----------
if (strcmp(coldef->colname, restdef->colname) == 0)
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_COLUMN),
- errmsg("column \"%s\" duplicated",
+ errmsg("column \"%s\" specified more than once",
coldef->colname)));
}
}
if (list_member_oid(parentOids, RelationGetRelid(relation)))
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_TABLE),
- errmsg("inherited relation \"%s\" duplicated",
- parent->relname)));
+ errmsg("relation \"%s\" would be inherited from more than once",
+ parent->relname)));
parentOids = lappend_oid(parentOids, RelationGetRelid(relation));
exist_attno = findAttrByName(attributeName, inhSchema);
if (exist_attno > 0)
{
+ Oid defTypeId;
+ int32 deftypmod;
+
/*
* Yes, try to merge the two column definitions. They must
* have the same type and typmod.
(errmsg("merging multiple inherited definitions of column \"%s\"",
attributeName)));
def = (ColumnDef *) list_nth(inhSchema, exist_attno - 1);
- if (typenameTypeId(NULL, def->typename) != attribute->atttypid ||
- def->typename->typmod != attribute->atttypmod)
+ defTypeId = typenameTypeId(NULL, def->typename, &deftypmod);
+ if (defTypeId != attribute->atttypid ||
+ deftypmod != attribute->atttypmod)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("inherited column \"%s\" has a type conflict",
}
/*
- * Now copy the constraints of this parent, adjusting attnos using the
- * completed newattno[] map
+ * 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)
{
for (i = 0; i < constr->num_check; i++)
{
- Constraint *cdef = makeNode(Constraint);
+ char *name = check[i].ccname;
Node *expr;
- cdef->contype = CONSTR_CHECK;
- cdef->name = pstrdup(check[i].ccname);
- cdef->raw_expr = NULL;
/* adjust varattnos of ccbin here */
expr = stringToNode(check[i].ccbin);
change_varattnos_of_a_node(expr, newattno);
- cdef->cooked_expr = nodeToString(expr);
- constraints = lappend(constraints, cdef);
+
+ /* 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->is_local = false;
+ cooked->inhcount = 1;
+ constraints = lappend(constraints, cooked);
+ }
}
}
if (exist_attno > 0)
{
ColumnDef *def;
+ Oid defTypeId,
+ newTypeId;
+ int32 deftypmod,
+ newtypmod;
/*
* Yes, try to merge the two column definitions. They must
(errmsg("merging column \"%s\" with inherited definition",
attributeName)));
def = (ColumnDef *) list_nth(inhSchema, exist_attno - 1);
- if (typenameTypeId(NULL, def->typename) != typenameTypeId(NULL, newdef->typename) ||
- def->typename->typmod != newdef->typename->typmod)
+ defTypeId = typenameTypeId(NULL, def->typename, &deftypmod);
+ newTypeId = typenameTypeId(NULL, newdef->typename, &newtypmod);
+ if (defTypeId != newTypeId || deftypmod != newtypmod)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("column \"%s\" has a type conflict",
return schema;
}
+
/*
- * Varattnos of pg_constraint.conbin must be rewritten when subclasses inherit
- * constraints from parent classes, since the inherited attributes could
- * be given different column numbers in multiple-inheritance cases.
+ * MergeCheckConstraint
+ * Try to merge an inherited CHECK constraint with previous ones
*
- * Note that the passed node tree is modified in place!
+ * If we inherit identically-named constraints from multiple parents, we must
+ * merge them, or throw an error if they don't have identical definitions.
*
- * This function is used elsewhere such as in analyze.c
+ * 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.
*/
-
-void
-change_varattnos_of_a_node(Node *node, const AttrNumber *newattno)
+static bool
+MergeCheckConstraint(List *constraints, char *name, Node *expr)
{
- change_varattnos_walker(node, newattno);
-}
+ ListCell *lc;
-/* Generate a map for change_varattnos_of_a_node from two tupledesc's. */
+ foreach(lc, constraints)
+ {
+ CookedConstraint *ccon = (CookedConstraint *) lfirst(lc);
-AttrNumber *
-varattnos_map(TupleDesc old, TupleDesc new)
-{
- int i,
- j;
- AttrNumber *attmap = palloc0(sizeof(AttrNumber) * old->natts);
+ Assert(ccon->contype == CONSTR_CHECK);
- for (i = 1; i <= old->natts; i++)
- {
- if (old->attrs[i - 1]->attisdropped)
- {
- attmap[i - 1] = 0;
+ /* Non-matching names never conflict */
+ if (strcmp(ccon->name, name) != 0)
continue;
+
+ if (equal(expr, ccon->expr))
+ {
+ /* OK to merge */
+ ccon->inhcount++;
+ return true;
}
- for (j = 1; j <= new->natts; j++)
- if (!strcmp(NameStr(old->attrs[i - 1]->attname), NameStr(new->attrs[j - 1]->attname)))
- attmap[i - 1] = j;
+
+ ereport(ERROR,
+ (errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("check constraint name \"%s\" appears multiple times but with different expressions",
+ name)));
}
- return attmap;
+
+ return false;
}
+
/*
- * Generate a map for change_varattnos_of_a_node from a tupledesc and a list of
- * ColumnDefs
+ * 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!
*/
-AttrNumber *
-varattnos_map_schema(TupleDesc old, List *schema)
+void
+change_varattnos_of_a_node(Node *node, const AttrNumber *newattno)
{
- int i;
- AttrNumber *attmap = palloc0(sizeof(AttrNumber) * old->natts);
-
- for (i = 1; i <= old->natts; i++)
- {
- if (old->attrs[i - 1]->attisdropped)
- {
- attmap[i - 1] = 0;
- continue;
- }
- attmap[i - 1] = findAttrByName(NameStr(old->attrs[i - 1]->attname), schema);
- }
- return attmap;
+ /* no setup needed, so away we go */
+ (void) change_varattnos_walker(node, newattno);
}
static bool
* currently.
*/
Assert(newattno[var->varattno - 1] > 0);
- var->varattno = newattno[var->varattno - 1];
+ var->varattno = var->varoattno = newattno[var->varattno - 1];
}
return false;
}
(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 old, TupleDesc new)
+{
+ AttrNumber *attmap;
+ int i,
+ j;
+
+ 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 */
+
+ for (j = 1; j <= new->natts; j++)
+ {
+ if (strcmp(NameStr(old->attrs[i - 1]->attname),
+ NameStr(new->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.
seqNumber = 1;
foreach(entry, supers)
{
- StoreCatalogInheritance1(relationId, lfirst_oid(entry), seqNumber, relation);
+ 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 relation)
+ int16 seqNumber, Relation inhRelation)
{
+ TupleDesc desc = RelationGetDescr(inhRelation);
Datum datum[Natts_pg_inherits];
- char nullarr[Natts_pg_inherits];
+ bool nullarr[Natts_pg_inherits];
ObjectAddress childobject,
parentobject;
HeapTuple tuple;
- TupleDesc desc = RelationGetDescr(relation);
- datum[0] = ObjectIdGetDatum(relationId); /* inhrel */
+ /*
+ * Make the pg_inherits entry
+ */
+ datum[0] = ObjectIdGetDatum(relationId); /* inhrelid */
datum[1] = ObjectIdGetDatum(parentOid); /* inhparent */
datum[2] = Int16GetDatum(seqNumber); /* inhseqno */
- nullarr[0] = ' ';
- nullarr[1] = ' ';
- nullarr[2] = ' ';
+ nullarr[0] = false;
+ nullarr[1] = false;
+ nullarr[2] = false;
- tuple = heap_formtuple(desc, datum, nullarr);
+ tuple = heap_form_tuple(desc, datum, nullarr);
- simple_heap_insert(relation, tuple);
+ simple_heap_insert(inhRelation, tuple);
- CatalogUpdateIndexes(relation, tuple);
+ CatalogUpdateIndexes(inhRelation, tuple);
heap_freetuple(tuple);
heap_close(attrelation, RowExclusiveLock);
+ relation_close(targetrelation, NoLock); /* close rel but keep lock */
+}
+
+
+/*
+ * Execute ALTER TABLE/INDEX/SEQUENCE/VIEW RENAME
+ *
+ * Caller has already done permissions checks.
+ */
+void
+RenameRelation(Oid myrelid, const char *newrelname, ObjectType reltype)
+{
+ Relation targetrelation;
+ Oid namespaceId;
+ char relkind;
+
/*
- * Update att name in any RI triggers associated with the relation.
+ * Grab an exclusive lock on the target table, index, sequence or view,
+ * which we will NOT release until end of transaction.
*/
- if (targetrelation->rd_rel->reltriggers > 0)
- {
- /* update tgargs column reference where att is primary key */
- update_ri_trigger_args(RelationGetRelid(targetrelation),
- oldattname, newattname,
- false, false);
- /* update tgargs column reference where att is foreign key */
- update_ri_trigger_args(RelationGetRelid(targetrelation),
- oldattname, newattname,
- true, false);
- }
+ targetrelation = relation_open(myrelid, AccessExclusiveLock);
- relation_close(targetrelation, NoLock); /* close rel but keep lock */
+ namespaceId = RelationGetNamespace(targetrelation);
+ relkind = targetrelation->rd_rel->relkind;
+
+ /*
+ * For compatibility with prior releases, we don't complain if ALTER TABLE
+ * or ALTER INDEX is used to rename a sequence or view.
+ */
+ if (reltype == OBJECT_SEQUENCE && relkind != RELKIND_SEQUENCE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a sequence",
+ RelationGetRelationName(targetrelation))));
+
+ if (reltype == OBJECT_VIEW && relkind != RELKIND_VIEW)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a view",
+ RelationGetRelationName(targetrelation))));
+
+ /*
+ * Don't allow ALTER TABLE on composite types.
+ * We want people to use ALTER TYPE for that.
+ */
+ if (relkind == RELKIND_COMPOSITE_TYPE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is a composite type",
+ RelationGetRelationName(targetrelation)),
+ errhint("Use ALTER TYPE instead.")));
+
+ /* Do the work */
+ RenameRelationInternal(myrelid, newrelname, namespaceId);
+
+ /*
+ * Close rel, but keep exclusive lock!
+ */
+ relation_close(targetrelation, NoLock);
}
/*
- * renamerel - change the name of a relation
+ * 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
* sequence, AFAIK there's no need for it to be there.
*/
void
-renamerel(Oid myrelid, const char *newrelname)
+RenameRelationInternal(Oid myrelid, const char *newrelname, Oid namespaceId)
{
Relation targetrelation;
Relation relrelation; /* for RELATION relation */
HeapTuple reltup;
- Oid namespaceId;
- char *oldrelname;
- char relkind;
- bool relhastriggers;
+ Form_pg_class relform;
/*
- * Grab an exclusive lock on the target table or index, which we will NOT
- * release until end of transaction.
+ * 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);
- oldrelname = pstrdup(RelationGetRelationName(targetrelation));
- namespaceId = RelationGetNamespace(targetrelation);
-
- if (!allowSystemTableMods && IsSystemRelation(targetrelation))
- ereport(ERROR,
- (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
- errmsg("permission denied: \"%s\" is a system catalog",
- RelationGetRelationName(targetrelation))));
-
- relkind = targetrelation->rd_rel->relkind;
- relhastriggers = (targetrelation->rd_rel->reltriggers > 0);
-
/*
* Find relation's pg_class tuple, and make sure newrelname isn't in use.
*/
relrelation = heap_open(RelationRelationId, RowExclusiveLock);
reltup = SearchSysCacheCopy(RELOID,
- PointerGetDatum(myrelid),
+ ObjectIdGetDatum(myrelid),
0, 0, 0);
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,
* Update pg_class tuple with new relname. (Scribbling on reltup is OK
* because it's a copy...)
*/
- namestrcpy(&(((Form_pg_class) GETSTRUCT(reltup))->relname), newrelname);
+ namestrcpy(&(relform->relname), newrelname);
simple_heap_update(relrelation, &reltup->t_self, reltup);
/*
* Also rename the associated type, if any.
*/
- if (relkind != RELKIND_INDEX)
- TypeRename(oldrelname, namespaceId, newrelname);
+ if (OidIsValid(targetrelation->rd_rel->reltype))
+ RenameTypeInternal(targetrelation->rd_rel->reltype,
+ newrelname, namespaceId);
/*
- * Update rel name in any RI triggers associated with the relation.
+ * Also rename the associated constraint, if any.
*/
- if (relhastriggers)
+ if (targetrelation->rd_rel->relkind == RELKIND_INDEX)
{
- /* update tgargs where relname is primary key */
- update_ri_trigger_args(myrelid,
- oldrelname,
- newrelname,
- false, true);
- /* update tgargs where relname is foreign key */
- update_ri_trigger_args(myrelid,
- oldrelname,
- newrelname,
- true, true);
+ Oid constraintId = get_index_constraint(myrelid);
+
+ if (OidIsValid(constraintId))
+ RenameConstraintById(constraintId, newrelname);
}
/*
}
/*
- * Scan pg_trigger for RI triggers that are on the specified relation
- * (if fk_scan is false) or have it as the tgconstrrel (if fk_scan
- * is true). Update RI trigger args fields matching oldname to contain
- * newname instead. If update_relname is true, examine the relname
- * fields; otherwise examine the attname fields.
+ * 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 AccessExclusiveLock 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.
*/
-static void
-update_ri_trigger_args(Oid relid,
- const char *oldname,
- const char *newname,
- bool fk_scan,
- bool update_relname)
+void
+CheckTableNotInUse(Relation rel, const char *stmt)
{
- Relation tgrel;
- ScanKeyData skey[1];
- SysScanDesc trigscan;
- HeapTuple tuple;
- Datum values[Natts_pg_trigger];
- char nulls[Natts_pg_trigger];
- char replaces[Natts_pg_trigger];
-
- tgrel = heap_open(TriggerRelationId, RowExclusiveLock);
- if (fk_scan)
- {
- ScanKeyInit(&skey[0],
- Anum_pg_trigger_tgconstrrelid,
- BTEqualStrategyNumber, F_OIDEQ,
- ObjectIdGetDatum(relid));
- trigscan = systable_beginscan(tgrel, TriggerConstrRelidIndexId,
- true, SnapshotNow,
- 1, skey);
- }
- else
- {
- ScanKeyInit(&skey[0],
- Anum_pg_trigger_tgrelid,
- BTEqualStrategyNumber, F_OIDEQ,
- ObjectIdGetDatum(relid));
- trigscan = systable_beginscan(tgrel, TriggerRelidNameIndexId,
- true, SnapshotNow,
- 1, skey);
- }
-
- while ((tuple = systable_getnext(trigscan)) != NULL)
- {
- Form_pg_trigger pg_trigger = (Form_pg_trigger) GETSTRUCT(tuple);
- bytea *val;
- bytea *newtgargs;
- bool isnull;
- int tg_type;
- bool examine_pk;
- bool changed;
- int tgnargs;
- int i;
- int newlen;
- const char *arga[RI_MAX_ARGUMENTS];
- const char *argp;
-
- tg_type = RI_FKey_trigger_type(pg_trigger->tgfoid);
- if (tg_type == RI_TRIGGER_NONE)
- {
- /* Not an RI trigger, forget it */
- continue;
- }
-
- /*
- * It is an RI trigger, so parse the tgargs bytea.
- *
- * NB: we assume the field will never be compressed or moved out of
- * line; so does trigger.c ...
- */
- tgnargs = pg_trigger->tgnargs;
- val = (bytea *)
- DatumGetPointer(fastgetattr(tuple,
- Anum_pg_trigger_tgargs,
- tgrel->rd_att, &isnull));
- if (isnull || tgnargs < RI_FIRST_ATTNAME_ARGNO ||
- tgnargs > RI_MAX_ARGUMENTS)
- {
- /* This probably shouldn't happen, but ignore busted triggers */
- continue;
- }
- argp = (const char *) VARDATA(val);
- for (i = 0; i < tgnargs; i++)
- {
- arga[i] = argp;
- argp += strlen(argp) + 1;
- }
-
- /*
- * Figure out which item(s) to look at. If the trigger is primary-key
- * type and attached to my rel, I should look at the PK fields; if it
- * is foreign-key type and attached to my rel, I should look at the FK
- * fields. But the opposite rule holds when examining triggers found
- * by tgconstrrel search.
- */
- examine_pk = (tg_type == RI_TRIGGER_PK) == (!fk_scan);
-
- changed = false;
- if (update_relname)
- {
- /* Change the relname if needed */
- i = examine_pk ? RI_PK_RELNAME_ARGNO : RI_FK_RELNAME_ARGNO;
- if (strcmp(arga[i], oldname) == 0)
- {
- arga[i] = newname;
- changed = true;
- }
- }
- else
- {
- /* Change attname(s) if needed */
- i = examine_pk ? RI_FIRST_ATTNAME_ARGNO + RI_KEYPAIR_PK_IDX :
- RI_FIRST_ATTNAME_ARGNO + RI_KEYPAIR_FK_IDX;
- for (; i < tgnargs; i += 2)
- {
- if (strcmp(arga[i], oldname) == 0)
- {
- arga[i] = newname;
- changed = true;
- }
- }
- }
-
- if (!changed)
- {
- /* Don't need to update this tuple */
- continue;
- }
-
- /*
- * Construct modified tgargs bytea.
- */
- newlen = VARHDRSZ;
- for (i = 0; i < tgnargs; i++)
- newlen += strlen(arga[i]) + 1;
- newtgargs = (bytea *) palloc(newlen);
- VARATT_SIZEP(newtgargs) = newlen;
- newlen = VARHDRSZ;
- for (i = 0; i < tgnargs; i++)
- {
- strcpy(((char *) newtgargs) + newlen, arga[i]);
- newlen += strlen(arga[i]) + 1;
- }
-
- /*
- * Build modified tuple.
- */
- for (i = 0; i < Natts_pg_trigger; i++)
- {
- values[i] = (Datum) 0;
- replaces[i] = ' ';
- nulls[i] = ' ';
- }
- values[Anum_pg_trigger_tgargs - 1] = PointerGetDatum(newtgargs);
- replaces[Anum_pg_trigger_tgargs - 1] = 'r';
-
- tuple = heap_modifytuple(tuple, RelationGetDescr(tgrel), values, nulls, replaces);
-
- /*
- * Update pg_trigger and its indexes
- */
- simple_heap_update(tgrel, &tuple->t_self, tuple);
-
- CatalogUpdateIndexes(tgrel, tuple);
-
- /*
- * Invalidate trigger's relation's relcache entry so that other
- * backends (and this one too!) are sent SI message to make them
- * rebuild relcache entries. (Ideally this should happen
- * automatically...)
- *
- * We can skip this for triggers on relid itself, since that relcache
- * flush will happen anyway due to the table or column rename. We
- * just need to catch the far ends of RI relationships.
- */
- pg_trigger = (Form_pg_trigger) GETSTRUCT(tuple);
- if (pg_trigger->tgrelid != relid)
- CacheInvalidateRelcacheByRelid(pg_trigger->tgrelid);
-
- /* free up our scratch memory */
- pfree(newtgargs);
- heap_freetuple(tuple);
- }
-
- systable_endscan(trigscan);
+ int expected_refcnt;
- heap_close(tgrel, RowExclusiveLock);
-
- /*
- * Increment cmd counter to make updates visible; this is needed in case
- * the same tuple has to be updated again by next pass (can happen in case
- * of a self-referential FK relationship).
- */
- CommandCounterIncrement();
+ 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))));
}
/*
* 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 do no explicit recursion, since phase 1 already did it).
+ * 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
void
AlterTable(AlterTableStmt *stmt)
{
- ATController(relation_openrv(stmt->relation, AccessExclusiveLock),
- stmt->cmds,
- interpretInhOption(stmt->relation->inhOpt));
+ Relation rel = relation_openrv(stmt->relation, AccessExclusiveLock);
+
+ CheckTableNotInUse(rel, "ALTER TABLE");
+
+ /* Check relation type against type specified in the ALTER command */
+ switch (stmt->relkind)
+ {
+ case OBJECT_TABLE:
+ /*
+ * For mostly-historical reasons, we allow ALTER TABLE to apply
+ * to all relation types.
+ */
+ break;
+
+ case OBJECT_INDEX:
+ if (rel->rd_rel->relkind != RELKIND_INDEX)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not an index",
+ RelationGetRelationName(rel))));
+ break;
+
+ case OBJECT_SEQUENCE:
+ if (rel->rd_rel->relkind != RELKIND_SEQUENCE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a sequence",
+ RelationGetRelationName(rel))));
+ break;
+
+ case OBJECT_VIEW:
+ if (rel->rd_rel->relkind != RELKIND_VIEW)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a view",
+ RelationGetRelationName(rel))));
+ break;
+
+ default:
+ elog(ERROR, "unrecognized object type: %d", (int) stmt->relkind);
+ }
+
+ ATController(rel, stmt->cmds, interpretInhOption(stmt->relation->inhOpt));
}
/*
* 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)
{
- ATController(relation_open(relid, AccessExclusiveLock),
- cmds,
- recurse);
+ Relation rel = relation_open(relid, AccessExclusiveLock);
+
+ ATController(rel, cmds, recurse);
}
static void
ATPrepAddColumn(wqueue, rel, recurse, cmd);
pass = AT_PASS_ADD_COL;
break;
+ case AT_AddColumnToView: /* add column via CREATE OR REPLACE VIEW */
+ ATSimplePermissions(rel, true);
+ /* Performs own recursion */
+ ATPrepAddColumn(wqueue, rel, recurse, cmd);
+ pass = AT_PASS_ADD_COL;
+ break;
case AT_ColumnDefault: /* ALTER COLUMN DEFAULT */
/*
break;
case AT_AddConstraint: /* ADD CONSTRAINT */
ATSimplePermissions(rel, false);
-
- /*
- * Currently we recurse only for CHECK constraints, never for
- * foreign-key constraints. UNIQUE/PKEY constraints won't be seen
- * here.
- */
- if (IsA(cmd->def, Constraint))
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
- /* No command-specific prep needed */
+ /* 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_DropConstraint: /* DROP CONSTRAINT */
ATSimplePermissions(rel, false);
- /* Performs own recursion */
- ATPrepDropConstraint(wqueue, rel, recurse, cmd);
- pass = AT_PASS_DROP;
- break;
- case AT_DropConstraintQuietly: /* DROP CONSTRAINT for child */
- ATSimplePermissions(rel, false);
- ATSimpleRecursion(wqueue, rel, cmd, recurse);
- /* No command-specific prep needed */
+ /* 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 */
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_AddInherits:
- case AT_DropInherits:
+ case AT_EnableRule: /* ENABLE/DISABLE RULE variants */
+ case AT_EnableAlwaysRule:
+ case AT_EnableReplicaRule:
+ case AT_DisableRule:
+ case AT_AddInherit: /* INHERIT / NO INHERIT */
+ case AT_DropInherit:
ATSimplePermissions(rel, false);
/* These commands never recurse */
/* No command-specific prep needed */
rel = relation_open(tab->relid, NoLock);
foreach(lcmd, subcmds)
- ATExecCmd(tab, rel, (AlterTableCmd *) lfirst(lcmd));
+ ATExecCmd(wqueue, tab, rel, (AlterTableCmd *) lfirst(lcmd));
/*
* After the ALTER TYPE pass, do cleanup work (this is not done in
* ATExecCmd: dispatch a subcommand to appropriate execution routine
*/
static void
-ATExecCmd(AlteredTableInfo *tab, Relation rel, AlterTableCmd *cmd)
+ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
+ AlterTableCmd *cmd)
{
switch (cmd->subtype)
{
case AT_AddColumn: /* ADD COLUMN */
+ case AT_AddColumnToView: /* add column via CREATE OR REPLACE VIEW */
ATExecAddColumn(tab, rel, (ColumnDef *) cmd->def);
break;
case AT_ColumnDefault: /* ALTER COLUMN DEFAULT */
ATExecAddIndex(tab, rel, (IndexStmt *) cmd->def, true);
break;
case AT_AddConstraint: /* ADD CONSTRAINT */
- ATExecAddConstraint(tab, rel, cmd->def);
+ ATExecAddConstraint(wqueue, tab, rel, cmd->def, false);
+ break;
+ case AT_AddConstraintRecurse: /* ADD CONSTRAINT with recursion */
+ ATExecAddConstraint(wqueue, tab, rel, cmd->def, true);
break;
case AT_DropConstraint: /* DROP CONSTRAINT */
- ATExecDropConstraint(rel, cmd->name, cmd->behavior, false);
+ ATExecDropConstraint(rel, cmd->name, cmd->behavior, false, false);
break;
- case AT_DropConstraintQuietly: /* DROP CONSTRAINT for child */
- ATExecDropConstraint(rel, cmd->name, cmd->behavior, true);
+ case AT_DropConstraintRecurse: /* DROP CONSTRAINT with recursion */
+ ATExecDropConstraint(rel, cmd->name, cmd->behavior, true, false);
break;
case AT_AlterColumnType: /* ALTER COLUMN TYPE */
ATExecAlterColumnType(tab, rel, cmd->name, (TypeName *) cmd->def);
case AT_ResetRelOptions: /* RESET (...) */
ATExecSetRelOptions(rel, (List *) cmd->def, true);
break;
+
case AT_EnableTrig: /* ENABLE TRIGGER name */
- ATExecEnableDisableTrigger(rel, cmd->name, true, false);
+ ATExecEnableDisableTrigger(rel, cmd->name,
+ TRIGGER_FIRES_ON_ORIGIN, false);
+ break;
+ case AT_EnableAlwaysTrig: /* ENABLE ALWAYS TRIGGER name */
+ ATExecEnableDisableTrigger(rel, cmd->name,
+ TRIGGER_FIRES_ALWAYS, false);
+ break;
+ case AT_EnableReplicaTrig: /* ENABLE REPLICA TRIGGER name */
+ ATExecEnableDisableTrigger(rel, cmd->name,
+ TRIGGER_FIRES_ON_REPLICA, false);
break;
case AT_DisableTrig: /* DISABLE TRIGGER name */
- ATExecEnableDisableTrigger(rel, cmd->name, false, false);
+ ATExecEnableDisableTrigger(rel, cmd->name,
+ TRIGGER_DISABLED, false);
break;
case AT_EnableTrigAll: /* ENABLE TRIGGER ALL */
- ATExecEnableDisableTrigger(rel, NULL, true, false);
+ ATExecEnableDisableTrigger(rel, NULL,
+ TRIGGER_FIRES_ON_ORIGIN, false);
break;
case AT_DisableTrigAll: /* DISABLE TRIGGER ALL */
- ATExecEnableDisableTrigger(rel, NULL, false, false);
+ ATExecEnableDisableTrigger(rel, NULL,
+ TRIGGER_DISABLED, false);
break;
case AT_EnableTrigUser: /* ENABLE TRIGGER USER */
- ATExecEnableDisableTrigger(rel, NULL, true, true);
+ ATExecEnableDisableTrigger(rel, NULL,
+ TRIGGER_FIRES_ON_ORIGIN, true);
break;
case AT_DisableTrigUser: /* DISABLE TRIGGER USER */
- ATExecEnableDisableTrigger(rel, NULL, false, true);
+ ATExecEnableDisableTrigger(rel, NULL,
+ TRIGGER_DISABLED, true);
break;
- case AT_DropInherits:
- ATExecDropInherits(rel, cmd->parent);
+
+ case AT_EnableRule: /* ENABLE RULE name */
+ ATExecEnableDisableRule(rel, cmd->name,
+ RULE_FIRES_ON_ORIGIN);
break;
- case AT_AddInherits:
- ATExecAddInherits(rel, cmd->parent);
+ case AT_EnableAlwaysRule: /* ENABLE ALWAYS RULE name */
+ ATExecEnableDisableRule(rel, cmd->name,
+ RULE_FIRES_ALWAYS);
+ break;
+ case AT_EnableReplicaRule: /* ENABLE REPLICA RULE name */
+ ATExecEnableDisableRule(rel, cmd->name,
+ RULE_FIRES_ON_REPLICA);
+ break;
+ case AT_DisableRule: /* DISABLE RULE name */
+ ATExecEnableDisableRule(rel, cmd->name,
+ RULE_DISABLED);
+ break;
+
+ case AT_AddInherit:
+ ATExecAddInherit(rel, (RangeVar *) cmd->def);
+ break;
+ case AT_DropInherit:
+ ATExecDropInherit(rel, (RangeVar *) cmd->def);
break;
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
*/
ATRewriteTable(tab, OIDNewHeap);
- /* Swap the physical files of the old and new heaps. */
- swap_relation_files(tab->relid, OIDNewHeap);
+ /*
+ * Swap the physical files of the old and new heaps. Since we are
+ * generating a new heap, we can use RecentXmin for the table's
+ * new relfrozenxid because we rewrote all the tuples on
+ * ATRewriteTable, so no older Xid remains on the table.
+ */
+ swap_relation_files(tab->relid, OIDNewHeap, RecentXmin);
CommandCounterIncrement();
refrel = heap_open(con->refrelid, RowShareLock);
- validateForeignKeyConstraint(fkconstraint, rel, refrel);
+ validateForeignKeyConstraint(fkconstraint, rel, refrel,
+ con->conid);
heap_close(refrel, NoLock);
}
*/
if (newrel)
find_composite_type_dependencies(oldrel->rd_rel->reltype,
- RelationGetRelationName(oldrel));
+ RelationGetRelationName(oldrel),
+ NULL);
/*
* Generate the constraint and default execution states
if (childrelid == relid)
continue;
childrel = relation_open(childrelid, AccessExclusiveLock);
+ CheckTableNotInUse(childrel, "ALTER TABLE");
ATPrepCmd(wqueue, childrel, cmd, false, true);
relation_close(childrel, NoLock);
}
Relation childrel;
childrel = relation_open(childrelid, AccessExclusiveLock);
+ CheckTableNotInUse(childrel, "ALTER TABLE");
ATPrepCmd(wqueue, childrel, cmd, true, true);
relation_close(childrel, NoLock);
}
/*
* find_composite_type_dependencies
*
- * Check to see if a table's rowtype is being used as a column in some
+ * Check to see if a composite type is being used as a column in some
* other table (possibly nested several levels deep in composite types!).
* Eventually, we'd like to propagate the check or rewrite operation
* into other such tables, but for now, just error out if we find any.
*
+ * Caller should provide either a table name or a type name (not both) to
+ * report in the error message, if any.
+ *
* We assume that functions and views depending on the type are not reasons
* to reject the ALTER. (How safe is this really?)
*/
-static void
-find_composite_type_dependencies(Oid typeOid, const char *origTblName)
+void
+find_composite_type_dependencies(Oid typeOid,
+ const char *origTblName,
+ const char *origTypeName)
{
Relation depRel;
ScanKeyData key[2];
SysScanDesc depScan;
HeapTuple depTup;
+ Oid arrayOid;
/*
* We scan pg_depend to find those things that depend on the rowtype. (We
if (rel->rd_rel->relkind == RELKIND_RELATION)
{
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot alter table \"%s\" because column \"%s\".\"%s\" uses its rowtype",
- origTblName,
- RelationGetRelationName(rel),
- NameStr(att->attname))));
+ if (origTblName)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot alter table \"%s\" because column \"%s\".\"%s\" uses its rowtype",
+ origTblName,
+ RelationGetRelationName(rel),
+ NameStr(att->attname))));
+ else
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot alter type \"%s\" because column \"%s\".\"%s\" uses it",
+ origTypeName,
+ RelationGetRelationName(rel),
+ NameStr(att->attname))));
}
else if (OidIsValid(rel->rd_rel->reltype))
{
* recursively check for indirect dependencies via its rowtype.
*/
find_composite_type_dependencies(rel->rd_rel->reltype,
- origTblName);
+ origTblName, origTypeName);
}
relation_close(rel, AccessShareLock);
systable_endscan(depScan);
relation_close(depRel, AccessShareLock);
+
+ /*
+ * If there's an array type for the rowtype, must check for uses of it,
+ * too.
+ */
+ arrayOid = get_array_type(typeOid);
+ if (OidIsValid(arrayOid))
+ find_composite_type_dependencies(arrayOid, origTblName, origTypeName);
}
*
* Adds an additional attribute to a relation making the assumption that
* CHECK, NOT NULL, and FOREIGN KEY constraints will be removed from the
- * AT_AddColumn AlterTableCmd by analyze.c and added as independent
+ * AT_AddColumn AlterTableCmd by parse_utilcmd.c and added as independent
* AlterTableCmd's.
*/
static void
Relation pgclass,
attrdesc;
HeapTuple reltup;
- HeapTuple attributeTuple;
- Form_pg_attribute attribute;
- FormData_pg_attribute attributeD;
+ FormData_pg_attribute attribute;
int i;
int minattnum,
maxatts;
+ char relkind;
HeapTuple typeTuple;
Oid typeOid;
+ int32 typmod;
Form_pg_type tform;
Expr *defval;
if (HeapTupleIsValid(tuple))
{
Form_pg_attribute childatt = (Form_pg_attribute) GETSTRUCT(tuple);
+ Oid ctypeId;
+ int32 ctypmod;
/* Okay if child matches by type */
- if (typenameTypeId(NULL, colDef->typename) != childatt->atttypid ||
- colDef->typename->typmod != childatt->atttypmod)
+ ctypeId = typenameTypeId(NULL, colDef->typename, &ctypmod);
+ if (ctypeId != childatt->atttypid ||
+ ctypmod != childatt->atttypmod)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("child table \"%s\" has different type for column \"%s\"",
colDef->colname, RelationGetRelationName(rel))));
minattnum = ((Form_pg_class) GETSTRUCT(reltup))->relnatts;
+ relkind = ((Form_pg_class) GETSTRUCT(reltup))->relkind;
maxatts = minattnum + 1;
if (maxatts > MaxHeapAttributeNumber)
ereport(ERROR,
MaxHeapAttributeNumber)));
i = minattnum + 1;
- typeTuple = typenameType(NULL, colDef->typename);
+ typeTuple = typenameType(NULL, colDef->typename, &typmod);
tform = (Form_pg_type) GETSTRUCT(typeTuple);
typeOid = HeapTupleGetOid(typeTuple);
/* make sure datatype is legal for a column */
CheckAttributeType(colDef->colname, typeOid);
- attributeTuple = heap_addheader(Natts_pg_attribute,
- false,
- ATTRIBUTE_TUPLE_SIZE,
- (void *) &attributeD);
-
- attribute = (Form_pg_attribute) GETSTRUCT(attributeTuple);
-
- attribute->attrelid = myrelid;
- namestrcpy(&(attribute->attname), colDef->colname);
- attribute->atttypid = typeOid;
- attribute->attstattarget = -1;
- attribute->attlen = tform->typlen;
- attribute->attcacheoff = -1;
- attribute->atttypmod = colDef->typename->typmod;
- attribute->attnum = i;
- attribute->attbyval = tform->typbyval;
- attribute->attndims = list_length(colDef->typename->arrayBounds);
- attribute->attstorage = tform->typstorage;
- attribute->attalign = tform->typalign;
- attribute->attnotnull = colDef->is_not_null;
- attribute->atthasdef = false;
- attribute->attisdropped = false;
- attribute->attislocal = colDef->is_local;
- attribute->attinhcount = colDef->inhcount;
+ attribute.attrelid = myrelid;
+ namestrcpy(&(attribute.attname), colDef->colname);
+ attribute.atttypid = typeOid;
+ attribute.attstattarget = -1;
+ attribute.attlen = tform->typlen;
+ attribute.attcacheoff = -1;
+ attribute.atttypmod = typmod;
+ attribute.attnum = i;
+ attribute.attbyval = tform->typbyval;
+ attribute.attndims = list_length(colDef->typename->arrayBounds);
+ attribute.attstorage = tform->typstorage;
+ attribute.attalign = tform->typalign;
+ attribute.attnotnull = colDef->is_not_null;
+ attribute.atthasdef = false;
+ attribute.attisdropped = false;
+ attribute.attislocal = colDef->is_local;
+ attribute.attinhcount = colDef->inhcount;
ReleaseSysCache(typeTuple);
- simple_heap_insert(attrdesc, attributeTuple);
-
- /* Update indexes on pg_attribute */
- CatalogUpdateIndexes(attrdesc, attributeTuple);
+ InsertPgAttributeTuple(attrdesc, &attribute, NULL);
heap_close(attrdesc, RowExclusiveLock);
RawColumnDefault *rawEnt;
rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));
- rawEnt->attnum = attribute->attnum;
+ rawEnt->attnum = attribute.attnum;
rawEnt->raw_default = copyObject(colDef->raw_default);
/*
* This function is intended for CREATE TABLE, so it processes a
* _list_ of defaults, but we just do one.
*/
- AddRelationRawConstraints(rel, list_make1(rawEnt), NIL);
+ AddRelationNewConstraints(rel, list_make1(rawEnt), NIL, false, true);
/* Make the additional catalog changes visible */
CommandCounterIncrement();
* the constraints more directly.)
*
* Note: we use build_column_default, and not just the cooked default
- * returned by AddRelationRawConstraints, so that the right thing happens
+ * returned by AddRelationNewConstraints, so that the right thing happens
* when a datatype's default applies.
+ *
+ * We skip this logic completely for views.
*/
- defval = (Expr *) build_column_default(rel, attribute->attnum);
+ if (relkind != RELKIND_VIEW) {
+ defval = (Expr *) build_column_default(rel, attribute.attnum);
- if (!defval && GetDomainConstraints(typeOid) != NIL)
- {
- Oid basetype = getBaseType(typeOid);
-
- defval = (Expr *) makeNullConst(basetype);
- defval = (Expr *) coerce_to_target_type(NULL,
- (Node *) defval,
- basetype,
- typeOid,
- colDef->typename->typmod,
- COERCION_ASSIGNMENT,
- COERCE_IMPLICIT_CAST);
- if (defval == NULL) /* should not happen */
- elog(ERROR, "failed to coerce base type to domain");
- }
+ if (!defval && GetDomainConstraints(typeOid) != NIL)
+ {
+ Oid baseTypeId;
+ int32 baseTypeMod;
+
+ baseTypeMod = typmod;
+ baseTypeId = getBaseTypeAndTypmod(typeOid, &baseTypeMod);
+ defval = (Expr *) makeNullConst(baseTypeId, baseTypeMod);
+ defval = (Expr *) coerce_to_target_type(NULL,
+ (Node *) defval,
+ baseTypeId,
+ typeOid,
+ typmod,
+ COERCION_ASSIGNMENT,
+ COERCE_IMPLICIT_CAST,
+ -1);
+ if (defval == NULL) /* should not happen */
+ elog(ERROR, "failed to coerce base type to domain");
+ }
- if (defval)
- {
- NewColumnValue *newval;
+ if (defval)
+ {
+ NewColumnValue *newval;
+
+ newval = (NewColumnValue *) palloc0(sizeof(NewColumnValue));
+ newval->attnum = attribute.attnum;
+ newval->expr = defval;
- newval = (NewColumnValue *) palloc0(sizeof(NewColumnValue));
- newval->attnum = attribute->attnum;
- newval->expr = defval;
+ tab->newvals = lappend(tab->newvals, newval);
+ }
- tab->newvals = lappend(tab->newvals, newval);
+ /*
+ * If the new column is NOT NULL, tell Phase 3 it needs to test that.
+ */
+ tab->new_notnull |= colDef->is_not_null;
}
/*
* Add needed dependency entries for the new column.
*/
- add_column_datatype_dependency(myrelid, i, attribute->atttypid);
+ add_column_datatype_dependency(myrelid, i, attribute.atttypid);
}
/*
* This function is intended for CREATE TABLE, so it processes a
* _list_ of defaults, but we just do one.
*/
- AddRelationRawConstraints(rel, list_make1(rawEnt), NIL);
+ AddRelationNewConstraints(rel, list_make1(rawEnt), NIL, false, true);
}
}
errmsg("statistics target %d is too low",
newtarget)));
}
- else if (newtarget > 1000)
+ else if (newtarget > 10000)
{
- newtarget = 1000;
+ newtarget = 10000;
ereport(WARNING,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("lowering statistics target to %d",
Form_pg_attribute childatt;
childrel = heap_open(childrelid, AccessExclusiveLock);
+ CheckTableNotInUse(childrel, "ALTER TABLE");
tuple = SearchSysCacheCopyAttName(childrelid, colName);
if (!HeapTupleIsValid(tuple)) /* shouldn't happen */
{
/*
* If we were told to drop ONLY in this table (no recursion),
- * we need to mark the inheritors' attribute as locally
+ * we need to mark the inheritors' attributes as locally
* defined rather than inherited.
*/
childatt->attinhcount--;
/*
* ALTER TABLE ADD INDEX
*
- * There is no such command in the grammar, but the parser converts UNIQUE
- * and PRIMARY KEY constraints into AT_AddIndex subcommands. This lets us
- * schedule creation of the index at the appropriate time during ALTER.
+ * There is no such command in the grammar, but parse_utilcmd.c converts
+ * UNIQUE and PRIMARY KEY constraints into AT_AddIndex subcommands. This lets
+ * us schedule creation of the index at the appropriate time during ALTER.
*/
static void
ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
/* suppress notices when rebuilding existing index */
quiet = is_rebuild;
+ /* The IndexStmt has already been through transformIndexStmt */
+
DefineIndex(stmt->relation, /* relation */
stmt->idxname, /* index name */
InvalidOid, /* no predefined OID */
stmt->tableSpace,
stmt->indexParams, /* parameters */
(Expr *) stmt->whereClause,
- stmt->rangetable,
stmt->options,
stmt->unique,
stmt->primary,
* ALTER TABLE ADD CONSTRAINT
*/
static void
-ATExecAddConstraint(AlteredTableInfo *tab, Relation rel, Node *newConstraint)
+ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
+ Node *newConstraint, bool recurse)
{
switch (nodeTag(newConstraint))
{
/*
* Currently, we only expect to see CONSTR_CHECK nodes
* arriving here (see the preprocessing done in
- * parser/analyze.c). Use a switch anyway to make it easier
- * to add more code later.
+ * parse_utilcmd.c). Use a switch anyway to make it easier to
+ * add more code later.
*/
switch (constr->contype)
{
case CONSTR_CHECK:
- {
- List *newcons;
- ListCell *lcon;
-
- /*
- * Call AddRelationRawConstraints to do the work.
- * It returns a list of cooked constraints.
- */
- newcons = AddRelationRawConstraints(rel, NIL,
- list_make1(constr));
- /* Add each constraint to Phase 3's queue */
- foreach(lcon, newcons)
- {
- CookedConstraint *ccon = (CookedConstraint *) lfirst(lcon);
- NewConstraint *newcon;
-
- newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
- newcon->name = ccon->name;
- newcon->contype = ccon->contype;
- /* ExecQual wants implicit-AND format */
- newcon->qual = (Node *)
- make_ands_implicit((Expr *) ccon->expr);
-
- tab->constraints = lappend(tab->constraints,
- newcon);
- }
- break;
- }
+ ATAddCheckConstraint(wqueue, tab, rel,
+ constr, recurse, false);
+ break;
default:
elog(ERROR, "unrecognized constraint type: %d",
(int) constr->contype);
FkConstraint *fkconstraint = (FkConstraint *) newConstraint;
/*
+ * Note that we currently never recurse for FK constraints,
+ * so the "recurse" flag is silently ignored.
+ *
* Assign or validate constraint name
*/
if (fkconstraint->constr_name)
}
}
+/*
+ * Add a check constraint to a single table and its children
+ *
+ * Subroutine for ATExecAddConstraint.
+ *
+ * We must recurse to child tables during execution, rather than using
+ * ALTER TABLE's normal prep-time recursion. The reason is that all the
+ * constraints *must* be given the same name, else they won't be seen as
+ * related later. If the user didn't explicitly specify a name, then
+ * AddRelationNewConstraints would normally assign different names to the
+ * child constraints. To fix that, we must capture the name assigned at
+ * the parent table and pass that down.
+ */
+static void
+ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
+ Constraint *constr, bool recurse, bool recursing)
+{
+ List *newcons;
+ ListCell *lcon;
+ List *children;
+ ListCell *child;
+
+ /* At top level, permission check was done in ATPrepCmd, else do it */
+ if (recursing)
+ ATSimplePermissions(rel, false);
+
+ /*
+ * Call AddRelationNewConstraints to do the work, making sure it works on
+ * a copy of the Constraint so transformExpr can't modify the original.
+ * It returns a list of cooked constraints.
+ *
+ * If the constraint ends up getting merged with a pre-existing one, it's
+ * omitted from the returned list, which is what we want: we do not need
+ * to do any validation work. That can only happen at child tables,
+ * though, since we disallow merging at the top level.
+ */
+ newcons = AddRelationNewConstraints(rel, NIL,
+ list_make1(copyObject(constr)),
+ recursing, !recursing);
+
+ /* Add each constraint to Phase 3's queue */
+ foreach(lcon, newcons)
+ {
+ CookedConstraint *ccon = (CookedConstraint *) lfirst(lcon);
+ NewConstraint *newcon;
+
+ newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
+ newcon->name = ccon->name;
+ newcon->contype = ccon->contype;
+ /* ExecQual wants implicit-AND format */
+ newcon->qual = (Node *) make_ands_implicit((Expr *) ccon->expr);
+
+ tab->constraints = lappend(tab->constraints, newcon);
+
+ /* Save the actually assigned name if it was defaulted */
+ if (constr->name == NULL)
+ constr->name = ccon->name;
+ }
+
+ /* At this point we must have a locked-down name to use */
+ Assert(constr->name != NULL);
+
+ /* Advance command counter in case same table is visited multiple times */
+ CommandCounterIncrement();
+
+ /*
+ * Propagate to children as appropriate. Unlike most other ALTER
+ * routines, we have to do this one level of recursion at a time; we can't
+ * use find_all_inheritors to do it in one pass.
+ */
+ children = find_inheritance_children(RelationGetRelid(rel));
+
+ /*
+ * If we are told not to recurse, there had better not be any child
+ * tables; else the addition would put them out of step.
+ */
+ if (children && !recurse)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+ errmsg("constraint must be added to child tables too")));
+
+ foreach(child, children)
+ {
+ Oid childrelid = lfirst_oid(child);
+ Relation childrel;
+ AlteredTableInfo *childtab;
+
+ childrel = heap_open(childrelid, AccessExclusiveLock);
+ CheckTableNotInUse(childrel, "ALTER TABLE");
+
+ /* Find or create work queue entry for this table */
+ childtab = ATGetQueueEntry(wqueue, childrel);
+
+ /* Recurse to child */
+ ATAddCheckConstraint(wqueue, childtab, childrel,
+ constr, recurse, true);
+
+ heap_close(childrel, NoLock);
+ }
+}
+
/*
* Add a foreign-key constraint to a single table
*
Oid pktypoid[INDEX_MAX_KEYS];
Oid fktypoid[INDEX_MAX_KEYS];
Oid opclasses[INDEX_MAX_KEYS];
+ Oid pfeqoperators[INDEX_MAX_KEYS];
+ Oid ppeqoperators[INDEX_MAX_KEYS];
+ Oid ffeqoperators[INDEX_MAX_KEYS];
int i;
int numfks,
numpks;
MemSet(pktypoid, 0, sizeof(pktypoid));
MemSet(fktypoid, 0, sizeof(fktypoid));
MemSet(opclasses, 0, sizeof(opclasses));
+ MemSet(pfeqoperators, 0, sizeof(pfeqoperators));
+ MemSet(ppeqoperators, 0, sizeof(ppeqoperators));
+ MemSet(ffeqoperators, 0, sizeof(ffeqoperators));
numfks = transformColumnNameList(RelationGetRelid(rel),
fkconstraint->fk_attrs,
opclasses);
}
- /* Be sure referencing and referenced column types are comparable */
+ /*
+ * Look up the equality operators to use in the constraint.
+ *
+ * Note that we have to be careful about the difference between the actual
+ * PK column type and the opclass' declared input type, which might be
+ * only binary-compatible with it. The declared opcintype is the right
+ * thing to probe pg_amop with.
+ */
if (numfks != numpks)
ereport(ERROR,
(errcode(ERRCODE_INVALID_FOREIGN_KEY),
for (i = 0; i < numpks; i++)
{
+ Oid pktype = pktypoid[i];
+ Oid fktype = fktypoid[i];
+ Oid fktyped;
+ HeapTuple cla_ht;
+ Form_pg_opclass cla_tup;
+ Oid amid;
+ Oid opfamily;
+ Oid opcintype;
+ Oid pfeqop;
+ Oid ppeqop;
+ Oid ffeqop;
+ int16 eqstrategy;
+
+ /* We need several fields out of the pg_opclass entry */
+ cla_ht = SearchSysCache(CLAOID,
+ ObjectIdGetDatum(opclasses[i]),
+ 0, 0, 0);
+ if (!HeapTupleIsValid(cla_ht))
+ elog(ERROR, "cache lookup failed for opclass %u", opclasses[i]);
+ cla_tup = (Form_pg_opclass) GETSTRUCT(cla_ht);
+ amid = cla_tup->opcmethod;
+ opfamily = cla_tup->opcfamily;
+ opcintype = cla_tup->opcintype;
+ ReleaseSysCache(cla_ht);
+
/*
- * pktypoid[i] is the primary key table's i'th key's type fktypoid[i]
- * is the foreign key table's i'th key's type
- *
- * Note that we look for an operator with the PK type on the left;
- * when the types are different this is critical because the PK index
- * will need operators with the indexkey on the left. (Ordinarily both
- * commutator operators will exist if either does, but we won't get
- * the right answer from the test below on opclass membership unless
- * we select the proper operator.)
+ * Check it's a btree; currently this can never fail since no other
+ * index AMs support unique indexes. If we ever did have other types
+ * of unique indexes, we'd need a way to determine which operator
+ * strategy number is equality. (Is it reasonable to insist that
+ * every such index AM use btree's number for equality?)
+ */
+ if (amid != BTREE_AM_OID)
+ elog(ERROR, "only b-tree indexes are supported for foreign keys");
+ eqstrategy = BTEqualStrategyNumber;
+
+ /*
+ * There had better be a primary equality operator for the index.
+ * We'll use it for PK = PK comparisons.
+ */
+ ppeqop = get_opfamily_member(opfamily, opcintype, opcintype,
+ eqstrategy);
+
+ if (!OidIsValid(ppeqop))
+ elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
+ eqstrategy, opcintype, opcintype, opfamily);
+
+ /*
+ * Are there equality operators that take exactly the FK type? Assume
+ * we should look through any domain here.
*/
- Operator o = oper(NULL, list_make1(makeString("=")),
- pktypoid[i], fktypoid[i],
- true, -1);
+ fktyped = getBaseType(fktype);
- if (o == NULL)
+ pfeqop = get_opfamily_member(opfamily, opcintype, fktyped,
+ eqstrategy);
+ if (OidIsValid(pfeqop))
+ ffeqop = get_opfamily_member(opfamily, fktyped, fktyped,
+ eqstrategy);
+ else
+ ffeqop = InvalidOid; /* keep compiler quiet */
+
+ if (!(OidIsValid(pfeqop) && OidIsValid(ffeqop)))
+ {
+ /*
+ * Otherwise, look for an implicit cast from the FK type to the
+ * opcintype, and if found, use the primary equality operator.
+ * This is a bit tricky because opcintype might be a polymorphic
+ * type such as ANYARRAY or ANYENUM; so what we have to test is
+ * whether the two actual column types can be concurrently cast to
+ * that type. (Otherwise, we'd fail to reject combinations such
+ * as int[] and point[].)
+ */
+ Oid input_typeids[2];
+ Oid target_typeids[2];
+
+ input_typeids[0] = pktype;
+ input_typeids[1] = fktype;
+ target_typeids[0] = opcintype;
+ target_typeids[1] = opcintype;
+ if (can_coerce_type(2, input_typeids, target_typeids,
+ COERCION_IMPLICIT))
+ pfeqop = ffeqop = ppeqop;
+ }
+
+ if (!(OidIsValid(pfeqop) && OidIsValid(ffeqop)))
ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_FUNCTION),
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("foreign key constraint \"%s\" "
"cannot be implemented",
fkconstraint->constr_name),
"are of incompatible types: %s and %s.",
strVal(list_nth(fkconstraint->fk_attrs, i)),
strVal(list_nth(fkconstraint->pk_attrs, i)),
- format_type_be(fktypoid[i]),
- format_type_be(pktypoid[i]))));
-
- /*
- * Check that the found operator is compatible with the PK index, and
- * generate a warning if not, since otherwise costly seqscans will be
- * incurred to check FK validity.
- */
- if (!op_in_opclass(oprid(o), opclasses[i]))
- ereport(WARNING,
- (errmsg("foreign key constraint \"%s\" "
- "will require costly sequential scans",
- fkconstraint->constr_name),
- errdetail("Key columns \"%s\" and \"%s\" "
- "are of different types: %s and %s.",
- strVal(list_nth(fkconstraint->fk_attrs, i)),
- strVal(list_nth(fkconstraint->pk_attrs, i)),
- format_type_be(fktypoid[i]),
- format_type_be(pktypoid[i]))));
-
- ReleaseSysCache(o);
- }
-
- /*
- * Tell Phase 3 to check that the constraint is satisfied by existing rows
- * (we can skip this during table creation).
- */
- if (!fkconstraint->skip_validation)
- {
- NewConstraint *newcon;
-
- newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
- newcon->name = fkconstraint->constr_name;
- newcon->contype = CONSTR_FOREIGN;
- newcon->refrelid = RelationGetRelid(pkrel);
- newcon->qual = (Node *) fkconstraint;
+ format_type_be(fktype),
+ format_type_be(pktype))));
- tab->constraints = lappend(tab->constraints, newcon);
+ pfeqoperators[i] = pfeqop;
+ ppeqoperators[i] = ppeqop;
+ ffeqoperators[i] = ffeqop;
}
/*
* constraint */
RelationGetRelid(pkrel),
pkattnum,
+ pfeqoperators,
+ ppeqoperators,
+ ffeqoperators,
numpks,
fkconstraint->fk_upd_action,
fkconstraint->fk_del_action,
indexOid,
NULL, /* no check constraint */
NULL,
- NULL);
+ NULL,
+ true, /* islocal */
+ 0); /* inhcount */
/*
* Create the triggers that will enforce the constraint.
*/
createForeignKeyTriggers(rel, fkconstraint, constrOid);
+ /*
+ * Tell Phase 3 to check that the constraint is satisfied by existing rows
+ * (we can skip this during table creation).
+ */
+ if (!fkconstraint->skip_validation)
+ {
+ NewConstraint *newcon;
+
+ newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));
+ newcon->name = fkconstraint->constr_name;
+ newcon->contype = CONSTR_FOREIGN;
+ newcon->refrelid = RelationGetRelid(pkrel);
+ newcon->conid = constrOid;
+ newcon->qual = (Node *) fkconstraint;
+
+ tab->constraints = lappend(tab->constraints, newcon);
+ }
+
/*
* Close pk table, but keep lock until we've committed.
*/
static void
validateForeignKeyConstraint(FkConstraint *fkconstraint,
Relation rel,
- Relation pkrel)
+ Relation pkrel,
+ Oid constraintOid)
{
HeapScanDesc scan;
HeapTuple tuple;
Trigger trig;
- ListCell *list;
- int count;
-
- /*
- * See if we can do it with a single LEFT JOIN query. A FALSE result
- * indicates we must proceed with the fire-the-trigger method.
- */
- if (RI_Initial_Check(fkconstraint, rel, pkrel))
- return;
/*
- * Scan through each tuple, calling RI_FKey_check_ins (insert trigger) as
- * if that tuple had just been inserted. If any of those fail, it should
- * ereport(ERROR) and that's that.
+ * Build a trigger call structure; we'll need it either way.
*/
MemSet(&trig, 0, sizeof(trig));
trig.tgoid = InvalidOid;
trig.tgname = fkconstraint->constr_name;
- trig.tgenabled = TRUE;
+ trig.tgenabled = TRIGGER_FIRES_ON_ORIGIN;
trig.tgisconstraint = TRUE;
trig.tgconstrrelid = RelationGetRelid(pkrel);
+ trig.tgconstraint = constraintOid;
trig.tgdeferrable = FALSE;
trig.tginitdeferred = FALSE;
+ /* we needn't fill in tgargs */
- trig.tgargs = (char **) palloc(sizeof(char *) *
- (4 + list_length(fkconstraint->fk_attrs)
- + list_length(fkconstraint->pk_attrs)));
-
- trig.tgargs[0] = trig.tgname;
- trig.tgargs[1] = RelationGetRelationName(rel);
- trig.tgargs[2] = RelationGetRelationName(pkrel);
- trig.tgargs[3] = fkMatchTypeToString(fkconstraint->fk_matchtype);
- count = 4;
- foreach(list, fkconstraint->fk_attrs)
- {
- char *fk_at = strVal(lfirst(list));
-
- trig.tgargs[count] = fk_at;
- count += 2;
- }
- count = 5;
- foreach(list, fkconstraint->pk_attrs)
- {
- char *pk_at = strVal(lfirst(list));
-
- trig.tgargs[count] = pk_at;
- count += 2;
- }
- trig.tgnargs = count - 1;
+ /*
+ * See if we can do it with a single LEFT JOIN query. A FALSE result
+ * indicates we must proceed with the fire-the-trigger method.
+ */
+ if (RI_Initial_Check(&trig, rel, pkrel))
+ return;
+ /*
+ * Scan through each tuple, calling RI_FKey_check_ins (insert trigger) as
+ * if that tuple had just been inserted. If any of those fail, it should
+ * ereport(ERROR) and that's that.
+ */
scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
}
heap_endscan(scan);
-
- pfree(trig.tgargs);
}
static void
CreateFKCheckTrigger(RangeVar *myRel, FkConstraint *fkconstraint,
- ObjectAddress *constrobj, ObjectAddress *trigobj,
- bool on_insert)
+ Oid constraintOid, bool on_insert)
{
CreateTrigStmt *fk_trigger;
- ListCell *fk_attr;
- ListCell *pk_attr;
fk_trigger = makeNode(CreateTrigStmt);
fk_trigger->trigname = fkconstraint->constr_name;
fk_trigger->deferrable = fkconstraint->deferrable;
fk_trigger->initdeferred = fkconstraint->initdeferred;
fk_trigger->constrrel = fkconstraint->pktable;
-
fk_trigger->args = NIL;
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkconstraint->constr_name));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(myRel->relname));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkconstraint->pktable->relname));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkMatchTypeToString(fkconstraint->fk_matchtype)));
- if (list_length(fkconstraint->fk_attrs) != list_length(fkconstraint->pk_attrs))
- ereport(ERROR,
- (errcode(ERRCODE_INVALID_FOREIGN_KEY),
- errmsg("number of referencing and referenced columns for foreign key disagree")));
-
- forboth(fk_attr, fkconstraint->fk_attrs,
- pk_attr, fkconstraint->pk_attrs)
- {
- fk_trigger->args = lappend(fk_trigger->args, lfirst(fk_attr));
- fk_trigger->args = lappend(fk_trigger->args, lfirst(pk_attr));
- }
- trigobj->objectId = CreateTrigger(fk_trigger, true);
-
- /* Register dependency from trigger to constraint */
- recordDependencyOn(trigobj, constrobj, DEPENDENCY_INTERNAL);
+ (void) CreateTrigger(fk_trigger, constraintOid);
/* Make changes-so-far visible */
CommandCounterIncrement();
*/
static void
createForeignKeyTriggers(Relation rel, FkConstraint *fkconstraint,
- Oid constrOid)
+ Oid constraintOid)
{
RangeVar *myRel;
CreateTrigStmt *fk_trigger;
- ListCell *fk_attr;
- ListCell *pk_attr;
- ObjectAddress trigobj,
- constrobj;
/*
* Reconstruct a RangeVar for my relation (not passed in, unfortunately).
*/
myRel = makeRangeVar(get_namespace_name(RelationGetNamespace(rel)),
- pstrdup(RelationGetRelationName(rel)));
-
- /*
- * Preset objectAddress fields
- */
- constrobj.classId = ConstraintRelationId;
- constrobj.objectId = constrOid;
- constrobj.objectSubId = 0;
- trigobj.classId = TriggerRelationId;
- trigobj.objectSubId = 0;
+ pstrdup(RelationGetRelationName(rel)),
+ -1);
/* Make changes-so-far visible */
CommandCounterIncrement();
* Build and execute a CREATE CONSTRAINT TRIGGER statement for the CHECK
* action for both INSERTs and UPDATEs on the referencing table.
*/
- CreateFKCheckTrigger(myRel, fkconstraint, &constrobj, &trigobj, true);
- CreateFKCheckTrigger(myRel, fkconstraint, &constrobj, &trigobj, false);
+ CreateFKCheckTrigger(myRel, fkconstraint, constraintOid, true);
+ CreateFKCheckTrigger(myRel, fkconstraint, constraintOid, false);
/*
* Build and execute a CREATE CONSTRAINT TRIGGER statement for the ON
(int) fkconstraint->fk_del_action);
break;
}
-
fk_trigger->args = NIL;
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkconstraint->constr_name));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(myRel->relname));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkconstraint->pktable->relname));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkMatchTypeToString(fkconstraint->fk_matchtype)));
- forboth(fk_attr, fkconstraint->fk_attrs,
- pk_attr, fkconstraint->pk_attrs)
- {
- fk_trigger->args = lappend(fk_trigger->args, lfirst(fk_attr));
- fk_trigger->args = lappend(fk_trigger->args, lfirst(pk_attr));
- }
-
- trigobj.objectId = CreateTrigger(fk_trigger, true);
- /* Register dependency from trigger to constraint */
- recordDependencyOn(&trigobj, &constrobj, DEPENDENCY_INTERNAL);
+ (void) CreateTrigger(fk_trigger, constraintOid);
/* Make changes-so-far visible */
CommandCounterIncrement();
(int) fkconstraint->fk_upd_action);
break;
}
-
fk_trigger->args = NIL;
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkconstraint->constr_name));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(myRel->relname));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkconstraint->pktable->relname));
- fk_trigger->args = lappend(fk_trigger->args,
- makeString(fkMatchTypeToString(fkconstraint->fk_matchtype)));
- forboth(fk_attr, fkconstraint->fk_attrs,
- pk_attr, fkconstraint->pk_attrs)
- {
- fk_trigger->args = lappend(fk_trigger->args, lfirst(fk_attr));
- fk_trigger->args = lappend(fk_trigger->args, lfirst(pk_attr));
- }
-
- trigobj.objectId = CreateTrigger(fk_trigger, true);
- /* Register dependency from trigger to constraint */
- recordDependencyOn(&trigobj, &constrobj, DEPENDENCY_INTERNAL);
-}
-
-/*
- * fkMatchTypeToString -
- * convert FKCONSTR_MATCH_xxx code to string to use in trigger args
- */
-static char *
-fkMatchTypeToString(char match_type)
-{
- switch (match_type)
- {
- case FKCONSTR_MATCH_FULL:
- return pstrdup("FULL");
- case FKCONSTR_MATCH_PARTIAL:
- return pstrdup("PARTIAL");
- case FKCONSTR_MATCH_UNSPECIFIED:
- return pstrdup("UNSPECIFIED");
- default:
- elog(ERROR, "unrecognized match type: %d",
- (int) match_type);
- }
- return NULL; /* can't get here */
+ (void) CreateTrigger(fk_trigger, constraintOid);
}
/*
* ALTER TABLE DROP CONSTRAINT
+ *
+ * Like DROP COLUMN, we can't use the normal ALTER TABLE recursion mechanism.
*/
static void
-ATPrepDropConstraint(List **wqueue, Relation rel,
- bool recurse, AlterTableCmd *cmd)
+ATExecDropConstraint(Relation rel, const char *constrName,
+ DropBehavior behavior,
+ bool recurse, bool recursing)
{
+ List *children;
+ ListCell *child;
+ Relation conrel;
+ Form_pg_constraint con;
+ SysScanDesc scan;
+ ScanKeyData key;
+ HeapTuple tuple;
+ bool found = false;
+ bool is_check_constraint = false;
+
+ /* At top level, permission check was done in ATPrepCmd, else do it */
+ if (recursing)
+ ATSimplePermissions(rel, false);
+
+ conrel = heap_open(ConstraintRelationId, RowExclusiveLock);
+
/*
- * We don't want errors or noise from child tables, so we have to pass
- * down a modified command.
+ * Find and drop the target constraint
*/
- if (recurse)
- {
- AlterTableCmd *childCmd = copyObject(cmd);
+ ScanKeyInit(&key,
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(rel)));
+ scan = systable_beginscan(conrel, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, &key);
- childCmd->subtype = AT_DropConstraintQuietly;
- ATSimpleRecursion(wqueue, rel, childCmd, recurse);
- }
-}
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ {
+ ObjectAddress conobj;
-static void
-ATExecDropConstraint(Relation rel, const char *constrName,
- DropBehavior behavior, bool quiet)
-{
- int deleted;
+ con = (Form_pg_constraint) GETSTRUCT(tuple);
- deleted = RemoveRelConstraints(rel, constrName, behavior);
+ if (strcmp(NameStr(con->conname), constrName) != 0)
+ continue;
- if (!quiet)
- {
- /* If zero constraints deleted, complain */
- if (deleted == 0)
+ /* Don't drop inherited constraints */
+ if (con->coninhcount > 0 && !recursing)
ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_OBJECT),
- errmsg("constraint \"%s\" does not exist",
- constrName)));
- /* Otherwise if more than one constraint deleted, notify */
- else if (deleted > 1)
- ereport(NOTICE,
- (errmsg("multiple constraints named \"%s\" were dropped",
- constrName)));
+ (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+ errmsg("cannot drop inherited constraint \"%s\" of relation \"%s\"",
+ constrName, RelationGetRelationName(rel))));
+
+ /* Right now only CHECK constraints can be inherited */
+ if (con->contype == CONSTRAINT_CHECK)
+ is_check_constraint = true;
+
+ /*
+ * Perform the actual constraint deletion
+ */
+ conobj.classId = ConstraintRelationId;
+ conobj.objectId = HeapTupleGetOid(tuple);
+ conobj.objectSubId = 0;
+
+ performDeletion(&conobj, behavior);
+
+ found = true;
}
-}
-/*
- * ALTER COLUMN TYPE
- */
-static void
-ATPrepAlterColumnType(List **wqueue,
- AlteredTableInfo *tab, Relation rel,
+ systable_endscan(scan);
+
+ if (!found)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("constraint \"%s\" of relation \"%s\" does not exist",
+ constrName, RelationGetRelationName(rel))));
+
+ /*
+ * Propagate to children as appropriate. Unlike most other ALTER
+ * routines, we have to do this one level of recursion at a time; we can't
+ * use find_all_inheritors to do it in one pass.
+ */
+ if (is_check_constraint)
+ children = find_inheritance_children(RelationGetRelid(rel));
+ else
+ children = NIL;
+
+ foreach(child, children)
+ {
+ Oid childrelid = lfirst_oid(child);
+ Relation childrel;
+
+ childrel = heap_open(childrelid, AccessExclusiveLock);
+ CheckTableNotInUse(childrel, "ALTER TABLE");
+
+ ScanKeyInit(&key,
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(childrelid));
+ scan = systable_beginscan(conrel, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, &key);
+
+ found = false;
+
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ {
+ HeapTuple copy_tuple;
+
+ con = (Form_pg_constraint) GETSTRUCT(tuple);
+
+ /* Right now only CHECK constraints can be inherited */
+ if (con->contype != CONSTRAINT_CHECK)
+ continue;
+
+ if (strcmp(NameStr(con->conname), constrName) != 0)
+ continue;
+
+ found = true;
+
+ if (con->coninhcount <= 0) /* shouldn't happen */
+ elog(ERROR, "relation %u has non-inherited constraint \"%s\"",
+ childrelid, constrName);
+
+ copy_tuple = heap_copytuple(tuple);
+ con = (Form_pg_constraint) GETSTRUCT(copy_tuple);
+
+ if (recurse)
+ {
+ /*
+ * If the child constraint has other definition sources,
+ * just decrement its inheritance count; if not, recurse
+ * to delete it.
+ */
+ if (con->coninhcount == 1 && !con->conislocal)
+ {
+ /* Time to delete this child constraint, too */
+ ATExecDropConstraint(childrel, constrName, behavior,
+ true, true);
+ }
+ else
+ {
+ /* Child constraint must survive my deletion */
+ con->coninhcount--;
+ simple_heap_update(conrel, ©_tuple->t_self, copy_tuple);
+ CatalogUpdateIndexes(conrel, copy_tuple);
+
+ /* Make update visible */
+ CommandCounterIncrement();
+ }
+ }
+ else
+ {
+ /*
+ * If we were told to drop ONLY in this table (no
+ * recursion), we need to mark the inheritors' constraints
+ * as locally defined rather than inherited.
+ */
+ con->coninhcount--;
+ con->conislocal = true;
+
+ simple_heap_update(conrel, ©_tuple->t_self, copy_tuple);
+ CatalogUpdateIndexes(conrel, copy_tuple);
+
+ /* Make update visible */
+ CommandCounterIncrement();
+ }
+
+ heap_freetuple(copy_tuple);
+ }
+
+ systable_endscan(scan);
+
+ if (!found)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("constraint \"%s\" of relation \"%s\" does not exist",
+ constrName,
+ RelationGetRelationName(childrel))));
+
+ heap_close(childrel, NoLock);
+ }
+
+ heap_close(conrel, RowExclusiveLock);
+}
+
+/*
+ * ALTER COLUMN TYPE
+ */
+static void
+ATPrepAlterColumnType(List **wqueue,
+ AlteredTableInfo *tab, Relation rel,
bool recurse, bool recursing,
AlterTableCmd *cmd)
{
Form_pg_attribute attTup;
AttrNumber attnum;
Oid targettype;
+ int32 targettypmod;
Node *transform;
NewColumnValue *newval;
ParseState *pstate = make_parsestate(NULL);
colName)));
/* Look up the target type */
- targettype = typenameTypeId(NULL, typename);
+ targettype = typenameTypeId(NULL, typename, &targettypmod);
/* make sure datatype is legal for a column */
CheckAttributeType(colName, targettype);
transform = coerce_to_target_type(pstate,
transform, exprType(transform),
- targettype, typename->typmod,
+ targettype, targettypmod,
COERCION_ASSIGNMENT,
- COERCE_IMPLICIT_CAST);
+ COERCE_IMPLICIT_CAST,
+ -1);
if (transform == NULL)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("column \"%s\" cannot be cast to type \"%s\"",
- colName, TypeNameToString(typename))));
+ errmsg("column \"%s\" cannot be cast to type %s",
+ colName, format_type_be(targettype))));
/*
* Add a work queue item to make ATRewriteTable update the column
HeapTuple typeTuple;
Form_pg_type tform;
Oid targettype;
+ int32 targettypmod;
Node *defaultexpr;
Relation attrelation;
Relation depRel;
colName)));
/* Look up the target type (should not fail, since prep found it) */
- typeTuple = typenameType(NULL, typename);
+ typeTuple = typenameType(NULL, typename, &targettypmod);
tform = (Form_pg_type) GETSTRUCT(typeTuple);
targettype = HeapTupleGetOid(typeTuple);
defaultexpr = strip_implicit_coercions(defaultexpr);
defaultexpr = coerce_to_target_type(NULL, /* no UNKNOWN params */
defaultexpr, exprType(defaultexpr),
- targettype, typename->typmod,
+ targettype, targettypmod,
COERCION_ASSIGNMENT,
- COERCE_IMPLICIT_CAST);
+ COERCE_IMPLICIT_CAST,
+ -1);
if (defaultexpr == NULL)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("default for column \"%s\" cannot be cast to type \"%s\"",
- colName, TypeNameToString(typename))));
+ errmsg("default for column \"%s\" cannot be cast to type %s",
+ colName, format_type_be(targettype))));
}
else
defaultexpr = NULL;
case OCLASS_LANGUAGE:
case OCLASS_OPERATOR:
case OCLASS_OPCLASS:
+ case OCLASS_OPFAMILY:
case OCLASS_TRIGGER:
case OCLASS_SCHEMA:
+ case OCLASS_TSPARSER:
+ case OCLASS_TSDICT:
+ case OCLASS_TSTEMPLATE:
+ case OCLASS_TSCONFIG:
/*
* We don't expect any of these sorts of objects to depend on
* copy of the syscache entry, so okay to scribble on.)
*/
attTup->atttypid = targettype;
- attTup->atttypmod = typename->typmod;
+ attTup->atttypmod = targettypmod;
attTup->attndims = list_length(typename->arrayBounds);
attTup->attlen = tform->typlen;
attTup->attbyval = tform->typbyval;
*/
RemoveAttrDefault(RelationGetRelid(rel), attnum, DROP_RESTRICT, true);
- StoreAttrDefault(rel, attnum, nodeToString(defaultexpr));
+ StoreAttrDefault(rel, attnum, defaultexpr);
}
/* Cleanup */
ListCell *list_item;
/*
- * We expect that we only have to do raw parsing and parse analysis, not
- * any rule rewriting, since these will all be utility statements.
+ * We expect that we will get only ALTER TABLE and CREATE INDEX
+ * statements. Hence, there is no need to pass them through
+ * parse_analyze() or the rewriter, but instead we need to pass them
+ * through parse_utilcmd.c to make them ready for execution.
*/
raw_parsetree_list = raw_parser(cmd);
querytree_list = NIL;
foreach(list_item, raw_parsetree_list)
{
- Node *parsetree = (Node *) lfirst(list_item);
-
- querytree_list = list_concat(querytree_list,
- parse_analyze(parsetree, cmd, NULL, 0));
+ Node *stmt = (Node *) lfirst(list_item);
+
+ if (IsA(stmt, IndexStmt))
+ querytree_list = lappend(querytree_list,
+ transformIndexStmt((IndexStmt *) stmt,
+ cmd));
+ else if (IsA(stmt, AlterTableStmt))
+ querytree_list = list_concat(querytree_list,
+ transformAlterTableStmt((AlterTableStmt *) stmt,
+ cmd));
+ else
+ querytree_list = lappend(querytree_list, stmt);
}
/*
*/
foreach(list_item, querytree_list)
{
- Query *query = (Query *) lfirst(list_item);
+ Node *stm = (Node *) lfirst(list_item);
Relation rel;
AlteredTableInfo *tab;
- Assert(IsA(query, Query));
- Assert(query->commandType == CMD_UTILITY);
- switch (nodeTag(query->utilityStmt))
+ switch (nodeTag(stm))
{
case T_IndexStmt:
{
- IndexStmt *stmt = (IndexStmt *) query->utilityStmt;
+ IndexStmt *stmt = (IndexStmt *) stm;
AlterTableCmd *newcmd;
rel = relation_openrv(stmt->relation, AccessExclusiveLock);
}
case T_AlterTableStmt:
{
- AlterTableStmt *stmt = (AlterTableStmt *) query->utilityStmt;
+ AlterTableStmt *stmt = (AlterTableStmt *) stm;
ListCell *lcmd;
rel = relation_openrv(stmt->relation, AccessExclusiveLock);
}
default:
elog(ERROR, "unexpected statement type: %d",
- (int) nodeTag(query->utilityStmt));
+ (int) nodeTag(stm));
}
}
}
* be changed separately from the parent table. Also, we can skip permission
* checks (this is necessary not just an optimization, else we'd fail to
* handle toast tables properly).
+ *
+ * recursing is also true if ALTER TYPE OWNER is calling us to fix up a
+ * free-standing composite type.
*/
void
ATExecChangeOwner(Oid relationOid, Oid newOwnerId, bool recursing)
get_rel_name(tableId))));
}
break;
+ case RELKIND_COMPOSITE_TYPE:
+ if (recursing)
+ break;
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is a composite type",
+ NameStr(tuple_class->relname)),
+ errhint("Use ALTER TYPE instead.")));
+ break;
case RELKIND_TOASTVALUE:
if (recursing)
break;
if (tuple_class->relowner != newOwnerId)
{
Datum repl_val[Natts_pg_class];
- char repl_null[Natts_pg_class];
- char repl_repl[Natts_pg_class];
+ bool repl_null[Natts_pg_class];
+ bool repl_repl[Natts_pg_class];
Acl *newAcl;
Datum aclDatum;
bool isNull;
}
}
- memset(repl_null, ' ', sizeof(repl_null));
- memset(repl_repl, ' ', sizeof(repl_repl));
+ memset(repl_null, false, sizeof(repl_null));
+ memset(repl_repl, false, sizeof(repl_repl));
- repl_repl[Anum_pg_class_relowner - 1] = 'r';
+ repl_repl[Anum_pg_class_relowner - 1] = true;
repl_val[Anum_pg_class_relowner - 1] = ObjectIdGetDatum(newOwnerId);
/*
{
newAcl = aclnewowner(DatumGetAclP(aclDatum),
tuple_class->relowner, newOwnerId);
- repl_repl[Anum_pg_class_relacl - 1] = 'r';
+ repl_repl[Anum_pg_class_relacl - 1] = true;
repl_val[Anum_pg_class_relacl - 1] = PointerGetDatum(newAcl);
}
- newtuple = heap_modifytuple(tuple, RelationGetDescr(class_rel), repl_val, repl_null, repl_repl);
+ newtuple = heap_modify_tuple(tuple, RelationGetDescr(class_rel), repl_val, repl_null, repl_repl);
simple_heap_update(class_rel, &newtuple->t_self, newtuple);
CatalogUpdateIndexes(class_rel, newtuple);
heap_freetuple(newtuple);
- /* Update owner dependency reference */
- changeDependencyOnOwner(RelationRelationId, relationOid, newOwnerId);
+ /*
+ * Update owner dependency reference, if any. A composite type has
+ * none, because it's tracked for the pg_type entry instead of here;
+ * indexes and TOAST tables don't have their own entries either.
+ */
+ if (tuple_class->relkind != RELKIND_COMPOSITE_TYPE &&
+ tuple_class->relkind != RELKIND_INDEX &&
+ tuple_class->relkind != RELKIND_TOASTVALUE)
+ changeDependencyOnOwner(RelationRelationId, relationOid,
+ newOwnerId);
/*
* Also change the ownership of the table's rowtype, if it has one
*/
if (tuple_class->relkind != RELKIND_INDEX)
- AlterTypeOwnerInternal(tuple_class->reltype, newOwnerId);
+ AlterTypeOwnerInternal(tuple_class->reltype, newOwnerId,
+ tuple_class->relkind == RELKIND_COMPOSITE_TYPE);
/*
* If we are operating on a table, also change the ownership of any
bool isnull;
Datum newOptions;
Datum repl_val[Natts_pg_class];
- char repl_null[Natts_pg_class];
- char repl_repl[Natts_pg_class];
+ bool repl_null[Natts_pg_class];
+ bool repl_repl[Natts_pg_class];
if (defList == NIL)
return; /* nothing to do */
* propagated into relcaches during post-commit cache inval.
*/
memset(repl_val, 0, sizeof(repl_val));
- memset(repl_null, ' ', sizeof(repl_null));
- memset(repl_repl, ' ', sizeof(repl_repl));
+ memset(repl_null, false, sizeof(repl_null));
+ memset(repl_repl, false, sizeof(repl_repl));
if (newOptions != (Datum) 0)
repl_val[Anum_pg_class_reloptions - 1] = newOptions;
else
- repl_null[Anum_pg_class_reloptions - 1] = 'n';
+ repl_null[Anum_pg_class_reloptions - 1] = true;
- repl_repl[Anum_pg_class_reloptions - 1] = 'r';
+ repl_repl[Anum_pg_class_reloptions - 1] = true;
- newtuple = heap_modifytuple(tuple, RelationGetDescr(pgclass),
+ newtuple = heap_modify_tuple(tuple, RelationGetDescr(pgclass),
repl_val, repl_null, repl_repl);
simple_heap_update(pgclass, &newtuple->t_self, newtuple);
Oid oldTableSpace;
Oid reltoastrelid;
Oid reltoastidxid;
+ Oid newrelfilenode;
RelFileNode newrnode;
SMgrRelation dstrel;
Relation pg_class;
HeapTuple tuple;
Form_pg_class rd_rel;
+ ForkNumber forkNum;
/*
* Need lock here in case we are recursing to toast table or index
errmsg("cannot move system relation \"%s\"",
RelationGetRelationName(rel))));
+ /* Can't move a non-shared relation into pg_global */
+ if (newTableSpace == GLOBALTABLESPACE_OID)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("only shared relations can be placed in pg_global tablespace")));
+
/*
* Don't allow moving temp tables of other backends ... their local buffer
* manager is not going to cope.
elog(ERROR, "cache lookup failed for relation %u", tableOid);
rd_rel = (Form_pg_class) GETSTRUCT(tuple);
- /* create another storage file. Is it a little ugly ? */
- /* NOTE: any conflict in relfilenode value will be caught here */
+ /*
+ * Since we copy the file directly without looking at the shared buffers,
+ * we'd better first flush out any pages of the source relation that are
+ * in shared buffers. We assume no new changes will be made while we are
+ * holding exclusive lock on the rel.
+ */
+ FlushRelationBuffers(rel);
+
+ /*
+ * Relfilenodes are not unique across tablespaces, so we need to allocate
+ * a new one in the new tablespace.
+ */
+ newrelfilenode = GetNewRelFileNode(newTableSpace,
+ rel->rd_rel->relisshared,
+ NULL);
+
+ /* Open old and new relation */
newrnode = rel->rd_node;
+ newrnode.relNode = newrelfilenode;
newrnode.spcNode = newTableSpace;
-
dstrel = smgropen(newrnode);
- smgrcreate(dstrel, rel->rd_istemp, false);
-
- /* copy relation data to the new physical file */
- copy_relation_data(rel, dstrel);
- /* schedule unlinking old physical file */
RelationOpenSmgr(rel);
- smgrscheduleunlink(rel->rd_smgr, rel->rd_istemp);
/*
- * Now drop smgr references. The source was already dropped by
- * smgrscheduleunlink.
+ * Create and copy all forks of the relation, and schedule unlinking
+ * of old physical files.
+ *
+ * NOTE: any conflict in relfilenode value will be caught in
+ * RelationCreateStorage().
*/
+ RelationCreateStorage(newrnode, rel->rd_istemp);
+
+ /* copy main fork */
+ copy_relation_data(rel->rd_smgr, dstrel, MAIN_FORKNUM, rel->rd_istemp);
+
+ /* copy those extra forks that exist */
+ for (forkNum = MAIN_FORKNUM + 1; forkNum <= MAX_FORKNUM; forkNum++)
+ {
+ if (smgrexists(rel->rd_smgr, forkNum))
+ {
+ smgrcreate(dstrel, forkNum, false);
+ copy_relation_data(rel->rd_smgr, dstrel, forkNum, rel->rd_istemp);
+ }
+ }
+
+ /* drop old relation, and close new one */
+ RelationDropStorage(rel);
smgrclose(dstrel);
/* update the pg_class row */
rd_rel->reltablespace = (newTableSpace == MyDatabaseTableSpace) ? InvalidOid : newTableSpace;
+ rd_rel->relfilenode = newrelfilenode;
simple_heap_update(pg_class, &tuple->t_self, tuple);
CatalogUpdateIndexes(pg_class, tuple);
* Copy data, block by block
*/
static void
-copy_relation_data(Relation rel, SMgrRelation dst)
+copy_relation_data(SMgrRelation src, SMgrRelation dst,
+ ForkNumber forkNum, bool istemp)
{
- SMgrRelation src;
bool use_wal;
BlockNumber nblocks;
BlockNumber blkno;
char buf[BLCKSZ];
Page page = (Page) buf;
- /*
- * Since we copy the file directly without looking at the shared buffers,
- * we'd better first flush out any pages of the source relation that are
- * in shared buffers. We assume no new changes will be made while we are
- * holding exclusive lock on the rel.
- */
- FlushRelationBuffers(rel);
-
/*
* We need to log the copied data in WAL iff WAL archiving is enabled AND
* it's not a temp rel.
*/
- use_wal = XLogArchivingActive() && !rel->rd_istemp;
+ use_wal = XLogArchivingActive() && !istemp;
- nblocks = RelationGetNumberOfBlocks(rel);
- /* RelationGetNumberOfBlocks will certainly have opened rd_smgr */
- src = rel->rd_smgr;
+ nblocks = smgrnblocks(src, forkNum);
for (blkno = 0; blkno < nblocks; blkno++)
{
- smgrread(src, blkno, buf);
+ smgrread(src, forkNum, blkno, buf);
/* XLOG stuff */
if (use_wal)
- {
- xl_heap_newpage xlrec;
- XLogRecPtr recptr;
- XLogRecData rdata[2];
-
- /* NO ELOG(ERROR) from here till newpage op is logged */
- START_CRIT_SECTION();
-
- xlrec.node = dst->smgr_rnode;
- xlrec.blkno = blkno;
-
- rdata[0].data = (char *) &xlrec;
- rdata[0].len = SizeOfHeapNewpage;
- rdata[0].buffer = InvalidBuffer;
- rdata[0].next = &(rdata[1]);
-
- rdata[1].data = (char *) page;
- rdata[1].len = BLCKSZ;
- rdata[1].buffer = InvalidBuffer;
- rdata[1].next = NULL;
-
- recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_NEWPAGE, rdata);
-
- PageSetLSN(page, recptr);
- PageSetTLI(page, ThisTimeLineID);
-
- END_CRIT_SECTION();
- }
+ log_newpage(&dst->smgr_rnode, forkNum, blkno, page);
/*
* Now write the page. We say isTemp = true even if it's not a temp
* rel, because there's no need for smgr to schedule an fsync for this
* write; we'll do it ourselves below.
*/
- smgrwrite(dst, blkno, buf, true);
+ smgrextend(dst, forkNum, blkno, buf, true);
}
/*
* wouldn't replay our earlier WAL entries. If we do not fsync those pages
* here, they might still not be on disk when the crash occurs.
*/
- if (!rel->rd_istemp)
- smgrimmedsync(dst);
+ if (!istemp)
+ smgrimmedsync(dst, forkNum);
}
/*
*/
static void
ATExecEnableDisableTrigger(Relation rel, char *trigname,
- bool enable, bool skip_system)
+ char fires_when, bool skip_system)
{
- EnableDisableTrigger(rel, trigname, enable, skip_system);
+ EnableDisableTrigger(rel, trigname, fires_when, skip_system);
}
-static char *
-decompile_conbin(HeapTuple contup, TupleDesc tupdesc)
+/*
+ * ALTER TABLE ENABLE/DISABLE RULE
+ *
+ * We just pass this off to rewriteDefine.c.
+ */
+static void
+ATExecEnableDisableRule(Relation rel, char *trigname,
+ char fires_when)
{
- Form_pg_constraint con;
- bool isnull;
- Datum attr;
- Datum expr;
-
- con = (Form_pg_constraint) GETSTRUCT(contup);
- attr = heap_getattr(contup, Anum_pg_constraint_conbin, tupdesc, &isnull);
- if (isnull)
- elog(ERROR, "null conbin for constraint %u", HeapTupleGetOid(contup));
-
- expr = DirectFunctionCall2(pg_get_expr, attr,
- ObjectIdGetDatum(con->conrelid));
- return DatumGetCString(DirectFunctionCall1(textout, expr));
+ EnableDisableRule(rel, trigname, fires_when);
}
/*
*
* Add a parent to the child's parents. This verifies that all the columns and
* check constraints of the parent appear in the child and that they have the
- * same data type and expressions.
+ * same data types and expressions.
*/
static void
-ATExecAddInherits(Relation child_rel, RangeVar *parent)
+ATExecAddInherit(Relation child_rel, RangeVar *parent)
{
Relation parent_rel,
catalogRelation;
SysScanDesc scan;
ScanKeyData key;
HeapTuple inheritsTuple;
- int4 inhseqno;
+ int32 inhseqno;
List *children;
+ /*
+ * AccessShareLock on the parent is what's obtained during normal CREATE
+ * TABLE ... INHERITS ..., so should be enough here.
+ */
parent_rel = heap_openrv(parent, AccessShareLock);
/*
- * Must be owner of both parent and child -- child is taken care of by
+ * Must be owner of both parent and child -- child was checked by
* ATSimplePermissions call in ATPrepCmd
*/
ATSimplePermissions(parent_rel, false);
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot inherit from temporary relation \"%s\"",
- parent->relname)));
-
- /* If parent has OIDs then all children must have OIDs */
- if (parent_rel->rd_rel->relhasoids && !child_rel->rd_rel->relhasoids)
- ereport(ERROR,
- (errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("table \"%s\" without OIDs cannot inherit from table \"%s\" with OIDs",
- RelationGetRelationName(child_rel), parent->relname)));
+ RelationGetRelationName(parent_rel))));
/*
- * Don't allow any duplicates in the list of parents. We scan through the
- * list of parents in pg_inherit and keep track of the first open inhseqno
- * slot found to use for the new parent.
+ * Check for duplicates in the list of parents, and determine the highest
+ * inhseqno already present; we'll use the next one for the new parent.
+ * (Note: get RowExclusiveLock because we will write pg_inherits below.)
+ *
+ * Note: we do not reject the case where the child already inherits from
+ * the parent indirectly; CREATE TABLE doesn't reject comparable cases.
*/
catalogRelation = heap_open(InheritsRelationId, RowExclusiveLock);
ScanKeyInit(&key,
if (inh->inhparent == RelationGetRelid(parent_rel))
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_TABLE),
- errmsg("inherited relation \"%s\" duplicated",
- parent->relname)));
- if (inh->inhseqno == inhseqno + 1)
+ errmsg("relation \"%s\" would be inherited from more than once",
+ RelationGetRelationName(parent_rel))));
+ if (inh->inhseqno > inhseqno)
inhseqno = inh->inhseqno;
}
systable_endscan(scan);
- heap_close(catalogRelation, RowExclusiveLock);
/*
- * If the new parent is found in our list of inheritors, we have a
- * circular structure
+ * Prevent circularity by seeing if proposed parent inherits from child.
+ * (In particular, this disallows making a rel inherit from itself.)
+ *
+ * This is not completely bulletproof because of race conditions: in
+ * multi-level inheritance trees, someone else could concurrently be
+ * making another inheritance link that closes the loop but does not join
+ * either of the rels we have locked. Preventing that seems to require
+ * exclusive locks on the entire inheritance tree, which is a cure worse
+ * than the disease. find_all_inheritors() will cope with circularity
+ * anyway, so don't sweat it too much.
*/
children = find_all_inheritors(RelationGetRelid(child_rel));
if (list_member_oid(children, RelationGetRelid(parent_rel)))
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_TABLE),
- errmsg("circular inheritance structure found"),
+ errmsg("circular inheritance not allowed"),
errdetail("\"%s\" is already a child of \"%s\".",
parent->relname,
RelationGetRelationName(child_rel))));
- /* Match up the columns and bump attinhcount and attislocal */
+ /* If parent has OIDs then child must have OIDs */
+ if (parent_rel->rd_rel->relhasoids && !child_rel->rd_rel->relhasoids)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("table \"%s\" without OIDs cannot inherit from table \"%s\" with OIDs",
+ RelationGetRelationName(child_rel),
+ RelationGetRelationName(parent_rel))));
+
+ /* Match up the columns and bump attinhcount as needed */
MergeAttributesIntoExisting(child_rel, parent_rel);
- /* Match up the constraints and make sure they're present in child */
+ /* Match up the constraints and bump coninhcount as needed */
MergeConstraintsIntoExisting(child_rel, parent_rel);
- catalogRelation = heap_open(InheritsRelationId, RowExclusiveLock);
+ /*
+ * OK, it looks valid. Make the catalog entries that show inheritance.
+ */
StoreCatalogInheritance1(RelationGetRelid(child_rel),
RelationGetRelid(parent_rel),
- inhseqno + 1, catalogRelation);
+ inhseqno + 1,
+ catalogRelation);
+
+ /* Now we're done with pg_inherits */
heap_close(catalogRelation, RowExclusiveLock);
/* keep our lock on the parent relation until commit */
}
/*
- * Check columns in child table match up with columns in parent
+ * Obtain the source-text form of the constraint expression for a check
+ * constraint, given its pg_constraint tuple
+ */
+static char *
+decompile_conbin(HeapTuple contup, TupleDesc tupdesc)
+{
+ Form_pg_constraint con;
+ bool isnull;
+ Datum attr;
+ Datum expr;
+
+ con = (Form_pg_constraint) GETSTRUCT(contup);
+ attr = heap_getattr(contup, Anum_pg_constraint_conbin, tupdesc, &isnull);
+ if (isnull)
+ elog(ERROR, "null conbin for constraint %u", HeapTupleGetOid(contup));
+
+ expr = DirectFunctionCall2(pg_get_expr, attr,
+ ObjectIdGetDatum(con->conrelid));
+ return TextDatumGetCString(expr);
+}
+
+/*
+ * Determine whether two check constraints are functionally equivalent
+ *
+ * The test we apply is to see whether they reverse-compile to the same
+ * source string. This insulates us from issues like whether attributes
+ * have the same physical column numbers in parent and child relations.
+ */
+static bool
+constraints_equivalent(HeapTuple a, HeapTuple b, TupleDesc tupleDesc)
+{
+ Form_pg_constraint acon = (Form_pg_constraint) GETSTRUCT(a);
+ Form_pg_constraint bcon = (Form_pg_constraint) GETSTRUCT(b);
+
+ if (acon->condeferrable != bcon->condeferrable ||
+ acon->condeferred != bcon->condeferred ||
+ strcmp(decompile_conbin(a, tupleDesc),
+ decompile_conbin(b, tupleDesc)) != 0)
+ return false;
+ else
+ return true;
+}
+
+/*
+ * Check columns in child table match up with columns in parent, and increment
+ * their attinhcount.
*
- * Called by ATExecAddInherits
+ * Called by ATExecAddInherit
*
- * Currently all columns must be found in child. Missing columns are an error.
- * One day we might consider creating new columns like CREATE TABLE does.
+ * Currently all parent columns must be found in child. Missing columns are an
+ * error. One day we might consider creating new columns like CREATE TABLE
+ * does. However, that is widely unpopular --- in the common use case of
+ * partitioned tables it's a foot-gun.
*
- * The data type must match perfectly. If the parent column is NOT NULL then
- * the child table must be as well. Defaults are ignored however.
+ * The data type must match exactly. If the parent column is NOT NULL then
+ * the child must be as well. Defaults are not compared, however.
*/
static void
MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel)
{
- Relation attrdesc;
+ Relation attrrel;
AttrNumber parent_attno;
int parent_natts;
TupleDesc tupleDesc;
TupleConstr *constr;
HeapTuple tuple;
+ attrrel = heap_open(AttributeRelationId, RowExclusiveLock);
+
tupleDesc = RelationGetDescr(parent_rel);
parent_natts = tupleDesc->natts;
constr = tupleDesc->constr;
if (attribute->attisdropped)
continue;
- /* Does it conflict with an existing column? */
- attrdesc = heap_open(AttributeRelationId, RowExclusiveLock);
-
+ /* Find same column in child (matching on column name). */
tuple = SearchSysCacheCopyAttName(RelationGetRelid(child_rel),
attributeName);
if (HeapTupleIsValid(tuple))
{
- /*
- * Yes, try to merge the two column definitions. They must have
- * the same type and typmod.
- */
+ /* Check they are same type and typmod */
Form_pg_attribute childatt = (Form_pg_attribute) GETSTRUCT(tuple);
if (attribute->atttypid != childatt->atttypid ||
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("child table \"%s\" has different type for column \"%s\"",
- RelationGetRelationName(child_rel), NameStr(attribute->attname))));
+ RelationGetRelationName(child_rel),
+ attributeName)));
if (attribute->attnotnull && !childatt->attnotnull)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("column \"%s\" in child table must be NOT NULL",
- NameStr(attribute->attname))));
-
- childatt->attinhcount++;
- simple_heap_update(attrdesc, &tuple->t_self, tuple);
- /* XXX strength reduce open indexes to outside loop? */
- CatalogUpdateIndexes(attrdesc, tuple);
- heap_freetuple(tuple);
+ errmsg("column \"%s\" in child table must be marked NOT NULL",
+ attributeName)));
/*
- * We don't touch default at all since we're not making any other
- * DDL changes to the child
+ * OK, bump the child column's inheritance count. (If we fail
+ * later on, this change will just roll back.)
*/
+ childatt->attinhcount++;
+ simple_heap_update(attrrel, &tuple->t_self, tuple);
+ CatalogUpdateIndexes(attrrel, tuple);
+ heap_freetuple(tuple);
}
else
{
- /*
- * Creating inherited columns in this case seems to be unpopular.
- * In the common use case of partitioned tables it's a foot-gun.
- */
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("child table missing column \"%s\"",
- NameStr(attribute->attname))));
+ errmsg("child table is missing column \"%s\"",
+ attributeName)));
}
-
- heap_close(attrdesc, RowExclusiveLock);
}
+
+ heap_close(attrrel, RowExclusiveLock);
}
/*
- * Check constraints in child table match up with constraints in parent
+ * Check constraints in child table match up with constraints in parent,
+ * and increment their coninhcount.
*
- * Called by ATExecAddInherits
+ * Called by ATExecAddInherit
*
* Currently all constraints in parent must be present in the child. One day we
* may consider adding new constraints like CREATE TABLE does. We may also want
* make it possible to ensure no records are mistakenly inserted into the
* master in partitioned tables rather than the appropriate child.
*
- * XXX this is O(n^2) which may be issue with tables with hundreds of
+ * XXX This is O(N^2) which may be an issue with tables with hundreds of
* constraints. As long as tables have more like 10 constraints it shouldn't be
- * an issue though. Even 100 constraints ought not be the end of the world.
+ * a problem though. Even 100 constraints ought not be the end of the world.
*/
static void
MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel)
{
- Relation catalogRelation;
- TupleDesc tupleDesc;
- SysScanDesc scan;
- ScanKeyData key;
- HeapTuple constraintTuple;
- ListCell *elem;
- List *constraints;
+ Relation catalog_relation;
+ TupleDesc tuple_desc;
+ SysScanDesc parent_scan;
+ ScanKeyData parent_key;
+ HeapTuple parent_tuple;
- /* First gather up the child's constraint definitions */
- catalogRelation = heap_open(ConstraintRelationId, AccessShareLock);
- tupleDesc = RelationGetDescr(catalogRelation);
+ catalog_relation = heap_open(ConstraintRelationId, RowExclusiveLock);
+ tuple_desc = RelationGetDescr(catalog_relation);
- ScanKeyInit(&key,
+ /* Outer loop scans through the parent's constraint definitions */
+ ScanKeyInit(&parent_key,
Anum_pg_constraint_conrelid,
- BTEqualStrategyNumber,
- F_OIDEQ,
- ObjectIdGetDatum(RelationGetRelid(child_rel)));
- scan = systable_beginscan(catalogRelation, ConstraintRelidIndexId,
- true, SnapshotNow, 1, &key);
- constraints = NIL;
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(parent_rel)));
+ parent_scan = systable_beginscan(catalog_relation, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, &parent_key);
- while (HeapTupleIsValid(constraintTuple = systable_getnext(scan)))
+ while (HeapTupleIsValid(parent_tuple = systable_getnext(parent_scan)))
{
- Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(constraintTuple);
+ Form_pg_constraint parent_con = (Form_pg_constraint) GETSTRUCT(parent_tuple);
+ SysScanDesc child_scan;
+ ScanKeyData child_key;
+ HeapTuple child_tuple;
+ bool found = false;
- if (con->contype != CONSTRAINT_CHECK)
+ if (parent_con->contype != CONSTRAINT_CHECK)
continue;
- constraints = lappend(constraints, heap_copytuple(constraintTuple));
- }
- systable_endscan(scan);
+ /* Search for a child constraint matching this one */
+ ScanKeyInit(&child_key,
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(child_rel)));
+ child_scan = systable_beginscan(catalog_relation, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, &child_key);
- /* Then loop through the parent's constraints looking for them in the list */
- ScanKeyInit(&key,
- Anum_pg_constraint_conrelid,
- BTEqualStrategyNumber,
- F_OIDEQ,
- ObjectIdGetDatum(RelationGetRelid(parent_rel)));
- scan = systable_beginscan(catalogRelation, ConstraintRelidIndexId, true,
- SnapshotNow, 1, &key);
- while (HeapTupleIsValid(constraintTuple = systable_getnext(scan)))
- {
- bool found = false;
- Form_pg_constraint parent_con = (Form_pg_constraint) GETSTRUCT(constraintTuple);
- Form_pg_constraint child_con = NULL;
- HeapTuple child_contuple = NULL;
+ while (HeapTupleIsValid(child_tuple = systable_getnext(child_scan)))
+ {
+ Form_pg_constraint child_con = (Form_pg_constraint) GETSTRUCT(child_tuple);
+ HeapTuple child_copy;
- if (parent_con->contype != CONSTRAINT_CHECK)
- continue;
+ if (child_con->contype != CONSTRAINT_CHECK)
+ continue;
- foreach(elem, constraints)
- {
- child_contuple = lfirst(elem);
- child_con = (Form_pg_constraint) GETSTRUCT(child_contuple);
- if (!strcmp(NameStr(parent_con->conname),
- NameStr(child_con->conname)))
- {
- found = true;
- break;
- }
+ if (strcmp(NameStr(parent_con->conname),
+ NameStr(child_con->conname)) != 0)
+ continue;
+
+ if (!constraints_equivalent(parent_tuple, child_tuple, tuple_desc))
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("child table \"%s\" has different definition for check constraint \"%s\"",
+ RelationGetRelationName(child_rel),
+ NameStr(parent_con->conname))));
+
+ /*
+ * OK, bump the child constraint's inheritance count. (If we fail
+ * later on, this change will just roll back.)
+ */
+ child_copy = heap_copytuple(child_tuple);
+ child_con = (Form_pg_constraint) GETSTRUCT(child_copy);
+ child_con->coninhcount++;
+ simple_heap_update(catalog_relation, &child_copy->t_self, child_copy);
+ CatalogUpdateIndexes(catalog_relation, child_copy);
+ heap_freetuple(child_copy);
+
+ found = true;
+ break;
}
- if (!found)
- ereport(ERROR,
- (errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("child table missing constraint matching parent table constraint \"%s\"",
- NameStr(parent_con->conname))));
+ systable_endscan(child_scan);
- if (parent_con->condeferrable != child_con->condeferrable ||
- parent_con->condeferred != child_con->condeferred ||
- parent_con->contypid != child_con->contypid ||
- strcmp(decompile_conbin(constraintTuple, tupleDesc),
- decompile_conbin(child_contuple, tupleDesc)))
+ if (!found)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("constraint definition for CHECK constraint \"%s\" doesn't match",
+ errmsg("child table is missing constraint \"%s\"",
NameStr(parent_con->conname))));
-
- /*
- * TODO: add conislocal,coninhcount to constraints. This is where we
- * would have to bump them just like attributes
- */
}
- systable_endscan(scan);
- heap_close(catalogRelation, AccessShareLock);
+ systable_endscan(parent_scan);
+ heap_close(catalog_relation, RowExclusiveLock);
}
/*
* and attislocal of the columns and removes the pg_inherit and pg_depend
* entries.
*
- * If attinhcount goes to 0 then attislocal gets set to true. If it goes back up
- * attislocal stays 0 which means if a child is ever removed from a parent then
- * its columns will never be automatically dropped which may surprise. But at
- * least we'll never surprise by dropping columns someone isn't expecting to be
- * dropped which would actually mean data loss.
+ * If attinhcount goes to 0 then attislocal gets set to true. If it goes back
+ * up attislocal stays true, which means if a child is ever removed from a
+ * parent then its columns will never be automatically dropped which may
+ * surprise. But at least we'll never surprise by dropping columns someone
+ * isn't expecting to be dropped which would actually mean data loss.
+ *
+ * coninhcount and conislocal for inherited constraints are adjusted in
+ * exactly the same way.
*/
static void
-ATExecDropInherits(Relation rel, RangeVar *parent)
+ATExecDropInherit(Relation rel, RangeVar *parent)
{
+ Relation parent_rel;
Relation catalogRelation;
SysScanDesc scan;
- ScanKeyData key[2];
+ ScanKeyData key[3];
HeapTuple inheritsTuple,
attributeTuple,
+ constraintTuple,
depTuple;
- Oid inhparent;
- Oid dropparent;
- int found = false;
+ List *connames;
+ bool found = false;
/*
- * Get the OID of parent -- if no schema is specified use the regular
- * search path and only drop the one table that's found. We could try to
- * be clever and look at each parent and see if it matches but that would
- * be inconsistent with other operations I think.
+ * AccessShareLock on the parent is probably enough, seeing that DROP
+ * TABLE doesn't lock parent tables at all. We need some lock since we'll
+ * be inspecting the parent's schema.
*/
- dropparent = RangeVarGetRelid(parent, false);
+ parent_rel = heap_openrv(parent, AccessShareLock);
- /* Search through the direct parents of rel looking for dropparent oid */
+ /*
+ * We don't bother to check ownership of the parent table --- ownership of
+ * the child is presumed enough rights.
+ */
+ /*
+ * Find and destroy the pg_inherits entry linking the two, or error out if
+ * there is none.
+ */
catalogRelation = heap_open(InheritsRelationId, RowExclusiveLock);
- ScanKeyInit(key,
+ ScanKeyInit(&key[0],
Anum_pg_inherits_inhrelid,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(RelationGetRelid(rel)));
while (HeapTupleIsValid(inheritsTuple = systable_getnext(scan)))
{
+ Oid inhparent;
+
inhparent = ((Form_pg_inherits) GETSTRUCT(inheritsTuple))->inhparent;
- if (inhparent == dropparent)
+ if (inhparent == RelationGetRelid(parent_rel))
{
simple_heap_delete(catalogRelation, &inheritsTuple->t_self);
found = true;
heap_close(catalogRelation, RowExclusiveLock);
if (!found)
- {
- if (parent->schemaname)
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_TABLE),
- errmsg("relation \"%s.%s\" is not a parent of relation \"%s\"",
- parent->schemaname, parent->relname, RelationGetRelationName(rel))));
- else
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_TABLE),
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_TABLE),
errmsg("relation \"%s\" is not a parent of relation \"%s\"",
- parent->relname, RelationGetRelationName(rel))));
- }
-
- /* Search through columns looking for matching columns from parent table */
+ RelationGetRelationName(parent_rel),
+ RelationGetRelationName(rel))));
+ /*
+ * Search through child columns looking for ones matching parent rel
+ */
catalogRelation = heap_open(AttributeRelationId, RowExclusiveLock);
- ScanKeyInit(key,
+ ScanKeyInit(&key[0],
Anum_pg_attribute_attrelid,
- BTEqualStrategyNumber,
- F_OIDEQ,
+ BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(RelationGetRelid(rel)));
scan = systable_beginscan(catalogRelation, AttributeRelidNumIndexId,
true, SnapshotNow, 1, key);
{
Form_pg_attribute att = (Form_pg_attribute) GETSTRUCT(attributeTuple);
- /*
- * Not an inherited column at all (do NOT use islocal for this
- * test--it can be true for inherited columns)
- */
- if (att->attinhcount == 0)
- continue;
+ /* Ignore if dropped or not inherited */
if (att->attisdropped)
continue;
+ if (att->attinhcount <= 0)
+ continue;
- if (SearchSysCacheExistsAttName(dropparent, NameStr(att->attname)))
+ if (SearchSysCacheExistsAttName(RelationGetRelid(parent_rel),
+ NameStr(att->attname)))
{
- /* Decrement inhcount and possibly set islocal to 1 */
+ /* Decrement inhcount and possibly set islocal to true */
HeapTuple copyTuple = heap_copytuple(attributeTuple);
Form_pg_attribute copy_att = (Form_pg_attribute) GETSTRUCT(copyTuple);
copy_att->attislocal = true;
simple_heap_update(catalogRelation, ©Tuple->t_self, copyTuple);
+ CatalogUpdateIndexes(catalogRelation, copyTuple);
+ heap_freetuple(copyTuple);
+ }
+ }
+ systable_endscan(scan);
+ heap_close(catalogRelation, RowExclusiveLock);
- /*
- * XXX "Avoid using it for multiple tuples, since opening the
- * indexes and building the index info structures is moderately
- * expensive." Perhaps this can be moved outside the loop or else
- * at least the CatalogOpenIndexes/CatalogCloseIndexes moved
- * outside the loop but when I try that it seg faults?!
- */
+ /*
+ * Likewise, find inherited check constraints and disinherit them.
+ * To do this, we first need a list of the names of the parent's check
+ * constraints. (We cheat a bit by only checking for name matches,
+ * assuming that the expressions will match.)
+ */
+ catalogRelation = heap_open(ConstraintRelationId, RowExclusiveLock);
+ ScanKeyInit(&key[0],
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(parent_rel)));
+ scan = systable_beginscan(catalogRelation, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, key);
+
+ connames = NIL;
+
+ while (HeapTupleIsValid(constraintTuple = systable_getnext(scan)))
+ {
+ Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(constraintTuple);
+
+ if (con->contype == CONSTRAINT_CHECK)
+ connames = lappend(connames, pstrdup(NameStr(con->conname)));
+ }
+
+ systable_endscan(scan);
+
+ /* Now scan the child's constraints */
+ ScanKeyInit(&key[0],
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(rel)));
+ scan = systable_beginscan(catalogRelation, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, key);
+
+ while (HeapTupleIsValid(constraintTuple = systable_getnext(scan)))
+ {
+ Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(constraintTuple);
+ bool match;
+ ListCell *lc;
+
+ if (con->contype != CONSTRAINT_CHECK)
+ continue;
+
+ match = false;
+ foreach (lc, connames)
+ {
+ if (strcmp(NameStr(con->conname), (char *) lfirst(lc)) == 0)
+ {
+ match = true;
+ break;
+ }
+ }
+
+ if (match)
+ {
+ /* Decrement inhcount and possibly set islocal to true */
+ HeapTuple copyTuple = heap_copytuple(constraintTuple);
+ Form_pg_constraint copy_con = (Form_pg_constraint) GETSTRUCT(copyTuple);
+ if (copy_con->coninhcount <= 0) /* shouldn't happen */
+ elog(ERROR, "relation %u has non-inherited constraint \"%s\"",
+ RelationGetRelid(rel), NameStr(copy_con->conname));
+
+ copy_con->coninhcount--;
+ if (copy_con->coninhcount == 0)
+ copy_con->conislocal = true;
+
+ simple_heap_update(catalogRelation, ©Tuple->t_self, copyTuple);
CatalogUpdateIndexes(catalogRelation, copyTuple);
heap_freetuple(copyTuple);
}
}
+
systable_endscan(scan);
heap_close(catalogRelation, RowExclusiveLock);
*
* There's no convenient way to do this, so go trawling through pg_depend
*/
-
catalogRelation = heap_open(DependRelationId, RowExclusiveLock);
ScanKeyInit(&key[0],
Anum_pg_depend_classid,
BTEqualStrategyNumber, F_OIDEQ,
- RelationRelationId);
+ ObjectIdGetDatum(RelationRelationId));
ScanKeyInit(&key[1],
Anum_pg_depend_objid,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(RelationGetRelid(rel)));
+ ScanKeyInit(&key[2],
+ Anum_pg_depend_objsubid,
+ BTEqualStrategyNumber, F_INT4EQ,
+ Int32GetDatum(0));
scan = systable_beginscan(catalogRelation, DependDependerIndexId, true,
- SnapshotNow, 2, key);
+ SnapshotNow, 3, key);
while (HeapTupleIsValid(depTuple = systable_getnext(scan)))
{
Form_pg_depend dep = (Form_pg_depend) GETSTRUCT(depTuple);
if (dep->refclassid == RelationRelationId &&
- dep->refobjid == dropparent &&
+ dep->refobjid == RelationGetRelid(parent_rel) &&
+ dep->refobjsubid == 0 &&
dep->deptype == DEPENDENCY_NORMAL)
- {
- /*
- * Only delete a single dependency -- there shouldn't be more but
- * just in case...
- */
simple_heap_delete(catalogRelation, &depTuple->t_self);
- break;
- }
}
systable_endscan(scan);
heap_close(catalogRelation, RowExclusiveLock);
+
+ /* keep our lock on the parent relation until commit */
+ heap_close(parent_rel, NoLock);
}
* Note: caller must have checked ownership of the relation already
*/
void
-AlterTableNamespace(RangeVar *relation, const char *newschema)
+AlterTableNamespace(RangeVar *relation, const char *newschema,
+ ObjectType stmttype)
{
Relation rel;
Oid relid;
Oid nspOid;
Relation classRel;
- rel = heap_openrv(relation, AccessExclusiveLock);
+ rel = relation_openrv(relation, AccessExclusiveLock);
relid = RelationGetRelid(rel);
oldNspOid = RelationGetNamespace(rel);
- /* heap_openrv allows TOAST, but we don't want to */
- if (rel->rd_rel->relkind == RELKIND_TOASTVALUE)
- ereport(ERROR,
- (errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("\"%s\" is a TOAST relation",
- RelationGetRelationName(rel))));
+ /* Check relation type against type specified in the ALTER command */
+ switch (stmttype)
+ {
+ case OBJECT_TABLE:
+ /*
+ * For mostly-historical reasons, we allow ALTER TABLE to apply
+ * to all relation types.
+ */
+ break;
- /* if it's an owned sequence, disallow moving it by itself */
- if (rel->rd_rel->relkind == RELKIND_SEQUENCE)
+ case OBJECT_SEQUENCE:
+ if (rel->rd_rel->relkind != RELKIND_SEQUENCE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a sequence",
+ RelationGetRelationName(rel))));
+ break;
+
+ case OBJECT_VIEW:
+ if (rel->rd_rel->relkind != RELKIND_VIEW)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a view",
+ RelationGetRelationName(rel))));
+ break;
+
+ default:
+ elog(ERROR, "unrecognized object type: %d", (int) stmttype);
+ }
+
+ /* Can we change the schema of this tuple? */
+ switch (rel->rd_rel->relkind)
{
- Oid tableId;
- int32 colId;
+ case RELKIND_RELATION:
+ case RELKIND_VIEW:
+ /* ok to change schema */
+ break;
+ case RELKIND_SEQUENCE:
+ {
+ /* if it's an owned sequence, disallow moving it by itself */
+ Oid tableId;
+ int32 colId;
- if (sequenceIsOwned(relid, &tableId, &colId))
+ if (sequenceIsOwned(relid, &tableId, &colId))
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot move an owned sequence into another schema"),
+ errdetail("Sequence \"%s\" is linked to table \"%s\".",
+ RelationGetRelationName(rel),
+ get_rel_name(tableId))));
+ }
+ break;
+ case RELKIND_COMPOSITE_TYPE:
ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot move an owned sequence into another schema"),
- errdetail("Sequence \"%s\" is linked to table \"%s\".",
- RelationGetRelationName(rel),
- get_rel_name(tableId))));
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is a composite type",
+ RelationGetRelationName(rel)),
+ errhint("Use ALTER TYPE instead.")));
+ break;
+ case RELKIND_INDEX:
+ case RELKIND_TOASTVALUE:
+ /* FALL THRU */
+ default:
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("\"%s\" is not a table, view, or sequence",
+ RelationGetRelationName(rel))));
}
/* get schema OID and check its permissions */
AlterRelationNamespaceInternal(classRel, relid, oldNspOid, nspOid, true);
/* Fix the table's rowtype too */
- AlterTypeNamespaceInternal(rel->rd_rel->reltype, nspOid, false);
+ AlterTypeNamespaceInternal(rel->rd_rel->reltype, nspOid, false, false);
/* Fix other dependent stuff */
if (rel->rd_rel->relkind == RELKIND_RELATION)
* them to the new namespace, too.
*/
AlterTypeNamespaceInternal(RelationGetForm(seqRel)->reltype,
- newNspOid, false);
+ newNspOid, false, false);
/* Now we can close it. Keep the lock till end of transaction. */
relation_close(seqRel, NoLock);
projects / postgresql / blobdiff