aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorTom Lane <tgl@sss.pgh.pa.us>2008-05-09 23:32:05 +0000
committerTom Lane <tgl@sss.pgh.pa.us>2008-05-09 23:32:05 +0000
commitcd902b331dc4b0c170e800441a98f9213d98b46b (patch)
treebef3eacf7ff474dd0fb96b368e80137f73658d52 /src
parentf8df836ae396be28a6c9e4f79a6adf3e5c0187b5 (diff)
downloadpostgresql-cd902b331dc4b0c170e800441a98f9213d98b46b.tar.gz
postgresql-cd902b331dc4b0c170e800441a98f9213d98b46b.zip
Change the rules for inherited CHECK constraints to be essentially the same
as those for inherited columns; that is, it's no longer allowed for a child table to not have a check constraint matching one that exists on a parent. This satisfies the principle of least surprise (rows selected from the parent will always appear to meet its check constraints) and eliminates some longstanding bogosity in pg_dump, which formerly had to guess about whether check constraints were really inherited or not. The implementation involves adding conislocal and coninhcount columns to pg_constraint (paralleling attislocal and attinhcount in pg_attribute) and refactoring various ALTER TABLE actions to be more like those for columns. Alex Hunsaker, Nikhil Sontakke, Tom Lane
Diffstat (limited to 'src')
-rw-r--r--src/backend/access/common/tupdesc.c49
-rw-r--r--src/backend/bootstrap/bootparse.y3
-rw-r--r--src/backend/catalog/heap.c372
-rw-r--r--src/backend/catalog/index.c6
-rw-r--r--src/backend/catalog/pg_constraint.c8
-rw-r--r--src/backend/catalog/toasting.c3
-rw-r--r--src/backend/commands/cluster.c10
-rw-r--r--src/backend/commands/tablecmds.c847
-rw-r--r--src/backend/commands/typecmds.c6
-rw-r--r--src/backend/executor/execMain.c5
-rw-r--r--src/bin/pg_dump/common.c47
-rw-r--r--src/bin/pg_dump/pg_dump.c125
-rw-r--r--src/bin/pg_dump/pg_dump.h4
-rw-r--r--src/include/catalog/catversion.h4
-rw-r--r--src/include/catalog/heap.h18
-rw-r--r--src/include/catalog/pg_constraint.h30
-rw-r--r--src/include/nodes/parsenodes.h6
-rw-r--r--src/test/regress/expected/alter_table.out28
-rw-r--r--src/test/regress/expected/inherit.out217
-rw-r--r--src/test/regress/sql/alter_table.sql23
-rw-r--r--src/test/regress/sql/inherit.sql80
21 files changed, 1334 insertions, 557 deletions
diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c
index 7f8df9f6829..1d5a98c069a 100644
--- a/src/backend/access/common/tupdesc.c
+++ b/src/backend/access/common/tupdesc.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/access/common/tupdesc.c,v 1.122 2008/01/01 19:45:46 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/access/common/tupdesc.c,v 1.123 2008/05/09 23:32:04 tgl Exp $
*
* NOTES
* some of the executor utility code such as "ExecTypeFromTL" should be
@@ -505,20 +505,18 @@ BuildDescForRelation(List *schema)
AttrNumber attnum;
ListCell *l;
TupleDesc desc;
- AttrDefault *attrdef = NULL;
- TupleConstr *constr = (TupleConstr *) palloc0(sizeof(TupleConstr));
+ bool has_not_null;
char *attname;
Oid atttypid;
int32 atttypmod;
int attdim;
- int ndef = 0;
/*
* allocate a new tuple descriptor
*/
natts = list_length(schema);
desc = CreateTemplateTupleDesc(natts, false);
- constr->has_not_null = false;
+ has_not_null = false;
attnum = 0;
@@ -547,52 +545,25 @@ BuildDescForRelation(List *schema)
atttypid, atttypmod, attdim);
/* Fill in additional stuff not handled by TupleDescInitEntry */
- if (entry->is_not_null)
- constr->has_not_null = true;
desc->attrs[attnum - 1]->attnotnull = entry->is_not_null;
-
- /*
- * Note we copy only pre-cooked default expressions. Digestion of raw
- * ones is someone else's problem.
- */
- if (entry->cooked_default != NULL)
- {
- if (attrdef == NULL)
- attrdef = (AttrDefault *) palloc(natts * sizeof(AttrDefault));
- attrdef[ndef].adnum = attnum;
- attrdef[ndef].adbin = pstrdup(entry->cooked_default);
- ndef++;
- desc->attrs[attnum - 1]->atthasdef = true;
- }
-
+ has_not_null |= entry->is_not_null;
desc->attrs[attnum - 1]->attislocal = entry->is_local;
desc->attrs[attnum - 1]->attinhcount = entry->inhcount;
}
- if (constr->has_not_null || ndef > 0)
+ if (has_not_null)
{
- desc->constr = constr;
+ TupleConstr *constr = (TupleConstr *) palloc0(sizeof(TupleConstr));
- if (ndef > 0) /* DEFAULTs */
- {
- if (ndef < natts)
- constr->defval = (AttrDefault *)
- repalloc(attrdef, ndef * sizeof(AttrDefault));
- else
- constr->defval = attrdef;
- constr->num_defval = ndef;
- }
- else
- {
- constr->defval = NULL;
- constr->num_defval = 0;
- }
+ constr->has_not_null = true;
+ constr->defval = NULL;
+ constr->num_defval = 0;
constr->check = NULL;
constr->num_check = 0;
+ desc->constr = constr;
}
else
{
- pfree(constr);
desc->constr = NULL;
}
diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y
index a6b904f83eb..da76d76d9d2 100644
--- a/src/backend/bootstrap/bootparse.y
+++ b/src/backend/bootstrap/bootparse.y
@@ -9,7 +9,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/bootstrap/bootparse.y,v 1.91 2008/01/01 19:45:48 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/bootstrap/bootparse.y,v 1.92 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -206,6 +206,7 @@ Boot_CreateStmt:
$6,
BOOTSTRAP_SUPERUSERID,
tupdesc,
+ NIL,
RELKIND_RELATION,
$3,
true,
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 567280e1529..d12e63445a5 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/catalog/heap.c,v 1.332 2008/03/27 03:57:33 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/catalog/heap.c,v 1.333 2008/05/09 23:32:04 tgl Exp $
*
*
* INTERFACE ROUTINES
@@ -77,9 +77,15 @@ static Oid AddNewRelationType(const char *typeName,
char new_rel_kind,
Oid new_array_type);
static void RelationRemoveInheritance(Oid relid);
-static void StoreRelCheck(Relation rel, char *ccname, char *ccbin);
-static void StoreConstraints(Relation rel, TupleDesc tupdesc);
+static void StoreRelCheck(Relation rel, char *ccname, Node *expr,
+ bool is_local, int inhcount);
+static void StoreConstraints(Relation rel, List *cooked_constraints);
+static bool MergeWithExistingConstraint(Relation rel, char *ccname, Node *expr,
+ bool allow_merge, bool is_local);
static void SetRelationNumChecks(Relation rel, int numchecks);
+static Node *cookConstraint(ParseState *pstate,
+ Node *raw_constraint,
+ char *relname);
static List *insert_ordered_unique_oid(List *list, Oid datum);
@@ -788,6 +794,7 @@ heap_create_with_catalog(const char *relname,
Oid relid,
Oid ownerid,
TupleDesc tupdesc,
+ List *cooked_constraints,
char relkind,
bool shared_relation,
bool oidislocal,
@@ -1004,13 +1011,13 @@ heap_create_with_catalog(const char *relname,
}
/*
- * store constraints and defaults passed in the tupdesc, if any.
+ * Store any supplied constraints and defaults.
*
* NB: this may do a CommandCounterIncrement and rebuild the relcache
* entry, so the relation must be valid and self-consistent at this point.
* In particular, there are not yet constraints and defaults anywhere.
*/
- StoreConstraints(new_rel_desc, tupdesc);
+ StoreConstraints(new_rel_desc, cooked_constraints);
/*
* If there's a special on-commit action, remember it
@@ -1426,12 +1433,11 @@ heap_drop_with_catalog(Oid relid)
/*
* Store a default expression for column attnum of relation rel.
- * The expression must be presented as a nodeToString() string.
*/
void
-StoreAttrDefault(Relation rel, AttrNumber attnum, char *adbin)
+StoreAttrDefault(Relation rel, AttrNumber attnum, Node *expr)
{
- Node *expr;
+ char *adbin;
char *adsrc;
Relation adrel;
HeapTuple tuple;
@@ -1445,12 +1451,12 @@ StoreAttrDefault(Relation rel, AttrNumber attnum, char *adbin)
defobject;
/*
- * Need to construct source equivalent of given node-string.
+ * Flatten expression to string form for storage.
*/
- expr = stringToNode(adbin);
+ adbin = nodeToString(expr);
/*
- * deparse it
+ * Also deparse it to form the mostly-obsolete adsrc field.
*/
adsrc = deparse_expression(expr,
deparse_context_for(RelationGetRelationName(rel),
@@ -1482,6 +1488,7 @@ StoreAttrDefault(Relation rel, AttrNumber attnum, char *adbin)
pfree(DatumGetPointer(values[Anum_pg_attrdef_adbin - 1]));
pfree(DatumGetPointer(values[Anum_pg_attrdef_adsrc - 1]));
heap_freetuple(tuple);
+ pfree(adbin);
pfree(adsrc);
/*
@@ -1525,27 +1532,27 @@ StoreAttrDefault(Relation rel, AttrNumber attnum, char *adbin)
/*
* Store a check-constraint expression for the given relation.
- * The expression must be presented as a nodeToString() string.
*
* Caller is responsible for updating the count of constraints
* in the pg_class entry for the relation.
*/
static void
-StoreRelCheck(Relation rel, char *ccname, char *ccbin)
+StoreRelCheck(Relation rel, char *ccname, Node *expr,
+ bool is_local, int inhcount)
{
- Node *expr;
+ char *ccbin;
char *ccsrc;
List *varList;
int keycount;
int16 *attNos;
/*
- * Convert condition to an expression tree.
+ * Flatten expression to string form for storage.
*/
- expr = stringToNode(ccbin);
+ ccbin = nodeToString(expr);
/*
- * deparse it
+ * Also deparse it to form the mostly-obsolete consrc field.
*/
ccsrc = deparse_expression(expr,
deparse_context_for(RelationGetRelationName(rel),
@@ -1553,7 +1560,7 @@ StoreRelCheck(Relation rel, char *ccname, char *ccbin)
false, false);
/*
- * Find columns of rel that are used in ccbin
+ * Find columns of rel that are used in expr
*
* NB: pull_var_clause is okay here only because we don't allow subselects
* in check constraints; it would fail to examine the contents of
@@ -1608,26 +1615,29 @@ StoreRelCheck(Relation rel, char *ccname, char *ccbin)
InvalidOid, /* no associated index */
expr, /* Tree form check constraint */
ccbin, /* Binary form check constraint */
- ccsrc); /* Source form check constraint */
+ ccsrc, /* Source form check constraint */
+ is_local, /* conislocal */
+ inhcount); /* coninhcount */
+ pfree(ccbin);
pfree(ccsrc);
}
/*
- * Store defaults and constraints passed in via the tuple constraint struct.
+ * Store defaults and constraints (passed as a list of CookedConstraint).
*
* NOTE: only pre-cooked expressions will be passed this way, which is to
* say expressions inherited from an existing relation. Newly parsed
* expressions can be added later, by direct calls to StoreAttrDefault
- * and StoreRelCheck (see AddRelationRawConstraints()).
+ * and StoreRelCheck (see AddRelationNewConstraints()).
*/
static void
-StoreConstraints(Relation rel, TupleDesc tupdesc)
+StoreConstraints(Relation rel, List *cooked_constraints)
{
- TupleConstr *constr = tupdesc->constr;
- int i;
+ int numchecks = 0;
+ ListCell *lc;
- if (!constr)
+ if (!cooked_constraints)
return; /* nothing to do */
/*
@@ -1637,33 +1647,46 @@ StoreConstraints(Relation rel, TupleDesc tupdesc)
*/
CommandCounterIncrement();
- for (i = 0; i < constr->num_defval; i++)
- StoreAttrDefault(rel, constr->defval[i].adnum,
- constr->defval[i].adbin);
+ foreach(lc, cooked_constraints)
+ {
+ CookedConstraint *con = (CookedConstraint *) lfirst(lc);
- for (i = 0; i < constr->num_check; i++)
- StoreRelCheck(rel, constr->check[i].ccname,
- constr->check[i].ccbin);
+ switch (con->contype)
+ {
+ case CONSTR_DEFAULT:
+ StoreAttrDefault(rel, con->attnum, con->expr);
+ break;
+ case CONSTR_CHECK:
+ StoreRelCheck(rel, con->name, con->expr,
+ con->is_local, con->inhcount);
+ numchecks++;
+ break;
+ default:
+ elog(ERROR, "unrecognized constraint type: %d",
+ (int) con->contype);
+ }
+ }
- if (constr->num_check > 0)
- SetRelationNumChecks(rel, constr->num_check);
+ if (numchecks > 0)
+ SetRelationNumChecks(rel, numchecks);
}
/*
- * AddRelationRawConstraints
+ * AddRelationNewConstraints
*
- * Add raw (not-yet-transformed) column default expressions and/or constraint
- * check expressions to an existing relation. This is defined to do both
- * for efficiency in DefineRelation, but of course you can do just one or
- * the other by passing empty lists.
+ * Add new column default expressions and/or constraint check expressions
+ * to an existing relation. This is defined to do both for efficiency in
+ * DefineRelation, but of course you can do just one or the other by passing
+ * empty lists.
*
* rel: relation to be modified
- * rawColDefaults: list of RawColumnDefault structures
- * rawConstraints: list of Constraint nodes
+ * newColDefaults: list of RawColumnDefault structures
+ * newConstraints: list of Constraint nodes
+ * allow_merge: TRUE if check constraints may be merged with existing ones
+ * is_local: TRUE if definition is local, FALSE if it's inherited
*
- * All entries in rawColDefaults will be processed. Entries in rawConstraints
- * will be processed only if they are CONSTR_CHECK type and contain a "raw"
- * expression.
+ * All entries in newColDefaults will be processed. Entries in newConstraints
+ * will be processed only if they are CONSTR_CHECK type.
*
* Returns a list of CookedConstraint nodes that shows the cooked form of
* the default and constraint expressions added to the relation.
@@ -1674,9 +1697,11 @@ StoreConstraints(Relation rel, TupleDesc tupdesc)
* tuples visible.
*/
List *
-AddRelationRawConstraints(Relation rel,
- List *rawColDefaults,
- List *rawConstraints)
+AddRelationNewConstraints(Relation rel,
+ List *newColDefaults,
+ List *newConstraints,
+ bool allow_merge,
+ bool is_local)
{
List *cookedConstraints = NIL;
TupleDesc tupleDesc;
@@ -1715,7 +1740,7 @@ AddRelationRawConstraints(Relation rel,
/*
* Process column default expressions.
*/
- foreach(cell, rawColDefaults)
+ foreach(cell, newColDefaults)
{
RawColumnDefault *colDef = (RawColumnDefault *) lfirst(cell);
Form_pg_attribute atp = rel->rd_att->attrs[colDef->attnum - 1];
@@ -1739,13 +1764,15 @@ AddRelationRawConstraints(Relation rel,
(IsA(expr, Const) &&((Const *) expr)->constisnull))
continue;
- StoreAttrDefault(rel, colDef->attnum, nodeToString(expr));
+ StoreAttrDefault(rel, colDef->attnum, expr);
cooked = (CookedConstraint *) palloc(sizeof(CookedConstraint));
cooked->contype = CONSTR_DEFAULT;
cooked->name = NULL;
cooked->attnum = colDef->attnum;
cooked->expr = expr;
+ cooked->is_local = is_local;
+ cooked->inhcount = is_local ? 0 : 1;
cookedConstraints = lappend(cookedConstraints, cooked);
}
@@ -1754,45 +1781,35 @@ AddRelationRawConstraints(Relation rel,
*/
numchecks = numoldchecks;
checknames = NIL;
- foreach(cell, rawConstraints)
+ foreach(cell, newConstraints)
{
Constraint *cdef = (Constraint *) lfirst(cell);
char *ccname;
- if (cdef->contype != CONSTR_CHECK || cdef->raw_expr == NULL)
+ if (cdef->contype != CONSTR_CHECK)
continue;
- Assert(cdef->cooked_expr == NULL);
- /*
- * Transform raw parsetree to executable expression.
- */
- expr = transformExpr(pstate, cdef->raw_expr);
-
- /*
- * Make sure it yields a boolean result.
- */
- expr = coerce_to_boolean(pstate, expr, "CHECK");
+ if (cdef->raw_expr != NULL)
+ {
+ Assert(cdef->cooked_expr == NULL);
- /*
- * Make sure no outside relations are referred to.
- */
- if (list_length(pstate->p_rtable) != 1)
- ereport(ERROR,
- (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
- errmsg("only table \"%s\" can be referenced in check constraint",
- RelationGetRelationName(rel))));
+ /*
+ * Transform raw parsetree to executable expression, and verify
+ * it's valid as a CHECK constraint.
+ */
+ expr = cookConstraint(pstate, cdef->raw_expr,
+ RelationGetRelationName(rel));
+ }
+ else
+ {
+ Assert(cdef->cooked_expr != NULL);
- /*
- * No subplans or aggregates, either...
- */
- if (pstate->p_hasSubLinks)
- ereport(ERROR,
- (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot use subquery in check constraint")));
- if (pstate->p_hasAggs)
- ereport(ERROR,
- (errcode(ERRCODE_GROUPING_ERROR),
- errmsg("cannot use aggregate function in check constraint")));
+ /*
+ * Here, we assume the parser will only pass us valid CHECK
+ * expressions, so we do no particular checking.
+ */
+ expr = stringToNode(cdef->cooked_expr);
+ }
/*
* Check name uniqueness, or generate a name if none was given.
@@ -1802,15 +1819,6 @@ AddRelationRawConstraints(Relation rel,
ListCell *cell2;
ccname = cdef->name;
- /* Check against pre-existing constraints */
- if (ConstraintNameIsUsed(CONSTRAINT_RELATION,
- RelationGetRelid(rel),
- RelationGetNamespace(rel),
- ccname))
- ereport(ERROR,
- (errcode(ERRCODE_DUPLICATE_OBJECT),
- errmsg("constraint \"%s\" for relation \"%s\" already exists",
- ccname, RelationGetRelationName(rel))));
/* Check against other new constraints */
/* Needed because we don't do CommandCounterIncrement in loop */
foreach(cell2, checknames)
@@ -1821,6 +1829,19 @@ AddRelationRawConstraints(Relation rel,
errmsg("check constraint \"%s\" already exists",
ccname)));
}
+
+ /* save name for future checks */
+ checknames = lappend(checknames, ccname);
+
+ /*
+ * Check against pre-existing constraints. If we are allowed
+ * to merge with an existing constraint, there's no more to
+ * do here. (We omit the duplicate constraint from the result,
+ * which is what ATAddCheckConstraint wants.)
+ */
+ if (MergeWithExistingConstraint(rel, ccname, expr,
+ allow_merge, is_local))
+ continue;
}
else
{
@@ -1855,15 +1876,15 @@ AddRelationRawConstraints(Relation rel,
"check",
RelationGetNamespace(rel),
checknames);
- }
- /* save name for future checks */
- checknames = lappend(checknames, ccname);
+ /* save name for future checks */
+ checknames = lappend(checknames, ccname);
+ }
/*
* OK, store it.
*/
- StoreRelCheck(rel, ccname, nodeToString(expr));
+ StoreRelCheck(rel, ccname, expr, is_local, is_local ? 0 : 1);
numchecks++;
@@ -1872,6 +1893,8 @@ AddRelationRawConstraints(Relation rel,
cooked->name = ccname;
cooked->attnum = 0;
cooked->expr = expr;
+ cooked->is_local = is_local;
+ cooked->inhcount = is_local ? 0 : 1;
cookedConstraints = lappend(cookedConstraints, cooked);
}
@@ -1888,6 +1911,90 @@ AddRelationRawConstraints(Relation rel,
}
/*
+ * Check for a pre-existing check constraint that conflicts with a proposed
+ * new one, and either adjust its conislocal/coninhcount settings or throw
+ * error as needed.
+ *
+ * Returns TRUE if merged (constraint is a duplicate), or FALSE if it's
+ * got a so-far-unique name, or throws error if conflict.
+ */
+static bool
+MergeWithExistingConstraint(Relation rel, char *ccname, Node *expr,
+ bool allow_merge, bool is_local)
+{
+ bool found;
+ Relation conDesc;
+ SysScanDesc conscan;
+ ScanKeyData skey[2];
+ HeapTuple tup;
+
+ /* Search for a pg_constraint entry with same name and relation */
+ conDesc = heap_open(ConstraintRelationId, RowExclusiveLock);
+
+ found = false;
+
+ ScanKeyInit(&skey[0],
+ Anum_pg_constraint_conname,
+ BTEqualStrategyNumber, F_NAMEEQ,
+ CStringGetDatum(ccname));
+
+ ScanKeyInit(&skey[1],
+ Anum_pg_constraint_connamespace,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetNamespace(rel)));
+
+ conscan = systable_beginscan(conDesc, ConstraintNameNspIndexId, true,
+ SnapshotNow, 2, skey);
+
+ while (HeapTupleIsValid(tup = systable_getnext(conscan)))
+ {
+ Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tup);
+
+ if (con->conrelid == RelationGetRelid(rel))
+ {
+ /* Found it. Conflicts if not identical check constraint */
+ if (con->contype == CONSTRAINT_CHECK)
+ {
+ Datum val;
+ bool isnull;
+
+ val = fastgetattr(tup,
+ Anum_pg_constraint_conbin,
+ conDesc->rd_att, &isnull);
+ if (isnull)
+ elog(ERROR, "null conbin for rel %s",
+ RelationGetRelationName(rel));
+ if (equal(expr, stringToNode(TextDatumGetCString(val))))
+ found = true;
+ }
+ if (!found || !allow_merge)
+ ereport(ERROR,
+ (errcode(ERRCODE_DUPLICATE_OBJECT),
+ errmsg("constraint \"%s\" for relation \"%s\" already exists",
+ ccname, RelationGetRelationName(rel))));
+ /* OK to update the tuple */
+ ereport(NOTICE,
+ (errmsg("merging constraint \"%s\" with inherited definition",
+ ccname)));
+ tup = heap_copytuple(tup);
+ con = (Form_pg_constraint) GETSTRUCT(tup);
+ if (is_local)
+ con->conislocal = true;
+ else
+ con->coninhcount++;
+ simple_heap_update(conDesc, &tup->t_self, tup);
+ CatalogUpdateIndexes(conDesc, tup);
+ break;
+ }
+ }
+
+ systable_endscan(conscan);
+ heap_close(conDesc, RowExclusiveLock);
+
+ return found;
+}
+
+/*
* Update the count of constraints in the relation's pg_class tuple.
*
* Caller had better hold exclusive lock on the relation.
@@ -2015,63 +2122,52 @@ cookDefault(ParseState *pstate,
return expr;
}
-
/*
- * Removes all constraints on a relation that match the given name.
- *
- * It is the responsibility of the calling function to acquire a suitable
- * lock on the relation.
+ * Take a raw CHECK constraint expression and convert it to a cooked format
+ * ready for storage.
*
- * Returns: The number of constraints removed.
+ * Parse state must be set up to recognize any vars that might appear
+ * in the expression.
*/
-int
-RemoveRelConstraints(Relation rel, const char *constrName,
- DropBehavior behavior)
+static Node *
+cookConstraint(ParseState *pstate,
+ Node *raw_constraint,
+ char *relname)
{
- int ndeleted = 0;
- Relation conrel;
- SysScanDesc conscan;
- ScanKeyData key[1];
- HeapTuple contup;
-
- /* Grab an appropriate lock on the pg_constraint relation */
- conrel = heap_open(ConstraintRelationId, RowExclusiveLock);
-
- /* Use the index to scan only constraints of the target relation */
- ScanKeyInit(&key[0],
- Anum_pg_constraint_conrelid,
- BTEqualStrategyNumber, F_OIDEQ,
- ObjectIdGetDatum(RelationGetRelid(rel)));
-
- conscan = systable_beginscan(conrel, ConstraintRelidIndexId, true,
- SnapshotNow, 1, key);
+ Node *expr;
/*
- * Scan over the result set, removing any matching entries.
+ * Transform raw parsetree to executable expression.
*/
- while ((contup = systable_getnext(conscan)) != NULL)
- {
- Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(contup);
+ expr = transformExpr(pstate, raw_constraint);
- if (strcmp(NameStr(con->conname), constrName) == 0)
- {
- ObjectAddress conobj;
-
- conobj.classId = ConstraintRelationId;
- conobj.objectId = HeapTupleGetOid(contup);
- conobj.objectSubId = 0;
-
- performDeletion(&conobj, behavior);
+ /*
+ * Make sure it yields a boolean result.
+ */
+ expr = coerce_to_boolean(pstate, expr, "CHECK");
- ndeleted++;
- }
- }
+ /*
+ * Make sure no outside relations are referred to.
+ */
+ if (list_length(pstate->p_rtable) != 1)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+ errmsg("only table \"%s\" can be referenced in check constraint",
+ relname)));
- /* Clean up after the scan */
- systable_endscan(conscan);
- heap_close(conrel, RowExclusiveLock);
+ /*
+ * No subplans or aggregates, either...
+ */
+ if (pstate->p_hasSubLinks)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot use subquery in check constraint")));
+ if (pstate->p_hasAggs)
+ ereport(ERROR,
+ (errcode(ERRCODE_GROUPING_ERROR),
+ errmsg("cannot use aggregate function in check constraint")));
- return ndeleted;
+ return expr;
}
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 2cb91eb69f3..4a68722fa68 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/catalog/index.c,v 1.296 2008/03/26 21:10:37 alvherre Exp $
+ * $PostgreSQL: pgsql/src/backend/catalog/index.c,v 1.297 2008/05/09 23:32:04 tgl Exp $
*
*
* INTERFACE ROUTINES
@@ -716,7 +716,9 @@ index_create(Oid heapRelationId,
InvalidOid, /* no associated index */
NULL, /* no check constraint */
NULL,
- NULL);
+ NULL,
+ true, /* islocal */
+ 0); /* inhcount */
referenced.classId = ConstraintRelationId;
referenced.objectId = conOid;
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 2cc51b01e2b..bb790e5fc41 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/catalog/pg_constraint.c,v 1.40 2008/03/26 21:10:37 alvherre Exp $
+ * $PostgreSQL: pgsql/src/backend/catalog/pg_constraint.c,v 1.41 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -60,7 +60,9 @@ CreateConstraintEntry(const char *constraintName,
Oid indexRelId,
Node *conExpr,
const char *conBin,
- const char *conSrc)
+ const char *conSrc,
+ bool conIsLocal,
+ int conInhCount)
{
Relation conDesc;
Oid conOid;
@@ -145,6 +147,8 @@ CreateConstraintEntry(const char *constraintName,
values[Anum_pg_constraint_confupdtype - 1] = CharGetDatum(foreignUpdateType);
values[Anum_pg_constraint_confdeltype - 1] = CharGetDatum(foreignDeleteType);
values[Anum_pg_constraint_confmatchtype - 1] = CharGetDatum(foreignMatchType);
+ values[Anum_pg_constraint_conislocal - 1] = BoolGetDatum(conIsLocal);
+ values[Anum_pg_constraint_coninhcount - 1] = Int32GetDatum(conInhCount);
if (conkeyArray)
values[Anum_pg_constraint_conkey - 1] = PointerGetDatum(conkeyArray);
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index 3f4d65a0383..0171d1f7338 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -8,7 +8,7 @@
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/catalog/toasting.c,v 1.9 2008/01/01 19:45:48 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/catalog/toasting.c,v 1.10 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -193,6 +193,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid)
toastOid,
rel->rd_rel->relowner,
tupdesc,
+ NIL,
RELKIND_TOASTVALUE,
shared_relation,
true,
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index 0c83e237fe8..196d6baaa41 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -11,7 +11,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/commands/cluster.c,v 1.173 2008/04/13 19:18:14 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/cluster.c,v 1.174 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -639,9 +639,12 @@ make_new_heap(Oid OIDOldHeap, const char *NewName, Oid NewTableSpace)
/*
* Need to make a copy of the tuple descriptor, since
- * heap_create_with_catalog modifies it.
+ * heap_create_with_catalog modifies it. Note that the NewHeap will
+ * not receive any of the defaults or constraints associated with the
+ * OldHeap; we don't need 'em, and there's no reason to spend cycles
+ * inserting them into the catalogs only to delete them.
*/
- tupdesc = CreateTupleDescCopyConstr(OldHeapDesc);
+ tupdesc = CreateTupleDescCopy(OldHeapDesc);
/*
* Use options of the old heap for new heap.
@@ -662,6 +665,7 @@ make_new_heap(Oid OIDOldHeap, const char *NewName, Oid NewTableSpace)
InvalidOid,
OldHeap->rd_rel->relowner,
tupdesc,
+ NIL,
OldHeap->rd_rel->relkind,
OldHeap->rd_rel->relisshared,
true,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 9d6939bbc11..1aaa0bd86fe 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.251 2008/04/24 20:17:50 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/tablecmds.c,v 1.252 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -169,11 +169,10 @@ typedef struct NewColumnValue
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 void add_nonduplicate_constraint(Constraint *cdef,
- ConstrCheck *check, int *ncheck);
+static bool MergeCheckConstraint(List *constraints, char *name, Node *expr);
static bool change_varattnos_walker(Node *node, const AttrNumber *newattno);
+static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel);
+static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel);
static void StoreCatalogInheritance(Oid relationId, List *supers);
static void StoreCatalogInheritance1(Oid relationId, Oid parentOid,
int16 seqNumber, Relation inhRelation);
@@ -201,7 +200,8 @@ 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);
@@ -232,14 +232,18 @@ static void ATExecDropColumn(Relation rel, const char *colName,
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,
@@ -287,6 +291,7 @@ DefineRelation(CreateStmt *stmt, char relkind)
bool localHasOids;
int parentOidCount;
List *rawDefaults;
+ List *cookedDefaults;
Datum reloptions;
ListCell *listptr;
AttrNumber attnum;
@@ -370,67 +375,78 @@ DefineRelation(CreateStmt *stmt, char relkind)
&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 || stmt->constraints)
+ /*
+ * 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;
- int ncheck = 0;
-
- /* make array that's certainly big enough */
- check = (ConstrCheck *)
- palloc((list_length(old_constraints) +
- list_length(stmt->constraints)) * sizeof(ConstrCheck));
- /* deal with constraints from MergeAttributes */
- foreach(listptr, old_constraints)
- {
- Constraint *cdef = (Constraint *) lfirst(listptr);
+ ColumnDef *colDef = lfirst(listptr);
- if (cdef->contype == CONSTR_CHECK)
- add_nonduplicate_constraint(cdef, check, &ncheck);
- }
+ attnum++;
- /*
- * parse_utilcmd.c might have passed some precooked constraints too,
- * due to LIKE tab INCLUDING CONSTRAINTS
- */
- foreach(listptr, stmt->constraints)
+ if (colDef->raw_default != NULL)
{
- Constraint *cdef = (Constraint *) lfirst(listptr);
+ RawColumnDefault *rawEnt;
- if (cdef->contype == CONSTR_CHECK && cdef->cooked_expr != NULL)
- add_nonduplicate_constraint(cdef, check, &ncheck);
+ Assert(colDef->cooked_default == NULL);
+
+ rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));
+ rawEnt->attnum = attnum;
+ rawEnt->raw_default = colDef->raw_default;
+ rawDefaults = lappend(rawDefaults, rawEnt);
+ descriptor->attrs[attnum - 1]->atthasdef = true;
}
- /* if we found any, insert 'em into the descriptor */
- 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,
@@ -463,36 +479,10 @@ DefineRelation(CreateStmt *stmt, char relkind)
* 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
@@ -1046,8 +1036,9 @@ MergeAttributes(List *schema, List *supers, bool istemp,
}
/*
- * 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)
{
@@ -1056,17 +1047,28 @@ MergeAttributes(List *schema, List *supers, bool istemp,
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);
+ }
}
}
@@ -1183,38 +1185,46 @@ MergeAttributes(List *schema, List *supers, bool istemp,
/*
- * In multiple-inheritance situations, it's possible to inherit
- * the same grandparent constraint through multiple parents.
- * Hence, we want to discard inherited constraints that match as to
- * both name and expression. Otherwise, gripe if there are conflicting
- * names. Nonconflicting constraints are added to the array check[]
- * of length *ncheck ... caller must ensure there is room!
+ * MergeCheckConstraint
+ * Try to merge an inherited CHECK constraint with previous ones
+ *
+ * If we inherit identically-named constraints from multiple parents, we must
+ * merge them, or throw an error if they don't have identical definitions.
+ *
+ * constraints is a list of CookedConstraint structs for previous constraints.
+ *
+ * Returns TRUE if merged (constraint is a duplicate), or FALSE if it's
+ * got a so-far-unique name, or throws error if conflict.
*/
-static void
-add_nonduplicate_constraint(Constraint *cdef, ConstrCheck *check, int *ncheck)
+static bool
+MergeCheckConstraint(List *constraints, char *name, Node *expr)
{
- int i;
-
- /* Should only see precooked constraints here */
- Assert(cdef->contype == CONSTR_CHECK);
- Assert(cdef->name != NULL);
- Assert(cdef->raw_expr == NULL && cdef->cooked_expr != NULL);
+ ListCell *lc;
- for (i = 0; i < *ncheck; i++)
+ foreach(lc, constraints)
{
- if (strcmp(check[i].ccname, cdef->name) != 0)
+ CookedConstraint *ccon = (CookedConstraint *) lfirst(lc);
+
+ Assert(ccon->contype == CONSTR_CHECK);
+
+ /* Non-matching names never conflict */
+ if (strcmp(ccon->name, name) != 0)
continue;
- if (strcmp(check[i].ccbin, cdef->cooked_expr) == 0)
- return; /* duplicate constraint, so ignore it */
+
+ if (equal(expr, ccon->expr))
+ {
+ /* OK to merge */
+ ccon->inhcount++;
+ return true;
+ }
+
ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("check constraint name \"%s\" appears multiple times but with different expressions",
- cdef->name)));
+ name)));
}
- /* No match on name, so add it to array */
- check[*ncheck].ccname = cdef->name;
- check[*ncheck].ccbin = pstrdup(cdef->cooked_expr);
- (*ncheck)++;
+
+ return false;
}
@@ -1252,7 +1262,7 @@ change_varattnos_walker(Node *node, const AttrNumber *newattno)
* currently.
*/
Assert(newattno[var->varattno - 1] > 0);
- var->varattno = newattno[var->varattno - 1];
+ var->varattno = var->varoattno = newattno[var->varattno - 1];
}
return false;
}
@@ -1887,8 +1897,9 @@ CheckTableNotInUse(Relation rel, const char *stmt)
* 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
@@ -2044,27 +2055,18 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
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 */
@@ -2181,7 +2183,7 @@ ATRewriteCatalogs(List **wqueue)
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
@@ -2215,7 +2217,8 @@ ATRewriteCatalogs(List **wqueue)
* 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)
{
@@ -2250,13 +2253,16 @@ ATExecCmd(AlteredTableInfo *tab, Relation rel, AlterTableCmd *cmd)
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);
@@ -3272,7 +3278,7 @@ ATExecAddColumn(AlteredTableInfo *tab, Relation rel,
* 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();
@@ -3297,7 +3303,7 @@ ATExecAddColumn(AlteredTableInfo *tab, Relation rel,
* 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.
*/
defval = (Expr *) build_column_default(rel, attribute->attnum);
@@ -3552,7 +3558,7 @@ ATExecColumnDefault(Relation rel, const char *colName,
* 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);
}
}
@@ -3829,7 +3835,7 @@ ATExecDropColumn(Relation rel, const char *colName,
{
/*
* 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--;
@@ -3936,7 +3942,8 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
* 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))
{
@@ -3953,34 +3960,9 @@ ATExecAddConstraint(AlteredTableInfo *tab, Relation rel, Node *newConstraint)
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);
@@ -3992,6 +3974,9 @@ ATExecAddConstraint(AlteredTableInfo *tab, Relation rel, Node *newConstraint)
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)
@@ -4025,6 +4010,107 @@ ATExecAddConstraint(AlteredTableInfo *tab, Relation rel, Node *newConstraint)
}
/*
+ * 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
*
* Subroutine for ATExecAddConstraint. Must already hold exclusive
@@ -4295,7 +4381,9 @@ ATAddForeignKeyConstraint(AlteredTableInfo *tab, Relation rel,
indexOid,
NULL, /* no check constraint */
NULL,
- NULL);
+ NULL,
+ true, /* islocal */
+ 0); /* inhcount */
/*
* Create the triggers that will enforce the constraint.
@@ -4813,46 +4901,186 @@ createForeignKeyTriggers(Relation rel, FkConstraint *fkconstraint,
/*
* 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)
+ ScanKeyInit(&key,
+ Anum_pg_constraint_conrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(rel)));
+ scan = systable_beginscan(conrel, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, &key);
+
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
{
- AlterTableCmd *childCmd = copyObject(cmd);
+ ObjectAddress conobj;
+
+ con = (Form_pg_constraint) GETSTRUCT(tuple);
+
+ if (strcmp(NameStr(con->conname), constrName) != 0)
+ continue;
+
+ /* Don't drop inherited constraints */
+ if (con->coninhcount > 0 && !recursing)
+ ereport(ERROR,
+ (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);
- childCmd->subtype = AT_DropConstraintQuietly;
- ATSimpleRecursion(wqueue, rel, childCmd, recurse);
+ found = true;
}
-}
-static void
-ATExecDropConstraint(Relation rel, const char *constrName,
- DropBehavior behavior, bool quiet)
-{
- int deleted;
+ systable_endscan(scan);
- deleted = RemoveRelConstraints(rel, constrName, behavior);
+ if (!found)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("constraint \"%s\" of relation \"%s\" does not exist",
+ constrName, RelationGetRelationName(rel))));
- if (!quiet)
+ /*
+ * 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)
{
- /* If zero constraints deleted, complain */
- if (deleted == 0)
+ 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, &copy_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, &copy_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\" 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)));
+ errmsg("constraint \"%s\" of relation \"%s\" does not exist",
+ constrName,
+ RelationGetRelationName(childrel))));
+
+ heap_close(childrel, NoLock);
}
+
+ heap_close(conrel, RowExclusiveLock);
}
/*
@@ -5314,7 +5542,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
*/
RemoveAttrDefault(RelationGetRelid(rel), attnum, DROP_RESTRICT, true);
- StoreAttrDefault(rel, attnum, nodeToString(defaultexpr));
+ StoreAttrDefault(rel, attnum, defaultexpr);
}
/* Cleanup */
@@ -6216,10 +6444,10 @@ ATExecAddInherit(Relation child_rel, RangeVar *parent)
RelationGetRelationName(child_rel),
RelationGetRelationName(parent_rel))));
- /* Match up the columns and bump attinhcount and attislocal */
+ /* 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);
/*
@@ -6260,6 +6488,28 @@ decompile_conbin(HeapTuple contup, TupleDesc tupdesc)
}
/*
+ * 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.
*
@@ -6342,7 +6592,8 @@ MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel)
}
/*
- * 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 ATExecAddInherit
*
@@ -6360,91 +6611,87 @@ MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel)
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);
+ ObjectIdGetDatum(RelationGetRelid(parent_rel)));
+ parent_scan = systable_beginscan(catalog_relation, ConstraintRelidIndexId,
+ true, SnapshotNow, 1, &parent_key);
- constraints = NIL;
- 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));
- }
+ /* 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);
- systable_endscan(scan);
+ while (HeapTupleIsValid(child_tuple = systable_getnext(child_scan)))
+ {
+ Form_pg_constraint child_con = (Form_pg_constraint) GETSTRUCT(child_tuple);
+ HeapTuple child_copy;
- /* Then scan through the parent's constraints looking for matches */
- ScanKeyInit(&key,
- Anum_pg_constraint_conrelid,
- BTEqualStrategyNumber, F_OIDEQ,
- ObjectIdGetDatum(RelationGetRelid(parent_rel)));
- scan = systable_beginscan(catalogRelation, ConstraintRelidIndexId, true,
- SnapshotNow, 1, &key);
+ if (child_con->contype != CONSTRAINT_CHECK)
+ continue;
- while (HeapTupleIsValid(constraintTuple = systable_getnext(scan)))
- {
- Form_pg_constraint parent_con = (Form_pg_constraint) GETSTRUCT(constraintTuple);
- bool found = false;
- Form_pg_constraint child_con = NULL;
- HeapTuple child_contuple = NULL;
+ if (strcmp(NameStr(parent_con->conname),
+ NameStr(child_con->conname)) != 0)
+ continue;
- if (parent_con->contype != CONSTRAINT_CHECK)
- 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))));
- foreach(elem, constraints)
- {
- child_contuple = (HeapTuple) lfirst(elem);
- child_con = (Form_pg_constraint) GETSTRUCT(child_contuple);
- if (strcmp(NameStr(parent_con->conname),
- NameStr(child_con->conname)) == 0)
- {
- found = true;
- break;
- }
+ /*
+ * 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;
}
+ systable_endscan(child_scan);
+
if (!found)
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("child table is missing constraint \"%s\"",
NameStr(parent_con->conname))));
-
- if (parent_con->condeferrable != child_con->condeferrable ||
- parent_con->condeferred != child_con->condeferred ||
- strcmp(decompile_conbin(constraintTuple, tupleDesc),
- decompile_conbin(child_contuple, tupleDesc)) != 0)
- ereport(ERROR,
- (errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("constraint definition for check constraint \"%s\" does not match",
- 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);
}
/*
@@ -6459,6 +6706,9 @@ MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel)
* 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
ATExecDropInherit(Relation rel, RangeVar *parent)
@@ -6469,7 +6719,9 @@ ATExecDropInherit(Relation rel, RangeVar *parent)
ScanKeyData key[3];
HeapTuple inheritsTuple,
attributeTuple,
+ constraintTuple,
depTuple;
+ List *connames;
bool found = false;
/*
@@ -6559,6 +6811,81 @@ ATExecDropInherit(Relation rel, RangeVar *parent)
heap_close(catalogRelation, RowExclusiveLock);
/*
+ * 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, &copyTuple->t_self, copyTuple);
+ CatalogUpdateIndexes(catalogRelation, copyTuple);
+ heap_freetuple(copyTuple);
+ }
+ }
+
+ systable_endscan(scan);
+ heap_close(catalogRelation, RowExclusiveLock);
+
+ /*
* Drop the dependency
*
* There's no convenient way to do this, so go trawling through pg_depend
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index 3f557b078b7..880788fd22b 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.117 2008/03/27 03:57:33 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/commands/typecmds.c,v 1.118 2008/05/09 23:32:04 tgl Exp $
*
* DESCRIPTION
* The "DefineFoo" routines take the parse tree and pick out the
@@ -2206,7 +2206,9 @@ domainAddConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid,
InvalidOid,
expr, /* Tree form check constraint */
ccbin, /* Binary form check constraint */
- ccsrc); /* Source form check constraint */
+ ccsrc, /* Source form check constraint */
+ true, /* is local */
+ 0); /* inhcount */
/*
* Return the compiled constraint expression so the calling routine can
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 8b14c04b56d..048542b99b9 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -26,7 +26,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.306 2008/04/21 03:49:45 tgl Exp $
+ * $PostgreSQL: pgsql/src/backend/executor/execMain.c,v 1.307 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -2626,7 +2626,7 @@ OpenIntoRel(QueryDesc *queryDesc)
false);
(void) heap_reloptions(RELKIND_RELATION, reloptions, true);
- /* have to copy the actual tupdesc to get rid of any constraints */
+ /* Copy the tupdesc because heap_create_with_catalog modifies it */
tupdesc = CreateTupleDescCopy(queryDesc->tupDesc);
/* Now we can actually create the new relation */
@@ -2636,6 +2636,7 @@ OpenIntoRel(QueryDesc *queryDesc)
InvalidOid,
GetUserId(),
tupdesc,
+ NIL,
RELKIND_RELATION,
false,
true,
diff --git a/src/bin/pg_dump/common.c b/src/bin/pg_dump/common.c
index 2d41a52e191..8950dee0ca8 100644
--- a/src/bin/pg_dump/common.c
+++ b/src/bin/pg_dump/common.c
@@ -11,7 +11,7 @@
*
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/bin/pg_dump/common.c,v 1.103 2008/03/27 03:57:33 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/pg_dump/common.c,v 1.104 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -62,8 +62,7 @@ static DumpableObject **oprinfoindex;
static void flagInhTables(TableInfo *tbinfo, int numTables,
InhInfo *inhinfo, int numInherits);
-static void flagInhAttrs(TableInfo *tbinfo, int numTables,
- InhInfo *inhinfo, int numInherits);
+static void flagInhAttrs(TableInfo *tblinfo, int numTables);
static DumpableObject **buildIndexArray(void *objArray, int numObjs,
Size objSize);
static int DOCatalogIdCompare(const void *p1, const void *p2);
@@ -191,7 +190,7 @@ getSchemaData(int *numTablesPtr)
if (g_verbose)
write_msg(NULL, "flagging inherited columns in subtables\n");
- flagInhAttrs(tblinfo, numTables, inhinfo, numInherits);
+ flagInhAttrs(tblinfo, numTables);
if (g_verbose)
write_msg(NULL, "reading indexes\n");
@@ -257,8 +256,7 @@ flagInhTables(TableInfo *tblinfo, int numTables,
* modifies tblinfo
*/
static void
-flagInhAttrs(TableInfo *tblinfo, int numTables,
- InhInfo *inhinfo, int numInherits)
+flagInhAttrs(TableInfo *tblinfo, int numTables)
{
int i,
j,
@@ -414,43 +412,6 @@ flagInhAttrs(TableInfo *tblinfo, int numTables,
tbinfo->inhAttrs[j] = false;
}
}
-
- /*
- * Check for inherited CHECK constraints. We assume a constraint is
- * inherited if its name matches the name of any constraint in the
- * parent. Originally this code tried to compare the expression
- * texts, but that can fail if the parent and child tables are in
- * different schemas, because reverse-listing of function calls may
- * produce different text (schema-qualified or not) depending on
- * search path. We really need a more bulletproof way of detecting
- * inherited constraints --- pg_constraint should record this
- * explicitly!
- */
- for (j = 0; j < tbinfo->ncheck; j++)
- {
- ConstraintInfo *constr;
-
- constr = &(tbinfo->checkexprs[j]);
-
- for (k = 0; k < numParents; k++)
- {
- int l;
-
- parent = parents[k];
- for (l = 0; l < parent->ncheck; l++)
- {
- ConstraintInfo *pconstr = &(parent->checkexprs[l]);
-
- if (strcmp(pconstr->dobj.name, constr->dobj.name) == 0)
- {
- constr->coninherited = true;
- break;
- }
- }
- if (constr->coninherited)
- break;
- }
- }
}
}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index f7ac1706de8..18e83f4a9ee 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -12,7 +12,7 @@
* by PostgreSQL
*
* IDENTIFICATION
- * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.c,v 1.489 2008/05/03 23:32:32 adunstan Exp $
+ * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.c,v 1.490 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -120,6 +120,7 @@ static void expand_table_name_patterns(SimpleStringList *patterns,
SimpleOidList *oids);
static NamespaceInfo *findNamespace(Oid nsoid, Oid objoid);
static void dumpTableData(Archive *fout, TableDataInfo *tdinfo);
+static void guessConstraintInheritance(TableInfo *tblinfo, int numTables);
static void dumpComment(Archive *fout, const char *target,
const char *namespace, const char *owner,
CatalogId catalogId, int subid, DumpId dumpId);
@@ -645,6 +646,9 @@ main(int argc, char **argv)
*/
tblinfo = getSchemaData(&numTables);
+ if (g_fout->remoteVersion < 80400)
+ guessConstraintInheritance(tblinfo, numTables);
+
if (!schemaOnly)
getTableData(tblinfo, numTables, oids);
@@ -1384,6 +1388,81 @@ getTableData(TableInfo *tblinfo, int numTables, bool oids)
/*
+ * guessConstraintInheritance:
+ * In pre-8.4 databases, we can't tell for certain which constraints
+ * are inherited. We assume a CHECK constraint is inherited if its name
+ * matches the name of any constraint in the parent. Originally this code
+ * tried to compare the expression texts, but that can fail for various
+ * reasons --- for example, if the parent and child tables are in different
+ * schemas, reverse-listing of function calls may produce different text
+ * (schema-qualified or not) depending on search path.
+ *
+ * In 8.4 and up we can rely on the conislocal field to decide which
+ * constraints must be dumped; much safer.
+ *
+ * This function assumes all conislocal flags were initialized to TRUE.
+ * It clears the flag on anything that seems to be inherited.
+ */
+static void
+guessConstraintInheritance(TableInfo *tblinfo, int numTables)
+{
+ int i,
+ j,
+ k;
+
+ for (i = 0; i < numTables; i++)
+ {
+ TableInfo *tbinfo = &(tblinfo[i]);
+ int numParents;
+ TableInfo **parents;
+ TableInfo *parent;
+
+ /* Sequences and views never have parents */
+ if (tbinfo->relkind == RELKIND_SEQUENCE ||
+ tbinfo->relkind == RELKIND_VIEW)
+ continue;
+
+ /* Don't bother computing anything for non-target tables, either */
+ if (!tbinfo->dobj.dump)
+ continue;
+
+ numParents = tbinfo->numParents;
+ parents = tbinfo->parents;
+
+ if (numParents == 0)
+ continue; /* nothing to see here, move along */
+
+ /* scan for inherited CHECK constraints */
+ for (j = 0; j < tbinfo->ncheck; j++)
+ {
+ ConstraintInfo *constr;
+
+ constr = &(tbinfo->checkexprs[j]);
+
+ for (k = 0; k < numParents; k++)
+ {
+ int l;
+
+ parent = parents[k];
+ for (l = 0; l < parent->ncheck; l++)
+ {
+ ConstraintInfo *pconstr = &(parent->checkexprs[l]);
+
+ if (strcmp(pconstr->dobj.name, constr->dobj.name) == 0)
+ {
+ constr->conislocal = false;
+ break;
+ }
+ }
+ if (!constr->conislocal)
+ break;
+ }
+ }
+ }
+}
+
+
+/*
* dumpDatabase:
* dump the database definition
*/
@@ -3522,7 +3601,7 @@ getIndexes(TableInfo tblinfo[], int numTables)
constrinfo[j].contype = contype;
constrinfo[j].condef = NULL;
constrinfo[j].conindex = indxinfo[j].dobj.dumpId;
- constrinfo[j].coninherited = false;
+ constrinfo[j].conislocal = true;
constrinfo[j].separate = true;
indxinfo[j].indexconstraint = constrinfo[j].dobj.dumpId;
@@ -3623,7 +3702,7 @@ getConstraints(TableInfo tblinfo[], int numTables)
constrinfo[j].contype = 'f';
constrinfo[j].condef = strdup(PQgetvalue(res, j, i_condef));
constrinfo[j].conindex = 0;
- constrinfo[j].coninherited = false;
+ constrinfo[j].conislocal = true;
constrinfo[j].separate = true;
}
@@ -3706,7 +3785,7 @@ getDomainConstraints(TypeInfo *tinfo)
constrinfo[i].contype = 'c';
constrinfo[i].condef = strdup(PQgetvalue(res, i, i_consrc));
constrinfo[i].conindex = 0;
- constrinfo[i].coninherited = false;
+ constrinfo[i].conislocal = true;
constrinfo[i].separate = false;
/*
@@ -4586,10 +4665,22 @@ getTableAttrs(TableInfo *tblinfo, int numTables)
tbinfo->dobj.name);
resetPQExpBuffer(q);
- if (g_fout->remoteVersion >= 70400)
+ if (g_fout->remoteVersion >= 80400)
+ {
+ appendPQExpBuffer(q, "SELECT tableoid, oid, conname, "
+ "pg_catalog.pg_get_constraintdef(oid) AS consrc, "
+ "conislocal "
+ "FROM pg_catalog.pg_constraint "
+ "WHERE conrelid = '%u'::pg_catalog.oid "
+ " AND contype = 'c' "
+ "ORDER BY conname",
+ tbinfo->dobj.catId.oid);
+ }
+ else if (g_fout->remoteVersion >= 70400)
{
appendPQExpBuffer(q, "SELECT tableoid, oid, conname, "
- "pg_catalog.pg_get_constraintdef(oid) AS consrc "
+ "pg_catalog.pg_get_constraintdef(oid) AS consrc, "
+ "true as conislocal "
"FROM pg_catalog.pg_constraint "
"WHERE conrelid = '%u'::pg_catalog.oid "
" AND contype = 'c' "
@@ -4600,7 +4691,8 @@ getTableAttrs(TableInfo *tblinfo, int numTables)
{
/* no pg_get_constraintdef, must use consrc */
appendPQExpBuffer(q, "SELECT tableoid, oid, conname, "
- "'CHECK (' || consrc || ')' AS consrc "
+ "'CHECK (' || consrc || ')' AS consrc, "
+ "true as conislocal "
"FROM pg_catalog.pg_constraint "
"WHERE conrelid = '%u'::pg_catalog.oid "
" AND contype = 'c' "
@@ -4612,7 +4704,8 @@ getTableAttrs(TableInfo *tblinfo, int numTables)
/* 7.2 did not have OIDs in pg_relcheck */
appendPQExpBuffer(q, "SELECT tableoid, 0 as oid, "
"rcname AS conname, "
- "'CHECK (' || rcsrc || ')' AS consrc "
+ "'CHECK (' || rcsrc || ')' AS consrc, "
+ "true as conislocal "
"FROM pg_relcheck "
"WHERE rcrelid = '%u'::oid "
"ORDER BY rcname",
@@ -4622,7 +4715,8 @@ getTableAttrs(TableInfo *tblinfo, int numTables)
{
appendPQExpBuffer(q, "SELECT tableoid, oid, "
"rcname AS conname, "
- "'CHECK (' || rcsrc || ')' AS consrc "
+ "'CHECK (' || rcsrc || ')' AS consrc, "
+ "true as conislocal "
"FROM pg_relcheck "
"WHERE rcrelid = '%u'::oid "
"ORDER BY rcname",
@@ -4634,7 +4728,8 @@ getTableAttrs(TableInfo *tblinfo, int numTables)
appendPQExpBuffer(q, "SELECT "
"(SELECT oid FROM pg_class WHERE relname = 'pg_relcheck') AS tableoid, "
"oid, rcname AS conname, "
- "'CHECK (' || rcsrc || ')' AS consrc "
+ "'CHECK (' || rcsrc || ')' AS consrc, "
+ "true as conislocal "
"FROM pg_relcheck "
"WHERE rcrelid = '%u'::oid "
"ORDER BY rcname",
@@ -4668,7 +4763,7 @@ getTableAttrs(TableInfo *tblinfo, int numTables)
constrs[j].contype = 'c';
constrs[j].condef = strdup(PQgetvalue(res, j, 3));
constrs[j].conindex = 0;
- constrs[j].coninherited = false;
+ constrs[j].conislocal = (PQgetvalue(res, j, 4)[0] == 't');
constrs[j].separate = false;
constrs[j].dobj.dump = tbinfo->dobj.dump;
@@ -4684,8 +4779,8 @@ getTableAttrs(TableInfo *tblinfo, int numTables)
/*
* If the constraint is inherited, this will be detected
- * later. We also detect later if the constraint must be
- * split out from the table definition.
+ * later (in pre-8.4 databases). We also detect later if the
+ * constraint must be split out from the table definition.
*/
}
PQclear(res);
@@ -8840,7 +8935,7 @@ dumpTableSchema(Archive *fout, TableInfo *tbinfo)
{
ConstraintInfo *constr = &(tbinfo->checkexprs[j]);
- if (constr->coninherited || constr->separate)
+ if (constr->separate || !constr->conislocal)
continue;
if (actual_atts > 0)
@@ -8955,7 +9050,7 @@ dumpTableSchema(Archive *fout, TableInfo *tbinfo)
{
ConstraintInfo *constr = &(tbinfo->checkexprs[j]);
- if (constr->coninherited || constr->separate)
+ if (constr->separate || !constr->conislocal)
continue;
dumpTableConstraintComment(fout, constr);
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index d4b40e9d887..8fbe98221b6 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -6,7 +6,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.h,v 1.139 2008/01/01 19:45:55 momjian Exp $
+ * $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.h,v 1.140 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -353,7 +353,7 @@ typedef struct _constraintInfo
char contype;
char *condef; /* definition, if CHECK or FOREIGN KEY */
DumpId conindex; /* identifies associated index if any */
- bool coninherited; /* TRUE if appears to be inherited */
+ bool conislocal; /* TRUE if constraint has local definition */
bool separate; /* TRUE if must dump as separate item */
} ConstraintInfo;
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 9b0cb80e174..fcc9c6c2340 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -37,7 +37,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.457 2008/05/08 08:58:59 mha Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.458 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -53,6 +53,6 @@
*/
/* yyyymmddN */
-#define CATALOG_VERSION_NO 200805081
+#define CATALOG_VERSION_NO 200805091
#endif
diff --git a/src/include/catalog/heap.h b/src/include/catalog/heap.h
index 727af295c76..52347a65522 100644
--- a/src/include/catalog/heap.h
+++ b/src/include/catalog/heap.h
@@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $PostgreSQL: pgsql/src/include/catalog/heap.h,v 1.87 2008/01/01 19:45:56 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/heap.h,v 1.88 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -29,6 +29,8 @@ typedef struct CookedConstraint
char *name; /* name, or NULL if none */
AttrNumber attnum; /* which attr (only for DEFAULT) */
Node *expr; /* transformed default or check expr */
+ bool is_local; /* constraint has local (non-inherited) def */
+ int inhcount; /* number of times constraint is inherited */
} CookedConstraint;
extern Relation heap_create(const char *relname,
@@ -46,6 +48,7 @@ extern Oid heap_create_with_catalog(const char *relname,
Oid relid,
Oid ownerid,
TupleDesc tupdesc,
+ List *cooked_constraints,
char relkind,
bool shared_relation,
bool oidislocal,
@@ -67,11 +70,13 @@ extern void InsertPgClassTuple(Relation pg_class_desc,
Oid new_rel_oid,
Datum reloptions);
-extern List *AddRelationRawConstraints(Relation rel,
- List *rawColDefaults,
- List *rawConstraints);
+extern List *AddRelationNewConstraints(Relation rel,
+ List *newColDefaults,
+ List *newConstraints,
+ bool allow_merge,
+ bool is_local);
-extern void StoreAttrDefault(Relation rel, AttrNumber attnum, char *adbin);
+extern void StoreAttrDefault(Relation rel, AttrNumber attnum, Node *expr);
extern Node *cookDefault(ParseState *pstate,
Node *raw_default,
@@ -79,9 +84,6 @@ extern Node *cookDefault(ParseState *pstate,
int32 atttypmod,
char *attname);
-extern int RemoveRelConstraints(Relation rel, const char *constrName,
- DropBehavior behavior);
-
extern void DeleteRelationTuple(Oid relid);
extern void DeleteAttributeTuples(Oid relid);
extern void RemoveAttributeById(Oid relid, AttrNumber attnum);
diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.h
index a05e2a4a40e..e94067ae953 100644
--- a/src/include/catalog/pg_constraint.h
+++ b/src/include/catalog/pg_constraint.h
@@ -8,7 +8,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $PostgreSQL: pgsql/src/include/catalog/pg_constraint.h,v 1.28 2008/03/27 03:57:34 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_constraint.h,v 1.29 2008/05/09 23:32:04 tgl Exp $
*
* NOTES
* the genbki.sh script reads this file and generates .bki
@@ -71,6 +71,12 @@ CATALOG(pg_constraint,2606)
char confdeltype; /* foreign key's ON DELETE action */
char confmatchtype; /* foreign key's match type */
+ /* Has a local definition (hence, do not drop when coninhcount is 0) */
+ bool conislocal;
+
+ /* Number of times inherited from direct parent relation(s) */
+ int4 coninhcount;
+
/*
* VARIABLE LENGTH FIELDS start here. These fields may be NULL, too.
*/
@@ -125,7 +131,7 @@ typedef FormData_pg_constraint *Form_pg_constraint;
* compiler constants for pg_constraint
* ----------------
*/
-#define Natts_pg_constraint 18
+#define Natts_pg_constraint 20
#define Anum_pg_constraint_conname 1
#define Anum_pg_constraint_connamespace 2
#define Anum_pg_constraint_contype 3
@@ -137,13 +143,15 @@ typedef FormData_pg_constraint *Form_pg_constraint;
#define Anum_pg_constraint_confupdtype 9
#define Anum_pg_constraint_confdeltype 10
#define Anum_pg_constraint_confmatchtype 11
-#define Anum_pg_constraint_conkey 12
-#define Anum_pg_constraint_confkey 13
-#define Anum_pg_constraint_conpfeqop 14
-#define Anum_pg_constraint_conppeqop 15
-#define Anum_pg_constraint_conffeqop 16
-#define Anum_pg_constraint_conbin 17
-#define Anum_pg_constraint_consrc 18
+#define Anum_pg_constraint_conislocal 12
+#define Anum_pg_constraint_coninhcount 13
+#define Anum_pg_constraint_conkey 14
+#define Anum_pg_constraint_confkey 15
+#define Anum_pg_constraint_conpfeqop 16
+#define Anum_pg_constraint_conppeqop 17
+#define Anum_pg_constraint_conffeqop 18
+#define Anum_pg_constraint_conbin 19
+#define Anum_pg_constraint_consrc 20
/* Valid values for contype */
@@ -192,7 +200,9 @@ extern Oid CreateConstraintEntry(const char *constraintName,
Oid indexRelId,
Node *conExpr,
const char *conBin,
- const char *conSrc);
+ const char *conSrc,
+ bool conIsLocal,
+ int conInhCount);
extern void RemoveConstraintById(Oid conId);
extern void RenameConstraintById(Oid conId, const char *newname);
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 61c3df53913..0875b99a224 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
- * $PostgreSQL: pgsql/src/include/nodes/parsenodes.h,v 1.364 2008/04/29 20:44:49 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/nodes/parsenodes.h,v 1.365 2008/05/09 23:32:04 tgl Exp $
*
*-------------------------------------------------------------------------
*/
@@ -892,11 +892,11 @@ typedef enum AlterTableType
AT_AddIndex, /* add index */
AT_ReAddIndex, /* internal to commands/tablecmds.c */
AT_AddConstraint, /* add constraint */
+ AT_AddConstraintRecurse, /* internal to commands/tablecmds.c */
AT_ProcessedConstraint, /* pre-processed add constraint (local in
* parser/parse_utilcmd.c) */
AT_DropConstraint, /* drop constraint */
- AT_DropConstraintQuietly, /* drop constraint, no error/warning (local in
- * commands/tablecmds.c) */
+ AT_DropConstraintRecurse, /* internal to commands/tablecmds.c */
AT_AlterColumnType, /* alter column type */
AT_ChangeOwner, /* change owner */
AT_ClusterOn, /* CLUSTER ON */
diff --git a/src/test/regress/expected/alter_table.out b/src/test/regress/expected/alter_table.out
index 51d5afa81fd..b5af16c558e 100644
--- a/src/test/regress/expected/alter_table.out
+++ b/src/test/regress/expected/alter_table.out
@@ -378,19 +378,21 @@ drop table atacc2 cascade;
NOTICE: drop cascades to table atacc3
NOTICE: drop cascades to constraint foo on table atacc3
drop table atacc1;
--- let's try only to add only to the parent
+-- adding only to a parent is disallowed as of 8.4
create table atacc1 (test int);
-create table atacc2 (test2 int);
-create table atacc3 (test3 int) inherits (atacc1, atacc2);
-alter table only atacc2 add constraint foo check (test2>0);
--- fail and then succeed on atacc2
-insert into atacc2 (test2) values (-3);
+create table atacc2 (test2 int) inherits (atacc1);
+-- fail:
+alter table only atacc1 add constraint foo check (test>0);
+ERROR: constraint must be added to child tables too
+-- ok:
+alter table only atacc2 add constraint foo check (test>0);
+-- check constraint not there on parent
+insert into atacc1 (test) values (-3);
+insert into atacc1 (test) values (3);
+-- check constraint is there on child
+insert into atacc2 (test) values (-3);
ERROR: new row for relation "atacc2" violates check constraint "foo"
-insert into atacc2 (test2) values (3);
--- both succeed on atacc3
-insert into atacc3 (test2) values (-3);
-insert into atacc3 (test2) values (3);
-drop table atacc3;
+insert into atacc2 (test) values (3);
drop table atacc2;
drop table atacc1;
-- test unique constraint adding
@@ -1230,7 +1232,7 @@ alter table p1 add column f2 text;
NOTICE: merging definition of column "f2" for child "c1"
insert into p1 values (1,2,'abc');
insert into c1 values(11,'xyz',33,0); -- should fail
-ERROR: new row for relation "c1" violates check constraint "c1_a1_check"
+ERROR: new row for relation "c1" violates check constraint "p1_a1_check"
insert into c1 values(11,'xyz',33,22);
select * from p1;
f1 | a1 | f2
@@ -1249,7 +1251,7 @@ select * from p1;
drop table p1 cascade;
NOTICE: drop cascades to table c1
-NOTICE: drop cascades to constraint c1_a1_check on table c1
+NOTICE: drop cascades to constraint p1_a1_check on table c1
-- test that operations with a dropped column do not try to reference
-- its datatype
create domain mytype as text;
diff --git a/src/test/regress/expected/inherit.out b/src/test/regress/expected/inherit.out
index f81776fe804..e3acd03c17d 100644
--- a/src/test/regress/expected/inherit.out
+++ b/src/test/regress/expected/inherit.out
@@ -692,3 +692,220 @@ drop function p2text(p2);
drop table c1;
drop table p2;
drop table p1;
+CREATE TABLE ac (aa TEXT);
+alter table ac add constraint ac_check check (aa is not null);
+CREATE TABLE bc (bb TEXT) INHERITS (ac);
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+----------+---------+------------+-------------+------------------
+ ac | ac_check | c | t | 0 | (aa IS NOT NULL)
+ bc | ac_check | c | f | 1 | (aa IS NOT NULL)
+(2 rows)
+
+insert into ac (aa) values (NULL);
+ERROR: new row for relation "ac" violates check constraint "ac_check"
+insert into bc (aa) values (NULL);
+ERROR: new row for relation "bc" violates check constraint "ac_check"
+alter table bc drop constraint ac_check; -- fail, disallowed
+ERROR: cannot drop inherited constraint "ac_check" of relation "bc"
+alter table ac drop constraint ac_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+---------+---------+------------+-------------+--------
+(0 rows)
+
+-- try the unnamed-constraint case
+alter table ac add check (aa is not null);
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+-------------+---------+------------+-------------+------------------
+ ac | ac_aa_check | c | t | 0 | (aa IS NOT NULL)
+ bc | ac_aa_check | c | f | 1 | (aa IS NOT NULL)
+(2 rows)
+
+insert into ac (aa) values (NULL);
+ERROR: new row for relation "ac" violates check constraint "ac_aa_check"
+insert into bc (aa) values (NULL);
+ERROR: new row for relation "bc" violates check constraint "ac_aa_check"
+alter table bc drop constraint ac_aa_check; -- fail, disallowed
+ERROR: cannot drop inherited constraint "ac_aa_check" of relation "bc"
+alter table ac drop constraint ac_aa_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+---------+---------+------------+-------------+--------
+(0 rows)
+
+alter table ac add constraint ac_check check (aa is not null);
+alter table bc no inherit ac;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+----------+---------+------------+-------------+------------------
+ ac | ac_check | c | t | 0 | (aa IS NOT NULL)
+ bc | ac_check | c | t | 0 | (aa IS NOT NULL)
+(2 rows)
+
+alter table bc drop constraint ac_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+----------+---------+------------+-------------+------------------
+ ac | ac_check | c | t | 0 | (aa IS NOT NULL)
+(1 row)
+
+alter table ac drop constraint ac_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+---------+---------+------------+-------------+--------
+(0 rows)
+
+drop table bc;
+drop table ac;
+create table ac (a int constraint check_a check (a <> 0));
+create table bc (a int constraint check_a check (a <> 0), b int constraint check_b check (b <> 0)) inherits (ac);
+NOTICE: merging column "a" with inherited definition
+NOTICE: merging constraint "check_a" with inherited definition
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+---------+---------+------------+-------------+----------
+ ac | check_a | c | t | 0 | (a <> 0)
+ bc | check_a | c | t | 1 | (a <> 0)
+ bc | check_b | c | t | 0 | (b <> 0)
+(3 rows)
+
+drop table bc;
+drop table ac;
+create table ac (a int constraint check_a check (a <> 0));
+create table bc (b int constraint check_b check (b <> 0));
+create table cc (c int constraint check_c check (c <> 0)) inherits (ac, bc);
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc', 'cc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+---------+---------+------------+-------------+----------
+ ac | check_a | c | t | 0 | (a <> 0)
+ bc | check_b | c | t | 0 | (b <> 0)
+ cc | check_a | c | f | 1 | (a <> 0)
+ cc | check_b | c | f | 1 | (b <> 0)
+ cc | check_c | c | t | 0 | (c <> 0)
+(5 rows)
+
+alter table cc no inherit bc;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc', 'cc') order by 1,2;
+ relname | conname | contype | conislocal | coninhcount | consrc
+---------+---------+---------+------------+-------------+----------
+ ac | check_a | c | t | 0 | (a <> 0)
+ bc | check_b | c | t | 0 | (b <> 0)
+ cc | check_a | c | f | 1 | (a <> 0)
+ cc | check_b | c | t | 0 | (b <> 0)
+ cc | check_c | c | t | 0 | (c <> 0)
+(5 rows)
+
+drop table cc;
+drop table bc;
+drop table ac;
+create table p1(f1 int);
+create table p2(f2 int);
+create table c1(f3 int) inherits(p1,p2);
+insert into c1 values(1,-1,2);
+alter table p2 add constraint cc check (f2>0); -- fail
+ERROR: check constraint "cc" is violated by some row
+alter table p2 add check (f2>0); -- check it without a name, too
+ERROR: check constraint "p2_f2_check" is violated by some row
+delete from c1;
+insert into c1 values(1,1,2);
+alter table p2 add check (f2>0);
+insert into c1 values(1,-1,2); -- fail
+ERROR: new row for relation "c1" violates check constraint "p2_f2_check"
+create table c2(f3 int) inherits(p1,p2);
+\d c2
+ Table "public.c2"
+ Column | Type | Modifiers
+--------+---------+-----------
+ f1 | integer |
+ f2 | integer |
+ f3 | integer |
+Check constraints:
+ "p2_f2_check" CHECK (f2 > 0)
+Inherits: p1,
+ p2
+
+create table c3 (f4 int) inherits(c1,c2);
+NOTICE: merging multiple inherited definitions of column "f1"
+NOTICE: merging multiple inherited definitions of column "f2"
+NOTICE: merging multiple inherited definitions of column "f3"
+\d c3
+ Table "public.c3"
+ Column | Type | Modifiers
+--------+---------+-----------
+ f1 | integer |
+ f2 | integer |
+ f3 | integer |
+ f4 | integer |
+Check constraints:
+ "p2_f2_check" CHECK (f2 > 0)
+Inherits: c1,
+ c2
+
+drop table p1 cascade;
+NOTICE: drop cascades to table c2
+NOTICE: drop cascades to table c3
+NOTICE: drop cascades to constraint p2_f2_check on table c3
+NOTICE: drop cascades to constraint p2_f2_check on table c2
+NOTICE: drop cascades to table c1
+NOTICE: drop cascades to constraint p2_f2_check on table c1
+drop table p2 cascade;
+create table pp1 (f1 int);
+create table cc1 (f2 text, f3 int) inherits (pp1);
+alter table pp1 add column a1 int check (a1 > 0);
+\d cc1
+ Table "public.cc1"
+ Column | Type | Modifiers
+--------+---------+-----------
+ f1 | integer |
+ f2 | text |
+ f3 | integer |
+ a1 | integer |
+Check constraints:
+ "pp1_a1_check" CHECK (a1 > 0)
+Inherits: pp1
+
+create table cc2(f4 float) inherits(pp1,cc1);
+NOTICE: merging multiple inherited definitions of column "f1"
+NOTICE: merging multiple inherited definitions of column "a1"
+\d cc2
+ Table "public.cc2"
+ Column | Type | Modifiers
+--------+------------------+-----------
+ f1 | integer |
+ a1 | integer |
+ f2 | text |
+ f3 | integer |
+ f4 | double precision |
+Check constraints:
+ "pp1_a1_check" CHECK (a1 > 0)
+Inherits: pp1,
+ cc1
+
+alter table pp1 add column a2 int check (a2 > 0);
+NOTICE: merging definition of column "a2" for child "cc2"
+NOTICE: merging constraint "pp1_a2_check" with inherited definition
+\d cc2
+ Table "public.cc2"
+ Column | Type | Modifiers
+--------+------------------+-----------
+ f1 | integer |
+ a1 | integer |
+ f2 | text |
+ f3 | integer |
+ f4 | double precision |
+ a2 | integer |
+Check constraints:
+ "pp1_a1_check" CHECK (a1 > 0)
+ "pp1_a2_check" CHECK (a2 > 0)
+Inherits: pp1,
+ cc1
+
+drop table pp1 cascade;
+NOTICE: drop cascades to table cc2
+NOTICE: drop cascades to constraint pp1_a1_check on table cc2
+NOTICE: drop cascades to constraint pp1_a2_check on table cc2
+NOTICE: drop cascades to table cc1
+NOTICE: drop cascades to constraint pp1_a1_check on table cc1
+NOTICE: drop cascades to constraint pp1_a2_check on table cc1
diff --git a/src/test/regress/sql/alter_table.sql b/src/test/regress/sql/alter_table.sql
index 81cc70612d5..46aacd1bef8 100644
--- a/src/test/regress/sql/alter_table.sql
+++ b/src/test/regress/sql/alter_table.sql
@@ -389,19 +389,20 @@ select test2 from atacc2;
drop table atacc2 cascade;
drop table atacc1;
--- let's try only to add only to the parent
+-- adding only to a parent is disallowed as of 8.4
create table atacc1 (test int);
-create table atacc2 (test2 int);
-create table atacc3 (test3 int) inherits (atacc1, atacc2);
-alter table only atacc2 add constraint foo check (test2>0);
--- fail and then succeed on atacc2
-insert into atacc2 (test2) values (-3);
-insert into atacc2 (test2) values (3);
--- both succeed on atacc3
-insert into atacc3 (test2) values (-3);
-insert into atacc3 (test2) values (3);
-drop table atacc3;
+create table atacc2 (test2 int) inherits (atacc1);
+-- fail:
+alter table only atacc1 add constraint foo check (test>0);
+-- ok:
+alter table only atacc2 add constraint foo check (test>0);
+-- check constraint not there on parent
+insert into atacc1 (test) values (-3);
+insert into atacc1 (test) values (3);
+-- check constraint is there on child
+insert into atacc2 (test) values (-3);
+insert into atacc2 (test) values (3);
drop table atacc2;
drop table atacc1;
diff --git a/src/test/regress/sql/inherit.sql b/src/test/regress/sql/inherit.sql
index b0499a64928..1730a485756 100644
--- a/src/test/regress/sql/inherit.sql
+++ b/src/test/regress/sql/inherit.sql
@@ -196,3 +196,83 @@ drop function p2text(p2);
drop table c1;
drop table p2;
drop table p1;
+
+CREATE TABLE ac (aa TEXT);
+alter table ac add constraint ac_check check (aa is not null);
+CREATE TABLE bc (bb TEXT) INHERITS (ac);
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+
+insert into ac (aa) values (NULL);
+insert into bc (aa) values (NULL);
+
+alter table bc drop constraint ac_check; -- fail, disallowed
+alter table ac drop constraint ac_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+
+-- try the unnamed-constraint case
+alter table ac add check (aa is not null);
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+
+insert into ac (aa) values (NULL);
+insert into bc (aa) values (NULL);
+
+alter table bc drop constraint ac_aa_check; -- fail, disallowed
+alter table ac drop constraint ac_aa_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+
+alter table ac add constraint ac_check check (aa is not null);
+alter table bc no inherit ac;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+alter table bc drop constraint ac_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+alter table ac drop constraint ac_check;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+
+drop table bc;
+drop table ac;
+
+create table ac (a int constraint check_a check (a <> 0));
+create table bc (a int constraint check_a check (a <> 0), b int constraint check_b check (b <> 0)) inherits (ac);
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc') order by 1,2;
+
+drop table bc;
+drop table ac;
+
+create table ac (a int constraint check_a check (a <> 0));
+create table bc (b int constraint check_b check (b <> 0));
+create table cc (c int constraint check_c check (c <> 0)) inherits (ac, bc);
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc', 'cc') order by 1,2;
+
+alter table cc no inherit bc;
+select pc.relname, pgc.conname, pgc.contype, pgc.conislocal, pgc.coninhcount, pgc.consrc from pg_class as pc inner join pg_constraint as pgc on (pgc.conrelid = pc.oid) where pc.relname in ('ac', 'bc', 'cc') order by 1,2;
+
+drop table cc;
+drop table bc;
+drop table ac;
+
+create table p1(f1 int);
+create table p2(f2 int);
+create table c1(f3 int) inherits(p1,p2);
+insert into c1 values(1,-1,2);
+alter table p2 add constraint cc check (f2>0); -- fail
+alter table p2 add check (f2>0); -- check it without a name, too
+delete from c1;
+insert into c1 values(1,1,2);
+alter table p2 add check (f2>0);
+insert into c1 values(1,-1,2); -- fail
+create table c2(f3 int) inherits(p1,p2);
+\d c2
+create table c3 (f4 int) inherits(c1,c2);
+\d c3
+drop table p1 cascade;
+drop table p2 cascade;
+
+create table pp1 (f1 int);
+create table cc1 (f2 text, f3 int) inherits (pp1);
+alter table pp1 add column a1 int check (a1 > 0);
+\d cc1
+create table cc2(f4 float) inherits(pp1,cc1);
+\d cc2
+alter table pp1 add column a2 int check (a2 > 0);
+\d cc2
+drop table pp1 cascade;